diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 652f6d45..e8fcf260 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -553,33 +553,13 @@ async fn run_meta_lock( // it up. Without this, a meta-input bump rebuilds every affected agent // against whatever `applied/` already happened to be locked to — // normally current, but silently stale forever if a past deploy failed - // and nothing since retried it. Relocks against each agent's *declared* - // input (the forge URL, not the local `applied/` mirror a rebuild - // relocks against) — this path has nothing to gate. One combined call, not per-agent: - // simpler, at the cost of one unreachable/broken agent repo failing the - // whole cascade relock rather than just that agent. - if !cascade.is_empty() { - let agent_inputs: Vec = - cascade.iter().map(|name| format!("agent-{name}")).collect(); - crate::meta::lock_update(&agent_inputs).await?; - // The relock above is a one-shot effect on *this* rebuild only — - // `applied/` itself hasn't moved, so the next `relock = true` - // rebuild trigger (the boot sweep, most notably) re-locks against - // `applied/` and reverts straight back to whatever it was - // stuck on. Fast-forward each cascade agent's own applied mirror - // too so a later rebuild through a different trigger doesn't undo - // this one. Best-effort per agent (unlike the relock above): one - // agent's forge repo being briefly unreachable here shouldn't be - // fatal to the cascade that already relocked its input above. - for name in &cascade { - if !crate::forge::fast_forward_applied_main(name).await { - tracing::warn!( - %name, - "meta-update cascade: applied/main fast-forward skipped or failed" - ); - } - } - } + // and nothing since retried it. + catch_up_cascade_agents( + &cascade, + crate::forge::fast_forward_applied_main, + |agent_inputs| async move { crate::meta::lock_update(&agent_inputs).await }, + ) + .await?; // Grow one rebuild subgraph per affected agent into *this* meta-update // DAG (rooted on this `MetaLock`, so they build against the post-bump // lock), rather than fanning out child DAGs. `relock = false` — the @@ -589,6 +569,41 @@ async fn run_meta_lock( Ok(cascade) } +/// Fast-forward each cascade agent's `applied//main` to its forge +/// config main, then relock every `agent-` input in one call. +/// +/// The declared `agent-` input is the `applied/` mirror itself, +/// and the cascade's rebuilds (`relock = false`) build whatever this relock +/// pins, so the mirror has to move first. The fast-forward is best-effort +/// per agent: a failed one leaves that agent relocked against whatever its +/// mirror already has. The relock is one combined call, so one broken input +/// fails it for the whole cascade. +/// +/// The fast-forward and the relock are passed in so the order is testable +/// without nix or a forge. +async fn catch_up_cascade_agents<'a, FF, Relock>( + cascade: &'a [String], + mut fast_forward: impl FnMut(&'a str) -> FF, + relock: impl FnOnce(Vec) -> Relock, +) -> Result<()> +where + FF: std::future::Future, + Relock: std::future::Future>, +{ + if cascade.is_empty() { + return Ok(()); + } + for name in cascade { + if !fast_forward(name).await { + tracing::warn!( + %name, + "meta-update cascade: applied/main fast-forward skipped or failed" + ); + } + } + relock(cascade.iter().map(|name| format!("agent-{name}")).collect()).await +} + /// Idempotent power-converge *planner*: compare `wanted` (durable /// intent) against observed state and, when they diverge, fan the /// mechanical `Start` / `Stop` out as a first-class node appended to @@ -893,7 +908,88 @@ fn validate_agent_names(names: impl IntoIterator) -> Vec #[cfg(test)] mod cascade_agent_tests { - use super::{parse_agent_input_names, validate_agent_names}; + use super::{catch_up_cascade_agents, parse_agent_input_names, validate_agent_names}; + use std::cell::RefCell; + use std::collections::BTreeMap; + + /// Runs [`catch_up_cascade_agents`] over `cascade` against a model of + /// the applied mirrors: every mirror starts at `"old"`, a fast-forward + /// moves it to `"new"` unless the agent is in `failing`. Returns the + /// mirror head the relock saw for each input it was given. + async fn relocked_heads(cascade: &[&str], failing: &[&str]) -> Vec<(String, &'static str)> { + let cascade: Vec = cascade.iter().map(|&n| n.to_owned()).collect(); + let heads = RefCell::new( + cascade + .iter() + .map(|n| (n.clone(), "old")) + .collect::>(), + ); + let relocked = RefCell::new(Vec::new()); + let fast_forward = |name: &str| { + let ok = !failing.contains(&name); + if ok { + heads.borrow_mut().insert(name.to_owned(), "new"); + } + async move { ok } + }; + let relock = |inputs: Vec| { + for input in inputs { + let name = input.strip_prefix("agent-").expect("agent input"); + let head = heads.borrow()[name]; + relocked.borrow_mut().push((input, head)); + } + async { anyhow::Ok(()) } + }; + catch_up_cascade_agents(&cascade, fast_forward, relock) + .await + .expect("catch up"); + relocked.into_inner() + } + + #[tokio::test] + async fn the_cascade_relock_pins_the_fast_forwarded_head() { + assert_eq!( + relocked_heads(&["iris", "atlas"], &[]).await, + vec![ + ("agent-iris".to_owned(), "new"), + ("agent-atlas".to_owned(), "new"), + ] + ); + } + + #[tokio::test] + async fn a_failed_fast_forward_still_relocks_that_agent() { + assert_eq!( + relocked_heads(&["iris", "atlas"], &["iris"]).await, + vec![ + ("agent-iris".to_owned(), "old"), + ("agent-atlas".to_owned(), "new"), + ] + ); + } + + /// A relock with no inputs is a bare `nix flake update`, which bumps + /// every meta input. + #[tokio::test] + async fn an_empty_cascade_moves_nothing_and_relocks_nothing() { + let fast_forwarded = std::cell::Cell::new(false); + let relocked = std::cell::Cell::new(false); + let outcome = catch_up_cascade_agents( + &[], + |_| { + fast_forwarded.set(true); + async { true } + }, + |_| { + relocked.set(true); + async { anyhow::Ok(()) } + }, + ) + .await; + outcome.expect("an empty cascade is not an error"); + assert!(!fast_forwarded.get(), "nothing to fast-forward"); + assert!(!relocked.get(), "an empty relock would bump every input"); + } #[test] fn well_formed_agent_inputs_pass_through() {