chorus_client/
error.rs

1use std::time::Duration;
2
3use crate::manifest_store::ManifestStoreError;
4use crate::protocol::ProtocolError;
5use crate::segment::WalSeqNo;
6use crate::transport::{TransportCode, TransportError};
7
8/// Failure returned by every fallible `chorus-client` public operation.
9///
10/// Admission errors are definitive and may be corrected by the caller. In
11/// particular, [`ActiveSegmentFull`](Self::ActiveSegmentFull) does not consume
12/// the supplied sequence number: truncate retained history so rotation can
13/// proceed, then retry the same append. Use [`may_have_committed`](Self::may_have_committed)
14/// on an append completion error before deciding whether recovery must resolve
15/// an ambiguous outcome.
16#[derive(Clone, Debug, thiserror::Error)]
17#[non_exhaustive]
18pub enum Error {
19    /// The background WAL task was aborted, failed, or shut down.
20    #[error("WAL engine is closed")]
21    Closed,
22    /// Graceful shutdown exceeded its configured deadline and aborted the
23    /// remaining owned tasks.
24    #[error("WAL shutdown exceeded its {timeout:?} deadline")]
25    #[non_exhaustive]
26    ShutdownTimeout {
27        /// Configured graceful-shutdown deadline.
28        timeout: Duration,
29    },
30    /// A gRPC channel could not be constructed or connected.
31    #[error("storage connection failed: {0}")]
32    Connection(String),
33    /// The configured identity provider could not issue a token.
34    #[error("access-token source failed: {0}")]
35    TokenSource(String),
36    /// A bearer token cannot be represented in an ASCII gRPC header.
37    #[error("access token cannot be encoded as gRPC metadata")]
38    InvalidToken,
39    /// Refreshing would busy-loop because an interval was zero.
40    #[error("auth refresh and retry intervals must be non-zero")]
41    InvalidRefreshInterval,
42    /// A WAL engine capacity or size limit was invalid or inconsistent.
43    #[error("invalid WAL configuration: {0}")]
44    InvalidConfig(&'static str),
45    /// Fewer than a strict-majority quorum of replicas could be read.
46    #[error("segment discovery could not read a quorum")]
47    NoReadQuorum,
48    /// An operation could not reach a strict-majority replica quorum.
49    #[error("operation did not reach a replica quorum")]
50    NoQuorum,
51    /// Listed objects or decoded finalized bytes did not form one chain.
52    #[error("invalid segment catalog: {0}")]
53    InvalidCatalog(String),
54    /// The application supplied a checkpoint below its prior checkpoint.
55    #[error("checkpoint regressed from {current:?} to {requested:?}")]
56    #[non_exhaustive]
57    CheckpointRegression {
58        /// Checkpoint previously supplied to this writer.
59        current: WalSeqNo,
60        /// Regressing checkpoint supplied by the caller.
61        requested: WalSeqNo,
62    },
63    /// Finalized bytes or inferred segment bounds disagree with record framing.
64    #[error("invalid sealed segment data: {0}")]
65    InvalidSegmentData(String),
66    /// Recovery witnesses contain different bytes at one record boundary.
67    #[error("recovery witnesses contain different bytes at record {record_index}")]
68    #[non_exhaustive]
69    ConflictingPrefix {
70        /// First record index where the recovered witnesses disagree.
71        record_index: usize,
72    },
73    /// Recovery did not find the committed number of records.
74    #[error("recovery prefix has {actual} records, expected at least {expected}")]
75    #[non_exhaustive]
76    RecoveryPrefixTooShort {
77        /// Minimum record count required by the committed boundary.
78        expected: usize,
79        /// Record count recovered from the available witnesses.
80        actual: usize,
81    },
82    /// Recovered bytes disagree with the manifest's committed SHA-256 seal.
83    #[error("recovered seal digest {actual} does not match committed digest {expected}")]
84    #[non_exhaustive]
85    SealDigestMismatch {
86        /// Digest committed by the manifest.
87        expected: String,
88        /// Digest computed from the recovered bytes.
89        actual: String,
90    },
91    /// Recovered bytes disagree with the manifest's committed CRC32C.
92    #[error("recovered seal CRC32C {actual:08x} does not match committed CRC32C {expected:08x}")]
93    #[non_exhaustive]
94    SealCrc32cMismatch {
95        /// CRC32C committed by the manifest.
96        expected: u32,
97        /// CRC32C computed from the recovered bytes.
98        actual: u32,
99    },
100    /// The manifest register is malformed or inconsistent with this volume.
101    #[error("manifest register is invalid: {0}")]
102    InvalidManifest(String),
103    /// The manifest cannot retain another sealed-segment directory entry.
104    #[error(
105        "the manifest segment directory is full: truncate the WAL to free retained sealed \
106         segments before sealing again"
107    )]
108    SegmentDirectoryFull,
109    /// The manifest register remained unavailable through its retry budget.
110    #[error("manifest register is unavailable")]
111    ManifestUnavailable,
112    /// A caller-supplied manifest register operation failed.
113    #[error(transparent)]
114    ManifestStore(#[from] ManifestStoreError),
115    /// An internal commit notification path closed before reporting a result.
116    #[error("commit pipeline closed before reporting a result")]
117    PipelineClosed,
118    /// The low-level segment writer is already finalized.
119    #[error("segment writer is sealed")]
120    Finalized,
121    /// An internal invariant failed without a more stable public taxonomy.
122    #[error("internal WAL invariant failed: {0}")]
123    Internal(String),
124    /// An indeterminate append broke the ordered commit prefix.
125    #[error("WAL writer is poisoned by an indeterminate record; restart recovery is required")]
126    Poisoned,
127    /// This writer lost ownership to a newer manifest epoch.
128    #[error("WAL writer was fenced: {0}")]
129    Fenced(String),
130    /// The caller supplied a sequence number other than the next admission.
131    #[error("WAL append is out of order: expected {expected:?}, received {actual:?}")]
132    #[non_exhaustive]
133    OutOfOrder {
134        /// Exact sequence number required for the next admission.
135        expected: WalSeqNo,
136        /// Sequence number supplied by the caller.
137        actual: WalSeqNo,
138    },
139    /// A record exceeded the configured application-payload limit.
140    #[error("WAL record contains {actual} payload bytes, exceeding configured maximum {max}")]
141    #[non_exhaustive]
142    RecordTooLarge {
143        /// Configured per-record payload-byte limit.
144        max: usize,
145        /// Payload bytes supplied by the caller.
146        actual: usize,
147    },
148    /// The active object cannot admit another encoded record without crossing
149    /// its configured hard ceiling.
150    ///
151    /// The writer remains healthy and the sequence number was not consumed.
152    /// This is backpressure, not poison: truncating old sealed segments may
153    /// free manifest-directory room so the engine can rotate and resume
154    /// admission. A permanently failed seal still requires restart recovery
155    /// before rotation can resume.
156    #[error(
157        "active WAL segment holds {current} encoded bytes; admitting {requested} more would \
158         exceed the configured ceiling {max}"
159    )]
160    #[non_exhaustive]
161    ActiveSegmentFull {
162        /// Configured hard active-object ceiling.
163        max: usize,
164        /// Encoded bytes already charged to the active segment.
165        current: usize,
166        /// Encoded bytes required by the rejected record.
167        requested: usize,
168    },
169    /// No further contiguous sequence number can be represented.
170    #[error("WAL sequence-number space is exhausted")]
171    SequenceExhausted,
172    /// Startup replay did not reach its fixed end successfully.
173    #[error("recovery replay must complete before starting the WAL")]
174    RecoveryIncomplete,
175    /// Provider operation failed with a stable retry/fencing classification.
176    #[error("zone {zone}: {code:?}: {message}")]
177    #[non_exhaustive]
178    Transport {
179        /// Replica zone where the operation failed.
180        zone: usize,
181        /// Stable protocol-facing classification.
182        code: TransportCode,
183        /// Provider diagnostic text; correctness never matches this string.
184        message: String,
185    },
186}
187
188impl Error {
189    /// Whether this error can be observed after an append was admitted but
190    /// before recovery determines its durable outcome.
191    ///
192    /// `true` is deliberately conservative. [`Closed`](Self::Closed),
193    /// [`Internal`](Self::Internal), and transport failures can also arise
194    /// outside append completion, but callers handling an admitted append must
195    /// assume it may replay after takeover. [`Poisoned`](Self::Poisoned),
196    /// [`Fenced`](Self::Fenced), and [`PipelineClosed`](Self::PipelineClosed)
197    /// always require that same treatment.
198    ///
199    /// Admission and configuration errors, sequence errors,
200    /// [`ActiveSegmentFull`](Self::ActiveSegmentFull), [`NoQuorum`](Self::NoQuorum),
201    /// manifest/catalog validation errors, and low-level finalized-writer
202    /// errors are definitive for the attempted append.
203    pub fn may_have_committed(&self) -> bool {
204        matches!(
205            self,
206            Self::Closed
207                | Self::Internal(_)
208                | Self::Poisoned
209                | Self::Fenced(_)
210                | Self::PipelineClosed
211                | Self::Transport { .. }
212        )
213    }
214}
215
216impl From<TransportError> for Error {
217    fn from(error: TransportError) -> Self {
218        Self::Transport {
219            zone: error.zone,
220            code: error.code,
221            message: error.message,
222        }
223    }
224}
225
226impl From<ProtocolError> for Error {
227    fn from(error: ProtocolError) -> Self {
228        match error {
229            ProtocolError::ReplicaCount => Self::InvalidConfig("replica count must be 1, 3, or 5"),
230            ProtocolError::NoQuorum => Self::NoQuorum,
231            ProtocolError::Poisoned => Self::Poisoned,
232            ProtocolError::Fenced(message) => Self::Fenced(message),
233            ProtocolError::ConflictingPrefix { record_index } => {
234                Self::ConflictingPrefix { record_index }
235            }
236            ProtocolError::RecoveryPrefixTooShort { expected, actual } => {
237                Self::RecoveryPrefixTooShort { expected, actual }
238            }
239            ProtocolError::SealDigestMismatch { expected, actual } => {
240                Self::SealDigestMismatch { expected, actual }
241            }
242            ProtocolError::SealCrc32cMismatch { expected, actual } => {
243                Self::SealCrc32cMismatch { expected, actual }
244            }
245            ProtocolError::InvalidManifest(message) => Self::InvalidManifest(message),
246            ProtocolError::SegmentDirectoryFull => Self::SegmentDirectoryFull,
247            ProtocolError::ManifestStore(error) => Self::ManifestStore(error),
248            ProtocolError::ManifestUnavailable => Self::ManifestUnavailable,
249            ProtocolError::PipelineClosed => Self::PipelineClosed,
250            ProtocolError::Finalized => Self::Finalized,
251            ProtocolError::Record(error) => Self::InvalidSegmentData(error.to_string()),
252            ProtocolError::Transport(error) => error.into(),
253        }
254    }
255}
256
257impl From<crate::record::RecordError> for Error {
258    fn from(error: crate::record::RecordError) -> Self {
259        Self::InvalidSegmentData(error.to_string())
260    }
261}
262
263#[cfg(test)]
264mod tests {
265    use super::{Error, WalSeqNo};
266    use crate::manifest_store::ManifestStoreError;
267    use crate::protocol::ProtocolError;
268    use crate::transport::{TransportCode, TransportError};
269
270    #[test]
271    fn append_outcome_predicate_is_conservative_only_for_ambiguous_classes() {
272        for error in [
273            Error::Closed,
274            Error::Internal("opaque invariant".into()),
275            Error::Poisoned,
276            Error::Fenced("new owner".into()),
277            Error::PipelineClosed,
278            Error::Transport {
279                zone: 0,
280                code: TransportCode::Unavailable,
281                message: "response lost".into(),
282            },
283        ] {
284            assert!(error.may_have_committed(), "{error}");
285        }
286
287        for error in [
288            Error::InvalidConfig("bad limit"),
289            Error::NoQuorum,
290            Error::SegmentDirectoryFull,
291            Error::ManifestUnavailable,
292            Error::Finalized,
293            Error::OutOfOrder {
294                expected: WalSeqNo::ZERO,
295                actual: WalSeqNo::record(1),
296            },
297            Error::ActiveSegmentFull {
298                max: 8,
299                current: 8,
300                requested: 1,
301            },
302        ] {
303            assert!(!error.may_have_committed(), "{error}");
304        }
305    }
306
307    #[test]
308    fn protocol_errors_keep_structured_public_classifications() {
309        assert!(matches!(
310            Error::from(ProtocolError::NoQuorum),
311            Error::NoQuorum
312        ));
313        assert!(matches!(
314            Error::from(ProtocolError::Fenced("new owner".into())),
315            Error::Fenced(message) if message == "new owner"
316        ));
317        assert!(matches!(
318            Error::from(ProtocolError::SegmentDirectoryFull),
319            Error::SegmentDirectoryFull
320        ));
321        assert!(matches!(
322            Error::from(ProtocolError::ManifestUnavailable),
323            Error::ManifestUnavailable
324        ));
325        assert!(matches!(
326            Error::from(ProtocolError::InvalidManifest("bad register".into())),
327            Error::InvalidManifest(message) if message == "bad register"
328        ));
329        assert!(matches!(
330            Error::from(ProtocolError::Finalized),
331            Error::Finalized
332        ));
333        assert!(matches!(
334            Error::from(ProtocolError::ConflictingPrefix { record_index: 7 }),
335            Error::ConflictingPrefix { record_index: 7 }
336        ));
337        assert!(matches!(
338            Error::from(ProtocolError::RecoveryPrefixTooShort {
339                expected: 4,
340                actual: 3,
341            }),
342            Error::RecoveryPrefixTooShort {
343                expected: 4,
344                actual: 3,
345            }
346        ));
347        assert!(matches!(
348            Error::from(ProtocolError::SealDigestMismatch {
349                expected: "expected".into(),
350                actual: "actual".into(),
351            }),
352            Error::SealDigestMismatch { expected, actual }
353                if expected == "expected" && actual == "actual"
354        ));
355        assert!(matches!(
356            Error::from(ProtocolError::SealCrc32cMismatch {
357                expected: 1,
358                actual: 2,
359            }),
360            Error::SealCrc32cMismatch {
361                expected: 1,
362                actual: 2,
363            }
364        ));
365        assert!(matches!(
366            Error::from(ProtocolError::ManifestStore(
367                ManifestStoreError::Backend("failed".into())
368            )),
369            Error::ManifestStore(ManifestStoreError::Backend(message)) if message == "failed"
370        ));
371        assert!(matches!(
372            Error::from(ProtocolError::PipelineClosed),
373            Error::PipelineClosed
374        ));
375        assert!(matches!(
376            Error::from(ProtocolError::Transport(TransportError {
377                zone: 2,
378                code: TransportCode::Unavailable,
379                message: "down".into(),
380            })),
381            Error::Transport {
382                zone: 2,
383                code: TransportCode::Unavailable,
384                message,
385            } if message == "down"
386        ));
387    }
388}
chorus_client/error.rs

chorus_client/
error.rs
