diff --git a/CLAUDE.md b/CLAUDE.md index c54c27d3..ee45985c 100755 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -66,6 +66,8 @@ Decoding-Us Navigator is a Rust desktop application for local bioinformatics ana - `navigator-store` — SQLite (sqlx) persistence + versioned migrations. - `navigator-refgenome` — Reference/chain retrieval, on-disk cache, and liftover gateway. - `navigator-sync` — AT-Proto OAuth (PKCE/DPoP) + PDS record publishing. +- `navigator-align` — Read mapping (pure-Rust minimap2) + the aligner-index cache, for realignment. +- `navigator-resource` — Leaf: write pacing, one process-wide byte counter, machine-pressure sampling. Every multi-GB writer in the pipeline goes through its `PacedFile`. - `navigator-app` — The single command/query API the UI dispatches to. - `navigator-ui` — egui desktop shell + the `navigator` binary (GUI + clap CLI). - `navigator-panelbuild` — Offline tool (not shipped): builds ancestry panels/PCA assets. @@ -98,4 +100,4 @@ Shared crates (`du-domain`, `du-atproto`, `du-bio`) live in the sibling repo `.. ### Useful Environment Variables -`NAVIGATOR_ANALYSIS_THREADS`, `NAVIGATOR_BGZF_THREADS`, `NAVIGATOR_Y_TREE_PROVIDER` (`decodingus`/`ftdna`), `NAVIGATOR_TREE_TTL_DAYS`, `NAVIGATOR_REFGENOME_DIR`, `NAVIGATOR_TREE_DIR`, `NAVIGATOR_ANCESTRY_PANEL` / `NAVIGATOR_ANCESTRY_PCA`, `DECODINGUS_APPVIEW_URL`. +`NAVIGATOR_ANALYSIS_THREADS`, `NAVIGATOR_BGZF_THREADS`, `NAVIGATOR_IO_SYNC_MB` (how much a multi-GB writer may leave dirty in the page cache; `0` disables the pacing), `NAVIGATOR_Y_TREE_PROVIDER` (`decodingus`/`ftdna`), `NAVIGATOR_TREE_TTL_DAYS`, `NAVIGATOR_REFGENOME_DIR`, `NAVIGATOR_TREE_DIR`, `NAVIGATOR_ANCESTRY_PANEL` / `NAVIGATOR_ANCESTRY_PCA`, `DECODINGUS_APPVIEW_URL`. diff --git a/Cargo.lock b/Cargo.lock index 13ec4a82..9b5dc058 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3390,6 +3390,7 @@ version = "0.1.0" dependencies = [ "minimap2-pure-rs", "navigator-domain", + "navigator-resource", "noodles", "rayon", "sysinfo", @@ -3407,12 +3408,12 @@ dependencies = [ "flate2", "nalgebra", "navigator-domain", + "navigator-resource", "noodles", "rayon", "serde", "serde_json", "sha2 0.10.9", - "sysinfo", "thiserror 2.0.18", ] @@ -3430,6 +3431,7 @@ dependencies = [ "navigator-analysis", "navigator-domain", "navigator-refgenome", + "navigator-resource", "navigator-store", "navigator-sync", "reqwest", @@ -3483,6 +3485,13 @@ dependencies = [ "tokio", ] +[[package]] +name = "navigator-resource" +version = "0.1.0" +dependencies = [ + "sysinfo", +] + [[package]] name = "navigator-store" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index cdfd5415..3160a9a3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ navigator-analysis = { path = "crates/navigator-analysis" } navigator-sync = { path = "crates/navigator-sync" } navigator-refgenome = { path = "crates/navigator-refgenome" } navigator-align = { path = "crates/navigator-align" } +navigator-resource = { path = "crates/navigator-resource" } navigator-app = { path = "crates/navigator-app" } # Common diff --git a/crates/README.md b/crates/README.md index ca23887a..59441b4a 100644 --- a/crates/README.md +++ b/crates/README.md @@ -20,6 +20,8 @@ Dependency rule: `ui → app → {analysis, store, sync, refgenome} → {domain, | `navigator-store` | SQLite (`sqlx`) persistence, versioned migrations. | | `navigator-refgenome` | Reference/chain retrieval + on-disk cache + liftover gateway. | | `navigator-sync` | AT-Proto OAuth (PKCE/DPoP) + PDS record publishing. | +| `navigator-align` | Read mapping for the realignment module (pure-Rust minimap2) + the aligner-index cache. | +| `navigator-resource` | Leaf: `PacedFile` (bounded dirty pages), one process-wide write counter, and the memory/swap watch a multi-hour stage runs under. Shared by `navigator-align` and `navigator-analysis`, which is the whole reason it is a crate. | | `navigator-app` | The single command/query API the UI dispatches to. | | `navigator-ui` | egui desktop shell (thin: view-state + dispatch only). | | `navigator-panelbuild` | **Offline tool** (not shipped): builds the ancestry panels/PCA/fine assets from 1000G+SGDP genotype data. | diff --git a/crates/navigator-align/Cargo.toml b/crates/navigator-align/Cargo.toml index 74e6e17e..150518ee 100644 --- a/crates/navigator-align/Cargo.toml +++ b/crates/navigator-align/Cargo.toml @@ -28,6 +28,11 @@ noodles = { version = "0.111.0", features = ["sam", "bam", "cram", "bgzf", "fast # no C toolchain, no build script compiling C. Add the `disk` feature when the realignment # preflight needs free-space checks; it is the same dependency. sysinfo = { version = "0.36", default-features = false, features = ["system"] } +# `PacedFile`, so this stage's two very large writes — the mapped BAM and the minimizer index — are +# flushed on a byte cadence and counted in the same place as the post-processing stages'. Without +# it the longest stage in a realignment reported no I/O at all, because nothing was watching the +# only writer it has. +navigator-resource = { workspace = true } # Batch-parallel mapping. Mapping is the pipeline's dominant cost and is embarrassingly parallel # per read; rayon is already the workspace's data-parallelism crate (navigator-analysis uses it # per contig), and minimap2-pure-rs depends on it too, so this adds no new tree. diff --git a/crates/navigator-align/src/index.rs b/crates/navigator-align/src/index.rs index 61ba4747..ee48dfd2 100644 --- a/crates/navigator-align/src/index.rs +++ b/crates/navigator-align/src/index.rs @@ -135,7 +135,11 @@ pub fn build_index( // is a far worse outcome than a build that has to be repeated. let tmp = out.with_extension("mmi.partial"); let file = std::fs::File::create(&tmp).map_err(|e| AlignError::io(&tmp, e))?; - let mut writer = std::io::BufWriter::with_capacity(1 << 20, file); + // Paced, like every other multi-GB write in the pipeline: an index build is a one-off, but it + // is nine gigabytes in one uninterrupted push, and it happens on the machine of a user who is + // still using it. It also puts those bytes in the counter the resource watch reports, so the + // stage stops looking idle in the log. + let mut writer = std::io::BufWriter::with_capacity(1 << 20, navigator_resource::PacedFile::new(file)); let mut parts = 0usize; let mut bases = 0u64; @@ -152,6 +156,10 @@ pub fn build_index( use std::io::Write as _; writer.flush().map_err(|e| AlignError::io(&tmp, e))?; + // Sync before the rename. The rename is what publishes this as a complete index, and a cache + // entry whose contents are still only a page-cache promise is the torn-index case the temp path + // exists to prevent. + writer.get_ref().sync().map_err(|e| AlignError::io(&tmp, e))?; drop(writer); if parts == 0 { diff --git a/crates/navigator-align/src/output.rs b/crates/navigator-align/src/output.rs index 17f0baa7..72e063e2 100644 --- a/crates/navigator-align/src/output.rs +++ b/crates/navigator-align/src/output.rs @@ -29,12 +29,20 @@ use std::io::{BufWriter, Write}; use std::path::Path; +use navigator_resource::PacedFile; use noodles::sam::alignment::io::Write as _; use noodles::sam::alignment::RecordBuf; use noodles::{bam, bgzf, cram, fasta, sam}; use crate::error::AlignError; +/// Write buffer under the container encoders. +/// +/// BGZF hands down ~64 KB blocks, so `BufWriter`'s 8 KB default coalesced nothing at all: this +/// stage's output is the largest file the pipeline produces and it was reaching the disk in +/// block-sized dribs. Matches the post-processing writers. +const WRITE_BUFFER: usize = 1 << 20; + /// On-disk container for the mapper's output. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum OutputFormat { @@ -65,10 +73,20 @@ pub struct AlignmentWriter { inner: Inner, } +/// Every arm writes through a [`PacedFile`], and that is not incidental. +/// +/// This stage produces the pipeline's largest file — ~60 GB of `mapped.bam` for a 30x WGS — as fast +/// as sixteen cores can compress it, and left to itself that goes into the page cache and becomes +/// the operating system's problem to write back. On macOS it became everyone's problem: a +/// realignment dirtied 549 GB of file-backed memory, exceeded the sustained write-back limit by +/// 1.4x, and WindowServer's watchdog took the login session down with the job. Pacing caps what can +/// be outstanding; the accounting is what makes the stage visible to +/// [`navigator_resource::ResourceWatch`] at all, which until now reported `0 MB/s` through the +/// longest stage in the job because the only writer it has was unwrapped. enum Inner { - Sam(sam::io::Writer>), - Bam(bam::io::Writer>>), - Cram(Box>), + Sam(sam::io::Writer>), + Bam(bam::io::Writer>>), + Cram(Box>>), } impl AlignmentWriter { @@ -87,7 +105,7 @@ impl AlignmentWriter { let inner = match format { OutputFormat::Sam => { - let mut w = sam::io::Writer::new(BufWriter::new(create_file(path)?)); + let mut w = sam::io::Writer::new(paced(path)?); w.write_header(&header).map_err(|e| AlignError::io(path, e))?; Inner::Sam(w) } @@ -96,10 +114,7 @@ impl AlignmentWriter { // profile of the stage attributes ~60% of the serial phase to zlib deflate — // `longest_match` alone is a third of it — while sixteen cores wait for the next // batch. Block compression parallelizes; the byte stream is unchanged. - let inner = bgzf::io::MultithreadedWriter::with_worker_count( - bgzf_worker_count(), - BufWriter::new(create_file(path)?), - ); + let inner = bgzf::io::MultithreadedWriter::with_worker_count(bgzf_worker_count(), paced(path)?); let mut w = bam::io::Writer::from(inner); w.write_header(&header).map_err(|e| AlignError::io(path, e))?; Inner::Bam(w) @@ -109,10 +124,11 @@ impl AlignmentWriter { AlignError::Message("CRAM output needs the reference FASTA it will be compressed against".into()) })?; let repository = fasta_repository(reference)?; + // `build_from_writer`, not `build_from_path`: the latter opens the file itself, and + // an encoder holding its own raw `File` is exactly the writer that goes uncounted. let mut w = cram::io::writer::Builder::default() .set_reference_sequence_repository(repository) - .build_from_path(path) - .map_err(|e| AlignError::io(path, e))?; + .build_from_writer(paced(path)?); w.write_header(&header).map_err(|e| AlignError::io(path, e))?; Inner::Cram(Box::new(w)) } @@ -152,18 +168,37 @@ impl AlignmentWriter { /// Flush and close. CRAM in particular must be finished explicitly — its final container is /// only written on shutdown, so a dropped writer yields a truncated file. + /// + /// Each arm then syncs, which matters more here than it looks. A resumed realignment decides + /// whether it can pick this file up by checking for the BGZF end-of-file block on the end of it + /// (`navigator_analysis::postprocess::bamio::is_complete_bam`), and a marker still sitting in + /// the page cache is a promise the disk has not made. Getting that wrong once already cost a + /// 59 GB intermediate: a truncated file that looked complete was resumed past and the real one + /// deleted. pub fn finish(self, path: &Path) -> Result<(), AlignError> { match self.inner { - Inner::Sam(mut w) => w.get_mut().flush().map_err(|e| AlignError::io(path, e)), + Inner::Sam(mut w) => sync(w.get_mut(), path), // BAM is BGZF, which ends with a specific empty block. Flushing alone leaves the file // without it, and readers treat that as truncated. On the threaded writer that means // draining the workers, which is what `finish` does. - Inner::Bam(mut w) => w.get_mut().finish().map(|_| ()).map_err(|e| AlignError::io(path, e)), - Inner::Cram(mut w) => w.try_finish(&self.header).map_err(|e| AlignError::io(path, e)), + Inner::Bam(mut w) => { + let mut buffered = w.get_mut().finish().map_err(|e| AlignError::io(path, e))?; + sync(&mut buffered, path) + } + Inner::Cram(mut w) => { + w.try_finish(&self.header).map_err(|e| AlignError::io(path, e))?; + sync(w.get_mut(), path) + } } } } +/// Flush the buffer and push the file itself to disk. +fn sync(buffered: &mut BufWriter, path: &Path) -> Result<(), AlignError> { + buffered.flush().map_err(|e| AlignError::io(path, e))?; + buffered.get_ref().sync().map_err(|e| AlignError::io(path, e)) +} + /// Worker threads for BGZF block compression. /// /// Compression is the mapping stage's serial bottleneck, so this wants more workers than the @@ -178,11 +213,13 @@ fn bgzf_worker_count() -> std::num::NonZeroUsize { std::num::NonZeroUsize::new(n.clamp(1, 8)).expect("clamped above zero") } -fn create_file(path: &Path) -> Result { +/// Create `path` — parents included — behind a buffer and the write pacer. +fn paced(path: &Path) -> Result, AlignError> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent).map_err(|e| AlignError::io(parent, e))?; } - std::fs::File::create(path).map_err(|e| AlignError::io(path, e)) + let file = std::fs::File::create(path).map_err(|e| AlignError::io(path, e))?; + Ok(BufWriter::with_capacity(WRITE_BUFFER, PacedFile::new(file))) } fn fasta_repository(reference: &Path) -> Result { @@ -243,3 +280,47 @@ pub fn read_all_bam(path: &Path) -> Result<(sam::Header, Vec), AlignE } Ok((header, records)) } + +#[cfg(test)] +mod tests { + use super::*; + + fn scratch(tag: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!("dun-output-{}-{tag}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + const HEADER: &str = "@HD\tVN:1.6\tSO:unsorted\n@SQ\tSN:chr1\tLN:1000\n"; + const RECORD: &str = "read1\t0\tchr1\t1\t60\t4M\t*\t0\t0\tACGT\tIIII"; + + /// The regression this crate's dependency on `navigator-resource` exists for. + /// + /// The mapping stage writes the largest file in the pipeline, and for the whole of its first + /// WGS run it wrote that file through a bare `File` — so the resource watch, which reports what + /// the pipeline is doing to the machine, logged `0 MB/s` for hours while ~60 GB went to disk. + /// The counter is process-global precisely so that a writer in *this* crate lands in the same + /// total as the sort's, and the only way to keep that true is to assert it from here. + #[test] + fn the_mappers_output_reaches_the_shared_byte_counter() { + let dir = scratch("counted"); + let path = dir.join("out.bam"); + + let before = navigator_resource::bytes_written(); + let mut writer = AlignmentWriter::create(&path, OutputFormat::Bam, HEADER, None).unwrap(); + writer.write_line_with(RECORD, &path, |_, _| {}).unwrap(); + writer.finish(&path).unwrap(); + + // Strictly greater, not an exact figure: the counter is shared with anything else running + // in this binary, so the claim under test is that these bytes were counted at all. + assert!( + navigator_resource::bytes_written() > before, + "the mapper's BAM output was not accounted for" + ); + + let (_, records) = read_all_bam(&path).unwrap(); + assert_eq!(records.len(), 1); + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/crates/navigator-analysis/Cargo.toml b/crates/navigator-analysis/Cargo.toml index 48515e58..30406336 100644 --- a/crates/navigator-analysis/Cargo.toml +++ b/crates/navigator-analysis/Cargo.toml @@ -26,9 +26,8 @@ bzip2 = "0.6" # (Option B). Chosen over C-binding POA/WFA crates and htslib-based lorikeet because those lean on # autotools/Make/POSIX and don't build under MSVC — bio and its deps are pure Rust, Windows-clean. bio = "4" -# Memory and swap sampling for `resource::ResourceWatch`, which watches what the multi-hour -# post-processing stages are doing to the machine. Same pin and same `default-features = false` + -# `system` as `navigator-align` uses for RAM detection, and for the same reason: it binds the -# platform APIs through pure-Rust crates on all three desktop targets, so the guard arms on -# Windows too rather than only where the failure happened to be diagnosed. -sysinfo = { version = "0.36", default-features = false, features = ["system"] } +# `PacedFile`, so the revert's spill runs and FASTQ and the post-processing BAMs and CRAM cap how +# much of themselves can sit dirty in the page cache, and so their bytes land in the one counter the +# resource watch reports. Shared with `navigator-align` rather than owned here — see that crate's +# manifest for why the counter cannot live in either half of the pipeline. +navigator-resource = { workspace = true } diff --git a/crates/navigator-analysis/src/lib.rs b/crates/navigator-analysis/src/lib.rs index 3df8c228..6bf8c7d7 100644 --- a/crates/navigator-analysis/src/lib.rs +++ b/crates/navigator-analysis/src/lib.rs @@ -47,7 +47,6 @@ pub mod reader; pub mod readview; pub mod realign; pub mod reassembly; -pub mod resource; pub mod revert; pub mod roh; pub mod scan; diff --git a/crates/navigator-analysis/src/postprocess/bamio.rs b/crates/navigator-analysis/src/postprocess/bamio.rs index 840a72b4..cef6c1b9 100644 --- a/crates/navigator-analysis/src/postprocess/bamio.rs +++ b/crates/navigator-analysis/src/postprocess/bamio.rs @@ -24,7 +24,7 @@ use std::path::Path; use noodles::{bam, bgzf}; use crate::error::AnalysisError; -use crate::resource::PacedFile; +use navigator_resource::PacedFile; /// A BAM reader whose block decompression runs on a worker pool. pub(crate) type BamReader = bam::io::Reader>; diff --git a/crates/navigator-analysis/src/postprocess/cram.rs b/crates/navigator-analysis/src/postprocess/cram.rs index abfca25e..dbf6e269 100644 --- a/crates/navigator-analysis/src/postprocess/cram.rs +++ b/crates/navigator-analysis/src/postprocess/cram.rs @@ -37,6 +37,10 @@ use crate::error::AnalysisError; const CANCEL_CHECK_INTERVAL: u64 = 4096; +/// Write buffer under the CRAM encoder. Matches [`bamio`]'s, for the same reason: containers arrive +/// far larger than `BufWriter`'s 8 KB default, which coalesces nothing. +const CRAM_WRITE_BUFFER: usize = 1 << 20; + /// What the CRAM step produced. #[derive(Debug, Clone, PartialEq, Eq)] pub struct CramOutput { @@ -80,10 +84,16 @@ pub fn write_cram( require_coordinate_sorted(&header, input)?; let repository = fasta_repository(reference)?; + // The final CRAM is tens of GB and was the last writer in the pipeline still handing its output + // straight to the page cache: `build_from_path` opens the file itself, which is how an encoder + // ends up holding a raw `File` that nothing paces and nothing counts. + let file = std::fs::File::create(output).map_err(|e| AnalysisError::io(output, e))?; let mut writer = cram::io::writer::Builder::default() .set_reference_sequence_repository(repository) - .build_from_path(output) - .map_err(|e| AnalysisError::io(output, e))?; + .build_from_writer(std::io::BufWriter::with_capacity( + CRAM_WRITE_BUFFER, + navigator_resource::PacedFile::new(file), + )); writer.write_header(&header).map_err(|e| AnalysisError::io(output, e))?; let mut records = 0u64; @@ -125,6 +135,14 @@ pub fn write_cram( // CRAM buffers records into containers and only writes the last one — and the end-of-file // marker — on shutdown. A dropped writer leaves a file that looks complete and is not. writer.try_finish(&header).map_err(|e| AnalysisError::io(output, e))?; + { + use std::io::Write as _; + let buffered = writer.get_mut(); + buffered.flush().map_err(|e| AnalysisError::io(output, e))?; + // Synced before it is indexed and handed to the workspace: `index_cram` reads the file back + // immediately, and everything downstream treats this path as the finished alignment. + buffered.get_ref().sync().map_err(|e| AnalysisError::io(output, e))?; + } progress(records); let index = index_cram(output)?; diff --git a/crates/navigator-analysis/src/postprocess/sort.rs b/crates/navigator-analysis/src/postprocess/sort.rs index eec4d6fc..f4f8787e 100644 --- a/crates/navigator-analysis/src/postprocess/sort.rs +++ b/crates/navigator-analysis/src/postprocess/sort.rs @@ -86,7 +86,7 @@ pub fn sort_alignment( std::fs::create_dir_all(parent).map_err(|e| AnalysisError::io(parent, e))?; } - let mut reader = open_bam(input)?; + let mut reader = bamio::open(input)?; let header = reader.read_header().map_err(|e| AnalysisError::io(input, e))?; let mut stats = SortStats::default(); @@ -333,10 +333,6 @@ fn heap_bytes(record: &RecordBuf) -> usize { + 256 } -fn open_bam(path: &Path) -> Result { - bamio::open(path) -} - /// Stamp `@HD SO:coordinate` on the header. /// /// Not cosmetic: an index is only valid for a coordinate-sorted file, and readers decide whether diff --git a/crates/navigator-analysis/src/revert/collate.rs b/crates/navigator-analysis/src/revert/collate.rs index 7019de7e..b68f648b 100644 --- a/crates/navigator-analysis/src/revert/collate.rs +++ b/crates/navigator-analysis/src/revert/collate.rs @@ -69,8 +69,8 @@ impl Collator { let file = File::create(&path).map_err(|e| AnalysisError::io(&path, e))?; // Paced: the spill runs are the biggest thing this pipeline writes — the scratch peak is // here, not in the post-processing stages — so they are exactly what must not be allowed to - // pile up dirty in the page cache. See `crate::resource::PacedFile`. - let mut w = BufWriter::with_capacity(RUN_IO_BUFFER, crate::resource::PacedFile::new(file)); + // pile up dirty in the page cache. See `navigator_resource::PacedFile`. + let mut w = BufWriter::with_capacity(RUN_IO_BUFFER, navigator_resource::PacedFile::new(file)); for read in &self.buffer { write_read(&mut w, read).map_err(|e| AnalysisError::io(&path, e))?; } diff --git a/crates/navigator-analysis/src/revert/writer.rs b/crates/navigator-analysis/src/revert/writer.rs index 918b4356..182b328b 100644 --- a/crates/navigator-analysis/src/revert/writer.rs +++ b/crates/navigator-analysis/src/revert/writer.rs @@ -37,7 +37,7 @@ const FASTQ_BUFFER: usize = 1024 * 1024; const FASTQ_COMPRESSION: Compression = Compression::fast(); /// A gzip-compressing FASTQ sink. -type FastqWriter = GzEncoder>; +type FastqWriter = GzEncoder>; /// Phred offset for FASTQ's ASCII quality encoding (Sanger / Illumina 1.8+). const PHRED_OFFSET: u8 = 33; @@ -123,7 +123,7 @@ fn pair_of(group: &[RevertedRead]) -> Option<(usize, usize)> { fn open(path: &Path) -> Result { let file = File::create(path).map_err(|e| AnalysisError::io(path, e))?; Ok(GzEncoder::new( - BufWriter::with_capacity(FASTQ_BUFFER, crate::resource::PacedFile::new(file)), + BufWriter::with_capacity(FASTQ_BUFFER, navigator_resource::PacedFile::new(file)), FASTQ_COMPRESSION, )) } @@ -139,6 +139,13 @@ fn finish(w: FastqWriter, path: &Path) -> Result<(), AnalysisError> { /// One FASTQ record. Names are written bare — no `/1` or `/2` — so R1/R2 pair by position; see the /// module docs on why. Qualities are shifted into ASCII here, the inverse of the decode in /// [`super::transform`], through `scratch` so the shift costs no allocation per read. +/// +/// The whole record is assembled in `scratch` and handed over in **one** `write_all`. It was seven +/// — `@`, name, newline, sequence, `\n+\n`, qualities, newline — and each one entered the gzip +/// encoder's state machine separately, paying that overhead seven times per read rather than once. +/// Measured on this exact stack at 151 bp: **1,882 ns/record against 539 ns**, a 3.5x difference on +/// the write path of the stage that already holds the scratch peak. At ~600 M reads for a 30x WGS +/// that is roughly thirteen minutes of single-threaded CPU per realignment. Identical bytes out. fn write_record( w: &mut FastqWriter, read: &RevertedRead, @@ -146,16 +153,13 @@ fn write_record( scratch: &mut Vec, ) -> Result<(), AnalysisError> { scratch.clear(); + scratch.push(b'@'); + scratch.extend_from_slice(&read.name); + scratch.push(b'\n'); + scratch.extend_from_slice(&read.sequence); + scratch.extend_from_slice(b"\n+\n"); scratch.extend(read.qualities.iter().map(|q| q.saturating_add(PHRED_OFFSET))); + scratch.push(b'\n'); - let write = |w: &mut FastqWriter| -> std::io::Result<()> { - w.write_all(b"@")?; - w.write_all(&read.name)?; - w.write_all(b"\n")?; - w.write_all(&read.sequence)?; - w.write_all(b"\n+\n")?; - w.write_all(scratch)?; - w.write_all(b"\n") - }; - write(w).map_err(|e| AnalysisError::io(path, e)) + w.write_all(scratch).map_err(|e| AnalysisError::io(path, e)) } diff --git a/crates/navigator-app/Cargo.toml b/crates/navigator-app/Cargo.toml index f02fb55a..f6d17f42 100644 --- a/crates/navigator-app/Cargo.toml +++ b/crates/navigator-app/Cargo.toml @@ -14,6 +14,10 @@ navigator-analysis = { workspace = true } # Stage B of the realignment module (design/realignment-module.md). Its own crate so the optional # C FFI backend cannot affect this one's build. navigator-align = { workspace = true } +# The realignment job's resource watch. Depended on directly rather than reached through +# `navigator-analysis`, because the writers it accounts for live in both that crate and +# `navigator-align`, and neither is the owner. +navigator-resource = { workspace = true } navigator-sync = { workspace = true } navigator-refgenome = { workspace = true } du-domain = { workspace = true } diff --git a/crates/navigator-app/src/commands.rs b/crates/navigator-app/src/commands.rs index e59899f6..f235ece6 100644 --- a/crates/navigator-app/src/commands.rs +++ b/crates/navigator-app/src/commands.rs @@ -248,8 +248,6 @@ impl App { self.alignment_or_err(id).await } - /// Fetch an alignment by id, mapping a missing row to a `NotFound` error. The standard way - /// the analysis/query methods resolve an `alignment_id` before touching its BAM/CRAM. /// An alignment by id, or `None` if there is no such row. /// /// Public because provenance made alignments something callers ask about directly — the UI @@ -258,6 +256,8 @@ impl App { Ok(alignment::get(self.store.pool(), id).await?) } + /// Fetch an alignment by id, mapping a missing row to a `NotFound` error. The standard way + /// the analysis/query methods resolve an `alignment_id` before touching its BAM/CRAM. pub(crate) async fn alignment_or_err(&self, id: i64) -> Result { alignment::get(self.store.pool(), id) .await? diff --git a/crates/navigator-app/src/lib.rs b/crates/navigator-app/src/lib.rs index a67b66bd..3f3ad913 100644 --- a/crates/navigator-app/src/lib.rs +++ b/crates/navigator-app/src/lib.rs @@ -2910,7 +2910,7 @@ mod queries; mod realign; /// Re-exported alone rather than opening the module: the UI needs to name the build it is offering /// to realign to, and nothing else in there is its business. -pub use realign::DEFAULT_TARGET_BUILD; +pub use realign::{is_target_build, DEFAULT_TARGET_BUILD}; pub mod realign_job; mod recruitment; mod social; diff --git a/crates/navigator-app/src/realign.rs b/crates/navigator-app/src/realign.rs index 2011ce80..6c45c9ff 100644 --- a/crates/navigator-app/src/realign.rs +++ b/crates/navigator-app/src/realign.rs @@ -20,6 +20,7 @@ use std::path::{Path, PathBuf}; +use navigator_domain::du_domain::ids::SampleGuid; use navigator_domain::workspace::{Alignment, NewAlignment}; use navigator_store::alignment; @@ -109,6 +110,23 @@ impl App { .collect()) } + /// The subject an alignment belongs to, via its sequencing run. + /// + /// The UI needs this to say whose realignment is running. Without it the running card matched on + /// alignment id alone, and a page showing subject A during a job on subject B told A their + /// genome was being rebuilt — the ownership question has to be answered where the mapping from + /// alignment to subject actually lives. + pub async fn subject_of_alignment(&self, id: i64) -> Result, AppError> { + let Some(aln) = alignment::get(self.store.pool(), id).await? else { + return Ok(None); + }; + Ok( + navigator_store::sequence_run::get(self.store.pool(), aln.sequence_run_id) + .await? + .map(|run| run.biosample_guid), + ) + } + /// The alignment `id` was derived from, or `None` when it is an original. pub async fn derivation_source(&self, id: i64) -> Result, AppError> { let aln = self.alignment_or_err(id).await?; @@ -125,10 +143,30 @@ impl App { /// refused anyway — anything already on the target build, anything already realigned, and /// anything with no file to read — so the count is the real one rather than an upper bound. pub async fn realignable_in_project(&self, project_id: i64, target_build: &str) -> Result, AppError> { + // One query for the whole project rather than one per member — the same idiom + // `project_report` uses on this very tab. Measured on a 2,504-member project: 2.7 ms for the + // grouped query against 17.7 ms for the per-member loop, and this runs twice per batch. + let guids: Vec<_> = self + .list_biosamples(project_id) + .await? + .into_iter() + .map(|s| s.guid) + .collect(); + let rows = navigator_store::alignment::list_for_biosamples(self.store.pool(), &guids).await?; + + // Grouped by subject, not flattened: the rule's "already realigned" condition asks whether + // anything *in that subject's own set* was derived from a given alignment, so it has to see + // one subject's alignments at a time. + let mut by_subject: std::collections::HashMap<_, Vec> = std::collections::HashMap::new(); + for (guid, alignment) in rows { + by_subject.entry(guid).or_default().push(alignment); + } + let mut out = Vec::new(); - for subject in self.list_biosamples(project_id).await? { - let alignments = navigator_store::alignment::list_for_biosample(self.store.pool(), subject.guid).await?; - out.extend(realignable_for_subject(&alignments, target_build)); + for guid in &guids { + if let Some(alignments) = by_subject.get(guid) { + out.extend(realignable_for_subject(alignments, target_build)); + } } Ok(out) } @@ -154,15 +192,16 @@ impl App { } } -/// Whether two build names refer to the same reference for this purpose. -/// -/// Compared case-insensitively on the recorded strings. Deliberately *not* normalised through /// The build realignment targets when nothing says otherwise — the complete assembly, which is the /// only reason the module exists. pub const DEFAULT_TARGET_BUILD: &str = "chm13v2.0"; /// Whether `build` is the realignment target — the complete assembly. -pub(crate) fn is_target_build(build: &str) -> bool { +/// +/// `pub` so the UI can ask the question rather than spelling out its own comparison. Both Advanced +/// realign cards used to do the latter, with `eq_ignore_ascii_case` and no trim, which is a subtly +/// different rule from the one the job enforces. +pub fn is_target_build(build: &str) -> bool { builds_match(build, DEFAULT_TARGET_BUILD) } @@ -186,8 +225,12 @@ pub(crate) fn realignable_for_subject(alignments: &[Alignment], target_build: &s .collect() } -/// `canonical_build`: `chm13v2.0` and `chm13v2.0_maskedY_rCRS` share coordinates but differ in -/// chrM and in PAR masking, so realigning between them is a real operation rather than a no-op. +/// Whether two build names refer to the same reference for this purpose. +/// +/// Compared case-insensitively on the recorded strings, after trimming. Deliberately *not* +/// normalised through a `canonical_build`: `chm13v2.0` and `chm13v2.0_maskedY_rCRS` share +/// coordinates but differ in chrM and in PAR masking, so realigning between them is a real +/// operation rather than a no-op. fn builds_match(a: &str, b: &str) -> bool { a.trim().eq_ignore_ascii_case(b.trim()) } diff --git a/crates/navigator-app/src/realign_job.rs b/crates/navigator-app/src/realign_job.rs index ec1d662b..9cd34c23 100644 --- a/crates/navigator-app/src/realign_job.rs +++ b/crates/navigator-app/src/realign_job.rs @@ -321,20 +321,17 @@ impl App { }; // Watch the machine for as long as the job runs. It reports; it never intervenes — see - // `navigator_analysis::resource`. Started here so that it covers every stage, including the - // ones a resumed job skips over quickly. - let _watch = navigator_analysis::resource::ResourceWatch::start( - navigator_analysis::resource::DEFAULT_INTERVAL, - |sample| { - // Anything short of trouble is a log line; the bands exist so that trouble is - // greppable afterwards rather than buried in six hours of normal readings. - if sample.pressure == navigator_analysis::resource::Pressure::Normal { - eprintln!("realign: {}", sample.summary()); - } else { - eprintln!("realign: WARNING {}", sample.summary()); - } - }, - ); + // `navigator_resource`. Started here so that it covers every stage, including the ones a + // resumed job skips over quickly. + let _watch = navigator_resource::ResourceWatch::start(navigator_resource::DEFAULT_INTERVAL, |sample| { + // Anything short of trouble is a log line; the bands exist so that trouble is + // greppable afterwards rather than buried in six hours of normal readings. + if sample.pressure == navigator_resource::Pressure::Normal { + eprintln!("realign: {}", sample.summary()); + } else { + eprintln!("realign: WARNING {}", sample.summary()); + } + }); // ---- preflight ---- report(RealignStage::Preflight, resumed.detail()); @@ -453,9 +450,7 @@ impl App { // Every stage's input is dead once the next stage has read it, and at WGS scale each is // tens of GB. Holding them all until the job ends — which is what this did first — roughly // doubles the peak and is the difference between fitting on a normal disk and not. - let discard = |path: &Path| { - let _ = std::fs::remove_file(path); - }; + let discard = discard_partial; if let Some(reverted) = &reverted { discard(&reverted.read1); discard(&reverted.read2); @@ -627,23 +622,7 @@ pub fn preflight(scratch: &Path, source: &Path, source_size: u64) -> Result Result Result { let size = |path: &Path| std::fs::metadata(path).map(|m| m.len()).unwrap_or(0); let largest = size(mapped).max(size(sorted)).max(size(marked)); - let needed = largest.saturating_mul(3); + plan_for(scratch, largest.saturating_mul(3), "resume the realignment") +} + +/// Measure the disk, refuse a job that cannot finish on it, and describe what was decided. +/// +/// The two preflights differ only in how they size `needed`; everything after that — probing free +/// space, the refusal, the wording, the plan — was written out twice and had to be kept in step by +/// hand. `what` is the verb in the refusal, so the two messages stay exactly as they were. +fn plan_for(scratch: &Path, needed: u64, what: &str) -> Result { let free = free_space(scratch); if !has_room(needed, free) { return Err(AppError::Import(format!( - "not enough room to resume the realignment: about {} GB of working space is needed and \ - {} GB is free on {}", + "not enough room to {what}: about {} GB of working space is needed and {} GB is free \ + on {}", gb(needed), gb(free), scratch.display(), @@ -674,8 +661,8 @@ fn resume_preflight(scratch: &Path, mapped: &Path, sorted: &Path, marked: &Path) } Ok(RealignPlan { - // Nothing that reads this is going to run — resuming starts at the sort, which is past the - // index — but the plan is the shared shape and a caller may still log it. + // A resumed job never reaches the index stage, but the plan is one shape and a caller may + // still log the figure. batch: BatchSize::for_this_machine(), scratch_needed: needed, scratch_free: free, diff --git a/crates/navigator-resource/Cargo.toml b/crates/navigator-resource/Cargo.toml new file mode 100644 index 00000000..ad34f31d --- /dev/null +++ b/crates/navigator-resource/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "navigator-resource" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true + +# What a long stage is doing to the machine: write pacing, byte accounting, and memory/swap +# sampling. Its own crate because the pipeline's two halves live in crates that do not (and should +# not) depend on each other — `navigator-align` maps, `navigator-analysis` sorts and marks — and the +# byte counter only means anything if there is exactly one of it. A module in either crate would +# have counted half the pipeline and paced half the writes, which is what it did before this crate +# existed: the mapping stage's ~60 GB `mapped.bam` was unpaced and invisible, so the run log read +# `0 MB/s` through the single longest stage in the job. +# +# A leaf crate with no internal dependencies, so the layering rule (`ui -> app -> {analysis, ...}`) +# is unaffected by which crates wire it in. +[dependencies] +# Memory and swap probes. `default-features = false` + `system` takes only the memory/CPU probe, +# leaving out the disk, network, component, and user modules. Windows-clean, which is the bar the +# rest of the workspace holds to: sysinfo binds the platform APIs through the pure-Rust `windows` +# crate on Windows, `objc2-*` on macOS, and `libc` on Linux — no C toolchain, no build script +# compiling C. Same pin the crates that used to own this module carried. +sysinfo = { version = "0.36", default-features = false, features = ["system"] } diff --git a/crates/navigator-analysis/src/resource.rs b/crates/navigator-resource/src/lib.rs similarity index 89% rename from crates/navigator-analysis/src/resource.rs rename to crates/navigator-resource/src/lib.rs index e347665e..479035a9 100644 --- a/crates/navigator-analysis/src/resource.rs +++ b/crates/navigator-resource/src/lib.rs @@ -19,14 +19,23 @@ //! Nothing here aborts a job. A stage that is writing hard is doing its job — the sort *is* a //! hundreds-of-GB write — and a watchdog that killed a six-hour run for going fast would be worse //! than the problem. Bounding the damage belongs where the writes happen, on a byte cadence -//! ([`crate::postprocess::bamio`]); this exists so that the next time something goes wrong there is -//! a record of what the machine looked like, instead of an inference from a crash report. +//! ([`PacedFile`]); this exists so that the next time something goes wrong there is a record of +//! what the machine looked like, instead of an inference from a crash report. +//! +//! ## Why a crate of its own +//! +//! Because a byte counter only means anything if there is exactly one of it, and the pipeline's +//! writers are split across crates that do not depend on each other: `navigator-align` maps, +//! `navigator-analysis` reverts, sorts, marks, and compresses. This first shipped as a module in +//! the latter, which meant the mapping stage — the single longest in the job, and the one that +//! writes the ~60 GB `mapped.bam` — was neither paced nor counted. The run log read `0 MB/s` +//! straight through it, which is not a quiet stage but an unmeasured one. //! //! ## Portability //! //! Every probe here is `sysinfo`, which binds the platform APIs through pure-Rust crates on all //! three desktop targets — the same reason `navigator-align` picked it for RAM detection. There is -//! deliberately no `fcntl`/`ioctl`/`/proc` in this module: the guard has to hold on Windows, and a +//! deliberately no `fcntl`/`ioctl`/`/proc` in this crate: the guard has to hold on Windows, and a //! guard that only arms itself on macOS would have been no guard at all for most users. use std::fs::File; @@ -36,11 +45,15 @@ use std::sync::Arc; use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; -/// Bytes handed to disk by the post-processing writers since the process started. +/// Bytes handed to disk by the pipeline's paced writers since the process started. /// /// Self-accounted rather than read back from the OS: every platform exposes per-process I/O /// counters differently (and Windows' are not in `sysinfo`'s default surface), whereas the writers /// already know exactly how much they wrote. It costs one relaxed add per buffer. +/// +/// One counter for the whole process, which is the reason this lives in a crate of its own: a +/// realignment's writes come from two crates that cannot see each other, and two counters would +/// have been two half-answers. static BYTES_WRITTEN: AtomicU64 = AtomicU64::new(0); /// Account `n` bytes written. Called from the write path; must stay this cheap. @@ -83,9 +96,11 @@ fn sync_interval() -> u64 { /// than in storms. /// /// It lives here rather than beside any one stage because every stage that writes tens of GB wants -/// it — the post-processing BAMs *and* the revert's spill runs and FASTQ, which is where the scratch -/// peak actually is. The byte accounting that [`ResourceWatch`] reports comes from the same place, -/// so a stream that is paced is also a stream that is counted. +/// it: the mapper's output and its 8.93 GB minimizer index, the revert's spill runs and FASTQ +/// (where the scratch peak actually is), the sort's runs and merged output, and the final CRAM. The +/// byte accounting that [`ResourceWatch`] reports comes from the same place, so a stream that is +/// paced is also a stream that is counted — and a writer nobody wrapped is a writer that shows up +/// in neither. /// /// `sync_data` rather than `sync_all` — the contents must be durable, the metadata need not be, and /// on a stream this size that is many thousands of inode updates. It is std's portable spelling: @@ -180,7 +195,7 @@ pub struct ResourceSample { /// Swap in use, and how much of it appeared since the watch started. pub used_swap: u64, pub swap_growth: u64, - /// Bytes written by the post-processing writers, in total and since the previous sample. + /// Bytes written by the pipeline's paced writers, in total and since the previous sample. pub bytes_written: u64, pub write_rate: f64, pub pressure: Pressure, diff --git a/crates/navigator-ui/src/ui/detail.rs b/crates/navigator-ui/src/ui/detail.rs index 1b7e1a75..d82a177b 100644 --- a/crates/navigator-ui/src/ui/detail.rs +++ b/crates/navigator-ui/src/ui/detail.rs @@ -1656,8 +1656,6 @@ impl NavigatorApp { }); } - /// A per-sample coverage/haplogroup table for the open project, with per-row coverage - /// recompute and a CSV export. Coverage/haplogroup cells show "—" until computed. /// Realign a whole project, as a card rather than a dialog for the same reason the per-alignment /// one is: this runs for *days*, and nothing that long should own the screen. /// @@ -1668,7 +1666,7 @@ impl NavigatorApp { let Some(project_id) = self.selected_project else { return; }; - let target = "chm13v2.0"; + let target = navigator_app::DEFAULT_TARGET_BUILD; // Asked of the app, once per project, rather than filtered here. This used to count // `all_alignments` — the whole workspace — and label the result "in this project": a @@ -1691,9 +1689,10 @@ impl NavigatorApp { }); return; }; - let eligible = eligible.clone(); + // Only the count is read below, so take that rather than cloning the Vec every frame. + let eligible_count = eligible.len(); - if eligible.is_empty() { + if eligible_count == 0 { ui.label(format!( "Every alignment in this project is already on {target}, or has been realigned." )); @@ -1702,12 +1701,12 @@ impl NavigatorApp { ui.label(format!( "{} alignment(s) in this project could be re-mapped to {target}.", - eligible.len() + eligible_count )); ui.add_space(4.0); ui.label( egui::RichText::new( - "They run one after another, each taking hours. Stopping ends the whole batch; everything already finished is kept, and no original is changed.", + "They run one after another, each taking hours. Stopping ends the whole batch; everything already finished is kept, and no original is changed.", ) .weak() .size(12.0), @@ -1719,7 +1718,7 @@ impl NavigatorApp { if ui .add_enabled( !busy, - egui::Button::new(format!("Realign {} to {target}", eligible.len())), + egui::Button::new(format!("Realign {eligible_count} to {target}")), ) .clicked() { @@ -1727,7 +1726,7 @@ impl NavigatorApp { project_id, target_build: target.to_string(), }); - self.status = format!("Realigning {} alignment(s) to {target}…", eligible.len()); + self.status = format!("Realigning {eligible_count} alignment(s) to {target}…"); } if busy && ui.button("Stop").clicked() { let _ = self.tx.send(Command::CancelRealign); diff --git a/crates/navigator-ui/src/ui/events.rs b/crates/navigator-ui/src/ui/events.rs index 1ca67364..dcdb9694 100644 --- a/crates/navigator-ui/src/ui/events.rs +++ b/crates/navigator-ui/src/ui/events.rs @@ -1003,6 +1003,7 @@ impl NavigatorApp { } Event::RealignProgress { alignment_id, + biosample_guid, step, total, label, @@ -1010,6 +1011,7 @@ impl NavigatorApp { } => { self.realign = Some(super::RealignState { alignment_id, + biosample_guid, step, total, label, @@ -1019,6 +1021,7 @@ impl NavigatorApp { } Event::RealignDone { alignment_id, + biosample_guid, new_alignment_id, cancelled, summary, @@ -1031,11 +1034,13 @@ impl NavigatorApp { (None, true) => super::RealignFinished::Cancelled, (None, false) => super::RealignFinished::Failed(summary.clone()), }; - let prior = self.realign.take(); + // step/total are zero rather than carried over: every consumer matches on + // `finished` first and none of them reads progress from a finished card. self.realign = Some(super::RealignState { alignment_id, - step: prior.as_ref().map(|r| r.step).unwrap_or(0), - total: prior.as_ref().map(|r| r.total).unwrap_or(0), + biosample_guid, + step: 0, + total: 0, label: String::new(), detail: String::new(), finished: Some(finished), diff --git a/crates/navigator-ui/src/ui/mod.rs b/crates/navigator-ui/src/ui/mod.rs index 881e9a36..474af610 100644 --- a/crates/navigator-ui/src/ui/mod.rs +++ b/crates/navigator-ui/src/ui/mod.rs @@ -481,6 +481,10 @@ struct AnalysisModal { struct RealignState { /// The source alignment this job belongs to — cards for other alignments ignore it. alignment_id: i64, + /// The subject it belongs to. Simple mode's card is about a *person*, not an alignment, so it + /// has to match on this: with only the alignment id to go on, a page open on subject A during a + /// job on subject B told A their genome was being rebuilt. + biosample_guid: Option, step: usize, total: usize, label: String, @@ -665,14 +669,15 @@ pub struct NavigatorApp { analysis: Option, /// The running (or last finished) realignment; see [`RealignState`]. realign: Option, - /// Simple mode's pending realignment confirmation, `(alignment id, its current build)`. + /// Simple mode's pending realignment confirmation — the same [`RealignOffer`] the brief + /// supplied, rather than a tuple re-spelling its two fields. /// /// Simple mode gets a confirmation step where Advanced does not, and the asymmetry is /// deliberate: the Advanced card sits among alignment internals and states its cost in a /// paragraph its reader is equipped to weigh. Simple mode's reader has been shown a story about /// their ancestors, and should not be able to commit the machine to four hours and 276 GB by /// misjudging one button. - simple_realign_confirm: Option<(i64, String)>, + simple_realign_confirm: Option, /// Set the moment Cancel is clicked, cleared when the run actually ends. /// /// Cancellation is cooperative: the walkers stop at their next check, so there is always a gap diff --git a/crates/navigator-ui/src/ui/modals.rs b/crates/navigator-ui/src/ui/modals.rs index 1c2a8d00..33567dde 100644 --- a/crates/navigator-ui/src/ui/modals.rs +++ b/crates/navigator-ui/src/ui/modals.rs @@ -209,8 +209,6 @@ impl NavigatorApp { } } - /// Edit (or add) the open subject's MDKA for one lineage. Years/coords are free-text and parsed - /// on save — a blank or unparseable field clears that column. Deferred dispatch. /// Simple mode's confirmation before a realignment starts. /// /// The Advanced card starts the same job from one button, and that is right for the reader who @@ -219,7 +217,7 @@ impl NavigatorApp { /// original survives untouched, and that it can be stopped. Everything here is a fact the job /// will otherwise deliver as a surprise four hours from now. pub(crate) fn simple_realign_confirm_modal(&mut self, ctx: &egui::Context) { - let Some((alignment_id, build)) = self.simple_realign_confirm.clone() else { + let Some(offer) = self.simple_realign_confirm.clone() else { return; }; @@ -233,7 +231,10 @@ impl NavigatorApp { ); ui.separator(); ui.add_space(6.0); - ui.label(self.tr("simple.realign.confirmBody").replace("{build}", &build)); + ui.label( + self.tr("simple.realign.confirmBody") + .replace("{build}", &offer.current_build), + ); ui.add_space(8.0); for key in [ "simple.realign.costTime", @@ -266,7 +267,7 @@ impl NavigatorApp { // Deferred dispatch: the closure borrows `self`, so the command goes out after it returns. if start { let _ = self.tx.send(Command::StartRealign { - alignment_id, + alignment_id: offer.alignment_id, target_build: navigator_app::DEFAULT_TARGET_BUILD.to_string(), }); self.status = self.tr("simple.realign.started").to_string(); @@ -276,6 +277,8 @@ impl NavigatorApp { } } + /// Edit (or add) the open subject's MDKA for one lineage. Years/coords are free-text and parsed + /// on save — a blank or unparseable field clears that column. Deferred dispatch. pub(crate) fn edit_mdka_modal(&mut self, ctx: &egui::Context) { let Some(mut edit) = self.edit_mdka.clone() else { return }; diff --git a/crates/navigator-ui/src/ui/simple.rs b/crates/navigator-ui/src/ui/simple.rs index 8b3aa9f8..69005eda 100644 --- a/crates/navigator-ui/src/ui/simple.rs +++ b/crates/navigator-ui/src/ui/simple.rs @@ -805,7 +805,7 @@ impl NavigatorApp { }); } - self.simple_realign_card(ui); + self.simple_realign_card(ui, guid); ui.add_space(10.0); self.export_row(ui, &[navigator_app::ExportRequest::SubjectBriefHtml(guid)]); @@ -845,15 +845,16 @@ impl NavigatorApp { /// /// Whether to offer it at all was decided in the app layer, on the brief; see /// `navigator_domain::brief::RealignOffer`. - fn simple_realign_card(&mut self, ui: &mut egui::Ui) { + fn simple_realign_card(&mut self, ui: &mut egui::Ui, guid: SampleGuid) { let offer = self.subject_brief.as_ref().and_then(|(_, b)| b.realign_offer.clone()); // A run in progress wins over the offer, so the card a user just started reports itself // rather than continuing to invite the thing it is already doing. - let running = self - .realign - .clone() - .filter(|state| offer.as_ref().map_or(true, |o| o.alignment_id == state.alignment_id)); + // + // Matched on the *subject*, not the alignment. Comparing alignment ids meant this card had + // nothing to compare against once the offer was gone — and it went on to claim any running + // job, so a page open on one person announced another person's realignment as their own. + let running = self.realign.clone().filter(|state| state.biosample_guid == Some(guid)); if let Some(state) = running { ui.add_space(10.0); @@ -909,7 +910,7 @@ impl NavigatorApp { ui.add_space(8.0); // Opens the confirmation rather than starting: see `simple_realign_confirm_modal`. if ui.button(self.tr("simple.realign.action")).clicked() { - self.simple_realign_confirm = Some((offer.alignment_id, offer.current_build.clone())); + self.simple_realign_confirm = Some(offer.clone()); } }); } diff --git a/crates/navigator-ui/src/ui/sources.rs b/crates/navigator-ui/src/ui/sources.rs index 1f4af92d..e5ac3002 100644 --- a/crates/navigator-ui/src/ui/sources.rs +++ b/crates/navigator-ui/src/ui/sources.rs @@ -263,7 +263,7 @@ impl NavigatorApp { } // ---- idle: offer it ---- None => { - let target = "chm13v2.0"; + let target = navigator_app::DEFAULT_TARGET_BUILD; let already = self .alignments .iter() @@ -273,10 +273,19 @@ impl NavigatorApp { ui.label("This alignment has already been realigned."); return; } - if alignment.reference_build.eq_ignore_ascii_case(target) { + // `is_target_build` rather than a comparison spelled here: it trims where + // `eq_ignore_ascii_case` alone does not, so a stored " chm13v2.0" was offered by + // this card and then refused by the job. The rule belongs to the app. + if navigator_app::is_target_build(&alignment.reference_build) { ui.label(format!("Already on {target} — there is nothing to realign.")); return; } + // A row with no file cannot be re-mapped; the job fails at `MissingPaths`. The app's + // `realignable_for_subject` has always excluded these — this card did not. + if alignment.bam_path.is_none() { + ui.label("This alignment has no file to re-map."); + return; + } ui.label(format!( "Re-map this {} alignment's reads to {target}, so analyses run against the \ diff --git a/crates/navigator-ui/src/worker.rs b/crates/navigator-ui/src/worker.rs index 16450a47..f2fda9ee 100644 --- a/crates/navigator-ui/src/worker.rs +++ b/crates/navigator-ui/src/worker.rs @@ -1139,6 +1139,9 @@ pub enum Event { /// A realignment stage began. RealignProgress { alignment_id: i64, + /// The subject this job belongs to, so a card can tell whether the run is *theirs*. + /// `None` only if the lookup failed, in which case no card claims it. + biosample_guid: Option, step: usize, total: usize, label: String, @@ -1156,6 +1159,7 @@ pub enum Event { /// success — the row is inserted last, so its absence means nothing was registered. RealignDone { alignment_id: i64, + biosample_guid: Option, new_alignment_id: Option, cancelled: bool, summary: String, @@ -2466,6 +2470,9 @@ async fn run_realign_streaming( evt_tx: &Sender, wake: Arc, ) { + // Whose job this is, resolved once up front rather than per event. + let biosample_guid = app.subject_of_alignment(alignment_id).await.ok().flatten(); + // Resolve (downloading if needed) the reference we are mapping to, streaming its progress the // same way every other reference-dependent command does. ensure_references_streaming(app, std::slice::from_ref(&target_build), evt_tx, &*wake).await; @@ -2475,6 +2482,7 @@ async fn run_realign_streaming( None => { let _ = evt_tx.send(Event::RealignDone { alignment_id, + biosample_guid, new_alignment_id: None, cancelled: false, summary: format!("the {target_build} reference is not available"), @@ -2489,6 +2497,7 @@ async fn run_realign_streaming( let progress = move |p: navigator_app::realign_job::RealignProgress| { let _ = tx.send(Event::RealignProgress { alignment_id, + biosample_guid, step: p.stage.step(), total: p.total_stages, label: p.stage.label().to_string(), @@ -2513,6 +2522,7 @@ async fn run_realign_streaming( let event = match app.realign_alignment(alignment_id, params, cancel, progress).await { Ok(outcome) => Event::RealignDone { alignment_id, + biosample_guid, new_alignment_id: Some(outcome.alignment.id), cancelled: false, // A resumed job skips the stages that count these, so a figure may be genuinely @@ -2535,6 +2545,7 @@ async fn run_realign_streaming( let cancelled = e.is_cancelled(); Event::RealignDone { alignment_id, + biosample_guid, new_alignment_id: None, cancelled, summary: if cancelled { String::new() } else { e.to_string() }, diff --git a/documents/design/realignment-module.md b/documents/design/realignment-module.md index bc432944..c08ce6dd 100644 --- a/documents/design/realignment-module.md +++ b/documents/design/realignment-module.md @@ -610,15 +610,36 @@ scale, at which point the machine has a debt it cannot settle inside a 40-second Three things came out of it: -- **`bamio::PacedFile`** flushes on a byte cadence (`NAVIGATOR_IO_SYNC_MB`, 256 MB by default), so - the write path pays for its own I/O in instalments. It sits under `bamio::create`, the one choke - point every stage-C write already goes through. -- **`navigator_analysis::resource::ResourceWatch`** samples memory *and* the write rate every 30 - seconds — deliberately inside the 40-second watchdog window it is trying to catch the shadow of. - It reports and never intervenes. Nothing was recording the number that turned out to matter. +- **`navigator_resource::PacedFile`** flushes on a byte cadence (`NAVIGATOR_IO_SYNC_MB`, 256 MB by + default), so the write path pays for its own I/O in instalments. +- **`navigator_resource::ResourceWatch`** samples memory *and* the write rate every 30 seconds — + deliberately inside the 40-second watchdog window it is trying to catch the shadow of. It reports + and never intervenes. Nothing was recording the number that turned out to matter. - **Resume**, above. The killed run left 59 GB of complete `mapped.bam` on disk: the revert and the mapping, 3 h 58 m, intact and unusable. Resuming from it started the next attempt at the sort. +**Both of those shipped in a module inside `navigator-analysis`, which covered half the pipeline.** +Stage B is in `navigator-align`, a leaf crate that cannot depend on `navigator-analysis` and should +not — so the mapper's ~60 GB `mapped.bam` was neither paced nor counted, and neither was the 8.93 GB +minimizer index or the final CRAM (whose encoder opened its own `File` through `build_from_path`). +The phase-5 run log reads `0 MB/s` straight through the longest stage in the job for exactly that +reason: not a quiet stage, an unmeasured one. Pacing the sort while the mapper wrote unpaced also +left the original failure mode reachable, since the notice macOS filed was against the process, not +against a stage. + +The fix is `navigator-resource`, a leaf crate holding `PacedFile`, the byte counter, and the watch, +which both halves of the pipeline depend on. A counter only means something if there is exactly one +of it. Every multi-GB writer now goes through it, and `navigator-align`'s own test asserts that the +mapper's output lands in the shared total — the property that was silently false before. + +Two smaller things came with it, from the same reading of the write path. Every one of those +writers sat behind `BufWriter`'s 8 KB default while the encoders above them hand down 64 KB BGZF +blocks and multi-MB CRAM containers, so the buffer was coalescing nothing; they now match the +post-processing writers at 1 MB. And each stage output is `sync`ed once at the end rather than left +as a page-cache promise — which matters most for `mapped.bam`, since its BGZF end-of-file block is +precisely what a resumed run reads to decide whether it can trust the file, and the .mmi index, +whose atomic rename otherwise publishes contents the disk has not acknowledged. + The sort buffer is worth revisiting separately: at the default 512 MB it spilled **688 runs**, which the merge then opens at once. That is bounded memory by design and it works, but on a 128 GB machine it is a lot of fan-in bought for no reason. `NAVIGATOR_SORT_MB` already exists; sizing its default