crates/mimir-ingest/tests/ingest/bounded.rs
//! The consumer as a server drives it (wiring map I1, I2, I4): bounded and
//! observed runs, cancellation between commits, N repos behind one consumer,
//! the first offset of a new repo (tip, decision D2) and the explicit replay.
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use mimir_ingest::{Consumer, DumperPool, Kind, RepoSpec, Run, Stopped, git};
use super::common::*;
fn spec(tmp: &Path, name: &str, src: &Scratch, initial_offset: Option<String>) -> RepoSpec {
RepoSpec {
name: name.into(),
source: src.root.to_string_lossy().into_owned(),
branch: "main".into(),
worktree: tmp.join(format!("clone-{name}")),
initial_offset,
}
}
/// `n` commits, each adding one fn to crate `a`; commit 3 (1-based) is docs-only.
fn grow(src: &Scratch, n: usize) -> Vec<String> {
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n");
let mut body = String::new();
let mut shas = Vec::new();
for i in 1..=n {
if i == 3 {
src.write("README.md", "docs\n");
} else {
body.push_str(&format!("fn f{i} () | b{i}\n"));
src.write("crates/a/src/lib.rs", &body);
}
shas.push(src.commit(&format!("c{i}")));
}
shas
}
fn open_fns(c: &Consumer<impl mimir_ingest::Dumper>, repo: &str) -> usize {
c.open_set().repo(Kind::Nodes, repo).map_or(0, |r| r.len())
}
#[tokio::test(flavor = "current_thread")]
async fn a_run_is_bounded_observed_and_resumes_where_it_stopped() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let shas = grow(&src, 6);
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), "r", &src, None);
nornir_git::gitio::clone_or_fetch(&rs.source, &rs.worktree, None).unwrap();
// ---- a budget of two: two releases, four still behind, per-commit progress
let (mut begun, mut landed) = (Vec::new(), Vec::new());
let out = c
.consume_run(
&rs,
Run {
max_commits: Some(2),
on_begin: Some(&mut |seq, sha: &str, behind| begun.push((seq, sha.to_string(), behind))),
on_commit: Some(&mut |r: &mimir_ingest::CommitReport| landed.push((r.seq, r.status.clone()))),
..Run::default()
},
)
.await
.unwrap();
assert_eq!((out.reports.len(), out.behind, out.stopped), (2, 4, Stopped::Bound));
assert_eq!(begun, vec![(1, shas[0].clone(), 5), (2, shas[1].clone(), 4)], "begin: seq, sha, commits behind it");
assert_eq!(landed, vec![(1, "ok".to_string()), (2, "ok".to_string())]);
assert_eq!(c.log().offset("r").await.unwrap().unwrap().sha, shas[1]);
// ---- cancelled before it starts: nothing consumed, nothing lost
let cancel = AtomicBool::new(true);
let out = c.consume_run(&rs, Run { cancel: Some(&cancel), ..Run::default() }).await.unwrap();
assert_eq!((out.reports.len(), out.behind, out.stopped), (0, 4, Stopped::Cancelled));
assert_eq!(c.log().offset("r").await.unwrap().unwrap().seq, 2);
// ---- cancel raised by the first commit's callback: that commit lands
// whole (a release is atomic), the next one is not begun
cancel.store(false, Ordering::Release);
let out = c
.consume_run(&rs, Run { cancel: Some(&cancel), on_commit: Some(&mut |_| cancel.store(true, Ordering::Release)), ..Run::default() })
.await
.unwrap();
assert_eq!((out.reports.len(), out.behind, out.stopped), (1, 3, Stopped::Cancelled));
assert_eq!(out.reports[0].status, "skipped_docs_only", "c3 is the docs-only commit");
// ---- the rest: resumes at the offset, catches up
let out = c.consume_run(&rs, Run::default()).await.unwrap();
assert_eq!((out.reports.iter().map(|r| r.seq).collect::<Vec<_>>(), out.behind, out.stopped), (vec![4, 5, 6], 0, Stopped::CaughtUp));
assert_eq!(open_fns(&c, "r"), 5, "f1 f2 f4 f5 f6");
let again = c.consume_run(&rs, Run { max_commits: Some(1), ..Run::default() }).await.unwrap();
assert_eq!((again.reports.len(), again.behind, again.stopped), (0, 0, Stopped::CaughtUp));
}
#[tokio::test(flavor = "current_thread")]
async fn one_consumer_serves_n_repos_with_a_dumper_each() {
let tmp = tempfile::tempdir().unwrap();
let (sa, sb) = (Scratch::new(tmp.path().join("src-a")), Scratch::new(tmp.path().join("src-b")));
grow(&sa, 2);
grow(&sb, 4);
let (ra, rb) = (spec(tmp.path(), "alpha", &sa, None), spec(tmp.path(), "beta", &sb, None));
let log = open_log(&tmp.path().join("lake")).await;
let mut pool = DumperPool::new();
pool.insert(ra.worktree.clone(), FakeDumper::default()).unwrap();
let mut c = Consumer::boot(log, pool).await.unwrap();
// a repo registered after boot gets its dumper then
c.dumper_mut().insert(rb.worktree.clone(), FakeDumper::default()).unwrap();
assert_eq!(c.dumper_mut().len(), 2);
// interleaved, one commit at a time: each repo keeps its own ordinals
for round in 1..=2 {
for rs in [&ra, &rb] {
if round == 1 {
nornir_git::gitio::clone_or_fetch(&rs.source, &rs.worktree, None).unwrap();
}
let out = c.consume_run(rs, Run { max_commits: Some(1), ..Run::default() }).await.unwrap();
assert_eq!(out.reports.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![round]);
}
}
let offsets = c.log().offsets().await.unwrap();
assert_eq!((offsets["alpha"].seq, offsets["beta"].seq), (2, 2));
assert_eq!(c.dumper_mut().get_mut(&ra.worktree).unwrap().calls.len(), 2, "alpha's dumper saw alpha's commits only");
assert_eq!(c.dumper_mut().get_mut(&rb.worktree).unwrap().calls.len(), 2);
assert_eq!((open_fns(&c, "alpha"), open_fns(&c, "beta")), (2, 2));
// a repo with no dumper in the pool is an infrastructure error, by name
let sc = Scratch::new(tmp.path().join("src-c"));
grow(&sc, 1);
let rc = spec(tmp.path(), "gamma", &sc, None);
nornir_git::gitio::clone_or_fetch(&rc.source, &rc.worktree, None).unwrap();
let err = c.consume_run(&rc, Run::default()).await.err().expect("no dumper for gamma");
assert!(format!("{err:#}").contains("no dumper registered"), "{err:#}");
assert!(c.log().offset("gamma").await.unwrap().is_none(), "the offset did not move");
}
/// D2: a newly registered repo starts at the TIP — one release holding the
/// whole tree as of the tip, also when the tip commit itself only touched docs.
#[tokio::test(flavor = "current_thread")]
async fn a_new_repo_starts_at_the_tip_and_an_explicit_replay_backfills() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let mut shas = grow(&src, 5);
src.write("NOTES.md", "more docs\n");
shas.push(src.commit("c6 docs")); // the tip is docs-only
let probe = spec(tmp.path(), "r", &src, None);
nornir_git::gitio::clone_or_fetch(&probe.source, &probe.worktree, None).unwrap();
let tip = git::tip(&probe.worktree, "main").unwrap();
assert_eq!(tip, shas[5]);
let rs = RepoSpec { initial_offset: git::offset_before(&probe.worktree, &tip).unwrap(), ..probe };
assert_eq!(rs.initial_offset.as_deref(), Some(shas[4].as_str()));
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
let out = c.consume_run(&rs, Run::default()).await.unwrap();
assert_eq!(out.reports.len(), 1, "ONE release: the tip, not the history");
assert_eq!((out.reports[0].seq, out.reports[0].sha.as_str(), out.reports[0].status.as_str()), (1, tip.as_str(), "ok"));
assert_eq!(open_fns(&c, "r"), 4, "the whole tree at the tip (f1 f2 f4 f5) — the docs-only tip is DUMPED, there is no earlier state");
// ---- explicit backfill: --from <sha> replays from that commit (inclusive)
let from = git::resolve_from(&rs.worktree, "main", &shas[1][..10]).unwrap();
assert_eq!(from, shas[1]);
let out = c.consume_run(&rs, Run { replay_from: Some(from), ..Run::default() }).await.unwrap();
let got: Vec<(i64, &str, &str)> = out.reports.iter().map(|r| (r.seq, r.sha.as_str(), r.status.as_str())).collect();
assert_eq!(
got,
vec![
(2, shas[1].as_str(), "ok"), // back to c2's tree: f4 f5 close
(3, shas[2].as_str(), "skipped_docs_only"),
(4, shas[3].as_str(), "ok"),
(5, shas[4].as_str(), "ok"),
(6, shas[5].as_str(), "skipped_docs_only"),
],
"ordinals keep counting; the history is in the log now"
);
assert_eq!((out.reports[0].rows_plus, out.reports[0].rows_minus), (0, 2));
assert_eq!(open_fns(&c, "r"), 4, "the same tree at the tip after walking forward again");
let state_at = |n| c.log().fold(Some("r"), Some(n));
assert_eq!(state_at(2).await.unwrap().repo(Kind::Nodes, "r").unwrap().len(), 2, "seq 2 = c2: f1 f2");
// ---- --from <date>: the oldest first-parent commit at or after it
assert_eq!(git::resolve_from(&rs.worktree, "main", "1970-01-02").unwrap(), shas[0]);
let far = git::resolve_from(&rs.worktree, "main", "2999-01-01").err().unwrap();
assert!(format!("{far:#}").contains("at or after"), "{far:#}");
let junk = git::resolve_from(&rs.worktree, "main", "not-a-ref").err().unwrap();
assert!(format!("{junk:#}").contains("neither a commit"), "{junk:#}");
let absent = git::resolve_from(&rs.worktree, "main", "0123456789abcdef").err().unwrap();
assert!(format!("{absent:#}").contains("not on the first-parent chain"), "{absent:#}");
}