znippy-plugin-git 0.1.1

//! `__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**.
//!
//! The ordinal space is the archive's oid-lexicographic ordering of its distinct
//! objects, which is also 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 to an object with the oid index alone.
//!
//! What it buys: `want − have` becomes `reach(want) ANDNOT reach(have)` — a
//! bitmap operation instead of a graph traversal. **No speedup is claimed here.**
//! Nothing in this repository has measured it against a real repository with real
//! merge history; the structure is built and correct, the number is not earned
//! (LAW 7).
//!
//! Cost, stated plainly: the builder is O(commits × bitmap width) in time and
//! holds live bitmaps for commits whose children are not yet processed, plus one
//! memoized tree-closure bitmap per distinct tree. On a wide DAG that is not
//! cheap. `ReachPolicy::max_commits` bounds how many bitmaps are *kept*, not how
//! many are computed.

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::ipc::reader::StreamReader;
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_common::read_reserved_section_bytes;

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

/// 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.
#[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),
    ]))
}

/// What the builder needs to know about every object in the archive.
pub struct ObjectFacts<'a> {
    /// Object ordinal for each oid hex — the bitmap space.
    pub ordinal: &'a HashMap<String, u32>,
    /// Payload of every tree object, by oid hex. Blobs are not needed (they are
    /// leaves) and are deliberately not held.
    pub trees: &'a HashMap<String, Vec<u8>>,
    /// Raw oid width, needed to walk tree records.
    pub oid_len: usize,
}

/// Build the reachability bitmaps.
///
/// `commits` must already carry generation numbers (see
/// [`crate::graph::assign_generations`]) and be ordered parents-before-children,
/// which that function guarantees.
pub fn build_reach(
    commits: &[CommitNode],
    facts: &ObjectFacts<'_>,
    policy: ReachPolicy,
) -> Vec<ReachEntry> {
    let n = commits.len();
    if n == 0 || policy.max_commits == 0 {
        return Vec::new();
    }
    let pos: HashMap<&str, usize> = commits
        .iter()
        .enumerate()
        .map(|(i, c)| (c.oid.as_str(), i))
        .collect();

    let selected = select_commits(commits, &pos, policy);
    let keep: Vec<bool> = {
        let mut k = vec![false; n];
        for &i in &selected {
            k[i] = true;
        }
        k
    };

    // How many children still need each commit's bitmap. Once it hits zero and
    // the commit is not selected, the bitmap is dropped — that is what keeps
    // peak memory to the DAG's frontier rather than the whole history.
    let mut pending: Vec<usize> = vec![0; n];
    let mut parent_idx: Vec<Vec<usize>> = Vec::with_capacity(n);
    for c in commits {
        let mut ps: Vec<usize> = c.parents.iter().filter_map(|p| pos.get(p.as_str()).copied()).collect();
        ps.sort_unstable();
        ps.dedup();
        for &p in &ps {
            pending[p] += 1;
        }
        parent_idx.push(ps);
    }

    let mut tree_memo: HashMap<String, RoaringBitmap> = HashMap::new();
    let mut live: HashMap<usize, RoaringBitmap> = HashMap::new();

    for i in 0..n {
        let c = &commits[i];
        let mut bm = RoaringBitmap::new();
        if let Some(&o) = facts.ordinal.get(c.oid.as_str()) {
            bm.insert(o);
        }
        if let Some(t) = &c.tree {
            // Borrowed out of the memo, not cloned out of it. See
            // [`tree_closure`]'s own note: this is once per commit and the thing
            // being copied is a whole tree closure, which on a repository-shaped
            // history is most of the object graph.
            if let Some(tc) = tree_closure(t, facts, &mut tree_memo) {
                bm |= tc;
            }
        }
        for &p in &parent_idx[i] {
            if let Some(pb) = live.get(&p) {
                bm |= pb;
            }
            pending[p] -= 1;
            if pending[p] == 0 && !keep[p] {
                live.remove(&p);
            }
        }
        // **Nothing is cloned here.** A kept commit's bitmap is inserted into
        // `live` like any other and harvested below; the `out.push(ReachEntry {
        // bitmap: bm.clone() })` that used to stand here made a second, deep
        // copy of every selected commit's roaring bitmap while `live` held the
        // first — and on the live selection path `keep` is *every* commit, so
        // that was one full duplicate of the entire index. MEASURED on oden
        // 2026-08-14, a 1 581-commit / 12 455-object store: the whole build was
        // 530 218 allocations.
        //
        // Safe by the retention rule directly above: an entry is dropped from
        // `live` only when `pending[p] == 0 && !keep[p]`, so a kept one is never
        // dropped and is still there when the harvest runs.
        if pending[i] > 0 || keep[i] {
            live.insert(i, bm);
        }
    }

    // The harvest: **move** each selected commit's bitmap out of `live`.
    //
    // `selected` is sorted ascending (`select_commits` sorts and dedups), and
    // the loop above pushed in `0..n` order filtered by `keep`, so this produces
    // the identical order the clone-per-iteration version did. That is not a
    // detail to leave implicit — `decode_reach`/`build_reach_batch` round-trip
    // this vector positionally.
    let mut out: Vec<ReachEntry> = Vec::with_capacity(selected.len());
    for &i in &selected {
        // `expect` rather than `unwrap_or_default`: a missing entry here means
        // the retention rule above let a kept commit be dropped, and answering
        // with an EMPTY bitmap would be a silent under-send — a clone served
        // one object per branch, exiting zero. Panicking names the commit.
        let bitmap = live.remove(&i).unwrap_or_else(|| {
            panic!(
                "commit {} was selected for a reachability bitmap and its bitmap is not live at \
                 the end of the build; an empty one here would under-send a clone",
                commits[i].oid
            )
        });
        out.push(ReachEntry {
            commit: commits[i].oid.clone(),
            bitmap,
        });
    }
    out
}

/// Tips first, then an even sample through the rest, capped at `max_commits`.
fn select_commits(
    commits: &[CommitNode],
    pos: &HashMap<&str, usize>,
    policy: ReachPolicy,
) -> Vec<usize> {
    let n = commits.len();
    let mut has_child = vec![false; n];
    for c in commits {
        for p in &c.parents {
            if let Some(&pi) = pos.get(p.as_str()) {
                has_child[pi] = true;
            }
        }
    }
    let mut chosen: Vec<usize> = (0..n).filter(|&i| !has_child[i]).collect();
    chosen.truncate(policy.max_commits);

    if chosen.len() < policy.max_commits {
        let room = policy.max_commits - chosen.len();
        let rest: Vec<usize> = (0..n).filter(|&i| has_child[i]).collect();
        if !rest.is_empty() {
            let stride = rest.len().div_ceil(room).max(1);
            for &i in rest.iter().step_by(stride).take(room) {
                chosen.push(i);
            }
        }
    }
    chosen.sort_unstable();
    chosen.dedup();
    chosen
}

/// Every object reachable from a tree, memoized per tree oid.
///
/// Iterative (an explicit post-order stack), not recursive: a deep source tree
/// would otherwise be a stack-depth bet.
///
/// # It **borrows** out of the memo rather than cloning out of it
///
/// It returned `RoaringBitmap` by value until 2026-08-14, which meant a deep
/// copy of a whole tree closure on every call — including, and especially, the
/// memo *hit* on the first line, which is the common case: consecutive commits
/// in a history overwhelmingly share a root tree or find one already computed.
/// The thing being copied is not small. On a repository-shaped history a root
/// tree's closure is most of the object graph, so this was one full copy of it
/// per commit, 1 581 times on the store this was measured against.
///
/// `None` means the memo has no entry for `root` even after the walk, which the
/// walk only produces for an oid this archive does not hold. The caller ORs
/// nothing in, which is what `unwrap_or_default()` did and is the same answer —
/// it is `Option` rather than an empty bitmap so that "absent" and "reaches
/// nothing" stay different words at the call site.
pub(crate) fn tree_closure<'m>(
    root: &str,
    facts: &ObjectFacts<'_>,
    memo: &'m mut HashMap<String, RoaringBitmap>,
) -> Option<&'m RoaringBitmap> {
    if memo.contains_key(root) {
        // Deliberately re-looked-up at the bottom rather than returned from
        // here: NLL cannot see that the early borrow ends on this path, and the
        // alternative is a second `HashMap` probe on the hit path against a deep
        // copy of the whole closure. The probe wins by orders of magnitude.
        return memo.get(root);
    }
    // (oid, children_already_expanded)
    let mut stack: Vec<(String, bool)> = vec![(root.to_string(), false)];
    // Trees currently on the stack. A tree graph cannot contain a cycle, but a
    // *corrupt* one can, and "cannot happen" is not a termination argument for
    // bytes that arrived over the wire.
    let mut visiting: std::collections::HashSet<String> = std::collections::HashSet::new();

    while let Some((oid, expanded)) = stack.pop() {
        if memo.contains_key(&oid) {
            visiting.remove(&oid);
            continue;
        }
        let Some(payload) = facts.trees.get(&oid) else {
            // Not a tree we hold (a blob, or an object outside the archive).
            let mut bm = RoaringBitmap::new();
            if let Some(&o) = facts.ordinal.get(oid.as_str()) {
                bm.insert(o);
            }
            memo.insert(oid, bm);
            continue;
        };
        let children: Vec<String> = tree_entries(payload, facts.oid_len)
            .iter()
            .map(|e| hex::encode(e.oid))
            .collect();
        if !expanded {
            visiting.insert(oid.clone());
            // A child that is already `visiting` is a back-edge: do not push it
            // again, or the stack never drains.
            let unresolved: Vec<String> = children
                .iter()
                .filter(|c| !memo.contains_key(*c) && !visiting.contains(*c))
                .cloned()
                .collect();
            if !unresolved.is_empty() {
                stack.push((oid, true));
                for c in unresolved {
                    stack.push((c, false));
                }
                continue;
            }
        }
        let mut bm = RoaringBitmap::new();
        if let Some(&o) = facts.ordinal.get(oid.as_str()) {
            bm.insert(o);
        }
        for c in &children {
            if let Some(cb) = memo.get(c) {
                bm |= cb;
            } else if let Some(&o) = facts.ordinal.get(c.as_str()) {
                // Back-edge or an object outside the archive: take the object
                // itself and stop rather than loop forever.
                bm.insert(o);
            }
        }
        memo.insert(oid.clone(), bm);
        visiting.remove(&oid);
    }
    memo.get(root)
}

/// **The bounded walk: OR into `acc` everything reachable from `tip`, stopping
/// at every commit that already has a bitmap.**
///
/// # Why this function is what makes a sampled bitmap table legal
///
/// Until 2026-08-14 this file had no walk at all, and the consequence ran all
/// the way up the stack. [`crate::git_ops::GitStore::reachable_oids`] answered a
/// `want` with no bitmap by contributing *the object itself and nothing else* —
/// a silent under-send, a clone that exits zero having served one object per
/// branch. `crate::serve`'s `select` therefore had to refuse any request whose
/// tips were not all bitmapped, and the only way to make that refusal never fire
/// was to bitmap **every** commit: `ReachPolicy { max_commits: usize::MAX }`, on
/// the request thread, holding the derived write lock, after every push (because
/// `refold` clears the table). MEASURED: 530 218 of a fetch's 575 349
/// allocations were that build, for a request that sent four objects.
///
/// With a walk, a missing bitmap is a *bounded amount of work* instead of a
/// wrong answer, and the table can be sampled like the sealed archive's already
/// is.
///
/// # Why it is bounded, stated as the property it rests on
///
/// **A stored bitmap is closed under reachability.** So reaching one accounts
/// for the entire history behind it, and the walk stops there rather than
/// descending. What it covers is therefore only what has been pushed since the
/// last time the table was built — not the repository — whatever the repository's
/// size. This is git's `add_to_include_set` rule, arrived at independently here
/// over a different object store: nothing of gitoxide's or of git's is copied,
/// and this crate links neither (LAW 5 applies to code, and these are different
/// codebases with different indexes).
///
/// # The two phases are an ordering, not a style
///
/// The commit walk runs to **completion** before a single tree is opened. Git
/// applies its skip rule one level down as well — `should_include_obj`, do not
/// descend into a tree whose bit is already set — and that rule is worth nothing
/// unless every stored bitmap this request will ever OR in is already in `acc`
/// when the tree walk starts. Get the order wrong and the first unbitmapped
/// commit costs a walk of every tree and blob in the repository, and the bounded
/// commit walk buys exactly nothing.
pub(crate) fn accumulate(
    tip: &str,
    by_commit: &HashMap<&str, &RoaringBitmap>,
    graph: &HashMap<&str, &CommitNode>,
    facts: &ObjectFacts<'_>,
    acc: &mut RoaringBitmap,
) {
    // ── phase 1: the commit walk, to completion ────────────────────────────
    //
    // Collect the commits that have no bitmap, ORing in every stored bitmap the
    // frontier touches. Nothing here opens a tree.
    let mut unbitmapped: Vec<&str> = Vec::new();
    let mut seen: std::collections::HashSet<&str> = std::collections::HashSet::new();
    let mut stack: Vec<&str> = vec![tip];
    while let Some(oid) = stack.pop() {
        if !seen.insert(oid) {
            continue;
        }
        // The stopping rule. A stored bitmap is closed, so everything behind
        // this commit is already accounted for and its parents are not walked.
        if let Some(bm) = by_commit.get(oid) {
            *acc |= *bm;
            continue;
        }
        let Some(node) = graph.get(oid) else {
            // Not a commit this archive holds as a graph row. It contributes
            // itself if the index knows it, and nothing else — there is no
            // history here to walk. A `want` that reaches this and is genuinely
            // absent is refused by the caller, which is the one place that can
            // tell "absent" from "not a commit".
            if let Some(&o) = facts.ordinal.get(oid) {
                acc.insert(o);
            }
            continue;
        };
        unbitmapped.push(oid);
        for p in &node.parents {
            stack.push(p.as_str());
        }
    }

    // ── phase 2: the trees, with the skip rule ─────────────────────────────
    for oid in unbitmapped {
        if let Some(&o) = facts.ordinal.get(oid) {
            acc.insert(o);
        }
        let Some(node) = graph.get(oid) else { continue };
        if let Some(tree) = &node.tree {
            accumulate_tree(tree.as_str(), facts, acc);
        }
    }
}

/// [`accumulate`]'s second phase for one root tree: every object under it that
/// is not already in `acc`.
///
/// The skip is git's `should_include_obj` and it is what keeps phase 2 bounded:
/// a set bit means that object — and, for a tree, everything under it — is
/// already accounted for, either by a stored bitmap ORed in during phase 1 or by
/// an earlier commit in this same walk. Descending anyway would be correct and
/// would also make the whole exercise pointless.
///
/// Iterative for the reason [`tree_closure`] is: a deep source tree is not a
/// stack-depth bet. It needs no `visiting` set of its own — a corrupt cycle
/// terminates here because a revisited oid's bit is already set.
fn accumulate_tree(root: &str, facts: &ObjectFacts<'_>, acc: &mut RoaringBitmap) {
    let mut stack: Vec<String> = vec![root.to_string()];
    while let Some(oid) = stack.pop() {
        let Some(&o) = facts.ordinal.get(oid.as_str()) else {
            // An object this archive does not hold. A thin push's base, or a
            // submodule gitlink, which is a commit oid that lives in another
            // repository entirely and must never be selected for emission.
            continue;
        };
        if !acc.insert(o) {
            // Already set: this object, and everything under it, is accounted
            // for. `RoaringBitmap::insert` returns false when the bit was
            // already present, so the test and the set are one operation.
            continue;
        }
        // Only trees carry pointers. A blob is a leaf and `facts.trees`
        // deliberately does not hold one.
        if let Some(payload) = facts.trees.get(oid.as_str()) {
            for e in tree_entries(payload, facts.oid_len) {
                stack.push(hex::encode(e.oid));
            }
        }
    }
}

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

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>> {
    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)
        .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),
    }
}

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

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

    /// Build a tiny repo: two commits, the second adding a file.
    /// Returns (commits, facts-owned-data).
    #[allow(clippy::type_complexity)]
    fn tiny_repo() -> (Vec<CommitNode>, HashMap<String, u32>, 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);

        let ordinal: HashMap<String, u32> = [
            (blob1, 0u32),
            (blob2, 1),
            (tree1, 2),
            (tree2, 3),
            (c1.clone(), 4),
            (c2.clone(), 5),
        ]
        .into_iter()
        .collect();

        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: &ordinal, 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: &ordinal, 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: &ordinal, 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: &ordinal, 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: HashMap<String, u32> = [(t.clone(), 0u32), (hexid('b'), 1)].into_iter().collect();
        let facts = ObjectFacts { ordinal: &ordinal, 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]);
    }
}