crates/mimir-ingest/tests/ingest/shared.rs
//! The handle model (wiring map R13): ONE writer, many readers at a committed
//! state, in one process — and a typed refusal for a second opener.
//!
//! A gated dumper holds the consumer INSIDE a commit (the dump step, where a
//! real consume spends its minutes) while the test reads the same lake from
//! another thread.
use std::path::Path;
use std::sync::mpsc::{Receiver, Sender, channel};
use std::time::Duration;
use anyhow::Result;
use mimir_ingest::{CommitDump, Consumer, Dumper, EventLog, LakeHeld, RepoSpec, WriterSeatTaken};
use super::common::*;
const REPO: &str = "fixture";
const WAIT: Duration = Duration::from_secs(60);
/// The fake dumper, stopping inside its `hold_at`-th dump until told to go.
struct Gated {
inner: FakeDumper,
hold_at: usize,
entered: Sender<String>,
go: Receiver<()>,
}
impl Dumper for Gated {
fn dump(&mut self, worktree: &Path, sha: &str) -> Result<CommitDump> {
if self.inner.calls.len() + 1 == self.hold_at {
self.entered.send(sha.to_string()).unwrap();
self.go.recv_timeout(WAIT).expect("the test never released the gated dump");
}
self.inner.dump(worktree, sha)
}
}
fn rt() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap()
}
/// c1 code · c2 docs · c3 code · c4 code — four releases.
fn history(src: &Scratch) -> Vec<String> {
let mut shas = Vec::new();
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n").write("crates/a/src/lib.rs", "fn f () | one\n");
shas.push(src.commit("c1"));
src.write("README.md", "docs\n");
shas.push(src.commit("c2 docs"));
src.write("crates/a/src/lib.rs", "fn f () | one\nfn g () | two\n");
shas.push(src.commit("c3"));
src.write("crates/a/src/lib.rs", "fn f () | one\nfn g () | two\nfn h () | three\n");
shas.push(src.commit("c4"));
shas
}
fn spec(tmp: &Path, src: &Scratch) -> RepoSpec {
RepoSpec { name: REPO.into(), source: src.root.to_string_lossy().into_owned(), branch: "main".into(), worktree: tmp.join("clone"), initial_offset: None }
}
/// (offset seq, ledger rows, open node keys) of the repo as `log` sees it.
fn state(log: &EventLog) -> (i64, usize, usize) {
rt().block_on(async {
let seq = log.offset(REPO).await.unwrap().map_or(0, |o| o.seq);
let runs = log.runs(REPO).await.unwrap().len();
let open = log.fold(Some(REPO), None).await.unwrap();
(seq, runs, open.repo(mimir_ingest::Kind::Nodes, REPO).map_or(0, |r| r.len()))
})
}
#[test]
fn a_reader_during_an_in_flight_consume_sees_the_last_committed_release() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let shas = history(&src);
let log = rt().block_on(open_log(&tmp.path().join("lake")));
let reader = log.clone();
let rs = spec(tmp.path(), &src);
let (entered_tx, entered) = channel();
let (go, go_rx) = channel();
// the ONE writer, on its own thread: held inside the dump of c3 (its 2nd
// dump — c2 is docs-only and never reaches the dumper)
let writer = std::thread::spawn(move || {
rt().block_on(async {
let gated = Gated { inner: FakeDumper::default(), hold_at: 2, entered: entered_tx, go: go_rx };
let mut c = Consumer::boot(log, gated).await.unwrap();
c.run(&rs).await.unwrap()
})
});
assert_eq!(entered.recv_timeout(WAIT).unwrap(), shas[2], "the consumer is inside the dump of c3");
// c1 and c2 landed, c3 is in flight: readers see exactly the two releases
assert!(reader.has_writer());
assert_eq!(state(&reader), (2, 2, 1), "offset 2, two ledger rows, the one fn of c1");
let pinned = rt().block_on(reader.pin()).unwrap();
assert!(pinned.is_pinned());
assert_eq!(state(&pinned), (2, 2, 1));
go.send(()).unwrap();
let reports = writer.join().unwrap();
assert_eq!(reports.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![1, 2, 3, 4]);
// the handle pinned at release 2 still answers release 2 — every table of
// it, offsets and ledger and events alike — while a new look sees the end
assert_eq!(state(&pinned), (2, 2, 1), "a pinned handle is one committed state, whatever lands after");
assert_eq!(state(&reader), (4, 4, 3));
assert_eq!(state(&rt().block_on(reader.pin()).unwrap()), (4, 4, 3));
assert!(!reader.has_writer(), "the consumer is gone: the seat is free");
}
#[test]
fn many_readers_pin_whole_releases_while_the_writer_commits() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n");
let mut body = String::new();
for i in 0..24 {
body.push_str(&format!("fn f{i} () | b{i}\n"));
src.write("crates/a/src/lib.rs", &body);
src.commit(&format!("c{i}"));
}
let log = rt().block_on(open_log(&tmp.path().join("lake")));
let rs = spec(tmp.path(), &src);
let done = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let readers: Vec<_> = (0..3)
.map(|_| {
let (log, done) = (log.clone(), done.clone());
std::thread::spawn(move || {
let rt = rt();
let mut looks = 0u32;
let mut last = 0;
loop {
let finished = done.load(std::sync::atomic::Ordering::Acquire);
let pinned = rt.block_on(log.pin()).unwrap();
let (seq, runs, open) = rt.block_on(async {
let seq = pinned.offset(REPO).await.unwrap().map_or(0, |o| o.seq);
let runs = pinned.runs(REPO).await.unwrap().len();
let open = pinned.fold(Some(REPO), None).await.unwrap();
(seq, runs, open.repo(mimir_ingest::Kind::Nodes, REPO).map_or(0, |r| r.len()))
});
// release N = offset N + ledger row N + the Nth fn: all or none
assert_eq!(runs as i64, seq, "a pinned state tore between consumer_offsets and ingest_runs");
assert_eq!(open as i64, seq, "a pinned state tore between consumer_offsets and code_node_events");
assert!(seq >= last, "a later pin went back in time");
last = seq;
looks += 1;
if finished {
return (looks, last);
}
}
})
})
.collect();
let wlog = log.clone();
let reports = rt().block_on(async {
let mut c = Consumer::boot(wlog, FakeDumper::default()).await.unwrap();
c.run(&rs).await.unwrap()
});
done.store(true, std::sync::atomic::Ordering::Release);
assert_eq!(reports.len(), 24);
for r in readers {
let (looks, last) = r.join().unwrap();
assert!(looks > 0);
assert_eq!(last, 24, "the look after the writer finished sees every release");
}
}
#[tokio::test(flavor = "current_thread")]
async fn the_log_has_one_writer_seat() {
let tmp = tempfile::tempdir().unwrap();
let log = open_log(&tmp.path().join("lake")).await;
let seat = log.writer().unwrap();
assert!(log.has_writer());
let second = log.clone().writer().err().expect("a second writer while the first lives");
assert!(second.downcast_ref::<WriterSeatTaken>().is_some(), "typed: {second:#}");
// a consumer is a writer: it cannot boot on a log that has one
let boot = Consumer::boot(log.clone(), FakeDumper::default()).await.err().expect("boot needs the seat");
assert!(boot.chain().any(|c| c.downcast_ref::<WriterSeatTaken>().is_some()), "{boot:#}");
drop(seat);
assert!(!log.has_writer());
// a pinned handle is a reader
let pinned = log.pin().await.unwrap();
assert!(pinned.writer().is_err(), "a pinned handle must not write");
// the consumer holds the seat for its lifetime and gives it back
let c = Consumer::boot(log.clone(), FakeDumper::default()).await.unwrap();
assert!(log.has_writer());
assert!(log.writer().is_err());
let (back, _) = c.into_parts();
assert!(!back.has_writer());
log.writer().unwrap();
}
/// The cross-process story, honestly: a lake has ONE opener. A second open —
/// from another process, or a second handle in this one — is the typed
/// `LakeHeld`, not an IO error and not a silent copy.
#[tokio::test(flavor = "current_thread")]
async fn a_second_open_of_a_held_lake_is_the_typed_lake_held() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
let first = EventLog::open_lake(&lake).await.unwrap();
let err = EventLog::open_lake(&lake).await.err().expect("the lake is held");
let held = err.downcast_ref::<LakeHeld>().unwrap_or_else(|| panic!("not typed: {err:#}"));
assert_eq!(held.lake, lake);
// readers share the holder's handle instead
assert_eq!(first.clone().offsets().await.unwrap().len(), 0);
drop(first);
EventLog::open_lake(&lake).await.expect("free again once every handle is gone");
}