use serde::Serialize;
use crate::release::doctor::{crate_publish_order, RepoGraph};
#[derive(Debug, Clone, Serialize)]
pub struct StageStep {
pub repo: String,
#[serde(default)]
pub deps: Vec<String>,
pub verify: Vec<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct StagePlan {
pub registry: String,
pub branch: String,
pub order: Vec<StageStep>,
pub cycle: Vec<String>,
}
impl StagePlan {
pub fn is_executable(&self) -> bool {
self.cycle.is_empty() && !self.order.is_empty()
}
}
fn direct_dependents(graphs: &[RepoGraph], repo: &str) -> Vec<String> {
let Some(target) = graphs.iter().find(|g| g.repo == repo) else {
return vec![];
};
let mut out: Vec<String> = graphs
.iter()
.filter(|g| g.repo != repo && g.deps.iter().any(|d| target.produces.contains(d)))
.map(|g| g.repo.clone())
.collect();
out.sort();
out
}
fn direct_dependencies(graphs: &[RepoGraph], repo: &str) -> Vec<String> {
let Some(me) = graphs.iter().find(|g| g.repo == repo) else {
return vec![];
};
let mut out: Vec<String> = graphs
.iter()
.filter(|g| g.repo != repo && me.deps.iter().any(|d| g.produces.contains(d)))
.map(|g| g.repo.clone())
.collect();
out.sort();
out
}
pub fn plan_stage(graphs: &[RepoGraph], registry: &str, branch: &str) -> StagePlan {
let topo = crate_publish_order(graphs);
let order = topo
.order
.iter()
.map(|repo| StageStep {
repo: repo.clone(),
deps: direct_dependencies(graphs, repo),
verify: direct_dependents(graphs, repo),
})
.collect();
StagePlan { registry: registry.to_string(), branch: branch.to_string(), order, cycle: topo.cycle }
}
pub fn format_plan(p: &StagePlan) -> String {
let mut s = String::new();
s.push_str("nornir release stage — plan (dry-run)\n\n");
s.push_str(&format!(" registry: {}\n", p.registry));
s.push_str(&format!(" branch: {}\n\n", p.branch));
if !p.cycle.is_empty() {
s.push_str(&format!(
" ⚠ dependency cycle — NOT executable until resolved: {}\n\n",
p.cycle.join(", ")
));
}
s.push_str(" Publish order (dependencies first):\n");
for (i, step) in p.order.iter().enumerate() {
s.push_str(&format!(" {}. cargo publish {} → /sparring\n", i + 1, step.repo));
if !step.verify.is_empty() {
s.push_str(&format!(" verify (build from /sparring): {}\n", step.verify.join(", ")));
}
}
if p.is_executable() {
s.push_str("\n → green plan; run with --execute to publish into /sparring, then promote.\n");
}
s
}
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::process::Command;
#[cfg(not(feature = "embed-holger"))]
use std::process::{Child, Stdio};
use anyhow::{Context, Result};
pub fn cargo_config_toml(http_addr: &str) -> String {
format!(
r#"[registries.sparring]
index = "sparse+http://{http_addr}/sparring/index/"
[source.crates-io]
replace-with = "sparring-mirror"
[source.sparring-mirror]
registry = "sparse+http://{http_addr}/sparring/index/"
[net]
retry = 2
"#
)
}
#[cfg(not(feature = "embed-holger"))]
struct HolgerProc(Child);
#[cfg(not(feature = "embed-holger"))]
impl Drop for HolgerProc {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
#[cfg(not(feature = "embed-holger"))]
fn spawn_holger(holger_bin: &Path, data_dir: &Path, grpc: &str, http: &str) -> Result<HolgerProc> {
let child = Command::new(holger_bin)
.arg("dev-pair")
.arg("--data")
.arg(data_dir)
.args(["--grpc", grpc, "--http", http])
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.with_context(|| format!("spawn holger dev-pair via {}", holger_bin.display()))?;
Ok(HolgerProc(child))
}
#[cfg(not(feature = "embed-holger"))]
fn wait_ready(http_addr: &str) -> Result<()> {
let url = format!("http://{http_addr}/cache/index/config.json");
for _ in 0..60 {
if ureq::get(&url)
.timeout(std::time::Duration::from_millis(500))
.call()
.is_ok()
{
return Ok(());
}
std::thread::sleep(std::time::Duration::from_millis(250));
}
anyhow::bail!("holger dev-pair never became ready at {url}")
}
#[derive(Debug, Default)]
pub struct StageOutcome {
pub published: Vec<String>,
pub already_staged: Vec<String>,
pub verified: Vec<String>,
pub errors: Vec<String>,
pub blocked: Vec<String>,
}
pub(crate) struct BranchGuard {
restores: Vec<(PathBuf, String)>,
branch: String,
}
impl BranchGuard {
pub(crate) fn new(branch: &str) -> Self {
BranchGuard { restores: Vec::new(), branch: branch.to_string() }
}
pub(crate) fn capture(&mut self, path: &Path) {
if let Ok(o) = Command::new("git")
.arg("-C")
.arg(path)
.args(["rev-parse", "--abbrev-ref", "HEAD"])
.output()
{
let prev = String::from_utf8_lossy(&o.stdout).trim().to_string();
if !prev.is_empty() && prev != self.branch {
self.restores.push((path.to_path_buf(), prev));
}
}
}
}
impl Drop for BranchGuard {
fn drop(&mut self) {
for (path, prev) in self.restores.iter().rev() {
let _ = Command::new("git").arg("-C").arg(path).args(["checkout", prev]).status();
let _ = Command::new("git")
.arg("-C")
.arg(path)
.args(["branch", "-D", &self.branch])
.status();
let _ = Command::new("git")
.arg("-C")
.arg(path)
.args(["checkout", "--", "Cargo.lock"])
.output();
eprintln!("stage: restored {} → {prev} (release branch cleaned)", path.display());
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum PublishAttempt {
Published,
AlreadyStaged,
Failed(String),
}
const MAX_ATTEMPTS: u32 = 2;
#[derive(Clone, Copy, PartialEq, Eq)]
enum CascadeState {
Pending,
Done,
Failed,
Blocked,
}
pub(crate) fn run_cascade<F>(order: &[StageStep], mut publish: F) -> StageOutcome
where
F: FnMut(&str) -> PublishAttempt,
{
let n = order.len();
let pos: BTreeMap<&str, usize> =
order.iter().enumerate().map(|(i, s)| (s.repo.as_str(), i)).collect();
let blocking: Vec<Vec<usize>> = order
.iter()
.enumerate()
.map(|(i, s)| {
s.deps
.iter()
.filter_map(|d| pos.get(d.as_str()).copied())
.filter(|&j| j < i)
.collect()
})
.collect();
let mut state = vec![CascadeState::Pending; n];
let mut attempts = vec![0u32; n];
let mut last_err: Vec<Option<String>> = vec![None; n];
let mut out = StageOutcome::default();
loop {
let mut progressed = false;
for i in 0..n {
if state[i] != CascadeState::Pending {
continue;
}
let dep_dead = blocking[i]
.iter()
.any(|&d| matches!(state[d], CascadeState::Failed | CascadeState::Blocked));
if dep_dead {
state[i] = CascadeState::Blocked;
progressed = true;
continue;
}
if !blocking[i].iter().all(|&d| state[d] == CascadeState::Done) {
continue;
}
if attempts[i] >= MAX_ATTEMPTS {
state[i] = CascadeState::Failed;
progressed = true;
continue;
}
attempts[i] += 1;
progressed = true;
match publish(&order[i].repo) {
PublishAttempt::Published => {
out.published.push(order[i].repo.clone());
state[i] = CascadeState::Done;
}
PublishAttempt::AlreadyStaged => {
out.already_staged.push(order[i].repo.clone());
state[i] = CascadeState::Done;
}
PublishAttempt::Failed(e) => {
last_err[i] = Some(e);
if attempts[i] >= MAX_ATTEMPTS {
state[i] = CascadeState::Failed; }
}
}
}
if !progressed {
break;
}
}
for i in 0..n {
match state[i] {
CascadeState::Failed => {
let why =
last_err[i].clone().unwrap_or_else(|| "cargo publish failed".to_string());
out.errors.push(format!("{}: {}", order[i].repo, why));
}
CascadeState::Blocked | CascadeState::Pending => {
let culprit = blocking[i]
.iter()
.find(|&&d| state[d] != CascadeState::Done)
.map(|&d| order[d].repo.clone())
.unwrap_or_else(|| "upstream".to_string());
out.blocked
.push(format!("{}: blocked by failed dependency {}", order[i].repo, culprit));
}
CascadeState::Done => {}
}
}
out
}
pub fn execute(
plan: &StagePlan,
repo_paths: &BTreeMap<String, PathBuf>,
private: &std::collections::BTreeSet<String>,
_holger_bin: &Path,
data_dir: &Path,
grpc: &str,
http: &str,
) -> Result<StageOutcome> {
if !plan.is_executable() {
anyhow::bail!("stage plan not executable (dependency cycle: {})", plan.cycle.join(", "));
}
std::fs::create_dir_all(data_dir)?;
let cfg_path = data_dir.join("cargo-config.toml");
std::fs::write(&cfg_path, cargo_config_toml(http))?;
#[cfg(feature = "embed-holger")]
let _holger = {
let h = crate::holger_embed::EmbeddedHolger::start(&data_dir.join("registry"), grpc, http)
.context("start embedded holger")?;
h.wait_ready(std::time::Duration::from_secs(15))
.context("embedded holger did not become ready")?;
h
};
#[cfg(not(feature = "embed-holger"))]
let _holger = {
let p = spawn_holger(_holger_bin, &data_dir.join("registry"), grpc, http)?;
wait_ready(http)?;
p
};
eprintln!("stage: holger /cache + /sparring ready on http://{http}");
let mut guard = BranchGuard::new(&plan.branch);
let mut out = run_cascade(&plan.order, |repo| {
if private.contains(repo) {
eprintln!("stage: skip {repo} (private — not published to crates.io)");
return PublishAttempt::AlreadyStaged;
}
let Some(path) = repo_paths.get(repo) else {
return PublishAttempt::AlreadyStaged;
};
guard.capture(path);
let _ = Command::new("git")
.arg("-C")
.arg(path)
.args(["checkout", "-B", &plan.branch])
.status(); let res = Command::new(crate::release::cargo::cargo_bin())
.current_dir(path)
.args(["publish", "--workspace", "--no-verify", "--allow-dirty", "--registry", "sparring"])
.arg("--config")
.arg(&cfg_path)
.env("CARGO_REGISTRIES_SPARRING_TOKEN", "stage-rehearsal")
.output();
let res = match res {
Ok(r) => r,
Err(e) => return PublishAttempt::Failed(format!("cargo publish spawn failed: {e}")),
};
if res.status.success() {
eprintln!("stage: published {repo} → /sparring");
PublishAttempt::Published
} else {
let err = String::from_utf8_lossy(&res.stderr);
if err.contains("already exists") || err.contains("already uploaded") {
eprintln!("stage: {repo} already staged in /sparring (skip; bump to re-stage)");
PublishAttempt::AlreadyStaged
} else {
eprint!("{err}");
PublishAttempt::Failed(format!("cargo publish failed ({})", res.status))
}
}
});
let done: std::collections::BTreeSet<String> =
out.published.iter().chain(out.already_staged.iter()).cloned().collect();
let mut seen = std::collections::BTreeSet::new();
for step in &plan.order {
if !done.contains(step.repo.as_str()) {
continue;
}
for dep in &step.verify {
if !seen.insert(dep.clone()) {
continue;
}
let Some(path) = repo_paths.get(dep) else { continue };
let status = Command::new(crate::release::cargo::cargo_bin())
.current_dir(path)
.args(["build", "--quiet"])
.arg("--config")
.arg(&cfg_path)
.status()
.with_context(|| format!("cargo build {dep}"))?;
if status.success() {
out.verified.push(dep.clone());
eprintln!("stage: verified {} builds from /sparring", dep);
} else {
out.errors.push(format!("{dep}: verify build failed ({status})"));
}
}
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeSet;
fn graph(repo: &str, produces: &[&str], deps: &[&str]) -> RepoGraph {
let dep_set = deps.iter().map(|s| s.to_string()).collect::<BTreeSet<_>>();
RepoGraph {
repo: repo.to_string(),
produces: produces.iter().map(|s| s.to_string()).collect::<BTreeSet<_>>(),
deps: dep_set.clone(),
crate_deps: produces
.iter()
.map(|p| (p.to_string(), dep_set.clone()))
.collect::<std::collections::BTreeMap<_, _>>(),
..Default::default()
}
}
#[test]
fn plan_orders_topologically_with_verify_targets() {
let graphs = [
graph("znippy", &["znippy-common"], &["serde"]),
graph("skade", &["skade-katalog"], &["arrow"]),
graph("nornir", &["nornir"], &["znippy-common", "skade-katalog"]),
];
let plan = plan_stage(&graphs, "sparse+http://h/sparring/index/", "release/arrow-58");
assert!(plan.is_executable(), "clean DAG must be executable");
assert_eq!(plan.branch, "release/arrow-58");
let order: Vec<&str> = plan.order.iter().map(|s| s.repo.as_str()).collect();
let pos = |r: &str| order.iter().position(|x| *x == r).unwrap();
assert!(pos("znippy") < pos("nornir") && pos("skade") < pos("nornir"));
let znippy = plan.order.iter().find(|s| s.repo == "znippy").unwrap();
assert_eq!(znippy.verify, vec!["nornir"], "publishing znippy must verify nornir");
let nornir = plan.order.iter().find(|s| s.repo == "nornir").unwrap();
assert!(nornir.verify.is_empty(), "nothing depends on nornir here");
}
#[test]
fn cargo_config_targets_sparring_and_replaces_crates_io() {
let c = cargo_config_toml("127.0.0.1:18464");
assert!(c.contains("[registries.sparring]"), "defines the publish registry");
assert!(c.contains("sparse+http://127.0.0.1:18464/sparring/index/"));
assert!(c.contains(r#"replace-with = "sparring-mirror""#), "resolves deps via /sparring");
}
#[test]
fn plan_with_cycle_is_not_executable() {
let graphs = [
graph("a", &["a-crate"], &["b-crate"]),
graph("b", &["b-crate"], &["a-crate"]),
];
let plan = plan_stage(&graphs, "reg", "rel");
assert!(!plan.is_executable(), "a cycle must block execution");
assert_eq!(plan.cycle.len(), 2);
}
#[test]
fn plan_records_in_constellation_deps() {
let graphs = [
graph("znippy", &["znippy-common"], &["serde"]),
graph("skade", &["skade-katalog"], &["arrow"]),
graph("nornir", &["nornir"], &["znippy-common", "skade-katalog"]),
];
let plan = plan_stage(&graphs, "reg", "rel");
let nornir = plan.order.iter().find(|s| s.repo == "nornir").unwrap();
assert_eq!(nornir.deps, vec!["skade", "znippy"], "nornir gates on both bases");
let znippy = plan.order.iter().find(|s| s.repo == "znippy").unwrap();
assert!(znippy.deps.is_empty(), "a base repo gates on nothing");
}
fn step(repo: &str, deps: &[&str], verify: &[&str]) -> StageStep {
StageStep {
repo: repo.to_string(),
deps: deps.iter().map(|s| s.to_string()).collect(),
verify: verify.iter().map(|s| s.to_string()).collect(),
}
}
#[test]
fn cascade_defers_dependent_until_dep_published() {
let order = [step("a", &[], &["b"]), step("b", &["a"], &[])];
let mut seen: Vec<String> = Vec::new();
let out = run_cascade(&order, |r| {
seen.push(r.to_string());
PublishAttempt::Published
});
assert_eq!(seen, ["a", "b"], "A must be attempted before B");
assert_eq!(out.published, ["a", "b"]);
assert!(out.errors.is_empty() && out.blocked.is_empty());
}
#[test]
fn cascade_blocks_dependent_when_dep_fails() {
let order = [step("a", &[], &["b"]), step("b", &["a"], &[])];
let mut attempted: Vec<String> = Vec::new();
let out = run_cascade(&order, |r| {
attempted.push(r.to_string());
if r == "a" {
PublishAttempt::Failed("boom".to_string())
} else {
PublishAttempt::Published
}
});
assert!(!attempted.iter().any(|r| r == "b"), "B must not be attempted once A fails");
assert!(out.published.is_empty());
assert!(out.errors.iter().any(|e| e.starts_with("a:")), "A is its own failure");
assert!(
out.blocked.iter().any(|e| e.starts_with("b:") && e.contains("a")),
"B is reported blocked-by-failed-dependency a, not as a failure: {:?}",
out.blocked
);
}
#[test]
fn cascade_retries_transient_failure_then_succeeds() {
let order = [step("a", &[], &[])];
let mut calls = 0u32;
let out = run_cascade(&order, |_| {
calls += 1;
if calls == 1 {
PublishAttempt::Failed("transient".to_string())
} else {
PublishAttempt::Published
}
});
assert!(calls >= 2, "a transient failure must get another attempt (got {calls})");
assert_eq!(out.published, ["a"]);
assert!(out.errors.is_empty() && out.blocked.is_empty());
}
#[test]
fn cascade_bounds_retries_of_permanent_failure() {
let order = [step("a", &[], &[])];
let mut calls = 0u32;
let out = run_cascade(&order, |_| {
calls += 1;
PublishAttempt::Failed("always".to_string())
});
assert_eq!(calls, MAX_ATTEMPTS, "retries are bounded to MAX_ATTEMPTS");
assert!(out.errors.iter().any(|e| e.starts_with("a:")));
assert!(out.published.is_empty() && out.blocked.is_empty());
}
#[test]
fn cascade_already_staged_satisfies_dependents() {
let order = [step("a", &[], &["b"]), step("b", &["a"], &[])];
let out = run_cascade(&order, |r| {
if r == "a" {
PublishAttempt::AlreadyStaged
} else {
PublishAttempt::Published
}
});
assert_eq!(out.already_staged, ["a"]);
assert_eq!(out.published, ["b"], "an already-staged dep unblocks its dependent");
assert!(out.blocked.is_empty() && out.errors.is_empty());
}
#[test]
fn cascade_ignores_back_edges_within_ordered_scc() {
let order = [step("a", &["b"], &["b"]), step("b", &["a"], &[])];
let out = run_cascade(&order, |_| PublishAttempt::Published);
assert_eq!(out.published, ["a", "b"], "the back-edge must not deadlock the SCC");
assert!(out.blocked.is_empty() && out.errors.is_empty());
}
}