refactor(consensus,correct): migrate hand-rolled rejects writers to first-class secondary_output

[nh13]

Closes #329.

Summary

• Adds secondary_output_header: Option<Header> to run_bam_pipeline_from_reader_with_secondary so the rejects BAM can carry a different header than the primary output. Defaults (None) reuse the resolved primary header — filter's prior behavior. Consensus commands pass Some(input_header) so the rejects BAM advertises the raw-input header rather than the consensus header.
• Migrates correct, simplex, duplex, codec (execute_threads_mode) from a hand-rolled Arc<Mutex<Option<RawBamWriter>>> to the unified pipeline's first-class secondary_output path. Each command's ProcessedBatch gains a rejected_records: Vec<Vec<u8>> field; rejects flow through a per-batch buffer and the pipeline drains them via secondary_serialize_fn in batch-input order.
• Drops header_as_unsorted from the four rejects sites: the input header's claimed sort order remains valid for the rejects subset because rejects are now in input/batch-serial order (a subset of an SO-X stream is still SO-X).
• Hoists the byte-identical secondary_serialize_fn framing into a single pub(crate) fn serialize_raw_bam_records<R: AsRef<[u8]>>(records: &[R], output: &mut Vec<u8>) -> io::Result<u64> in commands/common.rs. Filter's previously-private serialize_raw_records is removed in favor of the shared helper.

Order change (user-visible)

Rejects records previously landed in mutex-acquisition order across worker threads (arbitrary, non-deterministic). After this change they land in batch-serial order = input order, matching the primary BAM. Rejects content is unchanged; only the order changes.

Existing integration tests use content (set/count) assertions, not specific orderings, so they pass unchanged. The one helper that asserted the old hand-rolled SO:unsorted contract (assert_header_unsorted) is replaced by assert_rejects_header_matches_input(rejects, input), which verifies the rejects header inherits the input's sort fields verbatim.

Memory profile note

The hand-rolled path streamed rejects to disk during process_fn under a mutex — per-batch reject memory was effectively zero. The new path collects rejects in batch.rejected_records: Vec<Vec<u8>> until the pipeline's serialize step consumes them. At high reject rates this adds in-flight memory bounded by the pipeline's queue_memory_limit. Each migrated command's MemoryEstimate impl was updated to account for rejected_records.iter().map(Vec::capacity).sum() + rejected_records.capacity() * size_of::<Vec<u8>>(), so the pipeline correctly throttles.

Per-command perf bench: deferred

The issue's acceptance criteria include a per-command perf bench (≤5% wall + RSS regression). The local benchmark host wasn't quiet at the time of this PR (load average 32, BWA + indexer + media all running). The bench will land in a follow-up commit on this branch — either when the local host is quiet (`hyperfine` + `gtime` + `sudo purge` between trials) or on the documented `m7g.4xlarge` AWS instance (`/usr/bin/time -v`). The bench artifact will land at `docs/perf/issue-329-rejects-secondary.md`.

Test plan

• `cargo ci-test` (2070 tests pass, +1 new)
• `cargo ci-fmt && cargo ci-lint` (clean)
• `cargo doc --no-deps` (clean for changed items)
• New unit tests on `resolve_secondary_header` (None-default + Some-override)
• New integration test `test_simplex_command_rejects_inherits_input_read_groups` — exercises the secondary header override end-to-end (input has RG1+RG2; primary collapses to RG "A"; rejects must preserve both)
• Existing integration tests on `--rejects` for all four commands pass
• Per-command perf bench on quiet host (deferred to follow-up commit on this branch)

Out of scope (per issue)

• `commands/dedup.rs` and `commands/downsample.rs` (use `Arc` for unrelated purposes, not secondary BAM output).
• `fgumi runall --correct::rejects` wiring — separate follow-up.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 4000e385-86a3-4360-bd6e-efe67cabee7e

📥 Commits

Reviewing files that changed from the base of the PR and between a95618f and 5cd9d1a.

📒 Files selected for processing (12)

• src/lib/commands/codec.rs
• src/lib/commands/common.rs
• src/lib/commands/correct.rs
• src/lib/commands/duplex.rs
• src/lib/commands/filter.rs
• src/lib/commands/simplex.rs
• src/lib/unified_pipeline/bam.rs
• tests/integration/helpers/assertions.rs
• tests/integration/test_bgzf_eof.rs
• tests/integration/test_correct_command.rs
• tests/integration/test_duplex_command.rs
• tests/integration/test_simplex_command.rs

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

• tests/integration/test_bgzf_eof.rs
• tests/integration/helpers/assertions.rs
• tests/integration/test_correct_command.rs
• tests/integration/test_simplex_command.rs
• src/lib/commands/common.rs
• src/lib/commands/codec.rs
• tests/integration/test_duplex_command.rs
• src/lib/commands/simplex.rs
• src/lib/commands/duplex.rs
• src/lib/commands/correct.rs
• src/lib/unified_pipeline/bam.rs

---

📝 Walkthrough

Walkthrough

Consensus commands (correct, duplex, codec, simplex) now buffer rejected raw BAM records per processed batch (Vec<Vec>) and emit them via the unified pipeline’s secondary output instead of mutex‑protected streaming writers. Added serialize_raw_bam_records to frame records for secondary output. Pipeline runner accepts an optional secondary_output_header (defaults to resolved primary header). Heap-size estimates and tests were updated to account for buffered rejects.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	Title precisely describes the main refactoring: migrating hand-rolled rejects writers to secondary_output across consensus and correct commands.
Description check	✅ Passed	Description comprehensively covers changes: secondary header support, command migrations, batch buffering, order change, memory impact, tests, and deferred perf bench.
Linked Issues check	✅ Passed	PR fulfills #329 core requirements: migrates 4 commands from Arc<Mutex> to unified pipeline secondary_output, buffers rejects per-batch in input order, adds MemoryEstimate updates, shares serializer via serialize_raw_bam_records, updates tests; defers perf bench to follow-up.
Out of Scope Changes check	✅ Passed	All changes directly support the migration: secondary header plumbing, ProcessedBatch fields, serialize utility, header-matching assertions, integration tests for rejects behavior. No unrelated modifications.
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/issue-329-rejects-secondary-output

---

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.}

[codecov]

Codecov Report

❌ Patch coverage is 92.07048% with 36 lines in your changes missing coverage. Please review.
✅ Project coverage is 90.87%. Comparing base (abcb8e6) to head (5cd9d1a).

Files with missing lines	Patch %	Lines
src/lib/commands/common.rs	87.69%	8 Missing ⚠️
src/lib/commands/duplex.rs	93.16%	8 Missing ⚠️
src/lib/commands/codec.rs	92.04%	7 Missing ⚠️
src/lib/unified_pipeline/bam.rs	73.07%	7 Missing ⚠️
src/lib/commands/correct.rs	89.13%	5 Missing ⚠️
src/lib/commands/simplex.rs	99.08%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #332      +/-   ##
==========================================
+ Coverage   90.79%   90.87%   +0.08%     
==========================================
  Files          78       78              
  Lines       50801    50775      -26     
==========================================
+ Hits        46124    46143      +19     
+ Misses       4677     4632      -45     

☔ 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.

[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.

[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

Caution

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

⚠️ Outside diff range comments (2)

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

4232-4331: ⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Keep secondary serialization ordered the same way in single-threaded mode.

The parallel path mutates the batch before secondary_serialize_fn runs, but the single-threaded path still serializes rejects first. That makes rejects depend on --threads whenever mi_assign_fn and secondary output are both enabled.

Fix

@@
-                // Step 7a: Secondary serialize (borrows processed)
-                buffers.secondary.clear();
-                if let Some(ref secondary_fn) = fns.secondary_serialize_fn {
-                    (secondary_fn)(&processed, &mut buffers.secondary)?;
-                }
-
-                // Step 7b: Primary serialize (consumes processed, reuse buffer).
+                // Step 7a: Primary serialize (consumes processed, reuse buffer).
                 // Run the optional MI Assign hook first so any per-template
-                // rewriting it does is reflected in the bytes serialize
-                // emits.
+                // rewriting it does is reflected in both outputs.
                 buffers.serialized.clear();
                 let processed = run_mi_assign_single_threaded(
                     fns.mi_assign_fn.as_deref(),
                     &mut next_serial,
                     processed,
                 )?;
+                buffers.secondary.clear();
+                if let Some(ref secondary_fn) = fns.secondary_serialize_fn {
+                    (secondary_fn)(&processed, &mut buffers.secondary)?;
+                }

@@
-                buffers.secondary.clear();
-                if let Some(ref secondary_fn) = fns.secondary_serialize_fn {
-                    (secondary_fn)(&processed, &mut buffers.secondary)?;
-                }
-
-                buffers.serialized.clear();
+                buffers.serialized.clear();
                 let processed = run_mi_assign_single_threaded(
                     fns.mi_assign_fn.as_deref(),
                     &mut next_serial,
                     processed,
                 )?;
+                buffers.secondary.clear();
+                if let Some(ref secondary_fn) = fns.secondary_serialize_fn {
+                    (secondary_fn)(&processed, &mut buffers.secondary)?;
+                }

🤖 Prompt for AI Agents

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

In `@src/lib/unified_pipeline/bam.rs` around lines 4232 - 4331, In
run_bam_pipeline_from_reader_with_secondary ensure the single-threaded branch
mirrors the parallel ordering: call secondary_serialize_fn(&processed_batch,
&mut buf) using a borrow of the processed result before calling
serialize_fn(processed_batch, &header, &mut buf) so the secondary serialization
sees the same pre-consumption state as in the parallel path; locate the
single-threaded serialization block that currently writes rejects first/second
and swap or reorder the calls so secondary_serialize_fn runs with a &P borrow
immediately before the primary serialize_fn consumes the P produced by
process_fn.

tests/integration/test_duplex_command.rs (1)

195-202: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Update stale test docs about rejects header semantics.

The comment still describes old mutex-era SO:unsorted behavior, which no longer matches the asserted behavior.

Suggested wording update

 /// 1. The rejects BAM is created and has its `@HD` header rewritten to
-///    `SO:unsorted` because reject writes are emitted in mutex-acquisition
-///    order rather than input order.
+///    match the input BAM sort fields (`SO`/`GO`/`SS`) because rejects are
+///    emitted in batch/input order through unified secondary output.

🤖 Prompt for AI Agents

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

In `@tests/integration/test_duplex_command.rs` around lines 195 - 202, The test
docblock in tests/integration/test_duplex_command.rs still claims the rejects
BAM header is rewritten to `SO:unsorted` due to mutex-acquisition ordering;
update that sentence to match the current asserted behavior (remove the
`SO:unsorted` claim and the "mutex-acquisition order" rationale) so the comment
describes what the test actually asserts about the rejects BAM `@HD` header now
(e.g., that it preserves the expected sort-order/header semantics asserted in
the test). Edit the comment block above the duplex pipeline test accordingly and
remove any reference to the old mutex-era semantics.

🧹 Nitpick comments (4)

tests/integration/test_simplex_command.rs (1)

326-346: ⚡ Quick win

Assert at least one rejected record to avoid a header-only false positive.

This test can pass even if no read is actually rejected. Add a rejects-record assertion so it truly exercises the --min-reads rejection path.

Suggested patch

     let rejects_header =
         bam::io::Reader::new(fs::File::open(&rejects_bam).unwrap()).read_header().unwrap();
     let mut rejects_rgs: Vec<BString> = rejects_header.read_groups().keys().cloned().collect();
     rejects_rgs.sort();
     assert_eq!(
         rejects_rgs,
         vec![BString::from("RG1"), BString::from("RG2")],
         "rejects BAM should inherit the input header's read groups verbatim",
     );
+
+    let mut rejects_reader = bam::io::Reader::new(fs::File::open(&rejects_bam).unwrap());
+    let _ = rejects_reader.read_header().unwrap();
+    let rejected_count = rejects_reader
+        .records()
+        .map(|r| r.expect("Failed to read rejects record"))
+        .count();
+    assert!(rejected_count > 0, "Expected at least one rejected record");
 }

Based on learnings: In fgumi consensus-calling tests, treat a paired-end read pair as one template and verify `--min-reads` behavior against template-level rejection.

🤖 Prompt for AI Agents

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

In `@tests/integration/test_simplex_command.rs` around lines 326 - 346, The test
currently only checks headers and can falsely pass if no reads were rejected;
update the test to assert there is at least one rejected template in rejects_bam
by opening rejects_bam with bam::io::Reader, iterating its records to count
unique template names (QNAME) so paired-end reads count as one template, and add
an assert that this unique-template count is >= 1 before the header assertions
(use the existing rejects_bam variable and then proceed to
rejects_header/rejects_rgs checks).

src/lib/commands/duplex.rs (1)

1702-1717: ⚡ Quick win

Add a non-empty rejects case to memory-estimate test.

The test currently validates only consensus buffer capacity, not the new reject-buffer accounting branch.

Proposed test extension

 #[test]
 fn test_duplex_processed_batch_memory_estimate() {
@@
     let batch = DuplexProcessedBatch {
         consensus_output: ConsensusOutput { data, count: 1 },
         rejected_records: Vec::new(),
         groups_count: 1,
         stats: ConsensusCallingStats::default(),
         overlapping_stats: None,
     };
@@
     let estimate = batch.estimate_heap_size();
     assert_eq!(estimate, 1024, "estimate should match consensus_output capacity");
+
+    let mut rejected = Vec::with_capacity(192);
+    rejected.extend_from_slice(&[1u8; 8]);
+    let batch_with_rejects = DuplexProcessedBatch {
+        consensus_output: ConsensusOutput { data: Vec::with_capacity(64), count: 0 },
+        rejected_records: vec![rejected],
+        groups_count: 0,
+        stats: ConsensusCallingStats::default(),
+        overlapping_stats: None,
+    };
+    let estimate_with_rejects = batch_with_rejects.estimate_heap_size();
+    assert!(
+        estimate_with_rejects >= 64 + 192 + std::mem::size_of::<Vec<u8>>(),
+        "estimate should include rejected_records capacity and outer-vec overhead"
+    );
 }

🤖 Prompt for AI Agents

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

In `@src/lib/commands/duplex.rs` around lines 1702 - 1717, Extend the
test_duplex_processed_batch_memory_estimate to include a second sub-case where
DuplexProcessedBatch.rejected_records is non-empty so the estimate_heap_size
path that accounts for reject buffer is exercised; create a rejected_records Vec
with a known capacity/length (e.g., a few byte vectors or records), build a
DuplexProcessedBatch with that rejected_records plus the ConsensusOutput used
already, call estimate_heap_size() and assert the returned value includes both
consensus_output capacity and the reject buffer size (reference
DuplexProcessedBatch, rejected_records, ConsensusOutput and estimate_heap_size
to locate the code).

src/lib/commands/codec.rs (1)

1218-1232: ⚡ Quick win

Extend memory-estimate test to cover buffered rejects.

estimate_heap_size now includes rejected_records, but the test only covers an empty rejects vector.

Proposed test extension

 #[test]
 fn test_codec_processed_batch_memory_estimate() {
@@
     let batch = CodecProcessedBatch {
         consensus_output: ConsensusOutput { data, count: 1 },
         rejected_records: Vec::new(),
         groups_count: 1,
         stats: CodecConsensusStats::default(),
     };
@@
     let estimate = batch.estimate_heap_size();
     assert_eq!(estimate, 1024, "estimate should match consensus_output capacity");
+
+    let mut rejected = Vec::with_capacity(256);
+    rejected.extend_from_slice(&[1u8; 8]);
+    let batch_with_rejects = CodecProcessedBatch {
+        consensus_output: ConsensusOutput { data: Vec::with_capacity(128), count: 0 },
+        rejected_records: vec![rejected],
+        groups_count: 0,
+        stats: CodecConsensusStats::default(),
+    };
+    let estimate_with_rejects = batch_with_rejects.estimate_heap_size();
+    assert!(
+        estimate_with_rejects >= 128 + 256 + std::mem::size_of::<Vec<u8>>(),
+        "estimate should include rejected_records capacity and outer-vec overhead"
+    );
 }

🤖 Prompt for AI Agents

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

In `@src/lib/commands/codec.rs` around lines 1218 - 1232, Update the
test_codec_processed_batch_memory_estimate to include non-empty rejected_records
so estimate_heap_size covers buffered rejects: create one or more RejectedRecord
(or the concrete type used in rejected_records) containing a buffer with a known
capacity/length, add them to the CodecProcessedBatch.rejected_records alongside
the existing ConsensusOutput, call batch.estimate_heap_size() and assert the
expected value equals the sum of consensus_output capacity plus the heap
contribution of the rejected_records buffers; reference the test function name
test_codec_processed_batch_memory_estimate, the struct CodecProcessedBatch,
field rejected_records and the method estimate_heap_size when making the change.

tests/integration/test_bgzf_eof.rs (1)

851-853: ⚡ Quick win

Assert the single-threaded header expectation explicitly.

This test documents header behavior but doesn’t verify it.

Proposed assertion addition

     assert!(status.success(), "correct single-threaded with rejects failed");
     assert_has_bgzf_eof(&output_bam);
     assert_has_bgzf_eof(&rejects_bam);
-    // Single-threaded correct writes rejects in input order, so it keeps the
-    // input header as-is rather than marking it SO:unsorted.
+    assert_rejects_header_matches_input(&rejects_bam, &input_bam);
 }

🤖 Prompt for AI Agents

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

In `@tests/integration/test_bgzf_eof.rs` around lines 851 - 853, Add an explicit
assertion in the single-threaded branch of tests/integration/test_bgzf_eof.rs to
verify the output header was preserved (i.e., not changed to SO:unsorted).
Locate the place where the test reads or stores the resulting header after the
single-threaded correct-writes path and assert that it either equals the
original input header string or does not contain the "SO:unsorted" tag so the
test actually verifies the documented behavior.

🤖 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 4822-4836: The tests only compare comments() which is too narrow;
update both tests for resolve_secondary_header to assert equality of the entire
Header returned (not just comments) — e.g. replace the comments() assertions
with assertions that resolved equals the primary Header in
test_resolve_secondary_header_defaults_to_primary and that resolved equals the
override Header (and differs from primary) in
test_resolve_secondary_header_uses_override_when_provided; use Header's full
equality (or to_string()/serialization comparison if Header does not implement
PartialEq) so `@HD/`@SQ/@PG and all fields are validated.

---

Outside diff comments:
In `@src/lib/unified_pipeline/bam.rs`:
- Around line 4232-4331: In run_bam_pipeline_from_reader_with_secondary ensure
the single-threaded branch mirrors the parallel ordering: call
secondary_serialize_fn(&processed_batch, &mut buf) using a borrow of the
processed result before calling serialize_fn(processed_batch, &header, &mut buf)
so the secondary serialization sees the same pre-consumption state as in the
parallel path; locate the single-threaded serialization block that currently
writes rejects first/second and swap or reorder the calls so
secondary_serialize_fn runs with a &P borrow immediately before the primary
serialize_fn consumes the P produced by process_fn.

In `@tests/integration/test_duplex_command.rs`:
- Around line 195-202: The test docblock in
tests/integration/test_duplex_command.rs still claims the rejects BAM header is
rewritten to `SO:unsorted` due to mutex-acquisition ordering; update that
sentence to match the current asserted behavior (remove the `SO:unsorted` claim
and the "mutex-acquisition order" rationale) so the comment describes what the
test actually asserts about the rejects BAM `@HD` header now (e.g., that it
preserves the expected sort-order/header semantics asserted in the test). Edit
the comment block above the duplex pipeline test accordingly and remove any
reference to the old mutex-era semantics.

---

Nitpick comments:
In `@src/lib/commands/codec.rs`:
- Around line 1218-1232: Update the test_codec_processed_batch_memory_estimate
to include non-empty rejected_records so estimate_heap_size covers buffered
rejects: create one or more RejectedRecord (or the concrete type used in
rejected_records) containing a buffer with a known capacity/length, add them to
the CodecProcessedBatch.rejected_records alongside the existing ConsensusOutput,
call batch.estimate_heap_size() and assert the expected value equals the sum of
consensus_output capacity plus the heap contribution of the rejected_records
buffers; reference the test function name
test_codec_processed_batch_memory_estimate, the struct CodecProcessedBatch,
field rejected_records and the method estimate_heap_size when making the change.

In `@src/lib/commands/duplex.rs`:
- Around line 1702-1717: Extend the test_duplex_processed_batch_memory_estimate
to include a second sub-case where DuplexProcessedBatch.rejected_records is
non-empty so the estimate_heap_size path that accounts for reject buffer is
exercised; create a rejected_records Vec with a known capacity/length (e.g., a
few byte vectors or records), build a DuplexProcessedBatch with that
rejected_records plus the ConsensusOutput used already, call
estimate_heap_size() and assert the returned value includes both
consensus_output capacity and the reject buffer size (reference
DuplexProcessedBatch, rejected_records, ConsensusOutput and estimate_heap_size
to locate the code).

In `@tests/integration/test_bgzf_eof.rs`:
- Around line 851-853: Add an explicit assertion in the single-threaded branch
of tests/integration/test_bgzf_eof.rs to verify the output header was preserved
(i.e., not changed to SO:unsorted). Locate the place where the test reads or
stores the resulting header after the single-threaded correct-writes path and
assert that it either equals the original input header string or does not
contain the "SO:unsorted" tag so the test actually verifies the documented
behavior.

In `@tests/integration/test_simplex_command.rs`:
- Around line 326-346: The test currently only checks headers and can falsely
pass if no reads were rejected; update the test to assert there is at least one
rejected template in rejects_bam by opening rejects_bam with bam::io::Reader,
iterating its records to count unique template names (QNAME) so paired-end reads
count as one template, and add an assert that this unique-template count is >= 1
before the header assertions (use the existing rejects_bam variable and then
proceed to rejects_header/rejects_rgs checks).

🪄 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: 02430c5d-340f-4677-9849-a0c2a15cc5aa

📥 Commits

Reviewing files that changed from the base of the PR and between 3c3355e and 3278f35.

📒 Files selected for processing (12)

• src/lib/commands/codec.rs
• src/lib/commands/common.rs
• src/lib/commands/correct.rs
• src/lib/commands/duplex.rs
• src/lib/commands/filter.rs
• src/lib/commands/simplex.rs
• src/lib/unified_pipeline/bam.rs
• tests/integration/helpers/assertions.rs
• tests/integration/test_bgzf_eof.rs
• tests/integration/test_correct_command.rs
• tests/integration/test_duplex_command.rs
• tests/integration/test_simplex_command.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.

[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.

[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: 4

🧹 Nitpick comments (1)

src/lib/commands/simplex.rs (1)

1631-1676: ⚡ Quick win

Use one rstest for the memory-estimate cases.

These are the same test shape with different inputs. Folding them into a small #[rstest] table will reduce duplication and match the repo test convention.

As per coding guidelines, **/*.rs: Use rstest for parameterized tests and proptest for property-based testing.

🤖 Prompt for AI Agents

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

In `@src/lib/commands/simplex.rs` around lines 1631 - 1676, Combine the two
duplicate tests into a single parameterized rstest: replace the two functions
test_simplex_processed_batch_memory_estimate and
test_simplex_processed_batch_memory_estimate_with_rejects with one #[rstest]
that supplies two cases (one with consensus_output data cap 1024/len 100 and
empty rejected_records, and one with consensus cap 64/rej outer vec with two
vectors of capacities 256 and 128 and appropriate count values). In the new
test, construct SimplexProcessedBatch and call batch.estimate_heap_size() for
each case and assert equality against the expected value computed from the input
parameters (refer to ConsensusOutput, rejected_records, and use
std::mem::size_of::<Vec<u8>>() to compute outer-vec overhead). Ensure the test
function name remains descriptive (e.g.,
test_simplex_processed_batch_memory_estimates) and add the rstest crate
attribute imports if missing.

🤖 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/commands/codec.rs`:
- Around line 641-643: The current code logs MI groups processed but the
returned/used value is the number of consensus reads
(processed.consensus_output.count), so change the logging to report "consensus
reads" (or similar) instead of "MI groups" and ensure any variables or log
messages that currently reference groups_processed are replaced with the actual
consensus count (use processed.consensus_output.count or the local count
variable). Update the same pattern in the other occurrences mentioned (around
lines 656–699) so run_bam_pipeline_* return values and subsequent logs
consistently report consensus reads rather than MI groups.

In `@src/lib/commands/duplex.rs`:
- Around line 749-751: The code is logging consensus-read counts
(processed.consensus_output.count) as groups_processed; change this to count
actual MI groups produced by serialize_fn: use the MI-group count field (e.g.,
processed.mi_groups.len() or processed.group_count) and cast it to u64 for
groups_processed, and if no explicit group-count field exists fall back to
dividing consensus count by 2 (processed.consensus_output.count as u64 / 2) to
reflect that a group yields 0 or 2 consensus reads; update every occurrence that
uses processed.consensus_output.count for group counting in the duplex pipeline
(including the other spots referenced around 764–816) and keep types consistent.

In `@src/lib/commands/simplex.rs`:
- Around line 611-624: The code currently swallows errors from
apply_overlapping_consensus inside the threaded path by converting them into
per-group rejections and continuing; to match the single-threaded fast path you
must propagate the error instead of turning it into rejects. Modify the block
that handles apply_overlapping_consensus(&mut raw_records, oc) in the simplex
processing loop so that on Err(...) it returns or forwards the error upward
(preserving the original error value) rather than calling
batch_overlapping.merge(...), batch_stats.record_* and continue; keep resetting
oc.stats() and any necessary cleanup, but ensure the function/method returns the
error from apply_overlapping_consensus (or maps it into the enclosing error
type) so failures in overlapping_caller (overlapping_caller,
apply_overlapping_consensus) are not silently dropped in the threaded path.

In `@tests/integration/helpers/assertions.rs`:
- Around line 69-74: The test is reading SO/GO/SS from hdr_map.other_fields(),
but other_fields() only contains nonstandard tags so standard `@HD` sort tags are
never found; change the assertion to serialize/obtain the actual `@HD` header line
from the header object (e.g., use the header's serialization/to_string method or
API that returns the raw `@HD` line) and parse the SO, GO, SS values from that
serialized `@HD` line, then populate SortFields accordingly instead of using
other_fields(); update the code around hdr_map and SortFields construction to
extract the tags from the serialized header.

---

Nitpick comments:
In `@src/lib/commands/simplex.rs`:
- Around line 1631-1676: Combine the two duplicate tests into a single
parameterized rstest: replace the two functions
test_simplex_processed_batch_memory_estimate and
test_simplex_processed_batch_memory_estimate_with_rejects with one #[rstest]
that supplies two cases (one with consensus_output data cap 1024/len 100 and
empty rejected_records, and one with consensus cap 64/rej outer vec with two
vectors of capacities 256 and 128 and appropriate count values). In the new
test, construct SimplexProcessedBatch and call batch.estimate_heap_size() for
each case and assert equality against the expected value computed from the input
parameters (refer to ConsensusOutput, rejected_records, and use
std::mem::size_of::<Vec<u8>>() to compute outer-vec overhead). Ensure the test
function name remains descriptive (e.g.,
test_simplex_processed_batch_memory_estimates) and add the rstest crate
attribute imports if missing.

🪄 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: e35fad90-2e87-444c-b8a3-7266d87c65b8

📥 Commits

Reviewing files that changed from the base of the PR and between 3278f35 and 0ab78d1.

📒 Files selected for processing (12)

• src/lib/commands/codec.rs
• src/lib/commands/common.rs
• src/lib/commands/correct.rs
• src/lib/commands/duplex.rs
• src/lib/commands/filter.rs
• src/lib/commands/simplex.rs
• src/lib/unified_pipeline/bam.rs
• tests/integration/helpers/assertions.rs
• tests/integration/test_bgzf_eof.rs
• tests/integration/test_correct_command.rs
• tests/integration/test_duplex_command.rs
• tests/integration/test_simplex_command.rs

✅ Files skipped from review due to trivial changes (2)

• tests/integration/test_correct_command.rs
• tests/integration/test_duplex_command.rs

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

• tests/integration/test_simplex_command.rs
• src/lib/commands/common.rs
• src/lib/commands/filter.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.

[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

🧹 Nitpick comments (3)

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

4828-4842: 💤 Low value

Consider rstest for the two resolve_secondary_header cases.

The None-defaults-to-primary and Some-overrides tests exercise the same function over different inputs — a natural fit for rstest case parameters per the repo coding guidelines.

♻️ Optional refactor

-    #[test]
-    fn test_resolve_secondary_header_defaults_to_primary() {
-        let primary = Header::builder().add_comment("primary").build();
-        let resolved = resolve_secondary_header(None, &primary);
-        assert_eq!(resolved, primary);
-    }
-
-    #[test]
-    fn test_resolve_secondary_header_uses_override_when_provided() {
-        let primary = Header::builder().add_comment("primary").build();
-        let override_header = Header::builder().add_comment("rejects").build();
-        let resolved = resolve_secondary_header(Some(&override_header), &primary);
-        assert_eq!(resolved, override_header);
-        assert_ne!(resolved, primary);
-    }
+    #[rstest]
+    #[case(None, "primary", "primary")]   // None → falls back to primary
+    #[case(Some("rejects"), "primary", "rejects")] // Some → override wins
+    fn test_resolve_secondary_header(
+        #[case] override_comment: Option<&str>,
+        #[case] primary_comment: &str,
+        #[case] expected_comment: &str,
+    ) {
+        let primary = Header::builder().add_comment(primary_comment).build();
+        let override_header =
+            override_comment.map(|c| Header::builder().add_comment(c).build());
+        let resolved =
+            resolve_secondary_header(override_header.as_ref(), &primary);
+        let expected = Header::builder().add_comment(expected_comment).build();
+        assert_eq!(resolved, expected);
+    }

As per coding guidelines: "Use rstest for parameterized tests."

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/lib/unified_pipeline/bam.rs` around lines 4828 - 4842, Combine the two
unit tests into a single parameterized rstest for resolve_secondary_header:
replace test_resolve_secondary_header_defaults_to_primary and
test_resolve_secondary_header_uses_override_when_provided with one #[rstest]
test that takes an Option<&Header> input and an expected Header, constructing
primary via Header::builder().add_comment("primary").build() and override via
Header::builder().add_comment("rejects").build() for the Some case, then call
resolve_secondary_header(input, &primary) and assert_eq! against the expected;
remove the duplicate tests and import rstest::rstest as needed.

tests/integration/test_correct_command.rs (1)

234-257: ⚡ Quick win

Assert reject order, not just membership.

This proves “no drops / no dups”, but not the new contract that rejects are emitted in input order. A regression back to mutex-acquisition order would still pass here.

Possible assertion shape

     let mut reader = bam::io::Reader::new(fs::File::open(&rejects_bam).unwrap());
     let _header = reader.read_header().unwrap();
     let mut seen: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
+    let mut observed_names = Vec::new();
     let mut total: usize = 0;
     for result in reader.records() {
         let record = result.expect("Failed to read reject record");
         let name = record.name().expect("reject record missing read name").to_string();
+        observed_names.push(name.clone());
         *seen.entry(name).or_insert(0) += 1;
         total += 1;
     }
+
+    let expected_order: Vec<String> = ["far_a", "far_b", "far_c"]
+        .into_iter()
+        .flat_map(|base| (0..far_family_size).map(move |i| format!("{base}_{i}")))
+        .collect();
+    assert_eq!(observed_names, expected_order, "rejects should preserve input order");

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/integration/test_correct_command.rs` around lines 234 - 257, The test
currently verifies only membership/counts of rejects (variables seen, total,
expected_rejects) but not their input order; update the test in
test_correct_command.rs to assert ordered emission by reading the rejects BAM
into a Vec<String> (preserving read order from bam::io::Reader), then compare
that vector directly to expected_names (or expected_names[..expected_rejects])
to ensure exact sequence equality; keep or replace the current HashMap/member
checks (seen/total) with this ordered comparison and fail early with a clear
message if the sequences differ.

src/lib/commands/correct.rs (1)

325-332: ⚡ Quick win

Add a unit test for the new heap estimate.

This now drives pipeline throttling and includes a second buffered output path. A small CorrectProcessedBatch::estimate_heap_size() test would keep later tweaks honest.

As per coding guidelines, src/**/*.rs: Place unit tests alongside source code in src/ directory using #[cfg(test)] modules.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/lib/commands/correct.rs` around lines 325 - 332, Add a unit test in a
#[cfg(test)] mod in the same src file that constructs a CorrectProcessedBatch
with controlled kept_raw_records and rejected_records (populate RawRecord
instances and Vec<u8> with known lengths/capacities), calls
CorrectProcessedBatch::estimate_heap_size(), and asserts it equals the expected
sum computed from RawRecord::capacity() for each kept_raw_records,
Vec::capacity() for each rejected record plus the vector capacity overheads
(kept_raw_records.capacity() * size_of::<RawRecord>() and
rejected_records.capacity() * size_of::<Vec<u8>>()). Name the test clearly
(e.g., estimate_heap_size_counts_all_buffers) and keep it small and
deterministic.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/lib/commands/duplex.rs`:
- Around line 679-727: The batch statistics are being merged inside the MI-group
loop which double-counts cumulative stats; remove the per-group merge call
(remove batch_stats.merge(&caller.statistics()) from inside the for MiGroup
loop) and instead, after the loop completes, set/merge batch_stats once from the
caller's cumulative statistics (e.g., batch_stats = caller.statistics() or
batch_stats.merge(&caller.statistics()) once after the loop) so the threaded
path matches the single-threaded behavior; keep per-group rejected_records
collection and overlapping_stats handling unchanged and ensure the final
DuplexProcessedBatch uses the single aggregated batch_stats.

---

Nitpick comments:
In `@src/lib/commands/correct.rs`:
- Around line 325-332: Add a unit test in a #[cfg(test)] mod in the same src
file that constructs a CorrectProcessedBatch with controlled kept_raw_records
and rejected_records (populate RawRecord instances and Vec<u8> with known
lengths/capacities), calls CorrectProcessedBatch::estimate_heap_size(), and
asserts it equals the expected sum computed from RawRecord::capacity() for each
kept_raw_records, Vec::capacity() for each rejected record plus the vector
capacity overheads (kept_raw_records.capacity() * size_of::<RawRecord>() and
rejected_records.capacity() * size_of::<Vec<u8>>()). Name the test clearly
(e.g., estimate_heap_size_counts_all_buffers) and keep it small and
deterministic.

In `@src/lib/unified_pipeline/bam.rs`:
- Around line 4828-4842: Combine the two unit tests into a single parameterized
rstest for resolve_secondary_header: replace
test_resolve_secondary_header_defaults_to_primary and
test_resolve_secondary_header_uses_override_when_provided with one #[rstest]
test that takes an Option<&Header> input and an expected Header, constructing
primary via Header::builder().add_comment("primary").build() and override via
Header::builder().add_comment("rejects").build() for the Some case, then call
resolve_secondary_header(input, &primary) and assert_eq! against the expected;
remove the duplicate tests and import rstest::rstest as needed.

In `@tests/integration/test_correct_command.rs`:
- Around line 234-257: The test currently verifies only membership/counts of
rejects (variables seen, total, expected_rejects) but not their input order;
update the test in test_correct_command.rs to assert ordered emission by reading
the rejects BAM into a Vec<String> (preserving read order from bam::io::Reader),
then compare that vector directly to expected_names (or
expected_names[..expected_rejects]) to ensure exact sequence equality; keep or
replace the current HashMap/member checks (seen/total) with this ordered
comparison and fail early with a clear message if the sequences differ.

🪄 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: 95fe6704-e155-4221-b632-95ebf2693bfb

📥 Commits

Reviewing files that changed from the base of the PR and between 0ab78d1 and cad793f.

📒 Files selected for processing (11)

• src/lib/commands/codec.rs
• src/lib/commands/common.rs
• src/lib/commands/correct.rs
• src/lib/commands/duplex.rs
• src/lib/commands/filter.rs
• src/lib/commands/simplex.rs
• src/lib/unified_pipeline/bam.rs
• tests/integration/helpers/assertions.rs
• tests/integration/test_bgzf_eof.rs
• tests/integration/test_correct_command.rs
• tests/integration/test_simplex_command.rs

✅ Files skipped from review due to trivial changes (1)

• tests/integration/test_simplex_command.rs

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

• tests/integration/helpers/assertions.rs
• src/lib/commands/common.rs
• tests/integration/test_bgzf_eof.rs

[nh13]

@coderabbitai review

[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.

[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.

[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.