fix(sort): address CodeRabbit review + de-duplicate cli-common helpers (#446)

crates/fgumi-cli-common/src/lib.rs

@@ -5,7 +5,6 @@
 // Command trait
 // ─────────────────────────────────────────────────────────────────────────────
 
-use anyhow::Result;
 use enum_dispatch::enum_dispatch;
 
 /// Trait implemented by all fgumi CLI commands.
@@ -15,7 +14,7 @@ use enum_dispatch::enum_dispatch;
 #[enum_dispatch]
 pub trait Command {
     #[allow(clippy::missing_errors_doc)]
-    fn execute(&self, command_line: &str) -> Result<()>;
+    fn execute(&self, command_line: &str) -> anyhow::Result<()>;
 }
 
 // ─────────────────────────────────────────────────────────────────────────────
@@ -24,9 +23,16 @@ pub trait Command {
 
 use thiserror::Error;
 
-/// Result type alias for fgumi operations
+/// Result type alias for fgumi operations (preferred in standalone crates).
 pub type FgumiResult<T> = std::result::Result<T, FgumiError>;
 
+/// Unqualified result alias used by the umbrella crate's error/validation modules.
+///
+/// Both `FgumiResult<T>` and `Result<T>` are the same type; the two names exist
+/// so callers that shadow `std::result::Result` with `use crate::errors::Result`
+/// (umbrella convention) resolve to the same `FgumiError`-based alias.
+pub type Result<T> = std::result::Result<T, FgumiError>;
+
 /// Error type for fgumi operations
 #[derive(Error, Debug)]
 pub enum FgumiError {
@@ -99,7 +105,11 @@ pub fn detect_total_memory() -> usize {
     system.refresh_memory();
     let physical = system.total_memory();
     let bytes = system.cgroup_limits().map_or(physical, |c| c.total_memory.min(physical));
-    usize::try_from(bytes).unwrap_or(usize::MAX)
+    // Saturate at usize::MAX / 2 rather than usize::MAX on 32-bit platforms so
+    // the downstream `budget > total` overflow check in `resolve_memory_budget`
+    // can fire correctly (no value can exceed usize::MAX, so using it as the
+    // fallback renders the check dead).
+    usize::try_from(bytes).unwrap_or(usize::MAX / 2)
 }
 
 /// Returns the number of logical CPUs available to this process.
@@ -119,7 +129,8 @@ pub fn detect_cpu_count() -> usize {
 // ─────────────────────────────────────────────────────────────────────────────
 
 /// Format an integer with comma separators (e.g. 1234567 → "1,234,567").
-fn format_count(n: u64) -> String {
+#[must_use]
+pub fn format_count(n: u64) -> String {
     let s = n.to_string();
     let mut result = String::with_capacity(s.len() + s.len() / 3);
     let offset = s.len() % 3;
@@ -406,7 +417,7 @@ const AUTO_RESERVE_CAP: usize = 10 * 1024 * 1024 * 1024;
 /// # Errors
 ///
 /// Returns an error string if parsing fails.
-pub fn parse_memory(s: &str) -> Result<MemoryLimit, String> {
+pub fn parse_memory(s: &str) -> std::result::Result<MemoryLimit, String> {
     let s = s.trim();
     if s.eq_ignore_ascii_case("auto") {
         return Ok(MemoryLimit::Auto);
@@ -419,7 +430,7 @@ pub fn parse_memory(s: &str) -> Result<MemoryLimit, String> {
 /// # Errors
 ///
 /// Returns an error string if parsing fails.
-pub fn parse_memory_reserve(s: &str) -> Result<MemoryReserve, String> {
+pub fn parse_memory_reserve(s: &str) -> std::result::Result<MemoryReserve, String> {
     let s = s.trim();
     if s.eq_ignore_ascii_case("auto") {
         return Ok(MemoryReserve::Auto);
@@ -517,7 +528,7 @@ fn resolve_memory_budget_with_total(
 }
 
 /// Parse a memory size string into `usize` bytes (private helper).
-fn parse_memory_bytes(s: &str, label: &str) -> Result<usize, String> {
+fn parse_memory_bytes(s: &str, label: &str) -> std::result::Result<usize, String> {
     let bytes = parse_memory_size(s).map_err(|e| e.to_string())?;
     usize::try_from(bytes).map_err(|_| format!("{label} too large: {bytes}"))
 }
@@ -528,7 +539,7 @@ fn parse_memory_bytes(s: &str, label: &str) -> Result<usize, String> {
 /// # Errors
 ///
 /// Returns an error string if the input is not a recognized boolean.
-pub fn parse_bool(s: &str) -> Result<bool, String> {
+pub fn parse_bool(s: &str) -> std::result::Result<bool, String> {
     match s.to_ascii_lowercase().as_str() {
         "true" | "t" | "yes" | "y" => Ok(true),
         "false" | "f" | "no" | "n" => Ok(false),
@@ -696,4 +707,44 @@ mod tests {
         assert!(CompressionHarness::try_parse_from(["prog", "--compression-level", "13"]).is_err());
         assert!(CompressionHarness::try_parse_from(["prog", "--compression-level", "99"]).is_err());
     }
+
+    #[test]
+    fn test_resolve_memory_budget_auto_low_available() {
+        // When available memory per thread is below MIN_MEMORY_PER_THREAD, the
+        // budget is floored to MIN_MEMORY_PER_THREAD × threads, then capped at
+        // available (which is less), so the result equals available.
+        let total = 2 * MIN_MEMORY_PER_THREAD; // very tight: only 2 × floor per 4 threads
+        let reserve = 0;
+        let budget = resolve_memory_budget_with_total(
+            MemoryLimit::Auto,
+            MemoryReserve::Fixed(reserve),
+            4,
+            true,
+            total,
+        )
+        .unwrap();
+        // available = total - 0 = 2×MIN; per-thread = 2×MIN/4 < MIN → floored to MIN;
+        // target = MIN × 4 = 4×MIN > available → capped at available.
+        assert_eq!(budget, total);
+    }
+
+    #[test]
+    fn test_resolve_memory_budget_auto_margin_exceeds_total() {
+        // When the reserve margin >= total, saturating_sub → 0 available.
+        // Budget is then floored to MIN_MEMORY_PER_THREAD, capped at 0
+        // (available), so result == 0 (the cap wins).
+        let total = 512 * 1024 * 1024_usize; // 512 MiB
+        let margin = total + 1; // margin exceeds total
+        let budget = resolve_memory_budget_with_total(
+            MemoryLimit::Auto,
+            MemoryReserve::Fixed(margin),
+            1,
+            false,
+            total,
+        )
+        .unwrap();
+        // available = total.saturating_sub(margin) = 0; target = max(0, MIN) = MIN;
+        // budget = MIN.min(0) = 0.
+        assert_eq!(budget, 0);
+    }
 }

crates/fgumi-pipeline-core/Cargo.toml

@@ -15,6 +15,8 @@ log = "0"
 noodles = { version = "0.111.0", features = ["sam"] }
 
 [dev-dependencies]
+proptest = "1.10"
+rstest = "0"
 trybuild = "1.0"
 
 [lints.clippy]

crates/fgumi-pipeline-core/src/builder.rs

@@ -903,6 +903,10 @@ impl Pipeline {
                 (None, None)
             };
 
+        // Holds the first worker panic payload; re-raised after helper threads
+        // are cleaned up so monitor/rebalancer shutdown always executes.
+        let mut worker_panic: Option<Box<dyn std::any::Any + Send>> = None;
+
         if n_threads == 1 {
             // Single-threaded fast path: run the worker loop directly on
             // the caller's thread instead of spawning + joining a fresh
@@ -929,14 +933,26 @@ impl Pipeline {
             let sticky_owner = sticky_owners[0];
             let mut worker = WorkerCore::new(0, exclusive_owner, sticky_owner);
             let mut entries_local = entries;
-            run_worker_loop(
-                &mut worker,
-                &mut entries_local,
-                &contexts,
-                &drain_counters,
-                &signal_arc,
-                stats_arc.as_ref(),
-            );
+            // Defer a panic on the single-threaded fast path the same way the
+            // multi-worker join loop does: capture the payload, signal
+            // cancellation, and let the common monitor/rebalancer shutdown run
+            // before re-raising at step 7. Without this, a worker-loop panic
+            // unwinds straight through the caller and leaks the helper threads.
+            // `AssertUnwindSafe` is sound: after a panic we never touch `worker`
+            // or `entries_local` again — the run is shutting down.
+            if let Err(panic) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                run_worker_loop(
+                    &mut worker,
+                    &mut entries_local,
+                    &contexts,
+                    &drain_counters,
+                    &signal_arc,
+                    stats_arc.as_ref(),
+                );
+            })) {
+                signal_arc.cancel();
+                worker_panic = Some(panic);
+            }
         } else {
             // 5. Spawn worker threads.
             let mut handles = Vec::with_capacity(n_threads);
@@ -957,26 +973,43 @@ impl Pipeline {
                     .spawn(move || {
                         let mut worker = WorkerCore::new(worker_id, exclusive_owner, sticky_owner);
                         let mut entries_local = entries;
-                        run_worker_loop(
-                            &mut worker,
-                            &mut entries_local,
-                            &contexts_clone,
-                            &drain_counters_clone,
-                            &signal_clone,
-                            stats_clone.as_ref(),
-                        );
+                        // Catch a worker-loop panic so we can signal cancellation
+                        // *before* unwinding. A peer parked in its retry loop on a
+                        // full/empty queue only exits when it observes
+                        // `signal.is_done()`; without an early `cancel()` here, the
+                        // join loop below could block forever on an earlier,
+                        // now-wedged worker and never reach this thread's panic.
+                        // We re-raise after signalling so the join still collects
+                        // the payload (preserving the deferred re-raise at step 7).
+                        // `AssertUnwindSafe` is sound: on panic the run is tearing
+                        // down and neither local is used again.
+                        if let Err(panic) =
+                            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                                run_worker_loop(
+                                    &mut worker,
+                                    &mut entries_local,
+                                    &contexts_clone,
+                                    &drain_counters_clone,
+                                    &signal_clone,
+                                    stats_clone.as_ref(),
+                                );
+                            }))
+                        {
+                            signal_clone.cancel();
+                            std::panic::resume_unwind(panic);
+                        }
                     })
                     .expect("failed to spawn worker thread");
                 handles.push(handle);
             }
 
-            // 6. Join workers. If a worker panicked, re-raise its original
-            // payload via `resume_unwind` so the main thread aborts with the
-            // worker's actual panic message and location — not the opaque
-            // `Any { .. }` that `join().expect(...)` would print.
+            // 6. Join workers. Capture the first worker panic payload so cleanup
+            // can proceed; re-raise after monitor/rebalancer threads are stopped.
             for h in handles {
                 if let Err(panic) = h.join() {
-                    std::panic::resume_unwind(panic);
+                    if worker_panic.is_none() {
+                        worker_panic = Some(panic);
+                    }
                 }
             }
         }
@@ -1014,7 +1047,13 @@ impl Pipeline {
             }
         }
 
-        // 7. Surface error or cancellation. PipelineError isn't Clone
+        // 7. Re-raise worker panics after helper threads are cleaned up so
+        // monitor/rebalancer shutdown code always executes.
+        if let Some(panic) = worker_panic {
+            std::panic::resume_unwind(panic);
+        }
+
+        // 8. Surface error or cancellation. PipelineError isn't Clone
         // (io::Error isn't Clone); `to_result` reconstructs the recorded
         // outcome and, for an external cancel whose payload isn't yet visible
         // to this thread, synthesizes `Cancelled` from the terminal state.
@@ -1804,6 +1843,95 @@ mod tests {
         assert_eq!(received.load(AtomicOrd::Relaxed), 50);
     }
 
+    /// Sink whose worker loop panics the moment it pops an item — used to drive
+    /// the worker-panic deferral paths in `Pipeline::run`.
+    #[derive(Clone)]
+    struct PanickingSink;
+    impl Step for PanickingSink {
+        type Input = u32;
+        type Outputs = ();
+        fn profile(&self) -> StepProfile {
+            StepProfile {
+                name: "PanickingSink",
+                kind: StepKind::Parallel,
+                sticky: false,
+                output_queues: vec![],
+                branch_ordering: vec![],
+            }
+        }
+        fn try_run(&mut self, ctx: &mut StepCtx<'_, Self>) -> std::io::Result<StepOutcome> {
+            match ctx.input.pop() {
+                Some(_) => panic!("intentional worker panic for test"),
+                None if ctx.input.is_drained() => Ok(StepOutcome::Finished),
+                None => Ok(StepOutcome::NoProgress),
+            }
+        }
+        fn new_worker_copy(&self) -> Self {
+            self.clone()
+        }
+    }
+
+    /// Run `f` with the panic hook silenced so an *expected* worker panic does
+    /// not spew a backtrace into the test log. Safe under `nextest`, which runs
+    /// each test in its own process.
+    fn with_silenced_panic_hook<R>(f: impl FnOnce() -> R) -> R {
+        let prev = std::panic::take_hook();
+        std::panic::set_hook(Box::new(|_| {}));
+        let result = f();
+        std::panic::set_hook(prev);
+        result
+    }
+
+    #[test]
+    fn pipeline_run_reraises_worker_panic_single_threaded() {
+        // A worker-loop panic on the single-threaded fast path must propagate
+        // out of `run` (after the common monitor/rebalancer shutdown), not be
+        // swallowed. The test completing at all proves the run did not hang.
+        let remaining = Arc::new(AtomicU32::new(10));
+        let builder = PipelineBuilder::new();
+        builder
+            .chain(SharedCountingSource { remaining: Arc::clone(&remaining) })
+            .chain(PanickingSink)
+            .into_sink_marker();
+        let pipeline = builder.build().unwrap();
+
+        let result = with_silenced_panic_hook(|| {
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                pipeline.run(PipelineConfig { threads: 1, ..Default::default() })
+            }))
+        });
+        assert!(result.is_err(), "single-threaded worker panic must propagate out of run()");
+    }
+
+    #[test]
+    fn pipeline_run_reraises_worker_panic_with_monitor_enabled() {
+        // With the deadlock monitor enabled (stats + non-zero timeout), a
+        // multi-worker panic must still re-raise — after the monitor is stopped
+        // and joined — rather than deadlocking the join loop or leaking the
+        // helper thread. The panicking worker signals cancellation so any wedged
+        // peer observes `is_done()` and exits, letting every join complete.
+        let remaining = Arc::new(AtomicU32::new(1_000));
+        let builder = PipelineBuilder::new();
+        builder
+            .chain(SharedCountingSource { remaining: Arc::clone(&remaining) })
+            .chain(PanickingSink)
+            .into_sink_marker();
+        let pipeline = builder.build().unwrap();
+        let stats = pipeline.stats();
+
+        let result = with_silenced_panic_hook(|| {
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                pipeline.run(PipelineConfig {
+                    threads: 4,
+                    stats: Some(Arc::clone(&stats)),
+                    deadlock_timeout_secs: 5,
+                    ..Default::default()
+                })
+            }))
+        });
+        assert!(result.is_err(), "multi-worker panic must propagate out of run()");
+    }
+
     #[test]
     fn pipeline_stats_handle_matches_chain_size() {
         let builder = PipelineBuilder::new();

crates/fgumi-pipeline-core/src/erased.rs

@@ -530,13 +530,19 @@ impl<S: Step2> ErasedStep for TypedStep2<S> {
         p1_idx: usize,
         p1_branch: usize,
     ) -> Box<dyn Any + Send + Sync> {
-        assert_ne!(
-            p0_idx,
-            p1_idx,
-            "Step2 producers must be distinct steps (each upstream is its \
-             own subchain). Got p0_idx == p1_idx == {p0_idx} for step '{}'.",
-            self.profile().name
-        );
+        if p0_idx == p1_idx {
+            assert_ne!(
+                p0_branch,
+                p1_branch,
+                "Step2 inputs must consume distinct branches when they share \
+                 producer step {p0_idx} for step '{}'.",
+                self.profile().name
+            );
+            let set = &mut producer_sets[p0_idx];
+            let a: BranchInputHandle<S::InputA> = set.take_typed_input::<S::InputA>(p0_branch);
+            let b: BranchInputHandle<S::InputB> = set.take_typed_input::<S::InputB>(p1_branch);
+            return Box::new(TwoInputHandles::<S::InputA, S::InputB>::new(a, b));
+        }
         // Borrow two disjoint elements of `producer_sets` simultaneously.
         // `split_at_mut(lo+1)` puts producer_sets[lo] in the first half;
         // we index into the second half for the hi side.