znippy-plugin-git 0.1.1

//! **Two concurrent writers on one store root** — `P-014`.
//!
//! Nobody had ever run this. The defect was that `znippy` had no writer lock of
//! any kind: an exhaustive grep for `flock` / `LOCK_EX` / `try_lock` / `O_EXCL`
//! over `znippy-plugin-git/src/` returned zero hits, and two concurrent appends
//! to one store root were undefined.
//!
//! # The two cases are different and both are here
//!
//! 1. **Two processes, one store root.** There *was* accidental protection —
//!    redb takes `flock(LOCK_EX | LOCK_NB)` on its file
//!    (`redb-2.6.3/.../file_backend/unix.rs:37`) and a store opens two databases
//!    — so a second `GitStore::open` already failed. But that lock guards
//!    `objects.tail`, **not `objects.pack`**, and it is taken *after* the writer
//!    is already open. [`two_processes_appending_to_one_objects_pack_are_refused`]
//!    drives the writer layer with no redb anywhere near it, which is what that
//!    protection was hiding.
//! 2. **Two concurrent pushes inside ONE process**, one store handle, one set of
//!    open fds. `flock` **cannot** help here — an advisory lock is per open file
//!    *description*, so two threads sharing one fd are not excluded from each
//!    other. [`two_concurrent_pushes_on_one_store_handle_keep_every_object`] is
//!    that case, and what it found is written on it.
//!
//! # Every assertion here is on the ARCHIVE CONTENTS
//!
//! Not on the absence of an error, because **neither writer errors — that is the
//! whole problem.** [`audit`] reads the files back off the disk and asserts the
//! journal's extents are pairwise disjoint, that the bytes at each one are the
//! pack that was acked for it, that every object resolves to the body it was
//! pushed with, that the index's extents fall inside the journal's, and that
//! stock `git index-pack --strict` and `git fsck --strict` accept a pack emitted
//! from the result. A test that ran two pushes and checked both returned `Ok`
//! would prove nothing.
//!
//! No assertion here reads a clock and nothing sleeps. The two writers are
//! released by a **file rendezvous** the parent sequences; the only deadline in
//! the file exists to turn a hung child into a failure and can never turn one
//! into a pass.

use std::collections::BTreeMap;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{Duration, Instant};

use znippy_plugin_git::archive_write::{acked_packs, read_journal};
use znippy_plugin_git::{
    Caps, GitHashKind, GitObjectKind, GitOps, GitStore, WriterArm, canonical,
};

/// Packs per writer. Large enough that a collision cannot hide in one entry and
/// small enough that the whole file runs in a few seconds.
const PER_WRITER: usize = 24;

/// The env var that turns this test binary into one of the two writers.
const CHILD: &str = "AN_P014_CHILD";

// ── the fixture ─────────────────────────────────────────────────────────────

/// One real one-blob packfile and the oid of the blob inside it.
///
/// Built here rather than shelling to `git` so the two writers can be released
/// against each other without a subprocess of their own in the way.
fn one_blob_pack(body: &[u8]) -> (Vec<u8>, Vec<u8>) {
    let mut pack = b"PACK".to_vec();
    pack.extend_from_slice(&2u32.to_be_bytes());
    pack.extend_from_slice(&1u32.to_be_bytes());
    let mut size = body.len() as u64;
    let mut header = vec![(3u8 << 4) | (size as u8 & 0x0f)];
    size >>= 4;
    while size > 0 {
        let last = header.len() - 1;
        header[last] |= 0x80;
        header.push((size & 0x7f) as u8);
        size >>= 7;
    }
    pack.extend_from_slice(&header);
    let mut e = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
    e.write_all(body).unwrap();
    pack.extend_from_slice(&e.finish().unwrap());
    pack.extend_from_slice(&[0u8; 20]);
    let oid = GitHashKind::Sha1.oid_of(&canonical(GitObjectKind::Blob, body));
    (pack, oid)
}

/// This writer's `i`th blob. Distinct per writer and per index, and of a
/// *varying* length, so a pack that lands at the wrong offset cannot be mistaken
/// for the pack that belonged there.
fn body_of(tag: &str, i: usize) -> Vec<u8> {
    format!("P-014 writer {tag} object {i} {}", "z".repeat(64 + i * 13)).into_bytes()
}

/// One writer's whole workload: `(pack bytes, oid, body)`.
fn workload(tag: &str, n: usize) -> Vec<(Vec<u8>, Vec<u8>, Vec<u8>)> {
    (0..n)
        .map(|i| {
            let b = body_of(tag, i);
            let (p, o) = one_blob_pack(&b);
            (p, o, b)
        })
        .collect()
}

// ── what a writer reported it was told was durable ──────────────────────────

/// One acked push, as the writer that made it saw it: the oid it stored, the
/// pack bytes it handed over, and the extent it was told they live at.
#[derive(Clone, Debug)]
struct Acked {
    oid: Vec<u8>,
    body: Vec<u8>,
    pack: Vec<u8>,
    offset: u64,
    len: u64,
}

fn write_receipt(path: &Path, acked: &[Acked]) {
    let mut s = String::new();
    for a in acked {
        s.push_str(&format!("{} {} {}\n", hex::encode(&a.oid), a.offset, a.len));
    }
    std::fs::write(path, s).unwrap();
}

/// Read a child's receipt back and re-derive the bytes it must have pushed. The
/// pack is rebuilt from the oid rather than shipped through the file, so the
/// parent's expectation cannot be an echo of the child's claim.
fn read_receipt(path: &Path, tags: &[&str]) -> Vec<Acked> {
    let mut by_oid: BTreeMap<Vec<u8>, (Vec<u8>, Vec<u8>)> = BTreeMap::new();
    for t in tags {
        for (p, o, b) in workload(t, PER_WRITER) {
            by_oid.insert(o, (p, b));
        }
    }
    let Ok(text) = std::fs::read_to_string(path) else {
        return Vec::new();
    };
    text.lines()
        .filter(|l| !l.trim().is_empty())
        .map(|l| {
            let mut f = l.split_whitespace();
            let oid = hex::decode(f.next().unwrap()).unwrap();
            let offset: u64 = f.next().unwrap().parse().unwrap();
            let len: u64 = f.next().unwrap().parse().unwrap();
            let (pack, body) = by_oid
                .get(&oid)
                .unwrap_or_else(|| panic!("a writer acked an oid no workload produced"))
                .clone();
            Acked {
                oid,
                body,
                pack,
                offset,
                len,
            }
        })
        .collect()
}

// ── THE AUDIT — every assertion is on the archive as it lies on disk ────────

/// **The deliverable assertion.** `acked` is every push both writers were told
/// succeeded; this asserts the archive actually holds all of them.
///
/// It is deliberately split so a failure names which invariant broke:
///
/// 1. the journal claims exactly as many packs as were acked — nothing lost;
/// 2. no two claimed extents overlap, and none runs past the file — this is
///    where two processes sharing a stale cursor die;
/// 3. the bytes **at** each claimed extent are the pack that was acked for it —
///    this is where "the write landed, on top of somebody else's" dies;
/// 4. every object resolves, through a freshly opened store, to the body it was
///    pushed with;
/// 5. the index's extent for each object falls inside the journal's extent for
///    the pack that carried it — the index agrees with the pack;
/// 6. stock git accepts a pack emitted from the result: `index-pack --strict`
///    then `fsck --strict`.
fn audit(root: &Path, acked: &[Acked]) {
    assert!(
        !acked.is_empty(),
        "no writer acked anything — this test proves nothing unless at least one \
         of the two got through"
    );

    let blobs = root.join("objects.pack");
    let on_disk = std::fs::read(&blobs).expect("objects.pack");
    let journal = read_journal(&root.join("objects.pack.journal")).expect("journal");
    let claimed = acked_packs(&journal);

    // (1) nothing lost between the ack and the durable record of it.
    assert_eq!(
        claimed.len(),
        acked.len(),
        "the journal claims {} packs but {} pushes were acked — an acked push is \
         missing from the durable record",
        claimed.len(),
        acked.len()
    );

    // (2) the extents partition the blob region.
    let mut sorted = claimed.clone();
    sorted.sort_unstable();
    for w in sorted.windows(2) {
        let (a, b) = (w[0], w[1]);
        assert!(
            a.0 + a.1 <= b.0,
            "two acked packs share bytes: ({}, {}) overlaps ({}, {}). Both writers \
             were told their push was durable and they wrote on top of each other.",
            a.0,
            a.1,
            b.0,
            b.1
        );
    }
    if let Some(&(o, l)) = sorted.last() {
        assert!(
            o + l <= on_disk.len() as u64,
            "the journal claims bytes {}..{} but objects.pack is only {} B — an acked \
             extent points past the end of the archive",
            o,
            o + l,
            on_disk.len()
        );
    }

    // (3) THE CONTENT ASSERTION. The bytes at the extent are the pack that was
    //     acked for it, byte for byte.
    let mut lost = Vec::new();
    for a in acked {
        let end = (a.offset + a.len) as usize;
        if end > on_disk.len() || &on_disk[a.offset as usize..end] != &a.pack[..] {
            lost.push(hex::encode(&a.oid));
        }
    }
    assert!(
        lost.is_empty(),
        "{} of {} acked packs are NOT at the extent they were acked at — the bytes \
         were overwritten by the other writer and nothing errored. Lost oids: {lost:?}",
        lost.len(),
        acked.len()
    );

    // (4) + (5) a freshly opened store answers for every one of them.
    let store = GitStore::open(root, "rickard").expect("reopen the store over the result");
    store.wait_indexed();
    store.absorb_pending().expect("absorb");

    let mut unreadable = Vec::new();
    for a in acked {
        match store.content(&a.oid) {
            Ok(Some((GitObjectKind::Blob, got))) if got == a.body => {}
            other => unreadable.push((hex::encode(&a.oid), format!("{other:?}").len())),
        }
        let row = store
            .get(&a.oid)
            .expect("get")
            .unwrap_or_else(|| panic!("{} is not in the index", hex::encode(&a.oid)));
        let (ro, rl) = row.extent;
        assert!(
            ro >= a.offset && ro + rl <= a.offset + a.len,
            "the index puts {} at ({ro}, {rl}), outside the ({}, {}) the journal \
             claims for its pack — the index and the pack disagree",
            hex::encode(&a.oid),
            a.offset,
            a.len
        );
    }
    assert!(
        unreadable.is_empty(),
        "{} of {} acked objects do not read back as the body they were pushed with: {:?}",
        unreadable.len(),
        acked.len(),
        unreadable.iter().map(|(o, _)| o).collect::<Vec<_>>()
    );

    // (6) stock git, not our own reader agreeing with our own writer.
    let oids: Vec<&[u8]> = acked.iter().map(|a| a.oid.as_slice()).collect();
    let mut emitted = Vec::new();
    store
        .emit_oids(&oids, &[], &Caps::modern(), &mut emitted)
        .expect("emit a pack of everything that survived");
    fsck_strict(root, &emitted, acked.len());
}

/// `git index-pack --strict` and then `git fsck --strict`, on a pack emitted
/// from what survived.
fn fsck_strict(root: &Path, pack: &[u8], objects: usize) {
    let scratch = root.join("fsck");
    std::fs::create_dir_all(&scratch).unwrap();
    // 🔴 Through the shared oracle since 2026-08-11. This ran
    // `index-pack --strict` with `current_dir(&scratch)`, and `scratch` is a
    // plain directory: outside a repository that command **segfaults** on any
    // pack it would have rejected — exit 139 with no output — so this arbiter
    // could report a crash and never a reason. Everything that survived is
    // emitted, so the set is closed and `Connected` is the question a clone
    // would ask.
    znippy_plugin_git::git_oracle::assert_git_accepts(
        &scratch,
        "surviving.git",
        pack,
        znippy_plugin_git::git_oracle::Strictness::Connected,
    );

    let repo = scratch.join("repo");
    std::fs::create_dir_all(&repo).unwrap();
    let init = Command::new("git")
        .args(["init", "-q", "--bare", "."])
        .current_dir(&repo)
        .output()
        .expect("git init");
    assert!(init.status.success());
    let mut unpack = Command::new("git")
        .args(["unpack-objects", "-q"])
        .current_dir(&repo)
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .spawn()
        .expect("git unpack-objects");
    unpack.stdin.take().unwrap().write_all(pack).unwrap();
    let up = unpack.wait_with_output().unwrap();
    assert!(
        up.status.success(),
        "git unpack-objects refused the surviving archive:\n{}",
        String::from_utf8_lossy(&up.stderr)
    );

    let loose = count_loose(&repo.join("objects"));
    assert_eq!(
        loose, objects,
        "git unpacked {loose} objects out of an archive that acked {objects}"
    );

    let fsck = Command::new("git")
        .args(["fsck", "--strict", "--no-progress"])
        .current_dir(&repo)
        .output()
        .expect("git fsck");
    let stderr = String::from_utf8_lossy(&fsck.stderr);
    assert!(
        fsck.status.success(),
        "git fsck --strict is not clean:\n{stderr}"
    );
    for bad in ["error", "corrupt", "missing", "broken"] {
        assert!(
            !stderr.to_lowercase().contains(bad),
            "git fsck --strict reported `{bad}`:\n{stderr}"
        );
    }
}

fn count_loose(objects: &Path) -> usize {
    let mut n = 0;
    let Ok(rd) = std::fs::read_dir(objects) else {
        return 0;
    };
    for e in rd.flatten() {
        let name = e.file_name();
        let name = name.to_string_lossy();
        if name.len() == 2 && name.chars().all(|c| c.is_ascii_hexdigit()) {
            n += std::fs::read_dir(e.path()).map(|d| d.flatten().count()).unwrap_or(0);
        }
    }
    n
}

// ── the rendezvous, and the two children ────────────────────────────────────

/// Wait for a path to exist. This is a **rendezvous**, not a clock: the deadline
/// exists so a child that never arrives fails the run instead of hanging it, and
/// no assertion in this file can be satisfied by it elapsing.
fn await_path(p: &Path, what: &str) {
    let deadline = Instant::now() + Duration::from_secs(120);
    while !p.exists() {
        assert!(
            Instant::now() < deadline,
            "{what} never appeared at {} — the peer writer never reached the \
             rendezvous, so nothing was made concurrent and this test proves nothing",
            p.display()
        );
        std::thread::yield_now();
    }
}

fn tmproot(tag: &str) -> PathBuf {
    let d = std::env::temp_dir().join(format!(
        "p014-{tag}-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&d).unwrap();
    d
}

/// The other half of both cross-process tests. Re-executed by the parent with
/// [`CHILD`] set; a no-op in an ordinary run, which is why it asserts nothing.
#[test]
fn p014_child_writer() {
    let Ok(spec) = std::env::var(CHILD) else {
        return; // an ordinary `cargo test` run: this test does nothing.
    };
    // `<root>|<layer>|<arm>|<tag>`
    let f: Vec<&str> = spec.split('|').collect();
    let (root, layer, arm, tag) = (Path::new(f[0]), f[1], f[2], f[3]);
    let ready = root.join(format!("ready.{tag}"));
    let receipt = root.join(format!("receipt.{tag}"));
    let go = root.join("go");
    let work = workload(tag, PER_WRITER);
    let mut acked: Vec<Acked> = Vec::new();

    // Open FIRST, then announce. Both writers must hold their handle before
    // either appends — that is the whole condition the defect needs, and
    // announcing after the open is what makes it a state rather than a race.
    let opened: Result<Box<dyn FnMut(&[u8]) -> anyhow::Result<(u64, u64)>>, String> = match layer {
        "writer" => {
            let armv = WriterArm::parse(arm).unwrap();
            match armv.create(&root.join("objects.pack")) {
                Ok(w) => Ok(Box::new(move |b: &[u8]| w.append(b))),
                Err(e) => Err(format!("{e:#}")),
            }
        }
        "store" => match GitStore::open(root, "rickard") {
            Ok(s) => Ok(Box::new(move |b: &[u8]| {
                let tx = s.put(b, &[])?;
                Ok(tx.extent.expect("a pack push records its extent"))
            })),
            Err(e) => Err(format!("{e:#}")),
        },
        l => panic!("unknown layer {l}"),
    };

    let refusal = root.join(format!("refused.{tag}"));
    match opened {
        Err(ref why) => std::fs::write(&refusal, why).unwrap(),
        Ok(_) => {}
    }
    std::fs::write(&ready, b"").unwrap();
    await_path(&go, "the parent's go");

    if let Ok(mut append) = opened {
        for (pack, oid, body) in &work {
            match append(pack) {
                Ok((offset, len)) => acked.push(Acked {
                    oid: oid.clone(),
                    body: body.clone(),
                    pack: pack.clone(),
                    offset,
                    len,
                }),
                // A refusal is a correct outcome and is not an ack. It is simply
                // not written to the receipt, and the audit therefore never
                // expects it in the archive.
                Err(_) => break,
            }
        }
    }
    write_receipt(&receipt, &acked);
    std::fs::write(root.join(format!("done.{tag}")), b"").unwrap();
}

/// Spawn both writers, hold them at the rendezvous, release them together, and
/// hand back everything they were told was durable.
fn drive_two(root: &Path, layer: &str, arm: &str) -> Vec<Acked> {
    let exe = std::env::current_exe().expect("current_exe");
    let mut kids = Vec::new();
    for tag in ["A", "B"] {
        let child = Command::new(&exe)
            .args(["--exact", "p014_child_writer", "--nocapture"])
            .env(
                CHILD,
                format!("{}|{layer}|{arm}|{tag}", root.display()),
            )
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .expect("spawn the peer writer");
        // Sequenced, not timed: the next child is not started until this one is
        // holding its handle, and neither appends until `go`.
        await_path(&root.join(format!("ready.{tag}")), "a writer's handle");
        kids.push(child);
    }
    std::fs::write(root.join("go"), b"").unwrap();
    for (i, k) in kids.into_iter().enumerate() {
        let out = k.wait_with_output().expect("peer writer");
        assert!(
            out.status.success(),
            "peer writer {} died: {}",
            ["A", "B"][i],
            String::from_utf8_lossy(&out.stderr)
        );
    }
    let mut acked = read_receipt(&root.join("receipt.A"), &["A", "B"]);
    acked.extend(read_receipt(&root.join("receipt.B"), &["A", "B"]));
    acked
}

fn refusal(root: &Path, tag: &str) -> Option<String> {
    std::fs::read_to_string(root.join(format!("refused.{tag}"))).ok()
}

// ── CASE 1 — two processes, one store root ──────────────────────────────────

/// **Two processes appending to one `objects.pack`, with redb nowhere near it.**
///
/// This is the case the accidental protection was hiding. Every arm reserves its
/// extent from an `AtomicU64` seeded in `open_blobs` from the file's length *at
/// open time*, so two processes that both open before either appends hold two
/// cursors with the same value and hand out **the same offsets**.
///
/// # Seen RED, on t14s, 2026-08-11
///
/// With no lock, on `safe`, both writers acked all 24 of their packs and neither
/// errored:
///
/// ```text
/// APPEND A 0  37ac1ad6… (0, 72)      APPEND B 0  8b1d9014… (0, 72)
/// APPEND A 1  06e5f245… (72, 72)     APPEND B 1  62ea91f6… (72, 72)
/// …                                  …
/// ```
///
/// and the audit's step (2) fired:
///
/// ```text
/// two acked packs share bytes: (0, 72) overlaps (0, 72). Both writers were told
/// their push was durable and they wrote on top of each other.
/// ```
///
/// with, measured on the same run at 16 packs each, **16 of 32 acked objects
/// silently destroyed** and `objects.pack` never growing past one writer's worth.
///
/// # What must happen instead
///
/// One writer gets the archive; the other is **refused at open** and appends
/// nothing. The refusal must name the blob file, not redb's tail, because a
/// refusal that came from the tail would evaporate the day the tail moved.
#[test]
fn two_processes_appending_to_one_objects_pack_are_refused() {
    for arm in ["fast", "safe"] {
        let root = tmproot(&format!("proc-{arm}"));
        let acked = drive_two(&root, "writer", arm);

        // THE CONTENTS ASSERTION FIRST. Whatever the two writers were told was
        // durable, the archive must hold — that is the whole of P-014, and it is
        // what fires when the lock is taken out. The refusal accounting below is
        // the *mechanism*; this is the *property*.
        //
        // `fast` keeps no journal, so the audit's journal half does not apply to
        // it; the bytes half does, and the bytes are the half that was corrupt.
        if arm == "safe" {
            audit(&root, &acked);
        } else {
            let on_disk = std::fs::read(root.join("objects.pack")).unwrap();
            let mut lost = Vec::new();
            for a in &acked {
                let end = (a.offset + a.len) as usize;
                if end > on_disk.len() || on_disk[a.offset as usize..end] != a.pack[..] {
                    lost.push(hex::encode(&a.oid));
                }
            }
            assert!(
                lost.is_empty(),
                "[{arm}] {} of {} acked packs are NOT at the extent they were acked \
                 at — the bytes were overwritten by the other writer and nothing \
                 errored. Lost oids: {lost:?}",
                lost.len(),
                acked.len()
            );
        }

        // The mechanism: one writer owns the archive, the other never got a
        // handle. Both halves matter — a store that refused *both* would be
        // "safe" and useless.
        let refused: Vec<&str> = ["A", "B"]
            .into_iter()
            .filter(|t| refusal(&root, t).is_some())
            .collect();
        assert_eq!(
            refused.len(),
            1,
            "[{arm}] exactly one of the two writers must be refused the archive; \
             {} were. Refusals: {:?}",
            refused.len(),
            ["A", "B"].map(|t| refusal(&root, t))
        );
        let why = refusal(&root, refused[0]).unwrap();
        assert!(
            why.contains("objects.pack"),
            "[{arm}] the refusal must name the blob file it is protecting, or the \
             archive is only as safe as whatever else happened to be locked: {why}"
        );
        assert_eq!(
            acked.len(),
            PER_WRITER,
            "[{arm}] exactly one writer's whole workload must have been acked"
        );
    }
}

/// **Two processes calling `GitStore::open` on one root.**
///
/// This one was already green, and it is here to say *why* and to keep it that
/// way. redb's own `flock` on `objects.tail` refused the second open — verified,
/// not assumed:
///
/// ```text
/// OPEN-ERR B opening object tail at …/objects.tail: Database already open. Cannot acquire lock.
/// ```
///
/// That protection is **incidental**: it guards redb's files, it is taken *after*
/// the writer is already open, and it would vanish the day the tail moved off
/// redb. So the assertion is not "the second open failed" — it is that the second
/// open failed **naming the blob file**, which is only true if the writer lock is
/// the thing that refused it.
#[test]
fn two_processes_opening_one_store_root_are_refused_by_the_writer_lock() {
    let root = tmproot("proc-store");
    let acked = drive_two(&root, "store", "safe");

    let refused: Vec<&str> = ["A", "B"]
        .into_iter()
        .filter(|t| refusal(&root, t).is_some())
        .collect();
    assert_eq!(refused.len(), 1, "exactly one store open must be refused");
    let why = refusal(&root, refused[0]).unwrap();
    assert!(
        why.contains("objects.pack"),
        "the second open was refused by something other than the writer lock — \
         this store is protected by an accident that a change to the tail would \
         remove: {why}"
    );

    assert_eq!(acked.len(), PER_WRITER);
    audit(&root, &acked);
}

// ── CASE 2 — two concurrent pushes inside ONE process ───────────────────────

/// **One store handle, one set of open fds, two threads pushing at once.**
///
/// This is the case a real forge hits first, and `flock` **cannot** cover it: an
/// advisory lock is per open file description, and two threads sharing one fd are
/// not excluded from each other. Verified rather than recited — `File::try_lock`
/// on the same handle a second time returns `Ok`, and only a *second open*
/// returns `WouldBlock`.
///
/// # What this found, and it is a finding rather than a fix
///
/// **It does not corrupt, and it never could.** All three arms reserve their
/// extent with one `cursor.fetch_add` on a single `AtomicU64` that every thread
/// sharing the handle shares too, and then `pwrite` positionally — so two
/// concurrent appends touch provably disjoint ranges and never share a file
/// offset. The journal behind both durable arms is under a `Mutex`, the ref log
/// is under `GitStore::ref_gate`, and the absorber is under `Absorber::gate`.
///
/// So **no in-process lock was added**, because none was missing. Adding a mutex
/// around `append` here would have serialised the ack path — the latency path —
/// to fix a race that the atomic cursor had already made impossible.
///
/// The test stays, because that is a property of the code and not a fact about
/// it: the day somebody replaces `fetch_add` with a read-modify-write of a plain
/// `u64`, or moves the offset into the file, this goes red.
#[test]
fn two_concurrent_pushes_on_one_store_handle_keep_every_object() {
    let root = tmproot("threads");
    let store = GitStore::open(&root, "rickard").unwrap();
    let work: Vec<Vec<(Vec<u8>, Vec<u8>, Vec<u8>)>> = ["T0", "T1"]
        .iter()
        .map(|t| workload(t, PER_WRITER))
        .collect();

    // A rendezvous, not a delay: both threads are inside `put` before either is
    // allowed to finish opening the race window.
    let barrier = std::sync::Barrier::new(2);
    let acked = std::sync::Mutex::new(Vec::<Acked>::new());

    std::thread::scope(|s| {
        for w in &work {
            let (store, barrier, acked) = (&store, &barrier, &acked);
            s.spawn(move || {
                barrier.wait();
                for (pack, oid, body) in w {
                    let tx = store
                        .put(pack, &[])
                        .expect("a concurrent push must not be refused inside one process");
                    let (offset, len) = tx.extent.expect("a pack push records its extent");
                    acked.lock().unwrap().push(Acked {
                        oid: oid.clone(),
                        body: body.clone(),
                        pack: pack.clone(),
                        offset,
                        len,
                    });
                }
            });
        }
    });
    store.wait_indexed();
    store.absorb_pending().unwrap();

    let acked = acked.into_inner().unwrap();
    assert_eq!(
        acked.len(),
        2 * PER_WRITER,
        "both threads must have been served — this is not a place to serialise"
    );
    drop(store);
    audit(&root, &acked);
}