crates/mimir-ingest/tests/ingest/delta.rs
//! A tracked fold advanced by the lake's delta equals a fresh fold — the
//! room's advance must never differ from a full re-read (the oracle).
use std::collections::BTreeSet;
use super::common::*;
use mimir_ingest::schema::COL_REPO;
use mimir_ingest::{Consumer, FoldDelta, FoldScope, Kind, RepoSpec, Scan};
use skade::ScanFilter;
const REPO: &str = "fixture";
fn spec(tmp: &std::path::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 scope() -> FoldScope {
FoldScope { repos: [(REPO.to_string(), i64::MAX)].into(), universe: false }
}
fn scans() -> Vec<Scan> {
vec![Scan { filter: ScanFilter::is_in(COL_REPO, [REPO]), repo: None }]
}
/// `(moniker, body)` of every emitted row, by slot — what the caller's
/// records would hold.
fn keep(recs: &mut Vec<(String, String)>) -> impl FnMut(&skade::arrow_array::RecordBatch, &[u32]) -> anyhow::Result<()> + Send + '_ {
move |b, rows| {
let m = b.column_by_name("moniker").unwrap().as_any().downcast_ref::<skade::arrow_array::StringArray>().unwrap().clone();
let body = b.column_by_name("body").unwrap().as_any().downcast_ref::<skade::arrow_array::StringArray>().unwrap().clone();
for &r in rows {
recs.push((m.value(r as usize).to_string(), body.value(r as usize).to_string()));
}
Ok(())
}
}
#[tokio::test(flavor = "current_thread")]
async fn a_delta_fold_matches_a_fresh_fold() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
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");
src.commit("c1 init");
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log.clone(), FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), &src);
c.run(&rs).await.unwrap();
// the tracked fold at c1
let mut recs: Vec<(String, String)> = Vec::new();
let (open1, _events, mut track) = log.fold_stream_tracked(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut recs)).await.unwrap();
let at_c1: BTreeSet<&(String, String)> = open1.iter().map(|&s| &recs[s as usize]).collect();
assert!(at_c1.iter().any(|(m, b)| m == "a f" && b == "one"), "c1 holds f: {at_c1:?}");
// the history moves: a body edit, a new fn, crate b removed
src.write("crates/a/src/lib.rs", "fn f (x: i32) | uno\nfn g () | two\nfn k () | four\ncall g -> f\n");
src.commit("c2 edit + new fn");
src.remove("crates/b");
src.commit("c3 crate b removed");
c.run(&rs).await.unwrap();
// the delta advances the SAME records
let adv = log.fold_stream_delta(Kind::Nodes, &scope(), &scans(), &["body"], &mut track, keep(&mut recs)).await.unwrap();
let FoldDelta::Advanced { open, events } = adv else { panic!("pure appends advance, never a full reload") };
assert!(events > 0, "the delta read the new events");
let advanced: BTreeSet<(String, String)> = open.iter().map(|&s| recs[s as usize].clone()).collect();
// a fresh fold of the whole log is the oracle
let mut fresh: Vec<(String, String)> = Vec::new();
let (open_f, _) = log.fold_stream(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut fresh)).await.unwrap();
let oracle: BTreeSet<(String, String)> = open_f.iter().map(|&s| fresh[s as usize].clone()).collect();
assert_eq!(advanced, oracle, "the advanced open set IS the fresh fold's");
assert!(advanced.iter().any(|(m, b)| m == "a f" && b == "uno"), "the edit landed");
assert!(advanced.iter().any(|(m, _)| m == "a k"), "the new fn landed");
assert!(!advanced.iter().any(|(m, _)| m == "b h"), "crate b's fn is gone");
}
#[tokio::test(flavor = "current_thread")]
async fn a_caught_up_delta_is_empty_and_cheap() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n").write("crates/a/src/lib.rs", "fn f () | one\n");
src.commit("c1");
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log.clone(), FakeDumper::default()).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
let mut recs: Vec<(String, String)> = Vec::new();
let (open1, _, mut track) = log.fold_stream_tracked(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut recs)).await.unwrap();
let before = recs.len();
let adv = log.fold_stream_delta(Kind::Nodes, &scope(), &scans(), &["body"], &mut track, keep(&mut recs)).await.unwrap();
let FoldDelta::Advanced { open, events } = adv else { panic!("nothing new is still an advance") };
assert_eq!(events, 0, "nothing was read");
assert_eq!(recs.len(), before, "nothing was emitted");
assert_eq!(open, open1, "the open set stands");
}
/// Wave item 8 — COMPACTION-AWARE delta. A lake compaction commits a stamped
/// row-preserving `replace` (and expires the snapshots before it); a tracked
/// fold whose read mark is the compaction's base ADVANCES across it without
/// re-reading a row, and the open set equals a fresh fold (the oracle).
#[tokio::test(flavor = "current_thread")]
async fn a_delta_advances_across_a_lake_compaction() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
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");
src.commit("c1 init");
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log.clone(), FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), &src);
c.run(&rs).await.unwrap();
// the tracked fold, then a first delta so the track sits at the head
let mut recs: Vec<(String, String)> = Vec::new();
let (_open1, _, mut track) = log.fold_stream_tracked(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut recs)).await.unwrap();
src.write("crates/a/src/lib.rs", "fn f (x: i32) | uno\nfn g () | two\ncall g -> f\n");
src.commit("c2 edit");
c.run(&rs).await.unwrap();
let adv = log.fold_stream_delta(Kind::Nodes, &scope(), &scans(), &["body"], &mut track, keep(&mut recs)).await.unwrap();
assert!(matches!(adv, FoldDelta::Advanced { .. }), "pure appends advance");
// the REAL compaction: one stamped replace per table + expiry + sweep
let opts = mimir_ingest::CompactOptions { grace: std::time::Duration::ZERO, ..Default::default() };
let reports = log.compact(&opts).await.unwrap();
let nodes = reports.iter().find(|r| r.table == "code_node_events").unwrap();
assert!(nodes.snapshot.is_some(), "a replace landed on the node table: {nodes}");
// more history lands after the compaction
src.write("crates/a/src/lib.rs", "fn f (x: i32) | uno\nfn g () | two\nfn k () | four\ncall g -> f\n");
src.commit("c3 new fn");
c.run(&rs).await.unwrap();
// THE POINT: the delta advances ACROSS the compaction — no Full, no error
let adv = log.fold_stream_delta(Kind::Nodes, &scope(), &scans(), &["body"], &mut track, keep(&mut recs)).await.unwrap();
let FoldDelta::Advanced { open, events } = adv else {
panic!("a stamped row-preserving replace must not refold the track")
};
assert!(events > 0, "the post-compaction append was read");
let advanced: BTreeSet<(String, String)> = open.iter().map(|&s| recs[s as usize].clone()).collect();
// the oracle: a fresh fold of the whole log
let mut fresh: Vec<(String, String)> = Vec::new();
let (open_f, _) = log.fold_stream(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut fresh)).await.unwrap();
let oracle: BTreeSet<(String, String)> = open_f.iter().map(|&s| fresh[s as usize].clone()).collect();
assert_eq!(advanced, oracle, "the open set across the compaction IS the fresh fold's");
assert!(advanced.iter().any(|(m, _)| m == "a k"), "the post-compaction fn landed");
}
/// An append BETWEEN the track's read mark and the
/// compaction was rewritten INTO the replace's files — the live order after
/// every consume. The delta reads THROUGH the stamped replace: its merged
/// files bring rows the track already holds back, and the idempotent apply
/// lets them re-take their slots, so the open set still equals a fresh fold.
#[tokio::test(flavor = "current_thread")]
async fn an_append_swallowed_by_the_compaction_is_read_through_the_replace() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
src.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n")
.write("crates/a/src/lib.rs", "fn f (x: i32) | one\n");
src.commit("c1 init");
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log.clone(), FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), &src);
c.run(&rs).await.unwrap();
let mut recs: Vec<(String, String)> = Vec::new();
let (_o, _, mut track) = log.fold_stream_tracked(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut recs)).await.unwrap();
// an append the track has NOT seen, then the compaction rewrites it
src.write("crates/a/src/lib.rs", "fn f (x: i32) | one\nfn g () | two\n");
src.commit("c2 unseen");
c.run(&rs).await.unwrap();
let opts = mimir_ingest::CompactOptions { grace: std::time::Duration::ZERO, ..Default::default() };
let reports = log.compact(&opts).await.unwrap();
assert!(reports.iter().any(|r| r.snapshot.is_some()), "{reports:?}");
src.write("crates/a/src/lib.rs", "fn f (x: i32) | one\nfn g () | two\nfn h () | three\n");
src.commit("c3 after");
c.run(&rs).await.unwrap();
let adv = log.fold_stream_delta(Kind::Nodes, &scope(), &scans(), &["body"], &mut track, keep(&mut recs)).await.unwrap();
let FoldDelta::Advanced { open, .. } = adv else {
panic!("a swallowed append is read THROUGH the stamped replace, not refolded")
};
let advanced: BTreeSet<(String, String)> = open.iter().map(|&s| recs[s as usize].clone()).collect();
let mut fresh: Vec<(String, String)> = Vec::new();
let (open_f, _) = log.fold_stream(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut fresh)).await.unwrap();
let oracle: BTreeSet<(String, String)> = open_f.iter().map(|&s| fresh[s as usize].clone()).collect();
assert_eq!(advanced, oracle, "the replayed rows re-take their slots: the open set IS the fresh fold's");
assert!(advanced.iter().any(|(m, b)| m == "a g" && b == "two"), "the swallowed append landed");
assert!(advanced.iter().any(|(m, b)| m == "a h" && b == "three"), "the post-compaction append landed");
}
#[tokio::test(flavor = "current_thread")]
async fn a_zone_seeds_the_tracked_fold_and_the_delta_still_matches_a_fresh_fold() {
// the tracked fold used to scan the whole event history (no zone): with
// a materialized zone it must read the zone + the tail instead — fewer
// events — and the track it seeds must behave EXACTLY like the scanned
// one under a later delta (the standing oracle).
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
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");
src.commit("c1 init");
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log.clone(), FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), &src);
c.run(&rs).await.unwrap();
// churn so the zone buys something: five rewrites of the same fns
for n in 2..=6 {
src.write("crates/a/src/lib.rs", &format!("fn f (x: i32) | v{n}\nfn g () | two\ncall g -> f\n"));
src.commit(&format!("c{n}"));
}
c.run(&rs).await.unwrap();
log.materialize(REPO).await.unwrap();
// one commit after the zone: the tail
src.write("crates/a/src/lib.rs", "fn f (x: i32) | tail\nfn g () | two\nfn t () | tee\ncall g -> f\n");
src.commit("c7 tail");
c.run(&rs).await.unwrap();
// the un-zoned truth: a fresh fold's open set
let mut fresh: Vec<(String, String)> = Vec::new();
let (open_f, events_full) = log.fold_stream(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut fresh)).await.unwrap();
let oracle: std::collections::BTreeSet<(String, String)> = open_f.iter().map(|&s| fresh[s as usize].clone()).collect();
// the tracked fold with the zone present: same open set, and a read cut
// to the tail — strictly below the whole history (fold_stream's own
// count is zoned too, so the history is the honest yardstick)
let history = events(&log, Kind::Nodes).await.len() as u64;
let mut recs: Vec<(String, String)> = Vec::new();
let (open_t, events_tracked, mut track) = log.fold_stream_tracked(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut recs)).await.unwrap();
let tracked: std::collections::BTreeSet<(String, String)> = open_t.iter().map(|&s| recs[s as usize].clone()).collect();
assert_eq!(tracked, oracle, "the zone-seeded fold IS the fresh fold");
assert!(
events_tracked < history,
"the zone must cut the read: tracked {events_tracked} vs the whole history {history}"
);
let _ = events_full;
// and the seeded track advances like a scanned one (the standing oracle)
src.write("crates/a/src/lib.rs", "fn f (x: i32) | v8\nfn g () | two\nfn t () | tee\ncall g -> f\n");
src.commit("c8");
c.run(&rs).await.unwrap();
let adv = log.fold_stream_delta(Kind::Nodes, &scope(), &scans(), &["body"], &mut track, keep(&mut recs)).await.unwrap();
let FoldDelta::Advanced { open, .. } = adv else { panic!("appends advance") };
let advanced: std::collections::BTreeSet<(String, String)> = open.iter().map(|&s| recs[s as usize].clone()).collect();
let mut fresh2: Vec<(String, String)> = Vec::new();
let (open_f2, _) = log.fold_stream(Kind::Nodes, &scope(), &scans(), &["body"], keep(&mut fresh2)).await.unwrap();
let oracle2: std::collections::BTreeSet<(String, String)> = open_f2.iter().map(|&s| fresh2[s as usize].clone()).collect();
assert_eq!(advanced, oracle2, "a zone-seeded track advances to the fresh fold's open set");
assert!(advanced.iter().any(|(_, b)| b == "v8"), "the newest body landed");
}
/// The maintainer's body (the server's `lake_readfast` job): ONE call zones
/// EVERY consumed repo stream — none is left to chance, none hides behind
/// another — and the zone-seeded tracked fold over all of them IS the
/// whole-history fold (zero loss). A second call finds every anchor at its
/// tip and writes nothing.
#[tokio::test(flavor = "current_thread")]
async fn materialize_all_zones_every_repo_stream_and_the_seeded_fold_is_the_history_fold() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
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");
src.commit("c1 init");
let src2 = Scratch::new(tmp.path().join("src2"));
src2.write("Cargo.toml", "[workspace]\nmembers = [\"crates/*\"]\n").write("crates/z/src/lib.rs", "fn zf () | zed\n");
src2.commit("z1 init");
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log.clone(), FakeDumper::default()).await.unwrap();
let rs = spec(tmp.path(), &src);
let rs2 = RepoSpec {
name: "second".into(),
source: src2.root.to_string_lossy().into_owned(),
branch: "main".into(),
worktree: tmp.path().join("clone2"),
initial_offset: None,
};
c.run(&rs).await.unwrap();
c.run(&rs2).await.unwrap();
// churn on both streams so the zones buy something
for n in 2..=5 {
src.write("crates/a/src/lib.rs", &format!("fn f (x: i32) | v{n}\nfn g () | two\ncall g -> f\n"));
src.commit(&format!("c{n}"));
src2.write("crates/z/src/lib.rs", &format!("fn zf () | zv{n}\n"));
src2.commit(&format!("z{n}"));
}
c.run(&rs).await.unwrap();
c.run(&rs2).await.unwrap();
let both_scope = || FoldScope { repos: [(REPO.to_string(), i64::MAX), ("second".to_string(), i64::MAX)].into(), universe: false };
let both_scans = || vec![Scan { filter: ScanFilter::is_in(COL_REPO, [REPO, "second"]), repo: None }];
// the un-zoned truth FIRST: no stream has a zone yet, so this fold is the
// pure event-history answer — the oracle the zones must reproduce
for r in [REPO, "second"] {
assert_eq!(log.readfast_anchor(r).await.unwrap(), None, "{r}: no zone before the maintainer ran");
}
let mut fresh: Vec<(String, String)> = Vec::new();
let (open_f, _) = log.fold_stream(Kind::Nodes, &both_scope(), &both_scans(), &["body"], keep(&mut fresh)).await.unwrap();
let oracle: BTreeSet<(String, String)> = open_f.iter().map(|&s| fresh[s as usize].clone()).collect();
let history = events(&log, Kind::Nodes).await.len() as u64;
// THE JOB PATH: one sweep, every stream
let sweep = log.materialize_all().await.unwrap();
assert!(sweep.failed.is_empty(), "no stream failed: {:?}", sweep.failed);
let written: BTreeSet<&str> = sweep.written.iter().map(|r| r.repo.as_str()).collect();
assert_eq!(written, [REPO, "second"].into_iter().collect::<BTreeSet<_>>(), "EVERY consumed repo stream got its zone");
let offsets = log.offsets().await.unwrap();
for r in [REPO, "second"] {
assert_eq!(log.readfast_anchor(r).await.unwrap(), Some(offsets[r].seq), "{r}: anchored at the stream's tip");
}
// zero loss: the zone-seeded tracked fold IS the history fold, and reads less
let mut recs: Vec<(String, String)> = Vec::new();
let (open_t, events_tracked, _track) = log.fold_stream_tracked(Kind::Nodes, &both_scope(), &both_scans(), &["body"], keep(&mut recs)).await.unwrap();
let tracked: BTreeSet<(String, String)> = open_t.iter().map(|&s| recs[s as usize].clone()).collect();
assert_eq!(tracked, oracle, "the zone-seeded fold IS the whole-history fold");
assert!(events_tracked < history, "the zones cut the read: tracked {events_tracked} vs the whole history {history}");
assert!(tracked.iter().any(|(m, b)| m == "a f" && b == "v5"), "fixture's newest body");
assert!(tracked.iter().any(|(m, b)| m == "z zf" && b == "zv5"), "second's newest body");
// idempotent: a second sweep finds every anchor at its tip
let again = log.materialize_all().await.unwrap();
assert!(again.written.is_empty() && again.failed.is_empty(), "{again:?}");
assert_eq!(again.skipped, 2, "both streams already anchored");
}