dovecote_sqlx_postgres/
error.rs

1//! Typed errors at the `PostgreSQL` adapter boundary.
2
3use dovecote::{DeliveryState, RowId};
4use thiserror::Error;
5
6/// `PostgreSQL` SQLSTATE categories for failures callers may retry as a whole
7/// operation.  The original `SQLx` error remains available as the source.
8#[derive(Clone, Copy, Debug, Eq, PartialEq)]
9#[non_exhaustive]
10pub enum TransientKind {
11    /// Serialization failure (`40001`).
12    SerializationFailure,
13    /// Deadlock detected (`40P01`).
14    DeadlockDetected,
15    /// Statement/query cancellation or lock timeout (`57014`/`55P03`).
16    StatementOrLockTimeout,
17}
18
19impl std::fmt::Display for TransientKind {
20    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
21        let label = match self {
22            Self::SerializationFailure => "serialization failure",
23            Self::DeadlockDetected => "deadlock detected",
24            Self::StatementOrLockTimeout => "statement or lock timeout",
25        };
26        formatter.write_str(label)
27    }
28}
29
30impl TransientKind {
31    pub(crate) fn from_sqlx(source: &sqlx::Error) -> Option<Self> {
32        Self::from_sqlstate(source.as_database_error()?.code()?.as_ref())
33    }
34
35    pub(crate) fn from_sqlstate(code: &str) -> Option<Self> {
36        match code {
37            "40001" => Some(Self::SerializationFailure),
38            "40P01" => Some(Self::DeadlockDetected),
39            "57014" | "55P03" => Some(Self::StatementOrLockTimeout),
40            _ => None,
41        }
42    }
43}
44
45#[cfg(test)]
46mod tests {
47    use super::TransientKind;
48
49    #[test]
50    fn postgres_transient_sqlstates_have_typed_categories() {
51        assert_eq!(
52            TransientKind::from_sqlstate("40001"),
53            Some(TransientKind::SerializationFailure)
54        );
55        assert_eq!(
56            TransientKind::from_sqlstate("40P01"),
57            Some(TransientKind::DeadlockDetected)
58        );
59        for code in ["57014", "55P03"] {
60            assert_eq!(
61                TransientKind::from_sqlstate(code),
62                Some(TransientKind::StatementOrLockTimeout)
63            );
64        }
65        assert_eq!(TransientKind::from_sqlstate("23505"), None);
66    }
67}
68
69/// Errors returned while enqueueing an event.
70#[derive(Debug, Error)]
71#[non_exhaustive]
72pub enum EnqueueError {
73    /// The immutable event identity already exists with different content.
74    #[error("idempotency conflict for existing row {existing_row_id:?}")]
75    IdempotencyConflict {
76        /// The existing event row whose identity conflicted.
77        existing_row_id: RowId,
78    },
79    #[error("migration mismatch: {detail}")]
80    /// The installed schema does not satisfy this adapter's migration contract.
81    MigrationMismatch {
82        /// Details identifying the incompatible schema contract.
83        detail: String,
84    },
85    /// A database value could not be reconstructed as a valid domain value.
86    #[error("serialization: {detail}")]
87    Serialization {
88        /// Details describing the invalid stored value.
89        detail: String,
90    },
91    #[error("{operation}: {source}")]
92    /// A non-transient SQL operation failed.
93    Sql {
94        /// The adapter operation that failed.
95        operation: &'static str,
96        /// The underlying `SQLx` error.
97        #[source]
98        source: sqlx::Error,
99    },
100    #[error("{operation}: {kind}: {source}")]
101    /// A SQL operation failed with a retryable `PostgreSQL` condition.
102    Transient {
103        /// The adapter operation that failed.
104        operation: &'static str,
105        /// The retryable `PostgreSQL` failure category.
106        kind: TransientKind,
107        /// The underlying `SQLx` error.
108        #[source]
109        source: sqlx::Error,
110    },
111}
112
113impl EnqueueError {
114    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
115        match TransientKind::from_sqlx(&source) {
116            Some(kind) => Self::Transient {
117                operation,
118                kind,
119                source,
120            },
121            None => Self::Sql { operation, source },
122        }
123    }
124
125    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
126        Self::Serialization {
127            detail: detail.into(),
128        }
129    }
130}
131
132/// Errors returned by the migration-only importer.
133#[derive(Debug, Error)]
134#[non_exhaustive]
135pub enum ImportError {
136    /// The immutable event identity conflicts with an existing row.
137    #[error("immutable event identity conflict for existing row {existing_row_id:?}")]
138    IdentityConflict {
139        /// The existing event row whose identity conflicted.
140        existing_row_id: RowId,
141    },
142    #[error("imported delivery state conflict for existing row {existing_row_id:?}")]
143    /// The imported delivery state conflicts with an existing row.
144    ImportConflict {
145        /// The existing event row whose delivery state conflicted.
146        existing_row_id: RowId,
147    },
148    #[error("invalid imported delivery state: {source}")]
149    /// The supplied legacy delivery state is invalid.
150    InvalidState {
151        /// The validation failure from the domain state.
152        #[source]
153        source: dovecote::ValidationError,
154    },
155    #[error("migration mismatch: {detail}")]
156    /// The installed schema does not satisfy this adapter's migration contract.
157    MigrationMismatch {
158        /// Details identifying the incompatible schema contract.
159        detail: String,
160    },
161    #[error("serialization: {detail}")]
162    /// A database value could not be reconstructed as a valid domain value.
163    Serialization {
164        /// Details describing the invalid stored value.
165        detail: String,
166    },
167    #[error("{operation}: {source}")]
168    /// A non-transient SQL operation failed.
169    Sql {
170        /// The adapter operation that failed.
171        operation: &'static str,
172        /// The underlying `SQLx` error.
173        #[source]
174        source: sqlx::Error,
175    },
176    #[error("{operation}: {kind}: {source}")]
177    /// A SQL operation failed with a retryable `PostgreSQL` condition.
178    Transient {
179        /// The adapter operation that failed.
180        operation: &'static str,
181        /// The retryable `PostgreSQL` failure category.
182        kind: TransientKind,
183        /// The underlying `SQLx` error.
184        #[source]
185        source: sqlx::Error,
186    },
187}
188
189/// Errors returned by the migration-only delivery finalizer.
190#[derive(Debug, Error)]
191#[non_exhaustive]
192pub enum FinalizeError {
193    /// The requested event row does not exist.
194    #[error("event row not found")]
195    NotFound,
196    #[error("delivery row {row_id:?} is not a canonical imported pending delivery")]
197    /// The delivery row is not in the canonical imported pending state.
198    StateConflict {
199        /// The event row whose delivery state conflicted.
200        row_id: RowId,
201    },
202    #[error("invalid authoritative delivery timestamp: {source}")]
203    /// The supplied authoritative timestamp is invalid.
204    InvalidTimestamp {
205        /// The timestamp validation failure from the domain type.
206        #[source]
207        source: dovecote::ValidationError,
208    },
209    #[error("migration mismatch: {detail}")]
210    /// The installed schema does not satisfy this adapter's migration contract.
211    MigrationMismatch {
212        /// Details identifying the incompatible schema contract.
213        detail: String,
214    },
215    #[error("serialization: {detail}")]
216    /// A database value could not be reconstructed as a valid domain value.
217    Serialization {
218        /// Details describing the invalid stored value.
219        detail: String,
220    },
221    #[error("{operation}: {source}")]
222    /// A non-transient SQL operation failed.
223    Sql {
224        /// The adapter operation that failed.
225        operation: &'static str,
226        /// The underlying `SQLx` error.
227        #[source]
228        source: sqlx::Error,
229    },
230    #[error("{operation}: {kind}: {source}")]
231    /// A SQL operation failed with a retryable `PostgreSQL` condition.
232    Transient {
233        /// The adapter operation that failed.
234        operation: &'static str,
235        /// The retryable `PostgreSQL` failure category.
236        kind: TransientKind,
237        /// The underlying `SQLx` error.
238        #[source]
239        source: sqlx::Error,
240    },
241}
242
243impl FinalizeError {
244    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
245        match TransientKind::from_sqlx(&source) {
246            Some(kind) => Self::Transient {
247                operation,
248                kind,
249                source,
250            },
251            None => Self::Sql { operation, source },
252        }
253    }
254}
255
256impl ImportError {
257    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
258        match TransientKind::from_sqlx(&source) {
259            Some(kind) => Self::Transient {
260                operation,
261                kind,
262                source,
263            },
264            None => Self::Sql { operation, source },
265        }
266    }
267
268    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
269        Self::Serialization {
270            detail: detail.into(),
271        }
272    }
273}
274
275/// Errors returned while selecting and claiming a batch of events.
276#[derive(Debug, Error)]
277#[non_exhaustive]
278pub enum ClaimError {
279    /// The delivery attempt counter cannot be incremented safely.
280    #[error("attempt counter overflow for row {row_id:?}")]
281    CounterOverflow {
282        /// The event row whose attempt count overflowed.
283        row_id: RowId,
284    },
285    #[error("operating-system entropy unavailable: {source}")]
286    /// The operating system could not provide claim-token entropy.
287    EntropyUnavailable {
288        /// The underlying entropy-provider error.
289        #[source]
290        source: getrandom::Error,
291    },
292    #[error("serialization: {detail}")]
293    /// A database value could not be reconstructed as a valid domain value.
294    Serialization {
295        /// Details describing the invalid stored value.
296        detail: String,
297    },
298    #[error("migration mismatch: {detail}")]
299    /// The installed schema does not satisfy this adapter's migration contract.
300    MigrationMismatch {
301        /// Details identifying the incompatible schema contract.
302        detail: String,
303    },
304    #[error("{operation}: {source}")]
305    /// A non-transient SQL operation failed.
306    Sql {
307        /// The adapter operation that failed.
308        operation: &'static str,
309        /// The underlying `SQLx` error.
310        #[source]
311        source: sqlx::Error,
312    },
313    #[error("{operation}: {kind}: {source}")]
314    /// A SQL operation failed with a retryable `PostgreSQL` condition.
315    Transient {
316        /// The adapter operation that failed.
317        operation: &'static str,
318        /// The retryable `PostgreSQL` failure category.
319        kind: TransientKind,
320        /// The underlying `SQLx` error.
321        #[source]
322        source: sqlx::Error,
323    },
324}
325
326impl ClaimError {
327    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
328        match TransientKind::from_sqlx(&source) {
329            Some(kind) => Self::Transient {
330                operation,
331                kind,
332                source,
333            },
334            None => Self::Sql { operation, source },
335        }
336    }
337
338    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
339        Self::Serialization {
340            detail: detail.into(),
341        }
342    }
343}
344
345/// Errors returned by fenced post-claim mutations.
346#[derive(Debug, Error)]
347#[non_exhaustive]
348pub enum MutationError {
349    /// The requested event row does not exist.
350    #[error("event row not found")]
351    NotFound,
352    #[error("illegal delivery transition from {state:?}")]
353    /// The current delivery state cannot perform this mutation.
354    IllegalTransition {
355        /// The delivery state that rejected the mutation.
356        state: DeliveryState,
357    },
358    #[error("claim was lost")]
359    /// The claim token is stale or the lease has expired.
360    LostClaim,
361    #[error("migration mismatch: {detail}")]
362    /// The installed schema does not satisfy this adapter's migration contract.
363    MigrationMismatch {
364        /// Details identifying the incompatible schema contract.
365        detail: String,
366    },
367    #[error("serialization: {detail}")]
368    /// A database value could not be reconstructed as a valid domain value.
369    Serialization {
370        /// Details describing the invalid stored value.
371        detail: String,
372    },
373    #[error("{operation}: {source}")]
374    /// A non-transient SQL operation failed.
375    Sql {
376        /// The adapter operation that failed.
377        operation: &'static str,
378        /// The underlying `SQLx` error.
379        #[source]
380        source: sqlx::Error,
381    },
382    #[error("{operation}: {kind}: {source}")]
383    /// A SQL operation failed with a retryable `PostgreSQL` condition.
384    Transient {
385        /// The adapter operation that failed.
386        operation: &'static str,
387        /// The retryable `PostgreSQL` failure category.
388        kind: TransientKind,
389        /// The underlying `SQLx` error.
390        #[source]
391        source: sqlx::Error,
392    },
393}
394
395/// Errors returned by live and snapshot paging.
396#[derive(Debug, Error)]
397#[non_exhaustive]
398pub enum PageError {
399    /// A database value could not be reconstructed as a valid domain value.
400    #[error("serialization: {detail}")]
401    Serialization {
402        /// Details describing the invalid stored value.
403        detail: String,
404    },
405    #[error("{operation}: {source}")]
406    /// A non-transient SQL operation failed.
407    Sql {
408        /// The adapter operation that failed.
409        operation: &'static str,
410        /// The underlying `SQLx` error.
411        #[source]
412        source: sqlx::Error,
413    },
414    #[error("{operation}: {kind}: {source}")]
415    /// A SQL operation failed with a retryable `PostgreSQL` condition.
416    Transient {
417        /// The adapter operation that failed.
418        operation: &'static str,
419        /// The retryable `PostgreSQL` failure category.
420        kind: TransientKind,
421        /// The underlying `SQLx` error.
422        #[source]
423        source: sqlx::Error,
424    },
425}
426
427impl PageError {
428    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
429        match TransientKind::from_sqlx(&source) {
430            Some(kind) => Self::Transient {
431                operation,
432                kind,
433                source,
434            },
435            None => Self::Sql { operation, source },
436        }
437    }
438
439    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
440        Self::Serialization {
441            detail: detail.into(),
442        }
443    }
444}
445
446impl MutationError {
447    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
448        match TransientKind::from_sqlx(&source) {
449            Some(kind) => Self::Transient {
450                operation,
451                kind,
452                source,
453            },
454            None => Self::Sql { operation, source },
455        }
456    }
457
458    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
459        Self::Serialization {
460            detail: detail.into(),
461        }
462    }
463}
464
465/// Errors returned while checking the installed schema.
466#[derive(Debug, Error)]
467#[non_exhaustive]
468pub enum SchemaError {
469    /// The installed schema does not satisfy this adapter's migration contract.
470    #[error("migration mismatch: {detail}")]
471    MigrationMismatch {
472        /// Details identifying the incompatible schema contract.
473        detail: String,
474    },
475    #[error("{operation}: {source}")]
476    /// A non-transient SQL operation failed.
477    Sql {
478        /// The adapter operation that failed.
479        operation: &'static str,
480        /// The underlying `SQLx` error.
481        #[source]
482        source: sqlx::Error,
483    },
484    #[error("{operation}: {kind}: {source}")]
485    /// A SQL operation failed with a retryable `PostgreSQL` condition.
486    Transient {
487        /// The adapter operation that failed.
488        operation: &'static str,
489        /// The retryable `PostgreSQL` failure category.
490        kind: TransientKind,
491        /// The underlying `SQLx` error.
492        #[source]
493        source: sqlx::Error,
494    },
495}
496
497impl SchemaError {
498    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
499        match TransientKind::from_sqlx(&source) {
500            Some(kind) => Self::Transient {
501                operation,
502                kind,
503                source,
504            },
505            None => Self::Sql { operation, source },
506        }
507    }
508}
dovecote_sqlx_postgres/error.rs

dovecote_sqlx_postgres/
error.rs
