back off between failed broker recvs in the serve loop

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.

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<tokio::time::Instant>,
+}
+
+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<Output = RecvOutcome>) -> 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<Output = RecvOutcome>,
+    backoff: &mut RecvBackoff,
+    todo_wake: &tokio::sync::Notify,
+    local_msg_rx: &mut tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::inbox::DeliveredMessage>,
+) -> 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<S: Surface>(
     // 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<S: Surface>(
             was_paused = false;
             todo_miss_streak = 0;
         }
-        let next = match {
+        let next = match next_outcome(
-            // Idle wait: race the broker long-poll against a local
+            S::recv_next(socket),
-            // todo signal so an in-container producer's upsert drives a
+            &mut recv_backoff,
-            // turn without any broker round-trip. `biased` polls the
+            &todo_wake,
-            // broker recv first, so a genuinely-ready inbox message is
+            &mut local_msg_rx,
-            // never dropped in favour of the todo wake.
+        )
-            tokio::select! {
+        .await
-                biased;
+        {
-                o = S::recv_next(socket) => o,
-                () = todo_wake.notified() => RecvOutcome::LocalTodo,
-                Some(dm) = local_msg_rx.recv() => RecvOutcome::Message(dm),
-            }
-        } {
             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<S: Surface>(
                 post_compact_wake_message(prompt)
             }
             RecvOutcome::TransportError => {
-                // `recv_next` already logged the detail; just retry.
+                // `recv_next` already logged the detail; `recv_backoff`
-                // No backoff: the long-poll wait is itself the throttle.
+                // holds the next `Recv` back.
                 continue;
             }
             RecvOutcome::GracefulStop => {
@@ -1088,3 +1151,101 @@ async fn handle_turn_error_recovery<S: Surface>(
         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<Duration> {
+        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));
+    }
+}