alkblobs 0.2.0

//! The GC protocol's facade surface's executor side: the liveness
//! registration verbs, direct `delete`, `sweep`, the delete-window
//! forms (durable/inline/fleet) under the one-window-host rule, the
//! sweep executor (single-flight, protect consults, staged
//! re-arbitrated deletes), and the GC glue (`arbitrated_delete`,
//! `delete_everywhere`) — [`crate::store`]'s most intricate state
//! machine. The delete-window commit batch size rides here too.

#[cfg(any(feature = "mem", feature = "sqlite", feature = "postgres"))]
use super::engines::KvEngine;
#[cfg(any(all(feature = "mem", feature = "pg-lo"), all(test, feature = "sqlite")))]
use super::engines::LargeEngine;
use super::{boundary_key, typed_keys, Store};
#[cfg(all(test, feature = "postgres"))]
use crate::backend::Backend;
#[cfg(all(test, any(feature = "sqlite", feature = "postgres")))]
use crate::facade::LivenessSource;
use crate::facade::{
    GcAbortCause, LivenessRegistration, ProtectCallback, Registration, RegistrationKind,
    SweepReport,
};
#[cfg(feature = "postgres")]
use crate::gc::EngineFleetState;
#[cfg(all(test, feature = "postgres"))]
use crate::gc::PinRow;
use crate::gc::{Arbitration, EngineWindowState, WindowCandidate, WINDOW_TAKE_BATCH};
use crate::keys::Key;
#[cfg(feature = "postgres")]
use crate::postgres::PostgresBackend;
#[cfg(all(test, feature = "sqlite"))]
use crate::sqlite::SqliteBackend;
use crate::Error;
use std::collections::BTreeSet;
use std::sync::Arc;
#[cfg(all(test, any(feature = "sqlite", feature = "postgres")))]
use std::time::Duration;
use std::time::SystemTime;

/// **The delete-window commit batch size** — ~100/batch, iroh's proven
/// shape (ADR-005 §Sweep; POC #1 finding 3's named shape adopted).
const SWEEP_DELETE_BATCH: usize = 100;

/// The GC protocol's facade surface (ADR-005; gc-and-namespaces.md;
/// store-api.md §Lifecycle, §Invariants 4/5): the liveness
/// registration verbs, the direct `delete` with its arbitration-then-
/// delete refusal, and the `sweep()` delete-window protocol. This is
/// the architecture's most intricate state machine in its single-node
/// form — the fleet extension (DB-backed pins, sweep lease, SQL
/// arbitration) extends it without changing its invariants.
///
/// **The state machine, one instance at a time.** All shared GC state
/// rides the crate-internal `GcHome` behind the crate-internal
/// `Store::gc` accessor: one arbitration lock
/// over per-key pin refcounts, in-flight delete markers, and the
/// registered source lists ([`crate::gc`]'s module doc carries the
/// named design points — the marker mechanism that closes the
/// sweep-vs-put race without engine I/O under the lock, and the
/// single-node sweep lease).
///
/// **Abort-as-data** (ADR-012 §1): every sweep returns
/// `Ok(SweepReport)`; aborts are `GcAbortCause` values in the report,
/// never errors. `Error::GcRefuse` is the direct-delete refusal alone,
/// decided here at the facade — it never fires below the facade (the
/// error module's altitude discipline). The full protection set
/// (ADR-012 §6.4): pinned, re-observed live via a registered source,
/// or named by a protect callback.
impl Store {
    /// 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<dyn LivenessSource>`]); 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<Registration, Error>`: 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<Registration, Error> {
        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 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 {
        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
    /// source. Consulted pre-sweep (it may add externally-known hashes
    /// or fail the sweep into [`GcAbortCause::ProtectFailed`] report
    /// data) and at direct-delete arbitration, where a
    /// protect-callback-named key is refused (`Error::GcRefuse`, ADR-012
    /// §6.4's full protection set). Registration replaces nothing and
    /// composes: each callback's set unions into the protection.
    pub fn register_protect_callback(&self, callback: Arc<dyn ProtectCallback>) -> Registration {
        self.gc.register_protect(callback)
    }

    /// Deregister a protect callback previously registered on this
    /// store. `false` when the token was already consumed or never
    /// came from [`Store::register_protect_callback`].
    pub fn unregister_protect_callback(&self, registration: Registration) -> bool {
        self.gc.unregister_protect(registration)
    }

    /// **Direct `delete(key)`** — fall-through consult (kv first, then
    /// large; ADR-012 §6.4's "arbitration-then-delete where the entry
    /// lives"), then the full protection set arbitration at delete
    /// time: a key that is pinned, re-observed live via a registered
    /// source, or named by a protect callback is refused with
    /// [`Error::GcRefuse`] — nothing deleted. Unprotected keys delete
    /// (across both homes when a degenerate pool holds the key twice);
    /// absent keys carry [`Error::Missing`]; malformed key bytes
    /// [`Error::KeyInvalid`]. A failed protect callback is refused
    /// conservatively (`GcRefuse`, nothing deleted) — the direct-path
    /// reading of abort-as-data's safe side, recorded in the facade
    /// module doc.
    ///
    /// Outside-of-sweep deletion is unusual by posture — most deletion
    /// flows through sweeps (ADR-005); an embedder's correction flow
    /// deregisters/re-registers around the delete (ADR-012 §6.4).
    pub async fn delete(&self, key: &[u8]) -> Result<(), Error> {
        let key = boundary_key(key)?;
        let bytes = key.as_bytes();

        let present = if self.kv().has(&bytes).await? {
            true
        } else {
            match self.large() {
                Some(large) => large.has(&bytes).await?,
                None => false,
            }
        };
        if !present {
            return Err(Error::Missing);
        }

        match self.arbitrated_delete(&key, true).await? {
            crate::gc::DeleteOutcome::Cancelled => Err(Error::GcRefuse),
            crate::gc::DeleteOutcome::Deleted(true) => Ok(()),
            // The entry vanished between the existence probe and the
            // window: some other path resolved it after the probe.
            crate::gc::DeleteOutcome::Deleted(false) => Err(Error::Missing),
        }
    }

    /// **`sweep()`** — the explicit mark-and-sweep with the delete
    /// window (ADR-005 §Sweep), in two forms under one signature: the
    /// **durable form** runs on the delete window's SQL-backed host
    /// (ADR-012 §5's one-window-host rule — the SQL-backed kv engine
    /// when it is SQL-backed: sqlite and postgres; the pg-lo large
    /// host otherwise: kv = mem + large = pg-lo; the branch keys off
    /// the engine representation, never a flag), and the
    /// **inline form** (ADR-012 §5: `local`/`mem` stage candidates
    /// in-process and execute the window protocol inline in the same
    /// sweep) runs verbatim elsewhere. Sweeps are never errors:
    /// every abort is report data in `SweepReport.aborted`
    /// (`Ok(SweepReport)`), and a completed sweep's counts are exact
    /// relative to its form's schedule — the inline form reports
    /// `candidates_staged == deleted + cancelled_by_arbitration` over its
    /// same-sweep staged set; the durable form reports
    /// `candidates_staged` as the candidates staged *this* sweep and
    /// `deleted`/`cancelled` as the *overdue* set executed this sweep
    /// (staged and executed are different sets there — the delay
    /// separates them, ADR-012 §6.7; `deleted`/`cancelled` are both 0 on
    /// any abort).
    ///
    /// The protocol, as run here:
    ///
    /// 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. **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). Both
    ///    releases ride their Drop guards (review-001 N-6): a
    ///    panic or cancellation inside the sweep body releases the
    ///    lease(s), never a stuck single-flight state.
    /// 2. **Zero registered liveness sources** aborts with
    ///    [`GcAbortCause::NoLivenessSources`] before anything is marked —
    ///    the safe default (nothing deleted).
    /// 3. **Protect**: each registered [`ProtectCallback`] is consulted
    ///    pre-sweep; its keys union into the live set; a callback that
    ///    fails to answer aborts with [`GcAbortCause::ProtectFailed`] —
    ///    a flaky protection source skips the sweep rather than risk
    ///    deletion (iroh's `ProtectOutcome::Abort` conclusion). On the
    ///    durable form this consult's coverage extends to the executor
    ///    phase, so an abort here precedes any overdue execution too.
    /// 4. **Durable form (SQL-backed kv engine)**, per sweep under the
    ///    lease: **execute overdue** first (the bounded
    ///    `WINDOW_TAKE_BATCH` take looped until drained —
    ///    `window_take` re-arbitrates each overdue candidate at execution
    ///    time — pin-before-publish during the window delay cancels the
    ///    candidate — and deletes), then **mark** (unchanged), then
    ///    **stage** the fresh
    ///    candidate-dead set at `visible_after = mark_end + window` (the
    ///    constructor-tunable delay, ADR-008 §2). The executor of the
    ///    staged rows is the next sweep (ADR-008 §2; ADR-012 §6.7).
    ///    **Inline form**: the mark phase enumerates the whole pool
    ///    (both tiers' complete `list()`, ADR-003 — a malformed pool
    ///    list is fatal `Err(KeyInvalid)`); the live set computed from
    ///    all three sources (registered sources, pins, protect
    ///    callbacks); the candidate-dead set staged (sorted, in-process).
    /// 5. **Delete window**: per candidate, **arbitration at delete
    ///    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`) — on the durable form over the bounded
    ///    take's drained 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
    /// is never deleted by a sweep started before it committed — the
    /// in-flight delete markers plus the blocked pin mint are the
    /// mechanism ([`crate::gc`]'s module doc), and the concurrency
    /// 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<SweepReport, Error> {
        let started = SystemTime::now();

        if self.gc.registered_liveness_count() == 0 {
            return Ok(Self::sweep_report(
                0,
                0,
                0,
                Some(GcAbortCause::NoLivenessSources),
                started,
            ));
        }

        let Some(_sweep_lease) = self.gc.begin_sweep_guarded() else {
            return Ok(Self::sweep_report(
                0,
                0,
                0,
                Some(GcAbortCause::SweeperLock),
                started,
            ));
        };

        self.sweep_under_leases(started).await
    }

    /// The sweep body between the two single-flight domains: the
    /// in-process flag is already held (acquired in [`Store::sweep`],
    /// under the in-process lease's Drop guard); 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 awaited; panic and cancellation through
    /// [`FleetSweepLeaseGuard`]'s spawned best-effort release) — the
    /// held-forever alternative would keep both the advisory lock and
    /// its checkout live past the sweep (review-001 N-6).
    async fn sweep_under_leases(&self, started: SystemTime) -> Result<SweepReport, Error> {
        #[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),
            }
            // The lease's release-on-every-path guard (review-001 N-6):
            // a panic inside the sweep body (a panicking protect
            // callback is the plausible route) cannot reach the
            // explicit release below — the guard's Drop body spawns
            // the release best-effort, disarmed only on the normal
            // path, where the awaited release has already run. A
            // cancelled sweep future resolves the same way (the guard
            // drops with the release still owed).
            let mut fleet_lease = FleetSweepLeaseGuard::armed(Arc::clone(host));
            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;
            fleet_lease.disarm();
            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: 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<SweepReport, Error> {
        let mut live: BTreeSet<Key> = BTreeSet::new();

        for callback in self.gc.registered_protects() {
            match callback.protect() {
                Some(keys) => {
                    live.extend(keys);
                }
                None => {
                    return Ok(Self::sweep_report(
                        0,
                        0,
                        0,
                        Some(GcAbortCause::ProtectFailed),
                        started,
                    ));
                }
            }
        }

        match &self.kv {
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => match self.large() {
                #[cfg(feature = "pg-lo")]
                Some(LargeEngine::PgLo(host)) => {
                    self.sweep_window_durable(host, live, started).await
                }
                _ => self.sweep_window_inline(live, started).await,
            },
            #[cfg(feature = "postgres")]
            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,
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(super::engines::no_kv_engine()),
        }
    }

    /// **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<PostgresBackend>,
        live: BTreeSet<Key>,
        started: SystemTime,
    ) -> Result<SweepReport, Error> {
        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
    /// this runs on is the SQL-backed engine the window lives on: the
    /// kv engine's SQL-backed arms and pg-lo's large-host arm). The
    /// overdue staged rows execute first — each re-arbitrated at
    /// execution time — then this sweep's own mark and stage at
    /// `visible_after = mark_end + window`.
    // consumer: the SQL-backed window hosts' durable form (the default
    // build and the pg builds); a mem-only build has no SQL host.
    #[allow(dead_code)]
    async fn sweep_window_durable<W: EngineWindowState>(
        &self,
        host: &W,
        live: BTreeSet<Key>,
        started: SystemTime,
    ) -> Result<SweepReport, Error> {
        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 inline form (ADR-012 §5): the mark → arbitrate-at-delete-
    /// time → batched commit body, unchanged — the non-SQL engines
    /// stage candidates in-process and execute the window protocol
    /// inline in the same sweep. Preserved verbatim across the durable
    /// form's arrival.
    // consumer: the in-process window hosts (kv = mem shapes — the
    // reference/test builds); the SQL-backed default and pg builds run
    // the durable form.
    #[allow(dead_code)]
    async fn sweep_window_inline(
        &self,
        mut live: BTreeSet<Key>,
        started: SystemTime,
    ) -> Result<SweepReport, Error> {
        let mut pool = typed_keys(self.kv().list().await?)?;
        if let Some(large) = self.large() {
            pool.extend(typed_keys(large.list().await?)?);
        }
        live.extend(self.gc.pinned_keys());
        for source in self.gc.registered_sources() {
            live.extend(source.live_keys());
        }

        let candidates: Vec<Key> = {
            let enumerated: BTreeSet<Key> = pool.into_iter().collect();
            let mut candidates: Vec<Key> = enumerated.difference(&live).cloned().collect();
            candidates.sort();
            candidates
        };
        let candidates_staged = candidates.len() as u64;

        let mut deleted = 0u64;
        let mut cancelled_by_arbitration = 0u64;
        for chunk in candidates.chunks(SWEEP_DELETE_BATCH) {
            for key in chunk {
                match self.arbitrated_delete(key, false).await? {
                    // A candidate already resolved elsewhere (deleted
                    // between mark and its window turn) settles here
                    // too: the completed sweep's counts stay exact over
                    // the staged set.
                    crate::gc::DeleteOutcome::Deleted(true) => deleted += 1,
                    _ => cancelled_by_arbitration += 1,
                }
            }
        }

        Ok(Self::sweep_report(
            candidates_staged,
            deleted,
            cancelled_by_arbitration,
            None,
            started,
        ))
    }

    /// **The durable form's executor step** (ADR-012 §6.7): take the
    /// overdue staged candidates (`visible_after <= now`) off the window
    /// and execute each through the same
    /// [`Store::arbitrated_delete`] path — the re-arbitration happens
    /// at execution time, so a put landing (pin-before-publish) during
    /// the window delay cancels its candidate here. The take is
    /// **bounded** ([`WINDOW_TAKE_BATCH`], review-001 N-3 — a huge overdue
    /// backlog never materializes entirely in memory in one step) and the
    /// executor **loops the take until drained** (a full batch is followed
    /// by another; empty or short ends the loop), preserving the
    /// one-sweep-executes-all-overdue semantics the sweep report's counts
    /// assume. The consult posture on this path is the sweep's protect
    /// consult (recorded named design point): it already ran for this
    /// sweep, above, and its coverage extends to the executor phase —
    /// so these arbitrations carry `consult_protect = false`, the same
    /// posture as the inline form's delete-time arbitration. A vanished
    /// candidate (deleted during the delay by a direct delete) settles
    /// into the cancelled count.
    // consumer: the durable form's overdue execution (the SQL-backed
    // hosts; see `sweep_window_durable`).
    #[allow(dead_code)]
    async fn execute_overdue_window<W: EngineWindowState>(
        &self,
        host: &W,
    ) -> Result<(u64, u64), Error> {
        let mut deleted = 0u64;
        let mut cancelled_by_arbitration = 0u64;
        loop {
            let overdue: Vec<Key> = typed_keys(host.window_take(SystemTime::now()).await?)?;
            for key in &overdue {
                match self.arbitrated_delete(key, false).await? {
                    crate::gc::DeleteOutcome::Deleted(true) => deleted += 1,
                    _ => cancelled_by_arbitration += 1,
                }
            }
            if overdue.len() < WINDOW_TAKE_BATCH {
                break;
            }
        }
        Ok((deleted, cancelled_by_arbitration))
    }

    /// **The durable form's mark + stage step**: the mark phase exactly
    /// the inline form's (pool enumerated, live set from pins/registered
    /// sources plus the protect consult's keys, candidate-dead set
    /// computed), then the candidate-dead set staged into the window at
    /// `visible_after = mark_end + window` — the constructor-tunable
    /// delay (ADR-008 §2; the store carries the value its constructor
    /// took), so execution waits at earliest for the next sweep past
    /// the deadline. Returns the staged-candidate count.
    // consumer: the durable form's mark-and-stage step (the SQL-backed
    // hosts; see `sweep_window_durable`).
    #[allow(dead_code)]
    async fn mark_and_stage_window<W: EngineWindowState>(
        &self,
        host: &W,
        mut live: BTreeSet<Key>,
    ) -> Result<u64, Error> {
        let mut pool = typed_keys(self.kv().list().await?)?;
        if let Some(large) = self.large() {
            pool.extend(typed_keys(large.list().await?)?);
        }
        live.extend(self.gc.pinned_keys());
        for source in self.gc.registered_sources() {
            live.extend(source.live_keys());
        }

        let enumerated: BTreeSet<Key> = pool.into_iter().collect();
        let mut candidates: Vec<Key> = enumerated.difference(&live).cloned().collect();
        candidates.sort();

        let mark_end = SystemTime::now();
        let visible_after = mark_end.checked_add(self.window_delay).ok_or_else(|| {
            Error::Io("the delete-window delay overflows the system clock".to_owned())
        })?;
        let staged: Vec<WindowCandidate> = candidates
            .iter()
            .map(|key| WindowCandidate {
                key: key.key_bytes().to_vec(),
                visible_after,
            })
            .collect();
        host.window_stage(&staged).await?;
        Ok(candidates.len() as u64)
    }

    /// A finished report constructor — every sweep outcome is a
    /// completed record (ADR-012 §1: the report is data, not an error
    /// carrier).
    fn sweep_report(
        candidates_staged: u64,
        deleted: u64,
        cancelled_by_arbitration: u64,
        aborted: Option<GcAbortCause>,
        started: SystemTime,
    ) -> SweepReport {
        SweepReport {
            candidates_staged,
            deleted,
            cancelled_by_arbitration,
            aborted,
            started_at: started,
            finished_at: SystemTime::now(),
        }
    }

    /// Arbitrate, then delete where the entry lives, with the
    /// 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 — **and the failure arm concludes too**: the marker
    /// runs `conclude_delete` before the error is raised, so a
    /// transient SQL failure never leaves the marker held (a held
    /// marker would block every future pin mint and arbitration on
    /// the key — review-001 F-1; every arm past a `Proceed`
    /// arbitration concludes). Under `fleet: false` no SQL consult
    /// exists — the ADR-005 protocol verbatim.
    async fn arbitrated_delete(
        &self,
        key: &Key,
        consult_protect: bool,
    ) -> Result<crate::gc::DeleteOutcome, Error> {
        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() {
                    match EngineFleetState::arbitrate(&**host, &key.key_bytes()).await {
                        Ok(true) => {
                            self.gc.conclude_delete(key);
                            return Ok(crate::gc::DeleteOutcome::Cancelled);
                        }
                        Ok(false) => {}
                        Err(err) => {
                            self.gc.conclude_delete(key);
                            return Err(err);
                        }
                    }
                }
                let result = self.delete_everywhere(key).await;
                self.gc.conclude_delete(key);
                Ok(crate::gc::DeleteOutcome::Deleted(result?))
            }
        }
    }

    /// Delete the key's pool entry wherever it physically lives — the
    /// kv home and the large home (normal pools hold exactly one;
    /// immutability means both homes of a degenerate duplicate are the
    /// same dead content). `false` when the entry was already absent:
    /// per-home `Missing` is not an operational error here, it is the
    /// presence answer.
    ///
    /// **Partial-failure reporting (review-001 N-7; ADR-012 §6.5's
    /// altitude discipline):** when the large home's delete fails after
    /// the kv home's delete already ran, the returned `Err` wraps the
    /// underlying media failure in an [`Error::Io`] whose message names
    /// both homes' outcomes — the homes-divergent state the caller must
    /// know (kv half deleted, large half still holding the residue).
    /// The wrapper form is the recorded choice: the error family is
    /// ADR-pinned (new variants are new ADRs), so neither a payload
    /// variant nor a report field rides in — one honest line, not a
    /// diagnostics dump. The kv-failure short-circuit arm propagates
    /// unwrapped: nothing has moved yet, so there is no divergence to
    /// report. The self-healing posture itself is unchanged — the next
    /// sweep's re-mark reconciles the residue (reviewed-correct, stays).
    async fn delete_everywhere(&self, key: &Key) -> Result<bool, Error> {
        let bytes = key.as_bytes();
        let mut deleted = match self.kv().delete(&bytes).await {
            Ok(()) => true,
            Err(Error::Missing) => false,
            Err(err) => return Err(err),
        };
        if let Some(large) = self.large() {
            match large.delete(&bytes).await {
                Ok(()) => deleted = true,
                Err(Error::Missing) => {}
                Err(err) => {
                    let homes = if deleted {
                        "kv home deleted — the entry's homes diverged (residue; the next sweep's re-mark reconciles)"
                    } else {
                        "kv home absent (nothing moved there)"
                    };
                    return Err(Error::Io(format!(
                        "delete partial failure: large home delete failed ({err}); {homes}"
                    )));
                }
            }
        }
        Ok(deleted)
    }
}

/// The fleet sweep lease's release guard (review-001 N-6): the sweep
/// body between the lease's acquisition and the explicit awaited
/// release cannot reach that release on a panic (the plausible route:
/// a panicking consumer-supplied protect callback) — the Drop body
/// spawns `sweep_lock_release` best-effort on the held host instead,
/// the same detached-spill posture [`crate::pg_common::
/// detached_batch_execute`] carries (a `Drop` body is synchronous;
/// the release rides a runtime task, or a one-shot thread's
/// current-thread runtime when the runtime itself is going away).
/// Disarmed only after the normal path's awaited release has run, so
/// the two releases never overlap. `sweep_lock_release` is no-op-safe
/// by the seam's contract — the defensive extra release is harmless.
#[cfg(feature = "postgres")]
struct FleetSweepLeaseGuard {
    host: Option<Arc<PostgresBackend>>,
}

#[cfg(feature = "postgres")]
impl FleetSweepLeaseGuard {
    fn armed(host: Arc<PostgresBackend>) -> Self {
        Self { host: Some(host) }
    }

    fn disarm(&mut self) {
        self.host = None;
    }
}

#[cfg(feature = "postgres")]
impl Drop for FleetSweepLeaseGuard {
    fn drop(&mut self) {
        let Some(host) = self.host.take() else {
            return;
        };
        let release = async move {
            let _ = EngineFleetState::sweep_lock_release(&*host).await;
        };
        match tokio::runtime::Handle::try_current() {
            Ok(handle) => {
                handle.spawn(release);
            }
            Err(_) => {
                let _ = std::thread::spawn(move || {
                    let Ok(rt) = tokio::runtime::Builder::new_current_thread()
                        .enable_all()
                        .build()
                    else {
                        return;
                    };
                    rt.block_on(release);
                })
                .join();
            }
        }
    }
}

/// The sweep's durable form (feature `sqlite`): the staged
/// re-arbitrated delete (ADR-008 §2; ADR-012 §5/§6.7) — the first
/// durable delete window, single-node — beside the inline form. The
/// two-sweep exact-count shapes, the race across the delay, the
/// vanished-candidate form, the abort shapes, and the delay knob's
/// staging semantics.
#[cfg(all(test, feature = "sqlite"))]
mod sqlite_window_tests {
    use super::*;
    use crate::backend::Backend;
    use crate::hashing::Algorithm;
    use std::fs;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::{SystemTime, UNIX_EPOCH};

    /// A fresh temp site per test (the construction tests' helper
    /// shape); Drop removes the file tree and the WAL sidecars.
    struct TempSite {
        database: PathBuf,
        root: PathBuf,
    }

    impl TempSite {
        fn path(&self) -> &Path {
            &self.root
        }

        fn drop_sidecar(&self, suffix: &str) {
            let base = self.database.to_string_lossy().into_owned();
            let _ = fs::remove_file(format!("{base}{suffix}"));
        }
    }

    impl Drop for TempSite {
        fn drop(&mut self) {
            self.drop_sidecar("-wal");
            self.drop_sidecar("-shm");
            let _ = fs::remove_file(&self.database);
            let _ = fs::remove_dir_all(&self.root);
        }
    }

    fn temp_site(tag: &str) -> TempSite {
        static SEQ: AtomicU64 = AtomicU64::new(0);
        let seq = SEQ.fetch_add(1, Ordering::Relaxed);
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let base = std::env::temp_dir().join(format!(
            "alkblobs-store-window-{tag}-{}-{nanos:x}-{seq}",
            std::process::id()
        ));
        fs::create_dir_all(&base).expect("temp site created");
        TempSite {
            database: base.join("pool.db"),
            root: base.join("local-root"),
        }
    }

    fn key_for(content: &[u8]) -> Key {
        Key::from(Algorithm::GitBlobSha256.derive(content, content.len() as u64))
    }

    fn sqlite_engine(store: &Store) -> &SqliteBackend {
        match store.kv() {
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("these tests construct sqlite stores"),
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(_) => panic!("these tests construct sqlite stores"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => engine,
        }
    }

    async fn plant_kv(store: &Store, key: &Key, value: &[u8]) {
        sqlite_engine(store)
            .put(&key.key_bytes(), value)
            .await
            .expect("kv plant");
    }

    async fn plant_large(store: &Store, key: &Key, value: &[u8]) {
        match store.large() {
            Some(LargeEngine::Local(engine)) => engine
                .put(&key.key_bytes(), value)
                .await
                .expect("large plant"),
            #[cfg(feature = "pg-lo")]
            Some(LargeEngine::PgLo(_)) => panic!("these tests construct local-tier stores"),
            None => panic!("planting into a mode with no large tier"),
        }
    }

    /// A set-backed liveness source — the live-shared seam, shared
    /// with the wave-2 tests' fixture shape.
    #[derive(Default)]
    struct SetSource {
        live: parking_lot::Mutex<BTreeSet<Key>>,
    }

    impl SetSource {
        fn add(&self, key: Key) {
            self.live.lock().insert(key);
        }
    }

    impl LivenessSource for SetSource {
        fn live_keys(&self) -> Vec<Key> {
            self.live.lock().iter().cloned().collect()
        }

        fn contains(&self, key: &Key) -> bool {
            self.live.lock().contains(key)
        }
    }

    /// A set-backed protect callback; `None` models the
    /// failed-to-answer case.
    #[derive(Default)]
    struct SetProtect {
        set: parking_lot::Mutex<Option<Vec<Key>>>,
    }

    impl SetProtect {
        fn set_keys(&self, keys: Vec<Key>) {
            *self.set.lock() = Some(keys);
        }

        fn fail(&self) {
            *self.set.lock() = None;
        }
    }

    impl ProtectCallback for SetProtect {
        fn protect(&self) -> Option<Vec<Key>> {
            self.set.lock().clone()
        }
    }

    /// A protect callback that blocks mid-consult — the contention
    /// window for the single-flight assert on a sqlite store.
    struct LatchProtect {
        entered: std::sync::mpsc::Sender<()>,
        release: parking_lot::Mutex<Option<std::sync::mpsc::Receiver<()>>>,
    }

    impl ProtectCallback for LatchProtect {
        fn protect(&self) -> Option<Vec<Key>> {
            self.entered.send(()).ok();
            let receiver = self.release.lock().take();
            if let Some(receiver) = receiver {
                let _ = receiver.recv();
            }
            Some(Vec::new())
        }
    }

    /// A non-destructive read of the window is not on the seam — but a
    /// far-horizon `window_take` removes everything overdue, which is
    /// exactly the observation the assertions below need (and leaves the
    /// window clean for the next step of the test). The take is bounded
    /// (review-001 N-3) and only single-calls here, so this helper is
    /// exact just as long as the tests' window sets stay far below the
    /// batch — they do; the backlog test drives its own drain.
    async fn window_drain(store: &Store) -> Vec<Key> {
        let far = SystemTime::now()
            .checked_add(Duration::from_secs(365 * 24 * 3600))
            .unwrap();
        let raws = sqlite_engine(store).window_take(far).await.unwrap();
        typed_keys(raws).expect("the window holds typed keys")
    }

    /// The two-sweep exact-count shape at the degenerate delay (the
    /// test tuning, ADR-008 §2): sweep 1 stages N — nothing deleted,
    /// the pool intact; sweep 2 executes exactly N —
    /// `deleted + cancelled == N` against the staged set, staged-this-
    /// sweep and executed-this-sweep different sets, and the window
    /// left empty.
    #[tokio::test]
    async fn durable_two_sweep_exact_counts_at_delay_zero() {
        let site = temp_site("two-sweep");
        let store = Store::dual_tier_sqlite_local(&site.database, site.path(), 64, Duration::ZERO)
            .await
            .expect("dual-tier sqlite+local constructs");
        let source = Arc::new(SetSource::default());
        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;
        source.add(live.clone());

        let dead_kv = key_for(b"dead kv entry in the window");
        plant_kv(&store, &dead_kv, b"dead kv entry in the window").await;
        let dead_large = key_for(b"dead large entry in the window");
        plant_large(&store, &dead_large, b"dead large entry in the window").await;

        let first = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(first.aborted, None);
        assert_eq!(
            first.candidates_staged, 2,
            "both tiers' dead entries staged"
        );
        assert_eq!(
            first.deleted, 0,
            "staging is not executing — the delay separates"
        );
        assert_eq!(first.cancelled_by_arbitration, 0);

        assert!(
            store.has(&dead_kv.as_bytes()).await.unwrap(),
            "sweep 1 staged, did not execute: the kv entry is intact"
        );
        assert!(
            store.has(&dead_large.as_bytes()).await.unwrap(),
            "sweep 1 staged, did not execute: the large entry is intact"
        );
        assert!(store.has(&live.as_bytes()).await.unwrap());

        let second = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(second.aborted, None);
        assert_eq!(second.candidates_staged, 0, "nothing newly dead");
        assert_eq!(
            second.deleted + second.cancelled_by_arbitration,
            2,
            "sweep 2 executed exactly the staged set"
        );
        assert_eq!(second.deleted, 2);
        assert_eq!(second.cancelled_by_arbitration, 0);

        assert!(!store.has(&dead_kv.as_bytes()).await.unwrap());
        assert!(!store.has(&dead_large.as_bytes()).await.unwrap());
        assert!(store.has(&live.as_bytes()).await.unwrap());
        assert!(
            window_drain(&store).await.is_empty(),
            "the executed rows left the window"
        );
    }

    /// The durable executor's drain (review-001 N-3): a pre-staged
    /// overdue backlog larger than [`WINDOW_TAKE_BATCH`] executes
    /// entirely in one sweep — the executor loops the bounded take until
    /// drained — and the report's counts stay exact over the backlog
    /// (deleted == the backlog, nothing cancelled, nothing newly staged),
    /// the second sweep all-zero.
    #[tokio::test]
    async fn durable_backlog_larger_than_the_take_batch_drains_in_one_sweep() {
        let site = temp_site("backlog-drain");
        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
            .await
            .expect("kv-only sqlite constructs");
        let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
            SetSource::default(),
        )));

        let total = WINDOW_TAKE_BATCH + 13;
        let horizon = SystemTime::now()
            .checked_sub(Duration::from_secs(1))
            .expect("the test horizon precedes now");
        let mut staged = Vec::new();
        for i in 0..total {
            let content = format!("backlog-{i:05}");
            let key = key_for(content.as_bytes());
            plant_kv(&store, &key, content.as_bytes()).await;
            staged.push(WindowCandidate {
                key: key.key_bytes().to_vec(),
                visible_after: horizon,
            });
        }
        sqlite_engine(&store).window_stage(&staged).await.unwrap();

        let drained = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(drained.aborted, None);
        assert_eq!(
            drained.deleted, total as u64,
            "the executor looped the bounded take until drained"
        );
        assert_eq!(drained.cancelled_by_arbitration, 0);
        assert_eq!(
            drained.candidates_staged, 0,
            "the drained pool has nothing newly dead"
        );

        let second = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(second.aborted, None);
        assert_eq!(second.deleted, 0);
        assert_eq!(second.cancelled_by_arbitration, 0);
        assert_eq!(second.candidates_staged, 0);
    }

    /// The delay knob is the staging's cadence (ADR-008 §2): at a
    /// future-facing delay the staged rows are not overdue at
    /// `window_take(now)` — no execution, no deletion — and a repeated
    /// sweep re-stages the still-dead set without duplicating rows.
    #[tokio::test]
    async fn durable_stage_at_a_future_delay_waits_past_the_window() {
        let site = temp_site("future-delay");
        let store = Store::kv_only_sqlite(&site.database, 64, Duration::from_secs(600))
            .await
            .expect("kv-only sqlite constructs");
        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;
        let dead_b = key_for(b"dead under a future-facing delay b");
        plant_kv(&store, &dead_b, b"dead under a future-facing delay b").await;

        let first = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(first.aborted, None);
        assert_eq!(first.candidates_staged, 2);
        assert_eq!(first.deleted, 0, "nothing is overdue at staging time");

        let second = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(second.candidates_staged, 2, "still dead, re-staged");
        assert_eq!(
            second.deleted, 0,
            "the window delay pushed execution out again"
        );

        let drained = window_drain(&store).await;
        let mut expected = vec![dead_a.clone(), dead_b.clone()];
        expected.sort();
        assert_eq!(drained, expected);
        assert!(store.has(&dead_a.as_bytes()).await.unwrap());
        assert!(store.has(&dead_b.as_bytes()).await.unwrap());
    }

    /// The race across the delay (the window form of ADR-005's
    /// test-asserted property): a put landing during the window (the
    /// pin minted between the staging sweep and the executor sweep,
    /// pin-before-publish) survives — the execution-time re-arbitration
    /// cancels its candidate — while the unpinned twin deletes. The
    /// cancelled candidate re-stages (and executes) only once its
    /// protection drops.
    #[tokio::test]
    async fn durable_race_across_the_delay_put_during_window_survives() {
        let site = temp_site("race-across-delay");
        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
            .await
            .expect("kv-only sqlite constructs");
        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(
            &store,
            &pinned,
            b"the twin a put lands under during the window",
        )
        .await;
        let unpinned = key_for(b"the unpinned twin");
        plant_kv(&store, &unpinned, b"the unpinned twin").await;

        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(staged.candidates_staged, 2);
        assert_eq!(staged.deleted, 0);

        {
            let _pin = store.gc().acquire(&pinned).await;
            assert_eq!(store.gc().pin_count(&pinned), 1, "pin-before-publish");

            let executed = store.sweep().await.expect("sweeps are never errors");
            assert_eq!(executed.aborted, None);
            assert_eq!(
                executed.deleted + executed.cancelled_by_arbitration,
                2,
                "execution-time re-arbitration ran over the taken set"
            );
            assert_eq!(executed.deleted, 1, "the unpinned twin deleted");
            assert_eq!(
                executed.cancelled_by_arbitration, 1,
                "the pinned candidate cancelled"
            );
            assert!(
                store.has(&pinned.as_bytes()).await.unwrap(),
                "a put landing during the window survives the next sweep's execution"
            );
            assert!(!store.has(&unpinned.as_bytes()).await.unwrap());
        }

        // The pin dropped: the still-present candidate re-stages, and
        // then (and only then) executes — its window row was consumed
        // by the cancellation, not by a delete.
        let restaged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            restaged.candidates_staged, 1,
            "unpinned, the candidate re-stages"
        );
        assert_eq!(restaged.deleted, 0);

        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(executed.deleted, 1, "the re-staged candidate executes");
        assert!(!store.has(&pinned.as_bytes()).await.unwrap());
    }

    /// The vanished-candidate form: a staged candidate deleted during
    /// the delay by a direct delete settles into
    /// `cancelled_by_arbitration` at execution time — the vanished-
    /// candidate accounting carries over (nothing to delete, counts
    /// stay exact).
    #[tokio::test]
    async fn durable_vanished_candidate_settles_into_cancelled() {
        let site = temp_site("vanished");
        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
            .await
            .expect("kv-only sqlite constructs");
        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;

        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(staged.candidates_staged, 1);

        store
            .delete(&victim.as_bytes())
            .await
            .expect("the direct delete lands on the unprotected entry");

        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(executed.aborted, None);
        assert_eq!(executed.candidates_staged, 0);
        assert_eq!(executed.deleted, 0);
        assert_eq!(
            executed.cancelled_by_arbitration, 1,
            "the vanished candidate settles into the cancelled count"
        );
        assert!(window_drain(&store).await.is_empty());
    }

    /// The abort shapes on a sqlite store delete nothing, execute
    /// nothing overdue, and stage nothing: zero sources
    /// (`NoLivenessSources`), a failing protect callback
    /// (`ProtectFailed`), and lease contention (`SweeperLock`).
    #[tokio::test]
    async fn durable_aborts_delete_nothing_and_execute_nothing() {
        let site = temp_site("aborts-sources");
        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
            .await
            .expect("kv-only sqlite constructs");
        let dead = key_for(b"dead while no sources are registered");
        plant_kv(&store, &dead, b"dead while no sources are registered").await;

        let aborted = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(aborted.aborted, Some(GcAbortCause::NoLivenessSources));
        assert_eq!(aborted.candidates_staged, 0);
        assert_eq!(aborted.deleted, 0);
        assert!(store.has(&dead.as_bytes()).await.unwrap());
        assert!(window_drain(&store).await.is_empty(), "nothing staged");
    }

    /// `ProtectFailed` aborts before the executor step on the durable
    /// form: an overdue row staged by an earlier sweep is not taken
    /// while the protection consult fails — abort precedes execution.
    #[tokio::test]
    async fn durable_protect_failed_aborts_before_executing_the_overdue_set() {
        let site = temp_site("aborts-protect");
        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
            .await
            .expect("kv-only sqlite constructs");
        let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
            SetSource::default(),
        )));
        let protect = Arc::new(SetProtect::default());
        store.register_protect_callback(protect.clone());

        let staged_dead = key_for(b"staged dead while the consult answers");
        plant_kv(
            &store,
            &staged_dead,
            b"staged dead while the consult answers",
        )
        .await;
        protect.set_keys(Vec::new());
        let first = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(first.aborted, None);
        assert_eq!(first.candidates_staged, 1);

        protect.fail();
        let later_dead = key_for(b"dead while the consult fails");
        plant_kv(&store, &later_dead, b"dead while the consult fails").await;

        let aborted = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(aborted.aborted, Some(GcAbortCause::ProtectFailed));
        assert_eq!(aborted.candidates_staged, 0);
        assert_eq!(aborted.deleted, 0);
        assert_eq!(aborted.cancelled_by_arbitration, 0);
        assert!(
            store.has(&staged_dead.as_bytes()).await.unwrap(),
            "the overdue candidate was not executed past the abort"
        );
        assert!(
            store.has(&later_dead.as_bytes()).await.unwrap(),
            "and nothing was deleted"
        );
        assert_eq!(
            window_drain(&store).await,
            vec![staged_dead],
            "the overdue row still waits in the window"
        );
    }

    /// Contention on the single-node lease aborts the second sweeper
    /// as report data on the durable form too — nothing executed,
    /// nothing staged, nothing deleted (`Ok(SweepReport)`, ADR-012 §1).
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn durable_lease_contention_aborts_the_second_sweeper_as_data() {
        let site = temp_site("aborts-lease");
        let store = Arc::new(
            Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
                .await
                .expect("kv-only sqlite constructs"),
        );
        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::<()>();
        store.register_protect_callback(Arc::new(LatchProtect {
            entered: entered_tx,
            release: parking_lot::Mutex::new(Some(release_rx)),
        }));

        let dead = key_for(b"dead under the contended sweeps");
        plant_kv(&store, &dead, b"dead under the contended sweeps").await;

        let first = {
            let store = Arc::clone(&store);
            tokio::spawn(async move { store.sweep().await })
        };

        while entered_rx
            .recv_timeout(std::time::Duration::from_millis(500))
            .is_err()
        {
            tokio::task::yield_now().await;
        }

        let second = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(second.aborted, Some(GcAbortCause::SweeperLock));
        assert_eq!(second.deleted, 0);
        assert_eq!(second.candidates_staged, 0, "nothing staged");
        assert!(store.has(&dead.as_bytes()).await.unwrap());
        assert!(window_drain(&store).await.is_empty(), "nothing executed");

        release_tx.send(()).ok();
        let first = first
            .await
            .expect("first sweep task joined")
            .expect("first sweep returns a report");
        assert_eq!(first.aborted, None);
        assert_eq!(first.candidates_staged, 1, "the lease holder stages");
        assert_eq!(first.deleted, 0, "and executes nothing at staging time");
        assert!(
            store.has(&dead.as_bytes()).await.unwrap(),
            "the lease holder staged, did not execute"
        );
    }
}

/// 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::hashing::Algorithm;
    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);

    use std::sync::atomic::AtomicBool;

    fn env_nonempty(name: &str) -> Option<String> {
        std::env::var(name).ok().filter(|v| !v.is_empty())
    }

    fn harness_dsn() -> Option<String> {
        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<u8>, Vec<u8>, i64)>;

    struct TestSite {
        dsn: String,
        schema: String,
        _janitor: SchemaJanitor,
    }

    async fn fresh_site(tag: &str) -> Option<TestSite> {
        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::from(Algorithm::GitBlobSha256.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]) {
        store
            .fleet_host()
            .expect("these tests construct postgres fleet stores")
            .put(key, value)
            .await
            .expect("kv plant");
    }

    /// 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 = store
            .fleet_host()
            .expect("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<u8>>(0),
                        row.get::<_, Vec<u8>>(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 drop_liveness_table(&self, schema: &str, table: &str) {
            let sql = format!("DROP TABLE {}", qualified_table(schema, table));
            self.client
                .execute(sql.as_str(), &[])
                .await
                .expect("embedder DDL drop");
        }

        /// Take the `FOR UPDATE` finger-lock's row from an independent
        /// session (in a `BEGIN` that [`Self::release`] ends) — the
        /// mid-delete hold: every delete-time consult on the key blocks
        /// on this lock with its marker already inserted.
        async fn finger_lock(&self, schema: &str, key: &[u8]) {
            let sql = format!(
                "SELECT 1 FROM {} WHERE key = $1 AND expiry > $2 FOR UPDATE LIMIT 1",
                qualified_table(schema, "gc_pins")
            );
            let now = crate::pg_common::time_to_nanos(SystemTime::now());
            self.client
                .batch_execute("BEGIN")
                .await
                .expect("finger-lock begin");
            let hit = self
                .client
                .query_opt(sql.as_str(), &[&key, &now])
                .await
                .expect("finger lock");
            assert!(
                hit.is_some(),
                "the finger-lock instrument requires an un-expired pin row"
            );
        }

        /// Hold an `ACCESS EXCLUSIVE` table lock (in a `BEGIN` that
        /// [`Self::release`] ends) — the mid-delete hold for a delete
        /// with no pin row present: the delete-time consult's first
        /// table touch blocks on the lock, marker already inserted.
        async fn lock_table_access_exclusive(&self, schema: &str, table: &str) {
            let sql = format!(
                "LOCK TABLE {} IN ACCESS EXCLUSIVE MODE",
                qualified_table(schema, table)
            );
            self.client
                .batch_execute("BEGIN")
                .await
                .expect("lock begin");
            self.client
                .batch_execute(sql.as_str())
                .await
                .expect("lock table");
        }

        /// End the holding transaction — releasing the held lock.
        async fn release(&self) {
            self.client
                .batch_execute("COMMIT")
                .await
                .expect("lock release");
        }

        /// Expire every pin row on a key — the drop-the-guard aftermath:
        /// a dropped in-process guard stops its renewal but the durable
        /// row lives out its TTL, so the delete-able assertions expire it
        /// explicitly (the abandonment the sweeper's maintenance step
        /// would reap).
        async fn expire_pin_rows(&self, schema: &str, key: &[u8]) {
            let sql = format!(
                "UPDATE {} SET expiry = 1 WHERE key = $1",
                qualified_table(schema, "gc_pins")
            );
            let updated = self
                .client
                .execute(sql.as_str(), &[&key])
                .await
                .expect("pin row expiry");
            assert!(updated > 0, "the expiry instrument requires a pin row");
        }

        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<BTreeSet<Key>>,
    }

    impl LivenessSource for SetSource {
        fn live_keys(&self) -> Vec<Key> {
            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<Option<std::sync::mpsc::Receiver<()>>>,
    }

    impl ProtectCallback for LatchProtect {
        fn protect(&self) -> Option<Vec<Key>> {
            self.entered.send(()).ok();
            let receiver = self.release.lock().take();
            if let Some(receiver) = receiver {
                let _ = receiver.recv();
            }
            Some(Vec::new())
        }
    }

    /// A protect callback whose first consult panics — the sweep
    /// body's panic route (review-001 N-6's plausible route) with the
    /// consult already running under both leases. It panics exactly
    /// once; the recovery sweeps consult it clean.
    struct PanicOnceProtect {
        panicking: AtomicBool,
    }

    impl ProtectCallback for PanicOnceProtect {
        fn protect(&self) -> Option<Vec<Key>> {
            if self.panicking.swap(false, Ordering::SeqCst) {
                panic!("the protect consult panicked");
            }
            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());
    }

    /// The fleet executor's drain (review-001 N-3): a pre-staged overdue
    /// backlog larger than [`WINDOW_TAKE_BATCH`] executes entirely in one
    /// sweep — the executor loops the bounded take until drained — and
    /// the report's counts stay exact over the backlog (deleted == the
    /// backlog, nothing cancelled, nothing newly staged), the second
    /// sweep all-zero.
    #[tokio::test]
    async fn fleet_backlog_larger_than_the_take_batch_drains_in_one_sweep() {
        let Some(site) = fresh_site("fleet-backlog-drain").await else {
            skip_note();
            return;
        };
        let store = Arc::new(fleet_instance(&site, "fleet-backlog-drain", LONG_TTL).await);
        let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
            SetSource::default(),
        )));

        let total = WINDOW_TAKE_BATCH + 13;
        let horizon = SystemTime::now()
            .checked_sub(Duration::from_secs(1))
            .expect("the test horizon precedes now");
        let mut staged = Vec::new();
        for i in 0..total {
            let content = format!("fleet-backlog-{i:05}");
            let key = key_for(content.as_bytes());
            plant_kv(&store, &key.key_bytes(), content.as_bytes()).await;
            staged.push(WindowCandidate {
                key: key.key_bytes().to_vec(),
                visible_after: horizon,
            });
        }
        EngineWindowState::window_stage(&**store.fleet_host().unwrap(), &staged)
            .await
            .unwrap();

        let drained = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(drained.aborted, None);
        assert_eq!(
            drained.deleted, total as u64,
            "the executor looped the bounded take until drained"
        );
        assert_eq!(drained.cancelled_by_arbitration, 0);
        assert_eq!(
            drained.candidates_staged, 0,
            "the drained pool has nothing newly dead"
        );

        let second = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(second.aborted, None);
        assert_eq!(second.deleted, 0);
        assert_eq!(second.cancelled_by_arbitration, 0);
        assert_eq!(second.candidates_staged, 0);
    }

    /// 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.key_bytes(), content).await;

        let foreign_owner: Vec<u8> = (0u8..32).collect();
        let expiry = SystemTime::now() + Duration::from_millis(300);
        mint_pin_row(&store, &key.key_bytes(), &foreign_owner, expiry).await;

        assert_eq!(
            store.delete(&key.key_bytes()).await.unwrap_err(),
            Error::GcRefuse,
            "the foreign owner's un-expired pin row refuses: the cross-instance protection"
        );
        assert!(
            store.has(&key.key_bytes()).await.unwrap(),
            "nothing deleted"
        );

        tokio::time::sleep(Duration::from_millis(400)).await;
        store
            .delete(&key.key_bytes())
            .await
            .expect("the expired foreign row no longer protects");
        assert!(!store.has(&key.key_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_as(
                Algorithm::GitBlobSha256,
                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.key_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_as(
                Algorithm::GitBlobSha256,
                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.key_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.key_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}"
        );
    }

    // The review-001 F-1 regression round. The consult-failure
    // injection mechanism is the **dropped registered liveness table**
    // — the SQL `arbitrate` folds the un-expired pin rows and each
    // registered table reference under one finger-lock; dropping the
    // embedder's table mid-flight makes the consult fail with a real
    // engine error (`relation … does not exist`), exercising the exact
    // operational-error arm on both the direct-delete path and the
    // sweep executor. The contended-pin round drives gc.rs's never-
    // tested wait neighborhood (`acquire`'s notify loop and
    // `arbitrate`'s `Plan::Wait` loop with its post-wait re-checks)
    // through the finger-lock's controlled mid-delete window.

    /// The wait-latch bound: tests that block a pin mint or an
    /// arbitration on GC's notify loops bound the wait explicitly, so
    /// the wedged-marker regression fails as a panic, never a hang.
    const LATCH: Duration = Duration::from_secs(10);

    async fn latch<T>(what: &'static str, fut: impl std::future::Future<Output = T>) -> T {
        match tokio::time::timeout(LATCH, fut).await {
            Ok(value) => value,
            Err(_) => panic!("{what}: the GC wait-latch expired"),
        }
    }

    /// A protect callback that fails to answer — the arbitration-time
    /// conservative-refusal arm's instrument (gc.rs's `protected`
    /// break: a `None` consult composes protection, nothing deleted).
    struct RefuseProtect;

    impl ProtectCallback for RefuseProtect {
        fn protect(&self) -> Option<Vec<Key>> {
            None
        }
    }

    /// The SQL-arbitration consult fails mid-delete (the registered
    /// liveness table dropped — a transient, real engine error on the
    /// direct path): the error propagates **and** the in-flight delete
    /// marker concludes first — the key stays put-able (the re-put's
    /// pin mint proceeds past no wedged marker) and delete-able (the
    /// consult succeeding again deletes). The F-1 wedge — marker held
    /// forever, every mint and arbitration blocked — would fail these
    /// assertions by latch expiry.
    #[tokio::test]
    async fn consult_failure_on_the_direct_path_concludes_the_marker_and_keeps_the_key_operable() {
        let Some(site) = fresh_site("consult-fail").await else {
            skip_note();
            return;
        };
        let store = Arc::new(fleet_instance(&site, "consult-fail", 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 _ = store.register_liveness_source(LivenessRegistration::FleetTable(reference.clone()));

        let content = b"the key the failing consult must not wedge";
        let key = key_for(content);
        plant_kv(&store, &key.key_bytes(), content).await;

        probe
            .drop_liveness_table(&site.schema, "embedder_live")
            .await;
        let err = latch("the direct delete reports the consult failure", async {
            store.delete(&key.key_bytes()).await
        })
        .await
        .unwrap_err();
        assert!(
            err.to_string().contains("does not exist"),
            "the consult's engine error surfaces operationally: {err}"
        );

        let (re_key, guard) = latch("the key is put-able after the consult failure", async {
            store
                .put_as(
                    Algorithm::GitBlobSha256,
                    Some(content.len() as u64),
                    Cursor::new(content),
                )
                .await
        })
        .await
        .expect("the re-put proceeds past no wedged marker");
        assert_eq!(re_key, key);
        assert_eq!(
            store.gc().pin_count(&key),
            1,
            "no wedged marker: the pin mint landed"
        );
        drop(guard);
        assert_eq!(store.gc().pin_count(&key), 0);

        probe.expire_pin_rows(&site.schema, &key.key_bytes()).await;
        probe
            .create_liveness_table(&site.schema, "embedder_live")
            .await;
        latch("the key is delete-able after the consult failure", async {
            store.delete(&key.key_bytes()).await
        })
        .await
        .expect("the consult succeeding again, the delete proceeds");
        assert!(!store.has(&key.key_bytes()).await.unwrap());
    }

    /// The same consult failure driven through the sweep executor
    /// (`execute_overdue_window` → `arbitrated_delete`): the overdue
    /// candidate's execution aborts the sweep operationally, the
    /// marker concludes first, and the next sweeps recover — the
    /// candidate re-stages and deletes once the consult answers again.
    /// Pre-fix, the wedged marker would strand sweep 4's re-arbitration
    /// on the key (`Plan::Wait` forever) at the latch.
    #[tokio::test]
    async fn sweep_executor_consult_failure_concludes_the_marker_and_recovers() {
        let Some(site) = fresh_site("sweep-consult-fail").await else {
            skip_note();
            return;
        };
        let store = Arc::new(fleet_instance(&site, "sweep-consult-fail", 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 _ = store.register_liveness_source(LivenessRegistration::FleetTable(reference.clone()));

        let content = b"the candidate whose executor consult fails";
        let key = key_for(content);
        plant_kv(&store, &key.key_bytes(), content).await;

        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(staged.aborted, None);
        assert_eq!(staged.candidates_staged, 1);
        assert_eq!(staged.deleted, 0);

        probe
            .drop_liveness_table(&site.schema, "embedder_live")
            .await;
        let err = latch("the sweep reports the executor consult failure", async {
            store.sweep().await
        })
        .await
        .unwrap_err();
        assert!(
            err.to_string().contains("does not exist"),
            "the executor consult's engine error escapes the sweep: {err}"
        );

        probe
            .create_liveness_table(&site.schema, "embedder_live")
            .await;
        let restaged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(restaged.aborted, None);
        assert_eq!(
            (restaged.deleted, restaged.cancelled_by_arbitration),
            (0, 0),
            "the failed candidate was taken off the window before the failure; nothing \
             re-executes yet"
        );
        assert_eq!(restaged.candidates_staged, 1, "still dead, re-staged");

        let executed = latch(
            "the re-staged candidate executes past no wedged marker",
            async { store.sweep().await },
        )
        .await
        .expect("sweeps are never errors");
        assert_eq!(executed.aborted, None);
        assert_eq!(executed.deleted, 1, "the wedge-free executor recovers");
        assert!(!store.has(&key.key_bytes()).await.unwrap());
    }

    /// The `Plan::Wait`/protected contended-pin shape (never driven by
    /// any test before review-001 F-1's round): a foreign instance's
    /// un-expired pin row refuses a direct delete held mid-flight by
    /// the finger-lock (the marker is in flight across the blocked
    /// consult) — a concurrent pin mint blocked on the marker **proceeds
    /// against the live entry** when the refusal concludes, and a
    /// concurrent arbitration that waited on the marker re-checks after
    /// the release and finds the mint's protection composed (`GcRefuse`
    /// — the protected break), nothing deleted.
    #[tokio::test]
    async fn blocked_pin_mint_proceeds_against_the_live_entry_when_arbitration_refuses() {
        let Some(site) = fresh_site("contended-refuse").await else {
            skip_note();
            return;
        };
        let store = Arc::new(fleet_instance(&site, "contended-refuse", LONG_TTL).await);

        let content = b"protected mid-delete while a mint and an arbitration wait";
        let key = key_for(content);
        plant_kv(&store, &key.key_bytes(), content).await;

        let foreign_owner: Vec<u8> = (0u8..24).collect();
        mint_pin_row(
            &store,
            &key.key_bytes(),
            &foreign_owner,
            SystemTime::now() + Duration::from_secs(30),
        )
        .await;

        let lock = independent(&site.dsn).await;
        lock.finger_lock(&site.schema, &key.key_bytes()).await;

        let delete_task = {
            let store = Arc::clone(&store);
            let key = key.clone();
            tokio::spawn(async move { store.delete(&key.key_bytes()).await })
        };

        tokio::time::sleep(Duration::from_millis(300)).await;
        assert!(
            store.has(&key.key_bytes()).await.unwrap(),
            "the delete is held mid-flight by the finger-lock"
        );

        let mint_task = {
            let store = Arc::clone(&store);
            let key = key.clone();
            tokio::spawn(async move { store.gc().acquire(&key).await })
        };
        tokio::time::sleep(Duration::from_millis(150)).await;
        assert_eq!(
            store.gc().pin_count(&key),
            0,
            "the pin mint is blocked on the in-flight delete marker"
        );

        let waiter_task = {
            let store = Arc::clone(&store);
            let key = key.clone();
            tokio::spawn(async move { store.delete(&key.key_bytes()).await })
        };
        tokio::time::sleep(Duration::from_millis(150)).await;

        lock.release().await;

        assert_eq!(
            latch("the refused delete reports", delete_task)
                .await
                .expect("the delete task joined")
                .unwrap_err(),
            Error::GcRefuse,
            "the foreign row refuses: the marker concludes, the caller sees the refusal"
        );
        let guard = latch("the blocked mint proceeds", mint_task)
            .await
            .expect("the mint lands once the marker concludes");
        assert_eq!(
            store.gc().pin_count(&key),
            1,
            "the mint proceeded against the live entry"
        );
        assert!(
            store.has(&key.key_bytes()).await.unwrap(),
            "arbitration refused mid-flight: the entry stays live"
        );
        assert_eq!(
            latch("the waiter's re-check", waiter_task)
                .await
                .expect("the waiter task joined")
                .unwrap_err(),
            Error::GcRefuse,
            "the wait-loop re-check composes the mint's protection after the marker \
             released — the protected break"
        );

        drop(guard);
        assert_eq!(store.gc().pin_count(&key), 0);
    }

    /// The wait loop's conclude form: a delete held mid-flight by the
    /// finger-lock (no protection present) actually deletes — a pin
    /// mint (the re-put) blocked on the marker **waits through the
    /// engine delete and concludes correctly**: it lands as a fresh
    /// pinned put after the marker released, republishing the deleted
    /// content (ADR-005's re-put case (a) verbatim, fleet form).
    #[tokio::test]
    async fn the_delete_concluding_while_a_mint_waits_lands_it_as_a_fresh_pinned_put() {
        let Some(site) = fresh_site("contended-conclude").await else {
            skip_note();
            return;
        };
        let store = Arc::new(fleet_instance(&site, "contended-conclude", LONG_TTL).await);

        let content = b"deleted while a re-put waited on the marker";
        let key = key_for(content);
        plant_kv(&store, &key.key_bytes(), content).await;

        let lock = independent(&site.dsn).await;
        lock.lock_table_access_exclusive(&site.schema, "gc_pins")
            .await;

        let delete_task = {
            let store = Arc::clone(&store);
            let key = key.clone();
            tokio::spawn(async move { store.delete(&key.key_bytes()).await })
        };
        tokio::time::sleep(Duration::from_millis(300)).await;

        let put_task = {
            let store = Arc::clone(&store);
            let content = content.to_vec();
            tokio::spawn(async move {
                store
                    .put_as(
                        Algorithm::GitBlobSha256,
                        Some(content.len() as u64),
                        Cursor::new(&content[..]),
                    )
                    .await
            })
        };
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(
            store.gc().pin_count(&key),
            0,
            "the re-put's pin mint is blocked on the in-flight delete marker"
        );
        assert!(
            store.has(&key.key_bytes()).await.unwrap(),
            "the delete is still mid-flight, nothing deleted yet"
        );

        lock.release().await;

        latch("the delete completes", delete_task)
            .await
            .expect("the delete task joined")
            .expect("the unprotected delete proceeds and concludes");
        let (put_key, guard) = latch("the waited re-put lands", put_task)
            .await
            .expect("the put task joined")
            .expect("the mint landed once the delete concluded");
        assert_eq!(put_key, key);
        assert_eq!(store.gc().pin_count(&key), 1, "the fresh pinned put");
        assert!(
            store.has(&key.key_bytes()).await.unwrap(),
            "the delete concluded, the mint landed as a fresh pinned put"
        );
        let (len, blob) = store
            .get(&key.key_bytes())
            .await
            .expect("get succeeds")
            .expect("the re-put published");
        assert_eq!(len, content.len() as u64);
        assert_eq!(blob.read_at(0, len).await.unwrap(), content);

        drop(guard);
        assert_eq!(store.gc().pin_count(&key), 0);
    }

    /// The arbitration wait-loop's post-release re-check (driven at the
    /// gc seam, deterministically): an arbitration holding the marker
    /// (the emulated in-flight delete) blocks a concurrent direct
    /// delete in `Plan::Wait`; when the marker concludes, the waiter's
    /// re-loop re-consults protection — a protect callback that fails
    /// to answer composes the conservative refusal (`GcRefuse`, nothing
    /// deleted, no marker left); the key is immediately operable again
    /// (mint, delete) with no wedged marker.
    #[tokio::test]
    async fn the_wait_loop_re_checks_protection_after_the_marker_releases() {
        let Some(site) = fresh_site("wait-recheck").await else {
            skip_note();
            return;
        };
        let store = Arc::new(fleet_instance(&site, "wait-recheck", LONG_TTL).await);

        let content = b"the waiter whose re-loop sees protection appear";
        let key = key_for(content);
        plant_kv(&store, &key.key_bytes(), content).await;

        match store.gc().arbitrate(&key, false).await {
            Ok(Arbitration::Proceed) => {}
            Ok(Arbitration::Cancelled) => panic!("the unprotected arbitration proceeds"),
            Err(e) => panic!("the emulated arbitration errs: {e}"),
        }

        let waiter_task = {
            let store = Arc::clone(&store);
            let key = key.clone();
            tokio::spawn(async move { store.delete(&key.key_bytes()).await })
        };
        tokio::time::sleep(Duration::from_millis(300)).await;

        let token = store.register_protect_callback(Arc::new(RefuseProtect));
        store.gc().conclude_delete(&key);

        assert_eq!(
            latch("the waiter's post-release re-check", waiter_task)
                .await
                .expect("the waiter task joined")
                .unwrap_err(),
            Error::GcRefuse,
            "the re-loop recomposed protection from the failed consult: the conservative \
             refusal"
        );
        assert!(
            store.has(&key.key_bytes()).await.unwrap(),
            "nothing deleted by the waited arbitration"
        );

        assert!(store.unregister_protect_callback(token));
        let guard = latch("the mint proceeds past no wedged marker", async {
            store.gc().acquire(&key).await
        })
        .await;
        assert_eq!(store.gc().pin_count(&key), 1);
        drop(guard);
        store
            .delete(&key.key_bytes())
            .await
            .expect("the key is delete-able with no marker, no protection");
        assert!(!store.has(&key.key_bytes()).await.unwrap());
    }

    /// N-6, the fleet half: a panic inside the sweep body (the
    /// protect callback's panic route, riding the consult that runs
    /// under both leases) cannot leave either single-flight domain
    /// held — the in-process lease's guard ends the flag on the
    /// unwind, and the fleet guard's spawned release clears the
    /// advisory lock. The recovery sweep (the panicking consult
    /// consumed by its one-shot arm) completes: both acquisitions
    /// granted, nothing stuck.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn panicking_protect_cannot_leave_the_sweep_leases_held() {
        let Some(site) = fresh_site("panic-lease").await else {
            skip_note();
            return;
        };
        let store = Arc::new(fleet_instance(&site, "panic-lease", LONG_TTL).await);
        let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
            SetSource::default(),
        )));
        let protect = store.register_protect_callback(Arc::new(PanicOnceProtect {
            panicking: AtomicBool::new(true),
        }));

        let sweeper = Arc::clone(&store);
        let joined = tokio::spawn(async move { sweeper.sweep().await })
            .await
            .unwrap_err();
        assert!(
            joined.is_panic(),
            "the sweep panicked through the protect consult (the recorded panic route)"
        );
        assert!(store.unregister_protect_callback(protect));

        // The fleet lease was released by the guard's spawned release:
        // the re-acquire is granted once it lands.
        let host = Arc::clone(store.fleet_host().expect("the fleet host"));
        let mut acquired = false;
        for _ in 0..100 {
            if EngineFleetState::sweep_lock_acquire(&*host).await.unwrap() {
                acquired = true;
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert!(acquired, "the panicking sweep's advisory lease came free");
        EngineFleetState::sweep_lock_release(&*host).await.unwrap();

        let recovered = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            recovered.aborted, None,
            "both single-flight domains came free after the panic"
        );
    }
}