feat(store): the fleet put path — the joint entry+pin tx driver and the pin guard's fleet form (wave 6)

- 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<PostgresBackend> 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.
This commit is contained in:
glm-5.3-flash committed 2026-10-12 16:08:29 +00:00
1 parent 76be7b0273
commit 950af9ab0d
8 files changed
+2080 -52

No files matched your search

+1 -1
View File
@@ -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"
+108 -7
View File
@@ -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<RegistryCore>,
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<PinRenewal>,
}
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<PinRenewal> {
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<RegistryCore>,
keys: Vec<Key>,
/// 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<PinRenewal>,
}
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));
}
}
+13 -5
View File
@@ -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<F: FnOnce()>(&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();
+11
View File
@@ -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 {
+11
View File
@@ -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
+16 -4
View File
@@ -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<u64, Error> {
+1785 -23
View File
File diff suppressed because it is too large. Load diff
+135 -12
View File
@@ -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<PostgresBackend>` (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
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).