From 3b566e7122284ea1e6a6e22c95643c0aecd78a2c Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Tue, 20 Oct 2026 15:11:05 -0700 Subject: [PATCH] watch_applied_owned: hand the store back so a watch can resume in-process watch_applied closes its store on every exit, so a caller whose readers hold the store (a fold's FjallReader keeps its database alive) cannot re-open the path and has to restart the process on any watch error. A NATS heartbeat miss made instd do exactly that, three times in one day. watch_applied_owned is the same loop without the close. Its WatchEnd carries the result, the applied cursor committed with the store's data, and the store itself, intact on every exit, so the caller resumes by calling again with the same store and resume = applied. watch_applied is now watch_applied_owned followed by the off-runtime close. Release 0.8.5. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XoaQgogNq1z3dNoM9feGoH --- ARCHITECTURE.md | 2 + Cargo.lock | 2 +- Cargo.toml | 2 +- src/applied.rs | 212 ++++++++++++++++++++++++++++++++++++++++++++++-- src/lib.rs | 4 +- 5 files changed, 211 insertions(+), 11 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d91ecc9..46a7de8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -234,6 +234,8 @@ The one exception: an update carrying the **unknown** version (an unparseable AC **The store closes off the runtime.** However the watch ends — shutdown, stream close, or any error — `watch_applied` hands its store to a dedicated thread to drop, and waits up to 30 s for it (`close_off_runtime`). A store's destructor may block: fjall 3.1.x's `Database::drop` can deadlock joining its worker pool. Dropped on the runtime, that parked the only tokio worker of a `CPUQuota=100%` service and froze the whole process before it could act on the watch's error. A normal close finishes inside the wait, so the caller can reopen the same path; a hung close leaves its thread behind and the watch returns anyway. A std thread, not `spawn_blocking`, because runtime shutdown waits on blocking tasks (`a_store_that_hangs_on_close_cannot_freeze_the_runtime`, `the_store_is_closed_before_watch_applied_returns`). +**Resuming in-process (`watch_applied_owned`).** Closing on every exit makes a watch error fatal to a caller whose readers hold the store (a fold's `FjallReader` keeps its database alive). Such a caller can't re-open the path while those handles live, so its only recovery was a process restart, which a NATS heartbeat miss forced. `watch_applied_owned` is the same loop without the close. Its `WatchEnd` carries the result, the applied cursor (committed with the store's data), and the store itself, intact on every exit. The caller resumes by calling again with the same store and `resume = applied`: nothing is re-applied out of order and nothing is skipped, and cursor expiry and its repair take the same path as a fresh start. A store that comes back `None` was lost (a flush task panicked), and only a fresh open recovers it. `watch_applied` is `watch_applied_owned` followed by `close_off_runtime` (`owned_watch_hands_the_store_back_and_resumes_in_process`). + **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 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. diff --git a/Cargo.lock b/Cargo.lock index 6517acc..6dd135c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -143,7 +143,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" [[package]] name = "beyond-slipstream" -version = "0.8.4" +version = "0.8.5" dependencies = [ "aho-corasick", "async-nats", diff --git a/Cargo.toml b/Cargo.toml index d7e1155..89f6deb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ rust-version = "1.92" [package] name = "beyond-slipstream" -version = "0.8.4" +version = "0.8.5" edition.workspace = true license.workspace = true rust-version.workspace = true diff --git a/src/applied.rs b/src/applied.rs index 942cb25..a1facce 100644 --- a/src/applied.rs +++ b/src/applied.rs @@ -226,6 +226,78 @@ impl Default for BatchConfig { // would either box the closures or force a single generic bundle, losing that. #[allow(clippy::too_many_arguments)] pub async fn watch_applied( + watcher: Arc, + scope: WatchScope, + resume: Option, + // How an expired cursor is repaired (see the function docs). Only + // consulted on expiry (and, for a restore, at a cursor-less start on a + // bucket that has already evicted current values) — the hot path never + // touches it. + repair: R, + store: Option, + // `Some(rx)` arms an export-request arm in the select loop: each + // [`ExportRequest`] is handled between flushes (pending batch flushed + // first), so the exported artifact's cursor is the applied cursor (or, + // across a transiently failed store flush, the store's own lagging but + // self-consistent cursor — never a cursor past unfolded data). The + // artifact's manifest records this watch's scope, which a restore checks. + // `None` (or dropping the paired sender) leaves the loop's behavior + // unchanged. + exports: Option>, + config: BatchConfig, + parse: P, + apply: A, + on_applied: O, + shutdown: watch::Receiver, +) -> Result +where + U: Send, + // `Send + 'static`: each flush moves `store` onto a blocking task to run its + // (potentially blocking) `apply`, then takes it back — the same offload the + // append log's compaction always used. + S: SnapshotStore + Send + 'static, + P: FnMut(&KvUpdate) -> Option + Send, + A: FnMut(Vec) + Send, + O: FnMut(WatchCursor) + Send, + R: Into>, +{ + let end = watch_applied_owned( + watcher, scope, resume, repair, store, exports, config, parse, apply, on_applied, shutdown, + ) + .await; + if let Some(store) = end.store { + close_off_runtime(store).await; + } + end.result +} + +/// How a [`watch_applied_owned`] run ended. +pub struct WatchEnd { + /// The run's result: the applied cursor on a clean end or a shutdown, the + /// error otherwise. + pub result: Result, + /// The applied cursor at the end, error or not: everything up to it was + /// applied and, with a store, committed with it. The `resume` for the next + /// run. + pub applied: WatchCursor, + /// The store, intact. `None` when none was given, or it was lost (a + /// flush task panicked). + pub store: Option, +} + +/// [`watch_applied`], handing the store back instead of closing it. +/// +/// On every exit, a watch error included, the [`WatchEnd`] carries the store +/// intact and the applied cursor it committed with (data and cursor commit +/// together). A caller that keeps readers open on the store can therefore +/// resume in-process: call again with the same store and `resume` set to +/// [`WatchEnd::applied`]. Nothing the store holds is re-applied out of order and +/// nothing is skipped. A store that comes back `None` was lost (a flush task +/// panicked); only a fresh open recovers from that. Closing a returned store +/// is the caller's job: drop it off the runtime (fjall's `Database::drop` can +/// block; see [`watch_applied`]). +#[allow(clippy::too_many_arguments)] +pub async fn watch_applied_owned( watcher: Arc, scope: WatchScope, resume: Option, @@ -249,7 +321,7 @@ pub async fn watch_applied( apply: A, on_applied: O, mut shutdown: watch::Receiver, -) -> Result +) -> WatchEnd where U: Send, // `Send + 'static`: each flush moves `store` onto a blocking task to run its @@ -270,14 +342,27 @@ where // store to run it against (both repairs edit the fold). The watch task // sends the repair here and waits for the ack/reply before starting its // next watch. - let (repair, mut repairs) = repair::arm(repair.into(), store.is_some())?; + let start = resume.clone().unwrap_or_else(WatchCursor::none); + let early = |e: KvError, store: Option| WatchEnd { + result: Err(e), + applied: start.clone(), + store, + }; + let (repair, mut repairs) = match repair::arm(repair.into(), store.is_some()) { + Ok(armed) => armed, + Err(e) => return early(e, store), + }; // A fold with data but no cursor can't be re-listed blind (see // `repair::run_watch`). let unanchored = match &store { - Some(st) if resume.as_ref().is_none_or(WatchCursor::is_none) => st - .has_entries() - .map_err(|e| KvError::WatchError(format!("reading the fold at start failed: {e}")))?, + Some(st) if resume.as_ref().is_none_or(WatchCursor::is_none) => match st.has_entries() { + Ok(has) => has, + Err(e) => { + let e = KvError::WatchError(format!("reading the fold at start failed: {e}")); + return early(e, store); + } + }, _ => false, }; @@ -405,10 +490,11 @@ where } } .await; - if let Some(store) = fold.take_store() { - close_off_runtime(store).await; + WatchEnd { + result, + applied: fold.applied().clone(), + store: fold.take_store(), } - result } /// How long [`watch_applied`] waits for its store to close before returning @@ -1400,6 +1486,116 @@ mod tests { assert_eq!(snap.entries["node.a"].value, b"1b"); } + /// A watch that dies mid-stream (a NATS heartbeat miss) hands its store + /// back with the cursor it committed, and a second run on the same store, + /// resumed from that cursor, folds what changed during the gap: the caller + /// never has to close the store and re-open it. + #[tokio::test] + async fn owned_watch_hands_the_store_back_and_resumes_in_process() { + struct DiesAfter(Mutex>>); + #[async_trait] + impl KvWatcher for DiesAfter { + async fn watch_all(&self, tx: Sender) -> Result<(), KvError> { + let updates = self.0.lock().unwrap().take().unwrap_or_default(); + for u in updates { + let _ = tx.send(u).await; + } + Err(KvError::WatchError( + "watcher consumer error: missed idle heartbeat".into(), + )) + } + async fn watch_prefix(&self, _p: &str, tx: Sender) -> Result<(), KvError> { + self.watch_all(tx).await + } + async fn watch_prefixes( + &self, + _p: &[&str], + tx: Sender, + ) -> Result<(), KvError> { + self.watch_all(tx).await + } + } + + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().join("owned.snap"); + let (_r, store) = AppendLogSnapshot::open(&path, u64::MAX).unwrap(); + let (_sd_tx, sd_rx) = watch::channel(false); + + // Run 1: folds a@1 and b@2, then the watch dies. + let end = watch_applied_owned( + Arc::new(DiesAfter(Mutex::new(Some(vec![ + put("node.a", b"1", 1), + put("node.b", b"2", 2), + ])))), + WatchScope::All, + None, + None::>, + Some(store), + None, + BatchConfig::default(), + |_u: &KvUpdate| Some(()), + |_b: Vec<()>| {}, + |_| {}, + sd_rx.clone(), + ) + .await; + assert!( + matches!(&end.result, Err(KvError::WatchError(m)) if m.contains("missed idle heartbeat")), + "the watch error is surfaced" + ); + assert_eq!( + end.applied.as_u64(), + Some(2), + "everything delivered was applied" + ); + let store = end.store.expect("the store comes back intact"); + assert_eq!(store.get("node.a").unwrap().unwrap().value, b"1"); + + // Run 2, same store, resumed at the applied cursor: during the gap b + // was deleted and c written. The stream then closes cleanly. + let resumed = MockWatcher { + full: Mutex::new(Some(vec![])), + from: Mutex::new(Some(vec![ + KvUpdate::Delete { + key: "node.b".into(), + version: VersionToken::from_u64(3), + }, + put("node.c", b"3", 4), + ])), + from_expires: false, + hold: false, + }; + let end = watch_applied_owned( + Arc::new(resumed), + WatchScope::All, + Some(end.applied), + None::>, + Some(store), + None, + BatchConfig::default(), + |_u: &KvUpdate| Some(()), + |_b: Vec<()>| {}, + |_| {}, + sd_rx, + ) + .await; + assert_eq!(end.result.unwrap().as_u64(), Some(4)); + let store = end.store.expect("the store comes back on a clean end too"); + assert!( + store.get("node.b").unwrap().is_none(), + "the gap's delete landed" + ); + assert_eq!(store.get("node.c").unwrap().unwrap().value, b"3"); + assert_eq!(store.get("node.a").unwrap().unwrap().value, b"1"); + drop(store); + let snap = crate::snapshot::load(&path).unwrap().unwrap(); + assert_eq!( + snap.cursor.as_u64(), + Some(4), + "the cursor committed with the data" + ); + } + /// A prefix-scoped resync diffs only in-scope keys: an out-of-scope key the /// fold holds survives, the in-scope stale key is deleted, and a flush /// containing only synthetic deletes leaves the cursor untouched. diff --git a/src/lib.rs b/src/lib.rs index 179cfe4..0a69642 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -42,7 +42,9 @@ mod stores; #[cfg(feature = "transport")] mod transport; -pub use applied::{BatchConfig, ExportRequest, WatchScope, watch_applied}; +pub use applied::{ + BatchConfig, ExportRequest, WatchEnd, WatchScope, watch_applied, watch_applied_owned, +}; pub use artifact::{ARTIFACT_SCHEMA_VERSION, ArtifactFile, ExportManifest, MANIFEST_FILE}; pub use export_lease::{ExportLease, LeaseGuard, LeaseRecord}; pub use kv::{