From e827a63d7e3b9fb7475e5418c599c6955dfb8af1 Mon Sep 17 00:00:00 2001 From: Sandesh Devaraju Date: Fri, 28 Aug 2026 11:50:24 -0700 Subject: [PATCH 1/2] streams: cap JSONL line size and transcript batch bytes Transcript batches were bounded by event count only (1000), and read_jsonl_line buffered lines of unbounded length. A transcript whose events embed file contents (normal for agent tool results) could put hundreds of MB into a single batch, which downstream redaction and metrics conversion amplify several times over -- ballooning daemon RSS past the memory watchdog within seconds (#2244). - read_jsonl_line: cap a single line at MAX_JSONL_LINE_BYTES (8 MiB). The read is byte-based (read_until), NOT read_line: the cap can slice a multi-byte character, and read_line's UTF-8 validation would return InvalidData instead of classifying the line as Oversized -- wedging the stream at a fixed watermark. Oversized lines are skipped without being buffered (skip_until); callers advance their watermark past them via the new JsonlLineState::Oversized state. Non-UTF-8 content within the cap keeps read_line's InvalidData contract. - All JSONL byte-offset stream parsers (claude, codex, copilot, cursor, droid, gemini, pi, windsurf): stop a batch early once MAX_BATCH_BYTES (8 MiB) of raw JSON has been accepted; remaining events arrive in later batches. Not covered here: amp, continue_cli, and opencode parse whole files into a DOM (serde_json::from_reader) and need a separate treatment -- called out in #2244 as follow-up. Part 1/7 of the #2244 fix stack. --- src/streams/agents/claude.rs | 16 ++++- src/streams/agents/codex.rs | 16 ++++- src/streams/agents/copilot.rs | 15 ++++- src/streams/agents/cursor.rs | 16 ++++- src/streams/agents/droid.rs | 16 ++++- src/streams/agents/gemini.rs | 16 ++++- src/streams/agents/pi.rs | 16 ++++- src/streams/agents/windsurf.rs | 16 ++++- src/streams/types.rs | 114 ++++++++++++++++++++++++++++++++- 9 files changed, 230 insertions(+), 11 deletions(-) diff --git a/src/streams/agents/claude.rs b/src/streams/agents/claude.rs index b7f4e9632a..8ea28613a9 100644 --- a/src/streams/agents/claude.rs +++ b/src/streams/agents/claude.rs @@ -186,6 +186,7 @@ impl Agent for ClaudeAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -203,6 +204,17 @@ impl Agent for ClaudeAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -222,8 +234,10 @@ impl Agent for ClaudeAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/codex.rs b/src/streams/agents/codex.rs index eecb76a2b9..a074f88ec4 100644 --- a/src/streams/agents/codex.rs +++ b/src/streams/agents/codex.rs @@ -250,6 +250,7 @@ impl Agent for CodexAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; let mut line = String::new(); @@ -267,6 +268,17 @@ impl Agent for CodexAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -286,8 +298,10 @@ impl Agent for CodexAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/copilot.rs b/src/streams/agents/copilot.rs index dacd97dc88..9359854373 100644 --- a/src/streams/agents/copilot.rs +++ b/src/streams/agents/copilot.rs @@ -527,6 +527,7 @@ pub(super) fn read_event_stream( })?; let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -545,6 +546,17 @@ pub(super) fn read_event_stream( line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -564,8 +576,9 @@ pub(super) fn read_event_stream( } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES { break; } } diff --git a/src/streams/agents/cursor.rs b/src/streams/agents/cursor.rs index 606fc6634a..a9b64a1e95 100644 --- a/src/streams/agents/cursor.rs +++ b/src/streams/agents/cursor.rs @@ -158,6 +158,7 @@ impl Agent for CursorAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -175,6 +176,17 @@ impl Agent for CursorAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -194,8 +206,10 @@ impl Agent for CursorAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/droid.rs b/src/streams/agents/droid.rs index 6dc37a4276..1c4fccac5d 100644 --- a/src/streams/agents/droid.rs +++ b/src/streams/agents/droid.rs @@ -165,6 +165,7 @@ impl Agent for DroidAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; let mut latest_timestamp: Option> = @@ -185,6 +186,17 @@ impl Agent for DroidAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } // Skip empty lines @@ -225,8 +237,10 @@ impl Agent for DroidAgent { } // Push raw JSON entry + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/gemini.rs b/src/streams/agents/gemini.rs index 449414fe0f..4f59b7f3dd 100644 --- a/src/streams/agents/gemini.rs +++ b/src/streams/agents/gemini.rs @@ -160,6 +160,7 @@ impl Agent for GeminiAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -177,6 +178,17 @@ impl Agent for GeminiAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -196,8 +208,10 @@ impl Agent for GeminiAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/pi.rs b/src/streams/agents/pi.rs index 2aad9ceb81..ce60c88eb9 100644 --- a/src/streams/agents/pi.rs +++ b/src/streams/agents/pi.rs @@ -91,6 +91,7 @@ impl Agent for PiAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -108,6 +109,17 @@ impl Agent for PiAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -127,8 +139,10 @@ impl Agent for PiAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/agents/windsurf.rs b/src/streams/agents/windsurf.rs index cae5a720c3..65ebdc1716 100644 --- a/src/streams/agents/windsurf.rs +++ b/src/streams/agents/windsurf.rs @@ -91,6 +91,7 @@ impl Agent for WindsurfAgent { let batch_limit = self.batch_size_hint(); let mut events = Vec::with_capacity(batch_limit); + let mut batch_bytes: usize = 0; let mut current_offset = start_offset; let mut line_number = 0; @@ -108,6 +109,17 @@ impl Agent for WindsurfAgent { line_number += 1; current_offset += bytes_read as u64; } + crate::streams::types::JsonlLineState::Oversized(bytes_read) => { + line_number += 1; + current_offset += bytes_read as u64; + tracing::warn!( + line = line_number, + path = %path.display(), + max_bytes = crate::streams::types::MAX_JSONL_LINE_BYTES, + "skipping oversized transcript line" + ); + continue; + } } if line.trim().is_empty() { @@ -127,8 +139,10 @@ impl Agent for WindsurfAgent { } }; + batch_bytes += line.len(); events.push(entry); - if events.len() >= batch_limit { + if events.len() >= batch_limit || batch_bytes >= crate::streams::types::MAX_BATCH_BYTES + { break; } } diff --git a/src/streams/types.rs b/src/streams/types.rs index 8a799b5b54..a3d7637551 100644 --- a/src/streams/types.rs +++ b/src/streams/types.rs @@ -1,9 +1,24 @@ //! Core types for transcript processing. -use std::io::BufRead; +use std::io::{BufRead, Read}; use std::time::Duration; +/// Upper bound on a single JSONL line. Lines beyond this are skipped rather +/// than buffered: `read_line` is otherwise unbounded, and a single multi-hundred-MB +/// transcript line can balloon daemon RSS past the memory watchdog's hard limit +/// (git-ai#2244). 8 MiB comfortably fits any legitimate transcript event. +pub const MAX_JSONL_LINE_BYTES: u64 = 8 * 1024 * 1024; + +/// Upper bound on the total raw bytes carried by one transcript batch. +/// Batches were previously capped by event count only (1000), so a transcript +/// whose events embed file contents could put hundreds of MB into a single +/// batch — which downstream redaction + metrics conversion amplifies several +/// times over (git-ai#2244). The batch loop stops early once this budget is +/// spent; remaining events arrive in later batches. +pub const MAX_BATCH_BYTES: usize = 8 * 1024 * 1024; + /// Result of reading a single line from a JSONL reader. +#[derive(Debug)] pub enum JsonlLineState { /// End of file reached. Eof, @@ -11,21 +26,47 @@ pub enum JsonlLineState { Partial, /// Complete line ready for processing. Contains bytes read. Complete(usize), + /// Line exceeded [`MAX_JSONL_LINE_BYTES`] and was skipped without being + /// buffered. Contains total bytes consumed (cap + remainder up to and + /// including the newline) so callers can advance their watermark past it. + Oversized(usize), } /// Read a line from a BufReader, detecting partial writes from concurrent writers. /// /// Returns `Eof` if no more data, `Partial` if the line lacks a trailing newline, -/// or `Complete(bytes)` on success. +/// `Complete(bytes)` on success, or `Oversized(bytes)` when the line exceeded +/// [`MAX_JSONL_LINE_BYTES`] (content is discarded, reader advanced past the newline). pub fn read_jsonl_line( reader: &mut impl BufRead, line: &mut String, ) -> std::io::Result { line.clear(); - let bytes_read = reader.read_line(line)?; + // Read raw bytes, not via read_line: the byte cap can slice a multi-byte + // character, and read_line's UTF-8 validation would then return + // InvalidData instead of letting us classify the line as Oversized — + // wedging the stream at a fixed watermark. + let mut buf = Vec::new(); + let bytes_read = reader + .by_ref() + .take(MAX_JSONL_LINE_BYTES) + .read_until(b'\n', &mut buf)?; if bytes_read == 0 { return Ok(JsonlLineState::Eof); } + if buf.last() != Some(&b'\n') && bytes_read as u64 == MAX_JSONL_LINE_BYTES { + // The cap was hit mid-line: discard what we buffered (no UTF-8 + // conversion is attempted on discarded content) and skip the rest of + // the physical line without storing it. If EOF arrives before the + // newline (giant line still being written), we still report + // Oversized — re-reading it later would OOM anyway. + drop(buf); + let skipped = reader.skip_until(b'\n')?; + return Ok(JsonlLineState::Oversized(bytes_read + skipped)); + } + // Same contract as read_line: non-UTF-8 content is an InvalidData error. + *line = String::from_utf8(buf) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; if !line.ends_with('\n') { return Ok(JsonlLineState::Partial); } @@ -195,4 +236,71 @@ mod tests { let r2 = read_jsonl_line(&mut reader, &mut line).unwrap(); assert!(matches!(r2, JsonlLineState::Partial)); } + + #[test] + fn test_read_jsonl_line_oversized_is_skipped_and_reader_recovers() { + let big = "x".repeat(MAX_JSONL_LINE_BYTES as usize + 100); + let data = format!("{big}\n{{\"ok\":1}}\n"); + let mut reader = std::io::BufReader::new(data.as_bytes()); + let mut line = String::new(); + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Oversized(consumed) => { + assert_eq!(consumed, big.len() + 1); + assert!(line.is_empty(), "oversized content must not be retained"); + } + _ => panic!("expected Oversized for a line beyond MAX_JSONL_LINE_BYTES"), + } + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Complete(_) => assert_eq!(line.trim_end(), "{\"ok\":1}"), + _ => panic!("expected the next line to parse normally"), + } + } + + #[test] + fn test_read_jsonl_line_oversized_without_newline_at_eof() { + let big = "x".repeat(MAX_JSONL_LINE_BYTES as usize + 50); + let mut reader = std::io::BufReader::new(big.as_bytes()); + let mut line = String::new(); + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Oversized(consumed) => assert_eq!(consumed, big.len()), + _ => panic!("expected Oversized even when the giant line lacks a newline"), + } + } + + #[test] + fn test_read_jsonl_line_oversized_multibyte_at_cap_boundary() { + // 8 MiB is not divisible by 3, so a line of 3-byte characters + // guarantees the byte cap slices mid-character. The reader must + // classify Oversized — a read_line-based implementation returns + // InvalidData here and wedges the stream at a fixed watermark. + let euro = "€"; // 3 bytes in UTF-8 + let big = euro.repeat((MAX_JSONL_LINE_BYTES as usize / 3) + 50); + let data = format!("{big}\n{{\"ok\":1}}\n"); + let mut reader = std::io::BufReader::new(data.as_bytes()); + let mut line = String::new(); + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Oversized(consumed) => assert_eq!(consumed, big.len() + 1), + _ => panic!("expected Oversized for a multi-byte giant line"), + } + + match read_jsonl_line(&mut reader, &mut line).unwrap() { + JsonlLineState::Complete(_) => assert_eq!(line.trim_end(), "{\"ok\":1}"), + _ => panic!("expected the next line to parse normally"), + } + } + + #[test] + fn test_read_jsonl_line_invalid_utf8_within_cap_still_errors() { + // Non-UTF-8 content on a normal-sized line keeps read_line's + // InvalidData contract. + let data: &[u8] = b"\xff\xfe bad bytes\n"; + let mut reader = std::io::BufReader::new(data); + let mut line = String::new(); + let err = read_jsonl_line(&mut reader, &mut line).unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + } } From 14e112992b94ec94fcafafca0f339fa399134ff7 Mon Sep 17 00:00:00 2001 From: Sandesh Devaraju Date: Fri, 28 Aug 2026 11:51:59 -0700 Subject: [PATCH 2/2] daemon: serialize transcript metric events incrementally store_metrics_in_db materialized a Vec of every re-serialized event while the full Vec (each holding the redacted JSON tree) was still alive -- two complete copies of a transcript batch resident at once (#2244). The stream worker now serializes each MetricEvent as it is built and drops the tree immediately; persistence takes the pre-serialized rows via the new persist_metric_jsons_blocking. Existing callers of persist_metrics_blocking are unchanged (it now delegates to the same insert path). Behavior note: previously one unserializable event failed the whole batch (all-or-nothing insert); the stream-worker path now drops the failing event with a warning and persists the rest. These are best-effort diagnostics, and partial persistence beats losing the batch. Part 2/7 of the #2244 fix stack (stacked on 1/7). --- src/daemon/stream_worker.rs | 19 +++++++++++++++---- src/daemon/telemetry_worker.rs | 19 ++++++++++++++++++- 2 files changed, 33 insertions(+), 5 deletions(-) diff --git a/src/daemon/stream_worker.rs b/src/daemon/stream_worker.rs index 593e148b8e..d6f9152b9c 100644 --- a/src/daemon/stream_worker.rs +++ b/src/daemon/stream_worker.rs @@ -1128,7 +1128,11 @@ impl StreamWorker { let batch_count = batch.events.len(); let is_otel_stream = task.stream_kind == "otel_traces"; - let metric_events: Vec = batch + // Serialize each event as it is built and let the MetricEvent + // (which still holds the redacted JSON tree) drop before the next + // one is constructed. Collecting Vec and serializing + // afterwards kept two full copies of the batch alive at once (#2244). + let metric_event_jsons: Vec = batch .events .into_iter() .enumerate() @@ -1164,7 +1168,7 @@ impl StreamWorker { let attrs_sparse = event_attrs.to_sparse(); let raw_event = redact_json_secrets(raw_event); - Some(if is_otel_stream { + let metric_event = if is_otel_stream { MetricEvent::from_values_with_timestamp( OtelTraceValues::with_ids(raw_event, eid, pid, tid), attrs_sparse, @@ -1176,11 +1180,18 @@ impl StreamWorker { attrs_sparse, Some(event_ts), ) - }) + }; + match serde_json::to_string(&metric_event) { + Ok(json) => Some(json), + Err(e) => { + tracing::warn!(%e, "telemetry: failed to serialize transcript metric event; dropping"); + None + } + } }) .collect(); - if let Err(e) = telemetry.persist_metrics_blocking(&metric_events) { + if let Err(e) = telemetry.persist_metric_jsons_blocking(&metric_event_jsons) { tracing::warn!(%e, "telemetry: failed to persist transcript metrics locally"); } diff --git a/src/daemon/telemetry_worker.rs b/src/daemon/telemetry_worker.rs index e15fc220b0..0f4f8e47c4 100644 --- a/src/daemon/telemetry_worker.rs +++ b/src/daemon/telemetry_worker.rs @@ -435,6 +435,19 @@ impl DaemonTelemetryWorkerHandle { store_metrics_in_db(events) } + /// Persist pre-serialized metric-event JSON rows. + /// + /// Streaming variant of [`Self::persist_metrics_blocking`] for large + /// transcript batches: the caller serializes each event as it is built + /// and drops the event immediately, so a batch's `MetricEvent` trees and + /// their JSON copies are never all resident at the same time (#2244). + pub fn persist_metric_jsons_blocking( + &self, + event_jsons: &[String], + ) -> Result, GitAiError> { + store_metric_jsons_in_db(event_jsons) + } + /// Submit telemetry envelopes synchronously (best-effort, non-blocking). /// /// Used by the daemon process's own `observability::log_*()` calls which @@ -970,6 +983,10 @@ fn store_metrics_in_db(events: &[MetricEvent]) -> Result, GitAiError> { .map(serde_json::to_string) .collect::>()?; + store_metric_jsons_in_db(&event_jsons) +} + +fn store_metric_jsons_in_db(event_jsons: &[String]) -> Result, GitAiError> { if event_jsons.is_empty() { return Ok(Vec::new()); } @@ -978,7 +995,7 @@ fn store_metrics_in_db(events: &[MetricEvent]) -> Result, GitAiError> { let mut db_lock = db .lock() .map_err(|_| GitAiError::Generic("metrics DB lock poisoned".to_string()))?; - db_lock.insert_events(&event_jsons) + db_lock.insert_events(event_jsons) } #[derive(Debug, Default, PartialEq, Eq)]