From 6c35d6e8b6f8ecede5cf397fb753552164a7865b Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Thu, 1 Oct 2026 15:43:27 -0700 Subject: [PATCH 1/2] Retire the "fold is a cache of NATS" model the store layer still assumed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit slipstream's design is a bounded log: folds are the replicas of record and NATS holds only the retained tail. The store layer and its docs still assumed the older model — "NATS is the source of truth, the snapshot is a cache" — and every cursor-expiry bug fixed in the previous PR was a component running on it. This removes the remaining instances. Behavior: - A delete_with_version tombstone (an empty-value Put) now reaches watchers, and so folds, as a Delete carrying the tombstone's revision. The fold held it as a present-but-empty key while get/scan/keys and the repairs' key listings treated it as deleted, so the key-listing repair deleted it and then re-listed it. entry() still exposes the raw tombstone for CAS. - watch_applied refuses ExpiryRepair::Restore/Auto without a store (a restore needs a fold to restore into; such a consumer silently held NATS's retained view), and warns when a cursor-less start with no repair armed re-lists a bucket that has already evicted current values. Docs: the store invariants are now fold + tail = truth, cursor-after-apply, and replica of record. The durability rationale and recovery runbooks no longer say "delete the snapshot and replay NATS" (on a bounded log that loses everything evicted); a lost or corrupt fold is rebuilt by importing an artifact. Tests: live tombstone-as-delete through the plain watch and the multi-prefix resume consumer (made race-free: attach the resume consumer before anything supersedes its cursor; 100/100 under parallel load); store-less restore refused. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0152kQKDdP8XRhoeqisJYpWr --- ARCHITECTURE.md | 16 ++++---- README.md | 8 ++-- src/applied.rs | 69 +++++++++++++++++++++++++++++++-- src/kv.rs | 10 ++++- src/nats.rs | 28 ++++++++++++++ src/snapshot.rs | 39 +++++++++++-------- src/snapshot_fjall.rs | 4 +- src/snapshot_rocksdb.rs | 4 +- tests/integration.rs | 85 +++++++++++++++++++++++++++++++++++++++++ 9 files changed, 227 insertions(+), 36 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f957a23..5371c36 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -123,7 +123,7 @@ Every non-`_from` watch is a **state-sync** stream (NATS `DeliverPolicy::LastPer | `VersionToken` | Opaque per-key version (NATS: 8-byte u64; FDB: 10-byte versionstamp) | Not a wall-clock timestamp; not globally ordered | | `KvEntry` | One key + value + version from a read | Not a watch event; immutable once returned | | `KvUpdate` | One watch event: `Put`, `Delete`, or `Purge` | Not a read result; carries deletes too | -| `Snapshot` | Deduplicated KV state + cursor persisted to disk | Not the source of truth; a cache of NATS | +| `Snapshot` | Deduplicated KV state + cursor persisted to disk; on a bounded log, a replica of record | Not a cache: NATS no longer holds what it evicted | | `SnapshotWriter` | Append-only log of `KvUpdate`s; no in-memory state beyond a counter | Not the in-memory cache itself | | `SnapshotStore` | Trait: the durable-fold contract — atomic `apply(batch, cursor)`, `load`, `get`, `range` | Not a serving index; stops at fold + cursor + query | | `AppendLogSnapshot` | Default `SnapshotStore`: append-only log + in-RAM fold (pure-Rust) | Not for folds larger than RAM | @@ -417,9 +417,9 @@ fn range(prefix) -> Vec; // ordered prefix scan Three invariants bind every implementation: -- **Pure function of the log.** Delete the store, replay every update with revision `> cursor`, and the state is identical. The store caches the fold; NATS is the source of truth. +- **Fold + tail = truth.** The fold at cursor C, plus a resume from C, is every write minus every real delete. A NATS cursor means every *retained* message at or below it is applied — not "the truth at C": a key whose latest write is after C arrives with the resume. - **Cursor-after-apply.** `apply` makes data and cursor durable together, so the cursor never names a revision whose data is absent — one transaction on a transactional backend, data-then-cursor on the append log (a torn write leaves data *ahead* of the cursor, which replay re-folds, never skips). -- **Snapshot is a cache.** A tail lost to power loss (under a no-sync durability mode) is rebuilt by resuming the watch from the recovered cursor. +- **Replica of record.** NATS is a bounded log and keeps only the retained tail, so a fold can't be rebuilt from NATS alone. A tail lost to power loss (under a no-sync durability mode) leaves a consistent fold at an earlier cursor, rebuilt by resuming from the recovered cursor while it is inside NATS's retention, and by the cursor-expiry repair (an artifact restore) once it isn't. A lost or corrupt fold is rebuilt by importing an artifact, never by deleting it and re-listing NATS (complete only on a bucket that has never evicted a current value). `watch_applied` is generic over `SnapshotStore`: on each flush, after `apply` returns, it hands the raw batch + post-apply cursor to `store.apply(...)` on a blocking task. The trait stops at fold + cursor + query; serving structures built from the fold (routing rings, hashrings) live in the consumer, which reads them out via `get`/`range`. @@ -473,7 +473,7 @@ write_update() compact() [blocking: replay → dedup `compact()` flushes the BufWriter first so un-checkpointed records survive. It reads the current file, replays it, writes to a same-directory tempfile (same filesystem = atomic rename, no `EXDEV`), `sync_all`s, then renames. -`checkpoint()` writes only a cursor record and calls `BufWriter::flush()` — a `write(2)` into the page cache. This survives a process crash but NOT a power loss. The only `fsync` is in `compact()`. The snapshot is a cache; a lost tail is rebuilt from a NATS scan + watch replay. +`checkpoint()` writes only a cursor record and calls `BufWriter::flush()` — a `write(2)` into the page cache. This survives a process crash but NOT a power loss. The only `fsync` is in `compact()`. A tail lost to power loss leaves a consistent fold at an earlier cursor, rebuilt by resuming from the recovered cursor while it is inside NATS's retention, and by the cursor-expiry repair (an artifact restore) once it isn't. ### Export Lease (`export_lease.rs`) @@ -610,7 +610,7 @@ async-nats ≤0.46 has a race: the server can deliver the first batch of push-co ### Why checkpoint() does not fsync? -Checkpoints are frequent (every N watch events). An fsync per checkpoint would add milliseconds of disk-sync latency to the hot watch path. Since the snapshot is a cache backed by NATS, a tail lost to power loss is rebuilt from a NATS replay — not a correctness failure. The only `fsync` is in `compact_to_file()`, where it guarantees the new compact file is durable before the atomic rename replaces the old one. +Checkpoints are frequent (every N watch events). An fsync per checkpoint would add milliseconds of disk-sync latency to the hot watch path. A tail lost to power loss leaves a consistent fold at an earlier cursor (data and cursor commit together), rebuilt by resuming from the recovered cursor while it is inside NATS's retention, and by the cursor-expiry repair (an artifact restore) once it isn't — not a correctness failure. The only `fsync` is in `compact_to_file()`, where it guarantees the new compact file is durable before the atomic rename replaces the old one. ### Why raw backend-dir artifacts instead of a logical re-encode? @@ -708,9 +708,9 @@ Production code and an exhaustive model checker running the same logic closes th | `RevisionMismatch` on CAS | Concurrent writer won the race | Re-read with `entry()`, resolve, retry | | `AlreadyExists` on `create()` | Key already present; caller's create was not exclusive | Read live value, decide whether to proceed | | Snapshot truncated tail | `load()` discards partial final record; earlier records intact | Resume from recovered cursor; tail re-folded | -| Snapshot mid-file CRC mismatch | `SnapshotError::Corrupted` | Delete snapshot, full NATS replay | -| Snapshot wrong format version | `SnapshotError::InvalidFormat` | Delete snapshot, full NATS replay | -| `compact()` I/O error | Writer poisoned; subsequent writes return `Io` | Delete snapshot, rebuild from NATS | +| Snapshot mid-file CRC mismatch | `SnapshotError::Corrupted` | Import the latest artifact (a NATS re-list loses everything evicted; complete only on a bucket that never evicted a current value) | +| Snapshot wrong format version | `SnapshotError::InvalidFormat` | Same: import the latest artifact | +| `compact()` I/O error | Writer poisoned; subsequent writes return `Io` | Reopen (the old file is intact until the atomic rename); if unreadable, import the latest artifact | | Synadia Cloud stream limit | Raw API path treats as non-fatal; verifies bucket with `get_key_value` | Non-fatal if bucket exists | | Crash between payload upload and pointer swap | Old pointer remains fully consistent; payload orphaned | Next export round publishes new pointer; stale payload pruned after grace | | Slow exporter after newer round published | `pointer_publish_allowed` returns false → `SupersededByNewer` | Lease abandoned, local artifact deleted; payload orphaned until prune | diff --git a/README.md b/README.md index b0035b8..c7609a9 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ beyond-slipstream = { version = "0.7", features = ["transport"] } # export/impor | `VersionToken` | Opaque version — NATS: u64 revision; FDB: 10-byte versionstamp | | `KvEntry` | One key + value + version from a read | | `KvUpdate` | One watch event: `Put`, `Delete`, or `Purge` | -| `Snapshot` | Deduplicated KV state + cursor at a point in time. Disk cache, not source of truth | +| `Snapshot` | Deduplicated KV state + cursor at a point in time. On a bounded log, a replica of record: NATS no longer holds what it evicted | | `SnapshotWriter` | Append-only log of `KvUpdate`s; survives restarts without a full NATS scan | | `SnapshotStore` | Trait: the durable-fold contract — `apply` (data + cursor, atomically), `load`, `get`, `range` | | `AppendLogSnapshot` | Default `SnapshotStore`: the append-only log + an in-RAM fold (pure-Rust, small state) | @@ -216,6 +216,8 @@ if let Some(snap) = snapshot::load(Path::new("/var/lib/svc/state.snap"))? { } watcher.watch_all_from(&snap.cursor, tx).await?; } else { + // Complete only if the bucket has never evicted a current value; on a + // bounded log, seed a fresh node from an artifact instead. watcher.watch_all(tx).await?; } ``` @@ -244,7 +246,7 @@ while let Some(update) = rx.recv().await { This loop has a trap: `current_cursor` must track what `cache.apply()` has consumed, not what `rx.recv()` delivered. Get it wrong and a crash skips updates on resume. [`watch_applied`](#applied-watch) runs this loop for you with that invariant enforced. -The snapshot is a cache. Delete it and the service falls back to full replay on next start. +On a bounded log the snapshot is a replica of record, not a cache. Deleting it and re-listing NATS loses everything NATS has evicted: rebuild a lost or corrupt fold by importing the latest artifact. Re-listing is complete only on a bucket that never evicts current values. ### File format @@ -276,7 +278,7 @@ pub trait SnapshotStore: Sized + Send { } ``` -Every backend keeps the same invariants: the fold is a pure function of the log (delete the store, replay from the cursor, get identical state), the cursor never names a revision whose data isn't durable (cursor-after-apply), and the store is a cache — a tail lost to power loss is rebuilt by resuming the watch. +Every backend keeps the same invariants. **Fold + tail = truth**: the fold at cursor C, plus a resume from C, is every write minus every real delete (a cursor means every *retained* message at or below it is applied). **Cursor-after-apply**: the cursor never names a revision whose data isn't durable. **Replica of record**: on a bounded log the fold can't be rebuilt from NATS alone; a tail lost to power loss is rebuilt by resuming from the recovered cursor while it is inside NATS's retention, and by the cursor-expiry repair (an artifact restore) once it isn't, and a lost or corrupt fold is rebuilt by importing an artifact, never by re-listing NATS. | Backend | When | Notes | | ------- | ---- | ----- | diff --git a/src/applied.rs b/src/applied.rs index 1c0c8c7..b60f343 100644 --- a/src/applied.rs +++ b/src/applied.rs @@ -82,8 +82,8 @@ use crate::snapshot::{SnapshotError, SnapshotStore}; /// embedded cursor is exactly the applied cursor at the moment of export. /// /// The export result (or error) comes back on `reply`; an export failure is -/// reported there and the watch keeps running (the snapshot is a cache — a -/// failed artifact is the requester's problem, not the fold's). +/// reported there and the watch keeps running (a failed export leaves the fold +/// untouched; the next export round retries). pub struct ExportRequest { /// Where the artifact directory will be created. Must not exist (or be an /// empty directory); same filesystem as the fold for cheap hardlinks. @@ -328,10 +328,29 @@ where // Repair channel, armed only when there is a repair to run AND a store to // run it against (both repairs edit the fold). The watch task sends the // repair here and waits for the ack/reply before starting its next watch. + let mode: ExpiryRepair = repair.into(); + // A restore replaces the fold, so asking for one without a fold is a + // contradiction: on a bounded log such a consumer would silently hold + // NATS's retained view (missing everything evicted) while believing it + // repairs. Refuse it up front; `ExpiryRepair::None` is the explicit way + // to accept the retained view. + if store.is_none() + && matches!( + mode, + ExpiryRepair::Restore { .. } | ExpiryRepair::Auto { .. } + ) + { + return Err(KvError::WatchError( + "ExpiryRepair::Restore / ::Auto repair the fold, so they need a store: a consumer \ + without one cannot repair a bounded log. Pass a SnapshotStore, or \ + ExpiryRepair::None to accept NATS's retained view" + .into(), + )); + } let (repair_handle, mut repairs): ( Option>, Option>>, - ) = match repair.into() { + ) = match mode { ExpiryRepair::None => (None, None), mode if store.is_some() => { let (tx, rx) = mpsc::channel(1); @@ -963,7 +982,23 @@ async fn run_watch( } seed } - None => false, + None => { + // Nothing armed: the re-list stands as the fold. Say so when + // the bucket has already evicted current values — the copy + // then lacks whatever aged out, and nothing will repair it. + if watcher + .retention() + .await? + .is_some_and(|r| !listing_is_truth(&r)) + { + warn!( + "no resume cursor and no repair armed, on a bucket that has already \ + evicted current values: the re-list lacks whatever aged out. Wire a \ + store and ExpiryRepair::Auto to seed from an artifact" + ); + } + false + } }; loop { @@ -3583,4 +3618,30 @@ mod tests { "the restored fold must equal the artifact's" ); } + + /// A restore needs a fold to restore into: `Restore`/`Auto` without a + /// store is refused up front instead of silently running on NATS's + /// retained view. + #[tokio::test] + async fn restore_without_a_store_is_refused() { + let artifact = MockRestore::new(vec![], 10, Some(&[""])); + let watcher = Arc::new(ScriptedWatcher::new(evicting(9), vec![])); + let (_sd_tx, sd_rx) = watch::channel(false); + let err = watch_applied( + watcher, + WatchScope::All, + None, + restore_mode::(artifact), + None::, + None, + BatchConfig::default(), + parse_put, + |_b: Vec>| {}, + |_| {}, + sd_rx, + ) + .await + .expect_err("Restore without a store must be refused"); + assert!(err.to_string().contains("need a store"), "{err}"); + } } diff --git a/src/kv.rs b/src/kv.rs index 1ee88bd..8ed0dc5 100644 --- a/src/kv.rs +++ b/src/kv.rs @@ -216,7 +216,11 @@ pub struct KvEntry { pub enum KvUpdate { /// Key was created or updated. Put(KvEntry), - /// Key was deleted. + /// Key was deleted — by a delete marker, or (NATS) by a + /// [`delete_with_version`](KvWriter::delete_with_version) tombstone, which + /// watches report as a delete with the tombstone's revision: the same rule + /// `get`/`scan`/`keys` apply. Only [`entry`](KvReader::entry) exposes the raw + /// tombstone. Delete { key: String, version: VersionToken }, /// Key was purged (NATS-specific: all history removed). /// Stores without purge semantics should map this to Delete. @@ -393,6 +397,10 @@ pub trait KvWriter: Send + Sync { /// CAS-gated delete: delete only if current version matches `expected`. /// Returns `RevisionMismatch` on conflict. /// Writes an empty value (logical delete) so concurrent writers get a conflict. + /// Every read path treats it as a delete — `get`/`scan`/`keys` hide it, and + /// watches (hence folds) receive a [`KvUpdate::Delete`] — except + /// [`entry`](KvReader::entry), which exposes the tombstone and its version for + /// CAS callers. async fn delete_with_version( &self, key: &str, diff --git a/src/nats.rs b/src/nats.rs index d8376b9..5c0760f 100644 --- a/src/nats.rs +++ b/src/nats.rs @@ -990,6 +990,16 @@ fn nats_entry_to_kv_update(entry: async_nats::jetstream::kv::Entry) -> KvUpdate use async_nats::jetstream::kv::Operation; let version = VersionToken::from_u64(entry.revision); match entry.operation { + // An empty value is a `delete_with_version` tombstone: a delete, with + // the tombstone's own revision — the rule `get`/`scan`/`keys` apply. + // Watches, folds, and the repairs' key listings must all agree on it + // (a fold holding the key as present-but-empty disagreed with every + // read path and made the key-listing repair delete it, then re-list + // it). `entry()` still exposes the raw tombstone for CAS callers. + Operation::Put if entry.value.is_empty() => KvUpdate::Delete { + key: entry.key, + version, + }, Operation::Put => KvUpdate::Put(KvEntry { key: entry.key, value: entry.value.to_vec(), @@ -1248,6 +1258,9 @@ fn kv_message_to_update(msg: &async_nats::Message, kv_prefix: &str) -> Option KvUpdate::Delete { key, version }, Some("PURGE") => KvUpdate::Purge { key, version }, + // An empty value is a `delete_with_version` tombstone — a delete, as + // in `nats_entry_to_kv_update`. + _ if msg.payload.is_empty() => KvUpdate::Delete { key, version }, // No header (or an explicit "PUT") is a put — the common case carries // no KV-Operation header at all. _ => KvUpdate::Put(KvEntry { @@ -1913,6 +1926,21 @@ mod tests { )); } + /// A `delete_with_version` tombstone (empty-value Put) decodes as a + /// delete carrying its revision, so the fold drops the key exactly as + /// `get`/`scan`/`keys` do. + #[test] + fn kv_message_empty_value_is_a_delete() { + let msg = raw_kv_msg("$KV.certs.node.a", Some(ACK_42), b"", None); + match kv_message_to_update(&msg, "$KV.certs.") { + Some(KvUpdate::Delete { key, version }) => { + assert_eq!(key, "node.a"); + assert_eq!(version.as_u64(), Some(42), "the tombstone's own revision"); + } + other => panic!("expected Delete, got {other:?}"), + } + } + #[test] fn kv_message_outside_keyspace_is_skipped() { // A subject-filtered consumer should never deliver this; the decode diff --git a/src/snapshot.rs b/src/snapshot.rs index 21c789e..0c9eec8 100644 --- a/src/snapshot.rs +++ b/src/snapshot.rs @@ -23,8 +23,9 @@ //! silently flatten it to 0 and break every later CAS. //! //! Used by edge/tunnel services to survive restarts without a full -//! NATS KV scan. The snapshot is a cache — delete it and the system -//! falls back to `load_all()`. +//! NATS KV scan. On a bounded log the fold is a replica of record, not a +//! cache: NATS no longer holds what it evicted, so a lost fold is rebuilt by +//! importing an artifact, not by re-listing NATS. //! //! ## Blocking I/O //! @@ -49,8 +50,9 @@ use crate::kv::{KvEntry, KvUpdate, VersionToken, WatchCursor}; const MAGIC: &[u8; 4] = b"PGSS"; // v2: Put/Delete records store the version as length-prefixed raw bytes instead // of a fixed 8-byte u64, so non-u64 tokens (e.g. FDB versionstamps) survive a -// round-trip. v1 files are rejected by `replay_log`; the snapshot is a cache, so -// a rejected file just triggers a fresh NATS scan + watch replay. +// round-trip. v1 files are rejected by `replay_log` (InvalidFormat); recover +// such a fold by importing an artifact (a NATS re-list is complete only on a +// bucket that has never evicted a current value). const FORMAT_VERSION: u16 = 2; const HEADER_LEN: usize = 6; @@ -218,10 +220,10 @@ impl SnapshotWriter { /// /// The flush is a `write(2)` into the page cache — it survives a process /// crash, but NOT power loss: there is no `fsync` on this path. The durable - /// `sync_all` happens in [`compact`](Self::compact). That's deliberate — the - /// snapshot is a cache backed by NATS and checkpoints are frequent, so an - /// fsync per checkpoint isn't worth its latency; a tail lost to power loss is - /// rebuilt from a NATS scan + watch replay. + /// `sync_all` happens in [`compact`](Self::compact). That's deliberate — + /// checkpoints are frequent and an fsync per checkpoint isn't worth its + /// latency: a tail lost to power loss leaves a consistent fold at an earlier + /// cursor, rebuilt by resuming from the recovered cursor while it is inside NATS's retention, and by the cursor-expiry repair (an artifact restore) once it isn't. /// /// Returns `true` when the log has grown past the compaction threshold /// and the caller should run [`compact`](Self::compact). Separating @@ -326,18 +328,21 @@ pub fn load(path: &Path) -> Result, SnapshotError> { /// /// ## Invariants every implementation must hold /// -/// - **Pure function of the log.** Delete the store, replay every update with -/// revision `>` the persisted cursor, and you get byte-identical state. The -/// store caches the fold; it is never the source of truth (that is NATS). +/// - **Fold + tail = truth.** The fold at cursor `C`, plus a resume from `C`, +/// is every write minus every real delete. A NATS cursor means every +/// RETAINED message at or below it is applied — not "the truth at `C`": a key +/// whose latest write is after `C` arrives with the resume. /// - **Cursor-after-apply.** A persisted cursor `C` implies every update with /// revision `≤ C` is durably folded in. [`apply`](Self::apply) writes data and /// cursor together so the cursor never names a revision whose data is absent — /// on a transactional backend in one txn, on the append log data-then-cursor /// (a torn write leaves data *ahead* of the cursor, which replay re-folds, never /// skips). -/// - **Snapshot is a cache.** Any tail lost to power loss (under a no-sync -/// durability mode) is rebuilt by resuming the watch from the recovered cursor; -/// never a correctness failure. +/// - **Replica of record.** On a bounded log NATS keeps only the retained +/// tail, so the fold can't be rebuilt from NATS alone. A tail lost to power +/// loss (under a no-sync durability mode) leaves a consistent fold at an +/// earlier cursor, rebuilt by resuming from the recovered cursor while it is inside NATS's retention, and by the cursor-expiry repair (an artifact restore) once it isn't. A lost or corrupt fold is rebuilt by +/// importing an artifact, never by re-listing NATS. /// /// ## Threading /// @@ -716,8 +721,10 @@ fn replay_log(data: &[u8]) -> Result<(HashMap, WatchCursor, boo // land as `Truncated`, so we stop and silently drop everything // after this point. Detecting it would need a framed, separately- // checksummed length (a format change). Acceptable because the - // snapshot is a cache: a short read just triggers a fuller NATS - // scan + watch replay, never data loss of record. + // result is a consistent fold at an earlier cursor — the same + // shape as a tail lost to power loss — which the resume (inside + // retention) or the cursor-expiry restore (outside it) brings + // back to the truth. clean_eof = false; break; } diff --git a/src/snapshot_fjall.rs b/src/snapshot_fjall.rs index 98403db..97b6a17 100644 --- a/src/snapshot_fjall.rs +++ b/src/snapshot_fjall.rs @@ -107,8 +107,8 @@ const DATA_BLOCK_SIZE: u32 = 16 * 1024; pub struct FjallConfig { /// `fsync` every [`apply`](SnapshotStore::apply) commit when `true`. When /// `false` (the default), commits are not fsync'd (NO_SYNC): faster, and a - /// tail lost to power loss is rebuilt by resuming the watch from the recovered - /// cursor — the snapshot is a cache. + /// tail lost to power loss leaves a consistent fold at an earlier cursor, + /// rebuilt by resuming from the recovered cursor while it is inside NATS's retention, and by the cursor-expiry repair (an artifact restore) once it isn't. pub sync: bool, /// Block-cache capacity in bytes for the LSM. fjall's own default is 32 MiB, diff --git a/src/snapshot_rocksdb.rs b/src/snapshot_rocksdb.rs index e161303..466912d 100644 --- a/src/snapshot_rocksdb.rs +++ b/src/snapshot_rocksdb.rs @@ -158,8 +158,8 @@ pub struct RocksDbConfig { /// `fsync` the WAL on every [`apply`](SnapshotStore::apply) commit when /// `true`. When `false` (the default), commits are written to the WAL but not /// fsync'd (NO_SYNC): faster, survives a process crash via WAL replay, and a - /// tail lost to power loss is rebuilt by resuming the watch from the recovered - /// cursor — the snapshot is a cache. + /// tail lost to power loss leaves a consistent fold at an earlier cursor, + /// rebuilt by resuming from the recovered cursor while it is inside NATS's retention, and by the cursor-expiry repair (an artifact restore) once it isn't. pub sync: bool, /// Block-cache capacity in bytes. RocksDB's own default is 32 MiB — the diff --git a/tests/integration.rs b/tests/integration.rs index fd35900..c1c2098 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -548,6 +548,91 @@ async fn watch_all_streams_live_updates() { assert!(matches!(&updates[2], KvUpdate::Delete { key, .. } if key == "node.a")); } +/// A `delete_with_version` tombstone (an empty-value Put, so concurrent CAS +/// writers still conflict) reaches watchers — the plain watch AND the +/// multi-prefix resume consumer — as a DELETE carrying its own revision: the +/// same rule `get`/`scan`/`keys` apply. A fold therefore drops the key +/// instead of holding it present-but-empty, and agrees with every read path +/// and with the repairs' key listings. `entry()` still exposes the raw +/// tombstone (its version is what a CAS takeover needs). +#[tokio::test] +async fn cas_tombstone_reaches_watchers_as_a_delete() { + let nats = TestNats::start().await; + let (_conn, store) = nats.store("tomb").await; + let writer = store.writer().expect("writer"); + let watcher = store.watcher().expect("watcher"); + + let v1 = writer.put("node.a", b"1").await.expect("put a"); + // The resume consumer (hand-built, multi-prefix) from just before. + let (tx_from, mut rx_from) = mpsc::channel(64); + let (w, resume) = ( + Arc::clone(&watcher), + WatchCursor::from_u64(v1.as_u64().unwrap()), + ); + let from_task = tokio::spawn(async move { w.watch_prefixes_from(&["node."], &resume, tx_from).await }); + // Attach the resume consumer BEFORE anything supersedes node.a@1: with + // history 1, once every message after the cursor is superseded the + // stream's first sequence passes it, and the (conservative) resume check + // reports CursorExpired — it can't tell supersession from eviction. + establish_watch(writer.as_ref(), &mut rx_from, "node.__ready__").await; + let (tx, mut rx) = mpsc::channel(64); + let all_task = tokio::spawn(async move { watcher.watch_all(tx).await }); + establish_watch(writer.as_ref(), &mut rx, "__ready__").await; + + assert!( + writer + .delete_with_version("node.a", &v1) + .await + .expect("cas delete") + ); + let tomb = store + .reader() + .entry("node.a") + .await + .expect("entry") + .expect("raw tombstone"); + assert!(tomb.value.is_empty(), "entry() exposes the raw tombstone"); + let rev = tomb.version.as_u64().unwrap(); + + // Skip handshake echoes; the next node.a update is the tombstone. If a + // watch ends instead, report how. + async fn next_a( + rx: &mut mpsc::Receiver, + task: tokio::task::JoinHandle>, + which: &str, + ) -> KvUpdate { + let got = timeout(Duration::from_secs(5), async { + loop { + match rx.recv().await { + Some(u) if u.key() == "node.a" => return Some(u), + Some(_) => {} + None => return None, + } + } + }) + .await + .unwrap_or_else(|_| panic!("{which}: timed out waiting for the tombstone")); + match got { + Some(u) => { + task.abort(); + u + } + None => panic!("{which}: watch ended early: {:?}", task.await), + } + } + let live = next_a(&mut rx, all_task, "watch_all").await; + assert!( + matches!(&live, KvUpdate::Delete { version, .. } if version.as_u64() == Some(rev)), + "watch: {live:?}" + ); + let from = next_a(&mut rx_from, from_task, "watch_prefixes_from").await; + assert!( + matches!(&from, KvUpdate::Delete { version, .. } if version.as_u64() == Some(rev)), + "multi-prefix resume: {from:?}" + ); + assert!(store.reader().get("node.a").await.expect("get").is_none()); +} + #[tokio::test] async fn watch_prefix_filters_by_subject() { let nats = TestNats::start().await; From d7c49322c6a4a3210279141b138ab604269cafd1 Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Thu, 1 Oct 2026 15:44:36 -0700 Subject: [PATCH 2/2] Pin the supersession-only expiry fail-stop on evicting buckets The resume check sees only the stream's first retained revision, so a node whose post-cursor messages were all superseded (every key rewritten while it was down) looks expired though nothing was evicted. On an evicting bucket with no artifact ahead of the node, the restore refuses and the watch fail-stops: safe, but unavailable until the next export round. Pinned live and documented as a failure mode. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0152kQKDdP8XRhoeqisJYpWr --- ARCHITECTURE.md | 1 + tests/eviction.rs | 53 ++++++++++++++++++++++++++++++++++++++++++++ tests/integration.rs | 3 ++- 3 files changed, 56 insertions(+), 1 deletion(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 5371c36..a664f55 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -700,6 +700,7 @@ Production code and an exhaustive model checker running the same logic closes th | `CursorExpired`, bucket keeps current values | `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 | | `CursorExpired`, bucket evicts current values | `watch_applied` restores the in-scope fold from the newest artifact and resumes from its cursor | Automatic with `ExpiryRepair::Restore`/`Auto` | | Fold holds data but no cursor (torn first checkpoint) | Repaired like an expiry at revision 0 before watching | Automatic with a repair armed | +| Every message after a node's cursor SUPERSEDED (every key rewritten while it was down), evicting bucket | The conservative resume check sees the cursor as expired though nothing was evicted; the restore then needs an artifact ahead of the node, and without one the watch fail-stops (safe, unavailable) — pinned by `tests/eviction.rs::supersession_only_restart_fail_stops_without_a_newer_artifact` | Restarts once the next export round publishes; export more often on small hot buckets. Telling supersession from eviction needs the cursor's message timestamp against `max_age` (not stored today) | | A write ages out before ANY node folds it | Every node that needed it fail-stops (no fold anywhere holds it; `tests/model_fleet.rs` proves this is the only way the fleet fail-stops) | Data loss by retention; widen retention or export more often | | Key-listing resync on a bucket that evicts current values | Refused: `KvError::WatchError` ("evicts current values"); fold untouched | Wire `ExpiryRepair::Restore`/`Auto` with an artifact source | | Restore: no artifact ahead, stale artifact, or one that doesn't cover the scope | `KvError::WatchError`, reason logged at `error`; fold untouched | Fix the export pipeline (run exports well inside `max_age` / `discard: old` turnover); restart | diff --git a/tests/eviction.rs b/tests/eviction.rs index d4c431b..9a003fd 100644 --- a/tests/eviction.rs +++ b/tests/eviction.rs @@ -615,3 +615,56 @@ async fn live_floor_guard_trip_restores_in_process() { "the restored watcher is identical to the exporter" ); } + +/// PINNED BEHAVIOR (availability, not safety): the resume check is +/// conservative — it sees only the stream's first retained revision, so when +/// every message after a node's cursor has been SUPERSEDED (every key +/// rewritten since), the cursor looks expired though nothing was evicted. +/// On an evicting bucket that routes to a restore, and with no artifact +/// ahead of the node, the watch fail-stops until the next export round — +/// safe, but unavailable. A heartbeat-style bucket (few keys, all rewritten +/// often) hits this on a short restart. Telling supersession from eviction +/// needs information the stream doesn't expose (e.g. the cursor's message +/// timestamp against `max_age`); this test flips if that lands. +#[tokio::test(flavor = "multi_thread")] +async fn supersession_only_restart_fail_stops_without_a_newer_artifact() { + let h = Harness::new(StoreConfig { + name: "routes".into(), + max_age: Some(Duration::from_secs(3600)), + ..Default::default() + }) + .await; + let w = h.bucket.writer().unwrap(); + w.put("hb.a", b"1").await.unwrap(); + let rev = w.put("hb.b", b"1").await.unwrap().as_u64().unwrap(); + let (_c, fold) = h.open("node.snap"); + let node = Node::spawn(&h.bucket, fold, None, h.auto()); + node.wait_applied(rev).await; + let artifact = node.publish(&*h.transport, h.dir.path()).await; + let rev = w.put("hb.a", b"2").await.unwrap().as_u64().unwrap(); + node.wait_applied(rev).await; + let cursor = node.stop().await.unwrap().as_u64().unwrap(); + assert!(artifact < cursor, "the newest artifact is behind the node"); + + // While the node is down, every key is rewritten — including the one + // right after its cursor. Nothing ages out (max_age is an hour). + w.put("hb.a", b"3").await.unwrap(); + w.put("hb.b", b"3").await.unwrap(); + w.put("hb.a", b"4").await.unwrap(); + assert!( + h.first_sequence().await > cursor + 1, + "supersession alone passed the cursor" + ); + + let (c, fold) = h.open("node.snap"); + let node = Node::spawn(&h.bucket, fold, Some(c), h.auto()); + let err = tokio::time::timeout(Duration::from_secs(10), node.task) + .await + .expect("fails fast") + .unwrap() + .expect_err("no artifact ahead of the node: fail-stop"); + assert!( + err.to_string().contains("not ahead of the local fold"), + "{err}" + ); +} diff --git a/tests/integration.rs b/tests/integration.rs index c1c2098..6ee0290 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -569,7 +569,8 @@ async fn cas_tombstone_reaches_watchers_as_a_delete() { Arc::clone(&watcher), WatchCursor::from_u64(v1.as_u64().unwrap()), ); - let from_task = tokio::spawn(async move { w.watch_prefixes_from(&["node."], &resume, tx_from).await }); + let from_task = + tokio::spawn(async move { w.watch_prefixes_from(&["node."], &resume, tx_from).await }); // Attach the resume consumer BEFORE anything supersedes node.a@1: with // history 1, once every message after the cursor is superseded the // stream's first sequence passes it, and the (conservative) resume check