Watch
0
0
Fork
You've already forked hyperhive
0

swarm-controller: bound each periodic forge pass with a deadline

forgejo-api 0.11.1 builds its reqwest client with no timeout and offers
no way to hand it a caller-built one (the `client` field is private,
`new`/`with_user_agent` are the only constructors, `Endpoint` is
sealed). A forge that accepts a connection and then stalls therefore
held each of the six periodic reconcile loops on one request forever,
with no log line.

Wrap each loop's per-tick work in `forge::within_pass_deadline`
(`PASS_DEADLINE`, 2 min, under the 5-minute interval): on elapsed the
pass is dropped, a warn is logged, and the loop waits for its next tick.
Every pass is an idempotent reconcile, so abandoning one mid-way leaves
nothing the next pass does not redo.

Loops: config-pr poll, runner registration token, hive forge tokens,
agent forge tokens (observe + legacy sweep), docs grants, swarm forge
objects.

Refs #5060
This commit is contained in:
atlas 2026-10-09 14:49:50 +02:00 • committed by mara
commit 45f0582163
7 changed files with 85 additions and 9 deletions

View file

@ -227,7 +227,13 @@ pub fn spawn(client: Arc<Client>) -> Arc<ConfigPrCache> {
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!(

View file

@ -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<T>(
pass: &str,
fut: impl std::future::Future<Output = T>,
) -> Option<T> {
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)]

View file

@ -370,11 +370,18 @@ pub fn spawn(client: Arc<Client>, enqueue: impl Fn(Vec<String>) + 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!(

View file

@ -185,7 +185,13 @@ pub fn spawn(client: Arc<Client>, enqueue: impl Fn(Vec<String>) + 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);
}

View file

@ -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() {

View file

@ -1082,7 +1082,12 @@ pub fn spawn(client: Arc<Client>) {
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,

View file

@ -77,7 +77,15 @@ pub fn spawn(client: Arc<Client>) {
let mut last_written: Option<String> = 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);