1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
//! Crate error and result types.
use thiserror::Error;
use crate::types::TaskId;
/// Result type used by Steda.
pub type Result<T> = std::result::Result<T, Error>;
/// Errors returned by the queue handle, worker, and task context.
#[derive(Error, Debug)]
pub enum Error {
/// Error returned by `SQLx` while talking to Postgres.
#[error("database error: {0}")]
Database(#[from] sqlx::Error),
/// Invalid queue or worker options.
#[error("{0}")]
InvalidOptions(String),
/// Queue name was omitted.
#[error("queue name must be provided")]
MissingQueueName,
/// Queue name exceeds `PostgreSQL` identifier-derived byte limit.
#[error("queue name {name:?} is too long (max {max} bytes)")]
QueueNameTooLong {
/// Provided queue name.
name: String,
/// Maximum allowed UTF-8 byte length.
max: usize,
},
/// Task was not found in storage.
#[error("task {0} not found")]
TaskNotFound(TaskId),
/// A durable task reference names a different task definition than the persisted task.
#[error("task {task_id} is persisted as {actual:?}, but the task reference names {expected:?}")]
TaskNameMismatch {
/// Logical task identifier.
task_id: TaskId,
/// Task name carried by the durable reference.
expected: String,
/// Task name stored with the logical task.
actual: String,
},
/// Task reached a terminal failed state.
#[error("task failed: {failure}")]
TaskFailed {
/// Persisted task failure payload.
failure: serde_json::Value,
},
/// Task intentionally suspended itself while sleeping.
#[error("task suspended")]
Suspended,
/// Task or run was cancelled.
#[error("task cancelled")]
Cancelled,
/// Run had already failed when attempting a state transition.
#[error("task already failed")]
FailedRun,
/// The worker no longer owns the run because its finite lease expired.
#[error("task lease lost")]
LeaseLost,
/// An idempotency key was reused for a different spawn request.
#[error("idempotency key conflicts with an existing task request")]
IdempotencyConflict,
/// Generic timeout.
#[error("{0}")]
Timeout(String),
/// Headers in storage were not a JSON object.
#[error("invalid task headers: {0}")]
InvalidTaskHeaders(String),
/// JSON serialization or deserialization failed.
#[error("serialization error: {0}")]
Serialization(#[from] serde_json::Error),
/// Catch-all error for cases that do not deserve a dedicated variant.
#[error("{0}")]
Other(String),
}
impl Error {
/// Returns true when the error represents a deliberate task suspension.
pub const fn is_suspended(&self) -> bool {
matches!(self, Self::Suspended)
}
/// Returns true when the error represents task cancellation.
pub const fn is_cancelled(&self) -> bool {
matches!(self, Self::Cancelled)
}
/// Returns true when the error represents an already failed run.
pub const fn is_failed_run(&self) -> bool {
matches!(self, Self::FailedRun)
}
/// Returns true when the current worker has lost its finite run lease.
pub const fn is_lease_lost(&self) -> bool {
matches!(self, Self::LeaseLost)
}
}
/// Maps Steda queue SQLSTATE errors into typed Rust errors.
pub(crate) fn map_sqlx_error(e: sqlx::Error) -> Error {
if let sqlx::Error::Database(db_err) = &e {
match db_err.code().as_deref() {
// Task cancelled.
Some("ST001") => return Error::Cancelled,
// Run already failed.
Some("ST002") => return Error::FailedRun,
// Finite run lease expired.
Some("ST003") => return Error::LeaseLost,
// Idempotency key reused for a different request.
Some("ST004") => return Error::IdempotencyConflict,
_ => {}
}
}
Error::Database(e)
}