diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 526bffd..b1b229e 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -32,6 +32,8 @@ pub mod index; pub mod mind; pub mod persona; mod rows; +#[cfg(any(test, feature = "test-support"))] +pub mod oracle; pub mod query; pub mod receipt; pub mod refusal; diff --git a/crates/huginn-mind/src/oracle.rs b/crates/huginn-mind/src/oracle.rs new file mode 100644 index 0000000..33d89d1 --- /dev/null +++ b/crates/huginn-mind/src/oracle.rs @@ -0,0 +1,337 @@ +//! The read side's definition, frozen: decode every stored document and every +//! receipt on each call, derive status by a linear recursive scan, build every +//! row, and let the substrate select over all of them. It shares nothing with +//! the held `Image` or with `Docs`, so a test that sets the two side by side +//! compares two derivations, not one with itself. No production consumer. +//! +//! Also the synthetic history the floor test scales a store with. + +use std::collections::BTreeMap; +use std::path::Path; + +use chrono::{TimeZone, Utc}; +use cultcache_rs::{CultCacheEnvelope, DatabaseEntry}; +use cultnet_rs::{Selection, select, validate}; +use eureka_pipeline::{ + CodeLocation, CommitRange, Date, Evidence, EvidenceKind, FindingConfidence, FindingOrigin, FindingSeverity, Label, MutationRecord, + PipelineCutReport, PipelineDocument, PipelineFinding, PipelineKind, PipelineRef, PipelineResolution, ReportCommit, ResolutionOutcome, Sha, + Line, Para, Short, Slug, StructuralDelta, pipeline_key, +}; + +use crate::docs::{Held, citations, kind_of_type}; +use crate::mind::{Mind, schema_cache}; +use crate::persona::HuginnPersonaEntry; +use crate::query::{AdmissionFacts, PipelineDocumentView, PipelineSelectionPage, PipelineStatus}; +use crate::receipt::{self, Faculty, HuginnCommitReceipt, PipelineProvenance}; +use crate::refusal::MindRefusal; +use crate::rows::{SelectionRow, Vocabulary, refuse_values}; +use crate::store::MindStore; + +struct Doc { + kind: PipelineKind, + key: String, + document: PipelineDocument, + facts: AdmissionFacts, +} + +/// The mind as the store holds it, derived from nothing but the store. +struct Fresh { + docs: Vec, +} + +fn fault(detail: String) -> MindRefusal { + MindRefusal::Unavailable { detail } +} + +impl Fresh { + fn pull(mind: &Mind) -> Result { + let mut raw = mind.store().pull_all().map_err(|error| fault(format!("{error:#}")))?; + raw.sort_by(|left, right| (&left.r#type, &left.key).cmp(&(&right.r#type, &right.key))); + let mut held = Vec::new(); + for envelope in &raw { + let Some(kind) = kind_of_type(&envelope.r#type) else { continue }; + let document = PipelineDocument::decode(envelope) + .map_err(|refusal| fault(format!("stored document {} does not decode: {refusal}", envelope.key)))?; + held.push((kind, envelope.key.clone(), document)); + } + let mut receipts = Vec::new(); + for envelope in raw.iter().filter(|envelope| envelope.r#type == HuginnCommitReceipt::TYPE) { + receipts.push(rmp_serde::from_slice::(&envelope.payload).map_err(|error| { + fault(format!("receipt {} does not decode: {error}", envelope.key)) + })?); + } + let mut ordinals: Vec = receipts.iter().map(|receipt| receipt.ordinal).collect(); + ordinals.sort_unstable(); + if !ordinals.iter().copied().eq(1..=ordinals.len() as u64) { + return Err(fault("receipt ordinals are not 1..=N: found a duplicate or a gap".into())); + } + let mut facts: BTreeMap<(String, String), AdmissionFacts> = BTreeMap::new(); + for receipt in &receipts { + for write in receipt.writes.iter().filter(|write| write.document_type != HuginnPersonaEntry::TYPE) { + let landed = AdmissionFacts { + receipt_id: receipt.receipt_id.clone(), + admitted_at: receipt.committed_at.clone(), + provenance: receipt.provenance.clone(), + ordinal: receipt.ordinal, + }; + if facts.insert((write.document_type.clone(), write.document_key.clone()), landed).is_some() { + return Err(fault(format!("document {}/{} is written by two receipts", write.document_type, write.document_key))); + } + } + } + let mut docs = Vec::new(); + for (kind, key, document) in held { + let type_id = kind.type_id(); + let landed = facts + .get(&(type_id.to_string(), key.clone())) + .cloned() + .ok_or_else(|| fault(format!("document {type_id}/{key} has no commit receipt")))?; + docs.push(Doc { kind, key, document, facts: landed }); + } + Ok(Self { docs }) + } + + fn visible(&self, as_of: u64) -> Vec<&Doc> { + self.docs.iter().filter(|doc| doc.facts.ordinal <= as_of).collect() + } +} + +/// The recursive derivation, by linear scan over the visible documents. +fn closing<'a>(visible: &[&'a Doc], kind: PipelineKind, key: &str) -> Option<(&'a str, &'a PipelineResolution)> { + visible.iter().find_map(|doc| match &doc.document { + PipelineDocument::Resolution(resolution) + if resolution.subject.kind == kind + && resolution.subject.id.0 == key + && closing(visible, PipelineKind::Resolution, &doc.key).is_none() => + { + Some((doc.key.as_str(), resolution)) + } + _ => None, + }) +} + +fn view_of(visible: &[&Doc], doc: &Doc) -> PipelineDocumentView { + let status = match closing(visible, doc.kind, &doc.key) { + Some((key, record)) => PipelineStatus::Resolved { + resolution: PipelineRef { kind: PipelineKind::Resolution, id: Short(key.to_string()) }, + record: record.clone(), + }, + None => PipelineStatus::InForce, + }; + PipelineDocumentView { + id: PipelineRef { kind: doc.kind, id: Short(doc.key.clone()) }, + document: doc.document.clone(), + admission: doc.facts.clone(), + status, + } +} + +fn base<'a>(visible: &[&'a Doc], doc: &'a Doc) -> &'a Doc { + let mut current = doc; + while let PipelineDocument::Resolution(resolution) = ¤t.document { + let Some(subject) = visible.iter().find(|other| other.kind == resolution.subject.kind && other.key == resolution.subject.id.0) else { + return current; + }; + current = subject; + } + current +} + +fn row_of(visible: &[&Doc], doc: &Doc) -> SelectionRow { + let base = base(visible, doc); + let held = Held { kind: doc.kind, key: doc.key.clone(), document: doc.document.clone() }; + let base = Held { kind: base.kind, key: base.key.clone(), document: base.document.clone() }; + SelectionRow::new(&held, &base, view_of(visible, doc), citations(&doc.document)) +} + +/// The page a selection earns at `as_of`, derived from the store alone. +pub fn oracle_page(mind: &Mind, selection: &Selection, as_of: u64) -> Result { + validate(selection, &Vocabulary).map_err(cultnet_rs::SelectionRefusal::Invalid)?; + let selection = refuse_values(selection)?; + let fresh = Fresh::pull(mind)?; + let visible = fresh.visible(as_of); + let rows: Vec = visible.iter().map(|doc| row_of(&visible, doc)).collect(); + let evaluation = select(&Vocabulary, &rows, &selection, as_of, mind.cursor_key())?; + Mind::::page(&selection, evaluation, as_of, None) +} + +/// The view of one document at `as_of`, derived from the store alone. +pub fn oracle_view( + mind: &Mind, + id: &PipelineRef, + as_of: u64, +) -> Result, MindRefusal> { + let fresh = Fresh::pull(mind)?; + let visible = fresh.visible(as_of); + Ok(visible.iter().find(|doc| doc.kind == id.kind && doc.key == id.id.0).map(|doc| view_of(&visible, doc))) +} + +/// Every pipeline identity the store holds, in the snapshot's order. +pub fn oracle_ids(mind: &Mind, as_of: u64) -> Result, MindRefusal> { + let fresh = Fresh::pull(mind)?; + Ok(fresh.visible(as_of).iter().map(|doc| PipelineRef { kind: doc.kind, id: Short(doc.key.clone()) }).collect()) +} + +// --------------------------------------------------------------------------------------------- +// The synthetic history +// --------------------------------------------------------------------------------------------- + +const BATCH_RECEIPTS: usize = 700; + +fn pad(text: &str, bytes: usize) -> String { + let mut out = text.to_string(); + while out.len() < bytes { + out.push_str(" padding for payload size"); + } + out.truncate(bytes.max(text.len())); + out +} + +fn range() -> CommitRange { + CommitRange { base: Sha("5f98228d".into()), head: Sha("a1b2c3d4".into()) } +} + +fn finding(root: &str, unit: usize, n: usize) -> PipelineDocument { + PipelineDocument::Finding(PipelineFinding { + campaign: Slug(root.into()), + verdict: Short(format!("{root}:verdict:cut-u{unit}.s1")), + label: Label(format!("f{unit}-{n}")), + range: range(), + confidence: FindingConfidence::Confirmed, + severity: FindingSeverity::Medium, + claim: Line(pad("A synthetic claim.", 500)), + invariants: vec![Label("synthetic".into())], + locations: vec![CodeLocation { path: Short("crates/huginn-mind/src/query.rs".into()), line: 1, end_line: None, symbol: None }], + failure_scenario: Para(pad("A synthetic failure scenario.", 700)), + evidence: vec![Evidence { kind: EvidenceKind::Test, locator: "cargo test".into(), result: "ok".into() }], + precedents: vec![], + origin: FindingOrigin::Introduced, + }) +} + +fn cut_report(root: &str, unit: usize) -> PipelineDocument { + let delta = StructuralDelta { + lines_added: 0, + lines_removed: 0, + dependencies_added: vec![], + dependencies_removed: vec![], + formats_added: vec![], + formats_removed: vec![], + targets_added: vec![], + targets_removed: vec![], + }; + PipelineDocument::CutReport(PipelineCutReport { + campaign: Slug(root.into()), + cut_spec: Short(format!("{root}:cut_spec:cut-u{unit}.r1")), + attempt: 1, + repo: eureka_pipeline::OrgRepo("GameCult/Huginn".into()), + branch: Short("hands/synthetic".into()), + commits: vec![ReportCommit { sha: Sha("5f98228d".into()), subject: "Land the cut".into(), builds: true }], + range: range(), + verification: (0..8) + .map(|n| Evidence { kind: EvidenceKind::Command, locator: Line(format!("command {n}")), result: Line(pad("passed", 1000)) }) + .collect(), + mutations: vec![MutationRecord { + label: Label("M1".into()), + rule: "the rule".into(), + location: CodeLocation { path: Short("crates/huginn-mind/src/query.rs".into()), line: 1, end_line: None, symbol: None }, + before: "before".into(), + after: "after".into(), + commit: Sha("5f98228d".into()), + failed_as_expected: true, + }], + deviations: vec![], + forks: vec![], + structural_delta: delta, + landed_names: vec![], + undone: vec![], + promises: vec![], + }) +} + +fn resolution(subject: PipelineRef) -> PipelineDocument { + PipelineDocument::Resolution(PipelineResolution { + subject, + sequence: 1, + outcome: ResolutionOutcome::Recorded { reason: "Synthetic.".into() }, + rationale: Para(pad("Resolved.", 100)), + resolved_on: Date("2026-10-01".into()), + }) +} + +/// Grows a store to `factor` times its pipeline documents with synthetic +/// history: findings, cut reports and the resolutions that close four records +/// in five, about three writes a receipt, through the store's own +/// compare-and-swap in dense receipts. `spread` puts each hundred units under +/// its own `synth-` root; otherwise everything goes under +/// `eureka-substrate`, one root growing. Returns the documents written. +pub fn synth_history(store: &S, instance: &Slug, factor: u32, spread: bool) -> Result { + let pulled = store.pull_all().map_err(|error| fault(format!("{error:#}")))?; + let existing = pulled.iter().filter(|envelope| kind_of_type(&envelope.r#type).is_some()).count(); + let mut ordinal = pulled.iter().filter(|envelope| envelope.r#type == HuginnCommitReceipt::TYPE).count() as u64; + let want = existing * factor.saturating_sub(1) as usize; + // A unit is five records (four findings, one cut report) and the four + // resolutions that close four of them. + let units = want / 9; + let cache = schema_cache()?; + let provenance = PipelineProvenance { + faculty: Faculty::Hands, + agent: Short("synth".into()), + session: Short("synth".into()), + tool: Short("synth".into()), + }; + let now = Utc.with_ymd_and_hms(2026, 10, 1, 12, 0, 0).unwrap(); + let mut pending: Vec> = Vec::new(); + let mut written = 0; + let mut flush = |groups: &mut Vec>, ordinal: &mut u64| -> Result<(), MindRefusal> { + if groups.is_empty() { + return Ok(()); + } + let mut replacements = Vec::new(); + for writes in groups.drain(..) { + *ordinal += 1; + let receipt = receipt::candidate(instance, provenance.clone(), &[], &writes, *ordinal, now)?; + let (envelope, _) = cache.prepare_entry_named(&receipt.receipt_id, &receipt).map_err(|error| fault(format!("{error:#}")))?; + replacements.extend(writes); + replacements.push(envelope); + } + if !store.compare_and_swap_batch(&[], replacements).map_err(|error| fault(format!("{error:#}")))? { + return Err(fault("the synthetic batch lost its swap".into())); + } + Ok(()) + }; + let prepare = |document: &PipelineDocument| -> Result<(PipelineRef, CultCacheEnvelope), MindRefusal> { + let key = pipeline_key(document)?; + let envelope = document.prepare(&cache).map_err(|error| fault(format!("{error:#}")))?; + Ok((PipelineRef { kind: document.kind(), id: Short(key) }, envelope)) + }; + for unit in 0..units { + let root = if spread { format!("synth-{}", unit / 100) } else { "eureka-substrate".to_string() }; + let mut records = Vec::new(); + for n in 0..4 { + records.push(prepare(&finding(&root, unit, n))?); + } + records.push(prepare(&cut_report(&root, unit))?); + let mut closers = Vec::new(); + for (subject, _) in records.iter().take(4) { + closers.push(prepare(&resolution(subject.clone()))?); + } + let mut all = records.into_iter().chain(closers).map(|(_, envelope)| envelope).collect::>(); + written += all.len(); + let tail = all.split_off(6); + let middle = all.split_off(3); + pending.extend([all, middle, tail]); + if pending.len() >= BATCH_RECEIPTS { + flush(&mut pending, &mut ordinal)?; + } + } + flush(&mut pending, &mut ordinal)?; + Ok(written) +} + +/// Opens the store at `path` under its exclusive lock, grows it, and lets go +/// of the lock, so a `Mind::open` can follow. +pub fn synth_history_at(path: &Path, instance: &Slug, factor: u32, spread: bool) -> Result { + let store = crate::store::OwnedRedbMessagePackBackingStore::new(path).map_err(|error| fault(format!("{error:#}")))?; + synth_history(&store, instance, factor, spread) +} diff --git a/crates/huginn-mind/src/query.rs b/crates/huginn-mind/src/query.rs index 1897849..54d6704 100644 --- a/crates/huginn-mind/src/query.rs +++ b/crates/huginn-mind/src/query.rs @@ -771,7 +771,7 @@ impl Mind { /// The projection of an evaluation: a header or a whole view per row, and /// the typed edges a hop traversed. - fn page( + pub(crate) fn page( selection: &Selection, evaluation: Evaluation, as_of: u64, diff --git a/crates/huginn-mind/tests/read_floor.rs b/crates/huginn-mind/tests/read_floor.rs new file mode 100644 index 0000000..f324c68 --- /dev/null +++ b/crates/huginn-mind/tests/read_floor.rs @@ -0,0 +1,171 @@ +//! What a read costs, measured through the public API alone so the same file +//! runs at the base and at the branch. Ignored: it wants a copy of a real +//! store and says what it measured instead of asserting. +//! +//! `FLOOR_STATE` is a directory holding `minds//mind.redb`. It is +//! copied to a temp dir and never opened in place. `FLOOR_SCALE` (default 1) +//! grows the copy with synthetic history to that multiple of its documents; +//! `FLOOR_ROOTS` is `spread` (each hundred units under its own root) or +//! `same` (one root grows). `FLOOR_INSTANCE` defaults to `eureka`. +//! +//! cargo test --release -p huginn-mind --features test-support --test read_floor -- --ignored --nocapture +#![cfg(feature = "test-support")] + +use std::path::PathBuf; +use std::time::{Duration, Instant}; + +use chrono::Utc; +use cultnet_rs::{Citation, FieldPredicate, Incoming, RecordRef, Selection}; +use huginn_mind::eureka_pipeline::{PipelineDocument, PipelineKind, PipelinePulse, PipelineRef, Short, Slug}; +use huginn_mind::oracle::synth_history_at; +use huginn_mind::{Faculty, IndexStatus, Mind, PipelineAdmissionBatch, PipelinePageItems, PipelineProvenance}; + +fn env(name: &str, default: &str) -> String { + std::env::var(name).unwrap_or_else(|_| default.to_string()) +} + +fn rss_kib() -> u64 { + std::fs::read_to_string("/proc/self/status") + .ok() + .and_then(|status| { + status.lines().find_map(|line| line.strip_prefix("VmRSS:")?.split_whitespace().next()?.parse().ok()) + }) + .unwrap_or(0) +} + +fn any_of(index: &str, values: &[&str]) -> FieldPredicate { + FieldPredicate { index: index.into(), op: "any_of".into(), values: Some(values.iter().map(|v| v.to_string()).collect()), number: None } +} + +fn of(kind: PipelineKind, fields: Vec) -> Selection { + Selection { schemas: Some(vec![kind.type_id().to_string()]), fields: Some(fields), ..Selection::default() } +} + +fn median(mut samples: Vec) -> Duration { + samples.sort(); + samples[samples.len() / 2] +} + +fn time(mut read: impl FnMut() -> T) -> Duration { + median( + (0..7) + .map(|_| { + let started = Instant::now(); + std::hint::black_box(read()); + started.elapsed() + }) + .collect(), + ) +} + +fn keys_of(mind: &Mind, kind: PipelineKind) -> Vec { + let page = mind.query(&Selection { schemas: Some(vec![kind.type_id().to_string()]), limit: Some(5), ..Selection::default() }).unwrap(); + match page.items { + PipelinePageItems::Headers(headers) => headers.into_iter().map(|header| header.id).collect(), + PipelinePageItems::Documents(views) => views.into_iter().map(|view| view.id).collect(), + } +} + +#[test] +#[ignore] +fn read_floor() { + let instance = Slug(env("FLOOR_INSTANCE", "eureka")); + let scale: u32 = env("FLOOR_SCALE", "1").parse().unwrap(); + let spread = env("FLOOR_ROOTS", "spread") == "spread"; + let source = PathBuf::from(env("FLOOR_STATE", "/state")).join("minds").join(&instance.0).join("mind.redb"); + let temp = tempfile::tempdir().unwrap(); + let target = temp.path().join("minds").join(&instance.0); + std::fs::create_dir_all(&target).unwrap(); + std::fs::copy(&source, target.join("mind.redb")).unwrap(); + let path = target.join("mind.redb"); + + let synthetic = if scale > 1 { synth_history_at(&path, &instance, scale, spread).unwrap() } else { 0 }; + let size_mib = std::fs::metadata(&path).unwrap().len() / (1024 * 1024); + + let started = Instant::now(); + let mut mind = Mind::open(temp.path(), &instance).unwrap(); + let open = started.elapsed(); + let rss = rss_kib() / 1024; + let status = mind.status(IndexStatus::Current); + + let findings = keys_of(&mind, PipelineKind::Finding); + let finding = findings.first().expect("the store holds a finding").clone(); + let root = finding.id.0.split(':').next().unwrap().to_string(); + let mut one = Selection { keys: Some(vec![finding.id.0.clone()]), projection: "document".into(), ..Selection::default() }; + one.schemas = Some(vec![PipelineKind::Finding.type_id().to_string()]); + + let run_desc = Selection { descending: true, limit: Some(3), ..of(PipelineKind::Run, vec![any_of("root", &[&root]), any_of("operated_by", &["Mind"])]) }; + let cites = Selection { + cites: Some(Citation { target: RecordRef::new(PipelineKind::Finding.type_id(), finding.id.0.clone()), role: Some("subject".into()) }), + ..of(PipelineKind::Resolution, vec![]) + }; + let stewardship = of(PipelineKind::Stewardship, vec![any_of("root", &[&root])]); + let memory = of(PipelineKind::Memory, vec![any_of("root", &[&root]), any_of("memory_origin", &["Room"])]); + let follow_up = Selection { limit: Some(20), ..of(PipelineKind::FollowUp, vec![any_of("in_force", &["true"])]) }; + let three = of( + PipelineKind::Finding, + vec![any_of("severity", &["High"]), any_of("confidence", &["Confirmed"]), any_of("origin", &["Introduced"])], + ); + let unreported = Selection { + cited: Some(Incoming { role: "cut_spec".into(), exists: false }), + ..of(PipelineKind::CutSpec, vec![any_of("in_force", &["true"])]) + }; + + let paged = |selection: &Selection| { + let mut selection = selection.clone(); + let mut pages = 0; + loop { + let page = mind.query(&selection).unwrap(); + pages += 1; + match page.next { + Some(next) if pages < 3 => selection.cursor = Some(next), + _ => return pages, + } + } + }; + + let mut rows: Vec<(&str, Duration)> = vec![ + ("whoami/status", time(|| mind.status(IndexStatus::Current))), + ("persona", time(|| mind.persona().unwrap())), + ("view", time(|| mind.view(&finding).unwrap())), + ("keys=1 document", time(|| mind.query(&one).unwrap())), + ("run root+operated_by desc", time(|| mind.query(&run_desc).unwrap())), + ("resolution cites subject", time(|| mind.query(&cites).unwrap())), + ("stewardship root", time(|| mind.query(&stewardship).unwrap())), + ("memory two fields", time(|| mind.query(&memory).unwrap())), + ("follow_up in_force paged", time(|| paged(&follow_up))), + ("finding three fields", time(|| mind.query(&three).unwrap())), + ("cut_spec in_force cited", time(|| mind.query(&unreported).unwrap())), + ]; + let line = [&run_desc, &cites, &stewardship, &memory, &follow_up, &three, &unreported, &one]; + let first_line = time(|| { + for n in 0..25 { + let _ = mind.query(line[n % line.len()]); + } + }); + rows.push(("first line (25 reads)", first_line)); + + let pulse = PipelineDocument::Pulse(PipelinePulse { + instance: instance.clone(), + label: format!("floor-{}", Utc::now().timestamp_nanos_opt().unwrap_or_default()).as_str().into(), + note: "read floor".into(), + tempo_vote_beat_s: None, + tempo_vote_withdrawn: false, + urgency_permille: Some(1), + }); + let provenance = PipelineProvenance { faculty: Faculty::Hands, agent: Short("floor".into()), session: Short("floor".into()), tool: Short("floor".into()) }; + let started = Instant::now(); + let outcome = mind.admit(PipelineAdmissionBatch { instance: instance.clone(), provenance, documents: vec![pulse] }, Utc::now()); + let admit = started.elapsed(); + let started = Instant::now(); + let read_back = mind.read_back(); + let read_back_time = started.elapsed(); + + println!("\nFLOOR documents={} receipts={} store_mib={} synthetic_docs={} scale={} roots={}", status.documents, status.receipts, size_mib, synthetic, scale, if spread { "spread" } else { "same" }); + println!("FLOOR open_ms={:.0} rss_mib={}", open.as_secs_f64() * 1e3, rss); + for (name, took) in rows { + println!("FLOOR read {:<28} {:>10.1} us", name, took.as_secs_f64() * 1e6); + } + println!("FLOOR admit_ms={:.0} ({})", admit.as_secs_f64() * 1e3, if matches!(outcome, huginn_mind::PipelineAdmissionOutcome::Committed { .. }) { "committed".to_string() } else { format!("{outcome:?}") }); + println!("FLOOR read_back_ms={:.0} ({:?})", read_back_time.as_secs_f64() * 1e3, read_back.map(|count| count)); +}