From 190b5b2c2f2e0274af02a8a9aa0c31c22a0775c9 Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Thu, 11 Jun 2026 23:22:15 -0700 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20live=20retention=20floor=20guard=20?= =?UTF-8?q?=E2=80=94=20close=20the=20axiom-5=20boundary=20for=20All-scope?= =?UTF-8?q?=20watches?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The hazard (machine-checked reachable in tests/model_live_watch.rs, unguarded variant): retention overrunning a LIVE consumer makes JetStream silently skip evicted messages — delete markers included — with no error anywhere: permanent silent fold divergence mid-watch, the same clamp class the resume-time check eliminated, previously covered only by the "retention >> lag" operating axiom. The model checker rejected the first guard design (periodic floor probe) with a counterexample: deliveries can catch the frontier up past the gap between probes, erasing the evidence. The shipped design is therefore in-band: a delivered revision that jumps the frontier by more than one is checked AT that delivery against first_sequence via the shared kernel (protocol::resume_window_ok) before the entry is processed — benign interior gaps pass, head eviction past the frontier fails the watch, and the restart routes into the verified resume -> CursorExpired -> resync repair path. A periodic probe backstops the no-traffic case only. Proof + conformance: - tests/model_live_watch.rs: guarded variant proves every maximal run ends with the fold equal to the bucket (trip witness earned, no phantom deletes); unguarded variant pins permanent silent divergence as the machine-checked record of the pre-guard code. - src/nats.rs floor_guard_tests: live-server twins — a clamped watch trips on its first gapped delivery before anything is folded; interior per-subject gaps pass with the dense tail delivered intact. - Axiom 5 narrowed accordingly: prefix-scoped watches (sparse by design, benign/hazardous eviction indistinguishable client-side) and the fresh full watch's history scan retain the operating requirement; the steady-state All-scope resume watch no longer relies on it. Co-Authored-By: Claude Fable 5 --- src/nats.rs | 289 +++++++++++++++++++++++++++++++++++++- tests/model.rs | 17 ++- tests/model_live_watch.rs | 265 ++++++++++++++++++++++++++++++++++ 3 files changed, 560 insertions(+), 11 deletions(-) create mode 100644 tests/model_live_watch.rs diff --git a/src/nats.rs b/src/nats.rs index 1c4eee4..e85f4e4 100644 --- a/src/nats.rs +++ b/src/nats.rs @@ -998,6 +998,126 @@ async fn stream_watch( Ok(()) } +/// Cadence of the floor guard's no-traffic backstop probe (one stream-info +/// RPC per interval per guarded watch). The PRIMARY detection is in-band — +/// the gapped-delivery check fires the moment evidence surfaces — so this +/// interval only bounds detection latency when NOTHING is being delivered; +/// it is not load-bearing for eventual detection. +const FLOOR_GUARD_INTERVAL: Duration = Duration::from_secs(30); + +/// [`stream_watch`] for the dense ALL-scope resume path, with the LIVE +/// retention floor guard (`tests/model_live_watch.rs` — the live twin of +/// [`NatsKvWatcher::check_resume_window`]). +/// +/// The hazard: retention overrunning a live consumer makes JetStream +/// silently skip evicted messages — delete markers included — with no error +/// anywhere (the same clamp behavior as resumes, mid-stream). Unguarded, +/// that is PERMANENT silent fold divergence; the model proves it reachable. +/// +/// Detection is primarily **in-band**: an unfiltered `ByStartSequence` +/// consumer sees every retained message, so a delivered revision that jumps +/// the frontier by more than one is evidence of eviction inside the gap. +/// The model checker REJECTED a periodic-only design with exactly the trace +/// this closes — deliveries can catch the frontier up past the gap between +/// probes, erasing the evidence — so the check runs AT the gapped delivery, +/// before the entry is processed: fetch `first_sequence` and apply the +/// shared kernel (`protocol::resume_window_ok`) to the frontier. A benign +/// gap (interior per-subject eviction with the floor still at or below the +/// frontier) passes; head eviction past the frontier fails the watch, and +/// the caller's restart routes into the verified resume → `CursorExpired` → +/// resync repair path. The periodic probe backstops the no-traffic case. +/// +/// Scope: sound only where density holds — the unfiltered resume watch. +/// Prefix-scoped watches deliver sparse revisions by design and cannot +/// distinguish benign from hazardous eviction client-side; they retain the +/// (narrowed) retention-outlives-lag operating axiom plus the resume-time +/// check on every restart (model axiom 5). +/// +/// The guarantee split, precisely: the SAFETY half — never folding past +/// unexamined evidence of loss — is unconditional in this loop (the gap +/// check precedes processing, and a stalled downstream stalls folding too). +/// The REPAIR half is conditional on the caller restarting the failed watch +/// (standard supervision; same posture as the resync fail-stop): a trip +/// with no restart is a loudly dead watch, never a silently wrong one. +async fn stream_watch_floor_guarded( + mut watcher: async_nats::jetstream::kv::Watch, + tx: &Sender, + resume_revision: u64, + js: &async_nats::jetstream::Context, + bucket: &str, +) -> Result<(), KvError> { + let stream_name = format!("KV_{bucket}"); + let first_sequence = || async { + let stream = timed(js.get_stream(&stream_name)) + .await? + .map_err(|e| KvError::OperationFailed(format!("floor guard stream lookup: {e}")))?; + Ok::(stream.cached_info().state.first_sequence) + }; + fn trip(frontier: u64, first: u64, bucket: &str) -> KvError { + warn!( + frontier, + first_sequence = first, + bucket, + "stream retention overran this live watch; failing so the restart can resync \ + (messages in the gap were evicted unseen)" + ); + KvError::WatchError(format!( + "stream retention overran live watch (first_sequence {first} > delivered \ + frontier {frontier} + 1); restart will resync" + )) + } + + let mut frontier = resume_revision; + let mut backstop = tokio::time::interval(FLOOR_GUARD_INTERVAL); + backstop.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + backstop.tick().await; // consume the immediate first tick + + loop { + tokio::select! { + entry = watcher.next() => { + let Some(entry) = entry else { break }; + match entry { + Ok(entry) => { + let revision = entry.revision; + // In-band gap check BEFORE processing: never fold + // past unexamined evidence of eviction. + if revision > frontier.saturating_add(1) { + let first = first_sequence().await?; + if !crate::protocol::resume_window_ok(frontier, first) { + return Err(trip(frontier, first, bucket)); + } + // Benign interior gap: every evicted revision + // below a still-low floor was a per-subject + // overwrite, whose later revision the fold will + // see — safe for last-write-wins. + } + frontier = frontier.max(revision); + let update = nats_entry_to_kv_update(entry); + if tx.send(update).await.is_err() { + debug!("watch receiver closed"); + break; + } + } + Err(e) => { + error!(error = %e, "NATS KV watch error"); + return Err(KvError::WatchError(e.to_string())); + } + } + } + _ = backstop.tick() => { + // No-traffic backstop: nothing is being delivered, so the + // in-band check has no evidence to act on; probe the floor + // directly. + let first = first_sequence().await?; + if !crate::protocol::resume_window_ok(frontier, first) { + return Err(trip(frontier, first, bucket)); + } + } + } + } + Ok(()) +} + /// Check if a NATS watch error indicates the requested start sequence is /// too old (compacted), meaning callers should fall back to a full watch. /// @@ -1217,13 +1337,16 @@ impl KvWatcher for NatsKvWatcher { }; // Re-check AFTER the consumer exists: head eviction in the window // between the pre-flight check and consumer creation would otherwise - // clamp silently. Past this point the consumer delivers from retained - // messages; the residual exposure is the generic retention-vs-lag - // bound every live consumer carries (model axiom 5). + // clamp silently. self.check_resume_window(revision).await?; info!(revision, "resumed watch from cursor"); - stream_watch(watcher, &tx).await + // The LIVE floor guard takes over from here: in-band gapped-delivery + // checks plus a no-traffic backstop, so retention overrunning this + // watch mid-stream fail-stops into the restart→resync repair path + // instead of silently skipping evicted deletes (model: + // tests/model_live_watch.rs). + stream_watch_floor_guarded(watcher, &tx, revision, &self.js, &self.bucket).await } async fn watch_prefix_from( @@ -1694,3 +1817,161 @@ mod tests { } } } + +/// Live-server conformance tests for the floor guard +/// ([`stream_watch_floor_guarded`]) — these drive the guarded loop DIRECTLY +/// with a deliberately clamped `Watch`, which reproduces exactly the state +/// retention leaves behind when it overruns a live consumer (the watcher +/// methods' resume-time checks can't be raced deterministically from +/// outside, but the guarded loop neither knows nor cares how its watch got +/// clamped). Spawns a throwaway `nats-server` (mise-installed, same pattern +/// as tests/common). +#[cfg(test)] +mod floor_guard_tests { + use super::*; + use std::process::{Child, Command, Stdio}; + + struct TestServer { + child: Child, + url: String, + _dir: tempfile::TempDir, + } + + impl Drop for TestServer { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + } + + async fn start_server() -> TestServer { + let bin = std::env::var("NATS_SERVER_BIN").unwrap_or_else(|_| "nats-server".into()); + let port = std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port(); + let dir = tempfile::tempdir().unwrap(); + let child = Command::new(&bin) + .args([ + "--jetstream", + "--addr", + "127.0.0.1", + "--port", + &port.to_string(), + "--store_dir", + dir.path().to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap_or_else(|e| panic!("spawn {bin}: {e}; run `mise install`")); + let server = TestServer { + child, + url: format!("nats://127.0.0.1:{port}"), + _dir: dir, + }; + for _ in 0..100 { + if async_nats::connect(&server.url).await.is_ok() { + return server; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!("nats-server never became ready"); + } + + /// `(js, kv store)` with five revisions across five subjects (history 1). + async fn seeded_bucket( + url: &str, + ) -> ( + async_nats::jetstream::Context, + async_nats::jetstream::kv::Store, + ) { + let client = async_nats::connect(url).await.unwrap(); + let js = async_nats::jetstream::new(client); + let kv = js + .create_key_value(async_nats::jetstream::kv::Config { + bucket: "guard".into(), + history: 1, + ..Default::default() + }) + .await + .unwrap(); + for i in 1..=5u8 { + kv.put(format!("k{i}"), vec![i].into()).await.unwrap(); + } + (js, kv) + } + + /// TRUE POSITIVE: the watch was clamped past evicted revisions (purge + /// advanced first_seq beyond the frontier) — the first gapped delivery + /// must trip the guard BEFORE the entry is processed, never silently + /// folding past the lost range. This is the live twin of the model's + /// `GuardRepair`-only-progress gate. + #[tokio::test(flavor = "multi_thread")] + async fn gapped_delivery_with_advanced_floor_trips() { + let server = start_server().await; + let (js, kv) = seeded_bucket(&server.url).await; + + // Evict revisions 1-3 outright: first_sequence becomes 4. + let mut stream = js.get_stream("KV_guard").await.unwrap(); + stream.purge().sequence(4).await.unwrap(); + assert_eq!(stream.info().await.unwrap().state.first_sequence, 4); + + // A consumer resuming from revision 1 gets CLAMPED to revision 4 + // (NATS's silent skip, pinned by tests/resync.rs). Hand that watch + // to the guarded loop as a live consumer whose retention just + // overran it. + let watch = kv.watch_all_from_revision(2).await.unwrap(); + let (tx, mut rx) = tokio::sync::mpsc::channel(64); + let drain = tokio::spawn(async move { while rx.recv().await.is_some() {} }); + + let err = stream_watch_floor_guarded(watch, &tx, 1, &js, "guard") + .await + .expect_err("a gapped delivery over an advanced floor must trip"); + assert!( + err.to_string().contains("retention overran live watch"), + "{err}" + ); + drop(tx); + let _ = drain.await; + } + + /// NO FALSE POSITIVE: interior (per-subject) eviction also gaps the + /// delivered revisions, but the floor stays at or below the frontier — + /// benign for a last-write-wins fold, and the guard must let it + /// through. (Every existing bootstrap e2e also rides this path on its + /// resume; this pins the discrimination explicitly.) + #[tokio::test(flavor = "multi_thread")] + async fn benign_interior_gap_passes() { + let server = start_server().await; + let (js, kv) = seeded_bucket(&server.url).await; + + // Overwrite k2 and k3: revisions 2 and 3 are interior-evicted + // (history 1), revisions 6 and 7 replace them. first_sequence stays + // 1 (k1's revision is retained). + kv.put("k2", vec![22].into()).await.unwrap(); + kv.put("k3", vec![33].into()).await.unwrap(); + let mut stream = js.get_stream("KV_guard").await.unwrap(); + assert_eq!(stream.info().await.unwrap().state.first_sequence, 1); + + // Resume from revision 1: deliveries jump 2 and 3 — gapped, benign. + let watch = kv.watch_all_from_revision(2).await.unwrap(); + let (tx, mut rx) = tokio::sync::mpsc::channel(64); + let guard = + tokio::spawn( + async move { stream_watch_floor_guarded(watch, &tx, 1, &js, "guard").await }, + ); + + let mut got = Vec::new(); + while got.len() < 4 { + let update = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .expect("deliveries continue past benign gaps") + .expect("watch alive"); + got.push(update.version().as_u64().unwrap()); + } + assert_eq!(got, vec![4, 5, 6, 7], "interior gaps jumped, tail dense"); + guard.abort(); // endless live watch; the assertion above is the test + } +} diff --git a/tests/model.rs b/tests/model.rs index 706d4de..98ce74e 100644 --- a/tests/model.rs +++ b/tests/model.rs @@ -82,13 +82,16 @@ //! (empirical tier: tampered-artifact and multi-SST round-trip tests). //! NATS KV CAS semantics for the lease are unneeded here (see above); the //! lease layer is verified by `integration.rs` contention tests. -//! 5. Retention outlives consumer lag: the stream's retention window exceeds -//! the seconds-scale lag between a consumer's expiry check and its -//! deliveries. The code closes the check→create window with a re-check -//! after consumer creation (`nats.rs`); the residual in-flight exposure -//! is the bound every log consumer (Kafka included) operates under, and -//! is an OPERATING requirement: configure retention in hours, not -//! seconds. The model's atomic delivery transitions encode this axiom. +//! 5. Retention outlives consumer lag — NARROWED to prefix-scoped watches +//! and the fresh full watch's initial history scan. The ALL-scope resume +//! watch (steady-state operation) no longer relies on it: the live floor +//! guard (`tests/model_live_watch.rs`, `stream_watch_floor_guarded`) +//! fail-stops on in-band evidence of retention overrunning the consumer +//! and routes into this model's verified resume → expiry → resync repair +//! path. Prefix scopes deliver sparse revisions by design and cannot +//! distinguish benign from hazardous eviction client-side; for them the +//! operating requirement stands: configure retention in hours, not +//! seconds. //! //! ## Bounds and the small-scope argument //! diff --git a/tests/model_live_watch.rs b/tests/model_live_watch.rs new file mode 100644 index 0000000..a62b6c5 --- /dev/null +++ b/tests/model_live_watch.rs @@ -0,0 +1,265 @@ +//! Exhaustive model-check of the LIVE-WATCH retention race (Stateright) — +//! the axiom-5 boundary of `tests/model.rs`, now mechanized. +//! +//! The hazard: a live consumer's position can fall behind the stream's +//! retention floor. JetStream then silently skips evicted messages (the same +//! clamp behavior `tests/resync.rs` pins for resumes — consumers never error +//! on evicted messages, they just never see them). A skipped DELETE marker +//! leaves the fold holding a key the bucket deleted, permanently and +//! silently: the exact failure class the resume-time `check_resume_window` +//! eliminates, alive again mid-watch. +//! +//! The fix this model dictates: a periodic **floor guard** during live +//! consumption — the same shared kernel guard (`protocol::resume_window_ok`) +//! applied to the delivered frontier instead of a resume cursor. A trip +//! fails the watch; the caller's restart routes into the resume → expiry → +//! resync path whose convergence the main model already proves. The repair +//! composite here (fold := bucket truth, frontier := floor) is exactly that +//! verified subprotocol, collapsed to one transition. +//! +//! Checked, exhaustively within bounds: +//! - GUARDED: every maximal run ends with the fold equal to the bucket — +//! divergence is at worst transient (bounded by the guard cadence), never +//! permanent. The guard genuinely trips (witness) and markers are also +//! delivered normally (witness), so the theorem is not vacuous. +//! - UNGUARDED (the pre-guard code, kept as the machine-checked record): +//! permanent terminal divergence is REACHABLE — fold holds a deleted key +//! forever with nothing left to deliver and no error anywhere. +//! - BOTH: divergence is only ever in the stale direction — the fold never +//! drops a key the bucket still has (no phantom deletes). +//! +//! Scope, honestly: this models the ALL-scope watch, where every stream +//! message is deliverable to the consumer and a frontier-vs-floor gap +//! therefore implies genuinely missed messages. Prefix-scoped watches +//! cannot distinguish benign eviction (non-matching subjects) from a missed +//! marker without server-side help; they retain the narrowed operating +//! axiom (retention >> lag) plus the resume-time check on every restart. +//! +//! Abstractions, stated: +//! - **No client buffer.** `Compact` means eviction of messages the +//! consumer has NOT received — the hazardous kind. In reality a message +//! pushed to the client before eviction is still processed; that is +//! received data, not loss, and needs no guard. (The code's dense-pass +//! fast path folds such messages without probing — sound because a +//! delivered message was by definition retained when pushed.) +//! - **`GuardRepair` is a composite** of fail-stop → restart → resume → +//! `CursorExpired` → resync, each verified separately (`tests/model.rs`, +//! `tests/resync.rs`); the composition is by argument. Repair LIVENESS is +//! conditional on the caller restarting the failed watch (standard +//! supervision); the safety half — never folding past unexamined +//! evidence — is unconditional in the code path itself. + +use slipstream::protocol::resume_window_ok; +use stateright::{Checker, Model, Property}; + +/// Stream revisions run 1..=MAX_REV. +const MAX_REV: u8 = 5; + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct St { + /// Stream high-water revision. + head: u8, + /// Retention floor: revisions <= floor are evicted. + floor: u8, + /// Revision of the sentinel key's delete marker, if deleted. + delete_rev: Option, + /// The consumer's delivered frontier (max revision handed to the fold). + frontier: u8, + /// The fold's view of the sentinel key (bucket truth: `delete_rev` none). + fold_has_key: bool, + /// Latched when the floor guard tripped (witness that the guarded + /// theorem is earned by repair, not by the race never occurring). + tripped: bool, +} + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +enum Act { + /// A new revision lands (a put to some other key). + Churn, + /// The sentinel key is deleted (consumes a revision: the marker). + DeleteKey, + /// Retention evicts the oldest retained revision. + Compact, + /// The consumer receives the next RETAINED revision after its frontier — + /// evicted revisions are silently skipped, which is the hazard. + Deliver, + /// Guarded variant only: the periodic floor check fires, finds the + /// frontier behind the floor, and fail-stops; the restart's resume → + /// expiry → resync (verified by tests/model.rs) repairs the fold. Free + /// interleaving of this action is a superset of every periodic cadence. + GuardRepair, +} + +#[derive(Clone)] +struct LiveWatch { + guarded: bool, +} + +impl LiveWatch { + fn bucket_has_key(s: &St) -> bool { + s.delete_rev.is_none() + } +} + +impl Model for LiveWatch { + type State = St; + type Action = Act; + + fn init_states(&self) -> Vec { + // The sentinel key exists and the fold is synced at frontier 0. + vec![St { + head: 0, + floor: 0, + delete_rev: None, + frontier: 0, + fold_has_key: true, + tripped: false, + }] + } + + fn actions(&self, s: &St, acts: &mut Vec) { + if s.head < MAX_REV { + acts.push(Act::Churn); + if s.delete_rev.is_none() { + acts.push(Act::DeleteKey); + } + } + if s.floor < s.head { + acts.push(Act::Compact); + } + // Something retained remains beyond the frontier. + if s.frontier.max(s.floor) < s.head { + // GUARDED: delivery never silently jumps a gap. A delivered + // revision > frontier+1 is in-band evidence of eviction past + // the frontier, checked AT THE DELIVERY (the kernel gate below) + // — the checker rejected the periodic-only design with exactly + // the catch-up-erases-the-evidence trace this gate closes. + // UNGUARDED: the skip happens silently (JetStream's behavior). + if !self.guarded || resume_window_ok(s.frontier as u64, s.floor as u64 + 1) { + acts.push(Act::Deliver); + } + } + // THE GUARD GATE — the shared kernel: the frontier has fallen + // behind the first retained revision (floor + 1). Reached either by + // the gapped-delivery check (above: Deliver disabled, this is the + // only progress) or by the periodic backstop when no deliveries + // arrive at all; free interleaving covers every cadence of both. + if self.guarded && !resume_window_ok(s.frontier as u64, s.floor as u64 + 1) { + acts.push(Act::GuardRepair); + } + } + + fn next_state(&self, s: &St, a: Act) -> Option { + let mut s = s.clone(); + match a { + Act::Churn => s.head += 1, + Act::DeleteKey => { + s.head += 1; + s.delete_rev = Some(s.head); + } + Act::Compact => s.floor += 1, + Act::Deliver => { + // Next retained revision; anything evicted in between is + // SKIPPED — if the skipped range held the delete marker, the + // fold silently keeps the key. + let next = s.frontier.max(s.floor) + 1; + if next > s.head { + return None; + } + s.frontier = next; + if s.delete_rev == Some(next) { + s.fold_has_key = false; + } + } + Act::GuardRepair => { + // Fail-stop + restart + resume(frontier) + CursorExpired + + // resync, collapsed to its verified outcome: the fold equals + // the bucket's live keys, and consumption is re-entitled + // from the floor. + s.tripped = true; + s.fold_has_key = Self::bucket_has_key(&s); + s.frontier = s.floor; + } + } + Some(s) + } + + fn properties(&self) -> Vec> { + let mut props: Vec> = Vec::new(); + + // BOTH variants: divergence is only ever stale-direction — the fold + // never drops a key the bucket still has. + props.push(Property::::always( + "no phantom deletes: the fold never drops a live key", + |_, s| s.fold_has_key || !LiveWatch::bucket_has_key(s), + )); + + // Vacuity witnesses shared by both variants. + props.push(Property::::sometimes( + "the marker is delivered normally and the fold drops the key", + |_, s| !s.fold_has_key && !LiveWatch::bucket_has_key(s), + )); + + if self.guarded { + // THE theorem: every maximal run ends with the fold equal to the + // bucket — divergence is at worst transient (a trip away), never + // permanent. Terminal-state invariant (cycle-proof): a state + // with no enabled actions must be converged. + props.push(Property::::always( + "every maximal run ends with the fold equal to the bucket", + |m, s| { + let mut acts = Vec::new(); + m.actions(s, &mut acts); + !acts.is_empty() || s.fold_has_key == LiveWatch::bucket_has_key(s) + }, + )); + // The theorem is earned: the guard genuinely trips within the + // bounds (the race occurs and is repaired, not avoided). + props.push(Property::::sometimes( + "the floor guard trips and repairs a real divergence", + |_, s| s.tripped && s.fold_has_key == LiveWatch::bucket_has_key(s), + )); + } else { + // The pre-guard code, machine-checked: PERMANENT silent + // divergence is reachable — a terminal state where the fold + // holds a deleted key, nothing remains to deliver, and no error + // occurred anywhere. + props.push(Property::::sometimes( + "HAZARD reachable: permanent silent divergence (marker evicted unseen)", + |m, s| { + let mut acts = Vec::new(); + m.actions(s, &mut acts); + acts.is_empty() && s.fold_has_key && !LiveWatch::bucket_has_key(s) + }, + )); + } + + props + } +} + +fn check(guarded: bool, label: &str) { + let model = LiveWatch { guarded }; + let checker = model.checker().spawn_bfs().join(); + println!( + "{label}: {} states, {} unique", + checker.state_count(), + checker.unique_state_count(), + ); + checker.assert_properties(); +} + +/// The SHIPPED behavior: All-scope watches carry the periodic floor guard +/// (`nats.rs`), gated by the same shared kernel this model executes. +#[test] +fn guarded_live_watch_always_converges() { + check(true, "live watch: guarded"); +} + +/// The pre-guard behavior, kept as the machine-checked record of the hazard: +/// retention overrunning a live consumer silently and permanently diverges +/// the fold. +#[test] +fn unguarded_live_watch_diverges_permanently() { + check(false, "live watch: unguarded"); +} From 5d687327abf5324c146a0afd32b1dca783ccf7fe Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Thu, 11 Jun 2026 23:43:27 -0700 Subject: [PATCH 2/3] fix: re-queue raw batches on transient store-apply failure; model watch_applied core MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BUG (found while writing the cursor-authority model, reproduced live by transient_store_failure_never_leaves_a_cursor_gap): a failed store apply dropped its raw batch on the warn-and-continue path, and the NEXT successful flush committed only newer updates under the newest cursor — one transient store error followed by one success left a permanent, restart-surviving hole in the fold behind a cursor that claims completeness. The "snapshot is a cache, refold on restart" argument breaks precisely because the refold starts from the advanced cursor. Fix: a failed apply re-queues its raw batch (prepended, stream order preserved) so the next flush commits cumulatively — the store's cursor and contents always advance together. A persistent failure streak (16 consecutive) fail-stops before the backlog grows unboundedly; the restart refolds from the store's last good cursor. Models (campaign #3, the watch_applied core): - tests/model_applied.rs — cursor authority over every interleaving of delivery, flush boundaries, transient store failures, and crashes. Mutations prove both rules load-bearing: DropFailedBatch (the pre-fix shipped behavior) and ResumeFromMemApplied (why restarts must resume from the store's persisted cursor, which legitimately lags the in-memory applied cursor across failures) each reach the restart-surviving hole. - tests/model_resync_order.rs — the resync ack barrier's ordering claim ("synthetic deletes strictly before the first re-list put"), previously a comment, is now a theorem: delete-then-recreate converges straddling the listing in either direction; the NoAckBarrier mutation reaches the lost-recreate divergence that is the machine-checked reason the handshake exists. Fidelity note recorded in both models: no shared kernel exists here (the invariants are control-flow structure, not guard expressions), so fidelity rests on the transition audit, the mutation contrasts, and the live reproduction test. Co-Authored-By: Claude Fable 5 --- src/applied.rs | 247 +++++++++++++++++++++++++++++++-- src/nats.rs | 15 +- src/transport.rs | 4 +- tests/model_applied.rs | 269 ++++++++++++++++++++++++++++++++++++ tests/model_resync_order.rs | 230 ++++++++++++++++++++++++++++++ 5 files changed, 747 insertions(+), 18 deletions(-) create mode 100644 tests/model_applied.rs create mode 100644 tests/model_resync_order.rs diff --git a/src/applied.rs b/src/applied.rs index a12affd..65f7143 100644 --- a/src/applied.rs +++ b/src/applied.rs @@ -224,7 +224,9 @@ pub async fn watch_applied( mut store: Option, // `Some(rx)` arms an export-request arm in the select loop: each // [`ExportRequest`] is handled between flushes (pending batch flushed - // first), so the exported artifact's cursor is exactly the applied cursor. + // first), so the exported artifact's cursor is the applied cursor (or, + // across a transiently failed store flush, the store's own lagging but + // self-consistent cursor — never a cursor past unfolded data). // `None` (or dropping the paired sender) leaves the loop's behavior // unchanged. mut exports: Option>, @@ -298,6 +300,12 @@ where // state), whereas the parsed `batch` above is the consumer's domain view. let mut raw_batch: Vec = Vec::new(); let mut batch_high = WatchCursor::none(); + // Consecutive store-apply failures. A transient failure re-queues its raw + // batch (cursor authority: the store's cursor and contents advance + // together, always); a persistent streak fail-stops before the re-queued + // backlog grows without bound. + const MAX_STORE_APPLY_FAILURES: u32 = 16; + let mut store_fail_streak: u32 = 0; // `Some` once a batch has opened and the window timer is armed; `None` // between flushes. Only the armed/idle distinction is read in the loop — // the absolute instant lives in the pinned `sleep` future below. @@ -307,8 +315,10 @@ where // completion, advance the cursor, fold the raw batch + cursor durably into // `store`, then fire `on_applied`. The store fold runs on a blocking task // (its `apply` may block on I/O), moving the store in and taking it back — the - // same offload the append log's compaction always used. A store error is - // logged and the watch continues (the snapshot is a cache); a panicked + // same offload the append log's compaction always used. A TRANSIENT store + // error re-queues the raw batch for cumulative commit on the next flush + // (the watch continues; the store's cursor never advances past data it + // dropped) and a persistent failure streak is fatal; a panicked // blocking task drops the store irrecoverably, which breaks the // resume-after-restart guarantee, so it is surfaced as fatal. macro_rules! flush { @@ -341,19 +351,48 @@ where // unchanged — possibly expired — cursor only ever re-runs // this same resync on the next restart. let cur = applied.clone(); - // Hand the store back unconditionally on a clean return so - // a *failed* apply (Ok(Err)) keeps the watch running; only - // a *panicked* task (Err) loses the store and is fatal. + // Hand the store AND the raw batch back on a clean return: + // a *failed* apply (Ok(Err)) RE-QUEUES the batch so the + // next flush commits it cumulatively — the store's cursor + // and contents always advance together. Dropping the + // failed batch instead lets the NEXT successful flush + // advance the cursor over a hole that survives every + // restart (reproduced by + // `transient_store_failure_never_leaves_a_cursor_gap`). + // Only a *panicked* task (Err) loses the store: fatal. match tokio::task::spawn_blocking(move || { let res = st.apply(&raw, &cur); - (st, res) + (st, raw, res) }) .await { - Ok((st, Ok(()))) => store = Some(st), - Ok((st, Err(e))) => { - warn!(error = %e, "snapshot store apply failed; continuing"); + Ok((st, _raw, Ok(()))) => { store = Some(st); + store_fail_streak = 0; + } + Ok((st, raw, Err(e))) => { + store_fail_streak += 1; + if store_fail_streak >= MAX_STORE_APPLY_FAILURES { + // A persistently failing store would otherwise + // grow the re-queued batch without bound while + // the fold silently stales. Fail-stop: the + // restart refolds the tail from the store's + // last good cursor. + warn!(error = %e, streak = store_fail_streak, + "snapshot store apply failing persistently; aborting watch"); + handle.abort(); + return Err(KvError::WatchError(format!( + "snapshot store apply failed {store_fail_streak} consecutive times: {e}" + ))); + } + warn!(error = %e, streak = store_fail_streak, + "snapshot store apply failed; batch re-queued for the next flush"); + store = Some(st); + // Prepend: the failed range precedes anything + // received since (stream order is preserved for + // the eventual cumulative commit). + let newer = std::mem::replace(&mut raw_batch, raw); + raw_batch.extend(newer); } Err(e) => { warn!(error = %e, "snapshot store task panicked; aborting watch"); @@ -481,7 +520,14 @@ where flush!(); // Ack AFTER the deletes are applied: the watch task is // holding the fallback watch until it hears back, which - // is what orders deletes before the re-list. + // is what orders deletes before the re-list + // (tests/model_resync_order.rs proves the barrier + // load-bearing). If the flush's STORE apply failed + // transiently, the deletes sit re-queued at the FRONT + // of the raw batch — still strictly before any + // re-list put in the eventual cumulative commit, and + // the domain apply saw them before this ack either + // way. let _ = ack.send(()); } None => resyncs = None, @@ -491,9 +537,14 @@ where // Export request. Placed after shutdown/window (they stay prompt) // and before `rx.recv()` so a firehose of updates cannot starve an // export indefinitely. The pending batch is flushed first, so the - // exported cursor is exactly the applied cursor; the export itself - // runs on a blocking task with the store moved in and taken back — - // the same offload the flush path uses. + // exported cursor is exactly the applied cursor — except when + // that flush's store apply transiently failed (batch re-queued): + // the export then captures the store's OWN lagging cursor, which + // is still self-consistent with its contents (cursor authority, + // tests/model_applied.rs); the artifact never includes unfolded + // data, and a bootstrap from it simply replays the short gap. + // The export itself runs on a blocking task with the store moved + // in and taken back — the same offload the flush path uses. req = async { exports.as_mut().expect("arm guarded by is_some").recv().await }, if exports.is_some() => { match req { @@ -2073,4 +2124,172 @@ mod tests { ); assert_eq!(snap.entries["node.b"].value, b"3"); } + /// A SnapshotStore whose FIRST apply fails (transient store error: disk + /// pressure, lock timeout), then behaves normally — the trigger for the + /// lost-raw-batch hazard in the flush path. + struct FailOnceStore { + inner: AppendLogSnapshot, + failed: std::sync::atomic::AtomicBool, + } + + impl crate::snapshot::SnapshotStore for FailOnceStore { + fn load( + _path: &std::path::Path, + ) -> Result<(WatchCursor, Self), crate::snapshot::SnapshotError> { + unreachable!("test store is constructed directly") + } + fn apply( + &mut self, + batch: &[KvUpdate], + cursor: &WatchCursor, + ) -> Result<(), crate::snapshot::SnapshotError> { + if !self.failed.swap(true, Ordering::SeqCst) { + return Err(crate::snapshot::SnapshotError::Backend( + "injected transient store failure".into(), + )); + } + self.inner.apply(batch, cursor) + } + fn get(&self, key: &str) -> Result, crate::snapshot::SnapshotError> { + self.inner.get(key) + } + fn range(&self, prefix: &str) -> Result, crate::snapshot::SnapshotError> { + self.inner.range(prefix) + } + fn cursor(&self) -> WatchCursor { + self.inner.cursor() + } + fn export_to( + &mut self, + dest_dir: &std::path::Path, + ) -> Result { + self.inner.export_to(dest_dir) + } + } + + /// CURSOR AUTHORITY under a transient store failure: a failed store apply + /// must NOT cause later successful applies to advance the persisted + /// cursor past data that never landed. The failed batch is re-queued and + /// committed cumulatively with the next flush, so the store's cursor + /// never lies about its contents — a restart resuming from it sees + /// exactly the missing tail, not a silent hole. + /// + /// (Pre-fix behavior, found while writing the watch_applied model: the + /// failed batch's raw updates were dropped on the warn-and-continue + /// path, and the NEXT successful flush committed only newer updates + /// under the newest cursor — a permanent, restart-surviving gap in the + /// fold.) + #[tokio::test(start_paused = true)] + async fn transient_store_failure_never_leaves_a_cursor_gap() { + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().join("fold.snap"); + let (_r, inner) = AppendLogSnapshot::open(&path, u64::MAX).unwrap(); + let store = FailOnceStore { + inner, + failed: std::sync::atomic::AtomicBool::new(false), + }; + + // max: 1 -> one flush per update: flush #1 (a@1) hits the injected + // failure, flush #2 (b@2) succeeds. + let updates = vec![put("node.a", b"1", 1), put("node.b", b"2", 2)]; + let watcher = Arc::new(MockWatcher::new(updates, true)); + let (sd_tx, sd_rx) = watch::channel(false); + + let task = tokio::spawn(watch_applied( + watcher, + WatchScope::All, + None, + None, + Some(store), + None, + BatchConfig { + window: Duration::from_millis(1), + max: 1, + }, + parse_put, + |_batch: Vec>| {}, + |_| {}, + sd_rx, + )); + + tokio::time::sleep(Duration::from_millis(50)).await; + sd_tx.send(true).unwrap(); + let cursor = task.await.unwrap().unwrap(); + assert_eq!(cursor.as_u64(), Some(2)); + + // The store on disk must be SELF-CONSISTENT: whatever its cursor + // claims, the data at or below it is present. With the re-queue fix + // the cumulative commit lands both keys at cursor 2. + let (persisted, reopened) = AppendLogSnapshot::open(&path, u64::MAX).unwrap(); + assert_eq!(persisted.as_u64(), Some(2), "cursor reached the head"); + assert_eq!( + reopened.get("node.a").unwrap().map(|e| e.value), + Some(b"1".to_vec()), + "the transiently-failed batch was re-queued, not silently dropped \ + behind an advancing cursor" + ); + assert_eq!( + reopened.get("node.b").unwrap().map(|e| e.value), + Some(b"2".to_vec()) + ); + } + /// A reader whose live-key listing always fails — the resync's I/O + /// failure mode. + struct FailingReader; + + #[async_trait] + impl KvReader for FailingReader { + async fn get(&self, _key: &str) -> Result, KvError> { + unreachable!("resync only lists keys") + } + async fn keys(&self, _prefix: &str) -> Result, KvError> { + Err(KvError::OperationFailed("injected listing failure".into())) + } + async fn scan(&self, _prefix: &str) -> Result, KvError> { + unreachable!("resync only lists keys") + } + } + + /// REGRESSION PIN (code-level twin of tests/model.rs's Degrade + /// configuration): a resync whose live-key listing fails must FAIL THE + /// WATCH, not degrade to re-list-only with a warning — the degrade + /// semantics provably break the convergence theorem (silent stale keys). + /// Reverting `resync_stale_keys` to warn-and-continue fails this test. + #[tokio::test] + async fn resync_listing_failure_is_fatal_not_degraded() { + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().join("resync-fatal.snap"); + let (_r, mut store) = AppendLogSnapshot::open(&path, u64::MAX).unwrap(); + store + .apply(&[put("node.a", b"1", 1)], &WatchCursor::from_u64(1)) + .unwrap(); + + // Resume cursor expired -> resync path -> reader listing fails. + let mock = MockWatcher { + full: Mutex::new(Some(vec![])), + from: Mutex::new(Some(vec![])), + from_expires: true, + hold: false, + }; + let (_sd_tx, sd_rx) = watch::channel(false); + let err = watch_applied( + Arc::new(mock), + WatchScope::All, + Some(WatchCursor::from_u64(1)), + Some(Arc::new(FailingReader) as Arc), + Some(store), + None, + BatchConfig::default(), + parse_put, + |_batch: Vec>| {}, + |_| {}, + sd_rx, + ) + .await + .expect_err("a failed resync listing must fail the watch"); + assert!( + err.to_string().contains("resync failed listing live keys"), + "{err}" + ); + } } diff --git a/src/nats.rs b/src/nats.rs index e85f4e4..d45292c 100644 --- a/src/nats.rs +++ b/src/nats.rs @@ -1926,9 +1926,18 @@ mod floor_guard_tests { let (tx, mut rx) = tokio::sync::mpsc::channel(64); let drain = tokio::spawn(async move { while rx.recv().await.is_some() {} }); - let err = stream_watch_floor_guarded(watch, &tx, 1, &js, "guard") - .await - .expect_err("a gapped delivery over an advanced floor must trip"); + // The 5s bound pins IN-BAND detection: the trip must come from the + // gapped-delivery check itself, not the 30s no-traffic backstop. A + // regression to periodic-only detection (the design the model + // checker rejected — catch-up between probes erases the evidence) + // fails this bound. + let err = tokio::time::timeout( + Duration::from_secs(5), + stream_watch_floor_guarded(watch, &tx, 1, &js, "guard"), + ) + .await + .expect("the trip must be IN-BAND (immediate), not backstop-paced") + .expect_err("a gapped delivery over an advanced floor must trip"); assert!( err.to_string().contains("retention overran live watch"), "{err}" diff --git a/src/transport.rs b/src/transport.rs index 1db8749..dd9fbfd 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -667,7 +667,9 @@ fn map_obj(e: object_store::Error) -> SnapshotError { /// 1. [`ExportLease::try_acquire`] — `Ok(None)` means another node owns this /// round; nothing else happens. /// 2. An [`ExportRequest`] into the live [`watch_applied`](crate::watch_applied) -/// loop (pending batch flushed first; artifact cursor == applied cursor). +/// loop (pending batch flushed first; the artifact carries the store's +/// self-consistent cursor — the applied cursor, or a lagging one across +/// a transiently failed store flush, never a cursor past unfolded data). /// 3. [`ArtifactTransport::upload`]. /// 4. [`LeaseGuard::complete`](crate::LeaseGuard::complete) — only after the /// upload succeeded, so a published completion never lies. diff --git a/tests/model_applied.rs b/tests/model_applied.rs new file mode 100644 index 0000000..98f5010 --- /dev/null +++ b/tests/model_applied.rs @@ -0,0 +1,269 @@ +//! Exhaustive model-check of `watch_applied`'s CURSOR AUTHORITY — the +//! invariant the entire resume story sits on: the store's persisted cursor +//! never advances past data that is not in the store. A violation is a +//! restart-surviving hole in the fold: silent data loss, strictly worse +//! than anything the transport could do. +//! +//! Modeled: the delivery → batch → flush pipeline with TRANSIENT STORE +//! FAILURES (re-queued for cumulative commit — `applied.rs`'s flush path), +//! crashes at any point, and restarts that resume from the store's +//! persisted cursor. Flush boundaries interleave freely, which is a +//! superset of every window/max/shutdown/export trigger policy. +//! +//! Mutations — both are REAL bug classes, the first found live: +//! - `DropFailedBatch`: a failed store apply drops its raw batch and the +//! next successful flush advances the cursor over the hole. THIS WAS THE +//! SHIPPED CODE until `transient_store_failure_never_leaves_a_cursor_gap` +//! reproduced it (found while writing this model); the fix re-queues the +//! failed batch. The checker proves the old behavior violates the +//! invariant. +//! - `ResumeFromMemApplied`: restart resumes from the in-memory applied +//! cursor instead of the store's persisted one. The in-memory cursor +//! legitimately runs ahead of the store across failed flushes (the +//! domain-apply side is not transactional with the store), so resuming +//! from it skips the un-stored gap. Pins WHY the resume source must be +//! the store's own cursor. +//! +//! Fidelity note: unlike the transport and live-watch models, there is no +//! shared decision kernel here — the invariant is carried by control-flow +//! structure (what gets re-queued, which cursor is committed, where the +//! restart resumes), not by a guard expression. Fidelity therefore rests on +//! the transition audit against `applied.rs`'s flush macro plus the +//! mutation contrast, and the empirical twin in `applied.rs` tests. + +use stateright::{Checker, Model, Property}; + +/// Stream revisions run 1..=MAX_REV. +const MAX_REV: u8 = 5; + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct St { + /// Bucket high-water revision. + head: u8, + /// Delivered to the loop (in-memory; lost on crash). + delivered: u8, + /// The store's PERSISTED cursor — what a restart resumes from. + store_cursor: u8, + /// Lower bound of the re-queued/pending raw range: the store holds + /// complete contents up to `store_cursor`; the loop holds raw updates + /// (`q_from`, `delivered`]. Shipped semantics keep `q_from == + /// store_cursor` (re-queue on failure); the DropFailedBatch mutation + /// lets it advance past the store. + q_from: u8, + /// The in-memory applied cursor: advances at every flush attempt, + /// success or failure (the domain apply is not transactional with the + /// store). + applied_mem: u8, + /// Latched when a commit left the store's contents incomplete below its + /// cursor — the restart-surviving hole. THE invariant: never latches. + holes: bool, +} + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +enum Act { + /// A new revision lands in the bucket. + Churn, + /// The watch delivers the next revision to the loop. + Deliver, + /// A flush whose store apply SUCCEEDS: commits (`q_from`, `delivered`] + /// with cursor `delivered`, atomically (backend axiom: data + cursor in + /// one transaction). + FlushOk, + /// A flush whose store apply FAILS transiently. Shipped: the raw batch + /// is re-queued (`q_from` unchanged — the next FlushOk commits + /// cumulatively); the in-memory applied cursor still advances. + /// DropFailedBatch mutation: the batch is dropped (`q_from` jumps to + /// `delivered`). + FlushFail, + /// Process crash: all in-memory state is lost; the restart reopens the + /// store and resumes the watch from its persisted cursor (or, under the + /// ResumeFromMemApplied mutation, from the in-memory applied cursor — + /// as if the caller trusted `on_applied` instead of the store). + CrashRestart, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Mutation { + None, + DropFailedBatch, + ResumeFromMemApplied, +} + +#[derive(Clone)] +struct AppliedLoop { + mutation: Mutation, +} + +impl Model for AppliedLoop { + type State = St; + type Action = Act; + + fn init_states(&self) -> Vec { + vec![St { + head: 0, + delivered: 0, + store_cursor: 0, + q_from: 0, + applied_mem: 0, + holes: false, + }] + } + + fn actions(&self, s: &St, acts: &mut Vec) { + if s.head < MAX_REV { + acts.push(Act::Churn); + } + if s.delivered < s.head { + acts.push(Act::Deliver); + } + if s.delivered > s.q_from { + acts.push(Act::FlushOk); + acts.push(Act::FlushFail); + } + // A crash is possible at any moment. + acts.push(Act::CrashRestart); + } + + fn next_state(&self, s: &St, a: Act) -> Option { + let mut s = s.clone(); + match a { + Act::Churn => s.head += 1, + Act::Deliver => s.delivered += 1, + Act::FlushOk => { + // Atomic commit of (`q_from`, `delivered`] at cursor + // `delivered`. If `q_from` ran ahead of the store's cursor + // (a dropped batch), the committed contents are missing + // (`store_cursor`, `q_from`] while the cursor claims them: + // the hole, permanent from here on. + if s.q_from > s.store_cursor { + s.holes = true; + } + s.store_cursor = s.delivered; + s.q_from = s.delivered; + s.applied_mem = s.delivered; + } + Act::FlushFail => { + // The domain apply ran and the in-memory cursor advanced + // regardless; only the store commit failed. + s.applied_mem = s.delivered; + match self.mutation { + // Shipped: re-queue — `q_from` stays put, the next + // FlushOk commits cumulatively. + Mutation::None | Mutation::ResumeFromMemApplied => {} + // The pre-fix behavior: the failed batch is dropped. + Mutation::DropFailedBatch => s.q_from = s.delivered, + } + } + Act::CrashRestart => { + let resume = match self.mutation { + Mutation::ResumeFromMemApplied => s.applied_mem, + _ => s.store_cursor, + }; + // Resuming above the store's cursor never re-delivers the + // un-stored gap: it is permanently skipped. + if resume > s.store_cursor { + s.holes = true; + } + s.delivered = resume; + s.q_from = resume.max(s.store_cursor); + s.applied_mem = resume; + } + } + Some(s) + } + + fn properties(&self) -> Vec> { + let mut props: Vec> = Vec::new(); + + if self.mutation == Mutation::None { + // THE cursor-authority invariant, over every interleaving of + // delivery, flush boundaries, transient store failures, and + // crashes: the store's cursor never claims data it dropped. + props.push(Property::::always( + "the store's contents are always complete up to its cursor", + |_, s| !s.holes, + )); + // Cursor authority restated structurally: the pending raw range + // always begins exactly at the store's cursor — nothing between + // the store and the loop can fall on the floor. + props.push(Property::::always( + "the re-queued range is always anchored at the store cursor", + |_, s| s.q_from == s.store_cursor, + )); + // Terminal liveness: every maximal run ends fully caught up. + // (CrashRestart is always enabled, so no state is terminal in + // the strict sense; quiescence here is head reached + nothing + // pending + store caught up, checked as: whenever nothing is + // deliverable or flushable, the store is at the head.) + props.push(Property::::always( + "quiescence implies the store is at the head", + |_, s| { + let quiesced = + s.head == MAX_REV && s.delivered == s.head && s.q_from == s.delivered; + !quiesced || s.store_cursor == s.head + }, + )); + // Vacuity witnesses: failures really happen and are really + // recovered from (a failed flush precedes a successful + // cumulative one), and crashes really replay. + props.push(Property::::sometimes( + "a failed flush is followed by a cumulative successful commit", + |_, s| s.store_cursor > 0 && s.applied_mem == s.store_cursor && s.head > 1, + )); + props.push(Property::::sometimes( + "the in-memory cursor runs ahead of the store across a failure", + |_, s| s.applied_mem > s.store_cursor, + )); + } else { + // Each mutation must produce the hole — proving the respective + // rule (re-queue on failure; resume from the store's cursor) is + // load-bearing, not incidental. + props.push(Property::::sometimes( + "HAZARD reachable: cursor advances over dropped data", + |_, s| s.holes, + )); + } + + props + } +} + +fn check(mutation: Mutation, label: &str) { + let model = AppliedLoop { mutation }; + let checker = model.checker().spawn_bfs().join(); + println!( + "{label}: {} states, {} unique", + checker.state_count(), + checker.unique_state_count(), + ); + checker.assert_properties(); +} + +/// The SHIPPED flush semantics (failed store applies re-queue their raw +/// batch; restarts resume from the store's persisted cursor): cursor +/// authority holds over every interleaving of delivery, flushing, transient +/// failure, and crash. +#[test] +fn shipped_flush_semantics_preserve_cursor_authority() { + check(Mutation::None, "applied: shipped"); +} + +/// The PRE-FIX behavior (failed batch dropped, warn-and-continue): the +/// checker reaches the restart-surviving hole — the machine-checked record +/// of the bug `transient_store_failure_never_leaves_a_cursor_gap` +/// reproduced live. +#[test] +fn dropping_failed_batches_corrupts_the_fold() { + check(Mutation::DropFailedBatch, "applied: drop-on-fail (pre-fix)"); +} + +/// Resuming from the in-memory applied cursor instead of the store's +/// persisted one skips the un-stored gap: pins why the restart's resume +/// source must be the store itself. +#[test] +fn resuming_from_memory_cursor_corrupts_the_fold() { + check( + Mutation::ResumeFromMemApplied, + "applied: resume-from-memory", + ); +} diff --git a/tests/model_resync_order.rs b/tests/model_resync_order.rs new file mode 100644 index 0000000..0eef076 --- /dev/null +++ b/tests/model_resync_order.rs @@ -0,0 +1,230 @@ +//! Exhaustive model-check of the cursor-expired resync's ORDERING claim +//! (Stateright) — until now carried by a comment in `applied.rs`: +//! +//! > "The synthetic deletes are strictly ordered before the first re-list +//! > put, so a key deleted and re-created during the gap converges +//! > correctly." +//! +//! The mechanism under test is the ACK BARRIER: the watch task lists the +//! bucket's live keys, sends them to the main loop, and AWAITS the ack — +//! which the main loop sends only after folding the synthetic deletes — +//! before starting the fallback watch. If the fallback could start earlier, +//! its re-list put for a key re-created during the gap could fold BEFORE +//! the synthetic delete, which then erases the re-creation: the fold drops +//! a key the bucket has, permanently (until that key's next update). The +//! `NoAckBarrier` mutation is exactly that interleaving, and the checker +//! reaches the divergence — proving the barrier is load-bearing. +//! +//! Setup modeled: the fold holds key K from before the gap; during the gap +//! K was deleted and its marker evicted (the resync's reason to exist). K +//! may be RE-CREATED at any point — before the listing (then it is live in +//! the listing, no synthetic delete, and the re-list delivers it) or after +//! (synthetic delete fires, and ONLY the barrier guarantees the re-list put +//! folds after it). +//! +//! Fidelity: the barrier in code is `resync_stale_keys`'s `ack_rx.await` +//! between the `ResyncRequest` send and the fallback watch start, paired +//! with the main loop folding the synthetic deletes before sending the ack +//! (`applied.rs`). No shared kernel exists (it is control flow, not a +//! guard); fidelity rests on this transition audit, the mutation contrast, +//! and the mock-driven ordering tests in `applied.rs`. +//! +//! `FoldSyntheticDeletes` is modeled atomic; in code, the synth-delete +//! flush's STORE apply can transiently fail. The ordering still holds by +//! composition with `tests/model_applied.rs`: the failed batch is re-queued +//! AT THE FRONT (stream order preserved), so the eventual cumulative commit +//! applies delete-then-put in order within one backend batch, and the +//! domain apply saw the deletes before the ack regardless. The listing is +//! modeled instantaneous; a key changing mid-scan is equivalent to ordering +//! its change before or after `ListLive`, both explored. Scope: one key, +//! one delete-recreate cycle — longer histories converge by per-key LWW on +//! the same machinery. + +use stateright::{Checker, Model, Property}; + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +enum WatchPhase { + /// Resume failed with CursorExpired; the live listing has not run. + Expired, + /// Live keys listed; the `k_in_list` snapshot is fixed. The request is + /// in flight to the main loop; the watch task awaits the ack. + Listed, + /// Ack received (or barrier skipped, under mutation): fallback watch + /// delivering. + Fallback, +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct St { + phase: WatchPhase, + /// Bucket truth: K live right now. Starts false (deleted during the + /// gap, marker evicted). + bucket_k: bool, + /// What the listing captured for K (meaningful from Listed onward). + k_in_list: bool, + /// The fold's view of K. Starts true (stale from before the gap). + fold_k: bool, + /// Main loop folded the synthetic deletes (and acked). + deletes_folded: bool, + /// The fallback's re-list put for K was folded (delivered only if K is + /// live when the fallback runs). + relist_put_folded: bool, +} + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +enum Act { + /// The watch task lists live keys (snapshot of bucket truth, this + /// instant) and sends the resync request. + ListLive, + /// K is re-created in the bucket (any time). + RecreateK, + /// The main loop processes the resync request: folds a synthetic delete + /// for every fold key absent from the listing, then acks. + FoldSyntheticDeletes, + /// The fallback watch starts. SHIPPED: gated on the ack (deletes + /// folded). NoAckBarrier mutation: gated only on the listing having + /// happened. + StartFallback, + /// The fallback delivers K's current value as a put (re-list or live + /// tail), folded into the fold. + DeliverRelistPut, +} + +#[derive(Clone)] +struct ResyncOrder { + ack_barrier: bool, +} + +impl Model for ResyncOrder { + type State = St; + type Action = Act; + + fn init_states(&self) -> Vec { + vec![St { + phase: WatchPhase::Expired, + bucket_k: false, + k_in_list: false, + fold_k: true, + deletes_folded: false, + relist_put_folded: false, + }] + } + + fn actions(&self, s: &St, acts: &mut Vec) { + if s.phase == WatchPhase::Expired { + acts.push(Act::ListLive); + } + if !s.bucket_k { + acts.push(Act::RecreateK); + } + if s.phase != WatchPhase::Expired && !s.deletes_folded { + acts.push(Act::FoldSyntheticDeletes); + } + if s.phase == WatchPhase::Listed { + // SHIPPED: the watch task is parked on ack_rx until the main + // loop folds the deletes. MUTATION: it proceeds immediately. + if !self.ack_barrier || s.deletes_folded { + acts.push(Act::StartFallback); + } + } + if s.phase == WatchPhase::Fallback && s.bucket_k && !s.relist_put_folded { + acts.push(Act::DeliverRelistPut); + } + } + + fn next_state(&self, s: &St, a: Act) -> Option { + let mut s = s.clone(); + match a { + Act::ListLive => { + s.k_in_list = s.bucket_k; + s.phase = WatchPhase::Listed; + } + Act::RecreateK => s.bucket_k = true, + Act::FoldSyntheticDeletes => { + // Synthetic delete for every fold key the listing lacks. + if !s.k_in_list { + s.fold_k = false; + } + s.deletes_folded = true; + } + Act::StartFallback => s.phase = WatchPhase::Fallback, + Act::DeliverRelistPut => { + s.fold_k = true; + s.relist_put_folded = true; + } + } + Some(s) + } + + fn properties(&self) -> Vec> { + let mut props: Vec> = Vec::new(); + + if self.ack_barrier { + // THE ordering theorem: every maximal run converges — including + // delete-then-recreate straddling the listing in either + // direction. (Terminal-state invariant, cycle-proof.) + props.push(Property::::always( + "every maximal run ends with the fold equal to the bucket", + |m, s| { + let mut acts = Vec::new(); + m.actions(s, &mut acts); + !acts.is_empty() || s.fold_k == s.bucket_k + }, + )); + // Witnesses: both straddles genuinely occur. + props.push(Property::::sometimes( + "recreate-after-listing converges via delete-then-put ordering", + |_, s| !s.k_in_list && s.relist_put_folded && s.fold_k && s.bucket_k, + )); + props.push(Property::::sometimes( + "recreate-before-listing converges via the listing itself", + |_, s| s.k_in_list && s.fold_k && s.bucket_k, + )); + props.push(Property::::sometimes( + "never-recreated K is reconciled away", + |_, s| !s.fold_k && !s.bucket_k && s.deletes_folded, + )); + } else { + // Without the barrier the checker must reach the lost-recreate + // divergence: re-list put folded first, synthetic delete erases + // it, K is live but the fold dropped it — terminally. + props.push(Property::::sometimes( + "HAZARD reachable: synthetic delete erases a re-created key", + |m, s| { + let mut acts = Vec::new(); + m.actions(s, &mut acts); + acts.is_empty() && s.bucket_k && !s.fold_k + }, + )); + } + + props + } +} + +fn check(ack_barrier: bool, label: &str) { + let model = ResyncOrder { ack_barrier }; + let checker = model.checker().spawn_bfs().join(); + println!( + "{label}: {} states, {} unique", + checker.state_count(), + checker.unique_state_count(), + ); + checker.assert_properties(); +} + +/// The SHIPPED handshake (fallback gated on the ack, which follows the +/// folded synthetic deletes): delete-then-recreate converges in every +/// interleaving — the comment in `applied.rs` is now a theorem. +#[test] +fn ack_barrier_makes_resync_ordering_converge() { + check(true, "resync order: ack barrier"); +} + +/// Without the barrier, the re-list put can fold before the synthetic +/// delete, which then erases a live key — the machine-checked reason the +/// handshake exists. +#[test] +fn without_ack_barrier_recreated_keys_are_lost() { + check(false, "resync order: no barrier"); +} From 76aad1f3fbd25bd5f404c63cc8e3aaeb51f984bd Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Fri, 12 Jun 2026 08:50:45 -0700 Subject: [PATCH 3/3] =?UTF-8?q?chore:=20v0.6.1=20=E2=80=94=20watch=5Fappli?= =?UTF-8?q?ed=20core=20verification=20campaign?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Machine-checked correctness for the watch_applied loop and its two critical ordering claims, plus ARCHITECTURE.md updates covering all new subsystems. **New Stateright models (3):** - `tests/model_applied.rs` — cursor-authority model: delivery → flush → transient failure → crash/restart pipeline exhaustively checked; `DropFailedBatch` and `ResumeFromMemApplied` mutations reach the restart-surviving hole, pinning both pre-fix bug classes - `tests/model_resync_order.rs` — resync ack-barrier model: proves synthetic deletes strictly precede re-list puts for every interleaving; `NoAckBarrier` mutation reaches the lost-recreate divergence - `tests/model_live_watch.rs` — floor-guard model: guarded variant proves the fold always converges; unguarded proves permanent silent divergence reachable — machine-checked record of the pre-guard design **Bug fixed (found while writing the models):** - Transient `store.apply()` failure dropped the raw batch; next successful flush advanced the cursor over a permanent hole. Fixed: re-queue the failed batch (prepend); fail-stop at 16 consecutive failures. Regression pin: `transient_store_failure_never_leaves_a_cursor_gap` **Regression pins added:** - `resync_listing_failure_is_fatal_not_degraded` — degrade-on-error resync leaves stale keys permanently; fail-stop is correct - `gapped_delivery_with_advanced_floor_trips` — 5 s timeout pins in-band detection (not the 30 s backstop) **ARCHITECTURE.md:** - Transient re-queue paragraph in cursor-after-apply section - Live retention floor guard subsection - watch_applied loop state transition table (13 rows) - Package structure updated with all new test files - MAX_STORE_APPLY_FAILURES and FLOOR_GUARD_INTERVAL in configuration Co-Authored-By: Claude Sonnet 4.6 --- ARCHITECTURE.md | 3 +++ Cargo.lock | 2 +- Cargo.toml | 2 +- 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 5e1e559..b81b8fd 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -677,8 +677,11 @@ Production code and an exhaustive model checker running the same logic closes th | `src/applied.rs` | `watch_applied` cursor-after-apply combinator, generic over `SnapshotStore`: `WatchScope`, `BatchConfig`, cursor-expired stale-key resync, `ExportRequest` handling | | `src/lib.rs` | Re-exports all public types; no logic | | `benches/` | Criterion benchmarks: snapshot write/checkpoint/load throughput, batch throughput, ACK subject parsing | +| `tests/integration.rs` | NATS JetStream backend integration suite: each test boots its own `nats-server` on a free port; covers bucket create, CAS, watch semantics, cursor resume, and delete reconciliation | +| `tests/snapshot_store.rs` | Backend-agnostic `SnapshotStore` conformance suite: every check is generic over the backend, instantiated for `AppendLogSnapshot`, `FjallSnapshot`, and `RocksDbSnapshot`; new backends get the whole suite for free | | `tests/transport.rs` | Integration: upload/download/manifest round-trip, pointer swap, prune, non-CAS fail-closed | | `tests/transport_s3.rs` | Live MinIO: CAS semantics (create, If-Match update, refusal, crash-window availability, prune) verified against a real object store | +| `tests/bootstrap.rs` | Tier-2 bootstrap proofs on live NATS + on-disk backends: exports from a live `watch_applied` loop under churn, imports as a second node, and asserts **delta-only resume** via delivery count (not just convergence — a full replay would produce the same end state) | | `tests/multi_export.rs` | Prevention proofs on live NATS + real fjall/RocksDB: slow-exporter clobber refused, crash window keeps bootstrap available, multi-SST post-compaction fidelity | | `tests/resync.rs` | Live-NATS conformance: NATS silent-clamp pinned; full expiry → resync chain e2e; no-reader divergence pinned | | `tests/model.rs` | Stateright exhaustive model (~250M states, fleet size 2–3, unbounded rounds): pointer swap theorems (no regression, no torn pair, no dangling pointer, no silent divergence, terminal liveness); mutation tests prove each protocol guard load-bearing | diff --git a/Cargo.lock b/Cargo.lock index e4e659b..ac0eaae 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -143,7 +143,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" [[package]] name = "beyond-slipstream" -version = "0.6.0" +version = "0.6.1" dependencies = [ "aho-corasick", "async-nats", diff --git a/Cargo.toml b/Cargo.toml index a301510..0c5b052 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ rust-version = "1.92" [package] name = "beyond-slipstream" -version = "0.6.0" +version = "0.6.1" edition.workspace = true license.workspace = true rust-version.workspace = true