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#[derive(Debug, Error)]
70pub enum EnqueueError {
71    #[error("idempotency conflict for existing row {existing_row_id:?}")]
72    IdempotencyConflict { existing_row_id: RowId },
73    #[error("migration mismatch: {detail}")]
74    MigrationMismatch { detail: String },
75    #[error("serialization: {detail}")]
76    Serialization { detail: String },
77    #[error("{operation}: {source}")]
78    Sql {
79        operation: &'static str,
80        #[source]
81        source: sqlx::Error,
82    },
83    #[error("{operation}: {kind}: {source}")]
84    Transient {
85        operation: &'static str,
86        kind: TransientKind,
87        #[source]
88        source: sqlx::Error,
89    },
90}
91
92impl EnqueueError {
93    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
94        match TransientKind::from_sqlx(&source) {
95            Some(kind) => Self::Transient {
96                operation,
97                kind,
98                source,
99            },
100            None => Self::Sql { operation, source },
101        }
102    }
103
104    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
105        Self::Serialization {
106            detail: detail.into(),
107        }
108    }
109}
110
111/// Errors returned by the migration-only importer.
112#[derive(Debug, Error)]
113pub enum ImportError {
114    #[error("immutable event identity conflict for existing row {existing_row_id:?}")]
115    IdentityConflict { existing_row_id: RowId },
116    #[error("imported delivery state conflict for existing row {existing_row_id:?}")]
117    ImportConflict { existing_row_id: RowId },
118    #[error("invalid imported delivery state: {source}")]
119    InvalidState {
120        #[source]
121        source: dovecote::ValidationError,
122    },
123    #[error("migration mismatch: {detail}")]
124    MigrationMismatch { detail: String },
125    #[error("serialization: {detail}")]
126    Serialization { detail: String },
127    #[error("{operation}: {source}")]
128    Sql {
129        operation: &'static str,
130        #[source]
131        source: sqlx::Error,
132    },
133    #[error("{operation}: {kind}: {source}")]
134    Transient {
135        operation: &'static str,
136        kind: TransientKind,
137        #[source]
138        source: sqlx::Error,
139    },
140}
141
142/// Errors returned by the migration-only delivery finalizer.
143#[derive(Debug, Error)]
144pub enum FinalizeError {
145    #[error("event row not found")]
146    NotFound,
147    #[error("delivery row {row_id:?} is not a canonical imported pending delivery")]
148    StateConflict { row_id: RowId },
149    #[error("invalid authoritative delivery timestamp: {source}")]
150    InvalidTimestamp {
151        #[source]
152        source: dovecote::ValidationError,
153    },
154    #[error("migration mismatch: {detail}")]
155    MigrationMismatch { detail: String },
156    #[error("serialization: {detail}")]
157    Serialization { detail: String },
158    #[error("{operation}: {source}")]
159    Sql {
160        operation: &'static str,
161        #[source]
162        source: sqlx::Error,
163    },
164    #[error("{operation}: {kind}: {source}")]
165    Transient {
166        operation: &'static str,
167        kind: TransientKind,
168        #[source]
169        source: sqlx::Error,
170    },
171}
172
173impl FinalizeError {
174    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
175        match TransientKind::from_sqlx(&source) {
176            Some(kind) => Self::Transient {
177                operation,
178                kind,
179                source,
180            },
181            None => Self::Sql { operation, source },
182        }
183    }
184}
185
186impl ImportError {
187    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
188        match TransientKind::from_sqlx(&source) {
189            Some(kind) => Self::Transient {
190                operation,
191                kind,
192                source,
193            },
194            None => Self::Sql { operation, source },
195        }
196    }
197
198    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
199        Self::Serialization {
200            detail: detail.into(),
201        }
202    }
203}
204
205/// Errors returned while selecting and claiming a batch of events.
206#[derive(Debug, Error)]
207pub enum ClaimError {
208    #[error("attempt counter overflow for row {row_id:?}")]
209    CounterOverflow { row_id: RowId },
210    #[error("operating-system entropy unavailable: {source}")]
211    EntropyUnavailable {
212        #[source]
213        source: getrandom::Error,
214    },
215    #[error("serialization: {detail}")]
216    Serialization { detail: String },
217    #[error("migration mismatch: {detail}")]
218    MigrationMismatch { detail: String },
219    #[error("{operation}: {source}")]
220    Sql {
221        operation: &'static str,
222        #[source]
223        source: sqlx::Error,
224    },
225    #[error("{operation}: {kind}: {source}")]
226    Transient {
227        operation: &'static str,
228        kind: TransientKind,
229        #[source]
230        source: sqlx::Error,
231    },
232}
233
234impl ClaimError {
235    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
236        match TransientKind::from_sqlx(&source) {
237            Some(kind) => Self::Transient {
238                operation,
239                kind,
240                source,
241            },
242            None => Self::Sql { operation, source },
243        }
244    }
245
246    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
247        Self::Serialization {
248            detail: detail.into(),
249        }
250    }
251}
252
253/// Errors returned by fenced post-claim mutations.
254#[derive(Debug, Error)]
255pub enum MutationError {
256    #[error("event row not found")]
257    NotFound,
258    #[error("illegal delivery transition from {state:?}")]
259    IllegalTransition { state: DeliveryState },
260    #[error("claim was lost")]
261    LostClaim,
262    #[error("migration mismatch: {detail}")]
263    MigrationMismatch { detail: String },
264    #[error("serialization: {detail}")]
265    Serialization { detail: String },
266    #[error("{operation}: {source}")]
267    Sql {
268        operation: &'static str,
269        #[source]
270        source: sqlx::Error,
271    },
272    #[error("{operation}: {kind}: {source}")]
273    Transient {
274        operation: &'static str,
275        kind: TransientKind,
276        #[source]
277        source: sqlx::Error,
278    },
279}
280
281/// Errors returned by live and snapshot paging.
282#[derive(Debug, Error)]
283pub enum PageError {
284    #[error("serialization: {detail}")]
285    Serialization { detail: String },
286    #[error("{operation}: {source}")]
287    Sql {
288        operation: &'static str,
289        #[source]
290        source: sqlx::Error,
291    },
292    #[error("{operation}: {kind}: {source}")]
293    Transient {
294        operation: &'static str,
295        kind: TransientKind,
296        #[source]
297        source: sqlx::Error,
298    },
299}
300
301impl PageError {
302    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
303        match TransientKind::from_sqlx(&source) {
304            Some(kind) => Self::Transient {
305                operation,
306                kind,
307                source,
308            },
309            None => Self::Sql { operation, source },
310        }
311    }
312
313    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
314        Self::Serialization {
315            detail: detail.into(),
316        }
317    }
318}
319
320impl MutationError {
321    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
322        match TransientKind::from_sqlx(&source) {
323            Some(kind) => Self::Transient {
324                operation,
325                kind,
326                source,
327            },
328            None => Self::Sql { operation, source },
329        }
330    }
331
332    pub(crate) fn serialization(detail: impl Into<String>) -> Self {
333        Self::Serialization {
334            detail: detail.into(),
335        }
336    }
337}
338
339#[derive(Debug, Error)]
340pub enum SchemaError {
341    #[error("migration mismatch: {detail}")]
342    MigrationMismatch { detail: String },
343    #[error("{operation}: {source}")]
344    Sql {
345        operation: &'static str,
346        #[source]
347        source: sqlx::Error,
348    },
349    #[error("{operation}: {kind}: {source}")]
350    Transient {
351        operation: &'static str,
352        kind: TransientKind,
353        #[source]
354        source: sqlx::Error,
355    },
356}
357
358impl SchemaError {
359    pub(crate) fn sql(operation: &'static str, source: sqlx::Error) -> Self {
360        match TransientKind::from_sqlx(&source) {
361            Some(kind) => Self::Transient {
362                operation,
363                kind,
364                source,
365            },
366            None => Self::Sql { operation, source },
367        }
368    }
369}
dovecote_sqlx_postgres/error.rs

dovecote_sqlx_postgres/
error.rs
