From 3a51bb43fcc6ed711225862fe93fc1c76bc0e24e Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 16:54:21 +0000 Subject: [PATCH] =?UTF-8?q?feat(store):=20the=20fleet=20GC=20protocol=20?= =?UTF-8?q?=E2=80=94=20the=20sweep=20lease,=20SQL=20arbitration,=20pin=20r?= =?UTF-8?q?eaping,=20and=20the=20fleet=20liveness=20verb=20(wave=206)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Grow the sweep and delete paths into their fleet forms under fleet: true (ADR-008 §2, ADR-010 §2, ADR-012 §1/§5/§6.6/§6.7, ADR-013 §2); the non-fleet forms are byte-unchanged. - the sweep lease: sweep() composes the in-process single-flight flag with the SQL lease (sweep_lock_acquire on the fleet host), both held for the whole sweep and released on every exit path; contention at either domain aborts as SweeperLock report data, nothing deleted; the pool-sizing composition enforced (max_size >= 2 at the fleet constructors' entry — the lease holds one checkout; the body's clauses ride further checkouts) - SQL-level delete arbitration: arbitrated_delete composes the two domains (in-process first — cheap, local; the engine's arbitrate second — fleet-authoritative) on the direct path (GcRefuse) and the window execution's re-arbitration; a foreign instance's un-expired pin row or liveness-table row refuses cross-instance - the maintenance step: EngineFleetState::pin_reap (the seam as the named clause home) reaps exactly the expired pin rows, one statement on the postgres host, contract-tested, driven under the lease before the overdue execution - the fleet liveness verb: register_liveness_source takes the fleet-shaped LivenessRegistration argument (ADR-012 §6.6's same-verb wording) — the named-table form registers the embedder's own table (schema-ownership rule holds; idempotent; deregistration composes via a new engine-side unregister), the in-process form stays legal (instance-local protection, documented); Registration carries its kind; the verb is fallible (the fleet form's error surface) - the protect callback's fleet posture recorded: single-node-only, instance-local consult remains, materialization is the embedder's duty — no new machinery - the fleet sweep body: protect consult → reap → execute overdue (two-domain re-arbitration, fall-through delete across both tiers) → mark (pool + in-process sources; SQL liveness consulted at arbitration time, never enumerated) → stage → release; counts exact relative to the schedule - tests: lease contention/reaping between two instances, the foreign-row GcRefuse, the exact-set reap, the embedder-table protection + correction flow + malformed and fleet:false rejections, the two-instance exact-count shape, and the delay race in its fleet form; the wave-6 inert-machinery construction test rewritten to the now-live fleet semantics Verification: cargo test (default) green; cargo test --features postgres green server-less (skipping) and against pglo-poc :15432; cargo test --all-features green; cargo clippy --all-targets -D warnings green (default, postgres, all-features); cargo fmt --check green; cargo doc --no-deps clean (postgres-feature warnings unchanged at the pre-existing 21). --- src/facade.rs | 75 ++- src/gc.rs | 155 ++++- src/lib.rs | 4 +- src/postgres.rs | 145 ++++- src/store.rs | 1252 +++++++++++++++++++++++++++++++++--- tasks/store-fleet-sweep.md | 40 +- tests/conformance/rows.rs | 8 +- 7 files changed, 1550 insertions(+), 129 deletions(-) diff --git a/src/facade.rs b/src/facade.rs index 4ba8b59..d6ef617 100644 --- a/src/facade.rs +++ b/src/facade.rs @@ -500,18 +500,83 @@ pub trait ProtectCallback: Send + Sync + 'static { /// A registration token — the opaque handle the store's /// `unregister_liveness_source`/`unregister_protect_callback` verbs -/// consume to remove a -/// registered [`LivenessSource`] or [`ProtectCallback`]. ADR-012 §6.4's +/// consume to remove a registered [`LivenessSource`], a +/// [`ProtectCallback`], or a fleet liveness table registration (the +/// [`LivenessRegistration`] verb's fleet form). ADR-012 §6.4's /// embedder correction flow ("deregisters/re-registers around the -/// delete") is exactly this: the token makes deregistration expressible -/// without exposing the registry internals. +/// delete") is exactly this: the token makes deregistration +/// expressible without exposing the registry internals. /// /// Copyable, but each token deregisters at most once (the second call /// returns `false`). #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct Registration { pub(crate) id: u64, - pub(crate) protect: bool, + pub(crate) kind: RegistrationKind, +} + +/// The registration record's kind — crate-internal bookkeeping behind +/// [`Registration`], routing the deregistration verbs (a source token +/// never deregisters a protect callback, and so on). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) enum RegistrationKind { + /// An in-process [`LivenessSource`] registration. + Source, + /// A [`ProtectCallback`] registration. + Protect, + /// A fleet liveness table registration — the + /// [`LivenessRegistration::FleetTable`] form of + /// `register_liveness_source` (ADR-012 §6.6). + FleetSource, +} + +/// The liveness registration's **argument form** — ADR-012 §6.6's +/// fleet-shaped argument riding the same verb +/// (`register_liveness_source`): the two legal forms of registering a +/// liveness source, single-node and fleet. +/// +/// **Named design points (recorded):** +/// +/// - **The verb's shape and error surface.** The fleet form is not a +/// sibling method: ADR-012 §6.6's "same verb" wording pins the +/// argument-form reading, so this type IS the verb's argument and +/// the verb returns `Result` — the pre-fleet +/// signature's infallibility was an accident of the in-process form +/// alone (registration bookkeeping cannot fail there); the fleet +/// form validates the table reference (malformed references error +/// at registration, never interpolate unquoted) and forwards to the +/// shared engine, so the unified signature carries the fleet form's +/// fallibility. The in-process form never errors in practice. +/// - **In-process sources remain legal under `fleet: true`** +/// (allow-and-document): an instance-local registration's +/// protection is instance-local only — it covers *this* instance's +/// sweeps (mark-time enumeration) and *this* instance's direct +/// deletes (the in-process domain's probes); it is not a *fleet* +/// liveness form (ADR-008 §2: the SQL protocol sees tables, not +/// callbacks), so the embedder's fleet-wide protection must +/// materialize into its liveness table — or pin rows — before the +/// fleet's other instances sweep. The two domains compose: the +/// in-process registrations protect locally, the registered table +/// protects fleet-wide. +pub enum LivenessRegistration { + /// The single-node form: an in-process [`LivenessSource`] — the + /// live-shared seam consulted at mark time and at delete-time + /// arbitration (ADR-005 §Marks 1), exactly the original verb's + /// argument. Legal under every topology; under `fleet: true` its + /// protection is instance-local (see the type's design points). + InProcess(Arc), + /// The fleet form (ADR-008 §2; ADR-012 §6.6): a **named table + /// reference** in the shared engine — the embedder's own liveness + /// table (`(key bytea, …)`; schema-qualified name like + /// `"schema"."table"` — every dot-component rides the + /// quoted-identifier boundary, an empty component is an error). + /// The store core reads the table at delete-time arbitration and + /// **never owns it** — the schema-ownership rule (the engine never + /// creates, alters, or drops it); registration only records the + /// reference (idempotent re-registration of the same reference). + /// Only legal on a fleet store (`fleet: true` — under `fleet: false` + /// the fleet machinery is dormant and the verb errors). + FleetTable(String), } #[cfg(test)] diff --git a/src/gc.rs b/src/gc.rs index 98a12dd..ddceaf9 100644 --- a/src/gc.rs +++ b/src/gc.rs @@ -70,13 +70,14 @@ //! 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's put-path consumer has landed (the joint-tx put driver, -//! `store-fleet-put`: the store core's `begin_joint_tx` → engine +//! family's consumers have 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. +//! renewal task minted there) and the fleet sweep (`store-fleet-sweep` +//! — the sweep lease, `pin_reap`, and the SQL-level arbitration verbs +//! are driven: the sweep runs under the lease, the delete paths fold +//! the SQL arbitration behind the in-process domain, the maintenance +//! step reaps expired pin rows). #![allow(dead_code)] @@ -90,7 +91,9 @@ use parking_lot::Mutex; use tokio::sync::Notify; use crate::backend::Backend; -use crate::facade::{Batch, LivenessSource, Pin, PinState, ProtectCallback, Registration}; +use crate::facade::{ + Batch, LivenessSource, Pin, PinState, ProtectCallback, Registration, RegistrationKind, +}; use crate::keys::Key; use crate::Error; @@ -156,6 +159,35 @@ impl GcHome { self.core.register_source(source) } + /// Record a registered **fleet liveness table reference** (ADR-012 + /// §6.6's fleet form, the store-side half): the token deregisters + /// it via [`Self::unregister_fleet_source`]. The engine-side + /// forwarding has already succeeded when this runs. + pub(crate) fn register_fleet_source(&self, reference: String) -> Registration { + self.core.register_fleet_source(reference) + } + + /// Remove the fleet registration record for a fleet-source token, + /// handing back the named reference for the engine-side removal. + /// `None` when the token is not a fleet-source token or was + /// already consumed. + pub(crate) fn unregister_fleet_source(&self, registration: Registration) -> Option { + self.core.unregister_fleet_source(registration) + } + + /// The registered fleet liveness table count — the sweep's safe + /// default counts these beside the in-process sources. + pub(crate) fn fleet_source_count(&self) -> usize { + self.core.fleet_source_count() + } + + /// The registered liveness registrations' full count — in-process + /// sources plus fleet tables (the sweep's [`GcAbortCause::NoLivenessSources`] + /// safe default keys off zero). + pub(crate) fn registered_liveness_count(&self) -> usize { + self.registered_source_count() + self.fleet_source_count() + } + /// Deregister a previously registered liveness source; `false` when /// the token is already consumed or was never a source token. pub(crate) fn unregister_source(&self, registration: Registration) -> bool { @@ -291,6 +323,16 @@ struct RegistryState { sources: Vec<(u64, Arc)>, /// Registered protect callbacks with their registration ids. protects: Vec<(u64, Arc)>, + /// Registered **fleet liveness table references** (ADR-012 §6.6's + /// fleet form) with their registration ids — the store-side + /// bookkeeping beside the in-process registrations (the token's + /// id lives here; the named table itself is never owned by the + /// store core — the engine holds the quoted reference, the + /// schema-ownership rule keeps the table the embedder's). The + /// count feeds the sweep's safe default; deregistration removes + /// the record and hands the reference back for the engine-side + /// removal (the correction flow, ADR-012 §6.4). + fleet_sources: Vec<(u64, String)>, } impl RegistryCore { @@ -301,6 +343,7 @@ impl RegistryCore { deleting: BTreeSet::new(), sources: Vec::new(), protects: Vec::new(), + fleet_sources: Vec::new(), }), delete_done: Notify::new(), sweeping: AtomicBool::new(false), @@ -315,33 +358,97 @@ impl RegistryCore { pub(crate) fn register_source(&self, source: Arc) -> Registration { let id = self.allocate_registration(); self.state.lock().sources.push((id, source)); - Registration { id, protect: false } + Registration { + id, + kind: RegistrationKind::Source, + } } pub(crate) fn unregister_source(&self, registration: Registration) -> bool { - self.unregister(®istration, false) + self.unregister(®istration, RegistrationKind::Source) } pub(crate) fn register_protect(&self, callback: Arc) -> Registration { let id = self.allocate_registration(); self.state.lock().protects.push((id, callback)); - Registration { id, protect: true } + Registration { + id, + kind: RegistrationKind::Protect, + } } pub(crate) fn unregister_protect(&self, registration: Registration) -> bool { - self.unregister(®istration, true) + self.unregister(®istration, RegistrationKind::Protect) } - fn unregister(&self, registration: &Registration, protect: bool) -> bool { + /// Record a registered **fleet liveness table reference** (the + /// store-side half of the fleet registration verb — the engine-side + /// forwarding already succeeded): a fresh registration id beside + /// the in-process sources', the reference carried for the + /// deregistration flow. Re-registering the same reference returns + /// the original token — idempotent, mirroring the engine-side + /// registration's discipline. + pub(crate) fn register_fleet_source(&self, reference: String) -> Registration { let mut state = self.state.lock(); - if protect { - let before = state.protects.len(); - state.protects.retain(|(id, _)| *id != registration.id); - state.protects.len() != before - } else { - let before = state.sources.len(); - state.sources.retain(|(id, _)| *id != registration.id); - state.sources.len() != before + if let Some((id, _)) = state + .fleet_sources + .iter() + .find(|(_, held)| *held == reference) + { + return Registration { + id: *id, + kind: RegistrationKind::FleetSource, + }; + } + let id = self.allocate_registration(); + state.fleet_sources.push((id, reference)); + Registration { + id, + kind: RegistrationKind::FleetSource, + } + } + + /// Remove the fleet registration record for a + /// [`RegistrationKind::FleetSource`] token, handing back the named + /// reference for the engine-side removal (the correction flow, + /// ADR-012 §6.4). `None` when the token is not a fleet-source + /// token or was already consumed. + pub(crate) fn unregister_fleet_source(&self, registration: Registration) -> Option { + let mut state = self.state.lock(); + if registration.kind != RegistrationKind::FleetSource { + return None; + } + let position = state + .fleet_sources + .iter() + .position(|(id, _)| *id == registration.id)?; + Some(state.fleet_sources.remove(position).1) + } + + /// The registered fleet liveness table count — the sweep's safe + /// default counts these beside the in-process sources (a fleet + /// embedder may register nothing but its table). + pub(crate) fn fleet_source_count(&self) -> usize { + self.state.lock().fleet_sources.len() + } + + fn unregister(&self, registration: &Registration, kind: RegistrationKind) -> bool { + let mut state = self.state.lock(); + match kind { + RegistrationKind::Source => { + let before = state.sources.len(); + state.sources.retain(|(id, _)| *id != registration.id); + state.sources.len() != before + } + RegistrationKind::Protect => { + let before = state.protects.len(); + state.protects.retain(|(id, _)| *id != registration.id); + state.protects.len() != before + } + RegistrationKind::FleetSource => { + let _ = registration; + false + } } } @@ -661,6 +768,14 @@ pub(crate) trait EngineFleetState: EngineWindowState { /// and applies the decision. async fn pin_check(&self, key: &[u8]) -> Result; + /// The maintenance step (ADR-008 §2's GC-of-GC cleanup): delete the + /// **expired** pin rows (`expiry <= now`) — a pin whose holder + /// stopped renewing is abandoned, and this step reclaims its row. + /// Un-expired rows are never touched. Returns the count reaped. + /// Runs under the sweep lease (the fleet sweep body drives it — + /// the seam is the named home for fleet-clause growth, ADR-010 §2). + async fn pin_reap(&self, now: SystemTime) -> Result; + /// The full delete-time arbitration (ADR-008): the un-expired pin /// rows plus the embedder's registered liveness tables, under the /// `SELECT … FOR UPDATE` finger-lock keyed at delete time, folded diff --git a/src/lib.rs b/src/lib.rs index 3e2485f..e11ee5c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -11,8 +11,8 @@ pub mod store; pub use dispatch::DEFAULT_DISPATCH_THRESHOLD; pub use error::Error; pub use facade::{ - Batch, EntryMeta, GcAbortCause, LivenessSource, Pin, PinState, ProtectCallback, Registration, - SweepReport, + Batch, EntryMeta, GcAbortCause, LivenessRegistration, LivenessSource, Pin, PinState, + ProtectCallback, Registration, SweepReport, }; pub use hashing::{Algorithm, Digest}; pub use keys::Key; diff --git a/src/postgres.rs b/src/postgres.rs index 05974bc..a129056 100644 --- a/src/postgres.rs +++ b/src/postgres.rs @@ -93,10 +93,10 @@ //! **The crate-internal batch-delete helper** //! ([`PostgresBackend::delete_batch`]): `DELETE … WHERE key = ANY($1)` //! over a `bytea[]` parameter — ADR-007 §5's sweep-safety mechanics, -//! the sweep's batched delete form, contract-tested in-module. -//! Scope-noted under `#![allow(dead_code)]` until wave 6's fleet sweep -//! consumes it. Its virgin-store shape is a zero-row no-op (nothing to -//! delete in a table that does not exist). +//! the sweep's batched-delete shape (the store core's delete paths run +//! per-key arbitration and deletion; the helper remains available to +//! sweep-batch consumers). Its virgin-store shape is a zero-row no-op +//! (nothing to delete in a table that does not exist). //! //! **Errors.** The engine maps its failures onto the store error //! family: media/session/pool failure is [`Error::Io`], and @@ -110,13 +110,19 @@ //! fleet declaration) and stages/takes the durable delete window; the //! [`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's put driver consumes the families' joint-tx +//! with the engine-side liveness-table registration, the sweep lease, +//! and the maintenance step (`pin_reap`, the expired-pin reaping). +//! 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 +//! `pin_renew`); the store core's sweep driver consumes the sweep +//! lease, the maintenance step, and the SQL arbitration verbs (landed +//! — the `store-fleet-sweep` body: the sweep runs under the lease on +//! both the in-process flag and the advisory lock, the maintenance +//! step reaps the expired pin rows before the overdue execution, and +//! every delete-time arbitration folds the SQL answer behind the +//! in-process domain) — 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 @@ -396,8 +402,7 @@ impl PostgresBackend { /// actually removed — absent keys are simply not deleted, the sweep /// arbitrates above the trait. A `bytea[]` of zero keys removes /// nothing; a schema-less database holds nothing to remove (`Ok(0)`, - /// the read tolerance's delete mapping). Crate-internal until wave - /// 6's fleet sweep consumes it (the scope-noted `#![allow(dead_code)]`). + /// the read tolerance's delete mapping). pub(crate) async fn delete_batch(&self, keys: &[Vec]) -> Result { if keys.is_empty() { return Ok(0); @@ -429,6 +434,20 @@ impl PostgresBackend { Ok(()) } + /// The engine-side removal (the correction flow's other half, + /// ADR-012 §6.4): quote the reference the same way + /// [`Self::register_liveness_source`] did and drop it from the + /// consulted list. `false` when the reference (as quoted) is not + /// currently registered — a consumed or foreign reference is not + /// an error, deregistration stays a bool discipline. + pub(crate) fn unregister_liveness_source(&self, reference: &str) -> Result { + let quoted = quote_table_reference(reference)?; + let mut sources = self.liveness_sources.lock(); + let before = sources.len(); + sources.retain(|held| *held != quoted); + Ok(sources.len() != before) + } + /// The catalog table-existence probe — test-only: the store /// construction test's instrument (mirrors the sqlite engine's /// `sqlite_master` probe), answering for the engine-owned schema @@ -1019,6 +1038,20 @@ impl EngineFleetState for PostgresBackend { }) } + /// The maintenance step (ADR-008 §2's GC-of-GC cleanup): one + /// statement — `DELETE … WHERE expiry <= $1` — reclaiming exactly + /// the expired pin rows (the boundary matches `pin_renew`'s live + /// predicate: `expiry > now` is live, so `expiry <= now` is the + /// abandoned set). Un-expired rows are never touched; returns the + /// count reaped. The fleet sweep drives it under the lease. + async fn pin_reap(&self, now: SystemTime) -> Result { + let client = self.pool.get().await.map_err(err)?; + let now_nanos = time_to_nanos(now); + let sql = format!("DELETE FROM {} WHERE expiry <= $1", self.pins_table); + let stmt = client.prepare_cached(&sql).await.map_err(pg_err)?; + client.execute(&stmt, &[&now_nanos]).await.map_err(pg_err) + } + /// The full delete-time arbitration, folded under the `FOR UPDATE` /// finger-lock in one `BEGIN … COMMIT` step: un-expired pin rows /// for the key first, then each registered liveness-table @@ -2414,6 +2447,98 @@ mod tests { ); } + /// The maintenance step (`pin_reap`, ADR-008 §2's GC-of-GC + /// cleanup): exactly the expired set is reclaimed — rows whose + /// expiry is past `now` delete (count returned exact), un-expired + /// rows survive untouched, and per-row composition holds (a key + /// with one expired and one live row keeps the live row — + /// protection composes by row, the reaper reclaims by row). A + /// second reap of the same instant reclaims nothing. + #[tokio::test] + async fn pin_reap_reaps_exactly_the_expired_set() { + let Some(backend) = harness_gc("reap").await else { + return; + }; + let now = SystemTime::now(); + let expired_solo = pin_row( + b"\x01reap-expired-solo", + b"\x01owner-a", + now - Duration::from_secs(1), + ); + let expired_pair = pin_row( + b"\x01reap-pair-key", + b"\x01owner-b", + now - Duration::from_secs(300), + ); + let live_pair = pin_row( + b"\x01reap-pair-key", + b"\x01owner-c", + now + Duration::from_secs(30), + ); + let live_solo = pin_row( + b"\x01reap-live-solo", + b"\x01owner-d", + now + Duration::from_secs(400), + ); + + { + let mut tx = backend.begin_joint_tx().await.unwrap(); + EngineFleetState::joint_put(&*backend, &mut tx, &expired_solo.key, CONTENT) + .await + .unwrap(); + backend + .pin_commit( + tx, + &[ + expired_solo.clone(), + expired_pair.clone(), + live_pair.clone(), + ], + ) + .await + .unwrap(); + } + { + let tx = backend.begin_joint_tx().await.unwrap(); + backend + .pin_commit(tx, std::slice::from_ref(&live_solo)) + .await + .unwrap(); + } + + let reaped = EngineFleetState::pin_reap(&*backend, now).await.unwrap(); + assert_eq!( + reaped, 2, + "exactly the expired rows reclaimed: the solo and the pair-key's expired owner" + ); + + let after = pin_rows(&backend).await; + let mut expected: Vec<(Vec, Vec, i64)> = vec![ + ( + live_pair.key.clone(), + live_pair.owner.clone(), + time_to_nanos(live_pair.expiry), + ), + ( + live_solo.key.clone(), + live_solo.owner.clone(), + time_to_nanos(live_solo.expiry), + ), + ]; + expected.sort(); + assert_eq!( + after, expected, + "un-expired rows survive untouched — including the live row over the reaped key" + ); + + let second = EngineFleetState::pin_reap(&*backend, now).await.unwrap(); + assert_eq!( + second, 0, + "a second reap of the same instant reclaims nothing" + ); + assert_eq!(pin_rows(&backend).await, expected, "no further change"); + } + /// The constructor-carried pin TTL reaches the renewal body: a /// backend opened through `new_with_pin_ttl` (the fleet /// constructors' instrument) renews a live row to diff --git a/src/store.rs b/src/store.rs index 0abba19..620972f 100644 --- a/src/store.rs +++ b/src/store.rs @@ -327,11 +327,13 @@ use tokio::io::{AsyncRead, AsyncReadExt}; use crate::backend::{Backend, ReadCursor, StagingCursor}; use crate::dispatch::{route, Tier}; +#[cfg(test)] +use crate::facade::LivenessSource; #[cfg(feature = "postgres")] use crate::facade::PinRenewal; use crate::facade::{ - Batch, Blob, EntryMeta, GcAbortCause, LivenessSource, Pin, ProtectCallback, Registration, - SweepReport, + Batch, Blob, EntryMeta, GcAbortCause, LivenessRegistration, Pin, ProtectCallback, Registration, + RegistrationKind, SweepReport, }; use crate::gc::{Arbitration, EngineWindowState, GcHome, WindowCandidate}; use crate::hashing::{Algorithm, BlobHasher}; @@ -481,6 +483,31 @@ enum FleetLargeShape { PgLo, } +/// **The fleet sweep lease's pool-sizing guard** (the sweep-lease +/// design point's constructor-level enforcement, ADR-013 §2): the SQL +/// lease holds one pooled checkout for the sweep's duration, and the +/// fleet sweep body's other engine clauses ride further checkouts +/// under the held lease — a single-connection pool would make the +/// sweep acquire-wait against itself, a deadlock no sweep may +/// tolerate. A sequential fleet sweep body holds exactly two +/// concurrent checkouts (the lease's plus the current clause's), so +/// `max_size >= 2` is the floor; the shipped default (8) provides +/// headroom. Enforced at every `_fleet` constructor's entry — errors +/// closed, before any connection attempt. +#[cfg(feature = "postgres")] +fn fleet_pool_size_guard(opts: &PgOpts) -> Result<(), Error> { + if opts.max_size < 2 { + return Err(Error::Io( + "fleet: the kv pool must size at least 2 connections — the sweep lease holds \ + its checkout for the sweep's duration (ADR-013 §2) and the sweep body's other \ + clauses ride further checkouts under it; a single-connection pool would \ + deadlock the fleet sweep" + .to_owned(), + )); + } + Ok(()) +} + /// **The composite fleet-validity predicate** — backends-and-dispatch.md's /// one-sentence rule, both tiers, as code: a configuration is /// fleet-valid iff **kv = postgres AND (large = pg-lo on the same pg @@ -1509,6 +1536,7 @@ impl Store { .to_owned(), )); } + fleet_pool_size_guard(&opts)?; let kv = KvEngine::Postgres(Arc::new( PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?, )); @@ -1571,6 +1599,7 @@ impl Store { .to_owned(), )); } + fleet_pool_size_guard(&opts)?; let kv = KvEngine::Postgres(Arc::new( PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?, )); @@ -1641,6 +1670,7 @@ impl Store { .to_owned(), )); } + fleet_pool_size_guard(&opts)?; 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?; @@ -2410,6 +2440,22 @@ impl Store { } } + /// The fleet host when the **fleet machinery is active** — `Some` + /// exactly on a declared fleet store whose kv engine is the + /// postgres host (the composite fleet-validity predicate makes + /// that pairing structural). Every fleet-clause consumer branches + /// through this: under `fleet: false` (including every pre-existing + /// constructor, even postgres-kv ones) it answers `None` and the + /// in-process protocol applies verbatim. + #[cfg(feature = "postgres")] + fn fleet_sql_host(&self) -> Option<&Arc> { + if self.fleet.is_some() { + self.fleet_host() + } else { + 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 @@ -2841,26 +2887,105 @@ const SWEEP_DELETE_BATCH: usize = 100; /// (ADR-012 §6.4): pinned, re-observed live via a registered source, /// or named by a protect callback. impl Store { - /// Register a [`LivenessSource`] — the live-shared GC seam (ADR-005 - /// §Marks 1; `register_liveness_source` verbatim; POC #1 finding - /// 7's finding is the verb's pinning reason). The source is - /// consulted **live** at mark time and at delete-time arbitration — - /// never a snapshot at registration. The returned - /// [`Registration`] token deregisters via - /// [`Store::unregister_liveness_source`] (ADR-012 §6.4's correction - /// flow). Zero registered sources make every sweep abort with - /// [`GcAbortCause::NoLivenessSources`] — the safe default: a - /// consumer without registered liveness sources cannot have its - /// content deleted. - pub fn register_liveness_source(&self, source: Arc) -> Registration { - self.gc.register_source(source) + /// Register a [`LivenessRegistration`] — the live-shared GC seam + /// (ADR-005 §Marks 1; `register_liveness_source` verbatim; POC #1 + /// finding 7's finding is the verb's pinning reason). **ADR-012 + /// §6.6's fleet-shaped argument:** the same verb carries both + /// forms — + /// + /// - [`LivenessRegistration::InProcess`] — the original form (an + /// [`Arc`]); consulted **live** at mark time + /// and at delete-time arbitration, never a snapshot at + /// registration. Legal under every topology; under `fleet: true` + /// its protection is **instance-local only** (the allow-and- + /// document posture: the in-process callback view is not a + /// *fleet* liveness form — ADR-008 §2 — so this instance's + /// sweeps and direct deletes see it, other instances' SQL + /// arbitration does not; the embedder's fleet-wide protection + /// materializes into its liveness table or pin rows before the + /// fleet's other instances sweep). + /// - [`LivenessRegistration::FleetTable`] — the fleet form: a + /// named (schema-qualified) table reference registered into the + /// shared engine, read by the delete-time SQL arbitration and + /// **never owned by the store** (the schema-ownership rule — the + /// engine never creates the table). Malformed references error + /// here; re-registering the same reference is idempotent. Only + /// legal on a fleet store — the verb errors under `fleet: false` + /// (the fleet machinery is dormant there by declaration, ADR-012 + /// §3). + /// + /// The verb returns `Result`: the fleet form + /// validates the reference and forwards to the shared engine (both + /// fallible); the in-process form never errors in practice — the + /// pre-fleet signature's infallibility was the in-process form's + /// accident, retired by the unified surface (the recorded + /// verb-shape design point). The returned [`Registration`] token + /// deregisters via [`Store::unregister_liveness_source`] (ADR-012 + /// §6.4's correction flow). Zero registered liveness + /// registrations — in-process sources and fleet tables counted + /// together — make every sweep abort with + /// [`GcAbortCause::NoLivenessSources`] — the safe default. + /// + /// **The protect callback's fleet posture (recorded here where the + /// liveness verbs live):** [`ProtectCallback`] is single-node-only + /// (ADR-008 §2) — a fleet sweeper cannot consult another node's + /// in-process callback. A registered callback remains consultable + /// by *this* instance's sweeps (its protection is instance-local), + /// and the embedder materializes its fleet-wide protect set into + /// its liveness table (or pin rows) *before* invoking sweep; no + /// new machinery, the posture only. + pub fn register_liveness_source( + &self, + source: LivenessRegistration, + ) -> Result { + match source { + LivenessRegistration::InProcess(source) => Ok(self.gc.register_source(source)), + LivenessRegistration::FleetTable(reference) => { + #[cfg(feature = "postgres")] + if let Some(host) = self.fleet_sql_host() { + host.register_liveness_source(&reference)?; + return Ok(self.gc.register_fleet_source(reference)); + } + let _ = reference; + Err(Error::Io( + "fleet liveness table registration is a fleet: true form (ADR-012 §6.6): \ + no fleet declaration on this store — register an in-process source, or \ + construct through a _fleet constructor" + .to_owned(), + )) + } + } } - /// Deregister a liveness source previously registered on this - /// store. `false` when the token was already consumed or never - /// came from [`Store::register_liveness_source`]. + /// Deregister a liveness registration previously registered on + /// this store — the in-process sources and the fleet tables' + /// correction flow (ADR-012 §6.4): a fleet-table token removes the + /// store-side record **and** the engine-side reference (the SQL + /// arbitration stops consulting it). `false` when the token was + /// already consumed, was never a liveness token (a protect token), + /// or — fleet form — the reference was not registered. pub fn unregister_liveness_source(&self, registration: Registration) -> bool { - self.gc.unregister_source(registration) + match registration.kind { + RegistrationKind::Source => self.gc.unregister_source(registration), + RegistrationKind::Protect => false, + RegistrationKind::FleetSource => { + let Some(reference) = self.gc.unregister_fleet_source(registration) else { + return false; + }; + #[cfg(feature = "postgres")] + if let Some(host) = self.fleet_sql_host() { + // The engine-side answer is best-effort: the + // store-side record existed (the token was real), + // so the correction flow's goal — no further SQL + // consults — holds whether or not the reference + // was still on the engine's list. + let _ = host.unregister_liveness_source(&reference); + } + #[cfg(not(feature = "postgres"))] + let _ = reference; + true + } + } } /// Register a [`ProtectCallback`] — ADR-005 §Marks 3's protection @@ -2948,7 +3073,13 @@ impl Store { /// 1. **Single-flight**: the in-process sweep lease (ADR-013 §2's /// single-node analog); a contended second sweeper aborts with /// [`GcAbortCause::SweeperLock`], nothing deleted, nothing - /// staged or executed. + /// staged or executed. **Under `fleet: true` the SQL lease + /// composes** — both held for the whole sweep (`sweep_under_leases`, + /// ADR-013 §2's named exclusive semantics at the fleet host: + /// atomic test-and-set, one sweeper per pool across instances, + /// holdership ends at release, owner death, or engine-session + /// loss; contention — at either domain — is the + /// `SweeperLock` report abort with nothing deleted). /// 2. **Zero registered liveness sources** aborts with /// [`GcAbortCause::NoLivenessSources`] before anything is marked — /// the safe default (nothing deleted). @@ -2976,7 +3107,31 @@ impl Store { /// time** (not once upfront) under the arbitration lock — pinned /// or re-observed-live cancels the candidate (counted in /// `cancelled_by_arbitration`); otherwise the candidate deletes - /// (counted in `deleted`), in ~100-key batches. + /// (counted in `deleted`), in ~100-key batches. **Under + /// `fleet: true` the SQL arbitration composes behind the + /// in-process domain** (`arbitrated_delete`'s recorded + /// two-domain composition: in-process first — cheap, local; SQL + /// second — fleet-authoritative), on the direct-delete path and + /// at execution time alike; a foreign instance's un-expired pin + /// row or a liveness-table row cancels the candidate. + /// + /// **The fleet sweep body** (under `fleet: true` — + /// `sweep_window_fleet`, the `store-fleet-sweep` shape, + /// ADR-008 §2's mechanism as ADR-010 §2/ADR-012 shaped it): the + /// protect consult (instance-local, `ProtectFailed` abort precedes + /// everything in the body) → the **maintenance step** + /// (`pin_reap`: the expired pin rows are reaped — ADR-008 §2's + /// GC-of-GC cleanup, before the overdue execution; the ordering + /// recorded there) → **execute overdue** (`window_take` on the kv + /// host — every fleet-valid configuration's window hosts there, + /// the one-window-host rule — each candidate re-arbitrated at + /// execution time through the two-domain composition and deleted + /// where it lives, the fall-through across both tiers) → **mark** + /// (the pool enumerated; the live set from the in-process sources + /// — the SQL liveness tables are consulted at arbitration time, + /// never enumerated into the mark set) → **stage** at + /// `visible_after = mark_end + window` → release the lease. Under + /// `fleet: false` every form above is byte-unchanged. /// /// **The test-asserted property** (ADR-005): a visible pool entry is /// never deleted while liveness protects it, and an in-flight put @@ -2986,10 +3141,29 @@ impl Store { /// tests below assert it (the durable form's race holds *across the /// window delay* — a put landing during the delay pins in-process /// and the execution-time re-arbitration cancels the candidate). + /// **The single-flight, two-domain** (ADR-013 §2; the fleet + /// extension): the in-process flag composes with the SQL lease + /// under `fleet: true` — both held for the whole sweep, the + /// instance-local guard staying cheap and uniform beside the + /// fleet-authoritative lease. The contended second sweeper aborts + /// with [`GcAbortCause::SweeperLock`] at either domain, nothing + /// deleted, nothing staged or executed. + /// + /// **The lease's hold across the sweep body (named design point):** + /// the SQL lease holds **one pooled checkout** for the sweep's + /// duration (the session-scoped advisory lock's same-session + /// acquire/release discipline), and the sweep body's other engine + /// clauses — the maintenance step, `window_take`, the mark's `list` + /// enumeration, `window_stage` — ride **further checkouts** under + /// the held lease. The pool therefore must size ≥ 2 for a fleet + /// sweep not to acquire-wait against itself: the fleet constructors + /// enforce `max_size >= 2` at entry, and the shipped default (8) + /// provides headroom (a sequential sweep body holds exactly two + /// concurrent checkouts — the lease's and the current clause's). pub async fn sweep(&self) -> Result { let started = SystemTime::now(); - if self.gc.registered_source_count() == 0 { + if self.gc.registered_liveness_count() == 0 { return Ok(Self::sweep_report( 0, 0, @@ -3009,24 +3183,68 @@ impl Store { )); } - let result = self.sweep_window(started).await; + let result = self.sweep_under_leases(started).await; self.gc.end_sweep(); result } - /// The sweep body, under the sweep lease: the protect consult + /// The sweep body between the two single-flight domains: the + /// in-process flag is already held (acquired in [`Store::sweep`]); + /// under `fleet: true` the SQL lease is acquired on the fleet host + /// now — contention aborts as report data (`SweeperLock`, nothing + /// deleted), acquisition failure is operational (`Err`), and the + /// lease releases on **every** exit path (completion, operational + /// abort, protect abort) — the held-forever alternative would keep + /// both the advisory lock and its checkout live past the sweep. + async fn sweep_under_leases(&self, started: SystemTime) -> Result { + #[cfg(feature = "postgres")] + { + let Some(host) = self.fleet_sql_host() else { + return self.sweep_window(started).await; + }; + match EngineFleetState::sweep_lock_acquire(&**host).await { + Ok(true) => {} + Ok(false) => { + return Ok(Self::sweep_report( + 0, + 0, + 0, + Some(GcAbortCause::SweeperLock), + started, + )) + } + Err(e) => return Err(e), + } + let result = self.sweep_window(started).await; + // Release is no-op-safe by the seam's contract (holdership + // has ended either way — an unlock failure is the reaping + // case, not a fault to surface), so its answer is + // best-effort past the body's completion. + let _ = EngineFleetState::sweep_lock_release(&**host).await; + result + } + #[cfg(not(feature = "postgres"))] + { + self.sweep_window(started).await + } + } + + /// The sweep body, under the sweep lease(s): the protect consult /// (whose abort shape is [`GcAbortCause::ProtectFailed`] — and whose /// coverage extends to the durable form's execution phase), then /// the form branch on the delete window's host — the engine /// representation itself (ADR-012 §5's one-window-host rule), - /// never a flag: the **durable form** on a SQL-backed kv engine - /// (its generic host arm shared by every `EngineWindowState` - /// engine) and on the pg-lo large host when the kv engine is mem - /// (the mem arm's large-host completion), the **inline form** - /// verbatim on mem/local otherwise. Operational failures (engine - /// I/O, a malformed pool or window list) are the `Err` half of the - /// return shape (ADR-012 §1); every deliberate no-op is report data - /// instead. + /// never a flag: under `fleet: true` the **fleet sweep body** on + /// the fleet host (the kv-tier postgres engine — every fleet-valid + /// configuration's window hosts on the kv engine, the one-window- + /// host rule's fleet reading), the **durable form** on a SQL-backed + /// kv engine otherwise (its generic host arm shared by every + /// `EngineWindowState` engine) and on the pg-lo large host when the + /// kv engine is mem (the mem arm's large-host completion), the + /// **inline form** verbatim on mem/local otherwise. Operational + /// failures (engine I/O, a malformed pool or window list) are the + /// `Err` half of the return shape (ADR-012 §1); every deliberate + /// no-op is report data instead. async fn sweep_window(&self, started: SystemTime) -> Result { let mut live: BTreeSet = BTreeSet::new(); @@ -3057,12 +3275,80 @@ 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) => { + if self.fleet.is_some() { + self.sweep_window_fleet(engine, live, started).await + } else { + self.sweep_window_durable(&**engine, live, started).await + } + } #[cfg(feature = "sqlite")] KvEngine::Sqlite(engine) => self.sweep_window_durable(engine, live, started).await, } } + /// **The fleet sweep body** (`store-fleet-sweep`; ADR-008 §2 as + /// shaped by ADR-010 §2, ADR-012 §1/§5/§6.6/§6.7, ADR-013 §2) — + /// run on the fleet host under both single-flight domains, the + /// architecture's most intricate state machine across instances: + /// + /// 1. the **protect consult** already ran (above — instance-local, + /// aborting `ProtectFailed` before everything else in the body); + /// 2. the **maintenance step** (ADR-008 §2's GC-of-GC cleanup): + /// [`EngineFleetState::pin_reap`] deletes the expired pin rows — + /// a holder that stopped renewing is abandoned, and its row + /// stops being counted anywhere. The ordering is recorded: + /// reaping runs *before* the overdue execution (every later + /// arbitration sees the post-reap protection state; the + /// expired-row answer is identical either way — + /// `arbitrate` already requires `expiry > now` — so the placement + /// is a hygiene choice, made once and kept); + /// 3. **execute overdue** (`window_take` on the kv host — every + /// fleet-valid configuration's window hosts there, the + /// one-window-host rule): each overdue candidate re-arbitrated + /// at execution time — the in-process domain first (cheap, + /// local), the SQL arbitration second (fleet-authoritative: + /// un-expired pin rows of *any* instance plus the registered + /// liveness tables, under the `SELECT … FOR UPDATE` finger-lock) + /// — and deleted where it lives (the fall-through delete across + /// both tiers); + /// 4. **mark**: the pool enumerated (both tiers), the live set from + /// the in-process sources — pins, registered in-process sources, + /// the protect consult's keys. **The SQL liveness tables are not + /// enumerated into the mark set** — they are consulted at + /// arbitration time (step 3's and the direct path's SQL + /// arbitration), so a key protected only fleet-wide stages as a + /// candidate and is cancelled at its execution-time + /// re-arbitration (counted `cancelled_by_arbitration`, exact); + /// 5. **stage** the candidate-dead set at + /// `visible_after = mark_end + window` (the constructor's delay); + /// 6. the lease releases on every exit path (the caller's + /// `sweep_under_leases`). + /// + /// Report counts stay exact relative to the schedule — the durable + /// form's reading carries: staged-this-sweep vs executed-this-sweep + /// are different sets. + #[cfg(feature = "postgres")] + async fn sweep_window_fleet( + &self, + host: &Arc, + live: BTreeSet, + started: SystemTime, + ) -> Result { + let reaped = EngineFleetState::pin_reap(&**host, SystemTime::now()).await?; + let _ = reaped; + + let (deleted, cancelled_by_arbitration) = self.execute_overdue_window(&**host).await?; + let candidates_staged = self.mark_and_stage_window(&**host, live).await?; + Ok(Self::sweep_report( + candidates_staged, + deleted, + cancelled_by_arbitration, + None, + started, + )) + } + /// The durable form's body, shared by every window-host arm of /// the form branch (the branch keys off the engine /// representation, ADR-012 §5's one-window-host rule — the host @@ -3236,6 +3522,24 @@ impl Store { /// in-flight delete marker held across the delete and released on /// every path (the store-side composition of [`GcHome::arbitrate`] /// + [`GcHome::conclude_delete`]). + /// + /// **The two arbitration domains' composition (named design point; + /// ADR-008 §2's SQL-level fleet form) — in-process first, SQL + /// second:** under `fleet: true` a `Proceed` decision from the + /// in-process domain (pins, in-flight markers, the protect + /// callbacks, the in-process sources — cheap and local, guarding + /// this instance's in-flight state without engine I/O under the + /// lock) still consults the SQL `arbitrate` — the cross-instance + /// authority folded into one answer (un-expired pin rows of *any* + /// instance plus the registered liveness tables, under the `SELECT + /// … FOR UPDATE` finger-lock). Both domains must pass before any + /// delete: a foreign instance's pin row or a liveness-table row + /// refuses the candidate here, the marker releases (a blocked + /// pin mint proceeds against the live entry), and the caller sees + /// `Cancelled`. The SQL answer's failure is operational (`Err`) — + /// it propagates on the direct path and aborts a sweep's window + /// execution. Under `fleet: false` no SQL consult exists — the + /// ADR-005 protocol verbatim. async fn arbitrated_delete( &self, key: &Key, @@ -3244,6 +3548,13 @@ impl Store { match self.gc.arbitrate(key, consult_protect).await? { Arbitration::Cancelled => Ok(crate::gc::DeleteOutcome::Cancelled), Arbitration::Proceed => { + #[cfg(feature = "postgres")] + if let Some(host) = self.fleet_sql_host() { + if EngineFleetState::arbitrate(&**host, &key.as_bytes()).await? { + self.gc.conclude_delete(key); + return Ok(crate::gc::DeleteOutcome::Cancelled); + } + } let result = self.delete_everywhere(key).await; self.gc.conclude_delete(key); Ok(crate::gc::DeleteOutcome::Deleted(result?)) @@ -3952,7 +4263,9 @@ mod tests { #[tokio::test] async fn sweep_with_sources_and_an_empty_pool_is_a_zero_report() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let report = store.sweep().await.expect("sweeps are never errors"); assert_eq!(report.aborted, None); @@ -3972,7 +4285,7 @@ mod tests { .await .expect("dual-tier constructs"); let source = Arc::new(SetSource::default()); - store.register_liveness_source(source.clone()); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone())); let live = key_for(b"kept by the registered source"); plant_kv(&store, &live.as_bytes(), b"kept by the registered source").await; @@ -4016,7 +4329,9 @@ mod tests { #[tokio::test] async fn dedup_reput_guards_refcount_under_the_arbitration_lock() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let content = b"refcounted content"; let key = key_for(content); plant_kv(&store, &key.as_bytes(), content).await; @@ -4045,7 +4360,9 @@ mod tests { #[tokio::test] async fn batch_pins_drop_together_in_one_ordered_step() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let a = key_for(b"batch member a"); let b = key_for(b"batch member b"); plant_kv(&store, &a.as_bytes(), b"batch member a").await; @@ -4078,7 +4395,9 @@ mod tests { #[tokio::test] async fn protect_callback_adds_keys_and_aborts_as_data() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let protect = Arc::new(SetProtect::default()); store.register_protect_callback(protect.clone()); @@ -4121,7 +4440,7 @@ mod tests { async fn liveness_sources_are_consulted_live_not_snapshotted() { let store = Store::mem_only(); let source = Arc::new(SetSource::default()); - store.register_liveness_source(source.clone()); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone())); let key = key_for(b"live-shared content"); plant_kv(&store, &key.as_bytes(), b"live-shared content").await; @@ -4150,7 +4469,11 @@ mod tests { #[tokio::test] async fn unregistration_consumes_the_token_and_restores_the_safe_default() { let store = Store::mem_only(); - let registration = store.register_liveness_source(Arc::new(SetSource::default())); + let registration = store + .register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))) + .expect("the in-process registration never errors in practice"); assert!( store.unregister_liveness_source(registration), @@ -4184,7 +4507,7 @@ mod tests { .await .expect("dual-tier constructs"); let source = Arc::new(SetSource::default()); - store.register_liveness_source(source.clone()); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone())); let protect = Arc::new(SetProtect::default()); store.register_protect_callback(protect.clone()); @@ -4269,7 +4592,7 @@ mod tests { async fn pin_conversion_releases_and_registers_in_one_ordered_step() { let store = Store::mem_only(); let source = Arc::new(SetSource::default()); - store.register_liveness_source(source.clone()); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone())); let key = key_for(b"converted protection"); let pin = store.gc().acquire(&key).await; @@ -4303,7 +4626,7 @@ mod tests { async fn conversion_raced_against_direct_delete_never_deletes() { let store = Store::mem_only(); let source = Arc::new(SetSource::default()); - store.register_liveness_source(source.clone()); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone())); for round in 0..64 { let content = format!("converted protection round {round}"); @@ -4345,7 +4668,9 @@ mod tests { #[tokio::test] async fn sweep_racing_a_pinned_put_never_deletes_the_protected_entry() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); for round in 0..64 { let content = format!("race round {round}"); @@ -4400,7 +4725,9 @@ mod tests { #[tokio::test] async fn put_publishing_after_a_sweep_started_is_never_deleted() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let content = b"published after the mark"; let key = key_for(content); @@ -4445,7 +4772,9 @@ mod tests { #[tokio::test] async fn reads_during_a_delete_window_observe_no_torn_state() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let content = b"whole-content visibility under the window"; let key = key_for(content); @@ -4502,7 +4831,9 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn concurrent_sweeps_single_flight_and_the_second_aborts_as_data() { let store = Arc::new(Store::mem_only()); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); @@ -4559,7 +4890,9 @@ mod tests { #[tokio::test] async fn sweep_of_a_malformed_pool_list_is_an_operational_error() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); plant_kv(&store, &[0x07; 33], b"stray in the table").await; assert_eq!( @@ -4988,7 +5321,9 @@ mod tests { let store = Store::dual_tier_mem_local(root.path(), 64) .await .expect("dual-tier constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let big: Vec = (0..200u32).map(|i| (i % 241) as u8).collect(); let big_other: Vec = (0..288u32).map(|i| (i % 223) as u8).collect(); @@ -5050,7 +5385,9 @@ mod tests { let store = Store::dual_tier_mem_local(root.path(), 64) .await .expect("dual-tier constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let content = b"deduplicated content"; let (known_key, first) = store @@ -5114,7 +5451,9 @@ mod tests { let store = Store::dual_tier_mem_local(root.path(), 64) .await .expect("dual-tier constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let small_a = b"manifest member a"; let small_b = b"manifest member b"; @@ -5178,7 +5517,9 @@ mod tests { let store = Store::dual_tier_mem_local(root.path(), 64) .await .expect("dual-tier constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let landed = b"the member that landed"; let mut batch = store.batch(); @@ -5226,7 +5567,9 @@ mod tests { #[tokio::test] async fn kv_only_mode_rejects_over_threshold_puts_at_the_two_points() { let store = Store::kv_only_mem(64); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let err = store.put(Some(65), Unreadable).await.unwrap_err(); assert!( @@ -5343,7 +5686,9 @@ mod tests { #[tokio::test] async fn mem_only_accepts_every_length_on_both_put_paths() { let store = Store::mem_only(); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); assert_eq!(store.threshold(), None, "no threshold in effect"); let small = b"mem-only content"; @@ -5909,7 +6254,7 @@ mod sqlite_window_tests { .await .expect("dual-tier sqlite+local constructs"); let source = Arc::new(SetSource::default()); - store.register_liveness_source(source.clone()); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone())); let live = key_for(b"kept by the registered source"); plant_kv(&store, &live, b"kept by the registered source").await; @@ -5972,7 +6317,9 @@ mod sqlite_window_tests { let store = Store::kv_only_sqlite(&site.database, 64, Duration::from_secs(600)) .await .expect("kv-only sqlite constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let dead_a = key_for(b"dead under a future-facing delay a"); plant_kv(&store, &dead_a, b"dead under a future-facing delay a").await; @@ -6012,7 +6359,9 @@ mod sqlite_window_tests { let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO) .await .expect("kv-only sqlite constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let pinned = key_for(b"the twin a put lands under during the window"); plant_kv( @@ -6077,7 +6426,9 @@ mod sqlite_window_tests { let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO) .await .expect("kv-only sqlite constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let victim = key_for(b"deleted during the delay"); plant_kv(&store, &victim, b"deleted during the delay").await; @@ -6131,7 +6482,9 @@ mod sqlite_window_tests { let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO) .await .expect("kv-only sqlite constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let protect = Arc::new(SetProtect::default()); store.register_protect_callback(protect.clone()); @@ -6182,7 +6535,9 @@ mod sqlite_window_tests { .await .expect("kv-only sqlite constructs"), ); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); @@ -6655,7 +7010,7 @@ mod postgres_construction_tests { let store = Store::dual_tier_postgres_local(&dsn, opts, root.path(), 64, Duration::ZERO) .await .expect("dual-tier constructs"); - store.register_liveness_source(source.clone()); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone())); let dead = key_for(b"a dead candidate under the durable form"); store @@ -7028,7 +7383,9 @@ mod mem_pglo_construction_tests { let store = Store::dual_tier_mem_pglo(&site.dsn, PgOpts::default(), 64, Duration::ZERO) .await .expect("mem+pglo constructs"); - store.register_liveness_source(Arc::new(SetSource::default())); + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); let content = b"a dead candidate under the pg-lo host's window"; let (dead, _) = store @@ -7809,16 +8166,17 @@ mod postgres_fleet_construction_tests { ); } - /// 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. + /// The fleet declaration's scope after the sweep driver landed: + /// the kv put routes through the joint entry+pin tx (the pin row + /// lands with the entry — asserted against the pin check), and the + /// sweep runs the **fleet body** on the kv host — the form branch + /// keys off the declaration, the durable window stages there, and + /// the instance's own un-expired pin row protects the candidate at + /// the execution-time SQL re-arbitration (guard dropped, the row + /// still lives its TTL) until it expires and the maintenance step + /// reaps it — then the candidate executes. #[tokio::test] - async fn fleet_declaration_leaves_the_sweep_machinery_to_its_task() { + async fn fleet_declaration_runs_the_sweep_machinery_on_the_kv_host() { struct SetSource; impl LivenessSource for SetSource { fn live_keys(&self) -> Vec { @@ -7834,10 +8192,12 @@ mod postgres_fleet_construction_tests { return; }; let dsn = harness_dsn().expect("harness dsn re-read"); - let store = Store::kv_only_postgres_fleet(&dsn, opts, 64, Duration::ZERO, DEFAULT_PIN_TTL) - .await - .expect("constructs"); - store.register_liveness_source(Arc::new(SetSource)); + let store = + Store::kv_only_postgres_fleet(&dsn, opts, 64, Duration::ZERO, Duration::from_secs(2)) + .await + .expect("constructs"); + let _ = + store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(SetSource))); let content = b"a dead candidate on a declared fleet store"; let (key, pin) = store @@ -7852,15 +8212,24 @@ mod postgres_fleet_construction_tests { assert_eq!(staged.aborted, None); assert_eq!( staged.candidates_staged, 1, - "the durable form staged on the kv host (the form branch keys off the \ - engine representation)" + "the fleet body staged on the kv host (the one-window-host rule's fleet reading)" ); assert_eq!(staged.deleted, 0); let executed = store.sweep().await.expect("sweeps are never errors"); assert_eq!( - executed.deleted, 1, - "the next sweep executed the overdue set" + (executed.deleted, executed.cancelled_by_arbitration), + (0, 1), + "the instance's own un-expired pin row protects at the execution-time SQL \ + re-arbitration (guard dropped, the row lives its TTL)" + ); + assert!(store.has(&key.as_bytes()).await.unwrap()); + + tokio::time::sleep(Duration::from_secs(3)).await; + let reaped = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + reaped.deleted, 1, + "the expired row reaped by the maintenance step, the re-arbitration passes" ); assert!(!store.has(&key.as_bytes()).await.unwrap()); } @@ -9386,3 +9755,728 @@ mod postgres_pglo_fleet_put_tests { drop((guard_one, guard_two, guard_three, guard_four)); } } + +/// The wave-6 fleet sweep's store-core tests (feature `postgres`) — +/// the sweep lease's fleet form (contention between two instances of +/// one store shape → the second aborts `SweeperLock` with nothing +/// deleted; release reaping), SQL-level delete arbitration on the +/// direct path (a foreign owner's un-expired pin row refuses — +/// `GcRefuse`, the cross-instance protection proof), the maintenance +/// step reaping exactly the expired set, the fleet liveness verb (a +/// test-owned embedder table protects; the engine never creates the +/// table; deregistration composes), the two-instance exact-count +/// sweep shape (instance A pins, instance B sweeps), and the +/// sweep-vs-put race across the window delay in its fleet form. The +/// fleet liveness-table registration rejection under `fleet: false` +/// and the pool-sizing guard round out the constructor surfaces. +/// Env-driven (`ALKBLOBS_PG_*`), skipping cleanly server-less; every +/// test takes a fresh engine-owned schema on the harness database +/// with a janitor removing it in every exit path. +#[cfg(all(test, feature = "postgres"))] +mod postgres_fleet_sweep_tests { + use super::*; + use crate::pg_common::qualified_table; + use crate::postgres::test_support::SchemaJanitor; + use std::collections::BTreeSet; + 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" + ); + } + + /// The store's pin rows in their stored encoding, in the + /// `(key, owner)` sort order — read over the independent + /// connection (only durable state, outside every store's pool). + type StoredPinRows = Vec<(Vec, Vec, i64)>; + + struct TestSite { + dsn: String, + schema: String, + _janitor: SchemaJanitor, + } + + async fn fresh_site(tag: &str) -> Option { + let dsn = harness_dsn()?; + let schema = format!( + "abtfsweep_{}_{}_{:x}", + tag, + std::process::id(), + SCHEMA_SEQ.fetch_add(1, Ordering::Relaxed) + ); + Some(TestSite { + dsn: dsn.clone(), + schema: schema.clone(), + _janitor: SchemaJanitor::new(&dsn, &schema), + }) + } + + /// A fleet store instance over the site's shared schema — the + /// two-instance tests' instrument: two instances of one store + /// shape, one pool (the fleet's shared engine-owned schema, one + /// lock domain). + async fn fleet_instance(site: &TestSite, tag: &str, ttl: Duration) -> Store { + let opts = crate::PgOpts { + schema: site.schema.clone(), + ..crate::PgOpts::default() + }; + Store::kv_only_postgres_fleet(&site.dsn, opts, 64, Duration::ZERO, ttl) + .await + .unwrap_or_else(|e| panic!("{tag}: fleet store constructs: {e}")) + } + + fn key_for(content: &[u8]) -> Key { + Key::derive(content, content.len() as u64) + } + + /// Plant an entry straight into the fleet host's kv table, + /// bypassing the put seam — the dead-candidate instrument (no pin + /// row, no pin guard: the entry is genuinely unprotected). + async fn plant_kv(store: &Store, key: &[u8], value: &[u8]) { + match store.kv() { + KvEngine::Postgres(engine) => engine.put(key, value).await.expect("kv plant"), + _ => panic!("these tests construct postgres fleet stores"), + } + } + + /// Mint a pin row with an arbitrary owner id and expiry (row-only + /// joint tx, no `joint_put`) — the foreign-owner and expired-row + /// instruments. + async fn mint_pin_row(store: &Store, key: &[u8], owner: &[u8], expiry: SystemTime) { + let host = match store.kv() { + KvEngine::Postgres(engine) => engine, + _ => panic!("these tests construct postgres fleet stores"), + }; + let tx = EngineFleetState::begin_joint_tx(&**host).await.unwrap(); + EngineFleetState::pin_commit( + &**host, + tx, + &[PinRow { + key: key.to_vec(), + owner: owner.to_vec(), + expiry, + }], + ) + .await + .unwrap(); + } + + /// The independent connection — the durable-state probe (outside + /// every store's pool). + 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 { + async fn pin_rows(&self, schema: &str) -> StoredPinRows { + 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() + } + + async fn table_exists(&self, schema: &str, table: &str) -> bool { + self.client + .query_one( + "SELECT EXISTS ( + SELECT 1 FROM pg_tables WHERE schemaname = $1 AND tablename = $2 + )", + &[&schema, &table], + ) + .await + .expect("table probe") + .get::<_, bool>(0) + } + + async fn create_liveness_table(&self, schema: &str, table: &str) { + let sql = format!( + "CREATE TABLE {} (key bytea PRIMARY KEY)", + qualified_table(schema, table) + ); + self.client + .execute(sql.as_str(), &[]) + .await + .expect("embedder DDL"); + } + + async fn insert_live_row(&self, schema: &str, table: &str, key: &[u8]) { + let sql = format!( + "INSERT INTO {} (key) VALUES ($1)", + qualified_table(schema, table) + ); + self.client + .execute(sql.as_str(), &[&key]) + .await + .expect("live row"); + } + } + + /// A set-backed liveness source (the in-process registration + /// form's test shape; empty unless `add` is called). + #[derive(Default)] + 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) + } + } + + /// A protect callback latching the consult until released — the + /// lease holder's instrument (the protect consult runs under the + /// leases, so the latch proves the lease was held). + struct LatchProtect { + entered: std::sync::mpsc::Sender<()>, + release: parking_lot::Mutex>>, + } + + impl ProtectCallback for LatchProtect { + fn protect(&self) -> Option> { + self.entered.send(()).ok(); + let receiver = self.release.lock().take(); + if let Some(receiver) = receiver { + let _ = receiver.recv(); + } + Some(Vec::new()) + } + } + + /// A short fleet test TTL: minted rows expire well within the + /// sleeps the race test takes (the TTL/3 cadence ≈ 400 ms). + const SHORT_TTL: Duration = Duration::from_millis(1200); + /// A TTL that outlives every test that must keep its rows live. + const LONG_TTL: Duration = Duration::from_secs(60); + + /// The lease's fleet form (ADR-013 §2's named exclusive semantics): + /// two instances of one store shape — one sweeper per pool across + /// instances. The holder's consult latch holds the sweep (the + /// SQL lease is already acquired when the consult runs); the + /// contended second instance's sweep aborts as report data + /// (`SweeperLock`) with nothing deleted; the release reaps and the + /// released lease sweeps. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn fleet_sweep_lease_contention_between_instances_aborts_as_data() { + let Some(site) = fresh_site("lease").await else { + skip_note(); + return; + }; + let store_a = Arc::new(fleet_instance(&site, "lease-a", LONG_TTL).await); + let store_b = fleet_instance(&site, "lease-b", LONG_TTL).await; + let _ = store_a.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); + let _ = store_b.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); + + let dead = key_for(b"dead under the contended fleet sweeps"); + plant_kv( + &store_a, + &dead.as_bytes(), + b"dead under the contended fleet sweeps", + ) + .await; + + let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + store_a.register_protect_callback(Arc::new(LatchProtect { + entered: entered_tx, + release: parking_lot::Mutex::new(Some(release_rx)), + })); + + let holder = { + let store_a = Arc::clone(&store_a); + tokio::spawn(async move { store_a.sweep().await }) + }; + while entered_rx + .recv_timeout(std::time::Duration::from_millis(2000)) + .is_err() + { + tokio::task::yield_now().await; + } + + let contended = store_b.sweep().await.expect("sweeps are never errors"); + assert_eq!( + contended.aborted, + Some(GcAbortCause::SweeperLock), + "the contended fleet sweeper aborts as report data" + ); + assert_eq!(contended.deleted, 0); + assert_eq!(contended.cancelled_by_arbitration, 0); + assert_eq!(contended.candidates_staged, 0, "nothing staged either"); + assert!( + store_a.has(&dead.as_bytes()).await.unwrap(), + "the contended sweep deleted nothing" + ); + + release_tx.send(()).ok(); + let held_report = holder + .await + .expect("holder sweep task joined") + .expect("the holder returns a report"); + assert_eq!(held_report.aborted, None, "the lease holder sweeps"); + assert_eq!( + (held_report.deleted, held_report.candidates_staged), + (0, 1), + "the holder ran its body: the durable form stages, the next sweep executes" + ); + assert!(store_a.has(&dead.as_bytes()).await.unwrap()); + + let released = store_b.sweep().await.expect("sweeps are never errors"); + assert_eq!( + released.aborted, None, + "the released lease's reaping is observable: the second instance now sweeps" + ); + assert_eq!( + (released.deleted, released.candidates_staged), + (1, 0), + "the released lease executed the overdue candidate" + ); + assert!(!store_a.has(&dead.as_bytes()).await.unwrap()); + } + + /// SQL-level delete arbitration on the direct path: a pin row from + /// a *foreign* owner id refuses the delete — the cross-instance + /// protection proof — and the refusal relaxes exactly when the row + /// expires (`GcRefuse`, then the unprotected delete succeeds). + #[tokio::test] + async fn sql_arbitration_refuses_a_foreign_pin_row_on_the_direct_path() { + let Some(site) = fresh_site("foreign-row").await else { + skip_note(); + return; + }; + let store = fleet_instance(&site, "foreign", LONG_TTL).await; + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); + + let content = b"protected by a foreign instance's pin row"; + let key = key_for(content); + plant_kv(&store, &key.as_bytes(), content).await; + + let foreign_owner: Vec = (0u8..32).collect(); + let expiry = SystemTime::now() + Duration::from_millis(300); + mint_pin_row(&store, &key.as_bytes(), &foreign_owner, expiry).await; + + assert_eq!( + store.delete(&key.as_bytes()).await.unwrap_err(), + Error::GcRefuse, + "the foreign owner's un-expired pin row refuses: the cross-instance protection" + ); + assert!(store.has(&key.as_bytes()).await.unwrap(), "nothing deleted"); + + tokio::time::sleep(Duration::from_millis(400)).await; + store + .delete(&key.as_bytes()) + .await + .expect("the expired foreign row no longer protects"); + assert!(!store.has(&key.as_bytes()).await.unwrap()); + } + + /// The maintenance step under the fleet sweep: the abandoned + /// (expired) pin row is reaped by the sweep's maintenance step and + /// its candidate executes; the live row survives — and its + /// candidate is cancelled at the execution-time re-arbitration + /// (the two arbitration domains compose at the executor: the + /// in-process domain has none of it, the SQL row is the only + /// protection present). + #[tokio::test] + async fn the_fleet_sweep_reaps_the_expired_rows_and_arbitrates_the_rest() { + let Some(site) = fresh_site("reap").await else { + skip_note(); + return; + }; + let store = fleet_instance(&site, "reap", LONG_TTL).await; + let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); + + let doomed = key_for(b"whose pin expired and sweeps"); + let survivor = key_for(b"whose pin renews and survives"); + plant_kv(&store, &doomed.as_bytes(), b"whose pin expired and sweeps").await; + plant_kv( + &store, + &survivor.as_bytes(), + b"whose pin renews and survives", + ) + .await; + + let owner = b"\x01the-instance"; + let past = SystemTime::now() - Duration::from_secs(1); + let live = SystemTime::now() + Duration::from_secs(30); + mint_pin_row(&store, &doomed.as_bytes(), owner, past).await; + mint_pin_row(&store, &survivor.as_bytes(), owner, live).await; + + let first = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!(first.candidates_staged, 2, "both entries marked (dead set)"); + assert_eq!(first.deleted, 0, "nothing overdue yet"); + + let probe = independent(&site.dsn).await; + let rows = probe.pin_rows(&site.schema).await; + assert_eq!( + rows.len(), + 1, + "the expired row was reaped by the maintenance step" + ); + assert_eq!(rows[0].0, survivor.as_bytes(), "the live row survives"); + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.aborted, None); + assert_eq!( + (second.deleted, second.cancelled_by_arbitration), + (1, 1), + "the doomed entry executed; the live-pinned entry was cancelled at the SQL \ + re-arbitration" + ); + assert_eq!(second.candidates_staged, 1, "the survivor re-staged"); + assert!(store.has(&survivor.as_bytes()).await.unwrap()); + assert!(!store.has(&doomed.as_bytes()).await.unwrap()); + } + + /// The fleet liveness verb (ADR-012 §6.6's fleet form): a + /// test-owned embedder table's rows protect — through the + /// execution-time SQL re-arbitration and the direct path alike — + /// the engine never creates the table (the schema-ownership rule), + /// a fleet-only registration satisfies the safe default, malformed + /// references error at registration, `fleet: false` rejects the + /// fleet form, and deregistration composes (the correction flow). + #[tokio::test] + async fn the_fleet_liveness_verb_registers_the_embedders_table() { + let Some(site) = fresh_site("liveness").await else { + skip_note(); + return; + }; + let store = fleet_instance(&site, "liveness", LONG_TTL).await; + + let probe = independent(&site.dsn).await; + probe + .create_liveness_table(&site.schema, "embedder_live") + .await; + + let reference = format!("{}.embedder_live", site.schema); + let registration = store + .register_liveness_source(LivenessRegistration::FleetTable(reference.clone())) + .expect("the embedder table reference registers"); + let re_registered = store + .register_liveness_source(LivenessRegistration::FleetTable(reference.clone())) + .expect("re-registration is idempotent"); + assert_eq!( + registration, re_registered, + "re-registering the same reference returns the original token" + ); + + let malformed = store + .register_liveness_source(LivenessRegistration::FleetTable( + "bad..reference".to_owned(), + )) + .expect_err("a malformed reference errors at registration"); + + assert!( + malformed + .to_string() + .contains("invalid liveness table reference"), + "the malformed reference surfaces the engine error: {malformed}" + ); + assert!( + !probe.table_exists(&site.schema, "no_such_table").await, + "no table was ever created at registration time" + ); + + let protected = key_for(b"live in the embedder's table"); + let dead = key_for(b"dead with the table beside it"); + plant_kv( + &store, + &protected.as_bytes(), + b"live in the embedder's table", + ) + .await; + plant_kv(&store, &dead.as_bytes(), b"dead with the table beside it").await; + probe + .insert_live_row(&site.schema, "embedder_live", &protected.as_bytes()) + .await; + + let first = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + first.aborted, None, + "a fleet-table-only registration satisfies the safe default" + ); + assert_eq!( + first.candidates_staged, 2, + "both marked; the table is never enumerated" + ); + assert_eq!(first.deleted, 0, "nothing overdue yet"); + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (second.deleted, second.cancelled_by_arbitration), + (1, 1), + "the dead entry executed; the table-protected entry cancelled at the SQL \ + re-arbitration" + ); + assert!(store.has(&protected.as_bytes()).await.unwrap()); + assert!(!store.has(&dead.as_bytes()).await.unwrap()); + + assert_eq!( + store.delete(&protected.as_bytes()).await.unwrap_err(), + Error::GcRefuse, + "the liveness table protects the direct path too" + ); + + assert!( + store.unregister_liveness_source(registration), + "the fleet token deregisters once" + ); + assert!( + !store.unregister_liveness_source(re_registered), + "the idempotent re-registration consumed one token only" + ); + store + .delete(&protected.as_bytes()) + .await + .expect("the correction flow: deregistered, the delete proceeds") + } + + /// The two-instance exact-count sweep shape: instance A pins (a + /// fleet put's joint entry+pin commit, its guard held), instance B + /// sweeps — the pinned entry survives (staged by B's mark, since + /// the SQL liveness is never enumerated; cancelled at the + /// execution-time SQL re-arbitration by A's row) while the + /// unpinned twins delete. Counts exact relative to the schedule: + /// staged-this-sweep vs executed-this-sweep are different sets. + #[tokio::test] + async fn two_instance_exact_counts_instance_a_pins_instance_b_sweeps() { + let Some(site) = fresh_site("counts").await else { + skip_note(); + return; + }; + let store_a = fleet_instance(&site, "counts-a", LONG_TTL).await; + let store_b = fleet_instance(&site, "counts-b", LONG_TTL).await; + let _ = store_b.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); + + let pinned = key_for(b"pinned by instance a, swept by instance b"); + let dead = key_for(b"an unpinned twin"); + let dead_too = key_for(b"another unpinned twin"); + let (_, guard) = store_a + .put( + Some(b"pinned by instance a, swept by instance b".len() as u64), + Cursor::new(&b"pinned by instance a, swept by instance b"[..]), + ) + .await + .expect("instance a's fleet put publishes"); + plant_kv(&store_b, &dead.as_bytes(), b"an unpinned twin").await; + plant_kv(&store_b, &dead_too.as_bytes(), b"another unpinned twin").await; + + let first = store_b.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!( + first.candidates_staged, 3, + "instance b's mark stages all three (A's row is not enumerable live state)" + ); + assert_eq!(first.deleted, 0, "nothing overdue yet"); + + let second = store_b.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.aborted, None); + assert_eq!( + (second.deleted, second.cancelled_by_arbitration), + (2, 1), + "the twins executed; the pinned entry cancelled at the SQL re-arbitration" + ); + assert_eq!(second.candidates_staged, 1, "the survivor re-staged"); + assert!(store_a.has(&pinned.as_bytes()).await.unwrap()); + assert!(!store_b.has(&dead.as_bytes()).await.unwrap()); + assert!(!store_b.has(&dead_too.as_bytes()).await.unwrap()); + + drop(guard); + } + + /// The sweep-vs-put race across the window delay, fleet form (the + /// wave-3 durable-form race's fleet edition): an entry staged by + /// the first sweep, a *foreign* instance's fleet put landing + /// during the delay — its row committed atomically (the joint tx) + /// before the entry could ever be visible-unpinned — and the + /// execution-time SQL re-arbitration of the overdue candidate + /// cancels it. The entry survives byte-identical; once the foreign + /// guard's row expires and is reaped, the next sweep deletes. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn the_fleet_race_across_the_window_delay_is_closed_by_the_row() { + let Some(site) = fresh_site("race").await else { + skip_note(); + return; + }; + let delays = crate::PgOpts { + schema: site.schema.clone(), + ..crate::PgOpts::default() + }; + let sweeper = Store::kv_only_postgres_fleet( + &site.dsn, + delays, + 64, + Duration::from_millis(300), + SHORT_TTL, + ) + .await + .expect("the sweeping instance constructs"); + let putter = fleet_instance(&site, "race-b", SHORT_TTL).await; + let _ = sweeper.register_liveness_source(LivenessRegistration::InProcess(Arc::new( + SetSource::default(), + ))); + + let content = b"the race entry, re-put during the window delay"; + let key = key_for(content); + plant_kv(&sweeper, &key.as_bytes(), content).await; + + let first = sweeper.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!(first.candidates_staged, 1); + assert_eq!(first.deleted, 0); + + let (_, guard) = putter + .put(Some(content.len() as u64), Cursor::new(content)) + .await + .expect("the foreign instance's fleet put lands during the delay"); + + tokio::time::sleep(Duration::from_millis(500)).await; + + let second = sweeper.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.aborted, None); + assert_eq!( + (second.deleted, second.cancelled_by_arbitration), + (0, 1), + "the overdue candidate was cancelled by the row at execution time" + ); + assert_eq!(second.candidates_staged, 1, "the survivor re-staged"); + let (_, blob) = sweeper + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry survived"); + assert_eq!( + blob.read_at(0, content.len() as u64).await.unwrap(), + content, + "the survivor is byte-identical" + ); + + drop(guard); + + tokio::time::sleep(Duration::from_millis(2200)).await; + + let third = sweeper.sweep().await.expect("sweeps are never errors"); + assert_eq!(third.aborted, None); + assert_eq!( + third.deleted, 1, + "the guard's row expired and was reaped: the entry is unprotected now" + ); + assert!(!sweeper.has(&key.as_bytes()).await.unwrap()); + } + + /// The pool-sizing guard (the sweep-lease design point's + /// enforcement): a fleet constructor rejects a single-connection + /// pool at entry — the lease holds its checkout for the sweep's + /// duration and the sweep body's other clauses ride further + /// checkouts, so a one-pool fleet sweep would self-deadlock. The + /// check errors closed before any connection attempt (server-less + /// test). + #[tokio::test] + async fn fleet_constructors_reject_a_single_connection_pool() { + let err = Store::kv_only_postgres_fleet( + "host=no-such-host.invalid port=1 user=x password=x dbname=x", + crate::PgOpts { + max_size: 1, + ..crate::PgOpts::default() + }, + 64, + Duration::ZERO, + SHORT_TTL, + ) + .await + .err() + .expect("the entry guard rejects before any connection attempt"); + assert!( + err.to_string().contains("at least 2 connections"), + "the pool-sizing guard is the rejection: {err}" + ); + } + + /// The fleet liveness verb under `fleet: false` is rejected — the + /// fleet machinery is dormant there by declaration (ADR-012 §3), + /// and a silent no-op registration would leave the embedder + /// believing its table is consulted. Env-driven (the rejection is + /// the verb's, exercised against a constructed non-fleet store). + #[tokio::test] + async fn the_fleet_liveness_form_rejects_under_fleet_false() { + let Some(site) = fresh_site("no-fleet").await else { + skip_note(); + return; + }; + let opts = crate::PgOpts { + schema: site.schema.clone(), + ..crate::PgOpts::default() + }; + let store = Store::kv_only_postgres(&site.dsn, opts, 64, Duration::ZERO) + .await + .expect("the non-fleet store constructs on the harness site"); + let rejected = store + .register_liveness_source(LivenessRegistration::FleetTable("x.y".to_owned())) + .expect_err("the fleet form rejects without the declaration"); + assert!( + rejected.to_string().contains("fleet: true form"), + "the rejection is the verb's, spelled for the embedder: {rejected}" + ); + } +} diff --git a/tasks/store-fleet-sweep.md b/tasks/store-fleet-sweep.md index d115377..4efbf43 100644 --- a/tasks/store-fleet-sweep.md +++ b/tasks/store-fleet-sweep.md @@ -1,7 +1,7 @@ --- id: store-fleet-sweep name: the fleet GC protocol — SQL sweep lease, SQL-level arbitration, pin reaping, the fleet liveness verb -status: pending +status: completed depends_on: [store-fleet-construction] scope: broad risk: high @@ -128,29 +128,29 @@ deleted by a sweep started before it committed). The non-fleet forms ## Acceptance Criteria -- [ ] Under `fleet: true`, the sweep acquires the SQL lease (the +- [x] Under `fleet: true`, the sweep acquires the SQL lease (the ADR-013 §2 semantics: atomic test-and-set, one sweeper per pool across instances, contention = `SweeperLock` report abort with nothing deleted); the in-process flag composes; under `fleet: false` the sweep forms are byte-unchanged -- [ ] Delete-time arbitration is SQL-level under fleet mode on both +- [x] Delete-time arbitration is SQL-level under fleet mode on both the direct path (`GcRefuse`) and the window execution (execution-time re-arbitration); a foreign owner's un-expired pin row protects (cross-instance proof); the two-domain composition decided and recorded -- [ ] The expired-pin reaping maintenance step landed as a new +- [x] The expired-pin reaping maintenance step landed as a new `EngineFleetState` clause on the postgres host (the seam is the named home; no raw-SQL side-channel), contract-tested, run under the lease -- [ ] The fleet liveness registration verb landed (the named table +- [x] The fleet liveness registration verb landed (the named table reference form; the verb-shape and in-process-source design points decided and recorded; the schema-ownership rule holds) -- [ ] The protect callback's fleet posture recorded (instance-local; +- [x] The protect callback's fleet posture recorded (instance-local; materialization is the embedder's fleet-wide duty) -- [ ] The fleet race property unit-tested across the window delay (a +- [x] The fleet race property unit-tested across the window delay (a put landing during the delay survives via the row + the execution-time SQL re-arbitration) -- [ ] `cargo test` (default) green; `cargo test --features postgres` +- [x] `cargo test` (default) green; `cargo test --features postgres` green server-less (skipping) and against the harness server at acceptance; `cargo clippy --all-targets -- -D warnings` and `cargo fmt --check` green (default + `--features postgres`); @@ -182,8 +182,28 @@ deleted by a sweep started before it committed). The non-fleet forms ## Notes -> To be filled by implementation agent +Landed in `src/gc.rs`, `src/facade.rs`, `src/postgres.rs`, `src/store.rs`, `tests/conformance/rows.rs` (call-site mechanical only). Decisions and design points, as recorded: + +- **The lease's hold across the sweep body (pool sizing).** The SQL lease holds **one pooled checkout** for the sweep's duration (the session-scoped advisory lock's same-session acquire/release discipline); the fleet sweep body's other engine clauses — `pin_reap`, `window_take`, the mark's `list` enumeration, `window_stage` — ride **further checkouts** under the held lease. A sequential sweep body holds exactly two concurrent checkouts (the lease's plus the current clause's), so the pool must size ≥ 2 or the sweep acquire-waits against itself (a deadlock). Enforced: `fleet_pool_size_guard` rejects `max_size < 2` at every `_fleet` constructor's entry (errors closed, server-less-testable); the shipped default (8) provides headroom. Recorded in `sweep()`/`sweep_under_leases` docs. +- **The two arbitration domains' composition: in-process first, SQL second.** Implemented once, inside `arbitrated_delete`: a `Proceed` from the in-process domain (pins, in-flight markers, protect callbacks, in-process sources — the marker mechanism closes the local race without engine I/O under the lock) still consults the SQL `arbitrate` (un-expired pin rows of *any* instance + registered liveness tables, under the `SELECT … FOR UPDATE` finger-lock); both must pass. On a SQL refusal the marker releases (`conclude_delete`) and the caller sees `Cancelled` — a blocked pin mint proceeds against the live entry. Because the composition lives at this one seam, the direct path, the durable window's execution-time re-arbitration, and any future consumer share it uniformly; the SQL answer's failure is operational (`Err`). `pin_check` stays available for pins-only answers, unconsumed (its remaining consumer is wave 7's ops renewal probe). +- **The maintenance step's clause home.** `pin_reap(now) → count` is a new `EngineFleetState` clause (ADR-010 §2's "if fleet clauses ever grow, they grow here"); the postgres host body is one statement — `DELETE … WHERE expiry <= $1` — with the boundary matching `pin_renew`'s live predicate (`expiry > now`), never touching un-expired rows; contract-tested (`pin_reap_reaps_exactly_the_expired_set`, including per-row composition: a key with one expired and one live row keeps the live row). Ordering recorded in `sweep_window_fleet`: protect consult → reap → execute overdue → mark → stage → release. Reap *before* the overdue execution is a recorded hygiene choice, not a correctness one — `arbitrate` already requires `expiry > now`, so the protection answer is identical either way. +- **The fleet liveness verb's shape.** Same verb, fleet-shaped argument: `Store::register_liveness_source` takes a new public `LivenessRegistration` enum — `InProcess(Arc)` | `FleetTable(String)` (ADR-012 §6.6's "same verb" wording; a sibling method was rejected). The verb now returns `Result`: the fleet form validates the reference (malformed references error at registration, never interpolate unquoted — `postgres.rs`'s `quote_table_reference`) and forwards to the shared engine, so its fallibility is real; the in-process form never errors in practice — the pre-fleet signature's infallibility was that form's accident, retired by the unified surface. The store-side registration bookkeeping rides `RegistryCore` beside the in-process source records (`RegistrationKind::{Source, Protect, FleetSource}` behind `Registration`), re-registration of the same reference is idempotent (the original token returns), and deregistration composes the correction flow: the store-side record removal forwards to a new engine-side `PostgresBackend::unregister_liveness_source` (quote-then-remove). +- **In-process sources under `fleet: true`: allow-and-document.** An instance-local registration's protection is instance-local only — it covers this instance's sweeps (mark-time enumeration) and direct deletes (the in-process domain's probes); it is not a *fleet* liveness form (ADR-008 §2), so the embedder's fleet-wide protection materializes into its liveness table (or pin rows) before the fleet's other instances sweep. Recorded in the `LivenessRegistration` type docs. The sweep's safe default counts both forms together (`registered_liveness_count = sources + fleet tables`) — a fleet embedder may register nothing but its table. +- **The protect callback's fleet posture: single-node-only, no new machinery.** A registered callback remains consultable by *this* instance's sweeps (its protection is instance-local, abort `ProtectFailed` precedes everything in the body); the embedder materializes its fleet-wide set into its liveness table or pin rows before invoking sweep. Recorded in the `register_liveness_source` verb docs. +- **The SQL liveness tables are never enumerated into the mark set** — consulted at arbitration time only (the fleet body's mark enumerates the pool + in-process sources). Test-asserted: entries protected only by SQL state stage as candidates and cancel at their execution-time re-arbitration (`two_instance_exact_counts…`, `the_fleet_liveness_verb…`). +- **Superseded test rewritten:** `postgres_fleet_construction_tests::fleet_declaration_leaves_the_sweep_machinery_to_its_task` (its premise — inert machinery arbitrating in-process — was the pre-task state) became `fleet_declaration_runs_the_sweep_machinery_on_the_kv_host`, asserting the now-live fleet semantics: the instance's own un-expired row cancels the candidate at execution; expiry + reap → execution. +- **Verification:** `cargo test` (default) green; `cargo test --features postgres` green server-less (skipping) and against `pglo-poc` :15432; `cargo test --all-features` green; `clippy --all-targets -D warnings` green (default, `postgres`, `all-features`); `fmt --check` green; `cargo doc --no-deps` clean (postgres-feature doc warnings unchanged at the pre-existing 21). ## Summary -> To be filled on completion \ No newline at end of file +The sweep and delete paths grew their fleet forms under `fleet: true` — the architecture's most intricate state machine extended across instances, invariants unchanged (a visible pool entry is never deleted while liveness protects it; an in-flight put is never deleted by a sweep started before it committed), the non-fleet forms byte-unchanged: + +- **The sweep lease**: under fleet mode `sweep()` acquires the in-process single-flight flag *and* the SQL lease (`sweep_lock_acquire` on the fleet host — ADR-013 §2's named exclusive semantics), holding both for the whole sweep and releasing the lease on every exit path; contention at either domain aborts as `SweeperLock` report data with nothing deleted. The pool-sizing composition is verified and enforced (`max_size ≥ 2` at the fleet constructors' entry). +- **SQL-level delete arbitration**: `arbitrated_delete` composes the two domains — in-process first (cheap, local), the SQL `arbitrate` second (fleet-authoritative) — on the direct path (`GcRefuse`, facade-owned) and the durable window's execution-time re-arbitration alike; a foreign instance's un-expired pin row or a liveness-table row refuses cross-instance. +- **The maintenance step**: `EngineFleetState::pin_reap` (the seam as the named clause home) reaps exactly the expired pin rows, one statement on the postgres host, contract-tested, driven under the lease before the overdue execution. +- **The fleet liveness verb**: `register_liveness_source` takes the fleet-shaped `LivenessRegistration` argument (ADR-012 §6.6's same-verb wording): the named-table form registers the embedder's own table in the shared engine (schema-ownership rule holds — the engine never creates it; idempotent; deregistration composes); the in-process form remains legal (instance-local protection, documented). `Registration` now carries its kind; the verb is fallible (the fleet form's error surface). +- **The protect callback's fleet posture** recorded: single-node-only, instance-local consult remains, materialization is the embedder's duty — no new machinery. +- **The fleet sweep body**: protect consult → `pin_reap` → execute overdue (`window_take` on the kv host; execution-time two-domain re-arbitration; fall-through delete across both tiers) → mark (pool + in-process sources only; SQL liveness consulted at arbitration time, never enumerated) → stage at `visible_after = mark_end + window` → release. Counts exact relative to the schedule. +- **The fleet race across the delay** unit-tested: a foreign instance's fleet put landing during the window delay survives via its atomically-committed row plus the execution-time SQL re-arbitration. + +Tests: lease contention/reaping between two instances of one shape, the foreign-row `GcRefuse`, the exact-set reap, the embedder-table protection + correction flow + malformed-reference and `fleet: false` rejections, the two-instance exact-count shape, and the delay race — env-driven, skipping cleanly server-less, green against the harness server. \ No newline at end of file diff --git a/tests/conformance/rows.rs b/tests/conformance/rows.rs index e3c36cc..893e00f 100644 --- a/tests/conformance/rows.rs +++ b/tests/conformance/rows.rs @@ -44,8 +44,8 @@ use tokio::io::{AsyncRead, ReadBuf}; use crate::factory::StoreFactory; use alkblobs::{ - Algorithm, EntryMeta, Error, GcAbortCause, Key, LivenessSource, ProtectCallback, Store, - StoreMode, DEFAULT_DISPATCH_THRESHOLD, + Algorithm, EntryMeta, Error, GcAbortCause, Key, LivenessRegistration, LivenessSource, + ProtectCallback, Store, StoreMode, DEFAULT_DISPATCH_THRESHOLD, }; /// The shared fixtures the rows run on — set-backed liveness sources, @@ -64,7 +64,9 @@ pub(crate) struct SetSource { impl SetSource { pub(crate) fn register(store: &Store) -> Arc { let source = Arc::new(SetSource::default()); - store.register_liveness_source(source.clone()); + let _ = store.register_liveness_source(LivenessRegistration::InProcess( + Arc::clone(&source) as Arc, + )); source }