Wave-3 review gate: contract conformance clean; two findings fixed inline — substrate boundary move (open_writer_connection → seam.rs), writer-slot release on with_writer/begin/commit error arms + regression tests (task review-wave-3)

alkstore-sqlite/src/seam.rs

@@ -45,6 +45,17 @@ pub(crate) fn sqlite_error(e: rusqlite::Error) -> Error {
     Error::database(e)
 }
 
+/// Provision a fresh bootstrapped connection into the writer slot
+/// after a cancellation-path connection was consumed. The engine's tx
+/// handle captures this as its `reopen` closure's body; the
+/// substrate-outcome mapping stays engine-side (ADR-012 §2 — the
+/// taxonomy mapping is engine territory).
+pub(crate) fn open_writer_connection(path: &str) -> Result<rusqlite::Connection> {
+    crate::substrate::open_conn_bootstrapped(path).map_err(|e| match e {
+        crate::substrate::SchemaError::Sqlite(inner) => sqlite_error(inner),
+    })
+}
+
 /// The short-lived writer-slot lease for auto-commit ops (the
 /// `notify` path's shape): acquire, run the op inside the
 /// `spawn_blocking` seam, release — the slot is free the instant the
@@ -63,16 +74,9 @@ where
                 "the store is closed: writer slot unavailable",
             ))
         })?;
-        match f(&conn) {
-            Ok(out) => {
-                writer.release(conn);
-                Ok(out)
-            }
-            Err(e) => {
-                drop(conn);
-                Err(e)
-            }
-        }
+        let out = f(&conn);
+        writer.release(conn);
+        out
     })
     .await
 }

alkstore-sqlite/src/store.rs

@@ -120,7 +120,7 @@ impl Store for SqliteStore {
         let db_path = self.db_path.to_string_lossy().into_owned();
         let reopen: std::sync::Arc<
             dyn Fn() -> alkstore::Result<rusqlite::Connection> + Send + Sync,
-        > = std::sync::Arc::new(move || crate::substrate::open_writer_connection(&db_path));
+        > = std::sync::Arc::new(move || crate::seam::open_writer_connection(&db_path));
         crate::tx::SqliteTxHandle::begin(writer, reopen)
     }
 

alkstore-sqlite/src/store/notify_tests.rs

@@ -118,6 +118,47 @@ async fn notify_never_produces_payload_too_large() {
     cleanup(&dir);
 }
 
+/// Acceptance: an auto-commit op's *error* arm releases the writer
+/// slot — the lease-out-both-arms property (ADR-007's seam posture):
+/// after a notify whose statement fails at the storage layer, the slot
+/// is free again and a subsequent writer op succeeds without parking.
+#[tokio::test(flavor = "multi_thread")]
+async fn failed_notify_releases_the_writer_slot() {
+    let dir = temp_dir("err-arm-release");
+    let path = temp_path("err-arm-release");
+    let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
+
+    let side = rusqlite::Connection::open(&path).unwrap();
+    side.execute_batch(
+        "CREATE TRIGGER block_notify BEFORE INSERT ON __alkstore_notifications
+         BEGIN SELECT RAISE(ABORT, 'notify insert blocked'); END",
+    )
+    .unwrap();
+
+    let err = store.notify("orders", json!({"x": 1})).await.unwrap_err();
+    let source = match &err {
+        Error::Database(source) => source.to_string(),
+        other => panic!("the induced failure surfaces as Database, got {other:?}"),
+    };
+    assert!(
+        source.contains("notify insert blocked"),
+        "the source chain carries the trigger's message, got: {source}"
+    );
+    side.execute_batch("DROP TRIGGER block_notify;").unwrap();
+    drop(side);
+
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    let landed = tokio::time::timeout_at(deadline, store.notify("orders", json!({"x": 2})))
+        .await
+        .expect(
+            "the failed op must release the writer slot — the slot is free again, \
+             never parked",
+        );
+    landed.expect("the notify lands after a failed op released the slot");
+    store.close();
+    cleanup(&dir);
+}
+
 /// Acceptance: `listen` returns a working receiver — wakes arrive
 /// after commits by other connections, and `Wake { channel }` carries
 /// the channel name only. The honest posture: the watcher fans out on

alkstore-sqlite/src/store/tx_tests.rs

@@ -1015,6 +1015,49 @@ async fn failed_ops_inside_a_tx_keep_the_lease() {
     cleanup(&dir);
 }
 
+/// Acceptance: a failed `BEGIN IMMEDIATE` does not strand the writer
+/// slot — the acquired connection returns to the slot on the begin's
+/// error arm (the lease-out-both-arms property at the seam's entry)
+/// and the next `begin_tx` succeeds without parking. Induced with a
+/// concurrent exclusive holder at a shortened busy timeout, so the
+/// begin fails fast with `SQLITE_BUSY` instead of parking.
+#[tokio::test(flavor = "multi_thread")]
+async fn failed_begin_releases_the_writer_slot() {
+    let dir = temp_dir("begin-err-release");
+    let concrete = open_store(dir.join("store.db").to_str().unwrap(), Default::default()).unwrap();
+
+    {
+        let writer_conn = concrete.writer.acquire().unwrap();
+        writer_conn
+            .execute_batch("PRAGMA busy_timeout = 100;")
+            .unwrap();
+        concrete.writer.release(writer_conn);
+    }
+    let holder = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+    holder.execute_batch("BEGIN EXCLUSIVE; SELECT 1;").unwrap();
+
+    let begin = tokio::time::timeout(std::time::Duration::from_secs(5), concrete.begin_tx()).await;
+    match begin {
+        Err(_) => panic!("begin must fail fast at the 100 ms busy timeout, not park"),
+        Ok(Ok(_handle)) => {
+            panic!("begin must refuse under the exclusive holder, got a lease")
+        }
+        Ok(Err(e)) => assert!(
+            matches!(e, Error::Database(_)),
+            "a busy-refused begin errs opaque Database, got: {e:?}"
+        ),
+    }
+    drop(holder);
+
+    let next = tokio::time::timeout(std::time::Duration::from_secs(5), concrete.begin_tx())
+        .await
+        .expect("the slot must be grantable after the failed begin")
+        .unwrap();
+    drop(next);
+    drop(concrete);
+    cleanup(&dir);
+}
+
 /// A concurrent writer's commit blocks while a lease holds the slot —
 /// the parked-writer posture (ADR-007 negative consequence, honest).
 #[tokio::test(flavor = "multi_thread")]

alkstore-sqlite/src/substrate/mod.rs

@@ -66,13 +66,3 @@ pub(crate) use schema::Error as SchemaError;
 pub(crate) use schema::open_conn_bootstrapped;
 pub(crate) use schema::{Readers, Writer};
 pub(crate) use watcher::{SharedUpdateWatcher, WatcherConfig};
-
-/// Provision a fresh connection into the writer slot after a
-/// cancellation-path connection was consumed. Private helper over the
-/// substrate's bootstrapped open; the engine's tx handle captures it
-/// as its `reopen` closure's body.
-pub(crate) fn open_writer_connection(path: &str) -> Result<rusqlite::Connection, alkstore::Error> {
-    open_conn_bootstrapped(path).map_err(|e| match e {
-        SchemaError::Sqlite(inner) => alkstore::Error::database(inner),
-    })
-}

alkstore-sqlite/src/tx.rs

@@ -70,9 +70,13 @@ impl SqliteTxHandle {
                         "the store is closed: writer slot unavailable",
                     ))
                 })?;
-                conn.execute_batch("BEGIN IMMEDIATE")
-                    .map_err(sqlite_error)
-                    .map(|_| conn)
+                match conn.execute_batch("BEGIN IMMEDIATE").map_err(sqlite_error) {
+                    Ok(()) => Ok(conn),
+                    Err(e) => {
+                        writer2.release(conn);
+                        Err(e)
+                    }
+                }
             })
             .await?;
             Ok(Box::new(SqliteTxHandle {
@@ -387,12 +391,23 @@ impl TxHandle for SqliteTxHandle {
             return Box::pin(async { Err(Error::Closed) });
         };
         let writer = self.writer.clone();
+        let writer_reopen = self.reopen.clone();
         Box::pin(async move {
-            blocking(move || {
-                conn.execute_batch("COMMIT").map_err(sqlite_error)?;
-                writer.release(conn);
-                Ok(())
-            })
+            blocking(
+                move || match conn.execute_batch("COMMIT").map_err(sqlite_error) {
+                    Ok(()) => {
+                        writer.release(conn);
+                        Ok(())
+                    }
+                    Err(e) => {
+                        drop(conn);
+                        if let Ok(fresh) = (writer_reopen)() {
+                            writer.release(fresh);
+                        }
+                        Err(e)
+                    }
+                },
+            )
             .await
         })
     }