alkblobs/src/gc.rs

//! 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 SQL-backed kv engine; the sweep's durable
//! form drives `window_stage`/`window_take`), while the fleet family
//! waits for postgres's wave. It shrinks as that lands.

#![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};
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 **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)
    }

    /// 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()
    }

    /// **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 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>)>,
}

impl RegistryCore {
    fn new() -> RegistryCore {
        RegistryCore {
            state: Mutex::new(RegistryState {
                pins: BTreeMap::new(),
                deleting: BTreeSet::new(),
                sources: Vec::new(),
                protects: 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, protect: false }
    }

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

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

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

    fn unregister(&self, registration: &Registration, protect: bool) -> bool {
        let mut state = self.state.lock();
        if protect {
            let before = state.protects.len();
            state.protects.retain(|(id, _)| *id != registration.id);
            state.protects.len() != before
        } else {
            let before = state.sources.len();
            state.sources.retain(|(id, _)| *id != registration.id);
            state.sources.len() != before
        }
    }

    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) {
        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.
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 (idempotent
    /// staging per `(key, deadline)` is the body's discipline; the store
    /// stages each candidate once per sweep). 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`): 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 staged `lowrite` bodies (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 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>;
}