vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

retry: replace three hand-rolled doubling loops with backon

Browse source
This commit is contained in:
damocles 2026-10-17 09:23:46 +02:00 • committed by mara
commit 0f67ec34b1
7 changed files with 141 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -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",

1
Cargo.toml
View file

@ -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",

1
hive-agent/Cargo.toml
View file

@ -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

45
hive-agent/src/swarm_agent_icon.rs
View file

@ -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);
}
}

99
hive-agent/src/web_ui/mod.rs
View file

@ -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<tokio::net::TcpListener> {
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::<std::io::Error>()
.map(std::io::Error::kind),
Some(std::io::ErrorKind::AddrInUse)
);
drop(holder);
}
fn form(path: &str, body: &'static str) -> Request<Body> {
Request::post(path)
.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")

1
hive-c0re/Cargo.toml
View file

@ -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

75
hive-c0re/src/dashboard/mod.rs
View file

@ -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<Coordinator>) -> 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<tokio::net::TcpListener> {
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<tokio::net::TcpListener> {
@ -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);