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

retry every failed meta build each round, a timeout included, and drop the attempt count

Browse source
This commit is contained in:
atlas 2026-10-26 18:39:09 +01:00
commit 59456c3c58
4 changed files with 251 additions and 435 deletions
Download patch file Download diff file Expand all files Collapse all files

49
swarm-controller/src/held_deploys.rs
View file

@ -4,13 +4,12 @@
//!
//! An agent counts when its latest `BuildMeta` node is `Failed`, once per agent
//! under one of two `reason` labels:
//! - `red`: `nix` exited with an error twice. Nothing retries it; a config merge
//! builds again.
//! - `retrying`: the build gave no verdict or met its first `nix` error, so the
//! ten-minute timer builds it again.
//! - `failed`: `nix` failed, its timeout included.
//! - `unavailable`: the build got no verdict (no builder, a forge error, no
//! `internal/meta` head, a missing read grant).
//!
//! A pending, running or finished build does not count, so the value falls to
//! zero when the builder answers.
//! The ten-minute timer builds either again. A pending, running or finished
//! build does not count, so the value falls to zero when a build succeeds.
use std::sync::{Arc, Mutex};
@ -25,7 +24,7 @@ use crate::meta_build;
/// The gauge's name. The Grafana rule selects it by this name.
pub(crate) const METRIC: &str = "swarm.deploys.held";
/// Held agents as `(red, retrying)`.
/// Held agents as `(failed, unavailable)`.
fn count(graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>) -> (u64, u64) {
let mut latest = std::collections::BTreeMap::new();
for n in graph.nodes() {
@ -40,11 +39,11 @@ fn count(graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>) -> (u64, u6
latest
.into_values()
.filter(|n| n.state == State::Failed)
.fold((0, 0), |(red, retrying), n| {
if n.error.as_deref().is_some_and(meta_build::is_retried) {
(red, retrying + 1)
.fold((0, 0), |(failed, unavailable), n| {
if n.error.as_deref().is_some_and(meta_build::is_no_verdict) {
(failed, unavailable + 1)
} else {
(red + 1, retrying)
(failed + 1, unavailable)
}
})
}
@ -64,13 +63,13 @@ pub(crate) fn register(
.u64_observable_gauge(METRIC)
.with_description("agents whose deploy the build gate is holding back, by reason")
.with_callback(move |obs| {
let (red, retrying) = count(
let (failed, unavailable) = count(
jobq.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph(),
);
obs.observe(red, &[KeyValue::new("reason", "red")]);
obs.observe(retrying, &[KeyValue::new("reason", "retrying")]);
obs.observe(failed, &[KeyValue::new("reason", "failed")]);
obs.observe(unavailable, &[KeyValue::new("reason", "unavailable")]);
})
.build();
}
@ -97,7 +96,6 @@ mod tests {
.append(
SwarmNodeKind::BuildMeta {
agent: (*agent).to_owned(),
attempt: 0,
},
Vec::new(),
None,
@ -113,19 +111,19 @@ mod tests {
jobq
}
fn red() -> Outcome {
fn failed() -> Outcome {
Outcome::Failed("error: attribute missing".to_owned())
}
fn retrying() -> Outcome {
Outcome::Failed(format!("{}no builder", meta_build::RETRY_PREFIX))
fn unavailable() -> Outcome {
Outcome::Failed(format!("{}no builder", meta_build::NO_VERDICT_PREFIX))
}
#[tokio::test]
async fn a_failed_latest_build_holds_its_agent_by_reason() {
let jobq = after(&[
("atlas", red()),
("argus", retrying()),
("atlas", failed()),
("argus", unavailable()),
("iris", Outcome::Done),
])
.await;
@ -136,9 +134,9 @@ mod tests {
/// releases it, and a red one after a green one holds it.
#[tokio::test]
async fn only_the_latest_build_of_an_agent_counts() {
let released = after(&[("atlas", red()), ("atlas", Outcome::Done)]).await;
let released = after(&[("atlas", failed()), ("atlas", Outcome::Done)]).await;
assert_eq!(count(released.lock().unwrap().graph()), (0, 0));
let held = after(&[("atlas", Outcome::Done), ("atlas", red())]).await;
let held = after(&[("atlas", Outcome::Done), ("atlas", failed())]).await;
assert_eq!(count(held.lock().unwrap().graph()), (1, 0));
}
@ -146,7 +144,7 @@ mod tests {
/// included, so a rule sees the series before anything is held.
#[tokio::test]
async fn the_gauge_exports_both_reasons() {
let jobq = after(&[("atlas", red())]).await;
let jobq = after(&[("atlas", failed())]).await;
let exporter = InMemoryMetricExporter::default();
let reader = PeriodicReader::builder(exporter.clone()).build();
let provider = SdkMeterProvider::builder().with_reader(reader).build();
@ -174,6 +172,9 @@ mod tests {
}
}
points.sort();
assert_eq!(points, [("red".to_owned(), 1), ("retrying".to_owned(), 0)]);
assert_eq!(
points,
[("failed".to_owned(), 1), ("unavailable".to_owned(), 0)]
);
}
}

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

@ -115,10 +115,8 @@ pub(crate) enum SwarmNodeKind {
/// runs only when the build is green. See `meta_build`.
///
/// Carries no rev: it reads the head when it runs, which in most DAGs is
/// after the `UpdateMeta` that commits it. `attempt` is how many builds of
/// this deploy already ended in an error from `nix`; the build after the
/// second such error is red.
BuildMeta { agent: String, attempt: u32 },
/// after the `UpdateMeta` that commits it.
BuildMeta { agent: String },
/// Re-mint `agent`'s queue secret if it is still stored and old enough.
/// See `agent_renewal`.
///
@ -204,6 +202,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
| SwarmNodeKind::MintAgentIdentity { agent }
| SwarmNodeKind::MintAgentForgeToken { agent }
| SwarmNodeKind::MintAgentMatrixAccount { agent }
| SwarmNodeKind::BuildMeta { agent }
| SwarmNodeKind::RenewAgentQueueCredential { agent } => {
serde_json::json!({ "agent": agent })
}
@ -216,9 +215,6 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
SwarmNodeKind::BuildAgent { agent, rev, pr } => {
serde_json::json!({ "agent": agent, "rev": rev, "pr": pr })
}
SwarmNodeKind::BuildMeta { agent, attempt } => {
serde_json::json!({ "agent": agent, "attempt": attempt })
}
SwarmNodeKind::TriggerDeploy { hive, agent, rev } => {
serde_json::json!({ "agent": agent, "hive": hive, "rev": rev })
}
@ -418,9 +414,7 @@ async fn run_node_work(
SwarmNodeKind::BuildAgent { agent, rev, pr } => {
build_agent(&deps, forge, agent, rev, pr).await
}
SwarmNodeKind::BuildMeta { agent, attempt } => {
build_meta(&deps, forge, agent, attempt).await
}
SwarmNodeKind::BuildMeta { agent } => build_meta(&deps, forge, agent).await,
SwarmNodeKind::RenewAgentQueueCredential { agent } => renew_queue_credential(&agent).await,
SwarmNodeKind::SetAgentWanted { hive, agent } => {
set_agent_wanted(deps.wanted.as_deref(), &hive, &agent).await
@ -608,9 +602,8 @@ async fn build_meta(
deps: &WorkerDeps,
forge: Option<Arc<forge::Client>>,
agent: String,
attempt: u32,
) -> hive_jobq::scheduler::Outcome {
meta_build::run(forge.as_deref(), deps.queue.as_ref(), agent, attempt).await
meta_build::run(forge.as_deref(), deps.queue.as_ref(), agent).await
}
/// What an `UpdateMeta` pass decided: the deploys to send, and the placements
@ -748,21 +741,16 @@ async fn record_fanout(
/// parent. The build waits on it, since the build reads the head that update
/// commits. The deploy keeps its own edge to it too: `skipped_deploys` finds
/// the deploys a failed update skipped by that edge.
///
/// `attempt` is the build's `BuildMeta::attempt`: 0 for a deploy no build has
/// failed for yet.
fn gated_deploy<'a>(
b: &'a hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
hive: &str,
agent: &str,
rev: Option<String>,
update: Option<hive_jobq::builder::NodeGuid>,
attempt: u32,
) -> hive_jobq::builder::NodeRef<'a, SwarmNodeKind, SwarmResourceKind> {
let mut build = b
.node(SwarmNodeKind::BuildMeta {
agent: agent.to_owned(),
attempt,
})
.needs(SwarmResourceKind::Builder);
if let Some(update) = update {
@ -784,9 +772,9 @@ fn gated_deploy<'a>(
/// Declare one build and rebuild of each `(hive, agent)` placement, which
/// nothing else tells to build again, and for a [`FanOut::Record`] the
/// `RecordFanout` that waits on every one of them. A deploy the build gate
/// skipped counts as sent: the pin is recorded, so one red agent does not
/// resend every healthy one on each retry. A deploy skipped because the build
/// gave no verdict is re-declared by `queue_meta_update`.
/// skipped counts as sent: the pin is recorded, so one failing agent does
/// not resend every healthy one on each retry. A deploy skipped by a failed
/// build is re-declared by `queue_meta_update` on every round.
fn declare_fan_out(
grown: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
fan_out: &FanOut,
@ -797,7 +785,7 @@ fn declare_fan_out(
}
let deploys: Vec<hive_jobq::builder::NodeGuid> = placements
.iter()
.map(|(hive, agent)| gated_deploy(grown, hive, agent, None, None, 0).guid())
.map(|(hive, agent)| gated_deploy(grown, hive, agent, None, None).guid())
.collect();
if let FanOut::Record { pin } = fan_out {
let mut record = grown.node(SwarmNodeKind::RecordFanout { pin: pin.clone() });
@ -1235,7 +1223,7 @@ fn merged_deploy_job(
.node(SwarmNodeKind::UpdateMeta)
.needs(SwarmResourceKind::Meta);
let root = update_meta.guid();
let _deploy = gated_deploy(b, hive, agent, Some(rev.to_owned()), Some(root), 0);
let _deploy = gated_deploy(b, hive, agent, Some(rev.to_owned()), Some(root));
vec![root]
}
@ -1359,7 +1347,7 @@ pub(crate) fn declare_agent_job(
// queue at all fails this node — but `TriggerDeploy` has nothing to
// publish to on such a host either, so the strict edge cancels a node
// that could not have succeeded anyway.
let trigger_deploy = gated_deploy(b, hive, agent, None, Some(update_meta.guid()), 0)
let trigger_deploy = gated_deploy(b, hive, agent, None, Some(update_meta.guid()))
.after_any(mint_identity)
.after_any(mint_forge_token)
// `after_any` for the same reason: the container reads this token
@ -1564,9 +1552,8 @@ pub(crate) fn spawn_meta_update_timer(
enum SkipCause {
/// It waited on an `UpdateMeta` that `Failed`.
FailedUpdate,
/// It waited on a `BuildMeta` that gave no verdict or whose first `nix`
/// error asks for another build ([`meta_build::is_retried`]).
UnfinishedBuild,
/// It waited on a `BuildMeta` that `Failed`, whatever the reason.
FailedBuild,
}
/// A deploy [`skipped_deploys`] found.
@ -1575,18 +1562,15 @@ struct SkippedDeploy {
agent: String,
rev: Option<String>,
cause: SkipCause,
/// The `BuildMeta::attempt` of the build that replaces the failed one.
attempt: u32,
}
/// The deploys the timer declares again: for each agent, its latest
/// `TriggerDeploy` when the scheduler skipped it because an `UpdateMeta` it
/// waited on `Failed`, or because the `BuildMeta` it waited on gave no verdict
/// or met its first `nix` error. A deploy skipped for another reason is left
/// alone. A red build, the second `nix` error, is a verdict on the config, which
/// a config merge answers with a build of its own. The pause declaration or the
/// config seeding failing means the update that would unblock the deploy was
/// skipped too, and deploying past those is the race those edges close.
/// `TriggerDeploy` when the scheduler skipped it because an `UpdateMeta` or a
/// `BuildMeta` it waited on `Failed`. A failed build is built again on every
/// round until one succeeds. A deploy skipped for another reason is left alone:
/// the pause declaration or the config seeding failing means the update that
/// would unblock the deploy was skipped too, and deploying past those is the
/// race those edges close.
fn skipped_deploys(
graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>,
) -> Vec<SkippedDeploy> {
@ -1608,28 +1592,18 @@ fn skipped_deploys(
let SwarmNodeKind::TriggerDeploy { hive, agent, rev } = &n.payload else {
return None;
};
let deps = || {
n.deps.iter().filter_map(|d| match d {
hive_jobq::Dep::Node { id, .. } => graph.node(*id),
hive_jobq::Dep::Resource { .. } => None,
let failed = |is: fn(&SwarmNodeKind) -> bool| {
n.deps.iter().any(|d| match d {
hive_jobq::Dep::Node { id, .. } => graph.node(*id).is_some_and(|dep| {
is(&dep.payload) && dep.state == hive_jobq::State::Failed
}),
hive_jobq::Dep::Resource { .. } => false,
})
};
let failed_update = deps().any(|dep| {
matches!(dep.payload, SwarmNodeKind::UpdateMeta)
&& dep.state == hive_jobq::State::Failed
});
let unfinished_build = deps().find_map(|dep| match (&dep.payload, &dep.error) {
(SwarmNodeKind::BuildMeta { attempt, .. }, Some(error))
if dep.state == hive_jobq::State::Failed && meta_build::is_retried(error) =>
{
Some(meta_build::next_attempt(*attempt, error))
}
_ => None,
});
let (cause, attempt) = if failed_update {
(SkipCause::FailedUpdate, 0)
} else if let Some(attempt) = unfinished_build {
(SkipCause::UnfinishedBuild, attempt)
let cause = if failed(|p| matches!(p, SwarmNodeKind::UpdateMeta)) {
SkipCause::FailedUpdate
} else if failed(|p| matches!(p, SwarmNodeKind::BuildMeta { .. })) {
SkipCause::FailedBuild
} else {
return None;
};
@ -1638,7 +1612,6 @@ fn skipped_deploys(
agent: agent.clone(),
rev: rev.clone(),
cause,
attempt,
})
})
.collect()
@ -1650,11 +1623,10 @@ fn skipped_deploys(
/// run skipped ([`skipped_deploys`]), so an agent created or merged while the
/// update was failing is deployed by the run that succeeds.
///
/// When the latest `UpdateMeta` is `Done`, a deploy skipped because the builder
/// gave no verdict still gets a fresh build and deploy, with no update, so a
/// builder that was down or not yet updated is caught up once it answers. That
/// adds nodes on every tick while the builder stays away. Returns whether a job
/// was inserted.
/// When the latest `UpdateMeta` is `Done`, a deploy skipped by a failed build
/// still gets a fresh build and deploy, with no update, so a build that failed
/// is tried again on every round until it succeeds. That adds two nodes per
/// failing agent on every tick. Returns whether a job was inserted.
fn queue_meta_update(
sched: &Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
) -> Result<bool> {
@ -1680,7 +1652,7 @@ fn queue_meta_update(
.is_none_or(|n| n.state != hive_jobq::State::Done);
let skipped: Vec<SkippedDeploy> = skipped_deploys(sched.graph())
.into_iter()
.filter(|s| update_due || s.cause == SkipCause::UnfinishedBuild)
.filter(|s| update_due || s.cause == SkipCause::FailedBuild)
.collect();
if !update_due && skipped.is_empty() {
return Ok(false);
@ -1694,7 +1666,7 @@ fn queue_meta_update(
});
let deploys: Vec<_> = skipped
.into_iter()
.map(|s| gated_deploy(b, &s.hive, &s.agent, s.rev, update, s.attempt).guid())
.map(|s| gated_deploy(b, &s.hive, &s.agent, s.rev, update).guid())
.collect();
update.map_or(deploys, |update| vec![update])
})
@ -2869,81 +2841,66 @@ mod tests {
assert_eq!(fan_out, FanOut::Record { pin: "p1".into() });
}
/// A red build for one agent skips that agent's deploy only, and the pin
/// is still recorded: the red agent is not sent again with every healthy
/// one on the next retry. The timer declares nothing for it either, since a
/// red build is a verdict on its config.
/// A failed build for one agent skips that agent's deploy only, and the pin
/// is still recorded, so the healthy agents are not sent again. The timer
/// declares a build and deploy for the failed agent alone, whatever the
/// build failed with: `nix`'s error, a timeout, or no verdict.
#[tokio::test]
async fn a_red_build_skips_its_deploy_and_the_pin_is_still_recorded() {
use hive_jobq::State;
let pins = Arc::new(FakePins::holding("p0"));
let (_, sched) = pass_with(
&pins,
&report("p1", true),
&PLACED,
None,
false,
Some(("argus", "error: builder for argus failed".to_owned())),
)
.await;
let mut builds = states_of(&sched, "build_meta");
builds.sort_by_key(|s| format!("{s:?}"));
assert_eq!(builds, [State::Done, State::Done, State::Failed]);
let mut deploys = states_of(&sched, "trigger_deploy");
deploys.sort_by_key(|s| format!("{s:?}"));
assert_eq!(deploys, [State::Done, State::Done, State::Skipped]);
assert_eq!(states_of(&sched, "record_fanout"), [State::Done]);
assert_eq!(pins.held().as_deref(), Some("p1"));
// The failed build rolls the update up `Failed`, so the timer queues an
// update, and only an update: no deploy is declared again.
assert!(queue_meta_update(&sched).expect("tick"));
assert_eq!(states_of(&sched, "trigger_deploy").len(), 3);
assert_eq!(states_of(&sched, "build_meta").len(), 3);
assert_eq!(states_of(&sched, "update_meta").len(), 2);
}
/// A build that gave no verdict skips its deploy and still records the pin,
/// and the timer declares a build and deploy for that agent alone.
#[tokio::test]
async fn an_unfinished_build_is_declared_again_for_that_agent_alone() {
async fn a_failed_build_skips_its_deploy_and_the_pin_is_still_recorded() {
use hive_jobq::State;
use hive_jobq_wire::WireNode as _;
let pins = Arc::new(FakePins::holding("p0"));
let error = format!("{}no swarm builder is listening", meta_build::RETRY_PREFIX);
let (_, sched) = pass_with(
&pins,
&report("p1", true),
&PLACED,
None,
false,
Some(("argus", error)),
)
.await;
assert_eq!(states_of(&sched, "record_fanout"), [State::Done]);
assert_eq!(pins.held().as_deref(), Some("p1"));
for error in [
"error: builder for argus failed".to_owned(),
"timed out: nix did not finish".to_owned(),
format!(
"{}no swarm builder is listening",
meta_build::NO_VERDICT_PREFIX
),
] {
let pins = Arc::new(FakePins::holding("p0"));
let (_, sched) = pass_with(
&pins,
&report("p1", true),
&PLACED,
None,
false,
Some(("argus", error.clone())),
)
.await;
let mut builds = states_of(&sched, "build_meta");
builds.sort_by_key(|s| format!("{s:?}"));
assert_eq!(builds, [State::Done, State::Done, State::Failed], "{error}");
let mut deploys = states_of(&sched, "trigger_deploy");
deploys.sort_by_key(|s| format!("{s:?}"));
assert_eq!(
deploys,
[State::Done, State::Done, State::Skipped],
"{error}"
);
assert_eq!(states_of(&sched, "record_fanout"), [State::Done], "{error}");
assert_eq!(pins.held().as_deref(), Some("p1"), "{error}");
assert!(queue_meta_update(&sched).expect("tick"));
let guard = sched.lock().unwrap();
let again = |label: &str| {
let mut found: Vec<_> = guard
.graph()
.nodes()
.filter(|n| n.payload.label() == label)
.filter(|n| n.state == State::Pending)
.map(|n| match &n.payload {
SwarmNodeKind::BuildMeta { agent, .. }
| SwarmNodeKind::TriggerDeploy { agent, .. } => agent.clone(),
other => panic!("not a build or a deploy: {other:?}"),
})
.collect();
found.sort();
found
};
assert_eq!(again("build_meta"), ["argus"]);
assert_eq!(again("trigger_deploy"), ["argus"]);
assert!(queue_meta_update(&sched).expect("tick"), "{error}");
let guard = sched.lock().unwrap();
let again = |label: &str| {
let mut found: Vec<_> = guard
.graph()
.nodes()
.filter(|n| n.payload.label() == label)
.filter(|n| n.state == State::Pending)
.map(|n| match &n.payload {
SwarmNodeKind::BuildMeta { agent }
| SwarmNodeKind::TriggerDeploy { agent, .. } => agent.clone(),
other => panic!("not a build or a deploy: {other:?}"),
})
.collect();
found.sort();
found
};
assert_eq!(again("build_meta"), ["argus"], "{error}");
assert_eq!(again("trigger_deploy"), ["argus"], "{error}");
}
}
/// With nothing placed there is nothing to wait on, and the pin is
@ -3854,89 +3811,6 @@ mod tests {
nodes.into_iter().map(|(_, rev)| rev).collect()
}
/// A merged config's build that gave no verdict, with the update `Done`:
/// the deploy is skipped, and the timer declares a build and the same
/// deploy again, without an update, until the builder answers.
#[tokio::test]
async fn an_unfinished_build_is_retried_until_the_builder_answers() {
use hive_jobq::State;
let sched = fresh_sched();
sched
.lock()
.unwrap()
.insert_job(None, |b| merged_deploy_job(b, "pr1ma", "atlas", "abc"))
.expect("the graph inserts");
let unavailable = format!("{}no swarm builder is listening", meta_build::RETRY_PREFIX);
drain_failing_build(&sched, "atlas", unavailable.clone()).await;
assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]);
assert_eq!(update_metas(&sched), 1);
assert!(queue_meta_update(&sched).expect("tick"));
assert_eq!(update_metas(&sched), 1, "the retry needs no update");
drain_failing_build(&sched, "atlas", unavailable).await;
assert_eq!(
deploys_of(&sched, "atlas"),
[State::Skipped, State::Skipped]
);
assert!(queue_meta_update(&sched).expect("tick"));
drain(&sched, false, false).await;
assert_eq!(
deploys_of(&sched, "atlas"),
[State::Skipped, State::Skipped, State::Done]
);
// The merged deploy's rev survives every retry.
assert_eq!(
deploy_revs(&sched, "atlas"),
[
Some("abc".to_owned()),
Some("abc".to_owned()),
Some("abc".to_owned())
]
);
assert!(
!queue_meta_update(&sched).expect("tick"),
"the answer ends the retries"
);
}
/// A red build is a verdict: the timer declares nothing for it. The test
/// above is the control that the same shape with an unfinished build is
/// declared again.
#[tokio::test]
async fn a_red_build_is_not_declared_again() {
use hive_jobq::State;
let sched = fresh_sched();
sched
.lock()
.unwrap()
.insert_job(None, |b| merged_deploy_job(b, "pr1ma", "atlas", "abc"))
.expect("the graph inserts");
drain_failing_build(&sched, "atlas", "error: builder for x failed".to_owned()).await;
assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]);
assert!(!queue_meta_update(&sched).expect("tick"));
assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]);
}
/// The `attempt` of `agent`'s builds, oldest first.
fn build_attempts(sched: &SharedSched, agent: &str) -> Vec<u32> {
let guard = sched.lock().unwrap();
let mut nodes: Vec<_> = guard
.graph()
.nodes()
.filter_map(|n| match &n.payload {
SwarmNodeKind::BuildMeta { agent: a, attempt } if a == agent => {
Some((n.id, *attempt))
}
_ => None,
})
.collect();
nodes.sort_by_key(|(id, _)| *id);
nodes.into_iter().map(|(_, attempt)| attempt).collect()
}
fn merged_atlas() -> SharedSched {
let sched = fresh_sched();
sched
@ -3947,55 +3821,54 @@ mod tests {
sched
}
/// `nix` failing once is built again on the next tick, as the second
/// attempt; failing again is red, and the tick declares nothing more.
/// A merged config's build that fails, with the update `Done`, whatever it
/// fails with: the deploy is skipped, and the timer declares a build and
/// the same deploy again on every round, without an update, until a build
/// succeeds. Then it stops.
#[tokio::test]
async fn a_first_nix_error_is_built_once_more_and_the_second_is_red() {
async fn a_failed_build_is_built_again_every_round_until_it_succeeds() {
use hive_jobq::State;
let sched = merged_atlas();
let first = format!("{}error: fetch failed", meta_build::FIRST_FAILURE_PREFIX);
drain_failing_build(&sched, "atlas", first).await;
let errors = [
"error: attribute missing".to_owned(),
"timed out: nix did not finish".to_owned(),
format!(
"{}no swarm builder is listening",
meta_build::NO_VERDICT_PREFIX
),
];
drain_failing_build(&sched, "atlas", errors[0].clone()).await;
assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]);
assert_eq!(build_attempts(&sched, "atlas"), [0]);
assert_eq!(update_metas(&sched), 1);
assert!(queue_meta_update(&sched).expect("tick"));
assert_eq!(build_attempts(&sched, "atlas"), [0, 1]);
drain_failing_build(&sched, "atlas", "error: attribute missing".to_owned()).await;
for error in &errors[1..] {
assert!(queue_meta_update(&sched).expect("tick"));
assert_eq!(update_metas(&sched), 1, "the retry needs no update");
drain_failing_build(&sched, "atlas", error.clone()).await;
}
assert_eq!(
deploys_of(&sched, "atlas"),
[State::Skipped, State::Skipped]
[State::Skipped, State::Skipped, State::Skipped]
);
assert!(!queue_meta_update(&sched).expect("tick"));
assert_eq!(build_attempts(&sched, "atlas"), [0, 1]);
}
/// A build that passes after a first `nix` error sends the deploy, once.
#[tokio::test]
async fn a_build_that_passes_after_a_first_nix_error_deploys() {
use hive_jobq::State;
let sched = merged_atlas();
let first = format!("{}error: fetch failed", meta_build::FIRST_FAILURE_PREFIX);
drain_failing_build(&sched, "atlas", first).await;
// A success on a later round deploys, once.
assert!(queue_meta_update(&sched).expect("tick"));
drain(&sched, false, false).await;
assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped, State::Done]);
assert!(!queue_meta_update(&sched).expect("tick"));
}
/// A build with no verdict keeps its attempt, however many ticks it takes,
/// so a builder that is away does not use up the second `nix` attempt.
#[tokio::test]
async fn a_build_with_no_verdict_keeps_its_attempt() {
let sched = merged_atlas();
let away = format!("{}no builder", meta_build::RETRY_PREFIX);
drain_failing_build(&sched, "atlas", away.clone()).await;
assert!(queue_meta_update(&sched).expect("tick"));
drain_failing_build(&sched, "atlas", away).await;
assert!(queue_meta_update(&sched).expect("tick"));
assert_eq!(build_attempts(&sched, "atlas"), [0, 0, 0]);
assert_eq!(
deploys_of(&sched, "atlas"),
[State::Skipped, State::Skipped, State::Skipped, State::Done]
);
// The merged deploy's rev survives every retry.
assert!(
deploy_revs(&sched, "atlas")
.iter()
.all(|rev| rev.as_deref() == Some("abc"))
);
assert!(
!queue_meta_update(&sched).expect("tick"),
"the success ends the retries"
);
}
}

161
swarm-controller/src/meta_build.rs
View file

@ -8,17 +8,17 @@
//! key itself: the builder holds no grant on the bucket. A `Built` reply and
//! nothing else writes, so a red build leaves the previous record.
//!
//! The node ends one of three ways. `Done` is a green build. Every `Failed` skips
//! the deploy behind it, and the Jobs page shows why. What decides whether the
//! timer builds again (`jobq::queue_meta_update`) is the error text:
//! - No verdict (no builder, a timeout, a forge error, no `internal/meta` head,
//! a missing read grant, a failed closure write): [`RETRY_PREFIX`]. The next
//! build has the same `attempt`.
//! - `nix` exited with an error on the first attempt: [`FIRST_FAILURE_PREFIX`]
//! and `nix`'s tail. The next build is `attempt + 1`, which catches a fetch
//! hiccup inside `nix` without reading its output.
//! - `nix` exited with an error again: `nix`'s tail alone. That is a red build:
//! the deploy stays held until a config merge, and nothing retries it.
//! The node ends one of two ways. `Done` is a green build. Every `Failed` skips
//! the deploy behind it, and the Jobs page shows why:
//! - `nix` failed, its timeout included: `Failed` with `nix`'s tail, or `nix
//! timed out`. The deploy stays held and counts as a failed build.
//! - The build got no verdict (no builder, a forge error, no `internal/meta`
//! head, a missing read grant, a failed closure write): `Failed` starting with
//! [`NO_VERDICT_PREFIX`], which the held-deploys gauge reads to label the
//! reason.
//!
//! Both are built again on every round of the meta timer
//! (`jobq::queue_meta_update`) until a build succeeds.
use std::fmt::Display;
use std::future::Future;
@ -32,32 +32,16 @@ use crate::agent_renewal::unix_now;
use crate::forge::{self, objects::SHARED_ORG};
use crate::meta_flake::{META_REPO, MetaForge};
/// What a `Failed` outcome that is not a verdict on the config starts with.
pub(crate) const RETRY_PREFIX: &str = "not built, will be retried: ";
/// What a `Failed` outcome starts with when the build got no verdict.
pub(crate) const NO_VERDICT_PREFIX: &str = "no verdict: ";
/// What the `Failed` outcome of a first `nix` error starts with.
pub(crate) const FIRST_FAILURE_PREFIX: &str = "build failed, will be built once more: ";
/// How many times `nix` may exit with an error before the build is red.
const NIX_ATTEMPTS: u32 = 2;
/// Whether a `Failed` node's `error` asks for another build.
pub(crate) fn is_retried(error: &str) -> bool {
error.starts_with(RETRY_PREFIX) || error.starts_with(FIRST_FAILURE_PREFIX)
/// Whether a `Failed` node's `error` says the build got no verdict.
pub(crate) fn is_no_verdict(error: &str) -> bool {
error.starts_with(NO_VERDICT_PREFIX)
}
/// The `attempt` of the build that follows a `Failed` build of `attempt`
/// with `error`: one more after an error `nix` gave, the same after none.
pub(crate) fn next_attempt(attempt: u32, error: &str) -> u32 {
if error.starts_with(FIRST_FAILURE_PREFIX) {
attempt + 1
} else {
attempt
}
}
fn not_built(why: impl Display) -> Outcome {
Outcome::Failed(format!("{RETRY_PREFIX}{why}"))
fn no_verdict(why: impl Display) -> Outcome {
Outcome::Failed(format!("{NO_VERDICT_PREFIX}{why}"))
}
/// Where `internal/meta` is read and shared with the builder: the forge in
@ -136,47 +120,41 @@ impl MetaBuilds for async_nats::Client {
}
/// Build `agent` from `internal/meta`'s head and record the closure.
/// `attempt` is how many times `nix` has already exited with an error for this
/// deploy.
pub(crate) async fn run(
head: Option<&impl MetaRepo>,
queue: Option<&impl MetaBuilds>,
agent: String,
attempt: u32,
) -> Outcome {
let Some(head) = head else {
return not_built(
return no_verdict(
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token not in the \
swarm secret store), so internal/meta's head cannot be read",
);
};
let Some(queue) = queue else {
return not_built("no swarm queue is configured on this host, so no build can be sent");
return no_verdict("no swarm queue is configured on this host, so no build can be sent");
};
// The repo is private, and the grant is otherwise made only by a meta update.
if let Err(e) = head.grant_builder_read().await {
return not_built(format!(
return no_verdict(format!(
"the builder's read on internal/meta could not be granted: {e:#}"
));
}
let meta_rev = match head.head().await {
Ok(Some(rev)) => rev,
Ok(None) => return not_built("the swarm has not published internal/meta yet"),
Err(e) => return not_built(format!("{e:#}")),
Ok(None) => return no_verdict("the swarm has not published internal/meta yet"),
Err(e) => return no_verdict(format!("{e:#}")),
};
let request = match serde_json::to_vec(&BuildFromMeta {
agent: agent.clone(),
meta_rev: meta_rev.clone(),
}) {
Ok(request) => request,
Err(e) => return not_built(format!("encoding the build request: {e}")),
Err(e) => return no_verdict(format!("encoding the build request: {e}")),
};
match queue.build(request).await {
Err(why) => not_built(why),
Ok(FromMetaReply::Unavailable { reason }) => not_built(reason),
Ok(FromMetaReply::Failed { error_tail }) if attempt + 1 < NIX_ATTEMPTS => {
Outcome::Failed(format!("{FIRST_FAILURE_PREFIX}{error_tail}"))
}
Err(why) => no_verdict(why),
Ok(FromMetaReply::Unavailable { reason }) => no_verdict(reason),
Ok(FromMetaReply::Failed { error_tail }) => Outcome::Failed(error_tail),
Ok(FromMetaReply::Built { out_path, system }) => {
let closure = AgentClosure {
@ -187,7 +165,7 @@ pub(crate) async fn run(
};
match queue.record(&agent, &closure).await {
Ok(()) => Outcome::Done,
Err(e) => not_built(format!("built {out_path}, but recording it failed: {e:#}")),
Err(e) => no_verdict(format!("built {out_path}, but recording it failed: {e:#}")),
}
}
}
@ -278,11 +256,7 @@ mod tests {
}
async fn run_with(head: &FakeHead, queue: &FakeQueue) -> Outcome {
run_at(head, queue, 0).await
}
async fn run_at(head: &FakeHead, queue: &FakeQueue, attempt: u32) -> Outcome {
run(Some(head), Some(queue), "atlas".to_owned(), attempt).await
run(Some(head), Some(queue), "atlas".to_owned()).await
}
#[tokio::test]
@ -308,56 +282,30 @@ mod tests {
assert!(closure.built_unix > 0);
}
fn nix_error() -> FakeQueue {
FakeQueue::answering(Ok(FromMetaReply::Failed {
error_tail: "error: builder for foo failed".to_owned(),
}))
}
/// `nix` exiting with an error the first time is asked for once more: the
/// failure carries the tail and the retry marker, and the next build is the
/// second attempt. Nothing is recorded.
/// Any `nix` failure, a timeout included, fails the node at once with what
/// the builder said: not a no-verdict failure, and nothing is recorded. The
/// green test above is the control that a record is written when the build
/// is green.
#[tokio::test]
async fn the_first_nix_error_is_built_once_more() {
let queue = nix_error();
let Outcome::Failed(error) = run_with(&FakeHead::reading(Ok(Some(REV))), &queue).await
else {
panic!("not failed")
};
assert!(is_retried(&error), "{error}");
assert!(error.ends_with("error: builder for foo failed"), "{error}");
assert_eq!(next_attempt(0, &error), 1);
assert!(queue.recorded.lock().unwrap().is_empty());
}
/// The second `nix` error is a red build: the tail alone, nothing retries
/// it, nothing is recorded. The green test above is the control that a
/// record is written when the build is green.
#[tokio::test]
async fn the_second_nix_error_is_red_and_records_nothing() {
let queue = nix_error();
let outcome = run_at(&FakeHead::reading(Ok(Some(REV))), &queue, 1).await;
assert_eq!(
outcome,
Outcome::Failed("error: builder for foo failed".to_owned())
);
assert!(!is_retried("error: builder for foo failed"));
assert!(queue.recorded.lock().unwrap().is_empty());
}
/// A build with no verdict keeps its attempt, so a builder that is away for
/// hours does not use up the retry that a nix error is allowed.
#[test]
fn only_a_nix_error_uses_up_an_attempt() {
assert_eq!(next_attempt(0, &format!("{RETRY_PREFIX}no builder")), 0);
assert_eq!(next_attempt(1, &format!("{RETRY_PREFIX}no builder")), 1);
assert_eq!(next_attempt(0, &format!("{FIRST_FAILURE_PREFIX}tail")), 1);
async fn a_nix_failure_fails_the_node_with_its_tail_and_records_nothing() {
for tail in [
"error: builder for foo failed",
"timed out: nix took 60 minutes",
] {
let queue = FakeQueue::answering(Ok(FromMetaReply::Failed {
error_tail: tail.to_owned(),
}));
let outcome = run_with(&FakeHead::reading(Ok(Some(REV))), &queue).await;
assert_eq!(outcome, Outcome::Failed(tail.to_owned()));
assert!(!is_no_verdict(tail));
assert!(queue.recorded.lock().unwrap().is_empty(), "{tail}");
}
}
/// A read grant the forge will not make is the controller's precondition,
/// not the config's: nothing is sent and the failure is retried.
/// not the config's: nothing is sent and the failure has no verdict.
#[tokio::test]
async fn a_read_grant_that_cannot_be_made_is_retried_before_anything_is_sent() {
async fn a_read_grant_that_cannot_be_made_is_a_no_verdict_failure_before_anything_is_sent() {
let queue = FakeQueue::built();
let head = FakeHead {
grant_fails: true,
@ -366,13 +314,13 @@ mod tests {
let Outcome::Failed(error) = run_with(&head, &queue).await else {
panic!("not failed")
};
assert!(is_retried(&error), "{error}");
assert!(is_no_verdict(&error), "{error}");
assert!(error.contains("read on internal/meta"), "{error}");
assert!(queue.sent.lock().unwrap().is_empty());
}
#[tokio::test]
async fn a_builder_that_gave_no_verdict_is_a_retried_failure_and_records_nothing() {
async fn a_builder_that_gave_no_verdict_is_a_no_verdict_failure_and_records_nothing() {
for (name, reply) in [
(
"no responders",
@ -390,13 +338,13 @@ mod tests {
let Outcome::Failed(error) = outcome else {
panic!("{name}: not failed")
};
assert!(is_retried(&error), "{name}: {error}");
assert!(is_no_verdict(&error), "{name}: {error}");
assert!(queue.recorded.lock().unwrap().is_empty(), "{name}");
}
}
#[tokio::test]
async fn a_closure_that_cannot_be_recorded_is_a_retried_failure() {
async fn a_closure_that_cannot_be_recorded_is_a_no_verdict_failure() {
let queue = FakeQueue {
record_fails: true,
..FakeQueue::built()
@ -405,12 +353,12 @@ mod tests {
else {
panic!("not failed")
};
assert!(is_retried(&error), "{error}");
assert!(is_no_verdict(&error), "{error}");
assert!(error.contains("/nix/store/x-toplevel"), "{error}");
}
#[tokio::test]
async fn without_a_head_or_a_queue_nothing_is_sent_and_the_failure_is_retried() {
async fn without_a_head_or_a_queue_nothing_is_sent_and_the_failure_has_no_verdict() {
for (name, head, with_queue) in [
("no repo yet", Some(FakeHead::reading(Ok(None))), true),
(
@ -426,13 +374,12 @@ mod tests {
head.as_ref(),
with_queue.then_some(&queue),
"atlas".to_owned(),
0,
)
.await;
let Outcome::Failed(error) = outcome else {
panic!("{name}: not failed")
};
assert!(is_retried(&error), "{name}: {error}");
assert!(is_no_verdict(&error), "{name}: {error}");
assert!(queue.sent.lock().unwrap().is_empty(), "{name}");
}
}

6
swarm-controller/src/meta_fanout.rs
View file

@ -6,9 +6,9 @@
//! `TriggerDeploy` of that fan-out is done or skipped by its build gate. A pass
//! that dies, or a deploy that fails, before then leaves the record stale, so
//! the next pass sends the deploys again. A deploy the gate skipped counts as
//! sent: `jobq::queue_meta_update` declares it again when its build gave no
//! verdict, and a red build waits for the next config merge. A deploy is
//! idempotent per pin, so sending twice is safe and sending never is not.
//! sent: `jobq::queue_meta_update` declares it again on every round until its
//! build succeeds. A deploy is idempotent per pin, so sending twice is safe and
//! sending never is not.
//!
//! With no record, as on the first pass of a controller that has never written
//! one, the pin is not the recorded one and the agents are sent a deploy.