pipe_io/
error.rs

1//! Error model for `pipe-io`.
2//!
3//! The crate uses a single top-level [`Error`] type that wraps the boxed
4//! source error from any stage. Stage-supplied errors only need to
5//! satisfy [`StageError`] (`Debug + Display + Send + Sync + 'static`),
6//! which every `std::error::Error` type already does via the blanket
7//! impl below. The `core::error::Error` super-trait was avoided so the
8//! error model works at the crate's MSRV of 1.75.
9
10use alloc::boxed::Box;
11use core::fmt;
12
13use crate::StageId;
14
15/// Crate result alias.
16pub type Result<T> = core::result::Result<T, Error>;
17
18/// Marker trait for error types that stages can produce. Blanket-
19/// implemented for every `T: Debug + Display + Send + Sync + 'static`,
20/// which includes every `std::error::Error` implementor.
21pub trait StageError: fmt::Debug + fmt::Display + Send + Sync + 'static {}
22
23impl<T> StageError for T where T: fmt::Debug + fmt::Display + Send + Sync + 'static {}
24
25/// Boxed stage error. Carried by [`Error`] variants.
26pub type BoxError = Box<dyn StageError>;
27
28/// Kind of buffer-related failure surfaced by a driver.
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
30pub enum BufferErrorKind {
31    /// The downstream buffer was full and could not accept the item.
32    Full,
33    /// The buffer was closed (typically because a downstream stage
34    /// terminated unexpectedly).
35    Closed,
36}
37
38impl fmt::Display for BufferErrorKind {
39    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
40        match self {
41            Self::Full => f.write_str("buffer full"),
42            Self::Closed => f.write_str("buffer closed"),
43        }
44    }
45}
46
47/// Top-level error type returned by every pipeline operation.
48///
49/// Variants carry the [`StageId`] that produced them, and (where
50/// applicable) the underlying boxed error.
51#[non_exhaustive]
52pub enum Error {
53    /// A source produced an error or could not produce an item.
54    Source {
55        /// Stage identifier of the failing source.
56        stage: StageId,
57        /// Boxed source error.
58        source: BoxError,
59    },
60    /// A transform stage produced an error.
61    Stage {
62        /// Stage identifier of the failing stage.
63        stage: StageId,
64        /// Boxed stage error.
65        source: BoxError,
66    },
67    /// A sink produced an error while writing or flushing.
68    Sink {
69        /// Stage identifier of the failing sink.
70        stage: StageId,
71        /// Boxed sink error.
72        source: BoxError,
73    },
74    /// An inter-stage buffer failed.
75    Buffer {
76        /// Stage identifier of the affected stage.
77        stage: StageId,
78        /// Kind of buffer failure.
79        kind: BufferErrorKind,
80    },
81    /// The pipeline was cancelled before completion.
82    Cancelled,
83    /// The pipeline was already closed when an operation was attempted.
84    Closed,
85}
86
87impl fmt::Debug for Error {
88    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
89        match self {
90            Self::Source { stage, source } => f
91                .debug_struct("Source")
92                .field("stage", stage)
93                .field("source", &format_args!("{source}"))
94                .finish(),
95            Self::Stage { stage, source } => f
96                .debug_struct("Stage")
97                .field("stage", stage)
98                .field("source", &format_args!("{source}"))
99                .finish(),
100            Self::Sink { stage, source } => f
101                .debug_struct("Sink")
102                .field("stage", stage)
103                .field("source", &format_args!("{source}"))
104                .finish(),
105            Self::Buffer { stage, kind } => f
106                .debug_struct("Buffer")
107                .field("stage", stage)
108                .field("kind", kind)
109                .finish(),
110            Self::Cancelled => f.write_str("Cancelled"),
111            Self::Closed => f.write_str("Closed"),
112        }
113    }
114}
115
116impl fmt::Display for Error {
117    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
118        match self {
119            Self::Source { stage, source } => write!(f, "source `{stage}`: {source}"),
120            Self::Stage { stage, source } => write!(f, "stage `{stage}`: {source}"),
121            Self::Sink { stage, source } => write!(f, "sink `{stage}`: {source}"),
122            Self::Buffer { stage, kind } => write!(f, "buffer at `{stage}`: {kind}"),
123            Self::Cancelled => f.write_str("pipeline cancelled"),
124            Self::Closed => f.write_str("pipeline closed"),
125        }
126    }
127}
128
129#[cfg(feature = "std")]
130impl std::error::Error for Error {}
131
132impl Error {
133    /// Stage identifier associated with this error, if any.
134    #[must_use]
135    pub fn stage(&self) -> Option<StageId> {
136        match self {
137            Self::Source { stage, .. }
138            | Self::Stage { stage, .. }
139            | Self::Sink { stage, .. }
140            | Self::Buffer { stage, .. } => Some(*stage),
141            Self::Cancelled | Self::Closed => None,
142        }
143    }
144}
145
146/// Per-stage policy for how errors propagate.
147#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
148pub enum ErrorPolicy {
149    /// The first stage error cancels the entire pipeline and bubbles
150    /// out of [`crate::Pipeline::run`]. Default.
151    #[default]
152    FailFast,
153    /// Stage errors are dropped along with the failing item; downstream
154    /// continues to receive subsequent items.
155    Continue,
156    /// Stage errors are routed as [`StageFailure`] records to a
157    /// dead-letter sink installed via
158    /// [`crate::PipelineBuilder::dead_letter`]. If no dead-letter sink
159    /// is installed the failing record is silently dropped (same as
160    /// [`ErrorPolicy::Continue`]).
161    DeadLetter,
162}
163
164/// Record describing a per-item stage failure, routed to the
165/// dead-letter sink when [`ErrorPolicy::DeadLetter`] is active.
166///
167/// The struct is `#[non_exhaustive]`; future releases may add fields
168/// (for example, a type-erased copy of the failing input) without
169/// breaking SemVer.
170#[non_exhaustive]
171pub struct StageFailure {
172    /// Stage that failed.
173    pub stage: StageId,
174    /// Underlying error.
175    pub source: BoxError,
176}
177
178impl StageFailure {
179    /// Construct a new stage failure.
180    #[must_use]
181    pub fn new(stage: StageId, source: BoxError) -> Self {
182        Self { stage, source }
183    }
184}
185
186impl fmt::Debug for StageFailure {
187    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
188        f.debug_struct("StageFailure")
189            .field("stage", &self.stage)
190            .field("source", &format_args!("{}", self.source))
191            .finish()
192    }
193}
194
195impl fmt::Display for StageFailure {
196    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
197        write!(f, "stage `{}`: {}", self.stage, self.source)
198    }
199}
200
201#[cfg(test)]
202mod tests {
203    use super::*;
204    use alloc::format;
205    use alloc::string::ToString;
206
207    #[derive(Debug)]
208    struct Boom;
209    impl fmt::Display for Boom {
210        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
211            f.write_str("boom")
212        }
213    }
214
215    #[test]
216    fn display_source_error() {
217        let e = Error::Source {
218            stage: StageId::new("ingest"),
219            source: Box::new(Boom),
220        };
221        assert_eq!(e.to_string(), "source `ingest`: boom");
222    }
223
224    #[test]
225    fn display_stage_error() {
226        let e = Error::Stage {
227            stage: StageId::new("parse"),
228            source: Box::new(Boom),
229        };
230        assert_eq!(e.to_string(), "stage `parse`: boom");
231    }
232
233    #[test]
234    fn display_sink_error() {
235        let e = Error::Sink {
236            stage: StageId::new("writer"),
237            source: Box::new(Boom),
238        };
239        assert_eq!(e.to_string(), "sink `writer`: boom");
240    }
241
242    #[test]
243    fn display_buffer_error_kinds() {
244        let full = Error::Buffer {
245            stage: StageId::new("edge"),
246            kind: BufferErrorKind::Full,
247        };
248        let closed = Error::Buffer {
249            stage: StageId::new("edge"),
250            kind: BufferErrorKind::Closed,
251        };
252        assert_eq!(full.to_string(), "buffer at `edge`: buffer full");
253        assert_eq!(closed.to_string(), "buffer at `edge`: buffer closed");
254    }
255
256    #[test]
257    fn display_terminal_variants() {
258        assert_eq!(Error::Cancelled.to_string(), "pipeline cancelled");
259        assert_eq!(Error::Closed.to_string(), "pipeline closed");
260    }
261
262    #[test]
263    fn error_stage_extraction() {
264        let e = Error::Stage {
265            stage: StageId::new("middle"),
266            source: Box::new(Boom),
267        };
268        assert_eq!(e.stage(), Some(StageId::new("middle")));
269        assert_eq!(Error::Cancelled.stage(), None);
270    }
271
272    #[test]
273    fn error_policy_default_is_fail_fast() {
274        assert_eq!(ErrorPolicy::default(), ErrorPolicy::FailFast);
275    }
276
277    #[test]
278    fn stage_failure_display() {
279        let f = StageFailure {
280            stage: StageId::new("parse"),
281            source: Box::new(Boom),
282        };
283        assert_eq!(format!("{f}"), "stage `parse`: boom");
284    }
285}
pipe_io/error.rs

pipe_io/
error.rs
