diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 6f71fc03..714d4850 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -88,7 +88,12 @@ pub fn spawn( // paths independent reconnect state, so one could be serving while // the other was still down. `async_nats::Client` is a handle, so the // clone is cheap. - tokio::spawn(drain_swarm_events(client.clone(), coord, shutdown.clone())); + tokio::spawn(drain_swarm_events( + client.clone(), + coord, + hive.clone(), + shutdown.clone(), + )); let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER); loop { @@ -127,39 +132,40 @@ pub fn spawn( }); } -/// Listen on the swarm's knowledge-event subject and act on what arrives. +/// Listen on the swarm's event subjects and act on what arrives. /// /// The controller decides *what a forge delivery means* and addresses the -/// result here; this end does not know a forge exists. Today the one event is -/// **the knowledge repository changed**, and the response is the pull this -/// daemon already runs at boot. +/// result here; this end does not know a forge exists. Two events today: +/// **the knowledge repository changed**, answered by the pull this daemon +/// already runs at boot, and **deploy this agent**, answered by the same +/// rebuild insert the operator's own verb makes. /// -/// # There is no payload, and that is deliberate +/// Only the deploy event carries a payload, because only it is *addressed*: +/// a swarm-wide subject leaves the addressing nowhere to live but the body. +/// Even then it is a trigger — which hive, which agent — never the config, +/// which git owns. /// -/// The event carries nothing. The webhook handler this replaces read two -/// fields from Forgejo and used neither — both were filters — then ran -/// `git pull`, which re-derives everything from the repository. So it is an -/// edge trigger, and reading a body here would invent a contract nobody owes. +/// # A missed message costs the two events very differently /// -/// # What a missed message costs +/// Core NATS, so delivery is at-most-once. Harmless for knowledge: this +/// daemon pulls at startup regardless, and the webhook it replaced was lost +/// identically when a hive was down. /// -/// Core NATS, so delivery is at-most-once: a hive that is down when the -/// controller publishes never hears it, and its knowledge stays as of its last -/// pull until it next boots. **That is not a regression** — a webhook delivery -/// to a hive that is down is lost identically, and this daemon pulls at startup -/// regardless. `JetStream` would require this end to *publish* to -/// `$JS.API.CONSUMER.CREATE.`, which the callout policy does not grant, -/// so durability would cost a grant on both sides to remove a failure the boot -/// pull already covers. +/// ⚠️ **The deploy event has no such second path** — nothing else would ever +/// tell this hive to build that agent. Its backstop is the hive-side reconcile +/// loop ("hives pull and self-update"); until that exists this is a nudge with +/// no safety net. Not papered over with `JetStream`: durability on one subject +/// looks like a fix while the desired state still lives only in a message. /// /// ⚠️ **A refused subscription is indistinguishable from a quiet one.** NATS /// reports an authorization violation asynchronously on the connection, not as /// an error from `subscribe`, so this task cannot tell "no events published" -/// from "not allowed to hear them". If a hive stops picking up knowledge -/// changes, the server log is the thing that knows why — nothing here will say. +/// from "not allowed to hear them". If a hive stops picking up either event, +/// the server log is the thing that knows why — nothing here will say. async fn drain_swarm_events( client: async_nats::Client, coord: std::sync::Arc, + hive: String, mut shutdown: tokio::sync::watch::Receiver, ) { // One subject for the whole swarm, so this hive's own name never enters @@ -179,7 +185,18 @@ async fn drain_swarm_events( return; } }; - tracing::info!(%subject, "swarm events: listening"); + let deploy_subject = swarm_queue_client::DEPLOY_SUBJECT; + let mut deploy_sub = match client.subscribe(deploy_subject).await { + Ok(sub) => sub, + Err(e) => { + tracing::warn!( + subject = %deploy_subject, error = %e, + "swarm events: subscribe failed; this hive will not hear deploy requests" + ); + return; + } + }; + tracing::info!(%subject, %deploy_subject, "swarm events: listening"); loop { tokio::select! { @@ -197,6 +214,13 @@ async fn drain_swarm_events( tracing::warn!(error = ?e, "swarm events: knowledge pull failed"); } } + msg = deploy_sub.next() => { + let Some(msg) = msg else { + tracing::warn!(subject = %deploy_subject, "swarm events: deploy subscription closed"); + return; + }; + handle_deploy_request(&coord, &hive, &msg.payload); + } _ = shutdown.changed() => { tracing::info!("swarm events: shutdown signal received"); return; @@ -205,6 +229,46 @@ async fn drain_swarm_events( } } +/// Act on one deploy request, if it is addressed to this hive. +/// +/// Every hive receives every message — that is what the swarm-wide subject +/// buys — so *not* being the addressee is the ordinary case and logs at +/// `debug`, not `warn`. A malformed payload is different: the controller and +/// this end share one type, so a decode failure means they disagree about it, +/// and that is worth saying out loud. +fn handle_deploy_request( + coord: &std::sync::Arc, + hive: &str, + payload: &[u8], +) { + let request: swarm_queue_client::DeployRequest = match serde_json::from_slice(payload) { + Ok(request) => request, + Err(e) => { + tracing::warn!(error = %e, "swarm events: undecodable deploy request"); + return; + } + }; + if request.hive != hive { + tracing::debug!(target_hive = %request.hive, agent = %request.agent, "swarm events: deploy for another hive"); + return; + } + // The same insert the operator's own `rebuild` verb makes, relock and + // all: "deploy this agent" means here exactly what it already meant, and + // a swarm-triggered rebuild that quietly did something narrower would be + // a second definition of the word. + let agent = request.agent; + let inserted = coord + .job_queue + .insert_job(|b| crate::job_queue::templates::rebuild(b, &agent, true)); + match inserted { + Ok(_) => { + tracing::info!(%agent, "swarm events: deploy requested by the swarm, rebuild queued"); + coord.emit_rebuild_queue_snapshot(); + } + Err(e) => tracing::warn!(%agent, error = %e, "swarm events: queueing the deploy failed"), + } +} + /// Offer one snapshot: this hive's current readiness, under its own key. async fn publish(client: &async_nats::Client, hive: &str) -> Result<()> { // An unconnected client does not fail a JetStream request, it hangs