src/code_room/consumer_tests.rs
//! The workspace consumer end to end, without the nightly driver: scratch git
//! repos as members, a dumper that reads the checked-out tree and emits real
//! `mimir-schema` rows (so rooms open over what it wrote), the real log.
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use mimir_ingest::{BinaryDump, CommitDump, CrateDump, Dumper, StdDumper};
use mimir_room::Scope;
use mimir_schema::{EdgeKind, EdgeRow, Level, NodeRow, Origin};
use super::consumer_loop::{Deferred, DeferredWork, Dumpers, Event, Feed, FeedAt, FeedResult, Member, MemberStatus, Request};
use super::{CodeRoom, Layout, Refusal, State};
const WAIT: Duration = Duration::from_secs(60);
/// A member's source repo: `src/lib.rs` holds `fn NAME` and `call A B` lines;
/// `BROKEN` makes the crate fail to compile; `CRASH` makes the dumper die.
/// `needs NAME` names a sibling by path (the dump FAILS when its clone is not
/// there); `sees NAME FN` uses the item `FN` of the sibling `NAME` (the crate
/// does not compile against a sibling tree that lacks it).
/// `ffi A NAME` makes `A` call the foreign item `NAME` of the dependency
/// `libc` (linked as `__NAME`): the dump then carries that crate as a
/// universe crate, and the member's crate node with a DEPENDS_ON edge to it.
pub(crate) struct Repo {
pub(crate) dir: PathBuf,
}
impl Repo {
pub(crate) fn new(root: &Path, ws: &str, name: &str) -> Self {
let dir = root.join(ws).join("git").join(name);
crate::gitio::init(&dir).unwrap();
Self { dir }
}
pub(crate) fn commit(&self, lib: &str) -> String {
std::fs::create_dir_all(self.dir.join("src")).unwrap();
std::fs::write(self.dir.join("src/lib.rs"), lib).unwrap();
crate::gitio::commit_all(&self.dir, "c").unwrap()
}
pub(crate) fn docs(&self, text: &str) -> String {
std::fs::write(self.dir.join("README.md"), text).unwrap();
crate::gitio::commit_all(&self.dir, "docs").unwrap()
}
pub(crate) fn member(&self, name: &str) -> Member {
Member { name: name.into(), source: self.dir.clone(), branch: None, head: crate::gitio::head_sha(&self.dir).unwrap() }
}
}
fn node(krate: &str, name: &str) -> NodeRow {
NodeRow::new(format!("{krate} {name}"), "fn", Level::Item, name, krate, "0.1.0", Origin::Local)
}
/// Reads the checked-out tree; counts its dumps; can be held inside one.
struct TreeDumper {
krate: String,
dumps: Arc<AtomicUsize>,
gate: Option<Arc<Gate>>,
}
pub(crate) struct Gate {
hold_at: usize,
entered: Mutex<Sender<()>>,
go: Mutex<Receiver<()>>,
}
impl Dumper for TreeDumper {
fn dump(&mut self, worktree: &Path, _sha: &str) -> anyhow::Result<CommitDump> {
let n = self.dumps.fetch_add(1, Ordering::SeqCst) + 1;
if let Some(g) = &self.gate
&& n == g.hold_at
{
g.entered.lock().unwrap().send(()).unwrap();
g.go.lock().unwrap().recv_timeout(WAIT).expect("the test never released the gated dump");
}
let src = std::fs::read_to_string(worktree.join("src/lib.rs"))?;
if src.contains("CRASH") {
anyhow::bail!("the dumper died (infrastructure)");
}
// `needs NAME`: the tree names a sibling by path, as a manifest does (`../NAME`)
for sibling in src.lines().filter_map(|l| l.trim().strip_prefix("needs ")) {
let at = worktree.parent().expect("a clone has a parent").join(sibling);
anyhow::ensure!(at.join(".git").exists(), "failed to read `{}`: the sibling is not cloned", at.join("Cargo.toml").display());
}
// `sees NAME FN`: the tree uses the item FN of the sibling NAME — it
// compiles only against a sibling tree that has it
let blind = src.lines().filter_map(|l| l.trim().strip_prefix("sees ")).filter_map(|r| r.split_once(' ')).any(|(sibling, item)| {
let lib = worktree.parent().expect("a clone has a parent").join(sibling).join("src/lib.rs");
!std::fs::read_to_string(lib).is_ok_and(|s| s.lines().any(|l| l.trim() == format!("fn {item}")))
});
let k = &self.krate;
let owner = format!("{k}@0.1.0:lib");
let target: BTreeSet<String> = [format!("{k}:lib")].into();
if src.contains("BROKEN") || blind {
return Ok(CommitDump {
toolchain: "nightly-test".into(),
workspace_crates: BTreeSet::new(),
workspace_targets: target.clone(),
failed_units: target,
owner_disambiguations: 0,
crates: vec![],
binary: false,
binaries: vec![],
});
}
let (mut nodes, mut edges, mut foreign) = (Vec::new(), Vec::new(), Vec::new());
for line in src.lines().map(str::trim) {
if let Some(f) = line.strip_prefix("fn ") {
nodes.push(node(k, f));
} else if let Some(rest) = line.strip_prefix("call ") {
let (a, b) = rest.split_once(' ').unwrap();
edges.push(EdgeRow::new(format!("{k} {a}"), format!("{k} {b}"), EdgeKind::Calls));
} else if let Some(rest) = line.strip_prefix("ffi ") {
let (a, name) = rest.split_once(' ').unwrap();
edges.push(EdgeRow::new(format!("{k} {a}"), format!("libc {name}"), EdgeKind::Calls));
foreign.push(name.to_string());
}
}
let mut crates = Vec::new();
if !foreign.is_empty() {
// the dependency ring: `libc`, dumped as a universe crate, and the
// DEPENDS_ON edge that makes it part of this member's ring
nodes.push(NodeRow::new(format!("{k} crate"), "crate", Level::Crate, k.as_str(), k.as_str(), "0.1.0", Origin::Local));
edges.push(EdgeRow::new(format!("{k} crate"), "libc crate", EdgeKind::DependsOn));
let mut ring = vec![NodeRow::new("libc crate", "crate", Level::Crate, "libc", "libc", "0.2.0", Origin::Library)];
for name in &foreign {
let mut item = NodeRow::new(format!("libc {name}"), "ffi", Level::Item, name.as_str(), "libc", "0.2.0", Origin::Library);
item.parent = Some("libc crate".into());
item.link_name = Some(format!("__{name}"));
ring.push(item);
}
crates.push(CrateDump { krate: "libc@0.2.0:lib".into(), universe: true, nodes: vec![mimir_schema::nodes_batch(&ring)?], edges: vec![mimir_schema::edges_batch(&[])?], spans: vec![] });
}
crates.push(CrateDump {
krate: owner.clone(),
universe: false,
nodes: vec![mimir_schema::nodes_batch(&nodes)?],
edges: vec![mimir_schema::edges_batch(&edges)?],
spans: vec![],
});
Ok(CommitDump {
toolchain: "nightly-test".into(),
workspace_crates: [owner].into(),
workspace_targets: target,
failed_units: BTreeSet::new(),
owner_disambiguations: 0,
crates,
binary: false,
binaries: vec![],
})
}
}
/// Hands every member a [`TreeDumper`]; counts dumps per member.
#[derive(Default)]
pub(crate) struct TreeDumpers {
dumps: Mutex<std::collections::BTreeMap<String, Arc<AtomicUsize>>>,
gate: Option<(String, Arc<Gate>)>,
/// The toolchain dump this source offers (none by default: a scripted
/// dumper has no toolchain).
std: Option<StdDumper>,
}
impl TreeDumpers {
pub(crate) fn dumps(&self, member: &str) -> usize {
self.dumps.lock().unwrap().get(member).map_or(0, |n| n.load(Ordering::SeqCst))
}
/// Dumpers whose `n`-th dump of `member` stops inside the dump: it says so
/// on the first channel and goes on when the second is sent to.
pub(crate) fn held_at(member: &str, n: usize) -> (Self, Receiver<()>, Sender<()>) {
let (entered_tx, entered) = channel();
let (go, go_rx) = channel();
let gate = Arc::new(Gate { hold_at: n, entered: Mutex::new(entered_tx), go: Mutex::new(go_rx) });
(Self { gate: Some((member.into(), gate)), ..Default::default() }, entered, go)
}
}
impl Dumpers for TreeDumpers {
fn dumper(&self, _: &Layout, member: &Member) -> Result<Box<dyn Dumper + Send>, Refusal> {
let dumps = self.dumps.lock().unwrap().entry(member.name.clone()).or_default().clone();
let gate = self.gate.as_ref().filter(|(m, _)| m == &member.name).map(|(_, g)| g.clone());
Ok(Box::new(TreeDumper { krate: member.name.clone(), dumps, gate }))
}
fn describe(&self) -> String {
"tree dumper (test)".into()
}
fn std_dumper(&self, _: &Layout) -> Option<StdDumper> {
self.std.clone()
}
}
/// No driver on this box.
pub(crate) struct NoDriver;
impl Dumpers for NoDriver {
fn dumper(&self, _: &Layout, _: &Member) -> Result<Box<dyn Dumper + Send>, Refusal> {
Err(Refusal::new(State::DriverNotAvailable, "the Mímir driver is not installed"))
}
fn describe(&self) -> String {
"none".into()
}
}
fn quiet() -> impl FnMut(Event<'_>) {
|_| {}
}
/// (member, seq, status) of every commit event, in order.
fn commits_of(events: &[String]) -> Vec<&str> {
events.iter().filter(|e| e.starts_with("commit ")).map(String::as_str).collect()
}
fn recorder(into: &mut Vec<String>) -> impl FnMut(Event<'_>) + '_ {
|e| {
into.push(match e {
Event::Member { member, behind } => format!("member {member} behind={behind}"),
Event::Begin { member, seq, behind, .. } => format!("begin {member}@{seq} behind={behind}"),
Event::Commit { member, report } => format!("commit {member}@{} {}", report.seq, report.status),
Event::Feed { member, row } => format!("feed {member} {} {}", row.feed, row.status),
Event::Deferred(d) => format!("deferred {} {}", d.member, d.feed),
Event::StdBegin { toolchain } => format!("std-begin {toolchain}"),
Event::Std { toolchain, outcome, .. } => format!("std {toolchain} {}", if outcome.is_ok() { "ok" } else { "failed" }),
Event::Done { outcome } => format!("done {} {}", outcome.member, outcome.status.as_str()),
})
}
}
#[test]
fn a_new_workspace_starts_at_each_tip_then_follows_its_members() {
let tmp = tempfile::tempdir().unwrap();
let (a, b) = (Repo::new(tmp.path(), "ws", "alpha"), Repo::new(tmp.path(), "ws", "beta"));
a.commit("fn a1\n");
a.commit("fn a1\nfn a2\ncall a2 a1\n");
let a_tip = a.docs("the tip of alpha only touches docs\n");
let b_tip = b.commit("fn b1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha"), b.member("beta")];
assert_eq!(room.behind(&members, &[]).unwrap(), vec!["alpha", "beta"], "no offset = behind");
// ---- first run: ONE release per member, at its tip (decision D2)
let mut events = Vec::new();
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut recorder(&mut events)).unwrap();
assert_eq!(
events,
vec![
"member alpha behind=1",
"begin alpha@1 behind=0",
"commit alpha@1 ok",
"done alpha caught_up",
"member beta behind=1",
"begin beta@1 behind=0",
"commit beta@1 ok",
"done beta caught_up",
]
);
assert_eq!(rep.stamp(), "alpha@1,beta@1");
assert_eq!((rep.members[0].offset.as_ref().unwrap().sha.as_str(), rep.members[1].offset.as_ref().unwrap().sha.as_str()), (a_tip.as_str(), b_tip.as_str()));
assert_eq!(rep.members[0].reports[0].rows_plus, 3, "a1, a2, a2→a1: the whole tree at the tip, though the tip commit is docs-only");
assert!(room.behind(&members, &[]).unwrap().is_empty());
// the clones are where the layout says (no open-set cache any more:
// the READ-FAST zone in the lake replaced it)
assert!(room.layout().clone_of("alpha").join(".git").exists());
// ---- nothing moved: nothing is fetched, dumped or written
let again = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!(again.commits(), 0);
assert_eq!((dumpers.dumps("alpha"), dumpers.dumps("beta")), (1, 1));
// ---- alpha moves by two commits: alpha is behind, beta is not touched
a.commit("fn a1\nfn a2\nfn a3\ncall a2 a1\n");
a.commit("fn a1\nfn a3\n");
let members = [a.member("alpha"), b.member("beta")];
assert_eq!(room.behind(&members, &[]).unwrap(), vec!["alpha"]);
let mut events = Vec::new();
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut recorder(&mut events)).unwrap();
assert_eq!(commits_of(&events), vec!["commit alpha@2 ok", "commit alpha@3 ok"]);
assert_eq!(rep.stamp(), "alpha@3,beta@1");
assert_eq!((rep.members[0].reports[1].rows_plus, rep.members[0].reports[1].rows_minus), (0, 2), "a2 and a2→a1 close");
assert_eq!(dumpers.dumps("beta"), 1);
// ---- the room over what was written: both repos at their offsets
let r = room.room(&Scope::current()).unwrap();
assert_eq!((r.scope().repos["alpha"], r.scope().repos["beta"]), (3, 1));
assert!(r.ordinal("alpha a3").is_some() && r.ordinal("alpha a2").is_none() && r.ordinal("beta b1").is_some());
// ---- status: offsets, ledger rows, nothing behind
let st = room.status(&members).unwrap();
assert_eq!(st.members.iter().map(|m| (m.member.as_str(), m.offset.as_ref().unwrap().seq, m.runs, m.behind)).collect::<Vec<_>>(), vec![("alpha", 3, 3, false), ("beta", 1, 1, false)]);
assert!(st.behind().is_empty() && st.in_flight.is_none() && st.refusal.is_none());
assert!(st.generation().starts_with("lake ") && st.generation().contains("alpha@3") && st.generation().contains("beta@1"));
}
#[test]
fn a_run_is_bounded_per_member_and_from_backfills() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
let first = a.commit("fn a1\n");
a.commit("fn a1\nfn a2\n");
a.commit("fn a1\nfn a2\nfn a3\n");
a.commit("fn a1\nfn a2\nfn a3\nfn a4\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha")];
// ---- `--from <sha>` for a member with no offset: its history starts there
let rep = room
.consume(&dumpers, &[], &Request { members: &members, from: Some(&first[..12]), max_commits: Some(2), ..Default::default() }, &mut quiet())
.unwrap();
let m = &rep.members[0];
assert_eq!((m.status, m.reports.len(), m.behind, m.offset.as_ref().unwrap().seq), (MemberStatus::Behind, 2, 2, 2));
assert_eq!(m.reports[0].sha, first);
assert_eq!(room.behind(&members, &[]).unwrap(), vec!["alpha"], "the budget ran out: still behind");
assert!(m.line().contains("behind") && m.line().contains("2 commit(s)"), "{}", m.line());
// ---- the next run resumes at the offset (no `from`: that would replay)
let rep = room.consume(&dumpers, &[], &Request { members: &members, max_commits: Some(5), ..Default::default() }, &mut quiet()).unwrap();
let m = &rep.members[0];
assert_eq!((m.status, m.reports.iter().map(|r| r.seq).collect::<Vec<_>>(), m.behind), (MemberStatus::CaughtUp, vec![3, 4], 0));
// ---- `--from <date>` on a consumed member: a replay, ordinals keep counting
let rep = room.consume(&dumpers, &[], &Request { members: &members, from: Some("1970-01-02"), ..Default::default() }, &mut quiet()).unwrap();
let m = &rep.members[0];
assert_eq!(m.reports.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![5, 6, 7, 8]);
assert_eq!((m.reports[0].rows_plus, m.reports[0].rows_minus), (0, 3), "back to the first commit's tree: a2 a3 a4 close");
assert_eq!(room.room(&Scope::repo("alpha", Some(5))).unwrap().len(), 1);
assert_eq!(room.room(&Scope::current()).unwrap().len(), 4);
// ---- a `from` that names nothing is that member's failure, said by name
let rep = room.consume(&dumpers, &[], &Request { members: &members, from: Some("2999-01-01"), ..Default::default() }, &mut quiet()).unwrap();
assert_eq!(rep.members[0].status, MemberStatus::Failed);
assert!(rep.members[0].error.as_deref().unwrap().contains("at or after 2999-01-01"), "{:?}", rep.members[0].error);
// ---- `only` an unknown member: an argument error, not an empty report
let r = room.consume(&dumpers, &[], &Request { members: &members, only: Some("nope"), ..Default::default() }, &mut quiet()).unwrap_err();
assert_eq!(r.state, State::InvalidArgument);
assert!(r.note.contains("alpha"), "{r}");
}
/// HEAD on no branch, and no branch left in the clone — what a clone of a
/// source whose own HEAD is detached looks like.
fn detach(repo: &Path, sha: &str) {
std::fs::write(repo.join(".git/HEAD"), format!("{sha}\n")).unwrap();
let _ = std::fs::remove_file(repo.join(".git/refs/heads/main"));
assert_eq!(crate::gitio::head_branch(repo).unwrap(), "(detached)");
}
/// The branch a member follows is NAMED — by the workspace descriptor, the
/// clone's own branch or its remote's default — never guessed to be `main`.
/// With nothing to name it the member's turn ends `no_branch`, in a sentence
/// that says what to set: a state of the member, not a failure of the run.
#[test]
fn a_member_with_no_branch_to_follow_says_what_to_set() {
let tmp = tempfile::tempdir().unwrap();
let (a, b, c) = (Repo::new(tmp.path(), "ws", "alpha"), Repo::new(tmp.path(), "ws", "beta"), Repo::new(tmp.path(), "ws", "gamma"));
let sha = a.commit("fn a1\n");
b.commit("fn b1\n");
c.commit("fn c1\n");
detach(&a.dir, &sha);
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
// gamma: the descriptor names a branch its clone does not have
let gamma = Member { branch: Some("release".into()), ..c.member("gamma") };
let members = [a.member("alpha"), b.member("beta"), gamma];
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let st: Vec<_> = rep.members.iter().map(|m| (m.member.as_str(), m.status, m.status.as_str())).collect();
assert_eq!(st, vec![("alpha", MemberStatus::NoBranch, "no_branch"), ("beta", MemberStatus::CaughtUp, "caught_up"), ("gamma", MemberStatus::NoBranch, "no_branch")]);
assert!(rep.failed().is_empty(), "nothing failed: two members have no branch to follow");
let said = rep.members[0].error.as_deref().unwrap();
assert!(said.contains("`alpha` has no branch to follow") && said.contains("detached HEAD") && said.contains("set `branch = "), "{said}");
assert!(!said.contains("refs/") && !said.contains("main"), "plain, and no branch name is guessed: {said}");
assert!(rep.members[0].offset.is_none() && rep.members[0].reports.is_empty());
let said = rep.members[2].error.as_deref().unwrap();
assert!(said.contains("`gamma` is to follow branch `release` (the workspace descriptor)") && said.contains("no such branch") && said.contains("set `branch = "), "{said}");
// the remote names the default branch: a server clone that was detached
// after it was cloned is still followed on that branch
let delta = tmp.path().join("ws/git/delta");
crate::procreap::clone_or_fetch(&b.dir.to_string_lossy(), &delta, None).unwrap();
let tip = crate::gitio::head_sha(&delta).unwrap();
crate::gitio::detach_head_and_checkout(&delta, &tip).unwrap();
assert_eq!(crate::gitio::head_branch(&delta).unwrap(), "(detached)");
assert_eq!(mimir_ingest::git::remote_default_branch(&delta).unwrap().as_deref(), Some("main"));
assert_eq!(mimir_ingest::git::remote_default_branch(&a.dir).unwrap(), None, "no remote at all");
let members = [Member { name: "delta".into(), source: delta, branch: None, head: tip.clone() }];
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let m = &rep.members[0];
assert_eq!((m.status, m.offset.as_ref().map(|o| (o.branch.as_str(), o.sha.as_str()))), (MemberStatus::CaughtUp, Some(("main", tip.as_str()))), "{:?}", m.error);
// the descriptor names alpha's branch and the clone has it: consumed
std::fs::write(a.dir.join(".git/refs/heads/main"), format!("{sha}\n")).unwrap();
let members = [Member { branch: Some("main".into()), ..a.member("alpha") }];
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((rep.members[0].status, rep.members[0].offset.as_ref().map(|o| o.seq)), (MemberStatus::CaughtUp, Some(1)), "{:?}", rep.members[0].error);
}
#[test]
fn a_member_that_fails_does_not_stop_the_others_and_a_broken_commit_does_not_block() {
let tmp = tempfile::tempdir().unwrap();
let (a, b, c) = (Repo::new(tmp.path(), "ws", "alpha"), Repo::new(tmp.path(), "ws", "beta"), tmp.path().join("ws/git/gamma"));
a.commit("fn a1\nCRASH\n");
b.commit("fn b1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let gamma = Member { name: "gamma".into(), source: c, branch: None, head: String::new() };
let members = [a.member("alpha"), b.member("beta"), gamma.clone()];
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let st: Vec<_> = rep.members.iter().map(|m| (m.member.as_str(), m.status)).collect();
assert_eq!(st, vec![("alpha", MemberStatus::Failed), ("beta", MemberStatus::CaughtUp), ("gamma", MemberStatus::NoSource)]);
assert!(rep.members[0].error.as_deref().unwrap().contains("the dumper died"), "{:?}", rep.members[0].error);
assert!(rep.members[0].offset.is_none(), "an infrastructure failure does not move the offset");
assert_eq!(rep.stamp(), "beta@1");
assert_eq!(rep.failed().len(), 1);
// the failure is in the ledger (status reads it), and alpha is still behind
let status = room.status(&members).unwrap();
let alpha = &status.members[0];
assert_eq!((alpha.runs, alpha.last_run.as_ref().unwrap().status.as_str(), alpha.behind), (1, "error", true));
assert!(alpha.last_run.as_ref().unwrap().error.as_deref().unwrap().contains("the dumper died"));
assert_eq!(status.members[2].line(), "gamma: no clone yet");
assert_eq!(status.behind().iter().map(|m| m.member.as_str()).collect::<Vec<_>>(), vec!["alpha"], "a member with no clone is not `behind`: there is nothing to consume");
// ---- the next commit fixes the tree but does not COMPILE: recorded, offset advances
a.commit("fn a1\nBROKEN\n");
let members = [a.member("alpha"), b.member("beta"), gamma];
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let m = &rep.members[0];
assert_eq!((m.status, m.reports[0].status.as_str(), m.partial()), (MemberStatus::CaughtUp, "error", 1));
assert!(m.line().contains("1 partial/error"), "{}", m.line());
let status = room.status(&members).unwrap();
assert_eq!((status.members[0].partial, status.members[0].behind), (2, false));
}
/// A member whose tree names a sibling by path is dumped with that sibling's
/// clone in place — also on the FIRST run, and also when the sibling comes
/// later in the member order. Seen on a real server 2026-10-13: 17 members,
/// clones made one per turn, so `edda` was dumped before `znippy-zoomies`
/// existed and every member with a `../sibling` dependency failed in 35 ms.
#[test]
fn every_members_clone_is_there_before_any_member_is_dumped() {
let tmp = tempfile::tempdir().unwrap();
let (a, b) = (Repo::new(tmp.path(), "ws", "alpha"), Repo::new(tmp.path(), "ws", "beta"));
a.commit("fn a1\nneeds beta\n");
b.commit("fn b1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha"), b.member("beta")];
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let st: Vec<_> = rep.members.iter().map(|m| (m.member.as_str(), m.status, m.error.clone())).collect();
assert_eq!(st, vec![("alpha", MemberStatus::CaughtUp, None), ("beta", MemberStatus::CaughtUp, None)], "alpha is first in order and needs beta's clone");
assert_eq!(rep.stamp(), "alpha@1,beta@1");
// asking for ONE member still puts its siblings in place
let tmp = tempfile::tempdir().unwrap();
let (a, b) = (Repo::new(tmp.path(), "ws", "alpha"), Repo::new(tmp.path(), "ws", "beta"));
a.commit("fn a1\nneeds beta\n");
b.commit("fn b1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let members = [a.member("alpha"), b.member("beta")];
let rep = room.consume(&TreeDumpers::default(), &[], &Request { members: &members, only: Some("alpha"), ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((rep.members.len(), rep.members[0].status), (1, MemberStatus::CaughtUp), "{:?}", rep.members[0].error);
assert!(room.layout().clone_of("beta").join(".git").exists());
}
/// A feed that counts its runs and answers as told.
struct Counting {
name: &'static str,
runs: AtomicUsize,
answer: fn(usize) -> anyhow::Result<FeedResult>,
}
impl Feed for Counting {
fn name(&self) -> &'static str {
self.name
}
fn run(&self, at: &FeedAt<'_>) -> anyhow::Result<FeedResult> {
assert_eq!(crate::gitio::head_sha(&at.member.source).unwrap(), at.sha, "a feed reads the server's clone AT the offset");
(self.answer)(self.runs.fetch_add(1, Ordering::SeqCst) + 1)
}
}
#[test]
fn feeds_ride_the_offsets_once_per_sha_with_their_own_ledger_rows() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let heat = Counting { name: "git_heat", runs: AtomicUsize::new(0), answer: |n| Ok(FeedResult::Ok { rows: 10 * n as i64, detail: format!("run {n}") }) };
let embed = Counting { name: "embeddings", runs: AtomicUsize::new(0), answer: |_| Ok(FeedResult::Skipped("no embedder compiled".into())) };
let sbom = Counting { name: "sbom", runs: AtomicUsize::new(0), answer: |n| if n == 1 { anyhow::bail!("cargo metadata failed") } else { Ok(FeedResult::Ok { rows: 7, detail: String::new() }) } };
let feeds: [&dyn Feed; 3] = [&heat, &embed, &sbom];
let members = [a.member("alpha")];
let runs = |f: &Counting| f.runs.load(Ordering::SeqCst);
// ---- the first run: commit, then every feed for that (repo, sha)
let mut events = Vec::new();
let rep = room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut recorder(&mut events)).unwrap();
assert_eq!(
events,
vec!["member alpha behind=1", "begin alpha@1 behind=0", "commit alpha@1 ok", "feed alpha git_heat ok", "feed alpha embeddings skipped", "feed alpha sbom error", "done alpha caught_up"]
);
let sha = rep.members[0].offset.as_ref().unwrap().sha.clone();
let rows: Vec<_> = rep.members[0].feeds.iter().map(|r| (r.feed.as_str(), r.seq, r.sha.as_str(), r.status.as_str(), r.rows, r.detail.as_deref())).collect();
assert_eq!(
rows,
vec![
("git_heat", 1, sha.as_str(), "ok", 10, Some("run 1")),
("embeddings", 1, sha.as_str(), "skipped", 0, Some("no embedder compiled")),
("sbom", 1, sha.as_str(), "error", 0, Some("cargo metadata failed")),
]
);
// ---- every feed was TRIED for this sha: the member is not behind, and an
// unattended run does not try the skipped / failed ones again
assert!(room.behind(&members, &feeds).unwrap().is_empty());
let rep = room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert!(rep.members[0].feeds.is_empty());
assert_eq!((runs(&heat), runs(&embed), runs(&sbom)), (1, 1, 1));
// ---- an explicit refresh retries what is not `ok` — and only that
let rep = room.consume(&dumpers, &feeds, &Request { members: &members, retry_feeds: true, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!(rep.members[0].feeds.iter().map(|r| (r.feed.as_str(), r.status.as_str())).collect::<Vec<_>>(), vec![("embeddings", "skipped"), ("sbom", "ok")]);
assert_eq!((runs(&heat), runs(&embed), runs(&sbom)), (1, 2, 2));
// ---- a feed added later is behind for the sha the others already did
let late = Counting { name: "code_index", runs: AtomicUsize::new(0), answer: |_| Ok(FeedResult::Ok { rows: 3, detail: String::new() }) };
let more: [&dyn Feed; 4] = [&heat, &embed, &sbom, &late];
assert_eq!(room.behind(&members, &more).unwrap(), vec!["alpha"]);
room.consume(&dumpers, &more, &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((runs(&heat), runs(&late)), (1, 1));
// ---- a new commit: every feed runs for the new sha
a.commit("fn a1\nfn a2\n");
let members = [a.member("alpha")];
let rep = room.consume(&dumpers, &more, &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!(rep.members[0].feeds.iter().map(|r| (r.feed.as_str(), r.seq)).collect::<Vec<_>>(), vec![("git_heat", 2), ("embeddings", 2), ("sbom", 2), ("code_index", 2)]);
assert_eq!(runs(&heat), 2);
// ---- the ledger, through status: each feed's last row
let st = room.status(&members).unwrap();
let feeds: Vec<_> = st.members[0].feeds.values().map(|r| (r.feed.as_str(), r.seq, r.status.as_str())).collect();
assert_eq!(feeds, vec![("code_index", 2, "ok"), ("embeddings", 2, "skipped"), ("git_heat", 2, "ok"), ("sbom", 2, "ok")]);
// ---- a bounded run that leaves the member behind runs no feed: the
// server's clone is not at the offset
a.commit("fn a1\nfn a2\nfn a3\n");
a.commit("fn a1\nfn a2\nfn a3\nfn a4\n");
let members = [a.member("alpha")];
let rep = room.consume(&dumpers, &more, &Request { members: &members, max_commits: Some(1), ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((rep.members[0].status, rep.members[0].feeds.len(), runs(&heat)), (MemberStatus::Behind, 0, 2));
}
/// The dry-run delta through the handle (I3): a working copy that is not a
/// commit, diffed against the member's last release — nothing written, a
/// member that was never consumed refused by state.
#[test]
fn a_dry_run_diffs_an_uncommitted_tree_against_the_last_release_and_writes_nothing() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\nfn a2\ncall a2 a1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let members = [a.member("alpha")];
let dumpers = TreeDumpers::default();
let dump_of = |dir: &Path| {
let mut d = dumpers.dumper(room.layout(), &members[0]).unwrap();
let dir = dir.to_path_buf();
move || d.dump(&dir, "worktree")
};
// never consumed: there is nothing to diff against
let r = room.dry_delta_of("alpha", dump_of(&a.dir)).err().expect("never consumed");
assert_eq!(r.state, State::NotConsumed, "{r}");
room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let before = room.status(&members).unwrap();
// the developer's tree: edited, NOT committed
std::fs::write(a.dir.join("src/lib.rs"), "fn a1\nfn a3\ncall a3 a1\n").unwrap();
let d = room.dry_delta_of("alpha", dump_of(&a.dir)).unwrap();
assert_eq!((d.nodes.added.clone(), d.nodes.changed.len(), d.nodes.removed.clone()), (vec!["alpha a3".to_string()], 0, vec!["alpha a2".to_string()]));
assert_eq!((d.edges.added.len(), d.edges.removed.len()), (1, 1));
assert_eq!(super::DryDelta::edge_parts(&d.edges.added[0]), Some(("alpha a3", "alpha a1", "CALLS")));
assert_eq!(d.touched_nodes(), vec!["alpha a3", "alpha a2"]);
// nothing moved: same offset, same ledger, the room is still the release
assert_eq!(room.status(&members).unwrap(), before);
let r = room.room(&Scope::current()).unwrap();
assert!(r.ordinal("alpha a2").is_some() && r.ordinal("alpha a3").is_none());
// and the dry state has a place of its own in the layout
assert_eq!(room.layout().dry_of("alpha"), tmp.path().join("ws/builds/mimir_v2/dry/alpha"));
// a tree that cannot be dumped is an error with the reason, not an empty delta
std::fs::write(a.dir.join("src/lib.rs"), "CRASH\n").unwrap();
let r = room.dry_delta_of("alpha", dump_of(&a.dir)).err().expect("the dump failed");
assert_eq!(r.state, State::Internal);
assert!(r.note.contains("dump the tree") && r.note.contains("the dumper died"), "{r}");
}
#[test]
fn no_driver_is_a_typed_refusal_said_once_and_shown_in_status() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let members = [a.member("alpha")];
let r = room.consume(&NoDriver, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap_err();
assert_eq!(r.state, State::DriverNotAvailable);
assert_eq!(room.last_refusal().as_ref(), Some(&r));
assert!(!room.note_refusal(&r), "the same refusal again is not news");
assert!(room.note_refusal(&Refusal::new(State::DriverNotAvailable, "another reason")));
room.note_refusal(&r);
// nothing was consumed, and the status says why
let st = room.status(&members).unwrap();
assert_eq!((st.members[0].offset.is_none(), st.members[0].runs, st.members[0].behind), (true, 0, true));
assert_eq!(st.refusal.as_ref().unwrap().state, State::DriverNotAvailable);
assert_eq!(st.to_json()["refusal"]["state"], "driver_not_available");
assert_eq!(st.members[0].line(), format!("alpha: never consumed → head {}", &members[0].head[..10]));
// a reader of that repo gets `not_consumed`, not an empty room
assert_eq!(room.room(&Scope::repo("alpha", None)).err().expect("never consumed").state, State::NotConsumed);
// the driver arrives: the run goes through and the refusal is gone
room.consume(&TreeDumpers::default(), &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert!(room.last_refusal().is_none() && room.status(&members).unwrap().refusal.is_none());
}
/// The server's shape: a consume in flight on one thread, status and rooms
/// from another — and a cancel that lets the commit in flight land.
#[test]
fn status_and_rooms_answer_during_a_consume_and_cancel_stops_after_the_commit_in_flight() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
let first = a.commit("fn a1\n");
a.commit("fn a1\nfn a2\n");
a.commit("fn a1\nfn a2\nfn a3\n");
a.commit("fn a1\nfn a2\nfn a3\nfn a4\n");
let room = Arc::new(CodeRoom::new("ws", &tmp.path().join("ws/builds")));
let members = vec![a.member("alpha")];
let (entered_tx, entered) = channel();
let (go, go_rx) = channel();
let gate = Arc::new(Gate { hold_at: 3, entered: Mutex::new(entered_tx), go: Mutex::new(go_rx) });
let (r2, m2) = (room.clone(), members.clone());
let run = std::thread::spawn(move || {
let dumpers = TreeDumpers { gate: Some(("alpha".into(), gate)), ..Default::default() };
r2.consume(&dumpers, &[], &Request { members: &m2, from: Some(&first), ..Default::default() }, &mut quiet())
});
entered.recv_timeout(WAIT).expect("the consume never reached commit 3");
// ---- in flight: commit 3 is being dumped, 1 more behind it
let st = room.status(&members).unwrap();
let inf = st.in_flight.as_ref().expect("a consume is in flight");
assert_eq!((inf.member.as_str(), inf.commit.as_ref().unwrap().0, inf.behind), ("alpha", 3, 1));
assert!(inf.line().starts_with("alpha seq 3 @ "), "{}", inf.line());
assert_eq!((st.members[0].offset.as_ref().unwrap().seq, st.members[0].runs, st.members[0].behind), (2, 2, true), "the last committed release");
// ---- a room opens NOW: release 2
let r = room.room(&Scope::current()).unwrap();
assert_eq!((r.scope().repos["alpha"], r.len()), (2, 2));
// ---- cancel: the commit in flight lands whole, the next is not begun
room.cancel_consume();
go.send(()).unwrap();
let rep = run.join().unwrap().unwrap();
let m = &rep.members[0];
assert_eq!((m.status, m.reports.len(), m.behind, m.offset.as_ref().unwrap().seq), (MemberStatus::Cancelled, 3, 1, 3));
assert!(room.in_flight().is_none());
assert_eq!(room.room(&Scope::current()).unwrap().len(), 3);
// ---- the next run starts un-cancelled and finishes the member
let rep = room.consume(&TreeDumpers::default(), &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((rep.members[0].status, rep.members[0].offset.as_ref().unwrap().seq), (MemberStatus::CaughtUp, 4));
}
// ---------------------------------------------------------------- backfill
/// One turn of a backfill's job, as the server's lane takes it: ask the lake
/// what is next, consume ONE commit of the member. `None`: the walk is over.
fn backfill_turn(room: &CodeRoom, dumpers: &TreeDumpers, members: &[Member], member: &str) -> Option<String> {
use super::backfill::Next;
let m = members.iter().find(|m| m.name == member).unwrap();
match room.backfill_next(m).unwrap() {
Next::Step { from, .. } => {
let req = Request { members, max_commits: Some(1), only: Some(member), from: from.as_deref(), ..Default::default() };
let rep = room.consume(dumpers, &[], &req, &mut quiet()).unwrap();
assert_eq!(rep.members[0].reports.len(), 1, "one commit per turn: {}", rep.members[0].line());
Some(rep.members[0].reports[0].sha.clone())
}
Next::Finished(_) | Next::Nothing => None,
}
}
/// A member's history asked for from a commit is ONE row in the lake; its job
/// walks it one commit per run; the member is left out of every other run
/// meanwhile; a handle opened on the same lake after a "restart" knows where
/// the walk is; a commit pushed during the walk is its last commit, in order.
///
/// RED WHEN REVERTED: drop the `Backfilling` arm of `consume_locked` and the
/// ordinary run in the middle takes the member (`status` is not
/// `Backfilling`, its dump count moves).
#[test]
fn a_backfill_is_a_request_in_the_lake_walked_one_commit_per_turn_and_resumed_from_the_lake() {
let tmp = tempfile::tempdir().unwrap();
let (a, b) = (Repo::new(tmp.path(), "ws", "alpha"), Repo::new(tmp.path(), "ws", "beta"));
let c1 = a.commit("fn a1\n");
let c2 = a.commit("fn a1\nfn a2\n");
let c3 = a.commit("fn a1\nfn a2\nfn a3\n");
let c4 = a.commit("fn a1\nfn a2\nfn a3\nfn a4\n");
b.commit("fn b1\n");
let builds = tmp.path().join("ws/builds");
let room = CodeRoom::new("ws", &builds);
let dumpers = TreeDumpers::default();
let members = [a.member("alpha"), b.member("beta")];
fn all(members: &[Member]) -> Request<'_> {
Request { members, ..Default::default() }
}
room.consume(&dumpers, &[], &all(&members), &mut quiet()).unwrap();
assert!(room.backfills(&members).unwrap().is_empty(), "nobody asked for history");
// ---- the request: resolved on the branch, written, NOTHING consumed
let asked = room.request_backfill(&members[0], &c2[..10]).unwrap();
assert_eq!((asked.from_sha.as_str(), asked.is_open(), asked.started, asked.total, asked.done), (c2.as_str(), true, false, 3, 0));
assert!(asked.line().contains("queued, 3 commit(s) to walk"), "{}", asked.line());
assert_eq!(dumpers.dumps("alpha"), 1, "asking consumes nothing");
// one at a time per member: the second is refused and says where the first is
let r = room.request_backfill(&members[0], &c1[..10]).unwrap_err();
assert_eq!(r.state, State::InvalidArgument, "{r}");
assert!(r.note.contains("already open") && r.note.contains("backfill from"), "{r}");
// a `from` that names no commit of the branch: refused, nothing written
let r = room.request_backfill(&members[1], "2999-01-01").unwrap_err();
assert_eq!(r.state, State::InvalidArgument, "{r}");
assert_eq!(room.backfills(&members).unwrap().iter().map(|x| x.member.as_str()).collect::<Vec<_>>(), vec!["alpha"]);
// ---- the member is its job's: not behind, and no other run touches it
assert!(room.behind(&members, &[]).unwrap().is_empty());
let rep = room.consume(&dumpers, &[], &all(&members), &mut quiet()).unwrap();
assert_eq!((rep.members[0].status, rep.members[0].status.as_str(), rep.members[1].status), (MemberStatus::Backfilling, "backfilling", MemberStatus::CaughtUp));
assert_eq!(rep.stamp(), "alpha@1,beta@1", "its offset is still reported");
assert_eq!(dumpers.dumps("alpha"), 1);
// ---- turn 1: the replay starts at the commit asked for
assert_eq!(backfill_turn(&room, &dumpers, &members, "alpha").as_deref(), Some(c2.as_str()));
let st = room.backfills(&members).unwrap().remove(0);
assert_eq!((st.started, st.done, st.total, st.at.as_ref().unwrap().seq), (true, 1, 3, 2));
assert!(st.line().contains("1 of 3 commit(s), at seq 2 @ "), "{}", st.line());
assert_eq!(st.to_json()["status"], "running");
// ---- meanwhile beta's tip moves and alpha is pushed to. An ordinary run
// between two turns consumes beta's tip — the walk holds nothing between
// its commits — and leaves alpha alone
b.commit("fn b1\nfn b2\n");
let c5 = a.commit("fn a1\nfn a2\nfn a3\nfn a4\nfn a5\n");
let members = [a.member("alpha"), b.member("beta")];
assert_eq!(room.behind(&members, &[]).unwrap(), vec!["beta"]);
let rep = room.consume(&dumpers, &[], &all(&members), &mut quiet()).unwrap();
assert_eq!((rep.members[0].status, rep.members[1].status, rep.members[1].reports.len()), (MemberStatus::Backfilling, MemberStatus::CaughtUp, 1));
assert_eq!(dumpers.dumps("alpha"), 2, "the first tip and ONE turn");
// ---- a restart in the middle: a new handle on the same lake reads where
// the walk is — there is no other state
room.close();
drop(room);
let room = CodeRoom::new("ws", &builds);
let dumpers = TreeDumpers::default();
let st = room.backfills(&members).unwrap().remove(0);
assert_eq!((st.is_open(), st.started, st.done, st.from_sha.as_str(), st.at.as_ref().unwrap().sha.as_str()), (true, true, 1, c2.as_str(), c2.as_str()));
// ---- the rest of the walk: first-parent order, the pushed commit last
let mut walked = Vec::new();
while let Some(sha) = backfill_turn(&room, &dumpers, &members, "alpha") {
walked.push(sha);
}
assert_eq!(walked, vec![c3, c4, c5]);
let st = room.backfills(&members).unwrap().remove(0);
assert_eq!((st.status.as_str(), st.is_open(), st.total), ("done", false, 4));
assert!(st.line().contains("done (4 commit(s))"), "{}", st.line());
assert!(room.behind(&members, &[]).unwrap().is_empty());
assert!(backfill_turn(&room, &dumpers, &members, "alpha").is_none(), "nothing is open: nothing to do");
// ---- the ordinals of the walk ARE the history: 1 = the first tip, then
// c2 … c5; a room at an ordinal is that commit's tree
let status = room.status(&members).unwrap();
assert_eq!((status.members[0].offset.as_ref().unwrap().seq, status.members[0].runs), (5, 5));
assert_eq!(room.room(&Scope::repo("alpha", Some(2))).unwrap().len(), 2, "c2: a1 a2");
assert_eq!(room.room(&Scope::repo("alpha", Some(1))).unwrap().len(), 4, "the first tip: a1 … a4");
assert_eq!(room.room(&Scope::repo("alpha", None)).unwrap().len(), 5);
// ---- and the member is everybody's again
a.commit("fn a1\n");
let members = [a.member("alpha"), b.member("beta")];
assert_eq!(room.behind(&members, &[]).unwrap(), vec!["alpha"]);
let rep = room.consume(&dumpers, &[], &all(&members), &mut quiet()).unwrap();
assert_eq!((rep.members[0].status, rep.members[0].offset.as_ref().unwrap().seq), (MemberStatus::CaughtUp, 6));
}
// ---------------------------------------------------------- deferred feeds
/// A feed whose work is long (embeddings on a CPU): it tells the test when
/// the work starts and waits for the test to let it end. `deferred = false`
/// is the feed as it was before: all of it inside the run.
struct Slow {
deferred: bool,
prepared: AtomicUsize,
started: Mutex<Sender<String>>,
go: Arc<Mutex<Receiver<()>>>,
}
impl Slow {
fn new(deferred: bool) -> (Arc<Self>, Receiver<String>, Sender<()>) {
let (started_tx, started) = channel();
let (go, go_rx) = channel();
(Arc::new(Self { deferred, prepared: AtomicUsize::new(0), started: Mutex::new(started_tx), go: Arc::new(Mutex::new(go_rx)) }), started, go)
}
}
impl Feed for Slow {
fn name(&self) -> &'static str {
"embeddings"
}
fn run(&self, at: &FeedAt<'_>) -> anyhow::Result<FeedResult> {
self.started.lock().unwrap().send(at.sha.to_string()).unwrap();
self.go.lock().unwrap().recv_timeout(WAIT).expect("the test never let the feed end");
Ok(FeedResult::Ok { rows: 30, detail: "30 chunk(s), inside the run".into() })
}
fn prepare(&self, at: &FeedAt<'_>) -> Option<anyhow::Result<DeferredWork>> {
if !self.deferred {
return None;
}
// the read half: the tree IS the commit here
assert_eq!(crate::gitio::head_sha(&at.member.source).unwrap(), at.sha);
self.prepared.fetch_add(1, Ordering::SeqCst);
let (started, go, sha) = (self.started.lock().unwrap().clone(), self.go.clone(), at.sha.to_string());
Some(Ok(Box::new(move || {
started.send(sha).unwrap();
go.lock().unwrap().recv_timeout(WAIT).expect("the test never let the work end");
Ok(FeedResult::Ok { rows: 30, detail: "30 chunk(s)".into() })
})))
}
}
/// **A commit never waits for a deferred feed.** The feed's long half runs
/// with no lock of the code room held: while it is running, the member's next
/// commit is consumed, and the feed's row lands when its work is done.
///
/// The RED PROBE runs first, in the same test: the SAME feed with its work
/// inside the run (what `embeddings` was) — the next commit's consume does not
/// finish while the feed runs.
#[test]
fn a_commit_never_waits_for_a_deferred_feed_and_the_probe_without_deferral_does() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\n");
let room = Arc::new(CodeRoom::new("ws", &tmp.path().join("ws/builds")));
let consume = |room: &Arc<CodeRoom>, feed: &Arc<Slow>, members: Vec<Member>| {
let (room, feed) = (room.clone(), feed.clone());
let (done_tx, done) = channel();
let t = std::thread::spawn(move || {
let feeds: [&dyn Feed; 1] = [&*feed];
let mut handed: Vec<Deferred> = Vec::new();
let rep = room
.consume(&TreeDumpers::default(), &feeds, &Request { members: &members, ..Default::default() }, &mut |e| {
if let Event::Deferred(d) = e {
handed.push(d)
}
})
.unwrap();
let _ = done_tx.send(()); // a caller that only joins has dropped its end
(rep, handed)
});
(t, done)
};
// ---- RED PROBE: the feed's work inside the run holds the consumer
let (inline, started, go) = Slow::new(false);
let (first, _first_done) = consume(&room, &inline, vec![a.member("alpha")]);
let sha1 = started.recv_timeout(WAIT).expect("the feed never started");
a.commit("fn a1\nfn a2\n");
let (second, second_done) = consume(&room, &inline, vec![a.member("alpha")]);
assert!(
second_done.recv_timeout(Duration::from_millis(400)).is_err(),
"the probe did not block: the next commit was consumed while the feed ran INSIDE the run, so a green below would prove nothing"
);
go.send(()).unwrap(); // the first run's feed ends …
let (rep, _) = first.join().unwrap();
assert_eq!((rep.members[0].feeds[0].sha.as_str(), rep.members[0].feeds[0].detail.as_deref()), (sha1.as_str(), Some("30 chunk(s), inside the run")));
started.recv_timeout(WAIT).expect("the second run's feed never started"); // … only now the commit behind it was consumed
go.send(()).unwrap();
assert_eq!(second.join().unwrap().0.members[0].reports.len(), 1);
// ---- the deferred feed: the run hands the work out and returns
let (slow, started, go) = Slow::new(true);
let sha3 = a.commit("fn a1\nfn a2\nfn a3\n");
let members = vec![a.member("alpha")];
let feeds: [&dyn Feed; 1] = [&*slow];
let (run, _) = consume(&room, &slow, members.clone());
let (rep, mut handed) = run.join().unwrap();
assert_eq!((rep.members[0].reports.len(), rep.members[0].feeds.len(), handed.len()), (1, 0, 1), "consumed; no ledger row yet; the work is handed out");
assert_eq!((handed[0].member.as_str(), handed[0].feed, handed[0].seq, handed[0].sha.as_str()), ("alpha", "embeddings", 3, sha3.as_str()));
// in flight is not behind, and it is not read a second time
assert!(room.behind(&members, &feeds).unwrap().is_empty());
let (again, _) = consume(&room, &slow, members.clone());
assert!(again.join().unwrap().1.is_empty());
assert_eq!(slow.prepared.load(Ordering::SeqCst), 1);
// ---- its long half runs elsewhere …
let (r2, work) = (room.clone(), handed.pop().unwrap());
let embed = std::thread::spawn(move || r2.run_deferred(work));
assert_eq!(started.recv_timeout(WAIT).expect("the work never started"), sha3);
// … and while it runs, the tip moves and IS consumed
let sha4 = a.commit("fn a1\nfn a2\nfn a3\nfn a4\n");
let members = vec![a.member("alpha")];
let (run, done) = consume(&room, &slow, members.clone());
done.recv_timeout(WAIT).expect("the commit waited for the feed's work");
let (rep, mut handed) = run.join().unwrap();
assert!(!embed.is_finished(), "the work of the commit before is still running");
assert_eq!((rep.members[0].reports.len(), rep.members[0].offset.as_ref().unwrap().seq, handed.len()), (1, 4, 1), "the new commit's work is handed out too");
assert_eq!(room.room(&Scope::current()).unwrap().len(), 4);
// ---- the work ends: its row lands for the commit it read
go.send(()).unwrap();
let row = embed.join().unwrap().unwrap();
assert_eq!((row.feed.as_str(), row.seq, row.sha.as_str(), row.status.as_str(), row.rows, row.detail.as_deref()), ("embeddings", 3, sha3.as_str(), "ok", 30, Some("30 chunk(s)")));
// the new commit's work is still out: in flight, not behind
assert!(room.behind(&members, &feeds).unwrap().is_empty());
go.send(()).unwrap();
let row = room.run_deferred(handed.pop().unwrap()).unwrap();
assert_eq!((row.seq, row.sha.as_str(), row.status.as_str()), (4, sha4.as_str(), "ok"));
assert!(room.behind(&members, &feeds).unwrap().is_empty());
let st = room.status(&members).unwrap();
assert_eq!(st.members[0].feeds["embeddings"].seq, 4);
// ---- work given up (its unit was superseded): the feed is due again
a.commit("fn a1\n");
let members = vec![a.member("alpha")];
let (run, _) = consume(&room, &slow, members.clone());
let (_, handed) = run.join().unwrap();
assert!(room.behind(&members, &feeds).unwrap().is_empty());
room.drop_deferred(&handed[0]);
assert_eq!(room.behind(&members, &feeds).unwrap(), vec!["alpha"]);
}
// ------------------------------------------------------------ binary hook
/// A feed that found one linked artifact of the member: `alpha` links `a1`
/// (64 bytes) — the rows mimir-binary makes, handed to the consumer.
struct FoundBinary;
fn binary_of(member: &str, item: &str, bytes: u64) -> BinaryDump {
let symbol = mimir_schema::SymbolRow {
id: format!("{member} {item}"),
symbol: format!("_RNvCs1234_5{member}2{item}"),
demangled: Some(format!("{member}::{item}")),
role: mimir_schema::role::ITEM.into(),
krate: member.into(),
version: "0.1.0".into(),
binary: Some(format!("{member}-cli")),
section: Some(".text".into()),
addr: Some(0x1000),
size: Some(bytes),
provenance: mimir_schema::Provenance::Both,
valid_from_sha: None,
valid_to_sha: None,
};
BinaryDump {
artifact: BinaryDump::artifact_id(&format!("{member}-cli"), "x86_64-unknown-linux-gnu", "release"),
nodes: vec![mimir_schema::nodes_batch(&[]).unwrap()],
edges: vec![mimir_schema::edges_batch(&[]).unwrap()],
symbols: vec![mimir_schema::symbols_batch(&[symbol]).unwrap()],
ledger: Some("{}".into()),
ms: 1,
}
}
impl Feed for FoundBinary {
fn name(&self) -> &'static str {
super::feeds::Binary::NAME
}
fn run(&self, at: &FeedAt<'_>) -> anyhow::Result<FeedResult> {
Ok(FeedResult::Binaries {
dumps: vec![binary_of(&at.member.name, "a1", 64)],
declared_bins: [format!("{}_cli", at.member.name)].into(),
detail: "1 artifact(s)".into(),
})
}
fn retry(&self, _: &FeedAt<'_>, _: &mimir_ingest::FeedRow) -> bool {
true
}
}
/// The binary hook's two answers through the consumer: artifact rows a feed
/// found are written by the consumer's seat as the BINARY stream of the
/// member's commit (a room then has linked symbols); and the real hook, with
/// no artifact where it looks, writes the typed `no_artifact` row naming the
/// directory — and is asked again only once something is there.
///
/// RED WHEN REVERTED: make the `Binaries` arm of `run_feeds` an ordinary row
/// (drop the `ingest_binaries` call) and the room holds no symbol.
#[test]
fn the_binary_hook_writes_the_binary_stream_or_says_no_artifact_and_where_it_looked() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
let sha = a.commit("fn a1\nfn a2\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha")];
// ---- nothing was built for this commit: `no_artifact`, with the path
let hook = super::feeds::Binary;
let feeds: [&dyn Feed; 1] = [&hook];
let rep = room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let row = &rep.members[0].feeds[0];
let dir = room.layout().artifacts_of("alpha", &sha);
assert_eq!((row.feed.as_str(), row.status.as_str(), row.rows), ("binary", "no_artifact", 0));
assert!(row.detail.as_deref().unwrap().contains(&dir.display().to_string()), "{:?}", row.detail);
assert!(row.detail.as_deref().unwrap().contains("nothing here builds into it"), "{:?}", row.detail);
assert!(!dir.exists(), "the hook reads: it creates nothing");
// tried for this sha: not behind, and not run again by the next run
assert!(room.behind(&members, &feeds).unwrap().is_empty());
assert!(room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap().members[0].feeds.is_empty());
assert!(room.room(&Scope::current()).unwrap().symbols(None, None).unwrap().value.is_empty());
// ---- something appears there: the hook is due again for the SAME commit
std::fs::create_dir_all(dir.join("release")).unwrap();
assert_eq!(room.behind(&members, &feeds).unwrap(), vec!["alpha"]);
// ---- artifacts found: the consumer writes the binary stream
let found = FoundBinary;
let feeds: [&dyn Feed; 1] = [&found];
let rep = room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let row = &rep.members[0].feeds[0];
assert_eq!((row.feed.as_str(), row.status.as_str(), row.rows, rep.members[0].reports.len()), ("binary", "ok", 1, 0), "{:?}", row.detail);
assert!(row.detail.as_deref().unwrap().contains("alpha-cli@x86_64-unknown-linux-gnu:release ok +1 −0"), "{:?}", row.detail);
let r = room.room(&Scope::current()).unwrap();
let symbols = r.symbols(None, None).unwrap().value;
assert_eq!(symbols.iter().map(|s| (s.artifact, s.size, r.moniker(s.node))).collect::<Vec<_>>(), vec![("alpha-cli@x86_64-unknown-linux-gnu:release", Some(64), "alpha a1")]);
assert_eq!(r.linked_bytes(r.ordinal("alpha a1").unwrap()), 64);
assert_eq!(r.linked_bytes(r.ordinal("alpha a2").unwrap()), 0, "a source item nothing links is a fact, not an error");
// the source stream did not move: the same ordinal, one binary release
let st = room.status(&members).unwrap();
assert_eq!((st.members[0].offset.as_ref().unwrap().seq, st.members[0].runs), (1, 1));
let log = room.log().unwrap();
let bin = log.block_on(log.bin_runs("alpha")).unwrap();
assert_eq!(bin.iter().map(|b| (b.seq, b.rev, b.sha.as_str(), b.status.as_str())).collect::<Vec<_>>(), vec![(1, 1, sha.as_str(), "ok")]);
}
// ------------------------------------------------------- the toolchain's std
/// A stand-in for `cargo-mimir-dump --std`: a script that copies a made-up
/// dump of two toolchain crates into `--out` and counts its runs. `ok = false`
/// exits 1 instead.
fn fake_std_tool(dir: &Path, ok: bool) -> (PathBuf, PathBuf) {
use std::os::unix::fs::PermissionsExt;
let fixture = dir.join("fixture");
std::fs::create_dir_all(&fixture).unwrap();
let write = |p: PathBuf, b: &arrow::record_batch::RecordBatch| {
let mut w = arrow::ipc::writer::FileWriter::try_new(std::fs::File::create(p).unwrap(), &b.schema()).unwrap();
w.write(b).unwrap();
w.finish().unwrap();
};
for krate in ["std", "core"] {
let version = mimir_schema::moniker::lang_version(krate);
let stem = format!("{krate}@lang-0000");
let mut row = NodeRow::new(mimir_schema::moniker::moniker(krate, &version, "write()."), "fn", Level::Item, "write", krate, version, Origin::Lang);
row.defined_here = true;
write(fixture.join(format!("{stem}.nodes.arrow")), &mimir_schema::nodes_batch(&[row]).unwrap());
write(fixture.join(format!("{stem}.edges.arrow")), &mimir_schema::edges_batch(&[]).unwrap());
write(fixture.join(format!("{stem}.spans.arrow")), &mimir_schema::spans_batch(&[]).unwrap());
std::fs::write(fixture.join(format!("{stem}.stats.json")), format!(r#"{{"crate_name":"{krate}","origin":"lang","rustc_release":"1.101.0-nightly"}}"#)).unwrap();
}
let (tool, runs) = (dir.join("cargo-mimir-dump"), dir.join("runs"));
let body = if ok { format!("cp {}/* \"$out\"/", fixture.display()) } else { "echo 'error: the toolchain has no rust-src' >&2; exit 1".to_string() };
std::fs::write(
&tool,
format!("#!/bin/sh\nout=\nwhile [ $# -gt 0 ]; do [ \"$1\" = --out ] && out=\"$2\"; shift; done\necho run >> {}\n{body}\n", runs.display()),
)
.unwrap();
std::fs::set_permissions(&tool, std::fs::Permissions::from_mode(0o755)).unwrap();
(tool, runs)
}
fn lines(p: &Path) -> usize {
std::fs::read_to_string(p).map(|s| s.lines().count()).unwrap_or(0)
}
/// The first consume of a workspace brings the toolchain's std into the
/// universe ring — before the members, once per toolchain per lake, said as it
/// happens — and a dump that fails is said, does not stop the members, and is
/// not tried again on every run.
///
/// RED WHEN REVERTED: drop the `ensure_std` call of `consume_locked` and the
/// manifest has no `std@lang:lib` (and no `std` events).
#[test]
fn the_first_consume_brings_the_toolchains_std_into_the_ring_once() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\n");
let builds = tmp.path().join("ws/builds");
let room = CodeRoom::new("ws", &builds);
let (tool, runs) = fake_std_tool(&tmp.path().join("tool"), true);
let std_of = |toolchain: &str, tool: &Path| StdDumper { cargo_mimir_dump: tool.to_path_buf(), driver: None, toolchain: toolchain.into(), root: room.layout().std_root(), host: Default::default() };
let dumpers = TreeDumpers { std: Some(std_of("nightly-test", &tool)), ..Default::default() };
let members = [a.member("alpha")];
assert!(!room.std_due(&dumpers), "nothing consumed yet: no ring to fill");
// ---- the first run: std first, then the member
let mut events = Vec::new();
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut recorder(&mut events)).unwrap();
assert_eq!(events, vec!["std-begin nightly-test", "std nightly-test ok", "member alpha behind=1", "begin alpha@1 behind=0", "commit alpha@1 ok", "done alpha caught_up"]);
assert!(rep.std.as_deref().unwrap().starts_with("std of nightly-test: 2 crate(s), 2 row(s) into the ring"), "{:?}", rep.std);
assert_eq!(lines(&runs), 1);
let log = room.log().unwrap();
let manifest = log.block_on(log.universe_rows()).unwrap();
assert_eq!(manifest.iter().map(|m| (m.0.as_str(), m.1.as_str())).collect::<Vec<_>>(), vec![("core@lang:lib", "nightly-test"), ("std@lang:lib", "nightly-test")]);
assert!(room.std_settled() && !room.std_due(&dumpers));
// … as universe events, under no repo's ordinals
let ring = log.block_on(log.fold_rows(mimir_ingest::Kind::Nodes, &mimir_ingest::FoldScope { repos: Default::default(), universe: true }, &[])).unwrap();
assert_eq!(ring.open.len(), 2);
// ---- once: not on the next run, and not after a restart (the lake says so)
a.commit("fn a1\nfn a2\n");
let members = [a.member("alpha")];
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert!(rep.std.is_none());
drop(log);
room.close();
drop(room);
let room = CodeRoom::new("ws", &builds);
a.commit("fn a1\n");
let members = [a.member("alpha")];
let mut events = Vec::new();
room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut recorder(&mut events)).unwrap();
assert!(!events.iter().any(|e| e.starts_with("std")), "{events:?}");
assert_eq!(lines(&runs), 1, "one dump per toolchain per lake");
// ---- another toolchain whose dump FAILS: said, the member still consumed,
// not tried again by the next run — an explicit refresh tries again
let (bad, bad_runs) = fake_std_tool(&tmp.path().join("bad"), false);
let std_of = |toolchain: &str, tool: &Path| StdDumper { cargo_mimir_dump: tool.to_path_buf(), driver: None, toolchain: toolchain.into(), root: room.layout().std_root(), host: Default::default() };
let other = TreeDumpers { std: Some(std_of("nightly-other", &bad)), ..Default::default() };
assert!(room.std_due(&other));
a.commit("fn a1\nfn a3\n");
let members = [a.member("alpha")];
let mut events = Vec::new();
let rep = room.consume(&other, &[], &Request { members: &members, ..Default::default() }, &mut recorder(&mut events)).unwrap();
assert_eq!(&events[..2], ["std-begin nightly-other", "std nightly-other failed"]);
let said = rep.std.as_deref().unwrap();
assert!(said.contains("NOT in the ring") && said.contains("the toolchain has no rust-src"), "{said}");
assert_eq!((rep.members[0].status, rep.members[0].reports.len()), (MemberStatus::CaughtUp, 1), "the member is consumed all the same");
assert!(!room.std_due(&other), "a dump that failed is not due on every tick");
let rep = room.consume(&other, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert!(rep.std.as_deref().unwrap().contains("failed earlier in this process"), "{:?}", rep.std);
assert_eq!(lines(&bad_runs), 1);
room.consume(&other, &[], &Request { members: &members, retry_feeds: true, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!(lines(&bad_runs), 2, "an explicit refresh tries again");
}
// ------------------------------------------------------- the ring, read back
/// **A path from the workspace's code down to an `extern` item answers.**
/// The dependency a commit compiles is written into the universe ring by that
/// commit's own release (once per `crate@version`), with its foreign items
/// and the names they are linked under. The room of the repo's own code ends
/// at the border; `Knowledge.CallPath` then walks with the ring folded in.
///
/// RED WHEN REVERTED: make `call_path` answer from the repo's own room only
/// (drop the ring walk) and `write` names no node — `invalid_argument`.
#[test]
fn a_call_path_from_workspace_code_to_an_extern_item_walks_through_the_ring() {
use serde_json::json;
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\nfn a2\ncall a1 a2\nffi a2 write\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha")];
room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
// the ring crate is in the manifest, written by the member's own release
let log = room.log().unwrap();
assert_eq!(log.block_on(log.universe_rows()).unwrap().iter().map(|m| m.0.as_str()).collect::<Vec<_>>(), vec!["libc@0.2.0:lib"]);
let reach = super::Reach::room_only(&room);
let ask = |to: &str| super::reader::answer("Knowledge.CallPath", &reach, &json!({ "repo": "alpha", "from": "a1", "to": to }));
let monikers = |v: &serde_json::Value| v["path"].as_array().unwrap().iter().map(|s| s["moniker"].as_str().unwrap().to_string()).collect::<Vec<_>>();
// by the item's name …
let v = ask("write").unwrap();
assert_eq!(monikers(&v), vec!["alpha a1", "alpha a2", "libc write"], "{v}");
assert_eq!((&v["hops"], &v["ring"], &v["path"][2]["kind"]), (&json!(2), &json!(true), &json!("ffi")), "{v}");
assert!(v["evidence"]["notes"].as_array().unwrap().iter().any(|n| n.as_str().unwrap().contains("universe ring folded in")), "{v}");
assert!(v["evidence"]["scope"].as_str().unwrap().contains("universe on"), "{v}");
// … and by the name the linker knows it under (the `link_name` column)
let v = ask("__write").unwrap();
assert_eq!(monikers(&v), vec!["alpha a1", "alpha a2", "libc write"], "{v}");
// a path that stays in the repo's own code is answered by its own room
let v = super::reader::answer("Knowledge.CallPath", &reach, &json!({ "repo": "alpha", "from": "a1", "to": "a2" })).unwrap();
assert_eq!((monikers(&v), v.get("ring")), (vec!["alpha a1".to_string(), "alpha a2".to_string()], None), "{v}");
assert!(v["evidence"]["scope"].as_str().unwrap().contains("universe off"), "{v}");
// a name nothing has, ring or not, is still an argument error
assert_eq!(ask("no_such_item").unwrap_err().state, State::InvalidArgument);
}
// ------------------------------------------------- siblings, and what fails
/// **A member is dumped against its siblings' TIPS, not their offsets.** Two
/// members move in the same poll and the first one's new commit needs the
/// second one's: the sibling's consumer clone is at the commit its own
/// consumer left it on, which is a commit too early. The run fetches every
/// member first and puts every sibling at the tip of the branch it follows
/// before a member is dumped — and the sibling's own walk, later, starts from
/// wherever its tree is and still writes each of its commits' own tree.
///
/// RED WHEN REVERTED: drop the `siblings_at_tip` call and alpha's commit is
/// dumped against beta's old tree — `error`, nothing compiled, offset moved on.
#[test]
fn a_member_is_dumped_against_its_siblings_tips_not_their_offsets() {
let tmp = tempfile::tempdir().unwrap();
let (a, b) = (Repo::new(tmp.path(), "ws", "alpha"), Repo::new(tmp.path(), "ws", "beta"));
a.commit("fn a1\n");
b.commit("fn b1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha"), b.member("beta")];
room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
// ---- both move in one poll; alpha (first in the order) needs beta's new item
b.commit("fn b1\nfn b2\n");
a.commit("fn a1\nsees beta b2\n");
let members = [a.member("alpha"), b.member("beta")];
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let alpha = &rep.members[0].reports[0];
assert_eq!((alpha.status.as_str(), alpha.error.as_deref(), alpha.failed_units.len()), ("ok", None, 0), "alpha's commit was dumped against beta's tip");
assert_eq!(rep.stamp(), "alpha@2,beta@2");
// ---- the sibling stays BEHIND its tip (a bounded run), and its own walk is
// not disturbed by its tree having been at the tip for alpha's dump
b.commit("fn b1\nfn b2\nfn b3\n");
b.commit("fn b1\nfn b2\nfn b3\nfn b4\n");
b.commit("fn b1\nfn b2\nfn b3\nfn b4\nfn b5\n");
a.commit("fn a1\nsees beta b5\n");
let members = [a.member("alpha"), b.member("beta")];
let rep = room.consume(&dumpers, &[], &Request { members: &members, max_commits: Some(1), ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((rep.members[0].reports[0].status.as_str(), rep.members[1].status, rep.members[1].behind), ("ok", MemberStatus::Behind, 2));
let rep = room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((rep.members[1].status, rep.members[1].reports.len()), (MemberStatus::CaughtUp, 2));
// every ordinal of beta is its own commit's tree: b1 | b1 b2 | … | b1…b5
for (seq, items) in [(1, 1), (2, 2), (3, 3), (4, 4), (5, 5)] {
assert_eq!(room.room(&Scope::repo("beta", Some(seq))).unwrap().len(), items, "beta@{seq}");
}
}
/// **A dump that fails for a reason outside the commit does not move the
/// offset, keeps the member behind — and is not retried for ever.** Here the
/// member's tree names a sibling that is not a member, so its clone is never
/// there. Three runs fail the same way; then the member is `blocked` (said in
/// its outcome and in status) and no run dumps it again until a member of the
/// workspace moves or somebody asks.
///
/// RED WHEN REVERTED: drop the `blocked` arm of `consume_locked` and the
/// fourth run dumps the member again (and `behind` names it on every tick).
#[test]
fn a_dump_that_fails_for_a_reason_outside_the_commit_keeps_the_member_behind_and_is_not_retried_for_ever() {
use super::consumer_loop::STUCK_AFTER;
let tmp = tempfile::tempdir().unwrap();
let (a, b) = (Repo::new(tmp.path(), "ws", "alpha"), Repo::new(tmp.path(), "ws", "beta"));
a.commit("fn a1\nneeds gamma\n");
b.commit("fn b1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha"), b.member("beta")];
let all = |members: &[Member]| run_all(&room, &dumpers, members);
// ---- it fails, the offset does not move, the member stays behind
for run in 1..=STUCK_AFTER {
assert_eq!(room.behind(&members, &[]).unwrap().first().map(String::as_str), Some("alpha"));
let rep = all(&members);
let alpha = &rep.members[0];
assert_eq!((alpha.status, alpha.offset.is_none(), rep.members[1].status), (MemberStatus::Failed, true, MemberStatus::CaughtUp));
assert!(alpha.error.as_deref().unwrap().contains("the sibling is not cloned"), "{:?}", alpha.error);
assert_eq!(dumpers.dumps("alpha"), run as usize);
}
let st = room.status(&members).unwrap();
assert_eq!((st.members[0].offset.is_none(), st.members[0].runs, st.members[0].last_run.as_ref().unwrap().status.as_str(), st.members[0].behind), (true, 3, "error", true));
// ---- three times the same way: blocked — said, and not dumped again
assert!(room.behind(&members, &[]).unwrap().is_empty(), "a blocked member is not what a tick runs for");
let why = st.members[0].blocked.as_deref().expect("status says it is blocked, and why");
assert!(why.contains("3 run(s) failed the same way") && why.contains("the sibling is not cloned") && why.contains("workspace repopulate"), "{why}");
assert!(st.members[0].line().contains("BLOCKED") && st.to_json()["members"][0]["blocked"].is_string(), "{}", st.members[0].line());
let rep = all(&members);
assert_eq!((rep.members[0].status, rep.members[0].status.as_str(), dumpers.dumps("alpha")), (MemberStatus::Blocked, "blocked", 3));
assert!(rep.members[0].error.as_deref().unwrap().starts_with("not tried again"), "{:?}", rep.members[0].error);
assert!(rep.failed().is_empty() && rep.behind().len() == 1, "blocked is behind, and not a new failure");
// ---- a member of the workspace moves: that could change the outcome
b.commit("fn b1\nfn b2\n");
let members = [a.member("alpha"), b.member("beta")];
assert_eq!(room.behind(&members, &[]).unwrap(), vec!["alpha", "beta"]);
assert_eq!((all(&members).members[0].status, dumpers.dumps("alpha")), (MemberStatus::Failed, 4));
all(&members);
all(&members);
assert!(room.blocked(&members, "alpha").is_some());
// ---- the box is put right and somebody asks: consumed, and the block is gone
crate::gitio::init(&room.layout().clone_of("gamma")).unwrap();
let rep = room.consume(&dumpers, &[], &Request { members: &members, retry_feeds: true, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((rep.members[0].status, rep.members[0].offset.as_ref().map(|o| o.seq)), (MemberStatus::CaughtUp, Some(1)), "{:?}", rep.members[0].error);
let st = room.status(&members).unwrap();
assert!(st.members[0].blocked.is_none() && !st.members[0].line().contains("BLOCKED"));
}
/// **A member whose history was rewritten under its offset** (a forge move:
/// the same commits again, same message and date, other hashes) is read again
/// from its head as a fresh member — not failed for ever — and when that
/// read itself fails, its line says the history was rewritten.
#[test]
fn a_member_whose_history_was_rewritten_is_read_again_from_its_head() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\nfn a2\n");
let old = a.commit("fn a1\nfn a2\nfn a3\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let rep = run_all(&room, &dumpers, &[a.member("alpha")]);
assert_eq!(rep.members[0].offset.as_ref().map(|o| (o.seq, o.sha.clone())), Some((1, old.clone())), "a new member starts at its tip");
// ---- the move: the server's clone now holds a rewritten history
let rewrite = |head: &str| {
std::fs::remove_dir_all(&a.dir).unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\nfn a2\n# moved\n");
a.commit(head);
a
};
let a = rewrite("fn a1\nfn a3\nfn a4\n");
let members = [a.member("alpha")];
assert_ne!(members[0].head, old);
let rep = run_all(&room, &dumpers, &members);
let m = &rep.members[0];
assert_eq!((m.status, m.error.as_deref()), (MemberStatus::CaughtUp, None), "{}", m.line());
assert_eq!(m.offset.as_ref().map(|o| (o.seq, o.sha.clone())), Some((2, members[0].head.clone())), "ONE release at the new head");
assert_eq!((m.reports[0].rows_plus, m.reports[0].rows_minus), (1, 1), "a4 in, a2 (only in the old history) out");
assert_eq!(room.room(&Scope::current()).unwrap().len(), 3, "a1 a3 a4: nothing of the old history stays open");
// ---- rewritten again, and the read at the new head fails: said so
let a = rewrite("fn a1\nCRASH\n");
let rep = run_all(&room, &dumpers, &[a.member("alpha")]);
let m = &rep.members[0];
assert_eq!((m.status, m.offset.as_ref().map(|o| o.seq)), (MemberStatus::Failed, Some(2)));
assert!(m.line().contains("failed") && m.error.as_deref().unwrap().starts_with("history rewritten: "), "{}", m.line());
}
/// One unattended run over every member.
fn run_all(room: &CodeRoom, dumpers: &TreeDumpers, members: &[Member]) -> super::consumer_loop::ConsumeReport {
room.consume(dumpers, &[], &Request { members, ..Default::default() }, &mut quiet()).unwrap()
}
/// **A commit that is not `ok` says why** — in its report (the log line of
/// every front door), in the job detail and in status — not only that it is
/// an `error` with 0 crates.
#[test]
fn a_commit_that_is_not_ok_says_why_on_its_line_and_in_status() {
use super::consumer_loop::{commit_json, commit_reason};
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\nBROKEN\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let members = [a.member("alpha")];
let mut reports = Vec::new();
room.consume(&TreeDumpers::default(), &[], &Request { members: &members, ..Default::default() }, &mut |e| {
if let Event::Commit { report, .. } = e {
reports.push(report.clone())
}
})
.unwrap();
assert_eq!(reports[0].status, "error");
assert_eq!(commit_reason(&reports[0]), " — nothing compiled; failed units: alpha:lib");
assert_eq!(commit_json(&reports[0])["error"], "nothing compiled; failed units: alpha:lib");
let st = room.status(&members).unwrap();
assert!(st.members[0].line().ends_with("(error: nothing compiled; failed units: alpha:lib)"), "{}", st.members[0].line());
// a commit that IS ok adds nothing to its line
a.commit("fn a1\n");
let members = [a.member("alpha")];
let mut reports = Vec::new();
room.consume(&TreeDumpers::default(), &[], &Request { members: &members, ..Default::default() }, &mut |e| {
if let Event::Commit { report, .. } = e {
reports.push(report.clone())
}
})
.unwrap();
assert_eq!((reports[0].status.as_str(), commit_reason(&reports[0]).as_str()), ("ok", ""));
assert!(room.status(&members).unwrap().members[0].line().ends_with("(ok)"));
}
/// **A member whose descriptor names a branch is that branch's tip** — not
/// the server clone's HEAD. After the 2026-10-14 move the forge default was
/// knut's `ny_dok` and znippy's stale `main`; siblings were put at those and
/// every member that path-depends on them failed "outside the commit".
/// RED WHEN REVERTED: `of_workspace` taking `head_sha(source)` answers c2.
#[test]
fn a_named_branch_is_the_members_head_not_the_clones_head() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
let c1 = a.commit("fn a1\n");
{
// the reflog line needs a committer: the repo's own, not the box's (a
// build container has no global git identity)
use std::io::Write as _;
let mut cfg = std::fs::OpenOptions::new().append(true).open(a.dir.join(".git").join("config")).unwrap();
writeln!(cfg, "[user]\n\tname = consumer-test\n\temail = consumer-test@example.invalid").unwrap();
let repo = gix::open(&a.dir).unwrap();
let id = gix::ObjectId::from_hex(c1.as_bytes()).unwrap();
repo.reference("refs/heads/live", id, gix::refs::transaction::PreviousValue::Any, "test: the live line").unwrap();
}
let c2 = a.commit("fn a1\nfn a2\n"); // the clone's HEAD branch moves on; `live` does not
assert_ne!(c1, c2);
let ws = crate::registry::Workspace {
name: "ws".into(),
mode: crate::registry::Mode::Monitored,
descriptor: String::new(),
poll: String::new(),
current_snapshot: String::new(),
members: vec![
crate::registry::MemberState { name: "alpha".into(), git_ref: "live".into(), ..Default::default() },
crate::registry::MemberState { name: "plain".into(), ..Default::default() },
],
descriptor_content: String::new(),
created_at: String::new(),
updated_at: String::new(),
};
let members = Member::of_workspace(tmp.path(), &ws);
assert_eq!((members[0].branch.as_deref(), members[0].head.as_str()), (Some("live"), c1.as_str()), "the named branch's tip");
assert_eq!(members[1].branch, None, "no branch named: nothing guessed here");
}
/// `only: Affected`: the members a change reaches are the changed ones and
/// every member that depends on one of them — through another member too —
/// by the clones' manifests (Mímir's dependency graph); a member that does
/// not depend on the change is not in it.
#[test]
fn the_members_a_change_reaches_are_it_and_its_dependents() {
let tmp = tempfile::tempdir().unwrap();
let krate = |repo: &str, deps: &[&str]| {
let dir = tmp.path().join(repo);
std::fs::create_dir_all(dir.join("src")).unwrap();
let deps: String = deps.iter().map(|d| format!("{d} = {{ path = \"../{d}\" }}\n")).collect();
std::fs::write(dir.join("Cargo.toml"), format!("[package]\nname = \"{repo}\"\nversion = \"0.1.0\"\nedition = \"2021\"\n\n[dependencies]\n{deps}")).unwrap();
std::fs::write(dir.join("src/lib.rs"), "").unwrap();
(repo.to_string(), dir)
};
// c → b → a; d alone
let trees: std::collections::BTreeMap<String, PathBuf> = [krate("a", &[]), krate("b", &["a"]), krate("c", &["b"]), krate("d", &[])].into_iter().collect();
let got = |changed: &[&str]| -> BTreeSet<String> {
super::consumer_loop::affected_members(&trees, &changed.iter().map(|s| s.to_string()).collect::<Vec<_>>()).unwrap().into_iter().collect()
};
let set = |v: &[&str]| v.iter().map(|s| s.to_string()).collect::<BTreeSet<String>>();
assert_eq!(got(&["a"]), set(&["a", "b", "c"]));
assert_eq!(got(&["b"]), set(&["b", "c"]));
assert_eq!(got(&["d"]), set(&["d"]));
assert_eq!(got(&["c", "d"]), set(&["c", "d"]));
// in build order: what a member depends on comes first
let order = super::consumer_loop::affected_members(&trees, &["a".to_string()]).unwrap();
assert_eq!(order, vec!["a".to_string(), "b".to_string(), "c".to_string()]);
}
/// `only: Affected` in a run: the member that moved is read; the member that
/// depends on it is the one to compile (here, with no super build on a
/// scripted dumper, it says it was not compiled and why); a member the change
/// does not reach is not touched — no dump, no turn.
#[test]
fn an_affected_run_reads_what_moved_and_names_its_dependents_only() {
let tmp = tempfile::tempdir().unwrap();
let repo = |name: &str, deps: &[&str]| {
let r = Repo::new(tmp.path(), "ws", name);
let deps: String = deps.iter().map(|d| format!("{d} = {{ path = \"../{d}\" }}\n")).collect();
std::fs::write(r.dir.join("Cargo.toml"), format!("[package]\nname = \"{name}\"\nversion = \"0.1.0\"\nedition = \"2021\"\n\n[dependencies]\n{deps}")).unwrap();
r.commit(&format!("fn {name}1\n"));
r
};
let (a, b, g) = (repo("alpha", &[]), repo("beta", &["alpha"]), repo("gamma", &[]));
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha"), b.member("beta"), g.member("gamma")];
room.consume(&dumpers, &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!((dumpers.dumps("alpha"), dumpers.dumps("beta"), dumpers.dumps("gamma")), (1, 1, 1));
a.commit("fn alpha1\nfn alpha2\n");
let members = [a.member("alpha"), b.member("beta"), g.member("gamma")];
let rep = room.consume(&dumpers, &[], &Request { members: &members, affected: true, ..Default::default() }, &mut quiet()).unwrap();
assert_eq!(rep.members.iter().map(|m| m.member.as_str()).collect::<Vec<_>>(), vec!["alpha"], "only what moved takes a turn");
assert_eq!(rep.stamp(), "alpha@2");
assert_eq!(rep.checked.iter().map(|c| c.member.as_str()).collect::<Vec<_>>(), vec!["beta"], "the dependent, not gamma");
assert!(rep.checked[0].exit.is_none() && rep.checked[0].note.as_deref().is_some_and(|n| n.contains("no super build")), "{:?}", rep.checked);
assert_eq!((dumpers.dumps("alpha"), dumpers.dumps("beta"), dumpers.dumps("gamma")), (2, 1, 1));
// nothing moved: nothing to read, nothing to check
let rep = room.consume(&dumpers, &[], &Request { members: &members, affected: true, ..Default::default() }, &mut quiet()).unwrap();
assert!(rep.members.is_empty() && rep.checked.is_empty(), "{:?}", rep.checked);
}
/// Dumpers that plan (the driver's do: the build board hears their unit plans).
struct Planning(TreeDumpers);
impl Dumpers for Planning {
fn dumper(&self, layout: &Layout, member: &Member) -> Result<Box<dyn Dumper + Send>, Refusal> {
self.0.dumper(layout, member)
}
fn describe(&self) -> String {
"planning tree dumper (test)".into()
}
fn plans(&self) -> bool {
true
}
}
/// The board must not stay empty in a process that can plan: a member with a
/// clone and a manifest but no plan on the board makes the next run plan it,
/// also when nothing is behind (the sweep's early return); a scripted source
/// plans nothing and never asks for it.
#[test]
fn a_board_without_every_members_plan_asks_for_a_run() {
let tmp = tempfile::tempdir().unwrap();
let repo = |name: &str| {
let r = Repo::new(tmp.path(), "ws", name);
std::fs::write(r.dir.join("Cargo.toml"), format!("[package]\nname = \"{name}\"\nversion = \"0.1.0\"\nedition = \"2021\"\n")).unwrap();
r.commit(&format!("fn {name}1\n"));
r
};
let (a, b) = (repo("alpha"), repo("beta"));
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let members = [a.member("alpha"), b.member("beta")];
room.consume(&TreeDumpers::default(), &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let planning = Planning(TreeDumpers::default());
assert!(room.board_unplanned(&members, &planning), "a fresh process: nothing planned yet");
let board = super::units::board(&room.layout().clones());
let graph = |n: &str| serde_json::json!({ "units": [{ "pkg_id": format!("path+file://{}/{n}#0.1.0", room.layout().clones().display()), "target": { "kind": ["lib"], "name": n }, "profile": { "name": "dev" }, "platform": null, "mode": "check", "features": [], "dependencies": [] }] });
board.lock().unwrap().plan("alpha", &graph("alpha"));
assert!(room.board_unplanned(&members, &planning), "beta has no plan");
board.lock().unwrap().plan("beta", &graph("beta"));
assert!(!room.board_unplanned(&members, &planning), "every member planned");
assert!(!room.board_unplanned(&members, &TreeDumpers::default()), "a scripted source plans nothing");
// a member whose plan was asked for at its head (cargo could not give
// one) is not asked for again every tick — at its next head it is
let gamma = repo("gamma");
let members = [a.member("alpha"), b.member("beta"), gamma.member("gamma")];
room.consume(&TreeDumpers::default(), &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
assert!(room.board_unplanned(&members, &planning), "gamma has no plan");
board.lock().unwrap().try_plan("gamma", &members[2].head);
assert!(!room.board_unplanned(&members, &planning), "tried at this head");
gamma.commit("fn gamma1\nfn gamma2\n");
let members = [a.member("alpha"), b.member("beta"), gamma.member("gamma")];
assert!(room.board_unplanned(&members, &planning), "a new head: asked again");
}
// ------------------------------------------------- embeddings: hardware came
/// The real `embeddings` feed on a box without embedding hardware writes
/// `skipped` for the sha — and that sha is settled for as long as there is
/// none. Once there is (a GPU worker named with `$NORNIR_EMBED_REMOTE`, a GPU
/// that came up), the SAME sha is due again by itself: the member is behind,
/// and the next run hands the feed's work out — no new commit needed, no
/// operator verb. Any other skip stays settled.
///
/// RED WHEN REVERTED: drop `retry_due`'s `skipped` arm (or `Embeddings::retry`)
/// and the member is never behind again — the vectors of an unmoving member
/// wait for ever.
#[cfg(all(feature = "vector", any(feature = "embed-tract", feature = "embed-ort")))]
#[test]
fn a_sha_skipped_for_want_of_hardware_is_due_again_once_there_is_hardware() {
use super::feeds::{Embeddings, Sink};
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
let sha = a.commit("fn a1\nfn a2\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha")];
let sink = Sink {
warehouse: Arc::new(crate::warehouse::iceberg::IcebergWarehouse::open(&tmp.path().join("wh")).unwrap()),
index: Arc::new(crate::index::Index::open_at(tmp.path(), &tmp.path().join("index")).unwrap()),
git_dir: tmp.path().join("ws/git"),
index_dir: tmp.path().join("index"),
members: vec!["alpha".into()],
jobs: None,
ring_queue: None,
};
// ---- no hardware: `skipped`, and settled
let cpu = Embeddings { sink: sink.clone(), worker_up: || false, no_hardware: || Some("no embedding hardware here: no NVIDIA CUDA runtime and no AMD ROCm runtime is ready".into()) };
let feeds: [&dyn Feed; 1] = [&cpu];
let rep = room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let row = &rep.members[0].feeds[0];
assert_eq!((row.feed.as_str(), row.sha.as_str(), row.status.as_str()), ("embeddings", sha.as_str(), "skipped"));
assert!(room.behind(&members, &feeds).unwrap().is_empty(), "no hardware: the skip is settled");
assert!(room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap().members[0].feeds.is_empty());
// ---- hardware came: the same sha is due again, and its work is handed out
let gpu = Embeddings { sink, no_hardware: || None, worker_up: || false };
let feeds: [&dyn Feed; 1] = [&gpu];
assert_eq!(room.behind(&members, &feeds).unwrap(), vec!["alpha"], "the skipped sha is due again");
let mut handed: Vec<Deferred> = Vec::new();
room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut |e| {
if let Event::Deferred(d) = e {
handed.push(d)
}
})
.unwrap();
assert_eq!(handed.iter().map(|d| (d.feed, d.sha.as_str())).collect::<Vec<_>>(), vec![("embeddings", sha.as_str())], "the run hands the embed out");
assert!(room.behind(&members, &feeds).unwrap().is_empty(), "in flight is not behind");
}
/// A skip that is not the hardware gate's (another reason) is not retried
/// when hardware is there: the retry is for the reason that went away.
#[cfg(all(feature = "vector", any(feature = "embed-tract", feature = "embed-ort")))]
#[test]
fn a_skip_for_another_reason_is_not_retried_by_the_hardware_rule() {
let tmp = tempfile::tempdir().unwrap();
let sink = super::feeds::Sink {
warehouse: Arc::new(crate::warehouse::iceberg::IcebergWarehouse::open(&tmp.path().join("wh")).unwrap()),
index: Arc::new(crate::index::Index::open_at(tmp.path(), &tmp.path().join("index")).unwrap()),
git_dir: tmp.path().join("git"),
index_dir: tmp.path().join("index"),
members: vec!["alpha".into()],
jobs: None,
ring_queue: None,
};
let gpu = super::feeds::Embeddings { sink, no_hardware: || None, worker_up: || false };
let row = |status: &str, detail: &str| mimir_ingest::FeedRow { repo: "alpha".into(), seq: 1, sha: "s".into(), feed: "embeddings".into(), status: status.into(), rows: 0, ms: 0, detail: Some(detail.into()) };
let layout = Layout::under(&tmp.path().join("builds"));
let log = mimir_ingest::EventLog::open_lake_blocking(&tmp.path().join("lake")).unwrap();
let member = Member { name: "alpha".into(), source: tmp.path().join("git/alpha"), branch: None, head: "s".into() };
let at = FeedAt { workspace: "ws", member: &member, seq: 1, sha: "s", log: &log, layout: &layout };
assert!(gpu.retry(&at, &row("skipped", "no embedding hardware here: x. Not embedded unattended: `nornir workspace embed ws` embeds now")));
assert!(!gpu.retry(&at, &row("skipped", "the line-window half has nothing to read")));
assert!(!gpu.retry(&at, &row("error", "Not embedded unattended")), "an error row is not a skip");
assert!(!gpu.retry(&at, &row("ok", "")));
}
/// A dependent is judged against its last read: a unit that failed then and
/// fails now "was already failing" (with that read's seq); only a unit that
/// built then and fails now is the change's — a regression.
#[test]
fn a_dependent_is_judged_against_its_last_read() {
use super::consumer_loop::{Checked, judge_check};
let now: BTreeSet<String> = ["a:lib:a".to_string(), "a:test:t1".to_string()].into();
let partial = Some((7, "partial", Some("failed units (kept their previous state): a:test:t1, a:bin:x")));
let (reg, already, since) = judge_check(&now, partial);
assert_eq!((reg, already, since), (vec!["a:lib:a".to_string()], vec!["a:test:t1".to_string()], Some(7)));
let (reg, already, _) = judge_check(&now, Some((7, "ok", None)));
assert_eq!((reg.len(), already.len()), (2, 0), "it built: both are the change's");
let (reg, _, _) = judge_check(&now, None);
assert_eq!(reg.len(), 2, "never read: nothing to hold against");
// the line and the verdict
let only_old = Checked { member: "a".into(), exit: Some(101), ms: 5, already_failing: vec!["a:test:t1".into()], failing_since: Some(7), ..Default::default() };
assert!(only_old.ok() && only_old.line().contains("already failing at seq 7"), "{}", only_old.line());
let broke = Checked { member: "a".into(), exit: Some(101), ms: 5, regressions: vec!["a:lib:a".into()], already_failing: vec!["a:test:t1".into()], failing_since: Some(7), ..Default::default() };
assert!(!broke.ok() && broke.line().contains("breaks a:lib:a"), "{}", broke.line());
let unknown = Checked { member: "a".into(), exit: Some(101), ms: 5, ..Default::default() };
assert!(!unknown.ok(), "it failed and nothing says it failed before");
}
/// A clone that cannot be made (or checked out) is said as what it is —
/// "checkout failed: <why>" — on its member's turn AND on every member that
/// reaches into it, never as a sibling that is missing (loki 2026-10-17: an
/// LFS smudge stopped znippy-zoomies' checkout halfway and ten repos read
/// "sibling missing: …/znippy-zoomies/gatling").
#[test]
fn a_clone_that_fails_is_said_as_a_failed_checkout_on_every_member_it_reaches() {
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
a.commit("fn a1\nneeds beta\n");
let b = Repo::new(tmp.path(), "ws", "beta");
b.commit("fn b1\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
// beta's clone cannot be made: a file stands where it goes
let at = room.layout().clone_of("beta");
std::fs::create_dir_all(at.parent().unwrap()).unwrap();
std::fs::write(&at, "in the way").unwrap();
let members = [a.member("alpha"), b.member("beta")];
let rep = room.consume(&TreeDumpers::default(), &[], &Request { members: &members, ..Default::default() }, &mut quiet()).unwrap();
let err = |n: &str| rep.members.iter().find(|m| m.member == n).and_then(|m| m.error.clone()).unwrap_or_default();
assert!(err("beta").starts_with("checkout failed: "), "beta: {}", err("beta"));
assert!(err("alpha").contains("checkout failed: the clone of beta"), "alpha: {}", err("alpha"));
assert!(!err("alpha").contains("not cloned"), "alpha: {}", err("alpha"));
}
/// A sha whose embed failed because the GPU worker did not answer (it was
/// restarting, its box was off) is due again by itself once the worker
/// answers — not at the member's next commit, which may be days away. Any
/// other failure stays settled, and so does this one while the worker is
/// still away.
///
/// RED WHEN REVERTED: drop the `error` arm of `retry_due` (or
/// `Embeddings::worker_came_back`) and the member is never behind again.
#[cfg(all(feature = "vector", any(feature = "embed-tract", feature = "embed-ort")))]
#[test]
fn a_sha_that_failed_on_an_unreachable_worker_is_due_again_once_it_answers() {
use super::feeds::{Embeddings, Sink};
let tmp = tempfile::tempdir().unwrap();
let a = Repo::new(tmp.path(), "ws", "alpha");
let sha = a.commit("fn a1\nfn a2\n");
let room = CodeRoom::new("ws", &tmp.path().join("ws/builds"));
let dumpers = TreeDumpers::default();
let members = [a.member("alpha")];
let sink = Sink {
warehouse: Arc::new(crate::warehouse::iceberg::IcebergWarehouse::open(&tmp.path().join("wh")).unwrap()),
index: Arc::new(crate::index::Index::open_at(tmp.path(), &tmp.path().join("index")).unwrap()),
git_dir: tmp.path().join("ws/git"),
index_dir: tmp.path().join("index"),
members: vec!["alpha".into()],
jobs: None,
ring_queue: None,
};
// the run: consumed, the embed handed out — and it fails on the worker
let away = Embeddings { sink: sink.clone(), no_hardware: || None, worker_up: || false };
let feeds: [&dyn Feed; 1] = [&away];
let mut handed: Vec<Deferred> = Vec::new();
room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut |e| {
if let Event::Deferred(d) = e {
handed.push(d)
}
})
.unwrap();
let mut d = handed.pop().expect("the embed is handed out");
d.work = Box::new(|| anyhow::bail!("connect to the embed worker at http://10.0.0.9:7932: transport error: Connection refused"));
let row = room.run_deferred(d).unwrap();
assert_eq!((row.sha.as_str(), row.status.as_str()), (sha.as_str(), "error"));
assert!(room.behind(&members, &feeds).unwrap().is_empty(), "the worker is still away: settled");
// the worker answers again: the same sha is due by itself
let back = Embeddings { sink: sink.clone(), no_hardware: || None, worker_up: || true };
let feeds: [&dyn Feed; 1] = [&back];
assert_eq!(room.behind(&members, &feeds).unwrap(), vec!["alpha"], "the failed sha is due again");
// another failure is not the worker's: settled even with the worker up
let mut handed: Vec<Deferred> = Vec::new();
room.consume(&dumpers, &feeds, &Request { members: &members, ..Default::default() }, &mut |e| {
if let Event::Deferred(d) = e {
handed.push(d)
}
})
.unwrap();
let mut d = handed.pop().expect("handed out again");
d.work = Box::new(|| anyhow::bail!("the warehouse refused the append"));
assert_eq!(room.run_deferred(d).unwrap().status, "error");
assert!(room.behind(&members, &feeds).unwrap().is_empty(), "not the worker's failure: settled");
}