diff --git a/CLAUDE.md b/CLAUDE.md index ce731d655..5d54146b3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -137,6 +137,17 @@ The following external FFI calls are approved because they back core infrastruct - **`mach2`** (`crates/fgumi-sort/src/memory_probe.rs`, macOS only) — `task_info(TASK_VM_INFO)` is the only way to read `phys_footprint` (the RSS metric mimalloc reports accurately). Isolated to the same sub-module as the mimalloc FFI. +- **`libc::umask`** (`crates/fgumi-sort/src/external.rs`, Unix only) — the merge command writes + its output through an atomic temp+persist (`MergeOutputTarget`) so a rejected/failed merge + never leaves a partial file. A `NamedTempFile` is created `0600`, so the persisted output must + be re-stamped with the mode a plain `File::create` would have produced. For a new file that is + `0o666 & !umask`, and reading the process `umask` requires the libc binding (POSIX `umask(2)` + only *sets* the mask, returning the previous value, so `process_umask` sets it to `0` and + restores it in the next call). The call is isolated to a single `#[allow(unsafe_code)]` site + with a `// SAFETY:` note; it has no preconditions and cannot fail. The read-modify-restore is + not atomic, so a process-global `Mutex` (`UMASK_LOCK`) serializes concurrent probes — required + because `fgumi_lib` is a library and callers may run merges concurrently in one process; + without the lock two interleaved probes could leave the process mask permanently `0`. ### Approved hot-path unsafe (sort engine) diff --git a/Cargo.lock b/Cargo.lock index dd396e4f7..cc69da93f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -827,6 +827,7 @@ dependencies = [ "fgumi-raw-bam", "fgumi-sam", "fs4", + "libc", "libdeflater", "libmimalloc-sys", "log", diff --git a/crates/fgumi-sort/Cargo.toml b/crates/fgumi-sort/Cargo.toml index a6964db0d..ee62c6560 100644 --- a/crates/fgumi-sort/Cargo.toml +++ b/crates/fgumi-sort/Cargo.toml @@ -37,6 +37,9 @@ fs4 = { version = "1.1", default-features = false, features = ["sync"] } log = "0" zstd = "0.13" +[target.'cfg(unix)'.dependencies] +libc = "0.2" + [target.'cfg(target_os = "macos")'.dependencies] mach2 = "0.6" diff --git a/crates/fgumi-sort/src/external.rs b/crates/fgumi-sort/src/external.rs index 2392fc262..560efd286 100644 --- a/crates/fgumi-sort/src/external.rs +++ b/crates/fgumi-sort/src/external.rs @@ -40,7 +40,7 @@ use anyhow::{Context as _, Result}; use crossbeam_channel::{Receiver, Sender, bounded}; use fgumi_bam_io::ProgressTracker; use fgumi_bam_io::create_raw_bam_reader; -use fgumi_bam_io::{ChainedReader, TeeReader, is_stdin_path}; +use fgumi_bam_io::{ChainedReader, TeeReader, is_stdin_path, is_stdout_path}; use fgumi_raw_bam::SamTag; use log::{debug, info}; use noodles::sam::Header; @@ -228,6 +228,199 @@ pub fn cb_hasher() -> ahash::RandomState { ) } +/// Where a merge writes its output. +/// +/// For a regular file, the merge writes to an exclusive, uniquely-named +/// [`tempfile::NamedTempFile`] in the output's directory and atomically +/// [`persist`](MergeOutputTarget::persist)s it into place on success. Because the +/// temp is RAII-owned, *any* error path (a sort-order violation, a write failure, +/// `finish` failing) drops it and removes the file — a rejected or failed merge +/// never leaves a partial/corrupt output or a stray temp behind. The +/// same-directory temp keeps the rename atomic (same filesystem), and exclusive +/// creation avoids clobbering a concurrent run or a stale leftover. +/// +/// Stdout can't be renamed, so it is written directly (a mid-merge failure there +/// leaves whatever was already streamed, which is unavoidable for a pipe). +/// +/// A `NamedTempFile` is created `0600`, so persisting it verbatim would silently +/// make merged output owner-only. To preserve the semantics of the pre-atomic-temp +/// direct write (`File::create`), the `Temp` variant carries the mode the final +/// file must end up with (see [`target_file_mode`]) and applies it before the +/// rename. +enum MergeOutputTarget { + /// A regular-file output, staged in a same-directory temp. `dest` is the + /// resolved final path the temp is atomically renamed onto (a symlinked + /// output is followed to its real target, so the temp is staged next to — + /// and renamed onto — the linked file rather than the link itself). `mode` + /// is the Unix mode to stamp onto the temp before persisting (`None` on + /// non-Unix, where file modes are managed by the platform). + Temp { + temp: tempfile::NamedTempFile, + dest: PathBuf, + mode: Option, + }, + Stdout(PathBuf), +} + +impl MergeOutputTarget { + fn create(output: &Path) -> Result { + if is_stdout_path(output) { + return Ok(Self::Stdout(output.to_path_buf())); + } + // Follow a symlinked destination to the real file it points at, so the + // atomic rename updates the linked file in place (matching `File::create`, + // which follows symlinks) instead of replacing the symlink with a regular + // file. The temp is then staged next to — and renamed onto — that real + // target, keeping the rename same-directory (hence same-filesystem/atomic). + let dest = resolve_symlink_output(output)?; + // The temp+rename `persist` replaces `dest` with a regular file, so an + // existing FIFO, device, socket, or directory would be silently clobbered + // (a plain `File::create` would instead write *through* a FIFO/device). + // Only a missing path or an existing regular file can be staged this way; + // reject anything else with a clear error rather than corrupt it. + if let Ok(metadata) = std::fs::metadata(&dest) { + anyhow::ensure!( + metadata.file_type().is_file(), + "merge output must be a regular file or stdout: {}", + dest.display() + ); + } + let dir = dest + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .map_or_else(|| PathBuf::from("."), Path::to_path_buf); + let temp = tempfile::Builder::new() + .prefix(".fgumi-merge-") + .suffix(".tmp") + .tempfile_in(&dir) + .with_context(|| format!("failed to create a merge temp file in {}", dir.display()))?; + // Resolve the final mode now (before any BGZF writer threads spawn) so the + // umask read below is single-threaded and cannot race a concurrent create. + let mode = target_file_mode(&dest); + Ok(Self::Temp { temp, dest, mode }) + } + + /// The path the merged BAM is written to. + fn path(&self) -> &Path { + match self { + Self::Temp { temp, .. } => temp.path(), + Self::Stdout(path) => path, + } + } + + /// Atomically moves the finished temp onto its resolved destination (a no-op + /// for stdout), first stamping the resolved mode so the output is not left + /// temp-private (`0600`). Consumes `self`, disarming the RAII auto-remove. + fn persist(self) -> Result<()> { + if let Self::Temp { temp, dest, mode } = self { + if let Some(mode) = mode { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + temp.as_file() + .set_permissions(std::fs::Permissions::from_mode(mode)) + .with_context(|| { + format!("failed to set mode on merge temp for {}", dest.display()) + })?; + } + #[cfg(not(unix))] + let _ = mode; + } + temp.persist(&dest).map_err(|e| e.error).with_context(|| { + format!("failed to finalize merged output at {}", dest.display()) + })?; + } + Ok(()) + } +} + +/// Resolves a symlinked output path to the real file the atomic rename must +/// target, so a `latest.bam -> run.bam` symlink is followed and updated in +/// place (matching `File::create`, which follows symlinks) instead of being +/// replaced by a regular file. Non-symlink paths — including ones that do not +/// exist yet — are returned unchanged. Bails on a symlink cycle (or an +/// unreasonably long chain) rather than looping forever. +fn resolve_symlink_output(output: &Path) -> Result { + // POSIX ELOOP triggers around 40 levels; cap here at the same bound so a + // cyclic/pathological chain fails fast instead of spinning. + const MAX_LINKS: usize = 40; + let mut path = output.to_path_buf(); + for _ in 0..MAX_LINKS { + // `is_symlink` returns false for a nonexistent path, so a brand-new + // output (or its final component) short-circuits here unchanged. + if !path.is_symlink() { + return Ok(path); + } + let target = std::fs::read_link(&path) + .with_context(|| format!("failed to read symlink output {}", path.display()))?; + // Relative link targets resolve against the link's own directory. + path = if target.is_absolute() { + target + } else { + path.parent().map_or_else(|| target.clone(), |parent| parent.join(&target)) + }; + } + anyhow::bail!("too many levels of symbolic links while resolving output {}", output.display()) +} + +/// The mode the merged output file must carry, matching the pre-atomic-temp write +/// path (`File::create`): an existing destination keeps its current mode; a new +/// file gets `0o666 & !umask`. Returns `None` on non-Unix, where the temp's +/// platform-managed permissions are left as-is. +#[cfg(unix)] +// The `not(unix)` sibling returns `None`, so the `Option` is load-bearing there. +#[allow(clippy::unnecessary_wraps, reason = "non-unix sibling returns None")] +fn target_file_mode(output: &Path) -> Option { + use std::os::unix::fs::PermissionsExt; + let mode = match std::fs::metadata(output) { + // Overwriting: `File::create` never changes an existing file's mode, so keep it. + Ok(meta) => meta.permissions().mode() & 0o777, + // New file: `File::create` opens `0o666`, then the kernel masks it with umask. + Err(_) => 0o666 & !process_umask(), + }; + Some(mode) +} + +#[cfg(not(unix))] +fn target_file_mode(_output: &Path) -> Option { + None +} + +/// Reads the process file-creation mask (`umask`) without leaving it changed. +/// +/// `umask(2)` can only *set* the mask (returning the previous value), so reading it +/// means setting it to `0` and immediately restoring it. That read-modify-restore is +/// not atomic: two concurrent probes can interleave so the second reads the first's +/// transient `0` and restores `0`, permanently clearing the process mask (and mis- +/// computing every subsequent new-file mode). A process-global lock serializes the +/// probes so each is atomic with respect to every other — needed because +/// `fgumi_lib` is a library and callers may run merges concurrently in one process. +#[cfg(unix)] +fn process_umask() -> u32 { + use std::sync::Mutex; + + // Serializes the non-atomic umask read-restore against other concurrent probes. + static UMASK_LOCK: Mutex<()> = Mutex::new(()); + let _guard = UMASK_LOCK.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + + // SAFETY: `umask` has no preconditions and cannot fail; it is `unsafe` only + // because it is a raw libc binding. We restore the original mask on the very + // next call under `UMASK_LOCK`, so the process-wide umask is unchanged on return. + #[allow(unsafe_code)] + let previous = unsafe { + let previous = libc::umask(0); + libc::umask(previous); + previous + }; + // `libc::mode_t` is `u16` on macOS (the conversion widens) and `u32` on Linux + // (the conversion is a no-op), so `useless_conversion` fires only on Linux. + #[allow( + clippy::useless_conversion, + reason = "libc::mode_t is u16 on macOS and u32 on Linux; the From keeps this portable" + )] + u32::from(previous) +} + /// Maps read group ID -> library ordinal for O(1) comparison. /// /// Pre-computes ordinals by sorting library names alphabetically. @@ -1778,32 +1971,42 @@ impl RawExternalSorter { let lib_lookup = LibraryLookup::from_header(header); let cell_tag = self.cell_tag; let hasher = cb_hasher(); - self.run_merge_loop(&mut readers, &output_header, output, |bam| { + self.run_merge_loop(&mut readers, inputs, &output_header, output, |bam| { extract_template_key_inline(bam, &lib_lookup, cell_tag, &hasher) }) } SortOrder::Coordinate => { #[allow(clippy::cast_possible_truncation)] let nref = header.reference_sequences().len() as u32; - self.run_merge_loop(&mut readers, &output_header, output, |bam| RawCoordinateKey { - sort_key: extract_coordinate_key_inline(bam, nref), + self.run_merge_loop(&mut readers, inputs, &output_header, output, |bam| { + RawCoordinateKey { sort_key: extract_coordinate_key_inline(bam, nref) } }) } SortOrder::Queryname(QuerynameComparator::Lexicographic) => { let ctx = SortContext::from_header(header); - self.run_merge_loop(&mut readers, &output_header, output, |bam| { + self.run_merge_loop(&mut readers, inputs, &output_header, output, |bam| { RawQuerynameLexKey::extract(bam, &ctx) }) } SortOrder::Queryname(QuerynameComparator::Natural) => { let ctx = SortContext::from_header(header); - self.run_merge_loop(&mut readers, &output_header, output, |bam| { + self.run_merge_loop(&mut readers, inputs, &output_header, output, |bam| { RawQuerynameKey::extract(bam, &ctx) }) } } } + /// The `fgumi sort --order` value for this sorter's order, for error hints. + fn sort_order_flag_value(&self) -> &'static str { + match self.sort_order { + SortOrder::Coordinate => "coordinate", + SortOrder::Queryname(QuerynameComparator::Lexicographic) => "queryname", + SortOrder::Queryname(QuerynameComparator::Natural) => "queryname::natural", + SortOrder::TemplateCoordinate => "template-coordinate", + } + } + /// Open background prefetch readers for multiple BAM files. fn open_bam_prefetch_readers(inputs: &[PathBuf]) -> Result> { inputs @@ -1822,6 +2025,7 @@ impl RawExternalSorter { fn run_merge_loop( &self, readers: &mut [RawReadAheadReader], + inputs: &[PathBuf], output_header: &Header, output: &Path, extract_key: impl Fn(&[u8]) -> K, @@ -1844,22 +2048,30 @@ impl RawExternalSorter { } } + // Write to an exclusive temp and atomically persist on success, so a + // mid-merge sort-order violation (below) — or any other error — never + // leaves a partial/corrupt output or a stray temp behind. Chosen before + // the empty-input branch so header-only outputs are finalized through the + // same atomic path as non-empty merges. + let out_target = MergeOutputTarget::create(output)?; + if initial_keys.is_empty() { debug!("Merge complete: 0 records merged"); let writer = fgumi_bam_io::create_raw_bam_writer( - output, + out_target.path(), output_header, self.threads, self.output_compression, )?; writer.finish()?; + out_target.persist()?; return Ok(0); } let mut tree = LoserTree::new(initial_keys); let mut writer = fgumi_bam_io::create_raw_bam_writer( - output, + out_target.path(), output_header, self.threads, self.output_compression, @@ -1867,6 +2079,13 @@ impl RawExternalSorter { let mut records_merged = 0u64; let merge_progress = ProgressTracker::new("Merged records").with_interval(1_000_000); + // Set when an input is found not to be monotonic in the merge order: + // (source index into `inputs`, 1-based record number within that input). + let mut violation: Option<(usize, u64)> = None; + // Per-source count of records pulled so far (each source starts with its + // first record already loaded), used to report the offending record's + // 1-based position within its input. + let mut pulled_per_source = vec![1u64; records.len()]; while tree.winner_is_active() { let winner = tree.winner(); @@ -1881,13 +2100,44 @@ impl RawExternalSorter { buf.clear(); buf.extend_from_slice(raw_record.as_ref()); let new_key = extract_key(buf); + // MERGE3-01 streaming verify: the merge assumes each input is + // already sorted in the merge order. `tree.winner_key()` still + // holds the key of the record we just emitted from this source; + // if the next record sorts before it, the input isn't monotonic + // and the k-way merge would silently corrupt the output. + if new_key < *tree.winner_key() { + violation = Some((reader_idx, pulled_per_source[winner] + 1)); + break; + } + pulled_per_source[winner] += 1; tree.replace_winner(new_key); } else { tree.remove_winner(); } } + if let Some((reader_idx, record_no)) = violation { + // Close the writer's handle to the partial output. `out_target` is + // still owned here, so returning the error below drops it and removes + // the temp — nothing partial is left behind. Not finalizing a doomed + // output also avoids `finish` masking the real (sort-order) error. + drop(writer); + let source = inputs[reader_idx].display(); + let order = self.sort_order_flag_value(); + // The path is named for identification only; the remediation commands + // use a `` placeholder so a path with shell metacharacters + // can't turn the copy-pasteable hint into something unexpected. + anyhow::bail!( + "Input '{source}' is not sorted in {order} order: record {record_no} sorts \ + before a preceding record from the same input, so the k-way merge would \ + corrupt the output. Sort that input in {order} order first \ + (`fgumi sort -i -o sorted.bam --order {order}`), or locate the \ + violation with `fgumi sort -i --verify --order {order}`." + ); + } + writer.finish()?; + out_target.persist()?; merge_progress.log_final(); Ok(records_merged) @@ -3602,6 +3852,110 @@ mod tests { use noodles::sam::header::record::value::Map; use noodles::sam::header::record::value::map::ReadGroup; + // ======================================================================== + // process_umask concurrency + // ======================================================================== + + /// `process_umask` reads the mask via a non-atomic set-0-then-restore. Two + /// interleaved probes can leave the process mask permanently `0` (and every + /// probe observe the wrong value); `UMASK_LOCK` must serialize them. Read the + /// mask once, hammer it from many threads, and assert every probe — and the + /// final mask — matches the initial value. Without the lock this corrupts to + /// `0` reliably under contention. (Uses only `process_umask` so it introduces + /// no new `unsafe` site.) + #[cfg(unix)] + #[test] + fn test_process_umask_is_concurrency_safe() { + use std::thread; + + let expected = process_umask(); + let threads: Vec<_> = (0..8) + .map(|_| { + thread::spawn(move || { + for _ in 0..2000 { + assert_eq!( + process_umask(), + expected, + "a concurrent probe observed a corrupted umask" + ); + } + }) + }) + .collect(); + for t in threads { + t.join().expect("umask probe thread panicked"); + } + assert_eq!( + process_umask(), + expected, + "concurrent probes must leave the process umask unchanged" + ); + } + + // ======================================================================== + // resolve_symlink_output / target_file_mode + // ======================================================================== + + /// A non-symlink path — even one that does not exist yet — is returned + /// unchanged, so a brand-new merge output is not perturbed by the resolver. + #[cfg(unix)] + #[test] + fn test_resolve_symlink_output_passthrough_for_nonexistent() { + let dir = tempfile::tempdir().unwrap(); + let out = dir.path().join("merged.bam"); + assert_eq!(resolve_symlink_output(&out).unwrap(), out); + } + + /// A relative symlink target resolves against the link's own directory, so a + /// `latest.bam -> run.bam` link is followed to the sibling it points at. + #[cfg(unix)] + #[test] + fn test_resolve_symlink_output_follows_relative_target() { + let dir = tempfile::tempdir().unwrap(); + let target = dir.path().join("run.bam"); + std::fs::write(&target, b"x").unwrap(); + let link = dir.path().join("latest.bam"); + // Relative target ("run.bam") must resolve against the link's directory. + std::os::unix::fs::symlink("run.bam", &link).unwrap(); + assert_eq!(resolve_symlink_output(&link).unwrap(), target); + } + + /// A cyclic symlink chain bails at `MAX_LINKS` with an error instead of + /// looping forever. + #[cfg(unix)] + #[test] + fn test_resolve_symlink_output_detects_cycle() { + let dir = tempfile::tempdir().unwrap(); + let a = dir.path().join("a"); + let b = dir.path().join("b"); + std::os::unix::fs::symlink(&b, &a).unwrap(); + std::os::unix::fs::symlink(&a, &b).unwrap(); + assert!(resolve_symlink_output(&a).is_err()); + } + + /// Overwriting an existing destination keeps its current mode (matching + /// `File::create`, which never re-chmods an existing file). + #[cfg(unix)] + #[test] + fn test_target_file_mode_keeps_existing_file_mode() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("out.bam"); + std::fs::write(&path, b"x").unwrap(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o640)).unwrap(); + assert_eq!(target_file_mode(&path), Some(0o640)); + } + + /// A new destination gets `0o666 & !umask` — the mode `File::create` would + /// have produced — not the temp file's private `0600`. + #[cfg(unix)] + #[test] + fn test_target_file_mode_new_file_uses_umask() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("does-not-exist.bam"); + assert_eq!(target_file_mode(&path), Some(0o666 & !process_umask())); + } + // ======================================================================== // LibraryLookup tests // ======================================================================== diff --git a/src/lib/commands/merge.rs b/src/lib/commands/merge.rs index 5711e5440..c45645589 100644 --- a/src/lib/commands/merge.rs +++ b/src/lib/commands/merge.rs @@ -146,8 +146,9 @@ impl Command for Merge { } info!("Threads: {}", self.threads); - // Read and merge headers from all inputs - let header = merge_headers(&input_paths)?; + // Read and merge headers from all inputs (also validates each input's + // declared sort order against --order; MERGE3-01 fast check). + let header = merge_headers(&input_paths, self.order)?; let mut sorter = RawExternalSorter::new(self.order.into()) .threads(self.threads) @@ -168,13 +169,106 @@ impl Command for Merge { } } +/// A BAM header's *declared* sort order, as far as merge validation cares. +/// +/// `classify_declared_order` returns `None` when the header asserts no usable +/// order (`SO` absent, or `SO:unsorted` without the template-coordinate +/// sub-sort). Those inputs pass the fast header check and are instead verified +/// record-by-record during the merge (the streaming monotonicity check). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum DeclaredOrder { + Coordinate, + Queryname, + TemplateCoordinate, +} + +impl DeclaredOrder { + fn as_str(self) -> &'static str { + match self { + DeclaredOrder::Coordinate => "coordinate", + DeclaredOrder::Queryname => "queryname", + DeclaredOrder::TemplateCoordinate => "template-coordinate", + } + } +} + +/// Classifies a header's declared sort order for merge-input validation. +fn classify_declared_order(header: &Header) -> Option { + use noodles::sam::header::record::value::map::header::sort_order::{COORDINATE, QUERY_NAME}; + if fgumi_sam::is_template_coordinate_sorted(header) { + Some(DeclaredOrder::TemplateCoordinate) + } else if fgumi_sam::is_sorted(header, COORDINATE) { + Some(DeclaredOrder::Coordinate) + } else if fgumi_sam::is_sorted(header, QUERY_NAME) { + Some(DeclaredOrder::Queryname) + } else { + None + } +} + +/// The declared order an input must carry to be compatible with merge `order`. +fn expected_declared_order(order: SortOrderArg) -> DeclaredOrder { + match order { + SortOrderArg::Coordinate => DeclaredOrder::Coordinate, + SortOrderArg::Queryname | SortOrderArg::QuerynameNatural => DeclaredOrder::Queryname, + SortOrderArg::TemplateCoordinate => DeclaredOrder::TemplateCoordinate, + } +} + +/// The `fgumi sort --order` value string for an order (used in error hints). +fn order_flag_value(order: SortOrderArg) -> &'static str { + match order { + SortOrderArg::Coordinate => "coordinate", + SortOrderArg::Queryname => "queryname", + SortOrderArg::QuerynameNatural => "queryname::natural", + SortOrderArg::TemplateCoordinate => "template-coordinate", + } +} + +/// MERGE3-01 (fast header check): reject an input whose header *declares* a sort +/// order that conflicts with the requested merge `order`. +/// +/// The k-way merge only yields globally-sorted output if every input is already +/// sorted in `order`; a coordinate-sorted BAM fed to the default +/// `--order template-coordinate` merge (or any declared mismatch) silently +/// corrupts the output with a success exit. This catches the common footgun +/// before any records are read. Inputs that declare no usable order pass here +/// and are verified record-by-record during the merge instead. +/// +/// # Errors +/// +/// Returns an error if the header declares an order that conflicts with `order`. +fn check_input_declared_order(header: &Header, order: SortOrderArg, source: &str) -> Result<()> { + let expected = expected_declared_order(order); + if let Some(declared) = classify_declared_order(header) { + if declared != expected { + // `{source}` names the input for identification only; the remediation + // command uses a `` placeholder (matching the streaming-verify + // error) so a path with shell metacharacters can't turn the + // copy-pasteable hint into something unexpected. + bail!( + "Input '{source}' is sorted by {declared} but merge was asked for --order \ + {requested}. Every input must already be sorted in the merge order, or the \ + k-way merge silently corrupts the output.\n\nEither merge in the inputs' \ + existing order:\n fgumi merge --order {declared} ...\nor sort the inputs to \ + {requested} first:\n fgumi sort -i -o sorted.bam --order {requested}", + declared = declared.as_str(), + requested = order_flag_value(order), + ); + } + } + Ok(()) +} + /// Merge headers from multiple BAM files. /// /// Uses the first input's reference sequences and header line as the base. /// Combines read groups and program records from all inputs, with earlier /// inputs taking precedence for duplicate IDs (matching `samtools merge`). -/// Validates that all inputs share the same reference sequences (names and order). -fn merge_headers(input_paths: &[PathBuf]) -> Result
{ +/// Validates that all inputs share the same reference sequences (names and order) +/// and that each input's declared sort order is compatible with `order` +/// (MERGE3-01 fast check). +fn merge_headers(input_paths: &[PathBuf], order: SortOrderArg) -> Result
{ if input_paths.is_empty() { bail!("No input files to merge headers from"); } @@ -188,6 +282,13 @@ fn merge_headers(input_paths: &[PathBuf]) -> Result
{ }) .collect::>>()?; + // MERGE3-01: reject any input whose declared order conflicts with the merge + // order (validated for every input, including the single-input case, since + // the output header is stamped with the requested order regardless). + for (path, header) in input_paths.iter().zip(headers.iter()) { + check_input_declared_order(header, order, &path.display().to_string())?; + } + let first_header = &headers[0]; if headers.len() == 1 { @@ -309,8 +410,8 @@ mod tests { let dir = tempfile::tempdir().expect("failed to create temp dir"); let bam = write_bam_with_read_groups(dir.path(), "single", &["RG1", "RG2"]); - let header = - merge_headers(std::slice::from_ref(&bam)).expect("merge_headers should succeed"); + let header = merge_headers(std::slice::from_ref(&bam), SortOrderArg::Coordinate) + .expect("merge_headers should succeed"); // Should be the same header (same RGs) let rg_ids: Vec = @@ -326,7 +427,8 @@ mod tests { let bam_a = write_bam_with_read_groups(dir.path(), "a", &["RG1"]); let bam_b = write_bam_with_read_groups(dir.path(), "b", &["RG2"]); - let header = merge_headers(&[bam_a, bam_b]).expect("merge_headers should succeed"); + let header = merge_headers(&[bam_a, bam_b], SortOrderArg::Coordinate) + .expect("merge_headers should succeed"); let rg_ids: Vec = header.read_groups().iter().map(|(id, _)| id.to_string()).collect(); @@ -342,7 +444,8 @@ mod tests { let bam_a = write_bam_with_read_groups(dir.path(), "a", &["RG1"]); let bam_b = write_bam_with_read_groups(dir.path(), "b", &["RG1", "RG2"]); - let header = merge_headers(&[bam_a, bam_b]).expect("merge_headers should succeed"); + let header = merge_headers(&[bam_a, bam_b], SortOrderArg::Coordinate) + .expect("merge_headers should succeed"); let rg_ids: Vec = header.read_groups().iter().map(|(id, _)| id.to_string()).collect(); @@ -390,7 +493,7 @@ mod tests { let bam_a = write_bam_with_refs(dir.path(), "a", &["chr1", "chr2"]); let bam_b = write_bam_with_refs(dir.path(), "b", &["chr1"]); - let result = merge_headers(&[bam_a, bam_b]); + let result = merge_headers(&[bam_a, bam_b], SortOrderArg::Coordinate); assert!(result.is_err()); let msg = result.unwrap_err().to_string(); assert!(msg.contains("Reference sequence count mismatch"), "unexpected error: {msg}"); @@ -402,7 +505,7 @@ mod tests { let bam_a = write_bam_with_refs(dir.path(), "a", &["chr1", "chr2"]); let bam_b = write_bam_with_refs(dir.path(), "b", &["chr1", "chrX"]); - let result = merge_headers(&[bam_a, bam_b]); + let result = merge_headers(&[bam_a, bam_b], SortOrderArg::Coordinate); assert!(result.is_err()); let msg = result.unwrap_err().to_string(); assert!(msg.contains("Reference sequence mismatch"), "unexpected error: {msg}"); @@ -454,4 +557,88 @@ mod tests { header.read_groups().iter().map(|(id, _)| id.to_string()).collect(); assert_eq!(rg_ids.len(), 2); } + + /// MERGE3-01 fast header check: an input whose header *declares* an order + /// conflicting with `--order` is rejected; matching or undeclared inputs pass + /// (undeclared ones are verified record-by-record during the merge instead). + #[rstest::rstest] + // Declared coordinate. + #[case::coord_ok("@HD\tVN:1.6\tSO:coordinate\n", SortOrderArg::Coordinate, true)] + #[case::coord_into_tc("@HD\tVN:1.6\tSO:coordinate\n", SortOrderArg::TemplateCoordinate, false)] + #[case::coord_into_qname("@HD\tVN:1.6\tSO:coordinate\n", SortOrderArg::Queryname, false)] + // Declared queryname (both lex and natural merges accept a queryname header). + #[case::qname_ok("@HD\tVN:1.6\tSO:queryname\n", SortOrderArg::Queryname, true)] + #[case::qname_natural_ok("@HD\tVN:1.6\tSO:queryname\n", SortOrderArg::QuerynameNatural, true)] + #[case::qname_into_coord("@HD\tVN:1.6\tSO:queryname\n", SortOrderArg::Coordinate, false)] + #[case::qname_into_tc("@HD\tVN:1.6\tSO:queryname\n", SortOrderArg::TemplateCoordinate, false)] + // Declared template-coordinate. + #[case::tc_ok( + "@HD\tVN:1.6\tSO:unsorted\tGO:query\tSS:template-coordinate\n", + SortOrderArg::TemplateCoordinate, + true + )] + #[case::tc_into_coord( + "@HD\tVN:1.6\tSO:unsorted\tGO:query\tSS:template-coordinate\n", + SortOrderArg::Coordinate, + false + )] + #[case::tc_into_qname( + "@HD\tVN:1.6\tSO:unsorted\tGO:query\tSS:template-coordinate\n", + SortOrderArg::Queryname, + false + )] + // Undeclared: bare, plain unsorted, or query-grouped-without-SS → pass here. + #[case::bare_any("@HD\tVN:1.6\n", SortOrderArg::Coordinate, true)] + #[case::unsorted_any("@HD\tVN:1.6\tSO:unsorted\n", SortOrderArg::TemplateCoordinate, true)] + #[case::query_grouped_no_ss( + "@HD\tVN:1.6\tSO:unsorted\tGO:query\n", + SortOrderArg::TemplateCoordinate, + true + )] + fn test_check_input_declared_order( + #[case] header_str: &str, + #[case] order: SortOrderArg, + #[case] expect_ok: bool, + ) { + let header: Header = header_str.parse().expect("parse header"); + let result = check_input_declared_order(&header, order, "in.bam"); + assert_eq!(result.is_ok(), expect_ok, "header {header_str:?} into --order {order:?}"); + if let Err(e) = result { + let msg = e.to_string(); + assert!(msg.contains("in.bam"), "message missing source: {msg}"); + assert!(msg.contains(order_flag_value(order)), "message missing order hint: {msg}"); + } + } + + /// MERGE3-01 hardening: the copy-pasteable remediation command must use a + /// fixed `` placeholder, never the interpolated source path, so a + /// filename with shell metacharacters can't turn the hint into an unintended + /// command. The source is still named in the diagnostic text for identification. + #[test] + fn test_declared_order_error_does_not_interpolate_source_into_shell_command() { + // A coordinate header into a queryname merge triggers the declared-order + // conflict; the source path carries shell metacharacters. + let header: Header = "@HD\tVN:1.6\tSO:coordinate\n".parse().expect("parse header"); + let malicious = "a'; rm -rf ~; echo '.bam"; + let err = check_input_declared_order(&header, SortOrderArg::Queryname, malicious) + .expect_err("coordinate header into queryname merge must be rejected"); + let msg = err.to_string(); + + // The remediation command uses the fixed placeholder... + assert!( + msg.contains("fgumi sort -i -o sorted.bam"), + "remediation must use the placeholder: {msg}" + ); + // ...and the source path is never spliced into any `-i` argument. + assert!( + !msg.contains(&format!("-i '{malicious}'")), + "source path must not be interpolated into the sort command: {msg}" + ); + assert!( + !msg.contains(&format!("-i {malicious}")), + "source path must not be interpolated into the sort command: {msg}" + ); + // The source is still named for identification (in the descriptive text). + assert!(msg.contains(malicious), "message should still name the source: {msg}"); + } } diff --git a/tests/integration/main.rs b/tests/integration/main.rs index 72574e64c..e1a65dc40 100644 --- a/tests/integration/main.rs +++ b/tests/integration/main.rs @@ -31,6 +31,7 @@ mod test_fastq_command; mod test_filter_command; mod test_group_command; mod test_group_determinism; +mod test_merge_command; mod test_pipeline_concurrency; mod test_review_command; #[cfg(feature = "simplex")] diff --git a/tests/integration/test_merge_command.rs b/tests/integration/test_merge_command.rs new file mode 100644 index 000000000..7087da229 --- /dev/null +++ b/tests/integration/test_merge_command.rs @@ -0,0 +1,498 @@ +//! End-to-end tests for the merge command's input sort-order guard (MERGE3-01). +//! +//! Two mechanisms are exercised through `Merge::execute`: +//! 1. the fast header-declared check (rejects a declared-order conflict before +//! reading records, e.g. a coordinate BAM into the template-coordinate default), and +//! 2. the streaming monotonicity verify (catches actual disorder in a +//! bare/undeclared-header input, and leaves no partial output). + +use fgumi_lib::commands::command::Command; +use fgumi_lib::commands::merge::Merge; +use fgumi_lib::commands::sort::SortOrderArg; +use fgumi_raw_bam::{RawRecord, SamBuilder}; +use noodles::bam; +use noodles::sam::Header; +use noodles::sam::alignment::io::Write as AlignmentWrite; +use std::fs; +use std::path::Path; +use tempfile::TempDir; + +use crate::helpers::bam_generator::{create_coordinate_sorted_header, to_record_buf}; + +/// A mapped 20M record on chr1 (ref 0) at the given 1-based position. +fn mapped_record(name: &[u8], pos: i32) -> RawRecord { + let mut b = SamBuilder::new(); + b.read_name(name) + .ref_id(0) + .pos(pos - 1) // 1-based → 0-based + .mapq(60) + .cigar_ops(&[20 << 4]) // 20M + .sequence(b"ACGTACGTACGTACGTACGT") + .qualities(&[30; 20]); + b.build() +} + +/// Write `records` to `path` under `header`. +fn write_bam(path: &Path, header: &Header, records: &[RawRecord]) { + let mut writer = bam::io::Writer::new(fs::File::create(path).expect("create bam")); + writer.write_header(header).expect("write header"); + for r in records { + writer.write_alignment_record(header, &to_record_buf(r)).expect("write record"); + } + writer.try_finish().expect("finish bam"); +} + +/// Assert no `.fgumi-merge-*.tmp` staging sibling remains in `dir` (the atomic +/// temp+persist must clean up its temp on both the success and rejection paths). +fn assert_no_leftover_temp(dir: &Path) { + let leftover_temps: Vec<_> = fs::read_dir(dir) + .unwrap() + .filter_map(Result::ok) + .filter(|e| e.file_name().to_string_lossy().contains("fgumi-merge-")) + .map(|e| e.file_name()) + .collect(); + assert!(leftover_temps.is_empty(), "merge temp files must be cleaned up: {leftover_temps:?}"); +} + +/// Read a merged BAM into `(read name, 1-based alignment start)` pairs in file +/// order — an identity-aware oracle for asserting merge interleaving. +fn read_name_pos(path: &Path) -> Vec<(String, usize)> { + let mut reader = bam::io::Reader::new(fs::File::open(path).unwrap()); + let _ = reader.read_header().unwrap(); + reader + .records() + .map(|r| { + let rec = r.unwrap(); + let name = String::from_utf8(rec.name().unwrap().to_vec()).unwrap(); + let pos = rec.alignment_start().unwrap().unwrap().get(); + (name, pos) + }) + .collect() +} + +/// A header with a `chr1` reference and **no** `@HD` sort-order tag — its declared +/// order classifies as "none", so it passes the header check and is verified by +/// the streaming monotonicity check during the merge. +fn bare_header() -> Header { + use noodles::sam::header::record::value::Map; + use noodles::sam::header::record::value::map::ReferenceSequence; + use std::num::NonZeroUsize; + Header::builder() + .add_reference_sequence( + "chr1", + Map::::new(NonZeroUsize::new(10000).expect("nonzero")), + ) + .build() +} + +fn merge( + output: &Path, + inputs: Vec, + order: SortOrderArg, +) -> anyhow::Result<()> { + Merge { + output: output.to_path_buf(), + inputs, + input_list: None, + order, + threads: 1, + compression_level: 1, + } + .execute("fgumi merge") +} + +/// MERGE3-01 header check: a coordinate-sorted input fed to the default +/// `--order template-coordinate` merge is rejected before any records are read. +#[test] +fn test_merge_rejects_coordinate_input_into_template_coordinate_default() { + let dir = TempDir::new().unwrap(); + let header = create_coordinate_sorted_header("chr1", 10000); + let a = dir.path().join("a.bam"); + let b = dir.path().join("b.bam"); + write_bam(&a, &header, &[mapped_record(b"reada0", 100), mapped_record(b"reada1", 200)]); + write_bam(&b, &header, &[mapped_record(b"readb0", 150), mapped_record(b"readb1", 250)]); + let out = dir.path().join("merged.bam"); + + // Default order is template-coordinate; the coordinate inputs conflict. + let err = merge(&out, vec![a, b], SortOrderArg::TemplateCoordinate) + .expect_err("must reject coordinate input into template-coordinate merge"); + let msg = err.to_string(); + assert!(msg.contains("sorted by coordinate"), "unexpected error: {msg}"); + assert!(msg.contains("a.bam"), "error should name the offending input: {msg}"); + assert!(!out.exists(), "no output should be written on a declared-order conflict"); +} + +/// The CLI default for `--order` is template-coordinate — the fgumi-specific +/// footgun this guard defends (`fgumi merge a.bam b.bam` on plain +/// coordinate-sorted BAMs would otherwise silently corrupt). The reject test +/// above passes `TemplateCoordinate` through the `merge()` helper, bypassing +/// clap; parse without `--order` here so the default the guard relies on is +/// itself pinned. +#[test] +fn test_merge_order_defaults_to_template_coordinate() { + use clap::Parser; + let parsed = Merge::try_parse_from(["fgumi-merge", "-o", "out.bam", "in.bam"]) + .expect("merge parses without --order"); + assert_eq!(parsed.order, SortOrderArg::TemplateCoordinate); +} + +/// A valid merge (coordinate inputs, `--order coordinate`) succeeds and produces +/// globally coordinate-sorted output. +#[test] +fn test_merge_valid_coordinate_succeeds() { + let dir = TempDir::new().unwrap(); + let header = create_coordinate_sorted_header("chr1", 10000); + let a = dir.path().join("a.bam"); + let b = dir.path().join("b.bam"); + write_bam(&a, &header, &[mapped_record(b"reada0", 100), mapped_record(b"reada1", 300)]); + write_bam(&b, &header, &[mapped_record(b"readb0", 200), mapped_record(b"readb1", 400)]); + let out = dir.path().join("merged.bam"); + + merge(&out, vec![a, b], SortOrderArg::Coordinate).expect("valid coordinate merge"); + assert!(out.exists()); + + // The atomic temp+rename must leave no `.fgumi-merge-*.tmp` sibling on the + // success path either — same cleanup contract the failure test checks. + assert_no_leftover_temp(dir.path()); + + // Assert record *identity* at each position, not just that positions are + // sorted — a name-aware check fails on interleaving bugs that leave the + // coordinate column sorted but swap records between inputs. + let got = read_name_pos(&out); + assert_eq!( + got, + vec![ + ("reada0".to_string(), 100), + ("readb0".to_string(), 200), + ("reada1".to_string(), 300), + ("readb1".to_string(), 400), + ], + "merged output must interleave the two inputs in coordinate order" + ); +} + +/// MERGE3-01 streaming verify: a bare-header input whose records are actually +/// out of coordinate order passes the header check but is caught record-by-record +/// during the merge — and no partial/temp output is left behind (temp + rename). +#[test] +fn test_merge_streaming_verify_rejects_missorted_bare_header_no_partial_output() { + let dir = TempDir::new().unwrap(); + let coord = create_coordinate_sorted_header("chr1", 10000); + let good = dir.path().join("good.bam"); + write_bam(&good, &coord, &[mapped_record(b"readg0", 100)]); + + // Bare header (declared order "none"), records descending in position. + let bad = dir.path().join("bad.bam"); + write_bam(&bad, &bare_header(), &[mapped_record(b"readx", 300), mapped_record(b"ready", 100)]); + + let out = dir.path().join("merged.bam"); + let err = merge(&out, vec![good, bad.clone()], SortOrderArg::Coordinate) + .expect_err("must reject an actually-mis-sorted input"); + let msg = err.to_string(); + assert!(msg.contains("not sorted in coordinate order"), "unexpected error: {msg}"); + assert!(msg.contains("bad.bam"), "error should name the offending input: {msg}"); + + // No partial output and no leftover temp file. The merge temp is a uniquely + // named `.fgumi-merge-*.tmp` sibling, so scan the directory for that prefix + // rather than a fixed name (a RAII NamedTempFile removes it on any error). + assert!(!out.exists(), "a rejected merge must leave no output file"); + assert_no_leftover_temp(dir.path()); +} + +/// MERGE3-01 streaming verify, destination preservation: a streaming-order +/// failure must not touch a *pre-existing* destination. The atomic temp+persist +/// writes to a sibling temp and only renames on success, so a rejected merge must +/// leave an already-present output byte-for-byte unchanged (the sibling +/// absent-output test above only proves an absent output stays absent). +#[test] +fn test_merge_streaming_verify_preserves_existing_destination() { + let dir = TempDir::new().unwrap(); + let coord = create_coordinate_sorted_header("chr1", 10000); + let good = dir.path().join("good.bam"); + write_bam(&good, &coord, &[mapped_record(b"readg0", 100)]); + + // Bare header (declared order "none"), records descending in position — caught + // by the streaming monotonicity check, not the fast header-declared check. + let bad = dir.path().join("bad.bam"); + write_bam(&bad, &bare_header(), &[mapped_record(b"readx", 300), mapped_record(b"ready", 100)]); + + // Pre-create the destination with sentinel bytes that are *not* a valid BAM, + // so any accidental write (partial output, clobber) is detectable. + let out = dir.path().join("merged.bam"); + let sentinel: &[u8] = b"pre-existing sentinel output that must survive a rejected merge"; + fs::write(&out, sentinel).unwrap(); + + let err = merge(&out, vec![good, bad.clone()], SortOrderArg::Coordinate) + .expect_err("must reject an actually-mis-sorted input"); + assert!(err.to_string().contains("not sorted in coordinate order"), "unexpected error: {err}"); + + // The destination must still exist with exactly its original bytes... + assert!(out.exists(), "a rejected merge must not remove a pre-existing destination"); + assert_eq!( + fs::read(&out).unwrap(), + sentinel, + "a rejected merge must leave a pre-existing destination byte-for-byte unchanged" + ); + // ...and no leftover temp sibling from the discarded staged output. + assert_no_leftover_temp(dir.path()); +} + +/// MERGE3-01 streaming verify boundary: two consecutive records at the *same* +/// coordinate are a tie, not a decrease, so the monotonicity check (`<`, not +/// `<=`) must accept them. Guards the `>` vs `>=` off-by-one in the comparator +/// that the strictly-decreasing rejection test above cannot surface. +#[test] +fn test_merge_streaming_verify_accepts_equal_adjacent_positions() { + let dir = TempDir::new().unwrap(); + // Bare header (declared order "none") routes records through the streaming + // monotonicity check rather than the fast header-declared check. + let input = dir.path().join("ties.bam"); + write_bam( + &input, + &bare_header(), + &[mapped_record(b"read0", 100), mapped_record(b"read1", 100)], + ); + + let out = dir.path().join("merged.bam"); + merge(&out, vec![input], SortOrderArg::Coordinate) + .expect("equal adjacent positions must be accepted, not rejected as mis-sorted"); + assert!(out.exists()); + + // The tied records must pass through in their original order, unaltered. + let got = read_name_pos(&out); + assert_eq!( + got, + vec![("read0".to_string(), 100), ("read1".to_string(), 100)], + "tied-position records must be accepted and preserved in order" + ); +} + +/// The atomic temp+persist output must carry normal file permissions, not the +/// `0600` a `NamedTempFile` is created with. Overwriting an existing destination +/// keeps that destination's mode (matching the pre-atomic-temp `File::create` +/// path), which is deterministic regardless of the test process's umask. +#[cfg(unix)] +#[test] +fn test_merge_output_preserves_existing_destination_mode() { + use std::os::unix::fs::PermissionsExt; + + let dir = TempDir::new().unwrap(); + let header = create_coordinate_sorted_header("chr1", 10000); + let a = dir.path().join("a.bam"); + let b = dir.path().join("b.bam"); + write_bam(&a, &header, &[mapped_record(b"reada0", 100)]); + write_bam(&b, &header, &[mapped_record(b"readb0", 200)]); + + // Pre-create the destination with a group/other-readable mode (0644); the + // temp NamedTempFile is 0600, so without the mode fix the merge would leave + // the output owner-only. + let out = dir.path().join("merged.bam"); + fs::write(&out, b"placeholder").unwrap(); + fs::set_permissions(&out, fs::Permissions::from_mode(0o644)).unwrap(); + + merge(&out, vec![a, b], SortOrderArg::Coordinate).expect("valid coordinate merge"); + + let mode = fs::metadata(&out).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, 0o644, "merge must preserve the destination's existing mode, got {mode:#o}"); +} + +/// A newly created merge output (destination absent) must carry the mode a plain +/// `File::create` produces (`0o666 & !umask`), not the staging `NamedTempFile`'s +/// private `0600`. Compare against a control created via `File::create` in the +/// same process so the expectation tracks the test's umask deterministically — +/// an independent oracle for the new-file branch of `target_file_mode`. +#[cfg(unix)] +#[test] +fn test_merge_new_output_matches_file_create_mode() { + use std::os::unix::fs::PermissionsExt; + + let dir = TempDir::new().unwrap(); + let header = create_coordinate_sorted_header("chr1", 10000); + let a = dir.path().join("a.bam"); + let b = dir.path().join("b.bam"); + write_bam(&a, &header, &[mapped_record(b"reada0", 100)]); + write_bam(&b, &header, &[mapped_record(b"readb0", 200)]); + + // Oracle: the mode File::create yields under this process's umask. + let control = dir.path().join("control"); + fs::File::create(&control).unwrap(); + let expected = fs::metadata(&control).unwrap().permissions().mode() & 0o777; + + // Merge into a previously absent destination. + let out = dir.path().join("merged.bam"); + assert!(!out.exists(), "destination must not exist before the merge"); + merge(&out, vec![a, b], SortOrderArg::Coordinate).expect("valid coordinate merge"); + + let mode = fs::metadata(&out).unwrap().permissions().mode() & 0o777; + assert_eq!( + mode, expected, + "new merge output must match File::create's mode ({expected:#o}), got {mode:#o}" + ); +} + +/// The atomic temp+persist replaces its destination with a regular file, so a +/// non-regular output (FIFO, device, socket, directory) must be rejected up front +/// rather than clobbered. A directory is the portable stand-in for the whole +/// non-regular class — it exercises the same `is_file()` guard without needing a +/// platform-specific `mkfifo`. +#[test] +fn test_merge_rejects_non_regular_output_destination() { + let dir = TempDir::new().unwrap(); + let header = create_coordinate_sorted_header("chr1", 10000); + let a = dir.path().join("a.bam"); + let b = dir.path().join("b.bam"); + write_bam(&a, &header, &[mapped_record(b"reada0", 100)]); + write_bam(&b, &header, &[mapped_record(b"readb0", 200)]); + + // A directory sitting where the output path points is a non-regular file. + let out = dir.path().join("out_is_a_dir"); + fs::create_dir(&out).unwrap(); + + let err = merge(&out, vec![a, b], SortOrderArg::Coordinate) + .expect_err("must reject a non-regular output destination"); + assert!( + err.to_string().contains("must be a regular file or stdout"), + "unexpected error: {err}" + ); + // The destination directory must be left untouched, and no temp staged. + assert!(out.is_dir(), "the non-regular destination must be left in place"); + assert_no_leftover_temp(dir.path()); +} + +/// The atomic temp+persist must follow a symlinked output to its real target +/// (matching the pre-atomic-temp `File::create`, which follows symlinks) rather +/// than replacing the link with a regular file. A `latest.bam -> run.bam` +/// symlink must survive the merge, with the merged output landing in `run.bam`. +#[cfg(unix)] +#[test] +fn test_merge_output_follows_symlink_destination() { + let dir = TempDir::new().unwrap(); + let header = create_coordinate_sorted_header("chr1", 10000); + let a = dir.path().join("a.bam"); + let b = dir.path().join("b.bam"); + write_bam(&a, &header, &[mapped_record(b"reada0", 100)]); + write_bam(&b, &header, &[mapped_record(b"readb0", 200)]); + + // A pre-existing real target with a relative symlink pointing at it. + let real = dir.path().join("run.bam"); + write_bam(&real, &header, &[mapped_record(b"stale", 1)]); + let link = dir.path().join("latest.bam"); + std::os::unix::fs::symlink("run.bam", &link).unwrap(); + + merge(&link, vec![a, b], SortOrderArg::Coordinate).expect("valid coordinate merge"); + + // The symlink itself must be preserved, not clobbered by a regular file... + assert!( + fs::symlink_metadata(&link).unwrap().file_type().is_symlink(), + "symlink output must be preserved, not replaced by a regular file" + ); + // ...and the merged records must land in the real target it points to. + let mut reader = bam::io::Reader::new(fs::File::open(&real).unwrap()); + let _ = reader.read_header().unwrap(); + let names: Vec = reader + .records() + .map(|r| String::from_utf8(r.unwrap().name().unwrap().to_vec()).unwrap()) + .collect(); + assert_eq!( + names, + vec!["reada0".to_string(), "readb0".to_string()], + "merged records must land in the symlink's target file" + ); +} + +/// MERGE3-01 streaming verify across every `--order`, not just coordinate: a +/// bare-header input whose records are out of order *for the requested order* is +/// caught record-by-record during the merge, the offending input is named, and no +/// partial output or temp is left behind. The mis-ordered pair is order-specific +/// (descending position for coordinate/template-coordinate, a descending read +/// name for the two queryname orders) so each case actually exercises that order's +/// key comparator rather than re-testing the coordinate path. +#[rstest::rstest] +#[case::coordinate(SortOrderArg::Coordinate, "coordinate", "first", 300, "second", 100)] +#[case::queryname(SortOrderArg::Queryname, "queryname", "nameB", 100, "nameA", 100)] +#[case::queryname_natural( + SortOrderArg::QuerynameNatural, + "queryname::natural", + "read2", + 100, + "read1", + 100 +)] +#[case::template_coordinate( + SortOrderArg::TemplateCoordinate, + "template-coordinate", + "first", + 300, + "second", + 100 +)] +fn test_merge_streaming_verify_rejects_missorted_for_each_order( + #[case] order: SortOrderArg, + #[case] order_label: &str, + #[case] name0: &str, + #[case] pos0: i32, + #[case] name1: &str, + #[case] pos1: i32, +) { + let dir = TempDir::new().unwrap(); + // Bare header (declared order "none") routes records through the streaming + // monotonicity check rather than the fast header-declared check. + let bad = dir.path().join("bad.bam"); + write_bam( + &bad, + &bare_header(), + &[mapped_record(name0.as_bytes(), pos0), mapped_record(name1.as_bytes(), pos1)], + ); + let out = dir.path().join("merged.bam"); + + let err = merge(&out, vec![bad], order) + .expect_err("must reject an input mis-sorted for the requested order"); + let msg = err.to_string(); + assert!( + msg.contains(&format!("not sorted in {order_label} order")), + "error must name the {order_label} order: {msg}" + ); + assert!(msg.contains("bad.bam"), "error should name the offending input: {msg}"); + assert!(!out.exists(), "a rejected merge must leave no output file"); + assert_no_leftover_temp(dir.path()); +} + +/// The valid-order counterpart to the per-order rejection table: a correctly +/// sorted bare-header input merges for every `--order`, and the records pass +/// through with their identities intact. Asserting the `(name, position)` oracle +/// — not just a record count — catches an order-specific key bug that reorders or +/// drops records while leaving the count unchanged. +#[rstest::rstest] +#[case::coordinate(SortOrderArg::Coordinate, "low", 100, "high", 300)] +#[case::queryname(SortOrderArg::Queryname, "nameA", 100, "nameB", 100)] +#[case::queryname_natural(SortOrderArg::QuerynameNatural, "read1", 100, "read2", 100)] +#[case::template_coordinate(SortOrderArg::TemplateCoordinate, "low", 100, "high", 300)] +fn test_merge_streaming_verify_accepts_sorted_for_each_order( + #[case] order: SortOrderArg, + #[case] name0: &str, + #[case] pos0: i32, + #[case] name1: &str, + #[case] pos1: i32, +) { + let dir = TempDir::new().unwrap(); + let input = dir.path().join("sorted.bam"); + write_bam( + &input, + &bare_header(), + &[mapped_record(name0.as_bytes(), pos0), mapped_record(name1.as_bytes(), pos1)], + ); + let out = dir.path().join("merged.bam"); + + merge(&out, vec![input], order).expect("a correctly-ordered input must merge"); + assert!(out.exists()); + + let got = read_name_pos(&out); + assert_eq!( + got, + vec![ + (name0.to_string(), usize::try_from(pos0).unwrap()), + (name1.to_string(), usize::try_from(pos1).unwrap()), + ], + "merged records must retain their identity and input order" + ); +}