From f3e42c93b4abeae8a19d479e895416038c9eaaa4 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 19 Aug 2026 20:16:13 +0200 Subject: [PATCH] swarm-controller: wire the config-PR webhook, not just the poll --- swarm-controller/src/config_pr.rs | 155 ++++++++++++++++++++++++++++-- swarm-controller/src/main.rs | 9 +- swarm-controller/src/webhook.rs | 24 ++--- 3 files changed, 163 insertions(+), 25 deletions(-) diff --git a/swarm-controller/src/config_pr.rs b/swarm-controller/src/config_pr.rs index 4eef46e7..9cdd0590 100644 --- a/swarm-controller/src/config_pr.rs +++ b/swarm-controller/src/config_pr.rs @@ -1,4 +1,5 @@ -//! Swarm-level config-PR status, polled independently of any one hive. +//! Swarm-level config-PR status, kept current by both a webhook nudge and a +//! periodic poll. //! //! `hive-c0re::forge::config_pr_poll` already scans `agent-configs/*` for //! open PRs — but it does that once per hive, to queue that hive's own @@ -7,20 +8,56 @@ //! have an open config PR" that does not depend on which hive currently //! hosts X being reachable. //! -//! This mirrors `hive-c0re`'s poll shape (period, idempotent full rescan) -//! rather than consuming the swarm-wide `pull_request` webhook -//! (`crate::webhook::DeliveryKind::ConfigPr`) that already lands here: that -//! delivery is unparsed today, and building a reliable event path means -//! building this poll as a missed-delivery backstop anyway (the same reason -//! `hive-c0re`'s own poller exists). Shipping the backstop alone first -//! avoids paying for both at once; wiring the webhook as a low-latency nudge -//! on top is a cheap follow-up once this cache is the source of truth. +//! Both paths write the same [`ConfigPrCache`]: +//! +//! - [`spawn`] — a periodic full rescan, mirroring `hive-c0re`'s own poll +//! shape. The backstop: catches anything a missed delivery loses, and is +//! what populates the cache before the first delivery ever arrives. +//! - [`ConfigPrCache::apply_webhook_delivery`] — called from +//! `crate::webhook::post_webhook_forge` on a verified `ConfigPr` delivery. +//! The low-latency path: a PR opening or closing shows up immediately +//! instead of waiting up to [`POLL_INTERVAL`]. +//! +//! Per mara's review call: ship both from the start rather than the poll +//! alone — the eventual swarm-level replacement for `hive-c0re`'s own +//! poll+webhook pair needs both anyway, so building only half here would be +//! work redone rather than work reused. use std::collections::HashMap; use std::sync::{Arc, Mutex}; +use serde::Deserialize; + use crate::forge::{Client, ConfigPrStatus}; +/// The handful of fields this cache needs out of a Forgejo `pull_request` +/// webhook payload — not a full typed mirror of the event (Forgejo's own +/// schema has dozens more), just enough to know which agent, which PR, and +/// whether it's still open. +#[derive(Deserialize)] +pub struct ConfigPrWebhookPayload { + pull_request: WebhookPullRequest, + repository: WebhookRepository, +} + +#[derive(Deserialize)] +struct WebhookPullRequest { + number: u64, + /// Forgejo sends `"open"` or `"closed"` here — merged and + /// closed-without-merging are indistinguishable at this field, but this + /// cache only ever answers "is there an open PR," so the distinction + /// doesn't matter to it. + state: String, + html_url: Option, +} + +#[derive(Deserialize)] +struct WebhookRepository { + /// The config repo's name IS the agent's name — same convention + /// `Client::list_open_config_prs` relies on. + name: String, +} + /// How often to rescan `agent-configs/*`. Matches the interval named in /// `hive-c0re::forge::config_pr_poll`'s own doc comment — same org, same /// staleness tolerance, no reason for the two to disagree. @@ -62,6 +99,53 @@ impl ConfigPrCache { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) = scan; } + + /// Apply one verified `ConfigPr` webhook delivery's raw body to the + /// cache — the low-latency counterpart to [`spawn`]'s periodic rescan. + /// + /// A parse failure is logged and dropped, not propagated: the caller + /// (`crate::webhook::post_webhook_forge`) already returned 200 to + /// Forgejo (HMAC verification, not payload parsing, is what a retry + /// could fix), and the next poll tick reconciles whatever this delivery + /// would have changed — same "poll as backstop" property [`spawn`]'s + /// doc comment describes, just exercised on the failure path instead of + /// the steady-state one. + /// + /// An `open` PR is upserted unconditionally. A `closed` one is removed + /// only if the cached entry's PR number still matches — guards against + /// an out-of-order delivery (a stale `closed` for PR #1 arriving after a + /// newer `opened` for PR #2 on the same repo) wiping out a genuinely + /// current entry. + pub fn apply_webhook_delivery(&self, body: &[u8]) { + let payload: ConfigPrWebhookPayload = match serde_json::from_slice(body) { + Ok(p) => p, + Err(e) => { + tracing::warn!( + error = %e, + "config-pr webhook: payload did not parse, cache unchanged (next poll reconciles)" + ); + return; + } + }; + let mut map = self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if payload.pull_request.state == "open" { + map.insert( + payload.repository.name, + ConfigPrStatus { + pr_number: payload.pull_request.number, + html_url: payload.pull_request.html_url, + }, + ); + } else if map + .get(&payload.repository.name) + .is_some_and(|cached| cached.pr_number == payload.pull_request.number) + { + map.remove(&payload.repository.name); + } + } } /// Build an empty cache and spawn the periodic scan that keeps it current. @@ -89,3 +173,56 @@ pub fn spawn(client: Arc) -> Arc { }); cache } + +#[cfg(test)] +mod tests { + use super::ConfigPrCache; + + fn payload(agent: &str, number: u64, state: &str) -> Vec { + serde_json::json!({ + "pull_request": { "number": number, "state": state, "html_url": "https://forge.example/pr" }, + "repository": { "name": agent }, + }) + .to_string() + .into_bytes() + } + + #[test] + fn an_open_delivery_upserts_the_entry() { + let cache = ConfigPrCache::new(); + cache.apply_webhook_delivery(&payload("damocles", 5, "open")); + let status = cache.get("damocles").expect("entry inserted"); + assert_eq!(status.pr_number, 5); + } + + #[test] + fn a_closed_delivery_removes_a_matching_entry() { + let cache = ConfigPrCache::new(); + cache.apply_webhook_delivery(&payload("damocles", 5, "open")); + cache.apply_webhook_delivery(&payload("damocles", 5, "closed")); + assert!(cache.get("damocles").is_none()); + } + + #[test] + fn a_stale_closed_delivery_does_not_clobber_a_newer_open_pr() { + let cache = ConfigPrCache::new(); + // PR #5 opened, then closed, then a genuinely new PR #6 opens. + cache.apply_webhook_delivery(&payload("damocles", 5, "open")); + cache.apply_webhook_delivery(&payload("damocles", 6, "open")); + // The #5 close event arrives late, after #6 already replaced it. + cache.apply_webhook_delivery(&payload("damocles", 5, "closed")); + let status = cache.get("damocles").expect("PR #6 must survive"); + assert_eq!(status.pr_number, 6); + } + + #[test] + fn a_malformed_payload_leaves_the_cache_unchanged() { + let cache = ConfigPrCache::new(); + cache.apply_webhook_delivery(&payload("damocles", 5, "open")); + cache.apply_webhook_delivery(b"not json"); + let status = cache + .get("damocles") + .expect("prior entry must survive a bad delivery"); + assert_eq!(status.pr_number, 5); + } +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index e5280a7a..b20deba1 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -770,10 +770,11 @@ async fn get_jobq_rollup(State(state): State) -> Json announce_knowledge_change(&state).await, + DeliveryKind::ConfigPr => { + if let Some(cache) = state.config_prs.as_ref() { + cache.apply_webhook_delivery(&body); + } + } } (StatusCode::OK, "ok").into_response()