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