From 44b0e4bbd0c25f422435435717f06ac7852f74a1 Mon Sep 17 00:00:00 2001 From: James Kane Date: Sun, 16 Aug 2026 04:36:28 -0500 Subject: [PATCH 1/2] refactor(realign): one rule, one target constant, and the docs back on their own items MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A four-angle review of the realignment surface, prompted by two bugs that both traced to the same rule existing in several copies. **The duplication that caused those bugs was still there.** `DEFAULT_TARGET_BUILD` was exported so the UI could name the build it offers, and then only the Simple mode modal used it: both Advanced cards still carried the literal, and one compared builds with `eq_ignore_ascii_case` where the job uses `builds_match`, which trims. A stored " chm13v2.0" was offered by the card and refused by the job. `is_target_build` is now `pub` and the UI asks it instead of spelling out a comparison. The per-alignment card was also missing the `bam_path` condition, so it offered to re-map rows with no file. **Four doc comments documented the wrong item** — a new item inserted between an existing doc block and its function, four times, twice on public API. The worst had one sentence severed across two items and its tail orphaned twenty lines below. **Measured wins, benchmarked rather than guessed.** The revert's FASTQ writer issued seven `write_all` calls per read into the gzip encoder, paying the encoder's per-call overhead seven times: 1,882 ns/record against 539 for the same bytes assembled once, ~13 minutes of CPU per realignment. `realignable_in_project` was an N+1 over members where the grouped query already existed: 2.7 ms against 17.7 ms on a 2,504-member project, twice per batch. **And one bug, fixed at its own altitude.** Simple mode's running card matched on alignment id, and once the offer was gone it had nothing to match against — so it claimed *any* running job, and a page open on one person announced another person's realignment as theirs. `RealignState` now carries the owning subject, resolved once per job by `App::subject_of_alignment`, and the card matches on that. Tightening the old predicate would have broken the Done card instead; the ownership question needed answering where the mapping lives. Also: `preflight`/`resume_preflight` were the same function twice, `discard` had two implementations, the project card cloned a Vec per frame to read its length, finished states carried progress fields nobody reads, and a user-visible string held an 18-space run. 983 tests, clippy clean, fmt clean. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/postprocess/sort.rs | 6 +- .../navigator-analysis/src/revert/writer.rs | 24 +++++--- crates/navigator-app/src/commands.rs | 4 +- crates/navigator-app/src/lib.rs | 2 +- crates/navigator-app/src/realign.rs | 61 ++++++++++++++++--- crates/navigator-app/src/realign_job.rs | 40 +++++------- crates/navigator-ui/src/ui/detail.rs | 17 +++--- crates/navigator-ui/src/ui/events.rs | 11 +++- crates/navigator-ui/src/ui/mod.rs | 9 ++- crates/navigator-ui/src/ui/modals.rs | 13 ++-- crates/navigator-ui/src/ui/simple.rs | 15 ++--- crates/navigator-ui/src/ui/sources.rs | 13 +++- crates/navigator-ui/src/worker.rs | 11 ++++ 13 files changed, 146 insertions(+), 80 deletions(-) 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/writer.rs b/crates/navigator-analysis/src/revert/writer.rs index 918b4356..e0bf9dc2 100644 --- a/crates/navigator-analysis/src/revert/writer.rs +++ b/crates/navigator-analysis/src/revert/writer.rs @@ -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/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..17bdb039 100644 --- a/crates/navigator-app/src/realign_job.rs +++ b/crates/navigator-app/src/realign_job.rs @@ -453,9 +453,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 +625,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 +664,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-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() }, From a1c81dc70910e29b20ba619650601f2f78ca0715 Mon Sep 17 00:00:00 2001 From: James Kane Date: Sun, 16 Aug 2026 06:54:19 -0500 Subject: [PATCH 2/2] fix(realign): pace and count the mapper's writes, not just the sort's MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The write pacing and byte accounting that came out of the 2026-08-13 WindowServer teardown shipped as a module inside navigator-analysis. That crate holds the revert, sort, markdup and CRAM stages — but not the mapper, which lives in navigator-align, a leaf crate that cannot depend on it and should not. So half the pipeline was covered. The mapper's ~60 GB mapped.bam, the 8.93 GB minimizer index, and the final CRAM all went straight to the page cache, unpaced and uncounted. The phase-5 run log reads 0 MB/s through the longest stage in the job for exactly that reason: not a quiet stage, an unmeasured one. And pacing the sort while the mapper wrote unpaced left the original failure mode reachable, because the resource notice macOS filed was against the process, not against a stage. Move PacedFile, the byte counter and ResourceWatch into navigator-resource, a leaf crate both halves depend on. A counter only means something if there is exactly one of it. navigator-app takes it directly rather than reaching through navigator-analysis, since neither half is the owner. Also, from reading the same write path: - Every one of those writers sat behind BufWriter's 8 KB default while the encoders above hand down 64 KB BGZF blocks and multi-MB CRAM containers, so the buffer coalesced nothing. Now 1 MB, matching bamio. - Each stage output is synced once at the end. This matters most for mapped.bam, whose BGZF end-of-file block is what a resumed run reads to decide the file can be trusted — trusting a marker that was still a page-cache promise is how 59 GB was destroyed during phase 5 — and for the .mmi, whose atomic rename otherwise publishes contents the disk has not acknowledged. - The CRAM encoders took build_from_path, which opens the file themselves. build_from_writer, so the writer is ours to wrap. navigator-align carries the regression test: write a BAM through AlignmentWriter, assert the shared counter moved. It would have failed before this change. finalize.rs's .bai is left unpaced on purpose — a few MB, where the syscalls and the log noise would buy nothing. 984 tests pass (983 + the new one); clippy and fmt clean. Co-Authored-By: Claude Opus 5 (1M context) --- CLAUDE.md | 4 +- Cargo.lock | 11 +- Cargo.toml | 1 + crates/README.md | 2 + crates/navigator-align/Cargo.toml | 5 + crates/navigator-align/src/index.rs | 10 +- crates/navigator-align/src/output.rs | 111 +++++++++++++++--- crates/navigator-analysis/Cargo.toml | 11 +- crates/navigator-analysis/src/lib.rs | 1 - .../src/postprocess/bamio.rs | 2 +- .../src/postprocess/cram.rs | 22 +++- .../navigator-analysis/src/revert/collate.rs | 4 +- .../navigator-analysis/src/revert/writer.rs | 4 +- crates/navigator-app/Cargo.toml | 4 + crates/navigator-app/src/realign_job.rs | 25 ++-- crates/navigator-resource/Cargo.toml | 24 ++++ .../src/lib.rs} | 31 +++-- documents/design/realignment-module.md | 33 +++++- 18 files changed, 245 insertions(+), 60 deletions(-) create mode 100644 crates/navigator-resource/Cargo.toml rename crates/{navigator-analysis/src/resource.rs => navigator-resource/src/lib.rs} (89%) 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/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 e0bf9dc2..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, )) } 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/realign_job.rs b/crates/navigator-app/src/realign_job.rs index 17bdb039..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()); 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/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