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
use thiserror::Error;
/// Errors returned by Taquba's public API.
#[derive(Debug, Error)]
pub enum Error {
/// The underlying [SlateDB] storage layer reported a failure (transaction
/// commit failed, object-store I/O error, etc.).
///
/// [SlateDB]: https://github.com/slatedb/slatedb
#[error("storage error: {0}")]
Storage(#[from] slatedb::Error),
/// Failed to encode a job record to MessagePack before writing it to the
/// store. Typically indicates a payload that violates serde constraints.
#[error("serialization error: {0}")]
Serialization(#[from] rmp_serde::encode::Error),
/// Failed to decode a job record read back from the store. Usually means
/// the on-disk schema is from an incompatible Taquba version.
#[error("deserialization error: {0}")]
Deserialization(#[from] rmp_serde::decode::Error),
/// A job lookup by ID found no matching record.
#[error("job not found: {0}")]
JobNotFound(String),
/// An operation was issued against a job in the wrong state; for example,
/// `ack`-ing a record that is missing its `lease_expires_at`, or
/// `requeue_dead_job` on a record that is no longer in the dead state.
#[error("job is not in the expected state")]
InvalidState,
/// A value passed to [`crate::Queue::enqueue_with_kv`] exceeded the
/// configured maximum size for the user KV namespace. The cap is
/// enforced at the API boundary to keep bulk payload out of the LSM
/// tree; store large blobs in the underlying object store and put only
/// the pointer in KV. See [`crate::MAX_KV_VALUE_SIZE`].
#[error("kv value too large: {size} bytes (max {max})")]
KvValueTooLarge {
/// The value size that was rejected.
size: usize,
/// The configured maximum.
max: usize,
},
/// A caller-supplied [`crate::EnqueueOptions::id_override`] failed
/// validation. Caller-supplied ids must be 1-128 bytes of
/// `[A-Za-z0-9_-]`; ids that violate either bound are rejected at the
/// API boundary before any state is written.
#[error("invalid job id `{id}`: {reason}")]
InvalidId {
/// The id that was rejected.
id: String,
/// Why it was rejected.
reason: &'static str,
},
}
impl Error {
/// True if retrying the operation will not change the outcome; callers
/// should fast-fail rather than back off.
///
/// [`Self::Storage`] is conservatively treated as transient.
/// The remaining variants are programmer / data-shape errors
/// where retrying cannot help.
pub fn is_permanent(&self) -> bool {
match self {
Self::Serialization(_)
| Self::Deserialization(_)
| Self::JobNotFound(_)
| Self::InvalidState
| Self::KvValueTooLarge { .. }
| Self::InvalidId { .. } => true,
Self::Storage(_) => false,
}
}
}
/// Convenience alias for `Result<T, Error>` returned throughout the crate.
pub type Result<T> = std::result::Result<T, Error>;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn data_shape_and_state_variants_are_permanent() {
assert!(Error::JobNotFound("job-1".into()).is_permanent());
assert!(Error::InvalidState.is_permanent());
assert!(Error::KvValueTooLarge { size: 10, max: 5 }.is_permanent());
assert!(
Error::InvalidId {
id: "bad:id".into(),
reason: "invalid char",
}
.is_permanent()
);
}
}