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:
1 parent
76be7b0273
commit
950af9ab0d
8 files changed
+2080
-52
No files matched your search
+1
-1
@@ -26,7 +26,7 @@ pg-lo = ["dep:tokio-postgres", "dep:deadpool-postgres"]
|
|||||||
ops = ["dep:alkcall"]
|
ops = ["dep:alkcall"]
|
||||||
|
|
||||||
[dependencies]
|
[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"
|
thiserror = "2"
|
||||||
parking_lot = "0.12"
|
parking_lot = "0.12"
|
||||||
|
|
||||||
|
|||||||
+108
-7
@@ -147,10 +147,18 @@ pub struct EntryMeta {
|
|||||||
/// re-put acquires no new pool entry but refcounts the existing one
|
/// re-put acquires no new pool entry but refcounts the existing one
|
||||||
/// under the same arbitration lock (ADR-005).
|
/// under the same arbitration lock (ADR-005).
|
||||||
/// - Fleet correspondence (ADR-010 §4, ADR-012 §2): in fleet mode the
|
/// - Fleet correspondence (ADR-010 §4, ADR-012 §2): in fleet mode the
|
||||||
/// joint tx commits the expiry-carrying pin row and this guard drops
|
/// joint tx commits the expiry-carrying pin row alongside the entry,
|
||||||
/// at response time (the row's expiry is then the protection); in
|
/// and **the guard carries the row's renewal duty while it lives** —
|
||||||
/// non-fleet modes the ops layer's token is this guard's in-process
|
/// the store's TTL/3 schedule (ADR-008 §2; the landed
|
||||||
/// handle — nothing expires outside fleet mode.
|
/// `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 {
|
pub struct Pin {
|
||||||
registry: Arc<RegistryCore>,
|
registry: Arc<RegistryCore>,
|
||||||
key: Key,
|
key: Key,
|
||||||
@@ -158,6 +166,15 @@ pub struct Pin {
|
|||||||
/// release duty moved to the batch's one-step drop) or consumed by
|
/// release duty moved to the batch's one-step drop) or consumed by
|
||||||
/// the ordered-step conversion.
|
/// the ordered-step conversion.
|
||||||
armed: bool,
|
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 {
|
impl Pin {
|
||||||
@@ -174,6 +191,34 @@ impl Pin {
|
|||||||
registry,
|
registry,
|
||||||
key,
|
key,
|
||||||
armed: true,
|
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 {
|
if self.armed {
|
||||||
self.registry.release_pin(&self.key);
|
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 {
|
pub struct Batch {
|
||||||
registry: Arc<RegistryCore>,
|
registry: Arc<RegistryCore>,
|
||||||
keys: Vec<Key>,
|
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 {
|
impl Batch {
|
||||||
@@ -225,15 +312,22 @@ impl Batch {
|
|||||||
Batch {
|
Batch {
|
||||||
registry,
|
registry,
|
||||||
keys: Vec::new(),
|
keys: Vec::new(),
|
||||||
|
#[cfg(feature = "postgres")]
|
||||||
|
renewals: Vec::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Contribute a put's pin to the group (store-core call). The
|
/// Contribute a put's pin to the group (store-core call). The
|
||||||
/// contributed guard is defused — its own drop releases nothing;
|
/// contributed guard is defused — its own drop releases nothing;
|
||||||
/// the batch's one-step release owns the protection from here.
|
/// the batch's one-step release owns the protection from here —
|
||||||
pub(crate) fn add(&mut self, pin: Pin) {
|
/// and its renewal duty moves into the batch (release stops it;
|
||||||
let mut pin = pin;
|
/// while the batch lives, the member's row keeps renewing).
|
||||||
|
pub(crate) fn add(&mut self, mut pin: Pin) {
|
||||||
pin.armed = false;
|
pin.armed = false;
|
||||||
|
#[cfg(feature = "postgres")]
|
||||||
|
if let Some(renewal) = pin.take_renewal() {
|
||||||
|
self.renewals.push(renewal);
|
||||||
|
}
|
||||||
self.keys.push(pin.key().clone());
|
self.keys.push(pin.key().clone());
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -264,6 +358,13 @@ impl Batch {
|
|||||||
/// the batching put seam's commit step (which consumes the batch
|
/// the batching put seam's commit step (which consumes the batch
|
||||||
/// the same way).
|
/// the same way).
|
||||||
pub(crate) fn release_members(&mut self) {
|
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));
|
self.registry.release_pins(std::mem::take(&mut self.keys));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -70,11 +70,13 @@
|
|||||||
//! SQL-backed, pg-lo when the kv engine is not (the one-window-host
|
//! SQL-backed, pg-lo when the kv engine is not (the one-window-host
|
||||||
//! rule, ADR-012 §5); the sweep's durable form drives
|
//! rule, ADR-012 §5); the sweep's durable form drives
|
||||||
//! `window_stage`/`window_take` on that same host), while the fleet
|
//! `window_stage`/`window_take` on that same host), while the fleet
|
||||||
//! family waits for its store-core consumers — its postgres clause
|
//! family's put-path consumer has landed (the joint-tx put driver,
|
||||||
//! bodies landed with wave 4, and the store core consumes the family
|
//! `store-fleet-put`: the store core's `begin_joint_tx` → engine
|
||||||
//! only under the `fleet: true` declaration (the declaration's fleet
|
//! contribution → `pin_commit` flow, plus the pin guard's TTL/3
|
||||||
//! constructors landed; the put/sweep drivers are the remaining
|
//! renewal task minted there); the fleet sweep (`store-fleet-sweep`)
|
||||||
//! tasks). The note shrinks as those land.
|
//! 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)]
|
#![allow(dead_code)]
|
||||||
|
|
||||||
@@ -423,6 +425,12 @@ impl RegistryCore {
|
|||||||
/// Consumes the pin with its own drop suppressed (the release ran
|
/// Consumes the pin with its own drop suppressed (the release ran
|
||||||
/// inside the section).
|
/// inside the section).
|
||||||
pub(crate) fn convert<F: FnOnce()>(&self, pin: Pin, register: F) {
|
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 pin = std::mem::ManuallyDrop::new(pin);
|
||||||
let key = pin.key().clone();
|
let key = pin.key().clone();
|
||||||
let mut state = self.state.lock();
|
let mut state = self.state.lock();
|
||||||
|
|||||||
@@ -95,6 +95,17 @@ impl LocalBackend {
|
|||||||
fail_commit: false,
|
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 {
|
fn io_err(e: std::io::Error) -> Error {
|
||||||
|
|||||||
@@ -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()`,
|
/// The pg instance's `system_identifier` (`pg_control_system()`,
|
||||||
/// PG14+) — read across one pooled connection. The fleet
|
/// PG14+) — read across one pooled connection. The fleet
|
||||||
/// constructor's same-instance probe's large-side instrument
|
/// constructor's same-instance probe's large-side instrument
|
||||||
|
|||||||
+16
-4
@@ -111,10 +111,13 @@
|
|||||||
//! [`EngineFleetState`] impl carries the fleet family: the joint
|
//! [`EngineFleetState`] impl carries the fleet family: the joint
|
||||||
//! entry+pin transaction, the pin clauses, delete-time SQL arbitration
|
//! entry+pin transaction, the pin clauses, delete-time SQL arbitration
|
||||||
//! with the engine-side liveness-table registration, and the sweep
|
//! with the engine-side liveness-table registration, and the sweep
|
||||||
//! lease. The store core drives these clauses under the fleet
|
//! lease. The store core's put driver consumes the families' joint-tx
|
||||||
//! declaration (`fleet: true` — the `_fleet` constructors landed, with
|
//! and pin-renew clauses on the fleet put path (landed — the
|
||||||
//! the put/sweep drivers the fleet mechanism's remaining tasks) —
|
//! `store-fleet-put` driver: `begin_joint_tx` → joint contribution →
|
||||||
//! under `fleet: false` none of the fleet clauses run, while
|
//! `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
|
//! `ensure_gc_schema` runs on every postgres store construction (the
|
||||||
//! `store-postgres-construction` wiring, landed: the constructors are
|
//! `store-postgres-construction` wiring, landed: the constructors are
|
||||||
//! [`Store::dual_tier_postgres_local`] and [`Store::kv_only_postgres`],
|
//! [`Store::dual_tier_postgres_local`] and [`Store::kv_only_postgres`],
|
||||||
@@ -339,6 +342,15 @@ impl PostgresBackend {
|
|||||||
Self::system_identifier_of(&self.pool).await
|
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
|
/// The probe's body over a pool — shared with the pg-lo engine's
|
||||||
/// twin.
|
/// twin.
|
||||||
async fn system_identifier_of(pool: &Pool) -> Result<u64, Error> {
|
async fn system_identifier_of(pool: &Pool) -> Result<u64, Error> {
|
||||||
|
|||||||
+1785
-23
File diff suppressed because it is too large.
Load diff
+135
-12
@@ -1,7 +1,7 @@
|
|||||||
---
|
---
|
||||||
id: store-fleet-put
|
id: store-fleet-put
|
||||||
name: the fleet put path — the joint entry+pin tx driver and the pin guard's fleet form
|
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]
|
depends_on: [store-fleet-construction, engine-pg-lo-joint]
|
||||||
scope: broad
|
scope: broad
|
||||||
risk: high
|
risk: high
|
||||||
@@ -108,29 +108,29 @@ protocol verbatim; every wave-2/3/4/5 put test stays green).
|
|||||||
|
|
||||||
## Acceptance Criteria
|
## 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
|
(entry + pin row in one step; an independent connection never
|
||||||
observes one without the other); under `fleet: false` the put
|
observes one without the other); under `fleet: false` the put
|
||||||
paths are byte-unchanged (all existing suites green)
|
paths are byte-unchanged (all existing suites green)
|
||||||
- [ ] Amendment A1 landed: the driver terminates the tx on every path
|
- [x] Amendment A1 landed: the driver terminates the tx on every path
|
||||||
from `begin_joint_tx` onward (the termination design recorded;
|
from `begin_joint_tx` onward (the termination design recorded; a
|
||||||
a forced mid-driver failure leaves zero residue and returns no
|
forced mid-driver failure leaves zero residue and returns no
|
||||||
dirty session to the pool)
|
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
|
lives; drop/conversion/batch-contribution stops the renewal
|
||||||
with no leaked tasks; a failed renewal never resurrects an
|
with no leaked tasks; a failed renewal never resurrects an
|
||||||
expired row (the lifecycle design points decided and recorded)
|
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
|
renewal); a different instance's re-put composes protection by
|
||||||
row
|
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
|
(contract-tested — the pin row is durable before the committed
|
||||||
file exists; the orphan-pin crash window is the expiry
|
file exists; the orphan-pin crash window is the expiry
|
||||||
machinery's)
|
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
|
`put` (one tx over the shared instance; byte-exact large-entry
|
||||||
form)
|
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 `--features "postgres,pg-lo"` green server-less (skipping)
|
||||||
and against the harness server at acceptance;
|
and against the harness server at acceptance;
|
||||||
`cargo clippy --all-targets -- -D warnings` and
|
`cargo clippy --all-targets -- -D warnings` and
|
||||||
@@ -164,8 +164,131 @@ protocol verbatim; every wave-2/3/4/5 put test stays green).
|
|||||||
|
|
||||||
## Notes
|
## 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
|
## 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).
|
||||||
Reference in new issue
Block a user