feat(store): the fleet GC protocol — the sweep lease, SQL arbitration, pin reaping, and the fleet liveness verb (wave 6)

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

@@ -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<Registration, Error>` — 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<dyn LivenessSource>),
+    /// 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)]

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<String> {
+        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<dyn LivenessSource>)>,
     /// Registered protect callbacks with their registration ids.
     protects: Vec<(u64, Arc<dyn ProtectCallback>)>,
+    /// 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<dyn LivenessSource>) -> 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(&registration, false)
+        self.unregister(&registration, RegistrationKind::Source)
     }
 
     pub(crate) fn register_protect(&self, callback: Arc<dyn ProtectCallback>) -> 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(&registration, true)
+        self.unregister(&registration, 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<String> {
+        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<PinState, Error>;
 
+    /// 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<u64, Error>;
+
     /// 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

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;

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<u8>]) -> Result<u64, Error> {
         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<bool, Error> {
+        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<u64, Error> {
+        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<u8>, Vec<u8>, 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

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<dyn LivenessSource>)` | `FleetTable(String)` (ADR-012 §6.6's "same verb" wording; a sibling method was rejected). The verb now returns `Result<Registration, Error>`: 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
+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.

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<SetSource> {
         let source = Arc::new(SetSource::default());
-        store.register_liveness_source(source.clone());
+        let _ = store.register_liveness_source(LivenessRegistration::InProcess(
+            Arc::clone(&source) as Arc<dyn LivenessSource>,
+        ));
         source
     }