log failed controller jobs, re-run the meta update every ten minutes

docs/swarm/builder.md

@@ -143,7 +143,8 @@ running rev (`services.hyperhive.c0re.hyperhiveRev`, from `self.rev`):
 
 The controller commits `flake.nix` and the returned `flake.lock` to
 `internal/meta` in one commit, and gives each hive's forge account read on the
-repo. It re-renders when it starts, before it deploys a merged config PR, and
+repo. It re-renders when it starts and every ten minutes after (a tick skips
+while an update is pending or running), before it deploys a merged config PR, and
 before the first deploy of a new agent. Those deploys run whether or not the
 update succeeded: a hive that builds from `internal/meta` does so only while it
 pins the agent's current config (see

swarm-controller/README.md

@@ -22,7 +22,9 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients.
   reported takes its hive and declared state in `/api/agents/status` from the
   one hive whose declaration places it.
 - **Job graph** — a `hive-jobq` scheduler, served at `GET /api/jobq/graph`.
-  Every provisioning step below runs as a node in it.
+  Every provisioning step below runs as a node in it. A node that fails logs
+  one `warn` line, `swarm jobq: node failed`, with its `node` id, `kind` and
+  `error`.
 - **Agent creation** — `POST /api/agents` queues the SSO identity (through
   `swarm-authelia-bridge`), forge user, config repo, store identity, forge
   token and matrix account, declares the agent `paused`, then sends its hive

swarm-controller/src/jobq.rs

@@ -291,11 +291,28 @@ struct MetaDeps {
 const NEW_AGENT_WANTED_STATE: swarm_wire::wanted::AgentState =
     swarm_wire::wanted::AgentState::Paused;
 
+/// Run a claimed node and log one warn when it fails, whatever its kind.
+pub(crate) async fn run_swarm_node(
+    id: hive_jobq::NodeId,
+    kind: SwarmNodeKind,
+    builder: hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
+    deps: WorkerDeps,
+) -> (
+    hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
+    hive_jobq::scheduler::Outcome,
+) {
+    let node_kind = hive_jobq_wire::WireNode::label(&kind);
+    let (builder, outcome) = run_node_work(kind, builder, deps).await;
+    if let hive_jobq::scheduler::Outcome::Failed(error) = &outcome {
+        tracing::warn!(node = id.get(), kind = %node_kind, %error, "swarm jobq: node failed");
+    }
+    (builder, outcome)
+}
+
 /// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/
 /// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind`
 /// variant turns into a real effect.
-pub(crate) async fn run_swarm_node(
-    _id: hive_jobq::NodeId,
+async fn run_node_work(
     kind: SwarmNodeKind,
     builder: hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
     deps: WorkerDeps,
@@ -431,7 +448,7 @@ async fn forge_node<T>(
 /// together.
 const MINT_IDENTITY_FAILED: &str = "swarm jobq: mint_agent_identity failed";
 
-/// The `MintAgentIdentity` arm, lifted out so `run_swarm_node` stays under
+/// The `MintAgentIdentity` arm, lifted out so `run_node_work` stays under
 /// `clippy::too_many_lines`.
 pub(crate) async fn mint_identity(agent: &str) -> hive_jobq::scheduler::Outcome {
     use hive_jobq::scheduler::Outcome;
@@ -446,7 +463,7 @@ pub(crate) async fn mint_identity(agent: &str) -> hive_jobq::scheduler::Outcome
     }
 }
 
-/// The `MintAgentForgeToken` arm, lifted out so `run_swarm_node` stays under
+/// The `MintAgentForgeToken` arm, lifted out so `run_node_work` stays under
 /// `clippy::too_many_lines`.
 async fn mint_forge_token(
     forge: Option<Arc<forge::Client>>,
@@ -681,7 +698,7 @@ fn finish_update_meta(
     }
 }
 
-/// The `RecordFanout` arm, lifted out so `run_swarm_node` stays under the line limit.
+/// The `RecordFanout` arm, lifted out so `run_node_work` stays under the line limit.
 async fn record_fanout(
     queue: Option<&async_nats::Client>,
     pin: &str,
@@ -1450,18 +1467,42 @@ pub(crate) fn queue_builder_forge_token_mint(
         .map_err(|e| anyhow::anyhow!("{e}"))
 }
 
-/// Insert one `UpdateMeta` job unless one is already pending, which reads the
+/// How often [`spawn_meta_update_timer`] queues an `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.
+pub(crate) fn spawn_meta_update_timer(
+    sched: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
+) {
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(META_UPDATE_INTERVAL);
+        loop {
+            ticker.tick().await;
+            if let Err(e) = queue_meta_update(&sched) {
+                tracing::warn!(error = %format!("{e:#}"), "update_meta: queueing failed");
+            }
+        }
+    });
+}
+
+/// Insert one `UpdateMeta` job unless one is still in flight, which reads the
 /// same heads when it runs. Returns whether a job was inserted.
-pub(crate) fn queue_meta_update(
+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 pending = sched.graph().nodes().any(|n| {
-        n.state == hive_jobq::State::Pending && matches!(n.payload, SwarmNodeKind::UpdateMeta)
+    let in_flight = sched.graph().nodes().any(|n| {
+        matches!(
+            n.state,
+            hive_jobq::State::Pending | hive_jobq::State::Running | hive_jobq::State::Finishing
+        ) && matches!(n.payload, SwarmNodeKind::UpdateMeta)
     });
-    if pending {
+    if in_flight {
         return Ok(false);
     }
     sched
@@ -3073,6 +3114,42 @@ mod tests {
         labels.sort();
         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.
+    #[tokio::test]
+    async fn the_meta_timer_queues_unless_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");
+
+        let running = hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move {
+            (b, hive_jobq::scheduler::Outcome::Done)
+        })
+        .expect("runnable");
+        assert!(!super::queue_meta_update(&sched).expect("tick"));
+        assert_eq!(update_metas(), 1, "a running one blocks the tick");
+
+        running.await.1.expect("no growth");
+        assert!(super::queue_meta_update(&sched).expect("tick"));
+        assert_eq!(update_metas(), 2, "a finished one does not");
+    }
 }
 
 #[cfg(test)]
@@ -3082,7 +3159,7 @@ mod mint_identity_tests {
     use hive_jobq::scheduler::Outcome;
     use tracing_subscriber::layer::SubscriberExt as _;
 
-    use super::mint_identity;
+    use super::{SwarmNodeKind, WorkerDeps, mint_identity, run_swarm_node};
 
     /// One event's level and its fields, each by name.
     type Event = (tracing::Level, Vec<(String, String)>);
@@ -3139,4 +3216,62 @@ mod mint_identity_tests {
         assert_eq!(field("agent"), "a/b");
         assert_eq!(field("error"), error);
     }
+
+    /// A node of any kind whose work returns `Failed` logs one warn naming the
+    /// node's id, its kind and the error, from the one place that runs nodes.
+    #[tokio::test]
+    async fn a_failed_node_logs_its_kind_id_and_error() {
+        let events = Events::default();
+        let _guard =
+            tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone()));
+        let sched = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new(
+            hive_jobq::Graph::new(),
+            hive_jobq::resources::ResourceTable::new(),
+        )));
+        let ids = sched
+            .lock()
+            .unwrap()
+            .insert_job(None, |b| {
+                vec![
+                    b.node(SwarmNodeKind::SetAgentWanted {
+                        hive: "h".to_owned(),
+                        agent: "a".to_owned(),
+                    })
+                    .guid(),
+                ]
+            })
+            .expect("inserted");
+        let deps = WorkerDeps {
+            auth: None,
+            forge: super::Late::default(),
+            queue: None,
+            wanted: None,
+            matrix_homeserver: None,
+        };
+        hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
+            run_swarm_node(id, kind, builder, deps)
+        })
+        .expect("runnable")
+        .await
+        .1
+        .expect("no growth declared");
+
+        let got = events.0.lock().expect("unpoisoned").clone();
+        let [(level, fields)] = got.as_slice() else {
+            panic!("one event, got {got:?}");
+        };
+        assert_eq!(*level, tracing::Level::WARN);
+        let field = |name: &str| {
+            fields
+                .iter()
+                .find(|(n, _)| n == name)
+                .map_or_else(|| panic!("no {name} in {fields:?}"), |(_, v)| v.clone())
+        };
+        assert_eq!(field("kind"), "set_agent_wanted");
+        assert_eq!(field("node"), ids[0].get().to_string());
+        assert!(
+            field("error").contains("no swarm queue is configured"),
+            "{fields:?}"
+        );
+    }
 }

swarm-controller/src/startup.rs

@@ -288,10 +288,8 @@ fn spawn_forge_workers(
 ) {
     // `internal/meta` appears, catches up with every config head that moved
     // while this daemon was down, and takes this daemon's rev as the hyperhive
-    // pin when that is newer.
-    if let Err(e) = crate::jobq::queue_meta_update(jobq) {
-        tracing::warn!(error = %format!("{e:#}"), "update_meta: queueing failed");
-    }
+    // pin when that is newer. The timer's first tick is this boot run.
+    crate::jobq::spawn_meta_update_timer(Arc::clone(jobq));
     forge::objects::spawn(Arc::clone(&client));
     if let (Some(bridge), Some(queue)) = (auth, queue) {
         forge::operator_profiles::spawn(Arc::clone(&client), bridge, queue);