diff --git a/swarm-controller/src/status.rs b/swarm-controller/src/status.rs index eaff6038..5090cd7a 100644 --- a/swarm-controller/src/status.rs +++ b/swarm-controller/src/status.rs @@ -133,6 +133,23 @@ impl StatusReader { } } + /// A handle on the queue connection this reader holds. + /// + /// The controller has exactly **one** connection to the swarm queue and + /// more than one thing to do with it: status is read out of a KV bucket, + /// swarm events are published on a subject. Handing out a clone is cheap — + /// `async_nats::Client` is a handle, not a socket — and is strictly better + /// than opening a second connection, which would double the auth-callout + /// traffic and give the two paths independent reconnect state, so one could + /// be serving while the other was still down. + /// + /// That this lives on the *status* reader is an accident of who constructed + /// the connection first, not a claim that events are a kind of status. + #[must_use] + pub fn queue_client(&self) -> async_nats::Client { + self.client.clone() + } + /// Reads [`STALE_AFTER_ENV`], falling back to /// [`DEFAULT_STALE_AFTER`]. A zero or unparseable value takes the /// default rather than failing startup — same rule as `load_hives`: diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index 5be60268..5ad47f73 100644 --- a/swarm-controller/src/webhook.rs +++ b/swarm-controller/src/webhook.rs @@ -365,9 +365,74 @@ pub(super) async fn post_webhook_forge( bytes = body.len(), "webhook: verified delivery" ); + + match kind { + DeliveryKind::Knowledge => announce_knowledge_change(&state).await, + // Deploy coordination is a separate concern with its own issue: a + // hive does not want to hear that a config PR was opened, it wants + // to be told when to rebuild from main, and that is a decision the + // controller makes after a merge rather than a relay of this + // delivery. Logged above and deliberately not forwarded. + DeliveryKind::ConfigPr => {} + } + (StatusCode::OK, "ok").into_response() } +/// Tell every hive in the swarm that the knowledge repository changed. +/// +/// The event carries **no payload**, because there is nothing to carry: the +/// hive-side handler this replaces read two fields from Forgejo's webhook and +/// used neither — both were filters — and then ran `git pull`, which re-derives +/// everything from the repository itself. So what crosses the queue is an edge +/// trigger, and adding fields to it would invent a contract nobody reads. +/// +/// One subject per hive rather than one shared subject, so the swarm's callout +/// policy can express "this hive may hear its own events" at all; a subject +/// with no hive component is the same subject for everyone. +/// +/// # Failure +/// +/// Returns nothing and fails soft. A missed announcement costs a hive stale +/// knowledge until its next boot pull — the same cost as a webhook delivery to +/// a hive that happened to be down, which is what this replaces. +/// +/// ⚠️ A **permission** failure cannot be observed here. `publish` hands the +/// message to the connection's buffer, and a NATS authorization violation is +/// reported asynchronously on the connection rather than as an error on this +/// call — it reaches a client as a timeout, or as nothing at all. The `flush` +/// below therefore proves the bytes left this process, and nothing more; if +/// hives stop hearing events, the server log is the place that knows why. +async fn announce_knowledge_change(state: &AppState) { + let Some(status) = state.status.as_ref() else { + // Verified, accepted, and dropped. Worth a warning rather than + // silence: the forge will report a 200 and nobody would otherwise + // learn that the delivery reached a controller with nowhere to put it. + tracing::warn!( + "webhook: knowledge delivery verified but no swarm queue is \ + configured; no hive will be told" + ); + return; + }; + let client = status.queue_client(); + + for hive in state.hives.iter() { + let subject = swarm_queue_client::events::knowledge(&hive.name); + if let Err(e) = client.publish(subject.clone(), Vec::new().into()).await { + tracing::warn!( + hive = %hive.name, %subject, error = %e, + "webhook: publishing the knowledge event failed" + ); + } else { + tracing::info!(hive = %hive.name, %subject, "webhook: knowledge event published"); + } + } + + if let Err(e) = client.flush().await { + tracing::warn!(error = %e, "webhook: flushing knowledge events failed"); + } +} + #[cfg(test)] mod tests { use super::{DeliveryKind, Refusal, load_or_generate_at, secret_path_from, verify};