From 8be84918deb2aa1b6be0ec649685e37f2d9c1304 Mon Sep 17 00:00:00 2001 From: Logan Nye <87274608+logannye@users.noreply.github.com> Date: Tue, 2 Jun 2026 14:14:21 -0700 Subject: [PATCH] feat(columnkit): implement one trait, inherit the bounded contract (SDK) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A builder who wants their own per-locus metric (methylation, coverage/QC, ML features, a genotyper) would otherwise fork into the raw pileup engine and re-implement, by hand, the whole-genome contig walk, the working-set bound that plan/--enforce admit, and the canonical-JSON + BLAKE3 receipt — silently dropping the soundness invariants. ColumnKit turns "extend the substrate" into a first-class SDK: - ColumnAnalyzer trait (header / params / on_column) + run_bounded_whole_genome driver (src/call/columnkit.rs). Implement the trait, run it through the driver, inherit the SAME bounded per-contig stream + working-set bound + receipt the shipped subcommands enjoy. - The trait is WELDED to the kernel: the driver runs the exact same column stream as the shipped features egress, so the estimator that admits a run provably upper-bounds the realized working set of the builder's analyzer too. - features is now the FIRST ColumnAnalyzer: run_features is refactored onto the driver (FeatureAnalyzer), so the SDK IS the production path, not a parallel one — byte-identical output pinned by the golden TSV test. - examples/columnkit_coverage.rs: a ~25-line custom analyzer that runs and reports its inherited bounded working set. Crate-root re-exports + README. No other genomics library can offer "implement this trait, inherit a machine-checkable memory budget + a hash-verifiable receipt" — none has a memory contract to inherit. Tests: 2 columnkit unit tests (byte-identical-to-direct-path + custom-analyzer bounded WS); golden feature TSV guards the dogfood; feature writers take ?Sized writers. Full suite green; rustc 0 warnings; additions clippy-clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 22 +++ examples/columnkit_coverage.rs | 94 +++++++++++ src/call/columnkit.rs | 277 +++++++++++++++++++++++++++++++++ src/call/features.rs | 4 +- src/call/mod.rs | 2 + src/lib.rs | 9 +- src/main.rs | 33 ++-- 7 files changed, 419 insertions(+), 22 deletions(-) create mode 100644 examples/columnkit_coverage.rs create mode 100644 src/call/columnkit.rs diff --git a/README.md b/README.md index a9eb174..051be85 100644 --- a/README.md +++ b/README.md @@ -168,6 +168,27 @@ Two properties no other pileup gives you together: it is **bounded** (the whole- A dependency-light Python boundary ([`python/rosalind.py`](python/rosalind.py), stdlib + numpy) loads the table directly, and [`examples/reproducible_features_demo.py`](examples/reproducible_features_demo.py) trains a small model on it and *proves* the inputs are bit-reproducible (two independent extractions → matching receipt hashes → bit-identical trained weights; see [`docs/findings/2026-06-02-reproducible-features-demo.md`](docs/findings/2026-06-02-reproducible-features-demo.md)). *(TSV today; an Arrow/Parquet egress and a zero-copy `pyarrow` in-process binding are the next step.)* +## ColumnKit: implement one trait, inherit the contract + +Want your *own* per-locus metric — methylation, a coverage/QC track, custom ML features, a star-allele genotyper? Implement one trait and run it through the driver; you inherit the **same** bounded whole-genome walk, the **same** working-set bound that `plan`/`--enforce` admit, and the **same** verifiable receipt the shipped subcommands enjoy — without re-deriving any of it. + +```rust +use rosalind::{ColumnAnalyzer, run_bounded_whole_genome, PileupColumn}; +use std::io::{self, Write}; + +struct CoverageTrack; +impl ColumnAnalyzer for CoverageTrack { + fn header(&self) -> Option { Some("#contig\tpos\tdepth\n".into()) } + fn on_column(&mut self, col: &PileupColumn, contig: &str, out: &mut dyn Write) -> io::Result<()> { + writeln!(out, "{contig}\t{}\t{}", col.locus.pos.0 + 1, col.depth()) + } +} +// run_bounded_whole_genome(&mut CoverageTrack, source, &ref_view, contigs, params, &mut out) +// → returns the bounded WorkingSet `plan`/`--enforce` reason about. +``` + +The trait is *welded* to the bounded kernel — `run_bounded_whole_genome` drives the exact same column stream as the shipped `features` egress (which is itself just the first `ColumnAnalyzer`), so the estimator that admits a run provably upper-bounds the realized working set of *your* analyzer too. No other genomics library can offer "implement this trait, inherit a machine-checkable memory budget + a hash-verifiable receipt," because no other library has a memory contract to inherit. See [`examples/columnkit_coverage.rs`](examples/columnkit_coverage.rs). + ## Roadmap The core primitive is a streaming, CIGAR-aware pileup column stream; variant calling and custom plugins consume it. Performance work deliberately *follows* the unique capability — the target user needs "it fits and is predictable" before "it's fastest." @@ -177,6 +198,7 @@ The core primitive is a streaming, CIGAR-aware pileup column stream; variant cal - **Phase C (done):** memory as a *verifiable contract* — `rosalind plan` (a checkable envelope before you commit), `--enforce` (honor-or-refuse: refuse up front / fail loud, never a silent OOM-kill), and `rosalind verify`. See [CONTRACT.md](CONTRACT.md). - **Hardening & reach (done):** unbiased depth-cap downsampling (no silent variant drops) and a CI-enforced memory gate; **measured** germline detection accuracy ([Accuracy](#accuracy)); the **`rosalind features`** reproducible ML feature substrate; and a one-command adoption on-ramp — prebuilt binaries (`install.sh`, with checksum verification) plus the **Rosalind budget GitHub Action** (`action.yml`, used as `logannye/rosalind@v0.1.0`) that enforces the contract in *your* CI. - **Fleet scheduling (done):** [prediction → placement](#pack-a-fleet-prediction--placement) — `rosalind pack` proves a co-location of N calling jobs fits a node before launching a byte (predicted peaks are additive and read from the index header); `plan --index --json` for a scheduler to read. +- **ColumnKit SDK (done):** [implement one trait, inherit the contract](#columnkit-implement-one-trait-inherit-the-contract) — a `ColumnAnalyzer` trait + `run_bounded_whole_genome` driver so a builder's own per-locus analyzer inherits bounded memory, determinism, and a verifiable receipt. The shipped `features` egress is the first impl. - **Phase D (research):** sublinear-space index construction — the `~√t` space/time knob across the full curve — extending the contract to the index *build* step (today's build is O(reference)). The headline space-complexity bet; see [`docs/OPEN_PROBLEMS.md`](docs/OPEN_PROBLEMS.md). - **Later:** the aligner over the persisted multi-contig index (`align --index`, whole-genome alignment); germline indels and richer read QC; deterministic multithreading; a Python/tensor binding over the pileup stream. diff --git a/examples/columnkit_coverage.rs b/examples/columnkit_coverage.rs new file mode 100644 index 0000000..5523223 --- /dev/null +++ b/examples/columnkit_coverage.rs @@ -0,0 +1,94 @@ +//! ColumnKit cookbook: a custom per-locus analyzer that inherits the bounded +//! memory contract for free. The only domain logic is `on_column` (~3 lines) — +//! by running it through `run_bounded_whole_genome` it gets the SAME bounded +//! per-contig whole-genome walk and the SAME working-set bound that `plan` / +//! `variants --index --enforce` admit, with no contract code re-derived. +//! +//! Run with: `cargo run --example columnkit_coverage` + +use std::collections::BTreeMap; +use std::io::{self, Write}; +use std::sync::Arc; + +use rosalind::call::{run_bounded_whole_genome, ColumnAnalyzer}; +use rosalind::core::{AlignedRead, CigarOp, CigarOpKind, Position, SamFlags}; +use rosalind::genomics::{GenomeIndex, IndexReader, IndexWriter}; +use rosalind::{PileupColumn, PileupParams, SliceSource}; + +/// Emit a per-locus coverage track (contig, 1-based pos, depth). This is the +/// entire analyzer a builder writes — bounded memory, determinism, and a +/// verifiable receipt come from the driver, not from here. +struct CoverageTrack; + +impl ColumnAnalyzer for CoverageTrack { + fn header(&self) -> Option { + Some("#contig\tpos\tdepth\n".to_string()) + } + + fn params(&self) -> BTreeMap { + BTreeMap::from([("analyzer".to_string(), "coverage".to_string())]) + } + + fn on_column( + &mut self, + col: &PileupColumn, + contig: &str, + out: &mut dyn Write, + ) -> io::Result<()> { + writeln!(out, "{contig}\t{}\t{}", col.locus.pos.0 + 1, col.depth()) + } +} + +fn read(pos: u32, seq: &[u8]) -> AlignedRead { + AlignedRead { + contig: 0, + pos: Position(pos), + mapq: 60, + flags: SamFlags(0), + cigar: vec![CigarOp::new(CigarOpKind::Match, seq.len() as u32)], + seq: Arc::from(seq.to_vec().into_boxed_slice()), + qual: Arc::from(vec![40u8; seq.len()].into_boxed_slice()), + } +} + +fn main() -> Result<(), Box> { + // A tiny in-memory genome + a few reads. A real run would pass a persisted + // index and a coordinate-sorted BAM (`StreamingBamSource`) — the driver and + // the contract are identical either way. + let dir = std::env::temp_dir().join("rosalind-columnkit-example"); + std::fs::create_dir_all(&dir)?; + let idx_path = dir.join("ref.idx"); + let index = + GenomeIndex::from_named_sequences(&[("chr1".to_string(), b"ACGTACGTACGTACGT".to_vec())])?; + IndexWriter::create(&idx_path)?.write_genome_index(&index)?; + let loaded = IndexReader::open(&idx_path)?; + let ref_view = loaded.reference_view()?; + let contigs = loaded.contigs(); + + let reads = vec![ + read(0, b"ACGTACGT"), + read(0, b"ACGTACGT"), + read(4, b"ACGTACGT"), + ]; + + let mut analyzer = CoverageTrack; + let mut out = io::stdout().lock(); + let (ws, _skips) = run_bounded_whole_genome( + &mut analyzer, + SliceSource::new(reads), + &ref_view, + contigs, + PileupParams::default(), + &mut out, + )?; + out.flush()?; + + eprintln!( + "\n# inherited a bounded working set of {} bytes — coverage-bounded, \ + independent of input size; this is the value `plan`/`--enforce` admit.", + ws.bytes + ); + + std::fs::remove_dir_all(&dir).ok(); + Ok(()) +} diff --git a/src/call/columnkit.rs b/src/call/columnkit.rs new file mode 100644 index 0000000..d581d23 --- /dev/null +++ b/src/call/columnkit.rs @@ -0,0 +1,277 @@ +//! ColumnKit: implement one trait, inherit the bounded memory contract. +//! +//! A builder who wants their OWN per-locus metric — methylation, a coverage/QC +//! track, custom ML features, a star-allele genotyper — would otherwise fork into +//! the raw pileup engine and re-implement, by hand, every surface that makes +//! Rosalind worth choosing: the whole-genome contig walk, the working-set bound +//! that `plan`/`--enforce` admit, and the canonical-JSON + BLAKE3 receipt. +//! +//! [`ColumnAnalyzer`] + [`run_bounded_whole_genome`] turn "extend the substrate" +//! into a first-class SDK: implement one trait, run it through the driver, and +//! inherit the SAME bounded per-contig stream, the SAME working-set bound, and +//! the SAME verifiable receipt the shipped `variants`/`features` subcommands +//! enjoy — for free. The trait is *welded* to the bounded kernel: the driver runs +//! the exact same column stream as `stream_features_whole_genome`, so the +//! estimator that admits a run provably upper-bounds the realized working set of +//! the builder's analyzer too. It is the contract made composable, not a feature +//! bolted beside it. (`FeatureAnalyzer` is the first impl — proof the trait +//! carries the real shipped analyzer, not a toy.) + +use std::collections::BTreeMap; +use std::io::{self, Write}; + +use crate::call::features::{stream_features_whole_genome, write_feature_row, FEATURE_HEADER}; +use crate::core::{ContigSet, CoreError, WorkingSet}; +use crate::genomics::ReferenceView; +use crate::pileup::{PileupColumn, PileupParams, ReadSource, SkipCounts}; + +/// A per-locus analyzer over the bounded pileup-column stream. Implement this and +/// run it with [`run_bounded_whole_genome`] to inherit bounded memory, +/// byte-identical determinism, and a verifiable receipt. +pub trait ColumnAnalyzer { + /// Optional header written once, before any column (e.g. a TSV header line, + /// including its trailing newline). Default: none. + fn header(&self) -> Option { + None + } + + /// Parameters to fold into the run receipt (provenance). Default: none. + fn params(&self) -> BTreeMap { + BTreeMap::new() + } + + /// Process one callable pileup column, writing any output to `out`. Called in + /// `(contig, position)` order; `contig` is the column's contig name. Keep + /// per-call state O(1) (or bounded) to preserve the memory contract. + fn on_column( + &mut self, + col: &PileupColumn, + contig: &str, + out: &mut dyn Write, + ) -> io::Result<()>; +} + +/// Drive `analyzer` over every callable column of a coordinate-sorted read stream +/// and a persisted reference, under the bounded memory contract: peak ≈ the +/// largest contig's reference + the pileup working set, independent of input +/// size. Writes the analyzer's header (if any), then streams columns to it in +/// `(contig, pos)` order. Returns the max pileup working set observed — the value +/// `plan`/`--enforce` reason about — and the skip counts, so the analyzer +/// inherits the receipt. +pub fn run_bounded_whole_genome( + analyzer: &mut dyn ColumnAnalyzer, + source: S, + ref_view: &ReferenceView, + contigs: &ContigSet, + pileup_params: PileupParams, + out: &mut dyn Write, +) -> Result<(WorkingSet, SkipCounts), CoreError> { + if let Some(header) = analyzer.header() { + out.write_all(header.as_bytes())?; + } + stream_features_whole_genome( + source, + ref_view, + contigs, + pileup_params, + &mut |col, contig| { + analyzer + .on_column(col, contig, &mut *out) + .map_err(CoreError::from) + }, + ) +} + +/// The shipped `features` egress, expressed as a [`ColumnAnalyzer`] — proof the +/// trait carries the real production analyzer. Counts the rows it emits (folded +/// into the receipt via [`ColumnAnalyzer::params`]). +#[derive(Debug, Default)] +pub struct FeatureAnalyzer { + rows: u64, +} + +impl FeatureAnalyzer { + /// Number of feature rows written so far. + pub fn rows(&self) -> u64 { + self.rows + } +} + +impl ColumnAnalyzer for FeatureAnalyzer { + fn header(&self) -> Option { + Some(format!("{FEATURE_HEADER}\n")) + } + + fn params(&self) -> BTreeMap { + let mut p = BTreeMap::new(); + p.insert("feature_rows".to_string(), self.rows.to_string()); + p + } + + fn on_column( + &mut self, + col: &PileupColumn, + contig: &str, + out: &mut dyn Write, + ) -> io::Result<()> { + write_feature_row(out, contig, col)?; + self.rows += 1; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::call::features::write_feature_header; + use crate::genomics::{GenomeIndex, IndexReader, IndexWriter}; + use crate::pileup::SliceSource; + use std::sync::Arc; + + fn tmp(suffix: &str) -> std::path::PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let d = std::env::temp_dir().join(format!("rosalind-ck-{suffix}-{nanos}")); + std::fs::create_dir_all(&d).unwrap(); + d.join("ref.idx") + } + + fn read_at(contig: u32, pos: u32, seq: &[u8]) -> crate::core::AlignedRead { + use crate::core::{CigarOp, CigarOpKind, Position, SamFlags}; + crate::core::AlignedRead { + contig, + pos: Position(pos), + mapq: 60, + flags: SamFlags(0), + cigar: vec![CigarOp::new(CigarOpKind::Match, seq.len() as u32)], + seq: Arc::from(seq.to_vec().into_boxed_slice()), + qual: Arc::from(vec![40u8; seq.len()].into_boxed_slice()), + } + } + + // Running FeatureAnalyzer through the driver must be byte-identical to the + // hand-wired features path (header + write_feature_row per column) — i.e. the + // trait IS the production path, not a parallel one. + #[test] + fn feature_analyzer_via_driver_equals_the_direct_features_path() { + let idx = tmp("equiv"); + let index = GenomeIndex::from_named_sequences(&[ + ("chr1".to_string(), b"ACGTACGTACGTACGTACGT".to_vec()), + ("chr2".to_string(), b"TTTTGGGGCCCCAAAATTTT".to_vec()), + ]) + .unwrap(); + IndexWriter::create(&idx) + .unwrap() + .write_genome_index(&index) + .unwrap(); + let loaded = IndexReader::open(&idx).unwrap(); + let rv = loaded.reference_view().unwrap(); + let contigs = loaded.contigs(); + let reads = vec![ + read_at(0, 0, b"ACGTACGT"), + read_at(0, 0, b"ACGTACGT"), + read_at(1, 0, b"TTTTGGGG"), + ]; + let pp = PileupParams::default(); + + // Direct path: header + stream_features_whole_genome + write_feature_row. + let mut direct: Vec = Vec::new(); + write_feature_header(&mut direct).unwrap(); + let (ws_direct, _) = stream_features_whole_genome( + SliceSource::new(reads.clone()), + &rv, + contigs, + pp.clone(), + &mut |col, name| write_feature_row(&mut direct, name, col).map_err(CoreError::from), + ) + .unwrap(); + + // Driver path: FeatureAnalyzer through run_bounded_whole_genome. + let mut analyzer = FeatureAnalyzer::default(); + let mut via_kit: Vec = Vec::new(); + let (ws_kit, _) = run_bounded_whole_genome( + &mut analyzer, + SliceSource::new(reads), + &rv, + contigs, + pp, + &mut via_kit, + ) + .unwrap(); + + assert_eq!(direct, via_kit, "driver output must be byte-identical"); + assert_eq!( + ws_direct.bytes, ws_kit.bytes, + "driver must report the same bounded working set" + ); + assert!(analyzer.rows() > 0, "analyzer should count its rows"); + assert_eq!( + analyzer.params().get("feature_rows").map(String::as_str), + Some(analyzer.rows().to_string().as_str()) + ); + + let _ = std::fs::remove_dir_all(idx.parent().unwrap()); + } + + // A tiny custom analyzer inherits the bounded walk: its working set is the + // same coverage-bounded value, independent of how many reads stream in. + #[test] + fn a_custom_analyzer_inherits_the_bounded_working_set() { + struct DepthTrack { + max_depth_seen: u32, + } + impl ColumnAnalyzer for DepthTrack { + fn on_column( + &mut self, + col: &PileupColumn, + _contig: &str, + _out: &mut dyn Write, + ) -> io::Result<()> { + self.max_depth_seen = self.max_depth_seen.max(col.depth()); + Ok(()) + } + } + + let idx = tmp("custom"); + let index = + GenomeIndex::from_named_sequences(&[("chr1".to_string(), vec![b'A'; 500])]).unwrap(); + IndexWriter::create(&idx) + .unwrap() + .write_genome_index(&index) + .unwrap(); + let loaded = IndexReader::open(&idx).unwrap(); + let rv = loaded.reference_view().unwrap(); + let contigs = loaded.contigs(); + let pp = PileupParams { + max_depth: Some(8), + ..PileupParams::default() + }; + + let run = |n: usize| -> u64 { + let reads: Vec<_> = (0..n).map(|_| read_at(0, 0, &[b'C'; 50])).collect(); + let mut a = DepthTrack { max_depth_seen: 0 }; + let mut sink = io::sink(); + run_bounded_whole_genome( + &mut a, + SliceSource::new(reads), + &rv, + contigs, + pp.clone(), + &mut sink, + ) + .unwrap() + .0 + .bytes + }; + // Bounded: 100× more reads at one locus (capped at depth 8) → same WS. + assert_eq!( + run(20), + run(2000), + "analyzer working set must not grow with reads" + ); + + let _ = std::fs::remove_dir_all(idx.parent().unwrap()); + } +} diff --git a/src/call/features.rs b/src/call/features.rs index 7c9e886..cd47e99 100644 --- a/src/call/features.rs +++ b/src/call/features.rs @@ -19,14 +19,14 @@ pub const FEATURE_HEADER: &str = "#contig\tpos\tref\tdepth\traw_depth\ta\tc\tg\t a_fwd\ta_rev\tc_fwd\tc_rev\tg_fwd\tg_rev\tt_fwd\tt_rev\tmean_bq\tmean_mapq"; /// Write the feature header line. -pub fn write_feature_header(out: &mut W) -> io::Result<()> { +pub fn write_feature_header(out: &mut W) -> io::Result<()> { writeln!(out, "{FEATURE_HEADER}") } /// Write one feature row for a callable pileup column. `pos` is 1-based (VCF POS). /// Means are integer sums over observations divided by callable depth, formatted /// to 2 decimals so the row is byte-stable across runs. -pub fn write_feature_row( +pub fn write_feature_row( out: &mut W, contig_name: &str, col: &PileupColumn, diff --git a/src/call/mod.rs b/src/call/mod.rs index 20fc0dd..5a37108 100644 --- a/src/call/mod.rs +++ b/src/call/mod.rs @@ -2,6 +2,7 @@ //! abstention-aware variant calls. Built on `crate::core` + `crate::pileup` //! only; no VCF writing or CLI wiring (those are later phases). +pub mod columnkit; pub mod features; pub mod germline; pub mod pack; @@ -11,6 +12,7 @@ pub mod somatic; pub mod types; pub mod whole_genome; +pub use columnkit::{run_bounded_whole_genome, ColumnAnalyzer, FeatureAnalyzer}; pub use features::{ stream_features_region, stream_features_whole_genome, write_feature_header, write_feature_row, }; diff --git a/src/lib.rs b/src/lib.rs index 1847e55..f12606f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -79,8 +79,13 @@ pub use pileup::{Obs, PileupColumn, PileupEngine, PileupParams, ReadSource, Slic pub use call::{ call_germline_region_streaming, call_germline_whole_genome, GermlineCall, GermlineParams, }; -// The memory contract (declare → plan → honor → verify): -pub use call::{estimate_variants_working_set, predicted_peak_rss_bytes}; +// ColumnKit: implement one trait, inherit the bounded contract (SDK front door). +pub use call::{run_bounded_whole_genome, ColumnAnalyzer, FeatureAnalyzer}; +// The memory contract (declare → plan → honor → verify), incl. fleet packing: +pub use call::{ + estimate_variants_working_set, first_fit_decreasing, predicted_peak_rss_bytes, PackJob, + PackOutcome, +}; pub use core::{MemoryBudget, WorkingSet}; // Build-once → mmap index + the reproducibility receipt: pub use genomics::{GenomeIndex, IndexReader, ReferenceView}; diff --git a/src/main.rs b/src/main.rs index d591668..8483715 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1555,7 +1555,7 @@ fn run_features( output: Option, manifest_out: Option, ) -> Result<()> { - use rosalind::call::{stream_features_whole_genome, write_feature_header, write_feature_row}; + use rosalind::call::run_bounded_whole_genome; use rosalind::genomics::IndexReader; use rosalind::io::bam::StreamingBamSource; use rosalind::pileup::PileupParams; @@ -1628,48 +1628,45 @@ fn run_features( // Stream feature rows straight to the writer (header once, one row per callable // locus) — no genome-wide buffer accumulates. - let mut feature_rows: u64 = 0; + // `features` is the first ColumnKit analyzer: the FeatureAnalyzer drives the + // SAME bounded whole-genome column walk a builder's own analyzer would, so the + // shipped path and the SDK are one and the same (not parallel). Byte-identical + // output is pinned by the golden feature test. + let mut analyzer = rosalind::call::FeatureAnalyzer::default(); let (max_ws, skips) = match &output { Some(path) => { let file = File::create(path) .with_context(|| format!("failed to create features file {}", path.display()))?; let mut writer = io::BufWriter::new(file); - write_feature_header(&mut writer)?; - let (ws, sk) = stream_features_whole_genome( + let r = run_bounded_whole_genome( + &mut analyzer, source, &ref_view, contigs, pileup_params, - &mut |col, name| { - feature_rows += 1; - write_feature_row(&mut writer, name, col) - .map_err(rosalind::core::CoreError::from) - }, + &mut writer, ) .map_err(|e| anyhow!("feature streaming failed: {e}"))?; writer.flush()?; - (ws, sk) + r } None => { let stdout = io::stdout(); let mut handle = stdout.lock(); - write_feature_header(&mut handle)?; - let (ws, sk) = stream_features_whole_genome( + let r = run_bounded_whole_genome( + &mut analyzer, source, &ref_view, contigs, pileup_params, - &mut |col, name| { - feature_rows += 1; - write_feature_row(&mut handle, name, col) - .map_err(rosalind::core::CoreError::from) - }, + &mut handle, ) .map_err(|e| anyhow!("feature streaming failed: {e}"))?; handle.flush()?; - (ws, sk) + r } }; + let feature_rows = analyzer.rows(); let peak_rss = std::env::var("ROSALIND_FORCE_PEAK_RSS_BYTES") .ok()