Pace and count the mapper's writes, not just the sort's

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`.

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

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. |

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.

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 {

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<BufWriter<std::fs::File>>),
-    Bam(bam::io::Writer<bgzf::io::MultithreadedWriter<BufWriter<std::fs::File>>>),
-    Cram(Box<cram::io::Writer<std::fs::File>>),
+    Sam(sam::io::Writer<BufWriter<PacedFile>>),
+    Bam(bam::io::Writer<bgzf::io::MultithreadedWriter<BufWriter<PacedFile>>>),
+    Cram(Box<cram::io::Writer<BufWriter<PacedFile>>>),
 }
 
 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<PacedFile>, 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<std::fs::File, AlignError> {
+/// Create `path` — parents included — behind a buffer and the write pacer.
+fn paced(path: &Path) -> Result<BufWriter<PacedFile>, 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<fasta::Repository, AlignError> {
@@ -243,3 +280,47 @@ pub fn read_all_bam(path: &Path) -> Result<(sam::Header, Vec<RecordBuf>), 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);
+    }
+}

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 }

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;

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<bgzf::io::MultithreadedReader<File>>;

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)?;

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::BamReader, AnalysisError> {
-    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