feat(pg-lo): the large-tier engine — LOs + companion table behind the Backend trait (wave 5)

Implement the pg-lo engine (ADR-009, POC #7 C1-C7) in src/pg_lo.rs
behind feature pg-lo (default-off), plus the shared-helper home the
task's design points resolve into src/pg_common.rs.

- Engine: seven Backend methods + put_staged; the lo_entries companion
  table is the contract authority (list/size/has/CAS table-level;
  pg_largeobject read only by the orphan sweep's candidate enumeration);
  one-tx puts with 512 KiB lowrite statements (LOWRITE_CHUNK, the C2
  knee), the descriptorless lo_get window cursor (READ_CHUNK; EOF
  clamping + the unexpected-EOF clean failure), the held-tx staging
  cursor aggregating 64 KiB windows, one-tx deletes, the complete
  single-snapshot list, the 42P01 read-only virgin-store split, and the
  crate-internal reap_orphans (scope-noted, planted-orphan contract
  test). Posture deltas in the module doc (catalog space never
  returned / VACUUM FULL runbook line / autovacuum inherited / the
  cached-get capacity facts). Nothing of EngineFleetState; the window
  family remains engine-pg-lo-window.
- Shared helpers (pg_err, 42P01, quote_identifier, size clamp, i64-nanos
  encoding, build_pool, PgOpts, test janitors) move to src/pg_common.rs
  gated cfg(any(postgres, pg-lo)); PgOpts's public re-export widens
  (definition moved; crate::postgres::PgOpts path held by re-export);
  src/postgres.rs's changes are mechanical swaps.
- LO isolation (C7 lesson): per-test dedicated databases + DatabaseJanitor
  (WITH (FORCE)); zero LO/schema/database residue on the shared harness
  db (pre-existing public.lo_entries exploration residue found and
  removed).
- 18 in-module tests, env-driven (ALKBLOBS_PG_*), skipping cleanly
  server-less; all green against the pglo-poc harness server.

Verified: cargo test (69/19/3), --features pg-lo live (87/19/3, 18/18
pg-lo), --features postgres live (103/36/3), --all-features (176/71/4);
clippy -D warnings green for default, pg-lo, postgres, all-features,
--no-default-features --features "pg-lo,mem" and "postgres,pg-lo";
cargo fmt --check; cargo doc --no-deps.
This commit is contained in:
glm-5.3-flash committed 2026-10-12 07:31:46 +00:00
1 parent 5a217f7bc7
commit d69d9a345b
5 files changed
+2513 -225

No files matched your search

+5 -2
View File
@@ -27,11 +27,14 @@ pub mod pg_lo;
#[cfg(feature = "ops")] #[cfg(feature = "ops")]
pub mod ops; pub mod ops;
#[cfg(any(feature = "postgres", feature = "pg-lo"))]
mod pg_common;
#[cfg(feature = "postgres")] #[cfg(feature = "postgres")]
pub mod postgres; pub mod postgres;
#[cfg(feature = "postgres")] #[cfg(any(feature = "postgres", feature = "pg-lo"))]
pub use postgres::PgOpts; pub use pg_common::PgOpts;
#[cfg(feature = "sqlite")] #[cfg(feature = "sqlite")]
pub mod sqlite; pub mod sqlite;
+321
View File
@@ -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<Pool, Error> {
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();
}
}
}
+2056 -2
View File
File diff suppressed because it is too large. Load diff
+22 -218
View File
@@ -196,121 +196,28 @@
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; 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 parking_lot::Mutex;
use tokio_postgres::NoTls;
use crate::backend::{clamped, Backend, ReadCursor, StagingCursor}; use crate::backend::{clamped, Backend, ReadCursor, StagingCursor};
use crate::gc::{EngineFleetState, EngineWindowState, PinRow, WindowCandidate}; 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}; use crate::{Error, PinState};
fn err(e: impl std::fmt::Display) -> Error { /// The test-support helpers' shared home — the SchemaJanitor lives in
Error::Io(format!("postgres: {e}")) /// `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'
/// The engine's error mapping for the clause bodies: a SQL-level /// `crate::postgres::test_support` path holds.
/// failure surfaces with its server-side message and SQLSTATE (the #[cfg(test)]
/// `DbError` detail `Display` drops), anything else through [`err`]. pub(crate) use crate::pg_common::test_support;
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 `postgres` engine — a deadpool pool over tokio-postgres plus the /// The `postgres` engine — a deadpool pool over tokio-postgres plus the
/// quoted engine-owned table reference. Cloning shares the pool (the /// quoted engine-owned table reference. Cloning shares the pool (the
@@ -408,7 +315,7 @@ impl PostgresBackend {
/// `new_tableless` pattern. /// `new_tableless` pattern.
#[cfg(test)] #[cfg(test)]
async fn new_schemaless(config: &str, schema: &str) -> Result<Self, Error> { async fn new_schemaless(config: &str, schema: &str) -> Result<Self, Error> {
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 schema = schema.to_owned();
let pins_table = qualified_table(&schema, "gc_pins"); let pins_table = qualified_table(&schema, "gc_pins");
let window_table = qualified_table(&schema, "gc_window"); 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<Pool, Error> {
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 engine-owned bootstrap DDL — idempotent, no migration machinery.
/// The B5 row shape under the measured `fillfactor = 90`; column names /// The B5 row shape under the measured `fillfactor = 90`; column names
/// quote-follow the alkstore hazard-list discipline (`"key"` quoted; /// 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. /// source): every renewed row's expiry moves to `now` + this.
const DEFAULT_PIN_TTL: Duration = Duration::from_secs(DEFAULT_PIN_TTL_SECS); const DEFAULT_PIN_TTL: Duration = Duration::from_secs(DEFAULT_PIN_TTL_SECS);
/// The window row family's timestamp encoding (the recorded design /// The gc-window/pin rows' timestamp decoding — the test tunings'
/// point): signed nanoseconds since the Unix epoch — sub-second /// fixture helper (the sqlite engine's helpers are the reference —
/// precision for the test tunings, negative for pre-epoch instants, /// per-engine bodies, one encoding: [`crate::pg_common::time_to_nanos`]
/// i64 across the full range a `BIGINT` stores (±292 years around the /// is the shared forward direction).
/// epoch). The sqlite engine's helpers are this one's reference —
/// per-engine bodies, one encoding.
fn time_to_nanos(t: SystemTime) -> i64 {
match t.duration_since(UNIX_EPOCH) {
Ok(d) => d.as_nanos() as i64,
Err(e) => -(e.duration().as_nanos() as i64),
}
}
fn nanos_to_time(n: i64) -> SystemTime { fn nanos_to_time(n: i64) -> SystemTime {
if n >= 0 { if n >= 0 {
UNIX_EPOCH + std::time::Duration::from_nanos(n as u64) UNIX_EPOCH + std::time::Duration::from_nanos(n as u64)
@@ -1191,6 +1055,7 @@ mod tests {
use crate::gc::{EngineFleetState, EngineWindowState}; use crate::gc::{EngineFleetState, EngineWindowState};
use std::ops::Deref; use std::ops::Deref;
use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::atomic::{AtomicU64, Ordering};
use tokio_postgres::NoTls;
const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog"; 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(); 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();
}
}
}
+109 -3
View File
@@ -1,7 +1,7 @@
--- ---
id: engine-pg-lo id: engine-pg-lo
name: pg-lo large engine — Large Objects behind the Backend trait name: pg-lo large engine — Large Objects behind the Backend trait
status: pending status: completed
depends_on: [] depends_on: []
scope: moderate scope: moderate
risk: medium risk: medium
@@ -230,8 +230,114 @@ engine and its in-module unit tests only.
## Notes ## 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 ## Summary
> To be filled on completion > 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`.