znippy-plugin-git 0.1.0

//! The measured workloads behind the `znippy.git_*` benchmarks.
//!
//! These live in the crate rather than in `xtask/examples/nornir-bench.rs`
//! because of a hard constraint recorded in `CLAUDE.md`: `xtask` dev-depends on
//! nornir, nornir path-depends back into the canonical znippy checkout, and the
//! resulting lockfile collision means **no worktree can build xtask at all**. So
//! the workload lives where it can be compiled and run from anywhere, and the
//! registered `Bencher` impls are thin callers of these functions — one
//! implementation, two entry points (LAW 5), rather than a bench body that only
//! one checkout on the machine can execute.
//!
//! Every function here returns *measured* numbers or an error. None of them
//! returns a ratio it did not compute from two timings it took itself, and each
//! one that compares two strategies **asserts the two agree on the answer**
//! first — a faster path that returns a different set is not faster, it is
//! wrong (LAW 2).
//!
//! Unit convention follows nornir's: `_ns` is reciprocal throughput, ns per
//! operation, and renders "lower is better".

use std::collections::{HashMap, HashSet};
use std::time::Instant;

use anyhow::{Result, ensure};
use roaring::RoaringBitmap;

use crate::graph::{CommitNode, assign_generations};
use crate::object::{GitHashKind, GitObjectKind, canonical};
use crate::oid_index::{GitOidIndex, OidEntry, build_section, key_for_oid};
use crate::reach::{ObjectFacts, ReachEntry, ReachPolicy, build_reach, tree_closure};
use crate::sections::GitIndexBuilder;

// ── deterministic pseudo-random bytes ─────────────────────────────────────────

/// splitmix64 — seeded so every run of a bench is the same run, which is what
/// makes a historized series comparable at all.
struct Rng(u64);

impl Rng {
    fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
        let mut z = self.0;
        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
        z ^ (z >> 31)
    }

    fn fill(&mut self, out: &mut [u8]) {
        for c in out.chunks_mut(8) {
            let w = self.next().to_le_bytes();
            let n = c.len();
            c.copy_from_slice(&w[..n]);
        }
    }
}

/// `n` distinct uniformly random oids of `oid_len` bytes.
///
/// Uniformly random is not a convenience here, it is the property under test:
/// the whole argument for stree over the stock fst trie is that oids share no
/// prefixes, so a generator with structure would measure the wrong thing.
fn random_oids(n: usize, oid_len: usize, seed: u64) -> Vec<Vec<u8>> {
    let mut rng = Rng(seed);
    let mut seen: HashSet<Vec<u8>> = HashSet::with_capacity(n * 2);
    let mut out = Vec::with_capacity(n);
    while out.len() < n {
        let mut o = vec![0u8; oid_len];
        rng.fill(&mut o);
        if seen.insert(o.clone()) {
            out.push(o);
        }
    }
    out
}

/// Median of `reps` runs, in nanoseconds. Median rather than mean because one
/// scheduler hiccup in a five-run sample moves a mean and does not move a median.
fn time_ns<F: FnMut()>(reps: usize, mut f: F) -> f64 {
    let mut samples: Vec<f64> = Vec::with_capacity(reps);
    for _ in 0..reps {
        let t = Instant::now();
        f();
        samples.push(t.elapsed().as_nanos() as f64);
    }
    samples.sort_by(|a, b| a.partial_cmp(b).unwrap());
    samples[samples.len() / 2]
}

fn round(f: f64, places: i32) -> f64 {
    let m = 10f64.powi(places);
    (f * m).round() / m
}

// ── 1. oid lookup: stree vs binary search vs fst ──────────────────────────────

/// What one `oid_lookup` run measured.
#[derive(Debug, Clone)]
pub struct OidLookupReport {
    pub objects: usize,
    pub queries: usize,
    pub oid_len: usize,
    /// `__gunnar_oid__` section size.
    pub section_mb: f64,
    pub build_ms: f64,
    /// `std::binary_search` over the same keys + the same full-oid verify.
    pub binsearch_ns: f64,
    /// `stree::find_exact` + verify, one query at a time.
    pub serial_ns: f64,
    /// `stree::lookup_batch_pipeline::<8>` + verify — the pack-serving path.
    pub batch_ns: f64,
    /// An `fst::Map` over the same oids, built and queried the same way.
    pub fst_build_ms: f64,
    pub fst_mb: f64,
    pub fst_ns: f64,
    pub batch_vs_binsearch_x: f64,
    pub serial_vs_binsearch_x: f64,
    pub batch_vs_fst_x: f64,
}

/// Build a `__gunnar_oid__` index over `objects` random oids and time four ways
/// of resolving `queries` of them.
///
/// All four are held to the same contract — derive the key, find the run, verify
/// the full oid — and all four are checked against each other on every query
/// before any timing is reported.
pub fn oid_lookup(
    objects: usize,
    queries: usize,
    hash: GitHashKind,
    reps: usize,
    seed: u64,
) -> Result<OidLookupReport> {
    use std::hint::black_box;

    ensure!(objects > 0 && queries > 0, "oid_lookup needs a non-empty workload");
    let oid_len = hash.oid_len();
    let oids = random_oids(objects, oid_len, seed);

    let entries: Vec<OidEntry> = oids
        .iter()
        .enumerate()
        .map(|(i, o)| OidEntry { oid: o.clone(), lookup_row: i as u64, ordinal: i as u32 })
        .collect();

    let t0 = Instant::now();
    let section = build_section(&entries, hash)?;
    let index = GitOidIndex::parse(section.clone())?;
    let build_ms = t0.elapsed().as_secs_f64() * 1000.0;
    ensure!(index.len() == objects, "index lost objects while building");

    // Queries: random *present* oids, so every lookup is a full-depth traversal
    // rather than an early miss.
    let mut rng = Rng(seed ^ 0xC0FF_EE00);
    let qs: Vec<Vec<u8>> = (0..queries)
        .map(|_| oids[(rng.next() as usize) % oids.len()].clone())
        .collect();
    let qrefs: Vec<&[u8]> = qs.iter().map(|v| v.as_slice()).collect();

    // ── the fst rival, built over the same keys ──────────────────────────────
    // fst requires keys in lexicographic order; raw oids sorted lexicographically
    // is exactly the archive's own object ordering, so this is the index znippy
    // would have got for free had the stock trie been kept.
    let mut sorted: Vec<(&[u8], u64)> = oids
        .iter()
        .enumerate()
        .map(|(i, o)| (o.as_slice(), i as u64))
        .collect();
    sorted.sort_unstable_by(|a, b| a.0.cmp(b.0));
    let t1 = Instant::now();
    let mut fb = fst::MapBuilder::memory();
    for (k, v) in &sorted {
        fb.insert(k, *v)?;
    }
    let fst_map = fst::Map::new(fb.into_inner()?)?;
    let fst_build_ms = t1.elapsed().as_secs_f64() * 1000.0;
    let fst_mb = fst_map.as_fst().as_bytes().len() as f64 / 1_048_576.0;

    // ── LAW 2: the four paths must agree before any of them is timed ─────────
    let batched = index.lookup_batch(&qrefs);
    for (i, q) in qs.iter().enumerate() {
        let serial = index.lookup(q);
        ensure!(serial.is_some(), "a present oid missed the stree index");
        ensure!(batched[i] == serial, "batch and serial disagree at query {i}");
        ensure!(
            index.lookup_binary_search(q) == serial,
            "binary-search baseline disagrees with stree at query {i} — the \
             comparison below would be between two different answers"
        );
        let want_row = serial.unwrap().lookup_row;
        ensure!(
            fst_map.get(q) == Some(want_row),
            "fst rival disagrees with stree at query {i}"
        );
    }

    // Warm up every path once so the first timed rep is not paying page faults
    // that the others already paid.
    black_box(index.lookup_batch(&qrefs));

    let per_q = |ns: f64| ns / queries as f64;

    let binsearch_ns = per_q(time_ns(reps, || {
        let mut acc = 0u64;
        for q in &qs {
            acc ^= black_box(index.lookup_binary_search(q)).map_or(0, |h| h.lookup_row);
        }
        black_box(acc);
    }));
    let serial_ns = per_q(time_ns(reps, || {
        let mut acc = 0u64;
        for q in &qs {
            acc ^= black_box(index.lookup(q)).map_or(0, |h| h.lookup_row);
        }
        black_box(acc);
    }));
    let batch_ns = per_q(time_ns(reps, || {
        black_box(index.lookup_batch(&qrefs));
    }));
    let fst_ns = per_q(time_ns(reps, || {
        let mut acc = 0u64;
        for q in &qs {
            acc ^= black_box(fst_map.get(q)).unwrap_or(0);
        }
        black_box(acc);
    }));

    let ratio = |base: f64, new: f64| if new > 0.0 { round(base / new, 2) } else { 0.0 };
    Ok(OidLookupReport {
        objects,
        queries,
        oid_len,
        section_mb: round(section.len() as f64 / 1_048_576.0, 2),
        build_ms: round(build_ms, 1),
        binsearch_ns: round(binsearch_ns, 1),
        serial_ns: round(serial_ns, 1),
        batch_ns: round(batch_ns, 1),
        fst_build_ms: round(fst_build_ms, 1),
        fst_mb: round(fst_mb, 2),
        fst_ns: round(fst_ns, 1),
        batch_vs_binsearch_x: ratio(binsearch_ns, batch_ns),
        serial_vs_binsearch_x: ratio(binsearch_ns, serial_ns),
        batch_vs_fst_x: ratio(fst_ns, batch_ns),
    })
}

// ── 2. a synthetic repository ─────────────────────────────────────────────────

/// A generated object set with the shape of a git history: one blob and one tree
/// per commit, the tree also carrying a few carried-over blobs so tree closures
/// overlap the way they do in a real repository.
pub struct SynthRepo {
    pub hash: GitHashKind,
    /// Canonical bytes of every object, in creation order.
    pub objects: Vec<(String, Vec<u8>)>,
    pub commits: Vec<CommitNode>,
    pub tree_payloads: HashMap<String, Vec<u8>>,
    pub tip: String,
}

/// Generate `commits` commits. `width` is how many blobs each tree names, and
/// `merge_every` inserts a second parent that many commits apart (0 = linear).
///
/// Object count is roughly `commits * (2 + 1)` — one new blob, one tree and one
/// commit each — so `commits` is the knob and the object total is reported.
pub fn synth_repo(commits: usize, width: usize, merge_every: usize, hash: GitHashKind) -> SynthRepo {
    let oid_len = hash.oid_len();
    let mut objects: Vec<(String, Vec<u8>)> = Vec::with_capacity(commits * 3);
    let mut tree_payloads: HashMap<String, Vec<u8>> = HashMap::new();
    let mut nodes: Vec<CommitNode> = Vec::with_capacity(commits);
    let mut blob_pool: Vec<String> = Vec::new();
    let mut prev: Option<String> = None;
    let mut tip = String::new();

    for i in 0..commits {
        let blob = canonical(GitObjectKind::Blob, format!("content of object {i}\n").as_bytes());
        let blob_oid = hash.oid_hex_of(&blob);
        objects.push((blob_oid.clone(), blob));
        blob_pool.push(blob_oid);

        // The tree names the newest `width` blobs, so consecutive trees share
        // most of their entries — which is what makes the memoized tree closure
        // worth having, and what a synthetic repo without carry-over would hide.
        let start = blob_pool.len().saturating_sub(width.max(1));
        let mut payload = Vec::new();
        for (k, b) in blob_pool[start..].iter().enumerate() {
            payload.extend_from_slice(b"100644 ");
            payload.extend_from_slice(format!("f{k}.txt").as_bytes());
            payload.push(0);
            payload.extend_from_slice(&hex::decode(b).unwrap()[..oid_len]);
        }
        let tree = canonical(GitObjectKind::Tree, &payload);
        let tree_oid = hash.oid_hex_of(&tree);
        tree_payloads.insert(tree_oid.clone(), payload);
        objects.push((tree_oid.clone(), tree));

        let mut parents: Vec<String> = Vec::new();
        if let Some(p) = &prev {
            parents.push(p.clone());
        }
        if merge_every > 0 && i > merge_every && i % merge_every == 0 {
            // A second parent reaching back `merge_every` commits: a real merge
            // edge, which is what makes generation numbers non-trivial.
            parents.push(nodes[i - merge_every].oid.clone());
        }

        let mut body = format!("tree {tree_oid}\n");
        for p in &parents {
            body.push_str(&format!("parent {p}\n"));
        }
        let t = 1_700_000_000i64 + i as i64;
        body.push_str(&format!(
            "author A <a@x> {t} +0000\ncommitter A <a@x> {t} +0000\n\ncommit {i}\n"
        ));
        let commit = canonical(GitObjectKind::Commit, body.as_bytes());
        let commit_oid = hash.oid_hex_of(&commit);
        objects.push((commit_oid.clone(), commit));

        nodes.push(CommitNode {
            oid: commit_oid.clone(),
            parents,
            tree: Some(tree_oid),
            committer_time: Some(t),
            generation: 0,
        });
        prev = Some(commit_oid.clone());
        tip = commit_oid;
    }

    SynthRepo { hash, objects, commits: nodes, tree_payloads, tip }
}

impl SynthRepo {
    /// oid hex → ordinal, in the archive's oid-lexicographic order.
    pub fn ordinals(&self) -> HashMap<String, u32> {
        let mut sorted: Vec<&str> = self.objects.iter().map(|(o, _)| o.as_str()).collect();
        sorted.sort_unstable();
        sorted.dedup();
        sorted.iter().enumerate().map(|(i, o)| ((*o).to_string(), i as u32)).collect()
    }
}

// ── 3. commit-graph build ─────────────────────────────────────────────────────

#[derive(Debug, Clone)]
pub struct GraphReport {
    pub commits: usize,
    pub merge_every: usize,
    pub max_generation: u32,
    pub gen_assign_ms: f64,
    pub ns_per_commit: f64,
}

/// Time `assign_generations` over a history `commits` deep.
///
/// The generation number is the whole point of `__gunnar_graph__` — it is what
/// turns an ancestry test into an integer compare — so the cost of producing it
/// at repack time is the cost that has to be justified.
pub fn graph_build(commits: usize, merge_every: usize, reps: usize) -> Result<GraphReport> {
    use std::hint::black_box;
    let repo = synth_repo(commits, 4, merge_every, GitHashKind::Sha256);
    let nodes = repo.commits.clone();

    let assigned = assign_generations(nodes.clone());
    let max_generation = assigned.iter().map(|c| c.generation).max().unwrap_or(0);
    ensure!(
        max_generation > 0 && assigned.len() == commits,
        "generation assignment lost commits or produced no generations"
    );
    // A linear history's deepest commit must sit at generation == depth; if it
    // does not, the walk is not doing what the number claims.
    if merge_every == 0 {
        ensure!(
            max_generation as usize == commits,
            "a linear history {commits} deep should reach generation {commits}, got {max_generation}"
        );
    }

    let ns = time_ns(reps, || {
        black_box(assign_generations(black_box(nodes.clone())));
    });
    // The clone is inside the timed closure because `assign_generations` consumes
    // its input; subtract it so the number is the walk, not the memcpy.
    let clone_ns = time_ns(reps, || {
        black_box(nodes.clone());
    });
    let walk_ns = (ns - clone_ns).max(0.0);

    Ok(GraphReport {
        commits,
        merge_every,
        max_generation,
        gen_assign_ms: round(walk_ns / 1e6, 2),
        ns_per_commit: round(walk_ns / commits as f64, 1),
    })
}

// ── 4. reachability bitmaps: build cost, and whether the query pays ───────────

#[derive(Debug, Clone)]
pub struct ReachReport {
    pub commits: usize,
    pub objects: usize,
    pub bitmaps: usize,
    pub build_ms: f64,
    pub build_ns_per_commit: f64,
    pub bitmap_bytes: usize,
    /// `want − have` as a roaring ANDNOT over two sealed bitmaps.
    pub andnot_ns: f64,
    /// The same answer computed the way a server without bitmaps must: walk the
    /// commit ancestry and take every tree closure.
    pub traverse_ns: f64,
    pub query_speedup_x: f64,
    /// How many objects the answer actually contained — a speedup on an empty
    /// set would mean nothing.
    pub delta_objects: u64,
}

/// Walk `want`'s ancestry and subtract `have`'s, the way a server with no
/// bitmaps must. This is the baseline the bitmaps have to beat.
///
/// Deliberately unmemoized across calls: a fresh `tree_closure` memo per query
/// is exactly what "no precomputed reachability" means.
fn traverse_want_minus_have(
    commits: &[CommitNode],
    facts: &ObjectFacts<'_>,
    want: &str,
    have: &str,
) -> RoaringBitmap {
    let by: HashMap<&str, &CommitNode> = commits.iter().map(|c| (c.oid.as_str(), c)).collect();

    let mut closure = |root: &str| -> RoaringBitmap {
        let mut seen: HashSet<&str> = HashSet::new();
        let mut stack = vec![root];
        let mut memo: HashMap<String, RoaringBitmap> = HashMap::new();
        let mut bm = RoaringBitmap::new();
        while let Some(c) = stack.pop() {
            if !seen.insert(c) {
                continue;
            }
            let Some(node) = by.get(c) else { continue };
            if let Some(&o) = facts.ordinal.get(c) {
                bm.insert(o);
            }
            if let Some(t) = &node.tree {
                bm |= tree_closure(t, facts, &mut memo);
            }
            for p in &node.parents {
                stack.push(p.as_str());
            }
        }
        bm
    };
    let w = closure(want);
    let h = closure(have);
    w - h
}

/// Build the bitmaps over a synthetic repository, then answer one `want − have`
/// both ways and compare.
///
/// The two answers are asserted equal before either timing is reported — a
/// bitmap ANDNOT that returns a different object set is not a faster answer to
/// the same question.
pub fn reach_build_and_query(
    commits: usize,
    width: usize,
    merge_every: usize,
    policy: ReachPolicy,
    reps: usize,
) -> Result<ReachReport> {
    use std::hint::black_box;

    let repo = synth_repo(commits, width, merge_every, GitHashKind::Sha256);
    let ordinal = repo.ordinals();
    let nodes = assign_generations(repo.commits.clone());
    let facts = ObjectFacts {
        ordinal: &ordinal,
        trees: &repo.tree_payloads,
        oid_len: repo.hash.oid_len(),
    };

    let t0 = Instant::now();
    let entries: Vec<ReachEntry> = build_reach(&nodes, &facts, policy);
    let build_ms = t0.elapsed().as_secs_f64() * 1000.0;
    ensure!(!entries.is_empty(), "no bitmaps were built");

    let bitmap_bytes: usize = entries.iter().map(|e| e.bitmap.serialized_size()).sum();

    // want = the tip's bitmap; have = the earliest bitmapped commit, so the
    // delta is large enough that the comparison is about the algorithms and not
    // about two nearly identical sets.
    let by: HashMap<&str, &RoaringBitmap> =
        entries.iter().map(|e| (e.commit.as_str(), &e.bitmap)).collect();
    let tip_bm = by
        .get(repo.tip.as_str())
        .copied()
        .ok_or_else(|| anyhow::anyhow!("the tip commit did not get a bitmap"))?;
    let (have_oid, have_bm) = entries
        .iter()
        .filter(|e| e.commit != repo.tip)
        .min_by_key(|e| e.bitmap.len())
        .map(|e| (e.commit.clone(), &e.bitmap))
        .ok_or_else(|| anyhow::anyhow!("need at least two bitmaps to answer want-minus-have"))?;

    let via_bitmap = tip_bm.clone() - have_bm.clone();
    let via_walk = traverse_want_minus_have(&nodes, &facts, &repo.tip, &have_oid);
    ensure!(
        via_bitmap == via_walk,
        "the bitmap ANDNOT and the ancestry walk disagree ({} vs {} objects) — \
         timing them against each other would be meaningless",
        via_bitmap.len(),
        via_walk.len()
    );
    ensure!(via_bitmap.len() > 0, "the delta is empty; there is nothing to measure");

    let andnot_ns = time_ns(reps, || {
        black_box(tip_bm.clone() - have_bm.clone());
    });
    let traverse_ns = time_ns(reps.min(3).max(1), || {
        black_box(traverse_want_minus_have(&nodes, &facts, &repo.tip, &have_oid));
    });

    Ok(ReachReport {
        commits,
        objects: ordinal.len(),
        bitmaps: entries.len(),
        build_ms: round(build_ms, 1),
        build_ns_per_commit: round(build_ms * 1e6 / commits as f64, 1),
        bitmap_bytes,
        andnot_ns: round(andnot_ns, 1),
        traverse_ns: round(traverse_ns, 1),
        query_speedup_x: if andnot_ns > 0.0 { round(traverse_ns / andnot_ns, 2) } else { 0.0 },
        delta_objects: via_bitmap.len(),
    })
}

// ── 5. building the three sections at repack scale ────────────────────────────

#[derive(Debug, Clone)]
pub struct SectionsReport {
    pub objects: usize,
    pub commits: usize,
    pub push_ms: f64,
    pub build_sections_ms: f64,
    pub oid_section_mb: f64,
    pub objects_per_s: f64,
}

/// Feed `commits * 3` objects through `GitIndexBuilder` and build all three
/// reserved sections — the whole per-repack cost of the `git` format, minus the
/// archive write itself.
///
/// This is the arm that can be pushed to a million objects, because it needs no
/// filesystem: `archive_build` below does the same thing with a real
/// `compress_dir` at a size a filesystem can carry.
pub fn sections_build(commits: usize, width: usize, policy: ReachPolicy) -> Result<SectionsReport> {
    let repo = synth_repo(commits, width, 0, GitHashKind::Sha256);

    let t0 = Instant::now();
    let mut builder = GitIndexBuilder::new(GitHashKind::Sha256).with_reach_policy(policy);
    for (_, bytes) in &repo.objects {
        builder.push_canonical(bytes)?;
    }
    let push_ms = t0.elapsed().as_secs_f64() * 1000.0;

    let first_row: HashMap<&str, u64> = {
        let mut sorted: Vec<&str> = repo.objects.iter().map(|(o, _)| o.as_str()).collect();
        sorted.sort_unstable();
        sorted.dedup();
        sorted.iter().enumerate().map(|(i, o)| (*o, i as u64)).collect()
    };
    let n_objects = first_row.len();

    let t1 = Instant::now();
    let sections = builder.build_sections(&first_row)?;
    let build_sections_ms = t1.elapsed().as_secs_f64() * 1000.0;
    ensure!(sections.len() == 3, "expected all three reserved sections, got {}", sections.len());

    let oid_bytes = match &sections[0].payload {
        znippy_common::ReservedPayload::Raw(b) => b.len(),
        _ => anyhow::bail!("the oid section must be raw bytes"),
    };
    // Applied output: the section really indexes every object.
    let index = GitOidIndex::parse(match &sections[0].payload {
        znippy_common::ReservedPayload::Raw(b) => b.clone(),
        _ => unreachable!(),
    })?;
    ensure!(index.len() == n_objects, "oid section indexes {} of {n_objects}", index.len());

    let total_ms = push_ms + build_sections_ms;
    Ok(SectionsReport {
        objects: n_objects,
        commits,
        push_ms: round(push_ms, 1),
        build_sections_ms: round(build_sections_ms, 1),
        oid_section_mb: round(oid_bytes as f64 / 1_048_576.0, 2),
        objects_per_s: round(n_objects as f64 / (total_ms / 1000.0), 0),
    })
}

// ── 5b. the ref push log: how fast can a push land, and how fast is recovery ──

#[derive(Debug, Clone)]
pub struct PushLogReport {
    pub pushes: usize,
    pub refs_per_push: usize,
    pub log_mb: f64,
    /// Reciprocal throughput of one durable push: build the batch, append it,
    /// **fsync**. Lower is better.
    pub push_ns: f64,
    pub pushes_per_s: f64,
    /// Reciprocal throughput of recovering one push from a cold log.
    pub scan_ns: f64,
    /// Wall time to fold the whole log into the current ref namespace.
    pub fold_ms: f64,
    pub refs_live: usize,
    /// One compaction over the whole log, and what the log costs afterwards.
    pub compact_ms: f64,
    pub compact_mb: f64,
    pub compact_ratio: f64,
    /// Recovery of the COMPACTED log, per push. Directly comparable to
    /// `scan_ns` above, which is the same work on the uncompacted one.
    pub scan_compact_ns: f64,
}

/// Measure the two operations the ref format actually has to be fast at:
/// **landing a push** (which is fsync-bound, and is what a client waits on) and
/// **recovering the log** (which is what a cold start pays).
///
/// The two are reported separately and never summed, because they are paid by
/// different people at different times: a push cost is per-client latency, a
/// scan cost is once per process start.
///
/// RED for this bench: `push_ns` collapsing toward zero would mean the fsync
/// stopped happening, i.e. the durability claim the whole format rests on had
/// quietly been traded for the throughput number — so the kernel asserts the
/// bytes really reached the file and that recovery returns every push.
pub fn pushlog_throughput(
    pushes: usize,
    refs_per_push: usize,
    dir: &std::path::Path,
) -> Result<PushLogReport> {
    use crate::refs::{RefLog, RefUpdate};

    std::fs::create_dir_all(dir)?;
    let path = dir.join(format!("bench-refs-{pushes}-{refs_per_push}.log"));
    let _ = std::fs::remove_file(&path);
    let log = RefLog::new(&path);

    // Pre-build the update sets so the timed region is append+fsync, not
    // string formatting.
    let mut sets: Vec<Vec<RefUpdate>> = Vec::with_capacity(pushes);
    for p in 0..pushes {
        sets.push(
            (0..refs_per_push)
                .map(|r| {
                    RefUpdate::set(
                        format!("refs/heads/b{:06}", (p * refs_per_push + r) % 4096),
                        format!("{:064x}", p * refs_per_push + r),
                    )
                })
                .collect(),
        );
    }

    // `RefLog::push` re-derives push_seq by rescanning, which is O(log) per
    // push and would make this measure the scan, not the append. Drive the
    // batch + append directly with an explicit sequence — the same code path
    // the server takes when it holds its own counter.
    let t0 = Instant::now();
    for (seq, updates) in sets.iter().enumerate() {
        let batch = crate::refs::build_push_batch(updates, seq as u64, 1_700_000_000_000)?;
        log.append_batch(&batch)?;
    }
    let push_s = t0.elapsed().as_secs_f64();

    let log_bytes = std::fs::metadata(&path)?.len();
    ensure!(log_bytes > 0, "the log is empty — no push reached the filesystem");

    // Cold recovery.
    let t1 = Instant::now();
    let scan = log.scan()?;
    let scan_s = t1.elapsed().as_secs_f64();
    ensure!(
        scan.pushes.len() == pushes,
        "recovery returned {} of {pushes} pushes",
        scan.pushes.len()
    );
    ensure!(scan.is_clean(), "a log written without a crash must recover clean");

    let t2 = Instant::now();
    let live = crate::refs::fold(&scan.pushes)?;
    let fold_ms = t2.elapsed().as_secs_f64() * 1000.0;
    ensure!(!live.is_empty(), "the fold produced no refs");

    // Compaction: fold every frame into one, then pay recovery again on the
    // compacted log so the two scan numbers are the same work on the two
    // shapes.
    let t3 = Instant::now();
    let report = log.compact()?;
    let compact_ms = t3.elapsed().as_secs_f64() * 1000.0;
    ensure!(
        report.rows == pushes * refs_per_push,
        "compaction changed the row count: {} rows for {pushes}x{refs_per_push}",
        report.rows
    );

    let t4 = Instant::now();
    let scan2 = log.scan()?;
    let scan_compact_s = t4.elapsed().as_secs_f64();
    ensure!(scan2.is_clean(), "the compacted log did not scan clean");
    // Applied output: the fold is identical after compaction. A smaller log
    // that answers a different question is not a smaller log.
    let live2 = crate::refs::fold(&scan2.pushes)?;
    ensure!(
        live2 == live,
        "compaction changed the ref namespace: {} refs before, {} after",
        live.len(),
        live2.len()
    );

    let _ = std::fs::remove_file(&path);

    Ok(PushLogReport {
        pushes,
        refs_per_push,
        log_mb: round(log_bytes as f64 / 1_048_576.0, 2),
        push_ns: round(push_s * 1e9 / pushes as f64, 0),
        pushes_per_s: round(pushes as f64 / push_s, 0),
        scan_ns: round(scan_s * 1e9 / pushes as f64, 0),
        fold_ms: round(fold_ms, 2),
        refs_live: live.len(),
        compact_ms: round(compact_ms, 1),
        compact_mb: round(report.bytes_after as f64 / 1_048_576.0, 2),
        compact_ratio: round(report.bytes_before as f64 / report.bytes_after.max(1) as f64, 2),
        scan_compact_ns: round(scan_compact_s * 1e9 / pushes as f64, 0),
    })
}

// ── 6. a real archive, sealed through compress_dir ────────────────────────────

#[derive(Debug, Clone)]
pub struct ArchiveReport {
    pub objects: usize,
    pub source_mb: f64,
    pub archive_mb: f64,
    pub write_fixture_s: f64,
    pub seal_s: f64,
    pub objects_per_s: f64,
    pub source_mb_per_s: f64,
    /// Resolving every object through the sealed index, one at a time.
    pub reopen_lookup_ns: f64,
}

/// Seal `commits * 3` git objects into a real `.znippy` through the ordinary
/// `compress_dir` path with all three reserved sections attached, then reopen it
/// and resolve every object through the sealed oid index.
///
/// `dir` is where the object files are laid out; on this machine that must be a
/// path on md1, never `/tmp` (which is RAM).
pub fn archive_build(commits: usize, width: usize, dir: &std::path::Path) -> Result<ArchiveReport> {
    use std::fs;
    use std::hint::black_box;
    use znippy_common::plugin::PluginRegistry;
    use znippy_common::{ArchiveMetaSink, ArrowIpcSink};
    use znippy_compress::compress_dir;

    let repo = synth_repo(commits, width, 0, GitHashKind::Sha256);
    let src = dir.join("objects");
    let archive = dir.join("repo.znippy");
    if src.exists() {
        fs::remove_dir_all(&src)?;
    }
    fs::create_dir_all(&src)?;
    let _ = fs::remove_file(&archive);

    let t0 = Instant::now();
    let mut builder = GitIndexBuilder::new(GitHashKind::Sha256);
    let mut source_bytes = 0usize;
    let mut written: HashSet<&str> = HashSet::with_capacity(repo.objects.len());
    for (oid, bytes) in &repo.objects {
        if !written.insert(oid.as_str()) {
            continue;
        }
        fs::write(src.join(oid), bytes)?;
        source_bytes += bytes.len();
        builder.push_canonical(bytes)?;
    }
    let n_objects = written.len();
    let write_fixture_s = t0.elapsed().as_secs_f64();

    let reserved = builder.into_reserved_builder();
    let registry = PluginRegistry::with_plugin(Box::new(crate::NativeGitPlugin::new()));
    let t1 = Instant::now();
    let report = compress_dir(
        &src,
        &archive,
        false,
        Some(&registry),
        None,
        Some(Box::new(move |f, b| {
            Box::new(ArrowIpcSink::new(f, b).with_reserved_builder(reserved))
                as Box<dyn ArchiveMetaSink>
        })),
    )?;
    let seal_s = t1.elapsed().as_secs_f64();
    ensure!(
        report.total_files == n_objects as u64,
        "sealed {} of {n_objects} objects",
        report.total_files
    );

    let index = GitOidIndex::open(&archive)?
        .ok_or_else(|| anyhow::anyhow!("the sealed archive carries no __gunnar_oid__"))?;
    ensure!(index.len() == n_objects, "sealed index holds {} of {n_objects}", index.len());

    let oids: Vec<&str> = written.iter().copied().collect();
    let lookup_ns = time_ns(3, || {
        let mut acc = 0u64;
        for o in &oids {
            acc ^= black_box(index.lookup_hex(o)).map_or(0, |h| h.lookup_row);
        }
        black_box(acc);
    }) / oids.len() as f64;

    let archive_mb = fs::metadata(&archive)?.len() as f64 / 1_048_576.0;
    let source_mb = source_bytes as f64 / 1_048_576.0;
    let _ = fs::remove_dir_all(&src);
    let _ = fs::remove_file(&archive);

    Ok(ArchiveReport {
        objects: n_objects,
        source_mb: round(source_mb, 2),
        archive_mb: round(archive_mb, 2),
        write_fixture_s: round(write_fixture_s, 2),
        seal_s: round(seal_s, 2),
        objects_per_s: round(n_objects as f64 / seal_s, 0),
        source_mb_per_s: round(source_mb / seal_s, 1),
        reopen_lookup_ns: round(lookup_ns, 1),
    })
}

// ── the kernels' own guards ───────────────────────────────────────────────────
//
// Tiny sizes, so `cargo test` proves the workloads are wired correctly without
// paying for a benchmark. What is asserted is the thing that makes a bench
// trustworthy: that the paths being compared return the SAME answer.

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

    #[test]
    fn every_lookup_strategy_returns_the_same_row() {
        let r = oid_lookup(4_096, 512, GitHashKind::Sha256, 1, 42).unwrap();
        assert_eq!(r.objects, 4_096);
        // The cross-checks inside `oid_lookup` are the real assertion; here we
        // only prove the timings are populated and not identically zero, which
        // is how a silently skipped workload would look.
        assert!(r.binsearch_ns > 0.0 && r.serial_ns > 0.0 && r.batch_ns > 0.0 && r.fst_ns > 0.0);
    }

    #[test]
    fn a_linear_history_reaches_generation_equal_to_its_depth() {
        let r = graph_build(2_000, 0, 1).unwrap();
        assert_eq!(r.max_generation, 2_000);
        assert!(r.ns_per_commit > 0.0);
    }

    #[test]
    fn the_bitmap_andnot_and_the_ancestry_walk_agree() {
        let r = reach_build_and_query(200, 4, 0, ReachPolicy { max_commits: 64 }, 1).unwrap();
        assert!(r.delta_objects > 0, "the measured delta must not be empty");
        assert!(r.bitmaps > 1);
        // Equality of the two answers is asserted inside the kernel; if it were
        // ever relaxed to a subset test, this size check would still be blind —
        // so the kernel keeps the equality and this test keeps the shape.
        assert!(r.andnot_ns > 0.0 && r.traverse_ns > 0.0);
    }

    #[test]
    fn all_three_sections_are_built_and_the_oid_section_indexes_everything() {
        let r = sections_build(100, 4, ReachPolicy { max_commits: 32 }).unwrap();
        assert_eq!(r.objects, 300);
        assert!(r.oid_section_mb >= 0.0);
    }

    /// The push-log kernel must report a real fsync cost. A push that costs
    /// essentially nothing is a push that did not reach the disk, and the
    /// durability claim would be the thing that got optimised away.
    #[test]
    fn the_pushlog_kernel_measures_a_real_durable_push() {
        let dir = std::env::temp_dir().join(format!(
            "znippy_pushlog_bench_{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let r = pushlog_throughput(64, 4, &dir).unwrap();
        assert_eq!(r.pushes, 64);
        assert!(r.push_ns > 0.0, "a durable push cannot cost zero");
        assert!(r.scan_ns > 0.0);
        assert!(r.refs_live > 0, "the fold must produce refs");
        assert!(r.log_mb >= 0.0);
        std::fs::remove_dir_all(&dir).ok();
    }
}