diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index 8ef3482..e5e9a9f 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -792,6 +792,173 @@ mod tests { assert_eq!(documents(&mut daemon), 1, "the admission that crossed the wire landed"); } + fn question_doc(n: usize) -> huginn_mind::eureka_pipeline::PipelineDocument { + use huginn_mind::eureka_pipeline::{Date, PipelineDocument, PipelineQuestion, QuestionOption, Title}; + PipelineDocument::Question(PipelineQuestion { + campaign: slug(crate::daemon::tests::CAMPAIGN), + label: format!("Q{n}").as_str().into(), + title: Title("Who owns the state?".into()), + question: "Who owns the state?".into(), + options: vec![ + QuestionOption { label: "A".into(), text: "the organ".into() }, + QuestionOption { label: "B".into(), text: "the client".into() }, + ], + recommended: "A".into(), + depends: vec![], + raised_in: None, + asked_on: Date("2026-09-16".into()), + }) + } + + fn resolution_of(subject: PipelineRef, outcome: huginn_mind::eureka_pipeline::ResolutionOutcome) -> huginn_mind::eureka_pipeline::PipelineDocument { + huginn_mind::eureka_pipeline::PipelineDocument::Resolution(huginn_mind::eureka_pipeline::PipelineResolution { + subject, + sequence: 1, + outcome, + rationale: "Resolved.".into(), + resolved_on: huginn_mind::eureka_pipeline::Date("2026-09-16".into()), + }) + } + + /// One client session asks `request` and decodes the answer. + fn ask(client: &mut Client, message_id: &str, request: &HuginnMindRequest) -> HuginnMindResponse { + let reply = exchange(client, &encode_request(message_id, request, None).unwrap()).unwrap(); + decode_response(&reply).unwrap().1.unwrap_or_else(|failure| panic!("{message_id}: {failure:?}")) + } + + fn connect(endpoint: &str, name: &str) -> Client { + let mut client = CultMesh::create_rudp_client_for_endpoint( + format!("eureka-state-{name}"), + CULTNET_OPERATION_CONNECTION_ID, + endpoint, + CultMeshRudpSocketOptions::default(), + ) + .unwrap(); + client.connect(Vec::new()).unwrap(); + for _ in 0..1000 { + let _ = client.receive_once(); + if client.connected() { + return client; + } + std::thread::sleep(Duration::from_millis(2)); + } + panic!("{name} never connected"); + } + + /// One client admits batches that close, withdraw and re-open records while three others read + /// the same mind: fresh queries, cursor walks across commits, a hop and whoami. The loop is + /// serial, so every answer is one snapshot's; at quiesce each answer must equal what a fresh + /// derivation from the store gives at the snapshot it names, and no client sees the head go + /// backwards. The test pins the behaviour, not the mechanism. + #[test] + fn reads_racing_admissions_on_the_wire_answer_their_snapshot() { + use huginn_mind::eureka_pipeline::{PipelineKind as K, ResolutionOutcome}; + let root = tempfile::tempdir().unwrap(); + let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap(); + let seeded = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, crate::daemon::tests::campaign_seed())), now()).answered(); + assert!(matches!(seeded, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{seeded:?}"); + let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap(); + let endpoint = format!("rudp://{}", hub.local_addr().unwrap()); + let registry = schema_registry().unwrap(); + let stopping = Arc::new(AtomicBool::new(false)); + let loop_stopping = Arc::clone(&stopping); + let serving = std::thread::spawn(move || { + let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()); + (result, daemon) + }); + + let done = Arc::new(AtomicBool::new(false)); + let writer = { + let (endpoint, done) = (endpoint.clone(), Arc::clone(&done)); + std::thread::spawn(move || { + let mut client = connect(&endpoint, "writer"); + let admit = |client: &mut Client, n: usize, documents| { + let answer = ask(client, &format!("w-{n}"), &HuginnMindRequest::Admit(batch(INSTANCE, documents))); + assert!(matches!(answer, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{answer:?}"); + }; + let withdrawn = || ResolutionOutcome::Withdrawn { reason: "race".into() }; + let mut n = 0; + for cut in 0..14 { + let spec = PipelineRef { kind: K::Question, id: format!("{}:question:Q{cut}", crate::daemon::tests::CAMPAIGN).as_str().into() }; + admit(&mut client, n, vec![question_doc(cut)]); + admit(&mut client, n + 1, vec![resolution_of(spec, withdrawn())]); + if cut % 2 == 0 { + let closing = PipelineRef { kind: K::Resolution, id: format!("{}:resolution:question.Q{cut}.n1", crate::daemon::tests::CAMPAIGN).as_str().into() }; + admit(&mut client, n + 2, vec![resolution_of(closing, withdrawn())]); + } + n += 3; + } + done.store(true, Ordering::Relaxed); + }) + }; + let readers: Vec<_> = (0..3) + .map(|index| { + let (endpoint, done) = (endpoint.clone(), Arc::clone(&done)); + std::thread::spawn(move || { + let mut client = connect(&endpoint, &format!("reader-{index}")); + let of_specs = |extra: Selection| Selection { schemas: Some(vec![K::Question.type_id().into()]), ..extra }; + let in_force = FieldPredicate { index: "in_force".into(), op: "any_of".into(), values: Some(vec!["true".into()]), number: None }; + let mut recorded: Vec<(Selection, huginn_mind::PipelineSelectionPage)> = Vec::new(); + let mut status_walk = of_specs(Selection { fields: Some(vec![in_force.clone()]), limit: Some(1), projection: "document".into(), ..Selection::default() }); + let mut head = 0; + let mut round = 0; + while !done.load(Ordering::Relaxed) || round < 3 { + round += 1; + let query = |client: &mut Client, selection: &Selection| { + match ask(client, &format!("r-{index}-{round}"), &HuginnMindRequest::Query { instance: slug(INSTANCE), selection: selection.clone(), semantic: None }) { + HuginnMindResponse::Query(page) => page, + other => panic!("expected a page, got {other:?}"), + } + }; + let fresh = of_specs(Selection { fields: Some(vec![in_force.clone()]), limit: Some(5), ..Selection::default() }); + let page = query(&mut client, &fresh); + assert!(page.as_of >= head, "a client's head never goes backwards"); + head = page.as_of; + recorded.push((fresh, page)); + let mut walking = of_specs(Selection { limit: Some(1), ..Selection::default() }); + for _ in 0..3 { + let page = query(&mut client, &walking); + let next = page.next.clone(); + recorded.push((walking.clone(), page)); + match next { + Some(cursor) => walking.cursor = Some(cursor), + None => break, + } + } + // A status walk: its cursor pins the snapshot it began at while the writer + // closes and reopens the questions, so a page is read at an old snapshot + // through the status filter and the document, which only the row's + // status_changed can answer. + let page = query(&mut client, &status_walk); + let next = page.next.clone(); + recorded.push((status_walk.clone(), page)); + status_walk.cursor = next; + let hop = Selection { cited: Some(cultnet_rs::Incoming { role: "subject".into(), exists: true }), ..Selection::default() }; + let page = query(&mut client, &hop); + recorded.push((hop, page)); + assert!(matches!(ask(&mut client, "whoami", &HuginnMindRequest::Whoami), HuginnMindResponse::Whoami(_))); + } + recorded + }) + }) + .collect(); + + writer.join().unwrap(); + let recorded: Vec<_> = readers.into_iter().flat_map(|reader| reader.join().unwrap()).collect(); + stopping.store(true, Ordering::Relaxed); + let (result, daemon) = serving.join().unwrap(); + result.unwrap(); + let mind = daemon.mind(); + let final_head = mind.query(&Selection::default()).unwrap().as_of; + let mut snapshots = std::collections::BTreeSet::new(); + for (selection, page) in &recorded { + snapshots.insert(page.as_of); + assert_eq!(huginn_mind::oracle::oracle_page(mind, selection, page.as_of).as_ref(), Ok(page), "{selection:?} at {}", page.as_of); + } + assert!(snapshots.len() > 1, "the readers saw more than one snapshot: {snapshots:?}"); + assert!(snapshots.iter().all(|as_of| *as_of <= final_head)); + } + fn encoded_len(message: &CultNetMessage) -> u64 { encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap().len() as u64 } diff --git a/crates/huginn-mind/src/admission.rs b/crates/huginn-mind/src/admission.rs index 1aa2ad8..59fe6c7 100644 --- a/crates/huginn-mind/src/admission.rs +++ b/crates/huginn-mind/src/admission.rs @@ -124,7 +124,8 @@ impl Mind { return Err(MindRefusal::BatchSize { actual: envelopes.len() as u32 }); } let mind = self.instance().clone(); - let mut docs = Docs::from_image(self.envelopes())?; + let image = self.image()?; + let mut docs = Docs::at(image, image.head); // A3, A5, then A4 on push. for envelope in envelopes { docs.push(stage(envelope, &mind)?)?; @@ -366,7 +367,7 @@ fn references(staged: &Staged, mind: &Slug) -> Result, MindRefusa /// A7: every referent is in the batch or the image; the image ones are the /// receipt's strong reads. -fn resolve(docs: &Docs, staged: &Staged, mind: &Slug, strong: &mut BTreeSet<(String, String)>) -> Result<(), MindRefusal> { +fn resolve(docs: &Docs<'_>, staged: &Staged, mind: &Slug, strong: &mut BTreeSet<(String, String)>) -> Result<(), MindRefusal> { for reference in references(staged, mind)? { if docs.in_batch(reference.kind, &reference.id).is_some() { continue; @@ -389,7 +390,7 @@ fn resolve(docs: &Docs, staged: &Staged, mind: &Slug, strong: &mut BTreeSet<(Str /// the stewardship withdrawal on its source side and the stewardship /// assignment on its receiving side. One the batch already carries is not /// derived again. -fn derive(docs: &Docs, mind: &Slug) -> Vec { +fn derive(docs: &Docs<'_>, mind: &Slug) -> Vec { use PipelineDocument as D; use PipelineKind as K; let mut derived = Vec::new(); @@ -444,7 +445,7 @@ fn derive(docs: &Docs, mind: &Slug) -> Vec { } /// A8: the per-kind rules. -fn check(docs: &Docs, staged: &Staged, mind: &Slug) -> Result<(), MindRefusal> { +fn check(docs: &Docs<'_>, staged: &Staged, mind: &Slug) -> Result<(), MindRefusal> { use PipelineDocument as D; use PipelineKind as K; let key = staged.key.as_str(); @@ -629,7 +630,7 @@ fn check(docs: &Docs, staged: &Staged, mind: &Slug) -> Result<(), MindRefusal> { /// The one rule for a run opening or reinstated by withdrawing its close: every /// claim in force (a replay of an admitted run owes none) and held by no other /// live run, then at most one live run of hers per instance and turn. -fn run_rule(docs: &Docs, run_key: &str, run: &PipelineRun, replay: bool) -> Result<(), MindRefusal> { +fn run_rule(docs: &Docs<'_>, run_key: &str, run: &PipelineRun, replay: bool) -> Result<(), MindRefusal> { for claim in &run.claims { if !replay && !docs.in_force(claim.kind, &claim.id.0) { return Err(MindRefusal::CitesResolvedDocument { kind: claim.kind, id: claim.id.0.clone() }); @@ -647,7 +648,7 @@ fn run_rule(docs: &Docs, run_key: &str, run: &PipelineRun, replay: bool) -> Resu /// Revision 1 stands alone; revision N needs a resolution in the batch that /// supersedes revision N-1 by this document. The predecessor's key is the /// leaf's, from the same document one revision back. -fn revision_rule(docs: &Docs, kind: PipelineKind, key: &str, revision: u32, predecessor: &PipelineDocument) -> Result<(), MindRefusal> { +fn revision_rule(docs: &Docs<'_>, kind: PipelineKind, key: &str, revision: u32, predecessor: &PipelineDocument) -> Result<(), MindRefusal> { if revision == 1 { return Ok(()); } @@ -731,7 +732,7 @@ fn matrix(subject: PipelineKind, outcome: &ResolutionOutcome) -> bool { /// every referent in force, and the two /// coherence rules (an `Answered` ruling answers this question; a cut spec is /// superseded within its cut). -fn resolution_rule(docs: &Docs, resolution: &PipelineResolution) -> Result<(), MindRefusal> { +fn resolution_rule(docs: &Docs<'_>, resolution: &PipelineResolution) -> Result<(), MindRefusal> { use PipelineDocument as D; use PipelineKind as K; let subject_kind = resolution.subject.kind; @@ -830,7 +831,7 @@ fn resolution_rule(docs: &Docs, resolution: &PipelineResolution) -> Result<(), M /// A10: a write whose identity the image already holds. Every kind answers /// the same way now that a resolution key carries a sequence: the identity is /// taken. Whether a subject is closed is A8's question, not this one's. -fn refuse_collisions(docs: &Docs) -> Result<(), MindRefusal> { +fn refuse_collisions(docs: &Docs<'_>) -> Result<(), MindRefusal> { for staged in &docs.batch { if docs.in_image(staged.kind, &staged.key).is_some() { return Err(MindRefusal::IdentityCollision { kind: staged.kind, id: staged.key.clone() }); @@ -1070,8 +1071,8 @@ mod tests { let key = format!("{INSTANCE}:stewardship:gamecult_-epiphany.n1"); assert!(yggdrasil.envelope(K::Stewardship, &key).is_some()); assert!(thought_cage.envelope(K::Stewardship, &key).is_none()); - assert_eq!(yggdrasil.envelopes().len(), 4); - assert_eq!(thought_cage.envelopes().len(), 3); + assert_eq!(yggdrasil.envelopes().count(), 4); + assert_eq!(thought_cage.envelopes().count(), 3); } #[test] @@ -1577,7 +1578,7 @@ mod tests { refusal(admit(&mut mind, vec![resolution(r(K::Resolution, &withdrawal), withdrawn())])), MindRefusal::AlreadyStewarded { repo: REPO.into() } ); - let docs = Docs::from_image(mind.envelopes()).unwrap(); + let docs = docs_of(&mind); let in_force = docs.stewardships_of(&slug(INSTANCE), &repo(REPO), None).iter().map(|(key, _)| *key).collect::>(); assert_eq!(in_force, vec![again.as_str()], "one assignment stands, and it is the later one"); } @@ -1599,8 +1600,9 @@ mod tests { assert!(!docs_of(&mind).in_force(K::Target, &id("target", "r1"))); } - fn docs_of(mind: &Mind) -> Docs { - Docs::from_image(mind.envelopes()).unwrap() + fn docs_of(mind: &Mind) -> Docs<'_> { + let image = mind.image().unwrap(); + Docs::at(image, image.head) } /// The withdrawal a hand-off derives is a record of its subject like any diff --git a/crates/huginn-mind/src/docs.rs b/crates/huginn-mind/src/docs.rs index d6bff49..217a04e 100644 --- a/crates/huginn-mind/src/docs.rs +++ b/crates/huginn-mind/src/docs.rs @@ -13,6 +13,7 @@ use eureka_pipeline::{ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use crate::image::Image; use crate::refusal::MindRefusal; /// A batch document after A3 and A5: decoded, keyed, and its envelope. @@ -23,16 +24,12 @@ pub(crate) struct Staged { pub(crate) envelope: CultCacheEnvelope, } -/// An image document, decoded. -pub(crate) struct Held { - pub(crate) kind: PipelineKind, - pub(crate) key: String, - pub(crate) document: PipelineDocument, -} - -/// The image and the batch, the two places a rule may look. -pub(crate) struct Docs { - pub(crate) image: Vec, +/// The held image at one snapshot and the batch, the two places a rule or a view +/// may look. The one door of visibility: a document is there when its writing +/// receipt's ordinal is at or before `as_of`. Admission runs at the head. +pub(crate) struct Docs<'i> { + image: &'i Image, + as_of: u64, pub(crate) batch: Vec, } @@ -40,17 +37,13 @@ pub(crate) fn kind_of_type(type_id: &str) -> Option { PipelineKind::ALL.iter().copied().find(|kind| kind.type_id() == type_id) } -impl Docs { - pub(crate) fn from_image(envelopes: &[CultCacheEnvelope]) -> Result { - let mut image = Vec::new(); - for envelope in envelopes { - let Some(kind) = kind_of_type(&envelope.r#type) else { continue }; - let document = PipelineDocument::decode(envelope).map_err(|refusal| MindRefusal::Unavailable { - detail: format!("stored document {} does not decode: {refusal}", envelope.key), - })?; - image.push(Held { kind, key: envelope.key.clone(), document }); - } - Ok(Self { image, batch: Vec::new() }) +impl<'i> Docs<'i> { + pub(crate) fn at(image: &'i Image, as_of: u64) -> Self { + Self { image, as_of, batch: Vec::new() } + } + + fn visible(&self, id: u32) -> bool { + self.image.row(id).ordinal <= self.as_of } /// A4: one identity, once. @@ -67,7 +60,7 @@ impl Docs { } pub(crate) fn in_image(&self, kind: PipelineKind, id: &str) -> Option<&PipelineDocument> { - self.image.iter().find(|held| held.kind == kind && held.key == id).map(|held| &held.document) + self.image.id_of(kind, id).filter(|row| self.visible(*row)).map(|row| &self.image.row(row).view.document) } pub(crate) fn find(&self, kind: PipelineKind, id: &str) -> Option<&PipelineDocument> { @@ -76,17 +69,14 @@ impl Docs { pub(crate) fn of_kind(&self, kind: PipelineKind) -> impl Iterator { let batch = self.batch.iter().filter(move |staged| staged.kind == kind).map(|staged| (staged.key.as_str(), &staged.document)); - let image = self.image.iter().filter(move |held| held.kind == kind).map(|held| (held.key.as_str(), &held.document)); + let image = self + .image + .of_kind(kind) + .filter(move |(_, row)| self.visible(*row)) + .map(move |(key, row)| (key, &self.image.row(row).view.document)); batch.chain(image) } - pub(crate) fn resolutions(&self) -> impl Iterator { - self.of_kind(PipelineKind::Resolution).filter_map(|(key, document)| match document { - PipelineDocument::Resolution(resolution) => Some((key, resolution)), - _ => None, - }) - } - pub(crate) fn stewardships(&self) -> impl Iterator { self.of_kind(PipelineKind::Stewardship).filter_map(|(key, document)| match document { PipelineDocument::Stewardship(stewardship) => Some((key, stewardship)), @@ -97,7 +87,25 @@ impl Docs { /// Every resolution of one subject, in image and batch: the scope a /// subject's sequence counts in. pub(crate) fn resolutions_of(&self, subject: &PipelineRef) -> impl Iterator { - self.resolutions().filter(move |(_, resolution)| resolution.subject == *subject) + self.resolutions_naming(subject.kind, &subject.id.0) + } + + /// The resolutions naming one document: the batch's first, then the held ones visible at + /// `as_of` in resolution-key order, the order a closing resolution is found in. + fn resolutions_naming(&self, kind: PipelineKind, id: &str) -> impl Iterator { + let batch = self.batch.iter().filter_map(move |staged| match &staged.document { + PipelineDocument::Resolution(resolution) if resolution.subject.kind == kind && resolution.subject.id.0 == id => { + Some((staged.key.as_str(), resolution)) + } + _ => None, + }); + let held = self.image.closers_of(kind, id).filter(move |(_, row)| self.visible(*row)).filter_map(move |(key, row)| { + match &self.image.row(row).view.document { + PipelineDocument::Resolution(resolution) => Some((key, resolution)), + _ => None, + } + }); + batch.chain(held) } /// Every assignment of one repo to one mind, in image and batch, in force @@ -128,8 +136,8 @@ impl Docs { subject: &PipelineRef, derived: impl Fn(&ResolutionOutcome) -> bool, ) -> u32 { - self.resolutions() - .find(|(_, resolution)| resolution.subject == *subject && derived(&resolution.outcome)) + self.resolutions_of(subject) + .find(|(_, resolution)| derived(&resolution.outcome)) .map_or_else(|| self.latest_resolution(subject, None) + 1, |(_, resolution)| resolution.sequence) } @@ -174,12 +182,7 @@ impl Docs { id: &str, own: impl Fn(&PipelineResolution) -> bool, ) -> Option<(&str, &PipelineResolution)> { - self.resolutions().find(|&(key, resolution)| { - resolution.subject.kind == kind - && resolution.subject.id.0 == id - && !own(resolution) - && self.in_force(PipelineKind::Resolution, key) - }) + self.resolutions_naming(kind, id).find(|&(key, resolution)| !own(resolution) && self.in_force(PipelineKind::Resolution, key)) } /// The repos a campaign's cuts may name: the repo set of the highest-revision @@ -292,10 +295,8 @@ impl Docs { hand_off_key .and_then(|hand_off| { candidates.iter().find(|(key, _)| { - self.resolutions().any(|(_, resolution)| { - resolution.subject.kind == PipelineKind::Stewardship - && resolution.subject.id.0 == *key - && matches!(&resolution.outcome, ResolutionOutcome::Withdrawn { reason } if reason.0 == hand_off) + self.resolutions_naming(PipelineKind::Stewardship, key).any(|(_, resolution)| { + matches!(&resolution.outcome, ResolutionOutcome::Withdrawn { reason } if reason.0 == hand_off) }) }) }) @@ -560,10 +561,6 @@ mod tests { use PipelineDocument as D; use PipelineKind as K; - fn held(kind: K, key: String, document: D) -> Held { - Held { kind, key, document } - } - fn named(campaign: &str, revision: u32, repos: &[&str]) -> (String, D) { let D::Target(mut target) = target(revision, &[INVARIANT]) else { panic!("a target") }; target.campaign = campaign.into(); @@ -588,15 +585,13 @@ mod tests { let (closed_key, closed) = named(CAMPAIGN, 2, &["GameCult/Closed"]); let (foreign_key, foreign) = named("other-campaign", 3, &["GameCult/Foreign"]); let closing = resolution(r(K::Target, &closed_key), superseded(&[r(K::Target, &in_force_key)])); - let docs = Docs { - image: vec![ - held(K::Target, in_force_key, in_force), - held(K::Target, closed_key, closed), - held(K::Target, foreign_key, foreign), - held(K::Resolution, id("resolution", "target.r2.n1"), closing), - ], - batch: Vec::new(), - }; + let image = Image::of_documents(vec![ + (K::Target, in_force_key, in_force), + (K::Target, closed_key, closed), + (K::Target, foreign_key, foreign), + (K::Resolution, id("resolution", "target.r2.n1"), closing), + ]); + let docs = Docs::at(&image, image.head); assert_eq!(docs.repos_in_force(&this), [repo(REPO), repo(OTHER_REPO)], "neither the closed higher revision nor another campaign's higher target decides"); } } diff --git a/crates/huginn-mind/src/image.rs b/crates/huginn-mind/src/image.rs new file mode 100644 index 0000000..9d4c71b --- /dev/null +++ b/crates/huginn-mind/src/image.rs @@ -0,0 +1,378 @@ +//! The mind's one decoded, indexed image: a pure fold of the commit receipts +//! in ordinal order. It is built only by replaying the receipts and advanced +//! only by the function the replay calls, so a held image and a fresh fold of +//! the same store are equal by construction. Nothing here is stored: a restart +//! re-folds the store, which is the truth. +//! +//! A row is the only decoded copy of its document. Status is derived through +//! the closing resolutions at an `as_of` and is never stored in a document; +//! a row holds the head's answer and `status_changed` says from which +//! snapshot that answer holds. The postings, closers, citers and kind maps +//! only narrow which rows a read hands the substrate: `universe` is a superset +//! of what `select` can match, and `select` still decides membership, order, +//! cursors and edges. + +use std::collections::{BTreeMap, BTreeSet, HashMap}; + +use cultcache_rs::DatabaseEntry; +use cultnet_rs::Selection; +use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, Short}; + +use crate::docs::{Docs, citations, kind_of_id, kind_of_type}; +use crate::mind::Envelopes; +use crate::persona::HuginnPersonaEntry; +use crate::query::{AdmissionFacts, AdmissionIndex, PipelineDocumentView, PipelineStatus}; +use crate::receipt::HuginnCommitReceipt; +use crate::refusal::MindRefusal; +use crate::rows::{Alias, STATUS_ALIASES, SelectionRow, root_and_local}; + +type Postings = BTreeMap>>>; + +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct Image { + /// The newest receipt's ordinal: `N` of a dense `1..=N`. + pub(crate) head: u64, + /// Append-only; a row's index is its id. Receipts' order, then identity order within one. + pub(crate) rows: Vec, + /// Per row, the newest ordinal at which its status changed. + pub(crate) status_changed: Vec, + /// Kind, then key, to row id, in key order: the image's identity lookup and its per-kind order. + by_kind: BTreeMap>, + /// (kind, alias, value) to the rows answering it. + postings: Postings, + /// Subject kind and id to the resolutions naming it, by resolution key. + closers: BTreeMap>>, + /// Target id to the (role, citer) edges that name it. + citers: HashMap>, + /// Root to the newest row written under it. + roots: BTreeMap, + /// Whether every citation edge names a target of its role's kinds. When one does not, a hop + /// selection reads every row, so the substrate refuses it exactly as it would over the whole mind. + pub(crate) edges_in_target: bool, +} + +fn integrity(type_id: &str, key: &str, fault: &str) -> MindRefusal { + MindRefusal::Unavailable { detail: format!("document {type_id}/{key} {fault}") } +} + +pub(crate) fn status_of(closing: Option<(&str, &PipelineResolution)>) -> PipelineStatus { + match closing { + Some((key, record)) => PipelineStatus::Resolved { + resolution: PipelineRef { kind: PipelineKind::Resolution, id: Short(key.to_string()) }, + record: record.clone(), + }, + None => PipelineStatus::InForce, + } +} + +impl Image { + pub(crate) fn empty() -> Self { + Self { + head: 0, + rows: Vec::new(), + status_changed: Vec::new(), + by_kind: BTreeMap::new(), + postings: BTreeMap::new(), + closers: BTreeMap::new(), + citers: HashMap::new(), + roots: BTreeMap::new(), + edges_in_target: true, + } + } + + /// The fold of a store's envelopes. Faults, in order: a pipeline document that does not decode, + /// the receipts (undecodable, not dense, a document written twice), a document no receipt wrote. + /// There is no other way to make an image: this replays the receipts through `advance`. + pub(crate) fn build( + envelopes: &Envelopes, + receipts: &Result, MindRefusal>, + ) -> Result { + let mut decoded: HashMap<(&str, &str), PipelineDocument> = HashMap::new(); + for (type_id, keyed) in envelopes { + if kind_of_type(type_id).is_none() { + continue; + } + for (key, envelope) in keyed { + let document = PipelineDocument::decode(envelope).map_err(|refusal| MindRefusal::Unavailable { + detail: format!("stored document {key} does not decode: {refusal}"), + })?; + decoded.insert((type_id.as_str(), key.as_str()), document); + } + } + let receipts = receipts.clone()?; + let index = AdmissionIndex::build(&receipts)?; + for (type_id, keyed) in envelopes { + if kind_of_type(type_id).is_some() + && let Some(key) = keyed.keys().find(|key| !index.contains(type_id, key)) + { + return Err(integrity(type_id, key, "has no commit receipt")); + } + } + let mut ordered: Vec<&HuginnCommitReceipt> = receipts.iter().collect(); + ordered.sort_by_key(|receipt| receipt.ordinal); + let mut image = Self::empty(); + for receipt in ordered { + let mut documents = Vec::new(); + for write in receipt.writes.iter().filter(|write| write.document_type != HuginnPersonaEntry::TYPE) { + let Some(kind) = kind_of_type(&write.document_type) else { continue }; + // A receipted document the store no longer holds gets no row; read_back names it. + if let Some(document) = decoded.remove(&(write.document_type.as_str(), write.document_key.as_str())) { + documents.push((kind, write.document_key.clone(), document)); + } + } + image.advance(&AdmissionFacts::of(receipt), documents)?; + } + Ok(image) + } + + pub(crate) fn id_of(&self, kind: PipelineKind, key: &str) -> Option { + self.by_kind.get(&kind)?.get(key).copied() + } + + pub(crate) fn row(&self, id: u32) -> &SelectionRow { + &self.rows[id as usize] + } + + /// The ids of one kind in key order. + pub(crate) fn of_kind(&self, kind: PipelineKind) -> impl Iterator { + self.by_kind.get(&kind).into_iter().flat_map(|keyed| keyed.iter().map(|(key, id)| (key.as_str(), *id))) + } + + /// The resolutions naming one subject, in resolution-key order. + pub(crate) fn closers_of(&self, kind: PipelineKind, id: &str) -> impl Iterator { + self.closers.get(&kind).and_then(|named| named.get(id)).into_iter().flat_map(|keyed| keyed.iter().map(|(key, id)| (key.as_str(), *id))) + } + + /// Every root with the newest row under it. + pub(crate) fn roots(&self) -> impl Iterator { + self.roots.iter().map(|(root, id)| (root.as_str(), self.row(*id))) + } + + /// Folds one receipt's documents, given in the receipt's write order, into the image. The same + /// function replays the log and follows each commit. + pub(crate) fn advance( + &mut self, + facts: &AdmissionFacts, + documents: Vec<(PipelineKind, String, PipelineDocument)>, + ) -> Result<(), MindRefusal> { + let first = self.rows.len() as u32; + // Phase 1: every document of the receipt gets a row, so a subject that sorts after its + // resolution in one batch is found by phase 2. + for (kind, key, document) in documents { + if self.id_of(kind, &key).is_some() { + return Err(integrity(kind.type_id(), &key, "is written by two receipts")); + } + let id = self.rows.len() as u32; + let edges = citations(&document); + for (role, target) in &edges { + let in_target = kind_of_id(&target.id.0).is_some_and(|named| role.target_kinds().contains(&named)); + self.edges_in_target &= in_target; + self.citers.entry(target.id.0.clone()).or_default().insert((role.name(), id)); + } + if let PipelineDocument::Resolution(resolution) = &document { + self.closers + .entry(resolution.subject.kind) + .or_default() + .entry(resolution.subject.id.0.clone()) + .or_default() + .insert(key.clone(), id); + } + self.roots.insert(root_and_local(&key).0.to_string(), id); + self.by_kind.entry(kind).or_default().insert(key.clone(), id); + let view = PipelineDocumentView { + id: PipelineRef { kind, id: Short(key.clone()) }, + document, + admission: facts.clone(), + status: PipelineStatus::InForce, + }; + self.rows.push(SelectionRow::new(kind, &key, None, view, edges)); + self.status_changed.push(facts.ordinal); + } + self.head = facts.ordinal; + // Phase 2: a new resolution changes its subject's status, and a subject that is itself a + // resolution reopens or closes its own subject in turn, so the chain is walked to the first + // document that is not a resolution. + let mut touched: BTreeSet = (first..self.rows.len() as u32).collect(); + for id in first..self.rows.len() as u32 { + let mut current = id; + while let PipelineDocument::Resolution(resolution) = &self.row(current).view.document { + let Some(subject) = self.id_of(resolution.subject.kind, &resolution.subject.id.0) else { break }; + touched.insert(subject); + current = subject; + } + } + for id in touched { + self.rebuild(id, id >= first, facts.ordinal); + } + Ok(()) + } + + /// The row with `status`, its base walked through the subjects held. + pub(crate) fn make_row(&self, id: u32, status: PipelineStatus) -> SelectionRow { + let old = self.row(id); + let mut view = old.view.clone(); + view.status = status; + let mut current = id; + while let PipelineDocument::Resolution(resolution) = &self.row(current).view.document { + let Some(subject) = self.id_of(resolution.subject.kind, &resolution.subject.id.0) else { break }; + current = subject; + } + let base = self.row(current); + SelectionRow::new(old.kind, &old.key, Some((&base.key, &base.view.document)), view, citations(&old.view.document)) + } + + fn rebuild(&mut self, id: u32, fresh: bool, ordinal: u64) { + let (kind, key) = { + let row = self.row(id); + (row.kind, row.key.clone()) + }; + let status = status_of(Docs::at(self, u64::MAX).closing_resolution(kind, &key)); + let row = self.make_row(id, status); + if !fresh { + self.unindex(id); + } + if row.view.status != self.row(id).view.status { + self.status_changed[id as usize] = ordinal; + } + self.rows[id as usize] = row; + self.index(id); + } + + fn index(&mut self, id: u32) { + let row = &self.rows[id as usize]; + for (alias, values) in row.alias_values() { + for value in values { + self.postings + .entry(row.kind) + .or_default() + .entry(alias) + .or_default() + .entry(value.clone()) + .or_default() + .insert(id); + } + } + } + + fn unindex(&mut self, id: u32) { + let row = &self.rows[id as usize]; + for (alias, values) in row.alias_values() { + for value in values { + if let Some(posting) = self.postings.get_mut(&row.kind).and_then(|aliases| aliases.get_mut(alias)).and_then(|values| values.get_mut(value)) { + posting.remove(&id); + } + } + } + } + + fn posting(&self, kind: PipelineKind, alias: &str, value: &str) -> Option<&BTreeSet> { + self.postings.get(&kind)?.get(alias)?.get(value) + } + + /// The row ids a selection can match at `as_of`, ascending, with those written after `as_of` + /// among them for the caller to drop: a superset of what `select` matches, never a verdict. + pub(crate) fn universe(&self, selection: &Selection, as_of: u64) -> Vec { + let hop = selection.has_hop(); + if hop && !self.edges_in_target { + return (0..self.rows.len() as u32).collect(); + } + let hop_only_cites = selection.cites.is_some() && selection.cited.is_none(); + let base = if hop_only_cites { BTreeSet::new() } else { self.base_candidates(selection, as_of) }; + let mut ids: BTreeSet = BTreeSet::new(); + let citers_of = |key: &str, role: Option<&str>| -> Vec { + self.citers + .get(key) + .into_iter() + .flatten() + .filter(|(named, _)| role.is_none_or(|role| role == *named)) + .map(|(_, citer)| *citer) + .collect() + }; + match (&selection.cites, &selection.cited) { + (None, None) => ids = base, + (Some(cites), None) => { + let key = cites.target.record_key.as_str(); + // A row that passes the hop cites the target, so the citers are the superset; the kinds, + // keys and fields only the substrate decides. + ids.extend(citers_of(key, cites.role.as_deref())); + ids.extend(self.any_kind(key)); + } + (cites, Some(cited)) => { + for id in &base { + ids.extend(citers_of(&self.row(*id).key, Some(&cited.role))); + } + if let Some(cites) = cites { + ids.extend(self.any_kind(&cites.target.record_key)); + } + ids.extend(base); + } + } + ids.into_iter().collect() + } + + fn any_kind(&self, key: &str) -> Vec { + match kind_of_id(key) { + Some(kind) => self.id_of(kind, key).into_iter().collect(), + None => PipelineKind::ALL.iter().filter_map(|kind| self.id_of(*kind, key)).collect(), + } + } + + /// The smallest set the kinds, keys and any_of fields of a selection allow. + fn base_candidates(&self, selection: &Selection, as_of: u64) -> BTreeSet { + if let Some(keys) = &selection.keys { + return keys.iter().flat_map(|key| self.any_kind(key)).collect(); + } + let kinds: Vec = match &selection.schemas { + Some(schemas) => schemas.iter().filter_map(|schema| kind_of_type(schema)).collect(), + None => PipelineKind::ALL.to_vec(), + }; + let mut out = BTreeSet::new(); + for kind in kinds { + let mut best: Option<(usize, Vec<&BTreeSet>)> = None; + for field in selection.fields.iter().flatten() { + if field.op != "any_of" { + continue; + } + let Some(alias) = Alias::from_name(&field.index) else { continue }; + if STATUS_ALIASES.contains(&alias) && as_of != self.head { + continue; + } + let sets: Vec<&BTreeSet> = + field.values.iter().flatten().filter_map(|value| self.posting(kind, alias.name(), value)).collect(); + let size = sets.iter().map(|set| set.len()).sum(); + if best.as_ref().is_none_or(|(smallest, _)| size < *smallest) { + best = Some((size, sets)); + } + } + let whole = self.by_kind.get(&kind).map_or(0, BTreeMap::len); + match best { + Some((size, sets)) if size < whole => out.extend(sets.into_iter().flatten().copied()), + _ => out.extend(self.by_kind.get(&kind).into_iter().flat_map(|keyed| keyed.values().copied())), + } + } + out + } +} + +#[cfg(test)] +impl Image { + /// An image of hand-made documents, one receipt each in the order given. + pub(crate) fn of_documents(documents: Vec<(PipelineKind, String, PipelineDocument)>) -> Self { + let mut image = Self::empty(); + for (n, document) in documents.into_iter().enumerate() { + let facts = AdmissionFacts { + receipt_id: format!("receipt-{}", n + 1), + admitted_at: "2026-09-16T12:00:00Z".into(), + provenance: crate::receipt::PipelineProvenance { + faculty: crate::receipt::Faculty::Hands, + agent: Short("claude".into()), + session: Short("session-1".into()), + tool: Short("admit".into()), + }, + ordinal: n as u64 + 1, + }; + image.advance(&facts, vec![document]).unwrap(); + } + image + } +} diff --git a/crates/huginn-mind/src/index.rs b/crates/huginn-mind/src/index.rs index 99fc188..abcfd62 100644 --- a/crates/huginn-mind/src/index.rs +++ b/crates/huginn-mind/src/index.rs @@ -10,9 +10,7 @@ use eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef}; use sha2::{Digest, Sha256}; -use crate::docs::Docs; use crate::mind::Mind; -use crate::query::AdmissionIndex; use crate::refusal::MindRefusal; use crate::rows::root_and_local; use crate::store::MindStore; @@ -114,32 +112,31 @@ impl Mind { /// document the mind does not hold is a refusal, not a silent omission: a /// caller names what a commit just landed. pub fn index_entries(&self, only: Option<&[PipelineRef]>) -> Result, MindRefusal> { - let documents: Vec<(PipelineKind, String, PipelineDocument)> = match only { + let image = self.image()?; + let held: Vec<(PipelineKind, &str, &PipelineDocument, u64)> = match only { Some(refs) => { let mut held = Vec::with_capacity(refs.len()); for reference in refs { reference.validate_ref()?; - let document = self.get(reference.kind, &reference.id.0)?.ok_or_else(|| { + let row = image.id_of(reference.kind, &reference.id.0).map(|id| image.row(id)).ok_or_else(|| { MindRefusal::Unavailable { detail: format!("index: {} is not held", reference.id.0) } })?; - held.push((reference.kind, reference.id.0.clone(), document)); + held.push((reference.kind, row.key.as_str(), &row.view.document, row.ordinal)); } held } - None => Docs::from_image(self.envelopes())? - .image - .into_iter() - .map(|held| (held.kind, held.key, held.document)) - .collect(), + None => { + let mut held: Vec<_> = image.rows.iter().map(|row| (row.kind, row.key.as_str(), &row.view.document, row.ordinal)).collect(); + held.sort_by(|left, right| (left.0.type_id(), left.1).cmp(&(right.0.type_id(), right.1))); + held + } }; - let facts = AdmissionIndex::build(self)?; let mut entries = Vec::new(); - for (kind, key, document) in documents { - let Some(text) = index_text(&document) else { continue }; - let ordinal = facts.ordinal_of(kind, &key)?; + for (kind, key, document, ordinal) in held { + let Some(text) = index_text(document) else { continue }; entries.push(IndexEntry { - root: root_and_local(&key).0.to_string(), - id: PipelineRef { kind, id: key.as_str().into() }, + root: root_and_local(key).0.to_string(), + id: PipelineRef { kind, id: key.into() }, text_sha256: sha256_hex(&text), text, ordinal, diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 526bffd..a57e5d1 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -10,9 +10,10 @@ //! cross-document rule, every derived write, and the replay check, in one //! ordered path that `admit`, the daemon's sink and the hand-off all share. //! -//! The read side derives status, joins the receipts and answers one typed -//! selection (CultNet's, over rows this crate supplies) through the same -//! `docs` the rules use; nothing is stored for it. +//! The read side answers one typed selection (CultNet's, over rows this crate +//! supplies) from the mind's held image: a fold of the receipts, held by `Mind`, +//! advanced per commit, rebuilt on restart and on a lost swap. Status is derived +//! through the same `docs` the rules use; nothing is stored for it. //! //! The wire vocabulary and its operation envelope (`envelope`) are here too, //! so the daemon and the client share one set @@ -27,11 +28,14 @@ pub mod admission; mod docs; +mod image; pub mod envelope; 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; @@ -41,6 +45,9 @@ pub mod wire; #[cfg(test)] pub(crate) mod fixtures; +#[cfg(test)] +mod read_tests; + pub use docs::CitationRole; pub use index::{INDEX_TEXT_MAX_BYTES, INDEX_TEXT_VERSION, IndexEntry, index_text}; pub use admission::{BATCH_MAX, PipelineAdmissionBatch, PipelineAdmissionOutcome}; diff --git a/crates/huginn-mind/src/mind.rs b/crates/huginn-mind/src/mind.rs index 470bb83..5df8726 100644 --- a/crates/huginn-mind/src/mind.rs +++ b/crates/huginn-mind/src/mind.rs @@ -9,13 +9,16 @@ //! every gate before it attaches, so a refused store is never read into an //! image. +use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use cultcache_rs::{CultCache, CultCacheEnvelope, DatabaseEntry, OwnedRedbMessagePackBackingStore}; use cultnet_rs::CursorKey; use eureka_pipeline::{PIPELINE_SCHEMA_EPOCH, PipelineDocument, PipelineKind, Slug, register_pipeline_document_types}; -use crate::persona::HuginnPersonaEntry; +use crate::image::Image; +use crate::persona::{HuginnPersonaEntry, persona_receipt_of}; +use crate::query::AdmissionFacts; use crate::receipt::HuginnCommitReceipt; use crate::refusal::MindRefusal; use crate::store::MindStore; @@ -91,15 +94,53 @@ fn is_known_type(type_id: &str) -> bool { || PipelineKind::ALL.iter().any(|kind| kind.type_id() == type_id) } -/// One instance's mind: the declared instance, the store, the registered -/// cache attached to it, and the image (the store's envelopes as last pulled, -/// in identity order). The image is a cache of the store and is re-pulled -/// after every commit; nothing writes it directly. +/// The store's envelopes by type and key: the bytes as stored, in identity order. +pub(crate) type Envelopes = BTreeMap>; + +fn envelopes_of(raw: Vec) -> Envelopes { + let mut envelopes = Envelopes::new(); + for envelope in raw { + envelopes.entry(envelope.r#type.clone()).or_default().insert(envelope.key.clone(), envelope); + } + envelopes +} + +/// Every commit receipt in the envelopes, in receipt-id order. +pub(crate) fn decode_receipts(envelopes: &Envelopes) -> Result, MindRefusal> { + envelopes + .get(HuginnCommitReceipt::TYPE) + .into_iter() + .flat_map(|keyed| keyed.values()) + .map(|envelope| { + rmp_serde::from_slice::(&envelope.payload).map_err(|error| MindRefusal::Unavailable { + detail: format!("receipt {} does not decode: {error}", envelope.key), + }) + }) + .collect() +} + +/// The one fold: the image and the persona's receipt, both read out of the receipts. A fault is +/// held, not raised, so a faulted store still opens and every read answers the fault. +fn fold(envelopes: &Envelopes, instance: &Slug) -> (Result, Result, MindRefusal>) { + let receipts = decode_receipts(envelopes); + let persona = receipts.as_ref().map(|receipts| persona_receipt_of(receipts, instance)).map_err(Clone::clone); + (Image::build(envelopes, &receipts), persona) +} + +/// One instance's mind: the declared instance, the store, the registry of +/// the nineteen schemas, the store's envelopes as last pulled, and the image +/// folded from them. The envelopes are the store's bytes and the image is a +/// pure fold of the receipts among them: both advance with each commit and +/// are rebuilt from the store on a lost swap, a store error and a restart. +/// A cache of the store, never its authority; only `Mind::advance` and +/// `refresh` write them. pub struct Mind { instance: Slug, store: S, cache: CultCache, - image: Vec, + pub(crate) envelopes: Envelopes, + image: Result, + pub(crate) persona_receipt: Result, MindRefusal>, /// Mints and verifies this process's selection cursors. Never stored: a /// cursor does not survive a reopen, which the substrate accepts. cursor_key: CursorKey, @@ -186,8 +227,10 @@ impl Mind { refuse_foreign_epoch(&raw)?; refuse_foreign_types(&raw)?; refuse_foreign_identity(&raw, instance)?; - let (cache, image) = attach(store.clone())?; - Ok(Self { instance: instance.clone(), store, cache, image, cursor_key: CursorKey::random() }) + let cache = schema_cache()?; + let envelopes = envelopes_of(raw); + let (image, persona_receipt) = fold(&envelopes, instance); + Ok(Self { instance: instance.clone(), store, cache, envelopes, image, persona_receipt, cursor_key: CursorKey::random() }) } pub fn instance(&self) -> &Slug { @@ -209,10 +252,20 @@ impl Mind { Ok(()) } - /// The image: every envelope the store held at the last pull, in - /// identity order. Cut 10's snapshot source reads this. - pub fn envelopes(&self) -> &[CultCacheEnvelope] { - &self.image + /// Every envelope the store held at the last pull, in identity order. + pub fn envelopes(&self) -> impl Iterator { + self.envelopes.values().flat_map(|keyed| keyed.values()) + } + + /// The held image, or the fault the store holds: the same answer at any snapshot. + pub(crate) fn image(&self) -> Result<&Image, MindRefusal> { + self.image.as_ref().map_err(Clone::clone) + } + + /// The image folded from a fresh pull of the store: what a restart would hold. + pub(crate) fn fresh_image(&self) -> Result { + let envelopes = envelopes_of(self.store.pull_all().map_err(unavailable)?); + fold(&envelopes, &self.instance).0 } pub(crate) fn cursor_key(&self) -> &CursorKey { @@ -220,7 +273,7 @@ impl Mind { } pub fn is_empty(&self) -> bool { - self.image.is_empty() + self.envelopes.values().all(BTreeMap::is_empty) } /// The stored bytes of one document, by kind and id. @@ -237,15 +290,7 @@ impl Mind { /// Every commit receipt in the image, in receipt-id order. pub(crate) fn receipts(&self) -> Result, MindRefusal> { - self.image - .iter() - .filter(|envelope| envelope.r#type == HuginnCommitReceipt::TYPE) - .map(|envelope| { - rmp_serde::from_slice::(&envelope.payload).map_err(|error| { - MindRefusal::Unavailable { detail: format!("receipt {} does not decode: {error}", envelope.key) } - }) - }) - .collect() + decode_receipts(&self.envelopes) } /// The mind's own identity: the receipt id of its first admission @@ -259,7 +304,7 @@ impl Mind { } pub(crate) fn raw_envelope(&self, type_id: &str, key: &str) -> Option<&CultCacheEnvelope> { - self.image.iter().find(|envelope| envelope.r#type == type_id && envelope.key == key) + self.envelopes.get(type_id)?.get(key) } pub(crate) fn cache(&self) -> &CultCache { @@ -270,11 +315,55 @@ impl Mind { &self.store } - /// Re-pulls the image from the store through the attached cache. + /// Rebuilds the envelopes and the image from the store: the one rebuild, run only when a swap + /// was lost or the store errored, the two signals that the store moved beneath the holder. + /// One rule for a look that fails: the holder no longer knows the store, so it holds the fault + /// (reads and admission refuse with it) until a restart, exactly as a fold fault is held. pub(crate) fn refresh(&mut self) -> Result<(), MindRefusal> { - self.cache.pull_all_backing_stores().map_err(unavailable)?; - self.image = snapshot(&self.cache); - Ok(()) + match self.store.pull_all().map_err(unavailable) { + Ok(raw) => { + self.envelopes = envelopes_of(raw); + (self.image, self.persona_receipt) = fold(&self.envelopes, &self.instance); + Ok(()) + } + Err(fault) => { + self.image = Err(fault.clone()); + self.persona_receipt = Err(fault.clone()); + Err(fault) + } + } + } + + /// Folds a commit the compare-and-swap said landed into the held state. If the fold itself + /// faults on bytes that already landed, the state is rebuilt from the store, so the fault held + /// is exactly what a reopen would hold, and the commit stays committed. + pub(crate) fn advance(&mut self, receipt: &HuginnCommitReceipt, receipt_envelope: CultCacheEnvelope, writes: &[CultCacheEnvelope]) { + for envelope in writes.iter().chain(std::iter::once(&receipt_envelope)) { + self.envelopes.entry(envelope.r#type.clone()).or_default().insert(envelope.key.clone(), envelope.clone()); + } + if receipt.writes.iter().any(|write| write.document_type == HuginnPersonaEntry::TYPE && write.document_key == self.instance.0) + && let Ok(held) = &mut self.persona_receipt + { + *held = Some(receipt.receipt_id.clone()); + } + let Ok(image) = &mut self.image else { return }; + let mut documents = Vec::new(); + let mut faulted = false; + for write in &receipt.writes { + let Some(kind) = PipelineKind::ALL.iter().copied().find(|kind| kind.type_id() == write.document_type) else { continue }; + let Some(envelope) = writes.iter().find(|envelope| envelope.r#type == write.document_type && envelope.key == write.document_key) else { + faulted = true; + break; + }; + match PipelineDocument::decode(envelope) { + Ok(document) => documents.push((kind, write.document_key.clone(), document)), + Err(_) => faulted = true, + } + } + if faulted || image.advance(&AdmissionFacts::of(receipt), documents).is_err() { + // A failed look holds its own fault. + let _ = self.refresh(); + } } } @@ -343,22 +432,6 @@ fn refuse_foreign_identity(raw: &[CultCacheEnvelope], instance: &Slug) -> Result Ok(()) } -/// Step 5: the registered cache, the store as its one generic home, and the -/// image pulled through it. -fn attach(store: S) -> Result<(CultCache, Vec), MindRefusal> { - let mut cache = schema_cache()?; - cache.add_generic_backing_store(store).map_err(unavailable)?; - cache.pull_all_backing_stores().map_err(unavailable)?; - let image = snapshot(&cache); - Ok((cache, image)) -} - -fn snapshot(cache: &CultCache) -> Vec { - let mut image = cache.snapshot_envelopes(); - image.sort_by(|left, right| (&left.r#type, &left.key).cmp(&(&right.r#type, &right.key))); - image -} - #[cfg(test)] mod tests { use super::*; @@ -464,11 +537,11 @@ mod tests { Some(MindRefusal::ForeignEpoch { found: "2 records".into(), expected: PIPELINE_SCHEMA_EPOCH.into() }) ); } - // A store that passes every gate attaches, which is the second pull. + // A store that passes every gate is read once: the pull the gates ran on is the one the image folds. let store = planted(vec![epoch(), prepare(&instance(INSTANCE))]); let mind = Mind::open_with(store.clone(), &yggdrasil).unwrap(); - assert_eq!(store.pull_count(), 2); - assert_eq!(mind.envelopes().len(), 2); + assert_eq!(store.pull_count(), 1); + assert_eq!(mind.envelopes().count(), 2); assert!(!mind.is_empty()); let empty = MemoryStore::new(); assert!(Mind::open_with(empty, &yggdrasil).unwrap().is_empty()); @@ -559,7 +632,7 @@ mod tests { let store = OwnedRedbMessagePackBackingStore::new(&moved).unwrap(); let mind = Mind::open_with(store, &slug(INSTANCE)).unwrap(); assert_eq!(mind.instance(), &slug(INSTANCE)); - assert_eq!(mind.envelopes().len(), 2); + assert_eq!(mind.envelopes().count(), 2); } /// One check, one owner: the value admission refuses a foreign instance diff --git a/crates/huginn-mind/src/oracle.rs b/crates/huginn-mind/src/oracle.rs new file mode 100644 index 0000000..55b3b5d --- /dev/null +++ b/crates/huginn-mind/src/oracle.rs @@ -0,0 +1,336 @@ +//! 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::{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); + SelectionRow::new(doc.kind, &doc.key, Some((&base.key, &base.document)), 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 rows: Vec<&SelectionRow> = rows.iter().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 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/persona.rs b/crates/huginn-mind/src/persona.rs index 554149c..ac3e83c 100644 --- a/crates/huginn-mind/src/persona.rs +++ b/crates/huginn-mind/src/persona.rs @@ -19,7 +19,10 @@ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use serde_json::Value; +use eureka_pipeline::Slug; + use crate::mind::Mind; +use crate::receipt::HuginnCommitReceipt; use crate::refusal::MindRefusal; use crate::store::MindStore; @@ -124,6 +127,15 @@ pub(crate) fn updated_at(value: &Value) -> Option<&str> { value.get("updatedAt").and_then(Value::as_str) } +/// The receipt that wrote the persona's stored bytes: the newest receipt whose writes name it. +pub(crate) fn persona_receipt_of(receipts: &[HuginnCommitReceipt], instance: &Slug) -> Option { + receipts + .iter() + .filter(|receipt| receipt.writes.iter().any(|write| write.document_type == HuginnPersonaEntry::TYPE && write.document_key == instance.0)) + .max_by_key(|receipt| receipt.ordinal) + .map(|receipt| receipt.receipt_id.clone()) +} + impl Mind { /// This mind's persona envelope, if one is stored. pub(crate) fn persona_envelope(&self) -> Option<&CultCacheEnvelope> { @@ -146,14 +158,7 @@ impl Mind { let value = decode(envelope)?; let integrity = |detail: &str| MindRefusal::Unavailable { detail: detail.into() }; let updated_at = updated_at(&value).ok_or_else(|| integrity("the stored persona has no updatedAt"))?.to_string(); - let identity = (envelope.r#type.as_str(), envelope.key.as_str()); - let receipt_id = self - .receipts()? - .into_iter() - .filter(|receipt| receipt.writes.iter().any(|write| write.identity() == identity)) - .max_by_key(|receipt| receipt.ordinal) - .ok_or_else(|| integrity("the stored persona has no receipt"))? - .receipt_id; + let receipt_id = self.persona_receipt.clone()?.ok_or_else(|| integrity("the stored persona has no receipt"))?; Ok(Some(PersonaStateView { value, updated_at, receipt_id })) } } @@ -585,7 +590,7 @@ mod tests { let mut with_one = base(); with_one.push(entry(INSTANCE, persona(INSTANCE, T1))); let mind = Mind::open_with(planted(with_one.clone()), &yggdrasil).unwrap(); - assert_eq!(mind.envelopes().len(), 3); + assert_eq!(mind.envelopes().count(), 3); assert_eq!(persona_in_status(&mind), crate::wire::PersonaStatus::Unreadable); assert!(mind.persona().is_err()); @@ -606,6 +611,31 @@ mod tests { /// Status answers in the door's own terms: a stored persona the door /// cannot read is `Unreadable`, not the `Absent` of a mind with none, while /// `PersonaGet` answers `Unavailable` for the same bytes. + /// The persona's receipt is the newest put that wrote it, advanced with each put and the same + /// after a restart. + #[test] + fn the_persona_receipt_is_the_newest_put_held_and_after_a_restart() { + let store = MemoryStore::new(); + let mut mind = seeded_over(&store); + committed(put(&mut mind, persona(INSTANCE, T1), None)); + let first = mind.persona().unwrap().unwrap().receipt_id; + committed(put(&mut mind, persona(INSTANCE, T2), Some(T1))); + let second = mind.persona().unwrap().unwrap().receipt_id; + assert_ne!(first, second); + let newest = mind + .receipts() + .unwrap() + .into_iter() + .filter(|receipt| receipt.writes.iter().any(|write| write.document_type == HuginnPersonaEntry::TYPE)) + .max_by_key(|receipt| receipt.ordinal) + .unwrap() + .receipt_id; + assert_eq!(second, newest); + let reopened = Mind::open_with(store, &slug(INSTANCE)).unwrap(); + assert_eq!(reopened.persona().unwrap().unwrap().receipt_id, second); + assert_eq!(reopened.persona_receipt, mind.persona_receipt); + } + #[test] fn status_says_unreadable_where_the_door_says_unavailable() { let yggdrasil = slug(INSTANCE); diff --git a/crates/huginn-mind/src/query.rs b/crates/huginn-mind/src/query.rs index 1897849..93217a5 100644 --- a/crates/huginn-mind/src/query.rs +++ b/crates/huginn-mind/src/query.rs @@ -1,9 +1,12 @@ //! The read side of a mind: a document with the facts of its admission and //! its status, and one typed selection over them. //! -//! Nothing here is stored. Status is derived per call through the same -//! `docs::Docs` the admission rules use, so a withdrawn resolution reopens its -//! subject for a reader exactly when it reopens it for a rule. The facts come +//! Nothing derived here is stored in a document. The rows come from the mind's +//! held `image::Image`, a fold of the receipts; status is derived through the +//! same `docs::Docs` the admission rules use, at the head when the image was +//! advanced and at an old snapshot when a reader asks, so a withdrawn +//! resolution reopens its subject for a reader exactly when it reopens it for +//! a rule. The facts come //! from the commit receipts, from their `writes` and from nothing else: a //! document's `admitted_at` is the `now` its batch was admitted at //! (`committed_at`), never the store's `stored_at` stamp. A pipeline document @@ -13,18 +16,19 @@ //! `query` is CultNet's selection over rows this crate supplies (`rows.rs`). //! The substrate owns the filter, the order, the hop, the cursor and the //! paging; this module owns the snapshot (`Reader::at`: every derivation runs -//! over the documents admitted at or before the page's `asOf`), the +//! over the documents admitted at or before the page's `asOf`, and the image +//! only narrows which of its rows the substrate is handed), the //! projection to a header or a whole view, and the typing of the edges. //! //! `rank` is the semantic read: the daemon's index supplies candidate ids and //! scores, and this module joins them back through the mind and lets the //! selection decide. No clock, no store handle, no network. -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use cultcache_rs::DatabaseEntry; use cultnet_rs::{ - Cursor, EdgeAnchor, EdgeMatch, Evaluation, LIMIT_MAX, LIMIT_MIN, PROJECTION_DOCUMENT, Row, Selection, select, validate, + Cursor, EdgeAnchor, EdgeMatch, Evaluation, LIMIT_MAX, LIMIT_MIN, PROJECTION_DOCUMENT, Selection, select, validate, }; use eureka_pipeline::{ ClaimOutcome, CommitRange, Date, FindingConfidence, FindingOrigin, FindingSeverity, Label, Line, MemoryOrigin, MemorySort, OrgRepo, @@ -34,12 +38,13 @@ use eureka_pipeline::{ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use crate::docs::{CitationRole, Docs, Held, citations, kind_of_type}; +use crate::docs::{CitationRole, Docs, kind_of_type}; +use crate::image::{Image, status_of}; use crate::mind::{Mind, schema_cache}; use crate::persona::HuginnPersonaEntry; -use crate::receipt::{self, PipelineProvenance}; +use crate::receipt::{HuginnCommitReceipt, PipelineProvenance}; use crate::refusal::MindRefusal; -use crate::rows::{SelectionRow, Vocabulary, refuse_values, root_and_local}; +use crate::rows::{SelectionRow, Vocabulary, refuse_values}; use crate::store::MindStore; /// What admission recorded about a document when it landed: the receipt that @@ -325,57 +330,51 @@ pub struct Ranked { pub dropped: Vec, } -/// The admission facts of every stored document, built once per read call from -/// the receipts. `writes` only: a document a later batch cited as a strong read -/// keeps the receipt that wrote it. A10 admits one write per pipeline identity, -/// so two receipts naming one pipeline document is an integrity fault, not a -/// choice to make. The persona document is the one identity replaced by design -/// (`put_persona` is a compare-and-swap on `updatedAt`) and is not a pipeline -/// document: it has no admission facts here, and `Mind::persona` reads its own -/// receipt. -pub(crate) struct AdmissionIndex(BTreeMap<(String, String), AdmissionFacts>); +impl AdmissionFacts { + /// What a receipt records about every document it wrote. + pub(crate) fn of(receipt: &HuginnCommitReceipt) -> Self { + Self { + receipt_id: receipt.receipt_id.clone(), + admitted_at: receipt.committed_at.clone(), + provenance: receipt.provenance.clone(), + ordinal: receipt.ordinal, + } + } +} + +/// The identities the receipts wrote, checked once per build. A chain that is not exactly +/// `{1..=N}` refuses before a single ordinal is trusted. `writes` only: a document a later batch +/// cited as a strong read keeps the receipt that wrote it. A10 admits one write per pipeline +/// identity, so two receipts naming one pipeline document is an integrity fault, not a choice to +/// make. The persona document is the one identity replaced by design (`put_persona` is a +/// compare-and-swap on `updatedAt`) and is not a pipeline document: it has no admission facts +/// here, and the mind holds its receipt apart. +pub(crate) struct AdmissionIndex(HashSet<(String, String)>); impl AdmissionIndex { - pub(crate) fn build(mind: &Mind) -> Result { - // Every reader of the ordinal asks the same density check admission - // does, through `head`: a chain that is not exactly `{1..=N}` refuses - // here before a single ordinal is copied out, not only when admission - // happens to be the caller. - receipt::head(mind)?; - let mut facts = BTreeMap::new(); - for receipt in mind.receipts()? { + pub(crate) fn build(receipts: &[HuginnCommitReceipt]) -> Result { + let mut ordinals = receipts.iter().map(|receipt| receipt.ordinal).collect::>(); + ordinals.sort_unstable(); + if !ordinals.iter().copied().eq(1..=ordinals.len() as u64) { + return Err(MindRefusal::Unavailable { detail: "receipt ordinals are not 1..=N: found a duplicate or a gap".into() }); + } + let mut identities = HashSet::new(); + for receipt in receipts { for write in receipt.writes.iter().filter(|write| write.document_type != HuginnPersonaEntry::TYPE) { - let identity = (write.document_type.clone(), write.document_key.clone()); - 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(identity, landed).is_some() { + if !identities.insert((write.document_type.clone(), write.document_key.clone())) { return Err(integrity(&write.document_type, &write.document_key, "is written by two receipts")); } } } - Ok(Self(facts)) + Ok(Self(identities)) } - fn of(&self, held: &Held) -> Result { - self.facts_of(held.kind, &held.key) + pub(crate) fn contains(&self, type_id: &str, key: &str) -> bool { + self.0.contains(&(type_id.to_string(), key.to_string())) } - fn facts_of(&self, kind: PipelineKind, key: &str) -> Result { - let type_id = kind.type_id(); - self.0 - .get(&(type_id.to_string(), key.to_string())) - .cloned() - .ok_or_else(|| integrity(type_id, key, "has no commit receipt")) - } - - /// The ordinal of the receipt that wrote a document: the projection's - /// position for it. - pub(crate) fn ordinal_of(&self, kind: PipelineKind, key: &str) -> Result { - self.facts_of(kind, key).map(|facts| facts.ordinal) + pub(crate) fn identities(&self) -> impl Iterator { + self.0.iter() } } @@ -400,89 +399,75 @@ impl RootTouched { } } -/// One read call's working set: the image as of one snapshot and the facts it -/// was built from. Built once per call and thrown away with it; nothing here -/// outlives the answer. -struct Reader { - docs: Docs, - facts: AdmissionIndex, +/// One read call's view of the held image at one snapshot. Nothing is decoded or built here: a row +/// is the image's own, unless its status changed after `as_of`, in which case it is rebuilt once +/// with the status the closing resolutions give at `as_of`. +pub(crate) struct Reader<'i> { + docs: Docs<'i>, + image: &'i Image, + as_of: u64, + patched: HashMap, } -impl Reader { - /// The mind as it stood when receipt `as_of` landed: the documents whose - /// writing receipt has `ordinal <= as_of`, and every derivation (in - /// force, the closing resolution, `base`, citations) over that image - /// alone. The facts are built over all receipts, and every held document - /// must have its receipt whatever `as_of` is, so an orphan or a double - /// write refuses at any snapshot. - fn at(mind: &Mind, as_of: u64) -> Result { - let mut docs = Docs::from_image(mind.envelopes())?; - let facts = AdmissionIndex::build(mind)?; - let mut visible = Vec::with_capacity(docs.image.len()); - for held in docs.image { - if facts.of(&held)?.ordinal <= as_of { - visible.push(held); - } - } - docs.image = visible; - Ok(Self { docs, facts }) +impl<'i> Reader<'i> { + /// The mind as it stood when receipt `as_of` landed. The held fault, if the store has one, is + /// answered at any snapshot. + pub(crate) fn at(mind: &'i Mind, as_of: u64) -> Result { + let image = mind.image()?; + Ok(Self { docs: Docs::at(image, as_of), image, as_of, patched: HashMap::new() }) } - fn held(&self, kind: PipelineKind, key: &str) -> Option<&Held> { - self.docs.image.iter().find(|held| held.kind == kind && held.key == key) + fn status_at(&self, id: u32) -> PipelineStatus { + let row = self.image.row(id); + status_of(self.docs.closing_resolution(row.kind, &row.key)) + } + + /// Whether the row exists at `as_of`; when it does, its snapshot row is ready for `row`. + pub(crate) fn prepare(&mut self, id: u32) -> bool { + if self.image.row(id).ordinal > self.as_of { + return false; + } + if self.image.status_changed[id as usize] > self.as_of && !self.patched.contains_key(&id) { + let row = self.image.make_row(id, self.status_at(id)); + self.patched.insert(id, row); + } + true + } + + /// A prepared row. + pub(crate) fn row(&self, id: u32) -> &SelectionRow { + if self.image.status_changed[id as usize] > self.as_of { &self.patched[&id] } else { self.image.row(id) } + } + + /// The rows a selection can match at this snapshot: a superset the substrate then decides over. + pub(crate) fn universe(&mut self, selection: &Selection) -> Vec { + let ids = self.image.universe(selection, self.as_of); + ids.into_iter().filter(|id| self.prepare(*id)).collect() + } + + pub(crate) fn rows(&self, ids: &[u32]) -> Vec<&SelectionRow> { + ids.iter().map(|id| self.row(*id)).collect() } /// The lookup `Mind::view` answers through, so a caller holding one `Reader` /// views many documents from one snapshot. - fn view(&self, id: &PipelineRef) -> Result, MindRefusal> { - self.held(id.kind, &id.id.0).map(|held| self.view_of(held)).transpose() + pub(crate) fn id_of(&self, kind: PipelineKind, key: &str) -> Option { + self.image.id_of(kind, key) } - fn view_of(&self, held: &Held) -> Result { - let admission = self.facts.of(held)?; - let status = match self.docs.closing_resolution(held.kind, &held.key) { - Some((key, record)) => PipelineStatus::Resolved { - resolution: PipelineRef { kind: PipelineKind::Resolution, id: Short(key.to_string()) }, - record: record.clone(), - }, - None => PipelineStatus::InForce, - }; - Ok(PipelineDocumentView { - id: PipelineRef { kind: held.kind, id: Short(held.key.clone()) }, - document: held.document.clone(), - admission, - status, - }) - } - - /// The document `repo` and `cut` are read through: a resolution's - /// subject, and its subject's subject when that is a resolution too (the - /// Q19 cap stops the chain at two). A7 puts every subject in the image, - /// and a resolution key is strictly longer than the id it resolves, so - /// this walks down and stops. Anything that is not a resolution is its - /// own base. - fn base<'b>(&'b self, held: &'b Held) -> &'b Held { - let mut current = held; - while let PipelineDocument::Resolution(resolution) = ¤t.document { - let Some(subject) = self.held(resolution.subject.kind, &resolution.subject.id.0) else { - return current; - }; - current = subject; + pub(crate) fn view(&self, id: &PipelineRef) -> Result, MindRefusal> { + let Some(row) = self.image.id_of(id.kind, &id.id.0) else { return Ok(None) }; + let held = self.image.row(row); + if held.ordinal > self.as_of { + return Ok(None); } - current + if self.image.status_changed[row as usize] <= self.as_of { + return Ok(Some(held.view.clone())); + } + Ok(Some(self.image.make_row(row, self.status_at(row)).view)) } fn roots_touched(&self) -> Vec { - let mut newest: BTreeMap<&str, &AdmissionFacts> = BTreeMap::new(); - for ((type_id, key), facts) in &self.facts.0 { - if kind_of_type(type_id).is_none() { - continue; - } - let root = root_and_local(key).0; - if newest.get(root).is_none_or(|seen| seen.ordinal < facts.ordinal) { - newest.insert(root, facts); - } - } let docs = &self.docs; let campaigns: BTreeMap<&str, &PipelineCampaign> = docs .of_kind(PipelineKind::Campaign) @@ -491,19 +476,9 @@ impl Reader { _ => None, }) .collect(); - newest - .into_iter() - .map(|(root, facts)| RootTouched::of(root, facts, campaigns.get(root).map_or(&[][..], |campaign| docs.repos_in_force(campaign)))) - .collect() - } - - fn rows(&self) -> Result, MindRefusal> { - self.docs - .image - .iter() - .map(|held| { - Ok(SelectionRow::new(held, self.base(held), self.view_of(held)?, citations(&held.document))) - }) + self.image + .roots() + .map(|(root, newest)| RootTouched::of(root, &newest.view.admission, campaigns.get(root).map_or(&[][..], |campaign| docs.repos_in_force(campaign)))) .collect() } } @@ -524,7 +499,7 @@ impl Mind { /// vocabulary for one rule. pub fn view(&self, id: &PipelineRef) -> Result, MindRefusal> { id.validate_ref()?; - Reader::at(self, receipt::head(self)?)?.view(id) + Reader::at(self, self.image()?.head)?.view(id) } /// The upgrade gate (ruling readback-gate-and-materialized-image): every @@ -532,21 +507,29 @@ impl Mind { /// stored key, type and payload bytes, and views back unchanged through the /// read path `view` uses; every pipeline write a receipt names is stored, /// and every stored document has its receipt. Returns how many documents - /// it read back; an empty store is a fault. One `Reader` serves every view, - /// because `Reader::at` is the expensive build. Admits nothing. A fault - /// names the document's key and what failed, never its content. + /// it read back; an empty store is a fault. The held image is first set beside a + /// fresh fold of the store and must equal it. That clause proves the holder's advance + /// only for a mind that has committed since it opened; on the just-opened mind the + /// upgrade reads, both sides are one fold of the same bytes, so a wrong fold passes + /// it. What the gate proves today is the per-document read-back below. One pass over + /// the mind, O(N). Admits nothing. A fault names the document's key and what failed, + /// never its content. pub fn read_back(&self) -> Result { - self.read_back_through(Reader::view) + self.read_back_through(|reader, id| reader.view(id)) } /// The gate's one body; `lookup` is the read path under test, which tests /// replace with a lying one to prove the view comparison refuses. fn read_back_through( &self, - lookup: impl Fn(&Reader, &PipelineRef) -> Result, MindRefusal>, + lookup: impl Fn(&Reader<'_>, &PipelineRef) -> Result, MindRefusal>, ) -> Result { let cache = schema_cache()?; - let reader = Reader::at(self, receipt::head(self)?)?; + let image = self.image()?; + if *image != self.fresh_image()? { + return Err(MindRefusal::Unavailable { detail: "the held image differs from a fresh fold".into() }); + } + let reader = Reader::at(self, image.head)?; let mut count = 0; for envelope in self.envelopes() { let Some(kind) = kind_of_type(&envelope.r#type) else { continue }; @@ -566,10 +549,14 @@ impl Mind { } count += 1; } - for (type_id, key) in reader.facts.0.keys() { - if kind_of_type(type_id).is_some() && self.raw_envelope(type_id, key).is_none() { - return Err(integrity(type_id, key, "is named by a receipt but not stored")); - } + let receipted = AdmissionIndex::build(&self.receipts()?)?; + let mut missing: Vec<&(String, String)> = receipted + .identities() + .filter(|(type_id, key)| kind_of_type(type_id).is_some() && self.raw_envelope(type_id, key).is_none()) + .collect(); + missing.sort(); + if let Some((type_id, key)) = missing.first() { + return Err(integrity(type_id, key, "is named by a receipt but not stored")); } if count == 0 { return Err(MindRefusal::Unavailable { detail: "the store holds no pipeline document".into() }); @@ -587,7 +574,7 @@ impl Mind { pub fn query(&self, selection: &Selection) -> Result { validate(selection, &Vocabulary).map_err(cultnet_rs::SelectionRefusal::Invalid)?; let selection = refuse_values(selection)?; - let head = receipt::head(self)?; + let head = self.image()?.head; let as_of = match selection.cursor.as_deref().filter(|cursor| !cursor.is_empty()) { Some(cursor) => { let as_of = Cursor::parse(cursor)?.as_of; @@ -600,17 +587,31 @@ impl Mind { } None => head, }; - let rows = Reader::at(self, as_of)?.rows()?; - let evaluation = select(&Vocabulary, &rows, &selection, as_of, self.cursor_key())?; - Self::page(&selection, evaluation, as_of, None) + self.select_at(&selection, as_of) + } + + #[cfg(test)] + /// The doors `query` passes, then the one body at a snapshot the caller names: the entry the + /// oracle tests set against the oracle at every `as_of`. + pub(crate) fn query_at(&self, selection: &Selection, as_of: u64) -> Result { + validate(selection, &Vocabulary).map_err(cultnet_rs::SelectionRefusal::Invalid)?; + let selection = refuse_values(selection)?; + self.select_at(&selection, as_of) + } + + fn select_at(&self, selection: &Selection, as_of: u64) -> Result { + let mut reader = Reader::at(self, as_of)?; + let ids = reader.universe(selection); + let rows = reader.rows(&ids); + let evaluation = select(&Vocabulary, &rows, selection, as_of, self.cursor_key())?; + Self::page(selection, evaluation, as_of, None) } /// Every root this mind holds a document under, with its newest admission and, for a root that is an in-force /// campaign, the repos in force (`Docs::repos_in_force`, never the campaign row's opening set). A resolution's /// key carries its subject's root, so a later resolution is a touch of that root. One pass at the head, no rows. pub fn roots_touched(&self) -> Result, MindRefusal> { - let head = receipt::head(self)?; - Ok(Reader::at(self, head)?.roots_touched()) + Ok(Reader::at(self, self.image()?.head)?.roots_touched()) } /// What a semantic query must satisfy before anything is embedded: the @@ -655,8 +656,8 @@ impl Mind { /// ids this returns. pub fn members(&self, selection: &Selection, semantic: &SemanticQuery) -> Result { let selection = Self::semantic_selection(selection, semantic)?; - let head = receipt::head(self)?; - let rows = Reader::at(self, head)?.rows()?; + let head = self.image()?.head; + let mut reader = Reader::at(self, head)?; // The walk is built here, field by field, from the caller's filters // alone and never spread from the caller's selection: the page size // is ours (the caller's `limit` shapes an answer, not the walk's cost: @@ -675,9 +676,11 @@ impl Mind { let mut ids = Vec::new(); let mut pages = 0; loop { + let universe = reader.universe(&walk); + let rows = reader.rows(&universe); let evaluation = select(&Vocabulary, &rows, &walk, head, self.cursor_key())?; pages += 1; - ids.extend(evaluation.rows.iter().map(ref_of)); + ids.extend(evaluation.rows.iter().map(|row| ref_of(row))); match evaluation.next_cursor { Some(next) => walk.cursor = Some(next), None => return Ok(Members { as_of: head, ids, pages }), @@ -713,18 +716,19 @@ impl Mind { ) -> Result { let selection = Self::semantic_selection(selection, semantic)?; let as_of = members.as_of; - let rows = Reader::at(self, as_of)?.rows()?; - - let held: BTreeMap<(PipelineKind, &str), i64> = - rows.iter().map(|row| ((row.kind, row.key.as_str()), row.ordinal())).collect(); + let mut reader = Reader::at(self, as_of)?; let member: BTreeSet<(PipelineKind, &str)> = members.ids.iter().map(|id| (id.kind, id.id.0.as_str())).collect(); let mut candidates: Vec<(&str, f32, i64)> = Vec::new(); let mut dropped = Vec::new(); for (hit, score) in hits { let id = (hit.kind, hit.id.0.as_str()); - match held.get(&id) { - Some(&ordinal) if member.contains(&id) => candidates.push((id.1, *score, ordinal)), + let ordinal = match reader.id_of(hit.kind, id.1) { + Some(row) if reader.prepare(row) => Some(reader.row(row).ordinal as i64), + _ => None, + }; + match ordinal { + Some(ordinal) if member.contains(&id) => candidates.push((id.1, *score, ordinal)), _ => dropped.push(hit.clone()), } } @@ -744,16 +748,18 @@ impl Mind { Evaluation { rows: Vec::new(), matched: 0, edges: Vec::new(), next_cursor: None } } else { let narrowed = Selection { keys: Some(keys), limit: Some(LIMIT_MAX), ..selection.clone() }; + let ids = reader.universe(&narrowed); + let rows = reader.rows(&ids); select(&Vocabulary, &rows, &narrowed, as_of, self.cursor_key())? }; let score = |row: &SelectionRow| scores.get(row.key.as_str()).copied().unwrap_or(f32::NEG_INFINITY); - evaluation.rows.sort_by(|a, b| score(b).total_cmp(&score(a)).then(a.ordinal().cmp(&b.ordinal()))); + evaluation.rows.sort_by(|a, b| score(b).total_cmp(&score(a)).then(a.ordinal.cmp(&b.ordinal))); evaluation.rows.truncate(limit); evaluation.next_cursor = None; - let row_scores: Vec = evaluation.rows.iter().map(score).collect(); + let row_scores: Vec = evaluation.rows.iter().map(|row| score(row)).collect(); let position = |kind: PipelineKind, key: &str| evaluation.rows.iter().position(|row| row.kind == kind && row.key == key); - let mut kept: Vec<(usize, EdgeMatch)> = std::mem::take(&mut evaluation.edges) + let mut kept: Vec<(usize, EdgeMatch<&SelectionRow>)> = std::mem::take(&mut evaluation.edges) .into_iter() .filter_map(|edge| { let owner = match edge.anchor { @@ -771,9 +777,9 @@ 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, + evaluation: Evaluation<&SelectionRow>, as_of: u64, scores: Option>, ) -> Result { @@ -788,7 +794,7 @@ impl Mind { let role = CitationRole::from_name(&edge.role).ok_or_else(|| MindRefusal::Unavailable { detail: format!("the evaluator returned an edge in role {:?}, which is not a citation role", edge.role), })?; - typed.push(PipelineEdge { from: ref_of(&edge.from), role, to: ref_of(&edge.to) }); + typed.push(PipelineEdge { from: ref_of(edge.from), role, to: ref_of(edge.to) }); } Some(typed) } else { @@ -802,8 +808,15 @@ impl Mind { mod tests { use super::*; use crate::fixtures::*; - use crate::receipt::Faculty; + use crate::receipt::{self, Faculty}; use cultnet_rs::{Citation, FieldPredicate, Incoming, RecordRef}; + + /// Every row the mind holds at `as_of`, as the substrate is handed them. + fn all_rows(mind: &Mind, as_of: u64) -> Vec { + let mut reader = Reader::at(mind, as_of).unwrap(); + let ids = reader.universe(&Selection::default()); + reader.rows(&ids).into_iter().cloned().collect() + } use crate::store::test_stores::MemoryStore; use chrono::{TimeZone, Utc}; use eureka_pipeline::{ @@ -1339,7 +1352,7 @@ mod tests { committed(admit(&mut mind, vec![question_n(1), question_n(2)])); let selection = of_kinds(&[K::Question]); let head = 2; - let rows = Reader::at(&mind, head).unwrap().rows().unwrap(); + let rows = all_rows(&mind, head); let at = |as_of: u64| Selection { cursor: Some(Cursor::mint(as_of, &rows[0], &selection, mind.cursor_key())), ..selection.clone() @@ -1393,7 +1406,7 @@ mod tests { #[test] fn every_emitted_closed_value_passes_the_door() { let mind = worked(); - let rows = Reader::at(&mind, 100).unwrap().rows().unwrap(); + let rows = all_rows(&mind, 100); for row in &rows { for alias in ["root", "in_force", "faculty", "repo", "cut", "severity", "confidence", "origin", "authority", "claim_outcome", "outcome"] { for value in cultnet_rs::Row::values(row, alias) { @@ -2022,6 +2035,44 @@ mod tests { assert_eq!((ids(&page), page.matched), (vec![shelf.r4.clone()], plain_matched(&shelf.mind, &standing(Selection::default()))), "top_k bounds the page too"); } + /// Three equal scores in every arrival order rank by ordinal alone, so a lookup that gives + /// every hit the same row (or the wrong one) is seen. + #[test] + fn equal_scores_rank_by_each_documents_own_ordinal_in_any_arrival_order() { + let shelf = shelf(); + let image = shelf.mind.image().unwrap(); + let named = [(K::Question, shelf.q2.clone()), (K::Ruling, shelf.r4.clone()), (K::Resolution, shelf.withdrawn_resolution.clone())]; + let ordinal = |(kind, key): &(K, String)| image.row(image.id_of(*kind, key).unwrap()).ordinal; + let mut expected: Vec<(K, String)> = named.to_vec(); + expected.sort_by_key(ordinal); + assert!(ordinal(&expected[0]) < ordinal(&expected[1]) && ordinal(&expected[1]) < ordinal(&expected[2]), "the fixture's ordinals are distinct"); + for order in [[0, 1, 2], [0, 2, 1], [1, 0, 2], [1, 2, 0], [2, 0, 1], [2, 1, 0]] { + let hits: Vec<_> = order.iter().map(|at| hit(named[*at].0, &named[*at].1, 0.5)).collect(); + let page = rank_all(&shelf.mind, &Selection::default(), &asking(10), &hits).page; + assert_eq!(ids(&page), expected.iter().map(|(_, key)| key.clone()).collect::>(), "arrival order {order:?}"); + // The cut is where the tie-break decides: the page keeps the earliest ordinals. + let keys: Vec = expected.iter().map(|(_, key)| key.clone()).collect(); + let limited = Selection { limit: Some(1), ..Selection::default() }; + assert_eq!(ids(&rank_all(&shelf.mind, &limited, &asking(10), &hits).page), keys[..1].to_vec(), "limit 1, arrival order {order:?}"); + assert_eq!(ids(&rank_all(&shelf.mind, &Selection::default(), &asking(2), &hits).page), keys[..2].to_vec(), "top_k 2, arrival order {order:?}"); + } + } + + /// Past the candidate cut the tie-break decides who is kept: equal scores arriving in reverse + /// keep the earliest ordinals, which a lookup that gives every hit one row does not. + #[test] + fn equal_scores_past_the_cut_keep_the_earliest_ordinals() { + let mut mind = seeded(); + let total = LIMIT_MAX + 3; + for n in 1..=total { + committed(admit(&mut mind, vec![question_n(n)])); + } + let key = |n: u32| id("question", &format!("Q{n}")); + let hits: Vec<_> = (1..=total).rev().map(|n| hit(K::Question, &key(n), 0.5)).collect(); + let page = rank_all(&mind, &Selection::default(), &asking(LIMIT_MAX), &hits).page; + assert_eq!(ids(&page), (1..=LIMIT_MAX).map(key).collect::>()); + } + /// Membership is the selection's alone: a kind, a key allowlist and a /// citation predicate each cut the candidates exactly as they cut a plain /// query, and the citation predicate sees the whole mind, not only the @@ -2533,8 +2584,8 @@ mod tests { let absent = mind(gate_rows()).read_back_through(|_, _| Ok(None)); assert!(matches!(&absent, Err(MindRefusal::Unavailable { detail }) if detail.contains("does not view back")), "{absent:?}"); let other = mind(gate_rows()).read_back_through(|reader, id| { - let first = reader.docs.image.iter().find(|held| held.kind == id.kind).unwrap(); - reader.view_of(first).map(Some) + let (first, _) = reader.image.of_kind(id.kind).next().unwrap(); + reader.view(&PipelineRef { kind: id.kind, id: Short(first.to_string()) }) }); assert!(matches!(&other, Err(MindRefusal::Unavailable { detail }) if detail.contains("does not view back")), "{other:?}"); } @@ -2648,30 +2699,18 @@ mod tests { /// close a campaign, so the image is built by hand. #[test] fn a_closed_campaign_answers_no_repos() { - let held = |kind: K, key: String, document: D| Held { kind, key, document }; let (open_key, closed_key, closing_key) = ("open:campaign:open".to_string(), "closed:campaign:closed".to_string(), "closed:resolution:campaign.n1".to_string()); let named = |slug: &str, repo_name: &str| { let D::Campaign(mut c) = campaign(&[repo_name]) else { panic!("a campaign") }; c.slug = slug.into(); D::Campaign(c) }; - let image = vec![ - held(K::Campaign, open_key.clone(), named("open", REPO)), - held(K::Campaign, closed_key.clone(), named("closed", OTHER_REPO)), - held(K::Resolution, closing_key.clone(), resolution(r(K::Campaign, &closed_key), withdrawn())), - ]; - let landed = |ordinal: u64| AdmissionFacts { - receipt_id: format!("receipt-{ordinal}"), - admitted_at: "2026-09-16T12:00:00Z".into(), - provenance: provenance(Faculty::Hands), - ordinal, - }; - let facts = AdmissionIndex(BTreeMap::from([ - ((K::Campaign.type_id().to_string(), open_key), landed(1)), - ((K::Campaign.type_id().to_string(), closed_key), landed(2)), - ((K::Resolution.type_id().to_string(), closing_key), landed(3)), - ])); - let roots = Reader { docs: Docs { image, batch: Vec::new() }, facts }.roots_touched(); + let image = Image::of_documents(vec![ + (K::Campaign, open_key.clone(), named("open", REPO)), + (K::Campaign, closed_key.clone(), named("closed", OTHER_REPO)), + (K::Resolution, closing_key.clone(), resolution(r(K::Campaign, &closed_key), withdrawn())), + ]); + let roots = Reader { docs: Docs::at(&image, image.head), image: &image, as_of: image.head, patched: HashMap::new() }.roots_touched(); let by_root: BTreeMap<&str, &RootTouched> = roots.iter().map(|root| (root.root.0.as_str(), root)).collect(); assert_eq!(by_root["open"].repos, [repo(REPO)]); assert!(by_root["closed"].repos.is_empty(), "{roots:?}"); diff --git a/crates/huginn-mind/src/read_tests.rs b/crates/huginn-mind/src/read_tests.rs new file mode 100644 index 0000000..14837df --- /dev/null +++ b/crates/huginn-mind/src/read_tests.rs @@ -0,0 +1,768 @@ +//! The held image against the oracle. The oracle derives every answer from the store alone, per +//! call, by the linear derivation the read side had before it held an image; these tests set the +//! two side by side over a history with withdrawal chains, a re-resolution, hand-offs, a run and +//! its closure, at every snapshot, across a restart, and across the commit paths that do not +//! land. They fail when a held answer differs from a fresh one, not when a particular structure +//! changes. + +use std::collections::BTreeSet; +use std::sync::{Arc, atomic::{AtomicBool, Ordering}}; + +use anyhow::{Result, anyhow}; +use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, DatabaseEntry, PushAllOptions}; +use cultnet_rs::{Citation, FieldPredicate, Incoming, RecordRef, Row, Selection}; +use eureka_pipeline::{ClaimOutcome, FindingConfidence, PipelineDocument as D, PipelineKind as K, RunTurn, pipeline_key}; + +use crate::docs::{CitationRole, kind_of_id}; +use crate::fixtures::*; +use crate::image::Image; +use crate::mind::{Mind, schema_cache}; +use crate::oracle::{oracle_ids, oracle_page, oracle_view}; +use crate::query::{AdmissionIndex, PipelineSelectionPage, Reader}; +use crate::receipt::{self, Faculty, HuginnCommitReceipt}; +use crate::refusal::MindRefusal; +use crate::rows::{Alias, STATUS_ALIASES, SelectionRow}; +use crate::store::MindStore; +use crate::store::test_stores::{MemoryStore, RefusingStore, SwapCommand}; + +fn answering(label: &str, question: &str) -> D { + let mut ruling = ruling(label); + ruling.answers = Some(s(question)); + ruling.choice = Some(l("A")); + D::Ruling(ruling) +} + +fn key_of(document: &D) -> String { + pipeline_key(document).unwrap() +} + +/// The batches of a history, each with the faculty that admits it. +fn history() -> Vec<(Faculty, Vec)> { + let spec = |cut: &str, revision: u32| id("cut_spec", &format!("cut-{cut}.r{revision}")); + let run_a = mind_run("a", RunTurn::SelfRun, &[r(K::CutSpec, &spec("1", 1))]); + let run_key = key_of(&run_a); + let mut back = match hand_off(OTHER_INSTANCE, INSTANCE, REPO, &[]) { + D::HandOff(back) => back, + _ => unreachable!(), + }; + back.handed_on = eureka_pipeline::Date("2026-09-17".into()); + vec![ + (Faculty::Hands, vec![ + question("Q1", &["A", "B"], "A"), + D::Ruling(ruling("R1")), + D::CutSpec(cut_spec("1", 1)), + D::CutSpec(cut_spec("9", 1)), + D::CutSpec(cut_spec("10", 1)), + follow_up("FU-1", r(K::Question, &id("question", "Q1"))), + ]), + (Faculty::Hands, vec![D::CutReport(cut_report("9", 1)), D::CutReport(cut_report("10", 1))]), + (Faculty::Hands, vec![ + verdict("9", 1, vec![claim(ClaimOutcome::Falsified, &[&id("finding", "cut-9.s1.F1")], Some("P1"), &["M1"])]), + D::Finding(finding("9", 1, "F1", FindingConfidence::Confirmed)), + ]), + // A resolution that sorts before its subject's superseder in the batch it arrives in. + (Faculty::Hands, vec![ + D::CutSpec(cut_spec("9", 2)), + resolution(r(K::CutSpec, &spec("9", 1)), superseded(&[r(K::CutSpec, &spec("9", 2))])), + ]), + // The withdrawal of a resolution, and its withdrawal: the spec closes and reopens. + (Faculty::Hands, vec![resolution(r(K::CutSpec, &spec("10", 1)), withdrawn())]), + (Faculty::Hands, vec![resolution(r(K::Resolution, &id("resolution", "cut_spec.cut-10.r1.n1")), withdrawn())]), + (Faculty::Soul, vec![D::Ruling(ruling("R2"))]), + (Faculty::Hands, vec![answering("R3", &id("question", "Q1"))]), + // A hand-off away and back: a stewardship closes, its closure is withdrawn, a new one opens. + (Faculty::Hands, vec![hand_off(INSTANCE, OTHER_INSTANCE, REPO, &[])]), + (Faculty::Hands, vec![D::HandOff(back)]), + // A run that claims a spec, then is withdrawn unrun. + (Faculty::Hands, vec![run_a]), + (Faculty::Hands, vec![resolution(r(K::Run, &run_key), withdrawn())]), + (Faculty::Life, vec![question_n(2)]), + ] +} + +fn apply(mind: &mut Mind, faculty: Faculty, documents: Vec) { + committed(admit_as(mind, faculty, documents)); +} + +fn history_mind() -> Mind { + let mut mind = seeded(); + for (faculty, documents) in history() { + apply(&mut mind, faculty, documents); + } + mind +} + +fn held_equals_fresh(mind: &Mind) { + let held = mind.image().unwrap(); + let fresh = mind.fresh_image().unwrap(); + assert!(*held == fresh, "the held image differs from a fresh fold at head {}", held.head); +} + +fn of_kinds(kinds: &[K]) -> Selection { + Selection { schemas: Some(kinds.iter().map(|kind| kind.type_id().to_string()).collect()), ..Selection::default() } +} + +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 with(mut selection: Selection, predicate: FieldPredicate) -> Selection { + selection.fields.get_or_insert_with(Vec::new).push(predicate); + selection +} + +fn all_rows(mind: &Mind, as_of: u64) -> Vec { + let mut reader = Reader::at(mind, as_of).unwrap(); + let ids = reader.universe(&Selection::default()); + reader.rows(&ids).into_iter().cloned().collect() +} + +/// Selections that between them touch every kind, every alias value the history emits, keys, both +/// hops on every role with both `exists`, combinations, the other order and the smallest page. +fn corpus(mind: &Mind) -> Vec { + let head = mind.image().unwrap().head; + let rows = all_rows(mind, head); + let mut out = vec![Selection::default()]; + for kind in K::ALL.iter().copied() { + out.push(of_kinds(&[kind])); + out.push(Selection { descending: true, limit: Some(1), ..of_kinds(&[kind]) }); + for alias in ["in_force", "root", "faculty"] { + for value in ["true", "false", CAMPAIGN, "Hands"] { + out.push(with(of_kinds(&[kind]), any_of(alias, &[value]))); + } + } + } + let (mut pairs, mut edges) = (BTreeSet::new(), BTreeSet::new()); + for row in &rows { + for alias in Alias::ALL { + for value in Row::values(row, alias.name()) { + pairs.insert((alias.name(), value)); + } + } + for (role, target, _) in Row::references(row) { + edges.insert((role, target)); + } + out.push(Selection { keys: Some(vec![row.key.clone()]), projection: "document".into(), ..Selection::default() }); + } + out.push(Selection { keys: Some(vec!["nobody:question:nothing".into()]), ..Selection::default() }); + for (alias, value) in &pairs { + out.push(with(Selection::default(), any_of(alias, &[value]))); + } + for (role, target) in &edges { + let schema = kind_of_id(target).map_or("epiphany.pipeline.question.v1".to_string(), |kind| kind.type_id().to_string()); + for role_filter in [Some(role.clone()), None] { + let citation = Citation { target: RecordRef::new(schema.clone(), target.clone()), role: role_filter }; + out.push(Selection { cites: Some(citation.clone()), ..Selection::default() }); + out.push(with(Selection { cites: Some(citation), ..Selection::default() }, any_of("in_force", &["true"]))); + } + } + for role in CitationRole::ALL { + for exists in [true, false] { + let incoming = Incoming { role: role.name().into(), exists }; + out.push(Selection { cited: Some(incoming.clone()), ..Selection::default() }); + out.push(with(Selection { cited: Some(incoming.clone()), ..Selection::default() }, any_of("in_force", &["true"]))); + out.push(Selection { cited: Some(incoming), descending: true, limit: Some(1), ..of_kinds(&[K::CutSpec, K::Question]) }); + } + } + if let Some((role, target)) = edges.iter().next() { + let citation = Citation { target: RecordRef::new(K::Question.type_id(), target.clone()), role: Some(role.clone()) }; + out.push(Selection { cites: Some(citation), cited: Some(Incoming { role: "subject".into(), exists: false }), ..Selection::default() }); + } + out +} + +/// One page, both ways, whole page or whole refusal. +fn agree(mind: &Mind, selection: &Selection, as_of: u64) -> Result { + let held = mind.query_at(selection, as_of); + let fresh = oracle_page(mind, selection, as_of); + assert_eq!(held, fresh, "as_of {as_of}: {selection:?}"); + held +} + +fn walk(mind: &Mind, selection: &Selection, as_of: u64) { + let mut selection = Selection { limit: Some(1), ..selection.clone() }; + let mut at = as_of; + for _ in 0..80 { + let Ok(page) = agree(mind, &selection, at) else { return }; + let Some(next) = page.next else { return }; + at = page.as_of; + selection.cursor = Some(next); + } + panic!("a limit-1 walk did not end"); +} + +#[test] +fn held_image_equals_a_fresh_fold_after_every_commit() { + let mut mind = seeded(); + held_equals_fresh(&mind); + for (faculty, documents) in history() { + apply(&mut mind, faculty, documents); + held_equals_fresh(&mind); + } + let image = mind.image().unwrap(); + assert!(image.status_changed.iter().any(|at| *at > 1), "the history changes a status after the row was written"); + assert_eq!(image.head, history().len() as u64 + 1); +} + +#[test] +fn every_read_agrees_with_the_oracle_at_every_snapshot() { + let mind = history_mind(); + let head = mind.image().unwrap().head; + let corpus = corpus(&mind); + assert!(corpus.len() > 300, "the corpus is {}", corpus.len()); + let mut differs_from_head = 0; + for as_of in 1..=head { + for selection in &corpus { + let _ = agree(&mind, selection, as_of); + } + let open = Selection { fields: Some(vec![any_of("in_force", &["true"])]), ..Selection::default() }; + if mind.query_at(&open, as_of).unwrap().matched != mind.query_at(&open, head).unwrap().matched { + differs_from_head += 1; + } + } + assert!(differs_from_head > 0, "a status reversed by a later withdrawal shows at an old snapshot"); + for kind in K::ALL.iter().copied() { + walk(&mind, &of_kinds(&[kind]), head); + walk(&mind, &of_kinds(&[kind]), head - 3); + } + walk(&mind, &Selection::default(), head); + walk(&mind, &Selection { descending: true, ..Selection::default() }, 4); +} + +#[test] +fn views_and_status_agree_with_the_oracle_at_every_snapshot() { + let mind = history_mind(); + let head = mind.image().unwrap().head; + let every = oracle_ids(&mind, head).unwrap(); + assert_eq!(every.len(), mind.image().unwrap().rows.len(), "the oracle names every document, or the loop below proves nothing"); + for as_of in 1..=head { + let reader = Reader::at(&mind, as_of).unwrap(); + for id in &every { + assert_eq!(reader.view(id).unwrap(), oracle_view(&mind, id, as_of).unwrap(), "{} at {as_of}", id.id.0); + } + } + let spec = r(K::CutSpec, &id("cut_spec", "cut-10.r1")); + assert_eq!(mind.view(&spec).unwrap().unwrap().status, crate::query::PipelineStatus::InForce, "the reopened spec stands at the head"); + let closed = Reader::at(&mind, 6).unwrap().view(&spec).unwrap().unwrap(); + assert!(matches!(closed.status, crate::query::PipelineStatus::Resolved { .. }), "and is closed while its withdrawal is the head"); + let reopened = Mind::open_with(mind.store().clone(), &slug(INSTANCE)).unwrap(); + assert_eq!(reopened.status(crate::wire::IndexStatus::Current), mind.status(crate::wire::IndexStatus::Current)); + assert_eq!(reopened.persona().unwrap(), mind.persona().unwrap()); +} + +#[test] +fn a_restart_answers_what_the_held_image_answered() { + let dir = tempfile::tempdir().unwrap(); + let instance = slug(INSTANCE); + let mut mind = Mind::open(dir.path(), &instance).unwrap(); + seed(&mut mind); + for (faculty, documents) in history() { + apply(&mut mind, faculty, documents); + } + let head = mind.image().unwrap().head; + let corpus = corpus(&mind); + // A cursor is minted under the process's own key, so the answers are compared without it. + let plain = |page: Result| page.map(|page| PipelineSelectionPage { next: None, ..page }); + let answers: Vec<_> = corpus.iter().map(|selection| plain(mind.query_at(selection, head))).collect(); + let cursor = mind.query(&Selection { limit: Some(1), ..of_kinds(&[K::Question]) }).unwrap().next.unwrap(); + let held = mind.image().unwrap().clone(); + drop(mind); + let reopened = Mind::open(dir.path(), &instance).unwrap(); + assert!(*reopened.image().unwrap() == held, "the image a restart folds is the image the holder held"); + for (selection, before) in corpus.iter().zip(&answers) { + assert_eq!(&plain(reopened.query_at(selection, head)), before, "{selection:?}"); + } + let stale = Selection { limit: Some(1), cursor: Some(cursor), ..of_kinds(&[K::Question]) }; + assert!(matches!(reopened.query(&stale), Err(MindRefusal::CursorInvalid { .. })), "a cursor does not survive a restart"); +} + +#[test] +fn a_lost_swap_rebuilds_from_what_the_store_holds() { + let store = MemoryStore::new(); + let mut holder = opened(store.clone(), INSTANCE); + seed(&mut holder); + let mut other = Mind::open_with(store.clone(), &slug(INSTANCE)).unwrap(); + committed(admit(&mut other, vec![question_n(7)])); + let planted = r(K::Question, &id("question", "Q7")); + assert!(holder.view(&planted).unwrap().is_none(), "without a lost swap the plant stays unseen until restart: the stated contract"); + assert_eq!(Mind::open_with(store.clone(), &slug(INSTANCE)).unwrap().view(&planted).unwrap().map(|v| v.id), Some(planted.clone())); + let outcome = admit(&mut holder, vec![question_n(7)]); + assert!(matches!(outcome, crate::admission::PipelineAdmissionOutcome::Conflict { .. }), "{outcome:?}"); + held_equals_fresh(&holder); + assert!(holder.view(&planted).unwrap().is_some(), "the lost swap looked at the store"); + committed(admit(&mut holder, vec![question_n(8)])); + held_equals_fresh(&holder); +} + +/// A store that lands the swap and then reports an error, so the holder cannot know. +#[derive(Clone)] +struct LandsThenFails { + inner: MemoryStore, + armed: Arc, +} + +impl CacheBackingStore for LandsThenFails { + fn pull_all(&self) -> Result> { + self.inner.pull_all() + } + fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + self.inner.push(entry) + } + fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + self.inner.delete(entry) + } + fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> Result<()> { + self.inner.push_all(entries, options) + } +} + +impl MindStore for LandsThenFails { + fn compare_and_swap_batch(&self, expected: &[CultCacheEnvelope], replacements: Vec) -> Result { + let landed = self.inner.compare_and_swap_batch(expected, replacements)?; + if self.armed.swap(false, Ordering::SeqCst) { + return Err(anyhow!("the store failed after landing")); + } + Ok(landed) + } +} + +#[test] +fn a_store_error_during_commit_rebuilds() { + let inner = MemoryStore::new(); + let armed = Arc::new(AtomicBool::new(false)); + let mut mind = opened(LandsThenFails { inner: inner.clone(), armed: armed.clone() }, INSTANCE); + seed(&mut mind); + armed.store(true, Ordering::SeqCst); + let outcome = admit(&mut mind, vec![question_n(1)]); + assert!(matches!(outcome, crate::admission::PipelineAdmissionOutcome::Refused(MindRefusal::Unavailable { .. })), "{outcome:?}"); + held_equals_fresh(&mind); + assert!(mind.view(&r(K::Question, &id("question", "Q1"))).unwrap().is_some(), "the batch that landed is held"); + committed(admit(&mut mind, vec![question_n(2)])); + held_equals_fresh(&mind); + + let refusing = RefusingStore::new(); + let mut mind = opened(refusing.clone(), INSTANCE); + seed(&mut mind); + refusing.command(SwapCommand::Fail); + let before = mind.image().unwrap().clone(); + assert!(matches!(admit(&mut mind, vec![question_n(1)]), crate::admission::PipelineAdmissionOutcome::Refused(_))); + held_equals_fresh(&mind); + assert!(*mind.image().unwrap() == before, "a store that did not land leaves the image as it was"); + refusing.command(SwapCommand::Delegate); + committed(admit(&mut mind, vec![question_n(1)])); + held_equals_fresh(&mind); +} + +/// The rows of a healthy two-step mind, to be broken one way at a time. +fn healthy_rows() -> Vec { + let mut mind = seeded(); + committed(admit(&mut mind, vec![question_n(1)])); + mind.store().rows() +} + +fn receipt_of(rows: &[CultCacheEnvelope], ordinal: u64) -> HuginnCommitReceipt { + rows.iter() + .filter(|row| row.r#type == HuginnCommitReceipt::TYPE) + .map(|row| rmp_serde::from_slice::(&row.payload).unwrap()) + .find(|receipt| receipt.ordinal == ordinal) + .unwrap() +} + +fn faulted_mind(rows: Vec) -> (Mind, MemoryStore) { + let store = MemoryStore::new(); + rows.into_iter().for_each(|row| store.plant(row)); + (Mind::open_with(store.clone(), &slug(INSTANCE)).unwrap(), store) +} + +#[test] +fn a_faulted_store_refuses_alike_at_every_snapshot() { + let cache = schema_cache().unwrap(); + let mut cases: Vec<(&str, Vec)> = Vec::new(); + + let mut rows = healthy_rows(); + let at = rows.iter().position(|row| row.key == id("question", "Q1")).unwrap(); + rows[at].payload = vec![0xc1]; + cases.push(("an undecodable document", rows)); + + let mut rows = healthy_rows(); + let at = rows.iter().position(|row| row.r#type == HuginnCommitReceipt::TYPE).unwrap(); + rows[at].payload = vec![0xc1]; + cases.push(("an undecodable receipt", rows)); + + let mut rows = healthy_rows(); + let gone = receipt_of(&rows, 1).receipt_id; + rows.retain(|row| row.key != gone); + cases.push(("an undense chain", rows)); + + let mut rows = healthy_rows(); + let orphan = prepare(&question_n(5)); + rows.push(orphan); + cases.push(("an orphan", rows)); + + let mut rows = healthy_rows(); + let mut twice = receipt_of(&rows, 2); + twice.ordinal = 3; + twice.receipt_id = "mind-commit-twice".into(); + rows.push(cache.prepare_entry_named(&twice.receipt_id, &twice).unwrap().0); + cases.push(("a document written by two receipts", rows)); + + for (name, rows) in cases { + let (mind, store) = faulted_mind(rows); + let before = store.rows(); + let probe = of_kinds(&[K::Question]); + let seen = mind.query_at(&probe, 1); + let fault = seen.clone().expect_err(name); + for as_of in [1, 2, 5] { + assert_eq!(mind.query_at(&probe, as_of), seen, "{name} at {as_of}"); + assert_eq!(oracle_page(&mind, &probe, as_of).expect_err(name), fault, "{name}: the oracle names the same fault"); + } + assert_eq!(mind.query(&probe), seen, "{name}: query answers the held fault"); + assert_eq!(mind.view(&r(K::Question, &id("question", "Q1"))), Err(fault.clone()), "{name}"); + let outcome = admit(&mut Mind::open_with(store.clone(), &slug(INSTANCE)).unwrap(), vec![question_n(9)]); + assert!(matches!(outcome, crate::admission::PipelineAdmissionOutcome::Refused(ref refusal) if *refusal == fault), "{name}: admit refuses with the fault, {outcome:?}"); + assert_eq!(store.rows(), before, "{name}: a refused admission writes nothing"); + let status = mind.status(crate::wire::IndexStatus::Current); + assert_eq!(status.persona, crate::wire::PersonaStatus::Absent, "{name}: status stays total"); + assert_eq!(mind.persona().unwrap(), None, "{name}"); + } +} + +#[test] +fn an_edge_outside_its_target_falls_back_to_the_whole_universe() { + let mut rows = healthy_rows(); + let cache = schema_cache().unwrap(); + // A run claims a question: the role claims cut specs, findings and follow-ups. Admission would + // refuse it, so it is planted with its receipt. + let planted = mind_run("planted", RunTurn::SelfRun, &[r(K::Question, &id("question", "Q1"))]); + let envelope = planted.prepare(&cache).unwrap(); + let mut receipt = receipt::candidate(&slug(INSTANCE), provenance(Faculty::Hands), &[], std::slice::from_ref(&envelope), 3, now()).unwrap(); + receipt.receipt_id = receipt.digest().unwrap(); + rows.push(cache.prepare_entry_named(&receipt.receipt_id, &receipt).unwrap().0); + rows.push(envelope); + let (mind, _) = faulted_mind(rows); + assert!(!mind.image().unwrap().edges_in_target, "the planted edge is outside its role's kinds"); + let healthy = seeded(); + assert!(healthy.image().unwrap().edges_in_target); + let refused = MindRefusal::from(cultnet_rs::SelectionRefusal::ReferenceOutsideTarget { + from_schema_id: K::Run.type_id().into(), + from_key: key_of(&planted), + role: "claims".into(), + to_schema_id: K::Question.type_id().into(), + to_key: id("question", "Q1"), + }); + let elsewhere = Citation { target: RecordRef::new(K::CutSpec.type_id(), id("cut_spec", "cut-1.r1")), role: Some("claims".into()) }; + for selection in [ + Selection { cited: Some(Incoming { role: "claims".into(), exists: true }), ..Selection::default() }, + Selection { cites: Some(elsewhere), ..Selection::default() }, + ] { + assert_eq!(agree(&mind, &selection, 3), Err(refused.clone()), "the substrate refuses the edge exactly as it would over the whole mind"); + } +} + +#[test] +fn status_aliases_are_the_declared_ones() { + let mut mind = seeded(); + let mut before = all_rows(&mind, 1); + let mut status_changes = 0; + for (faculty, documents) in history() { + apply(&mut mind, faculty, documents); + let after = all_rows(&mind, mind.image().unwrap().head); + for (old, new) in before.iter().zip(&after) { + if old.view.status != new.view.status { + status_changes += 1; + } + for alias in Alias::ALL { + if !STATUS_ALIASES.contains(&alias) { + assert_eq!(Row::values(old, alias.name()), Row::values(new, alias.name()), "{} changed under a status change", alias.name()); + } + } + } + before = after; + } + assert!(status_changes >= 3, "the history changes statuses ({status_changes})"); +} + +#[test] +fn read_back_refuses_a_held_image_that_differs_from_a_fresh_fold() { + let store = MemoryStore::new(); + let mut holder = opened(store.clone(), INSTANCE); + seed(&mut holder); + assert_eq!(holder.read_back().unwrap(), 4, "a clean store counts every document"); + let mut other = Mind::open_with(store, &slug(INSTANCE)).unwrap(); + committed(admit(&mut other, vec![question_n(1)])); + let refused = holder.read_back(); + assert!(matches!(&refused, Err(MindRefusal::Unavailable { detail }) if detail.contains("held image differs from a fresh fold")), "{refused:?}"); + assert_eq!(other.read_back().unwrap(), 5); +} + +#[test] +fn the_admission_index_names_what_receipts_wrote() { + let mind = history_mind(); + let index = AdmissionIndex::build(&mind.receipts().unwrap()).unwrap(); + let image: &Image = mind.image().unwrap(); + for row in &image.rows { + assert!(index.contains(row.kind.type_id(), &row.key)); + } +} + +/// A store whose swap can land then err, be lost, and whose next pull can fail: a look that fails +/// after a store fault. +#[derive(Clone)] +struct FailsTheLook { + inner: MemoryStore, + land_then_err: Arc, + lose: Arc, + pull_err: Arc, +} + +impl CacheBackingStore for FailsTheLook { + fn pull_all(&self) -> Result> { + if self.pull_err.swap(false, Ordering::SeqCst) { + return Err(anyhow!("the look failed")); + } + self.inner.pull_all() + } + fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + self.inner.push(entry) + } + fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + self.inner.delete(entry) + } + fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> Result<()> { + self.inner.push_all(entries, options) + } +} + +impl MindStore for FailsTheLook { + fn compare_and_swap_batch(&self, expected: &[CultCacheEnvelope], replacements: Vec) -> Result { + if self.lose.swap(false, Ordering::SeqCst) { + return Ok(false); + } + let landed = self.inner.compare_and_swap_batch(expected, replacements)?; + if self.land_then_err.swap(false, Ordering::SeqCst) { + return Err(anyhow!("the store failed after landing")); + } + Ok(landed) + } +} + +fn fails_the_look() -> (Mind, FailsTheLook) { + let store = FailsTheLook { inner: MemoryStore::new(), land_then_err: Default::default(), lose: Default::default(), pull_err: Default::default() }; + let mut mind = opened(store.clone(), INSTANCE); + seed(&mut mind); + (mind, store) +} + +/// One rule for a failed look: the fault is held, reads and admission refuse with it, nothing is +/// written, and a restart holds what the store holds and admits at the next ordinal. +fn assert_fault_is_held(mind: &mut Mind, store: &FailsTheLook, landed: &[&str]) { + assert!(matches!(mind.image(), Err(MindRefusal::Unavailable { .. }))); + assert!(matches!(mind.query(&of_kinds(&[K::Question])), Err(MindRefusal::Unavailable { .. }))); + let rows = store.inner.rows(); + let next = admit(mind, vec![question_n(50)]); + assert!(matches!(next, crate::admission::PipelineAdmissionOutcome::Refused(MindRefusal::Unavailable { .. })), "{next:?}"); + assert_eq!(store.inner.rows(), rows, "a held fault writes nothing"); + let mut reopened = opened(store.clone(), INSTANCE); + held_equals_fresh(&reopened); + assert_eq!(reopened.query(&of_kinds(&[K::Question])).unwrap().matched, landed.len() as u32); + committed(admit(&mut reopened, vec![question_n(51)])); + held_equals_fresh(&reopened); + reopened.read_back().unwrap(); +} + +#[test] +fn a_failed_look_after_a_landed_error_holds_the_fault() { + let (mut mind, store) = fails_the_look(); + store.land_then_err.store(true, Ordering::SeqCst); + store.pull_err.store(true, Ordering::SeqCst); + let first = admit(&mut mind, vec![question_n(1)]); + assert!(matches!(first, crate::admission::PipelineAdmissionOutcome::Refused(MindRefusal::Unavailable { .. })), "{first:?}"); + assert_fault_is_held(&mut mind, &store, &["Q1"]); +} + +#[test] +fn a_failed_look_after_a_lost_swap_holds_the_fault() { + let (mut mind, store) = fails_the_look(); + let mut other = opened(store.clone(), INSTANCE); + committed(admit(&mut other, vec![question_n(7)])); + store.lose.store(true, Ordering::SeqCst); + store.pull_err.store(true, Ordering::SeqCst); + let lost = admit(&mut mind, vec![question_n(8)]); + assert!(matches!(lost, crate::admission::PipelineAdmissionOutcome::Refused(MindRefusal::Unavailable { .. })), "{lost:?}"); + assert_fault_is_held(&mut mind, &store, &["Q7"]); +} + +// ---- the floor: what a read and a landed commit may cost, counted ---- + +/// Reads and a landed admission never pull the store: a pull is a decode of the whole mind. +#[test] +fn reads_and_landed_admissions_pull_the_store_zero_times() { + let mut mind = history_mind(); + let pulls = mind.store().pull_count(); + let head = mind.image().unwrap().head; + mind.query(&Selection::default()).unwrap(); + mind.query_at(&of_kinds(&[K::Question]), head / 2).unwrap(); + mind.view(&r(K::Question, &id("question", "Q1"))).unwrap(); + let _ = mind.status(crate::wire::IndexStatus::Current); + let _ = mind.persona(); + assert_eq!(mind.store().pull_count(), pulls, "a read pulled the store"); + committed(admit(&mut mind, vec![question_n(99)])); + committed(admit(&mut mind, vec![resolution(r(K::Question, &id("question", "Q99")), withdrawn())])); + assert_eq!(mind.store().pull_count(), pulls, "a landed swap pulled the store"); + held_equals_fresh(&mind); +} + +/// A lost swap and a store error each look at the store exactly once. +#[test] +fn a_lost_swap_and_a_store_error_pull_the_store_once_each() { + let (mut mind, store) = fails_the_look(); + let mut other = opened(store.clone(), INSTANCE); + committed(admit(&mut other, vec![question_n(7)])); + let pulls = store.inner.pull_count(); + store.lose.store(true, Ordering::SeqCst); + admit(&mut mind, vec![question_n(8)]); + assert_eq!(store.inner.pull_count(), pulls + 1, "a lost swap looks once"); + store.land_then_err.store(true, Ordering::SeqCst); + admit(&mut mind, vec![question_n(9)]); + assert_eq!(store.inner.pull_count(), pulls + 2, "a store error looks once"); +} + +/// A history of many questions, one of them the only Soul's, one follow-up citing Q1. +fn many_rows_mind() -> Mind { + let mut mind = seeded(); + for chunk in (1..=40).collect::>().chunks(10) { + committed(admit(&mut mind, chunk.iter().map(|n| question_n(*n)).collect())); + } + committed(admit_as(&mut mind, Faculty::Soul, vec![question_n(41)])); + committed(admit(&mut mind, vec![follow_up("FU-1", r(K::Question, &id("question", "Q1")))])); + mind +} + +/// The universe is the narrowing: a selection that names a key, a rare field value or a cited +/// target hands the match only its candidates, never the mind. +#[test] +fn a_selection_reads_its_candidates_and_not_the_mind() { + let mind = many_rows_mind(); + let image = mind.image().unwrap(); + let (head, total) = (image.head, image.rows.len()); + assert!(total > 40); + assert_eq!(image.universe(&Selection::default(), head).len(), total, "the unfiltered selection is the whole mind"); + let keyed = Selection { keys: Some(vec![id("question", "Q3")]), ..Selection::default() }; + assert_eq!(image.universe(&keyed, head).len(), 1, "keys"); + let rare = with(of_kinds(&[K::Question]), any_of("faculty", &["Soul"])); + assert_eq!(image.universe(&rare, head).len(), 1, "a rare field value"); + assert_eq!(image.universe(&of_kinds(&[K::Question]), head).len(), 41, "a kind is its own rows"); + let citation = Citation { target: RecordRef::new(K::Question.type_id(), id("question", "Q1")), role: None }; + let citers = Selection { cites: Some(citation), ..Selection::default() }; + let universe = image.universe(&citers, head); + assert!(universe.len() <= 3 && !universe.is_empty(), "a hop starts from the target's citers: {}", universe.len()); + assert!(total > 10 * universe.len()); + assert!(mind.query(&citers).unwrap().matched >= 1); +} + +// ---- the narrowing lines the oracle corpus must reach ---- + +/// Both hops with the citer's schema excluding the target's: the target row is handed to the +/// match only by the line that adds the cites target to a both-hop universe. +#[test] +fn both_hops_with_a_schema_excluding_the_target_agree() { + let mind = history_mind(); + let head = mind.image().unwrap().head; + let rows = all_rows(&mind, head); + let mut edges = BTreeSet::new(); + for row in &rows { + for (role, target, _) in Row::references(row) { + edges.insert((row.kind, role, target)); + } + } + let (mut pages, mut hits) = (0, 0); + for (carrier, role, target) in &edges { + let schema = kind_of_id(target).map_or(K::Question.type_id().to_string(), |kind| kind.type_id().to_string()); + let citation = Citation { target: RecordRef::new(schema, target.clone()), role: Some(role.clone()) }; + for exists in [true, false] { + let incoming = Incoming { role: "subject".into(), exists }; + let plain = Selection { cites: Some(citation.clone()), cited: Some(incoming.clone()), ..of_kinds(&[*carrier]) }; + let narrowed = with(Selection { projection: "document".into(), ..plain.clone() }, any_of("in_force", &["true"])); + for as_of in [head, head / 2] { + for selection in [&plain, &narrowed] { + if let Ok(page) = agree(&mind, selection, as_of) { + pages += 1; + hits += usize::from(page.matched > 0); + } + } + } + } + } + assert!(pages > 0 && hits > 0, "no both-hop selection with a schema matched anything ({pages} pages)"); +} + +/// A resolution withdrawn by another: the Resolution schema excludes the cut spec it cites. +#[test] +fn a_both_hop_selection_finds_a_citer_whose_target_the_schema_excludes() { + let mind = history_mind(); + let selection = Selection { + cites: Some(Citation { target: RecordRef::new(K::CutSpec.type_id(), id("cut_spec", "cut-10.r1")), role: None }), + cited: Some(Incoming { role: "subject".into(), exists: true }), + ..of_kinds(&[K::Resolution]) + }; + let head = mind.image().unwrap().head; + let page = agree(&mind, &selection, head).unwrap(); + assert_eq!(page.matched, 1, "{page:?}"); +} + +fn planted_rows(rows: &mut Vec, ordinal: u64, document: &D) { + let cache = schema_cache().unwrap(); + let envelope = document.prepare(&cache).unwrap(); + let mut receipt = receipt::candidate(&slug(INSTANCE), provenance(Faculty::Hands), &[], std::slice::from_ref(&envelope), ordinal, now()).unwrap(); + receipt.receipt_id = receipt.digest().unwrap(); + rows.push(cache.prepare_entry_named(&receipt.receipt_id, &receipt).unwrap().0); + rows.push(envelope); +} + +/// Two closures in force over one subject, the later sequence arriving first. Admission refuses +/// this; a store that holds it must still answer as the oracle does, closers in key order. +#[test] +fn planted_double_closers_keep_key_order() { + let q = r(K::Question, &id("question", "Q1")); + let mut rows = healthy_rows(); + planted_rows(&mut rows, 3, &resolution_n(q.clone(), 2, withdrawn())); + planted_rows(&mut rows, 4, &resolution_n(q.clone(), 1, withdrawn())); + let (mind, _) = faulted_mind(rows); + for as_of in 1..=4 { + assert_eq!(Reader::at(&mind, as_of).unwrap().view(&q).unwrap(), oracle_view(&mind, &q, as_of).unwrap(), "at {as_of}"); + } +} + +// ---- Hands' own gaps ---- + +/// A lost swap names the identities that moved beneath it: a written key already stored, a read +/// document that changed, and not a read document that is the same. +#[test] +fn a_conflict_names_the_identities_that_changed() { + let (mut mind, store) = fails_the_look(); + let mut other = opened(store.clone(), INSTANCE); + committed(admit(&mut other, vec![question_n(7)])); + store.lose.store(true, Ordering::SeqCst); + let outcome = admit(&mut mind, vec![question_n(7)]); + let crate::admission::PipelineAdmissionOutcome::Conflict { identities } = outcome else { panic!("{outcome:?}") }; + assert_eq!(identities, vec![r(K::Question, &id("question", "Q7"))], "the written key that was already stored"); + + let cited = |mind: &mut Mind| { + let mut spec = cut_spec("1", 1); + spec.rulings = vec![s(&id("ruling", "R1"))]; + admit(mind, vec![D::CutSpec(spec)]) + }; + committed(admit(&mut mind, vec![D::Ruling(ruling("R1"))])); + store.lose.store(true, Ordering::SeqCst); + let outcome = cited(&mut mind); + let crate::admission::PipelineAdmissionOutcome::Conflict { identities } = outcome else { panic!("{outcome:?}") }; + assert!(identities.is_empty(), "a read document that did not change is not named: {identities:?}"); + + let mut bent = ruling("R1"); + bent.ruling = "A ruling that moved beneath the holder.".into(); + store.inner.plant(prepare(&D::Ruling(bent))); + store.lose.store(true, Ordering::SeqCst); + let outcome = cited(&mut mind); + let crate::admission::PipelineAdmissionOutcome::Conflict { identities } = outcome else { panic!("{outcome:?}") }; + assert_eq!(identities, vec![r(K::Ruling, &id("ruling", "R1"))], "the read document that changed"); +} diff --git a/crates/huginn-mind/src/receipt.rs b/crates/huginn-mind/src/receipt.rs index 446786e..4201958 100644 --- a/crates/huginn-mind/src/receipt.rs +++ b/crates/huginn-mind/src/receipt.rs @@ -190,26 +190,13 @@ pub(crate) fn candidate( Ok(receipt) } -/// This mind's admission order: `N`, the count of receipts, when their -/// ordinals are exactly `{1..=N}`. Every reader of the ordinal asks here, -/// never by counting receipts itself: admission takes `head(self)? + 1` for -/// the next one, and `AdmissionIndex::build` (`query.rs`) calls it too, so -/// `query` and `view` carry the same density check rather than trusting a -/// stored ordinal unchecked. `Mind::receipts` is `pub(crate)`, so this crate -/// is the only place raw ordinals can be read at all; that keeps this the -/// one door, rather than something a doc comment merely asks callers to -/// respect. `ordinal` stays display-only on the read side until RS-3 makes -/// it the order. A chain that is not dense refuses rather than guessing, on -/// both sides alike. +/// This mind's admission order: `N`, the count of receipts, when their ordinals are exactly +/// `{1..=N}`, read from the held image, which checked the chain once when it was built and +/// extends it by one with each commit. Every reader of the ordinal asks here, never by counting +/// receipts itself: admission takes `head(self)? + 1` for the next one. A chain that is not dense +/// is a held fault that refuses on both sides alike. pub(crate) fn head(mind: &Mind) -> Result { - let mut ordinals = mind.receipts()?.into_iter().map(|receipt| receipt.ordinal).collect::>(); - ordinals.sort_unstable(); - let n = ordinals.len() as u64; - if ordinals.into_iter().eq(1..=n) { - Ok(n) - } else { - Err(MindRefusal::Unavailable { detail: "receipt ordinals are not 1..=N: found a duplicate or a gap".into() }) - } + Ok(mind.image()?.head) } /// The versions of a set of envelopes in identity order, so the digest does @@ -265,24 +252,34 @@ pub(crate) fn commit( .filter(|read| !writes.iter().any(|write| write.r#type == read.r#type && write.key == read.key)) .cloned(), ); - replacements.push(receipt_envelope); + replacements.push(receipt_envelope.clone()); let expected: &[CultCacheEnvelope] = &strong_reads; - let landed = MindStore::compare_and_swap_batch(mind.store(), expected, replacements).map_err(unavailable)?; - mind.refresh()?; - if !landed { - let changed = strong_reads - .iter() - .filter(|read| mind.raw_envelope(&read.r#type, &read.key) != Some(read)) - .chain(writes.iter().filter(|write| mind.raw_envelope(&write.r#type, &write.key).is_some())); - let identities = changed - .filter_map(|envelope| { - let kind = PipelineKind::ALL.iter().find(|kind| kind.type_id() == envelope.r#type)?; - Some(PipelineRef { kind: *kind, id: Short(envelope.key.clone()) }) - }) - .collect(); - return Ok(CommitOutcome::Conflict(identities)); + let landed = match MindStore::compare_and_swap_batch(mind.store(), expected, replacements) { + Ok(landed) => landed, + // Whether the batch landed is unknown, so the holder looks at the store before it answers + // anything else; a failed look holds the fault (Mind::refresh), so nothing answers or + // writes on a stale image. + Err(error) => { + let _ = mind.refresh(); + return Err(unavailable(error)); + } + }; + if landed { + mind.advance(&receipt, receipt_envelope, &writes); + return Ok(CommitOutcome::Committed(receipt)); } - Ok(CommitOutcome::Committed(receipt)) + mind.refresh()?; + let changed = strong_reads + .iter() + .filter(|read| mind.raw_envelope(&read.r#type, &read.key) != Some(read)) + .chain(writes.iter().filter(|write| mind.raw_envelope(&write.r#type, &write.key).is_some())); + let identities = changed + .filter_map(|envelope| { + let kind = PipelineKind::ALL.iter().find(|kind| kind.type_id() == envelope.r#type)?; + Some(PipelineRef { kind: *kind, id: Short(envelope.key.clone()) }) + }) + .collect(); + Ok(CommitOutcome::Conflict(identities)) } fn unique_identities(versions: &[DocumentVersion]) -> bool { diff --git a/crates/huginn-mind/src/rows.rs b/crates/huginn-mind/src/rows.rs index 49adeab..a871f65 100644 --- a/crates/huginn-mind/src/rows.rs +++ b/crates/huginn-mind/src/rows.rs @@ -17,7 +17,7 @@ use eureka_pipeline::{ RulingAuthority, RunOperator, RunTurn, Slug, }; -use crate::docs::{CitationRole, Held, kind_of_id, kind_of_type}; +use crate::docs::{CitationRole, kind_of_id, kind_of_type}; use crate::query::PipelineDocumentView; use crate::refusal::MindRefusal; @@ -68,8 +68,14 @@ const MEMORY_ORIGINS: [&str; 4] = ["Room", "Direct", "Wake", "SelfRun"]; /// The one value a presence alias answers with; a row without it answers nothing. const PRESENT: [&str; 1] = ["true"]; +/// The aliases whose values follow a row's status. Every other alias's values are +/// fixed when the row is written, so the held image may narrow by them at any +/// snapshot and by these only at the head. A test pins that no other alias +/// changes across a status change. +pub(crate) const STATUS_ALIASES: [Alias; 2] = [Alias::InForce, Alias::Ran]; + impl Alias { - const ALL: [Alias; 19] = [ + pub(crate) const ALL: [Alias; 19] = [ Self::Root, Self::InForce, Self::Faculty, @@ -91,7 +97,7 @@ impl Alias { Self::MemoryOrigin, ]; - fn name(self) -> &'static str { + pub(crate) fn name(self) -> &'static str { match self { Self::Root => "root", Self::InForce => "in_force", @@ -115,7 +121,7 @@ impl Alias { } } - fn from_name(name: &str) -> Option { + pub(crate) fn from_name(name: &str) -> Option { Self::ALL.into_iter().find(|alias| alias.name() == name) } @@ -193,11 +199,11 @@ fn outcome_name(outcome: &ResolutionOutcome) -> &'static str { /// One document as the evaluator sees it: its identity, its ordinal, what it /// answers for each alias, its citation edges, and the view a page projects. -#[derive(Clone, Debug)] +#[derive(Clone, Debug, PartialEq)] pub(crate) struct SelectionRow { pub(crate) kind: PipelineKind, pub(crate) key: String, - ordinal: u64, + pub(crate) ordinal: u64, values: BTreeMap<&'static str, Vec>, refs: Vec<(CitationRole, PipelineRef)>, pub(crate) view: PipelineDocumentView, @@ -205,24 +211,28 @@ pub(crate) struct SelectionRow { impl SelectionRow { /// `base` is the first non-resolution the document reaches through its - /// subjects: the row a resolution's `repo` and `cut` are read from. + /// subjects: the row a resolution's `repo` and `cut` are read from; `None` + /// is a document that is its own base. The row is the only decoded copy of + /// its document: it lives in `view`. pub(crate) fn new( - held: &Held, - base: &Held, + kind: PipelineKind, + key: &str, + base: Option<(&str, &PipelineDocument)>, view: PipelineDocumentView, refs: Vec<(CitationRole, PipelineRef)>, ) -> Self { use PipelineDocument as D; + let (base_key, base_document) = base.unwrap_or((key, &view.document)); let mut values: BTreeMap<&'static str, Vec> = BTreeMap::new(); let mut put = |alias: Alias, found: Vec| { values.insert(alias.name(), found); }; - put(Alias::Root, vec![root_and_local(&held.key).0.to_string()]); + put(Alias::Root, vec![root_and_local(key).0.to_string()]); put(Alias::InForce, vec![matches!(view.status, crate::query::PipelineStatus::InForce).to_string()]); put(Alias::Faculty, vec![format!("{:?}", view.admission.provenance.faculty)]); put( Alias::Repo, - match &base.document { + match base_document { D::Campaign(campaign) => campaign.repos.iter().map(OrgRepo::identity).collect(), D::CutSpec(spec) => vec![spec.repo.identity()], D::CutReport(report) => vec![report.repo.identity()], @@ -236,7 +246,7 @@ impl SelectionRow { ); put( Alias::Cut, - root_and_local(&base.key) + root_and_local(base_key) .1 .strip_prefix("cut-") .and_then(|rest| rest.split_once('.')) @@ -244,7 +254,7 @@ impl SelectionRow { .into_iter() .collect(), ); - match &held.document { + match &view.document { D::Finding(finding) => { put(Alias::Severity, vec![format!("{:?}", finding.severity)]); put(Alias::Confidence, vec![format!("{:?}", finding.confidence)]); @@ -277,7 +287,12 @@ impl SelectionRow { } _ => {} } - Self { kind: held.kind, key: held.key.clone(), ordinal: view.admission.ordinal, values, refs, view } + Self { kind, key: key.to_string(), ordinal: view.admission.ordinal, values, refs, view } + } + + /// Each alias this row answers, with its values: what the held image indexes. + pub(crate) fn alias_values(&self) -> impl Iterator { + self.values.iter().map(|(alias, values)| (*alias, values.as_slice())) } } diff --git a/crates/huginn-mind/src/wire.rs b/crates/huginn-mind/src/wire.rs index 4f438ab..309bb6e 100644 --- a/crates/huginn-mind/src/wire.rs +++ b/crates/huginn-mind/src/wire.rs @@ -370,15 +370,9 @@ impl Mind { /// `whoami`, derived from the image every time it is asked, and from the /// index status the caller holds: the mind stores neither. pub fn status(&self, index: IndexStatus) -> MindStatus { - let mut documents = 0_u32; - let mut receipts = 0_u32; - for envelope in self.envelopes() { - if envelope.r#type == HuginnCommitReceipt::TYPE { - receipts += 1; - } else if PipelineKind::ALL.iter().any(|kind| kind.type_id() == envelope.r#type) { - documents += 1; - } - } + let count = |type_id: &str| self.envelopes.get(type_id).map_or(0, |keyed| keyed.len() as u32); + let receipts = count(HuginnCommitReceipt::TYPE); + let documents = PipelineKind::ALL.iter().map(|kind| count(kind.type_id())).sum(); MindStatus { instance: self.instance().clone(), schema_epoch: PIPELINE_SCHEMA_EPOCH.to_string(), diff --git a/crates/huginn-mind/tests/read_floor.rs b/crates/huginn-mind/tests/read_floor.rs new file mode 100644 index 0000000..b030376 --- /dev/null +++ b/crates/huginn-mind/tests/read_floor.rs @@ -0,0 +1,172 @@ +//! 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`; `FLOOR_SAMPLES` (default 7) is how many times each read is timed, for a base too slow to sample 7 times. +//! +//! 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 { + let samples: usize = env("FLOOR_SAMPLES", "7").parse().unwrap(); + median( + (0..samples) + .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); +}