From c8fc519978663df691ccaab39211436f35a446e5 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 07:40:59 +0000 Subject: [PATCH] =?UTF-8?q?feat(gc):=20the=20pg-lo=20engine-state=20window?= =?UTF-8?q?=20family=20=E2=80=94=20the=20conditional=20window=20host=20(wa?= =?UTF-8?q?ve=205)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 402 ++++++++++++++++++++++++++++++++++- tasks/engine-pg-lo-window.md | 79 ++++++- 2 files changed, 468 insertions(+), 13 deletions(-) diff --git a/src/pg_lo.rs b/src/pg_lo.rs index 23e13d1..48a1df9 100644 --- a/src/pg_lo.rs +++ b/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>, 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>(0).map_err(pg_err)) + .collect::>, _>>()?; + keys.sort(); + Ok(keys) + } +} + /// `lo_create` — a fresh LO oid inside the caller's tx. async fn lo_create(client: &Object) -> Result { 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 { + 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, 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, i64)> = rows + .iter() + .map(|row| (row.get::<_, Vec>(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 { + 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 = 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::>::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::>::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::>::new(), + "schema-less take is a no-op read" + ); + assert_eq!( + tableless.window_take(at(1_000_000_000)).await.unwrap(), + Vec::>::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::>::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 diff --git a/tasks/engine-pg-lo-window.md b/tasks/engine-pg-lo-window.md index 84a8e28..9cd1c6c 100644 --- a/tasks/engine-pg-lo-window.md +++ b/tasks/engine-pg-lo-window.md @@ -1,7 +1,7 @@ --- id: engine-pg-lo-window name: pg-lo engine-state window family — the conditional window host -status: pending +status: completed depends_on: [engine-pg-lo] scope: narrow risk: low @@ -70,22 +70,22 @@ engine-level contract tests. ## Acceptance Criteria -- [ ] The pg-lo backend implements `EngineWindowState` — the window +- [x] The pg-lo backend implements `EngineWindowState` — the window family only; no pin-clause or sweep-lock body exists anywhere in the pg-lo engine (nothing of `EngineFleetState`) -- [ ] `ensure_gc_schema` creates `gc_window` only; `gc_pins` is never +- [x] `ensure_gc_schema` creates `gc_window` only; `gc_pins` is never created by pg-lo (contract-tested); the embedder's liveness table is never created -- [ ] `window_stage` idempotent per `(key, deadline)`; a re-stage at a +- [x] `window_stage` idempotent per `(key, deadline)`; a re-stage at a new deadline updates the row -- [ ] `window_take` returns exactly the overdue set (`<= now` +- [x] `window_take` returns exactly the overdue set (`<= now` boundary), removes it atomically, leaves future rows; empty on none/virgin -- [ ] Contract tests env-driven (`ALKBLOBS_PG_*`), skipping cleanly +- [x] Contract tests env-driven (`ALKBLOBS_PG_*`), skipping cleanly server-less; green against the harness server (`pglo-poc`, :15432, db `blobs`, `postgres`/`poc`) at acceptance; the janitor discipline rides from the start (amendment A2) -- [ ] `cargo test` (default) green; `cargo test --features pg-lo` +- [x] `cargo test` (default) green; `cargo test --features pg-lo` green server-less (skipping); `cargo clippy --all-targets -- -D warnings` and `cargo fmt --check` green (default + `--features pg-lo`); the pg-lo pure builds clean (`--no-default-features @@ -110,8 +110,69 @@ engine-level contract tests. ## Notes -> To be filled by implementation agent +> Implemented in `src/pg_lo.rs` only. `PgLoBackend` gained a +> `window_table` field (the quoted `"schema"."gc_window"` reference, +> quoted-identifier boundary, constructed in `new` and `new_schemaless` +> — the two `#[cfg(test)]` failure-injection struct literals picked up +> the new field mechanically) and a `gc_window_ddl` body beside +> `bootstrap_ddl`. The `EngineWindowState` impl mirrors the postgres +> reference implementation statement-for-statement (same upsert / same +> single `DELETE … RETURNING` take, this engine's `pg_err`/`err` +> mapping and `time_to_nanos` from the shared `pg_common` home); the +> per-engine reverse decoder (`nanos_to_time`) lives in the test +> module like the postgres suite's. Recorded design points: +> +> - **The `gc_window`-only schema row** — `ensure_gc_schema` runs +> `CREATE SCHEMA IF NOT EXISTS` + `CREATE TABLE gc_window (key bytea +> PRIMARY KEY, visible_after bigint NOT NULL)` only; the schema's DDL +> names no `gc_pins` and no liveness table (ADR-010 §2's +> schema-ownership rule). The engine's constructor continues to +> create only `lo_entries` — the GC table is created by the clause +> alone, and whether the store core calls it at construction is the +> one-window-host rule's business (`store-pg-lo-construction` wires +> the active host) — untouched this task. +> - **Nothing of `EngineFleetState`** — no pin clause, no sweep-lock +> body anywhere in the module (grep-clean; only doc mentions carry +> the names), and `gc_pins`'s absence is contract-tested +> (`gc_schema_creates_window_table_only` asserts the engine-owned +> schema holds exactly `["gc_window", "lo_entries"]`). +> - **Dormancy posture recorded** — the impl doc states the +> one-window-host rule reading: clauses 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; the clause body +> exists and is contract-tested per applicability regardless. +> +> No store-core changes (`store.rs` untouched — the sweep's durable +> form drives the seam through `KvEngine`, which needs no new arm for +> this task). ## Summary -> To be filled on completion \ No newline at end of file +> Completed 2026-10-12 (wave 5). `src/pg_lo.rs` implements +> `EngineWindowState` — the window family only (ADR-012 §5's pg-lo +> row): `ensure_gc_schema` creates `gc_window (key bytea PRIMARY KEY, +> visible_after bigint)` in the engine's own schema and never +> `gc_pins` or the embedder's liveness table (contract-tested — the +> schema holds exactly `gc_window` + `lo_entries`); `window_stage` +> upserts idempotently per `(key, deadline)` with same-deadline +> idempotence and new-deadline update; `window_take(now)` is the one +> atomic `DELETE … RETURNING` step over `visible_after <= now` (the +> `<=` boundary pinned), keys sorted, with the 42P01 → empty +> virgin-take read tolerance and stage-less-write surfacing. Timestamps +> ride the crate-wide i64-nanos encoding (`pg_common::time_to_nanos`; +> the reverse decoder is the test module's per-engine body). Five +> contract tests mirror the postgres/sqlite suites: schema-only + +> row-shape + idempotence-of-call, stage idempotence/restage-at-new- +> deadline, take-exact-overdue-set/never-twice/future-left, the +> `<=` boundary with nanosecond slack, and both virgin shapes +> (schema-less take empty twice, schema-less stage surfaces the engine +> error, empty-window take). Env-driven (`ALKBLOBS_PG_*`), skipping +> cleanly server-less, green live against the harness server (23/23 +> in-module tests, per-test dedicated databases + `DatabaseJanitor`) +> with zero LO residue on the shared db gate green. Verified: `cargo +> test` (69/19/3 ok; 92/19/3 with `--features pg-lo` server-less), +> clippy `-D warnings` green (default, `--features pg-lo`, +> `--no-default-features --features "pg-lo,mem"`, `--no-default- +> features --features "postgres,pg-lo"`), `cargo fmt --check` green, +> `cargo doc --no-deps` clean. Remaining: constructor wiring is +> `store-pg-lo-construction`. \ No newline at end of file