fix(acp): pace targeted overflow recovery on consumer capacity

crates/buzz-acp/README.md

@@ -270,6 +270,10 @@ Forum event kinds:
 4. **Prompting** — When events are pending and no prompt is in flight for that channel, drains all queued events for the oldest channel into a single batched prompt via ACP `session/prompt`.
 5. **Agent response** — The agent processes the prompt and uses the Buzz CLI (`send_message`, `get_messages`, etc.) to interact with Buzz.
 6. **Recovery** — If the agent crashes, the harness respawns it. If the relay disconnects, the harness reconnects with a `since` filter to avoid missing events.
+   If the inbound queue overflows, the harness attempts replay for affected
+   subscriptions when capacity and relay quota permit, with at least five seconds
+   between attempts. Recovery depends on available relay history and the consumer
+   making progress; complete delivery is not guaranteed.
 
 Each channel has at most one prompt in flight. Multiple channels can be processed concurrently when agents > 1.
 

crates/buzz-acp/src/relay.rs

@@ -1160,10 +1160,8 @@ struct BgState {
     /// On reconnect resubscribe, `since` = min(last_seen, channel_dropped_since).
     /// Cleared per-channel after a successful resubscribe.
     channel_dropped_since: HashMap<Uuid, u64>,
-    /// Set by the backpressure handler when the event channel is full.
-    /// The main loop checks this flag and triggers a proactive resubscribe
-    /// (without waiting for a disconnect) so dropped events are replayed.
-    proactive_resubscribe_needed: bool,
+    /// Rate/fairness bookkeeping only; replay cursors retain baseline semantics.
+    recovery: recovery::RecoverySchedule,
     /// Unix timestamp captured just before the relay connection was established.
     /// Used as the floor `since` for membership notification replay so events
     /// predating this session are never re-delivered.
@@ -1237,7 +1235,7 @@ impl BgState {
             membership_sub_active: false,
             observer_control_sub_active: false,
             channel_dropped_since: HashMap::new(),
-            proactive_resubscribe_needed: false,
+            recovery: recovery::RecoverySchedule::default(),
             startup_watermark: None,
             subscribe_since: HashMap::new(),
             rate_limit_gate: None,
@@ -1298,6 +1296,9 @@ impl BgState {
     /// Prevents stale replay on re-subscribe and avoids unbounded state growth
     /// for channels that are removed and never re-added.
     fn clear_channel_state(&mut self, channel_id: &Uuid) {
+        self.recovery
+            .last_attempt
+            .remove(&channel_sub_id(*channel_id));
         self.last_seen.remove(channel_id);
         self.subscribe_since.remove(channel_id);
         self.channel_dropped_since.remove(channel_id);
@@ -1826,82 +1827,6 @@ async fn run_background_task(
     let mut drain_pacing_next: Option<tokio::time::Instant> = None;
 
     loop {
-        if state.proactive_resubscribe_needed {
-            state.proactive_resubscribe_needed = false;
-            info!("proactive resubscribe triggered by backpressure event loss");
-            // Proactive resubscribe runs on the EXISTING socket — do NOT clear the
-            // rate-limit gate or pending queues.
-            match resubscribe_after_reconnect(
-                &mut ws,
-                &mut cmd_rx,
-                &mut state,
-                &agent_pubkey_hex,
-                false, // existing socket — preserve gate state
-            )
-            .await
-            {
-                ResubscribeResult::Ok => {}
-                ResubscribeResult::Shutdown => return,
-                ResubscribeResult::RetryConnection => {
-                    warn!("proactive resubscribe had failures — triggering reconnect");
-                    let _ = event_tx.try_send(None);
-                    match try_autonomous_reconnect(
-                        &mut ws,
-                        &mut cmd_rx,
-                        &mut state,
-                        &keys,
-                        &relay_url,
-                        &agent_pubkey_hex,
-                        &event_tx,
-                        &observer_control_tx,
-                        auth_tag.as_ref(),
-                    )
-                    .await
-                    {
-                        ReconnectOutcome::Ok => {
-                            if matches!(
-                                drain_post_reconnect(
-                                    &mut ws,
-                                    &mut cmd_rx,
-                                    &mut state,
-                                    &agent_pubkey_hex
-                                )
-                                .await,
-                                ReconnectOutcome::Shutdown
-                            ) {
-                                return;
-                            }
-                        }
-                        ReconnectOutcome::Shutdown => return,
-                        ReconnectOutcome::Failed => {
-                            if matches!(
-                                wait_for_reconnect(
-                                    &mut ws,
-                                    &mut cmd_rx,
-                                    &mut state,
-                                    &keys,
-                                    &relay_url,
-                                    &agent_pubkey_hex,
-                                    &event_tx,
-                                    &observer_control_tx,
-                                    true,
-                                    auth_tag.as_ref(),
-                                )
-                                .await,
-                                ReconnectOutcome::Shutdown
-                            ) {
-                                return;
-                            }
-                        }
-                    }
-                    ping_sent = false;
-                    last_pong = Instant::now();
-                    connected_since = Instant::now();
-                    stable_logged = false;
-                }
-            }
-        }
-
         // Drain pending subs, one REQ per pacing tick within the relay's
         // admission window.
         let drain_window_open = drain_pacing_next.is_none_or(|t| tokio::time::Instant::now() >= t);
@@ -1982,7 +1907,11 @@ async fn run_background_task(
             }
         }
 
+        let recovery_at = recovery::ready_at(&mut state);
         tokio::select! {
+                   _ = recovery::ready(&event_tx, recovery_at) => {
+                       recovery::recover_one(&mut ws, &mut state, &event_tx, &agent_pubkey_hex).await;
+                   }
                    raw = ws.next() => {
                        // Determine if the socket is lost.
                        let socket_lost = match raw {
@@ -2358,12 +2287,10 @@ async fn handle_ws_message(
                                 // replay starts early enough to re-deliver it.
                                 state.membership_dropped_since =
                                     Some(state.membership_dropped_since.map_or(ts, |d| d.min(ts)));
-                                // Proactively trigger resubscribe without waiting for a disconnect.
-                                state.proactive_resubscribe_needed = true;
                                 warn!(
                                     channel_id = %channel_uuid,
                                     ts,
-                                    "membership notification dropped (backpressure) — proactive resubscribe queued"
+                                    "membership notification dropped (backpressure) — targeted recovery pending"
                                 );
                             }
                             Err(mpsc::error::TrySendError::Closed(_)) => return false,
@@ -2400,12 +2327,10 @@ async fn handle_ws_message(
                                         .entry(channel_id)
                                         .and_modify(|d| *d = (*d).min(ts))
                                         .or_insert(ts);
-                                    // Proactively trigger resubscribe without waiting for a disconnect.
-                                    state.proactive_resubscribe_needed = true;
                                     warn!(
                                         channel_id = %channel_id,
                                         ts,
-                                        "event channel full — dropping event for channel {channel_id} — proactive resubscribe queued"
+                                        "event channel full — dropping event for channel {channel_id} — targeted recovery pending"
                                     );
                                 }
                                 Err(mpsc::error::TrySendError::Closed(_)) => {
@@ -4248,6 +4173,11 @@ async fn wait_for_any_ok(
     }
 }
 
+mod recovery;
+
+#[cfg(test)]
+mod recovery_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -4775,7 +4705,7 @@ mod tests {
             .expect("signing should succeed")
     }
 
-    async fn test_ws_pair() -> (WsStream, WebSocketStream<tokio::net::TcpStream>) {
+    pub(super) async fn test_ws_pair() -> (WsStream, WebSocketStream<tokio::net::TcpStream>) {
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
             .await
             .expect("bind test websocket");
@@ -4792,7 +4722,7 @@ mod tests {
         (client, server.await.expect("join test websocket server"))
     }
 
-    async fn next_test_frame(
+    pub(super) async fn next_test_frame(
         server: &mut WebSocketStream<tokio::net::TcpStream>,
     ) -> serde_json::Value {
         let message = timeout(Duration::from_secs(1), server.next())
@@ -5035,14 +4965,14 @@ mod tests {
         ));
     }
 
-    fn test_channel_filter() -> ChannelFilter {
+    pub(super) fn test_channel_filter() -> ChannelFilter {
         ChannelFilter {
             kinds: Some(vec![9]),
             require_mention: false,
         }
     }
 
-    fn seed_test_subscription(state: &mut BgState, channel_id: Uuid) {
+    pub(super) fn seed_test_subscription(state: &mut BgState, channel_id: Uuid) {
         apply_command_to_state(
             state,
             RelayCommand::Subscribe {

crates/buzz-acp/src/relay/recovery.rs

@@ -0,0 +1,132 @@
+//! Overflow recovery is an attempted replay, not an EOSE/consumer receipt.
+//! Keep the existing IDs and cursor retirement rules; bound when work is sent.
+use super::*;
+
+pub(super) const RECOVERY_INTERVAL: Duration = Duration::from_secs(5);
+
+#[derive(Default)]
+pub(super) struct RecoverySchedule {
+    next_attempt: Option<tokio::time::Instant>,
+    pub(super) last_attempt: HashMap<String, tokio::time::Instant>,
+}
+
+/// Attempt at most one affected subscription, with space for replay to arrive.
+/// Failed writes retain the loss cursor and are paced too. No EOSE is interpreted
+/// as completion: overlapping requests keep their existing stable wire IDs.
+pub(super) async fn recover_one(
+    ws: &mut WsStream,
+    state: &mut BgState,
+    event_tx: &mpsc::Sender<Option<BuzzEvent>>,
+    agent_pubkey_hex: &str,
+) {
+    let now = tokio::time::Instant::now();
+    if event_tx.is_closed()
+        || event_tx.capacity() < event_tx.max_capacity().div_ceil(2)
+        || state.recovery.next_attempt.is_some_and(|next| now < next)
+        || state.check_rate_gate().is_some()
+    {
+        return;
+    }
+
+    let channel = next_channel(state);
+    let Some(channel) = channel else { return };
+    let sub = channel.map_or_else(|| MEMBERSHIP_NOTIF_SUB_ID.to_owned(), channel_sub_id);
+    state.recovery.last_attempt.insert(sub.clone(), now);
+    info!(subscription = sub, "attempting targeted overflow replay");
+
+    if let Some(ch) = channel {
+        if let Some(filter) = state.active_filters.get(&ch).cloned() {
+            let since = state.channel_since(&ch);
+            if send_subscribe(ws, state, ch, agent_pubkey_hex, since, &filter).await {
+                // Baseline retirement point: REQ write, NOT proven delivery.
+                // New overflow after this attempt creates another pending cursor.
+                state.channel_dropped_since.remove(&ch);
+            }
+        }
+    } else {
+        let since = match (state.membership_dropped_since, state.membership_last_seen) {
+            (Some(d), Some(l)) => Some(d.min(l)),
+            (Some(d), None) => Some(d),
+            (None, Some(l)) => Some(l),
+            (None, None) => state.startup_watermark,
+        };
+        if send_membership_subscribe(ws, agent_pubkey_hex, since).await {
+            state.membership_dropped_since = None;
+        }
+    }
+    // Pace from the end of a potentially backpressured write. No catch-up burst.
+    // The existing bounded write timeout and read/ping owner detect socket loss.
+    state.recovery.next_attempt = Some(tokio::time::Instant::now() + RECOVERY_INTERVAL);
+}
+
+/// No timer or capacity waiter when another authority owns all pending loss.
+/// Only actual attempts advance the cooldown; closed gates wake at expiry.
+pub(super) fn ready_at(state: &mut BgState) -> Option<tokio::time::Instant> {
+    next_channel(state)?;
+    Some(
+        state
+            .recovery
+            .next_attempt
+            .unwrap_or_else(tokio::time::Instant::now)
+            .max(
+                state
+                    .check_rate_gate()
+                    .unwrap_or_else(tokio::time::Instant::now),
+            ),
+    )
+}
+
+/// Select-local readiness, not a send or a reservation carried across reads.
+/// The socket task is the sole producer. `select!` drops this future (including
+/// partial permits) BEFORE handling another frame/command, so live try_send
+/// never competes with a recovery reservation. Receives only add capacity.
+/// Keep this future inside select!: awaiting it alone would block the reader;
+/// persisting it across iterations would steal capacity from live delivery.
+pub(super) async fn ready(
+    event_tx: &mpsc::Sender<Option<BuzzEvent>>,
+    at: Option<tokio::time::Instant>,
+) {
+    if let Some(at) = at {
+        if tokio::time::Instant::now() < at {
+            tokio::time::sleep_until(at).await;
+        }
+        // Use the channel's own race-free capacity wake, not periodic samples.
+        // Return ALL permits before recover_one rechecks capacity and intent.
+        if let Ok(permits) = event_tx
+            .reserve_many(event_tx.max_capacity().div_ceil(2))
+            .await
+        {
+            drop(permits);
+            return;
+        }
+    }
+    // No loss or a closed receiver: no immediate-ready/error wake loop.
+    std::future::pending::<()>().await;
+}
+
+fn next_channel(state: &BgState) -> Option<Option<Uuid>> {
+    // One record per active intent, not per loss or per request generation.
+    // Least recently attempted prevents a repeatedly overflowing channel from
+    // starving other channels or membership. Missing filters fail closed.
+    state
+        .channel_dropped_since
+        .keys()
+        .filter(|ch| {
+            state.active_subscriptions.contains_key(ch)
+                && state.active_filters.contains_key(ch)
+                && !state.rate_limited_pending.contains_key(ch)
+                && !state.resubscribe_retry.contains(ch)
+        })
+        .copied()
+        .map(Some)
+        .chain(
+            (state.membership_sub_active
+                && state.membership_dropped_since.is_some()
+                && !state.membership_resub_needed)
+                .then_some(None),
+        )
+        .min_by_key(|ch| {
+            let sub = ch.map_or_else(|| MEMBERSHIP_NOTIF_SUB_ID.to_owned(), channel_sub_id);
+            (state.recovery.last_attempt.get(&sub).copied(), sub)
+        })
+}