refactor(pipeline): PerThreadAccumulator for all per-batch metric collection

CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file.
 ### Bug Fixes
 
 - Cap `fgumi group` metric memory by merging per-position-group counts into per-thread accumulators instead of a `SegQueue` that grew one `AHashMap` per position group ([#285](https://github.com/fulcrumgenomics/fgumi/issues/285)).
+- Replace unbounded `SegQueue<CollectedXxxMetrics>` buffering in `filter`, `clip`, `correct`, `simplex`, `duplex`, and `codec` with a shared `PerThreadAccumulator` helper, capping retained metric memory at `O(threads × distinct keys)` across all pipeline commands. Rejects buffering in the consensus commands is unchanged and tracked separately.
 
 ### Breaking Changes
 

src/lib/commands/clip.rs

@@ -14,6 +14,7 @@ use crate::grouper::TemplateGrouper;
 use crate::logging::OperationTimer;
 use crate::metrics::clip::{ClipCounts, ClippingMetricsCollection};
 use crate::metrics::writer::write_metrics as write_metrics_tsv;
+use crate::per_thread_accumulator::PerThreadAccumulator;
 use crate::progress::ProgressTracker;
 use crate::reference::ReferenceReader;
 use crate::template::{TemplateBatch, TemplateIterator};
@@ -23,7 +24,6 @@ use crate::unified_pipeline::{
 use crate::validation::validate_file_exists;
 use anyhow::Result;
 use clap::Parser;
-use crossbeam_queue::SegQueue;
 use log::info;
 use noodles::sam::Header;
 use noodles::sam::alignment::io::Write as AlignmentWrite;
@@ -641,8 +641,8 @@ impl Clip {
             num_threads,
         )?;
 
-        // Lock-free metrics collection
-        let collected_metrics: Arc<SegQueue<CollectedClipMetrics>> = Arc::new(SegQueue::new());
+        // Per-thread metrics accumulator: bounded memory, no unbounded queue.
+        let collected_metrics = PerThreadAccumulator::<CollectedClipMetrics>::new(num_threads);
         let collected_for_serialize = Arc::clone(&collected_metrics);
 
         // Configuration for closures
@@ -789,11 +789,11 @@ impl Clip {
                                  header: &Header,
                                  output: &mut Vec<u8>|
               -> io::Result<u64> {
-            // Push metrics to lock-free queue
-            collected_for_serialize.push(CollectedClipMetrics {
-                total_templates: processed.templates_count,
-                overlap_clipped: processed.overlap_clipped_count,
-                extend_clipped: processed.extend_clipped_count,
+            // Merge per-batch counts into this worker's accumulator slot
+            collected_for_serialize.with_slot(|m| {
+                m.total_templates += processed.templates_count;
+                m.overlap_clipped += processed.overlap_clipped_count;
+                m.extend_clipped += processed.extend_clipped_count;
             });
 
             // Serialize clipped records into the provided buffer
@@ -817,10 +817,11 @@ impl Clip {
         let mut total_overlap_clipped = 0u64;
         let mut total_extend_clipped = 0u64;
 
-        while let Some(metrics) = collected_metrics.pop() {
-            total_templates += metrics.total_templates;
-            total_overlap_clipped += metrics.overlap_clipped;
-            total_extend_clipped += metrics.extend_clipped;
+        for slot in collected_metrics.slots() {
+            let m = slot.lock();
+            total_templates += m.total_templates;
+            total_overlap_clipped += m.overlap_clipped;
+            total_extend_clipped += m.extend_clipped;
         }
 
         info!("Total templates processed: {total_templates}");

src/lib/commands/codec.rs

@@ -12,9 +12,9 @@ use crate::bam_io::{
 };
 use crate::commands::command::Command;
 use crate::commands::consensus_runner::{ConsensusStatsOps, create_unmapped_consensus_header};
+use crate::per_thread_accumulator::PerThreadAccumulator;
 use anyhow::{Context, Result, bail};
 use clap::Parser;
-use crossbeam_queue::SegQueue;
 use fgoxide::io::DelimFile;
 
 use super::common::{
@@ -502,8 +502,9 @@ impl Codec {
             num_threads,
         )?;
 
-        // Lock-free metrics collection
-        let collected_metrics: Arc<SegQueue<CollectedCodecMetrics>> = Arc::new(SegQueue::new());
+        // Per-thread metrics accumulator: bounded metric memory, no unbounded
+        // queue. Rejects buffering semantics are preserved (see follow-up).
+        let collected_metrics = PerThreadAccumulator::<CollectedCodecMetrics>::new(num_threads);
         let collected_metrics_for_serialize = Arc::clone(&collected_metrics);
 
         // Parse cell tag
@@ -604,11 +605,14 @@ impl Codec {
                   _header: &Header,
                   output: &mut Vec<u8>|
                   -> io::Result<u64> {
-                // Collect metrics (lock-free)
-                collected_metrics_for_serialize.push(CollectedCodecMetrics {
-                    stats: processed.stats,
-                    groups_processed: processed.groups_count,
-                    rejects: processed.rejects,
+                // Merge per-batch metrics + rejects into this worker's slot
+                let batch_stats = processed.stats;
+                let groups_count = processed.groups_count;
+                let mut batch_rejects = processed.rejects;
+                collected_metrics_for_serialize.with_slot(|m| {
+                    m.stats.merge(&batch_stats);
+                    m.groups_processed += groups_count;
+                    m.rejects.append(&mut batch_rejects);
                 });
 
                 // Serialize consensus reads
@@ -624,11 +628,12 @@ impl Codec {
         let mut merged_stats = CodecConsensusStats::default();
         let mut all_rejects: Vec<Vec<u8>> = Vec::new();
 
-        while let Some(metrics) = collected_metrics.pop() {
-            total_groups += metrics.groups_processed;
-            merged_stats.merge(&metrics.stats);
+        for slot in collected_metrics.slots() {
+            let mut m = slot.lock();
+            total_groups += m.groups_processed;
+            merged_stats.merge(&m.stats);
             if track_rejects {
-                all_rejects.extend(metrics.rejects);
+                all_rejects.append(&mut m.rejects);
             }
         }
 

src/lib/commands/correct.rs

@@ -48,6 +48,7 @@ use crate::dna::reverse_complement_str;
 use crate::grouper::TemplateGrouper;
 use crate::logging::OperationTimer;
 use crate::metrics::correct::UmiCorrectionMetrics;
+use crate::per_thread_accumulator::PerThreadAccumulator;
 use crate::progress::ProgressTracker;
 use crate::sam::SamTag;
 use crate::sort::bam_fields;
@@ -57,7 +58,6 @@ use crate::validation::validate_file_exists;
 use ahash::AHashMap;
 use anyhow::{Result, bail};
 use clap::Parser;
-use crossbeam_queue::SegQueue;
 use fgumi_raw_bam::RawRecord;
 use log::{info, warn};
 use lru::LruCache;
@@ -788,8 +788,8 @@ impl CorrectUmis {
             pipeline_config.group_key_config = Some(GroupKeyConfig::new_raw_no_cell(library_index));
         }
 
-        // Lock-free metrics collection
-        let collected_metrics: Arc<SegQueue<CollectedCorrectMetrics>> = Arc::new(SegQueue::new());
+        // Per-thread metrics accumulator: bounded memory, no unbounded queue.
+        let collected_metrics = PerThreadAccumulator::<CollectedCorrectMetrics>::new(num_threads);
         let collected_for_serialize = Arc::clone(&collected_metrics);
 
         // Configuration for closures
@@ -959,18 +959,23 @@ impl CorrectUmis {
         };
 
         // Serialize function: convert records to bytes and collect metrics
-        let serialize_fn = move |processed: CorrectProcessedBatch,
+        let serialize_fn = move |mut processed: CorrectProcessedBatch,
                                  _header: &Header,
                                  output: &mut Vec<u8>|
               -> io::Result<u64> {
-            // Push metrics to lock-free queue
-            collected_for_serialize.push(CollectedCorrectMetrics {
-                templates_processed: processed.templates_count,
-                missing_umis: processed.missing_umis,
-                wrong_length: processed.wrong_length,
-                mismatched: processed.mismatched,
-                umi_matches: processed.umi_matches,
-                raw_rejects: processed.rejected_raw_records,
+            // Merge per-batch counts + UMI matches + rejects into this
+            // worker's accumulator slot.
+            let umi_matches = std::mem::take(&mut processed.umi_matches);
+            let mut rejects = std::mem::take(&mut processed.rejected_raw_records);
+            collected_for_serialize.with_slot(|m| {
+                m.templates_processed += processed.templates_count;
+                m.missing_umis += processed.missing_umis;
+                m.wrong_length += processed.wrong_length;
+                m.mismatched += processed.mismatched;
+                for (umi, counts) in umi_matches {
+                    merge_umi_counts(&mut m.umi_matches, umi, &counts);
+                }
+                m.raw_rejects.append(&mut rejects);
             });
 
             // Serialize kept records
@@ -1008,25 +1013,19 @@ impl CorrectUmis {
         let mut merged_umi_matches: AHashMap<String, UmiCorrectionMetrics> = AHashMap::new();
         let mut all_raw_rejects: Vec<Vec<u8>> = Vec::new();
 
-        while let Some(metrics) = collected_metrics.pop() {
-            total_templates += metrics.templates_processed;
-            total_missing += metrics.missing_umis;
-            total_wrong_length += metrics.wrong_length;
-            total_mismatched += metrics.mismatched;
-
-            for (umi, counts) in metrics.umi_matches {
-                let entry = merged_umi_matches
-                    .entry(umi.clone())
-                    .or_insert_with(|| UmiCorrectionMetrics::new(umi));
-                entry.total_matches += counts.total_matches;
-                entry.perfect_matches += counts.perfect_matches;
-                entry.one_mismatch_matches += counts.one_mismatch_matches;
-                entry.two_mismatch_matches += counts.two_mismatch_matches;
-                entry.other_matches += counts.other_matches;
+        for slot in collected_metrics.slots() {
+            let mut m = slot.lock();
+            total_templates += m.templates_processed;
+            total_missing += m.missing_umis;
+            total_wrong_length += m.wrong_length;
+            total_mismatched += m.mismatched;
+
+            for (umi, counts) in m.umi_matches.drain() {
+                merge_umi_counts(&mut merged_umi_matches, umi, &counts);
             }
 
             if track_rejects {
-                all_raw_rejects.extend(metrics.raw_rejects);
+                all_raw_rejects.append(&mut m.raw_rejects);
             }
         }
 
@@ -1362,6 +1361,22 @@ impl CorrectUmis {
 /// assert_eq!(count_mismatches_with_max(b"AAAAAA", b"AAAAAT", 10), 1);
 /// assert_eq!(count_mismatches_with_max(b"AAAAAA", b"CCCCCC", 2), 3);
 /// ```
+/// Merges `counts` into `dst[umi]`, creating a zero-initialized entry if the
+/// key is absent. Uses `or_insert_with_key` so `umi` is only cloned on insert,
+/// not on hit.
+fn merge_umi_counts(
+    dst: &mut AHashMap<String, UmiCorrectionMetrics>,
+    umi: String,
+    counts: &UmiCorrectionMetrics,
+) {
+    let entry = dst.entry(umi).or_insert_with_key(|k| UmiCorrectionMetrics::new(k.clone()));
+    entry.total_matches += counts.total_matches;
+    entry.perfect_matches += counts.perfect_matches;
+    entry.one_mismatch_matches += counts.one_mismatch_matches;
+    entry.two_mismatch_matches += counts.two_mismatch_matches;
+    entry.other_matches += counts.other_matches;
+}
+
 #[must_use]
 pub fn count_mismatches_with_max(a: &[u8], b: &[u8], max_mismatches: usize) -> usize {
     let mut mismatches = 0;

src/lib/commands/duplex.rs

@@ -29,6 +29,7 @@ use crate::overlapping_consensus::{
     AgreementStrategy, CorrectionStats, DisagreementStrategy, OverlappingBasesConsensusCaller,
     apply_overlapping_consensus_raw,
 };
+use crate::per_thread_accumulator::PerThreadAccumulator;
 use crate::progress::ProgressTracker;
 use crate::read_info::LibraryIndex;
 use crate::sam::SamTag;
@@ -38,7 +39,6 @@ use crate::unified_pipeline::{
     GroupKeyConfig, Grouper, MemoryEstimate, run_bam_pipeline_from_reader,
 };
 use crate::validation::validate_file_exists;
-use crossbeam_queue::SegQueue;
 use fgumi_raw_bam::{RawRecord, RawRecordView};
 use log::info;
 use noodles::sam::Header;
@@ -574,8 +574,9 @@ impl Duplex {
             num_threads,
         )?;
 
-        // Lock-free metrics collection
-        let collected_metrics: Arc<SegQueue<CollectedDuplexMetrics>> = Arc::new(SegQueue::new());
+        // Per-thread metrics accumulator: bounded metric memory, no unbounded
+        // queue. Rejects buffering semantics are preserved (see follow-up).
+        let collected_metrics = PerThreadAccumulator::<CollectedDuplexMetrics>::new(num_threads);
         let collected_metrics_for_serialize = Arc::clone(&collected_metrics);
 
         // Capture configuration for closures
@@ -731,12 +732,18 @@ impl Duplex {
                   _header: &Header,
                   output: &mut Vec<u8>|
                   -> io::Result<u64> {
-                // Collect metrics (lock-free)
-                collected_metrics_for_serialize.push(CollectedDuplexMetrics {
-                    stats: processed.stats,
-                    overlapping_stats: processed.overlapping_stats,
-                    groups_processed: processed.groups_count,
-                    rejects: processed.rejects,
+                // Merge per-batch metrics + rejects into this worker's slot
+                let batch_stats = processed.stats;
+                let batch_overlapping = processed.overlapping_stats;
+                let groups_count = processed.groups_count;
+                let mut batch_rejects = processed.rejects;
+                collected_metrics_for_serialize.with_slot(|m| {
+                    m.stats.merge(&batch_stats);
+                    if let Some(o) = batch_overlapping {
+                        m.overlapping_stats.get_or_insert_with(CorrectionStats::new).merge(&o);
+                    }
+                    m.groups_processed += groups_count;
+                    m.rejects.append(&mut batch_rejects);
                 });
 
                 let count = processed.consensus_output.count as u64;
@@ -752,14 +759,15 @@ impl Duplex {
         let mut merged_overlapping_stats = CorrectionStats::new();
         let mut all_rejects: Vec<Vec<u8>> = Vec::new();
 
-        while let Some(metrics) = collected_metrics.pop() {
-            total_groups += metrics.groups_processed;
-            merged_stats.merge(&metrics.stats);
-            if let Some(ref ocs) = metrics.overlapping_stats {
+        for slot in collected_metrics.slots() {
+            let mut m = slot.lock();
+            total_groups += m.groups_processed;
+            merged_stats.merge(&m.stats);
+            if let Some(ref ocs) = m.overlapping_stats {
                 merged_overlapping_stats.merge(ocs);
             }
             if track_rejects {
-                all_rejects.extend(metrics.rejects);
+                all_rejects.append(&mut m.rejects);
             }
         }
 

src/lib/commands/filter.rs

@@ -21,6 +21,7 @@ use crate::consensus_filter::{
 };
 use crate::grouper::{SingleRawRecordGrouper, TemplateGrouper};
 use crate::logging::OperationTimer;
+use crate::per_thread_accumulator::PerThreadAccumulator;
 use crate::read_info::LibraryIndex;
 use crate::reference::ReferenceReader;
 use crate::sort::bam_fields;
@@ -34,7 +35,6 @@ use crate::validation::validate_file_exists;
 use ahash::AHashMap;
 use anyhow::{Result, bail};
 use clap::Parser;
-use crossbeam_queue::SegQueue;
 use fgumi_raw_bam::RawRecordView;
 use log::info;
 use noodles::sam::Header;
@@ -260,7 +260,7 @@ fn serialize_raw_records(records: &[Vec<u8>], output: &mut Vec<u8>) -> io::Resul
     Ok(records.len() as u64)
 }
 
-/// Metrics collected from filter processing, aggregated post-pipeline.
+/// Per-thread accumulator merged into final counts after pipeline completion.
 #[derive(Default)]
 struct CollectedFilterMetrics {
     /// Total records processed.
@@ -278,7 +278,7 @@ struct FilterPipelineSetup {
     pipeline_config: BamPipelineConfig,
     config: Arc<FilterConfig>,
     reference: Option<Arc<ReferenceReader>>,
-    collected_metrics: Arc<SegQueue<CollectedFilterMetrics>>,
+    collected_metrics: Arc<PerThreadAccumulator<CollectedFilterMetrics>>,
     progress_counter: Arc<AtomicU64>,
 }
 
@@ -406,7 +406,7 @@ impl Filter {
             None => None,
         };
 
-        let collected_metrics: Arc<SegQueue<CollectedFilterMetrics>> = Arc::new(SegQueue::new());
+        let collected_metrics = PerThreadAccumulator::<CollectedFilterMetrics>::new(num_threads);
         let progress_counter = Arc::new(AtomicU64::new(0));
 
         Ok(FilterPipelineSetup {
@@ -475,11 +475,11 @@ impl Filter {
                                  _header: &Header,
                                  output: &mut Vec<u8>|
               -> io::Result<u64> {
-            collected_for_serialize.push(CollectedFilterMetrics {
-                total_records: processed.records_count,
-                passed_records: processed.passed_count,
-                failed_records: processed.records_count - processed.passed_count,
-                total_bases_masked: processed.bases_masked,
+            collected_for_serialize.with_slot(|m| {
+                m.total_records += processed.records_count;
+                m.passed_records += processed.passed_count;
+                m.failed_records += processed.records_count - processed.passed_count;
+                m.total_bases_masked += processed.bases_masked;
             });
 
             serialize_raw_records(&processed.kept_records, output)
@@ -523,11 +523,12 @@ impl Filter {
         let mut failed_reads = 0u64;
         let mut total_bases_masked = 0u64;
 
-        while let Some(metrics) = setup.collected_metrics.pop() {
-            total_reads += metrics.total_records;
-            passed_reads += metrics.passed_records;
-            failed_reads += metrics.failed_records;
-            total_bases_masked += metrics.total_bases_masked;
+        for slot in setup.collected_metrics.slots() {
+            let m = slot.lock();
+            total_reads += m.total_records;
+            passed_reads += m.passed_records;
+            failed_reads += m.failed_records;
+            total_bases_masked += m.total_bases_masked;
         }
 
         if let Some(stats_path) = &self.stats {