crates/mimir-ingest/src/consumer.rs
//! The consumer loop (Debezium shape): per repo, in order, each first-parent
//! commit after the offset → checkout → dump → delta vs the open set → ONE
//! atomic commit of events + offset + ledger row.
use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use std::time::Instant;
use anyhow::{Context, Result, bail};
use skade::arrow_array::RecordBatch;
use crate::delta::{self, Kind, OpenSet};
use crate::dump::{BinaryDump, CommitDump, CrateDump, Dumper};
use crate::git::{self, Change};
use crate::checkout;
use crate::log::{BinRunRow, CommitWrite, EDGE_SORT_COLUMN, EventLog, LogWriter, Offset, RunRow, UniverseRow};
use crate::schema::{STATUS_CLOSED, STATUS_ERROR, STATUS_OK, STATUS_PARTIAL, STATUS_SKIPPED_DOCS_ONLY, UNIVERSE_REPO};
/// One tracked repo.
#[derive(Debug, Clone)]
pub struct RepoSpec {
/// The repo's name in the log (`repo` column).
pub name: String,
/// Where to fetch from (a local path or a remote URL).
pub source: String,
/// The tracked branch (its first-parent history is the event stream).
pub branch: String,
/// The consumer's own clone of the repo: checkouts happen here, never in
/// anybody's working copy. Checkouts write only the paths a commit
/// changes; build output should still live outside it.
pub worktree: PathBuf,
/// Where a NEW consumer (no stored offset) starts: the first commit
/// consumed is the one after this sha. `None` = the root commit. Ignored
/// once the log holds an offset for the repo.
pub initial_offset: Option<String>,
}
/// Wall-clock split of one consumed commit.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PhaseMs {
pub classify: u64,
pub checkout: u64,
pub dump: u64,
pub delta: u64,
pub commit: u64,
/// Inside `dump`: the compile, the read-back, the second pass, and the
/// units (when the dumper measures them).
pub dump_detail: Option<crate::dump::DumpPhases>,
}
/// What happened to one first-parent commit.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommitReport {
pub seq: i64,
pub sha: String,
pub status: String,
pub crates_dumped: i64,
pub rows_plus: i64,
pub rows_minus: i64,
pub key_collisions: i64,
pub span_plus: i64,
pub span_minus: i64,
/// The binary stream's release ordinal, when the binary layer ran.
pub bin_rev: Option<i64>,
/// The binary stream's events (all three binary tables).
pub bin_plus: i64,
pub bin_minus: i64,
/// Units that did not compile (`partial` / commit-level `error`).
pub failed_units: Vec<String>,
/// Why the commit is `partial` / `error` — the ledger row's own sentence
/// (which units failed, or why the workspace does not load at the commit).
pub error: Option<String>,
/// Files the incremental checkout wrote / deleted.
pub files_written: usize,
pub files_deleted: usize,
pub ms: i64,
pub phase_ms: PhaseMs,
}
/// How one [`Consumer::consume_run`] is bounded and observed (wiring map I2).
/// `Run::default()` = everything behind the offset, unobserved.
#[derive(Default)]
pub struct Run<'a> {
/// Stop after this many commits; the rest stay behind the offset and the
/// next run resumes there (fairness between the repos of one log).
pub max_commits: Option<usize>,
/// Checked before each commit: once set, the run stops THERE — the commit
/// in flight is never abandoned half-way (its release is atomic).
pub cancel: Option<&'a std::sync::atomic::AtomicBool>,
/// A commit's step begins: (seq, sha, commits still behind it).
pub on_begin: Option<&'a mut dyn FnMut(i64, &str, usize)>,
/// A commit's release landed (or its error row was written).
pub on_commit: Option<&'a mut dyn FnMut(&CommitReport)>,
/// Replay from this first-parent commit (INCLUSIVE) instead of resuming
/// after the offset — the explicit backfill. The repo's ordinals keep
/// counting up: the first replayed commit is one more release, whose
/// delta takes the open state to that commit's, and the releases after
/// it walk forward to the tip again. Nothing is deleted.
pub replay_from: Option<String>,
/// The repo's history was rewritten under its offset and the run reads it
/// again from its head ([`Consumed::rewritten`]): called once, before that
/// release.
pub on_rewritten: Option<&'a mut dyn FnMut(&git::HistoryRewritten)>,
}
/// Why a run returned.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Stopped {
/// Nothing is behind the offset any more.
CaughtUp,
/// `max_commits` was reached.
Bound,
/// `cancel` was set.
Cancelled,
}
/// What one bounded run did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Consumed {
pub reports: Vec<CommitReport>,
/// First-parent commits still behind the offset when the run returned.
pub behind: usize,
pub stopped: Stopped,
/// The stored offset was not on the branch's first-parent chain any more
/// (history rewritten): the repo was read again from its head as a fresh
/// member — ONE release (seq = the old offset's + 1) whose delta takes the
/// open state from the old history's to the head's. What the head does not
/// have closes, so nothing of the old history stays open beside the new.
pub rewritten: Option<git::HistoryRewritten>,
}
/// The consumer: the log's ONE writer, the open set derived from the log, and
/// a dumper. It holds the writer seat ([`LogWriter`]) for its lifetime; any
/// number of readers open rooms on clones of [`Consumer::log`] meanwhile, at
/// the last committed release.
pub struct Consumer<D> {
writer: LogWriter,
open: OpenSet,
dumper: D,
}
impl<D: Dumper> Consumer<D> {
/// Boot: claim the log's writer seat (an error when another consumer
/// holds it) and derive the open set from the log — the READ-FAST zone
/// where a repo has one, the event fold elsewhere
/// ([`EventLog::fold_fast`]). The zone replaced the open-set cache file
/// (`open-set.bin`): the lake itself holds the anchor now.
pub async fn boot(log: EventLog, mut dumper: D) -> Result<Self> {
let writer = log.writer().context("boot a consumer")?;
dumper.attach(&log.dump_identity().await?).context("attach the dumper to the log")?;
let open = log.fold_fast().await?;
Ok(Self { writer, open, dumper })
}
/// The writer seat (for what else rides the consumer's offsets: the feed
/// ledger, the toolchain's std dump).
pub fn writer(&self) -> &LogWriter {
&self.writer
}
/// The log (a shareable handle: clone it for readers).
pub fn log(&self) -> &EventLog {
self.writer.log()
}
pub fn open_set(&self) -> &OpenSet {
&self.open
}
pub fn dumper_mut(&mut self) -> &mut D {
&mut self.dumper
}
/// Swap the dumper (one consumer serves N repos of one log, a dumper per
/// repo): the new one is attached to the log first.
pub async fn replace_dumper(&mut self, mut dumper: D) -> Result<D> {
dumper.attach(&self.log().dump_identity().await?).context("attach the dumper to the log")?;
Ok(std::mem::replace(&mut self.dumper, dumper))
}
/// Give the seat back: the log handle and the dumper.
pub fn into_parts(self) -> (EventLog, D) {
(self.writer.log().clone(), self.dumper)
}
/// Fetch the repo into its worktree clone, then consume.
pub async fn run(&mut self, repo: &RepoSpec) -> Result<Vec<CommitReport>> {
git::sync_clone(&repo.source, &repo.worktree).with_context(|| format!("sync {} from {}", repo.name, repo.source))?;
self.consume(repo).await
}
/// Consume every first-parent commit after the offset, oldest first.
///
/// A commit whose code does not compile never blocks: it is recorded
/// (`partial` / `error`) and the offset advances. Only an INFRASTRUCTURE
/// failure (dumper crash, IO, catalog conflict) stops here: its error is
/// recorded in the ledger and returned, the offset stays on the last good
/// commit.
pub async fn consume(&mut self, repo: &RepoSpec) -> Result<Vec<CommitReport>> {
Ok(self.consume_run(repo, Run::default()).await?.reports)
}
/// [`Self::consume`], bounded and observed ([`Run`]): per-commit progress,
/// a commit budget, cancellation between commits, and the explicit replay.
pub async fn consume_run(&mut self, repo: &RepoSpec, mut run: Run<'_>) -> Result<Consumed> {
let offset = self.log().offset(&repo.name).await?;
if let Some(o) = &offset
&& o.branch != repo.branch
{
bail!(
"{}: the offset tracks branch `{}`, the spec says `{}` — a branch switch is a new consumer, not a resume",
repo.name,
o.branch,
repo.branch
);
}
let replay_base = match &run.replay_from {
Some(sha) => Some(git::offset_before(&repo.worktree, sha).with_context(|| format!("replay {} from {sha}", repo.name))?),
None => None,
};
let from = match &replay_base {
Some(base) => base.as_deref(),
None => offset.as_ref().map(|o| o.sha.as_str()).or(repo.initial_offset.as_deref()),
};
let mut rewritten = None;
let commits = match git::first_parent_since(&repo.worktree, &repo.branch, from) {
Ok(commits) => commits,
// The offset is in a history the branch no longer has (a
// force-push, a repo moved with its history rewritten): read the
// repo again from its head, as a fresh member would start. The
// release's delta is against the repo's open state — the old
// history's — so what the head does not have closes; the ordinals
// keep counting up. Nothing is deleted (the old releases stay
// history, closed).
Err(e) if replay_base.is_none() && offset.is_some() && e.downcast_ref::<git::HistoryRewritten>().is_some() => {
let r = e.downcast::<git::HistoryRewritten>().expect("checked above");
if let Some(f) = run.on_rewritten.as_mut() {
f(&r);
}
let head = r.head.clone();
rewritten = Some(r);
vec![head]
}
Err(e) => return Err(e),
};
// The first commit of a stream that does not start at the root (a
// first offset, a replay, a rewritten history read from its head) has
// no previous state in the log to keep: it is dumped even when it only
// touched docs.
let mut jump = rewritten.is_some() || replay_base.is_some() || (offset.is_none() && repo.initial_offset.is_some());
let mut seq = offset.map(|o| o.seq).unwrap_or(0);
let total = commits.len();
let mut out = Vec::with_capacity(total.min(run.max_commits.unwrap_or(usize::MAX)));
let mut stopped = Stopped::CaughtUp;
for (i, sha) in commits.into_iter().enumerate() {
if run.cancel.is_some_and(|c| c.load(std::sync::atomic::Ordering::Acquire)) {
stopped = Stopped::Cancelled;
break;
}
if run.max_commits.is_some_and(|m| out.len() >= m) {
stopped = Stopped::Bound;
break;
}
seq += 1;
if let Some(f) = run.on_begin.as_mut() {
f(seq, &sha, total - i - 1);
}
let t0 = Instant::now();
match self.step(repo, seq, &sha, t0, std::mem::take(&mut jump)).await {
Ok(r) => {
if let Some(f) = run.on_commit.as_mut() {
f(&r);
}
out.push(r);
}
Err(e) => {
let e = match &rewritten {
Some(r) => e.context(format!("reading it again from its head {} (offset {} is not on `{}` any more)", r.head, r.offset, r.branch)),
None => e,
};
let row = RunRow {
repo: repo.name.clone(),
seq,
sha: sha.clone(),
status: STATUS_ERROR.into(),
crates_dumped: 0,
rows_plus: 0,
rows_minus: 0,
key_collisions: 0,
span_plus: 0,
span_minus: 0,
owner_disambiguations: 0,
ms: t0.elapsed().as_millis() as i64,
error: Some(format!("{e:#}")),
};
if let Err(le) = self.writer.record_error(&row).await {
return Err(e.context(format!("and the error ledger row failed too: {le:#}")));
}
return Err(e.context(format!("{} @ {sha} (seq {seq})", repo.name)));
}
}
}
let behind = total - out.len();
Ok(Consumed { reports: out, behind, stopped, rewritten })
}
/// One commit. `jump`: the worktree's previous state is not this commit's
/// first parent in the log (first offset, replay) — never skip the dump.
async fn step(&mut self, repo: &RepoSpec, seq: i64, sha: &str, t0: Instant, jump: bool) -> Result<CommitReport> {
let mut ph = PhaseMs::default();
let mut lap = Instant::now();
let mut tick = |slot: &mut u64| {
*slot = lap.elapsed().as_millis() as u64;
lap = Instant::now();
};
let change = git::classify(&repo.worktree, sha)?;
tick(&mut ph.classify);
let offset = Offset { branch: repo.branch.clone(), sha: sha.to_string(), seq };
if change == Change::DocsOnly && !jump {
let run = run_row(repo, seq, sha, STATUS_SKIPPED_DOCS_ONLY, Counts::default(), None, t0);
self.writer.commit(source_only_release(repo, offset, &run)).await?;
tick(&mut ph.commit);
return Ok(report(&run, Vec::new(), Default::default(), ph));
}
let co = checkout::checkout_changed(&repo.worktree, sha)?;
tick(&mut ph.checkout);
let dump = self.dumper.dump(&repo.worktree, sha).context("dump")?;
tick(&mut ph.dump);
ph.dump_detail = self.dumper.phases();
let failed: Vec<String> = dump.failed_units.iter().cloned().collect();
// Nothing compiled at all: no events, the offset still advances (the
// next commit's delta corrects the state).
if !failed.is_empty() && dump.workspace_crates.is_empty() {
let err = format!("nothing compiled; failed units: {}", failed.join(", "));
let run = run_row(repo, seq, sha, STATUS_ERROR, Counts::default(), Some(err), t0);
self.writer.commit(source_only_release(repo, offset, &run)).await?;
tick(&mut ph.commit);
return Ok(report(&run, failed, co, ph));
}
let rev = if dump.binary { Some(self.writer.next_rev(&repo.name).await?) } else { None };
let planned = self.plan(&repo.name, seq, rev, sha, &dump)?;
tick(&mut ph.delta);
let (status, err) = if failed.is_empty() {
(STATUS_OK, None)
} else {
(STATUS_PARTIAL, Some(format!("failed units (kept their previous state): {}", failed.join(", "))))
};
let mut counts = planned.counts;
counts.crates = dump.crates.len() as i64;
counts.owner_disambiguations = dump.owner_disambiguations as i64;
let run = run_row(repo, seq, sha, status, counts, err, t0);
let (bin_plus, bin_minus) = (planned.bin_counts.plus, planned.bin_counts.minus);
self.writer
.commit(CommitWrite {
repo: repo.name.clone(),
sha: sha.to_string(),
offset: Some(offset),
run: Some(run.clone()),
bin_runs: planned.bin_runs,
events: planned.events,
ring_events: planned.ring,
universe: planned.universe,
})
.await?;
// Only now — the commit landed — does the open set move, and may the
// dumper treat this dump as consumed.
for (kind, d) in &planned.local {
self.apply_open(*kind, &repo.name, d);
}
self.dumper.committed().context("after the release landed")?;
tick(&mut ph.commit);
Ok(CommitReport {
ms: t0.elapsed().as_millis() as i64,
bin_rev: rev,
bin_plus,
bin_minus,
..report(&run, failed, co, ph)
})
}
/// A binary-only re-ingest: the artifacts were rebuilt at the repo's
/// CURRENT offset (same sha — another profile, flags, linker), the source
/// did not change. Runs only the binary layer and commits ONLY the binary
/// stream (events + `bin_ingest_runs`) as its own atomic release with the
/// next `rev`. Artifacts not rebuilt keep their state. Idempotent: the same
/// artifacts again write ledger rows and no events.
pub async fn reingest_binaries(&mut self, repo: &RepoSpec) -> Result<Vec<BinRunRow>> {
let Some(offset) = self.log().offset(&repo.name).await? else {
bail!("{}: no offset yet — a binary re-ingest needs a consumed commit", repo.name);
};
// the clone may sit on the last CODE commit (docs-only ones are not
// checked out): move it to the offset (writes only what differs)
checkout::checkout_changed(&repo.worktree, &offset.sha)?;
let bins = self.dumper.binaries(&repo.worktree, &offset.sha).context("binary layer")?;
let rev = self.writer.next_rev(&repo.name).await?;
let mut p = Planned::default();
self.plan_binary(&repo.name, offset.seq, rev, &offset.sha, &bins, &BTreeSet::new(), &mut p)?;
let rows = p.bin_runs.clone();
self.writer
.commit(CommitWrite {
repo: repo.name.clone(),
sha: offset.sha.clone(),
offset: None,
run: None,
bin_runs: p.bin_runs,
events: p.events,
ring_events: vec![],
universe: vec![],
})
.await?;
for (kind, d) in &p.local {
self.apply_open(*kind, &repo.name, d);
}
self.dumper.committed().context("after the release landed")?;
Ok(rows)
}
/// The binary stream of `repo` from artifacts that ALREADY EXIST for its
/// current offset — linked by somebody else's build of that commit, joined
/// by the caller (`crate::artifact`). Nothing is compiled and no dumper is
/// asked: `bins` are the rows of every artifact found; an artifact that is
/// open in the log and whose bin is not in `declared_bins` (crate names, `-`
/// as `_`: the workspace no longer declares it) closes. Artifacts neither given nor gone keep their
/// keys — the binary stream then still describes the commit they were
/// linked at (`bin_offsets` says which). ONE atomic release with the next
/// `rev`; idempotent (the same artifacts again write ledger rows only).
///
/// `at` must be the repo's offset: the caller read the artifacts for that
/// commit, and a commit that landed since makes them another commit's.
pub async fn ingest_binaries(&mut self, repo: &str, at: &Offset, bins: &[BinaryDump], declared_bins: &BTreeSet<String>) -> Result<Vec<BinRunRow>> {
let Some(offset) = self.log().offset(repo).await? else {
bail!("{repo}: no offset yet — a binary ingest needs a consumed commit");
};
if offset.sha != at.sha || offset.seq != at.seq {
bail!("{repo}: the artifacts are of seq {} @ {}, the offset is seq {} @ {} — not written", at.seq, at.sha, offset.seq, offset.sha);
}
let mut gone = BTreeSet::new();
for kind in Kind::BINARY {
for owner in self.open.repo(kind, repo).into_iter().flat_map(|o| o.krates()) {
// a bin `my-tool` is the crate `my_tool`
if !declared_bins.contains(&BinaryDump::name_of(owner).replace('-', "_")) {
gone.insert(owner.clone());
}
}
}
let rev = self.writer.next_rev(repo).await?;
let mut p = Planned::default();
self.plan_binary(repo, offset.seq, rev, &offset.sha, bins, &gone, &mut p)?;
let rows = p.bin_runs.clone();
if rows.is_empty() {
return Ok(rows);
}
self.writer
.commit(CommitWrite {
repo: repo.to_string(),
sha: offset.sha.clone(),
offset: None,
run: None,
bin_runs: p.bin_runs,
events: p.events,
ring_events: vec![],
universe: vec![],
})
.await?;
for (kind, d) in &p.local {
self.apply_open(*kind, repo, d);
}
Ok(rows)
}
/// BUILDER 2: land what one Build Thing run built ([`crate::product`]) —
/// its artifact nodes, BUILT_FROM edges and provenance — in the binary
/// stream's tables under [`crate::product::PRODUCTS_REPO`], one atomic
/// release with that repo's next `rev` (which is also its `seq`: products
/// have no consumer offset). Each artifact owns its keys; nothing closes
/// (another build is another owner). Idempotent: the same artifacts again
/// write ledger rows and no events.
pub async fn ingest_products(&mut self, facts: &crate::product::ProductFacts) -> Result<Vec<BinRunRow>> {
let repo = crate::product::PRODUCTS_REPO;
let dumps = facts.dumps()?;
if dumps.is_empty() {
bail!("build {}: no artifact to land", facts.build_id);
}
let rev = self.writer.next_rev(repo).await?;
let mut p = Planned::default();
self.plan_binary(repo, rev, rev, facts.sha(), &dumps, &BTreeSet::new(), &mut p)?;
let rows = p.bin_runs.clone();
self.writer
.commit(CommitWrite {
repo: repo.to_string(),
sha: facts.sha().to_string(),
offset: None,
run: None,
bin_runs: p.bin_runs,
events: p.events,
ring_events: vec![],
universe: vec![],
})
.await?;
for (kind, d) in &p.local {
self.apply_open(*kind, repo, d);
}
Ok(rows)
}
/// Delta of one dump against the open set: event batches + what to apply
/// to the open set once they are committed. The source stream writes only
/// [`Kind::SOURCE`] tables; with the binary layer (`rev`), the binary
/// stream's delta ([`Self::plan_binary`]) joins the same release.
fn plan(&self, repo: &str, seq: i64, rev: Option<i64>, sha: &str, dump: &CommitDump) -> Result<Planned> {
let universe: Vec<&CrateDump> = dump.crates.iter().filter(|c| c.universe).collect();
let mut p = Planned::default();
for (kind, d) in source_delta(self.log(), &self.open, repo, dump)? {
let spec = self.log().spec(kind);
p.counts.add(kind, d.plus.len(), d.minus.len(), d.collisions);
if let Some(b) = delta::event_batch(spec, repo, seq, None, sha, &d)? {
p.push(kind, b)?;
}
p.local.push((kind, d));
}
// Universe ring: once per crate@version + toolchain, all `+`.
let mut fresh: BTreeMap<&str, &CrateDump> = BTreeMap::new();
for c in universe {
if !self.log().universe_has(&c.krate, &dump.toolchain) {
fresh.entry(c.krate.as_str()).or_insert(c);
}
}
for kind in Kind::SOURCE {
let spec = self.log().spec(kind);
let keyed = delta::key_rows(spec, fresh.values().map(|c| (c.krate.as_str(), rows_of(c, kind))))?;
let d = delta::universe_delta(keyed);
p.counts.add(kind, d.plus.len(), 0, d.collisions);
if let Some(b) = delta::event_batch(spec, UNIVERSE_REPO, seq, None, sha, &d)? {
// a split lake keeps the ring in its own tables; a legacy
// lake rides it in the source tables until it is split
if self.log().ring_split() { p.push_ring(kind, b)? } else { p.push(kind, b)? }
}
}
for c in fresh.values() {
let rows = Kind::SOURCE.iter().flat_map(|k| rows_of(c, *k)).map(|b| b.num_rows() as i64).sum();
p.universe.push(UniverseRow { krate: c.krate.clone(), toolchain: dump.toolchain.clone(), rows, rustc_release: None });
}
if let Some(rev) = rev {
// Artifacts whose keys close: the bin left the workspace, or it was
// linked this time under another artifact id (profile / triple). An
// artifact whose bin FAILED keeps its state.
let built: BTreeSet<&str> = dump.binaries.iter().map(|b| BinaryDump::name_of(&b.artifact)).collect();
let present: BTreeSet<&str> = dump.binaries.iter().map(|b| b.artifact.as_str()).collect();
let mut gone_bins = BTreeSet::new();
// a listed artifact whose bin the workspace does not declare (a
// stale executable) is gone too, open or not
for b in &dump.binaries {
let bin = format!(":bin:{}", BinaryDump::name_of(&b.artifact));
if !dump.workspace_targets.iter().any(|t| t.ends_with(&bin)) {
gone_bins.insert(b.artifact.clone());
}
}
for kind in Kind::BINARY {
for owner in self.open.repo(kind, repo).into_iter().flat_map(|o| o.krates()) {
let name = BinaryDump::name_of(owner);
let bin = format!(":bin:{name}");
let declared = dump.workspace_targets.iter().any(|t| t.ends_with(&bin));
let failed = dump.failed_units.iter().any(|t| t.ends_with(&bin));
let replaced = !failed && built.contains(name) && !present.contains(owner.as_str());
if !declared || replaced {
gone_bins.insert(owner.clone());
}
}
}
self.plan_binary(repo, seq, rev, sha, &dump.binaries, &gone_bins, &mut p)?;
}
Ok(p)
}
/// The binary stream's delta: per artifact (the OWNER of its keys), its
/// nodes / INSTANCE_OF edges / symbols against the open set, written ONLY
/// to [`Kind::BINARY`] tables with release ordinal `rev`, one
/// `bin_ingest_runs` row per artifact; every key of a `gone` artifact
/// closes. Artifacts neither given nor gone keep their keys.
#[allow(clippy::too_many_arguments)]
fn plan_binary(&self, repo: &str, seq: i64, rev: i64, sha: &str, bins: &[BinaryDump], gone: &BTreeSet<String>, p: &mut Planned) -> Result<()> {
let mut seen = BTreeSet::new();
for b in bins {
if !seen.insert(b.artifact.as_str()) {
bail!("artifact {} given twice in one binary release", b.artifact);
}
}
let empty = BTreeSet::new();
// A gone artifact that the dump still lists (a stale executable of a
// bin the workspace no longer declares) is gone: diffing it as dumped
// too would close its keys in this release and reopen them in the
// next, flapping every commit.
let jobs = bins
.iter()
.filter(|b| !gone.contains(&b.artifact))
.map(|b| (b.artifact.clone(), Some(b)))
.chain(gone.iter().map(|g| (g.clone(), None)));
for (artifact, b) in jobs {
let t0 = Instant::now();
let (mut plus, mut minus, mut collisions) = (0i64, 0i64, 0i64);
let owner: BTreeSet<String> = [artifact.clone()].into();
for kind in Kind::BINARY {
let spec = self.log().spec(kind);
let keyed = delta::key_rows(spec, b.map(|b| (artifact.as_str(), bin_rows_of(b, kind))))?;
let (dumped, closed) = if b.is_some() { (&owner, &empty) } else { (&empty, &owner) };
let d = delta::delta(keyed, dumped, closed, self.open.repo(kind, repo));
plus += d.plus.len() as i64;
minus += d.minus.len() as i64;
collisions += d.collisions as i64;
if let Some(batch) = delta::event_batch(spec, repo, seq, Some(rev), sha, &d)? {
p.push(kind, batch)?;
}
p.local.push((kind, d));
}
p.bin_counts.plus += plus;
p.bin_counts.minus += minus;
p.bin_runs.push(BinRunRow {
repo: repo.to_string(),
seq,
rev,
sha: sha.to_string(),
artifact,
status: if b.is_some() { STATUS_OK.into() } else { STATUS_CLOSED.into() },
rows_plus: plus,
rows_minus: minus,
key_collisions: collisions,
ms: b.map_or(0, |b| b.ms) + t0.elapsed().as_millis() as i64,
ledger: b.and_then(|b| b.ledger.clone()),
});
}
Ok(())
}
/// What `dump` WOULD change in `repo` against this consumer's open set
/// ([`dry_delta`]). Read-only: no event, no offset, no ledger row.
pub fn dry_delta(&self, repo: &str, dump: &CommitDump) -> Result<DryDelta> {
dry_delta(self.log(), &self.open, repo, dump)
}
/// Apply a delta whose release landed to the open set.
fn apply_open(&mut self, kind: Kind, repo: &str, d: &delta::KindDelta) {
self.open.apply(kind, repo, d);
}
}
/// The source stream's delta of one dump against `open`: per table, the keys
/// that open or change payload (`+`) and the ones that close (`-`). Pure — the
/// consumer writes it as events, a dry run only reads it.
fn source_delta(log: &EventLog, open: &OpenSet, repo: &str, dump: &CommitDump) -> Result<Vec<(Kind, delta::KindDelta)>> {
let local: Vec<&CrateDump> = dump.crates.iter().filter(|c| !c.universe).collect();
// The log's owners are STABLE ids (version-free): a version bump keeps
// every key the owner holds.
let owners: Vec<String> = local.iter().map(|c| CrateDump::stable_owner(&c.krate).into_owned()).collect();
let mut dumped = BTreeSet::new();
for (c, owner) in local.iter().zip(&owners) {
if !dump.workspace_crates.contains(&c.krate) {
bail!("dumped local crate {} is not in the dump's workspace_crates", c.krate);
}
if !dumped.insert(owner.clone()) {
bail!("local crate {} dumped twice in one commit", c.krate);
}
}
// Owners whose keys close: the crate left the workspace, or it compiled
// this time under another id (metadata hash moved — a version alone no
// longer re-keys anything). A crate that FAILED keeps every owner it had.
let stable_crates: BTreeSet<String> = dump.workspace_crates.iter().map(|o| CrateDump::stable_owner(o).into_owned()).collect();
let compiled: BTreeSet<String> = dump.workspace_crates.iter().map(|o| CrateDump::target_of(o)).collect();
let mut gone = BTreeSet::new();
for kind in Kind::SOURCE {
if let Some(open) = open.repo(kind, repo) {
for owner in open.krates() {
let target = CrateDump::target_of(owner);
let left = !dump.workspace_targets.contains(&target);
let replaced = !dump.failed_units.contains(&target)
&& compiled.contains(&target)
&& !stable_crates.contains(owner.as_str());
if left || replaced {
gone.insert(owner.clone());
}
}
}
}
let mut out = Vec::new();
for kind in Kind::SOURCE {
let spec = log.spec(kind);
let keyed =
delta::key_rows(spec, local.iter().zip(&owners).map(|(c, owner)| (owner.as_str(), rows_of(c, kind))))?;
out.push((kind, delta::delta(keyed, &dumped, &gone, open.repo(kind, repo))));
}
Ok(out)
}
/// The keys of one table a dry run would move.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct KeyDelta {
/// Not open today.
pub added: Vec<String>,
/// Open today with another payload.
pub changed: Vec<String>,
/// Open today, gone from the tree.
pub removed: Vec<String>,
}
impl KeyDelta {
pub fn is_empty(&self) -> bool {
self.added.is_empty() && self.changed.is_empty() && self.removed.is_empty()
}
pub fn len(&self) -> usize {
self.added.len() + self.changed.len() + self.removed.len()
}
}
/// What a tree that is NOT a commit would change (wiring map I3): its dump
/// against the open set, as the consumer would diff it — and nothing written.
///
/// Keys are the log's identities: a node's is its id (moniker); an edge's is
/// `from ⟨U+001F⟩ to ⟨U+001F⟩ kind` ([`DryDelta::edge_parts`]). Each key is
/// listed once, whatever number of units own it (a crate's lib and its test
/// harness dump the same items). A key that only moved owner (its crate's
/// version changed, the row did not) is in none of the three lists.
///
/// EXACT only for a dump that holds EVERY local crate whose rows may differ
/// from the open state — a crate absent from the dump counts as unchanged, as
/// in a consumed commit. `DriverDumper::dump_tree` dumps them all.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DryDelta {
pub repo: String,
/// Local units of the tree that did not compile: their rows are NOT in
/// the delta (they keep their open state) — the answer is then partial.
pub failed_units: Vec<String>,
pub nodes: KeyDelta,
pub edges: KeyDelta,
/// Positions (a moved line): never a semantic change on their own.
pub spans: KeyDelta,
/// STABLE node key → the RAW id the TREE carries (added and changed
/// nodes): where the version the dump moved to matters — a reader that
/// wants to tell a version bump from a real change compares it with the
/// room's raw id.
pub raw_nodes: std::collections::BTreeMap<String, String>,
}
impl DryDelta {
/// No node and no edge would change (positions may have moved).
pub fn is_semantically_empty(&self) -> bool {
self.nodes.is_empty() && self.edges.is_empty()
}
/// The ids of every node the tree adds, changes or removes — the seeds of
/// a blast-radius walk.
pub fn touched_nodes(&self) -> Vec<&str> {
let n = &self.nodes;
n.added.iter().chain(&n.changed).chain(&n.removed).map(String::as_str).collect()
}
/// `(from, to, kind)` of an edge key.
pub fn edge_parts(key: &str) -> Option<(&str, &str, &str)> {
let mut it = key.split(crate::schema::KEY_SEP);
Some((it.next()?, it.next()?, it.next()?))
}
}
/// [`DryDelta`] of `dump` for `repo` against `open` (the fold of `log`).
pub fn dry_delta(log: &EventLog, open: &OpenSet, repo: &str, dump: &CommitDump) -> Result<DryDelta> {
let mut out = DryDelta {
repo: repo.to_string(),
failed_units: dump.failed_units.iter().cloned().collect(),
nodes: KeyDelta::default(),
edges: KeyDelta::default(),
spans: KeyDelta::default(),
raw_nodes: BTreeMap::new(),
};
for (kind, d) in source_delta(log, open, repo, dump)? {
if kind == Kind::Nodes {
// the RAW id each + row carries (the key column before the
// stable canonicalization)
let id_col = log.spec(kind).key_columns[0].clone();
for p in &d.plus {
let b = &d.batches[p.batch];
if let Some(c) = b.column_by_name(&id_col)
&& let Ok(c) = skade::arrow_cast::cast(c, &skade::arrow_schema::DataType::Utf8)
&& let Some(c) = c.as_any().downcast_ref::<skade::arrow_array::StringArray>()
{
out.raw_nodes.insert(p.key.clone(), c.value(p.row).to_string());
}
}
}
let open_repo = open.repo(kind, repo);
// key → payload hash on each side; a key on both sides moved owner
let plus: BTreeMap<&str, delta::PayloadHash> = d.plus.iter().map(|p| (p.key.as_str(), p.hash)).collect();
let mut minus: BTreeMap<&str, Option<delta::PayloadHash>> = BTreeMap::new();
for m in &d.minus {
minus.insert(m.key.as_str(), open_repo.and_then(|o| o.get(&m.krate, &m.key)).copied());
}
let slot = match kind {
Kind::Nodes => &mut out.nodes,
Kind::Edges => &mut out.edges,
Kind::Spans => &mut out.spans,
Kind::BinNodes | Kind::BinEdges | Kind::Symbols => continue,
};
for p in &d.plus {
let same_owner_before = open_repo.and_then(|o| o.get(&p.krate, &p.key));
match (same_owner_before, minus.get(p.key.as_str())) {
// open under this owner with another payload
(Some(_), _) => slot.changed.push(p.key.clone()),
// closes under another owner and opens here: moved owner
(None, Some(old)) if *old == Some(p.hash) => {}
(None, Some(_)) => slot.changed.push(p.key.clone()),
(None, None) => slot.added.push(p.key.clone()),
}
}
for m in &d.minus {
if !plus.contains_key(m.key.as_str()) {
slot.removed.push(m.key.clone());
}
}
// One unit of a crate is often dumped under several owners (its lib
// and its test-harness build): a key is listed once, and a key that is
// new under one owner but already open under another is a change.
let set = |v: &mut Vec<String>| std::mem::take(v).into_iter().collect::<BTreeSet<String>>();
let (mut added, mut changed, mut removed) = (set(&mut slot.added), set(&mut slot.changed), set(&mut slot.removed));
changed.extend(added.intersection(&removed).cloned().collect::<Vec<_>>());
added.retain(|k| !changed.contains(k));
removed.retain(|k| !changed.contains(k));
slot.added = added.into_iter().collect();
slot.changed = changed.into_iter().collect();
slot.removed = removed.into_iter().collect();
}
Ok(out)
}
/// Ledger counters of one commit.
#[derive(Debug, Clone, Copy, Default)]
struct Counts {
crates: i64,
plus: i64,
minus: i64,
collisions: i64,
span_plus: i64,
span_minus: i64,
owner_disambiguations: i64,
}
impl Counts {
fn add(&mut self, kind: Kind, plus: usize, minus: usize, collisions: usize) {
self.collisions += collisions as i64;
if kind == Kind::Spans {
self.span_plus += plus as i64;
self.span_minus += minus as i64;
} else {
self.plus += plus as i64;
self.minus += minus as i64;
}
}
}
fn run_row(repo: &RepoSpec, seq: i64, sha: &str, status: &str, c: Counts, error: Option<String>, t0: Instant) -> RunRow {
RunRow {
repo: repo.name.clone(),
seq,
sha: sha.to_string(),
status: status.to_string(),
crates_dumped: c.crates,
rows_plus: c.plus,
rows_minus: c.minus,
key_collisions: c.collisions,
span_plus: c.span_plus,
span_minus: c.span_minus,
owner_disambiguations: c.owner_disambiguations,
ms: t0.elapsed().as_millis() as i64,
error,
}
}
/// A release of the source stream with no events (docs-only, nothing compiled).
fn source_only_release(repo: &RepoSpec, offset: Offset, run: &RunRow) -> CommitWrite {
CommitWrite {
repo: repo.name.clone(),
sha: offset.sha.clone(),
offset: Some(offset),
run: Some(run.clone()),
bin_runs: vec![],
events: vec![],
ring_events: vec![],
universe: vec![],
}
}
/// A source dump's rows of one SOURCE table (a binary kind has none: the
/// source stream never writes a binary table).
fn rows_of(c: &CrateDump, kind: Kind) -> &[RecordBatch] {
match kind {
Kind::Nodes => &c.nodes,
Kind::Edges => &c.edges,
Kind::Spans => &c.spans,
Kind::BinNodes | Kind::BinEdges | Kind::Symbols => &[],
}
}
/// An artifact's rows of one BINARY table (none for a source kind).
fn bin_rows_of(b: &BinaryDump, kind: Kind) -> &[RecordBatch] {
match kind {
Kind::BinNodes => &b.nodes,
Kind::BinEdges => &b.edges,
Kind::Symbols => &b.symbols,
Kind::Nodes | Kind::Edges | Kind::Spans => &[],
}
}
fn report(run: &RunRow, failed_units: Vec<String>, co: checkout::CheckoutStats, phase_ms: PhaseMs) -> CommitReport {
CommitReport {
seq: run.seq,
sha: run.sha.clone(),
status: run.status.clone(),
crates_dumped: run.crates_dumped,
rows_plus: run.rows_plus,
rows_minus: run.rows_minus,
key_collisions: run.key_collisions,
span_plus: run.span_plus,
span_minus: run.span_minus,
bin_rev: None,
bin_plus: 0,
bin_minus: 0,
failed_units,
error: run.error.clone(),
files_written: co.written,
files_deleted: co.deleted,
ms: run.ms,
phase_ms,
}
}
#[derive(Default)]
struct Planned {
events: Vec<(Kind, Vec<RecordBatch>)>,
/// The ring's batches, for the ring tables (a split lake).
ring: Vec<(Kind, Vec<RecordBatch>)>,
universe: Vec<UniverseRow>,
local: Vec<(Kind, delta::KindDelta)>,
counts: Counts,
bin_runs: Vec<BinRunRow>,
bin_counts: BinCounts,
}
#[derive(Default)]
struct BinCounts {
plus: i64,
minus: i64,
}
impl Planned {
/// Add an event batch; edge events go in sorted by `from` (the tables'
/// declared sort order).
fn push(&mut self, kind: Kind, b: RecordBatch) -> Result<()> {
Self::push_into(&mut self.events, kind, b)
}
/// A ring batch, for the ring table of `kind`.
fn push_ring(&mut self, kind: Kind, b: RecordBatch) -> Result<()> {
Self::push_into(&mut self.ring, kind, b)
}
fn push_into(into: &mut Vec<(Kind, Vec<RecordBatch>)>, kind: Kind, b: RecordBatch) -> Result<()> {
let b = if matches!(kind, Kind::Edges | Kind::BinEdges) { sort_by(&b, EDGE_SORT_COLUMN)? } else { b };
match into.iter_mut().find(|(k, _)| *k == kind) {
Some((_, v)) => v.push(b),
None => into.push((kind, vec![b])),
}
Ok(())
}
}
pub(crate) fn sort_by(b: &RecordBatch, column: &str) -> Result<RecordBatch> {
let col = b.column_by_name(column).with_context(|| format!("sort column `{column}`"))?;
let idx = arrow_ord::sort::sort_to_indices(col, None, None)?;
Ok(arrow_select::take::take_record_batch(b, &idx)?)
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use skade::arrow_array::{Array, Int64Array, RecordBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use super::{Kind, Planned};
#[test]
fn edge_events_are_written_sorted_by_from() {
let schema = Arc::new(Schema::new(vec![Field::new("from", DataType::Utf8, false), Field::new("n", DataType::Int64, false)]));
let b = RecordBatch::try_new(
schema,
vec![Arc::new(StringArray::from(vec!["c", "a", "b"])), Arc::new(Int64Array::from(vec![3, 1, 2]))],
)
.unwrap();
let mut p = Planned::default();
p.push(Kind::Edges, b.clone()).unwrap();
p.push(Kind::Nodes, b).unwrap();
let col = |k: Kind| {
let (_, v) = p.events.iter().find(|(kk, _)| *kk == k).unwrap();
let n = v[0].column(1).as_any().downcast_ref::<Int64Array>().unwrap().clone();
(0..n.len()).map(|i| n.value(i)).collect::<Vec<_>>()
};
assert_eq!(col(Kind::Edges), vec![1, 2, 3], "edges sorted by `from`, rows kept whole");
assert_eq!(col(Kind::Nodes), vec![3, 1, 2], "other tables keep their order");
}
}