diff --git a/alkstore-postgres/src/forwarder.rs b/alkstore-postgres/src/forwarder.rs index 7e1f8d2..9a88c87 100644 --- a/alkstore-postgres/src/forwarder.rs +++ b/alkstore-postgres/src/forwarder.rs @@ -218,6 +218,12 @@ pub(crate) struct Forwarder { /// `#[cfg(test)]` swap accessors). #[cfg(test)] reconnect_config: ReconnectConfigSlot, + /// The test-only generation-top LISTEN pause (review 003 Finding + /// 2a's seam) — the loop's own [`LoopCtx`] clone is the live read + /// path; the field exists for the test arm accessor below. Dead + /// in production builds. + #[cfg(test)] + listen_pause: crate::seam::listen_pause::Flag, shutdown: watch::Sender, } @@ -360,6 +366,8 @@ impl Forwarder { let generation: GenerationSlot = Arc::new(AtomicU64::new(0)); let reconnect_config_slot: ReconnectConfigSlot = Arc::new(std::sync::Mutex::new(reconnect_config)); + #[cfg(test)] + let listen_pause = crate::seam::listen_pause::disarmed(); // The connection handed in was just established — the liveness // flag starts true and the loop clears it at each generation's // death (the register path's honest transient state). @@ -372,6 +380,8 @@ impl Forwarder { client, connection, reconnect_config: reconnect_config_slot.clone(), + #[cfg(test)] + listen_pause: listen_pause.clone(), fanout: fanout.clone(), fanout_slot: fanout_slot.clone(), channels: channels.clone(), @@ -389,6 +399,8 @@ impl Forwarder { connected, #[cfg(test)] reconnect_config: reconnect_config_slot, + #[cfg(test)] + listen_pause, shutdown: shutdown_tx, } } @@ -505,6 +517,17 @@ impl Forwarder { .expect("reconnect config mutex poisoned") .clone() } + + /// Arm the test-only generation-top LISTEN pause (one-shot) — the + /// connection-death-during-LISTEN seam (review 003 Finding 2a): + /// the loop sleeps a short window at the next generation's top, + /// *before* issuing its snapshot LISTEN, so a test can kill the + /// fresh generation's backend inside the window and make the + /// LISTEN fail honestly (the real fall-through arm runs). + #[cfg(test)] + pub(crate) fn arm_generation_listen_pause(&self) { + crate::seam::listen_pause::arm(&self.listen_pause) + } } /// The reconnect arm's retry core (review 002 Finding 1's fix): sleep @@ -600,6 +623,10 @@ struct LoopCtx { client: tokio_postgres::Client, connection: ListenerConnection, reconnect_config: ReconnectConfigSlot, + /// The test-only generation-top LISTEN pause (review 003 Finding + /// 2a's seam); dead in production builds. + #[cfg(test)] + listen_pause: crate::seam::listen_pause::Flag, fanout: broadcast::Sender, fanout_slot: FanoutSlot, channels: Arc, @@ -614,6 +641,8 @@ async fn forwarder_loop( client, connection, reconnect_config, + #[cfg(test)] + listen_pause, fanout, fanout_slot, channels, @@ -639,6 +668,22 @@ async fn forwarder_loop( // through the Client. let mut poll_task = tokio::spawn(poll_loop(connection, fanout.clone())); + // The test-only LISTEN-window seam (review 003 Finding 2a): + // one-shot armed, this generation's top pauses here for the + // window a backend-kill test can land in — the LISTEN below + // then runs against a dead session and fails honestly, the + // exact "connection died before/during the LISTEN" shape the + // fall-through arm exists for. Production builds compile a + // plain `false`. + #[cfg(test)] + let pre_listen_pause = crate::seam::listen_pause::take(&listen_pause); + #[cfg(not(test))] + let pre_listen_pause = false; + if pre_listen_pause { + #[cfg(test)] + tokio::time::sleep(crate::seam::listen_pause::WINDOW).await; + } + // Issue LISTENs from the dynamic channel set's snapshot (on // first connect it carries pre-registered channels; on // reconnect it carries everything `listen` ever registered). diff --git a/alkstore-postgres/src/seam.rs b/alkstore-postgres/src/seam.rs index e954205..865be62 100644 --- a/alkstore-postgres/src/seam.rs +++ b/alkstore-postgres/src/seam.rs @@ -42,6 +42,44 @@ pub(crate) fn database_error(message: impl Into) -> Error { Error::database(std::io::Error::other(message.into())) } +/// The `cfg(test)` generation-top LISTEN pause seam (review 003 +/// Finding 2a): a one-shot arm that makes the forwarder loop sleep a +/// short window at the next connection generation's top, **before** +/// issuing its snapshot LISTEN — the "connection died before/during +/// the LISTEN" window (the fall-through to the reconnect path at +/// `forwarder.rs`'s `listen_ok == false` arm). A test arms it, forces +/// the reconnect (a backend kill), then kills the fresh generation's +/// backend inside the window: the fresh session's LISTEN batch +/// executes against a dead connection and fails **honestly** — the +/// error is real driver-against-dead-session, only the timing is +/// fabricated — and the production fall-through arm runs unmodified. +/// One-store scoping (disarms on first consumption, the commit-fault +/// seam's shape) keeps parallel test stores from interfering. +#[cfg(test)] +pub(crate) mod listen_pause { + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::time::Duration; + + pub(crate) type Flag = Arc; + + pub(crate) fn disarmed() -> Flag { + Arc::new(AtomicBool::new(false)) + } + + pub(crate) fn arm(flag: &Flag) { + flag.store(true, Ordering::Release); + } + + pub(crate) fn take(flag: &Flag) -> bool { + flag.swap(false, Ordering::AcqRel) + } + + /// The pause window: wide enough for a kill probe to land inside + /// it, narrow enough that the test's waits stay bounded. + pub(crate) const WINDOW: Duration = Duration::from_millis(500); +} + /// The `cfg(test)` tick-fault seam (`pg-fix-scheduler-tick-seam`, /// review 003 Finding 1): a per-store arm that, when set, substitutes /// a fabricated pool/database-shaped error for one `tick_once` diff --git a/alkstore-postgres/src/store/open_tests.rs b/alkstore-postgres/src/store/open_tests.rs index bb52739..f89c535 100644 --- a/alkstore-postgres/src/store/open_tests.rs +++ b/alkstore-postgres/src/store/open_tests.rs @@ -108,6 +108,22 @@ async fn listener_sessions(admin: &tokio_postgres::Client, app_name: &str) -> i6 .get(0) } +/// Kill every listener session of `app_name` — the backend-killing +/// probe the reconnect tests ride (`false` when no listener backend +/// is currently live, e.g. while the forwarder is mid-reconnect). +async fn kill_listener(admin: &tokio_postgres::Client, app_name: &str) -> bool { + admin + .query_one( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity + WHERE application_name = $1 AND pid <> pg_backend_pid()", + &[&app_name], + ) + .await + .ok() + .and_then(|row| row.try_get::<_, bool>(0).ok()) + .unwrap_or(false) +} + /// The forwarder's fanout subscription surface for the minimal-delivery /// proof (bridging into `WakeReceiver`s is the notify-listen task's; /// the skeleton's proof fans the raw notification channel names out). @@ -314,6 +330,183 @@ async fn forwarder_reconnects_after_backend_kill_and_delivers() { drop_schema(&admin, &schema).await; } +/// Acceptance (review 003 Finding 2a): the "connection died +/// before/during the LISTEN" fall-through arm — a generation whose +/// LISTEN issue fails because the session died inside the LISTEN +/// window — falls through to the reconnect path with the honest +/// transient state (`connected` cleared: registrations during the gap +/// fail the transient `Database`, never a spurious success), reaps +/// the dead generation's poll task, reconnects, and recovers fully: +/// the reconnect-wake broadcasts, the pre-death registration's +/// NOTIFY delivers (the re-issue), and the mid-gap registration — +/// refused with the transient but registry-written-first — is +/// recovered by the re-issue too. +/// +/// The seam is the `cfg(test)` generation-top LISTEN pause +/// (`seam.rs`'s one-shot `listen_pause`): the fresh reconnect +/// generation pauses before its LISTEN issue, and the test kills that +/// fresh backend inside the window — the LISTEN batch then executes +/// against a dead session and fails *honestly* (real driver error, +/// fabricated timing only), exercising the production fall-through +/// arm unmodified. +#[tokio::test(flavor = "multi_thread")] +async fn forwarder_recovers_from_connection_death_during_the_listen_window() { + let Some(dsn) = harness_dsn() else { + eprintln!("skip: no harness server"); + return; + }; + let schema = instance_namer("listen-death")(); + let store = open_store(&dsn, test_opts(&schema)).await.unwrap(); + let forwarder = store.forwarder().clone(); + let admin = harness_client().await.unwrap(); + + let channel = format!("lstndeath_probe_{}", instance_namer("ch")()); + let mut rx = forwarder.subscribe().unwrap(); + forwarder.register(&channel).await.unwrap(); + + // Pre-death delivery verified (the normal path is not what broke). + let notify_sql = |ch: &str, payload: &str| { + format!( + "NOTIFY {}, '{payload}'", + crate::schema::quote_identifier(ch) + ) + }; + admin + .batch_execute(¬ify_sql(&channel, "pre-death")) + .await + .unwrap(); + assert!( + wait_for_channel(&mut rx, &channel, Duration::from_secs(3)).await, + "pre-death delivery must work first" + ); + + // A dedicated wake counter (subscribed before the kills): the + // fall-through arm's differentiator — see the exactly-one-wake + // pin after recovery. + let mut wake_counter = forwarder.subscribe().unwrap(); + + // Arm the seam, then kill the listener's first-generation backend: + // the reconnect's fresh generation pauses before its LISTEN + // issue — the window the second kill lands in. + forwarder.arm_generation_listen_pause(); + let app_name = store.listener_application_name().to_string(); + assert!( + kill_listener(&admin, &app_name).await, + "the first listener backend was killed" + ); + + // Kill the fresh reconnect generation's backend inside its + // pre-LISTEN pause (retry until a listener backend is found + // again — the fresh connection lives only behind the pause). + let mut fresh_killed = false; + for _ in 0..80 { + if kill_listener(&admin, &app_name).await { + fresh_killed = true; + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + assert!( + fresh_killed, + "the fresh reconnect generation's backend was killed during its \ + pre-LISTEN pause (the connection death inside the LISTEN window)" + ); + + // The fall-through's honest transient: a registration during the + // gap fails with the transient `Database` (the connection is + // gone; never a spurious success) — and its registry entry stays + // written-first, recovered by the fresh generation's re-issue. + let interrupted = instance_namer("interrupted-ch")(); + let err = forwarder.register(&interrupted).await.unwrap_err(); + assert!( + matches!(err, alkstore::Error::Database(_)), + "registration during the LISTEN-death gap fails with the honest \ + transient, got: {err:?}" + ); + + // Recovery: the fall-through reaps the dead generation, reconnects, + // and re-issues the snapshot — reconnect-wake, both channels' + // delivery back, fresh registrations healthy again. + let got_wake = wait_for_channel( + &mut rx, + alkstore::RESERVED_LISTENER_RECONNECTED, + Duration::from_secs(10), + ) + .await; + let mut delivered = false; + let mut interrupted_delivered = false; + for _ in 0..30 { + let _ = admin + .batch_execute(¬ify_sql(&channel, "post-recovery")) + .await; + let _ = admin + .batch_execute(¬ify_sql(&interrupted, "interrupted-recovered")) + .await; + if !delivered && wait_for_channel(&mut rx, &channel, Duration::from_millis(150)).await { + delivered = true; + } + if !interrupted_delivered + && wait_for_channel(&mut rx, &interrupted, Duration::from_millis(150)).await + { + interrupted_delivered = true; + } + if delivered && interrupted_delivered { + break; + } + } + assert!( + got_wake && delivered && interrupted_delivered, + "after the LISTEN-window death the forwarder must fall through to \ + reconnect and recover (wake {got_wake}, pre-death channel \ + {delivered}, mid-gap registration recovered {interrupted_delivered})" + ); + + // Exactly one reconnect-wake for the whole kill pair, observed on + // the dedicated counter subscribed before the kills: the paused + // generation's LISTEN never succeeded, so it never verified and + // never broadcast — the count is the fall-through arm's + // observable (a kill landing *after* a healthy LISTEN would have + // produced a verified generation and a second wake). First the + // single wake is received; then silence is pinned. + match tokio::time::timeout(Duration::from_secs(10), wake_counter.recv()).await { + Ok(Ok(n)) => assert_eq!( + n.channel, + alkstore::RESERVED_LISTENER_RECONNECTED, + "the counter's first delivery is the single reconnect-wake, got: {n:?}" + ), + other => panic!("the reconnect wake must reach the pre-kill counter: {other:?}"), + } + let quiet_until = tokio::time::Instant::now() + Duration::from_secs(1); + let second_wake = loop { + let remaining = quiet_until.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + break false; + } + match tokio::time::timeout(remaining, wake_counter.recv()).await { + Ok(Ok(n)) if n.channel == alkstore::RESERVED_LISTENER_RECONNECTED => { + break true; + } + Ok(Ok(_)) => continue, + Ok(Err(_)) | Err(_) => break false, + } + }; + assert!( + !second_wake, + "exactly one reconnect-wake expected — the LISTEN-failed generation \ + never verified (a second wake would mean the kill landed after a \ + healthy LISTEN, not during the LISTEN window)" + ); + + let recovery_channel = instance_namer("recovery-ch")(); + forwarder + .register(&recovery_channel) + .await + .expect("registrations succeed again post-recovery"); + + store.close(); + drop_schema(&admin, &schema).await; +} + /// Acceptance: pool bounds — `max_size` caps concurrent checkouts (the /// +1 listener sits outside the pool). #[tokio::test(flavor = "multi_thread")] @@ -926,6 +1119,34 @@ fn fanout_release_guard_empties_the_shared_slot_on_drop() { const FANOUT_TEST_CAPACITY: usize = 1; +/// Unit pin (server-less): the LISTEN-window pause seam (review 003 +/// Finding 2a) — born disarmed, arming causes exactly one generation- +/// top pause (disarms on consumption), and re-arming re-fires exactly +/// once: parallel-armed test stores cannot interfere and a paused run +/// cannot leak pauses into later generations. +#[test] +fn generation_listen_pause_consumes_once_per_arm() { + let flag = crate::seam::listen_pause::disarmed(); + assert!( + !crate::seam::listen_pause::take(&flag), + "born disarmed — no pause without arming" + ); + crate::seam::listen_pause::arm(&flag); + assert!( + crate::seam::listen_pause::take(&flag), + "one pause after the arm" + ); + assert!( + !crate::seam::listen_pause::take(&flag), + "the pause is one-shot — disarmed by its consumption" + ); + crate::seam::listen_pause::arm(&flag); + assert!( + crate::seam::listen_pause::take(&flag), + "re-arming fires exactly once again" + ); +} + /// Unit pin (server-less) — the stale-command reconcile (review 002 /// Finding 4's decision), **all four command/staleness combinations**: /// a stale-generation UNLISTEN is dropped (the hazard — replayed after diff --git a/alkstore-sqlite/src/seam.rs b/alkstore-sqlite/src/seam.rs index 3e86b49..30713f4 100644 --- a/alkstore-sqlite/src/seam.rs +++ b/alkstore-sqlite/src/seam.rs @@ -173,6 +173,24 @@ pub(crate) fn closed_store_error() -> Error { /// holds a flag per [`crate::SqliteStore`] (born disarmed), so /// parallel test stores cannot interfere; `take` disarms on read — /// exactly one commit faults, and the post-failure commits are real. +/// The `cfg(test)` rollback-fault seam (review 003 Finding 3): the +/// commit-fault mechanics pointed at the drop paths — when armed, one +/// `ROLLBACK` is substituted with a `SQLITE_FULL`-shaped driver error, +/// feeding the *real* rollback-failure drop arms (the tx handle's and +/// the cancelled op's connection lease: the error printed, the +/// unknown-state connection dropped, the writer slot replenished via +/// `reopen`) rather than replicating their behavior. No PRAGMA-space +/// probe was needed on this one: SQLite exposes no deterministic +/// in-process failure shape for `ROLLBACK` itself (its reachable +/// failure is an environment I/O error), so a seam substitution is +/// the only honest route — the same conclusion the commit seam +/// reached after probing. One-store scoping; disarms on first +/// consumption. +#[cfg(test)] +pub(crate) mod rollback_fault { + pub(crate) use super::commit_fault::{Flag, arm, disarmed, failure, take}; +} + #[cfg(test)] pub(crate) mod commit_fault { use std::sync::Arc; diff --git a/alkstore-sqlite/src/store.rs b/alkstore-sqlite/src/store.rs index 2a9e732..2f39efa 100644 --- a/alkstore-sqlite/src/store.rs +++ b/alkstore-sqlite/src/store.rs @@ -59,6 +59,14 @@ pub struct SqliteStore { /// arm. Dead in production builds. #[cfg(test)] commit_fault: crate::seam::commit_fault::Flag, + /// The store's rollback-fault arm (review 003 Finding 3) — born + /// disarmed; a test arms it through + /// [`SqliteStore::arm_rollback_fault`] so the next ROLLBACK on a + /// drop path (the handle's or the cancelled op's lease's) fails + /// through the real rollback-failure drop arms. Dead in production + /// builds. + #[cfg(test)] + rollback_fault: crate::seam::rollback_fault::Flag, } impl std::fmt::Debug for SqliteStore { @@ -90,6 +98,18 @@ impl SqliteStore { pub(crate) fn arm_commit_fault(&self) { crate::seam::commit_fault::arm(&self.commit_fault) } + + /// Arm the rollback-fault seam (review 003 Finding 3): the next + /// `ROLLBACK` on a drop path of this store (a transaction handle + /// dropped open, or a tx-op future dropped mid-op) fails with a + /// `SQLITE_FULL`-shaped driver error and replenishes the writer + /// slot through the real rollback-failure drop arm. One-store + /// scoping (disarms on first consumption) keeps parallel test + /// stores unaffected. + #[cfg(test)] + pub(crate) fn arm_rollback_fault(&self) { + crate::seam::rollback_fault::arm(&self.rollback_fault) + } } impl Drop for SqliteStore { @@ -130,6 +150,8 @@ pub(crate) fn open_store(path: &str, opts: SqliteOpts) -> alkstore::Result alkstore::Result + Send + Sync, > = std::sync::Arc::new(move || crate::seam::open_writer_connection(&db_path)); #[cfg(test)] - let fut = crate::tx::SqliteTxHandle::begin(writer, reopen, self.commit_fault.clone()); + let fut = crate::tx::SqliteTxHandle::begin( + writer, + reopen, + self.commit_fault.clone(), + self.rollback_fault.clone(), + ); #[cfg(not(test))] let fut = crate::tx::SqliteTxHandle::begin(writer, reopen); fut diff --git a/alkstore-sqlite/src/store/tx_tests.rs b/alkstore-sqlite/src/store/tx_tests.rs index cf14dfb..96ef077 100644 --- a/alkstore-sqlite/src/store/tx_tests.rs +++ b/alkstore-sqlite/src/store/tx_tests.rs @@ -1267,6 +1267,113 @@ async fn failed_commit_replenishes_the_writer_slot() { cleanup(&dir); } +/// Acceptance (review 003 Finding 3): failure of `ROLLBACK` **itself** +/// on the drop paths — the arms the commit-fault test's machinery +/// mirrors but never exercised. When the rollback fails, the unknown- +/// state connection is dropped (its error printed), and the writer +/// slot is replenished via `reopen` (the wave-3 fix's property, whose +/// commit arm the commit-fault seam proves): the next `begin_tx` +/// proceeds, the dropped connection's uncommitted writes leave no +/// ghosts (SQLite rolls an uncommitted transaction back at connection +/// close even though the `ROLLBACK` statement failed), and the store +/// stays fully usable — the next handle commits cleanly and +/// auto-commit ops work. +/// +/// **Both drop arms in one test**: `SqliteTxHandle::drop`'s +/// failed-`ROLLBACK` path (a handle dropped with an open transaction) +/// and `ConnLease::drop`'s (a tx-op future dropped mid-op — its lease +/// dies holding the connection, the mid-op-cancellation shape). +/// +/// Injection is the engine's `cfg(test)` rollback-fault seam +/// (seam.rs, the commit-fault seam's mechanics): a `SQLITE_FULL` +/// -shaped driver error substituted for one `ROLLBACK` per arming — +/// SQLite exposes no deterministic in-process failure shape for +/// `ROLLBACK` itself (its reachable failure is an environment I/O +/// error), so the seam feeds the production drop arms rather than +/// replicating them. The flag is per-store and disarms on its first +/// consumption, so parallel test stores cannot interfere. +#[tokio::test(flavor = "multi_thread")] +async fn failed_rollback_on_drop_replenishes_the_writer_slot() { + let dir = temp_dir("rollback-err"); + let concrete = open_store(dir.join("store.db").to_str().unwrap(), Default::default()).unwrap(); + + // Arm 1: the handle drop's failed-`ROLLBACK` path. + concrete.arm_rollback_fault(); + let mut tx = concrete.begin_tx().await.unwrap(); + let id = tx + .enqueue_tx("q", EnqueueOpts::default(), serde_json::json!({"n": 1})) + .await + .unwrap(); + assert!( + id > 0, + "the in-tx write landed before the failed-rollback drop" + ); + drop(tx); + + let next = tokio::time::timeout(std::time::Duration::from_secs(5), concrete.begin_tx()) + .await + .expect( + "the slot must be grantable after the failed-rollback drop \ + (replenished via reopen)", + ) + .unwrap(); + let mut next = next; + let ghost = next.get_job_tx("q", id).await.unwrap(); + assert_eq!( + ghost, None, + "the rollback-failed drop left no ghost — the dropped connection's \ + uncommitted transaction rolled back at connection close" + ); + next.commit() + .await + .expect("the post-failure handle commits cleanly (fault disarmed)"); + + // Arm 2: the cancelled op's connection-lease drop path. Exactly + // one poll fires the op's `spawn_blocking` (the blocking join's + // first poll is always Pending — it just spawned), then the + // future's drop cancels the op mid-flight: the lease that die + // holding the connection runs the `ConnLease::drop` fault arm. + concrete.arm_rollback_fault(); + let mut tx2 = concrete.begin_tx().await.unwrap(); + let mut fut = tx2.enqueue_tx("q", EnqueueOpts::default(), serde_json::json!({"n": 2})); + let waker = std::task::Waker::noop(); + let mut cx = std::task::Context::from_waker(waker); + assert!( + matches!( + std::pin::Pin::new(&mut fut).poll(&mut cx), + std::task::Poll::Pending + ), + "the op's first poll fires its blocking round trip and parks on it" + ); + drop(fut); + + let next2 = tokio::time::timeout(std::time::Duration::from_secs(5), concrete.begin_tx()) + .await + .expect( + "the slot must be grantable after the cancelled op's lease drop \ + (replenished via reopen)", + ) + .unwrap(); + let mut next2 = next2; + let ghost2 = next2.get_job_tx("q", 1).await.unwrap(); + assert_eq!( + ghost2, None, + "no ghost from the cancelled op either — whatever the race between \ + the blocking op and the drop, an uncommitted row never surfaces" + ); + next2 + .commit() + .await + .expect("the post-lease-drop handle commits cleanly"); + + concrete + .notify("probe", serde_json::json!({"after": "rollback-error"})) + .await + .expect("auto-commit ops work after both rollback-failure arms"); + drop(concrete); + cleanup(&dir); +} + /// Acceptance: a tokio task cancelled *mid-op* (its future dropped /// while an op's blocking round trip is in flight) cannot strand the /// writer slot — the op's RAII connection lease replenishes the slot diff --git a/alkstore-sqlite/src/tx.rs b/alkstore-sqlite/src/tx.rs index 3d6d5a1..7d54ba8 100644 --- a/alkstore-sqlite/src/tx.rs +++ b/alkstore-sqlite/src/tx.rs @@ -52,6 +52,11 @@ pub struct SqliteTxHandle { /// production builds. #[cfg(test)] commit_fault: crate::seam::commit_fault::Flag, + /// This store's rollback-fault arm (review 003 Finding 3) — read + /// once per drop/lease by the test-only fault branches; dead in + /// production builds. + #[cfg(test)] + rollback_fault: crate::seam::rollback_fault::Flag, } impl std::fmt::Debug for SqliteTxHandle { @@ -67,6 +72,7 @@ impl SqliteTxHandle { writer: Arc, reopen: Arc Result + Send + Sync>, #[cfg(test)] commit_fault: crate::seam::commit_fault::Flag, + #[cfg(test)] rollback_fault: crate::seam::rollback_fault::Flag, ) -> BoxedFuture<'static, Result>> { Box::pin(async move { let writer2 = writer.clone(); @@ -91,6 +97,8 @@ impl SqliteTxHandle { reopen, #[cfg(test)] commit_fault, + #[cfg(test)] + rollback_fault, }) as _) }) } @@ -114,11 +122,15 @@ impl SqliteTxHandle { }; let writer = self.writer.clone(); let reopen = self.reopen.clone(); + #[cfg(test)] + let rollback_fault = self.rollback_fault.clone(); let fut = blocking(move || { let lease = ConnLease { conn: Some(conn), writer: writer.clone(), reopen: reopen.clone(), + #[cfg(test)] + rollback_fault, }; let out = match lease.conn() { Some(conn) => f(conn), @@ -148,6 +160,11 @@ struct ConnLease { conn: Option, writer: std::sync::Arc, reopen: Arc Result + Send + Sync>, + /// The store's rollback-fault arm (review 003 Finding 3), read + /// once by the lease's test-only drop branch; dead in production + /// builds. + #[cfg(test)] + rollback_fault: crate::seam::rollback_fault::Flag, } impl ConnLease { @@ -163,7 +180,15 @@ impl ConnLease { impl Drop for ConnLease { fn drop(&mut self) { if let Some(conn) = self.conn.take() { - if let Err(rollback) = conn.execute_batch("ROLLBACK") { + #[cfg(test)] + let rolled_back = if crate::seam::rollback_fault::take(&self.rollback_fault) { + Err(crate::seam::rollback_fault::failure()) + } else { + conn.execute_batch("ROLLBACK") + }; + #[cfg(not(test))] + let rolled_back = conn.execute_batch("ROLLBACK"); + if let Err(rollback) = rolled_back { eprintln!( "alkstore: error: rollback on a cancelled transaction op \ left the connection in an unknown state: {rollback}" @@ -434,7 +459,15 @@ impl TxHandle for SqliteTxHandle { impl Drop for SqliteTxHandle { fn drop(&mut self) { if let Some(conn) = self.conn.take() { - match conn.execute_batch("ROLLBACK") { + #[cfg(test)] + let rolled_back = if crate::seam::rollback_fault::take(&self.rollback_fault) { + Err(crate::seam::rollback_fault::failure()) + } else { + conn.execute_batch("ROLLBACK") + }; + #[cfg(not(test))] + let rolled_back = conn.execute_batch("ROLLBACK"); + match rolled_back { Ok(()) => self.writer.release(conn), Err(rollback) => { eprintln!( diff --git a/tasks/release-findings-2-3-disposition.md b/tasks/release-findings-2-3-disposition.md index 9837d9d..669ee0f 100644 --- a/tasks/release-findings-2-3-disposition.md +++ b/tasks/release-findings-2-3-disposition.md @@ -1,7 +1,7 @@ --- id: release-findings-2-3-disposition name: Review 003 Findings 2–3 — the parked pre-release dispositions -status: pending +status: completed depends_on: [] scope: narrow risk: low @@ -46,14 +46,14 @@ disposition" is this task's judgment call, made explicitly in Notes. ## Acceptance Criteria -- [ ] Finding 2's (a) arm seam-tested (or its not-taken disposition +- [x] Finding 2's (a) arm seam-tested (or its not-taken disposition recorded with the invasiveness reason); the race arm + defensive arms' parked disposition recorded -- [ ] Finding 3's rollback-fault test landed covering both drop arms +- [x] Finding 3's rollback-fault test landed covering both drop arms (or its not-taken disposition recorded) -- [ ] The disposition record written so a future general review reads +- [x] The disposition record written so a future general review reads the inventory as closed, not re-derivable -- [ ] `cargo test -p alkstore-sqlite` green server-less; `cargo test +- [x] `cargo test -p alkstore-sqlite` green server-less; `cargo test -p alkstore-postgres` green server-less (SKIP posture) and — if the pg seam landed — green against the harness server; clippy `-D warnings`; fmt clean @@ -68,8 +68,179 @@ disposition" is this task's judgment call, made explicitly in Notes. ## Notes -> To be filled by implementation agent +### Finding 2(a) — landed: the generation-top LISTEN-pause seam + +**Both Findings 2(a) and 3 taken — landed, not parked. The parked +dispositions below are recorded so the next general review reads this +inventory as closed.** + +- **Finding 2(a) — taken.** The "connection died before/during the + LISTEN" fall-through (`alkstore-postgres/src/forwarder.rs`'s + `listen_ok == false` arm, review-cited at 693–698 pre-diff) is + seam-tested. The seam is a one-shot `cfg(test)` flag + (`seam.rs`'s `listen_pause`, the commit-fault shape): + `Forwarder::arm_generation_listen_pause()` arms it, and the loop + sleeps a 500 ms window at the **next** generation's top *before* + issuing its snapshot LISTEN. The test kills the fresh reconnect + generation's backend inside that window — the LISTEN batch then + executes against a dead session and fails **honestly** (real driver + error — `Error::Db(code 57P01)`, "terminating connection due to + administrator command"; only the timing is fabricated), and the + production fall-through arm runs unmodified + (`forwarder_recovers_from_connection_death_during_the_listen_window` + in `store/open_tests.rs`, run against the harness server). +- **Why this seam shape (not a blind kill-loop)**: the honest + connect→LISTEN window is microseconds, so killing in a loop hoping + to land there is sampling luck, not a test. Replay-proofed both + ways: with the pause window zeroed, the test's key differing pin + becomes non-deterministic and fails; with the 500 ms pause, 6/6 + runs green. +- **The fall-through's differentiating pin — the wake count.** The + arm's observables (fall-through → honest transient + `connected=false` state → reconnect → recovery) are pinned by: + a mid-gap `register` failing the transient `Database`, its + registry-entry recovered by the re-issue (its NOTIFY delivers + post-recovery), and — the differentiator — **exactly one** + reconnect-wake on a dedicated pre-kill counter subscription (a + generation whose LISTEN succeeded broadcasts a wake; the + LISTEN-failed one never verifies and never broadcasts; a kill + landing *after* a healthy LISTEN would produce a second). The + counter must be consumed first, then silence pinned for 1 s — the + delivery-loop receiver would otherwise swallow the wake before any + count (the first draft of this pin did exactly that; replay caught + it). +- **Small server-less unit pin** rides alongside: + `generation_listen_pause_consumes_once_per_arm` (the flag's + one-shot mechanics). + +### Finding 2(b) + defensive arms — parked with the code-read as pin + +Recorded per the review's "either way the disposition should be +recorded" — line numbers current, with the review's pre-diff sites +in parentheses (the diff added ~15 lines to `forwarder.rs`): + +- `forwarder.rs:763-771` (was 719–727) — the mid-recv-loop + stale-command abandonment (`command.generation() < generation` + inside the serve loop): **parked — race-shaped.** Deterministic + reach needs a send-racing-the-reconnect seam (a queued command + delivered mid-generational swap); the fix's *other* half is already + unit-pinned (`stale_generation_commands_drop_and_current_generation_ + replay` pins the reconcile helper, both drop outcomes, queue-order + preservation, and the dropped-LISTEN's transient ack via + `abandon_stale`), and this arm consumes the same helper's decision + at a second site — the code-read plus the shared-helper pin is the + disposition of record. +- `forwarder.rs:340` (was 330–338) — `run_command`'s + failed-`batch_execute` ack (connection dies mid-LISTEN command): + **parked — defensive.** Needs a command racing session death; + behaviorally the same transient ack `abandon_stale` already pins. +- `forwarder.rs:456,460` (was 444) — `register`'s forwarder-gone + arms: **parked — defensive.** Reachable only with the loop dead + without a shutdown flip; the wave-4 `FanoutRelease` hardening makes + exactly that the guarded anomaly, and `register` itself is + exercised live in the LISTEN-window test's mid-gap arm. +- `forwarder.rs:757,789` (was 744–745/762) — shutdown abort / poll + await arms: **parked — defensive residue.** The shutdown flip's + loop-level exit is pinned end-to-end by the existing shutdown + tests; these are the intra-select abort/await edges of the same + exit. + +None is consumer-facing; none guards a wave-4-fix property that +isn't already pinned elsewhere (the findings-class caution the +review applied to Finding 1 was about *untested hardening* — these +are the tested-elsewhere edges, which is why parked-with-code-read +is the honest line here). + +### Finding 3 — taken: the rollback-fault seam, both drop arms + +- **Taken:** a `cfg(test)` `rollback_fault` module in + `alkstore-sqlite/src/seam.rs` — the commit-fault mechanics + re-exported (`Flag`/`arm`/`take`/`disarmed`/`failure`) with the + `SQLITE_FULL`-shaped substituted error feeding the drop paths. + No PRAGMA probe was run for this one: SQLite has no deterministic + in-process failure shape for `ROLLBACK` itself (its reachable + failure is an environment I/O error), so the same seam conclusion + the commit arm reached after probing holds trivially — recorded + rather than re-probed. +- **Plumbing** mirrors `sqlite-commit-error-arm` exactly + (`#[cfg(test)]` fields on `SqliteStore`/`SqliteTxHandle`/ + `ConnLease`, the `arm_rollback_fault` store surface, cfg'd + branches in both `Drop` impls). The one-shot arm is consumed by + whichever drop path fires first — the test arms twice, once per + arm. +- **One test, both arms** (`store/tx_tests.rs:: + failed_rollback_on_drop_replenishes_the_writer_slot`): phase 1 + arms and drops an open handle (`SqliteTxHandle::drop`'s + failed-`ROLLBACK` path, was tx.rs:439–447); phase 2 arms and + cancels a tx-op **mid-flight** — exactly one poll of the op future + (fires its `spawn_blocking`; a blocking join's first poll is + always Pending, pinned by the test) then `drop` — so the lease + dies holding the connection (`ConnLease::drop`'s, was tx.rs: + 166–175). Both phases pin: slot replenished via `reopen` (next + `begin_tx` proceeds under timeout), no ghost rows (the dropped + connection's uncommitted tx rolled back at connection close even + though the `ROLLBACK` statement "failed"), and the store stays + usable (clean commits + auto-commit `notify` afterward). The + error-printed pin is the *executed* production message (verified + live via `--nocapture`: both eprintlns fired with the + `SQLITE_FULL` source) — output capture to assert it in-test is + not a house posture, so the behavioral pins carry the acceptance. +- One first-draft flaw worth noting for future seam authors: dropping + the op future **without polling** never engages the lease (the + `spawn_blocking` only fires on first poll) — the writer slot then + strands because the connection dies inside the never-polled future + with `ConnLease` never constructed. The single-poll shape is what + makes the mid-flight cancellation honest. ## Summary -> To be filled on completion \ No newline at end of file +> What landed, verified how: + +- **Finding 2(a) landed** — `alkstore-postgres/src/seam.rs`: the + one-shot `cfg(test)` `listen_pause` module (flag + 500 ms + `WINDOW`); `alkstore-postgres/src/forwarder.rs`: the `listen_pause` + flag on `Forwarder` and `LoopCtx` (cfg'd — dead in production), the + generation-top pause branch, and + `Forwarder::arm_generation_listen_pause`. + `alkstore-postgres/src/store/open_tests.rs`: + `forwarder_recovers_from_connection_death_during_the_listen_window` + (harness server) — the pause-armed kill pair puts the fresh + reconnect generation's LISTEN against a dead session (honest + `57P01` terminating-connection failure); pins the fall-through's + honest transient (mid-gap `register` errs `Database`), full + recovery (reconnect-wake, the pre-death channel's delivery back, + the mid-gap registration recovered by the re-issue, fresh + registrations healthy), and the exactly-one-wake differentiator (a + dedicated pre-kill counter consumed, then 1 s of silence — a second + wake would mean the kill landed after a healthy LISTEN). Also the + server-less unit pin + `generation_listen_pause_consumes_once_per_arm`. Replay-proofed: + window zeroed ⇒ the one-wake pin becomes non-deterministic and + fails (verified live); pause restored ⇒ 6/6 green. The kill rides a + shared `kill_listener` helper (the repeated `pg_terminate_backend` + inline shape of this file's reconnect tests, extracted). +- **Finding 2(b) + the defensive arms parked, disposition recorded + above** — mid-recv-loop stale-command abandonment (race arm), + `run_command`'s failed-batch ack, `register`'s forwarder-gone arms, + shutdown abort/await edges — each with its current line ref, the + reach reason, and the tested-elsewhere coverage it overlaps. +- **Finding 3 landed** — `alkstore-sqlite/src/seam.rs`: the + `rollback_fault` module (commit-fault mechanics pointed at the drop + paths); `alkstore-sqlite/src/tx.rs`: cfg'd fields on + `SqliteTxHandle`/`ConnLease` + cfg'd fault branches in both drop + paths; `alkstore-sqlite/src/store.rs`: `arm_rollback_fault` + the + `begin` plumbing. + `alkstore-sqlite/src/store/tx_tests.rs`: + `failed_rollback_on_drop_replenishes_the_writer_slot` — both drop + arms in one test (open-handle drop; mid-op cancellation via the + exactly-one-poll-then-drop shape), pinning slot replenish via + `reopen`, no ghosts, store usable. The fault arms' production + eprintlns verified firing live (`--nocapture`). +- **Gates**: `cargo build`; workspace `cargo test` server-less green + (no failures: core 25 + harness 3, postgres 127 lib + 25 suite + 9 + schema server-less, sqlite 193 lib + 25 suite); the pg column live + against the harness server green (127 lib + 25 suite + 9 schema); + pg server-less green (the documented SKIP posture); `cargo clippy + --all-targets -- -D warnings`; `cargo fmt --check`; the fuzz corpus + gate (`cargo test --manifest-path fuzz/shared/Cargo.toml`) — all + green. \ No newline at end of file