alkblobs 0.1.0

//! 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/`'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 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 below is the store-held tx edition of this staged put; the
//! engine-level staging cursor stays the non-fleet put path's own.
//!
//! **The fleet joint form — the store-held tx (ADR-012 §4's one-tx
//! ruling, wave 6; feature `postgres`).** Under the consolidation
//! posture (kv = postgres + large = pg-lo, same pg instance), a large
//! entry's content reaches the *same store-held transaction* the pin
//! row commits in: `put_joint_staged` hands a `PgLoJointStaging`
//! cursor against the held `PgJointTx` — the postgres engine's held
//! checkout, wave 4's recorded shape — and `pin_commit` (the store's
//! terminal step) publishes entry + pin row in one step over the
//! shared instance. **Named design point — the contribution's API
//! home (decided):** the *joint staging-cursor variant*, not a
//! whole-value method taking `&mut PgJointTx` — the large tier's put
//! contract is the streaming staging cursor, the fleet put driver
//! routes to the joint form exactly as the non-fleet path routes to
//! the staged form (stage-while-hashing, the key resolving at stream
//! end), and the unknown-length flow's derivation pass re-reads the
//! staged artifact through the cursor's read seam — none of which a
//! whole-value method can serve. The cursor is the staged put above
//! with inverted ownership: it owns no checkout, opens no
//! transaction, never COMMITs, never ROLLBACKs (the store core's
//! driver owns termination on every path, amendment A1), and borrows
//! the held checkout for every statement; the staged-put mechanics
//! are the landed bodies re-homed — the 512 KiB `lowrite` knee, the
//! CAS existence check, the companion-row insert, transactional LO
//! creation (the LO lives and dies with the store-held tx, C6's
//! structural zero-orphan story), read-your-own-writes staging reads.
//! **The joint CAS form:** the key resolves at publish, so
//! first-writer-wins lands there — a re-put publishes an **empty
//! contribution** (the staged LO unlinked *inside the store-held tx*:
//! the committed tx carries nothing of the losing put, zero residue,
//! and the driver's `pin_commit` upsert-renews the pin row), while a
//! concurrent CAS miss surfaces the companion key's unique violation
//! (the landed puts' posture). The store's one-tx ruling is what the
//! consolidation admits: the same held tx also carries the kv host's
//! `joint_put` row write for entries that land on that tier, one tx
//! over one instance. The engine-level held-tx staging cursor above
//! is unchanged (the non-fleet put path keeps its own tx). Nothing of
//! `EngineFleetState` exists on this engine (ADR-010 §2's correction
//! of ADR-009's ownership statement): the contribution is
//! crate-internal mechanics the store core drives.
//!
//! **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-50x 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, declaration-gated per ADR-016): 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). The sweep is gated by
//! `PgOpts.orphan_reap_permitted` (**default false**): the flag is the
//! operator's assertion that this database hosts no other large-object
//! consumer — `pg_largeobject` is database-level, not schema-level,
//! while the claim probe is companion-table-scoped, so an undeclared
//! sweep would `lo_unlink` a co-tenant consumer's committed LOs (the
//! review-001 F-4 data-loss shape); the sweep refuses with a named
//! error while the declaration is absent. ADR-016's riding fixes, all
//! landed: the `lo_entries(loid)` supporting index (the claim probe is
//! `WHERE loid = $1` per oid and the PK is `key` — bootstrap-created,
//! so existing databases gain it on next construction), and the
//! missing-LO pass (a loid vanishing between the candidate snapshot and
//! its unlink — a concurrent same-engine delete, or a foreign
//! consumer's lifecycle — rides a savepoint: the SQLSTATE 42704 pass
//! continues the sweep and is not counted reaped, instead of a plain
//! unlink's statement error aborting the whole tx). Candidates enumerate
//! `SELECT DISTINCT loid FROM pg_largeobject` as a **candidate list
//! only**, re-validated at delete time against `lo_entries.loid` (the
//! truth) — cost scales with oid count, not content size (C6: ~7 ms/oid).
//! Scope-noted under the module-level allow; the landed ops module
//! (wave 7) consumes none of it — the sweep's consumer remains the
//! embedder's operator runbook, not a store-core verb; contract-tested
//! with a planted orphan, a byte-exact survival assertion for committed
//! content, the tested refusal without the declaration, and the stale-
//! candidate pass.
//!
//! **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 implemented here and wired
//! by construction (landed with the constructor wiring): the store
//! core's construction call follows the one-window-host rule — active
//! on the kv=mem + large=pg-lo shape (this module's `ensure_gc_schema`
//! runs at construction; its `gc_window` hosts the durable window),
//! dormant on a dual-SQL instance (kv=sqlite/postgres + large=pg-lo:
//! the kv host's call runs, this schema carries no `gc_window`);
//! `ensure_gc_schema` creates `gc_window` only — never `gc_pins`,
//! never the embedder's liveness table.
//!
//! 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).
//!
//! The pre-split file's blanket `#![allow(dead_code)]` is gone (the
//! wave-8 split discipline: no blanket allowances survive). The
//! remaining allowance is scoped to the crate-internal orphan-recovery
//! sweep (`PgLoBackend::reap_orphans`/`reap_orphans_tx` — the
//! maintenance verb whose consumer is the embedder's operator runbook,
//! not the in-crate store paths) with the consumer named at it; the
//! joint form's feature-gated shapes need none (the `postgres` gate
//! compiles them out of the pg-lo-pure build and the fleet consumers
//! use them where they compile in); the test-only mid-driver
//! failure-injection knob needs one (its store-core consolidation
//! driver is the `all(postgres, pg-lo)` build's consumer).
//!
//! **The split layout (wave-8 structure, task `split-postgres-pg-lo`):**
//! the engine is a directory module — `staging.rs` the held-tx staging
//! cursor's bodies (with its `Drop` body) and the fleet joint form
//! (`PgLoJointStaging` + the `put_joint_staged` handout, feature
//! `postgres`), `windows.rs` the descriptorless `lo_get` window
//! machinery, the window-family seam clauses, the streaming cursor, and
//! the whole `Backend` companion-row body, `reap.rs` the orphan-recovery
//! sweep, `testing.rs` the shared test fixtures. The two
//! cross-file-shared structs (`PgLoBackend`, `PgLoStaging`) stay on the
//! module root so their private fields remain sibling-visible; the
//! public path `crate::pg_lo::PgLoCursor` is re-exported from
//! `windows.rs`.

/// 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 GC window-table reference, quoted once at
    /// construction: `"schema"."gc_window"` — the window family's row
    /// home (ADR-012 §5's pg-lo row: the window clauses and the window
    /// table only).
    window_table: String,
    /// The engine-owned schema name (the test-only catalog probes').
    schema: String,
    /// The orphan-reap declaration ([`PgOpts::orphan_reap_permitted`],
    /// ADR-016) — read at the sweep's entry ([`PgLoBackend::reap_orphans`]);
    /// the dangerous method refuses while it is absent.
    orphan_reap_permitted: bool,
    /// 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,
    /// Test-only mid-loop stage-failure injection (the transactional
    /// window staging test's instrument — the failure lands after prior
    /// candidates' upserts have run inside the open transaction).
    #[cfg(test)]
    fail_window_stage: bool,
}

/// 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<Object>,
    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<u8>,
    #[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.
    // consumer: the kv+large=pg-lo pair constructors (feature `postgres`
    // or features `pg-lo`+`mem`) — a pg-lo-only build has no constructible
    // store, so this construction entry point compiles dead.
    #[cfg_attr(not(any(feature = "mem", feature = "postgres")), allow(dead_code))]
    pub(crate) async fn new(config: &str, opts: PgOpts) -> Result<Self, Error> {
        let pool = build_pool(config, &opts)?;
        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"),
            window_table: qualified_table(&schema, "gc_window"),
            schema,
            orphan_reap_permitted: opts.orphan_reap_permitted,
            #[cfg(test)]
            fail_commit: false,
            #[cfg(test)]
            fail_window_stage: false,
        })
    }
}

impl PgLoBackend {
    /// Test-only injection arm: force the joint contribution's
    /// publish-time failure form (the mid-driver failure the
    /// store-core driver's A1 termination design is exercised against
    /// through the public `put` — the contribution returns with the
    /// tx the way it found it, the driver owes the termination).
    /// Mirrors the other engines' cfg(test) knobs.
    #[cfg(test)]
    #[allow(dead_code)]
    pub(crate) fn inject_fail_commit(&mut self) {
        self.fail_commit = true;
    }

    /// The pg instance's `system_identifier` (`pg_control_system()`,
    /// PG14+) — read across one pooled connection. The fleet
    /// constructor's same-instance probe's large-side instrument
    /// (ADR-012 §4): the kv engine's identifier compared against this
    /// — server-side instance identity, immune to DSN aliases. A probe
    /// failure surfaces as an error (the probe errors closed — never a
    /// silent pass). pg-lo never hosts fleet state; it only answers
    /// the probe.
    #[allow(dead_code)]
    pub(crate) async fn system_identifier(&self) -> Result<u64, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let row = client
            .query_one("SELECT system_identifier FROM pg_control_system()", &[])
            .await
            .map_err(pg_err)?;
        let id: i64 = row.try_get(0).map_err(pg_err)?;
        Ok(id as u64)
    }

    /// 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<bool, Error> {
        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 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.
/// The `(loid)` supporting index (ADR-016's riding fix): the orphan-reap
/// claim probe is `WHERE loid = $1` per oid and the PK is `key` —
/// without the index every probe seq-scans; idempotent DDL, so existing
/// databases gain it on next construction.
// consumer: `PgLoBackend::new` (the only caller); the pg-lo-only build
// compiles the body dead alongside its constructor.
#[cfg_attr(not(any(feature = "mem", feature = "postgres")), allow(dead_code))]
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
);

CREATE INDEX IF NOT EXISTS lo_entries_loid_idx ON {table} (loid);
"
    )
}

/// The GC window schema's DDL — idempotent, the window-only row (ADR-012
/// §5's pg-lo row): `gc_window` keyed by the candidate key, the
/// timestamp `BIGINT` nanos. `gc_pins` is never named here (fleet pins
/// live on the kv host) and neither is the embedder's liveness table
/// (the schema-ownership rule — the sweep reads it and never owns it).
pub(super) fn gc_window_ddl(schema: &str) -> String {
    let schema_q = quote_identifier(schema);
    let window = qualified_table(schema, "gc_window");
    format!(
        "CREATE SCHEMA IF NOT EXISTS {schema_q};

CREATE TABLE IF NOT EXISTS {window} (
    key bytea PRIMARY KEY,
    visible_after bigint NOT NULL
);
"
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::{Backend, StagingCursor};
    use crate::pg_common::test_support::DatabaseJanitor;
    use tokio_postgres::NoTls;

    use crate::pg_lo::testing::{
        catalog_rows, create_db, dsn_for_db, fresh_db, harness, harness_admin_dsn, harness_dsn,
        harness_schemaless, harness_with, shown_synchronous_commit, skip_note, stored_value,
        CONTENT,
    };

    /// 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<String> = 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.)
    ///
    /// **The ADR-016 §3 framing (revised):** the co-tenancy this test
    /// demonstrates — two engines' schemas, one LO-level database — is
    /// exactly the configuration the orphan-reap declaration bounds,
    /// and it bounds only the *reaping* method. Row isolation across
    /// the schemas is unaffected and stays what this test proves: the
    /// storage verbs (put/get/list/size/delete) ride the
    /// schema-qualified companion table and never cross; the sweep's
    /// claim probe is the one operation whose reach crosses the
    /// database-level catalog, and it refuses on the default
    /// (declaration-absent) construction.
    #[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());

        let err = hostile.reap_orphans().await.unwrap_err();
        assert!(
            err.to_string().contains("orphan reap refused"),
            "storage isolates; reaping is what the declaration gates: {err}"
        );
    }

    /// `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"
        );
    }

    /// 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)"
        );
    }
}

mod reap;
mod staging;
#[cfg(test)]
pub(crate) mod testing;
mod windows;

pub use windows::PgLoCursor;

use deadpool_postgres::{Object, Pool};

use crate::pg_common::{
    build_pool, err, pg_err, qualified_table, quote_identifier, PgOpts, DEFAULT_SCHEMA,
};
use crate::Error;