diff --git a/src/rust/shard_manager/src/lib.rs b/src/rust/shard_manager/src/lib.rs index b96b30bb..46c26697 100644 --- a/src/rust/shard_manager/src/lib.rs +++ b/src/rust/shard_manager/src/lib.rs @@ -26,17 +26,76 @@ pub type OpsCb = extern "C" fn(*const c_char) -> *const c_char; pub const OP_ENSURE_SCHEMA: &str = "ensure_schema"; pub const OP_COUNT_ENTRIES: &str = "count_entries"; -// ── doc-envelope FFI plumbing ──────────────────────────────────────────────── - -fn read_arg(ptr: *const c_char) -> Result { - if ptr.is_null() { - return Err("null argument pointer".into()); - } - let x = unsafe { CStr::from_ptr(ptr) }.to_str().map_err(|e| e.to_string())?; - pyjson::parse(x) +/// Typed kernel error (Phase B): every failure mode the port can produce, +/// exhaustively matched at the envelope boundary. +#[derive(Debug)] +pub enum SmError { + /// fs / os failure (makedirs, read, write) + Io(std::io::Error), + /// doc-envelope JSON problem (parse, missing/badly-typed field) + Doc(String), + /// the ops seam rejected an op the kernel had to have + OpRejected(&'static str), } -fn envelope(r: Result) -> *mut c_char { +impl std::fmt::Display for SmError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SmError::Io(e) => write!(f, "io error: {}", e), + SmError::Doc(m) => write!(f, "doc error: {}", m), + SmError::OpRejected(op) => write!(f, "{} op failed", op), + } + } +} + +impl std::error::Error for SmError {} + +impl From for SmError { + fn from(e: std::io::Error) -> Self { + SmError::Io(e) + } +} + +fn doc_err(msg: impl Into) -> Result { + Err(SmError::Doc(msg.into())) +} + +/// One seam op the kernel can issue. Exhaustive: adding an op forces a +/// compile error here (name + args live in one place). +enum OpKind { + EnsureSchema { path: String }, + CountEntries { path: String }, +} + +impl OpKind { + fn name(&self) -> &'static str { + match self { + OpKind::EnsureSchema { .. } => OP_ENSURE_SCHEMA, + OpKind::CountEntries { .. } => OP_COUNT_ENTRIES, + } + } + + fn args(&self) -> Json { + let path = match self { + OpKind::EnsureSchema { path } | OpKind::CountEntries { path } => path, + }; + Json::Obj(vec![("path".into(), Json::Str(path.clone()))]) + } +} + +// ── doc-envelope FFI plumbing ──────────────────────────────────────────────── + +fn read_arg(ptr: *const c_char) -> Result { + if ptr.is_null() { + return doc_err("null argument pointer"); + } + let x = unsafe { CStr::from_ptr(ptr) } + .to_str() + .map_err(|e| SmError::Doc(e.to_string()))?; + pyjson::parse(x).map_err(SmError::Doc) +} + +fn envelope(r: Result) -> *mut c_char { let v = match r { Ok(x) => Json::Obj(vec![ ("ok".into(), Json::Bool(true)), @@ -44,7 +103,7 @@ fn envelope(r: Result) -> *mut c_char { ]), Err(e) => Json::Obj(vec![ ("ok".into(), Json::Bool(false)), - ("err".into(), Json::Str(e)), + ("err".into(), Json::Str(e.to_string())), ]), }; CString::new(pyjson::dumps(&v)) @@ -54,10 +113,10 @@ fn envelope(r: Result) -> *mut c_char { /// Issue one op through the callback; `None` on any failure (kernel then /// degrades exactly like the oracle's best-effort branches). -fn issue_op(cb: OpsCb, op: &str, args: Json) -> Option { +fn try_op(cb: OpsCb, kind: &OpKind) -> Option { let payload = CString::new(pyjson::dumps(&Json::Obj(vec![ - ("op".into(), Json::Str(op.into())), - ("args".into(), args), + ("op".into(), Json::Str(kind.name().into())), + ("args".into(), kind.args()), ]))) .ok()?; let ret = cb(payload.as_ptr()); @@ -99,29 +158,38 @@ fn join_path(dir: &str, name: &str) -> String { // ── kernel (name-preserving port of the 9 oracle fns) ─────────────────────── /// __init__: makedirs(SHARD_DIR, exist_ok=True) + _load_index() -pub fn init(shard_dir: &str, index_path: &str) -> Result { - std::fs::create_dir_all(shard_dir).map_err(|e| e.to_string())?; +pub fn init(shard_dir: &str, index_path: &str) -> Result { + std::fs::create_dir_all(shard_dir)?; load_index(index_path) } /// _load_index: missing file → {}; JSONDecodeError → {}; else the parsed dict. -pub fn load_index(index_path: &str) -> Result { +pub fn load_index(index_path: &str) -> Result { let p = std::path::Path::new(index_path); if !p.exists() { return Ok(Json::Obj(Vec::new())); } - let raw = std::fs::read_to_string(p).map_err(|e| e.to_string())?; + let raw = std::fs::read_to_string(p)?; Ok(pyjson::parse(&raw).unwrap_or(Json::Obj(Vec::new()))) } /// _save_index: json.dump(self.index, f, indent=2) — no trailing newline. -pub fn save_index(index_path: &str, index: &Json) -> Result<(), String> { - std::fs::write(index_path, pyjson::dump_indent2(index)).map_err(|e| e.to_string()) +pub fn save_index(index_path: &str, index: &Json) -> Result<(), SmError> { + std::fs::write(index_path, pyjson::dump_indent2(index))?; + Ok(()) } -/// Shard file name for a topic (oracle: replace(' ', '_').replace('/', '_')). +/// Shard file name for a topic (oracle: replace(' ', '_').replace('/', '_') +/// — both chars map to the same target, so one pass is faithful). pub fn shard_name(topic: &str) -> String { - format!("shard_{}.db", topic.replace([' ', '/'], "_")) + let normalized: String = topic + .chars() + .map(|c| match c { + ' ' | '/' => '_', + other => other, + }) + .collect(); + format!("shard_{}.db", normalized) } /// create_shard: schema creation via the ensure_schema op; early-exists @@ -133,16 +201,15 @@ pub fn create_shard( topic: &str, index: &Json, cb: OpsCb, -) -> Result<(String, Json), String> { +) -> Result<(String, Json), SmError> { let shard_path = join_path(shard_dir, &shard_name(topic)); if std::path::Path::new(&shard_path).exists() { return Ok((shard_path, index.clone())); } - issue_op(cb, OP_ENSURE_SCHEMA, Json::Obj(vec![( - "path".into(), - Json::Str(shard_path.clone()), - )])) - .ok_or_else(|| "ensure_schema op failed".to_string())?; + try_op(cb, &OpKind::EnsureSchema { + path: shard_path.clone(), + }) + .ok_or(SmError::OpRejected(OP_ENSURE_SCHEMA))?; let mut idx = index.clone(); index_set(&mut idx, topic, &shard_path); save_index(index_path, &idx)?; @@ -150,25 +217,23 @@ pub fn create_shard( } /// route_query: keyword overlap over the index (insertion order, strictly -/// greater wins → first topic on ties), then the >3-char 3-word fallback. +/// greater wins → first topic on ties — hence fold, not max_by), then the +/// >3-char 3-word fallback. pub fn route_query(query: &str, index: &Json) -> String { let query_lower = query.to_lowercase(); let query_words: HashSet<&str> = query_lower.split_whitespace().collect(); - let mut best_topic: Option<&str> = None; - let mut best_score: i64 = 0; - if let Json::Obj(pairs) = index { - for (topic, _) in pairs { - let topic_lower = topic.to_lowercase(); - let topic_words: HashSet<&str> = topic_lower.split('_').collect(); - let overlap = topic_words.intersection(&query_words).count() as i64; - if overlap > best_score { - best_score = overlap; - best_topic = Some(topic.as_str()); - } + let best = index_pairs(index).fold((None, 0i64), |(best_topic, best_score), topic| { + let topic_lower = topic.to_lowercase(); + let topic_words: HashSet<&str> = topic_lower.split('_').collect(); + let overlap = topic_words.intersection(&query_words).count() as i64; + if overlap > best_score { + (Some(topic), overlap) + } else { + (best_topic, best_score) } - } - if let (Some(topic), true) = (best_topic, best_score > 0) { + }); + if let (Some(topic), true) = (best.0, best.1 > 0) { return topic.to_string(); } @@ -183,6 +248,15 @@ pub fn route_query(query: &str, index: &Json) -> String { "general".to_string() } +/// Borrowed (topic, path) iteration over an index doc — zero-copy view. +fn index_pairs(index: &Json) -> impl Iterator { + let mut topics: Vec<&str> = Vec::new(); + if let Json::Obj(pairs) = index { + topics.extend(pairs.iter().map(|(k, _)| k.as_str())); + } + topics.into_iter() +} + fn merge_key(x: &Json) -> f64 { for key in ["relevance", "wash_count"] { if let Some(Json::Num(n)) = x.get(key) { @@ -228,31 +302,29 @@ pub fn get_shard( topic: &str, index: &Json, cb: OpsCb, -) -> Result<(String, Json), String> { - let existing = index.get(topic).and_then(|v| match v { - Json::Str(s) => Some(s.clone()), - _ => None, - }); - if let Some(path) = existing { - return Ok((path, index.clone())); +) -> Result<(String, Json), SmError> { + // zero-copy hit path: borrow the stored path, clone only at the boundary + if let Json::Obj(pairs) = index { + if let Some((_, Json::Str(path))) = pairs.iter().find(|(k, _)| k == topic) { + return Ok((path.clone(), index.clone())); + } } create_shard(shard_dir, index_path, topic, index, cb) } /// get_shard_stats: one count_entries op per topic, index order; op failure /// degrades to 0 (oracle except branch). -pub fn get_shard_stats(index: &Json, cb: OpsCb) -> Result { +pub fn get_shard_stats(index: &Json, cb: OpsCb) -> Result { let mut stats = Vec::new(); if let Json::Obj(pairs) = index { for (topic, path) in pairs { let path = match path { - Json::Str(s) => s.clone(), - _ => String::new(), + Json::Str(s) => s.as_str(), + _ => "", }; - let count = issue_op(cb, OP_COUNT_ENTRIES, Json::Obj(vec![( - "path".into(), - Json::Str(path), - )])) + let count = try_op(cb, &OpKind::CountEntries { + path: path.to_string(), + }) .and_then(|v| match v { Json::Num(n) => Some(n as i64), _ => None, @@ -265,16 +337,18 @@ pub fn get_shard_stats(index: &Json, cb: OpsCb) -> Result { } /// total_entries: sum of get_shard_stats values. -pub fn total_entries(index: &Json, cb: OpsCb) -> Result { +pub fn total_entries(index: &Json, cb: OpsCb) -> Result { let stats = get_shard_stats(index, cb)?; - let mut total: i64 = 0; - if let Json::Obj(pairs) = &stats { - for (_, v) in pairs { - if let Json::Num(n) = v { - total += *n as i64; - } - } - } + let total = match &stats { + Json::Obj(pairs) => pairs + .iter() + .filter_map(|(_, v)| match v { + Json::Num(n) => Some(*n as i64), + _ => None, + }) + .sum(), + _ => 0, + }; Ok(total) } @@ -282,7 +356,7 @@ pub fn total_entries(index: &Json, cb: OpsCb) -> Result { #[no_mangle] pub extern "C" fn sm_init(doc_ptr: *const c_char) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; let dir = str_field(&doc, "shard_dir")?; let idx_path = str_field(&doc, "index_path")?; @@ -293,7 +367,7 @@ pub extern "C" fn sm_init(doc_ptr: *const c_char) -> *mut c_char { #[no_mangle] pub extern "C" fn sm_load_index(doc_ptr: *const c_char) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; load_index(str_field(&doc, "index_path")?) }; @@ -302,7 +376,7 @@ pub extern "C" fn sm_load_index(doc_ptr: *const c_char) -> *mut c_char { #[no_mangle] pub extern "C" fn sm_save_index(doc_ptr: *const c_char) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; let idx_path = str_field(&doc, "index_path")?; let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new())); @@ -314,7 +388,7 @@ pub extern "C" fn sm_save_index(doc_ptr: *const c_char) -> *mut c_char { #[no_mangle] pub extern "C" fn sm_create_shard(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; let dir = str_field(&doc, "shard_dir")?; let idx_path = str_field(&doc, "index_path")?; @@ -331,7 +405,7 @@ pub extern "C" fn sm_create_shard(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_c #[no_mangle] pub extern "C" fn sm_route_query(doc_ptr: *const c_char) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; let query = str_field(&doc, "query")?; let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new())); @@ -342,7 +416,7 @@ pub extern "C" fn sm_route_query(doc_ptr: *const c_char) -> *mut c_char { #[no_mangle] pub extern "C" fn sm_merge_results(doc_ptr: *const c_char) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; let results_list = doc.get("results_list").cloned().unwrap_or(Json::Arr(Vec::new())); Ok(merge_results(&results_list)) @@ -352,7 +426,7 @@ pub extern "C" fn sm_merge_results(doc_ptr: *const c_char) -> *mut c_char { #[no_mangle] pub extern "C" fn sm_get_shard(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; let dir = str_field(&doc, "shard_dir")?; let idx_path = str_field(&doc, "index_path")?; @@ -369,7 +443,7 @@ pub extern "C" fn sm_get_shard(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char #[no_mangle] pub extern "C" fn sm_get_shard_stats(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new())); get_shard_stats(&index, cb) @@ -379,7 +453,7 @@ pub extern "C" fn sm_get_shard_stats(doc_ptr: *const c_char, cb: OpsCb) -> *mut #[no_mangle] pub extern "C" fn sm_total_entries(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char { - let run = || -> Result { + let run = || -> Result { let doc = read_arg(doc_ptr)?; let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new())); Ok(Json::Num(total_entries(&index, cb)? as f64)) @@ -387,10 +461,10 @@ pub extern "C" fn sm_total_entries(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_ envelope(run()) } -fn str_field<'a>(doc: &'a Json, key: &str) -> Result<&'a str, String> { +fn str_field<'a>(doc: &'a Json, key: &str) -> Result<&'a str, SmError> { match doc.get(key) { Some(Json::Str(s)) => Ok(s), - _ => Err(format!("missing string field '{}'", key)), + _ => doc_err(format!("missing string field '{}'", key)), } }