alkgit_substrate/

liveness.rs

//! The GC liveness sources over alkgit's own durable rows (waves 7
//! and 9): the ref-table closure — every ref row's full reachable
//! graph walk over the substrate object front — and, under the `lfs`
//! feature, the media rows source ([`LfsMedia`], substrate.md §GC
//! sources 1 and 2; ADR-019 §6) — both registered as in-process
//! [`alkblobs::LivenessSource`]s on the composition's store at the
//! composition entry ([`crate::compose`]).
//!
//! # The walk (the owning discipline)
//!
//! The live set is the union over all repos' oid rows of the full
//! graph closure of the row's tip oid: commits (parents included —
//! the owning walk; gix-fsck's `Connectivity::check_commit` does not
//! walk parents, so it is not the walk here), trees (sub-trees and
//! blobs), blobs, and tags (peeled to their target and walked on).
//! Gitlink tree entries (submodule modes) neither mark nor walk: the
//! pinned commit belongs to its other repo, whose own refs cover it.
//! Symref rows (the seeded unborn `HEAD`) carry no oid — unborn
//! contributes nothing; when the symref's target is a real row, that
//! row's own closure covers the object. Keys are assembled map-free
//! ([`crate::odb::oid_key`]): `[0x03]` + oid bytes, the git-object
//! family's physical encoding.
//!
//! The walk is best-effort by the seam's shape
//! ([`alkblobs::LivenessSource::live_keys`] returns keys, not
//! errors): a missing or unreadable object stops that branch (the key
//! is absent from the pool or unread — both safe) and the walk
//! continues; the next rebuild retries. A corrupt entry surfaces the
//! front's verification error at the walk read — the same
//! [`StorageError::Invalid`] the serving reads carry.
//!
//! # The LFS media source (the second alkgit-owned liveness source,
//! the `lfs` feature)
//!
//! [`LfsMedia`] (wave 9, ADR-022 §4; substrate.md §GC source 2) serves
//! the LFS media family's liveness from the pointer-lifecycle rows
//! ([`crate::lfs::LfsRows`]). The mark consult is **a row read, not a
//! graph walk** — media is unstructured content, there is nothing
//! reachable *through* it — and the keys form **map-free**: the row's
//! oid *is* the digest, so the key is `[0x04] + oid bytes`
//! ([`lfs_media_key`], the odb front's `oid_key` analog for the
//! 32-byte sha256 form; the oid is the raw sha256 of the content,
//! byte-exact the `GitObjectSha256` derivation — lfs.md §2, ADR-022
//! §1). One key with N rows is one live key (rows dedup through the
//! key set — the cross-reference refcount's mark-side reading; *any*
//! row re-protects the key, the refcount column's contract realized).
//!
//! The invalidation posture is the pinned simple-honest shape: the
//! mark consult **rebuilds per mark** (the row read is cheap; the
//! refs closure's fingerprint machinery buy the walk's cost — nothing
//! to skip here, so nothing to key a memo on). What the arbitration
//! discipline *does* need is the refs closure's **write-counter**
//! half: `RefClosure`'s `contains` consult cannot answer exactly without an
//! epoch signal, and the LFS rows' dangerous window is a row
//! *landing* between a sweep's mark and a staged candidate's
//! arbitration (the stale index says unreferenced; deleting it eats
//! referenced media). [`crate::lfs::VersionedLfsRows`] bumps the
//! shared counter on every `assert` — the coverage-increasing event,
//! and the only one whose staleness is dangerous: a removal's stale
//! answers are the safe side (a stale "live" holds a candidate; a
//! stale "dead" is exact for a candidate whose rows left). The
//! same-window answer pattern as the refs source: index hit exactly,
//! cold memo live (the safe default), miss after a moved counter live
//! until the next mark's rebuild resolves it.
//!
//! The mark consult bridges with the front's
//! [`block_in_place`](crate::odb::ObjectFront::block_in_place)
//! posture (the sqlite/pg rows twins are async — spawn_blocking /
//! native I/O inside); the arbitration consult never bridges. Same
//! floor: the multi-thread runtime (test-pinned). The failed mark
//! consult answers empty with the memo untouched — the same residual
//! class the refs closure documents (the stale memo answers at
//! arbitration; the next mark corrects).
//!
//! The cross-process posture is the refs closure's, stated honestly:
//! the in-process source is instance-local (alkblobs ADR-012 §6.6's
//! allow-and-document reading); the fleet-wide answer is the
//! materialized liveness table
//! ([`alkblobs::LivenessRegistration::FleetTable`] — wave-10 fleet
//! assembly's wiring, not this module's). The in-flight LFS uploads
//! are protected fleet-wide by the DB-backed pins (alkblobs ADR-008)
//! — the put side is pins', not this source's.
//!
//! # The sync verb / async store bridge
//!
//! [`alkblobs::LivenessSource`]'s verbs are sync (the sweep consults
//! `live_keys` from its async body, `contains` under the arbitration
//! lock). The walk bridges with
//! [`crate::odb::ObjectFront::block_in_place`] — the odb-front task's
//! async-context posture (the POC-O panic shape; the blocking half is
//! the pack pipelines'). Consequence: the multi-thread runtime is the
//! floor — a current-thread runtime panics (test-pinned).
//! `contains` never bridges: it answers from the memo's in-memory
//! index only ([`RefClosure::keys_builds`] proves no consult derives).
//! The walk consumes the front's async read legs directly (the same
//! verified reads the sync Find legs bridge to) — the walk runs
//! inside the bridge's future, where the sync legs' own `block_on`
//! is illegal (the POC-O panic shape).
//!
//! # Memoization and the invalidation decision (the honest posture)
//!
//! POC-O's shape was the closure memoized by the refs table's version
//! counter. The wave-6 refs rows carry no version counter, so this
//! task pins the mechanism; the three candidates resolved as two
//! composable half-decisions:
//!
//! - **The memo's rebuild key is a durable refs fingerprint** (the
//!   durable-marker option, realized read-side — no schema change).
//!   [`RefRows::snapshot`] returns the rows plus a fingerprint over
//!   their canonical form; a sweep's mark consult reads it (one cheap
//!   refs-rows read, ordered by repo/name) and rebuilds the walk only
//!   when it moved. Because the fingerprint is computed from the
//!   durable rows, cross-process ref writes are seen at mark time by
//!   construction — the fleet-shape invalidation is the *snapshot
//!   read*, not process-local state.
//! - **The arbitration window is covered by an in-process write
//!   counter** (the POC-O shape carried). A push landing between this
//!   sweep's mark and the next sweep's execution-time arbitration
//!   cannot move the fingerprint consult (arbitration runs under the
//!   store's short-held lock — no bridging, no reads). The
//!   composition's refs writes therefore bump a shared in-process
//!   counter ([`crate::refs::VersionedRefTable`] on the sqlite
//!   assembly), and [`alkblobs::LivenessSource::contains`] answers
//!   conservatively — "live" (unknown) — whenever the counter moved
//!   past the memo's build: the affected candidates cancel and
//!   resolve on the next sweep's mark, which rebuilds against the new
//!   rows. Over-invalidation costs one delayed sweep; the alternative
//!   deletes a pushed object. A cold (never-built) memo answers
//!   "live" too — the safe default until the first mark; direct
//!   deletes under it refuse ([`alkblobs::Error::GcRefuse`]), the
//!   ADR-012 §6.4 deregister-delete-re-register correction flow is
//!   the escape hatch, and an index hit always answers exactly.
//! - **The cross-process posture, stated honestly**: the in-process
//!   source is instance-local (alkblobs ADR-012 §6.6's allow-and-
//!   document reading — the SQL protocol sees tables, not callbacks).
//!   Single-node shapes (the default sqlite assembly) are exact.
//!   Fleet deployments carry one residual window: a completed push
//!   from *another* instance between mark and execution-time
//!   arbitration (this instance's counter does not move and the mark
//!   has not come yet). The in-flight push itself is protected
//!   fleet-wide by the DB-backed pins (alkblobs ADR-008), and the
//!   fleet-wide answer is the materialized liveness table
//!   ([`alkblobs::LivenessRegistration::FleetTable`] — the embedder's
//!   own table the store core reads at SQL arbitration); that
//!   materialization is the fleet assembly's wiring — the wave-8
//!   composition entry is the unifier and the wave-10 pg assembly
//!   exercises it (the pg `GitRefs` twin landed with the wave-7 pg
//!   refs task), not this module's.
//!
//! # Registration
//!
//! [`RefClosure::register`] registers the source on the store
//! (`Store::register_liveness_source`, the live-shared seam —
//! consulted live every sweep, never a snapshot at registration).
//! The composition's construction wires [`RefClosure::new`] over the
//! object front, the refs rows seam, and the assembly's write
//! counter, then registers: zero registered liveness sources make
//! every sweep abort with
//! [`alkblobs::GcAbortCause::NoLivenessSources`] — this source is
//! what the wave-7 maintenance chain's sweep dispatch requires. The
//! wave-8 composition entry unifies the construction; until then the
//! pieces + this wiring are the proof surface (the tests run the full
//! chain, sqlite co-tenancy included).

use std::collections::HashSet;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use alkblobs::Key;
use async_trait::async_trait;
use gix_hash::ObjectId;
use gix_object::Kind;

use alkgit::StorageError;

use crate::odb::oid_key;

/// One refs row as the liveness seam reads it: the resolved repo id,
/// the validated ref name, and the row's state — an oid tip or the
/// symref (unborn) shape. The rows cross [`RefRows::snapshot`] as
/// they are, not as [`alkgit::RefLine`] listings: liveness walks
/// tips, it does not advertise.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RefRow {
    pub repo_id: String,
    pub name: String,
    pub state: RefRowState,
}

/// A refs row's state as the CAS tables store it (the refs module's
/// row shapes, surfaced for the liveness seam).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RefRowState {
    /// The ordinary ref row: the tip oid.
    Oid(ObjectId),
    /// The symref row (the seeded `HEAD → refs/heads/main`): no oid
    /// of its own — unborn, contributes nothing to the closure.
    Symref(String),
}

/// A refs-rows snapshot: every row of every repo, in the canonical
/// order (repo id, then name — sorted), plus the fingerprint the
/// memo keys on.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RefRowsSnapshot {
    pub rows: Vec<RefRow>,
    pub fingerprint: u64,
}

/// The refs rows' read seam — the durable state the closure walks and
/// fingerprints. Implemented by the refs tables
/// ([`crate::refs::InMemoryRefTable`], the `sqlite` twin) and
/// forwarded by the composition's write-counting wrapper
/// ([`crate::refs::VersionedRefTable`]). Engine twins carry their
/// own: snapshotting is a plain ordered read of the rows the CAS
/// writes.
#[async_trait]
pub trait RefRows: Send + Sync {
    /// The full canonical snapshot: all repos' rows, name-sorted,
    /// with the fingerprint over the canonical form.
    async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError>;
}

/// The canonical fingerprint over a snapshot's rows: the row count
/// folded into an FNV-1a pass over each row's canonical bytes
/// (length-prefixed strings, the state discriminant, oid bytes /
/// target bytes). The order is the canonical one (repo id, then
/// name), which the impls guarantee (B-tree maps; `ORDER BY
/// repo_id, name`); a fingerprint move means the durable rows moved
/// (cross-process — the snapshot reads the rows, never local state).
pub(crate) fn fingerprint_rows(rows: &[RefRow]) -> u64 {
    fn fold(state: &mut u64, part: &[u8]) {
        for b in part {
            *state ^= u64::from(*b);
            *state = state.wrapping_mul(0x0100_0000_0193_e739);
        }
        *state ^= 0x0f4e_a559_f1bf_2acd;
        *state = state.wrapping_mul(0x0100_0000_0193_e739);
    }
    // Canonical (repo id, then name) order regardless of what the
    // caller passes — the impls hand sorted rows; the fold is
    // order-normalized anyway so the fingerprint cannot drift on
    // input order.
    let mut sorted: Vec<&RefRow> = rows.iter().collect();
    sorted.sort_by(|a, b| (&a.repo_id, &a.name).cmp(&(&b.repo_id, &b.name)));
    let mut state = 0xcbf2_9ce4_8422_2325u64;
    fold(&mut state, &(sorted.len() as u64).to_le_bytes());
    for row in sorted {
        fold(&mut state, row.repo_id.as_bytes());
        fold(&mut state, row.name.as_bytes());
        match &row.state {
            RefRowState::Oid(oid) => {
                fold(&mut state, &[0x01]);
                fold(&mut state, oid.as_bytes());
            }
            RefRowState::Symref(target) => {
                fold(&mut state, &[0x02]);
                fold(&mut state, target.as_bytes());
            }
        }
    }
    state
}

/// The memo — the closure the sweep's mark consults and the
/// arbitration's `contains` answers from, keyed on the fingerprint
/// and the write counter. Held under a short lock only: `contains`
/// reads it under the store's arbitration lock, the rebuild replaces
/// it back, and the walk itself never holds it.
struct Memo {
    fingerprint: u64,
    build_writes: u64,
    keys: HashSet<Key>,
}

/// The ref-table-closure [`alkblobs::LivenessSource`] — the first
/// alkgit-owned liveness source (substrate.md §GC; ADR-019 §6). See
/// the module docs for the walk, the bridge, and the invalidation
/// decision.
pub struct RefClosure {
    front: crate::odb::ObjectFront,
    rows: Arc<dyn RefRows>,
    writes: Arc<AtomicU64>,
    memo: parking_lot::Mutex<Option<Memo>>,
    builds: AtomicU64,
}

impl RefClosure {
    /// The closure source over the object front, the refs rows seam,
    /// and the composition's shared write counter (bumped by the
    /// refs writes — [`crate::refs::VersionedRefTable`] on the sqlite
    /// assembly).
    pub fn new(
        front: crate::odb::ObjectFront,
        rows: Arc<dyn RefRows>,
        writes: Arc<AtomicU64>,
    ) -> Arc<Self> {
        Arc::new(RefClosure {
            front,
            rows,
            writes,
            memo: parking_lot::Mutex::new(None),
            builds: AtomicU64::new(0),
        })
    }

    /// Register the source with the store — the live-shared seam
    /// (`Store::register_liveness_source`; alkblobs store-api.md
    /// §Lifecycle). Registered at the composition's construction;
    /// deregistration/re-registration around a direct delete is the
    /// ADR-012 §6.4 embedder correction flow (out of scope here).
    ///
    /// The composition carries this source for the store's lifetime
    /// (registration once, at construction — the returned token is
    /// deregister-at-most-once and is the composition's to keep).
    pub fn register(
        self: &Arc<Self>,
        store: &alkblobs::Store,
    ) -> Result<alkblobs::Registration, alkblobs::Error> {
        store.register_liveness_source(alkblobs::LivenessRegistration::InProcess(
            self.clone() as Arc<dyn alkblobs::LivenessSource>
        ))
    }

    /// How many closure walks have run — the memoization pin (a
    /// fingerprint hit consumes no walk; the arbitration consult
    /// never triggers one).
    pub fn keys_builds(&self) -> u64 {
        self.builds.load(Ordering::Relaxed)
    }

    /// The mark consult's build half: the cheap refs snapshot, the
    /// fingerprint hit-or-rebuild, the walk over the front. The
    /// counter loads **before** the snapshot (the epoch discipline:
    /// every write's bump lands *after* its commit, so a memo's
    /// counter-load can never pair an included bump with a
    /// snapshot-read that missed the same write's committed rows —
    /// the poisoned memo shape; anything the read missed moves the
    /// counter past the captured epoch, which answers
    /// conservatively-live at arbitration).
    async fn rebuild_or_cached_keys(&self) -> Result<Vec<Key>, StorageError> {
        let build_writes = self.writes.load(Ordering::Relaxed);
        let snapshot = self.rows.snapshot().await?;
        {
            let mut memo = self.memo.lock();
            if let Some(memo) = memo.as_mut() {
                if memo.fingerprint == snapshot.fingerprint {
                    // The rows are unchanged — the index stays exact.
                    // A write landing between the counter load and the
                    // fingerprint hit would have moved the fingerprint,
                    // so the captured epoch is still current here.
                    memo.build_writes = build_writes;
                    return Ok(memo.keys.iter().cloned().collect());
                }
            }
        }
        let (keys, fingerprint) = self.walk(&snapshot).await;
        self.builds.fetch_add(1, Ordering::Relaxed);
        *(self.memo.lock()) = Some(Memo {
            fingerprint,
            build_writes,
            keys: keys.iter().cloned().collect(),
        });
        Ok(keys)
    }

    /// The graph walk: tips from the oid rows, BFS over the object
    /// front's async read legs (the same verified reads the sync Find
    /// legs bridge to — the walk runs *inside* the
    /// [`block_in_place`](crate::odb::ObjectFront::block_in_place)
    /// bridge, where the sync legs' own `block_on` is illegal, the
    /// POC-O panic shape). Parents included; gitlinks skipped;
    /// map-free keys. Best-effort on per-node failures (the module
    /// docs' walk section). The fingerprint travels: it is the
    /// snapshot's, so the memo it lands in keys on exactly the rows
    /// the tips came from.
    async fn walk(&self, snapshot: &RefRowsSnapshot) -> (Vec<Key>, u64) {
        let fingerprint = snapshot.fingerprint;
        let mut tips: Vec<ObjectId> = snapshot
            .rows
            .iter()
            .filter_map(|row| match &row.state {
                RefRowState::Oid(oid) => Some(*oid),
                RefRowState::Symref(_) => None,
            })
            .collect();
        tips.sort();
        tips.dedup();

        let mut visited: HashSet<ObjectId> = HashSet::new();
        let mut keys: Vec<Key> = Vec::new();
        let mut stack = tips;
        while let Some(oid) = stack.pop() {
            if !visited.insert(oid) {
                continue;
            }
            let Ok(key) = oid_key(&oid) else {
                continue;
            };
            let Some((kind, body)) = self.front.get(&oid).await.ok().flatten() else {
                continue;
            };
            keys.push(key);
            let body = body.as_slice();
            match kind {
                Kind::Blob => {}
                Kind::Commit => {
                    let Ok(commit) = gix_object::CommitRef::from_bytes(body, gix_hash::Kind::Sha1)
                    else {
                        continue;
                    };
                    for parent in std::iter::once(&commit.tree).chain(commit.parents.iter()) {
                        if let Ok(id) = ObjectId::from_hex(parent) {
                            stack.push(id);
                        }
                    }
                }
                Kind::Tree => {
                    for entry in gix_object::TreeRefIter::from_bytes(body, gix_hash::Kind::Sha1) {
                        let Ok(entry) = entry else { continue };
                        match entry.mode.kind() {
                            gix_object::tree::EntryKind::Tree
                            | gix_object::tree::EntryKind::Blob
                            | gix_object::tree::EntryKind::BlobExecutable
                            | gix_object::tree::EntryKind::Link => {
                                stack.push(entry.oid.to_owned());
                            }
                            gix_object::tree::EntryKind::Commit => {}
                        }
                    }
                }
                Kind::Tag => {
                    let Ok(tag) = gix_object::TagRef::from_bytes(body, gix_hash::Kind::Sha1) else {
                        continue;
                    };
                    if let Ok(target) = ObjectId::from_hex(tag.target) {
                        stack.push(target);
                    }
                }
            }
        }
        (keys, fingerprint)
    }
}

impl alkblobs::LivenessSource for RefClosure {
    /// The mark-time consult — the whole live set of the refs
    /// closure, bridged through the front's async-context posture
    /// (multi-thread runtime floor). Fingerprint hit = no walk; a
    /// failed snapshot consult answers empty (the next mark corrects
    /// it). The delete leg is not widened by it: arbitration consults
    /// this impl's trait-method `contains`, which answers against the
    /// *same* memo the
    /// failed mark left untouched — the staged pool-wide candidates
    /// are guarded by the stale-epoch answer, so the deletion set is
    /// the residual class documented in the module docs, not the
    /// snapshot failure's. The alternative postures (blocking or
    /// panicking the sweep's mark) are worse.
    fn live_keys(&self) -> Vec<Key> {
        self.front
            .block_in_place(self.rebuild_or_cached_keys())
            .unwrap_or_default()
    }

    /// The delete-time arbitration consult — under the store's
    /// arbitration lock, quick, indexed, never blocking (no bridge,
    /// no reads, no recompute):
    ///
    /// - an index hit answers exactly (`true` — the key was live at
    ///   the memo's build);
    /// - a cold (never-built) memo answers `true` — the safe default
    ///   (unknown is live);
    /// - an index miss after an in-process refs write answers `true`
    ///   too — the write counter moved past the build, so the
    ///   candidate's coverage state is unknown: cancel it now, let
    ///   the next mark's rebuild decide. A ref delete between mark
    ///   and arbitration lands here the safe way too (kept one more
    ///   sweep, reclaimed after the rebuilt mark confirms).
    fn contains(&self, key: &Key) -> bool {
        let memo = self.memo.lock();
        let Some(memo) = memo.as_ref() else {
            return true;
        };
        if memo.keys.contains(key) {
            return true;
        }
        let writes_now = self.writes.load(Ordering::Relaxed);
        writes_now != memo.build_writes
    }
}

// The liveness seam's row reads and the wrapper: [`RefRows`] is
// implemented by the refs tables and forwarded by
// [`crate::refs::VersionedRefTable`] (the composition hands the
// liveness wiring one handle for both the writes — the bump — and the
// rows — the snapshot). The seam type lives here (the walk's durable
// state); its trait impls live beside their tables ([`crate::refs`]).

#[cfg(feature = "lfs")]
use crate::lfs::LfsRows;

/// The media family's key formation from a row's oid — the map-free
/// half the walk's [`oid_key`] is, sha256-swapped
/// (lfs.md §2; ADR-022 §1: the oid *is* the digest, byte-exact the
/// `GitObjectSha256` derivation): the key is
/// `[Algorithm::GitObjectSha256] + oid bytes`, decoded through
/// `Key::from_bytes` over the physical encoding — family-explicit,
/// rejects-never-guesses. A non-64-lowercase-hex oid is
/// [`StorageError::Invalid`] (the rows store validates the same form
/// on write; the read-side check is the front's OQ-05 posture).
#[cfg(feature = "lfs")]
pub fn lfs_media_key(oid_hex: &str) -> Result<Key, StorageError> {
    let hex_ok = oid_hex.len() == 64
        && oid_hex
            .bytes()
            .all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'));
    if !hex_ok {
        return Err(StorageError::Invalid(format!(
            "the LFS media oid is not a 64-lowercase-hex sha256 form: {oid_hex:?}"
        )));
    }
    let mut bytes = Vec::with_capacity(1 + 32);
    bytes.push(alkblobs::Algorithm::GitObjectSha256.byte());
    let pair_value = |pair: &[u8]| {
        let hi = (pair[0] as char).to_digit(16).ok_or_else(|| {
            StorageError::Invalid("the media-family digest half is not hex".to_owned())
        })?;
        let lo = (pair[1] as char).to_digit(16).ok_or_else(|| {
            StorageError::Invalid("the media-family digest half is not hex".to_owned())
        })?;
        Ok((hi * 16 + lo) as u8)
    };
    for pair in oid_hex.as_bytes().chunks(2) {
        bytes.push(pair_value(pair)?);
    }
    Key::from_bytes(&bytes)
        .map_err(|e| StorageError::Invalid(format!("the media-family key does not decode: {e}")))
}

/// The memo of the LFS-media liveness source: the live keys built at
/// the last mark consult plus the rows surface's write counter at that
/// build (the arbitration epoch). Held under a short lock only;
/// `contains` reads it under the store's arbitration lock, the
/// rebuild replaces it back, and the rebuild itself never holds it.
#[cfg(feature = "lfs")]
struct MediaMemo {
    build_writes: u64,
    keys: HashSet<Key>,
}

/// The LFS-media [`alkblobs::LivenessSource`] — the second
/// alkgit-owned liveness source (substrate.md §GC source 2; ADR-022
/// §4): the pointer-lifecycle rows as the media family's live set, a
/// mark consult that reads the rows (no graph walk) and an arbitration
/// consult that answers from the mark's index. See the module docs'
/// LFS section for the walk-not-walk posture, the invalidation
/// decision, and the cross-process residual.
#[cfg(feature = "lfs")]
pub struct LfsMedia {
    rows: Arc<dyn LfsRows>,
    writes: Arc<AtomicU64>,
    runtime: tokio::runtime::Handle,
    memo: parking_lot::Mutex<Option<MediaMemo>>,
    builds: AtomicU64,
}

#[cfg(feature = "lfs")]
impl LfsMedia {
    /// The source over the rows surface and the composition's shared
    /// assert counter (the [`VersionedLfsRows`](crate::lfs::VersionedLfsRows)
    /// wrapper bumps the same counter per committed `assert` through
    /// it — the bump-after-commit half of the epoch discipline).
    /// Construct within the running runtime (the assembly does): the
    /// constructor captures `Handle::current()` for the mark consult's
    /// bridge.
    pub fn new(rows: Arc<dyn LfsRows>, writes: Arc<AtomicU64>) -> Arc<Self> {
        Arc::new(LfsMedia {
            rows,
            writes,
            runtime: tokio::runtime::Handle::current(),
            memo: parking_lot::Mutex::new(None),
            builds: AtomicU64::new(0),
        })
    }

    /// Register the source with the store — the live-shared seam
    /// ([`Store::register_liveness_source`](`alkblobs::Store::register_liveness_source`);
    /// the composition entry's second registration leg, the fail-safe
    /// default ADR-022 §6: an unregistered media source lets sweeps
    /// eat referenced media on the first mixed-feature deployment).
    /// Registration once, at construction; the returned token is
    /// deregister-at-most-once and is the composition's to keep.
    pub fn register(
        self: &Arc<Self>,
        store: &alkblobs::Store,
    ) -> Result<alkblobs::Registration, alkblobs::Error> {
        store.register_liveness_source(alkblobs::LivenessRegistration::InProcess(
            self.clone() as Arc<dyn alkblobs::LivenessSource>
        ))
    }

    /// How many mark consults have run — the rebuild-per-mark posture's
    /// pin (no fingerprint memo: every mark consult re-reads the rows;
    /// the arbitration consult never triggers one).
    pub fn keys_builds(&self) -> u64 {
        self.builds.load(Ordering::Relaxed)
    }

    /// The front's async-context bridge, shared posture (the POC-O
    /// panic shape): parks the worker and serves the rows read through
    /// the blocking pool; the multi-thread runtime is the floor.
    fn block_in_place<F>(&self, fut: F) -> F::Output
    where
        F: std::future::Future + Send,
        F::Output: Send,
    {
        tokio::task::block_in_place(move || self.runtime.block_on(fut))
    }

    /// The mark consult's build half — the rows read, the keys formed
    /// map-free and deduped (one key with N rows is one live key), the
    /// memo replaced with the fresh epoch. The counter loads
    /// **before** the rows read (the epoch discipline, [`RefClosure`]'s
    /// build half): every assert's bump lands *after* its committed
    /// rows ([`crate::lfs::VersionedLfsRows`]), so a memo's captured
    /// epoch can never include a bump whose rows the row read missed —
    /// the poisoned memo that would answer exact-dead for
    /// just-referenced media. An assert landing between the load and
    /// the row read has its bump land after the load: the memo's
    /// epoch reads stale at the subsequent arbitration, which answers
    /// conservatively-live; an assert landing after the read moves the
    /// counter past the build's epoch the same way.
    async fn rebuild_keys(&self) -> Result<Vec<Key>, StorageError> {
        let build_writes = self.writes.load(Ordering::Relaxed);
        let rows = self.rows.rows().await?;
        let mut keys: HashSet<Key> = HashSet::new();
        for row in &rows {
            if let Ok(key) = lfs_media_key(&row.oid) {
                keys.insert(key);
            }
        }
        *(self.memo.lock()) = Some(MediaMemo {
            build_writes,
            keys: keys.clone(),
        });
        self.builds.fetch_add(1, Ordering::Relaxed);
        let mut out: Vec<Key> = keys.into_iter().collect();
        out.sort();
        Ok(out)
    }
}

#[cfg(feature = "lfs")]
impl alkblobs::LivenessSource for LfsMedia {
    /// The mark-time consult — rebuild-per-mark (the module docs' LFS
    /// section; the row read is the whole walk). A failed rows read
    /// answers empty with the memo untouched: the staged candidates
    /// arbitrate against the *stale* index (the previous mark's) —
    /// conservative for keys live at that build, and the next mark
    /// corrects. The alternative postures (blocking or panicking the
    /// sweep's mark) are worse.
    fn live_keys(&self) -> Vec<Key> {
        self.block_in_place(self.rebuild_keys()).unwrap_or_default()
    }

    /// The delete-time arbitration consult — under the store's
    /// arbitration lock, quick, indexed, never blocking (no bridge, no
    /// reads, no recompute):
    ///
    /// - an index hit answers exactly (`true` — the key was live at
    ///   the last mark);
    /// - a cold (never-marked) source answers `true` — the safe
    ///   default (unknown is live); direct deletes under it refuse
    ///   ([`alkblobs::Error::GcRefuse`], the ADR-012 §6.4 flow);
    /// - an index miss after a moved assert counter answers `true`
    ///   too — a row may have landed since the build (unknown is
    ///   live): the candidate cancels and resolves on the next mark's
    ///   rebuild, one delayed sweep.
    fn contains(&self, key: &Key) -> bool {
        let memo = self.memo.lock();
        let Some(memo) = memo.as_ref() else {
            return true;
        };
        if memo.keys.contains(key) {
            return true;
        }
        self.writes.load(Ordering::Relaxed) != memo.build_writes
    }
}

#[cfg(test)]
mod tests {
    use alkblobs::Store;
    use alkgit::{GitRefs, RefUpdate};
    use gix_hash::ObjectId;
    use gix_object::Kind;
    use std::sync::atomic::AtomicU64;
    use std::sync::Arc;

    use crate::odb::{key_to_oid, ObjectFront};
    use crate::refs::InMemoryRefTable;
    use crate::testutil::private_record as repo;
    #[cfg(feature = "sqlite")]
    use crate::testutil::scratch_dir;

    use super::*;

    fn writes() -> Arc<AtomicU64> {
        Arc::new(AtomicU64::new(0))
    }

    /// A multi_thread runtime is the floor (the block_in_place
    /// posture); every live-verb-driving test runs under it.
    fn mem_fixture() -> (ObjectFront, Arc<InMemoryRefTable>, Arc<AtomicU64>) {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        (front, Arc::new(InMemoryRefTable::new()), writes())
    }

    async fn put_object(front: &ObjectFront, kind: Kind, content: &[u8]) -> ObjectId {
        let (key, _pin) = front.put(kind, content).await.unwrap();
        key_to_oid(&key).expect("the put's key is the family key")
    }

    async fn create_ref(table: &dyn GitRefs, id: &str, name: &str, oid: ObjectId) {
        let outcomes = table
            .apply_updates(
                &repo(id),
                vec![RefUpdate {
                    name: name.to_owned(),
                    expected: None,
                    new: Some(oid),
                }],
                false,
            )
            .await
            .unwrap();
        assert!(outcomes.iter().all(|o| o.result.is_ok()));
    }

    fn blob_content() -> &'static [u8] {
        b"hello\n"
    }

    enum TreeEntry<'a> {
        Blob(&'a str, &'a ObjectId),
        Link(&'a str, &'a ObjectId),
        Gitlink(&'a str, &'a ObjectId),
    }

    fn tree_bytes(entries: &[TreeEntry<'_>]) -> Vec<u8> {
        let mode_of = |e: &TreeEntry<'_>| match e {
            TreeEntry::Blob(..) => "100644 ",
            TreeEntry::Link(..) => "120000 ",
            TreeEntry::Gitlink(..) => "160000 ",
        };
        let mut tree = Vec::new();
        let mut sorted: Vec<(&str, &str, &ObjectId)> = entries
            .iter()
            .map(|e| match e {
                TreeEntry::Blob(n, o) | TreeEntry::Link(n, o) | TreeEntry::Gitlink(n, o) => {
                    (mode_of(e), *n, *o)
                }
            })
            .collect();
        sorted.sort();
        for (mode, name, oid) in sorted {
            tree.extend_from_slice(mode.as_bytes());
            tree.extend_from_slice(name.as_bytes());
            tree.push(0);
            tree.extend_from_slice(oid.as_bytes());
        }
        tree
    }

    fn commit_bytes(tree: &ObjectId, parents: &[&ObjectId]) -> Vec<u8> {
        let mut body = format!("tree {tree}\n");
        for parent in parents {
            body.push_str(&format!("parent {parent}\n"));
        }
        body.push_str(
            "author A <a@example.invalid> 1750000000 +0000\n\
             committer A <a@example.invalid> 1750000000 +0000\n\n",
        );
        body.into_bytes()
    }

    fn tag_bytes(target: &ObjectId, target_kind: Kind) -> Vec<u8> {
        format!(
            "object {target}\n\
             type {target_kind}\n\
             tag v1\n\
             tagger A <a@example.invalid> 1750000000 +0000\n\nmessage\n"
        )
        .into_bytes()
    }

    /// A two-commit ancestry with one tree/blob, a tag on the tip,
    /// and gitlink + a second blob in a separate tree (the gitlink
    /// mode shape).
    async fn fixture_graph(front: &ObjectFront) -> GraphFixture {
        let blob = put_object(front, Kind::Blob, blob_content()).await;
        let blob2 = put_object(front, Kind::Blob, b"other\n").await;
        let tree = front
            .put(Kind::Tree, &tree_bytes(&[TreeEntry::Blob("hello", &blob)]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let gitlink_subtree_oid = front
            .put(Kind::Tree, &tree_bytes(&[TreeEntry::Blob("g", &blob)]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let gitlink_target = put_object(
            front,
            Kind::Commit,
            &commit_bytes(&gitlink_subtree_oid, &[]),
        )
        .await;
        let gitlink_tree = front
            .put(
                Kind::Tree,
                &tree_bytes(&[
                    TreeEntry::Blob("other", &blob2),
                    TreeEntry::Link("sym", &blob2),
                    TreeEntry::Gitlink("sub", &gitlink_target),
                ]),
            )
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let base = front
            .put(Kind::Commit, &commit_bytes(&tree, &[]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let tip = front
            .put(Kind::Commit, &commit_bytes(&tree, &[&base]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let tag = front
            .put(Kind::Tag, &tag_bytes(&tip, Kind::Commit))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        GraphFixture {
            blob,
            blob2,
            tree,
            base,
            tip,
            tag,
            gitlink_target,
            gitlink_tree,
        }
    }

    #[derive(Clone, Copy)]
    struct GraphFixture {
        blob: ObjectId,
        blob2: ObjectId,
        tree: ObjectId,
        base: ObjectId,
        tip: ObjectId,
        tag: ObjectId,
        gitlink_target: ObjectId,
        gitlink_tree: ObjectId,
    }

    /// The minimal covered closure — blob, tree, base commit, tip
    /// commit; no tag, no gitlink shapes. The sweep-focused tests'
    /// fixture (every pooled object accounted for in the assertions).
    async fn covered_fixture(front: &ObjectFront) -> GraphFixture {
        let blob = put_object(front, Kind::Blob, blob_content()).await;
        let tree = front
            .put(Kind::Tree, &tree_bytes(&[TreeEntry::Blob("hello", &blob)]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let base = front
            .put(Kind::Commit, &commit_bytes(&tree, &[]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let tip = front
            .put(Kind::Commit, &commit_bytes(&tree, &[&base]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let oid = ObjectId::from([0u8; 20]);
        GraphFixture {
            blob,
            blob2: oid,
            tree,
            base,
            tip,
            tag: oid,
            gitlink_target: oid,
            gitlink_tree: oid,
        }
    }

    async fn live_oids(source: &Arc<RefClosure>) -> Vec<ObjectId> {
        let mut oids: Vec<ObjectId> = alkblobs::LivenessSource::live_keys(source.as_ref())
            .iter()
            .filter_map(key_to_oid)
            .collect();
        oids.sort();
        oids
    }

    fn sorted(mut oids: Vec<ObjectId>) -> Vec<ObjectId> {
        oids.sort();
        oids
    }

    /// The full closure walk: two repos, commits (parents included),
    /// trees/blobs, a peeled tag, gitlink entries neither marked nor
    /// walked, the symref row contributing nothing, and the keys
    /// map-free (`[0x03] + oid`).
    #[tokio::test(flavor = "multi_thread")]
    async fn the_closure_walk_covers_the_full_graph() {
        let (front, refs, writes) = mem_fixture();
        let fix = fixture_graph(&front).await;
        create_ref(refs.as_ref(), "r/one", "refs/heads/main", fix.tip).await;
        create_ref(refs.as_ref(), "r/one", "refs/tags/v1", fix.tag).await;
        create_ref(refs.as_ref(), "r/two", "refs/heads/wip", fix.gitlink_tree).await;
        let orphan = put_object(&front, Kind::Blob, b"orphan\n").await;

        // A symref row in the snapshot (the seeded unborn HEAD shape):
        // it contributes nothing to the closure — and its target's own
        // row (r/one's main) covers the object when it exists.
        let seeded_rows = Arc::new(SymrefSeededRows::new(refs.clone()));
        seeded_rows.seed_head();
        let rows: Arc<dyn RefRows> = seeded_rows;
        let source = RefClosure::new(front.clone(), rows, writes);
        assert_eq!(source.keys_builds(), 0, "construction walks nothing");

        let oids = live_oids(&source).await;
        let mut expected = sorted(vec![
            fix.blob,
            fix.blob2,
            fix.tree,
            fix.base,
            fix.tip,
            fix.tag,
            fix.gitlink_tree,
        ]);
        expected.dedup();
        assert_eq!(
            oids, expected,
            "the closure covers the commit ancestry (parents included), the trees and blobs, and the peeled tag"
        );
        assert!(
            !oids.contains(&fix.gitlink_target),
            "a gitlink entry neither marks nor walks (the other repo's refs cover it)"
        );
        assert!(!oids.contains(&orphan), "an unreachable object is not live");
        assert_eq!(source.keys_builds(), 1);

        // Map-free keys: every live key carries the family byte.
        for key in alkblobs::LivenessSource::live_keys(source.as_ref()) {
            assert_eq!(key.as_bytes()[0], 0x03, "the sha1 family byte");
        }
    }

    /// The refs-rows seam standing in front of the CAS table to seed
    /// a HEAD symref row (the unborn shape the tables' CAS cannot
    /// write): the snapshot surfaces it, and the walk skips it.
    struct SymrefSeededRows {
        inner: Arc<InMemoryRefTable>,
        head: parking_lot::Mutex<Option<RefRow>>,
    }

    impl SymrefSeededRows {
        fn new(inner: Arc<InMemoryRefTable>) -> Self {
            SymrefSeededRows {
                inner,
                head: parking_lot::Mutex::new(None),
            }
        }

        fn seed_head(self: &Arc<Self>) {
            *(self.head.lock()) = Some(RefRow {
                repo_id: "r/one".to_owned(),
                name: "HEAD".to_owned(),
                state: RefRowState::Symref("refs/heads/main".to_owned()),
            });
        }
    }

    #[async_trait]
    impl RefRows for SymrefSeededRows {
        async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
            let mut rows = self.inner.snapshot().await?.rows;
            if let Some(head) = self.head.lock().clone() {
                rows.push(head);
            }
            rows.sort_by(|a, b| (&a.repo_id, &a.name).cmp(&(&b.repo_id, &b.name)));
            Ok(RefRowsSnapshot {
                fingerprint: fingerprint_rows(&rows),
                rows,
            })
        }
    }

    /// Orphan reclamation end-to-end at the source level: objects
    /// unreachable from any ref are absent from `live_keys`, the
    /// sweep consumes them (no `NoLivenessSources` abort), and the
    /// closure's objects survive.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_sweep_reclaims_orphans_when_the_source_is_registered() {
        let (front, refs, writes) = mem_fixture();
        let fix = covered_fixture(&front).await;
        create_ref(refs.as_ref(), "repo", "refs/heads/main", fix.tip).await;
        let rows: Arc<dyn RefRows> = refs.clone();

        let orphan_one = put_object(&front, Kind::Blob, b"orphan one\n").await;
        let orphan_two = put_object(&front, Kind::Blob, b"orphan two\n").await;

        let store = front.store().clone();
        let source = RefClosure::new(front.clone(), rows, writes);
        source.register(&store).unwrap();

        let report = store.sweep().await.unwrap();
        assert!(
            report.aborted.is_none(),
            "a registered source never trips NoLivenessSources"
        );
        assert_eq!(
            report.deleted, 2,
            "both orphans reclaimed (the pool: the covered closure + the two orphans)"
        );
        assert!(front.stat(&orphan_one).await.unwrap().is_none(), "gone");
        assert!(front.stat(&orphan_two).await.unwrap().is_none(), "gone");
        for live in [fix.tip, fix.base, fix.tree, fix.blob] {
            assert!(
                front.stat(&live).await.unwrap().is_some(),
                "the closure stays"
            );
        }
    }

    /// A ref delete uncovers its closure: the formerly-covered set is
    /// gone from `live_keys` after the fingerprint moves, and the
    /// sweep reclaims it (the repo-deletion-is-GC shape, at the
    /// source level — the wave-7 chain's sweep consumes it).
    #[tokio::test(flavor = "multi_thread")]
    async fn a_ref_delete_uncovers_its_closure_for_the_sweep() {
        let (front, refs, writes) = mem_fixture();
        let fix = fixture_graph(&front).await;
        create_ref(refs.as_ref(), "repo", "refs/heads/main", fix.tip).await;
        let rows: Arc<dyn RefRows> = refs.clone();

        let store = front.store().clone();
        let source = RefClosure::new(front.clone(), rows, writes);
        source.register(&store).unwrap();

        let first = alkblobs::LivenessSource::live_keys(source.as_ref()).len();
        assert_eq!(first, 4, "tip, base, tree, blob — the covered closure");
        for live in [fix.tip, fix.base, fix.tree, fix.blob] {
            assert!(front.stat(&live).await.unwrap().is_some());
        }

        let outcomes = refs
            .apply_updates(
                &repo("repo"),
                vec![RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: Some(fix.tip),
                    new: None,
                }],
                false,
            )
            .await
            .unwrap();
        assert!(outcomes.iter().all(|o| o.result.is_ok()));

        let after = alkblobs::LivenessSource::live_keys(source.as_ref());
        assert_eq!(
            after.len(),
            0,
            "the delete-uncovered closure leaves the live set after the rebuild"
        );

        let report = store.sweep().await.unwrap();
        assert!(report.aborted.is_none());
        for gone in [fix.tip, fix.base, fix.tree, fix.blob] {
            assert!(front.stat(&gone).await.unwrap().is_none(), "reclaimed");
        }
    }

    /// The memoization: an unchanged rows fingerprint consumes no
    /// walk; a changed one rebuilds; a counter bump alone (a denied
    /// CAS call) does not.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_memo_is_keyed_on_the_rows_fingerprint() {
        let (front, refs, writes) = mem_fixture();
        let fix = fixture_graph(&front).await;
        create_ref(refs.as_ref(), "repo", "refs/heads/main", fix.tip).await;
        let rows: Arc<dyn RefRows> = refs.clone();
        let source = RefClosure::new(front.clone(), rows, writes);

        alkblobs::LivenessSource::live_keys(source.as_ref());
        assert_eq!(source.keys_builds(), 1, "the first consult walks");

        alkblobs::LivenessSource::live_keys(source.as_ref());
        assert_eq!(
            source.keys_builds(),
            1,
            "an unchanged fingerprint consumes no walk"
        );

        // A counter bump without a row change (a denied call's bump)
        // does not rebuild — the fingerprint is the rows.
        source.writes.fetch_add(1, Ordering::Relaxed);
        let memoized = alkblobs::LivenessSource::live_keys(source.as_ref()).len();
        assert_eq!(memoized, 4, "the memo answers");
        assert_eq!(
            source.keys_builds(),
            1,
            "a counter bump alone does not rebuild"
        );

        // A row change (the same write, landed) moves the fingerprint:
        // rebuild.
        create_ref(refs.as_ref(), "repo", "refs/heads/topic", fix.base).await;
        alkblobs::LivenessSource::live_keys(source.as_ref());
        assert_eq!(source.keys_builds(), 2, "a fingerprint move rebuilds");
        let oids = live_oids(&source).await;
        assert_eq!(oids, sorted(vec![fix.tip, fix.base, fix.tree, fix.blob]));
    }

    /// The arbitration consult's discipline: indexed, quick, never
    /// deriving — exact on index hits, safe-default on a cold memo,
    /// unknown-safe after an in-process write, and the counter-driven
    /// answer never triggers a rebuild.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_contains_consult_never_derives() {
        let (front, refs, writes) = mem_fixture();
        let fix = fixture_graph(&front).await;
        create_ref(refs.as_ref(), "repo", "refs/heads/main", fix.tip).await;
        let rows: Arc<dyn RefRows> = refs.clone();
        let source = RefClosure::new(front.clone(), rows, writes);

        let tip_key = oid_key(&fix.tip).unwrap();
        let absent_key = oid_key(&ObjectId::from([1u8; 20])).unwrap();
        assert!(
            alkblobs::LivenessSource::contains(source.as_ref(), &tip_key),
            "a cold memo answers live — the safe default"
        );
        assert_eq!(source.keys_builds(), 0, "contains never derives");
        assert!(alkblobs::LivenessSource::contains(
            source.as_ref(),
            &absent_key
        ));

        alkblobs::LivenessSource::live_keys(source.as_ref());
        assert_eq!(source.keys_builds(), 1);
        assert!(alkblobs::LivenessSource::contains(
            source.as_ref(),
            &tip_key
        ));
        assert!(
            !alkblobs::LivenessSource::contains(source.as_ref(), &absent_key),
            "a built memo answers the index exactly"
        );

        // An in-process write moves the counter: unknown → live, no rebuild.
        source.writes.fetch_add(1, Ordering::Relaxed);
        assert!(
            alkblobs::LivenessSource::contains(source.as_ref(), &absent_key),
            "a moved write counter answers conservatively (unknown is live)"
        );
        assert_eq!(
            source.keys_builds(),
            1,
            "the arbitration consult never triggers the walk"
        );

        // The next mark rebuilds and the counter catches up at the
        // memo's build: exact answers resume.
        alkblobs::LivenessSource::live_keys(source.as_ref());
        assert!(!alkblobs::LivenessSource::contains(
            source.as_ref(),
            &absent_key
        ));
        assert!(alkblobs::LivenessSource::contains(
            source.as_ref(),
            &tip_key
        ));
    }

    /// The walk is best-effort: a ref tip with no object and a commit
    /// with a missing parent each stop their branch and never fail
    /// the consult (the corruption class rides the odb suite's
    /// verification tests — the same read legs).
    #[tokio::test(flavor = "multi_thread")]
    async fn the_walk_degrades_best_effort_on_missing_nodes() {
        let (front, refs, writes) = mem_fixture();
        let blob = put_object(&front, Kind::Blob, blob_content()).await;
        let tree = front
            .put(Kind::Tree, &tree_bytes(&[TreeEntry::Blob("hello", &blob)]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let missing_parent = ObjectId::from([9u8; 20]);
        let tip = front
            .put(Kind::Commit, &commit_bytes(&tree, &[&missing_parent]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let dangling = ObjectId::from([5u8; 20]);
        create_ref(refs.as_ref(), "repo", "refs/heads/main", tip).await;
        create_ref(refs.as_ref(), "repo", "refs/heads/dangling", dangling).await;

        let rows: Arc<dyn RefRows> = refs.clone();
        let source = RefClosure::new(front.clone(), rows, writes);
        let oids = live_oids(&source).await;
        assert_eq!(oids, sorted(vec![tip, tree, blob]));
    }

    /// The postures pinned by odb's bridge tests, re-pinned at the
    /// source: the live verb bridges with `block_in_place` — a
    /// current-thread runtime panics (the multi-thread runtime is
    /// the floor).
    #[tokio::test]
    #[should_panic]
    async fn the_live_verb_panics_on_a_current_thread_runtime() {
        let (front, refs, writes) = mem_fixture();
        let tip = put_object(&front, Kind::Blob, b"tip\n").await;
        create_ref(refs.as_ref(), "repo", "refs/heads/main", tip).await;
        let rows: Arc<dyn RefRows> = refs.clone();
        let source = RefClosure::new(front, rows, writes);
        alkblobs::LivenessSource::live_keys(source.as_ref());
    }

    /// The FNV fingerprint's determinism and sensitivity: equal rows
    /// fingerprint equal across calls; any row change moves it
    /// (name, oid, state, count); the canonical fingerprint is
    /// order-normalized (the impls pass sorted rows).
    #[test]
    fn the_fingerprint_is_deterministic_and_sensitive() {
        let row = |name: &str, oid: u8| RefRow {
            repo_id: "repo".to_owned(),
            name: name.to_owned(),
            state: RefRowState::Oid(ObjectId::from([oid; 20])),
        };
        let rows = vec![row("a", 1), row("b", 2)];
        let again = vec![row("a", 1), row("b", 2)];
        assert_eq!(fingerprint_rows(&rows), fingerprint_rows(&again));

        let moved_oid = vec![row("a", 1), row("b", 3)];
        let moved_name = vec![row("a", 1), row("c", 2)];
        let reordered = vec![row("b", 2), row("a", 1)];
        assert_ne!(fingerprint_rows(&rows), fingerprint_rows(&moved_oid));
        assert_ne!(fingerprint_rows(&rows), fingerprint_rows(&moved_name));
        assert_eq!(
            fingerprint_rows(&rows),
            fingerprint_rows(&reordered),
            "the canonical fingerprint is order-normalized"
        );

        let symref = RefRow {
            repo_id: "repo".to_owned(),
            name: "HEAD".to_owned(),
            state: RefRowState::Symref("refs/heads/main".to_owned()),
        };
        let with_symref = vec![row("a", 1), symref];
        assert_ne!(fingerprint_rows(&rows), fingerprint_rows(&with_symref));
        assert_ne!(fingerprint_rows(&[]), fingerprint_rows(&[row("a", 1)]));
    }

    /// The sqlite twin's snapshot seam: rows arrive in canonical
    /// order with the fingerprint over them; an unchanged table
    /// fingerprints equal, a row change moves it.
    #[cfg(feature = "sqlite")]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_sqlite_rows_seam_snapshots_canonically() {
        use crate::migrations::CoTenantSqlite;
        use crate::refs::SqliteRefTable;

        let dir = scratch_dir("liveness", "rows");
        let handle = CoTenantSqlite::open(dir.join("rows.sqlite3")).unwrap();
        let table = Arc::new(SqliteRefTable::new(handle));
        let tip = ObjectId::from([4u8; 20]);
        create_ref(table.as_ref(), "repo", "refs/heads/main", tip).await;
        create_ref(table.as_ref(), "other", "refs/heads/wip", tip).await;

        let rows: Arc<dyn RefRows> = table.clone();
        let first = rows.snapshot().await.unwrap();
        assert_eq!(first.rows.len(), 2);
        assert_eq!(first.rows[0].repo_id, "other", "repo id sorts first");
        assert_eq!(first.rows[1].repo_id, "repo");
        assert_eq!(first.fingerprint, fingerprint_rows(&first.rows));

        let second = rows.snapshot().await.unwrap();
        assert_eq!(
            first.fingerprint, second.fingerprint,
            "an unchanged table fingerprints equal"
        );
        create_ref(table.as_ref(), "repo", "refs/heads/topic", tip).await;
        let third = rows.snapshot().await.unwrap();
        assert_ne!(third.fingerprint, second.fingerprint);
        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }

    /// The versioned wrapper forwards the snapshot and bumps per
    /// **green** CAS call — the bump follows the committed rows (the
    /// epoch discipline's bump-after-commit half; a per-ref denial
    /// still returns a green call, so it bumps — over-invalidation,
    /// the safe direction — while an `Err` call bumps nothing: no
    /// rows landed); the assembly wiring test proves the composition
    /// path (sqlite.rs).
    #[cfg(feature = "sqlite")]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_versioned_wrapper_forwards_and_bumps() {
        use crate::migrations::CoTenantSqlite;
        use crate::refs::{SqliteRefTable, VersionedRefTable};

        let dir = scratch_dir("liveness", "wrapper");
        let handle = CoTenantSqlite::open(dir.join("wrapper.sqlite3")).unwrap();
        let inner = Arc::new(SqliteRefTable::new(handle));
        let counter = Arc::new(AtomicU64::new(0));
        let wrapper = Arc::new(VersionedRefTable::new(inner.clone(), counter.clone()));
        assert_eq!(counter.load(Ordering::Relaxed), 0, "listing does not bump");

        let tip = ObjectId::from([4u8; 20]);
        wrapper
            .apply_updates(
                &repo("repo"),
                vec![RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: None,
                    new: Some(tip),
                }],
                false,
            )
            .await
            .unwrap();
        assert_eq!(
            counter.load(Ordering::Relaxed),
            1,
            "an applied CAS call bumps the counter"
        );
        let rows: Arc<dyn RefRows> = wrapper.clone();
        assert_eq!(
            rows.snapshot().await.unwrap().rows.len(),
            1,
            "the snapshot forwards through the wrapper"
        );

        let denied = wrapper
            .apply_updates(
                &repo("repo"),
                vec![RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: Some(ObjectId::from([9u8; 20])),
                    new: Some(tip),
                }],
                false,
            )
            .await
            .unwrap();
        assert!(denied.iter().all(|o| o.result.is_err()));
        assert_eq!(
            counter.load(Ordering::Relaxed),
            2,
            "a green call bumps (the bump follows the committed rows; the denials ride the outcomes)"
        );
        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }

    /// The end-to-end wiring over the default engine composition:
    /// the sqlite assembly's co-tenancy (the alkstore + alkblobs
    /// stores, the co-tenant twins — one file), the object front, the
    /// versioned refs (the composition's `GitRefs` impl — the pushes'
    /// path), the registered `RefClosure`, and the sweep consuming
    /// it. Orphans reclaim across the zero-length durable window; a
    /// ref create covering a staged candidate cancels it (the
    /// arbitration window's counter answer — one delayed sweep); a
    /// ref delete uncovers and the closure reclaims.
    #[cfg(feature = "sqlite")]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_composition_wiring_serves_the_sweep_end_to_end() {
        use std::time::Duration;

        use alkblobs::StoreMode;
        use alkgit::GitRegistryStore as _;

        use crate::sqlite::SqliteRegistryAssembly;

        let dir = scratch_dir("liveness", "wiring");
        let db = dir.join("composition.sqlite3");
        let assembly = SqliteRegistryAssembly::open(&db).await.unwrap();
        let record = repo("owner/wired");

        let store = Arc::new(
            Store::kv_only_sqlite(&db, 128 * 1024, Duration::ZERO)
                .await
                .unwrap(),
        );
        assert_eq!(store.mode(), StoreMode::KvOnly);
        let front = ObjectFront::new(store.clone());

        // The wiring: rows and the write counter come straight from
        // the assembly's versioned refs — one handle for both.
        let rows: Arc<dyn RefRows> = assembly.refs().clone();
        let source = RefClosure::new(front.clone(), rows, assembly.refs_version().clone());
        source.register(&store).unwrap();

        // `put_repo` seeds the unborn HEAD symref row — part of the
        // snapshot, contributes nothing (the unborn shape).
        assembly.backend().put_repo(&record).await.unwrap();

        let covered = put_object(&front, Kind::Blob, b"covered\n").await;
        let orphan = put_object(&front, Kind::Blob, b"orphan\n").await;
        assert!(
            front.get(&covered).await.unwrap().is_some()
                && front.get(&orphan).await.unwrap().is_some(),
            "both land in the pool"
        );

        // Sweep 1 — mark with both objects uncovered: the durable
        // form stages both (the SQLite-hosted window; the zero-length
        // delay makes staged candidates overdue at the next sweep).
        let first = store.sweep().await.unwrap();
        assert!(first.aborted.is_none());
        assert_eq!(first.candidates_staged, 2, "both orphans stage");
        assert_eq!(first.deleted, 0);
        assert!(
            front.get(&orphan).await.unwrap().is_some(),
            "staged, not deleted"
        );
        assert!(front.get(&covered).await.unwrap().is_some());

        // The push path lands the creation through the composition's
        // versioned refs — the counter bumps and the coverage moves.
        assembly
            .refs()
            .apply_updates(
                &record,
                vec![RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: None,
                    new: Some(covered),
                }],
                false,
            )
            .await
            .unwrap();

        // Sweep 2 — the executor's re-arbitration consults `contains`;
        // the counter moved past the memo's build (unknown is live),
        // so BOTH staged candidates cancel — the pushed object cannot
        // be deleted here, and the over-conservative answer holds the
        // still-uncovered orphan for one sweep too. The mark rebuilds
        // (the rows moved; the memo's epoch catches up) and re-stages
        // the still-uncovered orphan.
        let second = store.sweep().await.unwrap();
        assert!(second.aborted.is_none());
        assert_eq!(second.deleted, 0);
        assert_eq!(
            second.cancelled_by_arbitration, 2,
            "the push-path bump cancels the staged candidates"
        );
        assert_eq!(second.candidates_staged, 1);
        assert!(front.get(&orphan).await.unwrap().is_some());
        assert!(front.get(&covered).await.unwrap().is_some());

        // Sweep 3 — the memo now keys on the post-create counter and
        // rows: the orphan's consult answers exactly, and it is gone.
        let third = store.sweep().await.unwrap();
        assert!(third.aborted.is_none());
        assert_eq!(
            third.deleted, 1,
            "the orphan reclaims once the rebuild caught up"
        );
        assert!(front.get(&orphan).await.unwrap().is_none());
        assert!(front.get(&covered).await.unwrap().is_some());

        // The ref delete uncovers the closure: sweep 4's mark rebuilds
        // (the rows moved) and stages it; sweep 5's consult answers
        // exactly (the epoch already caught up) and the closure
        // reclaims — the repo-deletion-is-GC shape through the full
        // chain.
        assembly
            .refs()
            .apply_updates(
                &record,
                vec![RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: Some(covered),
                    new: None,
                }],
                false,
            )
            .await
            .unwrap();
        let fourth = store.sweep().await.unwrap();
        assert!(fourth.aborted.is_none());
        assert_eq!(fourth.candidates_staged, 1, "the uncovered closure stages");
        assert_eq!(fourth.deleted, 0);
        assert!(
            front.get(&covered).await.unwrap().is_some(),
            "still in the window"
        );
        let fifth = store.sweep().await.unwrap();
        assert_eq!(
            fifth.deleted, 1,
            "the overdue uncovered closure reclaims (the epoch already caught up at the delete-time mark)"
        );
        assert!(front.get(&covered).await.unwrap().is_none());

        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }
}

/// The LFS media source's pins (the `lfs` feature): map-free keys, the
/// rebuild-per-mark row read, the arbitration consult's discipline,
/// and the composition wiring's durable-window sweep contract over the
/// sqlite rows twin (the mem shape's end-to-end rows ride the
/// composition entry's tests in [`crate::compose`]; the pg rows on the
/// live-pg harness ride the same entry pattern there).
#[cfg(all(test, feature = "lfs"))]
mod media_tests {
    use std::sync::atomic::AtomicU64;
    use std::sync::Arc;

    use alkblobs::{Algorithm, Store};
    use alkgit::{GitRefs, RefUpdate, RepoRecord};
    use gix_hash::ObjectId;
    use gix_object::Kind;
    use std::io::Cursor;

    use crate::lfs::InMemoryLfsRows;
    use crate::liveness::{lfs_media_key, oid_key, LfsMedia};
    use crate::odb::{key_to_oid, ObjectFront};
    use crate::refs::InMemoryRefTable;
    use crate::testutil::private_record as repo;
    #[cfg(feature = "sqlite")]
    use crate::testutil::scratch_dir;

    use super::*;

    async fn put_object(front: &ObjectFront, kind: Kind, content: &[u8]) -> ObjectId {
        let (key, _pin) = front.put(kind, content).await.unwrap();
        key_to_oid(&key).expect("the put's key is the family key")
    }

    fn writes() -> Arc<AtomicU64> {
        Arc::new(AtomicU64::new(0))
    }

    fn media_fixture() -> (Arc<Store>, Arc<dyn crate::lfs::LfsRows>, Arc<AtomicU64>) {
        let store = Arc::new(Store::mem_only());
        let counter = writes();
        let rows: Arc<dyn crate::lfs::LfsRows> = Arc::new(crate::lfs::VersionedLfsRows::new(
            Arc::new(InMemoryLfsRows::new()),
            Arc::clone(&counter),
        ));
        (store, rows, counter)
    }

    /// Put one LFS media entry under its own family: the store derives
    /// the key (the facade seam's put), the oid read off it is the
    /// digest's hex — the row store's oid shape.
    async fn put_media(store: &Store, content: &[u8]) -> (String, Key) {
        let (key, _pin) = store
            .put_as(
                Algorithm::GitObjectSha256,
                Some(content.len() as u64),
                Cursor::new(content),
            )
            .await
            .unwrap();
        let hex: String = key
            .digest()
            .bytes()
            .iter()
            .map(|b| format!("{b:02x}"))
            .collect();
        (hex, key)
    }

    fn contains_of(source: &Arc<LfsMedia>, key: &Key) -> bool {
        alkblobs::LivenessSource::contains(source.as_ref(), key)
    }

    /// The map-free formation: the put's derived key is re-formable
    /// from the digest hex alone — `[0x04] + oid bytes`, byte-exact —
    /// and every malformed oid form is a reject.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_media_keys_form_map_free_and_reject_the_rest() {
        let store = Store::mem_only();
        let content = b"lfs media bytes\n";
        let (hex, put_key) = put_media(&store, content).await;
        assert_eq!(hex.len(), 64, "the digest's hex is the row oid");
        assert_eq!(put_key.as_bytes()[0], 0x04, "the media family byte");

        let key = lfs_media_key(&hex).unwrap();
        assert_eq!(
            key, put_key,
            "the oid IS the digest — the key forms map-free"
        );
        assert_eq!(key.algorithm(), Algorithm::GitObjectSha256);
        assert_eq!(
            key.as_bytes(),
            put_key.as_bytes(),
            "the formed key is the physical encoding byte-exact"
        );

        let bad_forms: Vec<String> = vec![
            "".to_owned(),
            "nothex".to_owned(),
            "AB".repeat(32),
            "1".repeat(63),
            "1".repeat(65),
            "z".repeat(64),
        ];
        for bad in bad_forms {
            assert!(
                matches!(lfs_media_key(&bad), Err(StorageError::Invalid(_))),
                "a non-sha256-hex oid rejects: {bad:?}"
            );
        }
    }

    /// The mark consult reads the rows (rebuild-per-mark — every
    /// consult, no fingerprint memo); the dedup collapses one key with
    /// N rows to one live key; `contains` answers the index exactly,
    /// cold-live, and unknown-live after the assert counter moved —
    /// never deriving.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_media_source_marks_the_rows_and_answers_the_arbitration_from_the_index() {
        let (store, rows, counter) = media_fixture();
        let source = LfsMedia::new(rows.clone(), Arc::clone(&counter));

        let (_, key_absent) = put_media(&store, b"unasserted\n").await;
        assert!(
            contains_of(&source, &key_absent),
            "a cold source answers live — the safe default"
        );
        assert_eq!(source.keys_builds(), 0, "contains never derives");

        let (oid_a, key_a) = put_media(&store, b"shared\n").await;
        let (oid_b, key_b) = put_media(&store, b"other\n").await;
        rows.assert("r/one", &oid_a, 10).await.unwrap();
        rows.assert("r/two", &oid_a, 10).await.unwrap();
        rows.assert("r/one", &oid_b, 6).await.unwrap();

        let keys = alkblobs::LivenessSource::live_keys(source.as_ref());
        assert_eq!(
            keys.len(),
            2,
            "one key with two rows is one live key (the dedup)"
        );
        assert!(keys.contains(&key_a) && keys.contains(&key_b));
        assert_eq!(source.keys_builds(), 1);

        alkblobs::LivenessSource::live_keys(source.as_ref());
        assert_eq!(
            source.keys_builds(),
            2,
            "rebuild-per-mark: a consult re-reads the rows even unchanged"
        );

        assert!(contains_of(&source, &key_a), "the index answer is exact");
        assert!(
            !contains_of(&source, &key_absent),
            "an unasserted oid is exactly absent while the epoch is caught up"
        );

        // The dangerous window's safe side: a row lands after this
        // mark — the moved counter turns the miss conservative until
        // the next mark's rebuild resolves it.
        let (oid_c, key_c) = put_media(&store, b"landed-late\n").await;
        rows.assert("r/one", &oid_c, 12).await.unwrap();
        assert!(contains_of(&source, &key_c), "unknown is live");
        let keys = alkblobs::LivenessSource::live_keys(source.as_ref());
        assert_eq!(keys.len(), 3, "the next mark's rebuild answers exactly");
        assert!(contains_of(&source, &key_c), "now in the index");
    }

    /// The straddle regression (the epoch discipline's poison shape):
    /// an `assert` parks **inside its committing call** — bump not yet
    /// landed (bump-after-commit) — while the mark consult runs its
    /// load-before-read rows read; the mark therefore captures the
    /// pre-write rows with the pre-write epoch. The write lands, the
    /// bump follows, and the arbitration consult answers LIVE (the
    /// moved counter). The bump-before shape this test replaced
    /// answered exact-DEAD here: its bump landed ahead of the gated
    /// commit, the mark's load caught it while the rows read missed
    /// the uncommitted row, and the memo paired the included bump with
    /// the missed rows — the sweep would delete just-referenced media.
    struct GatedAssertLfsRows {
        inner: Arc<crate::lfs::InMemoryLfsRows>,
        gate: std::sync::Mutex<Option<tokio::sync::oneshot::Receiver<()>>>,
        arrived: tokio::sync::watch::Sender<bool>,
    }

    #[async_trait::async_trait]
    impl crate::lfs::LfsRows for GatedAssertLfsRows {
        async fn assert(&self, repo_id: &str, oid: &str, size: u64) -> Result<(), StorageError> {
            let rx = self.gate.lock().unwrap_or_else(|e| e.into_inner()).take();
            if let Some(rx) = rx {
                let _ = self.arrived.send(true);
                let _ = rx.await;
            }
            self.inner.assert(repo_id, oid, size).await
        }
        async fn remove(&self, repo_id: &str, oid: &str) -> Result<(), StorageError> {
            self.inner.remove(repo_id, oid).await
        }
        async fn remove_repo(&self, repo_id: &str) -> Result<(), StorageError> {
            self.inner.remove_repo(repo_id).await
        }
        async fn rows(&self) -> Result<Vec<crate::lfs::LfsRow>, StorageError> {
            self.inner.rows().await
        }
        async fn repo_rows(&self, repo_id: &str) -> Result<Vec<crate::lfs::LfsRow>, StorageError> {
            self.inner.repo_rows(repo_id).await
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn an_in_flight_assert_straddling_a_mark_answers_live_not_dead() {
        let (gate_tx, gate_rx) = tokio::sync::oneshot::channel();
        let (arrived_tx, mut arrived_rx) = tokio::sync::watch::channel(false);
        let counter = writes();
        let rows: Arc<dyn crate::lfs::LfsRows> = Arc::new(crate::lfs::VersionedLfsRows::new(
            Arc::new(GatedAssertLfsRows {
                inner: Arc::new(crate::lfs::InMemoryLfsRows::new()),
                gate: std::sync::Mutex::new(Some(gate_rx)),
                arrived: arrived_tx,
            }),
            Arc::clone(&counter),
        ));
        let source = LfsMedia::new(rows.clone(), Arc::clone(&counter));

        let oid = "ab".repeat(32);
        let key = lfs_media_key(&oid).unwrap();
        let task = tokio::spawn({
            let rows = rows.clone();
            async move { rows.assert("r/one", &oid, 10).await }
        });
        loop {
            if *arrived_rx.borrow_and_update() {
                break;
            }
            arrived_rx
                .changed()
                .await
                .expect("the gated assert arrives");
        }

        let keys = alkblobs::LivenessSource::live_keys(source.as_ref());
        assert!(
            keys.is_empty(),
            "the mark's rows read misses the in-flight (uncommitted) row"
        );
        let before = counter.load(Ordering::Relaxed);
        assert_eq!(
            before, 0,
            "bump-after-commit: no bump while the write is in flight"
        );

        gate_tx.send(()).expect("the gate opens");
        task.await
            .expect("the assert completes")
            .expect("the assert commits");
        assert_eq!(
            counter.load(Ordering::Relaxed),
            1,
            "the bump follows the landed row"
        );
        assert!(
            contains_of(&source, &key),
            "just-referenced media answers live — never exact-dead"
        );
    }

    /// The same straddle on the refs closure (the wave-7 source the
    /// wave-9 discipline fixed beside it): a gated `apply_updates`
    /// parks inside the write; the mark consult's snapshot read misses
    /// the uncommitted ref while its counter load catches
    /// (bump-before's) epoch — the fixed pairing answers live after
    /// the commit lands.
    struct GatedApplyRefTable {
        inner: Arc<InMemoryRefTable>,
        gate: std::sync::Mutex<Option<tokio::sync::oneshot::Receiver<()>>>,
        arrived: tokio::sync::watch::Sender<bool>,
    }

    #[async_trait::async_trait]
    impl GitRefs for GatedApplyRefTable {
        async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<alkgit::RefLine>, StorageError> {
            self.inner.list_refs(repo).await
        }
        async fn apply_updates(
            &self,
            repo: &RepoRecord,
            updates: Vec<RefUpdate>,
            atomic: bool,
        ) -> Result<Vec<alkgit::RefOutcome>, StorageError> {
            let rx = self.gate.lock().unwrap_or_else(|e| e.into_inner()).take();
            if let Some(rx) = rx {
                let _ = self.arrived.send(true);
                let _ = rx.await;
            }
            self.inner.apply_updates(repo, updates, atomic).await
        }
    }

    #[async_trait::async_trait]
    impl super::RefRows for GatedApplyRefTable {
        async fn snapshot(&self) -> Result<super::RefRowsSnapshot, StorageError> {
            self.inner.snapshot().await
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn an_in_flight_ref_write_straddling_a_mark_answers_live_not_dead() {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let tip = put_object(&front, Kind::Blob, b"straddled\n").await;
        let key = oid_key(&tip).expect("the blob oid's family key");

        let (gate_tx, gate_rx) = tokio::sync::oneshot::channel();
        let (arrived_tx, mut arrived_rx) = tokio::sync::watch::channel(false);
        let counter = writes();
        let wrapper = Arc::new(crate::refs::VersionedRefTable::new(
            Arc::new(GatedApplyRefTable {
                inner: Arc::new(InMemoryRefTable::new()),
                gate: std::sync::Mutex::new(Some(gate_rx)),
                arrived: arrived_tx,
            }),
            Arc::clone(&counter),
        ));
        let rows: Arc<dyn super::RefRows> = wrapper.clone();
        let source = RefClosure::new(front.clone(), rows, Arc::clone(&counter));

        let task = tokio::spawn({
            let wrapper = wrapper.clone();
            async move {
                wrapper
                    .apply_updates(
                        &repo("r/one"),
                        vec![RefUpdate {
                            name: "refs/heads/main".to_owned(),
                            expected: None,
                            new: Some(tip),
                        }],
                        false,
                    )
                    .await
            }
        });
        loop {
            if *arrived_rx.borrow_and_update() {
                break;
            }
            arrived_rx.changed().await.expect("the gated write arrives");
        }

        let keys = alkblobs::LivenessSource::live_keys(source.as_ref());
        assert!(
            keys.is_empty(),
            "the snapshot read misses the in-flight write"
        );
        assert_eq!(
            counter.load(Ordering::Relaxed),
            0,
            "bump-after-commit: no bump while the write is in flight"
        );

        gate_tx.send(()).expect("the gate opens");
        task.await
            .expect("the write completes")
            .expect("the write commits");
        assert_eq!(
            counter.load(Ordering::Relaxed),
            1,
            "the bump follows the row"
        );
        assert!(
            alkblobs::LivenessSource::contains(source.as_ref(), &key),
            "just-pushed refs' objects answer live — never exact-dead"
        );
    }

    /// The registration + sweep integration at the source level over a
    /// registered mem pool: the never-asserted media reclaims (POC-L's
    /// sweep datum: orphaned deleted, referenced survive), the
    /// referenced media survives.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_registered_media_source_serves_the_sweep_orphan_reclaim() {
        let (store, rows, counter) = media_fixture();
        let (oid_ref, _) = put_media(&store, b"referenced\n").await;
        let (oid_orphan, _) = put_media(&store, b"orphan\n").await;
        rows.assert("r/one", &oid_ref, 10).await.unwrap();

        let source = LfsMedia::new(rows, Arc::clone(&counter));
        source.register(&store).unwrap();

        let report = store.sweep().await.unwrap();
        assert!(
            report.aborted.is_none(),
            "a registered source never trips NoLivenessSources"
        );
        assert_eq!(
            report.deleted, 1,
            "the never-asserted media reclaims; the referenced survives"
        );
        let stat = |oid_hex: &str| {
            let hex = oid_hex.to_owned();
            let store = Arc::clone(&store);
            async move {
                store
                    .stat(&lfs_media_key(&hex).unwrap().as_bytes())
                    .await
                    .unwrap()
                    .is_some()
            }
        };
        assert!(stat(&oid_ref).await, "the referenced media stays");
        assert!(!stat(&oid_orphan).await, "the orphan is gone");
    }

    /// The composition's LFS wiring leg over the durable rows — the
    /// sqlite rows twin under the wrapper + the register, exactly the
    /// entry's construction shape (the mem shape's own end-to-end rows
    /// ride the compose entry's test): re-protect while any row
    /// remains (the cross-reference refcount realized across a
    /// durable sweep window — stage, then executed-overdue), the
    /// assert-across-window cancel (the counter's dangerous-direction
    /// coverage), and the repo-deletion reclaim through the twin's
    /// record-removal leg.
    #[cfg(feature = "sqlite")]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_composition_wiring_serves_the_lfs_sweep_across_the_durable_window() {
        use std::time::Duration;

        use alkgit::GitRegistryStore as _;

        use crate::sqlite::SqliteRegistryAssembly;

        let dir = scratch_dir("liveness", "media-wiring");
        let db = dir.join("lfs-sweep.sqlite3");
        let assembly = SqliteRegistryAssembly::open(&db).await.unwrap();
        let store = Arc::new(
            Store::kv_only_sqlite(&db, 128 * 1024, Duration::ZERO)
                .await
                .unwrap(),
        );

        let lfs_version = assembly.lfs_version().clone();
        let rows: Arc<dyn crate::lfs::LfsRows> = assembly.lfs_rows().clone();
        let source = super::LfsMedia::new(rows.clone(), Arc::clone(&lfs_version));
        source.register(&store).unwrap();

        async fn pool_has(store: &Store, hex: &str) -> bool {
            store
                .stat(&lfs_media_key(hex).unwrap().as_bytes())
                .await
                .unwrap()
                .is_some()
        }
        let put = |store: Arc<Store>, content: &'static [u8]| async move {
            let (hex, key) = put_media(&store, content).await;
            (hex, key)
        };

        let (shared_oid, _) = put(Arc::clone(&store), b"shared\n").await;
        rows.assert("repo-a", &shared_oid, 16).await.unwrap();
        rows.assert("repo-b", &shared_oid, 16).await.unwrap();

        let first = store.sweep().await.unwrap();
        assert!(first.aborted.is_none());
        assert_eq!(
            first.candidates_staged, 0,
            "the cross-referenced media never stages while any row stands"
        );
        assert_eq!(first.deleted, 0);

        rows.remove("repo-a", &shared_oid).await.unwrap();
        let second = store.sweep().await.unwrap();
        assert!(
            second.deleted == 0 && second.candidates_staged == 0,
            "one row removed — the survivor re-protects the key across the durable window"
        );
        assert!(pool_has(&store, &shared_oid).await);

        rows.remove("repo-b", &shared_oid).await.unwrap();
        let third = store.sweep().await.unwrap();
        assert_eq!(
            third.candidates_staged, 1,
            "the last row's removal makes the media sweep-eligible — staged into the window"
        );
        assert_eq!(third.deleted, 0);
        assert!(pool_has(&store, &shared_oid).await, "staged, not deleted");
        let fourth = store.sweep().await.unwrap();
        assert_eq!(
            fourth.deleted, 1,
            "the overdue candidate executes — the arbitration's index answered exact"
        );
        assert!(!pool_has(&store, &shared_oid).await);

        let (late_oid, _) = put(Arc::clone(&store), b"late orphan\n").await;
        let fifth = store.sweep().await.unwrap();
        assert_eq!(fifth.candidates_staged, 1, "the bare orphan stages");
        rows.assert("repo-c", &late_oid, 16).await.unwrap();
        let sixth = store.sweep().await.unwrap();
        assert_eq!(
            sixth.cancelled_by_arbitration, 1,
            "the row landed across the window — the moved counter cancels the \
             candidate (the dangerous direction the counter exists for)"
        );
        assert!(pool_has(&store, &late_oid).await, "the media survives");
        let seventh = store.sweep().await.unwrap();
        assert_eq!(
            seventh.deleted, 0,
            "the mark's rebuild-per-mark saw the row — protected while it stands"
        );

        let (bare_oid, _) = put(Arc::clone(&store), b"bare\n").await;
        let eighth = store.sweep().await.unwrap();
        assert_eq!(
            eighth.candidates_staged, 1,
            "the never-asserted orphan stages"
        );
        let ninth = store.sweep().await.unwrap();
        assert_eq!(
            ninth.deleted, 1,
            "orphan reclaim across the durable window (POC-L's datum)"
        );
        assert!(!pool_has(&store, &bare_oid).await);

        let (victim_oid, _) = put(Arc::clone(&store), b"victim media\n").await;
        use alkgit::RepoRecord;
        assembly
            .backend()
            .put_repo(&RepoRecord {
                repo_id: "victim".to_owned(),
                visibility: alkgit::Visibility::Private,
                grants: Default::default(),
            })
            .await
            .unwrap();
        rows.assert("victim", &victim_oid, 16).await.unwrap();
        let tenth = store.sweep().await.unwrap();
        assert_eq!(
            tenth.candidates_staged, 0,
            "the repo-referenced media re-protects"
        );
        assembly.backend().remove_repo("victim").await.unwrap();
        let eleventh = store.sweep().await.unwrap();
        assert_eq!(
            eleventh.candidates_staged, 1,
            "the repo deletion's rows leg leaves the media unreferenced — staged"
        );
        let twelfth = store.sweep().await.unwrap();
        assert_eq!(
            twelfth.deleted, 1,
            "nothing LFS-specific remains — the sweep reclaims"
        );

        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }
}