From a314975ce7bf9aa00baebfef874b6e69474a95cf Mon Sep 17 00:00:00 2001 From: Sanket Kanjalkar Date: Thu, 12 Nov 2026 14:37:33 -0800 Subject: [PATCH] perf(acp): reduce observer CPU and lock overhead (#9381) ## Summary The Agent Client Protocol (ACP) observer records agent activity for the owner's viewer. Sizing currently allocates JSON strings and rescans whole frames for each trimmed field. Count JSON bytes directly, reuse fitted sizes, and count changed fields when cheaper. Skip raw-message copies for disabled observers and copy replay payloads outside the mutex. Keep sequence numbers, timestamps, replay insertion, and publication under one lock to preserve replay/live ordering. This adds one `Arc` allocation per replay entry, and snapshots pin entries until copying completes. Replay retains the existing 1,000-event limit. ### Related issue Related to #7987. This PR targets `main` independently of #9366, which owns the byte-budget fix. The changes overlap in observer plumbing, so the second PR to land will need a rebase. ### Testing The new ignored probe exercises the production publish queue. Timing includes template clones and queue ingestion, before encryption. Five alternating fresh release processes per arm/case compared `75d38b6c2` with this commit on a busy host. Every pair retained identical queue bytes and event counts. ```sh cargo test --release -p buzz-acp --lib observer::perf_tests::publish_queue_probe -- --exact --ignored --nocapture ``` Set `BUZZ_OBSERVER_PROBE_EVENTS`, `BUZZ_OBSERVER_PROBE_BYTES`, and `BUZZ_OBSERVER_PROBE_FIELDS` for each row. Medians: | Events / string bytes per event / fields | Before | After | | --- | ---: | ---: | | 200 / 1 MiB / 64 | 5605 ms | 228 ms | | 1,000 / 1 MiB / 1 | 714 ms | 352 ms | | 10,000 / 1 KiB / 1 | 13.01 ms | 6.90 ms | A separate 20-trial integration run used the same production Chromium app build (`block/buzz-app@128cd4c2`) and four real ACP subprocesses. Three peers emitted 225 synthetic messages over five seconds, each with 1 MiB across 64 strings. A fourth emitted 50 markers, 100 ms apart. Each arm/mode used five fresh processes and browser contexts, with alternating arm order. The real observer, publisher, encryption/signing, browser broker, and Activity UI ran through a modeled relay transport. Marker timing used one controller clock, from release through transcript DOM commit plus two animation frames. Observer timestamps were not the timing origin. Observation-enabled medians: | Metric | Before | After | | --- | ---: | ---: | | Rust CPU time | 4.49 s | 0.45 s | | Marker-to-render p95 | 4471 ms | 1018 ms | | Sampled peak RSS | 342.3 MiB | 296.9 MiB | All 500 enabled-mode markers rendered exactly once; all 2,250 large events were delivered with the existing payload elision. Observation-disabled CPU stayed roughly flat (0.13 s to 0.12 s). The controlled publication-order regression also fails original `main` and passes this commit. These are stress-workload measurements, not a general app speedup. Native hosting and deployed-relay performance remain unmeasured. The publish cadence remains one frame per second. No CI timing threshold was added. Generated with Codex Signed-off-by: Sanket Kanjalkar Co-authored-by: Codex --- crates/buzz-acp/TESTING.md | 23 ++++ crates/buzz-acp/src/acp.rs | 18 ++- crates/buzz-acp/src/lib.rs | 142 ++++++++++++++++----- crates/buzz-acp/src/observer.rs | 39 ++++-- crates/buzz-acp/src/observer/perf_tests.rs | 134 +++++++++++++++++++ 5 files changed, 304 insertions(+), 52 deletions(-) create mode 100644 crates/buzz-acp/src/observer/perf_tests.rs diff --git a/crates/buzz-acp/TESTING.md b/crates/buzz-acp/TESTING.md index 528bac2138..00f3225aaf 100644 --- a/crates/buzz-acp/TESTING.md +++ b/crates/buzz-acp/TESTING.md @@ -59,3 +59,26 @@ Goose and uses its provider to invoke the native developer shell. Both operate only on temporary local repositories, verify commit/tag signatures and identity, check unrelated-remote credential scoping, and assert keyfile removal. They do not replace authenticated relay clone/push/readback testing. + +## Observer performance measurement + +An ignored local probe exercises the production publish queue and its JSON +fitter. The default workload is 200 events, each with 1 MiB across 64 string +leaves. It prints elapsed milliseconds and retained queue bytes/events; there +is no timing threshold or additional CI job. + +```sh +cargo test --release -p buzz-acp --lib observer::perf_tests::publish_queue_probe -- \ + --exact --ignored --nocapture + +# Under-budget control: +BUZZ_OBSERVER_PROBE_EVENTS=10000 BUZZ_OBSERVER_PROBE_BYTES=1024 \ + BUZZ_OBSERVER_PROBE_FIELDS=1 \ + cargo test --release -p buzz-acp --lib observer::perf_tests::publish_queue_probe -- \ + --exact --ignored --nocapture +``` + +Elapsed time includes cloning the template payload and queue ingestion; it +excludes template construction and later publication/encryption. Use release +builds, fresh processes, identical workloads, and repeated alternating runs +when comparing revisions. Queue bytes are serialized JSON, not heap or RSS. diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 026b2344a0..d01127dc15 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -660,12 +660,16 @@ impl AcpClient { /// Emit a semantic event to the local observer feed, if enabled. pub fn observe(&self, kind: impl Into, payload: serde_json::Value) { + self.observe_with(kind, || payload); + } + + fn observe_with(&self, kind: impl Into, payload: impl FnOnce() -> serde_json::Value) { if let Some(observer) = &self.observer { observer.emit( kind, self.observer_agent_index, &self.observer_context, - payload, + payload(), ); } } @@ -1157,7 +1161,7 @@ impl AcpClient { .await .map_err(|_| AcpError::WriteTimeout(WRITE_TIMEOUT))? .map_err(AcpError::Io)?; - self.observe("acp_write", value.clone()); + self.observe_with("acp_write", || value.clone()); Ok(()) } @@ -1314,7 +1318,7 @@ impl AcpClient { continue; } }; - self.observe("acp_read", msg.clone()); + self.observe_with("acp_read", || msg.clone()); // Check if this is a response to our expected request (has matching id // AND no `method` field — a `method` field means it's an agent-initiated @@ -1637,7 +1641,7 @@ impl AcpClient { continue; } }; - self.observe("acp_read", msg.clone()); + self.observe_with("acp_read", || msg.clone()); let activity_now = Instant::now(); idle_deadline = activity_now + idle_timeout; @@ -3399,6 +3403,8 @@ mod tests { let mut client = spawn_script(r#"while IFS= read -r line; do printf '%s\n' "$line"; done"#).await; let mut stdin = client.stdin.take().expect("test stdin"); + // Raw wire payload copies must be lazy when observation is disabled. + client.observe_with("acp_read", || panic!("disabled observer copied payload")); let observer = ObserverHandle::in_process(); let mut events = observer.subscribe(); client.set_observer(Some(observer), 0); @@ -3441,6 +3447,10 @@ mod tests { ); if let Ok(event) = events.try_recv() { assert_eq!(event.kind, "acp_read"); + assert_eq!( + event.payload, + serde_json::from_str::(&frame).unwrap() + ); break; } assert!( diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index e9f2eb7dbc..b9aa0cba96 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1221,8 +1221,7 @@ impl ObserverPublishQueue { // Pre-trim at enqueue so (a) byte accounting reflects what will ship // and (b) one oversized leaf cannot force every frame it touches into // whole-envelope elision downstream. - fit_observer_event_to_budget(&mut event); - let bytes = serialized_len(&event); + let bytes = fit_observer_event_to_budget(&mut event); self.pending_bytes += bytes; self.events.push_back((bytes, source_events, event)); } @@ -1638,44 +1637,55 @@ const OBSERVER_LEAF_RETAIN_BYTES: usize = 3_000; /// byte-identical. /// /// The cap is measured in SERIALIZED bytes (JSON escaping makes serialized -/// length differ from raw), so the stop condition is always a full reserialize -/// of the whole frame: that counts the envelope, the variable `Option` -/// IDs, and any elision markers exactly. No separate margin constant is needed. +/// length differ from raw). For small leaves, subtract the old serialized +/// size and add the replacement's size. JSON delimiters and the envelope stay +/// unchanged, so this accounting is exact. For a dominant leaf, recount the +/// fitted frame instead of traversing the large original leaf again. /// /// Termination is provable: each iteration elides the largest string leaf that -/// would STRICTLY shrink the serialized frame, then reserializes. Shrinkability -/// is re-evaluated against each leaf's CURRENT value, so a leaf already at its -/// retained floor can never be re-elided — the loop strictly decreases the -/// serialized length each pass and is bounded by the leaf count. When no leaf -/// can shrink the frame and it still overflows, the payload is replaced with a -/// tiny stub, which trivially fits. Monotone decrease, bounded below by the stub. +/// would STRICTLY shrink the serialized frame, then updates its size. +/// Shrinkability is re-evaluated against each leaf's CURRENT value: a shorter +/// removed-byte count can shorten an existing marker once more. Every pass +/// strictly decreases serialized length, so the loop terminates. When no leaf +/// can shrink the frame and it still overflows, replace the payload with a +/// small stub. Callers must still reject an oversized metadata envelope. /// -/// **Signature choice (`&mut`, double-serialize accepted):** on the common -/// under-budget path this serializes the frame once to decide it fits, then -/// `encrypt_observer_payload` serializes it again — one extra `to_string` of an -/// already-small frame. Reusing that string would mean changing buzz-core's -/// `encrypt_observer_payload` signature or adding a parallel encrypt path; both -/// are out of this change's scope (buzz-core stays untouched). The clean `&mut` -/// signature with one cheap redundant serialize is the deliberate tradeoff. -fn fit_observer_event_to_budget(event: &mut observer::ObserverEvent) { - if serialized_len(event) <= OBSERVER_MAX_PLAINTEXT_LEN { - return; +/// Returns the fitted serialized size so retention can account for bytes +/// without another traversal. Counting writes avoid allocating a second copy +/// of a potentially multi-megabyte source frame just to measure its size. +fn fit_observer_event_to_budget(event: &mut observer::ObserverEvent) -> usize { + let mut bytes = serialized_len(event); + if bytes <= OBSERVER_MAX_PLAINTEXT_LEN { + return bytes; } - // Raw size of the payload we are about to trim, captured before mutation so - // the stub's `originalBytes` reports source bytes discarded, not serialized - // overflow — consistent with the per-leaf marker's raw byte count. - let original_payload_bytes = serde_json::to_string(&event.payload) - .map(|s| s.len()) - .unwrap_or(0); + // Derive the original payload size from the measured frame and its small + // envelope, avoiding a second traversal of the oversized source payload. + // JSON null occupies four bytes in the temporary envelope. + let payload = event.payload.take(); + let envelope_bytes = serialized_len(event).saturating_sub(4); + event.payload = payload; + let original_payload_bytes = bytes.saturating_sub(envelope_bytes); - // Elide the largest shrinkable leaf, reserialize, repeat. Each successful - // elision strictly shrinks the serialized frame, and a floored leaf can - // never be re-elided, so the loop is bounded by the leaf count. + // Each successful elision strictly shrinks the serialized frame, so the + // loop terminates even when a shortened marker permits another elision. while let Some(leaf) = largest_shrinkable_leaf(&mut event.payload) { - elide_leaf(leaf); - if serialized_len(event) <= OBSERVER_MAX_PLAINTEXT_LEN { - return; + // A JSON string uses at most six bytes per source byte (\u0000), + // plus quotes. Use deltas only when even that upper bound is small + // relative to the frame; otherwise recount the now-smaller frame. + let leaf_bound = leaf + .as_str() + .map_or(0, |s| s.len().saturating_mul(6).saturating_add(2)); + if leaf_bound < bytes / 2 { + let old_leaf_bytes = serialized_len(leaf); + elide_leaf(leaf); + bytes = bytes - old_leaf_bytes + serialized_len(leaf); + } else { + elide_leaf(leaf); + bytes = serialized_len(event); + } + if bytes <= OBSERVER_MAX_PLAINTEXT_LEN { + return bytes; } } @@ -1685,10 +1695,27 @@ fn fit_observer_event_to_budget(event: &mut observer::ObserverEvent) { "elided": format!("{} payload too large", event.kind), "originalBytes": original_payload_bytes, }); + serialized_len(event) } -fn serialized_len(event: &observer::ObserverEvent) -> usize { - serde_json::to_string(event).map(|s| s.len()).unwrap_or(0) +fn serialized_len(value: &impl serde::Serialize) -> usize { + let mut counter = ObserverByteCounter(0); + serde_json::to_writer(&mut counter, value) + .map(|()| counter.0) + .unwrap_or(usize::MAX) +} + +struct ObserverByteCounter(usize); + +impl std::io::Write for ObserverByteCounter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0 = self.0.saturating_add(bytes.len()); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } } /// Find the longest string leaf that would STRICTLY shrink if elided, returning @@ -13566,6 +13593,38 @@ mod observer_payload_trim_tests { 20_000, "the smaller leaf is left untouched once the frame fits" ); + + // Many escaped leaves exercise incremental size accounting; enough + // shrink to fit while untouched leaves and section order survive. + for text in ["x", "\n\"\\", "🐝\u{0}"] { + let leaf = format!( + "{}{}{}", + "x".repeat(3_000), + text.repeat(4_000), + "y".repeat(3_000) + ); + let mut event = event_with_payload( + "acp_read", + serde_json::json!({"items": vec![leaf.clone(); 9], "small": "keep"}), + ); + let fitted_bytes = fit_observer_event_to_budget(&mut event); + assert_eq!(fitted_bytes, serialized(&event).len()); + assert!(fitted_bytes <= OBSERVER_MAX_PLAINTEXT_LEN); + let items = event.payload["items"] + .as_array() + .expect("trimmed sections survive"); + assert_eq!(items.len(), 9); + assert_eq!(event.payload["small"], "keep"); + let mut saw_untouched = false; + for item in items { + if item == &leaf { + saw_untouched = true; + } else { + assert!(!saw_untouched, "equal-size leaves elide in traversal order"); + assert!(item.as_str().unwrap().contains("…[elided")); + } + } + } } #[test] @@ -13617,6 +13676,19 @@ mod observer_payload_trim_tests { "fell back to the stub" ); assert!(event.payload.get("originalBytes").is_some()); + + // An escaped leaf just above its retained floor is small relative to + // a non-shrinkable numeric array. Delta accounting must include the + // replacement, or it would incorrectly accept this oversized frame. + let mut event = event_with_payload( + "acp_read", + serde_json::json!({"numbers": vec![0; 18_500], "edge": "\u{0}".repeat(6_100)}), + ); + let original_payload_bytes = serialized_len(&event.payload); + let fitted_bytes = fit_observer_event_to_budget(&mut event); + assert_eq!(fitted_bytes, serialized(&event).len()); + assert!(fitted_bytes <= OBSERVER_MAX_PLAINTEXT_LEN); + assert_eq!(event.payload["originalBytes"], original_payload_bytes); } #[test] diff --git a/crates/buzz-acp/src/observer.rs b/crates/buzz-acp/src/observer.rs index 7029e5af6d..1861c55e13 100644 --- a/crates/buzz-acp/src/observer.rs +++ b/crates/buzz-acp/src/observer.rs @@ -15,6 +15,9 @@ use std::{ use serde::Serialize; use tokio::sync::broadcast; +#[cfg(test)] +mod perf_tests; + const OBSERVER_BUFFER_CAP: usize = 1_000; /// Best-effort metadata attached to observer events. @@ -38,7 +41,7 @@ pub struct ObserverHandle { struct ObserverInner { tx: broadcast::Sender, - buffer: Mutex>, + buffer: Mutex>>, seq: AtomicU64, } @@ -91,13 +94,15 @@ impl ObserverHandle { /// Return the current replay buffer. pub fn snapshot(&self) -> Vec { - match self.inner.buffer.lock() { + // Pin the replay window under the lock, then copy JSON after releasing it. + let events: Vec<_> = match self.inner.buffer.lock() { Ok(buffer) => buffer.iter().cloned().collect(), Err(error) => { tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}"); Vec::new() } - } + }; + events.into_iter().map(|event| (*event).clone()).collect() } /// Emit a local observer event. @@ -108,9 +113,9 @@ impl ObserverHandle { context: &ObserverContext, payload: serde_json::Value, ) { - let event = ObserverEvent { - seq: self.inner.seq.fetch_add(1, Ordering::Relaxed), - timestamp: chrono::Utc::now().to_rfc3339(), + let mut event = ObserverEvent { + seq: 0, + timestamp: String::new(), kind: kind.into(), agent_index, channel_id: context.channel_id.clone(), @@ -120,17 +125,25 @@ impl ObserverHandle { payload, }; - match self.inner.buffer.lock() { - Ok(mut buffer) => { - if buffer.len() >= OBSERVER_BUFFER_CAP { - buffer.pop_front(); - } - buffer.push_back(event.clone()); - } + // Copy the payload before locking so other emitters can proceed. + let mut replay = event.clone(); + let mut buffer = match self.inner.buffer.lock() { + Ok(buffer) => buffer, Err(error) => { tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}"); + return; } + }; + // Assign both fields with publication: desktop sorts by timestamp, + // and the publisher deduplicates snapshots by sequence high-water mark. + event.seq = self.inner.seq.fetch_add(1, Ordering::Relaxed); + event.timestamp = chrono::Utc::now().to_rfc3339(); + replay.seq = event.seq; + replay.timestamp.clone_from(&event.timestamp); + if buffer.len() >= OBSERVER_BUFFER_CAP { + buffer.pop_front(); } + buffer.push_back(Arc::new(replay)); let _ = self.inner.tx.send(event); } diff --git a/crates/buzz-acp/src/observer/perf_tests.rs b/crates/buzz-acp/src/observer/perf_tests.rs new file mode 100644 index 0000000000..1f5e44228d --- /dev/null +++ b/crates/buzz-acp/src/observer/perf_tests.rs @@ -0,0 +1,134 @@ +use super::*; +use serde_json::json; +use tokio::sync::broadcast::error::TryRecvError; + +struct SlowKind { + started: std::sync::mpsc::Sender<()>, + resume: std::sync::mpsc::Receiver<()>, +} + +impl From for String { + fn from(kind: SlowKind) -> Self { + kind.started.send(()).unwrap(); + kind.resume.recv().unwrap(); + "slow".into() + } +} + +#[test] +fn concurrent_emitters_keep_live_and_replay_sequence_order() { + let observer = ObserverHandle::in_process(); + let mut live = observer.subscribe(); + let context = context_for_turn( + None, + Some("session".into()), + "turn".into(), + "2026-11-01T00:00:00Z".into(), + ); + let payload = json!({"text":"hello 🐝\n\"world\"", "body":"x".repeat(70_000)}); + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let (resume_tx, resume_rx) = std::sync::mpsc::channel(); + let (observer_ref, context_ref, payload_ref) = (&observer, &context, &payload); + let (mut handoff_live, mut handoff_snapshot) = std::thread::scope(move |scope| { + scope.spawn(move || { + observer_ref.emit( + SlowKind { + started: started_tx, + resume: resume_rx, + }, + Some(0), + context_ref, + payload_ref.clone(), + ) + }); + // Pause one real emit during preparation, then let another publish. + // No timing assertion: the channel controls this interleaving exactly. + started_rx + .recv_timeout(std::time::Duration::from_secs(10)) + .unwrap(); + observer_ref.emit("fast", Some(1), context_ref, payload_ref.clone()); + let handoff_live = observer_ref.subscribe(); + let snapshot = observer_ref.snapshot(); + resume_tx.send(()).unwrap(); + (handoff_live, snapshot) + }); + let snapshot = observer.snapshot(); + assert_eq!(snapshot.len(), 2); + for (index, kind) in ["fast", "slow"].into_iter().enumerate() { + let event = live.try_recv().unwrap(); + assert_eq!(event.seq, index as u64 + 1); + assert_eq!(event.kind, kind); + assert!(chrono::DateTime::parse_from_rfc3339(&event.timestamp).is_ok()); + assert_eq!(event.payload, payload); + assert_eq!(event.session_id, context.session_id); + assert_eq!(event.turn_id, context.turn_id); + assert_eq!(event.started_at, context.started_at); + assert_eq!( + serde_json::to_value(&event).unwrap(), + serde_json::to_value(&snapshot[index]).unwrap() + ); + } + assert!(matches!(live.try_recv(), Err(TryRecvError::Empty))); + let high_water = handoff_snapshot.last().map_or(0, |event| event.seq); + while let Ok(event) = handoff_live.try_recv() { + if event.seq > high_water { + handoff_snapshot.push(event); + } + } + assert_eq!( + serde_json::to_value(&handoff_snapshot).unwrap(), + serde_json::to_value(&snapshot).unwrap(), + "snapshot/live handoff must neither skip nor duplicate the slow event" + ); + for n in 0..1_010 { + observer.emit("marker", None, &context, json!({"n":n})); + } + let snapshot = observer.snapshot(); + assert_eq!(snapshot.len(), 1_000); + assert_eq!(snapshot.first().unwrap().payload["n"], 10); + assert_eq!(snapshot.last().unwrap().payload["n"], 1_009); +} + +/// Local measurement only: the production publish queue, with no timing gate. +#[test] +#[ignore = "manual observer fitting measurement"] +fn publish_queue_probe() { + let count: usize = std::env::var("BUZZ_OBSERVER_PROBE_EVENTS") + .unwrap_or_else(|_| "200".into()) + .parse() + .unwrap(); + let payload_bytes: usize = std::env::var("BUZZ_OBSERVER_PROBE_BYTES") + .unwrap_or_else(|_| "1048576".into()) + .parse() + .unwrap(); + let fields: usize = std::env::var("BUZZ_OBSERVER_PROBE_FIELDS") + .unwrap_or_else(|_| "64".into()) + .parse() + .unwrap(); + assert!(fields > 0); + let mut template = ObserverEvent { + seq: 0, + timestamp: "2026-11-01T00:00:00Z".into(), + kind: "acp_read".into(), + agent_index: None, + channel_id: None, + session_id: None, + turn_id: None, + started_at: None, + payload: json!({"body": (0..fields).map(|n| + "x".repeat(payload_bytes / fields + usize::from(n < payload_bytes % fields)) + ).collect::>() }), + }; + let mut queue = crate::ObserverPublishQueue::default(); + let start = std::time::Instant::now(); + for n in 0..count { + template.seq = n as u64 + 1; + queue.ingest(template.clone()); + } + let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0; + println!( + "{}", + json!({"events": count, "payloadBytes": payload_bytes, "fields": fields, + "elapsedMs": elapsed_ms, "queuedEvents": queue.events.len(), "queuedBytes": queue.pending_bytes}) + ); +}