perf(ledger): stream rows to disk instead of collecting them first (#155)

Scoring built the whole ledger as a `Vec<CallRecord>` and wrote it afterwards.
Every resolved row carries the recorded side's full `args` and `result`, so a run
with thousands of resolved calls held a second copy of its own recording in
memory, alongside the parsed lookup table and record graph, before a byte reached
disk.

That is what OOMKilled the runner. Established by controlled comparison rather
than by argument: on one 287-correlation recording, at one 16Gi limit, with one
deja build, the candidate with the `imc` seam disabled COMPLETES in 1m51 while
the unmodified candidate dies at 16Gi. Same 214 MB lookup table, same 100
correlations driven, same 67s seeding, same 8s driving. The only difference is
what the ledger holds: 82 resolved calls against thousands, because the disabled
seam makes almost every row `omitted` or `pruned_subtree` and those carry no
payload.

Every other explanation was eliminated by test first: table size (the passing run
loads the same one), deja revision (same build), correlation count (a
427-correlation recording scores in 88s), render and scoping (3.2s, timed), the
`Expected` result clone (removed in #152, still OOMs), and a hang (exit 137, the
container terminated).

`build_with_inconclusive` and `build_with_plan` now emit through a sink; the
collecting form is kept for `build_ledger`, which the `/calls` API genuinely
needs whole. `detect_and_score` streams straight into a `BufWriter`, so peak is
bounded by one row rather than by the run. Flat-tier rows stream first, exactly
as the collecting version ordered them.

Behaviour is unchanged — 443 tests pass, including the two that assert the
scorecard and the ledger classify identically, which is what would break if the
row set or its order moved.

Author: maverox

crates/deja-orchestrator/src/divergence/ledger.rs

@@ -199,6 +199,8 @@ pub fn build(
     )
 }
 
+/// Collecting wrapper over [`build_with_inconclusive_into`], for callers that genuinely
+/// want every row in memory (the `/calls` API).
 pub(crate) fn build_with_inconclusive(
     events: &[BoundaryEvent],
     observed: &[ObservedCall],
@@ -207,6 +209,39 @@ pub(crate) fn build_with_inconclusive(
     inconclusive_race: &InconclusiveRaceEvidence,
     tail_gap: &TailGapEvidence,
 ) -> Vec<CallRecord> {
+    let mut rows = Vec::new();
+    let _ = build_with_inconclusive_into(
+        events,
+        observed,
+        table,
+        idempotent_delete_demote,
+        inconclusive_race,
+        tail_gap,
+        &mut |row| {
+            rows.push(row);
+            Ok(())
+        },
+    );
+    rows
+}
+
+/// Emit each ledger row to `sink` as it is produced.
+///
+/// Streaming rather than returning a `Vec<CallRecord>`: every resolved row
+/// carries the recorded side's full `args` and `result`, so a run with
+/// thousands of resolved calls held a second copy of its own recording in
+/// memory before a byte reached disk. That OOMKilled the runner at 16 GiB on a
+/// 287-correlation tape, while the SAME tape scored fine for a candidate whose
+/// rows were overwhelmingly payload-free — 82 resolved calls against thousands.
+pub(crate) fn build_with_inconclusive_into(
+    events: &[BoundaryEvent],
+    observed: &[ObservedCall],
+    table: &deja::LookupTable,
+    idempotent_delete_demote: &HashSet<u64>,
+    inconclusive_race: &InconclusiveRaceEvidence,
+    tail_gap: &TailGapEvidence,
+    sink: &mut dyn FnMut(CallRecord) -> std::io::Result<()>,
+) -> std::io::Result<()> {
     let expected_seqs = &expected_sequences(table);
     let span_paths = &recorded_span_paths(table);
     let by_seq: HashMap<u64, &BoundaryEvent> =
@@ -222,7 +257,6 @@ pub(crate) fn build_with_inconclusive(
         })
     };
 
-    let mut rows: Vec<CallRecord> = Vec::new();
     let mut consumed: HashSet<u64> = HashSet::new();
 
     // Args-free pairing of recorded twins for execute-mode write consequences:
@@ -274,7 +308,7 @@ pub(crate) fn build_with_inconclusive(
             } else {
                 ("value_diverged".to_owned(), true)
             };
-            rows.push(CallRecord {
+            sink(CallRecord {
                 correlation_id: obs.correlation_id.clone(),
                 source_event_global_sequence: obs.source_event_global_sequence,
                 served_event_global_sequence: None,
@@ -287,7 +321,7 @@ pub(crate) fn build_with_inconclusive(
                 resolved_rank: obs.resolved_rank,
                 recorded,
                 observed: observed_side(obs).or_none(),
-            });
+            })?;
             continue;
         }
 
@@ -366,7 +400,7 @@ pub(crate) fn build_with_inconclusive(
                         }
                         _ => None,
                     });
-                rows.push(CallRecord {
+                sink(CallRecord {
                     correlation_id: obs.correlation_id.clone(),
                     source_event_global_sequence: Some(twin_seq),
                     served_event_global_sequence: None,
@@ -384,7 +418,7 @@ pub(crate) fn build_with_inconclusive(
                     resolved_rank: obs.resolved_rank,
                     recorded,
                     observed: observed.or_none(),
-                });
+                })?;
                 continue;
             }
         }
@@ -423,7 +457,7 @@ pub(crate) fn build_with_inconclusive(
             .source_event_global_sequence
             .and_then(recorded_for)
             .and_then(CallSide::or_none);
-        rows.push(CallRecord {
+        sink(CallRecord {
             correlation_id: obs.correlation_id.clone(),
             source_event_global_sequence: obs.source_event_global_sequence,
             served_event_global_sequence: None,
@@ -436,7 +470,7 @@ pub(crate) fn build_with_inconclusive(
             resolved_rank: obs.resolved_rank,
             recorded,
             observed: observed_side(obs).or_none(),
-        });
+        })?;
     }
 
     // --- omitted: expected (table-covered) recorded events never consumed ----
@@ -452,7 +486,7 @@ pub(crate) fn build_with_inconclusive(
             &ev.boundary,
             ev.role.as_deref(),
         );
-        rows.push(CallRecord {
+        sink(CallRecord {
             correlation_id: ev.correlation_id.clone(),
             source_event_global_sequence: Some(ev.global_sequence),
             served_event_global_sequence: None,
@@ -465,25 +499,38 @@ pub(crate) fn build_with_inconclusive(
             resolved_rank: None,
             recorded: recorded_for(ev.global_sequence),
             observed: None,
-        });
+        })?;
     }
 
-    rows
+    Ok(())
 }
 
 /// Build a ledger through the same per-correlation graph/flat seam as the
 /// scorecard. The all-flat arm delegates directly to the legacy builder; mixed
 /// runs remove graph correlations before doing so, so args-free pairing cannot
 /// claim an event owned by graph alignment.
-pub(crate) fn build_with_plan(
+/// Emit each ledger row to `sink` as it is produced.
+///
+/// Streaming rather than returning a `Vec<CallRecord>`: every resolved row
+/// carries the recorded side's full `args` and `result`, so a run with
+/// thousands of resolved calls held a second copy of its own recording in
+/// memory before a byte reached disk. That OOMKilled the runner at 16 GiB on a
+/// 287-correlation tape, while the SAME tape scored fine for a candidate whose
+/// rows were overwhelmingly payload-free — 82 resolved calls against thousands.
+// Eight because the sink is an eighth parameter on a function that already took
+// seven; bundling them into a struct would be a larger change than the one being
+// made and would obscure that this is the same function, streaming.
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn build_with_plan_into(
     events: &[BoundaryEvent],
     observed: &[ObservedCall],
     table: &deja::LookupTable,
     idempotent_delete_demote: &HashSet<u64>,
     inconclusive_race: &InconclusiveRaceEvidence,
     tail_gap: &TailGapEvidence,
     plan: &GraphScoringPlan,
-) -> Vec<CallRecord> {
+    sink: &mut dyn FnMut(CallRecord) -> std::io::Result<()>,
+) -> std::io::Result<()> {
     let graph_correlations: BTreeSet<&str> = events
         .iter()
         .filter_map(|event| event.correlation_id.as_deref())
@@ -496,13 +543,14 @@ pub(crate) fn build_with_plan(
         .collect();
 
     if graph_correlations.is_empty() {
-        return build_with_inconclusive(
+        return build_with_inconclusive_into(
             events,
             observed,
             table,
             idempotent_delete_demote,
             inconclusive_race,
             tail_gap,
+            sink,
         );
     }
     let by_seq: HashMap<u64, &BoundaryEvent> = events
@@ -581,14 +629,17 @@ pub(crate) fn build_with_plan(
         .entries
         .retain(|entry| !graph_sequence.contains(&entry.source_event_global_sequence));
 
-    let mut rows = build_with_inconclusive(
+    // Flat-tier rows stream through the same sink, and FIRST — preserving the
+    // order the collecting version produced (flat rows, then graph rows).
+    build_with_inconclusive_into(
         &flat_events,
         &flat_observed,
         &flat_table,
         idempotent_delete_demote,
         inconclusive_race,
         &flat_tail_gap,
-    );
+        sink,
+    )?;
     let span_paths = recorded_span_paths(table);
 
     for correlation_id in graph_correlations {
@@ -607,7 +658,7 @@ pub(crate) fn build_with_plan(
                         let event = by_seq[&sequence];
                         let mut side = recorded_side(event);
                         side.span_path = span_paths.get(&sequence).cloned();
-                        rows.push(CallRecord {
+                        sink(CallRecord {
                             correlation_id: event.correlation_id.clone(),
                             source_event_global_sequence: Some(sequence),
                             served_event_global_sequence: None,
@@ -624,7 +675,7 @@ pub(crate) fn build_with_plan(
                             resolved_rank: None,
                             recorded: side.or_none(),
                             observed: None,
-                        });
+                        })?;
                     }
                 }
                 NodeOutcome::NovelSubtree { events_below } => {
@@ -636,7 +687,7 @@ pub(crate) fn build_with_plan(
                         // space the evidence was measured in — no remap here.
                         let unrecorded_tail =
                             tail_gap.covers(call.correlation_id.as_deref(), index);
-                        rows.push(CallRecord {
+                        sink(CallRecord {
                             correlation_id: call.correlation_id.clone(),
                             source_event_global_sequence: None,
                             served_event_global_sequence: None,
@@ -655,7 +706,7 @@ pub(crate) fn build_with_plan(
                             resolved_rank: call.resolved_rank,
                             recorded: None,
                             observed: observed_side(call).or_none(),
-                        });
+                        })?;
                     }
                 }
                 outcome => {
@@ -792,7 +843,7 @@ pub(crate) fn build_with_plan(
                             )
                         };
 
-                    rows.push(CallRecord {
+                    sink(CallRecord {
                         correlation_id: call.correlation_id.clone(),
                         source_event_global_sequence: aligned_sequence,
                         served_event_global_sequence: served_sequence,
@@ -805,13 +856,13 @@ pub(crate) fn build_with_plan(
                         resolved_rank: call.resolved_rank,
                         recorded: recorded.and_then(CallSide::or_none),
                         observed: observed.or_none(),
-                    });
+                    })?;
                 }
             }
         }
     }
 
-    rows
+    Ok(())
 }
 
 /// The set of `global_sequence`s the lookup table covers (so http_incoming and

crates/deja-orchestrator/src/divergence/mod.rs

@@ -5632,13 +5632,11 @@ pub fn detect_and_score(root: &HarnessRoot, run_id: &str) -> io::Result<Scorecar
     crate::write_json(&path, &card)?;
 
     // Ledger: the per-call detail the scorecard summary drops. Best-effort.
-    match build_ledger_with_plan(&art, &graph_plan) {
-        Ok(rows) => {
-            if let Err(e) = write_ledger(&root.call_ledger_path(run_id), &rows) {
-                eprintln!("divergence: ledger write failed for {run_id}: {e}");
-            }
-        }
-        Err(e) => eprintln!("divergence: ledger build failed for {run_id}: {e}"),
+    // Streamed straight to the file. Collecting first held every resolved row —
+    // each carrying the recorded side's full `args` and `result` — alongside the
+    // parsed table and graph, which is what OOMKilled the runner on a dense tape.
+    if let Err(e) = stream_ledger(&root.call_ledger_path(run_id), &art, &graph_plan) {
+        eprintln!("divergence: ledger stream failed for {run_id}: {e}");
     }
     Ok(card)
 }
@@ -5654,13 +5652,19 @@ pub fn detect_and_score(root: &HarnessRoot, run_id: &str) -> io::Result<Scorecar
 /// drove attached to its ledger rows.
 pub fn build_ledger(art: &RunArtifacts) -> io::Result<Vec<CallRecord>> {
     let graph_plan = GraphScoringPlan::build(art);
-    build_ledger_with_plan(art, &graph_plan)
+    let mut rows = Vec::new();
+    build_ledger_into(art, &graph_plan, &mut |row| {
+        rows.push(row);
+        Ok(())
+    })?;
+    Ok(rows)
 }
 
-pub(crate) fn build_ledger_with_plan(
+pub(crate) fn build_ledger_into(
     art: &RunArtifacts,
     graph_plan: &GraphScoringPlan,
-) -> io::Result<Vec<CallRecord>> {
+    sink: &mut dyn FnMut(CallRecord) -> io::Result<()>,
+) -> io::Result<()> {
     let events = &art.events;
     let span_paths = ledger::recorded_span_paths(&art.table);
     // Mirror scorecard classification: discover race evidence under status-clean
@@ -5705,15 +5709,16 @@ pub(crate) fn build_ledger_with_plan(
             &document_clauses_for(&art.reply_canons, "http_incoming"),
         ),
     );
-    Ok(ledger::build_with_plan(
+    ledger::build_with_plan_into(
         events,
         &art.observed,
         &art.table,
         &idempotent_delete,
         &inconclusive_race,
         &tail_gap,
         graph_plan,
-    ))
+        sink,
+    )
 }
 
 /// Read-through ledger for `GET /runs/{id}/calls` (recomputes from artifacts;
@@ -5723,17 +5728,22 @@ pub fn call_ledger(root: &HarnessRoot, run_id: &str) -> io::Result<Vec<CallRecor
     build_ledger(&art)
 }
 
-fn write_ledger(path: &std::path::Path, rows: &[CallRecord]) -> io::Result<()> {
+/// Build and write the ledger without ever holding it whole.
+fn stream_ledger(
+    path: &std::path::Path,
+    art: &RunArtifacts,
+    graph_plan: &GraphScoringPlan,
+) -> io::Result<()> {
     use std::io::Write as _;
     if let Some(parent) = path.parent() {
         std::fs::create_dir_all(parent)?;
     }
     let mut out = std::io::BufWriter::new(std::fs::File::create(path)?);
-    for row in rows {
-        let line = serde_json::to_vec(row).map_err(io::Error::other)?;
+    build_ledger_into(art, graph_plan, &mut |row| {
+        let line = serde_json::to_vec(&row).map_err(io::Error::other)?;
         out.write_all(&line)?;
-        out.write_all(b"\n")?;
-    }
+        out.write_all(b"\n")
+    })?;
     out.flush()
 }