From 387aa1cec67165323a6e5aa76da4247a51f71b01 Mon Sep 17 00:00:00 2001 From: Nils Homer Date: Thu, 20 Aug 2026 01:07:08 -0700 Subject: [PATCH] fix(cli): reject colliding output paths across every command output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #689 guarded only `--output` vs `--rejects`, and only in the consensus commands. The same collision — two of a command's outputs resolving to one destination — was unguarded for `--stats`, `--metrics`, histograms, and the `--metrics PREFIX` expansions, and is worse there: the write happens after the BAM is complete, so `fgumi simplex -o out.bam --stats out.bam` writes a correct BAM and then truncates it to a TSV. Generalize the pairwise guard into a set-based `reject_output_collisions` that takes every output path a command will open as `(path, flag)` pairs and rejects any two that resolve to one destination (stdout may be named at most once; the null device stays exempt). Wire it into simplex, duplex, codec, filter, correct, downsample, group (including the three `--metrics PREFIX` files, `-f`, and `-g`), dedup, and clip, each handing the guard its full output list before any writer opens. Also fix `open_pipeline_output` to delegate to `open_output_writer` rather than returning a `LineWriter`-backed `std::io::stdout()`, which tears every BGZF flush at each 0x0a; `open_output_writer` is the single place the `-` convention is honoured for BAM output. Addresses #715 (items 2 and 3). Item 1 — moving identity into the writer layer with a `dev`+`ino` check to catch symlink/hardlink/case-insensitive collisions a pre-open path comparison cannot — remains open as a follow-up. --- src/lib/commands/clip.rs | 8 + src/lib/commands/codec.rs | 12 +- src/lib/commands/common.rs | 168 +++++++++++++++------ src/lib/commands/correct.rs | 14 +- src/lib/commands/dedup.rs | 14 ++ src/lib/commands/downsample.rs | 16 +- src/lib/commands/duplex.rs | 14 +- src/lib/commands/filter.rs | 13 +- src/lib/commands/group.rs | 21 +++ src/lib/commands/simplex.rs | 12 +- src/lib/unified_pipeline/bam.rs | 19 +-- tests/integration/test_streaming_output.rs | 49 +++++- 12 files changed, 285 insertions(+), 75 deletions(-) diff --git a/src/lib/commands/clip.rs b/src/lib/commands/clip.rs index c817b8e5e..7d5283a1d 100644 --- a/src/lib/commands/clip.rs +++ b/src/lib/commands/clip.rs @@ -499,6 +499,14 @@ impl ClipParams { impl Command for Clip { #[allow(clippy::too_many_lines)] fn execute(&self, command_line: &str) -> Result<()> { + // Reject two outputs resolving to one destination before any writer opens. + let mut outputs: Vec<(&std::path::Path, &str)> = + vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.metrics { + outputs.push((path.as_path(), "--metrics")); + } + crate::commands::common::reject_output_collisions(&outputs)?; + // Validate the input exists (stdin paths are exempt). self.io.validate()?; validate_file_exists(&self.reference, "Reference FASTA")?; diff --git a/src/lib/commands/codec.rs b/src/lib/commands/codec.rs index b91f5cb41..1180a3319 100644 --- a/src/lib/commands/codec.rs +++ b/src/lib/commands/codec.rs @@ -25,11 +25,12 @@ use fgumi_bam_io::{ create_bam_reader_for_pipeline_with_opts, create_bam_writer, create_optional_bam_writer, create_raw_bam_reader_with_opts, }; +use std::path::Path; use super::common::{ AllowUnmappedOptions, BamIoOptions, CompressionOptions, ConsensusCallingOptions, QueueMemoryOptions, ReadGroupOptions, RejectsOptions, SchedulerOptions, StatsOptions, - ThreadingOptions, build_pipeline_config, reject_colliding_outputs, serialize_raw_bam_records, + ThreadingOptions, build_pipeline_config, reject_output_collisions, serialize_raw_bam_records, }; use crate::consensus::codec_caller::{ CodecConsensusCaller, CodecConsensusError, CodecConsensusOptions, CodecConsensusStats, @@ -318,7 +319,14 @@ impl Command for Codec { // Validate the input exists (stdin paths are exempt — the reader // streams them in a single pass). self.io.validate()?; - reject_colliding_outputs(&self.io.output, self.rejects_opts.rejects.as_ref(), "--rejects")?; + let mut outputs: Vec<(&Path, &str)> = vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.rejects_opts.rejects { + outputs.push((path.as_path(), "--rejects")); + } + if let Some(path) = &self.stats_opts.stats { + outputs.push((path.as_path(), "--stats")); + } + reject_output_collisions(&outputs)?; let timer = OperationTimer::new("Calling CODEC consensus"); diff --git a/src/lib/commands/common.rs b/src/lib/commands/common.rs index 9597cd34e..3ec74d4d4 100644 --- a/src/lib/commands/common.rs +++ b/src/lib/commands/common.rs @@ -3,6 +3,8 @@ //! This module provides shared argument structures that can be composed into //! command structs using `#[command(flatten)]`. +use std::collections::HashMap; +use std::ffi::OsString; use std::path::{Path, PathBuf}; #[cfg(feature = "simplex")] use std::sync::Arc; @@ -17,7 +19,6 @@ use crate::unified_pipeline::{ use crate::validation::validate_input_exists; use bytesize::ByteSize; use clap::Args; -#[cfg(feature = "simplex")] use fgumi_bam_io::is_stdout_path; use fgumi_consensus::methylation::RefBaseProvider; use fgumi_umi::IndexThreshold; @@ -584,12 +585,17 @@ impl RejectsOptions { } } -/// Refuse a secondary output that resolves to the same destination as `--output`. +/// Refuse a command whose output paths do not all resolve to distinct +/// destinations. /// -/// A command's primary and secondary writers are both opened up front and -/// written from the same loop, so pointing them at one destination does not make -/// them take turns — it interleaves two BAMs. How that fails depends on the -/// destination, and both ways are silent: +/// Given every output a command will open — `--output`, `--rejects`, `--stats`, +/// `--metrics` (including a prefix's expanded files), histograms — as +/// `(path, flag-label)` pairs, this rejects any two that resolve to the same +/// destination. A command's writers are opened up front and written +/// independently, so pointing two at one destination does not make them take +/// turns — it interleaves or overwrites, corrupting the output (a metrics/stats +/// path equal to `--output` truncates a finished BAM to a TSV). How that fails +/// depends on the destination, and both ways are silent: /// /// - **stdout.** `-` and `/dev/stdout` both resolve to fd 1, where the two /// writers share one file description. Every byte lands, but the stream @@ -613,8 +619,7 @@ impl RejectsOptions { /// hard links, or one name in two cases on a case-insensitive filesystem. A true /// `dev`+`ino` comparison needs both files to exist, and by the time they do both /// `File::create` calls have already truncated; closing that gap means moving the -/// check into the writer layer. See #715, which also covers `--stats`/`--metrics` -/// sharing a path with `--output`. +/// check into the writer layer — a follow-up to #715 (item 1) tracked separately. /// /// The null device is the one exempt destination: it discards every byte, so two /// writers on it cannot corrupt each other and `-o /dev/null --rejects /dev/null` @@ -624,43 +629,67 @@ impl RejectsOptions { /// two block sequences, and two EOF markers into a stream no reader can parse. /// That is the same failure as `-o - --rejects -`, so it is rejected the same way. /// -/// `secondary_flag` names the offending option in the error, since a command may -/// have more than one secondary output. +/// Each target carries the flag label that named it, so the error can point at +/// the offending options. Absent (`None`) outputs are simply left out of the +/// slice by the caller. /// /// # Errors /// -/// Returns an error if `output` and `secondary` both name stdout, or both resolve -/// to the same destination and that destination is not the null device. -pub fn reject_colliding_outputs( - output: &Path, - secondary: Option<&PathBuf>, - secondary_flag: &str, -) -> anyhow::Result<()> { - let Some(secondary) = secondary else { return Ok(()) }; - - if is_stdout_path(output) || is_stdout_path(secondary) { - anyhow::ensure!( - !(is_stdout_path(output) && is_stdout_path(secondary)), - "--output and {secondary_flag} cannot both write to stdout: the two BAM streams \ - would interleave into one unreadable stream; give {secondary_flag} a path" +/// Returns an error if more than one target names stdout, or if two targets +/// resolve to the same non-null destination. +pub fn reject_output_collisions(targets: &[(&Path, &str)]) -> anyhow::Result<()> { + // stdout is a single shared stream, so at most one target may name it — + // regardless of spelling (`-`, `/dev/stdout`). Two writers on it would + // interleave two headers and two block sequences into one unreadable stream. + let stdout_flags: Vec<&str> = + targets.iter().filter(|(path, _)| is_stdout_path(path)).map(|&(_, flag)| flag).collect(); + if stdout_flags.len() > 1 { + anyhow::bail!( + "{} cannot all write to stdout: the streams would interleave into one unreadable \ + output; give all but one an explicit path", + join_flags(&stdout_flags) ); - return Ok(()); } - if is_null_device(output) && is_null_device(secondary) { - return Ok(()); + // Group the remaining file targets by the identity that can be established + // before the file exists. The null device is the one destination multiple + // writers may share (it discards every byte), so it is skipped rather than + // grouped. The first flag to claim an identity wins the error's "prior" slot. + let mut seen: HashMap<(PathBuf, Option), &str> = HashMap::new(); + for &(path, flag) in targets { + if is_stdout_path(path) || is_null_device(path) { + continue; + } + let identity = resolve_output_identity_owned(path); + if let Some(&prior_flag) = seen.get(&identity) { + anyhow::bail!( + "{prior_flag} and {flag} both write to {}: two outputs on one destination would \ + overwrite each other byte for byte, or interleave on a shared stream, and either \ + way nothing can read the result; give one a different path", + path.display() + ); + } + seen.insert(identity, flag); } + Ok(()) +} - if resolve_output_identity(output) == resolve_output_identity(secondary) { - anyhow::bail!( - "--output and {secondary_flag} both write to {}: the two BAM streams would \ - interleave on a shared stream, or overwrite each other byte for byte on a \ - regular file, and either way nothing can read the result; give \ - {secondary_flag} a different path", - secondary.display() - ); +/// Join flag labels into an English list for an error message: `"--output and +/// --stats"`, or `"--output, --stats, and --metrics"`. +fn join_flags(flags: &[&str]) -> String { + match flags { + [] => String::new(), + [only] => (*only).to_string(), + [a, b] => format!("{a} and {b}"), + [rest @ .., last] => format!("{}, and {last}", rest.join(", ")), } - Ok(()) +} + +/// [`resolve_output_identity`] with the file name owned, so the identity can key +/// a map across targets rather than borrowing each path. +fn resolve_output_identity_owned(path: &Path) -> (PathBuf, Option) { + let (parent, name) = resolve_output_identity(path); + (parent, name.map(std::ffi::OsStr::to_os_string)) } /// Whether `path` is the null device, the one destination two writers can share. @@ -1669,7 +1698,7 @@ mod tests { assert_eq!(io.effective_check_crc(), expected); } - /// `reject_colliding_outputs` compares destinations, not the strings naming + /// `reject_output_collisions` compares destinations, not the strings naming /// them: `out.bam` and `./out.bam` are one file under two names, and two /// writers on one file overwrite each other byte for byte. /// @@ -1686,7 +1715,7 @@ mod tests { #[case::distinct_names("out.bam", "rejects.bam", false)] #[case::identical_under_a_missing_parent("missing/out.bam", "missing/out.bam", true)] #[case::distinct_under_a_missing_parent("missing/out.bam", "missing/rejects.bam", false)] - fn reject_colliding_outputs_compares_resolved_files( + fn reject_output_collisions_compares_resolved_files( #[case] output: &str, #[case] secondary: &str, #[case] collides: bool, @@ -1696,7 +1725,10 @@ mod tests { let output = dir.path().join(output); let secondary = dir.path().join(secondary); - let result = reject_colliding_outputs(&output, Some(&secondary), "--rejects"); + let result = reject_output_collisions(&[ + (output.as_path(), "--output"), + (secondary.as_path(), "--rejects"), + ]); assert_eq!( result.is_err(), @@ -1731,14 +1763,17 @@ mod tests { #[case::a_file_then_stdout("out.bam", "-", false)] #[case::both_dev_null("/dev/null", "/dev/null", false)] #[case::dev_null_then_a_file("/dev/null", "rejects.bam", false)] - fn reject_colliding_outputs_handles_stdout_and_the_null_device( + fn reject_output_collisions_handles_stdout_and_the_null_device( #[case] output: &str, #[case] secondary: &str, #[case] collides: bool, ) { let secondary = PathBuf::from(secondary); - let result = reject_colliding_outputs(Path::new(output), Some(&secondary), "--rejects"); + let result = reject_output_collisions(&[ + (Path::new(output), "--output"), + (secondary.as_path(), "--rejects"), + ]); assert_eq!( result.is_err(), @@ -1764,8 +1799,8 @@ mod tests { #[case::stdout("-")] #[case::dev_stdout("/dev/stdout")] #[case::a_file("out.bam")] - fn reject_colliding_outputs_allows_an_absent_secondary(#[case] output: &str) { - let result = reject_colliding_outputs(Path::new(output), None, "--rejects"); + fn reject_output_collisions_allows_an_absent_secondary(#[case] output: &str) { + let result = reject_output_collisions(&[(Path::new(output), "--output")]); assert!(result.is_ok(), "`-o {output}` with no --rejects must be allowed: {result:?}"); } @@ -1781,7 +1816,7 @@ mod tests { #[rstest] #[case::one_fifo_named_twice("primary", "primary", true)] #[case::two_distinct_fifos("primary", "secondary", false)] - fn reject_colliding_outputs_rejects_a_shared_fifo( + fn reject_output_collisions_rejects_a_shared_fifo( #[case] output: &str, #[case] secondary: &str, #[case] collides: bool, @@ -1794,7 +1829,10 @@ mod tests { make_fifo(&secondary); } - let result = reject_colliding_outputs(&output, Some(&secondary), "--rejects"); + let result = reject_output_collisions(&[ + (output.as_path(), "--output"), + (secondary.as_path(), "--rejects"), + ]); assert_eq!( result.is_err(), @@ -1812,6 +1850,46 @@ mod tests { } } + /// The guard checks every pair among a command's outputs, not just + /// output-vs-rejects, and a distinct set of paths is always allowed. + #[rstest] + #[case::stats_hits_output(&[("out.bam", "--output"), ("r.bam", "--rejects"), ("out.bam", "--stats")], true)] + #[case::metrics_hits_rejects(&[("out.bam", "--output"), ("m.tsv", "--rejects"), ("m.tsv", "--metrics")], true)] + #[case::all_distinct(&[("out.bam", "--output"), ("r.bam", "--rejects"), ("s.tsv", "--stats")], false)] + #[case::single_output(&[("out.bam", "--output")], false)] + #[case::no_outputs(&[], false)] + fn reject_output_collisions_checks_every_pair( + #[case] specs: &[(&str, &str)], + #[case] collides: bool, + ) { + let dir = tempfile::tempdir().expect("create temp dir"); + let owned: Vec<(PathBuf, &str)> = + specs.iter().map(|(p, flag)| (dir.path().join(p), *flag)).collect(); + let targets: Vec<(&Path, &str)> = + owned.iter().map(|(p, flag)| (p.as_path(), *flag)).collect(); + + let result = reject_output_collisions(&targets); + assert_eq!(result.is_err(), collides, "{specs:?}: got {result:?}"); + } + + /// stdout is one shared stream, so at most one output may name it — and the + /// error names every flag that tried to. + #[test] + fn reject_output_collisions_rejects_more_than_one_stdout() { + let result = reject_output_collisions(&[ + (Path::new("-"), "--output"), + (Path::new("out.bam"), "--rejects"), + (Path::new("/dev/stdout"), "--stats"), + ]); + let message = result.expect_err("two stdout writers must be rejected").to_string(); + assert!( + message.contains("stdout") + && message.contains("--output") + && message.contains("--stats"), + "the error must name stdout and both offending flags, got: {message}" + ); + } + /// Create a FIFO at `path`. /// /// The standard library has no `mkfifo(3)` binding, and the crate's `nix` diff --git a/src/lib/commands/correct.rs b/src/lib/commands/correct.rs index 2425d4934..abad8cae5 100644 --- a/src/lib/commands/correct.rs +++ b/src/lib/commands/correct.rs @@ -21,7 +21,7 @@ //! # Example //! //! ```no_run -//! use std::path::PathBuf; +//! use std::path::{Path, PathBuf}; //! use fgumi_lib::commands::correct::{CorrectUmis, Target}; //! use fgumi_lib::commands::command::Command; //! use fgumi_lib::commands::common::{ @@ -85,6 +85,7 @@ use noodles::sam::Header; use noodles::sam::alignment::record::data::field::Tag; use std::io; use std::num::NonZero; +use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -92,7 +93,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use crate::commands::command::Command; use crate::commands::common::{ BamIoOptions, CompressionOptions, QueueMemoryOptions, RejectsOptions, SchedulerOptions, - ThreadingOptions, build_pipeline_config, reject_colliding_outputs, serialize_raw_bam_records, + ThreadingOptions, build_pipeline_config, reject_output_collisions, serialize_raw_bam_records, }; /// Which SAM tag `correct` operates on. @@ -471,7 +472,14 @@ impl Command for CorrectUmis { /// ``` fn execute(&self, command_line: &str) -> Result<()> { self.validate()?; - reject_colliding_outputs(&self.io.output, self.rejects_opts.rejects.as_ref(), "--rejects")?; + let mut outputs: Vec<(&Path, &str)> = vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.rejects_opts.rejects { + outputs.push((path.as_path(), "--rejects")); + } + if let Some(path) = &self.metrics { + outputs.push((path.as_path(), "--metrics")); + } + reject_output_collisions(&outputs)?; let timer = OperationTimer::new("Correcting UMIs"); diff --git a/src/lib/commands/dedup.rs b/src/lib/commands/dedup.rs index bd5f65ec9..033692bd0 100644 --- a/src/lib/commands/dedup.rs +++ b/src/lib/commands/dedup.rs @@ -1151,6 +1151,20 @@ pub struct MarkDuplicates { impl Command for MarkDuplicates { fn execute(&self, command_line: &str) -> Result<()> { + // Reject two outputs resolving to one destination before any writer opens. + let mut outputs: Vec<(&std::path::Path, &str)> = + vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.metrics { + outputs.push((path.as_path(), "--metrics")); + } + if let Some(path) = &self.family_size_histogram { + outputs.push((path.as_path(), "--family-size-histogram")); + } + if let Some(path) = &self.duplication_ladder { + outputs.push((path.as_path(), "--duplication-ladder")); + } + crate::commands::common::reject_output_collisions(&outputs)?; + // Validate strategy/min-umi-length combination if self.min_umi_length.is_some() && matches!(self.strategy, Strategy::Paired) { bail!("Paired strategy cannot be used with --min-umi-length"); diff --git a/src/lib/commands/downsample.rs b/src/lib/commands/downsample.rs index 1389084e4..f1635ceb8 100644 --- a/src/lib/commands/downsample.rs +++ b/src/lib/commands/downsample.rs @@ -20,10 +20,10 @@ use rand::rngs::StdRng; use std::collections::{BTreeMap, HashSet}; use std::fs::File; use std::io::Write; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use crate::commands::command::Command; -use crate::commands::common::{BamIoOptions, CompressionOptions, reject_colliding_outputs}; +use crate::commands::common::{BamIoOptions, CompressionOptions, reject_output_collisions}; /// Downsample a BAM file by UMI family using streaming. /// @@ -130,7 +130,17 @@ impl Command for Downsample { // is exempt from the file-existence check, matching every other streaming // command; the BAM reader already handles stdin. self.io.validate()?; - reject_colliding_outputs(&self.io.output, self.rejects.as_ref(), "--rejects")?; + let mut outputs: Vec<(&Path, &str)> = vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.rejects { + outputs.push((path.as_path(), "--rejects")); + } + if let Some(path) = &self.histogram_kept { + outputs.push((path.as_path(), "--histogram-kept")); + } + if let Some(path) = &self.histogram_rejected { + outputs.push((path.as_path(), "--histogram-rejected")); + } + reject_output_collisions(&outputs)?; // Validate fraction Self::validate_fraction(self.fraction)?; diff --git a/src/lib/commands/duplex.rs b/src/lib/commands/duplex.rs index 3803a9151..3e9858371 100644 --- a/src/lib/commands/duplex.rs +++ b/src/lib/commands/duplex.rs @@ -20,12 +20,13 @@ use fgumi_bam_io::{ create_bam_reader_for_pipeline_with_opts, create_bam_writer, create_optional_bam_writer, create_raw_bam_reader_with_opts, }; +use std::path::Path; use super::common::{ AllowUnmappedOptions, BamIoOptions, CompressionOptions, ConsensusCallingOptions, OverlappingConsensusOptions, QueueMemoryOptions, ReadGroupOptions, RejectsOptions, SchedulerOptions, StatsOptions, ThreadingOptions, build_pipeline_config, - consensus_pregroup_keep_flags, consensus_pregroup_keep_raw, reject_colliding_outputs, + consensus_pregroup_keep_flags, consensus_pregroup_keep_raw, reject_output_collisions, serialize_raw_bam_records, }; use crate::commands::consensus_runner::{ @@ -274,7 +275,7 @@ impl Command for Duplex { /// # OverlappingConsensusOptions, QueueMemoryOptions, ReadGroupOptions, RejectsOptions, /// # SchedulerOptions, StatsOptions, ThreadingOptions, /// # }; - /// # use std::path::PathBuf; + /// # use std::path::{Path, PathBuf}; /// let duplex = Duplex { /// io: BamIoOptions { /// input: PathBuf::from("grouped.bam"), @@ -315,7 +316,14 @@ impl Command for Duplex { // Validate the input exists (stdin paths are exempt — the reader // streams them in a single pass). self.io.validate()?; - reject_colliding_outputs(&self.io.output, self.rejects_opts.rejects.as_ref(), "--rejects")?; + let mut outputs: Vec<(&Path, &str)> = vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.rejects_opts.rejects { + outputs.push((path.as_path(), "--rejects")); + } + if let Some(path) = &self.stats_opts.stats { + outputs.push((path.as_path(), "--stats")); + } + reject_output_collisions(&outputs)?; // Validate consensus arguments (e.g. error rates must be > 0). self.validate()?; diff --git a/src/lib/commands/filter.rs b/src/lib/commands/filter.rs index dbc62c2a0..7c2759c48 100644 --- a/src/lib/commands/filter.rs +++ b/src/lib/commands/filter.rs @@ -40,7 +40,7 @@ use fgumi_raw_bam::{RawRecord, RawRecordView}; use log::info; use noodles::sam::Header; use std::io; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Instant; @@ -48,7 +48,7 @@ use std::time::Instant; use crate::commands::command::Command; use crate::commands::common::{ BamIoOptions, CompressionOptions, QueueMemoryOptions, SchedulerOptions, ThreadingOptions, - build_pipeline_config, reject_colliding_outputs, serialize_raw_bam_records, + build_pipeline_config, reject_output_collisions, serialize_raw_bam_records, }; /// Filters and masks consensus reads based on various quality metrics. @@ -292,7 +292,14 @@ impl Command for Filter { // Validate the input exists (stdin paths are exempt — the reader // streams them in a single pass). self.io.validate()?; - reject_colliding_outputs(&self.io.output, self.rejects.as_ref(), "--rejects")?; + let mut outputs: Vec<(&Path, &str)> = vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.rejects { + outputs.push((path.as_path(), "--rejects")); + } + if let Some(path) = &self.stats { + outputs.push((path.as_path(), "--stats")); + } + reject_output_collisions(&outputs)?; if let Some(ref reference) = self.reference { validate_file_exists(reference, "Reference FASTA")?; diff --git a/src/lib/commands/group.rs b/src/lib/commands/group.rs index d131b09ca..d1f439991 100644 --- a/src/lib/commands/group.rs +++ b/src/lib/commands/group.rs @@ -767,6 +767,27 @@ impl Command for GroupReadsByUmi { /// Execute the tool using the 7-step unified pipeline. #[allow(clippy::too_many_lines)] fn execute(&self, command_line: &str) -> Result<()> { + // Reject two outputs resolving to one destination before any writer opens + // (e.g. a `--metrics PREFIX` file, or `-f`/`-g`, landing on `--output`). + let metrics_files: Vec = self.metrics.as_ref().map_or_else(Vec::new, |prefix| { + vec![ + with_extension(prefix, "family_sizes.txt"), + with_extension(prefix, "grouping_metrics.txt"), + with_extension(prefix, "position_group_sizes.txt"), + ] + }); + let mut outputs: Vec<(&Path, &str)> = vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.family_size_histogram { + outputs.push((path.as_path(), "--family-size-histogram")); + } + if let Some(path) = &self.grouping_metrics { + outputs.push((path.as_path(), "--grouping-metrics")); + } + for path in &metrics_files { + outputs.push((path.as_path(), "--metrics")); + } + crate::commands::common::reject_output_collisions(&outputs)?; + // Validate inputs if self.min_umi_length.is_some() && matches!(self.strategy, Strategy::Paired) { bail!("Paired strategy cannot be used with --min-umi-length"); diff --git a/src/lib/commands/simplex.rs b/src/lib/commands/simplex.rs index 2af31af29..72bc995e2 100644 --- a/src/lib/commands/simplex.rs +++ b/src/lib/commands/simplex.rs @@ -35,6 +35,7 @@ use fgumi_bam_io::{ create_raw_bam_reader_with_opts, }; use fgumi_raw_bam::{RawRecord, RawRecordView}; +use std::path::Path; // RejectionTracker now used via ConsensusStatsOps trait in consensus_runner use crate::per_thread_accumulator::PerThreadAccumulator; use crate::sam::SamTag; @@ -52,7 +53,7 @@ use crate::commands::common::{ AllowUnmappedOptions, BamIoOptions, CompressionOptions, ConsensusCallingOptions, OverlappingConsensusOptions, QueueMemoryOptions, ReadGroupOptions, RejectsOptions, SchedulerOptions, StatsOptions, ThreadingOptions, build_pipeline_config, - consensus_pregroup_keep_flags, consensus_pregroup_keep_raw, reject_colliding_outputs, + consensus_pregroup_keep_flags, consensus_pregroup_keep_raw, reject_output_collisions, serialize_raw_bam_records, }; use crate::commands::consensus_runner::{ @@ -253,7 +254,14 @@ impl Command for Simplex { // Validate inputs self.io.validate()?; - reject_colliding_outputs(&self.io.output, self.rejects_opts.rejects.as_ref(), "--rejects")?; + let mut outputs: Vec<(&Path, &str)> = vec![(self.io.output.as_path(), "--output")]; + if let Some(path) = &self.rejects_opts.rejects { + outputs.push((path.as_path(), "--rejects")); + } + if let Some(path) = &self.stats_opts.stats { + outputs.push((path.as_path(), "--stats")); + } + reject_output_collisions(&outputs)?; self.validate_read_bounds()?; diff --git a/src/lib/unified_pipeline/bam.rs b/src/lib/unified_pipeline/bam.rs index 428ff7335..865cd30fa 100644 --- a/src/lib/unified_pipeline/bam.rs +++ b/src/lib/unified_pipeline/bam.rs @@ -8,7 +8,6 @@ use noodles::bam::{self}; use noodles::sam::{Header, alignment::RecordBuf}; use parking_lot::Mutex; use std::collections::VecDeque; -use std::fs::File; use std::io::{self, BufReader, BufWriter, Read, Write}; use std::path::Path; use std::sync::Arc; @@ -21,7 +20,6 @@ use crate::bgzf_writer::InlineBgzfCompressor; use crate::sam::SamTag; use fgumi_bam_io::ProgressTracker; use fgumi_bam_io::ReorderBuffer; -use fgumi_bam_io::is_stdout_path; use noodles::sam::alignment::record::data::field::Tag; use super::base::{ @@ -4202,14 +4200,17 @@ impl BamPipelineConfig { } /// Open an output writer for pipeline use, supporting stdout via "-" or "/dev/stdout". +/// +/// Delegates to [`fgumi_bam_io::open_output_writer`], the single place the `-` +/// convention is honoured for BAM output. This previously re-implemented the +/// dispatch and returned a `LineWriter`-backed [`std::io::stdout`], which tears +/// every BGZF flush at each `0x0a`; `open_output_writer` returns a block-buffered +/// stdout handle instead. fn open_pipeline_output(output_path: &Path) -> io::Result> { - if is_stdout_path(output_path) { - Ok(Box::new(std::io::stdout())) - } else { - let file = File::create(output_path) - .map_err(|e| io::Error::new(e.kind(), format!("Failed to create output: {e}")))?; - Ok(Box::new(file)) - } + fgumi_bam_io::open_output_writer(output_path).map_err(|e| { + let kind = e.downcast_ref::().map_or(io::ErrorKind::Other, io::Error::kind); + io::Error::new(kind, format!("{e:#}")) + }) } /// Convert an input-open failure into an `io::Error`, preserving its kind. diff --git a/tests/integration/test_streaming_output.rs b/tests/integration/test_streaming_output.rs index 3afadedb4..33bc0571e 100644 --- a/tests/integration/test_streaming_output.rs +++ b/tests/integration/test_streaming_output.rs @@ -259,13 +259,19 @@ fn sort_rejects_write_index_with_stdout(#[case] spelling: &str) { /// regardless of the tags each would otherwise require. The reference and UMI /// list are supplied because the two commands that take them reject a missing /// file before the guard under test would ever run. -fn colliding_output_args(dir: &Path, command: &[&str], output: &str, rejects: &str) -> Vec { +fn colliding_output_args( + dir: &Path, + command: &[&str], + secondary_flag: &str, + output: &str, + secondary: &str, +) -> Vec { let input = dir.join("input.bam"); write_test_bam(&input, 4); let mut args: Vec = command.iter().map(|a| (*a).to_string()).collect(); args.extend( - ["-i", input.to_str().unwrap(), "-o", output, "--rejects", rejects] + ["-i", input.to_str().unwrap(), "-o", output, secondary_flag, secondary] .map(ToString::to_string), ); if command[0] == "correct" { @@ -273,7 +279,7 @@ fn colliding_output_args(dir: &Path, command: &[&str], output: &str, rejects: &s std::fs::write(&umis, "ACGTACGT\n").expect("write UMI list"); args.extend(["-U".to_string(), umis.display().to_string()]); } - if command[0] == "filter" { + if command[0] == "filter" || command[0] == "clip" { let reference = create_test_reference(dir); args.extend(["-r".to_string(), reference.display().to_string()]); } @@ -305,7 +311,7 @@ fn rejects_output_and_rejects_both_on_stdout( #[values("-", "/dev/stdout")] rejects: &str, ) { let dir = TempDir::new().expect("create temp dir"); - let args = colliding_output_args(dir.path(), command, output, rejects); + let args = colliding_output_args(dir.path(), command, "--rejects", output, rejects); let run = run_in(dir.path(), &args); let label = format!("{} -o {output} --rejects {rejects}", command[0]); @@ -360,7 +366,7 @@ fn rejects_output_and_rejects_on_the_same_file( ) { let dir = TempDir::new().expect("create temp dir"); std::fs::create_dir_all(dir.path().join("sub")).expect("create subdirectory"); - let args = colliding_output_args(dir.path(), command, "out.bam", rejects); + let args = colliding_output_args(dir.path(), command, "--rejects", "out.bam", rejects); let run = run_in(dir.path(), &args); let label = format!("{} -o out.bam --rejects {rejects}", command[0]); @@ -380,6 +386,39 @@ fn rejects_output_and_rejects_on_the_same_file( ); } +/// The collision guard covers a command's *other* secondary outputs, not just +/// `--rejects` (#715): a `--stats`/`--metrics`/histogram path equal to `--output` +/// used to write a correct BAM and then truncate it to a TSV. Each command here +/// carries a different secondary output; pointing it at `--output` must be +/// rejected before either writer opens, so the BAM is never created. +#[rstest] +#[case::simplex_stats(&["simplex", "--min-reads", "1"], "--stats")] +#[case::duplex_stats(&["duplex", "--min-reads", "1"], "--stats")] +#[case::codec_stats(&["codec", "--min-reads", "1", "--min-duplex-length", "1"], "--stats")] +#[case::filter_stats(&["filter", "--min-reads", "1"], "--stats")] +#[case::correct_metrics(&["correct", "-d", "1"], "--metrics")] +#[case::dedup_metrics(&["dedup"], "--metrics")] +#[case::clip_metrics(&["clip"], "--metrics")] +#[case::group_histogram(&["group", "--strategy", "adjacency"], "--family-size-histogram")] +fn rejects_secondary_output_on_the_output_file(#[case] command: &[&str], #[case] flag: &str) { + let dir = TempDir::new().expect("create temp dir"); + let args = colliding_output_args(dir.path(), command, flag, "out.bam", "out.bam"); + + let run = run_in(dir.path(), &args); + let label = format!("{} -o out.bam {flag} out.bam", command[0]); + + assert!(!run.status.success(), "{label}: a secondary output on `--output` must be rejected"); + let stderr = String::from_utf8_lossy(&run.stderr); + assert!( + stderr.contains(flag) && stderr.contains("out.bam"), + "{label}: the error must name the colliding option and the path, got: {stderr}" + ); + assert!( + !dir.path().join("out.bam").exists(), + "{label}: the rejected run still created (and would have truncated) the output BAM" + ); +} + /// `extract` reaches stdout whether or not `--threads` makes it parallel. /// /// The two cases select different writers — a plain raw-BAM writer when the flag