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

queue the meta update only until one run succeeds

Browse source
This commit is contained in:
atlas 2026-10-25 12:56:30 +01:00
commit 31cb6bfa42
2 changed files with 91 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1467,13 +1467,14 @@ pub(crate) fn queue_builder_forge_token_mint(
.map_err(|e| anyhow::anyhow!("{e}"))
}
/// How often [`spawn_meta_update_timer`] queues an `UpdateMeta`.
/// How often [`spawn_meta_update_timer`] looks at the last `UpdateMeta`.
const META_UPDATE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(10);
/// Queue an `UpdateMeta` now and every [`META_UPDATE_INTERVAL`] after, so a
/// boot run that failed or hung, and drift the config-merge path did not see,
/// are picked up. A tick that finds the previous run pending, running or
/// finishing queues nothing.
/// Look at the last `UpdateMeta` now and every [`META_UPDATE_INTERVAL`] after,
/// and queue one when none has run yet or the latest did not succeed, so a
/// boot run that failed or hung is retried. The first tick is the boot run.
/// Once a run has succeeded a tick adds nothing: the graph keeps every node,
/// so a clock-driven node per tick would grow it without bound.
pub(crate) fn spawn_meta_update_timer(
sched: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
) {
@ -1488,21 +1489,32 @@ pub(crate) fn spawn_meta_update_timer(
});
}
/// Insert one `UpdateMeta` job unless one is still in flight, which reads the
/// same heads when it runs. Returns whether a job was inserted.
/// Insert one `UpdateMeta` job when none is in flight and the latest one, by
/// node id, is not `Done` (or none exists). Returns whether a job was inserted.
fn queue_meta_update(
sched: &Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
) -> Result<bool> {
let mut sched = sched
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let in_flight = sched.graph().nodes().any(|n| {
let mut update_metas = sched
.graph()
.nodes()
.filter(|n| matches!(n.payload, SwarmNodeKind::UpdateMeta))
.collect::<Vec<_>>();
if update_metas.iter().any(|n| {
matches!(
n.state,
hive_jobq::State::Pending | hive_jobq::State::Running | hive_jobq::State::Finishing
) && matches!(n.payload, SwarmNodeKind::UpdateMeta)
});
if in_flight {
)
}) {
return Ok(false);
}
update_metas.sort_by_key(|n| n.id);
if update_metas
.last()
.is_some_and(|n| n.state == hive_jobq::State::Done)
{
return Ok(false);
}
sched
@ -3115,40 +3127,86 @@ mod tests {
assert_eq!(labels, ["create_forge_user", "mint_agent_forge_token"]);
}
/// The timer's tick queues an `UpdateMeta` when none is in flight, and
/// queues nothing while one is pending or running; once it is done the
/// next tick queues again.
/// A scheduler holding one `UpdateMeta` that ran and ended `outcome`.
async fn after_one_update_meta(
outcome: hive_jobq::scheduler::Outcome,
) -> std::sync::Arc<
std::sync::Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
> {
let sched =
std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
hive_jobq::Graph::new(),
hive_jobq::resources::ResourceTable::new(),
)));
assert!(super::queue_meta_update(&sched).expect("tick"));
hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move { (b, outcome) })
.expect("runnable")
.await
.1
.expect("no growth");
sched
}
fn update_metas(
sched: &std::sync::Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
) -> usize {
sched
.lock()
.unwrap()
.graph()
.nodes()
.filter(|n| matches!(n.payload, SwarmNodeKind::UpdateMeta))
.count()
}
/// With none run yet the tick queues one, and queues no second while that
/// one is pending or running.
#[tokio::test]
async fn the_meta_timer_queues_unless_one_is_in_flight() {
async fn the_meta_timer_queues_first_and_skips_while_one_is_in_flight() {
let sched =
std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
hive_jobq::Graph::new(),
hive_jobq::resources::ResourceTable::new(),
)));
let update_metas = || {
sched
.lock()
.unwrap()
.graph()
.nodes()
.filter(|n| matches!(n.payload, SwarmNodeKind::UpdateMeta))
.count()
};
assert!(super::queue_meta_update(&sched).expect("tick"));
assert_eq!(update_metas(), 1);
assert!(!super::queue_meta_update(&sched).expect("tick"));
assert_eq!(update_metas(), 1, "a pending one blocks the tick");
assert_eq!(update_metas(&sched), 1, "a pending one blocks the tick");
let running = hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move {
(b, hive_jobq::scheduler::Outcome::Done)
(b, hive_jobq::scheduler::Outcome::Failed("late".to_owned()))
})
.expect("runnable");
assert!(!super::queue_meta_update(&sched).expect("tick"));
assert_eq!(update_metas(), 1, "a running one blocks the tick");
assert_eq!(update_metas(&sched), 1, "a running one blocks the tick");
running.await.1.expect("no growth");
}
/// Once the latest run succeeded, ticks add no node.
#[tokio::test]
async fn the_meta_timer_adds_nothing_after_a_success() {
let sched = after_one_update_meta(hive_jobq::scheduler::Outcome::Done).await;
for _ in 0..3 {
assert!(!super::queue_meta_update(&sched).expect("tick"));
}
assert_eq!(update_metas(&sched), 1);
}
/// A failed latest run is retried; a success after it ends the retries.
#[tokio::test]
async fn the_meta_timer_retries_after_a_failure_until_a_success() {
let sched =
after_one_update_meta(hive_jobq::scheduler::Outcome::Failed("no".to_owned())).await;
assert!(super::queue_meta_update(&sched).expect("tick"));
assert_eq!(update_metas(), 2, "a finished one does not");
assert_eq!(update_metas(&sched), 2);
hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move {
(b, hive_jobq::scheduler::Outcome::Done)
})
.expect("runnable")
.await
.1
.expect("no growth");
assert!(!super::queue_meta_update(&sched).expect("tick"));
assert_eq!(update_metas(&sched), 2);
}
}