diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 23e3ab6..b81b8fd 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -209,10 +209,23 @@ This is the lesson of Saltzer, Reed & Clark, *End-to-End Arguments in System Des **Flush triggers.** A batch flushes when any of these fires: the `window` elapses, `batch.len()` reaches `config.max`, a shutdown is signalled, or the channel closes with a pending batch (the remainder is flushed before returning). +**Transient store failures re-queue, not drop.** If `store.apply()` returns an error, the raw batch is prepended to the next flush's accumulation and the watch continues. The streak counter increments; at 16 consecutive failures the watch fail-stops with `KvError::WatchError`. Dropping the failed batch and continuing was the shipped behavior until `transient_store_failure_never_leaves_a_cursor_gap` reproduced the bug: a transient failure followed by a successful flush advanced the cursor over a hole that survived every restart, because the restart re-folds from the advanced cursor, skipping the missing range. Cursor authority requires the store's cursor and contents to advance together, always (`applied.rs:303–394`, `tests/model_applied.rs`). + **Cursor-expired resync.** On `CursorExpired` from the resume path the combinator falls back to the full-scope watch (`watch_all` / `watch_prefix` / `watch_prefixes`), whose state-sync re-list re-delivers every live key as puts. The re-list cannot cover keys deleted during the gap whose markers were evicted with the cursor, so — when the combinator is given a `KvReader` and a store — it closes that hole first: the watch task lists the bucket's live keys, hands them to the main loop, and waits for an ack; the main loop flushes, diffs the fold's in-scope keys against the listing, and runs a synthetic `KvUpdate::Delete` (unknown version — never advances the cursor) through `parse`/`apply`/store for each key that vanished; only then does the fallback watch start. That ack ordering is the invariant: a synthetic delete always precedes the re-list put for the same key, so delete-then-recreate during the gap converges. Without a reader the fallback is re-list-only and logs the possible stale keys. This is the layer the tunnel router (swap route table) and edge origin watcher (rebuild hashrings) both collapse onto: `parse` extracts the domain registration, `apply` swaps the live state, `on_applied` persists the cursor. +### Live retention floor guard (`stream_watch_floor_guarded`, `nats.rs:1042`) + +NATS does **not** error when a live consumer's position is overrun by retention — it silently skips evicted messages (the same silent-clamp behaviour that `resume_window_ok` closes at resume time). For an All-scope watch, a skipped delete marker permanently diverges the fold, with zero log lines. The floor guard closes this mid-stream: + +- **In-band (primary):** when a delivered revision jumps the frontier by more than 1, fetch `first_sequence` and call `resume_window_ok(frontier, first_seq)`. A benign interior gap — per-subject overwrites below the floor threshold — passes. Head eviction past the frontier fails the watch into the restart → `CursorExpired` → resync repair path. The check runs *before* the entry is processed: the fold never advances past unexamined evidence of loss. +- **Backstop (30 s):** when no deliveries arrive, the periodic probe catches the no-traffic case where there is no in-band evidence to act on. + +A periodic-only design was rejected by the model checker: deliveries catching the frontier up past a gap between probes erase the evidence, leaving permanent silent divergence with the guard running. The in-band check is what makes the design correct (`tests/model_live_watch.rs`). + +Scope: this guard is sound only for the All-scope unfiltered resume watch, where every stream message is deliverable and a delivery gap implies a missed message. Prefix-scoped watches deliver sparse revisions — gaps from non-matching subjects are indistinguishable from eviction gaps client-side — and retain the retention-outlives-lag operating axiom. + ### scan() and keys() via Ephemeral Push Consumer Both use `DeliverPolicy::LastPerSubject` — one ephemeral push consumer delivers the latest value per key in a single streaming operation, rather than N sequential `get()` calls. `keys()` adds `headers_only: true` so no value bytes cross the wire. @@ -330,6 +343,24 @@ Machine-checked as: _"bootstrap never silently diverges."_ Empirically pinned by | Completing | `Published` | Prune/complete failures are non-fatal | | Abandoning | `SupersededByNewer` or upload failed | CAS delete failure is non-fatal | +### watch_applied loop state transitions + +| From | Trigger | Guard | What Actually Happens | +|---|---|---|---| +| Watching | `rx.recv()` delivers a `KvUpdate` | — | `batch_high` updated; raw update pushed to `raw_batch`; `parse()` called; window timer armed on first update | +| Batching | Window elapses | — | `apply(batch)`; cursor = `batch_high`; `store.apply(raw_batch, cursor)` on blocking task; `on_applied` fired | +| Batching | `batch.len() == config.max` | — | Same as window flush | +| Flush | `store.apply()` succeeds | streak < 16 | Streak reset; `on_applied` fired; continue watching | +| Flush | `store.apply()` transient error | streak < 16 | Raw batch prepended to next `raw_batch`; streak++; warn; continue | +| Flush | `store.apply()` error | streak ≥ 16 | `KvError::WatchError` returned; watch fail-stops | +| Watching | Watch task returns `CursorExpired` | first_seq > cursor+1 | Resync (if reader wired): list keys → synthetic deletes → ack; then `watch_all()` fallback | +| Resync | `reader.keys()` fails | — | Fatal: `KvError::WatchError`; restart re-runs the full resume → expiry → resync path | +| Resync | Fold range fails | — | Fatal: same | +| Watching | `GuardTrip` (in-band gap or 30s backstop) | `!resume_window_ok(frontier, first_seq)` | Watch task fails; `KvError::WatchError` surfaces; caller's restart hits `CursorExpired` and runs resync | +| Watching | `ExportRequest` | — | Flush pending batch; `store.export_to()` on blocking task; reply on oneshot; continue | +| Any | Shutdown signal | — | Flush pending batch; abort watch task; return applied cursor | +| Any | Channel closes cleanly | — | Flush remaining batch; return applied cursor | + ## Snapshot Subsystem ### The durable-fold trait @@ -604,6 +635,10 @@ Production code and an exhaustive model checker running the same logic closes th | Failure | What Actually Happens | Recovery | | ------------------------------------ | --------------------------------------------------------------------------- | ------------------------------------------------- | +| Transient `store.apply()` failure | Raw batch re-queued at front; next flush commits cumulatively; streak counter incremented | Automatic up to 16 consecutive failures | +| 16 consecutive `store.apply()` failures | Watch fail-stops: `KvError::WatchError` returned | Restart re-folds from last good cursor; investigate store (disk, permissions) | +| Live watch retention overrun | Floor guard detects in-band (gapped delivery) or via 30 s backstop; watch fails with `KvError::WatchError` | Caller restarts; resume hits `CursorExpired`; resync repairs stale keys | +| Resync `reader.keys()` failure | Watch fail-stops (fail-stop, not degrade — degrade semantics proven to leave stale keys permanently) | Restart re-runs full resume → expiry → resync | | `CursorExpired` | `watch_applied` lists live keys, diffs fold, applies synthetic deletes, then falls back to state-sync re-list | Automatic; raw callers use `stale_keys()` manually | | `WatchError` | Watch stream dropped (NATS restart, reconnect) | Re-subscribe | | `Timeout` on any NATS op | CLOSE_WAIT half-dead TCP parks the `await` without this guard | Call `shutdown()` + `connect()` | @@ -642,12 +677,18 @@ 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/transport.rs` | Integration: upload/download/manifest round-trip, pointer swap, prune | -| `tests/transport_s3.rs` | Live MinIO / S3: CAS semantics (create, update, precondition) verified against real object stores | -| `tests/multi_export.rs` | Concurrent exporters, lease contention, fjall + RocksDB backends, multi-file artifacts under churn | -| `tests/resync.rs` | Cursor expiry, stale-key resync, NATS silent-clamp pinning across 3 backends × reader modes | -| `tests/model.rs` | Stateright exhaustive model: proves pointer monotonicity, no dangling pointer, no silent divergence; mutation tests prove each protocol guard is load-bearing | -| `tests/common/mod.rs` | Shared test helpers: ephemeral NATS server, temp dirs, assertion helpers | +| `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 | +| `tests/model_applied.rs` | Stateright cursor-authority model: delivery → flush → transient failure → crash/restart; `DropFailedBatch` and `ResumeFromMemApplied` mutations pin both pre-fix bug classes | +| `tests/model_resync_order.rs` | Stateright resync ack-barrier model: proves synthetic deletes strictly precede re-list puts; `NoAckBarrier` mutation reaches lost-recreate divergence | +| `tests/model_live_watch.rs` | Stateright floor-guard model: guarded variant proves fold always converges; unguarded variant pins permanent silent divergence as the machine-checked record of the pre-guard code | +| `tests/common/mod.rs` | Shared test helpers: ephemeral NATS server, MinIO harness, `ManifestPutCrash` transport injection | ## Configuration @@ -685,6 +726,11 @@ Credentials priority: `creds` > `creds_file` > URL-embedded > no auth. | `window` | 10 ms | Max time a batch stays open before flush | | `max` | 100 | Max updates per batch before early flush | +| Constant | Value | Effect | +|---|---|---| +| `MAX_STORE_APPLY_FAILURES` | 16 | Consecutive transient store-apply failures before the watch fail-stops. Prevents the re-queued batch backlog from growing unboundedly under a persistently failing store. | +| `FLOOR_GUARD_INTERVAL` | 30 s | Period of the no-traffic backstop probe in the live floor guard. Primary detection is in-band (gapped delivery); this only fires when no deliveries arrive. | + ### ObjectStoreTransport / ExportLease | Setting | Default | Effect | 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 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"); +}