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
138
139
140
141
142
143
144
145
146
147
use thiserror::Error;
/// Errors returned by the runtime's submission and worker paths.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum Error {
/// A step job is missing the [`crate::HEADER_RUN_ID`] header.
/// Permanent: a misconfigured job will not become valid on retry.
#[error("step job is missing header `{0}`")]
MissingHeader(&'static str),
/// A step job's [`crate::HEADER_STEP`] header is not a valid `u32`.
/// Permanent: header value won't change across retries.
#[error("step job has invalid `{header}` header `{value}`")]
InvalidStepHeader {
/// Header name.
header: &'static str,
/// Offending value.
value: String,
},
/// A submission included a user header starting with the reserved
/// `workflow.*` prefix. The runtime owns that prefix; submitters must use
/// any other key.
#[error("submission header `{0}` uses the reserved `workflow.*` prefix")]
ReservedHeaderInSubmit(String),
/// A re-submission of an active `run_id` carried `spec.input` bytes
/// that differ from the original submission's. Reusing a `run_id`
/// with new input is treated as a programmer error: pick a fresh
/// `run_id` for a new run, or wait for the active one to terminate.
#[error("run `{0}` is active with a different input; pick a fresh run_id")]
InputMismatch(String),
/// A caller KV key passed via [`crate::RunSpec::kv_writes`] or staged
/// through an [`crate::EffectsHandle`] starts with the reserved
/// `workflow/` prefix. The runtime owns that prefix; callers must use
/// any other key.
#[error("kv key `{0}` uses the reserved `workflow/` prefix")]
ReservedKvKey(String),
/// A key was staged through an [`crate::EffectsHandle`] for both a
/// write and a delete within one step. The combination has no defined
/// order in the settlement transaction and is rejected when the
/// second operation is staged.
#[error("kv key `{0}` is staged for both a write and a delete")]
ConflictingKvEffect(String),
/// An effect was staged through an [`crate::EffectsHandle`] or
/// [`crate::TerminalEffects`] clone after its delivery returned.
/// Effects are collected when the runner or hook returns; an effect
/// staged after that point cannot join the settlement.
#[error("the effects handle is sealed; its delivery has returned")]
EffectsSealed,
/// Underlying error from a Taquba queue operation.
#[error(transparent)]
Queue(#[from] taquba::Error),
/// Reading or writing a blob in object storage failed.
#[error("object store error: {0}")]
Store(#[from] taquba::object_store::Error),
/// Serializing a value for workflow storage failed.
#[error("serialization error: {0}")]
Serialization(#[from] rmp_serde::encode::Error),
}
impl Error {
/// True if retrying the operation will not change the outcome; callers
/// should fast-fail (e.g. dead-letter a step, mark a submission as
/// failed) rather than back off and try again.
///
/// [`Self::Queue`] delegates to [`taquba::Error::is_permanent`].
pub fn is_permanent(&self) -> bool {
match self {
Self::MissingHeader(_)
| Self::InvalidStepHeader { .. }
| Self::ReservedHeaderInSubmit(_)
| Self::InputMismatch(_)
| Self::ReservedKvKey(_)
| Self::ConflictingKvEffect(_)
| Self::EffectsSealed
| Self::Serialization(_) => true,
Self::Queue(e) => e.is_permanent(),
Self::Store(_) => false,
}
}
}
/// Result alias used throughout the crate.
pub type Result<T> = std::result::Result<T, Error>;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn workflow_variants_are_permanent() {
assert!(Error::MissingHeader("workflow.run_id").is_permanent());
assert!(
Error::InvalidStepHeader {
header: "workflow.step",
value: "not-a-u32".into(),
}
.is_permanent()
);
assert!(Error::ReservedHeaderInSubmit("workflow.foo".into()).is_permanent());
assert!(Error::InputMismatch("run-1".into()).is_permanent());
assert!(Error::ReservedKvKey("workflow/x".into()).is_permanent());
assert!(Error::ConflictingKvEffect("k".into()).is_permanent());
assert!(Error::EffectsSealed.is_permanent());
}
#[test]
fn queue_classifies_per_inner_variant() {
assert!(Error::Queue(taquba::Error::JobNotFound("job-1".into())).is_permanent());
assert!(Error::Queue(taquba::Error::InvalidState).is_permanent());
assert!(Error::Queue(taquba::Error::KvValueTooLarge { size: 10, max: 5 }).is_permanent());
}
#[test]
fn store_is_transient() {
let store_err = taquba::object_store::Error::NotFound {
path: "x".into(),
source: "missing".into(),
};
assert!(!Error::Store(store_err).is_permanent());
}
#[test]
fn serialization_is_permanent() {
struct BadSerialize;
impl serde::Serialize for BadSerialize {
fn serialize<S>(&self, _serializer: S) -> std::result::Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
Err(serde::ser::Error::custom("serialization failed"))
}
}
let err = rmp_serde::to_vec_named(&BadSerialize).unwrap_err();
assert!(Error::Serialization(err).is_permanent());
}
}