feat(pipeline-core): add fgumi-pipeline-core, the typed-step pipeline framework

.coderabbit.yaml

@@ -49,11 +49,14 @@ reviews:
 
   # MAINTENANCE: these instructions name specific modules, invariants, and code
   # constructs. Re-read and update them whenever those modules are refactored or
-  # renamed — stale instructions silently misdirect the reviewer. (Both the
-  # `unified_pipeline` and `pipeline` paths are listed: issue #330 renames the
-  # former to the latter, so the same instructions apply to whichever is present.)
+  # renamed — stale instructions silently misdirect the reviewer. (All three
+  # pipeline paths are listed: issue #330 renames `src/lib/unified_pipeline` to
+  # `src/lib/pipeline` and extracts its engine into the `fgumi-pipeline-core`
+  # crate, so the same instructions apply to whichever is present. A glob that
+  # names only the `src/lib/` homes silently stops reviewing the engine the day
+  # it moves into the crate.)
   path_instructions:
-    - path: "src/lib/{unified_pipeline,pipeline}/**/*.rs"
+    - path: "{src/lib/unified_pipeline,src/lib/pipeline,crates/fgumi-pipeline-core/src}/**/*.rs"
       instructions: >-
         This is the hand-rolled concurrent step pipeline; its bugs are deadlocks,
         lost output, and unbounded memory, not style. Treat any change to the

.github/workflows/miri.yml

@@ -5,12 +5,19 @@ name: Miri (undefined-behavior check)
 # undefined behavior (out-of-bounds, invalid pointer use, aliasing violations),
 # turning those invariants into a machine-checked gate.
 #
-# Scope: `fgumi-raw-bam`, which contains the raw-pointer queryname comparator
-# (`natural_compare` / `natural_compare_nul`, via `get_unchecked` and `*const u8`
-# walks) and has no FFI. `fgumi-sort` is intentionally not covered yet: its
-# `memory_probe` module calls the mimalloc / mach2 FFI, which Miri cannot execute,
-# so covering its radix-sort unsafe needs `#[cfg(not(miri))]` guards first
-# (tracked as a follow-up).
+# Scope: the two crates whose approved `unsafe` is pure Rust (no FFI), each
+# narrowed to the module that carries it —
+#   - `fgumi-raw-bam::sort` — the raw-pointer queryname comparator
+#     (`natural_compare` / `natural_compare_nul`, via `get_unchecked` and
+#     `*const u8` walks).
+#   - `fgumi-pipeline-core::erased` — the typed-handle dispatch cache, four
+#     `mem::transmute`s that widen `&'a Handle` to `&'static` for storage and
+#     narrow it back on read. Aliasing/lifetime UB is exactly what Stacked
+#     Borrows checks, so this is the one place Miri adds signal the type system
+#     cannot.
+# `fgumi-sort` is intentionally not covered yet: its `memory_probe` module calls
+# the mimalloc / mach2 FFI, which Miri cannot execute, so covering its radix-sort
+# unsafe needs `#[cfg(not(miri))]` guards first (tracked as a follow-up).
 #
 # Runs on a schedule (and on demand) rather than per-PR: Miri needs the nightly
 # toolchain (which can regress independently of this repo) and is much slower than
@@ -61,4 +68,65 @@ jobs:
         # cannot interpret, and the BAM encode/decode roundtrip tests are slow and
         # exercise no `unsafe` of ours. The `sort` tests (comparator + coordinate/
         # queryname raw-key tests) run clean under Miri in a few seconds.
-        run: cargo +nightly miri test -p fgumi-raw-bam sort
+        #
+        # `--list` first and fail on an empty selection: `cargo test <filter>`
+        # exits 0 when the filter matches nothing, so renaming or moving the
+        # module would silently retire this gate while CI stayed green — the same
+        # zero-match hazard `compile_fail.rs`'s `EXPECTED_FIXTURES` guards.
+        run: |
+          set -euo pipefail
+          # The zero-match guard below proves the filter selects tests; it cannot
+          # prove the filter still COVERS the crate's `unsafe`. Pin that too: if an
+          # `#[allow(unsafe_code)]` site moves out of `sort.rs`, the filter keeps
+          # matching and Miri silently stops checking the moved site.
+          stray=$(grep -rl 'allow(unsafe_code)' crates/fgumi-raw-bam/src \
+            | grep -v '^crates/fgumi-raw-bam/src/sort.rs$' || true)
+          if [ -n "${stray}" ]; then
+            echo "::error::unsafe_code outside the Miri-scoped 'sort' module: ${stray}" >&2
+            exit 1
+          fi
+          cargo +nightly miri test -p fgumi-raw-bam sort -- --list > "${RUNNER_TEMP}/miri-raw-bam-sort.list"
+          n=$(grep -c ': test$' "${RUNNER_TEMP}/miri-raw-bam-sort.list" || true)
+          echo "miri: the 'sort' filter matched ${n} test(s) in fgumi-raw-bam"
+          if [ "${n}" -eq 0 ]; then
+            echo "::error::the 'sort' filter matched no tests in fgumi-raw-bam — the Miri scope is stale" >&2
+            exit 1
+          fi
+          cargo +nightly miri test -p fgumi-raw-bam sort
+      - name: Miri — fgumi-pipeline-core typed-handle dispatch cache
+        # Scope to the `erased` module for the same reason: all four
+        # `#[allow(unsafe_code)]` sites in the crate live there
+        # (`TypedStep::resolve_input`/`resolve_outputs` and the `TypedStep2`
+        # pair). These 25 tests run clean under Miri in ~6s. The rest of the
+        # crate is deliberately excluded: the `builder` / `runtime` tests spawn
+        # worker threads and run a full pipeline, which takes Miri well over ten
+        # minutes, and three of them (`detached_two_sided_no_deadlock`,
+        # `driver_round_robins_all_live_before_parking`,
+        # `sticky_holding_source_yields_to_its_draining_consumer`) guard against
+        # a wedge with a WALL-CLOCK watchdog that `process::abort()`s — under
+        # Miri's slowdown that fires on a healthy run. Widening this scope means
+        # giving those watchdogs a `#[cfg(miri)]` budget first.
+        #
+        # Zero-match guard, as on the `fgumi-raw-bam` step above: this filter is
+        # the only thing pointing Miri at the crate's four `unsafe` sites, and an
+        # empty selection would pass silently.
+        run: |
+          set -euo pipefail
+          # Location guard, as on the `fgumi-raw-bam` step above: the step comment
+          # claims all four `#[allow(unsafe_code)]` sites live in `erased`, and
+          # nothing else enforces it. A site that moves elsewhere would leave the
+          # filter matching and the moved site unchecked.
+          stray=$(grep -rl 'allow(unsafe_code)' crates/fgumi-pipeline-core/src \
+            | grep -v '^crates/fgumi-pipeline-core/src/erased.rs$' || true)
+          if [ -n "${stray}" ]; then
+            echo "::error::unsafe_code outside the Miri-scoped 'erased' module: ${stray}" >&2
+            exit 1
+          fi
+          cargo +nightly miri test -p fgumi-pipeline-core erased -- --list > "${RUNNER_TEMP}/miri-pipeline-core-erased.list"
+          n=$(grep -c ': test$' "${RUNNER_TEMP}/miri-pipeline-core-erased.list" || true)
+          echo "miri: the 'erased' filter matched ${n} test(s) in fgumi-pipeline-core"
+          if [ "${n}" -eq 0 ]; then
+            echo "::error::the 'erased' filter matched no tests in fgumi-pipeline-core — the Miri scope is stale" >&2
+            exit 1
+          fi
+          cargo +nightly miri test -p fgumi-pipeline-core erased

.github/workflows/publish.yml

@@ -87,6 +87,7 @@ jobs:
             fgumi-fmt
             fgumi-cli-macros
             fgumi-cli-common
+            fgumi-pipeline-core
             fgumi-raw-bam
             fgumi-bam-io
             fgumi-umi

CLAUDE.md

@@ -210,6 +210,37 @@ regresses `samtools sort -n`–style throughput.
     SAFETY: the buffers are `to_vec()` + push, so the pointer is valid and
     null-terminated for the call's lifetime.
 
+### Approved typed-handle cache (fgumi-pipeline-core)
+
+The pipeline runtime hands each step its input/output handles as
+`&dyn Any`, so the `ErasedStep` adapter must `downcast_ref` them back to
+their concrete types. That downcast sits on the per-item dispatch path: a
+4-thread CODEC 8M run spends ≈1.6% of samples on `TypeId` compares alone.
+The adapter resolves the handles once and caches them, which requires
+storing a reference whose real lifetime the struct cannot name.
+
+- **`crates/fgumi-pipeline-core/src/erased.rs`** — four `#[allow(unsafe_code)]`
+  sites, two on `TypedStep<S>` (`resolve_input`, `resolve_outputs`) and two on
+  `TypedStep2<S>` (`resolve_inputs`, `resolve_outputs`). Each is a
+  `std::mem::transmute` that extends a `&'a Handle` to `&'static Handle` for
+  storage in the cache slot, and narrows it back to `&'a` on read. No pointer
+  is dereferenced through the `'static` form. SAFETY rests on three invariants
+  documented on `TypedStep` itself: the handle boxes are owned by
+  `ChainContexts` (alive for the whole `Pipeline::run`), every step instance is
+  dropped before those contexts are, and every dispatch passes the *same* box
+  for a given `step_idx`. The third is the one the compiler cannot check, so
+  each cache slot stores the address of the erased box it was resolved from and
+  every cached hit `assert!`s it — **unconditionally, release included**, so a
+  step reused across two *live* `ChainContexts` panics instead of reading
+  through the wrong one. The check is one load and one compare, not the
+  `downcast_ref` `TypeId` probe the cache exists to elide. A debug-only
+  `debug_assert!` keeps re-resolving the typed pointer as a second diagnostic.
+  Treat the assert as defence in depth, not as the safety argument: it compares
+  data addresses only, so an allocator that reuses a freed box's address defeats
+  it. Soundness rests on the second invariant — every step instance is dropped
+  before the contexts it cached from — which is what must be preserved by any
+  future change.
+
 Any new `unsafe` site must extend this list and explain why the safe
 alternative is unacceptable. Do not introduce `unsafe` outside the crates
 listed in this section.

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/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/xtask"]
 resolver = "2"
 
 [workspace.package]
@@ -28,6 +28,7 @@ fgumi-fmt = { version = "0.5.0", path = "crates/fgumi-fmt" }
 fgumi-consensus = { version = "0.5.0", path = "crates/fgumi-consensus", default-features = false }
 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-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" }
@@ -82,6 +83,7 @@ serde = { version = "1.0.228", features = ["derive"] }
 sysinfo = { version = "0.38", default-features = false, features = ["system"] }
 tempfile = "3.3.0"
 thiserror = "2"
+trybuild = "1.0"
 wide = "1.5"
 
 [package]

crates/fgumi-cli-macros/Cargo.toml

@@ -19,7 +19,7 @@ proc-macro2 = "1"
 clap = { workspace = true }
 anyhow = { workspace = true }
 rstest = { workspace = true }
-trybuild = "1.0"
+trybuild = { workspace = true }
 
 [lints.clippy]
 pedantic = { level = "deny", priority = -1 }

crates/fgumi-pipeline-core/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "fgumi-pipeline-core"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+description = "Typed-step pipeline framework core (steps, queues, reorder stage, scheduler) for fgumi"
+repository.workspace = true
+license.workspace = true
+
+[dependencies]
+ahash = { workspace = true }
+anyhow = { workspace = true }
+crossbeam-queue = { workspace = true }
+log = { workspace = true }
+noodles = { workspace = true, features = ["sam"] }
+parking_lot = { workspace = true }
+
+[dev-dependencies]
+proptest = { workspace = true }
+rstest = { workspace = true }
+trybuild = { workspace = true }
+
+[lints.clippy]
+pedantic = { level = "deny", priority = -1 }