chainfold/
error.rs

1use core::fmt;
2
3use crate::{
4    batch::BatchShapeError,
5    position::{
6        BlockRef,
7        Position,
8    },
9};
10
11/// Classified fold failure; the variant fixes the engine's recovery obligation.
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum FoldError<E> {
14    /// Event is not mine; engine counts it and advances the cursor.
15    Skip(E),
16    /// State is clean up to the previous position; recovery is rollback or resync.
17    Halt(E),
18    /// Apply mutated state partially; state is untrusted until a restore.
19    Poison(E),
20}
21
22impl<E: fmt::Display> fmt::Display for FoldError<E> {
23    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
24        match self {
25            Self::Skip(error) => write!(f, "skip: {error}"),
26            Self::Halt(error) => write!(f, "halt: {error}"),
27            Self::Poison(error) => write!(f, "poison: {error}"),
28        }
29    }
30}
31
32impl<E: fmt::Debug + fmt::Display> core::error::Error for FoldError<E> {}
33
34/// Coarse-grained state of the fold engine.
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub enum EngineStatus {
37    /// Applying batches and advancing the cursor.
38    Active,
39    /// Fold refused an event with clean state below it.
40    Halted {
41        /// Position the fold halted at.
42        at: Position,
43    },
44    /// Fold refused an event after mutating state.
45    Poisoned {
46        /// Position the fold poisoned at.
47        at: Position,
48    },
49    /// Automated recovery is exhausted; only a reset leaves this state.
50    Unrecoverable {
51        /// Divergence that ended automated recovery.
52        cause: DivergenceCause,
53    },
54}
55
56impl EngineStatus {
57    /// True while the engine still accepts batches.
58    pub const fn is_active(&self) -> bool {
59        matches!(self, Self::Active)
60    }
61}
62
63impl fmt::Display for EngineStatus {
64    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
65        match self {
66            Self::Active => write!(f, "active"),
67            Self::Halted { at } => {
68                write!(f, "halted at block {} log index {}", at.block, at.log_index)
69            }
70            Self::Poisoned { at } => {
71                write!(
72                    f,
73                    "poisoned at block {} log index {}",
74                    at.block, at.log_index
75                )
76            }
77            Self::Unrecoverable { cause } => write!(f, "unrecoverable: {cause}"),
78        }
79    }
80}
81
82impl core::error::Error for EngineStatus {}
83
84/// Terminal divergence reason the engine cannot recover from automatically.
85#[derive(Debug, Clone, Copy, PartialEq, Eq)]
86pub enum DivergenceCause {
87    /// Fork deeper than the oldest observed block in the ring.
88    ForkBeyondWindow,
89    /// Replay is required but the source's horizon no longer covers the start block.
90    HorizonExceeded {
91        /// Block replay must start from.
92        needed: u64,
93        /// Oldest block the source still serves.
94        horizon: u64,
95    },
96}
97
98impl fmt::Display for DivergenceCause {
99    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
100        match self {
101            Self::ForkBeyondWindow => write!(f, "fork deeper than the observed window"),
102            Self::HorizonExceeded { needed, horizon } => {
103                write!(
104                    f,
105                    "replay needs block {needed}, source horizon is {horizon}"
106                )
107            }
108        }
109    }
110}
111
112impl core::error::Error for DivergenceCause {}
113
114/// Durability sink refused a snapshot offer; persistence has stopped.
115#[derive(Debug, Clone, Copy, PartialEq, Eq)]
116pub struct DurabilityLost;
117
118impl fmt::Display for DurabilityLost {
119    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
120        write!(
121            f,
122            "durability sink refused a snapshot; snapshots are no longer persisted"
123        )
124    }
125}
126
127impl core::error::Error for DurabilityLost {}
128
129/// Rejected engine construction configuration.
130#[derive(Debug, Clone, Copy, PartialEq, Eq)]
131pub enum ConfigError {
132    /// Ring capacity is not a power of two.
133    RingCapacityNotPowerOfTwo {
134        /// Capacity the caller asked for.
135        got: usize,
136    },
137    /// Ring capacity is outside the allowed range.
138    RingCapacityOutOfRange {
139        /// Capacity the caller asked for.
140        got: usize,
141    },
142    /// Source cannot replay from the configured start block.
143    HorizonExceedsStart {
144        /// Block the driver folds from.
145        start: u64,
146        /// Oldest block the source still serves.
147        horizon: u64,
148    },
149}
150
151impl fmt::Display for ConfigError {
152    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
153        match self {
154            Self::RingCapacityNotPowerOfTwo { got } => {
155                write!(f, "ring capacity {got} is not a power of two")
156            }
157            Self::RingCapacityOutOfRange { got } => {
158                write!(f, "ring capacity {got} is out of the allowed range")
159            }
160            Self::HorizonExceedsStart { start, horizon } => {
161                write!(f, "replay horizon {horizon} exceeds start block {start}")
162            }
163        }
164    }
165}
166
167impl core::error::Error for ConfigError {}
168
169/// Rejected or failed batch application.
170#[derive(Debug, PartialEq, Eq)]
171pub enum ApplyError<E> {
172    /// Engine is halted, poisoned, or unrecoverable.
173    NotActive {
174        /// Status that refused the batch.
175        status: EngineStatus,
176    },
177    /// Batch violates the flat layout rules.
178    Shape(BatchShapeError),
179    /// Cursor is set but the batch carries no boundary header.
180    MissingBoundary,
181    /// Boundary header names a block other than the cursor block.
182    BoundaryNumberMismatch {
183        /// Cursor block the boundary must name.
184        expected: u64,
185        /// Block the boundary named.
186        got: u64,
187    },
188    /// Refetched header disagrees with the observed block of that number.
189    ForkSuspected {
190        /// Block as the engine observed it.
191        observed: BlockRef,
192        /// Block as the source refetched it.
193        refetched: BlockRef,
194    },
195    /// Observed ring holds no entry for the cursor block, so no boundary is verifiable.
196    CursorBlockUnobserved {
197        /// Cursor block absent from the ring.
198        block: u64,
199    },
200    /// Fold halted; state is clean up to the position before `at`.
201    Halted {
202        /// Position the fold refused.
203        at: Position,
204        /// Error the fold reported.
205        error: E,
206    },
207    /// Fold poisoned; state is untrusted until a restore.
208    Poisoned {
209        /// Position the fold refused after mutating.
210        at: Position,
211        /// Error the fold reported.
212        error: E,
213    },
214}
215
216impl<E: fmt::Display> fmt::Display for ApplyError<E> {
217    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
218        match self {
219            Self::NotActive { status } => write!(f, "engine not active: {status}"),
220            Self::Shape(error) => write!(f, "batch shape invalid: {error}"),
221            Self::MissingBoundary => {
222                write!(f, "cursor is set but the batch carries no boundary")
223            }
224            Self::BoundaryNumberMismatch { expected, got } => {
225                write!(
226                    f,
227                    "boundary block {got} does not match cursor block {expected}"
228                )
229            }
230            Self::ForkSuspected {
231                observed,
232                refetched,
233            } => {
234                write!(
235                    f,
236                    "fork suspected at block {}: observed hash {}, refetched hash {}",
237                    observed.number,
238                    HexHash(&observed.hash),
239                    HexHash(&refetched.hash)
240                )
241            }
242            Self::CursorBlockUnobserved { block } => {
243                write!(f, "observed ring holds no entry for cursor block {block}")
244            }
245            Self::Halted { at, error } => {
246                write!(
247                    f,
248                    "halted at block {} log index {}: {error}",
249                    at.block, at.log_index
250                )
251            }
252            Self::Poisoned { at, error } => {
253                write!(
254                    f,
255                    "poisoned at block {} log index {}: {error}",
256                    at.block, at.log_index
257                )
258            }
259        }
260    }
261}
262
263impl<E: fmt::Debug + fmt::Display> core::error::Error for ApplyError<E> {}
264
265/// Rejected rollback request.
266#[derive(Debug, Clone, Copy, PartialEq, Eq)]
267pub enum RollbackError {
268    /// No retained checkpoint sits at or below the requested block.
269    NoCheckpointAtOrBelow {
270        /// Block the rollback targeted.
271        block: u64,
272    },
273    /// Engine is unrecoverable; only a reset leaves this state.
274    Unrecoverable {
275        /// Divergence that ended automated recovery.
276        cause: DivergenceCause,
277    },
278}
279
280impl fmt::Display for RollbackError {
281    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
282        match self {
283            Self::NoCheckpointAtOrBelow { block } => {
284                write!(f, "no retained checkpoint at or below block {block}")
285            }
286            Self::Unrecoverable { cause } => write!(f, "unrecoverable: {cause}"),
287        }
288    }
289}
290
291impl core::error::Error for RollbackError {}
292
293/// Renders a 32-byte hash as lowercase hex.
294struct HexHash<'a>(&'a [u8; 32]);
295
296impl fmt::Display for HexHash<'_> {
297    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
298        for byte in self.0 {
299            write!(f, "{byte:02x}")?;
300        }
301        Ok(())
302    }
303}
chainfold/error.rs

chainfold/
error.rs
