alkgit-substrate 0.1.0

//! The pin bridge's coordination half — the pending-pin registry and
//! the receive flow's committed-ref side (substrate.md §object storage,
//! "the pin bridge's mechanism and failure semantics" + "Pin custody
//! across the frozen trait seam"; ADR-020 §2; ADR-019 §7; ADR-013 §7–8
//! — the CAS timing the order realizes). Composition-internal by pin:
//! substrate.md §public API carries the receive-pack composition as
//! "internal to the trait impls, not a public member" — none of this
//! module's names join the OQ-03 inventory.
//!
//!
//! # The frozen seam and the custody problem
//!
//! The pinned trait signatures carry no guards:
//! [`GitPackIngest::prepare`](alkgit::GitPackIngest::prepare) returns
//! `PreparedPush` (the unpack report plus the verbatim update list) and
//! the wire layer then calls
//! [`GitRefs::apply_updates`](alkgit::GitRefs::apply_updates) with the
//! same list — the just-put objects' pin guards (one `Pin` per object
//! put; the store's `Batch` group for the batch-form puts) have nowhere
//! to ride the seam. The composition owns the custody internally: the
//! substrate `GitPackIngest` and `GitRefs` impls are constructed
//! together and share the one [`PendingPinRegistry`].
//!
//! # The entry lifecycle (substrate.md's pinned rules)
//!
//! - **retain (from prepare)** — the ingest implement retains the
//!   just-put objects' guards ([`PinCustody`]) into the registry,
//!   keyed by `repo_id` and carrying the prepared update list. The
//!   ng-leg prepare (pack-level validation failed) retains nothing —
//!   its guards drop with the prepare task, and nothing the push could
//!   reference ever enters the store's reachable set.
//! - **drain-with-correlation (from apply)** — each apply drains the
//!   pending entry whose update list matches the call's:
//!   content-addressed correlation over the prepared list (the verbatim
//!   handoff — exact ordered equality of [`RefUpdate`]s), because
//!   identical update lists guard identical CAS requirements (same
//!   `expected`s, same new tips, same closure) and so which twin
//!   drains is immaterial (substrate.md's pinned rule). The first
//!   matching entry drains (the oldest custody first); a twin whose
//!   sibling's entry got drained leaves its own entry — its apply
//!   arrives after the sibling's commit, reports the pinned stale
//!   denials, and the leftover prunes on the next apply (the bounded
//!   sweep-eligible custody the doc names).
//! - **prune-stale (from any apply)** — every apply with an update
//!   list first prunes the repo's stale entries: prepare-returned,
//!   apply-never-arrived (the client-disconnect class, the
//!   malformed-push-between-the-calls class) — identified as the
//!   entries whose update list does not match the incoming call's (the
//!   matching twin, if any, is left for the drain; an empty update
//!   list prunes nothing). The frozen seam carries no session
//!   identity, so this is the discriminating rule the doc pins; an
//!   in-flight sibling push whose entry lands under this rule releases
//!   its guards into the same sweep-eligible recovery the crash case
//!   rides — bounded by a sweep striking inside that push's remaining
//!   window, and never touching the *drained* entry's custody, which
//!   stays registry-held across the whole CAS.
//! - **drop-on-abort** — on the aborted CAS (the atomic-denial
//!   rollback the table layer pins as `atomic` + any denial) the
//!   drained entry's guards drop; a substrate `Err` drops them the
//!   same. Nothing partial survives: a failed push is a failed push
//!   (the entries are the same sweep-eligible recovery the crash case
//!   rides).
//!
//! # The committed-ref side ([`CommittedRefs`])
//!
//! The wrapped [`GitRefs`] implement — the receive flow's
//! committed-ref side, the one place the wire layer's frozen seam and
//! the receive flow's post-CAS legs can meet (the wire layer reaches
//! the backend only through `GitRefs::apply_updates`). Every
//! `apply_updates` call runs the composed order of substrate.md
//! §the receive-pack flow:
//!
//! 1. **stale-prune** — the registry's prune-stale rule above;
//! 2. **the inner CAS** — verbatim delegation to the wrapped table
//!    (one transaction per call; the drained entry's guards are still
//!    registry-held across it);
//! 3. **drain-only-after-commit** — on success, the matching entry
//!    drains and its guards release (commit custody) — the CAS
//!    transaction has committed inside the inner call by then; on the
//!    aborted CAS and on a substrate `Err`, the release is the
//!    drop-on-abort drop;
//! 4. **the post-CAS legs** — on a committed call with a nonempty
//!    update list, the per-push mutation-marker write and its
//!    post-commit announce (see below), then the band-1 status rides
//!    the returned outcomes.
//!
//! `list_refs` delegates verbatim (the advertisement path has no
//! custody and no legs).
//!
//! # The post-CAS legs (the pinned decisions — the wave-8 gate's
//! flagged rows)
//!
//! The legs ride [`RegistryMutations`], the legs surface the
//! composition entry wires per engine ([`MemMutationLegs`] on the mem
//! shape; the engine-twin assemblies implement the same trait over
//! their fused `record_mutation`). The pinned decisions:
//!
//! - **The announce-event shape**: the twins' existing fused shape —
//!   one marker write, then [`RegistryEvent::Mutation`] `{seq,
//!   repo_id}` announced (publish + notify + enqueue, fail-soft each).
//!   The seq is transactionally exact on the durable side; the event
//!   is the watermark leg that drives diff-driven replication's prompt
//!   pass without record-change noise (ADR-019 §7).
//! - **The leg-firing rule**: the legs ride exactly the committed
//!   ref-table transaction — a committed call with a nonempty update
//!   list, per-ref denials included (the transaction committed; the
//!   marker exists to say "this push landed here"). The aborted CAS
//!   (atomic rollback), a substrate `Err`, and the empty-update call
//!   carry no legs.
//! - **The error posture when the marker write fails after a committed
//!   CAS**: fail-soft, observation-only — the push outcome is the
//!   CAS's and is untouched; the mutation's `Mutation` event is simply
//!   not announced, and the reconcile sweep plus the next push's
//!   marker re-anchor the watermark (the wave-6 recipe: every announce
//!   leg's loss is the sweep's closure). No error surfaces to the wire
//!   layer.
//!
//! # The custody posture over a sweep window
//!
//! A pending entry's guards hold across GC until drained or dropped —
//! never partial repo state; the cost of a leaked guard is bounded
//! sweep-eligible custody (substrate.md's sentence, test-pinned in
//! this module's battery: a sweep cannot reclaim a bridged push's
//! objects mid-window). A pruned or dropped entry's objects recover
//! the crash way — the next sweep reclaims them (recover-without-
//! restore: the pins' durability is the process's).

use std::collections::HashMap;
use std::sync::Arc;

use alkblobs::{Batch, Pin};
use alkgit::{GitRefs, RefLine, RefOutcome, RefUpdate, RegistryError, RepoRecord, StorageError};
use async_trait::async_trait;

use crate::reactor::{Reactor, RegistryEvent};
use crate::registry::InMemoryRegistry;

/// The guard set a pending entry holds across the ingest → CAS span:
/// the lone-put pins ([`Pin`] per object put) or the store's batch
/// form (the [`Batch`] group — the ingest's `put_in_batch` working
/// set, one commit discipline). Both release by dropping: the batch's
/// drop is its commit's one-step release, the pins' drops decrement
/// their refcounts — a released entry's objects are sweep-eligible,
/// never partial repo state.
pub(crate) enum PinCustody {
    /// One pin per lone put ([`ObjectFront::put`]'s returned guards).
    Pins(Vec<Pin>),
    /// The batch-form put's group (the ingest implement's
    /// `put_in_batch` working set — the batch-form retains land with
    /// the ingest implement).
    Batch(Batch),
}

/// The registry's per-repo pending entries.
struct PendingEntry {
    updates: Vec<RefUpdate>,
    custody: PinCustody,
}

/// The pending-pin registry — the composition-internal custody seam
/// over the frozen trait signatures (the module doc carries the pinned
/// lifecycle). Keyed by `repo_id`, entries correlated by content (the
/// verbatim prepared update list); the whole lifecycle runs under one
/// short-held lock ([`parking_lot::Mutex`]) — retain/drain/prune are
/// sync, no-await sections.
pub(crate) struct PendingPinRegistry {
    repos: parking_lot::Mutex<HashMap<String, Vec<PendingEntry>>>,
}

impl Default for PendingPinRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl PendingPinRegistry {
    /// A fresh registry with no pending custody.
    pub(crate) fn new() -> Self {
        PendingPinRegistry {
            repos: parking_lot::Mutex::new(HashMap::new()),
        }
    }

    /// Retain (from prepare): the just-put objects' custody rides into
    /// the registry keyed by repo id, correlated with the prepared
    /// update list. Identical update lists guard identical CAS
    /// requirements — twins are separate entries, and which twin
    /// drains is immaterial.
    pub(crate) fn retain(&self, repo_id: &str, updates: Vec<RefUpdate>, custody: PinCustody) {
        self.repos
            .lock()
            .entry(repo_id.to_owned())
            .or_default()
            .push(PendingEntry { updates, custody });
    }

    /// Drain-with-correlation (from apply): the pending entry whose
    /// update list matches `updates` comes out, guards and all — the
    /// applying call owns the custody from here (it releases only
    /// after the CAS transaction commits, or drops it on abort). The
    /// first match drains (the oldest custody first).
    pub(crate) fn drain(&self, repo_id: &str, updates: &[RefUpdate]) -> Option<PinCustody> {
        let mut repos = self.repos.lock();
        let drained = {
            let entries = repos.get_mut(repo_id)?;
            let index = entries.iter().position(|entry| entry.updates == updates)?;
            Some(entries.remove(index))
        };
        if drained.is_some() {
            self.remove_empty_locked(&mut repos, repo_id);
        }
        drained.map(|entry| entry.custody)
    }

    /// Prune-stale (from any apply): the repo's entries whose update
    /// list does not match the incoming call's are gone — the
    /// prepare-returned, apply-never-arrived class; their guards drop
    /// here (the sweep-eligible recovery the doc names). The matching
    /// twin, if any, stays for the drain. Returns the pruned count
    /// (observation data).
    pub(crate) fn prune_stale(&self, repo_id: &str, updates: &[RefUpdate]) -> usize {
        let mut repos = self.repos.lock();
        let Some(entries) = repos.get_mut(repo_id) else {
            return 0;
        };
        let before = entries.len();
        entries.retain(|entry| entry.updates == updates);
        let pruned = before - entries.len();
        self.remove_empty_locked(&mut repos, repo_id);
        pruned
    }

    /// The repo's pending entry count (observation data — the tests'
    /// probe surface).
    #[cfg(test)]
    pub(crate) fn pending_count(&self, repo_id: &str) -> usize {
        self.repos.lock().get(repo_id).map(Vec::len).unwrap_or(0)
    }

    fn remove_empty_locked(&self, repos: &mut HashMap<String, Vec<PendingEntry>>, repo_id: &str) {
        if repos.get(repo_id).is_some_and(Vec::is_empty) {
            repos.remove(repo_id);
        }
    }
}

/// The receive flow's post-CAS legs surface — the per-push
/// mutation-marker write fused with its post-commit announce ([`Reactor`]'s
/// publish + notify + enqueue, fail-soft each). The composition entry
/// wires one implement per engine: the engine-twin assemblies (their
/// fused `record_mutation` over the durable twins' marker write) and
/// [`MemMutationLegs`] on the mem shape. The pinned decisions — the
/// announce-event shape, the leg-firing rule, and the marker-failure
/// posture — live in the module doc.
#[async_trait]
pub(crate) trait RegistryMutations: Send + Sync {
    /// The per-push mutation-marker write, announced
    /// ([`RegistryEvent::Mutation`] post-commit, fail-soft) — the
    /// receive flow's watermark leg (ADR-019 §7). The durable leg
    /// carries no marker-driven record change: `rows_since` never
    /// carries markers. A removed repo's marker is refused
    /// ([`RegistryError::NotFound`]).
    async fn record_mutation(&self, repo_id: &str) -> Result<i64, RegistryError>;
}

/// The mem shape's fused mutation legs — the in-memory default
/// adapter's durable marker write (the process-local sequence pointer)
/// fused with the reactor's post-commit announce, the twins'
/// assembly-level shape on the completeness engine. The composition
/// entry constructs it next to its mem registry wiring.
pub(crate) struct MemMutationLegs {
    durable: Arc<InMemoryRegistry>,
    reactor: Arc<Reactor>,
}

impl MemMutationLegs {
    /// Fuse the durable write side and the reactor that announces for
    /// it (the [`crate::reactor::AnnouncingRegistry`] shape, at the
    /// mutations surface).
    pub(crate) fn new(durable: Arc<InMemoryRegistry>, reactor: Arc<Reactor>) -> Self {
        MemMutationLegs { durable, reactor }
    }
}

#[async_trait]
impl RegistryMutations for MemMutationLegs {
    async fn record_mutation(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let seq = self.durable.record_mutation(repo_id).await?;
        self.reactor
            .announce(&RegistryEvent::Mutation {
                seq,
                repo_id: repo_id.to_owned(),
            })
            .await;
        Ok(seq)
    }
}

/// The receive flow's committed-ref side — the composition-internal
/// [`GitRefs`] implement wrapping any inner table (the engine twins'
/// row-CAS behind the composition's write-counter wrapper included),
/// the one place the wire layer's frozen seam and the receive flow's
/// post-CAS legs meet. The construction is the composition entry's
/// (substrate.md §public API: internal to the trait impls, not a
/// public member); the composed order of every `apply_updates` call
/// and the pinned leg decisions live in the module doc.
pub(crate) struct CommittedRefs {
    inner: Arc<dyn GitRefs>,
    pins: Arc<PendingPinRegistry>,
    mutations: Arc<dyn RegistryMutations>,
}

impl CommittedRefs {
    /// The wrapped table, the shared pending-pin registry, and the
    /// wired mutation legs (constructed together — the custody seam's
    /// one share).
    pub(crate) fn new(
        inner: Arc<dyn GitRefs>,
        pins: Arc<PendingPinRegistry>,
        mutations: Arc<dyn RegistryMutations>,
    ) -> Self {
        CommittedRefs {
            inner,
            pins,
            mutations,
        }
    }

    /// The drain-and-release at the call's two endings: on the
    /// committed CAS the drained entry's guards release (commit
    /// custody — the transaction committed inside the inner call); on
    /// the aborted CAS and on a substrate `Err` the release is the
    /// drop-on-abort drop (the same sweep-eligible recovery the crash
    /// case rides). Either way the entry is gone; a push with no
    /// matching entry (the packless, CAS-only path) releases nothing.
    fn release_drained(&self, repo_id: &str, updates: &[RefUpdate]) {
        match self.pins.drain(repo_id, updates) {
            Some(PinCustody::Pins(pins)) => drop(pins),
            Some(PinCustody::Batch(batch)) => drop(batch),
            None => {}
        }
    }
}

#[async_trait]
impl GitRefs for CommittedRefs {
    async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
        self.inner.list_refs(repo).await
    }

    async fn apply_updates(
        &self,
        repo: &RepoRecord,
        updates: Vec<RefUpdate>,
        atomic: bool,
    ) -> Result<Vec<RefOutcome>, StorageError> {
        let list = updates.clone();
        if !list.is_empty() {
            self.pins.prune_stale(&repo.repo_id, &list);
        }
        match self.inner.apply_updates(repo, updates, atomic).await {
            Err(e) => {
                self.release_drained(&repo.repo_id, &list);
                Err(e)
            }
            Ok(outcomes) => {
                let aborted = atomic && outcomes.iter().any(|outcome| outcome.result.is_err());
                self.release_drained(&repo.repo_id, &list);
                if !aborted && !list.is_empty() {
                    let _ = self.mutations.record_mutation(&repo.repo_id).await;
                }
                Ok(outcomes)
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
    use std::sync::Arc;
    use std::time::Duration;

    use alkstore::Store as _;
    use gix_hash::ObjectId;
    use gix_object::Kind;

    use alkgit::ATOMIC_PUSH_FAILURE_REASON;

    use super::*;
    use crate::odb::{key_to_oid, ObjectFront};
    use crate::refs::{InMemoryRefTable, CAS_STALE_REASON};
    use crate::registry::{CachedRegistryProvider, RegistryWrite};
    use crate::testutil::record as repo;

    fn up(name: &str, expected: Option<u8>, new: Option<u8>) -> RefUpdate {
        RefUpdate {
            name: name.to_owned(),
            expected: expected.map(|b| ObjectId::from([b; 20])),
            new: new.map(|b| ObjectId::from([b; 20])),
        }
    }

    fn up_oid(name: &str, expected: Option<ObjectId>, new: Option<ObjectId>) -> RefUpdate {
        RefUpdate {
            name: name.to_owned(),
            expected,
            new,
        }
    }

    fn mem_front() -> ObjectFront {
        ObjectFront::new(Arc::new(alkblobs::Store::mem_only()))
    }

    /// The ref-table-closure liveness source over an empty (or the
    /// test's own) in-memory refs table — the sweep-driving fixture
    /// (a registered source keeps the sweep un-aborted).
    fn closure_of(
        front: &ObjectFront,
        rows: &Arc<InMemoryRefTable>,
    ) -> Arc<crate::liveness::RefClosure> {
        crate::liveness::RefClosure::new(
            front.clone(),
            rows.clone() as Arc<dyn crate::liveness::RefRows>,
            Arc::new(AtomicU64::new(0)),
        )
    }

    /// The registry's observation helpers (the tests' probe surface —
    /// same-module privacy; not composition API).
    fn pending_count(pins: &PendingPinRegistry, repo_id: &str) -> usize {
        pins.repos.lock().get(repo_id).map(Vec::len).unwrap_or(0)
    }

    fn entry_present(pins: &PendingPinRegistry, repo_id: &str, updates: &[RefUpdate]) -> bool {
        pins.repos
            .lock()
            .get(repo_id)
            .is_some_and(|entries| entries.iter().any(|entry| entry.updates == updates))
    }

    /// The legs double: a call counter, an optional failure, and the
    /// seqs it returned — the marker-write probe.
    struct LegProbe {
        called: AtomicU64,
        fail: bool,
        seqs: parking_lot::Mutex<Vec<i64>>,
    }

    impl LegProbe {
        fn failing() -> Arc<Self> {
            Arc::new(LegProbe {
                called: AtomicU64::new(0),
                fail: true,
                seqs: parking_lot::Mutex::new(Vec::new()),
            })
        }

        fn recording() -> Arc<Self> {
            Arc::new(LegProbe {
                called: AtomicU64::new(0),
                fail: false,
                seqs: parking_lot::Mutex::new(Vec::new()),
            })
        }

        fn calls(&self) -> u64 {
            self.called.load(Ordering::Relaxed)
        }

        fn seqs(&self) -> Vec<i64> {
            self.seqs.lock().clone()
        }
    }

    #[async_trait]
    impl RegistryMutations for LegProbe {
        async fn record_mutation(&self, _repo_id: &str) -> Result<i64, RegistryError> {
            let call = self.called.fetch_add(1, Ordering::Relaxed) as i64 + 1;
            if self.fail {
                return Err(RegistryError::Io(
                    "marker leg suppressed (probe)".to_owned(),
                ));
            }
            self.seqs.lock().push(call);
            Ok(call)
        }
    }

    /// An inner table double that runs its probe while the CAS is in
    /// flight (between the entry into the call and the commit the
    /// wrapped side sees on return) — the drain-ordering pin's driver.
    struct ProbedRefs {
        probe: Box<dyn Fn() + Send + Sync>,
        outcome: Result<(), StorageError>,
    }

    #[async_trait]
    impl GitRefs for ProbedRefs {
        async fn list_refs(&self, _repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
            Ok(Vec::new())
        }

        async fn apply_updates(
            &self,
            _repo: &RepoRecord,
            _updates: Vec<RefUpdate>,
            _atomic: bool,
        ) -> Result<Vec<RefOutcome>, StorageError> {
            (self.probe)();
            match &self.outcome {
                Err(e) => Err(e.clone()),
                Ok(()) => Ok(Vec::new()),
            }
        }
    }

    /// The correlation rules: drain by content, one entry per apply,
    /// repo-keyed, and the custody rides out of the drain.
    #[tokio::test(flavor = "multi_thread")]
    async fn drain_correlates_by_content_one_entry_per_apply() {
        let pins = Arc::new(PendingPinRegistry::new());
        pins.retain(
            "r",
            vec![up("refs/heads/a", None, Some(1))],
            PinCustody::Pins(Vec::new()),
        );
        pins.retain(
            "r",
            vec![up("refs/heads/b", None, Some(2))],
            PinCustody::Pins(Vec::new()),
        );
        pins.retain(
            "other",
            vec![up("refs/heads/a", None, Some(1))],
            PinCustody::Pins(Vec::new()),
        );
        assert_eq!(pending_count(&pins, "r"), 2);

        assert!(
            pins.drain("r", &[up("refs/heads/c", None, Some(9))])
                .is_none(),
            "a non-matching list drains nothing"
        );
        assert!(
            pins.drain("r", &[up("refs/heads/a", None, Some(1))])
                .is_some(),
            "the repo's own matching entry drains"
        );
        assert!(
            pins.drain("r", &[up("refs/heads/a", None, Some(1))])
                .is_none(),
            "the other repo's twin is not this repo's entry"
        );
        assert!(
            pending_count(&pins, "r") > 0,
            "the other list's entry stays"
        );
        assert!(
            pins.drain("r", &[up("refs/heads/b", None, Some(2))])
                .is_some(),
            "the other list drains by its content"
        );
        assert_eq!(
            pending_count(&pins, "r"),
            0,
            "a fully drained repo tier is gone"
        );
    }

    /// Identical twins: two retains under one list, two drains each
    /// taking exactly one (which one is immaterial — both custodies
    /// come out), and a third apply drains nothing. Identical CAS
    /// requirements: after one twin's CAS (say) denied, the other's
    /// drain still carries the same custody out.
    #[tokio::test(flavor = "multi_thread")]
    async fn twin_entries_drain_one_per_apply() {
        let pins = Arc::new(PendingPinRegistry::new());
        let list = vec![up("refs/heads/main", None, Some(1))];
        pins.retain("repo", list.clone(), PinCustody::Pins(Vec::new()));
        pins.retain("repo", list.clone(), PinCustody::Pins(Vec::new()));

        assert!(pins.drain("repo", &list).is_some());
        assert!(pins.drain("repo", &list).is_some());
        assert!(pins.drain("repo", &list).is_none(), "exactly two drains");
    }

    /// The prune rule: the repo's non-matching entries go (the stale
    /// class) with their guards released — the sweep-eligible recovery
    /// the doc pins — while the matching twin stays for the drain.
    #[tokio::test(flavor = "multi_thread")]
    async fn prune_stale_releases_the_non_matching_class() {
        let front = mem_front();
        let store = front.store().clone();
        let rows = Arc::new(InMemoryRefTable::new());
        let source = closure_of(&front, &rows);
        source.register(&store).expect("the closure registers");

        let (_, pin_a) = front.put(Kind::Blob, b"bridged a\n").await.unwrap();
        let (_, pin_b) = front.put(Kind::Blob, b"bridged b\n").await.unwrap();
        let key_a = pin_a.key().clone();
        let key_b = pin_b.key().clone();

        let pins = Arc::new(PendingPinRegistry::new());
        pins.retain(
            "repo",
            vec![up("refs/heads/gone", None, Some(1))],
            PinCustody::Pins(vec![pin_a]),
        );
        pins.retain(
            "repo",
            vec![up("refs/heads/gone", None, Some(1))],
            PinCustody::Pins(vec![pin_b]),
        );
        pins.retain(
            "repo",
            vec![up("refs/heads/keep", None, Some(2))],
            PinCustody::Pins(Vec::new()),
        );

        let pruned = pins.prune_stale("repo", &[up("refs/heads/keep", None, Some(2))]);
        assert_eq!(pruned, 2, "both non-matching entries pruned");
        assert!(
            pins.drain("repo", &[up("refs/heads/keep", None, Some(2))])
                .is_some(),
            "the matching entry stayed for the drain"
        );
        assert_eq!(pending_count(&pins, "repo"), 0);

        let report = store.sweep().await.expect("the sweep runs");
        assert!(report.aborted.is_none());
        assert_eq!(
            report.deleted, 2,
            "the pruned custody's objects are sweep-eligible"
        );
        assert!(front
            .stat(&key_to_oid(&key_a).unwrap())
            .await
            .unwrap()
            .is_none());
        assert!(front
            .stat(&key_to_oid(&key_b).unwrap())
            .await
            .unwrap()
            .is_none());
    }

    /// The custody posture over a sweep window (substrate.md's
    /// test-pinned sentence): a pending entry's guards hold across GC
    /// until drained or dropped — a sweep cannot reclaim a bridged
    /// push's objects mid-window — and a drained-and-released entry's
    /// objects recover the crash way on the next sweep (release alone
    /// reclaims nothing).
    #[tokio::test(flavor = "multi_thread")]
    async fn a_pending_entry_holds_across_the_sweep_until_drained() {
        let front = mem_front();
        let store = front.store().clone();
        let rows = Arc::new(InMemoryRefTable::new());
        let source = closure_of(&front, &rows);
        source.register(&store).expect("the closure registers");

        let (_, pin) = front.put(Kind::Blob, b"bridged\n").await.unwrap();
        let oid = key_to_oid(pin.key()).unwrap();

        let pins = Arc::new(PendingPinRegistry::new());
        let list = vec![up("refs/heads/main", None, Some(1))];
        pins.retain("repo", list.clone(), PinCustody::Pins(vec![pin]));

        let report = store.sweep().await.expect("the sweep runs");
        assert!(report.aborted.is_none());
        assert_eq!(
            report.deleted, 0,
            "the bridged push's objects are not reclaimable"
        );
        assert!(
            front.stat(&oid).await.unwrap().is_some(),
            "held across the window"
        );

        let custody = pins.drain("repo", &list).expect("the entry drains");
        drop(custody);

        assert!(
            front.stat(&oid).await.unwrap().is_some(),
            "release alone reclaims nothing"
        );
        let report = store.sweep().await.expect("the sweep runs");
        assert_eq!(
            report.deleted, 1,
            "the released objects recover the crash way"
        );
        assert!(front.stat(&oid).await.unwrap().is_none());
    }

    /// The committed-ref side's composed order: the inner CAS runs
    /// while the matching entry's guards are still registry-held, and
    /// the drain lands only after the CAS returns (the transaction
    /// committed inside it).
    #[tokio::test(flavor = "multi_thread")]
    async fn the_drain_lands_only_after_the_cas_returns() {
        let front = mem_front();
        let pins = Arc::new(PendingPinRegistry::new());
        let list = vec![up("refs/heads/main", None, Some(1))];
        let (_, pin) = front.put(Kind::Blob, b"held\n").await.unwrap();
        pins.retain("repo", list.clone(), PinCustody::Pins(vec![pin]));

        let saw_held = Arc::new(AtomicBool::new(false));
        let during_cas = pins.clone();
        let key_at_cas = list.clone();
        let saw = saw_held.clone();
        let probe = move || {
            saw.store(
                entry_present(&during_cas, "repo", &key_at_cas),
                Ordering::Relaxed,
            );
        };
        let inner: Arc<dyn GitRefs> = Arc::new(ProbedRefs {
            probe: Box::new(probe),
            outcome: Ok(()),
        });
        let wrapped = CommittedRefs::new(inner, pins.clone(), LegProbe::recording());

        let outcomes = wrapped
            .apply_updates(&repo("repo"), list.clone(), false)
            .await
            .expect("the committed CAS");
        assert!(outcomes.is_empty());
        assert!(
            saw_held.load(Ordering::Relaxed),
            "the matching entry's guards were registry-held across the CAS"
        );
        assert!(
            !entry_present(&pins, "repo", &list),
            "the drain landed only after the CAS returned"
        );
        assert_eq!(pending_count(&pins, "repo"), 0);
    }

    /// The committed-CAS ending over the real machinery: custody
    /// commits post-CAS and the legs ride — the marker write anchors
    /// the watermark and the `Mutation` announce is in the durable
    /// feed (the mem shape's fused legs, replayed from head). A
    /// packless-path push (no matching entry) rides the legs the same
    /// way; release alone reclaims nothing.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_committed_cas_drains_and_rides_the_legs() {
        let registry = Arc::new(InMemoryRegistry::new());
        registry.put(&repo("repo")).await.unwrap();
        let provider = Arc::new(CachedRegistryProvider::new());
        let mem_store = Arc::new(alkstore_mem::MemStore::new());
        let reactor = Arc::new(Reactor::new(
            mem_store.clone(),
            registry.clone(),
            provider.clone(),
        ));
        let legs = Arc::new(MemMutationLegs::new(registry.clone(), reactor.clone()));

        let front = mem_front();
        let (_, pin) = front.put(Kind::Blob, b"landed\n").await.unwrap();
        let key = pin.key().clone();

        let pins = Arc::new(PendingPinRegistry::new());
        let inner: Arc<dyn GitRefs> = Arc::new(InMemoryRefTable::new());
        let wrapped = CommittedRefs::new(inner, pins.clone(), legs);
        let list = vec![up("refs/heads/main", None, Some(1))];
        pins.retain("repo", list.clone(), PinCustody::Pins(vec![pin]));

        let outcomes = wrapped
            .apply_updates(&repo("repo"), list, false)
            .await
            .expect("the committed CAS");
        assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));
        assert_eq!(pending_count(&pins, "repo"), 0, "the entry drained");

        let seq = registry.watermark().await.unwrap();
        assert_eq!(
            seq, 2,
            "the marker write anchored the watermark (1 = the put)"
        );

        let stream = mem_store
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .unwrap();
        let mut rx = stream.subscribe("gate").await.unwrap();
        let event = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .expect("the announce lands inside the deadline")
            .expect("the feed delivered")
            .expect("the event is clean");
        match RegistryEvent::decode_event(&event) {
            Some(RegistryEvent::Mutation { seq: 2, repo_id }) => {
                assert_eq!(repo_id, "repo");
            }
            other => panic!("the Mutation announce rides the marker write: {other:?}"),
        }
        assert!(front
            .stat(&key_to_oid(&key).unwrap())
            .await
            .unwrap()
            .is_some());
    }

    /// The stale-entry class, wrapper-level (the pinned acceptance
    /// row): a prepare whose apply never arrives leaves a stale entry,
    /// the next apply prunes it alongside its own drain, and the
    /// pruned custody's objects recover GC-decreed — the following
    /// sweep reclaims the orphan.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_next_apply_prunes_the_stale_entry() {
        let front = mem_front();
        let store = front.store().clone();
        let rows = Arc::new(InMemoryRefTable::new());
        let source = closure_of(&front, &rows);
        source.register(&store).expect("the closure registers");

        let legs = LegProbe::recording();
        let pins = Arc::new(PendingPinRegistry::new());
        let inner: Arc<dyn GitRefs> = Arc::new(InMemoryRefTable::new());
        let wrapped = CommittedRefs::new(inner, pins.clone(), legs);

        let (_, stale_pin) = front.put(Kind::Blob, b"never applied\n").await.unwrap();
        let stale_oid = key_to_oid(stale_pin.key()).unwrap();
        pins.retain(
            "repo",
            vec![up("refs/heads/stale", None, Some(1))],
            PinCustody::Pins(vec![stale_pin]),
        );

        let (_, live_pin) = front
            .put(Kind::Blob, b"the live push's object\n")
            .await
            .unwrap();
        let live_oid = key_to_oid(live_pin.key()).unwrap();
        let live_list = vec![up_oid("refs/heads/live", None, Some(live_oid))];

        let outcomes = wrapped
            .apply_updates(&repo("repo"), live_list, false)
            .await
            .expect("the next apply runs");
        assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));
        assert_eq!(
            pending_count(&pins, "repo"),
            0,
            "the stale entry pruned alongside the drain"
        );

        let report = store.sweep().await.expect("the sweep runs");
        assert_eq!(report.deleted, 1, "the pruned guards recovered GC-decreed");
        assert!(front.stat(&stale_oid).await.unwrap().is_none());
        assert!(
            front.stat(&live_oid).await.unwrap().is_some(),
            "the landed ref's object covered"
        );
    }

    /// The aborted-CAS ending (the atomic-denial rollback): the
    /// drained entry's guards drop, no legs ride — a failed push is a
    /// failed push, and the objects recover GC-decreed.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_aborted_cas_drops_the_guards_and_skips_the_legs() {
        let front = mem_front();
        let store = front.store().clone();
        let rows = Arc::new(InMemoryRefTable::new());
        let source = closure_of(&front, &rows);
        source.register(&store).expect("the closure registers");

        let legs = LegProbe::recording();
        let pins = Arc::new(PendingPinRegistry::new());
        let inner: Arc<dyn GitRefs> = Arc::new(InMemoryRefTable::new());
        let wrapped = CommittedRefs::new(inner, pins.clone(), legs.clone());

        let (_, base_pin) = front.put(Kind::Blob, b"the seed's object\n").await.unwrap();
        let base_oid = key_to_oid(base_pin.key()).unwrap();
        let seed = wrapped
            .apply_updates(
                &repo("repo"),
                vec![up_oid("refs/heads/base", None, Some(base_oid))],
                false,
            )
            .await
            .unwrap();
        assert!(seed.iter().all(|outcome| outcome.result.is_ok()));
        assert_eq!(legs.calls(), 1);

        let (_, pin) = front.put(Kind::Blob, b"aborted\n").await.unwrap();
        let aborted_oid = key_to_oid(pin.key()).unwrap();
        let list = vec![
            up_oid(
                "refs/heads/aborted-sibling",
                None,
                Some(ObjectId::from([9; 20])),
            ),
            up_oid(
                "refs/heads/base",
                Some(ObjectId::from([8; 20])),
                Some(ObjectId::from([9; 20])),
            ),
        ];
        pins.retain("repo", list.clone(), PinCustody::Pins(vec![pin]));

        let outcomes = wrapped
            .apply_updates(&repo("repo"), list, true)
            .await
            .expect("the aborted CAS is a business outcome");
        assert_eq!(
            outcomes[0].result,
            Err(ATOMIC_PUSH_FAILURE_REASON.to_owned())
        );
        assert_eq!(pending_count(&pins, "repo"), 0, "the drained entry is gone");
        assert_eq!(legs.calls(), 1, "the aborted CAS carries no legs");

        let report = store.sweep().await.expect("the sweep runs");
        assert_eq!(report.deleted, 1, "the dropped guards' objects are orphans");
        assert!(front.stat(&aborted_oid).await.unwrap().is_none());
        assert!(
            front.stat(&base_oid).await.unwrap().is_some(),
            "the seeded ref's object covered"
        );
    }

    /// The substrate-`Err` ending: the transaction never committed —
    /// the guards drop, no legs ride, and the error propagates.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_substrate_err_drops_the_guards_and_skips_the_legs() {
        let legs = LegProbe::recording();
        let pins = Arc::new(PendingPinRegistry::new());
        let list = vec![up("refs/heads/main", None, Some(1))];
        pins.retain("repo", list.clone(), PinCustody::Pins(Vec::new()));

        let inner: Arc<dyn GitRefs> = Arc::new(ProbedRefs {
            probe: Box::new(|| ()),
            outcome: Err(StorageError::Io("engine failure".to_owned())),
        });
        let wrapped = CommittedRefs::new(inner, pins.clone(), legs.clone());
        let err = wrapped
            .apply_updates(&repo("repo"), list, false)
            .await
            .expect_err("the substrate failure propagates");
        assert!(err.to_string().contains("engine failure"));
        assert_eq!(pending_count(&pins, "repo"), 0, "the guards dropped");
        assert_eq!(legs.calls(), 0, "the Err ending carries no legs");
    }

    /// The marker-failure posture after a committed CAS: fail-soft,
    /// observation-only — the push outcome is the CAS's and is
    /// untouched, and the custody releases regardless.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_failed_marker_write_never_fails_the_push() {
        let legs = LegProbe::failing();
        let pins = Arc::new(PendingPinRegistry::new());
        let inner: Arc<dyn GitRefs> = Arc::new(InMemoryRefTable::new());
        let wrapped = CommittedRefs::new(inner, pins.clone(), legs.clone());

        let outcomes = wrapped
            .apply_updates(
                &repo("repo"),
                vec![up("refs/heads/main", None, Some(1))],
                false,
            )
            .await
            .expect("the committed CAS is the push outcome");
        assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));
        assert_eq!(legs.calls(), 1);
        assert_eq!(pending_count(&pins, "repo"), 0);
    }

    /// The leg-firing rule: the empty-update call commits but carries
    /// no legs; a denied non-atomic call (a committed transaction with
    /// zero landed refs) still rides them.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_leg_firing_rule_pins_the_committed_transactions() {
        let legs = LegProbe::recording();
        let pins = Arc::new(PendingPinRegistry::new());
        let inner: Arc<dyn GitRefs> = Arc::new(InMemoryRefTable::new());
        let wrapped = CommittedRefs::new(inner, pins.clone(), legs.clone());

        let outcomes = wrapped
            .apply_updates(&repo("repo"), Vec::new(), false)
            .await
            .expect("the empty call is a committed nothing");
        assert!(outcomes.is_empty());
        assert_eq!(legs.calls(), 0, "the empty-update call carries no legs");

        let denied = wrapped
            .apply_updates(
                &repo("repo"),
                vec![up("refs/heads/base", Some(3), Some(4))],
                false,
            )
            .await
            .expect("denials are outcomes");
        assert_eq!(denied[0].result, Err(CAS_STALE_REASON.to_owned()));
        assert_eq!(legs.calls(), 1);
        assert_eq!(legs.seqs(), vec![1], "one marker per committed push");
    }

    /// `list_refs` delegates verbatim (no custody, no legs on the
    /// advertisement path).
    #[tokio::test(flavor = "multi_thread")]
    async fn list_refs_delegates_verbatim() {
        let rows = Arc::new(InMemoryRefTable::new());
        let list = vec![up("refs/heads/main", None, Some(7))];
        rows.apply_updates(&repo("repo"), list, false)
            .await
            .unwrap();
        let wrapped = CommittedRefs::new(
            rows.clone() as Arc<dyn GitRefs>,
            Arc::new(PendingPinRegistry::new()),
            LegProbe::recording(),
        );
        let lines = wrapped.list_refs(&repo("repo")).await.unwrap();
        assert_eq!(lines.len(), 1);
        assert_eq!(lines[0].name, "refs/heads/main");
        assert_eq!(lines[0].oid, Some(ObjectId::from([7; 20])));
    }
}