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.
This commit is contained in:
parent
8baf1899d8
commit
bac4a8b6a1
2 changed files with 82 additions and 0 deletions
|
|
@ -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
|
/// Reads [`STALE_AFTER_ENV`], falling back to
|
||||||
/// [`DEFAULT_STALE_AFTER`]. A zero or unparseable value takes the
|
/// [`DEFAULT_STALE_AFTER`]. A zero or unparseable value takes the
|
||||||
/// default rather than failing startup — same rule as `load_hives`:
|
/// default rather than failing startup — same rule as `load_hives`:
|
||||||
|
|
|
||||||
|
|
@ -365,9 +365,74 @@ pub(super) async fn post_webhook_forge(
|
||||||
bytes = body.len(),
|
bytes = body.len(),
|
||||||
"webhook: verified delivery"
|
"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()
|
(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)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::{DeliveryKind, Refusal, load_or_generate_at, secret_path_from, verify};
|
use super::{DeliveryKind, Refusal, load_or_generate_at, secret_path_from, verify};
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue