From 22f25ffafd2a0ec9f40eb3c40116ce5e0f7a0daf Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 11 Oct 2026 09:55:45 +0000 Subject: [PATCH] sqlite: panic-safety armor for the auto-commit seams (review 004 item 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit with_writer/with_reader/blocking_read release their slot/pool connection only on the normal return — a panic inside the closure wedged the single writer slot forever (Writer::acquire waits on a condvar) and permanently shrank the reader pool (outstanding never decremented on abandonment). The seams now ride RAII drop-guards (ConnLease's idiom): the normal arm defuses and releases unchanged; the unwind arm drops the unknown-state connection and takes the established abandonment dispositions — Writer::replenish() (the tick's poisoned arm as a substrate method, the writer now holding its path) and Readers::abandon() (outstanding decrement + pool-locked notify). Three panic-injection tests pin the properties; no reachable panic path exists in production — this is the defense-in-depth armor the review scoped. Task: sqlite-fix-seam-panic-safety. --- alkstore-sqlite/src/seam.rs | 120 +++++++++++++++-- alkstore-sqlite/src/store.rs | 4 +- alkstore-sqlite/src/store/seam_tests.rs | 163 ++++++++++++++++++++++++ alkstore-sqlite/src/substrate/schema.rs | 48 ++++++- tasks/sqlite-fix-seam-panic-safety.md | 84 +++++++++++- 5 files changed, 403 insertions(+), 16 deletions(-) create mode 100644 alkstore-sqlite/src/store/seam_tests.rs diff --git a/alkstore-sqlite/src/seam.rs b/alkstore-sqlite/src/seam.rs index 30713f4..2d4b1f4 100644 --- a/alkstore-sqlite/src/seam.rs +++ b/alkstore-sqlite/src/seam.rs @@ -62,7 +62,13 @@ pub(crate) fn open_writer_connection(path: &str) -> Result /// op completes (no lease held across a consumer's `await` points; /// that is the long-transaction posture, not this path's). A closed /// store's acquire fails closed with `Database` (the engine-wide -/// closed-store shape). +/// closed-store shape). The connection rides an RAII [`WriterLease`]: +/// a panic inside the closure unwinds through the lease's drop, which +/// drops the connection (unknown state — dropped, not released) and +/// replenishes the slot via a fresh bootstrapped open +/// ([`Writer::replenish`], the substrate's abandonment arm), so the +/// slot stays grantable after the unwind and the `JoinError` maps to +/// `Database` below. pub(crate) async fn with_writer(writer: Arc, f: F) -> alkstore::Result where T: Send + 'static, @@ -74,17 +80,64 @@ where "the store is closed: writer slot unavailable", )) })?; - let out = f(&conn); - writer.release(conn); + let lease = WriterLease { + conn: Some(conn), + writer: writer.clone(), + }; + let out = match lease.conn() { + Some(conn) => f(conn), + None => Err(Error::database(std::io::Error::other( + "the store is closed: writer slot unavailable", + ))), + }; + if let Some(conn) = lease.defuse() { + writer.release(conn); + } out }) .await } +/// The writer connection inside the auto-commit seam: an RAII guard so +/// the leased slot connection always has an owner that accounts for +/// the slot. Handled normally, [`WriterLease::defuse`] takes the +/// connection out and the caller releases it to the slot exactly as +/// before; on an unwind (a panicked op — no reachable path today, but +/// the seams take no consumer code they can't trust to stay that way) +/// the guard drops instead, taking the drop-and-replenish disposition +/// ([`Writer::replenish`]) rather than releasing the connection into +/// an unknown state. The same class as the tx seam's `ConnLease`. +struct WriterLease { + conn: Option, + writer: Arc, +} + +impl WriterLease { + fn conn(&self) -> Option<&rusqlite::Connection> { + self.conn.as_ref() + } + + fn defuse(mut self) -> Option { + self.conn.take() + } +} + +impl Drop for WriterLease { + fn drop(&mut self) { + if let Some(conn) = self.conn.take() { + drop(conn); + self.writer.replenish(); + } + } +} + /// A reader-pool read through the `spawn_blocking` seam: acquire a /// pooled reader, run the body, release on both arms (a failed SELECT -/// leaves the pooled connection healthy). A closed pool (store closed) -/// fails closed with `Database`. +/// leaves the pooled connection healthy; a panicked body unwinds +/// through an RAII [`ReaderLease`], which drops the connection and +/// decrements the pool's open-budget — [`Readers::abandon`] — so the +/// pool's capacity is restored and a later reader acquires). A closed +/// pool (store closed) fails closed with `Database`. pub(crate) async fn with_reader(readers: Arc, f: F) -> Result where T: Send + 'static, @@ -92,8 +145,17 @@ where { blocking(move || { let conn = acquire_reader(&readers)?; - let out = f(&conn); - readers.release(conn); + let lease = ReaderLease { + conn: Some(conn), + readers: readers.clone(), + }; + let out = match lease.conn() { + Some(conn) => f(conn), + None => Err(closed_store_error()), + }; + if let Some(conn) = lease.defuse() { + readers.release(conn); + } out }) .await @@ -135,8 +197,17 @@ pub(crate) fn blocking_read( } }, }; - let out = f(&conn); - readers.release(conn); + let lease = ReaderLease { + conn: Some(conn), + readers: readers.clone(), + }; + let out = match lease.conn() { + Some(conn) => f(conn), + None => Err(closed_store_error()), + }; + if let Some(conn) = lease.defuse() { + readers.release(conn); + } match out { Ok(value) => Ok(value), Err(_e) if store_closed.load(Ordering::Acquire) => Err(ReaderLoopExit::Closed), @@ -150,6 +221,37 @@ pub(crate) enum ReaderLoopExit { Failed(Error), } +/// The pooled reader connection inside the reader seams +/// ([`with_reader`] and [`blocking_read`]): the same RAII class as +/// [`WriterLease`] — handled normally the guard defuses and the +/// connection releases to the pool exactly as before; on an unwind +/// (a panicked read) the guard drops instead, taking the drop-and- +/// decrement disposition ([`Readers::abandon`]): the connection is +/// dropped, not released, and the pool's open-budget is restored. +struct ReaderLease { + conn: Option, + readers: Arc, +} + +impl ReaderLease { + fn conn(&self) -> Option<&rusqlite::Connection> { + self.conn.as_ref() + } + + fn defuse(mut self) -> Option { + self.conn.take() + } +} + +impl Drop for ReaderLease { + fn drop(&mut self) { + if let Some(conn) = self.conn.take() { + drop(conn); + self.readers.abandon(); + } + } +} + pub(crate) fn is_closed_err(e: &rusqlite::Error) -> bool { matches!( e, diff --git a/alkstore-sqlite/src/store.rs b/alkstore-sqlite/src/store.rs index 59d1e4e..44105a1 100644 --- a/alkstore-sqlite/src/store.rs +++ b/alkstore-sqlite/src/store.rs @@ -143,7 +143,7 @@ pub(crate) fn open_store(path: &str, opts: SqliteOpts) -> alkstore::Result PathBuf { + let d = std::env::temp_dir().join(format!( + "alkstore-seam-panic-{tag}-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + let _ = std::fs::remove_dir_all(&d); + std::fs::create_dir_all(&d).unwrap(); + d +} + +fn cleanup(dir: &PathBuf) { + let _ = std::fs::remove_dir_all(dir); +} + +/// A full store operation that must succeed after each abandonment — +/// the write path (the slot's real client) and its read-back. +async fn full_store_operation(store: &dyn Store) { + let queue = store + .queue("after-panic", QueueOpts::default()) + .await + .unwrap(); + let id = queue + .enqueue(json!({"survived": true}), EnqueueOpts::default()) + .await + .unwrap(); + let job = queue + .get_job(id) + .await + .unwrap() + .expect("the enqueued row must read back after the panic"); + assert_eq!(job.id, id); +} + +/// Acceptance (writer): a panic inside the `with_writer` closure +/// unwinds through [`crate::seam::WriterLease`] — the JoinError maps +/// to `Database`, the slot is replenished, and a later acquire plus a +/// full store operation succeed (before the armor, the slot stayed +/// empty forever: `Writer::acquire` waits on a condvar with no +/// timeout). +#[tokio::test(flavor = "multi_thread")] +async fn panicked_writer_op_leaves_the_slot_grantable() { + let dir = temp_dir("writer-slot"); + let path = dir.join("store.db"); + let store = open_store(path.to_str().unwrap(), Default::default()).unwrap(); + + let writer = store.writer.clone(); + let outcome = crate::seam::with_writer(writer, |_conn| -> alkstore::Result<()> { + panic!("injected seam panic (writer)") + }) + .await; + match outcome { + Err(Error::Database(_)) => {} + other => panic!("expected Database error from the panic, got {other:?}"), + } + + let conn = store + .writer + .acquire() + .expect("the slot must be grantable after the unwind"); + store.writer.release(conn); + + full_store_operation(&store).await; + store.close(); + cleanup(&dir); +} + +/// Acceptance (reader, async seam): a panic inside the `with_reader` +/// closure unwinds through [`crate::seam::ReaderLease`] — the +/// connection is dropped, not released, and the pool's open-budget is +/// restored ([`crate::substrate::Readers::abandon`]), so a later +/// reader acquires (before the armor, `max_readers: 1`'s connection +/// was gone and one abandonment permanently shrank the pool to zero). +#[tokio::test(flavor = "multi_thread")] +async fn panicked_reader_op_restores_the_pool_capacity() { + let dir = temp_dir("reader-pool"); + let path = dir.join("store.db"); + let store = open_store( + path.to_str().unwrap(), + crate::opts::SqliteOpts { + max_readers: 1, + ..Default::default() + }, + ) + .unwrap(); + + let readers = store.readers.clone(); + let outcome = crate::seam::with_reader(readers, |_conn| -> alkstore::Result<()> { + panic!("injected seam panic (reader)") + }) + .await; + match outcome { + Err(Error::Database(_)) => {} + other => panic!("expected Database error from the panic, got {other:?}"), + } + + let conn = store + .readers + .acquire() + .expect("a later reader must acquire after the abandonment"); + store.readers.release(conn); + + full_store_operation(&store).await; + store.close(); + cleanup(&dir); +} + +/// Acceptance (reader, sync bridge seam): the same property through +/// `blocking_read` — the unwind is caught at the test seam (the panic +/// propagates past the synchronous bridge), the pool's capacity is +/// restored, and a later reader acquires. +#[test] +fn panicked_blocking_read_restores_the_pool_capacity() { + let dir = temp_dir("blocking-read"); + let path = dir.join("store.db"); + let store = open_store( + path.to_str().unwrap(), + crate::opts::SqliteOpts { + max_readers: 1, + ..Default::default() + }, + ) + .unwrap(); + + let readers = store.readers.clone(); + let closed: Arc = store.closed.clone(); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || { + crate::seam::blocking_read(&readers, &closed, |_conn| -> alkstore::Result<()> { + panic!("injected seam panic (blocking read)") + }) + })); + assert!(result.is_err(), "the panic must propagate past the seam"); + + let conn = store + .readers + .acquire() + .expect("a later reader must acquire after the abandonment"); + store.readers.release(conn); + + store.close(); + cleanup(&dir); +} diff --git a/alkstore-sqlite/src/substrate/schema.rs b/alkstore-sqlite/src/substrate/schema.rs index aaddf1c..da2b12f 100644 --- a/alkstore-sqlite/src/substrate/schema.rs +++ b/alkstore-sqlite/src/substrate/schema.rs @@ -348,14 +348,39 @@ pub(crate) struct Writer { slot: Mutex>, available: Condvar, closed: AtomicBool, + path: String, } impl Writer { - pub(crate) fn new(conn: Connection) -> Self { + pub(crate) fn new(path: String, conn: Connection) -> Self { Self { slot: Mutex::new(Some(conn)), available: Condvar::new(), closed: AtomicBool::new(false), + path, + } + } + + /// The abandonment arm (the auto-commit seam's panic-safety armor, + /// review 004 item 1): a connection that cannot return to the slot + /// — a panicked op leaves its statement state unknown — is dropped, + /// not released, and the slot is replenished with a fresh + /// bootstrapped open from the writer's own path. The release's slot + /// notify re-opens the waiters. A failed replenish leaves the slot + /// empty (the established arms' tolerance — the tick's + /// poisoned-disposition arm and the tx seam's `ConnLease` drop arm + /// both take `if let Ok(fresh)`), with the failure printed for + /// diagnosability. + pub(crate) fn replenish(&self) { + if self.closed.load(Ordering::Acquire) { + return; + } + match open_conn_bootstrapped(&self.path) { + Ok(fresh) => self.release(fresh), + Err(e) => eprintln!( + "alkstore: error: writer slot replenish after an abandoned \ + operation failed to open a fresh connection: {e}" + ), } } @@ -453,6 +478,23 @@ impl Readers { self.available.notify_one(); } + /// The abandonment arm (the auto-commit seam's panic-safety + /// armor, review 004 item 1): the abandoned connection is dropped, + /// not released. `outstanding` counts connections in existence — + /// every connection carries its fresh-open increment for life, and + /// release/pop never decrement — so the abandoned connection's + /// decrement restores the pool's open-budget, and a waiter wakes + /// under the pool lock (the release's notify discipline) to + /// re-evaluate: pool miss + under `max` ⇒ fresh open. + pub(crate) fn abandon(&self) { + { + let mut out = self.outstanding.lock(); + *out = out.saturating_sub(1); + } + let _pool = self.pool.lock(); + self.available.notify_one(); + } + pub(crate) fn close(&self) { self.closed.store(true, Ordering::Release); self.pool.lock().clear(); @@ -490,14 +532,14 @@ mod writer_reader_tests { #[test] fn writer_close_drops_idle_connection() { - let w = Writer::new(mem()); + let w = Writer::new(":memory:".to_string(), mem()); w.close(); assert!(w.acquire().is_none()); } #[test] fn writer_close_drops_returned_connection() { - let w = Writer::new(mem()); + let w = Writer::new(":memory:".to_string(), mem()); let conn = w.acquire().unwrap(); w.close(); w.release(conn); diff --git a/tasks/sqlite-fix-seam-panic-safety.md b/tasks/sqlite-fix-seam-panic-safety.md index c9cca39..0e4e71a 100644 --- a/tasks/sqlite-fix-seam-panic-safety.md +++ b/tasks/sqlite-fix-seam-panic-safety.md @@ -1,7 +1,7 @@ --- id: sqlite-fix-seam-panic-safety name: Panic-safety armor for the sqlite auto-commit seams (review 004 item 1) -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -77,8 +77,86 @@ restored and a later reader acquires. ## Notes -> To be filled by implementation agent +**Design of record** (decided before implementation): + +- **Mechanism**: RAII drop-guards in `seam.rs`, following the tx + seam's established `ConnLease` idiom (`tx.rs::ConnLease`). Two + guards: `WriterLease` (for `with_writer`) and a shared `ReaderLease` + (for `with_reader` and `blocking_read`), each holding the lease + connection in an `Option` plus an owned + `Arc` of the substrate object. The normal arm defuses the guard + (`lease.defuse()` yields the connection; the caller releases it to + the slot/pool exactly as before the armor — normal paths are + behaviorally unchanged). The unwind arm is `Drop`: the connection + is **dropped, not released** (a panicked op leaves its statement + state unknown; SQLite rolls back its uncommitted state at close — + no rollback attempt is made, unlike `ConnLease`, these are + auto-commit ops with no intentional transaction), then the + abandonment disposition runs. +- **Writer disposition**: drop + replenish, per the tick's + established poisoned arm (review 004 inline fix 1, + `scheduler.rs:235-244`). The replenish lives on the substrate: + `Writer` gains a `path: String` field (symmetric with `Readers`, + which already held its path — this avoided threading a reopen + closure through ~20 `with_writer` call sites) and a + `Writer::replenish()` method that opens a fresh bootstrapped + connection from its own path and `release`s it into the slot; on + open failure the slot stays empty with an `eprintln` for + diagnosability — the established arms' tolerance (`ConnLease`'s + drop and the tick's arm both take `if let Ok(fresh)`). + `Writer::new(path, conn)` now takes the path (call site: + `store.rs` only). A failed replenish with a `:memory:` path opens a + fresh private db — acceptable, since `:memory:` connections are + documented as each-private anyway (the store open test's posture). +- **Reader disposition**: outstanding decrement, connection dropped + not released. `Readers::abandon()` decrements `outstanding` + (saturating — the accounting precises out: every connection carries + its fresh-open increment for life, release/pop never decrement, so + `outstanding` counts connections in existence and an abandonment is + exactly one destruction) and wakes one waiter **under the pool + lock** (`Readers::available` waits on the pool mutex — the notify + must be race-safe against a waiter between its pool-empty check and + `wait`, matching `release`'s notify discipline); the woken waiter + re-evaluates pool-miss + under-`max` and opens fresh, restoring + capacity. No immediate replacement open — the next acquire repairs + lazily. +- **Testability**: panics injected by passing panicking closures + directly to the `pub(crate)` seam functions from `store/seam_tests.rs` + (same crate). `with_writer`/`with_reader` panics surface as + `Err(Database)` via the existing `JoinError` mapping — the unwound + panic payload is consumed by tokio; `blocking_read` is synchronous + (propagates past the seam), so its test wraps the call in + `catch_unwind(AssertUnwindSafe(...))`. Post-unwind assertions run + behaviorally: slot reacquired, `max_readers: 1` reader re-acquired + (the pre-armor state would hang the acquire on both — the condvars + have no timeout), and a full store enqueue/read-back succeeds. ## Summary -> To be filled on completion \ No newline at end of file +Landed the auto-commit seams' panic-safety armor (review 004 +"Findings recorded, not fixed" item 1): + +- `alkstore-sqlite/src/substrate/schema.rs`: + `Writer` carries its path and gained `replenish()` (the tick's + poisoned-disposition arm as a substrate method); `Readers` gained + `abandon()` (outstanding decrement + pool-lock-held notify_one). +- `alkstore-sqlite/src/seam.rs`: `WriterLease` and `ReaderLease` + drop-guards now own the connection across all three seams + (`with_writer`, `with_reader`, `blocking_read`); the normal arm + defuses and releases exactly as before; the unwind arm drops the + connection and takes the replenish/decrement disposition. The three + seams' doc comments name the armor. +- `alkstore-sqlite/src/store/seam_tests.rs` (new): three + panic-injection tests pinning writer-slot grantability (re-acquire + + full store op after), reader-pool capacity restoration + (`max_readers: 1` re-acquire + full store op), and the synchronous + `blocking_read` path via `catch_unwind`. + +Verified: `cargo test -p alkstore-sqlite` (197 lib + 25 suite, green — +the pre-existing seam/slot/pool tests untouched and passing, normal +paths behaviorally unchanged), `cargo test --workspace` (all columns +green server-less), `cargo clippy --all-targets -- -D warnings`, `cargo +fmt --check`, the fuzz corpus replay (`fuzz/shared`, 6/6), and the doc +gate (`RUSTDOCFLAGS="-D warnings" cargo doc --no-deps` — zero warnings). +No reachable panic path exists in production today; this is the +defense-in-depth armor the review scoped. \ No newline at end of file