From bdf6d284e7bb1bdec906a9971bb906ceb5a2d93f Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Sat, 3 Oct 2026 11:31:02 -0700 Subject: [PATCH] Fix three cursor-expiry repair gaps: eviction history, supersession-only gaps, restore memory - A fold now remembers, on its cursor, that it saw its bucket evict current values. Turning max_age / discard:old off later (or recreating the bucket) no longer makes the key listing look trustworthy, so Auto keeps restoring instead of deleting keys that aged out. The memo is carried in artifacts. - On a max_age-only bucket, a resume whose gap was all overwrites (nothing young enough to have aged out) resumes from the first retained revision instead of fail-stopping for want of a newer artifact. Uses the cursor message's server timestamp and the server's clock, with half of max_age as headroom. - The restore no longer holds every live key name: rewrites spill to scratch, only delete candidates stay in memory, and the listing is streamed past them (and re-checked against retention, which the model checker showed is needed once the listing moves after the diff). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0152kQKDdP8XRhoeqisJYpWr --- ARCHITECTURE.md | 15 +- Cargo.lock | 1 + Cargo.toml | 3 + README.md | 6 + src/applied.rs | 35 ++- src/artifact.rs | 13 +- src/kv.rs | 235 +++++++++++++- src/nats.rs | 124 +++++++- src/protocol.rs | 85 ++++- src/repair.rs | 632 ++++++++++++++++++++++++++++---------- src/snapshot.rs | 27 +- src/snapshot_fjall.rs | 18 +- src/snapshot_rocksdb.rs | 18 +- src/transport.rs | 4 +- tests/eviction.rs | 190 ++++++++++-- tests/model.rs | 1 + tests/model_fleet.rs | 6 +- tests/model_live_watch.rs | 1 + tests/model_repair.rs | 81 ++++- 19 files changed, 1222 insertions(+), 273 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 3153312..fca4019 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -234,8 +234,8 @@ The one exception: an update carrying the **unknown** version (an unparseable AC **Cursor-expiry repair.** The cursor expires at resume (`CursorExpired` from the `*_from` watch) or mid-watch (the live floor guard ends an All-scope watch with `CursorExpired` when retention overruns it). Both take one path. Correct means **every write minus every real delete** — not "whatever NATS still lists"; the two agree only when retention never evicts current values. The repair (`ExpiryRepair`, `src/repair.rs`) is chosen accordingly: -- **Key-listing diff** (`Relist`; `Auto` on a bucket that keeps current values). The watch task lists the bucket's live keys, hands them to the main loop, and waits for an ack; the main loop 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 full-scope re-list (`watch_all` / `watch_prefix` / `watch_prefixes`) 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. **Refused** — the watch fails — when the watcher reports retention that evicts current values (`KvWatcher::retention`, read live from the stream config): there it deletes valid keys. `tests/eviction.rs` reproduced exactly that against a live server before the fix. -- **Artifact restore** (`Restore`; `Auto` on a bucket that evicts current values or can't say). The watch task peeks the newest artifact's manifest and checks it before downloading: it must cover the watch scope (recorded in the manifest by the exporting `watch_applied`), and pass `protocol::restore_allowed` — strictly ahead of the local fold, cursor still inside retention. It then downloads and imports it into scratch as a temporary fold of the consumer's backend, re-checks what actually arrived, lists the bucket's live keys, and hands both to the main loop. The main loop re-checks "ahead" against its own applied cursor (a mid-watch expiry delivered past the resume cursor), diffs the in-scope fold against the artifact (keys only), and folds the difference through `parse`/`apply`/store in `max`-sized chunks, then advances the cursor to the artifact's in one final commit and replies. The diff takes an artifact entry only when it is newer than the local one, and deletes a key the artifact lacks only when the bucket does not list it live. An artifact is complete *together with the log after its cursor* — a NATS cursor C means every retained message ≤ C is applied, not "the truth at C": a key whose latest write is after C may be missing or older in an artifact exported mid catch-up — and in both skipped cases the key's latest write is after C, so the resume delivers it. The consumer therefore never sees a key move backward or a live key deleted, not even transiently. The watch resumes from that cursor (which can itself expire later and go round again, each time only for a strictly newer artifact). A failed check fails the watch and logs the reason at `error`: a stale artifact has no safe recovery. +- **Key-listing diff** (`Relist`; `Auto` on a bucket that keeps current values). The watch task lists the bucket's live keys, hands them to the main loop, and waits for an ack; the main loop 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 full-scope re-list (`watch_all` / `watch_prefix` / `watch_prefixes`) 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. **Refused** — the watch fails — when the watcher reports retention that evicts current values (`KvWatcher::retention`, read live from the stream config), or the fold remembers seeing it do so (below): there it deletes valid keys. `tests/eviction.rs` reproduced exactly that against a live server before the fix. +- **Artifact restore** (`Restore`; `Auto` on a bucket that evicts current values or can't say). The watch task peeks the newest artifact's manifest and checks it before downloading: it must cover the watch scope (recorded in the manifest by the exporting `watch_applied`), and pass `protocol::restore_allowed` — strictly ahead of the local fold, cursor still inside retention. It then downloads and imports it into scratch as a temporary fold of the consumer's backend, re-checks what actually arrived, and hands it to the main loop. The main loop re-checks "ahead" against its own applied cursor (a mid-watch expiry delivered past the resume cursor) and diffs the in-scope fold against the artifact in one key-ordered pass: keys the artifact's entry should replace are spilled to a scratch file next to the imported artifact (`RestoredFold::scratch_in`), and local keys the artifact lacks are collected as delete candidates. Only if there are candidates does it stream the bucket's live keys past them (`KvReader::for_each_key`) and then re-read retention: the listing vouches for writes after the artifact's cursor only while the log still holds all of them, so if retention passed that cursor meanwhile the restore is refused before anything is folded (`tests/model_repair.rs`, `NoListingRecheck`). It then folds the difference through `parse`/`apply`/store in `max`-sized chunks, read back from the spill, then advances the cursor to the artifact's in one final commit and replies. Memory is the delete candidates — the keys deleted between the two cursors — never the fold, the artifact, or the bucket's listing, so a billion-key fold (or a fresh node seeding from an artifact) restores in bounded memory. The diff takes an artifact entry only when it is newer than the local one, and deletes a key the artifact lacks only when the bucket does not list it live. An artifact is complete *together with the log after its cursor* — a NATS cursor C means every retained message ≤ C is applied, not "the truth at C": a key whose latest write is after C may be missing or older in an artifact exported mid catch-up — and in both skipped cases the key's latest write is after C, so the resume delivers it. The consumer therefore never sees a key move backward or a live key deleted, not even transiently. The watch resumes from that cursor (which can itself expire later and go round again, each time only for a strictly newer artifact). A failed check fails the watch and logs the reason at `error`: a stale artifact has no safe recovery. - **None** (`None`, or no store): the re-list alone, with a warning that deleted keys may persist. Whichever runs, the main loop first **drains** the watch channel: a mid-watch expiry leaves delivered updates buffered there, and the repair must fold them first — one applied after the repair would resurrect pre-repair state and could move the cursor backward. The watch task is parked on the ack/reply, so the drain takes exactly that backlog. @@ -244,6 +244,10 @@ A cursor-less start is repaired before it watches in two cases. A fold that hold The listing counts as the truth — "not listed" means "deleted" — when retention never evicts current values, or when it can but has never evicted anything yet (first retained revision ≤ 1). Otherwise the key-listing diff is refused. +**Retention can be edited, so the fold remembers what it saw.** A key that aged out while `max_age` or `discard: old` was in force is missing from the listing forever after, even once the operator turns eviction off or recreates the bucket — and the fold, the only place it still exists, would have it deleted by a diff against that listing. So the first time the watch task reads retention and finds it has evicted (eviction on and the log past its first revision), it remembers: the main loop commits a memo on the fold's cursor (`WatchCursor::to_bytes`: a 12- or 20-byte encoding, longer than any bare version; a cursor without the memo keeps the old encoding byte for byte), every later cursor carries it, and so does every artifact exported from the fold, so a node restored from one inherits it. A fold with the memo never treats the listing as the truth again (`listing_is_truth`'s `seen_evicting`; `Snapshot::stale_keys` reads it too). Retention is read at every watch start, at every expiry, and every 60 s during a live watch while a key-listing repair could still be planned (`Relist`/`Auto`) and the memo is unset. The axiom: a fold reads retention while eviction is in force. Eviction that starts, drops a key, and is turned off again entirely between two reads (within a minute, or while the node is down) is not seen. `tests/model_repair.rs` (`eviction_turned_off_repair_steps_are_correct`, mutation `NoEvictionMemo`) and `tests/eviction.rs::eviction_turned_off_never_trusts_the_listing_again` (live) pin it. + +**A supersession-only gap resumes** (`NatsKvWatcher::resume_point`). The resume check sees only the stream's first retained revision, so when every message after a node's cursor was superseded while it was down — a small hot bucket on a short restart — the cursor looks expired though nothing aged out, and on an evicting bucket the restore needs an artifact newer than the node. A cursor with the memo also keeps the server time its message was written. When the window check fails, the resume reads the stream's config and the server's clock (`ts` in the raw stream-info reply, NATS 2.10+); if age is the bucket's only way to evict a current value (`max_age`, no per-message TTLs, no `discard: old` limit) and the cursor's message is at most half of `max_age` old (`protocol::gap_holds_no_eviction`: half is headroom for clock steps and leader changes), every later message is younger still, none has aged out, and each missing one was superseded by a later write the resume delivers — so it resumes from the first retained revision instead. Assumes the config didn't switch from a `discard: old` limit to age-only while the node was down. + Before diffing, both repairs flush until the store holds every delivered update: a batch re-queued by a transient store failure is invisible to a diff of the store, and a re-queued put of a key deleted during the gap would otherwise commit after the diff and resurrect it. **The key-listing repair's one axiom.** Between taking the listing and the full re-list delivering what the bucket holds, retention must not evict a delete marker (axiom 6 of `tests/model.rs`, narrowed to this window). On a bucket that keeps current values, the only such eviction is an admin purge of markers, and NATS's `purge_deletes` keeps markers younger than 30 minutes by default. `tests/model_repair.rs` drops the axiom and the checker reaches the purge-inside-the-window trace; the artifact restore resumes on a floor-guarded watch and holds without it. @@ -358,9 +362,9 @@ artifact_revision > local_revision && resume_window_ok(artifact_revision, first_ Machine-checked as: _"bootstrap never silently diverges from every write minus every real delete"_ on an evicting bucket (`tests/model.rs`), and _"a repair never moves the fold backward"_ (`tests/model_live_watch.rs`, where removing the guard is a mutation the checker catches). -**`listing_is_truth(evicts_current_values, first_revision) → bool`** and **`plan_repair(mode, listing_truth) → RepairPlan`** +**`listing_is_truth(evicts_current_values, first_revision, seen_evicting) → bool`**, **`listing_truth(retention, seen_evicting)`** and **`plan_repair(mode, listing_truth) → RepairPlan`** -Whether "not listed" means "deleted": yes when retention never evicts current values, or when it can but has evicted nothing yet (first retained revision ≤ 1). The planner turns that (or `None`, a backend that can't say) and the armed `ExpiryRepair` into the repair: `Relist` where the listing isn't the truth is `RefuseRelist`; `Auto` re-lists only on `Some(true)` and restores otherwise. +Whether "not listed" means "deleted": yes when retention never evicts current values, or when it can but has evicted nothing yet (first retained revision ≤ 1) — and never once the fold has seen the answer be no (`seen_evicting`, the cursor's memo). The planner turns that (or `None`, a backend that can't say) and the armed `ExpiryRepair` into the repair: `Relist` where the listing isn't the truth is `RefuseRelist`; `Auto` re-lists only on `Some(true)` and restores otherwise. **`cursorless_start_needs_repair(mode, fold_has_data, listing_truth) → bool`** @@ -729,7 +733,8 @@ 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) | +| Every message after a node's cursor SUPERSEDED (every key rewritten while it was down), evicting bucket | On an age-only (`max_age`) bucket within half of `max_age` of the cursor's message: resumes from the first retained revision (`tests/eviction.rs::supersession_only_restart_resumes_on_a_max_age_bucket`). Otherwise (a `discard: old` limit or per-message TTLs, an older cursor, or one without the memo) the cursor reads as expired, the restore needs an artifact ahead of the node, and without one the watch fail-stops (safe, unavailable) | Restarts once the next export round publishes; export more often on small hot buckets | +| Eviction turned off (or the bucket recreated) after it dropped keys | The fold remembers it saw eviction (cursor memo) and never trusts the key listing again: `Auto` restores, `Relist` is refused | Automatic with `ExpiryRepair::Auto`; an artifact source is needed for good | | 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/Cargo.lock b/Cargo.lock index 1bd5259..44c3cd6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -162,6 +162,7 @@ dependencies = [ "tar", "tempfile", "thiserror 2.0.18", + "time", "tokio", "tracing", "url", diff --git a/Cargo.toml b/Cargo.toml index 4d887c9..757fca9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,6 +34,9 @@ serde_json = "1" tar = { version = "0.4.46", optional = true } tempfile = "3" thiserror = "2" +# Parses the server time in a raw JetStream stream-info reply (`ts`), for the +# supersession-only resume check. Already in the tree via async-nats. +time = { version = "0.3", features = ["parsing"] } tokio = { version = "1", features = ["macros", "rt", "sync", "time"] } tracing = "0.1" url = "2" diff --git a/README.md b/README.md index 2334549..c26fe92 100644 --- a/README.md +++ b/README.md @@ -374,6 +374,12 @@ A restore only accepts an artifact that is ahead of the local fold, covers the w A node with no cursor on a bucket that has already evicted current values seeds from the artifact too, since a re-list alone would miss whatever aged out. +Turning eviction off later doesn't make the key list trustworthy again: keys that aged out while it was on stay missing from it. A node that ever saw its bucket evict remembers that in its saved cursor (and in every artifact it exports), so `Auto` keeps restoring and `Relist` keeps refusing. + +On a bucket that evicts only by `max_age`, a node restarted within half of `max_age` resumes even when every message after its cursor was overwritten while it was down: none of those can have aged out, so there is nothing to restore. + +A restore holds only the keys it will delete in memory; the keys it rewrites are spilled to the scratch directory, so it works at any fold size. + `Some(reader)` from older code still compiles and means `Relist`. `None` (or no `store`) falls back to the re-list alone and warns that deleted keys may persist. ## NATS mapping diff --git a/src/applied.rs b/src/applied.rs index 7db2fe5..d1eb817 100644 --- a/src/applied.rs +++ b/src/applied.rs @@ -434,6 +434,8 @@ pub(crate) struct Fold { store_fail_streak: u32, // A batch opened since the last flush (the window timer is armed). open: bool, + // `applied` gained the eviction memo and the store doesn't have it yet. + memo_dirty: bool, } impl Fold @@ -467,6 +469,7 @@ where applied, store_fail_streak: 0, open: false, + memo_dirty: false, } } @@ -528,7 +531,11 @@ where // Nothing received since the last flush → nothing to do at all. // (`raw` can be non-empty with no cursor advance only via a repair's // corrections, or a re-queued batch.) - if self.batch.is_empty() && self.raw.is_empty() && self.batch_high.is_none() { + if self.batch.is_empty() + && self.raw.is_empty() + && self.batch_high.is_none() + && !self.memo_dirty + { return Ok(()); } if !self.batch.is_empty() { @@ -546,13 +553,16 @@ where } let advanced = !self.batch_high.is_none(); if advanced { - self.applied = self.batch_high.clone(); + // The eviction memo never clears: a stream position doesn't carry + // it, so it rides over from the cursor it replaces. + let seen = self.applied.seen_evicting(); + self.applied = self.batch_high.clone().remembering(seen); } // A cursor-only advance (nothing raw to fold) still commits to the // store: a restore's final step moves the cursor to the artifact's // with no updates left. In the steady state every advance carries raw // updates, so this changes nothing there. - if (!self.raw.is_empty() || advanced) + if (!self.raw.is_empty() || advanced || self.memo_dirty) && let Some(mut st) = self.store.take() { let raw = std::mem::take(&mut self.raw); @@ -579,6 +589,7 @@ where Ok((st, _raw, Ok(()))) => { self.store = Some(st); self.store_fail_streak = 0; + self.memo_dirty = false; } Ok((st, raw, Err(e))) => { self.store_fail_streak += 1; @@ -618,6 +629,19 @@ where Ok(()) } + /// Remember, durably, that the bucket's retention evicts current values: + /// from now on every cursor this fold commits carries the memo, and so + /// does every artifact exported from it. Commits at once, even with + /// nothing else to fold. + pub(crate) async fn remember_evicting(&mut self) -> Result<(), KvError> { + if self.applied.seen_evicting() { + return Ok(()); + } + self.applied = self.applied.clone().remembering(true); + self.memo_dirty = self.store.is_some(); + self.flush().await + } + /// Flush until the store holds every update received: a transient store /// failure re-queues the batch, so retry; a persistent one hits the /// flush's own fail-stop bound. @@ -2562,11 +2586,12 @@ mod tests { run.seen, vec![ ("node.changed".into(), false), - ("node.gone".into(), true), ("node.late".into(), false), + ("node.gone".into(), true), ("node.tail".into(), false), ], - "the diff (key order), then the delta from the artifact's cursor" + "the diff (rewrites in key order, then deletes), then the delta from the \ + artifact's cursor" ); assert_eq!( run.cursors, diff --git a/src/artifact.rs b/src/artifact.rs index f5e549b..b3c0bee 100644 --- a/src/artifact.rs +++ b/src/artifact.rs @@ -39,7 +39,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; -use crate::kv::{VersionToken, WatchCursor}; +use crate::kv::WatchCursor; use crate::snapshot::SnapshotError; /// Version of the artifact layout itself (`MANIFEST.json` schema + `data/` @@ -174,18 +174,17 @@ pub(crate) fn hex_decode(s: &str) -> Option> { } fn cursor_to_hex(cursor: &WatchCursor) -> String { - hex_encode(cursor.version().as_bytes()) + hex_encode(&cursor.to_bytes()) } fn cursor_from_hex(s: &str) -> Result { let bytes = hex_decode(s).ok_or_else(|| invalid(format!("malformed cursor_hex: {s:?}")))?; - let token = VersionToken::from_raw(&bytes).ok_or_else(|| { + WatchCursor::from_bytes(&bytes).ok_or_else(|| { invalid(format!( - "cursor_hex decodes to {} bytes, exceeds version token capacity", + "cursor_hex decodes to {} bytes, not a cursor this build can read", bytes.len() )) - })?; - Ok(WatchCursor::from_version(token)) + }) } // --------------------------------------------------------------------------- @@ -814,7 +813,7 @@ mod tests { fn manifest_round_trips_fdb_width_cursor() { // 10-byte tokens have no u64 form; the hex path must carry them intact. let raw = [1u8, 2, 3, 4, 5, 6, 7, 8, 9, 10]; - let cursor = WatchCursor::from_version(VersionToken::from_raw(&raw).unwrap()); + let cursor = WatchCursor::from_version(crate::kv::VersionToken::from_raw(&raw).unwrap()); let dir = TempDir::new().unwrap(); write_manifest(dir.path(), &manifest_with(vec![], cursor.clone())).unwrap(); let got = read_manifest(dir.path()).unwrap(); diff --git a/src/kv.rs b/src/kv.rs index 2e4513b..439f0ef 100644 --- a/src/kv.rs +++ b/src/kv.rs @@ -7,39 +7,144 @@ use tokio::sync::mpsc::Sender; /// Backends store whatever they need to resume (NATS: u64 revision). /// Callers should treat this as opaque and only pass it back to /// `watch_all_from` / `watch_prefix_from`. -#[derive(Debug, Clone, Default, PartialEq, Eq)] -pub struct WatchCursor(VersionToken); +/// +/// A cursor that [`watch_applied`](crate::watch_applied) produced may also +/// remember that the bucket's retention was once seen evicting current +/// values. That memo outlives the retention setting: a key that aged out +/// while it was in force is missing from the bucket's listing forever after, +/// so the cursor-expiry repair must never again treat the listing as the +/// truth. A store persisting a cursor should write +/// [`to_bytes`](Self::to_bytes) and read [`from_bytes`](Self::from_bytes); +/// [`as_u64`](Self::as_u64) keeps only the position. +#[derive(Clone, Default, PartialEq, Eq)] +pub struct WatchCursor { + version: VersionToken, + seen_evicting: bool, +} + +impl fmt::Debug for WatchCursor { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + if self.seen_evicting { + write!(f, "WatchCursor({:?}, seen evicting)", self.version) + } else { + write!(f, "WatchCursor({:?})", self.version) + } + } +} + +/// The longest encoding [`WatchCursor::to_bytes`] produces. +pub(crate) const MAX_CURSOR_BYTES: usize = CURSOR_MEMO_LEN + 8; + +/// A cursor with its memo: the version padded to the token's 10-byte +/// capacity, its length, a flags byte, and — when the flags say so — the +/// server time its message was written (u64 LE Unix milliseconds). Longer +/// than any bare version, so the encodings never collide, and a cursor +/// without the memo keeps the pre-memo encoding byte for byte (older builds +/// read it; they refuse this one as too long rather than misread it). +const CURSOR_MEMO_LEN: usize = 12; +const CURSOR_SEEN_EVICTING: u8 = 0b01; +const CURSOR_WRITTEN_AT: u8 = 0b10; impl WatchCursor { /// No cursor — forces a full watch on next connect. pub fn none() -> Self { - Self(VersionToken::unknown()) + Self::default() } /// Returns true if this cursor has no position (will trigger full watch). pub fn is_none(&self) -> bool { - self.0.is_unknown() + self.version.is_unknown() } /// Create a cursor from a version token. pub fn from_version(token: VersionToken) -> Self { - Self(token) + Self { + version: token, + seen_evicting: false, + } } /// Create a cursor from a u64 revision (convenience for NATS). pub fn from_u64(rev: u64) -> Self { - Self(VersionToken::from_u64(rev)) + Self::from_version(VersionToken::from_u64(rev)) + } + + /// The cursor as a store should persist it, memo included. The time its + /// message was written is kept only alongside the memo: it matters only + /// on a bucket that evicts (`NatsKvWatcher`'s resume check). + pub fn to_bytes(&self) -> Vec { + let v = self.version.as_bytes(); + if !self.seen_evicting { + return v.to_vec(); + } + let mut out = vec![0u8; CURSOR_MEMO_LEN]; + out[..v.len()].copy_from_slice(v); + out[10] = v.len() as u8; + out[11] = CURSOR_SEEN_EVICTING; + if let Some(ms) = self.version.written_ms() { + out[11] |= CURSOR_WRITTEN_AT; + out.extend_from_slice(&ms.to_le_bytes()); + } + out + } + + /// Read back [`to_bytes`](Self::to_bytes). `None` for bytes no build + /// wrote: a length or flags this build doesn't know. + pub fn from_bytes(bytes: &[u8]) -> Option { + if bytes.len() <= 10 { + return VersionToken::from_raw(bytes).map(Self::from_version); + } + let (head, tail) = bytes.split_at_checked(CURSOR_MEMO_LEN)?; + let (len, flags) = (head[10] as usize, head[11]); + if len > 10 || flags & !(CURSOR_SEEN_EVICTING | CURSOR_WRITTEN_AT) != 0 { + return None; + } + let mut version = VersionToken::from_raw(&head[..len])?; + match (flags & CURSOR_WRITTEN_AT != 0, tail.len()) { + (false, 0) => {} + (true, 8) => { + let ms = u64::from_le_bytes(tail.try_into().ok()?); + if ms == 0 { + return None; + } + version = version.written_at(ms); + } + _ => return None, + } + Some(Self { + version, + seen_evicting: flags & CURSOR_SEEN_EVICTING != 0, + }) + } + + /// The server time this cursor's message was written, when known (Unix + /// milliseconds). + pub(crate) fn written_ms(&self) -> Option { + self.version.written_ms() + } + + /// The bucket's retention was seen evicting current values while this + /// cursor's fold — or one it was restored from — tracked it. + pub(crate) fn seen_evicting(&self) -> bool { + self.seen_evicting + } + + /// This cursor, remembering that retention evicts current values (or + /// unchanged when `seen` is false: the memo never clears). + pub(crate) fn remembering(mut self, seen: bool) -> Self { + self.seen_evicting |= seen; + self } /// Try to extract as u64 revision. #[must_use] pub fn as_u64(&self) -> Option { - self.0.as_u64() + self.version.as_u64() } /// Access the underlying version token. pub(crate) fn version(&self) -> &VersionToken { - &self.0 + &self.version } /// The cursor's position for ordering — the artifact pointer swap, the @@ -123,10 +228,31 @@ pub struct Retention { /// /// Stored inline (no heap allocation) — fits up to 10 bytes, which covers /// every current backend. -#[derive(Clone, Default, PartialEq, Eq, Hash)] +/// +/// A token read off a watch may also carry the server time its message was +/// written (NATS: the message timestamp). That time is not part of the +/// version: equality and hashing ignore it, and it isn't in +/// [`as_bytes`](Self::as_bytes). +#[derive(Clone, Default)] pub struct VersionToken { len: u8, buf: [u8; 10], + // Unix milliseconds of the message's server timestamp; 0 = unknown. + written_ms: u64, +} + +impl PartialEq for VersionToken { + fn eq(&self, other: &Self) -> bool { + self.as_bytes() == other.as_bytes() + } +} + +impl Eq for VersionToken {} + +impl std::hash::Hash for VersionToken { + fn hash(&self, state: &mut H) { + self.as_bytes().hash(state); + } } impl fmt::Debug for VersionToken { @@ -157,7 +283,24 @@ impl VersionToken { pub fn from_u64(rev: u64) -> Self { let mut buf = [0u8; 10]; buf[..8].copy_from_slice(&rev.to_be_bytes()); - Self { len: 8, buf } + Self { + len: 8, + buf, + written_ms: 0, + } + } + + /// This version, carrying the server time its message was written + /// (Unix milliseconds). + pub(crate) fn written_at(mut self, unix_ms: u64) -> Self { + self.written_ms = unix_ms; + self + } + + /// The server time this version's message was written, when the backend + /// reported it (Unix milliseconds). + pub(crate) fn written_ms(&self) -> Option { + (self.written_ms != 0).then_some(self.written_ms) } /// Create from FDB versionstamp (10 bytes). @@ -170,7 +313,11 @@ impl VersionToken { /// second constructor. #[cfg(test)] pub(crate) fn from_fdb_versionstamp(vs: &[u8; 10]) -> Self { - Self { len: 10, buf: *vs } + Self { + len: 10, + buf: *vs, + written_ms: 0, + } } /// Try to extract as u64 (for NATS compatibility). @@ -208,7 +355,11 @@ impl VersionToken { let len = bytes.len() as u8; let mut buf = [0u8; 10]; buf[..len as usize].copy_from_slice(bytes); - Some(Self { len, buf }) + Some(Self { + len, + buf, + written_ms: 0, + }) } } @@ -272,6 +423,23 @@ pub trait KvReader: Send + Sync { /// Get all keys matching a prefix. Returns keys only, not values. async fn keys(&self, prefix: &str) -> Result, KvError>; + /// Hand every key [`keys`](Self::keys) would list to `f`, one at a time, + /// without collecting them: the cursor-expiry restore checks a large + /// bucket's listing this way rather than holding every key name at once. + /// + /// The provided implementation collects [`keys`](Self::keys) and iterates + /// it. A backend that can stream should override it. + async fn for_each_key( + &self, + prefix: &str, + f: &mut (dyn FnMut(String) + Send), + ) -> Result<(), KvError> { + for key in self.keys(prefix).await? { + f(key); + } + Ok(()) + } + /// Get multiple entries by prefix. Useful for bulk loading. async fn scan(&self, prefix: &str) -> Result, KvError>; @@ -451,6 +619,49 @@ pub trait KvPurge: KvWriter { mod tests { use super::*; + /// A cursor without the eviction memo encodes exactly as before the memo + /// existed (older builds keep reading it); with the memo it round-trips + /// and is longer than any bare version; unknown encodings are refused. + #[test] + fn cursor_bytes_round_trip_and_stay_compatible() { + let c = WatchCursor::from_u64(42); + assert_eq!(c.to_bytes(), VersionToken::from_u64(42).as_bytes()); + assert_eq!(WatchCursor::from_bytes(&c.to_bytes()), Some(c.clone())); + assert!(WatchCursor::from_bytes(&[]).is_some_and(|c| c.is_none())); + + let m = c.clone().remembering(true); + assert!(m.seen_evicting() && m.as_u64() == Some(42)); + assert!(m.to_bytes().len() > 10); + assert_eq!(WatchCursor::from_bytes(&m.to_bytes()), Some(m.clone())); + assert!( + m.clone().remembering(false).seen_evicting(), + "the memo never clears" + ); + + let fdb = WatchCursor::from_version(VersionToken::from_fdb_versionstamp(&[7; 10])) + .remembering(true); + assert_eq!(WatchCursor::from_bytes(&fdb.to_bytes()), Some(fdb)); + + let t = WatchCursor::from_version(VersionToken::from_u64(42).written_at(1_700_000_000_123)); + assert_eq!(t.to_bytes(), c.to_bytes(), "no memo: the time isn't kept"); + let t = t.remembering(true); + let back = WatchCursor::from_bytes(&t.to_bytes()).unwrap(); + assert_eq!(back.written_ms(), Some(1_700_000_000_123)); + assert!(back.seen_evicting() && back.as_u64() == Some(42)); + assert_eq!(back, t); + + let mut bad = m.to_bytes(); + bad[11] |= 0b100; + assert_eq!(WatchCursor::from_bytes(&bad), None, "unknown flag"); + let mut bad = m.to_bytes(); + bad[11] |= CURSOR_WRITTEN_AT; + assert_eq!(WatchCursor::from_bytes(&bad), None, "time flagged, missing"); + let mut bad = m.to_bytes(); + bad[10] = 11; + assert_eq!(WatchCursor::from_bytes(&bad), None, "version too long"); + assert_eq!(WatchCursor::from_bytes(&[0; 11]), None, "no such length"); + } + #[test] fn from_raw_roundtrips_within_capacity() { // The largest token any backend uses is a 10-byte FDB versionstamp. diff --git a/src/nats.rs b/src/nats.rs index a81348a..bad9f04 100644 --- a/src/nats.rs +++ b/src/nats.rs @@ -755,8 +755,17 @@ impl KvReader for NatsKvReader { async fn keys(&self, prefix: &str) -> Result, KvError> { debug!(prefix = %prefix, "listing keys with prefix"); - let mut keys = Vec::new(); + self.for_each_key(prefix, &mut |key| keys.push(key)).await?; + debug!(prefix = %prefix, keys = keys.len(), "keys listing complete"); + Ok(keys) + } + + async fn for_each_key( + &self, + prefix: &str, + f: &mut (dyn FnMut(String) + Send), + ) -> Result<(), KvError> { self.consume_last_per_subject(prefix, true, |msg, key| { // Skip both real KV deletes and CAS tombstones (empty-value Puts // written by delete_with_version). get()/scan() hide the latter, so @@ -764,13 +773,10 @@ impl KvReader for NatsKvReader { // With headers_only the payload is stripped, but NATS adds a // `Nats-Msg-Size` header we use to detect the empty value. if !is_kv_delete(&msg) && !is_empty_value(&msg) { - keys.push(key); + f(key); } }) - .await?; - - debug!(prefix = %prefix, keys = keys.len(), "keys listing complete"); - Ok(keys) + .await } async fn scan(&self, prefix: &str) -> Result, KvError> { @@ -815,11 +821,19 @@ impl KvReader for NatsKvReader { /// version on every scanned entry. We instead parse from the front, accounting /// for the optional `.` prefix that modern servers prepend. fn stream_sequence_from_ack(reply: &str) -> Option { - // The stream-seq field sits at index 5 (legacy) or 7 (modern), so we only - // ever read the first 8 tokens. Keep those in a stack array and count the + ack_fields(reply).map(|(seq, _)| seq) +} + +/// The stream sequence and the message's server timestamp (Unix nanoseconds) +/// from a JetStream ACK subject ([`stream_sequence_from_ack`] has the shapes; +/// the timestamp is two tokens after the sequence). +fn ack_fields(reply: &str) -> Option<(u64, Option)> { + // The stream-seq field sits at index 5 (legacy) or 7 (modern) and the + // timestamp two after it, so we only ever read the first 10 tokens. Keep + // those in a stack array and count the // remainder with the iterator — no heap `Vec`, which on a large `scan()` // would be one allocation per delivered message. - let mut head = [""; 8]; + let mut head = [""; 10]; let mut count = 0usize; for (i, token) in reply.split('.').enumerate() { if i < head.len() { @@ -833,7 +847,13 @@ fn stream_sequence_from_ack(reply: &str) -> Option { // Legacy form has exactly 9 tokens with no domain/account; anything longer // carries the two-token `.` prefix, shifting fields right. let stream_seq_idx = if count == 9 { 5 } else { 7 }; - head[stream_seq_idx].parse::().ok() + let seq = head[stream_seq_idx].parse::().ok()?; + Some((seq, head[stream_seq_idx + 2].parse::().ok())) +} + +/// Unix milliseconds of a server timestamp; 0 (unknown) before the epoch. +fn unix_ms(t: time::OffsetDateTime) -> u64 { + u64::try_from(t.unix_timestamp_nanos() / 1_000_000).unwrap_or(0) } /// Check if a NATS message represents a KV delete/purge operation. @@ -988,7 +1008,7 @@ impl NatsKvReader { /// instead of allocating a fresh copy per watch event. 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); + let version = VersionToken::from_u64(entry.revision).written_at(unix_ms(entry.created)); 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. @@ -1247,8 +1267,8 @@ fn kv_message_to_update(msg: &async_nats::Message, kv_prefix: &str) -> Option Result { + let Some(written_ms) = cursor.written_ms() else { + return Ok(revision); + }; + let subject = format!("STREAM.INFO.KV_{}", self.bucket); + let raw: serde_json::Value = timed(self.js.request(subject, &())) + .await? + .map_err(|e| KvError::OperationFailed(format!("stream info for resume: {e}")))?; + let info: async_nats::jetstream::stream::Info = match serde_json::from_value(raw.clone()) { + Ok(info) => info, + // An API error reply (or a shape we can't read): no shortcut; the + // window check reports what it finds. + Err(_) => return Ok(revision), + }; + let first = info.state.first_sequence; + if crate::protocol::resume_window_ok(revision, first) { + return Ok(revision); + } + let Some(now_ms) = raw + .get("ts") + .and_then(serde_json::Value::as_str) + .and_then(|ts| { + time::OffsetDateTime::parse(ts, &time::format_description::well_known::Rfc3339).ok() + }) + .map(unix_ms) + else { + return Ok(revision); + }; + let max_age_ms = u64::try_from(info.config.max_age.as_millis()).unwrap_or(u64::MAX); + let age_ms = now_ms.saturating_sub(written_ms); + if !crate::protocol::gap_holds_no_eviction( + evicts_only_by_age(&info.config), + max_age_ms, + age_ms, + ) { + return Ok(revision); + } + let resume = first.saturating_sub(1); + info!( + revision, + first_sequence = first, + cursor_age_ms = age_ms, + max_age_ms, + "every message between the cursor and the first retained one was superseded \ + (none is old enough to have aged out); resuming from the first retained" + ); + Ok(resume.max(revision)) + } + /// Create a resume consumer (or watch) for `revision`, the ONE way every /// `*_from` path does it: the window checked before creation /// ([`check_resume_window`](Self::check_resume_window); `create` gets the @@ -1372,9 +1454,18 @@ impl NatsKvWatcher { /// counts. When this returns false, a key missing from the bucket was deleted; /// when it returns true, it may simply have aged out. fn evicts_current_values(config: &async_nats::jetstream::stream::Config) -> bool { + !config.max_age.is_zero() || evicts_other_than_by_age(config) +} + +/// Current values go only by age: `max_age`, and neither per-message TTLs nor +/// a `discard: old` limit. +fn evicts_only_by_age(config: &async_nats::jetstream::stream::Config) -> bool { + !config.max_age.is_zero() && !evicts_other_than_by_age(config) +} + +fn evicts_other_than_by_age(config: &async_nats::jetstream::stream::Config) -> bool { use async_nats::jetstream::stream::DiscardPolicy as NatsDiscard; - !config.max_age.is_zero() - || config.allow_message_ttl + config.allow_message_ttl || (config.discard == NatsDiscard::Old && (config.max_bytes > 0 || config.max_messages > 0)) } @@ -1439,6 +1530,7 @@ impl KvWatcher for NatsKvWatcher { Some(rev) if rev > 0 => rev, _ => return self.watch_all(tx).await, }; + let revision = self.resume_point(cursor, revision).await?; let start = resume_start(revision)?; let watcher = self .create_resume(revision, &"all", |_| self.kv.watch_all_from_revision(start)) @@ -1463,6 +1555,7 @@ impl KvWatcher for NatsKvWatcher { Some(rev) if rev > 0 => rev, _ => return self.watch_prefix(prefix, tx).await, }; + let revision = self.resume_point(cursor, revision).await?; let nats_key = format!("{prefix}>"); let start = resume_start(revision)?; let watcher = self @@ -1492,6 +1585,7 @@ impl KvWatcher for NatsKvWatcher { Some(rev) if rev > 0 => rev, _ => return self.watch_prefixes(prefixes, tx).await, }; + let revision = self.resume_point(cursor, revision).await?; // async-nats has `watch_many` (multi-filter) and `watch_from_revision` // (seek) but no combination of the two, so build the multi-filter diff --git a/src/protocol.rs b/src/protocol.rs index 05465f5..b73ad4e 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -10,10 +10,11 @@ //! | [`resume_window_ok`] | `nats` resume paths (`check_resume_window`), the live floor guard | every model | //! | [`restore_allowed`] | `repair::check_restore` | `model.rs` `Act::RestoreRead`, live `GuardRepair`, repair-steps `WPlan`, fleet `Repair` | //! | [`restore_ahead`] | `repair::check_restore`, the main loop's re-check (`repair::fold_in`) | repair-steps restore reply | -//! | [`listing_is_truth`] | `repair` (plan, cursor-less start), `Snapshot::stale_keys` | every model's view of retention | +//! | [`listing_is_truth`] / [`listing_truth`] | `repair` (plan, cursor-less start), `Snapshot::stale_keys` | every model's view of retention; repair-steps `ConfigForgot` | //! | [`plan_repair`] | `repair::plan`, `Snapshot::stale_keys` | expiry dispatch in every model | //! | [`cursorless_start_needs_repair`] | `repair::run_watch` | repair-steps `WStart`, fleet `Start` | //! | [`restore_key`] | `repair::restore_diff` | repair-steps restore op | +//! | [`gap_holds_no_eviction`] | `nats` resume (`NatsKvWatcher::resume_point`) | — (unit-tested; `tests/eviction.rs` live) | //! //! Because the model transitions call these very functions, a change to any //! of them is re-verified against the full bounded state space on the next @@ -27,6 +28,8 @@ //! revision, with revisionless cursors ranked 0 by the callers) so they stay //! free of I/O types and usable from the checker's `u8`-bounded state space. +use crate::kv::Retention; + /// What the publisher observed at the pointer key before deciding. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum PointerState { @@ -151,10 +154,33 @@ pub fn restore_ahead(artifact_revision: u64, local_revision: u64) -> bool { /// Is the bucket's key listing the truth — does "not listed" mean /// "deleted"? Yes when retention never evicts current values, and also when -/// it can but has evicted nothing yet (first retained revision ≤ 1). +/// it can but has evicted nothing yet (first retained revision ≤ 1) — unless +/// the fold has ever seen this answer be no (`seen_evicting`). Retention can be edited: a key that aged out while +/// `max_age` or `discard: old` was in force stays missing from the listing +/// after the setting is removed (or the bucket recreated), so the current +/// config alone can't vouch for the listing. /// Erring toward `false` is safe: it only routes a repair to an artifact. -pub fn listing_is_truth(evicts_current_values: bool, first_revision: u64) -> bool { - !evicts_current_values || resume_window_ok(0, first_revision) +pub fn listing_is_truth( + evicts_current_values: bool, + first_revision: u64, + seen_evicting: bool, +) -> bool { + !seen_evicting && (!evicts_current_values || resume_window_ok(0, first_revision)) +} + +/// [`listing_is_truth`] for retention as a watcher reports it: `None` when +/// the backend can't say and the fold never saw it evict — then the caller's +/// chosen repair decides ([`plan_repair`]). +pub fn listing_truth(retention: Option, seen_evicting: bool) -> Option { + match retention { + Some(r) => Some(listing_is_truth( + r.evicts_current_values, + r.first_revision, + seen_evicting, + )), + None if seen_evicting => Some(false), + None => None, + } } /// The repair a watch armed for an expired cursor, without its payloads @@ -270,6 +296,20 @@ pub fn restore_key( } } +/// Can a resume gap that looks expired (the log's first retained revision is +/// past the cursor's successor) hold no evicted message? Only when age is the +/// log's sole way to evict a current value (`age_only`: `max_age`, no +/// per-message TTLs, no `discard: old` limit) and the cursor's own message +/// is at most half of `max_age` old: every later message is younger still, so +/// none has aged out, and each one missing was superseded by a later write +/// the resume delivers. Half rather than all of `max_age` is headroom for +/// clock steps and leader changes between the server that stamped the +/// message and the one reporting the time. `cursor_age_ms` is 0 when the +/// stamp is ahead of the clock. +pub fn gap_holds_no_eviction(age_only: bool, max_age_ms: u64, cursor_age_ms: u64) -> bool { + age_only && max_age_ms > 0 && cursor_age_ms.saturating_mul(2) <= max_age_ms +} + /// Sequence the watch should start at after applying `revision`. /// /// `None` at `u64::MAX`: wrapping to 0 would silently replay the bucket @@ -318,6 +358,19 @@ mod tests { assert_eq!(resume_start_sequence(u64::MAX), None); } + #[test] + fn gap_holds_no_eviction_boundaries() { + assert!(gap_holds_no_eviction(true, 60_000, 0)); + assert!(gap_holds_no_eviction(true, 60_000, 30_000)); + assert!(!gap_holds_no_eviction(true, 60_000, 30_001)); + assert!(!gap_holds_no_eviction(false, 60_000, 0), "other eviction"); + assert!(!gap_holds_no_eviction(true, 0, 0), "no max_age"); + assert!( + !gap_holds_no_eviction(true, 60_000, u64::MAX), + "no overflow" + ); + } + #[test] fn repair_planner_is_total() { use RepairMode::*; @@ -339,12 +392,30 @@ mod tests { RepairPlan::Restore, "unknown: restore" ); - assert!(listing_is_truth(false, 99)); + assert!(listing_is_truth(false, 99, false)); assert!( - listing_is_truth(true, 1), + listing_is_truth(true, 1, false), "evicting, but nothing evicted yet" ); - assert!(!listing_is_truth(true, 2)); + assert!(!listing_is_truth(true, 2, false)); + assert!( + !listing_is_truth(false, 99, true), + "eviction removed from the config, but the fold saw it in force" + ); + let r = |evicts| { + Some(Retention { + evicts_current_values: evicts, + first_revision: 9, + }) + }; + assert_eq!(listing_truth(r(false), false), Some(true)); + assert_eq!(listing_truth(r(false), true), Some(false)); + assert_eq!(listing_truth(Option::None, false), Option::None); + assert_eq!( + listing_truth(Option::None, true), + Some(false), + "the memo speaks even when the backend can't" + ); assert!( cursorless_start_needs_repair(Relist, true, Some(true)), "data, no cursor" diff --git a/src/repair.rs b/src/repair.rs index 5d87c08..2c3acb2 100644 --- a/src/repair.rs +++ b/src/repair.rs @@ -26,7 +26,11 @@ use std::any::Any; use std::collections::HashSet; +use std::fs::File; +use std::io::{BufReader, BufWriter, Read, Seek, Write}; +use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::Duration; use async_trait::async_trait; use tokio::sync::{mpsc, oneshot}; @@ -34,9 +38,9 @@ use tracing::{error, info, warn}; use crate::applied::{Fold, WatchScope}; use crate::artifact::ExportManifest; -use crate::kv::{KvEntry, KvError, KvReader, KvUpdate, KvWatcher, VersionToken, WatchCursor}; +use crate::kv::{KvError, KvReader, KvUpdate, KvWatcher, Retention, VersionToken, WatchCursor}; use crate::protocol::{ - KeyRestore, KeyState, RepairMode, RepairPlan, cursorless_start_needs_repair, listing_is_truth, + KeyRestore, KeyState, RepairMode, RepairPlan, cursorless_start_needs_repair, listing_truth, plan_repair, restore_ahead, restore_allowed, restore_key, resume_window_ok, }; use crate::snapshot::{SnapshotError, SnapshotStore}; @@ -86,7 +90,9 @@ pub enum ExpiryRepair { }, /// Decide by the bucket's live retention, read at the moment of expiry: /// [`Relist`](Self::Relist) when it never evicts current values, - /// [`Restore`](Self::Restore) when it does or can't say. + /// [`Restore`](Self::Restore) when it does or can't say — and always + /// Restore once the fold has seen it evict, even after eviction is + /// turned off (a key that aged out stays missing from the listing). Auto { /// Lists live keys (for either repair). reader: Arc, @@ -138,6 +144,7 @@ pub trait RestoreSource: Send + Sync { /// first, then the guard. pub struct RestoredFold { manifest: ExportManifest, + scratch: Option, // Declaration order is drop order, and it is load-bearing: the fold (an // open LSM, possibly) must close before its guard deletes its directory. fold: S, @@ -149,11 +156,21 @@ impl RestoredFold { pub fn new(manifest: ExportManifest, fold: S) -> Self { Self { manifest, + scratch: None, fold, guard: None, } } + /// Where the restore may spill its working set (the keys it will + /// rewrite) instead of holding it in memory. Put it on the fold's + /// filesystem, not a RAM-backed temp dir; without it the restore spills + /// to the system temp dir. + pub fn scratch_in(mut self, dir: impl Into) -> Self { + self.scratch = Some(dir.into()); + self + } + /// Keep `guard` (e.g. the `TempDir` holding the fold's files) alive until /// after the fold is dropped. pub fn holding(mut self, guard: impl Any + Send) -> Self { @@ -231,39 +248,42 @@ pub(crate) fn check_restore( Ok(()) } -/// One key of a restore: take the artifact's entry, or delete the key. -pub(crate) enum RestoreOp { - Put(String), - Delete(String), -} - -impl RestoreOp { - fn key(&self) -> &str { - match self { - RestoreOp::Put(k) | RestoreOp::Delete(k) => k, +/// `prefixes` with every prefix that another one covers dropped, so no key +/// matches two of them: a scan per prefix then visits each key once. +pub(crate) fn disjoint(prefixes: &[String]) -> Vec { + let mut sorted: Vec<&String> = prefixes.iter().collect(); + sorted.sort_by_key(|p| p.len()); + let mut kept: Vec = Vec::new(); + for p in sorted { + if !kept.iter().any(|k| p.starts_with(k.as_str())) { + kept.push(p.clone()); } } + kept } -/// The in-scope ops that bring the `local` fold to the `artifact` fold's -/// state at its cursor `C`, without ever moving a key backward or deleting a -/// live one. Keys only, in key order — values are read back in bounded -/// chunks ([`materialize`]), so a restore never holds the whole changed set's -/// values at once (the same discipline as the key-listing diff). +/// One pass over a restore's in-scope keys, in key order, deciding each with +/// [`restore_key`] (which the repair model, `tests/model_repair.rs`, runs too). +/// Keys the artifact's entry should replace go to `take`, one at a time, so the +/// caller can spill them rather than hold them. Local keys the artifact lacks +/// are returned: the restore deletes them unless the bucket lists them live, +/// and only these need checking against the listing. Out-of-scope keys are +/// untouched on both sides. /// -/// Each in-scope key is decided by [`restore_key`], which the repair model -/// (`tests/model_repair.rs`) runs too. Out-of-scope keys are untouched on -/// both sides. +/// Memory is the returned set, which holds the keys deleted between the local +/// fold's cursor and the artifact's: never the fold, the artifact, or the +/// bucket's listing. pub(crate) fn restore_diff( local: &S, artifact: &S, prefixes: &[String], - live: &HashSet, -) -> Result, SnapshotError> { - let mut ops = Vec::new(); - for prefix in prefixes { - // Every key the artifact holds. - artifact.for_each_in_range(prefix, |entry| { + mut take: impl FnMut(String) -> Result<(), SnapshotError>, +) -> Result, SnapshotError> { + let mut candidates = HashSet::new(); + for prefix in disjoint(prefixes) { + // Every key the artifact holds. Whether the bucket lists it doesn't + // matter when the artifact has it. + artifact.for_each_in_range(&prefix, |entry| { let at = KeyState::At(entry.version.as_u64()); let (state, identical) = match local.get(&entry.key)? { Some(l) => ( @@ -272,56 +292,125 @@ pub(crate) fn restore_diff( ), None => (KeyState::Absent, false), }; - let listed = live.contains(&entry.key); - if restore_key(state, at, identical, listed) == KeyRestore::TakeArtifact { - ops.push(RestoreOp::Put(entry.key)); + if restore_key(state, at, identical, false) == KeyRestore::TakeArtifact { + take(entry.key)?; } Ok(()) })?; // Local keys the artifact lacks. The kernel is asked first, as if the - // artifact lacked the key, so the artifact is read only for keys that - // would be deleted. - local.for_each_in_range(prefix, |entry| { + // artifact lacked the key and the bucket didn't list it, so the + // artifact is read only for keys that could be deleted. + local.for_each_in_range(&prefix, |entry| { let state = KeyState::At(entry.version.as_u64()); - let listed = live.contains(&entry.key); - if restore_key(state, KeyState::Absent, false, listed) == KeyRestore::Delete + if restore_key(state, KeyState::Absent, false, false) == KeyRestore::Delete && artifact.get(&entry.key)?.is_none() { - ops.push(RestoreOp::Delete(entry.key)); + candidates.insert(entry.key); } Ok(()) })?; } - // Overlapping prefixes visit a key more than once; a key is a Put or a - // Delete, never both (Put iff the artifact has it). - ops.sort_unstable_by(|a, b| a.key().cmp(b.key())); - ops.dedup_by(|a, b| a.key() == b.key()); - Ok(ops) + Ok(candidates) +} + +/// Keys written to a scratch file and read back in chunks: a restore's +/// rewrite set can be the whole fold (a fresh node seeding from an artifact), +/// so it never sits in memory. Each key is `len: u32 LE ++ bytes`. +pub(crate) struct KeySpill { + w: BufWriter, + len: u64, +} + +impl KeySpill { + /// A spill in `dir`, or the system temp dir. The file is unlinked from the + /// start, so nothing is left behind. + pub(crate) fn new(dir: Option<&Path>) -> std::io::Result { + let file = match dir { + Some(dir) => tempfile::tempfile_in(dir)?, + None => tempfile::tempfile()?, + }; + Ok(Self { + w: BufWriter::new(file), + len: 0, + }) + } + + pub(crate) fn push(&mut self, key: &str) -> std::io::Result<()> { + let n = u32::try_from(key.len()).map_err(|_| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "key longer than 4 GiB") + })?; + self.w.write_all(&n.to_le_bytes())?; + self.w.write_all(key.as_bytes())?; + self.len += 1; + Ok(()) + } + + pub(crate) fn len(&self) -> u64 { + self.len + } + + /// Rewind for reading. + pub(crate) fn into_reader(self) -> std::io::Result { + let mut file = self.w.into_inner().map_err(|e| e.into_error())?; + file.rewind()?; + Ok(SpillReader { + r: BufReader::new(file), + left: self.len, + }) + } +} + +/// The read side of a [`KeySpill`]. +pub(crate) struct SpillReader { + r: BufReader, + left: u64, } -/// Read one chunk of a restore's values back from the artifact fold. Deletes -/// carry the unknown version: like the key-listing diff's synthetic deletes, -/// they are a state correction, not a log entry, and never move a cursor. +impl SpillReader { + /// Up to `max` keys, in the order they were pushed; empty when done. + pub(crate) fn take(&mut self, max: usize) -> std::io::Result> { + let mut keys = Vec::new(); + while self.left > 0 && keys.len() < max { + let mut n = [0u8; 4]; + self.r.read_exact(&mut n)?; + let mut key = vec![0u8; u32::from_le_bytes(n) as usize]; + self.r.read_exact(&mut key)?; + keys.push( + String::from_utf8(key) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?, + ); + self.left -= 1; + } + Ok(keys) + } +} + +/// Read one chunk of a restore's rewritten entries back from the artifact +/// fold. pub(crate) fn materialize( artifact: &S, - ops: Vec, + keys: Vec, ) -> Result, SnapshotError> { - ops.into_iter() - .map(|op| match op { - RestoreOp::Put(key) => match artifact.get(&key)? { - Some(entry @ KvEntry { .. }) => Ok(KvUpdate::Put(entry)), - None => Err(SnapshotError::Backend(format!( - "restored artifact fold lost key {key:?} mid-restore" - ))), - }, - RestoreOp::Delete(key) => Ok(KvUpdate::Delete { - key, - version: VersionToken::unknown(), - }), + keys.into_iter() + .map(|key| match artifact.get(&key)? { + Some(entry) => Ok(KvUpdate::Put(entry)), + None => Err(SnapshotError::Backend(format!( + "restored artifact fold lost key {key:?} mid-restore" + ))), }) .collect() } +/// A restore's delete: like the key-listing diff's synthetic deletes, a state +/// correction rather than a log entry, so it carries the unknown version and +/// never moves a cursor. +fn restore_delete(key: String) -> KvUpdate { + KvUpdate::Delete { + key, + version: VersionToken::unknown(), + } +} + // --------------------------------------------------------------------------- // The repair flow. Two tasks share it: the watch task, which owns the watch // and decides on a repair when the cursor expires, and `watch_applied`'s main @@ -333,6 +422,9 @@ pub(crate) fn materialize( /// A cursor-expired repair handoff from the watch task to the main loop. pub(crate) enum RepairRequest { + /// The bucket's retention was seen evicting current values: the fold + /// commits the memo on its cursor. No reply; the watch keeps running. + Evicting, /// The key-listing diff: the bucket's live keys for the watch scope. The /// main loop applies synthetic deletes for in-scope keys missing from it, /// then acks so the watch task can start the fallback re-list. @@ -340,13 +432,15 @@ pub(crate) enum RepairRequest { live_keys: Vec, ack: oneshot::Sender<()>, }, - /// The artifact restore: a verified artifact fold, and the bucket's live - /// keys for the scope (listed after the fetch). The main loop folds the - /// in-scope difference, advances to the artifact's cursor, and replies - /// with it so the watch task resumes from there. + /// The artifact restore: a verified artifact fold, the reader that lists + /// the bucket's live keys, and the watcher whose retention vouches for + /// that listing. The main loop folds the in-scope difference, advances to + /// the artifact's cursor, and replies with it so the watch task resumes + /// from there. Restore { restored: RestoredFold, - live: HashSet, + reader: Arc, + watcher: Arc, reply: oneshot::Sender>, }, } @@ -408,13 +502,97 @@ pub(crate) fn arm( } } -/// [`listing_is_truth`] for the watcher's live retention; `None` when the -/// backend can't say. -async fn listing_truth(watcher: &dyn KvWatcher) -> Result, KvError> { - Ok(watcher - .retention() - .await? - .map(|r| listing_is_truth(r.evicts_current_values, r.first_revision))) +/// How often a live watch re-reads the bucket's retention, until it first +/// sees current-value eviction (only while a key-listing repair could still +/// be planned: `Relist` or `Auto`). +const RETENTION_POLL: Duration = Duration::from_secs(60); + +/// The watch task's copy of the eviction memo ([`WatchCursor`]'s): has this +/// fold ever seen the bucket's retention evict current values? The fold +/// keeps it durably; this copy decides the task's repairs. Retention can be +/// edited, and a key that aged out while eviction was in force stays missing +/// from the bucket's listing after it is turned off, so the listing is +/// trusted only by a fold that never saw eviction ([`listing_truth`]). +struct EvictionMemo<'a, S> { + seen: bool, + repair: Option<&'a RepairHandle>, +} + +impl EvictionMemo<'_, S> { + /// Read the bucket's retention. The first time it shows current values + /// may have been evicted (eviction on, and the log already past its first + /// revision), remember it here and have the fold commit it. Eviction that + /// is configured but hasn't dropped anything yet isn't remembered: the + /// listing is still complete, and a fresh node on a fresh bucket must not + /// be sent to an artifact that doesn't exist yet. + async fn observe(&mut self, watcher: &dyn KvWatcher) -> Result, KvError> { + let retention = watcher.retention().await?; + if !self.seen && listing_truth(retention, false) == Some(false) { + self.seen = true; + if let Some(h) = self.repair { + info!( + "the bucket's retention has evicted current values; this fold will not \ + trust the bucket's key listing again, even if eviction is turned off" + ); + h.tx.send(RepairRequest::Evicting).await.map_err(|_| { + KvError::WatchError("watch loop ended while recording eviction".into()) + })?; + } + } + Ok(retention) + } + + /// [`listing_truth`] for the retention read now and the memo. + async fn listing_truth(&mut self, watcher: &dyn KvWatcher) -> Result, KvError> { + let retention = self.observe(watcher).await?; + Ok(listing_truth(retention, self.seen)) + } + + /// Could a key-listing repair still be planned, so retention is worth + /// watching? + fn polls(&self) -> bool { + !self.seen + && matches!( + self.repair.map(|h| h.mode.mode()), + Some(RepairMode::Relist | RepairMode::Auto) + ) + } + + /// Read retention without failing the watch on an error: the read only + /// feeds the memo, and the next one retries. + async fn poll(&mut self, watcher: &dyn KvWatcher) { + if let Err(e) = self.observe(watcher).await { + warn!(error = %e, "reading the bucket's retention failed; retrying later"); + } + } + + /// Run a live watch, re-reading retention every [`RETENTION_POLL`] + /// while [`polls`](Self::polls): eviction turned on and off again while + /// the watch runs must still be seen. + async fn watching( + &mut self, + watcher: &dyn KvWatcher, + watch: impl std::future::Future>, + ) -> Result<(), KvError> { + if !self.polls() { + return watch.await; + } + tokio::pin!(watch); + let mut tick = tokio::time::interval(RETENTION_POLL); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + tick.tick().await; + loop { + tokio::select! { + res = &mut watch => return res, + _ = tick.tick() => { + self.poll(watcher).await; + if !self.polls() { + return watch.await; + } + } + } + } + } } /// The watch task: run the underlying watch for `scope`, resuming from @@ -437,7 +615,12 @@ pub(crate) async fn run_watch( repair: Option>, tx: mpsc::Sender, ) -> Result<(), KvError> { + let shared = Arc::clone(&watcher); let (watcher, scope) = (watcher.as_ref(), &scope); + let mut memo = EvictionMemo { + seen: resume.as_ref().is_some_and(WatchCursor::seen_evicting), + repair: repair.as_ref(), + }; // Resume only when the cursor carries a real position; an absent or `none()` // cursor falls through to a full watch. Binding `cursor` here makes "we have a // resume position" structural — there is no separate bool whose truth a later @@ -453,7 +636,7 @@ pub(crate) async fn run_watch( let truth = match mode { RepairMode::Relist => None, RepairMode::Restore | RepairMode::Auto if unanchored => None, - _ => listing_truth(watcher).await?, + _ => memo.listing_truth(watcher).await?, }; repair_first = cursorless_start_needs_repair(mode, unanchored, truth); if repair_first && unanchored { @@ -477,38 +660,49 @@ pub(crate) async fn run_watch( } } + if memo.polls() { + memo.poll(watcher).await; + } + loop { let cursor = if std::mem::take(&mut repair_first) { WatchCursor::none() } else { let Some(cursor) = resume.take() else { - return watch_scope(watcher, scope, tx).await; + return memo + .watching(watcher, watch_scope(watcher, scope, tx)) + .await; }; - match watch_scope_from(watcher, scope, &cursor, tx.clone()).await { + let watch = watch_scope_from(watcher, scope, &cursor, tx.clone()); + match memo.watching(watcher, watch).await { Err(KvError::CursorExpired) => cursor, other => return other, } }; - match plan(watcher, repair.as_ref()).await? { + match plan(watcher, repair.as_ref(), &mut memo).await? { Plan::ReListOnly => { warn!( "watch cursor expired with no repair armed (needs a store and a reader or \ restore source); falling back to the re-list alone — keys deleted during \ the gap may persist in the fold" ); - return watch_scope(watcher, scope, tx).await; + return memo + .watching(watcher, watch_scope(watcher, scope, tx)) + .await; } Plan::Relist(reader, repairs) => { warn!( "watch cursor expired; resyncing stale keys, then falling back to the full re-list" ); resync_stale_keys(scope, reader, repairs).await?; - return watch_scope(watcher, scope, tx).await; + return memo + .watching(watcher, watch_scope(watcher, scope, tx)) + .await; } Plan::Restore(reader, source, repairs) => { warn!("watch cursor expired; restoring the fold from the latest artifact"); let restored = - restore_from_artifact(watcher, scope, reader, source, repairs, &cursor) + restore_from_artifact(&shared, scope, reader, source, repairs, &cursor) .await .map_err(|e| match e { KvError::WatchError(msg) if cursor.is_none() => { @@ -522,6 +716,9 @@ pub(crate) async fn run_watch( } other => other, })?; + // An artifact from a fold that saw eviction carries the memo + // (the fold took it with the artifact's cursor). + memo.seen |= restored.seen_evicting(); resume = Some(restored); } } @@ -566,22 +763,24 @@ async fn watch_scope_from( async fn plan<'a, S>( watcher: &dyn KvWatcher, repair: Option<&'a RepairHandle>, + memo: &mut EvictionMemo<'_, S>, ) -> Result, KvError> { let Some(h) = repair else { return Ok(Plan::ReListOnly); }; let mode = h.mode.mode(); let truth = match mode { - RepairMode::Relist | RepairMode::Auto => listing_truth(watcher).await?, + RepairMode::Relist | RepairMode::Auto => memo.listing_truth(watcher).await?, RepairMode::None | RepairMode::Restore => None, }; Ok(match (plan_repair(mode, truth), &h.mode) { (RepairPlan::ReListOnly, _) => Plan::ReListOnly, (RepairPlan::RefuseRelist, _) => { let msg = "watch cursor expired on a bucket whose retention evicts current \ - values (max_age, per-message TTL, or discard:old): its key listing \ - can't tell a deleted key from an aged-out one, so the key-listing \ - resync would delete valid keys. Refusing. Repair from artifacts with \ + values (max_age, per-message TTL, or discard:old), or did while this \ + fold tracked it: its key listing can't tell a deleted key from an \ + aged-out one, so the key-listing resync would delete valid keys. \ + Refusing. Repair from artifacts with \ ExpiryRepair::Restore or ExpiryRepair::Auto (or accept a re-list-only \ fallback with ExpiryRepair::None)"; error!(msg); @@ -603,7 +802,7 @@ async fn plan<'a, S>( /// retention), download it, check what actually arrived, then hand it to the /// main loop and wait for the cursor it restored to. async fn restore_from_artifact( - watcher: &dyn KvWatcher, + watcher: &Arc, scope: &WatchScope, reader: &Arc, source: &Arc>, @@ -645,23 +844,12 @@ async fn restore_from_artifact( "restoring the fold from the latest artifact" ); - // The bucket's live keys, listed after the artifact's cursor is fixed: - // the restore never deletes one of these (see `restore_diff`). A failed - // listing fails the watch, like the key-listing repair's. - let mut live = HashSet::new(); - for prefix in &prefixes { - let keys = reader - .keys(prefix) - .await - .map_err(|e| fail(format!("listing live keys under {prefix:?} failed: {e}")))?; - live.extend(keys); - } - let (reply_tx, reply_rx) = oneshot::channel(); repairs .send(RepairRequest::Restore { restored, - live, + reader: Arc::clone(reader), + watcher: Arc::clone(watcher), reply: reply_tx, }) .await @@ -746,6 +934,10 @@ where A: FnMut(Vec) + Send, O: FnMut(WatchCursor) + Send, { + let req = match req { + RepairRequest::Evicting => return fold.remember_evicting().await, + req => req, + }; // Fold everything the watch task delivered before it asked for this // repair. A floor-guard trip ends a live watch mid-stream with updates // still buffered in the channel: the repair must see them (it diffs the @@ -763,6 +955,7 @@ where // removed, and the put would then commit and resurrect the key. fold.settle().await?; match req { + RepairRequest::Evicting => unreachable!("handled above"), RepairRequest::Relist { live_keys, ack } => { relist(fold, &live_keys, prefixes).await?; // Ack AFTER the deletes are applied: the watch task is holding @@ -778,7 +971,8 @@ where } RepairRequest::Restore { restored, - live, + reader, + watcher, reply, } => { let target = restored.manifest().cursor.clone(); @@ -797,8 +991,15 @@ where let _ = reply.send(Err(KvError::WatchError(msg))); return Ok(()); } - restore(fold, restored, live, prefixes).await?; - let _ = reply.send(Ok(target)); + match restore(fold, restored, &*reader, &*watcher, prefixes).await? { + Ok(()) => { + let _ = reply.send(Ok(target)); + } + Err(msg) => { + error!(%msg, "cursor-expiry repair refused"); + let _ = reply.send(Err(KvError::WatchError(msg))); + } + } Ok(()) } } @@ -870,12 +1071,18 @@ where /// The artifact restore, main-loop half: replace the in-scope fold with the /// artifact's ([`restore_diff`]), then take its cursor. +/// +/// Memory stays bounded at any fold size. The keys to rewrite are spilled to +/// scratch and read back a chunk at a time; only the keys the restore would +/// delete are held, and the bucket's listing is streamed past them rather +/// than collected. With no delete candidates the listing is skipped. async fn restore( fold: &mut Fold, mut restored: RestoredFold, - live: HashSet, + reader: &dyn KvReader, + watcher: &dyn KvWatcher, prefixes: &[String], -) -> Result<(), KvError> +) -> Result, KvError> where U: Send, S: SnapshotStore + Send + 'static, @@ -887,53 +1094,115 @@ where let Some(st) = fold.take_store() else { unreachable!("repairs are armed only with a store") }; - let (prefixes, max) = (prefixes.to_vec(), fold.batch_cap()); + let prefixes = disjoint(prefixes); + let max = fold.batch_cap(); + let scope = prefixes.clone(); let (st, r, diff) = tokio::task::spawn_blocking(move || { - let diff = restore_diff(&st, restored.fold(), &prefixes, &live); + let diff = (|| { + let mut spill = KeySpill::new(restored.scratch.as_deref())?; + let candidates = restore_diff(&st, restored.fold(), &scope, |key| { + spill.push(&key).map_err(SnapshotError::from) + })?; + Ok::<_, SnapshotError>((spill, candidates)) + })(); (st, restored, diff) }) .await .map_err(|e| repair_fatal(format!("cursor-expired restore: diff task panicked: {e}")))?; fold.put_store(st); restored = r; - let diff = diff.map_err(|e| { + let (spill, mut deletes) = diff.map_err(|e| { repair_fatal(format!( "cursor-expired restore: diffing the fold against the artifact failed: {e}" )) })?; + + // A delete candidate the bucket lists live is kept: its current value is a + // write after the artifact's cursor, which the resume delivers + // (`restore_key`'s `listed_live`). A failed listing fails the watch, like + // the key-listing repair's. + if !deletes.is_empty() { + for prefix in &prefixes { + reader + .for_each_key(prefix, &mut |key| { + deletes.remove(&key); + }) + .await + .map_err(|e| { + repair_fatal(format!( + "cursor-expired restore: listing live keys under {prefix:?} failed: {e}" + )) + })?; + } + // The listing vouches for every write after the artifact's cursor only + // while the log still holds all of them: one that aged out before the + // listing is unlisted, and would be deleted. Retention only moves + // forward, so a check after the listing covers it + // (`tests/model_repair.rs`, `NoListingRecheck`). Nothing is folded yet. + if let Some(first) = watcher.retention().await?.map(|r| r.first_revision) + && !resume_window_ok(target.rank(), first) + { + let _ = tokio::task::spawn_blocking(move || drop(restored)).await; + return Ok(Err(format!( + "cursor-expired restore: the log evicted writes after the artifact's cursor \ + ({:?}, first retained revision {first}) while the restore was listing live \ + keys, so the listing can't vouch for them; nothing was changed. Exports must \ + run well inside the bucket's max_age / discard turnover", + target + ))); + } + } info!( from = ?fold.applied(), to = ?target, - changed = diff.len(), + rewritten = spill.len(), + deleted = deletes.len(), "cursor-expired restore: replacing the in-scope fold with the artifact's" ); + // Fold the difference through parse/apply/store in `max`-sized chunks, // values read back per chunk. These carry no stream position: every chunk // commits under the old cursor, so a crash part-way re-runs the // (idempotent) restore on restart. - let mut ops = diff.into_iter(); + let mut spill = spill + .into_reader() + .map_err(|e| repair_fatal(format!("cursor-expired restore: reading scratch: {e}")))?; loop { - let chunk: Vec<_> = ops.by_ref().take(max).collect(); - if chunk.is_empty() { - break; - } - let (r, updates) = tokio::task::spawn_blocking(move || { - let updates = materialize(restored.fold(), chunk); - (restored, updates) + let (r, sp, updates) = tokio::task::spawn_blocking(move || { + let updates = spill + .take(max) + .map_err(SnapshotError::from) + .and_then(|keys| materialize(restored.fold(), keys)); + (restored, spill, updates) }) .await .map_err(|e| repair_fatal(format!("cursor-expired restore: read task panicked: {e}")))?; restored = r; + spill = sp; let updates = updates.map_err(|e| { repair_fatal(format!( "cursor-expired restore: reading the artifact failed: {e}" )) })?; + if updates.is_empty() { + break; + } for u in updates { fold.correct(u); } fold.flush().await?; } + let mut deletes = deletes.into_iter(); + loop { + let chunk: Vec<_> = deletes.by_ref().take(max).collect(); + if chunk.is_empty() { + break; + } + for key in chunk { + fold.correct(restore_delete(key)); + } + fold.flush().await?; + } // Now the fold IS the artifact's (in scope): take its cursor, committed to // the store with the final chunk's re-queued remainder, if any. fold.advance_to(target); @@ -941,12 +1210,13 @@ where // Closing an LSM can block; drop the temporary fold and its scratch dir // off the async thread. let _ = tokio::task::spawn_blocking(move || drop(restored)).await; - Ok(()) + Ok(Ok(())) } #[cfg(test)] mod tests { use super::*; + use crate::kv::KvEntry; use crate::snapshot::AppendLogSnapshot; fn put(key: &str, value: &[u8], rev: u64) -> KvUpdate { @@ -1028,9 +1298,41 @@ mod tests { assert!(err.contains("outside the log's retention window"), "{err}"); } + /// The restore's updates in production order — rewritten entries from + /// the spill, then deletes for candidates the listing lacks — through the + /// real `restore_diff`, `KeySpill`, and `materialize`. + fn restore_ops( + local: &AppendLogSnapshot, + artifact: &AppendLogSnapshot, + prefixes: &[String], + live: &HashSet, + ) -> Vec { + let mut spill = KeySpill::new(None).unwrap(); + let candidates = restore_diff(local, artifact, prefixes, |k| { + spill.push(&k).map_err(SnapshotError::from) + }) + .unwrap(); + let mut reader = spill.into_reader().unwrap(); + let mut updates = Vec::new(); + loop { + let chunk = reader.take(2).unwrap(); + if chunk.is_empty() { + break; + } + updates.extend(materialize(artifact, chunk).unwrap()); + } + let mut deletes: Vec = candidates + .into_iter() + .filter(|k| !live.contains(k)) + .collect(); + deletes.sort(); + updates.extend(deletes.into_iter().map(restore_delete)); + updates + } + /// The diff turns the local in-scope fold into the artifact's: changed and - /// new keys are Puts, keys the artifact lacks are Deletes, identical - /// entries and out-of-scope keys are left alone. + /// new keys are rewritten, keys the artifact lacks are delete candidates, + /// identical entries and out-of-scope keys are left alone. #[test] fn diff_is_wholesale_replacement_within_scope() { let dir = tempfile::TempDir::new().unwrap(); @@ -1056,27 +1358,16 @@ mod tests { ], 9, ); - let ops = restore_diff( - &local, - &artifact, - &["node.".to_string()], - &Default::default(), - ) + let mut rewritten = Vec::new(); + let candidates = restore_diff(&local, &artifact, &["node.".to_string()], |k| { + rewritten.push(k); + Ok(()) + }) .unwrap(); - let got: Vec<(String, bool)> = ops - .iter() - .map(|o| (o.key().to_string(), matches!(o, RestoreOp::Put(_)))) - .collect(); - assert_eq!( - got, - vec![ - ("node.changed".into(), true), - ("node.gone".into(), false), - ("node.new".into(), true), - ] - ); + assert_eq!(rewritten, vec!["node.changed", "node.new"]); + assert_eq!(candidates, HashSet::from(["node.gone".to_string()])); - let updates = materialize(&artifact, ops).unwrap(); + let updates = restore_ops(&local, &artifact, &["node.".to_string()], &HashSet::new()); match &updates[0] { KvUpdate::Put(e) => { assert_eq!(e.value, b"new"); @@ -1089,24 +1380,62 @@ mod tests { other => panic!("expected put, got {other:?}"), } assert!( - matches!(&updates[1], KvUpdate::Delete { version, .. } if version.is_unknown()), + matches!(&updates[2], KvUpdate::Delete { version, .. } if version.is_unknown()), "restore deletes never carry a revision" ); } #[test] - fn diff_dedups_overlapping_prefixes() { + fn diff_visits_each_key_once_under_overlapping_prefixes() { let dir = tempfile::TempDir::new().unwrap(); let local = fold(&dir, "local", &[put("node.a", b"1", 1)], 1); let artifact = fold(&dir, "artifact", &[put("node.b", b"2", 2)], 2); - let ops = restore_diff( + let updates = restore_ops( &local, &artifact, &["node.".into(), "node".into()], - &Default::default(), - ) - .unwrap(); - assert_eq!(ops.len(), 2, "each key once despite two covering prefixes"); + &HashSet::new(), + ); + assert_eq!( + updates.len(), + 2, + "each key once despite two covering prefixes" + ); + } + + #[test] + fn disjoint_drops_covered_prefixes() { + let s = |v: &[&str]| v.iter().map(|p| p.to_string()).collect::>(); + assert_eq!( + disjoint(&s(&["node.", "node", "edge."])), + s(&["node", "edge."]) + ); + assert_eq!(disjoint(&s(&["a.", ""])), s(&[""])); + assert_eq!(disjoint(&s(&["a.", "a."])), s(&["a."])); + } + + #[test] + fn spill_round_trips_in_order_and_chunks() { + let mut spill = KeySpill::new(None).unwrap(); + let keys: Vec = (0..10).map(|i| format!("k.{i}")).collect(); + for k in &keys { + spill.push(k).unwrap(); + } + spill.push("").unwrap(); + assert_eq!(spill.len(), 11); + let mut r = spill.into_reader().unwrap(); + let mut got = Vec::new(); + loop { + let chunk = r.take(3).unwrap(); + if chunk.is_empty() { + break; + } + assert!(chunk.len() <= 3); + got.extend(chunk); + } + let mut want = keys; + want.push(String::new()); + assert_eq!(got, want); } // --- Exhaustive small-domain checks of the real functions --------------- @@ -1244,7 +1573,7 @@ mod tests { /// THE RESTORE, exhaustively over every pair of 3-key folds (each key /// absent or at one of two revisions: 729 pairs), four scopes including /// overlapping prefixes, and every live-key listing (8), using the real - /// `restore_diff` and `materialize` and a real store. Per in-scope key, + /// `restore_diff`, `KeySpill` and `materialize` and a real store. Per in-scope key, /// the restored fold holds: /// /// - the artifact's entry, unless the local one is NEWER (or the same) — @@ -1278,7 +1607,7 @@ mod tests { for scope in scopes { let prefixes: Vec = scope.iter().map(|p| p.to_string()).collect(); for (li, live) in lives.iter().enumerate() { - let full = restore_diff( + let full = restore_ops( &fold_of( &dir, &format!("probe-{li}-{}", scope.join("|")), @@ -1288,19 +1617,16 @@ mod tests { &artifact, &prefixes, live, - ) - .unwrap(); + ); // Crash after `cut` ops: apply them, then re-diff. for cut in 0..=full.len() { let name = format!("local-{li}-{cut}-{}", scope.join("|")); let mut local = fold_of(&dir, &name, &local_states, 3); - let first = restore_diff(&local, &artifact, &prefixes, live).unwrap(); - let partial: Vec = first.into_iter().take(cut).collect(); - let updates = materialize(&artifact, partial).unwrap(); - local.apply(&updates, &WatchCursor::from_u64(3)).unwrap(); - let rest = restore_diff(&local, &artifact, &prefixes, live).unwrap(); - let updates = materialize(&artifact, rest).unwrap(); - local.apply(&updates, &WatchCursor::from_u64(9)).unwrap(); + let first = restore_ops(&local, &artifact, &prefixes, live); + let partial: Vec = first.into_iter().take(cut).collect(); + local.apply(&partial, &WatchCursor::from_u64(3)).unwrap(); + let rest = restore_ops(&local, &artifact, &prefixes, live); + local.apply(&rest, &WatchCursor::from_u64(9)).unwrap(); for (i, key) in KEYS.iter().enumerate() { let (l, a) = (local_states[i], artifact_states[i]); @@ -1326,9 +1652,7 @@ mod tests { ); } assert!( - restore_diff(&local, &artifact, &prefixes, live) - .unwrap() - .is_empty(), + restore_ops(&local, &artifact, &prefixes, live).is_empty(), "a converged fold must diff empty" ); cases += 1; diff --git a/src/snapshot.rs b/src/snapshot.rs index c4fb9cd..212b31c 100644 --- a/src/snapshot.rs +++ b/src/snapshot.rs @@ -45,8 +45,8 @@ use std::io::{self, Write}; use std::path::{Path, PathBuf}; use crate::artifact::{ExportManifest, ExportStage, verify_and_stage_import}; -use crate::kv::{KvEntry, KvUpdate, Retention, VersionToken, WatchCursor}; -use crate::protocol::{RepairMode, RepairPlan, listing_is_truth, plan_repair}; +use crate::kv::{KvEntry, KvUpdate, MAX_CURSOR_BYTES, Retention, VersionToken, WatchCursor}; +use crate::protocol::{RepairMode, RepairPlan, listing_truth, plan_repair}; const MAGIC: &[u8; 4] = b"PGSS"; // v2: Put/Delete records store the version as length-prefixed raw bytes instead @@ -126,7 +126,7 @@ impl Snapshot { where I: IntoIterator, { - let truth = retention.map(|r| listing_is_truth(r.evicts_current_values, r.first_revision)); + let truth = listing_truth(retention, self.cursor.seen_evicting()); if plan_repair(RepairMode::Relist, truth) == RepairPlan::RefuseRelist { return None; } @@ -953,12 +953,11 @@ fn parse_cursor(data: &[u8], stored_crc: u32) -> Result<(Record<'_>, usize), Rec return Err(RecordError::Truncated); } let cursor_len = data[5] as usize; - // A version token holds at most 10 bytes inline. A larger length means a - // corrupt or incompatible record — reject it rather than letting it reach - // `VersionToken::from_raw`, which would panic. - if cursor_len > 10 { + // No cursor encoding is longer than this. A larger length means a corrupt + // or incompatible record — reject it before framing on it. + if cursor_len > MAX_CURSOR_BYTES { return Err(RecordError::Invalid(format!( - "cursor length {cursor_len} exceeds max version token size (10)" + "cursor length {cursor_len} exceeds the longest cursor encoding ({MAX_CURSOR_BYTES})" ))); } let total = 6 + cursor_len; @@ -972,15 +971,13 @@ fn parse_cursor(data: &[u8], stored_crc: u32) -> Result<(Record<'_>, usize), Rec return Err(RecordError::CrcMismatch { consumed: total }); } - // The `cursor_len > 10` check above bounds this to the inline capacity, so - // `from_raw` always returns `Some` here; the guard makes that explicit. - let version = VersionToken::from_raw(&data[6..total]).ok_or_else(|| { + let cursor = WatchCursor::from_bytes(&data[6..total]).ok_or_else(|| { RecordError::Invalid(format!( - "cursor length {cursor_len} exceeds max version token size (10)" + "cursor record ({cursor_len} bytes) is not a cursor this build can read" )) })?; - Ok((Record::Cursor(WatchCursor::from_version(version)), total)) + Ok((Record::Cursor(cursor), total)) } // --------------------------------------------------------------------------- @@ -1093,7 +1090,7 @@ fn write_delete_record( } fn write_cursor_record(w: &mut impl Write, cursor: &WatchCursor) -> Result { - let cb = cursor.version().as_bytes(); + let cb = &cursor.to_bytes()[..]; // The record encodes the cursor length as a single byte. `VersionToken` // caps inline storage at 10 bytes, so this never trips today — but checking // here rather than casting means a future backend that widens the token @@ -1165,7 +1162,7 @@ fn compact_to_file( + if cursor.is_none() { 0 } else { - 4 + 1 + 1 + cursor.version().as_bytes().len() + 4 + 1 + 1 + cursor.to_bytes().len() }; let capacity = estimated.clamp(8 * 1024, 1024 * 1024); let mut buf = io::BufWriter::with_capacity(capacity, tempfile::NamedTempFile::new_in(dir)?); diff --git a/src/snapshot_fjall.rs b/src/snapshot_fjall.rs index 4ff0ae2..d4d307a 100644 --- a/src/snapshot_fjall.rs +++ b/src/snapshot_fjall.rs @@ -64,7 +64,7 @@ use fjall::config::{BlockSizePolicy, PinningPolicy}; use fjall::{Database, Keyspace, KeyspaceCreateOptions, PersistMode}; use crate::artifact::{ExportManifest, ExportStage, verify_and_stage_import}; -use crate::kv::{KvEntry, KvUpdate, VersionToken, WatchCursor}; +use crate::kv::{KvEntry, KvUpdate, WatchCursor}; use crate::snapshot::{SnapshotError, SnapshotStore}; use crate::snapshot_record::{decode_entry, encode_value_into}; @@ -189,14 +189,12 @@ impl FjallSnapshot { .map_err(map_fjall)?; let cursor = match meta.get(CURSOR_KEY).map_err(map_fjall)? { - Some(raw) => VersionToken::from_raw(&raw) - .map(WatchCursor::from_version) - .ok_or_else(|| { - SnapshotError::InvalidFormat(format!( - "stored cursor is {} bytes, exceeds version token capacity", - raw.len() - )) - })?, + Some(raw) => WatchCursor::from_bytes(&raw).ok_or_else(|| { + SnapshotError::InvalidFormat(format!( + "stored cursor ({} bytes) is not a cursor this build can read", + raw.len() + )) + })?, None => WatchCursor::none(), }; @@ -469,7 +467,7 @@ impl SnapshotStore for FjallSnapshot { } } // Cursor in the SAME batch as the data it names. - wb.insert(&self.meta, CURSOR_KEY, cursor.version().as_bytes()); + wb.insert(&self.meta, CURSOR_KEY, cursor.to_bytes()); wb.commit().map_err(map_fjall)?; self.cursor = cursor.clone(); diff --git a/src/snapshot_rocksdb.rs b/src/snapshot_rocksdb.rs index 00786be..f619ee5 100644 --- a/src/snapshot_rocksdb.rs +++ b/src/snapshot_rocksdb.rs @@ -83,7 +83,7 @@ use rocksdb::{ }; use crate::artifact::{ExportManifest, ExportStage, verify_and_stage_import}; -use crate::kv::{KvEntry, KvUpdate, VersionToken, WatchCursor}; +use crate::kv::{KvEntry, KvUpdate, WatchCursor}; use crate::snapshot::{SnapshotError, SnapshotStore}; use crate::snapshot_record::{decode_entry, encode_value_into}; @@ -334,14 +334,12 @@ impl RocksDbSnapshot { .get_cf(cf(&db, META_CF)?, CURSOR_KEY) .map_err(map_rocksdb)? { - Some(raw) => VersionToken::from_raw(&raw) - .map(WatchCursor::from_version) - .ok_or_else(|| { - SnapshotError::InvalidFormat(format!( - "stored cursor is {} bytes, exceeds version token capacity", - raw.len() - )) - })?, + Some(raw) => WatchCursor::from_bytes(&raw).ok_or_else(|| { + SnapshotError::InvalidFormat(format!( + "stored cursor ({} bytes) is not a cursor this build can read", + raw.len() + )) + })?, None => WatchCursor::none(), }; @@ -561,7 +559,7 @@ impl SnapshotStore for RocksDbSnapshot { } } // Cursor in the SAME batch as the data it names. - wb.put_cf(meta, CURSOR_KEY, cursor.version().as_bytes()); + wb.put_cf(meta, CURSOR_KEY, cursor.to_bytes()); // The WAL is always on; `set_sync` only toggles the per-commit fsync. // NO_SYNC (sync: false) reaches the OS — survives a process crash via WAL diff --git a/src/transport.rs b/src/transport.rs index 232e6f7..fa473a7 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -858,7 +858,9 @@ impl RestoreSource for ArtifactRestore { }) .await .map_err(|e| SnapshotError::Backend(format!("restore import task panicked: {e}")))??; - Ok(RestoredFold::new(manifest, fold).holding(tmp)) + Ok(RestoredFold::new(manifest, fold) + .scratch_in(tmp.path()) + .holding(tmp)) } } diff --git a/tests/eviction.rs b/tests/eviction.rs index 9a003fd..5ce781d 100644 --- a/tests/eviction.rs +++ b/tests/eviction.rs @@ -446,7 +446,9 @@ async fn relist_on_evicting_bucket_is_refused() { assert!(err.to_string().contains("evicts current values"), "{err}"); let (after, fold) = h.open("watcher.snap"); - assert_eq!(after, cursor, "cursor untouched"); + // The refusal read retention and saved the eviction memo; the position + // is untouched. + assert_eq!(after.as_u64(), cursor.as_u64(), "cursor untouched"); assert_eq!( fold.get("route.keep").unwrap().map(|e| e.value), Some(b"v1".to_vec()), @@ -477,7 +479,7 @@ async fn stale_artifact_fails_the_watch() { .expect_err("a stale artifact must fail the watch"); assert!(err.to_string().contains("retention window"), "{err}"); let (after, fold) = h.open("watcher.snap"); - assert_eq!(after, cursor); + assert_eq!(after.as_u64(), cursor.as_u64(), "position untouched"); assert!(fold.get("route.keep").unwrap().is_some()); } @@ -595,7 +597,8 @@ async fn live_floor_guard_trip_restores_in_process() { let (wc, wfold) = h.open("watcher.snap"); let (xc, xfold) = h.open("exporter.snap"); - assert_eq!(wc, xc); + // Same position; only the watcher has read retention since it evicted. + assert_eq!(wc.as_u64(), xc.as_u64()); assert_eq!( wfold.get("route.keep").unwrap().map(|e| e.value), Some(b"v1".to_vec()), @@ -616,18 +619,39 @@ async fn live_floor_guard_trip_restores_in_process() { ); } -/// 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. +/// A node on a heartbeat bucket that has read retention after the log moved +/// past its first revision, so its cursor keeps the eviction memo and the +/// time its message was written. (A node starting with no cursor on such a +/// log would seed from an artifact instead.) +async fn seed_heartbeat_node(h: &Harness) -> Node { + let w = h.bucket.writer().unwrap(); + w.put("hb.a", b"0").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 rev = w.put("hb.a", b"1").await.unwrap().as_u64().unwrap(); + node.wait_applied(rev).await; + let c = node.stop().await.unwrap(); + // The restart reads retention first thing: `max_age`, log past revision 1. + let (_c, fold) = h.open("node.snap"); + let node = Node::spawn(&h.bucket, fold, Some(c), h.auto()); + let rev = w.put("hb.b", b"1").await.unwrap().as_u64().unwrap(); + node.wait_applied(rev).await; + node +} + +/// A supersession-only gap on a `max_age` bucket resumes instead of +/// fail-stopping. The resume check sees only the stream's first retained +/// revision, so when every message after a node's cursor was SUPERSEDED +/// while it was down (every key rewritten since — a heartbeat bucket on a +/// short restart), the cursor looks expired though nothing aged out. With no +/// artifact ahead of the node, a restore can't run. But the cursor keeps the +/// time its message was written: it is far younger than `max_age`, so no +/// later message can have aged out, and the watch resumes from the first +/// retained revision. #[tokio::test(flavor = "multi_thread")] -async fn supersession_only_restart_fail_stops_without_a_newer_artifact() { +async fn supersession_only_restart_resumes_on_a_max_age_bucket() { let h = Harness::new(StoreConfig { name: "routes".into(), max_age: Some(Duration::from_secs(3600)), @@ -635,11 +659,7 @@ async fn supersession_only_restart_fail_stops_without_a_newer_artifact() { }) .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 node = seed_heartbeat_node(&h).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; @@ -650,7 +670,7 @@ async fn supersession_only_restart_fail_stops_without_a_newer_artifact() { // 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(); + let head = w.put("hb.a", b"4").await.unwrap().as_u64().unwrap(); assert!( h.first_sequence().await > cursor + 1, "supersession alone passed the cursor" @@ -658,7 +678,47 @@ async fn supersession_only_restart_fail_stops_without_a_newer_artifact() { 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) + node.wait_applied(head).await; + node.stop().await.expect("resumed, no fail-stop"); + let (c, fold) = h.open("node.snap"); + assert_eq!(c.as_u64(), Some(head)); + assert_eq!( + fold.get("hb.a").unwrap().map(|e| e.value), + Some(b"4".to_vec()) + ); + assert_eq!( + fold.get("hb.b").unwrap().map(|e| e.value), + Some(b"3".to_vec()) + ); +} + +/// The same gap where `discard: old` can also evict: the time proves nothing +/// about a limit, so the resume still reports the cursor expired and, with no +/// artifact ahead of the node, fail-stops. +#[tokio::test(flavor = "multi_thread")] +async fn supersession_gap_still_expires_when_eviction_is_not_only_by_age() { + let h = Harness::new(StoreConfig { + name: "routes".into(), + max_age: Some(Duration::from_secs(3600)), + max_bytes: Some(1 << 20), + discard: DiscardPolicy::Old, + ..Default::default() + }) + .await; + let w = h.bucket.writer().unwrap(); + let node = seed_heartbeat_node(&h).await; + 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(); + 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); + + let (c, fold) = h.open("node.snap"); + let node = Node::spawn(&h.bucket, fold, Some(c), h.auto()); + let err = timeout(Duration::from_secs(10), node.task) .await .expect("fails fast") .unwrap() @@ -668,3 +728,91 @@ async fn supersession_only_restart_fail_stops_without_a_newer_artifact() { "{err}" ); } + +/// Eviction turned off after it dropped a key: the bucket's config now says +/// it keeps current values, so a planner reading only the config trusts the +/// key listing and the key-listing diff deletes `route.keep`, which aged out +/// while eviction was on. The fold remembers that it saw eviction (the memo +/// on its cursor), so `Auto` routes the expiry to a restore instead: with no +/// artifact ahead it fails, and `route.keep` stays. +#[tokio::test(flavor = "multi_thread")] +async fn eviction_turned_off_never_trusts_the_listing_again() { + let h = Harness::new(Eviction::DiscardOld.config()).await; + let w = h.bucket.writer().unwrap(); + w.put("route.keep", b"v1").await.unwrap(); + let (_c, fold) = h.open("node.snap"); + let node = Node::spawn(&h.bucket, fold, None, h.auto()); + let rev = Eviction::DiscardOld.evict(&h.bucket).await; + node.wait_applied(rev).await; + let cursor = node.stop().await.expect("ran clean"); + assert!( + h.bucket.reader().get("route.keep").await.unwrap().is_none(), + "route.keep must have aged out" + ); + + // A restart while eviction is still on reads retention, sees that it has + // evicted, and commits the memo before the watch delivers anything. + let (c, fold) = h.open("node.snap"); + assert_eq!(c, cursor); + let node = Node::spawn(&h.bucket, fold, Some(c), h.auto()); + let rev = w.put("route.marker", b"m").await.unwrap().as_u64().unwrap(); + node.wait_applied(rev).await; + let cursor = node.stop().await.expect("ran clean"); + let (c, _) = h.open("node.snap"); + assert!( + format!("{c:?}").contains("seen evicting"), + "the fold remembers the eviction: {c:?}" + ); + let cursor = cursor.as_u64().unwrap(); + + // The operator turns eviction off. + let js = async_nats::jetstream::new(async_nats::connect(&h._nats.url).await.unwrap()); + let mut stream = js.get_stream("KV_routes").await.unwrap(); + let mut config = stream.info().await.unwrap().config.clone(); + config.max_bytes = -1; + js.update_stream(config).await.unwrap(); + let retention = h + .bucket + .watcher() + .unwrap() + .retention() + .await + .unwrap() + .unwrap(); + assert!( + !retention.evicts_current_values, + "the config no longer evicts" + ); + + // While the node is down, every key is rewritten: supersession alone + // expires its cursor. + // Twice, so even the first rewrite (the message right after the cursor) + // is superseded. + let reader = h.bucket.reader(); + let keys = reader.keys("").await.unwrap(); + for round in [b"again".as_slice(), b"again2"] { + for key in &keys { + w.put(key, round).await.unwrap(); + } + } + assert!( + h.first_sequence().await > cursor + 1, + "supersession passed the cursor" + ); + + let (c, fold) = h.open("node.snap"); + let node = Node::spawn(&h.bucket, fold, Some(c.clone()), h.auto()); + let err = timeout(Duration::from_secs(15), node.task) + .await + .expect("fails fast") + .unwrap() + .expect_err("no artifact to restore from: the listing must not be trusted"); + assert!(err.to_string().contains("artifact"), "{err}"); + let (after, fold) = h.open("node.snap"); + assert_eq!(after.as_u64(), c.as_u64(), "cursor untouched"); + assert_eq!( + fold.get("route.keep").unwrap().map(|e| e.value), + Some(b"v1".to_vec()), + "the key that aged out while eviction was on is still there" + ); +} diff --git a/tests/model.rs b/tests/model.rs index 434129d..d6a1998 100644 --- a/tests/model.rs +++ b/tests/model.rs @@ -779,6 +779,7 @@ impl Model for SnapshotProtocol { || listing_is_truth( self.retention == BucketRetention::EvictsCurrent, s.floor as u64 + 1, + false, ); s.importer = match plan_repair(mode, Some(truth)) { RepairPlan::ReListOnly => { diff --git a/tests/model_fleet.rs b/tests/model_fleet.rs index 44fac85..120c889 100644 --- a/tests/model_fleet.rs +++ b/tests/model_fleet.rs @@ -119,7 +119,11 @@ impl Fleet { if self.mutation == Mutation::RelistOnEvicting { return Some(true); } - Some(listing_is_truth(self.evicting, Self::first_revision(s))) + Some(listing_is_truth( + self.evicting, + Self::first_revision(s), + false, + )) } /// Does an expired node repair by the artifact restore (else the /// key-listing diff)? The shared planner, under the shipped diff --git a/tests/model_live_watch.rs b/tests/model_live_watch.rs index e006396..a7bfa11 100644 --- a/tests/model_live_watch.rs +++ b/tests/model_live_watch.rs @@ -198,6 +198,7 @@ impl LiveWatch { || listing_is_truth( self.retention == BucketRetention::EvictsCurrent, s.floor as u64 + 1, + false, ); match plan_repair(RepairMode::Auto, Some(truth)) { RepairPlan::Restore => true, diff --git a/tests/model_repair.rs b/tests/model_repair.rs index 8c4461f..daa52dd 100644 --- a/tests/model_repair.rs +++ b/tests/model_repair.rs @@ -72,9 +72,9 @@ struct Upd { enum Req { /// Key-listing diff; what the listing showed for the key. Relist { listed: Option }, - /// Artifact restore to this cursor, with what the live-key listing (taken - /// after the fetch) showed for the key. - Restore { target: u8, listed: Option }, + /// Artifact restore to this cursor. The main loop lists the live keys + /// itself, after the diff and before folding it (`repair::restore`). + Restore { target: u8 }, } #[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] @@ -205,6 +205,15 @@ enum Mutation { RestoreIgnoresVersions, /// The restore deletes keys the bucket lists live (the pre-fix diff). RestoreIgnoresListing, + /// The restore trusts its live-key listing without re-checking, after + /// taking it, that the log still holds everything past the artifact's + /// cursor: a write after the cursor that aged out first is unlisted, and + /// deleted. + NoListingRecheck, + /// The fold forgets that it saw retention evict current values: once the + /// operator turns eviction off (`config_forgot`), the planner trusts the + /// listing again, and the key-listing diff deletes a key that aged out. + NoEvictionMemo, } #[derive(Clone)] @@ -216,6 +225,11 @@ struct RepairModel { mutation: Mutation, /// Let retention evict during a key-listing repair (see `actions`). drop_axiom_6: bool, + /// The bucket evicts (`evicting`), but its config no longer says so: the + /// operator turned `max_age` / `discard: old` off after the fold saw it + /// evict. The memo axiom: a fold reads retention while eviction is in + /// force, so it has seen any eviction that happened. + config_forgot: bool, /// The key's revisions run 1..=max_rev (revision 1 is its creation). max_rev: u8, /// Crash budget (process deaths, including store-apply crashes). @@ -250,7 +264,9 @@ impl RepairModel { if self.mutation == Mutation::RelistOnEvicting { return Some(true); } - Some(listing_is_truth(self.evicting, Self::first_revision(s))) + let reported = self.evicting && !self.config_forgot; + let seen = self.evicting && s.evicted && self.mutation != Mutation::NoEvictionMemo; + Some(listing_is_truth(reported, Self::first_revision(s), seen)) } fn listed(s: &St) -> Option { @@ -636,10 +652,7 @@ impl Model for RepairModel { // (the code re-checks what it fetched against now). let target = s.pointer; if self.restore_ok(&s, target, local) && target > 0 { - s.req = Some(Req::Restore { - target, - listed: Self::listed(&s), - }); + s.req = Some(Req::Restore { target }); s.w = WPhase::AwaitReply; } else { s.w = WPhase::Failed; @@ -728,18 +741,28 @@ impl Model for RepairModel { s.relisted = true; s.m = MPhase::RelistFlush(false); } - MPhase::PreFlush(Req::Restore { target, listed }, _) => { + MPhase::PreFlush(Req::Restore { target }, _) => { // The authoritative ahead check. if self.mutation != Mutation::NoRestoreGuard && !restore_ahead(target as u64, s.applied as u64) { s.reply = Some(Reply::Refused); s.m = MPhase::Idle; + } else if !matches!( + self.mutation, + Mutation::NoListingRecheck | Mutation::NoRestoreGuard + ) && !resume_window_ok(target as u64, Self::first_revision(&s)) + { + // The listing (taken here) can only be trusted for + // writes after the artifact's cursor while the log + // still holds all of them. + s.reply = Some(Reply::Refused); + s.m = MPhase::Idle; } else if self.mutation == Mutation::CursorBeforeDiff { s.batch_high = Some(target); s.m = MPhase::RestoreCommit(target, false); } else { - if let Some(u) = self.restore_op(&s, target, listed) { + if let Some(u) = self.restore_op(&s, target, Self::listed(&s)) { s.batch.push(u); s.raw.push(u); } @@ -855,6 +878,7 @@ fn shipped(evicting: bool, fresh: bool) -> RepairModel { fresh, mutation: Mutation::None, drop_axiom_6: false, + config_forgot: false, max_rev: 4, max_crashes: 1, max_transient: 2, @@ -879,6 +903,40 @@ fn evicting_repair_steps_are_correct() { run(shipped(true, false), "repair steps: evicting").assert_properties(); } +/// Eviction turned off after it dropped a key: the planner sees a config +/// that keeps current values, but the fold remembers what it saw, so `Auto` +/// still restores rather than diffing the listing. +#[test] +fn eviction_turned_off_repair_steps_are_correct() { + run( + RepairModel { + config_forgot: true, + ..shipped(true, false) + }, + "repair steps: eviction turned off", + ) + .assert_properties(); +} + +/// Without the memo, the same run deletes the key that aged out. +#[test] +fn eviction_memo_is_load_bearing() { + let checker = run( + RepairModel { + config_forgot: true, + mutation: Mutation::NoEvictionMemo, + ..shipped(true, false) + }, + "repair steps: eviction turned off, no memo", + ); + assert!( + checker + .discovery("a repair never deletes the key's current value") + .is_some(), + "the checker found no deleted aged-out key without the memo" + ); +} + #[test] fn fresh_start_repair_steps_are_correct() { run(shipped(false, true), "repair steps: keeps-current, fresh").assert_properties(); @@ -901,6 +959,7 @@ fn every_repair_step_is_load_bearing() { (Mutation::UnanchoredRelist, false, true), (Mutation::RestoreIgnoresVersions, true, false), (Mutation::RestoreIgnoresListing, true, false), + (Mutation::NoListingRecheck, true, false), ]; let names = [ "while the store's cursor is resumable, the store plus the tail is the truth", @@ -924,6 +983,8 @@ fn every_repair_step_is_load_bearing() { Mutation::UnanchoredRelist => "unanchored fold re-listed blind", Mutation::RestoreIgnoresVersions => "restore ignores versions", Mutation::RestoreIgnoresListing => "restore ignores the live listing", + Mutation::NoListingRecheck => "restore trusts a listing taken after eviction", + Mutation::NoEvictionMemo => "no eviction memo", }, if evicting { "evicting"