From ec69e83353cf2f4970d737ccbd35e304c863ed6f Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Sat, 17 Oct 2026 09:08:34 -0700 Subject: [PATCH] Close the snapshot store off the runtime when watch_applied exits watch_applied dropped its store wherever the loop exited, on the tokio worker. fjall 3.1.x's Database::drop can deadlock joining its worker pool (a blocking send(Close) into its full channel after the last worker exits), and on a single-worker runtime (a CPUQuota=100% service) that froze the whole process: instd stopped pulling NATS, logging and handling signals after a KV-watch heartbeat miss, and never reached the error path that restarts it. The loop now only borrows the fold; on every exit the store is dropped on a dedicated thread and the return waits up to 30s for it. A normal close still completes before return, so callers can reopen the path; a hung close leaves its thread behind instead of the runtime. A std thread, not spawn_blocking, since runtime shutdown waits on blocking tasks. Release 0.8.4. 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 | 372 ++++++++++++++++++++++++++++++++++++------------ 4 files changed, 283 insertions(+), 95 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 75efe9c..d91ecc9 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -232,6 +232,8 @@ The one exception: an update carrying the **unknown** version (an unparseable AC **Transient store failures re-queue, not drop.** If `store.apply()` returns an error, the raw batch is prepended to the next flush's accumulation and the watch continues. The streak counter increments; at 16 consecutive failures the watch fail-stops with `KvError::WatchError`. Dropping the failed batch and continuing was the shipped behavior until `transient_store_failure_never_leaves_a_cursor_gap` reproduced the bug: a transient failure followed by a successful flush advanced the cursor over a hole that survived every restart, because the restart re-folds from the advanced cursor, skipping the missing range. Cursor authority requires the store's cursor and contents to advance together, always (`Fold::flush` in `applied.rs`, `tests/model_applied.rs`). +**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`). + **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 1da7844..6517acc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -143,7 +143,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" [[package]] name = "beyond-slipstream" -version = "0.8.3" +version = "0.8.4" dependencies = [ "aho-corasick", "async-nats", diff --git a/Cargo.toml b/Cargo.toml index 6f9a158..d7e1155 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ rust-version = "1.92" [package] name = "beyond-slipstream" -version = "0.8.3" +version = "0.8.4" edition.workspace = true license.workspace = true rust-version.workspace = true diff --git a/src/applied.rs b/src/applied.rs index 23d8c9d..942cb25 100644 --- a/src/applied.rs +++ b/src/applied.rs @@ -161,6 +161,11 @@ impl Default for BatchConfig { /// Returns the final applied cursor when the watch ends (shutdown signalled, or /// the underlying stream closed). /// +/// On every exit, `Ok` or `Err`, the store is closed on its own thread, never +/// on the runtime, and the return waits up to 30 s for that close — so a +/// caller may reopen the same path, and a store whose close hangs cannot +/// freeze the runtime or swallow the watch's result. +/// /// `store` is any [`SnapshotStore`] backend the consumer chose (the in-RAM /// [`AppendLogSnapshot`](crate::AppendLogSnapshot) default, an on-disk backend, or /// its own impl) — or `None` to run without persistence. On each flush, *after* @@ -283,117 +288,169 @@ where let applied = resume.clone().unwrap_or_else(WatchCursor::none); let mut fold = Fold::new(store, parse, apply, on_applied, config.max, applied); - // Spawn the watch task. It owns the cursor-expiry handling so the main loop - // only ever sees a clean ordered stream of updates on `rx`, plus repairs. - let (tx, mut rx) = mpsc::channel::(config.channel_capacity.max(1)); - let handle = tokio::spawn(repair::run_watch( - watcher, scope, resume, unanchored, repair, tx, - )); - // A fatal error ends the watch, and the watch task with it. - let stop = |e: KvError| { - handle.abort(); - e - }; + // The loop only borrows `fold`, so every exit path — `?`, `return`, a + // clean shutdown — comes back here with the store still owned, to be + // closed off the runtime (see `close_off_runtime`). + let result: Result = async { + // Spawn the watch task. It owns the cursor-expiry handling so the main loop + // only ever sees a clean ordered stream of updates on `rx`, plus repairs. + let (tx, mut rx) = mpsc::channel::(config.channel_capacity.max(1)); + let handle = tokio::spawn(repair::run_watch( + watcher, scope, resume, unanchored, repair, tx, + )); + // A fatal error ends the watch, and the watch task with it. + let stop = |e: KvError| { + handle.abort(); + e + }; - // A single timer future, reset in place each time a batch opens. The old - // `tokio::time::sleep(timeout)` lived inside the select arm, so it was - // re-created on every loop iteration — one Arc-backed timer-wheel entry - // allocated, registered, and immediately dropped per received update. - // Pinning one future and `reset`-ing it reuses that single allocation; the - // `if fold.is_open()` guard keeps it from firing while idle, so its - // initial already-elapsed deadline is never observed. - let sleep = tokio::time::sleep(Duration::ZERO); - tokio::pin!(sleep); - - loop { - tokio::select! { - biased; - - // Shutdown wins: flush whatever is batched (so the cursor reflects - // it), abandon any updates still in flight on the channel — they - // weren't applied, the cursor doesn't claim them, and they'll be - // re-delivered on the next resume — and return the applied cursor. - res = shutdown.changed() => { - if res.is_err() || *shutdown.borrow() { - fold.flush().await.map_err(stop)?; - handle.abort(); - // Observe the task's terminal state. An abort surfaces as a - // cancelled JoinError, which we ignore; a genuine panic that - // raced ahead of the abort is logged rather than silently lost. - if let Err(join) = handle.await - && !join.is_cancelled() - { - warn!(error = %join, "watch task panicked at shutdown"); + // A single timer future, reset in place each time a batch opens. The old + // `tokio::time::sleep(timeout)` lived inside the select arm, so it was + // re-created on every loop iteration — one Arc-backed timer-wheel entry + // allocated, registered, and immediately dropped per received update. + // Pinning one future and `reset`-ing it reuses that single allocation; the + // `if fold.is_open()` guard keeps it from firing while idle, so its + // initial already-elapsed deadline is never observed. + let sleep = tokio::time::sleep(Duration::ZERO); + tokio::pin!(sleep); + + loop { + tokio::select! { + biased; + + // Shutdown wins: flush whatever is batched (so the cursor reflects + // it), abandon any updates still in flight on the channel — they + // weren't applied, the cursor doesn't claim them, and they'll be + // re-delivered on the next resume — and return the applied cursor. + res = shutdown.changed() => { + if res.is_err() || *shutdown.borrow() { + fold.flush().await.map_err(stop)?; + handle.abort(); + // Observe the task's terminal state. An abort surfaces as a + // cancelled JoinError, which we ignore; a genuine panic that + // raced ahead of the abort is logged rather than silently lost. + if let Err(join) = handle.await + && !join.is_cancelled() + { + warn!(error = %join, "watch task panicked at shutdown"); + } + return Ok(fold.applied().clone()); } - return Ok(fold.applied().clone()); } - } - // Batch window elapsed. - () = &mut sleep, if fold.is_open() => { - fold.flush().await.map_err(stop)?; - } + // Batch window elapsed. + () = &mut sleep, if fold.is_open() => { + fold.flush().await.map_err(stop)?; + } - // Cursor-expiry repair. Placed before `rx.recv()` (biased), but - // ordering against the update stream comes from the backlog drain - // in `repair::fold_in` and the ack/reply protocol: the watch task - // is parked until this arm finishes, so nothing from its next - // watch can interleave. - req = async { repairs.as_mut().expect("arm guarded by is_some").recv().await }, - if repairs.is_some() => { - match req { - Some(req) => { - repair::fold_in(&mut fold, req, &mut rx, &prefixes) - .await - .map_err(stop)?; + // Cursor-expiry repair. Placed before `rx.recv()` (biased), but + // ordering against the update stream comes from the backlog drain + // in `repair::fold_in` and the ack/reply protocol: the watch task + // is parked until this arm finishes, so nothing from its next + // watch can interleave. + req = async { repairs.as_mut().expect("arm guarded by is_some").recv().await }, + if repairs.is_some() => { + match req { + Some(req) => { + repair::fold_in(&mut fold, req, &mut rx, &prefixes) + .await + .map_err(stop)?; + } + None => repairs = None, } - None => repairs = None, } - } - // Export request. Placed after shutdown/window (they stay prompt) - // and before `rx.recv()` so a firehose of updates cannot starve an - // export indefinitely. - req = async { exports.as_mut().expect("arm guarded by is_some").recv().await }, - if exports.is_some() => { - match req { - Some(req) => fold.export(req, &prefixes).await.map_err(stop)?, - // Sender dropped: disarm the arm for the rest of the run. - None => exports = None, + // Export request. Placed after shutdown/window (they stay prompt) + // and before `rx.recv()` so a firehose of updates cannot starve an + // export indefinitely. + req = async { exports.as_mut().expect("arm guarded by is_some").recv().await }, + if exports.is_some() => { + match req { + Some(req) => fold.export(req, &prefixes).await.map_err(stop)?, + // Sender dropped: disarm the arm for the rest of the run. + None => exports = None, + } } - } - update = rx.recv() => { - match update { - Some(u) => { - // Arm the window on the first received update of a - // batch — even a parse-rejected one, so the cursor - // advances within `window` even through a run of - // irrelevant keys. - if !fold.is_open() { - sleep.as_mut().reset(tokio::time::Instant::now() + config.window); + update = rx.recv() => { + match update { + Some(u) => { + // Arm the window on the first received update of a + // batch — even a parse-rejected one, so the cursor + // advances within `window` even through a run of + // irrelevant keys. + if !fold.is_open() { + sleep.as_mut().reset(tokio::time::Instant::now() + config.window); + } + if fold.ingest(u) { + fold.flush().await.map_err(stop)?; + } } - if fold.ingest(u) { + None => { + // Stream closed. Flush the remainder, then surface the + // watch task's terminal result: a clean end returns the + // applied cursor, an error propagates. fold.flush().await.map_err(stop)?; + return match handle.await { + Ok(Ok(())) => Ok(fold.applied().clone()), + Ok(Err(e)) => Err(e), + Err(join) => Err(KvError::WatchError(format!( + "watch task panicked: {join}" + ))), + }; } } - None => { - // Stream closed. Flush the remainder, then surface the - // watch task's terminal result: a clean end returns the - // applied cursor, an error propagates. - fold.flush().await.map_err(stop)?; - return match handle.await { - Ok(Ok(())) => Ok(fold.applied().clone()), - Ok(Err(e)) => Err(e), - Err(join) => Err(KvError::WatchError(format!( - "watch task panicked: {join}" - ))), - }; - } } } } } + .await; + if let Some(store) = fold.take_store() { + close_off_runtime(store).await; + } + result +} + +/// How long [`watch_applied`] waits for its store to close before returning +/// without it. +const STORE_CLOSE_TIMEOUT: Duration = Duration::from_secs(30); + +/// Drop `store` on a detached thread, waiting up to [`STORE_CLOSE_TIMEOUT`]. +/// +/// A store's destructor can block: fjall's `Database::drop` joins its worker +/// pool, and in fjall 3.1.x can deadlock doing so — a blocking `send(Close)` +/// into its full worker channel after the last worker has already exited. +/// Dropped on the runtime, that parks a tokio worker forever; on a +/// single-worker runtime (a service under `CPUQuota=100%`) it freezes the +/// whole process, and the caller never sees the error this watch returned. +/// +/// A std thread, not `spawn_blocking`: runtime shutdown waits on blocking +/// tasks, so a hung close there would only move the hang to process exit. A +/// normal close finishes well inside the timeout, so a caller that reopens +/// the same path after we return still finds it unlocked; past the timeout we +/// return anyway and leave the closing thread behind. +async fn close_off_runtime(store: S) { + let (done_tx, done_rx) = oneshot::channel::<()>(); + let spawned = std::thread::Builder::new() + .name("fold-close".into()) + .spawn(move || { + drop(store); + let _ = done_tx.send(()); + }); + if let Err(e) = spawned { + // `spawn` dropped the closure, and the store with it, on this thread. + warn!(error = %e, "could not spawn the store-close thread; closed inline"); + return; + } + if tokio::time::timeout(STORE_CLOSE_TIMEOUT, done_rx) + .await + .is_err() + { + warn!( + timeout_secs = STORE_CLOSE_TIMEOUT.as_secs(), + "snapshot store did not close in time; returning without it" + ); + } } /// Consecutive store-apply failures that fail-stop the watch. A transient @@ -2243,6 +2300,135 @@ mod tests { Some(b"2".to_vec()) ); } + /// A SnapshotStore that runs `on_drop` when dropped — to observe where + /// and when `watch_applied` closes its store. + struct DropProbeStore { + inner: AppendLogSnapshot, + on_drop: Option>, + } + + impl Drop for DropProbeStore { + fn drop(&mut self) { + if let Some(f) = self.on_drop.take() { + f(); + } + } + } + + impl crate::snapshot::SnapshotStore for DropProbeStore { + fn load( + _path: &std::path::Path, + ) -> Result<(WatchCursor, Self), crate::snapshot::SnapshotError> { + unreachable!("test store is constructed directly") + } + fn apply( + &mut self, + batch: &[KvUpdate], + cursor: &WatchCursor, + ) -> Result<(), crate::snapshot::SnapshotError> { + self.inner.apply(batch, cursor) + } + fn get(&self, key: &str) -> Result, crate::snapshot::SnapshotError> { + self.inner.get(key) + } + fn range(&self, prefix: &str) -> Result, crate::snapshot::SnapshotError> { + self.inner.range(prefix) + } + fn cursor(&self) -> WatchCursor { + self.inner.cursor() + } + fn export_to( + &mut self, + dest_dir: &std::path::Path, + ) -> Result { + self.inner.export_to(dest_dir) + } + } + + fn drop_probe_store( + dir: &tempfile::TempDir, + on_drop: impl FnOnce() + Send + 'static, + ) -> DropProbeStore { + let (_c, inner) = AppendLogSnapshot::open(&dir.path().join("fold.snap"), u64::MAX).unwrap(); + DropProbeStore { + inner, + on_drop: Some(Box::new(on_drop)), + } + } + + /// A store whose close never finishes (fjall 3.1.x `Database::drop` can + /// deadlock) must not take the runtime with it. This runs on a + /// current-thread runtime — the single-worker shape of a service under + /// `CPUQuota=100%` — where a store dropped on the runtime parks the only + /// worker: pre-fix, `watch_applied` never returned its error and this + /// test hung. + #[tokio::test(start_paused = true)] + async fn a_store_that_hangs_on_close_cannot_freeze_the_runtime() { + let dir = tempfile::TempDir::new().unwrap(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let store = drop_probe_store(&dir, move || { + let _ = release_rx.recv(); + }); + let (_sd_tx, sd_rx) = watch::channel(false); + + let res = watch_applied( + Arc::new(ErrorWatcher), + WatchScope::All, + None, + None, + Some(store), + None, + BatchConfig::default(), + parse_put, + |_batch: Vec>| {}, + |_| {}, + sd_rx, + ) + .await; + + assert!( + matches!(&res, Err(KvError::WatchError(m)) if m.contains("injected watch failure")), + "the watch's own error reaches the caller, got {res:?}" + ); + // Let the abandoned close thread finish. + release_tx.send(()).unwrap(); + } + + /// A store that closes normally is closed BEFORE `watch_applied` returns, + /// so a caller can reopen the same path (fjall holds a lock file until + /// its `Database` drops). Real time, not paused: a paused clock + /// auto-advances while the runtime idles on the close thread, firing the + /// close timeout at once. + #[tokio::test] + async fn the_store_is_closed_before_watch_applied_returns() { + let dir = tempfile::TempDir::new().unwrap(); + let closed = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let c = Arc::clone(&closed); + let store = drop_probe_store(&dir, move || c.store(true, Ordering::SeqCst)); + let watcher = Arc::new(MockWatcher::new(vec![put("a", b"1", 1)], true)); + let (sd_tx, sd_rx) = watch::channel(false); + + let task = tokio::spawn(watch_applied( + watcher, + WatchScope::All, + None, + None, + Some(store), + None, + BatchConfig::default(), + parse_put, + |_batch: Vec>| {}, + |_| {}, + sd_rx, + )); + tokio::time::sleep(Duration::from_millis(50)).await; + sd_tx.send(true).unwrap(); + let cursor = task.await.unwrap().unwrap(); + + assert_eq!(cursor.as_u64(), Some(1)); + assert!(closed.load(Ordering::SeqCst), "store closed before return"); + } + /// A reader whose live-key listing always fails — the resync's I/O /// failure mode. struct FailingReader;