From 950af9ab0d37131e39d4e99029a76baf596f3060 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 16:08:29 +0000 Subject: [PATCH] =?UTF-8?q?feat(store):=20the=20fleet=20put=20path=20?= =?UTF-8?q?=E2=80=94=20the=20joint=20entry+pin=20tx=20driver=20and=20the?= =?UTF-8?q?=20pin=20guard's=20fleet=20form=20(wave=206)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Under fleet: true, puts route through the joint entry+pin tx (store-fleet-put): kv tier via begin_joint_tx -> joint_put (CAS; a dedup re-put commits the pin row alone, upsert-renewing (key, owner)) -> pin_commit — entry + pin row in one step, verified against an independent connection; large = pg-lo rides the one-tx consolidation form (known- and unknown-length; the 512 KiB knee, the empty-contribution re-put form, read-your-own-writes derivation); large = local on declared shared media is pin-first, publish-second (the row-only joint tx, then the staged commit-rename; the orphan-pin crash window is the expiry machinery's). - Amendment A1 landed: catch-propagate-terminate — every error path between begin_joint_tx and pin_commit terminates the held tx before propagating (no mid-flight PgJointTx returns dirty to the pool, PG-2); tested with a forced mid-driver failure (zero residue, the pool's idle-in-transaction count zero, a follow-up put clean). - The pin guard's fleet form: the returned guard carries the row's TTL/3 renewal duty — a per-guard tokio task (spawned at the row's durability), silent on Err, ended by the never-resurrect Ok(false), aborted at every guard-life boundary (drop, batch contribution, conversion); PinRenewal gated feature postgres. - KvEngine::Postgres holds Arc so detached renewal tasks share the fleet host's engine; tokio gains the time feature for the cadence. - Non-fleet put paths byte-unchanged (all existing suites green); a non-fleet test pins that no pin rows are ever committed. - Tests: new postgres_fleet_put_tests (fresh schema + SchemaJanitor; kv joint atomicity both-or-neither, forced pin_commit failure, renewal cadence/drop/conversion stop, never-resurrect, batch release ends renewals, dedup upsert-renewal + by-row composition, local pin-first success + forced-failure forms, non-fleet non-event) and postgres_pglo_fleet_put_tests (fresh db + DatabaseJanitor; the A1 pg-lo failure form, the consolidation round-trip through the public put). Env-driven, skipping cleanly server-less. Verified: cargo test (default) green; --features postgres and --features "postgres,pg-lo" green server-less (skipping) and against the harness server (pglo-poc, :15432); --all-features green live; pg-pure and mem,pg-lo pure builds green; clippy -D warnings green (all combos); cargo fmt --check green; cargo doc --no-deps clean (default; the gated builds' pre-existing warnings unchanged). Schema/db residue 0. --- Cargo.toml | 2 +- src/facade.rs | 115 ++- src/gc.rs | 18 +- src/local.rs | 11 + src/pg_lo.rs | 11 + src/postgres.rs | 20 +- src/store.rs | 1808 +++++++++++++++++++++++++++++++++++++- tasks/store-fleet-put.md | 147 +++- 8 files changed, 2080 insertions(+), 52 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 94fe681..2ab0c08 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,7 +26,7 @@ pg-lo = ["dep:tokio-postgres", "dep:deadpool-postgres"] ops = ["dep:alkcall"] [dependencies] -tokio = { version = "1", default-features = false, features = ["rt", "sync", "io-util", "macros"] } +tokio = { version = "1", default-features = false, features = ["rt", "sync", "io-util", "macros", "time"] } thiserror = "2" parking_lot = "0.12" diff --git a/src/facade.rs b/src/facade.rs index 229d0ce..4ba8b59 100644 --- a/src/facade.rs +++ b/src/facade.rs @@ -147,10 +147,18 @@ pub struct EntryMeta { /// re-put acquires no new pool entry but refcounts the existing one /// under the same arbitration lock (ADR-005). /// - Fleet correspondence (ADR-010 §4, ADR-012 §2): in fleet mode the -/// joint tx commits the expiry-carrying pin row and this guard drops -/// at response time (the row's expiry is then the protection); in -/// non-fleet modes the ops layer's token is this guard's in-process -/// handle — nothing expires outside fleet mode. +/// joint tx commits the expiry-carrying pin row alongside the entry, +/// and **the guard carries the row's renewal duty while it lives** — +/// the store's TTL/3 schedule (ADR-008 §2; the landed +/// `store-fleet-put` design point: a per-guard tokio task, aborted on +/// the guard's drop/conversion/batch contribution, silent on a failed +/// renewal, ended by the never-resurrect `Ok(false)` — the row's +/// expiry is the once-protection after the guard leaves). A guard +/// that outlives its store instance keeps renewing through the +/// registry-adjacent engine handle it carries (same detachment as the +/// refcount bookkeeping); in non-fleet modes the ops layer's token is +/// this guard's in-process handle — nothing expires outside fleet +/// mode. pub struct Pin { registry: Arc, key: Key, @@ -158,6 +166,15 @@ pub struct Pin { /// release duty moved to the batch's one-step drop) or consumed by /// the ordered-step conversion. armed: bool, + /// The fleet pin row's renewal duty — `Some` exactly on a guard a + /// fleet put returned (store-fleet-put's fleet form of + /// pin-before-publish, ADR-008 §2). Aborted wherever the guard's + /// life or ownership ends: drop, batch contribution (the batch's + /// one-step release), or the ordered-step conversion. Gated to the + /// fleet-capable build (feature `postgres`) — fleet state can only + /// ride the postgres host. + #[cfg(feature = "postgres")] + renewal: Option, } impl Pin { @@ -174,6 +191,34 @@ impl Pin { registry, key, armed: true, + #[cfg(feature = "postgres")] + renewal: None, + } + } + + /// Attach the pin row's TTL/3 renewal duty (store-core call, fleet + /// puts only): while the guard lives, its task moves the row's + /// expiry forward — see the struct's fleet-correspondence bullet. + #[cfg(feature = "postgres")] + pub(crate) fn attach_renewal(&mut self, renewal: PinRenewal) { + self.renewal = Some(renewal); + } + + /// Take the renewal duty out (the batch contribution's move: the + /// batch's one-step release owns it from here). + #[cfg(feature = "postgres")] + pub(crate) fn take_renewal(&mut self) -> Option { + self.renewal.take() + } + + /// Stop the renewal duty without moving it — the ordered-step + /// conversion's pre-stop (the pin's own drop is suppressed there, + /// so the abort must happen before [`RegistryCore::convert`]'s + /// `ManuallyDrop` takes over). + #[cfg(feature = "postgres")] + pub(crate) fn stop_renewal(&mut self) { + if let Some(renewal) = self.renewal.take() { + renewal.stop(); } } } @@ -192,6 +237,40 @@ impl Drop for Pin { if self.armed { self.registry.release_pin(&self.key); } + #[cfg(feature = "postgres")] + if let Some(renewal) = self.renewal.take() { + renewal.stop(); + } + } +} + +/// The fleet pin row's renewal duty — the guard's TTL/3 renewal task's +/// handle (store-fleet-put's named design point: a **per-guard tokio +/// task**; the shared-renewal-sweep keyed off a guard registry was the +/// rejected alternative — it needs the same lifecycle notifications an +/// abort handle gives for free). The task moves the row's expiry +/// forward every TTL/3 while it lives; wherever the guard's life or +/// ownership ends (drop, batch contribution, conversion), the handle +/// aborts it — no leaked tasks, no renewal past the guard's life. +/// Crate-internal, minted by the store core's fleet put path only. +#[cfg(feature = "postgres")] +pub(crate) struct PinRenewal { + handle: tokio::task::JoinHandle<()>, +} + +#[cfg(feature = "postgres")] +impl PinRenewal { + /// Store-core constructor: the duty wraps the renewal task the + /// fleet put path spawned. + pub(crate) fn spawned(handle: tokio::task::JoinHandle<()>) -> PinRenewal { + PinRenewal { handle } + } + /// Stop the duty — aborts the task at its next await point. A + /// renewal caught mid-flight is harmless: the clause's expiry + /// update is idempotent, and a task killed between its sleep and + /// the UPDATE leaves the row exactly as it was. + pub(crate) fn stop(&self) { + self.handle.abort(); } } @@ -216,6 +295,14 @@ impl Drop for Pin { pub struct Batch { registry: Arc, keys: Vec, + /// The members' renewal duties, moved out of each contributed pin + /// ([`Batch::add`]) and aborted by the one-step release — the batch + /// group's life ends the members' fleet renewals with it + /// (store-fleet-put: no renewal past the batch's life). Gated to + /// the fleet-capable build, feature `postgres` ([`PinRenewal`]'s + /// gate). + #[cfg(feature = "postgres")] + renewals: Vec, } impl Batch { @@ -225,15 +312,22 @@ impl Batch { Batch { registry, keys: Vec::new(), + #[cfg(feature = "postgres")] + renewals: Vec::new(), } } /// Contribute a put's pin to the group (store-core call). The /// contributed guard is defused — its own drop releases nothing; - /// the batch's one-step release owns the protection from here. - pub(crate) fn add(&mut self, pin: Pin) { - let mut pin = pin; + /// the batch's one-step release owns the protection from here — + /// and its renewal duty moves into the batch (release stops it; + /// while the batch lives, the member's row keeps renewing). + pub(crate) fn add(&mut self, mut pin: Pin) { pin.armed = false; + #[cfg(feature = "postgres")] + if let Some(renewal) = pin.take_renewal() { + self.renewals.push(renewal); + } self.keys.push(pin.key().clone()); } @@ -264,6 +358,13 @@ impl Batch { /// the batching put seam's commit step (which consumes the batch /// the same way). pub(crate) fn release_members(&mut self) { + #[cfg(feature = "postgres")] + { + let renewals = std::mem::take(&mut self.renewals); + for renewal in &renewals { + renewal.stop(); + } + } self.registry.release_pins(std::mem::take(&mut self.keys)); } } diff --git a/src/gc.rs b/src/gc.rs index bc374b7..98a12dd 100644 --- a/src/gc.rs +++ b/src/gc.rs @@ -70,11 +70,13 @@ //! 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 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. +//! family's put-path consumer has landed (the joint-tx put driver, +//! `store-fleet-put`: the store core's `begin_joint_tx` → engine +//! contribution → `pin_commit` flow, plus the pin guard's TTL/3 +//! renewal task minted there); the fleet sweep (`store-fleet-sweep`) +//! holds the family's remaining scope — the sweep lease and the +//! SQL-level arbitration verbs are contract-tested but not yet driven. +//! The note shrinks as those land. #![allow(dead_code)] @@ -423,6 +425,12 @@ impl RegistryCore { /// Consumes the pin with its own drop suppressed (the release ran /// inside the section). pub(crate) fn convert(&self, pin: Pin, register: F) { + #[cfg(feature = "postgres")] + let pin = { + let mut pin = pin; + pin.stop_renewal(); + pin + }; let pin = std::mem::ManuallyDrop::new(pin); let key = pin.key().clone(); let mut state = self.state.lock(); diff --git a/src/local.rs b/src/local.rs index 93a6a47..6d90c03 100644 --- a/src/local.rs +++ b/src/local.rs @@ -95,6 +95,17 @@ impl LocalBackend { fail_commit: false, }) } + + /// Test-only injection arm: force the staged put's commit-rename + /// failure form so the store-core driver's pin-first/publish-second + /// ordering (store-fleet-put, ADR-012 §4) is exercised against the + /// real engine — the pin row lands, the publish fails, the orphan + /// pin is the expiry machinery's. Mirrors the other engines' + /// cfg(test) knobs. + #[cfg(test)] + pub(crate) fn inject_fail_commit(&mut self) { + self.fail_commit = true; + } } fn io_err(e: std::io::Error) -> Error { diff --git a/src/pg_lo.rs b/src/pg_lo.rs index 6f6c172..b8dc1a8 100644 --- a/src/pg_lo.rs +++ b/src/pg_lo.rs @@ -336,6 +336,17 @@ impl PgLoBackend { }) } + /// Test-only injection arm: force the joint contribution's + /// publish-time failure form (the mid-driver failure the + /// store-core driver's A1 termination design is exercised against + /// through the public `put` — the contribution returns with the + /// tx the way it found it, the driver owes the termination). + /// Mirrors the other engines' cfg(test) knobs. + #[cfg(test)] + pub(crate) fn inject_fail_commit(&mut self) { + self.fail_commit = true; + } + /// 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 diff --git a/src/postgres.rs b/src/postgres.rs index 53ddd7e..05974bc 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -111,10 +111,13 @@ //! [`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 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 +//! lease. The store core's put driver consumes the families' joint-tx +//! and pin-renew clauses on the fleet put path (landed — the +//! `store-fleet-put` driver: `begin_joint_tx` → joint contribution → +//! `pin_commit`, plus the guard's TTL/3 renewal task driving +//! `pin_renew`); the sweep lease and the SQL arbitration verbs are the +//! fleet sweep's (`store-fleet-sweep`) — 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 //! [`Store::dual_tier_postgres_local`] and [`Store::kv_only_postgres`], @@ -339,6 +342,15 @@ impl PostgresBackend { Self::system_identifier_of(&self.pool).await } + /// Test-only injection arm: force [`EngineFleetState::pin_commit`]'s + /// failure form (the kv joint tx's terminal step) so the store-core + /// driver's A1 propagation path is exercised through the public + /// `put`. Mirrors the other engines' cfg(test) knobs. + #[cfg(test)] + pub(crate) fn inject_fail_joint_commit(&mut self) { + self.fail_joint_commit = true; + } + /// The probe's body over a pool — shared with the pg-lo engine's /// twin. async fn system_identifier_of(pool: &Pool) -> Result { diff --git a/src/store.rs b/src/store.rs index a8c7ac1..0abba19 100644 --- a/src/store.rs +++ b/src/store.rs @@ -202,6 +202,20 @@ //! Members publish pin-before-publish exactly as lone puts do; the //! group's pins drop together on [`Batch::commit`] or drop (the //! `gc-protocol` machinery drives). +//! - **The fleet put path (store-fleet-put; landed):** under `fleet: +//! true` the put seam's routing branches on the fleet flag only — +//! kv-tier puts drive the joint entry+pin tx (`begin_joint_tx` → +//! `joint_put` (CAS; a dedup re-put commits the pin row alone) → +//! `pin_commit`; publish + protection in one step), the pg-lo tier's +//! staged form *is* the joint form (one tx over the shared +//! instance), and the local tier's staged form is pin-first, +//! publish-second (the row-only tx, then the commit-rename). The +//! returned `Pin` carries the row's TTL/3 renewal duty (a per-guard +//! tokio task, aborted at every guard-life boundary); the non-fleet +//! paths are the ADR-005 in-process protocol verbatim — byte- +//! unchanged. The named design points (A1's catch-propagate- +//! terminate, the renewal lifecycle, the pin-first ordering) are +//! recorded on the fleet impl block in this module. //! //! **The fleet declaration (ADR-012 §3; landed — wave 6's //! `store-fleet-construction`).** Fleet mode is a constructor @@ -291,10 +305,13 @@ //! 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 -//! 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`). +//! declaration and the fleet put path have landed (the constructors' +//! predicate, parameters, and fleet state; the joint-tx put driver — +//! `begin_joint_tx` → the engine's contribution → `pin_commit`, with +//! the pin guard's TTL/3 renewal duty); the fleet sweep +//! (`store-fleet-sweep`) holds the fleet mechanism's remaining +//! store-core scope — the sweep lease and the SQL-level arbitration +//! clauses are contract-tested but not yet driven. #![allow(dead_code)] @@ -310,6 +327,8 @@ use tokio::io::{AsyncRead, AsyncReadExt}; use crate::backend::{Backend, ReadCursor, StagingCursor}; use crate::dispatch::{route, Tier}; +#[cfg(feature = "postgres")] +use crate::facade::PinRenewal; use crate::facade::{ Batch, Blob, EntryMeta, GcAbortCause, LivenessSource, Pin, ProtectCallback, Registration, SweepReport, @@ -326,10 +345,13 @@ use crate::pg_common::PgOpts; #[cfg(feature = "pg-lo")] use crate::pg_lo::{PgLoBackend, PgLoCursor, PgLoStaging}; #[cfg(feature = "postgres")] -use crate::postgres::{PostgresBackend, PostgresCursor}; +use crate::postgres::{PgJointTx, PostgresBackend, PostgresCursor}; #[cfg(feature = "sqlite")] use crate::sqlite::{SqliteBackend, SqliteCursor}; +#[cfg(feature = "postgres")] +use crate::gc::{EngineFleetState, PinRow}; + #[cfg(feature = "postgres")] use sha2::{Digest as _, Sha256}; @@ -624,9 +646,15 @@ pub(crate) enum KvEngine { /// a one-window-host alongside sqlite: its constructors run the /// GC schema's engine-neutral call (the postgres row's body /// creates `gc_pins` + `gc_window`) and its sweeps run the durable - /// window form on it. Feature `postgres`, default-off. + /// window form on it. Feature `postgres`, default-off. The payload + /// is an `Arc` shared with the store core's detached pin-renewal + /// tasks (fleet puts mint one per guard — the task outlives the + /// `Store`, renewing its row through the engine handle it carries; + /// `PostgresBackend` itself is not `Clone` — its lease checkout + /// must not be), so the fleet constructor's one backend serves both + /// the in-line dispatch and the detached duties. #[cfg(feature = "postgres")] - Postgres(PostgresBackend), + Postgres(Arc), } /// The large-tier engine set. `local` is ungated (dependency-free, @@ -1292,7 +1320,7 @@ impl Store { threshold: u64, window_delay: Duration, ) -> Result { - let kv = KvEngine::Postgres(PostgresBackend::new(config, opts).await?); + let kv = KvEngine::Postgres(Arc::new(PostgresBackend::new(config, opts).await?)); ensure_gc_schema_at_construction(&kv, None).await?; Ok(Store { kv, @@ -1352,7 +1380,7 @@ impl Store { threshold: u64, window_delay: Duration, ) -> Result { - let kv = KvEngine::Postgres(PostgresBackend::new(config, opts).await?); + let kv = KvEngine::Postgres(Arc::new(PostgresBackend::new(config, opts).await?)); let large = LargeEngine::Local(LocalBackend::new(root).await?); ensure_gc_schema_at_construction(&kv, Some(&large)).await?; Ok(Store { @@ -1423,7 +1451,7 @@ impl Store { threshold: u64, window_delay: Duration, ) -> Result { - let kv = KvEngine::Postgres(PostgresBackend::new(config, opts).await?); + let kv = KvEngine::Postgres(Arc::new(PostgresBackend::new(config, opts).await?)); let large = LargeEngine::PgLo(PgLoBackend::new(large_config, large_opts).await?); ensure_gc_schema_at_construction(&kv, Some(&large)).await?; Ok(Store { @@ -1481,8 +1509,9 @@ impl Store { .to_owned(), )); } - let kv = - KvEngine::Postgres(PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?); + let kv = KvEngine::Postgres(Arc::new( + 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); } @@ -1542,8 +1571,9 @@ impl Store { .to_owned(), )); } - let kv = - KvEngine::Postgres(PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?); + let kv = KvEngine::Postgres(Arc::new( + 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) = @@ -1611,7 +1641,7 @@ impl Store { .to_owned(), )); } - let kv_backend = PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?; + let kv_backend = Arc::new(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?; @@ -1965,6 +1995,17 @@ impl Store { /// mid-stream threshold overflow — lengths are not known at /// construction, so put time is the only possible rejection point. /// Zero residue on either rejection. + /// + /// **Fleet puts (ADR-012 §3/§4):** under a `fleet: true` + /// declaration the same signature routes the fleet forms — the kv + /// tier through the joint entry+pin tx (the joint-tx driver's + /// named design points, the termination design included, are + /// recorded on this module's fleet impl block), the pg-lo tier + /// through the one-tx consolidation form, the local tier + /// pin-first/publish-second; the returned guard carries the pin + /// row's TTL/3 renewal duty. Under `fleet: false` (every + /// pre-existing constructor) this doc applies verbatim and nothing + /// of the fleet machinery runs. pub async fn put( &self, len: Option, @@ -2048,6 +2089,9 @@ impl Store { return Err(Error::Verification); } let key = Key::from(hasher.finish()); + if let Some(fleet) = self.fleet() { + return self.fleet_put_kv(fleet, key, &value).await; + } let pin = self.gc().acquire(&key).await; self.kv().put(&key.as_bytes(), &value).await?; Ok((key, pin)) @@ -2057,6 +2101,14 @@ impl Store { Some(engine) => engine, None => return Err(Error::Io("no large tier in this mode".to_owned())), }; + #[cfg(all(feature = "postgres", feature = "pg-lo"))] + if let Some(fleet) = self.fleet() { + if let LargeEngine::PgLo(lo) = engine { + // The consolidation posture: the staged form *is* + // the joint form (one tx; ADR-012 §4). + return self.fleet_put_pglo_known(fleet, lo, len, stream).await; + } + } let mut stage = engine.put_staged().await?; let fed: Result<(), Error> = 'stream: loop { let n = match stream.read(&mut chunk).await { @@ -2086,6 +2138,11 @@ impl Store { } Ok(()) => { let key = Key::from(hasher.finish()); + if let Some(fleet) = self.fleet() { + // The shared-media ordering (ADR-012 §4): + // pin row first, publish-second. + return self.fleet_staged_commit(fleet, key, stage).await; + } let pin = self.gc().acquire(&key).await; stage.commit(&key.as_bytes()).await?; Ok((key, pin)) @@ -2107,6 +2164,12 @@ impl Store { &self, mut stream: R, ) -> Result<(Key, Pin), Error> { + #[cfg(all(feature = "postgres", feature = "pg-lo"))] + if let (Some(fleet), Some(LargeEngine::PgLo(lo))) = (self.fleet(), self.large()) { + // The consolidation posture: the unknown-length staged form + // *is* the joint form (one tx; ADR-012 §4). + return self.fleet_put_pglo_unknown(fleet, lo, stream).await; + } let mut buffered = Vec::with_capacity(PUT_CHUNK); let mut stage: Option = None; let mut streamed_total = 0u64; @@ -2209,6 +2272,11 @@ impl Store { } Ok(()) => { let key = Key::from(hasher.finish()); + if let Some(fleet) = self.fleet() { + // The shared-media ordering (ADR-012 + // §4): pin row first, publish-second. + return self.fleet_staged_commit(fleet, key, staged).await; + } let pin = self.gc().acquire(&key).await; staged.commit(&key.as_bytes()).await?; Ok((key, pin)) @@ -2227,6 +2295,9 @@ impl Store { let mut hasher = BlobHasher::new(Algorithm::GitBlobSha256, streamed_total); hasher.update(&buffered); let key = Key::from(hasher.finish()); + if let Some(fleet) = self.fleet() { + return self.fleet_put_kv(fleet, key, &buffered).await; + } let pin = self.gc().acquire(&key).await; self.kv().put(&key.as_bytes(), &buffered).await?; Ok((key, pin)) @@ -2236,6 +2307,511 @@ impl Store { } } +/// **The fleet put path (store-fleet-put; ADR-010 §2, ADR-012 §3/§4)** — +/// the joint entry+pin transaction's store-core driver and the pin +/// guard's fleet form (the TTL/3 renewal duty, ADR-008 §2), behind the +/// public `put`/`put_in_batch` signatures' unchanged routing: the paths +/// branch on [`Store::fleet`]'s presence (the declaration's flag, never +/// re-derived from the engine set), and every non-fleet path is this +/// module's ADR-005 in-process protocol verbatim. +/// +/// **Named design point — the termination design (amendment A1, the +/// wave-4 gate's owner landed here): catch-propagate-terminate.** The +/// driver's steps between [`Store::begin` of the joint tx] and +/// [`pin_commit`] run inside one catch block (an async block's `?` +/// surface); any error terminates the held tx (a `ROLLBACK` through +/// [`PgJointTx::client`]) *before* propagating — no mid-flight +/// `PgJointTx` ever returns to the pool (deadpool's `Fast` recycle +/// never resets transaction state, PG-2's hazard). An explicit abort +/// verb was rejected — it would need a new ADR (the seam shape is +/// pinned; `pin_commit`'s terminal-consume reading stays), and the +/// rollback here issues nothing the clause bodies did not already +/// guarantee: a `ROLLBACK` on a session with no active transaction is +/// a completed, warning-level no-op, so the blanket termination after +/// any failure (including the clause bodies' own rollback paths) is +/// always safe; `pin_commit`'s failure paths (the terminal step, which +/// consumes the tx) roll back per the wave-4 clause contract. The +/// zero-residue atomicity of the tx itself is the engine clause's +/// contract (wave 4); the driver's duty is only never abandoning the +/// tx mid-flight — its row-only and one-LO drivers below each state +/// their interior `?` surface. +/// +/// **The fleet put's per-shape drivers:** +/// +/// - **kv tier (every fleet shape's sub-threshold route; the +/// re-routing posture's every put):** the in-process refcount mints +/// (pin-before-publish, unchanged), then `begin_joint_tx` → +/// `joint_put` (the engine's CAS row write inside the held tx; a +/// dedup re-put contributes nothing — the recorded CAS form) → +/// `pin_commit(tx, [row])` — publish and protection in one step. +/// - **large = pg-lo (the consolidation posture):** one tx over the +/// shared instance — the driver begins it and hands the large tier +/// its joint staging cursor; the cursor streams the content inside +/// the held tx (the 512 KiB `lowrite` knee rides), the key resolves +/// at stream end, the cursor's publish is the CAS contribution (a +/// re-put: the empty-contribution form — the staged LO unlinked +/// in-tx), and `pin_commit` lands the row with the entry. +/// - **large = local on declared shared media:** the joint tx is +/// unachievable across media, so the ordering is pin-first, +/// publish-second — a row-only joint tx (no `joint_put`) commits the +/// pin row **before** the staged commit-rename publishes the entry; +/// the invariant survives (a visible entry always has a durable pin +/// row) and the crash window moves to the harmless side (an orphan +/// pin row, reaped by the expiry machinery — never the reverse +/// ordering, which is rejected). +/// +/// **Named design points — the renewal lifecycle (the pin guard's +/// fleet form, ADR-008 §2):** +/// +/// - **The duty's mechanism: a per-guard tokio task on a TTL/3 +/// interval.** The shared-renewal-sweep keyed off a guard registry +/// was the rejected alternative: it needs precisely the lifecycle +/// notifications (mint/drop/convert/batch-contribute) that an abort +/// handle gives for free, wrapped in a second registry keyed by +/// `(key, owner)` — machinery without a behavior difference, with +/// the per-guard task count bounded by the live pin count. +/// - **The spawn point is the pin row's durability, not the guard's +/// mint:** the task is spawned right after the row's `pin_commit` +/// succeeded (a renewal arriving at a nonexistent row would read the +/// not-yet-pinned window), and the first cadence tick is a full +/// TTL/3 after that — the row's initial `now + TTL` expiry is never +/// raced. +/// - **The abort-on-drop discipline:** the task is aborted by +/// [`Pin`]'s drop, by [`Batch`]'s one-step release (a contributed +/// member's duty moved into the batch at `add` time), and by the +/// ordered-step conversion (`RegistryCore::convert` stops the duty +/// before its `ManuallyDrop` suppresses the guard's drop) — no +/// leaked tasks, no renewal past the guard's life. A task aborted +/// mid-renewal is harmless: the clause's expiry update is +/// idempotent, and the row is exactly as it was. +/// - **The failure posture: silent-and-continued on `Err`, ended on +/// `Ok(false)`.** The row's expiry is the protection — a missed +/// renewal degrades to ADR-008's expiry semantics, never an error on +/// the put path (the crate carries no logging dependency; an embedder +/// needing visibility into renewals observes the row). A `Ok(false)` +/// — an expired or absent `(key, owner)` row — ends the duty: the +/// engine's `pin_renew` never resurrects (wave 4's recorded +/// semantics), so continuing would only churn. +impl Store { + /// The fleet host — the kv-tier postgres backend behind its `Arc` + /// (the detached renewal tasks share it). Every `fleet: true` + /// store's kv engine is postgres (the composite fleet-validity + /// predicate's kv clause), so a `None` here is only the + /// declaration-absent case; where the fleet put paths read it, the + /// branch on `Store::fleet` has already admitted the posture. + #[cfg(feature = "postgres")] + fn fleet_host(&self) -> Option<&Arc> { + match &self.kv { + KvEngine::Postgres(engine) => Some(engine), + #[cfg(feature = "sqlite")] + KvEngine::Sqlite(_) => None, + #[cfg(feature = "mem")] + KvEngine::Mem(_) => None, + } + } + + /// The pin row a fleet put mints: the store-instance id as the + /// `owner` half, `expiry = now + pin_ttl` (the store resolves the + /// expiry; the TTL lives with the store's renewal schedule — ADR-012 + /// §2's rename). + #[cfg(feature = "postgres")] + fn fleet_pin_row(&self, fleet: &FleetState, key: &Key) -> Result { + let expiry = SystemTime::now() + .checked_add(fleet.pin_ttl()) + .ok_or_else(|| { + Error::Io("fleet: the pin expiry overflows the system clock".to_owned()) + })?; + Ok(PinRow { + key: key.as_bytes(), + owner: fleet.owner().to_vec(), + expiry, + }) + } + + /// Mint the pin guard's renewal duty: the per-guard tokio task on + /// the TTL/3 cadence (the lifecycle's named design points above). + /// + /// The cadence floor (1 ms) is the degenerate-tuning guard: a TTL + /// below 3 ms would floor the interval to zero and churn the engine + /// — the TTL/3 cadence must remain a *cadence*. + #[cfg(feature = "postgres")] + fn spawn_pin_renewal( + &self, + host: &Arc, + fleet: &FleetState, + key: &Key, + ) -> PinRenewal { + let cadence = (fleet.pin_ttl() / 3).max(Duration::from_millis(1)); + let host = Arc::clone(host); + let key = key.as_bytes(); + let owner = fleet.owner().to_vec(); + let handle = tokio::spawn(async move { + loop { + tokio::time::sleep(cadence).await; + match EngineFleetState::pin_renew(&*host, &key, &owner).await { + Ok(true) => {} + Ok(false) => break, + Err(_) => {} + } + } + }); + PinRenewal::spawned(handle) + } + + /// Amendment A1's termination (the design's one verb): a step's + /// failure between `begin_joint_tx` and `pin_commit` lands here — + /// terminate the held tx before propagating. A `ROLLBACK` on a + /// session with no active transaction (the clause bodies' own + /// rollback paths having already closed it) is a completed + /// warning-level no-op, never an error. + #[cfg(feature = "postgres")] + async fn terminate_joint_tx(tx: &mut PgJointTx) { + let _ = tx.client().batch_execute("ROLLBACK").await; + } + + /// The kv tier's fleet put: the joint entry+pin tx with the CAS + /// row write inside it (`joint_put` — a dedup re-put contributes + /// nothing, the recorded CAS form) and the pin row's commit — + /// publish and protection in one step. The returned guard carries + /// the in-process refcount's release duty (unchanged) and the + /// fleet row's renewal duty. The body's error paths terminate the + /// held tx before propagating (A1, above) — the put paths' route + /// into this is only ever a declared fleet store, which the + /// `_fleet` constructors make impossible without the postgres + /// host; the non-postgres build's arm is the dead shape, surfaced + /// as an error never a panic. + async fn fleet_put_kv( + &self, + fleet: &FleetState, + key: Key, + value: &[u8], + ) -> Result<(Key, Pin), Error> { + #[cfg(feature = "postgres")] + { + let host = match self.fleet_host() { + Some(host) => host, + None => { + return Err(Error::Io( + "fleet put without the postgres kv host".to_owned(), + )) + } + }; + let pin = self.gc().acquire(&key).await; + let mut tx = match host.begin_joint_tx().await { + Ok(tx) => tx, + Err(e) => return Err(e), + }; + match EngineFleetState::joint_put(&**host, &mut tx, &key.as_bytes(), value).await { + Ok(()) => {} + Err(e) => { + Self::terminate_joint_tx(&mut tx).await; + return Err(e); + } + } + let row = self.fleet_pin_row(fleet, &key)?; + match EngineFleetState::pin_commit(&**host, tx, std::slice::from_ref(&row)).await { + Ok(()) => {} + Err(e) => return Err(e), + } + let mut pin = pin; + pin.attach_renewal(self.spawn_pin_renewal(host, fleet, &key)); + Ok((key, pin)) + } + #[cfg(not(feature = "postgres"))] + { + let _ = (fleet, key, value); + Err(Error::Io( + "fleet put without the postgres kv host: no fleet constructor exists \ + without it" + .to_owned(), + )) + } + } + + /// The row-only joint tx (pin row alone, no `joint_put`) — the + /// large=local shared-media ordering's pin half (ADR-012 §4) and + /// the terminal-consume seam's edge: between the tx's begin and + /// its `pin_commit` there are no interior steps, so the A1 + /// discipline here is the clause's own failure rollback. + #[cfg(feature = "postgres")] + async fn fleet_pin_only_commit( + &self, + host: &Arc, + row: PinRow, + ) -> Result<(), Error> { + let tx = match host.begin_joint_tx().await { + Ok(tx) => tx, + Err(e) => return Err(e), + }; + match EngineFleetState::pin_commit(&**host, tx, std::slice::from_ref(&row)).await { + Ok(()) => Ok(()), + Err(e) => Err(e), + } + } + + /// The shared-media ordering's staged commit (ADR-012 §4: pin + /// row first, publish-second): the pin row's joint tx commits + /// **before** the staged commit-rename publishes the entry; the + /// crash window is the harmless side (an orphan pin row whose + /// renewal stops at the guard's drop — the put's failure return — + /// and whose expiry reclaims it). A pin-commit failure discards + /// the stage and propagates with zero published residue. + async fn fleet_staged_commit( + &self, + fleet: &FleetState, + key: Key, + stage: AnyStaging, + ) -> Result<(Key, Pin), Error> { + #[cfg(feature = "postgres")] + { + let host = match self.fleet_host() { + Some(host) => host, + None => { + stage.discard().await; + return Err(Error::Io( + "fleet put without the postgres kv host".to_owned(), + )); + } + }; + let mut pin = self.gc().acquire(&key).await; + let row = self.fleet_pin_row(fleet, &key)?; + match self.fleet_pin_only_commit(host, row).await { + Ok(()) => { + pin.attach_renewal(self.spawn_pin_renewal(host, fleet, &key)); + } + Err(e) => { + stage.discard().await; + return Err(e); + } + } + stage.commit(&key.as_bytes()).await?; + Ok((key, pin)) + } + #[cfg(not(feature = "postgres"))] + { + let _ = (fleet, key, stage); + Err(Error::Io( + "fleet put without the postgres kv host: no fleet constructor exists \ + without it" + .to_owned(), + )) + } + } + + /// The joint publish's tail (shared by both pg-lo drivers): the + /// in-process pin mint, the row's `pin_commit` (the entry + pin + /// row in one step — the CAS contribution, fresh or empty, is + /// already inside the tx), and the renewal's spawn onto the durable + /// row. + #[cfg(feature = "postgres")] + async fn fleet_joint_publish( + &self, + fleet: &FleetState, + tx: PgJointTx, + key: Key, + ) -> Result<(Key, Pin), Error> { + let host = match self.fleet_host() { + Some(host) => host, + None => { + return Err(Error::Io( + "fleet put without the postgres kv host".to_owned(), + )) + } + }; + let row = self.fleet_pin_row(fleet, &key)?; + match EngineFleetState::pin_commit(&**host, tx, std::slice::from_ref(&row)).await { + Ok(()) => {} + Err(e) => return Err(e), + } + let mut pin = self.gc().acquire(&key).await; + pin.attach_renewal(self.spawn_pin_renewal(host, fleet, &key)); + Ok((key, pin)) + } + + /// The pg-lo consolidation put's known-length driver: the joint tx + /// begins first, the content streams into the tier's joint staging + /// cursor while hashing (the declared-len `Verification` checks + /// unchanged), the key resolves at stream end, the cursor's + /// publish is the CAS contribution (a re-put takes the empty- + /// contribution form — the staged LO unlinked in-tx), and + /// `pin_commit` lands the row with the entry. + /// + /// Amendment A1's catch-propagate-terminate wraps exactly the + /// steps between the tx's begin and `pin_commit`: any error + /// (stream, staged append, the contribution's publish) is caught, + /// terminates the tx (the staged LO's pages die with it — + /// transactional LO creation, C6), and propagates. + #[cfg(all(feature = "postgres", feature = "pg-lo"))] + async fn fleet_put_pglo_known( + &self, + fleet: &FleetState, + large: &PgLoBackend, + len: u64, + mut stream: R, + ) -> Result<(Key, Pin), Error> { + let host = match self.fleet_host() { + Some(host) => host, + None => { + return Err(Error::Io( + "fleet put without the postgres kv host".to_owned(), + )) + } + }; + let mut tx = match host.begin_joint_tx().await { + Ok(tx) => tx, + Err(e) => return Err(e), + }; + let driven = async { + let mut stage = large.put_joint_staged(&mut tx).await?; + let mut hasher = BlobHasher::new(Algorithm::GitBlobSha256, len); + let mut chunk = vec![0u8; PUT_CHUNK]; + let mut received = 0u64; + let fed: Result<(), Error> = 'stream: loop { + let n = match stream.read(&mut chunk).await { + Ok(n) => n, + Err(e) => break 'stream Err(Error::Io(e.to_string())), + }; + if n == 0 { + break 'stream Ok(()); + } + received = received.saturating_add(n as u64); + if received > len { + break 'stream Err(Error::Verification); + } + hasher.update(&chunk[..n]); + if let Err(e) = stage.append(&chunk[..n]).await { + break 'stream Err(e); + } + }; + fed?; + if received != len { + return Err(Error::Verification); + } + let key = Key::from(hasher.finish()); + let _fresh = stage.publish(&key.as_bytes()).await?; + Ok(key) + } + .await; + match driven { + Ok(key) => self.fleet_joint_publish(fleet, tx, key).await, + Err(e) => { + Self::terminate_joint_tx(&mut tx).await; + Err(e) + } + } + } + + /// The pg-lo consolidation put's unknown-length driver: the + /// buffer-to-threshold phase first, exactly `put_unknown`'s rule 1 + /// (a stream that ends sub-threshold stays on the kv tier's joint + /// buffer form); the tx begins only when the threshold overflows, + /// the buffered prefix flushes into the joint cursor and the + /// stream continues inside the held tx; the derivation pass + /// re-runs over the staged artifact through the joint cursor's + /// read seam (rule 2 — read-your-own-writes inside the held tx); + /// the CAS publish and `pin_commit` close the flow in the one tx. + #[cfg(all(feature = "postgres", feature = "pg-lo"))] + async fn fleet_put_pglo_unknown( + &self, + fleet: &FleetState, + large: &PgLoBackend, + mut stream: R, + ) -> Result<(Key, Pin), Error> { + let host = match self.fleet_host() { + Some(host) => host, + None => { + return Err(Error::Io( + "fleet put without the postgres kv host".to_owned(), + )) + } + }; + let Some(threshold) = self.threshold() else { + return Err(Error::Io( + "no dispatch threshold in effect; large-tier routing cannot decide".to_owned(), + )); + }; + let mut buffered = Vec::with_capacity(PUT_CHUNK); + let mut streamed_total = 0u64; + let mut chunk = vec![0u8; PUT_CHUNK]; + let overflow_n; + + 'feed: loop { + let n = match stream.read(&mut chunk).await { + Ok(n) => n, + Err(e) => return Err(Error::Io(e.to_string())), + }; + if n == 0 { + let mut hasher = BlobHasher::new(Algorithm::GitBlobSha256, streamed_total); + hasher.update(&buffered); + let key = Key::from(hasher.finish()); + return self.fleet_put_kv(fleet, key, &buffered).await; + } + streamed_total = streamed_total.saturating_add(n as u64); + let overflow = (buffered.len() as u64).saturating_add(n as u64) > threshold; + if !overflow { + buffered.extend_from_slice(&chunk[..n]); + continue; + } + overflow_n = n; + break 'feed; + } + + let mut tx = match host.begin_joint_tx().await { + Ok(tx) => tx, + Err(e) => return Err(e), + }; + let driven = async { + let mut stage = large.put_joint_staged(&mut tx).await?; + stage.append(&buffered).await?; + stage.append(&chunk[..overflow_n]).await?; + let fed: Result<(), Error> = 'stream: loop { + let n = match stream.read(&mut chunk).await { + Ok(n) => n, + Err(e) => break 'stream Err(Error::Io(e.to_string())), + }; + if n == 0 { + break 'stream Ok(()); + } + if let Err(e) = stage.append(&chunk[..n]).await { + break 'stream Err(e); + } + }; + fed?; + let mut hasher = BlobHasher::new(Algorithm::GitBlobSha256, streamed_total); + let mut offset = 0u64; + let derived: Result<(), Error> = 'pass: loop { + if offset >= streamed_total { + break 'pass Ok(()); + } + let want = (streamed_total - offset).min(PUT_CHUNK as u64); + let bytes = match stage.read_at(offset, want).await { + Ok(bytes) => bytes, + Err(e) => break 'pass Err(e), + }; + if bytes.is_empty() { + break 'pass Err(Error::Io( + "staged artifact under-read during the derivation pass".to_owned(), + )); + } + hasher.update(&bytes); + offset += bytes.len() as u64; + }; + derived?; + let key = Key::from(hasher.finish()); + let _fresh = stage.publish(&key.as_bytes()).await?; + Ok(key) + } + .await; + match driven { + Ok(key) => self.fleet_joint_publish(fleet, tx, key).await, + Err(e) => { + Self::terminate_joint_tx(&mut tx).await; + Err(e) + } + } + } +} + /// **The delete-window commit batch size** — ~100/batch, iroh's proven /// shape (ADR-005 §Sweep; POC #1 finding 3's named shape adopted). const SWEEP_DELETE_BATCH: usize = 100; @@ -2481,7 +3057,7 @@ impl Store { _ => self.sweep_window_inline(live, started).await, }, #[cfg(feature = "postgres")] - KvEngine::Postgres(engine) => self.sweep_window_durable(engine, live, started).await, + KvEngine::Postgres(engine) => self.sweep_window_durable(&**engine, live, started).await, #[cfg(feature = "sqlite")] KvEngine::Sqlite(engine) => self.sweep_window_durable(engine, live, started).await, } @@ -7233,13 +7809,16 @@ mod postgres_fleet_construction_tests { ); } - /// 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). + /// The fleet declaration's scope discipline after the put driver + /// landed: the kv put routes through the joint entry+pin tx (the + /// pin row lands with the entry — asserted against the pin check), + /// while the *sweep* machinery stays off (the store-fleet-sweep + /// task) — the sweep's form branch keys off the engine + /// representation, never the declaration, and runs the durable + /// form on the postgres host (stage this sweep, execute the next), + /// arbitrating in-process. #[tokio::test] - async fn fleet_declaration_turns_nothing_else_on_yet() { + async fn fleet_declaration_leaves_the_sweep_machinery_to_its_task() { struct SetSource; impl LivenessSource for SetSource { fn live_keys(&self) -> Vec { @@ -7264,7 +7843,7 @@ mod postgres_fleet_construction_tests { let (key, pin) = store .put(Some(content.len() as u64), Cursor::new(&content[..])) .await - .expect("the put publishes through the in-process protocol"); + .expect("the fleet put publishes through the joint tx"); assert_eq!(key, key_for(content)); assert!(store.has(&key.as_bytes()).await.unwrap()); drop(pin); @@ -7624,3 +8203,1186 @@ mod postgres_pglo_fleet_construction_tests { ); } } + +/// The wave-6 fleet put path's store-core tests (feature `postgres`) +/// for the kv tier and the local shared-media tier: the joint +/// entry+pin tx driver exercised through the public `put` / +/// `put_in_batch` — the kv tier's joint form (entry + pin row in one +/// step, seen by an independent connection), the A1 termination +/// discipline's forced kv failure form, the pin guard's TTL/3 renewal +/// duty (cadence, drop/conversion/batch stop, the never-resurrect +/// form), the dedup re-put's row-alone upsert-renewal and the by-row +/// protection composition, the shared-media pin-first / +/// publish-second ordering, and the `fleet: false` non-event (no pin +/// rows at all). Env-driven (`ALKBLOBS_PG_*`), skipping cleanly +/// server-less; every test takes a fresh engine-owned schema on the +/// harness database (no LOs involved here — the LO-bearing shapes' own +/// module is `postgres_pglo_fleet_put_tests`) with a janitor removing +/// it in every exit path. +#[cfg(all(test, feature = "postgres"))] +mod postgres_fleet_put_tests { + use super::*; + use crate::pg_common::qualified_table; + use crate::postgres::test_support::SchemaJanitor; + use std::io::Cursor; + use std::sync::atomic::{AtomicU64, Ordering}; + use tokio_postgres::NoTls; + + static SCHEMA_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" + ); + } + + struct TestSite { + dsn: String, + db: String, + schema: String, + _janitor: SchemaJanitor, + } + + async fn fresh_site(tag: &str) -> Option { + let dsn = harness_dsn()?; + let schema = format!( + "abtfput_{}_{}_{:x}", + tag, + std::process::id(), + SCHEMA_SEQ.fetch_add(1, Ordering::Relaxed) + ); + Some(TestSite { + dsn: dsn.clone(), + db: env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()), + schema: schema.clone(), + _janitor: SchemaJanitor::new(&dsn, &schema), + }) + } + + /// A fresh temp root per test (the local tier's medium); Drop + /// removes the whole 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-fput-{tag}-{}-{:x}-{seq}", + std::process::id(), + seq, + )); + std::fs::create_dir_all(&dir).expect("temp root created"); + TempRoot(dir) + } + + impl Drop for TempRoot { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + /// A short test pin TTL: the TTL/3 cadence observable within a + /// test's runtime (≈667 ms renewals, ≈2 s expiry horizon). + const TEST_TTL: Duration = Duration::from_secs(2); + /// One renewal cadence plus margin: any interval this long admits + /// at least one TTL/3 tick (TEST_TTL / 3 ≈ 667 ms). + const ONE_CADENCE: Duration = Duration::from_millis(900); + /// More than a cadence beyond a stopped renewal: the frozen row + /// proves the task ended (no leaked renewals). + const BEYOND_CADENCE: Duration = Duration::from_millis(1200); + + fn key_for(content: &[u8]) -> Key { + Key::derive(content, content.len() as u64) + } + + /// The independent connection — the "sees both or neither" + /// acceptance's instrument: every probe below runs outside the + /// store's pool, over its own connection, seeing only durable + /// state. + struct Independent { + client: tokio_postgres::Client, + _conn: tokio::task::JoinHandle<()>, + } + + async fn independent(dsn: &str) -> Independent { + let (client, conn) = tokio_postgres::connect(dsn, NoTls) + .await + .expect("independent connect"); + Independent { + client, + _conn: tokio::spawn(async move { + let _ = conn.await; + }), + } + } + + impl Independent { + /// The gc_pins rows in their stored encoding, in the + /// `(key, owner)` sort order — the pool's durable protection + /// set as a second process sees it. + async fn pin_rows(&self, schema: &str) -> Vec<(Vec, Vec, i64)> { + let sql = format!( + "SELECT key, owner, expiry FROM {} ORDER BY key, owner", + qualified_table(schema, "gc_pins") + ); + let rows = self + .client + .query(sql.as_str(), &[]) + .await + .expect("pin rows"); + rows.iter() + .map(|row| { + ( + row.get::<_, Vec>(0), + row.get::<_, Vec>(1), + row.get::<_, i64>(2), + ) + }) + .collect() + } + + /// The kv engine's entry rows `(key, size)`. + async fn kv_rows(&self, schema: &str) -> Vec<(Vec, i64)> { + let sql = format!( + "SELECT key, size FROM {} ORDER BY key", + qualified_table(schema, "blobs") + ); + let rows = self.client.query(sql.as_str(), &[]).await.expect("kv rows"); + rows.iter() + .map(|row| (row.get::<_, Vec>(0), row.get::<_, i64>(1))) + .collect() + } + + /// The pooled sessions sitting in a not-yet-committed + /// transaction — the dirty-session observable the A1 + /// termination design's test asserts to zero (deadpool's + /// `Fast` recycle never resets transaction state, PG-2). + async fn idle_in_transaction_sessions(&self, db: &str) -> i64 { + self.client + .query_one( + "SELECT count(*) FROM pg_stat_activity \ + WHERE datname = $1 AND state IS NOT NULL \ + AND state LIKE 'idle in transaction%'", + &[&db], + ) + .await + .expect("pg_stat_activity probe") + .get::<_, i64>(0) + } + } + + /// The store's pin rows for its own instance id — the filters + /// keeping multi-instance tests' assertions per-instance + /// (protection composes by row). + fn instance_rows(rows: &[(Vec, Vec, i64)], owner: &[u8]) -> Vec<(Vec, i64)> { + rows.iter() + .filter(|(_, row_owner, _)| row_owner == owner) + .map(|(key, _, expiry)| (key.clone(), *expiry)) + .collect() + } + + /// A store instance's sole row's expiry (the single-row shape's + /// tests); `None` when the row is not there. + fn expiry_of_row(rows: &[(Vec, Vec, i64)], owner: &[u8]) -> Option { + instance_rows(rows, owner) + .into_iter() + .next() + .map(|(_, expiry)| expiry) + } + + /// A particular `(key, owner)` row's expiry. + fn row_expiry(rows: &[(Vec, Vec, i64)], owner: &[u8], key: &Key) -> Option { + rows.iter() + .find(|(k, o, _)| k == &key.as_bytes() && o == owner) + .map(|(_, _, expiry)| *expiry) + } + + async fn fleet_kv_only_store(site: &TestSite, ttl: Duration) -> (Store, Vec) { + let store = Store::kv_only_postgres_fleet( + &site.dsn, + PgOpts { + schema: site.schema.clone(), + ..Default::default() + }, + 64, + DEFAULT_DELETE_WINDOW_DELAY, + ttl, + ) + .await + .expect("the fleet kv-only store constructs"); + let owner = store.fleet().expect("fleet state").owner().to_vec(); + (store, owner) + } + + async fn sleep_of(d: Duration) { + tokio::time::sleep(d).await; + } + + /// The kv tier's joint form (the re-routing client posture): the + /// put returns byte-exact, and an independent connection sees + /// **both** the entry row and its pin row — one committed step. + /// The pin row carries the store-instance id as its owner and an + /// expiry at `now + TTL`. + #[tokio::test] + async fn fleet_kv_put_publishes_entry_and_pin_row_together() { + let Some(site) = fresh_site("fput-kv-joint").await else { + skip_note(); + return; + }; + let (store, owner) = fleet_kv_only_store(&site, DEFAULT_PIN_TTL).await; + + let content = b"a sub-threshold blob under the fleet joint tx".to_vec(); + let (key, pin) = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("the fleet kv put publishes"); + assert_eq!(key, key_for(&content)); + + // The in-process refcount machinery continues to function + // (instance-local bookkeeping; unchanged shape). + assert_eq!(store.gc().pin_count(&key), 1); + drop(pin); + assert_eq!(store.gc().pin_count(&key), 0); + + let probe = independent(&site.dsn).await; + assert_eq!( + probe.kv_rows(&site.schema).await, + vec![(key.as_bytes(), content.len() as i64)], + "the entry row is durable" + ); + let now = SystemTime::now(); + let rows = instance_rows(&probe.pin_rows(&site.schema).await, &owner); + assert_eq!(rows.len(), 1, "exactly this instance's pin row"); + assert_eq!(rows[0].0, key.as_bytes()); + assert!( + rows[0].1 + > crate::pg_common::time_to_nanos(now + DEFAULT_PIN_TTL - Duration::from_secs(5)), + "expiry = now + TTL at the joint tx's commit" + ); + + let (len, blob) = store + .get(&key.as_bytes()) + .await + .unwrap() + .expect("the entry reads back"); + assert_eq!(len, content.len() as u64); + assert_eq!(blob.read_at(0, len).await.unwrap(), content); + } + + /// The kv joint tx's zero-residue failure form (A1's propagation + /// through the public `put`; the clause's own rollback is part of + /// the wave-4 contract): a forced `pin_commit` failure — publish + + /// protection landing in one step, so both miss together — leaves + /// the pool's sessions committed-transaction-free and the store's + /// tables empty. + #[tokio::test] + async fn fleet_kv_joint_commit_failure_leaves_neither_entry_nor_pin() { + let Some(site) = fresh_site("fput-kv-fail").await else { + skip_note(); + return; + }; + let (mut store, _owner) = fleet_kv_only_store(&site, DEFAULT_PIN_TTL).await; + match &mut store.kv { + KvEngine::Postgres(engine) => match Arc::get_mut(engine) { + Some(backend) => backend.inject_fail_joint_commit(), + None => panic!("the store uniquely holds its own engine"), + }, + #[cfg(feature = "mem")] + KvEngine::Mem(_) => panic!("a postgres store is required"), + #[cfg(feature = "sqlite")] + KvEngine::Sqlite(_) => panic!("a postgres store is required"), + } + + let content = b"content whose joint tx is forced to fail".to_vec(); + let err = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .unwrap_err(); + assert!( + err.to_string().contains("injected"), + "the engine's forced failure surfaces: {err}" + ); + + let probe = independent(&site.dsn).await; + assert_eq!( + probe.kv_rows(&site.schema).await, + vec![], + "no published entry" + ); + assert_eq!(probe.pin_rows(&site.schema).await, vec![], "no pin rows"); + assert_eq!( + probe.idle_in_transaction_sessions(&site.db).await, + 0, + "no session returned to the pool holding a transaction (A1)" + ); + assert_eq!( + store.list().await.unwrap(), + Vec::::new(), + "zero residue" + ); + } + + /// The pin guard's renewal duty (a short test TTL; the cadence + /// observed): while the guard lives, its row's expiry moves + /// forward every TTL/3; the guard's drop stops the renewal; the + /// ordered-step conversion stops it likewise — no leaked tasks, no + /// renewal past the guard's life. + #[tokio::test] + async fn fleet_pin_row_renews_until_the_guard_drops_or_converts() { + let Some(site) = fresh_site("fput-renewal").await else { + skip_note(); + return; + }; + let (store, owner) = fleet_kv_only_store(&site, TEST_TTL).await; + let probe = independent(&site.dsn).await; + + // While the guard lives, the cadence moves the expiry; the + // drop stops the duty and the row freezes. + let content = b"a guard whose row renews while it lives".to_vec(); + let (_, pin) = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("the fleet kv put publishes"); + let initial = expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row"); + sleep_of(ONE_CADENCE).await; + let renewed = expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row"); + assert!( + renewed > initial, + "the guard's TTL/3 duty moved the row's expiry forward" + ); + drop(pin); + let at_drop = expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row"); + assert_eq!( + renewed, at_drop, + "no renewal between the last tick and the drop" + ); + sleep_of(BEYOND_CADENCE).await; + assert_eq!( + at_drop, + expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row"), + "the dropped guard's renewal is stopped — no leaked tasks, \ + no renewal past the guard's life" + ); + + // The conversion boundary: a converted guard stops its duty — + // the ordered-step conversion's pre-stop (RegistryCore:: + // convert's ManuallyDrop would otherwise suppress the guard's + // own stop). + struct SetSource { + live: parking_lot::Mutex>, + } + impl LivenessSource for SetSource { + fn live_keys(&self) -> Vec { + self.live.lock().iter().cloned().collect() + } + fn contains(&self, key: &Key) -> bool { + self.live.lock().contains(key) + } + } + let source = Arc::new(SetSource { + live: parking_lot::Mutex::new(BTreeSet::new()), + }); + let content_two = b"a guard converted into a registered source".to_vec(); + let (second_key, guard) = store + .put( + Some(content_two.len() as u64), + Cursor::new(&content_two[..]), + ) + .await + .expect("the second put publishes"); + let at_mint = row_expiry(&probe.pin_rows(&site.schema).await, &owner, &second_key) + .expect("the second row"); + let registered = second_key.clone(); + let source_clone = Arc::clone(&source); + store.gc().convert(guard, move || { + source_clone.live.lock().insert(registered); + }); + assert_eq!( + store.gc().pin_count(&second_key), + 0, + "the conversion released the refcount in the ordered step" + ); + sleep_of(BEYOND_CADENCE).await; + assert_eq!( + at_mint, + row_expiry(&probe.pin_rows(&site.schema).await, &owner, &second_key).expect("the row"), + "the converted guard's duty is stopped with its life" + ); + } + + /// The never-resurrect form: an expired row is never renewed back + /// to life — the guard's renewal observes `pin_renew`'s `false` + /// and the duty ends (the delete-then-recover backstop's + /// territory, ADR-008's expiry semantics verbatim). + #[tokio::test] + async fn fleet_pin_renewal_never_resurrects_an_expired_row() { + let Some(site) = fresh_site("fput-expired").await else { + skip_note(); + return; + }; + let (store, owner) = fleet_kv_only_store(&site, TEST_TTL).await; + let content = b"a guard whose row a manual expiry abandons".to_vec(); + let (key, _pin) = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("the fleet kv put publishes"); + + // Expire the row out from under the guard: the abandonment the + // sweeper's maintenance step would reap. + let expired = crate::pg_common::time_to_nanos(SystemTime::now() - Duration::from_secs(1)); + let probe = independent(&site.dsn).await; + probe + .client + .execute( + &format!( + "UPDATE {} SET expiry = $1 WHERE key = $2 AND owner = $3", + qualified_table(site.schema.as_str(), "gc_pins") + ), + &[&expired, &key.as_bytes(), &owner], + ) + .await + .expect("the manual expiry"); + + sleep_of(ONE_CADENCE).await; + let rows = probe.pin_rows(&site.schema).await; + let row = rows + .iter() + .find(|(k, o, _)| k == &key.as_bytes() && o == &owner) + .expect("the row still holds its slot"); + assert_eq!( + row.2, expired, + "the expired row was never renewed back to life" + ); + } + + /// The batch form: each member's joint tx commits its row; while + /// the batch lives the members' rows renew; the batch's one-step + /// release ends the members' renewals (the rows expire and the + /// sweeper's maintenance step reaps them — `store-fleet-sweep`'s). + #[tokio::test] + async fn fleet_batch_release_ends_the_members_renewals() { + let Some(site) = fresh_site("fput-batch").await else { + skip_note(); + return; + }; + let (store, owner) = fleet_kv_only_store(&site, TEST_TTL).await; + let first = b"batch member one".to_vec(); + let second = b"batch member two".to_vec(); + + let mut batch = store.batch(); + let k1 = store + .put_in_batch( + &mut batch, + Some(first.len() as u64), + Cursor::new(&first[..]), + ) + .await + .expect("member one publishes"); + let k2 = store + .put_in_batch( + &mut batch, + Some(second.len() as u64), + Cursor::new(&second[..]), + ) + .await + .expect("member two publishes"); + + let probe = independent(&site.dsn).await; + let initial = instance_rows(&probe.pin_rows(&site.schema).await, &owner); + assert_eq!(initial.len(), 2, "each member's joint tx committed its row"); + + sleep_of(ONE_CADENCE).await; + let renewed = instance_rows(&probe.pin_rows(&site.schema).await, &owner); + assert!( + renewed + .iter() + .zip(initial.iter()) + .all(|((_, r), (_, i))| r > i), + "the members' rows renew while the batch lives" + ); + + batch.commit(); + assert_eq!(store.gc().pin_count(&k1), 0, "the batch's one-step release"); + assert_eq!(store.gc().pin_count(&k2), 0); + let at_release = instance_rows(&probe.pin_rows(&site.schema).await, &owner); + sleep_of(BEYOND_CADENCE).await; + let frozen = instance_rows(&probe.pin_rows(&site.schema).await, &owner); + assert_eq!( + at_release, frozen, + "the batch's release ended the members' renewals" + ); + } + + /// The dedup re-put under fleet: CAS dedup means no new entry — + /// the same store's re-put commits the pin row alone (the joint + /// tx's row upsert renewing `(key, owner)`, the wave-4 recorded + /// shape), and a different instance's re-put is a different pin + /// row — protection composes by row (ADR-010 §2). + #[tokio::test] + async fn fleet_dedup_reput_commits_the_pin_row_alone_and_composes_by_row() { + let Some(site) = fresh_site("fput-dedup").await else { + skip_note(); + return; + }; + let (store, owner) = fleet_kv_only_store(&site, TEST_TTL).await; + let content = b"put twice; addressed once".to_vec(); + + let (key, pin_one) = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("the first put publishes"); + let probe = independent(&site.dsn).await; + let first_rows = instance_rows(&probe.pin_rows(&site.schema).await, &owner); + assert_eq!(probe.kv_rows(&site.schema).await.len(), 1, "one entry row"); + drop(pin_one); + + let (key_again, pin_two) = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("the dedup re-put publishes"); + assert_eq!(key_again, key, "CAS dedup under the same canonical key"); + assert_eq!( + probe.kv_rows(&site.schema).await.len(), + 1, + "the dedup re-put committed the pin row alone — no new entry row" + ); + let renewed_rows = instance_rows(&probe.pin_rows(&site.schema).await, &owner); + assert_eq!( + renewed_rows.len(), + 1, + "the same store's row, upsert-renewed" + ); + assert!( + renewed_rows[0].1 >= first_rows[0].1, + "the row upsert-renews (key, owner): expiry moves forward" + ); + + // A different instance's re-put: a different pin row. + let other_store = Store::kv_only_postgres_fleet( + &site.dsn, + PgOpts { + schema: site.schema.clone(), + ..Default::default() + }, + 64, + DEFAULT_DELETE_WINDOW_DELAY, + TEST_TTL, + ) + .await + .expect("the second instance constructs"); + let other_owner = other_store.fleet().expect("fleet state").owner().to_vec(); + assert_ne!(owner, other_owner, "two instances, two store-instance ids"); + other_store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("the foreign instance's re-put publishes"); + assert_eq!( + probe.kv_rows(&site.schema).await.len(), + 1, + "still one entry row across instances" + ); + let rows = probe.pin_rows(&site.schema).await; + assert_eq!( + rows.len(), + 2, + "a different instance's re-put is a different pin row — \ + protection composes by row" + ); + assert!( + row_expiry(&rows, &owner, &key).is_some() + && row_expiry(&rows, &other_owner, &key).is_some(), + "both instances' rows present, keyed by their own owner" + ); + let _ = pin_two; + } + + /// The shared-media ordering (ADR-012 §4): pin row first, + /// publish-second. The success form lands both; the forced publish + /// failure is the ordering's proof — the pin row **existed and was + /// durable before the committed file came to be**, and the failed + /// put leaves no committed entry, no stage file, and no in-process + /// refcount behind (the orphan pin row the expiry machinery reaps; + /// the harmless crash window). + #[tokio::test] + async fn fleet_local_shared_media_put_is_pin_first_publish_second() { + let Some(site) = fresh_site("fput-local").await else { + skip_note(); + return; + }; + let (store, owner) = { + let store = Store::dual_tier_postgres_local_fleet( + &site.dsn, + PgOpts { + schema: site.schema.clone(), + ..Default::default() + }, + { + let root = temp_root("fput-local-root"); + root.0.clone() + }, + 64, + DEFAULT_DELETE_WINDOW_DELAY, + DEFAULT_PIN_TTL, + FleetSharedMedia::asserted(), + ) + .await + .expect("the fleet shared-media store constructs"); + let owner = store.fleet().expect("fleet state").owner().to_vec(); + (store, owner) + }; + + // Success form: the pin row is durable, then the commit-rename + // publishes — after the put, both are there, byte-exact. + let content = vec![7u8; 200]; + let (key, pin) = store + .put(Some(200), Cursor::new(&content[..])) + .await + .expect("the shared-media put publishes"); + let probe = independent(&site.dsn).await; + let rows = instance_rows(&probe.pin_rows(&site.schema).await, &owner); + assert_eq!(rows.len(), 1, "the pin row is durable"); + assert_eq!(rows[0].0, key.as_bytes()); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .unwrap() + .expect("the entry is published"); + assert_eq!(len, 200); + assert_eq!(blob.read_at(0, len).await.unwrap(), content); + drop(pin); + + // The forced publish failure: pin-first means the row landed + // before the publish could. + let (mut store2, owner2) = { + let store = Store::dual_tier_postgres_local_fleet( + &site.dsn, + PgOpts { + schema: site.schema.clone(), + ..Default::default() + }, + { + let root = temp_root("fput-local-fail"); + root.0.clone() + }, + 64, + DEFAULT_DELETE_WINDOW_DELAY, + DEFAULT_PIN_TTL, + FleetSharedMedia::asserted(), + ) + .await + .expect("second instance"); + let owner = store.fleet().expect("fleet state").owner().to_vec(); + (store, owner) + }; + match &mut store2.large { + Some(LargeEngine::Local(engine)) => engine.inject_fail_commit(), + _ => panic!("a local large tier is required"), + } + let failed = b"content whose commit-rename is forced to fail, over the threshold".to_vec(); + assert!( + failed.len() > 64, + "the failed put must route to the large tier" + ); + let err = store2 + .put(Some(failed.len() as u64), Cursor::new(&failed[..])) + .await + .unwrap_err(); + assert!( + err.to_string().contains("injected"), + "the engine's forced publish failure surfaces: {err}" + ); + let key = key_for(&failed); + let probe = independent(&site.dsn).await; + let orphan = instance_rows(&probe.pin_rows(&site.schema).await, &owner2); + assert_eq!( + orphan.len(), + 1, + "pin-first: the pin row committed before the failed publish \ + (the orphan the expiry machinery reaps)" + ); + assert_eq!(orphan[0].0, key.as_bytes()); + assert!( + !store2.has(&key.as_bytes()).await.unwrap(), + "no committed file exists — the publish never happened" + ); + assert_eq!( + store2.gc().pin_count(&key), + 0, + "the failed put's in-process refcount released with the guard" + ); + } + + /// Under `fleet: false` the put paths are byte-unchanged: the + /// ADR-005 in-process protocol verbatim — no pin rows are ever + /// committed (the pin machinery is the in-process guard only), and + /// the puts/gets behave exactly as before this task. + #[tokio::test] + async fn non_fleet_put_commits_no_pin_rows() { + let Some(site) = fresh_site("fput-nonfleet").await else { + skip_note(); + return; + }; + let store = Store::dual_tier_postgres_local( + &site.dsn, + PgOpts { + schema: site.schema.clone(), + ..Default::default() + }, + { + let root = temp_root("fput-nonfleet-root"); + root.0.clone() + }, + 64, + DEFAULT_DELETE_WINDOW_DELAY, + ) + .await + .expect("the non-fleet store constructs"); + assert!(store.fleet().is_none(), "implicitly fleet: false"); + + let small = b"a non-fleet kv put".to_vec(); + let (k_small, pin_one) = store + .put(Some(small.len() as u64), Cursor::new(&small[..])) + .await + .expect("the put publishes"); + let big = vec![5u8; 300]; + let (k_big, pin_two) = store + .put(Some(300), Cursor::new(&big[..])) + .await + .expect("the staged put publishes, unchanged"); + + for (key, want) in [(&k_small, &small[..]), (&k_big, &big[..])] { + let (len, blob) = store + .get(&key.as_bytes()) + .await + .unwrap() + .expect("the entry reads back"); + assert_eq!(blob.read_at(0, len).await.unwrap(), want); + } + + let probe = independent(&site.dsn).await; + assert_eq!( + probe.pin_rows(&site.schema).await, + vec![], + "fleet: false pins nothing durable — the in-process guard only" + ); + drop(pin_one); + drop(pin_two); + } +} + +/// The wave-6 fleet put path's store-core tests for the consolidation +/// posture (features `postgres` + `pg-lo`): the A1 termination +/// discipline's mid-driver failure form through the public `put` (the +/// contribution returns with the held tx; the driver terminates — zero +/// residue, no dirty pooled session), and the one-tx round-trips +/// (both tiers, known- and unknown-length, byte-exact; the pool's pin +/// rows exactly one per member). Every test takes a fresh dedicated +/// database (large objects are database-level — the C7 lesson) with a +/// janitor removing it in every exit path. Env-driven +/// (`ALKBLOBS_PG_*`), skipping cleanly server-less. +#[cfg(all(test, feature = "postgres", feature = "pg-lo"))] +mod postgres_pglo_fleet_put_tests { + use super::*; + use crate::backend::Backend; + use crate::pg_common::{qualified_table, test_support::DatabaseJanitor, DEFAULT_SCHEMA}; + use std::io::Cursor; + 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, + db: 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!( + "abtfputpglo_{}_{}_{: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), + db: db.clone(), + _janitor: DatabaseJanitor::new(&admin, &db), + }) + } + + /// A short test pin TTL: the TTL/3 cadence observable within a + /// test's runtime (≈667 ms renewals, ≈2 s expiry horizon). + const TEST_TTL: Duration = Duration::from_secs(2); + + fn key_for(content: &[u8]) -> Key { + Key::derive(content, content.len() as u64) + } + + /// C2's catalog math: LO pages are 2048-byte (BLCKSZ/4). + fn pages_for(len: usize) -> i64 { + (len as i64 + 2047) / 2048 + } + + async fn sleep_of(d: Duration) { + tokio::time::sleep(d).await; + } + + async fn fleet_pglo_store( + site: &TestDatabase, + threshold: u64, + ttl: Duration, + ) -> (Store, Vec) { + let store = Store::dual_tier_postgres_pglo_fleet( + &site.dsn, + PgOpts::default(), + &site.dsn, + PgOpts::default(), + threshold, + DEFAULT_DELETE_WINDOW_DELAY, + ttl, + ) + .await + .expect("the fleet consolidation store constructs"); + let owner = store.fleet().expect("fleet state").owner().to_vec(); + (store, owner) + } + + fn pglo_engine(store: &Store) -> &PgLoBackend { + match store.large() { + Some(LargeEngine::PgLo(engine)) => engine, + Some(LargeEngine::Local(_)) | None => panic!("these tests construct pg-lo stores"), + } + } + + /// The independent connection — every probe runs outside the + /// store's pool, over its own connection, seeing only durable + /// state. + struct Independent { + client: tokio_postgres::Client, + _conn: tokio::task::JoinHandle<()>, + } + + async fn independent(dsn: &str) -> Independent { + let (client, conn) = tokio_postgres::connect(dsn, NoTls) + .await + .expect("independent connect"); + Independent { + client, + _conn: tokio::spawn(async move { + let _ = conn.await; + }), + } + } + + impl Independent { + /// The gc_pins rows in their stored encoding, in the + /// `(key, owner)` sort order. + async fn pin_rows(&self) -> Vec<(Vec, Vec, i64)> { + let sql = format!( + "SELECT key, owner, expiry FROM {} ORDER BY key, owner", + qualified_table(DEFAULT_SCHEMA, "gc_pins") + ); + let rows = self + .client + .query(sql.as_str(), &[]) + .await + .expect("pin rows"); + rows.iter() + .map(|row| { + ( + row.get::<_, Vec>(0), + row.get::<_, Vec>(1), + row.get::<_, i64>(2), + ) + }) + .collect() + } + + /// The kv engine's entry rows `(key, size)`. + async fn kv_rows(&self) -> Vec<(Vec, i64)> { + let sql = format!( + "SELECT key, size FROM {} ORDER BY key", + qualified_table(DEFAULT_SCHEMA, "blobs") + ); + let rows = self.client.query(sql.as_str(), &[]).await.expect("kv rows"); + rows.iter() + .map(|row| (row.get::<_, Vec>(0), row.get::<_, i64>(1))) + .collect() + } + + /// The LO catalog's page count (LOs are database-level, C7 — + /// the pages every failure form's zero-residue assertion + /// counts). + async fn lo_pages(&self) -> i64 { + self.client + .query_one("SELECT count(*) FROM pg_largeobject", &[]) + .await + .expect("LO catalog probe") + .get::<_, i64>(0) + } + + /// The pooled sessions sitting in a not-yet-committed + /// transaction — the dirty-session observable the A1 + /// termination design's test asserts to zero (deadpool's + /// `Fast` recycle never resets transaction state, PG-2). + async fn idle_in_transaction_sessions(&self, db: &str) -> i64 { + self.client + .query_one( + "SELECT count(*) FROM pg_stat_activity \ + WHERE datname = $1 AND state IS NOT NULL \ + AND state LIKE 'idle in transaction%'", + &[&db], + ) + .await + .expect("pg_stat_activity probe") + .get::<_, i64>(0) + } + } + + /// The store's pin rows for its own instance id — the filter + /// keeping multi-instance assertions per-instance. + fn instance_rows(rows: &[(Vec, Vec, i64)], owner: &[u8]) -> Vec<(Vec, i64)> { + rows.iter() + .filter(|(_, row_owner, _)| row_owner == owner) + .map(|(key, _, expiry)| (key.clone(), *expiry)) + .collect() + } + + /// The A1 termination discipline's mid-driver failure form through + /// the public `put` (the consolidation posture): the tier's + /// injected contribution failure — LO pages already staged into + /// the held tx — is caught by the driver, which terminates the + /// tx (the contribution never does) before propagating: zero + /// residue everywhere (no kv row, no pin row, no companion row, no + /// LO pages) and no dirty session returned to the pool (deadpool's + /// `Fast` recycle never resets transaction state, PG-2; a follow-up + /// put on the pool round-trips clean). + #[tokio::test] + async fn fleet_pglo_mid_driver_failure_terminates_with_zero_residue() { + let Some(site) = fresh_db("fput-pglo-fail").await else { + skip_note(); + return; + }; + let (mut store, _owner) = fleet_pglo_store(&site, 64, DEFAULT_PIN_TTL).await; + match &mut store.large { + Some(LargeEngine::PgLo(engine)) => engine.inject_fail_commit(), + _ => panic!("the consolidation posture's large tier is required"), + } + + // ≥ two 512 KiB blocks flushed into the held tx before the + // injected publish failure — real staged LO state for the + // termination to take away. + let content = vec![9u8; 1_500_000]; + let err = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .unwrap_err(); + assert!( + err.to_string().contains("injected"), + "the contribution's forced failure surfaces: {err}" + ); + + let probe = independent(&site.dsn).await; + assert_eq!(probe.kv_rows().await, vec![], "no kv row"); + assert_eq!(probe.pin_rows().await, vec![], "no pin row"); + assert_eq!( + probe.lo_pages().await, + 0, + "the staged LO's pages died with the driver's termination" + ); + assert_eq!( + probe.idle_in_transaction_sessions(&site.db).await, + 0, + "no dirty pooled session (A1's catch-propagate-terminate)" + ); + + // The session the pool hands back is workable: a follow-up + // put/get round-trips. + let fresh = b"the pool's session came back clean".to_vec(); + let (key, pin) = store + .put(Some(fresh.len() as u64), Cursor::new(&fresh[..])) + .await + .expect("the follow-up put publishes"); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .unwrap() + .expect("the entry reads back"); + assert_eq!(blob.read_at(0, len).await.unwrap(), fresh); + drop(pin); + } + + /// The consolidation posture's round-trip through the public + /// `put` (both tiers, one tx each): the sub-threshold puts ride + /// the kv tier's joint form (known- and unknown-length alike), + /// the over-threshold puts ride the pg-lo tier's joint form (the + /// staged LO + companion row + pin row + commit inside the + /// store-held tx), every entry reads back byte-exact, and the + /// pool's pin rows are exactly the four members' — same owner, + /// row-per-key. + #[tokio::test] + async fn fleet_pglo_consolidation_round_trips_through_the_public_put() { + let Some(site) = fresh_db("fput-pglo-rt").await else { + skip_note(); + return; + }; + let (store, owner) = fleet_pglo_store(&site, 64, TEST_TTL).await; + + let small = b"a sub-threshold consolidation put".to_vec(); + let (k_small, guard_one) = store + .put(Some(small.len() as u64), Cursor::new(&small[..])) + .await + .expect("the kv joint put publishes"); + let small_streamed = b"an unknown-length sub-threshold put".to_vec(); + let (k_kv_streamed, guard_two) = store + .put(None, Cursor::new(&small_streamed[..])) + .await + .expect("the unknown-length kv joint put publishes"); + + let big = vec![7u8; 1_500_000]; + let (k_big, guard_three) = store + .put(Some(big.len() as u64), Cursor::new(&big[..])) + .await + .expect("the large joint put publishes"); + let big_streamed = vec![11u8; 300_000]; + let (k_large_streamed, guard_four) = store + .put(None, Cursor::new(&big_streamed[..])) + .await + .expect("the unknown-length large joint put publishes"); + + for (key, want) in [ + (&k_small, &small[..]), + (&k_kv_streamed, &small_streamed[..]), + (&k_big, &big[..]), + (&k_large_streamed, &big_streamed[..]), + ] { + let (len, blob) = store + .get(&key.as_bytes()) + .await + .unwrap() + .expect("the entry reads back"); + assert_eq!( + blob.read_at(0, len).await.unwrap(), + want, + "byte-exact round-trip (the tier's one-tx form)" + ); + } + + let probe = independent(&site.dsn).await; + let kv_rows = probe.kv_rows().await; + assert_eq!( + kv_rows.len(), + 2, + "exactly the two kv-tier entries (the large tier's one-tx form \ + carries no kv row)" + ); + let rows = probe.pin_rows().await; + let mine = instance_rows(&rows, &owner); + assert_eq!( + mine.len(), + 4, + "exactly the four members' pin rows, one tx apiece" + ); + for key in [&k_small, &k_kv_streamed, &k_big, &k_large_streamed] { + assert!( + mine.iter().any(|(k, _)| k == &key.as_bytes()), + "each member's row carries its key" + ); + } + + let lo = pglo_engine(&store); + assert_eq!(lo.size(&k_big.as_bytes()).await.unwrap(), Some(1_500_000)); + assert_eq!( + lo.size(&k_large_streamed.as_bytes()).await.unwrap(), + Some(300_000), + "the unknown-length staged form's derivation resolved exactly" + ); + assert_eq!( + probe.lo_pages().await, + pages_for(1_500_000) + pages_for(300_000), + "exactly the two entries' LO pages" + ); + + // The renewal duty rides the durable rows: after a cadence the + // members' expiries move (the guards are still held by this + // test's drops-never bindings). + sleep_of(Duration::from_millis(1200)).await; + let moved = instance_rows(&probe.pin_rows().await, &owner); + for ((_, expiry), (_, initial)) in moved.iter().zip(mine.iter()) { + assert!( + expiry > initial, + "the guard's TTL/3 duty keeps the row moving forward" + ); + } + drop((guard_one, guard_two, guard_three, guard_four)); + } +} diff --git a/tasks/store-fleet-put.md b/tasks/store-fleet-put.md index c688b71..bdc2b28 100644 --- a/tasks/store-fleet-put.md +++ b/tasks/store-fleet-put.md @@ -1,7 +1,7 @@ --- id: store-fleet-put name: the fleet put path — the joint entry+pin tx driver and the pin guard's fleet form -status: pending +status: completed depends_on: [store-fleet-construction, engine-pg-lo-joint] scope: broad risk: high @@ -108,29 +108,29 @@ protocol verbatim; every wave-2/3/4/5 put test stays green). ## Acceptance Criteria -- [ ] Under `fleet: true`, kv-tier puts route through the joint tx +- [x] Under `fleet: true`, kv-tier puts route through the joint tx (entry + pin row in one step; an independent connection never observes one without the other); under `fleet: false` the put paths are byte-unchanged (all existing suites green) -- [ ] Amendment A1 landed: the driver terminates the tx on every path - from `begin_joint_tx` onward (the termination design recorded; - a forced mid-driver failure leaves zero residue and returns no +- [x] Amendment A1 landed: the driver terminates the tx on every path + from `begin_joint_tx` onward (the termination design recorded; a + forced mid-driver failure leaves zero residue and returns no dirty session to the pool) -- [ ] The pin guard's fleet form: renewal every TTL/3 while the guard +- [x] The pin guard's fleet form: renewal every TTL/3 while the guard lives; drop/conversion/batch-contribution stops the renewal with no leaked tasks; a failed renewal never resurrects an expired row (the lifecycle design points decided and recorded) -- [ ] The dedup re-put commits the pin row alone (CAS; upsert- +- [x] The dedup re-put commits the pin row alone (CAS; upsert- renewal); a different instance's re-put composes protection by row -- [ ] `large = local` fleet puts: pin-row-first, publish-second +- [x] `large = local` fleet puts: pin-row-first, publish-second (contract-tested — the pin row is durable before the committed file exists; the orphan-pin crash window is the expiry machinery's) -- [ ] The pg-lo consolidation put round-trips through the public +- [x] The pg-lo consolidation put round-trips through the public `put` (one tx over the shared instance; byte-exact large-entry form) -- [ ] `cargo test` (default) green; `cargo test --features postgres` +- [x] `cargo test` (default) green; `cargo test --features postgres` and `--features "postgres,pg-lo"` green server-less (skipping) and against the harness server at acceptance; `cargo clippy --all-targets -- -D warnings` and @@ -164,8 +164,131 @@ protocol verbatim; every wave-2/3/4/5 put test stays green). ## Notes -> To be filled by implementation agent +Landed 2026-10-12. Design decisions (each recorded in-code as a named +design point on the fleet impl block of `src/store.rs`; all keyed off +the landed ADR text): + +- **A1's termination design — catch-propagate-terminate, no abort + verb.** The driver's steps between `begin_joint_tx` and `pin_commit` + run inside one catch block (an async block's `?` surface); any error + terminates the held tx (a `ROLLBACK` through the + `PgJointTx::client()` checkout — the wave-5 landed accessor) before + propagating; `pin_commit` remains the terminal-consume step (its own + failure paths roll back per the wave-4 clause contract). An explicit + abort verb was rejected per A1's recorded options (it would need a + new ADR; the seam shape is pinned). A ROLLBACK on a session with no + active transaction (the clause bodies' having already rolled back) + is a completed, warning-level no-op — the blanket termination is + always safe. `pin_commit`-consuming txs on failure are the clause's + own; the driver never re-uses a consumed tx. +- **The renewal duty's mechanism — a per-guard tokio task on a TTL/3 + interval**, holding an `Arc` (the kv-tier enum's + postgres payload moved to an `Arc` for this: the detached task + outlives the `Store` and keeps the backend's pool exactly as long as + the pin lives; `PostgresBackend` itself is not `Clone` — its lease + checkout must not be). The shared-renewal-sweep keyed off a guard + registry was the rejected alternative: it needs precisely the + lifecycle notifications (mint/drop/convert/batch-contribute) that an + abort handle gives for free. The task is spawned **at the pin row's + durability** (after `pin_commit`, never at guard mint — a renewal + arriving at a nonexistent row would read the not-yet-pinned window), + on the cadence `(ttl / 3).max(1 ms)` (the 1 ms floor is the + degenerate-tuning guard: a TTL below 3 ms would floor to a zero + poll). The store resolves the expiry (`now + ttl`) at the joint tx's + `pin_commit`; successive renewals ride the engine's `pin_renew` + (the successor-time body reads the TTL the fleet constructor passed + the engine — `store-fleet-construction`'s wiring). +- **The failure posture: silent-and-continued on `Err`, ended on + `Ok(false)`.** The row's expiry is the protection — a missed renewal + degrades to ADR-008's expiry semantics, never an error on the put + path (the crate carries no logging dependency; embedder visibility + into renewals observes the row). `Ok(false)` — an expired or absent + `(key, owner)` row — ends the duty: the engine never resurrects + (wave 4's recorded semantics), so continuing would only churn; the + duty ends, the delete-then-recover backstop's territory begins. +- **The abort-on-drop discipline:** the duty is a + `PinRenewal` (a `JoinHandle` wrapper, crate-internal, gated feature + `postgres`) attached to the returned `Pin`; it is aborted by the + guard's drop, by `Batch`'s one-step release (the duty moves into the + batch at `Batch::add` — a member's row keeps renewing while the + batch lives), and by the ordered-step conversion + (`RegistryCore::convert` pre-stops before its `ManuallyDrop` + suppresses the guard's drop). No leaked tasks; no renewal past the + guard's life. A task aborted mid-renewal is harmless (the clause's + expiry update is idempotent). +- **The routing never re-derives from the engine set:** every fleet + branch keys off `Store::fleet`'s presence first; the pg-lo tier's + joint-vs-staged choice inside fleet mode rides the engine enum (the + tier mechanics' selector, the same dispatch as every other fleet + branch shape). The public `put`/`put_in_batch` signatures are + unchanged; the non-fleet paths are byte-unchanged (all existing + suites green untouched). +- **The kv joint form's mechanics:** the in-process refcount mints + first (unchanged bookkeeping), then `begin_joint_tx` → `joint_put` + (the engine's CAS row write; a dedup re-put contributes nothing — + the recorded CAS form; the row upsert-renews via `pin_commit`'s + `ON CONFLICT (key, owner)` upsert) → `pin_commit(tx, [row])` — + publish + protection in one step. The row's `owner` is the + store-instance id; `expiry = now + pin_ttl`. +- **The consolidated (pg-lo) form:** the tx begins, the tier's joint + staging cursor streams the content inside it (the engine-level + contract: the 512 KiB knee rides, the unknown-length form's + buffer-to-threshold phase is put_unknown's verbatim shape and its + derivation pass re-reads the artifact through the cursor's + read-your-own-writes seam), the cursor's publish is the CAS + contribution (a re-put: the empty-contribution form — the staged LO + unlinked in-tx, zero residue), and `pin_commit` lands the row with + the entry. +- **Test isolation:** every store test takes a fresh engine-owned + schema (the LO-free kv tier; `SchemaJanitor`) or a fresh dedicated + database (the LO-bearing consolidation tests; `DatabaseJanitor`) + with a janitor removing it in every exit path — amendment A2's + lesson honored from the start. The A1 dirty-session observable is + `pg_stat_activity`'s `idle in transaction%` count asserted to zero + (plus the follow-up put's clean round-trip and the zero-residue + table assertions). ## Summary -> To be filled on completion \ No newline at end of file +The fleet put path landed. Under `fleet: true`, `put`/`put_in_batch` +(unchanged signatures) route: **kv tier** — the joint entry+pin tx +(`begin_joint_tx` → the engine's CAS `joint_put` → `pin_commit`, +publish + protection in one step; a dedup re-put commits the pin row +alone, the row upsert-renewing `(key, owner)`; a different instance's +re-put mints a different row — protection composes by row); **large = +pg-lo (the consolidation posture)** — one tx over the shared instance, +the tier's joint staging cursor streaming the content inside the +store-held tx (known- and unknown-length; the 512 KiB knee, the CAS +publish, the empty-contribution re-put form, read-your-own-writes +derivation passes), `pin_commit` closing; **large = local (declared +shared media)** — pin-first, publish-second: a row-only joint tx +commits the pin row before the staged commit-rename (the +forced-publish-failure test proving the ordering: the pin row is +durable with no committed file, the orphan reclaimed by the expiry +machinery; never the reverse ordering). Amendment A1 landed as its +recorded owner — catch-propagate-terminate: every error path between +`begin_joint_tx` and `pin_commit` terminates the held tx before +propagating (tested: a mid-driver contribution failure leaves zero +residue and returns no dirty pooled session — `pg_stat_activity`'s +idle-in-transaction count zero — and the pool's follow-up put +round-trips clean). The pin guard's fleet form landed: the returned +guard carries the row's TTL/3 renewal duty (a per-guard tokio task, +spawned at the row's durability, silent on `Err`, ended by the +never-resurrect `Ok(false)`, aborted at every guard-life boundary — +drop, batch contribution, conversion). The in-process refcount +machinery continues to function (instance-local bookkeeping, the +batch's one-step release, the ordered-step conversion); the row is +the cross-instance protection; nothing expires in non-fleet modes. +Non-fleet put paths are byte-unchanged (all existing suites green +untouched; a non-fleet test additionally pins that no pin rows are +ever committed). 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 — +live, all fleet-put tests green); `--all-features` and the pg-lo-pure +`--features mem,pg-lo` builds green; clippy `-D warnings` green +(default, mem, postgres, all-features, pg-pure, mem+pg-lo, and +postgres+pg-lo); `cargo fmt --check` green; `cargo doc --no-deps` +zero warnings in the default build (the gated builds' pre-existing +warnings counted before and after — identical). The fleet sweep +(`store-fleet-sweep`) holds the fleet mechanism's remaining +store-core scope (the sweep lease, the SQL arbitration). \ No newline at end of file