diff --git a/src/lib.rs b/src/lib.rs index 3434850..2d899a0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,11 +27,14 @@ pub mod pg_lo; #[cfg(feature = "ops")] pub mod ops; +#[cfg(any(feature = "postgres", feature = "pg-lo"))] +mod pg_common; + #[cfg(feature = "postgres")] pub mod postgres; -#[cfg(feature = "postgres")] -pub use postgres::PgOpts; +#[cfg(any(feature = "postgres", feature = "pg-lo"))] +pub use pg_common::PgOpts; #[cfg(feature = "sqlite")] pub mod sqlite; diff --git a/src/pg_common.rs b/src/pg_common.rs new file mode 100644 index 0000000..6d05a46 --- /dev/null +++ b/src/pg_common.rs @@ -0,0 +1,321 @@ +//! The crate-internal pg-common module — the shared-helper design point +//! resolved (engine-pg-lo): the pieces both postgres-attached engines +//! (`postgres` kv, `pg-lo` large) share, gated +//! `cfg(any(feature = "postgres", feature = "pg-lo"))` so each engine's +//! pure build (ADR-013 §4: `pg-lo` does not imply `postgres`; the +//! pg-lo-pure build) compiles without duplication. Held here: the +//! deadpool pool construction (the shipped driver posture), +//! [`quote_identifier`] (the quoted-identifier boundary), the SQLSTATE +//! 42P01 (undefined table) virgin-store discriminant, the error mapping, +//! the metadata-size clamp, the i64-nanos timestamp encoding, and +//! [`PgOpts`] — the shared construction-opts type whose public +//! re-export gate is widened to `any(postgres, pg-lo)` with the +//! definition's move here (the type is unchanged; `alkblobs::PgOpts` +//! now exists in the pg-lo-pure build too). +//! +//! The error mapping keeps the `postgres:` prefix for both engines: each +//! attaches to a PostgreSQL server, so family attribution reads the same +//! from either engine's failures. + +use std::time::{SystemTime, UNIX_EPOCH}; + +use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod}; +use tokio_postgres::NoTls; + +use crate::Error; + +/// The error mapping's base form (the postgres engine's): a failure +/// surfaces as [`Error::Io`] with the engine family prefix. +pub(crate) fn err(e: impl std::fmt::Display) -> Error { + Error::Io(format!("postgres: {e}")) +} + +/// The engine's error mapping for statement bodies: a SQL-level failure +/// surfaces with its server-side message and SQLSTATE (the `DbError` +/// detail `Display` drops), anything else through [`err`]. +pub(crate) fn pg_err(e: tokio_postgres::Error) -> Error { + match e.as_db_error() { + Some(db) => Error::Io(format!("postgres: {db}")), + None => err(e), + } +} + +/// The virgin-store tolerance's discriminant (POC #1 finding 2): a +/// statement naming a table that does not exist maps on the read paths +/// to absent/empty (and delete to Missing), never an error. The stable +/// surface PostgreSQL carries is the SQLSTATE class+code (42P01, +/// undefined table); a missing *schema* reads identically. +pub(crate) fn is_undefined_table(e: &tokio_postgres::Error) -> bool { + e.code() == Some(&tokio_postgres::error::SqlState::UNDEFINED_TABLE) +} + +/// Quote a SQL identifier (the Postgres rule: double the embedded `"`). +/// +/// The engines compose schema-qualified SQL from consumer-supplied +/// schema names; this is the quoted-identifier boundary — an identifier +/// is never interpolated into SQL text unquoted (the alkstore +/// `quote_identifier` pattern). +pub(crate) fn quote_identifier(name: &str) -> String { + let mut quoted = String::with_capacity(name.len() + 2); + quoted.push('"'); + for c in name.chars() { + if c == '"' { + quoted.push('"'); + } + quoted.push(c); + } + quoted.push('"'); + quoted +} + +/// A schema-qualified table reference: `"schema"."table"`. +pub(crate) fn qualified_table(schema: &str, table: &str) -> String { + format!("{}.{}", quote_identifier(schema), quote_identifier(table)) +} + +/// The default engine-owned schema name (the recorded design point's +/// default). Both postgres-attached engines default to it — the +/// consolidation shape (ADR-012 §4's same-pg-instance rule) holds both +/// engines' tables in one engine-owned schema without name collision +/// (`blobs`, `lo_entries`, and the GC tables are distinct names). +pub(crate) const DEFAULT_SCHEMA: &str = "alkblobs"; + +/// The shipped default pool size ([`PgOpts::max_size`]) — the alkstore +/// `DEFAULT_MAX_SIZE = 8` precedent. +pub(crate) const DEFAULT_MAX_SIZE: usize = 8; + +/// The `size` metadata's integer, clamped to the domain the trait probes +/// in. +pub(crate) fn size_u64(raw: i64) -> u64 { + raw.max(0) as u64 +} + +/// The crate-wide timestamp encoding (the recorded design point): signed +/// i64 nanoseconds from the Unix epoch — sub-second precision for the +/// test tunings, negative for pre-epoch instants, i64 across the full +/// range a `BIGINT` stores (±292 years around the epoch). +pub(crate) fn time_to_nanos(t: SystemTime) -> i64 { + match t.duration_since(UNIX_EPOCH) { + Ok(d) => d.as_nanos() as i64, + Err(e) => -(e.duration().as_nanos() as i64), + } +} + +/// Build the deadpool pool with the shipped posture: per-session +/// `synchronous_commit` merged into every connection's connect options +/// (appended to the consumer's own `options=` DSN values — postgres +/// parses repeated `-c` flags), `RecyclingMethod::Fast`, the configured +/// `max_size`. `max_size = 0` fails at entry, before any round trip. +pub(crate) fn build_pool( + config: &str, + max_size: usize, + synchronous_commit: bool, +) -> Result { + if max_size == 0 { + return Err(Error::Io(format!( + "postgres: max_size must be positive, got {max_size}" + ))); + } + let mut cfg: tokio_postgres::Config = config.parse().map_err(err)?; + let engine_opt = format!( + "-c synchronous_commit={}", + if synchronous_commit { "on" } else { "off" } + ); + let merged = match cfg.get_options() { + Some(existing) => format!("{existing} {engine_opt}"), + None => engine_opt, + }; + cfg.options(merged); + let manager = Manager::from_config( + cfg, + NoTls, + ManagerConfig { + recycling_method: RecyclingMethod::Fast, + }, + ); + Pool::builder(manager) + .max_size(max_size) + .build() + .map_err(err) +} + +/// Construction options shared by the crate's postgres-attached engines +/// (`postgres` kv, `pg-lo` large) — the PgOpts design point resolved: +/// one type, its public re-export gate widened to +/// `any(postgres, pg-lo)` alongside the definition's move to this +/// module (ADR-013 §4's pg-lo-pure build carries the same opts surface). +/// Durability knobs, pool sizing, and the engine-owned schema name are +/// engine-configuration concerns — they ride the store constructors' +/// parameters, never invented defaults (the alkstore `PgOpts` +/// precedent; `open(config, PgOpts)` adapted to the named-constructor +/// shape — the constructors are [`crate::Store::dual_tier_postgres_local`] +/// and [`crate::Store::kv_only_postgres`]). +/// +/// The connection settings themselves do not ride this struct — they +/// are the constructor's separate connection-settings argument (the +/// tokio-postgres [`tokio_postgres::Config`]-parseable form); the +/// engine never invents defaults for them. TLS: the engines hardwire +/// `NoTls` on every connection path regardless of the config's +/// sslmode — v1 TLS is a post-v1 deployment concern, mirroring the +/// alkstore posture. +#[derive(Debug, Clone)] +pub struct PgOpts { + /// The engine-owned PostgreSQL schema holding the engine's tables. + /// Both engines default to [`DEFAULT_SCHEMA`] (the consolidation + /// shape holds both engines' tables in one schema); an empty string + /// selects the default. The name is a quoted-identifier boundary + /// ([`quote_identifier`]) — reserved-word and hostile names are + /// safe. + pub schema: String, + /// Upper bound on pooled connections (deadpool's `max_size`). + /// Default [`DEFAULT_MAX_SIZE`] = 8. A value of `0` fails + /// construction immediately (validated at entry, no round trip — + /// deadpool cannot grant checkouts from an empty pool; the + /// alkstore precedent). + pub max_size: usize, + /// The per-session `synchronous_commit` durability knob, wired as a + /// connect option (`-c synchronous_commit=…`) on every pooled + /// connection (POC-verified per-session SET mechanics; the + /// consumer's own `options=` DSN values are preserved alongside — + /// postgres parses repeated `-c` flags). Default `true` — the ship + /// config; `false` is the documented sqlite-NORMAL-alike latency + /// trade (B5 ~2-3×). Both engines carry the knob; the pg-lo POC ran + /// its server with `synchronous_commit=off` to measure LO shape + /// without disk fsync dominating (POC #7), and the shipped knob + /// posture is the kv engine's parity default `on`. + pub synchronous_commit: bool, +} + +impl Default for PgOpts { + fn default() -> Self { + Self { + schema: DEFAULT_SCHEMA.to_owned(), + max_size: DEFAULT_MAX_SIZE, + synchronous_commit: true, + } + } +} + +/// Test-support helpers shared with the crate's postgres-attached test +/// modules (`src/postgres.rs`'s and `src/pg_lo.rs`'s test suites, +/// `src/store.rs`'s construction tests) — crate-internal and compiled +/// out of every non-test build. Each janitor is gated by the feature +/// whose tests consume it. +#[cfg(test)] +pub(crate) mod test_support { + use super::quote_identifier; + use tokio_postgres::NoTls; + + /// Drops a test's engine-owned schema when the test ends — success, + /// early return, or unwind (Drop runs in all three), so a fresh + /// schema per test never accumulates on the shared harness + /// database. No async teardown exists at Drop, so the drop executes + /// synchronously from a std thread over a one-shot current-thread + /// runtime; an unreachable harness server is tolerated (the residue + /// stays; nothing else breaks). + /// + /// Construct it (or a wrapper holding it) *before* the store or + /// backend it outlives: drop order is reverse of declaration, so + /// the backend's pool drains before the schema drops. + #[cfg(feature = "postgres")] + pub(crate) struct SchemaJanitor { + dsn: String, + schema: String, + } + + #[cfg(feature = "postgres")] + impl SchemaJanitor { + pub(crate) fn new(dsn: &str, schema: &str) -> SchemaJanitor { + SchemaJanitor { + dsn: dsn.to_owned(), + schema: schema.to_owned(), + } + } + } + + #[cfg(feature = "postgres")] + impl Drop for SchemaJanitor { + fn drop(&mut self) { + let dsn = self.dsn.clone(); + let schema = self.schema.clone(); + let _ = std::thread::spawn(move || { + let Ok(rt) = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + else { + return; + }; + rt.block_on(async move { + let Ok((client, conn)) = tokio_postgres::connect(&dsn, NoTls).await else { + return; + }; + tokio::spawn(conn); + let sql = format!( + "DROP SCHEMA IF EXISTS {} CASCADE", + quote_identifier(&schema) + ); + let _ = client.execute(sql.as_str(), &[]).await; + }); + }) + .join(); + } + } + + /// Drops a test's dedicated database when the test ends — the + /// pg-lo engine tests' janitor. Large Objects are **database-level** + /// (the C7 lesson, recorded design point): the schema-per-test + /// convention does not isolate LOs across tests sharing a database, + /// and a `DROP SCHEMA` leaves the LO's pages in `pg_largeobject` — + /// so the pg-lo tests isolate per dedicated database (the POC's + /// proven `open_fresh` shape) and this guard removes it, zero LO + /// residue on the shared harness database by construction. + /// Same Drop mechanics as [`SchemaJanitor`]; `WITH (FORCE)` reaps + /// any connection the pool has not yet drained (postgres ≥ 13). + #[cfg(feature = "pg-lo")] + pub(crate) struct DatabaseJanitor { + admin_dsn: String, + db: String, + } + + #[cfg(feature = "pg-lo")] + impl DatabaseJanitor { + /// `admin_dsn` must name a database to connect *through* (the + /// `postgres` maintenance database); `db` is the test's + /// dedicated database to drop. + pub(crate) fn new(admin_dsn: &str, db: &str) -> DatabaseJanitor { + DatabaseJanitor { + admin_dsn: admin_dsn.to_owned(), + db: db.to_owned(), + } + } + } + + #[cfg(feature = "pg-lo")] + impl Drop for DatabaseJanitor { + fn drop(&mut self) { + let admin_dsn = self.admin_dsn.clone(); + let db = self.db.clone(); + let _ = std::thread::spawn(move || { + let Ok(rt) = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + else { + return; + }; + rt.block_on(async move { + let Ok((client, conn)) = tokio_postgres::connect(&admin_dsn, NoTls).await + else { + return; + }; + tokio::spawn(conn); + let sql = format!( + "DROP DATABASE IF EXISTS {} WITH (FORCE)", + quote_identifier(&db) + ); + let _ = client.execute(sql.as_str(), &[]).await; + }); + }) + .join(); + } + } +} diff --git a/src/pg_lo.rs b/src/pg_lo.rs index 9c4785c..23e13d1 100644 --- a/src/pg_lo.rs +++ b/src/pg_lo.rs @@ -1,2 +1,2056 @@ -//! The `pg-lo` large-tier engine (ADR-009): postgres Large Objects -//! storage with descriptorless `lo_get` read windows over companion rows. +//! The `pg-lo` large-tier engine (ADR-009): postgres Large Objects hold +//! the bytes; the companion table +//! `lo_entries (key bytea PRIMARY KEY, loid oid, size bigint, +//! committed_at)` is the **contract authority** — descriptorless +//! `lo_get(oid, off, len)` read windows over companion rows, one- +//! transaction puts (lo_create → chunked `lowrite` → companion-row +//! insert → COMMIT), and a held-transaction staging cursor. Feature +//! `pg-lo`, **default-off** (ADR-013 §4: `pg-lo` does not imply +//! `postgres` — the pg-lo-pure build compiles this module without +//! `feature = "postgres"`). The driver posture is `src/postgres.rs`'s +//! structural twin (deadpool pool over tokio-postgres, per-connection +//! `prepare_cached` discipline — statements die with sessions — +//! natively async, no `spawn_blocking` seam, no blocking I/O on the +//! async path); the large-tier bodies are `src/local.rs`'s streaming +//! twin (streaming read cursor, staged-put hygiene shapes). +//! +//! **Engine shape (POC #7 C1 is the contract).** Two artifacts: LOs +//! hold the bytes; the companion table is the addressing authority — +//! `list`, `size`, `has`, and the CAS existence probe are all +//! table-level. `pg_largeobject` is **never read for addressing**; the +//! sole sanctioned read of it is the orphan-recovery sweep's candidate +//! enumeration. `get`'s cursor closes over the companion row (oid + +//! length; content immutable, so the row is authoritative) — never +//! over a tx-scoped descriptor. +//! +//! **`lo_get` windows are the shipped get shape (ADR-009 §3, POC #7 +//! C4).** Range reads are single stateless statements, pool-friendly +//! (handle-acquire p50 0.7–1.6 ms, p99 1.5–5.6 ms at ≤8 readers); a +//! read beyond the cursor's length clamps at EOF per the cursor +//! contract, and a window that comes up short *below* the length the +//! companion row claims (a corrupted half-state, which healthy stores +//! never hold) fails cleanly with an unexpected-EOF-shaped +//! [`Error::Io`]. **The held-descriptor tx posture exists as the +//! fallback only** — measured (C4): no advantage even for sequential +//! full reads, and it pins one pool connection per open handle — named +//! here as not-shipped, never coded. +//! +//! **The 64-bit-variants discipline (POC #7 C7's trap not inherited).** +//! Every offset this engine issues is 64-bit end-to-end: `lo_get`'s +//! offset parameter is `bigint` by definition, and the engine performs +//! **no int4 positioning at all** — there is no `lo_lseek` anywhere +//! (its int4 offset caps at 2 GiB; the large tier serves larger), and +//! `lowrite` rides the descriptor's implicit sequential position (one +//! writer per LO, appends only; no repositioning is ever needed). The +//! same discipline makes the descriptor a non-requirement for staged +//! writes: the descriptor advances with each `lowrite`, sequential by +//! construction. +//! +//! **512 KiB statements (POC #7 C2's measured knee).** The put cost is +//! statement count, not LOBLKSIZE (catalog rows are ceil(len/2048) +//! regardless of transport chunking): `lowrite` sweeps are flat at +//! 39–43 MB/s from 512 KiB chunks and collapse below (6 MB/s @8 KiB). +//! [`LOWRITE_CHUNK`] is the named shipped constant (C2's citation), +//! covering the whole-value put directly and the staging cursor's +//! append aggregation (the store core's put loop hands +//! `crate::dispatch::PUT_CHUNK`-sized 64 KiB windows; the engine +//! aggregates appends into 512 KiB `lowrite` statements). [`READ_CHUNK`] +//! bounds the `lo_get` window: larger windows don't beat the +//! server-side per-statement ceiling (~65–70 MB/s per in-flight +//! statement, C2/C3) and the int4 length parameter caps at 2 GiB. +//! +//! **Put = one transaction (ADR-009 §2).** BEGIN → `lo_create` → +//! chunked `lowrite` → companion-row insert → COMMIT, on one exclusive +//! pooled checkout. CAS first-writer-wins: the companion-table +//! existence check runs first — a re-put of an existing key is a +//! no-op tx (COMMIT of an empty transaction), never an overwrite, and +//! never a `lo_create` (no LO is born for a losing put). +//! +//! **Staged put: the held-transaction staging cursor (the engine-level +//! `put_staged` contract).** The handout opens the checkout + the tx + +//! `lo_create`; `append` rides the chunked `lowrite` aggregation; the +//! verification pass reads the staged bytes back through the read seam +//! *inside the writing tx* (read-your-own-writes: the LO's pages are +//! visible to `lo_get` from the tx that wrote them); `commit` flushes +//! the aggregation buffer, re-checks CAS existence (a re-put of an +//! existing key ROLLBACKs the staged LO away and publishes nothing — +//! first-writer-wins, `Ok`), inserts the companion row, and COMMITs; +//! `discard` ROLLBACKs. The row and the LO's pages land in one step, +//! so the crash story is structurally clean (C6: zero orphans) — and +//! every failure path converges on ROLLBACK: a failed commit +//! self-discards (there is no post-failure discard to drive), an +//! explicit discard removes LO and row-to-be together (no LO pages, no +//! row), and a cursor dropped without either is best-effort rolled +//! back by [`Drop`] (a spawned ROLLBACK when a runtime is live — +//! deadpool's `Fast` recycle never resets transaction state, so a +//! dirty checkout must never be returned silently). The held tx pins +//! one pool connection per active stage — the held-tx posture's named +//! cost, bounded by the pool size. `#[cfg(test)] fail_commit` +//! injection stands in for the media failure (the +//! mem/sqlite/postgres pattern). **Scope boundary:** the *fleet* joint +//! form (wave 6's store-held tx, ADR-012 §4/§5's ruling that pg-lo +//! contributes LO writes + companion-row insert into it) is a wave-6 +//! design point; this is the engine-level staging cursor only. +//! +//! **Delete = one transaction** (`lo_unlink` + companion-row delete) +//! with the rows-affected discipline (`Err(Error::Missing)` when +//! absent); GC-participating — the delete window's arbitration is +//! store-core and table-level, nothing fs-specific (ADR-009 §4). `list` +//! is `SELECT key FROM lo_entries` — complete by contract, the POC's +//! exact-count gate (list == companion table after mixed ops). +//! +//! **`size` is metadata only (ADR-008's discipline):** the companion +//! row's `size` column — never LO content. The cursor's `len()` reads +//! the same column at `get` time. +//! +//! **Virgin-store posture.** The constructor creates the engine-owned +//! schema and the companion table (idempotent DDL at the constructor; +//! no migration machinery); schema-less reads map to absent/empty via +//! the SQLSTATE 42P01 discriminant (the shared +//! `crate::pg_common::is_undefined_table`), schema-less writes surface +//! the engine error — the recorded read-only split. +//! +//! **The shared-helper design point (resolved).** The postgres engine's +//! crate-internal helpers now live in [`crate::pg_common`], gated +//! `cfg(any(feature = "postgres", feature = "pg-lo"))`, one module per +//! file — the pg-lo-pure build compiles without duplication: +//! [`quote_identifier`], the 42P01 discriminant, the error mapping, the +//! i64-nanos timestamp encoding, the pool construction, and [`PgOpts`]. +//! **PgOpts is shared** — the widening (its public re-export gate moved +//! from `postgres` to `any(postgres, pg-lo)`) was chosen over a +//! pg-lo-local opts type: the knobs are identical, the public surface +//! (one opts type per consumer story) stays single, and ADR-013 §4's +//! pg-lo-pure build carries the same opts type. The +//! `synchronous_commit` parity knob **carries**: the shipped default is +//! `on` (the kv engine's ship config); the POC #7 runs set it +//! server-side `off` to measure LO *shape* without disk fsync +//! dominating — a measurement posture, not the shipped posture. +//! +//! **LO isolation is database-level (the C7 lesson, resolved design +//! point).** Large Objects live per-database, not per-schema: the +//! schema-per-test convention does not isolate them (a `DROP SCHEMA` +//! leaves the LO's pages in `pg_largeobject`, and shared-database tests +//! race each other into spurious "large object NNNN does not exist" +//! aborts). The tests therefore run on **per-test dedicated databases** +//! (the POC's proven `open_fresh` shape): each test `CREATE DATABASE`s +//! a unique database, builds the engine inside it, and a +//! `DatabaseJanitor` drop-guard (`WITH (FORCE)`) removes the database. +//! The shared harness database (`blobs`) receives **zero LO residue by +//! construction** — no pg-lo test ever opens a pool against it. +//! +//! **Posture deltas, stated not hidden (ADR-009 §5/§6).** LO catalog +//! space is reused but never returned — the engine's inverted-fs +//! property: deleted LOs' pages free at delete-commit and are reused by +//! later puts, but `pg_largeobject`'s rel file only shrinks by +//! `VACUUM FULL`/`pg_repack` — an **operator runbook line, never +//! automatic**; monitor with `pg_total_relation_size('pg_largeobject')`. +//! Autovacuum is inherited — `pg_largeobject` is an ordinary catalog +//! heap autovacuum processes; keep it on (a deployment disabling it +//! gets the operator-visible note). Capacity facts, named not solved: +//! single-stream cached get runs 20–50× behind page-cache local fs at +//! ≥1 MiB (server-side LO page-walk + copy ceiling, ~65–70 MB/s per +//! in-flight statement); the fleet serving picture aggregates ~700 MB/s +//! at 16 readers (server-side parallelism) — the REQ-2-relevant number; +//! durable puts land 60–65 MB/s at ≥1 MiB (within 5–10% of durable +//! local fs). Deployments dominated by single-stream page-cache-bound +//! reads stay better served by `local`-on-shared-media. +//! +//! **The crate-internal orphan-recovery sweep** ([`PgLoBackend::reap_orphans`], +//! ADR-009 §5): for the legacy/bypass class only — committed LOs +//! without companion rows. Crash orphans are structurally zero (C6: +//! LO lifecycle is transactional in modern postgres; kill mid-tx leaves +//! zero pages). Candidates enumerate `SELECT DISTINCT loid FROM +//! pg_largeobject` as a **candidate list only**; each is validated +//! against `lo_entries.loid` (the truth) and only unclaimed oids are +//! unlinked — cost scales with oid count, not content size (C6: +//! ~7 ms/oid). Scope-noted under the module-level allow until a +//! consumer exists (the ops/maintenance surface is wave 7+; no +//! store-core verb is scoped this wave); contract-tested with a planted +//! orphan and a byte-exact survival assertion for committed content. +//! +//! **Errors.** The engine maps its failures onto the store error +//! family: media/session/pool failure is [`Error::Io`], and +//! [`Error::Missing`] is the rows-affected delete mapping and the +//! virgin-store read-only split — no panics, no unwrap outside tests. +//! 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. +//! +//! 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`, +//! user/password `postgres`/`poc` — verified live 2026-10-12). + +#![allow(dead_code)] + +use std::time::SystemTime; + +use deadpool_postgres::{Object, Pool}; + +use crate::backend::{clamped, Backend, ReadCursor, StagingCursor}; +use crate::pg_common::{ + build_pool, err, is_undefined_table, pg_err, qualified_table, quote_identifier, size_u64, + time_to_nanos, PgOpts, DEFAULT_SCHEMA, +}; +use crate::Error; + +/// The `lowrite` statement chunk — POC #7 C2's measured knee (512 KiB +/// statements; statement count, not LOBLKSIZE, is what costs: 39–43 +/// MB/s flat from here, 6 MB/s @8 KiB). The store core's put loop hands +/// 64 KiB windows; both put bodies aggregate into statements of this +/// size. +pub(crate) const LOWRITE_CHUNK: usize = 512 * 1024; + +/// The `lo_get` window bound — one read statement's maximum length +/// (POC #7 C2/C3: larger windows don't beat the server-side per-statement +/// ceiling; also keeps the int4 `len` parameter in domain). +pub(crate) const READ_CHUNK: usize = 512 * 1024; + +/// `lo_open`'s write mode (the libpq `INV_WRITE` bit). +const INV_WRITE: i32 = 0x0002_0000; + +/// The `pg-lo` engine — a deadpool pool over tokio-postgres plus the +/// quoted engine-owned companion-table reference. Cloning shares the +/// pool; the pool is the concurrency posture. +pub(crate) struct PgLoBackend { + pool: Pool, + /// The engine-owned companion-table reference, quoted once at + /// construction: `"schema"."lo_entries"` — the quoted-identifier + /// boundary every statement composes through. + table: String, + /// The engine-owned schema name (the test-only catalog probes'). + schema: String, + /// Test-only commit-failure injection, standing in for the media + /// failure the large tier cannot reliably produce in tests; mirrors + /// the mem/sqlite/postgres/local knobs so the stage-hygiene failure + /// form is exercised against the real engine. + #[cfg(test)] + fail_commit: bool, +} + +impl PgLoBackend { + /// Open (or create) the engine-owned schema and companion table and + /// establish the shipped posture: the deadpool pool + /// (`RecyclingMethod::Fast`, the shared [`build_pool`] posture) with + /// the per-session `synchronous_commit` knob ([`PgOpts`], shared + /// with the postgres engine; default `on`), then the idempotent + /// bootstrap DDL on a pool connection. Construction fails — never + /// silently downgrades posture — when the server is unreachable, + /// the credentials/config are unparseable, the pool cannot be built + /// (`max_size = 0` fails at entry), or the DDL errors. + pub(crate) async fn new(config: &str, opts: PgOpts) -> Result { + let pool = build_pool(config, opts.max_size, opts.synchronous_commit)?; + let schema = if opts.schema.is_empty() { + DEFAULT_SCHEMA.to_owned() + } else { + opts.schema + }; + { + let client = pool.get().await.map_err(err)?; + client + .batch_execute(&bootstrap_ddl(&schema)) + .await + .map_err(pg_err)?; + } + Ok(Self { + pool, + table: qualified_table(&schema, "lo_entries"), + schema, + #[cfg(test)] + fail_commit: false, + }) + } + + /// Open a pool without schema creation — the virgin-store tolerance + /// test's instrument (the postgres engine's `new_schemaless` + /// pattern): the read paths' no-schema-yet semantics are tested + /// against a schema that was never bootstrapped. + #[cfg(test)] + async fn new_schemaless(config: &str, schema: &str) -> Result { + let pool = build_pool(config, crate::pg_common::DEFAULT_MAX_SIZE, true)?; + Ok(Self { + pool, + table: qualified_table(schema, "lo_entries"), + schema: schema.to_owned(), + #[cfg(test)] + fail_commit: false, + }) + } + + /// The catalog table-existence probe — test-only: the bootstrap + /// test's instrument, answering for the engine-owned schema the + /// backend was constructed with. + #[cfg(test)] + pub(crate) async fn table_exists(&self, table: &str) -> Result { + let client = self.pool.get().await.map_err(err)?; + let row = client + .query_one( + "SELECT EXISTS ( + SELECT 1 FROM pg_tables WHERE schemaname = $1 AND tablename = $2 + )", + &[&self.schema, &table], + ) + .await + .map_err(pg_err)?; + let present: bool = row.try_get(0).map_err(pg_err)?; + Ok(present) + } + + /// The crate-internal orphan-recovery sweep (ADR-009 §5): reap the + /// legacy/bypass class — committed LOs without companion rows. One + /// tx: candidates enumerate `pg_largeobject`'s distinct oids + /// (candidate list only), each validated against + /// `lo_entries.loid`, the unclaimed unlinked. Returns the count + /// reaped. See the module doc for the scope note and the C6 + /// evidence. + pub(crate) async fn reap_orphans(&self) -> Result { + let client = self.pool.get().await.map_err(err)?; + client.batch_execute("BEGIN").await.map_err(pg_err)?; + let reaped = self.reap_orphans_tx(&client).await; + match reaped { + Ok(n) => match client.batch_execute("COMMIT").await { + Ok(()) => Ok(n), + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + Err(pg_err(e)) + } + }, + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + Err(e) + } + } + } + + async fn reap_orphans_tx(&self, client: &Object) -> Result { + let rows = client + .query("SELECT DISTINCT loid FROM pg_largeobject", &[]) + .await + .map_err(pg_err)?; + let claim_sql = format!("SELECT 1 FROM {} WHERE loid = $1", self.table); + let claim_stmt = client.prepare_cached(&claim_sql).await.map_err(pg_err)?; + let unlink_stmt = client + .prepare_cached("SELECT lo_unlink($1)") + .await + .map_err(pg_err)?; + let mut reaped = 0u64; + for row in &rows { + let loid: u32 = row.try_get(0).map_err(pg_err)?; + let claimed = client + .query_opt(&claim_stmt, &[&loid]) + .await + .map_err(pg_err)? + .is_some(); + if !claimed { + client + .query_one(&unlink_stmt, &[&loid]) + .await + .map_err(pg_err)?; + reaped += 1; + } + } + Ok(reaped) + } + + /// The put's tx body (over the opened transaction): CAS existence + /// (a re-put is a no-op tx — the COMMIT of an empty transaction — + /// never an overwrite, never a `lo_create`), then lo_create → + /// chunked `lowrite` → companion-row insert. Errors bubble to + /// [`Backend::put`]'s ROLLBACK convergence point. + async fn put_in_tx(&self, client: &Object, key: &[u8], value: &[u8]) -> Result<(), Error> { + let exists_sql = format!("SELECT 1 FROM {} WHERE key = $1", self.table); + let exists_stmt = client.prepare_cached(&exists_sql).await.map_err(err)?; + let present = client + .query_opt(&exists_stmt, &[&key]) + .await + .map_err(pg_err)? + .is_some(); + if present { + return Ok(()); + } + let loid = lo_create(client).await?; + let fd = lo_open(client, loid, INV_WRITE).await?; + lowrite_all(client, fd, value).await?; + let size = value.len() as i64; + let committed_at = time_to_nanos(SystemTime::now()); + let insert_sql = format!( + "INSERT INTO {} (key, loid, size, committed_at) VALUES ($1, $2, $3, $4)", + self.table + ); + let insert_stmt = client.prepare_cached(&insert_sql).await.map_err(err)?; + client + .execute(&insert_stmt, &[&key, &loid, &size, &committed_at]) + .await + .map_err(pg_err)?; + Ok(()) + } +} + +/// The engine-owned bootstrap DDL — idempotent, no migration machinery. +/// The C1 companion-table shape; column names quote-follow the alkstore +/// hazard-list discipline (`"key"` quoted; `loid`/`size`/`committed_at` +/// non-reserved). `committed_at` is the crate-wide i64-nanos encoding +/// (`bigint`), engine-stamped at insert — not a server-default column. +fn bootstrap_ddl(schema: &str) -> String { + let schema_q = quote_identifier(schema); + let table = qualified_table(schema, "lo_entries"); + format!( + "CREATE SCHEMA IF NOT EXISTS {schema_q}; + +CREATE TABLE IF NOT EXISTS {table} ( + \"key\" bytea PRIMARY KEY, + loid oid NOT NULL, + size bigint NOT NULL, + committed_at bigint NOT NULL +); +" + ) +} + +/// `lo_create` — a fresh LO oid inside the caller's tx. +async fn lo_create(client: &Object) -> Result { + let stmt = client + .prepare_cached("SELECT lo_create(0) AS loid") + .await + .map_err(err)?; + let row = client.query_one(&stmt, &[]).await.map_err(pg_err)?; + row.try_get(0).map_err(pg_err) +} + +/// `lo_open` — the tx-scoped descriptor for in-tx `lowrite` (reads never +/// hold one; the shipped get shape is descriptorless windows). +async fn lo_open(client: &Object, loid: u32, mode: i32) -> Result { + let stmt = client + .prepare_cached("SELECT lo_open($1, $2) AS fd") + .await + .map_err(err)?; + let row = client + .query_one(&stmt, &[&loid, &mode]) + .await + .map_err(pg_err)?; + row.try_get(0).map_err(pg_err) +} + +/// Chunked `lowrite` — statements of [`LOWRITE_CHUNK`] (the C2 knee); +/// rides the descriptor's implicit sequential position (no repositioning, +/// the 64-bit-variants discipline). Fails on a negative or short count. +async fn lowrite_all(client: &Object, fd: i32, mut data: &[u8]) -> Result<(), Error> { + if data.is_empty() { + return Ok(()); + } + let stmt = client + .prepare_cached("SELECT lowrite($1, $2) AS n") + .await + .map_err(err)?; + while !data.is_empty() { + let take = data.len().min(LOWRITE_CHUNK); + let chunk = &data[..take]; + let n: i32 = client + .query_one(&stmt, &[&fd, &chunk]) + .await + .map_err(pg_err)? + .try_get(0) + .map_err(pg_err)?; + if n < 0 || (n as usize) != take { + return Err(Error::Io(format!( + "postgres: lowrite wrote {n} of {take} staged bytes" + ))); + } + data = &data[take..]; + } + Ok(()) +} + +/// One descriptorless `lo_get` window — the shipped read shape (C4): +/// stateless, pool-friendly, bigint offset. Window length stays within +/// [`READ_CHUNK`] (and the int4 parameter domain). +async fn lo_get_window( + client: &Object, + loid: u32, + offset: u64, + len: usize, +) -> Result, Error> { + let stmt = client + .prepare_cached("SELECT lo_get($1, $2, $3) AS b") + .await + .map_err(err)?; + let row = client + .query_one(&stmt, &[&loid, &(offset as i64), &(len as i32)]) + .await + .map_err(pg_err)?; + row.try_get(0).map_err(pg_err) +} + +/// The streamed read over [`lo_get_window`] windows: the cursor +/// contract's clamp against `body_len` first, then the window loop; a +/// window coming up empty before the body is complete is the +/// unexpected-EOF clean failure (C4's shape — never surfaced for a +/// healthy store, whose companion row is authoritative). +async fn read_windows(client: &Object, loid: u32, start: u64, end: u64) -> Result, Error> { + let wanted = (end - start) as usize; + let mut out = Vec::with_capacity(wanted); + let mut pos = start; + while pos < end { + let window = ((end - pos) as usize).min(READ_CHUNK); + let bytes = lo_get_window(client, loid, pos, window).await?; + if bytes.is_empty() { + return Err(Error::Io(format!( + "pg-lo: large object {loid} ended {} bytes short of the \ + committed window (unexpected EOF, companion row overclaims)", + wanted - out.len() + ))); + } + out.extend_from_slice(&bytes); + pos += bytes.len() as u64; + } + Ok(out) +} + +/// The streaming read cursor (the large body family): descriptorless — +/// it closes over the companion row (oid + length; content immutable, +/// so the row is authoritative), never a tx-scoped descriptor, never a +/// content load at `get` time. `len()` is the companion row's `size` — +/// the same metadata-only fact [`Backend::size`] probes. Range reads +/// are [`lo_get_window`] windows. +pub struct PgLoCursor { + pool: Pool, + loid: u32, + len: u64, +} + +impl ReadCursor for PgLoCursor { + fn len(&self) -> u64 { + self.len + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + let (start, end) = clamped(offset, len, self.len as usize); + if start == end { + return Ok(Vec::new()); + } + let client = self.pool.get().await.map_err(err)?; + read_windows(&client, self.loid, start as u64, end as u64).await + } +} + +/// The held-transaction staging cursor: the handout opens the checkout + +/// the tx + `lo_create` (the exclusive checkout is what carries the tx — +/// no other statement interleaves on the session); `append` rides the +/// chunked `lowrite` aggregation (the flush boundary is +/// [`LOWRITE_CHUNK`]; bytes below it sit in the aggregation buffer and +/// serve the read seam from memory until flushed); the verification pass +/// reads staged bytes back through the read seam inside the writing tx +/// (read-your-own-writes over `lo_get` + the unflushed tail); +/// `commit` publishes (row insert + COMMIT — row and pages land in one +/// step, C6's zero-orphan shape); every other path — explicit +/// `discard`, a failed commit, the `Drop` cancellation story — +/// converges on ROLLBACK with zero residue (no LO pages, no row). +pub struct PgLoStaging { + /// The exclusive checkout carrying the held tx; `Some` exactly while + /// the cursor is live (consumed paths take it before returning). + client: Option, + loid: u32, + fd: i32, + table: String, + /// Bytes flushed to the LO via `lowrite` — the read seam's LO part + /// boundary; the aggregation buffer holds everything above it. + flushed_len: u64, + pending: Vec, + #[cfg(test)] + fail_commit: bool, +} + +impl PgLoStaging { + /// The live checkout — always `Some` on a live cursor; the `None` + /// arm is the defensively-typed consumed state, never a panic. + fn client(&self) -> Result<&Object, Error> { + self.client + .as_ref() + .ok_or_else(|| Error::Io("pg-lo: staging cursor already consumed".to_owned())) + } + + /// Flush full [`LOWRITE_CHUNK`] blocks out of the aggregation buffer + /// (the store core's 64 KiB put-loop windows aggregate here into the + /// C2 knee's statement size). On failure the buffer is intact and + /// `flushed_len` unclaimed — the caller converges on ROLLBACK. + async fn flush_blocks(&mut self) -> Result<(), Error> { + while self.pending.len() >= LOWRITE_CHUNK { + let block = self.pending[..LOWRITE_CHUNK].to_vec(); + lowrite_all(self.client()?, self.fd, &block).await?; + self.pending.drain(..LOWRITE_CHUNK); + self.flushed_len += LOWRITE_CHUNK as u64; + } + Ok(()) + } + + /// Flush the aggregation tail (called by publish over the taken + /// checkout; the tail is < [`LOWRITE_CHUNK`]). + async fn flush_tail(&mut self, client: &Object) -> Result<(), Error> { + if self.pending.is_empty() { + return Ok(()); + } + let tail = std::mem::take(&mut self.pending); + lowrite_all(client, self.fd, &tail).await?; + self.flushed_len += tail.len() as u64; + Ok(()) + } + + /// The one ROLLBACK convergence point: every non-publish path runs + /// this — the staged LO's pages vanish with the tx, zero residue. + async fn rollback(mut self) { + if let Some(client) = self.client.take() { + let _ = client.batch_execute("ROLLBACK").await; + } + } +} + +impl ReadCursor for PgLoStaging { + fn len(&self) -> u64 { + self.flushed_len + self.pending.len() as u64 + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + let (start, end) = clamped(offset, len, self.len() as usize); + if start == end { + return Ok(Vec::new()); + } + let client = self.client()?; + let lo_end = self.flushed_len.min(end as u64) as usize; + let mut out = Vec::with_capacity(end - start); + if (start as u64) < self.flushed_len { + out.extend(read_windows(client, self.loid, start as u64, lo_end as u64).await?); + } + if end > lo_end { + let pending_start = (start.max(lo_end)) - lo_end; + out.extend_from_slice(&self.pending[pending_start..(end - lo_end)]); + } + Ok(out) + } +} + +impl StagingCursor for PgLoStaging { + async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> { + if chunk.is_empty() { + return Ok(()); + } + self.pending.extend_from_slice(chunk); + self.flush_blocks().await + } + + /// Publish: flush the tail, CAS existence against the companion + /// table (a re-put of an existing key ROLLBACKs the staged LO away + /// and returns `Ok` — first-writer-wins dedup, never an overwrite), + /// insert the companion row, COMMIT. Every failure path — including + /// the injected one — converges on `rollback` (self-discard; the + /// consuming call took the cursor). + async fn commit(mut self, key: &[u8]) -> Result<(), Error> { + #[cfg(test)] + if self.fail_commit { + self.rollback().await; + return Err(Error::Io( + "engine commit failure (pg-lo, injected)".to_owned(), + )); + } + let client = match self.client.take() { + Some(client) => client, + None => { + return Err(Error::Io( + "pg-lo: staging cursor already consumed".to_owned(), + )) + } + }; + let result = self.publish(&client, key).await; + match result { + Ok(true) => match client.batch_execute("COMMIT").await { + Ok(()) => Ok(()), + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + Err(pg_err(e)) + } + }, + Ok(false) => { + let _ = client.batch_execute("ROLLBACK").await; + Ok(()) + } + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + Err(e) + } + } + } + + /// Discard the stage: ROLLBACK — the staged LO's pages vanish with + /// the tx, no row existed to remove. Zero residue. + async fn discard(self) { + self.rollback().await; + } +} + +impl PgLoStaging { + /// The publish body (over the taken checkout): tail flush, CAS + /// existence against the companion table (a re-put returns + /// `Ok(false)` — first-writer-wins dedup, commit ROLLBACKs the + /// staged LO away, never an overwrite), then the companion-row + /// insert (`Ok(true)`: commit proceeds). Errors bubble to + /// [`StagingCursor::commit`]'s ROLLBACK convergence point. + async fn publish(&mut self, client: &Object, key: &[u8]) -> Result { + self.flush_tail(client).await?; + let exists_sql = format!("SELECT 1 FROM {} WHERE key = $1", self.table); + let exists_stmt = client.prepare_cached(&exists_sql).await.map_err(err)?; + let present = client + .query_opt(&exists_stmt, &[&key]) + .await + .map_err(pg_err)? + .is_some(); + if present { + return Ok(false); + } + let size = self.flushed_len as i64; + let committed_at = time_to_nanos(SystemTime::now()); + let insert_sql = format!( + "INSERT INTO {} (key, loid, size, committed_at) VALUES ($1, $2, $3, $4)", + self.table + ); + let insert_stmt = client.prepare_cached(&insert_sql).await.map_err(err)?; + client + .execute(&insert_stmt, &[&key, &self.loid, &size, &committed_at]) + .await + .map_err(pg_err)?; + Ok(true) + } +} + +impl Drop for PgLoStaging { + /// Best-effort residue cleanup for the cancellation story: a cursor + /// dropped without reaching `commit` or `discard` holds an open tx + /// on a pooled session — hand it to a spawned ROLLBACK (the POC's + /// `release_client` shape) so the checkout never returns dirty; + /// with no runtime live, degrade to a one-shot thread doing the + /// same (deadpool's `Fast` recycle never resets transaction state — + /// the recorded pitfall). Cursors reaching `commit`/`discard` have + /// already taken the checkout; this is a no-op on those paths. + fn drop(&mut self) { + if let Some(client) = self.client.take() { + match tokio::runtime::Handle::try_current() { + Ok(handle) => { + handle.spawn(async move { + let _ = client.batch_execute("ROLLBACK").await; + }); + } + Err(_) => { + let _ = std::thread::spawn(move || { + let Ok(rt) = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + else { + return; + }; + rt.block_on(async move { + let _ = client.batch_execute("ROLLBACK").await; + }); + }) + .join(); + } + } + } + } +} + +impl Backend for PgLoBackend { + type Cursor = PgLoCursor; + type Staging = PgLoStaging; + + fn name(&self) -> &'static str { + "pg-lo" + } + + /// The companion-table EXISTS probe — never a value load, never LO + /// content. Virgin store: `false`. + async fn has(&self, key: &[u8]) -> Result { + let client = self.pool.get().await.map_err(err)?; + let sql = format!("SELECT EXISTS(SELECT 1 FROM {} WHERE key = $1)", self.table); + let stmt = match client.prepare_cached(&sql).await { + Ok(stmt) => stmt, + Err(e) if is_undefined_table(&e) => return Ok(false), + Err(e) => return Err(err(e)), + }; + let present = client + .query_one(&stmt, &[&key]) + .await + .map_err(pg_err)? + .try_get::<_, bool>(0) + .map_err(pg_err)?; + Ok(present) + } + + /// The descriptorless cursor body: the companion row is read once + /// (oid + size — metadata only); the bytes stream through + /// [`lo_get_window`] windows afterwards. Virgin store: `None`. + async fn get(&self, key: &[u8]) -> Result, Error> { + let client = self.pool.get().await.map_err(err)?; + let sql = format!("SELECT loid, size FROM {} WHERE key = $1", self.table); + let stmt = match client.prepare_cached(&sql).await { + Ok(stmt) => stmt, + Err(e) if is_undefined_table(&e) => return Ok(None), + Err(e) => return Err(err(e)), + }; + match client.query_opt(&stmt, &[&key]).await.map_err(pg_err)? { + Some(row) => { + let loid = row.try_get::<_, u32>(0).map_err(pg_err)?; + let len = size_u64(row.try_get::<_, i64>(1).map_err(pg_err)?); + Ok(Some(PgLoCursor { + pool: self.pool.clone(), + loid, + len, + })) + } + None => Ok(None), + } + } + + /// The whole-value put — one tx (BEGIN → CAS existence → `lo_create` + /// → chunked `lowrite` → companion-row insert → COMMIT), CAS + /// first-writer-wins: an existing key is a no-op tx (nothing is + /// staged, no LO is born), never an overwrite. + async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> { + let client = self.pool.get().await.map_err(err)?; + client.batch_execute("BEGIN").await.map_err(pg_err)?; + match self.put_in_tx(&client, key, value).await { + Ok(()) => match client.batch_execute("COMMIT").await { + Ok(()) => Ok(()), + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + Err(pg_err(e)) + } + }, + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + Err(e) + } + } + } + + /// The held-tx stage's handout: checkout + BEGIN + `lo_create` + + /// `lo_open(INV_WRITE)`. Every failure inside the handout converges + /// on ROLLBACK before returning the checkout. + async fn put_staged(&self) -> Result { + let client = self.pool.get().await.map_err(err)?; + client.batch_execute("BEGIN").await.map_err(pg_err)?; + let loid = match lo_create(&client).await { + Ok(loid) => loid, + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + return Err(e); + } + }; + let fd = match lo_open(&client, loid, INV_WRITE).await { + Ok(fd) => fd, + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + return Err(e); + } + }; + Ok(PgLoStaging { + client: Some(client), + loid, + fd, + table: self.table.clone(), + flushed_len: 0, + pending: Vec::new(), + #[cfg(test)] + fail_commit: self.fail_commit, + }) + } + + /// Unconditional delete — one tx (`lo_unlink` + companion-row + /// delete), rows-affected discipline. Virgin store (42P01): + /// `Err(Error::Missing)`. + async fn delete(&self, key: &[u8]) -> Result<(), Error> { + let client = self.pool.get().await.map_err(err)?; + client.batch_execute("BEGIN").await.map_err(pg_err)?; + let loid_sql = format!("SELECT loid FROM {} WHERE key = $1", self.table); + let loid_stmt = match client.prepare_cached(&loid_sql).await { + Ok(stmt) => stmt, + Err(e) if is_undefined_table(&e) => { + let _ = client.batch_execute("ROLLBACK").await; + return Err(Error::Missing); + } + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + return Err(err(e)); + } + }; + let row = match client.query_opt(&loid_stmt, &[&key]).await { + Ok(row) => row, + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + return Err(pg_err(e)); + } + }; + let Some(row) = row else { + let _ = client.batch_execute("ROLLBACK").await; + return Err(Error::Missing); + }; + let loid: u32 = match row.try_get(0) { + Ok(loid) => loid, + Err(e) => { + let e = pg_err(e); + let _ = client.batch_execute("ROLLBACK").await; + return Err(e); + } + }; + let result = async { + let unlink_stmt = client + .prepare_cached("SELECT lo_unlink($1)") + .await + .map_err(pg_err)?; + client + .query_one(&unlink_stmt, &[&loid]) + .await + .map_err(pg_err)?; + let delete_sql = format!("DELETE FROM {} WHERE key = $1", self.table); + let delete_stmt = client.prepare_cached(&delete_sql).await.map_err(err)?; + client + .execute(&delete_stmt, &[&key]) + .await + .map_err(pg_err)?; + Ok(()) + } + .await; + match result { + Ok(()) => match client.batch_execute("COMMIT").await { + Ok(()) => Ok(()), + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + Err(pg_err(e)) + } + }, + Err(e) => { + let _ = client.batch_execute("ROLLBACK").await; + Err(e) + } + } + } + + /// Whole-engine key enumeration — the single-statement MVCC-snapshot + /// cursor over the companion table (the contract authority), ordered + /// by key. Virgin store: empty. + async fn list(&self) -> Result>, Error> { + let client = self.pool.get().await.map_err(err)?; + let sql = format!("SELECT key FROM {} ORDER BY key", self.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, &[]).await.map_err(pg_err)?; + rows.iter() + .map(|row| row.try_get::<_, Vec>(0).map_err(pg_err)) + .collect() + } + + /// The metadata-only length probe: the companion row's `size` + /// column — never LO content, never a value load. Virgin store: + /// `None`. + async fn size(&self, key: &[u8]) -> Result, Error> { + let client = self.pool.get().await.map_err(err)?; + let sql = format!("SELECT size FROM {} WHERE key = $1", self.table); + let stmt = match client.prepare_cached(&sql).await { + Ok(stmt) => stmt, + Err(e) if is_undefined_table(&e) => return Ok(None), + Err(e) => return Err(err(e)), + }; + match client.query_opt(&stmt, &[&key]).await.map_err(pg_err)? { + Some(row) => Ok(Some(size_u64(row.try_get::<_, i64>(0).map_err(pg_err)?))), + None => Ok(None), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::pg_common::test_support::DatabaseJanitor; + use std::ops::Deref; + use std::sync::atomic::{AtomicU64, Ordering}; + use tokio_postgres::NoTls; + + const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog"; + + /// The guarded harness backend: derefs to the engine, and the test's + /// own database is dropped when the guard drops — success, early + /// return, or unwind included (the janitor is declared after the + /// backend, so the pool drains before the database drops). + struct Harness { + /// The per-test database's DSN (test-only probes: e.g. opening a + /// second engine over the same database with hostile opts). + dsn: String, + backend: PgLoBackend, + _janitor: DatabaseJanitor, + } + + impl Deref for Harness { + type Target = PgLoBackend; + + fn deref(&self) -> &PgLoBackend { + &self.backend + } + } + + /// A unique per-test database name — LOs are database-level (the C7 + /// lesson), so every test owns a dedicated database on the shared + /// harness server; the pid + sequence + wall-clock components make + /// the name collision-free across parallel tests and reruns. + static DB_SEQ: AtomicU64 = AtomicU64::new(0); + + fn fresh_db(tag: &str) -> String { + let nanos = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + format!( + "abtlo_{}_{}_{}_{nanos:x}", + tag, + std::process::id(), + DB_SEQ.fetch_add(1, Ordering::Relaxed) + ) + } + + fn env_nonempty(name: &str) -> Option { + std::env::var(name).ok().filter(|v| !v.is_empty()) + } + + /// The harness connection settings ride the environment + /// (`ALKBLOBS_PG_HOST` required; port/user/password/db default to + /// the harness server's values). Absent → the caller skips. + fn harness_dsn() -> Option { + let host = env_nonempty("ALKBLOBS_PG_HOST")?; + Some(format!( + "host={host} port={} user={} password={} dbname={}", + env_nonempty("ALKBLOBS_PG_PORT").unwrap_or_else(|| "15432".to_owned()), + env_nonempty("ALKBLOBS_PG_USER").unwrap_or_else(|| "postgres".to_owned()), + env_nonempty("ALKBLOBS_PG_PASSWORD").unwrap_or_else(|| "poc".to_owned()), + env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()), + )) + } + + /// The maintenance-database DSN the tests connect *through* to + /// create/drop the per-test databases (the same server credentials). + fn harness_admin_dsn() -> Option { + let dsn = harness_dsn()?; + Some(format!("{dsn} dbname=postgres")) + } + + fn dsn_for_db(db: &str) -> String { + let dsn = harness_dsn().expect("harness dsn"); + format!( + "{} dbname={db}", + dsn.rsplit_once("dbname=").expect("dbname").0 + ) + } + + /// The skip note — printed exactly once per skipped test. + fn skip_note() { + eprintln!( + "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB" + ); + } + + /// The POC's proven `open_fresh` shape: create the test's dedicated + /// database (the LO-level isolation decision), then open the engine + /// inside it. No engine state is left on the shared harness + /// database; the janitor in [`Harness`] removes the database. + async fn create_db(admin_dsn: &str, db: &str) { + let (client, conn) = tokio_postgres::connect(admin_dsn, NoTls) + .await + .expect("admin connect"); + tokio::spawn(async move { + let _ = conn.await; + }); + client + .execute(&format!("CREATE DATABASE {}", quote_identifier(db)), &[]) + .await + .expect("create test database"); + } + + async fn harness(tag: &str) -> Option { + harness_with(tag, PgOpts::default()).await + } + + async fn harness_with(tag: &str, opts: PgOpts) -> Option { + let admin = harness_admin_dsn()?; + let db = fresh_db(tag); + create_db(&admin, &db).await; + let dsn = dsn_for_db(&db); + let backend = PgLoBackend::new(&dsn, opts).await.expect("open backend"); + Some(Harness { + dsn, + backend, + _janitor: DatabaseJanitor::new(&admin, &db), + }) + } + + /// The virgin-store fixture: a dedicated database whose schema was + /// never bootstrapped (the postgres engine's `new_schemaless` + /// pattern). The janitor still owns the whole database. + async fn harness_schemaless(tag: &str) -> Option { + let admin = harness_admin_dsn()?; + let db = fresh_db(tag); + create_db(&admin, &db).await; + let dsn = dsn_for_db(&db); + let backend = PgLoBackend::new_schemaless(&dsn, "alkblobs") + .await + .expect("open schemaless backend"); + Some(Harness { + dsn, + backend, + _janitor: DatabaseJanitor::new(&admin, &db), + }) + } + + /// The real stored value, observed through the read seam — the + /// mirror of the postgres engine's helper. + async fn stored_value(backend: &PgLoBackend, key: &[u8]) -> Option> { + let cursor = backend.get(key).await.expect("get succeeds")?; + let (bytes, _) = cursor.read_range(0, cursor.len()).await.expect("read"); + Some(bytes) + } + + async fn staged(backend: &PgLoBackend, key: &[u8], content: &[u8]) -> Result<(), Error> { + let mut staging = backend.put_staged().await?; + staging.append(content).await?; + staging.commit(key).await + } + + async fn staged_in_64k_windows( + backend: &PgLoBackend, + key: &[u8], + content: &[u8], + ) -> Result<(), Error> { + let mut staging = backend.put_staged().await?; + for chunk in content.chunks(64 * 1024) { + staging.append(chunk).await?; + } + staging.commit(key).await + } + + /// The LO catalog's distinct oids — the tests' instrumentation for + /// the zero-residue gates and the catalog-agreement checks. The + /// engine never reads this for addressing; tests may. + async fn catalog_loids(backend: &PgLoBackend) -> Vec { + let client = backend.pool.get().await.expect("checkout"); + let rows = client + .query("SELECT DISTINCT loid FROM pg_largeobject", &[]) + .await + .expect("catalog loids"); + let mut loids: Vec = rows.iter().map(|row| row.get::<_, u32>(0)).collect(); + loids.sort(); + loids + } + + /// The LO catalog's page-row count — the exact-count residue gate + /// (no LO pages, no residue). + async fn catalog_rows(backend: &PgLoBackend) -> i64 { + let client = backend.pool.get().await.expect("checkout"); + client + .query_one("SELECT count(*) FROM pg_largeobject", &[]) + .await + .expect("catalog count") + .get::<_, i64>(0) + } + + /// The companion table's oids, sorted — the truth side of the + /// catalog-agreement check. + async fn companion_loids_sorted(backend: &PgLoBackend) -> Vec { + let client = backend.pool.get().await.expect("checkout"); + let sql = format!("SELECT loid FROM {} ORDER BY loid", backend.table); + let rows = client + .query(sql.as_str(), &[]) + .await + .expect("companion loids"); + rows.iter().map(|row| row.get::<_, u32>(0)).collect() + } + + /// The live session's `synchronous_commit` — the parity-knob + /// posture test's probe (the knob wires per session; a pooled + /// checkout must be asked). + async fn shown_synchronous_commit(backend: &PgLoBackend) -> String { + let client = backend.pool.get().await.expect("checkout"); + let messages = client + .simple_query("SHOW synchronous_commit") + .await + .expect("SHOW"); + for message in &messages { + if let tokio_postgres::SimpleQueryMessage::Row(row) = message { + return row.get(0).expect("SHOW row").to_owned(); + } + } + panic!("SHOW returned no row: {messages:?}") + } + + /// Patterned content large enough to span multiple 512 KiB windows. + fn pattern(len: usize) -> Vec { + (0..len).map(|i| (i % 251) as u8).collect() + } + + /// The LO catalog's page-row count for a content of `len` bytes — + /// ceil(len/2048), C2's catalog math. + fn pages_for(len: usize) -> i64 { + (len as i64 + 2047) / 2048 + } + + /// The Backend method contract: the name, the seven methods' + /// absence/value/error disciplines, CAS first-writer-wins, and + /// virgin-store reads as no-ops — mirroring the postgres engine's + /// suite. + #[tokio::test] + async fn backend_method_contract() { + let Some(backend) = harness("engine").await else { + return; + }; + let key = b"\x01opaque-key-bytes".to_vec(); + assert_eq!(backend.name(), "pg-lo", "reporting identity"); + + assert!( + !backend.has(&key).await.unwrap(), + "virgin store: has is a no-op read" + ); + assert!( + backend.get(&key).await.unwrap().is_none(), + "virgin store: get is a no-op read" + ); + assert_eq!( + backend.size(&key).await.unwrap(), + None, + "virgin store: size is a no-op read" + ); + assert!( + backend.list().await.unwrap().is_empty(), + "virgin store: list is empty" + ); + assert_eq!( + backend.delete(&key).await.unwrap_err(), + Error::Missing, + "virgin store: delete reports Missing" + ); + assert_eq!( + catalog_rows(&backend).await, + 0, + "virgin operations touched no large object pages" + ); + + backend.put(&key, CONTENT).await.unwrap(); + assert!(backend.has(&key).await.unwrap(), "put lands"); + assert_eq!( + backend.size(&key).await.unwrap(), + Some(CONTENT.len() as u64), + "size is metadata-only and exact" + ); + backend.put(&key, b"overwritten").await.unwrap(); + assert_eq!( + stored_value(&backend, &key).await.as_deref(), + Some(CONTENT), + "CAS first-writer wins: re-put is a no-op" + ); + assert_eq!( + backend.list().await.unwrap(), + vec![key.clone()], + "list is complete, keys not values" + ); + assert_eq!(catalog_loids(&backend).await.len(), 1, "one LO exactly"); + assert_eq!( + catalog_rows(&backend).await, + pages_for(CONTENT.len()), + "the LO catalog holds ceil(len/2048) page rows (C2's catalog math)" + ); + + staged_in_64k_windows(&backend, &key, b"overwritten-by-stage") + .await + .expect("staged re-commit is dedup, not failure"); + assert_eq!( + stored_value(&backend, &key).await.as_deref(), + Some(CONTENT), + "staged re-commit does not overwrite" + ); + + backend.delete(&key).await.unwrap(); + assert!(!backend.has(&key).await.unwrap(), "delete lands"); + assert_eq!( + backend.delete(&key).await.unwrap_err(), + Error::Missing, + "delete reports Missing when absent" + ); + assert_eq!( + catalog_rows(&backend).await, + 0, + "delete freed the LO's pages at commit (C5: space is reused)" + ); + } + + /// The `list` contract is complete AND well-formed after mixed ops + /// (POC #7's exact-count gate: list == companion table), and in a + /// healthy store the companion table's oids agree exactly with the + /// LO catalog's distinct oids — the C1 shape (the catalog read here + /// is test instrumentation; the engine never addresses through it). + #[tokio::test] + async fn list_is_complete_after_mixed_ops() { + let Some(backend) = harness("list").await else { + return; + }; + let mut keys: Vec> = vec![ + b"\x02b-key".to_vec(), + b"\x02a-key".to_vec(), + b"\x02c-key".to_vec(), + ]; + for key in &keys[..2] { + backend.put(key, CONTENT).await.unwrap(); + } + staged(&backend, &keys[2], b"staged-content") + .await + .expect("staged commit lands"); + backend.delete(&keys[1]).await.unwrap(); + + keys.remove(1); + let mut listed = keys.clone(); + listed.sort(); + assert_eq!( + backend.list().await.unwrap(), + listed, + "list is complete after put/staged-commit/delete mixed ops" + ); + for key in &listed { + assert!(backend.has(key).await.unwrap(), "no spurious keys"); + } + + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "SELECT count(*) FROM (\ + SELECT loid FROM {t} EXCEPT SELECT DISTINCT loid FROM pg_largeobject\ + ) missing, (\ + SELECT DISTINCT loid FROM pg_largeobject EXCEPT SELECT loid FROM {t}\ + ) orphaned", + t = backend.table + ); + let disagreements: i64 = client.query_one(sql.as_str(), &[]).await.unwrap().get(0); + assert_eq!( + disagreements, 0, + "companion table and LO catalog agree exactly (no missing pages, no orphans)" + ); + assert_eq!( + companion_loids_sorted(&backend).await, + catalog_loids(&backend).await, + "the companion table's loids are the catalog's oids" + ); + } + + /// The streaming window cursor body's read contract: `len()` exact + /// (the companion row's size), EOF-clamped reads (over-read, tail, + /// at-EOF empty, zero-length), window-boundary crossings (reads that + /// straddle the 512 KiB window edge), and the ADR-006 range seam's + /// digest over exactly the returned bytes. + #[tokio::test] + async fn cursor_contract_streams_windows() { + let Some(backend) = harness("cursor").await else { + return; + }; + let content = pattern(2 * READ_CHUNK + 100); + let key = b"\x01cursor-key".to_vec(); + backend.put(&key, &content).await.unwrap(); + + let cursor = backend + .get(&key) + .await + .expect("get succeeds") + .expect("entry present"); + let len = content.len() as u64; + assert_eq!(cursor.len(), len, "len is the companion row's size"); + + assert_eq!( + cursor.read_at(0, len).await.unwrap(), + content, + "full multi-window read assembles byte-exact" + ); + assert_eq!( + cursor.read_at(0, len + 4097).await.unwrap(), + content, + "over-read clamps at EOF, pread-style" + ); + let mid = (content.len() / 2) as u64; + assert_eq!( + cursor.read_at(mid, len).await.unwrap(), + content[(content.len() / 2)..], + "tail read clamps" + ); + assert_eq!( + cursor.read_at(len, 5).await.unwrap(), + Vec::::new(), + "read at EOF is empty, never an error" + ); + assert_eq!( + cursor.read_at(u64::MAX, 5).await.unwrap(), + Vec::::new(), + "an overflowing u64 offset clamps in u64 arithmetic — the C7 trap not inherited" + ); + assert_eq!( + cursor.read_at(0, 0).await.unwrap(), + Vec::::new(), + "zero-length read" + ); + + let boundary = READ_CHUNK as u64; + let windowed = cursor.read_at(boundary - 100, 300).await.unwrap(); + assert_eq!( + windowed, + content[(READ_CHUNK - 100)..(READ_CHUNK + 200)], + "a read straddling the window edge crosses two lo_get statements byte-exact" + ); + let (slice, digest) = cursor.read_range(boundary, 7).await.unwrap(); + assert_eq!(slice, content[READ_CHUNK..READ_CHUNK + 7]); + assert_eq!(digest, crate::backend::slice_digest(&slice)); + } + + /// The degenerate case: a zero-byte entry round-trips through put, + /// the staged form, and the committed cursor (size exact-zero, every + /// read empty). + #[tokio::test] + async fn empty_entry_round_trips() { + let Some(backend) = harness("empty").await else { + return; + }; + let key = b"\x01empty-key".to_vec(); + + backend.put(&key, b"").await.unwrap(); + assert!(backend.has(&key).await.unwrap(), "empty value still lands"); + assert_eq!(backend.size(&key).await.unwrap(), Some(0), "exact zero"); + let cursor = backend.get(&key).await.unwrap().expect("entry present"); + assert_eq!(cursor.len(), 0); + assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::::new()); + + let staged_key = b"\x01empty-staged-key".to_vec(); + let mut staging = backend.put_staged().await.unwrap(); + staging.append(b"").await.unwrap(); + assert_eq!(staging.len(), 0, "empty append stays empty"); + staging + .commit(&staged_key) + .await + .expect("empty commit publishes"); + assert!( + backend.has(&staged_key).await.unwrap(), + "empty entry landed" + ); + assert_eq!(backend.size(&staged_key).await.unwrap(), Some(0)); + let committed = backend + .get(&staged_key) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!( + committed.read_at(0, 8).await.unwrap(), + Vec::::new(), + "committed empty body reads empty" + ); + } + + /// `size` reads the companion row's `size` column only (metadata + /// only, the C1 table-level authority) — exact for a large value, + /// riding the PK index (EXPLAIN proof), and never touching LO + /// content. + #[tokio::test] + async fn size_probe_is_metadata_only_and_exact() { + let Some(backend) = harness("size").await else { + return; + }; + let key = b"\x01size-probe-key".to_vec(); + let large_value = pattern(1024 * 1024); + backend.put(&key, &large_value).await.unwrap(); + + assert_eq!( + backend.size(&key).await.unwrap(), + Some(large_value.len() as u64), + "size is metadata-only and exact" + ); + assert_eq!(backend.size(b"\x01absent").await.unwrap(), None); + + let client = backend.pool.get().await.unwrap(); + let explain = format!( + "EXPLAIN SELECT size FROM {} WHERE key = '\\x{}'", + backend.table, + hex::encode(&key) + ); + let rows = client.query(explain.as_str(), &[]).await.unwrap(); + let plan: String = rows + .iter() + .map(|row| row.get::<_, String>(0)) + .collect::>() + .join(" "); + assert!( + plan.contains("Index Scan") && plan.contains("lo_entries_pkey"), + "the size probe searches the companion table's PK index: {plan}" + ); + } + + /// The staged put's happy path: 64 KiB appends (the store core's + /// put-loop window size) aggregate into 512 KiB `lowrite` statements + /// (the C2 knee), the verification pass reads the staged artifact + /// back as a whole through the read seam — including reads crossing + /// the flushed/pending aggregation boundary — commit publishes, and + /// the committed entry round-trips through get's cursor. + #[tokio::test] + async fn staged_put_round_trips_through_the_read_seam() { + let Some(backend) = harness("staged").await else { + return; + }; + let content = pattern(1_500_000); + let key = b"\x01staged-key".to_vec(); + + let mut staging = backend.put_staged().await.unwrap(); + for chunk in content.chunks(64 * 1024) { + staging.append(chunk).await.unwrap(); + } + + assert_eq!( + staging.len() as usize, + content.len(), + "staged length is the appended length" + ); + let (bytes, digest) = staging + .read_range(0, staging.len()) + .await + .expect("staged bytes readable"); + assert_eq!(bytes, content, "staged artifact readable as a whole"); + assert_eq!(digest, crate::backend::slice_digest(&content)); + + let boundary = LOWRITE_CHUNK as u64; + assert!( + staging.len() > boundary, + "the staged content crossed the first aggregation flush" + ); + let crossed = staging + .read_at(boundary - 64, 256) + .await + .expect("boundary-crossing staged read"); + assert_eq!( + crossed, + content[(LOWRITE_CHUNK - 64)..(LOWRITE_CHUNK + 192)], + "a staged read crossing the flushed/pending boundary is byte-exact \ + (LO windows over committed-into-tx pages + the in-memory tail)" + ); + + staging.commit(&key).await.expect("commit publishes"); + assert!(backend.has(&key).await.unwrap(), "entry published"); + + let cursor = backend + .get(&key) + .await + .expect("get succeeds") + .expect("entry present"); + let (bytes, digest) = cursor + .read_range(0, cursor.len()) + .await + .expect("committed read"); + assert_eq!(bytes, content); + assert_eq!(digest, crate::backend::slice_digest(&content)); + + backend + .put(&key, b"overwritten") + .await + .expect("whole-value re-put lands as dedup"); + assert_eq!( + stored_value(&backend, &key).await.as_deref(), + Some(&content[..]), + "whole-value re-put does not overwrite" + ); + staged(&backend, &key, b"overwritten-by-stage") + .await + .expect("staged re-commit is dedup, not failure"); + assert_eq!( + stored_value(&backend, &key).await.as_deref(), + Some(&content[..]), + "staged re-commit does not overwrite" + ); + assert_eq!( + catalog_rows(&backend).await, + pages_for(content.len()), + "the dedup re-puts created no orphan LO pages \ + (a re-put never births an LO)" + ); + } + + /// Stage hygiene, discard form: every non-commit path converges on + /// discard with zero residue — no companion row, no LO pages. + #[tokio::test] + async fn staged_discard_leaves_zero_residue() { + let Some(backend) = harness("discard").await else { + return; + }; + let key = b"\x01discard-key".to_vec(); + + let mut staging = backend.put_staged().await.unwrap(); + staging.append(b"residue").await.unwrap(); + staging.discard().await; + + assert!( + !backend.has(&key).await.unwrap(), + "discarded stage published nothing" + ); + assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed"); + assert!( + backend.list().await.unwrap().is_empty(), + "the pool holds exactly zero entries after discard" + ); + assert_eq!( + catalog_rows(&backend).await, + 0, + "discard left zero LO pages (the tx rolled back)" + ); + } + + /// Stage hygiene, failure form: a failed commit is already + /// self-discarded — zero residue, entry unpublished, no post-failure + /// discard left to drive (the consuming call took the cursor). The + /// injection stands in for the media failure. + #[tokio::test] + async fn commit_failure_self_discards() { + let Some(backend) = harness("fail").await else { + return; + }; + let failing = PgLoBackend { + pool: backend.pool.clone(), + table: backend.table.clone(), + schema: backend.schema.clone(), + #[cfg(test)] + fail_commit: true, + }; + let key = b"\x01failure-key".to_vec(); + + let mut staging = failing.put_staged().await.unwrap(); + staging.append(CONTENT).await.unwrap(); + let err = staging.commit(&key).await.unwrap_err(); + assert!(matches!(err, Error::Io(_)), "engine media failure"); + + assert!( + !backend.has(&key).await.unwrap(), + "failed commit published nothing" + ); + assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue"); + assert!( + backend.list().await.unwrap().is_empty(), + "zero residue in the companion table itself" + ); + assert_eq!( + catalog_rows(&backend).await, + 0, + "zero residue in the LO catalog itself" + ); + } + + /// The engine's own cancellation story: a staging cursor dropped + /// without commit or discard leaves no residue (the Drop body's + /// spawned ROLLBACK — the pool never receives a dirty checkout). + #[tokio::test] + async fn drop_cancellation_leaves_zero_residue() { + let Some(backend) = harness("drop").await else { + return; + }; + let key = b"\x01cancelled-key".to_vec(); + + { + let mut staging = backend.put_staged().await.unwrap(); + staging.append(b"never published").await.unwrap(); + // Dropped without commit or discard. + } + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + + assert!( + !backend.has(&key).await.unwrap(), + "cancelled stage published nothing" + ); + assert!( + backend.list().await.unwrap().is_empty(), + "the pool holds exactly zero entries after cancellation" + ); + assert_eq!( + catalog_rows(&backend).await, + 0, + "the spawned ROLLBACK removed the staged LO's pages" + ); + } + + /// Exact-count stage hygiene (store-api.md invariant 3): N failed + /// stages — over all three forms, explicit discard, failed commit, + /// and Drop cancellation — leave zero companion rows and zero LO + /// pages, and N successful stages leave exactly N entries. + #[tokio::test] + async fn exact_count_hygiene_forms_leave_zero() { + const N: usize = 3; + let Some(backend) = harness("exact").await else { + return; + }; + for i in 0..N { + let content = format!("discard-{i}"); + let mut staging = backend.put_staged().await.unwrap(); + staging.append(content.as_bytes()).await.unwrap(); + staging.discard().await; + + let content = format!("commit-failure-{i}"); + let key = format!("failure-key-{i}").into_bytes(); + let failing = PgLoBackend { + pool: backend.pool.clone(), + table: backend.table.clone(), + schema: backend.schema.clone(), + #[cfg(test)] + fail_commit: true, + }; + let mut staging = failing.put_staged().await.unwrap(); + staging.append(content.as_bytes()).await.unwrap(); + staging.commit(&key).await.unwrap_err(); + + let content = format!("cancelled-{i}"); + { + let mut staging = backend.put_staged().await.unwrap(); + staging.append(content.as_bytes()).await.unwrap(); + drop(staging); + } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } + + assert!( + backend.list().await.unwrap().is_empty(), + "{} failed stages left zero companion rows", + 3 * N + ); + assert_eq!( + catalog_rows(&backend).await, + 0, + "{} failed stages left zero LO pages", + 3 * N + ); + + for i in 0..N { + let content = format!("success-{i}"); + let key = format!("success-key-{i}").into_bytes(); + staged(&backend, &key, content.as_bytes()) + .await + .expect("commit publishes"); + } + let listed = backend.list().await.unwrap(); + assert_eq!( + listed.len(), + N, + "N successful stages leave exactly N entries" + ); + for key in &listed { + assert!(backend.has(key).await.unwrap(), "every entry is real"); + } + assert_eq!( + companion_loids_sorted(&backend).await.len(), + N, + "every entry owns an LO" + ); + assert_eq!( + catalog_rows(&backend).await, + pages_for(CONTENT.len()) * (N as i64), + "success leaves exactly the live entries' LO pages" + ); + } + + /// Virgin-store no-schema-yet tolerance (the recorded read-only + /// split): every read path against a schema that was never + /// bootstrapped is a no-op — absent/empty, never "table does not + /// exist" — delete reports Missing; writes surface the engine error + /// and leave zero residue (including the staged handout, whose + /// commit cannot publish). + #[tokio::test] + async fn virgin_store_no_schema_yet_reads_are_noops() { + let Some(backend) = harness_schemaless("virgin").await else { + return; + }; + let key = b"\x01opaque-key".to_vec(); + + assert!( + !backend.has(&key).await.unwrap(), + "has on a schema-less database is a no-op read" + ); + assert!( + backend.get(&key).await.unwrap().is_none(), + "get is a no-op read" + ); + assert_eq!( + backend.size(&key).await.unwrap(), + None, + "size is a no-op read" + ); + assert!( + backend.list().await.unwrap().is_empty(), + "list is empty on a schema-less database" + ); + assert_eq!( + backend.delete(&key).await.unwrap_err(), + Error::Missing, + "delete on a schema-less database reports Missing" + ); + + assert!( + backend.put(&key, CONTENT).await.is_err(), + "the tolerance is read-only: writes surface the engine error" + ); + let mut staging = backend.put_staged().await.unwrap(); + staging.append(CONTENT).await.unwrap(); + let err = staging.commit(&key).await; + assert!( + err.is_err(), + "a staged commit on a schema-less database surfaces the engine error" + ); + assert!( + backend.list().await.unwrap().is_empty(), + "the failed writes left zero residue" + ); + assert_eq!( + catalog_rows(&backend).await, + 0, + "zero LO residue from the schema-less probes" + ); + } + + /// The bootstrap creates the C1 companion-table shape — column + /// types bytea/oid/bigint/bigint — and is idempotent (a second open + /// over the same schema is a no-op; the store then keeps working). + #[tokio::test] + async fn bootstrap_creates_the_c1_row_shape_idempotently() { + let Some(admin) = harness_admin_dsn() else { + skip_note(); + return; + }; + let db = fresh_db("bootstrap"); + create_db(&admin, &db).await; + let _janitor = DatabaseJanitor::new(&admin, &db); + let dsn = dsn_for_db(&db); + let opts = PgOpts::default(); + let first = PgLoBackend::new(&dsn, opts.clone()).await.unwrap(); + let second = PgLoBackend::new(&dsn, opts).await.unwrap(); + assert!( + first.table_exists("lo_entries").await.unwrap(), + "the companion table exists" + ); + + let client = first.pool.get().await.unwrap(); + let columns: Vec = client + .query( + "SELECT data_type::text FROM information_schema.columns \ + WHERE table_schema = $1 AND table_name = 'lo_entries' \ + ORDER BY ordinal_position", + &[&first.schema], + ) + .await + .unwrap() + .iter() + .map(|row| row.get::<_, String>(0)) + .collect(); + assert_eq!( + columns, + ["bytea", "oid", "bigint", "bigint"], + "the C1 row shape: key/loid/size/committed_at column types" + ); + + let key = b"\x01bootstrap-key".to_vec(); + first.put(&key, CONTENT).await.unwrap(); + assert!(second.has(&key).await.unwrap(), "the second open sees it"); + } + + /// The quoted-identifier boundary: a hostile schema name — + /// reserved-word-carved, containing an embedded double-quote — + /// round-trips storage through the schema-qualified statements + /// without ever entering SQL text unquoted. (The test's whole + /// database is janitor-dropped, so the hostile schema needs no + /// cleanup of its own.) + #[tokio::test] + async fn hostile_schema_name_is_quoted_safely() { + let Some(backend) = harness("hostile-schema").await else { + return; + }; + let hostile_schema = "se\"lect"; + let hostile = PgLoBackend::new( + &backend.dsn, + PgOpts { + schema: hostile_schema.to_owned(), + ..Default::default() + }, + ) + .await + .expect("open with a hostile schema name"); + + let key = b"\x01hostile-schema-key".to_vec(); + hostile.put(&key, CONTENT).await.unwrap(); + assert!(hostile.has(&key).await.unwrap()); + assert_eq!(stored_value(&hostile, &key).await.as_deref(), Some(CONTENT)); + assert_eq!( + hostile.size(&key).await.unwrap(), + Some(CONTENT.len() as u64) + ); + assert_eq!(hostile.list().await.unwrap(), vec![key.clone()]); + hostile.delete(&key).await.unwrap(); + assert!(!hostile.has(&key).await.unwrap()); + } + + /// `max_size` of zero fails construction at entry, before any round + /// trip — deadpool cannot grant checkouts from an empty pool (the + /// alkstore precedent). + #[tokio::test] + async fn constructor_rejects_zero_pool_size() { + let Some(dsn) = harness_dsn() else { + skip_note(); + return; + }; + let empty_db = format!("{dsn} dbname={}", fresh_db("zero-pool")); + let opened = PgLoBackend::new( + &empty_db, + PgOpts { + max_size: 0, + ..Default::default() + }, + ) + .await; + match opened { + Ok(_) => panic!("max_size 0 must fail at entry"), + Err(Error::Io(message)) => assert!( + message.contains("max_size must be positive"), + "typed entry validation: {message}" + ), + Err(other) => panic!("max_size 0 must fail typed, got: {other:?}"), + } + } + + /// The shipped `synchronous_commit` posture (the shared PgOpts + /// design point): the parity knob defaults to `on` (the ship + /// config) on the pool's connections. + #[tokio::test] + async fn opens_with_synchronous_commit_on_by_default() { + let Some(backend) = harness("sync-default").await else { + return; + }; + assert_eq!( + shown_synchronous_commit(&backend).await, + "on", + "the shipped tier is synchronous_commit = on" + ); + } + + /// The parity knob is configurable per engine instance (the shared + /// PgOpts surface): `off` wires onto the pooled connections. + #[tokio::test] + async fn synchronous_commit_knob_wires_per_session() { + let Some(backend) = harness_with( + "sync-off", + PgOpts { + synchronous_commit: false, + ..Default::default() + }, + ) + .await + else { + return; + }; + assert_eq!( + shown_synchronous_commit(&backend).await, + "off", + "the instance's knob wires onto pooled connections" + ); + } + + /// The orphan-recovery sweep (ADR-009 §5, the C6/pgdiag10 shape): + /// a planted orphan (`lo_create` + `lowrite` + COMMIT, no companion + /// row — the legacy/bypass class crash-free stores never produce) + /// is reaped by `reap_orphans`; committed content survives the + /// sweep byte-exact; nothing is reaped when there are no orphans. + #[tokio::test] + async fn orphan_recovery_sweep_reaps_the_bypass_class() { + let Some(backend) = harness("orphan").await else { + return; + }; + let committed_key = b"\x01committed-key".to_vec(); + backend.put(&committed_key, CONTENT).await.unwrap(); + + { + let client = backend.pool.get().await.unwrap(); + client.batch_execute("BEGIN").await.unwrap(); + let loid: u32 = client + .query_one("SELECT lo_create(0) AS loid", &[]) + .await + .unwrap() + .get(0); + let fd: i32 = client + .query_one("SELECT lo_open($1, $2)", &[&loid, &INV_WRITE]) + .await + .unwrap() + .get(0); + client + .query_one("SELECT lowrite($1, $2)", &[&fd, &b"orphan bytes".to_vec()]) + .await + .unwrap(); + client.batch_execute("COMMIT").await.unwrap(); + let orphan = loid; + drop(client); + + assert_eq!( + backend.reap_orphans().await.unwrap(), + 1, + "exactly the planted orphan is reaped" + ); + let client = backend.pool.get().await.unwrap(); + let gone: i64 = client + .query_one( + "SELECT count(*) FROM pg_largeobject WHERE loid = $1", + &[&orphan], + ) + .await + .unwrap() + .get(0); + assert_eq!(gone, 0, "the orphan's pages are unlinked"); + drop(client); + } + + assert_eq!( + stored_value(&backend, &committed_key).await.as_deref(), + Some(CONTENT), + "committed content survives the sweep byte-exact (validated against lo_entries)" + ); + assert_eq!( + backend.reap_orphans().await.unwrap(), + 0, + "no orphans remain: committed LOs are never reaped" + ); + } + + /// 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 + /// database's `pg_largeobject` must hold exactly zero LO pages + /// after any full run of this suite. + #[tokio::test] + async fn shared_harness_db_holds_zero_lo_residue() { + let Some(dsn) = harness_dsn() else { + skip_note(); + return; + }; + let (client, conn) = tokio_postgres::connect(&dsn, NoTls) + .await + .expect("shared db connect"); + tokio::spawn(async move { + let _ = conn.await; + }); + let pages: i64 = client + .query_one("SELECT count(*) FROM pg_largeobject", &[]) + .await + .expect("shared catalog count") + .get(0); + assert_eq!( + pages, 0, + "the shared harness database holds zero LO pages (per-test-database isolation)" + ); + } +} diff --git a/src/postgres.rs b/src/postgres.rs index d27569d..7ef55bd 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -196,121 +196,28 @@ use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use deadpool_postgres::{Manager, ManagerConfig, Object, Pool, RecyclingMethod}; +use deadpool_postgres::{Object, Pool}; use parking_lot::Mutex; -use tokio_postgres::NoTls; use crate::backend::{clamped, Backend, ReadCursor, StagingCursor}; use crate::gc::{EngineFleetState, EngineWindowState, PinRow, WindowCandidate}; +/// The shared opts type's re-export home for the `crate::postgres::PgOpts` +/// path (its definition lives in [`crate::pg_common`]; the crate-root +/// re-export is `any(postgres, pg-lo)` — the widened gate). +pub use crate::pg_common::PgOpts; +use crate::pg_common::{ + build_pool, err, is_undefined_table, pg_err, qualified_table, quote_identifier, size_u64, + time_to_nanos, DEFAULT_SCHEMA, +}; use crate::{Error, PinState}; -fn err(e: impl std::fmt::Display) -> Error { - Error::Io(format!("postgres: {e}")) -} - -/// The engine's error mapping for the clause bodies: a SQL-level -/// failure surfaces with its server-side message and SQLSTATE (the -/// `DbError` detail `Display` drops), anything else through [`err`]. -fn pg_err(e: tokio_postgres::Error) -> Error { - match e.as_db_error() { - Some(db) => Error::Io(format!("postgres: {db}")), - None => err(e), - } -} - -/// The virgin-store tolerance's discriminant (POC #1 finding 2): a -/// statement naming a table that does not exist maps on the read paths -/// to absent/empty (and delete to Missing), never an error. The stable -/// surface PostgreSQL carries is the SQLSTATE class+code (42P01, -/// undefined table); a missing *schema* reads identically. -fn is_undefined_table(e: &tokio_postgres::Error) -> bool { - e.code() == Some(&tokio_postgres::error::SqlState::UNDEFINED_TABLE) -} - -/// The `size` column's integer, clamped to the domain the trait probes in. -fn size_u64(raw: i64) -> u64 { - raw.max(0) as u64 -} - -/// The default engine-owned schema name (the recorded design point's -/// default). -pub(crate) const DEFAULT_SCHEMA: &str = "alkblobs"; - -/// The shipped default pool size ([`PgOpts::max_size`]) — the alkstore -/// `DEFAULT_MAX_SIZE = 8` precedent. -pub(crate) const DEFAULT_MAX_SIZE: usize = 8; - -/// Quote a SQL identifier (the Postgres rule: double the embedded `"`). -/// -/// The engine composes schema-qualified SQL from a consumer-supplied -/// schema name; this is the quoted-identifier boundary — an identifier -/// is never interpolated into SQL text unquoted (the alkstore -/// `quote_identifier` pattern). -pub(crate) fn quote_identifier(name: &str) -> String { - let mut quoted = String::with_capacity(name.len() + 2); - quoted.push('"'); - for c in name.chars() { - if c == '"' { - quoted.push('"'); - } - quoted.push(c); - } - quoted.push('"'); - quoted -} - -/// A schema-qualified table reference: `"schema"."table"`. -pub(crate) fn qualified_table(schema: &str, table: &str) -> String { - format!("{}.{}", quote_identifier(schema), quote_identifier(table)) -} - -/// Construction options for the postgres engine. Durability knobs, -/// pool sizing, and the engine-owned schema name are -/// engine-configuration concerns — they ride the store constructors' -/// parameters, never invented defaults (the alkstore `PgOpts` -/// precedent; `open(config, PgOpts)` adapted to the named-constructor -/// shape — the constructors are [`Store::dual_tier_postgres_local`] and -/// [`Store::kv_only_postgres`]). -/// -/// The connection settings themselves do not ride this struct — they -/// are the constructor's separate connection-settings argument (the -/// tokio-postgres [`tokio_postgres::Config`]-parseable form); the -/// engine never invents defaults for them. TLS: the engine hardwires -/// `NoTls` on every connection path regardless of the config's -/// sslmode — v1 TLS is a post-v1 deployment concern, mirroring the -/// alkstore posture. -#[derive(Debug, Clone)] -pub struct PgOpts { - /// The engine-owned PostgreSQL schema holding the pool table. - /// Default [`DEFAULT_SCHEMA`]; an empty string selects the default. - /// The name is a quoted-identifier boundary ([`quote_identifier`]) - /// — reserved-word and hostile names are safe. - pub schema: String, - /// Upper bound on pooled connections (deadpool's `max_size`). - /// Default [`DEFAULT_MAX_SIZE`] = 8. A value of `0` fails - /// construction immediately (validated at entry, no round trip — - /// deadpool cannot grant checkouts from an empty pool; the - /// alkstore precedent). - pub max_size: usize, - /// The per-session `synchronous_commit` durability knob, wired as a - /// connect option (`-c synchronous_commit=…`) on every pooled - /// connection (POC-verified per-session SET mechanics; the - /// consumer's own `options=` DSN values are preserved alongside — - /// postgres parses repeated `-c` flags). Default `true` — the ship - /// config; `false` is the documented sqlite-NORMAL-alike latency - /// trade (B5 ~2-3×). - pub synchronous_commit: bool, -} - -impl Default for PgOpts { - fn default() -> Self { - Self { - schema: DEFAULT_SCHEMA.to_owned(), - max_size: DEFAULT_MAX_SIZE, - synchronous_commit: true, - } - } -} +/// The test-support helpers' shared home — the SchemaJanitor lives in +/// `src/pg_common.rs`'s `#[cfg(test)] test_support` (gated +/// `cfg(test)` + `any(postgres, pg-lo)`, the shared-helper design +/// point's home); re-exported here so the sibling test suites' +/// `crate::postgres::test_support` path holds. +#[cfg(test)] +pub(crate) use crate::pg_common::test_support; /// The `postgres` engine — a deadpool pool over tokio-postgres plus the /// quoted engine-owned table reference. Cloning shares the pool (the @@ -408,7 +315,7 @@ impl PostgresBackend { /// `new_tableless` pattern. #[cfg(test)] async fn new_schemaless(config: &str, schema: &str) -> Result { - let pool = build_pool(config, DEFAULT_MAX_SIZE, true)?; + let pool = build_pool(config, crate::pg_common::DEFAULT_MAX_SIZE, true)?; let schema = schema.to_owned(); let pins_table = qualified_table(&schema, "gc_pins"); let window_table = qualified_table(&schema, "gc_window"); @@ -488,40 +395,6 @@ impl PostgresBackend { } } -/// Build the deadpool pool with the shipped posture: per-session -/// `synchronous_commit` merged into every connection's connect options -/// (appended to the consumer's own `options=` DSN values — postgres -/// parses repeated `-c` flags), `RecyclingMethod::Fast`, the configured -/// `max_size`. `max_size = 0` fails at entry, before any round trip. -fn build_pool(config: &str, max_size: usize, synchronous_commit: bool) -> Result { - if max_size == 0 { - return Err(Error::Io(format!( - "postgres: max_size must be positive, got {max_size}" - ))); - } - let mut cfg: tokio_postgres::Config = config.parse().map_err(err)?; - let engine_opt = format!( - "-c synchronous_commit={}", - if synchronous_commit { "on" } else { "off" } - ); - let merged = match cfg.get_options() { - Some(existing) => format!("{existing} {engine_opt}"), - None => engine_opt, - }; - cfg.options(merged); - let manager = Manager::from_config( - cfg, - NoTls, - ManagerConfig { - recycling_method: RecyclingMethod::Fast, - }, - ); - Pool::builder(manager) - .max_size(max_size) - .build() - .map_err(err) -} - /// The engine-owned bootstrap DDL — idempotent, no migration machinery. /// The B5 row shape under the measured `fillfactor = 90`; column names /// quote-follow the alkstore hazard-list discipline (`"key"` quoted; @@ -549,19 +422,10 @@ pub(crate) const DEFAULT_PIN_TTL_SECS: u64 = 60; /// source): every renewed row's expiry moves to `now` + this. const DEFAULT_PIN_TTL: Duration = Duration::from_secs(DEFAULT_PIN_TTL_SECS); -/// The window row family's timestamp encoding (the recorded design -/// point): signed nanoseconds since the Unix epoch — sub-second -/// precision for the test tunings, negative for pre-epoch instants, -/// i64 across the full range a `BIGINT` stores (±292 years around the -/// epoch). The sqlite engine's helpers are this one's reference — -/// per-engine bodies, one encoding. -fn time_to_nanos(t: SystemTime) -> i64 { - match t.duration_since(UNIX_EPOCH) { - Ok(d) => d.as_nanos() as i64, - Err(e) => -(e.duration().as_nanos() as i64), - } -} - +/// The gc-window/pin rows' timestamp decoding — the test tunings' +/// fixture helper (the sqlite engine's helpers are the reference — +/// per-engine bodies, one encoding: [`crate::pg_common::time_to_nanos`] +/// is the shared forward direction). fn nanos_to_time(n: i64) -> SystemTime { if n >= 0 { UNIX_EPOCH + std::time::Duration::from_nanos(n as u64) @@ -1191,6 +1055,7 @@ mod tests { use crate::gc::{EngineFleetState, EngineWindowState}; use std::ops::Deref; use std::sync::atomic::{AtomicU64, Ordering}; + use tokio_postgres::NoTls; const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog"; @@ -2794,64 +2659,3 @@ mod tests { backend.sweep_lock_release().await.unwrap(); } } - -/// Test-support helpers shared with the crate's other postgres-attached -/// test modules (`src/store.rs`'s construction tests) — crate-internal -/// and compiled out of every non-test build. -#[cfg(test)] -pub(crate) mod test_support { - use super::quote_identifier; - use tokio_postgres::NoTls; - - /// Drops a test's engine-owned schema when the test ends — success, - /// early return, or unwind (Drop runs in all three), so a fresh - /// schema per test never accumulates on the shared harness - /// database. No async teardown exists at Drop, so the drop executes - /// synchronously from a std thread over a one-shot current-thread - /// runtime; an unreachable harness server is tolerated (the residue - /// stays; nothing else breaks). - /// - /// Construct it (or a wrapper holding it) *before* the store or - /// backend it outlives: drop order is reverse of declaration, so - /// the backend's pool drains before the schema drops. - pub(crate) struct SchemaJanitor { - dsn: String, - schema: String, - } - - impl SchemaJanitor { - pub(crate) fn new(dsn: &str, schema: &str) -> SchemaJanitor { - SchemaJanitor { - dsn: dsn.to_owned(), - schema: schema.to_owned(), - } - } - } - - impl Drop for SchemaJanitor { - fn drop(&mut self) { - let dsn = self.dsn.clone(); - let schema = self.schema.clone(); - let _ = std::thread::spawn(move || { - let Ok(rt) = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - else { - return; - }; - rt.block_on(async move { - let Ok((client, conn)) = tokio_postgres::connect(&dsn, NoTls).await else { - return; - }; - tokio::spawn(conn); - let sql = format!( - "DROP SCHEMA IF EXISTS {} CASCADE", - quote_identifier(&schema) - ); - let _ = client.execute(sql.as_str(), &[]).await; - }); - }) - .join(); - } - } -} diff --git a/tasks/engine-pg-lo.md b/tasks/engine-pg-lo.md index 96ea691..5e447d2 100644 --- a/tasks/engine-pg-lo.md +++ b/tasks/engine-pg-lo.md @@ -1,7 +1,7 @@ --- id: engine-pg-lo name: pg-lo large engine — Large Objects behind the Backend trait -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -230,8 +230,114 @@ engine and its in-module unit tests only. ## Notes -> To be filled by implementation agent +> Implemented in `src/pg_lo.rs` (the engine, its cursors, the orphan +> sweep, and 18 in-module tests), `src/pg_common.rs` (new — the +> shared-helper home), and `src/{lib,postgres}.rs` (the shared-helper +> move + gate widening; `src/postgres.rs`'s changes are mechanical +> import swaps plus path-stable re-exports). Recorded design points, +> as resolved: +> +> - **Shared-helper shape** — `src/pg_common.rs`, gated +> `cfg(any(feature = "postgres", feature = "pg-lo"))`, one module per +> file: `quote_identifier`/`qualified_table`, the 42P01 discriminant, +> `err`/`pg_err` (the mapping keeps the `postgres:` prefix — both +> engines attach to the same server family), `size_u64`, +> `time_to_nanos` (the i64-nanos encoding), `build_pool`, [`PgOpts`], +> and the `#[cfg(test)] test_support` janitors (`SchemaJanitor` +> gated `#[cfg(feature = "postgres")]`, new `DatabaseJanitor` gated +> `#[cfg(feature = "pg-lo")]`; `crate::postgres::test_support` remains +> a path-stable re-export, so `src/store.rs`'s import needed no edit). +> - **PgOpts** — one shared type; its definition moved to `pg_common`, +> the public re-export widened to `any(postgres, pg-lo)` (the type is +> unchanged; `alkblobs::postgres::PgOpts` still resolves through the +> path-stable re-export). The `synchronous_commit` knob **carries** +> (default `on`; the POC's server-side `off` was a measurement +> posture, not the ship config — recorded in the module doc). +> - **LO isolation (the C7 lesson resolved)** — Large Objects are +> database-level; schema-per-test does not isolate them. The tests +> run on **per-test dedicated databases** (the POC's proven +> `open_fresh` shape): unique `abtlo_{tag}_{pid}_{seq}_{nanos}` +> names, `CREATE DATABASE` through the `postgres` maintenance db, a +> `DatabaseJanitor` drop-guard (`DROP DATABASE … WITH (FORCE)`) after +> the backend's pool. Zero LO residue on the shared `blobs` db is +> structural (no pg-lo test ever opens a pool against it) and gate- +> tested (`shared_harness_db_holds_zero_lo_residue`). +> - **`committed_at`** — the crate-wide i64-nanos encoding as `bigint`, +> engine-stamped at insert (not a server-default column); the crate's +> one timestamp encoding rather than the POC's `timestamptz default +> now()`. +> - **Staged reads** — verified live: `lo_get` inside the writing tx +> sees its own uncommitted pages (read-your-own-writes), so the +> verification pass reads the staged artifact through descriptorless +> windows + the unflushed in-memory tail; no descriptor reads, no +> `lo_lseek` anywhere (the C7 64-bit discipline: offsets ride bigint +> only, `lowrite` is implicitly sequential). +> - **Stage hygiene mechanics** — the staging cursor takes its +> checkout in `commit` (`publish` over the taken `Object`), so every +> > failure path converges on one ROLLBACK point; the dedup path +> (CAS hit) ROLLBACKs and returns `Ok` (the staged LO is the residue +> removed); `Drop` hands an abandoned cursor's open tx to a spawned +> ROLLBACK (never returns a dirty checkout to the pool — deadpool's +> `Fast` recycle pitfall), degrading to a one-shot thread when no +> runtime is live. `#[cfg(test)] fail_commit` knob mirrors +> mem/sqlite/postgres/local. +> - **Harness-server residue found and cleaned** — the shared `blobs` +> db carried pre-existing exploration residue (`public.lo_entries` +> + 21 LOs, 9 472 catalog pages) from pre-task pg-lo probing; +> unlinked + dropped (count now 0) so the +> `pg_largeobject`-probe gate measures this suite's own runs. The +> `public.blobs` kv table was left untouched (kv-engine content, out +> of scope). +> - **The literal `--no-default-features --features pg-lo` config +> cannot compile the whole crate** — `store.rs`'s `KvEngine`/`KvCursor` +> enums are empty there (E0004), identically on the pre-task tree +> (verified by stash). Every constructor mode pairs a kv engine +> (ADR-011 §4's table), so a kv-less store is not a shippable shape; +> the clean configs are `--no-default-features --features "pg-lo,mem"` +> and ADR-013 §4's pg-pure build +> `--no-default-features --features "postgres,pg-lo"` — both clippy +> clean. In the literal config the *shared pieces* compile (the 10 +> store.rs E0004s are the crate's only errors; nothing in +> pg_common/pg_lo errors). Flagged for `store-pg-lo-construction` to +> keep in view. ## Summary -> To be filled on completion \ No newline at end of file +> Completed 2026-10-12 (wave 5). `src/pg_lo.rs` implements the +> seven-method `Backend` + `put_staged` behind feature `pg-lo` +> (default-off): LOs hold the bytes, the `lo_entries` companion table is +> the contract authority (list/size/has/CAS all table-level; `pg_largeobject` +> read by the engine only inside the orphan sweep's candidate +> enumeration), one-tx puts (lo_create → 512 KiB chunked `lowrite` +> (`LOWRITE_CHUNK`, the C2 knee) → row insert → COMMIT), the +> descriptorless `lo_get(oid, off, len)` window cursor +> (`READ_CHUNK`; `len()` metadata-only from the companion row; EOF +> clamping per the cursor contract plus the unexpected-EOF clean +> failure for a body short of its claimed length), the held-tx staging +> cursor with aggregation of the store core's 64 KiB windows into +> 512 KiB statements and both hygiene forms exact-count-tested, one-tx +> deletes with the Missing discipline, the complete single-snapshot +> `list`, idempotent schema+table bootstrap at the constructor with the +> quoted-identifier boundary (hostile names tested), the 42P01 +> read-only virgin-store split (schema-less form tested), the +> crate-internal `reap_orphans` sweep (scope-noted; planted-orphan +> contract test with byte-exact committed-content survival), and the +> posture deltas in the module doc (catalog space never returned + +> `VACUUM FULL` runbook line + rel-size monitoring; autovacuum +> inherited; the 20–50× cached-get capacity facts). Shared helpers +> (`pg_err`, 42P01, quoting, timestamps, pool, `PgOpts`) moved to the +> new `src/pg_common.rs` gated `any(postgres, pg-lo)`; `PgOpts`'s +> public re-export widened (shared opts decision; the knob carries). +> 18 in-module tests, per-test dedicated databases (`DatabaseJanitor`), +> env-driven (`ALKBLOBS_PG_*`), skipping cleanly server-less. Verified: +> `cargo test` (69/19/3 ok), `cargo test --features pg-lo` live +> (87/19/3 ok, 18/18 pg-lo), `cargo test --features postgres` live +> (103/36/3 ok), `cargo test --all-features` (176/71/4 ok), clippy +> `-D warnings` green (default, `--features pg-lo`, `--features +> postgres`, `--all-features`, `--no-default-features --features +> "pg-lo,mem"`, `--no-default-features --features "postgres,pg-lo"`), +> `cargo fmt --check` green, `cargo doc --no-deps` clean, zero LO + +> schema + database residue on the harness server after full runs. +> Engine-state window family remains `engine-pg-lo-window` +> (`EngineWindowState` untouched; nothing of `EngineFleetState` exists +> on this engine); constructor wiring remains `store-pg-lo-construction`. \ No newline at end of file