crates/mimir-ingest/tests/ingest/binary.rs
//! The binary layer as its OWN producer stream (plan L9-dec, LAW 5): its
//! tables, the derived provenance, the binary-only re-ingest, and the
//! toolchain dump that runs once per toolchain.
use std::collections::BTreeSet;
use std::path::Path;
use super::common::*;
use mimir_ingest::{Consumer, EventLog, Kind, RepoSpec, StdDumper, StdStatus, Stream};
use mimir_schema::Provenance;
const REPO: &str = "binfix";
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,
}
}
/// c1: crate `a` with f (linked), g (not linked: inlined / dead), generic t
/// with one instance, an ffi import and a linked universe item; crate `b`
/// has no bin. c2: `a`'s bin no longer links f.
fn history(src: &Scratch) -> Vec<String> {
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n")
.write(
"crates/a/src/lib.rs",
"fn f () | one\nfn g () | two\nfn t (x: T) | gen\ncall f -> g\nlink f\nmono t u32\nimport getpid\nlinkdep helper\n",
)
.write("crates/b/src/lib.rs", "fn h () | three\n");
let c1 = src.commit("c1");
src.write("crates/a/src/lib.rs", "fn f () | one\nfn g () | two\nfn t (x: T) | gen\ncall f -> g\nmono t u32\nimport getpid\nlinkdep helper\n");
let c2 = src.commit("c2 f no longer linked");
vec![c1, c2]
}
async fn boot(dir: &Path, binary: bool) -> Consumer<FakeDumper> {
let log = open_log(&dir.join("lake")).await;
Consumer::boot(log, FakeDumper { binary, ..Default::default() }).await.unwrap()
}
/// (owner column) of every event of `kind`.
async fn owners(log: &EventLog, kind: Kind) -> BTreeSet<String> {
events(log, kind).await.into_iter().map(|e| e.4).collect()
}
#[tokio::test(flavor = "current_thread")]
async fn provenance_is_derived_by_joining_the_two_streams_on_the_moniker() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let shas = history(&src);
let mut c = boot(tmp.path(), true).await;
let rs = spec(tmp.path(), &src);
let rep = c.run(&rs).await.unwrap();
assert_eq!(rep.iter().map(|r| r.bin_rev).collect::<Vec<_>>(), vec![Some(1), Some(2)]);
let at1 = c.log().provenance(REPO, Some(1)).await.unwrap();
let p = |v: &mimir_ingest::ProvenanceView, id: &str| v.get(id);
assert_eq!(p(&at1, "a f"), Some(Provenance::Both), "linked source item");
assert_eq!(p(&at1, "a g"), Some(Provenance::SourceOnly), "no binary twin");
assert_eq!(p(&at1, "b h"), Some(Provenance::SourceOnly), "crate without a bin");
assert_eq!(p(&at1, "a t"), Some(Provenance::Both), "generic item with a linked instance");
assert_eq!(p(&at1, "a t<u32>"), Some(Provenance::BinaryOnly), "mono instance");
assert_eq!(p(&at1, "extern getpid"), Some(Provenance::BinaryOnly), "ffi import without a source decl");
assert_eq!(p(&at1, "dep@1.0.0 helper"), Some(Provenance::Both), "universe item the artifact links");
assert_eq!(p(&at1, "dep@1.0.0 other"), None);
assert_eq!(
at1.counts().into_iter().collect::<Vec<_>>(),
vec![(Provenance::SourceOnly, 2), (Provenance::BinaryOnly, 2), (Provenance::Both, 3)]
);
// c2: f is no longer in the artifact → source_only again (derived, no
// source event was written for it)
let now = c.log().provenance(REPO, None).await.unwrap();
assert_eq!(p(&now, "a f"), Some(Provenance::SourceOnly));
assert_eq!(p(&now, "a t"), Some(Provenance::Both));
let node_events = events(c.log(), Kind::Nodes).await;
assert!(node_events.iter().all(|e| e.1 == 1 || e.5 != "a f"), "c2 wrote no source event for f: {node_events:?}");
let sym = shape(&events(c.log(), Kind::Symbols).await);
assert!(sym.contains(&(REPO.into(), 2, "-".into(), artifact("a"), "_Raf | a f".into())), "{sym:?}");
let _ = shas;
}
#[test]
fn every_event_table_has_exactly_one_writer_stream() {
let source: BTreeSet<&str> = Kind::SOURCE.iter().map(|k| EventLog::table_name(*k)).collect();
let binary: BTreeSet<&str> = Kind::BINARY.iter().map(|k| EventLog::table_name(*k)).collect();
assert_eq!((source.len(), binary.len()), (3, 3), "one table per kind");
assert!(source.is_disjoint(&binary), "a table shared by both streams: {:?}", source.intersection(&binary).collect::<Vec<_>>());
for k in Kind::ALL {
let expected = if Kind::SOURCE.contains(&k) { Stream::Source } else { Stream::Binary };
assert_eq!(k.stream(), expected, "{k:?}");
}
assert_eq!(Kind::ALL.len(), Kind::SOURCE.len() + Kind::BINARY.len());
}
#[tokio::test(flavor = "current_thread")]
async fn the_two_streams_never_write_the_same_table() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
history(&src);
let mut c = boot(tmp.path(), true).await;
c.run(&spec(tmp.path(), &src)).await.unwrap();
let crates: BTreeSet<String> = ["a:lib", "b:lib", DEP].iter().map(|s| s.to_string()).collect();
let arts: BTreeSet<String> = [artifact("a")].into();
for kind in Kind::ALL {
let got = owners(c.log(), kind).await;
let (want, other) = match kind.stream() {
Stream::Source => (&crates, &arts),
Stream::Binary => (&arts, &crates),
};
assert!(got.is_disjoint(other), "{kind:?} ({}) carries the other stream's owners: {got:?}", EventLog::table_name(kind));
if kind != Kind::BinEdges || !got.is_empty() {
assert!(got.is_subset(want), "{kind:?}: {got:?}");
}
}
// the binary stream wrote no source node: no source id in bin_node_events
let bin_nodes: BTreeSet<String> = events(c.log(), Kind::BinNodes).await.into_iter().map(|e| e.5).collect();
assert_eq!(bin_nodes, ["a t<u32>", "extern getpid"].iter().map(|s| s.to_string()).collect());
// and the source stream wrote none of the binary's
let src_nodes: BTreeSet<String> = events(c.log(), Kind::Nodes).await.into_iter().map(|e| e.5).collect();
assert!(src_nodes.is_disjoint(&bin_nodes));
}
#[tokio::test(flavor = "current_thread")]
async fn a_binary_reingest_is_its_own_release_and_idempotent() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let shas = history(&src);
let mut c = boot(tmp.path(), true).await;
let rs = spec(tmp.path(), &src);
c.run(&rs).await.unwrap();
let before: Vec<_> = futures_all(c.log()).await;
let runs_before = c.log().runs(REPO).await.unwrap();
// artifact rebuilt, same sha: same rows → ledger row, NO events
let r = c.reingest_binaries(&rs).await.unwrap();
assert_eq!(r.len(), 1);
assert_eq!((r[0].rev, r[0].seq, r[0].sha.as_str(), r[0].rows_plus, r[0].rows_minus), (3, 2, shas[1].as_str(), 0, 0));
assert_eq!(futures_all(c.log()).await, before, "an identical re-ingest writes no event");
assert_eq!(c.log().runs(REPO).await.unwrap(), runs_before, "the source ledger is not touched");
assert_eq!(c.log().offset(REPO).await.unwrap().unwrap().seq, 2, "the offset does not move");
let again = c.reingest_binaries(&rs).await.unwrap();
assert_eq!((again[0].rev, again[0].rows_plus, again[0].rows_minus), (4, 0, 0));
assert_eq!(futures_all(c.log()).await, before);
assert_eq!(c.log().bin_runs(REPO).await.unwrap().iter().map(|r| r.rev).collect::<Vec<_>>(), vec![1, 2, 3, 4]);
assert_eq!(c.dumper_mut().binary_calls, 2);
// a re-ingest that DOES differ (the worktree's bin lines changed without a
// commit — a rebuilt artifact) lands at the same seq with a higher rev,
// and the fold sees it: the open set from the log == the live one
std::fs::write(
rs.worktree.join("crates/a/src/lib.rs"),
// appended: the other symbols keep their addresses
"fn f () | one\nfn g () | two\nfn t (x: T) | gen\ncall f -> g\nmono t u32\nimport getpid\nlinkdep helper\nlink f\nlink g\n",
)
.unwrap();
let r = c.reingest_binaries(&rs).await.unwrap();
assert_eq!((r[0].rev, r[0].seq, r[0].rows_plus, r[0].rows_minus), (5, 2, 2, 0));
let folded = c.log().fold(None, None).await.unwrap();
assert_eq!(&folded, c.open_set());
let p = c.log().provenance(REPO, None).await.unwrap();
assert_eq!((p.get("a f"), p.get("a g")), (Some(Provenance::Both), Some(Provenance::Both)));
}
/// Every event of every table, for equality.
async fn futures_all(log: &EventLog) -> Vec<Vec<Ev>> {
let mut v = Vec::new();
for k in Kind::ALL {
v.push(events(log, k).await);
}
v
}
#[tokio::test(flavor = "current_thread")]
async fn an_artifact_whose_bin_left_the_workspace_closes_and_a_check_run_leaves_the_binary_stream_alone() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
history(&src);
// c3: a's bin lines gone (the bin left), c4 docs only
src.write("crates/a/src/lib.rs", "fn f () | one\nfn g () | two\nfn t (x: T) | gen\ncall f -> g\n");
src.commit("c3 bin removed");
let mut c = boot(tmp.path(), true).await;
let rs = spec(tmp.path(), &src);
let rep = c.run(&rs).await.unwrap();
assert!(rep[2].bin_minus > 0);
let runs = c.log().bin_runs(REPO).await.unwrap();
assert_eq!(runs.last().map(|r| (r.rev, r.status.as_str())), Some((3, "closed")));
for k in Kind::BINARY {
assert!(c.open_set().repo(k, REPO).is_none_or(|o| o.is_empty()), "{k:?} still open");
}
assert_eq!(c.log().provenance(REPO, None).await.unwrap().get("a f"), Some(Provenance::SourceOnly));
// the same repo consumed WITHOUT the binary layer: the binary tables stay empty
let tmp2 = tempfile::tempdir().unwrap();
let mut plain = boot(tmp2.path(), false).await;
plain.run(&spec(tmp2.path(), &src)).await.unwrap();
for k in Kind::BINARY {
assert!(events(plain.log(), k).await.is_empty(), "{k:?}");
}
assert!(plain.log().bin_runs(REPO).await.unwrap().is_empty());
}
/// A fake `cargo mimir-dump --std` output: lang crates `std` and `core`.
fn fake_std(out: &Path) -> anyhow::Result<()> {
use mimir_schema::{Level, NodeRow, Origin};
for (krate, item) in [("std", "io/stdout()."), ("core", "fmt/Debug#")] {
let ver = mimir_schema::moniker::lang_version(krate);
let id = mimir_schema::moniker::moniker(krate, &ver, item);
let stem = format!("{krate}@lang-0000");
let row = NodeRow {
id,
kind: "fn".into(),
level: Level::Item,
name: item.into(),
parent: None,
repo: None,
krate: krate.into(),
version: ver,
file: None,
start_line: None,
end_line: None,
cfg: None,
origin: Origin::Lang,
provenance: Provenance::SourceOnly,
valid_from_sha: None,
valid_to_sha: None,
ty: None,
is_test: false,
defined_here: true,
visibility: None,
effective_visibility: None,
doc: None,
has_doc: None,
ignored: None,
doc_hidden: None,
link_name: None,
};
write_ipc(&out.join(format!("{stem}.nodes.arrow")), &mimir_schema::nodes_batch(&[row])?)?;
write_ipc(&out.join(format!("{stem}.edges.arrow")), &mimir_schema::edges_batch(&[])?)?;
write_ipc(&out.join(format!("{stem}.spans.arrow")), &mimir_schema::spans_batch(&[])?)?;
std::fs::write(
out.join(format!("{stem}.stats.json")),
format!(r#"{{"crate_name":"{krate}","origin":"lang","rustc_release":"1.101.0-nightly"}}"#),
)?;
}
Ok(())
}
fn write_ipc(p: &Path, b: &skade::arrow_array::RecordBatch) -> anyhow::Result<()> {
let mut w = arrow_ipc::writer::FileWriter::try_new(std::fs::File::create(p)?, &b.schema())?;
w.write(b)?;
w.finish()?;
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn the_toolchain_dump_runs_once_per_toolchain() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
// the MIMIR payloads: the std dump is mimir-schema rows
let wh = skade::open(&lake).await.unwrap();
let mut log = EventLog::open(wh, mimir_ingest::Specs::mimir().unwrap()).await.unwrap();
let sd = StdDumper { cargo_mimir_dump: "/nonexistent".into(), driver: None, toolchain: "nightly-fixture".into(), root: tmp.path().join("std"), host: Default::default() };
let mut runs = 0;
let o = sd.ensure_with(&mut log, |out| { runs += 1; fake_std(out) }).await.unwrap();
assert_eq!((o.status, o.crates, o.rows, o.rustc_release.as_deref()), (StdStatus::Dumped, 2, 2, Some("1.101.0-nightly")));
assert_eq!(runs, 1);
let manifest = log.universe_rows().await.unwrap();
assert_eq!(
manifest.iter().map(|m| (m.0.as_str(), m.1.as_str(), m.2, m.3.as_deref())).collect::<Vec<_>>(),
vec![("core@lang:lib", "nightly-fixture", 1, Some("1.101.0-nightly")), ("std@lang:lib", "nightly-fixture", 1, Some("1.101.0-nightly"))]
);
let n_events = log.events(Kind::Nodes).await.unwrap().iter().map(|b| b.num_rows()).sum::<usize>();
assert_eq!(n_events, 2);
drop(log);
// second run, a fresh process (log re-opened): skipped, the dump never runs
let wh = skade::open(&lake).await.unwrap();
let mut log = EventLog::open(wh, mimir_ingest::Specs::mimir().unwrap()).await.unwrap();
let o = sd.ensure_with(&mut log, |_| panic!("the std dump must not run again")).await.unwrap();
assert_eq!(o.status, StdStatus::Skipped);
assert_eq!(log.events(Kind::Nodes).await.unwrap().iter().map(|b| b.num_rows()).sum::<usize>(), 2);
assert_eq!(log.universe_rows().await.unwrap().len(), 2);
// files lost (another box), log has it: files re-made, NO new events
std::fs::remove_dir_all(sd.dir()).unwrap();
let mut runs = 0;
let o = sd.ensure_with(&mut log, |out| { runs += 1; fake_std(out) }).await.unwrap();
assert_eq!((o.status, runs), (StdStatus::FilesRestored, 1));
assert_eq!(log.events(Kind::Nodes).await.unwrap().iter().map(|b| b.num_rows()).sum::<usize>(), 2);
assert_eq!(log.universe_rows().await.unwrap().len(), 2);
// another toolchain: dumped again
let sd2 = StdDumper { toolchain: "nightly-other".into(), ..sd.clone() };
let o = sd2.ensure_with(&mut log, fake_std).await.unwrap();
assert_eq!(o.status, StdStatus::Dumped);
assert_eq!(log.universe_rows().await.unwrap().len(), 4);
}
#[tokio::test(flavor = "current_thread")]
async fn the_next_rev_comes_from_bin_offsets_written_in_the_same_release() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let shas = history(&src);
// b gets a bin too: two artifacts
src.write("crates/b/src/lib.rs", "fn h () | three\nlink h\n");
let c3 = src.commit("c3 b links h");
let mut c = boot(tmp.path(), true).await;
let rs = spec(tmp.path(), &src);
c.run(&rs).await.unwrap();
let offs = c.log().bin_offsets(REPO).await.unwrap();
let got: Vec<(&str, i64, i64, &str)> = offs.iter().map(|(a, o)| (a.as_str(), o.rev, o.seq, o.sha.as_str())).collect();
let (a, b) = (artifact("a"), artifact("b"));
assert_eq!(got, vec![(a.as_str(), 3, 3, c3.as_str()), (b.as_str(), 3, 3, c3.as_str())]);
let _ = shas;
// a restart: the log re-opened, the rev continues from bin_offsets
let (log, dumper) = c.into_parts();
drop(log);
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, dumper).await.unwrap();
let r = c.reingest_binaries(&rs).await.unwrap();
assert_eq!(r.iter().map(|x| x.rev).collect::<Vec<_>>(), vec![4, 4]);
let offs = c.log().bin_offsets(REPO).await.unwrap();
assert!(offs.values().all(|o| (o.rev, o.seq) == (4, 3)), "{offs:?}");
assert_eq!(c.log().bin_runs(REPO).await.unwrap().iter().map(|r| r.rev).max(), Some(4));
}
/// `fold_rows` (the payload-keeping fold readers use) opens exactly the
/// identities `fold` opens, per table and per consumer ordinal.
#[tokio::test(flavor = "current_thread")]
async fn fold_rows_opens_exactly_what_fold_opens() {
use skade::arrow_array::{Array, StringArray};
use skade::arrow_schema::DataType;
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let _ = history(&src);
let mut c = boot(tmp.path(), true).await;
c.run(&spec(tmp.path(), &src)).await.unwrap();
let log = c.log();
assert_eq!(log.offsets().await.unwrap().get(REPO).map(|o| o.seq), Some(2));
for upto in [1i64, 2] {
let open = log.fold(Some(REPO), Some(upto)).await.unwrap();
let scope = mimir_ingest::FoldScope { repos: [(REPO.to_string(), upto)].into(), universe: false };
for kind in Kind::ALL {
let want: BTreeSet<(String, String)> =
open.repo(kind, REPO).map(|o| o.iter().map(|(k, key, _)| (k.clone(), key.clone())).collect()).unwrap_or_default();
let rows = log.fold_rows(kind, &scope, &[]).await.unwrap();
let col = |b: &skade::arrow_array::RecordBatch, n: &str| -> StringArray {
let c = skade::arrow_cast::cast(b.column_by_name(n).unwrap(), &DataType::Utf8).unwrap();
c.as_any().downcast_ref::<StringArray>().unwrap().clone()
};
let mut got = BTreeSet::new();
for &(bi, ri) in &rows.open {
let b = &rows.batches[bi as usize];
let mut key: Vec<String> = log.spec(kind).key_columns.iter().map(|k| col(b, k).value(ri as usize).to_string()).collect();
let dup = col(b, "key_dup");
if !dup.is_null(ri as usize) {
key.push(dup.value(ri as usize).to_string());
}
got.insert((col(b, "krate").value(ri as usize).to_string(), key.join("\u{1f}")));
}
assert_eq!(got, want, "{kind:?} at seq {upto}");
assert!(!got.is_empty() || kind == Kind::Spans, "{kind:?} at seq {upto} must hold rows in this fixture");
}
}
// the universe ring is read only when asked for
let uni = mimir_ingest::FoldScope { repos: Default::default(), universe: true };
assert_eq!(log.fold_rows(Kind::Nodes, &uni, &[]).await.unwrap().open.len(), 1, "dep@1.0.0 helper");
let none = mimir_ingest::FoldScope::default();
assert!(log.fold_rows(Kind::Nodes, &none, &[]).await.unwrap().open.is_empty());
}
/// Lists every artifact, but from dump number `from` on declares no bin
/// target: the bins "left the workspace" while their executables are still
/// found.
struct Undeclared(FakeDumper, usize, usize);
impl mimir_ingest::Dumper for Undeclared {
fn dump(&mut self, worktree: &Path, sha: &str) -> anyhow::Result<mimir_ingest::CommitDump> {
let mut d = self.0.dump(worktree, sha)?;
self.2 += 1;
if self.2 >= self.1 {
d.workspace_targets.retain(|t| !t.contains(":bin:"));
}
Ok(d)
}
}
/// A listed artifact whose bin the workspace no longer declares closes ONCE
/// and stays closed (it used to flap: closed at one commit, reopened at the
/// next, because it was diffed both as dumped and as gone).
#[tokio::test(flavor = "current_thread")]
async fn a_stale_artifact_of_an_undeclared_bin_closes_once_and_does_not_flap() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
history(&src);
src.write("crates/b/src/lib.rs", "fn h () | four\n");
src.commit("c3 b edit");
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, Undeclared(FakeDumper { binary: true, ..Default::default() }, 2, 0)).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
let per_seq = |evs: Vec<Ev>| -> Vec<(i64, String)> {
let mut v: Vec<(i64, String)> = evs.into_iter().map(|e| (e.1, e.3)).collect();
v.dedup();
v
};
// seq 1: declared → written; seq 2: undeclared (still listed) → closed; seq 3: nothing
assert_eq!(per_seq(events(c.log(), Kind::BinNodes).await), vec![(1, "+".to_string()), (2, "-".to_string())]);
for k in Kind::BINARY {
assert!(c.open_set().repo(k, REPO).is_none_or(|o| o.is_empty()), "{k:?} still open");
}
}
/// An artifact listed while its bin is never declared never enters.
#[tokio::test(flavor = "current_thread")]
async fn an_artifact_of_a_never_declared_bin_never_enters() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
history(&src);
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, Undeclared(FakeDumper { binary: true, ..Default::default() }, 1, 0)).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
for k in Kind::BINARY {
assert!(events(c.log(), k).await.is_empty(), "{k:?}");
}
}