feat(pipeline-io): add the fgumi-pipeline-io crate (P4)

.cargo/config.toml

@@ -34,7 +34,7 @@ ci-test-samtools = "nextest run --workspace --features compare,simulate,profile-
 # Run the concurrency stress tests (behind the `stress-tests` feature). Timing-
 # sensitive, so run on a nightly schedule (see .github/workflows/stress.yml) rather
 # than per-PR, where a flake would block unrelated changes.
-ci-test-stress = "nextest run --workspace --features compare,simulate,profile-adjacency,stress-tests --locked"
+ci-test-stress = "nextest run --workspace --features compare,simulate,profile-adjacency,stress-tests,fgumi-pipeline-io/stress-tests --locked"
 # Run tests with test-utils feature enabled (allows binary tests to use library test utilities)
 t = "test --features test-utils"
 # Generate and serve documentation locally (runs xtask then mdbook serve)

.github/workflows/check.yml

@@ -109,7 +109,11 @@ jobs:
         with:
           tool: nextest
       - name: Generate coverage
-        run: cargo llvm-cov nextest --workspace --features compare,simulate,profile-adjacency --no-tests=pass --lcov --output-path lcov.info
+        # Includes `fgumi-pipeline-io/stress-tests`: the soak/matrix/proptest
+        # suites are gated off the fast `test` job, but they cover ~200 lines that
+        # nothing else reaches. Measuring without them under-reports patch
+        # coverage for code that IS tested, just not on the PR-latency path.
+        run: cargo llvm-cov nextest --workspace --features compare,simulate,profile-adjacency,fgumi-pipeline-io/stress-tests --no-tests=pass --lcov --output-path lcov.info
       - name: Upload to Codecov
         uses: codecov/codecov-action@fb8b3582c8e4def4969c97caa2f19720cb33a72f # v6.0.2
         with:

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = [".", "crates/fgumi-raw-bam", "crates/fgumi-dna", "crates/fgumi-bgzf", "crates/fgumi-fmt", "crates/fgumi-metrics", "crates/fgumi-sam", "crates/fgumi-simd-fastq", "crates/fgumi-tag", "crates/fgumi-umi", "crates/fgumi-consensus", "crates/fgumi-bam-io", "crates/fgumi-sort", "crates/fgumi-cli-common", "crates/fgumi-cli-macros", "crates/fgumi-pipeline-core", "crates/xtask"]
+members = [".", "crates/fgumi-raw-bam", "crates/fgumi-dna", "crates/fgumi-bgzf", "crates/fgumi-fmt", "crates/fgumi-metrics", "crates/fgumi-sam", "crates/fgumi-simd-fastq", "crates/fgumi-tag", "crates/fgumi-umi", "crates/fgumi-consensus", "crates/fgumi-bam-io", "crates/fgumi-sort", "crates/fgumi-cli-common", "crates/fgumi-cli-macros", "crates/fgumi-pipeline-core", "crates/fgumi-pipeline-io", "crates/xtask"]
 resolver = "2"
 
 [workspace.package]
@@ -29,6 +29,7 @@ fgumi-consensus = { version = "0.5.0", path = "crates/fgumi-consensus", default-
 fgumi-dna = { version = "0.5.0", path = "crates/fgumi-dna" }
 fgumi-metrics = { version = "0.5.0", path = "crates/fgumi-metrics" }
 fgumi-pipeline-core = { version = "0.5.0", path = "crates/fgumi-pipeline-core" }
+fgumi-pipeline-io = { version = "0.5.0", path = "crates/fgumi-pipeline-io" }
 fgumi-raw-bam = { version = "0.5.0", path = "crates/fgumi-raw-bam" }
 fgumi-sam = { version = "0.5.0", path = "crates/fgumi-sam" }
 fgumi-simd-fastq = { version = "0.5.0", path = "crates/fgumi-simd-fastq" }

crates/fgumi-bam-io/src/prefetch_reader.rs

@@ -205,7 +205,6 @@ impl PrefetchReader {
 
     /// Total bytes served to callers of [`Read::read`] so far.
     #[must_use]
-    #[allow(dead_code)]
     pub fn bytes_consumed(&self) -> u64 {
         self.bytes_consumed
     }
@@ -214,7 +213,6 @@ impl PrefetchReader {
     /// to deliver the next chunk. Useful as a prototype-phase signal for
     /// whether [`DEFAULT_PREFETCH_DEPTH`] is large enough.
     #[must_use]
-    #[allow(dead_code)]
     pub fn consumer_stalls(&self) -> u64 {
         self.consumer_stalls
     }

crates/fgumi-pipeline-io/Cargo.toml

@@ -0,0 +1,43 @@
+[package]
+name = "fgumi-pipeline-io"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+description = "BAM pipeline I/O types and steps for fgumi"
+repository.workspace = true
+license.workspace = true
+
+[dependencies]
+fgumi-bam-io = { workspace = true }
+fgumi-bgzf = { workspace = true }
+fgumi-pipeline-core = { workspace = true }
+fgumi-raw-bam = { workspace = true, features = ["noodles"] }
+fgumi-sort = { workspace = true }
+anyhow = "1.0.102"
+log = "0"
+noodles = { version = "0.111.0", features = ["bam", "sam", "bgzf"] }
+parking_lot = "0.12"
+rayon = "1.10"
+tempfile = "3.4"
+
+[features]
+test-utils = []
+# Multi-minute soak / matrix / proptest suites. Off by default so PR CI stays
+# fast; run on the nightly schedule via `cargo ci-test-stress`, matching the
+# root crate's `stress-tests` convention.
+stress-tests = []
+
+[dev-dependencies]
+fgumi-bam-io = { workspace = true }
+fgumi-raw-bam = { workspace = true, features = ["noodles", "test-utils"] }
+fgumi-sort = { workspace = true, features = ["test-utils"] }
+tempfile = "3.4"
+noodles = { version = "0.111.0", features = ["bam", "sam"] }
+fgumi-bgzf = { workspace = true }
+rstest = "0.26"
+proptest = "1.10"
+criterion = { version = "0.8", features = ["html_reports"] }
+
+[[bench]]
+name = "serial_ingest"
+harness = false

crates/fgumi-pipeline-io/benches/serial_ingest.rs

@@ -0,0 +1,119 @@
+#![deny(unsafe_code)]
+
+//! Go/no-go microbenchmark for the parallel-inflate redesign (increment 3b.0).
+//!
+//! The redesign's whole wall-clock premise is that the *serial* Phase-1 ingest
+//! step gets cheaper per record by NOT copying each record's bytes: today's
+//! `SortBlockBuffer` does `BoundaryState::scan` → `CoordinateChunkSorter::push`,
+//! and `push` memcpys every record body into the sorter's own arena (~the
+//! 121 ns/rec the profile flagged). The redesign instead builds a lightweight
+//! `(key, offset, len)` ref over bytes already in a shared arena — no copy.
+//!
+//! This bench measures EXACTLY that per-record delta with today's public API —
+//! no new pipeline steps, no arena, no visibility changes:
+//!   * `scan_copy_push`  — scan + `CoordinateChunkSorter::push` (copies body)
+//!   * `scan_refbuild`   — scan + `extract_coordinate_key_inline` + `RecordRef::new` (no copy)
+//!
+//! It is a NECESSARY-condition gate, not the full wall proof: it shows whether
+//! removing the copy cuts serial per-record cost. If `scan_refbuild` is not
+//! materially faster than `scan_copy_push`, the redesign cannot beat legacy on
+//! wall and we stop before building the pipeline. If it is faster, the full wall
+//! win still has to be confirmed by the end-to-end gp3 measurement (the overlap
+//! the freeze-per-run model trades away is not captured here).
+//!
+//!     cargo bench -p fgumi-pipeline-io --bench serial_ingest
+
+use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main};
+use fgumi_pipeline_io::boundaries::BoundaryState;
+use fgumi_raw_bam::testutil::make_bam_bytes;
+use fgumi_sort::{RawExternalSorter, RecordRef, SortOrder, extract_coordinate_key_inline};
+use noodles::sam::Header;
+
+/// Build a decompressed-BAM-style buffer of `n` framed records: each record is
+/// `[block_size: u32 LE][body]`, exactly what `BoundaryState::scan` walks. Bodies
+/// are ~100 bp aligned reads at scattered positions on tid 0 (realistic per-record
+/// size; the sort would do real reordering work).
+fn synth_framed(n: usize) -> Vec<u8> {
+    let mut buf = Vec::new();
+    for i in 0..n {
+        let pos = (i as u64).wrapping_mul(2_654_435_761) % 5_000_000;
+        let name = format!("r{i:08}");
+        let body = make_bam_bytes(0, pos as i32, 0, name.as_bytes(), &[], 100, -1, -1, &[]);
+        let block_size = u32::try_from(body.len()).expect("record fits u32");
+        buf.extend_from_slice(&block_size.to_le_bytes());
+        buf.extend_from_slice(&body);
+    }
+    buf
+}
+
+fn bench_serial_ingest(c: &mut Criterion) {
+    const N: usize = 200_000;
+    let framed = synth_framed(N);
+    let n_ref = 1u32;
+
+    // Sanity: both paths see the same record count (so the throughput is over the
+    // same work). Computed once, untimed.
+    {
+        let mut bs = BoundaryState::new_no_header();
+        let (offsets, _range) = bs.scan(&framed).expect("scan");
+        assert_eq!(offsets.len().saturating_sub(1), N, "scan must yield N records");
+    }
+
+    let mut group = c.benchmark_group("serial_ingest");
+    group.throughput(Throughput::Elements(N as u64));
+    group.sample_size(20);
+
+    // OLD path: scan + CoordinateChunkSorter::push — `push` copies each body into
+    // the sorter's arena (plus key-extract + internal ref-push).
+    group.bench_function("scan_copy_push", |b| {
+        b.iter_batched(
+            || {
+                RawExternalSorter::new(SortOrder::Coordinate)
+                    .threads(1)
+                    .into_coordinate_chunk_sorter(&Header::default())
+                    .expect("build coordinate chunk sorter")
+            },
+            |mut sorter| {
+                let mut bs = BoundaryState::new_no_header();
+                let (offsets, range) = bs.scan(&framed).expect("scan");
+                let recs = bs.records_bytes(range);
+                for w in offsets.windows(2) {
+                    let body = &recs[w[0] + 4..w[1]];
+                    sorter.push(body).expect("push");
+                }
+                std::hint::black_box(&mut sorter);
+            },
+            BatchSize::PerIteration,
+        );
+    });
+
+    // NEW path: scan + key-extract + ref-build — NO body copy. This is the
+    // per-record work the redesign's serial FindBoundariesAndSort step does.
+    group.bench_function("scan_refbuild", |b| {
+        b.iter_batched(
+            || Vec::<RecordRef>::with_capacity(N),
+            |mut refs| {
+                refs.clear();
+                let mut bs = BoundaryState::new_no_header();
+                let (offsets, range) = bs.scan(&framed).expect("scan");
+                let recs = bs.records_bytes(range);
+                for w in offsets.windows(2) {
+                    let body = &recs[w[0] + 4..w[1]];
+                    let key = extract_coordinate_key_inline(body, n_ref);
+                    // offset/len point at the body (prefix skipped) — the redesign's
+                    // arena ref representation; here the offset is illustrative.
+                    let offset = u64::try_from(w[0] + 4).expect("offset fits u64");
+                    let len = u32::try_from(w[1] - w[0] - 4).expect("len fits u32");
+                    refs.push(RecordRef::new(key, offset, len));
+                }
+                std::hint::black_box(&mut refs);
+            },
+            BatchSize::PerIteration,
+        );
+    });
+
+    group.finish();
+}
+
+criterion_group!(benches, bench_serial_ingest);
+criterion_main!(benches);