gunnar-wire 1.5.1

//! Answering `want`/`have` from **the store's own** reachability index.
//!
//! [`crate::bitmap`] is this module's sibling and they answer the same question
//! for two different kinds of store. The difference is where the set algebra
//! happens:
//!
//! ```text
//!   crate::bitmap    store hands out BITS  → this crate does the algebra
//!                    (git's EWAH over one pack's positions, plus the .idx
//!                     that gives those positions names — so the caller can
//!                     resolve them and a bounded walk can add to them)
//!
//!   this module      store hands out the ANSWER
//!                    (roaring over the store's own ordinals; only the store
//!                     can resolve an ordinal, so only the store can do the
//!                     algebra)
//! ```
//!
//! # Why the second shape exists at all
//!
//! gunnar's znippy cold tier has no `.pack` on disk and no `.idx`, by design:
//! the storage contract replaced them with an Arrow objects table, and the end-to-end proof of
//! that store asserts those files do **not** exist. Its reachability is real,
//! exact and already built — roaring bitmaps over object ordinals — but it is
//! not addressed by pack position and cannot be, because there are no pack
//! positions.
//!
//! Encoding it into git's format to satisfy a `.bitmap`-shaped accessor's
//! return type was the first thing tried, and it is a **tax with no payer**:
//! serialise EWAH over synthetic positions so that [`crate::bitmap`] can decode
//! it back into the set algebra the store had already done. It also needs three
//! facts the store does not have — a resolved object kind per object (the objects
//! table stores
//! the *pack entry* type, so a delta's is `OfsDelta`), per-commit stored bitmaps
//! (`GitOps` exposes a query, not the table), and a single pack position space
//! (the store holds N concatenated verbatim packs). The seam was widened
//! instead of the answer being bent — and it is `GitServe::select` now, on the
//! contract itself rather than on a gunnar-owned trait above it.
//!
//! # What this module actually does
//!
//! Very little, and that is the point. Steps 1, 2, 8 and 9 below are lifted
//! from [`crate::bitmap::select`] because they are the same steps and must
//! stay the same (one implementation — the difference is one call, not a
//! second selection algorithm). Everything between them is
//! [`GitServe::select`](gunnar_store::git_storage_trait::GitServe::select).
//!
//! # What it declines
//!
//! Rather than approximate — the same list [`crate::bitmap`] keeps, for the same
//! reason, plus one more:
//!
//! * a **partial-clone filter**, a **shallow boundary**, a **deepening**'s
//!   `restrict` set: membership decided per object by facts no reachability
//!   index carries;
//! * a **tip that does not peel to a commit**. `GitServe::select`'s contract
//!   is commits in, closure out; a `want` naming a tree or a blob
//!   outright — a partial clone's lazy fetch does exactly this — needs a tree
//!   closure the index was never asked for. [`crate::bitmap`] can serve that
//!   case because it can *add* to a bit set; this one cannot add to an answer.
//!
//! Every one of them falls through to `select_objects`' walk and is served
//! exactly as it was before this module existed.
//!
//! # What it does **not** decline, and why that is deliberate
//!
//! The **object ceiling**. `Limits::object_bytes` is `Some` by default, so
//! declining on it would turn this path off on every serve path there is while
//! every test stayed green — which is precisely what happened to
//! [`crate::bitmap`] on 2026-08-04, four end-to-end cases against stock git
//! reporting green while measuring the walk. So it is enforced here, in
//! [`enforce_object_ceiling`], and [`crate::graph::Selection::answered_from_bitmap`]
//! is set so a guard can assert which path answered rather than infer it.

use gix_hash::ObjectId;
use gunnar_store::object;

use crate::error::{Error, Result};
use crate::graph::{backend, oid, ClientHas, Graph, OidSet, SelectOptions, Selection};

/// `select_objects`, answered by the store.
///
/// `Ok(None)` means "this request is not one this index can answer exactly";
/// the caller walks. See the module documentation for the list.
pub(crate) fn select(
    graph: &Graph<'_>,
    tips: &[ObjectId],
    excludes: &[ObjectId],
    options: &SelectOptions<'_>,
) -> Result<Option<Selection>> {
    if options.filter.is_some()
        || options.restrict.is_some()
        || !options.boundary.is_empty()
        || !options.have_boundary.is_empty()
    {
        return Ok(None);
    }

    // 1. Peel the tips, exactly as the walk does and exactly as
    //    `crate::bitmap::select` does. A `want` may name a tag, and the tag
    //    objects along the way are part of the answer — they are in no commit's
    //    closure, because a closure is built by walking trees and a tree never
    //    points at a tag.
    let mut commit_tips = Vec::new();
    let mut tags: Vec<ObjectId> = Vec::new();
    let mut tagged = std::collections::HashSet::new();
    for tip in tips {
        let peeled = graph.peel(tip)?;
        for tag in peeled.tags {
            if tagged.insert(tag) {
                tags.push(tag);
            }
        }
        if peeled.kind != object::Kind::Commit {
            // A tree or a blob asked for outright. See the module header: this
            // index answers commit closures and cannot be added to.
            return Ok(None);
        }
        commit_tips.push(peeled.id);
    }

    // 2. The same for the excludes, tolerating gaps: a `have` from a shallow
    //    client can name a commit whose ancestry that client does not hold.
    let mut exclude_commits = Vec::new();
    for id in excludes {
        if !graph.contains(id)? {
            continue;
        }
        match graph.peel(id) {
            Ok(peeled) if peeled.kind == object::Kind::Commit => exclude_commits.push(peeled.id),
            Ok(_) => {}
            Err(Error::ObjectMissing { .. }) => {}
            Err(err) => return Err(err),
        }
    }

    // 3. The whole selection, in one call. `want` closure minus `have` closure,
    //    over the store's own bitmaps, with no object opened on either side.
    let want: Vec<&[u8]> = commit_tips.iter().map(|id| id.as_bytes()).collect();
    let have: Vec<&[u8]> = exclude_commits.iter().map(|id| id.as_bytes()).collect();
    let set = match graph.store().select(&want, &have) {
        Ok(Some(set)) => set,
        // The store declined: its index does not cover this request. Walking is
        // the answer, and it is the same answer as before this path existed.
        Ok(None) => return Ok(None),
        // The store's index is being built (a pack is mid-absorb). That is the
        // index saying "not right now", which is `Ok(None)` in the contract's
        // own words: the walk reads objects one by one and needs no index row.
        // Refusing here told a CLIENT "the repository is unavailable; try
        // again" for every fetch that met a push being indexed (gunnar.rs
        // 2026-10-25, eleven in 45 minutes). `tests/index_in_progress.rs`.
        Err(err) if is_index_in_progress(&err) => {
            tracing::debug!(
                event = "git.select.index_in_progress",
                error = %format!("{err:#}"),
                "the reachability index is being built; the walk serves this fetch"
            );
            return Ok(None);
        }
        Err(err) => return Err(backend(err)),
    };

    // 3a. The operator's ceiling, before this answer is turned into ids.
    //
    //     `Graph::charge_walk` cannot see this path — nothing walked, so
    //     `meter::live()` is zero — and `--fetch-budget` was therefore accepted
    //     and ignored by the storage engine the appliance actually runs. The
    //     count the store just handed back is the exact number the flag is
    //     about, so it is charged here and refused before the repository-sized
    //     `Vec<ObjectId>` below is built. See `Graph::charge_answer`.
    graph.charge_answer(set.objects.len() + tags.len())?;

    // 4. Names. Tags first — the same order `crate::bitmap::select` emits and
    //    the walk emits — then the store's own order, which it is asked to keep
    //    pack-copy-friendly and which this module does not reorder.
    let mut selection = Selection {
        objects: tags,
        ..Selection::default()
    };
    selection.objects.reserve(set.objects.len());
    for bytes in &set.objects {
        let id = oid(bytes)?;
        // A tag object is reachable from nothing, so a store cannot have put it
        // in the closure; the guard is against a store that returned one anyway
        // rather than against a case this tree produces.
        if !tagged.contains(&id) {
            selection.objects.push(id);
        }
    }
    selection.commits = set
        .commits
        .iter()
        .map(|bytes| oid(bytes))
        .collect::<Result<Vec<_>>>()?;

    // 5. The voucher: what the receiver held **before** this transfer, and only
    //    that. `Selection::client_has` records the object that refuted the wider
    //    reading, and `ReachSet::client_has` is documented to the narrow one.
    let mut held = OidSet::new();
    for bytes in set.client_has.iter() {
        held.insert(oid(bytes)?);
    }
    selection.client_has = ClientHas::Ids(held);

    // 6. The object ceiling, which the walk applies per blob as it goes and
    //    this has to apply to the answer.
    if let Some(limit) = options.max_blob_bytes {
        enforce_object_ceiling(graph, &selection, limit)?;
    }

    selection.answered_from_bitmap = true;
    // Nothing was opened to reach the answer, which is what this counts.
    selection.bitmap_commits_walked = 0;
    Ok(Some(selection))
}

/// `Limits::object_bytes`, applied to a selection that was decided without ever
/// opening an object.
///
/// # Why every object and not only the blobs
///
/// [`crate::bitmap::enforce_object_ceiling`]'s equivalent asks the pack's four
/// type indexes which positions are blobs and pays a header read only for
/// those. There is no type index here:
/// [`ReachSet`](gunnar_store::git_storage_trait::ReachSet) names the
/// commits, because the store's commit graph is what built the bitmaps, and
/// says nothing about the rest. So the commits are skipped and everything else
/// gets a header read whose `kind` decides whether the size matters — the same
/// `known_blob: false` branch the bitmap path uses for its extended index, and
/// the same reason: the type is not known, so the header answers it.
///
/// A header read is not an object read. Every backend that reaches this path
/// answers `header` without inflating a payload.
/// # The bulk pre-pass, and why it is exactly equivalent
///
/// The loop below asks one thing per object: *is it over the limit?* An object
/// **under** the limit passes whatever its kind is — the `kind != Blob` arm only
/// ever turns a would-be refusal into a pass. So a size alone decides every
/// object under the ceiling, and the kind is needed only for the ones over it.
///
/// That matters because `header` is cheap on a packfile backend and expensive on
/// an archive-backed one, where it is a single-oid index lookup plus a walk of
/// the object's whole delta chain plus a partial inflate. MEASURED on the `h2h`
/// fixture: **34 124** of those per clone, to establish that a repository whose
/// largest blob is 16 KiB holds nothing over the 1 GiB default. Not one could
/// return true.
///
/// [`GitServe::sizes`](gunnar_store::git_storage_trait::GitServe::sizes) answers
/// all of them in one batched index pass when the backend has a column for it.
/// `None` for an individual row — unknown, not "fine" — falls through to the
/// `header` arm exactly as before.
///
/// **An unknown size falls through to `header`; it is never a pass.** A store
/// one drain behind reports a just-pushed object absent, and treating that as
/// "under the limit" would skip the ceiling for precisely the objects a push had
/// introduced. One writer decides the outcome — the loop below — and the bulk
/// read only ever spares it work (one implementation).
fn enforce_object_ceiling(graph: &Graph<'_>, selection: &Selection, limit: u64) -> Result<()> {
    let commits: std::collections::HashSet<&ObjectId> = selection.commits.iter().collect();
    let candidates: Vec<ObjectId> = selection
        .objects
        .iter()
        .filter(|id| !commits.contains(id))
        .copied()
        .collect();

    // Positionally answers `candidates`.
    //
    // `ObjectStore::sizes` returned `Option<Vec<..>>`, where the outer `None`
    // meant *"no bulk path, loop over `header`"*. `GitServe::sizes` as frozen
    // has no outer option: it is required, so a backend without a bulk
    // path loops internally and every caller keeps one code path. The inner
    // `Option` — this row's size is not known — survives and still matters, and
    // an unknown size still falls through to `header` rather than passing. A
    // store one drain behind reports a just-pushed object absent, and treating
    // that as "under the limit" would skip the ceiling for exactly the objects
    // a push had introduced.
    let refs: Vec<&[u8]> = candidates.iter().map(|id| id.as_bytes()).collect();
    let bulk = graph.store().sizes(&refs).map_err(backend)?;

    for (i, id) in candidates.iter().enumerate() {
        // Known and under the ceiling: decided, whatever the kind is.
        if let Some(Some(size)) = bulk.get(i) {
            if *size <= limit {
                continue;
            }
        }
        // ONE header read, kind and size together. Two calls here would be the
        // same defect removed from the commit walk one file over.
        let Some((kind, size)) = graph.store().header(id.as_bytes()).map_err(backend)? else {
            continue;
        };
        if kind != gunnar_store::git_storage_trait::ObjType::Blob {
            continue;
        }
        if size > limit {
            return Err(Error::ObjectTooLarge {
                oid: *id,
                size,
                limit,
            });
        }
    }
    Ok(())
}

/// The znippy arm's words when its absorb gate is held — a pack is being
/// indexed and its reachability rows do not exist yet
/// (`znippy-plugin-git` `GitStore::index_in_progress`).
pub(crate) const INDEX_IN_PROGRESS: &str = "index in progress for this repository";

/// Is `err` the store saying its index is being built (and nothing worse)?
fn is_index_in_progress(err: &dyn std::fmt::Display) -> bool {
    format!("{err:#}").contains(INDEX_IN_PROGRESS)
}