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 {