alkgit-substrate 0.1.0

//! The substrate's `GitPackGen` implement ([`PackGen`]) — pack
//! generation over the real Find front ([`ObjectFront`]'s per-session
//! [`SessionFind`] views) and the negotiation ack seam's real predicate
//! (the trait's `common_haves`, ADR-014 §5) — substrate.md §object
//! storage "Packs"; ADR-004 is authoritative for the pipeline.
//!
//! # The pipeline (ADR-004, composed exactly)
//!
//! `run_pipeline` (the module's pipeline function) is the wave-2
//! doubles' composition re-landed on the substrate Find (the POC-O
//! `packgen.rs` reference shape):
//!
//! ```text
//! peel wants to commit tips (tag chains walked; peeled tag objects kept)
//!   boundary_haves peeled identically (the closure-subtraction sets)
//!   → commit-ancestry walk (gix_traverse::commit::Simple, Parents::All,
//!     non-commit tips as-is)
//!   → count::objects_unthreaded (TreeContents) over the closure inputs
//!   [+ AsIs over the peel-exposed tag chains]
//!   → counted-object existence pre-check
//!   → entry::iter_from_counts → InOrderIter
//!   → bytes::FromEntriesIter (V2, sha1) → the sink
//! ```
//!
//! **The two-stage closure is mandatory** (ADR-004: TreeContents does
//! not follow commit parents; feeding tip commits alone produces
//! incomplete packs — POC-2's course correction). Missing objects abort
//! with [`StorageError::ObjectMissing`] naming the missing object
//! (never a broken pack emitted downstream): every counted object is
//! existence-checked before the byte pipeline opens, and the entry
//! stage's own `missing_objects` statistics remain the abort posture.
//!
//! Generation is negotiation-agnostic: boundary sets in, pack out. The
//! `limits` parameter is accepted and unused for v1 — the doubles' own
//! posture; ADR-009 budgets the receive side, and ADR-004 pins no
//! generation-side size budget (`max_pack_size` is
//! [`GitPackIngest`](alkgit::GitPackIngest) territory).
//!
//! # Both serving shapes (substrate.md §object storage "Packs")
//!
//! The mode selection is the substrate impl's internal posture — the
//! constructor level (the trade: a per-call selection would surface a
//! mode parameter the frozen trait signature does not carry).
//!
//! - *Direct* ([`PackGen::direct`]): the generated pack streams through
//!   the sink, no storage round-trip.
//! - *Through-store* ([`PackGen::through_store`]): the pack is put once
//!   into the store (the pack rides the canonical blob family —
//!   `PACK_ALGORITHM` below — with the store's own tier dispatch
//!   routing big packs to the large tier), then served through
//!   `read_range` windows into the sink — every window's
//!   out-of-band slice digest is re-derived and verified mid-flight
//!   (the store does not verify a range read; the front-line impl
//!   re-derives, the same posture [`ObjectFront`]'s loose reads carry).
//!   A request-digest [`PackCache`] lets repeat fetches ride the stored
//!   pack without regeneration; releases and evictions are pins the
//!   next sweep reclaims ([`PackCache::release_all`] is the
//!   `PackHousekeeping` consumer's release path).
//!   *Large tier required for through-store serving* — a kv-only shape
//!   rejects over-threshold packs, the rejection riding the store's own
//!   put-time kv-only overflow (mapped to [`StorageError::Io`]).
//!
//! # Threading and the one bridge
//!
//! The ADR-004 pipeline 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 inside the pipeline goes through [`ObjectFront::block_on`] — the
//! odb-front task's one-place bridge rule ([`SessionFind`]'s sync legs
//! already own theirs). The through-store put rides the async path
//! directly (generate's own async context): the pipeline streams the
//! pack through a bounded duplex pipe (`PACK_PIPE_CAPACITY`, the
//! backpressure boundary) into `Store::put_as` — no unbounded buffer on
//! the async path.
//!
//! # Failure posture on the pack put
//!
//! A generation that fails *after* the put began leaves the pack put's
//! fate to the stream: an early pipeline end closes the pipe and the
//! put commits what it received — a truncated or empty orphan pack that
//! holds its pin only across the call. The error is returned and any
//! put-returned pin is dropped, so the orphan is sweep-eligible: a
//! failed fetch is a failed fetch, not partial repo state (the
//! substrate's GC-decreed recovery shape). Nothing partial is ever
//! *served*: the pack never reaches the sink on an error path, and a
//! cached pack is inserted only once both the pipeline and the put
//! succeeded. The put error wins over the pipeline's pipe-closed noise
//! — the honest cause (a kv-only over-threshold rejection, a backing
//! failure) is what the caller sees.

use std::collections::{HashMap, HashSet, VecDeque};
use std::io;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use alkblobs::{Algorithm, Key, Pin};
use gix_hash::ObjectId;
use gix_object::Kind;
use gix_pack::data::output::bytes::FromEntriesIter;
use gix_pack::data::output::count;
use gix_pack::data::output::entry::iter_from_counts;
use gix_pack::data::output::entry::iter_from_counts::{Mode, Options, Outcome};
use gix_traverse::commit::{Parents, Simple};
use sha2::Digest as _;

use alkgit::{GitPackGen, Limits, RepoRecord, StorageError};
use async_trait::async_trait;

use crate::odb::{store_leg_error, ObjectFront, SessionFind};

/// The derivation family the through-store pack rides: the canonical
/// blob family's sha256 member (an ordinary canonical-key blob — the
/// pack is cache, not a git object; git objects address through the
/// git-object family alone, and the pack key is the store's own
/// content derivation, never an oid).
pub(crate) const PACK_ALGORITHM: Algorithm = Algorithm::GitBlobSha256;

/// The bounded pipe between the blocking pipeline and the async pack
/// put (`tokio::io::duplex`) — the through-store path's backpressure
/// boundary (no unbounded buffer: a store slower than the pipeline
/// throttles generation to the pipe width).
const PACK_PIPE_CAPACITY: usize = 256 * 1024;

/// The through-store serving window (`read_range` slice, 1 MiB — the
/// POC-O datum). Each window's out-of-band SHA-256 slice digest is
/// re-derived and compared mid-flight.
const SERVE_WINDOW: u64 = 1024 * 1024;

/// The default request-digest cache capacity (the POC-O datum: a small
/// bound — the pack cache holds at most this many generated packs).
pub(crate) const DEFAULT_CACHE_ENTRIES: usize = 4;

/// The request digest of a fetch pack request: SHA-256 over the
/// canonical (sorted, tagged) want and have lists — two requests with
/// the same wants-and-recognized-haves set are the same request, so
/// repeat fetches of the same shape ride the stored pack. POC-O's
/// recipe verbatim.
fn request_digest(wants: &[ObjectId], boundary_haves: &[ObjectId]) -> [u8; 32] {
    let mut digest = sha2::Sha256::new();
    let mut wants = wants.to_vec();
    wants.sort_unstable();
    for want in wants {
        digest.update(b"W");
        digest.update(want.as_bytes());
    }
    let mut haves = boundary_haves.to_vec();
    haves.sort_unstable();
    for have in haves {
        digest.update(b"H");
        digest.update(have.as_bytes());
    }
    digest.finalize().into()
}

/// The peel result of one want or have: the tag objects walked through
/// (the chain's closure members) and the terminal object's id.
struct PeeledTip {
    chain: Vec<ObjectId>,
    terminal: ObjectId,
    terminal_is_commit: bool,
}

/// Peels `tip` through its tag chain (ADR-004: tips must be peeled
/// before the commit-ancestry walk), returning the chain and the
/// terminal kind. Errors on the first missing object of the chain with
/// [`StorageError::ObjectMissing`] — a tip that cannot be resolved is
/// the abort, never a partial pack.
fn peel_tip(find: &SessionFind, tip: ObjectId) -> Result<PeeledTip, StorageError> {
    let mut chain = Vec::new();
    let mut current = tip;
    loop {
        let mut buf = Vec::new();
        let data = match gix_object::FindExt::find(find, &current, &mut buf) {
            Ok(data) => data,
            Err(error) => return Err(from_pipeline_error(find, error, tip)),
        };
        let decoded = data.decode();
        let next = match decoded {
            Ok(gix_object::ObjectRef::Tag(tag)) => {
                chain.push(current);
                tag.target()
            }
            Ok(gix_object::ObjectRef::Commit(_)) => {
                return Ok(PeeledTip {
                    chain,
                    terminal: current,
                    terminal_is_commit: true,
                });
            }
            Ok(_) => {
                return Ok(PeeledTip {
                    chain,
                    terminal: current,
                    terminal_is_commit: false,
                });
            }
            Err(error) => return Err(StorageError::Io(error.to_string())),
        };
        current = next;
    }
}

/// The missing-object abort (ADR-004): the recorded miss names the
/// missing object, falling back to the id the stage was working on.
fn pipeline_missing(find: &SessionFind, fallback: ObjectId) -> StorageError {
    StorageError::ObjectMissing {
        oid: find.missed_object().unwrap_or(fallback),
    }
}

/// Maps a pipeline error to the seam's error model: a not-found is the
/// missing-object abort (ADR-004), anything else is a backing-store
/// failure (ADR-018 §7).
fn from_pipeline_error(
    find: &SessionFind,
    error: gix_error::Error,
    fallback: ObjectId,
) -> StorageError {
    if error.is_not_found() {
        pipeline_missing(find, fallback)
    } else {
        StorageError::Io(error.to_string())
    }
}

/// The ADR-004 generation pipeline over one session's Find — the
/// doubles' wave-2 composition re-landed on the substrate Find front.
/// `W` is any blocking sink (the wire's sideband bridge on the direct
/// shape; the pack-put pipe on the through-store one). The two-stage
/// closure, the missing-object aborts and the entry-statistics check
/// are ADR-004's pinned shape; O(counts) memory throughout.
fn run_pipeline<W: io::Write>(
    find: &SessionFind,
    wants: Vec<ObjectId>,
    boundary_haves: Vec<ObjectId>,
    sink: &mut W,
) -> Result<(), StorageError> {
    let no_want = ObjectId::null(gix_hash::Kind::Sha1);
    let first_want = wants.first().copied().unwrap_or(no_want);

    let mut boundary_commits = HashSet::new();
    let mut boundary_objects = HashSet::new();
    for have in boundary_haves {
        let peeled = peel_tip(find, have)?;
        boundary_objects.extend(peeled.chain.iter().copied());
        boundary_objects.insert(peeled.terminal);
        if peeled.terminal_is_commit {
            boundary_commits.insert(peeled.terminal);
        }
    }

    let mut commit_tips = Vec::new();
    let mut non_commit_tips = Vec::new();
    let mut tag_chains = Vec::new();
    for want in wants {
        let peeled = peel_tip(find, want)?;
        for chain_oid in peeled.chain {
            if !boundary_objects.contains(&chain_oid) {
                tag_chains.push(chain_oid);
            }
        }
        if peeled.terminal_is_commit {
            commit_tips.push(peeled.terminal);
        } else if !boundary_objects.contains(&peeled.terminal) {
            non_commit_tips.push(peeled.terminal);
        }
    }

    let mut walk = Simple::filtered(commit_tips, find.clone(), |tip: &gix_hash::oid| {
        !boundary_commits.contains(tip)
    })
    .parents(Parents::All);
    let mut survivors = Vec::new();
    for info in walk.by_ref() {
        let info = info.map_err(|error| from_pipeline_error(find, error, first_want))?;
        survivors.push(info.id);
    }

    let counter = gix_utils::progress::Discard;
    let interrupt = AtomicBool::new(false);
    let mut closure_inputs = survivors;
    closure_inputs.extend(non_commit_tips);
    let (mut counts, _count_outcome) = count::objects_unthreaded(
        find,
        &mut closure_inputs
            .into_iter()
            .map(Ok::<ObjectId, gix_error::Error>),
        &counter,
        &interrupt,
        count::objects::ObjectExpansion::TreeContents,
    )
    .map_err(|error| from_pipeline_error(find, error, first_want))?;
    let (tag_counts, _tag_outcome) = count::objects_unthreaded(
        find,
        &mut tag_chains.into_iter().map(Ok::<ObjectId, gix_error::Error>),
        &counter,
        &interrupt,
        count::objects::ObjectExpansion::AsIs,
    )
    .map_err(|error| from_pipeline_error(find, error, first_want))?;
    let mut counted: HashSet<ObjectId> = counts.iter().map(|count| count.id).collect();
    for tag_count in tag_counts {
        if counted.contains(&tag_count.id) {
            continue;
        }
        counted.insert(tag_count.id);
        counts.push(tag_count);
    }

    for count in &counts {
        if !gix_pack::Find::contains(find, &count.id) {
            return Err(StorageError::ObjectMissing { oid: count.id });
        }
    }

    let counts_len = counts.len();
    let mut entries = gix_parallel::InOrderIter::from(
        iter_from_counts(
            counts,
            find.clone(),
            Box::new(gix_utils::progress::Discard),
            Options {
                thread_limit: Some(1),
                mode: Mode::PackCopyAndBaseObjects,
                allow_thin_pack: false,
                chunk_size: 10,
                version: gix_pack::data::Version::V2,
                compression: gix_zlib::Compression::DEFAULT,
            },
        )
        .map_err(|error| from_pipeline_error(find, error, first_want))?,
    );
    let mut from_entries = FromEntriesIter::new(
        entries.by_ref(),
        sink,
        u32::try_from(counts_len)
            .map_err(|_| StorageError::Invalid("the pack's object count exceeds u32".to_owned()))?,
        gix_pack::data::Version::V2,
        gix_hash::Kind::Sha1,
    );
    for step in from_entries.by_ref() {
        step.map_err(|error| from_pipeline_error(find, error, first_want))?;
    }
    if from_entries.digest().is_none() {
        return Err(StorageError::Io(
            "the pack trailer was never written".to_string(),
        ));
    }
    let entry_statistics: Outcome = {
        use gix_parallel::Finalize as _;
        entries
            .inner
            .finalize()
            .map_err(|error| from_pipeline_error(find, error, first_want))?
    };
    if entry_statistics.missing_objects > 0 {
        let missing = find
            .missed_object()
            .unwrap_or_else(|| ObjectId::null(gix_hash::Kind::Sha1));
        return Err(StorageError::ObjectMissing { oid: missing });
    }
    Ok(())
}

/// The blocking pipeline's through-store end: a sync writer over the
/// pack-put pipe's write half. Every write bridges through
/// [`ObjectFront::block_on`] — the one-place rule (this sink exists
/// only inside the `spawn_blocking` pipeline task; in an async context
/// the bridge is the POC-O panic shape and never taken).
struct PipeSink {
    stream: tokio::io::DuplexStream,
    front: ObjectFront,
}

impl io::Write for PipeSink {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        use tokio::io::AsyncWriteExt;
        self.front
            .block_on(AsyncWriteExt::write(&mut self.stream, buf))
    }

    fn flush(&mut self) -> io::Result<()> {
        use tokio::io::AsyncWriteExt;
        self.front.block_on(AsyncWriteExt::flush(&mut self.stream))
    }
}

/// The pack cache's entry: the stored pack's canonical key, the pin
/// holding it against the sweep for exactly the cache's custody span,
/// and the byte length the windows serve against.
struct CachedPack {
    key: Key,
    _pin: Pin,
    len: u64,
}

/// The request-digest pack cache of the through-store shape: repeat
/// fetches of the same (canonicalized) want/have set ride the stored
/// pack without regeneration. Evictions (capacity bounded) and
/// releases ([`PackCache::release_all`]) drop the entry's pins — the
/// sweep reclaims the pack blobs the next time it runs (substrate.md
/// §object storage: "releases/evictions are pins the next sweep
/// reclaims").
pub struct PackCache {
    max_entries: usize,
    entries: parking_lot::Mutex<HashMap<[u8; 32], CachedPack>>,
    order: parking_lot::Mutex<VecDeque<[u8; 32]>>,
}

impl PackCache {
    /// The cache holding at most `max_entries` generated packs (a `0`
    /// clamps to 1 — a zero-capacity cache would evict on every insert
    /// but must not loop). The store handle stays with the
    /// [`PackGen`] front; the cache is the digest-to-pack mapping.
    pub fn new(max_entries: usize) -> Self {
        PackCache {
            max_entries: max_entries.max(1),
            entries: parking_lot::Mutex::new(HashMap::new()),
            order: parking_lot::Mutex::new(VecDeque::new()),
        }
    }

    /// The cached pack behind `digest`, if any: the pack's key (the
    /// windows serve against it through the store) and its length.
    fn lookup(&self, digest: [u8; 32]) -> Option<(Key, u64)> {
        self.entries
            .lock()
            .get(&digest)
            .map(|cached| (cached.key.clone(), cached.len))
    }

    /// Caches the just-generated pack. When the cache is at capacity
    /// the oldest digest's entry (and the pin protecting it) drops
    /// first — the eviction's pack becomes sweep-eligible.
    fn insert(&self, digest: [u8; 32], key: Key, len: u64, pin: Pin) {
        let mut entries = self.entries.lock();
        let mut order = self.order.lock();
        while entries.len() >= self.max_entries {
            match order.pop_front() {
                Some(evict) => {
                    entries.remove(&evict);
                }
                None => break,
            }
        }
        entries.insert(
            digest,
            CachedPack {
                key,
                _pin: pin,
                len,
            },
        );
        order.push_back(digest);
    }

    /// The cache's release path — the `PackHousekeeping` consumer's
    /// action: drop every cached entry (their pins release) and return
    /// the released count. The released packs are reclaimed by the next
    /// sweep.
    pub fn release_all(&self) -> usize {
        let mut entries = self.entries.lock();
        let mut order = self.order.lock();
        order.clear();
        let released = entries.len();
        entries.clear();
        released
    }
}

/// The through-store shape's serving mode enum for [`PackGen`].
enum Serving {
    /// The generated pack streams through the sink; no storage
    /// round-trip, no cache.
    Direct,
    /// The pack is put once, then served ranged from the store through
    /// the request-digest cache — shared with the composition's
    /// maintenance chain, whose `PackHousekeeping` job runs the
    /// cache's release path (`Arc`: one cache, two holders).
    ThroughStore(Arc<PackCache>),
}

/// The substrate's [`GitPackGen`] implement: ADR-004's pipeline over
/// the object front's real Find, the ADR-014 §5 ack predicate, and one
/// of the two serving shapes (constructor-level posture — through the
/// frozen trait signature runs no mode parameter; the wire layer is
/// unchanged either way).
///
/// The object pool is global, not per-repo-namespaced (the wave-7
/// recorded interpretation): the `&RepoRecord` parameter is accepted
/// and unused here — the Find front reads the git-object family
/// directly, per-repo coverage is liveness-decided.
pub struct PackGen {
    front: ObjectFront,
    serving: Serving,
}

impl PackGen {
    /// The direct shape: the generated pack streams through the sink,
    /// no storage round-trip.
    pub fn direct(front: ObjectFront) -> Self {
        PackGen {
            front,
            serving: Serving::Direct,
        }
    }

    /// The through-store shape with the default request-digest cache
    /// (4 entries).
    ///
    /// The fresh private cache is a deliberate per-piece probe surface
    /// (the documented-`durable()` posture, review 006 R-6's second
    /// instance): the cache's pins are reachable by no maintenance
    /// chain, so its release path is eviction alone and served packs
    /// go reclaim-only at sweep — an honest probe shape, not a
    /// production composition. The production through-store serving
    /// shape is the composition entry's ([`crate::compose`]'s
    /// `PackServing::ThroughStore`, the cache shared with the chain's
    /// housekeeping), and a hand-wired production embedder passes the
    /// chain-held handle explicitly ([`PackGen::through_store_with`]).
    pub fn through_store(front: ObjectFront) -> Self {
        PackGen::through_store_with(front, Arc::new(PackCache::new(DEFAULT_CACHE_ENTRIES)))
    }

    /// The through-store shape with an explicit cache — shared
    /// (`Arc`): the same handle the composition's maintenance chain
    /// holds, whose `PackHousekeeping` job runs
    /// [`PackCache::release_all`].
    pub fn through_store_with(front: ObjectFront, cache: Arc<PackCache>) -> Self {
        PackGen {
            front,
            serving: Serving::ThroughStore(cache),
        }
    }

    /// One through-store generation: the pipeline streams the pack
    /// through the bounded pipe into the store's put (tier dispatch
    /// runs on the store's side), and the put-returned pin comes back
    /// for the cache to hold. On any failure the pin drops (an orphan
    /// pack is sweep-eligible) and the error surfaces.
    async fn store_pack(
        &self,
        find: SessionFind,
        wants: Vec<ObjectId>,
        boundary_haves: Vec<ObjectId>,
    ) -> Result<(Key, Pin, u64), StorageError> {
        let store = Arc::clone(self.front.store());
        let (pipe_write, pipe_read) = tokio::io::duplex(PACK_PIPE_CAPACITY);
        let sink = PipeSink {
            stream: pipe_write,
            front: self.front.clone(),
        };
        let pipeline = tokio::task::spawn_blocking(move || {
            let mut sink = sink;
            run_pipeline(&find, wants, boundary_haves, &mut sink)
        });
        let put = store.put_as(PACK_ALGORITHM, None, pipe_read);
        let put = put.await;
        let piped = pipeline
            .await
            .map_err(|join| StorageError::Io(format!("the pack generation task failed: {join}")))?;
        match (put, piped) {
            (Ok((key, pin)), Ok(())) => {
                let meta = store
                    .stat(&key.as_bytes())
                    .await
                    .map_err(|e| store_leg_error("alkblobs stat", e))?
                    .ok_or_else(|| {
                        StorageError::Invalid(
                            "the pack put returned a key the store no longer stats".to_owned(),
                        )
                    })?;
                Ok((key, pin, meta.len))
            }
            (Err(put_error), _) => Err(store_leg_error("the pack put", put_error)),
            (Ok((_key, pin)), Err(pipeline_error)) => {
                drop(pin);
                Err(pipeline_error)
            }
        }
    }

    /// Serves the stored pack's bytes through `read_range` windows into
    /// the sink, every window's out-of-band slice digest re-derived and
    /// verified mid-flight. One spawn_blocking task owns the sink; the
    /// store reads bridge through [`ObjectFront::block_on`].
    async fn serve_stored(
        &self,
        key: &Key,
        len: u64,
        sink: Box<dyn io::Write + Send>,
    ) -> Result<(), StorageError> {
        let front = self.front.clone();
        let key_bytes = key.as_bytes().to_vec();
        let store = Arc::clone(self.front.store());
        let serving = tokio::task::spawn_blocking(move || -> Result<(), StorageError> {
            let mut sink = sink;
            let mut offset = 0u64;
            while offset < len {
                let window = SERVE_WINDOW.min(len - offset);
                let (slice, digest) = front
                    .block_on(store.read_range(&key_bytes, offset, window))
                    .map_err(|e| store_leg_error("alkblobs read_range", e))?;
                let derived: [u8; 32] = sha2::Sha256::digest(&slice).into();
                if derived != digest {
                    return Err(StorageError::Invalid(format!(
                        "the stored pack's window at {offset} fails its slice digest (mid-flight corruption)"
                    )));
                }
                sink.write_all(&slice)
                    .map_err(|e| StorageError::Io(format!("the pack sink failed: {e}")))?;
                offset += window;
            }
            sink.flush()
                .map_err(|e| StorageError::Io(format!("the pack sink failed: {e}")))
        });
        serving
            .await
            .map_err(|join| StorageError::Io(format!("the pack serving task failed: {join}")))?
    }
}

#[async_trait]
impl GitPackGen for PackGen {
    /// closure(wants) − closure(boundary_haves) → pack streamed into
    /// the sink (ADR-004). Missing objects abort with
    /// [`StorageError::ObjectMissing`] — never a broken pack on the
    /// sink.
    async fn generate(
        &self,
        _repo: &RepoRecord,
        wants: Vec<ObjectId>,
        boundary_haves: Vec<ObjectId>,
        _limits: Limits,
        sink: Box<dyn io::Write + Send>,
    ) -> Result<(), StorageError> {
        match &self.serving {
            Serving::Direct => {
                let front = self.front.clone();
                let direct = tokio::task::spawn_blocking(move || {
                    let find = front.session();
                    let mut sink = sink;
                    run_pipeline(&find, wants, boundary_haves, &mut sink)?;
                    sink.flush()
                        .map_err(|e| StorageError::Io(format!("the pack sink failed: {e}")))
                });
                direct.await.map_err(|join| {
                    StorageError::Io(format!("the pack generation task failed: {join}"))
                })?
            }
            Serving::ThroughStore(cache) => {
                let digest = request_digest(&wants, &boundary_haves);
                if let Some((key, len)) = cache.lookup(digest) {
                    return self.serve_stored(&key, len, sink).await;
                }
                let find = self.front.session();
                let (key, pin, len) = self.store_pack(find, wants, boundary_haves).await?;
                cache.insert(digest, key.clone(), len, pin);
                self.serve_stored(&key, len, sink).await
            }
        }
    }

    /// The negotiation ack seam's real predicate (ADR-014 §5, the same
    /// predicate on the ack rounds and the done-round boundary alike): a
    /// have is recognized iff it **exists in the object store and is a
    /// commit** — the is-commit half over the loose header the odb layer
    /// owns ([`ObjectFront::kind`], the bounded header peek — the
    /// predicate never reads a have's body, so a hostile have set cannot
    /// walk the store's object contents). A missing have is
    /// `kind`'s `None` (unrecognized — subtracts nothing); a non-sha1
    /// have is `StorageError::Invalid` (OQ-05). One seam predicate, no
    /// wire-side refinement (ADR-010). The caller's order is preserved
    /// verbatim.
    async fn common_haves(
        &self,
        _repo: &RepoRecord,
        haves: Vec<ObjectId>,
    ) -> Result<Vec<ObjectId>, StorageError> {
        let mut recognized = Vec::new();
        for oid in haves {
            if self.front.kind(&oid).await? == Some(Kind::Commit) {
                recognized.push(oid);
            }
        }
        Ok(recognized)
    }
}

#[cfg(test)]
mod tests {
    use alkblobs::Store;
    use alkgit::{GitPackGen as _, Limits, RepoRecord, StorageError, Visibility};
    use gix_hash::ObjectId;
    use gix_object::Kind;
    use std::io::{self, Write};
    use std::sync::Arc;

    use super::{request_digest, run_pipeline, PackCache, PackGen};
    use crate::odb::{key_to_oid, ObjectFront};
    use crate::testutil::scratch_dir as scratch_root;

    fn repo() -> RepoRecord {
        RepoRecord {
            repo_id: "substrate/packgen-tests".to_owned(),
            visibility: Visibility::Public,
            grants: Default::default(),
        }
    }

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

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

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

    /// The wave-7 fixture, built into `front`: two commits with distinct
    /// root trees, an annotated tag on the base commit (the tag-chain
    /// path's member), and a stray blob (the non-commit want and the
    /// unrecognized-have member).
    struct Fixture {
        base: ObjectId,
        tip: ObjectId,
        tag: ObjectId,
        stray: ObjectId,
    }

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

    /// The test sink: a shared buffer behind the seam's `Box<dyn
    /// io::Write + Send>` (the bytes come back out through the clone).
    #[derive(Clone, Default)]
    struct BytesSink(Arc<parking_lot::Mutex<Vec<u8>>>);

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

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

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

    async fn generate(
        packgen: &PackGen,
        wants: Vec<ObjectId>,
        boundary: Vec<ObjectId>,
    ) -> Result<Vec<u8>, StorageError> {
        let sink = BytesSink::default();
        packgen
            .generate(
                &repo(),
                wants,
                boundary,
                Limits::default(),
                Box::new(sink.clone()),
            )
            .await?;
        Ok(sink.bytes())
    }

    fn cached_count(packgen: &PackGen) -> usize {
        match &packgen.serving {
            super::Serving::ThroughStore(cache) => cache.cached(),
            super::Serving::Direct => panic!("the packgen is not through-store"),
        }
    }

    #[tokio::test]
    async fn generate_packs_the_fixture_over_the_real_find() {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let fx = build_fixture(&front).await;
        let packgen = PackGen::direct(front.clone());
        let pack = generate(&packgen, vec![fx.tip], vec![]).await.unwrap();

        assert!(pack.starts_with(b"PACK"), "the pack header wrote");
        let object_count = u32::from_be_bytes([pack[8], pack[9], pack[10], pack[11]]);
        assert_eq!(
            object_count, 6,
            "closure(tip) − closure(nothing) = tip, base, two trees, two blobs"
        );
        let trailer = &pack[pack.len() - 20..];
        assert!(
            trailer.iter().any(|byte| *byte != 0),
            "the trailer carries the sha1 digest"
        );
    }

    #[tokio::test]
    async fn direct_output_is_byte_identical_to_the_doubles_pipeline() {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let fx = build_fixture(&front).await;
        let wants = vec![fx.tip, fx.tag];
        let boundary = vec![fx.base];
        let packgen = PackGen::direct(front.clone());
        let pack = generate(&packgen, wants.clone(), boundary.clone())
            .await
            .unwrap();

        let find = front.session();
        let reference = tokio::task::spawn_blocking(move || {
            let mut reference = Vec::new();
            run_pipeline(&find, wants, boundary, &mut reference)?;
            Ok::<_, StorageError>(reference)
        })
        .await
        .unwrap()
        .unwrap();
        assert_eq!(
            pack, reference,
            "the direct shape is the doubles' ADR-004 pipeline, byte for byte \
             (the boundary have trims the commit walk; the tag want rides the AsIs stage)"
        );
    }

    #[tokio::test]
    async fn a_missing_want_aborts_naming_the_missing_object() {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let missing = ObjectId::from([0xA5u8; 20]);
        let packgen = PackGen::direct(front.clone());
        let error = generate(&packgen, vec![missing], vec![]).await.unwrap_err();
        match error {
            StorageError::ObjectMissing { oid } => {
                assert_eq!(oid, missing, "the abort names the missing object");
            }
            other => panic!("a missing want is the ObjectMissing abort, not {other}"),
        }
    }

    /// A commit whose tree references an oid the store does not hold:
    /// the peel and the commit walk succeed, TreeContents misses — the
    /// abort names the missing closure member, never a partial pack.
    #[tokio::test]
    async fn a_missing_closure_member_aborts_naming_it() {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let ghost = ObjectId::from([0xEEu8; 20]);
        let tree = put_object(&front, Kind::Tree, &tree_bytes(&[("ghost", &ghost)])).await;
        let commit = put_object(&front, Kind::Commit, &commit_bytes(&tree, &[])).await;
        let packgen = PackGen::direct(front);
        let error = generate(&packgen, vec![commit], vec![]).await.unwrap_err();
        match error {
            StorageError::ObjectMissing { oid } => {
                assert_eq!(oid, ghost, "the abort names the missing member");
            }
            other => panic!("a missing member is the ObjectMissing abort, not {other}"),
        }
    }

    #[tokio::test]
    async fn the_ack_predicate_recognizes_only_existing_commits() {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let fx = build_fixture(&front).await;
        let missing = ObjectId::from([0x3Cu8; 20]);
        let packgen = PackGen::direct(front.clone());
        let recognized = packgen
            .common_haves(&repo(), vec![fx.tip, fx.stray, fx.tag, missing, fx.base])
            .await
            .unwrap();
        assert_eq!(
            recognized,
            vec![fx.tip, fx.base],
            "exists + is-commit: the blob and the tag are unrecognized, a \
             missing have subtracts nothing, the caller's order is verbatim"
        );
    }

    #[tokio::test]
    async fn the_through_store_shape_serves_the_stored_pack() {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let fx = build_fixture(&front).await;
        let wants = vec![fx.tip];
        let boundary: Vec<ObjectId> = vec![];
        let direct_packgen = PackGen::direct(front.clone());
        let direct = generate(&direct_packgen, wants.clone(), boundary.clone())
            .await
            .unwrap();

        let packgen = PackGen::through_store(front.clone());
        let first = generate(&packgen, wants.clone(), boundary.clone())
            .await
            .unwrap();
        assert_eq!(first, direct, "the stored pack is the generated pack");
        assert_eq!(cached_count(&packgen), 1, "one request, one cached pack");

        let digest = request_digest(&wants, &boundary);
        let (key, len) = match &packgen.serving {
            super::Serving::ThroughStore(cache) => cache.lookup(digest).unwrap(),
            super::Serving::Direct => unreachable!(),
        };
        let stored = front.store().stat(&key.as_bytes()).await.unwrap();
        assert_eq!(stored.map(|meta| meta.len), Some(len), "put-once stat");

        let second = generate(&packgen, wants, boundary).await.unwrap();
        assert_eq!(second, direct, "the repeat fetch rides the stored pack");
        assert_eq!(cached_count(&packgen), 1, "no regeneration on the hit");
    }

    #[tokio::test]
    async fn cache_eviction_and_release_are_the_pin_paths() {
        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let fx = build_fixture(&front).await;
        let packgen = PackGen::through_store_with(front.clone(), Arc::new(PackCache::new(1)));
        let first = generate(&packgen, vec![fx.tip], vec![]).await.unwrap();
        assert_eq!(cached_count(&packgen), 1);

        let second = generate(&packgen, vec![fx.tag], vec![]).await.unwrap();
        assert_ne!(&first, &second, "the two requests differ");
        assert_eq!(
            cached_count(&packgen),
            1,
            "capacity held; the first evicted"
        );

        let cache = match &packgen.serving {
            super::Serving::ThroughStore(cache) => cache,
            super::Serving::Direct => unreachable!(),
        };
        assert!(
            cache
                .lookup(request_digest(std::slice::from_ref(&fx.tip), &[]))
                .is_none(),
            "the evicted digest is gone (its pin dropped — sweep-eligible)"
        );
        assert!(
            cache
                .lookup(request_digest(std::slice::from_ref(&fx.tag), &[]))
                .is_some(),
            "the newest entry is the cache's"
        );
        assert_eq!(cache.release_all(), 1, "the housekeeping release path");
        assert_eq!(cache.cached(), 0, "the release path drains the cache");
    }

    #[tokio::test]
    async fn a_kv_only_store_rejects_an_over_threshold_pack() {
        let front = ObjectFront::new(Arc::new(Store::kv_only_mem(256)));
        let fx = build_fixture(&front).await;
        assert!(front.stat(&fx.tip).await.unwrap().is_some());
        let packgen = PackGen::through_store(front.clone());
        let error = generate(&packgen, vec![fx.tip], vec![]).await.unwrap_err();
        match &error {
            StorageError::Io(reason) => {
                assert!(
                    reason.contains("kv-only"),
                    "the kv-only put-time rejection is what surfaces: {reason}"
                );
            }
            other => panic!("the over-threshold pack is the put's rejection, not {other}"),
        }
    }

    #[tokio::test]
    async fn the_direct_shape_never_rounds_through_the_store() {
        let front = ObjectFront::new(Arc::new(Store::kv_only_mem(256)));
        let fx = build_fixture(&front).await;
        let packgen = PackGen::direct(front.clone());
        let pack = generate(&packgen, vec![fx.tip], vec![]).await.unwrap();
        assert!(
            pack.starts_with(b"PACK"),
            "a pack the through-store shape would have rejected streams direct"
        );
    }

    /// The interop row (alkblobs' `git_interop.rs` standard): the
    /// generated direct pack indexes into a real git repo and fscks
    /// clean — skipped with a message when no `git` is installed
    /// (tests only; the serving path spawns nothing).
    #[tokio::test]
    async fn the_generated_pack_indexes_clean_under_git() {
        if std::process::Command::new("git")
            .arg("--version")
            .output()
            .is_err()
        {
            eprintln!("skipping (no git binary installed)");
            return;
        }
        let dir = scratch_root("packgen", "git-index");
        let init = std::process::Command::new("git")
            .args(["init", "-q"])
            .current_dir(&dir)
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_CONFIG_SYSTEM", "/dev/null")
            .output()
            .unwrap();
        assert!(init.status.success(), "git init failed: {init:?}");

        let front = ObjectFront::new(Arc::new(Store::mem_only()));
        let fx = build_fixture(&front).await;
        let packgen = PackGen::direct(front.clone());
        let pack = generate(&packgen, vec![fx.tip], vec![]).await.unwrap();
        let pack_path = dir.join("incoming.pack");
        std::fs::write(&pack_path, &pack).unwrap();

        let index = std::process::Command::new("git")
            .args(["index-pack", "--stdin"])
            .current_dir(&dir)
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_CONFIG_SYSTEM", "/dev/null")
            .stdin(std::process::Stdio::from(
                std::fs::File::open(&pack_path).unwrap(),
            ))
            .output()
            .unwrap();
        assert!(
            index.status.success(),
            "git index-pack failed: {}",
            String::from_utf8_lossy(&index.stderr)
        );
        let fsck = std::process::Command::new("git")
            .args(["fsck", "--strict"])
            .current_dir(&dir)
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_CONFIG_SYSTEM", "/dev/null")
            .output()
            .unwrap();
        assert!(
            fsck.status.success(),
            "git fsck --strict failed: {}",
            String::from_utf8_lossy(&fsck.stderr)
        );
        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }

    // The `cached` probe is test-only: the pack cache's occupancy is
    // observable nowhere else on the public surface.
    impl PackCache {
        fn cached(&self) -> usize {
            self.entries.lock().len()
        }
    }
}