Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/swarm-controller/src/forge/runner_token.rs
atlas 6ba29c4d33 swarm-controller: re-read the forge token from the store when the forge rejects it
The controller read its forge token from the swarm secret store once, when
its client slot filled, and kept sending it until the daemon restarted. A
token the forge re-minted or revoked left every forge-backed path failing
with 401 until someone restarted the controller, and a restart drops the
runtime-only job graph.

Every forge request now goes through `forge::api::Api::call`. On a 401 it
asks a `StoredSecret` (the queue clients' debounced, single-flight re-read
with backoff) for a token to try instead, rebuilds the forgejo client with
it, and sends the request once more. A second 401, or a store still holding
the refused token, surfaces the error as before. Only 401 counts: unlike an
IdP, the forge answers 400 to a malformed request, and a re-read started by
one would put a real 401 that follows inside the backoff.

forgejo-api bakes the token into its own reqwest client's default headers
and takes no client or middleware from outside, so there is no single
request function to hook. Each `self.api.<endpoint>(..)` site becomes
`self.api.call(|api| api.<endpoint>(..).into_future())`; option structs a
site moved into the call are cloned, since the closure can run twice. The
`OnceLock<Arc<Client>>` slot and its `get()` callers are unchanged.

Refs #5017
2026-10-14 10:16:18 +02:00

124 lines
4.2 KiB
Rust

//! The forge's runner registration token, published to the swarm secret store
//! at [`ci::registration_token_path`], where the CI runner's fetch unit
//! (`nix/host-modules/swarm-ci-runner.nix`) reads it under the runner's own
//! certificate.
//!
//! Forgejo answers `GET /admin/runners/registration-token` with the same
//! instance-wide token until someone resets it. The runner re-registers on
//! every new value it reads, which leaves its previous entry on the forge, so
//! a pass writes only when the forge's token differs from the one this process
//! last wrote. The controller's store grant on the path is write-only, so that
//! last write is held in memory: a restarted controller writes the current
//! token once.
use std::sync::Arc;
use anyhow::{Context, Result};
use swarm_secret_client::ci;
use super::Client;
/// How often [`spawn`] re-reads the forge's token.
const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);
/// Whether `current`, the forge's token now, has to be written, given the one
/// this process last wrote (`None` before its first write).
fn needs_write(last_written: Option<&str>, current: &str) -> bool {
last_written != Some(current)
}
impl Client {
/// The forge's instance-wide runner registration token.
///
/// # Errors
/// When the forge refuses the call, or answers without a token.
pub async fn runner_registration_token(&self) -> Result<String> {
let answer = self
.api
.call(|api| api.admin_get_registration_token().into_future())
.await
.context("GET /admin/runners/registration-token")?;
answer
.token
.filter(|t| !t.is_empty())
.context("the forge answered without a runner registration token")
}
}
/// One pass: read the forge's token and write it when it is not the one
/// `last_written` holds. Returns the token it wrote, if it wrote one.
async fn pass(client: &Client, last_written: Option<&str>) -> Result<Option<String>> {
let token = client.runner_registration_token().await?;
if !needs_write(last_written, &token) {
return Ok(None);
}
let path = ci::registration_token_path()?;
let store = crate::store::connect()
.await
.context("logging in to the swarm secret store")?;
store
.write(
&path,
&ci::Credential {
value: token.clone(),
},
)
.await
.with_context(|| format!("writing {path}"))?;
Ok(Some(token))
}
/// Publish the forge's runner registration token now and every
/// [`RECONCILE_INTERVAL`] after. A pass that fails is logged and retried on
/// the next tick; it never stops the daemon.
pub fn spawn(client: Arc<Client>) {
tokio::spawn(async move {
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
let mut last_written: Option<String> = None;
loop {
ticker.tick().await;
let Some(outcome) = super::within_forge_deadline(
"runner registration token",
pass(&client, last_written.as_deref()),
)
.await
else {
continue;
};
match outcome {
Ok(Some(written)) => {
tracing::info!("runner registration token: written to the store");
last_written = Some(written);
}
Ok(None) => tracing::debug!("runner registration token: unchanged"),
Err(e) => tracing::warn!(
error = %format!("{e:#}"),
retry_in_s = RECONCILE_INTERVAL.as_secs(),
"runner registration token: pass failed; retrying next tick"
),
}
}
});
}
#[cfg(test)]
mod tests {
use super::needs_write;
#[test]
fn the_first_pass_writes() {
assert!(needs_write(None, "t1"));
}
#[test]
fn an_unchanged_token_is_not_rewritten() {
// A rewrite of the same value is harmless to the runner, but it is a
// new store version every five minutes for nothing.
assert!(!needs_write(Some("t1"), "t1"));
}
#[test]
fn a_reset_token_is_written() {
assert!(needs_write(Some("t1"), "t2"));
}
}