From 0f67ec34b14d54f63f7bd54bb42230f051878ed0 Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 17 Oct 2026 09:23:46 +0200 Subject: [PATCH] retry: replace three hand-rolled doubling loops with backon --- Cargo.lock | 2 + Cargo.toml | 1 + hive-agent/Cargo.toml | 1 + hive-agent/src/swarm_agent_icon.rs | 45 ++++++++------ hive-agent/src/web_ui/mod.rs | 99 ++++++++++++++++++++---------- hive-c0re/Cargo.toml | 1 + hive-c0re/src/dashboard/mod.rs | 75 ++++++++++++---------- 7 files changed, 141 insertions(+), 83 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c830a4f9..2d9b9336 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1664,6 +1664,7 @@ dependencies = [ "anyhow", "async-nats", "axum", + "backon", "chrono", "clap", "futures-util", @@ -1762,6 +1763,7 @@ dependencies = [ "anyhow", "async-nats", "axum", + "backon", "base64", "chrono", "clap", diff --git a/Cargo.toml b/Cargo.toml index b9a22798..c8e7968e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -81,6 +81,7 @@ unescaped_backticks = "deny" anyhow = "1" libc = "0.2" axum = { version = "0.8", features = ["ws"] } +backon = "1" base64 = "0.22" chrono = { version = "0.4", default-features = false, features = [ "clock", diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 725630c3..b31612c3 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -14,6 +14,7 @@ anyhow.workspace = true # handling live in `swarm-queue-client` below. async-nats = { workspace = true, features = ["jetstream"] } axum.workspace = true +backon.workspace = true chrono.workspace = true reqwest.workspace = true hyper.workspace = true diff --git a/hive-agent/src/swarm_agent_icon.rs b/hive-agent/src/swarm_agent_icon.rs index 45f8d458..a752fda8 100644 --- a/hive-agent/src/swarm_agent_icon.rs +++ b/hive-agent/src/swarm_agent_icon.rs @@ -22,6 +22,7 @@ use std::path::Path; use std::time::Duration; use async_nats::jetstream::context::PublishErrorKind; +use backon::{ExponentialBuilder, Retryable}; /// Where the agent's NixOS config puts the icon /// (`services.hyperhive.agent.icon`), and the file `GET /icon` serves. Absent @@ -29,7 +30,7 @@ use async_nats::jetstream::context::PublishErrorKind; const ICON_PATH: &str = "/etc/hyperhive/icon.svg"; /// First retry delay after a failed write, doubled per failure up to -/// [`MAX_RETRY`]. +/// [`MAX_RETRY`], with no cap on the number of tries. const FIRST_RETRY: Duration = Duration::from_secs(5); /// Longest wait between retries. @@ -80,27 +81,33 @@ async fn run(agent: String) { } }; let js = async_nats::jetstream::new(client.clone()); - let mut delay = FIRST_RETRY; - loop { - match write(&client, &js, &subject, &icon).await { - Ok(()) => { - tracing::info!( - subject, - set = matches!(icon, Icon::Set(_)), - "published this agent's icon" - ); - return; - } - Err(Failure::TooLarge(e)) => { - tracing::warn!(error = %e, "the agent icon is over the queue's payload limit; not publishing it"); - return; - } - Err(Failure::Retry(e)) => { + let written = (|| write(&client, &js, &subject, &icon)) + .retry( + ExponentialBuilder::default() + .with_min_delay(FIRST_RETRY) + .with_max_delay(MAX_RETRY) + .without_max_times(), + ) + .when(|failure| matches!(failure, Failure::Retry(_))) + .notify(|failure, delay| { + if let Failure::Retry(e) = failure { tracing::warn!(error = e, retry_in = ?delay, "publishing the agent icon failed"); } + }) + .await; + match written { + Ok(()) => tracing::info!( + subject, + set = matches!(icon, Icon::Set(_)), + "published this agent's icon" + ), + Err(Failure::TooLarge(e)) => { + tracing::warn!(error = %e, "the agent icon is over the queue's payload limit; not publishing it"); + } + // Only a `TooLarge` ends the retries. + Err(Failure::Retry(e)) => { + tracing::error!(error = e, "gave up publishing the agent icon"); } - tokio::time::sleep(delay).await; - delay = (delay * 2).min(MAX_RETRY); } } diff --git a/hive-agent/src/web_ui/mod.rs b/hive-agent/src/web_ui/mod.rs index a9fc5e7f..dc4b2331 100644 --- a/hive-agent/src/web_ui/mod.rs +++ b/hive-agent/src/web_ui/mod.rs @@ -21,6 +21,7 @@ use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::sync::atomic::AtomicBool; use std::sync::{Arc, Mutex}; +use std::time::Duration; use anyhow::{Context, Result}; use axum::{ @@ -29,6 +30,7 @@ use axum::{ response::{IntoResponse, Response}, routing::{get, post}, }; +use backon::{ExponentialBuilder, Retryable}; use tower_http::services::ServeDir; use crate::events::Bus; @@ -221,38 +223,34 @@ const MAX_BIND_ATTEMPTS: u32 = 12; /// Retry rationale + dashboard-banner-on-real-collision: /// see `docs/web-ui/shape.md::Listener bind`. async fn bind_with_retry(addr: SocketAddr, label: &str) -> Result { - let mut delay_ms = 250u64; - let mut attempts = 0u32; - loop { - match try_bind(addr) { - Ok(l) => { - if attempts > 0 { - tracing::info!( - %addr, attempts, - "{label}: bind succeeded after retry" - ); - } - return Ok(l); - } - Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => { - let attempt = attempts + 1; - if attempt >= MAX_BIND_ATTEMPTS { - return Err(e).with_context(|| { - format!("bind {label} on {addr}: still AddrInUse after {attempt} attempts") - }); - } - tracing::warn!( - %addr, attempt, - "{label}: AddrInUse, retrying in {delay_ms}ms" - ); - tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; - attempts += 1; - delay_ms = (delay_ms * 2).min(2000); - } - Err(e) => { - return Err(e).with_context(|| format!("bind {label} on {addr}")); + let mut retries = 0u32; + let bound = (|| async { try_bind(addr) }) + .retry( + ExponentialBuilder::default() + .with_min_delay(Duration::from_millis(250)) + .with_max_delay(Duration::from_secs(2)) + .with_max_times((MAX_BIND_ATTEMPTS - 1) as usize), + ) + .when(|e| e.kind() == std::io::ErrorKind::AddrInUse) + .notify(|_, delay| { + retries += 1; + tracing::warn!( + %addr, attempt = retries, + "{label}: AddrInUse, retrying in {}ms", delay.as_millis() + ); + }) + .await; + match bound { + Ok(l) => { + if retries > 0 { + tracing::info!(%addr, attempts = retries, "{label}: bind succeeded after retry"); } + Ok(l) } + Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => Err(e).with_context(|| { + format!("bind {label} on {addr}: still AddrInUse after {MAX_BIND_ATTEMPTS} attempts") + }), + Err(e) => Err(e).with_context(|| format!("bind {label} on {addr}")), } } @@ -484,10 +482,49 @@ mod tests { use axum::http::{Request, StatusCode, header}; use tower::ServiceExt; - use super::{AppState, SigintOutcome, routes, sigint_turn}; + use super::{AppState, SigintOutcome, bind_with_retry, routes, sigint_turn, try_bind}; use crate::events::Bus; use crate::login::LoginState; + /// A port another listener holds for a moment is waited out. + #[tokio::test(start_paused = true)] + async fn bind_waits_out_a_port_that_is_briefly_held() { + let holder = try_bind("127.0.0.1:0".parse().expect("addr")).expect("holder"); + let addr = holder.local_addr().expect("local addr"); + let release = tokio::spawn(async move { + tokio::time::sleep(Duration::from_secs(1)).await; + drop(holder); + }); + let bound = bind_with_retry(addr, "test") + .await + .expect("bind after release"); + assert_eq!(bound.local_addr().expect("local addr"), addr); + release.await.expect("release task"); + } + + /// A port that stays held ends in an `AddrInUse` error after 12 attempts: + /// 11 waits of 250ms doubling to the 2s cap. + #[tokio::test(start_paused = true)] + async fn bind_gives_up_when_the_port_stays_held() { + let holder = try_bind("127.0.0.1:0".parse().expect("addr")).expect("holder"); + let addr = holder.local_addr().expect("local addr"); + let start = tokio::time::Instant::now(); + let err = bind_with_retry(addr, "test") + .await + .expect_err("port is held"); + assert_eq!( + start.elapsed(), + Duration::from_millis(250 + 500 + 1000 + 8 * 2000) + ); + assert!(err.to_string().contains("after 12 attempts"), "{err:#}"); + assert_eq!( + err.downcast_ref::() + .map(std::io::Error::kind), + Some(std::io::ErrorKind::AddrInUse) + ); + drop(holder); + } + fn form(path: &str, body: &'static str) -> Request { Request::post(path) .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded") diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 5fe285be..423c5c9f 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -16,6 +16,7 @@ anyhow.workspace = true # around; the connect itself lives in `swarm-queue-client` below. async-nats.workspace = true axum.workspace = true +backon.workspace = true chrono.workspace = true base64.workspace = true reqwest.workspace = true diff --git a/hive-c0re/src/dashboard/mod.rs b/hive-c0re/src/dashboard/mod.rs index b5234a37..fa11d40c 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -3,6 +3,7 @@ use std::net::SocketAddr; use std::sync::Arc; +use std::time::Duration; use anyhow::{Context, Result}; use axum::{ @@ -11,6 +12,7 @@ use axum::{ response::{IntoResponse, Response}, routing::get, }; +use backon::{ExponentialBuilder, Retryable}; use utoipa::OpenApi; use utoipa_axum::{router::OpenApiRouter, routes}; @@ -166,41 +168,33 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { /// `SO_REUSEADDR` bind with retry. Retry mechanics, attempt-cap /// rationale, and log-level cadence: `docs/web-ui/shape.md::Listener bind`. async fn bind_with_retry(addr: SocketAddr) -> Result { - let mut delay_ms = 250u64; - let mut attempts = 0u32; - loop { - match try_bind(addr) { - Ok(l) => { - if attempts > 0 { - tracing::info!( - %addr, attempts, - "dashboard: bind succeeded after retry" - ); - } - return Ok(l); + let mut retries = 0u32; + let listener = (|| async { try_bind(addr) }) + .retry( + ExponentialBuilder::default() + .with_min_delay(Duration::from_millis(250)) + .with_max_delay(Duration::from_secs(2)) + .without_max_times(), + ) + .when(|e| e.kind() == std::io::ErrorKind::AddrInUse) + .notify(|_, delay| { + retries += 1; + let ms = delay.as_millis(); + if retries <= 12 { + tracing::warn!(%addr, attempt = retries, "dashboard: AddrInUse, retrying in {ms}ms"); + } else { + tracing::info!( + %addr, attempt = retries, + "dashboard: AddrInUse still holding, retrying in {ms}ms" + ); } - Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => { - let attempt = attempts + 1; - if attempt <= 12 { - tracing::warn!( - %addr, attempt, - "dashboard: AddrInUse, retrying in {delay_ms}ms" - ); - } else { - tracing::info!( - %addr, attempt, - "dashboard: AddrInUse still holding, retrying in {delay_ms}ms" - ); - } - tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; - attempts += 1; - delay_ms = (delay_ms * 2).min(2000); - } - Err(e) => { - return Err(e).with_context(|| format!("bind dashboard on {addr}")); - } - } + }) + .await + .with_context(|| format!("bind dashboard on {addr}"))?; + if retries > 0 { + tracing::info!(%addr, attempts = retries, "dashboard: bind succeeded after retry"); } + Ok(listener) } fn try_bind(addr: SocketAddr) -> std::io::Result { @@ -312,6 +306,21 @@ mod tests { .unwrap_or_else(|| panic!("no status line in {reply:?}")) } + /// The dashboard never gives up on a held port: a holder that lets go well + /// after the attempt count where the log level drops still gets waited out. + #[tokio::test(start_paused = true)] + async fn bind_keeps_retrying_until_the_port_is_free() { + let holder = try_bind("127.0.0.1:0".parse().expect("addr")).expect("holder"); + let addr = holder.local_addr().expect("local addr"); + let release = tokio::spawn(async move { + tokio::time::sleep(Duration::from_mins(5)).await; + drop(holder); + }); + let bound = bind_with_retry(addr).await.expect("bind after release"); + assert_eq!(bound.local_addr().expect("local addr"), addr); + release.await.expect("release task"); + } + #[tokio::test] async fn operator_inbox_route_is_gone() { assert_eq!(status_of("/health/live").await, 200);