plugins/native/znippy-plugin-git/tests/concurrent_push.rs
//! **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, "alice").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, "alice") {
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, "alice").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);
}