diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 1b4477f6..1ad2af1d 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -86,6 +86,16 @@ enum SwarmNodeKind { /// swarm routes a deploy message to — it belongs on the node that /// sends that message, not on this one. InitAgentConfigRepo { agent: String }, + /// Tell `hive` to rebuild `agent`, by publishing on the swarm's deploy + /// subject. The one node kind whose effect leaves this host. + /// + /// Carries the hive, unlike every variant above — this is the node + /// `InitAgentConfigRepo`'s doc points at when it says the hive belongs + /// on the node that sends the deploy message. The subject is swarm-wide + /// and the addressing rides in the payload; see + /// `swarm_queue_client::DEPLOY_SUBJECT` for why it is not a per-hive + /// family. + TriggerDeploy { hive: String, agent: String }, } impl hive_jobq_wire::WireNode for SwarmNodeKind { @@ -96,15 +106,17 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { SwarmNodeKind::CreateForgeUser { .. } => "create_forge_user".to_owned(), SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(), SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(), + SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(), } } fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value { - // Every node in this graph is about exactly one agent, so they all - // render the same and `label()` is what distinguishes them. Spelled - // out as an or-pattern rather than a catch-all on purpose: a fifth - // variant then fails to compile here instead of silently rendering - // as an agent name. + // Spelled out as an or-pattern rather than a catch-all on purpose: a + // new variant then fails to compile here instead of silently + // rendering as an agent name. That gate has now fired once — + // `TriggerDeploy` is the first node that is about an agent *and a + // hive*, and a catch-all would have dropped the hive from the + // viewer without anyone noticing. match self { SwarmNodeKind::CreateIdentity { agent } | SwarmNodeKind::CreateRepo { agent } @@ -113,6 +125,9 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { | SwarmNodeKind::InitAgentConfigRepo { agent } => { serde_json::json!({ "agent": agent }) } + SwarmNodeKind::TriggerDeploy { hive, agent } => { + serde_json::json!({ "agent": agent, "hive": hive }) + } } } } @@ -140,6 +155,13 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind { struct WorkerDeps { auth: Option>, forge: Option>, + /// The swarm queue connection, for nodes whose effect is a published + /// event rather than an API call. A handle rather than the status + /// reader it is cloned from: a node that announces a deploy has no + /// business reading hive status, and `status.rs`'s own doc says the + /// connection living there is an accident of construction order, not a + /// claim that events are a kind of status. + queue: Option, } /// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/ @@ -226,10 +248,51 @@ async fn run_swarm_node( Err(e) => Outcome::Failed(format!("{e:#}")), }, }, + SwarmNodeKind::TriggerDeploy { hive, agent } => match deps.queue { + None => Outcome::Failed( + "no swarm queue is configured on this host, so no hive can be told to deploy" + .to_owned(), + ), + Some(client) => publish_deploy(&client, &hive, &agent).await, + }, }; (builder, outcome) } +/// Publish one deploy request and report whether it left this process. +/// +/// The flush is not belt-and-braces: `publish` hands the message to the +/// client's write buffer and returns, so a node that reported `Done` on +/// that alone would be claiming a delivery it has no evidence for — the +/// same ordering `webhook::announce_knowledge_change` documents. +async fn publish_deploy( + client: &async_nats::Client, + hive: &str, + agent: &str, +) -> hive_jobq::scheduler::Outcome { + use hive_jobq::scheduler::Outcome; + + let subject = swarm_queue_client::DEPLOY_SUBJECT; + let request = swarm_queue_client::DeployRequest { + hive: hive.to_owned(), + agent: agent.to_owned(), + }; + let payload = match serde_json::to_vec(&request) { + Ok(payload) => payload, + Err(e) => return Outcome::Failed(format!("encoding the deploy request failed: {e}")), + }; + if let Err(e) = client.publish(subject, payload.into()).await { + return Outcome::Failed(format!("publishing to {subject} failed: {e}")); + } + if let Err(e) = client.flush().await { + return Outcome::Failed(format!( + "flushing the deploy event to {subject} failed: {e}" + )); + } + tracing::info!(%subject, %hive, %agent, "swarm jobq: deploy requested"); + Outcome::Done +} + /// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/ /// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node, /// spawn the future that runs + completes it, loop again immediately if @@ -1151,6 +1214,7 @@ async fn main() -> Result<()> { let deps = WorkerDeps { auth: auth.clone(), forge: forge_client.clone(), + queue: status.as_ref().map(|s| s.queue_client()), }; let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new(