feat(pg-lo): the large-tier engine — LOs + companion table behind the Backend trait (wave 5)

Implement the pg-lo engine (ADR-009, POC #7 C1-C7) in src/pg_lo.rs
behind feature pg-lo (default-off), plus the shared-helper home the
task's design points resolve into src/pg_common.rs.

- Engine: seven Backend methods + put_staged; the lo_entries companion
  table is the contract authority (list/size/has/CAS table-level;
  pg_largeobject read only by the orphan sweep's candidate enumeration);
  one-tx puts with 512 KiB lowrite statements (LOWRITE_CHUNK, the C2
  knee), the descriptorless lo_get window cursor (READ_CHUNK; EOF
  clamping + the unexpected-EOF clean failure), the held-tx staging
  cursor aggregating 64 KiB windows, one-tx deletes, the complete
  single-snapshot list, the 42P01 read-only virgin-store split, and the
  crate-internal reap_orphans (scope-noted, planted-orphan contract
  test). Posture deltas in the module doc (catalog space never
  returned / VACUUM FULL runbook line / autovacuum inherited / the
  cached-get capacity facts). Nothing of EngineFleetState; the window
  family remains engine-pg-lo-window.
- Shared helpers (pg_err, 42P01, quote_identifier, size clamp, i64-nanos
  encoding, build_pool, PgOpts, test janitors) move to src/pg_common.rs
  gated cfg(any(postgres, pg-lo)); PgOpts's public re-export widens
  (definition moved; crate::postgres::PgOpts path held by re-export);
  src/postgres.rs's changes are mechanical swaps.
- LO isolation (C7 lesson): per-test dedicated databases + DatabaseJanitor
  (WITH (FORCE)); zero LO/schema/database residue on the shared harness
  db (pre-existing public.lo_entries exploration residue found and
  removed).
- 18 in-module tests, env-driven (ALKBLOBS_PG_*), skipping cleanly
  server-less; all green against the pglo-poc harness server.

Verified: cargo test (69/19/3), --features pg-lo live (87/19/3, 18/18
pg-lo), --features postgres live (103/36/3), --all-features (176/71/4);
clippy -D warnings green for default, pg-lo, postgres, all-features,
--no-default-features --features "pg-lo,mem" and "postgres,pg-lo";
cargo fmt --check; cargo doc --no-deps.

src/lib.rs

@@ -27,11 +27,14 @@ pub mod pg_lo;
 #[cfg(feature = "ops")]
 pub mod ops;
 
+#[cfg(any(feature = "postgres", feature = "pg-lo"))]
+mod pg_common;
+
 #[cfg(feature = "postgres")]
 pub mod postgres;
 
-#[cfg(feature = "postgres")]
-pub use postgres::PgOpts;
+#[cfg(any(feature = "postgres", feature = "pg-lo"))]
+pub use pg_common::PgOpts;
 
 #[cfg(feature = "sqlite")]
 pub mod sqlite;

src/pg_common.rs

@@ -0,0 +1,321 @@
+//! The crate-internal pg-common module — the shared-helper design point
+//! resolved (engine-pg-lo): the pieces both postgres-attached engines
+//! (`postgres` kv, `pg-lo` large) share, gated
+//! `cfg(any(feature = "postgres", feature = "pg-lo"))` so each engine's
+//! pure build (ADR-013 §4: `pg-lo` does not imply `postgres`; the
+//! pg-lo-pure build) compiles without duplication. Held here: the
+//! deadpool pool construction (the shipped driver posture),
+//! [`quote_identifier`] (the quoted-identifier boundary), the SQLSTATE
+//! 42P01 (undefined table) virgin-store discriminant, the error mapping,
+//! the metadata-size clamp, the i64-nanos timestamp encoding, and
+//! [`PgOpts`] — the shared construction-opts type whose public
+//! re-export gate is widened to `any(postgres, pg-lo)` with the
+//! definition's move here (the type is unchanged; `alkblobs::PgOpts`
+//! now exists in the pg-lo-pure build too).
+//!
+//! The error mapping keeps the `postgres:` prefix for both engines: each
+//! attaches to a PostgreSQL server, so family attribution reads the same
+//! from either engine's failures.
+
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
+use tokio_postgres::NoTls;
+
+use crate::Error;
+
+/// The error mapping's base form (the postgres engine's): a failure
+/// surfaces as [`Error::Io`] with the engine family prefix.
+pub(crate) fn err(e: impl std::fmt::Display) -> Error {
+    Error::Io(format!("postgres: {e}"))
+}
+
+/// The engine's error mapping for statement bodies: a SQL-level failure
+/// surfaces with its server-side message and SQLSTATE (the `DbError`
+/// detail `Display` drops), anything else through [`err`].
+pub(crate) fn pg_err(e: tokio_postgres::Error) -> Error {
+    match e.as_db_error() {
+        Some(db) => Error::Io(format!("postgres: {db}")),
+        None => err(e),
+    }
+}
+
+/// The virgin-store tolerance's discriminant (POC #1 finding 2): a
+/// statement naming a table that does not exist maps on the read paths
+/// to absent/empty (and delete to Missing), never an error. The stable
+/// surface PostgreSQL carries is the SQLSTATE class+code (42P01,
+/// undefined table); a missing *schema* reads identically.
+pub(crate) fn is_undefined_table(e: &tokio_postgres::Error) -> bool {
+    e.code() == Some(&tokio_postgres::error::SqlState::UNDEFINED_TABLE)
+}
+
+/// Quote a SQL identifier (the Postgres rule: double the embedded `"`).
+///
+/// The engines compose schema-qualified SQL from consumer-supplied
+/// schema names; this is the quoted-identifier boundary — an identifier
+/// is never interpolated into SQL text unquoted (the alkstore
+/// `quote_identifier` pattern).
+pub(crate) fn quote_identifier(name: &str) -> String {
+    let mut quoted = String::with_capacity(name.len() + 2);
+    quoted.push('"');
+    for c in name.chars() {
+        if c == '"' {
+            quoted.push('"');
+        }
+        quoted.push(c);
+    }
+    quoted.push('"');
+    quoted
+}
+
+/// A schema-qualified table reference: `"schema"."table"`.
+pub(crate) fn qualified_table(schema: &str, table: &str) -> String {
+    format!("{}.{}", quote_identifier(schema), quote_identifier(table))
+}
+
+/// The default engine-owned schema name (the recorded design point's
+/// default). Both postgres-attached engines default to it — the
+/// consolidation shape (ADR-012 §4's same-pg-instance rule) holds both
+/// engines' tables in one engine-owned schema without name collision
+/// (`blobs`, `lo_entries`, and the GC tables are distinct names).
+pub(crate) const DEFAULT_SCHEMA: &str = "alkblobs";
+
+/// The shipped default pool size ([`PgOpts::max_size`]) — the alkstore
+/// `DEFAULT_MAX_SIZE = 8` precedent.
+pub(crate) const DEFAULT_MAX_SIZE: usize = 8;
+
+/// The `size` metadata's integer, clamped to the domain the trait probes
+/// in.
+pub(crate) fn size_u64(raw: i64) -> u64 {
+    raw.max(0) as u64
+}
+
+/// The crate-wide timestamp encoding (the recorded design point): signed
+/// i64 nanoseconds from the Unix epoch — sub-second precision for the
+/// test tunings, negative for pre-epoch instants, i64 across the full
+/// range a `BIGINT` stores (±292 years around the epoch).
+pub(crate) fn time_to_nanos(t: SystemTime) -> i64 {
+    match t.duration_since(UNIX_EPOCH) {
+        Ok(d) => d.as_nanos() as i64,
+        Err(e) => -(e.duration().as_nanos() as i64),
+    }
+}
+
+/// Build the deadpool pool with the shipped posture: per-session
+/// `synchronous_commit` merged into every connection's connect options
+/// (appended to the consumer's own `options=` DSN values — postgres
+/// parses repeated `-c` flags), `RecyclingMethod::Fast`, the configured
+/// `max_size`. `max_size = 0` fails at entry, before any round trip.
+pub(crate) fn build_pool(
+    config: &str,
+    max_size: usize,
+    synchronous_commit: bool,
+) -> Result<Pool, Error> {
+    if max_size == 0 {
+        return Err(Error::Io(format!(
+            "postgres: max_size must be positive, got {max_size}"
+        )));
+    }
+    let mut cfg: tokio_postgres::Config = config.parse().map_err(err)?;
+    let engine_opt = format!(
+        "-c synchronous_commit={}",
+        if synchronous_commit { "on" } else { "off" }
+    );
+    let merged = match cfg.get_options() {
+        Some(existing) => format!("{existing} {engine_opt}"),
+        None => engine_opt,
+    };
+    cfg.options(merged);
+    let manager = Manager::from_config(
+        cfg,
+        NoTls,
+        ManagerConfig {
+            recycling_method: RecyclingMethod::Fast,
+        },
+    );
+    Pool::builder(manager)
+        .max_size(max_size)
+        .build()
+        .map_err(err)
+}
+
+/// Construction options shared by the crate's postgres-attached engines
+/// (`postgres` kv, `pg-lo` large) — the PgOpts design point resolved:
+/// one type, its public re-export gate widened to
+/// `any(postgres, pg-lo)` alongside the definition's move to this
+/// module (ADR-013 §4's pg-lo-pure build carries the same opts surface).
+/// Durability knobs, pool sizing, and the engine-owned schema name are
+/// engine-configuration concerns — they ride the store constructors'
+/// parameters, never invented defaults (the alkstore `PgOpts`
+/// precedent; `open(config, PgOpts)` adapted to the named-constructor
+/// shape — the constructors are [`crate::Store::dual_tier_postgres_local`]
+/// and [`crate::Store::kv_only_postgres`]).
+///
+/// The connection settings themselves do not ride this struct — they
+/// are the constructor's separate connection-settings argument (the
+/// tokio-postgres [`tokio_postgres::Config`]-parseable form); the
+/// engine never invents defaults for them. TLS: the engines hardwire
+/// `NoTls` on every connection path regardless of the config's
+/// sslmode — v1 TLS is a post-v1 deployment concern, mirroring the
+/// alkstore posture.
+#[derive(Debug, Clone)]
+pub struct PgOpts {
+    /// The engine-owned PostgreSQL schema holding the engine's tables.
+    /// Both engines default to [`DEFAULT_SCHEMA`] (the consolidation
+    /// shape holds both engines' tables in one schema); an empty string
+    /// selects the default. The name is a quoted-identifier boundary
+    /// ([`quote_identifier`]) — reserved-word and hostile names are
+    /// safe.
+    pub schema: String,
+    /// Upper bound on pooled connections (deadpool's `max_size`).
+    /// Default [`DEFAULT_MAX_SIZE`] = 8. A value of `0` fails
+    /// construction immediately (validated at entry, no round trip —
+    /// deadpool cannot grant checkouts from an empty pool; the
+    /// alkstore precedent).
+    pub max_size: usize,
+    /// The per-session `synchronous_commit` durability knob, wired as a
+    /// connect option (`-c synchronous_commit=…`) on every pooled
+    /// connection (POC-verified per-session SET mechanics; the
+    /// consumer's own `options=` DSN values are preserved alongside —
+    /// postgres parses repeated `-c` flags). Default `true` — the ship
+    /// config; `false` is the documented sqlite-NORMAL-alike latency
+    /// trade (B5 ~2-3×). Both engines carry the knob; the pg-lo POC ran
+    /// its server with `synchronous_commit=off` to measure LO shape
+    /// without disk fsync dominating (POC #7), and the shipped knob
+    /// posture is the kv engine's parity default `on`.
+    pub synchronous_commit: bool,
+}
+
+impl Default for PgOpts {
+    fn default() -> Self {
+        Self {
+            schema: DEFAULT_SCHEMA.to_owned(),
+            max_size: DEFAULT_MAX_SIZE,
+            synchronous_commit: true,
+        }
+    }
+}
+
+/// Test-support helpers shared with the crate's postgres-attached test
+/// modules (`src/postgres.rs`'s and `src/pg_lo.rs`'s test suites,
+/// `src/store.rs`'s construction tests) — crate-internal and compiled
+/// out of every non-test build. Each janitor is gated by the feature
+/// whose tests consume it.
+#[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.
+    #[cfg(feature = "postgres")]
+    pub(crate) struct SchemaJanitor {
+        dsn: String,
+        schema: String,
+    }
+
+    #[cfg(feature = "postgres")]
+    impl SchemaJanitor {
+        pub(crate) fn new(dsn: &str, schema: &str) -> SchemaJanitor {
+            SchemaJanitor {
+                dsn: dsn.to_owned(),
+                schema: schema.to_owned(),
+            }
+        }
+    }
+
+    #[cfg(feature = "postgres")]
+    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();
+        }
+    }
+
+    /// Drops a test's dedicated database when the test ends — the
+    /// pg-lo engine tests' janitor. Large Objects are **database-level**
+    /// (the C7 lesson, recorded design point): the schema-per-test
+    /// convention does not isolate LOs across tests sharing a database,
+    /// and a `DROP SCHEMA` leaves the LO's pages in `pg_largeobject` —
+    /// so the pg-lo tests isolate per dedicated database (the POC's
+    /// proven `open_fresh` shape) and this guard removes it, zero LO
+    /// residue on the shared harness database by construction.
+    /// Same Drop mechanics as [`SchemaJanitor`]; `WITH (FORCE)` reaps
+    /// any connection the pool has not yet drained (postgres ≥ 13).
+    #[cfg(feature = "pg-lo")]
+    pub(crate) struct DatabaseJanitor {
+        admin_dsn: String,
+        db: String,
+    }
+
+    #[cfg(feature = "pg-lo")]
+    impl DatabaseJanitor {
+        /// `admin_dsn` must name a database to connect *through* (the
+        /// `postgres` maintenance database); `db` is the test's
+        /// dedicated database to drop.
+        pub(crate) fn new(admin_dsn: &str, db: &str) -> DatabaseJanitor {
+            DatabaseJanitor {
+                admin_dsn: admin_dsn.to_owned(),
+                db: db.to_owned(),
+            }
+        }
+    }
+
+    #[cfg(feature = "pg-lo")]
+    impl Drop for DatabaseJanitor {
+        fn drop(&mut self) {
+            let admin_dsn = self.admin_dsn.clone();
+            let db = self.db.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(&admin_dsn, NoTls).await
+                    else {
+                        return;
+                    };
+                    tokio::spawn(conn);
+                    let sql = format!(
+                        "DROP DATABASE IF EXISTS {} WITH (FORCE)",
+                        quote_identifier(&db)
+                    );
+                    let _ = client.execute(sql.as_str(), &[]).await;
+                });
+            })
+            .join();
+        }
+    }
+}

src/postgres.rs

@@ -196,121 +196,28 @@
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
-use deadpool_postgres::{Manager, ManagerConfig, Object, Pool, RecyclingMethod};
+use deadpool_postgres::{Object, Pool};
 use parking_lot::Mutex;
-use tokio_postgres::NoTls;
 
 use crate::backend::{clamped, Backend, ReadCursor, StagingCursor};
 use crate::gc::{EngineFleetState, EngineWindowState, PinRow, WindowCandidate};
+/// The shared opts type's re-export home for the `crate::postgres::PgOpts`
+/// path (its definition lives in [`crate::pg_common`]; the crate-root
+/// re-export is `any(postgres, pg-lo)` — the widened gate).
+pub use crate::pg_common::PgOpts;
+use crate::pg_common::{
+    build_pool, err, is_undefined_table, pg_err, qualified_table, quote_identifier, size_u64,
+    time_to_nanos, DEFAULT_SCHEMA,
+};
 use crate::{Error, PinState};
 
-fn err(e: impl std::fmt::Display) -> Error {
-    Error::Io(format!("postgres: {e}"))
-}
-
-/// The engine's error mapping for the clause bodies: a SQL-level
-/// failure surfaces with its server-side message and SQLSTATE (the
-/// `DbError` detail `Display` drops), anything else through [`err`].
-fn pg_err(e: tokio_postgres::Error) -> Error {
-    match e.as_db_error() {
-        Some(db) => Error::Io(format!("postgres: {db}")),
-        None => err(e),
-    }
-}
-
-/// The virgin-store tolerance's discriminant (POC #1 finding 2): a
-/// statement naming a table that does not exist maps on the read paths
-/// to absent/empty (and delete to Missing), never an error. The stable
-/// surface PostgreSQL carries is the SQLSTATE class+code (42P01,
-/// undefined table); a missing *schema* reads identically.
-fn is_undefined_table(e: &tokio_postgres::Error) -> bool {
-    e.code() == Some(&tokio_postgres::error::SqlState::UNDEFINED_TABLE)
-}
-
-/// The `size` column's integer, clamped to the domain the trait probes in.
-fn size_u64(raw: i64) -> u64 {
-    raw.max(0) as u64
-}
-
-/// The default engine-owned schema name (the recorded design point's
-/// default).
-pub(crate) const DEFAULT_SCHEMA: &str = "alkblobs";
-
-/// The shipped default pool size ([`PgOpts::max_size`]) — the alkstore
-/// `DEFAULT_MAX_SIZE = 8` precedent.
-pub(crate) const DEFAULT_MAX_SIZE: usize = 8;
-
-/// Quote a SQL identifier (the Postgres rule: double the embedded `"`).
-///
-/// The engine composes schema-qualified SQL from a consumer-supplied
-/// schema name; this is the quoted-identifier boundary — an identifier
-/// is never interpolated into SQL text unquoted (the alkstore
-/// `quote_identifier` pattern).
-pub(crate) fn quote_identifier(name: &str) -> String {
-    let mut quoted = String::with_capacity(name.len() + 2);
-    quoted.push('"');
-    for c in name.chars() {
-        if c == '"' {
-            quoted.push('"');
-        }
-        quoted.push(c);
-    }
-    quoted.push('"');
-    quoted
-}
-
-/// A schema-qualified table reference: `"schema"."table"`.
-pub(crate) fn qualified_table(schema: &str, table: &str) -> String {
-    format!("{}.{}", quote_identifier(schema), quote_identifier(table))
-}
-
-/// Construction options for the postgres engine. Durability knobs,
-/// pool sizing, and the engine-owned schema name are
-/// engine-configuration concerns — they ride the store constructors'
-/// parameters, never invented defaults (the alkstore `PgOpts`
-/// precedent; `open(config, PgOpts)` adapted to the named-constructor
-/// shape — the constructors are [`Store::dual_tier_postgres_local`] and
-/// [`Store::kv_only_postgres`]).
-///
-/// The connection settings themselves do not ride this struct — they
-/// are the constructor's separate connection-settings argument (the
-/// tokio-postgres [`tokio_postgres::Config`]-parseable form); the
-/// engine never invents defaults for them. TLS: the engine hardwires
-/// `NoTls` on every connection path regardless of the config's
-/// sslmode — v1 TLS is a post-v1 deployment concern, mirroring the
-/// alkstore posture.
-#[derive(Debug, Clone)]
-pub struct PgOpts {
-    /// The engine-owned PostgreSQL schema holding the pool table.
-    /// Default [`DEFAULT_SCHEMA`]; an empty string selects the default.
-    /// The name is a quoted-identifier boundary ([`quote_identifier`])
-    /// — reserved-word and hostile names are safe.
-    pub schema: String,
-    /// Upper bound on pooled connections (deadpool's `max_size`).
-    /// Default [`DEFAULT_MAX_SIZE`] = 8. A value of `0` fails
-    /// construction immediately (validated at entry, no round trip —
-    /// deadpool cannot grant checkouts from an empty pool; the
-    /// alkstore precedent).
-    pub max_size: usize,
-    /// The per-session `synchronous_commit` durability knob, wired as a
-    /// connect option (`-c synchronous_commit=…`) on every pooled
-    /// connection (POC-verified per-session SET mechanics; the
-    /// consumer's own `options=` DSN values are preserved alongside —
-    /// postgres parses repeated `-c` flags). Default `true` — the ship
-    /// config; `false` is the documented sqlite-NORMAL-alike latency
-    /// trade (B5 ~2-3×).
-    pub synchronous_commit: bool,
-}
-
-impl Default for PgOpts {
-    fn default() -> Self {
-        Self {
-            schema: DEFAULT_SCHEMA.to_owned(),
-            max_size: DEFAULT_MAX_SIZE,
-            synchronous_commit: true,
-        }
-    }
-}
+/// The test-support helpers' shared home — the SchemaJanitor lives in
+/// `src/pg_common.rs`'s `#[cfg(test)] test_support` (gated
+/// `cfg(test)` + `any(postgres, pg-lo)`, the shared-helper design
+/// point's home); re-exported here so the sibling test suites'
+/// `crate::postgres::test_support` path holds.
+#[cfg(test)]
+pub(crate) use crate::pg_common::test_support;
 
 /// The `postgres` engine — a deadpool pool over tokio-postgres plus the
 /// quoted engine-owned table reference. Cloning shares the pool (the
@@ -408,7 +315,7 @@ impl PostgresBackend {
     /// `new_tableless` pattern.
     #[cfg(test)]
     async fn new_schemaless(config: &str, schema: &str) -> Result<Self, Error> {
-        let pool = build_pool(config, DEFAULT_MAX_SIZE, true)?;
+        let pool = build_pool(config, crate::pg_common::DEFAULT_MAX_SIZE, true)?;
         let schema = schema.to_owned();
         let pins_table = qualified_table(&schema, "gc_pins");
         let window_table = qualified_table(&schema, "gc_window");
@@ -488,40 +395,6 @@ impl PostgresBackend {
     }
 }
 
-/// Build the deadpool pool with the shipped posture: per-session
-/// `synchronous_commit` merged into every connection's connect options
-/// (appended to the consumer's own `options=` DSN values — postgres
-/// parses repeated `-c` flags), `RecyclingMethod::Fast`, the configured
-/// `max_size`. `max_size = 0` fails at entry, before any round trip.
-fn build_pool(config: &str, max_size: usize, synchronous_commit: bool) -> Result<Pool, Error> {
-    if max_size == 0 {
-        return Err(Error::Io(format!(
-            "postgres: max_size must be positive, got {max_size}"
-        )));
-    }
-    let mut cfg: tokio_postgres::Config = config.parse().map_err(err)?;
-    let engine_opt = format!(
-        "-c synchronous_commit={}",
-        if synchronous_commit { "on" } else { "off" }
-    );
-    let merged = match cfg.get_options() {
-        Some(existing) => format!("{existing} {engine_opt}"),
-        None => engine_opt,
-    };
-    cfg.options(merged);
-    let manager = Manager::from_config(
-        cfg,
-        NoTls,
-        ManagerConfig {
-            recycling_method: RecyclingMethod::Fast,
-        },
-    );
-    Pool::builder(manager)
-        .max_size(max_size)
-        .build()
-        .map_err(err)
-}
-
 /// The engine-owned bootstrap DDL — idempotent, no migration machinery.
 /// The B5 row shape under the measured `fillfactor = 90`; column names
 /// quote-follow the alkstore hazard-list discipline (`"key"` quoted;
@@ -549,19 +422,10 @@ pub(crate) const DEFAULT_PIN_TTL_SECS: u64 = 60;
 /// source): every renewed row's expiry moves to `now` + this.
 const DEFAULT_PIN_TTL: Duration = Duration::from_secs(DEFAULT_PIN_TTL_SECS);
 
-/// The window row family's timestamp encoding (the recorded design
-/// point): signed nanoseconds since the Unix epoch — sub-second
-/// precision for the test tunings, negative for pre-epoch instants,
-/// i64 across the full range a `BIGINT` stores (±292 years around the
-/// epoch). The sqlite engine's helpers are this one's reference —
-/// per-engine bodies, one encoding.
-fn time_to_nanos(t: SystemTime) -> i64 {
-    match t.duration_since(UNIX_EPOCH) {
-        Ok(d) => d.as_nanos() as i64,
-        Err(e) => -(e.duration().as_nanos() as i64),
-    }
-}
-
+/// The gc-window/pin rows' timestamp decoding — the test tunings'
+/// fixture helper (the sqlite engine's helpers are the reference —
+/// per-engine bodies, one encoding: [`crate::pg_common::time_to_nanos`]
+/// is the shared forward direction).
 fn nanos_to_time(n: i64) -> SystemTime {
     if n >= 0 {
         UNIX_EPOCH + std::time::Duration::from_nanos(n as u64)
@@ -1191,6 +1055,7 @@ mod tests {
     use crate::gc::{EngineFleetState, EngineWindowState};
     use std::ops::Deref;
     use std::sync::atomic::{AtomicU64, Ordering};
+    use tokio_postgres::NoTls;
 
     const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";
 
@@ -2794,64 +2659,3 @@ mod tests {
         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();
-        }
-    }
-}

tasks/engine-pg-lo.md

@@ -1,7 +1,7 @@
 ---
 id: engine-pg-lo
 name: pg-lo large engine — Large Objects behind the Backend trait
-status: pending
+status: completed
 depends_on: []
 scope: moderate
 risk: medium
@@ -230,8 +230,114 @@ engine and its in-module unit tests only.
 
 ## Notes
 
-> To be filled by implementation agent
+> Implemented in `src/pg_lo.rs` (the engine, its cursors, the orphan
+> sweep, and 18 in-module tests), `src/pg_common.rs` (new — the
+> shared-helper home), and `src/{lib,postgres}.rs` (the shared-helper
+> move + gate widening; `src/postgres.rs`'s changes are mechanical
+> import swaps plus path-stable re-exports). Recorded design points,
+> as resolved:
+>
+> - **Shared-helper shape** — `src/pg_common.rs`, gated
+>   `cfg(any(feature = "postgres", feature = "pg-lo"))`, one module per
+>   file: `quote_identifier`/`qualified_table`, the 42P01 discriminant,
+>   `err`/`pg_err` (the mapping keeps the `postgres:` prefix — both
+>   engines attach to the same server family), `size_u64`,
+>   `time_to_nanos` (the i64-nanos encoding), `build_pool`, [`PgOpts`],
+>   and the `#[cfg(test)] test_support` janitors (`SchemaJanitor`
+>   gated `#[cfg(feature = "postgres")]`, new `DatabaseJanitor` gated
+>   `#[cfg(feature = "pg-lo")]`; `crate::postgres::test_support` remains
+>   a path-stable re-export, so `src/store.rs`'s import needed no edit).
+> - **PgOpts** — one shared type; its definition moved to `pg_common`,
+>   the public re-export widened to `any(postgres, pg-lo)` (the type is
+>   unchanged; `alkblobs::postgres::PgOpts` still resolves through the
+>   path-stable re-export). The `synchronous_commit` knob **carries**
+>   (default `on`; the POC's server-side `off` was a measurement
+>   posture, not the ship config — recorded in the module doc).
+> - **LO isolation (the C7 lesson resolved)** — Large Objects are
+>   database-level; schema-per-test does not isolate them. The tests
+>   run on **per-test dedicated databases** (the POC's proven
+>   `open_fresh` shape): unique `abtlo_{tag}_{pid}_{seq}_{nanos}`
+>   names, `CREATE DATABASE` through the `postgres` maintenance db, a
+>   `DatabaseJanitor` drop-guard (`DROP DATABASE … WITH (FORCE)`) after
+>   the backend's pool. Zero LO residue on the shared `blobs` db is
+>   structural (no pg-lo test ever opens a pool against it) and gate-
+>   tested (`shared_harness_db_holds_zero_lo_residue`).
+> - **`committed_at`** — the crate-wide i64-nanos encoding as `bigint`,
+>   engine-stamped at insert (not a server-default column); the crate's
+>   one timestamp encoding rather than the POC's `timestamptz default
+>   now()`.
+> - **Staged reads** — verified live: `lo_get` inside the writing tx
+>   sees its own uncommitted pages (read-your-own-writes), so the
+>   verification pass reads the staged artifact through descriptorless
+>   windows + the unflushed in-memory tail; no descriptor reads, no
+>   `lo_lseek` anywhere (the C7 64-bit discipline: offsets ride bigint
+>   only, `lowrite` is implicitly sequential).
+> - **Stage hygiene mechanics** — the staging cursor takes its
+>   checkout in `commit` (`publish` over the taken `Object`), so every
+>   >   failure path converges on one ROLLBACK point; the dedup path
+>   (CAS hit) ROLLBACKs and returns `Ok` (the staged LO is the residue
+>   removed); `Drop` hands an abandoned cursor's open tx to a spawned
+>   ROLLBACK (never returns a dirty checkout to the pool — deadpool's
+>   `Fast` recycle pitfall), degrading to a one-shot thread when no
+>   runtime is live. `#[cfg(test)] fail_commit` knob mirrors
+>   mem/sqlite/postgres/local.
+> - **Harness-server residue found and cleaned** — the shared `blobs`
+>   db carried pre-existing exploration residue (`public.lo_entries`
+>   + 21 LOs, 9 472 catalog pages) from pre-task pg-lo probing;
+>   unlinked + dropped (count now 0) so the
+>   `pg_largeobject`-probe gate measures this suite's own runs. The
+>   `public.blobs` kv table was left untouched (kv-engine content, out
+>   of scope).
+> - **The literal `--no-default-features --features pg-lo` config
+>   cannot compile the whole crate** — `store.rs`'s `KvEngine`/`KvCursor`
+>   enums are empty there (E0004), identically on the pre-task tree
+>   (verified by stash). Every constructor mode pairs a kv engine
+>   (ADR-011 §4's table), so a kv-less store is not a shippable shape;
+>   the clean configs are `--no-default-features --features "pg-lo,mem"`
+>   and ADR-013 §4's pg-pure build
+>   `--no-default-features --features "postgres,pg-lo"` — both clippy
+>   clean. In the literal config the *shared pieces* compile (the 10
+>   store.rs E0004s are the crate's only errors; nothing in
+>   pg_common/pg_lo errors). Flagged for `store-pg-lo-construction` to
+>   keep in view.
 
 ## Summary
 
-> To be filled on completion
+> Completed 2026-10-12 (wave 5). `src/pg_lo.rs` implements the
+> seven-method `Backend` + `put_staged` behind feature `pg-lo`
+> (default-off): LOs hold the bytes, the `lo_entries` companion table is
+> the contract authority (list/size/has/CAS all table-level; `pg_largeobject`
+> read by the engine only inside the orphan sweep's candidate
+> enumeration), one-tx puts (lo_create → 512 KiB chunked `lowrite`
+> (`LOWRITE_CHUNK`, the C2 knee) → row insert → COMMIT), the
+> descriptorless `lo_get(oid, off, len)` window cursor
+> (`READ_CHUNK`; `len()` metadata-only from the companion row; EOF
+> clamping per the cursor contract plus the unexpected-EOF clean
+> failure for a body short of its claimed length), the held-tx staging
+> cursor with aggregation of the store core's 64 KiB windows into
+> 512 KiB statements and both hygiene forms exact-count-tested, one-tx
+> deletes with the Missing discipline, the complete single-snapshot
+> `list`, idempotent schema+table bootstrap at the constructor with the
+> quoted-identifier boundary (hostile names tested), the 42P01
+> read-only virgin-store split (schema-less form tested), the
+> crate-internal `reap_orphans` sweep (scope-noted; planted-orphan
+> contract test with byte-exact committed-content survival), and the
+> posture deltas in the module doc (catalog space never returned +
+> `VACUUM FULL` runbook line + rel-size monitoring; autovacuum
+> inherited; the 20–50× cached-get capacity facts). Shared helpers
+> (`pg_err`, 42P01, quoting, timestamps, pool, `PgOpts`) moved to the
+> new `src/pg_common.rs` gated `any(postgres, pg-lo)`; `PgOpts`'s
+> public re-export widened (shared opts decision; the knob carries).
+> 18 in-module tests, per-test dedicated databases (`DatabaseJanitor`),
+> env-driven (`ALKBLOBS_PG_*`), skipping cleanly server-less. Verified:
+> `cargo test` (69/19/3 ok), `cargo test --features pg-lo` live
+> (87/19/3 ok, 18/18 pg-lo), `cargo test --features postgres` live
+> (103/36/3 ok), `cargo test --all-features` (176/71/4 ok), clippy
+> `-D warnings` green (default, `--features pg-lo`, `--features
+> postgres`, `--all-features`, `--no-default-features --features
+> "pg-lo,mem"`, `--no-default-features --features "postgres,pg-lo"`),
+> `cargo fmt --check` green, `cargo doc --no-deps` clean, zero LO +
+> schema + database residue on the harness server after full runs.
+> Engine-state window family remains `engine-pg-lo-window`
+> (`EngineWindowState` untouched; nothing of `EngineFleetState` exists
+> on this engine); constructor wiring remains `store-pg-lo-construction`.