vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

deploy only after a successful meta update

Browse source 
A hive builds only from internal/meta, so a TriggerDeploy after a failed UpdateMeta could only fail its build. Both DAGs that queue an update (merged config PR, agent create) now gate the deploy with after_ok: the update shows Failed and the deploy Skipped.
This commit is contained in:
atlas 2026-10-25 13:55:10 +01:00
commit 61e35ab603
3 changed files with 90 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

137
swarm-controller/src/jobq.rs
View file

@ -1139,10 +1139,10 @@ pub(crate) async fn queue_merged_config_deploy(
/// The sub-DAG a merged config PR is: the meta update, then the deploy of
/// `rev` on `hive`.
///
/// `after_any`: the deploy is asked for even when the update failed (no
/// builder, an unreadable config repo). A hive builds only from
/// `internal/meta`, so it fails that build, naming the agent, while the repo
/// does not pin the agent's config commit.
/// `after_ok`: a hive builds only from `internal/meta`, so a deploy asked for
/// after a failed update (no builder, an unreadable config repo) could only
/// fail its build. The failed update is `Failed` in the graph and the deploy
/// `Skipped`.
fn merged_deploy_job(
b: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
hive: &str,
@ -1159,7 +1159,7 @@ fn merged_deploy_job(
agent: agent.to_owned(),
rev: Some(rev.to_owned()),
})
.after_any(update_meta);
.after_ok(update_meta);
vec![root]
}
@ -1251,10 +1251,9 @@ pub(crate) fn declare_agent_job(
hive: hive.to_owned(),
agent: agent.to_owned(),
});
// The hive builds the agent from `internal/meta` and fails that build,
// naming the agent, until the repo pins the agent's config commit.
// `after_any` on the update, so one that failed (no builder, an unreadable
// config repo) still asks for the deploy.
// The hive builds the agent from `internal/meta` only, so the deploy needs
// the update to succeed (`after_ok`): after a failed one (no builder, an
// unreadable config repo) the hive's build could only fail.
let update_meta = b
.node(SwarmNodeKind::UpdateMeta)
.needs(SwarmResourceKind::Meta)
@ -1289,7 +1288,7 @@ pub(crate) fn declare_agent_job(
agent: agent.to_owned(),
rev: None,
})
.after_any(update_meta)
.after_ok(update_meta)
.after_any(mint_identity)
.after_any(mint_forge_token)
// `after_any` for the same reason: the container reads this token
@ -2142,39 +2141,50 @@ mod tests {
.accepts(hive_jobq::TerminalState::Failed)
}
/// A deploy that waits for the meta update still runs when the update
/// fails.
type TestBuilder = hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>;
fn create_dag(b: &TestBuilder) {
let _ = super::declare_agent_job(b, "atlas", "pr1ma", false);
}
fn merged_dag(b: &TestBuilder) {
let _ = super::merged_deploy_job(b, "pr1ma", "atlas", "abc");
}
/// A deploy needs the meta update to succeed: its edge refuses a failed
/// update, in both DAGs that queue one.
#[test]
fn a_failed_meta_update_does_not_hold_back_the_deploys() {
let create = |b: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>| {
let _ = super::declare_agent_job(b, "atlas", "pr1ma", false);
};
let merge = |b: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>| {
let _ = super::merged_deploy_job(b, "pr1ma", "atlas", "abc");
};
assert!(edge_accepts_failure(
create,
fn a_deploy_needs_the_meta_update_to_succeed() {
assert!(!edge_accepts_failure(
create_dag,
"trigger_deploy",
"update_meta"
));
assert!(edge_accepts_failure(merge, "trigger_deploy", "update_meta"));
// The same lookup finds the edges that do hold a deploy back.
assert!(!edge_accepts_failure(
create,
merged_dag,
"trigger_deploy",
"set_agent_wanted"
"update_meta"
));
// Controls: the same lookup finds the edges that do accept a failure.
assert!(edge_accepts_failure(
create_dag,
"trigger_deploy",
"mint_agent_identity"
));
assert!(!edge_accepts_failure(
create,
"update_meta",
create_dag,
"trigger_deploy",
"set_agent_wanted"
));
}
/// The scheduler's side of the same edge: with the update failed, the
/// deploy is the next node to run.
#[tokio::test]
async fn the_merged_deploy_runs_after_a_failed_meta_update() {
/// Every node of the DAG `declare` builds, run to the end with the meta
/// update failing or succeeding and every other node succeeding: the
/// final state of each, by label.
async fn run_dag(
declare: impl FnOnce(&TestBuilder),
update_meta_fails: bool,
) -> Vec<(String, hive_jobq::State)> {
use hive_jobq::scheduler::{Outcome, Scheduler};
use hive_jobq_wire::WireNode as _;
@ -2184,39 +2194,62 @@ mod tests {
);
sched
.insert_job(None, |b| {
super::merged_deploy_job(b, "pr1ma", "atlas", "abc")
declare(b);
Vec::new()
})
.expect("the graph inserts");
let sched = std::sync::Arc::new(std::sync::Mutex::new(sched));
let mut ran = Vec::new();
while let Some(runner) = Scheduler::claim_next(&sched, |_id, kind, builder| async move {
let outcome = match kind {
SwarmNodeKind::UpdateMeta => Outcome::Failed("no builder".to_owned()),
SwarmNodeKind::UpdateMeta if update_meta_fails => {
Outcome::Failed("no builder".to_owned())
}
_ => Outcome::Done,
};
(builder, outcome)
}) {
runner.await.1.expect("no growth declared");
ran.push(
sched
.lock()
.unwrap()
.graph()
.nodes()
.filter(|n| n.state.is_terminal())
.map(|n| (n.payload.label(), n.state))
.collect::<Vec<_>>(),
}
let guard = sched.lock().unwrap();
guard
.graph()
.nodes()
.map(|n| (n.payload.label(), n.state))
.collect()
}
fn state_of(states: &[(String, hive_jobq::State)], label: &str) -> hive_jobq::State {
states
.iter()
.find(|(l, _)| l == label)
.unwrap_or_else(|| panic!("the graph holds a {label} node"))
.1
}
/// The scheduler's side of the same edge, for both DAGs that queue an
/// update: with it failed, the deploy never runs. The graph shows the
/// update `Failed` and the deploy `Skipped`; the first run in each pair is
/// the control, where the deploy does run.
#[tokio::test]
async fn a_failed_meta_update_means_no_deploy_runs() {
use hive_jobq::State;
for (name, declare) in [
("merged", merged_dag as fn(&TestBuilder)),
("create", create_dag as fn(&TestBuilder)),
] {
let ok = run_dag(declare, false).await;
assert_eq!(state_of(&ok, "update_meta"), State::Done, "{name}");
assert_eq!(state_of(&ok, "trigger_deploy"), State::Done, "{name}");
let failed = run_dag(declare, true).await;
assert_eq!(state_of(&failed, "update_meta"), State::Failed, "{name}");
assert_eq!(
state_of(&failed, "trigger_deploy"),
State::Skipped,
"{name}: the deploy must not run after a failed update"
);
}
let last = ran.last().expect("both nodes ran");
assert!(
last.contains(&("update_meta".to_owned(), hive_jobq::State::Failed)),
"{last:?}"
);
assert!(
last.contains(&("trigger_deploy".to_owned(), hive_jobq::State::Done)),
"{last:?}"
);
}
/// A fan-out record held in memory; `unreadable` stands for a bucket that