reader (Rust): stream JSONL hot loops; use memchr crate (#372)

* reader (Rust): stream JSONL hot loops; use memchr crate

The Claude / Codex incremental and prescan paths previously built a
multi-GB up-front buffer (`vec![0u8; (size - start_offset) as usize]`,
`Vec::with_capacity((size - start_offset) as usize)` + `read_to_end`)
to scan a session log. Switch to BufReader + `read_until(b'\n', ...)`
into a reused line buffer so only the longest single line stays
resident, regardless of file size.

The main `parse_claude_session` loop also moves off
`BufReader::lines()` (a fresh `String` per line) onto `read_line` into
a reused `String`, keeping per-line allocation bounded by the longest
line for sessions with tens of thousands of turns.

`memchr_newline` in the codex parser was named for `memchr` but did
`buf.iter().position(|&b| b == b'\n')`. Wire the actual `memchr` crate
(already a transitive dep through `regex`) for SIMD-accelerated line
splits; pinned at the workspace root and depended on directly from
`relayburn-sdk` so we own the version.

Closes #323.

* codex reader: propagate read_until I/O errors

Address review feedback on #372: the codex streaming loop was
swallowing `read_until` failures via `Err(_) => break`, which would
silently truncate the parse at a transient mid-file read error and
advance the resume cursor as if the bytes had been processed. The
claude.rs equivalents already use `?` to propagate.

Bubble the error by changing `parse_codex_buffer` to return
`std::io::Result<ParseCodexIncrementalResult>` and using `?` in the
loop, matching `parse_codex_session_incremental`'s outer signature.

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: willwashburn

CHANGELOG.md

@@ -4,6 +4,8 @@ Cross-package release notes for relayburn. Package changelogs contain package-le
 
 ## [Unreleased]
 
+- `relayburn-sdk` (Rust): reader hot loops in `claude.rs` and `codex.rs` now stream JSONL line-by-line via `BufReader::read_until` instead of pre-allocating a `(size - start_offset)`-byte buffer up front; only the longest single line stays resident. `memchr_newline` in the codex parser now actually uses the `memchr` crate for SIMD-accelerated newline scanning. The main `parse_claude_session` loop also drops `BufReader::lines()` in favor of `read_line` into a reused `String`. (#323)
+
 ## [2.0.0] - 2026-05-07
 
 - `relayburn-sdk` (Rust): default ledger home moves from `~/.relayburn` to `~/.agentworkforce/burn` so the Rust 2.0 port and the TS 1.x package can coexist on disk during the #249 cutover. `RELAYBURN_HOME` (and the per-DB path overrides) continue to override the path; TS 1.x users on `~/.relayburn` are unaffected. Rust-port testers with data under the old path can `mv ~/.relayburn ~/.agentworkforce/burn` to carry it over (formats are not compatible — Rust treats any non-2.0 layout as empty and requires a `burn ingest` re-population).

Cargo.lock

@@ -17,3 +17,4 @@ anyhow = "1"
 thiserror = "2"
 clap = { version = "4", features = ["derive"] }
 rusqlite = { version = "0.32", features = ["bundled"] }
+memchr = "2"

Cargo.toml

@@ -33,6 +33,10 @@ sha2 = "0.10"
 hex = "0.4"
 regex = "1"
 phf = { version = "0.11", features = ["macros"] }
+# reader: SIMD-accelerated newline split for the JSONL hot loops in
+# `reader/{claude,codex}.rs`. `regex` already pulls memchr in
+# transitively; we depend on it directly so we own the version.
+memchr = { workspace = true }
 
 # ledger: SQLite events + content store
 rusqlite = { workspace = true }

crates/relayburn-sdk/Cargo.toml

@@ -96,13 +96,21 @@ pub fn parse_claude_session_with_counter<P: AsRef<Path>, C: TokenCounter + ?Size
     let capture_content = matches!(content_mode, ContentStoreMode::Full);
 
     let file = File::open(path)?;
-    let reader = BufReader::new(file);
+    let mut reader = BufReader::new(file);
 
     let mut state = ParseState::new(options, path);
 
-    for line in reader.lines() {
-        let line = line?;
-        state.ingest_line(&line, counter, capture_content);
+    // `BufReader::lines()` allocates a fresh `String` per line; for sessions
+    // with tens of thousands of turns that's pure churn. `read_line` into a
+    // single reused buffer keeps allocation bounded by the longest line.
+    let mut line = String::new();
+    loop {
+        line.clear();
+        match reader.read_line(&mut line) {
+            Ok(0) => break,
+            Ok(_) => state.ingest_line(&line, counter, capture_content),
+            Err(err) => return Err(err),
+        }
     }
 
     Ok(state.finish(options, capture_content))
@@ -2290,7 +2298,7 @@ fn prescan_nodes(
             next_event_index: 0,
         });
     }
-    let mut file = File::open(path)?;
+    let file = File::open(path)?;
     let size = file.metadata()?.len();
     let length = end_offset.min(size);
     if length == 0 {
@@ -2299,18 +2307,29 @@ fn prescan_nodes(
             next_event_index: 0,
         });
     }
-    let mut buf = vec![0u8; length as usize];
-    file.read_exact(&mut buf)?;
-    let mut p: usize = 0;
+    // Stream the prefix line-by-line rather than reading `[0, length)`
+    // into memory all at once. For multi-GB sessions the up-front
+    // `vec![0u8; length as usize]` was a multi-GB allocation we never
+    // need — only the longest single line has to fit in memory.
+    let mut reader = BufReader::new(file).take(length);
+    let mut line_buf: Vec<u8> = Vec::new();
     let mut last_assistant_message_id: Option<String> = None;
     let mut next_event_index: u64 = 0;
-    while p < buf.len() {
-        let nl_idx = match buf[p..].iter().position(|&b| b == b'\n') {
-            Some(i) => p + i,
-            None => break,
-        };
-        let raw = std::str::from_utf8(&buf[p..nl_idx]).unwrap_or("").trim();
-        p = nl_idx + 1;
+    loop {
+        line_buf.clear();
+        let n = reader.read_until(b'\n', &mut line_buf)?;
+        if n == 0 {
+            break;
+        }
+        // A trailing partial line (no `\n`) inside the prescan window
+        // should never happen — incremental ingest only commits cursors
+        // at newline boundaries — but guard anyway.
+        if line_buf.last() != Some(&b'\n') {
+            break;
+        }
+        let raw = std::str::from_utf8(&line_buf[..n - 1])
+            .unwrap_or("")
+            .trim();
         if raw.is_empty() {
             continue;
         }
@@ -2499,20 +2518,30 @@ fn run_incremental<C: TokenCounter + ?Sized>(
 
     let mut file = File::open(path)?;
     file.seek(SeekFrom::Start(start_offset))?;
-    let mut buf: Vec<u8> = Vec::with_capacity((size - start_offset) as usize);
-    file.read_to_end(&mut buf)?;
-
-    let mut p: usize = 0;
+    // Stream from `start_offset` line-by-line. The previous implementation
+    // allocated `Vec::with_capacity((size - start_offset) as usize)` and
+    // `read_to_end` into it — for a multi-GB session this was a multi-GB
+    // up-front allocation. With BufReader + `read_until` only the longest
+    // single line stays resident.
+    let mut reader = BufReader::new(file);
+    let mut line_buf: Vec<u8> = Vec::new();
     let mut cursor_offset: u64 = start_offset; // position past last complete \n
-    while p < buf.len() {
-        let nl_idx = match buf[p..].iter().position(|&b| b == b'\n') {
-            Some(i) => p + i,
-            None => break,
-        };
-        let line_start_offset = start_offset + p as u64;
-        let line_end_offset = start_offset + nl_idx as u64 + 1;
-        let trimmed = std::str::from_utf8(&buf[p..nl_idx]).unwrap_or("").trim();
-        p = nl_idx + 1;
+    loop {
+        line_buf.clear();
+        let n = reader.read_until(b'\n', &mut line_buf)?;
+        if n == 0 {
+            break;
+        }
+        // Drop trailing partial lines — the next incremental call resumes
+        // from `cursor_offset`, which we only advance past complete `\n`.
+        if line_buf.last() != Some(&b'\n') {
+            break;
+        }
+        let line_start_offset = cursor_offset;
+        let line_end_offset = cursor_offset + n as u64;
+        let trimmed = std::str::from_utf8(&line_buf[..n - 1])
+            .unwrap_or("")
+            .trim();
         cursor_offset = line_end_offset;
         if trimmed.is_empty() {
             continue;

crates/relayburn-sdk/src/reader/claude.rs

@@ -12,7 +12,7 @@
 
 use std::collections::{BTreeMap, BTreeSet, HashMap};
 use std::fs::File;
-use std::io::{Read, Seek, SeekFrom};
+use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
 use std::path::Path;
 
 use serde_json::Value;
@@ -192,17 +192,16 @@ pub fn parse_codex_session_incremental(
             resume: clone_resume(options.resume.as_ref()),
         });
     }
-    let mut buf = vec![0u8; (size - start_offset) as usize];
     file.seek(SeekFrom::Start(start_offset))?;
-    file.read_exact(&mut buf)?;
+    // Stream from `start_offset` line-by-line. The previous implementation
+    // pre-allocated `vec![0u8; (size - start_offset) as usize]` and
+    // `read_exact` into it — for a multi-GB session that was a multi-GB
+    // up-front allocation. With BufReader + `read_until` only the longest
+    // single line stays resident.
+    let reader = BufReader::new(file);
 
     let project_resolver = ProjectResolver::new();
-    Ok(parse_codex_buffer(
-        &buf,
-        start_offset,
-        options,
-        &project_resolver,
-    ))
+    parse_codex_buffer(reader, start_offset, options, &project_resolver)
 }
 
 // ---------------------------------------------------------------------------
@@ -335,12 +334,12 @@ struct Pending<T> {
     record: T,
 }
 
-fn parse_codex_buffer(
-    buf: &[u8],
+fn parse_codex_buffer<R: BufRead>(
+    mut reader: R,
     start_offset: u64,
     options: &ParseCodexIncrementalOptions,
     project_resolver: &ProjectResolver,
-) -> ParseCodexIncrementalResult {
+) -> std::io::Result<ParseCodexIncrementalResult> {
     let capture_content = matches!(options.content_mode, Some(ContentStoreMode::Full));
     // Validated by `resolve_token_counter` at the public entry point.
     let counter = HeuristicCounter;
@@ -392,16 +391,24 @@ fn parse_codex_buffer(
     let mut committed_tool_result_counters = tool_result_counters.clone();
     let mut committed_last_completed_turn = last_completed_turn.clone();
 
-    let mut p: usize = 0;
-    while p < buf.len() {
-        let nl_idx = match find_newline(&buf[p..]) {
-            Some(idx) => p + idx,
-            None => break,
-        };
-        let line_end_offset = start_offset + (nl_idx as u64) + 1;
-        let raw = &buf[p..nl_idx];
-        p = nl_idx + 1;
-        let text = std::str::from_utf8(raw).unwrap_or("").trim();
+    let mut line_buf: Vec<u8> = Vec::new();
+    let mut current_offset: u64 = start_offset;
+    loop {
+        line_buf.clear();
+        let n = reader.read_until(b'\n', &mut line_buf)?;
+        if n == 0 {
+            break;
+        }
+        // Drop trailing partial lines — the next incremental call resumes
+        // from the committed end offset, which only advances past `\n`.
+        if line_buf.last() != Some(&b'\n') {
+            break;
+        }
+        let line_end_offset = current_offset + n as u64;
+        current_offset = line_end_offset;
+        let text = std::str::from_utf8(&line_buf[..n - 1])
+            .unwrap_or("")
+            .trim();
         if text.is_empty() {
             continue;
         }
@@ -1185,7 +1192,19 @@ fn parse_codex_buffer(
         }
     }
 
-    ParseCodexIncrementalResult {
+    // Silence unused-mutable warnings for snapshot mirrors that are written
+    // but only read indirectly.
+    let _ = (
+        cumulative,
+        session_id,
+        session_cwd,
+        turn_contexts,
+        seen_session_meta_keys,
+        root_session_emitted,
+        last_completed_turn,
+    );
+
+    Ok(ParseCodexIncrementalResult {
         turns,
         content: content_out,
         events: events_out,
@@ -1194,7 +1213,7 @@ fn parse_codex_buffer(
         tool_result_events: tool_events_out,
         end_offset: committed_end_offset,
         resume,
-    }
+    })
 }
 
 // ---------------------------------------------------------------------------
@@ -1217,10 +1236,6 @@ fn resolve_token_counter(
     }
 }
 
-fn find_newline(buf: &[u8]) -> Option<usize> {
-    buf.iter().position(|&b| b == b'\n')
-}
-
 fn session_meta_payload_id(payload: &Value) -> Option<String> {
     let id = payload.get("id")?.as_str()?;
     if id.is_empty() {