From 8e4a0d73ccfbfca4771661d66745997a1fa8c5ee Mon Sep 17 00:00:00 2001 From: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Date: Sat, 18 Jul 2026 09:13:34 -0400 Subject: [PATCH 1/4] fix(acp): pace relay observer frames Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-acp/src/lib.rs | 101 ++++++++++++++++++++++++++++++++++--- 1 file changed, 94 insertions(+), 7 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 8e50a7ce76..33827bdd48 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -13,7 +13,7 @@ mod usage; pub use usage::TurnUsage; -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::Arc; use std::time::Duration; @@ -287,6 +287,53 @@ async fn check_sibling_via_profile( false } +const OBSERVER_PUBLISH_INTERVAL: Duration = Duration::from_millis(167); +const OBSERVER_PUBLISH_LIMIT_PER_MINUTE: usize = 90; + +struct ObserverPublishPacer { + next_publish: tokio::time::Instant, + published: VecDeque, +} + +impl ObserverPublishPacer { + fn new() -> Self { + Self { + // No initial burst: even the first snapshot frame waits for its slot. + next_publish: tokio::time::Instant::now() + OBSERVER_PUBLISH_INTERVAL, + published: VecDeque::with_capacity(OBSERVER_PUBLISH_LIMIT_PER_MINUTE), + } + } + + async fn wait(&mut self) { + loop { + let now = tokio::time::Instant::now(); + while self + .published + .front() + .is_some_and(|sent| now.duration_since(*sent) >= Duration::from_secs(60)) + { + self.published.pop_front(); + } + + let minute_slot = self.published.front().and_then(|sent| { + (self.published.len() >= OBSERVER_PUBLISH_LIMIT_PER_MINUTE) + .then_some(*sent + Duration::from_secs(60)) + }); + let publish_at = + minute_slot.map_or(self.next_publish, |slot| slot.max(self.next_publish)); + if publish_at > now { + tokio::time::sleep_until(publish_at).await; + continue; + } + + let published_at = tokio::time::Instant::now(); + self.published.push_back(published_at); + self.next_publish = published_at + OBSERVER_PUBLISH_INTERVAL; + return; + } + } +} + fn spawn_relay_observer_publisher( observer: observer::ObserverHandle, publisher: RelayEventPublisher, @@ -297,7 +344,11 @@ fn spawn_relay_observer_publisher( ) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { let mut coalescer = ObserverChunkCoalescer::default(); - for event in observer.snapshot() { + let mut pacer = ObserverPublishPacer::new(); + // Subscribe before pacing so live events remain buffered while the snapshot drains. + let snapshot = observer.snapshot(); + let mut rx = observer.subscribe(); + for event in snapshot { for event in coalescer.ingest(event) { publish_relay_observer_event( &publisher, @@ -305,13 +356,13 @@ fn spawn_relay_observer_publisher( &agent_pubkey_hex, &owner_pubkey_hex, &owner_pubkey, + &mut pacer, event, ) .await; } } - let mut rx = observer.subscribe(); let mut flush_interval = tokio::time::interval(std::time::Duration::from_millis(500)); flush_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { @@ -322,7 +373,7 @@ fn spawn_relay_observer_publisher( for event in coalescer.ingest(event) { publish_relay_observer_event( &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, event, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, ).await; } } @@ -330,7 +381,7 @@ fn spawn_relay_observer_publisher( for event in coalescer.flush() { publish_relay_observer_event( &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, event, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, ).await; } tracing::warn!(dropped = count, "relay observer publisher lagged"); @@ -339,7 +390,7 @@ fn spawn_relay_observer_publisher( for event in coalescer.flush() { publish_relay_observer_event( &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, event, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, ).await; } break; @@ -351,7 +402,7 @@ fn spawn_relay_observer_publisher( for event in coalescer.flush() { publish_relay_observer_event( &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, event, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, ).await; } } @@ -638,8 +689,10 @@ async fn publish_relay_observer_event( agent_pubkey_hex: &str, owner_pubkey_hex: &str, owner_pubkey: &PublicKey, + pacer: &mut ObserverPublishPacer, mut event: observer::ObserverEvent, ) { + pacer.wait().await; // Trim oversized frames to fit the plaintext cap rather than letting // encrypt_observer_payload reject and drop them whole (silent telemetry loss). fit_observer_event_to_budget(&mut event); @@ -4070,6 +4123,40 @@ mod author_gate_tests { } } +#[cfg(test)] +mod observer_publish_pacer_tests { + use super::*; + + #[tokio::test(start_paused = true)] + async fn starts_without_a_burst_and_spaces_frames() { + let started = tokio::time::Instant::now(); + let mut pacer = ObserverPublishPacer::new(); + + pacer.wait().await; + let first = tokio::time::Instant::now(); + pacer.wait().await; + let second = tokio::time::Instant::now(); + + assert_eq!(first.duration_since(started), OBSERVER_PUBLISH_INTERVAL); + assert_eq!(second.duration_since(first), OBSERVER_PUBLISH_INTERVAL); + } + + #[tokio::test(start_paused = true)] + async fn limits_frames_in_each_rolling_minute() { + let mut pacer = ObserverPublishPacer::new(); + pacer.wait().await; + let first = tokio::time::Instant::now(); + for _ in 1..OBSERVER_PUBLISH_LIMIT_PER_MINUTE { + pacer.wait().await; + } + + pacer.wait().await; + let ninety_first = tokio::time::Instant::now(); + + assert_eq!(ninety_first.duration_since(first), Duration::from_secs(60)); + } +} + #[cfg(test)] mod observer_chunk_coalescer_tests { use super::*; From 5916ce975bc8886d28cb74612f1ce50ad66f5003 Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Tue, 21 Jul 2026 07:34:40 -0700 Subject: [PATCH 2/4] fix(acp): close observer snapshot/live handoff race Subscribe to the live observer feed BEFORE taking the replay snapshot, so an event emitted between the two calls can never miss both. The overlap this creates is deduped in the run loop via the snapshot's high-water seq (ObserverEvent.seq is monotonic, assigned at emit). The publisher body moves into run_relay_observer_publisher so the exactly-once property is testable without a spawned task; the new test emits an event inside the race window and asserts before/overlap/after each publish exactly once, in order. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-acp/src/lib.rs | 205 +++++++++++++++++++++++++++---------- 1 file changed, 150 insertions(+), 55 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 33827bdd48..923e132167 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -343,72 +343,102 @@ fn spawn_relay_observer_publisher( owner_pubkey: PublicKey, ) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { - let mut coalescer = ObserverChunkCoalescer::default(); - let mut pacer = ObserverPublishPacer::new(); - // Subscribe before pacing so live events remain buffered while the snapshot drains. + // Subscribe BEFORE snapshotting so an event emitted between the two + // calls is never lost: it lands in the snapshot, the live receiver, or + // both. The overlap is deduped in the run loop via the snapshot's + // high-water `seq` (monotonic, assigned at emit). + let rx = observer.subscribe(); let snapshot = observer.snapshot(); - let mut rx = observer.subscribe(); - for event in snapshot { - for event in coalescer.ingest(event) { - publish_relay_observer_event( - &publisher, - &keys, - &agent_pubkey_hex, - &owner_pubkey_hex, - &owner_pubkey, - &mut pacer, - event, - ) - .await; - } + run_relay_observer_publisher( + snapshot, + rx, + publisher, + keys, + agent_pubkey_hex, + owner_pubkey_hex, + owner_pubkey, + ) + .await; + }) +} + +async fn run_relay_observer_publisher( + snapshot: Vec, + mut rx: tokio::sync::broadcast::Receiver, + publisher: RelayEventPublisher, + keys: nostr::Keys, + agent_pubkey_hex: String, + owner_pubkey_hex: String, + owner_pubkey: PublicKey, +) { + let mut coalescer = ObserverChunkCoalescer::default(); + let mut pacer = ObserverPublishPacer::new(); + let max_snapshot_seq = snapshot.iter().map(|event| event.seq).max().unwrap_or(0); + for event in snapshot { + for event in coalescer.ingest(event) { + publish_relay_observer_event( + &publisher, + &keys, + &agent_pubkey_hex, + &owner_pubkey_hex, + &owner_pubkey, + &mut pacer, + event, + ) + .await; } + } - let mut flush_interval = tokio::time::interval(std::time::Duration::from_millis(500)); - flush_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - loop { - tokio::select! { - result = rx.recv() => { - match result { - Ok(event) => { - for event in coalescer.ingest(event) { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, - ).await; - } + let mut flush_interval = tokio::time::interval(std::time::Duration::from_millis(500)); + flush_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + result = rx.recv() => { + match result { + Ok(event) => { + // Skip live events already delivered via the snapshot + // (the subscribe-before-snapshot overlap). + if event.seq <= max_snapshot_seq { + continue; } - Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { - for event in coalescer.flush() { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, - ).await; - } - tracing::warn!(dropped = count, "relay observer publisher lagged"); + for event in coalescer.ingest(event) { + publish_relay_observer_event( + &publisher, &keys, &agent_pubkey_hex, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + ).await; } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - for event in coalescer.flush() { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, - ).await; - } - break; + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => { + for event in coalescer.flush() { + publish_relay_observer_event( + &publisher, &keys, &agent_pubkey_hex, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + ).await; } + tracing::warn!(dropped = count, "relay observer publisher lagged"); } - } - _ = flush_interval.tick() => { - // Periodic flush ensures live streaming even during continuous chunk delivery. - for event in coalescer.flush() { - publish_relay_observer_event( - &publisher, &keys, &agent_pubkey_hex, - &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, - ).await; + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + for event in coalescer.flush() { + publish_relay_observer_event( + &publisher, &keys, &agent_pubkey_hex, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + ).await; + } + break; } } } + _ = flush_interval.tick() => { + // Periodic flush ensures live streaming even during continuous chunk delivery. + for event in coalescer.flush() { + publish_relay_observer_event( + &publisher, &keys, &agent_pubkey_hex, + &owner_pubkey_hex, &owner_pubkey, &mut pacer, event, + ).await; + } + } } - }) + } } #[derive(Default)] @@ -4123,6 +4153,71 @@ mod author_gate_tests { } } +#[cfg(test)] +mod observer_snapshot_race_tests { + use super::*; + use nostr::Keys; + + fn emit_marker(observer: &observer::ObserverHandle, marker: &str) { + observer.emit( + "test_event", + None, + &observer::context_for(None, None, None), + serde_json::json!({ "marker": marker }), + ); + } + + /// An event emitted between `subscribe()` and `snapshot()` lands in BOTH + /// the snapshot and the live receiver; the seq high-water dedupe must + /// deliver it exactly once — and never lose events on either side of it. + #[tokio::test(start_paused = true)] + async fn overlap_between_subscribe_and_snapshot_publishes_exactly_once() { + let observer = observer::ObserverHandle::in_process(); + let agent_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let (publisher, mut published_rx) = RelayEventPublisher::test_pair(); + + // Before the publisher starts: replay-buffer only. + emit_marker(&observer, "before"); + // The race window: emitted after subscribe() but before snapshot(), + // so it is present in the snapshot AND queued on the receiver. + let rx = observer.subscribe(); + emit_marker(&observer, "overlap"); + let snapshot = observer.snapshot(); + assert_eq!(snapshot.len(), 2, "overlap event must be in the snapshot"); + // After the snapshot: live receiver only. + emit_marker(&observer, "after"); + // Close the broadcast channel so the run loop drains and exits. + drop(observer); + + run_relay_observer_publisher( + snapshot, + rx, + publisher, + agent_keys.clone(), + agent_keys.public_key().to_hex(), + owner_keys.public_key().to_hex(), + owner_keys.public_key(), + ) + .await; + + // The run loop has exited, dropping the publisher; drain the forwarded + // events until the channel closes (deterministic — no try_recv race + // with the test_pair forwarding task). + let mut markers = Vec::new(); + while let Some(event) = published_rx.recv().await { + let payload: serde_json::Value = + decrypt_observer_payload(&owner_keys, &event).expect("decrypt published frame"); + markers.push(payload["payload"]["marker"].as_str().unwrap().to_string()); + } + assert_eq!( + markers, + ["before", "overlap", "after"], + "each event must be published exactly once, in order" + ); + } +} + #[cfg(test)] mod observer_publish_pacer_tests { use super::*; From 6c0fcae66ad1495d89f53181d6cdfb8f79f25583 Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Tue, 21 Jul 2026 07:34:57 -0700 Subject: [PATCH 3/4] fix(acp): park observer frames while rate-gated instead of dropping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The generic PublishEvent arm silently drops every publish while the rate-limit gate (#2199) is armed. That is correct for typing indicators, but observer telemetry frames (kind 24200) are durable: one NOTICE from any traffic source erased ~30 frames of turn history over the default ~5s gate. Observer frames now park in a bounded drop-oldest FIFO on BgState (cap 256 ≈ 40s of upstream pacing at 6/s) whenever the gate is armed or earlier parked frames are still draining (order preserved). The existing main-loop drain machinery delivers them one frame per 125ms pacing tick once the gate clears, arming the pacing timer to the gate deadline so drains fire even with no other traffic. Overflow evicts oldest and is counted + logged — visible loss, never silent. The same parking now also covers observer frames hit by a live send failure or emitted while disconnected, so they survive reconnect via the post-reconnect drain. Typing indicators keep the existing drop-while-gated behavior. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-acp/src/relay.rs | 372 +++++++++++++++++++++++++++++++++-- 1 file changed, 351 insertions(+), 21 deletions(-) diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index b1497b6be7..99efa62a92 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -105,6 +105,11 @@ const REQ_PACING_INTERVAL: Duration = Duration::from_millis(125); /// below the relay's 50-frames/5s budget, and ensures the select! loop is never /// blocked for more than one REQ's worth of I/O between drain ticks. const DRAIN_BUDGET_PER_ITER: usize = 1; +/// Maximum observer telemetry frames parked while the rate-limit gate is armed +/// (or the socket is down). The upstream pacer feeds at most ~6 frames/s, so +/// this covers ~40 s of gating; beyond that the oldest frames are dropped with +/// visible accounting (`gated_observer_dropped`). +const GATED_OBSERVER_QUEUE_CAP: usize = 256; use std::time::Instant; @@ -551,6 +556,24 @@ impl RelayEventPublisher { .await .map_err(|_| RelayError::ConnectionClosed) } + + /// Test-only publisher pair: published events are forwarded to the + /// returned receiver instead of a live relay socket. + #[cfg(test)] + pub(crate) fn test_pair() -> (Self, mpsc::Receiver) { + let (cmd_tx, mut cmd_rx) = mpsc::channel::(64); + let (event_tx, event_rx) = mpsc::channel(64); + tokio::spawn(async move { + while let Some(cmd) = cmd_rx.recv().await { + if let RelayCommand::PublishEvent { event } = cmd { + if event_tx.send(*event).await.is_err() { + break; + } + } + } + }); + (Self { cmd_tx }, event_rx) + } } impl HarnessRelay { @@ -998,6 +1021,15 @@ struct BgState { /// subscription. The main-loop drain re-sends the REQ once the gate clears, /// even when `rate_limited_pending` is empty. observer_resub_needed: bool, + /// Observer telemetry frames (kind 24200) parked while the rate-limit gate + /// is armed. Unlike typing indicators, these frames are durable telemetry: + /// dropping them silently loses turn history in the Desktop observer. + /// Bounded at `GATED_OBSERVER_QUEUE_CAP` (drop-oldest); drained by the + /// main loop one frame per pacing tick once the gate clears. + gated_observer_pending: VecDeque>, + /// Frames evicted from `gated_observer_pending` since the last drain-empty + /// summary log. Makes overflow loss visible instead of silent. + gated_observer_dropped: u64, /// Channels whose REQ failed during `resubscribe_after_reconnect`. /// /// A single failed channel REQ is parked here instead of aborting the whole @@ -1030,6 +1062,8 @@ impl BgState { rate_limited_pending: HashMap::new(), membership_resub_needed: false, observer_resub_needed: false, + gated_observer_pending: VecDeque::new(), + gated_observer_dropped: 0, resubscribe_retry: HashSet::new(), backoff_step: 0, } @@ -1128,13 +1162,30 @@ impl BgState { } None } + + /// Park an observer telemetry frame while the rate-limit gate is armed. + /// + /// Bounded drop-oldest queue: overflow evicts the oldest frame and counts + /// it in `gated_observer_dropped` so the loss is visible, never silent. + fn park_gated_observer_frame(&mut self, event: Box) { + if self.gated_observer_pending.len() >= GATED_OBSERVER_QUEUE_CAP { + self.gated_observer_pending.pop_front(); + self.gated_observer_dropped += 1; + warn!( + dropped_total = self.gated_observer_dropped, + "gated observer queue full — dropped oldest frame" + ); + } + self.gated_observer_pending.push_back(event); + } } /// Record a command's intent in state while disconnected (no WebSocket). /// /// Subscribe/Unsubscribe/SubscribeMembership record intent so reconnect /// restores the right subscriptions. SetStartupWatermark floors the replay -/// window. PublishEvent and Reconnect are no-ops while disconnected. +/// window. Observer telemetry publishes are parked for post-reconnect drain; +/// other PublishEvent and Reconnect are no-ops while disconnected. /// /// Callers MUST handle `Shutdown` before calling — reaching the Shutdown /// arm here is a logic error. @@ -1174,8 +1225,15 @@ fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) { state.membership_last_seen = Some(ts); } } - // Ephemeral events are meaningless while disconnected. - RelayCommand::PublishEvent { .. } => {} + // Observer telemetry frames are durable: park them (bounded, visible + // overflow) so they are delivered by the post-reconnect drain. Other + // ephemeral publishes (typing indicators) are meaningless while + // disconnected and are dropped. + RelayCommand::PublishEvent { event } => { + if event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME { + state.park_gated_observer_frame(event); + } + } // Already reconnecting — redundant. RelayCommand::Reconnect => {} // Callers MUST handle Shutdown before calling this function. @@ -1190,11 +1248,17 @@ fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) { /// Retain command intent after a live send failure. /// -/// Subscription state must survive reconnect; ephemeral publishes are deliberately +/// Subscription state must survive reconnect. Observer telemetry publishes are +/// parked for post-reconnect drain; other ephemeral publishes are deliberately /// discarded because replaying a typing indicator after reconnect is meaningless. /// `Shutdown` and `Reconnect` are handled by the caller. fn retain_failed_command_intent(state: &mut BgState, cmd: RelayCommand) { match cmd { + RelayCommand::PublishEvent { event } + if event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME => + { + state.park_gated_observer_frame(event); + } RelayCommand::PublishEvent { .. } => {} cmd => apply_command_to_state(state, cmd), } @@ -1350,28 +1414,41 @@ async fn execute_connected_command( } } RelayCommand::PublishEvent { event } => { - // Drop ephemeral publishes while rate-gated. Stale typing indicators - // are worthless and sending them would consume admission budget the - // relay already rejected us on. + // Observer telemetry frames (kind 24200) are durable telemetry, not + // droppable ephemera: park them while the rate-limit gate is armed — + // and while earlier parked frames are still draining, so relative + // order is preserved — then let the main-loop drain deliver them + // one per pacing tick once the gate clears. + if event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME + && (state.check_rate_gate().is_some() || !state.gated_observer_pending.is_empty()) + { + debug!( + pending = state.gated_observer_pending.len(), + "rate-gated: parking observer frame for paced drain" + ); + state.park_gated_observer_frame(event); + return true; + } + // Drop remaining ephemeral publishes while rate-gated. Stale typing + // indicators are worthless and sending them would consume admission + // budget the relay already rejected us on. // - // INVARIANT: the WS publish path carries only ephemeral kinds (typing - // indicators). The silent drop-while-gated relies on that invariant. If a - // future caller publishes durable events through this path, it must add a - // kind guard before this branch to avoid silently discarding user data. + // INVARIANT: apart from observer frames (parked above), the WS publish + // path carries only ephemeral kinds (typing indicators). The silent + // drop-while-gated relies on that invariant. If a future caller + // publishes durable events through this path, it must extend the + // kind guard above to avoid silently discarding user data. if state.check_rate_gate().is_some() { debug!("rate-gated: dropping ephemeral PublishEvent (typing indicator)"); return true; } - let msg = json!(["EVENT", event]); - if let Ok(text) = serde_json::to_string(&msg) { - if let Err(e) = - ws_send_timeout(ws, Message::Text(text.into()), WS_SEND_TIMEOUT_SECS).await - { - // Ephemeral events (typing indicators) are best-effort. - // Log the failure but don't trigger reconnect — the next - // ping or read will detect the dead socket. - warn!("failed to publish event: {e}"); - } + // Best-effort: log a send failure but don't trigger reconnect — the + // next ping or read will detect the dead socket. A failed observer + // frame is parked so the post-reconnect drain redelivers it. + if !send_publish_event_frame(ws, &event).await + && event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME + { + state.park_gated_observer_frame(event); } true } @@ -1624,6 +1701,14 @@ async fn run_background_task( if budget > 0 && !state.resubscribe_retry.is_empty() { let sent = drain_resubscribe_retry(&mut ws, &mut state, &agent_pubkey_hex, budget).await; + budget = budget.saturating_sub(sent); + if sent > 0 { + any_sent = true; + } + } + + if budget > 0 && !state.gated_observer_pending.is_empty() { + let sent = drain_gated_observer_pending(&mut ws, &mut state, budget).await; if sent > 0 { any_sent = true; } @@ -1631,6 +1716,13 @@ async fn run_background_task( if any_sent { drain_pacing_next = Some(tokio::time::Instant::now() + REQ_PACING_INTERVAL); + } else if !state.gated_observer_pending.is_empty() { + // Nothing sent because the gate is still armed. Arm the pacing + // timer to the gate deadline so parked observer frames drain + // promptly even when no other traffic wakes the select loop. + drain_pacing_next = state + .check_rate_gate() + .or_else(|| Some(tokio::time::Instant::now() + REQ_PACING_INTERVAL)); } } @@ -2455,6 +2547,59 @@ async fn resubscribe_after_reconnect( } } +/// Send a signed EVENT frame on the live socket. Returns `false` on send failure. +/// +/// Best-effort at the socket level: a failure is logged but does not trigger +/// reconnect — the next ping or read will detect the dead socket. +async fn send_publish_event_frame(ws: &mut WsStream, event: &Event) -> bool { + let msg = json!(["EVENT", event]); + if let Ok(text) = serde_json::to_string(&msg) { + if let Err(e) = ws_send_timeout(ws, Message::Text(text.into()), WS_SEND_TIMEOUT_SECS).await + { + warn!("failed to publish event: {e}"); + return false; + } + } + true +} + +/// Drain parked observer telemetry frames once the rate-limit gate clears. +/// +/// Called by the main loop pacing timer. Sends at most `budget` frames without +/// sleeping — pacing is enforced by the caller via `drain_pacing_next`. Stops +/// immediately if the gate re-arms mid-drain. When the queue empties, any +/// overflow loss is summarized in one warning. Returns the number of frames sent. +async fn drain_gated_observer_pending( + ws: &mut WsStream, + state: &mut BgState, + budget: usize, +) -> usize { + let mut sent = 0; + while sent < budget { + if state.check_rate_gate().is_some() { + break; + } + let Some(event) = state.gated_observer_pending.pop_front() else { + break; + }; + if !send_publish_event_frame(ws, &event).await { + // Socket may be dead — re-park at the front so the frame survives + // reconnect (the post-reconnect drain will retry it in order). + state.gated_observer_pending.push_front(event); + break; + } + sent += 1; + } + if state.gated_observer_pending.is_empty() && state.gated_observer_dropped > 0 { + warn!( + observer_frames_dropped = state.gated_observer_dropped, + "observer frames lost to gated-queue overflow" + ); + state.gated_observer_dropped = 0; + } + sent +} + /// Drain `rate_limited_pending` channels whose retry deadline has passed. /// /// Called by the main loop pacing timer. Sends at most `budget` REQs without @@ -5564,6 +5709,191 @@ mod tests { ); } + /// Build a signed observer telemetry frame (kind 24200) for gate tests. + fn make_observer_frame(keys: &Keys) -> Event { + let recipient = Keys::generate(); + let encrypted = buzz_core::observer::encrypt_observer_payload( + keys, + &recipient.public_key(), + &json!({"type": "test"}), + ) + .expect("encrypt test observer payload"); + buzz_sdk::build_agent_observer_frame( + &recipient.public_key().to_hex(), + &keys.public_key().to_hex(), + "telemetry", + &encrypted, + ) + .expect("build test observer frame") + .sign_with_keys(keys) + .expect("sign test observer frame") + } + + /// While the rate-limit gate is armed, an observer frame (kind 24200) is + /// parked — not silently dropped — and delivered by the drain once the + /// gate clears. A typing indicator in the same window stays dropped. + #[tokio::test] + async fn gated_observer_frame_is_parked_then_drained_not_dropped() { + let (mut client, mut server) = test_ws_pair().await; + let mut state = BgState::new(); + let keys = Keys::generate(); + state.rate_limit_gate = Some(tokio::time::Instant::now() + Duration::from_millis(150)); + + // Observer frame while gated: parked, nothing on the wire. + let observer_frame = make_observer_frame(&keys); + let ok = execute_connected_command( + &mut client, + &mut state, + "agent-pubkey", + RelayCommand::PublishEvent { + event: Box::new(observer_frame.clone()), + }, + ) + .await; + assert!(ok); + assert_eq!( + state.gated_observer_pending.len(), + 1, + "observer frame must be parked while gated" + ); + + // Typing indicator while gated: still dropped, not parked. + let typing = EventBuilder::new(Kind::Custom(KIND_TYPING_INDICATOR as u16), "") + .tags([Tag::parse(["h", &Uuid::new_v4().to_string()]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign typing indicator"); + let ok = execute_connected_command( + &mut client, + &mut state, + "agent-pubkey", + RelayCommand::PublishEvent { + event: Box::new(typing), + }, + ) + .await; + assert!(ok); + assert_eq!( + state.gated_observer_pending.len(), + 1, + "typing indicators must not be parked" + ); + assert!( + timeout(Duration::from_millis(50), server.next()) + .await + .is_err(), + "nothing may reach the wire while the gate is armed" + ); + + // Gate expires — the drain delivers the parked frame. + tokio::time::sleep(Duration::from_millis(160)).await; + assert_eq!( + drain_gated_observer_pending(&mut client, &mut state, 1).await, + 1 + ); + assert!(state.gated_observer_pending.is_empty()); + let frame = next_test_frame(&mut server).await; + assert_eq!(frame[0], "EVENT"); + assert_eq!(frame[1]["id"], observer_frame.id.to_hex()); + assert_eq!( + frame[1]["kind"], + u64::from(KIND_AGENT_OBSERVER_FRAME), + "delivered frame must be the parked observer frame" + ); + } + + /// Observer frames arriving while earlier parked frames are still queued + /// are appended behind them (order preserved), even if the gate has + /// already expired. + #[tokio::test] + async fn observer_frames_queue_behind_parked_backlog_in_order() { + let (mut client, mut server) = test_ws_pair().await; + let mut state = BgState::new(); + let keys = Keys::generate(); + state.rate_limit_gate = Some(tokio::time::Instant::now() + Duration::from_millis(50)); + + let first = make_observer_frame(&keys); + let second = make_observer_frame(&keys); + for event in [&first, &second] { + let ok = execute_connected_command( + &mut client, + &mut state, + "agent-pubkey", + RelayCommand::PublishEvent { + event: Box::new(event.clone()), + }, + ) + .await; + assert!(ok); + } + assert_eq!(state.gated_observer_pending.len(), 2); + + // Gate expires but the backlog is not drained yet — a third frame must + // queue behind it rather than jumping ahead on the wire. + tokio::time::sleep(Duration::from_millis(60)).await; + let third = make_observer_frame(&keys); + let ok = execute_connected_command( + &mut client, + &mut state, + "agent-pubkey", + RelayCommand::PublishEvent { + event: Box::new(third.clone()), + }, + ) + .await; + assert!(ok); + assert_eq!( + state.gated_observer_pending.len(), + 3, + "frame must queue behind undrained backlog to preserve order" + ); + + for expected in [&first, &second, &third] { + assert_eq!( + drain_gated_observer_pending(&mut client, &mut state, 1).await, + 1 + ); + let frame = next_test_frame(&mut server).await; + assert_eq!(frame[1]["id"], expected.id.to_hex(), "order preserved"); + } + assert!(state.gated_observer_pending.is_empty()); + } + + /// The parked-frame queue is bounded: overflow evicts the oldest frame and + /// counts it; the drain resets the counter after logging the summary. + #[tokio::test] + async fn gated_observer_queue_drops_oldest_on_overflow() { + let mut state = BgState::new(); + let keys = Keys::generate(); + let first = make_observer_frame(&keys); + state.park_gated_observer_frame(Box::new(first.clone())); + for _ in 1..GATED_OBSERVER_QUEUE_CAP { + state.park_gated_observer_frame(Box::new(make_observer_frame(&keys))); + } + assert_eq!(state.gated_observer_pending.len(), GATED_OBSERVER_QUEUE_CAP); + assert_eq!(state.gated_observer_dropped, 0); + + let overflow = make_observer_frame(&keys); + state.park_gated_observer_frame(Box::new(overflow.clone())); + assert_eq!( + state.gated_observer_pending.len(), + GATED_OBSERVER_QUEUE_CAP, + "queue must stay bounded" + ); + assert_eq!(state.gated_observer_dropped, 1, "loss must be counted"); + assert!( + !state + .gated_observer_pending + .iter() + .any(|e| e.id == first.id), + "oldest frame must be the one evicted" + ); + assert_eq!( + state.gated_observer_pending.back().map(|e| e.id), + Some(overflow.id), + "newest frame must be retained" + ); + } + /// is_dns_error correctly classifies platform resolver strings, including /// the production shape: a WebSocket I/O error wrapping the OS message. #[test] From a1cf7493938adb43c1dcde5d6f2e02a669c2c884 Mon Sep 17 00:00:00 2001 From: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Date: Tue, 21 Jul 2026 09:03:50 -0700 Subject: [PATCH 4/4] fix(acp): retry unacknowledged observer frames Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> --- crates/buzz-acp/src/relay.rs | 79 ++++++++++++++++++++++++++++++++++-- 1 file changed, 75 insertions(+), 4 deletions(-) diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 99efa62a92..ce7d565f12 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -1027,7 +1027,11 @@ struct BgState { /// Bounded at `GATED_OBSERVER_QUEUE_CAP` (drop-oldest); drained by the /// main loop one frame per pacing tick once the gate clears. gated_observer_pending: VecDeque>, - /// Frames evicted from `gated_observer_pending` since the last drain-empty + /// Observer frames written to the socket but not yet acknowledged. The + /// relay's rate-limit NOTICE does not carry an event ID, so all unresolved + /// observer writes are moved back ahead of the parked FIFO when one arrives. + observer_in_flight: VecDeque>, + /// Frames evicted from the bounded pending/in-flight observer buffers since /// summary log. Makes overflow loss visible instead of silent. gated_observer_dropped: u64, /// Channels whose REQ failed during `resubscribe_after_reconnect`. @@ -1063,6 +1067,7 @@ impl BgState { membership_resub_needed: false, observer_resub_needed: false, gated_observer_pending: VecDeque::new(), + observer_in_flight: VecDeque::new(), gated_observer_dropped: 0, resubscribe_retry: HashSet::new(), backoff_step: 0, @@ -1178,6 +1183,41 @@ impl BgState { } self.gated_observer_pending.push_back(event); } + + /// Restore unresolved observer writes ahead of frames parked after the + /// gate armed. NOTICE has no event ID, so conservatively retry every frame + /// without an OK; duplicate IDs are harmless at the relay. + fn requeue_observer_in_flight(&mut self) { + while let Some(event) = self.observer_in_flight.pop_back() { + self.gated_observer_pending.push_front(event); + } + while self.gated_observer_pending.len() > GATED_OBSERVER_QUEUE_CAP { + self.gated_observer_pending.pop_front(); + self.gated_observer_dropped += 1; + } + } + + fn track_observer_in_flight(&mut self, event: Box) { + if self.observer_in_flight.len() >= GATED_OBSERVER_QUEUE_CAP { + self.observer_in_flight.pop_front(); + self.gated_observer_dropped += 1; + warn!( + dropped_total = self.gated_observer_dropped, + "observer acknowledgment window full — dropped oldest frame" + ); + } + self.observer_in_flight.push_back(event); + } + + fn acknowledge_observer_frame(&mut self, event_id: &str) { + if let Some(index) = self + .observer_in_flight + .iter() + .position(|event| event.id.to_hex() == event_id) + { + self.observer_in_flight.remove(index); + } + } } /// Record a command's intent in state while disconnected (no WebSocket). @@ -1445,9 +1485,12 @@ async fn execute_connected_command( // Best-effort: log a send failure but don't trigger reconnect — the // next ping or read will detect the dead socket. A failed observer // frame is parked so the post-reconnect drain redelivers it. - if !send_publish_event_frame(ws, &event).await - && event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME - { + let is_observer = event.kind.as_u16() as u32 == KIND_AGENT_OBSERVER_FRAME; + if send_publish_event_frame(ws, &event).await { + if is_observer { + state.track_observer_in_flight(event); + } + } else if is_observer { state.park_gated_observer_frame(event); } true @@ -2142,6 +2185,7 @@ async fn handle_ws_message( if message.starts_with("rate-limited:") { let secs = parse_rate_limit_retry_secs(&message).unwrap_or(0); let deadline = state.set_rate_limit_gate(secs); + state.requeue_observer_in_flight(); warn!( "rate-limit gate armed via NOTICE until ~{:.1}s from now", deadline @@ -2305,6 +2349,7 @@ async fn handle_ws_message( warn!("mid-session AUTH rejected (event {event_id}): {message} — triggering reconnect"); return false; } + state.acknowledge_observer_frame(&event_id); debug!("OK for event {event_id}: accepted={accepted} message={message}"); } } @@ -2588,6 +2633,7 @@ async fn drain_gated_observer_pending( state.gated_observer_pending.push_front(event); break; } + state.track_observer_in_flight(event); sent += 1; } if state.gated_observer_pending.is_empty() && state.gated_observer_dropped > 0 { @@ -2843,6 +2889,7 @@ async fn try_autonomous_reconnect( observer_control_tx: &mpsc::Sender, auth_tag: Option<&nostr::Tag>, ) -> ReconnectOutcome { + state.requeue_observer_in_flight(); // 5 attempts, up to 16s base backoff. Shares delay values with the // initial-connect retry in `HarnessRelay::connect()` (STARTUP_CONNECT_BACKOFFS) — // see its doc comment for how the two loops consume the array differently. @@ -2972,6 +3019,7 @@ async fn wait_for_reconnect( skip_drain: bool, auth_tag: Option<&nostr::Tag>, ) -> ReconnectOutcome { + state.requeue_observer_in_flight(); if !skip_drain { // Drain commands until we get Reconnect (or Shutdown). // Other commands update state so reconnect reflects latest intent. @@ -5858,6 +5906,29 @@ mod tests { assert!(state.gated_observer_pending.is_empty()); } + #[test] + fn observer_notice_requeues_unacknowledged_frames_and_ok_retires_them() { + let mut state = BgState::new(); + let keys = Keys::generate(); + let accepted = make_observer_frame(&keys); + let rejected = make_observer_frame(&keys); + let later = make_observer_frame(&keys); + + state.track_observer_in_flight(Box::new(accepted.clone())); + state.track_observer_in_flight(Box::new(rejected.clone())); + state.acknowledge_observer_frame(&accepted.id.to_hex()); + state.park_gated_observer_frame(Box::new(later.clone())); + state.requeue_observer_in_flight(); + + let ids: Vec<_> = state + .gated_observer_pending + .iter() + .map(|event| event.id) + .collect(); + assert_eq!(ids, [rejected.id, later.id]); + assert!(state.observer_in_flight.is_empty()); + } + /// The parked-frame queue is bounded: overflow evicts the oldest frame and /// counts it; the drain resets the counter after logging the summary. #[tokio::test]