feat(pipeline): add the typed-step pipeline tree with its foundational steps (R1a) - #735
Merged
Conversation
|
Note Reviews pausedUse the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (2)
WalkthroughThe root crate now exposes a typed pipeline framework with BGZF, BAM, and SAM processing steps, ordered batch types, tuning configuration, sink and sort re-exports, backpressure handling, validation, and capacity-based memory accounting. ChangesTyped pipeline integration
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant BgzfInput
participant BgzfDecompress
participant FindBamBoundaries
participant ParseBamRecords
participant DecodeRecords
BgzfInput->>BgzfDecompress: BgzfBlock input
BgzfDecompress->>FindBamBoundaries: DecompressedBlock output
FindBamBoundaries->>ParseBamRecords: Record-aligned block
ParseBamRecords->>DecodeRecords: RecordBatch
DecodeRecords->>DecodeRecords: Validate and compute group keys
Possibly related PRs
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
Comment |
Member
Author
|
@coderabbitai pause |
✅ Action performedReviews paused. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main-runall #735 +/- ##
==============================================
Coverage ? 93.96%
==============================================
Files ? 241
Lines ? 128338
Branches ? 0
==============================================
Hits ? 120598
Misses ? 7740
Partials ? 0 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
nh13
force-pushed
the
nh/runall-12-bam-io-grouping
branch
from
August 10, 2026 01:47
f3698f5 to
804bb73
Compare
nh13
force-pushed
the
nh/runall-11-pipeline-steps
branch
from
August 10, 2026 01:49
f9755bb to
964b42c
Compare
nh13
force-pushed
the
nh/runall-12-bam-io-grouping
branch
from
August 10, 2026 02:47
804bb73 to
fbc534f
Compare
nh13
force-pushed
the
nh/runall-11-pipeline-steps
branch
from
August 10, 2026 02:48
964b42c to
0c3ef9f
Compare
The typed-step pipeline budgets its queues on real allocated bytes, so `BamTemplateBatch` needs a `Template` heap footprint it can sum. The capacity-based computation already existed, but only inside the `MemoryEstimate` impl, where the pipeline cannot reach it. Lift it to an inherent `heap_size` and make `estimate_heap_size` delegate, so the queue budget and `MemoryEstimate` cannot drift apart. Adds the agreement test, which also re-derives the expected floor from the components rather than from `estimate_heap_size` (that would be tautological once the trait delegates).
nh13
force-pushed
the
nh/runall-11-pipeline-steps
branch
from
August 10, 2026 03:38
0c3ef9f to
4355d9a
Compare
Member
Author
|
@coderabbitai review |
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 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/pipeline/steps/boundaries/state.rs`:
- Around line 97-125: Bound all framing lengths with checked arithmetic and the
configured carry limit in src/lib/pipeline/steps/boundaries/state.rs:97-125,
including l_text and l_name before offset updates or slicing; reject invalid
values. In src/lib/pipeline/steps/boundaries/state.rs:170-225, validate each
record’s computed length before retaining incomplete input and return
InvalidData when it exceeds the limit. Apply the same checked calculation and
limit during EOF validation in
src/lib/pipeline/steps/boundaries/state.rs:274-290 so retained memory remains
bounded.
In `@src/lib/pipeline/steps/parse/bam.rs`:
- Around line 139-141: Update parse_record_ranges to initialize ranges with
capacity estimated from bytes.len(), accounting for the 32-byte BAM fixed header
and 4-byte record-length prefix; preserve the existing push and parsing behavior
while avoiding Vec growth during normal batches.
In `@src/lib/pipeline/steps/parse/sam.rs`:
- Around line 99-104: Expose cache_umi_position from decode.rs as pub(crate),
then call it in the SAM parsing flow immediately after
DecodedRecord::from_raw_bytes in the shown logic, using key_config.umi_tag and
the constructed record so SAM records cache their UMI position before being
pushed. Keep the existing name-hash/group-key behavior unchanged.
- Around line 62-71: Validate each range in the loop before slicing bytes: use
bytes.get(...) to confirm start..end is in bounds and non-reversed, returning an
io::Error with ErrorKind::InvalidData when validation fails. Apply the existing
empty-line check only after successful validation, and preserve the current
continue behavior for spans of one byte or less.
In `@src/lib/pipeline/steps/types.rs`:
- Around line 68-70: Update src/lib/pipeline/steps/types.rs:68-70 in the batch
constructor to include records.capacity() multiplied by
size_of::<DecodedRecord>() in total_bytes; update both template constructors at
src/lib/pipeline/steps/types.rs:136-148 to include templates.capacity()
multiplied by size_of::<Template>(), while keeping from_parts limited to summed
template heap usage; and update src/lib/pipeline/steps/types.rs:219-223 to
account for bytes.capacity() and line_offsets.capacity() instead of logical
lengths. Add tests covering excess reserved container capacity for each
accounting path.
🪄 Autofix
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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 2736e68a-2c0c-470e-9f3f-3d906ce19510
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock,!**/*.lock
📒 Files selected for processing (21)
Cargo.tomlsrc/lib/mod.rssrc/lib/pipeline/mod.rssrc/lib/pipeline/steps/bgzf/compress.rssrc/lib/pipeline/steps/bgzf/decompress.rssrc/lib/pipeline/steps/bgzf/mod.rssrc/lib/pipeline/steps/boundaries/bam.rssrc/lib/pipeline/steps/boundaries/mod.rssrc/lib/pipeline/steps/boundaries/state.rssrc/lib/pipeline/steps/mod.rssrc/lib/pipeline/steps/parse/bam.rssrc/lib/pipeline/steps/parse/decode.rssrc/lib/pipeline/steps/parse/mod.rssrc/lib/pipeline/steps/parse/sam.rssrc/lib/pipeline/steps/sink/mod.rssrc/lib/pipeline/steps/sink/write_bgzf.rssrc/lib/pipeline/steps/sink/write_raw.rssrc/lib/pipeline/steps/sort/mod.rssrc/lib/pipeline/steps/tuning.rssrc/lib/pipeline/steps/types.rssrc/lib/template.rs
nh13
force-pushed
the
nh/runall-11-pipeline-steps
branch
from
August 11, 2026 00:57
4355d9a to
1abf0ae
Compare
nh13
force-pushed
the
nh/runall-11-pipeline-steps
branch
from
August 11, 2026 01:22
1abf0ae to
3c5ff4d
Compare
Member
Author
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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/pipeline/steps/parse/bam.rs`:
- Around line 242-264: Update the error-message prefixes in parse_records to use
the function’s own name, parse_records, instead of ParseBamRecords. Apply this
to both the “record extends past buffer end” and “trailing partial record”
errors, leaving the existing error kinds and details unchanged.
In `@src/lib/pipeline/steps/parse/decode.rs`:
- Around line 86-99: Update the read-name bound calculation in the affected
decode function to derive its fixed-record offset from
fgumi_raw_bam::MIN_BAM_RECORD_LEN instead of the hardcoded 32, keeping it
consistent with the existing length guards and preserving the current
bounds-check behavior.
- Around line 543-562: Add a test alongside
cache_umi_position_leaves_cache_unset_when_tag_missing that uses
SamBuilder::add_int_tag to assign RX a non-Z integer value, then calls
cache_umi_position and asserts cached_umi_position().0 remains
UMI_OFFSET_UNCACHED and cached_umi() is None.
🪄 Autofix
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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: bc6126d2-7aee-4416-9283-b41b7597f146
📒 Files selected for processing (8)
src/lib/pipeline/steps/boundaries/state.rssrc/lib/pipeline/steps/chain_tests.rssrc/lib/pipeline/steps/mod.rssrc/lib/pipeline/steps/parse/bam.rssrc/lib/pipeline/steps/parse/decode.rssrc/lib/pipeline/steps/parse/sam.rssrc/lib/pipeline/steps/tuning.rssrc/lib/pipeline/steps/types.rs
nh13
force-pushed
the
nh/runall-11-pipeline-steps
branch
from
August 12, 2026 06:19
3c5ff4d to
444bb61
Compare
Member
Author
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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/pipeline/steps/parse/bam.rs`:
- Around line 161-170: Update both framing parsers around the visible loop and
parse_records frame-end calculations to use checked_add for body_start plus
block_size (and the equivalent cursor plus header plus block_size computation).
Treat arithmetic overflow as an invalid/truncated frame and reject it before
comparing bounds or slicing, ensuring every length-derived range is validated
against the remaining buffer.
🪄 Autofix
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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: ddd1bee5-6ec5-4261-a7e9-6b33b7927502
📒 Files selected for processing (3)
src/lib/pipeline/steps/parse/bam.rssrc/lib/pipeline/steps/parse/decode.rssrc/lib/pipeline/steps/parse/sam.rs
nh13
force-pushed
the
nh/runall-11-pipeline-steps
branch
from
August 12, 2026 07:19
444bb61 to
17cc974
Compare
…l steps
First slice of the umbrella-side port. Additive: nothing in
`src/lib/commands` routes through this tree yet, so every command still runs
on `unified_pipeline`.
`pipeline/mod.rs` re-exports `fgumi_pipeline_core as core`, which is what lets
the step sources compile unmodified — their `crate::pipeline::core::…` paths
resolve through the shim rather than needing a rewrite.
Ports `steps/{types,tuning}` plus the `sink/`, `sort/`, `bgzf/`,
`boundaries/` and `parse/` subtrees. `source/`, `group/`, `correct/` and the
closure-driven mid-steps follow; `align_and_merge` waits on `crate::aligner`.
Two deliberate deltas from the source branch:
- `boundaries/bam.rs` uses a let-chain instead of nested `if`s. The source
predates the toolchain that unlocked them, and `ci-lint` rejects the nested
form as `collapsible_if`; per CLAUDE.md we adopt the idiom rather than
suppress the lint.
- `tuning.rs` demotes the `PipelineConfig::auto_tuned` intra-doc link to a
code span. The referent lives in `unified_pipeline`, which this port exists
to retire, so linking it would only break again on deletion.
nh13
force-pushed
the
nh/runall-11-pipeline-steps
branch
from
August 12, 2026 07:32
17cc974 to
8a51a99
Compare
Member
Author
|
@coderabbitai review |
✅ Action performedReview finished.
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #734 (R0). Review that first; this PR's diff is against it.
Additive: nothing in
src/lib/commandsroutes through the new tree, so every command still runs onunified_pipeline. No behaviour changes.Why this is cheap
feat-runall'ssrc/lib/pipeline/mod.rsdoespub use fgumi_pipeline_core as core;. Everycrate::pipeline::core::…path insidesteps/resolves through that shim, so the step sources compile unmodified — this is a near-verbatim port plus a module root, not a rewrite.What
src/lib/pipeline/mod.rs— thecorere-export shim +steps. (chainsfollows in a later PR; until it lands nothing routes here.)src/lib/pipeline/steps/—mod,types,tuning, and thesink/,sort/,bgzf/,boundaries/,parse/subtrees.src/lib/template.rs—Template::heap_sizelifted out of theMemoryEstimateimpl so the pipeline can budget on it, withestimate_heap_sizedelegating. Separate commit.Cargo.toml— depends onfgumi-pipeline-coreandfgumi-pipeline-io. This edge did not previously exist: both crates had landed with no consumer.Still to come:
source/,group/,correct/, the closure-driven mid-steps, andalign_and_merge(which needscrate::aligner, not yet ported).Two deliberate deltas from
feat-runallBoth are noted in the commit message:
boundaries/bam.rsuses a let-chain instead of nestedifs. The source predates the toolchain that unlocked them andci-lintrejects the nested form ascollapsible_if. PerCLAUDE.mdwe adopt the idiom rather than suppress the lint.tuning.rsdemotes an intra-doc link to a code span.[PipelineConfig::auto_tuned]resolves tounified_pipeline::base— the module this whole port exists to retire. Linking it would break again on deletion, so the reference is now prose.Verification
Full local gate:
cargo ci-fmt,cargo ci-lint,cargo ci-tag-literals— cleanRUSTDOCFLAGS="-D warnings" cargo ci-doc— clean./scripts/publish-crates.sh --check— cleancargo ci-test— 8234 passed, 30 skipped (up 42 from feat(bam-io): add the shared grouping and library-lookup domain types (R0) #734's 8192; the ported steps bring their own tests)Risk: command output changes: none;
unsafe: none, andCLAUDE.mdallowlist unchanged; memory, queue, and backpressure policy: changed throughTemplate::heap_sizeand typed-step byte limits. Fix: route the typed pipeline only after command-output parity is validated.src/lib/pipeline/.Template::heap_size;MemoryEstimate::estimate_heap_sizenow delegates to it.unified_pipeline.