znippy-plugin-git 0.1.1

//! **What the redb tail buys on the WRITE side** — the one claim the read stack
//! is justified by and the one that had never been measured.
//!
//! ```text
//! CARGO_TARGET_DIR=/home/rickard/scratch/cargo/znippy-plugin-git-tailwrite \
//!   cargo run --release --no-default-features --example tail_write_bench -- \
//!   bases=10000,100000,1000000 pushes=50 push=300 runs=3
//! ```
//!
//! Arguments (`key=value`): `bases=`, `pushes=`, `push=`, `runs=`, `oid=20`,
//! `real=<path>` (a `git cat-file --batch-all-objects --batch-check` dump — see
//! [`load_real`]), `shuffle=0|1`, `dir=<scratch>`, `seed=`.
//!
//! ## The claim
//!
//! [`crate::read_stack`]'s own module doc says reads are a wash — 108 ns/oid with
//! the stack against 109 ns for the Arrow tables alone — and rests the whole
//! design on the write side instead:
//!
//! > Without the tail, every push must rebuild the Arrow index, so push cost
//! > grows with **repository size**. With it, a push appends only what changed to
//! > redb and the projection catches up on a ratio trigger.
//!
//! The only figure ever offered for that was arithmetic: ~45 bytes/row × 1M rows
//! = a ~45 MB table rewritten per push against ~13.5 KB appended. Arithmetic is
//! not a measurement. This is the measurement.
//!
//! ## The two arms
//!
//! Same entries, same order, same process, same run — the arms differ in one
//! thing and it is the tail.
//!
//! * **A, with the tail.** [`ObjectReadStack::append`] into a redb file on disk,
//!   with the shipped [`RebuildTriggers::default()`] — `tail_hits_per_row = 1.0`,
//!   floor 4096, volume 64 MiB. A rebuild that the volume trigger fires runs
//!   **inline inside the timed push**, which is where `read_stack` actually puts
//!   it (its own doc: "There are none here", of threads). So arm A is charged
//!   for its own catch-up, in the push that pays for it. The task that
//!   commissioned this called the rebuild "background"; the code is inline, and
//!   inline is what is measured — see the report's note.
//! * **B, without the tail.** The rows live in a `Vec<IndexEntry>` in memory —
//!   B is *not* made to re-read them from disk, because a real no-tail
//!   implementation would not — and every push extends that vector, rebuilds
//!   [`OneTableFourColumns`] over **all** of it, and writes the resulting Arrow
//!   IPC stream to disk atomically (`tmp` → `write_all` → `sync_all` → `rename`
//!   → directory fsync). That last part is not optional: arm A's redb commit is
//!   fully durable, so an arm B that left its index in the page cache would be
//!   winning on a durability it never paid for.
//!
//! ### Arm A is not allowed to win by deferring
//!
//! A 50-push window of ordinary objects never reaches the 64 MiB volume trigger,
//! so arm A would otherwise be timed over exactly the window in which it owes a
//! rebuild and has not paid for it. After the timed loop the harness therefore
//! runs one explicit [`ObjectReadStack::rebuild`] over the repository as it then
//! stands, times it, and prints it **amortised over the trigger period** —
//! `64 MiB / bytes per push` — beside the raw per-push figure. Both numbers are
//! shown; neither is folded into the other.
//!
//! B rebuilds the `stree` too, because [`ObjectIndex::build`] does, and that is
//! the honest cost of "rebuild the Arrow index". B does **not** persist the
//! stree section — it can be re-derived from the oid column at open — so the
//! bytes reported for B are the smallest a real no-tail implementation could
//! write, not the largest. Every asymmetry left in this harness favours B.
//!
//! ## What is counted
//!
//! * **wall** — `Instant`, per push.
//! * **cpu** — `CLOCK_PROCESS_CPUTIME_ID`, per push. Wall minus cpu is fsync.
//! * **bytes written** — `/proc/self/io`, per push, both fields:
//!   * `wchar`, bytes this process handed to `write(2)`. Deterministic and
//!     wholly attributable; this is the number quoted. redb 2.6.3 writes through
//!     `File`/`pwrite` in `page_store::file_backend::unix` and not through a
//!     mapping, so its writes land in `wchar` exactly as arm B's do.
//!   * `write_bytes`, the storage-layer estimate, reported beside it as a
//!     cross-check. Buffered writeback can be charged to a flusher thread, so
//!     this one is a floor, not a total.
//!   * plus the on-disk size of each arm's artefact at the end of the run.
//!   Nothing here is inferred from a row width.
//!
//! ## Method
//!
//! * **Arm order rotates by run** (A,B / B,A / A,B …). A previous sweep in this
//!   crate measured a real ~1.1% penalty for whichever arm ran second; without
//!   rotation that bias lands entirely on one arm.
//! * **`/proc/loadavg` is read before every arm** and printed with its figures.
//!   oden is shared. A cell taken above `LOAD_QUIET` is marked `[LOADED]` in the
//!   output rather than silently averaged in.
//! * **3 runs per cell, median quoted, min/max printed.** The spread is the
//!   noise band and nothing smaller than it is claimed.
//! * **Per-push, not just total** — the claim is about the *shape* of the curve,
//!   so every push's cost is kept and the first/median/last thirds are printed
//!   separately.
//!
//! ## MEASURED — oden, 2026-08-08, 1-min loadavg 1.51–1.58, 3 runs/cell
//!
//! 50 pushes × 300 objects, sha1, `--no-default-features`, built from inside
//! `/home/rickard/git/znippy` so `.cargo/config.toml`'s `target-cpu=native` and
//! the root `lto="thin", codegen-units=1` both applied.
//!
//! **Per-push wall, median of 3** (worst run-to-run band 21.6%, so nothing under
//! ~22% is claimed):
//!
//! | repository | A with tail | B no tail | B/A |
//! |---|---:|---:|---:|
//! | 10 000 | 2.115 ms | 4.176 ms | **2.0×** |
//! | 100 000 | 2.917 ms | 33.37 ms | **11.4×** |
//! | 1 000 000 | 5.731 ms | 726.5 ms | **126.8×** |
//!
//! **Per-push bytes written (`wchar`), median of 3, run-to-run band 0.0% at
//! every cell** — this counter is deterministic and load-independent:
//!
//! | repository | A with tail | B no tail | B/A |
//! |---|---:|---:|---:|
//! | 10 000 | 930 KB | 950 KB | 1.0× |
//! | 100 000 | 1.375 MB | 5.787 MB | 4.2× |
//! | 1 000 000 | 2.215 MB | 54.16 MB | **24.5×** |
//!
//! **The claim holds, and the shape is the reason.** B's per-push cost is the
//! whole repository every time: 4.2 → 33.4 → 726 ms across two decades of
//! repository size, and its written bytes are exactly the Arrow table (54.16 MB
//! at 1M rows, against the ~45 MB the arithmetic predicted — the arithmetic was
//! 20% low because it forgot `delta_base`). A's cost rises too, but only as the
//! tail B-tree deepens: 2.1 → 2.9 → 5.7 ms, a **2.7× rise for a 100× repository**
//! against B's 174×. Within a single cell A is flat (last-third / first-third
//! 1.03× at 1M) where B is not.
//!
//! **Two things the arithmetic got wrong, both against the tail.**
//!
//! 1. **A does not write 13.5 KB per push, it writes 0.9–2.2 MB.** 300 rows ×
//!    41 B is 12.3 KB of payload; redb is a copy-on-write B-tree keyed by oid,
//!    oids are random, so each of the 300 inserts dirties its own root-to-leaf
//!    path and the commit fsyncs all of them. That is **75× write amplification
//!    at 10 000 rows and 180× at 1 000 000** — and the amplification is what
//!    makes A's byte cost grow with `log(rows)` instead of staying flat.
//! 2. **At 10 000 objects the tail buys nothing on bytes at all** (1.0×), and
//!    the crossover is near 20 000 rows: below it, redb's per-key amplification
//!    exceeds the whole Arrow table. The tail is a large-repository mechanism,
//!    and on a small repository it is paying for a curve it has not reached yet.
//!
//! **Deferral does not rescue the ratio and does not need to.** Arm A's unpaid
//! catch-up, timed explicitly after the loop and amortised over the 64 MiB
//! trigger period, adds 0.067 / 0.247 / 3.409 ms per push at the three sizes —
//! so the honest A figures are 2.18 / 3.16 / 9.14 ms and B/A at a million
//! objects is **79× rather than 127×**. Still two orders of magnitude.
//!
//! ### Real objects agree with the generator, once arrival order is fixed
//!
//! 116 345 real objects from every repository under `/home/rickard/git`
//! (`shuffle=1`, loadavg 2.58–2.63, so slightly noisier): 2.183 ms and 925 KB
//! per push at 10 000, 3.959 ms and 1.380 MB at 100 000 — within 35% of the
//! synthetic wall figures and within **0.5% of the synthetic byte figures**. The
//! generator is representative of this workload.
//!
//! The **first** real sweep was not, and the way it failed is worth keeping.
//! Fed in `sort -u` order, arm A wrote 66 KB per push instead of 925 KB and its
//! cost stopped growing with repository size entirely — because oid order *is*
//! the tail's B-tree key order, so every insert landed on the rightmost leaves.
//! A 14× byte difference and a changed asymptote, from arrival order alone. It
//! is reproducible with `shuffle=0`, and it says something real: the tail's
//! write cost is a fact about key locality, and a git push has none.
//!
//! ## LAW 2 — every guard here was seen RED before it was trusted
//!
//! Four guards, four deliberate breakages on oden 2026-08-08, each one restored
//! immediately after. A benchmark's asserts are the only thing standing between
//! "arm B is 40× more expensive" and "arm B quietly did less work", so they are
//! the part that most needs to have been watched failing.
//!
//! | # | the guard | the edit that broke it | what it said |
//! |---|---|---|---|
//! | 1 | arm B's file is byte-for-byte its index | `write_durable`: `f.write_all(&bytes[..bytes.len() / 2])` | `arm B's file is 70468 bytes but its index is 140936 — the write did not land` |
//! | 2 | arm A answers for every sampled oid | `arm_with_tail`: `stack.append(&p[..1])` | `arm A lost 3c53775474… — the tail is supposed to answer for everything ever appended` |
//! | 3 | the byte counter is not blind | `Io::now`: parse every field as `0` | `arm B charged 0 write bytes for an index of 140936 — a rewrite cannot cost less than one copy` |
//! | 4 | one timed sample per push | `arm_no_tail`: `for p in pushes.iter().take(pushes.len() - 1)` | `B  no tail timed 2 pushes, not 3 — a missing sample would flatten the curve` |
//! | 5 | arm A's pushes reached the tail | `arm_with_tail`: `let _ = p;` in place of the `append` | `arm A owed no catch-up after 3 pushes and never rebuilt — the appends did not reach the tail` |
//!
//! Guard 5's first red run **reported green, and it was the harness lying, not
//! the guard**. The crate did not compile at that moment — a concurrent change
//! in `src/git_ops.rs` was mid-flight — so `cargo run` printed compiler errors,
//! the `grep panicked` found nothing, and "no panic" was read as "guard did not
//! fire". LAW 2's fourth clause exactly: *check output, not exit code*; a
//! malformed run's empty output is shaped precisely like a clean result. The
//! run was repeated once the crate built and the guard spoke on the first try.
//!
//! Guard 4 is the one that had to be moved to earn its keep. In its first
//! position — after the statistics — the short series panicked inside
//! `slice_mean` with `range end index 3 out of range for slice of length 2`
//! first, so the guard never spoke. It is now the first thing `report` does,
//! ahead of every `Stat::of`. That is the LAW 2 trap in its own right: a guard
//! that is *shadowed* by an earlier crash is a guard that has never been seen
//! red, only inferred.
//!
//! Guard 3 exists because of the specific way this measurement could be hollow:
//! `/proc/self/io` is parsed by hand, and a parser that matched no field would
//! report zero bytes for both arms — a perfectly shaped, perfectly wrong "the
//! arms are identical". The identity value here is 0, and 0 is exactly what an
//! unmeasured arm looks like.
//!
//! Single-threaded throughout. (No rayon here or anywhere — LAW 3.)

use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::time::Instant;

use znippy_plugin_git::index_layout::{
    IndexEntry, ObjType, ObjectIndex, OneTableFourColumns, Rng, synthetic_entries,
};
use znippy_plugin_git::read_stack::{ObjectReadStack, RebuildTriggers};

type Stack = ObjectReadStack<OneTableFourColumns>;

/// Above this 1-minute loadavg a cell is reported but flagged. oden runs several
/// agents; refusing to measure at all would mean never measuring, so the figure
/// is taken and the condition is recorded with it.
const LOAD_QUIET: f64 = 1.5;

// ── the counters ──────────────────────────────────────────────────────────────

fn loadavg() -> (f64, String) {
    let s = fs::read_to_string("/proc/loadavg").unwrap_or_default();
    let one = s
        .split_whitespace()
        .next()
        .and_then(|x| x.parse::<f64>().ok())
        .unwrap_or(f64::NAN);
    (one, s.trim().to_string())
}

/// The two byte counters this benchmark lives or dies by, straight out of
/// `/proc/self/io`. `wchar` is what the process handed to `write(2)`;
/// `write_bytes` is what the kernel believes reached storage.
#[derive(Clone, Copy, Default, Debug)]
struct Io {
    wchar: u64,
    write_bytes: u64,
}

impl Io {
    fn now() -> Io {
        let mut io = Io::default();
        let s = fs::read_to_string("/proc/self/io").unwrap_or_default();
        for line in s.lines() {
            let mut it = line.split(':');
            let (Some(k), Some(v)) = (it.next(), it.next()) else {
                continue;
            };
            let v = v.trim().parse::<u64>().unwrap_or(0);
            match k {
                "wchar" => io.wchar = v,
                "write_bytes" => io.write_bytes = v,
                _ => {}
            }
        }
        io
    }
    fn since(self, t0: Io) -> Io {
        Io {
            wchar: self.wchar.saturating_sub(t0.wchar),
            write_bytes: self.write_bytes.saturating_sub(t0.write_bytes),
        }
    }
}

fn cpu_ns() -> u64 {
    let mut ts = libc::timespec {
        tv_sec: 0,
        tv_nsec: 0,
    };
    // SAFETY: `ts` is a live, correctly typed, writable `timespec`.
    unsafe { libc::clock_gettime(libc::CLOCK_PROCESS_CPUTIME_ID, &mut ts) };
    ts.tv_sec as u64 * 1_000_000_000 + ts.tv_nsec as u64
}

/// Bytes under `path`, whether it is one file or a directory of them.
fn disk_bytes(path: &Path) -> u64 {
    let Ok(md) = fs::metadata(path) else {
        return 0;
    };
    if md.is_file() {
        return md.len();
    }
    fs::read_dir(path)
        .map(|rd| {
            rd.filter_map(|e| e.ok())
                .map(|e| disk_bytes(&e.path()))
                .sum()
        })
        .unwrap_or(0)
}

// ── statistics ────────────────────────────────────────────────────────────────

#[derive(Clone, Copy)]
struct Stat {
    med: f64,
    lo: f64,
    hi: f64,
}

impl Stat {
    fn of(mut v: Vec<f64>) -> Stat {
        assert!(!v.is_empty(), "a statistic over no samples is not a number");
        v.sort_by(|a, b| a.partial_cmp(b).unwrap());
        Stat {
            med: v[v.len() / 2],
            lo: v[0],
            hi: v[v.len() - 1],
        }
    }
    /// The noise band: full spread as a fraction of the median.
    fn band(&self) -> f64 {
        if self.med == 0.0 {
            0.0
        } else {
            (self.hi - self.lo) / self.med
        }
    }
}

fn median(mut v: Vec<f64>) -> f64 {
    v.sort_by(|a, b| a.partial_cmp(b).unwrap());
    v[v.len() / 2]
}

// ── one arm's result ──────────────────────────────────────────────────────────

/// Per-push series. Every vector has one entry per push, in push order, which is
/// what makes the growth curve visible rather than an average of it.
struct ArmRun {
    wall_ms: Vec<f64>,
    cpu_ms: Vec<f64>,
    wchar: Vec<u64>,
    write_bytes: Vec<u64>,
    /// Rebuilds the arm ran during the timed loop. Arm B rebuilds on every push
    /// by definition; arm A only when a trigger fires, and that is the number
    /// the trigger defaults are supposed to keep small.
    rebuilds: u64,
    /// Bytes on disk at the end, the artefact the arm would restart from.
    disk: u64,
    /// Rows the arm can answer for at the end — the LAW 2 anchor: an arm that
    /// silently skipped its work has the wrong count here.
    rows: u64,
    /// **The catch-up arm A has not paid for yet**, measured after the timed
    /// loop: one explicit [`ObjectReadStack::rebuild`] over the repository as it
    /// then stands. Zero for arm B, which has no debt because it rebuilds inside
    /// every push.
    ///
    /// This is here so arm A cannot win by deferring. A 50-push run of small
    /// synthetic objects never reaches the 64 MiB volume trigger, so arm A's
    /// measured pushes would show a debt-free cost that a longer run would not
    /// have; amortising this over the trigger period is what makes the two arms
    /// comparable over a repository's life rather than over one window.
    rebuild_debt_ms: f64,
}

impl ArmRun {
    fn total_wall(&self) -> f64 {
        self.wall_ms.iter().sum()
    }
    fn total_wchar(&self) -> u64 {
        self.wchar.iter().sum()
    }
    /// Mean over pushes `[a, b)` — used to cut the run into thirds so the shape
    /// of the curve is readable without printing fifty lines.
    fn slice_mean(v: &[f64], a: usize, b: usize) -> f64 {
        let s: f64 = v[a..b].iter().sum();
        s / (b - a) as f64
    }
}

// ── the arms ──────────────────────────────────────────────────────────────────

/// **Arm A.** redb tail, shipped triggers, projection catches up inline when a
/// trigger fires.
fn arm_with_tail(
    dir: &Path,
    base: &[IndexEntry],
    pushes: &[Vec<IndexEntry>],
    sample: &[Vec<u8>],
) -> ArmRun {
    let _ = fs::remove_dir_all(dir);
    fs::create_dir_all(dir).expect("arm A scratch");
    let tail = dir.join("tail.redb");

    let stack = Stack::open(
        &tail,
        RebuildTriggers::default(),
        znippy_plugin_git::DEFAULT_REDB_CACHE_BYTES,
    )
    .expect("open tail");
    // Setup, untimed: the repository as it stood before the first push. Chunked
    // only so one redb transaction does not have to hold a million rows.
    for chunk in base.chunks(50_000) {
        stack.append(chunk).expect("seed the tail");
    }
    stack.rebuild().expect("seal the seeded projection");
    let rebuilds0 = stack.stats().rebuilds;

    let mut run = ArmRun {
        wall_ms: Vec::new(),
        cpu_ms: Vec::new(),
        wchar: Vec::new(),
        write_bytes: Vec::new(),
        rebuilds: 0,
        disk: 0,
        rows: 0,
        rebuild_debt_ms: 0.0,
    };

    for p in pushes {
        let io0 = Io::now();
        let c0 = cpu_ns();
        let t0 = Instant::now();
        stack.append(p).expect("append a push");
        let wall = t0.elapsed();
        let cpu = cpu_ns() - c0;
        let io = Io::now().since(io0);

        run.wall_ms.push(wall.as_secs_f64() * 1e3);
        run.cpu_ms.push(cpu as f64 / 1e6);
        run.wchar.push(io.wchar);
        run.write_bytes.push(io.write_bytes);
    }

    // The catch-up arm A still owes. Measured outside the timed loop and
    // reported separately, never folded silently into the per-push figure.
    let pre = stack.stats();
    let t0 = Instant::now();
    stack.rebuild().expect("the deferred rebuild");
    run.rebuild_debt_ms = t0.elapsed().as_secs_f64() * 1e3;
    // Arm A must end either holding un-absorbed rows or having just absorbed
    // them on a trigger — anything else means the appends never reached the
    // tail. Written as the disjunction on purpose: "unabsorbed > 0" alone is a
    // guard that fires spuriously whenever the final push happens to be the one
    // that trips the volume trigger, and a guard that cries wolf gets deleted.
    assert!(
        pre.unabsorbed_rows > 0 || pre.rebuilds > rebuilds0,
        "arm A owed no catch-up after {} pushes and never rebuilt — the appends \
         did not reach the tail",
        pushes.len()
    );

    let st = stack.stats();
    run.rebuilds = st.rebuilds - rebuilds0 - 1;
    run.rows = st.total_rows;
    run.disk = disk_bytes(dir);

    // LAW 2 anchor — applied output, not a state round-trip. Every sampled oid,
    // base and pushed alike, must resolve to the facts it was appended with.
    for oid in sample {
        let got = stack.lookup(oid);
        assert!(
            got.is_some(),
            "arm A lost {} — the tail is supposed to answer for everything ever appended",
            hex::encode(oid)
        );
    }
    run
}

/// **Arm B.** No tail: the whole Arrow index is rebuilt and rewritten per push.
///
/// The rows are kept in memory and extended in place, which is the strongest
/// honest form of this arm — a version that re-read them from disk each time
/// would be a strawman and is not what is measured here.
fn arm_no_tail(
    dir: &Path,
    base: &[IndexEntry],
    pushes: &[Vec<IndexEntry>],
    sample: &[Vec<u8>],
) -> ArmRun {
    let _ = fs::remove_dir_all(dir);
    fs::create_dir_all(dir).expect("arm B scratch");
    let index_path = dir.join("index.arrow");

    let mut rows: Vec<IndexEntry> = base.to_vec();
    // Setup, untimed: the index as it stood before the first push, on disk.
    let seed = OneTableFourColumns::build(&rows).expect("seed the index");
    write_durable(&index_path, seed.ipc_slice());
    drop(seed);

    let mut run = ArmRun {
        wall_ms: Vec::new(),
        cpu_ms: Vec::new(),
        wchar: Vec::new(),
        write_bytes: Vec::new(),
        rebuilds: 0,
        disk: 0,
        rows: 0,
        rebuild_debt_ms: 0.0,
    };
    let mut last: Option<OneTableFourColumns> = None;

    for p in pushes {
        let io0 = Io::now();
        let c0 = cpu_ns();
        let t0 = Instant::now();
        rows.extend_from_slice(p);
        let idx = OneTableFourColumns::build(&rows).expect("rebuild the index");
        write_durable(&index_path, idx.ipc_slice());
        let wall = t0.elapsed();
        let cpu = cpu_ns() - c0;
        let io = Io::now().since(io0);

        run.wall_ms.push(wall.as_secs_f64() * 1e3);
        run.cpu_ms.push(cpu as f64 / 1e6);
        run.wchar.push(io.wchar);
        run.write_bytes.push(io.write_bytes);
        run.rebuilds += 1;
        last = Some(idx);
    }

    let idx = last.expect("at least one push");
    run.rows = idx.len() as u64;
    run.disk = disk_bytes(dir);

    // LAW 2 anchor, three ways, all applied output:
    //  1. the file on disk is exactly the index's own IPC stream — an arm that
    //     wrote a short or stale file would differ here;
    //  2. the index really covers every row;
    //  3. every sampled oid resolves.
    let on_disk = fs::metadata(&index_path).expect("arm B index exists").len();
    assert_eq!(
        on_disk,
        idx.ipc_slice().len() as u64,
        "arm B's file is {on_disk} bytes but its index is {} — the write did not land",
        idx.ipc_slice().len()
    );
    assert!(
        run.total_wchar() >= idx.ipc_slice().len() as u64,
        "arm B charged {} write bytes for an index of {} — a rewrite cannot cost less than one copy",
        run.total_wchar(),
        idx.ipc_slice().len()
    );
    for oid in sample {
        assert!(
            idx.lookup(oid).is_some(),
            "arm B lost {} — a full rebuild cannot drop a row it was given",
            hex::encode(oid)
        );
    }
    run
}

/// Atomic, durable replacement — the same guarantee redb's commit gives arm A.
/// Anything less would let arm B win on durability it never paid for.
fn write_durable(path: &Path, bytes: &[u8]) {
    let tmp = path.with_extension("tmp");
    {
        let mut f = fs::File::create(&tmp).expect("create the temp index");
        f.write_all(bytes).expect("write the index");
        f.sync_all().expect("fsync the index");
    }
    fs::rename(&tmp, path).expect("rename the index into place");
    // The rename itself needs the directory entry on disk, or a crash leaves the
    // old file. redb fsyncs; so does this.
    if let Some(d) = path.parent()
        && let Ok(dh) = fs::File::open(d)
    {
        let _ = dh.sync_all();
    }
}

// ── workloads ─────────────────────────────────────────────────────────────────

/// Real objects, from
/// `git cat-file --batch-all-objects --batch-check='%(objectname) %(objecttype)
/// %(objectsize) %(objectsize:disk) %(deltabase)'` over the repositories under
/// `/home/rickard/git`, deduplicated by oid.
///
/// `objectsize:disk` is the stored (compressed / delta-encoded) length, which is
/// exactly [`IndexEntry::len`]; offsets are assigned by laying the objects out
/// back to back, which is what an archive does. `deltabase` is resolved to the
/// base's assigned offset in a second pass, so [`IndexEntry::delta_base`] is a
/// real offset of a real row of the same workload and not the identity value.
///
/// `git cat-file` reports the **resolved** type, so no `OfsDelta` / `RefDelta`
/// codes appear on this path — recorded rather than faked, since the type column
/// is one byte and cannot change what this benchmark measures.
fn load_real(path: &Path, oid_len: usize, shuffle: bool, seed: u64) -> Vec<IndexEntry> {
    let text = fs::read_to_string(path)
        .unwrap_or_else(|e| panic!("reading the real-object dump {}: {e}", path.display()));
    let mut out: Vec<IndexEntry> = Vec::new();
    let mut bases: Vec<Vec<u8>> = Vec::new();
    let mut off = 12u64;
    for line in text.lines() {
        let f: Vec<&str> = line.split_whitespace().collect();
        if f.len() < 5 {
            continue;
        }
        let Ok(oid) = hex::decode(f[0]) else { continue };
        if oid.len() != oid_len {
            continue;
        }
        let obj_type = match f[1] {
            "commit" => ObjType::Commit,
            "tree" => ObjType::Tree,
            "blob" => ObjType::Blob,
            "tag" => ObjType::Tag,
            _ => continue,
        };
        let uncompressed_size: u64 = f[2].parse().unwrap_or(0);
        let len: u64 = f[3].parse().unwrap_or(0);
        let base = hex::decode(f[4]).unwrap_or_default();
        out.push(IndexEntry {
            oid,
            offset: off,
            len,
            obj_type,
            uncompressed_size,
            delta_base: 0,
        });
        bases.push(base);
        off += len.max(1);
    }
    let mut where_: std::collections::HashMap<&[u8], u64> =
        std::collections::HashMap::with_capacity(out.len() * 2);
    for e in &out {
        where_.insert(e.oid.as_slice(), e.offset);
    }
    let resolved: Vec<u64> = bases
        .iter()
        .map(|b| {
            if b.iter().all(|&x| x == 0) {
                0
            } else {
                where_.get(b.as_slice()).copied().unwrap_or(0)
            }
        })
        .collect();
    for (e, base_off) in out.iter_mut().zip(resolved) {
        e.delta_base = base_off;
    }

    // **Arrival order, and why it is not the file's order.**
    //
    // The dump is deduplicated with `sort -u`, so it arrives in oid order — and
    // a tail fed in oid order is fed in *its own B-tree's* key order, which is
    // the single most flattering input redb can be given: every insert lands on
    // the rightmost leaves and dirties a handful of pages instead of one
    // root-to-leaf path per key. The first sweep taken this way measured 66 KB
    // per push against the shuffled 930 KB, a 14× difference that is a fact
    // about key locality and NOT a fact about git.
    //
    // A real push carries new objects whose sha1s are uniform over the whole
    // keyspace, in no order at all. So the corpus is shuffled, and the sorted
    // order is kept reachable (`shuffle=0`) because the contrast between the two
    // is itself the finding.
    //
    // Only the *order of the vector* changes. Offsets stay as assigned, so
    // `delta_base` still names a real row's real offset; `build` sorts by oid
    // and redb keys by oid, so nothing downstream depends on this order except
    // the thing under test.
    if shuffle {
        let mut rng = Rng(seed);
        for i in (1..out.len()).rev() {
            let j = (rng.next_u64() % (i as u64 + 1)) as usize;
            out.swap(i, j);
        }
    }
    out
}

// ── driver ────────────────────────────────────────────────────────────────────

fn arg(args: &[String], key: &str) -> Option<String> {
    args.iter()
        .find_map(|a| a.strip_prefix(&format!("{key}=")).map(|v| v.to_string()))
}

fn main() {
    // S-070: agent sandboxes set PR_SET_THP_DISABLE and children inherit it, so
    // every MADV_HUGEPAGE this bench's trees issue is a silent no-op and every
    // number here was taken THP-less without saying so. Clearing the flag is a
    // per-binary decision (the library must never flip process state); this
    // binary wants real numbers, and prints which kind it got.
    let thp = znippy_zoomies::stree::thp_enable_for_process();
    eprintln!("thp_enabled={}", thp);
    let args: Vec<String> = std::env::args().skip(1).collect();
    let bases: Vec<usize> = arg(&args, "bases")
        .unwrap_or_else(|| "10000,100000,1000000".into())
        .split(',')
        .filter_map(|s| s.trim().parse().ok())
        .collect();
    let pushes_n: usize = arg(&args, "pushes").and_then(|s| s.parse().ok()).unwrap_or(50);
    let push_n: usize = arg(&args, "push").and_then(|s| s.parse().ok()).unwrap_or(300);
    let runs: usize = arg(&args, "runs").and_then(|s| s.parse().ok()).unwrap_or(3);
    let oid_len: usize = arg(&args, "oid").and_then(|s| s.parse().ok()).unwrap_or(20);
    let seed: u64 = arg(&args, "seed").and_then(|s| s.parse().ok()).unwrap_or(0xB0BA);
    let scratch = PathBuf::from(
        arg(&args, "dir").unwrap_or_else(|| "/home/rickard/scratch/gitbench/tailwrite".into()),
    );
    let real = arg(&args, "real").map(PathBuf::from);
    let shuffle = arg(&args, "shuffle").map(|v| v != "0").unwrap_or(true);

    let (l0, lfull) = loadavg();
    println!("# tail_write_bench — what the redb tail buys on the write side");
    println!(
        "# loadavg at start: {lfull}  ({})",
        if l0 <= LOAD_QUIET { "quiet" } else { "LOADED" }
    );
    println!(
        "# workload: {} pushes of {push_n} objects, {runs} runs/cell, oid {oid_len}B, arm order rotates per run",
        pushes_n
    );

    let corpus: Option<Vec<IndexEntry>> = real.as_ref().map(|p| {
        let v = load_real(p, oid_len, shuffle, seed);
        println!(
            "# real corpus: {} objects from {}, mean stored len {} B, arrival order {}",
            v.len(),
            p.display(),
            v.iter().map(|e| e.len).sum::<u64>() / v.len().max(1) as u64,
            if shuffle {
                "SHUFFLED (what a push looks like)"
            } else {
                "oid-sorted (flatters the tail's B-tree — see load_real)"
            }
        );
        v
    });
    let kind = if corpus.is_some() { "real" } else { "synthetic" };
    println!("# entries: {kind}");
    println!();

    for &base_n in &bases {
        let total = base_n + pushes_n * push_n;
        let all: Vec<IndexEntry> = match &corpus {
            Some(c) => {
                if c.len() < total {
                    println!(
                        "## base {base_n}: SKIPPED — the real corpus has {} objects, this cell needs {total}",
                        c.len()
                    );
                    continue;
                }
                c[..total].to_vec()
            }
            None => synthetic_entries(total, oid_len, seed),
        };
        let base: Vec<IndexEntry> = all[..base_n].to_vec();
        let pushes: Vec<Vec<IndexEntry>> = all[base_n..]
            .chunks(push_n)
            .map(|c| c.to_vec())
            .collect();
        // A LAW 2 sample: oids from the seeded base, the first push and the last
        // push. Base-only would pass even for an arm that dropped every push.
        let sample: Vec<Vec<u8>> = [
            base.first(),
            base.last(),
            pushes[0].first(),
            pushes[0].last(),
            pushes[pushes.len() - 1].first(),
            pushes[pushes.len() - 1].last(),
        ]
        .iter()
        .filter_map(|e| e.map(|e| e.oid.clone()))
        .collect();
        let push_bytes: u64 = pushes[0].iter().map(|e| e.len).sum();

        println!(
            "## base {base_n} objects ({kind}) — {} pushes × {push_n}, {} stored bytes per push",
            pushes.len(),
            push_bytes
        );

        let mut a_runs: Vec<ArmRun> = Vec::new();
        let mut b_runs: Vec<ArmRun> = Vec::new();
        let mut loads: Vec<String> = Vec::new();

        for run in 0..runs {
            // Rotate: run 0 is A,B; run 1 is B,A; … Position bias is real here
            // (a measured ~1.1% second-position penalty in this crate) and this
            // is what cancels it instead of assuming it away.
            let a_first = run % 2 == 0;
            let dir_a = scratch.join(format!("a-{base_n}-{run}"));
            let dir_b = scratch.join(format!("b-{base_n}-{run}"));

            let (la, sa) = loadavg();
            loads.push(format!("run{run}:{sa}"));
            if la > LOAD_QUIET {
                println!("   [LOADED] run {run} starts at loadavg {la}");
            }
            if a_first {
                a_runs.push(arm_with_tail(&dir_a, &base, &pushes, &sample));
                b_runs.push(arm_no_tail(&dir_b, &base, &pushes, &sample));
            } else {
                b_runs.push(arm_no_tail(&dir_b, &base, &pushes, &sample));
                a_runs.push(arm_with_tail(&dir_a, &base, &pushes, &sample));
            }
            let _ = fs::remove_dir_all(&dir_a);
            let _ = fs::remove_dir_all(&dir_b);
        }

        report("A  with tail   ", &a_runs, pushes.len(), base_n as u64);
        report("B  no tail     ", &b_runs, pushes.len(), base_n as u64);
        curve(&a_runs[0], &b_runs[0], base_n, push_n);

        // Arm A's deferred catch-up, priced. The volume trigger fires once per
        // 64 MiB of appended object bytes, so a repository taking pushes of
        // `push_bytes` owes one rebuild every `period` of them; spreading the
        // measured rebuild over that period is arm A's true per-push cost, and
        // it is printed beside the undeferred one rather than instead of it.
        let debt = median(a_runs.iter().map(|r| r.rebuild_debt_ms).collect());
        let period = (RebuildTriggers::DEFAULT_TAIL_BYTES as f64 / push_bytes.max(1) as f64).max(1.0);
        let a_per_push = median(a_runs.iter().map(|r| r.total_wall()).collect()) / pushes.len() as f64;
        println!(
            "   A's deferred rebuild: {debt:8.1} ms over {} rows, one per {period:.1} pushes at this \
             push size → +{:.3} ms/push amortised (per-push {:.3} → {:.3} ms)",
            a_runs[0].rows,
            debt / period,
            a_per_push,
            a_per_push + debt / period,
        );

        let a_tot = median(a_runs.iter().map(|r| r.total_wall()).collect());
        let b_tot = median(b_runs.iter().map(|r| r.total_wall()).collect());
        let a_w = median(a_runs.iter().map(|r| r.total_wchar() as f64).collect());
        let b_w = median(b_runs.iter().map(|r| r.total_wchar() as f64).collect());
        println!(
            "   ratio B/A: wall {:.1}×, bytes written {:.1}×",
            b_tot / a_tot,
            b_w / a_w
        );
        for l in &loads {
            println!("   loadavg {l}");
        }
        println!();
    }
}

/// **The curve itself**, sampled — the claim is about shape, and a mean cannot
/// show a shape. Eight points from run 0 of each arm, with the repository size
/// each push was made against, so "cost tracks what changed" and "cost tracks
/// total size" are distinguishable by eye and not only by a ratio.
fn curve(a: &ArmRun, b: &ArmRun, base_n: usize, push_n: usize) {
    let n = a.wall_ms.len();
    let step = (n / 8).max(1);
    println!(
        "   {:>6}  {:>10}  {:>10} {:>12}   {:>10} {:>12}",
        "push", "repo rows", "A wall ms", "A bytes", "B wall ms", "B bytes"
    );
    let mut i = 0;
    while i < n {
        println!(
            "   {:>6}  {:>10}  {:>10.3} {:>12}   {:>10.3} {:>12}",
            i,
            base_n + (i + 1) * push_n,
            a.wall_ms[i],
            a.wchar[i],
            b.wall_ms[i],
            b.wchar[i]
        );
        i += step;
    }
    if n > 0 && (n - 1) % step != 0 {
        let i = n - 1;
        println!(
            "   {:>6}  {:>10}  {:>10.3} {:>12}   {:>10.3} {:>12}",
            i,
            base_n + (i + 1) * push_n,
            a.wall_ms[i],
            a.wchar[i],
            b.wall_ms[i],
            b.wchar[i]
        );
    }
}

/// One arm's cell: the totals, the noise band across runs, and the per-push
/// curve cut into thirds — which is the only part that can tell a flat cost from
/// one that tracks repository size.
fn report(name: &str, runs: &[ArmRun], n_pushes: usize, base_n: u64) {
    // First, before any statistic is taken over the series: every run must have
    // one sample per push and must have ended larger than it started. Both are
    // applied output — samples that exist, rows the arm can answer for — and
    // both come first because a statistic over a short series is still a
    // number, and an arm that quietly skipped its last push would otherwise be
    // reported as merely a little cheaper.
    for r in runs {
        assert_eq!(
            r.wall_ms.len(),
            n_pushes,
            "{name} timed {} pushes, not {n_pushes} — a missing sample would flatten the curve",
            r.wall_ms.len()
        );
        assert!(
            r.rows > base_n,
            "{name} ended with {} rows over a base of {base_n} — the pushes did not land",
            r.rows
        );
    }

    let wall = Stat::of(runs.iter().map(|r| r.total_wall()).collect());
    let cpu = Stat::of(runs.iter().map(|r| r.cpu_ms.iter().sum::<f64>()).collect());
    let wchar = Stat::of(runs.iter().map(|r| r.total_wchar() as f64).collect());
    let wbytes = Stat::of(
        runs.iter()
            .map(|r| r.write_bytes.iter().sum::<u64>() as f64)
            .collect(),
    );
    let third = (n_pushes / 3).max(1);
    let first3 = Stat::of(
        runs.iter()
            .map(|r| ArmRun::slice_mean(&r.wall_ms, 0, third))
            .collect(),
    );
    let last3 = Stat::of(
        runs.iter()
            .map(|r| ArmRun::slice_mean(&r.wall_ms, n_pushes - third, n_pushes))
            .collect(),
    );
    let r0 = &runs[0];

    println!(
        "   {name} total wall {:8.1} ms [{:.1}–{:.1}, band {:>5.1}%]  cpu {:8.1} ms  \
         wchar {:>12} B [band {:>5.1}%]  write_bytes {:>12} B  disk {:>12} B  rows {}  rebuilds {}",
        wall.med,
        wall.lo,
        wall.hi,
        wall.band() * 100.0,
        cpu.med,
        wchar.med as u64,
        wchar.band() * 100.0,
        wbytes.med as u64,
        r0.disk,
        r0.rows,
        r0.rebuilds,
    );
    println!(
        "   {:<15} per push: mean {:8.3} ms   first {third} pushes {:8.3} ms   last {third} pushes {:8.3} ms   \
         growth last/first {:5.2}×   bytes/push {:>10} B",
        "",
        wall.med / n_pushes as f64,
        first3.med,
        last3.med,
        last3.med / first3.med,
        (wchar.med / n_pushes as f64) as u64,
    );
}