From b66b020e185684e41de2d07377194255af9c40ad Mon Sep 17 00:00:00 2001 From: Nils Homer Date: Sun, 13 Sep 2026 12:30:30 -0700 Subject: [PATCH] fix(extract): honor --check-crc/--no-check-crc on the BGZF FASTQ split decode The all-file BGZF parallel-decode split for FASTQ input (build_bgzf_fastq_split -> FastqDecompress) reads its per-block CRC32 policy from ChainSpec.verify_crc, but Extract::execute_chain hardcoded that field to false. As a result --check-crc and the file-input default silently skipped CRC verification on that path, accepting corrupted FASTQ blocks. This path shipped to main with the StepK FASTQ decode work (via #952, merged into #951). Resolve verify_crc through the shared resolve_check_crc policy (--check-crc wins, --no-check-crc disables, file inputs verify by default). The spec construction is extracted into a testable Extract::build_extract_chain_spec, mirroring Sort::build_sort_chain_spec. test_bgzf_fastq_honors_check_crc did not catch this: its small input is read in full during quality-encoding detection (sample_detection_quals), which opens its own policy-honoring reader and rejects the corrupted trailing block before the split decoder runs. Add split_decoder_honors_check_crc_past_detection_window, which places the corruption past the detection window so only the split decode reaches it, exercising the decoder's own policy. Also align the parse_and_zip_two_streams_preserves_ordinal doc with what it actually covers (the primitives, not the step; step wiring is covered by the chain tests). --- src/lib/commands/extract.rs | 151 +++++++++++++++--- .../pipeline/steps/source/parse_zip_fastq.rs | 17 +- 2 files changed, 142 insertions(+), 26 deletions(-) diff --git a/src/lib/commands/extract.rs b/src/lib/commands/extract.rs index 82ad48df2..23abf1cac 100644 --- a/src/lib/commands/extract.rs +++ b/src/lib/commands/extract.rs @@ -721,26 +721,35 @@ impl Extract { } } - /// Run extract on the declarative chain builder — the only execution path. + /// Hand-build the extract [`ChainSpec`] (FASTQ source → unmapped-BAM sink). /// - /// [`Command::execute`] always dispatches here, with or without `--threads` - /// (absent `--threads` runs the chain at a single worker). Hand-builds the - /// [`ChainSpec`] (rather than using [`ChainSpec::single_stage`], which is + /// Hand-built (rather than [`ChainSpec::single_stage`], which is /// BAM-in/BAM-out): the source is a FASTQ source — /// [`SourceSpec::InterleavedFastq`] for `--interleaved`, else /// [`SourceSpec::Fastqs`] — and the sink is a BAM. The chain opens its own - /// readers and detects the quality encoding in `ChainBuilder::open_source`; - /// `read_streams`/`verify_crc` are BAM-reader knobs and inert here (the FASTQ - /// source carries its CRC policy in [`ExtractOptions`]). + /// readers and detects the quality encoding in `ChainBuilder::open_source`. + /// + /// Split out from [`Self::execute_chain`] so the flag-derived spec fields — + /// notably the resolved CRC policy — can be unit-tested without running the + /// pipeline, mirroring `Sort::build_sort_chain_spec`. + /// + /// `read_streams` is a seekable-BAM knob and inert for a FASTQ source, but + /// `verify_crc` is NOT inert: the all-file BGZF parallel-decode split + /// (`ChainBuilder::build_bgzf_fastq_split` → `FastqDecompress`) reads it as + /// its per-block CRC32 policy. It is resolved here through the shared + /// [`resolve_check_crc`] so `--check-crc` / `--no-check-crc` reach that + /// decoder instead of it silently skipping verification. /// /// [`ChainSpec`]: crate::pipeline::chains::ChainSpec /// [`ChainSpec::single_stage`]: crate::pipeline::chains::ChainSpec::single_stage /// [`SourceSpec::Fastqs`]: crate::pipeline::chains::SourceSpec::Fastqs /// [`SourceSpec::InterleavedFastq`]: crate::pipeline::chains::SourceSpec::InterleavedFastq - fn execute_chain(&self, command_line: &str) -> Result<()> { - use crate::pipeline::chains::{ - ChainSpec, SinkSpec, SourceSpec, Stage, StageOptionsBag, build_for, - }; + /// [`resolve_check_crc`]: crate::commands::common::resolve_check_crc + fn build_extract_chain_spec( + &self, + command_line: &str, + ) -> Result { + use crate::pipeline::chains::{ChainSpec, SinkSpec, SourceSpec, Stage, StageOptionsBag}; let read_structures = self.get_read_structures()?; let source = if self.interleaved { @@ -752,7 +761,20 @@ impl Extract { let stage_opts = StageOptionsBag { extract: Some(self.to_extract_options()), ..Default::default() }; - let spec = ChainSpec { + // `verify_crc` is consumed only by the all-file BGZF parallel-decode + // split; the fused reader path derives its own per-path policy directly + // from `check_crc`/`no_check_crc`. The split's eligibility gate + // guarantees every input is a non-stdin BGZF file, so + // `resolve_check_crc`'s stdin-trust branch never applies on that path — + // resolve against the first input (a real file whenever the split runs; + // the value is inert when the fused path handles the source). + let verify_crc = crate::commands::common::resolve_check_crc( + self.check_crc, + self.no_check_crc, + &self.inputs[0], + ); + + Ok(ChainSpec { stages: vec![Stage::Extract], source, sink: SinkSpec::Bam(self.output.clone()), @@ -762,15 +784,19 @@ impl Extract { scheduler: self.scheduler_opts.clone(), queue_memory: self.queue_memory.clone(), async_reader: self.async_reader, - // Inert for a FASTQ source: `read_streams` is a seekable-BAM knob and - // `verify_crc` is the BAM BGZF-decode policy; the FASTQ source carries - // its own CRC policy in `ExtractOptions` (`check_crc`/`no_check_crc`). read_streams: fgumi_bam_io::ReadStreams::Fixed(1), - verify_crc: false, + verify_crc, command_line: command_line.to_string(), - }; + }) + } - build_for(spec)?.run() + /// Run extract on the declarative chain builder — the only execution path. + /// + /// [`Command::execute`] always dispatches here, with or without `--threads` + /// (absent `--threads` runs the chain at a single worker). Builds the spec + /// via [`Self::build_extract_chain_spec`] and runs it. + fn execute_chain(&self, command_line: &str) -> Result<()> { + crate::pipeline::chains::build_for(self.build_extract_chain_spec(command_line)?)?.run() } /// Get actual read structures (default to +T if none provided). @@ -1903,9 +1929,15 @@ mod tests { /// A corrupted-CRC BGZF FASTQ must be rejected by default and with /// `--check-crc`, and read clean with `--no-check-crc` (#819). Parameterized - /// over threading so both BGZF decode paths are covered: single-threaded goes - /// through `open_fastq_reader`'s fgumi-bgzf arm, while `--threads N` decodes - /// the pure-BGZF input in the pipeline honoring `FastqPipelineConfig::verify_crc`. + /// over threading at both thread counts. + /// + /// NB: this small input is read in full during quality-encoding detection + /// (`sample_detection_quals`, which opens its own policy-honoring reader), so + /// the corrupted trailing block is caught there before the all-BGZF split + /// decoder runs. It therefore pins the command's end-to-end CRC behavior but + /// does NOT prove the split decoder itself honors the policy — that is + /// covered by `split_decoder_honors_check_crc_past_detection_window`, whose + /// corruption sits past the detection window. #[rstest] #[case::single_threaded(ThreadingOptions::none())] #[case::multi_threaded(ThreadingOptions::new(2))] @@ -1950,6 +1982,83 @@ mod tests { assert_crc_error_msg(&format!("{check_err:#}")); } + /// `--check-crc` / `--no-check-crc` reach the extract `ChainSpec.verify_crc` + /// through the shared [`resolve_check_crc`] policy, so the all-file BGZF + /// parallel-decode split (`ChainBuilder::build_bgzf_fastq_split` → + /// `FastqDecompress`) honors the flags instead of silently skipping CRC + /// verification. The field was previously hardcoded `false`, disabling CRC + /// on that path regardless of the flags. Mirrors + /// `Sort::build_sort_chain_spec_resolves_verify_crc`. + #[rstest] + #[case::file_default_verifies(false, false, true)] + #[case::file_no_check_crc_skips(false, true, false)] + #[case::file_check_crc_verifies(true, false, true)] + fn build_extract_chain_spec_resolves_verify_crc( + #[case] check_crc: bool, + #[case] no_check_crc: bool, + #[case] expected: bool, + ) { + // A file path (never stdin) — the split's eligibility gate the field + // feeds only fires on real files, so the file-default policy applies. + let extract = bgzf_crc_extract( + PathBuf::from("in.fq.gz"), + PathBuf::from("out.bam"), + ThreadingOptions::none(), + check_crc, + no_check_crc, + ); + let spec = extract + .build_extract_chain_spec("fgumi extract (test)") + .expect("spec build should succeed"); + assert_eq!(spec.verify_crc, expected, "file input CRC policy on the FASTQ split"); + } + + /// End-to-end guard that the BGZF FASTQ split decoder itself honors the CRC + /// policy, not just that the spec carries it. `test_bgzf_fastq_honors_check_crc` + /// cannot prove this: its small input is fully read during quality-encoding + /// detection (`sample_detection_quals`, which opens its own policy-honoring + /// reader), so detection catches a corrupted trailing block before the split + /// decoder ever runs. Here the corruption sits far past the detection window + /// (`QUALITY_DETECTION_SAMPLE_SIZE` records / `BUFFER_SIZE` bytes), so only + /// the full split decode reaches it — isolating the decoder's own policy. + #[rstest] + #[case::single_threaded(ThreadingOptions::none())] + #[case::multi_threaded(ThreadingOptions::new(2))] + fn split_decoder_honors_check_crc_past_detection_window(#[case] threading: ThreadingOptions) { + // ~350k records * ~29 bytes ≈ 10 MiB uncompressed. Detection reads only + // ~1 MiB (BUFFER_SIZE) after sampling 400 records, so the corrupted last + // block is untouched by detection and reached only by the split decode. + const NUM_RECORDS: usize = 350_000; + let tmp = TempDir::new().expect("temp dir"); + let input = create_bgzf_fastq(&tmp, "big.fq.gz", NUM_RECORDS); + corrupt_last_block_crc(&input); + + // --no-check-crc: the split decoder skips the corrupted CRC32 and emits + // every record. + let out_ok = tmp.path().join("ok.bam"); + bgzf_crc_extract(input.clone(), out_ok.clone(), threading.clone(), false, true) + .execute("test") + .expect("--no-check-crc must accept a corrupted trailing block on the split path"); + assert_eq!(read_bam_records(&out_ok).len(), NUM_RECORDS, "all records extracted"); + + // Default (file => verify on): the split decoder rejects the corrupted + // trailing block. Before the fix, `verify_crc` was hardcoded `false` on + // the split, so this corruption was silently accepted. + let out_default = tmp.path().join("default.bam"); + let default_err = + bgzf_crc_extract(input.clone(), out_default, threading.clone(), false, false) + .execute("test") + .expect_err("the split decoder must reject a corrupted CRC32 by default"); + assert_crc_error_msg(&format!("{default_err:#}")); + + // --check-crc: same rejection, explicitly requested. + let out_check = tmp.path().join("check.bam"); + let check_err = bgzf_crc_extract(input, out_check, threading, true, false) + .execute("test") + .expect_err("the split decoder must reject a corrupted CRC32 under --check-crc"); + assert_crc_error_msg(&format!("{check_err:#}")); + } + /// `open_fastq_reader` must honor `--no-check-crc` on BGZF input at **any** /// thread count. At `threads > 1` the parallel noodles decoder always /// verifies, so the skip has to route through fgumi-bgzf instead — a diff --git a/src/lib/pipeline/steps/source/parse_zip_fastq.rs b/src/lib/pipeline/steps/source/parse_zip_fastq.rs index cc9b14dd6..03dd5f98d 100644 --- a/src/lib/pipeline/steps/source/parse_zip_fastq.rs +++ b/src/lib/pipeline/steps/source/parse_zip_fastq.rs @@ -136,11 +136,18 @@ mod tests { assert_eq!(p.branch_ordering, vec![BranchOrdering::ByItemOrdinal]); } - /// End-to-end of the step body's parse+zip: an `NRawFastqBatch` of two - /// aligned raw byte-chunks parses and zips into templates, and the batch's - /// `ordinal` is preserved verbatim as the output `batch_serial`. (The parse - /// and zip primitives themselves are unit-tested in `super::parse_fastq_chunk` - /// and `super::fastq_zip::zip_streams`; this pins the wiring in this step.) + /// Composes the primitives `ParseAndZipFastqN::try_run` chains — parse each + /// stream's chunk (`super::super::parse_fastq_chunk`), zip the per-stream + /// records into templates (`super::super::fastq_zip::zip_streams`), and carry + /// the input `ordinal` verbatim into the output batch's `batch_serial` — and + /// asserts the resulting templates and preserved ordinal. + /// + /// This is a primitive-level check: it does NOT construct `ParseAndZipFastqN` + /// or drive `try_run`, so it cannot detect step-level changes to stream + /// forwarding or ordinal propagation. That step wiring is exercised through + /// `Pipeline::run` by the chain tests + /// `unified_fastq_chain_recovers_every_template_in_order` and + /// `bgzf_split_mismatched_block_layouts_stay_in_lockstep`. #[test] fn parse_and_zip_two_streams_preserves_ordinal() { let mut data_a = record_bytes("read1/1", "ACGT");