From 7519d9b904c0ca2e48a328a1a7be206d968e3f79 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 30 Aug 2026 22:51:55 +0200 Subject: [PATCH] hive-c0re: act on a deploy request addressed to this hive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The second subject on the connection this task already drains. Every hive receives every message — that is what a swarm-wide subject buys — so not being the addressee is the ordinary case and logs at `debug`. A payload that will not decode is not: both ends share one type, so a decode failure means they disagree about it. The rebuild is the same insert the operator's own `rebuild` verb makes, relock and all. "Deploy this agent" already means something here, and a swarm-triggered rebuild that quietly did something narrower would be a second definition of the word. ⚠️ The at-most-once argument in this function's docs does NOT transfer to the new subject, and the docs now say so. A missed knowledge event is repaired by the pull this daemon does at startup regardless; a missed deploy event has no second path — nothing else would ever tell this hive to build that agent. Closing that is the hive-side reconcile loop the issue's other half calls for; until it exists this is a nudge with no safety net. Not papered over with `JetStream` here: durability on one subject would look like a fix while the desired state still lived only in a message. Swept the surrounding prose rather than only the lines I touched. Two sections had gone quietly false: the summary said this listens on "the knowledge-event subject" and named one event, and a whole section argued "there is no payload, and that is deliberate" — true of the knowledge event and the opposite of true for a deploy request, which is addressed and has nowhere but the body to say so. --- hive-c0re/src/swarm_status.rs | 108 +++++++++++++++++++++++++++------- 1 file changed, 86 insertions(+), 22 deletions(-) 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