crates/mimir-ingest/tests/ingest/consumer.rs
//! The consumer end to end over a scratch repo + the fake dumper: exact
//! events and ledger rows, crash safety, idempotent replay, open-set rebuild.
use std::path::Path;
use super::common::*;
use mimir_ingest::{Consumer, Kind, Offset, RepoSpec};
use skade::arrow_schema::{DataType, Field, Schema};
const REPO: &str = "fixture";
/// c1 init · c2 docs-only · c3 body edit · c4 signature change (+ call graph
/// change) · c5 crate `b` removed. Returns the five shas.
fn build_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 (x: i32) | one\nfn g () | two\ncall g -> f\n")
.write("crates/b/src/lib.rs", "fn h () | three\n");
shas.push(src.commit("c1 init"));
src.write("README.md", "# docs only\n");
shas.push(src.commit("c2 docs"));
src.write("crates/a/src/lib.rs", "fn f (x: i32) | uno\nfn g () | two\ncall g -> f\n");
shas.push(src.commit("c3 body edit"));
src.write("crates/a/src/lib.rs", "fn f (x: i64) | uno\nfn g () | two\ncall f -> g\n");
shas.push(src.commit("c4 signature change"));
src.remove("crates/b");
shas.push(src.commit("c5 crate b removed"));
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,
}
}
fn t(repo: &str, seq: i64, op: &str, krate: &str, key: &str) -> (String, i64, String, String, String) {
(repo.into(), seq, op.into(), krate.into(), key.into())
}
#[tokio::test(flavor = "current_thread")]
async fn scenario_writes_exact_events_and_ledger() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let shas = build_history(&src);
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), &src);
let report = c.run(&rs).await.unwrap();
let got: Vec<(i64, &str, i64, i64, i64)> = report
.iter()
.map(|r| (r.seq, r.status.as_str(), r.crates_dumped, r.rows_plus, r.rows_minus))
.collect();
assert_eq!(
got,
vec![
(1, "ok", 3, 5, 0), // a.f a.g b.h + g→f + dep.helper (universe)
(2, "skipped_docs_only", 0, 0, 0), // README only: no checkout, no dump
(3, "ok", 3, 1, 0), // body of f
(4, "ok", 3, 2, 1), // f's signature; g→f gone, f→g new
(5, "ok", 2, 0, 1), // b left the workspace: b.h closes
]
);
assert_eq!(report.iter().map(|r| r.sha.clone()).collect::<Vec<_>>(), shas);
// The dumper never ran for the docs-only commit.
let (log, dumper) = c.into_parts();
assert_eq!(dumper.calls, vec![shas[0].clone(), shas[2].clone(), shas[3].clone(), shas[4].clone()]);
let nodes = events(&log, Kind::Nodes).await;
assert_eq!(
shape(&nodes),
vec![
t("@universe", 1, "+", DEP, "dep@1.0.0 helper"),
t(REPO, 1, "+", "a:lib", "a f"),
t(REPO, 1, "+", "a:lib", "a g"),
t(REPO, 1, "+", "b:lib", "b h"),
t(REPO, 3, "+", "a:lib", "a f"),
t(REPO, 4, "+", "a:lib", "a f"),
t(REPO, 5, "-", "b:lib", "b h"),
]
);
// Every `+` of `a f` carries a different payload hash; `-` carries none.
let f_hashes: Vec<_> = nodes.iter().filter(|e| e.5 == "a f").map(|e| e.6.clone().unwrap()).collect();
assert_eq!(f_hashes.len(), 3);
assert!(f_hashes[0] != f_hashes[1] && f_hashes[1] != f_hashes[2]);
assert!(nodes.iter().filter(|e| e.3 == "-").all(|e| e.6.is_none()));
// Each event row names the commit it came from.
assert!(nodes.iter().filter(|e| e.0 == REPO).all(|e| e.2 == shas[(e.1 - 1) as usize]));
let edges = events(&log, Kind::Edges).await;
assert_eq!(
shape(&edges),
vec![
t(REPO, 1, "+", "a:lib", "a g | a f | CALLS"),
t(REPO, 4, "+", "a:lib", "a f | a g | CALLS"),
t(REPO, 4, "-", "a:lib", "a g | a f | CALLS"),
]
);
// Ledger and offset, as stored.
let runs = log.runs(REPO).await.unwrap();
assert_eq!(
runs.iter().map(|r| (r.seq, r.sha.as_str(), r.status.as_str(), r.rows_plus, r.rows_minus, r.error.is_none())).collect::<Vec<_>>(),
vec![
(1, shas[0].as_str(), "ok", 5, 0, true),
(2, shas[1].as_str(), "skipped_docs_only", 0, 0, true),
(3, shas[2].as_str(), "ok", 1, 0, true),
(4, shas[3].as_str(), "ok", 2, 1, true),
(5, shas[4].as_str(), "ok", 0, 1, true),
]
);
assert_eq!(log.offset(REPO).await.unwrap(), Some(Offset { branch: "main".into(), sha: shas[4].clone(), seq: 5 }));
assert!(log.universe_has(DEP, TOOLCHAIN));
assert!(!log.universe_has(DEP, "another-toolchain"));
// valid_from / valid_to, derived: b.h is open up to seq 4, closed at 5;
// f's payload at seq 3 is not its payload at seq 4.
let at3 = log.fold(Some(REPO), Some(3)).await.unwrap();
let at4 = log.fold(Some(REPO), Some(4)).await.unwrap();
let at5 = log.fold(Some(REPO), Some(5)).await.unwrap();
assert!(at4.repo(Kind::Nodes, REPO).unwrap().get("b:lib", "b h").is_some());
assert!(at5.repo(Kind::Nodes, REPO).unwrap().get("b:lib", "b h").is_none());
assert_ne!(
at3.repo(Kind::Nodes, REPO).unwrap().get("a:lib", "a f").unwrap(),
at4.repo(Kind::Nodes, REPO).unwrap().get("a:lib", "a f").unwrap()
);
let keys = |s: &mimir_ingest::OpenSet, k| {
let mut v: Vec<String> = s.repo(k, REPO).map(|o| o.iter().map(|(_, k, _)| k.replace('\u{1f}', " | ")).collect()).unwrap_or_default();
v.sort();
v
};
assert_eq!(keys(&at5, Kind::Nodes), vec!["a f", "a g"]);
assert_eq!(keys(&at5, Kind::Edges), vec!["a f | a g | CALLS"]);
assert_eq!(keys(&at3, Kind::Edges), vec!["a g | a f | CALLS"]);
}
#[tokio::test(flavor = "current_thread")]
async fn dumper_error_commits_nothing_and_resume_matches_a_clean_run() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let shas = build_history(&src);
// Reference: a clean run in its own lake.
let clean = {
let log = open_log(&tmp.path().join("lake-clean")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
let mut rs = spec(tmp.path(), &src);
rs.worktree = tmp.path().join("clone-clean");
c.run(&rs).await.unwrap();
let (log, _) = c.into_parts();
(shape(&events(&log, Kind::Nodes).await), shape(&events(&log, Kind::Edges).await))
};
let log = open_log(&tmp.path().join("lake")).await;
let dumper = FakeDumper { fail_at: Some(shas[2].clone()), ..Default::default() };
let mut c = Consumer::boot(log, dumper).await.unwrap();
let rs = spec(tmp.path(), &src);
let err = c.run(&rs).await.unwrap_err();
assert!(format!("{err:#}").contains("rustc died"), "{err:#}");
// Offset stays on the last good commit (the docs-only c2), no event of seq 3.
let log = c.log();
assert_eq!(log.offset(REPO).await.unwrap().map(|o| (o.seq, o.sha)), Some((2, shas[1].clone())));
let nodes = events(log, Kind::Nodes).await;
assert!(nodes.iter().all(|e| e.1 < 3), "{nodes:?}");
let runs = log.runs(REPO).await.unwrap();
assert_eq!(runs.len(), 3);
assert_eq!((runs[2].seq, runs[2].status.as_str()), (3, "error"));
assert!(runs[2].error.as_deref().unwrap().contains("rustc died"));
// Fix the dumper, run again: resumes at c3, ends where a clean run ends.
c.dumper_mut().fail_at = None;
let report = c.run(&rs).await.unwrap();
assert_eq!(report.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![3, 4, 5]);
let (log, _) = c.into_parts();
assert_eq!((shape(&events(&log, Kind::Nodes).await), shape(&events(&log, Kind::Edges).await)), clean);
let statuses: Vec<(i64, String)> = log.runs(REPO).await.unwrap().into_iter().map(|r| (r.seq, r.status)).collect();
assert_eq!(
statuses,
vec![(1, "ok".into()), (2, "skipped_docs_only".into()), (3, "error".into()), (3, "ok".into()), (4, "ok".into()), (5, "ok".into())]
);
}
#[tokio::test(flavor = "current_thread")]
async fn failure_inside_the_atomic_release_moves_no_table() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
build_history(&src);
let lake = tmp.path().join("lake");
let log = open_log(&lake).await;
// Sabotage the LAST table of the batch: universe_manifest gets a schema the
// manifest row cannot be cast into. Events, offset and ledger tables are
// encoded and their files written BEFORE the manifest fails.
{
use skade::iceberg::Catalog;
let wh = log.warehouse();
let ident = wh.table_ident("mimir_v2.universe_manifest").unwrap();
wh.catalog().drop_table(&ident).await.unwrap();
let bad = Schema::new(vec![Field::new("only", DataType::Int64, false)]);
wh.create_table("mimir_v2.universe_manifest", &bad).await.unwrap();
}
let before = log.stamp().await.unwrap();
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), &src);
let err = c.run(&rs).await.unwrap_err();
assert!(format!("{err:#}").contains("atomic release"), "{err:#}");
let log = c.log();
assert_eq!(log.stamp().await.unwrap(), before, "an event table advanced");
assert!(events(log, Kind::Nodes).await.is_empty());
assert!(events(log, Kind::Edges).await.is_empty());
assert_eq!(log.offset(REPO).await.unwrap(), None);
let runs = log.runs(REPO).await.unwrap();
assert_eq!(runs.iter().map(|r| (r.seq, r.status.as_str())).collect::<Vec<_>>(), vec![(1, "error")]);
// The open set did not move either.
assert!(c.open_set().repo(Kind::Nodes, REPO).is_none_or(|o| o.is_empty()));
}
#[tokio::test(flavor = "current_thread")]
async fn replay_is_idempotent() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
build_history(&src);
let lake = tmp.path().join("lake");
let rs = spec(tmp.path(), &src);
let log = open_log(&lake).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
assert_eq!(c.run(&rs).await.unwrap().len(), 5);
let after_first = (events(c.log(), Kind::Nodes).await, events(c.log(), Kind::Edges).await);
let stamp = c.log().stamp().await.unwrap();
// Same consumer, again: nothing to consume, nothing written.
assert!(c.run(&rs).await.unwrap().is_empty());
// A fresh boot (open set folded from the log), again: same.
let (log, _) = c.into_parts();
drop(log);
let log = open_log(&lake).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
assert!(c.run(&rs).await.unwrap().is_empty());
assert_eq!(c.log().stamp().await.unwrap(), stamp);
// A code commit whose dump is unchanged (a trailing comment: nothing
// moves): the dumper runs, the delta is empty, only offset + ledger move.
src.write("crates/a/src/lib.rs", "fn f (x: i64) | uno\nfn g () | two\ncall f -> g\n// a comment rustc ignores\n");
let sha = src.commit("c6 comment");
let report = c.run(&rs).await.unwrap();
assert_eq!(report.len(), 1);
assert_eq!((report[0].seq, report[0].status.as_str(), report[0].rows_plus, report[0].rows_minus), (6, "ok", 0, 0));
assert_eq!(c.dumper_mut().calls, vec![sha]);
assert_eq!((events(c.log(), Kind::Nodes).await, events(c.log(), Kind::Edges).await), after_first);
assert_eq!(c.log().stamp().await.unwrap(), stamp);
}
/// A boot derives the same open set whichever zone exists: from the events
/// alone (no anchor), and from the READ-FAST zone once the repo is
/// materialized — `open-set.bin` is retired, the lake holds the anchor.
#[tokio::test(flavor = "current_thread")]
async fn a_boot_from_the_zone_is_the_same_open_set() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
build_history(&src);
let lake = tmp.path().join("lake");
let rs = spec(tmp.path(), &src);
let mut c = Consumer::boot(open_log(&lake).await, FakeDumper::default()).await.unwrap();
c.run(&rs).await.unwrap();
let live = c.open_set().clone();
assert_eq!(live.repo(Kind::Nodes, REPO).unwrap().len(), 2);
drop(c);
// no zone: folded from the events — identical
let c = Consumer::boot(open_log(&lake).await, FakeDumper::default()).await.unwrap();
assert_eq!(c.open_set(), &live);
drop(c);
// the zone: materialized, the boot reads interval rows + tail — identical
let log = open_log(&lake).await;
let r = log.materialize(REPO).await.unwrap();
assert!(!r.skipped && r.rows > 0, "{r:?}");
let c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
assert_eq!(c.open_set(), &live);
drop(c);
// the log moves past the anchor: the tail completes the boot — identical
src.write("crates/a/src/lib.rs", "fn f (x: i64) | dos\nfn g () | two\ncall f -> g\n");
src.commit("c6 body");
let mut c = Consumer::boot(open_log(&lake).await, FakeDumper::default()).await.unwrap();
c.run(&rs).await.unwrap();
let moved = c.open_set().clone();
assert_ne!(moved, live);
drop(c);
let c = Consumer::boot(open_log(&lake).await, FakeDumper::default()).await.unwrap();
assert_eq!(c.open_set(), &moved);
}
/// Six repos in one lake: a boot after materializing SOME of them mixes the
/// zone (anchored repos) and the event fold (the rest) into the same open
/// set a full event fold derives.
#[tokio::test(flavor = "current_thread")]
async fn a_boot_mixes_zone_and_event_repos() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
let mut c = Consumer::boot(open_log(&lake).await, FakeDumper::default()).await.unwrap();
for i in 0..6 {
let src = Scratch::new(tmp.path().join(format!("src{i}")));
let body: String = (0..30).map(|f| format!("fn f{f} () | body {f}\n")).collect();
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n").write(&format!("crates/m{i}/src/lib.rs"), &body);
src.commit("init");
let rs = RepoSpec {
name: format!("m{i}"),
source: src.root.to_string_lossy().into_owned(),
branch: "main".into(),
worktree: tmp.path().join(format!("clone{i}")),
initial_offset: None,
};
c.run(&rs).await.unwrap();
}
assert_eq!(c.open_set().repo(Kind::Nodes, "m5").unwrap().len(), 30);
drop(c);
// the yardstick: a boot before any zone exists (the event fold)
let c = Consumer::boot(open_log(&lake).await, FakeDumper::default()).await.unwrap();
let rebuilt = c.open_set().clone();
drop(c);
let log = open_log(&lake).await;
for repo in ["m0", "m2", "m4"] {
assert!(!log.materialize(repo).await.unwrap().skipped);
}
let c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
assert_eq!(c.open_set(), &rebuilt);
}
/// Dumps a fixed `mimir-schema` batch per commit, in commit order.
type Script = (Vec<mimir_schema::NodeRow>, Vec<mimir_schema::EdgeRow>, Vec<mimir_schema::SpanRow>);
struct Scripted(Vec<Script>);
impl mimir_ingest::Dumper for Scripted {
fn dump(&mut self, _: &Path, _: &str) -> anyhow::Result<mimir_ingest::CommitDump> {
let (n, e, s) = self.0.remove(0);
Ok(mimir_ingest::CommitDump {
toolchain: TOOLCHAIN.into(),
workspace_crates: ["k@0.1.0:lib".to_string()].into(),
workspace_targets: ["k:lib".to_string()].into(),
failed_units: Default::default(),
owner_disambiguations: 0,
crates: vec![mimir_ingest::CrateDump {
krate: "k@0.1.0:lib".into(),
universe: false,
nodes: vec![mimir_schema::nodes_batch(&n).unwrap()],
edges: vec![mimir_schema::edges_batch(&e).unwrap()],
spans: vec![mimir_schema::spans_batch(&s).unwrap()],
}],
binary: false,
binaries: vec![],
})
}
}
#[tokio::test(flavor = "current_thread")]
async fn real_mimir_schema_rows_round_trip_through_the_log() {
use mimir_schema::{EdgeKind, EdgeRow, Level, NodeRow, Origin, Provenance, SPAN_EDGE, SPAN_NODE, SpanRow, edge_key};
let node = |id: &str, ty: Option<&str>, start: u32, end: u32| NodeRow {
id: id.into(),
kind: "fn".into(),
level: Level::Item,
name: id.into(),
parent: None,
repo: Some(REPO.into()),
krate: "k".into(),
version: "0.1.0".into(),
file: Some("src/lib.rs".into()),
start_line: Some(start),
end_line: Some(end),
cfg: None,
origin: Origin::Local,
provenance: Provenance::SourceOnly,
valid_from_sha: None,
valid_to_sha: None,
ty: ty.map(str::to_string),
is_test: false,
defined_here: true,
visibility: None,
effective_visibility: None,
doc: None,
has_doc: None,
ignored: None,
doc_hidden: None,
link_name: None,
};
let call = |count: u32| EdgeRow {
from: "f".into(),
to: "g".into(),
kind: EdgeKind::Calls,
count,
provenance: Provenance::SourceOnly,
valid_from_sha: None,
valid_to_sha: None,
negated: None,
partial: None,
access: None,
};
let span = |target: &'static str, key: &str, start: u32, end: u32| SpanRow {
target,
key: key.into(),
file: "src/lib.rs".into(),
start_line: start,
end_line: end,
};
let site = edge_key("f", "g", EdgeKind::Calls);
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
src.write("src/lib.rs", "1\n");
src.commit("one");
src.write("src/lib.rs", "2\n");
src.commit("two");
let wh = skade::open(tmp.path().join("lake")).await.unwrap();
let log = mimir_ingest::EventLog::open(wh, mimir_ingest::Specs::mimir().unwrap()).await.unwrap();
// 1: f, g, and `dup` — a producer collision (one id, two semantic
// payloads); f calls g at two sites (count 2, two edge spans).
// 2: g MOVES down a line and grows (positions only), the call sites move
// with it, the collision resolves to one payload.
let script = Scripted(vec![
(
vec![node("f", None, 1, 2), node("g", None, 3, 5), node("dup", Some("A"), 9, 9), node("dup", Some("B"), 9, 9)],
vec![call(2)],
vec![span(SPAN_NODE, "f", 1, 2), span(SPAN_NODE, "g", 3, 5), span(SPAN_EDGE, &site, 3, 3), span(SPAN_EDGE, &site, 4, 4)],
),
(
vec![node("f", None, 1, 2), node("g", None, 4, 6), node("dup", Some("A"), 9, 9)],
vec![call(2)],
vec![span(SPAN_NODE, "f", 1, 2), span(SPAN_NODE, "g", 4, 6), span(SPAN_EDGE, &site, 4, 4), span(SPAN_EDGE, &site, 5, 5)],
),
]);
let mut c = Consumer::boot(log, script).await.unwrap();
let report = c.run(&spec(tmp.path(), &src)).await.unwrap();
assert_eq!(
report.iter().map(|r| (r.rows_plus, r.rows_minus, r.key_collisions, r.span_plus, r.span_minus)).collect::<Vec<_>>(),
vec![
// f g dup#A dup#B + one semantic edge; 4 spans
(5, 0, 1, 4, 0),
// g moved: NO node or edge event for it; dup: + plain, − both
// disambiguated; spans: g's span and the site at line 3 → 4 / 5
(1, 2, 0, 2, 2),
]
);
let live = c.open_set().clone();
let (log, _) = c.into_parts();
assert_eq!(log.fold(None, None).await.unwrap(), live, "the fold must rebuild every identity, incl. disambiguated and span ones");
let dup_events: Vec<_> = events(&log, Kind::Nodes).await.into_iter().filter(|e| e.5 == "dup").map(|e| (e.1, e.3)).collect();
assert_eq!(dup_events, vec![(1, "+".into()), (1, "+".into()), (2, "+".into()), (2, "-".into()), (2, "-".into())]);
assert!(events(&log, Kind::Nodes).await.iter().all(|e| e.5 != "g" || e.1 == 1), "a moved item writes no node event");
assert_eq!(shape(&events(&log, Kind::Edges).await), vec![t(REPO, 1, "+", "k:lib", "f | g | CALLS")]);
// Positions are not in the node table; edges carry count, no line; the
// edge table declares its sort order on `from`.
let node_t = log.warehouse().table("mimir_v2.code_node_events").await.unwrap();
let stored = skade::arrow_schema_of(node_t.inner()).unwrap();
assert!(stored.field_with_name("valid_from_sha").is_err() && stored.field_with_name("id").is_ok());
assert!(stored.field_with_name("start_line").is_err() && stored.field_with_name("file").is_err());
let edge_t = log.warehouse().table("mimir_v2.code_edge_events").await.unwrap();
let stored = skade::arrow_schema_of(edge_t.inner()).unwrap();
assert_eq!(stored.field_with_name("count").unwrap().data_type(), &DataType::Int64);
assert!(stored.field_with_name("line").is_err());
assert_eq!(mimir_ingest::log::sorted_by(edge_t.inner()), vec!["from"]);
assert!(mimir_ingest::log::sorted_by(node_t.inner()).is_empty());
}
#[tokio::test(flavor = "current_thread")]
async fn a_new_consumer_can_start_after_an_initial_offset() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let shas = build_history(&src);
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
let mut rs = spec(tmp.path(), &src);
rs.initial_offset = Some(shas[2].clone());
let report = c.run(&rs).await.unwrap();
// c4 is the first event: seq 1, cold (everything it dumps is new).
assert_eq!(
report.iter().map(|r| (r.seq, r.sha.clone(), r.rows_plus, r.rows_minus)).collect::<Vec<_>>(),
vec![(1, shas[3].clone(), 5, 0), (2, shas[4].clone(), 0, 1)]
);
// Once an offset is stored, the initial offset no longer matters.
rs.initial_offset = Some(shas[0].clone());
assert!(c.run(&rs).await.unwrap().is_empty());
}
#[tokio::test(flavor = "current_thread")]
async fn moving_code_writes_span_events_only() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
build_history(&src);
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), &src);
c.run(&rs).await.unwrap();
let before = (events(c.log(), Kind::Nodes).await, events(c.log(), Kind::Edges).await);
// Two lines pushed in above everything in `a`: every item and site moves.
src.write("crates/a/src/lib.rs", "// moved\n// down\nfn f (x: i64) | uno\nfn g () | two\ncall f -> g\n");
src.commit("c6 move");
let r = c.run(&rs).await.unwrap();
assert_eq!(
r.iter().map(|r| (r.status.as_str(), r.rows_plus, r.rows_minus, r.span_plus, r.span_minus)).collect::<Vec<_>>(),
// f, g and the f→g site: each a − at the old line, a + at the new one
vec![("ok", 0, 0, 3, 3)]
);
assert_eq!((events(c.log(), Kind::Nodes).await, events(c.log(), Kind::Edges).await), before);
let spans = shape(&events(c.log(), Kind::Spans).await);
let at6: Vec<_> = spans.iter().filter(|e| e.1 == 6).map(|e| (e.2.clone(), e.4.replace('\u{1e}', ">"))).collect();
assert_eq!(
at6,
vec![
("+".into(), "edge | a f>a g>CALLS | crates/a/src/lib.rs | 5 | 5".into()),
("+".into(), "node | a f | crates/a/src/lib.rs | 3 | 3".into()),
("+".into(), "node | a g | crates/a/src/lib.rs | 4 | 4".into()),
("-".into(), "edge | a f>a g>CALLS | crates/a/src/lib.rs | 3 | 3".into()),
("-".into(), "node | a f | crates/a/src/lib.rs | 1 | 1".into()),
("-".into(), "node | a g | crates/a/src/lib.rs | 2 | 2".into()),
]
);
let live = c.open_set().clone();
assert_eq!(c.log().fold(None, None).await.unwrap(), live);
}
#[tokio::test(flavor = "current_thread")]
async fn a_crate_that_does_not_compile_never_blocks_the_repo() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
src.write("Cargo.toml", "[workspace]\n")
.write("crates/a/src/lib.rs", "fn f () | one\n")
.write("crates/b/src/lib.rs", "fn h () | three\n");
src.commit("c1");
// b breaks while a changes: a's change lands, b keeps its state.
src.write("crates/a/src/lib.rs", "fn f () | two\n").write("crates/b/src/lib.rs", "fn h () | four\nBROKEN\n");
src.commit("c2 b broken");
// b is fixed: its delta (three → five) appears now.
src.write("crates/b/src/lib.rs", "fn h () | five\n");
src.commit("c3 b fixed");
// Everything broken: nothing compiled, no events, the offset still moves.
src.write("crates/a/src/lib.rs", "BROKEN\n").write("crates/b/src/lib.rs", "BROKEN\n");
let c4 = src.commit("c4 all broken");
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
let report = c.run(&spec(tmp.path(), &src)).await.unwrap();
assert_eq!(
report.iter().map(|r| (r.seq, r.status.as_str(), r.rows_plus, r.rows_minus, r.failed_units.clone())).collect::<Vec<_>>(),
vec![
(1, "ok", 3, 0, vec![]),
(2, "partial", 1, 0, vec!["b:lib".to_string()]),
(3, "ok", 1, 0, vec![]),
(4, "error", 0, 0, vec!["a:lib".to_string(), "b:lib".to_string()]),
]
);
let nodes = shape(&events(c.log(), Kind::Nodes).await);
let local: Vec<_> = nodes.iter().filter(|e| e.0 == REPO).map(|e| (e.1, e.2.as_str(), e.4.as_str())).collect();
assert_eq!(local, vec![(1, "+", "a f"), (1, "+", "b h"), (2, "+", "a f"), (3, "+", "b h")]);
// b's state through the broken commit is its c1 state; at c3 it moved on.
let h = |s: &mimir_ingest::OpenSet| *s.repo(Kind::Nodes, REPO).unwrap().get("b:lib", "b h").unwrap();
let (at1, at2, at3) = (
c.log().fold(Some(REPO), Some(1)).await.unwrap(),
c.log().fold(Some(REPO), Some(2)).await.unwrap(),
c.log().fold(Some(REPO), Some(3)).await.unwrap(),
);
assert_eq!(h(&at1), h(&at2));
assert_ne!(h(&at2), h(&at3));
// The all-broken commit closed nothing and is the offset.
assert_eq!(c.open_set().repo(Kind::Nodes, REPO).unwrap().len(), 2);
assert_eq!(c.log().offset(REPO).await.unwrap().map(|o| (o.seq, o.sha)), Some((4, c4)));
let runs = c.log().runs(REPO).await.unwrap();
assert!(runs[1].error.as_deref().unwrap().contains("b"), "{:?}", runs[1].error);
assert!(runs[3].error.as_deref().unwrap().starts_with("nothing compiled"), "{:?}", runs[3].error);
}
/// The commit's message and committer time (what a history rewrite keeps).
fn message_and_time(repo: &Path, sha: &str) -> (String, i64) {
let repo = gix::open(repo).unwrap();
let c = repo.find_commit(gix::ObjectId::from_hex(sha.as_bytes()).unwrap()).unwrap();
(c.message_raw_sloppy().to_string(), c.time().unwrap().seconds)
}
/// Every open identity of `repo`, all six tables (a table with none: `None`).
fn open_of(open: &mimir_ingest::OpenSet, repo: &str) -> Vec<Option<mimir_ingest::RepoOpen>> {
Kind::ALL.iter().map(|k| open.repo(*k, repo).filter(|o| !o.is_empty()).cloned()).collect()
}
/// The repo's history is REWRITTEN under the consumer (a forge move: every
/// commit again — same message, same date — with another hash; here the move
/// also scrubbed crate `b` out of history): the stored offset is on no
/// first-parent chain of the branch any more. The consumer reads the repo
/// again from its head as a fresh member — ONE release, the ordinals counting
/// on — and nothing of the old history stays open beside the new: the open
/// state is exactly a fresh member's at that head.
#[tokio::test(flavor = "current_thread")]
async fn a_rewritten_history_is_read_again_from_its_head() {
let tmp = tempfile::tempdir().unwrap();
let src_dir = tmp.path().join("src");
let src = Scratch::new(src_dir.clone());
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n# repository = codeberg\n")
.write("crates/a/src/lib.rs", "fn f (x: i32) | one\nfn g () | two\ncall g -> f\n")
.write("crates/b/src/lib.rs", "fn h () | three\nlink h\n");
let o1 = src.commit("init");
src.write("crates/a/src/lib.rs", "fn f (x: i32) | one\nfn g () | two\ncall g -> f\nfn k () | four\n");
let o2 = src.commit("more");
let dumper = || FakeDumper { binary: true, ..Default::default() };
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, dumper()).await.unwrap();
let rs = spec(tmp.path(), &src);
let first = c.run(&rs).await.unwrap();
assert_eq!(first.iter().map(|r| (r.seq, r.sha.clone())).collect::<Vec<_>>(), vec![(1, o1.clone()), (2, o2.clone())]);
assert!(c.open_set().repo(Kind::Nodes, REPO).unwrap().get("b:lib", "b h").is_some());
assert!(c.open_set().repo(Kind::Symbols, REPO).is_some_and(|o| !o.is_empty()), "b's bin is in the binary stream");
// ---- the move: the same source path now holds a rewritten history
std::fs::remove_dir_all(&src_dir).unwrap();
let src = Scratch::new(src_dir);
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n# repository = gunnar.rs\n")
.write("crates/a/src/lib.rs", "fn f (x: i32) | one\nfn g () | two\ncall g -> f\n");
src.commit("init");
src.write("crates/a/src/lib.rs", "fn f (x: i32) | one\nfn g () | two\ncall g -> f\nfn k () | four\n")
.write("crates/c/src/lib.rs", "fn m () | five\n");
let n2 = src.commit("more");
assert_ne!(n2, o2);
assert_eq!(message_and_time(&src.root, &n2), message_and_time(&rs.worktree, &o2), "same message, same date, another hash");
mimir_ingest::git::sync_clone(&rs.source, &rs.worktree).unwrap();
let mut seen = Vec::new();
let out = c
.consume_run(&rs, mimir_ingest::Run { on_rewritten: Some(&mut |r| seen.push((r.offset.clone(), r.head.clone()))), ..Default::default() })
.await
.unwrap();
assert_eq!(seen, vec![(o2.clone(), n2.clone())], "said once: the old offset, the new head");
let r = out.rewritten.as_ref().expect("the run says the history was rewritten");
assert_eq!((r.offset.as_str(), r.head.as_str(), r.branch.as_str()), (o2.as_str(), n2.as_str(), "main"));
assert_eq!(
out.reports.iter().map(|r| (r.seq, r.sha.clone(), r.status.clone())).collect::<Vec<_>>(),
vec![(3, n2.clone(), "ok".to_string())],
"ONE release at the head, the ordinals counting on"
);
assert_eq!(c.log().offset(REPO).await.unwrap(), Some(Offset { branch: "main".into(), sha: n2.clone(), seq: 3 }));
// the old history's rows are not open any more; the new head's are
let nodes = c.open_set().repo(Kind::Nodes, REPO).unwrap();
let mut keys: Vec<String> = nodes.iter().map(|(_, k, _)| k).collect();
keys.sort();
assert_eq!(keys, vec!["a f", "a g", "a k", "c m"]);
assert!(c.open_set().repo(Kind::Symbols, REPO).is_none_or(|o| o.is_empty()), "b's bin closed with b");
let closed: Vec<_> = events(c.log(), Kind::Nodes).await.into_iter().filter(|e| e.0 == REPO && e.1 == 3).map(|e| (e.3, e.5)).collect();
assert_eq!(closed, vec![("+".to_string(), "c m".to_string()), ("-".to_string(), "b h".to_string())], "the delta: c in, b out");
// what the log folds to is that state, and it is exactly a FRESH member's at the head
assert_eq!(open_of(&c.log().fold(None, None).await.unwrap(), REPO), open_of(c.open_set(), REPO));
let fresh_log = open_log(&tmp.path().join("lake-fresh")).await;
let mut f = Consumer::boot(fresh_log, dumper()).await.unwrap();
let mut fresh = RepoSpec { worktree: tmp.path().join("clone-fresh"), ..spec(tmp.path(), &src) };
mimir_ingest::git::sync_clone(&fresh.source, &fresh.worktree).unwrap();
fresh.initial_offset = mimir_ingest::git::offset_before(&fresh.worktree, &n2).unwrap();
assert_eq!(f.consume(&fresh).await.unwrap().len(), 1);
assert_eq!(open_of(c.open_set(), REPO), open_of(f.open_set(), REPO), "the same open state as a fresh member at the head");
// and it is an ordinary member again: caught up, nothing rewritten
let again = c.consume_run(&rs, mimir_ingest::Run::default()).await.unwrap();
assert!(again.reports.is_empty() && again.rewritten.is_none());
// the streamed fold agrees with the fold, over a history rewritten and read again
fold_each_agrees_with_fold_rows(c.log(), REPO).await;
}