crates/dwarves/src/incremental.rs
//! **Change-based incremental builds + the action-cache walk** — dwarves's Phase-2
//! differentiators over stock TeamCity.
//!
//! [`run_pipeline_cached`] is a drop-in richer sibling of
//! [`crate::brokkr::engine::run_pipeline`]: it walks the same topo-ordered [`Dag`], but
//! before running each step it
//!
//! 1. **consults change detection** — if an [`IncrementalOptions::affected`] blast
//! radius (from nornir's [`crate::change`-style `ChangeSet::affected`]) is
//! supplied and neither the step nor any upstream step that actually re-ran is
//! in it, the step is *proven unaffected*: it is **skipped as `Unchanged`** and
//! its previously-cached artifact is fed forward to satisfy downstream deps;
//! 2. **consults the action cache** — otherwise it computes the step's input
//! content-id ([`crate::cache::step_input_id`]) and, on a **hit**, skips
//! execution as a `CacheHit`, reusing the recorded result + artifact; on a
//! **miss** it runs the step for real and records `(input_id → result)`.
//!
//! Every step is tagged with a precise [`StepDisposition`] so the run report /
//! `state_json` distinguishes **ran / cache-hit / unchanged / when-skipped /
//! upstream-skipped / failed** — the viz DAG can then render cached/skipped
//! stations differently from freshly-run ones. The embedded [`crate::brokkr::RunReport`]
//! stays a fully valid Phase-0 report (skips map to [`RunStatus::Skipped`], reuse
//! to [`RunStatus::Passed`]), so historization, the facets and every existing
//! call site keep working unchanged.
//!
//! **Phase-0 invariant:** with `use_cache = false` and `affected = None` (an empty
//! cache, incremental off) this produces byte-identical results to
//! [`crate::brokkr::engine::run_pipeline`] — every step simply `Ran`.
use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use std::sync::Arc;
use chrono::Utc;
use serde::Serialize;
use crate::artifact_store::ArtifactStore;
use crate::brokkr::dag::Dag;
use crate::brokkr::engine::{RunReport, RunRequest};
use crate::brokkr::history::{BuildRow, PipelineRunRow, RunStatus};
use crate::brokkr::model::{Pipeline, Step};
use crate::brokkr::runner::{StepContext, StepOutcome, StepRunner};
use crate::cache::{ActionCache, ActionCacheEntry, step_input_id};
/// How a step was handled this run — the Phase-2 status taxonomy the run report
/// and `state_json` surface (finer than [`RunStatus`], which stays the durable
/// warehouse shape).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum StepDisposition {
/// Executed for real this run (a cache miss / forced / genuinely affected).
Ran,
/// Skipped: its input content-id hit the action cache; recorded result reused.
CacheHit,
/// Skipped: change detection proved it unaffected (outside the blast radius);
/// its cached artifact was fed forward.
Unchanged,
/// Skipped: a `when` gate did not match this run's branch/event context.
WhenSkipped,
/// Skipped: an upstream step failed (fail-fast cascade).
UpstreamSkipped,
/// Executed and failed.
Failed,
}
impl StepDisposition {
/// The status-chip label.
pub fn label(&self) -> &'static str {
match self {
StepDisposition::Ran => "ran",
StepDisposition::CacheHit => "cache-hit",
StepDisposition::Unchanged => "unchanged",
StepDisposition::WhenSkipped => "when-skipped",
StepDisposition::UpstreamSkipped => "upstream-skipped",
StepDisposition::Failed => "failed",
}
}
/// Whether the result was reused from the cache (a `CacheHit`/`Unchanged`
/// station the viz can tint distinctly from a freshly-`Ran` one).
pub fn cached(&self) -> bool {
matches!(self, StepDisposition::CacheHit | StepDisposition::Unchanged)
}
/// The durable warehouse [`RunStatus`] this disposition historizes as.
pub fn status(&self) -> RunStatus {
match self {
StepDisposition::Ran | StepDisposition::CacheHit | StepDisposition::Unchanged => {
RunStatus::Passed
}
StepDisposition::Failed => RunStatus::Failed,
StepDisposition::WhenSkipped | StepDisposition::UpstreamSkipped => RunStatus::Skipped,
}
}
}
/// Knobs controlling the cached / incremental walk.
#[derive(Debug, Clone, Default)]
pub struct IncrementalOptions {
/// Consult + populate the action cache. `false` ⇒ every step runs and nothing
/// is recorded (Phase-0 behavior).
pub use_cache: bool,
/// Force everything to run, ignoring both the cache and the blast radius (the
/// `--full` path). Steps still record their results into the cache.
pub full: bool,
/// The change blast radius (nornir `ChangeSet::affected`, mapped to the step
/// identities this pipeline cares about — step names and/or the pipeline's
/// project/source keys). `None` ⇒ no incremental filtering (treat every step
/// as possibly-affected; the action cache still de-dupes unchanged content).
pub affected: Option<BTreeSet<String>>,
/// Per-step source-input fingerprint (step-name → source subtree hash / sha).
/// Absent steps fall back to [`default_source_id`](Self::default_source_id).
pub source_ids: BTreeMap<String, String>,
/// The pipeline-level source fingerprint applied to steps without a specific
/// [`source_ids`](Self::source_ids) entry (e.g. the built sha). `""` ⇒ none.
pub default_source_id: String,
/// The artifact store to push produced artifact **bytes** into (holger, sealed
/// via znippy). `None` (the default) keeps the id-only behavior — a hit feeds
/// forward the artifact *label* only. When set, a step that ran and reported one
/// or more
/// [`artifact_files`](crate::brokkr::runner::StepOutcome::artifact_files) has EVERY
/// file's bytes pushed (a single file's own content id, or a multi-artifact build's
/// manifest id, becomes the recorded `artifact_id`), so a later hit on ANY machine
/// can [`get`](crate::artifact_store::ArtifactStore::get) the real output(s) back.
/// Additive; every existing caller leaves this `None`.
pub artifact_store: Option<Arc<dyn ArtifactStore>>,
/// A workspace directory into which a **REUSED** step's artifact bytes are
/// auto-restored on a cache **HIT** / **UNCHANGED** skip — the inverse of the
/// [`artifact_store`](Self::artifact_store) push. A skipped step never re-produces
/// its output, yet a downstream *real* step may consume it; with this set, the
/// reused artifact is [`materialize`](crate::artifact_store::materialize)d back
/// onto disk here so the downstream step finds it (even on a fresh workspace / a
/// second machine). `None` (the default) keeps the id-only reuse — no disk write.
/// A no-op unless [`artifact_store`](Self::artifact_store) is also set (the bytes
/// live there). Best-effort: a restore error never fails the run. Additive.
pub materialize_dir: Option<PathBuf>,
}
impl IncrementalOptions {
/// The default Phase-2 standalone run: action cache on, no incremental
/// filtering, no force. (An empty cache + this = one full run that populates
/// the cache; a second identical run then hits it.)
pub fn cached() -> Self {
Self {
use_cache: true,
..Default::default()
}
}
/// The Phase-0 identity: cache off, incremental off — byte-identical to
/// [`crate::brokkr::engine::run_pipeline`].
pub fn off() -> Self {
Self::default()
}
/// This run forces a full rebuild (still recording into the cache).
pub fn forced() -> Self {
Self {
use_cache: true,
full: true,
..Default::default()
}
}
/// Wire an artifact store (holger) into this run so produced artifact bytes are
/// pushed to the shared registry (and resolvable on a later hit, cross-machine).
/// Builder-style, additive.
pub fn with_artifact_store(mut self, store: Arc<dyn ArtifactStore>) -> Self {
self.artifact_store = Some(store);
self
}
/// Wire a workspace dir into which a reused (cache-hit / unchanged) step's
/// artifact bytes are auto-restored on reuse, so a downstream real step finds
/// the upstream output on disk. Builder-style, additive; a no-op without an
/// [`artifact_store`](Self::artifact_store).
pub fn with_materialize_dir(mut self, dir: impl Into<PathBuf>) -> Self {
self.materialize_dir = Some(dir.into());
self
}
/// The source fingerprint for `step`.
fn source_for(&self, step: &str) -> &str {
self.source_ids
.get(step)
.map(String::as_str)
.unwrap_or(&self.default_source_id)
}
/// Whether `step` is within the change blast radius (or `affected` is unset ⇒
/// everything is a candidate). Matches on the step name or its source key.
fn in_blast_radius(&self, step: &str) -> bool {
match &self.affected {
None => true,
Some(a) => a.contains(step) || a.contains(self.source_for(step)),
}
}
}
/// One executed pipeline plus its per-step [`StepDisposition`] taxonomy.
#[derive(Debug, Clone)]
pub struct CachedRunReport {
/// The Phase-0-shaped run report (valid for historization + the facets).
pub report: RunReport,
/// `(step_name → disposition)` in topo order, aligned to `report.steps`.
pub dispositions: Vec<(String, StepDisposition)>,
/// `(step_name → restored file names)` for every reused step whose artifact was
/// auto-materialized back onto disk this run (see
/// [`IncrementalOptions::materialize_dir`]). Empty unless a materialize dir + an
/// artifact store were both wired in.
pub materialized: Vec<(String, Vec<String>)>,
}
impl CachedRunReport {
/// True when the run is green (no failed step).
pub fn is_green(&self) -> bool {
self.report.is_green()
}
/// How many steps actually executed for real this run.
pub fn ran(&self) -> usize {
self.count(StepDisposition::Ran)
}
/// How many steps were served from the action cache (content-id hit).
pub fn cache_hits(&self) -> usize {
self.count(StepDisposition::CacheHit)
}
/// How many steps were skipped as change-proven-unchanged.
pub fn unchanged(&self) -> usize {
self.count(StepDisposition::Unchanged)
}
fn count(&self, want: StepDisposition) -> usize {
self.dispositions.iter().filter(|(_, d)| *d == want).count()
}
/// How many artifact files were auto-restored to disk on reuse this run (the
/// total across every reused step whose output was materialized).
pub fn materialized_files(&self) -> usize {
self.materialized.iter().map(|(_, files)| files.len()).sum()
}
/// The `state_json` slice: per step **ran / cache-hit / unchanged /
/// when-skipped / upstream-skipped / failed**, plus a `cached` flag the viz
/// DAG reads to tint reused stations. Distinct from a plain
/// [`crate::brokkr::facets::DagFacet`] so the finer taxonomy is never flattened.
pub fn state_json(&self) -> serde_json::Value {
let steps: Vec<serde_json::Value> = self
.report
.steps
.iter()
.zip(&self.dispositions)
.map(|(row, (_, disp))| {
serde_json::json!({
"step": row.step,
"stage": row.stage,
"disposition": disp.label(),
"status": row.status.label(),
"cached": disp.cached(),
"summary": row.summary,
"artifact_id": row.artifact_id,
})
})
.collect();
serde_json::json!({
"run_id": self.report.run.run_id,
"pipeline": self.report.run.pipeline,
"status": self.report.run.status.label(),
"steps": steps,
"ran": self.ran(),
"cache_hits": self.cache_hits(),
"unchanged": self.unchanged(),
"materialized_files": self.materialized_files(),
"materialized": self.materialized.iter().map(|(step, files)| serde_json::json!({
"step": step,
"files": files,
})).collect::<Vec<_>>(),
})
}
}
/// Look up a step by name in the pipeline (mirrors the engine's private helper).
fn find_step<'a>(pipeline: &'a Pipeline, name: &str) -> Option<&'a Step> {
pipeline
.stages
.iter()
.flat_map(|s| s.steps.iter())
.find(|st| st.name == name)
}
/// Render a matrix map as a stable `k=v,k=v` string.
fn render_matrix(m: &BTreeMap<String, String>) -> String {
m.iter()
.map(|(k, v)| format!("{k}={v}"))
.collect::<Vec<_>>()
.join(",")
}
/// Run `pipeline`'s `dag` through `runner`, applying the action cache and the
/// change blast radius per [`IncrementalOptions`]. Records results into `cache`.
///
/// The walk is the engine's fail-fast topo walk, extended per step with:
/// upstream-failed → `UpstreamSkipped`; `when` gate miss → `WhenSkipped`; a
/// content-id computed from {argv, matrix, upstream artifact pins, source
/// fingerprint}; then either a change-proven `Unchanged` skip, a cache `CacheHit`
/// skip (both feed the cached artifact forward), or a real `Ran`/`Failed`
/// execution recorded back into the cache.
pub fn run_pipeline_cached(
pipeline: &Pipeline,
dag: &Dag,
runner: &dyn StepRunner,
req: &RunRequest,
run_id: &str,
cache: &mut ActionCache,
opts: &IncrementalOptions,
) -> CachedRunReport {
run_pipeline_cached_reported(
pipeline,
dag,
runner,
req,
run_id,
cache,
opts,
&jera::jobs::Reporter::noop(),
)
}
/// [`run_pipeline_cached`] with progress. ONE implementation — the unreported
/// entry point above delegates here with [`jera::jobs::Reporter::noop`], so there
/// is no second walk to keep in step (L5).
///
/// The phase name carries the DISPOSITION as well as the position, because in this
/// walk "step 7 of 30" is nearly meaningless: most steps are cache hits that take
/// microseconds. A reader wants to know which steps actually ran. Emitted per step
/// AFTER the disposition is decided, plus one terminal phase folding the
/// ran/cache-hit/unchanged tally.
#[allow(clippy::too_many_arguments)]
pub fn run_pipeline_cached_reported(
pipeline: &Pipeline,
dag: &Dag,
runner: &dyn StepRunner,
req: &RunRequest,
run_id: &str,
cache: &mut ActionCache,
opts: &IncrementalOptions,
reporter: &jera::jobs::Reporter,
) -> CachedRunReport {
let matrix_str = render_matrix(&req.matrix);
let step_total = dag.order.len() as u64;
let ctx = StepContext {
matrix: req.matrix.clone(),
};
let started = std::time::Instant::now();
let mut done: BTreeSet<String> = BTreeSet::new();
let mut failed: BTreeSet<String> = BTreeSet::new();
// step → the artifact id it produced/reused (fed forward as an upstream pin).
let mut artifacts: BTreeMap<String, Option<String>> = BTreeMap::new();
// steps that actually re-ran this walk (propagate the blast radius downstream).
let mut reran: BTreeSet<String> = BTreeSet::new();
let mut rows: Vec<BuildRow> = Vec::new();
let mut dispositions: Vec<(String, StepDisposition)> = Vec::new();
// reused steps whose artifact bytes were auto-restored to disk this run.
let mut materialized: Vec<(String, Vec<String>)> = Vec::new();
let mut passed = 0usize;
for (idx, node) in dag.order.iter().enumerate() {
let done_count = idx as u64;
reporter.progress_note(
done_count,
Some(step_total),
format!("{}/{}", node.stage, node.step),
);
reporter.phase(
format!(
"{}/{} {}/{}",
done_count + 1,
step_total,
node.stage,
node.step
),
failed.is_empty(),
);
// ── fail-fast cascade: an upstream failure/skip poisons this step ──
if node.needs.iter().any(|n| failed.contains(n)) {
failed.insert(node.step.clone());
rows.push(skip_row(
run_id,
node,
&matrix_str,
"skipped: an upstream step failed",
));
dispositions.push((node.step.clone(), StepDisposition::UpstreamSkipped));
continue;
}
let step = find_step(pipeline, &node.step).expect("dag node names a declared step");
// ── `when` gate: off-branch/off-event steps are WhenSkipped (not run) ──
if let Some(when) = &step.when
&& !when.passes(req.branch.as_deref(), req.event.as_deref())
{
done.insert(node.step.clone());
artifacts.insert(node.step.clone(), None);
rows.push(skip_row(
run_id,
node,
&matrix_str,
"skipped: `when` gate did not match",
));
dispositions.push((node.step.clone(), StepDisposition::WhenSkipped));
continue;
}
// Resolved upstream artifact pins this step consumes (its `needs`' outputs).
let upstream_pins: Vec<String> = node
.needs
.iter()
.filter_map(|n| artifacts.get(n).cloned().flatten())
.collect();
let input_id = step_input_id(
&step.run,
&req.matrix,
&upstream_pins,
opts.source_for(&node.step),
);
// Did any upstream step actually re-run? Then this step is downstream of a
// real change and must be a candidate even if change-detection excluded it.
let upstream_reran = node.needs.iter().any(|n| reran.contains(n));
// ── (1) change-based skip: proven unaffected → Unchanged (reuse cache) ──
if !opts.full
&& !opts.in_blast_radius(&node.step)
&& !upstream_reran
&& let Some(entry) = cache
.get(&input_id)
.filter(|e| e.status.is_green())
.cloned()
{
done.insert(node.step.clone());
artifacts.insert(node.step.clone(), entry.artifact_id.clone());
passed += 1;
auto_materialize(
opts,
&node.step,
entry.artifact_id.as_deref(),
&mut materialized,
);
rows.push(reuse_row(run_id, node, &matrix_str, "unchanged", &entry));
dispositions.push((node.step.clone(), StepDisposition::Unchanged));
continue;
}
// Proven-unchanged but no cached output to feed forward → fall through
// and run it (a skip we cannot safely satisfy downstream is unsafe).
// ── (2) action cache: content-id hit → CacheHit (reuse result) ──
if opts.use_cache
&& !opts.full
&& let Some(entry) = cache
.get(&input_id)
.filter(|e| e.status.is_green())
.cloned()
{
done.insert(node.step.clone());
artifacts.insert(node.step.clone(), entry.artifact_id.clone());
passed += 1;
auto_materialize(
opts,
&node.step,
entry.artifact_id.as_deref(),
&mut materialized,
);
rows.push(reuse_row(run_id, node, &matrix_str, "cache-hit", &entry));
dispositions.push((node.step.clone(), StepDisposition::CacheHit));
continue;
}
// ── (3) cache miss / forced / affected → run it for real ──
let step_started = std::time::Instant::now();
let StepOutcome {
ok,
summary,
artifact_id,
artifact_files,
} = runner.run(step, &ctx);
let dur = step_started.elapsed().as_millis() as u64;
reran.insert(node.step.clone());
// If an artifact store is wired in and the step produced artifact bytes, push
// EVERY produced file to holger and key the recorded artifact on their TRUE
// content id — a single file records its own id, a multi-artifact build records
// a manifest of the whole set (the manifest's id becomes the artifact_id), so a
// later hit on ANY machine can resolve every real output. Best-effort: on any
// push error we fall back to the runner's own artifact_id (no regression, the
// run stays green — a store the forge can't reach never fails a build).
let artifact_id = match (ok, &opts.artifact_store) {
(true, Some(store)) if !artifact_files.is_empty() => {
match crate::artifact_store::put_artifacts(store.as_ref(), &artifact_files) {
Ok(Some(id)) => Some(id),
_ => artifact_id,
}
}
_ => artifact_id,
};
if ok {
done.insert(node.step.clone());
artifacts.insert(node.step.clone(), artifact_id.clone());
passed += 1;
if opts.use_cache {
cache.insert(ActionCacheEntry {
input_id,
step: node.step.clone(),
status: RunStatus::Passed,
summary: summary.clone(),
artifact_id: artifact_id.clone(),
recorded_at: Utc::now().to_rfc3339(),
});
}
} else {
failed.insert(node.step.clone());
}
rows.push(BuildRow {
run_id: run_id.to_string(),
step: node.step.clone(),
stage: node.stage.clone(),
backend: node.backend.to_string(),
matrix: matrix_str.clone(),
status: if ok {
RunStatus::Passed
} else {
RunStatus::Failed
},
summary,
artifact_id,
job_ref: None,
duration_ms: dur,
finished_at: Utc::now().to_rfc3339(),
});
dispositions.push((
node.step.clone(),
if ok {
StepDisposition::Ran
} else {
StepDisposition::Failed
},
));
}
let total = dag.order.len();
let status = if failed.is_empty() {
RunStatus::Passed
} else {
RunStatus::Failed
};
let run = PipelineRunRow {
run_id: run_id.to_string(),
pipeline: pipeline.name.clone(),
project: req.project.clone(),
trigger: req.trigger.clone(),
sha: req.sha.clone(),
status,
steps: total,
passed,
duration_ms: started.elapsed().as_millis() as u64,
job_ref: None,
finished_at: Utc::now().to_rfc3339(),
};
let out = CachedRunReport {
report: RunReport { run, steps: rows },
dispositions,
materialized,
};
// The terminal phase folds the taxonomy, so a reader polling after the run
// sees what the walk actually DID, not the last step still in flight.
reporter.progress_note(
step_total,
Some(step_total),
format!(
"{} ran, {} cached, {} unchanged",
out.ran(),
out.cache_hits(),
out.unchanged()
),
);
reporter.phase(
if out.is_green() {
format!("done: {} ran, {} cached", out.ran(), out.cache_hits())
} else {
format!(
"failed: {}/{} passed",
out.report.run.passed, out.report.run.steps
)
},
out.is_green(),
);
out
}
/// Auto-restore a REUSED (cache-hit / unchanged) step's artifact bytes back onto
/// disk under [`IncrementalOptions::materialize_dir`], so a downstream *real* step
/// finds the upstream output even though this step was skipped. Records the restored
/// file names against `step` in `materialized`. A no-op unless BOTH a materialize
/// dir and an artifact store are wired in and the reused step actually pinned an
/// `artifact_id`. Best-effort: a restore failure is logged and skipped — a store the
/// forge can't restore from never fails the run (downstream simply rebuilds).
fn auto_materialize(
opts: &IncrementalOptions,
step: &str,
artifact_id: Option<&str>,
materialized: &mut Vec<(String, Vec<String>)>,
) {
let (Some(dir), Some(store), Some(id)) =
(&opts.materialize_dir, &opts.artifact_store, artifact_id)
else {
return;
};
match crate::artifact_store::materialize(store.as_ref(), id, dir, step) {
Ok(files) if !files.is_empty() => {
materialized.push((
step.to_string(),
files.into_iter().map(|(n, _)| n).collect(),
));
}
Ok(_) => {} // an id-only label / miss: nothing restored, no regression.
Err(e) => {
eprintln!(
"dwarves: auto-materialize of reused step '{step}' ({id}) failed: {e:#}; \
downstream will rebuild if it needs the bytes"
);
}
}
}
/// A `Skipped` build row with a reason (upstream-failed / when-gated).
fn skip_row(
run_id: &str,
node: &crate::brokkr::dag::DagNode,
matrix: &str,
reason: &str,
) -> BuildRow {
BuildRow {
run_id: run_id.to_string(),
step: node.step.clone(),
stage: node.stage.clone(),
backend: node.backend.to_string(),
matrix: matrix.to_string(),
status: RunStatus::Skipped,
summary: reason.to_string(),
artifact_id: None,
job_ref: None,
duration_ms: 0,
finished_at: Utc::now().to_rfc3339(),
}
}
/// A reused (cache-hit / unchanged) build row — green, artifact fed forward, its
/// summary prefixed with the reuse reason so even the plain facets/log show it.
fn reuse_row(
run_id: &str,
node: &crate::brokkr::dag::DagNode,
matrix: &str,
reason: &str,
entry: &ActionCacheEntry,
) -> BuildRow {
BuildRow {
run_id: run_id.to_string(),
step: node.step.clone(),
stage: node.stage.clone(),
backend: node.backend.to_string(),
matrix: matrix.to_string(),
status: RunStatus::Passed,
summary: format!("{reason}: {}", entry.summary),
artifact_id: entry.artifact_id.clone(),
job_ref: None,
duration_ms: 0,
finished_at: Utc::now().to_rfc3339(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::brokkr::PipelineDescriptor;
use crate::brokkr::runner::FakeRunner;
const PIPE: &str = r#"
[pipeline]
name = "p"
[[pipeline.stage]]
name = "build"
[[pipeline.stage.step]]
name = "compile"
run = { variant = "server" }
[[pipeline.stage]]
name = "test"
[[pipeline.stage.step]]
name = "unit"
needs = ["compile"]
run = { variant = "server" }
"#;
fn setup(toml: &str) -> (Pipeline, Dag) {
let d = PipelineDescriptor::from_toml_str(toml).unwrap();
let dag = Dag::build(&d.pipeline).unwrap();
(d.pipeline, dag)
}
fn req() -> RunRequest {
RunRequest {
project: "proj".into(),
trigger: "manual".into(),
..Default::default()
}
}
#[test]
fn cache_off_is_phase0_identical_every_step_ran() {
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
let out = run_pipeline_cached(
&p,
&dag,
&FakeRunner::new(),
&req(),
"r0",
&mut cache,
&IncrementalOptions::off(),
);
assert!(out.is_green());
assert_eq!(out.ran(), 2, "both steps executed");
assert_eq!(out.cache_hits(), 0);
assert!(cache.is_empty(), "cache-off records nothing");
// The embedded report matches the Phase-0 shape.
assert_eq!(out.report.run.passed, 2);
assert_eq!(out.report.steps.len(), 2);
}
#[test]
fn second_identical_run_hits_the_cache_on_every_step() {
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
let opts = IncrementalOptions::cached();
let runner = FakeRunner::new();
let first = run_pipeline_cached(&p, &dag, &runner, &req(), "r1", &mut cache, &opts);
assert_eq!(first.ran(), 2, "first run executes both steps");
assert_eq!(runner.order().len(), 2);
// Second identical run: byte-identical inputs → every step is a cache hit.
let second = run_pipeline_cached(&p, &dag, &runner, &req(), "r2", &mut cache, &opts);
assert_eq!(second.cache_hits(), 2, "both steps served from cache");
assert_eq!(second.ran(), 0, "nothing re-ran");
assert!(second.is_green(), "a cache-hit run is still green");
assert_eq!(runner.order().len(), 2, "the runner was NOT invoked again");
// Downstream still satisfied: `unit` reused compile's fed-forward artifact.
let unit = second
.report
.steps
.iter()
.find(|r| r.step == "unit")
.unwrap();
assert!(
unit.artifact_id.is_some(),
"cached artifact fed forward to downstream"
);
}
#[test]
fn changed_argv_misses_and_reruns() {
let (p1, dag1) = setup(PIPE);
let mut cache = ActionCache::in_memory();
let opts = IncrementalOptions::cached();
run_pipeline_cached(
&p1,
&dag1,
&FakeRunner::new(),
&req(),
"r1",
&mut cache,
&opts,
);
// Change `compile`'s argv (variant server → client): its content-id changes
// → miss → it re-runs (and `unit`, downstream of a re-run, also re-runs).
let changed = PIPE.replacen(
"name = \"compile\"\nrun = { variant = \"server\" }",
"name = \"compile\"\nrun = { variant = \"client\" }",
1,
);
let (p2, dag2) = setup(&changed);
let runner = FakeRunner::new();
let out = run_pipeline_cached(&p2, &dag2, &runner, &req(), "r2", &mut cache, &opts);
let compile = out
.dispositions
.iter()
.find(|(s, _)| s == "compile")
.unwrap()
.1;
assert_eq!(
compile,
StepDisposition::Ran,
"changed argv misses the cache"
);
assert!(runner.order().contains(&"compile".to_string()));
}
/// A runner that publishes a caller-chosen artifact id per step name (so a
/// test can make an upstream step produce a *different* output on a later run).
struct MapRunner {
arts: BTreeMap<String, String>,
}
impl StepRunner for MapRunner {
fn run(&self, step: &Step, _ctx: &StepContext) -> StepOutcome {
let art = self
.arts
.get(&step.name)
.cloned()
.unwrap_or_else(|| format!("art-{}", step.name));
StepOutcome::ok(format!("built {}", step.name)).with_artifact(art)
}
}
#[test]
fn changed_upstream_artifact_misses_the_downstream_step() {
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
let opts = IncrementalOptions::cached();
// Run 1: compile publishes artifact "A"; `unit` caches keyed off upstream
// pin "A".
let r1 = MapRunner {
arts: BTreeMap::from([("compile".to_string(), "A".to_string())]),
};
run_pipeline_cached(&p, &dag, &r1, &req(), "r1", &mut cache, &opts);
// Run 2: invalidate ONLY compile (a source-id change) so it re-runs and now
// publishes a *different* artifact "B". `unit`'s own argv/source are
// unchanged, but its upstream artifact pin shifts A→B, so its input
// content-id changes → it MUST miss the cache and re-run.
let opts2 = IncrementalOptions {
source_ids: BTreeMap::from([("compile".to_string(), "rev2".to_string())]),
..IncrementalOptions::cached()
};
let r2 = MapRunner {
arts: BTreeMap::from([("compile".to_string(), "B".to_string())]),
};
let out = run_pipeline_cached(&p, &dag, &r2, &req(), "r2", &mut cache, &opts2);
let compile = out
.dispositions
.iter()
.find(|(s, _)| s == "compile")
.unwrap()
.1;
let unit = out
.dispositions
.iter()
.find(|(s, _)| s == "unit")
.unwrap()
.1;
assert_eq!(
compile,
StepDisposition::Ran,
"compile source changed → miss"
);
assert_eq!(
unit,
StepDisposition::Ran,
"changed upstream artifact (A→B) shifts unit's pin → downstream misses"
);
// The fed-forward pin is the NEW artifact.
let unit_row = out.report.steps.iter().find(|r| r.step == "unit").unwrap();
assert_eq!(unit_row.artifact_id.as_deref(), Some("art-unit"));
}
#[test]
fn incremental_skips_unaffected_steps_as_unchanged_and_feeds_artifacts_forward() {
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
// Warm the cache with a full run.
run_pipeline_cached(
&p,
&dag,
&FakeRunner::new(),
&req(),
"warm",
&mut cache,
&IncrementalOptions::cached(),
);
// Now a change-detection blast radius that excludes both steps: everything
// is proven unaffected → skipped as Unchanged, artifacts fed forward.
let opts = IncrementalOptions {
affected: Some(BTreeSet::new()),
..IncrementalOptions::cached()
};
let runner = FakeRunner::new();
let out = run_pipeline_cached(&p, &dag, &runner, &req(), "inc", &mut cache, &opts);
assert_eq!(out.unchanged(), 2, "both steps proven unaffected");
assert_eq!(out.ran(), 0, "nothing executed");
assert!(runner.order().is_empty(), "the runner was never invoked");
assert!(out.is_green());
// Downstream `unit` still got compile's artifact fed forward.
let unit = out.report.steps.iter().find(|r| r.step == "unit").unwrap();
assert!(unit.artifact_id.is_some());
}
#[test]
fn affected_but_content_identical_early_cuts_and_downstream_stays_unchanged() {
// Early-cutoff: a step inside the blast radius is a *candidate* to run, but
// if its input content-id is byte-identical the action cache still serves
// it (a CacheHit) — and because it did not actually re-run, a downstream
// step outside the radius stays a proven-Unchanged skip. This is the
// blast-radius (coarse) + content-cache (fine) interplay.
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
run_pipeline_cached(
&p,
&dag,
&FakeRunner::new(),
&req(),
"warm",
&mut cache,
&IncrementalOptions::cached(),
);
let mut affected = BTreeSet::new();
affected.insert("compile".to_string());
let opts = IncrementalOptions {
affected: Some(affected),
..IncrementalOptions::cached()
};
let runner = FakeRunner::new();
let out = run_pipeline_cached(&p, &dag, &runner, &req(), "inc", &mut cache, &opts);
let compile = out
.dispositions
.iter()
.find(|(s, _)| s == "compile")
.unwrap()
.1;
let unit = out
.dispositions
.iter()
.find(|(s, _)| s == "unit")
.unwrap()
.1;
assert_eq!(
compile,
StepDisposition::CacheHit,
"affected but identical → early cutoff"
);
assert_eq!(
unit,
StepDisposition::Unchanged,
"excluded + upstream not re-run → unchanged"
);
assert!(runner.order().is_empty(), "nothing executed");
}
#[test]
fn affected_step_with_changed_source_reruns_and_propagates_via_artifact() {
// The genuine affected path: a step is affected *because* its source moved,
// so its content-id changes → it really re-runs. When its output artifact
// also changes, the downstream step's upstream pin shifts → it re-runs too.
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
let r1 = MapRunner {
arts: BTreeMap::from([("compile".to_string(), "A".to_string())]),
};
run_pipeline_cached(
&p,
&dag,
&r1,
&req(),
"warm",
&mut cache,
&IncrementalOptions::cached(),
);
let opts = IncrementalOptions {
affected: Some(BTreeSet::from(["compile".to_string()])),
source_ids: BTreeMap::from([("compile".to_string(), "rev2".to_string())]),
..IncrementalOptions::cached()
};
let r2 = MapRunner {
arts: BTreeMap::from([("compile".to_string(), "B".to_string())]),
};
let out = run_pipeline_cached(&p, &dag, &r2, &req(), "inc", &mut cache, &opts);
let compile = out
.dispositions
.iter()
.find(|(s, _)| s == "compile")
.unwrap()
.1;
let unit = out
.dispositions
.iter()
.find(|(s, _)| s == "unit")
.unwrap()
.1;
assert_eq!(
compile,
StepDisposition::Ran,
"affected + source changed → re-runs"
);
assert_eq!(
unit,
StepDisposition::Ran,
"changed upstream artifact propagates downstream"
);
}
#[test]
fn full_forces_every_step_to_rerun_even_with_a_warm_cache() {
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
run_pipeline_cached(
&p,
&dag,
&FakeRunner::new(),
&req(),
"warm",
&mut cache,
&IncrementalOptions::cached(),
);
let runner = FakeRunner::new();
let out = run_pipeline_cached(
&p,
&dag,
&runner,
&req(),
"full",
&mut cache,
&IncrementalOptions::forced(),
);
assert_eq!(out.ran(), 2, "--full ignores the warm cache");
assert_eq!(out.cache_hits(), 0);
assert_eq!(runner.order().len(), 2, "both steps really executed");
}
#[test]
fn a_failed_step_is_never_cached_and_reruns_next_time() {
// The core safety invariant of the action cache: only GREEN steps are
// recorded. A failed step must leave NO entry, so a later run re-executes
// it (never serving a stale/failed result as a hit) — and its downstream
// dependents, upstream-skipped this run, also carry no poison.
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
let opts = IncrementalOptions::cached();
// Run 1: compile fails → unit is upstream-skipped. Nothing green ran, so
// the cache stays EMPTY (no failed entry, no skipped entry).
let bad = FakeRunner::new().fail_step("compile");
let r1 = run_pipeline_cached(&p, &dag, &bad, &req(), "r1", &mut cache, &opts);
assert!(!r1.is_green(), "run failed");
let compile1 = r1
.dispositions
.iter()
.find(|(s, _)| s == "compile")
.unwrap()
.1;
let unit1 = r1.dispositions.iter().find(|(s, _)| s == "unit").unwrap().1;
assert_eq!(compile1, StepDisposition::Failed);
assert_eq!(unit1, StepDisposition::UpstreamSkipped);
assert!(
cache.is_empty(),
"a failed run records NOTHING into the action cache"
);
// Run 2: same inputs, now the runner greens compile. Because run 1 cached
// nothing, compile is a genuine miss and RE-RUNS (never a cache hit off a
// prior failure); unit then runs for real too.
let good = FakeRunner::new();
let r2 = run_pipeline_cached(&p, &dag, &good, &req(), "r2", &mut cache, &opts);
assert!(r2.is_green());
let compile2 = r2
.dispositions
.iter()
.find(|(s, _)| s == "compile")
.unwrap()
.1;
assert_eq!(
compile2,
StepDisposition::Ran,
"a previously-failed step re-runs, never hits"
);
assert_eq!(r2.ran(), 2, "both steps executed for real");
assert_eq!(r2.cache_hits(), 0);
assert!(good.order().contains(&"compile".to_string()));
assert_eq!(cache.len(), 2, "now both greens are recorded");
}
#[test]
fn a_global_source_fingerprint_bump_busts_every_step() {
// `default_source_id` is the pipeline-wide source fingerprint (e.g. the
// built tree sha) applied to every step without a specific `source_ids`
// entry. Bumping it must change EVERY step's input content-id → the whole
// pipeline misses and re-runs, even against a fully warm cache. This is the
// "the source moved, rebuild it all" path.
let (p, dag) = setup(PIPE);
let mut cache = ActionCache::in_memory();
let rev1 = IncrementalOptions {
default_source_id: "rev1".into(),
..IncrementalOptions::cached()
};
// Warm at rev1: both steps run + record keyed off source=rev1.
let warm = run_pipeline_cached(
&p,
&dag,
&FakeRunner::new(),
&req(),
"warm",
&mut cache,
&rev1,
);
assert_eq!(warm.ran(), 2);
// Re-run at the SAME rev1: byte-identical inputs → every step is a hit.
let same = run_pipeline_cached(
&p,
&dag,
&FakeRunner::new(),
&req(),
"same",
&mut cache,
&rev1,
);
assert_eq!(
same.cache_hits(),
2,
"unchanged global fingerprint still hits"
);
// Bump the global fingerprint to rev2: every step's content-id shifts →
// the warm cache is bypassed and everything re-runs.
let rev2 = IncrementalOptions {
default_source_id: "rev2".into(),
..IncrementalOptions::cached()
};
let runner = FakeRunner::new();
let out = run_pipeline_cached(&p, &dag, &runner, &req(), "rev2", &mut cache, &rev2);
assert_eq!(out.ran(), 2, "a global source bump re-runs every step");
assert_eq!(
out.cache_hits(),
0,
"no step served from the (now stale-keyed) cache"
);
assert_eq!(runner.order().len(), 2, "both steps really executed");
}
/// A runner that writes a real file (the "built artifact") to a temp dir and
/// reports its path — so the artifact-store push path can capture genuine bytes.
struct FileRunner {
dir: std::path::PathBuf,
/// step-name → the bytes that step "builds".
bytes: BTreeMap<String, Vec<u8>>,
}
impl StepRunner for FileRunner {
fn run(&self, step: &Step, _ctx: &StepContext) -> StepOutcome {
let content = self
.bytes
.get(&step.name)
.cloned()
.unwrap_or_else(|| step.name.as_bytes().to_vec());
let path = self.dir.join(format!("{}.bin", step.name));
std::fs::write(&path, &content).unwrap();
StepOutcome::ok(format!("built {}", step.name))
.with_artifact("runner-label-ignored")
.with_artifact_file(path)
}
}
/// An in-memory [`ArtifactStore`] test double: same content-id discipline as the
/// real holger-backed store (the id IS the brokkr content id), but bytes live in
/// a map so the forge wiring is provable without linking holger. Stands in for
/// holger in the incremental-wiring roundtrip; the real holger seal/resolve is
/// covered by the `holger`-feature roundtrip test.
#[derive(Debug, Default)]
struct MemStore {
objects: std::sync::Mutex<BTreeMap<String, Vec<u8>>>,
}
impl crate::artifact_store::ArtifactStore for MemStore {
fn put_bytes(&self, bytes: &[u8]) -> anyhow::Result<String> {
let id = crate::brokkr::artifact::content_id(bytes);
self.objects
.lock()
.unwrap()
.insert(id.clone(), bytes.to_vec());
Ok(id)
}
fn get(&self, content_id: &str) -> anyhow::Result<Option<Vec<u8>>> {
Ok(self.objects.lock().unwrap().get(content_id).cloned())
}
}
#[test]
fn artifact_store_captures_produced_bytes_and_a_hit_resolves_the_real_output() {
// The whole point of the store: after a real run, the recorded artifact_id is
// the TRUE content id of the produced bytes (not the runner's label), the
// bytes live in the store, and a later cache HIT resolves them byte-exact.
let (p, dag) = setup(PIPE);
let tmp = tempfile::tempdir().unwrap();
let store = Arc::new(MemStore::default());
let mut cache = ActionCache::in_memory();
let opts = IncrementalOptions::cached().with_artifact_store(store.clone());
let runner = FileRunner {
dir: tmp.path().to_path_buf(),
bytes: BTreeMap::from([
("compile".to_string(), b"COMPILED-OUTPUT".to_vec()),
("unit".to_string(), b"TEST-REPORT".to_vec()),
]),
};
let first = run_pipeline_cached(&p, &dag, &runner, &req(), "r1", &mut cache, &opts);
assert_eq!(first.ran(), 2);
// The recorded artifact_id is the real content id, and the bytes are stored.
let compile_id = first
.report
.steps
.iter()
.find(|r| r.step == "compile")
.unwrap()
.artifact_id
.clone()
.unwrap();
assert_eq!(
compile_id,
crate::brokkr::artifact::content_id(b"COMPILED-OUTPUT")
);
assert!(
store.has(&compile_id),
"produced bytes are pushed to the store"
);
assert_eq!(
store.objects.lock().unwrap().len(),
2,
"both step outputs captured"
);
// Second identical run: every step is a cache hit — and the fed-forward
// artifact_id still resolves to the exact original bytes in the store.
let second = run_pipeline_cached(&p, &dag, &runner, &req(), "r2", &mut cache, &opts);
assert_eq!(second.cache_hits(), 2, "content-identical inputs hit");
let compile_hit = second
.report
.steps
.iter()
.find(|r| r.step == "compile")
.unwrap()
.artifact_id
.clone()
.unwrap();
assert_eq!(
compile_hit, compile_id,
"the hit feeds forward the same content id"
);
let resolved = store
.get(&compile_hit)
.unwrap()
.expect("cached output resolves");
assert_eq!(resolved, b"COMPILED-OUTPUT", "byte-exact resolve on a hit");
crate::selftest::emit(
"dwarves::artifact_store",
"push_resolve_roundtrip",
resolved == b"COMPILED-OUTPUT" && compile_hit == compile_id,
"produced bytes push to the store and a hit resolves them byte-exact",
);
}
/// A runner whose step emits **several** artifact files (a multi-artifact variant
/// build — e.g. holger's embedded/mini/full binaries in one step). Writes each to
/// a temp dir and reports ALL of them via `with_artifact_files`.
struct MultiFileRunner {
dir: std::path::PathBuf,
/// step-name → the `(name, bytes)` set that step "builds".
outputs: BTreeMap<String, Vec<(String, Vec<u8>)>>,
}
impl StepRunner for MultiFileRunner {
fn run(&self, step: &Step, _ctx: &StepContext) -> StepOutcome {
let set = self.outputs.get(&step.name).cloned().unwrap_or_default();
let files = set.into_iter().map(|(name, bytes)| {
let path = self.dir.join(&name);
std::fs::write(&path, &bytes).unwrap();
crate::brokkr::runner::ArtifactFile::new(name, path)
});
StepOutcome::ok(format!("built {}", step.name))
.with_artifact("runner-label-ignored")
.with_artifact_files(files)
}
}
#[test]
fn a_multi_artifact_step_captures_every_output_and_records_a_manifest() {
// The multi-artifact fix: a step that emits N>1 files must push ALL N to the
// store (not just artifact[0]) AND record a manifest of the whole set whose id
// becomes the step's artifact_id. RED-when-broken: capturing only the first
// artifact drops members / mis-keys the id; both assertions below then fail.
let (p, dag) = setup(PIPE);
let tmp = tempfile::tempdir().unwrap();
let store = Arc::new(MemStore::default());
let mut cache = ActionCache::in_memory();
let opts = IncrementalOptions::cached().with_artifact_store(store.clone());
// `compile` emits THREE binaries; `unit` emits ONE (the single-artifact path).
let runner = MultiFileRunner {
dir: tmp.path().to_path_buf(),
outputs: BTreeMap::from([
(
"compile".to_string(),
vec![
("holger-embedded".to_string(), b"EMBEDDED-BIN".to_vec()),
("holger-mini".to_string(), b"MINI-BIN".to_vec()),
("holger-full".to_string(), b"FULL-BIN".to_vec()),
],
),
(
"unit".to_string(),
vec![("report".to_string(), b"TEST-REPORT".to_vec())],
),
]),
};
let first = run_pipeline_cached(&p, &dag, &runner, &req(), "r1", &mut cache, &opts);
assert_eq!(first.ran(), 2);
// Every one of the three produced binaries is in the store, byte-exact.
for bytes in [&b"EMBEDDED-BIN"[..], &b"MINI-BIN"[..], &b"FULL-BIN"[..]] {
let id = crate::brokkr::artifact::content_id(bytes);
assert!(store.has(&id), "member {:?} pushed to the store", bytes);
}
// The step's recorded artifact_id is the MANIFEST id (not any single member),
// and resolving it yields a manifest listing all three (name → content_id).
let compile_id = first
.report
.steps
.iter()
.find(|r| r.step == "compile")
.unwrap()
.artifact_id
.clone()
.unwrap();
let manifest_bytes = store
.get(&compile_id)
.unwrap()
.expect("manifest object stored");
let manifest =
crate::artifact_store::ArtifactManifest::from_bytes(&manifest_bytes).unwrap();
assert_eq!(
manifest.artifacts.len(),
3,
"manifest carries the whole set, not one"
);
let names: Vec<&str> = manifest.artifacts.iter().map(|a| a.name.as_str()).collect();
assert_eq!(
names,
["holger-embedded", "holger-full", "holger-mini"],
"sorted, all present"
);
for a in &manifest.artifacts {
assert!(
store.has(&a.content_id),
"manifest member {} resolves",
a.name
);
}
// The single-artifact step is byte-identical to the pre-multi path: its own
// content id is recorded, and NO manifest object is written for it.
let unit_id = first
.report
.steps
.iter()
.find(|r| r.step == "unit")
.unwrap()
.artifact_id
.clone()
.unwrap();
assert_eq!(
unit_id,
crate::brokkr::artifact::content_id(b"TEST-REPORT"),
"single artifact records its own content id — no manifest"
);
// A second identical build hits the cache and feeds the SAME manifest id
// forward (deterministic manifest ⇒ cross-machine hit works).
let second = run_pipeline_cached(&p, &dag, &runner, &req(), "r2", &mut cache, &opts);
assert_eq!(
second.cache_hits(),
2,
"identical multi-artifact rebuild hits"
);
let compile_hit = second
.report
.steps
.iter()
.find(|r| r.step == "compile")
.unwrap()
.artifact_id
.clone()
.unwrap();
assert_eq!(
compile_hit, compile_id,
"the hit feeds forward the same manifest id"
);
crate::selftest::emit(
"dwarves::artifact_store",
"multi_artifact_manifest_captures_all",
manifest.artifacts.len() == 3 && compile_hit == compile_id,
"a multi-artifact build captures every output + a manifest keying the set",
);
}
#[test]
fn a_cache_hit_auto_materializes_the_reused_output_to_a_fresh_workspace() {
// The deferral this closes: on a cache HIT the artifact_id is fed forward, but
// the BYTES were never re-produced — a downstream real step (or a fresh
// workspace / second machine) needs them on disk. With a materialize_dir wired
// in, the reused output is restored there. RED-when-broken: without the
// auto-restore the fresh dir stays empty and every read below fails.
let (p, dag) = setup(PIPE);
let produce_dir = tempfile::tempdir().unwrap();
let store = Arc::new(MemStore::default());
let mut cache = ActionCache::in_memory();
// Run 1: produce + push bytes to the store (no materialize dir yet).
let runner = FileRunner {
dir: produce_dir.path().to_path_buf(),
bytes: BTreeMap::from([
("compile".to_string(), b"COMPILED-OUTPUT".to_vec()),
("unit".to_string(), b"TEST-REPORT".to_vec()),
]),
};
let opts1 = IncrementalOptions::cached().with_artifact_store(store.clone());
let first = run_pipeline_cached(&p, &dag, &runner, &req(), "r1", &mut cache, &opts1);
assert_eq!(first.ran(), 2);
assert_eq!(
first.materialized_files(),
0,
"a first RAN run restores nothing"
);
// Run 2: a FRESH, EMPTY workspace dir + a materialize dir. Every step is a
// cache hit and its reused bytes are restored into the fresh dir.
let workspace = tempfile::tempdir().unwrap();
assert_eq!(
std::fs::read_dir(workspace.path()).unwrap().count(),
0,
"workspace starts empty"
);
let opts2 = IncrementalOptions::cached()
.with_artifact_store(store.clone())
.with_materialize_dir(workspace.path());
let second = run_pipeline_cached(&p, &dag, &runner, &req(), "r2", &mut cache, &opts2);
assert_eq!(second.cache_hits(), 2, "both steps hit");
assert_eq!(
second.materialized_files(),
2,
"both reused outputs restored to disk"
);
// The single-artifact steps restore under the step name (the caller basename).
assert_eq!(
std::fs::read(workspace.path().join("compile")).unwrap(),
b"COMPILED-OUTPUT"
);
assert_eq!(
std::fs::read(workspace.path().join("unit")).unwrap(),
b"TEST-REPORT"
);
// Observable via state_json (the viz / status seam).
let js = second.state_json();
assert_eq!(js["materialized_files"], 2);
crate::selftest::emit(
"dwarves::incremental",
"cache_hit_auto_materializes",
second.materialized_files() == 2
&& std::fs::read(workspace.path().join("compile")).unwrap() == b"COMPILED-OUTPUT",
"a cache hit restores the reused artifact bytes into the workspace dir",
);
}
#[test]
fn a_hit_on_a_multi_artifact_step_restores_every_member_by_name() {
// The manifest leg: a reused multi-artifact step restores ALL members under
// their recorded names (not just the first, not the manifest bytes).
let (p, dag) = setup(PIPE);
let produce_dir = tempfile::tempdir().unwrap();
let store = Arc::new(MemStore::default());
let mut cache = ActionCache::in_memory();
let runner = MultiFileRunner {
dir: produce_dir.path().to_path_buf(),
outputs: BTreeMap::from([
(
"compile".to_string(),
vec![
("holger-embedded".to_string(), b"EMBEDDED-BIN".to_vec()),
("holger-mini".to_string(), b"MINI-BIN".to_vec()),
("holger-full".to_string(), b"FULL-BIN".to_vec()),
],
),
(
"unit".to_string(),
vec![("report".to_string(), b"TEST-REPORT".to_vec())],
),
]),
};
let opts1 = IncrementalOptions::cached().with_artifact_store(store.clone());
run_pipeline_cached(&p, &dag, &runner, &req(), "r1", &mut cache, &opts1);
let workspace = tempfile::tempdir().unwrap();
let opts2 = IncrementalOptions::cached()
.with_artifact_store(store.clone())
.with_materialize_dir(workspace.path());
let second = run_pipeline_cached(&p, &dag, &runner, &req(), "r2", &mut cache, &opts2);
assert_eq!(second.cache_hits(), 2);
// compile (3 members) + unit (1) = 4 restored files.
assert_eq!(
second.materialized_files(),
4,
"every member of every reused step restored"
);
// The multi-artifact members restore under their RECORDED names (the manifest
// carries them); the single-artifact `unit` step has no recorded filename, so
// it restores under the step name (the caller basename).
assert_eq!(
std::fs::read(workspace.path().join("holger-embedded")).unwrap(),
b"EMBEDDED-BIN"
);
assert_eq!(
std::fs::read(workspace.path().join("holger-mini")).unwrap(),
b"MINI-BIN"
);
assert_eq!(
std::fs::read(workspace.path().join("holger-full")).unwrap(),
b"FULL-BIN"
);
assert_eq!(
std::fs::read(workspace.path().join("unit")).unwrap(),
b"TEST-REPORT"
);
}
#[test]
fn materialize_dir_without_a_store_is_a_noop() {
// Additive invariant: a materialize dir with NO store restores nothing (the
// bytes live in the store) — identical to the pre-materialize path.
let (p, dag) = setup(PIPE);
let tmp = tempfile::tempdir().unwrap();
let workspace = tempfile::tempdir().unwrap();
let mut cache = ActionCache::in_memory();
let runner = FileRunner {
dir: tmp.path().to_path_buf(),
bytes: BTreeMap::new(),
};
// Warm, then a hit run with a materialize dir but no store.
run_pipeline_cached(
&p,
&dag,
&runner,
&req(),
"warm",
&mut cache,
&IncrementalOptions::cached(),
);
let opts = IncrementalOptions::cached().with_materialize_dir(workspace.path());
let out = run_pipeline_cached(&p, &dag, &runner, &req(), "hit", &mut cache, &opts);
assert_eq!(out.cache_hits(), 2, "steps still hit");
assert_eq!(out.materialized_files(), 0, "no store ⇒ nothing restored");
assert_eq!(
std::fs::read_dir(workspace.path()).unwrap().count(),
0,
"workspace untouched"
);
}
#[test]
fn without_a_store_behavior_is_unchanged_bytes_are_not_captured() {
// Additive invariant: artifact_store = None ⇒ the artifact_id is the runner's
// own label and nothing is pushed (identical to the pre-store path).
let (p, dag) = setup(PIPE);
let tmp = tempfile::tempdir().unwrap();
let mut cache = ActionCache::in_memory();
let runner = FileRunner {
dir: tmp.path().to_path_buf(),
bytes: BTreeMap::new(),
};
let out = run_pipeline_cached(
&p,
&dag,
&runner,
&req(),
"r1",
&mut cache,
&IncrementalOptions::cached(),
);
let compile = out
.report
.steps
.iter()
.find(|r| r.step == "compile")
.unwrap();
assert_eq!(
compile.artifact_id.as_deref(),
Some("runner-label-ignored"),
"no store ⇒ the runner's label is kept verbatim, bytes untouched"
);
crate::selftest::emit(
"dwarves::artifact_store",
"no_endpoint_is_noop",
compile.artifact_id.as_deref() == Some("runner-label-ignored"),
"no store configured ⇒ pre-store behaviour (label kept, no push)",
);
}
#[test]
fn state_json_distinguishes_the_full_taxonomy() {
// A pipeline exercising ran + when-skip; then a second run for cache-hit.
let toml = r#"
[pipeline]
name = "p"
[[pipeline.stage]]
name = "build"
[[pipeline.stage.step]]
name = "compile"
run = { variant = "server" }
[[pipeline.stage.step]]
name = "deploy"
needs = ["compile"]
run = { variant = "server" }
when = { branch = "main" }
"#;
let (p, dag) = setup(toml);
let mut cache = ActionCache::in_memory();
// On a feature branch: deploy is when-skipped, compile runs.
let r = RunRequest {
project: "proj".into(),
trigger: "manual".into(),
branch: Some("feature/x".into()),
..Default::default()
};
let out = run_pipeline_cached(
&p,
&dag,
&FakeRunner::new(),
&r,
"r1",
&mut cache,
&IncrementalOptions::cached(),
);
let js = out.state_json();
let steps = js["steps"].as_array().unwrap();
let disp = |name: &str| {
steps.iter().find(|s| s["step"] == name).unwrap()["disposition"]
.as_str()
.unwrap()
.to_string()
};
assert_eq!(disp("compile"), "ran");
assert_eq!(disp("deploy"), "when-skipped");
assert_eq!(js["ran"], 1);
// Second run on the same branch: compile is a cache hit; deploy still gated.
let out2 = run_pipeline_cached(
&p,
&dag,
&FakeRunner::new(),
&r,
"r2",
&mut cache,
&IncrementalOptions::cached(),
);
let js2 = out2.state_json();
let steps2 = js2["steps"].as_array().unwrap();
let compile2 = steps2.iter().find(|s| s["step"] == "compile").unwrap();
assert_eq!(compile2["disposition"], "cache-hit");
assert_eq!(
compile2["cached"], true,
"the viz tints this station as reused"
);
}
}