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:
1 parent
5a217f7bc7
commit
d69d9a345b
5 files changed
+2513
-225
No files matched your search
+5
-2
@@ -27,11 +27,14 @@ pub mod pg_lo;
|
||||
#[cfg(feature = "ops")]
|
||||
pub mod ops;
|
||||
|
||||
#[cfg(any(feature = "postgres", feature = "pg-lo"))]
|
||||
mod pg_common;
|
||||
|
||||
#[cfg(feature = "postgres")]
|
||||
pub mod postgres;
|
||||
|
||||
#[cfg(feature = "postgres")]
|
||||
pub use postgres::PgOpts;
|
||||
#[cfg(any(feature = "postgres", feature = "pg-lo"))]
|
||||
pub use pg_common::PgOpts;
|
||||
|
||||
#[cfg(feature = "sqlite")]
|
||||
pub mod sqlite;
|
||||
@@ -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
File diff suppressed because it is too large.
Load diff
+22
-218
@@ -196,121 +196,28 @@
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use deadpool_postgres::{Manager, ManagerConfig, Object, Pool, RecyclingMethod};
|
||||
use deadpool_postgres::{Object, Pool};
|
||||
use parking_lot::Mutex;
|
||||
use tokio_postgres::NoTls;
|
||||
|
||||
use crate::backend::{clamped, Backend, ReadCursor, StagingCursor};
|
||||
use crate::gc::{EngineFleetState, EngineWindowState, PinRow, WindowCandidate};
|
||||
/// The shared opts type's re-export home for the `crate::postgres::PgOpts`
|
||||
/// path (its definition lives in [`crate::pg_common`]; the crate-root
|
||||
/// re-export is `any(postgres, pg-lo)` — the widened gate).
|
||||
pub use crate::pg_common::PgOpts;
|
||||
use crate::pg_common::{
|
||||
build_pool, err, is_undefined_table, pg_err, qualified_table, quote_identifier, size_u64,
|
||||
time_to_nanos, DEFAULT_SCHEMA,
|
||||
};
|
||||
use crate::{Error, PinState};
|
||||
|
||||
fn err(e: impl std::fmt::Display) -> Error {
|
||||
Error::Io(format!("postgres: {e}"))
|
||||
}
|
||||
|
||||
/// The engine's error mapping for the clause bodies: a SQL-level
|
||||
/// failure surfaces with its server-side message and SQLSTATE (the
|
||||
/// `DbError` detail `Display` drops), anything else through [`err`].
|
||||
fn pg_err(e: tokio_postgres::Error) -> Error {
|
||||
match e.as_db_error() {
|
||||
Some(db) => Error::Io(format!("postgres: {db}")),
|
||||
None => err(e),
|
||||
}
|
||||
}
|
||||
|
||||
/// The virgin-store tolerance's discriminant (POC #1 finding 2): a
|
||||
/// statement naming a table that does not exist maps on the read paths
|
||||
/// to absent/empty (and delete to Missing), never an error. The stable
|
||||
/// surface PostgreSQL carries is the SQLSTATE class+code (42P01,
|
||||
/// undefined table); a missing *schema* reads identically.
|
||||
fn is_undefined_table(e: &tokio_postgres::Error) -> bool {
|
||||
e.code() == Some(&tokio_postgres::error::SqlState::UNDEFINED_TABLE)
|
||||
}
|
||||
|
||||
/// The `size` column's integer, clamped to the domain the trait probes in.
|
||||
fn size_u64(raw: i64) -> u64 {
|
||||
raw.max(0) as u64
|
||||
}
|
||||
|
||||
/// The default engine-owned schema name (the recorded design point's
|
||||
/// default).
|
||||
pub(crate) const DEFAULT_SCHEMA: &str = "alkblobs";
|
||||
|
||||
/// The shipped default pool size ([`PgOpts::max_size`]) — the alkstore
|
||||
/// `DEFAULT_MAX_SIZE = 8` precedent.
|
||||
pub(crate) const DEFAULT_MAX_SIZE: usize = 8;
|
||||
|
||||
/// Quote a SQL identifier (the Postgres rule: double the embedded `"`).
|
||||
///
|
||||
/// The engine composes schema-qualified SQL from a consumer-supplied
|
||||
/// schema name; this is the quoted-identifier boundary — an identifier
|
||||
/// is never interpolated into SQL text unquoted (the alkstore
|
||||
/// `quote_identifier` pattern).
|
||||
pub(crate) fn quote_identifier(name: &str) -> String {
|
||||
let mut quoted = String::with_capacity(name.len() + 2);
|
||||
quoted.push('"');
|
||||
for c in name.chars() {
|
||||
if c == '"' {
|
||||
quoted.push('"');
|
||||
}
|
||||
quoted.push(c);
|
||||
}
|
||||
quoted.push('"');
|
||||
quoted
|
||||
}
|
||||
|
||||
/// A schema-qualified table reference: `"schema"."table"`.
|
||||
pub(crate) fn qualified_table(schema: &str, table: &str) -> String {
|
||||
format!("{}.{}", quote_identifier(schema), quote_identifier(table))
|
||||
}
|
||||
|
||||
/// Construction options for the postgres engine. Durability knobs,
|
||||
/// pool sizing, and the engine-owned schema name are
|
||||
/// engine-configuration concerns — they ride the store constructors'
|
||||
/// parameters, never invented defaults (the alkstore `PgOpts`
|
||||
/// precedent; `open(config, PgOpts)` adapted to the named-constructor
|
||||
/// shape — the constructors are [`Store::dual_tier_postgres_local`] and
|
||||
/// [`Store::kv_only_postgres`]).
|
||||
///
|
||||
/// The connection settings themselves do not ride this struct — they
|
||||
/// are the constructor's separate connection-settings argument (the
|
||||
/// tokio-postgres [`tokio_postgres::Config`]-parseable form); the
|
||||
/// engine never invents defaults for them. TLS: the engine hardwires
|
||||
/// `NoTls` on every connection path regardless of the config's
|
||||
/// sslmode — v1 TLS is a post-v1 deployment concern, mirroring the
|
||||
/// alkstore posture.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct PgOpts {
|
||||
/// The engine-owned PostgreSQL schema holding the pool table.
|
||||
/// Default [`DEFAULT_SCHEMA`]; an empty string selects the default.
|
||||
/// The name is a quoted-identifier boundary ([`quote_identifier`])
|
||||
/// — reserved-word and hostile names are safe.
|
||||
pub schema: String,
|
||||
/// Upper bound on pooled connections (deadpool's `max_size`).
|
||||
/// Default [`DEFAULT_MAX_SIZE`] = 8. A value of `0` fails
|
||||
/// construction immediately (validated at entry, no round trip —
|
||||
/// deadpool cannot grant checkouts from an empty pool; the
|
||||
/// alkstore precedent).
|
||||
pub max_size: usize,
|
||||
/// The per-session `synchronous_commit` durability knob, wired as a
|
||||
/// connect option (`-c synchronous_commit=…`) on every pooled
|
||||
/// connection (POC-verified per-session SET mechanics; the
|
||||
/// consumer's own `options=` DSN values are preserved alongside —
|
||||
/// postgres parses repeated `-c` flags). Default `true` — the ship
|
||||
/// config; `false` is the documented sqlite-NORMAL-alike latency
|
||||
/// trade (B5 ~2-3×).
|
||||
pub synchronous_commit: bool,
|
||||
}
|
||||
|
||||
impl Default for PgOpts {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
schema: DEFAULT_SCHEMA.to_owned(),
|
||||
max_size: DEFAULT_MAX_SIZE,
|
||||
synchronous_commit: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
/// The test-support helpers' shared home — the SchemaJanitor lives in
|
||||
/// `src/pg_common.rs`'s `#[cfg(test)] test_support` (gated
|
||||
/// `cfg(test)` + `any(postgres, pg-lo)`, the shared-helper design
|
||||
/// point's home); re-exported here so the sibling test suites'
|
||||
/// `crate::postgres::test_support` path holds.
|
||||
#[cfg(test)]
|
||||
pub(crate) use crate::pg_common::test_support;
|
||||
|
||||
/// The `postgres` engine — a deadpool pool over tokio-postgres plus the
|
||||
/// quoted engine-owned table reference. Cloning shares the pool (the
|
||||
@@ -408,7 +315,7 @@ impl PostgresBackend {
|
||||
/// `new_tableless` pattern.
|
||||
#[cfg(test)]
|
||||
async fn new_schemaless(config: &str, schema: &str) -> Result<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 pins_table = qualified_table(&schema, "gc_pins");
|
||||
let window_table = qualified_table(&schema, "gc_window");
|
||||
@@ -488,40 +395,6 @@ impl PostgresBackend {
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the deadpool pool with the shipped posture: per-session
|
||||
/// `synchronous_commit` merged into every connection's connect options
|
||||
/// (appended to the consumer's own `options=` DSN values — postgres
|
||||
/// parses repeated `-c` flags), `RecyclingMethod::Fast`, the configured
|
||||
/// `max_size`. `max_size = 0` fails at entry, before any round trip.
|
||||
fn build_pool(config: &str, max_size: usize, synchronous_commit: bool) -> Result<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 B5 row shape under the measured `fillfactor = 90`; column names
|
||||
/// quote-follow the alkstore hazard-list discipline (`"key"` quoted;
|
||||
@@ -549,19 +422,10 @@ pub(crate) const DEFAULT_PIN_TTL_SECS: u64 = 60;
|
||||
/// source): every renewed row's expiry moves to `now` + this.
|
||||
const DEFAULT_PIN_TTL: Duration = Duration::from_secs(DEFAULT_PIN_TTL_SECS);
|
||||
|
||||
/// The window row family's timestamp encoding (the recorded design
|
||||
/// point): signed nanoseconds since the Unix epoch — sub-second
|
||||
/// precision for the test tunings, negative for pre-epoch instants,
|
||||
/// i64 across the full range a `BIGINT` stores (±292 years around the
|
||||
/// epoch). The sqlite engine's helpers are this one's reference —
|
||||
/// per-engine bodies, one encoding.
|
||||
fn time_to_nanos(t: SystemTime) -> i64 {
|
||||
match t.duration_since(UNIX_EPOCH) {
|
||||
Ok(d) => d.as_nanos() as i64,
|
||||
Err(e) => -(e.duration().as_nanos() as i64),
|
||||
}
|
||||
}
|
||||
|
||||
/// The gc-window/pin rows' timestamp decoding — the test tunings'
|
||||
/// fixture helper (the sqlite engine's helpers are the reference —
|
||||
/// per-engine bodies, one encoding: [`crate::pg_common::time_to_nanos`]
|
||||
/// is the shared forward direction).
|
||||
fn nanos_to_time(n: i64) -> SystemTime {
|
||||
if n >= 0 {
|
||||
UNIX_EPOCH + std::time::Duration::from_nanos(n as u64)
|
||||
@@ -1191,6 +1055,7 @@ mod tests {
|
||||
use crate::gc::{EngineFleetState, EngineWindowState};
|
||||
use std::ops::Deref;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use tokio_postgres::NoTls;
|
||||
|
||||
const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";
|
||||
|
||||
@@ -2794,64 +2659,3 @@ mod tests {
|
||||
backend.sweep_lock_release().await.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
/// Test-support helpers shared with the crate's other postgres-attached
|
||||
/// test modules (`src/store.rs`'s construction tests) — crate-internal
|
||||
/// and compiled out of every non-test build.
|
||||
#[cfg(test)]
|
||||
pub(crate) mod test_support {
|
||||
use super::quote_identifier;
|
||||
use tokio_postgres::NoTls;
|
||||
|
||||
/// Drops a test's engine-owned schema when the test ends — success,
|
||||
/// early return, or unwind (Drop runs in all three), so a fresh
|
||||
/// schema per test never accumulates on the shared harness
|
||||
/// database. No async teardown exists at Drop, so the drop executes
|
||||
/// synchronously from a std thread over a one-shot current-thread
|
||||
/// runtime; an unreachable harness server is tolerated (the residue
|
||||
/// stays; nothing else breaks).
|
||||
///
|
||||
/// Construct it (or a wrapper holding it) *before* the store or
|
||||
/// backend it outlives: drop order is reverse of declaration, so
|
||||
/// the backend's pool drains before the schema drops.
|
||||
pub(crate) struct SchemaJanitor {
|
||||
dsn: String,
|
||||
schema: String,
|
||||
}
|
||||
|
||||
impl SchemaJanitor {
|
||||
pub(crate) fn new(dsn: &str, schema: &str) -> SchemaJanitor {
|
||||
SchemaJanitor {
|
||||
dsn: dsn.to_owned(),
|
||||
schema: schema.to_owned(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for SchemaJanitor {
|
||||
fn drop(&mut self) {
|
||||
let dsn = self.dsn.clone();
|
||||
let schema = self.schema.clone();
|
||||
let _ = std::thread::spawn(move || {
|
||||
let Ok(rt) = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
else {
|
||||
return;
|
||||
};
|
||||
rt.block_on(async move {
|
||||
let Ok((client, conn)) = tokio_postgres::connect(&dsn, NoTls).await else {
|
||||
return;
|
||||
};
|
||||
tokio::spawn(conn);
|
||||
let sql = format!(
|
||||
"DROP SCHEMA IF EXISTS {} CASCADE",
|
||||
quote_identifier(&schema)
|
||||
);
|
||||
let _ = client.execute(sql.as_str(), &[]).await;
|
||||
});
|
||||
})
|
||||
.join();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user