From 6f8a6ab8e06cbb9ceb93425609efbb62045395e4 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 9 Oct 2026 20:00:19 +0200 Subject: [PATCH] swarm-controller: bound forge calls in job executors forgejo-api's client has no request timeout, so a forge that accepts a connection and never answers held a forge-calling job node forever: the node never reached Done or Failed, and kept its executor task. Every executor arm that awaits the forge now runs its work under the same deadline the periodic passes use. On elapsing, the node fails through its existing Outcome::Failed path with ": no answer from the forge within 120s". Nodes: create_repo, create_forge_user, add_repo_member, grant_docs_read, init_agent_config_repo, mint_agent_forge_token, mint_hive_forge_token. The bound is per node, not per request: the nodes make 1 to about 15 sequential forge and store requests each (mint_hive_forge_token the most), and the requests live inside forge::Client methods the periodic passes share. One 2-minute bound covers both, so #5065's within_pass_deadline / PASS_DEADLINE become within_forge_deadline / FORGE_DEADLINE, with a log line that no longer says "retrying next tick" (true of a pass, not of a node). The InitAgentConfigRepo arm is lifted out of run_swarm_node, like its siblings, to keep that function under clippy::too_many_lines. Refs #5066 --- swarm-controller/src/config_pr.rs | 8 +- swarm-controller/src/forge.rs | 46 +++--- swarm-controller/src/forge/agent_token.rs | 2 +- swarm-controller/src/forge/docs_grant.rs | 2 +- swarm-controller/src/forge/hive_token.rs | 2 +- swarm-controller/src/forge/objects.rs | 2 +- swarm-controller/src/forge/runner_token.rs | 2 +- swarm-controller/src/main.rs | 174 ++++++++++++++++++--- 8 files changed, 189 insertions(+), 49 deletions(-) diff --git a/swarm-controller/src/config_pr.rs b/swarm-controller/src/config_pr.rs index 7c2def65..8788864f 100644 --- a/swarm-controller/src/config_pr.rs +++ b/swarm-controller/src/config_pr.rs @@ -227,9 +227,11 @@ pub fn spawn(client: Arc) -> Arc { let mut ticker = tokio::time::interval(POLL_INTERVAL); loop { ticker.tick().await; - let Some(scan) = - crate::forge::within_pass_deadline("config-pr poll", client.list_open_config_prs()) - .await + let Some(scan) = crate::forge::within_forge_deadline( + "config-pr poll", + client.list_open_config_prs(), + ) + .await else { continue; }; diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index bfcb5f88..89a9efd1 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -231,26 +231,27 @@ where } } -/// The longest one periodic forge pass may run before it is abandoned until -/// its next tick. forgejo-api's client has no request timeout, so without -/// this a forge that accepts a connection and never answers holds the pass, -/// and the loop running it, forever. Under every pass's 5-minute interval and -/// far over what a pass that is making progress takes. -pub(crate) const PASS_DEADLINE: std::time::Duration = std::time::Duration::from_mins(2); +/// The longest one unit of forge work, a periodic pass or a job node, may run +/// before it is abandoned. forgejo-api's client has no request timeout, so +/// without this a forge that accepts a connection and never answers holds the +/// pass's loop, or the node, forever. Under every pass's 5-minute interval, +/// and far over what the longest node (`MintHiveForgeToken`, about fifteen +/// sequential forge and store requests) takes while it is making progress. +pub(crate) const FORGE_DEADLINE: std::time::Duration = std::time::Duration::from_mins(2); -/// Run `fut`, one periodic pass named `pass` in the log, under -/// [`PASS_DEADLINE`]. `None` means it was abandoned at the deadline; that is -/// logged here, and the caller moves on to its next tick. -pub(crate) async fn within_pass_deadline( - pass: &str, +/// Run `fut`, the forge work named `work` in the log, under +/// [`FORGE_DEADLINE`]. `None` means it was abandoned at the deadline; that is +/// logged here. +pub(crate) async fn within_forge_deadline( + work: &str, fut: impl std::future::Future, ) -> Option { - let result = tokio::time::timeout(PASS_DEADLINE, fut).await.ok(); + let result = tokio::time::timeout(FORGE_DEADLINE, fut).await.ok(); if result.is_none() { tracing::warn!( - pass, - deadline_s = PASS_DEADLINE.as_secs(), - "forge pass: no answer within the deadline; abandoned, retrying next tick" + work, + deadline_s = FORGE_DEADLINE.as_secs(), + "forge work: no answer within the deadline; abandoned" ); } result @@ -1411,6 +1412,15 @@ fn conflict_error() -> ForgejoError { ForgejoError::ApiError(ApiErrorKind::Other(StatusCode::CONFLICT).into()) } +#[cfg(test)] +impl Client { + /// A client for a test's own forge at `url`, holding a placeholder token. + pub(crate) fn for_test_forge(url: url::Url) -> Self { + let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap(); + Self { api, url } + } +} + #[cfg(test)] mod tests { use super::*; @@ -1483,14 +1493,14 @@ mod tests { #[tokio::test(start_paused = true)] async fn a_pass_that_never_answers_is_abandoned_at_the_deadline() { let start = tokio::time::Instant::now(); - let got = within_pass_deadline("test", std::future::pending::<()>()).await; + let got = within_forge_deadline("test", std::future::pending::<()>()).await; assert_eq!(got, None); - assert_eq!(start.elapsed(), PASS_DEADLINE); + assert_eq!(start.elapsed(), FORGE_DEADLINE); } #[tokio::test(start_paused = true)] async fn a_pass_that_answers_returns_its_result() { - assert_eq!(within_pass_deadline("test", async { 7 }).await, Some(7)); + assert_eq!(within_forge_deadline("test", async { 7 }).await, Some(7)); } #[tokio::test] diff --git a/swarm-controller/src/forge/agent_token.rs b/swarm-controller/src/forge/agent_token.rs index 5f4c4fbd..ebe3991b 100644 --- a/swarm-controller/src/forge/agent_token.rs +++ b/swarm-controller/src/forge/agent_token.rs @@ -370,7 +370,7 @@ pub fn spawn(client: Arc, enqueue: impl Fn(Vec) + Send + 'static let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); loop { ticker.tick().await; - let Some(outcome) = super::within_pass_deadline("agent forge tokens", async { + let Some(outcome) = super::within_forge_deadline("agent forge tokens", async { let observed = client.observe_all().await?; let legacy = super::legacy_tokens::sweepable(&observed); client.sweep_legacy_tokens(&legacy).await; diff --git a/swarm-controller/src/forge/docs_grant.rs b/swarm-controller/src/forge/docs_grant.rs index 39810be0..435b5fea 100644 --- a/swarm-controller/src/forge/docs_grant.rs +++ b/swarm-controller/src/forge/docs_grant.rs @@ -185,7 +185,7 @@ pub fn spawn(client: Arc, enqueue: impl Fn(Vec) + Send + 'static let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); loop { ticker.tick().await; - let Some(agents) = super::within_pass_deadline("docs grants", async { + let Some(agents) = super::within_forge_deadline("docs grants", async { run_pass(&client, read_roster().await).await }) .await diff --git a/swarm-controller/src/forge/hive_token.rs b/swarm-controller/src/forge/hive_token.rs index ed7cb9fa..3ce1e39d 100644 --- a/swarm-controller/src/forge/hive_token.rs +++ b/swarm-controller/src/forge/hive_token.rs @@ -387,7 +387,7 @@ pub fn spawn( loop { ticker.tick().await; let Some(outcome) = - super::within_pass_deadline("hive forge tokens", client.observe_hives(&hives)) + super::within_forge_deadline("hive forge tokens", client.observe_hives(&hives)) .await else { continue; diff --git a/swarm-controller/src/forge/objects.rs b/swarm-controller/src/forge/objects.rs index ee562c65..ab349631 100644 --- a/swarm-controller/src/forge/objects.rs +++ b/swarm-controller/src/forge/objects.rs @@ -1083,7 +1083,7 @@ pub fn spawn(client: Arc) { loop { ticker.tick().await; let Some(outcome) = - super::within_pass_deadline("swarm forge objects", client.reconcile(&desired)) + super::within_forge_deadline("swarm forge objects", client.reconcile(&desired)) .await else { continue; diff --git a/swarm-controller/src/forge/runner_token.rs b/swarm-controller/src/forge/runner_token.rs index d34dea88..75aa255f 100644 --- a/swarm-controller/src/forge/runner_token.rs +++ b/swarm-controller/src/forge/runner_token.rs @@ -77,7 +77,7 @@ pub fn spawn(client: Arc) { let mut last_written: Option = None; loop { ticker.tick().await; - let Some(outcome) = super::within_pass_deadline( + let Some(outcome) = super::within_forge_deadline( "runner registration token", pass(&client, last_written.as_deref()), ) diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index a421bd25..32b79f2c 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -324,7 +324,7 @@ async fn run_swarm_node( not in the swarm secret store)" .to_owned(), ), - Some(client) => match client.create_repo(&agent).await { + Some(client) => match forge_node("create_repo", client.create_repo(&agent)).await { Ok(full_name) => { tracing::info!(%full_name, "swarm jobq: create_repo done"); Outcome::Done @@ -338,10 +338,12 @@ async fn run_swarm_node( not in the swarm secret store)" .to_owned(), ), - Some(client) => match client.ensure_agent_user(&agent).await { - Ok(()) => Outcome::Done, - Err(e) => Outcome::Failed(format!("{e:#}")), - }, + Some(client) => { + match forge_node("create_forge_user", client.ensure_agent_user(&agent)).await { + Ok(()) => Outcome::Done, + Err(e) => Outcome::Failed(format!("{e:#}")), + } + } }, SwarmNodeKind::AddRepoMember { agent } => match forge { None => Outcome::Failed( @@ -349,23 +351,15 @@ async fn run_swarm_node( not in the swarm secret store)" .to_owned(), ), - Some(client) => match client.add_repo_member(&agent, &agent).await { - Ok(()) => Outcome::Done, - Err(e) => Outcome::Failed(format!("{e:#}")), - }, + Some(client) => { + match forge_node("add_repo_member", client.add_repo_member(&agent, &agent)).await { + Ok(()) => Outcome::Done, + Err(e) => Outcome::Failed(format!("{e:#}")), + } + } }, 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)" - .to_owned(), - ), - Some(client) => match client.seed_agent_config(&agent, &agent).await { - Ok(()) => Outcome::Done, - Err(e) => Outcome::Failed(format!("{e:#}")), - }, - }, + SwarmNodeKind::InitAgentConfigRepo { agent } => init_agent_config_repo(forge, &agent).await, SwarmNodeKind::MintAgentIdentity { agent } => mint_identity(&agent).await, SwarmNodeKind::MintAgentForgeToken { agent } => mint_forge_token(forge, &agent).await, SwarmNodeKind::MintAgentMatrixAccount { agent } => { @@ -395,6 +389,23 @@ async fn run_swarm_node( (builder, outcome) } +/// A forge-calling node's work, run under [`forge::FORGE_DEADLINE`]. Elapsing +/// is the node's own error, naming `node`, so it fails like any other refused +/// step instead of holding its executor slot. +async fn forge_node( + node: &str, + work: impl std::future::Future>, +) -> Result { + forge::within_forge_deadline(node, work) + .await + .unwrap_or_else(|| { + Err(anyhow::anyhow!( + "{node}: no answer from the forge within {}s", + forge::FORGE_DEADLINE.as_secs() + )) + }) +} + /// The `MintAgentIdentity` arm, lifted out so `run_swarm_node` stays under /// `clippy::too_many_lines`. async fn mint_identity(agent: &str) -> hive_jobq::scheduler::Outcome { @@ -421,7 +432,12 @@ async fn mint_forge_token( .to_owned(), ); }; - match client.ensure_agent_forge_token(agent).await { + match forge_node( + "mint_agent_forge_token", + client.ensure_agent_forge_token(agent), + ) + .await + { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } @@ -441,7 +457,32 @@ async fn grant_docs_read( .to_owned(), ); }; - match client.grant_docs_read(agent).await { + match forge_node("grant_docs_read", client.grant_docs_read(agent)).await { + Ok(()) => Outcome::Done, + Err(e) => Outcome::Failed(format!("{e:#}")), + } +} + +/// The `InitAgentConfigRepo` arm, lifted out for the same reason. +async fn init_agent_config_repo( + forge: Option>, + agent: &str, +) -> hive_jobq::scheduler::Outcome { + use hive_jobq::scheduler::Outcome; + + let Some(client) = forge else { + return Outcome::Failed( + "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ + not in the swarm secret store)" + .to_owned(), + ); + }; + match forge_node( + "init_agent_config_repo", + client.seed_agent_config(agent, agent), + ) + .await + { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } @@ -471,7 +512,12 @@ async fn mint_hive_forge_token( .to_owned(), ); }; - match client.ensure_hive_forge_token(hive).await { + match forge_node( + "mint_hive_forge_token", + client.ensure_hive_forge_token(hive), + ) + .await + { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } @@ -4188,6 +4234,88 @@ mod tests { assert_eq!(queued.node_id, node.id.get()); } + /// Every node whose first request goes to the forge, run against a forge + /// that accepts the connection and never answers: each ends `Failed` + /// with an error naming the node instead of holding its executor slot. + /// `MintAgentForgeToken` is not in the list: it logs in to the secret + /// store before its first forge request, and no test here has a store. + #[tokio::test(start_paused = true)] + async fn a_forge_node_fails_when_the_forge_never_answers() { + // Bound but never accepted: the kernel completes the handshake and + // buffers the request, and no response ever comes. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); + let forge = super::Late::>::default(); + assert!( + forge + .set(std::sync::Arc::new(crate::forge::Client::for_test_forge( + url + ))) + .is_ok() + ); + let agent = || "atlas".to_owned(); + let nodes = [ + ("create_repo", SwarmNodeKind::CreateRepo { agent: agent() }), + ( + "create_forge_user", + SwarmNodeKind::CreateForgeUser { agent: agent() }, + ), + ( + "add_repo_member", + SwarmNodeKind::AddRepoMember { agent: agent() }, + ), + ( + "grant_docs_read", + SwarmNodeKind::GrantDocsRead { agent: agent() }, + ), + ( + "init_agent_config_repo", + SwarmNodeKind::InitAgentConfigRepo { agent: agent() }, + ), + ( + "mint_hive_forge_token", + SwarmNodeKind::MintHiveForgeToken { + hive: "pr1ma".to_owned(), + }, + ), + ]; + for (label, kind) in nodes { + let mut sched = hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ); + let id = sched.append(kind, Vec::new(), None).expect("insert"); + let sched = std::sync::Arc::new(std::sync::Mutex::new(sched)); + let deps = WorkerDeps { + auth: None, + forge: forge.clone(), + queue: None, + wanted: None, + matrix_homeserver: None, + }; + let runner = + hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { + run_swarm_node(id, kind, builder, deps) + }) + .expect("the node just inserted is runnable"); + tokio::time::timeout(std::time::Duration::from_mins(10), runner) + .await + .unwrap_or_else(|_| panic!("{label}: node still running after 10 minutes")) + .1 + .expect("no growth declared, nothing to reject"); + + let guard = sched.lock().unwrap(); + let node = guard.graph().node(id).expect("node still present"); + assert_eq!(node.state, hive_jobq::State::Failed, "{label}"); + let error = node.error.as_deref().unwrap_or_default(); + assert!( + error.contains(label) && error.contains("no answer from the forge"), + "{label}: expected a deadline error naming the node, got {error:?}" + ); + } + drop(listener); + } + #[tokio::test] async fn create_repo_node_runs_end_to_end_and_fails_without_forge_configured() { unsafe {