From 87ed8349f8c5c18597f34315097f827033a4e180 Mon Sep 17 00:00:00 2001 From: Nils Homer Date: Mon, 15 Jun 2026 20:21:54 -0700 Subject: [PATCH] refactor(pipeline): add PipelineError::reconstruct to de-duplicate outcome rematch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Pipeline::run` (core/builder.rs) and the fused single-thread driver (core/runtime/fused.rs) each open-coded the same per-variant match to turn the borrowed `signal.outcome(): Option<&PipelineError>` into the owned error they return — necessary because `PipelineError` is not `Clone` (its `Io` variant holds a non-`Clone` `io::Error`). Extract that into `PipelineError::reconstruct(&self) -> PipelineError`; both call sites collapse to `Some(err) => Err(err.reconstruct())`. Behavior-identical (the `Io` source is still rebuilt from `kind()` + display string), and a future error variant now only needs handling in one place instead of two. Fused-pipeline and pipeline error-path tests pass unchanged. --- src/lib/pipeline/core/builder.rs | 13 +------ src/lib/pipeline/core/runtime/fused.rs | 14 ++------ src/lib/pipeline/core/signal.rs | 47 ++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 24 deletions(-) diff --git a/src/lib/pipeline/core/builder.rs b/src/lib/pipeline/core/builder.rs index b6bbb8267..49bd9005c 100644 --- a/src/lib/pipeline/core/builder.rs +++ b/src/lib/pipeline/core/builder.rs @@ -751,7 +751,6 @@ impl Pipeline { build_chain_contexts, build_worker_storage, is_fusible_chain, run_fused_single_thread, run_worker_loop, }; - use super::signal::PipelineError; use super::step::StepKind; use super::topology::StepIdx; @@ -1022,17 +1021,7 @@ impl Pipeline { // (io::Error isn't Clone); reconstruct from the recorded outcome. let _ = graph; match signal.outcome() { - Some(PipelineError::Cancelled) => Err(PipelineError::Cancelled), - Some(PipelineError::Io { step, source }) => Err(PipelineError::Io { - step, - source: std::io::Error::new(source.kind(), format!("{source}")), - }), - Some(PipelineError::NotEnoughThreads { required, available }) => { - Err(PipelineError::NotEnoughThreads { required: *required, available: *available }) - } - Some(PipelineError::TimedOut { stalled_secs }) => { - Err(PipelineError::TimedOut { stalled_secs: *stalled_secs }) - } + Some(err) => Err(err.reconstruct()), None => Ok(()), } } diff --git a/src/lib/pipeline/core/runtime/fused.rs b/src/lib/pipeline/core/runtime/fused.rs index 7cdde19c1..deb265826 100644 --- a/src/lib/pipeline/core/runtime/fused.rs +++ b/src/lib/pipeline/core/runtime/fused.rs @@ -189,19 +189,9 @@ pub fn run_fused_single_thread( } // Map the recorded outcome to the run result (same shape as `Pipeline::run`; - // `PipelineError` is not `Clone`, so reconstruct the `Io` variant). + // `PipelineError` is not `Clone`, so reconstruct via `PipelineError::reconstruct`). match signal.outcome() { - Some(PipelineError::Cancelled) => Err(PipelineError::Cancelled), - Some(PipelineError::Io { step, source }) => Err(PipelineError::Io { - step, - source: std::io::Error::new(source.kind(), format!("{source}")), - }), - Some(PipelineError::NotEnoughThreads { required, available }) => { - Err(PipelineError::NotEnoughThreads { required: *required, available: *available }) - } - Some(PipelineError::TimedOut { stalled_secs }) => { - Err(PipelineError::TimedOut { stalled_secs: *stalled_secs }) - } + Some(err) => Err(err.reconstruct()), None => Ok(()), } } diff --git a/src/lib/pipeline/core/signal.rs b/src/lib/pipeline/core/signal.rs index 389779341..cb9115788 100644 --- a/src/lib/pipeline/core/signal.rs +++ b/src/lib/pipeline/core/signal.rs @@ -26,6 +26,29 @@ pub enum PipelineError { TimedOut { stalled_secs: u64 }, } +impl PipelineError { + /// Reconstruct an owned copy of this error. + /// + /// `PipelineError` is not `Clone` because its `Io` variant holds an + /// [`io::Error`], which is not `Clone`; this rebuilds that source from its + /// `kind()` + display string. Used to turn the borrowed `signal.outcome()` + /// into the owned error returned from the run drivers (`Pipeline::run` and + /// the fused single-thread driver), which previously open-coded the same + /// per-variant rematch in two places. + pub(crate) fn reconstruct(&self) -> PipelineError { + match self { + Self::Cancelled => Self::Cancelled, + Self::Io { step, source } => { + Self::Io { step, source: io::Error::new(source.kind(), format!("{source}")) } + } + Self::NotEnoughThreads { required, available } => { + Self::NotEnoughThreads { required: *required, available: *available } + } + Self::TimedOut { stalled_secs } => Self::TimedOut { stalled_secs: *stalled_secs }, + } + } +} + impl std::fmt::Display for PipelineError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { @@ -210,4 +233,28 @@ mod tests { assert!(handle.is_cancelled()); assert!(signal.is_cancelled()); } + + #[test] + fn reconstruct_preserves_every_variant() { + // `reconstruct` rebuilds an owned error from a borrow (PipelineError + // isn't Clone). Each variant must round-trip; the `Io` source is + // rebuilt from kind + display, the rest are copied verbatim. + let io = PipelineError::Io { step: "s", source: io::Error::other("boom") }.reconstruct(); + match io { + PipelineError::Io { step, source } => { + assert_eq!(step, "s"); + assert_eq!(source.kind(), io::ErrorKind::Other); + assert_eq!(source.to_string(), "boom"); + } + other => panic!("expected Io, got {other:?}"), + } + + assert!(matches!(PipelineError::Cancelled.reconstruct(), PipelineError::Cancelled)); + + let net = PipelineError::NotEnoughThreads { required: 4, available: 2 }.reconstruct(); + assert!(matches!(net, PipelineError::NotEnoughThreads { required: 4, available: 2 })); + + let to = PipelineError::TimedOut { stalled_secs: 60 }.reconstruct(); + assert!(matches!(to, PipelineError::TimedOut { stalled_secs: 60 })); + } }