alkgit_substrate/

ingest.rs

//! The substrate's `GitPackIngest` implement ([`PackIngest`]) — ADR-020
//! §2's ingestion composition, for real: the received pack indexes into
//! a budgeted host working area (thin bases resolving through the
//! substrate Find), the indexed objects unpack into alkblobs as
//! loose-form objects under the git-object family (the wave-7 object
//! front's family put — the store's verification-on-put *is* the oid
//! re-derivation, corruption cannot enter undetected), the guards ride
//! the w8-pin-registry's custody seam, connectivity closes the graph
//! from the pushed tips, and the working area is discarded. The
//! reference shape is POC-O's `ingest.rs`; the reference semantics are
//! the wave-3 doubles' `MemPackIngest` (`tests/common/mod.rs` in the
//! core crate).
//!
//! # The recorded design decisions (the wave-8 gate's reading list)
//!
//! - **The decompressed-size cap's enforcement, layered** (the task's
//!   named design point — how the `Limits::max_object_size` bound
//!   holds through gix's own index/inflate machinery):
//!   1. the *counting reader* is the defense-in-depth layer the
//!      wave-3 gate confirmed: the pack stream's byte total is bounded
//!      by `Limits::max_pack_size`; a breach is a pack-level failure
//!      mapping into the `unpack` ng leg, never `Err` (the wire side's
//!      relay breaches separately fail the session closed);
//!   2. the *indexing leg* rides gix's indexing machinery
//!      (`gix_pack::Bundle::write_to_directory`) with its
//!      `alloc_limit_bytes` knob set to the cap: gix's decode paths vet
//!      every allocation-relevant pack-declared quantity against the
//!      limit before reserving, so a hostile declaration beyond the cap
//!      errors before any allocation, and the machinery's
//!      exact-decompression discipline (a stream that stops early or
//!      overproduces against its declared buffer is corruption) makes a
//!      lying-small declaration fail too — the hostile-declared-size
//!      class maps to the unpack ng leg over the real machinery;
//!   3. the *unpack leg* is this module's own incremental machinery —
//!      the wave-5 doubles' guard verbatim: each entry's declared size
//!      is vetted against the cap before anything is read, the inflate
//!      reads in bounded chunks accumulating only produced bytes (the
//!      running total aborts at the cap), and each delta's materialized
//!      result is checked against its declaration and produced
//!      incrementally — nothing on the unpack path ever sizes an
//!      allocation from the pack's own declaration.
//! - **The working area's shape**: a fresh per-push directory under the
//!   constructor's root knob (no env — conventions 4/5); the disk
//!   shadow is the pack the counting reader bounds (pack + index ≈ the
//!   pack budget — no new `Limits` field, ADR-009's no-new-kinds rule).
//!   The authoritative object home is the store; the area (temporary
//!   pack + index) is discarded after unpack — on every path, the
//!   explicit return paths' removal riding a best-effort residue
//!   removal that also covers a panic's unwind (Drop's bounded
//!   directory rm — the one blocking best-effort on the async side).
//! - **Unpack order and the store's role**: the indexing machinery
//!   normalizes the pack (thin-base entries inserted physically and
//!   their ref-deltas rewritten, in-pack ref-deltas resolved through
//!   the index), so the unpack iterates the written pack in pack order
//!   over the index's entries; every materialized object re-derives
//!   (the pack-side oid answer), cross-checks against the index's own
//!   oid, and goes through the store's batch form — one
//!   [`alkblobs::Batch`] group over the working set (the store's
//!   pin-before-publish ordering, one commit discipline). Existing
//!   content dedups in the store; re-putting every indexed object
//!   (rather than skipping known ones) keeps the sweep race closed: an
//!   object present-but-sweep-eligible could otherwise vanish
//!   mid-window under a bare presence check. Delta bases read back
//!   through the Find front (verification-on-get is the same
//!   re-derivation guard), one base body at a time — the unpack
//!   stage's memory is the cap's working set, never O(pack).
//! - **Connectivity**: the full-graph walk over the substrate Find
//!   from the pushed tips — commits walk parents and trees, trees walk
//!   entries, tags walk targets, blobs terminate (`gix-fsck`'s
//!   `Connectivity::check_commit` does not walk parents; this is the
//!   owning walk, the same two-stage discipline as generation). A
//!   missing closure member fails into the `unpack` ng leg naming the
//!   missing object.
//! - **Status shapes**: pack-level validation failures (parse, index,
//!   thin-base miss, the cap, connectivity) land in
//!   [`PreparedPush::unpack`](alkgit::PreparedPush)'s ng leg with a
//!   server-chosen reason; `updates` returns the input list verbatim
//!   on every path (ADR-018 §5); `Err` stays
//!   `StorageError` — infrastructure only
//!   (working-area construction, task joins, backing-store legs). The
//!   empty pack (zero objects) is valid and reports as such; the
//!   push's tips must then exist in the store (the same connectivity
//!   walk, vacuous over deletes-only updates). The ng legs retain
//!   nothing — their guards drop with the prepare call and the
//!   just-put objects are the same sweep-eligible recovery the crash
//!   case rides; a fully ingested push retains its [`alkblobs::Batch`]
//!   into the composition-internal `PendingPinRegistry` keyed by
//!   `repo_id`, correlated content-wise with the verbatim prepared
//!   update list (the custody seam the w8-pin-registry task owns).
//!
//! # Threading
//!
//! The whole composition runs on `tokio::task::spawn_blocking` with
//! owned handles moved in (backend.md §concurrency model — the permit
//! stays wire-side, the impl's threading invisible); every sync→async
//! hop goes through [`ObjectFront::block_on`] from the blocking
//! contexts — no consumer constructs its own bridge.

use std::collections::{HashMap, HashSet};
use std::io::{self, Read as _};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use alkblobs::Batch;
use gix_hash::ObjectId;
use gix_object::Kind;
use gix_pack::bundle::write::Options as BundleOptions;
use gix_pack::data::entry::Header as PackHeader;

use alkgit::{
    GitPackIngest, Limits, PreparedPush, PushOptions, RefUpdate, RepoRecord, StorageError,
};
use async_trait::async_trait;

use crate::odb::{key_to_oid, loose_form, store_leg_error, ObjectFront, SessionFind};
use crate::pins::{PendingPinRegistry, PinCustody};

/// The derive algorithm every family put carries (the odb layer's
/// serving-path member — the sha1 git-object family, OQ-05).
use crate::odb::GIT_OBJECT_ALGORITHM;

/// The inflate window the unpack leg reads in — bounded output slices;
/// the produced bytes accumulate beyond the window under the cap.
const INFLATE_CHUNK_BYTES: usize = 65536;

/// The per-process uniqueness counter behind the working-area
/// directory names (pid + nanoseconds + the counter — no client input
/// ever reaches a filesystem path).
static WORK_AREA_COUNTER: AtomicU64 = AtomicU64::new(0);

/// The substrate's [`GitPackIngest`] implement.
///
/// The construction is the composition entry's (substrate.md §public
/// API — the receive-pack composition is internal to the trait impls,
/// not a public member; the pending-pin registry is shared with the
/// wrapped `GitRefs` implement the same entry wires — the w8
/// composition-entry task joins the names into the OQ-03 inventory).
/// The working-area root is the constructor knob.
pub struct PackIngest {
    front: ObjectFront,
    pins: Arc<PendingPinRegistry>,
    root: PathBuf,
}

impl PackIngest {
    /// The implement over the object front, the pending-pin registry
    /// shared with the wired `GitRefs` implement, and the working-area
    /// root — the assembler-level knob the composition entry's config
    /// carries (no env).
    pub(crate) fn new(front: ObjectFront, pins: Arc<PendingPinRegistry>, root: PathBuf) -> Self {
        PackIngest { front, pins, root }
    }
}

/// One leg's failure split: every pack-level validation problem rides
/// the ng leg as its server-chosen reason; a backing-store failure is
/// the seam's infrastructure class ([`StorageError`], the `Err`).
enum LegFailure {
    Pack(String),
    Backing(StorageError),
}

impl From<StorageError> for LegFailure {
    fn from(error: StorageError) -> Self {
        LegFailure::Backing(error)
    }
}

impl From<String> for LegFailure {
    fn from(reason: String) -> Self {
        LegFailure::Pack(reason)
    }
}

/// The best-effort residue remover on the working area: held across
/// the prepare call, dropped on every exit — the return paths' removal
/// and the panic-path's unwind reach the same cleanup.
struct WorkArea(PathBuf);

impl Drop for WorkArea {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}

/// The counting reader (the defense-in-depth layer the wave-3 gate
/// confirmed): past `limit` every read errors — the indexing machinery
/// fails and the push reports the `unpack` ng leg, never `Err`. EOF
/// before the budget is the ordinary stream end (an under-length input
/// is the machinery's own parse error); a pack of exactly the budget
/// consumes without a further read.
struct BudgetedReader<R> {
    inner: R,
    bytes_seen: u64,
    limit: u64,
}

impl<R> BudgetedReader<R> {
    fn new(inner: R, limit: u64) -> Self {
        BudgetedReader {
            inner,
            bytes_seen: 0,
            limit,
        }
    }
}

impl<R: io::Read> io::Read for BudgetedReader<R> {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if self.bytes_seen >= self.limit {
            return Err(io::Error::other(format!(
                "the pack stream exceeded the maximum pack size budget of {} bytes",
                self.limit
            )));
        }
        let read = self.inner.read(buf)?;
        self.bytes_seen += read as u64;
        Ok(read)
    }
}

/// A store read's failure classification along the ingest legs: a
/// corruption-classified backing state is a pack-level failure (the ng
/// leg names the corruption), every other backing failure is
/// infrastructure.
fn store_read_failure(oid: &ObjectId, error: StorageError) -> LegFailure {
    match error {
        StorageError::Invalid(message) => {
            LegFailure::Pack(format!("corrupt object {oid}: {message}"))
        }
        other => LegFailure::Backing(other),
    }
}

/// The indexing leg (blocking): the budgeted stream indexes into the
/// working area — pack + index on disk over a `Verify` iteration, thin
/// bases resolving through the substrate Find — returning the written
/// pair. A zero-object pack yields `None`: the machinery never reaches
/// a zero-object pack's trailer, so the leg verifies it itself (the
/// doubles' zero-object branch). Every error here is pack-level (the
/// ng leg's server-chosen reason).
fn index_leg(
    pack: Box<dyn io::Read + Send>,
    dir: PathBuf,
    thin: SessionFind,
    budget: u64,
    cap: u64,
) -> Result<Option<(PathBuf, PathBuf)>, String> {
    let mut counted = BudgetedReader::new(pack, budget);
    let mut header = [0u8; gix_pack::data::header::SIZE];
    counted
        .read_exact(&mut header)
        .map_err(|_| "the pack stream ended in the header".to_string())?;
    let (_version, object_count) = gix_pack::data::header::decode(&header)
        .map_err(|error| format!("the pack header does not decode: {error}"))?;

    if object_count == 0 {
        let mut trailer = [0u8; gix_hash::Kind::Sha1.len_in_bytes()];
        counted
            .read_exact(&mut trailer)
            .map_err(|_| "the zero-object pack stream ends before the trailer".to_string())?;
        let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha1);
        hasher.update(&header);
        let digest = hasher
            .try_finalize()
            .map_err(|error| format!("the pack trailer does not hash: {error}"))?;
        if digest != ObjectId::from(trailer) {
            return Err(
                "the zero-object pack's trailer does not verify against the pack bytes".to_owned(),
            );
        }
        return Ok(None);
    }

    let interrupt = AtomicBool::new(false);
    let mut progress = gix_utils::progress::Discard;
    let mut pack_stream = io::BufReader::new(io::Cursor::new(header.to_vec()).chain(counted));
    let outcome = gix_pack::Bundle::write_to_directory(
        &mut pack_stream,
        Some(&dir),
        &mut progress,
        &interrupt,
        Some(thin),
        gix_hash::Kind::Sha1,
        BundleOptions {
            thread_limit: Some(1),
            iteration_mode: gix_pack::data::input::Mode::Verify,
            index_version: Default::default(),
            alloc_limit_bytes: Some(usize::try_from(cap).unwrap_or(usize::MAX)),
            compression: gix_zlib::Compression::DEFAULT,
        },
    )
    .map_err(|error| format!("the pack does not index: {error}"))?;
    match (outcome.index_path, outcome.data_path) {
        (Some(index_path), Some(pack_path)) => Ok(Some((pack_path, index_path))),
        _ => Ok(None),
    }
}

/// A materialized pack entry's oid (the pack-side re-derivation; the
/// store's verification-on-put repeats it at put time).
fn rederive_object(kind: Kind, body: &[u8]) -> Result<ObjectId, String> {
    gix_object::compute_hash(gix_hash::Kind::Sha1, kind, body)
        .map_err(|error| format!("the pack entry object does not hash: {error}"))
}

/// The async-hop discipline at the put seam, in one function: the
/// loose form goes through the store's batch form
/// ([`alkblobs::Store::put_in_batch`]) via [`ObjectFront::block_on`]
/// from the blocking context (the one-place bridge rule), and the
/// returned family key must re-derive to the expected oid — the
/// verification-on-put's cross-check on the batch path.
fn put_in_batch(
    front: &ObjectFront,
    batch: &mut Batch,
    kind: Kind,
    body: &[u8],
    expected: ObjectId,
) -> Result<(), StorageError> {
    let form = loose_form(kind, body);
    let store = front.store().clone();
    let future = async {
        store
            .put_in_batch(
                batch,
                GIT_OBJECT_ALGORITHM,
                Some(form.len() as u64),
                io::Cursor::new(&form),
            )
            .await
    };
    let key = front
        .block_on(future)
        .map_err(|error| store_leg_error("alkblobs put_in_batch", error))?;
    if key_to_oid(&key) != Some(expected) {
        return Err(StorageError::Invalid(format!(
            "the put returned a key that does not re-derive to {expected}"
        )));
    }
    Ok(())
}

/// One delta base read back through the Find front
/// ([`ObjectFront::get`]'s verification-on-get is the read's
/// re-derivation); the delivered kind must match the delta chain's.
fn base_object(
    front: &ObjectFront,
    base_oid: ObjectId,
    base_kind: Kind,
) -> Result<(Kind, Vec<u8>), LegFailure> {
    let found = front.block_on(front.get(&base_oid));
    match found {
        Ok(None) => Err(LegFailure::Pack(format!("missing object {base_oid}"))),
        Err(error) => Err(store_read_failure(&base_oid, error)),
        Ok(Some((delivered_kind, body))) => {
            if delivered_kind != base_kind {
                return Err(LegFailure::Pack(format!(
                    "the base object {base_oid} delivered kind {delivered_kind:?} against \
                     the delta chain's kind {base_kind:?}"
                )));
            }
            Ok((base_kind, body))
        }
    }
}

/// One pack-entry size field of a delta stream header: 7-bit
/// little-endian groups with the continuation bit.
fn delta_size_at(delta: &[u8], offset: usize) -> Result<(u64, usize), String> {
    let mut size = 0u64;
    let mut shift = 0u32;
    let mut cursor = offset;
    loop {
        let Some(&byte) = delta.get(cursor) else {
            return Err("the delta header is truncated".to_string());
        };
        cursor += 1;
        size |= u64::from(byte & 0x7f) << shift;
        shift += 7;
        if byte & 0x80 == 0 {
            return Ok((size, cursor - offset));
        }
        if shift >= 64 {
            return Err("the delta header size overflow".to_string());
        }
    }
}

/// Applies a git delta stream to its base: the declared base size must
/// match, the declared result stays under `cap` (the decompressed-size
/// cap — the delta's materialized result is one of the objects the
/// pack produces), and the instructions must produce exactly the
/// declared result size. The result accumulates incrementally — the
/// declared size is checked against, never pre-allocated; the loop
/// itself aborts once the produced bytes pass the declaration.
fn apply_delta(base: &[u8], delta: &[u8], cap: u64) -> Result<Vec<u8>, String> {
    let (base_size, base_header_len) = delta_size_at(delta, 0)?;
    let (result_size, result_header_len) = delta_size_at(delta, base_header_len)?;
    if base_size != u64::try_from(base.len()).unwrap_or(u64::MAX) {
        return Err(format!(
            "the delta declares a base size of {base_size} but the base object is {} bytes",
            base.len()
        ));
    }
    if result_size > u64::from(u32::MAX) {
        return Err("the delta declares an out-of-bounds result size".to_string());
    }
    if result_size > cap {
        return Err(format!(
            "the delta declares a materialized result of {result_size} bytes beyond \
             the ingestion decompressed-size cap of {cap} bytes"
        ));
    }
    let mut out = Vec::new();
    let mut cursor = base_header_len + result_header_len;
    while let Some(&cmd) = delta.get(cursor) {
        cursor += 1;
        if cmd & 0x80 != 0 {
            let mut ofs = 0u64;
            let mut size = 0u64;
            for (bit, shift) in [(0x01, 0), (0x02, 8), (0x04, 16), (0x08, 24)] {
                if cmd & bit != 0 {
                    let Some(&byte) = delta.get(cursor) else {
                        return Err("the delta copy instruction is truncated".to_string());
                    };
                    cursor += 1;
                    ofs |= u64::from(byte) << shift;
                }
            }
            for (bit, shift) in [(0x10, 0), (0x20, 8), (0x40, 16)] {
                if cmd & bit != 0 {
                    let Some(&byte) = delta.get(cursor) else {
                        return Err("the delta copy instruction is truncated".to_string());
                    };
                    cursor += 1;
                    size |= u64::from(byte) << shift;
                }
            }
            if size == 0 {
                size = 0x10000;
            }
            let (ofs, size) = (
                usize::try_from(ofs).map_err(|_| "the delta copy offset overflows".to_string())?,
                usize::try_from(size).map_err(|_| "the delta copy size overflows".to_string())?,
            );
            let Some(end) = ofs.checked_add(size) else {
                return Err("the delta copy range overflows".to_string());
            };
            let Some(bytes) = base.get(ofs..end) else {
                return Err("the delta copy range exceeds the base object size".to_string());
            };
            out.extend_from_slice(bytes);
        } else if cmd == 0 {
            return Err("the delta command 0 is reserved and invalid".to_string());
        } else {
            let end = cursor + usize::from(cmd);
            let Some(bytes) = delta.get(cursor..end) else {
                return Err("the delta insert data is truncated".to_string());
            };
            cursor = end;
            out.extend_from_slice(bytes);
        }
        if u64::try_from(out.len()).unwrap_or(u64::MAX) > result_size {
            return Err(format!(
                "the delta instructions produce more than the declared {result_size} bytes"
            ));
        }
    }
    if out.len() as u64 != result_size {
        return Err(format!(
            "the delta instructions produce {} bytes but the delta declares {result_size}",
            out.len()
        ));
    }
    Ok(out)
}

/// The cap-vetted, chunked inflate of one pack entry's compressed
/// stream — the unpack path's incremental reading: the produced bytes
/// accumulate under the cap, the declared length is only *checked
/// against* at the stream's end, never pre-allocated.
fn inflate_slice(compressed: &[u8], declared: u64, cap: u64) -> Result<Vec<u8>, String> {
    let mut out: Vec<u8> = Vec::new();
    let mut decompressor = gix_zlib::Decompress::new();
    let mut reader = io::Cursor::new(compressed);
    let mut chunk = vec![0u8; INFLATE_CHUNK_BYTES];
    loop {
        let written = gix_zlib::stream::inflate::read(&mut reader, &mut decompressor, &mut chunk)
            .map_err(|error| format!("the pack entry does not inflate: {error}"))?;
        out.extend_from_slice(&chunk[..written]);
        if u64::try_from(out.len()).unwrap_or(u64::MAX) > cap {
            return Err(format!(
                "the pack entry inflates past the ingestion decompressed-size cap of {cap} bytes"
            ));
        }
        if written < chunk.len() {
            break;
        }
    }
    if out.len() as u64 != declared {
        return Err(format!(
            "the pack entry inflates to {} bytes but {declared} were declared",
            out.len()
        ));
    }
    Ok(out)
}

/// The unpack-and-connectivity leg (blocking): every indexed object
/// materializes incrementally under the cap over the working-area
/// pack, re-derives, cross-checks against the index's own oid, and
/// goes through the family put inside the one [`Batch`]; then the
/// connectivity walk closes from the pushed tips. `Ok` carries the
/// batch out (the pin bridge's retained custody); every pack-level
/// failure abandons the batch (its pins drop together — sweep-eligible
/// recovery) and names the reason.
fn ingest_leg(
    front: ObjectFront,
    index_path: PathBuf,
    tips: Vec<ObjectId>,
    cap: u64,
) -> Result<Batch, LegFailure> {
    let bundle = gix_pack::Bundle::at(&index_path, gix_hash::Kind::Sha1)
        .map_err(|error| LegFailure::Pack(format!("the indexed pack does not open: {error}")))?;
    let mut batch = front.store().batch();
    let indexed_by_offset: HashMap<u64, ObjectId> = bundle
        .index
        .iter()
        .map(|entry| (entry.pack_offset, entry.oid))
        .collect();
    let mut staged: HashMap<u64, (ObjectId, Kind)> = HashMap::new();
    let mut parked: Vec<(u64, ObjectId)> = Vec::new();

    for offset in bundle.index.sorted_offsets() {
        stage_entry(
            &bundle,
            &front,
            &mut batch,
            &mut staged,
            &mut parked,
            offset,
            cap,
        )?;
    }
    while !parked.is_empty() {
        let drained = std::mem::take(&mut parked);
        let before = staged.len();
        for (offset, _base_id) in drained {
            stage_entry(
                &bundle,
                &front,
                &mut batch,
                &mut staged,
                &mut parked,
                offset,
                cap,
            )?;
        }
        if staged.len() == before {
            let (_, base_id) = parked[0];
            return Err(LegFailure::Pack(format!(
                "the delta base {base_id} does not resolve within the pack"
            )));
        }
    }

    for (offset, indexed) in &indexed_by_offset {
        match staged.get(offset) {
            Some((materialized, _)) if materialized == indexed => {}
            Some((materialized, _)) => {
                return Err(LegFailure::Pack(format!(
                    "the pack entry at offset {offset} materialized to {materialized}, \
                     not its indexed object {indexed}"
                )));
            }
            None => {
                let base = *indexed;
                return Err(LegFailure::Pack(format!(
                    "the delta base {base} does not resolve within the pack"
                )));
            }
        }
    }

    connectivity_leg(&front, &tips)?;
    Ok(batch)
}

/// Stages one pack entry: base entries inflate incrementally (the
/// declared size vetted against the cap first, the produced bytes
/// accumulated in bounded chunks); delta entries resolve their base —
/// by pack offset (ofs, always backward and staged) or by oid through
/// the index (ref, possibly forward and parked for the fixpoint) — and
/// apply the delta stream incrementally against the base read back
/// through the Find; every staged object goes through the family put.
/// `Ok` for a staged or parked entry, the ng leg's failure for a
/// resolution failure.
fn stage_entry(
    bundle: &gix_pack::Bundle,
    front: &ObjectFront,
    batch: &mut Batch,
    staged: &mut HashMap<u64, (ObjectId, Kind)>,
    parked: &mut Vec<(u64, ObjectId)>,
    offset: u64,
    cap: u64,
) -> Result<(), LegFailure> {
    let entry = bundle.pack.entry(offset).map_err(|error| {
        LegFailure::Pack(format!(
            "the pack entry at {offset} does not parse: {error}"
        ))
    })?;
    let declared = entry.decompressed_size;
    if declared > cap {
        return Err(LegFailure::Pack(format!(
            "the pack entry at {offset} declares {declared} decompressed bytes beyond \
             the ingestion decompressed-size cap of {cap} bytes"
        )));
    }
    let compressed = bundle
        .pack
        .entry_slice(entry.data_offset..bundle.pack.pack_end() as u64)
        .ok_or_else(|| {
            LegFailure::Pack(format!(
                "the pack entry at {offset} reads outside the pack data file"
            ))
        })?;

    let staged_object = match entry.header {
        PackHeader::Blob | PackHeader::Tree | PackHeader::Commit | PackHeader::Tag => {
            let kind = entry.header.as_kind().ok_or_else(|| {
                LegFailure::Pack(format!("the pack entry at {offset} carries no object kind"))
            })?;
            let body = inflate_slice(compressed, declared, cap)?;
            let oid = rederive_object(kind, &body)?;
            put_in_batch(front, batch, kind, &body, oid)?;
            Some((oid, kind))
        }
        PackHeader::OfsDelta { base_distance } => {
            let base_offset = entry
                .checked_base_pack_offset(base_distance)
                .ok_or_else(|| {
                    LegFailure::Pack(format!(
                        "the ofs delta at pack offset {offset} names an impossible base \
                         distance {base_distance}"
                    ))
                })?;
            let Some((base_oid, base_kind)) = staged.get(&base_offset).copied() else {
                return Err(LegFailure::Pack(format!(
                    "the ofs delta's base at distance {base_distance} does not resolve \
                     within the pack"
                )));
            };
            let (_, base_body) = base_object(front, base_oid, base_kind)?;
            let payload = inflate_slice(compressed, declared, cap)?;
            let body = apply_delta(&base_body, &payload, cap)?;
            let oid = rederive_object(base_kind, &body)?;
            put_in_batch(front, batch, base_kind, &body, oid)?;
            Some((oid, base_kind))
        }
        PackHeader::RefDelta { base_id } => {
            let base_offset = match bundle.index.lookup(base_id) {
                Some(base_index) => bundle.index.pack_offset_at_index(base_index),
                None => {
                    return Err(LegFailure::Pack(format!(
                        "the delta base {base_id} is neither in the pack nor in the repository"
                    )));
                }
            };
            match staged.get(&base_offset).copied() {
                Some((base_oid, base_kind)) => {
                    let (_, base_body) = base_object(front, base_oid, base_kind)?;
                    let payload = inflate_slice(compressed, declared, cap)?;
                    let body = apply_delta(&base_body, &payload, cap)?;
                    let oid = rederive_object(base_kind, &body)?;
                    put_in_batch(front, batch, base_kind, &body, oid)?;
                    Some((oid, base_kind))
                }
                None => {
                    parked.push((offset, base_id));
                    None
                }
            }
        }
    };
    if let Some((oid, kind)) = staged_object {
        staged.insert(offset, (oid, kind));
    }
    Ok(())
}

/// The full-graph connectivity walk from the pushed tips: commits walk
/// parents and trees, trees walk entries, tags walk targets, blobs
/// close the walk (the owning implementation walks — `gix-fsck`'s
/// `Connectivity::check_commit` does not walk parents). A missing
/// closure member is the ng leg's failure naming the object; a
/// corruption found on the walk is the same class.
fn connectivity_leg(front: &ObjectFront, tips: &[ObjectId]) -> Result<(), LegFailure> {
    let mut seen: HashSet<ObjectId> = HashSet::new();
    let mut stack: Vec<ObjectId> = tips.to_vec();
    while let Some(oid) = stack.pop() {
        if !seen.insert(oid) {
            continue;
        }
        let (kind, body) = {
            let found = front.block_on(front.get(&oid));
            match found {
                Ok(None) => return Err(LegFailure::Pack(format!("missing object {oid}"))),
                Err(error) => return Err(store_read_failure(&oid, error)),
                Ok(Some((kind, body))) => (kind, body),
            }
        };
        match kind {
            Kind::Commit => {
                let commit = gix_object::CommitRef::from_bytes(&body, gix_hash::Kind::Sha1)
                    .map_err(|error| LegFailure::Pack(format!("corrupt commit {oid}: {error}")))?;
                stack.push(commit.tree());
                for parent in commit.parents() {
                    stack.push(parent);
                }
            }
            Kind::Tree => {
                let tree = gix_object::TreeRef::from_bytes(&body, gix_hash::Kind::Sha1)
                    .map_err(|error| LegFailure::Pack(format!("corrupt tree {oid}: {error}")))?;
                for entry in &tree.entries {
                    stack.push(entry.oid.into());
                }
            }
            Kind::Tag => {
                let tag = gix_object::TagRef::from_bytes(&body, gix_hash::Kind::Sha1)
                    .map_err(|error| LegFailure::Pack(format!("corrupt tag {oid}: {error}")))?;
                stack.push(tag.target());
            }
            Kind::Blob => {}
        }
    }
    Ok(())
}

#[async_trait]
impl GitPackIngest for PackIngest {
    /// Unpack + index + connectivity; validates and prepares the ref
    /// updates, never applies them. Pack-level validation failures
    /// land in [`PreparedPush::unpack`]'s ng leg — parse, budget, the
    /// decompressed-size cap, thin-base and delta-base resolution,
    /// connectivity; a backing-store corruption or a
    /// working-area/task infrastructure failure is the
    /// [`StorageError`]. `updates` returns verbatim on every path. The
    /// successful legs retain the unpacked working set's [`Batch`]
    /// (the pack bridge's custody) into the pending-pin registry,
    /// content-correlated with the prepared list — the ref CAS's
    /// drain commits it after the CAS commits.
    async fn prepare(
        &self,
        repo: &RepoRecord,
        updates: Vec<RefUpdate>,
        pack: Box<dyn io::Read + Send>,
        _push_options: Option<PushOptions>,
        limits: Limits,
    ) -> Result<PreparedPush, StorageError> {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|duration| duration.as_nanos())
            .unwrap_or_default();
        let dir = self.root.join(format!(
            "ingest-{}-{}-{}",
            std::process::id(),
            nanos,
            WORK_AREA_COUNTER.fetch_add(1, Ordering::Relaxed)
        ));
        tokio::fs::create_dir_all(&dir).await.map_err(|error| {
            StorageError::Io(format!("the working area failed to create: {error}"))
        })?;
        let _area = WorkArea(dir.clone());

        let indexed = {
            let front = self.front.clone();
            let dir = dir.clone();
            let thin = front.session();
            let budget = limits.max_pack_size;
            let cap = limits.max_object_size;
            tokio::task::spawn_blocking(move || index_leg(pack, dir, thin, budget, cap))
                .await
                .map_err(|join| {
                    StorageError::Io(format!("the pack indexing task failed: {join}"))
                })?
        };

        match indexed {
            Ok(None) => {
                let front = self.front.clone();
                let tips: Vec<ObjectId> = updates.iter().filter_map(|update| update.new).collect();
                let verdict = tokio::task::spawn_blocking(move || connectivity_leg(&front, &tips))
                    .await
                    .map_err(|join| {
                        StorageError::Io(format!("the pack connectivity task failed: {join}"))
                    })?;
                match verdict {
                    Ok(()) => {
                        self.pins.retain(
                            &repo.repo_id,
                            updates.clone(),
                            PinCustody::Pins(Vec::new()),
                        );
                        Ok(PreparedPush {
                            unpack: Ok(()),
                            updates,
                        })
                    }
                    Err(LegFailure::Pack(reason)) => Ok(PreparedPush {
                        unpack: Err(reason),
                        updates,
                    }),
                    Err(LegFailure::Backing(error)) => Err(error),
                }
            }
            Ok(Some(pack_and_index)) => {
                let front = self.front.clone();
                let tips: Vec<ObjectId> = updates.iter().filter_map(|update| update.new).collect();
                let cap = limits.max_object_size;
                let leg = move || {
                    let (_pack_path, index_path) = pack_and_index;
                    ingest_leg(front, index_path, tips, cap)
                };
                let verdict = tokio::task::spawn_blocking(leg).await.map_err(|join| {
                    StorageError::Io(format!("the pack ingestion task failed: {join}"))
                })?;
                match verdict {
                    Ok(batch) => {
                        self.pins
                            .retain(&repo.repo_id, updates.clone(), PinCustody::Batch(batch));
                        Ok(PreparedPush {
                            unpack: Ok(()),
                            updates,
                        })
                    }
                    Err(LegFailure::Pack(reason)) => Ok(PreparedPush {
                        unpack: Err(reason),
                        updates,
                    }),
                    Err(LegFailure::Backing(error)) => Err(error),
                }
            }
            Err(reason) => Ok(PreparedPush {
                unpack: Err(reason),
                updates,
            }),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use crate::testutil::record as repo;
    use crate::testutil::scratch_dir as scratch_root;
    use std::path::Path;
    use std::process::Stdio;
    use std::sync::atomic::AtomicU64;
    use std::sync::Arc;

    use alkcall::core::Identity;
    use gix_hash::Kind as HashKind;
    use gix_hash::ObjectId;
    use gix_object::Kind;
    use gix_pack::data::entry::Header as PackHeader;
    use gix_pack::data::input::Entry as PackEntry;
    use tokio::net::TcpListener;
    use tokio::sync::Semaphore;

    use alkgit::{
        serve_duplex_resolved, AccessAction, GitPackGen, GitPackIngest, GitRefs, GitRegistry,
        GitRegistryStore, Limits, Service, Visibility,
    };

    use super::*;
    use crate::liveness::RefClosure;
    use crate::pins::{CommittedRefs, MemMutationLegs};
    use crate::reactor::Reactor;
    use crate::refs::InMemoryRefTable;
    use crate::registry::{
        CachedRegistryProvider, InMemoryRegistry, RegistryBackend, RegistryWrite,
    };

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

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

    fn tree_bytes(entries: &[(&str, &ObjectId)]) -> Vec<u8> {
        let mut tree = Vec::new();
        for (name, oid) in entries {
            tree.extend_from_slice(format!("100644 {name}\0").as_bytes());
            tree.extend_from_slice(oid.as_bytes());
        }
        tree
    }

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

    async fn push(
        ingest: &PackIngest,
        repo: &RepoRecord,
        updates: Vec<RefUpdate>,
        pack: Vec<u8>,
    ) -> PreparedPush {
        ingest
            .prepare(
                repo,
                updates,
                Box::new(std::io::Cursor::new(pack)),
                None,
                Limits::default(),
            )
            .await
            .expect("the pack-level session never errors here")
    }

    /// The test sink: a shared buffer behind the boxed write seam.
    #[derive(Clone, Default)]
    struct BytesSink(Arc<parking_lot::Mutex<Vec<u8>>>);

    impl BytesSink {
        fn bytes(&self) -> Vec<u8> {
            self.0.lock().clone()
        }
    }

    impl std::io::Write for BytesSink {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().extend_from_slice(buf);
            Ok(buf.len())
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    async fn generate_pack(front: &ObjectFront, wants: Vec<ObjectId>) -> Vec<u8> {
        let packgen = crate::packgen::PackGen::direct(front.clone());
        let sink = BytesSink::default();
        packgen
            .generate(
                &repo("origin"),
                wants,
                Vec::new(),
                Limits::default(),
                Box::new(sink.clone()),
            )
            .await
            .expect("the fixture pack generates");
        sink.bytes()
    }

    struct Fixture {
        blob: ObjectId,
        tree_base: ObjectId,
        tree_tip: ObjectId,
        base: ObjectId,
        tip: ObjectId,
        tag: ObjectId,
    }

    async fn build_fixture(front: &ObjectFront) -> Fixture {
        let blob = put_object(front, Kind::Blob, b"hello\n").await;
        let second = put_object(front, Kind::Blob, b"second\n").await;
        let tree_base = put_object(front, Kind::Tree, &tree_bytes(&[("hello", &blob)])).await;
        let tree_tip = put_object(
            front,
            Kind::Tree,
            &tree_bytes(&[("hello", &blob), ("second", &second)]),
        )
        .await;
        let base = put_object(front, Kind::Commit, &commit_bytes(&tree_base, &[])).await;
        let tip = put_object(front, Kind::Commit, &commit_bytes(&tree_tip, &[&base])).await;
        let tag = put_object(front, Kind::Tag, &tag_bytes(&base)).await;
        Fixture {
            blob,
            tree_base,
            tree_tip,
            base,
            tip,
            tag,
        }
    }

    fn tag_bytes(target: &ObjectId) -> Vec<u8> {
        format!(
            "object {target}\ntype commit\ntag v1.0.0\n\
             tagger A <a@example.invalid> 1750000000 +0000\n\nannotated\n"
        )
        .into_bytes()
    }

    fn deflate_bytes(bytes: &[u8]) -> Vec<u8> {
        let mut deflate =
            gix_zlib::stream::deflate::Write::new(Vec::new(), gix_zlib::Compression::DEFAULT);
        std::io::Write::write_all(&mut deflate, bytes).expect("fixture bytes deflate");
        std::io::Write::flush(&mut deflate).expect("the fixture deflate stream finishes");
        deflate.into_inner()
    }

    fn pack_entry_base(kind: Kind, body: &[u8], pack_offset: u64) -> PackEntry {
        let header = match kind {
            Kind::Blob => PackHeader::Blob,
            Kind::Tree => PackHeader::Tree,
            Kind::Commit => PackHeader::Commit,
            Kind::Tag => PackHeader::Tag,
        };
        let compressed = deflate_bytes(body);
        PackEntry {
            header,
            header_size: header.size(body.len() as u64) as u16,
            pack_offset,
            compressed: Some(compressed.clone()),
            compressed_size: compressed.len() as u64,
            crc32: None,
            decompressed_size: body.len() as u64,
            trailer: None,
        }
    }

    fn pack_entry_delta(header: PackHeader, payload: &[u8], pack_offset: u64) -> PackEntry {
        let compressed = deflate_bytes(payload);
        PackEntry {
            header,
            header_size: header.size(payload.len() as u64) as u16,
            pack_offset,
            compressed: Some(compressed.clone()),
            compressed_size: compressed.len() as u64,
            crc32: None,
            decompressed_size: payload.len() as u64,
            trailer: None,
        }
    }

    fn pack_bytes(entries: &[PackEntry]) -> Vec<u8> {
        let mut out = Vec::new();
        out.extend_from_slice(&gix_pack::data::header::encode(
            gix_pack::data::Version::V2,
            entries.len() as u32,
        ));
        for entry in entries {
            entry
                .header
                .write_to(entry.decompressed_size, &mut out)
                .expect("fixture entry headers write");
            out.extend_from_slice(
                entry
                    .compressed
                    .as_ref()
                    .expect("fixture entries keep bytes"),
            );
        }
        let mut hasher = gix_hash::hasher(HashKind::Sha1);
        hasher.update(&out);
        let digest = hasher
            .try_finalize()
            .expect("fixture pack trailers hash over the pack bytes");
        out.extend_from_slice(digest.as_slice());
        out
    }

    fn delta_size_varint(size: u64) -> Vec<u8> {
        let mut out = Vec::new();
        let mut rest = size;
        loop {
            let mut byte = rest as u8 & 0x7f;
            rest >>= 7;
            if rest != 0 {
                byte |= 0x80;
            }
            out.push(byte);
            if rest == 0 {
                return out;
            }
        }
    }

    fn delta_stream(base_size: u64, result: &[u8]) -> Vec<u8> {
        let mut out = delta_size_varint(base_size);
        out.extend(delta_size_varint(base_size + result.len() as u64));
        if base_size > 0 {
            let mut cmd = 0x80u8;
            let mut size_bytes = Vec::new();
            let mut rest = base_size;
            for bit in [0x10u8, 0x20, 0x40] {
                if rest & 0xff != 0 {
                    cmd |= bit;
                    size_bytes.push(rest as u8);
                }
                rest >>= 8;
            }
            out.push(cmd);
            out.extend_from_slice(&size_bytes);
        }
        for chunk in result.chunks(127) {
            out.push(chunk.len() as u8);
            out.extend_from_slice(chunk);
        }
        out
    }

    fn hostile_delta_stream(base_size: u64, declared_result: u64) -> Vec<u8> {
        let mut out = delta_size_varint(base_size);
        out.extend(delta_size_varint(declared_result));
        out
    }

    /// The liveness source over the (empty refs) table — the
    /// sweep-driving registration.
    fn closure_of(front: &ObjectFront, rows: &Arc<InMemoryRefTable>) -> Arc<RefClosure> {
        RefClosure::new(
            front.clone(),
            rows.clone() as Arc<dyn crate::liveness::RefRows>,
            Arc::new(AtomicU64::new(0)),
        )
    }

    /// The ordinary path end-to-end in-process: a generated pack for a
    /// fixture chain ingests into a fresh store, every object verified
    /// under its git oid, the working set's batch retained into the
    /// pending-pin registry (the custody holding across a sweep), the
    /// ref CAS drains it after committing (the legs ride), and the
    /// working area is gone on the way out.
    #[tokio::test(flavor = "multi_thread")]
    async fn an_ordinary_pack_ingests_pins_and_discards_its_area() {
        let temp = scratch_root("ingest", "ordinary");
        let origin = mem_front();
        let fx = build_fixture(&origin).await;
        let pack = generate_pack(&origin, vec![fx.tip, fx.tag]).await;
        assert!(pack.starts_with(b"PACK"));

        let front = mem_front();
        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let updates = vec![
            up_oid("refs/heads/main", None, Some(fx.tip)),
            up_oid("refs/tags/v1", None, Some(fx.tag)),
        ];
        let prepared = push(&ingest, &repo("r"), updates.clone(), pack).await;
        assert!(prepared.unpack.is_ok(), "{:?}", prepared.unpack);
        assert_eq!(prepared.updates, updates, "verbatim on the ok path");
        assert_eq!(pins.pending_count("r"), 1, "the custody retained");

        for oid in [fx.tip, fx.base, fx.tag, fx.blob, fx.tree_base, fx.tree_tip] {
            assert!(
                front.stat(&oid).await.unwrap().is_some(),
                "object {oid} landed and stats"
            );
        }

        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 report = store.sweep().await.expect("the sweep runs");
        assert_eq!(
            report.deleted, 0,
            "the retained custody holds the push's objects across the sweep window"
        );

        let durable = Arc::new(InMemoryRegistry::new());
        durable.put(&repo("r")).await.expect("the repo registers");
        let provider = Arc::new(CachedRegistryProvider::new());
        let mem_store = Arc::new(alkstore_mem::MemStore::new());
        let reactor = Arc::new(Reactor::new(
            mem_store.clone(),
            durable.clone(),
            provider.clone(),
        ));
        let legs = Arc::new(MemMutationLegs::new(durable.clone(), reactor));
        let inner: Arc<dyn GitRefs> = rows.clone();
        let wrapped = CommittedRefs::new(inner, pins.clone(), legs);
        let outcomes = wrapped
            .apply_updates(&repo("r"), updates.clone(), false)
            .await
            .expect("the CAS commits");
        assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));
        assert_eq!(pins.pending_count("r"), 0, "the entry drained post-CAS");
        let watermark = durable.watermark().await.expect("the watermark reads");
        assert_eq!(
            watermark, 2,
            "the marker write anchored the watermark (1 = the put)"
        );

        let work = temp.join("work");
        assert!(
            dir_is_empty(&work),
            "the working area is discarded (its root may remain, empty)"
        );
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// The thin push: one ref-delta entry whose base exists only in the
    /// store (the default thin-pack behavior) resolves through the
    /// substrate Find; the pack's own object lands under its oid.
    #[tokio::test]
    async fn a_thin_pack_resolves_its_bases_through_the_find() {
        let temp = scratch_root("ingest", "thin");
        let front = mem_front();
        let base_body = b"base\n".as_slice();
        let base_oid = put_object(&front, Kind::Blob, base_body).await;
        let delta = delta_stream(base_body.len() as u64, b"-delta");
        let result_body = {
            let mut body = base_body.to_vec();
            body.extend_from_slice(b"-delta");
            body
        };
        let result_oid = gix_object::compute_hash(HashKind::Sha1, Kind::Blob, &result_body)
            .expect("the fixture result hashes");
        let entry = pack_entry_delta(PackHeader::RefDelta { base_id: base_oid }, &delta, 0);
        let pack = pack_bytes(&[entry]);

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let updates = vec![up_oid("refs/heads/main", None, Some(result_oid))];
        let prepared = push(&ingest, &repo("r"), updates.clone(), pack).await;
        assert!(
            prepared.unpack.is_ok(),
            "the thin base resolves through the Find: {:?}",
            prepared.unpack
        );
        assert_eq!(prepared.updates, updates);
        assert_eq!(pins.pending_count("r"), 1, "the custody retained");
        assert!(
            front.stat(&result_oid).await.unwrap().is_some(),
            "the pack's own object landed"
        );
        assert!(
            front.stat(&base_oid).await.unwrap().is_some(),
            "the thin base's normalized entry landed through the store's dedup"
        );
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    fn dir_is_empty(path: &Path) -> bool {
        match std::fs::read_dir(path) {
            Ok(mut entries) => entries.next().is_none(),
            Err(_) => true,
        }
    }

    /// The empty pack (zero objects) is valid and reports as such: the
    /// trailer the machinery never reaches is verified by the leg
    /// itself, the push reports ok, the tips' connectivity closes over
    /// the store, and the custody retains the (empty) prepared list.
    #[tokio::test]
    async fn the_empty_pack_is_valid_and_reports_as_such() {
        let temp = scratch_root("ingest", "empty-pack");
        let front = mem_front();
        let tip = put_object(&front, Kind::Blob, b"already here\n").await;
        let pack = pack_bytes(&[]);

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let updates = vec![up_oid("refs/heads/main", None, Some(tip))];
        let prepared = push(&ingest, &repo("r"), updates.clone(), pack).await;
        assert!(prepared.unpack.is_ok(), "{:?}", prepared.unpack);
        assert_eq!(prepared.updates, updates, "verbatim on the empty path");
        assert_eq!(pins.pending_count("r"), 1, "the empty custody retained");
        assert!(dir_is_empty(&temp.join("work")), "nothing was written");
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// The zero-object pack's *trailer* is itself invalid: the leg's
    /// own verification fails it into the ng leg (the machinery is
    /// never reached), never `Err`, updates verbatim, nothing retained.
    #[tokio::test]
    async fn a_zero_pack_with_a_bad_trailer_fails_into_unpack_ng() {
        let temp = scratch_root("ingest", "empty-bad");
        let front = mem_front();
        let mut pack = Vec::new();
        pack.extend_from_slice(&gix_pack::data::header::encode(
            gix_pack::data::Version::V2,
            0,
        ));
        pack.extend_from_slice(&[7u8; 20]);

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let updates = vec![up("refs/heads/main", None, None)];
        let prepared = push(&ingest, &repo("r"), updates.clone(), pack).await;
        let reason = prepared.unpack.expect_err("the bad trailer fails the pack");
        assert!(!reason.is_empty(), "the ng reason is server-chosen");
        assert_eq!(prepared.updates, updates, "verbatim on the ng path");
        assert_eq!(pins.pending_count("r"), 0, "nothing retained on ng");
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// A pack entry whose *declared* decompressed size is hostile (a
    /// bomb-shaped declaration far above anything the compressed
    /// stream can produce) fails into the `unpack` ng leg — the
    /// doubles' hostile-declared-size row re-landed over the real
    /// machinery: the declaration never reaches an allocation, `Err`
    /// stays unreachable, nothing lands, nothing is retained.
    #[tokio::test]
    async fn a_hostile_declared_decompressed_size_fails_into_unpack_ng() {
        let temp = scratch_root("ingest", "hostile-declared");
        let front = mem_front();
        let mut hostile = pack_entry_base(Kind::Blob, b"", 0);
        hostile.decompressed_size = 1u64 << 40;
        let pack = pack_bytes(&[hostile]);

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let updates = vec![up_oid(
            "refs/heads/main",
            None,
            Some(ObjectId::from([6u8; 20])),
        )];
        let prepared = push(&ingest, &repo("r"), updates.clone(), pack).await;
        let reason = prepared
            .unpack
            .expect_err("the hostile declaration fails into the ng leg");
        assert!(
            !reason.is_empty(),
            "the ng reason is server-chosen: {reason}"
        );
        assert_eq!(prepared.updates, updates, "verbatim on the ng path");
        assert_eq!(pins.pending_count("r"), 0, "nothing retained on ng");
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// A delta stream whose declared materialized result exceeds the
    /// ingestion cap fails into the `unpack` ng leg — the applier's
    /// cap check names the decompressed-size cap before any result
    /// bytes accumulate — never `Err`, nothing lands, nothing retained.
    #[tokio::test]
    async fn a_delta_declaring_a_result_beyond_the_cap_fails_into_unpack_ng() {
        let temp = scratch_root("ingest", "delta-cap");
        let front = mem_front();
        let base_body = b"delta-base".to_vec();
        let base_entry = pack_entry_base(Kind::Blob, &base_body, 0);
        let delta_entry = pack_entry_delta(
            PackHeader::OfsDelta {
                base_distance: base_entry.bytes_in_pack(),
            },
            &hostile_delta_stream(base_body.len() as u64, 200_000),
            base_entry.bytes_in_pack(),
        );
        let pack = pack_bytes(&[base_entry, delta_entry]);
        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let limits = Limits {
            max_object_size: 65536,
            ..Limits::default()
        };
        let updates = vec![up_oid(
            "refs/heads/main",
            None,
            Some(ObjectId::from([7u8; 20])),
        )];
        let prepared = ingest
            .prepare(
                &repo("r"),
                updates.clone(),
                Box::new(std::io::Cursor::new(pack)),
                None,
                limits,
            )
            .await
            .unwrap();
        let reason = prepared
            .unpack
            .expect_err("the hostile delta result fails into the ng leg");
        assert!(
            reason.contains("fit in memory") || reason.contains("cap"),
            "the machinery's vet (or the applier's cap check) carries the reason: {reason}"
        );
        assert_eq!(prepared.updates, updates, "verbatim on the ng path");
        assert_eq!(pins.pending_count("r"), 0, "nothing retained on ng");
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// The unpack leg's own incremental inflate: a stream that
    /// *produces* beyond the cap (a lying-small declaration cannot
    /// overrun — the machinery catches the mismatch — so the row
    /// drives the applier directly) aborts at the cap naming it; the
    /// produced bytes never exceed the cap plus one chunk.
    #[tokio::test]
    async fn the_incremental_inflate_aborts_at_the_cap() {
        let big = vec![0u8; 200_000];
        let compressed = deflate_bytes(&big);
        let error = inflate_slice(&compressed, big.len() as u64, 65536)
            .expect_err("the stream's production exceeds the cap");
        assert!(error.contains("cap"), "the reason names the cap: {error}");
    }

    /// The unpack leg's own delta applier: a *lying* declaration (the
    /// declared result stays under the cap but the instructions
    /// produce more) aborts as produced-beyond-declared; a hostile
    /// declaration beyond the cap aborts on the declaration before any
    /// produced bytes.
    #[tokio::test]
    async fn the_delta_applier_vets_the_result_incrementally() {
        let base = b"0123456789".to_vec();
        let mut delta = delta_size_varint(base.len() as u64);
        delta.extend(delta_size_varint(20));
        delta.push(0x90);
        delta.push(10);
        let mut instructions = vec![15u8];
        instructions.extend_from_slice(&[b'X'; 15]);
        delta.extend_from_slice(&instructions);
        let error = apply_delta(&base, &delta, 1024).expect_err("overproducing fails");
        assert!(
            error.contains("declared"),
            "the reason names the declaration: {error}"
        );
        let error = apply_delta(
            &base,
            &hostile_delta_stream(base.len() as u64, 1 << 20),
            1024,
        )
        .expect_err("the hostile declaration fails on the cap");
        assert!(
            error.contains("cap"),
            "the reason names the decompressed-size cap: {error}"
        );
    }

    /// The in-pack *forward* ref-delta: the base later in the pack is
    /// parked in the unpack leg's fixpoint and resolves once staged.
    #[tokio::test]
    async fn an_in_pack_forward_ref_delta_resolves() {
        let temp = scratch_root("ingest", "forward-ref");
        let front = mem_front();
        let base_body = b"the later base\n".to_vec();
        let delta_result = {
            let mut body = base_body.clone();
            body.extend_from_slice(b"-delta");
            body
        };
        let delta_oid = gix_object::compute_hash(HashKind::Sha1, Kind::Blob, &delta_result)
            .expect("the fixture hashes");
        let base_oid = gix_object::compute_hash(HashKind::Sha1, Kind::Blob, &base_body)
            .expect("the fixture hashes");
        let delta_payload = delta_stream(base_body.len() as u64, b"-delta");
        let delta = pack_entry_delta(
            PackHeader::RefDelta { base_id: base_oid },
            &delta_payload,
            0,
        );
        let base = pack_entry_base(Kind::Blob, &base_body, delta.bytes_in_pack());
        let pack = pack_bytes(&[delta, base]);

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let updates = vec![up_oid("refs/heads/main", None, Some(delta_oid))];
        let prepared = push(&ingest, &repo("r"), updates.clone(), pack).await;
        assert!(
            prepared.unpack.is_ok(),
            "the forward ref-delta parks and resolves: {:?}",
            prepared.unpack
        );
        assert!(front.stat(&delta_oid).await.unwrap().is_some());
        assert!(front.stat(&base_oid).await.unwrap().is_some());
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// An ofs-delta pack resolves backward in one pass: the delta's
    /// materialized result re-derives to git's oid and lands.
    #[tokio::test]
    async fn an_ofs_delta_pack_resolves_backward() {
        let temp = scratch_root("ingest", "ofs");
        let front = mem_front();
        let base_body = b"the base body\n".to_vec();
        let result_body = {
            let mut body = base_body.clone();
            body.extend_from_slice(b"-delta");
            body
        };
        let result_oid = gix_object::compute_hash(HashKind::Sha1, Kind::Blob, &result_body)
            .expect("the fixture hashes");
        let base_entry = pack_entry_base(Kind::Blob, &base_body, 0);
        let delta = pack_entry_delta(
            PackHeader::OfsDelta {
                base_distance: base_entry.bytes_in_pack(),
            },
            &delta_stream(base_body.len() as u64, b"-delta"),
            base_entry.bytes_in_pack(),
        );
        let pack = pack_bytes(&[base_entry, delta]);

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let updates = vec![up_oid("refs/heads/main", None, Some(result_oid))];
        let prepared = push(&ingest, &repo("r"), updates.clone(), pack).await;
        assert!(
            prepared.unpack.is_ok(),
            "the ofsdelta resolves: {:?}",
            prepared.unpack
        );
        assert!(front.stat(&result_oid).await.unwrap().is_some());
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// The connectivity walk is the whole closure over the store
    /// (parents included): a commit whose tree references a ghost
    /// fails into the `unpack` ng leg naming the missing member; the
    /// abandoned batch's objects are sweep-eligible (a failed push is
    /// a failed push; a sweep reclaims them GC-decreed).
    #[tokio::test(flavor = "multi_thread")]
    async fn a_missing_closure_member_fails_the_connectivity_ng() {
        let temp = scratch_root("ingest", "connectivity");
        let front = mem_front();
        let ghost = ObjectId::from([0xE9u8; 20]);
        let commit = put_object(&front, Kind::Commit, &commit_bytes(&ghost, &[])).await;
        let pack = pack_bytes(&[pack_entry_base(Kind::Commit, &commit_bytes(&ghost, &[]), 0)]);
        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(front.clone(), pins.clone(), temp.join("work"));
        let updates = vec![up_oid("refs/heads/main", None, Some(commit))];
        let prepared = push(&ingest, &repo("r"), updates.clone(), pack).await;
        let reason = prepared
            .unpack
            .expect_err("the missing tree fails the walk");
        assert!(
            reason.contains("missing object"),
            "the reason names the missing member: {reason}"
        );
        assert_eq!(prepared.updates, updates, "verbatim on the ng path");
        assert_eq!(pins.pending_count("r"), 0, "nothing retained on ng");

        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 report = store.sweep().await.expect("the sweep runs");
        assert_eq!(report.deleted, 1, "the abandoned batch's objects recover");
        assert!(front.stat(&commit).await.unwrap().is_none());
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// Garbage, truncated, and digest-corrupt pack streams all fail
    /// into the `unpack` ng leg (server-chosen reasons, never `Err`,
    /// updates verbatim, nothing landed or retained, the working area
    /// discarded on every path).
    #[tokio::test]
    async fn malformed_packs_fail_into_unpack_ng() {
        let front = mem_front();

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(
            front.clone(),
            pins.clone(),
            scratch_root("ingest", "malformed").join("work"),
        );
        let updates = vec![up("refs/heads/main", None, Some(1))];
        let prepared = push(
            &ingest,
            &repo("r"),
            updates.clone(),
            b"garbage garbage ga".to_vec(),
        )
        .await;
        let reason = prepared.unpack.expect_err("garbage fails into the ng leg");
        assert!(!reason.is_empty(), "the ng reason is server-chosen");
        assert_eq!(prepared.updates, updates, "verbatim on the ng path");
        assert_eq!(pins.pending_count("r"), 0, "nothing retained on ng");

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(
            front.clone(),
            pins.clone(),
            scratch_root("ingest", "truncated").join("work"),
        );
        let legit = pack_bytes(&[pack_entry_base(Kind::Blob, b"truncated away\n", 0)]);
        let truncated = legit[..legit.len() - 24].to_vec();
        let prepared = push(&ingest, &repo("r"), updates.clone(), truncated).await;
        assert!(prepared.unpack.is_err(), "a truncated stream cannot pass");

        let pins = Arc::new(PendingPinRegistry::new());
        let ingest = PackIngest::new(
            front,
            pins.clone(),
            scratch_root("ingest", "digest").join("work"),
        );
        let mut digest_broken = pack_bytes(&[pack_entry_base(Kind::Blob, b"payload\n", 0)]);
        let last = digest_broken.len() - 1;
        digest_broken[last] ^= 0xff;
        let prepared = push(&ingest, &repo("r"), updates.clone(), digest_broken).await;
        assert!(prepared.unpack.is_err(), "a bad digest cannot pass");
        assert_eq!(pins.pending_count("r"), 0, "nothing retained on ng");
    }

    /// The real client drives the implement: `git push` over the
    /// git-protocol door (the test's own transport shape — doors are
    /// family infrastructure) through the core's published serve entry
    /// — the fresh push (full pack) lands; the incremental push rides
    /// the thin-pack path; the pending custody drains through the
    /// committed-ref side's CAS; the marker legs anchor the
    /// watermark. Skipped quietly when no `git` binary is installed.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_real_git_push_round_trips_through_the_implement() {
        if std::process::Command::new("git")
            .arg("--version")
            .output()
            .is_err()
        {
            eprintln!("skipping row: no `git` binary on PATH");
            return;
        }
        let temp = scratch_root("ingest", "git-push");
        std::fs::create_dir_all(&temp).expect("the scratch root creates");
        let front = mem_front();
        let pins = Arc::new(PendingPinRegistry::new());
        let ingest: Arc<dyn GitPackIngest> = Arc::new(PackIngest::new(
            front.clone(),
            pins.clone(),
            temp.join("ingest-root"),
        ));
        let durable = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let backend = Arc::new(RegistryBackend::with_cached_provider(
            durable.clone(),
            provider.clone(),
        ));
        let registry: Arc<dyn GitRegistry> = backend.clone();
        let registry_store: Arc<dyn GitRegistryStore> = backend;
        let identity = Identity {
            id: "u-push".to_owned(),
            scopes: Vec::new(),
            resources: Default::default(),
        };
        let record = RepoRecord {
            repo_id: "repo".to_owned(),
            visibility: Visibility::Private,
            grants: BTreeMap::from([(
                identity.id.clone(),
                vec![AccessAction::Read, AccessAction::Write],
            )]),
        };
        registry_store.put_repo(&record).await.unwrap();

        let rows = Arc::new(InMemoryRefTable::new());
        let mem_store = Arc::new(alkstore_mem::MemStore::new());
        let reactor = Arc::new(Reactor::new(
            mem_store.clone(),
            durable.clone(),
            provider.clone(),
        ));
        let legs = Arc::new(MemMutationLegs::new(durable.clone(), reactor));
        let refs: Arc<dyn GitRefs> = Arc::new(CommittedRefs::new(rows.clone(), pins.clone(), legs));

        let packgen: Arc<dyn GitPackGen> = Arc::new(crate::packgen::PackGen::direct(front.clone()));
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let limits = Limits::default();
        let pipeline = Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks));
        let push_ref = refs.clone();
        let push_registry = registry.clone();
        let push_gen = packgen.clone();
        let push_ingest = ingest.clone();
        let ident = identity.clone();
        let door = tokio::spawn(async move {
            loop {
                let Ok((mut conn, _peer)) = listener.accept().await else {
                    break;
                };
                let refs = push_ref.clone();
                let registry = push_registry.clone();
                let gen = push_gen.clone();
                let ingest = push_ingest.clone();
                let identity = ident.clone();
                let pipeline = pipeline.clone();
                tokio::spawn(async move {
                    let Some(frame) = read_daemon_frame(&mut conn).await else {
                        return;
                    };
                    let Some(repo_id) = parse_daemon_frame(&frame) else {
                        return;
                    };
                    let _ = serve_duplex_resolved(
                        &repo_id,
                        Some(&identity),
                        Service::ReceivePack,
                        conn,
                        limits,
                        registry,
                        refs,
                        gen,
                        ingest,
                        false,
                        pipeline,
                    )
                    .await;
                });
            }
        });

        let work = temp.join("work");
        run_git(
            &temp,
            &["init", "-q", "-b", "main", work.to_string_lossy().as_ref()],
        )
        .await;
        std::fs::write(work.join("a.txt"), "alpha\n").unwrap();
        run_git(&work, &["add", "."]).await;
        run_git(
            &work,
            &[
                "-c",
                "user.name=A",
                "-c",
                "user.email=a@example.invalid",
                "commit",
                "-q",
                "-m",
                "one",
            ],
        )
        .await;
        let url = format!("git://127.0.0.1:{port}/repo");
        run_git(&work, &["push", "-q", &url, "main"]).await;

        let tip1 = run_git(&work, &["rev-parse", "HEAD"]).await;
        let tree1 = run_git(&work, &["rev-parse", "HEAD^{tree}"]).await;
        let blob1 = run_git(&work, &["rev-parse", "HEAD:a.txt"]).await;
        let parse = |hex_text: &String| {
            ObjectId::from_hex(hex_text.trim().as_bytes()).expect("the git output is hex")
        };
        let tip1 = parse(&tip1);
        let tree1 = parse(&tree1);
        let blob1 = parse(&blob1);
        let main = rows
            .list_refs(&record)
            .await
            .unwrap()
            .into_iter()
            .find(|line| line.name == "refs/heads/main")
            .expect("the ref row landed");
        assert_eq!(main.oid, Some(tip1), "the push's tip is the ref row");
        for (oid, kind) in [
            (tip1, Kind::Commit),
            (tree1, Kind::Tree),
            (blob1, Kind::Blob),
        ] {
            assert_eq!(
                front.kind(&oid).await.unwrap(),
                Some(kind),
                "the pushed object {oid} verified into the store"
            );
        }
        assert_eq!(
            pins.pending_count("repo"),
            0,
            "the custody drained through the committed-ref side's CAS"
        );
        let watermark_one = durable.watermark().await.unwrap();

        std::fs::write(work.join("b.txt"), "beta\n").unwrap();
        run_git(&work, &["add", "."]).await;
        run_git(
            &work,
            &[
                "-c",
                "user.name=A",
                "-c",
                "user.email=a@example.invalid",
                "commit",
                "-q",
                "-m",
                "two",
            ],
        )
        .await;
        run_git(&work, &["push", "-q", &url, "main"]).await;
        let tip2 = run_git(&work, &["rev-parse", "HEAD"]).await;
        let tip2 = parse(&tip2);
        let main = rows
            .list_refs(&record)
            .await
            .unwrap()
            .into_iter()
            .find(|line| line.name == "refs/heads/main")
            .expect("the ref row");
        assert_eq!(
            main.oid,
            Some(tip2),
            "the incremental push advanced the row"
        );
        assert_eq!(front.kind(&tip2).await.unwrap(), Some(Kind::Commit));
        assert_eq!(
            pins.pending_count("repo"),
            0,
            "the second push's custody drained"
        );
        let watermark_two = durable.watermark().await.unwrap();
        assert_eq!(
            watermark_two,
            watermark_one + 1,
            "each committed push rode its marker leg"
        );

        door.abort();
        let _ = tokio::fs::remove_dir_all(&temp).await;
    }

    /// The pkt-line's read (bounded); `None` when the peer closed.
    async fn read_daemon_frame(conn: &mut tokio::net::TcpStream) -> Option<Vec<u8>> {
        let mut header = [0u8; 4];
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            tokio::io::AsyncReadExt::read_exact(conn, &mut header).await
        })
        .await
        .ok()?
        .ok()?;
        let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
            .expect("hex length");
        if !(4..=(1 << 16)).contains(&len) {
            return None;
        }
        let mut payload = vec![0u8; len - 4];
        tokio::io::AsyncReadExt::read_exact(conn, &mut payload)
            .await
            .ok()?;
        Some(payload)
    }

    /// The daemon frame's request line: `<service> <repo>` NUL-attached
    /// `host=…` — only the receive-pack service is served here (the
    /// row's shape).
    fn parse_daemon_frame(frame: &[u8]) -> Option<String> {
        let text = String::from_utf8(frame.to_vec()).ok()?;
        let mut segments = text.split('\0');
        let (service, repo_path) = segments.next()?.split_once(' ')?;
        if service != "git-receive-pack" {
            return None;
        }
        let repo_id = repo_path.strip_prefix('/').unwrap_or(repo_path);
        if repo_id.is_empty() {
            return None;
        }
        Some(repo_id.to_owned())
    }

    /// One isolated real-git command; its stdout (the rows assert
    /// exit status themselves).
    async fn run_git(dir: &Path, args: &[&str]) -> String {
        let mut command = tokio::process::Command::new("git");
        command
            .arg("-C")
            .arg(dir)
            .args(args)
            .env_clear()
            .env("PATH", std::env::var("PATH").unwrap_or_default())
            .env("GIT_CONFIG_NOSYSTEM", "1")
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_TERMINAL_PROMPT", "0")
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        let output = command.output().await.expect("the git command spawns");
        assert!(
            output.status.success(),
            "git {args:?} failed: {}{}",
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
        String::from_utf8_lossy(&output.stdout).into_owned()
    }
}