From 22decafce495a76b92dd5c2174c0ed96e1dfd595 Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Sat, 3 Oct 2026 14:29:48 -0700 Subject: [PATCH] Bound the key-listing resync's memory: sort the listing on disk, merge-join the fold The cursor-expiry key-listing diff held every live key name (a Vec plus a HashSet), which at a billion keys exhausts memory on exactly the on-disk backends built for that size. The listing now streams into an external sort (64 MiB runs spilled to scratch) and is merge-joined against the fold's own key-ordered scan; only the stale keys are held. Scratch is the restore source's directory under Auto, TMPDIR under Relist. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0152kQKDdP8XRhoeqisJYpWr --- ARCHITECTURE.md | 2 +- README.md | 2 +- src/applied.rs | 5 - src/repair.rs | 356 +++++++++++++++++++++++++++++++++--------- src/transport.rs | 4 + tests/model_repair.rs | 43 +++-- 6 files changed, 316 insertions(+), 96 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index fca4019..89a2206 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -234,7 +234,7 @@ 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), 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. +- **Key-listing diff** (`Relist`; `Auto` on a bucket that keeps current values). The watch task hands the main loop the reader and waits for an ack; the main loop lists the bucket's live keys, sorting them externally (`KeySorter`: 64 MiB runs spilled to the restore source's scratch dir under `Auto`, else `TMPDIR`), merge-joins that against the fold's own key-ordered scan — so neither the listing nor the fold is held in memory, only the stale keys — 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. (With the listing taken by the main loop after its drain, `tests/model_repair.rs` shows axiom 6 alone already keeps it correct; the barrier is defense in depth.) **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. diff --git a/README.md b/README.md index c26fe92..7dd95c8 100644 --- a/README.md +++ b/README.md @@ -378,7 +378,7 @@ Turning eviction off later doesn't make the key list trustworthy again: keys tha 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. +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. The key-listing repair likewise sorts the bucket's listing on disk and walks it alongside the fold, holding only the keys it deletes. Under `Auto` it uses the restore source's scratch directory; under `Relist` it uses `TMPDIR`, so point that at disk, not tmpfs, for large buckets. `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. diff --git a/src/applied.rs b/src/applied.rs index d1eb817..23d8c9d 100644 --- a/src/applied.rs +++ b/src/applied.rs @@ -717,11 +717,6 @@ where self.open } - /// The store, between flushes. - pub(crate) fn store(&self) -> Option<&S> { - self.store.as_ref() - } - /// Move the store out (to a blocking task); [`put_store`](Self::put_store) /// hands it back. A flush in between would skip the store fold. pub(crate) fn take_store(&mut self) -> Option { diff --git a/src/repair.rs b/src/repair.rs index 2c3acb2..8f32ae5 100644 --- a/src/repair.rs +++ b/src/repair.rs @@ -137,6 +137,14 @@ pub trait RestoreSource: Send + Sync { /// temporary fold. It may be newer than what [`latest`](Self::latest) /// returned; the caller re-checks it. async fn fetch(&self) -> Result, SnapshotError>; + + /// A directory on the fold's filesystem for a repair's scratch files. + /// [`ExpiryRepair::Auto`]'s key-listing diff sorts the bucket's listing + /// there; without one (and always under [`ExpiryRepair::Relist`]) it uses + /// the system temp dir (`TMPDIR`), which may be RAM-backed. + fn scratch_dir(&self) -> Option { + None + } } /// A verified artifact opened as a temporary, read-only fold, plus anything @@ -367,24 +375,140 @@ pub(crate) struct SpillReader { } impl SpillReader { + /// The next key, in the order they were pushed. + pub(crate) fn next_key(&mut self) -> std::io::Result> { + if self.left == 0 { + return Ok(None); + } + 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)?; + self.left -= 1; + String::from_utf8(key) + .map(Some) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e)) + } + /// 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; + while keys.len() < max { + match self.next_key()? { + Some(key) => keys.push(key), + None => break, + } } Ok(keys) } } +/// How many bytes of keys a [`KeySorter`] holds before spilling a sorted run. +const SORT_RUN_BYTES: usize = 64 << 20; + +/// An external sort of key names: keys accumulate up to a run's worth, each +/// full run is sorted and spilled ([`KeySpill`]), and [`sorted`](Self::sorted) +/// merges the runs. Memory is one run, however many keys pass through; a +/// listing that fits in one run never touches disk. +pub(crate) struct KeySorter { + dir: Option, + run_bytes: usize, + buf: Vec, + held: usize, + runs: Vec, +} + +impl KeySorter { + pub(crate) fn new(dir: Option) -> Self { + Self::with_run_bytes(dir, SORT_RUN_BYTES) + } + + pub(crate) fn with_run_bytes(dir: Option, run_bytes: usize) -> Self { + Self { + dir, + run_bytes, + buf: Vec::new(), + held: 0, + runs: Vec::new(), + } + } + + pub(crate) fn push(&mut self, key: String) -> std::io::Result<()> { + self.held += key.len() + std::mem::size_of::(); + self.buf.push(key); + if self.held >= self.run_bytes { + self.spill_run()?; + } + Ok(()) + } + + fn spill_run(&mut self) -> std::io::Result<()> { + self.buf.sort_unstable(); + self.buf.dedup(); + let mut spill = KeySpill::new(self.dir.as_deref())?; + for key in self.buf.drain(..) { + spill.push(&key)?; + } + self.runs.push(spill.into_reader()?); + self.held = 0; + Ok(()) + } + + /// Every key pushed, ascending, each once. + pub(crate) fn sorted(mut self) -> std::io::Result { + if self.runs.is_empty() { + self.buf.sort_unstable(); + self.buf.dedup(); + return Ok(SortedKeys::Held(self.buf.into_iter())); + } + if !self.buf.is_empty() { + self.spill_run()?; + } + let mut heads = std::collections::BinaryHeap::new(); + for (i, run) in self.runs.iter_mut().enumerate() { + if let Some(key) = run.next_key()? { + heads.push(std::cmp::Reverse((key, i))); + } + } + Ok(SortedKeys::Merged { + runs: self.runs, + heads, + last: None, + }) + } +} + +/// [`KeySorter`]'s output: ascending, without duplicates. +pub(crate) enum SortedKeys { + Held(std::vec::IntoIter), + Merged { + runs: Vec, + heads: std::collections::BinaryHeap>, + last: Option, + }, +} + +impl SortedKeys { + pub(crate) fn next_key(&mut self) -> std::io::Result> { + match self { + SortedKeys::Held(keys) => Ok(keys.next()), + SortedKeys::Merged { runs, heads, last } => loop { + let Some(std::cmp::Reverse((key, i))) = heads.pop() else { + return Ok(None); + }; + if let Some(next) = runs[i].next_key()? { + heads.push(std::cmp::Reverse((next, i))); + } + // Runs are deduplicated within, not across. + if last.as_deref() != Some(key.as_str()) { + *last = Some(key.clone()); + return Ok(Some(key)); + } + }, + } + } +} + /// Read one chunk of a restore's rewritten entries back from the artifact /// fold. pub(crate) fn materialize( @@ -425,11 +549,13 @@ 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. + /// The key-listing diff: the reader that lists the bucket's live keys, + /// and where to sort that listing. 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. Relist { - live_keys: Vec, + reader: Arc, + scratch: Option, ack: oneshot::Sender<()>, }, /// The artifact restore: a verified artifact fold, the reader that lists @@ -457,7 +583,11 @@ enum Plan<'a, S> { /// Fall back to the re-list alone (nothing armed). ReListOnly, /// The key-listing diff, then the re-list. - Relist(&'a Arc, &'a mpsc::Sender>), + Relist( + &'a Arc, + Option, + &'a mpsc::Sender>, + ), /// The artifact restore, then a resume from the artifact's cursor. Restore( &'a Arc, @@ -690,11 +820,11 @@ pub(crate) async fn run_watch( .watching(watcher, watch_scope(watcher, scope, tx)) .await; } - Plan::Relist(reader, repairs) => { + Plan::Relist(reader, scratch, repairs) => { warn!( "watch cursor expired; resyncing stale keys, then falling back to the full re-list" ); - resync_stale_keys(scope, reader, repairs).await?; + resync_stale_keys(reader, scratch, repairs).await?; return memo .watching(watcher, watch_scope(watcher, scope, tx)) .await; @@ -760,7 +890,7 @@ async fn watch_scope_from( /// Decide how to repair one expiry ([`plan_repair`]), reading the bucket's /// retention live when the choice depends on it. -async fn plan<'a, S>( +async fn plan<'a, S: Send>( watcher: &dyn KvWatcher, repair: Option<&'a RepairHandle>, memo: &mut EvictionMemo<'_, S>, @@ -786,8 +916,9 @@ async fn plan<'a, S>( error!(msg); return Err(KvError::WatchError(msg.into())); } - (RepairPlan::Relist, ExpiryRepair::Relist(reader) | ExpiryRepair::Auto { reader, .. }) => { - Plan::Relist(reader, &h.tx) + (RepairPlan::Relist, ExpiryRepair::Relist(reader)) => Plan::Relist(reader, None, &h.tx), + (RepairPlan::Relist, ExpiryRepair::Auto { reader, restore }) => { + Plan::Relist(reader, restore.scratch_dir(), &h.tx) } ( RepairPlan::Restore, @@ -860,8 +991,8 @@ async fn restore_from_artifact( } /// Cursor-expired stale-key resync (the key-listing diff), run BEFORE the -/// fallback watch is established: list the scope's live keys, hand them to the -/// main loop (which diffs them against the fold and applies synthetic +/// fallback watch is established: hand the main loop the reader (it lists the +/// scope's live keys, diffs them against the fold, and applies synthetic /// deletes), and wait for the ack. That ordering — deletes applied, then /// fallback watch armed — is what makes a delete-then-recreate during the gap /// converge: the synthetic delete always lands before the re-list put. @@ -878,33 +1009,22 @@ async fn restore_from_artifact( /// restart re-resumes, hits `CursorExpired` again, and retries the resync from /// scratch. async fn resync_stale_keys( - scope: &WatchScope, reader: &Arc, + scratch: Option, repairs: &mpsc::Sender>, ) -> Result<(), KvError> { - let mut live_keys = Vec::new(); - for prefix in scope.prefixes() { - match reader.keys(&prefix).await { - Ok(keys) => live_keys.extend(keys), - Err(e) => { - return Err(KvError::WatchError(format!( - "cursor-expired resync failed listing live keys under {prefix:?}: {e}; \ - failing the watch rather than silently keeping stale keys" - ))); - } - } - } let (ack_tx, ack_rx) = oneshot::channel(); if repairs .send(RepairRequest::Relist { - live_keys, + reader: Arc::clone(reader), + scratch, ack: ack_tx, }) .await .is_ok() { - // A dropped ack (main loop shutting down) just means the fallback watch - // is about to die with it; nothing to recover. + // A dropped ack (main loop shut down, or the resync failed and ended + // it) means the fallback watch is about to die with it. let _ = ack_rx.await; } Ok(()) @@ -956,8 +1076,12 @@ where fold.settle().await?; match req { RepairRequest::Evicting => unreachable!("handled above"), - RepairRequest::Relist { live_keys, ack } => { - relist(fold, &live_keys, prefixes).await?; + RepairRequest::Relist { + reader, + scratch, + ack, + } => { + relist(fold, &*reader, scratch, prefixes).await?; // Ack AFTER the deletes are applied: the watch task is holding // the fallback watch until it hears back, which is what orders // deletes before the re-list (tests/model_resync_order.rs proves @@ -1009,9 +1133,15 @@ where /// the fold holds and the bucket no longer lists — they vanished during the /// gap (their delete markers evicted with the cursor), and the re-list can't /// deliver a delete. +/// +/// Memory stays bounded at any fold size: the listing is sorted externally +/// ([`KeySorter`], runs spilled to `scratch`) and merge-joined against the +/// fold's own key-ordered scan, so neither the listing nor the fold is ever +/// held. Only the stale keys are — the keys deleted during the gap. async fn relist( fold: &mut Fold, - live_keys: &[String], + reader: &dyn KvReader, + scratch: Option, prefixes: &[String], ) -> Result<(), KvError> where @@ -1021,50 +1151,95 @@ where A: FnMut(Vec) + Send, O: FnMut(WatchCursor) + Send, { - let live: HashSet<&str> = live_keys.iter().map(String::as_str).collect(); - let mut stale: Vec = Vec::new(); - if let Some(st) = fold.store() { - for prefix in prefixes { - // Stream the fold's keys rather than `range()`, which buffers every - // in-scope entry — values included — into one Vec. On an on-disk - // backend holding a fold larger than RAM (the case those backends - // exist for), an All-scope resync would materialize the entire - // fold on the repair path. Only the keys matter. - if let Err(e) = st.for_each_in_range(prefix, |entry| { - if !live.contains(entry.key.as_str()) { - stale.push(entry.key); - } - Ok(()) - }) { - // FATAL, not a degrade: an incomplete diff silently leaves - // deleted keys in the fold forever (tests/model.rs proves the - // divergence reachable under degrade semantics). Fail the - // watch; the restart re-runs the resume → expiry → resync from - // scratch. - warn!(error = %e, prefix = %prefix, - "resync fold scan failed; aborting watch rather than diverging"); - return Err(KvError::WatchError(format!( - "cursor-expired resync failed listing fold prefix {prefix:?}: {e}" - ))); - } + // Disjoint and in order, so the fold's per-prefix scans run in one + // ascending sequence the sorted listing can be walked alongside. + let mut prefixes = disjoint(prefixes); + prefixes.sort_unstable(); + + // The listing streams into a sorter on a blocking thread. The send blocks + // only if sorting falls behind the network. + let (tx, rx) = std::sync::mpsc::sync_channel::(8192); + let sorter = tokio::task::spawn_blocking(move || { + let mut sorter = KeySorter::new(scratch); + for key in rx { + sorter.push(key)?; } + sorter.sorted() + }); + for prefix in &prefixes { + reader + .for_each_key(prefix, &mut |key| { + let _ = tx.send(key); + }) + .await + .map_err(|e| { + repair_fatal(format!( + "cursor-expired resync failed listing live keys under {prefix:?}: {e}; \ + failing the watch rather than silently keeping stale keys" + )) + })?; } - // Overlapping prefixes can list a key twice. - stale.sort_unstable(); - stale.dedup(); + drop(tx); + let mut live = sorter + .await + .map_err(|e| repair_fatal(format!("cursor-expired resync: sort task panicked: {e}")))? + .map_err(|e| repair_fatal(format!("cursor-expired resync: sorting the listing: {e}")))?; + + let Some(st) = fold.take_store() else { + unreachable!("repairs are armed only with a store") + }; + let (st, stale) = tokio::task::spawn_blocking(move || { + let mut stale: Vec = Vec::new(); + let res = (|| { + let mut next = live.next_key()?; + for prefix in &prefixes { + // Stream the fold's keys rather than `range()`, which buffers + // every in-scope entry — values included. + st.for_each_in_range(prefix, |entry| { + while next.as_ref().is_some_and(|k| *k < entry.key) { + next = live.next_key()?; + } + if next.as_deref() != Some(entry.key.as_str()) { + stale.push(entry.key); + } + Ok(()) + })?; + } + Ok::<_, SnapshotError>(()) + })(); + (st, res.map(|()| stale)) + }) + .await + .map_err(|e| repair_fatal(format!("cursor-expired resync: diff task panicked: {e}")))?; + fold.put_store(st); + // FATAL, not a degrade: an incomplete diff silently leaves deleted keys in + // the fold forever (tests/model.rs proves the divergence reachable under + // degrade semantics). Fail the watch; the restart re-runs the resume → + // expiry → resync from scratch. + let stale = stale.map_err(|e| { + repair_fatal(format!( + "cursor-expired resync failed diffing the fold against the listing: {e}" + )) + })?; if !stale.is_empty() { warn!( stale = stale.len(), "cursor-expired resync: deleting keys that vanished during the gap" ); } - for key in stale { - // Synthetic: carries no revision (unknown version) and so never - // advances the cursor. - fold.correct(KvUpdate::Delete { - key, - version: VersionToken::unknown(), - }); + // Synthetic deletes: no revision (unknown version), so they never advance + // the cursor. Flushed in batch-sized chunks, all before the ack. + let max = fold.batch_cap(); + let mut stale = stale.into_iter(); + loop { + let chunk: Vec = stale.by_ref().take(max).collect(); + if chunk.is_empty() { + break; + } + for key in chunk { + fold.correct(restore_delete(key)); + } + fold.flush().await?; } fold.flush().await } @@ -1414,6 +1589,33 @@ mod tests { assert_eq!(disjoint(&s(&["a.", "a."])), s(&["a."])); } + /// The external sort against an in-memory sort, across run sizes that + /// force one run, a few, and one per key, with duplicates within and + /// across runs. + #[test] + fn key_sorter_matches_sort_dedup_at_every_run_size() { + let mut keys: Vec = (0..300u32) + .map(|i| format!("k.{}", (i * 7919) % 211)) + .collect(); + keys.push(String::new()); + keys.push("k.".into()); + let mut want = keys.clone(); + want.sort(); + want.dedup(); + for run_bytes in [usize::MAX, 4096, 512, 1] { + let mut sorter = KeySorter::with_run_bytes(None, run_bytes); + for k in &keys { + sorter.push(k.clone()).unwrap(); + } + let mut sorted = sorter.sorted().unwrap(); + let mut got = Vec::new(); + while let Some(k) = sorted.next_key().unwrap() { + got.push(k); + } + assert_eq!(got, want, "run_bytes {run_bytes}"); + } + } + #[test] fn spill_round_trips_in_order_and_chunks() { let mut spill = KeySpill::new(None).unwrap(); diff --git a/src/transport.rs b/src/transport.rs index fa473a7..9c0602d 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -862,6 +862,10 @@ impl RestoreSource for ArtifactRestore { .scratch_in(tmp.path()) .holding(tmp)) } + + fn scratch_dir(&self) -> Option { + Some(self.scratch_dir.clone()) + } } impl crate::AppendLogSnapshot { diff --git a/tests/model_repair.rs b/tests/model_repair.rs index daa52dd..882ddc7 100644 --- a/tests/model_repair.rs +++ b/tests/model_repair.rs @@ -70,8 +70,9 @@ struct Upd { #[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] enum Req { - /// Key-listing diff; what the listing showed for the key. - Relist { listed: Option }, + /// Key-listing diff. The main loop lists the live keys itself, after the + /// pre-repair flush and before the diff (`repair::relist`). + Relist, /// 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 }, @@ -495,11 +496,11 @@ impl Model for RepairModel { let relist_in_flight = !self.drop_axiom_6 && (matches!(s.w, WPhase::Watch { f, guarded: false } if f < s.head) || matches!(s.w, WPhase::AwaitAck) - || matches!(s.req, Some(Req::Relist { .. })) + || matches!(s.req, Some(Req::Relist)) || matches!( s.m, - MPhase::Drain(Req::Relist { .. }) - | MPhase::PreFlush(Req::Relist { .. }, _) + MPhase::Drain(Req::Relist) + | MPhase::PreFlush(Req::Relist, _) | MPhase::RelistFlush(_) )); let evictable = if self.evicting { @@ -659,9 +660,7 @@ impl Model for RepairModel { } } RepairPlan::Relist => { - s.req = Some(Req::Relist { - listed: Self::listed(&s), - }); + s.req = Some(Req::Relist); s.w = if self.mutation == Mutation::NoAckBarrier { WPhase::Watch { f: 0, @@ -728,9 +727,9 @@ impl Model for RepairModel { } } Act::MNext => match s.m { - MPhase::PreFlush(Req::Relist { listed }, _) => { - // Diff the STORE against the listing. - if s.store_val.is_some() && listed.is_none() { + MPhase::PreFlush(Req::Relist, _) => { + // List, then diff the STORE against the listing. + if s.store_val.is_some() && Self::listed(&s).is_none() { let u = Upd { pos: None, value: None, @@ -918,6 +917,21 @@ fn eviction_turned_off_repair_steps_are_correct() { .assert_properties(); } +/// The key-listing resync stays correct without the ack barrier now that the +/// main loop takes the listing after its drain (axiom 6 suffices); the +/// barrier is defense in depth. +#[test] +fn relist_holds_without_the_ack_barrier() { + run( + RepairModel { + mutation: Mutation::NoAckBarrier, + ..shipped(false, false) + }, + "repair steps: keeps-current, no ack barrier", + ) + .assert_properties(); +} + /// Without the memo, the same run deletes the key that aged out. #[test] fn eviction_memo_is_load_bearing() { @@ -953,7 +967,12 @@ fn every_repair_step_is_load_bearing() { (Mutation::NoPreFlushRetry, false, false), (Mutation::NoPreFlushRetry, true, false), (Mutation::CursorBeforeDiff, true, false), - (Mutation::NoAckBarrier, false, false), + // Not `NoAckBarrier`: the main loop lists the live keys itself, after + // draining the backlog, so a re-list put that races the diff is + // followed by its key's delete marker on the same watch; axiom 6 + // alone keeps the resync correct. The barrier stays as defense in + // depth (and is still load-bearing in `tests/model_resync_order.rs`, + // which models a listing taken before the handoff). (Mutation::NoRestoreGuard, true, false), (Mutation::RelistOnEvicting, true, false), (Mutation::UnanchoredRelist, false, true),