feat(store): the postgres constructor wiring — the fleet-capable store shapes (wave 4)
- `KvEngine::Postgres` slotted in: all six kv-op dispatch arms, the `KvCursor::Postgres` materializing body behind `AnyCursor`'s kv arm (the recorded sub-enum design point, landed as predicted), and the engine-neutral `ensure_gc_schema` call creating `gc_pins` + `gc_window` in the fresh schema at construction (the fleet host's both-tables row, ADR-012 §5) - The two named constructors behind feature `postgres` (default-off): `dual_tier_postgres_local(config, opts, root, threshold, window_delay)` — the fleet-capable-kv production shape — and `kv_only_postgres(config, opts, threshold, window_delay)`; the DSN connection settings + engine opts ride the constructor parameters, never invented defaults. `PgOpts` is pub and re-exported from `lib.rs` (it is on the public constructor signature now) - The sweep's durable form runs on the postgres host: the form body extracted into `sweep_window_durable<W: EngineWindowState>` (sqlite's arm delegates, behavior unchanged), both SQL-backed arms keyed off the kv-engine enum, never a flag — wave 5's pg-lo dormancy slots in without reshaping - Non-fleet posture recorded on both constructors (ADR-012 §3): implicitly `fleet: false`, the ADR-005 in-process protocol verbatim, the `fleet: true` declaration is wave 6 and slots in without reshaping - Stale scope notes updated to the landed state (engine-representation doc, `KvCursor` design point, fleet-declaration paragraph, one-window-host paragraph, postgres module doc, gc.rs companion-seam note); the batch-delete helper's scope note stays (wave 6 consumer) - 5 new env-driven construction tests (skipping server-less): the modes construct and introspect, both GC tables exist in the fresh schema, the window-delay default is the pinned 5-min constant and tunable (`0` legal), the kv-only rejection wiring is unchanged, the dual-tier shape round-trips through both tiers, and the durable form is observed staging/executing across sweeps on the postgres host Verification: `cargo test` 69/19/3 ok; `cargo test --features postgres` 102/19/3 ok live (pglo-poc harness) and green server-less; `cargo test --no-default-features --features postgres` (the pg-pure build, ADR-013 §4) 75/2/3 ok server-less; `cargo test --all-features` 157/54/4 ok; clippy `-D warnings` green (default, postgres, pg-pure, all-features); `cargo fmt --check` green; `cargo doc --no-deps` clean.
This commit is contained in:
1 parent
29209fe53f
commit
67b927b13c
5 files changed
+842
-72
No files matched your search
@@ -66,9 +66,11 @@
|
||||
//!
|
||||
//! The `#![allow(dead_code)]` below covers the companion-seam scope:
|
||||
//! the window family has its in-tree callers (store construction runs
|
||||
//! `ensure_gc_schema` on the SQL-backed kv engine; the sweep's durable
|
||||
//! `ensure_gc_schema` on every SQL-backed kv engine; the sweep's durable
|
||||
//! form drives `window_stage`/`window_take`), while the fleet family
|
||||
//! waits for postgres's wave. It shrinks as that lands.
|
||||
//! waits for wave 6 — its postgres clause bodies landed with wave 4,
|
||||
//! but the store core consumes the family only under the `fleet: true`
|
||||
//! declaration. It shrinks as that lands.
|
||||
|
||||
#![allow(dead_code)]
|
||||
|
||||
|
||||
@@ -30,5 +30,8 @@ pub mod ops;
|
||||
#[cfg(feature = "postgres")]
|
||||
pub mod postgres;
|
||||
|
||||
#[cfg(feature = "postgres")]
|
||||
pub use postgres::PgOpts;
|
||||
|
||||
#[cfg(feature = "sqlite")]
|
||||
pub mod sqlite;
|
||||
+36
-11
@@ -114,7 +114,9 @@
|
||||
//! lease. The store core drives these clauses in wave 6 (fleet mode) —
|
||||
//! under `fleet: false` none of the fleet clauses run, while
|
||||
//! `ensure_gc_schema` runs on every postgres store construction (the
|
||||
//! `store-postgres-construction` wiring). The embedder's liveness table
|
||||
//! `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
|
||||
@@ -262,20 +264,23 @@ pub(crate) fn qualified_table(schema: &str, table: &str) -> String {
|
||||
format!("{}.{}", quote_identifier(schema), quote_identifier(table))
|
||||
}
|
||||
|
||||
/// Construction options for the [`PostgresBackend`]. Durability knobs,
|
||||
/// Construction options for the postgres engine. Durability knobs,
|
||||
/// pool sizing, and the engine-owned schema name are
|
||||
/// engine-configuration concerns — they never appear on the contract
|
||||
/// surface (the alkstore `PgOpts` precedent; the constructor wiring
|
||||
/// itself is `store-postgres-construction`).
|
||||
/// 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 first 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.
|
||||
/// 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(crate) struct PgOpts {
|
||||
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`])
|
||||
@@ -461,6 +466,26 @@ impl PostgresBackend {
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The catalog table-existence probe — test-only: the store
|
||||
/// construction test's instrument (mirrors the sqlite engine's
|
||||
/// `sqlite_master` probe), 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)
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the deadpool pool with the shipped posture: per-session
|
||||
|
||||
+703
-49
@@ -26,10 +26,11 @@
|
||||
//! collapsing to the concrete facade type across all factories"). The
|
||||
//! public surface is engine-blind: consumers name modes and engines
|
||||
//! only through named constructors, feature-gated with their engine
|
||||
//! (`mem_only` needs feature `mem`; the sqlite/postgres/pg-lo
|
||||
//! constructors arrive with waves 3–5 and slot into the same enums and
|
||||
//! constructor pattern without reshaping — adding an enum variant and
|
||||
//! a named constructor is slotting in).
|
||||
//! (`mem_only` needs feature `mem`; the sqlite constructors are the
|
||||
//! default build's — wave 3; the postgres constructors arrived with
|
||||
//! wave 4 behind feature `postgres`; pg-lo slots in at wave 5 — and
|
||||
//! each slotting is exactly adding an enum variant and a named
|
||||
//! constructor, without reshaping).
|
||||
//!
|
||||
//! **Constructor modes** (ADR-011 §4; ADR-012 §6.3):
|
||||
//!
|
||||
@@ -39,14 +40,18 @@
|
||||
//! ADR-012 §6.3 names (both tiers load-bearing); the wave-3
|
||||
//! constructible shape is `kv = sqlite` + `large = local` — the
|
||||
//! **default-build production shape** (feature `sqlite` is
|
||||
//! default-on, ADR-013 §4). Over-threshold content routes to the
|
||||
//! large tier; at-or-below to kv.
|
||||
//! default-on, ADR-013 §4); the wave-4 shape is `kv = postgres` +
|
||||
//! `large = local` — the **fleet-capable-kv production shape** (the
|
||||
//! engine is the fleet host, ADR-012 §4, but the constructor is the
|
||||
//! non-fleet declaration until wave 6). Over-threshold content
|
||||
//! routes to the large tier; at-or-below to kv.
|
||||
//! - **kv-only**: no large tier; over-threshold puts reject **at put
|
||||
//! time** (ADR-012 §6.2 — known-length immediately; unknown-length at
|
||||
//! mid-stream threshold overflow). The rejection behavior itself
|
||||
//! lands with the put seam; the threshold this mode carries is the
|
||||
//! rejection point's value. The wave-2 engine is mem (testing);
|
||||
//! the wave-3 engine is sqlite (the durable single-node medium).
|
||||
//! the wave-3 engine is sqlite (the durable single-node medium); the
|
||||
//! wave-4 engine is postgres (the fleet host's kv-only shape).
|
||||
//! - **mem-only**: the mem engine alone, **no dispatch threshold in
|
||||
//! effect** (everything is local and small) — a testing/embedder-
|
||||
//! ephemeral posture, never production. Structurally distinct from
|
||||
@@ -179,15 +184,21 @@
|
||||
//! group's pins drop together on [`Batch::commit`] or drop (the
|
||||
//! `gc-protocol` machinery drives).
|
||||
//!
|
||||
//! **The fleet declaration defers to the engine waves.** ADR-012 §3
|
||||
//! rules fleet mode a constructor declaration (`fleet: true`), never
|
||||
//! inferred — but it is a constructor *error* for every non-fleet-valid
|
||||
//! configuration, and no fleet-valid configuration is constructible yet
|
||||
//! (mem is never fleet-valid; `local` needs the shared-media
|
||||
//! declaration of ADR-008 §3; the fleet-valid engines — postgres, pg-lo
|
||||
//! — are waves 4–5). Adding the declaration then is additive; carrying
|
||||
//! a flag that could only ever be rejected today would be dead
|
||||
//! machinery.
|
||||
//! **The fleet declaration defers to wave 6.** ADR-012 §3 rules fleet
|
||||
//! mode a constructor declaration (`fleet: true`), never inferred —
|
||||
//! but it is a constructor *error* for every non-fleet-valid
|
||||
//! configuration, and no fleet-valid configuration is constructible
|
||||
//! yet (mem is never fleet-valid; `local` needs the shared-media
|
||||
//! declaration of ADR-008 §3; pg-lo's wave-5 constructor completes the
|
||||
//! fleet-valid engine set). The fleet-capable engines' constructors
|
||||
//! — postgres's pair, landed wave 4 — are **implicitly `fleet: false`**
|
||||
//! (ADR-012 §3): the ADR-005 in-process protocol applies verbatim
|
||||
//! regardless of engine — pins are in-process guards, the sweep lease
|
||||
//! is the in-process lock, and none of the `EngineFleetState` clauses
|
||||
//! run — with the posture recorded in each constructor's doc so wave
|
||||
//! 6's `fleet: true` declaration (the fleet-validity predicate plus
|
||||
//! the constructor declarations) slots in without reshaping: no fleet
|
||||
//! parameter is carried that could only ever be rejected today.
|
||||
//!
|
||||
//! **The dispatch seam the ops tasks program through:** tier selection
|
||||
//! for a given length (`Store::tier_for` composing
|
||||
@@ -199,9 +210,9 @@
|
||||
//! the gc-protocol task fills.
|
||||
//!
|
||||
//! **The durable delete window hosts on the kv engine when it is
|
||||
//! SQL-backed (ADR-012 §5's one-window-host rule).** Sqlite — the
|
||||
//! default build's kv engine — hosts: its constructors run the GC
|
||||
//! schema's engine-neutral call (`ensure_gc_schema` on the
|
||||
//! SQL-backed (ADR-012 §5's one-window-host rule).** Sqlite and
|
||||
//! postgres — the SQL-backed kv engines — host: their constructors run
|
||||
//! the GC schema's engine-neutral call (`ensure_gc_schema` on the
|
||||
//! `EngineWindowState` seam) at construction, dispatched on the
|
||||
//! kv-engine enum — a mem arm is structurally nothing, the in-process
|
||||
//! engines execute the ADR-005 window protocol inline, so there is no
|
||||
@@ -209,12 +220,12 @@
|
||||
//! form**, ADR-008 §2 staged re-arbitrated delete; ADR-012 §6.7 the
|
||||
//! executor is the next sweep) stages its candidates on the kv engine
|
||||
//! the constructor built. The form branch keys off the engine
|
||||
//! representation (`sweep_window`'s one kv-engine-enum match
|
||||
//! arm, never a flag): the durable form runs where the host is
|
||||
//! SQL-backed, the inline form verbatim on mem/local, so wave 5's
|
||||
//! pg-lo dormancy (the dual-SQL instance's large-engine window clauses
|
||||
//! are dormant when the kv engine already hosts) slots in without
|
||||
//! reshaping. Both sqlite constructors carry the delete-window delay
|
||||
//! representation (`sweep_window`'s kv-engine-enum match — each
|
||||
//! SQL-backed arm runs the durable form on its host, never a flag):
|
||||
//! the inline form verbatim on mem/local, so wave 5's pg-lo dormancy
|
||||
//! (the dual-SQL instance's large-engine window clauses are dormant
|
||||
//! when the kv engine already hosts) slots in without reshaping. Every
|
||||
//! SQL-backed constructor carries the delete-window delay
|
||||
//! (ADR-008 §2: default 5 min, constructor-tunable per deployment —
|
||||
//! [`DEFAULT_DELETE_WINDOW_DELAY`]; `0` is a legal degenerate tuning,
|
||||
//! execute-on-next-sweep) as the cadence knob the durable form's
|
||||
@@ -247,12 +258,14 @@ use crate::local::{LocalBackend, LocalCursor, LocalStaging};
|
||||
|
||||
#[cfg(feature = "mem")]
|
||||
use crate::mem::{MemBackend, MemCursor};
|
||||
#[cfg(feature = "postgres")]
|
||||
use crate::postgres::{PgOpts, PostgresBackend, PostgresCursor};
|
||||
#[cfg(feature = "sqlite")]
|
||||
use crate::sqlite::{SqliteBackend, SqliteCursor};
|
||||
|
||||
use crate::Error;
|
||||
|
||||
#[cfg(any(feature = "mem", feature = "sqlite"))]
|
||||
#[cfg(any(feature = "mem", feature = "postgres", feature = "sqlite"))]
|
||||
use std::path::PathBuf;
|
||||
|
||||
use self::StoreMode::{DualTier, KvOnly, MemOnly};
|
||||
@@ -319,7 +332,14 @@ pub enum StoreMode {
|
||||
/// The kv-tier engine set — the closed per-tier engine enum of the
|
||||
/// engine-representation design point. Feature-gated variants slot in
|
||||
/// as their engine waves land: sqlite (wave 3 — landed, the default),
|
||||
/// postgres (wave 4).
|
||||
/// postgres (wave 4 — landed, feature-gated default-off).
|
||||
///
|
||||
/// The variant payloads are structurally dissimilar by design — the
|
||||
/// in-process engine holds one `Arc`, the SQL-backed engines hold
|
||||
/// their connection postures inline — and every dispatch rides the
|
||||
/// enum by reference (`&self` matches only), so the size difference
|
||||
/// costs nothing.
|
||||
#[allow(clippy::large_enum_variant)]
|
||||
pub(crate) enum KvEngine {
|
||||
/// The contract-reference ephemeral engine (ADR-011 §2).
|
||||
#[cfg(feature = "mem")]
|
||||
@@ -330,6 +350,13 @@ pub(crate) enum KvEngine {
|
||||
/// SQL-backed).
|
||||
#[cfg(feature = "sqlite")]
|
||||
Sqlite(SqliteBackend),
|
||||
/// The second SQL-backed engine — the fleet host (ADR-012 §4) and
|
||||
/// a one-window-host alongside sqlite: its constructors run the
|
||||
/// GC schema's engine-neutral call (the postgres row's body
|
||||
/// creates `gc_pins` + `gc_window`) and its sweeps run the durable
|
||||
/// window form on it. Feature `postgres`, default-off.
|
||||
#[cfg(feature = "postgres")]
|
||||
Postgres(PostgresBackend),
|
||||
}
|
||||
|
||||
/// The large-tier engine set. `local` is ungated (dependency-free,
|
||||
@@ -346,7 +373,8 @@ pub(crate) enum LargeEngine {
|
||||
/// get body, `Blob`'s drive) per engine without adding a behavior;
|
||||
/// the sub-enum keeps [`AnyCursor`] a two-arm tier dispatch (kv /
|
||||
/// large) while kv engines multiply inside this one. Recorded for the
|
||||
/// review gate: postgres (wave 4) slots one arm in here.
|
||||
/// review gate and landed with wave 4: postgres slotted one arm in
|
||||
/// here (exactly the predicted shape, no reshaping).
|
||||
pub(crate) enum KvCursor {
|
||||
/// The contract-reference ephemeral body (ADR-011 §2).
|
||||
#[cfg(feature = "mem")]
|
||||
@@ -355,6 +383,11 @@ pub(crate) enum KvCursor {
|
||||
/// once, reads in-memory slices (the sqlite engine's record).
|
||||
#[cfg(feature = "sqlite")]
|
||||
Sqlite(SqliteCursor),
|
||||
/// The postgres body — the same materializing shape, the value
|
||||
/// loaded once from the engine's pooled row (the postgres
|
||||
/// engine's record).
|
||||
#[cfg(feature = "postgres")]
|
||||
Postgres(PostgresCursor),
|
||||
}
|
||||
|
||||
impl ReadCursor for KvCursor {
|
||||
@@ -362,6 +395,8 @@ impl ReadCursor for KvCursor {
|
||||
match self {
|
||||
#[cfg(feature = "mem")]
|
||||
Self::Mem(cursor) => cursor.len(),
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Postgres(cursor) => cursor.len(),
|
||||
#[cfg(feature = "sqlite")]
|
||||
Self::Sqlite(cursor) => cursor.len(),
|
||||
}
|
||||
@@ -371,6 +406,8 @@ impl ReadCursor for KvCursor {
|
||||
match self {
|
||||
#[cfg(feature = "mem")]
|
||||
Self::Mem(cursor) => cursor.read_at(offset, len).await,
|
||||
#[cfg(feature = "postgres")]
|
||||
Self::Postgres(cursor) => cursor.read_at(offset, len).await,
|
||||
#[cfg(feature = "sqlite")]
|
||||
Self::Sqlite(cursor) => cursor.read_at(offset, len).await,
|
||||
}
|
||||
@@ -461,6 +498,8 @@ impl KvEngine {
|
||||
match self {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(engine) => engine.put(key, value).await,
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => engine.put(key, value).await,
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => engine.put(key, value).await,
|
||||
}
|
||||
@@ -473,6 +512,11 @@ impl KvEngine {
|
||||
.get(key)
|
||||
.await?
|
||||
.map(|c| AnyCursor::Kv(KvCursor::Mem(c)))),
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => Ok(engine
|
||||
.get(key)
|
||||
.await?
|
||||
.map(|c| AnyCursor::Kv(KvCursor::Postgres(c)))),
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => Ok(engine
|
||||
.get(key)
|
||||
@@ -483,15 +527,19 @@ impl KvEngine {
|
||||
|
||||
/// The GC schema's engine-neutral store-core call (ADR-010 §2's
|
||||
/// companion seam; ADR-012 §5's applicability table), run at
|
||||
/// construction: a SQL-backed kv engine creates its window
|
||||
/// table(s) — sqlite's body creates `gc_window` only — and an
|
||||
/// in-process kv engine is structurally nothing (the
|
||||
/// construction: a SQL-backed kv engine creates the tables its
|
||||
/// applicability row names — sqlite's body creates `gc_window`
|
||||
/// only, postgres's creates `gc_pins` + `gc_window` (the fleet
|
||||
/// host's both-tables row) — and an in-process kv engine is
|
||||
/// structurally nothing (the
|
||||
/// local/mem row of the table: those engines execute the window
|
||||
/// protocol inline, ADR-005, and carry no GC schema).
|
||||
async fn ensure_gc_schema(&self) -> Result<(), Error> {
|
||||
match self {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(_) => Ok(()),
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => engine.ensure_gc_schema().await,
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => engine.ensure_gc_schema().await,
|
||||
}
|
||||
@@ -501,6 +549,8 @@ impl KvEngine {
|
||||
match self {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(engine) => engine.has(key).await,
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => engine.has(key).await,
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => engine.has(key).await,
|
||||
}
|
||||
@@ -510,6 +560,8 @@ impl KvEngine {
|
||||
match self {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(engine) => engine.size(key).await,
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => engine.size(key).await,
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => engine.size(key).await,
|
||||
}
|
||||
@@ -519,6 +571,8 @@ impl KvEngine {
|
||||
match self {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(engine) => engine.delete(key).await,
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => engine.delete(key).await,
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => engine.delete(key).await,
|
||||
}
|
||||
@@ -528,6 +582,8 @@ impl KvEngine {
|
||||
match self {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(engine) => engine.list().await,
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => engine.list().await,
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => engine.list().await,
|
||||
}
|
||||
@@ -728,6 +784,114 @@ impl Store {
|
||||
})
|
||||
}
|
||||
|
||||
/// **kv-only mode** (ADR-011 §4; ADR-012 §6.2) with the postgres kv
|
||||
/// engine — the fleet host's kv-only shape: no large tier;
|
||||
/// `threshold` is the put-time rejection point (known-length puts
|
||||
/// over it reject immediately, unknown-length puts at mid-stream
|
||||
/// threshold overflow — the put seam's behavior, identical to the
|
||||
/// mem and sqlite kv-only wiring; zero residue on either
|
||||
/// rejection). Routing in this mode is kv for every length.
|
||||
///
|
||||
/// `config` is the connection settings — the tokio-postgres
|
||||
/// [`tokio_postgres::Config`]-parseable DSN form; `opts` carries
|
||||
/// the engine-configuration knobs (schema name, pool size,
|
||||
/// `synchronous_commit`), riding the constructor parameters, never
|
||||
/// invented defaults. Opening the engine (config parse, pool
|
||||
/// build, connect, the bootstrap DDL) is fallible I/O; the
|
||||
/// constructor also runs the GC schema's engine-neutral call
|
||||
/// (`ensure_gc_schema` — the postgres row's body creates
|
||||
/// `gc_pins` + `gc_window`, the fleet host's both-tables row,
|
||||
/// ADR-012 §5), so a construction failure is `Err(Error::Io)` per
|
||||
/// the fallible-constructor pattern `dual_tier_mem_local` set.
|
||||
///
|
||||
/// `window_delay` (ADR-008 §2) is the durable window's cadence
|
||||
/// knob — [`DEFAULT_DELETE_WINDOW_DELAY`] is the shipped default,
|
||||
/// `Duration::ZERO` the legal degenerate tuning.
|
||||
///
|
||||
/// **Non-fleet (ADR-012 §3):** this constructor is implicitly
|
||||
/// `fleet: false` — the ADR-005 in-process protocol applies
|
||||
/// verbatim regardless of engine (pins are in-process guards, the
|
||||
/// sweep lease is the in-process lock, none of the
|
||||
/// `EngineFleetState` clauses run). The `fleet: true` declaration
|
||||
/// is wave 6 (the fleet-validity predicate plus the constructor
|
||||
/// declarations) and slots in without reshaping — no fleet
|
||||
/// parameter is carried here.
|
||||
#[cfg(feature = "postgres")]
|
||||
pub async fn kv_only_postgres(
|
||||
config: &str,
|
||||
opts: PgOpts,
|
||||
threshold: u64,
|
||||
window_delay: Duration,
|
||||
) -> Result<Store, Error> {
|
||||
let kv = KvEngine::Postgres(PostgresBackend::new(config, opts).await?);
|
||||
kv.ensure_gc_schema().await?;
|
||||
Ok(Store {
|
||||
kv,
|
||||
large: None,
|
||||
threshold,
|
||||
window_delay,
|
||||
mode: KvOnly,
|
||||
gc: GcHome::default(),
|
||||
})
|
||||
}
|
||||
|
||||
/// **dual-tier mode** (the default, ADR-011 §4) in the
|
||||
/// fleet-capable-kv production shape — `kv = postgres` +
|
||||
/// `large = local`: size-threshold dispatch at `threshold`
|
||||
/// (constructor-tunable;
|
||||
/// [`crate::dispatch::DEFAULT_DISPATCH_THRESHOLD`] is the shipped
|
||||
/// default), at-or-below on the kv tier, over-threshold staged
|
||||
/// into the large tier.
|
||||
///
|
||||
/// `config` is the connection settings — the tokio-postgres
|
||||
/// [`tokio_postgres::Config`]-parseable DSN form; `opts` carries
|
||||
/// the engine-configuration knobs (schema name, pool size,
|
||||
/// `synchronous_commit`), riding the constructor parameters, never
|
||||
/// invented defaults.
|
||||
///
|
||||
/// Fallible on two accounts (I/O): opening the engine (config
|
||||
/// parse, pool build, connect, the bootstrap DDL) — which also
|
||||
/// runs the GC schema's engine-neutral call (`ensure_gc_schema`;
|
||||
/// the postgres row's body creates `gc_pins` + `gc_window`,
|
||||
/// ADR-012 §5's one-window-host rule: the durable window is the
|
||||
/// kv engine's when the kv engine is SQL-backed, so the postgres
|
||||
/// host creates both GC tables per its applicability row) — and
|
||||
/// creating the local root. A construction failure is
|
||||
/// `Err(Error::Io)` per the fallible-constructor pattern
|
||||
/// `dual_tier_mem_local` set.
|
||||
///
|
||||
/// `window_delay` (ADR-008 §2) is the durable window's cadence
|
||||
/// knob — [`DEFAULT_DELETE_WINDOW_DELAY`] is the shipped default,
|
||||
/// `Duration::ZERO` the legal degenerate tuning.
|
||||
///
|
||||
/// **Non-fleet (ADR-012 §3):** this constructor is implicitly
|
||||
/// `fleet: false` — the ADR-005 in-process protocol applies
|
||||
/// verbatim regardless of engine (pins are in-process guards, the
|
||||
/// sweep lease is the in-process lock, none of the
|
||||
/// `EngineFleetState` clauses run). The `fleet: true` declaration
|
||||
/// is wave 6 (the fleet-validity predicate plus the constructor
|
||||
/// declarations) and slots in without reshaping — no fleet
|
||||
/// parameter is carried here.
|
||||
#[cfg(feature = "postgres")]
|
||||
pub async fn dual_tier_postgres_local(
|
||||
config: &str,
|
||||
opts: PgOpts,
|
||||
root: impl Into<PathBuf>,
|
||||
threshold: u64,
|
||||
window_delay: Duration,
|
||||
) -> Result<Store, Error> {
|
||||
let kv = KvEngine::Postgres(PostgresBackend::new(config, opts).await?);
|
||||
kv.ensure_gc_schema().await?;
|
||||
Ok(Store {
|
||||
kv,
|
||||
large: Some(LargeEngine::Local(LocalBackend::new(root).await?)),
|
||||
threshold,
|
||||
window_delay,
|
||||
mode: DualTier,
|
||||
gc: GcHome::default(),
|
||||
})
|
||||
}
|
||||
|
||||
/// The constructor mode (ADR-011 §4's vocabulary) — introspection
|
||||
/// for embedders and tests; the dispatch/put/read paths branch on
|
||||
/// it crate-internally.
|
||||
@@ -1374,8 +1538,9 @@ impl Store {
|
||||
/// **`sweep()`** — the explicit mark-and-sweep with the delete
|
||||
/// window (ADR-005 §Sweep), in two forms under one signature: the
|
||||
/// **durable form** runs where the kv engine is SQL-backed and hosts
|
||||
/// the window (ADR-012 §5's one-window-host rule — sqlite in today's
|
||||
/// set; the branch keys off the engine enum, never a flag), and the
|
||||
/// the window (ADR-012 §5's one-window-host rule — sqlite and
|
||||
/// postgres in today's set; the branch keys off the engine enum,
|
||||
/// never a flag), and the
|
||||
/// **inline form** (ADR-012 §5: `local`/`mem` stage candidates
|
||||
/// in-process and execute the window protocol inline in the same
|
||||
/// sweep) runs verbatim elsewhere. Sweeps are never errors:
|
||||
@@ -1494,26 +1659,37 @@ impl Store {
|
||||
match &self.kv {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(_) => self.sweep_window_inline(live, started).await,
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => self.sweep_window_durable(engine, live, started).await,
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => {
|
||||
// The durable form (ADR-008 §2; ADR-012 §6.7): the
|
||||
// overdue staged rows execute first — each re-arbitrated
|
||||
// at execution time — then this sweep's own mark and
|
||||
// stage at `visible_after = mark_end + window`.
|
||||
let (deleted, cancelled_by_arbitration) =
|
||||
self.execute_overdue_window(engine).await?;
|
||||
let candidates_staged = self.mark_and_stage_window(engine, live).await?;
|
||||
Ok(Self::sweep_report(
|
||||
candidates_staged,
|
||||
deleted,
|
||||
cancelled_by_arbitration,
|
||||
None,
|
||||
started,
|
||||
))
|
||||
}
|
||||
KvEngine::Sqlite(engine) => self.sweep_window_durable(engine, live, started).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// The durable form's body, shared by every SQL-backed kv-engine
|
||||
/// arm of the form branch (the branch keys off the kv-engine
|
||||
/// enum, ADR-012 §5's one-window-host rule — the host this runs
|
||||
/// on is the kv engine the constructor built). The overdue staged
|
||||
/// rows execute first — each re-arbitrated at execution time —
|
||||
/// then this sweep's own mark and stage at
|
||||
/// `visible_after = mark_end + window`.
|
||||
async fn sweep_window_durable<W: EngineWindowState>(
|
||||
&self,
|
||||
host: &W,
|
||||
live: BTreeSet<Key>,
|
||||
started: SystemTime,
|
||||
) -> Result<SweepReport, Error> {
|
||||
let (deleted, cancelled_by_arbitration) = self.execute_overdue_window(host).await?;
|
||||
let candidates_staged = self.mark_and_stage_window(host, live).await?;
|
||||
Ok(Self::sweep_report(
|
||||
candidates_staged,
|
||||
deleted,
|
||||
cancelled_by_arbitration,
|
||||
None,
|
||||
started,
|
||||
))
|
||||
}
|
||||
|
||||
/// The inline form (ADR-012 §5): the mark → arbitrate-at-delete-
|
||||
/// time → batched commit body, unchanged — the non-SQL engines
|
||||
/// stage candidates in-process and execute the window protocol
|
||||
@@ -1861,6 +2037,8 @@ mod tests {
|
||||
match store.kv() {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(engine) => engine.put(key, value).await.expect("kv plant"),
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => engine.put(key, value).await.expect("kv plant"),
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => engine.put(key, value).await.expect("kv plant"),
|
||||
}
|
||||
@@ -3905,6 +4083,8 @@ mod sqlite_construction_tests {
|
||||
KvEngine::Sqlite(engine) => engine,
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(_) => panic!("these tests construct sqlite stores"),
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(_) => panic!("these tests construct sqlite stores"),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4214,6 +4394,8 @@ mod sqlite_window_tests {
|
||||
match store.kv() {
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(_) => panic!("these tests construct sqlite stores"),
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(_) => panic!("these tests construct sqlite stores"),
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(engine) => engine,
|
||||
}
|
||||
@@ -4643,3 +4825,475 @@ mod sqlite_window_tests {
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// The wave-4 constructor wiring's tests (feature `postgres`): the
|
||||
/// postgres modes construct and introspect correctly, the GC schema
|
||||
/// ran at construction (both GC tables in the fresh schema — the
|
||||
/// fleet host's both-tables row), the window-delay default is the
|
||||
/// pinned constant and is constructor-tunable, the kv-only rejection
|
||||
/// wiring is the mode's put seam unchanged, the fleet-capable-kv
|
||||
/// shape (kv = postgres + large = local) round-trips end-to-end
|
||||
/// through both tiers, and the sweep's form branch runs the durable
|
||||
/// form on the postgres host. Env-driven — the engine tests'
|
||||
/// convention — skipping cleanly server-less.
|
||||
#[cfg(all(test, feature = "postgres"))]
|
||||
mod postgres_construction_tests {
|
||||
use super::*;
|
||||
use std::fs;
|
||||
use std::io::Cursor;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
/// A fresh temp root per test (mirrors the mem construction
|
||||
/// tests' helper); Drop removes the tree idempotently.
|
||||
struct TempRoot(PathBuf);
|
||||
|
||||
fn temp_root(tag: &'static str) -> TempRoot {
|
||||
static SEQ: AtomicU64 = AtomicU64::new(0);
|
||||
let seq = SEQ.fetch_add(1, Ordering::Relaxed);
|
||||
let nanos = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_nanos())
|
||||
.unwrap_or(0);
|
||||
let dir = std::env::temp_dir().join(format!(
|
||||
"alkblobs-store-pg-{tag}-{}-{:x}-{seq}",
|
||||
std::process::id(),
|
||||
nanos
|
||||
));
|
||||
fs::create_dir_all(&dir).expect("temp root created");
|
||||
TempRoot(dir)
|
||||
}
|
||||
|
||||
impl TempRoot {
|
||||
fn path(&self) -> &Path {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TempRoot {
|
||||
fn drop(&mut self) {
|
||||
let _ = fs::remove_dir_all(&self.0);
|
||||
}
|
||||
}
|
||||
|
||||
/// A fresh engine-owned schema per test — the postgres suite
|
||||
/// factory's fresh-schema-per-open convention — so the GC-schema
|
||||
/// assertions observe genuine creation.
|
||||
static SCHEMA_SEQ: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
fn fresh_schema(tag: &str) -> String {
|
||||
format!(
|
||||
"abt_sctor_{}_{}_{}",
|
||||
tag,
|
||||
std::process::id(),
|
||||
SCHEMA_SEQ.fetch_add(1, Ordering::Relaxed)
|
||||
)
|
||||
}
|
||||
|
||||
fn env_nonempty(name: &str) -> Option<String> {
|
||||
std::env::var(name).ok().filter(|v| !v.is_empty())
|
||||
}
|
||||
|
||||
/// The harness connection settings ride the environment
|
||||
/// (`ALKBLOBS_PG_HOST` required; port/user/password/db default to
|
||||
/// the harness server's values). Absent → the caller skips.
|
||||
fn harness_dsn() -> Option<String> {
|
||||
let host = env_nonempty("ALKBLOBS_PG_HOST")?;
|
||||
Some(format!(
|
||||
"host={host} port={} user={} password={} dbname={}",
|
||||
env_nonempty("ALKBLOBS_PG_PORT").unwrap_or_else(|| "15432".to_owned()),
|
||||
env_nonempty("ALKBLOBS_PG_USER").unwrap_or_else(|| "postgres".to_owned()),
|
||||
env_nonempty("ALKBLOBS_PG_PASSWORD").unwrap_or_else(|| "poc".to_owned()),
|
||||
env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()),
|
||||
))
|
||||
}
|
||||
|
||||
/// The skip note — printed exactly once per skipped test.
|
||||
fn skip_note() {
|
||||
eprintln!(
|
||||
"skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
|
||||
);
|
||||
}
|
||||
|
||||
/// A fresh-schema opts for the harness server, or `None` (skip
|
||||
/// note printed) server-less.
|
||||
fn fresh_opts(tag: &str) -> Option<PgOpts> {
|
||||
harness_dsn()?;
|
||||
Some(PgOpts {
|
||||
schema: fresh_schema(tag),
|
||||
..Default::default()
|
||||
})
|
||||
}
|
||||
|
||||
fn key_for(content: &[u8]) -> Key {
|
||||
Key::derive(content, content.len() as u64)
|
||||
}
|
||||
|
||||
/// The postgres kv engine behind the constructed store — the
|
||||
/// schema observations' instrument.
|
||||
fn postgres_engine(store: &Store) -> &PostgresBackend {
|
||||
match store.kv() {
|
||||
#[cfg(feature = "postgres")]
|
||||
KvEngine::Postgres(engine) => engine,
|
||||
#[cfg(feature = "mem")]
|
||||
KvEngine::Mem(_) => panic!("these tests construct postgres stores"),
|
||||
#[cfg(feature = "sqlite")]
|
||||
KvEngine::Sqlite(_) => panic!("these tests construct postgres stores"),
|
||||
}
|
||||
}
|
||||
|
||||
/// The engine-neutral GC-schema call ran at construction on the
|
||||
/// fleet host's applicability row: the pool table exists (the
|
||||
/// engine's own bootstrap) and *both* GC tables exist in the
|
||||
/// fresh schema (postgres's both-tables row, ADR-012 §5).
|
||||
async fn assert_gc_schema_ran(store: &Store) {
|
||||
let engine = postgres_engine(store);
|
||||
assert!(
|
||||
engine.table_exists("blobs").await.unwrap(),
|
||||
"the pool table exists at construction"
|
||||
);
|
||||
assert!(
|
||||
engine.table_exists("gc_window").await.unwrap(),
|
||||
"ensure_gc_schema ran at construction: the window table exists in the fresh schema"
|
||||
);
|
||||
assert!(
|
||||
engine.table_exists("gc_pins").await.unwrap(),
|
||||
"ensure_gc_schema ran at construction: the pins table exists (the fleet host's both-tables row)"
|
||||
);
|
||||
}
|
||||
|
||||
/// The dual-tier construction: both tiers present, mode/threshold/
|
||||
/// window-delay introspection exact, boundary routing
|
||||
/// deterministic, and the GC schema ran (both GC tables in the
|
||||
/// fresh schema).
|
||||
#[tokio::test]
|
||||
async fn dual_tier_postgres_local_constructs_and_routes() {
|
||||
let Some(opts) = fresh_opts("dual-tier") else {
|
||||
skip_note();
|
||||
return;
|
||||
};
|
||||
let root = temp_root("dual-tier");
|
||||
let dsn = harness_dsn().expect("harness dsn re-read");
|
||||
let store = Store::dual_tier_postgres_local(
|
||||
&dsn,
|
||||
opts,
|
||||
root.path(),
|
||||
1024,
|
||||
DEFAULT_DELETE_WINDOW_DELAY,
|
||||
)
|
||||
.await
|
||||
.expect("dual-tier postgres+local constructs");
|
||||
assert_eq!(store.mode(), DualTier);
|
||||
assert_eq!(store.threshold(), Some(1024));
|
||||
assert_eq!(store.window_delay(), DEFAULT_DELETE_WINDOW_DELAY);
|
||||
assert!(store.large().is_some(), "both tiers present");
|
||||
|
||||
for (len, tier) in [
|
||||
(0u64, Tier::Kv),
|
||||
(1023, Tier::Kv),
|
||||
(1024, Tier::Kv),
|
||||
(1025, Tier::Large),
|
||||
(u64::MAX, Tier::Large),
|
||||
] {
|
||||
assert_eq!(
|
||||
store.tier_for(len),
|
||||
tier,
|
||||
"length {len} routes deterministically"
|
||||
);
|
||||
}
|
||||
|
||||
assert_gc_schema_ran(&store).await;
|
||||
}
|
||||
|
||||
/// The kv-only construction: no large tier, the threshold is the
|
||||
/// rejection point's carrier, and the rejection wiring is the put
|
||||
/// seam's unchanged — known-length over-threshold rejects
|
||||
/// immediately (zero residue), unknown-length at mid-stream
|
||||
/// overflow, and exactly-at-threshold puts land.
|
||||
#[tokio::test]
|
||||
async fn kv_only_postgres_constructs_and_rejects_over_threshold_puts() {
|
||||
let Some(opts) = fresh_opts("kv-only") else {
|
||||
skip_note();
|
||||
return;
|
||||
};
|
||||
let dsn = harness_dsn().expect("harness dsn re-read");
|
||||
let store = Store::kv_only_postgres(&dsn, opts, 64, DEFAULT_DELETE_WINDOW_DELAY)
|
||||
.await
|
||||
.expect("kv-only postgres constructs");
|
||||
assert_eq!(store.mode(), KvOnly);
|
||||
assert_eq!(store.threshold(), Some(64));
|
||||
assert_eq!(store.window_delay(), DEFAULT_DELETE_WINDOW_DELAY);
|
||||
assert!(store.large().is_none(), "kv-only has no large tier");
|
||||
assert_gc_schema_ran(&store).await;
|
||||
|
||||
for len in [0u64, 63, 64, 65, u64::MAX] {
|
||||
assert_eq!(
|
||||
store.tier_for(len),
|
||||
Tier::Kv,
|
||||
"kv-only routes every length kv"
|
||||
);
|
||||
}
|
||||
|
||||
let err = store
|
||||
.put(Some(65), Cursor::new(&b"never read"[..]))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
err.to_string().contains("kv-only mode rejects"),
|
||||
"the immediate rejection is the mode's: {err}"
|
||||
);
|
||||
assert_eq!(
|
||||
store.list().await.unwrap(),
|
||||
Vec::<Key>::new(),
|
||||
"zero residue"
|
||||
);
|
||||
|
||||
let streamed = [3u8; 100];
|
||||
let err = store
|
||||
.put(None, Cursor::new(&streamed[..]))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("kv-only mode rejects an over-threshold put"),
|
||||
"the overflow rejection is the mode's, not the stream's: {err}"
|
||||
);
|
||||
assert_eq!(
|
||||
store.list().await.unwrap(),
|
||||
Vec::<Key>::new(),
|
||||
"zero residue"
|
||||
);
|
||||
|
||||
let at_threshold = vec![9u8; 64];
|
||||
let (key, pin) = store
|
||||
.put(Some(64), Cursor::new(&at_threshold[..]))
|
||||
.await
|
||||
.expect("exactly-at-threshold known put lands");
|
||||
assert_eq!(key, key_for(&at_threshold));
|
||||
assert!(store.has(&key.as_bytes()).await.unwrap());
|
||||
drop(pin);
|
||||
}
|
||||
|
||||
/// The window-delay parameter: the shipped default is the pinned
|
||||
/// 5-minute constant, the constructor's value passes through the
|
||||
/// store's introspection, and `Duration::ZERO` is a legal
|
||||
/// degenerate tuning (execute-on-next-sweep).
|
||||
#[tokio::test]
|
||||
async fn window_delay_default_is_pinned_and_constructor_tunable() {
|
||||
assert_eq!(
|
||||
DEFAULT_DELETE_WINDOW_DELAY,
|
||||
Duration::from_secs(5 * 60),
|
||||
"the shipped default is 5 minutes (ADR-008 §2)"
|
||||
);
|
||||
|
||||
let Some(opts) = fresh_opts("delay-default") else {
|
||||
skip_note();
|
||||
return;
|
||||
};
|
||||
let dsn = harness_dsn().expect("harness dsn re-read");
|
||||
let root = temp_root("delay-default");
|
||||
let store = Store::dual_tier_postgres_local(
|
||||
&dsn,
|
||||
opts,
|
||||
root.path(),
|
||||
64,
|
||||
DEFAULT_DELETE_WINDOW_DELAY,
|
||||
)
|
||||
.await
|
||||
.expect("constructs at the shipped default");
|
||||
assert_eq!(store.window_delay(), Duration::from_secs(300));
|
||||
|
||||
let Some(opts) = fresh_opts("delay-tuned") else {
|
||||
return;
|
||||
};
|
||||
let store = Store::kv_only_postgres(&dsn, opts, 64, Duration::from_secs(90))
|
||||
.await
|
||||
.expect("constructs at a custom tuning");
|
||||
assert_eq!(store.window_delay(), Duration::from_secs(90));
|
||||
|
||||
let Some(opts) = fresh_opts("delay-zero") else {
|
||||
return;
|
||||
};
|
||||
let root = temp_root("delay-zero");
|
||||
let store = Store::dual_tier_postgres_local(&dsn, opts, root.path(), 64, Duration::ZERO)
|
||||
.await
|
||||
.expect("the degenerate zero tuning is legal");
|
||||
assert_eq!(store.window_delay(), Duration::ZERO);
|
||||
|
||||
let Some(opts) = fresh_opts("delay-long") else {
|
||||
return;
|
||||
};
|
||||
Store::kv_only_postgres(&dsn, opts, 64, Duration::from_secs(6000))
|
||||
.await
|
||||
.expect("a long delay is a legal tuning too");
|
||||
}
|
||||
|
||||
/// The fleet-capable-kv shape round-trips end-to-end: a
|
||||
/// sub-threshold put lands on the kv tier (postgres), an
|
||||
/// over-threshold put stages into the large tier (local), both
|
||||
/// get back byte-identical, and the pool list is the sorted union
|
||||
/// of both tiers.
|
||||
#[tokio::test]
|
||||
async fn dual_tier_postgres_shape_round_trips_end_to_end_through_both_tiers() {
|
||||
let Some(opts) = fresh_opts("round-trip") else {
|
||||
skip_note();
|
||||
return;
|
||||
};
|
||||
let dsn = harness_dsn().expect("harness dsn re-read");
|
||||
let root = temp_root("round-trip");
|
||||
let store = Store::dual_tier_postgres_local(&dsn, opts, root.path(), 64, Duration::ZERO)
|
||||
.await
|
||||
.expect("dual-tier constructs");
|
||||
|
||||
let small = b"a sub-threshold blob";
|
||||
let (key, pin) = store
|
||||
.put(Some(small.len() as u64), Cursor::new(&small[..]))
|
||||
.await
|
||||
.expect("kv put publishes");
|
||||
assert_eq!(key, key_for(small));
|
||||
assert!(
|
||||
store.kv().has(&key.as_bytes()).await.unwrap(),
|
||||
"landed on the kv tier (postgres)"
|
||||
);
|
||||
assert!(
|
||||
!store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
|
||||
"no large-tier copy on the kv route"
|
||||
);
|
||||
let (len, blob) = store
|
||||
.get(&key.as_bytes())
|
||||
.await
|
||||
.expect("get succeeds")
|
||||
.expect("kv entry present");
|
||||
assert_eq!(len, small.len() as u64, "whole-blob length exact");
|
||||
assert_eq!(
|
||||
blob.read_at(0, len).await.expect("full window"),
|
||||
small,
|
||||
"kv-tier round-trip byte-identical"
|
||||
);
|
||||
drop(pin);
|
||||
|
||||
let large = vec![7u8; 128];
|
||||
let (key, pin) = store
|
||||
.put(Some(large.len() as u64), Cursor::new(&large[..]))
|
||||
.await
|
||||
.expect("staged put publishes");
|
||||
assert_eq!(key, key_for(&large));
|
||||
assert!(
|
||||
store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
|
||||
"landed on the large tier (local)"
|
||||
);
|
||||
assert!(
|
||||
!store.kv().has(&key.as_bytes()).await.unwrap(),
|
||||
"no kv-tier copy on the large route"
|
||||
);
|
||||
let (len, blob) = store
|
||||
.get(&key.as_bytes())
|
||||
.await
|
||||
.expect("get succeeds")
|
||||
.expect("large entry present");
|
||||
assert_eq!(len, large.len() as u64);
|
||||
assert_eq!(
|
||||
blob.read_at(0, len).await.expect("full window"),
|
||||
large,
|
||||
"large-tier round-trip byte-identical"
|
||||
);
|
||||
drop(pin);
|
||||
|
||||
let mut expected: Vec<Key> = [key_for(small), key_for(&large)].to_vec();
|
||||
expected.sort();
|
||||
assert_eq!(
|
||||
store.list().await.expect("list succeeds"),
|
||||
expected,
|
||||
"the pool list is the sorted union of both tiers"
|
||||
);
|
||||
}
|
||||
|
||||
/// A set-backed liveness source (the construction tests' shared
|
||||
/// sweep instrument).
|
||||
#[derive(Default)]
|
||||
struct SetSource {
|
||||
live: parking_lot::Mutex<BTreeSet<Key>>,
|
||||
}
|
||||
|
||||
impl LivenessSource for SetSource {
|
||||
fn live_keys(&self) -> Vec<Key> {
|
||||
self.live.lock().iter().cloned().collect()
|
||||
}
|
||||
|
||||
fn contains(&self, key: &Key) -> bool {
|
||||
self.live.lock().contains(key)
|
||||
}
|
||||
}
|
||||
|
||||
/// The sweep's form branch runs the **durable form** on the
|
||||
/// postgres host (the branch keyed off the kv-engine enum,
|
||||
/// ADR-012 §5's one-window-host rule): a zero-delay sweep stages
|
||||
/// this sweep's dead candidates and executes none of them (the
|
||||
/// inline form would delete in the same sweep), and the *next*
|
||||
/// sweep executes the overdue set through time-of-delete
|
||||
/// re-arbitration.
|
||||
#[tokio::test]
|
||||
async fn sweep_runs_the_durable_form_on_the_postgres_host() {
|
||||
let source = Arc::new(SetSource::default());
|
||||
|
||||
let Some(opts) = fresh_opts("durable-sweep") else {
|
||||
drop(source);
|
||||
skip_note();
|
||||
return;
|
||||
};
|
||||
let dsn = harness_dsn().expect("harness dsn re-read");
|
||||
let root = temp_root("durable-sweep");
|
||||
let store = Store::dual_tier_postgres_local(&dsn, opts, root.path(), 64, Duration::ZERO)
|
||||
.await
|
||||
.expect("dual-tier constructs");
|
||||
store.register_liveness_source(source.clone());
|
||||
|
||||
let dead = key_for(b"a dead candidate under the durable form");
|
||||
store
|
||||
.put(
|
||||
Some(dead_content().len() as u64),
|
||||
Cursor::new(&dead_content()[..]),
|
||||
)
|
||||
.await
|
||||
.expect("the candidate's put publishes");
|
||||
assert_eq!(store.list().await.unwrap(), vec![dead.clone()]);
|
||||
|
||||
let first = store.sweep().await.expect("sweeps are never errors");
|
||||
assert_eq!(first.aborted, None);
|
||||
assert_eq!(
|
||||
first.candidates_staged, 1,
|
||||
"the dead candidate staged into the durable window"
|
||||
);
|
||||
assert_eq!(
|
||||
first.deleted, 0,
|
||||
"the durable form executes nothing in its own staging sweep"
|
||||
);
|
||||
assert!(
|
||||
store.has(&dead.as_bytes()).await.unwrap(),
|
||||
"the staged candidate still waits out its window"
|
||||
);
|
||||
|
||||
let second = store.sweep().await.expect("sweeps are never errors");
|
||||
assert_eq!(second.aborted, None);
|
||||
assert_eq!(
|
||||
second.deleted, 1,
|
||||
"the next sweep executed the overdue candidate"
|
||||
);
|
||||
assert_eq!(
|
||||
second.candidates_staged, 0,
|
||||
"the pool holds nothing dead after the execution"
|
||||
);
|
||||
assert!(
|
||||
!store.has(&dead.as_bytes()).await.unwrap(),
|
||||
"the expired candidate is gone"
|
||||
);
|
||||
}
|
||||
|
||||
/// The content the durable-sweep test's candidate carries — the
|
||||
/// key derives over it, the entry is genuinely dead (nothing
|
||||
/// liveness-protects it; the registered source set stays empty).
|
||||
fn dead_content() -> Vec<u8> {
|
||||
b"a dead candidate under the durable form".to_vec()
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
---
|
||||
id: store-postgres-construction
|
||||
name: postgres constructor wiring — the fleet-capable store shapes
|
||||
status: pending
|
||||
status: completed
|
||||
depends_on: [engine-postgres, engine-postgres-fleet]
|
||||
scope: narrow
|
||||
risk: low
|
||||
@@ -79,24 +79,24 @@ a named constructor is slotting in"):
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- [ ] `KvEngine::Postgres` slots in; all six kv ops dispatch; the kv
|
||||
- [x] `KvEngine::Postgres` slots in; all six kv ops dispatch; the kv
|
||||
cursor body slots into `KvCursor` (the design-point note updated
|
||||
to the landed state)
|
||||
- [ ] `dual_tier_postgres_local` + `kv_only_postgres` constructors
|
||||
- [x] `dual_tier_postgres_local` + `kv_only_postgres` constructors
|
||||
exist behind feature `postgres`, take connection settings +
|
||||
engine opts + threshold + window delay, fallible where I/O is
|
||||
fallible
|
||||
- [ ] `ensure_gc_schema` runs at construction (both GC tables in the
|
||||
- [x] `ensure_gc_schema` runs at construction (both GC tables in the
|
||||
fresh schema); the window-delay shipped default is 5 min
|
||||
(pinned constant) and constructor-tunable; `0` legal
|
||||
- [ ] The non-fleet posture recorded (in-process protocol verbatim;
|
||||
- [x] The non-fleet posture recorded (in-process protocol verbatim;
|
||||
the `fleet: true` declaration is wave 6, slotting in without
|
||||
reshaping)
|
||||
- [ ] The one-window-host posture verified to cover postgres; the form
|
||||
- [x] The one-window-host posture verified to cover postgres; the form
|
||||
branch keys off the kv-engine enum
|
||||
- [ ] The pg-pure build (`--no-default-features --features postgres`)
|
||||
- [x] The pg-pure build (`--no-default-features --features postgres`)
|
||||
compiles clean
|
||||
- [ ] `cargo test` (default) green; `cargo test --features postgres`
|
||||
- [x] `cargo test` (default) green; `cargo test --features postgres`
|
||||
green server-less (skipping); `cargo clippy --all-targets -- -D
|
||||
warnings` and `cargo fmt --check` green (default +
|
||||
`--features postgres`)
|
||||
@@ -122,8 +122,94 @@ a named constructor is slotting in"):
|
||||
|
||||
## Notes
|
||||
|
||||
> To be filled by implementation agent
|
||||
> Implemented in `src/store.rs` + `src/postgres.rs` + `src/lib.rs`
|
||||
> (wave 4, following `store-sqlite-construction` — the structural
|
||||
> twin — point for point). Recorded design points, as resolved:
|
||||
>
|
||||
> - **Opts on the public surface** — `PgOpts` went `pub` (and is
|
||||
> re-exported from `lib.rs` behind the feature gate): the
|
||||
> constructors' engine-opts parameter is on the public constructor
|
||||
> signature, so the type must be nameable there (the alkstore
|
||||
> `open(config, PgOpts)` pin adapted to the named-constructor shape;
|
||||
> the connection settings remain the separate DSN-string argument).
|
||||
> - **`KvCursor` slotting** — postgres's materializing body slotted
|
||||
> into the recorded sub-enum exactly as predicted: one variant, two
|
||||
> `ReadCursor` impl arms, no reshaping of `AnyCursor`/`Blob`. The
|
||||
> store.rs design-point note reads landed.
|
||||
> - **`ensure_gc_schema`** rides the engine-neutral `KvEngine` call —
|
||||
> the postgres arm forwards to the `EngineWindowState` clause (the
|
||||
> fleet host's both-tables body: `gc_pins` + `gc_window` in the
|
||||
> fresh schema), and both constructors dispatch through it (the
|
||||
> store core calls the kv engine, never a typed postgres path).
|
||||
> Construction failure is `Err(Error::Io)` per the fallible-
|
||||
> constructor pattern (`?` throughout).
|
||||
> - **The sweep's form branch** — the durable-form body was extracted
|
||||
> into `sweep_window_durable<W: EngineWindowState>` (behavior
|
||||
> unchanged for sqlite, whose arm now delegates); both SQL-backed
|
||||
> arms call it, so the form branch still keys off the kv-engine
|
||||
> enum, never a flag, and wave 5's pg-lo dormancy slots in without
|
||||
> reshaping. The one-window-host posture's wording now names both
|
||||
> SQL-backed hosts (module doc + the sweep doc).
|
||||
> - **Non-fleet posture** — both constructors' docs carry the
|
||||
> ADR-012 §3 recording verbatim: implicitly `fleet: false`, the
|
||||
> ADR-005 in-process protocol applies regardless of engine (in-
|
||||
> process pins, in-process sweep lease, no `EngineFleetState` clause
|
||||
> runs), and the `fleet: true` declaration is wave 6, slotting in
|
||||
> without reshaping — no fleet parameter carried.
|
||||
> - **Signature shape** — `dual_tier_postgres_local(config, opts,
|
||||
> root, threshold, window_delay)` / `kv_only_postgres(config, opts,
|
||||
> threshold, window_delay)`: connection settings first (the
|
||||
> tokio-postgres `Config`-parseable DSN string), engine opts second,
|
||||
> then the sqlite twins' remaining parameters (root/threshold/
|
||||
> window delay). The DSN is `&str` — `PostgresBackend::new` parses;
|
||||
> no invented defaults anywhere.
|
||||
> - **Clippy notes** — `KvEngine` needed a scoped
|
||||
> `#[allow(clippy::large_enum_variant)]` under the postgres
|
||||
> configs (the payloads are structurally dissimilar — one `Arc` vs
|
||||
> the pooled-connection posture inline — and all dispatch rides the
|
||||
> enum by reference, so the size difference costs nothing; recorded
|
||||
> on the enum doc). The store-construction `table_exists` probe
|
||||
> (pg_tables, schema-scoped) mirrors the sqlite engine's
|
||||
> `#[cfg(test)]` instrument.
|
||||
> - **Tests (5 new, env-driven, skipping server-less)** — dual-tier
|
||||
> construction + routing + both-GC-tables schema check; kv-only
|
||||
> construction + rejection wiring (immediate, overflow, zero
|
||||
> residue, at-threshold lands); the window-delay default/tunings/
|
||||
> zero; the dual-tier round-trip through both tiers with tier-
|
||||
> placement assertions; and a durable-form sweep smoke (zero-delay:
|
||||
> first sweep stages and executes nothing — the inline form would
|
||||
> delete in-sweep, so this is the form-branch-is-postgres proof —
|
||||
> the second sweep executes the overdue set re-arbitrated at
|
||||
> execution time). The test-side `KvEngine` matches in the existing
|
||||
> test modules gained `Postgres` arms for `--all-features`
|
||||
> exhaustiveness.
|
||||
> - Stale scope notes updated to the landed state: the engine-
|
||||
> representation doc, the `KvCursor` design point, the fleet-
|
||||
> declaration paragraph, the one-window-host paragraph, the
|
||||
> postgres module doc's forward reference — and `gc.rs`'s
|
||||
> companion-seam note (the fleet family now "waits for wave 6", its
|
||||
> clause bodies landed with wave 4). The batch-delete helper's
|
||||
> scope note stays (its consumer is still wave 6).
|
||||
|
||||
## Summary
|
||||
|
||||
> To be filled on completion
|
||||
> Completed 2026-10-12 (wave 4). Postgres slotted into the facade:
|
||||
> `KvEngine::Postgres` with all six kv-op dispatch arms, the
|
||||
> `KvCursor::Postgres` materializing body, the `ensure_gc_schema`
|
||||
> engine-neutral call creating `gc_pins` + `gc_window` in the fresh
|
||||
> schema at construction, and the two named feature-gated
|
||||
> constructors (`dual_tier_postgres_local`, `kv_only_postgres`) taking
|
||||
> the DSN connection settings + `PgOpts` (now public and re-exported) +
|
||||
> threshold + the constructor-tunable delete-window delay
|
||||
> (`DEFAULT_DELETE_WINDOW_DELAY` pinned at 5 min, `0` legal) — both
|
||||
> implicitly non-fleet per ADR-012 §3 with wave 6's declaration
|
||||
> recorded as a non-reshaping slotting. The sweep's durable form runs
|
||||
> on the postgres host through the kv-engine-enum-keyed branch
|
||||
> (`sweep_window_durable`, shared with sqlite's arm). Verification:
|
||||
> `cargo test` 69/19/3 ok (default), `cargo test --features postgres`
|
||||
> 102/19/3 ok live and green server-less, `cargo test
|
||||
> --no-default-features --features postgres` (the pg-pure build, ADR-
|
||||
> 013 §4's first constructible configuration) 75/2/3 ok server-less,
|
||||
> `cargo test --all-features` 157/54/4 ok, clippy `-D warnings` green
|
||||
> on default + postgres + pg-pure + all-features, `cargo fmt --check`
|
||||
> green, `cargo doc --no-deps` clean.
|
||||
Reference in new issue
Block a user