From 0599c132e579b547e60f126ea5b3728a6c75d388 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 17 Oct 2026 00:35:20 +0200 Subject: [PATCH] back off between failed broker recvs in the serve loop --- docs/turn-loop/README.md | 5 +- hive-agent/src/main.rs | 191 ++++++++++++++++++++++++++++++++++++--- 2 files changed, 180 insertions(+), 16 deletions(-) diff --git a/docs/turn-loop/README.md b/docs/turn-loop/README.md index b686c942..8d8da6f6 100644 --- a/docs/turn-loop/README.md +++ b/docs/turn-loop/README.md @@ -31,7 +31,10 @@ agents) runs: 1. Long-poll `Recv` on its socket. The host-side broker (`broker.rs::recv_blocking_batch`) returns immediately if there's a pending message, otherwise waits up to 30 s for a broker `Sent` - event for this recipient. + event for this recipient. After a failed `Recv` (a socket error, or + a broker error, which comes back at once) the loop waits before the + next one: 250 ms, doubling up to 30 s, and no wait again once a + `Recv` succeeds. Todo wakes and local messages don't wait for it. 2. Pop one message. Peek the remaining inbox depth with `Status`. 3. Emit `LiveEvent::TurnStart { from, body, unread }` onto the SSE bus. diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index e857f517..81108225 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -65,6 +65,14 @@ const RECV_WAIT: Duration = Duration::from_mins(3); /// hive-c0re. const RECV_RESPONSE_TIMEOUT: Duration = RECV_WAIT.saturating_add(Duration::from_secs(30)); +/// Wait before the `Recv` after a failed one. A broker error comes back at +/// once rather than after the long poll, so without a wait the serve loop +/// would retry as fast as hive-c0re answers. +const RECV_RETRY_FLOOR: Duration = Duration::from_millis(250); + +/// Ceiling for that wait as it doubles over consecutive failures. +const RECV_RETRY_CAP: Duration = Duration::from_secs(30); + /// Default web UI port — used when `HIVE_PORT` env is unset. const DEFAULT_WEB_PORT: u16 = 8042; @@ -627,6 +635,65 @@ fn share_session_handles(bus: &Bus, session: &turn::AgentSession) { } } +/// When the serve loop may send its next `Recv`: at once while `Recv`s +/// succeed, and after each failed one [`RECV_RETRY_FLOOR`] later, doubling +/// up to [`RECV_RETRY_CAP`]. One that succeeds resets it. +#[derive(Default)] +struct RecvBackoff { + /// Failed `Recv`s since the last one that succeeded. + failures: u32, + /// No `Recv` goes out before this. A deadline rather than a duration, + /// so a local wake that cancels the wait doesn't restart it. + retry_at: Option, +} + +impl RecvBackoff { + /// The wait after the `failures`-th failure in a row. + fn delay(failures: u32) -> Duration { + RECV_RETRY_FLOOR + .saturating_mul(2u32.saturating_pow(failures.saturating_sub(1))) + .min(RECV_RETRY_CAP) + } + + /// Wait out the backoff, run `recv`, and record whether it failed. + async fn recv(&mut self, recv: impl Future) -> RecvOutcome { + if let Some(at) = self.retry_at { + tokio::time::sleep_until(at).await; + } + let outcome = recv.await; + if matches!(outcome, RecvOutcome::TransportError) { + self.failures = self.failures.saturating_add(1); + self.retry_at = Some(tokio::time::Instant::now() + Self::delay(self.failures)); + } else { + if self.failures > 0 { + tracing::info!(failures = self.failures, "recv succeeded again"); + } + *self = Self::default(); + } + outcome + } +} + +/// The serve loop's idle wait: race the broker `Recv`, held back by +/// `backoff`, against a local todo signal and local messages, so an +/// in-container producer drives a turn without any broker round-trip and +/// isn't held up by a broker that is failing. `biased` polls the broker +/// recv first, so a genuinely-ready inbox message is never dropped in +/// favour of the todo wake. +async fn next_outcome( + recv: impl Future, + backoff: &mut RecvBackoff, + todo_wake: &tokio::sync::Notify, + local_msg_rx: &mut tokio::sync::mpsc::UnboundedReceiver, +) -> RecvOutcome { + tokio::select! { + biased; + o = backoff.recv(recv) => o, + () = todo_wake.notified() => RecvOutcome::LocalTodo, + Some(dm) = local_msg_rx.recv() => RecvOutcome::Message(dm), + } +} + /// The long-running message loop. Long-polls the broker via /// `S::recv_next`, drives a turn per message, parks on auth-failed, /// otherwise retries. @@ -664,6 +731,7 @@ async fn serve_loop( // too, so a just-unparked agent gets a clean slate rather than being // one miss away from an instant re-pause. let mut todo_miss_streak: u32 = 0; + let mut recv_backoff = RecvBackoff::default(); loop { // Pause gate. While the marker is present this loop drives no // turns at all. @@ -704,19 +772,14 @@ async fn serve_loop( was_paused = false; todo_miss_streak = 0; } - let next = match { - // Idle wait: race the broker long-poll against a local - // todo signal so an in-container producer's upsert drives a - // turn without any broker round-trip. `biased` polls the - // broker recv first, so a genuinely-ready inbox message is - // never dropped in favour of the todo wake. - tokio::select! { - biased; - o = S::recv_next(socket) => o, - () = todo_wake.notified() => RecvOutcome::LocalTodo, - Some(dm) = local_msg_rx.recv() => RecvOutcome::Message(dm), - } - } { + let next = match next_outcome( + S::recv_next(socket), + &mut recv_backoff, + &todo_wake, + &mut local_msg_rx, + ) + .await + { RecvOutcome::Message(first) => first, RecvOutcome::LocalTodo => { // Gate on `has_any()` before spawning a turn: a burst of @@ -762,8 +825,8 @@ async fn serve_loop( post_compact_wake_message(prompt) } RecvOutcome::TransportError => { - // `recv_next` already logged the detail; just retry. - // No backoff: the long-poll wait is itself the throttle. + // `recv_next` already logged the detail; `recv_backoff` + // holds the next `Recv` back. continue; } RecvOutcome::GracefulStop => { @@ -1088,3 +1151,101 @@ async fn handle_turn_error_recovery( S::requeue_inflight(socket).await; } } + +#[cfg(test)] +mod tests { + use super::{RECV_RETRY_CAP, RecvBackoff, RecvOutcome, next_outcome}; + use std::time::Duration; + use tokio::time::Instant; + + /// Run one `Recv` through `backoff` that answers `outcome`, and return + /// when it went out. + async fn recv_at(backoff: &mut RecvBackoff, outcome: RecvOutcome) -> Instant { + let mut sent = None; + backoff + .recv(async { + sent = Some(Instant::now()); + outcome + }) + .await; + sent.expect("the recv ran") + } + + fn gaps(sent: &[Instant]) -> Vec { + sent.windows(2).map(|w| w[1] - w[0]).collect() + } + + #[tokio::test(start_paused = true)] + async fn failed_recvs_back_off_doubling_up_to_the_cap() { + let mut backoff = RecvBackoff::default(); + let mut sent = Vec::new(); + for _ in 0..10 { + sent.push(recv_at(&mut backoff, RecvOutcome::TransportError).await); + } + let ms = Duration::from_millis; + assert_eq!( + gaps(&sent), + [ + ms(250), + ms(500), + ms(1000), + ms(2000), + ms(4000), + ms(8000), + ms(16000), + RECV_RETRY_CAP, + RECV_RETRY_CAP, + ] + ); + } + + #[tokio::test(start_paused = true)] + async fn a_successful_recv_resets_the_backoff() { + let mut backoff = RecvBackoff::default(); + for _ in 0..4 { + recv_at(&mut backoff, RecvOutcome::TransportError).await; + } + let sent = vec![ + recv_at(&mut backoff, RecvOutcome::Empty).await, + recv_at(&mut backoff, RecvOutcome::TransportError).await, + recv_at(&mut backoff, RecvOutcome::Empty).await, + ]; + assert_eq!(gaps(&sent), [Duration::ZERO, Duration::from_millis(250)]); + } + + /// A todo wake or a local message during a backoff is taken at once, + /// without sending the held-back `Recv`; with no backoff the `Recv` + /// goes first. + #[tokio::test(start_paused = true)] + async fn a_local_wake_does_not_wait_out_the_backoff() { + let todo_wake = tokio::sync::Notify::new(); + let (local_tx, mut local_rx) = tokio::sync::mpsc::unbounded_channel(); + let mut backoff = RecvBackoff::default(); + for _ in 0..10 { + recv_at(&mut backoff, RecvOutcome::TransportError).await; + } + let retry_at = backoff.retry_at; + let held_back = || async { panic!("recv sent inside its backoff") }; + + let start = Instant::now(); + todo_wake.notify_one(); + let woke = next_outcome(held_back(), &mut backoff, &todo_wake, &mut local_rx).await; + assert!(matches!(woke, RecvOutcome::LocalTodo)); + local_tx.send(super::synthetic_todo_message(false)).unwrap(); + let woke = next_outcome(held_back(), &mut backoff, &todo_wake, &mut local_rx).await; + assert!(matches!(woke, RecvOutcome::Message(_))); + assert_eq!(start.elapsed(), Duration::ZERO); + assert_eq!(backoff.retry_at, retry_at); + + let mut backoff = RecvBackoff::default(); + todo_wake.notify_one(); + let woke = next_outcome( + async { RecvOutcome::Empty }, + &mut backoff, + &todo_wake, + &mut local_rx, + ) + .await; + assert!(matches!(woke, RecvOutcome::Empty)); + } +}