ADR-0017: Per-Runner Durability Ordering Map

Status: Accepted Date: 2026-05-30

Context

Eight runners produce output parts (file + manifest entry + state row + journal event) on the destination + state-store boundary:

• pipeline::single::run_single_export (Snapshot, Incremental)
• pipeline::keyset::run_keyset (Keyset / Page)
• pipeline::chunked::exec::run_chunked_sequential (Chunked, single thread)
• pipeline::chunked::exec::run_chunked_parallel (Chunked, thread pool)
• pipeline::chunked::sequential_checkpoint::run_chunked_sequential_checkpoint (Checkpoint, single thread)
• pipeline::chunked::parallel_checkpoint::run_chunked_parallel_checkpoint (Checkpoint, worker pool)
• pipeline::keyset::run_keyset_parallel (Keyset, worker pool — added 2026-07)
• pipeline::mongo_parallel::run_mongo_parallel (Mongo, worker pool)

All eight share pipeline::commit::record_part for the ordered tail (I2/M1 manifest + I7 file-log + counters + journal), introduced by the commit_part seam work. The cursor + progression writes that follow share pipeline::run_store::RunStore (ADR-0018).

But the timing of the file-log write (the I7 step inside record_part) varies across runners. Four runners call record_part synchronously per part inside the runner’s main loop; one runner — run_chunked_parallel_checkpoint — writes the file-log synchronously inside the worker (before pushing the part to a shared Vec), then has the parent thread call record_part with state = None during the post-scope drain (the parent drain populates manifest_parts + counters + journal; the file-log write is already durable from the worker).

This ADR documents the asymmetry, why it exists, and what invariants each variant satisfies.

The asymmetry

The odd rows are chunked_parallel, parallel_checkpoint, keyset_parallel (feat/parallel-keyset), and mongo_parallel (whose main-thread drain calls record_part(Some(state)) post-scope — the chunked_parallel shape; src/pipeline/mongo_parallel.rs). They are odd for different reasons.

keyset_parallel — per-RANGE worker-sync file_log (added 2026-07)

The 7th runner. Like parallel_checkpoint it writes file_log synchronously in the worker (so a crash-resume can rehydrate) and drains manifest_parts + counters + journal post-scope (record_part(state=None)). It differs on granularity and atomicity: the worker-sync write is per-RANGE, not per-chunk — a range’s parts + its keyset_range.done=1 flip go in ONE transaction (commit_keyset_range), the atomic checkpoint boundary (see dev/parallel_keyset/design_iter2.md). Because an incomplete range writes NO file_log rows, the resume rehydrate pulls only done ranges with no filtering. This sidesteps ADR-0017’s per-chunk StateStore::open smell — the reconnect amortizes over a whole range, not every part — but it is the SAME *_at_ref reconnect pattern (see the smell section; a shared with_ref helper would deepen all four *_at_ref sites).

chunked_parallel — three writes coalesced post-scope

Background: this runner has no chunk_task persistence (it is the non-resumable parallel engine; resumability is parallel_checkpoint’s job). All three writes (file_log + manifest_parts + journal) move to a post-scope drain because:

• The worker has &mut RunSummary access via shared agg_* atomics + a Mutex<Vec<(PartRecord, chunk_index)>>, but summary.journal is not Send + Sync for ordered append.
• The post-scope parent has the &mut summary borrow, can drain the shared Vec, and calls record_part(Some(state), …) once per collected part.

Effect: if the process crashes between scope-join and drain-end, the file is durable at the destination but file_log, manifest_parts, and journal have no entry for that chunk. This is the standard “crash after destination write, before manifest” window (ADR-0001 I2 → I3): the file is recoverable from the destination listing, the manifest is reconstructed from the file_log on resume.

Crash window is the scope-join to drain-end interval — typically microseconds (the drain is a tight CPU loop over an in-memory Vec). The window in practice is dominated by the destination write, not by this drain.

parallel_checkpoint — file_log split from manifest_parts

Background: this runner does have chunk_task persistence and is the resumable parallel engine. Each chunk’s success / failure flips a chunk_task row, and a crash mid-run drops back into a “resume from chunk_task state” code path.

Live test live_chunked_recovery.rs::parallel_chunked_crash_after_chunk_complete_resume_finishes_with_no_duplicates (C3) panics the parent process after the worker has marked its chunk_task row as completed. The resume code must rebuild the manifest from the per-chunk durable file_log rows — there is no in-memory RunSummary to drain, the prior process is gone.

If parallel_checkpoint wrote file_log only in the post-scope drain (like chunked_parallel), a crash at this fault point would leave chunk_task.status = 'completed' but file_log empty for that chunk. On resume, the M8 manifest-reconcile path (pipeline::chunked::resume_m8) would see the chunk_task as done but have no file_log entry to feed back into manifest_parts. The C3 test detects this directly: it asserts post-resume file_log and manifest_parts are coherent.

The migration commit (e9b0796) initially moved file_log writes to the post-scope drain (matching chunked_parallel’s shape). C3 failed immediately. The fix: write file_log synchronously per chunk in the worker (via StateStore::open_at_ref(&state_ref) — see “Known performance smell” below; the earlier StateStore::open(&config_path_w) variant was removed because rivet apply dispatches the runner with an empty config_path, so StateStore::open("") resolved to a stray ./.rivet_state.db and silently stranded the durable-part rows), push the PartRecord to the shared Vec, and have the parent drain call record_part(state=None, …) so the manifest_parts + counters + journal half runs once without double-writing the file_log.

Effect: per-chunk durability of file_log survives any crash. The crash window for manifest_parts is the same scope-join-to-drain-end microsecond interval as chunked_parallel’s, but the file_log half that’s actually needed for resume is already durable.

Decision

The asymmetry is kept because each side optimizes for its runner’s resume semantics:

• chunked_parallel has no resume — the only consumer of file_log during this run is the post-run report. Coalescing all three writes into the drain is correct and simpler.
• parallel_checkpoint has resume — the resume path reads file_log directly without rebuilding from in-memory state. Per-chunk file_log durability is load-bearing for C3-class crashes.

The other four runners are inline per-part because they are single-threaded — there is no worker/parent split forcing the question.

Known performance smell: per-chunk StateStore::open

The parallel_checkpoint worker opens a fresh StateStore connection per chunk just to call record_durable_part. StateStore::open on SQLite takes ~1-5 ms (cold) — for a 1000-chunk run that amortizes to ~1-5 seconds of overhead.

This is not fixed in this release. The clean fix is a record_durable_part_at_ref(&StateRef, …) helper in state::file_log that uses the shared StateRef::Sqlite(path) without re-opening — matching the surviving *_at_ref pattern (claim_next_chunk_task_at_ref; complete_chunk_task / fail_chunk_task are now plain methods invoked after StateStore::open_at_ref). ~50 lines of state-crate work.

Tracked as follow-up; this ADR exists so future readers see the smell was conscious and addressable, not a hidden footgun.

Consequences

Positive

• Each runner’s crash-window semantics are explicit and matched to its resume contract.
• The eight runners share the maximum possible code (commit::record_part body) — the asymmetry is at the caller layer, not in record_part itself.
• C3 at-least-once durability under worker-crash is preserved for the resumable parallel engine without forcing the non-resumable one to pay the per-chunk-open cost.

Negative

• Two runners have non-obvious file_log timing. New contributors who read chunked_parallel first might assume the same pattern applies to parallel_checkpoint, or vice versa. This ADR is the short-circuit.
• The StateStore::open per chunk in parallel_checkpoint is a known performance smell, not fixed in this release.

Amendment 2026-09-24 — the drain is one module

The four parallel runners (plain chunked, parallel checkpoint, parallel keyset, parallel Mongo) no longer write their own post-join drain. Workers publish parts, observations, committed checksums and failures to pipeline::fan_in::FanIn, and FanIn::finish drains them on the parent in one fixed order: governor log, observations, every durable part through commit::record_part, committed checksums, then the bail. The asymmetry this ADR keeps is now the file_log argument of finish — None where the worker already wrote file_log (parallel checkpoint; parallel keyset on a checkpoint run), Some(state) where the drain writes it (plain chunked, parallel Mongo, non-checkpoint keyset). The timing is unchanged; it is stated at one call site per runner instead of implied by a loop.

Measured reason, not a refactor for its own sake: three of the four drains had already diverged — observations fed below the bail in both chunked runners (the ledger’s contract is above it), inline copies of the governor guards in keyset, and a panicking Mongo worker handing back nothing it had written. Work distribution (spawner, pool, per-range) and the reads (ADR-0028) stay per runner.

Amendment 2026-10-03 — the sequential runners commit through the same module

The four sequential runners (single, plain sequential chunked, sequential checkpoint, sequential keyset) no longer spell out the unit commit by hand. Each calls pipeline::fan_in::commit_unit once per unit (the run, a chunk, a page) with the parts commit::write_sink_parts made durable and the unit’s outcome. commit_unit and FanIn::finish share one private order — observations, every durable part through record_part (on the failure path too), then the committed checksums — and only a committed unit goes on to the validate verdict and the runner’s CHECKPOINT, passed in as a closure: the chunk_task flip for sequential checkpoint, the high-water cursor for sequential keyset, nothing for single and plain sequential chunked. So “checkpoint after the parts are durable” is stated once, not four times. The file_log timing of the table above is unchanged: inline, per unit. One visible difference: single now records its parts after the unit’s last write instead of after each write (the order the chunked runners already had), and its fault hook is the shared sink_part_write:N.

References

• src/pipeline/commit.rs — the shared record_part body that runs in all eight runners.
• src/pipeline/run_store.rs — cursor + progression ordering at the next layer (see ADR-0018).
• src/pipeline/chunked/exec.rs — chunked_parallel runner with post-scope drain.
• src/pipeline/chunked/parallel_checkpoint.rs — split-write runner with worker-sync file_log + post-scope drain for the rest.
• tests/live_chunked_recovery.rs::parallel_chunked_crash_after_chunk_complete_resume_finishes_with_no_duplicates (C3) — the test that pins per-chunk file_log durability for the resumable engine.
• ADR-0001 — state invariants I1-I8; this ADR specializes the I2 → I3 crash window per runner.
• ADR-0010 — two parallel engines (in-process chunked vs subprocess fan-out); this ADR is about a different parallelism axis (chunked workers within one process).