From d57367624132c07768aca2b4354f183464d825e0 Mon Sep 17 00:00:00 2001
From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com>
Date: Mon, 14 Sep 2026 19:02:28 +0000
Subject: [PATCH] fix(streams): stop a bad byte from stalling transcript
 ingestion

read_jsonl_line read each line with BufRead::read_line into a String, so
any byte that was not valid UTF-8 returned an InvalidData io error. All
eight agent readers wrapped that as StreamError::Transient, which the
stream worker retried four times and then dropped. The condition is a
property of the bytes at a fixed offset, not a temporary fault, so no
retry could clear it. The watermark never advanced past that offset, and
every later event in the session transcript was lost silently.

Read the line as raw bytes with read_until(b'\n') and decode with
String::from_utf8_lossy. A complete line that is not valid UTF-8 now
decodes lossily and advances the offset; if the result is not valid JSON
the reader skips it through the existing malformed-JSON path. A line
without a trailing newline stays Partial, which also covers a half-written
multi-byte character at the tail of a growing file.

The signature is unchanged, so all eight readers keep working as before.

Generated-By: PostHog Desktop
Task-Id: e8a82875-b221-4b69-bc9d-0dcc050819ca
---
 src/streams/types.rs | 53 ++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 51 insertions(+), 2 deletions(-)

diff --git a/src/streams/types.rs b/src/streams/types.rs
index 8a799b5..55e156f 100644
--- a/src/streams/types.rs
+++ b/src/streams/types.rs
@@ -17,18 +17,27 @@ pub enum JsonlLineState {
 ///
 /// Returns `Eof` if no more data, `Partial` if the line lacks a trailing newline,
 /// or `Complete(bytes)` on success.
+///
+/// The line is read as raw bytes and decoded with `String::from_utf8_lossy`, so a
+/// byte sequence that is not valid UTF-8 yields replacement characters instead of an
+/// error. This keeps a single bad byte from permanently stalling ingestion at its
+/// offset; the caller advances by the returned raw byte count regardless. A line
+/// without a trailing newline is still `Partial`, which also covers a half-written
+/// multi-byte character at the tail of a growing file.
 pub fn read_jsonl_line(
     reader: &mut impl BufRead,
     line: &mut String,
 ) -> std::io::Result<JsonlLineState> {
     line.clear();
-    let bytes_read = reader.read_line(line)?;
+    let mut buf = Vec::new();
+    let bytes_read = reader.read_until(b'\n', &mut buf)?;
     if bytes_read == 0 {
         return Ok(JsonlLineState::Eof);
     }
-    if !line.ends_with('\n') {
+    if buf.last() != Some(&b'\n') {
         return Ok(JsonlLineState::Partial);
     }
+    line.push_str(&String::from_utf8_lossy(&buf));
     Ok(JsonlLineState::Complete(bytes_read))
 }
 
@@ -195,4 +204,44 @@ mod tests {
         let r2 = read_jsonl_line(&mut reader, &mut line).unwrap();
         assert!(matches!(r2, JsonlLineState::Partial));
     }
+
+    #[test]
+    fn test_read_jsonl_line_invalid_utf8_complete() {
+        // A complete line (trailing newline) with a byte that is not valid UTF-8.
+        // The old read_line path returned InvalidData here, which stalled ingestion
+        // forever. It must now decode lossily and report the raw byte count.
+        let data = b"{\"a\":\"\xff\"}\n";
+        let mut reader = std::io::BufReader::new(&data[..]);
+        let mut line = String::new();
+        let result = read_jsonl_line(&mut reader, &mut line).unwrap();
+        assert!(matches!(result, JsonlLineState::Complete(n) if n == data.len()));
+        assert!(line.ends_with('\n'));
+        assert!(line.contains('\u{FFFD}'));
+    }
+
+    #[test]
+    fn test_read_jsonl_line_invalid_utf8_then_next_line() {
+        // A bad byte on one line must not block the following lines from being read.
+        let data = b"{\"a\":\"\xff\"}\n{\"b\":2}\n";
+        let mut reader = std::io::BufReader::new(&data[..]);
+        let mut line = String::new();
+
+        let r1 = read_jsonl_line(&mut reader, &mut line).unwrap();
+        assert!(matches!(r1, JsonlLineState::Complete(_)));
+
+        let r2 = read_jsonl_line(&mut reader, &mut line).unwrap();
+        assert!(matches!(r2, JsonlLineState::Complete(8)));
+        assert_eq!(line, "{\"b\":2}\n");
+    }
+
+    #[test]
+    fn test_read_jsonl_line_partial_invalid_utf8() {
+        // A half-written multi-byte character at the tail of a growing file has no
+        // trailing newline, so it stays Partial (a genuinely transient state).
+        let data = b"{\"a\":\"\xe2\x82";
+        let mut reader = std::io::BufReader::new(&data[..]);
+        let mut line = String::new();
+        let result = read_jsonl_line(&mut reader, &mut line).unwrap();
+        assert!(matches!(result, JsonlLineState::Partial));
+    }
 }