diff --git a/swarm-controller/src/config_pr.rs b/swarm-controller/src/config_pr.rs index 2d772d7a..7c2def65 100644 --- a/swarm-controller/src/config_pr.rs +++ b/swarm-controller/src/config_pr.rs @@ -227,7 +227,13 @@ pub fn spawn(client: Arc) -> Arc { let mut ticker = tokio::time::interval(POLL_INTERVAL); loop { ticker.tick().await; - match client.list_open_config_prs().await { + let Some(scan) = + crate::forge::within_pass_deadline("config-pr poll", client.list_open_config_prs()) + .await + else { + continue; + }; + match scan { Ok(scan) => cache_for_task.replace(scan), Err(e) => { tracing::warn!( diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index b9d7892a..bfcb5f88 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -231,6 +231,31 @@ 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); + +/// 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, + fut: impl std::future::Future, +) -> Option { + let result = tokio::time::timeout(PASS_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" + ); + } + result +} + /// 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 @@ -1455,6 +1480,19 @@ mod tests { assert_eq!(start.elapsed().as_secs(), 30 + 60 + 120); } + #[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; + assert_eq!(got, None); + assert_eq!(start.elapsed(), PASS_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)); + } + #[tokio::test] async fn page_search_results_pages_past_a_full_first_page() { #[derive(Clone)] diff --git a/swarm-controller/src/forge/agent_token.rs b/swarm-controller/src/forge/agent_token.rs index b18d7451..5f4c4fbd 100644 --- a/swarm-controller/src/forge/agent_token.rs +++ b/swarm-controller/src/forge/agent_token.rs @@ -370,11 +370,18 @@ pub fn spawn(client: Arc, enqueue: impl Fn(Vec) + Send + 'static let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); loop { ticker.tick().await; - match client.observe_all().await { + let Some(outcome) = super::within_pass_deadline("agent forge tokens", async { + let observed = client.observe_all().await?; + let legacy = super::legacy_tokens::sweepable(&observed); + client.sweep_legacy_tokens(&legacy).await; + anyhow::Ok(observed) + }) + .await + else { + continue; + }; + match outcome { Ok(observed) => { - let legacy = super::legacy_tokens::sweepable(&observed); - client.sweep_legacy_tokens(&legacy).await; - let agents = plan(&observed); if agents.is_empty() { tracing::debug!( diff --git a/swarm-controller/src/forge/docs_grant.rs b/swarm-controller/src/forge/docs_grant.rs index 6595cb1d..39810be0 100644 --- a/swarm-controller/src/forge/docs_grant.rs +++ b/swarm-controller/src/forge/docs_grant.rs @@ -185,7 +185,13 @@ pub fn spawn(client: Arc, enqueue: impl Fn(Vec) + Send + 'static let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); loop { ticker.tick().await; - let agents = run_pass(&client, read_roster().await).await; + let Some(agents) = super::within_pass_deadline("docs grants", async { + run_pass(&client, read_roster().await).await + }) + .await + else { + continue; + }; if !agents.is_empty() { enqueue(agents); } diff --git a/swarm-controller/src/forge/hive_token.rs b/swarm-controller/src/forge/hive_token.rs index d2e0867f..ed7cb9fa 100644 --- a/swarm-controller/src/forge/hive_token.rs +++ b/swarm-controller/src/forge/hive_token.rs @@ -386,7 +386,13 @@ pub fn spawn( let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); loop { ticker.tick().await; - match client.observe_hives(&hives).await { + let Some(outcome) = + super::within_pass_deadline("hive forge tokens", client.observe_hives(&hives)) + .await + else { + continue; + }; + match outcome { Ok((team, observed)) => { let queued = plan(team, &observed); if queued.is_empty() { diff --git a/swarm-controller/src/forge/objects.rs b/swarm-controller/src/forge/objects.rs index 1b98f3d0..ee562c65 100644 --- a/swarm-controller/src/forge/objects.rs +++ b/swarm-controller/src/forge/objects.rs @@ -1082,7 +1082,12 @@ pub fn spawn(client: Arc) { let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); loop { ticker.tick().await; - let outcome = client.reconcile(&desired).await; + let Some(outcome) = + super::within_pass_deadline("swarm forge objects", client.reconcile(&desired)) + .await + else { + continue; + }; if outcome.failed > 0 { tracing::warn!( applied = outcome.applied, diff --git a/swarm-controller/src/forge/runner_token.rs b/swarm-controller/src/forge/runner_token.rs index ff04dbb0..d34dea88 100644 --- a/swarm-controller/src/forge/runner_token.rs +++ b/swarm-controller/src/forge/runner_token.rs @@ -77,7 +77,15 @@ pub fn spawn(client: Arc) { let mut last_written: Option = None; loop { ticker.tick().await; - match pass(&client, last_written.as_deref()).await { + let Some(outcome) = super::within_pass_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);