diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f957a23..a664f55 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? @@ -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 | @@ -708,9 +709,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/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 fd35900..6ee0290 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -548,6 +548,92 @@ 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;