znippy-plugin-git 0.2.3

//! `__gunnar_reach__` — reachability bitmaps.
//!
//! For each *selected* commit, the set of objects reachable from it — the commit
//! itself, its ancestors, and every tree and blob those commits point at —
//! recorded as a roaring bitmap over **object ordinals**.
//!
//! # Two ordinal spaces, one view
//!
//! A **sealed archive** numbers its objects in oid-lexicographic order: the order
//! the sorted lookup sub-index is in and the order `__gunnar_oid__` records in
//! its `ordinal` column, so a bitmap position is resolvable with the oid index
//! alone. [`OrdinalSpace::new`] is that space.
//!
//! A **live store** numbers its objects in an append-only space
//! ([`OrdinalTable`]): a sorted base laid down by the first fold, then every
//! object a later push brings in appended behind it in arrival order. The point
//! is that a push never renumbers anything — a bitmap is a fact about an
//! immutable commit, and under the sorted space every push shifted every ordinal
//! above the new oids, so every stored bitmap had to be rewritten bit by bit
//! (`carry_reach`, 512 entries × every object, after every push). Under the
//! stable space a push appends a few ordinals and the table is untouched. The
//! sorted order is restored only when the store seals or the appended part
//! outgrows the base, both through one remap table ([`OrdinalTable::remap_to`]).
//!
//! # No hex on any hot path
//!
//! Everything below the public builders is in integers: the graph is
//! [`OrdGraph`] (rows and ordinals), the table is [`ReachTable`] (ordinals and
//! bitmaps), a tree is read by its raw oid and each of its entries is resolved
//! with one [`OrdinalSpace::ordinal_of_raw`] — a 16-bit fanout and a short
//! binary search, no decode. GUN-PERF measured the hex form on git.git
//! (402 227 objects): the build took 85 to 150 s and `ordinal_of_hex` alone was
//! 51 % of a 65 s clone (gunnar 18c10860).
//!
//! # The builder is git's, not a closure memo
//!
//! [`build_reach_table`] walks the commits parents-before-children, starts each
//! commit's bitmap from its parents' and adds the trees with the skip rule
//! (`should_include_obj`: a set bit means that object AND everything under it
//! is already accounted for). Work is O(objects + commits), and the memory is
//! the frontier — the bitmaps of commits whose children are still to come — not
//! one closure bitmap per distinct tree. The old builder kept the latter and
//! paid a hex `String` per tree entry on top.

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

use anyhow::{Result, anyhow};
use roaring::RoaringBitmap;
use znippy_common::GUNNAR_REACH_MODULE;
use znippy_common::arrow::array::{Array, BinaryArray, BinaryBuilder, StringArray, StringBuilder};
use znippy_common::arrow::datatypes::{DataType, Field, Schema};
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_common::read_reserved_section_bytes;

use crate::graph::CommitNode;
use crate::object::{GitObjectKind, tree_entries};

/// "No ordinal" / "no row" in the integer tables.
pub const NONE: u32 = u32::MAX;

/// Which commits get a bitmap.
#[derive(Debug, Clone, Copy)]
pub struct ReachPolicy {
    /// Upper bound on how many bitmaps are sealed. Tips are taken first (a
    /// commit with no child inside the archive — the branch heads a clone
    /// negotiates against), then the remaining commits are sampled at an even
    /// stride through generation order.
    pub max_commits: usize,
}

impl Default for ReachPolicy {
    fn default() -> Self {
        // git's own bitmap selection is of this order: cover the tips, then
        // sample the history so a `have` usually lands near a bitmapped commit.
        Self { max_commits: 512 }
    }
}

/// One sealed bitmap, in the sealed archive's spelling (hex commit oid). The
/// live store keeps a [`ReachTable`]; this is the `__gunnar_reach__` row.
#[derive(Debug, Clone, PartialEq)]
pub struct ReachEntry {
    pub commit: String,
    pub bitmap: RoaringBitmap,
}

pub fn reach_schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("commit_oid", DataType::Utf8, false),
        Field::new("cardinality", DataType::UInt64, false),
        Field::new("bitmap", DataType::Binary, false),
    ]))
}

/// **Where a tree's bytes come from** — a lookup, not a table, keyed by the
/// **raw** oid.
///
/// It was `&HashMap<String, Vec<u8>>`: every tree payload in the repository,
/// resident, keyed by a hex `String`. MEASURED on oden 2026-08-30, the real
/// `torvalds/linux` mirror through [`crate::git_ops::GitStore`]: **28.24 GB**
/// of tree payloads held on disk so that map could be rebuilt into RAM on every
/// fold; peak RSS 50.9 GiB on a 16 GB appliance. So the map became a lookup,
/// and the store's arm resolves a tree the way any other read does.
///
/// `None` is *"not a tree this archive holds"* and is an ordinary answer, not
/// a fault: a blob is a leaf, a gitlink is a commit in another repository, and
/// a thin push's base may be outside the archive. Every caller treats it as
/// "stop here".
pub trait TreeSource: Sync {
    fn tree(&self, oid: &[u8]) -> Option<std::borrow::Cow<'_, [u8]>>;
}

/// The eager arm keyed by hex (the sealed builder and the tests): one encode
/// per tree *visited*, never per entry.
impl TreeSource for HashMap<String, Vec<u8>> {
    fn tree(&self, oid: &[u8]) -> Option<std::borrow::Cow<'_, [u8]>> {
        self.get(&hex::encode(oid)).map(|v| std::borrow::Cow::Borrowed(v.as_slice()))
    }
}

impl TreeSource for HashMap<Vec<u8>, Vec<u8>> {
    fn tree(&self, oid: &[u8]) -> Option<std::borrow::Cow<'_, [u8]>> {
        self.get(oid).map(|v| std::borrow::Cow::Borrowed(v.as_slice()))
    }
}

/// Number of 16-bit prefixes plus one: `fanout[p]` is the first base index whose
/// two leading bytes are `>= p`, and `fanout[65536]` is the base length. git's
/// `.idx` has the 8-bit version of the same table; two bytes cut a 400 000-oid
/// search to about six entries.
const FANOUT_LEN: usize = (1 << 16) + 1;

fn build_fanout(base: &[u8], oid_len: usize) -> Vec<u32> {
    let n = if oid_len == 0 { 0 } else { base.len() / oid_len };
    let mut fan = vec![0u32; FANOUT_LEN];
    let mut i = 0usize;
    for p in 0..(1usize << 16) {
        while i < n {
            let at = i * oid_len;
            let key = (usize::from(base[at]) << 8) | usize::from(base[at + 1]);
            if key >= p {
                break;
            }
            i += 1;
        }
        fan[p] = i as u32;
    }
    fan[1 << 16] = n as u32;
    fan
}

/// **oid ⇄ ordinal, as a `Copy` view that owns nothing.**
///
/// Two layouts share it. A sorted buffer alone ([`Self::new`]) is the sealed
/// archive's space. An [`OrdinalTable`]'s view adds the fanout and the appended
/// part: ordinal `o < base_len` is the slice at `o * oid_len` of `base`, and
/// `o >= base_len` is the slice at `(o - base_len) * oid_len` of `extra`, which
/// is in arrival order and is searched through `extra_sorted` (its ordinals,
/// sorted by oid).
#[derive(Clone, Copy)]
pub struct OrdinalSpace<'a> {
    base: &'a [u8],
    /// Empty for a plain sorted buffer; `FANOUT_LEN` entries otherwise.
    fanout: &'a [u32],
    extra: &'a [u8],
    extra_sorted: &'a [u32],
    oid_len: usize,
}

impl<'a> OrdinalSpace<'a> {
    /// Borrow a flat, **sorted**, fixed-width oid buffer as the ordinal space.
    /// The sort is the caller's invariant (every sealed builder produces it in
    /// oid-lexicographic order); the binary search below rests on it.
    pub fn new(raw: &'a [u8], oid_len: usize) -> Self {
        debug_assert!(oid_len > 0 && raw.len() % oid_len == 0, "ragged ordinal space");
        Self { base: raw, fanout: &[], extra: &[], extra_sorted: &[], oid_len }
    }

    /// How many oids the space holds.
    pub fn len(&self) -> usize {
        if self.oid_len == 0 { 0 } else { (self.base.len() + self.extra.len()) / self.oid_len }
    }

    pub fn is_empty(&self) -> bool {
        self.base.is_empty() && self.extra.is_empty()
    }

    pub fn oid_len(&self) -> usize {
        self.oid_len
    }

    fn base_len(&self) -> usize {
        if self.oid_len == 0 { 0 } else { self.base.len() / self.oid_len }
    }

    /// ordinal → raw oid, a slice into the flat buffers. `None` for an ordinal
    /// outside the space.
    pub fn oid(&self, ordinal: u32) -> Option<&'a [u8]> {
        let o = ordinal as usize;
        let b = self.base_len();
        if o < b {
            let at = o * self.oid_len;
            self.base.get(at..at + self.oid_len)
        } else {
            let at = (o - b).checked_mul(self.oid_len)?;
            self.extra.get(at..at + self.oid_len)
        }
    }

    /// raw oid → ordinal: the fanout narrows the base to one 16-bit prefix, a
    /// binary search finishes it, and the appended part is searched through
    /// its sorted index. This is the exact inverse of [`Self::oid`].
    pub fn ordinal_of_raw(&self, oid: &[u8]) -> Option<u32> {
        if oid.len() != self.oid_len {
            return None;
        }
        let (mut lo, mut hi) = if self.fanout.len() == FANOUT_LEN && self.oid_len >= 2 {
            let p = (usize::from(oid[0]) << 8) | usize::from(oid[1]);
            (self.fanout[p] as usize, self.fanout[p + 1] as usize)
        } else {
            (0usize, self.base_len())
        };
        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            let at = mid * self.oid_len;
            match self.base[at..at + self.oid_len].cmp(oid) {
                std::cmp::Ordering::Less => lo = mid + 1,
                std::cmp::Ordering::Greater => hi = mid,
                std::cmp::Ordering::Equal => return Some(mid as u32),
            }
        }
        if self.extra_sorted.is_empty() {
            return None;
        }
        let b = self.base_len();
        let (mut lo, mut hi) = (0usize, self.extra_sorted.len());
        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            let o = self.extra_sorted[mid];
            let at = (o as usize - b) * self.oid_len;
            match self.extra[at..at + self.oid_len].cmp(oid) {
                std::cmp::Ordering::Less => lo = mid + 1,
                std::cmp::Ordering::Greater => hi = mid,
                std::cmp::Ordering::Equal => return Some(o),
            }
        }
        None
    }

    /// hex oid → ordinal. Decodes the hex into a **stack** buffer (≤ 32 bytes),
    /// so a lookup allocates nothing, then searches the raw form. Not on any
    /// serving path: the walk and the builder carry raw oids.
    pub fn ordinal_of_hex(&self, hex: &str) -> Option<u32> {
        let mut buf = [0u8; 32];
        let dst = buf.get_mut(..self.oid_len)?;
        hex::decode_to_slice(hex, dst).ok()?;
        self.ordinal_of_raw(dst)
    }

    /// The oids in ordinal order, concatenated — what a sealed builder sorts
    /// and what [`OrdinalTable::from_sorted`] takes.
    pub fn concat(&self) -> Vec<u8> {
        let mut v = Vec::with_capacity(self.base.len() + self.extra.len());
        v.extend_from_slice(self.base);
        v.extend_from_slice(self.extra);
        v
    }
}

/// **The live store's ordinal space: a sorted base, then arrivals appended.**
/// Owned; [`Self::view`] is the lookup. See the module header for why it is
/// append-only.
#[derive(Clone, Default)]
pub struct OrdinalTable {
    oid_len: usize,
    base: Vec<u8>,
    fanout: Vec<u32>,
    extra: Vec<u8>,
    extra_sorted: Vec<u32>,
}

impl OrdinalTable {
    pub fn empty(oid_len: usize) -> Self {
        Self { oid_len, ..Default::default() }
    }

    /// A fresh table over a **sorted** full oid list: base only, nothing appended.
    pub fn from_sorted(raw: Vec<u8>, oid_len: usize) -> Self {
        debug_assert!(oid_len > 0 && raw.len() % oid_len == 0, "ragged ordinal table");
        let fanout = build_fanout(&raw, oid_len);
        Self { oid_len, base: raw, fanout, extra: Vec::new(), extra_sorted: Vec::new() }
    }

    /// The persisted form, replayed: `full_sorted` is the store's current sorted
    /// oid list and `extra` the appended part in arrival order. The base is
    /// `full − extra`. `None` when the two do not agree (an oid in `extra` that
    /// the store no longer holds, or a width mismatch) — rebuild from scratch.
    pub fn from_parts(full_sorted: &[u8], extra: &[u8], oid_len: usize) -> Option<Self> {
        if oid_len == 0 || full_sorted.len() % oid_len != 0 || extra.len() % oid_len != 0 {
            return None;
        }
        let mut ex: Vec<&[u8]> = extra.chunks_exact(oid_len).collect();
        ex.sort_unstable();
        ex.dedup();
        if ex.len() * oid_len != extra.len() {
            return None;
        }
        let mut base = Vec::with_capacity(full_sorted.len().saturating_sub(extra.len()));
        let mut j = 0usize;
        for oid in full_sorted.chunks_exact(oid_len) {
            if j < ex.len() && ex[j] == oid {
                j += 1;
            } else {
                base.extend_from_slice(oid);
            }
        }
        if j != ex.len() {
            return None;
        }
        let mut t = Self::from_sorted(base, oid_len);
        t.extra = extra.to_vec();
        t.reindex_extra();
        Some(t)
    }

    /// The persisted form, replayed from the two raw parts: `base` must be
    /// sorted and distinct (the binary search rests on it), `extra` distinct
    /// and disjoint from it. `None` otherwise — a malformed file is a rebuild,
    /// never a wrong ordinal.
    pub fn from_base_and_extra(base: Vec<u8>, extra: Vec<u8>, oid_len: usize) -> Option<Self> {
        if oid_len == 0 || base.len() % oid_len != 0 || extra.len() % oid_len != 0 {
            return None;
        }
        if base.chunks_exact(oid_len).zip(base.chunks_exact(oid_len).skip(1)).any(|(a, b)| a >= b) {
            return None;
        }
        let mut t = Self::from_sorted(base, oid_len);
        t.extra = extra;
        t.reindex_extra();
        let v = t.view();
        for (i, o) in t.extra.chunks_exact(oid_len).enumerate() {
            // Each appended oid must resolve to ITS ordinal: absent from the
            // base, and not a duplicate inside `extra`.
            if v.ordinal_of_raw(o) != Some((t.base_len() + i) as u32) {
                return None;
            }
        }
        Some(t)
    }

    /// The sorted base, raw — what the persisted form stores.
    pub fn base_raw(&self) -> &[u8] {
        &self.base
    }

    fn reindex_extra(&mut self) {
        let b = (self.base.len() / self.oid_len) as u32;
        let n = self.extra.len() / self.oid_len;
        let mut idx: Vec<u32> = (0..n as u32).map(|i| i + b).collect();
        let (extra, w, bb) = (&self.extra, self.oid_len, b as usize);
        idx.sort_unstable_by(|&x, &y| {
            let a = &extra[(x as usize - bb) * w..(x as usize - bb + 1) * w];
            let c = &extra[(y as usize - bb) * w..(y as usize - bb + 1) * w];
            a.cmp(c)
        });
        self.extra_sorted = idx;
    }

    pub fn len(&self) -> usize {
        if self.oid_len == 0 { 0 } else { (self.base.len() + self.extra.len()) / self.oid_len }
    }

    pub fn is_empty(&self) -> bool {
        self.base.is_empty() && self.extra.is_empty()
    }

    pub fn oid_len(&self) -> usize {
        self.oid_len
    }

    pub fn base_len(&self) -> usize {
        if self.oid_len == 0 { 0 } else { self.base.len() / self.oid_len }
    }

    /// How many ordinals were appended behind the sorted base.
    pub fn extra_len(&self) -> usize {
        if self.oid_len == 0 { 0 } else { self.extra.len() / self.oid_len }
    }

    /// The appended part, arrival order — what the persisted form stores.
    pub fn extra_raw(&self) -> &[u8] {
        &self.extra
    }

    pub fn view(&self) -> OrdinalSpace<'_> {
        OrdinalSpace {
            base: &self.base,
            fanout: &self.fanout,
            extra: &self.extra,
            extra_sorted: &self.extra_sorted,
            oid_len: self.oid_len,
        }
    }

    /// **Follow the store**: `full_sorted` is every oid the store holds now, in
    /// sorted order. Every oid in it that this table does not number is
    /// appended (in sorted order among themselves, after everything already
    /// numbered) and the range of new ordinals is returned. `None` when an oid
    /// the table numbers is **absent** from `full_sorted` — something was
    /// removed under the table (a gc, a retain) and the numbering cannot be
    /// kept; the caller rebuilds from scratch.
    pub fn follow(&mut self, full_sorted: &[u8]) -> Option<std::ops::Range<u32>> {
        let w = self.oid_len;
        if w == 0 || full_sorted.len() % w != 0 {
            return None;
        }
        let before = self.len() as u32;
        // The known oids in sorted order: a merge of the base and the sorted
        // view of the appended part.
        let mut known = KnownIter { table: self, i: 0, j: 0 };
        let mut appended: Vec<u8> = Vec::new();
        let mut cur = known.next();
        for oid in full_sorted.chunks_exact(w) {
            loop {
                match cur {
                    None => {
                        appended.extend_from_slice(oid);
                        break;
                    }
                    Some(k) if k == oid => {
                        cur = known.next();
                        break;
                    }
                    Some(k) if k > oid => {
                        appended.extend_from_slice(oid);
                        break;
                    }
                    // A known oid smaller than the next store oid: the store
                    // no longer holds it.
                    Some(_) => return None,
                }
            }
        }
        if cur.is_some() {
            return None;
        }
        if appended.is_empty() {
            return Some(before..before);
        }
        self.extra.extend_from_slice(&appended);
        self.reindex_extra();
        Some(before..self.len() as u32)
    }

    /// **old ordinal → ordinal in `target`**, as one table, for a seal or a
    /// compaction. [`NONE`] where `target` lacks the oid. O(n) lookups through
    /// the target's fanout, once — not once per set bit.
    pub fn remap_to(&self, target: &OrdinalSpace<'_>) -> Vec<u32> {
        let me = self.view();
        (0..me.len() as u32)
            .map(|o| me.oid(o).and_then(|oid| target.ordinal_of_raw(oid)).unwrap_or(NONE))
            .collect()
    }
}

/// The oids an [`OrdinalTable`] numbers, in sorted order: a merge of the base
/// and the sorted view of the appended part.
struct KnownIter<'a> {
    table: &'a OrdinalTable,
    i: usize,
    j: usize,
}

impl<'a> Iterator for KnownIter<'a> {
    type Item = &'a [u8];
    fn next(&mut self) -> Option<&'a [u8]> {
        let t = self.table;
        let w = t.oid_len;
        let base_n = t.base_len();
        let a = (self.i < base_n).then(|| &t.base[self.i * w..(self.i + 1) * w]);
        let b = t.extra_sorted.get(self.j).map(|&o| {
            let at = (o as usize - base_n) * w;
            &t.extra[at..at + w]
        });
        match (a, b) {
            (Some(a), Some(b)) => {
                if a <= b {
                    self.i += 1;
                    Some(a)
                } else {
                    self.j += 1;
                    Some(b)
                }
            }
            (Some(a), None) => {
                self.i += 1;
                Some(a)
            }
            (None, Some(b)) => {
                self.j += 1;
                Some(b)
            }
            (None, None) => None,
        }
    }
}

/// **The commit graph in integers**: one row per commit, rows in the order the
/// [`CommitNode`] list came in (parents before children, as
/// [`crate::graph::assign_generations`] guarantees).
#[derive(Clone, Default)]
pub struct OrdGraph {
    /// row → the commit's ordinal.
    pub ords: Vec<u32>,
    /// row → the root tree's ordinal, or [`NONE`].
    pub tree: Vec<u32>,
    /// row → parent **rows** (a parent outside the graph contributes nothing).
    pub parents: Vec<Vec<u32>>,
    /// row → generation.
    pub generation: Vec<u32>,
    /// The commit ordinals, as a set.
    commits: RoaringBitmap,
    /// rank in `commits` → row.
    row_by_rank: Vec<u32>,
}

impl OrdGraph {
    /// Decode each node's hex **once** and resolve it against the space. A node
    /// whose oid the space does not hold is kept as a row with no ordinal (it
    /// cannot be asked for by ordinal, and its children still see it as a
    /// parent row).
    pub fn from_nodes(nodes: &[CommitNode], space: &OrdinalSpace<'_>) -> Self {
        let n = nodes.len();
        let mut ords = Vec::with_capacity(n);
        let mut commits = RoaringBitmap::new();
        for node in nodes {
            let o = space.ordinal_of_hex(&node.oid).unwrap_or(NONE);
            if o != NONE {
                commits.insert(o);
            }
            ords.push(o);
        }
        let mut row_by_rank = vec![NONE; commits.len() as usize];
        for (row, &o) in ords.iter().enumerate() {
            if o != NONE {
                row_by_rank[commits.rank(o) as usize - 1] = row as u32;
            }
        }
        let g = Self { ords, tree: Vec::new(), parents: Vec::new(), generation: Vec::new(), commits, row_by_rank };
        let mut tree = Vec::with_capacity(n);
        let mut parents = Vec::with_capacity(n);
        let mut generation = Vec::with_capacity(n);
        for node in nodes {
            tree.push(node.tree.as_deref().and_then(|t| space.ordinal_of_hex(t)).unwrap_or(NONE));
            let mut ps: Vec<u32> = node
                .parents
                .iter()
                .filter_map(|p| space.ordinal_of_hex(p))
                .filter_map(|o| g.row_of(o))
                .collect();
            ps.sort_unstable();
            ps.dedup();
            parents.push(ps);
            generation.push(node.generation);
        }
        Self { tree, parents, generation, ..g }
    }

    pub fn len(&self) -> usize {
        self.ords.len()
    }

    pub fn is_empty(&self) -> bool {
        self.ords.is_empty()
    }

    /// The set of commit ordinals.
    pub fn commits(&self) -> &RoaringBitmap {
        &self.commits
    }

    #[inline]
    pub fn is_commit(&self, ordinal: u32) -> bool {
        self.commits.contains(ordinal)
    }

    /// ordinal → row, for a commit.
    #[inline]
    pub fn row_of(&self, ordinal: u32) -> Option<u32> {
        if !self.commits.contains(ordinal) {
            return None;
        }
        self.row_by_rank.get(self.commits.rank(ordinal) as usize - 1).copied()
    }

    /// Rows with no child in the graph — the tips.
    pub fn tip_rows(&self) -> Vec<u32> {
        let mut has_child = vec![false; self.len()];
        for ps in &self.parents {
            for &p in ps {
                has_child[p as usize] = true;
            }
        }
        (0..self.len() as u32).filter(|&r| !has_child[r as usize]).collect()
    }
}

/// **The live reachability table**: entry `i` is the closure of the commit at
/// ordinal `commits[i]`.
#[derive(Clone, Default, Debug, PartialEq)]
pub struct ReachTable {
    pub commits: Vec<u32>,
    pub bitmaps: Vec<RoaringBitmap>,
}

impl ReachTable {
    pub fn len(&self) -> usize {
        self.commits.len()
    }

    pub fn is_empty(&self) -> bool {
        self.commits.is_empty()
    }

    /// row → entry index (or [`NONE`]), over `graph`. Rebuilt at every fold
    /// beside the graph; the table itself is not.
    pub fn by_row(&self, graph: &OrdGraph) -> Vec<u32> {
        let mut v = vec![NONE; graph.len()];
        for (i, &o) in self.commits.iter().enumerate() {
            if let Some(r) = graph.row_of(o) {
                v[r as usize] = i as u32;
            }
        }
        v
    }

    /// The sealed spelling, with every ordinal mapped through `remap` (old →
    /// sorted). `None` if a set bit maps to [`NONE`] — an entry naming an object
    /// the target space lacks would be a silent over-claim.
    pub fn to_entries(&self, space: &OrdinalSpace<'_>, remap: Option<&[u32]>) -> Option<Vec<ReachEntry>> {
        let mut out = Vec::with_capacity(self.len());
        for (i, &o) in self.commits.iter().enumerate() {
            // The commit's name comes out of the TARGET space, through the
            // same remap its bits go through.
            let c = match remap {
                None => o,
                Some(m) => match m.get(o as usize).copied() {
                    Some(n) if n != NONE => n,
                    _ => return None,
                },
            };
            let commit = hex::encode(space.oid(c)?);
            let bitmap = match remap {
                None => self.bitmaps[i].clone(),
                Some(m) => {
                    let mut mapped: Vec<u32> = Vec::with_capacity(self.bitmaps[i].len() as usize);
                    for b in self.bitmaps[i].iter() {
                        match m.get(b as usize).copied() {
                            Some(n) if n != NONE => mapped.push(n),
                            _ => return None,
                        }
                    }
                    mapped.sort_unstable();
                    RoaringBitmap::from_sorted_iter(mapped).ok()?
                }
            };
            out.push(ReachEntry { commit, bitmap });
        }
        Some(out)
    }

    /// Renumber every bit through `remap` (old ordinal → new ordinal); an entry
    /// whose commit maps to [`NONE`] is dropped, a BIT that does is a refusal
    /// (`None`: rebuild). One array index per bit.
    pub fn remapped(&self, remap: &[u32]) -> Option<ReachTable> {
        let mut out = ReachTable::default();
        for (i, &o) in self.commits.iter().enumerate() {
            let c = *remap.get(o as usize)?;
            if c == NONE {
                continue;
            }
            let mut mapped: Vec<u32> = Vec::with_capacity(self.bitmaps[i].len() as usize);
            for b in self.bitmaps[i].iter() {
                match remap.get(b as usize).copied() {
                    Some(n) if n != NONE => mapped.push(n),
                    _ => return None,
                }
            }
            mapped.sort_unstable();
            out.commits.push(c);
            out.bitmaps.push(RoaringBitmap::from_sorted_iter(mapped).ok()?);
        }
        Some(out)
    }

    /// Keep at most `max` entries: the current tips first, then an even sample
    /// of the rest in table order.
    pub fn pruned(&self, graph: &OrdGraph, max: usize) -> ReachTable {
        if self.len() <= max {
            return self.clone();
        }
        let mut is_tip = vec![false; graph.len()];
        for r in graph.tip_rows() {
            is_tip[r as usize] = true;
        }
        let tip_idx: Vec<usize> = (0..self.len())
            .filter(|&i| graph.row_of(self.commits[i]).is_some_and(|r| is_tip[r as usize]))
            .collect();
        let mut chosen: Vec<usize> = tip_idx.iter().copied().take(max).collect();
        if chosen.len() < max {
            let room = max - chosen.len();
            let rest: Vec<usize> = (0..self.len()).filter(|i| !tip_idx.contains(i)).collect();
            if !rest.is_empty() {
                let stride = rest.len().div_ceil(room).max(1);
                chosen.extend(rest.iter().step_by(stride).take(room));
            }
        }
        chosen.sort_unstable();
        chosen.dedup();
        ReachTable {
            commits: chosen.iter().map(|&i| self.commits[i]).collect(),
            bitmaps: chosen.iter().map(|&i| self.bitmaps[i].clone()).collect(),
        }
    }
}

/// **Carry a sealed-spelling table across a renumbering** — kept for the sealed
/// builders and the tests that speak hex; the live store uses
/// [`ReachTable::remapped`] with one remap table. `keep` decides which commits
/// survive. `None` means the carry cannot be exact (a set bit's object is gone).
pub fn carry_reach(
    old: &[ReachEntry],
    old_space: &OrdinalSpace<'_>,
    new_space: &OrdinalSpace<'_>,
    keep: impl Fn(&str) -> bool,
) -> Option<Vec<ReachEntry>> {
    if old.is_empty() || old_space.is_empty() {
        return None;
    }
    let remap: Vec<u32> = (0..old_space.len() as u32)
        .map(|o| old_space.oid(o).and_then(|oid| new_space.ordinal_of_raw(oid)).unwrap_or(NONE))
        .collect();
    let mut out = Vec::with_capacity(old.len());
    for e in old {
        if !keep(&e.commit) {
            continue;
        }
        let mut mapped = Vec::with_capacity(e.bitmap.len() as usize);
        for b in e.bitmap.iter() {
            match remap.get(b as usize).copied() {
                Some(n) if n != NONE => mapped.push(n),
                _ => return None,
            }
        }
        mapped.sort_unstable();
        let bitmap = RoaringBitmap::from_sorted_iter(mapped).ok()?;
        out.push(ReachEntry { commit: e.commit.clone(), bitmap });
    }
    Some(out)
}

/// What the builder needs to know about every object in the archive.
pub struct ObjectFacts<'a> {
    /// oid ⇄ ordinal — the bitmap space.
    pub ordinal: OrdinalSpace<'a>,
    /// Tree payloads, by raw oid. Blobs are not needed (they are leaves) and
    /// are deliberately not asked for.
    pub trees: &'a dyn TreeSource,
    /// Raw oid width, needed to walk tree records.
    pub oid_len: usize,
}

/// **Every object under `root` that `acc` does not already hold** — git's
/// `fill_bitmap_tree` with `should_include_obj`.
///
/// A set bit means that object and, for a tree, everything under it, is already
/// accounted for: either by a parent's closure the caller started `acc` from, or
/// by an earlier tree in this same walk. Descending anyway would be correct and
/// would make the whole exercise O(commits × tree). Iterative: a deep source
/// tree is not a stack-depth bet, and a corrupt cycle terminates because the
/// revisited oid's bit is already set.
pub(crate) fn fill_tree(root: u32, facts: &ObjectFacts<'_>, acc: &mut RoaringBitmap) {
    fill_tree_counted(root, facts, acc, &mut BuildCount::default());
}

/// What a build spent, for the opt-in report (`ZNIPPY_REACH_TIMING=1`):
/// counters, not clocks, except the one clock around the tree reads.
#[derive(Default)]
struct BuildCount {
    trees_read: u64,
    tree_bytes: u64,
    entries: u64,
    tree_read: std::time::Duration,
}

fn fill_tree_counted(root: u32, facts: &ObjectFacts<'_>, acc: &mut RoaringBitmap, count: &mut BuildCount) {
    let mut stack: Vec<u32> = vec![root];
    while let Some(o) = stack.pop() {
        if !acc.insert(o) {
            continue;
        }
        let Some(oid) = facts.ordinal.oid(o) else { continue };
        let t0 = std::time::Instant::now();
        let payload = facts.trees.tree(oid);
        count.tree_read += t0.elapsed();
        count.trees_read += 1;
        if let Some(payload) = payload {
            count.tree_bytes += payload.len() as u64;
            for e in tree_entries(&payload, facts.oid_len) {
                count.entries += 1;
                // An entry the archive does not hold (a gitlink, a thin base
                // outside) has no ordinal and is not selected for emission.
                if let Some(c) = facts.ordinal.ordinal_of_raw(e.oid) {
                    stack.push(c);
                }
            }
        }
    }
}

/// Tips first, then an even sample through the rest, capped at `max_commits`.
fn select_rows(graph: &OrdGraph, policy: ReachPolicy) -> Vec<u32> {
    let n = graph.len();
    let mut has_child = vec![false; n];
    for ps in &graph.parents {
        for &p in ps {
            has_child[p as usize] = true;
        }
    }
    let mut chosen: Vec<u32> = (0..n as u32).filter(|&i| !has_child[i as usize]).collect();
    chosen.truncate(policy.max_commits);
    if chosen.len() < policy.max_commits {
        let room = policy.max_commits - chosen.len();
        let rest: Vec<u32> = (0..n as u32).filter(|&i| has_child[i as usize]).collect();
        if !rest.is_empty() {
            let stride = rest.len().div_ceil(room).max(1);
            chosen.extend(rest.iter().step_by(stride).take(room));
        }
    }
    chosen.sort_unstable();
    chosen.dedup();
    chosen
}

/// **Build the table from scratch**, parents before children, each bitmap
/// started from its parents' and completed by [`fill_tree`].
///
/// Memory is the frontier: a commit's bitmap lives only while a child still
/// needs it (or it was selected). The last child to need a parent takes the
/// parent's bitmap by move; every other child clones it. Rows that are not in
/// the ordinal space (no oid) still get a bitmap so their children inherit the
/// history, but they are never selected.
pub fn build_reach_table(graph: &OrdGraph, facts: &ObjectFacts<'_>, policy: ReachPolicy) -> ReachTable {
    let n = graph.len();
    if n == 0 || policy.max_commits == 0 {
        return ReachTable::default();
    }
    let selected = select_rows(graph, policy);
    let mut keep = vec![false; n];
    for &r in &selected {
        keep[r as usize] = graph.ords[r as usize] != NONE;
    }
    let mut pending: Vec<u32> = vec![0; n];
    for ps in &graph.parents {
        for &p in ps {
            pending[p as usize] += 1;
        }
    }
    let mut live: Vec<Option<RoaringBitmap>> = (0..n).map(|_| None).collect();
    let report = std::env::var_os(crate::arms::ENV_REACH_TIMING).is_some();
    let started = std::time::Instant::now();
    let mut count = BuildCount::default();
    let mut live_peak = 0usize;
    let mut live_now = 0usize;
    for i in 0..n {
        let ps = &graph.parents[i];
        let mut bm: Option<RoaringBitmap> = None;
        for &p in ps {
            let p = p as usize;
            pending[p] -= 1;
            let last = pending[p] == 0 && !keep[p];
            if last && live[p].is_some() {
                live_now -= 1;
            }
            match (&mut bm, last) {
                (None, true) => bm = live[p].take(),
                (None, false) => bm = live[p].clone(),
                (Some(acc), true) => {
                    if let Some(pb) = live[p].take() {
                        *acc |= pb;
                    }
                }
                (Some(acc), false) => {
                    if let Some(pb) = &live[p] {
                        *acc |= pb;
                    }
                }
            }
        }
        let mut bm = bm.unwrap_or_default();
        if graph.ords[i] != NONE {
            bm.insert(graph.ords[i]);
        }
        if graph.tree[i] != NONE {
            fill_tree_counted(graph.tree[i], facts, &mut bm, &mut count);
        }
        if pending[i] > 0 || keep[i] {
            live[i] = Some(bm);
            live_now += 1;
            live_peak = live_peak.max(live_now);
        }
    }
    if report {
        eprintln!(
            "reach build: {} commits, {} objects, {} entries; {} trees read ({} MB, {} entries) in {:.2} s of {:.2} s; live bitmaps peak {live_peak}, end {live_now}",
            n,
            facts.ordinal.len(),
            selected.len(),
            count.trees_read,
            count.tree_bytes >> 20,
            count.entries,
            count.tree_read.as_secs_f64(),
            started.elapsed().as_secs_f64(),
        );
    }
    let mut out = ReachTable::default();
    for &r in &selected {
        if !keep[r as usize] {
            continue;
        }
        // `expect`, not `unwrap_or_default`: a missing bitmap here means the
        // retention rule let a kept row be dropped, and an EMPTY bitmap would
        // be a silent under-send — a clone served one object per branch.
        let bitmap = live[r as usize].take().unwrap_or_else(|| {
            panic!("row {r} was selected for a reachability bitmap and its bitmap is not live at the end of the build")
        });
        out.commits.push(graph.ords[r as usize]);
        out.bitmaps.push(bitmap);
    }
    out
}

/// Give every `EXTEND_STRIDE`-th new commit an entry besides the new tips, so a
/// later walk down a long pushed branch is bounded by this many commits' trees.
const EXTEND_STRIDE: usize = 64;

/// **Extend a table after a push**: entries for the new tips (and a sample of
/// the other new commits), each computed by the bounded walk down to the
/// entries that already exist. Nothing already in the table is touched or
/// recomputed; the cost is the pushed commits' trees, whatever the repository's
/// size. `new_rows` must be in parents-before-children order (the caller takes
/// them off the generation-ordered graph).
pub fn extend_reach_table(
    table: &ReachTable,
    graph: &OrdGraph,
    facts: &ObjectFacts<'_>,
    new_rows: &[u32],
    policy: ReachPolicy,
) -> ReachTable {
    if new_rows.is_empty() || policy.max_commits == 0 {
        return table.clone();
    }
    let mut out = table.clone();
    let mut by_row = out.by_row(graph);
    let mut has_child = vec![false; graph.len()];
    for ps in &graph.parents {
        for &p in ps {
            has_child[p as usize] = true;
        }
    }
    for (k, &r) in new_rows.iter().enumerate() {
        let row = r as usize;
        if graph.ords[row] == NONE || by_row[row] != NONE {
            continue;
        }
        if has_child[row] && k % EXTEND_STRIDE != EXTEND_STRIDE - 1 {
            continue;
        }
        let mut bm = RoaringBitmap::new();
        {
            let view = ReachView { table: &out, by_row: &by_row, graph };
            accumulate(r, &view, facts, &mut bm);
        }
        by_row[row] = out.len() as u32;
        out.commits.push(graph.ords[row]);
        out.bitmaps.push(bm);
    }
    if out.len() > policy.max_commits.saturating_mul(2) {
        out = out.pruned(graph, policy.max_commits);
    }
    out
}

/// The table, its row index and the graph, paired by the fold that built them
/// so a lookup can never mix a table with another fold's rows.
#[derive(Clone, Copy)]
pub struct ReachView<'a> {
    pub table: &'a ReachTable,
    pub by_row: &'a [u32],
    pub graph: &'a OrdGraph,
}

impl<'a> ReachView<'a> {
    /// The stored bitmap of the commit at `row`, if it has one.
    #[inline]
    pub fn bitmap_of_row(&self, row: u32) -> Option<&'a RoaringBitmap> {
        match self.by_row.get(row as usize).copied() {
            Some(i) if i != NONE => self.table.bitmaps.get(i as usize),
            _ => None,
        }
    }
}

/// **The bounded walk: OR into `acc` everything reachable from the commit at
/// `row`, stopping at every commit that already has a bitmap.**
///
/// A stored bitmap is closed under reachability, so reaching one accounts for
/// the entire history behind it; what the walk covers is only what has been
/// pushed since the table last covered it, whatever the repository's size.
/// This is git's `add_to_include_set` rule.
///
/// The two phases are an ordering, not a style: the commit walk runs to
/// completion before a single tree is opened, so that every stored bitmap the
/// request will OR in is already in `acc` when the tree walk's skip rule runs.
/// Get it wrong and the first unbitmapped commit costs a walk of every tree in
/// the repository.
pub fn accumulate(row: u32, view: &ReachView<'_>, facts: &ObjectFacts<'_>, acc: &mut RoaringBitmap) {
    // ── phase 1: the commit walk, to completion ────────────────────────────
    let mut unbitmapped: Vec<u32> = Vec::new();
    let mut seen = RoaringBitmap::new();
    let mut stack: Vec<u32> = vec![row];
    while let Some(r) = stack.pop() {
        if !seen.insert(r) {
            continue;
        }
        // **Already covered.** `acc` only ever holds closed sets — stored
        // bitmaps, and the results of completed walks — so a commit whose bit
        // is set has its whole history in `acc` and there is nothing to walk.
        // This is what makes a clone's 1 006 wants (most of them tags on old,
        // unbitmapped commits) cost nothing once the branch tips are in: the
        // caller ORs the bitmapped wants first, then walks the rest.
        let o = view.graph.ords[r as usize];
        if o != NONE && acc.contains(o) {
            continue;
        }
        if let Some(bm) = view.bitmap_of_row(r) {
            *acc |= bm;
            continue;
        }
        unbitmapped.push(r);
        if let Some(ps) = view.graph.parents.get(r as usize) {
            stack.extend(ps.iter().copied());
        }
    }
    // ── phase 2: the trees, with the skip rule ─────────────────────────────
    for r in unbitmapped {
        let r = r as usize;
        if view.graph.ords[r] != NONE {
            acc.insert(view.graph.ords[r]);
        }
        if view.graph.tree[r] != NONE {
            fill_tree(view.graph.tree[r], facts, acc);
        }
    }
}

/// Which objects a repack must hold to make the reachability pass exact.
///
/// Only commits and trees carry pointers; blobs are leaves. The builder
/// therefore needs tree payloads and nothing else, which is why
/// [`ObjectFacts::trees`] exists and there is no `blobs` field.
pub fn needs_payload(kind: GitObjectKind) -> bool {
    matches!(kind, GitObjectKind::Commit | GitObjectKind::Tree)
}

/// **The sealed builder**: hex nodes in, hex entries out, over a sorted space.
/// Decodes each node once into an [`OrdGraph`] and runs [`build_reach_table`].
pub fn build_reach(commits: &[CommitNode], facts: &ObjectFacts<'_>, policy: ReachPolicy) -> Vec<ReachEntry> {
    let graph = OrdGraph::from_nodes(commits, &facts.ordinal);
    let table = build_reach_table(&graph, facts, policy);
    table.to_entries(&facts.ordinal, None).unwrap_or_default()
}

pub fn build_reach_batch(entries: &[ReachEntry]) -> Result<RecordBatch> {
    use znippy_common::arrow::array::UInt64Builder;
    let n = entries.len();
    let mut oid_b = StringBuilder::with_capacity(n, n * 64);
    let mut card_b = UInt64Builder::with_capacity(n);
    let mut bm_b = BinaryBuilder::with_capacity(n, n * 128);
    for e in entries {
        oid_b.append_value(&e.commit);
        card_b.append_value(e.bitmap.len());
        let mut buf = Vec::new();
        e.bitmap
            .serialize_into(&mut buf)
            .map_err(|err| anyhow!("reach bitmap serialize: {err}"))?;
        bm_b.append_value(&buf);
    }
    RecordBatch::try_new(
        reach_schema(),
        vec![Arc::new(oid_b.finish()), Arc::new(card_b.finish()), Arc::new(bm_b.finish())],
    )
    .map_err(|e| anyhow!("reach batch: {e}"))
}

pub fn decode_reach(bytes: &[u8]) -> Result<Vec<ReachEntry>> {
    // Through the IPC guard: the section comes out of an archive.
    let reader = znippy_common::ipc_guard::stream_reader(bytes)
        .map_err(|e| anyhow!("reach reader: {e}"))?;
    let mut out = Vec::new();
    for batch in reader {
        let batch = batch.map_err(|e| anyhow!("reach batch read: {e}"))?;
        let oids = batch
            .column_by_name("commit_oid")
            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
            .ok_or_else(|| anyhow!("reach: no `commit_oid` column"))?;
        let bms = batch
            .column_by_name("bitmap")
            .and_then(|c| c.as_any().downcast_ref::<BinaryArray>())
            .ok_or_else(|| anyhow!("reach: no `bitmap` column"))?;
        for i in 0..batch.num_rows() {
            let bitmap = RoaringBitmap::deserialize_from(bms.value(i))
                .map_err(|e| anyhow!("reach bitmap deserialize: {e}"))?;
            out.push(ReachEntry { commit: oids.value(i).to_string(), bitmap });
        }
    }
    Ok(out)
}

/// Read the reachability bitmaps out of a sealed archive. `Ok(None)` when absent.
pub fn read_reach(archive: &Path) -> Result<Option<Vec<ReachEntry>>> {
    match read_reserved_section_bytes(archive, GUNNAR_REACH_MODULE)? {
        Some(b) => Ok(Some(decode_reach(&b)?)),
        None => Ok(None),
    }
}

// ── the live table's persisted form ─────────────────────────────────────────

fn live_schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("commit_ordinal", DataType::UInt32, false),
        Field::new("bitmap", DataType::Binary, false),
    ]))
}

/// The live table as Arrow IPC stream bytes: `(commit_ordinal, bitmap)`.
pub fn encode_live_table(table: &ReachTable) -> Result<Vec<u8>> {
    use znippy_common::arrow::array::UInt32Builder;
    let n = table.len();
    let mut ord_b = UInt32Builder::with_capacity(n);
    let mut bm_b = BinaryBuilder::with_capacity(n, n * 128);
    for (i, &o) in table.commits.iter().enumerate() {
        ord_b.append_value(o);
        let mut buf = Vec::new();
        table.bitmaps[i]
            .serialize_into(&mut buf)
            .map_err(|err| anyhow!("reach bitmap serialize: {err}"))?;
        bm_b.append_value(&buf);
    }
    let batch = RecordBatch::try_new(live_schema(), vec![Arc::new(ord_b.finish()), Arc::new(bm_b.finish())])
        .map_err(|e| anyhow!("live reach batch: {e}"))?;
    let mut buf = Vec::new();
    {
        let mut w = znippy_common::arrow::ipc::writer::StreamWriter::try_new(&mut buf, &live_schema())
            .map_err(|e| anyhow!("live reach writer: {e}"))?;
        w.write(&batch).map_err(|e| anyhow!("live reach write: {e}"))?;
        w.finish().map_err(|e| anyhow!("live reach finish: {e}"))?;
    }
    Ok(buf)
}

/// The inverse of [`encode_live_table`], through the IPC guard: hostile bytes
/// are an `Err`, never a panic or an allocation abort. `space_len` bounds the
/// ordinals: an entry naming one outside the space is refused.
pub fn decode_live_table(bytes: &[u8], space_len: u32) -> Result<ReachTable> {
    use znippy_common::arrow::array::UInt32Array;
    let reader = znippy_common::ipc_guard::stream_reader(bytes).map_err(|e| anyhow!("live reach reader: {e}"))?;
    let mut out = ReachTable::default();
    for batch in reader {
        let batch = batch.map_err(|e| anyhow!("live reach batch read: {e}"))?;
        let ords = batch
            .column_by_name("commit_ordinal")
            .and_then(|c| c.as_any().downcast_ref::<UInt32Array>())
            .ok_or_else(|| anyhow!("live reach: no `commit_ordinal` column"))?;
        let bms = batch
            .column_by_name("bitmap")
            .and_then(|c| c.as_any().downcast_ref::<BinaryArray>())
            .ok_or_else(|| anyhow!("live reach: no `bitmap` column"))?;
        for i in 0..batch.num_rows() {
            let o = ords.value(i);
            if o >= space_len {
                return Err(anyhow!("live reach: commit ordinal {o} is outside the space ({space_len})"));
            }
            let bitmap = RoaringBitmap::deserialize_from(bms.value(i))
                .map_err(|e| anyhow!("live reach bitmap deserialize: {e}"))?;
            if bitmap.max().is_some_and(|m| m >= space_len) {
                return Err(anyhow!("live reach: a bitmap names an ordinal outside the space ({space_len})"));
            }
            out.commits.push(o);
            out.bitmaps.push(bitmap);
        }
    }
    Ok(out)
}

#[cfg(test)]
mod carry_tests {
    use super::*;

    fn space(oids: &[u8]) -> Vec<u8> {
        // 1-byte "oids" are enough for the arithmetic; the space is sorted.
        let mut v = oids.to_vec();
        v.sort_unstable();
        v
    }

    /// The carry renumbers every bit through the remap, drops an entry whose
    /// commit is gone, and refuses when a set bit's object is gone.
    #[test]
    fn a_carry_renumbers_bits_drops_dead_commits_and_refuses_a_missing_object() {
        let old_raw = space(&[10, 20, 30, 40]);
        let old = OrdinalSpace::new(&old_raw, 1);
        // Two oids inserted below 30 and one above 40: ordinals shift by 2 and 2.
        let new_raw = space(&[10, 15, 20, 25, 30, 40, 50]);
        let new = OrdinalSpace::new(&new_raw, 1);
        let entries = vec![
            ReachEntry { commit: "keep".into(), bitmap: [0u32, 2, 3].into_iter().collect() },
            ReachEntry { commit: "gone".into(), bitmap: [1u32].into_iter().collect() },
        ];
        let carried = carry_reach(&entries, &old, &new, |c| c == "keep").expect("exact carry");
        assert_eq!(carried.len(), 1, "the dead commit's entry is dropped, not carried");
        assert_eq!(carried[0].commit, "keep");
        let bits: Vec<u32> = carried[0].bitmap.iter().collect();
        assert_eq!(bits, vec![0, 4, 5], "10→0, 30→4, 40→5 in the new numbering");

        // An object removed from the space under a kept entry: refuse the carry.
        let shrunk_raw = space(&[10, 20, 40]);
        let shrunk = OrdinalSpace::new(&shrunk_raw, 1);
        assert!(carry_reach(&entries, &old, &shrunk, |_| true).is_none(), "30 is gone; a bitmap naming it must not survive");
        // Nothing to carry is not a carry.
        assert!(carry_reach(&[], &old, &new, |_| true).is_none());
    }
}

#[cfg(test)]
mod table_tests {
    use super::*;

    fn oid(i: u32) -> [u8; 20] {
        let mut o = [0u8; 20];
        // Spread the two fanout bytes so the prefix table does real work.
        o[0] = (i.wrapping_mul(2654435761) >> 24) as u8;
        o[1] = (i.wrapping_mul(40503) >> 8) as u8;
        o[16..20].copy_from_slice(&i.to_be_bytes());
        o
    }

    fn sorted_concat(ids: &[u32]) -> Vec<u8> {
        let mut v: Vec<[u8; 20]> = ids.iter().map(|&i| oid(i)).collect();
        v.sort_unstable();
        v.dedup();
        v.concat()
    }

    /// **Stable ordinals**: a follow appends behind the base and moves nothing;
    /// every oid is found both ways through the fanout and the appended index;
    /// a removed oid refuses the follow; the persisted form replays exactly.
    /// Seen RED with `fanout[p+1]` off by one (misses in the last bucket) and
    /// with `extra_sorted` left stale after a follow (appended oids absent).
    #[test]
    fn an_ordinal_table_appends_and_never_renumbers() {
        let first: Vec<u32> = (0..3000).collect();
        let mut t = OrdinalTable::from_sorted(sorted_concat(&first), 20);
        assert_eq!(t.len(), 3000);
        let v = t.view();
        for o in 0..3000u32 {
            let raw = v.oid(o).unwrap();
            assert_eq!(v.ordinal_of_raw(raw), Some(o), "base inverse at {o}");
        }
        let before: Vec<[u8; 20]> = (0..3000u32).map(|o| v.oid(o).unwrap().try_into().unwrap()).collect();

        // Push 1: 7 new oids.
        let second: Vec<u32> = (0..3007).collect();
        let r = t.follow(&sorted_concat(&second)).expect("a superset follows");
        assert_eq!(r, 3000..3007);
        assert_eq!(t.extra_len(), 7);
        // Nothing moved.
        let v = t.view();
        for (o, raw) in before.iter().enumerate() {
            assert_eq!(v.oid(o as u32), Some(raw.as_slice()), "ordinal {o} moved");
            assert_eq!(v.ordinal_of_raw(raw), Some(o as u32));
        }
        for o in 3000..3007u32 {
            let raw = v.oid(o).unwrap();
            assert_eq!(v.ordinal_of_raw(raw), Some(o), "appended inverse at {o}");
        }
        // Push 2: no change is an empty range; then 5 more.
        assert_eq!(t.follow(&sorted_concat(&second)), Some(3007..3007));
        let third: Vec<u32> = (0..3012).collect();
        assert_eq!(t.follow(&sorted_concat(&third)), Some(3007..3012));
        let v = t.view();
        assert_eq!(v.len(), 3012);
        for o in 0..3012u32 {
            assert_eq!(v.ordinal_of_raw(v.oid(o).unwrap()), Some(o));
        }
        assert_eq!(v.ordinal_of_raw(&oid(999_999)), None, "an absent oid is None, not a neighbour");
        assert_eq!(v.ordinal_of_raw(&[0u8; 19]), None);
        assert_eq!(v.oid(3012), None);

        // The persisted form: (full sorted now, extra in arrival order).
        let replay = OrdinalTable::from_parts(&sorted_concat(&third), t.extra_raw(), 20).expect("replays");
        let (a, b) = (t.view(), replay.view());
        for o in 0..3012u32 {
            assert_eq!(a.oid(o), b.oid(o), "replayed ordinal {o} differs");
        }
        // An extra the store no longer holds does not replay.
        let shrunk: Vec<u32> = (0..3011).collect();
        assert!(OrdinalTable::from_parts(&sorted_concat(&shrunk), t.extra_raw(), 20).is_none());

        // A removal under the table refuses the follow (a gc, a retain).
        let mut gone: Vec<u32> = (0..3012).collect();
        gone.remove(1500);
        assert!(t.follow(&sorted_concat(&gone)).is_none(), "an oid the table numbers is gone");
        let mut gone_extra: Vec<u32> = (0..3012).collect();
        gone_extra.remove(3005);
        assert!(t.follow(&sorted_concat(&gone_extra)).is_none(), "an appended oid is gone");

        // The remap to a sorted space over the same oids is a permutation.
        let sorted = sorted_concat(&third);
        let target = OrdinalSpace::new(&sorted, 20);
        let remap = t.remap_to(&target);
        let mut seen: Vec<u32> = remap.clone();
        seen.sort_unstable();
        assert_eq!(seen, (0..3012u32).collect::<Vec<_>>());
        for (o, &n) in remap.iter().enumerate() {
            assert_eq!(t.view().oid(o as u32), target.oid(n));
        }
    }

    /// The live table round-trips through its IPC form, and hostile bytes are
    /// an `Err` (the ipc guard), including an ordinal outside the space.
    #[test]
    fn the_live_table_roundtrips_and_refuses_hostile_bytes() {
        let t = ReachTable {
            commits: vec![5, 9],
            bitmaps: vec![[1u32, 5].into_iter().collect(), [0u32, 1, 5, 9].into_iter().collect()],
        };
        let bytes = encode_live_table(&t).unwrap();
        assert_eq!(decode_live_table(&bytes, 10).unwrap(), t);
        assert!(decode_live_table(&bytes, 9).is_err(), "ordinal 9 is outside a space of 9");
        assert!(decode_live_table(b"not arrow at all, not even close to a stream", 10).is_err());
        let mut torn = bytes.clone();
        torn.truncate(bytes.len() / 2);
        assert!(decode_live_table(&torn, 10).is_err());
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::graph::assign_generations;

    fn hexid(c: char) -> String {
        std::iter::repeat_n(c, 40).collect()
    }

    /// A flat, sorted, fixed-width ordinal buffer from oids **already in
    /// ordinal order** — the shape [`OrdinalSpace`] borrows.
    fn ordinal_raw(oids_in_order: &[&str]) -> Vec<u8> {
        let mut v = Vec::new();
        for o in oids_in_order {
            v.extend_from_slice(&hex::decode(o).unwrap());
        }
        v
    }

    /// Returns (commits, facts-owned-data).
    #[allow(clippy::type_complexity)]
    fn tiny_repo() -> (Vec<CommitNode>, Vec<u8>, HashMap<String, Vec<u8>>) {
        let blob1 = hexid('1');
        let blob2 = hexid('2');
        let tree1 = hexid('3');
        let tree2 = hexid('4');
        let c1 = hexid('5');
        let c2 = hexid('6');

        let mut trees: HashMap<String, Vec<u8>> = HashMap::new();
        let mut t1 = Vec::new();
        t1.extend_from_slice(b"100644 a\0");
        t1.extend_from_slice(&hex::decode(&blob1).unwrap());
        trees.insert(tree1.clone(), t1);

        let mut t2 = Vec::new();
        t2.extend_from_slice(b"100644 a\0");
        t2.extend_from_slice(&hex::decode(&blob1).unwrap());
        t2.extend_from_slice(b"100644 b\0");
        t2.extend_from_slice(&hex::decode(&blob2).unwrap());
        trees.insert(tree2.clone(), t2);

        // Ordinal = oid-lexicographic rank: hexid('1') < … < hexid('6'), so the
        // sorted raw buffer is simply these six in this order.
        let ordinal = ordinal_raw(&[&blob1, &blob2, &tree1, &tree2, &c1, &c2]);

        let commits = assign_generations(vec![
            CommitNode {
                oid: c1.clone(),
                parents: vec![],
                tree: Some(hexid('3')),
                committer_time: Some(1),
                generation: 0,
            },
            CommitNode {
                oid: c2.clone(),
                parents: vec![c1.clone()],
                tree: Some(hexid('4')),
                committer_time: Some(2),
                generation: 0,
            },
        ]);
        (commits, ordinal, trees)
    }

    #[test]
    fn bitmap_contains_exactly_the_reachable_objects() {
        let (commits, ordinal, trees) = tiny_repo();
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let entries = build_reach(&commits, &facts, ReachPolicy { max_commits: 16 });

        let by_commit: HashMap<&str, &RoaringBitmap> =
            entries.iter().map(|e| (e.commit.as_str(), &e.bitmap)).collect();

        // c1 reaches: blob1(0), tree1(2), c1(4).
        let b1: Vec<u32> = by_commit[hexid('5').as_str()].iter().collect();
        assert_eq!(b1, vec![0, 2, 4], "c1 must not reach blob2/tree2/c2");

        // c2 reaches everything.
        let b2: Vec<u32> = by_commit[hexid('6').as_str()].iter().collect();
        assert_eq!(b2, vec![0, 1, 2, 3, 4, 5]);
    }

    #[test]
    fn want_minus_have_is_an_andnot() {
        let (commits, ordinal, trees) = tiny_repo();
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let entries = build_reach(&commits, &facts, ReachPolicy { max_commits: 16 });
        let by: HashMap<&str, &RoaringBitmap> =
            entries.iter().map(|e| (e.commit.as_str(), &e.bitmap)).collect();

        let want = by[hexid('6').as_str()].clone();
        let have = by[hexid('5').as_str()].clone();
        let delta: Vec<u32> = (want - have).iter().collect();
        // Exactly the objects the second commit introduced: blob2(1), tree2(3), c2(5).
        assert_eq!(delta, vec![1, 3, 5]);
    }

    #[test]
    fn selection_takes_tips_first_and_respects_the_cap() {
        let (commits, ordinal, trees) = tiny_repo();
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let entries = build_reach(&commits, &facts, ReachPolicy { max_commits: 1 });
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].commit, hexid('6'), "the tip, not the root");
    }

    #[test]
    fn batch_roundtrips_through_arrow_ipc() {
        let (commits, ordinal, trees) = tiny_repo();
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let entries = build_reach(&commits, &facts, ReachPolicy::default());
        let batch = build_reach_batch(&entries).unwrap();
        let mut buf = Vec::new();
        {
            let mut w = znippy_common::arrow::ipc::writer::StreamWriter::try_new(
                &mut buf,
                &reach_schema(),
            )
            .unwrap();
            w.write(&batch).unwrap();
            w.finish().unwrap();
        }
        let back = decode_reach(&buf).unwrap();
        assert_eq!(back, entries);
    }

    #[test]
    fn a_corrupt_tree_cycle_terminates() {
        // t -> t (self-referential): must not hang or blow the stack.
        let t = hexid('a');
        let mut payload = Vec::new();
        payload.extend_from_slice(b"40000 self\0");
        payload.extend_from_slice(&hex::decode(&t).unwrap());
        let trees: HashMap<String, Vec<u8>> = [(t.clone(), payload)].into_iter().collect();
        let ordinal = ordinal_raw(&[&t, &hexid('b')]);
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let commits = assign_generations(vec![CommitNode {
            oid: hexid('b'),
            parents: vec![],
            tree: Some(t),
            committer_time: None,
            generation: 0,
        }]);
        let entries = build_reach(&commits, &facts, ReachPolicy::default());
        assert_eq!(entries.len(), 1);
        let bits: Vec<u32> = entries[0].bitmap.iter().collect();
        assert_eq!(bits, vec![0, 1]);
    }

    /// **The binary search is the exact inverse of the slice, and it is the same
    /// answer a `HashMap<String, u32>` over the sorted list would give.**
    #[test]
    fn ordinal_space_is_the_inverse_of_the_slice() {
        // 32 distinct sha1-width oids, deliberately built OUT of order so the
        // sort is doing real work, then sorted into the flat space.
        let mut oids: Vec<[u8; 20]> = (0..32u8)
            .map(|i| {
                let mut o = [0u8; 20];
                o[0] = i.wrapping_mul(37);
                o[19] = i;
                o
            })
            .collect();
        oids.sort_unstable();
        oids.dedup();
        let mut raw = Vec::new();
        for o in &oids {
            raw.extend_from_slice(o);
        }
        let space = OrdinalSpace::new(&raw, 20);
        assert_eq!(space.len(), oids.len());
        for (o, oid) in oids.iter().enumerate() {
            assert_eq!(space.oid(o as u32), Some(oid.as_slice()));
            assert_eq!(space.ordinal_of_raw(oid), Some(o as u32));
            assert_eq!(space.ordinal_of_hex(&hex::encode(oid)), Some(o as u32));
        }
        // Absent, wrong width, and out of range are all None — not a neighbour.
        assert_eq!(space.ordinal_of_raw(&[0xffu8; 20]), None);
        assert_eq!(space.ordinal_of_raw(&[0u8; 19]), None);
        assert_eq!(space.oid(oids.len() as u32), None);
        assert_eq!(space.ordinal_of_hex("not-hex"), None);
    }

    /// **The view owns nothing.** `OrdinalSpace` is a `Copy` view of borrowed
    /// slices, so building one at open allocates zero bytes — the reason the
    /// hex `Vec<String>` and the `HashMap<String, u32>` were removed. It grew
    /// from one slice to four (the fanout and the appended part came with the
    /// stable live space); red if an owned field creeps in or `Copy` is lost.
    #[test]
    fn ordinal_space_is_a_borrowed_view() {
        assert_eq!(
            std::mem::size_of::<OrdinalSpace<'_>>(),
            4 * std::mem::size_of::<&[u8]>() + std::mem::size_of::<usize>(),
            "OrdinalSpace grew an owned field — it must stay a borrow"
        );
        fn assert_copy<T: Copy>() {}
        assert_copy::<OrdinalSpace<'_>>();
    }

    // ── property tests: bitmap = graph walk, on random DAGs ─────────────────

    /// A deterministic xorshift, so a failing seed is reproducible.
    struct Rng(u64);
    impl Rng {
        fn next(&mut self) -> u64 {
            let mut x = self.0;
            x ^= x << 13;
            x ^= x >> 7;
            x ^= x << 17;
            self.0 = x;
            x
        }
        fn below(&mut self, n: usize) -> usize {
            (self.next() % n.max(1) as u64) as usize
        }
    }

    /// A synthetic repository with a real object graph: nested trees, shared
    /// blobs, merges and unrelated roots.
    struct Dag {
        nodes: Vec<CommitNode>,
        /// Every object: oid → (kind, payload). Trees and commits carry
        /// pointers; blobs are leaves.
        objects: HashMap<Vec<u8>, (GitObjectKind, Vec<u8>)>,
        sorted: Vec<u8>,
        trees: HashMap<Vec<u8>, Vec<u8>>,
    }

    fn fake_oid(rng: &mut Rng) -> Vec<u8> {
        let mut o = vec![0u8; 20];
        for b in o.iter_mut() {
            *b = rng.next() as u8;
        }
        o
    }

    fn random_dag(seed: u64, commits: usize) -> Dag {
        let mut rng = Rng(seed | 1);
        let mut objects: HashMap<Vec<u8>, (GitObjectKind, Vec<u8>)> = HashMap::new();
        let mut blobs: Vec<Vec<u8>> = Vec::new();
        let mut subtrees: Vec<Vec<u8>> = Vec::new();
        let mut trees: HashMap<Vec<u8>, Vec<u8>> = HashMap::new();
        let new_tree = |rng: &mut Rng,
                            objects: &mut HashMap<Vec<u8>, (GitObjectKind, Vec<u8>)>,
                            blobs: &mut Vec<Vec<u8>>,
                            subtrees: &[Vec<u8>],
                            trees: &mut HashMap<Vec<u8>, Vec<u8>>|
         -> Vec<u8> {
            let mut payload = Vec::new();
            let entries = 1 + rng.below(4);
            for e in 0..entries {
                // Reuse a blob 50 % of the time, else make one.
                let blob = if !blobs.is_empty() && rng.below(2) == 0 {
                    blobs[rng.below(blobs.len())].clone()
                } else {
                    let b = fake_oid(rng);
                    objects.insert(b.clone(), (GitObjectKind::Blob, vec![e as u8]));
                    blobs.push(b.clone());
                    b
                };
                payload.extend_from_slice(format!("100644 f{e}\0").as_bytes());
                payload.extend_from_slice(&blob);
            }
            if !subtrees.is_empty() && rng.below(3) == 0 {
                let st = &subtrees[rng.below(subtrees.len())];
                payload.extend_from_slice(b"40000 d\0");
                payload.extend_from_slice(st);
            }
            if rng.below(7) == 0 {
                // A gitlink: a commit in another repository, not in this space.
                payload.extend_from_slice(b"160000 sub\0");
                payload.extend_from_slice(&fake_oid(rng));
            }
            let t = fake_oid(rng);
            objects.insert(t.clone(), (GitObjectKind::Tree, payload.clone()));
            trees.insert(t.clone(), payload);
            t
        };
        let mut nodes: Vec<CommitNode> = Vec::new();
        for i in 0..commits {
            let tree = new_tree(&mut rng, &mut objects, &mut blobs, &subtrees, &mut trees);
            if rng.below(2) == 0 {
                subtrees.push(tree.clone());
            }
            let mut parents = Vec::new();
            if i > 0 && rng.below(5) != 0 {
                parents.push(nodes[rng.below(i)].oid.clone());
                if i > 1 && rng.below(4) == 0 {
                    parents.push(nodes[rng.below(i)].oid.clone());
                }
            }
            let oid = fake_oid(&mut rng);
            let mut payload = format!("tree {}\n", hex::encode(&tree)).into_bytes();
            for p in &parents {
                payload.extend_from_slice(format!("parent {p}\n").as_bytes());
            }
            objects.insert(oid.clone(), (GitObjectKind::Commit, payload));
            nodes.push(CommitNode {
                oid: hex::encode(&oid),
                parents,
                tree: Some(hex::encode(&tree)),
                committer_time: Some(i as i64),
                generation: 0,
            });
        }
        let mut all: Vec<&Vec<u8>> = objects.keys().collect();
        all.sort_unstable();
        let sorted: Vec<u8> = all.iter().flat_map(|o| o.iter().copied()).collect();
        Dag { nodes: assign_generations(nodes), objects, sorted, trees }
    }

    /// The oracle: a plain object walk from `tip`, over the object map.
    fn walk_closure(dag: &Dag, tip: &[u8]) -> std::collections::BTreeSet<Vec<u8>> {
        let mut seen = std::collections::BTreeSet::new();
        let mut stack = vec![tip.to_vec()];
        while let Some(o) = stack.pop() {
            let Some((kind, payload)) = dag.objects.get(&o) else { continue };
            if !seen.insert(o.clone()) {
                continue;
            }
            match kind {
                GitObjectKind::Commit => {
                    for line in payload.split(|b| *b == b'\n') {
                        if let Some(h) = line.strip_prefix(b"tree ").or_else(|| line.strip_prefix(b"parent ")) {
                            stack.push(hex::decode(h).unwrap());
                        }
                    }
                }
                GitObjectKind::Tree => {
                    for e in tree_entries(payload, 20) {
                        stack.push(e.oid.to_vec());
                    }
                }
                _ => {}
            }
        }
        seen
    }

    fn bits_as_oids(bm: &RoaringBitmap, space: &OrdinalSpace<'_>) -> std::collections::BTreeSet<Vec<u8>> {
        bm.iter().map(|o| space.oid(o).unwrap().to_vec()).collect()
    }

    /// **Every stored bitmap equals the graph walk, at every cap, on random
    /// DAGs** — the full builder, and the walk-covered answer for every commit
    /// with or without an entry.
    #[test]
    fn the_full_table_equals_a_graph_walk_on_random_dags() {
        for seed in 1..=24u64 {
            let n = 1 + (seed as usize * 7) % 60;
            let dag = random_dag(seed, n);
            let space = OrdinalSpace::new(&dag.sorted, 20);
            let facts = ObjectFacts { ordinal: space, trees: &dag.trees, oid_len: 20 };
            let graph = OrdGraph::from_nodes(&dag.nodes, &space);
            for cap in [usize::MAX, 1, 3, 7] {
                let table = build_reach_table(&graph, &facts, ReachPolicy { max_commits: cap });
                let by_row = table.by_row(&graph);
                let view = ReachView { table: &table, by_row: &by_row, graph: &graph };
                for (i, &o) in table.commits.iter().enumerate() {
                    let oracle = walk_closure(&dag, space.oid(o).unwrap());
                    assert_eq!(bits_as_oids(&table.bitmaps[i], &space), oracle, "seed {seed} cap {cap}: stored entry {i}");
                }
                // Every commit, bitmapped or not, through the bounded walk.
                for row in 0..graph.len() as u32 {
                    let mut acc = RoaringBitmap::new();
                    accumulate(row, &view, &facts, &mut acc);
                    let oracle = walk_closure(&dag, space.oid(graph.ords[row as usize]).unwrap());
                    assert_eq!(bits_as_oids(&acc, &space), oracle, "seed {seed} cap {cap}: walk from row {row}");
                }
            }
        }
    }

    /// **A table extended after a "push" equals a table built from scratch**,
    /// on random DAGs split into an old and a new part, over a stable
    /// (appended) ordinal space: entries for the new tips are exact, old
    /// entries are untouched, and the walk from every commit still equals the
    /// oracle.
    #[test]
    fn an_extended_table_equals_a_graph_walk_on_random_dags() {
        for seed in 1..=24u64 {
            let n = 4 + (seed as usize * 11) % 70;
            let dag = random_dag(seed * 7919, n);
            // The old part: the first k commits (in generation order they are
            // closed under parents only if we take a prefix of the ORIGINAL
            // creation order, so take the objects reachable from the first k
            // tips of creation order instead).
            let k = 1 + (seed as usize) % (n - 1);
            let old_nodes: Vec<CommitNode> = dag.nodes.iter().filter(|c| c.committer_time.unwrap() < k as i64).cloned().collect();
            let mut old_objs: std::collections::BTreeSet<Vec<u8>> = std::collections::BTreeSet::new();
            for c in &old_nodes {
                old_objs.extend(walk_closure(&dag, &hex::decode(&c.oid).unwrap()));
            }
            let old_sorted: Vec<u8> = old_objs.iter().flat_map(|o| o.iter().copied()).collect();
            let mut table_space = OrdinalTable::from_sorted(old_sorted, 20);
            let old_graph = OrdGraph::from_nodes(&old_nodes, &table_space.view());
            let facts_old = ObjectFacts { ordinal: table_space.view(), trees: &dag.trees, oid_len: 20 };
            let policy = ReachPolicy { max_commits: 5 };
            let old_table = build_reach_table(&old_graph, &facts_old, policy);

            // The push: the rest arrives; the space appends.
            let appended = table_space.follow(&dag.sorted).expect("a superset follows");
            let space = table_space.view();
            let graph = OrdGraph::from_nodes(&dag.nodes, &space);
            let facts = ObjectFacts { ordinal: space, trees: &dag.trees, oid_len: 20 };
            let new_rows: Vec<u32> = (0..graph.len() as u32).filter(|&r| graph.ords[r as usize] >= appended.start).collect();
            let table = extend_reach_table(&old_table, &graph, &facts, &new_rows, policy);

            // Old entries untouched (an entry may be PRUNED once the table
            // outgrows twice the cap — never rewritten), new tips covered.
            assert!(table.len() <= 2 * policy.max_commits, "seed {seed}: the table outgrew twice the cap");
            for (i, &o) in old_table.commits.iter().enumerate() {
                if let Some(j) = table.commits.iter().position(|&c| c == o) {
                    assert_eq!(table.bitmaps[j], old_table.bitmaps[i], "seed {seed}: old entry rewritten");
                }
            }
            let by_row = table.by_row(&graph);
            // Every new tip has an entry, unless the cap forced a prune (then
            // at least `max` entries stand and the walk covers the rest).
            for r in graph.tip_rows() {
                if new_rows.contains(&r) && table.len() < policy.max_commits {
                    assert_ne!(by_row[r as usize], NONE, "seed {seed}: new tip row {r} has no entry");
                }
            }
            let view = ReachView { table: &table, by_row: &by_row, graph: &graph };
            for (i, &o) in table.commits.iter().enumerate() {
                let oracle = walk_closure(&dag, space.oid(o).unwrap());
                assert_eq!(bits_as_oids(&table.bitmaps[i], &space), oracle, "seed {seed}: extended entry {i}");
            }
            for row in 0..graph.len() as u32 {
                let mut acc = RoaringBitmap::new();
                accumulate(row, &view, &facts, &mut acc);
                let oracle = walk_closure(&dag, space.oid(graph.ords[row as usize]).unwrap());
                assert_eq!(bits_as_oids(&acc, &space), oracle, "seed {seed}: walk from row {row} after the extension");
            }
            // And the sealed spelling, renumbered into the sorted space, equals
            // a sealed build over the same nodes.
            let sorted_space = OrdinalSpace::new(&dag.sorted, 20);
            let remap = table_space.remap_to(&sorted_space);
            let sealed = table.to_entries(&sorted_space, Some(&remap)).expect("exact remap");
            for e in &sealed {
                let oracle = walk_closure(&dag, &hex::decode(&e.commit).unwrap());
                assert_eq!(bits_as_oids(&e.bitmap, &sorted_space), oracle, "seed {seed}: sealed entry {}", e.commit);
            }
        }
    }
}