Compare commits

...
Author SHA1 Message Date
atlas
3c390f8132 docs(#3255): the announce doc described the design the rework replaced
The function's own doc comment still said one subject per hive, four
lines above an inline comment correctly saying one publish for all of
them. Left behind by the single-writer rework.

Swept the rest of the touched files for the old design's vocabulary
rather than fixing only the line argus quoted: every other per-hive
mention is about webhook registration or the callout policy's
extra-subject template, both of which are still per-hive and still
accurate.
2026-08-19 21:05:52 +02:00
atlas
33958d78ae refactor(#3255): a constant, not a module with an essay
Review call: 46 lines of documentation around a single constant, part of
it already stale. The worst paragraph explained why the earlier per-hive
shape had been justified wrongly — history of a design that never
shipped, written into the file within an hour of that design being
dropped. A file is not a changelog; why it was wrong belongs in the PR.

The constant moves to lib.rs beside the status bucket name, keeping only
the rationale that stays true: three crates must agree on the string, and
the one that agrees hardest speaks neither jetstream nor kv, which is why
it cannot sit behind a feature gate.

status earns a module of its own because it holds a bucket name AND the
functions that open it. This held a constant.
2026-08-19 21:05:52 +02:00
atlas
9b939f4626 refactor(#3255): one knowledge subject, single writer and many readers
Review call: the event was addressed per hive — `$SWARM.events.<hive>.knowledge`,
published in a loop over the roster, granted through a wildcard. It does not
need to be. The payload is empty and the event means the same thing to every
hive, so one publish to one subject delivers exactly what N publishes to N
subjects did, and core NATS already fans out to whoever is subscribed. A hive
that was down misses it either way and reconciles on its next periodic pull.

That deletes rather than reshuffles: the roster loop, the wildcard, and the
shared subject-building function whose entire purpose was keeping the grant and
the publish from drifting apart. With one literal there is nothing to disagree
about.

The per-hive shape was justified by the callout policy's rule that an extra
subject must contain the hive name. That rule governs `extra_hive_subjects` —
what a HIVE may publish. This subject lives in the controller's reader grant,
which the rule does not constrain, so a real rule was carried across into a
decision it had no authority over.

Knowledge becomes its own category rather than a leaf under a general event
namespace, since a namespace shaped for events that do not exist yet is a
decision made before there is anything to decide from. The empty config-PR match
arm goes with it: an arm with no body claims this is where the deploy path is
handled, and it is not.

The deny test stays and matters more, not less: with one shared subject a forged
event would reach the whole swarm where a per-hive one reached a single hive.
2026-08-19 21:05:52 +02:00
atlas
84125b0806 docs(#3255): correct the reaper invariant the swarm webhook test guards
The test's doc named knowledge::ensure_webhook as one of two hive-side
reapers. That function is gone; its replacement matches the full URL and
cannot touch another hive's hook.

The assertion arm stays. The hazard is not what this repository's source
says, it is what is deployed — a hive still running the previous version
reaps by suffix until it is upgraded, so the arm guards the transition
rather than a current code path. Recorded when to drop it.
2026-08-19 21:05:52 +02:00
atlas
d2a550e685 feat(#3255): hives stop owning the knowledge webhook, and clean up their own
A webhook has exactly one target URL, so every hive registering one
against the shared internal/knowledge repository was last-writer-wins
rather than idempotent: all but the most recent silently stopped
receiving deliveries. The swarm controller holds the single registration
and now addresses an event to each hive over the queue instead.

This is a migration, not a deletion. Not registering any more fixes
nothing on a hive that has already run — the hook it created persists on
the forge, so the contention would survive on exactly the deployments
that have it while fresh installs looked fixed. The hive that created a
hook removes it.

It removes only its OWN, matched on the full URL rather than the
/webhook/knowledge suffix. A hook with that suffix and a different base
belongs to another hive, possibly one not yet upgraded, and deleting it
would break that hive's knowledge sync until it caught up. Reaping a
neighbour's registration is the behaviour being removed here; doing it
while fixing it would only invert the direction.

The predecessor did reap by suffix, to clear loopback hooks left by an
older single-hive layout. That was safe when a hive was alone on its
forge and is not safe now. The hive-side registrars also acted as reapers
of hooks under their own path, which is why the swarm hook lives under
/webhook/forge/; removing this registrar removes that reaper too.
Intended, and stated because no reviewer would infer it from the diff.

The receive endpoint goes with it. A live HMAC-verified
/webhook/knowledge that nothing can legitimately reach would tell the
next reader that this is how a hive learns about knowledge changes.

Docs move in the same commit: docs/swarm/README.md said two hooks exist
per swarm-wide repo and neither should be deleted, which is now true for
agent-configs and wrong for internal/knowledge — a half-correct
description being worse than an uncorrected one.
2026-08-19 21:05:52 +02:00
atlas
89050ef34b feat(#3255): hives subscribe to their own knowledge event
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.<stream>`, 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.
2026-08-19 21:05:52 +02:00
atlas
bac4a8b6a1 feat(#3255): announce a knowledge change to every hive on the queue
The controller verified a knowledge delivery, logged it, and returned OK.
Nothing downstream ever heard about it, so a hive learned the repository
had changed only by registering its own webhook — which is the
last-writer-wins contention this issue is about.

The event carries no payload. The hive-side handler this replaces reads
two fields from Forgejo's push webhook and uses neither — both are
filters — then runs `git pull`, which re-derives everything from the
repository. What crosses the queue is an edge trigger, and fields would
invent a contract nobody reads.

One subject per hive, so the callout policy can express "this hive may
hear its own events" at all; a subject with no hive component is the same
subject for every hive.

`ConfigPr` deliveries are deliberately not forwarded. A hive does not
want to hear that a config PR opened — it wants to be told when to
rebuild from main, which the controller decides after a merge rather than
by relaying this delivery. That is deploy coordination's job, and the
empty arm is there so the omission reads as scoped rather than forgotten.

Fails soft: a missed announcement costs a hive stale knowledge until its
next boot pull, which is the same cost as a webhook delivery to a hive
that was down — what this replaces. A permission failure cannot be
observed at the call site (a NATS authorization violation is reported
asynchronously on the connection, reaching a client as a timeout or as
nothing), so the doc says the flush proves only that the bytes left this
process and points at the server log.
2026-08-19 21:05:52 +02:00
atlas
8baf1899d8 feat(#3255): name the swarm event subjects, and let the controller publish them
The controller could not emit an event at all: a reader's grant is
`reader_subjects()`, which is `$JS.API.*` only, so a publish to any event
subject would be refused — and a NATS refusal reaches the client as a
timeout, so the visible symptom would have been a hive that never hears
about a change, with nothing in any log naming a permission.

Adds `swarm_queue_client::events`, following `status::BUCKET`: three
crates must agree on these strings (the controller publishes, a hive
subscribes, the callout responder decides whether the publish is
permitted), and a literal repeated across crates is an agreement nothing
checks. The responder speaks neither jetstream nor kv, so the module is
unconditional and carries no NATS types, exactly as the bucket name is.

The grant takes the wildcard form from the same function the publisher
calls, so the two cannot drift; a separate wildcard constant would have
re-created the disagreement this module exists to prevent.

Tests pin that a reader gets the subject and that a hive does NOT — a
hive able to publish here could tell a neighbour the knowledge repo
changed when it had not, which is an unauthenticated write into someone
else's control path. That one asserts on the subject root rather than a
rendered subject, so a future event leaf fails it too instead of passing
because the test only knew about `knowledge`.

Both assertions mutation-tested: removing the grant fails the reader
test, granting a hive the subject fails the denial test, each on its own
assertion line, and the unmutated tree is green.
2026-08-19 21:05:52 +02:00
15 changed files with 375 additions and 241 deletions

1
Cargo.lock generated
View file

@ -1673,6 +1673,7 @@ dependencies = [
"clap-markdown",
"clap_complete",
"forgejo-api",
"futures-util",
"hive-agent-sock",
"hive-core-agent-sock",
"hive-host-sock",

View file

@ -38,18 +38,23 @@ create a new document.
hive-c0re maintains the local clone at
`/var/lib/hyperhive/knowledge` via two paths:
1. **Forgejo push webhook**`ensure_webhook` registers a push
hook on `internal/knowledge` at startup pointing at
`https://<hive_domain>/webhook/knowledge` (routed through the
gateway, avoiding the Forgejo SSRF guard that blocks loopback
delivery). On any push to main (including merge commits) hive-c0re
runs `git pull` so agents see the new content on their next turn.
The endpoint is protected by an auto-generated HMAC secret that
hive-c0re verifies on every delivery.
1. **Swarm event** — the swarm controller holds the single push hook on
`internal/knowledge` (see `docs/swarm/README.md` § Swarm-wide forge
webhooks). On any push to main, including merge commits, it sends an
event to every hive over the swarm queue and each hive runs `git
pull`, so agents see the new content on their next turn.
On a swarm, a **second** hook on the same repo points at the
swarm controller — see `docs/swarm/README.md` § Swarm-wide forge
webhooks. Both are expected; neither should be deleted.
A hive that is offline when the event is sent does not get it on
reconnect — the periodic pull below is what closes that gap. So one
hive briefly showing older `/knowledge` content than another is
expected, and resolves by itself within the fallback interval.
**Do not add a per-hive hook.** A webhook has exactly one target
URL, so a second registration against the same repo does not add a
recipient — it takes delivery away from whoever registered first.
Earlier versions had each hive register its own; hive-c0re now
removes its own leftover at startup, so no operator step is needed
to migrate.
2. **Periodic pull** — a background task in `hive-c0re::main`
pulls on a fixed cadence as a fallback (webhook missed, c0re

View file

@ -356,18 +356,23 @@ itself — a `push` hook on `internal/knowledge` and a `pull_request` hook
on the `agent-configs` org, both under
`https://<swarm.ui.domain>/webhook/forge/`.
**Two hooks exist per swarm-wide repo: each hive's own, plus the
controller's.** Both are expected — **do not delete either.** Removing a
hive's hook stops that hive acting on knowledge pushes and config PRs;
removing the controller's just gets recreated on its next start.
The controller **interprets** a delivery and sends hives a specific
message — *the knowledge repo changed*, *deploy agent `foo` at rev
`abc123`* — rather than forwarding forge payloads for each hive to
re-derive. Approval happens once, at the swarm level: a hive receives a
decision, not an event to adjudicate. Today the controller logs each
verified delivery and sends nothing, because the swarm→hive channel does
not exist yet; the hive-side hooks are what act in the meantime.
decision, not an event to adjudicate.
**`internal/knowledge` is on that path.** The controller's is the only
hook on it: hives no longer register their own, and each removes its
leftover at startup. A webhook has exactly one target URL, so per-hive
registration never added a recipient — it took delivery away from
whichever hive registered before it.
**The `agent-configs` org is not yet.** Each hive still registers its own
`pull_request` hook there, so that repo has two — the hive's and the
controller's — and **both are expected; do not delete either.** Removing
a hive's stops it acting on config PRs; removing the controller's just
gets recreated on its next start.
Nothing to configure. The hooks are registered only when this host also
serves the swarm UI vhost — that is what publishes the endpoint, and a

View file

@ -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.

View file

@ -184,7 +184,6 @@ pub async fn serve(
.routes(routes!(schedules::post_schedule_resume))
.routes(routes!(schedules::post_schedule_fire_now))
.routes(routes!(schedules::post_rebuild_queue_cancel))
.routes(routes!(webhook::post_webhook_knowledge))
.routes(routes!(webhook::post_webhook_config_pr))
.routes(routes!(approvals::post_approve))
.routes(routes!(approvals::post_deny))
@ -430,7 +429,6 @@ mod router_build_probe {
.routes(routes!(schedules::post_schedule_resume))
.routes(routes!(schedules::post_schedule_fire_now))
.routes(routes!(schedules::post_rebuild_queue_cancel))
.routes(routes!(webhook::post_webhook_knowledge))
.routes(routes!(webhook::post_webhook_config_pr))
.routes(routes!(approvals::post_approve))
.routes(routes!(approvals::post_deny))

View file

@ -1,12 +1,18 @@
//! Forgejo webhook endpoints.
//!
//! - **`/webhook/knowledge`** — push events on `internal/knowledge` trigger a
//! `git pull` on the local clone so agents see up-to-date docs.
//! - **`/webhook/config-pr`** — `pull_request` events on any `agent-configs/*`
//! repo queue a [`hive_sh4re::approvals::ApprovalKind::MergeConfigPr`] approval row
//! so the operator can review + approve the merge from the dashboard.
//!
//! Both endpoints are reached via the gateway (HTTPS, public domain URL) so
//! There was a second endpoint here, `/webhook/knowledge`, which pulled the
//! local `internal/knowledge` clone on push. It is gone along with the
//! per-hive registration that fed it: a webhook has exactly one target URL,
//! so every hive registering one against the shared repository was
//! last-writer-wins. The swarm controller now holds the single registration
//! and addresses an event to each hive over the queue, which
//! [`crate::workers::knowledge`] documents.
//!
//! The endpoint is reached via the gateway (HTTPS, public domain URL) so
//! Forgejo's SSRF guard does not block delivery. Each delivery is verified
//! against the `X-Hub-Signature-256` HMAC header Forgejo attaches; the
//! shared secret is auto-generated at startup and persisted to
@ -42,104 +48,6 @@ fn verify_hmac(state: &AppState, headers: &HeaderMap, body: &Bytes) -> Result<()
crate::webhook_secret::verify_signature(secret, body, sig).map_err(|e| e.to_string())
}
// ── knowledge webhook ──────────────────────────────────────────────────────────
/// Minimal Forgejo push-webhook payload — only the fields we care about.
#[derive(Deserialize)]
pub(super) struct PushWebhookPayload {
#[serde(rename = "ref")]
git_ref: Option<String>,
repository: Option<PushWebhookRepo>,
}
#[derive(Deserialize)]
pub(super) struct PushWebhookRepo {
full_name: Option<String>,
}
/// POST `/webhook/knowledge` — Forgejo push webhook for
/// `internal/knowledge`.
///
/// Runs `git pull` on the local clone so agents see up-to-date documents
/// on their next turn.
///
/// Expected Forgejo webhook configuration:
/// - URL: `https://<HYPERHIVE_HIVE_DOMAIN>/webhook/knowledge`
/// - Content type: `application/json`
/// - Event: "Push" (fires on merge commits to main as well)
/// - Secret: auto-generated HMAC key (see [`crate::webhook_secret`])
///
/// The gateway routes `/webhook/` → hive-c0re; the HMAC secret protects
/// the endpoint from unauthenticated callers.
#[utoipa::path(
post,
path = "/webhook/knowledge",
request_body(
content = String,
content_type = "application/json",
description = "Forgejo push-webhook payload, taken as raw bytes \
(not a typed extractor) so HMAC verification runs \
over the exact wire bytes before any JSON parsing"
),
responses(
(status = 200, description = "processed (pull triggered or ignored)", body = String),
(status = 400, description = "invalid JSON payload"),
(status = 401, description = "bad or missing HMAC signature"),
(status = 503, description = "HMAC secret unavailable at startup"),
),
tag = "webhook"
)]
pub(super) async fn post_webhook_knowledge(
State(state): State<AppState>,
headers: HeaderMap,
body: Bytes,
) -> Response {
if let Err(e) = verify_hmac(&state, &headers, &body) {
tracing::warn!("webhook/knowledge: HMAC verification failed: {e}");
let status = if e.contains("unavailable") {
StatusCode::SERVICE_UNAVAILABLE
} else {
StatusCode::UNAUTHORIZED
};
return (status, e).into_response();
}
let payload = match serde_json::from_slice::<PushWebhookPayload>(&body) {
Ok(p) => p,
Err(e) => {
tracing::warn!("webhook/knowledge: JSON parse error: {e}");
return (StatusCode::BAD_REQUEST, "invalid JSON").into_response();
}
};
let expected_repo = format!("{}/{}", crate::knowledge::ORG, crate::knowledge::REPO);
let full_name = payload
.repository
.as_ref()
.and_then(|r| r.full_name.as_deref())
.unwrap_or("");
if full_name != expected_repo {
tracing::debug!(
full_name,
"webhook/knowledge: ignoring push from unexpected repo"
);
return (StatusCode::OK, "ignored").into_response();
}
let git_ref = payload.git_ref.as_deref().unwrap_or("");
if git_ref != "refs/heads/main" {
tracing::debug!(git_ref, "webhook/knowledge: ignoring non-main push");
return (StatusCode::OK, "ignored").into_response();
}
tracing::info!("webhook/knowledge: pull triggered by push to {expected_repo}");
let coord = state.coord.clone();
tokio::spawn(async move {
if let Err(e) = crate::knowledge::pull(&coord).await {
tracing::warn!(error = ?e, "webhook/knowledge: pull failed");
}
});
(StatusCode::OK, "ok").into_response()
}
// ── config-PR webhook ──────────────────────────────────────────────────────────
/// Minimal Forgejo `pull_request`-webhook payload.

View file

@ -483,8 +483,7 @@ pub async fn ensure_all() {
/// existing hooks first and skips creation when one is already targeting the
/// correct URL.
///
/// `hive_domain` is the public domain name of the hive (e.g.
/// `pr1ma.darkest.space`); the webhook URL is
/// `hive_domain` is the public domain name of the hive; the webhook URL is
/// `https://<hive_domain>/webhook/config-pr` (routed through the gateway,
/// avoiding the Forgejo SSRF guard that blocks loopback delivery).
///
@ -495,8 +494,9 @@ pub async fn ensure_all() {
/// An org-level hook covers every repo in `agent-configs` automatically,
/// so no per-repo setup is needed as new agents are provisioned.
///
/// Called at startup alongside `knowledge::ensure_webhook`. No-op when the
/// core token is absent (forge not yet provisioned).
/// Called at startup beside `knowledge::remove_webhook`, its opposite: that
/// repo's one hook is the controller's now, this one has not moved yet. No-op
/// when the core token is absent (forge not yet provisioned).
///
/// # Errors
///

View file

@ -179,10 +179,18 @@ async fn run_matrix_sweep() -> Result<()> {
}
}
/// Boot-time Forgejo webhook registration as a DAG node — see
/// Boot-time Forgejo webhook management as a DAG node — see
/// [`NodeKind::WebhookRegister`]. Mirrors the guard chain the
/// `tokio::spawn` block it replaced used: no-op (not an error) when the
/// HMAC secret, core token, or hive domain aren't available yet.
///
/// The node now does one of each: it still registers the config-PR hook,
/// and it *removes* the knowledge one. A knowledge push is delivered to
/// the swarm controller, which addresses an event to each hive over the
/// queue — so a hive holding its own registration is holding a shared
/// resource only one party can own. The removal runs every boot rather
/// than behind a marker because it is already idempotent: it is a no-op
/// the moment the hook is gone.
async fn run_webhook_register() -> Result<()> {
let Ok(webhook_secret) = crate::webhook_secret::load_or_generate() else {
tracing::debug!("webhook secret unavailable; skipping hook registration");
@ -198,10 +206,8 @@ async fn run_webhook_register() -> Result<()> {
tracing::debug!("HYPERHIVE_HIVE_DOMAIN unset; skipping webhook registration");
return Ok(());
};
if let Err(e) =
crate::workers::knowledge::ensure_webhook(&token, &domain, &webhook_secret).await
{
tracing::warn!(error = ?e, "knowledge: ensure_webhook failed");
if let Err(e) = crate::workers::knowledge::remove_webhook(&token, &domain).await {
tracing::warn!(error = ?e, "knowledge: remove_webhook failed");
}
if let Err(e) = crate::forge::ensure_config_pr_webhook(&token, &domain, &webhook_secret).await {
tracing::warn!(error = ?e, "forge: ensure_config_pr_webhook failed");

View file

@ -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

View file

@ -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<bool>) {
pub fn spawn(
coord: std::sync::Arc<crate::coordinator::Coordinator>,
mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
let cfg = match swarm_queue_client::QueueConfig::from_env(ENV_PREFIX) {
Ok(Some(cfg)) => cfg,
Ok(None) => {
@ -123,6 +129,14 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
}
};
// 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(), coord, shutdown.clone()));
let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER);
loop {
match publish(&client, &hive).await {
@ -160,6 +174,84 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
});
}
/// Listen on the swarm's knowledge-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.<stream>`, 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,
coord: std::sync::Arc<crate::coordinator::Coordinator>,
mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
// One subject for the whole swarm, so this hive's own name never enters
// it: the controller publishes once and core NATS fans out to whoever is
// subscribed.
let subject = swarm_queue_client::KNOWLEDGE_SUBJECT;
let mut sub = match client.subscribe(subject).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

View file

@ -6,15 +6,18 @@
//! contribute by forking the repo and opening PRs — they never write
//! to the bind-mounted path inside the container.
//!
//! hive-c0re maintains the local clone. A Forgejo webhook notifies it
//! on push to main so agents always see an up-to-date snapshot. The
//! webhook is auto-created by [`ensure_webhook`] at startup. A
//! hive-c0re maintains the local clone. It learns that the repository
//! moved from the **swarm controller**, which owns the one Forgejo
//! webhook and addresses an event to each hive over the queue; a
//! periodic pull in `main.rs` provides a fallback cadence.
use std::collections::BTreeMap;
//!
//! A hive used to register that webhook itself, pointing at its own
//! `/webhook/knowledge`. A webhook has exactly one target URL, so with
//! more than one hive that was last-writer-wins rather than idempotent —
//! every hive but the most recent silently stopped receiving deliveries.
//! [`remove_webhook`] is the migration off it.
use anyhow::{Context, Result};
use forgejo_api::structs::{CreateHookOption, CreateHookOptionConfig, CreateHookOptionType};
use crate::coordinator::Coordinator;
use crate::forge::{core_auth_header, forge_git_url};
@ -142,103 +145,71 @@ async fn seed_readme(core_token: &str) -> Result<()> {
}
}
/// Ensure a Forgejo push webhook for `internal/knowledge` exists and
/// points at hive-c0re's `/webhook/knowledge` endpoint. Idempotent —
/// lists existing hooks first and skips creation when one is already
/// targeting the correct URL.
/// Delete this hive's own `internal/knowledge` push webhook if it is
/// still registered, so the swarm controller is the only party holding
/// one.
///
/// `hive_domain` is the public domain name of the hive; the webhook URL is
/// `https://<hive_domain>/webhook/knowledge` (routed through the gateway,
/// avoiding the Forgejo SSRF guard that blocks loopback delivery).
/// # Why this is a migration and not just a deletion
///
/// `webhook_secret` is the HMAC secret Forgejo will attach as
/// `X-Hub-Signature-256` on each delivery; hive-c0re verifies this header
/// in the dashboard webhook handler (`post_webhook_knowledge`).
/// Not registering any more fixes nothing on a hive that has already
/// run: the hook it created persists on the forge, so the contention
/// this removes would survive on exactly the deployments that have it
/// while fresh installs looked fixed. The hive that created a hook is
/// the one that removes it.
///
/// Called at startup alongside [`ensure_local_clone`]. No-op when the
/// core token is absent (forge not yet provisioned).
pub async fn ensure_webhook(
core_token: &str,
hive_domain: &str,
webhook_secret: &str,
) -> Result<()> {
/// # It removes only its OWN hook, never a neighbour's
///
/// The match is the full URL, not the `/webhook/knowledge` suffix. A
/// hook with that suffix and a different base belongs to *another hive* —
/// one that may not have been upgraded yet — and deleting it would break
/// its knowledge sync until it was. Reaping a neighbour's registration is
/// the very behaviour this issue is about; doing it in the name of fixing
/// it would just invert the direction.
///
/// (The predecessor did reap by suffix, to clear loopback hooks left by
/// an older single-hive layout. That was safe when a hive was alone on
/// its forge and is not safe now.)
///
/// A listing failure is an error rather than a silent skip: there is no
/// create attempt left to fall through to, so swallowing it would leave
/// the hook in place with nothing said. The caller logs and continues —
/// boot does not depend on this.
pub async fn remove_webhook(core_token: &str, hive_domain: &str) -> Result<()> {
// The typed client carries no per-request timeout, so each call is
// wrapped in one: this runs as a detached startup task, and a forge
// that accepts connections but never answers would otherwise hang
// it forever (and the hourly pull fallback masks the missing hook).
// that accepts connections but never answers would otherwise hang it
// forever.
const HTTP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
let target_url = format!("https://{hive_domain}/webhook/knowledge");
let own_url = format!("https://{hive_domain}/webhook/knowledge");
let client = crate::forge::api(core_token)?;
// List existing hooks — skip creation if ours is already there.
// Best-effort like the raw-HTTP predecessor: a listing failure
// falls through to the create attempt.
let listed = tokio::time::timeout(HTTP_TIMEOUT, client.repo_list_hooks(ORG, REPO).all())
.await
.map_err(anyhow::Error::from)
.and_then(|r| r.map_err(anyhow::Error::from));
match listed {
Ok(hooks) => {
let already_exists = hooks.iter().any(|h| {
h.config
.as_ref()
.and_then(|c| c.get("url"))
.map(String::as_str)
== Some(target_url.as_str())
});
if already_exists {
tracing::debug!(%target_url, "knowledge: push webhook already configured");
return Ok(());
}
// Delete stale hooks that point at our path but a different base
// (e.g. old loopback hooks from before the SSRF-bypass migration).
for h in &hooks {
let hook_url = h
.config
.as_ref()
.and_then(|c| c.get("url"))
.map_or("", String::as_str);
if hook_url.ends_with("/webhook/knowledge")
&& hook_url != target_url
&& let Some(id) = h.id
{
tracing::info!(hook_url, "knowledge: deleting stale webhook (wrong base)");
let _ = tokio::time::timeout(
HTTP_TIMEOUT,
client.repo_delete_hook(ORG, REPO, id).send(),
)
.await;
}
}
}
Err(e) => {
tracing::debug!(error = %e, "knowledge: listing hooks failed; attempting create");
}
}
// Create the webhook.
let mut additional = BTreeMap::new();
additional.insert("secret".to_owned(), webhook_secret.to_owned());
let hook = CreateHookOption {
active: Some(true),
authorization_header: None,
branch_filter: None,
config: CreateHookOptionConfig {
content_type: "json".to_owned(),
url: url::Url::parse(&target_url).context("parse webhook target url")?,
additional,
},
events: Some(vec!["push".to_owned()]),
r#type: CreateHookOptionType::Forgejo,
};
tokio::time::timeout(HTTP_TIMEOUT, client.repo_create_hook(ORG, REPO, hook))
let hooks = tokio::time::timeout(HTTP_TIMEOUT, client.repo_list_hooks(ORG, REPO).all())
.await
.map_err(anyhow::Error::from)
.and_then(|r| r.map_err(anyhow::Error::from))
.with_context(|| format!("create webhook for {ORG}/{REPO}"))?;
tracing::info!(%target_url, "knowledge: push webhook created");
.with_context(|| format!("list webhooks for {ORG}/{REPO}"))?;
for h in &hooks {
let hook_url = h
.config
.as_ref()
.and_then(|c| c.get("url"))
.map_or("", String::as_str);
if hook_url == own_url
&& let Some(id) = h.id
{
tokio::time::timeout(HTTP_TIMEOUT, client.repo_delete_hook(ORG, REPO, id).send())
.await
.map_err(anyhow::Error::from)
.and_then(|r| r.map_err(anyhow::Error::from))
.with_context(|| format!("delete webhook {id} for {ORG}/{REPO}"))?;
tracing::info!(
%own_url,
"knowledge: removed this hive's push webhook — the swarm controller owns it now"
);
}
}
Ok(())
}

View file

@ -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`:

View file

@ -170,13 +170,14 @@ pub(super) enum DeliveryKind {
/// The route prefix a registered `target_url` must point at.
///
/// ⚠️ Deliberately **not** `/webhook/knowledge` or `/webhook/config-pr`, the
/// paths the per-hive receivers use. Both hive-side registrars delete any
/// hook whose URL ends with *their* path but has a different base — see
/// `hive-c0re`'s `forge::ensure_config_pr_webhook` and
/// `workers::knowledge::ensure_webhook`. A swarm-level hook under those
/// paths would therefore be deleted by every hive on every boot, and the
/// symptom is a hook that silently stops existing. `webhook_urls_survive_the_hive_side_reapers`
/// pins that.
/// paths the per-hive receivers use. A hive-side registrar deletes any hook
/// whose URL ends with *its* path but has a different base — see
/// `hive-c0re`'s `forge::ensure_config_pr_webhook`. A swarm-level hook under
/// such a path would therefore be deleted by every hive on every boot, and
/// the symptom is a hook that silently stops existing.
/// `webhook_urls_survive_the_hive_side_reapers` pins that, and its own doc
/// records why `/webhook/knowledge` stays in the check even though the
/// knowledge registrar no longer reaps.
const ROUTE_PREFIX: &str = "/webhook/forge/";
impl DeliveryKind {
@ -365,9 +366,75 @@ pub(super) async fn post_webhook_forge(
bytes = body.len(),
"webhook: verified delivery"
);
// Only the knowledge delivery is acted on. 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. Written as a condition rather than a match arm holding
// an empty body, which would claim this is where that path is handled.
if kind == DeliveryKind::Knowledge {
announce_knowledge_change(&state).await;
}
(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 publish to one shared subject, not one per hive: every subscriber gets
/// the same empty event, so the roster is never consulted and this daemon does
/// not need to know who the hives are in order to tell them.
///
/// # 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();
let subject = swarm_queue_client::KNOWLEDGE_SUBJECT;
// One publish, not one per hive: every subscriber gets the same empty
// event, so the roster is not consulted at all. The controller does not
// need to know who the hives are in order to say the repository moved.
if let Err(e) = client.publish(subject, Vec::new().into()).await {
tracing::warn!(%subject, error = %e, "webhook: publishing the knowledge event failed");
return;
}
// Logged after the flush rather than after the publish: `publish` only
// hands the message to the client's write buffer, so a line printed there
// would claim delivery this end cannot yet know about.
if let Err(e) = client.flush().await {
tracing::warn!(%subject, error = %e, "webhook: flushing the knowledge event failed");
return;
}
tracing::info!(%subject, "webhook: knowledge event published");
}
#[cfg(test)]
mod tests {
use super::{DeliveryKind, Refusal, load_or_generate_at, secret_path_from, verify};
@ -567,17 +634,22 @@ mod tests {
}
}
/// A cross-daemon invariant with nothing else to enforce it: both
/// per-hive registrars in `hive-c0re` **delete** hooks whose URL ends
/// with their own path but carries a different base — see
/// `forge::ensure_config_pr_webhook` and `knowledge::ensure_webhook`.
/// A cross-daemon invariant with nothing else to enforce it: a per-hive
/// registrar in `hive-c0re` **deletes** hooks whose URL ends with its
/// own path but carries a different base. A controller URL matching such
/// a suffix would be deleted by every hive on every boot — the swarm hook
/// would simply cease to exist, with the cause in a different daemon's
/// startup sweep. Serving these under `/webhook/forge/` is what avoids
/// it, and this is the only place that says so in a form that fails.
///
/// While the swarm-level hooks live alongside the per-hive ones, a
/// controller URL matching either suffix would be deleted by every hive
/// on every boot: the swarm hook would simply cease to exist, with the
/// cause in a different daemon's startup sweep. Serving these under
/// `/webhook/forge/` is what avoids it, and this is the only place that
/// says so in a form that fails.
/// `forge::ensure_config_pr_webhook` still reaps that way. The knowledge
/// registrar no longer does — it was replaced by a removal that matches
/// the full URL, so it cannot touch another hive's hook. **The
/// `/webhook/knowledge` arm is kept anyway**, because the hazard is not
/// this repository's current code: it is whatever is *deployed*, and a
/// hive still running the previous version reaps by suffix until it is
/// upgraded. Drop that arm once no such hive can exist, not when the
/// source stops mentioning it.
#[test]
fn webhook_urls_survive_the_hive_side_reapers() {
for suffix in ["/webhook/knowledge", "/webhook/config-pr"] {

View file

@ -234,6 +234,19 @@ impl Policy {
// stays for a named/durable consumer.
format!("$JS.API.CONSUMER.CREATE.{stream}"),
format!("$JS.API.CONSUMER.CREATE.{stream}.>"),
// The knowledge event. One writer, many readers: the controller is
// the only publisher and every hive subscribes, so this is one
// literal subject rather than a per-hive family — named from the
// crate the publisher and the subscriber also name, so a rename
// cannot leave the grant pointing at a subject nobody uses.
//
// This is the reader's only non-JetStream subject, and without it
// the controller cannot emit the event at all. Worth stating because
// the symptom is unhelpful: a refused publish reaches the client as
// a **timeout**, so the visible failure is a hive that never hears
// about a change, with nothing in the controller's log to say a
// permission was the reason.
swarm_queue_client::KNOWLEDGE_SUBJECT.to_owned(),
]);
subjects
}
@ -295,6 +308,40 @@ mod tests {
assert!(!p.publish.iter().any(|s| s.contains("beta")));
}
#[test]
fn a_reader_may_publish_the_knowledge_event() {
let p = policy().permissions("swarm-controller").expect("a reader");
assert!(
p.publish
.contains(&swarm_queue_client::KNOWLEDGE_SUBJECT.to_owned()),
"the controller is the only publisher of this event; without the \
grant its publish is refused, and a refusal arrives as a timeout"
);
}
#[test]
fn a_hive_may_not_publish_the_knowledge_event_to_anyone_including_itself() {
// The controller *interprets* what a delivery means; a hive receives
// that verdict. A hive able to publish here could tell every other hive
// in the swarm — or itself — that the knowledge repo changed when it did
// not, which is an unauthenticated write into someone else's control
// path wearing an event's shape.
//
// One writer and many readers makes this arm matter MORE, not less: with
// a single shared subject a forged event reaches the whole swarm, where
// a per-hive subject would have reached one.
let p = policy()
.permissions("hive-alpha")
.expect("a hive is admitted");
assert!(
!p.publish
.iter()
.any(|s| s == swarm_queue_client::KNOWLEDGE_SUBJECT),
"a hive must not publish the knowledge event: {:?}",
p.publish
);
}
#[test]
fn a_hive_grant_never_includes_the_jetstream_wildcard() {
// `$JS.API.>` also covers `$JS.API.STREAM.DELETE.KV_hive-status`, with

View file

@ -156,6 +156,15 @@ pub fn chain(error: &dyn std::error::Error) -> String {
/// which is the disagreement this module exists to prevent.
pub mod status;
/// The subject the swarm controller publishes on when the hive-wide knowledge
/// repository has changed. One writer, many readers — every hive subscribes.
///
/// Here rather than in a module of its own for the same reason as the bucket
/// name above: three crates must agree on the string, and the one that agrees
/// hardest — the auth-callout responder, which decides whether a publish is
/// permitted at all — speaks neither `jetstream` nor `kv`.
pub const KNOWLEDGE_SUBJECT: &str = "$SWARM.knowledge";
/// Only the fields this needs; authelia returns several.
#[derive(serde::Deserialize)]
struct TokenResponse {