mkit-cli 0.5.0

//! The packmap-chain transfer layer for `mkit push` / `mkit fetch`.
//!
//! A branch's transfer history is a content-addressed singly-linked list
//! of packlist nodes — the *packmap* — advertised through the
//! `refs/mkit/packmap/<branch>` metadata ref. Each push appends one node
//! (a `prev` pointer plus the pack(s) it added); the fetch side walks the
//! chain oldest-first so every delta's base is already in the store when
//! its pack is unpacked.
//!
//! This module co-locates both ends of that machinery so the push-side
//! *advance* and the fetch-side *resolve / fetch* logic — which share the
//! same chain-walk integrity rules — sit next to each other:
//!
//! * push side: [`advance_packmap`] (chains one or more new packs on, gated by a CAS
//!   on the packmap ref).
//! * fetch side: [`resolve_pack_chain`] (walk + integrity check),
//!   [`resolve_and_download_chain`] (walk + download + private staging, no
//!   object-store writes), and [`apply_fetched_chain`] (unpack + verify + publish
//!   pre-check — must run under the repo lock).
//!
//! The fetch side is deliberately split into a network phase and a
//! local-write phase (#642): a branch's packs are fully downloaded before
//! the repo lock is taken, and the lock is held only from the first local
//! write through the branch's ref publish (in
//! [`super::fetch_objects_inner`]) — narrow enough that a slow transfer no
//! longer serializes out unrelated commands for its whole duration, while
//! still closing the #267 GC-prune race (a concurrent `gc` needs the same
//! lock, so it can never observe a downloaded-but-unpublished object).
//!
//! All entry points keep `pub(crate)` visibility so the orchestration in
//! the parent [`super`] module (`push_branch`, `fetch_objects`) can call
//! them.

use std::io::{Read as _, Write as _};

use mkit_core::hash::{self, Hash};
use mkit_core::object::Object;
use mkit_core::pack::{self, PackReader};
use mkit_core::protocol::{AdvanceOutcome, CommitOutcome, PackKey, Transport, TransportError};
use mkit_core::refs;
use mkit_core::sign;
use mkit_core::store::ObjectStore;
use mkit_core::transfer;
use rayon::prelude::*;

use super::DispatchError;
use super::applied_packs::AppliedPacks;

/// Ref under which a branch's transfer packlist key is advertised. The
/// value is the BLAKE3 of the packlist chain head — an auxiliary
/// content-addressed blob (moved via [`Transport::upload_blob`] /
/// [`Transport::download_blob`], not a packfile); the fetch side reads it to
/// discover every pack needed to reconstruct the branch. Lives in the
/// `refs/mkit/` metadata namespace so it never collides with real
/// heads/tags. See [`mkit_core::transfer`].
pub(crate) fn packmap_ref(branch: &str) -> String {
    format!("refs/mkit/packmap/{branch}")
}

/// Push and fetch callers may carry a short branch name or its full ref name.
fn branch_ref_hint(branch: &str) -> String {
    format!(
        "refs/heads/{}",
        branch.strip_prefix("refs/heads/").unwrap_or(branch)
    )
}

/// Number of read-modify-write attempts when chaining a new pack onto the
/// packmap. Each conflict means another pusher advanced the packmap; we
/// re-read and retry. Exhaustion (sustained contention) aborts the push
/// *before* the branch ref moves — see [`advance_packmap`].
const PACKMAP_CAS_ATTEMPTS: u32 = 8;

/// Hard cap on packmap **chain length** (number of linked nodes), i.e.
/// chain DEPTH — distinct from [`transfer::PACKLIST_MAX_ENTRIES`], which
/// caps the packs recorded in a single node.
///
/// A `prev`-linked chain is attacker-influenced: a malicious or corrupt
/// remote can advertise an arbitrarily long (or cyclic) chain to make the
/// fetcher walk forever issuing one blob download per node. We bound the
/// walk so such a remote surfaces as [`DispatchError::PackChainInvalid`]
/// instead of hanging. (A cycle is already caught by the `seen` set; this
/// bound additionally caps an *acyclic* but pathologically long chain.)
///
/// `100_000` nodes is well beyond any honest history — one node is appended
/// per push, and periodic re-baselining (#406, see [`rebaseline_depth`])
/// now collapses a healthy chain back to a single node long before it gets
/// anywhere near this cap — while still being cheap to reject. It is
/// deliberately NOT tied to the per-node entry cap: conflating "packs in one
/// node" with "nodes in the chain" overloads one number for two unrelated
/// bounds. This constant remains the runaway/cycle guard for a chain that
/// re-baselining never got a chance to bound (e.g. a hostile or corrupt
/// remote advertising an ever-growing or cyclic chain) — AND, since mkit
/// #521, the *only* bound on a healthy chain's growth on a transport whose
/// `advance_refs` is not transactional (see
/// [`Transport::supports_atomic_advance`]): such a transport never
/// re-baselines (a reset is unsafe there), so it keeps appending past
/// [`rebaseline_depth`] until this cap.
const MAX_PACK_CHAIN_DEPTH: usize = 100_000;

/// Bounds retained chain/stage metadata across all nodes, not only within one
/// packlist. At most one million pack keys plus one hundred thousand nodes.
const MAX_FETCH_PACKS: usize = 1_000_000;
/// Private download staging has a separate disk ceiling from one-pack memory.
/// A larger history must be fetched incrementally or re-baselined by its owner.
const MAX_FETCH_STAGED_BYTES: u64 = 64 * 1024 * 1024 * 1024;

/// Default chain depth at which a push re-baselines: resets the packlist
/// chain to a single fresh self-contained node instead of appending to it
/// (#406). Bounds clone cost, which otherwise grows with the chain length
/// (one node walked/downloaded per push since the last re-baseline).
/// Overridable via `MKIT_PACK_REBASELINE_DEPTH` for tests; `0` disables
/// re-baselining entirely.
const DEFAULT_REBASELINE_DEPTH: usize = 64;

/// Resolve the configured re-baseline threshold: the `MKIT_PACK_REBASELINE_DEPTH`
/// environment variable if present and parsable, else [`DEFAULT_REBASELINE_DEPTH`].
/// A value of `0` disables re-baselining (the push path never forces a
/// full-closure reset on depth alone).
pub(crate) fn rebaseline_depth() -> usize {
    // Unset (or non-UTF-8) → the default. A present-but-unparsable value is
    // an operator mistake (`-1`, `off`, `""` — none of which disable; only
    // `0` does), so warn loudly instead of silently re-enabling the default.
    match std::env::var("MKIT_PACK_REBASELINE_DEPTH") {
        Err(_) => DEFAULT_REBASELINE_DEPTH,
        Ok(s) => s.parse::<usize>().unwrap_or_else(|_| {
            eprintln!(
                "warning: MKIT_PACK_REBASELINE_DEPTH='{s}' is not a valid non-negative \
                 integer; using the default {DEFAULT_REBASELINE_DEPTH} (set it to 0 to \
                 disable re-baselining)"
            );
            DEFAULT_REBASELINE_DEPTH
        }),
    }
}

/// Download and decode one packlist node by key. Packlist nodes are
/// auxiliary transfer metadata, not packfiles, so they travel over the
/// dedicated [`Transport::download_blob_via_ref`] verb, with the branch hint.
fn download_packlist_node(
    tx: &dyn Transport,
    key: Hash,
    ref_name: &str,
) -> Result<transfer::PackListNode, DispatchError> {
    let requested = PackKey::from_hash(key);
    let bytes = tx.download_blob_via_ref(&requested, ref_name)?;
    requested.verify_bytes(&bytes)?;
    Ok(transfer::decode_packlist(&bytes)?)
}

/// Walk a branch's packlist chain from `head_key` (newest node) following
/// `prev` pointers, applying the shared cycle / runaway-depth guard, and
/// return every node visited in newest-to-oldest walk order.
///
/// This is the ONE place the chain walk is written: [`resolve_pack_chain`]
/// (which flattens the visited nodes' packs into the oldest-first fetch
/// order) and [`probe_chain`] (which needs both the visited-node count and
/// the flattened packs, for the push-side pre-plan re-baseline probe) both
/// call through here rather than re-implementing the walk.
///
/// This walks the WHOLE chain, so it doubles as the push-side integrity
/// check: a node the chain references but the remote can't deliver/decode,
/// a cycle, or an over-deep chain all surface as
/// [`DispatchError::PackChainInvalid`]. A transient transport error (network
/// blip) propagates unchanged so it is never mistaken for corruption. The
/// walk is `O(chain length)` node downloads; chain depth is bounded by
/// periodic re-baselining (#406, see [`rebaseline_depth`]), and a
/// server-side atomic advance (#408) could move this check off the hot path.
fn walk_pack_chain(
    tx: &dyn Transport,
    branch: &str,
    head_key: Hash,
) -> Result<Vec<transfer::PackListNode>, DispatchError> {
    let invalid = || DispatchError::PackChainInvalid {
        branch: branch.to_owned(),
    };
    let mut nodes = Vec::new();
    let mut pack_count = 0usize;
    let mut seen = std::collections::HashSet::new();
    let mut cursor = Some(head_key);
    let ref_name = branch_ref_hint(branch);
    while let Some(key) = cursor {
        if crate::signal::is_shutdown() {
            return Err(DispatchError::Interrupted);
        }
        // Cycle / runaway-depth guard.
        if !seen.insert(key) || seen.len() > MAX_PACK_CHAIN_DEPTH {
            return Err(invalid());
        }
        let node = match download_packlist_node(tx, key, &ref_name) {
            Ok(n) => n,
            // A referenced-but-undeliverable / undecodable node = broken
            // chain (distinct from a transient transport error).
            Err(
                DispatchError::Transport(
                    TransportError::PackNotFound | TransportError::InvalidResponse,
                )
                | DispatchError::PackList(_),
            ) => return Err(invalid()),
            Err(e) => return Err(e),
        };
        pack_count = pack_count
            .checked_add(node.packs.len())
            .ok_or_else(invalid)?;
        if pack_count > MAX_FETCH_PACKS {
            return Err(invalid());
        }
        cursor = node.prev;
        nodes.push(node);
    }
    Ok(nodes)
}

/// Walk a branch's packlist chain from `head_key` and return the flat
/// **oldest-first** list of pack keys it references — the order a fetcher
/// must unpack so each delta's base is already present. See
/// [`walk_pack_chain`] for the walk / integrity-check semantics.
pub(crate) fn resolve_pack_chain(
    tx: &dyn Transport,
    branch: &str,
    head_key: Hash,
) -> Result<Vec<Hash>, DispatchError> {
    // Defined in terms of `probe_chain` so the flattened pack list is
    // identical BY CONSTRUCTION to `probe_chain(..).packs` (the equivalence
    // the `probe_chain_*_matches_resolve_pack_chain` test asserts) — the two
    // can't drift. `probe_chain` can't delegate the other way: it also needs
    // the node count for `depth`.
    Ok(probe_chain(tx, branch, head_key)?.packs)
}

/// A branch's packmap chain, walked exactly once, paired with the packmap
/// value it was walked from.
///
/// Produced by [`probe_chain`] (the push-side pre-plan re-baseline probe in
/// `push_branch`) and consumed by [`advance_packmap`]'s first CAS attempt —
/// so a healthy chain is walked once per push, not twice (mkit #521 perf
/// fix; previously the pre-plan depth probe and `advance_packmap`'s own
/// [`resolve_pack_chain`] call each walked the whole chain independently).
/// `advance_packmap` only reuses this when the packmap's live value still
/// equals `head`; a mismatch (the packmap moved between the probe and the
/// CAS attempt, or a prior CAS attempt already lost the race) falls back to
/// a fresh [`resolve_pack_chain`] call, preserving the existing retry
/// semantics.
#[derive(Debug)]
pub(crate) struct ResolvedChain {
    /// The packmap value this chain was walked from.
    pub(crate) head: Hash,
    /// Chain depth (node count) at `head`.
    pub(crate) depth: usize,
    /// Flattened oldest-first pack keys — identical to what
    /// [`resolve_pack_chain`] would return for `head`.
    pub(crate) packs: Vec<Hash>,
}

/// Walk a branch's packlist chain from `head_key` ONCE, returning both its
/// depth and its flattened oldest-first pack keys as a single
/// [`ResolvedChain`]. Sole caller is `push_branch`'s pre-plan re-baseline
/// probe (#406); see [`ResolvedChain`] for why threading its result into
/// [`advance_packmap`] matters.
pub(crate) fn probe_chain(
    tx: &dyn Transport,
    branch: &str,
    head_key: Hash,
) -> Result<ResolvedChain, DispatchError> {
    let mut nodes = walk_pack_chain(tx, branch, head_key)?;
    let depth = nodes.len();
    nodes.reverse(); // oldest node first
    let packs = nodes.into_iter().flat_map(|n| n.packs).collect();
    Ok(ResolvedChain {
        head: head_key,
        depth,
        packs,
    })
}

/// The push-side decision for how a new pack extends (or resets) the
/// packmap chain, computed once in `push_branch` and threaded into
/// [`advance_packmap`].
///
/// Replaces a former `self_contained: bool, rebaseline: bool` parameter
/// pair, which let "`rebaseline` == true but `self_contained` == false" —
/// an invariant violation `advance_packmap` had to police with a
/// `debug_assert` PLUS a defensive runtime error
/// (`DispatchError::RebaselineNotSelfContained`) — be constructed at all.
/// [`Self::ResetSelfContained`] carries no `self_contained` field, so that
/// illegal combination now has no representation (mkit #521).
#[derive(Debug, Clone, Copy)]
pub(crate) enum ChainAction {
    /// Validate the prior chain (if any) and append onto it. If the prior
    /// chain is broken, a self-contained pack (`self_contained == true`)
    /// resets to escape it; otherwise the push blocks
    /// ([`DispatchError::PackChainInvalid`]).
    Append {
        /// Whether the pack being chained on reconstructs the whole
        /// closure with no external base — the precondition for the
        /// broken-chain escape-hatch reset described above.
        self_contained: bool,
    },
    /// Proactive re-baseline (#406): unconditionally reset the chain to a
    /// single fresh self-contained node, skipping prior-chain validation
    /// entirely. `push_branch` only ever picks this alongside a
    /// full-closure pack plan (always self-contained), and only when the
    /// transport reports [`Transport::supports_atomic_advance`] AND the head
    /// write is CAS-conditioned (mkit #521) — see [`advance_packmap`]'s doc
    /// comment for why that gate matters.
    ///
    /// # Fetch-side cost (mkit #521)
    ///
    /// A reset replaces the whole chain with one new full-closure pack whose
    /// digest no existing fetcher has in its applied-pack record. So every
    /// incremental fetcher must re-download AND re-unpack the ENTIRE branch
    /// closure once per re-baseline cycle (~every [`DEFAULT_REBASELINE_DEPTH`]
    /// pushes), silently defeating #520's applied-packs skip optimization for
    /// that fetch. Worse, the applied-packs record then accumulates the
    /// orphaned pre-reset digests (one full cycle's worth per reset) with no
    /// pruning — an unbounded, full-file-rewrite-per-fetch growth. (The
    /// remote-side orphaning is documented in `transfer.rs` / makechain#849;
    /// this note covers the fetcher impact, which nothing else did.)
    ResetSelfContained,
}

/// Chain `pack_keys` (in apply order — a push over the payload cap
/// produces more than one, see [`super::build_and_upload_packs`]) onto
/// the branch's packmap, recorded as a single new node, and CAS-advance
/// the `refs/mkit/packmap/<branch>` pointer to it.
///
/// This MUST succeed before the branch ref is moved: the invariant is
/// "if `refs/heads/<branch>` resolves to T, the packmap reconstructs
/// closure(T)". We can't update both refs in one transaction against the
/// CAS-only ref API, so we order them — packmap first, gated — and on a
/// concurrent conflict we re-read the (only-growing) packmap and retry.
/// If we can't land it within [`PACKMAP_CAS_ATTEMPTS`], the caller aborts
/// the push without moving the head, so the head never advances past a
/// packmap that fails to reconstruct it.
///
/// The **entire** prior chain is validated (not just its head node) before
/// we build on it, so a deeper missing/corrupt node can't slip a head onto a
/// chain that fetch cannot walk:
///
/// * No prior packmap → start a fresh chain (`prev = None`).
/// * Prior chain fully walks → append (`prev = prior`). If every key in
///   our set is already somewhere in the chain the push is idempotent
///   and we stop.
/// * Prior chain broken at any depth → we must not append a chain whose tail
///   can't be resolved. If `self_contained` (this pack reconstructs the whole
///   closure with no external base) we **reset** to a fresh chain — the only
///   safe way to escape a broken chain. Otherwise we **block** the push
///   ([`DispatchError::PackChainInvalid`]) before the head moves, because the
///   deltas' bases live in the unreachable prior chain.
///
/// `ChainAction::ResetSelfContained` (#406) is the *proactive* counterpart
/// to the broken-chain reset above: the caller (`push_branch`) picks it
/// when the prior chain is perfectly healthy but has simply grown past the
/// re-baseline threshold (see [`rebaseline_depth`]) — AND, since mkit
/// #521, only when the transport's [`Transport::advance_refs`] is
/// transactional (see [`Transport::supports_atomic_advance`]); a reset is
/// not a superset of the prior chain, so committing one while losing the
/// paired head CAS is safe only when both writes land as one transaction.
/// When picked, prior-chain validation is skipped entirely and `prev` is
/// unconditionally `None` — a re-baseline always resets, whether or not
/// the prior chain would otherwise resolve. `push_branch` only ever picks
/// this action alongside a pack planned with an empty `have_old` set
/// (always self-contained), so — unlike the old `self_contained: bool,
/// rebaseline: bool` pair this enum replaces — "a reset of a
/// non-self-contained pack" has no representation to defensively guard
/// against: [`ChainAction::ResetSelfContained`] carries no
/// `self_contained` field at all.
///
/// `resolved` (#521 perf fix) is an optional chain walk the caller already
/// performed (`push_branch`'s pre-plan re-baseline probe, see
/// [`probe_chain`]) for the SAME packmap value this loop's first iteration
/// will read. When it's still fresh (the packmap hasn't moved since), the
/// first iteration reuses it instead of re-walking the chain via
/// [`resolve_pack_chain`]; a lost CAS race (or a stale/absent `resolved`)
/// falls back to a fresh walk on retry, exactly as before this
/// optimization.
pub(crate) fn advance_packmap(
    tx: &dyn Transport,
    branch: &str,
    pack_keys: &[Hash],
    action: ChainAction,
    resolved: Option<ResolvedChain>,
    head_condition: refs::RefWriteCondition,
    tip: Hash,
) -> Result<(), DispatchError> {
    debug_assert!(
        !pack_keys.is_empty(),
        "advance_packmap requires at least one pack key — callers only reach this with a \
         non-empty plan"
    );
    let packmap_name = packmap_ref(branch);
    let head_name = format!("refs/heads/{branch}");
    // Consumed by (at most) the first iteration that reaches the "resolve
    // the prior chain" branch below; a retry (another loop pass) always
    // finds this already `None` and re-walks fresh, per the doc comment.
    let mut cached = resolved;
    // `action` (Append vs. Reset) is FROZEN by `push_branch` before this CAS
    // loop and never re-evaluated across retries — the depth probe ran once,
    // pre-plan. So under contention just below the re-baseline threshold, N
    // racers can each independently decide to Append (none saw the others'
    // node yet) and the chain overshoots the bound by ~N nodes. That is
    // bounded (by the number of concurrent pushers) and self-correcting (the
    // next push over the now-higher depth re-baselines), so we accept it
    // rather than re-probe depth inside the loop.
    for _ in 0..PACKMAP_CAS_ATTEMPTS {
        if crate::signal::is_shutdown() {
            return Err(DispatchError::Interrupted);
        }
        let prior = tx.read_ref(&packmap_name)?;
        // Decide the new node's `prev`. A re-baseline always resets
        // (skipping prior-chain validation); otherwise validate the WHOLE
        // prior chain before deciding to append vs. reset.
        let prev = match action {
            ChainAction::ResetSelfContained => {
                // Invariant guard (mkit #521): `push_branch` only ever picks a
                // reset on a transport whose `advance_refs` is transactional
                // AND whose head write is CAS-conditioned (not `Any`). A reset
                // is not a superset of the prior chain, so committing the
                // packmap while a paired ORDERED head PUT is lost/crashes
                // would strand the head at a closure the reset can't rebuild.
                // Assert both here so a FUTURE direct caller fails loudly
                // instead of silently reintroducing that stranded-head bug.
                debug_assert!(
                    tx.supports_atomic_advance(),
                    "re-baseline reset requires a transactional advance_refs (mkit #521)"
                );
                debug_assert!(
                    !matches!(head_condition, refs::RefWriteCondition::Any),
                    "re-baseline reset must not run with an `Any` head condition — the \
                     ordered advance_refs fallback would strand the head (mkit #521)"
                );
                None
            }
            ChainAction::Append { self_contained } => match prior {
                None => None,
                Some(p) => {
                    // Reuse the pre-plan walk if it's still for the packmap
                    // value we just read; otherwise walk it fresh. Either
                    // way this converges on the same `Result` shape
                    // `resolve_pack_chain` alone used to produce here.
                    let packs = match cached.take() {
                        Some(c) if c.head == p => Ok(c.packs),
                        _ => resolve_pack_chain(tx, branch, p),
                    };
                    match packs {
                        // Idempotency: a previous attempt already advertised
                        // EVERY key in our set (any position — the node
                        // that carried them was written atomically before
                        // its ref CAS, so a committed prior attempt from
                        // THIS push contributed all keys at once, already
                        // in apply order; determinism — same plan, same
                        // greedy split, same pack bytes — means a retry
                        // regenerates the identical key set). The packmap
                        // is already correct, so only the head still needs
                        // to move — commit it alone. Partial overlap can
                        // only arise from a different push that happened
                        // to produce byte-identical packs for a subset;
                        // appending a fresh node in that case is harmless
                        // (fetch applies content-addressed objects
                        // idempotently, and the applied-packs record skips
                        // already-applied digests).
                        Ok(packs) => {
                            let have: std::collections::HashSet<&Hash> = packs.iter().collect();
                            if pack_keys.iter().all(|k| have.contains(k)) {
                                return commit_head(tx, &head_name, head_condition, &tip, branch);
                            }
                            Some(p) // healthy chain — append onto it
                        }
                        // Broken chain: a self-contained pack can reset to escape it;
                        // a delta push must block (its bases live in the broken tail).
                        // Transient transport errors propagate (don't reset on a blip).
                        Err(DispatchError::PackChainInvalid { .. }) if self_contained => None,
                        Err(e) => return Err(e),
                    }
                }
            },
        };
        let node = transfer::encode_packlist(prev, pack_keys)?;
        let node_key = pack::pack_key(&node);
        tx.upload_blob_via_ref(&node, &PackKey::from_hash(node_key), &head_name)
            .map_err(|error| super::repository_operation_error(tx, error))?;
        // CAS off the packmap's CURRENT value (`prior`), independent of the
        // node's `prev` — a reset still has to win the race for the ref.
        let packmap_condition = match prior {
            Some(k) => refs::RefWriteCondition::Match(k),
            None => refs::RefWriteCondition::Missing,
        };
        let commit_keys: Vec<PackKey> = pack_keys
            .iter()
            .copied()
            .map(PackKey::from_hash)
            .chain(std::iter::once(PackKey::from_hash(node_key)))
            .collect();
        // Commit the packmap AND the head together (#408). A transactional
        // transport applies both atomically; the default does packmap-then-
        // head — still safe, the head never lands past an unadvanced packmap.
        match tx
            .advance_refs_committing(
                &head_name,
                head_condition,
                &tip,
                &packmap_name,
                packmap_condition,
                &node_key,
                &commit_keys,
            )
            .map_err(|error| super::repository_operation_error(tx, error))?
        {
            CommitOutcome::Advanced(AdvanceOutcome::Committed) => return Ok(()),
            // Another pusher advanced the packmap under us — re-read and retry.
            CommitOutcome::Advanced(AdvanceOutcome::PackmapConflict) => {}
            // The head precondition failed. Either the branch moved under
            // us (the push is stale) or the retry ladder re-issued a write
            // that had already landed (SPEC-TRANSPORT §7) — disambiguate.
            CommitOutcome::Advanced(AdvanceOutcome::HeadConflict) => {
                return head_conflict(tx, &head_name, &tip, branch);
            }
            CommitOutcome::TicketRejected => {
                if tx.read_ref(&head_name)? == Some(tip) {
                    return Ok(());
                }
                return Err(DispatchError::TicketRejected);
            }
            CommitOutcome::PacklistNotInRepository => {
                if tx.read_ref(&head_name)? == Some(tip) {
                    return Ok(());
                }
                return Err(DispatchError::PacklistNotInRepository);
            }
            CommitOutcome::DeltaBaseUnavailable => return Err(DispatchError::DeltaBaseUnavailable),
            _ => return Err(DispatchError::Transport(TransportError::InvalidResponse)),
        }
    }
    Err(DispatchError::PackmapContended {
        branch: branch.to_owned(),
    })
}

/// Move just the branch head under its CAS condition, mapping a conflict to
/// the actionable non-fast-forward error. Used when the packmap already
/// advertises the pack (idempotent retry) or a push has nothing to send.
pub(crate) fn commit_head(
    tx: &dyn Transport,
    head_name: &str,
    condition: refs::RefWriteCondition,
    tip: &Hash,
    branch: &str,
) -> Result<(), DispatchError> {
    match tx.update_ref(head_name, condition, tip) {
        Ok(()) => Ok(()),
        Err(TransportError::RefConflict) => head_conflict(tx, head_name, tip, branch),
        Err(e) => Err(super::repository_operation_error(tx, e)),
    }
}

/// Resolve a failed head precondition (`RefConflict` from `update_ref`,
/// `HeadConflict` from `advance_refs`) per SPEC-TRANSPORT §7: the retry
/// ladder may have re-issued a write whose first attempt landed, so the
/// conflict can be our own write. Re-read the head; if it already holds
/// `tip` the push landed, otherwise it is a genuine non-fast-forward.
///
/// Only the head is checked. Whoever moved the head to `tip` also advanced
/// the packmap to reconstruct it first (packmap-then-head, or both at once),
/// so a head at `tip` already satisfies the push's invariant even if a later
/// pusher has since moved the packmap past our node.
fn head_conflict(
    tx: &dyn Transport,
    head_name: &str,
    tip: &Hash,
    branch: &str,
) -> Result<(), DispatchError> {
    let current = tx
        .read_ref(head_name)
        .map_err(|error| super::repository_operation_error(tx, error))?;
    if current == Some(*tip) {
        return Ok(());
    }
    Err(DispatchError::NonFastForwardPush {
        branch: branch.to_owned(),
    })
}

/// Walk a branch's packlist chain from `head_key` and unpack every pack in
/// dependency order (oldest node first), so each delta's base is already in
/// the store when its pack is unpacked. Chain resolution + integrity is
/// shared with the push side via [`resolve_pack_chain`].
///
/// Within a single packmap chain a delta's base always arrives in an
/// earlier pack (the push planner only deltas against bases the remote — and
/// therefore the chain — already holds), so unpacking oldest-first is
/// sufficient: there is no per-object base prefetch.
///
/// Packs already recorded in `applied` (the caller-owned, in-memory
/// applied-pack record for `remote`; see [`applied_packs`]) are skipped —
/// neither downloaded nor unpacked — so a steady-state fetch only pays for
/// packs new since the last fetch (#409). The chain itself is still
/// resolved in full every time: node downloads are small blobs and remain
/// the source of truth for chain shape, independent of what's locally
/// applied.
///
/// This function never loads or persists `applied` itself — it only mutates
/// the in-memory set; see [`super::fetch_objects`] for the load-once /
/// persist-once contract it runs under.
///
/// # Self-heal
///
/// A run first downloads/unpacks every non-skipped pack, then asserts
/// `tip`'s closure is fully present via [`super::verify_closure_present`]
/// (folded in here rather than sequenced by the caller — see that function's
/// doc comment for why). Exactly ONE failure mode is treated as local
/// staleness: the closure check reporting [`DispatchError::RemoteMissingObject`]
/// on a run that skipped at least one recorded pack. That means the record
/// claimed packs were applied but their objects aren't in the store (e.g.
/// `.mkit/objects` was wiped out-of-band while `applied-packs/` survived), so
/// `applied` is cleared in memory ([`AppliedPacks::clear`]) and the whole
/// chain is retried once with no skips; the caller's single end-of-fetch
/// persist durably reflects this post-heal state.
///
/// Every other failure propagates without triggering self-heal, because none
/// is evidence the local object store is stale:
///
/// * A download/unpack failure — a freshly-downloaded corrupt pack
///   ([`DispatchError::Pack`] from [`PackReader::read`]), a genuinely
///   incomplete remote ([`DispatchError::AdvertisedPackMissing`]), or a
///   transient [`DispatchError::Transport`] error — is a remote-side or
///   network problem. Wiping the cache would destroy a valid record on a
///   corrupt-remote's behalf, so these propagate as-is (a corrupt pack must
///   surface, not be papered over by a full re-download).
/// * A [`DispatchError::ClosureTooLarge`] means the closure exceeded the
///   verification cap — a scale limit, not staleness.
/// * A [`DispatchError::Interrupted`] (user-requested shutdown) is never a
///   self-heal trigger.
/// One branch's pack chain, resolved and downloaded over the network with
/// **no repo lock held** — see [`resolve_and_download_chain`]. Consumed by
/// [`apply_fetched_chain`], which the caller must run under the repo lock.
pub(crate) struct FetchedChain {
    /// Oldest-first flattened chain, as returned by [`resolve_pack_chain`].
    /// Retained (not just `downloaded`) so a self-heal retry inside
    /// [`apply_fetched_chain`] can re-download without re-walking the
    /// packlist chain.
    chain: Vec<Hash>,
    /// Private, verified files for missing packs in apply order. Payload
    /// bytes are never retained across download iterations.
    downloaded: StagedPacks,
}

/// Phase 1 of a branch fetch (#642): walk `branch`'s packmap chain from
/// `head_key` and download every pack in it not already recorded in
/// `applied`. Pure network I/O — resolving the chain shape reads small
/// auxiliary blobs and downloading packs never touches the local object
/// store — so the caller does **not** need the repo lock for this call.
/// The repo lock is only required for [`apply_fetched_chain`], which
/// unpacks the result.
///
/// # Errors
/// Propagates chain-walk failures ([`DispatchError::PackChainInvalid`],
/// [`DispatchError::Interrupted`]) and download failures
/// ([`DispatchError::AdvertisedPackMissing`], transport errors) unchanged.
pub(crate) fn resolve_and_download_chain(
    tx: &dyn Transport,
    branch: &str,
    head_key: Hash,
    applied: &AppliedPacks,
) -> Result<FetchedChain, DispatchError> {
    // Chain shape is always resolved fresh and in full — see the doc
    // comment on `fetch_pack_chain`'s prior single-function form. Only the
    // per-pack download loop below consults the applied-pack record (it
    // does not mutate it; mutation happens once the pack is actually
    // unpacked, in `unpack_downloaded_packs`).
    let chain = resolve_pack_chain(tx, branch, head_key)?;
    let downloaded = download_pack_chain(tx, branch, &chain, applied)?;
    Ok(FetchedChain { chain, downloaded })
}

/// Phase 2 of a branch fetch (#642): unpack `fetched`'s downloaded packs
/// into `store` and verify the closure at `tip` is complete, running the
/// applied-pack self-heal retry on a stale-record failure.
///
/// **The caller MUST hold the repo lock across this call, through the
/// branch's subsequent ref publish** — see [`super::fetch_objects_inner`]
/// and the module docs on `#642`. Unpacking writes objects to the local
/// store before this branch's ref makes them reachable; a concurrent `gc`
/// must not be able to observe that store state, so the lock has to stay
/// held continuously from the first write here through the ref write in
/// the caller. Self-heal (rare: only on a stale `applied` record) is the
/// one path that still does network I/O under that lock — an accepted
/// trade, since it is a recovery path, not the routine one.
///
/// # Signature verification (issue #692)
///
/// Every commit/remix/tag in each freshly unpacked pack is run through
/// [`verify_new_object_signatures`] before recording that pack as applied
/// when `require_signed` is `true` (the CLI's default; `false` is the
/// explicit `--no-verify-signatures` / `pull.require_signed = false`
/// opt-out). A failure is [`DispatchError::UnsignedOrInvalidObject`],
/// which — like [`DispatchError::ClosureTooLarge`] — is deliberately NOT
/// a self-heal trigger: an invalid signature is not evidence of local
/// staleness. On the self-heal path every re-downloaded object (the whole
/// chain, not just the delta) is re-verified, since self-heal only runs
/// when the local store's contents are already suspect.
///
/// # Errors
/// [`DispatchError::RemoteMissingObject`] if the closure is still
/// incomplete after self-heal (or immediately, when self-heal doesn't
/// apply); pack-decode / store errors from the unpack; download errors
/// from the self-heal retry; [`DispatchError::UnsignedOrInvalidObject`]
/// if `require_signed` is `true` and a newly-fetched object's signature
/// does not verify.
pub(crate) fn apply_fetched_chain(
    store: &ObjectStore,
    tx: &dyn Transport,
    remote: &str,
    branch: &str,
    fetched: FetchedChain,
    tip: Hash,
    applied: &mut AppliedPacks,
    require_signed: bool,
) -> Result<(), DispatchError> {
    let FetchedChain { chain, downloaded } = fetched;
    let skipped = chain.len() - downloaded.packs.len();
    unpack_downloaded_packs(store, downloaded, applied, require_signed)?;

    // Closure completeness. With skips this is the sole guarantee the
    // store is whole, and a `RemoteMissingObject` here is the ONLY
    // self-heal trigger.
    match super::verify_closure_present(store, &tip) {
        Ok(()) => Ok(()),
        Err(e @ DispatchError::RemoteMissingObject(_)) if skipped > 0 => {
            eprintln!(
                "note: applied-packs record for remote '{remote}' branch '{branch}' looks stale ({e}); clearing it and re-fetching the full pack chain"
            );
            // Clear the suspected-stale record in memory and retry the whole
            // chain with no skips. This is infallible — the caller's single
            // end-of-fetch persist durably reflects the post-heal state.
            // Clearing intentionally discards digests inserted by other
            // branches earlier in this same fetch: the store wipe that trips
            // self-heal makes those entries just as stale.
            applied.clear();
            let downloaded = download_pack_chain(tx, branch, &chain, applied)?;
            unpack_downloaded_packs(store, downloaded, applied, require_signed)?;
            super::verify_closure_present(store, &tip)
        }
        Err(e) => Err(e),
    }
}

/// Entries-per-thread budget below which [`verify_new_object_signatures`]
/// checks signatures sequentially instead of fanning out across rayon's
/// thread pool, for a pool of a given size. Same crossover shape as
/// `remote_dispatch/mod.rs`'s `PACK_FANOUT_ENTRIES_PER_THREAD` (rayon's
/// pool-dispatch overhead loses to a plain loop below a few entries per
/// thread and wins clearly above it) but tuned lower: each entry here is
/// only a `store.read_object` (disk read + BLAKE3 re-hash) plus one
/// Ed25519 `verify_strict` — cheaper per item than the zstd-compression
/// fan-out `PACK_FANOUT_ENTRIES_PER_THREAD` guards, so fewer entries per
/// thread are enough to amortize dispatch. Measured with `cargo bench -p
/// mkit-benches --bench verify_fanout`.
const VERIFY_FANOUT_ENTRIES_PER_THREAD: usize = 2;

/// The total entry count [`verify_new_object_signatures`] fans out across
/// rayon at — see [`VERIFY_FANOUT_ENTRIES_PER_THREAD`].
fn verify_fanout_threshold() -> usize {
    crate::fanout::threshold(VERIFY_FANOUT_ENTRIES_PER_THREAD)
}

/// Chunk size [`verify_new_object_signatures`]'s parallel path processes
/// `stored` in — deliberately decoupled from [`verify_fanout_threshold`]
/// (a small, pool-size-scaled crossover picked to amortize rayon's
/// per-dispatch overhead) rather than reusing it as the chunk size too.
/// The two constants answer different questions: `verify_fanout_threshold`
/// asks "is this worth parallelizing at all?"; this asks "how much wasted
/// verification work should a hostile fetch's first bad signature be
/// allowed to force?" A large legitimate fetch (thousands of newly-signed
/// objects) pays one rayon dispatch per chunk, so reusing the tiny
/// crossover threshold as the chunk size (as few as `2 * num_threads`,
/// e.g. 16 on an 8-core host) meant a 10,000-object fetch paid roughly
/// 625 separate dispatches — thread-pool coordination overhead with no
/// benefit, since the hostile-input bound doesn't need a chunk anywhere
/// near that small. 512 keeps the bound meaningful (a hostile remote can
/// force at most 512 extra reads/Ed25519-verifies past the object that
/// actually fails — a small, fixed amount of wasted CPU regardless of
/// fetch size) while cutting a 10,000-object fetch to ~20 dispatches.
const VERIFY_CHUNK_CAP: usize = 512;

/// The chunk size [`verify_new_object_signatures`]'s parallel path uses —
/// see [`VERIFY_CHUNK_CAP`]. `.max(verify_fanout_threshold())` guards the
/// (currently unreachable on any real host) case of a thread pool large
/// enough that the crossover threshold itself would exceed the cap —
/// a chunk should never be smaller than the count that justified
/// parallelizing it in the first place.
fn verify_chunk_size() -> usize {
    VERIFY_CHUNK_CAP.max(verify_fanout_threshold())
}

/// Verify the Ed25519 signature on every commit/remix/tag in `stored` —
/// the digests [`unpack_downloaded_packs`] just wrote, i.e. the objects
/// this fetch actually introduced (issue #692). Uses the exact same check
/// `mkit verify <rev>` runs manually
/// ([`mkit_core::sign::verify_commit`]/`verify_remix`/`verify_tag`), so
/// clone/pull/fetch cannot publish a remote-tracking ref to a hostile
/// remote's unsigned or forged history (THREAT-MODEL §3.1) without the
/// caller explicitly opting out. Blob/Tree/ChunkedBlob/Delta objects carry
/// no signature and are skipped.
///
/// `require_signed = false` (the explicit opt-out) short-circuits to
/// `Ok(())` without reading any object — a no-op, not a "verify but
/// ignore the result".
///
/// Each entry's read + verify is independent of every other's, so at or
/// above [`verify_fanout_threshold`] entries this fans out across rayon's
/// global thread pool (mirroring `add.rs`'s hash fan-out and
/// `remote_dispatch/mod.rs`'s pack-build fan-out) instead of running one
/// entry at a time — `store.read_object`'s BLAKE3 re-hash and
/// `verify_commit`/`verify_remix`/`verify_tag`'s Ed25519 check are both
/// CPU-bound, so a large fetch's verification pass parallelizes cleanly
/// across cores.
///
/// The parallel path processes `stored` in fixed-size chunks of
/// [`verify_chunk_size`] entries, verifying each chunk in full before
/// starting the next, rather than fanning the whole slice out in one
/// `par_iter` — rayon's `try_for_each` only best-effort short-circuits
/// (already-dispatched work keeps running once an error is found), so a
/// single flat fan-out over a very large hostile fetch could still force
/// reading and Ed25519-verifying a large fraction of the batch past the
/// first invalid signature before the rejection propagates. Chunking
/// bounds that wasted work to at most one chunk: a hostile remote can
/// force at most `verify_chunk_size()` extra reads/verifies beyond the
/// object that actually fails, never the rest of `stored`. Which entry's
/// error surfaces first is not guaranteed to match `stored`'s order
/// *within* a chunk, but the chunk containing the first invalid entry
/// (in `stored`'s order) is always the one whose error is returned,
/// since later chunks are never started.
///
/// Within each chunk (or the whole slice, below the fan-out threshold),
/// [`verify_slice`] tries a single [`mkit_core::sign::verify_batch`] pass
/// over every commit/remix/tag it read before falling back to
/// [`verify_one_object`]'s per-object loop — see [`verify_slice`]'s docs.
fn verify_new_object_signatures(
    store: &ObjectStore,
    stored: &[Hash],
    require_signed: bool,
) -> Result<(), DispatchError> {
    if !require_signed {
        return Ok(());
    }
    if stored.len() < verify_fanout_threshold() {
        return verify_slice(store, stored, false);
    }
    stored
        .chunks(verify_chunk_size())
        .try_for_each(|chunk| verify_slice(store, chunk, true))
}

/// Verify the signature on a single already-read object through
/// [`sign::verify_object_signature`] (commit/remix/tag; Blob/Tree/
/// ChunkedBlob/Delta carry no signature and pass). The exact per-object
/// check [`verify_new_object_signatures`] used unconditionally before
/// batch verification was added, now used both as [`verify_slice`]'s
/// fallback and — via that fallback — the sole check whenever the batch
/// fast path doesn't apply or doesn't succeed.
fn verify_one_object(h: Hash, obj: &Object) -> Result<(), DispatchError> {
    sign::verify_object_signature(obj).map_err(|e| DispatchError::UnsignedOrInvalidObject {
        hash: hash::to_hex(&h),
        reason: e.to_string(),
    })
}

/// Collect `(public_key, digest, signature)` triples for every signed
/// object in `entries`, for a single [`mkit_core::sign::verify_batch`]
/// call — `None` if computing any entry's signing digest itself failed,
/// which [`verify_slice`] treats as "skip the batch attempt", since
/// [`verify_one_object`] will independently hit and correctly attribute
/// that same failure in its fallback loop.
fn collect_batch_entries(
    entries: &[(Hash, Object)],
) -> Option<Vec<(sign::PublicKey, Hash, sign::Signature)>> {
    let mut batch = Vec::new();
    for (_, obj) in entries {
        match obj {
            Object::Commit(c) => batch.push((
                sign::PublicKey(c.signer),
                sign::commit_signing_hash(c).ok()?,
                sign::Signature(c.signature),
            )),
            Object::Remix(r) => batch.push((
                sign::PublicKey(r.signer),
                sign::remix_signing_hash(r).ok()?,
                sign::Signature(r.signature),
            )),
            Object::Tag(t) => batch.push((
                sign::PublicKey(t.signer),
                sign::tag_signing_hash(t).ok()?,
                sign::Signature(t.signature),
            )),
            Object::Blob(_) | Object::Tree(_) | Object::ChunkedBlob(_) | Object::Delta(_) => {}
        }
    }
    Some(batch)
}

/// Splits `batch` into one sub-batch per rayon worker thread and
/// verifies the sub-batches in parallel, instead of one whole-slice
/// [`sign::verify_batch`] call on a single thread.
///
/// `cargo bench -p mkit-benches --bench verify_fanout` on a 4-core host
/// found that a single-threaded batch call, despite doing less total
/// scalar-multiplication work than one `verify_strict` per entry, loses
/// to today's per-object rayon fan-out once the fan-out has enough
/// entries to keep every core busy (256 entries: 9.6ms rayon vs. 11.9ms
/// one whole-slice batch) — batching's ~2x reduction in total work
/// doesn't make up for using only one of four cores. Chunking the batch
/// itself across rayon combines both effects instead of trading one for
/// the other: same bench, same 256 entries, 7.4ms — faster than either
/// alone, and consistently faster than plain rayon fan-out from 8
/// entries up (verified at 8/16/32/64/128/256).
fn verify_batch_parallel(batch: &[(sign::PublicKey, Hash, sign::Signature)]) -> bool {
    let threads = rayon::current_num_threads().max(1);
    let chunk_size = batch.len().div_ceil(threads).max(1);
    batch
        .par_chunks(chunk_size)
        .all(|sub| sign::verify_batch(sub).is_ok())
}

/// Read every object in `hashes` (in parallel iff `parallel`, matching
/// [`verify_new_object_signatures`]'s existing sequential/rayon split),
/// then verify their signatures. Every hash is read (there's no cheaper
/// way to learn an object's type), but only Commit/Remix/Tag entries are
/// retained afterward — see the read step's own comment for why that
/// matters for a chunk dominated by large blobs.
///
/// Tries [`mkit_core::sign::verify_batch`] first over every commit/
/// remix/tag found — via [`verify_batch_parallel`] when `parallel` is
/// set, a single whole-slice call otherwise (see that function's docs
/// for why the two cases need different strategies) — which does
/// strictly less total Ed25519 work than verifying each individually
/// and is the overwhelmingly common case: every object a well-behaved
/// remote sends is validly signed. Batch verification only proves "all
/// valid" or "at least one isn't", never *which* one, so on failure (or
/// if any entry's signing digest itself couldn't be computed) this
/// falls back to [`verify_one_object`]'s per-object loop — identical to
/// [`verify_new_object_signatures`]'s behavior before batch verification
/// was added — to locate and report the exact offending hash. A hostile
/// remote therefore pays for both the batch attempt and the fallback
/// loop on its one bad chunk, same bound [`verify_new_object_signatures`]
/// already documents (at most [`verify_chunk_size`] extra reads/verifies
/// past the object that actually fails); a well-behaved remote pays for
/// the batch attempt only.
fn verify_slice(store: &ObjectStore, hashes: &[Hash], parallel: bool) -> Result<(), DispatchError> {
    // Every hash in `hashes` still has to be *read* to learn its type —
    // the pack format doesn't index that separately — but only
    // Commit/Remix/Tag objects are ever inspected below, by both
    // `collect_batch_entries` and `verify_one_object`. Dropping a
    // Blob/Tree/ChunkedBlob/Delta right after the type check, instead of
    // keeping it in `entries` for the rest of this call, matters
    // concretely: a single `Blob` can be up to `worktree::CHUNK_THRESHOLD`
    // (1 MiB — larger files are chunked, per this crate's whole design),
    // so a `verify_chunk_size()`-sized chunk (up to 512 entries)
    // dominated by such blobs previously held up to ~512 MiB of fully
    // decoded, never-inspected object bytes resident at once. This way,
    // at most one blob per rayon worker (or one at all, sequentially) is
    // ever live simultaneously.
    let read_signed_one = |h: &Hash| -> Result<Option<(Hash, Object)>, DispatchError> {
        let obj = store.read_object(h)?;
        Ok(match obj {
            Object::Commit(_) | Object::Remix(_) | Object::Tag(_) => Some((*h, obj)),
            Object::Blob(_) | Object::Tree(_) | Object::ChunkedBlob(_) | Object::Delta(_) => None,
        })
    };
    let entries: Vec<(Hash, Object)> = if parallel {
        hashes
            .par_iter()
            .filter_map(|h| read_signed_one(h).transpose())
            .collect::<Result<_, _>>()?
    } else {
        hashes
            .iter()
            .filter_map(|h| read_signed_one(h).transpose())
            .collect::<Result<_, _>>()?
    };

    if let Some(batch) = collect_batch_entries(&entries) {
        let batch_ok = batch.is_empty()
            || if parallel {
                verify_batch_parallel(&batch)
            } else {
                sign::verify_batch(&batch).is_ok()
            };
        if batch_ok {
            return Ok(());
        }
    }

    if parallel {
        entries
            .par_iter()
            .try_for_each(|(h, obj)| verify_one_object(*h, obj))
    } else {
        entries
            .iter()
            .try_for_each(|(h, obj)| verify_one_object(*h, obj))
    }
}

/// Owns all staging paths; success, cancellation and every error remove the
/// private directory. Files are closed between operations, so a large history
/// does not also require one live file descriptor per pack.
struct StagedPacks {
    directory: tempfile::TempDir,
    packs: Vec<StagedPack>,
    bytes: u64,
}

struct StagedPack {
    key: PackKey,
    bytes: usize,
}

/// Download to private files without holding the repo lock. Only one returned
/// pack buffer exists at a time; chain metadata and staging disk have explicit
/// aggregate limits. No authoritative objects or applied records are changed.
fn download_pack_chain(
    tx: &dyn Transport,
    branch: &str,
    chain: &[Hash],
    applied: &AppliedPacks,
) -> Result<StagedPacks, DispatchError> {
    download_pack_chain_with_limits(tx, branch, chain, applied, MAX_FETCH_STAGED_BYTES, None)
}

fn download_pack_chain_with_limits(
    tx: &dyn Transport,
    branch: &str,
    chain: &[Hash],
    applied: &AppliedPacks,
    disk_limit: u64,
    stage_parent: Option<&std::path::Path>,
) -> Result<StagedPacks, DispatchError> {
    if chain.len() > MAX_FETCH_PACKS {
        return Err(DispatchError::PackChainInvalid {
            branch: branch.to_owned(),
        });
    }
    let directory = match stage_parent {
        Some(parent) => tempfile::Builder::new()
            .prefix("mkit-fetch-")
            .tempdir_in(parent)?,
        None => tempfile::Builder::new().prefix("mkit-fetch-").tempdir()?,
    };
    let mut staged = StagedPacks {
        directory,
        packs: Vec::new(),
        bytes: 0,
    };
    let ref_name = branch_ref_hint(branch);
    for &pk in chain {
        if crate::signal::is_shutdown() {
            return Err(DispatchError::Interrupted);
        }
        let key = PackKey::from_hash(pk);
        if applied.contains(&key) {
            continue;
        }
        let pack = match tx.download_pack_via_ref(&key, &ref_name) {
            Ok(b) => b,
            Err(TransportError::PackNotFound) => {
                return Err(DispatchError::AdvertisedPackMissing {
                    branch: branch.to_owned(),
                    pack: mkit_core::hash::to_hex(&pk),
                });
            }
            Err(e) => return Err(e.into()),
        };
        #[cfg(test)]
        observe_retained_pack_bytes(pack.capacity());
        if pack.len() as u64 > mkit_core::protocol::PACK_BODY_LIMIT {
            return Err(TransportError::PayloadTooLarge(pack.len()).into());
        }
        key.verify_bytes(&pack)?;
        let next_bytes = staged
            .bytes
            .checked_add(pack.len() as u64)
            .filter(|&bytes| bytes <= disk_limit)
            .ok_or_else(|| {
                std::io::Error::new(
                    std::io::ErrorKind::StorageFull,
                    format!("fetch staging exceeds the {disk_limit}-byte disk budget"),
                )
            })?;
        let path = staged.directory.path().join(staged.packs.len().to_string());
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(path)?;
        file.write_all(&pack)?;
        staged.packs.push(StagedPack {
            key,
            bytes: pack.len(),
        });
        staged.bytes = next_bytes;
        // `pack` and the file handle drop before the next download. No fsync
        // is needed for disposable staging: a crash never publishes a ref.
    }
    Ok(staged)
}

#[cfg(test)]
std::thread_local! {
    // Tracks owned download payloads at the retention boundary. Thread-local
    // state keeps concurrently running unit tests independent.
    static PEAK_RETAINED_PACK_BYTES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

#[cfg(test)]
fn observe_retained_pack_bytes(bytes: usize) {
    PEAK_RETAINED_PACK_BYTES.with(|peak| peak.set(peak.get().max(bytes)));
}

/// Apply staged packs in order while the caller holds the repo lock through
/// ref publication. Signature checks run per pack, so the full history's
/// newly stored-object IDs are not accumulated in memory. The applied key is
/// recorded only after digest, unpack and signature checks succeed.
fn unpack_downloaded_packs(
    store: &ObjectStore,
    staged: StagedPacks,
    applied: &mut AppliedPacks,
    require_signed: bool,
) -> Result<(), DispatchError> {
    for (index, entry) in staged.packs.iter().enumerate() {
        if crate::signal::is_shutdown() {
            return Err(DispatchError::Interrupted);
        }
        let path = staged.directory.path().join(index.to_string());
        let file = std::fs::File::open(&path)?;
        let mut pack = Vec::new();
        // Bound the read even if local temporary bytes were modified. The
        // independently requested digest is rechecked before store effects.
        file.take(entry.bytes as u64 + 1).read_to_end(&mut pack)?;
        if pack.len() != entry.bytes {
            return Err(TransportError::InvalidResponse.into());
        }
        entry.key.verify_bytes(&pack)?;
        let report = PackReader::read(&pack, store)?;
        let unpacked = (report.raw_count + report.delta_count) as usize;
        if unpacked > 0 {
            crate::progress::report(crate::progress::Event::ObjectsUnpacked(unpacked));
        }
        verify_new_object_signatures(store, &report.stored, require_signed)?;
        applied.insert(&entry.key);
        std::fs::remove_file(path)?;
    }
    drop(staged);
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use mkit_core::hash;
    use mkit_transport_memory::MemoryTransport;

    fn h(seed: &str) -> Hash {
        hash::hash(seed.as_bytes())
    }

    #[test]
    fn requested_pack_identity_rejects_valid_substitute_before_unpack() {
        use mkit_core::layout::RepoLayout;
        use mkit_core::object::Blob;
        use mkit_core::pack::PackWriter;

        let objects = [
            Object::Blob(Blob {
                data: b"first".to_vec(),
            }),
            Object::Blob(Blob {
                data: b"second".to_vec(),
            }),
        ];
        let mut a = PackWriter::new();
        let mut b = PackWriter::new();
        for object in &objects {
            a.push_raw(
                object.id().unwrap(),
                &mkit_core::serialize::serialize(object).unwrap(),
            )
            .unwrap();
        }
        for object in objects.iter().rev() {
            b.push_raw(
                object.id().unwrap(),
                &mkit_core::serialize::serialize(object).unwrap(),
            )
            .unwrap();
        }
        let a = a.finish().unwrap();
        let b = b.finish().unwrap();
        let key = pack::pack_key(&a);
        assert_ne!(key, pack::pack_key(&b));
        let tx = MemoryTransport::new();
        tx.upload_pack(&b, &PackKey::from_hash(key)).unwrap();
        let td = tempfile::tempdir().unwrap();
        let applied = AppliedPacks::load(&RepoLayout::single(td.path()), "origin").unwrap();

        assert!(matches!(
            download_pack_chain(&tx, "main", &[key], &applied),
            Err(DispatchError::Transport(TransportError::InvalidResponse))
        ));
        assert!(!applied.contains(&PackKey::from_hash(key)));
    }

    #[test]
    fn requested_packlist_identity_rejects_substitute() {
        let tx = MemoryTransport::new();
        let a = transfer::encode_packlist(None, &[h("pack-a")]).unwrap();
        let b = transfer::encode_packlist(None, &[h("pack-b")]).unwrap();
        let key = hash::hash(&a);
        tx.upload_blob(&b, &PackKey::from_hash(key)).unwrap();
        assert!(matches!(
            download_packlist_node(&tx, key, "refs/heads/main"),
            Err(DispatchError::Transport(TransportError::InvalidResponse))
        ));
    }

    #[test]
    fn fetch_retains_at_most_one_downloaded_pack() {
        let tx = MemoryTransport::new();
        let bytes = vec![0xAB; 1024 * 1024];
        let key = hash::hash(&bytes);
        tx.upload_pack(&bytes, &PackKey::from_hash(key)).unwrap();
        let td = tempfile::tempdir().unwrap();
        let applied =
            AppliedPacks::load(&mkit_core::layout::RepoLayout::single(td.path()), "origin")
                .unwrap();
        PEAK_RETAINED_PACK_BYTES.with(|peak| peak.set(0));
        let downloaded = download_pack_chain(&tx, "main", &vec![key; 32], &applied).unwrap();
        let peak = PEAK_RETAINED_PACK_BYTES.with(std::cell::Cell::get);
        assert!(
            peak <= bytes.len(),
            "retained {peak} bytes for one-MiB packs"
        );
        drop(downloaded);
    }

    #[test]
    fn fetch_staging_budget_failure_cleans_every_partial_file() {
        let tx = MemoryTransport::new();
        let bytes = vec![0xAB; 128];
        let key = hash::hash(&bytes);
        tx.upload_pack(&bytes, &PackKey::from_hash(key)).unwrap();
        let td = tempfile::tempdir().unwrap();
        let applied =
            AppliedPacks::load(&mkit_core::layout::RepoLayout::single(td.path()), "origin")
                .unwrap();
        let result = download_pack_chain_with_limits(
            &tx,
            "main",
            &[key, key],
            &applied,
            128,
            Some(td.path()),
        );
        assert!(
            matches!(result, Err(DispatchError::Io(ref error)) if error.kind() == std::io::ErrorKind::StorageFull)
        );
        assert_eq!(std::fs::read_dir(td.path()).unwrap().count(), 0);
        assert!(!applied.contains(&PackKey::from_hash(key)));
    }

    #[test]
    fn fetch_staging_drop_removes_successful_downloads() {
        let tx = MemoryTransport::new();
        let bytes = b"downloaded object bytes";
        let key = hash::hash(bytes);
        tx.upload_pack(bytes, &PackKey::from_hash(key)).unwrap();
        let td = tempfile::tempdir().unwrap();
        let applied =
            AppliedPacks::load(&mkit_core::layout::RepoLayout::single(td.path()), "origin")
                .unwrap();
        let staged =
            download_pack_chain_with_limits(&tx, "main", &[key], &applied, 1024, Some(td.path()))
                .unwrap();
        assert_eq!(
            std::fs::read(staged.directory.path().join("0")).unwrap(),
            bytes
        );
        assert_eq!(staged.bytes, bytes.len() as u64);
        drop(staged);
        assert_eq!(std::fs::read_dir(td.path()).unwrap().count(), 0);
    }

    #[test]
    fn fetch_staging_digest_failure_cleans_prior_downloads() {
        let tx = MemoryTransport::new();
        let valid = b"valid first bytes";
        let first = hash::hash(valid);
        let wrong = h("missing expected bytes");
        tx.upload_pack(valid, &PackKey::from_hash(first)).unwrap();
        tx.upload_pack(b"substitute", &PackKey::from_hash(wrong))
            .unwrap();
        let td = tempfile::tempdir().unwrap();
        let applied =
            AppliedPacks::load(&mkit_core::layout::RepoLayout::single(td.path()), "origin")
                .unwrap();
        assert!(matches!(
            download_pack_chain_with_limits(
                &tx,
                "main",
                &[first, wrong],
                &applied,
                1024,
                Some(td.path())
            ),
            Err(DispatchError::Transport(TransportError::InvalidResponse))
        ));
        assert_eq!(std::fs::read_dir(td.path()).unwrap().count(), 0);
    }

    /// Publish a real content-addressed node. Corruption tests insert
    /// forged bytes explicitly instead of teaching valid fixtures wrong IDs.
    fn put_node(tx: &MemoryTransport, prev: Option<Hash>, packs: &[Hash]) -> Hash {
        let bytes = transfer::encode_packlist(prev, packs).unwrap();
        let key = hash::hash(&bytes);
        tx.upload_blob(&bytes, &PackKey::from_hash(key)).unwrap();
        key
    }

    #[test]
    fn probe_chain_depth_counts_nodes_and_matches_resolve_pack_chain() {
        let tx = MemoryTransport::new();
        let n1 = put_node(&tx, None, &[h("pack1")]);
        let n2 = put_node(&tx, Some(n1), &[h("pack2")]);
        let n3 = put_node(&tx, Some(n2), &[h("pack3")]);

        let probed = probe_chain(&tx, "main", n3).unwrap();
        assert_eq!(probed.depth, 3);

        // Same node count as the packs resolve_pack_chain flattens, and the
        // packs themselves come back oldest-first — matching `probe_chain`'s
        // own `packs` field too.
        let packs = resolve_pack_chain(&tx, "main", n3).unwrap();
        assert_eq!(packs, vec![h("pack1"), h("pack2"), h("pack3")]);
        assert_eq!(probed.depth, packs.len());
        assert_eq!(probed.packs, packs);
        assert_eq!(probed.head, n3);
    }

    #[test]
    fn probe_chain_depth_of_a_single_node_chain_is_one() {
        let tx = MemoryTransport::new();
        let solo = put_node(&tx, None, &[h("pack-solo")]);
        assert_eq!(probe_chain(&tx, "main", solo).unwrap().depth, 1);
    }

    #[test]
    fn probe_chain_errors_on_a_cycle_exactly_like_resolve_pack_chain() {
        let tx = MemoryTransport::new();
        let a = h("cycle-a");
        let b = h("cycle-b");
        // a -> b -> a: only reachable via hand-built (non-content-addressed)
        // nodes, exercising the shared cycle guard in `walk_pack_chain`.
        let a_bytes = transfer::encode_packlist(Some(b), &[h("pack-a")]).unwrap();
        let b_bytes = transfer::encode_packlist(Some(a), &[h("pack-b")]).unwrap();
        tx.upload_blob(&a_bytes, &PackKey::from_hash(a)).unwrap();
        tx.upload_blob(&b_bytes, &PackKey::from_hash(b)).unwrap();

        assert!(matches!(
            probe_chain(&tx, "main", a).unwrap_err(),
            DispatchError::PackChainInvalid { .. }
        ));
        assert!(matches!(
            resolve_pack_chain(&tx, "main", a).unwrap_err(),
            DispatchError::PackChainInvalid { .. }
        ));
    }

    #[test]
    fn probe_chain_errors_on_an_undownloadable_node_like_resolve_pack_chain() {
        let tx = MemoryTransport::new();
        let ghost = h("never-uploaded");
        assert!(matches!(
            probe_chain(&tx, "main", ghost).unwrap_err(),
            DispatchError::PackChainInvalid { .. }
        ));
        assert!(matches!(
            resolve_pack_chain(&tx, "main", ghost).unwrap_err(),
            DispatchError::PackChainInvalid { .. }
        ));
    }

    // ---- advance_packmap — multi-key (issue #831) -------------------

    fn decode_node_at(tx: &MemoryTransport, key: Hash) -> transfer::PackListNode {
        let bytes = tx.download_blob(&PackKey::from_hash(key)).unwrap();
        transfer::decode_packlist(&bytes).unwrap()
    }

    #[test]
    fn advance_packmap_multi_key_first_push_writes_one_node_in_order() {
        let tx = MemoryTransport::new();
        let (k1, k2, k3) = (h("k1"), h("k2"), h("k3"));
        let tip = h("tip");

        advance_packmap(
            &tx,
            "main",
            &[k1, k2, k3],
            ChainAction::Append {
                self_contained: true,
            },
            None,
            refs::RefWriteCondition::Missing,
            tip,
        )
        .unwrap();

        let pm_head = tx.read_ref(&packmap_ref("main")).unwrap().unwrap();
        let node = decode_node_at(&tx, pm_head);
        assert_eq!(node.prev, None, "first push has no prior chain");
        assert_eq!(
            node.packs,
            vec![k1, k2, k3],
            "all keys land on ONE node, in build/apply order"
        );
        assert_eq!(tx.read_ref("refs/heads/main").unwrap(), Some(tip));
    }

    #[test]
    fn advance_packmap_single_key_matches_pre_831_shape() {
        // Regression: the common case (a push that fits one pack) must
        // still produce exactly the one-pack node it always did.
        let tx = MemoryTransport::new();
        let k1 = h("only-key");
        let tip = h("tip");

        advance_packmap(
            &tx,
            "main",
            &[k1],
            ChainAction::Append {
                self_contained: true,
            },
            None,
            refs::RefWriteCondition::Missing,
            tip,
        )
        .unwrap();

        let pm_head = tx.read_ref(&packmap_ref("main")).unwrap().unwrap();
        assert_eq!(decode_node_at(&tx, pm_head).packs, vec![k1]);
    }

    #[test]
    fn advance_packmap_is_idempotent_when_every_key_already_chained() {
        // A retried push (e.g. a lost head-CAS race after the packmap
        // already landed) must not append a redundant node when the
        // prior chain already carries every key this attempt would add
        // — only the head still needs to move.
        let tx = MemoryTransport::new();
        let (k1, k2, k3) = (h("k1"), h("k2"), h("k3"));
        let prior_head = put_node(&tx, None, &[k1, k2, k3]);
        tx.update_ref(
            &packmap_ref("main"),
            refs::RefWriteCondition::Missing,
            &prior_head,
        )
        .unwrap();
        let tip = h("tip");

        advance_packmap(
            &tx,
            "main",
            &[k1, k2, k3],
            ChainAction::Append {
                self_contained: true,
            },
            None,
            refs::RefWriteCondition::Missing,
            tip,
        )
        .unwrap();

        assert_eq!(
            tx.read_ref(&packmap_ref("main")).unwrap(),
            Some(prior_head),
            "idempotent retry must not write a new node"
        );
        assert_eq!(tx.read_ref("refs/heads/main").unwrap(), Some(tip));
    }

    #[test]
    fn advance_packmap_appends_when_only_some_keys_already_chained() {
        // Partial overlap is NOT idempotency — a different push that
        // happened to produce one identical pack is not the same push
        // retried. A fresh node carrying the full new set is appended.
        let tx = MemoryTransport::new();
        let (k1, k2, k3) = (h("k1"), h("k2"), h("k3"));
        let prior_head = put_node(&tx, None, &[k1]);
        tx.update_ref(
            &packmap_ref("main"),
            refs::RefWriteCondition::Missing,
            &prior_head,
        )
        .unwrap();
        let tip = h("tip");

        advance_packmap(
            &tx,
            "main",
            &[k1, k2, k3],
            ChainAction::Append {
                self_contained: true,
            },
            None,
            refs::RefWriteCondition::Missing,
            tip,
        )
        .unwrap();

        let new_head = tx.read_ref(&packmap_ref("main")).unwrap().unwrap();
        assert_ne!(new_head, prior_head, "a new node must be appended");
        let node = decode_node_at(&tx, new_head);
        assert_eq!(node.prev, Some(prior_head));
        assert_eq!(node.packs, vec![k1, k2, k3]);

        // The chain still resolves and flattens both nodes' packs.
        assert_eq!(
            resolve_pack_chain(&tx, "main", new_head).unwrap(),
            vec![k1, k1, k2, k3],
        );
    }

    // ---- lost-response retries — SPEC-TRANSPORT §7 (MKIT-58) --------

    /// How [`LostResponseTransport`] models the retry ladder re-issuing a
    /// CAS write whose first attempt landed but whose response was lost.
    #[derive(Clone, Copy)]
    enum Reissue {
        /// `update_ref` on a `refs/heads/` name is applied twice and the
        /// caller sees only the second result. `advance_refs` keeps the
        /// ordered packmap-then-head default, so the re-issued head write
        /// surfaces as `AdvanceOutcome::HeadConflict` after the packmap
        /// write landed.
        HeadUpdate,
        /// `advance_refs` is transactional and applied twice; the caller
        /// sees only the second outcome (`PackmapConflict`: both refs
        /// already moved).
        Advance,
        /// Nothing is re-issued: a plain transport, for the genuine
        /// non-fast-forward cases.
        None,
    }

    struct LostResponseTransport {
        inner: MemoryTransport,
        reissue: Reissue,
    }

    impl LostResponseTransport {
        fn new(reissue: Reissue) -> Self {
            Self {
                inner: MemoryTransport::new(),
                reissue,
            }
        }

        fn atomic_advance_once(
            &self,
            head: (&str, refs::RefWriteCondition, &Hash),
            packmap: (&str, refs::RefWriteCondition, &Hash),
        ) -> Result<AdvanceOutcome, TransportError> {
            fn holds(c: refs::RefWriteCondition, cur: Option<Hash>) -> bool {
                match c {
                    refs::RefWriteCondition::Any => true,
                    refs::RefWriteCondition::Missing => cur.is_none(),
                    refs::RefWriteCondition::Match(h) => cur == Some(h),
                }
            }
            if !holds(packmap.1, self.inner.read_ref(packmap.0)?) {
                return Ok(AdvanceOutcome::PackmapConflict);
            }
            if !holds(head.1, self.inner.read_ref(head.0)?) {
                return Ok(AdvanceOutcome::HeadConflict);
            }
            self.inner.update_ref(packmap.0, packmap.1, packmap.2)?;
            self.inner.update_ref(head.0, head.1, head.2)?;
            Ok(AdvanceOutcome::Committed)
        }
    }

    impl Transport for LostResponseTransport {
        fn upload_pack(&self, bytes: &[u8], key: &PackKey) -> Result<(), TransportError> {
            self.inner.upload_pack(bytes, key)
        }
        fn download_pack(&self, key: &PackKey) -> Result<Vec<u8>, TransportError> {
            self.inner.download_pack(key)
        }
        fn pack_exists(&self, key: &PackKey) -> Result<bool, TransportError> {
            self.inner.pack_exists(key)
        }
        fn upload_blob(&self, bytes: &[u8], key: &PackKey) -> Result<(), TransportError> {
            self.inner.upload_blob(bytes, key)
        }
        fn download_blob(&self, key: &PackKey) -> Result<Vec<u8>, TransportError> {
            self.inner.download_blob(key)
        }
        fn update_ref(
            &self,
            name: &str,
            condition: refs::RefWriteCondition,
            hash: &Hash,
        ) -> Result<(), TransportError> {
            if matches!(self.reissue, Reissue::HeadUpdate) && name.starts_with("refs/heads/") {
                // First attempt lands; its response is lost.
                let _ = self.inner.update_ref(name, condition, hash);
            }
            self.inner.update_ref(name, condition, hash)
        }
        fn read_ref(&self, name: &str) -> Result<Option<Hash>, TransportError> {
            self.inner.read_ref(name)
        }
        fn list_refs(&self, prefix: &str) -> Result<Vec<refs::Ref>, TransportError> {
            self.inner.list_refs(prefix)
        }
        fn advance_refs(
            &self,
            head_ref: &str,
            head_condition: refs::RefWriteCondition,
            head_value: &Hash,
            packmap_ref: &str,
            packmap_condition: refs::RefWriteCondition,
            packmap_value: &Hash,
        ) -> Result<AdvanceOutcome, TransportError> {
            let head = (head_ref, head_condition, head_value);
            let packmap = (packmap_ref, packmap_condition, packmap_value);
            match self.reissue {
                Reissue::Advance => {
                    // First attempt lands; its response is lost.
                    let _ = self.atomic_advance_once(head, packmap)?;
                    self.atomic_advance_once(head, packmap)
                }
                Reissue::HeadUpdate | Reissue::None => {
                    // The trait's ordered default: packmap, then head.
                    match self.update_ref(packmap_ref, packmap_condition, packmap_value) {
                        Ok(()) => {}
                        Err(TransportError::RefConflict) => {
                            return Ok(AdvanceOutcome::PackmapConflict);
                        }
                        Err(e) => return Err(e),
                    }
                    match self.update_ref(head_ref, head_condition, head_value) {
                        Ok(()) => Ok(AdvanceOutcome::Committed),
                        Err(TransportError::RefConflict) => Ok(AdvanceOutcome::HeadConflict),
                        Err(e) => Err(e),
                    }
                }
            }
        }
        fn supports_atomic_advance(&self) -> bool {
            matches!(self.reissue, Reissue::Advance)
        }
    }

    fn advance_first_push(
        tx: &dyn Transport,
        keys: &[Hash],
        tip: Hash,
    ) -> Result<(), DispatchError> {
        advance_packmap(
            tx,
            "main",
            keys,
            ChainAction::Append {
                self_contained: true,
            },
            None,
            refs::RefWriteCondition::Missing,
            tip,
        )
    }

    #[test]
    fn advance_packmap_head_conflict_from_reissued_head_write_is_success() {
        // Ordered advance: the packmap write lands, the head write lands
        // but its response is lost, and the ladder's re-issue reports
        // `HeadConflict` for our own write. SPEC-TRANSPORT §7: read the
        // head; it holds our tip, so the push landed.
        let tx = LostResponseTransport::new(Reissue::HeadUpdate);
        let tip = h("tip");
        advance_first_push(&tx, &[h("k1")], tip).unwrap();
        assert_eq!(tx.read_ref("refs/heads/main").unwrap(), Some(tip));
        let pm = tx.read_ref(&packmap_ref("main")).unwrap().unwrap();
        assert_eq!(decode_node_at(&tx.inner, pm).packs, vec![h("k1")]);
    }

    #[test]
    fn advance_packmap_reissued_atomic_advance_is_success() {
        // Transactional advance: the first attempt commits both refs and
        // its response is lost. The re-issue reports `PackmapConflict`; the
        // re-read finds our node already chained, so only the head write
        // remains, and that write's `RefConflict` is our own landed head.
        let tx = LostResponseTransport::new(Reissue::Advance);
        let tip = h("tip");
        advance_first_push(&tx, &[h("k1"), h("k2")], tip).unwrap();
        assert_eq!(tx.read_ref("refs/heads/main").unwrap(), Some(tip));
        let pm = tx.read_ref(&packmap_ref("main")).unwrap().unwrap();
        let node = decode_node_at(&tx.inner, pm);
        assert_eq!(node.prev, None, "no redundant node was appended");
        assert_eq!(node.packs, vec![h("k1"), h("k2")]);
    }

    #[test]
    fn advance_packmap_reissued_rebaseline_reset_is_success() {
        // A re-baseline reset re-issued after it landed: the retry loop
        // re-reads the packmap (now our node) and CASes off it, and the
        // head precondition then fails against our own landed tip.
        let tx = LostResponseTransport::new(Reissue::Advance);
        let old = h("old-tip");
        let prior = put_node(&tx.inner, None, &[h("old-pack")]);
        tx.inner
            .update_ref(
                &packmap_ref("main"),
                refs::RefWriteCondition::Missing,
                &prior,
            )
            .unwrap();
        tx.inner
            .update_ref("refs/heads/main", refs::RefWriteCondition::Missing, &old)
            .unwrap();
        let tip = h("tip");
        advance_packmap(
            &tx,
            "main",
            &[h("full")],
            ChainAction::ResetSelfContained,
            None,
            refs::RefWriteCondition::Match(old),
            tip,
        )
        .unwrap();
        assert_eq!(tx.read_ref("refs/heads/main").unwrap(), Some(tip));
    }

    #[test]
    fn advance_packmap_genuine_head_conflict_stays_non_fast_forward() {
        // Another writer moved the head to a different commit: the re-read
        // does not find our tip, so the push is still rejected.
        for reissue in [Reissue::None, Reissue::Advance] {
            let tx = LostResponseTransport::new(reissue);
            let theirs = h("their-tip");
            tx.inner
                .update_ref("refs/heads/main", refs::RefWriteCondition::Missing, &theirs)
                .unwrap();
            let err = advance_first_push(&tx, &[h("k1")], h("tip")).unwrap_err();
            assert!(
                matches!(err, DispatchError::NonFastForwardPush { ref branch } if branch == "main"),
                "{err:?}"
            );
            assert_eq!(tx.read_ref("refs/heads/main").unwrap(), Some(theirs));
        }
    }

    #[test]
    fn commit_head_reissued_write_is_success() {
        let tx = LostResponseTransport::new(Reissue::HeadUpdate);
        let tip = h("tip");
        commit_head(
            &tx,
            "refs/heads/main",
            refs::RefWriteCondition::Missing,
            &tip,
            "main",
        )
        .unwrap();
        assert_eq!(tx.read_ref("refs/heads/main").unwrap(), Some(tip));
    }

    #[test]
    fn commit_head_genuine_conflict_stays_non_fast_forward() {
        let tx = LostResponseTransport::new(Reissue::None);
        let theirs = h("their-tip");
        tx.inner
            .update_ref("refs/heads/main", refs::RefWriteCondition::Missing, &theirs)
            .unwrap();
        let err = commit_head(
            &tx,
            "refs/heads/main",
            refs::RefWriteCondition::Match(h("stale-lease")),
            &h("tip"),
            "main",
        )
        .unwrap_err();
        assert!(
            matches!(err, DispatchError::NonFastForwardPush { ref branch } if branch == "main"),
            "{err:?}"
        );
        assert_eq!(tx.read_ref("refs/heads/main").unwrap(), Some(theirs));
    }

    // ---- verify_new_object_signatures — sequential/rayon fan-out parity ----

    /// A validly-signed, independently-hashing commit — every `seed`
    /// produces a distinct object, matching the bench fixtures'
    /// no-dedup convention.
    fn signed_commit_bytes(kp: &mkit_core::sign::KeyPair, seed: usize) -> Vec<u8> {
        use mkit_core::object::{Commit, Identity};
        let mut c = Commit::new_unannotated(
            hash::hash(format!("tree #{seed}").as_bytes()),
            Vec::new(),
            Identity::ed25519(kp.public.0),
            kp.public.0,
            format!("packmap fanout test fixture #{seed}").into_bytes(),
            1_700_000_000 + seed as u64,
            [0u8; 64],
        );
        c.signature = mkit_core::sign::sign_commit(&c, kp).unwrap().0;
        mkit_core::serialize::serialize(&Object::Commit(c)).unwrap()
    }

    /// Large enough that `stored.len()` clears
    /// [`verify_fanout_threshold`] on any realistic thread-pool size
    /// (`VERIFY_FANOUT_ENTRIES_PER_THREAD` * rayon's pool size), so
    /// these tests exercise the `par_iter` branch rather than the
    /// sequential one the integration tests in
    /// `tests/verify_signatures_on_fetch.rs` already cover.
    const LARGE_BATCH: usize = 512;

    #[test]
    fn verify_new_object_signatures_accepts_a_large_validly_signed_batch() {
        let dir = tempfile::tempdir().unwrap();
        let layout = mkit_core::layout::RepoLayout::single(dir.path());
        let store = ObjectStore::init(&layout).unwrap();
        let kp = mkit_core::sign::KeyPair::generate().unwrap();
        let stored: Vec<Hash> = (0..LARGE_BATCH)
            .map(|i| store.write(&signed_commit_bytes(&kp, i)).unwrap())
            .collect();
        assert!(stored.len() >= verify_fanout_threshold());

        verify_new_object_signatures(&store, &stored, true)
            .expect("every commit in the batch is validly signed");
    }

    /// Regression for `verify_slice`'s read step: a chunk mixing unsigned
    /// object kinds (Blob/Tree — neither carries a signature) with signed
    /// commits must still verify correctly. `verify_slice` reads every
    /// hash to learn its type but only *retains* Commit/Remix/Tag entries
    /// afterward (dropping Blob/Tree/ChunkedBlob/Delta immediately) — this
    /// pins that the filtering doesn't drop a commit it should have kept,
    /// doesn't get confused by interleaving, and both a validly-signed and
    /// a tampered commit are still correctly accepted/rejected when
    /// surrounded by unsigned entries on both sides.
    #[test]
    fn verify_new_object_signatures_mixed_with_unsigned_object_kinds() {
        let dir = tempfile::tempdir().unwrap();
        let layout = mkit_core::layout::RepoLayout::single(dir.path());
        let store = ObjectStore::init(&layout).unwrap();
        let kp = mkit_core::sign::KeyPair::generate().unwrap();

        let blob = |seed: usize| -> Hash {
            store
                .write(
                    &mkit_core::serialize::serialize(&Object::Blob(mkit_core::object::Blob {
                        data: format!("unsigned blob fixture #{seed}").into_bytes(),
                    }))
                    .unwrap(),
                )
                .unwrap()
        };

        // Blob, commit, blob, commit, ... — unsigned entries on both
        // sides of every signed one, so a filtering bug that drops or
        // misattributes a neighbor would show up either direction.
        let mut stored: Vec<Hash> = Vec::with_capacity(2 * LARGE_BATCH);
        for i in 0..LARGE_BATCH {
            stored.push(blob(i));
            stored.push(store.write(&signed_commit_bytes(&kp, i)).unwrap());
        }
        assert!(stored.len() >= verify_fanout_threshold());

        verify_new_object_signatures(&store, &stored, true)
            .expect("unsigned entries must not affect verifying the signed ones");

        // Tamper one of the commits — must still be caught even though
        // it's surrounded by unsigned entries the batch/fallback paths
        // both skip.
        let commit_index = 2 * (LARGE_BATCH / 2) + 1;
        let Object::Commit(mut c) = store.read_object(&stored[commit_index]).unwrap() else {
            panic!("expected commit");
        };
        c.signature[0] ^= 0xff;
        let tampered_bytes = mkit_core::serialize::serialize(&Object::Commit(c)).unwrap();
        stored[commit_index] = store.write(&tampered_bytes).unwrap();

        let err = verify_new_object_signatures(&store, &stored, true)
            .expect_err("a tampered commit must still be caught alongside unsigned entries");
        assert!(matches!(err, DispatchError::UnsignedOrInvalidObject { .. }));
    }

    #[test]
    fn verify_new_object_signatures_rejects_one_bad_signature_in_a_large_batch() {
        let dir = tempfile::tempdir().unwrap();
        let layout = mkit_core::layout::RepoLayout::single(dir.path());
        let store = ObjectStore::init(&layout).unwrap();
        let kp = mkit_core::sign::KeyPair::generate().unwrap();
        let mut stored: Vec<Hash> = (0..LARGE_BATCH)
            .map(|i| store.write(&signed_commit_bytes(&kp, i)).unwrap())
            .collect();
        assert!(stored.len() >= verify_fanout_threshold());

        // Tamper one commit's signature in place — content-addressing stays
        // intact (re-written under a new hash), only the crypto is wrong,
        // same attacker model as `tamper_head_commit_signature` in the
        // integration test.
        let Object::Commit(mut c) = store.read_object(&stored[LARGE_BATCH / 2]).unwrap() else {
            panic!("expected commit");
        };
        c.signature[0] ^= 0xff;
        let tampered_bytes = mkit_core::serialize::serialize(&Object::Commit(c)).unwrap();
        let tampered_hash = store.write(&tampered_bytes).unwrap();
        stored[LARGE_BATCH / 2] = tampered_hash;

        let err = verify_new_object_signatures(&store, &stored, true)
            .expect_err("a batch with one bad signature must be rejected");
        assert!(matches!(err, DispatchError::UnsignedOrInvalidObject { .. }));
    }

    #[test]
    fn verify_new_object_signatures_never_reads_past_the_chunk_containing_the_first_bad_signature()
    {
        // Regression for the chunked fan-out: a hostile remote must not be
        // able to force verification work past the chunk containing the
        // first invalid signature. The first `chunk_size` entries are a
        // validly-signed batch with one tampered signature; every entry
        // after that names a digest that was NEVER written to the store.
        // If the implementation ever started a second chunk, reading one
        // of those would surface as a store error (`ObjectNotFound`), not
        // `UnsignedOrInvalidObject` — so this test fails deterministically
        // if chunking regresses back to a single flat fan-out (or any
        // other shape that can read beyond the first failing chunk).
        let dir = tempfile::tempdir().unwrap();
        let layout = mkit_core::layout::RepoLayout::single(dir.path());
        let store = ObjectStore::init(&layout).unwrap();
        let kp = mkit_core::sign::KeyPair::generate().unwrap();
        let chunk_size = verify_chunk_size();

        let mut stored: Vec<Hash> = (0..chunk_size)
            .map(|i| store.write(&signed_commit_bytes(&kp, i)).unwrap())
            .collect();
        let Object::Commit(mut c) = store.read_object(&stored[0]).unwrap() else {
            panic!("expected commit");
        };
        c.signature[0] ^= 0xff;
        let tampered_bytes = mkit_core::serialize::serialize(&Object::Commit(c)).unwrap();
        let tampered_hash = store.write(&tampered_bytes).unwrap();
        stored[0] = tampered_hash;
        stored.extend((0..chunk_size * 3).map(|i| h(&format!("never-written-{i}"))));

        let err = verify_new_object_signatures(&store, &stored, true)
            .expect_err("the tampered signature in the first chunk must reject the fetch");
        match err {
            DispatchError::UnsignedOrInvalidObject { hash: got, .. } => {
                assert_eq!(got, hash::to_hex(&tampered_hash));
            }
            other => panic!("expected UnsignedOrInvalidObject, got {other:?}"),
        }
    }

    #[test]
    fn verify_new_object_signatures_require_signed_false_skips_a_large_batch() {
        // `require_signed = false` must short-circuit to `Ok(())` without
        // reading the store at all, regardless of batch size — pass a
        // digest list naming objects that were never written, so any
        // attempted read would panic/error and fail the test.
        let dir = tempfile::tempdir().unwrap();
        let layout = mkit_core::layout::RepoLayout::single(dir.path());
        let store = ObjectStore::init(&layout).unwrap();
        let stored: Vec<Hash> = (0..LARGE_BATCH)
            .map(|i| h(&format!("never-written-{i}")))
            .collect();

        verify_new_object_signatures(&store, &stored, false)
            .expect("require_signed=false must no-op regardless of batch size");
    }
}