swarm-controller: a node kind that asks a hive to deploy an agent

`TriggerDeploy` is the first `SwarmNodeKind` whose effect leaves this
host, so it is also the first to need the queue connection: `WorkerDeps`
grows a `queue` handle, cloned from the status reader whose own doc says
the connection living there is an accident of construction order rather
than a claim that events are a kind of status.

`publish_deploy` publishes and then flushes before reporting `Done`.
`publish` only hands the message to the client's write buffer, so a node
that reported success on that alone would be claiming a delivery it has
no evidence for — the ordering `webhook::announce_knowledge_change`
already documents.

The `WireNode::data` or-pattern did the job it was written for: its
comment says a new variant should fail to compile there rather than
silently render as an agent name, and `TriggerDeploy` is the first node
about an agent *and a hive*. A catch-all would have dropped the hive
from the viewer with nothing to notice it.
This commit is contained in:
atlas 2026-08-30 22:43:31 +02:00 committed by mara
commit 38ccef987e

View file

@ -86,6 +86,16 @@ enum SwarmNodeKind {
/// swarm routes a deploy message to — it belongs on the node that /// swarm routes a deploy message to — it belongs on the node that
/// sends that message, not on this one. /// sends that message, not on this one.
InitAgentConfigRepo { agent: String }, 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 { 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::CreateForgeUser { .. } => "create_forge_user".to_owned(),
SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(), SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(),
SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".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 { fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value {
// Every node in this graph is about exactly one agent, so they all // Spelled out as an or-pattern rather than a catch-all on purpose: a
// render the same and `label()` is what distinguishes them. Spelled // new variant then fails to compile here instead of silently
// out as an or-pattern rather than a catch-all on purpose: a fifth // rendering as an agent name. That gate has now fired once —
// variant then fails to compile here instead of silently rendering // `TriggerDeploy` is the first node that is about an agent *and a
// as an agent name. // hive*, and a catch-all would have dropped the hive from the
// viewer without anyone noticing.
match self { match self {
SwarmNodeKind::CreateIdentity { agent } SwarmNodeKind::CreateIdentity { agent }
| SwarmNodeKind::CreateRepo { agent } | SwarmNodeKind::CreateRepo { agent }
@ -113,6 +125,9 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
| SwarmNodeKind::InitAgentConfigRepo { agent } => { | SwarmNodeKind::InitAgentConfigRepo { agent } => {
serde_json::json!({ "agent": 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 { struct WorkerDeps {
auth: Option<std::sync::Arc<auth::AuthBridge>>, auth: Option<std::sync::Arc<auth::AuthBridge>>,
forge: Option<std::sync::Arc<forge::Client>>, forge: Option<std::sync::Arc<forge::Client>>,
/// 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<async_nats::Client>,
} }
/// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/ /// 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:#}")), 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) (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/ /// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/
/// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node, /// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node,
/// spawn the future that runs + completes it, loop again immediately if /// spawn the future that runs + completes it, loop again immediately if
@ -1151,6 +1214,7 @@ async fn main() -> Result<()> {
let deps = WorkerDeps { let deps = WorkerDeps {
auth: auth.clone(), auth: auth.clone(),
forge: forge_client.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( let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new(