fix(postgres): the wave-4 review gate's inline findings

- PG-1: sweep_lock_acquire reaps a lease whose held session died
  (in-instance session-loss reaping, ADR-013 §2) — previously the
  mutex answered Ok(false) until an explicit release; new test
  terminates the lease's own backend via pg_terminate_backend and
  re-acquires without a release
- PG-3: the engine-level pg tests now drop their per-test schemas
  (test_support::SchemaJanitor, the Harness deref-guard wrapper in
  src/postgres.rs's tests, fresh_opts carrying a janitor in the
  construction tests) — 380 leaked abt_* schemas on the shared
  harness db are 0 after every run now
- PG-2: PgJointTx's doc records the mid-flight-drop hazard (deadpool
  Fast recycle never resets transaction state; the clause bodies'
  failure paths all ROLLBACK — the store core's joint-tx driver owes
  pin_commit on every path from begin_joint_tx; wave 6 amendment)

Verification: cargo test (69/19/3), --features mem (124/54/4),
--features postgres vs server (103/36/3, 51 clean skips server-less),
--all-features vs server (158/71/4), pg-pure (76/19/3);
clippy -D warnings green (default/postgres/all-features/pg-pure);
fmt --check green; shared-db schema residue 0; cargo doc clean

src/postgres.rs

@@ -848,7 +848,16 @@ impl Backend for PostgresBackend {
 /// on every failure path). Exclusive checkout ownership is what makes
 /// the session-level transaction sound; dropping the value before
 /// `pin_commit` is outside the seam's contract (the store core holds
-/// the tx — ADR-010 §2's ownership rule, pinned in the trait).
+/// the tx — ADR-010 §2's ownership rule, pinned in the trait), and it
+/// is a load-bearing exclusion: the clause bodies' failure paths close
+/// the tx themselves (`ROLLBACK`, then `Err`), but a *mid-flight* drop
+/// (an error path between [`PostgresBackend::begin_joint_tx`] and
+/// [`PostgresBackend::pin_commit`] leaking through `?`) strands the
+/// open transaction on its checkout — deadpool's `Fast` recycle never
+/// resets transaction state, so the pool would hand the dirty session
+/// to the next checkout. The store core's joint-tx driver (wave 6)
+/// therefore owes pin_commit on **every** path from `begin_joint_tx`
+/// onward; the recorded wave-4 review amendment names this duty.
 pub(crate) struct PgJointTx {
     client: Object,
     blobs_table: String,
@@ -1118,11 +1127,18 @@ impl EngineFleetState for PostgresBackend {
     /// contention — either a second call on this backend (the one-
     /// sweeper-per-instance mutex answers before any pool checkout) or
     /// another session on the same schema holding the lock. Session
-    /// loss reaps a held lease server-side; owner death reaps it at
-    /// checkout drop.
+    /// loss reaps a held lease server-side and **in-instance**: a lease
+    /// whose held checkout is closed is dropped here (ADR-013 §2's
+    /// holdership rule — the acquisition answers against a live
+    /// session only, so a fresh acquire succeeds without a prior
+    /// release); owner death reaps it at checkout drop.
     async fn sweep_lock_acquire(&self) -> Result<bool, Error> {
-        if self.lease.lock().is_some() {
-            return Ok(false);
+        {
+            let mut lease = self.lease.lock();
+            if lease.as_ref().is_some_and(|client| !client.is_closed()) {
+                return Ok(false);
+            }
+            *lease = None;
         }
         let client = self.pool.get().await.map_err(err)?;
         let stmt = client
@@ -1170,12 +1186,31 @@ impl EngineFleetState for PostgresBackend {
 
 #[cfg(test)]
 mod tests {
+    use super::test_support::SchemaJanitor;
     use super::*;
     use crate::gc::{EngineFleetState, EngineWindowState};
+    use std::ops::Deref;
     use std::sync::atomic::{AtomicU64, Ordering};
 
     const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";
 
+    /// The guarded harness backend: derefs to the engine, and the test's
+    /// own schema is dropped when the guard drops — success, early
+    /// return, or unwind included (the janitor is declared after the
+    /// backend, so the pool drains first).
+    struct Harness {
+        backend: PostgresBackend,
+        _janitor: SchemaJanitor,
+    }
+
+    impl Deref for Harness {
+        type Target = PostgresBackend;
+
+        fn deref(&self) -> &PostgresBackend {
+            &self.backend
+        }
+    }
+
     /// A fresh engine-owned schema per test — the alkstore suite
     /// factory's fresh-schema-per-open convention: every test opens its
     /// own schema on the shared harness database, and the virgin-store
@@ -1217,35 +1252,45 @@ mod tests {
     }
 
     /// Open a backend on a fresh schema — or `None` (skip note printed)
-    /// when the harness server is not configured.
-    async fn harness(tag: &str) -> Option<PostgresBackend> {
+    /// when the harness server is not configured. The backend rides a
+    /// guard that drops the test's schema when the test ends.
+    async fn harness(tag: &str) -> Option<Harness> {
         harness_with(tag, PgOpts::default()).await
     }
 
-    async fn harness_with(tag: &str, opts: PgOpts) -> Option<PostgresBackend> {
+    async fn harness_with(tag: &str, opts: PgOpts) -> Option<Harness> {
         let Some(dsn) = harness_dsn() else {
             skip_note();
             return None;
         };
+        let schema = fresh_schema(tag);
         let opts = PgOpts {
-            schema: fresh_schema(tag),
+            schema: schema.clone(),
             ..opts
         };
-        Some(PostgresBackend::new(&dsn, opts).await.expect("open"))
+        let backend = PostgresBackend::new(&dsn, opts).await.expect("open");
+        Some(Harness {
+            backend,
+            _janitor: SchemaJanitor::new(&dsn, &schema),
+        })
     }
 
     /// The virgin-store fixture: a pool connected to a schema that was
-    /// never bootstrapped (the sqlite `new_tableless` pattern).
-    async fn harness_schemaless(tag: &str) -> Option<PostgresBackend> {
+    /// never bootstrapped (the sqlite `new_tableless` pattern). The
+    /// guard drops the (table-less) schema at test end.
+    async fn harness_schemaless(tag: &str) -> Option<Harness> {
         let Some(dsn) = harness_dsn() else {
             skip_note();
             return None;
         };
-        Some(
-            PostgresBackend::new_schemaless(&dsn, &fresh_schema(tag))
-                .await
-                .expect("open"),
-        )
+        let schema = fresh_schema(tag);
+        let backend = PostgresBackend::new_schemaless(&dsn, &schema)
+            .await
+            .expect("open");
+        Some(Harness {
+            backend,
+            _janitor: SchemaJanitor::new(&dsn, &schema),
+        })
     }
 
     /// The real stored value, observed through the read seam — the
@@ -1748,6 +1793,7 @@ mod tests {
         };
         let first = PostgresBackend::new(&dsn, opts.clone()).await.unwrap();
         let second = PostgresBackend::new(&dsn, opts).await.unwrap();
+        let _janitor = SchemaJanitor::new(&dsn, &schema);
 
         let client = first.pool.get().await.unwrap();
         let columns: Vec<String> = client
@@ -1848,12 +1894,13 @@ mod tests {
         let backend = PostgresBackend::new(
             &dsn,
             PgOpts {
-                schema,
+                schema: schema.clone(),
                 ..Default::default()
             },
         )
         .await
         .expect("open with a hostile schema name");
+        let _janitor = SchemaJanitor::new(&dsn, &schema);
 
         let key = b"\x01hostile-schema-key".to_vec();
         backend.put(&key, CONTENT).await.unwrap();
@@ -1906,7 +1953,7 @@ mod tests {
     /// A constructed backend with the GC schema ensured — the clause
     /// tests' fixture (store construction always wires the ensure call;
     /// the backend constructor itself never creates the GC tables).
-    async fn harness_gc(tag: &str) -> Option<PostgresBackend> {
+    async fn harness_gc(tag: &str) -> Option<Harness> {
         let backend = harness(tag).await?;
         backend.ensure_gc_schema().await.unwrap();
         Some(backend)
@@ -2233,7 +2280,7 @@ mod tests {
         let owner = b"\x01instance-id".to_vec();
 
         let mut tx = backend.begin_joint_tx().await.unwrap();
-        EngineFleetState::joint_put(&backend, &mut tx, &key, CONTENT)
+        EngineFleetState::joint_put(&*backend, &mut tx, &key, CONTENT)
             .await
             .unwrap();
         let expiry = SystemTime::now() + DEFAULT_PIN_TTL;
@@ -2257,7 +2304,7 @@ mod tests {
         );
 
         let mut tx = backend.begin_joint_tx().await.unwrap();
-        EngineFleetState::joint_put(&backend, &mut tx, &key, b"overwritten")
+        EngineFleetState::joint_put(&*backend, &mut tx, &key, b"overwritten")
             .await
             .unwrap();
         let renewed = SystemTime::now() + DEFAULT_PIN_TTL + Duration::from_secs(7);
@@ -2703,4 +2750,108 @@ mod tests {
         );
         backend.sweep_lock_release().await.unwrap();
     }
+
+    /// In-instance session-loss reaping (ADR-013 §2): a lease whose
+    /// *own* held session dies is reaped at the next acquire — no
+    /// release call is owed first. The lease's held checkout is
+    /// terminated server-side (`pg_terminate_backend`), so its
+    /// advisory lock is gone; the mutex-held closed session must not
+    /// block re-acquisition forever.
+    #[tokio::test]
+    async fn sweep_lock_in_instance_session_loss_reaps_the_lease() {
+        let Some(backend) = harness_gc("sweep-reap-own").await else {
+            return;
+        };
+
+        assert!(backend.sweep_lock_acquire().await.unwrap(), "held");
+        let held_client = backend
+            .lease
+            .lock()
+            .take()
+            .expect("the acquire stored its checkout");
+        let pid = held_client
+            .query_one("SELECT pg_backend_pid()", &[])
+            .await
+            .unwrap()
+            .get::<_, i32>(0);
+        *backend.lease.lock() = Some(held_client);
+
+        let dsn = harness_dsn().unwrap();
+        let (raw, _conn) = raw_connection(&dsn).await;
+        raw.query_one("SELECT pg_terminate_backend($1)", &[&pid])
+            .await
+            .unwrap();
+        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+        assert!(
+            backend.sweep_lock_acquire().await.unwrap(),
+            "the dead in-instance lease is reaped without a release call"
+        );
+        assert!(
+            !backend.sweep_lock_acquire().await.unwrap(),
+            "the re-acquired lease contends again"
+        );
+        backend.sweep_lock_release().await.unwrap();
+    }
+}
+
+/// Test-support helpers shared with the crate's other postgres-attached
+/// test modules (`src/store.rs`'s construction tests) — crate-internal
+/// and compiled out of every non-test build.
+#[cfg(test)]
+pub(crate) mod test_support {
+    use super::quote_identifier;
+    use tokio_postgres::NoTls;
+
+    /// Drops a test's engine-owned schema when the test ends — success,
+    /// early return, or unwind (Drop runs in all three), so a fresh
+    /// schema per test never accumulates on the shared harness
+    /// database. No async teardown exists at Drop, so the drop executes
+    /// synchronously from a std thread over a one-shot current-thread
+    /// runtime; an unreachable harness server is tolerated (the residue
+    /// stays; nothing else breaks).
+    ///
+    /// Construct it (or a wrapper holding it) *before* the store or
+    /// backend it outlives: drop order is reverse of declaration, so
+    /// the backend's pool drains before the schema drops.
+    pub(crate) struct SchemaJanitor {
+        dsn: String,
+        schema: String,
+    }
+
+    impl SchemaJanitor {
+        pub(crate) fn new(dsn: &str, schema: &str) -> SchemaJanitor {
+            SchemaJanitor {
+                dsn: dsn.to_owned(),
+                schema: schema.to_owned(),
+            }
+        }
+    }
+
+    impl Drop for SchemaJanitor {
+        fn drop(&mut self) {
+            let dsn = self.dsn.clone();
+            let schema = self.schema.clone();
+            let _ = std::thread::spawn(move || {
+                let Ok(rt) = tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                else {
+                    return;
+                };
+                rt.block_on(async move {
+                    let Ok((client, conn)) = tokio_postgres::connect(&dsn, NoTls).await else {
+                        return;
+                    };
+                    tokio::spawn(conn);
+                    let sql = format!(
+                        "DROP SCHEMA IF EXISTS {} CASCADE",
+                        quote_identifier(&schema)
+                    );
+                    let _ = client.execute(sql.as_str(), &[]).await;
+                });
+            })
+            .join();
+        }
+    }
 }

src/store.rs

@@ -4839,6 +4839,7 @@ mod sqlite_window_tests {
 #[cfg(all(test, feature = "postgres"))]
 mod postgres_construction_tests {
     use super::*;
+    use crate::postgres::test_support::SchemaJanitor;
     use std::fs;
     use std::io::Cursor;
     use std::path::{Path, PathBuf};
@@ -4916,14 +4917,19 @@ mod postgres_construction_tests {
         );
     }
 
-    /// A fresh-schema opts for the harness server, or `None` (skip
-    /// note printed) server-less.
-    fn fresh_opts(tag: &str) -> Option<PgOpts> {
-        harness_dsn()?;
-        Some(PgOpts {
-            schema: fresh_schema(tag),
-            ..Default::default()
-        })
+    /// A fresh-schema opts for the harness server — with the schema
+    /// janitor that drops it at test end (success, early return, or
+    /// unwind) — or `None` (skip note printed) server-less.
+    fn fresh_opts(tag: &str) -> Option<(PgOpts, SchemaJanitor)> {
+        let dsn = harness_dsn()?;
+        let schema = fresh_schema(tag);
+        Some((
+            PgOpts {
+                schema: schema.clone(),
+                ..Default::default()
+            },
+            SchemaJanitor::new(&dsn, &schema),
+        ))
     }
 
     fn key_for(content: &[u8]) -> Key {
@@ -4969,7 +4975,7 @@ mod postgres_construction_tests {
     /// fresh schema).
     #[tokio::test]
     async fn dual_tier_postgres_local_constructs_and_routes() {
-        let Some(opts) = fresh_opts("dual-tier") else {
+        let Some((opts, _janitor)) = fresh_opts("dual-tier") else {
             skip_note();
             return;
         };
@@ -5013,7 +5019,7 @@ mod postgres_construction_tests {
     /// overflow, and exactly-at-threshold puts land.
     #[tokio::test]
     async fn kv_only_postgres_constructs_and_rejects_over_threshold_puts() {
-        let Some(opts) = fresh_opts("kv-only") else {
+        let Some((opts, _janitor)) = fresh_opts("kv-only") else {
             skip_note();
             return;
         };
@@ -5087,7 +5093,7 @@ mod postgres_construction_tests {
             "the shipped default is 5 minutes (ADR-008 §2)"
         );
 
-        let Some(opts) = fresh_opts("delay-default") else {
+        let Some((opts, _janitor)) = fresh_opts("delay-default") else {
             skip_note();
             return;
         };
@@ -5104,7 +5110,7 @@ mod postgres_construction_tests {
         .expect("constructs at the shipped default");
         assert_eq!(store.window_delay(), Duration::from_secs(300));
 
-        let Some(opts) = fresh_opts("delay-tuned") else {
+        let Some((opts, _janitor)) = fresh_opts("delay-tuned") else {
             return;
         };
         let store = Store::kv_only_postgres(&dsn, opts, 64, Duration::from_secs(90))
@@ -5112,7 +5118,7 @@ mod postgres_construction_tests {
             .expect("constructs at a custom tuning");
         assert_eq!(store.window_delay(), Duration::from_secs(90));
 
-        let Some(opts) = fresh_opts("delay-zero") else {
+        let Some((opts, _janitor)) = fresh_opts("delay-zero") else {
             return;
         };
         let root = temp_root("delay-zero");
@@ -5121,7 +5127,7 @@ mod postgres_construction_tests {
             .expect("the degenerate zero tuning is legal");
         assert_eq!(store.window_delay(), Duration::ZERO);
 
-        let Some(opts) = fresh_opts("delay-long") else {
+        let Some((opts, _janitor)) = fresh_opts("delay-long") else {
             return;
         };
         Store::kv_only_postgres(&dsn, opts, 64, Duration::from_secs(6000))
@@ -5136,7 +5142,7 @@ mod postgres_construction_tests {
     /// of both tiers.
     #[tokio::test]
     async fn dual_tier_postgres_shape_round_trips_end_to_end_through_both_tiers() {
-        let Some(opts) = fresh_opts("round-trip") else {
+        let Some((opts, _janitor)) = fresh_opts("round-trip") else {
             skip_note();
             return;
         };
@@ -5237,7 +5243,7 @@ mod postgres_construction_tests {
     async fn sweep_runs_the_durable_form_on_the_postgres_host() {
         let source = Arc::new(SetSource::default());
 
-        let Some(opts) = fresh_opts("durable-sweep") else {
+        let Some((opts, _janitor)) = fresh_opts("durable-sweep") else {
             drop(source);
             skip_note();
             return;