a3s_flow/
error.rs

1use std::fmt;
2
3use thiserror::Error;
4
5use crate::runtime_build::RuntimeBuildId;
6
7/// Crate-local result type.
8pub type Result<T> = std::result::Result<T, FlowError>;
9
10/// Errors surfaced by the workflow engine and runtime adapters.
11#[derive(Error)]
12#[non_exhaustive]
13pub enum FlowError {
14    /// A requested workflow run does not exist.
15    #[error("workflow run not found: {0}")]
16    RunNotFound(String),
17
18    /// An operation attempted to append work to a terminal run.
19    #[error("workflow run {0} is already terminal")]
20    RunTerminal(String),
21
22    /// A run identifier violates the storage-safe identifier contract.
23    #[error("workflow run id is invalid: {0}")]
24    InvalidRunId(String),
25
26    /// A continue-as-new chain contains a repeated run.
27    #[error("continue-as-new chain contains a cycle at workflow run {0}")]
28    ContinueAsNewCycle(String),
29
30    /// A continue-as-new chain exceeds the configured traversal bound.
31    #[error("continue-as-new chain exceeded the configured limit of {0} hops")]
32    ContinueAsNewLimitExceeded(usize),
33
34    /// Parent and child ownership links contain a cycle.
35    #[error("child workflow graph contains a cycle at workflow run {0}")]
36    ChildWorkflowCycle(String),
37
38    /// A child workflow request exceeds the configured nesting bound.
39    #[error("child workflow nesting exceeded the configured depth of {0}")]
40    ChildWorkflowDepthExceeded(usize),
41
42    /// An idempotent run start disagrees with the existing run definition.
43    #[error("workflow run {run_id} conflicts with existing run: {reason}")]
44    RunConflict {
45        /// Existing run identifier.
46        run_id: String,
47        /// Description of the conflicting immutable input.
48        reason: String,
49    },
50
51    /// A runtime build identity is empty or malformed.
52    #[error("invalid runtime build identity: {0}")]
53    InvalidRuntimeBuildId(String),
54
55    /// A workflow patch identity is empty, malformed, or too long.
56    #[error("invalid workflow patch identity: {0}")]
57    InvalidWorkflowPatchId(String),
58
59    /// No configured runtime can replay the run's pinned build.
60    #[error(
61        "workflow run {run_id} requires runtime build {required_build_id:?}, but the configured current build is {current_build_id:?}"
62    )]
63    RuntimeBuildUnavailable {
64        /// Run that requires replay admission.
65        run_id: String,
66        /// Runtime build pinned by the run, if any.
67        required_build_id: Option<RuntimeBuildId>,
68        /// Runtime build configured on the attempted executor, if any.
69        current_build_id: Option<RuntimeBuildId>,
70    },
71
72    /// No task queue route is registered for a required runtime build.
73    #[error("no Flow task route is registered for runtime build {required_build_id:?}")]
74    RuntimeBuildRouteNotFound {
75        /// Runtime build required by the task, if pinned.
76        required_build_id: Option<RuntimeBuildId>,
77    },
78
79    /// Runtime replay emitted a command that conflicts with durable history.
80    #[error("non-deterministic workflow replay for run {run_id}: {reason}")]
81    NonDeterministic {
82        /// Run whose replay diverged.
83        run_id: String,
84        /// Description of the durable command mismatch.
85        reason: String,
86    },
87
88    /// An optimistic event append used a stale expected sequence.
89    #[error(
90        "event sequence conflict for run {run_id}: expected {expected_sequence}, actual {actual_sequence}"
91    )]
92    EventConflict {
93        /// Run whose history changed concurrently.
94        run_id: String,
95        /// Last sequence assumed by the caller.
96        expected_sequence: u64,
97        /// Last sequence currently stored.
98        actual_sequence: u64,
99    },
100
101    /// The original token remains available for programmatic routing, while
102    /// `Display` and `Debug` deliberately redact it.
103    #[error("active hook token not found (value redacted)")]
104    HookTokenNotFound(String),
105
106    /// A queue lease was lost before its task could be acknowledged.
107    #[error("workflow task lease is no longer active: {0}")]
108    LeaseLost(String),
109
110    /// The conflicting token remains available for programmatic handling,
111    /// while `Display` and `Debug` deliberately redact it.
112    #[error(
113        "active hook token is already used by run {existing_run_id} hook {existing_hook_id} (value redacted)"
114    )]
115    HookTokenConflict {
116        /// Conflicting bearer token, retained only for programmatic recovery.
117        token: String,
118        /// Run that already owns the token.
119        existing_run_id: String,
120        /// Hook that already owns the token.
121        existing_hook_id: String,
122    },
123
124    /// A hook retry conflicts with its durable identity or resolution.
125    #[error("hook {hook_id} for workflow run {run_id} conflicts with request: {reason}")]
126    HookConflict {
127        /// Run that owns the hook.
128        run_id: String,
129        /// Replay-stable hook identity.
130        hook_id: String,
131        /// Description of the conflicting request.
132        reason: String,
133    },
134
135    /// A signal retry conflicts with its durable identity or payload.
136    #[error("signal {signal_id} for workflow run {run_id} conflicts with request: {reason}")]
137    SignalConflict {
138        /// Run targeted by the signal.
139        run_id: String,
140        /// Caller-owned signal identity.
141        signal_id: String,
142        /// Description of the conflicting delivery.
143        reason: String,
144    },
145
146    /// A workflow definition violates a static invariant.
147    #[error("invalid workflow definition: {0}")]
148    InvalidWorkflow(String),
149
150    /// An event or command violates the current durable state.
151    #[error("invalid state transition: {0}")]
152    InvalidTransition(String),
153
154    /// Worker settings cannot provide the requested execution guarantees.
155    #[error("invalid worker configuration: {0}")]
156    InvalidWorkerConfiguration(String),
157
158    /// An external task manager rejected or failed an operation.
159    #[error("task manager error: {0}")]
160    TaskManagement(String),
161
162    /// A durable event store rejected or failed an operation.
163    #[error("event store error: {0}")]
164    Store(String),
165
166    /// A workflow or step runtime failed outside application commands.
167    #[error("runtime error: {0}")]
168    Runtime(String),
169
170    /// JSON serialization or deserialization failed.
171    #[error("serialization error: {0}")]
172    Serialization(#[from] serde_json::Error),
173
174    /// Filesystem, process, or stream I/O failed.
175    #[error("io error: {0}")]
176    Io(#[from] std::io::Error),
177
178    /// Replay emitted more commands than the configured safety bound.
179    #[error("workflow replay exceeded {0} iterations")]
180    ReplayLimitExceeded(usize),
181}
182
183// Error values can retain callback tokens for programmatic recovery, but
184// diagnostics must never reveal those bearer credentials. Keep ordinary
185// variants structurally useful while replacing token fields in Debug output.
186impl fmt::Debug for FlowError {
187    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
188        match self {
189            Self::RunNotFound(run_id) => {
190                formatter.debug_tuple("RunNotFound").field(run_id).finish()
191            }
192            Self::RunTerminal(run_id) => {
193                formatter.debug_tuple("RunTerminal").field(run_id).finish()
194            }
195            Self::InvalidRunId(run_id) => {
196                formatter.debug_tuple("InvalidRunId").field(run_id).finish()
197            }
198            Self::ContinueAsNewCycle(run_id) => formatter
199                .debug_tuple("ContinueAsNewCycle")
200                .field(run_id)
201                .finish(),
202            Self::ContinueAsNewLimitExceeded(limit) => formatter
203                .debug_tuple("ContinueAsNewLimitExceeded")
204                .field(limit)
205                .finish(),
206            Self::ChildWorkflowCycle(run_id) => formatter
207                .debug_tuple("ChildWorkflowCycle")
208                .field(run_id)
209                .finish(),
210            Self::ChildWorkflowDepthExceeded(limit) => formatter
211                .debug_tuple("ChildWorkflowDepthExceeded")
212                .field(limit)
213                .finish(),
214            Self::RunConflict { run_id, reason } => formatter
215                .debug_struct("RunConflict")
216                .field("run_id", run_id)
217                .field("reason", reason)
218                .finish(),
219            Self::InvalidRuntimeBuildId(reason) => formatter
220                .debug_tuple("InvalidRuntimeBuildId")
221                .field(reason)
222                .finish(),
223            Self::InvalidWorkflowPatchId(reason) => formatter
224                .debug_tuple("InvalidWorkflowPatchId")
225                .field(reason)
226                .finish(),
227            Self::RuntimeBuildUnavailable {
228                run_id,
229                required_build_id,
230                current_build_id,
231            } => formatter
232                .debug_struct("RuntimeBuildUnavailable")
233                .field("run_id", run_id)
234                .field("required_build_id", required_build_id)
235                .field("current_build_id", current_build_id)
236                .finish(),
237            Self::RuntimeBuildRouteNotFound { required_build_id } => formatter
238                .debug_struct("RuntimeBuildRouteNotFound")
239                .field("required_build_id", required_build_id)
240                .finish(),
241            Self::NonDeterministic { run_id, reason } => formatter
242                .debug_struct("NonDeterministic")
243                .field("run_id", run_id)
244                .field("reason", reason)
245                .finish(),
246            Self::EventConflict {
247                run_id,
248                expected_sequence,
249                actual_sequence,
250            } => formatter
251                .debug_struct("EventConflict")
252                .field("run_id", run_id)
253                .field("expected_sequence", expected_sequence)
254                .field("actual_sequence", actual_sequence)
255                .finish(),
256            Self::HookTokenNotFound(_) => formatter
257                .debug_tuple("HookTokenNotFound")
258                .field(&"<redacted>")
259                .finish(),
260            Self::LeaseLost(lease_id) => {
261                formatter.debug_tuple("LeaseLost").field(lease_id).finish()
262            }
263            Self::HookTokenConflict {
264                existing_run_id,
265                existing_hook_id,
266                ..
267            } => formatter
268                .debug_struct("HookTokenConflict")
269                .field("token", &"<redacted>")
270                .field("existing_run_id", existing_run_id)
271                .field("existing_hook_id", existing_hook_id)
272                .finish(),
273            Self::HookConflict {
274                run_id,
275                hook_id,
276                reason,
277            } => formatter
278                .debug_struct("HookConflict")
279                .field("run_id", run_id)
280                .field("hook_id", hook_id)
281                .field("reason", reason)
282                .finish(),
283            Self::SignalConflict {
284                run_id,
285                signal_id,
286                reason,
287            } => formatter
288                .debug_struct("SignalConflict")
289                .field("run_id", run_id)
290                .field("signal_id", signal_id)
291                .field("reason", reason)
292                .finish(),
293            Self::InvalidWorkflow(message) => formatter
294                .debug_tuple("InvalidWorkflow")
295                .field(message)
296                .finish(),
297            Self::InvalidTransition(message) => formatter
298                .debug_tuple("InvalidTransition")
299                .field(message)
300                .finish(),
301            Self::InvalidWorkerConfiguration(message) => formatter
302                .debug_tuple("InvalidWorkerConfiguration")
303                .field(message)
304                .finish(),
305            Self::TaskManagement(message) => formatter
306                .debug_tuple("TaskManagement")
307                .field(message)
308                .finish(),
309            Self::Store(message) => formatter.debug_tuple("Store").field(message).finish(),
310            Self::Runtime(message) => formatter.debug_tuple("Runtime").field(message).finish(),
311            Self::Serialization(error) => {
312                formatter.debug_tuple("Serialization").field(error).finish()
313            }
314            Self::Io(error) => formatter.debug_tuple("Io").field(error).finish(),
315            Self::ReplayLimitExceeded(limit) => formatter
316                .debug_tuple("ReplayLimitExceeded")
317                .field(limit)
318                .finish(),
319        }
320    }
321}
a3s_flow/error.rs

a3s_flow/
error.rs
