diff --git a/alkstore-sqlite/src/notify.rs b/alkstore-sqlite/src/notify.rs index 3329b6e..21f79ea 100644 --- a/alkstore-sqlite/src/notify.rs +++ b/alkstore-sqlite/src/notify.rs @@ -25,7 +25,11 @@ //! (it never reopens); `try_recv` / `recv_timeout` carry //! [`Error::Closed`], distinguishable from idle's `Ok(None)`; //! `recv_timeout`'s timeout expiry without a wake is `Ok(None)` — the -//! documented idle arm, never a timeout signal. +//! documented idle arm, never a timeout signal. A `listen` issued +//! *after* a watcher death fails closed with +//! [`Error::Database`](alkstore::Error) — no new enrollment onto a +//! dead feed (which would park forever); recovery is a fresh store +//! open. //! //! Unsubscribe is the receiver's own [`Drop`]: the substrate prunes //! disconnected subscribers lazily (at the next fanout), so the @@ -40,7 +44,7 @@ use serde_json::Value; use alkstore::{BoxedFuture, Error, Result, Wake, WakeReceiver}; -use crate::seam::{sqlite_error, with_writer}; +use crate::seam::{database_error, sqlite_error, with_writer}; use crate::substrate::{SharedUpdateWatcher, Writer}; /// The bridge's tokio-channel capacity. The substrate's 1-slot sync @@ -80,7 +84,10 @@ pub(crate) fn notify( /// The listen path (see the module docs). Name validation fires at the /// entry point; a listen against a closed store fails closed with /// [`Error::Database`](alkstore::Error) (the engine-wide closed-store -/// posture — `begin_tx`/`notify` yield the same shape). +/// posture — `begin_tx`/`notify` yield the same shape); a listen +/// against a store whose watcher has *died* (the db file was replaced) +/// fails closed the same way — a new subscription never enrolls onto a +/// dead feed to park forever (review 004 item 2's new-subscriber arm). pub(crate) fn listen( watcher: Arc, closed: Arc, @@ -92,7 +99,7 @@ pub(crate) fn listen( "the store is closed", ))); } - let (id, rx) = watcher.subscribe(); + let (id, rx) = watcher.subscribe().map_err(database_error)?; let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(WAKE_CHANNEL_CAPACITY); let channel = channel.to_string(); let _bridge = tokio::task::spawn_blocking(move || { diff --git a/alkstore-sqlite/src/store.rs b/alkstore-sqlite/src/store.rs index 44105a1..9b5c9f9 100644 --- a/alkstore-sqlite/src/store.rs +++ b/alkstore-sqlite/src/store.rs @@ -13,12 +13,13 @@ //! bootstrap. //! //! Watcher spawn is fallible in the substrate (ADR-012 §4's W-2); the -//! failure maps into [`Error::Database`](alkstore::Error::Database) -//! engine-side and `open` fails if the watcher cannot start. On -//! watcher death the substrate's `WatcherDeathGuard` closes every -//! subscriber (ADR-006) — the death-guard plumbing exists from open -//! onward; the consumer-visible receiver bridging arrives with -//! `listen`. +//! failure maps into [`Error::Database`](alkstore::Error) engine-side +//! and `open` fails if the watcher cannot start. On watcher death the +//! substrate's `WatcherDeathGuard` closes every subscriber (ADR-006) +//! and new subscriptions fail closed with `Error::Database` (the +//! dead-watcher sign at the substrate's `subscribe`; review 004 +//! item 2) — a listen or stream subscribe on a dead-watcher store +//! never parks. //! //! Drop/close: the watcher's poll thread is joined (releasing its //! connection and the db file handle) and the writer slot + reader diff --git a/alkstore-sqlite/src/store/notify_tests.rs b/alkstore-sqlite/src/store/notify_tests.rs index 8a36706..5ec17ce 100644 --- a/alkstore-sqlite/src/store/notify_tests.rs +++ b/alkstore-sqlite/src/store/notify_tests.rs @@ -262,6 +262,50 @@ async fn watcher_death_closes_the_receiver_terminally() { cleanup(&dir); } +/// Acceptance: `listen` *after* a watcher death fails closed with +/// `Error::Database` — a new subscription never enrolls onto a dead +/// feed (which would park forever; review 004 item 2's +/// new-subscriber arm). Driven through the substrate's real +/// dead-man's switch (db-file replacement), unix-gated like the other +/// file-replacement drives. The pre-death subscribers' terminals stay +/// pinned above (`watcher_death_closes_the_receiver_terminally`). +#[tokio::test(flavor = "multi_thread")] +#[cfg(unix)] +async fn listen_after_watcher_death_fails_closed() { + let dir = temp_dir("death-listen"); + let path = temp_path("death-listen"); + let store = open_store(path.to_str().unwrap(), Default::default()).unwrap(); + + // Kill the watcher: replace the db file (the identity mismatch + // drives the substrate's watcher-fatal death). + let other = dir.join("store.db.replacement"); + std::fs::File::create(&other).unwrap(); + std::fs::rename(&other, &path).unwrap(); + until("death-listen", || store.watcher.is_dead()).await; + + let err = match store.listen("c").await { + Err(e) => e, + Ok(_) => panic!("listen on a dead-watcher store must fail closed, not park"), + }; + match &err { + Error::Database(source) => { + let message = source.to_string(); + assert!( + message.contains("the update watcher has died"), + "the source chain must name the watcher death: {message}" + ); + } + other => panic!("listen after watcher death is Database, got {other:?}"), + } + assert_eq!( + store.watcher.subscriber_count(), + 0, + "the failed listen enrolled no subscription" + ); + store.close(); + cleanup(&dir); +} + /// Acceptance: `try_recv` distinguishes closed from idle pre-close — /// `Ok(None)` with no wake pending (idle), `Some` when one arrived, /// `Err(Closed)` only after the source closed (the pinned arms). diff --git a/alkstore-sqlite/src/store/open_tests.rs b/alkstore-sqlite/src/store/open_tests.rs index 5a430dc..cd7663b 100644 --- a/alkstore-sqlite/src/store/open_tests.rs +++ b/alkstore-sqlite/src/store/open_tests.rs @@ -82,7 +82,7 @@ async fn open_leaves_the_watcher_running_and_fanning_out() { let path = dir.join("store.db"); let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); - let (_id, rx) = store.watcher.subscribe(); + let (_id, rx) = store.watcher.subscribe().unwrap(); let writer_conn = store.writer.acquire().unwrap(); writer_conn @@ -124,7 +124,7 @@ async fn closing_the_store_closes_watcher_subscribers() { let path = dir.join("store.db"); let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); - let (_id, rx) = store.watcher.subscribe(); + let (_id, rx) = store.watcher.subscribe().unwrap(); store.close(); assert!( rx.recv().is_err(), @@ -356,7 +356,7 @@ async fn drop_runs_the_full_teardown() { let path = dir.join("store.db"); { let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); - let (_id, rx) = store.watcher.subscribe(); + let (_id, rx) = store.watcher.subscribe().unwrap(); drop(store); assert!( rx.recv().is_err(), diff --git a/alkstore-sqlite/src/store/stream_tests.rs b/alkstore-sqlite/src/store/stream_tests.rs index f076c0e..69034d6 100644 --- a/alkstore-sqlite/src/store/stream_tests.rs +++ b/alkstore-sqlite/src/store/stream_tests.rs @@ -417,6 +417,51 @@ async fn watcher_death_closes_the_subscription_terminally() { cleanup(&dir); } +/// Acceptance: a stream `subscribe` *after* a watcher death fails +/// closed with `Error::Database` — a new subscription never enrolls +/// onto a dead feed (which would park forever; review 004 item 2's +/// new-subscriber arm). Driven through the substrate's real +/// dead-man's switch (db-file replacement), unix-gated like the other +/// file-replacement drives. The pre-death subscribers' terminals stay +/// pinned above (`watcher_death_closes_the_subscription_terminally`). +#[tokio::test(flavor = "multi_thread")] +#[cfg(unix)] +async fn subscribe_after_watcher_death_fails_closed() { + let dir = temp_dir("death-subscribe"); + let path = temp_path("death-subscribe"); + let store = open_store(path.to_str().unwrap(), Default::default()).unwrap(); + let stream = store.stream("s").await.unwrap(); + + // Kill the watcher: replace the db file (the identity mismatch + // drives the substrate's watcher-fatal death). + let other = dir.join("store.db.replacement"); + std::fs::File::create(&other).unwrap(); + std::fs::rename(&other, &path).unwrap(); + until("death-subscribe", || store.watcher.is_dead()).await; + + let err = match stream.subscribe("c").await { + Err(e) => e, + Ok(_) => panic!("subscribe on a dead-watcher store must fail closed, not park"), + }; + match &err { + Error::Database(source) => { + let message = source.to_string(); + assert!( + message.contains("the update watcher has died"), + "the source chain must name the watcher death: {message}" + ); + } + other => panic!("subscribe after watcher death is Database, got {other:?}"), + } + assert_eq!( + store.watcher.subscriber_count(), + 0, + "the failed subscribe enrolled no subscription" + ); + store.close(); + cleanup(&dir); +} + /// Acceptance: a subscribe attached while the stream is at tail idles /// (`Ok(None)`), then delivers when the next publish lands. #[tokio::test(flavor = "multi_thread")] diff --git a/alkstore-sqlite/src/stream.rs b/alkstore-sqlite/src/stream.rs index a9a83ea..0a94c83 100644 --- a/alkstore-sqlite/src/stream.rs +++ b/alkstore-sqlite/src/stream.rs @@ -27,9 +27,11 @@ //! stored offset to the current tail, then wake-driven re-reads off //! the watcher fanout (the same bridge shape as `listen`, yielding //! `Result` per `EventReceiver`). Watcher death closes -//! it terminally (`recv() -> None`); recovery is a fresh -//! `subscribe(consumer)` resuming from the saved checkpoint -//! (ADR-021 §5). +//! it terminally (`recv() -> None`); a subscribe issued after a +//! watcher death fails closed with +//! ([`Error::Database`](alkstore::Error)) (never a parked enrollment +//! onto a dead feed); recovery is a fresh `subscribe` on a reopened +//! store, resuming from the saved checkpoint (ADR-021 §5). //! //! Subscribe specifics: //! @@ -70,8 +72,8 @@ use alkstore::{ }; use crate::seam::{ - ReaderLoopExit as LoopExit, blocking_read, closed_store_error, sqlite_error, with_reader, - with_writer, + ReaderLoopExit as LoopExit, blocking_read, closed_store_error, database_error, sqlite_error, + with_reader, with_writer, }; use crate::substrate::{ Readers, SharedUpdateWatcher, Writer, stream_get_offset, stream_publish, stream_read_since, @@ -347,7 +349,12 @@ impl StreamHandle for SqliteStreamHandle { /// The `subscribe` bridge construction: the watcher subscription is /// taken before any read, the bridge thread then replays and parks on /// the wake feed. A thread-spawn failure unsubscribes and fails with -/// `Database` (no panics cross the seam, the W-2 posture). +/// `Database` (no panics cross the seam, the W-2 posture). A watcher +/// that has *died* (db-file replacement) fails the subscription +/// closed with `Database` too — a new subscription never enrolls onto +/// a dead feed to park forever (the death-guard closes *pre-death* +/// subscribers; this closes the *new-subscriber* arm, review 004 +/// item 2). fn spawn_bridge( name: String, consumer: String, @@ -356,7 +363,7 @@ fn spawn_bridge( watcher: Arc, store_closed: Arc, ) -> Result { - let (id, wake_rx) = watcher.subscribe(); + let (id, wake_rx) = watcher.subscribe().map_err(database_error)?; let wake_rx = Arc::new(Mutex::new(wake_rx)); let (event_tx, event_rx) = tokio::sync::mpsc::channel::>(EVENT_CHANNEL_CAPACITY); diff --git a/alkstore-sqlite/src/substrate/PROVENANCE.md b/alkstore-sqlite/src/substrate/PROVENANCE.md index 1d641e0..bf2b75c 100644 --- a/alkstore-sqlite/src/substrate/PROVENANCE.md +++ b/alkstore-sqlite/src/substrate/PROVENANCE.md @@ -131,6 +131,7 @@ Cherry-picks: empty at scaffold, append-only forever. | D-36 | drop | `UpdateWatcher::spawn` and `SharedUpdateWatcher::new` cut (the default-cadence convenience constructors): upstream's embedding harnesses opened watchers without explicit config; this engine's single open path always constructs a `WatcherConfig` — from `SqliteOpts` (ADR-023 §4's `poll_interval` knob, `None` = the 1 ms default) — and passes it through `*_with_config`; a config-less twin would re-create two open postures (the mixed-flag inconsistency D-29 resolved at the URI layer). `WatcherConfig::default()` remains the default's single owner; the tests driving the cut constructors now build the config explicitly, pinning the same behavior | `watcher.rs` (both constructors + five watcher-test call sites) | ADR-023 §4 (the cadence knob rides the engine's opts — one open posture carries config); ADR-012 §2 (one construction path the engine reviews); wave-3 lint-removal obligation — cut, not suppressed | ours | | D-37 | re-derivation | visibility-only widening on the re-derived `@every` surface: `parse_every_interval` (`pub(crate)` → `pub`) and `skip_forward` (module-private → `pub`) in `queue_ops.rs`, plus this subtree's facade re-export (`mod.rs`: `pub use queue_ops::{parse_every_interval, skip_forward}`), so the engine crate root can `#[doc(hidden)] pub use` both as the wave-7 fuzzing differential target's drive surface (`fuzz-parser-exposure`, target `every_parse`). Zero logic change, zero signature change, zero name change; `mod substrate` stays private, so containment holds — nothing leaves the crate except the doc-hidden root re-exports. A `pub use` re-export mechanically requires item-level `pub` (E0364/E0603), which is why the facade line is in `mod.rs`; the module doc's "nothing re-exported" boundary sentence was amended in the same commit to name this sole exception | `queue_ops.rs` (two visibility keywords); `mod.rs` (facade re-export line + boundary-doc exception sentence) | ADR-012 §3 (fidelity governs the ported honker body — `watcher.rs`/`schema.rs`/`ops.rs` untouched; `queue_ops.rs` is the re-derived half, owned contract-v1 code, so a visibility-only delta is legitimate); docs/research/fuzzing.md §3.2 (option (b): the `#[doc(hidden)]` narrow exposure); ADR-017 §3 (doc-hidden test surface, no stability guarantee — the `Job::from_row` precedent) | ours | | D-38 | port delta | degenerate/adversarial-triple hardening on `skip_forward` (two legs, both aligning to the pg engine's committed clamp shape where it existed and extending pg itself): (1) the `interval` is clamped (`interval.max(1) as i128`) — pre-fix, `interval == 0` panicked on i128 division-by-zero, reachable only through degenerate/non-grammar callers (the parsers reject `n <= 0`, so every engine-internal call site passes `interval >= 1`); (2) the gap is computed in i128 (`now as i128 - next_fire_at as i128`) — pre-fix, `now - next_fire_at` was an **i64 subtraction** that panics in debug (overflow) / silently wraps in release when `(next_fire_at, now)` straddle the full domain (`now=i64::MIN`-adjacent, `next_fire_at=i64::MAX`-adjacent), feeding a wrapped gap to the 128-bit arithmetic and breaking the saturation guarantee. Found by the `every_parse` differential target (wave 7, tasks/fuzz-target-every-parse) driving the adversarial `(next_fire_at, now, interval)` space; pg needed leg (2) as well. Observable behavior unaffected in the reachable domain (|gap| < i64 range), saturation posture now holds over the whole domain. Zero signature change; lineage body untouched (`skip_forward` is the re-derived scheduler half, D-12's own) | `queue_ops.rs` (`skip_forward`, two lines) | docs/research/fuzzing.md §3.2 (the overflow-posture invariant: reject-or-saturate, never wrap or panic, over the whole domain); wave-6's register-overflow alignment (tasks/review-wave-6); alkstore-postgres/src/scheduler.rs `skip_forward` (the pinned clamp shape) | ours | +| D-39 | port delta | watcher-death state machine (review 004 items 2+3, two legs on the ported watcher): (1) **identity-unknown baseline** — the loop's initial identity `stat` failure seeds an unknown baseline instead of the false `(0, 0)`; the first *successful* stat captures it without driving the death path (pre-fix, a transient startup stat hiccup killed the watcher — and closed every subscriber — when the first later stat succeeded); (2) **dead-watcher fail-closed subscribe** — the loop sets a `dead` sign on the caller (`SharedUpdateWatcher`, an `Arc`) before exiting through the watcher-fatal path, and `subscribe` becomes fallible, failing closed on the sign (or a taken watcher slot) checked atomically with the enrollment under the senders lock — new subscriptions after a death fail instead of enrolling onto a feed with no poller alive to park forever (the death-guard's clear of *pre-death* subscribers unchanged, D-07 still holds); the engine maps the failure into `Error::Database` at `listen`/stream `subscribe` (ADR-008 §5's opaque fallback). Lineage behavior for both legs was the (0,0)-seed and the infallible enroll | `watcher.rs` (`run_poll_loop`'s baseline/identity arms, `SharedUpdateWatcher`'s dead sign + fallible `subscribe`); `notify.rs`/`stream.rs` (the engine-side mapping) | engine-sqlite.md's watcher-death posture ("new subscriptions fail closed"; identity unknown until the first successful stat); review 004 items 2–3 (tasks/sqlite-fix-watcher-death-subscribe); ADR-006 (the close/fail-closed posture); ADR-012 §4 (delta discipline; D-07's death path refined) | ours | ### Cherry-picks diff --git a/alkstore-sqlite/src/substrate/watcher.rs b/alkstore-sqlite/src/substrate/watcher.rs index 4b1cd3e..db4ceb8 100644 --- a/alkstore-sqlite/src/substrate/watcher.rs +++ b/alkstore-sqlite/src/substrate/watcher.rs @@ -135,15 +135,23 @@ fn open_watcher_conn(path: &Path) -> rusqlite::Result { /// vanished db file produces bounded open attempts (W-1), not the /// lineage's one attempt per millisecond. /// 3. **Identity check (about every 100 ms):** `stat(db_path)` to -/// compare `(dev, ino)`. If the file was replaced, log the precise -/// diagnostic and exit through the ordinary death path (the fork's -/// dead-man's-switch delta — the lineage panicked here); continuing -/// would silently watch stale data. `WatcherDeathGuard` closes -/// every subscriber exactly as it does for any watcher death. +/// compare `(dev, ino)` against the baseline. If the file was +/// replaced, log the precise diagnostic and exit through the +/// ordinary death path (the fork's dead-man's-switch delta — the +/// lineage panicked here); continuing would silently watch stale +/// data. `WatcherDeathGuard` closes every subscriber exactly as it +/// does for any watcher death, and the caller's `dead` flag is set +/// before the exit — new subscriptions on the dead store fail +/// closed, they never park on a dead feed. The baseline starts +/// unknown when the initial stat fails (the first *successful* +/// stat captures it, without driving death — a failed startup stat +/// must not seed a false identity that a later transient recovery +/// would trip). pub(crate) fn run_poll_loop( db_path: PathBuf, on_change: F, stop: Arc, + dead: Arc, ready: std::sync::mpsc::SyncSender<()>, poll_interval: Duration, open_conn_fn: ReconnectOpenFn, @@ -161,11 +169,16 @@ pub(crate) fn run_poll_loop( .as_ref() .and_then(|c| poll_data_version(c).ok()) .unwrap_or(0); - let initial_identity = match stat_identity(&db_path) { - Ok(id) => id, + // The file identity baseline, seeded by the loop's initial stat: + // `None` while unknown — a failed startup stat must not seed the + // `(0, 0)` false identity that the first successful later stat + // would trip into the death path; the first successful stat + // captures the baseline instead, without driving death. + let mut baseline = match stat_identity(&db_path) { + Ok(id) => Some(id), Err(e) => { eprintln!("alkstore: failed to stat database for identity check: {e}"); - (0, 0) + None } }; let mut next_identity_check = Instant::now() + UPDATE_WATCHER_IDENTITY_INTERVAL; @@ -211,19 +224,23 @@ pub(crate) fn run_poll_loop( if now >= next_identity_check { next_identity_check = now + UPDATE_WATCHER_IDENTITY_INTERVAL; match stat_identity(&db_path) { - Ok(current) => { - if current != initial_identity { - eprintln!( - "alkstore: database file replaced: \ - expected (dev={}, ino={}), \ - found (dev={}, ino={}) at {:?}. \ - The watcher cannot recover; it has stopped and \ - closed all subscribers; close the store and reopen.", - initial_identity.0, initial_identity.1, current.0, current.1, db_path - ); - return; + Ok(current) => match baseline { + Some(expected) => { + if current != expected { + eprintln!( + "alkstore: database file replaced: \ + expected (dev={}, ino={}), \ + found (dev={}, ino={}) at {:?}. \ + The watcher cannot recover; it has stopped and \ + closed all subscribers; close the store and reopen.", + expected.0, expected.1, current.0, current.1, db_path + ); + dead.store(true, Ordering::Release); + return; + } } - } + None => baseline = Some(current), + }, Err(e) => { eprintln!("alkstore: stat identity check failed: {e}"); conn = None; @@ -256,6 +273,7 @@ impl UpdateWatcher { db_path: PathBuf, on_change: F, config: WatcherConfig, + dead: Arc, ) -> Result where F: Fn() + Send + 'static, @@ -269,6 +287,7 @@ impl UpdateWatcher { stop_t, ready_tx, config.poll_interval, + dead, None, )?; let _ = ready_rx.recv(); @@ -288,6 +307,7 @@ impl UpdateWatcher { stop_t: Arc, ready_tx: std::sync::mpsc::SyncSender<()>, poll_interval: Duration, + dead: Arc, stack_size: Option, ) -> Result, String> where @@ -304,6 +324,7 @@ impl UpdateWatcher { db_path, on_change, stop_t, + dead, ready_tx, poll_interval, open_watcher_conn, @@ -345,6 +366,12 @@ impl Drop for WatcherDeathGuard { pub(crate) struct SharedUpdateWatcher { watcher: Mutex>, senders: Arc>>>, + /// The death sign the poll loop sets before exiting through the + /// watcher-fatal path (db-file replacement). New subscriptions + /// fail closed on it — they must never enroll onto a feed with no + /// poller alive (the death-guard clears *existing* subscribers; + /// this closes the *new-subscriber* arm). + dead: Arc, next_id: AtomicU64, } @@ -358,6 +385,7 @@ impl SharedUpdateWatcher { let senders: Arc>>> = Arc::new(Mutex::new(HashMap::new())); let senders_t = senders.clone(); + let dead: Arc = Arc::new(AtomicBool::new(false)); let death_guard = WatcherDeathGuard { senders: senders.clone(), }; @@ -372,10 +400,12 @@ impl SharedUpdateWatcher { }); }, config, + dead.clone(), )?; Ok(Self { watcher: Mutex::new(Some(watcher)), senders, + dead, next_id: AtomicU64::new(0), }) } @@ -388,11 +418,32 @@ impl SharedUpdateWatcher { /// /// Channel capacity is 1: bursts coalesce into one wake per drain /// cycle. Wakes are "go re-read state" signals. - pub(crate) fn subscribe(&self) -> (u64, std::sync::mpsc::Receiver<()>) { + /// + /// Fallible, failing closed: on a dead watcher (the poll thread + /// exited through the watcher-fatal path — the store's db file was + /// replaced; close the store and reopen) or a closed watcher (the + /// watcher slot was taken by [`close`](Self::close)). Both checks + /// hold the senders lock together with the enrollment, so the + /// sign and the enroll are one step — the death-guard's clear + /// races the check as a pre-death subscriber (cleared → the + /// receiver sees the disconnect), never as a parked ghost. The + /// engine layer maps the failure into the contract taxonomy. + pub(crate) fn subscribe(&self) -> Result<(u64, std::sync::mpsc::Receiver<()>), String> { + let mut senders = self.senders.lock(); + if self.dead.load(Ordering::Acquire) { + return Err( + "the update watcher has died: the database file was replaced; \ + new subscriptions fail closed — close the store and reopen" + .to_string(), + ); + } + if self.watcher.lock().is_none() { + return Err("the update watcher is closed".to_string()); + } let id = self.next_id.fetch_add(1, Ordering::Relaxed); let (tx, rx) = std::sync::mpsc::sync_channel(1); - self.senders.lock().insert(id, tx); - (id, rx) + senders.insert(id, tx); + Ok((id, rx)) } /// Remove a subscriber. The corresponding receiver sees @@ -407,6 +458,11 @@ impl SharedUpdateWatcher { self.senders.lock().len() } + #[cfg(test)] + pub(crate) fn is_dead(&self) -> bool { + self.dead.load(Ordering::Acquire) + } + /// Disconnect all subscribers and synchronously join the poll /// thread. The thread owns the watcher's connection; joining drops /// that connection and releases the file handle. Idempotent — safe @@ -464,6 +520,7 @@ mod watcher_tests { Arc::new(AtomicBool::new(false)), tx, Duration::from_millis(1), + Arc::new(AtomicBool::new(false)), Some(usize::MAX), ) .expect_err("an oversized stack reservation must fail to spawn"); @@ -494,7 +551,7 @@ mod watcher_tests { let shared = SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap(); let subs: Vec<(u64, std::sync::mpsc::Receiver<()>)> = - (0..50).map(|_| shared.subscribe()).collect(); + (0..50).map(|_| shared.subscribe().unwrap()).collect(); let writer = Connection::open(&tmp).unwrap(); writer @@ -532,7 +589,7 @@ mod watcher_tests { let shared = SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap(); - let (id, rx) = shared.subscribe(); + let (id, rx) = shared.subscribe().unwrap(); assert_eq!(shared.subscriber_count(), 1); shared.unsubscribe(id); @@ -564,7 +621,7 @@ mod watcher_tests { let shared = SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap(); - let (_id, rx) = shared.subscribe(); + let (_id, rx) = shared.subscribe().unwrap(); // Let the watcher snapshot the initial identity. std::thread::sleep(Duration::from_millis(200)); @@ -586,6 +643,10 @@ mod watcher_tests { WatcherDeathGuard didn't fire?" ); } + assert!( + shared.is_dead(), + "the death sign must be set once the poll thread died" + ); shared.close().unwrap(); let _ = std::fs::remove_file(&tmp); @@ -603,8 +664,13 @@ mod watcher_tests { let tmp = temp_db("watcher-replace"); wal_db(&tmp); - let watcher = - UpdateWatcher::spawn_with_config(tmp.clone(), || {}, WatcherConfig::default()).unwrap(); + let watcher = UpdateWatcher::spawn_with_config( + tmp.clone(), + || {}, + WatcherConfig::default(), + Arc::new(AtomicBool::new(false)), + ) + .unwrap(); std::thread::sleep(Duration::from_millis(200)); @@ -634,7 +700,7 @@ mod watcher_tests { let shared = SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap(); { - let _subs: Vec<_> = (0..10).map(|_| shared.subscribe()).collect(); + let _subs: Vec<_> = (0..10).map(|_| shared.subscribe().unwrap()).collect(); assert_eq!(shared.subscriber_count(), 10); } @@ -858,6 +924,8 @@ mod watcher_tests { let wakes_t = wakes.clone(); let stop = Arc::new(AtomicBool::new(false)); let stop_t = stop.clone(); + let dead = Arc::new(AtomicBool::new(false)); + let dead_t = dead.clone(); let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<()>(1); let db_path = tmp.clone(); let handle = std::thread::spawn(move || { @@ -867,6 +935,7 @@ mod watcher_tests { wakes_t.fetch_add(1, Ordering::Relaxed); }, stop_t, + dead_t, ready_tx, Duration::from_millis(5), instrumented_open, @@ -930,12 +999,15 @@ mod watcher_tests { )); let stop = Arc::new(AtomicBool::new(false)); let stop_t = stop.clone(); + let dead = Arc::new(AtomicBool::new(false)); + let dead_t = dead.clone(); let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<()>(1); let handle = std::thread::spawn(move || { run_poll_loop( db_path, || {}, stop_t, + dead_t, ready_tx, Duration::from_millis(tick_ms), counting_open, @@ -968,4 +1040,163 @@ mod watcher_tests { window at {tick_ms}ms/tick; the backoff admits at most {bound}" ); } + + /// Review 004 item 2 (substrate arm): after the poll thread exits + /// through the watcher-fatal path (file replacement), `subscribe` + /// fails closed — no new sender enrolls onto a feed with no poller + /// alive (which would park its receiver forever). Driven through + /// the same seam as the death-signal test: a pre-death subscriber + /// observes the death first, then the dead store rejects a fresh + /// subscription. + #[test] + #[cfg(unix)] + fn subscribe_fails_closed_after_watcher_death() { + let tmp = temp_db("death-subscribe"); + wal_db(&tmp); + + let shared = + SharedUpdateWatcher::new_with_config(tmp.clone(), WatcherConfig::default()).unwrap(); + let (_id, rx) = shared.subscribe().unwrap(); + + std::thread::sleep(Duration::from_millis(200)); + + let other = temp_db("death-subscribe-other"); + let _ = std::fs::File::create(&other).unwrap(); + std::fs::rename(&other, &tmp).unwrap(); + + let deadline = std::time::Instant::now() + Duration::from_secs(2); + loop { + match rx.recv_timeout(Duration::from_millis(100)) { + Ok(()) => continue, + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} + } + assert!( + std::time::Instant::now() < deadline, + "watcher died but subscriber's channel never disconnected" + ); + } + assert!( + shared.is_dead(), + "the death sign must be set once the poll thread died" + ); + + let err = shared + .subscribe() + .expect_err("a subscription on a dead-watcher store must fail closed"); + assert!( + err.contains("the update watcher has died"), + "the failure must name the dead watcher: {err}" + ); + assert_eq!( + shared.subscriber_count(), + 0, + "a rejected subscribe must enroll nothing" + ); + + shared.close().unwrap(); + let _ = std::fs::remove_file(&tmp); + let _ = std::fs::remove_file(format!("{}-wal", tmp.display())); + let _ = std::fs::remove_file(format!("{}-shm", tmp.display())); + } + + /// Review 004 item 3: a failed identity stat at watcher start must + /// leave the baseline *unknown*, not seeded with the `(0, 0)` + /// false identity — the first successful later stat captures the + /// baseline and the watcher survives, and only a later true + /// mismatch drives the death path. Driven through `run_poll_loop` + /// with the db file absent at start (the initial stat genuinely + /// fails on unix; on non-unix `stat_identity` never fails, so the + /// scenario cannot arise there and this test is unix-only), the + /// file then created and committed to (alive: the wake lands — + /// with the `(0, 0)` seeding the watcher would have died on the + /// first successful stat and no wake could arrive), and replaced + /// afterwards (the death path still fires on a real mismatch). + #[test] + #[cfg(unix)] + fn initial_stat_failure_does_not_seed_a_false_identity() { + let tmp = temp_db("identity-unknown"); + + let wakes = Arc::new(AtomicU64::new(0)); + let wakes_t = wakes.clone(); + let stop = Arc::new(AtomicBool::new(false)); + let stop_t = stop.clone(); + let dead = Arc::new(AtomicBool::new(false)); + let dead_t = dead.clone(); + let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<()>(1); + let db_path = tmp.clone(); + let handle = std::thread::spawn(move || { + run_poll_loop( + db_path, + move || { + wakes_t.fetch_add(1, Ordering::Relaxed); + }, + stop_t, + dead_t, + ready_tx, + Duration::from_millis(5), + open_watcher_conn, + ) + }); + ready_rx + .recv_timeout(Duration::from_secs(2)) + .expect("watcher loop baseline capture"); + + // The file is absent: several identity checks fail; the + // baseline must stay unknown (never seeded with (0, 0)). + std::thread::sleep(Duration::from_millis(300)); + assert!( + !dead.load(Ordering::Acquire), + "identity-unknown must not drive the death path" + ); + + // First successful stat: the baseline is captured. The watcher + // survives it — a later commit's wake lands. (Wakes counted + // *before* the snapshot include the transient on_change calls + // the failing identity stats fired; the proof is relative to + // the post-recovery snapshot.) + wal_db(&tmp); + let writer = Connection::open(&tmp).unwrap(); + writer + .execute("CREATE TABLE IF NOT EXISTS _test_ident(x INTEGER)", []) + .unwrap(); + + std::thread::sleep(Duration::from_millis(400)); + assert!( + !dead.load(Ordering::Acquire), + "the first successful stat must not trigger the death path" + ); + let before = wakes.load(Ordering::Relaxed); + writer + .execute("INSERT INTO _test_ident(x) VALUES (1)", []) + .unwrap(); + let deadline = std::time::Instant::now() + Duration::from_secs(3); + while wakes.load(Ordering::Relaxed) == before && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(5)); + } + assert!( + wakes.load(Ordering::Relaxed) > before, + "the watcher must still be delivering wakes after the \ + identity-unknown recovery" + ); + + // A real mismatch still drives the death path. + let tmp2 = temp_db("identity-replacement"); + let _ = std::fs::File::create(&tmp2).unwrap(); + std::fs::rename(&tmp2, &tmp).unwrap(); + let deadline = std::time::Instant::now() + Duration::from_secs(3); + while !dead.load(Ordering::Acquire) && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + } + assert!( + dead.load(Ordering::Acquire), + "a genuine identity mismatch must still drive the death path" + ); + + stop.store(true, Ordering::Release); + handle.join().unwrap(); + let _ = std::fs::remove_file(&tmp); + let _ = std::fs::remove_file(format!("{}-wal", tmp.display())); + let _ = std::fs::remove_file(format!("{}-shm", tmp.display())); + } } diff --git a/docs/architecture/engine-sqlite.md b/docs/architecture/engine-sqlite.md index c981dc9..2a50d89 100644 --- a/docs/architecture/engine-sqlite.md +++ b/docs/architecture/engine-sqlite.md @@ -97,7 +97,16 @@ per-contract obligations live in the core spec and are not restated here. - Watcher failure handling is inherited: on watcher death, every subscriber's receiver closes (`WatcherDeathGuard` behavior) — consumers see the close, never a silent hang - ([ADR-006](decisions/006-wake-and-delivery-contract.md)). + ([ADR-006](decisions/006-wake-and-delivery-contract.md)). New + subscriptions issued *after* a watcher death (a `listen()` or a + stream `subscribe()`) fail closed with the opaque `Error::Database` + fallback — they never enroll onto a dead feed to park forever + (review 004 item 2); recovery is a fresh store open. The watcher + also treats the db file's identity as unknown until its first + successful stat: a failed startup stat does not seed a false + identity that the first later stat would spuriously trip into the + death path (review 004 item 3 — a transient stat hiccup at start + closes nothing). - Wake coalescing: bursts inside one poll tick produce one wake; correctness is preserved by the re-read contract (POC-pinned: missed-wake stress with correct post-burst re-reads). diff --git a/tasks/sqlite-fix-watcher-death-subscribe.md b/tasks/sqlite-fix-watcher-death-subscribe.md index cfeb265..1ce8563 100644 --- a/tasks/sqlite-fix-watcher-death-subscribe.md +++ b/tasks/sqlite-fix-watcher-death-subscribe.md @@ -1,7 +1,7 @@ --- id: sqlite-fix-watcher-death-subscribe name: Dead-watcher subscribe hang + initial-stat self-kill (review 004 items 2+3) -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -46,17 +46,17 @@ subscribers stay open. ## Acceptance Criteria -- [ ] Subscribing (listen or stream) after watcher death fails closed +- [x] Subscribing (listen or stream) after watcher death fails closed with a typed error instead of parking — pinned by test -- [ ] Pre-death subscribers still close at watcher death — the +- [x] Pre-death subscribers still close at watcher death — the existing pins green unchanged -- [ ] An initial identity-stat failure no longer seeds a false +- [x] An initial identity-stat failure no longer seeds a false identity: the first successful stat does not trigger the death path and subscribers stay open — pinned by injected-failure test -- [ ] engine-sqlite.md's watcher-death posture text covers the +- [x] engine-sqlite.md's watcher-death posture text covers the new-subscriber arm (and the identity-unknown state if it changes observable behavior) -- [ ] `cargo test -p alkstore-sqlite`, clippy `-D warnings`, fmt +- [x] `cargo test -p alkstore-sqlite`, clippy `-D warnings`, fmt clean; workspace gates green server-less ## References @@ -71,6 +71,79 @@ subscribers stay open. > To be filled by implementation agent +Decisions of record the description didn't pin: + +- The dead sign lives on `SharedUpdateWatcher` as an + `Arc`, set by `run_poll_loop` itself just before the + watcher-fatal return — it is set *before* the death-guard's senders + clear (which happens after the loop returns), so the guard-side + ordering is fixed by program order. +- `SharedUpdateWatcher::subscribe` became fallible + (`Result<_, String>`) and checks **two** closed arms atomically with + the enrollment under the senders lock: the dead sign *and* the + taken watcher slot (the post-`close` enroll-onto-empty-watcher race + of the same shape). Failing closed on the taken slot is new + belt-and-braces beyond the finding's shape; the store's `closed` + flag already guarded the normal path. Rationale for checking under + the lock: a check-then-enroll done outside the lock could interleave + between the death and the guard's clear and enroll a ghost. +- The substrate stays contract-blind: the failure is a `String` + naming the death, mapped into the taxonomy engine-side — + `Error::Database` (the opaque fallback with the source chain), the + same shape the engine's closed-store checks use. `Error::Closed` + was rejected: the contract docs scope that variant to *receiver* + closed arms (`try_recv`/`recv_timeout`), not entry points. +- The initial-identity fix is an in-place re-derivation of the same + code (`Option<(u64, u64)>` baseline vs. the `(0, 0)` seed); the + death-log message keeps its precise diagnostic (now printing the + captured baseline). +- The injected-failure test for (b) is the substrate-level drive + (`run_poll_loop` with the db file absent at start — the initial + stat genuinely fails; on non-unix `stat_identity` never fails, so + the scenario cannot arise and the test is unix-gated like the other + file-replacement drives); no `cfg(test)` injection seam was added to + `stat_identity` itself. Aliveness after the first successful stat is + proven by wake delivery relative to a pre-insert snapshot (failing + identity stats fire `on_change`, so absolute wake counts are not a + clean liveness signal). +- Also updated: engine-sqlite.md's watcher-failure posture bullet + (both arms), notify.rs/stream.rs/store.rs module docs, open_tests' + three direct `watcher.subscribe()` call sites (now `.unwrap()`d), + and one PROVENANCE.md row (D-39 — the register wants divergence to + the ported watcher recorded even though the task text didn't name it). + ## Summary -> To be filled on completion \ No newline at end of file +> To be filled on completion + +Both defects landed with tests: + +- (a) Post-watcher-death subscribe hang: the poll loop sets the dead + sign before exiting through the identity-replacement death path; + `SharedUpdateWatcher::subscribe` fails closed on it (and on a taken + watcher slot) atomically with enrollment under the senders lock; + `notify.rs::listen` and `stream.rs::spawn_bridge` map the failure to + `Error::Database` with the death named in the source chain. Pinned + by four tests: substrate-level + (`subscribe_fails_closed_after_watcher_death`, unix file-replacement + drive) and engine-level (`listen_after_watcher_death_fails_closed`, + `subscribe_after_watcher_death_fails_closed` — same drive through + the store surface, error-shape + no-enrollment assertions). The + pre-death subscriber close-at-death pins + (`watcher_death_closes_the_receiver_terminally`, + `watcher_death_closes_the_subscription_terminally`, and the + substrate death-signal tests) are green unchanged — the death-signal + substrate test also now pins the dead sign being set. +- (b) Initial-stat self-kill: the loop tracks the identity baseline as + `Option<(u64, u64)>`; a failed startup stat leaves it unknown, the + first successful stat captures it without driving death, and a later + genuine mismatch still kills. Pinned by + `initial_stat_failure_does_not_seed_a_false_identity` + (`run_poll_loop` drive: absent file → unknown baseline → file + created → no death + wake delivery lands → replacement → death + fires; unix-gated). +- Verified: `cargo build`, `cargo test` (workspace, all green), + `cargo test -p alkstore-sqlite` (201 tests), fuzz gate + (`cargo test --manifest-path fuzz/shared/Cargo.toml`), clippy + `-D warnings` clean, `cargo fmt --check` clean, rustdoc + `-D warnings` doc gate clean. \ No newline at end of file