fix: prevent pipeline deadlock when held items block reorder buffer progress

[nh13]

Summary

• Fix deadlock in pipeline steps that feed reorder buffers (compress, decompress, decode)
• When a worker holds an out-of-order batch blocked by reorder buffer backpressure, the step function no longer returns early at Priority 1 — it continues to Priority 3 to pop and process new work, including the critical next-in-sequence batch that unblocks the pipeline
• Three instances of the same bug pattern fixed: shared_try_step_compress (base.rs, BAM + FASTQ), try_step_decompress (bam.rs), try_step_decode (bam.rs)

Deadlock chain (before fix)

1. Parallel workers produce out-of-order batches that accumulate in the reorder buffer, pushing heap_bytes above the 50% backpressure threshold
2. All workers hold batches with serial != next_seq — blocked by can_proceed() returning false
3. Priority 1 returns immediately (OutputFull / false), never reaching Priority 3
4. The next_seq batch sits unprocessed in the input queue — no worker can pop it
5. The downstream consumer (FindBoundaries / Group / Write) starves — deadlock

Fix

• At Priority 1, when the held item is blocked, set held_blocked = true instead of returning early
• At Priority 2, skip the soft output-full check when held_blocked (must still try to process new work)
• At Priority 5, when held_blocked is true, bypass the can_proceed backpressure check for the freshly processed batch — push it directly to the output queue to break the deadlock

Test plan

• Added unit test test_compress_held_item_does_not_block_next_seq — reproduces the exact deadlock scenario (RED before fix, GREEN after)
• Added unit test test_compress_held_next_seq_advances_normally — verifies held next-seq items still advance correctly
• Added unit test test_compress_no_held_empty_q5_returns_input_empty — verifies normal behavior when no held items
• All 2350 existing tests pass
• cargo ci-fmt clean
• cargo ci-lint clean

[codecov]

Codecov Report

❌ Patch coverage is 69.88636% with 53 lines in your changes missing coverage. Please review.
✅ Project coverage is 89.26%. Comparing base (91c0231) to head (46af9f0).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
src/lib/unified_pipeline/bam.rs	6.66%	28 Missing ⚠️
src/lib/unified_pipeline/base.rs	82.87%	25 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #251      +/-   ##
==========================================
- Coverage   89.28%   89.26%   -0.02%     
==========================================
  Files         119      119              
  Lines       57770    57914     +144     
==========================================
+ Hits        51582    51699     +117     
- Misses       6188     6215      +27     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[coderabbitai]

📝 Walkthrough

Walkthrough

Decompression, decoding, and compression steps were changed to track a held_blocked condition when a worker holds a batch that cannot proceed due to memory admission/backpressure. When held_blocked is true, priority-1 early returns and certain queue-capacity or admission checks are skipped so the worker can attempt next-in-sequence work; pushing a second result into a physically full queue now sets an explicit error instead of overwriting. Compression tests were added to validate held-item behavior and error handling.

🚥 Pre-merge checks | ✅ 3

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	Title clearly and concisely describes the main fix: preventing pipeline deadlock caused by held items blocking reorder buffer progress.
Description check	✅ Passed	Description is comprehensive and directly related to the changeset, explaining the deadlock scenario, the fix applied, and test coverage.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch nh/fix-pipeline-deadlock

Warning

Review ran into problems

🔥 Problems

Git: Failed to clone repository. Please run the @coderabbitai full review command to re-trigger a full review. If the issue persists, set path_filters to include or exclude specific files.

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 2

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

src/lib/unified_pipeline/base.rs (1)

4726-4736: ⚠️ Potential issue | 🟠 Major

Potential data loss when Q6 push fails with held_blocked=true.

If held_blocked is true (line 4652 already stored the original held item back), and q6_push fails here, line 4733 overwrites held_compressed with the new batch, silently losing the original held item.

Race scenario: Q6 fills between Priority 2 check (skipped when held_blocked) and this push.

Consider detecting and handling this edge case:

Suggested defensive check

     Err((serial, batch)) => {
+        if held_blocked {
+            // Already holding something - can't store two batches
+            // This shouldn't happen if Q6 is adequately sized, but if it does,
+            // fail explicitly rather than lose the original held batch
+            state.set_error(io::Error::other(
+                "compress: Q6 full while held_blocked - cannot hold two batches"
+            ));
+            return StepResult::OutputFull;
+        }
         // Output full - hold the result for next attempt
         let heap_size = batch.estimate_heap_size();
         *worker.held_compressed_mut() = Some((serial, batch, heap_size));
         StepResult::OutputFull
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/lib/unified_pipeline/base.rs` around lines 4726 - 4736, The current
q6_push error path unconditionally sets worker.held_compressed_mut() which can
overwrite an original held item when held_blocked is true; modify the
Err((serial, batch)) branch in the match after state.q6_push to first check
worker.held_compressed() and worker.held_blocked (or held_blocked flag) and if
there is already Some(...) and held_blocked is true, avoid overwriting—either
return StepResult::OutputFull without replacing the held item or merge/queue the
new (serial, batch) into a safe buffer; ensure you still preserve heap_size via
batch.estimate_heap_size() only when storing a new held item, and keep existing
behavior for record_q7_push_progress/StepResult::Success triggered only on Ok.

🧹 Nitpick comments (1)

src/lib/unified_pipeline/base.rs (1)

6180-6213: Test coverage looks good for the happy path.

Verifies the core fix: held out-of-order batch doesn't block compressing the next-seq batch. Correctly asserts serial 0 reaches Q6 while serial 5 remains held.

Missing edge case: test where Q6 becomes full during the compress operation (triggering the potential data loss path mentioned above). Consider adding:

#[test]
fn test_compress_held_blocked_q6_full_no_data_loss() {
    let state = MockCompressState::new(1, 1000); // capacity=1
    state.reorder.add_heap_bytes(600);
    
    // Fill Q6 with a dummy batch
    let dummy = CompressedBlockBatch::new();
    assert!(state.q6.push((99, dummy)).is_ok());
    
    // Put next-seq in Q5
    let batch_0 = SerializedBatch { data: vec![0u8; 10], record_count: 1, secondary_data: None };
    assert!(state.q5.push((0, batch_0)).is_ok());
    
    let mut worker = MockCompressWorker::new();
    worker.held_compressed = Some((5, CompressedBlockBatch::new(), 100));
    
    let result = shared_try_step_compress(&state, &mut worker);
    // Should NOT lose serial 5
    // (Behavior depends on fix applied)
}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/lib/unified_pipeline/base.rs` around lines 6180 - 6213, Add a new unit
test that exercises the edge case where Q6 is full while compressing the
next-seq batch to ensure held out-of-order batches are not lost: create
MockCompressState with capacity 1 (MockCompressState::new(1, 1000)), mark heap
bytes high via reorder.add_heap_bytes(600), push a dummy CompressedBlockBatch
into state.q6, push the next-seq SerializedBatch into state.q5, set
worker.held_compressed = Some((5, CompressedBlockBatch::new(), 100)) on a
MockCompressWorker, call shared_try_step_compress(&state, &mut worker) and then
assert that the held_compressed still contains serial 5 and that the next-seq
batch was either moved to q6 (if space was freed) or appropriately handled per
current behavior (use StepResult and state.q6.pop()/state.q6.peek() to validate
no data loss).

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/lib/unified_pipeline/bam.rs`:
- Around line 1824-1825: When held_blocked is true the code restores the old
held batch into worker.held_decompressed (worker.held_decompressed =
Some((serial, held, heap_size))) but then later on a failed push path can
overwrite that same slot (causing a dropped batch); change the flow so that when
held_blocked is true you do not reuse worker.held_decompressed for the newly
decompressed batch—store the fresh batch into a separate temporary slot or
insert it directly into the reorder/queue path instead of writing into
worker.held_decompressed. Locate the restore and reassign sites (the assignments
to worker.held_decompressed around the held_blocked branch and the push error
handling code that currently writes back at the later path) and implement a
second storage variable or direct reorder insertion for the new batch to avoid
overwriting the restored held batch.
- Around line 2163-2164: The decode path currently overwrites
worker.held_decoded (set via worker.held_decoded = Some((serial, held,
heap_size)) using serial/held/heap_size) which can lose an already-held decoded
batch when a later push fails and the same slot is replaced; instead preserve
both the existing held batch and the new decoded result: change the storage to
hold multiple entries (e.g., Vec or a small pair) or append the new tuple rather
than replace, and update the push-failure path that currently assigns to
worker.held_decoded to merge/append into that collection (mirror the decompress
fix behavior used elsewhere), ensuring both batches are retained and later
re-queued rather than silently overwritten.

---

Outside diff comments:
In `@src/lib/unified_pipeline/base.rs`:
- Around line 4726-4736: The current q6_push error path unconditionally sets
worker.held_compressed_mut() which can overwrite an original held item when
held_blocked is true; modify the Err((serial, batch)) branch in the match after
state.q6_push to first check worker.held_compressed() and worker.held_blocked
(or held_blocked flag) and if there is already Some(...) and held_blocked is
true, avoid overwriting—either return StepResult::OutputFull without replacing
the held item or merge/queue the new (serial, batch) into a safe buffer; ensure
you still preserve heap_size via batch.estimate_heap_size() only when storing a
new held item, and keep existing behavior for
record_q7_push_progress/StepResult::Success triggered only on Ok.

---

Nitpick comments:
In `@src/lib/unified_pipeline/base.rs`:
- Around line 6180-6213: Add a new unit test that exercises the edge case where
Q6 is full while compressing the next-seq batch to ensure held out-of-order
batches are not lost: create MockCompressState with capacity 1
(MockCompressState::new(1, 1000)), mark heap bytes high via
reorder.add_heap_bytes(600), push a dummy CompressedBlockBatch into state.q6,
push the next-seq SerializedBatch into state.q5, set worker.held_compressed =
Some((5, CompressedBlockBatch::new(), 100)) on a MockCompressWorker, call
shared_try_step_compress(&state, &mut worker) and then assert that the
held_compressed still contains serial 5 and that the next-seq batch was either
moved to q6 (if space was freed) or appropriately handled per current behavior
(use StepResult and state.q6.pop()/state.q6.peek() to validate no data loss).

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: b32ccbba-f358-4053-9756-63601cd199a9

📥 Commits

Reviewing files that changed from the base of the PR and between 91c0231 and 5a04b4a.

📒 Files selected for processing (2)

• src/lib/unified_pipeline/bam.rs
• src/lib/unified_pipeline/base.rs

[nh13]

@coderabbitai review

[coderabbitai]

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/lib/unified_pipeline/base.rs`:
- Around line 4717-4726: The current bypass admits any Q5 head when held_blocked
is true and can push out-of-order batches into Q6; modify the branch that
currently does match state.q6_push((serial, batch)) so it first verifies the
batch is the writer's expected next sequence (e.g., compare serial to the
writer's next_seq via an existing accessor like state.next_expected_seq() or
state.writer_next_seq()); if serial == next_seq then proceed with
state.q6_push((serial, batch)), otherwise preserve the non-next_seq batch
instead of force-admitting it (for example, store it back into
worker.held_compressed_mut() or requeue it to Q5/Q5-backlog) and return
StepResult::OutputFull, ensuring you do not bypass
state.write_reorder_can_proceed(serial) for arbitrary Q5 items and add an
assertion/log if you prefer to fail-fast when assumptions are violated.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: e81faf25-e3ca-4944-89bf-4c20bb43f958

📥 Commits

Reviewing files that changed from the base of the PR and between 5a04b4a and 46af9f0.

📒 Files selected for processing (2)

• src/lib/unified_pipeline/bam.rs
• src/lib/unified_pipeline/base.rs

🚧 Files skipped from review as they are similar to previous changes (1)

• src/lib/unified_pipeline/bam.rs