crates/mimir-ingest/src/dump.rs
//! The pluggable dump step: whatever turns a checked-out worktree into Arrow
//! rows per crate. Production = the lane-3 rustc driver; tests inject a fake.
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use anyhow::{Context, Result, bail};
use skade::arrow_array::RecordBatch;
/// The rows one compiled crate produced.
#[derive(Debug, Clone)]
pub struct CrateDump {
/// The unit's OWNER id in the log: `package@version:kind[+test]` (see
/// `driver`), stable across dependency changes. Every key it emits is
/// owned by it until a later dump moves or drops it.
pub krate: String,
/// A dependency / std crate (the universe ring): written once per
/// `krate` + toolchain, never diffed per repo.
pub universe: bool,
/// Node rows, in the node payload schema.
pub nodes: Vec<RecordBatch>,
/// Edge rows, in the edge payload schema.
pub edges: Vec<RecordBatch>,
/// Position rows, in the span payload schema.
pub spans: Vec<RecordBatch>,
}
impl CrateDump {
/// The owner id without its version: `package@version:kind[+test][#…]` →
/// `package:kind[+test][#…]`. The log's identities use this form, so a
/// version bump (own or dependency) keeps every key the owner holds. An
/// owner already version-free comes back unchanged.
pub fn stable_owner(owner: &str) -> std::borrow::Cow<'_, str> {
match owner.split_once('@') {
Some((pkg, rest)) => match rest.split_once(':') {
Some((_, tail)) => std::borrow::Cow::Owned(format!("{pkg}:{tail}")),
None => std::borrow::Cow::Borrowed(owner),
},
None => std::borrow::Cow::Borrowed(owner),
}
}
/// The target of an owner id (versioned or stable), as `package:kind`
/// (version, harness flag and disambiguator dropped).
pub fn target_of(owner: &str) -> String {
let s = Self::stable_owner(owner);
let s = s.split('#').next().unwrap_or(&s);
s.strip_suffix("+test").unwrap_or(s).to_string()
}
}
/// The rows mimir-binary produced for ONE linked artifact: the BINARY
/// stream's payload (never source rows — provenance is derived at read time).
#[derive(Debug, Clone)]
pub struct BinaryDump {
/// The artifact, OWNER of its keys in the binary tables:
/// `name@target-triple:profile` ([`BinaryDump::artifact_id`]).
pub artifact: String,
/// Nodes only the binary knows (bin node payload).
pub nodes: Vec<RecordBatch>,
/// INSTANCE_OF edges (bin edge payload).
pub edges: Vec<RecordBatch>,
/// `code_symbols` rows (symbol payload).
pub symbols: Vec<RecordBatch>,
/// mimir-binary's join ledger (JSON), kept in `bin_ingest_runs`.
pub ledger: Option<String>,
pub ms: i64,
}
impl BinaryDump {
/// `name@target-triple:profile`.
pub fn artifact_id(name: &str, triple: &str, profile: &str) -> String {
format!("{name}@{triple}:{profile}")
}
/// The bin name of an artifact id.
pub fn name_of(artifact: &str) -> &str {
artifact.split('@').next().unwrap_or(artifact)
}
}
/// Everything one dump of one commit produced.
#[derive(Debug, Clone)]
pub struct CommitDump {
/// The toolchain that compiled it (keys the universe manifest).
pub toolchain: String,
/// Every LOCAL unit that COMPILED at this commit (rebuilt or fresh), as
/// owner ids. A unit missing here closes its keys only if its crate is
/// gone from `workspace_targets`, or its crate compiled under another id.
pub workspace_crates: BTreeSet<String>,
/// `package:kind` of every local target the workspace declares at this
/// commit (cargo metadata), compiled or not.
pub workspace_targets: BTreeSet<String>,
/// Local units that did NOT compile (errors, or skipped because a crate
/// they depend on failed). Their keys keep their previous state.
pub failed_units: BTreeSet<String>,
/// Owner ids that needed a feature digest (or hash) to stay apart.
pub owner_disambiguations: usize,
/// The crates rustc actually compiled this time. A local crate absent
/// here (still in `workspace_crates`) did not change: its keys stay open.
pub crates: Vec<CrateDump>,
/// The binary layer ran (`--with-binary`): `binaries` is then EVERY
/// artifact linked at this commit and the binary stream is diffed in the
/// same atomic release. `false` leaves the binary stream untouched.
pub binary: bool,
pub binaries: Vec<BinaryDump>,
}
/// Where the last dump's time went and what it compiled (a dumper that
/// cannot tell reports none).
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DumpPhases {
/// The compile: cargo + the driver's walk + its IPC files (one process).
pub cargo_ms: u64,
/// Reading the IPC files and cargo's report back.
pub read_ms: u64,
/// The second pass for units with no dump (clean + recompile + read).
pub retry_ms: u64,
/// Units cargo reported (fresh or not).
pub units: u64,
/// Units dumped by THIS run (compiled now).
pub compiled: u64,
/// Units not compiled now, covered by what landed before.
pub fresh: u64,
/// Units recompiled by the second pass.
pub recompiled: u64,
}
/// The dump step. Runs with the worktree checked out at `sha`.
///
/// An `Err` is an INFRASTRUCTURE failure (tool crash, IO): nothing of the
/// commit is written, the offset does not move, the ledger records `error`.
/// A commit whose code does not (fully) compile is NOT an `Err`: return the
/// units that compiled plus `failed_units` (ledger `partial`, or `error` when
/// nothing compiled) — the offset advances either way.
pub trait Dumper {
fn dump(&mut self, worktree: &Path, sha: &str) -> Result<CommitDump>;
/// The log this dumper writes for ([`crate::EventLog::dump_identity`]): state
/// the dumper keeps across commits belongs to exactly that log.
fn attach(&mut self, log_identity: &str) -> Result<()> {
let _ = log_identity;
Ok(())
}
/// The release built from the last [`Dumper::dump`] landed in the log:
/// whatever the dumper keeps across commits may now treat that dump as
/// consumed (never before — a dump whose release did not land is redone).
fn committed(&mut self) -> Result<()> {
Ok(())
}
/// ONLY the binary layer, on the worktree as checked out at `sha` (the
/// artifacts were rebuilt — another profile, flags, linker — the source
/// did not change): the rows of every linked artifact.
fn binaries(&mut self, worktree: &Path, sha: &str) -> Result<Vec<BinaryDump>> {
let _ = (worktree, sha);
bail!("this dumper has no binary layer")
}
/// Where the last [`Dumper::dump`]'s time went (`None`: not measured).
fn phases(&self) -> Option<DumpPhases> {
None
}
/// **The super build** (all repos compiled at once over one target dir,
/// [`crate::driver::run_super`]): this repo's line of the job for the
/// commit `sha` checked out at `worktree` — `None` when this dumper does
/// not take part (the binary layer, a scripted dumper). Its later
/// [`Dumper::dump`] of `sha` reads what the super build left.
fn super_entry(&mut self, worktree: &Path, sha: &str) -> Result<Option<crate::driver::SuperEntry>> {
let _ = (worktree, sha);
Ok(None)
}
/// **A check line of the super build**: this repo's tree, as checked out
/// at `worktree`, compiled with the driver dumping NOTHING (`MIMIR_DUMP=off`)
/// — a dependent of a change (`BuildWorkspace { only: Affected }`), at a
/// commit already read, compiled only to see that it still builds against
/// the changed trees beside it. `None` when this dumper does not take part.
fn super_check_entry(&mut self, worktree: &Path) -> Result<Option<crate::driver::SuperEntry>> {
let _ = worktree;
Ok(None)
}
/// The unit keys every repo of the super build dumped or covered.
fn super_adopt(&mut self, cover: std::sync::Arc<std::collections::BTreeSet<String>>) {
let _ = cover;
}
/// A super build runs BESIDE the reads: `outs` are its repos' out dirs,
/// `done` turns true when it has ended (a dump of a commit it compiles
/// waits for that repo's record).
fn super_watch(&mut self, outs: std::sync::Arc<Vec<std::path::PathBuf>>, done: std::sync::Arc<std::sync::atomic::AtomicBool>) {
let _ = (outs, done);
}
/// Run the super build of `entries` (`line`: repo, cargo JSON line).
fn super_run(&self, job_dir: &Path, entries: &[crate::driver::SuperEntry], jobs: usize, parallel_repos: usize, line: &mut dyn FnMut(&str, &str)) -> Result<crate::driver::SuperOutcome> {
let _ = (job_dir, entries, jobs, parallel_repos, line);
bail!("this dumper runs no super build")
}
/// A read of this repo is coming: tell whoever watches the builds what
/// the tree at `worktree` will compile (its unit plan), before its turn.
fn preplan(&mut self, worktree: &Path) {
let _ = worktree;
}
}
/// A boxed dumper is a dumper: a pool can then hold different kinds (the
/// driver in production, a scripted one in a test) behind one type.
impl<D: Dumper + ?Sized> Dumper for Box<D> {
fn dump(&mut self, worktree: &Path, sha: &str) -> Result<CommitDump> {
(**self).dump(worktree, sha)
}
fn attach(&mut self, log_identity: &str) -> Result<()> {
(**self).attach(log_identity)
}
fn committed(&mut self) -> Result<()> {
(**self).committed()
}
fn binaries(&mut self, worktree: &Path, sha: &str) -> Result<Vec<BinaryDump>> {
(**self).binaries(worktree, sha)
}
fn phases(&self) -> Option<DumpPhases> {
(**self).phases()
}
fn preplan(&mut self, worktree: &Path) {
(**self).preplan(worktree)
}
fn super_entry(&mut self, worktree: &Path, sha: &str) -> Result<Option<crate::driver::SuperEntry>> {
(**self).super_entry(worktree, sha)
}
fn super_check_entry(&mut self, worktree: &Path) -> Result<Option<crate::driver::SuperEntry>> {
(**self).super_check_entry(worktree)
}
fn super_adopt(&mut self, cover: std::sync::Arc<std::collections::BTreeSet<String>>) {
(**self).super_adopt(cover)
}
fn super_watch(&mut self, outs: std::sync::Arc<Vec<std::path::PathBuf>>, done: std::sync::Arc<std::sync::atomic::AtomicBool>) {
(**self).super_watch(outs, done)
}
fn super_run(&self, job_dir: &Path, entries: &[crate::driver::SuperEntry], jobs: usize, parallel_repos: usize, line: &mut dyn FnMut(&str, &str)) -> Result<crate::driver::SuperOutcome> {
(**self).super_run(job_dir, entries, jobs, parallel_repos, line)
}
}
/// One dumper per repo behind ONE consumer (wiring map I1): a workspace has N
/// members on one log, and a dumper is per repo (its target dir, its out dir,
/// its pending dump, its mirror). The pool IS a [`Dumper`]: it routes each
/// call to the dumper registered for the worktree the consumer checked out.
pub struct DumperPool<D> {
by_worktree: BTreeMap<PathBuf, D>,
log_identity: Option<String>,
/// The dumper whose dump is waiting for its release to land.
last: Option<PathBuf>,
}
impl<D> Default for DumperPool<D> {
fn default() -> Self {
Self { by_worktree: BTreeMap::new(), log_identity: None, last: None }
}
}
impl<D: Dumper> DumperPool<D> {
pub fn new() -> Self {
Self::default()
}
/// Register the dumper of the repo checked out at `worktree` (replacing an
/// earlier one). It is attached to the pool's log when the pool has one.
pub fn insert(&mut self, worktree: PathBuf, mut dumper: D) -> Result<()> {
if let Some(id) = &self.log_identity {
dumper.attach(id)?;
}
self.by_worktree.insert(worktree, dumper);
Ok(())
}
pub fn contains(&self, worktree: &Path) -> bool {
self.by_worktree.contains_key(worktree)
}
pub fn get_mut(&mut self, worktree: &Path) -> Option<&mut D> {
self.by_worktree.get_mut(worktree)
}
pub fn len(&self) -> usize {
self.by_worktree.len()
}
pub fn is_empty(&self) -> bool {
self.by_worktree.is_empty()
}
fn of(&mut self, worktree: &Path) -> Result<&mut D> {
self.by_worktree
.get_mut(worktree)
.with_context(|| format!("no dumper registered for the worktree {}", worktree.display()))
}
}
impl<D: Dumper> Dumper for DumperPool<D> {
fn dump(&mut self, worktree: &Path, sha: &str) -> Result<CommitDump> {
let dump = self.of(worktree)?.dump(worktree, sha)?;
self.last = Some(worktree.to_path_buf());
Ok(dump)
}
fn attach(&mut self, log_identity: &str) -> Result<()> {
self.log_identity = Some(log_identity.to_string());
for d in self.by_worktree.values_mut() {
d.attach(log_identity)?;
}
Ok(())
}
fn committed(&mut self) -> Result<()> {
match self.last.take() {
Some(w) => self.of(&w)?.committed(),
None => Ok(()),
}
}
fn binaries(&mut self, worktree: &Path, sha: &str) -> Result<Vec<BinaryDump>> {
let bins = self.of(worktree)?.binaries(worktree, sha)?;
self.last = Some(worktree.to_path_buf());
Ok(bins)
}
fn phases(&self) -> Option<DumpPhases> {
self.last.as_ref().and_then(|w| self.by_worktree.get(w)).and_then(Dumper::phases)
}
fn preplan(&mut self, worktree: &Path) {
if let Some(d) = self.by_worktree.get_mut(worktree) {
d.preplan(worktree);
}
}
fn super_entry(&mut self, worktree: &Path, sha: &str) -> Result<Option<crate::driver::SuperEntry>> {
match self.by_worktree.get_mut(worktree) {
Some(d) => d.super_entry(worktree, sha),
None => Ok(None),
}
}
fn super_check_entry(&mut self, worktree: &Path) -> Result<Option<crate::driver::SuperEntry>> {
match self.by_worktree.get_mut(worktree) {
Some(d) => d.super_check_entry(worktree),
None => Ok(None),
}
}
fn super_adopt(&mut self, cover: std::sync::Arc<std::collections::BTreeSet<String>>) {
for d in self.by_worktree.values_mut() {
d.super_adopt(cover.clone());
}
}
fn super_watch(&mut self, outs: std::sync::Arc<Vec<std::path::PathBuf>>, done: std::sync::Arc<std::sync::atomic::AtomicBool>) {
for d in self.by_worktree.values_mut() {
d.super_watch(outs.clone(), done.clone());
}
}
fn super_run(&self, job_dir: &Path, entries: &[crate::driver::SuperEntry], jobs: usize, parallel_repos: usize, line: &mut dyn FnMut(&str, &str)) -> Result<crate::driver::SuperOutcome> {
// any repo's dumper runs it: one host, one driver, one target dir
let Some(d) = self.by_worktree.values().next() else { bail!("no dumper to run the super build") };
d.super_run(job_dir, entries, jobs, parallel_repos, line)
}
}