From f195759989f2720cbf5b7d71a642aa7fd6c42284 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 07:06:03 +0000 Subject: [PATCH] fix(postgres): the wave-4 review gate's inline findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 193 ++++++++++++++++++++++++++++++++++++++++++------ src/store.rs | 38 ++++++---- 2 files changed, 194 insertions(+), 37 deletions(-) diff --git a/src/postgres.rs b/src/postgres.rs index 4478d04..d27569d 100644 --- a/src/postgres.rs +++ b/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 { - 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 { + /// 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_with(tag, PgOpts::default()).await } - async fn harness_with(tag: &str, opts: PgOpts) -> Option { + async fn harness_with(tag: &str, opts: PgOpts) -> Option { 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 { + /// never bootstrapped (the sqlite `new_tableless` pattern). The + /// guard drops the (table-less) schema at test end. + async fn harness_schemaless(tag: &str) -> Option { 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 = 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 { + async fn harness_gc(tag: &str) -> Option { 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(); + } + } } diff --git a/src/store.rs b/src/store.rs index a191b47..6be34c7 100644 --- a/src/store.rs +++ b/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 { - 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;