diff --git a/cubestore/src/lib.rs b/cubestore/src/lib.rs index da65d94..b725eb0 100644 --- a/cubestore/src/lib.rs +++ b/cubestore/src/lib.rs @@ -411,6 +411,13 @@ impl CubeStore { let raw_fallback = |bytes: &[u8]| { let mut h = CubeHeader::new(); h.size_bytes = Some(bytes.len() as u64); + // Keep the synthesized header's flag bits consistent with its + // fields, so associative queries (`scan_by_flag(SIZE_BYTES)`) see + // raw records too. Only the derived low bits are recomputed; any + // out-of-band bits already on the payload-derived header (none + // here, since raw payloads carry no header) are preserved by + // `refresh_flags`. + h.refresh_flags(); Some((h, bytes.to_vec())) }; if raw.len() < 4 { @@ -500,6 +507,78 @@ impl CubeStore { out.sort(); out } + + /// PDF Package 2 API: `scan_by_flag` — associative storage lookup. + /// + /// Returns every coordinate whose decoded [`CubeHeader`] carries `flag` + /// set. This is the "query by metatag/flag, not by path" primitive the + /// source PDF describes (§"associative storage"): records are addressed + /// by *what they are* (a flag bit) rather than *where they live* (a + /// coordinate path). `cube-os-*` services that emit typed records + /// (e.g. a log line tagged `doc_type = "klog"`) become discoverable by + /// event type without knowing their CZYX address in advance. + /// + /// Records written through the RAW path (`rawput`, used by the OS layers) + /// carry only a synthesized `size_bytes` header, so they will NOT match + /// a content flag unless the writer also set one. The associative query + /// is therefore most powerful when records are written through + /// [`CubeStore::put_record`] with a populated [`CubeHeader`]. + pub fn scan_by_flag(&self, flag: u16) -> Vec { + let mut out: Vec = self + .backend + .keys() + .into_iter() + .filter(|k| { + self.get_record(k) + .map(|(h, _)| h.flags.has(flag)) + .unwrap_or(false) + }) + .collect(); + out.sort(); + out + } + + /// PDF Package 2 API: `scan_by_type` — query records by `doc_type` + /// (the "event type" / file-extension analogue the PDF calls Flag 2). + /// + /// This is the concrete "log lookup by query" the OS layers want: instead + /// of addressing `/cubefs/c200/z004/...` directly, you ask "every record + /// whose doc_type is `klog`" and get back all matching coordinates. The + /// comparison is exact-match (case-sensitive) against the header field. + pub fn scan_by_type(&self, doc_type: &str) -> Vec { + let mut out: Vec = self + .backend + .keys() + .into_iter() + .filter(|k| { + self.get_record(k) + .map(|(h, _)| h.doc_type.as_deref() == Some(doc_type)) + .unwrap_or(false) + }) + .collect(); + out.sort(); + out + } + + /// Return `(label, header, body)` for every record matching `flag`. + /// Convenience wrapper over [`CubeStore::scan_by_flag`] that also pulls + /// the decoded payload so a caller (e.g. a `cubelog` query tool) can + /// present the matching records directly. + pub fn query_by_flag(&self, flag: u16) -> Vec<(Czyx, CubeHeader, Vec)> { + self.scan_by_flag(flag) + .into_iter() + .filter_map(|k| self.get_record(&k).map(|(h, b)| (k, h, b))) + .collect() + } + + /// Return `(label, header, body)` for every record whose `doc_type` + /// matches. See [`CubeStore::scan_by_type`] for the matching semantics. + pub fn query_by_type(&self, doc_type: &str) -> Vec<(Czyx, CubeHeader, Vec)> { + self.scan_by_type(doc_type) + .into_iter() + .filter_map(|k| self.get_record(&k).map(|(h, b)| (k, h, b))) + .collect() + } } #[cfg(test)] @@ -594,4 +673,64 @@ mod tests { "must decode the true header, not fall back to raw" ); } + + /// PDF Package 2: `scan_by_flag` finds records by flag bit, independent of + /// their coordinate address. A record tagged by `DOC_TYPE` must surface + /// when queried for that flag and be absent otherwise. + #[test] + fn scan_by_flag_finds_typed_records() { + let mut store = CubeStore::new(HashBackend::new()); + + // A "klog" typed event record. + let mut klog = CubeHeader::new(); + klog.doc_type = Some("klog".into()); + klog.refresh_flags(); + store.put_record(Czyx::new(200, 4, 2, 1), &klog, b"kernel: eth0 up"); + + // An untyped raw record (the common OS-layer case). + store.put_raw(Czyx::new(200, 70, 1, 1), b"118".to_vec()); + + let by_doc_type = store.scan_by_flag(cubecoords::HeaderFlags::DOC_TYPE); + assert_eq!(by_doc_type, vec![Czyx::new(200, 4, 2, 1)]); + + // The raw record carries only a synthesized size flag, so it must NOT + // match DOC_TYPE. + assert!(!by_doc_type.contains(&Czyx::new(200, 70, 1, 1))); + + let by_size = store.scan_by_flag(cubecoords::HeaderFlags::SIZE_BYTES); + assert!( + !by_size.is_empty(), + "synthesized raw headers carry SIZE_BYTES" + ); + } + + /// `scan_by_type` is the concrete "log lookup by query" — pull every record + /// of a given event type without knowing its coordinate. + #[test] + fn scan_by_type_event_lookup() { + let mut store = CubeStore::new(HashBackend::new()); + let mut a = CubeHeader::new(); + a.doc_type = Some("klog".into()); + let mut b = CubeHeader::new(); + b.doc_type = Some("klog".into()); + let mut c = CubeHeader::new(); + c.doc_type = Some("state".into()); + store.put_record(Czyx::new(200, 4, 2, 1), &a, b"k1"); + store.put_record(Czyx::new(200, 4, 2, 2), &b, b"k2"); + store.put_record(Czyx::new(200, 1, 1, 1), &c, b"s1"); + + let klogs = store.scan_by_type("klog"); + assert_eq!(klogs.len(), 2); + assert!(klogs.contains(&Czyx::new(200, 4, 2, 1))); + assert!(klogs.contains(&Czyx::new(200, 4, 2, 2))); + + // query_by_type returns the decoded payloads too. + let klogs_full = store.query_by_type("klog"); + assert_eq!(klogs_full.len(), 2); + assert!(klogs_full.iter().any(|(_, _, b)| b == b"k1")); + assert!(klogs_full.iter().any(|(_, _, b)| b == b"k2")); + + assert_eq!(store.scan_by_type("state"), vec![Czyx::new(200, 1, 1, 1)]); + assert!(store.scan_by_type("nonexistent").is_empty()); + } } diff --git a/cubesys/src/commands.rs b/cubesys/src/commands.rs index cf2c767..29feea6 100644 --- a/cubesys/src/commands.rs +++ b/cubesys/src/commands.rs @@ -939,6 +939,62 @@ impl Session { store.delete_raw(&coord); Ok(format!("ok: deleted {}", coord.pack_u32())) } + // --- Enveloped coordinate API (PDF Package 2: `put(C,Z,Y,X, + // bytes, flags)`). Unlike `rawput`, this writes a full record + // with a `CubeHeader`, so the record carries metatags + // (doc_type / title) that the associative `query` verb can + // target by event type. `rawput` stays for bulk/append payloads + // (e.g. the rolling kernel-log window) that do not need header + // metadata; `put` is for addressed, discoverable records. --- + "put" => { + let c = parse_u8(it.next(), "put needs ")?; + let z = parse_u8(it.next(), "put needs ")?; + let y = parse_u8(it.next(), "put needs ")?; + let x = parse_u8(it.next(), "put needs ")?; + let hex = it + .next() + .ok_or_else(|| "put needs ".to_string())?; + let val = hex_decode(hex).ok_or_else(|| "put: value must be hex".to_string())?; + let coord = Czyx::new(c, z, y, x); + // Optional metatags: `doc_type=` and/or `title=` may + // follow. These are what `query ` matches on. + let mut header = CubeHeader::new(); + while let Some(tok) = it.next() { + if let Some(v) = tok.strip_prefix("doc_type=") { + header.doc_type = Some(v.to_string()); + } else if let Some(v) = tok.strip_prefix("title=") { + header.title = Some(v.to_string()); + } else { + return Err(format!("put: unknown option {tok}")); + } + } + header.created_at = Some( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0), + ); + header.refresh_flags(); + let doc_type_dbg = header.doc_type.clone(); + if let Some(txn) = self.txn.as_mut() { + txn.ops.push(TxnOp { + coord, + put: Some((val, header)), + }); + Ok(format!( + "buffered put {} (doc_type={:?}) — commit to apply", + coord.pack_u32(), + doc_type_dbg + )) + } else { + store.put_record(coord, &header, &val); + Ok(format!( + "ok: wrote {} (doc_type={:?})", + coord.pack_u32(), + doc_type_dbg + )) + } + } "rawkeys" => { let ks: Vec = store .keys()