test: close spill/cancellation coverage gaps + rstest-ify reject loop (audit E1/E2/E3)

tests/integration/test_runall_parity.rs

@@ -1229,6 +1229,143 @@ fn runall_reads_bam_from_stdin_once() {
     assert_bams_record_equivalent_nonempty(&from_file, &from_stdin);
 }
 
+/// The `(offset, len)` of every BGZF block in `bam`, in file order.
+///
+/// Only the block framing is read: `BSIZE` lives at offset 16 of each block
+/// header and holds the total block size minus one. That is enough to pick a
+/// truncation point by structure rather than by guessing at a byte fraction.
+fn bgzf_block_bounds(bam: &[u8]) -> Vec<(usize, usize)> {
+    const BGZF_HEADER_SIZE: usize = 18;
+    const BSIZE_OFFSET: usize = 16;
+
+    let mut bounds = Vec::new();
+    let mut offset = 0;
+    while offset + BGZF_HEADER_SIZE <= bam.len() {
+        let bsize =
+            u16::from_le_bytes([bam[offset + BSIZE_OFFSET], bam[offset + BSIZE_OFFSET + 1]]);
+        let block_len = usize::from(bsize) + 1;
+        // A malformed or truncated trailing block would otherwise loop forever or
+        // report a block running past the buffer.
+        if block_len < BGZF_HEADER_SIZE || offset + block_len > bam.len() {
+            break;
+        }
+        bounds.push((offset, block_len));
+        offset += block_len;
+    }
+    bounds
+}
+
+/// CG-5 (audit E2): every runall failure test rejects *before* the pipeline runs
+/// (invalid stage pair, missing `--unmapped`/`--ref`/`--umis`, bad flag combo).
+/// None checks that the fused BAM-in pipeline fails *cleanly mid-run* when a
+/// downstream stage hits malformed input. Feed `runall --start-from group
+/// --stop-after consensus --consensus simplex` a BAM truncated mid-stream and
+/// assert a clean non-zero exit — no panic / `unreachable`, and (via a
+/// wall-clock watchdog) no hang.
+#[test]
+fn runall_group_simplex_fails_cleanly_on_truncated_bam() {
+    use std::io::Read as _;
+    use std::process::{Command, Stdio};
+    use std::time::{Duration, Instant};
+
+    let tmp = TempDir::new().unwrap();
+    let fixture = sorted_simplex_fixture(tmp.path());
+
+    // Truncate the sorted input BAM so the reader errors partway through a real
+    // record stream rather than failing an up-front header parse.
+    //
+    // The cut point is derived from the file's actual BGZF block boundaries, not
+    // from a fraction of its length. A fraction cannot state the property this
+    // test needs — "past the header block, inside a record block" — and on this
+    // fixture it very nearly fails to hold it: the file is ~592 bytes with its
+    // header block ending at ~329, so a 60% cut landed only ~26 bytes inside the
+    // first record block. Any growth in the header (a longer `@PG CL` from a
+    // longer temp path, a longer version string) would push the cut back into the
+    // header block, silently turning this into the up-front-parse-failure case
+    // that other tests already cover — while still passing.
+    let bytes = fs::read(&fixture).expect("read fixture");
+    let blocks = bgzf_block_bounds(&bytes);
+    assert!(
+        blocks.len() >= 3,
+        "fixture must have a header block, at least one record block, and the EOF block to \
+         truncate meaningfully; got {} block(s) in {} bytes",
+        blocks.len(),
+        bytes.len()
+    );
+    // Halfway into the second block: past the header, inside real record data.
+    let (record_block_start, record_block_len) = blocks[1];
+    let keep = record_block_start + record_block_len / 2;
+    assert!(
+        keep > record_block_start && keep < record_block_start + record_block_len,
+        "cut point {keep} must fall strictly inside the first record block \
+         [{record_block_start}, {})",
+        record_block_start + record_block_len
+    );
+    let truncated = tmp.path().join("truncated.bam");
+    fs::write(&truncated, &bytes[..keep]).expect("write truncated bam");
+
+    let out_path = tmp.path().join("out.bam");
+    let args = ParityArgs::for_simplex_like();
+    let mut child = Command::new(fgumi_binary())
+        .args([
+            "runall",
+            "--start-from",
+            "group",
+            "--stop-after",
+            "consensus",
+            "--consensus",
+            "simplex",
+            "--input",
+            truncated.to_str().unwrap(),
+            "--output",
+            out_path.to_str().unwrap(),
+            "--group::strategy",
+            args.strategy,
+            "--group::edits",
+            &args.edits.to_string(),
+            "--simplex::min-reads",
+            args.min_reads,
+            // Deliberately more than `args.threads` (1): a mid-run failure has to
+            // stay clean when several workers are in flight, which is where a
+            // partial-shutdown panic or a hang would actually show up.
+            "--threads",
+            "2",
+        ])
+        .stdout(Stdio::null())
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("spawn fgumi runall");
+
+    // Drain stderr on a worker so a hung child (which never closes the pipe)
+    // cannot block the watchdog loop below.
+    let mut stderr_pipe = child.stderr.take().expect("child stderr");
+    let reader = std::thread::spawn(move || {
+        let mut s = String::new();
+        let _ = stderr_pipe.read_to_string(&mut s);
+        s
+    });
+
+    // Wall-clock watchdog: a mid-run corruption must fail fast, never hang.
+    let deadline = Instant::now() + Duration::from_secs(60);
+    let status = loop {
+        if let Some(status) = child.try_wait().expect("try_wait") {
+            break status;
+        }
+        if Instant::now() >= deadline {
+            let _ = child.kill();
+            panic!("runall did not exit within 60s on truncated input (hang?)");
+        }
+        std::thread::sleep(Duration::from_millis(20));
+    };
+    let stderr = reader.join().expect("join stderr reader");
+
+    assert!(!status.success(), "runall must fail on a truncated BAM; stderr:\n{stderr}");
+    assert!(
+        !stderr.contains("panicked") && !stderr.contains("unreachable"),
+        "runall must fail cleanly (no panic/unreachable) on truncated input; stderr:\n{stderr}"
+    );
+}
+
 /// Write a gzip-compressed FASTQ from `(name, seq, qual)` records.
 fn write_gzip_fastq(path: &Path, records: &[(&str, &str, &str)]) {
     use std::io::Write as _;

tests/integration/test_sort_correctness.rs

@@ -505,6 +505,121 @@ fn coordinate_sort_matrix(#[case] spill: Spill) {
     assert_records_preserved(&input, &output);
 }
 
+/// CG-3 (audit E1): the spill matrix drives `-m` down to *force* spills but
+/// never asserts one actually happened. If memory accounting regressed so the
+/// engine ignored `-m` and held everything in RAM, every matrix case would still
+/// pass (output is still correct) — silently deleting all spill/merge coverage
+/// *and* the memory-bound guarantee. Pin the lever directly by observing the
+/// arena `SortMerge` step's INFO log: a tiny `-m` must take the disk-spill merge
+/// path ("Sort merge complete"), while a large `-m` must take the single-source
+/// in-memory fast path ("in-memory fast path complete"). Run as a subprocess so
+/// `RUST_LOG=info` is honored, with `--threads 1` so the in-memory case is a
+/// single source (deterministic fast path).
+/// `expect_multiple_sources` is what separates `single_spill` from `many_spill`:
+/// both take the merge path, so asserting only `expect_spill` would make the two
+/// cases byte-identical and their labels would claim a distinction nothing checks
+/// — `many_spill` would still pass if `-m 64K` regressed to a single chunk. The
+/// merge's own INFO line reports the chunk count (`… (N records, M sources)`), so
+/// the spill *count* is observable: measured here as 1 source at `-m 200K` and 4
+/// at `-m 64K`. Asserted as one-vs-many rather than an exact 4, which would break
+/// on any record-size or compression tuning without indicating a real regression.
+#[rstest]
+#[case::in_memory("256M", false, None)]
+#[case::single_spill("200K", true, Some(false))]
+#[case::many_spill("64K", true, Some(true))]
+fn sort_spill_actually_occurs_under_small_memory(
+    #[case] max_memory: &str,
+    #[case] expect_spill: bool,
+    #[case] expect_multiple_sources: Option<bool>,
+) {
+    let (dir, input) = build_unsorted_fixture(1500);
+    let output = dir.path().join("sorted.bam");
+
+    let out = Command::new(env!("CARGO_BIN_EXE_fgumi"))
+        .env("RUST_LOG", "info")
+        .args([
+            "sort",
+            "-i",
+            input.to_str().unwrap(),
+            "-o",
+            output.to_str().unwrap(),
+            "--order",
+            "coordinate",
+            "--threads",
+            "1",
+            "-m",
+            max_memory,
+            "--temp-compression",
+            "1",
+        ])
+        .output()
+        .expect("spawn fgumi sort");
+    assert!(
+        out.status.success(),
+        "fgumi sort failed (-m {max_memory}): {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let stderr = String::from_utf8_lossy(&out.stderr);
+
+    let merged = stderr.contains("Sort merge complete");
+    let in_memory = stderr.contains("in-memory fast path complete");
+    if expect_spill {
+        assert!(
+            merged && !in_memory,
+            "-m {max_memory} must force a disk-spill merge, not the in-memory fast path.\n\
+             stderr:\n{stderr}"
+        );
+    } else {
+        assert!(
+            in_memory && !merged,
+            "-m {max_memory} must sort entirely in memory (no spill/merge).\nstderr:\n{stderr}"
+        );
+    }
+
+    // Pin how *many* chunks spilled, not just that spilling happened, so the
+    // `single_spill` / `many_spill` labels mean something. The merge logs
+    // `… (N records, M sources)`; `M` is its chunk count.
+    if let Some(expect_multiple) = expect_multiple_sources {
+        let sources = stderr
+            .lines()
+            .find_map(|line| {
+                // "… Sort merge diag: … (3020 records, 4 sources)"
+                let (_, tail) = line.split_once("Sort merge diag:")?;
+                let (_, after_records) = tail.split_once("records, ")?;
+                let (count, _) = after_records.split_once(" sources")?;
+                count.parse::<usize>().ok()
+            })
+            .unwrap_or_else(|| {
+                panic!(
+                    "-m {max_memory} spilled, so the merge must log its source count.\n\
+                     stderr:\n{stderr}"
+                )
+            });
+        if expect_multiple {
+            assert!(
+                sources > 1,
+                "-m {max_memory} must spill more than one chunk (this is the `many_spill` case, \
+                 and a single chunk would make it a duplicate of `single_spill`); got {sources} \
+                 source(s).\nstderr:\n{stderr}"
+            );
+        } else {
+            assert_eq!(
+                sources, 1,
+                "-m {max_memory} must spill exactly one chunk (the `single_spill` case); got \
+                 {sources} source(s).\nstderr:\n{stderr}"
+            );
+        }
+    }
+
+    // The output must still be correct regardless of regime: the tool's own
+    // --verify guards against write corruption, and an INDEPENDENT in-test
+    // coordinate-order oracle (re-derived from the records via noodles, not the
+    // tool's --verify path) proves the order — mirroring `coordinate_sort_matrix`.
+    assert!(verify_sorted(&output, "coordinate"), "coordinate --verify guard failed");
+    assert_coordinate_ordered(&read_records(&output));
+    assert_records_preserved(&input, &output);
+}
+
 // ---------------------------------------------------------------------------
 // Queryname-lexicographic sort: full spill matrix, independent in-test oracle
 // ---------------------------------------------------------------------------

tests/integration/test_streaming_input.rs

@@ -356,44 +356,45 @@ fn test_simplex_single_threaded_reads_stdin_once() {
 /// and the multi-threaded path is the SAM-capable one. Guards the `with_context`
 /// message that replaced the removed SAM pre-sniff (codec/simplex) and the
 /// matching hint added to duplex.
-#[test]
-fn test_single_threaded_consensus_rejects_sam_with_threads_hint() {
+///
+/// Each single-threaded consensus command must reject SAM input with a
+/// `--threads` hint. One `#[case]` per command so a regression names the
+/// command instead of dying on the first bad assert (repo rstest convention).
+#[rstest]
+#[case::codec("codec", &["--min-reads", "1", "--min-duplex-length", "1"])]
+#[case::simplex("simplex", &["--min-reads", "1"])]
+#[case::duplex("duplex", &["--min-reads", "1"])]
+fn test_single_threaded_consensus_rejects_sam_with_threads_hint(
+    #[case] cmd: &str,
+    #[case] extra: &[&str],
+) {
     let temp_dir = TempDir::new().expect("Failed to create temp dir");
     let bam = temp_dir.path().join("grouped.bam");
     let sam = temp_dir.path().join("grouped.sam");
     create_grouped_test_bam(&bam);
     convert_bam_to_sam(&bam, &sam);
     let input = sam.to_str().unwrap();
 
-    // Per-command minimal args; each runs single-threaded (no `--threads`).
-    let extra_args: [(&str, &[&str]); 3] = [
-        ("codec", &["--min-reads", "1", "--min-duplex-length", "1"]),
-        ("simplex", &["--min-reads", "1"]),
-        ("duplex", &["--min-reads", "1"]),
-    ];
-    for (cmd, extra) in extra_args {
-        let out = temp_dir.path().join(format!("{cmd}_out.bam"));
-        let mut args: Vec<&str> = vec![
-            cmd,
-            "--input",
-            input,
-            "--output",
-            out.to_str().unwrap(),
-            "--compression-level",
-            "1",
-        ];
-        args.extend_from_slice(extra);
-        let output = Command::new(env!("CARGO_BIN_EXE_fgumi"))
-            .args(&args)
-            .output()
-            .unwrap_or_else(|e| panic!("failed to spawn {cmd}: {e}"));
-        assert!(!output.status.success(), "single-threaded {cmd} must reject SAM input");
-        let stderr = String::from_utf8_lossy(&output.stderr);
-        assert!(
-            stderr.contains("--threads"),
-            "single-threaded {cmd} SAM error should hint at --threads; stderr: {stderr}"
-        );
-    }
+    let out = temp_dir.path().join(format!("{cmd}_out.bam"));
+    // `--threads` is deliberately absent, and must stay absent: omitting it is the
+    // *only* way to select the single-threaded fast path this test is about.
+    // `--threads 1` is not a synonym — per `--threads`' own help, "when specified
+    // (even with --threads 1), uses the typed-step work-stealing pipeline", which is
+    // the SAM-capable path, so adding it makes all three cases accept the SAM input
+    // and the assertion below fail.
+    let mut args: Vec<&str> =
+        vec![cmd, "--input", input, "--output", out.to_str().unwrap(), "--compression-level", "1"];
+    args.extend_from_slice(extra);
+    let output = Command::new(env!("CARGO_BIN_EXE_fgumi"))
+        .args(&args)
+        .output()
+        .unwrap_or_else(|e| panic!("failed to spawn {cmd}: {e}"));
+    assert!(!output.status.success(), "single-threaded {cmd} must reject SAM input");
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(
+        stderr.contains("--threads"),
+        "single-threaded {cmd} SAM error should hint at --threads; stderr: {stderr}"
+    );
 }
 
 /// `--async-reader` over stdin must spawn the prefetch thread (the stdin branch
@@ -407,7 +408,8 @@ fn test_simplex_single_threaded_async_reader_over_stdin() {
     let output_from_pipe = temp_dir.path().join("output_pipe.bam");
     create_grouped_test_bam(&input_bam);
 
-    // Baseline: single-threaded + --async-reader from a file.
+    // Baseline: single-threaded (no `--threads`, which is what selects the
+    // single-threaded fast path) + --async-reader from a file.
     let status = Command::new(env!("CARGO_BIN_EXE_fgumi"))
         .args([
             "simplex",