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"); +}