diff --git a/crates/huginn-mind/src/mind.rs b/crates/huginn-mind/src/mind.rs index 660a813..5df8726 100644 --- a/crates/huginn-mind/src/mind.rs +++ b/crates/huginn-mind/src/mind.rs @@ -317,10 +317,21 @@ impl Mind { /// 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.envelopes = envelopes_of(self.store.pull_all().map_err(unavailable)?); - (self.image, self.persona_receipt) = fold(&self.envelopes, &self.instance); - 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 @@ -349,11 +360,9 @@ impl Mind { Err(_) => faulted = true, } } - if (faulted || image.advance(&AdmissionFacts::of(receipt), documents).is_err()) - && let Err(fault) = self.refresh() - { - self.image = Err(fault.clone()); - self.persona_receipt = Err(fault); + if faulted || image.advance(&AdmissionFacts::of(receipt), documents).is_err() { + // A failed look holds its own fault. + let _ = self.refresh(); } } } diff --git a/crates/huginn-mind/src/read_tests.rs b/crates/huginn-mind/src/read_tests.rs index ef3605b..14837df 100644 --- a/crates/huginn-mind/src/read_tests.rs +++ b/crates/huginn-mind/src/read_tests.rs @@ -234,6 +234,7 @@ 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 { @@ -503,3 +504,265 @@ fn the_admission_index_names_what_receipts_wrote() { 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 a1f0fff..4201958 100644 --- a/crates/huginn-mind/src/receipt.rs +++ b/crates/huginn-mind/src/receipt.rs @@ -257,7 +257,8 @@ pub(crate) fn commit( 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 leaves the next lost swap or restart to heal it. + // 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));