From b6b353242ba911ab650c66b1792646050977837f Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 14:53:55 +0000 Subject: [PATCH] =?UTF-8?q?feat(store):=20the=20fleet=20declaration=20?= =?UTF-8?q?=E2=80=94=20the=20composite=20fleet-validity=20predicate=20at?= =?UTF-8?q?=20the=20constructor=20(wave=206)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Three `_fleet` twin constructors slot the explicit `fleet: true` declaration in without reshaping the fleet-capable constructors; every pre-existing constructor stays implicitly `fleet: false` with byte-unchanged behavior (the recorded non-event landed as recorded): `kv_only_postgres_fleet` (the re-routing client posture, large = none), `dual_tier_postgres_local_fleet` (the shared-media posture), `dual_tier_postgres_pglo_fleet` (the consolidation posture). - The composite fleet-validity predicate (`fleet_validity_error`) is one function, one sentence, both tiers, run at the constructor keyed off the engine representation: kv = postgres AND (pg-lo on the same pg instance OR local on declared shared media OR none). The structural clauses (mem never, sqlite never, local needs the declaration) are enforced by the constructor set's shape and tested per clause at the predicate level; the pg-lo same-instance clause is the runtime arm: `pg_control_system().system_identifier` compared across one pooled connection per engine — mismatch or failed probe is a constructor error, never a silent pass. - The named design points decided and recorded: the shared-media assertion is a sealed dedicated type (`FleetSharedMedia::asserted()`, never a bare flag); the pin TTL is a fleet constructor parameter (default `DEFAULT_PIN_TTL` 60 s, `Duration::ZERO` rejected at entry) wired through `PostgresBackend::new_with_pin_ttl` to the renewal body (proven against the live server); the owner id mints 32 raw bytes at construction (SHA-256 fold over pid/nanos/instance counter/ASLR anchor), stable per instance, crate-internal. - The store carries its fleet state (`FleetState`: the declaration flag, the TTL, the owner id) — crate-internal (fleet parameters, not API); the one-window-host posture verified intact under fleet (kv host's both-tables GC schema, pg-lo window clauses dormant, the form branch keyed off the engine representation), and the declaration turns nothing else on yet (put/sweep run the non-fleet machinery until the driver tasks). - Constructor docs updated to the landed state (module doc's fleet-declaration paragraph, the three non-fleet recordings, the scope notes, gc.rs/postgres.rs/pg_lo.rs wave-6 notes); `cargo doc --no-deps` clean in the default build. Verification: cargo test (default) green; --features postgres and --features "postgres,pg-lo" green server-less (skipping cleanly) and against the harness server (pglo-poc, :15432 — verified live 2026- 10-12); --all-features green; clippy -D warnings green (default + postgres + postgres,pg-lo + all-features); cargo fmt --check green; cargo doc --no-deps clean. --- src/gc.rs | 8 +- src/lib.rs | 3 + src/pg_lo.rs | 18 + src/postgres.rs | 147 +++- src/store.rs | 1320 ++++++++++++++++++++++++++++- tasks/store-fleet-construction.md | 112 ++- 6 files changed, 1544 insertions(+), 64 deletions(-) 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