feat(gc): the pg-lo engine-state window family — the conditional window host (wave 5)

- impl EngineWindowState on PgLoBackend: ensure_gc_schema creates
  gc_window only (never gc_pins, never the embedder's liveness
  table), window_stage upserts idempotent per (key, deadline),
  window_take is the one atomic DELETE ... RETURNING step over the
  <= boundary with the 42P01 -> empty virgin-take split
- nothing of EngineFleetState exists on this engine (ADR-012 §5's
  pg-lo row: fleet pins live on the kv host, sweep lock no); the
  one-window-host dormancy posture is recorded in the impl doc
- five contract tests mirror the postgres/sqlite suites (schema-only
  + row shape, stage idempotence/restage, take-exact-set/never-twice,
  the <= boundary with nanosecond slack, both virgin shapes);
  env-driven, skipping cleanly server-less, green live against the
  harness server

Verification: cargo test (default + --features pg-lo, server-less
and live), clippy -D warnings (default, pg-lo, + the pure builds
"pg-lo,mem" and "postgres,pg-lo"), cargo fmt --check, cargo doc --no-deps

src/pg_lo.rs

@@ -175,10 +175,16 @@
 //! Nothing of [`crate::gc::EngineFleetState`] exists on this engine
 //! (ADR-012 §5's pg-lo row: pin clauses **no** — fleet pins live on
 //! the kv host; sweep lock **no**). The window-family seam clauses
-//! ([`crate::gc::EngineWindowState`]) are `engine-pg-lo-window`
-//! (separate task); the constructor wiring is
-//! `store-pg-lo-construction` — this module ships the engine and its
-//! in-module unit tests only.
+//! ([`crate::gc::EngineWindowState`]) are implemented here (the
+//! engine-pg-lo-window task): `ensure_gc_schema` creates `gc_window`
+//! only — never `gc_pins`, never the embedder's liveness table — and
+//! the window clauses are conditional per the one-window-host rule
+//! (dormant on a dual-SQL instance — kv=sqlite/postgres + large=pg-lo
+//! stages on the kv engine; active on the kv=mem + large=pg-lo shape).
+//! Whether the store core drives them at construction is the
+//! constructor wiring's business (`store-pg-lo-construction`); this
+//! module ships the clauses, their engine-level contract tests, and
+//! the rest of the engine (unit tests only).
 //!
 //! Tests need a live server: connection settings ride the environment
 //! (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB`); tests skip cleanly
@@ -193,6 +199,7 @@ use std::time::SystemTime;
 use deadpool_postgres::{Object, Pool};
 
 use crate::backend::{clamped, Backend, ReadCursor, StagingCursor};
+use crate::gc::{EngineWindowState, WindowCandidate};
 use crate::pg_common::{
     build_pool, err, is_undefined_table, pg_err, qualified_table, quote_identifier, size_u64,
     time_to_nanos, PgOpts, DEFAULT_SCHEMA,
@@ -223,6 +230,11 @@ pub(crate) struct PgLoBackend {
     /// construction: `"schema"."lo_entries"` — the quoted-identifier
     /// boundary every statement composes through.
     table: String,
+    /// The engine-owned GC window-table reference, quoted once at
+    /// construction: `"schema"."gc_window"` — the window family's row
+    /// home (ADR-012 §5's pg-lo row: the window clauses and the window
+    /// table only).
+    window_table: String,
     /// The engine-owned schema name (the test-only catalog probes').
     schema: String,
     /// Test-only commit-failure injection, standing in for the media
@@ -260,6 +272,7 @@ impl PgLoBackend {
         Ok(Self {
             pool,
             table: qualified_table(&schema, "lo_entries"),
+            window_table: qualified_table(&schema, "gc_window"),
             schema,
             #[cfg(test)]
             fail_commit: false,
@@ -276,6 +289,7 @@ impl PgLoBackend {
         Ok(Self {
             pool,
             table: qualified_table(schema, "lo_entries"),
+            window_table: qualified_table(schema, "gc_window"),
             schema: schema.to_owned(),
             #[cfg(test)]
             fail_commit: false,
@@ -412,6 +426,103 @@ CREATE TABLE IF NOT EXISTS {table} (
     )
 }
 
+/// The GC window schema's DDL — idempotent, the window-only row (ADR-012
+/// §5's pg-lo row): `gc_window` keyed by the candidate key, the
+/// timestamp `BIGINT` nanos. `gc_pins` is never named here (fleet pins
+/// live on the kv host) and neither is the embedder's liveness table
+/// (the schema-ownership rule — the sweep reads it and never owns it).
+fn gc_window_ddl(schema: &str) -> String {
+    let schema_q = quote_identifier(schema);
+    let window = qualified_table(schema, "gc_window");
+    format!(
+        "CREATE SCHEMA IF NOT EXISTS {schema_q};
+
+CREATE TABLE IF NOT EXISTS {window} (
+    key bytea PRIMARY KEY,
+    visible_after bigint NOT NULL
+);
+"
+    )
+}
+
+/// The window family (ADR-012 §5's pg-lo row): the window-table-only
+/// schema and the durable staged delete window. Same semantics as the
+/// sqlite and postgres reference implementations — different dialect;
+/// nothing of the fleet family exists on this engine.
+impl EngineWindowState for PgLoBackend {
+    /// Create `gc_window (key bytea PRIMARY KEY, visible_after bigint)`
+    /// — this engine's applicability row is the window table only;
+    /// `gc_pins` is never created here (ADR-012 §5: fleet pins live on
+    /// the kv host) and the embedder's liveness table is never created
+    /// in any case (ADR-008's schema-ownership rule). Idempotent, so
+    /// the twice-at-open call form is a no-op. Whether the store core
+    /// calls this at construction is the one-window-host rule's
+    /// business (`store-pg-lo-construction` wires the active host) —
+    /// the clause body exists and is contract-tested per applicability
+    /// regardless of dormancy.
+    async fn ensure_gc_schema(&self) -> Result<(), Error> {
+        let client = self.pool.get().await.map_err(err)?;
+        client
+            .batch_execute(&gc_window_ddl(&self.schema))
+            .await
+            .map_err(pg_err)
+    }
+
+    /// Stage candidates into the durable window — the upsert
+    /// discipline: a re-stage of the same key at the same deadline is
+    /// idempotent (one row, value unchanged), a re-stage at a new
+    /// deadline updates the row's `visible_after` rather than
+    /// duplicating it (the sqlite/postgres engines' recorded
+    /// semantics).
+    async fn window_stage(&self, candidates: &[WindowCandidate]) -> Result<(), Error> {
+        if candidates.is_empty() {
+            return Ok(());
+        }
+        let client = self.pool.get().await.map_err(err)?;
+        let sql = format!(
+            "INSERT INTO {} (key, visible_after) VALUES ($1, $2) \
+             ON CONFLICT (key) DO UPDATE SET visible_after = excluded.visible_after",
+            self.window_table
+        );
+        let stmt = client.prepare_cached(&sql).await.map_err(pg_err)?;
+        for candidate in candidates {
+            let visible_after = time_to_nanos(candidate.visible_after);
+            client
+                .execute(&stmt, &[&candidate.key, &visible_after])
+                .await
+                .map_err(pg_err)?;
+        }
+        Ok(())
+    }
+
+    /// Take overdue candidates (`visible_after <= now` — the `<=`
+    /// boundary is the seam doc's pin): one atomic step, the single
+    /// `DELETE … RETURNING` statement selecting and removing the rows
+    /// together, so a row is never handed out twice and an aborted body
+    /// leaves no partial take. A missing table maps to empty (the
+    /// postgres twin's virgin take; the recorded read-only split).
+    async fn window_take(&self, now: SystemTime) -> Result<Vec<Vec<u8>>, Error> {
+        let client = self.pool.get().await.map_err(err)?;
+        let horizon = time_to_nanos(now);
+        let sql = format!(
+            "DELETE FROM {} WHERE visible_after <= $1 RETURNING key",
+            self.window_table
+        );
+        let stmt = match client.prepare_cached(&sql).await {
+            Ok(stmt) => stmt,
+            Err(e) if is_undefined_table(&e) => return Ok(Vec::new()),
+            Err(e) => return Err(err(e)),
+        };
+        let rows = client.query(&stmt, &[&horizon]).await.map_err(pg_err)?;
+        let mut keys = rows
+            .iter()
+            .map(|row| row.try_get::<_, Vec<u8>>(0).map_err(pg_err))
+            .collect::<Result<Vec<Vec<u8>>, _>>()?;
+        keys.sort();
+        Ok(keys)
+    }
+}
+
 /// `lo_create` — a fresh LO oid inside the caller's tx.
 async fn lo_create(client: &Object) -> Result<u32, Error> {
     let stmt = client
@@ -1622,6 +1733,7 @@ mod tests {
         let failing = PgLoBackend {
             pool: backend.pool.clone(),
             table: backend.table.clone(),
+            window_table: backend.window_table.clone(),
             schema: backend.schema.clone(),
             #[cfg(test)]
             fail_commit: true,
@@ -1702,6 +1814,7 @@ mod tests {
             let failing = PgLoBackend {
                 pool: backend.pool.clone(),
                 table: backend.table.clone(),
+                window_table: backend.window_table.clone(),
                 schema: backend.schema.clone(),
                 #[cfg(test)]
                 fail_commit: true,
@@ -2026,6 +2139,287 @@ mod tests {
         );
     }
 
+    /// The window-family helpers (the postgres engine's suite is the
+    /// reference — same semantics, this dialect's harness). A
+    /// fixed-instant deadline family for the window tests (epoch-based
+    /// signed nanos, so pre-epoch coverage rides the same encoding):
+    /// -10s (past, pre-epoch), +1s (past), +2s (future), +1h (far
+    /// future).
+    fn nanos_to_time(n: i64) -> SystemTime {
+        if n >= 0 {
+            std::time::UNIX_EPOCH + std::time::Duration::from_nanos(n as u64)
+        } else {
+            std::time::UNIX_EPOCH - std::time::Duration::from_nanos((-n) as u64)
+        }
+    }
+
+    fn at(nanos: i64) -> SystemTime {
+        nanos_to_time(nanos)
+    }
+
+    /// A constructed backend with the GC schema ensured — the clause
+    /// tests' fixture (store construction wires the ensure call where
+    /// the one-window-host rule makes the window active; the engine
+    /// constructor itself never creates the GC table).
+    async fn harness_gc(tag: &str) -> Option<Harness> {
+        let backend = harness(tag).await?;
+        backend.ensure_gc_schema().await.unwrap();
+        Some(backend)
+    }
+
+    fn stages(key: &[u8], visible_after: SystemTime) -> WindowCandidate {
+        WindowCandidate {
+            key: key.to_vec(),
+            visible_after,
+        }
+    }
+
+    /// The gc_window rows as the engine stores them — the raw encoding,
+    /// read back through the engine-owned table itself.
+    async fn window_rows(backend: &PgLoBackend) -> Vec<(Vec<u8>, i64)> {
+        let client = backend.pool.get().await.expect("checkout");
+        let sql = format!(
+            "SELECT key, visible_after FROM {} ORDER BY key",
+            backend.window_table
+        );
+        let rows = client.query(sql.as_str(), &[]).await.expect("window rows");
+        let mut out: Vec<(Vec<u8>, i64)> = rows
+            .iter()
+            .map(|row| (row.get::<_, Vec<u8>>(0), row.get::<_, i64>(1)))
+            .collect();
+        out.sort();
+        out
+    }
+
+    /// The engine-owned schema's full table list — the schema-ownership
+    /// test's instrument (exactly the engine's tables, nothing more).
+    async fn schema_tables(backend: &PgLoBackend) -> Vec<String> {
+        let client = backend.pool.get().await.expect("checkout");
+        let rows = client
+            .query(
+                "SELECT table_name::text FROM information_schema.tables \
+                 WHERE table_schema = $1 ORDER BY table_name",
+                &[&backend.schema],
+            )
+            .await
+            .expect("schema tables");
+        rows.iter().map(|row| row.get::<_, String>(0)).collect()
+    }
+
+    /// Schema contract (the pg-lo row): `ensure_gc_schema` creates
+    /// `gc_window` only — `gc_pins` is never created by pg-lo (fleet
+    /// pins live on the kv host, ADR-012 §5) and the embedder's
+    /// liveness table is never created in any case (the
+    /// schema-ownership rule: exactly the engine's two tables exist) —
+    /// the call is idempotent, and the gc_window row shape is
+    /// bytea/bigint.
+    #[tokio::test]
+    async fn gc_schema_creates_window_table_only() {
+        let Some(backend) = harness("window-schema").await else {
+            return;
+        };
+        backend.ensure_gc_schema().await.unwrap();
+        backend.ensure_gc_schema().await.unwrap();
+        assert_eq!(
+            schema_tables(&backend).await,
+            vec!["gc_window", "lo_entries"],
+            "the engine's two tables exactly — no gc_pins, no embedder liveness table"
+        );
+        assert!(
+            !backend.table_exists("gc_pins").await.unwrap(),
+            "gc_pins is never created by pg-lo (fleet pins live on the kv host)"
+        );
+
+        let client = backend.pool.get().await.unwrap();
+        let columns: Vec<String> = client
+            .query(
+                "SELECT data_type::text FROM information_schema.columns \
+                 WHERE table_schema = $1 AND table_name = 'gc_window' \
+                 ORDER BY ordinal_position",
+                &[&backend.schema],
+            )
+            .await
+            .unwrap()
+            .iter()
+            .map(|row| row.get::<_, String>(0))
+            .collect();
+        assert_eq!(
+            columns,
+            ["bytea", "bigint"],
+            "the gc_window row shape: key/visible_after"
+        );
+    }
+
+    /// Staging contract: idempotent per `(key, deadline)` — the same
+    /// candidate staged twice leaves exactly one row — and a re-stage
+    /// at a new deadline updates the row (upsert on the key) rather
+    /// than duplicating it.
+    #[tokio::test]
+    async fn window_stage_is_idempotent_and_restage_updates_the_deadline() {
+        let Some(backend) = harness_gc("window-stage").await else {
+            return;
+        };
+        let key = b"\x01window-key".to_vec();
+        let deadline = at(1_000_000_000);
+
+        backend
+            .window_stage(&[stages(&key, deadline)])
+            .await
+            .unwrap();
+        backend
+            .window_stage(&[stages(&key, deadline)])
+            .await
+            .unwrap();
+        assert_eq!(
+            window_rows(&backend).await,
+            vec![(key.clone(), 1_000_000_000)],
+            "re-stage at the same deadline is one idempotent row"
+        );
+
+        backend
+            .window_stage(&[stages(&key, at(2_000_000_000))])
+            .await
+            .unwrap();
+        assert_eq!(
+            window_rows(&backend).await,
+            vec![(key.clone(), 2_000_000_000)],
+            "a re-stage at a new deadline updates the row"
+        );
+
+        let other = b"\x01window-other".to_vec();
+        backend
+            .window_stage(&[stages(&other, deadline)])
+            .await
+            .unwrap();
+        assert_eq!(
+            window_rows(&backend).await.len(),
+            2,
+            "distinct keys are distinct rows"
+        );
+    }
+
+    /// Take contract: exactly the overdue set (`visible_after <= now`),
+    /// removed atomically — a second take returns nothing — with future
+    /// rows left in place, and the key round-trips through the raw
+    /// encoding (signed nanos, so a pre-epoch deadline rides along).
+    #[tokio::test]
+    async fn window_take_returns_the_overdue_set_and_leaves_future_rows() {
+        let Some(backend) = harness_gc("window-take").await else {
+            return;
+        };
+        let past_pre_epoch = b"\x01win-past-pre-epoch".to_vec();
+        let past = b"\x01win-past".to_vec();
+        let future = b"\x01win-future".to_vec();
+        let far_future = b"\x01win-far-future".to_vec();
+
+        backend
+            .window_stage(&[
+                stages(&past_pre_epoch, at(-1_000_000_000)),
+                stages(&past, at(1_000_000_000)),
+                stages(&future, at(2_000_000_000)),
+                stages(&far_future, at(3_600_000_000_000)),
+            ])
+            .await
+            .unwrap();
+
+        let taken = backend.window_take(at(1_000_000_001)).await.unwrap();
+        let mut expected = vec![past_pre_epoch.clone(), past.clone()];
+        expected.sort();
+        assert_eq!(taken, expected, "exactly the overdue set");
+        assert_eq!(
+            backend.window_take(at(1_000_000_001)).await.unwrap(),
+            Vec::<Vec<u8>>::new(),
+            "the rows left the table: never handed out twice"
+        );
+        assert_eq!(
+            window_rows(&backend).await,
+            vec![
+                (far_future.clone(), 3_600_000_000_000),
+                (future.clone(), 2_000_000_000)
+            ],
+            "future rows left in place"
+        );
+    }
+
+    /// The take boundary is `visible_after == now` (the seam doc's
+    /// `<=`): a candidate visible exactly at the horizon is overdue;
+    /// one nanosecond later it is not.
+    #[tokio::test]
+    async fn window_take_boundary_visible_after_equal_now_is_overdue() {
+        let Some(backend) = harness_gc("window-boundary").await else {
+            return;
+        };
+        let equal = b"\x01win-equal-boundary".to_vec();
+        let just_later = b"\x01win-nanosecond-later".to_vec();
+
+        backend
+            .window_stage(&[
+                stages(&equal, at(1_500_000_000)),
+                stages(&just_later, at(1_500_000_001)),
+            ])
+            .await
+            .unwrap();
+
+        assert_eq!(
+            backend.window_take(at(1_500_000_000)).await.unwrap(),
+            vec![equal.clone()],
+            "visible_after == now is overdue (the `<=` boundary)"
+        );
+        assert_eq!(
+            backend.window_take(at(1_500_000_000)).await.unwrap(),
+            Vec::<Vec<u8>>::new(),
+            "the not-yet-overdue row stays (one nanosecond of slack)"
+        );
+        assert_eq!(
+            backend.window_take(at(1_500_000_001)).await.unwrap(),
+            vec![just_later.clone()],
+            "the horizon itself moved, so the row is now overdue"
+        );
+    }
+
+    /// Virgin-store shapes (the recorded read-only split): on a
+    /// schema-less database `window_take` maps a missing table to empty
+    /// (the read-tolerance parity with the postgres twin), while
+    /// `window_stage` is a write and surfaces the engine error without
+    /// silently creating schema. A constructed backend with an empty
+    /// window table also takes empty.
+    #[tokio::test]
+    async fn window_virgin_store_shapes() {
+        let Some(tableless) = harness_schemaless("window-virgin").await else {
+            return;
+        };
+        let key = b"\x01virgin-window-key".to_vec();
+
+        assert_eq!(
+            tableless.window_take(at(1_000_000_000)).await.unwrap(),
+            Vec::<Vec<u8>>::new(),
+            "schema-less take is a no-op read"
+        );
+        assert_eq!(
+            tableless.window_take(at(1_000_000_000)).await.unwrap(),
+            Vec::<Vec<u8>>::new(),
+            "still empty on a repeat"
+        );
+        match tableless.window_stage(&[stages(&key, at(1))]).await {
+            Err(Error::Io(message)) => {
+                assert!(
+                    message.contains("does not exist"),
+                    "the write surfaces the engine error, not a silent schema create: {message}"
+                )
+            }
+            other => panic!("schema-less stage is the engine error, got {other:?}"),
+        }
+
+        let Some(backend) = harness_gc("window-empty").await else {
+            return;
+        };
+        assert_eq!(
+            backend.window_take(SystemTime::now()).await.unwrap(),
+            Vec::<Vec<u8>>::new(),
+            "empty window table takes empty"
+        );
+    }
+
     /// Zero LO residue on the shared harness database (the C7 gate's
     /// permanent probe): every pg-lo test runs on its own dedicated
     /// database and never opens a pool against `blobs`, so the shared