From 7fb3a63ddc7516d5da561e8a8bb912b931e380d2 Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Fri, 3 Jul 2026 11:17:39 -0400 Subject: [PATCH] =?UTF-8?q?fix(buzz-acp):=20recover=20steered=20?= =?UTF-8?q?=F0=9F=91=80=20from=20acks=20that=20cross=20shutdown?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review (Wren) found a final shutdown race in 47193621: the one-time TaskMeta drain only captures steered ids whose SteerAck::Success was already processed by the main loop. A watcher can resolve during the 30s grace period — after select! stopped polling steer_ack_rx — so a successful pending steer in that window was attached to nothing and its 👀 died stale at process exit. Fix: after the grace-period drain settles every prompt task (so every ack watcher has resolved or dropped), shutdown drops its steer_ack_tx and drains the channel (drain_crossing_steer_acks), collecting Success event ids and removing their 👀 alongside the TaskMeta drain's ids — both bound-awaited before exit. A per-recv 2s timeout backstops a wedged watcher. Non-Success acks are dropped: their events were released to the queue being discarded, the same pre-existing stale-👀 fate as any queued-but-never-dispatched event at exit. Tests: crossing_success_ack_is_drained_and_cleaned_on_the_wire pins the drain (Success kept, non-Success ignored, channel-close exit) and the wire-level kind:5 deletion; crossing_ack_drain_times_out_on_wedged_watcher (paused time) pins the no-hang backstop. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-acp/src/lib.rs | 133 +++++++++++++++++++++++++++++++++++-- 1 file changed, 127 insertions(+), 6 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 18edd6be4..87834c9e8 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2306,10 +2306,12 @@ async fn tokio_main() -> Result<()> { // tasks that outlive the grace period are aborted — either way every // TaskMeta drops with the pool and its steered_event_ids with it. All // in-flight turns are stopping now, so the user-visible contract (eyes - // clear when the turn stops) says remove them. No new steer acks can be - // recorded after the main loop exits, so draining once here is complete. - // Runs concurrently with the grace-period drain; awaited (bounded, each - // reaction_remove is internally timeout-capped) before exit. + // clear when the turn stops) says remove them. This one-time drain only + // covers acks already processed by the SteerAck arm; acks still pending + // when the loop exited are handled by drain_crossing_steer_acks below, + // after the grace period resolves every watcher. Runs concurrently with + // the grace-period drain; awaited (bounded, each reaction_remove is + // internally timeout-capped) before exit. let steered_cleanup = spawn_steered_eyes_cleanup(Some(&ctx.rest_client), pool.drain_all_steered_event_ids()); // 30 s is generous for in-flight prompts to be cancelled; using @@ -2365,11 +2367,21 @@ async fn tokio_main() -> Result<()> { } drop(pool); - // Wait (bounded) for the steered-👀 cleanup spawned above — each + // Resolve success acks that crossed the loop exit (watcher resolved + // during the grace period, after `select!` stopped polling). All prompt + // tasks have now settled, so every watcher is done or doomed — drop our + // sender and drain the channel, then remove the 👀 those events carry. + drop(steer_ack_tx); + let crossing_cleanup = spawn_steered_eyes_cleanup( + Some(&ctx.rest_client), + drain_crossing_steer_acks(&mut steer_ack_rx).await, + ); + + // Wait (bounded) for the steered-👀 cleanups spawned above — each // reaction_remove is internally capped at ~2 s of HTTP, so this cannot // hang shutdown meaningfully. Best-effort: on timeout the reactions // stay stale, same class of loss as any other kill-during-cleanup. - if let Some(handle) = steered_cleanup { + for handle in [steered_cleanup, crossing_cleanup].into_iter().flatten() { if tokio::time::timeout(Duration::from_secs(10), handle) .await .is_err() @@ -2722,6 +2734,53 @@ fn spawn_steered_eyes_cleanup( })) } +/// Drain `steer_ack_rx` after the main loop has exited and every prompt +/// task has settled, returning the event ids of `SteerAck::Success` acks +/// that crossed the loop exit. +/// +/// A watcher can resolve during the shutdown grace period — after the +/// `select!` stopped polling `steer_ack_rx` but while its turn's read loop +/// was still running. Such a Success ack was never seen by the SteerAck +/// arm: its event id is on no `TaskMeta` (the one-time drain missed it) and +/// no immediate cleanup fired — without this pass its 👀 would die stale at +/// process exit. +/// +/// Caller must drop its `steer_ack_tx` clone first and call this only after +/// the grace-period drain: with all prompt tasks finished or aborted, every +/// watcher's oneshot has resolved (or dropped → `RecvError`), so each +/// watcher sends promptly and releases its sender clone — `recv()` then +/// yields `None` once the channel empties. The per-recv timeout is a +/// backstop against a wedged watcher; on timeout we keep what was drained. +/// +/// Non-Success acks are dropped: their events were released back to the +/// queue (which is being discarded) — the same stale-👀-at-exit fate as any +/// queued-but-never-dispatched event, a pre-existing shutdown limitation +/// owned by the relay-side cleanup. +async fn drain_crossing_steer_acks( + steer_ack_rx: &mut mpsc::UnboundedReceiver, +) -> Vec { + let mut ids = Vec::new(); + loop { + match tokio::time::timeout(Duration::from_secs(2), steer_ack_rx.recv()).await { + Ok(Some(ev)) => { + if matches!(ev.ack, Ok(pool::SteerAck::Success)) { + ids.push(ev.event_id); + } + } + Ok(None) => break, // channel closed — every watcher finished + Err(_) => { + tracing::warn!( + drained = ids.len(), + "steer ack watcher still pending at shutdown deadline — \ + proceeding with acks drained so far" + ); + break; + } + } + } + ids +} + #[allow(clippy::too_many_arguments)] fn handle_prompt_result( pool: &mut AgentPool, @@ -4579,6 +4638,68 @@ mod steered_eyes_lifecycle_tests { tokio::time::sleep(Duration::from_millis(50)).await; assert!(captured.lock().await.is_empty()); } + + /// Build a `SteerAckEvent` with a real (already-finished) task id. + async fn ack_event(event_id: &str, ack: pool::SteerAck) -> SteerAckEvent { + let handle = tokio::spawn(async {}); + let task_id = handle.id(); + let _ = handle.await; + SteerAckEvent { + channel_id: Uuid::new_v4(), + event_id: event_id.to_string(), + task_id, + ack: Ok(ack), + } + } + + /// Crossing-ack shutdown race: a `SteerAck::Success` whose watcher + /// resolves after the main loop stopped polling `steer_ack_rx` was + /// never attached to any TaskMeta — the post-grace drain must still + /// recover its event id (and only Success ids) and remove the 👀 on + /// the wire before process exit. + #[tokio::test] + async fn crossing_success_ack_is_drained_and_cleaned_on_the_wire() { + let reaction_id = "5a".repeat(32); + let (rest, captured) = spawn_mock_relay(&reaction_id).await; + + let (tx, mut rx) = mpsc::unbounded_channel::(); + let steered_event = "77".repeat(32); + // Watcher resolved during the grace period: Success ack sitting in + // the channel, unseen by the (exited) main loop. + tx.send(ack_event(&steered_event, pool::SteerAck::Success).await) + .unwrap(); + // Non-Success acks in the same window must be ignored. + tx.send(ack_event("ignored", pool::SteerAck::PromptCompletedNeutral).await) + .unwrap(); + drop(tx); // shutdown drops its sender before draining + + let ids = drain_crossing_steer_acks(&mut rx).await; + assert_eq!(ids, vec![steered_event]); + + let handle = spawn_steered_eyes_cleanup(Some(&rest), ids) + .expect("crossing success ack must spawn cleanup"); + handle.await.expect("cleanup task must not panic"); + + let deletions = captured.lock().await.clone(); + assert_eq!(deletions.len(), 1); + assert_is_deletion_of(&deletions[0], &reaction_id); + } + + /// Backstop: a wedged watcher (sender clone alive, never sends) must + /// not hang shutdown — the drain times out and keeps what it has. + /// Paused time makes the 2 s recv deadline resolve instantly. + #[tokio::test(start_paused = true)] + async fn crossing_ack_drain_times_out_on_wedged_watcher() { + let (tx, mut rx) = mpsc::unbounded_channel::(); + let done = "99".repeat(32); + tx.send(ack_event(&done, pool::SteerAck::Success).await) + .unwrap(); + // `tx` intentionally kept alive: simulates a watcher that never + // resolves, so the channel never closes. + let ids = drain_crossing_steer_acks(&mut rx).await; + assert_eq!(ids, vec![done], "must keep acks drained before timeout"); + drop(tx); + } } #[cfg(test)]