From 89050ef34bb336bcc9f0dba579cef0b9dd345959 Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 19 Aug 2026 19:06:52 +0200 Subject: [PATCH] feat(#3255): hives subscribe to their own knowledge event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A hive learned the knowledge repository had changed only by registering its own forge webhook. This subscribes to the per-hive subject the controller now publishes on and calls the pull this daemon already runs at boot. Shares the hive's ONE queue connection rather than opening a second: a second connect would double the auth-callout traffic against authelia and give the two paths independent reconnect state, so one could be serving while the other was still down. Same argument as the controller side. No payload is read, because there is none to read — the webhook handler this replaces took two fields from Forgejo and used neither, then ran `git pull`, which re-derives everything from the repository. At-most-once, and that is not a regression: a webhook delivery to a hive that is down is lost identically today, and the boot pull covers it. JetStream would require this end to publish to `$JS.API.CONSUMER.CREATE.`, which the callout policy does not grant, so durability would cost grants on both sides to remove a failure the boot pull already handles. ⚠️ Documented at the call site rather than left implicit: a refused subscription is indistinguishable from a quiet one, because NATS reports authorization violations asynchronously on the connection. If hives stop hearing events, the server log is the thing that knows. futures-util comes from the workspace (same version swarm-controller already uses), not a new dependency version. --- Cargo.lock | 1 + hive-c0re/Cargo.toml | 3 ++ hive-c0re/src/main.rs | 2 +- hive-c0re/src/swarm_status.rs | 97 ++++++++++++++++++++++++++++++++++- 4 files changed, 101 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d95ae5b5..abcdcced 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1673,6 +1673,7 @@ dependencies = [ "clap-markdown", "clap_complete", "forgejo-api", + "futures-util", "hive-agent-sock", "hive-core-agent-sock", "hive-host-sock", diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 6ffc9549..68440cc2 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -8,6 +8,9 @@ readme = "README.md" workspace = true [dependencies] +# For `StreamExt::next` on the swarm-event subscription in `swarm_status`. +# Workspace-level, same version swarm-controller already uses — not a second copy. +futures-util.workspace = true anyhow.workspace = true # Named directly only for the client type the swarm status publisher passes # around; the connect itself lives in `swarm-queue-client` below. diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 9efe1d51..5e5d6f06 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -483,7 +483,7 @@ async fn cmd_serve( // A no-op on a standalone hive (no queue env, logged once) — see // swarm_status, which owns the whole task including its own decision // not to start. - swarm_status::spawn(coord.shutdown_rx()); + swarm_status::spawn(std::sync::Arc::clone(&coord), coord.shutdown_rx()); // Per-agent events.sqlite + bash-tasks file cleanup now runs // agent-side in the harness (`hive_agent::vacuum`): the files are // agent-owned, so host-side deletes hit PermissionDenied / readonly-db diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 822b8564..f4b66698 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -31,6 +31,9 @@ use std::time::Duration; use anyhow::{Context, Result}; +// `Subscriber` is a `Stream`, so reading the next event needs the extension +// trait — there is no inherent `next()` on it. +use futures_util::StreamExt as _; use crate::stats::sweep_health::{self, SweepHealth}; @@ -62,7 +65,10 @@ const FAILURES_BEFORE_BANNER: u32 = 3; /// makes it a hard error; it is bannered here rather than swallowed, /// because the failure it otherwise produces is a hive that looks fine /// and silently never reports. -pub fn spawn(mut shutdown: tokio::sync::watch::Receiver) { +pub fn spawn( + coord: std::sync::Arc, + mut shutdown: tokio::sync::watch::Receiver, +) { let cfg = match swarm_queue_client::QueueConfig::from_env(ENV_PREFIX) { Ok(Some(cfg)) => cfg, Ok(None) => { @@ -123,6 +129,19 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver) { } }; + // The hive's ONE queue connection, now serving both directions: + // status goes up, swarm events come down. A second `connect` would + // double the auth-callout traffic against authelia and give the two + // 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(), + hive.clone(), + coord, + shutdown.clone(), + )); + let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER); loop { match publish(&client, &hive).await { @@ -160,6 +179,82 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver) { }); } +/// Listen on this hive's swarm-event subject 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. +/// +/// # There is no payload, and that is deliberate +/// +/// 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. +/// +/// # What a missed message costs +/// +/// 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. +/// +/// ⚠️ **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. +async fn drain_swarm_events( + client: async_nats::Client, + hive: String, + coord: std::sync::Arc, + mut shutdown: tokio::sync::watch::Receiver, +) { + let subject = swarm_queue_client::events::knowledge(&hive); + let mut sub = match client.subscribe(subject.clone()).await { + Ok(sub) => sub, + Err(e) => { + // Warn rather than a boot banner: the hive is fully functional + // without this, it just falls back to learning about knowledge + // changes at its next boot. + tracing::warn!( + %subject, error = %e, + "swarm events: subscribe failed; this hive will not hear knowledge changes" + ); + return; + } + }; + tracing::info!(%subject, "swarm events: listening"); + + loop { + tokio::select! { + msg = sub.next() => { + if msg.is_none() { + // The subscription ended — the connection went away for + // good. Returning is right: `async-nats` reconnects + // underneath a live subscription, so a closed stream is + // not a blip this should spin on. + tracing::warn!(%subject, "swarm events: subscription closed"); + return; + } + tracing::info!(%subject, "swarm events: knowledge change announced, pulling"); + if let Err(e) = crate::workers::knowledge::pull(&coord).await { + tracing::warn!(error = ?e, "swarm events: knowledge pull failed"); + } + } + _ = shutdown.changed() => { + tracing::info!("swarm events: shutdown signal received"); + return; + } + } + } +} + /// 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