alkblobs 0.2.2

//! GC state machinery (ADR-005/012): the crate-internal engine-state
//! companion seam (ADR-010 §2, as reduced by ADR-012 §5) and — in wave 2 —
//! the sweep/liveness mechanism the store core drives through it.
//!
//! **The seam's shape is ADR-012 §5's per-engine applicability table,
//! expressed structurally — there is no dead machinery.** The window
//! family (GC schema + staged delete window) is `EngineWindowState`,
//! implemented by every SQL-backed engine: postgres, sqlite, pg-lo. The
//! fleet family (pin clauses + the sweep lease) is `EngineFleetState`,
//! implemented by the postgres host only — all fleet GC state hosts on
//! the fleet's kv-tier engine (ADR-012 §4), sqlite's pin/sweep-lock
//! bodies were dropped as dead machinery (ADR-012 §5), and a sqlite
//! fleet is a recorded non-option (ADR-010 §3). `local` and `mem` are
//! excluded by construction: they stage delete-window candidates
//! in-process and execute the ADR-005 protocol inline within the same
//! sweep, and they are never fleet engines.
//!
//! **Ownership rule (ADR-010 §2, verbatim posture):** the store core owns
//! all GC state — the in-process pin and liveness maps and the fleet
//! state's rows and locks. Engines execute clauses; the store decides.
//! The public-side `Backend` trait (module [`crate::backend`]) stays
//! "plain content-addressed rows, nothing else": pins, window rows, and
//! the sweep lease are invisible at the public trait, and backends never
//! learn liveness (ADR-005). Every clause is contract-tested per its
//! applicability row in waves 3-5; `local`/`mem` are contract-excluded.
//!
//! **The in-process machinery (this module's `GcHome`).** The whole
//! in-process GC state sits behind **one arbitration lock** — a
//! short-held [`parking_lot::Mutex`] over the per-key pin refcounts,
//! the in-flight delete markers, and the registered liveness/protect
//! source lists (ADR-005's "the arbitration lock the sweep consults";
//! store-api.md §Concurrency posture; AGENTS convention 3). parking_lot
//! does not poison, and these sections hold no `RwLock` other lock
//! kind, so there is no `unwrap_or_else(|e| e.into_inner())` recovery
//! posture to apply (the mem engine's module doc records the same
//! posture). Consumer consults ([`LivenessSource::contains`]/
//! [`ProtectCallback::protect`], the trait shapes on
//! [`crate::facade`]) run under that lock at delete-time arbitration
//! and outside it at mark time — both documented as quick, non-blocking
//! duties on the traits. Lock order is one-directional everywhere:
//! arbitration lock → consumer locks (the ordered-step conversion
//! primitive is the only nesting, same direction).
//!
//! **Named design points (recorded for the review gate):**
//!
//! - **The sweep-vs-put race is closed with in-flight delete markers,**
//!   not by holding the arbitration lock across engine I/O. A sweep's
//!   (or direct delete's) arbitration decides *and* inserts the
//!   key's delete marker inside one lock section; the engine delete
//!   runs outside the lock; the marker removal notifies any put-path
//!   pin mint blocked on the key. A re-put whose pin mint lands while
//!   the marker is held blocks until the deletion commits, then lands
//!   as a fresh pinned put — ADR-005's case (a) delete-then-recover
//!   verbatim; a pin mint that lands before the arbitration is case
//!   (b), its protection composed at the arbitration itself. Holding
//!   the parking_lot lock across the engine delete was rejected: the
//!   guard is not `Send`, the sweep future must be, and AGENTS
//!   convention 3 forbids blocking I/O under internal locks.
//! - **The single-node sweep is single-flight** — an in-process
//!   test-and-set flag is the single-node analog of ADR-013 §2's named
//!   exclusive sweep lease (ADR-012 §5's "single-flighter is the
//!   in-process lock" posture): one sweeper per arbitration domain,
//!   atomic test-and-set, holdership ends at release; a contended
//!   second sweeper aborts with [`GcAbortCause::SweeperLock`][crate::GcAbortCause::SweeperLock] as
//!   report data (`Ok(SweepReport)` — ADR-012 §1), nothing deleted.
//!
//! The `#![allow(dead_code)]` below covers the companion-seam scope:
//! the window family has its in-tree callers (store construction runs
//! `ensure_gc_schema` on the window's host — the kv engine when it is
//! SQL-backed, pg-lo when the kv engine is not (the one-window-host
//! rule, ADR-012 §5); the sweep's durable form drives
//! `window_stage`/`window_take` on that same host), and the fleet
//! family's consumers have all landed: the joint-tx put driver
//! (`store-fleet-put`: the store core's `begin_joint_tx` → engine
//! contribution → `pin_commit` flow, plus the pin guard's TTL/3
//! renewal task minted there) and the fleet sweep (`store-fleet-sweep`
//! — the sweep lease, `pin_reap`, and the SQL-level arbitration verbs
//! are driven: the sweep runs under the lease, the delete paths fold
//! the SQL arbitration behind the in-process domain, the maintenance
//! step reaps expired pin rows). What the allow still covers: the
//! fleet trait and its registration bookkeeping in non-postgres
//! builds (no implementer or consumer compiles without feature
//! `postgres`), the seam's `pin_check` clause — the pins-only
//! classifier, whose consumer has landed (wave 7's
//! ops-foundation: the ops module's renewal probe reaches it through
//! the store core's crate-internal `ops_pin_check` crossing, the
//! have handler's renewal-verification step), and the ADR-005
//! conversion-ordering primitive (`GcHome::convert`/
//! `RegistryCore::convert`) plus the
//! `GcHome::pin_count` observation aid — the conversion's
//! guard-custody consumer has also landed (the ops module's confirm
//! verb converts held guards at the embedder's confirmed
//! registration): in the default build `convert`/`pin_count` are
//! test-consumed; in the ops build the confirm verb is the
//! conversion-ordering primitive's non-test consumer.

#![allow(dead_code)]

use std::collections::{BTreeMap, BTreeSet};
use std::pin::pin;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::SystemTime;

use parking_lot::Mutex;
use tokio::sync::Notify;

use crate::backend::Backend;
use crate::facade::{
    Batch, LivenessSource, Pin, PinState, ProtectCallback, Registration, RegistrationKind,
};
use crate::keys::Key;
use crate::Error;

/// The store-instance GC state home — held as a field of the public
/// facade type ([`Store`](crate::store::Store)). ADR-011 §1's vocabulary
/// and ADR-010 §2's ownership rule together: the store instance owns
/// all GC state, so in-process GC state carries on the facade
/// instance itself — the pin registry (per-key refcounts under the
/// arbitration lock), registered liveness sources, the protect
/// callbacks, and the in-process delete window (ADR-005's protocol
/// inline, ADR-012 §5's non-SQL staging), shared with the public pin
/// guards ([`Pin`]/[`Batch`]) through the [`RegistryCore`] `Arc`.
/// Fleet state never lands here: it rides the engine-state companion
/// seam below (ADR-010 §2), hosted on the fleet's kv engine (ADR-012 §4).
pub(crate) struct GcHome {
    core: Arc<RegistryCore>,
}

impl Default for GcHome {
    fn default() -> GcHome {
        GcHome {
            core: Arc::new(RegistryCore::new()),
        }
    }
}

impl GcHome {
    /// Mint a pin: acquire the key's refcount under the arbitration
    /// lock, blocking (async) while a delete of the key is in flight —
    /// ADR-005's re-put case (a): the mint waits until the deletion
    /// commits, then the guard protects the fresh put. The put path
    /// acquires this **before** publishing (pin-before-publish).
    pub(crate) async fn acquire(&self, key: &Key) -> Pin {
        self.core.acquire(key).await
    }

    /// A fresh batch-scoped pin group (ADR-005 §Marks 2): member pins
    /// release together in one ordered step on commit or drop.
    pub(crate) fn new_batch(&self) -> Batch {
        Batch::created(Arc::clone(&self.core))
    }

    /// The registry core Arc behind this home — the batching put
    /// seam's foreign-Batch identity check (review-001 N-5) compares
    /// it against the batch's.
    pub(crate) fn core(&self) -> &Arc<RegistryCore> {
        &self.core
    }

    /// The **conversion ordering primitive** (ADR-005, pinned): the
    /// pin's release and the replacement closure's registration into a
    /// liveness source are one ordered step under the same arbitration
    /// lock — a consumer may not release a pin before its registered
    /// source observes the replacement. Consumes the pin (its own drop
    /// is suppressed; the release happens inside the section, before
    /// the closure runs).
    pub(crate) fn convert<F: FnOnce()>(&self, pin: Pin, register: F) {
        self.core.convert(pin, register)
    }

    /// The registered liveness source count — the sweep's
    /// [`GcAbortCause::NoLivenessSources`] safe default keys off zero.
    pub(crate) fn registered_source_count(&self) -> usize {
        self.core.registered_source_count()
    }

    /// Register a [`LivenessSource`] — the live-shared seam; the token
    /// deregisters it.
    pub(crate) fn register_source(&self, source: Arc<dyn LivenessSource>) -> Registration {
        self.core.register_source(source)
    }

    /// Record a registered **fleet liveness table reference** (ADR-012
    /// §6.6's fleet form, the store-side half): the token deregisters
    /// it via [`Self::unregister_fleet_source`]. The engine-side
    /// forwarding has already succeeded when this runs.
    pub(crate) fn register_fleet_source(&self, reference: String) -> Registration {
        self.core.register_fleet_source(reference)
    }

    /// Remove the fleet registration record for a fleet-source token,
    /// handing back the named reference for the engine-side removal.
    /// `None` when the token is not a fleet-source token or was
    /// already consumed.
    pub(crate) fn unregister_fleet_source(&self, registration: Registration) -> Option<String> {
        self.core.unregister_fleet_source(registration)
    }

    /// The registered fleet liveness table count — the sweep's safe
    /// default counts these beside the in-process sources.
    pub(crate) fn fleet_source_count(&self) -> usize {
        self.core.fleet_source_count()
    }

    /// The registered liveness registrations' full count — in-process
    /// sources plus fleet tables (the sweep's [`GcAbortCause::NoLivenessSources`]
    /// safe default keys off zero).
    pub(crate) fn registered_liveness_count(&self) -> usize {
        self.registered_source_count() + self.fleet_source_count()
    }

    /// Deregister a previously registered liveness source; `false` when
    /// the token is already consumed or was never a source token.
    pub(crate) fn unregister_source(&self, registration: Registration) -> bool {
        self.core.unregister_source(registration)
    }

    /// Register a [`ProtectCallback`]; the token deregisters it.
    pub(crate) fn register_protect(&self, callback: Arc<dyn ProtectCallback>) -> Registration {
        self.core.register_protect(callback)
    }

    /// Deregister a previously registered protect callback; `false`
    /// when the token is already consumed or was never a protect token.
    pub(crate) fn unregister_protect(&self, registration: Registration) -> bool {
        self.core.unregister_protect(registration)
    }

    /// Snapshots of the registered liveness sources — cloned under the
    /// arbitration lock, consulted outside it (mark-time enumeration
    /// may be slow; the delete-time probe path re-reads the live list
    /// under the lock instead).
    pub(crate) fn registered_sources(&self) -> Vec<Arc<dyn LivenessSource>> {
        self.core.registered_sources()
    }

    /// Snapshot of the registered protect callbacks — consulted
    /// pre-sweep, outside the arbitration lock.
    pub(crate) fn registered_protects(&self) -> Vec<Arc<dyn ProtectCallback>> {
        self.core.registered_protects()
    }

    /// The keys currently carrying pins (refcount > 0) — the pin half
    /// of the mark-phase live set.
    pub(crate) fn pinned_keys(&self) -> Vec<Key> {
        self.core.pinned_keys()
    }

    /// The key's current pin refcount (a testing/put-path observation
    /// aid; zero when unpinned).
    pub(crate) fn pin_count(&self, key: &Key) -> u64 {
        self.core.pin_count(key)
    }

    /// The single-node sweep lease: `true` = acquired (this caller
    /// sweeps); `false` = contended — the caller aborts with
    /// [`GcAbortCause::SweeperLock`] report data.
    pub(crate) fn try_begin_sweep(&self) -> bool {
        self.core.try_begin_sweep()
    }

    /// Release the single-node sweep lease (ADR-013 §2's holdership
    /// rule, in-process: holdership ends at release).
    pub(crate) fn end_sweep(&self) {
        self.core.end_sweep()
    }

    /// Take the sweep lease — acquiring is [`Self::try_begin_sweep`]'s
    /// atomic test-and-set; the returned guard (the lease's
    /// release-on-every-path form, review-001 N-6) ends it on **every**
    /// exit from the sweep scope, unwind (a panicking protect callback)
    /// and cancellation (an aborted sweep task) included — the
    /// stick-`true`-forever alternative was the recorded gap.
    pub(crate) fn begin_sweep_guarded(&self) -> Option<SweepLeaseGuard<'_>> {
        if self.try_begin_sweep() {
            Some(SweepLeaseGuard { home: self })
        } else {
            None
        }
    }

    /// **Delete-time arbitration** (ADR-005 §Sweep step 2). Decides the
    /// key's fate in one arbitration-lock section — pins, in-flight
    /// markers (wait and retry), and — per the `consult_protect` flag —
    /// the registered liveness/protect sources' delete-time probes:
    ///
    /// - [`Arbitration::Cancelled`] — protection composed over the
    ///   candidate (pinned, re-observed live, protect-named/failed);
    ///   nothing deleted, no marker left.
    /// - [`Arbitration::Proceed`] — the key's delete marker is now
    ///   held; the caller **must** run [`Self::conclude_delete`] on
    ///   every path after its engine delete (success, failure, or
    ///   absent) — the marker-release notifies blocked pin mints.
    pub(crate) async fn arbitrate(
        &self,
        key: &Key,
        consult_protect: bool,
    ) -> Result<Arbitration, Error> {
        self.core.arbitrate(key, consult_protect).await
    }

    /// Release the delete marker taken by a [`Arbitration::Proceed`]
    /// arbitration and notify blocked pin mints. Mandatory on every
    /// post-arbitration path.
    pub(crate) fn conclude_delete(&self, key: &Key) {
        self.core.conclude_delete(key)
    }
}

/// The arbitration vocabulary between a delete and a pin mint: the
/// delete is cancelled by protection, or it proceeds holding the key's
/// in-flight delete marker.
pub(crate) enum Arbitration {
    /// Protection composed over the key at delete time (ADR-005's
    /// "pinned or re-observed live, cancel the candidate").
    Cancelled,
    /// The delete marker is held; run the engine delete, then
    /// [`GcHome::conclude_delete`].
    Proceed,
}

/// The outcome of a store-side arbitrated delete: cancelled by
/// protection (nothing deleted), or executed — with whether the entry
/// was still present when the window reached it.
pub(crate) enum DeleteOutcome {
    Cancelled,
    Deleted(bool),
}

/// The in-process sweep lease's guard (review-001 N-6's drop-deferred
/// release): created only by [`GcHome::begin_sweep_guarded`], it owns
/// the lease for the sweep scope and ends it on every exit — normal
/// return, unwind, cancellation — not just the awaited return. The
/// fleet lease's twin (its Drop spawns the release past the synchronous
/// body) rides the store sweep driver.
pub(crate) struct SweepLeaseGuard<'a> {
    home: &'a GcHome,
}

impl Drop for SweepLeaseGuard<'_> {
    fn drop(&mut self) {
        self.home.end_sweep();
    }
}

/// The GC registry core — the shared state behind [`GcHome`], the
/// public pin guards' back-reference ([`Pin`] holds this `Arc`), and
/// **the arbitration lock's owner**. One [`parking_lot::Mutex`] guards
/// the per-key pin refcounts, the in-flight delete markers, and the
/// registered source lists together, so a delete-time decision (pins /
/// markers / sources) and its marker insert are one atomic section
/// against the put path's pin mints.
pub(crate) struct RegistryCore {
    /// The arbitration lock's state (see the module doc).
    state: Mutex<RegistryState>,
    /// Wakeup for pin mints blocked on an in-flight delete marker.
    delete_done: Notify,
    /// The single-node sweep lease (the in-process analog of ADR-013
    /// §2's named exclusive sweep lease).
    sweeping: AtomicBool,
    /// Registration id allocation.
    next_registration: AtomicU64,
}

struct RegistryState {
    /// Per-key pin refcounts (ADR-005 §Marks 2); zero-count keys are
    /// removed, so the map is exactly the pinned-key set at mark time.
    pins: BTreeMap<Key, u64>,
    /// In-flight delete markers: keys between an arbitration's Proceed
    /// decision and its marker release. Pin mints block on these
    /// (ADR-005's re-put case (a)).
    deleting: BTreeSet<Key>,
    /// Registered liveness sources with their registration ids.
    sources: Vec<(u64, Arc<dyn LivenessSource>)>,
    /// Registered protect callbacks with their registration ids.
    protects: Vec<(u64, Arc<dyn ProtectCallback>)>,
    /// Registered **fleet liveness table references** (ADR-012 §6.6's
    /// fleet form) with their registration ids — the store-side
    /// bookkeeping beside the in-process registrations (the token's
    /// id lives here; the named table itself is never owned by the
    /// store core — the engine holds the quoted reference, the
    /// schema-ownership rule keeps the table the embedder's). The
    /// count feeds the sweep's safe default; deregistration removes
    /// the record and hands the reference back for the engine-side
    /// removal (the correction flow, ADR-012 §6.4).
    fleet_sources: Vec<(u64, String)>,
}

impl RegistryCore {
    fn new() -> RegistryCore {
        RegistryCore {
            state: Mutex::new(RegistryState {
                pins: BTreeMap::new(),
                deleting: BTreeSet::new(),
                sources: Vec::new(),
                protects: Vec::new(),
                fleet_sources: Vec::new(),
            }),
            delete_done: Notify::new(),
            sweeping: AtomicBool::new(false),
            next_registration: AtomicU64::new(1),
        }
    }

    fn allocate_registration(&self) -> u64 {
        self.next_registration.fetch_add(1, Ordering::Relaxed)
    }

    pub(crate) fn register_source(&self, source: Arc<dyn LivenessSource>) -> Registration {
        let id = self.allocate_registration();
        self.state.lock().sources.push((id, source));
        Registration {
            id,
            kind: RegistrationKind::Source,
        }
    }

    pub(crate) fn unregister_source(&self, registration: Registration) -> bool {
        self.unregister(&registration, RegistrationKind::Source)
    }

    pub(crate) fn register_protect(&self, callback: Arc<dyn ProtectCallback>) -> Registration {
        let id = self.allocate_registration();
        self.state.lock().protects.push((id, callback));
        Registration {
            id,
            kind: RegistrationKind::Protect,
        }
    }

    pub(crate) fn unregister_protect(&self, registration: Registration) -> bool {
        self.unregister(&registration, RegistrationKind::Protect)
    }

    /// Record a registered **fleet liveness table reference** (the
    /// store-side half of the fleet registration verb — the engine-side
    /// forwarding already succeeded): a fresh registration id beside
    /// the in-process sources', the reference carried for the
    /// deregistration flow. Re-registering the same reference returns
    /// the original token — idempotent, mirroring the engine-side
    /// registration's discipline.
    pub(crate) fn register_fleet_source(&self, reference: String) -> Registration {
        let mut state = self.state.lock();
        if let Some((id, _)) = state
            .fleet_sources
            .iter()
            .find(|(_, held)| *held == reference)
        {
            return Registration {
                id: *id,
                kind: RegistrationKind::FleetSource,
            };
        }
        let id = self.allocate_registration();
        state.fleet_sources.push((id, reference));
        Registration {
            id,
            kind: RegistrationKind::FleetSource,
        }
    }

    /// Remove the fleet registration record for a
    /// [`RegistrationKind::FleetSource`] token, handing back the named
    /// reference for the engine-side removal (the correction flow,
    /// ADR-012 §6.4). `None` when the token is not a fleet-source
    /// token or was already consumed.
    pub(crate) fn unregister_fleet_source(&self, registration: Registration) -> Option<String> {
        let mut state = self.state.lock();
        if registration.kind != RegistrationKind::FleetSource {
            return None;
        }
        let position = state
            .fleet_sources
            .iter()
            .position(|(id, _)| *id == registration.id)?;
        Some(state.fleet_sources.remove(position).1)
    }

    /// The registered fleet liveness table count — the sweep's safe
    /// default counts these beside the in-process sources (a fleet
    /// embedder may register nothing but its table).
    pub(crate) fn fleet_source_count(&self) -> usize {
        self.state.lock().fleet_sources.len()
    }

    fn unregister(&self, registration: &Registration, kind: RegistrationKind) -> bool {
        let mut state = self.state.lock();
        match kind {
            RegistrationKind::Source => {
                let before = state.sources.len();
                state.sources.retain(|(id, _)| *id != registration.id);
                state.sources.len() != before
            }
            RegistrationKind::Protect => {
                let before = state.protects.len();
                state.protects.retain(|(id, _)| *id != registration.id);
                state.protects.len() != before
            }
            RegistrationKind::FleetSource => {
                let _ = registration;
                false
            }
        }
    }

    pub(crate) fn registered_source_count(&self) -> usize {
        self.state.lock().sources.len()
    }

    fn registered_sources(&self) -> Vec<Arc<dyn LivenessSource>> {
        self.state
            .lock()
            .sources
            .iter()
            .map(|(_, s)| Arc::clone(s))
            .collect()
    }

    fn registered_protects(&self) -> Vec<Arc<dyn ProtectCallback>> {
        self.state
            .lock()
            .protects
            .iter()
            .map(|(_, c)| Arc::clone(c))
            .collect()
    }

    pub(crate) fn pinned_keys(&self) -> Vec<Key> {
        self.state.lock().pins.keys().cloned().collect()
    }

    /// The key's current pin refcount (a testing/put-path observation
    /// aid; zero when unpinned).
    pub(crate) fn pin_count(&self, key: &Key) -> u64 {
        self.state.lock().pins.get(key).copied().unwrap_or(0)
    }

    pub(crate) async fn acquire(self: &Arc<Self>, key: &Key) -> Pin {
        loop {
            let notified = self.delete_done.notified();
            let mut notified = pin!(notified);
            notified.as_mut().enable();
            {
                let mut state = self.state.lock();
                if !state.deleting.contains(key) {
                    *state.pins.entry(key.clone()).or_insert(0) += 1;
                    return Pin::minted(Arc::clone(self), key.clone());
                }
            }
            notified.as_mut().await;
        }
    }

    pub(crate) fn release_pin(&self, key: &Key) {
        let mut state = self.state.lock();
        Self::release_in(&mut state, key);
    }

    /// The batch's one-step release: every member refcount decremented
    /// inside one arbitration-lock section, in the keys' total order
    /// (ADR-005: no early single-pin release).
    pub(crate) fn release_pins(&self, mut keys: Vec<Key>) {
        keys.sort();
        let mut state = self.state.lock();
        for key in &keys {
            Self::release_in(&mut state, key);
        }
    }

    fn release_in(state: &mut RegistryState, key: &Key) {
        if let Some(count) = state.pins.get_mut(key) {
            *count -= 1;
            if *count == 0 {
                state.pins.remove(key);
            }
        }
    }

    /// The ordered-step conversion (ADR-005's pinned conversion
    /// ordering): within one arbitration-lock section, release the
    /// pin's refcount and run the registration closure. A concurrent
    /// arbitration therefore never observes neither-pin-nor-registered.
    /// Consumes the pin with its own drop suppressed (the release ran
    /// inside the section).
    pub(crate) fn convert<F: FnOnce()>(&self, pin: Pin, register: F) {
        #[cfg(feature = "postgres")]
        let pin = {
            let mut pin = pin;
            pin.stop_renewal();
            pin
        };
        let pin = std::mem::ManuallyDrop::new(pin);
        let key = pin.key().clone();
        let mut state = self.state.lock();
        Self::release_in(&mut state, &key);
        register();
    }

    pub(crate) fn try_begin_sweep(&self) -> bool {
        self.sweeping
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_ok()
    }

    pub(crate) fn end_sweep(&self) {
        self.sweeping.store(false, Ordering::Release);
    }

    pub(crate) async fn arbitrate(
        &self,
        key: &Key,
        consult_protect: bool,
    ) -> Result<Arbitration, Error> {
        loop {
            let notified = self.delete_done.notified();
            let mut notified = pin!(notified);
            notified.as_mut().enable();

            enum Plan {
                Cancel,
                Wait,
                Proceed,
            }

            let plan = {
                let mut state = self.state.lock();
                if state.pins.contains_key(key) {
                    Plan::Cancel
                } else if state.deleting.contains(key) {
                    Plan::Wait
                } else {
                    let mut protected = false;
                    if consult_protect {
                        for (_, callback) in &state.protects {
                            match callback.protect() {
                                Some(keys) if keys.iter().any(|live| live == key) => {
                                    protected = true;
                                    break;
                                }
                                None => {
                                    // A failed protection source refuses
                                    // conservatively: nothing deleted.
                                    protected = true;
                                    break;
                                }
                                Some(_) => {}
                            }
                        }
                    }
                    if !protected {
                        for (_, source) in &state.sources {
                            if source.contains(key) {
                                protected = true;
                                break;
                            }
                        }
                    }
                    if protected {
                        Plan::Cancel
                    } else {
                        state.deleting.insert(key.clone());
                        Plan::Proceed
                    }
                }
            };

            match plan {
                Plan::Cancel => return Ok(Arbitration::Cancelled),
                Plan::Proceed => return Ok(Arbitration::Proceed),
                Plan::Wait => notified.as_mut().await,
            }
        }
    }

    pub(crate) fn conclude_delete(&self, key: &Key) {
        let removed = self.state.lock().deleting.remove(key);
        if removed {
            self.delete_done.notify_waiters();
        }
    }
}

/// A delete-window candidate staged into the durable window: the key and
/// the time it becomes visible for deletion — exactly the
/// `gc_window (key, visible_after)` row pair (ADR-005's delete-window
/// protocol; ADR-008 §2).
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct WindowCandidate {
    /// The opaque key bytes (the seam's boundary discipline: opaque byte
    /// keys, like the `Backend` trait).
    pub key: Vec<u8>,
    /// When the candidate becomes deletable (the staged window's
    /// expiry; the calling sweep passes `now` to [`EngineWindowState::window_take`]
    /// and executes whatever has become overdue).
    pub visible_after: SystemTime,
}

/// A fleet pin row to commit inside the joint entry+pin transaction —
/// exactly the `gc_pins (key, owner, expiry)` row triple (ADR-008;
/// ADR-010 §2). `owner` is the **store-instance id** (ADR-011's
/// vocabulary): the row's identity is `(key, owner)`, a dedup re-put by
/// a different instance is a different row, and protection composes by
/// row — an entry is protected while any un-expired pin row exists.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PinRow {
    /// The opaque key bytes of the entry being pinned.
    pub key: Vec<u8>,
    /// The store-instance id minting the pin (ADR-011's vocabulary).
    pub owner: Vec<u8>,
    /// The row's expiry; the TTL and the renewal schedule (every TTL/3)
    /// are the store's, not the clause's (ADR-012 §2's rename).
    pub expiry: SystemTime,
}

/// The engine-state companion seam's **window family**: the GC schema
/// clause and the staged delete window — on every SQL-backed engine
/// (ADR-010 §2; ADR-012 §5's table: postgres, sqlite, pg-lo). Postgres is
/// fleet-active on these when declared; sqlite stages durably
/// single-node; pg-lo hosts the window where the kv engine is not
/// SQL-backed — under the one-window-host rule (at most one SQL engine
/// per store instance carries an active window; the others' clauses are
/// dormant). Crate-internal — never exported; the store core calls the
/// clauses, engines execute the SQL, and each clause is contract-tested
/// per its applicability row.
/// The window take's batch bound (review-001 N-3, the recorded design
/// point — a named const, not a knob): `window_take` deletes and returns
/// at most this many overdue rows per call, so an enormous overdue
/// backlog never materializes entirely in memory in one step. The store's
/// sweep executor consumes the take in a loop until drained (the
/// one-sweep-executes-all-overdue semantics ride the loop, never the
/// batch size); the value is sized equal to the sweep body's own delete
/// batch (`SWEEP_DELETE_BATCH`) so one take feeds one execution batch.
pub(crate) const WINDOW_TAKE_BATCH: usize = 100;

pub(crate) trait EngineWindowState: Backend {
    /// Create the GC table(s) this engine's applicability row names:
    /// postgres — `gc_pins` + `gc_window` (the fleet host); sqlite and
    /// pg-lo — `gc_window` only (ADR-012 §5). Called at store
    /// construction; dialect-specific bodies, engine-neutral call. The
    /// embedder's liveness table is never created here — the sweep reads
    /// it and never owns it (ADR-008's schema-ownership rule).
    async fn ensure_gc_schema(&self) -> Result<(), Error>;

    /// Stage candidates into this engine's durable delete window. The
    /// staging is **transactional** (review-001 N-2): one transaction per
    /// call — all the candidates' upserts commit together, and a failure
    /// anywhere in the loop leaves zero staged rows. The upsert discipline
    /// is idempotent per `(key, deadline)` — a re-stage of the same key at
    /// the same deadline is one unchanged row — and the conflict update is
    /// **forward-only** (review-001 N-4): a re-stage at an earlier
    /// deadline never moves `visible_after` backwards (the dialect-carried
    /// `GREATEST`/`max` semantics), so a re-stage can never accelerate
    /// deletion beyond the embedder's staging intent. The window's schema
    /// runs at store construction (the store-core [`Self::ensure_gc_schema`]
    /// call, the constructor-vs-lazy timing design point), so staging
    /// always sees its table in real service — no lazy create. A genuinely
    /// tableless database surfaces the engine error on stage rather than
    /// silently creating schema; staging tolerance beyond that is each
    /// engine's recorded posture (the Backend trait's engine posture
    /// carries; sqlite's is recorded in src/sqlite.rs's module doc).
    async fn window_stage(&self, candidates: &[WindowCandidate]) -> Result<(), Error>;

    /// Take overdue candidates (`visible_after <= now`): bounded to
    /// [`WINDOW_TAKE_BATCH`] rows per call (review-001 N-3 — a huge
    /// overdue backlog never materializes entirely in memory in one step;
    /// the stale-before-newest deadline order drains the oldest first),
    /// and the calling sweep executor consumes the take in a loop until
    /// drained, preserving the one-sweep-executes-all-overdue semantics.
    /// The rows leave the window table and return as keys for the calling
    /// sweep to re-arbitrate and delete — the executor is the next sweep
    /// (ADR-012 §6.7). Empty when nothing is overdue; virgin store: empty.
    async fn window_take(&self, now: SystemTime) -> Result<Vec<Vec<u8>>, Error>;
}

/// The engine-state companion seam's **fleet family**: the pin clauses
/// and the sweep lease — the postgres host only (ADR-012 §4: all fleet
/// GC state hosts on the fleet's kv-tier engine, large tier included;
/// ADR-012 §5: sqlite's and pg-lo's pin/sweep-lock bodies are dropped).
/// Fleet activation is a constructor declaration (`fleet: true`), never
/// inferred from the engine choice (ADR-012 §3): under `fleet: false`
/// none of this runs and the ADR-005 in-process protocol applies
/// verbatim, regardless of engine. Crate-internal — never exported.
pub(crate) trait EngineFleetState: EngineWindowState {
    /// The joint entry+pin transaction — the declared holder type whose
    /// concrete payloads are per-engine bodies (waves 3-5: the pg
    /// transaction the pooled client carries; in the pg-lo consolidation
    /// posture, the same tx the LO writes ride, ADR-012 §4).
    ///
    /// **Ownership rule, pinned in this signature:** the store core
    /// holds the tx from [`Self::begin_joint_tx`] until it hands the
    /// value to [`Self::pin_commit`] (which consumes it as the joint
    /// terminal step); the engine contributes mechanics into the tx and
    /// never owns it. This is ADR-010 §2's ruling — "the store holds the
    /// tx; the engine contributes mechanics" — which corrected ADR-009's
    /// put-transaction ownership statement.
    type JointTx: Send;

    /// Begin the joint tx (store-core call): open on the engine's
    /// connection/session, uncommitted until [`Self::pin_commit`].
    async fn begin_joint_tx(&self) -> Result<Self::JointTx, Error>;

    /// The entry half the engine contributes into the held tx: its put
    /// mechanics (row shape, CAS clause) under the same first-writer-wins
    /// semantics as the public `put` (ADR-010 §2). For the pg-lo
    /// consolidation posture this clause is the kv host's row write;
    /// the large entry's content reaches the same tx through that
    /// engine's joint contribution — the landed `put_joint_staged`
    /// handout in `crate::pg_lo`, whose joint staging cursor writes
    /// `lo_create`/chunked `lowrite` + the companion-row insert into
    /// this tx type (an engine-task contribution against this
    /// `JointTx` type, per ADR-012 §4's one-tx ruling).
    async fn joint_put(
        &self,
        tx: &mut Self::JointTx,
        key: &[u8],
        value: &[u8],
    ) -> Result<(), Error>;

    /// The joint terminal step: the pin-row inserts **and** the commit —
    /// publish and protection land in one step (ADR-005's
    /// pin-before-publish, fleet form). On any failure the entire tx
    /// aborts: no pin rows, no published entry, zero residue — the
    /// stage-hygiene invariant's tx form. Consumes the tx; the store
    /// core never re-uses it after this call.
    async fn pin_commit(&self, tx: Self::JointTx, entries: &[PinRow]) -> Result<(), Error>;

    /// Renew the `(key, owner)` pin row: `true` when a **live**
    /// (un-expired) row existed and its expiry moved to this call's
    /// successor time (the store computes cadence — every TTL/3 — and
    /// passes the renewal's shape through the row update). An expired or
    /// absent row is never resurrected: `false`, nothing updated —
    /// ADR-012 §2's expiry semantics verbatim (a putter paused past TTL
    /// falls to the delete-then-recover backstop; the ops handler reads
    /// this bool as the renewal probe's `present` answer).
    async fn pin_renew(&self, key: &[u8], owner: &[u8]) -> Result<bool, Error>;

    /// Classify the key against un-expired pin rows: any un-expired
    /// `(key, *)` row exists → [`PinState::Protected`]; else
    /// [`PinState::Unprotected`]. The engine never decides deletion — it
    /// only classifies (ADR-010 §2); the store core receives the answer
    /// and applies the decision.
    async fn pin_check(&self, key: &[u8]) -> Result<PinState, Error>;

    /// The maintenance step (ADR-008 §2's GC-of-GC cleanup): delete the
    /// **expired** pin rows (`expiry <= now`) — a pin whose holder
    /// stopped renewing is abandoned, and this step reclaims its row.
    /// Un-expired rows are never touched. Returns the count reaped.
    /// Runs under the sweep lease (the fleet sweep body drives it —
    /// the seam is the named home for fleet-clause growth, ADR-010 §2).
    async fn pin_reap(&self, now: SystemTime) -> Result<u64, Error>;

    /// The full delete-time arbitration (ADR-008): the un-expired pin
    /// rows plus the embedder's registered liveness tables, under the
    /// `SELECT … FOR UPDATE` finger-lock keyed at delete time, folded
    /// into one answer — `true` = liveness protects the key at this
    /// instant (the caller refuses the direct delete / cancels the
    /// window candidate); `false` = nothing does and the caller may
    /// proceed. The engine never decides; the facade decides and owns
    /// the refusal (`Error::GcRefuse` never fires below the facade).
    async fn arbitrate(&self, key: &[u8]) -> Result<bool, Error>;

    /// Acquire the sweep lease — the pg advisory lock with ADR-013 §2's
    /// named exclusive semantics: atomic test-and-set, one sweeper per
    /// pool, holdership ends at release, owner death, or engine-session
    /// loss. `Ok(true)` = held; `Ok(false)` = contention — the second
    /// sweeper's sweep aborts with `GcAbortCause::SweeperLock` as report
    /// data (`Ok(SweepReport)`; sweeps are never errors, ADR-012 §1).
    async fn sweep_lock_acquire(&self) -> Result<bool, Error>;

    /// Release the sweep lease. Releasing an unheld lease is a no-op;
    /// owner death and engine-session loss reap it without this call
    /// (ADR-013 §2's holdership rule).
    async fn sweep_lock_release(&self) -> Result<(), Error>;
}