diff --git a/crates/nexum-sdk-test/src/lib.rs b/crates/nexum-sdk-test/src/lib.rs index 1da359e2..7dfa2255 100644 --- a/crates/nexum-sdk-test/src/lib.rs +++ b/crates/nexum-sdk-test/src/lib.rs @@ -803,6 +803,124 @@ impl LocalStoreHost for MockLocalStore { } } +// ---------------------------------------------------------------- trap store + +/// Trap-injection wrapper over a [`LocalStoreHost`]: counts `set` and +/// `delete` calls and, once armed, simulates a guest trap mid-flow. +/// [`arm_after`](Self::arm_after)`(n)` lets the next `n` writes land; +/// the write after that trips the trap, and from then on every +/// operation - reads included - faults until +/// [`disarm`](Self::disarm), because nothing past a trap executes. +/// Sweeping `n` over a flow's write count drives the store through +/// every torn prefix a trap can strand, so a recovery pass can be +/// held to convergence from each one. +/// +/// Review rule this harness enforces (#609): no in-store invariant +/// may span two `set` calls unless the intermediate state is +/// self-healing or the writes ride the atomic `apply` batch verb. +pub struct TrapStore { + inner: H, + /// Write calls the trap let through. + writes: Cell, + /// Writes still allowed before the trap trips; `None` when unarmed. + remaining: Cell>, + tripped: Cell, +} + +impl TrapStore { + /// Wrap `inner`, unarmed: every operation delegates, writes are + /// counted. + pub fn new(inner: H) -> Self { + Self { + inner, + writes: Cell::new(0), + remaining: Cell::new(None), + tripped: Cell::new(false), + } + } + + /// Arm the trap: the next `n` writes land, the one after trips it. + pub fn arm_after(&self, n: u64) { + self.remaining.set(Some(n)); + self.tripped.set(false); + } + + /// Clear the trap and the tripped state; operations resume. The + /// write count keeps accumulating. + pub fn disarm(&self) { + self.remaining.set(None); + self.tripped.set(false); + } + + /// `set`/`delete` calls the trap let through since construction. + pub fn writes(&self) -> u64 { + self.writes.get() + } + + /// Whether the trap has fired. + pub fn tripped(&self) -> bool { + self.tripped.get() + } + + /// The wrapped store. + pub fn inner(&self) -> &H { + &self.inner + } + + /// Fault unless still executing: past the trap nothing runs. + fn read_gate(&self) -> Result<(), Fault> { + if self.tripped.get() { + return Err(Fault::Internal("TrapStore: trapped".into())); + } + Ok(()) + } + + /// Spend one write from the armed budget, tripping at zero. + fn write_gate(&self) -> Result<(), Fault> { + self.read_gate()?; + if let Some(remaining) = self.remaining.get() { + if remaining == 0 { + self.tripped.set(true); + return Err(Fault::Internal("TrapStore: trapped".into())); + } + self.remaining.set(Some(remaining - 1)); + } + self.writes.set(self.writes.get() + 1); + Ok(()) + } +} + +impl LocalStoreHost for TrapStore { + fn get(&self, key: &str) -> Result>, Fault> { + self.read_gate()?; + self.inner.get(key) + } + fn set(&self, key: &str, value: &[u8]) -> Result<(), Fault> { + self.write_gate()?; + self.inner.set(key, value) + } + fn delete(&self, key: &str) -> Result<(), Fault> { + self.write_gate()?; + self.inner.delete(key) + } + fn list_keys(&self, prefix: &str) -> Result, Fault> { + self.read_gate()?; + self.inner.list_keys(prefix) + } + fn contains(&self, key: &str) -> Result { + self.read_gate()?; + self.inner.contains(key) + } + fn len(&self, key: &str) -> Result, Fault> { + self.read_gate()?; + self.inner.len(key) + } + fn count(&self, prefix: &str) -> Result { + self.read_gate()?; + self.inner.count(prefix) + } +} + // ---------------------------------------------------------------- logging /// One recorded log line. @@ -1203,6 +1321,59 @@ mod tests { assert!(store.list_keys("bad:").is_err()); } + #[test] + fn trap_store_counts_writes_unarmed() { + let store = TrapStore::new(MockLocalStore::default()); + store.set("a", b"1").unwrap(); + store.set("b", b"2").unwrap(); + store.delete("a").unwrap(); + assert_eq!(store.writes(), 3); + assert!(!store.tripped()); + assert_eq!(store.get("b").unwrap().as_deref(), Some(&b"2"[..])); + } + + #[test] + fn trap_store_trips_after_the_armed_budget() { + let store = TrapStore::new(MockLocalStore::default()); + store.arm_after(2); + store.set("a", b"1").unwrap(); + store.delete("a").unwrap(); + // The third write trips; the row never lands. + assert!(store.set("b", b"2").is_err()); + assert!(store.tripped()); + assert_eq!(store.writes(), 2); + assert!(store.inner().get("b").unwrap().is_none()); + } + + #[test] + fn trap_store_faults_every_operation_once_tripped() { + let store = TrapStore::new(MockLocalStore::default()); + store.set("a", b"1").unwrap(); + store.arm_after(0); + assert!(store.set("b", b"2").is_err()); + // Nothing past a trap executes, reads included. + assert!(store.get("a").is_err()); + assert!(store.list_keys("").is_err()); + assert!(store.contains("a").is_err()); + assert!(store.delete("a").is_err()); + } + + #[test] + fn trap_store_disarm_resumes_over_the_surviving_rows() { + let store = TrapStore::new(MockLocalStore::default()); + store.set("a", b"1").unwrap(); + store.arm_after(0); + assert!(store.set("b", b"2").is_err()); + + store.disarm(); + assert!(!store.tripped()); + // The torn prefix survives: `a` landed, `b` never did. + assert_eq!(store.get("a").unwrap().as_deref(), Some(&b"1"[..])); + assert!(store.get("b").unwrap().is_none()); + store.set("b", b"2").unwrap(); + assert_eq!(store.writes(), 2); + } + #[test] fn local_store_max_entries_enforced() { let store = MockLocalStore::default(); diff --git a/crates/videre-sdk/src/keeper.rs b/crates/videre-sdk/src/keeper.rs index 084d2959..621d80c3 100644 --- a/crates/videre-sdk/src/keeper.rs +++ b/crates/videre-sdk/src/keeper.rs @@ -362,7 +362,7 @@ mod tests { use nexum_sdk::host::{Fault, LocalStoreHost as _}; use nexum_sdk::keeper::{Gates, Journal, Mark, Tick, WatchRef, WatchSet}; use nexum_sdk::prelude::{Address, B256, hex, keccak256}; - use nexum_sdk_test::MockLocalStore; + use nexum_sdk_test::{MockLocalStore, TrapStore}; use super::{Keeper, Outcome, RunReport, submission_key}; use crate::client::{VenueId, VenueTransport}; @@ -1069,4 +1069,85 @@ mod tests { assert_eq!(mark(&host, b"order"), Some(Mark::Reserved)); assert_ne!(mark(&host, b"order"), None); } + + // ---- #609 trap-injection: convergence from every torn prefix ---- + // + // Review rule the sweep enforces: no in-store invariant may span + // two `set` calls unless the intermediate state is self-healing or + // the writes ride the atomic `apply` batch verb (#609). The + // reserve/commit journal passes because each of its intermediate + // states - nothing, RESERVED, COMMITTED-sans-marker-clear - is one + // the next sweep's reconcile pass resolves on its own. + + /// Seed one watch on a trap store and pair it with an accepting + /// venue and a submitting keeper. + fn trap_rig( + venue: &CountingVenue, + ) -> ( + TrapStore, + Keeper, + ) { + let host = TrapStore::new(MockLocalStore::default()); + WatchSet::new(&host) + .put(&Address::ZERO, &B256::ZERO, b"params") + .expect("watch writes"); + let keeper = Keeper::new(StubSource(Outcome::Submit(b"order".to_vec())), venue, STUB); + (host, keeper) + } + + /// Writes one accepted submit tick performs, pinned by a dry run: + /// the reserve set, the commit set, and the refusal-marker delete. + fn accepted_tick_writes() -> u64 { + let venue = CountingVenue::accepting(); + let (host, keeper) = trap_rig(&venue); + let seeded = host.writes(); + run(keeper.run(&host, &TICK)).expect("dry run completes"); + assert_eq!(mark(&host, b"order"), Some(Mark::Committed)); + host.writes() - seeded + } + + #[test] + fn trap_at_every_write_prefix_reconciles_to_exactly_one_held_order() { + let total = accepted_tick_writes(); + assert_eq!(total, 3, "reserve set, commit set, refusal-marker delete"); + + // Trap the tick after each n of its writes: every torn prefix + // from nothing-landed through all-but-the-last. + for n in 0..total { + let venue = CountingVenue::accepting(); + let (host, keeper) = trap_rig(&venue); + host.arm_after(n); + let _ = run(keeper.run(&host, &TICK)); + assert!(host.tripped(), "prefix {n}: the trap must fire mid-tick"); + + // Restart from the torn store: the next sweep's reconcile + // pass plus the fresh-watch loop must converge. + host.disarm(); + run(keeper.run(&host, &TICK)).expect("recovery tick runs"); + assert_eq!( + mark(&host, b"order"), + Some(Mark::Committed), + "prefix {n}: the journal must end COMMITTED", + ); + assert_eq!( + venue.held_count(), + 1, + "prefix {n}: exactly one held order, whatever the POST count", + ); + assert!( + Journal::submitted(&host) + .pending() + .expect("journal reads") + .is_empty(), + "prefix {n}: no reservation may stay stranded", + ); + + // Steady state: a further tick is a pure idempotent skip. + let posts = venue.post_count(); + let report = run(keeper.run(&host, &TICK)).expect("steady tick runs"); + assert_eq!(report.duplicates, 1, "prefix {n}: COMMITTED skips"); + assert_eq!(venue.post_count(), posts, "prefix {n}: no further POST"); + assert_eq!(venue.held_count(), 1, "prefix {n}: still one held order"); + } + } }