alkblobs 0.2.2

//! The `postgres` kv-tier engine (ADR-007): POC #5 finding B5's row shape
//! held in one engine-owned PostgreSQL schema, content-addressed rows over
//! pooled tokio-postgres sessions. Feature `postgres`, **default-off**
//! (ADR-013 §4): the kv tier's second SQL-backed engine, the fleet host
//! (ADR-012 §4), and `src/sqlite.rs`'s structural twin — same contract,
//! different driver cluster and posture deltas.
//!
//! **Pooling is load-bearing, not an optimization (POC #5 B2, recorded
//! posture).** A fresh PostgreSQL session costs ~20 ms; every engine
//! operation rides [`Pool`] checkouts, sized by [`PgOpts::max_size`]
//! (shipped default `DEFAULT_MAX_SIZE` = 8, the alkstore precedent).
//! The pool is the concurrency posture in full — tokio-postgres is
//! natively async, so there is **no `spawn_blocking` seam** here (the
//! sqlite engine's blocking discipline does not carry;
//! store-api.md §Concurrency posture).
//!
//! **Per-connection prepared-statement discipline (POC #5 B5, recorded
//! design point).** Prepared statements die with sessions — the POC
//! probe hit "prepared statement s_1 does not exist" exactly when a
//! pool-recycled connection lost its statement. Every engine statement
//! rides `prepare_cached` on the pooled client, whose statement cache is
//! per-connection by construction: a checkout whose session lost its
//! statements simply misses the cache and re-prepares. Raw `prepare`
//! never appears below.
//!
//! **`synchronous_commit` parity knob (ADR-007 §4, recorded posture).**
//! Postgres has no table-level `synchronous=NORMAL` equivalent; the
//! durability parity with sqlite's shipped `synchronous=NORMAL`-WAL tier
//! is the session-level `synchronous_commit`;
//! [`PgOpts::synchronous_commit`] wires it as a connect option on
//! **every pooled connection** (the POC-verified per-session
//! connect-options mechanics; the alkstore `src/opts.rs` precedent).
//! Default `on` — the ship config; `off` is the documented
//! sqlite-NORMAL-alike latency trade (B5: latency moves ~2-3x across
//! settings on the POC disk). Recorded posture, not hidden.
//!
//! **Schema (recorded design point).** One engine-owned PostgreSQL
//! schema — [`PgOpts::schema`], default `DEFAULT_SCHEMA` — holds the
//! B5 row shape: `blobs (key bytea PRIMARY KEY, value bytea NOT NULL,
//! size bigint NOT NULL) WITH (fillfactor = 90)`. The constructor runs
//! idempotent DDL (`CREATE … IF NOT EXISTS`) at open on a pool
//! connection; there is no migration machinery. The only identifier
//! entering SQL text from outside is the schema name, and it rides
//! `quote_identifier` (the alkstore `quote_identifier` boundary —
//! double the embedded `"`; reserved-word and hostile names are safe)
//! on **every** statement, bootstrap and DML alike. Engine-owned column
//! names are fixed and unquoted in DML (`key`, `value`, `size` are
//! non-reserved); in DDL they follow the alkstore hazard-list quote.
//!
//! **Autovacuum / storage-overhead posture (ADR-007 §4/§6, stated —
//! not silence, per B5/B6).** A continuously-churning CAS pool leaves
//! dead tuples with every CAS loser and every delete; autovacuum is a
//! real maintenance surface sqlite does not have. `fillfactor = 90` is
//! the POC-measured posture for a mostly-append immutable workload
//! (the default 100 produced a bloated-looking heap in early B5 runs).
//! The ~40x small-tier storage overhead (B6: page-fill + block floor at
//! 1-4 KiB values) is accepted — deployments dominated by tiny blobs on
//! constrained disks stay better served by sqlite. The B5 tuning
//! numbers are POC-derived posture defaults (ADR-007 §6's
//! deferred-cost pattern): the first real pg deployment verifies them
//! against its media and records deltas.
//!
//! **The dedicated `size` column is the metadata-only probe** (ADR-008's
//! size discipline): `SELECT size … WHERE key = $1` rides the PK index
//! (gate-tested below with `EXPLAIN`) and never loads the value blob.
//! `has` is an EXISTS probe, never a value load. `get` is the kv
//! family's materializing cursor body: the whole value loaded once
//! (bounded by tier policy), `len()` from the row's size column, range
//! reads slicing the materialized value through
//! `crate::backend::clamped`.
//!
//! **CAS first-writer-wins** on both put forms (`INSERT … ON CONFLICT DO
//! NOTHING`, ADR-007 §3's mapping of sqlite's `INSERT OR IGNORE`): a
//! re-put of an existing key is dedup (`Ok`), never an overwrite.
//!
//! **`list` is complete and well-formed as a stable cursor over a
//! vacuuming table (ADR-007 §5, recorded design point).** One statement
//! executes on one MVCC snapshot: ghost rows from concurrent vacuuming
//! never surface. The single `SELECT key … ORDER BY key` executes
//! through tokio-postgres's row-stream machinery — server-side portal
//! batches drained incrementally, the collection materializing only the
//! key column — the recorded fetch shape, bounded-memory on huge pools.
//!
//! **Virgin-store posture.** The constructor creates the schema and
//! tables. The tolerance is the sqlite engine's recorded read-only
//! split: a nonexistent table (SQLSTATE 42P01, undefined table) on the
//! read paths maps to absent/empty, and `delete` reports Missing —
//! never a "table does not exist" error; a write against a schema-less
//! database surfaces the engine error rather than silently creating
//! schema. The test fixture connects without running the bootstrap
//! (the sqlite `new_tableless` pattern).
//!
//! **The crate-internal batch-delete helper**
//! (`PostgresBackend::delete_batch`): `DELETE … WHERE key = ANY($1)`
//! over a `bytea[]` parameter — ADR-007 §5's sweep-safety mechanics,
//! the sweep's batched-delete shape (the store core's delete paths run
//! per-key arbitration and deletion; the helper remains available to
//! sweep-batch consumers). Its virgin-store shape is a zero-row no-op
//! (nothing to delete in a table that does not exist).
//!
//! **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.
//!
//! **The engine-state clause families (ADR-010 §2 as reduced by ADR-012
//! §5 — the postgres row is all yes: this is the fleet host).** The
//! `EngineWindowState` impl creates `gc_pins` + `gc_window`
//! (unconditionally — the applicability row does not condition on the
//! fleet declaration) and stages/takes the durable delete window; the
//! `EngineFleetState` impl carries the fleet family: the joint
//! entry+pin transaction, the pin clauses, delete-time SQL arbitration
//! with the engine-side liveness-table registration, the sweep lease,
//! and the maintenance step (`pin_reap`, the expired-pin reaping).
//! The store core's put driver consumes the families' joint-tx
//! and pin-renew clauses on the fleet put path (landed — the
//! `store-fleet-put` driver: `begin_joint_tx` → joint contribution →
//! `pin_commit`, plus the guard's TTL/3 renewal task driving
//! `pin_renew`); the store core's sweep driver consumes the sweep
//! lease, the maintenance step, and the SQL arbitration verbs (landed
//! — the `store-fleet-sweep` body: the sweep runs under the lease on
//! both the in-process flag and the advisory lock, the maintenance
//! step reaps the expired pin rows before the overdue execution, and
//! every delete-time arbitration folds the SQL answer behind the
//! in-process domain) — under `fleet: false` none of
//! the fleet clauses run, while
//! `ensure_gc_schema` runs on every postgres store construction (the
//! `store-postgres-construction` wiring, landed: the constructors are
//! `Store::dual_tier_postgres_local` and `Store::kv_only_postgres`,
//! both implicitly `fleet: false` per ADR-012 §3). The embedder's liveness table
//! is never created here (the sweep reads it and never owns it —
//! ADR-008's schema-ownership rule); the engine-side registration of
//! named liveness-table references (ADR-012 §6.6) is
//! `PostgresBackend::register_liveness_source` (crate-internal), a
//! mechanism the store core's fleet registration verb forwards to.
//! Every clause is contract-tested in-module per its applicability row.
//!
//! **Named design points (recorded for the review gate):**
//!
//! - **`JointTx` is PgJointTx** — the pooled checkout
//!   ([`Object`]) held for the tx's life, with the transaction opened
//!   as an explicit session-level `BEGIN` on that checkout and closed
//!   by `COMMIT`/`ROLLBACK` in the clause bodies. tokio-postgres's
//!   `Transaction` borrows its client (`&mut`), so it cannot be
//!   co-owned with the checkout behind one `Send` field — the
//!   exclusive checkout is what makes "the pooled client carries the
//!   tx" real (no other statement can interleave on the session), and
//!   ownership stays with the store core from `begin_joint_tx` until
//!   `pin_commit` consumes it (ADR-010 §2's pinned ruling).
//! - **`pin_renew`'s successor time** — the clause signature carries
//!   no expiry, so the engine computes `expiry = now + pin_ttl` (the
//!   backend's pin-TTL field: the shipped default 60 s, constructor-
//!   tunable through the fleet constructors' `new_with_pin_ttl`
//!   instrument — the store's fleet `pin_ttl` parameter, ADR-008 §2).
//!   The store owns the TTL and the TTL/3 cadence — `pin_commit`'s pin
//!   rows carry the store-resolved expiry; renewals take the
//!   successor-time shape recorded here.
//! - **Timestamp encoding** — signed i64 nanoseconds from the Unix
//!   epoch (`time_to_nanos`), carried as `BIGINT` — the crate-wide
//!   encoding (the sqlite engine's helpers are reference, not a
//!   dependency: per-engine bodies, one encoding).
//! - **The sweep lease** — the session-scoped pg advisory lock:
//!   `pg_try_advisory_lock` (atomic test-and-set) /
//!   `pg_advisory_unlock`, session scope being the only scope whose
//!   holdership ends at owner death or engine-session loss (ADR-013
//!   §2). The lock key is a fixed, namespaced integer (`sweep_lock_key`:
//!   FNV-1a over `alkblobs:sweep:{schema}`, folded to positive i64) —
//!   co-tenant engines on one instance (different schemas) never
//!   collide, while the fleet's shared engine-owned schema is one lock
//!   domain (one sweeper per pool). **The lease holds its connection**:
//!   the acquired checkout is stored under a short-held
//!   [`parking_lot::Mutex`] until release, so acquire and release run
//!   on the same session — advisory locks are per-session, and a
//!   release from another pooled connection would leave the holder's
//!   lock live (the recorded pitfall). A second `sweep_lock_acquire`
//!   on the same backend contended: the mutex answers `Ok(false)`
//!   before any pool checkout. Release is no-op-safe and never errors:
//!   by the call's end holdership has ended — unlocked, or already
//!   reaped by the holder's session loss (an unlock failure is the
//!   reaping case itself, not a fault to surface).
//! - **Pin rows upsert-renew (the re-put form)** — `pin_commit`'s
//!   insert rides `ON CONFLICT (key, owner) DO UPDATE SET expiry`, and
//!   `gc_pins`' primary key is `(key, owner)` (the row identity):
//!   a re-put by the same instance renews its row (ADR-010 §4), and
//!   protection composes by row (ADR-010 §2).
//! - **`arbitrate` folds under the `FOR UPDATE` finger-lock** — one
//!   `BEGIN … COMMIT` step: un-expired pin rows for the key, then each
//!   registered liveness-table reference, all `SELECT … FOR UPDATE`
//!   keyed at delete time, folded into one bool before the commit. The
//!   engine only classifies — it never decides deletion; the referenced
//!   liveness tables are never created by the engine, and their
//!   references are quoted (`quote_table_reference`) at registration.
//! - **Virgin-store tolerance asymmetry (recorded posture):**
//!   `window_take` maps a missing table (SQLSTATE 42P01) to empty —
//!   the read-tolerance parity with the sqlite twin; the fleet clauses
//!   carry no tolerance (they run in fleet mode, where store
//!   construction has ensured the GC schema), so a missing GC table
//!   surfaces the engine error.
//!
//! Tests need a live server: connection settings ride the environment
//! (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB`); tests skip cleanly
//! server-less; acceptance requires green against the harness server
//! (dockerized `pglo-poc`, `postgres:16-alpine`, :15432, db `blobs`,
//! 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 allowances are scoped to the genuinely not-yet-consumed
//! items with the consumer named at each: the crate-internal
//! batch-delete helper (`PostgresBackend::delete_batch` — the
//! sweep-batch shape remaining available to batch consumers; the store
//! core's delete paths arbitrate and delete per key) and the
//! same-instance probe's engine read (`system_identifier` — consumed
//! by the pg-lo fleet constructor, feature `pg-lo`).
//!
//! **The split layout (wave-8 structure, task `split-postgres-pg-lo`):**
//! the engine is a directory module — `pool.rs` the constructor family
//! and the shipped pin-TTL consts, `schema.rs` the bootstrap/GC DDL and
//! the catalog probe, `cas.rs` the `Backend` row body (put, get, the
//! stable-cursor list, size/delete) with the cursor pair and the
//! crate-internal batch-delete helper, `pins.rs` the pin clause family
//! and the liveness-registration half, `windows.rs` the window family,
//! `joint.rs` the joint tx and the sweep lease, `testing.rs` the shared
//! test fixtures. The two shared structs (`PostgresBackend`,
//! `PgJointTx`) stay on the module root so their private fields remain
//! sibling-visible; the public paths (`crate::postgres::PostgresCursor`
//! and friends) are the intra-module re-exports below.

/// 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;

/// 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;

mod cas;
mod joint;
mod pins;
mod pool;
mod schema;
#[cfg(test)]
pub(crate) mod testing;
mod windows;

pub use cas::{PostgresCursor, PostgresStaging};
pub(crate) use pool::DEFAULT_PIN_TTL_SECS;

use std::sync::Arc;
use std::time::Duration;

use crate::pg_common::detached_batch_execute;
use crate::Error;

use deadpool_postgres::{Object, Pool};
use parking_lot::Mutex;

/// The `postgres` engine — a deadpool pool over tokio-postgres plus the
/// quoted engine-owned table reference. Cloning shares the pool (the
/// staging cursor handout's out-of-band commit requires it); the pool
/// is the concurrency posture.
pub(crate) struct PostgresBackend {
    pool: Pool,
    /// The engine-owned table reference, quoted once at construction:
    /// `"schema"."blobs"` — the quoted-identifier boundary every
    /// statement composes through.
    table: String,
    /// The engine-owned schema name (the sweep lease's lock-key
    /// namespace and the qualified GC tables' parent).
    schema: String,
    /// The GC pin table reference, quoted once at construction,
    /// `"schema"."gc_pins"` (the fleet family's row home).
    pins_table: String,
    /// The GC window table reference, quoted once at construction,
    /// `"schema"."gc_window"` (the window family's row home).
    window_table: String,
    /// The pin-TTL the successor-time source renews against (the
    /// recorded design point: ADR-008's default 60 s, shipped as a
    /// backend constant; the fleet constructors tune it through
    /// [`PostgresBackend::new_with_pin_ttl`] — the store's fleet
    /// `pin_ttl` parameter landed).
    pin_ttl: Duration,
    /// The sweep lease's held checkout: `Some` exactly while the lease
    /// is held — the same-connection acquire/release discipline (the
    /// advisory lock is per-session; the release-on-another-connection
    /// hazard is recorded in the module doc).
    lease: Mutex<Option<Object>>,
    /// The registered liveness-table references (ADR-012 §6.6's fleet
    /// registration's engine-side home): quoted at registration, held
    /// as references only — the tables are never created by the engine
    /// (the schema-ownership rule).
    liveness_sources: Arc<Mutex<Vec<String>>>,
    /// Test-only commit-failure injection, standing in for the media
    /// failure the kv tier cannot reliably produce in tests; mirrors
    /// the sqlite engine's knob so the stage-hygiene failure form is
    /// exercised against the real engine.
    #[cfg(test)]
    fail_commit: bool,
    /// Test-only joint-commit-failure injection for the `pin_commit`
    /// clause (the joint tx atomicity test's forced-failure form).
    #[cfg(test)]
    fail_joint_commit: bool,
    /// 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 `postgres` engine's [`EngineFleetState::JointTx`] — the pooled
/// checkout held for the tx's life, carrying a session-level `BEGIN`
/// opened by [`PostgresBackend::begin_joint_tx`] and closed by the
/// clause bodies (`COMMIT` in `pin_commit`'s success path, `ROLLBACK`
/// on every failure path). Exclusive checkout ownership is what makes
/// the session-level transaction sound; dropping the value before
/// `pin_commit` is outside the seam's contract (the store core holds
/// the tx — ADR-010 §2's ownership rule, pinned in the trait), and it
/// is a load-bearing exclusion: the clause bodies' failure paths close
/// the tx themselves (`ROLLBACK`, then `Err`), but a *mid-flight* drop
/// (an error path between [`PostgresBackend::begin_joint_tx`] and
/// [`PostgresBackend::pin_commit`] leaking through `?`) strands the
/// open transaction on its checkout — deadpool's `Fast` recycle never
/// resets transaction state, so the pool would hand the dirty session
/// to the next checkout. The store core's joint-tx driver (wave 6)
/// therefore owes pin_commit on **every** path from `begin_joint_tx`
/// onward; the recorded wave-4 review amendment names this duty. The
/// pg-lo joint contribution composes with the duty: its failure paths
/// never close the tx themselves — the contribution returns with the
/// tx open (no partial LO state that outlives the termination's
/// rollback; transactional LO creation), and the driver terminates.
/// A mid-flight drop (a *cancellation* — a task abort, a `select!`, a
/// `timeout` — dropping the future between
/// [`PostgresBackend::begin_joint_tx`] and
/// [`PostgresBackend::pin_commit`] is the recorded hazard, review-001
/// F-2: the store core's catch-propagate-terminate covers an **error**
/// path but cannot catch a cancellation) is defended by an `impl Drop`
/// aliasing `PgLoStaging`'s spawned-ROLLBACK defense: the tx's held
/// checkout is handed to a spawned best-effort `ROLLBACK`
/// ([`crate::pg_common::detached_batch_execute`]) before the pool ever
/// receives the session back — the open transaction never rides an
/// idle session back into the pool (deadpool's `Fast` recycle never
/// resets transaction state — the recorded pitfall).
///
/// **The guard's trigger conditions, on the type:** the guard fires
/// exactly when the tx is dropped with its checkout still in hand —
/// i.e. dropped without [`EngineFleetState::pin_commit`] having run.
/// `pin_commit` consumes the checkout itself (it takes the client and
/// issues `COMMIT` on its success path, `ROLLBACK` on every failure
/// path), so a tx that reached `pin_commit` — normally or through its
/// error arms — drops checkout-less and the guard is a structural
/// no-op. A terminated-by-the-driver tx (`terminate_joint_tx`'s
/// `ROLLBACK`) drops after that rollback succeeded, also checkout-less.
/// There is no path where the checkout is in hand while the
/// transaction is already terminated, so "checkout present at drop" is
/// the exact "open transaction stranded" condition.
pub(crate) struct PgJointTx {
    /// The held pooled checkout — `Some` exactly while the tx is open
    /// (the guard's trigger condition, documented on the type).
    /// `None` once `pin_commit` (or — defensively — the Drop body) has
    /// taken it for termination.
    client: Option<Object>,
    blobs_table: String,
    pins_table: String,
    #[cfg(test)]
    fail_commit: bool,
}

impl PgJointTx {
    /// The held checkout, by reference — the statement surface every
    /// joint contribution rides. An already-consumed tx (dropped
    /// defensively-typed here, never a panic) surfaces as
    /// [`Error::Io`]; the state is unreachable on any live path
    /// (`pin_commit` consumes the whole value).
    pub(crate) fn client(&self) -> Result<&Object, Error> {
        self.client
            .as_ref()
            .ok_or_else(|| Error::Io("postgres: joint tx already consumed".to_owned()))
    }
}

impl Drop for PgJointTx {
    /// The cancellation guard (review-001 F-2): a tx dropped with its
    /// checkout still in hand holds an open transaction — hand the
    /// checkout to a spawned best-effort `ROLLBACK` (aliasing
    /// `PgLoStaging`'s defense via
    /// [`crate::pg_common::detached_batch_execute`]) so the pool never
    /// receives an idle-in-transaction session. Txs reaching
    /// `pin_commit` drop checkout-less; this is a no-op on those paths.
    fn drop(&mut self) {
        if let Some(client) = self.client.take() {
            detached_batch_execute(client, "ROLLBACK".to_owned());
        }
    }
}