From e9f78fc4e80334b8370d2ab258fda188492121c0 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 3 Nov 2026 07:45:23 +0100 Subject: [PATCH] read-floor: a hop-only cites starts from the target's citers; reads racing admissions on the wire answer their snapshot; the persona receipt is pinned held and across a restart; FLOOR_SAMPLES; header comments restated Co-Authored-By: Claude Opus 5.5 --- crates/huginn-daemon/src/serve.rs | 158 +++++++++++++++++++++++++ crates/huginn-mind/src/image.rs | 8 +- crates/huginn-mind/src/lib.rs | 7 +- crates/huginn-mind/src/persona.rs | 25 ++++ crates/huginn-mind/src/query.rs | 17 ++- crates/huginn-mind/tests/read_floor.rs | 5 +- 6 files changed, 206 insertions(+), 14 deletions(-) diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index 8ef3482..d980b18 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -792,6 +792,164 @@ 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 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, + } + } + 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/image.rs b/crates/huginn-mind/src/image.rs index 8bab7c4..5710926 100644 --- a/crates/huginn-mind/src/image.rs +++ b/crates/huginn-mind/src/image.rs @@ -276,7 +276,8 @@ impl Image { if hop && !self.edges_in_target { return (0..self.rows.len() as u32).collect(); } - let base = self.base_candidates(selection, as_of); + 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 @@ -291,8 +292,9 @@ impl Image { (None, None) => ids = base, (Some(cites), None) => { let key = cites.target.record_key.as_str(); - let citing: BTreeSet = citers_of(key, cites.role.as_deref()).into_iter().collect(); - ids.extend(base.intersection(&citing)); + // 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)) => { diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 179bfa3..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 diff --git a/crates/huginn-mind/src/persona.rs b/crates/huginn-mind/src/persona.rs index 3ae796d..ac3e83c 100644 --- a/crates/huginn-mind/src/persona.rs +++ b/crates/huginn-mind/src/persona.rs @@ -611,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 b7741ce..f906191 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,7 +16,8 @@ //! `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 @@ -503,8 +507,9 @@ 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 + /// 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, so every upgrade proves the held + /// state against the store. 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, id| reader.view(id)) diff --git a/crates/huginn-mind/tests/read_floor.rs b/crates/huginn-mind/tests/read_floor.rs index f324c68..1b248d0 100644 --- a/crates/huginn-mind/tests/read_floor.rs +++ b/crates/huginn-mind/tests/read_floor.rs @@ -6,7 +6,7 @@ //! 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`. +//! `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")] @@ -47,8 +47,9 @@ fn median(mut samples: Vec) -> Duration { } fn time(mut read: impl FnMut() -> T) -> Duration { + let samples: usize = env("FLOOR_SAMPLES", "7").parse().unwrap(); median( - (0..7) + (0..samples) .map(|_| { let started = Instant::now(); std::hint::black_box(read());