From 29209fe53fbbe169fc55a5fd3e1c26c360f0a316 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 06:26:23 +0000 Subject: [PATCH] =?UTF-8?q?feat(gc):=20the=20postgres=20engine-state=20cla?= =?UTF-8?q?use=20families=20=E2=80=94=20the=20fleet=20host's=20full=20seam?= =?UTF-8?q?=20(wave=204)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - EngineWindowState: the both-tables ensure_gc_schema (gc_pins with the (key, owner) identity + gc_window), idempotent window_stage (the upsert discipline per (key, deadline)), and window_take as one atomic DELETE ... RETURNING with the <= boundary - EngineFleetState: PgJointTx (the pooled checkout carrying a session-level BEGIN), joint_put (the CAS clause inside the held tx), pin_commit (pin-row upsert-renewals + commit, rollback-on-failure => zero residue), pin_renew (expiry = now + DEFAULT_PIN_TTL, never resurrects), pin_check (the two-variant classification), arbitrate (the FOR UPDATE finger-lock folding pins + registered liveness-table references), and the engine-side register_liveness_source home - The sweep lease: the session-scoped pg advisory lock with the namespaced FNV lock key and the same-connection acquire/release holding discipline (the lease's checkout stored until release) - 13 contract tests (28 postgres tests total), env-driven, skipping cleanly server-less; verified green against the live pglo-poc harness server; clippy/fmt green default + --features postgres Verification: cargo test (69/19/3 ok), cargo test --features postgres server-less (97 ok, 28 skips) and live (97 ok), cargo test --all-features (152/54/4 ok), cargo clippy -D warnings (both configs), cargo fmt --check, cargo doc --no-deps. --- src/postgres.rs | 1406 +++++++++++++++++++++++++++++++- tasks/engine-postgres-fleet.md | 97 ++- 2 files changed, 1492 insertions(+), 11 deletions(-) diff --git a/src/postgres.rs b/src/postgres.rs index 0431b24..229ba99 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -103,11 +103,87 @@ //! [`Error::Missing`] is the rows-affected delete mapping and the //! virgin-store read-only split — no panics, no unwrap outside tests. //! -//! The engine-state clause families ([`crate::gc`]) are -//! `engine-postgres-fleet` (a separate task); the store-facade -//! constructor wiring is `store-postgres-construction`. This module -//! ships the engine and its in-module unit tests only. Tests need a -//! live server: connection settings ride the environment +//! **The engine-state clause families (ADR-010 §2 as reduced by ADR-012 +//! §5 — the postgres row is all yes: this is the fleet host).** The +//! [`EngineWindowState`] impl creates `gc_pins` + `gc_window` +//! (unconditionally — the applicability row does not condition on the +//! fleet declaration) and stages/takes the durable delete window; the +//! [`EngineFleetState`] impl carries the fleet family: the joint +//! entry+pin transaction, the pin clauses, delete-time SQL arbitration +//! with the engine-side liveness-table registration, and the sweep +//! lease. The store core drives these clauses in wave 6 (fleet mode) — +//! under `fleet: false` none of the fleet clauses run, while +//! `ensure_gc_schema` runs on every postgres store construction (the +//! `store-postgres-construction` wiring). The embedder's liveness table +//! is never created here (the sweep reads it and never owns it — +//! ADR-008's schema-ownership rule); the engine-side registration of +//! named liveness-table references (ADR-012 §6.6) is +//! [`PostgresBackend::register_liveness_source`], a crate-internal +//! mechanism the store core's fleet registration verb forwards to. +//! Every clause is contract-tested in-module per its applicability row. +//! +//! **Named design points (recorded for the review gate):** +//! +//! - **`JointTx` is [`PgJointTx`]** — the pooled checkout +//! ([`Object`]) held for the tx's life, with the transaction opened +//! as an explicit session-level `BEGIN` on that checkout and closed +//! by `COMMIT`/`ROLLBACK` in the clause bodies. tokio-postgres's +//! `Transaction` borrows its client (`&mut`), so it cannot be +//! co-owned with the checkout behind one `Send` field — the +//! exclusive checkout is what makes "the pooled client carries the +//! tx" real (no other statement can interleave on the session), and +//! ownership stays with the store core from `begin_joint_tx` until +//! `pin_commit` consumes it (ADR-010 §2's pinned ruling). +//! - **`pin_renew`'s successor time** — the clause signature carries +//! no expiry, so the engine computes `expiry = now + DEFAULT_PIN_TTL` +//! (ADR-008's default 60 s; the shipped constant is a backend field, +//! constructor-tunable when wave 6's fleet constructor lands). The +//! store owns the TTL and the TTL/3 cadence — `pin_commit`'s pin +//! rows carry the store-resolved expiry; renewals take the +//! successor-time shape recorded here. +//! - **Timestamp encoding** — signed i64 nanoseconds from the Unix +//! epoch ([`time_to_nanos`]), carried as `BIGINT` — the crate-wide +//! encoding (the sqlite engine's helpers are reference, not a +//! dependency: per-engine bodies, one encoding). +//! - **The sweep lease** — the session-scoped pg advisory lock: +//! `pg_try_advisory_lock` (atomic test-and-set) / +//! `pg_advisory_unlock`, session scope being the only scope whose +//! holdership ends at owner death or engine-session loss (ADR-013 +//! §2). The lock key is a fixed, namespaced integer ([`sweep_lock_key`]: +//! FNV-1a over `alkblobs:sweep:{schema}`, folded to positive i64) — +//! co-tenant engines on one instance (different schemas) never +//! collide, while the fleet's shared engine-owned schema is one lock +//! domain (one sweeper per pool). **The lease holds its connection**: +//! the acquired checkout is stored under a short-held +//! [`parking_lot::Mutex`] until release, so acquire and release run +//! on the same session — advisory locks are per-session, and a +//! release from another pooled connection would leave the holder's +//! lock live (the recorded pitfall). A second `sweep_lock_acquire` +//! on the same backend contended: the mutex answers `Ok(false)` +//! before any pool checkout. Release is no-op-safe and never errors: +//! by the call's end holdership has ended — unlocked, or already +//! reaped by the holder's session loss (an unlock failure is the +//! reaping case itself, not a fault to surface). +//! - **Pin rows upsert-renew (the re-put form)** — `pin_commit`'s +//! insert rides `ON CONFLICT (key, owner) DO UPDATE SET expiry`, and +//! `gc_pins`' primary key is `(key, owner)` (the row identity): +//! a re-put by the same instance renews its row (ADR-010 §4), and +//! protection composes by row (ADR-010 §2). +//! - **`arbitrate` folds under the `FOR UPDATE` finger-lock** — one +//! `BEGIN … COMMIT` step: un-expired pin rows for the key, then each +//! registered liveness-table reference, all `SELECT … FOR UPDATE` +//! keyed at delete time, folded into one bool before the commit. The +//! engine only classifies — it never decides deletion; the referenced +//! liveness tables are never created by the engine, and their +//! references are quoted ([`quote_table_reference`]) at registration. +//! - **Virgin-store tolerance asymmetry (recorded posture):** +//! `window_take` maps a missing table (SQLSTATE 42P01) to empty — +//! the read-tolerance parity with the sqlite twin; the fleet clauses +//! carry no tolerance (they run in fleet mode, where store +//! construction has ensured the GC schema), so a missing GC table +//! surfaces the engine error. +//! +//! Tests need a live server: connection settings ride the environment //! (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB`); tests skip cleanly //! server-less; acceptance requires green against the harness server //! (dockerized `pglo-poc`, `postgres:16-alpine`, :15432, db `blobs`, @@ -115,16 +191,31 @@ #![allow(dead_code)] -use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use deadpool_postgres::{Manager, ManagerConfig, Object, Pool, RecyclingMethod}; +use parking_lot::Mutex; use tokio_postgres::NoTls; use crate::backend::{clamped, Backend, ReadCursor, StagingCursor}; -use crate::Error; +use crate::gc::{EngineFleetState, EngineWindowState, PinRow, WindowCandidate}; +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 @@ -226,12 +317,40 @@ pub(crate) struct PostgresBackend { /// `"schema"."blobs"` — the quoted-identifier boundary every /// statement composes through. table: String, + /// The engine-owned schema name (the sweep lease's lock-key + /// namespace and the qualified GC tables' parent). + schema: String, + /// The GC pin table reference, quoted once at construction, + /// `"schema"."gc_pins"` (the fleet family's row home). + pins_table: String, + /// The GC window table reference, quoted once at construction, + /// `"schema"."gc_window"` (the window family's row home). + window_table: String, + /// The pin-TTL the successor-time source renews against (the + /// recorded design point: ADR-008's default 60 s, shipped as a + /// backend constant; constructor-tunability arrives with wave 6's + /// fleet constructor). + pin_ttl: Duration, + /// The sweep lease's held checkout: `Some` exactly while the lease + /// is held — the same-connection acquire/release discipline (the + /// advisory lock is per-session; the release-on-another-connection + /// hazard is recorded in the module doc). + lease: Mutex>, + /// The registered liveness-table references (ADR-012 §6.6's fleet + /// registration's engine-side home): quoted at registration, held + /// as references only — the tables are never created by the engine + /// (the schema-ownership rule). + liveness_sources: Arc>>, /// Test-only commit-failure injection, standing in for the media /// failure the kv tier cannot reliably produce in tests; mirrors /// the sqlite engine's knob so the stage-hygiene failure form is /// exercised against the real engine. #[cfg(test)] fail_commit: bool, + /// Test-only joint-commit-failure injection for the `pin_commit` + /// clause (the joint tx atomicity test's forced-failure form). + #[cfg(test)] + fail_joint_commit: bool, } impl PostgresBackend { @@ -259,11 +378,21 @@ impl PostgresBackend { .await .map_err(err)?; } + let pins_table = qualified_table(&schema, "gc_pins"); + let window_table = qualified_table(&schema, "gc_window"); Ok(Self { pool, table: qualified_table(&schema, "blobs"), + schema, + pins_table, + window_table, + pin_ttl: DEFAULT_PIN_TTL, + lease: Mutex::new(None), + liveness_sources: Arc::new(Mutex::new(Vec::new())), #[cfg(test)] fail_commit: false, + #[cfg(test)] + fail_joint_commit: false, }) } @@ -275,11 +404,22 @@ impl PostgresBackend { #[cfg(test)] async fn new_schemaless(config: &str, schema: &str) -> Result { let pool = build_pool(config, 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"); Ok(Self { pool, - table: qualified_table(schema, "blobs"), + table: qualified_table(&schema, "blobs"), + schema, + pins_table, + window_table, + pin_ttl: DEFAULT_PIN_TTL, + lease: Mutex::new(None), + liveness_sources: Arc::new(Mutex::new(Vec::new())), #[cfg(test)] fail_commit: false, + #[cfg(test)] + fail_joint_commit: false, }) } @@ -304,6 +444,23 @@ impl PostgresBackend { }; client.execute(&stmt, &[&keys]).await.map_err(err) } + + /// The engine-side registration of named liveness-table references + /// (ADR-012 §6.6's fleet registration's home): the store core's + /// fleet registration verb forwards here. The reference is a + /// (schema-qualified) table name, quoted at registration and held + /// as a reference only — the engine never creates the table (the + /// schema-ownership rule) and never learns liveness semantics; the + /// [`EngineFleetState::arbitrate`] clause is the only consumer. + /// Re-registering the same reference is idempotent. + pub(crate) fn register_liveness_source(&self, reference: &str) -> Result<(), Error> { + let quoted = quote_table_reference(reference)?; + let mut sources = self.liveness_sources.lock(); + if !sources.contains("ed) { + sources.push(quoted); + } + Ok(()) + } } /// Build the deadpool pool with the shipped posture: per-session @@ -359,6 +516,95 @@ CREATE TABLE IF NOT EXISTS {table} ( ) } +/// The pin-TTL constant's seconds ([`DEFAULT_PIN_TTL`]) — ADR-008's +/// default, constructor-tunable when wave 6's fleet constructor lands. +pub(crate) const DEFAULT_PIN_TTL_SECS: u64 = 60; + +/// The shipped pin-TTL ([`PostgresBackend::pin_renew`]'s successor-time +/// 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), + } +} + +fn nanos_to_time(n: i64) -> SystemTime { + if n >= 0 { + UNIX_EPOCH + std::time::Duration::from_nanos(n as u64) + } else { + UNIX_EPOCH - std::time::Duration::from_nanos((-n) as u64) + } +} + +/// The GC schema's DDL — idempotent, the fleet host's both-tables row +/// (ADR-012 §5): `gc_pins` with its `(key, owner)` row identity and +/// `gc_window` keyed by the candidate key, both timestamps `BIGINT` +/// nanos. The embedder's liveness table is never named here (the +/// schema-ownership rule). +fn gc_ddl(schema: &str) -> String { + let schema_q = quote_identifier(schema); + let pins = qualified_table(schema, "gc_pins"); + let window = qualified_table(schema, "gc_window"); + format!( + "CREATE SCHEMA IF NOT EXISTS {schema_q}; + +CREATE TABLE IF NOT EXISTS {pins} ( + key bytea NOT NULL, + owner bytea NOT NULL, + expiry bigint NOT NULL, + PRIMARY KEY (key, owner) +); + +CREATE TABLE IF NOT EXISTS {window} ( + key bytea PRIMARY KEY, + visible_after bigint NOT NULL +); +" + ) +} + +/// The sweep lease's advisory-lock key: FNV-1a over the namespaced +/// string `alkblobs:sweep:{schema}`, folded to positive i64. Deterministic +/// across processes and nodes — the fleet's shared schema derives one +/// key (one sweeper per pool), co-tenant schemas derive distinct keys. +fn sweep_lock_key(schema: &str) -> i64 { + let mut state: u64 = 0xcbf2_9ce4_8422_2325; + for byte in format!("alkblobs:sweep:{schema}").as_bytes() { + state ^= u64::from(*byte); + state = state.wrapping_mul(0x0000_0100_0000_01b3); + } + (state & !(1_u64 << 63)) as i64 +} + +/// Quote a (possibly schema-qualified) liveness-table reference at +/// registration: every dot-component rides the quoted-identifier +/// boundary; an empty component is a malformed reference, surfaced as +/// the engine error rather than interpolated unquoted. +fn quote_table_reference(reference: &str) -> Result { + let mut quoted = String::new(); + for part in reference.split('.') { + if part.is_empty() { + return Err(Error::Io(format!( + "postgres: invalid liveness table reference {reference:?}" + ))); + } + if !quoted.is_empty() { + quoted.push('.'); + } + quoted.push_str("e_identifier(part)); + } + Ok(quoted) +} + /// The materializing read cursor (ADR-010 §1's kv body family): the /// whole stored value in memory — `get` loads the row once (the kv tier /// holds only ≤ threshold content); `len()` reads the row's `size` @@ -570,9 +816,337 @@ impl Backend for PostgresBackend { } } +/// The `postgres` engine's [`EngineFleetState::JointTx`] — the pooled +/// checkout held for the tx's life, carrying a session-level `BEGIN` +/// opened by [`PostgresBackend::begin_joint_tx`] and closed by the +/// clause bodies (`COMMIT` in `pin_commit`'s success path, `ROLLBACK` +/// 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). +pub(crate) struct PgJointTx { + client: Object, + blobs_table: String, + pins_table: String, + #[cfg(test)] + fail_commit: bool, +} + +/// The window family (ADR-012 §5's postgres row): the fleet host's +/// both-tables schema and the durable staged delete window. Same +/// semantics as the sqlite engine's reference implementation — +/// different dialect. +impl EngineWindowState for PostgresBackend { + /// Create `gc_pins (key, owner, expiry)` + `gc_window (key, + /// visible_after)` — unconditionally, both tables (the applicability + /// row does not condition on the fleet declaration); idempotent, so + /// the twice-at-open call form is a no-op. The embedder's liveness + /// table is never created here (the schema-ownership rule). + async fn ensure_gc_schema(&self) -> Result<(), Error> { + let client = self.pool.get().await.map_err(err)?; + client + .batch_execute(&gc_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 engine's 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 + /// sqlite twin's virgin take; the read-tolerance 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) + } +} + +/// The fleet family (ADR-012 §5's postgres row — all yes): the joint +/// entry+pin transaction, the pin clauses, delete-time SQL arbitration, +/// and the sweep lease. Fleet activation is wave 6's constructor +/// declaration — under `fleet: false` the store core never calls these; +/// the clauses exist and are contract-tested per their applicability. +impl EngineFleetState for PostgresBackend { + type JointTx = PgJointTx; + + /// Open the joint tx on a pooled connection: checkout held for the + /// tx's life, session-level `BEGIN` issued, uncommitted until + /// [`Self::pin_commit`]. + async fn begin_joint_tx(&self) -> Result { + let client = self.pool.get().await.map_err(err)?; + client.batch_execute("BEGIN").await.map_err(pg_err)?; + Ok(PgJointTx { + client, + blobs_table: self.table.clone(), + pins_table: self.pins_table.clone(), + #[cfg(test)] + fail_commit: self.fail_joint_commit, + }) + } + + /// The entry half, executed inside the held tx: the engine's put + /// mechanics (the `ON CONFLICT DO NOTHING` CAS clause), same + /// first-writer-wins semantics as the public `put`. + async fn joint_put(&self, tx: &mut PgJointTx, key: &[u8], value: &[u8]) -> Result<(), Error> { + let sql = format!( + "INSERT INTO {} (key, value, size) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", + tx.blobs_table + ); + let stmt = tx.client.prepare_cached(&sql).await.map_err(pg_err)?; + let size = value.len() as i64; + tx.client + .execute(&stmt, &[&key, &value, &size]) + .await + .map_err(pg_err)?; + Ok(()) + } + + /// The joint terminal step: the pin-row inserts **and** the commit + /// — publish and protection land in one step. Every failure path + /// rolls the entire tx back (no pin rows, no published entry, zero + /// residue) before surfacing the error; consumes the tx. + async fn pin_commit(&self, tx: PgJointTx, entries: &[PinRow]) -> Result<(), Error> { + #[cfg(test)] + if tx.fail_commit { + tx.client.batch_execute("ROLLBACK").await.map_err(pg_err)?; + return Err(Error::Io( + "engine joint commit failure (postgres, injected)".to_owned(), + )); + } + let sql = format!( + "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3) \ + ON CONFLICT (key, owner) DO UPDATE SET expiry = excluded.expiry", + tx.pins_table + ); + let stmt = match tx.client.prepare_cached(&sql).await { + Ok(stmt) => stmt, + Err(e) => { + let _ = tx.client.batch_execute("ROLLBACK").await; + return Err(err(e)); + } + }; + for entry in entries { + let expiry = time_to_nanos(entry.expiry); + match tx + .client + .execute(&stmt, &[&entry.key, &entry.owner, &expiry]) + .await + { + Ok(_) => {} + Err(e) => { + let _ = tx.client.batch_execute("ROLLBACK").await; + return Err(err(e)); + } + } + } + match tx.client.batch_execute("COMMIT").await { + Ok(_) => Ok(()), + Err(e) => { + let _ = tx.client.batch_execute("ROLLBACK").await; + Err(err(e)) + } + } + } + + /// Renew the `(key, owner)` pin row: `expiry = now + DEFAULT_PIN_TTL` + /// (the recorded successor-time source) — and only against a live + /// row: the update's predicate requires the existing row un-expired, + /// so an expired or absent row is never resurrected (`false`, + /// nothing updated). + async fn pin_renew(&self, key: &[u8], owner: &[u8]) -> Result { + let client = self.pool.get().await.map_err(err)?; + let now = SystemTime::now(); + let successor = now + .checked_add(self.pin_ttl) + .ok_or_else(|| Error::Io("postgres: pin expiry overflow".to_owned()))?; + let successor_nanos = time_to_nanos(successor); + let now_nanos = time_to_nanos(now); + let sql = format!( + "UPDATE {} SET expiry = $1 WHERE key = $2 AND owner = $3 AND expiry > $4", + self.pins_table + ); + let stmt = client.prepare_cached(&sql).await.map_err(pg_err)?; + let updated = client + .execute(&stmt, &[&successor_nanos, &key, &owner, &now_nanos]) + .await + .map_err(pg_err)?; + Ok(updated == 1) + } + + /// Classify against un-expired pin rows: any un-expired `(key, *)` + /// row exists → `Protected`, else `Unprotected`. The engine never + /// decides deletion — it only classifies. + async fn pin_check(&self, key: &[u8]) -> Result { + let client = self.pool.get().await.map_err(err)?; + let now_nanos = time_to_nanos(SystemTime::now()); + let sql = format!( + "SELECT EXISTS(SELECT 1 FROM {} WHERE key = $1 AND expiry > $2)", + self.pins_table + ); + let stmt = client.prepare_cached(&sql).await.map_err(pg_err)?; + let present = client + .query_one(&stmt, &[&key, &now_nanos]) + .await + .map_err(pg_err)? + .try_get::<_, bool>(0) + .map_err(pg_err)?; + Ok(if present { + PinState::Protected + } else { + PinState::Unprotected + }) + } + + /// The full delete-time arbitration, folded under the `FOR UPDATE` + /// finger-lock in one `BEGIN … COMMIT` step: un-expired pin rows + /// for the key first, then each registered liveness-table + /// reference, all keyed at delete time — `true` = liveness protects + /// the key at this instant. The registration list is snapshotted + /// under its (process-local) mutex before the server round trip. + async fn arbitrate(&self, key: &[u8]) -> Result { + let sources = self.liveness_sources.lock().clone(); + let client = self.pool.get().await.map_err(err)?; + client.batch_execute("BEGIN").await.map_err(pg_err)?; + let folded: Result = async { + let now_nanos = time_to_nanos(SystemTime::now()); + let pins_sql = format!( + "SELECT 1 FROM {} WHERE key = $1 AND expiry > $2 FOR UPDATE LIMIT 1", + self.pins_table + ); + let stmt = client.prepare_cached(&pins_sql).await.map_err(pg_err)?; + let mut protected = client + .query_opt(&stmt, &[&key, &now_nanos]) + .await + .map_err(pg_err)? + .is_some(); + if !protected { + for table in &sources { + let sql = format!("SELECT 1 FROM {table} WHERE key = $1 FOR UPDATE LIMIT 1"); + let liveness_stmt = client.prepare_cached(&sql).await.map_err(pg_err)?; + let live = client + .query_opt(&liveness_stmt, &[&key]) + .await + .map_err(pg_err)? + .is_some(); + if live { + protected = true; + break; + } + } + } + client.batch_execute("COMMIT").await.map_err(pg_err)?; + Ok(protected) + } + .await; + if folded.is_err() { + let _ = client.batch_execute("ROLLBACK").await; + } + folded + } + + /// Acquire the sweep lease: the session-scoped pg advisory lock, + /// atomic test-and-set. `Ok(true)` = held (the lease's checkout is + /// stored so release rides the same session); `Ok(false)` = + /// 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. + async fn sweep_lock_acquire(&self) -> Result { + if self.lease.lock().is_some() { + return Ok(false); + } + let client = self.pool.get().await.map_err(err)?; + let stmt = client + .prepare_cached("SELECT pg_try_advisory_lock($1)") + .await + .map_err(pg_err)?; + let held = client + .query_one(&stmt, &[&sweep_lock_key(&self.schema)]) + .await + .map_err(pg_err)? + .try_get::<_, bool>(0) + .map_err(pg_err)?; + if held { + *self.lease.lock() = Some(client); + } + Ok(held) + } + + /// Release the sweep lease on the same session it was acquired on + /// (the mutex-held checkout). Releasing an unheld lease is a + /// no-op; a holder whose session died is the reaping case itself — + /// the call never errors, holdership has ended either way (an + /// unlock failure is that reaping, not a fault to surface). + async fn sweep_lock_release(&self) -> Result<(), Error> { + let held = self.lease.lock().take(); + let Some(client) = held else { + return Ok(()); + }; + let _ = async { + let stmt = client + .prepare_cached("SELECT pg_advisory_unlock($1)") + .await + .map_err(pg_err)?; + client + .query_one(&stmt, &[&sweep_lock_key(&self.schema)]) + .await + .map_err(pg_err)? + .try_get::<_, bool>(0) + .map_err(pg_err) + } + .await; + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; + use crate::gc::{EngineFleetState, EngineWindowState}; use std::sync::atomic::{AtomicU64, Ordering}; const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog"; @@ -967,8 +1541,16 @@ mod tests { let failing = PostgresBackend { pool: backend.pool.clone(), table: backend.table.clone(), + schema: backend.schema.clone(), + pins_table: backend.pins_table.clone(), + window_table: backend.window_table.clone(), + pin_ttl: backend.pin_ttl, + lease: Mutex::new(None), + liveness_sources: Arc::new(Mutex::new(Vec::new())), #[cfg(test)] fail_commit: true, + #[cfg(test)] + fail_joint_commit: false, }; let key = b"\x01failure-key".to_vec(); @@ -1288,4 +1870,812 @@ mod tests { Err(other) => panic!("max_size 0 must fail typed, got: {other}"), } } + + /// 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 at(nanos: i64) -> SystemTime { + nanos_to_time(nanos) + } + + /// 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 { + 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, + } + } + + fn pin_row(key: &[u8], owner: &[u8], expiry: SystemTime) -> PinRow { + PinRow { + key: key.to_vec(), + owner: owner.to_vec(), + expiry, + } + } + + /// 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: &PostgresBackend) -> Vec<(Vec, i64)> { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "SELECT key, visible_after FROM {} ORDER BY key", + backend.window_table + ); + let rows = client.query(sql.as_str(), &[]).await.unwrap(); + let mut out: Vec<(Vec, i64)> = rows + .iter() + .map(|row| (row.get::<_, Vec>(0), row.get::<_, i64>(1))) + .collect(); + out.sort(); + out + } + + /// The gc_pins rows as the engine stores them — the raw encoding. + async fn pin_rows(backend: &PostgresBackend) -> Vec<(Vec, Vec, i64)> { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "SELECT key, owner, expiry FROM {} ORDER BY key, owner", + backend.pins_table + ); + let rows = client.query(sql.as_str(), &[]).await.unwrap(); + rows.iter() + .map(|row| { + ( + row.get::<_, Vec>(0), + row.get::<_, Vec>(1), + row.get::<_, i64>(2), + ) + }) + .collect() + } + + /// A standalone pg session outside the backend's pool — the fleet + /// tests' instrument for cross-session facts (advisory-lock + /// contention, session-loss reaping, and the embedder's own + /// schema work). + async fn raw_connection(dsn: &str) -> (tokio_postgres::Client, tokio::task::JoinHandle<()>) { + let (client, conn) = tokio_postgres::connect(dsn, NoTls) + .await + .expect("raw harness connect"); + let handle = tokio::spawn(async move { + let _ = conn.await; + }); + (client, handle) + } + + /// The schema's table set — the schema-ownership test's probe. + async fn schema_tables(backend: &PostgresBackend) -> Vec { + let client = backend.pool.get().await.unwrap(); + let rows = client + .query( + "SELECT table_name::text FROM information_schema.tables \ + WHERE table_schema = $1 ORDER BY table_name", + &[&backend.schema], + ) + .await + .unwrap(); + rows.iter().map(|row| row.get::<_, String>(0)).collect() + } + + /// Schema contract (the fleet host's both-tables row): `ensure_gc_schema` + /// creates `gc_pins` (the `(key, owner)` identity) + `gc_window`, the + /// call is idempotent — and the embedder's liveness table is never + /// created by the engine (the schema-ownership rule: exactly the + /// engine's three tables exist, no `alive_keys`). + #[tokio::test] + async fn gc_schema_creates_both_tables_and_never_the_embedders() { + let Some(backend) = harness("gc-schema").await else { + return; + }; + backend.ensure_gc_schema().await.unwrap(); + backend.ensure_gc_schema().await.unwrap(); + assert_eq!( + schema_tables(&backend).await, + vec!["blobs", "gc_pins", "gc_window"], + "the engine's three tables exactly — no embedder liveness table" + ); + + 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_pins' \ + ORDER BY ordinal_position", + &[&backend.schema], + ) + .await + .unwrap() + .iter() + .map(|row| row.get::<_, String>(0)) + .collect(); + assert_eq!( + columns, + ["bytea", "bytea", "bigint"], + "the gc_pins row shape: key/owner/expiry" + ); + 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").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("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("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: on a schema-less database `window_take` maps + /// a missing table to empty (the read-tolerance parity with the + /// sqlite twin), while `window_stage` surfaces the engine error + /// without silently creating schema. A freshly constructed backend + /// with an empty window table also takes empty. + #[tokio::test] + async fn window_virgin_store_shapes() { + let Some(dsn) = harness_dsn() else { + skip_note(); + return; + }; + let tableless = PostgresBackend::new_schemaless(&dsn, &fresh_schema("window-virgin")) + .await + .unwrap(); + 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" + ); + } + + /// The joint entry+pin transaction publishes together: the entry row + /// and its pin row commit in one step (pin-before-publish, fleet + /// form), the joint put rides the same first-writer-wins CAS + /// semantics as the public `put` (a re-put is dedup), and a re-put's + /// pin row upsert renews the same instance's row. + #[tokio::test] + async fn joint_put_and_pin_commit_publish_together() { + let Some(backend) = harness_gc("joint").await else { + return; + }; + let key = b"\x01joint-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + + let mut tx = backend.begin_joint_tx().await.unwrap(); + EngineFleetState::joint_put(&backend, &mut tx, &key, CONTENT) + .await + .unwrap(); + let expiry = SystemTime::now() + DEFAULT_PIN_TTL; + backend + .pin_commit(tx, &[pin_row(&key, &owner, expiry)]) + .await + .unwrap(); + assert!(backend.has(&key).await.unwrap(), "entry published"); + assert_eq!( + backend.pin_check(&key).await.unwrap(), + PinState::Protected, + "the pin row landed with the entry" + ); + assert_eq!(stored_value(&backend, &key).await.as_deref(), Some(CONTENT)); + + let before = pin_rows(&backend).await; + assert_eq!( + before, + vec![(key.clone(), owner.clone(), time_to_nanos(expiry))], + "the pin row's identity is (key, owner), expiry the store-resolved one" + ); + + let mut tx = backend.begin_joint_tx().await.unwrap(); + EngineFleetState::joint_put(&backend, &mut tx, &key, b"overwritten") + .await + .unwrap(); + let renewed = SystemTime::now() + DEFAULT_PIN_TTL + Duration::from_secs(7); + backend + .pin_commit(tx, &[pin_row(&key, &owner, renewed)]) + .await + .unwrap(); + assert_eq!( + stored_value(&backend, &key).await.as_deref(), + Some(CONTENT), + "joint re-put does not overwrite (first writer wins)" + ); + assert_eq!( + pin_rows(&backend).await, + vec![(key, owner, time_to_nanos(renewed))], + "the re-put's pin row upsert renews the same instance's row" + ); + } + + /// The joint tx atomicity (the stage-hygiene invariant's tx form): a + /// forced failure at `pin_commit` aborts the entire tx — no pin + /// rows, no published entry, zero residue. + #[tokio::test] + async fn pin_commit_failure_aborts_with_zero_residue() { + let Some(backend) = harness_gc("joint-fail").await else { + return; + }; + let key = b"\x01joint-failure-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + + let failing = PostgresBackend { + pool: backend.pool.clone(), + table: backend.table.clone(), + schema: backend.schema.clone(), + pins_table: backend.pins_table.clone(), + window_table: backend.window_table.clone(), + pin_ttl: backend.pin_ttl, + lease: Mutex::new(None), + liveness_sources: Arc::new(Mutex::new(Vec::new())), + #[cfg(test)] + fail_commit: false, + #[cfg(test)] + fail_joint_commit: true, + }; + let mut tx = failing.begin_joint_tx().await.unwrap(); + EngineFleetState::joint_put(&failing, &mut tx, &key, CONTENT) + .await + .unwrap(); + let expiry = SystemTime::now() + DEFAULT_PIN_TTL; + let err = failing + .pin_commit(tx, &[pin_row(&key, &owner, expiry)]) + .await + .unwrap_err(); + assert!( + matches!(err, Error::Io(_)), + "engine joint media failure surfaces as Io" + ); + + assert!( + !backend.has(&key).await.unwrap(), + "the aborted tx published nothing" + ); + assert_eq!( + backend.pin_check(&key).await.unwrap(), + PinState::Unprotected, + "no pin row survived" + ); + assert_eq!(pin_rows(&backend).await, vec![], "zero residue in gc_pins"); + assert!( + backend.list().await.unwrap().is_empty(), + "zero residue in the blobs table itself" + ); + } + + /// `pin_renew`'s expiry semantics (ADR-012 §2 verbatim): a live row + /// renews to the successor time (`now + DEFAULT_PIN_TTL`, the + /// recorded source); an expired or absent row is never resurrected + /// — `false`, nothing updated. + #[tokio::test] + async fn pin_renew_never_resurrects_an_expired_or_absent_row() { + let Some(backend) = harness_gc("renew").await else { + return; + }; + let key = b"\x01renew-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + + assert!( + !backend.pin_renew(&key, &owner).await.unwrap(), + "absent row: no renewal, no creation" + ); + assert_eq!(pin_rows(&backend).await, vec![], "absent stays absent"); + + let live = SystemTime::now() + Duration::from_secs(30); + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)", + backend.pins_table + ); + client + .execute(sql.as_str(), &[&key, &owner, &time_to_nanos(live)]) + .await + .unwrap(); + } + assert!( + backend.pin_renew(&key, &owner).await.unwrap(), + "live row renews" + ); + let after = pin_rows(&backend).await; + let expected = + time_to_nanos(SystemTime::now()) + DEFAULT_PIN_TTL_SECS as i64 * 1_000_000_000; + let got = after[0].2; + assert!( + (got - expected).abs() < 3_000_000_000, + "renewed to the successor time now + {DEFAULT_PIN_TTL_SECS}s: {got} vs {expected}" + ); + + let expired_nanos = time_to_nanos(at(-1_000_000_000)); + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "UPDATE {} SET expiry = $1 WHERE key = $2 AND owner = $3", + backend.pins_table + ); + client + .execute(sql.as_str(), &[&expired_nanos, &key, &owner]) + .await + .unwrap(); + } + assert!( + !backend.pin_renew(&key, &owner).await.unwrap(), + "expired row: never resurrected" + ); + assert_eq!( + pin_rows(&backend).await, + vec![(key, owner, expired_nanos)], + "the expired row's expiry is untouched" + ); + } + + /// `pin_check`'s two-variant classification against un-expired pin + /// rows: absent or only-expired rows → `Unprotected`; any live row + /// → `Protected`. Protection composes by row, so any owner's live + /// row counts. + #[tokio::test] + async fn pin_check_classifies_the_two_variants() { + let Some(backend) = harness_gc("check").await else { + return; + }; + let key = b"\x01check-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + let other_owner = b"\x02instance-id".to_vec(); + + assert_eq!( + backend.pin_check(&key).await.unwrap(), + PinState::Unprotected, + "no rows: Unprotected" + ); + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)", + backend.pins_table + ); + client + .execute( + sql.as_str(), + &[&key, &owner, &time_to_nanos(at(-5_000_000_000))], + ) + .await + .unwrap(); + } + assert_eq!( + backend.pin_check(&key).await.unwrap(), + PinState::Unprotected, + "only an expired row: Unprotected (abandoned pins never protect)" + ); + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)", + backend.pins_table + ); + client + .execute( + sql.as_str(), + &[ + &key, + &other_owner, + &time_to_nanos(SystemTime::now() + DEFAULT_PIN_TTL), + ], + ) + .await + .unwrap(); + } + assert_eq!( + backend.pin_check(&key).await.unwrap(), + PinState::Protected, + "any owner's live row protects (protection composes by row)" + ); + } + + /// `arbitrate` folds the full answer under the `FOR UPDATE` + /// finger-lock: un-expired pin rows plus registered liveness-table + /// references — here a test-owned embedder table the engine never + /// created. `true` = liveness protects the key at this instant; + /// expired pins and absent rows protect nothing. + #[tokio::test] + async fn arbitrate_folds_pins_and_registered_liveness_tables() { + let Some(backend) = harness_gc("arbitrate").await else { + return; + }; + let key = b"\x01arbitrate-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + + let embedder_table = format!("{}.alive_keys", backend.schema); + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "CREATE TABLE {} (key bytea PRIMARY KEY, seq bigint)", + quote_table_reference(&embedder_table).unwrap() + ); + client.execute(sql.as_str(), &[]).await.unwrap(); + } + backend.register_liveness_source(&embedder_table).unwrap(); + backend.register_liveness_source(&embedder_table).unwrap(); + let tables = schema_tables(&backend).await; + assert_eq!( + tables, + vec!["alive_keys", "blobs", "gc_pins", "gc_window"], + "the embedder's table exists only because the test (the embedder) created it" + ); + + assert!( + !backend.arbitrate(&key).await.unwrap(), + "nothing protects an absent key" + ); + + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "INSERT INTO {} (key, seq) VALUES ($1, $2)", + quote_table_reference(&embedder_table).unwrap() + ); + client.execute(sql.as_str(), &[&key, &1_i64]).await.unwrap(); + } + assert!( + backend.arbitrate(&key).await.unwrap(), + "a registered liveness row protects the key" + ); + + { + let client = backend.pool.get().await.unwrap(); + client + .execute( + &format!( + "DELETE FROM {} WHERE key = $1", + quote_table_reference(&embedder_table).unwrap() + ), + &[&key], + ) + .await + .unwrap(); + } + assert!( + !backend.arbitrate(&key).await.unwrap(), + "liveness row gone, no pin rows: unprotected" + ); + + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)", + backend.pins_table + ); + client + .execute( + sql.as_str(), + &[&key, &owner, &time_to_nanos(at(-1_000_000_000))], + ) + .await + .unwrap(); + } + assert!( + !backend.arbitrate(&key).await.unwrap(), + "an expired pin row is abandoned: it protects nothing" + ); + + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "UPDATE {} SET expiry = $1 WHERE key = $2 AND owner = $3", + backend.pins_table + ); + client + .execute( + sql.as_str(), + &[ + &time_to_nanos(SystemTime::now() + DEFAULT_PIN_TTL), + &key, + &owner, + ], + ) + .await + .unwrap(); + } + assert!( + backend.arbitrate(&key).await.unwrap(), + "a live pin row protects the key" + ); + } + + /// The `FOR UPDATE` finger-lock is real: a raw session holding the + /// pin row's row lock at delete time blocks a concurrent + /// `pin_renew` (which targets the same row) until the holder + /// commits — the arbitration's exclusion of the renewal path. + #[tokio::test] + async fn arbitrate_finger_lock_blocks_a_concurrent_renewal() { + let Some(backend) = harness_gc("finger-lock").await else { + return; + }; + let key = b"\x01finger-lock-key".to_vec(); + let owner = b"\x01instance-id".to_vec(); + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)", + backend.pins_table + ); + client + .execute( + sql.as_str(), + &[ + &key, + &owner, + &time_to_nanos(SystemTime::now() + DEFAULT_PIN_TTL), + ], + ) + .await + .unwrap(); + } + + let client = backend.pool.get().await.unwrap(); + client.batch_execute("BEGIN").await.unwrap(); + let lock_sql = format!( + "SELECT expiry FROM {} WHERE key = $1 AND owner = $2 FOR UPDATE", + backend.pins_table + ); + let stmt = client.prepare(&lock_sql).await.unwrap(); + let locked = client.query(&stmt, &[&key, &owner]).await.unwrap().len(); + assert_eq!(locked, 1, "the raw session holds the finger-lock"); + + let (renewed_tx, mut renewed_rx) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + let renewed = backend.pin_renew(&key, &owner).await; + let _ = renewed_tx.send(renewed); + }); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + assert!( + matches!( + renewed_rx.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Empty) + ), + "the renewal is still blocked on the finger-lock" + ); + + client.batch_execute("COMMIT").await.unwrap(); + assert!( + renewed_rx.await.unwrap().unwrap(), + "renew lands once the holder releases" + ); + } + + /// The sweep lease (ADR-013 §2's named exclusive semantics, shipped + /// as the session-scoped pg advisory lock): atomic test-and-set + /// (`Ok(true)` = held), one sweep per backend and per schema — a + /// second acquire and a raw session's try both contend — release is + /// no-op-safe, and a release actually frees the lock for the next + /// acquirer (the same-connection discipline did its job). + #[tokio::test] + async fn sweep_lock_acquire_contention_release() { + let Some(backend) = harness_gc("sweep-lock").await else { + return; + }; + + assert!( + backend.sweep_lock_acquire().await.unwrap(), + "first acquire: held" + ); + assert!( + !backend.sweep_lock_acquire().await.unwrap(), + "second acquire on the held backend contended" + ); + + let dsn = harness_dsn().unwrap(); + let (raw, _conn) = raw_connection(&dsn).await; + let key = sweep_lock_key(&backend.schema); + let held = raw + .query_one("SELECT pg_try_advisory_lock($1)", &[&key]) + .await + .unwrap() + .get::<_, bool>(0); + assert!(!held, "a concurrent session's try contends on the same key"); + + backend.sweep_lock_release().await.unwrap(); + backend.sweep_lock_release().await.unwrap(); + assert!( + backend.sweep_lock_acquire().await.unwrap(), + "acquire after release: held again (release freed the lock)" + ); + backend.sweep_lock_release().await.unwrap(); + } + + /// Session-loss reaping: a lock held by a foreign session reaps at + /// that session's death without any release call — the scope whose + /// holdership ends at engine-session loss (ADR-013 §2). + #[tokio::test] + async fn sweep_lock_session_loss_reaps_the_lease() { + let Some(backend) = harness_gc("sweep-reap").await else { + return; + }; + let dsn = harness_dsn().unwrap(); + let key = sweep_lock_key(&backend.schema); + let (holder, conn) = raw_connection(&dsn).await; + let held = holder + .query_one("SELECT pg_try_advisory_lock($1)", &[&key]) + .await + .unwrap() + .get::<_, bool>(0); + assert!(held, "the foreign session holds the lock"); + + assert!( + !backend.sweep_lock_acquire().await.unwrap(), + "the backend's acquire contends while the foreign session lives" + ); + + drop(holder); + let _ = conn.await; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!( + backend.sweep_lock_acquire().await.unwrap(), + "session loss reaped the lease — the backend now holds it" + ); + backend.sweep_lock_release().await.unwrap(); + } } diff --git a/tasks/engine-postgres-fleet.md b/tasks/engine-postgres-fleet.md index 8efc6ef..39f5107 100644 --- a/tasks/engine-postgres-fleet.md +++ b/tasks/engine-postgres-fleet.md @@ -1,7 +1,7 @@ --- id: engine-postgres-fleet name: postgres engine-state clause families — the fleet host's full seam -status: pending +status: completed depends_on: [engine-postgres] scope: moderate risk: medium @@ -191,8 +191,99 @@ engine-level contract tests. ## Notes -> To be filled by implementation agent +> Implemented in `src/postgres.rs` (wave 4, this task's only code +> change; both impls sit behind the pre-existing `#[cfg(feature = +> "postgres")]` module gate). Named design points, as resolved: +> +> - **`JointTx`** — a crate-internal `PgJointTx`: the deadpool checkout +> (`deadpool_postgres::Object`) held for the tx's life, with the +> transaction opened as an explicit session-level `BEGIN` and closed +> by `COMMIT`/`ROLLBACK` in the clause bodies. tokio-postgres's +> `Transaction` borrows its client (`&mut`), so it cannot be +> co-owned with the checkout behind one `Send` field; the exclusive +> checkout *is* what "the pooled client carries the tx" means +> (nothing else can interleave on the session), and ownership stays +> with the store core from `begin_joint_tx` to `pin_commit` per the +> trait's pinned ruling. Every `pin_commit` failure path rolls the +> tx back before surfacing `Err` (zero residue); dropping a tx +> mid-flight is outside the seam contract (the store core holds it). +> - **`pin_renew` successor time** — the engine computes +> `expiry = now + DEFAULT_PIN_TTL` from the shipped 60 s constant +> (`DEFAULT_PIN_TTL_SECS`, a backend field, constructor-tunable when +> wave 6's fleet constructor lands); the store owns the TTL and the +> TTL/3 cadence (`pin_commit`'s rows carry the store-resolved +> expiry). The update's predicate requires un-expired rows only, so +> expired/absent rows are never resurrected. +> - **Timestamp encoding** — signed i64 nanos from the Unix epoch, +> `time_to_nanos`/`nanos_to_time` mirrored as per-engine bodies (the +> sqlite helpers are reference, one encoding). +> - **Sweep lease** — session-scoped pg advisory lock +> (`pg_try_advisory_lock`/`pg_advisory_unlock`): the only scope whose +> holdership ends at owner death/session loss (ADR-013 §2). Lock key +> = FNV-1a over `alkblobs:sweep:{schema}` folded to positive i64 — +> the fleet's shared schema is one lock domain (one sweeper per +> pool), co-tenant schemas derive distinct keys. **Connection +> holding**: the acquired checkout is stored under a +> `parking_lot::Mutex>` until release, so +> acquire/release ride the same session (the recorded pitfall); +> a second acquire answers `Ok(false)` from the mutex before any +> pool checkout; release is no-op-safe and never errors (an unlock +> failure is the session-loss reaping case itself). +> - **Pin rows** — `gc_pins` carries `PRIMARY KEY (key, owner)` (the +> row identity) and `pin_commit` inserts with `ON CONFLICT (key, +> owner) DO UPDATE SET expiry`: a re-put by the same instance +> upsert-renews its row (ADR-010 §4), protection composes by row. +> - **`arbitrate`** — one `BEGIN … COMMIT` fold: un-expired pin rows +> first, then each registered liveness-table reference, all +> `SELECT … FOR UPDATE LIMIT 1` keyed at delete time, one bool +> answer. The registration list (a `parking_lot::Mutex>` +> behind an `Arc`, shared with clones) is snapshotted before the +> server round trip. +> - **Liveness-table registration** — +> `PostgresBackend::register_liveness_source(ref)` is the engine-side +> home of ADR-012 §6.6's fleet registration: every dot-component of +> the reference rides `quote_identifier` at registration (hostile +> names safe, malformed refs error), re-registration is idempotent, +> the engine holds references only and never creates the tables. +> - **Virgin tolerance asymmetry** — `window_take` maps SQLSTATE 42P01 +> to empty (the sqlite twin's read parity); the fleet clauses carry +> no tolerance — they run in fleet mode where store construction has +> ensured the GC schema. +> - **Error mapping** — new `pg_err` for the clause bodies surfaces +> the `DbError` message + SQLSTATE (plain `Display` is just "db +> error"); the landed engine's sites keep the original mapping. +> - **Contract tests (13 new; 28 postgres tests total)** — the +> both-tables schema + the never-created embedder column probe, +> stage idempotence/re-stage, take's exact set / atomic re-take / +> future rows / `<=` boundary / pre-epoch encoding / virgin shapes, +> the joint publish-together path with CAS dedup + upsert renewal, +> the injected joint-failure zero-residue abort, `pin_renew`'s +> live/expired/absent trio, `pin_check`'s two variants, arbitrate +> folding pins + a test-owned embedder table plus the live +> finger-lock-blocks-concurrent-renewal proof, and the advisory +> lock's acquire/contention/release + foreign-session-loss reaping. +> Harness server verified live 2026-10-12. ## Summary -> To be filled on completion \ No newline at end of file +> Completed 2026-10-12 (wave 4). `src/postgres.rs` implements +> `EngineWindowState` (the fleet host's both-tables `ensure_gc_schema`, +> idempotent `window_stage`, atomic `DELETE … RETURNING` +> `window_take` with the `<=` boundary) and `EngineFleetState` +> (PgJointTx joint entry+pin tx, `pin_commit` with rollback-on-failure +> and upsert-renewal pin-row inserts, `pin_renew`'s no-resurrection +> expiry semantics, two-variant `pin_check`, the `FOR UPDATE` +> finger-lock `arbitrate` folding pin rows + registered liveness-table +> references, and the session-scoped advisory sweep lease with its +> same-connection holding discipline) — every clause family of +> ADR-012 §5's postgres row, contract-tested in-module with the +> established env-driven harness convention (skip cleanly server-less, +> green against the live pglo-poc server). 13 new tests; fleet +> activation remains wave 6 (the store core never calls the fleet +> clauses under `fleet: false`), and the `ensure_gc_schema` +> constructor wiring remains `store-postgres-construction`. Verified: +> `cargo test` (69/19/3 ok), `cargo test --features postgres` +> server-less (97 ok, 28 skips; with server 97 ok, all live), +> `cargo test --all-features` (152/54/4 ok), `cargo clippy +> --all-targets -- -D warnings` and `--all-features --all-targets` +> green, `cargo fmt --check` green, `cargo doc --no-deps` clean. \ No newline at end of file