redeclare the deploys a failed meta update skipped on the timer's retry

queue_meta_update adds a TriggerDeploy behind each retry UpdateMeta (after_ok) for every agent whose latest deploy waited on a failed update. A created or merged agent whose deploy was skipped is deployed by the run that succeeds; a deploy skipped for another reason, or an agent already deployed, gets nothing.

swarm-controller/src/jobq.rs

@@ -1488,8 +1488,53 @@ pub(crate) fn spawn_meta_update_timer(
     });
 }
 
+/// The deploys a failed `UpdateMeta` skipped, as `(hive, agent, rev)`: for each
+/// agent, its latest `TriggerDeploy` when it waited on an `UpdateMeta` that
+/// `Failed` (so the scheduler skipped it). A deploy skipped for another reason
+/// (the pause declaration or the config seeding failed) is left alone: the
+/// update that would unblock it was skipped too, and deploying past those is
+/// the race those edges close.
+fn skipped_deploys(
+    graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>,
+) -> Vec<(String, String, Option<String>)> {
+    let mut latest = std::collections::BTreeMap::new();
+    for n in graph.nodes() {
+        if let SwarmNodeKind::TriggerDeploy { hive, agent, .. } = &n.payload {
+            let key = (hive.clone(), agent.clone());
+            if latest
+                .get(&key)
+                .is_none_or(|l: &&hive_jobq::Node<_, _>| l.id < n.id)
+            {
+                latest.insert(key, n);
+            }
+        }
+    }
+    latest
+        .into_values()
+        .filter(|n| {
+            n.deps.iter().any(|d| match d {
+                hive_jobq::Dep::Node { id, .. } => graph.node(*id).is_some_and(|dep| {
+                    matches!(dep.payload, SwarmNodeKind::UpdateMeta)
+                        && dep.state == hive_jobq::State::Failed
+                }),
+                hive_jobq::Dep::Resource { .. } => false,
+            })
+        })
+        .filter_map(|n| match &n.payload {
+            SwarmNodeKind::TriggerDeploy { hive, agent, rev } => {
+                Some((hive.clone(), agent.clone(), rev.clone()))
+            }
+            _ => None,
+        })
+        .collect()
+}
+
 /// 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.
+/// node id, is not `Done` (or none exists), with a fresh `TriggerDeploy`
+/// behind it (`after_ok`) for every deploy the failed run skipped
+/// ([`skipped_deploys`]), so an agent created or merged while the update was
+/// failing is deployed by the run that succeeds. Returns whether a job was
+/// inserted.
 fn queue_meta_update(
     sched: &Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
 ) -> Result<bool> {
@@ -1516,13 +1561,18 @@ fn queue_meta_update(
     {
         return Ok(false);
     }
+    let skipped = skipped_deploys(sched.graph());
     sched
         .insert_job(None, |b| {
-            vec![
-                b.node(SwarmNodeKind::UpdateMeta)
-                    .needs(SwarmResourceKind::Meta)
-                    .guid(),
-            ]
+            let update = b
+                .node(SwarmNodeKind::UpdateMeta)
+                .needs(SwarmResourceKind::Meta);
+            for (hive, agent, rev) in skipped {
+                let _ = b
+                    .node(SwarmNodeKind::TriggerDeploy { hive, agent, rev })
+                    .after_ok(update);
+            }
+            vec![update.guid()]
         })
         .map_err(|e| anyhow::anyhow!("{e}"))?;
     Ok(true)
@@ -3241,6 +3291,143 @@ mod tests {
         assert!(!super::queue_meta_update(&sched).expect("tick"));
         assert_eq!(update_metas(&sched), 2);
     }
+
+    type SharedSched = std::sync::Arc<
+        std::sync::Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
+    >;
+
+    /// Declare `agent`'s creation, as the endpoint does, and run every node to
+    /// the end: the meta update fails when `update_meta_fails`, the pause
+    /// declaration when `set_wanted_fails`, everything else succeeds.
+    async fn create_and_run(
+        sched: &SharedSched,
+        agent: &str,
+        update_meta_fails: bool,
+        set_wanted_fails: bool,
+    ) {
+        sched
+            .lock()
+            .unwrap()
+            .insert_job(None, |b| super::declare_agent_job(b, agent, "pr1ma", false))
+            .expect("the graph inserts");
+        drain(sched, update_meta_fails, set_wanted_fails).await;
+    }
+
+    /// Run every runnable node; outcomes as for [`create_and_run`].
+    async fn drain(sched: &SharedSched, update_meta_fails: bool, set_wanted_fails: bool) {
+        use hive_jobq::scheduler::{Outcome, Scheduler};
+
+        while let Some(runner) = Scheduler::claim_next(sched, |_id, kind, builder| async move {
+            let outcome = match kind {
+                SwarmNodeKind::UpdateMeta if update_meta_fails => {
+                    Outcome::Failed("no builder".to_owned())
+                }
+                SwarmNodeKind::SetAgentWanted { .. } if set_wanted_fails => {
+                    Outcome::Failed("no queue".to_owned())
+                }
+                _ => Outcome::Done,
+            };
+            (builder, outcome)
+        }) {
+            runner.await.1.expect("no growth declared");
+        }
+    }
+
+    /// The states of `agent`'s `TriggerDeploy` nodes, oldest first.
+    fn deploys_of(sched: &SharedSched, agent: &str) -> Vec<hive_jobq::State> {
+        let guard = sched.lock().unwrap();
+        let mut nodes: Vec<_> = guard
+            .graph()
+            .nodes()
+            .filter(|n| matches!(&n.payload, SwarmNodeKind::TriggerDeploy { agent: a, .. } if a == agent))
+            .collect();
+        nodes.sort_by_key(|n| n.id);
+        nodes.into_iter().map(|n| n.state).collect()
+    }
+
+    /// `bob` deployed under a working update; then `atlas` was created while
+    /// the update failed, so its deploy is `Skipped`.
+    async fn bob_deployed_and_atlas_skipped() -> SharedSched {
+        use hive_jobq::State;
+
+        let sched: SharedSched =
+            std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
+                hive_jobq::Graph::new(),
+                hive_jobq::resources::ResourceTable::new(),
+            )));
+        create_and_run(&sched, "bob", false, false).await;
+        create_and_run(&sched, "atlas", true, false).await;
+        assert_eq!(deploys_of(&sched, "bob"), [State::Done]);
+        assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]);
+        sched
+    }
+
+    /// An agent created while the update was failing is deployed by the run
+    /// that succeeds, exactly once.
+    #[tokio::test]
+    async fn the_next_successful_update_deploys_the_agent_a_failed_one_skipped() {
+        use hive_jobq::State;
+
+        let sched = bob_deployed_and_atlas_skipped().await;
+        assert!(super::queue_meta_update(&sched).expect("tick"));
+        // A retry whose update fails too skips the new deploy as well.
+        drain(&sched, true, false).await;
+        assert_eq!(
+            deploys_of(&sched, "atlas"),
+            [State::Skipped, State::Skipped]
+        );
+        assert!(super::queue_meta_update(&sched).expect("tick"));
+        drain(&sched, false, false).await;
+        assert_eq!(
+            deploys_of(&sched, "atlas"),
+            [State::Skipped, State::Skipped, State::Done]
+        );
+        // The success ends the retries: no tick adds a node, so no second deploy.
+        assert!(!super::queue_meta_update(&sched).expect("tick"));
+        assert_eq!(
+            deploys_of(&sched, "atlas"),
+            [State::Skipped, State::Skipped, State::Done]
+        );
+    }
+
+    /// An agent whose latest deploy ran gets no extra one from the retry, even
+    /// when an older deploy of its was skipped.
+    #[tokio::test]
+    async fn an_agent_that_was_deployed_gets_no_extra_deploy_from_the_retry() {
+        use hive_jobq::State;
+
+        let sched = bob_deployed_and_atlas_skipped().await;
+        assert!(super::queue_meta_update(&sched).expect("tick"));
+        drain(&sched, false, false).await;
+        assert_eq!(deploys_of(&sched, "bob"), [State::Done]);
+
+        // atlas now holds an old Skipped deploy and a Done one. Another
+        // creation fails its update; the retry deploys carol and only carol.
+        create_and_run(&sched, "carol", true, false).await;
+        assert!(super::queue_meta_update(&sched).expect("tick"));
+        drain(&sched, false, false).await;
+        assert_eq!(deploys_of(&sched, "carol"), [State::Skipped, State::Done]);
+        assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped, State::Done]);
+        assert_eq!(deploys_of(&sched, "bob"), [State::Done]);
+    }
+
+    /// A deploy skipped because the pause declaration failed stays skipped: the
+    /// retry must not deploy past the declaration it waits for.
+    #[tokio::test]
+    async fn a_deploy_skipped_for_another_reason_is_not_redeclared() {
+        use hive_jobq::State;
+
+        let sched: SharedSched =
+            std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
+                hive_jobq::Graph::new(),
+                hive_jobq::resources::ResourceTable::new(),
+            )));
+        create_and_run(&sched, "atlas", false, true).await;
+        assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]);
+        assert!(super::queue_meta_update(&sched).expect("tick"));
+        drain(&sched, false, false).await;
+        assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]);
+    }
 }
 
 #[cfg(test)]