swarm-controller gains forge access when the token arrives, without a restart
A controller that started before the forge's host had published its token ran without forge access until someone restarted it: on a split-host first deploy that is a manual step. Now the forge client lives in a slot that is filled once (`Late<T>`, an `Arc<OnceLock<T>>`): - `forge::enable` tries the store once at start. On failure it logs once and re-reads in the background with backoff (30s doubling to a 5-minute cap, no log per retry), fills the slot and logs once when the token is there. - The forge passes (#5014's per-hive token mint among them), the config-PR poll and the webhook registration start from that moment, through `start_forge`'s callback. - Job nodes and HTTP handlers read the slot when they run, so they answer "no forge" before and work after. A token replaced in the store after the slot is filled still reaches the controller with its next restart. The round-1 change that shared `queue_identity::fetch_with_retry` is reverted; nothing else uses it. Refs #4998
This commit is contained in:
atlas
parent
121c4678aa
commit
c45f7a5330
11 changed files with 243 additions and 135 deletions
|
|
@ -6,13 +6,17 @@
|
|||
//!
|
||||
//! The forge's URL comes from `swarm-controller.nix`'s `forgeEnv`
|
||||
//! (`SWARM_CONTROLLER_FORGE_URL`). The token comes from the swarm's secret
|
||||
//! store, at `swarm_secret_client::forge::controller_token_path`, read once at
|
||||
//! start under this daemon's own certificate and held only in memory. The
|
||||
//! forge mints it inside its container (`hive-forge/default.nix`'s
|
||||
//! store, at `swarm_secret_client::forge::controller_token_path`, read under
|
||||
//! this daemon's own certificate and held only in memory. The forge mints it
|
||||
//! inside its container (`hive-forge/default.nix`'s
|
||||
//! `forgejo-swarm-controller-account`) and `swarm-bao-forge-publisher`, on the
|
||||
//! forge's host, puts it there. An unset URL means [`Client::from_env`]
|
||||
//! returns `None` — every caller treats that as "no forge access here" and
|
||||
//! continues, the same graceful-absence shape the queue coordinates use.
|
||||
//! forge's host, puts it there.
|
||||
//!
|
||||
//! [`enable`] fills the daemon's one client slot: at start when the token is
|
||||
//! there, later from a background re-read when it is not. Every caller reads
|
||||
//! the slot and treats an empty one as "no forge access here" — the same
|
||||
//! graceful-absence shape the queue coordinates use — so the daemon serves
|
||||
//! from the start and gains forge access without a restart.
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use forgejo_api::structs::{
|
||||
|
|
@ -28,6 +32,7 @@ use futures_util::{StreamExt as _, TryStreamExt as _};
|
|||
use reqwest::StatusCode;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::{BTreeMap, HashMap, HashSet};
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use utoipa::ToSchema;
|
||||
|
||||
use crate::webhook::DeliveryKind;
|
||||
|
|
@ -130,7 +135,7 @@ const KNOWLEDGE_REPO: &str = "knowledge";
|
|||
const SWARM_CONTROLLER_FORGE_USER: &str = "swarm-controller";
|
||||
|
||||
/// The env var `swarm-controller.nix`'s `forgeEnv` sets. Named here once so
|
||||
/// [`Client::from_env`] and the nix module can't drift silently.
|
||||
/// [`url_from_env`] and the nix module can't drift silently.
|
||||
const URL_ENV: &str = "SWARM_CONTROLLER_FORGE_URL";
|
||||
|
||||
/// One login and one read. An empty value is refused like an absent one: it
|
||||
|
|
@ -163,34 +168,104 @@ pub struct Client {
|
|||
url: url::Url,
|
||||
}
|
||||
|
||||
/// The forge's URL. `Ok(None)` (not an error) when `SWARM_CONTROLLER_FORGE_URL`
|
||||
/// is unset — no forge is configured for this daemon.
|
||||
///
|
||||
/// # Errors
|
||||
/// A URL that does not parse. Retrying cannot fix that, so it is reported once
|
||||
/// rather than waited out.
|
||||
pub fn url_from_env() -> Result<Option<url::Url>> {
|
||||
let Some(url) = std::env::var(URL_ENV).ok() else {
|
||||
return Ok(None);
|
||||
};
|
||||
url::Url::parse(&url)
|
||||
.with_context(|| format!("parse {URL_ENV} ({url}) as a URL"))
|
||||
.map(Some)
|
||||
}
|
||||
|
||||
/// How long the background re-read waits before its `n`-th attempt: 30s,
|
||||
/// doubling, capped at 5 minutes.
|
||||
fn background_delay(n: u32) -> std::time::Duration {
|
||||
const FIRST_S: u64 = 30;
|
||||
const CAP_S: u64 = 300;
|
||||
std::time::Duration::from_secs(
|
||||
FIRST_S
|
||||
.saturating_mul(1u64.checked_shl(n).unwrap_or(u64::MAX))
|
||||
.min(CAP_S),
|
||||
)
|
||||
}
|
||||
|
||||
/// Fill `slot` with the first value `attempt` yields, sleeping
|
||||
/// [`background_delay`] before each attempt. Never gives up, and logs no
|
||||
/// failure: the caller logs the first one and the recovery.
|
||||
async fn fill_when_ready<T, F, Fut>(slot: &OnceLock<T>, mut attempt: F) -> &T
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: std::future::Future<Output = Result<T>>,
|
||||
{
|
||||
let mut n = 0;
|
||||
loop {
|
||||
tokio::time::sleep(background_delay(n)).await;
|
||||
match attempt().await {
|
||||
Ok(value) => return slot.get_or_init(|| value),
|
||||
Err(e) => tracing::debug!(
|
||||
error = %format!("{e:#}"),
|
||||
"the controller's forge token is still unavailable"
|
||||
),
|
||||
}
|
||||
n = n.saturating_add(1);
|
||||
}
|
||||
}
|
||||
|
||||
/// Put a client for `url` into `slot` and hand it to `on_ready`: at once when
|
||||
/// the store has the token, otherwise from a background task that re-reads it
|
||||
/// until it does. Either way the daemon starts, and every forge-backed path
|
||||
/// reads `slot`, so it answers "no forge" until the token lands and works
|
||||
/// from then on.
|
||||
///
|
||||
/// A token replaced in the store after the slot is filled reaches this daemon
|
||||
/// with its next restart.
|
||||
pub async fn enable(
|
||||
url: url::Url,
|
||||
slot: Arc<OnceLock<Arc<Client>>>,
|
||||
on_ready: impl FnOnce(Arc<Client>) + Send + 'static,
|
||||
) {
|
||||
match Client::connect(&url).await {
|
||||
Ok(client) => on_ready(Arc::clone(slot.get_or_init(|| Arc::new(client)))),
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
error = %format!("{e:#}"),
|
||||
"no forge access yet; re-reading the forge token from the swarm secret store in \
|
||||
the background"
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
let client = Arc::clone(
|
||||
fill_when_ready(&slot, || async {
|
||||
Client::connect(&url).await.map(Arc::new)
|
||||
})
|
||||
.await,
|
||||
);
|
||||
tracing::info!("the forge token is in the swarm secret store; forge access is on");
|
||||
on_ready(client);
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Client {
|
||||
/// Build from env and the secret store. `Ok(None)` (not an error) when
|
||||
/// `SWARM_CONTROLLER_FORGE_URL` is unset — no forge is configured for this
|
||||
/// daemon (see the module doc comment).
|
||||
/// One read of the token from the secret store, and a client for `url`
|
||||
/// holding it.
|
||||
///
|
||||
/// # Errors
|
||||
/// An unparseable URL, or a token the store still would not give up after
|
||||
/// `queue_identity::RETRY_DELAYS_S`: the store refusing the login, or
|
||||
/// nothing at the path because the forge's host has not published it.
|
||||
pub async fn from_env() -> Result<Option<Self>> {
|
||||
let Some(url) = std::env::var(URL_ENV).ok() else {
|
||||
return Ok(None);
|
||||
};
|
||||
let parsed_url =
|
||||
url::Url::parse(&url).with_context(|| format!("parse {URL_ENV} ({url}) as a URL"))?;
|
||||
let delays = crate::queue_identity::RETRY_DELAYS_S.map(std::time::Duration::from_secs);
|
||||
let token = crate::queue_identity::fetch_with_retry(
|
||||
&delays,
|
||||
"the controller's forge token",
|
||||
fetch_token,
|
||||
)
|
||||
.await?;
|
||||
let api = Forgejo::new(Auth::Token(&token), parsed_url.clone())
|
||||
.context("build forgejo client")?;
|
||||
Ok(Some(Self {
|
||||
/// The store refusing the login, nothing at the path because the forge's
|
||||
/// host has not published it yet, or the client failing to build.
|
||||
async fn connect(url: &url::Url) -> Result<Self> {
|
||||
let token = fetch_token().await?;
|
||||
let api = Forgejo::new(Auth::Token(&token), url.clone()).context("build forgejo client")?;
|
||||
Ok(Self {
|
||||
api,
|
||||
url: parsed_url,
|
||||
}))
|
||||
url: url.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Creation options for an empty repo defaulting to `main`.
|
||||
|
|
@ -1306,33 +1381,59 @@ mod tests {
|
|||
// the same process-global env var would race each other. A single
|
||||
// test's body is one thread, so sequential set/clear within it is
|
||||
// race-free without needing a cross-test lock.
|
||||
#[tokio::test]
|
||||
async fn from_env_is_none_without_a_url_and_refuses_a_bad_one_before_the_store() {
|
||||
#[test]
|
||||
fn url_from_env_is_none_without_a_url_and_refuses_a_bad_one() {
|
||||
// SAFETY: test-only mutation of the env var this module itself
|
||||
// owns (`URL_ENV`), scoped to this one test function so there's no
|
||||
// cross-test race (see comment above).
|
||||
unsafe {
|
||||
std::env::remove_var(URL_ENV);
|
||||
}
|
||||
assert!(Client::from_env().await.unwrap().is_none());
|
||||
assert_eq!(url_from_env().unwrap(), None);
|
||||
|
||||
unsafe {
|
||||
std::env::set_var(URL_ENV, "not a url");
|
||||
}
|
||||
// Not `.unwrap_err()`: that needs `Client: Debug` for its panic
|
||||
// message, and the wrapped `forgejo_api::Forgejo` doesn't derive
|
||||
// it. Returning at once, with no `BAO_*` set, is what shows the URL
|
||||
// is checked before the store is asked.
|
||||
match Client::from_env().await {
|
||||
Err(e) => assert!(e.to_string().contains(URL_ENV), "{e:#}"),
|
||||
Ok(_) => panic!("expected an error for an unparseable forge URL"),
|
||||
}
|
||||
let err = url_from_env().expect_err("an unparseable forge URL");
|
||||
assert!(err.to_string().contains(URL_ENV), "{err:#}");
|
||||
|
||||
unsafe {
|
||||
std::env::remove_var(URL_ENV);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_background_re_read_backs_off_from_30s_to_a_5_minute_cap() {
|
||||
let s = |n| background_delay(n).as_secs();
|
||||
assert_eq!([s(0), s(1), s(2), s(3), s(4)], [30, 60, 120, 240, 300]);
|
||||
assert_eq!(s(40), 300);
|
||||
assert_eq!(s(u32::MAX), 300);
|
||||
}
|
||||
|
||||
// Paused time: each sleep completes at once and advances the clock, so the
|
||||
// elapsed time is the backoff schedule itself.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn a_token_absent_at_start_fills_the_slot_once_it_arrives() {
|
||||
let slot = OnceLock::new();
|
||||
let start = tokio::time::Instant::now();
|
||||
let mut calls = 0;
|
||||
let got = fill_when_ready(&slot, || {
|
||||
calls += 1;
|
||||
let n = calls;
|
||||
async move {
|
||||
if n < 3 {
|
||||
anyhow::bail!("nothing at the controller's forge token path")
|
||||
}
|
||||
Ok("client")
|
||||
}
|
||||
})
|
||||
.await;
|
||||
assert_eq!(*got, "client");
|
||||
assert_eq!(slot.get(), Some(&"client"));
|
||||
assert_eq!(calls, 3);
|
||||
assert_eq!(start.elapsed().as_secs(), 30 + 60 + 120);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn page_search_results_pages_past_a_full_first_page() {
|
||||
#[derive(Clone)]
|
||||
|
|
|
|||
|
|
@ -157,10 +157,10 @@ mod tests {
|
|||
hive_jobq::resources::ResourceTable::new(),
|
||||
))),
|
||||
webhook_secret: None,
|
||||
config_prs: None,
|
||||
config_prs: std::sync::Arc::default(),
|
||||
swarm_name: None,
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
create_gate: std::sync::Arc::default(),
|
||||
accounts: Accounts::Fake(Arc::clone(store)),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -127,10 +127,10 @@ mod tests {
|
|||
hive_jobq::resources::ResourceTable::new(),
|
||||
))),
|
||||
webhook_secret: None,
|
||||
config_prs: None,
|
||||
config_prs: std::sync::Arc::default(),
|
||||
swarm_name: None,
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
create_gate: std::sync::Arc::default(),
|
||||
accounts: Accounts::Fake(Arc::clone(store)),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@ use crate::forge;
|
|||
pub async fn get_repos(
|
||||
State(state): State<AppState>,
|
||||
) -> Result<Json<Vec<String>>, StatusUnavailable> {
|
||||
let Some(client) = state.forge.as_ref() else {
|
||||
let Some(client) = state.forge.get() else {
|
||||
return Err(StatusUnavailable(
|
||||
"no forge is configured on this host".to_owned(),
|
||||
));
|
||||
|
|
@ -64,7 +64,7 @@ pub async fn get_repos(
|
|||
pub async fn get_issue_report_all(
|
||||
State(state): State<AppState>,
|
||||
) -> Result<Json<Vec<forge::IssueReportRow>>, StatusUnavailable> {
|
||||
let Some(client) = state.forge.as_ref() else {
|
||||
let Some(client) = state.forge.get() else {
|
||||
return Err(StatusUnavailable(
|
||||
"no forge is configured on this host".to_owned(),
|
||||
));
|
||||
|
|
@ -109,7 +109,7 @@ pub async fn get_issue_report(
|
|||
State(state): State<AppState>,
|
||||
Path((org, repo)): Path<(String, String)>,
|
||||
) -> Result<Json<Vec<forge::IssueReportRow>>, problem_details::ProblemDetails> {
|
||||
let Some(client) = state.forge.as_ref() else {
|
||||
let Some(client) = state.forge.get() else {
|
||||
return Err(error_problem(
|
||||
axum::http::StatusCode::SERVICE_UNAVAILABLE,
|
||||
"no forge is configured on this host",
|
||||
|
|
|
|||
|
|
@ -962,10 +962,10 @@ pub(crate) mod tests {
|
|||
hive_jobq::resources::ResourceTable::new(),
|
||||
))),
|
||||
webhook_secret: None,
|
||||
config_prs: None,
|
||||
config_prs: std::sync::Arc::default(),
|
||||
swarm_name: None,
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
create_gate: std::sync::Arc::default(),
|
||||
accounts: super::Accounts::Fake(Arc::clone(store)),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -234,6 +234,10 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind {
|
|||
}
|
||||
}
|
||||
|
||||
/// A dependency this daemon may only gain after it starts serving: empty
|
||||
/// until then, filled once. Readers take what is there at the moment they ask.
|
||||
type Late<T> = Arc<std::sync::OnceLock<T>>;
|
||||
|
||||
/// Everything a claimed node's executor arm might need to reach outside
|
||||
/// this process — bundled into one `Clone` struct rather than growing
|
||||
/// `run_swarm_node`'s parameter list per node kind (three forge-shaped
|
||||
|
|
@ -245,7 +249,9 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind {
|
|||
#[derive(Clone)]
|
||||
struct WorkerDeps {
|
||||
auth: Option<std::sync::Arc<auth::AuthBridge>>,
|
||||
forge: Option<std::sync::Arc<forge::Client>>,
|
||||
/// Empty until the forge token has been read (`forge::enable`), so a node
|
||||
/// run before then fails as "no forge access", and one run after succeeds.
|
||||
forge: Late<Arc<forge::Client>>,
|
||||
/// The swarm queue connection, for nodes whose effect is a published
|
||||
/// event rather than an API call. A handle rather than the status
|
||||
/// reader it is cloned from: a node that announces a deploy has no
|
||||
|
|
@ -288,6 +294,7 @@ async fn run_swarm_node(
|
|||
) {
|
||||
use hive_jobq::scheduler::Outcome;
|
||||
|
||||
let forge = deps.forge.get().cloned();
|
||||
let outcome = match kind {
|
||||
SwarmNodeKind::CreateIdentity { agent } => match deps.auth {
|
||||
None => {
|
||||
|
|
@ -311,7 +318,7 @@ async fn run_swarm_node(
|
|||
Err(e) => Outcome::Failed(format!("{e:#}")),
|
||||
},
|
||||
},
|
||||
SwarmNodeKind::CreateRepo { agent } => match deps.forge {
|
||||
SwarmNodeKind::CreateRepo { agent } => match forge {
|
||||
None => Outcome::Failed(
|
||||
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
|
||||
not in the swarm secret store)"
|
||||
|
|
@ -325,7 +332,7 @@ async fn run_swarm_node(
|
|||
Err(e) => Outcome::Failed(format!("{e:#}")),
|
||||
},
|
||||
},
|
||||
SwarmNodeKind::CreateForgeUser { agent } => match deps.forge {
|
||||
SwarmNodeKind::CreateForgeUser { agent } => match forge {
|
||||
None => Outcome::Failed(
|
||||
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
|
||||
not in the swarm secret store)"
|
||||
|
|
@ -336,7 +343,7 @@ async fn run_swarm_node(
|
|||
Err(e) => Outcome::Failed(format!("{e:#}")),
|
||||
},
|
||||
},
|
||||
SwarmNodeKind::AddRepoMember { agent } => match deps.forge {
|
||||
SwarmNodeKind::AddRepoMember { agent } => match forge {
|
||||
None => Outcome::Failed(
|
||||
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
|
||||
not in the swarm secret store)"
|
||||
|
|
@ -347,8 +354,8 @@ async fn run_swarm_node(
|
|||
Err(e) => Outcome::Failed(format!("{e:#}")),
|
||||
},
|
||||
},
|
||||
SwarmNodeKind::GrantDocsRead { agent } => grant_docs_read(deps.forge, &agent).await,
|
||||
SwarmNodeKind::InitAgentConfigRepo { agent } => match deps.forge {
|
||||
SwarmNodeKind::GrantDocsRead { agent } => grant_docs_read(forge, &agent).await,
|
||||
SwarmNodeKind::InitAgentConfigRepo { agent } => match forge {
|
||||
None => Outcome::Failed(
|
||||
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
|
||||
not in the swarm secret store)"
|
||||
|
|
@ -360,16 +367,14 @@ async fn run_swarm_node(
|
|||
},
|
||||
},
|
||||
SwarmNodeKind::MintAgentIdentity { agent } => mint_identity(&agent).await,
|
||||
SwarmNodeKind::MintAgentForgeToken { agent } => mint_forge_token(deps.forge, &agent).await,
|
||||
SwarmNodeKind::MintAgentForgeToken { agent } => mint_forge_token(forge, &agent).await,
|
||||
SwarmNodeKind::MintAgentMatrixAccount { agent } => {
|
||||
mint_matrix_account(deps.matrix_homeserver.as_deref(), &agent).await
|
||||
}
|
||||
SwarmNodeKind::MintHiveSenderToken { hive } => {
|
||||
mint_hive_sender(deps.matrix_homeserver.as_deref(), &hive).await
|
||||
}
|
||||
SwarmNodeKind::MintHiveForgeToken { hive } => {
|
||||
mint_hive_forge_token(deps.forge, &hive).await
|
||||
}
|
||||
SwarmNodeKind::MintHiveForgeToken { hive } => mint_hive_forge_token(forge, &hive).await,
|
||||
SwarmNodeKind::RenewAgentQueueCredential { agent } => renew_queue_credential(&agent).await,
|
||||
SwarmNodeKind::SetAgentWanted { hive, agent } => match deps.wanted {
|
||||
None => Outcome::Failed(
|
||||
|
|
@ -774,10 +779,10 @@ struct AppState {
|
|||
/// takes without a second hop through `String`.
|
||||
webhook_secret: Option<Arc<str>>,
|
||||
/// Last successful `agent-configs/*` scan, kept current by
|
||||
/// [`config_pr::spawn`]. `None` when this host has no forge configured
|
||||
/// — same "absent means don't ask" shape as `status` and `forge` above,
|
||||
/// not a startup failure.
|
||||
config_prs: Option<Arc<config_pr::ConfigPrCache>>,
|
||||
/// [`config_pr::spawn`]. Empty while this daemon has no forge access —
|
||||
/// same "absent means don't ask" shape as `status` and `forge`, not a
|
||||
/// startup failure — and filled when the forge client is.
|
||||
config_prs: Late<Arc<config_pr::ConfigPrCache>>,
|
||||
/// The swarm's human display name (`services.hyperhive.swarm.name`),
|
||||
/// loaded once at startup (`load_swarm_name`). `None` when the
|
||||
/// operator never set it — a swarm without a display name is a
|
||||
|
|
@ -788,10 +793,9 @@ struct AppState {
|
|||
/// The forge client itself, for the read-only repo/issue-report
|
||||
/// routes (`GET /api/repos`, `GET /api/repos/{org}/{repo}/issue-report`)
|
||||
/// and `POST /api/forge/users/{name}/admin` — distinct from `config_prs`, which holds a *cache* built off this
|
||||
/// same client rather than the client. `None` under the same
|
||||
/// "no forge configured on this host" shape every other
|
||||
/// forge-backed field here uses.
|
||||
forge: Option<Arc<forge::Client>>,
|
||||
/// same client rather than the client. Empty while this daemon has no
|
||||
/// forge access, and filled the moment `forge::enable` builds the client.
|
||||
forge: Late<Arc<forge::Client>>,
|
||||
/// Held by `create_agent` from reading where a name is placed until its
|
||||
/// graph is queued, and by `set_agent_state` from that read until its
|
||||
/// declaration is written, so two placements of one name cannot both
|
||||
|
|
@ -1433,7 +1437,7 @@ async fn get_agents_status(
|
|||
})?;
|
||||
match reader.view(&roster, std::time::SystemTime::now()).await {
|
||||
Ok(mut rows) => {
|
||||
if let Some(cache) = state.config_prs.as_ref() {
|
||||
if let Some(cache) = state.config_prs.get() {
|
||||
let mut config_prs = cache.snapshot();
|
||||
for row in &mut rows {
|
||||
row.config_pr = config_prs.remove(&row.name);
|
||||
|
|
@ -2178,7 +2182,7 @@ async fn get_agent_config_pr(
|
|||
State(state): State<AppState>,
|
||||
Path(name): Path<String>,
|
||||
) -> Result<Json<Option<forge::ConfigPrStatus>>, StatusUnavailable> {
|
||||
let Some(cache) = state.config_prs.as_ref() else {
|
||||
let Some(cache) = state.config_prs.get() else {
|
||||
return Err(StatusUnavailable(
|
||||
"no forge is configured on this host".to_owned(),
|
||||
));
|
||||
|
|
@ -2443,16 +2447,13 @@ fn spawn_matrix_account_backfill(
|
|||
/// registration token, hives' read tokens, agents' `internal/docs` grants, and
|
||||
/// agents' tokens (which also
|
||||
/// sweeps legacy tokens; see `forge::legacy_tokens`) —
|
||||
/// when a forge is configured. Lifted out of `main` for
|
||||
/// once there is a forge client. Lifted out of `main` for
|
||||
/// `clippy::too_many_lines`.
|
||||
fn spawn_forge_workers(
|
||||
jobq: &Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
||||
forge_client: Option<Arc<forge::Client>>,
|
||||
client: Arc<forge::Client>,
|
||||
hives: &[HiveEntry],
|
||||
) {
|
||||
let Some(client) = forge_client else {
|
||||
return;
|
||||
};
|
||||
forge::objects::spawn(Arc::clone(&client));
|
||||
forge::runner_token::spawn(Arc::clone(&client));
|
||||
let sched = Arc::clone(jobq);
|
||||
|
|
@ -2648,7 +2649,7 @@ async fn make_forge_admin(
|
|||
let name = hive_types::Ident::parse(&name)
|
||||
.map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
|
||||
.into_string();
|
||||
let Some(forge) = state.forge.clone() else {
|
||||
let Some(forge) = state.forge.get().cloned() else {
|
||||
return Err(error_problem(
|
||||
axum::http::StatusCode::SERVICE_UNAVAILABLE,
|
||||
"no forge is configured on this host",
|
||||
|
|
@ -2828,7 +2829,7 @@ async fn get_agent_icon(
|
|||
async fn get_config_prs(
|
||||
State(state): State<AppState>,
|
||||
) -> Result<Json<std::collections::HashMap<String, forge::ConfigPrStatus>>, StatusUnavailable> {
|
||||
let Some(cache) = state.config_prs.as_ref() else {
|
||||
let Some(cache) = state.config_prs.get() else {
|
||||
return Err(StatusUnavailable(
|
||||
"no forge is configured on this host".to_owned(),
|
||||
));
|
||||
|
|
@ -2862,12 +2863,11 @@ const PUBLIC_URL_ENV: &str = "SWARM_CONTROLLER_PUBLIC_URL";
|
|||
/// daemon keeps running — and the failure is silent at both ends, because the
|
||||
/// forge has nothing to report about a call that never arrived.
|
||||
///
|
||||
/// Silently does nothing when any of the three preconditions is missing —
|
||||
/// each is a legitimate deployment shape (no forge here, no state directory
|
||||
/// to hold a secret, no public vhost), and each is already logged where it
|
||||
/// is discovered.
|
||||
fn register_swarm_webhooks(forge: Option<Arc<forge::Client>>, secret: Option<Arc<str>>) {
|
||||
let Some(forge) = forge else { return };
|
||||
/// Runs once there is a forge client, and silently does nothing when either
|
||||
/// other precondition is missing — each is a legitimate deployment shape (no
|
||||
/// state directory to hold a secret, no public vhost), and each is already
|
||||
/// logged where it is discovered.
|
||||
fn register_swarm_webhooks(forge: Arc<forge::Client>, secret: Option<Arc<str>>) {
|
||||
let Some(secret) = secret else { return };
|
||||
let Ok(public_url) = std::env::var(PUBLIC_URL_ENV) else {
|
||||
tracing::info!(
|
||||
|
|
@ -2953,18 +2953,32 @@ async fn connect_status_reader() -> Result<Option<Arc<status::StatusReader>>> {
|
|||
}
|
||||
}
|
||||
|
||||
/// Clones `forge_client` for [`AppState::forge`] before handing the
|
||||
/// original off to [`register_swarm_webhooks`], which takes it by value —
|
||||
/// split out of `main` purely to keep that function under clippy's
|
||||
/// line-count lint (same rationale as [`connect_status_reader`]'s own doc
|
||||
/// comment), no behavior change.
|
||||
fn keep_forge_for_state(
|
||||
forge_client: Option<Arc<forge::Client>>,
|
||||
/// Fill the forge slots and start everything that runs off the forge client —
|
||||
/// its periodic passes, the config-PR poll and the webhook registration — now
|
||||
/// if the token is in the store, otherwise when `forge::enable`'s background
|
||||
/// re-read finds it. Nothing to start when no forge URL is configured. Lifted
|
||||
/// out of `main` for `clippy::too_many_lines`.
|
||||
async fn start_forge(
|
||||
forge_client: Late<Arc<forge::Client>>,
|
||||
config_prs: Late<Arc<config_pr::ConfigPrCache>>,
|
||||
jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
||||
hives: Vec<HiveEntry>,
|
||||
webhook_secret: Option<Arc<str>>,
|
||||
) -> Option<Arc<forge::Client>> {
|
||||
let state_forge = forge_client.clone();
|
||||
register_swarm_webhooks(forge_client, webhook_secret);
|
||||
state_forge
|
||||
) {
|
||||
let url = match forge::url_from_env() {
|
||||
Ok(Some(url)) => url,
|
||||
Ok(None) => return,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %format!("{e:#}"), "forge misconfigured; repo provisioning is off");
|
||||
return;
|
||||
}
|
||||
};
|
||||
forge::enable(url, forge_client, move |client| {
|
||||
spawn_forge_workers(&jobq, Arc::clone(&client), &hives);
|
||||
config_prs.get_or_init(|| config_pr::spawn(Arc::clone(&client)));
|
||||
register_swarm_webhooks(client, webhook_secret);
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
|
|
@ -3019,18 +3033,13 @@ async fn main() -> Result<()> {
|
|||
};
|
||||
// Same shape again: a controller with no forge access still serves
|
||||
// everything else, and the forge-shaped node kinds give an honest
|
||||
// per-job failure rather than this fn refusing to start. A token missing
|
||||
// from the store lands here too, after `from_env`'s retries.
|
||||
let forge_client = match forge::Client::from_env().await {
|
||||
Ok(client) => client.map(Arc::new),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %format!("{e:#}"), "no forge access; repo provisioning is off");
|
||||
None
|
||||
}
|
||||
};
|
||||
// per-job failure rather than this fn refusing to start. Both slots stay
|
||||
// empty until `start_forge` below fills them.
|
||||
let forge_client: Late<Arc<forge::Client>> = Arc::default();
|
||||
let config_prs: Late<Arc<config_pr::ConfigPrCache>> = Arc::default();
|
||||
let deps = WorkerDeps {
|
||||
auth: auth.clone(),
|
||||
forge: forge_client.clone(),
|
||||
forge: Arc::clone(&forge_client),
|
||||
queue: status.as_ref().map(|s| s.queue_client()),
|
||||
wanted: wanted_writer(status.as_ref()),
|
||||
matrix_homeserver: configured_matrix_homeserver(),
|
||||
|
|
@ -3094,9 +3103,14 @@ async fn main() -> Result<()> {
|
|||
);
|
||||
}
|
||||
|
||||
spawn_forge_workers(&jobq, forge_client.clone(), &hives);
|
||||
let config_prs = forge_client.clone().map(config_pr::spawn);
|
||||
let state_forge = keep_forge_for_state(forge_client, webhook_secret.clone());
|
||||
start_forge(
|
||||
Arc::clone(&forge_client),
|
||||
Arc::clone(&config_prs),
|
||||
Arc::clone(&jobq),
|
||||
hives.clone(),
|
||||
webhook_secret.clone(),
|
||||
)
|
||||
.await;
|
||||
|
||||
spawn_agent_renewal(&jobq, wanted_writer(status.as_ref()), &hives);
|
||||
spawn_per_agent_upkeep(status.as_ref(), &hives);
|
||||
|
|
@ -3123,7 +3137,7 @@ async fn main() -> Result<()> {
|
|||
config_prs,
|
||||
swarm_name: load_swarm_name().map(Arc::from),
|
||||
auth,
|
||||
forge: state_forge,
|
||||
forge: forge_client,
|
||||
create_gate: Arc::default(),
|
||||
accounts: linked_accounts::Accounts::Live,
|
||||
};
|
||||
|
|
@ -3295,13 +3309,13 @@ mod tests {
|
|||
agent_icons: None,
|
||||
jobq: std::sync::Arc::clone(&sched),
|
||||
webhook_secret: None,
|
||||
config_prs: None,
|
||||
config_prs: std::sync::Arc::default(),
|
||||
swarm_name: None,
|
||||
// No bridge: these tests drive agent *creation*, which queues a
|
||||
// job and consults one only for a rule-breaking name. The roster
|
||||
// read is the verb that needs it, and it has its own test below.
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
create_gate: std::sync::Arc::default(),
|
||||
accounts: super::linked_accounts::Accounts::Fake(std::sync::Arc::default()),
|
||||
};
|
||||
|
|
@ -4197,7 +4211,7 @@ mod tests {
|
|||
|
||||
let deps = WorkerDeps {
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
queue: None,
|
||||
wanted: None,
|
||||
matrix_homeserver: None,
|
||||
|
|
@ -4250,7 +4264,7 @@ mod tests {
|
|||
|
||||
let deps = WorkerDeps {
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
queue: None,
|
||||
wanted: None,
|
||||
matrix_homeserver: None,
|
||||
|
|
@ -4313,7 +4327,7 @@ mod tests {
|
|||
|
||||
let deps = WorkerDeps {
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
queue: None,
|
||||
wanted: None,
|
||||
matrix_homeserver: None,
|
||||
|
|
@ -4364,7 +4378,7 @@ mod tests {
|
|||
|
||||
let deps = WorkerDeps {
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
queue: None,
|
||||
wanted: None,
|
||||
matrix_homeserver: None,
|
||||
|
|
|
|||
|
|
@ -571,10 +571,10 @@ mod tests {
|
|||
hive_jobq::resources::ResourceTable::new(),
|
||||
))),
|
||||
webhook_secret: None,
|
||||
config_prs: None,
|
||||
config_prs: std::sync::Arc::default(),
|
||||
swarm_name: None,
|
||||
auth: None,
|
||||
forge: None,
|
||||
forge: std::sync::Arc::default(),
|
||||
create_gate: std::sync::Arc::default(),
|
||||
accounts: Accounts::Fake(Arc::clone(store)),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ const PREFIX: &str = "SWARM_CONTROLLER";
|
|||
/// Seconds to wait after each failed fetch before the next one. About a minute
|
||||
/// in all, which rides out a store restarting alongside this daemon. A longer
|
||||
/// outage fails the start, and the unit's `Restart=` tries again from the top.
|
||||
pub(crate) const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30];
|
||||
const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30];
|
||||
|
||||
static IDENTITY: OnceLock<Option<QueueConfig>> = OnceLock::new();
|
||||
|
||||
|
|
@ -61,9 +61,7 @@ pub async fn init() -> Result<()> {
|
|||
None => None,
|
||||
Some(coordinates) => {
|
||||
let delays = RETRY_DELAYS_S.map(Duration::from_secs);
|
||||
Some(coordinates.with_secret(
|
||||
fetch_with_retry(&delays, "the queue client secret", fetch_secret).await?,
|
||||
))
|
||||
Some(coordinates.with_secret(fetch_with_retry(&delays, fetch_secret).await?))
|
||||
}
|
||||
};
|
||||
IDENTITY
|
||||
|
|
@ -120,13 +118,8 @@ async fn fetch_secret() -> Result<String> {
|
|||
}
|
||||
|
||||
/// Run `attempt`, waiting each of `delays` after a failure, then once more.
|
||||
/// Answers with the first success, or the last attempt's error. `what` names the
|
||||
/// secret in the log line and the error.
|
||||
pub(crate) async fn fetch_with_retry<F, Fut>(
|
||||
delays: &[Duration],
|
||||
what: &str,
|
||||
mut attempt: F,
|
||||
) -> Result<String>
|
||||
/// Answers with the first success, or the last attempt's error.
|
||||
async fn fetch_with_retry<F, Fut>(delays: &[Duration], mut attempt: F) -> Result<String>
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: Future<Output = Result<String>>,
|
||||
|
|
@ -138,7 +131,7 @@ where
|
|||
tracing::warn!(
|
||||
error = %format!("{e:#}"),
|
||||
retry_in = ?delay,
|
||||
"fetching {what} from the swarm secret store failed"
|
||||
"fetching the queue client secret from the swarm secret store failed"
|
||||
);
|
||||
tokio::time::sleep(*delay).await;
|
||||
}
|
||||
|
|
@ -146,7 +139,7 @@ where
|
|||
}
|
||||
attempt().await.with_context(|| {
|
||||
format!(
|
||||
"fetching {what} from the swarm secret store, after {} attempts",
|
||||
"fetching the queue client secret from the swarm secret store, after {} attempts",
|
||||
delays.len() + 1
|
||||
)
|
||||
})
|
||||
|
|
@ -224,7 +217,7 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn a_store_that_comes_back_within_the_window_is_waited_for() {
|
||||
let mut calls = 0;
|
||||
let got = fetch_with_retry(&[Duration::ZERO; 3], "a secret", || {
|
||||
let got = fetch_with_retry(&[Duration::ZERO; 3], || {
|
||||
calls += 1;
|
||||
let n = calls;
|
||||
async move {
|
||||
|
|
@ -243,7 +236,7 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn a_store_that_stays_down_fails_after_one_attempt_per_delay_plus_one() {
|
||||
let mut calls = 0;
|
||||
let err = fetch_with_retry(&[Duration::ZERO; 3], "a secret", || {
|
||||
let err = fetch_with_retry(&[Duration::ZERO; 3], || {
|
||||
calls += 1;
|
||||
async { bail!("store sealed") }
|
||||
})
|
||||
|
|
|
|||
|
|
@ -364,7 +364,7 @@ pub(super) async fn post_webhook_forge(
|
|||
match kind {
|
||||
DeliveryKind::Knowledge => announce_knowledge_change(&state).await,
|
||||
DeliveryKind::ConfigPr => {
|
||||
if let Some(cache) = state.config_prs.as_ref() {
|
||||
if let Some(cache) = state.config_prs.get() {
|
||||
cache.apply_webhook_delivery(&body);
|
||||
}
|
||||
if let Some(merged) = crate::config_pr::merged(&body) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue