feat(pipeline): typed-step pipeline foundation

.cargo/config.toml

@@ -14,7 +14,14 @@ ci-publish-order = "run --package xtask -- check-publish-order"
 ci-doctest = "test --doc --workspace --features compare,simulate,profile-adjacency --locked"
 # Build the docs and fail on any rustdoc warning (e.g. broken intra-doc links).
 # RUSTDOCFLAGS="-D warnings" is set by the caller (CI job / pre-push).
-ci-doc = "doc --no-deps --workspace --features compare,simulate,profile-adjacency --locked"
+#
+# `--document-private-items` is load-bearing, not cosmetic. Without it rustdoc
+# only checks links on items it renders — i.e. public ones — so a broken link in
+# the doc comment of a private fn, a private field, or a `#[cfg(test)]` helper
+# passes silently. That is not hypothetical: several such links accumulated
+# undetected and had to be found by hand. The flag makes the docs job check every
+# doc comment in the workspace, which is what the job's name implies it does.
+ci-doc = "doc --no-deps --workspace --document-private-items --features compare,simulate,profile-adjacency --locked"
 # Run only the #[ignore]-d integration tests (the sort-correctness suites in
 # tests/integration/{test_sort_correctness,test_async_reader,test_sort_write_index}.rs).
 # They require the `samtools` binary on PATH — samtools builds the BAM fixtures,
@@ -34,7 +41,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)

.coderabbit.yaml

@@ -164,22 +164,28 @@ reviews:
   # constructs. Re-read and update them whenever those modules are refactored or
   # renamed — a stale instruction silently misdirects the reviewer, and the
   # reviewer will never tell you it is chasing a symbol that no longer exists.
-  # Every symbol named below was verified with `git grep` to resolve on `main`
-  # at the commit that introduced this revision; re-verify when touching these
-  # modules.
+  # Every symbol named below was verified with `git grep` to resolve at the
+  # commit that introduced this revision; re-verify when touching these modules.
   #
-  # Deliberately NOT listed, because they do not exist on `main` and a glob that
-  # matches nothing is exactly the failure mode above — add an entry in the same
-  # PR that lands the code:
-  #   - `crates/fgumi-pipeline-core/**`, `crates/fgumi-pipeline-io/**` (on
-  #     `feat-runall`; the byte-bound and unbounded-drain rules in the
-  #     `unified_pipeline` entry apply to them verbatim once merged)
+  # The two legacy homes share one entry: issue #330 renames
+  # `src/lib/unified_pipeline` to `src/lib/pipeline`, so the same instructions
+  # apply to whichever is present. `crates/fgumi-pipeline-core` is the extracted
+  # typed-step engine and gets its OWN entry below: it is a different design and
+  # names different symbols (`queues.rs` / `ByteBoundedQueue`, not `queue.rs` /
+  # `OrderedQueue`), so folding it into the legacy entry would point the reviewer
+  # at symbols that do not exist there.
+  #
+  # Not yet listed, and a glob that matches nothing is exactly the failure mode
+  # above — add an entry in the same PR that lands the code:
+  #   - `crates/fgumi-pipeline-io/**` — does not exist here yet; the byte-bound
+  #     and unbounded-drain rules in the pipeline entry apply to it verbatim once
+  #     it lands.
   #   - `crates/fgumi-fmt/**`, `crates/fgumi-cli-common/**`,
-  #     `crates/fgumi-cli-macros/**` (in flight)
-  # `src/lib/pipeline/**` is listed alongside `unified_pipeline` only because
-  # issue #330 renames the latter to the former; today it matches nothing.
+  #     `crates/fgumi-cli-macros/**` — these DO exist here but still have no
+  #     instructions of their own, so they are reviewed by the generic themes
+  #     only.
   path_instructions:
-    - path: "src/lib/{unified_pipeline,pipeline}/**/*.rs"
+    - path: "{src/lib/unified_pipeline,src/lib/pipeline}/**/*.rs"
       instructions: >-
         This is the hand-rolled concurrent step pipeline; its bugs are deadlocks,
         lost output, unbounded memory, and panics on malformed input — not style.
@@ -207,6 +213,35 @@ reviews:
         a real, previously-shipped crash on corruption-controlled input. This
         module contains no `unsafe`; treat newly introduced `unsafe` here as out
         of policy and require it to be justified in CLAUDE.md first.
+    - path: "crates/fgumi-pipeline-core/src/**/*.rs"
+      instructions: >-
+        This is the extracted typed-step pipeline engine (the successor to the
+        `src/lib/*_pipeline` design above); its bugs are deadlocks, lost output,
+        unbounded memory, and panics on malformed input — not style. It names
+        DIFFERENT symbols than the legacy entry: queues live in `queues.rs`
+        (`ItemQueue` trait, `QueueSpec`, `CountBoundedQueue`, `ByteBoundedQueue`,
+        `UnboundedQueue`), NOT `queue.rs`/`OrderedQueue`. Require that no queue or
+        reorder buffer grows without a bound: byte-bounded transports go through
+        `ByteBoundedQueue<T: HeapSize>` (steady-state memory a function of config,
+        not input size), and consumers/drains must stay unbounded — bound the
+        producers instead. Flag a cap checked on only one sub-condition so another
+        path bypasses it, and byte accounting that measures logical length where
+        the memory held is allocation capacity. Cancellation and error propagation
+        run through `PipelineSignal` / `CancelHandle` (`signal.rs`): require every
+        worker loop and the fused driver to observe `signal.is_done()` promptly and
+        never block forever on a channel whose peer has exited. Liveness for the
+        deadlock monitor is the per-worker `LivenessCounter` (`liveness.rs`), bumped
+        only on productive `StepOutcome::Progress`/`Finished`; the fused
+        single-thread path (`runtime/fused.rs`) is NOT monitored and carries its own
+        stall budget instead. The ONLY approved `unsafe` in this crate is the typed-
+        handle dispatch cache in `erased.rs` (`TypedStep`/`TypedStep2`
+        `resolve_input`/`resolve_outputs`), whose soundness rests on the invariants
+        documented in CLAUDE.md — chiefly that every step instance is dropped before
+        the `ChainContexts` it cached from; flag any change that weakens those
+        invariants or adds `unsafe` elsewhere without a CLAUDE.md allowlist entry.
+        Byte-level framing that parses length/offset fields out of the input stream
+        must validate against the remaining buffer with checked arithmetic before
+        slicing.
     - path: "crates/fgumi-sort/**/*.rs"
       instructions: >-
         This is the sort engine, including approved `unsafe` hot paths (LSD radix

.github/workflows/check.yml

@@ -61,6 +61,39 @@ jobs:
       - name: Clippy check
         run: cargo ci-lint
 
+  # `merge_slots.rs` swaps `std::sync` for `loom::sync` under `--cfg loom`, and
+  # `tests/loom_merge_slots.rs` is `#![cfg(loom)]`. Neither is built by any other
+  # job, so without this one a broken model — or a `cfg(loom)` build that stopped
+  # compiling — is invisible: the test target compiles to an empty binary under a
+  # normal build and reports success. That silence is the whole reason this job
+  # exists.
+  #
+  # Per-PR rather than on a schedule, unlike `miri.yml` and `stress.yml`. Those
+  # two are scheduled because they are non-deterministic in ways unrelated to a
+  # given change (a nightly-toolchain regression; timing sensitivity). Loom
+  # explores a bounded state space deterministically — four of the five models
+  # under a preemption bound, see the test's "Preemption-bounded exploration"
+  # note — so the same input yields the same verdict with no flake, which is
+  # what lets it gate a PR, and a gate is what stops it rotting.
+  loom:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+      - name: Set up compilation cache (sccache)
+        uses: mozilla-actions/sccache-action@fc920bf0ec8de6ee65d409111f7ec508035751ba # v0.0.11
+      # Release: the models explore enough interleavings that a debug build takes
+      # several times as long for the same verdict.
+      - name: Loom model check (SortMergeSlot)
+        run: cargo test -p fgumi-sort --test loom_merge_slots --release
+        env:
+          RUSTFLAGS: --cfg loom
+
   coverage:
     runs-on: ubuntu-latest
     timeout-minutes: 20
@@ -84,7 +117,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:

.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).
 #
 # Two triggers, two jobs to do.
 #
@@ -117,7 +124,80 @@ jobs:
           # discarding); and reading — on failure, writing — the
           # `.proptest-regressions` file the lookup resolves to.
           MIRIFLAGS: -Zmiri-disable-isolation
-        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 a
+          # NEW `#[allow(unsafe_code)]` site appears outside the modules known to
+          # carry approved unsafe, the `sort` filter keeps matching and Miri
+          # silently stops checking the moved/new site — so flag it.
+          #
+          # Two modules are excluded because both carry documented, approved
+          # unsafe (see CLAUDE.md → "Unsafe Code"):
+          #   - `sort.rs` — the natural-order queryname comparator, which the
+          #     `sort` filter below DOES exercise under Miri.
+          #   - `raw_bam_record.rs` — `read_raw_record`'s spare-capacity read.
+          #     The `sort` filter does not run its tests, so this unsafe is not
+          #     Miri-checked here (matching `main`, whose miri job runs only the
+          #     `sort` filter too). Bringing it under Miri needs a separate
+          #     `read_raw_record`-filtered run; tracked as a follow-up. Excluded
+          #     here so this PRE-EXISTING approved site does not fail the guard.
+          stray=$(grep -rlE '^[[:space:]]*#!?\[[^]]*allow\(unsafe_code\)' crates/fgumi-raw-bam/src \
+            | grep -vE '^crates/fgumi-raw-bam/src/(sort|raw_bam_record)\.rs$' || true)
+          if [ -n "${stray}" ]; then
+            echo "::error::unsafe_code outside the Miri-scoped fgumi-raw-bam modules: ${stray}"
+            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"
+            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 -rlE '^[[:space:]]*#!?\[[^]]*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}"
+            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"
+            exit 1
+          fi
+          cargo +nightly miri test -p fgumi-pipeline-core erased
 
   # Nothing watches a cron's result, so a failure has to announce itself. See the
   # action for why it reuses one issue per workflow.