dex_sdk/
sdk_error.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Super Durable Source License 1.0.
4// You may not use this file except in compliance with the License.
5// See the LICENSE file in the repository root.
6//
7// SPDX-License-Identifier: LicenseRef-Super-Durable-1.0
8
9use std::error::Error;
10use std::fmt::{Display, Formatter};
11
12use dex_protocol::dex::{ErrorResponse, ErrorSubStatus as ProtoErrorSubStatus};
13use prost::Message;
14use prost_types::Any;
15use tonic::Status;
16
17use crate::{FlowErrorType, FlowStatus, GrpcCode};
18
19/// Result returned by fallible Dex SDK operations.
20pub type SdkResult<T> = Result<T, SdkError>;
21
22#[derive(Clone, Copy, Debug, Eq, PartialEq)]
23/// Provides Dex-specific detail beyond a gRPC status code.
24pub enum ErrorSubStatus {
25    /// Dex returned no specific status.
26    Uncategorized,
27    /// A start request targeted an already-existing Flow ID.
28    FlowAlreadyStarted,
29    /// The requested Flow does not exist or is no longer active.
30    FlowNotExists,
31    /// A Worker Step or RPC invocation failed.
32    WorkerApiError,
33    /// A long-poll deadline elapsed while the Flow remained active.
34    LongPollTimeout,
35}
36
37#[derive(Debug)]
38/// Preserves one failed Dex service operation and its transport source.
39pub struct ServiceError {
40    code: GrpcCode,
41    sub_status: ErrorSubStatus,
42    detail: String,
43    operation: &'static str,
44    flow_id: Option<String>,
45    source: Status,
46}
47
48impl ServiceError {
49    /// Returns the outer gRPC status code.
50    pub fn code(&self) -> GrpcCode {
51        self.code
52    }
53
54    /// Returns the decoded Dex-specific status.
55    pub fn sub_status(&self) -> ErrorSubStatus {
56        self.sub_status
57    }
58
59    /// Returns the most specific server or transport detail available.
60    pub fn detail(&self) -> &str {
61        &self.detail
62    }
63
64    /// Returns the SDK operation name that failed.
65    pub fn operation(&self) -> &str {
66        self.operation
67    }
68
69    /// Returns the targeted Flow ID when the operation had one.
70    pub fn flow_id(&self) -> Option<&str> {
71        self.flow_id.as_deref()
72    }
73
74    pub(crate) fn local(operation: &'static str, detail: impl Into<String>) -> Self {
75        let detail = detail.into();
76        Self {
77            code: GrpcCode::Unknown,
78            sub_status: ErrorSubStatus::Uncategorized,
79            detail: detail.clone(),
80            operation,
81            flow_id: None,
82            source: Status::unknown(detail),
83        }
84    }
85}
86
87impl Display for ServiceError {
88    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
89        formatter.write_str(&self.detail)
90    }
91}
92
93impl Error for ServiceError {
94    fn source(&self) -> Option<&(dyn Error + 'static)> {
95        Some(&self.source)
96    }
97}
98
99#[derive(Clone, Debug, Eq, PartialEq)]
100/// Preserves the original Worker-side error carried by a service failure.
101pub struct WorkerError {
102    code: Option<GrpcCode>,
103    error_type: String,
104    detail: String,
105}
106
107impl WorkerError {
108    /// Returns the Worker gRPC status when available.
109    pub fn code(&self) -> Option<GrpcCode> {
110        self.code
111    }
112
113    /// Returns the Worker exception or error type name.
114    pub fn error_type(&self) -> &str {
115        &self.error_type
116    }
117
118    /// Returns the Worker-provided failure detail.
119    pub fn detail(&self) -> &str {
120        &self.detail
121    }
122}
123
124#[derive(Debug)]
125/// Reports service, definition, value, handler, and Flow-completion failures.
126///
127/// Match specific variants when recovery differs. [`Self::service_error`] extracts common gRPC
128/// metadata from every service-backed variant.
129pub enum SdkError {
130    /// A general Dex service or transport call failed.
131    Service {
132        /// Structured service failure.
133        service: ServiceError,
134    },
135    /// Starting a Flow conflicted with an existing Flow ID.
136    FlowAlreadyStarted {
137        /// Structured start-operation failure.
138        service: ServiceError,
139    },
140    /// An operation requiring an existing Flow targeted an unknown ID.
141    FlowNotFound {
142        /// Structured lookup failure.
143        service: ServiceError,
144    },
145    /// An operation requiring an active Flow targeted a terminal or unknown ID.
146    FlowNotActive {
147        /// Structured active-Flow failure.
148        service: ServiceError,
149    },
150    /// A Worker Step or RPC handler failed.
151    WorkerInvocation {
152        /// Outer Dex service failure.
153        service: ServiceError,
154        /// Original Worker-side failure metadata.
155        worker: Box<WorkerError>,
156    },
157    /// A locking RPC could not acquire all requested Attribute locks.
158    RpcLockConflict {
159        /// Structured aborted-operation failure.
160        service: ServiceError,
161    },
162    /// A long-poll timeout elapsed without indicating a Flow failure.
163    LongPollTimeout {
164        /// Structured timeout failure.
165        service: ServiceError,
166    },
167    /// A waited Flow reached a terminal status other than Completed.
168    FlowUncompleted {
169        /// Run ID that reached the terminal status.
170        run_id: String,
171        /// Terminal Flow status.
172        status: FlowStatus,
173        /// Dex failure category, when available.
174        error_type: Option<FlowErrorType>,
175        /// Server completion detail, when available.
176        message: Option<String>,
177        /// Number of completed Step outputs returned with the failure.
178        result_count: usize,
179    },
180    /// A registered Flow, Step, RPC, or persistence definition is invalid.
181    FlowDefinition {
182        /// Developer-actionable contract violation.
183        message: String,
184    },
185    /// An SDK method received an invalid argument.
186    InvalidArgument {
187        /// Developer-actionable argument violation.
188        message: String,
189    },
190    /// An application value could not be serialized or decoded.
191    ValueMapping {
192        /// Developer-actionable mapping failure.
193        message: String,
194    },
195    /// A Step returned a value that violates its handler contract.
196    InvalidStepResult {
197        /// Flow type containing the Step.
198        flow_type: String,
199        /// Step type that returned the invalid value.
200        step_type: String,
201        /// Developer-actionable result violation.
202        detail: String,
203    },
204}
205
206impl Display for SdkError {
207    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
208        match self {
209            Self::Service { service }
210            | Self::FlowAlreadyStarted { service }
211            | Self::FlowNotFound { service }
212            | Self::FlowNotActive { service }
213            | Self::RpcLockConflict { service }
214            | Self::LongPollTimeout { service } => Display::fmt(service, formatter),
215            Self::WorkerInvocation { service, .. } => Display::fmt(service, formatter),
216            Self::InvalidStepResult { detail, .. } => formatter.write_str(detail),
217            Self::FlowUncompleted { message, .. } => {
218                formatter.write_str(message.as_deref().unwrap_or("Flow did not complete"))
219            }
220            Self::FlowDefinition { message }
221            | Self::InvalidArgument { message }
222            | Self::ValueMapping { message } => formatter.write_str(message),
223        }
224    }
225}
226
227impl Error for SdkError {
228    fn source(&self) -> Option<&(dyn Error + 'static)> {
229        self.service_error()
230            .map(|error| error as &(dyn Error + 'static))
231    }
232}
233
234impl SdkError {
235    /// Returns shared service metadata for service-backed variants.
236    ///
237    /// Local definition, argument, mapping, invalid-result, and uncompleted-Flow errors return
238    /// `None`.
239    pub fn service_error(&self) -> Option<&ServiceError> {
240        match self {
241            Self::Service { service }
242            | Self::FlowAlreadyStarted { service }
243            | Self::FlowNotFound { service }
244            | Self::FlowNotActive { service }
245            | Self::RpcLockConflict { service }
246            | Self::LongPollTimeout { service } => Some(service),
247            Self::WorkerInvocation { service, .. } => Some(service),
248            _ => None,
249        }
250    }
251
252    pub(crate) fn from_status(
253        status: Status,
254        operation: &'static str,
255        flow_id: Option<&str>,
256        requirement: FlowTargetRequirement,
257    ) -> Self {
258        let code = status.code();
259        let decoded = decode_details(status.details());
260        let details = decoded.as_ref().ok().and_then(|details| details.as_ref());
261        let detail = details
262            .filter(|details| !details.detail.is_empty())
263            .map_or_else(
264                || {
265                    if status.message().is_empty() {
266                        format!("{code:?}")
267                    } else {
268                        status.message().to_string()
269                    }
270                },
271                |details| details.detail.clone(),
272            );
273        let sub_status = details
274            .map(|details| map_sub_status(details.sub_status))
275            .unwrap_or(ErrorSubStatus::Uncategorized);
276        let service = ServiceError {
277            code,
278            sub_status,
279            detail,
280            operation,
281            flow_id: flow_id.map(str::to_string),
282            source: status,
283        };
284        if decoded.is_err() {
285            return Self::Service { service };
286        }
287        match sub_status {
288            ErrorSubStatus::FlowAlreadyStarted => Self::FlowAlreadyStarted { service },
289            ErrorSubStatus::FlowNotExists => match requirement {
290                FlowTargetRequirement::Active => Self::FlowNotActive { service },
291                FlowTargetRequirement::Existing => Self::FlowNotFound { service },
292                FlowTargetRequirement::None => Self::Service { service },
293            },
294            ErrorSubStatus::WorkerApiError if code == GrpcCode::Aborted => {
295                Self::RpcLockConflict { service }
296            }
297            ErrorSubStatus::WorkerApiError => {
298                let details = details.expect("worker details were decoded");
299                Self::WorkerInvocation {
300                    service,
301                    worker: Box::new(WorkerError {
302                        code: (details.original_worker_error_status != 0)
303                            .then(|| GrpcCode::from_i32(details.original_worker_error_status)),
304                        error_type: details.original_worker_error_type.clone(),
305                        detail: details.original_worker_error_detail.clone(),
306                    }),
307                }
308            }
309            ErrorSubStatus::LongPollTimeout => Self::LongPollTimeout { service },
310            ErrorSubStatus::Uncategorized => Self::Service { service },
311        }
312    }
313}
314
315#[derive(Clone, Copy)]
316pub(crate) enum FlowTargetRequirement {
317    None,
318    Existing,
319    Active,
320}
321
322#[derive(Clone, PartialEq, Message)]
323struct GoogleRpcStatus {
324    #[prost(int32, tag = "1")]
325    code: i32,
326    #[prost(string, tag = "2")]
327    message: String,
328    #[prost(message, repeated, tag = "3")]
329    details: Vec<Any>,
330}
331
332fn decode_details(bytes: &[u8]) -> Result<Option<ErrorResponse>, prost::DecodeError> {
333    if bytes.is_empty() {
334        return Ok(None);
335    }
336    match GoogleRpcStatus::decode(bytes) {
337        Ok(status) => {
338            for detail in status.details {
339                if detail.type_url.ends_with("/dex.ErrorResponse") {
340                    return ErrorResponse::decode(detail.value.as_slice()).map(Some);
341                }
342            }
343            Ok(None)
344        }
345        Err(status_error) => ErrorResponse::decode(bytes).map(Some).or(Err(status_error)),
346    }
347}
348
349fn map_sub_status(value: i32) -> ErrorSubStatus {
350    match ProtoErrorSubStatus::try_from(value).ok() {
351        Some(ProtoErrorSubStatus::FlowAlreadyStarted) => ErrorSubStatus::FlowAlreadyStarted,
352        Some(ProtoErrorSubStatus::FlowNotExists) => ErrorSubStatus::FlowNotExists,
353        Some(ProtoErrorSubStatus::WorkerApiError) => ErrorSubStatus::WorkerApiError,
354        Some(ProtoErrorSubStatus::LongPollTimeOut) => ErrorSubStatus::LongPollTimeout,
355        _ => ErrorSubStatus::Uncategorized,
356    }
357}
358
359#[cfg(test)]
360mod tests {
361    use super::*;
362
363    #[test]
364    fn missing_flow_uses_endpoint_lifecycle_requirement() {
365        let found = SdkError::from_status(
366            rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
367            "describe_flow",
368            Some("flow-id"),
369            FlowTargetRequirement::Existing,
370        );
371        assert!(matches!(found, SdkError::FlowNotFound { .. }));
372
373        let active = SdkError::from_status(
374            rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
375            "publish",
376            Some("flow-id"),
377            FlowTargetRequirement::Active,
378        );
379        assert!(matches!(active, SdkError::FlowNotActive { .. }));
380    }
381
382    #[test]
383    fn worker_details_and_lock_conflicts_are_distinct() {
384        let invocation = SdkError::from_status(
385            worker_status(GrpcCode::FailedPrecondition),
386            "invoke_rpc",
387            Some("flow-id"),
388            FlowTargetRequirement::Active,
389        );
390        match invocation {
391            SdkError::WorkerInvocation { service, worker } => {
392                assert_eq!(GrpcCode::FailedPrecondition, service.code());
393                assert_eq!(Some(GrpcCode::InvalidArgument), worker.code());
394                assert_eq!("ApplicationError", worker.error_type());
395                assert_eq!("invalid order", worker.detail());
396            }
397            error => panic!("expected WorkerInvocation, got {error:?}"),
398        }
399
400        let conflict = SdkError::from_status(
401            worker_status(GrpcCode::Aborted),
402            "invoke_rpc",
403            Some("flow-id"),
404            FlowTargetRequirement::Active,
405        );
406        assert!(matches!(conflict, SdkError::RpcLockConflict { .. }));
407    }
408
409    #[test]
410    fn missing_and_malformed_details_use_generic_fallback() {
411        let missing = SdkError::from_status(
412            Status::internal("missing"),
413            "search_flows",
414            None,
415            FlowTargetRequirement::None,
416        );
417        assert!(matches!(missing, SdkError::Service { .. }));
418
419        let malformed = SdkError::from_status(
420            Status::with_details(GrpcCode::Internal, "malformed", vec![255].into()),
421            "search_flows",
422            None,
423            FlowTargetRequirement::None,
424        );
425        assert!(matches!(malformed, SdkError::Service { .. }));
426    }
427
428    fn rich_status(code: GrpcCode, sub_status: ProtoErrorSubStatus) -> Status {
429        status_with_response(
430            code,
431            ErrorResponse {
432                detail: "service detail".to_string(),
433                sub_status: sub_status as i32,
434                ..ErrorResponse::default()
435            },
436        )
437    }
438
439    fn worker_status(code: GrpcCode) -> Status {
440        status_with_response(
441            code,
442            ErrorResponse {
443                detail: "worker failure".to_string(),
444                sub_status: ProtoErrorSubStatus::WorkerApiError as i32,
445                original_worker_error_detail: "invalid order".to_string(),
446                original_worker_error_type: "ApplicationError".to_string(),
447                original_worker_error_status: GrpcCode::InvalidArgument as i32,
448                original_worker_error_stack_trace: String::new(),
449            },
450        )
451    }
452
453    fn status_with_response(code: GrpcCode, response: ErrorResponse) -> Status {
454        let status = GoogleRpcStatus {
455            code: code as i32,
456            message: "gRPC detail".to_string(),
457            details: vec![Any {
458                type_url: "type.googleapis.com/dex.ErrorResponse".to_string(),
459                value: response.encode_to_vec(),
460            }],
461        };
462        Status::with_details(code, "gRPC detail", status.encode_to_vec().into())
463    }
464}
dex_sdk/sdk_error.rs

dex_sdk/
sdk_error.rs
