diff --git a/src/gc.rs b/src/gc.rs index 5663b69..2b740cb 100644 --- a/src/gc.rs +++ b/src/gc.rs @@ -70,9 +70,11 @@ //! SQL-backed, pg-lo when the kv engine is not (the one-window-host //! rule, ADR-012 §5); the sweep's durable form drives //! `window_stage`/`window_take` on that same host), while the fleet -//! family 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. +//! family waits for its store-core consumers — its postgres clause +//! bodies landed with wave 4, and the store core consumes the family +//! only under the `fleet: true` declaration (the declaration's fleet +//! constructors landed; the put/sweep drivers are the remaining +//! tasks). The note shrinks as those land. #![allow(dead_code)] diff --git a/src/lib.rs b/src/lib.rs index 2d899a0..3e2485f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,6 +18,9 @@ pub use hashing::{Algorithm, Digest}; pub use keys::Key; pub use store::{Store, StoreMode, DEFAULT_DELETE_WINDOW_DELAY}; +#[cfg(feature = "postgres")] +pub use store::{FleetSharedMedia, DEFAULT_PIN_TTL}; + #[cfg(feature = "mem")] pub mod mem; diff --git a/src/pg_lo.rs b/src/pg_lo.rs index b8d9dd0..3d54609 100644 --- a/src/pg_lo.rs +++ b/src/pg_lo.rs @@ -295,6 +295,24 @@ impl PgLoBackend { }) } + /// The pg instance's `system_identifier` (`pg_control_system()`, + /// PG14+) — read across one pooled connection. The fleet + /// constructor's same-instance probe's large-side instrument + /// (ADR-012 §4): the kv engine's identifier compared against this + /// — server-side instance identity, immune to DSN aliases. A probe + /// failure surfaces as an error (the probe errors closed — never a + /// silent pass). pg-lo never hosts fleet state; it only answers + /// the probe. + pub(crate) async fn system_identifier(&self) -> Result { + let client = self.pool.get().await.map_err(err)?; + let row = client + .query_one("SELECT system_identifier FROM pg_control_system()", &[]) + .await + .map_err(pg_err)?; + let id: i64 = row.try_get(0).map_err(pg_err)?; + Ok(id as u64) + } + /// The catalog table-existence probe — test-only: the bootstrap /// test's instrument, answering for the engine-owned schema the /// backend was constructed with. diff --git a/src/postgres.rs b/src/postgres.rs index 7ef55bd..2a491fe 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -111,7 +111,9 @@ //! [`EngineFleetState`] impl carries the fleet family: the joint //! entry+pin transaction, the pin clauses, delete-time SQL arbitration //! with the engine-side liveness-table registration, and the sweep -//! lease. The store core drives these clauses in wave 6 (fleet mode) — +//! lease. The store core drives these clauses under the fleet +//! declaration (`fleet: true` — the `_fleet` constructors landed, with +//! the put/sweep drivers the fleet mechanism's remaining tasks) — //! under `fleet: false` none of the fleet clauses run, while //! `ensure_gc_schema` runs on every postgres store construction (the //! `store-postgres-construction` wiring, landed: the constructors are @@ -137,10 +139,11 @@ //! ownership stays with the store core from `begin_joint_tx` until //! `pin_commit` consumes it (ADR-010 §2's pinned ruling). //! - **`pin_renew`'s successor time** — the clause signature carries -//! no expiry, so the engine computes `expiry = now + DEFAULT_PIN_TTL` -//! (ADR-008's default 60 s; the shipped constant is a backend field, -//! constructor-tunable when wave 6's fleet constructor lands). The -//! store owns the TTL and the TTL/3 cadence — `pin_commit`'s pin +//! no expiry, so the engine computes `expiry = now + pin_ttl` (the +//! backend's pin-TTL field: the shipped default 60 s, constructor- +//! tunable through the fleet constructors' `new_with_pin_ttl` +//! instrument — the store's fleet `pin_ttl` parameter, ADR-008 §2). +//! The store owns the TTL and the TTL/3 cadence — `pin_commit`'s pin //! rows carry the store-resolved expiry; renewals take the //! successor-time shape recorded here. //! - **Timestamp encoding** — signed i64 nanoseconds from the Unix @@ -240,8 +243,9 @@ pub(crate) struct PostgresBackend { window_table: String, /// The pin-TTL the successor-time source renews against (the /// recorded design point: ADR-008's default 60 s, shipped as a - /// backend constant; constructor-tunability arrives with wave 6's - /// fleet constructor). + /// backend constant; the fleet constructors tune it through + /// [`PostgresBackend::new_with_pin_ttl`] — the store's fleet + /// `pin_ttl` parameter landed). pin_ttl: Duration, /// The sweep lease's held checkout: `Some` exactly while the lease /// is held — the same-connection acquire/release discipline (the @@ -277,6 +281,22 @@ impl PostgresBackend { /// cannot be built (`max_size = 0` fails at entry), or the DDL /// errors. pub(crate) async fn new(config: &str, opts: PgOpts) -> Result { + Self::new_with_pin_ttl(config, opts, DEFAULT_PIN_TTL).await + } + + /// Open the engine with a constructor-carried pin TTL — the fleet + /// constructors' instrument (wave 6's declaration landed): the + /// store's fleet `pin_ttl` (ADR-008 §2, default [`DEFAULT_PIN_TTL`]) + /// configures the backend so [`EngineFleetState::pin_renew`]'s + /// successor-time body reads the TTL it was constructed with. The + /// non-fleet constructors ride [`Self::new`] (the shipped default; + /// nothing renews under `fleet: false`, so the value is dormant + /// there). + pub(crate) async fn new_with_pin_ttl( + config: &str, + opts: PgOpts, + pin_ttl: Duration, + ) -> Result { let pool = build_pool(config, opts.max_size, opts.synchronous_commit)?; let schema = if opts.schema.is_empty() { DEFAULT_SCHEMA.to_owned() @@ -298,7 +318,7 @@ impl PostgresBackend { schema, pins_table, window_table, - pin_ttl: DEFAULT_PIN_TTL, + pin_ttl, lease: Mutex::new(None), liveness_sources: Arc::new(Mutex::new(Vec::new())), #[cfg(test)] @@ -308,6 +328,29 @@ impl PostgresBackend { }) } + /// The pg instance's `system_identifier` (`pg_control_system()`, + /// PG14+) — read across one pooled connection. The fleet + /// constructor's same-instance probe's instrument (ADR-012 §4): + /// the kv engine's and the pg-lo engine's identifiers compared — + /// server-side instance identity, immune to DSN aliases. A probe + /// failure surfaces as an error (the probe errors closed — never + /// a silent pass). + pub(crate) async fn system_identifier(&self) -> Result { + Self::system_identifier_of(&self.pool).await + } + + /// The probe's body over a pool — shared with the pg-lo engine's + /// twin. + async fn system_identifier_of(pool: &Pool) -> Result { + let client = pool.get().await.map_err(err)?; + let row = client + .query_one("SELECT system_identifier FROM pg_control_system()", &[]) + .await + .map_err(pg_err)?; + let id: i64 = row.try_get(0).map_err(pg_err)?; + Ok(id as u64) + } + /// Open a pool without schema creation — the virgin-store /// tolerance test's instrument (constructor-vs-lazy is a recorded /// design point; the read paths' no-schema-yet semantics are tested @@ -415,7 +458,10 @@ CREATE TABLE IF NOT EXISTS {table} ( } /// The pin-TTL constant's seconds ([`DEFAULT_PIN_TTL`]) — ADR-008's -/// default, constructor-tunable when wave 6's fleet constructor lands. +/// default; the store re-exports it as the fleet constructors' shipped +/// `pin_ttl` value ([`crate::store::DEFAULT_PIN_TTL`]), and the fleet +/// constructors tune the backend's `pin_ttl` through +/// [`PostgresBackend::new_with_pin_ttl`]. pub(crate) const DEFAULT_PIN_TTL_SECS: u64 = 60; /// The shipped pin-TTL ([`PostgresBackend::pin_renew`]'s successor-time @@ -804,9 +850,12 @@ impl EngineWindowState for PostgresBackend { /// The fleet family (ADR-012 §5's postgres row — all yes): the joint /// entry+pin transaction, the pin clauses, delete-time SQL arbitration, -/// and the sweep lease. Fleet activation is wave 6's constructor -/// declaration — under `fleet: false` the store core never calls these; -/// the clauses exist and are contract-tested per their applicability. +/// and the sweep lease. Fleet activation is the constructor +/// declaration (`fleet: true` — the `_fleet` constructors' explicit +/// declaration, ADR-012 §3) — under `fleet: false` the store core +/// never calls these; the clauses exist and are contract-tested per +/// their applicability, and the store core's put/sweep drivers +/// consume them as the fleet mechanism's tasks land. impl EngineFleetState for PostgresBackend { type JointTx = PgJointTx; @@ -889,11 +938,13 @@ impl EngineFleetState for PostgresBackend { } } - /// Renew the `(key, owner)` pin row: `expiry = now + DEFAULT_PIN_TTL` - /// (the recorded successor-time source) — and only against a live - /// row: the update's predicate requires the existing row un-expired, - /// so an expired or absent row is never resurrected (`false`, - /// nothing updated). + /// Renew the `(key, owner)` pin row: `expiry = now + pin_ttl` (the + /// recorded successor-time source — the TTL the backend was + /// constructed with: the shipped default, or the fleet + /// constructor's value through `new_with_pin_ttl`) — and only + /// against a live row: the update's predicate requires the + /// existing row un-expired, so an expired or absent row is never + /// resurrected (`false`, nothing updated). async fn pin_renew(&self, key: &[u8], owner: &[u8]) -> Result { let client = self.pool.get().await.map_err(err)?; let now = SystemTime::now(); @@ -1123,6 +1174,29 @@ mod tests { harness_with(tag, PgOpts::default()).await } + /// The TTL-tuned twin of [`harness`]: the backend opens through + /// `new_with_pin_ttl` (the fleet constructors' instrument), so the + /// renewal-body wiring tests the constructor-carried value. + async fn harness_with_ttl(tag: &str, pin_ttl: Duration) -> Option { + let Some(dsn) = harness_dsn() else { + skip_note(); + return None; + }; + let schema = fresh_schema(tag); + let opts = PgOpts { + schema: schema.clone(), + ..PgOpts::default() + }; + let backend = PostgresBackend::new_with_pin_ttl(&dsn, opts, pin_ttl) + .await + .expect("open"); + backend.ensure_gc_schema().await.expect("gc schema"); + Some(Harness { + backend, + _janitor: SchemaJanitor::new(&dsn, &schema), + }) + } + async fn harness_with(tag: &str, opts: PgOpts) -> Option { let Some(dsn) = harness_dsn() else { skip_note(); @@ -2310,6 +2384,45 @@ mod tests { ); } + /// The constructor-carried pin TTL reaches the renewal body: a + /// backend opened through `new_with_pin_ttl` (the fleet + /// constructors' instrument) renews a live row to + /// `now + pin_ttl` — the constructor's value, not the shipped + /// default (the wave-4 recorded design point, landed with the + /// fleet declaration). + #[tokio::test] + async fn pin_renew_renews_to_the_constructor_carried_ttl() { + let Some(backend) = harness_with_ttl("renew-ttl", Duration::from_secs(3600)).await else { + return; + }; + let key = b"\x01renew-ttl-key".to_vec(); + let owner = b"\x0ainstance-id".to_vec(); + + let live = SystemTime::now() + Duration::from_secs(30); + { + let client = backend.pool.get().await.unwrap(); + let sql = format!( + "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)", + backend.pins_table + ); + client + .execute(sql.as_str(), &[&key, &owner, &time_to_nanos(live)]) + .await + .unwrap(); + } + assert!( + backend.pin_renew(&key, &owner).await.unwrap(), + "live row renews" + ); + let after = pin_rows(&backend).await; + let expected = time_to_nanos(SystemTime::now()) + 3600i64 * 1_000_000_000; + let got = after[0].2; + assert!( + (got - expected).abs() < 3_000_000_000, + "renewed to the constructor-carried successor time now + 3600s: {got} vs {expected}" + ); + } + /// `pin_check`'s two-variant classification against un-expired pin /// rows: absent or only-expired rows → `Unprotected`; any live row /// → `Protected`. Protection composes by row, so any owner's live diff --git a/src/store.rs b/src/store.rs index 9f188f7..a8c7ac1 100644 --- a/src/store.rs +++ b/src/store.rs @@ -42,8 +42,10 @@ //! **default-build production shape** (feature `sqlite` is //! 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). The wave-5 shapes complete +//! engine is the fleet host, ADR-012 §4; the constructor stays the +//! non-fleet declaration — the fleet declaration is the `_fleet` +//! twin constructors, the module doc's fleet-declaration +//! paragraph). The wave-5 shapes complete //! the dual-tier constructor set: `kv = postgres` + `large = pg-lo` //! — the **consolidation posture** (ADR-009's "one engine to //! operate"; that constructor completes the fleet-valid engine @@ -201,27 +203,52 @@ //! group's pins drop together on [`Batch::commit`] or drop (the //! `gc-protocol` machinery drives). //! -//! **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 the fleet-valid engine set is now complete in -//! the constructors (pg-lo's consolidation constructor landed wave 5, -//! the module doc's recorded forward reference — there is no -//! fleet-valid engine left to land): mem is never fleet-valid, -//! `local` needs the shared-media declaration of ADR-008 §3, and -//! pg-lo requires the kv host's same-instance clause (ADR-012 §4) — -//! so wave 6 is the declaration's predicate and constructor -//! parameter slotting in, nothing more. The fleet-capable engines' -//! constructors — postgres's pair (wave 4) and the pg-lo -//! consolidation constructor (wave 5) — 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 fleet declaration (ADR-012 §3; landed — wave 6's +//! `store-fleet-construction`).** Fleet mode is a constructor +//! declaration, never inferred: every pre-existing 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 — behavior byte-unchanged), and the **`_fleet` +//! twin constructors** declare `fleet: true`: +//! `Store::kv_only_postgres_fleet` (the re-routing client posture — +//! large = none, pool large content rides the ops surface, wave 7's +//! wiring), `Store::dual_tier_postgres_local_fleet` (the shared-media +//! posture — the declaration requires the sealed +//! `FleetSharedMedia` assertion, ADR-008 §3), and +//! `Store::dual_tier_postgres_pglo_fleet` (the consolidation posture — +//! ADR-012 §4's same-instance clause runtime-probed). **The composite +//! fleet-validity predicate** (`fleet_validity_error` — the one- +//! sentence rule of backends-and-dispatch.md, both tiers) is enforced +//! at the constructor, keyed off the engine representation like every +//! other dispatch in this store: a configuration is fleet-valid iff +//! `kv = postgres` AND (`large = pg-lo` on the same pg instance as the +//! kv engine OR `large = local` on the declared shared media OR +//! `large = none`) — plus the explicit declaration. `fleet: true` is a +//! constructor error for every non-fleet-valid configuration; the +//! predicate's mem/sqlite clauses are enforced structurally (no +//! `_fleet` constructor is named for a mem or sqlite kv engine — the +//! recorded "not fleet-anything" postures) with the predicate's +//! rejection clauses unit-tested per clause at the predicate level, +//! and the pg-lo same-instance clause is the one runtime-checkable +//! clause: the instance's `system_identifier` (via +//! `pg_control_system()`, PG14+) compared across one connection per +//! engine — a mismatch **or a failed probe** is a constructor error, +//! never a silent pass. **The fleet constructor parameters, not API** +//! (ADR-008 §2): the pin TTL (default `DEFAULT_PIN_TTL`, 60 s — the +//! store owns the TTL and the TTL/3 renewal cadence; the engine's +//! `pin_renew` successor-time body reads the TTL it was constructed +//! with) and the large-tier declaration (the mode's name, the sealed +//! shared-media assertion, or the pg-lo shape). The store carries the +//! fleet state once declared — the flag (the put/sweep paths branch on +//! it, never on re-deriving from the engine set), the pin TTL, and the +//! store-instance id (the pin row's `owner` half, minted at +//! construction, 32 raw bytes, crate-internal): `FleetState`, never +//! public API (ADR-008 §3: fleet constructor parameters, not API). The +//! fleet declaration slots in without reshaping the fleet-capable +//! constructors (the wave-4/5 recorded non-event landed as recorded): +//! the pre-existing constructors' signatures and behavior are +//! unchanged and remain implicitly `fleet: false`. //! //! **The dispatch seam the ops tasks program through:** tier selection //! for a given length (`Store::tier_for` composing @@ -263,13 +290,20 @@ //! //! The `#![allow(dead_code)]` below marks the work still ahead, not //! the GC protocol (the window host's construction call and the -//! durable form are driven below on either host). The fleet clauses -//! hold the remaining scope until wave 6's fleet declaration. +//! durable form are driven below on either host). The fleet +//! declaration has landed (the constructors' predicate, parameters, +//! and fleet state above); the fleet mechanism's store-core consumers +//! hold the remaining scope — the joint-tx put driver and the fleet +//! sweep (`store-fleet-put`, `store-fleet-sweep`). #![allow(dead_code)] use std::collections::BTreeSet; +#[cfg(feature = "postgres")] +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +#[cfg(feature = "postgres")] +use std::time::UNIX_EPOCH; use std::time::{Duration, SystemTime}; use tokio::io::{AsyncRead, AsyncReadExt}; @@ -296,6 +330,9 @@ use crate::postgres::{PostgresBackend, PostgresCursor}; #[cfg(feature = "sqlite")] use crate::sqlite::{SqliteBackend, SqliteCursor}; +#[cfg(feature = "postgres")] +use sha2::{Digest as _, Sha256}; + use crate::Error; #[cfg(any(feature = "mem", feature = "postgres", feature = "sqlite"))] @@ -313,6 +350,195 @@ use self::StoreMode::{DualTier, KvOnly, MemOnly}; /// delay in effect and the constant is irrelevant to it. pub const DEFAULT_DELETE_WINDOW_DELAY: Duration = Duration::from_secs(5 * 60); +/// The shipped default **pin TTL** (ADR-008 §2): a fleet mode's pin +/// rows renew to `now + pin_ttl` on the guard's TTL/3 cadence — 60 +/// seconds, the fleet constructors' default tuning (`FleetState`'s +/// carried value; the engine's `pin_renew` successor-time body reads +/// the TTL its backend was constructed with). Constructor-tunable +/// per deployment, never API (ADR-008 §2: fleet constructor +/// parameters, not API). `Duration::ZERO` is a constructor error — +/// the TTL/3 renewal cadence degenerates at zero and a zero TTL +/// expires every pin at once. +#[cfg(feature = "postgres")] +pub const DEFAULT_PIN_TTL: Duration = Duration::from_secs(crate::postgres::DEFAULT_PIN_TTL_SECS); + +/// The operator's explicit **shared-media assertion** (ADR-008 §3's +/// enforceable-seam bullet): the declaration a `local` large tier +/// needs under `fleet: true` — the operator's assertion that the +/// constructor's `local` root is fleet-shared media. **Named design +/// point — the declaration's type shape:** a dedicated sealed type +/// with a private seal field, not a bare flag — the caller cannot +/// construct it accidentally and must name [`FleetSharedMedia::asserted`] +/// at the call site (the ADR's "explicit declaration" wording's +/// reading: a named type the call site cannot misread), while the +/// predicate-level alternative (`LocalUndeclared`) has no constructor +/// to reach it from. The assertion asserts — cross-node truth ("is +/// the declared media actually shared?") remains the deployment's +/// verified invariant, the detection symptom documented (ADR-008 §3's +/// seam; ADR-012 §3's misconfiguration reading): the constructor +/// cannot prove cross-node fact, only admit the declared posture. +#[cfg(feature = "postgres")] +pub struct FleetSharedMedia { + _seal: (), +} + +#[cfg(feature = "postgres")] +impl FleetSharedMedia { + /// Mint the shared-media assertion — the operator's explicit + /// declaration at the fleet constructor's call site (the type's + /// sealed shape makes this the only minting path). + pub fn asserted() -> Self { + Self { _seal: () } + } +} + +/// The store's **fleet state** — what the constructor's `fleet: true` +/// declaration leaves the store carrying (crate-internal, never +/// public API — ADR-008 §3: fleet constructor parameters, not API): +/// the declaration's flag (the put/sweep paths branch on the state's +/// presence, never on re-deriving from the engine set), the pin TTL +/// the constructor took, and the **store-instance id** (`owner`, +/// ADR-011's vocabulary — the pin row's identity half). +pub(crate) struct FleetState { + /// The pin TTL the fleet constructor took (default + /// [`DEFAULT_PIN_TTL`]) — the store owns the TTL and the TTL/3 + /// renewal cadence; the engine's `pin_renew` successor-time body + /// reads the TTL its backend was constructed with. + pin_ttl: Duration, + /// The store-instance id ([`mint_owner_id`]'s minting) — carried + /// as the `owner` bytes of every [`crate::gc::PinRow`] the store + /// mints. + owner: Vec, +} + +impl FleetState { + /// The pin TTL the fleet constructor took. + pub(crate) fn pin_ttl(&self) -> Duration { + self.pin_ttl + } + + /// The store-instance id — the `owner` bytes of the pin rows this + /// instance mints. + pub(crate) fn owner(&self) -> &[u8] { + &self.owner + } +} + +/// The composite fleet-validity predicate's **kv descriptor** — the +/// engine-set shape the predicate reasons over, derived off the +/// engine representation ([`FleetKvShape::of`]) like every other +/// dispatch in this store, never a flag. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FleetKvShape { + /// The fleet kv engine (ADR-011 §4's constructor table's only + /// fleet-valid kv row). + Postgres, + /// The durable single-node kv engine — never fleet-valid (a + /// sqlite fleet is a recorded non-option, ADR-010 §3). + Sqlite, + /// The contract-reference ephemeral engine — never fleet-valid + /// (ADR-011 §4). + Mem, +} + +/// The composite fleet-validity predicate's **large descriptor**. +#[derive(Debug, Clone, Copy)] +enum FleetLargeShape { + /// No large tier — the kv-only re-routing posture (ADR-008 §3): + /// pool large content rides the ops surface. + None, + /// The `local` engine with the operator's shared-media + /// declaration ([`FleetSharedMedia`]). + LocalSharedMedia, + /// The `local` engine undeclared — the predicate's rejected arm: + /// a fleet-mode constructor requires the explicit shared-media + /// assertion (ADR-008 §3). + LocalUndeclared, + /// The pg-lo engine — fleet-valid under ADR-012 §4's + /// same-instance clause (the constructor-level probe). + PgLo, +} + +/// **The composite fleet-validity predicate** — backends-and-dispatch.md's +/// one-sentence rule, both tiers, as code: a configuration is +/// fleet-valid iff **kv = postgres AND (large = pg-lo on the same pg +/// instance as the kv engine OR large = `local` on declared shared +/// media OR large = none)**. `None` = fleet-valid; `Some` = the +/// constructor error `fleet: true` carries (ADR-012 §3). The +/// enforcement map: the `_fleet` constructors are named only for the +/// shapes the predicate accepts — the structural clauses (kv ≠ +/// postgres, mem anywhere, `local` without the shared-media +/// declaration) have no fleet constructor to reach them through (the +/// recorded "not fleet-anything" postures carry no fleet parameter) +/// and are enforced by that constructor-set shape plus this +/// predicate's rejection arms (unit-tested per clause, the +/// predicate's error being the constructor error a would-be +/// construction carries); the pg-lo shape's runtime-checkable clause +/// — the same-instance probe (ADR-012 §4) — runs at its constructor +/// alongside this call. Derived off the engine representation; +/// crate-internal (fleet parameters, not API). +fn fleet_validity_error(kv: FleetKvShape, large: FleetLargeShape) -> Option { + match (kv, large) { + (FleetKvShape::Postgres, FleetLargeShape::None) => None, + (FleetKvShape::Postgres, FleetLargeShape::LocalSharedMedia) => None, + (FleetKvShape::Postgres, FleetLargeShape::PgLo) => None, + (FleetKvShape::Postgres, FleetLargeShape::LocalUndeclared) => Some(Error::Io( + "fleet: the local large tier rejects without the shared-media declaration \ + (ADR-008 \u{a7}3): a fleet-mode constructor requires the explicit \ + FleetSharedMedia assertion" + .to_owned(), + )), + (kv, _) => Some(Error::Io(format!( + "fleet: the kv engine {kv:?} rejects the fleet declaration (ADR-012 \u{a7}3): \ + kv = postgres is the composite fleet-validity rule's kv clause — \ + {kv:?} is never fleet-valid" + ))), + } +} + +impl FleetKvShape { + /// The predicate kv descriptor derived off the engine + /// representation. + #[cfg(any(feature = "mem", feature = "postgres", feature = "sqlite"))] + fn of(engine: &KvEngine) -> FleetKvShape { + match engine { + #[cfg(feature = "mem")] + KvEngine::Mem(_) => FleetKvShape::Mem, + #[cfg(feature = "postgres")] + KvEngine::Postgres(_) => FleetKvShape::Postgres, + #[cfg(feature = "sqlite")] + KvEngine::Sqlite(_) => FleetKvShape::Sqlite, + } + } +} + +/// **The store-instance id's minting** (named design point): 32 raw +/// bytes — a SHA-256 fold over the process id, the minting instant's +/// wall-clock nanos, a process-global instance counter, and an +/// address-space-randomized anchor — minted once at construction, +/// stable for the instance's life, carried as the `owner` bytes of +/// every [`crate::gc::PinRow`] the store mints. Uniqueness is the +/// contract, not secrecy: the counter distinguishes instances in one +/// process by construction; time + pid + the ASLR anchor distinguish +/// processes. The id is crate-internal (never public API — ADR-008: +/// fleet constructor parameters, not API). +#[cfg(feature = "postgres")] +fn mint_owner_id() -> Vec { + static INSTANCE_SEQ: AtomicU64 = AtomicU64::new(0); + let seq = INSTANCE_SEQ.fetch_add(1, Ordering::Relaxed); + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let anchor = &seq as *const u64 as usize; + let mut hasher = Sha256::new(); + hasher.update(u64::from(std::process::id()).to_le_bytes()); + hasher.update(nanos.to_le_bytes()); + hasher.update(seq.to_le_bytes()); + hasher.update((anchor as u64).to_le_bytes()); + hasher.finalize().to_vec() +} + /// The store instance — the public facade above the tier engines /// (ADR-011 §1's vocabulary: *one running instantiation of the store /// facade — one engine set per the constructor mode, plus dispatch and @@ -348,6 +574,15 @@ pub struct Store { /// The GC state home (ADR-010 §2: the store core owns GC state) /// — the gc-protocol task's machinery lands here. gc: GcHome, + /// The constructor's fleet declaration — `Some` exactly when a + /// `_fleet` constructor declared `fleet: true` (never inferred, + /// ADR-012 §3): the flag the put/sweep paths branch on (crate- + /// internally, never re-derived from the engine set), the pin + /// TTL, and the store-instance id (`FleetState`). Under + /// `fleet: false` — every pre-existing constructor, behavior + /// byte-unchanged — the ADR-005 in-process protocol applies + /// verbatim regardless of engine. + fleet: Option, } /// The constructor modes (ADR-011 §4's pinned set; ADR-012 §6.3's @@ -888,6 +1123,7 @@ impl Store { window_delay: DEFAULT_DELETE_WINDOW_DELAY, mode: MemOnly, gc: GcHome::default(), + fleet: None, } } @@ -906,6 +1142,7 @@ impl Store { window_delay: DEFAULT_DELETE_WINDOW_DELAY, mode: KvOnly, gc: GcHome::default(), + fleet: None, } } @@ -933,6 +1170,7 @@ impl Store { window_delay: DEFAULT_DELETE_WINDOW_DELAY, mode: DualTier, gc: GcHome::default(), + fleet: None, }) } @@ -969,6 +1207,7 @@ impl Store { window_delay, mode: KvOnly, gc: GcHome::default(), + fleet: None, }) } @@ -1009,6 +1248,7 @@ impl Store { window_delay, mode: DualTier, gc: GcHome::default(), + fleet: None, }) } @@ -1040,10 +1280,11 @@ impl Store { /// `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. + /// `EngineFleetState` clauses run), behavior byte-unchanged. The + /// `fleet: true` declaration is this constructor's [`Store::kv_only_postgres_fleet`] + /// twin (wave 6's landed declaration — the composite + /// fleet-validity predicate plus the fleet parameters), slotted in + /// without reshaping; this signature carries no fleet parameter. #[cfg(feature = "postgres")] pub async fn kv_only_postgres( config: &str, @@ -1060,6 +1301,7 @@ impl Store { window_delay, mode: KvOnly, gc: GcHome::default(), + fleet: None, }) } @@ -1096,10 +1338,12 @@ impl Store { /// `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. + /// `EngineFleetState` clauses run), behavior byte-unchanged. The + /// `fleet: true` declaration is this constructor's [`Store::dual_tier_postgres_local_fleet`] + /// twin (wave 6's landed declaration — the composite + /// fleet-validity predicate plus the fleet parameters, the + /// shared-media assertion included), slotted in without + /// reshaping; this signature carries no fleet parameter. #[cfg(feature = "postgres")] pub async fn dual_tier_postgres_local( config: &str, @@ -1118,6 +1362,7 @@ impl Store { window_delay, mode: DualTier, gc: GcHome::default(), + fleet: None, }) } @@ -1159,14 +1404,16 @@ impl Store { /// `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 composite fleet-validity predicate plus the - /// constructor declarations) and slots in without reshaping — no - /// fleet parameter is carried here. **This constructor completes - /// the fleet-valid engine set** — the module doc's recorded - /// forward reference: with pg-lo's landing, every fleet-valid - /// engine has its named constructor, and wave 6 is the - /// declaration's predicate and parameter alone. + /// `EngineFleetState` clauses run), behavior byte-unchanged. The + /// `fleet: true` declaration is this constructor's [`Store::dual_tier_postgres_pglo_fleet`] + /// twin (wave 6's landed declaration — the composite fleet-validity + /// predicate plus the fleet parameters, the same-instance probe + /// included), slotted in without reshaping; this signature carries + /// no fleet parameter. This constructor completed + /// the fleet-valid engine set — the module doc's recorded forward + /// reference, resolved: with pg-lo's landing, every fleet-valid + /// engine had its named constructor before the declaration's + /// predicate and parameters followed. #[cfg(all(feature = "postgres", feature = "pg-lo"))] pub async fn dual_tier_postgres_pglo( config: &str, @@ -1186,6 +1433,212 @@ impl Store { window_delay, mode: DualTier, gc: GcHome::default(), + fleet: None, + }) + } + + /// **kv-only mode, `fleet: true`** (ADR-012 §3's declaration; + /// ADR-011 §4; ADR-012 §6.2) — the fleet re-routing client posture + /// (ADR-008 §3): the postgres kv engine alone, no large tier — the + /// composite fleet-validity predicate's `large = none` arm, the + /// client node's shape whose over-threshold pool content rides the + /// ops surface (the re-routing composition, wave 7's wiring; the + /// over-threshold put-time rejection, ADR-012 §6.2, unchanged + /// under fleet). + /// + /// `config`/`opts`/`threshold`/`window_delay` carry exactly the + /// [`Store::kv_only_postgres`] twin's meaning (the non-fleet + /// posture stays byte-unchanged there). `pin_ttl` (ADR-008 §2) is + /// the fleet constructor parameter — the pin rows' renewal window + /// the store owns and renews on the TTL/3 cadence, the engine's + /// `pin_renew` successor-time body reading the TTL this + /// constructor passed to its backend; [`DEFAULT_PIN_TTL`] is the + /// shipped default (60 s). `Duration::ZERO` is a constructor + /// error at entry (the TTL/3 cadence degenerates at zero; a zero + /// TTL expires every pin at once) — validated before any I/O. + /// + /// **The composite fleet-validity predicate runs at the + /// constructor** (`fleet_validity_error`, keyed off the engine + /// representation like every dispatch in this store): kv = postgres + /// (structural — the only kv engine a fleet constructor names; + /// the predicate's mem/sqlite rejection arms are the clause's + /// testing home) and `large = none` — the re-routing arm. The + /// store carries the fleet state on success (`Store::fleet`): + /// the declaration's flag, the pin TTL, and the store-instance id + /// (`mint_owner_id`). + #[cfg(feature = "postgres")] + pub async fn kv_only_postgres_fleet( + config: &str, + opts: PgOpts, + threshold: u64, + window_delay: Duration, + pin_ttl: Duration, + ) -> Result { + if pin_ttl.is_zero() { + return Err(Error::Io( + "fleet: pin_ttl must be positive — the TTL/3 renewal cadence degenerates \ + at zero, and a zero-TTL pin expires at once" + .to_owned(), + )); + } + let kv = + KvEngine::Postgres(PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?); + if let Some(err) = fleet_validity_error(FleetKvShape::of(&kv), FleetLargeShape::None) { + return Err(err); + } + ensure_gc_schema_at_construction(&kv, None).await?; + Ok(Store { + kv, + large: None, + threshold, + window_delay, + mode: KvOnly, + gc: GcHome::default(), + fleet: Some(FleetState { + pin_ttl, + owner: mint_owner_id(), + }), + }) + } + + /// **dual-tier mode, `fleet: true`** (ADR-012 §3's declaration; + /// ADR-011 §4) — the fleet shared-media posture: `kv = postgres` + + /// `large = local`, the composite fleet-validity predicate's + /// declared-shared-media arm (ADR-008 §3's enforceable seam). The + /// joint entry+pin transaction is unachievable across media, so + /// the put path's ordering under fleet is pin-row-first, + /// publish-second (ADR-012 §4; the put path's wiring is + /// `store-fleet-put`'s). + /// + /// `config`/`opts`/`root`/`threshold`/`window_delay` carry exactly + /// the [`Store::dual_tier_postgres_local`] twin's meaning (the + /// non-fleet posture stays byte-unchanged there). `pin_ttl` is the + /// fleet constructor parameter ([`DEFAULT_PIN_TTL`] the shipped + /// default; `Duration::ZERO` a constructor error at entry). **The + /// large-tier declaration** rides [`FleetSharedMedia`] — the + /// operator's explicit shared-media assertion, a named type the + /// call site cannot construct accidentally; `local` without it + /// never reaches a fleet constructor (the predicate's + /// `LocalUndeclared` arm is the rule's textual form). The + /// composite fleet-validity predicate runs at the constructor + /// (kv = postgres and the declared shared media); the cross-node + /// truth behind the assertion stays the deployment's verified + /// invariant (ADR-008 §3's seam). The store carries the fleet + /// state on success (`Store::fleet`). + #[cfg(feature = "postgres")] + pub async fn dual_tier_postgres_local_fleet( + config: &str, + opts: PgOpts, + root: impl Into, + threshold: u64, + window_delay: Duration, + pin_ttl: Duration, + shared_media: FleetSharedMedia, + ) -> Result { + if pin_ttl.is_zero() { + return Err(Error::Io( + "fleet: pin_ttl must be positive — the TTL/3 renewal cadence degenerates \ + at zero, and a zero-TTL pin expires at once" + .to_owned(), + )); + } + let kv = + KvEngine::Postgres(PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?); + let large = LargeEngine::Local(LocalBackend::new(root).await?); + let _ = shared_media; + if let Some(err) = + fleet_validity_error(FleetKvShape::of(&kv), FleetLargeShape::LocalSharedMedia) + { + return Err(err); + } + ensure_gc_schema_at_construction(&kv, Some(&large)).await?; + Ok(Store { + kv, + large: Some(large), + threshold, + window_delay, + mode: DualTier, + gc: GcHome::default(), + fleet: Some(FleetState { + pin_ttl, + owner: mint_owner_id(), + }), + }) + } + + /// **dual-tier mode, `fleet: true`** (ADR-012 §3's declaration; + /// ADR-009's consolidation posture) — `kv = postgres` + `large = + /// pg-lo`, **the same pg instance** (ADR-012 §4's checkable + /// clause, the predicate's runtime arm): the joint entry+pin tx + /// the fleet put path drives (`store-fleet-put`) is expressible + /// over the shared instance. + /// + /// `config`/`opts`/`large_config`/`large_opts`/`threshold`/ + /// `window_delay` carry exactly the [`Store::dual_tier_postgres_pglo`] + /// twin's meaning (cross-instance kv=postgres + large=pg-lo stays + /// valid only as the non-fleet twin). `pin_ttl` is the fleet + /// constructor parameter ([`DEFAULT_PIN_TTL`] the shipped + /// default; `Duration::ZERO` a constructor error at entry) — it + /// configures the **kv engine's** backend (`pin_renew` is the kv + /// host's clause; the pg-lo engine hosts no fleet state, ADR-012 + /// §5). + /// + /// **The pg-lo same-instance probe (named design point):** the + /// instance's `system_identifier` (via `pg_control_system()`, + /// PG14+) read across one pooled connection per engine and + /// compared — server-side, immune to DSN aliases (textual DSN + /// comparison was rejected: hostname vs IP vs alias cannot decide + /// instance identity). A mismatch, **or any probe failure** (a + /// pre-PG14 server, a denied function call, an unreachable + /// instance), is a constructor error — the probe errors closed, + /// never a silent pass. The composite fleet-validity predicate + /// runs at the constructor alongside the probe; the store carries + /// the fleet state on success (`Store::fleet`). + #[cfg(all(feature = "postgres", feature = "pg-lo"))] + pub async fn dual_tier_postgres_pglo_fleet( + config: &str, + opts: PgOpts, + large_config: &str, + large_opts: PgOpts, + threshold: u64, + window_delay: Duration, + pin_ttl: Duration, + ) -> Result { + if pin_ttl.is_zero() { + return Err(Error::Io( + "fleet: pin_ttl must be positive — the TTL/3 renewal cadence degenerates \ + at zero, and a zero-TTL pin expires at once" + .to_owned(), + )); + } + let kv_backend = PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?; + let large_backend = PgLoBackend::new(large_config, large_opts).await?; + let kv_id = kv_backend.system_identifier().await?; + let large_id = large_backend.system_identifier().await?; + let kv = KvEngine::Postgres(kv_backend); + let large = LargeEngine::PgLo(large_backend); + if let Some(err) = fleet_validity_error(FleetKvShape::of(&kv), FleetLargeShape::PgLo) { + return Err(err); + } + if kv_id != large_id { + return Err(Error::Io(format!( + "fleet: the kv engine and the pg-lo large engine resolve to different pg \ + instances (system_identifier {kv_id} vs {large_id}) — ADR-012 \u{a7}4's \ + same-instance clause; cross-instance pg-lo is valid only as fleet: false" + ))); + } + ensure_gc_schema_at_construction(&kv, Some(&large)).await?; + Ok(Store { + kv, + large: Some(large), + threshold, + window_delay, + mode: DualTier, + gc: GcHome::default(), + fleet: Some(FleetState { + pin_ttl, + owner: mint_owner_id(), + }), }) } @@ -1234,6 +1687,7 @@ impl Store { window_delay, mode: DualTier, gc: GcHome::default(), + fleet: None, }) } @@ -1264,6 +1718,16 @@ impl Store { self.window_delay } + /// The store's fleet state — `Some` exactly when the constructor + /// declared `fleet: true` (the declaration is explicit, never + /// inferred, ADR-012 §3): the put/sweep paths branch on the + /// state's presence, never on re-deriving from the engine set. + /// Crate-internal introspection (`FleetState`): the fleet + /// parameters are constructor parameters, not API (ADR-008 §3). + pub(crate) fn fleet(&self) -> Option<&FleetState> { + self.fleet.as_ref() + } + /// **Write-side tier selection for a given content length** — the /// dispatch seam's routing half (the put seam programs through /// this; read paths do not consult it). Dual-tier: @@ -6386,3 +6850,777 @@ mod postgres_pglo_construction_tests { .expect("a long delay is a legal tuning too"); } } + +/// The composite fleet-validity predicate's per-clause tests (the +/// predicate is the fleet constructors' admission gate; the mem and +/// sqlite clauses have no fleet constructor to reach them through — +/// the recorded "not fleet-anything" postures — so the rejection arms +/// are exercised at the predicate level, the same error the would-be +/// construction carries). Runs in every build: pure vocabulary, no +/// engine I/O. +#[cfg(test)] +mod fleet_validity_predicate_tests { + use super::*; + + /// Every fleet-valid shape admits: kv = postgres and (large = none + /// — the re-routing arm, OR `local` on the declared shared media, + /// OR pg-lo — the same-instance clause's runtime probe rides its + /// constructor). (backends-and-dispatch.md, "The composite + /// fleet-validity rule", one sentence, both tiers.) + #[test] + fn every_fleet_valid_shape_admits() { + assert_eq!( + fleet_validity_error(FleetKvShape::Postgres, FleetLargeShape::None), + None, + "the re-routing posture (large = none) is fleet-valid with kv = postgres" + ); + assert_eq!( + fleet_validity_error(FleetKvShape::Postgres, FleetLargeShape::LocalSharedMedia), + None, + "the declared shared-media arm is fleet-valid" + ); + assert_eq!( + fleet_validity_error(FleetKvShape::Postgres, FleetLargeShape::PgLo), + None, + "the pg-lo arm is fleet-valid (its same-instance clause probes at the constructor)" + ); + } + + /// `mem` anywhere rejects (ADR-012 §3): mem is never fleet-valid + /// (ADR-011 §4) — every large shape over a mem kv engine is the + /// constructor error. + #[test] + fn mem_anywhere_rejects() { + for large in [ + FleetLargeShape::None, + FleetLargeShape::LocalSharedMedia, + FleetLargeShape::LocalUndeclared, + FleetLargeShape::PgLo, + ] { + let err = fleet_validity_error(FleetKvShape::Mem, large) + .expect("mem is never fleet-valid, any large shape"); + assert!( + err.to_string().contains("Mem is never fleet-valid"), + "mem is the kv engine, large = {large:?}: {err}" + ); + } + } + + /// The sqlite kv engine rejects (ADR-012 §3's "any kv engine other + /// than postgres"): a sqlite fleet is a recorded non-option (ADR-010 + /// §3). + #[test] + fn sqlite_rejects() { + for large in [ + FleetLargeShape::None, + FleetLargeShape::LocalSharedMedia, + FleetLargeShape::LocalUndeclared, + FleetLargeShape::PgLo, + ] { + assert!( + fleet_validity_error(FleetKvShape::Sqlite, large).is_some(), + "the sqlite kv engine, large = {large:?}, rejects" + ); + } + } + + /// `local` without the shared-media declaration rejects (ADR-008 + /// §3): a fleet-mode constructor requires the explicit + /// [`FleetSharedMedia`] assertion — the sealed declaration type + /// keeps every real call site on the declared arm, and this is the + /// rule's rejected arm. + #[test] + fn local_without_the_shared_media_declaration_rejects() { + let err = fleet_validity_error(FleetKvShape::Postgres, FleetLargeShape::LocalUndeclared) + .expect("the undeclared arm rejects"); + assert!( + err.to_string().contains("shared-media declaration"), + "the error names the missing declaration: {err}" + ); + } +} + +/// The wave-6 fleet declaration's constructor tests (feature +/// `postgres`): the `_fleet` constructors' fleet state introspection +/// (flag, TTL, owner id), the TTL's engine wiring, the admission gate +/// (a zero TTL errors closed at entry), and the declaration's behavioral +/// nothing-else posture — put/sweep on a fleet store run exactly the +/// non-fleet store's machinery (the fleet mechanism turns on with the +/// later put/sweep driver tasks, not with the declaration). +/// Env-driven (`ALKBLOBS_PG_*`), skipping cleanly server-less. +#[cfg(all(test, feature = "postgres"))] +mod postgres_fleet_construction_tests { + use super::*; + use crate::postgres::test_support::SchemaJanitor; + use std::fs; + use std::io::Cursor; + use std::path::{Path, PathBuf}; + use std::sync::atomic::{AtomicU64, Ordering}; + + /// A fresh temp root per test; 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 dir = std::env::temp_dir().join(format!( + "alkblobs-store-fleet-{tag}-{}-{:x}-{seq}", + std::process::id(), + seq, + )); + 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); + } + } + + static SCHEMA_SEQ: AtomicU64 = AtomicU64::new(0); + + fn fresh_schema(tag: &str) -> String { + format!( + "abtfleet_{}_{}_{:x}", + tag, + std::process::id(), + SCHEMA_SEQ.fetch_add(1, Ordering::Relaxed) + ) + } + + fn env_nonempty(name: &str) -> Option { + std::env::var(name).ok().filter(|v| !v.is_empty()) + } + + fn harness_dsn() -> Option { + 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()), + )) + } + + fn skip_note() { + eprintln!( + "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB" + ); + } + + fn fresh_opts(tag: &str) -> Option<(PgOpts, SchemaJanitor)> { + let dsn = harness_dsn()?; + let schema = fresh_schema(tag); + Some(( + PgOpts { + schema: schema.clone(), + ..Default::default() + }, + SchemaJanitor::new(&dsn, &schema), + )) + } + + fn key_for(content: &[u8]) -> Key { + Key::derive(content, content.len() as u64) + } + + fn postgres_engine(store: &Store) -> &PostgresBackend { + match store.kv() { + 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"), + } + } + + 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" + ); + assert!( + engine.table_exists("gc_pins").await.unwrap(), + "the fleet host's both-tables row: gc_pins exists" + ); + } + + /// The shipped default pin TTL is the pinned 60-second constant + /// (ADR-008 §2) — the fleet constructors' default tuning, the + /// constant the engine's `pin_renew` successor-time body renews + /// against. + #[test] + fn default_pin_ttl_is_the_pinned_sixty_seconds() { + assert_eq!( + DEFAULT_PIN_TTL, + Duration::from_secs(crate::postgres::DEFAULT_PIN_TTL_SECS), + "the store's shipped default is the engine constant's seconds" + ); + assert_eq!( + DEFAULT_PIN_TTL, + Duration::from_secs(60), + "the shipped default is 60 s (ADR-008 §2)" + ); + } + + /// The store-instance id's minting: 32 raw bytes, unique per + /// instance (the process-global counter distinguishes instances in + /// one process; time + pid + the ASLR anchor distinguish + /// processes) — minted at construction, carried as the pin rows' + /// `owner` half, crate-internal. + #[test] + fn owner_id_mints_unique_and_fixed_width() { + let first = mint_owner_id(); + let second = mint_owner_id(); + assert_eq!(first.len(), 32, "the SHA-256 fold's width"); + assert_eq!(second.len(), 32); + assert_ne!( + first, second, + "two instances in one process mint distinct ids" + ); + } + + /// A zero pin TTL errors closed at entry — before any I/O — never + /// a silent pass (the TTL/3 renewal cadence degenerates at zero + /// and a zero-TTL pin expires at once). + #[tokio::test] + async fn zero_pin_ttl_is_a_constructor_error_at_entry() { + let err = Store::kv_only_postgres_fleet( + "host=no-such-host.invalid port=1 user=x password=x dbname=x", + PgOpts::default(), + 64, + DEFAULT_DELETE_WINDOW_DELAY, + Duration::ZERO, + ) + .await + .err() + .expect("the entry validation rejects before any connection attempt"); + assert!( + err.to_string().contains("pin_ttl must be positive"), + "the zero TTL rejects before any connection attempt: {err}" + ); + + let err = Store::dual_tier_postgres_local_fleet( + "host=no-such-host.invalid port=1 user=x password=x dbname=x", + PgOpts::default(), + std::env::temp_dir().join("alkblobs-fleet-zero-ttl"), + 64, + DEFAULT_DELETE_WINDOW_DELAY, + Duration::ZERO, + FleetSharedMedia::asserted(), + ) + .await + .err() + .expect("the entry validation rejects on the shared-media shape too"); + assert!( + err.to_string().contains("pin_ttl must be positive"), + "the zero TTL rejects at entry on the shared-media shape too: {err}" + ); + } + + /// The fleet re-routing client posture constructs and introspects: + /// the fleet state carries the declaration's flag, the pin TTL + /// (default and constructor-tuned), and the store-instance id — + /// distinct across two instances — and the GC schema ran (the + /// fleet host's both-tables row). The over-threshold put-time + /// rejection (ADR-012 §6.2) is unchanged under fleet. + #[tokio::test] + async fn kv_only_postgres_fleet_constructs_and_introspects_fleet_state() { + let Some((opts, _janitor)) = fresh_opts("fleet-kv-only") else { + skip_note(); + return; + }; + let dsn = harness_dsn().expect("harness dsn re-read"); + let store = Store::kv_only_postgres_fleet( + &dsn, + opts, + 64, + DEFAULT_DELETE_WINDOW_DELAY, + DEFAULT_PIN_TTL, + ) + .await + .expect("the fleet kv-only posture constructs"); + assert_eq!(store.mode(), KvOnly); + assert!(store.large().is_none(), "no large tier (large = none)"); + let fleet = store + .fleet() + .expect("fleet: true declares: the state carries"); + assert_eq!(fleet.pin_ttl(), DEFAULT_PIN_TTL); + assert_eq!(fleet.owner().len(), 32); + assert_gc_schema_ran(&store).await; + + for (len, tier) in [ + (0u64, Tier::Kv), + (63, Tier::Kv), + (64, Tier::Kv), + (65, Tier::Kv), + (u64::MAX, Tier::Kv), + ] { + assert_eq!( + store.tier_for(len), + tier, + "kv-only routes every length kv, fleet or not" + ); + } + + let err = store + .put(Some(65), Cursor::new(&b"never read"[..])) + .await + .unwrap_err(); + assert!( + err.to_string().contains("kv-only mode rejects"), + "the put-time rejection is unchanged under the declaration: {err}" + ); + assert_eq!( + store.list().await.unwrap(), + Vec::::new(), + "zero residue" + ); + drop(store); + + let Some((opts, _janitor)) = fresh_opts("fleet-kv-ttl") else { + skip_note(); + return; + }; + let store = Store::kv_only_postgres_fleet( + &dsn, + opts, + 64, + DEFAULT_DELETE_WINDOW_DELAY, + Duration::from_secs(90), + ) + .await + .expect("constructs at a custom pin TTL"); + let fleet = store.fleet().expect("fleet state carries"); + assert_eq!( + fleet.pin_ttl(), + Duration::from_secs(90), + "the TTL is a fleet constructor parameter" + ); + let same_instance_owner = fleet.owner().to_vec(); + drop(store); + + let Some((opts, _janitor)) = fresh_opts("fleet-kv-owner2") else { + skip_note(); + return; + }; + let other = Store::kv_only_postgres_fleet( + &dsn, + opts, + 64, + DEFAULT_DELETE_WINDOW_DELAY, + DEFAULT_PIN_TTL, + ) + .await + .expect("a second instance constructs"); + assert_ne!( + other.fleet().expect("fleet state carries").owner(), + same_instance_owner, + "two instances never share the store-instance id" + ); + } + + /// A declared fleet store behaves exactly as the non-fleet store + /// does until the fleet machinery's drivers land: a put lands on + /// the kv tier, and the sweep's form branch — keyed off the engine + /// representation, never the declaration — runs the durable form + /// on the postgres host (stage this sweep, execute the next). + #[tokio::test] + async fn fleet_declaration_turns_nothing_else_on_yet() { + struct SetSource; + impl LivenessSource for SetSource { + fn live_keys(&self) -> Vec { + Vec::new() + } + fn contains(&self, _key: &Key) -> bool { + false + } + } + + let Some((opts, _janitor)) = fresh_opts("fleet-inert") else { + skip_note(); + return; + }; + let dsn = harness_dsn().expect("harness dsn re-read"); + let store = Store::kv_only_postgres_fleet(&dsn, opts, 64, Duration::ZERO, DEFAULT_PIN_TTL) + .await + .expect("constructs"); + store.register_liveness_source(Arc::new(SetSource)); + + let content = b"a dead candidate on a declared fleet store"; + let (key, pin) = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("the put publishes through the in-process protocol"); + assert_eq!(key, key_for(content)); + assert!(store.has(&key.as_bytes()).await.unwrap()); + drop(pin); + + let staged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(staged.aborted, None); + assert_eq!( + staged.candidates_staged, 1, + "the durable form staged on the kv host (the form branch keys off the \ + engine representation)" + ); + assert_eq!(staged.deleted, 0); + + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + executed.deleted, 1, + "the next sweep executed the overdue set" + ); + assert!(!store.has(&key.as_bytes()).await.unwrap()); + } + + /// The fleet shared-media posture constructs and introspects; the + /// GC schema ran on the kv host (the one-window-host rule intact + /// under fleet — kv = postgres hosts the window; there is no + /// second SQL engine to host anything else). + #[tokio::test] + async fn dual_tier_postgres_local_fleet_constructs_and_introspects() { + let Some((opts, _janitor)) = fresh_opts("fleet-shared-media") else { + skip_note(); + return; + }; + let dsn = harness_dsn().expect("harness dsn re-read"); + let root = temp_root("fleet-shared-media"); + let store = Store::dual_tier_postgres_local_fleet( + &dsn, + opts, + root.path(), + 1024, + DEFAULT_DELETE_WINDOW_DELAY, + DEFAULT_PIN_TTL, + FleetSharedMedia::asserted(), + ) + .await + .expect("the fleet shared-media posture constructs"); + assert_eq!(store.mode(), DualTier); + assert!(store.large().is_some(), "both tiers present"); + let fleet = store + .fleet() + .expect("fleet: true declares: the state carries"); + assert_eq!(fleet.pin_ttl(), DEFAULT_PIN_TTL); + assert_eq!(fleet.owner().len(), 32); + assert_gc_schema_ran(&store).await; + + 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, + "the declaration does not disturb the dispatch" + ); + } + } +} + +/// The wave-6 fleet declaration's pg-lo-side constructor tests +/// (features `postgres` + `pg-lo`): the consolidation posture's fleet +/// twin — the pg-lo same-instance probe (ADR-012 §4) green on the +/// same instance (including the instance-level shape: two distinct +/// *databases* on one instance are one probe-matching instance), the +/// probe's error-closed negative (a DSN that cannot resolve to the kv +/// engine's instance is a constructor error, never a silent pass), and +/// the one-window-host posture (the window hosts on the kv engine; the +/// pg-lo window clauses stay dormant). Env-driven (`ALKBLOBS_PG_*`), +/// skipping cleanly server-less. +#[cfg(all(test, feature = "pg-lo", feature = "postgres"))] +mod postgres_pglo_fleet_construction_tests { + use super::*; + use crate::pg_common::test_support::DatabaseJanitor; + use std::sync::atomic::{AtomicU64, Ordering}; + use tokio_postgres::NoTls; + + static DB_SEQ: AtomicU64 = AtomicU64::new(0); + + fn env_nonempty(name: &str) -> Option { + std::env::var(name).ok().filter(|v| !v.is_empty()) + } + + fn harness_dsn() -> Option { + 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()), + )) + } + + fn skip_note() { + eprintln!( + "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB" + ); + } + + /// A fresh dedicated database per test (LOs are database-level — + /// the C7 lesson); the drop-guard removes it in every exit path. + struct TestDatabase { + dsn: String, + _janitor: DatabaseJanitor, + } + + fn admin_dsn(dsn: &str) -> String { + let base = dsn.rsplit_once("dbname=").expect("dbname").0; + format!("{base} dbname=postgres") + } + + fn db_dsn(dsn: &str, db: &str) -> String { + let base = dsn.rsplit_once("dbname=").expect("dbname").0; + format!("{base} dbname={db}") + } + + async fn fresh_db(tag: &str) -> Option { + let dsn = harness_dsn()?; + let db = format!( + "abtfleetlo_{}_{}_{:x}", + tag, + std::process::id(), + DB_SEQ.fetch_add(1, Ordering::Relaxed) + ); + let admin = admin_dsn(&dsn); + { + let (client, conn) = tokio_postgres::connect(&admin, NoTls) + .await + .expect("admin connect"); + tokio::spawn(async move { + let _ = conn.await; + }); + client + .execute( + &format!( + "CREATE DATABASE {}", + crate::pg_common::quote_identifier(&db) + ), + &[], + ) + .await + .expect("create test database"); + } + Some(TestDatabase { + dsn: db_dsn(&dsn, &db), + _janitor: DatabaseJanitor::new(&admin, &db), + }) + } + + fn fresh_schema(tag: &str) -> String { + format!( + "abtfleetlo_{}_{}_{:x}", + tag, + std::process::id(), + DB_SEQ.fetch_add(1, Ordering::Relaxed) + ) + } + + fn pglo_engine(store: &Store) -> &crate::pg_lo::PgLoBackend { + match store.large() { + Some(LargeEngine::PgLo(engine)) => engine, + Some(LargeEngine::Local(_)) | None => panic!("these tests construct pg-lo-tier stores"), + } + } + + fn postgres_engine(store: &Store) -> &PostgresBackend { + match store.kv() { + 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 consolidation posture's fleet twin constructs on the same + /// pg instance: the fleet state carries (flag, TTL, owner id), the + /// kv host's GC schema ran (the fleet host's both-tables row), and + /// the one-window-host posture composes — the pg-lo window clauses + /// stay dormant (no `gc_window` in the pg-lo schema; two distinct + /// schemas so the dormancy assertion observes genuine absence), + /// `gc_pins` never in the pg-lo schema. + #[tokio::test] + async fn dual_tier_postgres_pglo_fleet_constructs_same_instance() { + let Some(site) = fresh_db("fleet-consolidation").await else { + skip_note(); + return; + }; + let kv_schema = fresh_schema("kv"); + let lo_schema = fresh_schema("lo"); + let store = Store::dual_tier_postgres_pglo_fleet( + &site.dsn, + PgOpts { + schema: kv_schema.clone(), + ..Default::default() + }, + &site.dsn, + PgOpts { + schema: lo_schema.clone(), + ..Default::default() + }, + 1024, + DEFAULT_DELETE_WINDOW_DELAY, + DEFAULT_PIN_TTL, + ) + .await + .expect("the same-instance consolidation posture constructs"); + + assert_eq!(store.mode(), DualTier); + assert!(store.large().is_some()); + let fleet = store + .fleet() + .expect("fleet: true declares: the state carries"); + assert_eq!(fleet.pin_ttl(), DEFAULT_PIN_TTL); + assert_eq!(fleet.owner().len(), 32); + + 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, + "the declaration does not disturb the dispatch" + ); + } + + let kv = postgres_engine(&store); + assert!(kv.table_exists("blobs").await.unwrap()); + assert!( + kv.table_exists("gc_window").await.unwrap(), + "the window hosts on the kv engine (the one-window-host rule under fleet)" + ); + assert!( + kv.table_exists("gc_pins").await.unwrap(), + "the fleet host's both-tables row: gc_pins in the kv schema" + ); + + let lo = pglo_engine(&store); + assert!(lo.table_exists("lo_entries").await.unwrap()); + assert!( + !lo.table_exists("gc_window").await.unwrap(), + "the pg-lo window clauses stay dormant on the dual-SQL instance" + ); + assert!( + !lo.table_exists("gc_pins").await.unwrap(), + "gc_pins is never created by pg-lo (fleet pins live on the kv host)" + ); + drop(store); + let _ = (kv_schema, lo_schema); + } + + /// The probe is instance-level, not database-level: two distinct + /// *databases* on one pg instance carry one `system_identifier` + /// (the data directory's), so the fleet twin accepts the kv engine + /// and the pg-lo engine living in different databases of the same + /// instance — the textual-DSN-comparison alternative could not + /// have answered this shape. + #[tokio::test] + async fn dual_tier_postgres_pglo_fleet_two_databases_one_instance() { + let Some(kv_site) = fresh_db("fleet-two-db-kv").await else { + skip_note(); + return; + }; + let Some(lo_site) = fresh_db("fleet-two-db-lo").await else { + skip_note(); + return; + }; + let store = Store::dual_tier_postgres_pglo_fleet( + &kv_site.dsn, + PgOpts::default(), + &lo_site.dsn, + PgOpts::default(), + 64, + DEFAULT_DELETE_WINDOW_DELAY, + DEFAULT_PIN_TTL, + ) + .await + .expect("two databases on one instance are one fleet-valid instance"); + + let fleet = store + .fleet() + .expect("fleet: true declares: the state carries"); + assert_eq!(fleet.pin_ttl(), DEFAULT_PIN_TTL); + } + + /// The probe errors closed: a large-side DSN that cannot resolve + /// to the kv engine's instance (a database that does not exist on + /// the dev harness — the recorded negative, one instance only) is + /// a constructor error, never a silent pass. The failing point + /// rides the engine open or the probe — either way the + /// constructor rejects. + #[tokio::test] + async fn dual_tier_postgres_pglo_fleet_unresolvable_instance_errors_closed() { + let Some(kv_site) = fresh_db("fleet-closed-kv").await else { + skip_note(); + return; + }; + let nonexistent = db_dsn(&kv_site.dsn, "abtnosuchdb_fleet_negative"); + let err = Store::dual_tier_postgres_pglo_fleet( + &kv_site.dsn, + PgOpts::default(), + &nonexistent, + PgOpts::default(), + 64, + DEFAULT_DELETE_WINDOW_DELAY, + DEFAULT_PIN_TTL, + ) + .await + .err() + .expect("the cross-instance (unresolvable) negative errors closed"); + assert!( + err.to_string().contains("postgres"), + "the rejection is a pg-side failure surfaced as the constructor error: {err}" + ); + assert!( + !matches!( + err, + Error::Missing | Error::Verification | Error::KeyInvalid | Error::GcRefuse + ), + "the failure never masquerades as a non-Io member" + ); + } + + /// A zero pin TTL errors closed at entry — before any I/O. The + /// entry check runs server-less. + #[tokio::test] + async fn zero_pin_ttl_is_a_constructor_error_at_entry() { + let err = Store::dual_tier_postgres_pglo_fleet( + "host=no-such-host.invalid port=1 user=x password=x dbname=x", + PgOpts::default(), + "host=no-such-host.invalid port=1 user=x password=x dbname=x", + PgOpts::default(), + 64, + DEFAULT_DELETE_WINDOW_DELAY, + Duration::ZERO, + ) + .await + .err() + .expect("the entry validation rejects before any connection attempt"); + assert!( + err.to_string().contains("pin_ttl must be positive"), + "the zero TTL rejects at entry on the consolidation shape too: {err}" + ); + } +} diff --git a/tasks/store-fleet-construction.md b/tasks/store-fleet-construction.md index 1985887..dc7a4ec 100644 --- a/tasks/store-fleet-construction.md +++ b/tasks/store-fleet-construction.md @@ -1,7 +1,7 @@ --- id: store-fleet-construction name: the fleet declaration — the composite fleet-validity predicate at the constructor -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -173,8 +173,114 @@ is the admission gate and the state those tasks consume. ## Notes -> To be filled by implementation agent +Landed 2026-10-12. Design decisions (each recorded in-code as a named +design point; all keyed off the landed ADR text): + +- **The declaration slots in as `_fleet` twin constructors, without + reshaping (`kv_only_postgres_fleet`, `dual_tier_postgres_local_fleet`, + `dual_tier_postgres_pglo_fleet`).** Every pre-existing constructor's + signature and behavior is byte-unchanged and stays implicitly + `fleet: false` (the ADR-005 in-process protocol verbatim) — the + recorded non-event landed as recorded; no fleet parameter rides the + non-fleet signatures. The store carries `fleet: + Option` — `Some` exactly under the declaration, the flag + the put/sweep paths branch on (never re-derived from the engine set). +- **The declaration's type shape (named design point):** the pin TTL + rides as a `Duration` parameter (mirroring `window_delay`'s shape, + shipped default the re-exported `DEFAULT_PIN_TTL`, 60 s), and the + shared-media assertion is a **sealed dedicated type** — + `FleetSharedMedia` with a private seal field; the caller cannot + construct it accidentally and must name `FleetSharedMedia::asserted()` + at the call site (the reading ADR-008 §3's "explicit declaration" + wording supports). No bare flag anywhere. +- **The composite fleet-validity predicate** is one function + (`fleet_validity_error` over engine-representation-derived shape + descriptors `FleetKvShape`/`FleetLargeShape`), one sentence, both + tiers, run inside every fleet constructor keyed off the engine + representation like every other dispatch in this store. **Enforcement + map (recorded):** the structural clauses (kv = postgres, mem never, + sqlite never, `local` needs the declaration) are enforced by the + constructor set's shape itself — no `_fleet` constructor is named for + a mem/sqlite kv engine (the recorded "not fleet-anything" postures + carry no fleet parameter) — with the predicate's rejection arms + unit-tested per clause at the predicate level (the same error a + would-be construction carries); the pg-lo shape's runtime-checkable + clause probes at its constructor. +- **The same-instance probe (named design point):** the instance's + `system_identifier` via `pg_control_system()` (PG14+) read across + one pooled connection per engine (`PostgresBackend::system_identifier` + / `PgLoBackend::system_identifier`) and compared — server-side + instance identity, immune to DSN aliases (textual DSN comparison + rejected as specified). **A mismatch or any probe failure is a + constructor error (errors closed, never a silent pass).** The probe + is instance-level, not database-level: verified positively with two + distinct *databases* on one instance (one `system_identifier` — the + data directory's). +- **The owner id's minting (named design point):** 32 raw bytes, a + SHA-256 fold over process id + wall-clock nanos + a process-global + instance counter + an ASLR anchor; minted once at construction, + stable for the instance's life; uniqueness is the contract (the + counter separates instances in one process; time+pid+ASLR separate + processes), not secrecy; crate-internal, never public API. Carried + as the `owner` bytes of every `PinRow` the store will mint. +- **The pin TTL's wiring:** `PostgresBackend::new_with_pin_ttl` (the + fleet constructors' instrument; `new` keeps the shipped default) — + proven against the live server: a backend opened through the + constructor renews a live pin row to `now + pin_ttl` (the + constructor-carried value, not the default). + `Duration::ZERO` is a constructor error at entry, before any I/O + (the TTL/3 cadence degenerates at zero). +- **The one-window-host posture verified intact:** every fleet-valid + configuration has kv = postgres, and the fleet constructors' tests + assert the kv host's both-tables GC schema and the pg-lo schema's + dormancy (no `gc_window`, no `gc_pins`) — the form branch untouched + (keys off the engine representation, never a flag); a behavioral + test additionally proves the declaration turns nothing else on (a + fleet store's put/sweep run exactly the non-fleet machinery until + the put/sweep driver tasks land). +- **Fleet constructor parameters, not API (ADR-008 §3):** `FleetState`, + `fleet_validity_error`, the shape descriptors, and `mint_owner_id` + are crate-internal; the public surface adds only the three + constructors, `FleetSharedMedia`, and `DEFAULT_PIN_TTL` + (feature `postgres`). +- **Docs updated to the landed state:** the store.rs module doc's + fleet-declaration paragraph, the three postgres constructors' "Non-fleet + (ADR-012 §3)" recordings, the `#![allow(dead_code)]` scope note, the + postgres.rs pin-TTL design points (`new_with_pin_ttl` landed), the + gc.rs fleet-family note, and the pg-lo probe note. The remaining + wave-6 forward references (joint-tx put driver, fleet sweep, joint + contribution) are the later tasks', accurately scoped. +- **Doc hygiene:** `cargo doc --no-deps` zero warnings in the default + build (the PGLO-1 class avoided: the module doc de-links the + feature-gated/private fleet items rather than widening gates; the + pre-existing gated-build doc-link warnings were counted before and + after — identical, not this change's). ## Summary -> To be filled on completion \ No newline at end of file +The fleet declaration landed: three `_fleet` constructors +(`kv_only_postgres_fleet`, `dual_tier_postgres_local_fleet`, +`dual_tier_postgres_pglo_fleet`) carry the explicit `fleet: true` +declaration with the composite fleet-validity predicate enforced at the +constructor (kv = postgres AND large = pg-lo same-instance / local on +the sealed shared-media declaration / none-the-re-routing posture) and +the fleet parameters (pin TTL, default 60 s, wired through +`new_with_pin_ttl` to the engine's renewal body; `FleetSharedMedia` +sealed assertion; the probe = `pg_control_system().system_identifier` +compared across one connection per engine, mismatch or failed probe a +constructor error). The store carries its fleet state (`FleetState`: +flag, TTL, the 32-byte minted owner id) — `fleet: false` postures +byte-unchanged (all existing suites green untouched). Per-clause tests: +every fleet-valid shape constructs and introspects against the live +harness server (including the two-databases-one-instance probe shape +and the kv-host window/dormancy assertions); mem, sqlite, and local- +without-declaration reject per the predicate's tested arms; a +unresolvable-instance DSN errors closed; a zero TTL rejects at entry. +Verified: `cargo test` (default) green; `--features postgres` and +`--features "postgres,pg-lo"` green server-less (skipping cleanly) and +against the harness server (pglo-poc, :15432, PG 16.15 — verified live +2026-10-12); `--all-features` green; clippy `-D warnings` green (all +four feature combinations); `cargo fmt --check` green; `cargo doc +--no-deps` clean. Wave 6's remaining tasks (`store-fleet-put`, +`store-fleet-sweep`, `engine-pg-lo-joint`, `suite-column-fleet`) consume +this task's declaration and state. \ No newline at end of file