src/code_room/consumer_loop.rs
//! The workspace consumer — what replaced the poll loop's populate.
//!
//! ```text
//! registry → members { name, git/<name> clone, branch, head }
//! per member, in order:
//! behind? no offset in the log, or offset.sha != head (boot, a wiped
//! lake and "a week away" are this same test)
//! fetch the consumer's OWN clone ← the server's clone (a local path:
//! auth + network stay in the one fetch door)
//! start a member with no offset starts at the TIP (decision D2);
//! `--from <sha|date>` names another first commit, or — for a
//! member that already has an offset — replays from it
//! consume Consumer::consume_run: per commit, ONE atomic release
//! (events + offset + ledger), bounded, cancellable, observed
//! feeds once the member is caught up: every non-graph feed whose own
//! offset (its last `ok` ledger row) is not this sha
//! ```
//!
//! Before the members, once per toolchain: the toolchain's own crates (std,
//! core, alloc …) are dumped into the universe ring ([`Dumpers::std_dumper`]).
//! The dependencies are in the ring already — the driver dumps every crate it
//! compiles, and a `crate@version` is written once.
//!
//! Three things do NOT happen inside a run, because a run holds the one
//! consumer (and, on a server, the workspace's gate):
//!
//! * a **deferred feed** ([`Feed::prepare`]: embeddings) only reads what it
//! needs of the tree; the long part is handed out as [`Event::Deferred`] and
//! runs where the caller wants (the server's background lane), then writes
//! its ledger row through [`CodeRoom::run_deferred`];
//! * a **backfill** (`super::backfill`) — a member's history asked for from a
//! commit — is walked ONE commit per run by its own job, and the member is
//! left out of every other run meanwhile ([`MemberStatus::Backfilling`]);
//! * a build: the binary hook reads artifacts that already exist.
//!
//! ## One tree per member, and what its siblings are at
//!
//! A member's manifest names its siblings by path (`../edda`), so a member is
//! never dumped alone. Two rules, both applied before any member is dumped:
//!
//! * **every member's clone is there and fetched** — also the ones this run
//! does not consume, also the ones that come later in the member order;
//! * **every sibling is at the tip of the branch it follows** while a member
//! is dumped (the commit its server clone is at) — not at the commit its own
//! consumer happens to be on. When two members move in the same poll and the
//! first one's new commit needs the second one's, the first is dumped
//! against the sibling's NEW tree. (There is no order between the commits of
//! two repos: a commit of the past, replayed by a backfill, is dumped
//! against its siblings as they are now, and says so if it does not compile.
//! One exception: a tip world that refuses a SIBLING's version — a commit
//! that pins an older sibling version can never load against the tips — is
//! retried once by the driver with every sibling dated at the commit's own
//! time, and the siblings are put back at their heads after; see
//! `mimir_ingest::driver`.)
//!
//! ## A dump that fails for a reason outside the commit
//!
//! A commit that does not compile is recorded and the offset moves on. A dump
//! that fails because of the BOX — a sibling that is not a member, a toolchain
//! that is not installed, the tool crashing — is not such a commit: the offset
//! stays, the member stays behind, and the run says why. It is tried again,
//! but not for ever: after [`STUCK_AFTER`] runs that failed the same way the
//! member is [`MemberStatus::Blocked`] until something that could change the
//! outcome happens — a member of the workspace moves, somebody asks
//! (`workspace repopulate`), the server restarts. Status says so.
//!
//! ONE consumer per workspace, kept for the life of the handle: it holds the
//! log's writer seat and the open set, and a [`DumperPool`] with one dumper
//! per member. Readers never wait for it (they pin the last release).
//!
//! Nothing here knows gRPC, jobs or the registry's storage: a caller hands
//! in [`Member`]s and gets [`Event`]s while it runs and a [`ConsumeReport`]
//! at the end. The server's sweep, `Workspaces.Register` / `Fetch` /
//! `Repopulate` and the CLI's fat-mode `workspace populate` all call the one
//! [`CodeRoom::consume`].
use std::cell::RefCell;
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering;
use std::time::Instant;
use mimir_ingest::{
BinaryDump, CommitReport, Consumer, DriverReady, Dumper, DumperPool, EventLog, FeedRow, Offset, RepoSpec, Run, StdDumper, StdOutcome, StdStatus, Stopped,
checkout, git, schema,
};
use serde_json::{Value, json};
use super::{CodeRoom, Layout, Refusal, State};
/// The pooled consumer of one workspace.
pub(super) type Pooled = Consumer<DumperPool<Box<dyn Dumper + Send>>>;
/// One workspace member as the consumer sees it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Member {
pub name: String,
/// The server's clone (`<root>/<ws>/git/<name>`): the consumer fetches its
/// own clone from here, and the feeds read this tree.
pub source: PathBuf,
/// The tracked branch, when the registry names one (`git_ref`).
pub branch: Option<String>,
/// The commit the server's clone is at — what "caught up" means.
pub head: String,
}
impl Member {
/// The members of a registered workspace, as they are on disk under
/// `<root>/<ws>/git/`. `head` is the tip of the branch the descriptor
/// names, when it names one — the line the member is read from, and the
/// commit a sibling is put at — else the clone's HEAD (what the tree
/// really is), else the registry's last seen sha. Taking HEAD for a named
/// branch put knut's siblings at the forge default `ny_dok` and znippy's at
/// a stale `main` after the 2026-10-14 move.
pub fn of_workspace(root: &Path, ws: &crate::registry::Workspace) -> Vec<Member> {
let git_dir = root.join(&ws.name).join("git");
ws.members
.iter()
.map(|m| {
let source = git_dir.join(&m.name);
let branch = (!m.git_ref.trim().is_empty()).then(|| m.git_ref.trim().to_string());
let tip = branch.as_deref().and_then(|b| {
let repo = gix::open(&source).ok()?;
git::branch_tip(&repo, b).ok().map(|id| id.to_string())
});
let head = tip
.or_else(|| crate::gitio::head_sha(&source).ok())
.unwrap_or_else(|| m.last_seen_sha.clone());
Member { name: m.name.clone(), source, branch, head }
})
.collect()
}
/// Is there a clone to consume from?
pub fn has_source(&self) -> bool {
self.source.join(".git").exists()
}
/// The branch whose first-parent history is this member's event stream,
/// and who named it: the registry (the workspace descriptor's `branch`),
/// else the branch the server's clone is on, else the one its remote names
/// as default (`git::remote_default_branch`). `None` when nothing names one
/// — a clone on a detached HEAD of a remote that has no default: there is
/// no stream to follow, and no name is guessed.
pub(super) fn tracked_branch(&self) -> Option<(String, &'static str)> {
if let Some(b) = &self.branch {
return Some((b.clone(), "the workspace descriptor"));
}
// `(detached)` is gitio's word for a clone on no branch
if let Some(b) = crate::gitio::head_branch(&self.source).ok().filter(|b| !b.is_empty() && !b.starts_with('(')) {
return Some((b, "the branch its clone is on"));
}
git::remote_default_branch(&self.source).ok().flatten().map(|b| (b, "its remote's default branch"))
}
}
/// Where a member's dumper comes from: the probed driver in production, a
/// scripted one in a test.
pub trait Dumpers {
fn dumper(&self, layout: &Layout, member: &Member) -> Result<Box<dyn Dumper + Send>, Refusal>;
/// Where a super build runs (the driver's host, its `cargo-mimir-dump`):
/// `None` = no super build here (a scripted source).
fn super_host(&self) -> Option<(mimir_ingest::DriverHost, PathBuf)> {
None
}
/// One line for reports: which driver, where.
fn describe(&self) -> String;
/// The dump of the toolchain's own crates (std, core, alloc and std's
/// private dependencies) for the universe ring — once per toolchain per
/// lake. `None`: there is no toolchain to dump here (a scripted source).
fn std_dumper(&self, layout: &Layout) -> Option<StdDumper> {
let _ = layout;
None
}
/// Do this source's dumpers tell the build board their unit plans
/// ([`super::units`])? A scripted source does not.
fn plans(&self) -> bool {
false
}
/// The system boundary to write into the universe after a run
/// ([`mimir_ingest::boundary::ensure`]): the target (its syscall table,
/// its libc) and which kernel's BTF to read. `None`: no boundary here (a
/// scripted source).
fn boundary(&self) -> Option<(mimir_ingest::boundary::Target, Option<mimir_ingest::boundary::KernelSource>)> {
None
}
}
/// The nightly driver, probed ([`super::driver::Settings::probe`]).
pub struct DriverDumpers(pub DriverReady);
impl Dumpers for DriverDumpers {
fn dumper(&self, layout: &Layout, member: &Member) -> Result<Box<dyn Dumper + Send>, Refusal> {
let mut d = self.0.spec.dumper(layout.driver_target(&member.name), layout.out_of(&member.name), member.name.clone());
// one linker flag set for every member: their units hash alike
d.flags = Some(super::units::flag_policy());
// the build board ([`super::units`]) watches every member's builds
d.progress = Some(mimir_ingest::ProgressSink(std::sync::Arc::new(super::units::Sink {
board: super::units::board(&layout.clones()),
member: member.name.clone(),
})));
Ok(Box::new(d))
}
fn describe(&self) -> String {
format!("{} · {} · {}", self.0.spec.cargo_mimir_dump.display(), self.0.rustc, self.0.spec.host.describe())
}
fn super_host(&self) -> Option<(mimir_ingest::DriverHost, PathBuf)> {
Some((self.0.spec.host.clone(), self.0.spec.cargo_mimir_dump.clone()))
}
fn plans(&self) -> bool {
true
}
/// The host's target (the std dump's `--target`) and the kernel: the
/// configured BTF (`$NORNIR_KERNEL_BTF`, `none` = no kernel) or the box's
/// own (`/sys/kernel/btf/vmlinux`).
fn boundary(&self) -> Option<(mimir_ingest::boundary::Target, Option<mimir_ingest::boundary::KernelSource>)> {
Some((mimir_ingest::boundary::Target::host()?, mimir_ingest::boundary::KernelSource::configured_or_box()))
}
fn std_dumper(&self, layout: &Layout) -> Option<StdDumper> {
let spec = &self.0.spec;
Some(StdDumper {
cargo_mimir_dump: spec.cargo_mimir_dump.clone(),
driver: spec.driver.clone(),
toolchain: spec.toolchain.clone(),
root: layout.std_root(),
host: spec.host.clone(),
})
}
}
/// Where a feed runs: one member, caught up at (seq, sha), its server clone
/// checked out there.
pub struct FeedAt<'a> {
pub workspace: &'a str,
pub member: &'a Member,
pub seq: i64,
pub sha: &'a str,
/// The workspace's event log (a feed that reads the code graph opens its
/// room from here, or pins it: the one open lake of the process).
pub log: &'a EventLog,
/// Where the code room's own directories are.
pub layout: &'a Layout,
}
/// What a feed run produced.
pub enum FeedResult {
/// It ran: how many rows / files / components it wrote, and a line.
Ok { rows: i64, detail: String },
/// It cannot run in this build / on this tree, and why. Not an error and
/// not done: the ledger row says `skipped`.
Skipped(String),
/// What it reads is not there (the binary hook: no linked artifact of the
/// commit where it looks). `state` is the ledger row's status
/// (`no_artifact`), `detail` names what was looked at. Not an error and
/// not done: [`Feed::retry`] says when to look again.
Absent { state: &'static str, detail: String },
/// The binary layer's rows of the artifacts found for this commit. A feed
/// has no writer seat: the consumer writes them — the binary stream, ONE
/// atomic release — and the ledger row says what landed.
/// `declared_bins`: the bins the workspace declares at the commit (an
/// artifact of any other bin that is open in the log closes).
Binaries { dumps: Vec<BinaryDump>, declared_bins: BTreeSet<String>, detail: String },
}
/// The long part of a deferred feed ([`Feed::prepare`]): everything it needs
/// of the tree is already inside.
pub type DeferredWork = Box<dyn FnOnce() -> anyhow::Result<FeedResult> + Send>;
/// A deferred feed's work for one (member, sha), handed out of the run
/// ([`Event::Deferred`]). Run it with [`CodeRoom::run_deferred`].
pub struct Deferred {
pub member: String,
pub seq: i64,
pub sha: String,
pub feed: &'static str,
pub work: DeferredWork,
}
/// A non-graph feed that rides the consumer's offsets (git heat, sbom, the
/// code index, embeddings): an existing producer, triggered per (repo, sha).
pub trait Feed {
/// Its key in the `consumer_feed_runs` ledger.
fn name(&self) -> &'static str;
fn run(&self, at: &FeedAt<'_>) -> anyhow::Result<FeedResult>;
/// A feed whose run is LONG (embeddings on a CPU: minutes per member)
/// must not run inside a consume, which holds the workspace's one
/// consumer. `Some`: this feed is deferred — `prepare` is called inside
/// the run instead of [`Feed::run`], reads what it needs of the tree as
/// it is at `at.sha`, and returns the work to do afterwards, outside
/// every lock. `None` (the default): [`Feed::run`], inline.
fn prepare(&self, at: &FeedAt<'_>) -> Option<anyhow::Result<DeferredWork>> {
let _ = at;
None
}
/// The feed's last row for this sha is not `ok` ([`FeedResult::Absent`],
/// `skipped`, `error`): is there something to read now, or is what stood
/// in the way (the embedding hardware, an unreachable GPU worker) gone? Asked on every "is anything
/// behind" check, so it must be cheap (a directory listing). Default: no.
fn retry(&self, at: &FeedAt<'_>, last: &FeedRow) -> bool {
let _ = (at, last);
false
}
}
/// What a run may consume.
#[derive(Debug, Clone, Default)]
pub struct Request<'a> {
pub members: &'a [Member],
/// Commits per member in this run (fairness: the rest stay behind the
/// offset and the next run resumes there). `None` = until caught up.
pub max_commits: Option<usize>,
/// `--from <sha|date>`: the first commit of a member with no offset, or
/// the commit an already consumed member is replayed from.
pub from: Option<&'a str>,
/// Only this member.
pub only: Option<&'a str>,
/// `BuildWorkspace { only: Affected }`: only the members behind are
/// read, and only the members that depend on them (Mímir's
/// `affected_by_change` over the clones' manifests) are compiled — beside
/// the reads, in the super build, dumping nothing — to see that they
/// still build ([`ConsumeReport::checked`]). The rest are not touched.
pub affected: bool,
/// Run again the feeds whose last row for the member's sha is not `ok`
/// (an explicit refresh). Without it a feed is tried ONCE per sha: a feed
/// that cannot run here (`skipped`) or failed is not retried every tick.
pub retry_feeds: bool,
}
/// What happens while a run goes on.
pub enum Event<'a> {
/// A member's turn begins: `behind` commits to consume.
Member { member: &'a str, behind: usize },
/// A commit's step begins.
Begin { member: &'a str, seq: i64, sha: &'a str, behind: usize },
/// A commit's release landed.
Commit { member: &'a str, report: &'a CommitReport },
/// A feed ran for the member's offset.
Feed { member: &'a str, row: &'a FeedRow },
/// A deferred feed read what it needs; its long part is the caller's to
/// run ([`CodeRoom::run_deferred`]) — after the run, outside its locks.
Deferred(Deferred),
/// The toolchain's own crates are about to be dumped into the ring (once
/// per toolchain; minutes).
StdBegin { toolchain: &'a str },
/// That dump ended: what it wrote, or why it failed.
Std { toolchain: &'a str, outcome: Result<&'a StdOutcome, &'a str>, ms: u64 },
/// The member's turn is over.
Done { outcome: &'a MemberOutcome },
}
/// How a member's turn ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MemberStatus {
/// Its offset is its head.
CaughtUp,
/// The commit budget ran out: commits are still behind the offset.
Behind,
/// The run was cancelled before it caught up.
Cancelled,
/// An infrastructure failure stopped it (the ledger has the error row).
Failed,
/// No clone under `git/` yet: nothing to consume from.
NoSource,
/// There is no branch to follow: nothing names one (the clone is on a
/// detached HEAD, the registry names none), or the named one is not in
/// the clone. `error` says what to set.
NoBranch,
/// A backfill of the member is open (`super::backfill`): its job walks it
/// one commit per run, and no other run touches it.
Backfilling,
/// The member's last [`STUCK_AFTER`] runs failed the same way for a reason
/// outside its commits (`error` says which): it is behind and NOT tried
/// again until a member of the workspace moves, somebody asks, or the
/// process restarts.
Blocked,
}
impl MemberStatus {
pub fn as_str(self) -> &'static str {
match self {
MemberStatus::CaughtUp => "caught_up",
MemberStatus::Behind => "behind",
MemberStatus::Cancelled => "cancelled",
MemberStatus::Failed => "failed",
MemberStatus::NoSource => "no_source",
MemberStatus::NoBranch => "no_branch",
MemberStatus::Backfilling => "backfilling",
MemberStatus::Blocked => "blocked",
}
}
}
/// One member's part of a run.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberOutcome {
pub member: String,
pub status: MemberStatus,
/// The commit the server's clone is at.
pub head: String,
/// The member's offset after the run.
pub offset: Option<Offset>,
/// First-parent commits still behind the offset.
pub behind: usize,
/// The commits consumed in this run.
pub reports: Vec<CommitReport>,
/// The feeds that ran in this run.
pub feeds: Vec<FeedRow>,
pub error: Option<String>,
pub ms: u64,
}
impl MemberOutcome {
/// Commits of this run whose code did not (fully) compile.
pub fn partial(&self) -> usize {
self.reports.iter().filter(|r| r.status == schema::STATUS_PARTIAL || r.status == schema::STATUS_ERROR).count()
}
pub fn to_json(&self) -> Value {
json!({
"member": self.member,
"status": self.status.as_str(),
"head": self.head,
"offset": self.offset.as_ref().map(|o| json!({ "seq": o.seq, "sha": o.sha, "branch": o.branch })),
"behind": self.behind,
"commits": self.reports.iter().map(commit_json).collect::<Vec<_>>(),
"feeds": self.feeds.iter().map(feed_json).collect::<Vec<_>>(),
"error": self.error,
"ms": self.ms,
})
}
/// `skade@7 (3 commits, +120 −4, 1 partial) caught_up`
pub fn line(&self) -> String {
let at = match &self.offset {
Some(o) => format!("{}@{} {}", self.member, o.seq, short(&o.sha)),
None => format!("{} (no offset)", self.member),
};
let (plus, minus): (i64, i64) = self.reports.iter().fold((0, 0), |a, r| (a.0 + r.rows_plus, a.1 + r.rows_minus));
let mut s = format!("{at}: {} — {} commit(s), +{plus} −{minus}", self.status.as_str(), self.reports.len());
if self.partial() > 0 {
s.push_str(&format!(", {} partial/error", self.partial()));
}
if self.behind > 0 {
s.push_str(&format!(", {} behind", self.behind));
}
if let Some(e) = &self.error {
s.push_str(&format!(" — {e}"));
}
s
}
}
/// One ledger row of a consumed commit, as JSON (job detail, reports).
pub fn commit_json(r: &CommitReport) -> Value {
json!({
"seq": r.seq,
"sha": r.sha,
"status": r.status,
"crates_dumped": r.crates_dumped,
"rows_plus": r.rows_plus,
"rows_minus": r.rows_minus,
"span_plus": r.span_plus,
"span_minus": r.span_minus,
"key_collisions": r.key_collisions,
"failed_units": r.failed_units,
"error": r.error,
"files_written": r.files_written,
"files_deleted": r.files_deleted,
"ms": r.ms,
"phase_ms": {
"classify": r.phase_ms.classify,
"checkout": r.phase_ms.checkout,
"dump": r.phase_ms.dump,
"delta": r.phase_ms.delta,
"commit": r.phase_ms.commit,
"dump_detail": r.phase_ms.dump_detail.map(|d| json!({
"cargo": d.cargo_ms, "read": d.read_ms, "retry": d.retry_ms,
"units": d.units, "compiled": d.compiled, "fresh": d.fresh, "recompiled": d.recompiled,
})),
},
})
}
/// One feed ledger row as JSON.
pub fn feed_json(r: &FeedRow) -> Value {
json!({ "feed": r.feed, "seq": r.seq, "sha": r.sha, "status": r.status, "rows": r.rows, "ms": r.ms, "detail": r.detail })
}
pub(super) fn short(sha: &str) -> &str {
&sha[..sha.len().min(10)]
}
/// The first line of a commit's `partial` / `error` sentence, for a log line:
/// ` — nothing compiled; failed units: (workspace does not load: …)`. Empty for
/// a commit that has none.
pub fn commit_reason(r: &CommitReport) -> String {
r.error.as_deref().and_then(|e| e.lines().next()).map(|l| format!(" — {l}")).unwrap_or_default()
}
/// Runs of a member that may fail the same way, for a reason outside its
/// commits, before it is [`MemberStatus::Blocked`].
pub const STUCK_AFTER: u32 = 3;
/// A member whose runs keep failing for a reason outside its commits.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct Stuck {
/// The heads of every member when it last failed: a member that moves is
/// a reason to try again.
world: String,
tries: u32,
error: String,
}
/// Every member's head, as one string: what the outcome of a dump can depend
/// on besides the box itself.
fn world_of(members: &[Member]) -> String {
members.iter().map(|m| format!("{}@{}", m.name, m.head)).collect::<Vec<_>>().join(",")
}
/// What one run did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConsumeReport {
pub workspace: String,
/// Which driver dumped (one line).
pub driver: String,
/// The toolchain's std dump, when this run made (or tried) it: one line.
pub std: Option<String>,
pub members: Vec<MemberOutcome>,
/// The dependents of a change compiled to see that they still build
/// ([`Request::affected`]), in build order.
pub checked: Vec<Checked>,
}
/// One dependent of a change, compiled against the changed trees.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Checked {
pub member: String,
/// cargo's exit code; `None`: not compiled (no super build here, or it failed).
pub exit: Option<i32>,
pub ms: u64,
/// Why it was not compiled, when it was not.
pub note: Option<String>,
/// Units that built at the member's last read and fail now: what the
/// change broke ([`judge_check`]).
pub regressions: Vec<String>,
/// Units that fail now and already failed at the member's last read
/// (`failing_since`: that read's seq) — not the change's.
pub already_failing: Vec<String>,
pub failing_since: Option<i64>,
}
impl Checked {
/// The change did not break it: it builds, or only what already failed fails.
pub fn ok(&self) -> bool {
match self.exit {
Some(0) => true,
Some(_) => self.regressions.is_empty() && !self.already_failing.is_empty(),
None => false,
}
}
pub fn line(&self) -> String {
match (self.exit, &self.note) {
(Some(0), _) => format!("{}: builds ({} ms)", self.member, self.ms),
(Some(_), _) if self.ok() => format!(
"{}: builds as before ({} ms) — {} unit(s) were already failing at seq {}: {}",
self.member,
self.ms,
self.already_failing.len(),
self.failing_since.map(|s| s.to_string()).unwrap_or_else(|| "?".into()),
self.already_failing.join(", ")
),
(Some(c), _) if !self.regressions.is_empty() => format!("{}: does NOT build — the change breaks {} (cargo exit {c}, {} ms)", self.member, self.regressions.join(", "), self.ms),
(Some(c), _) => format!("{}: does NOT build (cargo exit {c}, {} ms)", self.member, self.ms),
(None, Some(n)) => format!("{}: not compiled — {n}", self.member),
(None, None) => format!("{}: not compiled", self.member),
}
}
}
/// **Judge a dependent's build against its last known state**: of the units
/// that fail now (`failed_now`, cargo's errored units), the ones its last
/// read already listed as failed (`last`: that ledger row's seq, status and
/// error) were already failing; the rest are regressions.
pub fn judge_check(failed_now: &BTreeSet<String>, last: Option<(i64, &str, Option<&str>)>) -> (Vec<String>, Vec<String>, Option<i64>) {
// what the last read said failed: a partial run's error names them
let (seq, was): (Option<i64>, BTreeSet<String>) = match last {
Some((seq, _status, Some(err))) => match err.split_once("failed units (kept their previous state):") {
Some((_, list)) => (Some(seq), list.split(',').map(|u| u.trim().to_string()).filter(|u| !u.is_empty()).collect()),
None => (None, BTreeSet::new()),
},
_ => (None, BTreeSet::new()),
};
let (already, regressions): (Vec<String>, Vec<String>) = failed_now.iter().cloned().partition(|u| was.contains(u));
let since = if already.is_empty() { None } else { seq };
(regressions, already, since)
}
/// How the dependents of a change ended (`checks`: the ones the super build
/// compiled; `outcome`: its per-repo exit codes).
pub fn check_outcomes(check: &[&Member], checks: &[String], outcome: Option<&mimir_ingest::SuperOutcome>) -> Vec<Checked> {
check
.iter()
.map(|m| {
let ran = outcome.and_then(|o| o.repos.iter().find(|(r, _, _)| *r == m.name));
match (checks.contains(&m.name), ran) {
(true, Some((_, code, ms))) => Checked { member: m.name.clone(), exit: Some(*code), ms: *ms, note: None, ..Default::default() },
(true, None) => Checked { member: m.name.clone(), exit: None, ms: 0, note: Some("the super build did not run it".into()), ..Default::default() },
(false, _) => Checked { member: m.name.clone(), exit: None, ms: 0, note: Some("no super build here: nothing it depends on was compiled beside it".into()), ..Default::default() },
}
})
.collect()
}
/// **The members a change reaches**: `changed` and every member that
/// (transitively) depends on one of them, by Mímir's dependency graph of the
/// members' manifests ([`nornir_depgraph::WorkspaceGraph::affected_by_change`]
/// over `trees`: member → its clone). `Err`: the graph could not be read.
pub fn affected_members(trees: &BTreeMap<String, PathBuf>, changed: &[String]) -> anyhow::Result<Vec<String>> {
let graph = nornir_depgraph::WorkspaceGraph::build_from_members(trees)?;
Ok(graph.affected_by_change(changed))
}
impl ConsumeReport {
/// `member@seq` of every member that has an offset — the registry's
/// `current_snapshot` (what a v1 snapshot id was: "which state is served").
pub fn stamp(&self) -> String {
self.members.iter().filter_map(|m| m.offset.as_ref().map(|o| format!("{}@{}", m.member, o.seq))).collect::<Vec<_>>().join(",")
}
pub fn commits(&self) -> usize {
self.members.iter().map(|m| m.reports.len()).sum()
}
pub fn failed(&self) -> Vec<&MemberOutcome> {
self.members.iter().filter(|m| m.status == MemberStatus::Failed).collect()
}
pub fn behind(&self) -> Vec<&MemberOutcome> {
self.members.iter().filter(|m| matches!(m.status, MemberStatus::Behind | MemberStatus::Cancelled | MemberStatus::Blocked)).collect()
}
pub fn to_json(&self) -> Value {
json!({
"workspace": self.workspace,
"driver": self.driver,
"std": self.std,
"stamp": self.stamp(),
"commits": self.commits(),
"members": self.members.iter().map(MemberOutcome::to_json).collect::<Vec<_>>(),
})
}
}
/// What the consumer is doing right now (read by status, never blocked by it).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InFlight {
pub member: String,
/// The commit being consumed (`None`: fetching / between commits / feeds).
pub commit: Option<(i64, String)>,
/// Commits behind the one in flight.
pub behind: usize,
pub since: Instant,
}
impl InFlight {
/// `skade seq 12 @ 0b07d30f37 (41 s, 3 behind)`
pub fn line(&self) -> String {
let secs = self.since.elapsed().as_secs();
match &self.commit {
Some((seq, sha)) => format!("{} seq {seq} @ {} ({secs} s, {} behind)", self.member, short(sha), self.behind),
None => format!("{} ({secs} s)", self.member),
}
}
}
/// The consumer side of the engine's state.
#[derive(Default)]
pub(super) struct ConsumerState {
/// The booted consumer. Held for a whole run: one run at a time.
consumer: std::sync::Mutex<Option<Pooled>>,
cancel: std::sync::atomic::AtomicBool,
in_flight: std::sync::Mutex<Option<InFlight>>,
/// Why the last attempt to consume could not start (cleared by a run).
refusal: std::sync::Mutex<Option<Refusal>>,
/// Deferred feeds whose work was handed out and whose ledger row has not
/// landed yet: (member, feed, sha). Such a feed is not "behind" and is not
/// prepared a second time.
deferred: std::sync::Mutex<BTreeSet<(String, String, String)>>,
/// The toolchain whose std dump failed in this process, and why: not tried
/// again on every tick — a restart or an explicit refresh tries again.
std_failed: std::sync::Mutex<Option<(String, String)>>,
/// The toolchain whose std is known to be in the ring (asked every tick).
std_settled: std::sync::Mutex<Option<String>>,
/// Members whose runs fail for a reason outside their commits ([`Stuck`]).
stuck: std::sync::Mutex<BTreeMap<String, Stuck>>,
}
impl ConsumerState {
/// Drop the consumer (its writer seat and its handle on the lake). Waits
/// for a run in flight.
pub(super) fn close(&self) {
self.consumer.lock().unwrap_or_else(|e| e.into_inner()).take();
}
fn set_in_flight(&self, v: Option<InFlight>) {
*self.in_flight.lock().unwrap_or_else(|e| e.into_inner()) = v;
}
/// Why `member` is not tried again, when it is blocked: its last
/// [`STUCK_AFTER`] runs failed the same way and no member moved since.
fn blocked(&self, member: &str, world: &str) -> Option<String> {
let stuck = self.stuck.lock().unwrap_or_else(|e| e.into_inner());
let s = stuck.get(member).filter(|s| s.tries >= STUCK_AFTER && s.world == world)?;
Some(format!(
"not tried again: its last {} run(s) failed the same way, for a reason outside its commits — {}. It is tried again when a member of the workspace moves, on `nornir workspace repopulate`, or after a restart",
s.tries, s.error
))
}
/// A member's turn ended: count a failure, forget the count on anything else.
fn note_turn(&self, outcome: &MemberOutcome, world: &str) {
let mut stuck = self.stuck.lock().unwrap_or_else(|e| e.into_inner());
match (&outcome.status, &outcome.error) {
(MemberStatus::Failed, Some(error)) => {
let tries = stuck.get(&outcome.member).filter(|s| s.world == world).map_or(0, |s| s.tries) + 1;
stuck.insert(outcome.member.clone(), Stuck { world: world.to_string(), tries, error: error.clone() });
}
(MemberStatus::Blocked | MemberStatus::Cancelled | MemberStatus::Backfilling, _) => {}
_ => {
stuck.remove(&outcome.member);
}
}
}
fn is_deferred(&self, member: &str, feed: &str, sha: &str) -> bool {
self.deferred.lock().unwrap_or_else(|e| e.into_inner()).contains(&(member.to_string(), feed.to_string(), sha.to_string()))
}
}
/// A feed row whose status is none of `ok` / `error` / `skipped`: the feed
/// answered [`FeedResult::Absent`] (its state is the status).
fn is_absent(row: &FeedRow) -> bool {
![schema::STATUS_OK, schema::STATUS_ERROR, schema::STATUS_SKIPPED].contains(&row.status.as_str())
}
/// Is the feed's last row for this sha due again by itself: it found
/// nothing to read ([`FeedResult::Absent`]), could not run here (`skipped`)
/// or failed (`error`), and the feed says what stood in the way is gone
/// ([`Feed::retry`]).
fn retry_due(feed: &dyn Feed, at: &FeedAt<'_>, row: &FeedRow) -> bool {
row.status != schema::STATUS_OK && feed.retry(at, row)
}
/// The ledger row of a feed run that did not write the binary stream.
fn feed_row(member: &str, seq: i64, sha: &str, feed: &str, ran: anyhow::Result<FeedResult>, t0: Instant) -> FeedRow {
let (status, rows, detail) = match ran {
Ok(FeedResult::Ok { rows, detail }) => (schema::STATUS_OK, rows, Some(detail).filter(|d| !d.is_empty())),
Ok(FeedResult::Skipped(why)) => (schema::STATUS_SKIPPED, 0, Some(why)),
Ok(FeedResult::Absent { state, detail }) => (state, 0, Some(detail)),
Ok(FeedResult::Binaries { .. }) => (schema::STATUS_ERROR, 0, Some("the feed handed binary rows to a place that has no writer seat (a deferred feed cannot write the binary stream)".to_string())),
Err(e) => (schema::STATUS_ERROR, 0, Some(format!("{e:#}"))),
};
FeedRow { repo: member.to_string(), seq, sha: sha.to_string(), feed: feed.to_string(), status: status.to_string(), rows, ms: t0.elapsed().as_millis() as i64, detail }
}
/// `std of nightly-…: 21 crate(s), 184 201 row(s) into the ring in 96 s`
fn std_line(toolchain: &str, o: &StdOutcome, ms: u64) -> String {
match o.status {
StdStatus::Dumped => format!("std of {toolchain}: {} crate(s), {} row(s) into the ring in {} s", o.crates, o.rows, ms / 1000),
StdStatus::Skipped => format!("std of {toolchain}: already in the ring"),
StdStatus::FilesRestored => format!("std of {toolchain}: in the ring; its dump files were made again in {} s", ms / 1000),
}
}
impl CodeRoom {
/// What the consumer is doing right now (`None`: idle).
pub fn in_flight(&self) -> Option<InFlight> {
self.engine.consumer.in_flight.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// Ask the run in flight to stop after the commit it is on.
pub fn cancel_consume(&self) {
self.engine.consumer.cancel.store(true, Ordering::Release);
}
/// Why the last attempt to consume could not start, if it could not.
pub fn last_refusal(&self) -> Option<Refusal> {
self.engine.consumer.refusal.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// Record why a run could not start (the driver is not there, …). Returns
/// `true` when this is news — the same refusal again is not, so a poll
/// loop can say it once.
pub fn note_refusal(&self, refusal: &Refusal) -> bool {
let mut slot = self.engine.consumer.refusal.lock().unwrap_or_else(|e| e.into_inner());
let news = slot.as_ref() != Some(refusal);
*slot = Some(refusal.clone());
news
}
/// The members with an OPEN backfill (`super::backfill`): asked for, not
/// yet walked to the tip.
pub(super) fn backfilling(&self, log: &EventLog) -> Result<BTreeSet<String>, Refusal> {
let last = log
.block_on(log.backfill_last())
.map_err(|e| Refusal::new(State::Internal, format!("code room of `{}`: read the backfills: {e:#}", self.workspace)))?;
Ok(last.into_values().filter(|r| r.is_open()).map(|r| r.repo).collect())
}
/// The members of `members` that are behind AND would be consumed by a run
/// now: no offset, the offset is not the head, or a feed in `feeds` was
/// never tried for the offset's sha (or answered "not there" and now has
/// something to read). A member whose backfill is open is its job's, a
/// deferred feed whose work is out is in flight, and a member that is
/// blocked ([`MemberStatus::Blocked`]) is not tried until something
/// changes: none of the three is in the answer.
pub fn behind(&self, members: &[Member], feeds: &[&dyn Feed]) -> Result<Vec<String>, Refusal> {
let log = self.log()?;
let internal = |e: anyhow::Error| Refusal::new(State::Internal, format!("code room of `{}`: {e:#}", self.workspace));
let offsets = log.block_on(log.offsets()).map_err(internal)?;
let backfilling = self.backfilling(&log)?;
let world = world_of(members);
let mut out = Vec::new();
for m in members.iter().filter(|m| m.has_source() && !backfilling.contains(&m.name) && self.engine.consumer.blocked(&m.name, &world).is_none()) {
let behind = match offsets.get(&m.name) {
None => true,
Some(o) if o.sha != m.head => true,
Some(o) => {
let last = log.block_on(log.feed_last(&m.name)).map_err(internal)?;
let at = FeedAt { workspace: &self.workspace, member: m, seq: o.seq, sha: &o.sha, log: &log, layout: &self.layout };
feeds.iter().any(|f| {
if self.engine.consumer.is_deferred(&m.name, f.name(), &o.sha) {
return false;
}
match last.get(f.name()) {
Some(r) if r.sha == o.sha => retry_due(&**f, &at, r),
_ => true,
}
})
}
};
if behind {
out.push(m.name.clone());
}
}
Ok(out)
}
/// Why `member` is blocked, when it is ([`MemberStatus::Blocked`]).
pub fn blocked(&self, members: &[Member], member: &str) -> Option<String> {
self.engine.consumer.blocked(member, &world_of(members))
}
/// **Is the build board missing a plan** this process could give it? A
/// member with a clone and a manifest that has no plan on the board,
/// with dumpers that plan. A run then plans it (and settles it when it is
/// caught up), also when nothing is behind: once the process has read the
/// workspace's plan the board never answers empty — after a restart too
/// (live dwarf 2026-10-16: the Tree and the Explorer showed "no plan yet"
/// between consume runs; a sweep with nothing behind returned before
/// planning anything).
pub fn board_unplanned(&self, members: &[Member], dumpers: &dyn Dumpers) -> bool {
dumpers.plans() && self.board_gaps(members)
}
/// [`Self::board_unplanned`] without asking the dumpers (no driver probe):
/// a member with a clone and a manifest has no plan on the board, and none
/// was tried at its head in this process (a tree whose plan cargo cannot
/// give is not asked for again every tick).
pub fn board_gaps(&self, members: &[Member]) -> bool {
let board = super::units::board(&self.layout.clones());
let b = board.lock().unwrap_or_else(|e| e.into_inner());
members.iter().filter(|m| m.has_source()).any(|m| !b.has_plan(&m.name) && !b.tried(&m.name, &m.head) && self.layout.clone_of(&m.name).join("Cargo.toml").is_file())
}
/// Is the toolchain's std still to be dumped into this workspace's ring?
/// Only once something was consumed (a lake nobody consumes into needs no
/// ring), and not while a dump that failed in this process stands.
pub fn std_due(&self, dumpers: &dyn Dumpers) -> bool {
let Some(sd) = dumpers.std_dumper(&self.layout) else { return false };
let state = &self.engine.consumer;
if state.std_settled.lock().unwrap_or_else(|e| e.into_inner()).as_deref() == Some(sd.toolchain.as_str()) {
return false;
}
let Ok(log) = self.log() else { return false };
if log.universe_has(mimir_ingest::std_dump::STD_MARKER, &sd.toolchain) {
*state.std_settled.lock().unwrap_or_else(|e| e.into_inner()) = Some(sd.toolchain);
return false;
}
if state.std_failed.lock().unwrap_or_else(|e| e.into_inner()).as_ref().is_some_and(|(tc, _)| *tc == sd.toolchain) {
return false;
}
log.block_on(log.offsets()).is_ok_and(|o| !o.is_empty())
}
/// Is the std of the toolchain this process consumes with known to be in
/// the ring? (`false` until a check found it: the answer costs a probe.)
pub fn std_settled(&self) -> bool {
self.engine.consumer.std_settled.lock().unwrap_or_else(|e| e.into_inner()).is_some()
}
/// The booted consumer, for what else writes through its seat (a backfill
/// row, a deferred feed's ledger row). Waits for a run in flight.
pub(super) fn with_consumer<T>(&self, f: impl FnOnce(&EventLog, &mut Pooled) -> Result<T, Refusal>) -> Result<T, Refusal> {
let log = self.log()?;
let mut slot = self.engine.consumer.consumer.lock().unwrap_or_else(|e| e.into_inner());
self.boot(&log, &mut slot)?;
f(&log, slot.as_mut().expect("booted above"))
}
fn boot(&self, log: &EventLog, slot: &mut Option<Pooled>) -> Result<(), Refusal> {
if slot.is_some() {
return Ok(());
}
let internal = |what: &str, e: anyhow::Error| Refusal::new(State::Internal, format!("code room of `{}`: {what}: {e:#}", self.workspace));
std::fs::create_dir_all(self.layout.base()).map_err(|e| internal("create the state dir", e.into()))?;
// the retired open-set cache (the READ-FAST zone replaced it): a
// leftover base or journal is dead weight, swept once here
for stale in ["open-set.bin", "open-set.journal"] {
let _ = std::fs::remove_file(self.layout.base().join(stale));
}
// the open-set load is one of the process's heavy transients: never
// beside a room build (the room's heavy gate)
let consumer = self
.engine
.heavy
.enter_as("open-set load", || log.block_on(Consumer::boot(log.clone(), DumperPool::new())))
.map_err(|e| internal("boot the consumer", e))?;
*slot = Some(consumer);
Ok(())
}
/// Consume every member of `req` that is behind, then run the feeds of
/// the members that are caught up. One run at a time per workspace (a
/// second caller waits); readers are never blocked.
///
/// SYNC and long (a `cargo check` per commit): call it from a thread that
/// may block, never from an async task.
///
/// A member's failure does not stop the others: it is in its outcome (and
/// in the ledger). The `Err` is for what stops the whole run before it
/// starts — the lake is held elsewhere, the dumper cannot be made.
pub fn consume(&self, dumpers: &dyn Dumpers, feeds: &[&dyn Feed], req: &Request<'_>, on: &mut dyn FnMut(Event<'_>)) -> Result<ConsumeReport, Refusal> {
let state = &self.engine.consumer;
let mut slot = state.consumer.lock().unwrap_or_else(|e| e.into_inner());
state.cancel.store(false, Ordering::Release);
let run = self.consume_locked(&mut slot, dumpers, feeds, req, on);
state.set_in_flight(None);
match &run {
Ok(_) => *state.refusal.lock().unwrap_or_else(|e| e.into_inner()) = None,
Err(r) => {
self.note_refusal(r);
}
}
run
}
fn consume_locked(
&self,
slot: &mut Option<Pooled>,
dumpers: &dyn Dumpers,
feeds: &[&dyn Feed],
req: &Request<'_>,
on: &mut dyn FnMut(Event<'_>),
) -> Result<ConsumeReport, Refusal> {
let log = self.log()?;
self.boot(&log, slot)?;
let consumer = slot.as_mut().expect("booted above");
let on = RefCell::new(on);
let selected: Vec<&Member> = req.members.iter().filter(|m| req.only.is_none_or(|o| o == m.name)).collect();
if let Some(only) = req.only
&& selected.is_empty()
{
return Err(Refusal::new(
State::InvalidArgument,
format!("`{only}` is not a member of `{}` (members: {})", self.workspace, req.members.iter().map(|m| m.name.as_str()).collect::<Vec<_>>().join(", ")),
));
}
// every member's clone is there AND FETCHED before ANY member is
// dumped: a member's manifest names its siblings by path (`../edda`),
// a clone made only on its own turn does not exist yet for the members
// before it, and one fetched only on its own turn does not hold the
// commit a sibling's new commit needs. A clone that has its member's
// head needs no fetch. One that cannot be made here is its member's
// own failure, said on that member's turn.
// A clone that cannot be made is said as a failed checkout — on its
// member's turn and on every member that reaches into it (never as a
// sibling that is missing: loki 2026-10-17).
let mut clone_failed: BTreeMap<String, String> = BTreeMap::new();
for m in req.members.iter().filter(|m| m.has_source()) {
let worktree = self.layout.clone_of(&m.name);
if worktree.join(".git").exists() && git::has_commit(&worktree, &m.head) {
continue;
}
if let Some(parent) = worktree.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Err(e) = git::sync_clone(&m.source.to_string_lossy(), &worktree) {
clone_failed.insert(m.name.clone(), format!("{e:#}"));
}
}
// `only: Affected`: the members behind are read; the members that
// depend on them are compiled (checked); nothing else is touched
let (selected, check): (Vec<&Member>, Vec<&Member>) = if req.affected && req.from.is_none() {
let behind: Vec<&Member> =
selected.iter().copied().filter(|m| m.has_source() && log.block_on(log.offset(&m.name)).ok().flatten().is_none_or(|o| o.sha != m.head)).collect();
let changed: Vec<String> = behind.iter().map(|m| m.name.clone()).collect();
let trees: BTreeMap<String, PathBuf> = req.members.iter().filter(|m| m.has_source()).map(|m| (m.name.clone(), self.layout.clone_of(&m.name))).collect();
let affected = if changed.is_empty() {
Vec::new()
} else {
affected_members(&trees, &changed).unwrap_or_else(|e| {
// no graph: every member may be reached
eprintln!("nornir-consumer: {}: the dependency graph of the members did not read ({e:#}); every member is checked", self.workspace);
trees.keys().cloned().collect()
})
};
let check: Vec<&Member> = affected
.iter()
.filter(|a| !changed.contains(a))
.filter_map(|a| selected.iter().copied().find(|m| &m.name == a))
.filter(|m| self.layout.clone_of(&m.name).join("Cargo.toml").is_file())
.collect();
eprintln!(
"nornir-consumer: {}: affected build — changed: {}; dependents compiled: {}",
self.workspace,
if changed.is_empty() { "none".to_string() } else { changed.join(", ") },
if check.is_empty() { "none".to_string() } else { check.iter().map(|m| m.name.as_str()).collect::<Vec<_>>().join(", ") }
);
(behind, check)
} else {
(selected, Vec::new())
};
// the build board: every member of this run is on it, and the plan of
// each that is behind is read now — the whole tree before any build
{
let board = super::units::board(&self.layout.clones());
for m in selected.iter().filter(|m| m.has_source()) {
let offset = log.block_on(log.offset(&m.name)).ok().flatten();
let worktree = self.layout.clone_of(&m.name);
if offset.as_ref().is_some_and(|o| o.sha == m.head) && req.from.is_none() {
// caught up: on the board too — its plan read ONCE per
// process (the tree and the super build need every
// member's units), every unit in the log
let planned = board.lock().unwrap_or_else(|e| e.into_inner()).has_plan(&m.name);
if !planned && worktree.join("Cargo.toml").is_file() {
board.lock().unwrap_or_else(|e| e.into_inner()).try_plan(&m.name, &m.head);
if !consumer.dumper_mut().contains(&worktree)
&& let Ok(dumper) = dumpers.dumper(&self.layout, m)
{
let _ = consumer.dumper_mut().insert(worktree.clone(), dumper);
}
consumer.dumper_mut().preplan(&worktree);
}
board.lock().unwrap_or_else(|e| e.into_inner()).settle(&m.name, &m.head);
continue;
}
if !worktree.join("Cargo.toml").is_file() {
continue;
}
if !consumer.dumper_mut().contains(&worktree)
&& let Ok(dumper) = dumpers.dumper(&self.layout, m)
{
let _ = consumer.dumper_mut().insert(worktree.clone(), dumper);
}
board.lock().unwrap_or_else(|e| e.into_inner()).queue(&m.name);
consumer.dumper_mut().preplan(&worktree);
}
}
// the dependents to compile: their dumpers and plans (keys, waits)
for m in &check {
let worktree = self.layout.clone_of(&m.name);
if !consumer.dumper_mut().contains(&worktree)
&& let Ok(dumper) = dumpers.dumper(&self.layout, m)
{
let _ = consumer.dumper_mut().insert(worktree.clone(), dumper);
}
consumer.dumper_mut().preplan(&worktree);
}
// the PLANNER's order: the members whose units others share go first,
// so the shared leaves are compiled once, early, into the one target dir
// (a stable sort: members with no plan, and ties, keep the descriptor's order)
let selected: Vec<&Member> = {
let board = super::units::board(&self.layout.clones());
let b = board.lock().unwrap_or_else(|e| e.into_inner());
let mut s = selected;
s.sort_by_key(|m| std::cmp::Reverse(b.shared_count(&m.name).unwrap_or(0)));
s
};
// the SUPER BUILD: every member whose next commit is its head (a cold
// read, or one new commit each) compiled AT ONCE — one target dir,
// cargo's fine-grain locking, one jobserver — before the members are
// read one by one from what it left
let super_job = if req.from.is_none() && super_enabled() { self.super_prepare(consumer, &log, &selected, &check, dumpers) } else { None };
let checks: Vec<String> = super_job.as_ref().map(|j| j.checks.clone()).unwrap_or_default();
// the super build compiles BESIDE the reads: a member is read as soon
// as its repo's record is there (its dump waits for it)
std::thread::scope(|scope| {
let super_thread = super_job.map(|job| {
let done = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
consumer.dumper_mut().super_watch(std::sync::Arc::new(job.outs.clone()), done.clone());
let (board, ws, job_dir) = (super::units::board(&self.layout.clones()), self.workspace.clone(), self.layout.base().join("super"));
scope.spawn(move || {
let ran = mimir_ingest::run_super(&job.host, &job.tool, &job_dir, &job.entries, job.jobs, job.parallel, |repo, line| {
board.lock().unwrap_or_else(|e| e.into_inner()).cargo_line(repo, line);
});
done.store(true, std::sync::atomic::Ordering::SeqCst);
let outcome = ran.as_ref().ok().cloned();
match ran {
Ok(o) => eprintln!(
"nornir-consumer: {ws}: super build done in {} ms — {}",
o.ms,
o.repos.iter().map(|(r, c, ms)| format!("{r} {} {} s", if *c == 0 { "ok".to_string() } else { format!("exit {c}") }, ms / 1000)).collect::<Vec<_>>().join(", ")
),
Err(e) => eprintln!("nornir-consumer: {ws}: the super build failed — {e:#}; every member compiles on its own turn"),
}
outcome
})
});
// the ring's toolchain half first: a member's feeds (the binary hook)
// join against it
let std = if selected.iter().any(|m| m.has_source()) { self.ensure_std(consumer, &log, dumpers, req.retry_feeds, &on) } else { None };
let backfilling = self.backfilling(&log)?;
let world = world_of(req.members);
let mut members = Vec::new();
for member in selected {
let t0 = Instant::now();
let offset_of = || log.block_on(log.offset(&member.name)).ok().flatten();
let mut outcome = if self.engine.consumer.cancel.load(Ordering::Acquire) {
Ok(MemberOutcome { status: MemberStatus::Cancelled, ..blank(member) })
} else if backfilling.contains(&member.name) && req.only != Some(member.name.as_str()) {
// its backfill's job walks it, one commit per run
Ok(MemberOutcome { status: MemberStatus::Backfilling, offset: offset_of(), ..blank(member) })
} else if let Some(why) = self.engine.consumer.blocked(&member.name, &world).filter(|_| !req.retry_feeds) {
// it failed the same way too often: not again until something changes
Ok(MemberOutcome { status: MemberStatus::Blocked, offset: offset_of(), error: Some(why), ..blank(member) })
} else {
self.consume_member(consumer, &log, dumpers, feeds, member, req, &on)
}?;
if let Some(why) = clone_failed.get(&member.name) {
outcome.status = MemberStatus::Failed;
outcome.error = Some(format!("checkout failed: {why}"));
} else if let Some(err) = outcome.error.as_deref()
&& let Some((sibling, why)) = clone_failed.iter().find(|(s, _)| err.contains(self.layout.clone_of(s).to_string_lossy().as_ref()))
{
outcome.error = Some(format!("checkout failed: the clone of {sibling} — {why}"));
}
self.engine.consumer.note_turn(&outcome, &world);
outcome.ms = t0.elapsed().as_millis() as u64;
(on.borrow_mut())(Event::Done { outcome: &outcome });
members.push(outcome);
}
// the system boundary: what the run brought into the universe gets its
// map (once per owner), the table and the kernel once
if let Some(line) = self.ensure_boundary(consumer, &log, dumpers) {
eprintln!("nornir-consumer: {}: {line}", self.workspace);
}
// the dependents compiled beside the reads: how their builds ended
let outcome = super_thread.and_then(|h| h.join().ok()).flatten();
let mut checked = check_outcomes(&check, &checks, outcome.as_ref());
// a dependent that does not build: what failed, against its last read
for c in checked.iter_mut().filter(|c| c.exit.is_some_and(|e| e != 0)) {
let tree = self.layout.clone_of(&c.member);
let json = std::fs::read_to_string(self.layout.out_of(&c.member).join(mimir_ingest::driver::CHECK_LABEL).join(mimir_ingest::driver::SUPER_DIR).join("cargo.json")).unwrap_or_default();
let failed_now = mimir_ingest::driver::cargo_build(&json, &tree).map(|b| b.errored).unwrap_or_default();
let runs = log.block_on(log.runs(&c.member)).unwrap_or_default();
let last = runs.iter().max_by_key(|r| r.seq).map(|r| (r.seq, r.status.as_str(), r.error.as_deref()));
(c.regressions, c.already_failing, c.failing_since) = judge_check(&failed_now, last);
}
{
let board = super::units::board(&self.layout.clones());
let mut b = board.lock().unwrap_or_else(|e| e.into_inner());
for c in &checked {
if let Some(m) = check.iter().find(|m| m.name == c.member) {
b.phase(&m.name, if c.ok() { super::units::Phase::Landed } else { super::units::Phase::Failed }, &m.head);
}
}
}
for c in &checked {
eprintln!("nornir-consumer: {}: checked {}", self.workspace, c.line());
}
Ok(ConsumeReport { workspace: self.workspace.clone(), driver: dumpers.describe(), std, members, checked })
})
}
/// **The super build** (dv2-b1-super): every qualifying member — its next
/// commit is its head — gets its line of ONE job (all repos compiled at
/// once over one target dir, run beside the reads). A member that does not
/// qualify, a dumper that does not take part: that member compiles on its
/// own turn, as before. `None`: fewer than two members, or no host.
fn super_prepare(&self, consumer: &mut Pooled, log: &EventLog, members: &[&Member], check: &[&Member], dumpers: &dyn Dumpers) -> Option<SuperJob> {
let (host, tool) = dumpers.super_host()?;
let mut entries = Vec::new();
let mut left_out: Vec<String> = Vec::new();
for m in members.iter().filter(|m| m.has_source() && !m.head.is_empty()) {
let worktree = self.layout.clone_of(&m.name);
if !worktree.join("Cargo.toml").is_file() || !consumer.dumper_mut().contains(&worktree) {
continue;
}
// the commit this member's turn will read; a member that does not
// take part is said, with why (never dropped silently)
let next = match log.block_on(log.offset(&m.name)).ok().flatten() {
Some(o) if o.sha == m.head => continue,
Some(o) => match git::first_parent_since(&worktree, &o.branch, Some(&o.sha)) {
Ok(c) if c.len() == 1 => c[0].clone(),
Ok(c) => {
left_out.push(format!("{} ({} commits behind: read one by one)", m.name, c.len()));
continue;
}
Err(e) => {
left_out.push(format!("{} (its commits since {} on {} do not read: {e:#})", m.name, &o.sha[..o.sha.len().min(10)], o.branch));
continue;
}
},
None => match m.tracked_branch().map(|(b, _)| git::tip(&worktree, &b)) {
Some(Ok(tip)) => tip,
_ => {
left_out.push(format!("{} (no branch tip to start from)", m.name));
continue;
}
},
};
if next != m.head {
left_out.push(format!("{} (the clone's branch is at {}, the member's head is {})", m.name, &next[..next.len().min(10)], &m.head[..m.head.len().min(10)]));
continue;
}
if let Err(e) = checkout::checkout_changed(&worktree, &next) {
left_out.push(format!("{} (checkout of {} failed: {e:#})", m.name, &next[..next.len().min(10)]));
continue;
}
match consumer.dumper_mut().super_entry(&worktree, &next) {
Ok(Some(e)) => entries.push(e),
Ok(None) => {}
Err(e) => eprintln!("nornir-consumer: {}: {} is left out of the super build: {e:#}", self.workspace, m.name),
}
}
// the dependents of a change, compiled after (waiting for) the units
// of the trees they depend on — only beside a read in this build: a
// tree read on its own turn walks its commits one by one
let mut checks = Vec::new();
if !entries.is_empty() {
for m in check {
let worktree = self.layout.clone_of(&m.name);
match consumer.dumper_mut().super_check_entry(&worktree) {
Ok(Some(e)) => {
checks.push(e.repo.clone());
entries.push(e);
}
Ok(None) => {}
Err(e) => eprintln!("nornir-consumer: {}: {} is not checked: {e:#}", self.workspace, m.name),
}
}
}
if !left_out.is_empty() {
eprintln!("nornir-consumer: {}: left out of the super build (compiled on their own turn): {}", self.workspace, left_out.join("; "));
}
if entries.len() < 2 {
return None;
}
let jobs = super_jobs(std::env::var("CARGO_BUILD_JOBS").ok(), std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4));
let parallel = std::env::var("NORNIR_MIMIR_SUPER_REPOS").ok().and_then(|v| v.parse().ok()).unwrap_or(0);
eprintln!(
"nornir-consumer: {}: SUPER BUILD of {} repo(s) at once ({jobs} jobs, one target dir): {}",
self.workspace,
entries.len(),
entries.iter().map(|e| e.repo.as_str()).collect::<Vec<_>>().join(", ")
);
let board = super::units::board(&self.layout.clones());
for e in &entries {
board.lock().unwrap_or_else(|e| e.into_inner()).phase(&e.repo, super::units::Phase::Compiling, "");
}
let outs = entries.iter().map(|e| e.out.clone()).collect();
Some(SuperJob { entries, checks, outs, host, tool, jobs, parallel })
}
/// The toolchain's own crates into the universe ring, once per toolchain
/// per lake ([`StdDumper`]): a `cargo check -Zbuild-std` of a trivial crate
/// under the driver, then ONE release. Minutes, once. A failure is said
/// (the line, the event) and does not stop the members: the ring then lacks
/// std until a later try — a restart, or an explicit refresh (`retry`).
fn ensure_std(&self, consumer: &mut Pooled, log: &EventLog, dumpers: &dyn Dumpers, retry: bool, on: &RefCell<&mut dyn FnMut(Event<'_>)>) -> Option<String> {
let sd = dumpers.std_dumper(&self.layout)?;
let state = &self.engine.consumer;
let tc = sd.toolchain.as_str();
if log.universe_has(mimir_ingest::std_dump::STD_MARKER, tc) {
*state.std_settled.lock().unwrap_or_else(|e| e.into_inner()) = Some(tc.to_string());
return None;
}
if !retry
&& let Some((failed, why)) = state.std_failed.lock().unwrap_or_else(|e| e.into_inner()).as_ref()
&& failed == tc
{
return Some(format!("std of {tc}: NOT in the ring — its dump failed earlier in this process ({why}); `workspace repopulate` tries again"));
}
state.set_in_flight(Some(InFlight { member: format!("std@{tc}"), commit: None, behind: 0, since: Instant::now() }));
(on.borrow_mut())(Event::StdBegin { toolchain: tc });
let t0 = Instant::now();
let ran = log.block_on(sd.ensure_seated(consumer.writer()));
let ms = t0.elapsed().as_millis() as u64;
match ran {
Ok(o) => {
*state.std_failed.lock().unwrap_or_else(|e| e.into_inner()) = None;
*state.std_settled.lock().unwrap_or_else(|e| e.into_inner()) = Some(tc.to_string());
(on.borrow_mut())(Event::Std { toolchain: tc, outcome: Ok(&o), ms });
Some(std_line(tc, &o, ms))
}
Err(e) => {
let why = format!("{e:#}");
(on.borrow_mut())(Event::Std { toolchain: tc, outcome: Err(&why), ms });
let line = format!("std of {tc}: NOT in the ring — the dump failed after {} s: {why}", ms / 1000);
*state.std_failed.lock().unwrap_or_else(|e| e.into_inner()) = Some((tc.to_string(), why));
Some(line)
}
}
}
/// The system boundary into the universe ([`mimir_ingest::boundary`]):
/// the target's syscall table and the kernel's BTF once, a map per
/// dependency owner the universe gained. Nothing new = a manifest read,
/// no write. A failure is said, never fatal to the run.
fn ensure_boundary(&self, consumer: &mut Pooled, log: &EventLog, dumpers: &dyn Dumpers) -> Option<String> {
let (target, kernel) = dumpers.boundary()?;
let toolchain = dumpers.std_dumper(&self.layout).map_or_else(|| "boundary".to_string(), |s| s.toolchain);
match log.block_on(mimir_ingest::boundary::ensure(consumer.writer(), &target, &toolchain, kernel.as_ref())) {
Ok(o) if o.table_written.is_none() && o.maps_written == 0 && !o.kernel.as_ref().is_some_and(|k| k.written) => None,
Ok(o) => Some(o.line()),
Err(e) => Some(format!("boundary {}: NOT written — {e:#}", target.triple)),
}
}
/// One member's turn. `Err` only for what stops the whole run.
#[allow(clippy::too_many_arguments)]
fn consume_member(
&self,
consumer: &mut Pooled,
log: &EventLog,
dumpers: &dyn Dumpers,
feeds: &[&dyn Feed],
member: &Member,
req: &Request<'_>,
on: &RefCell<&mut dyn FnMut(Event<'_>)>,
) -> Result<MemberOutcome, Refusal> {
let state = &self.engine.consumer;
let mut out = blank(member);
if !member.has_source() {
out.status = MemberStatus::NoSource;
out.error = Some(format!("no clone at {} yet", member.source.display()));
return Ok(out);
}
state.set_in_flight(Some(InFlight { member: member.name.clone(), commit: None, behind: 0, since: Instant::now() }));
// a member's own failure is its outcome, not the run's
let failed = |mut out: MemberOutcome, what: &str, e: anyhow::Error| {
out.status = MemberStatus::Failed;
out.error = Some(format!("{what}: {e:#}"));
out
};
let offset = match log.block_on(log.offset(&member.name)) {
Ok(o) => o,
Err(e) => return Ok(failed(out, "read the offset", e)),
};
out.offset = offset.clone();
let caught_up = offset.as_ref().is_some_and(|o| o.sha == member.head);
if !caught_up || req.from.is_some() {
let worktree = self.layout.clone_of(&member.name);
if let Some(parent) = worktree.parent()
&& let Err(e) = std::fs::create_dir_all(parent)
{
return Ok(failed(out, "create the clone dir", e.into()));
}
// a member's stream is the first-parent history of ONE branch:
// the one its offset was consumed on, else the one something names
let set = format!("set `branch = \"…\"` for `{}` in the workspace descriptor and register the workspace again", member.name);
let no_branch = |mut out: MemberOutcome, why: String| {
out.status = MemberStatus::NoBranch;
out.error = Some(format!("{why} — {set}"));
out
};
let (branch, named_by) = match (&offset, member.tracked_branch()) {
(Some(o), _) => (o.branch.clone(), "the branch its offset was consumed on"),
(None, Some(named)) => named,
(None, None) => {
let why = format!(
"`{}` has no branch to follow: its clone at {} is on a detached HEAD, the workspace descriptor names no branch for it and its remote names no default branch",
member.name,
member.source.display()
);
return Ok(no_branch(out, why));
}
};
if let Err(e) = git::sync_clone(&member.source.to_string_lossy(), &worktree) {
return Ok(failed(out, "fetch the consumer's clone", e));
}
// decision D2: a member with no offset starts at the tip — or at
// the commit `--from` names; one with an offset replays from it
let mut spec = RepoSpec { name: member.name.clone(), source: member.source.to_string_lossy().into_owned(), branch, worktree: worktree.clone(), initial_offset: None };
let mut replay_from = None;
let start = match req.from {
Some(from) => git::resolve_from(&worktree, &spec.branch, from),
None => git::tip(&worktree, &spec.branch),
};
match (start, &offset) {
(Err(e), _) if e.downcast_ref::<git::NoSuchBranch>().is_some() => {
let why = format!(
"`{}` is to follow branch `{}` ({named_by}), and its clone at {} has no such branch",
member.name,
spec.branch,
member.source.display()
);
return Ok(no_branch(out, why));
}
(Err(e), _) => return Ok(failed(out, "resolve where to start", e)),
(Ok(first), None) => match git::offset_before(&worktree, &first) {
Ok(before) => spec.initial_offset = before,
Err(e) => return Ok(failed(out, "resolve where to start", e)),
},
(Ok(first), Some(_)) if req.from.is_some() => replay_from = Some(first),
(Ok(_), Some(_)) => {}
}
// the member is dumped with every sibling at the tip of the branch
// it follows (see the module docs)
self.siblings_at_tip(req.members, member);
if !consumer.dumper_mut().contains(&worktree) {
let dumper = dumpers.dumper(&self.layout, member)?;
if let Err(e) = consumer.dumper_mut().insert(worktree.clone(), dumper) {
return Ok(failed(out, "attach the dumper", e));
}
}
let name = member.name.as_str();
let mut began = false;
let mut on_begin = |seq: i64, sha: &str, behind: usize| {
if !std::mem::replace(&mut began, true) {
(on.borrow_mut())(Event::Member { member: name, behind: behind + 1 });
}
state.set_in_flight(Some(InFlight { member: name.to_string(), commit: Some((seq, sha.to_string())), behind, since: Instant::now() }));
(on.borrow_mut())(Event::Begin { member: name, seq, sha, behind });
};
let mut on_commit = |r: &CommitReport| (on.borrow_mut())(Event::Commit { member: name, report: r });
// the member's history was rewritten under its offset (a force-push,
// a forge move): it is read again from its head as a fresh member
let rewritten = std::cell::Cell::new(false);
let mut on_rewritten = |r: &git::HistoryRewritten| {
rewritten.set(true);
eprintln!(
"nornir-consumer: {}: {name}: history rewritten — offset {} is not on `{}` any more; reading it again from its head {} as a fresh member",
self.workspace,
short(&r.offset),
r.branch,
short(&r.head)
);
};
let run = Run {
max_commits: req.max_commits,
cancel: Some(&state.cancel),
on_begin: Some(&mut on_begin),
on_commit: Some(&mut on_commit),
replay_from,
on_rewritten: Some(&mut on_rewritten),
};
match log.block_on(consumer.consume_run(&spec, run)) {
Ok(consumed) => {
out.behind = consumed.behind;
out.status = match consumed.stopped {
Stopped::CaughtUp => MemberStatus::CaughtUp,
Stopped::Bound => MemberStatus::Behind,
Stopped::Cancelled => MemberStatus::Cancelled,
};
out.reports = consumed.reports;
}
// the ledger has the error row; the offset is on the last good commit
Err(e) => {
// the typed error says "history rewritten" itself; a failed
// read from the head after one does not, so the line does
let what = if rewritten.get() && e.downcast_ref::<git::HistoryRewritten>().is_none() { "history rewritten" } else { "consume" };
out = failed(out, what, e)
}
}
state.set_in_flight(Some(InFlight { member: member.name.clone(), commit: None, behind: out.behind, since: Instant::now() }));
match log.block_on(log.offset(&member.name)) {
Ok(o) => out.offset = o,
Err(e) => return Ok(failed(out, "read the offset", e)),
}
} else {
out.status = MemberStatus::CaughtUp;
}
if out.status == MemberStatus::CaughtUp
&& let Some(at) = out.offset.clone()
{
// the feeds read the server's clone: only when it IS the offset
if at.sha == member.head {
out.feeds = self.run_feeds(consumer, log, feeds, member, &at, req.retry_feeds, on);
}
}
Ok(out)
}
/// Put every OTHER member's consumer clone at the commit its server clone
/// is at — the tip of the branch it follows — before `member` is dumped.
/// A sibling's clone is otherwise wherever its own consumer left it: at
/// its offset, which is behind its tip whenever the sibling has moved and
/// was not consumed yet (it comes later in the order, its budget ran out,
/// a backfill is walking it). Only what differs is written; a sibling
/// whose clone lacks the commit is left as it is (its fetch failed: said
/// on its own turn). The sibling's own next step checks out the commit it
/// consumes, from wherever its tree is.
fn siblings_at_tip(&self, members: &[Member], member: &Member) {
for sibling in members.iter().filter(|m| m.name != member.name && m.has_source() && !m.head.is_empty()) {
let worktree = self.layout.clone_of(&sibling.name);
if !worktree.join(".git").exists() || git::head(&worktree).as_deref() == Some(sibling.head.as_str()) || !git::has_commit(&worktree, &sibling.head) {
continue;
}
if let Err(e) = checkout::checkout_changed(&worktree, &sibling.head) {
eprintln!("nornir-consumer: {}/{}: its sibling {} could not be put at its tip {}: {e:#}", self.workspace, member.name, sibling.name, short(&sibling.head));
}
}
}
/// Every feed whose own offset is not the member's: run it, write its
/// ledger row. A feed's failure is its row — it never stops the others.
/// `retry`: also the feeds whose row for this sha is not `ok`. A deferred
/// feed ([`Feed::prepare`]) only reads here; its work goes out as
/// [`Event::Deferred`] and its row lands when that work is done.
#[allow(clippy::too_many_arguments)]
fn run_feeds(&self, consumer: &mut Pooled, log: &EventLog, feeds: &[&dyn Feed], member: &Member, at: &Offset, retry: bool, on: &RefCell<&mut dyn FnMut(Event<'_>)>) -> Vec<FeedRow> {
let state = &self.engine.consumer;
let last: BTreeMap<String, FeedRow> = match log.block_on(log.feed_last(&member.name)) {
Ok(d) => d,
Err(e) => {
eprintln!("nornir-consumer: {}/{}: feed ledger unreadable, feeds not run: {e:#}", self.workspace, member.name);
return Vec::new();
}
};
let feed_at = FeedAt { workspace: &self.workspace, member, seq: at.seq, sha: &at.sha, log, layout: &self.layout };
let mut rows = Vec::new();
for feed in feeds {
let settled = last.get(feed.name()).is_some_and(|r| r.sha == at.sha && (r.status == schema::STATUS_OK || !retry) && !retry_due(*feed, &feed_at, r));
if settled || state.cancel.load(Ordering::Acquire) || state.is_deferred(&member.name, feed.name(), &at.sha) {
continue;
}
state.set_in_flight(Some(InFlight { member: format!("{} feed {}", member.name, feed.name()), commit: None, behind: 0, since: Instant::now() }));
let t0 = Instant::now();
let ran = match feed.prepare(&feed_at) {
Some(Ok(work)) => {
state.deferred.lock().unwrap_or_else(|e| e.into_inner()).insert((member.name.clone(), feed.name().to_string(), at.sha.clone()));
(on.borrow_mut())(Event::Deferred(Deferred { member: member.name.clone(), seq: at.seq, sha: at.sha.clone(), feed: feed.name(), work }));
continue;
}
Some(Err(e)) => Err(e),
None => feed.run(&feed_at),
};
let row = match ran {
// the binary stream: the consumer's seat writes it
Ok(FeedResult::Binaries { dumps, declared_bins, detail }) => match log.block_on(consumer.ingest_binaries(&member.name, at, &dumps, &declared_bins)) {
Ok(written) => {
let per: Vec<String> = written.iter().map(|r| format!("{} {} +{} −{}", r.artifact, r.status, r.rows_plus, r.rows_minus)).collect();
let n = written.iter().map(|r| r.rows_plus + r.rows_minus).sum();
feed_row(&member.name, at.seq, &at.sha, feed.name(), Ok(FeedResult::Ok { rows: n, detail: format!("{detail} — {}", per.join("; ")) }), t0)
}
Err(e) => feed_row(&member.name, at.seq, &at.sha, feed.name(), Err(e.context("write the binary stream")), t0),
},
other => feed_row(&member.name, at.seq, &at.sha, feed.name(), other, t0),
};
if let Err(e) = log.block_on(consumer.writer().record_feed(&row)) {
eprintln!("nornir-consumer: {}/{}: feed `{}` ran but its ledger row failed: {e:#}", self.workspace, member.name, row.feed);
continue;
}
(on.borrow_mut())(Event::Feed { member: &member.name, row: &row });
rows.push(row);
}
rows
}
/// Run a deferred feed's work ([`Event::Deferred`]) and write its ledger
/// row. The work runs on the calling thread with NO lock of the code room
/// held — a consume goes on beside it; only the row's write waits for the
/// consumer (one commit at most on a server's sweep, never the reverse).
///
/// SYNC and as long as the work (embeddings: minutes).
pub fn run_deferred(&self, d: Deferred) -> Result<FeedRow, Refusal> {
let Deferred { member, seq, sha, feed, work } = d;
let t0 = Instant::now();
// a panic in the work must still free the (member, feed, sha) slot
let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(work)).unwrap_or_else(|_| Err(anyhow::anyhow!("the feed's work panicked")));
let row = feed_row(&member, seq, &sha, feed, ran, t0);
let wrote = self.with_consumer(|log, consumer| {
log.block_on(consumer.writer().record_feed(&row))
.map_err(|e| Refusal::new(State::Internal, format!("code room of `{}`: feed `{feed}` of `{member}` ran but its ledger row failed: {e:#}", self.workspace)))
});
self.engine.consumer.deferred.lock().unwrap_or_else(|e| e.into_inner()).remove(&(member.clone(), feed.to_string(), sha.clone()));
wrote.map(|()| row)
}
/// Give up a deferred feed's work without running it (its unit was
/// superseded): the (member, feed, sha) is due again.
pub fn drop_deferred(&self, d: &Deferred) {
self.engine.consumer.deferred.lock().unwrap_or_else(|e| e.into_inner()).remove(&(d.member.clone(), d.feed.to_string(), d.sha.clone()));
}
}
impl CodeRoom {
/// **The dry-run delta** (wiring map I3, decision D3): what `tree` — a
/// working copy of `member` that is NOT a commit, dirty or not — would
/// change in the code graph, against the last release the consumer
/// committed for that member. Nothing is written: no event, no offset, no
/// ledger row, and `tree` is only read.
///
/// The nightly driver dumps EVERY local unit of the tree (a dumper of its
/// own under `<builds>/mimir_v2/dry/<member>/`; the first run compiles the
/// dependencies there, the next ones the members only), then the dump is
/// diffed as the consumer would diff a commit. It needs no consumer and
/// does not wait for one: it reads a pinned state of the lake.
///
/// SYNC and as long as a `cargo check` of the members. A member the log
/// never consumed is [`State::NotConsumed`]: against nothing, everything
/// would be "added", which is not an answer.
pub fn dry_run(&self, driver: &DriverReady, member: &str, tree: &Path) -> Result<mimir_ingest::DryDelta, Refusal> {
let _one = self.engine.dry.lock().unwrap_or_else(|e| e.into_inner());
let state = self.layout.dry_of(member);
let mut dumper = driver.spec.dumper(state.join("target"), state.join("out"), member.to_string());
self.dry_delta_of(member, || dumper.dump_tree(tree))
}
/// [`Self::dry_run`] with the dump step given (the diff half, on its own).
pub fn dry_delta_of(&self, member: &str, dump: impl FnOnce() -> anyhow::Result<mimir_ingest::CommitDump>) -> Result<mimir_ingest::DryDelta, Refusal> {
let log = self.log()?;
let internal = |what: &str, e: anyhow::Error| Refusal::new(State::Internal, format!("dry run of `{member}` in `{}`: {what}: {e:#}", self.workspace));
// against ONE committed state, whatever lands while the tree compiles
let pinned = log.block_on(log.pin()).map_err(|e| internal("pin the lake", e))?;
if log.block_on(pinned.offset(member)).map_err(|e| internal("read the offset", e))?.is_none() {
return Err(Refusal::new(
State::NotConsumed,
format!("dry run of `{member}` in `{}`: the member was never consumed — there is no state to diff the tree against", self.workspace),
));
}
let dump = dump().map_err(|e| internal("dump the tree", e))?;
log.block_on(pinned.dry_delta(member, &dump)).map_err(|e| internal("diff against the open state", e))
}
}
fn blank(member: &Member) -> MemberOutcome {
MemberOutcome {
member: member.name.clone(),
status: MemberStatus::CaughtUp,
head: member.head.clone(),
offset: None,
behind: 0,
reports: Vec::new(),
feeds: Vec::new(),
error: None,
ms: 0,
}
}
/// Is the super build on? (`NORNIR_MIMIR_SUPER=off` turns it off: every
/// member compiles on its own turn.)
fn super_enabled() -> bool {
!matches!(std::env::var("NORNIR_MIMIR_SUPER").as_deref().map(str::trim), Ok("off" | "0" | "false"))
}
/// One super build, ready to run ([`CodeRoom::super_prepare`]).
struct SuperJob {
entries: Vec<mimir_ingest::SuperEntry>,
/// The repos of `entries` compiled only (dependents of a change).
checks: Vec<String>,
outs: Vec<PathBuf>,
host: mimir_ingest::DriverHost,
tool: PathBuf,
jobs: usize,
parallel: usize,
}
/// The super build's cargo jobs: the host's bound (`CARGO_BUILD_JOBS`) when
/// it names one, else one per core. Live on dwarf the super build ran 12 jobs
/// beside a server already near its host's memory while every other compile
/// there was held to 4.
pub fn super_jobs(bound: Option<String>, cores: usize) -> usize {
bound.and_then(|j| j.trim().parse::<usize>().ok()).filter(|j| *j > 0).unwrap_or(cores)
}
#[cfg(test)]
mod super_jobs_tests {
/// RED before: the super build took every core whatever the host's bound.
#[test]
fn the_super_build_keeps_the_hosts_cargo_job_bound() {
assert_eq!(super::super_jobs(Some("4".into()), 12), 4);
assert_eq!(super::super_jobs(None, 12), 12);
assert_eq!(super::super_jobs(Some("x".into()), 12), 12);
}
}