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::{ErrorSubStatus as ProtoErrorSubStatus, ServiceErrorResponse};
13use prost::Message;
14use prost_types::Any;
15use tonic::Status;
16
17use crate::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 registered Flow, Step, RPC, or persistence definition is invalid.
168    FlowDefinition {
169        /// Developer-actionable contract violation.
170        message: String,
171    },
172    /// An SDK method received an invalid argument.
173    InvalidArgument {
174        /// Developer-actionable argument violation.
175        message: String,
176    },
177    /// An application value could not be serialized or decoded.
178    ValueMapping {
179        /// Developer-actionable mapping failure.
180        message: String,
181    },
182    /// A Step returned a value that violates its handler contract.
183    InvalidStepResult {
184        /// Flow type containing the Step.
185        flow_type: String,
186        /// Step type that returned the invalid value.
187        step_type: String,
188        /// Developer-actionable result violation.
189        detail: String,
190    },
191}
192
193impl Display for SdkError {
194    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
195        match self {
196            Self::Service { service }
197            | Self::FlowAlreadyStarted { service }
198            | Self::FlowNotFound { service }
199            | Self::FlowNotActive { service }
200            | Self::RpcLockConflict { service }
201            | Self::LongPollTimeout { service } => Display::fmt(service, formatter),
202            Self::WorkerInvocation { service, .. } => Display::fmt(service, formatter),
203            Self::InvalidStepResult { detail, .. } => formatter.write_str(detail),
204            Self::FlowDefinition { message }
205            | Self::InvalidArgument { message }
206            | Self::ValueMapping { message } => formatter.write_str(message),
207        }
208    }
209}
210
211impl Error for SdkError {
212    fn source(&self) -> Option<&(dyn Error + 'static)> {
213        self.service_error()
214            .map(|error| error as &(dyn Error + 'static))
215    }
216}
217
218impl SdkError {
219    /// Returns shared service metadata for service-backed variants.
220    ///
221    /// Local definition, argument, mapping, and invalid-result errors return `None`.
222    pub fn service_error(&self) -> Option<&ServiceError> {
223        match self {
224            Self::Service { service }
225            | Self::FlowAlreadyStarted { service }
226            | Self::FlowNotFound { service }
227            | Self::FlowNotActive { service }
228            | Self::RpcLockConflict { service }
229            | Self::LongPollTimeout { service } => Some(service),
230            Self::WorkerInvocation { service, .. } => Some(service),
231            _ => None,
232        }
233    }
234
235    pub(crate) fn from_status(
236        status: Status,
237        operation: &'static str,
238        flow_id: Option<&str>,
239        requirement: FlowTargetRequirement,
240    ) -> Self {
241        let code = status.code();
242        let decoded = decode_details(status.details());
243        let details = decoded.as_ref().ok().and_then(|details| details.as_ref());
244        let detail = details
245            .filter(|details| !details.detail.is_empty())
246            .map_or_else(
247                || {
248                    if status.message().is_empty() {
249                        format!("{code:?}")
250                    } else {
251                        status.message().to_string()
252                    }
253                },
254                |details| details.detail.clone(),
255            );
256        let sub_status = details
257            .map(|details| map_sub_status(details.sub_status))
258            .unwrap_or(ErrorSubStatus::Uncategorized);
259        let service = ServiceError {
260            code,
261            sub_status,
262            detail,
263            operation,
264            flow_id: flow_id.map(str::to_string),
265            source: status,
266        };
267        if decoded.is_err() {
268            return Self::Service { service };
269        }
270        match sub_status {
271            ErrorSubStatus::FlowAlreadyStarted => Self::FlowAlreadyStarted { service },
272            ErrorSubStatus::FlowNotExists => match requirement {
273                FlowTargetRequirement::Active => Self::FlowNotActive { service },
274                FlowTargetRequirement::Existing => Self::FlowNotFound { service },
275                FlowTargetRequirement::None => Self::Service { service },
276            },
277            ErrorSubStatus::WorkerApiError if code == GrpcCode::Aborted => {
278                Self::RpcLockConflict { service }
279            }
280            ErrorSubStatus::WorkerApiError => {
281                let details = details.expect("worker details were decoded");
282                Self::WorkerInvocation {
283                    service,
284                    worker: Box::new(WorkerError {
285                        code: (details.original_worker_error_status != 0)
286                            .then(|| GrpcCode::from_i32(details.original_worker_error_status)),
287                        error_type: details.original_worker_error_type.clone(),
288                        detail: details.original_worker_error_detail.clone(),
289                    }),
290                }
291            }
292            ErrorSubStatus::LongPollTimeout => Self::LongPollTimeout { service },
293            ErrorSubStatus::Uncategorized => Self::Service { service },
294        }
295    }
296}
297
298#[derive(Clone, Copy)]
299pub(crate) enum FlowTargetRequirement {
300    None,
301    Existing,
302    Active,
303}
304
305#[derive(Clone, PartialEq, Message)]
306struct GoogleRpcStatus {
307    #[prost(int32, tag = "1")]
308    code: i32,
309    #[prost(string, tag = "2")]
310    message: String,
311    #[prost(message, repeated, tag = "3")]
312    details: Vec<Any>,
313}
314
315fn decode_details(bytes: &[u8]) -> Result<Option<ServiceErrorResponse>, prost::DecodeError> {
316    if bytes.is_empty() {
317        return Ok(None);
318    }
319    match GoogleRpcStatus::decode(bytes) {
320        Ok(status) => {
321            for detail in status.details {
322                if detail.type_url.ends_with("/dex.ServiceErrorResponse") {
323                    return ServiceErrorResponse::decode(detail.value.as_slice()).map(Some);
324                }
325            }
326            Ok(None)
327        }
328        Err(status_error) => ServiceErrorResponse::decode(bytes)
329            .map(Some)
330            .or(Err(status_error)),
331    }
332}
333
334fn map_sub_status(value: i32) -> ErrorSubStatus {
335    match ProtoErrorSubStatus::try_from(value).ok() {
336        Some(ProtoErrorSubStatus::FlowAlreadyStarted) => ErrorSubStatus::FlowAlreadyStarted,
337        Some(ProtoErrorSubStatus::FlowNotExists) => ErrorSubStatus::FlowNotExists,
338        Some(ProtoErrorSubStatus::WorkerApiError) => ErrorSubStatus::WorkerApiError,
339        Some(ProtoErrorSubStatus::LongPollTimeOut) => ErrorSubStatus::LongPollTimeout,
340        _ => ErrorSubStatus::Uncategorized,
341    }
342}
343
344#[cfg(test)]
345mod tests {
346    use super::*;
347
348    #[test]
349    fn missing_flow_uses_endpoint_lifecycle_requirement() {
350        let found = SdkError::from_status(
351            rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
352            "describe_flow",
353            Some("flow-id"),
354            FlowTargetRequirement::Existing,
355        );
356        assert!(matches!(found, SdkError::FlowNotFound { .. }));
357
358        let active = SdkError::from_status(
359            rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
360            "publish",
361            Some("flow-id"),
362            FlowTargetRequirement::Active,
363        );
364        assert!(matches!(active, SdkError::FlowNotActive { .. }));
365    }
366
367    #[test]
368    fn worker_details_and_lock_conflicts_are_distinct() {
369        let invocation = SdkError::from_status(
370            worker_status(GrpcCode::FailedPrecondition),
371            "invoke_rpc",
372            Some("flow-id"),
373            FlowTargetRequirement::Active,
374        );
375        match invocation {
376            SdkError::WorkerInvocation { service, worker } => {
377                assert_eq!(GrpcCode::FailedPrecondition, service.code());
378                assert_eq!(Some(GrpcCode::InvalidArgument), worker.code());
379                assert_eq!("ApplicationError", worker.error_type());
380                assert_eq!("invalid order", worker.detail());
381            }
382            error => panic!("expected WorkerInvocation, got {error:?}"),
383        }
384
385        let conflict = SdkError::from_status(
386            worker_status(GrpcCode::Aborted),
387            "invoke_rpc",
388            Some("flow-id"),
389            FlowTargetRequirement::Active,
390        );
391        assert!(matches!(conflict, SdkError::RpcLockConflict { .. }));
392    }
393
394    #[test]
395    fn missing_and_malformed_details_use_generic_fallback() {
396        let missing = SdkError::from_status(
397            Status::internal("missing"),
398            "search_flows",
399            None,
400            FlowTargetRequirement::None,
401        );
402        assert!(matches!(missing, SdkError::Service { .. }));
403
404        let malformed = SdkError::from_status(
405            Status::with_details(GrpcCode::Internal, "malformed", vec![255].into()),
406            "search_flows",
407            None,
408            FlowTargetRequirement::None,
409        );
410        assert!(matches!(malformed, SdkError::Service { .. }));
411    }
412
413    fn rich_status(code: GrpcCode, sub_status: ProtoErrorSubStatus) -> Status {
414        status_with_response(
415            code,
416            ServiceErrorResponse {
417                detail: "service detail".to_string(),
418                sub_status: sub_status as i32,
419                ..ServiceErrorResponse::default()
420            },
421        )
422    }
423
424    fn worker_status(code: GrpcCode) -> Status {
425        status_with_response(
426            code,
427            ServiceErrorResponse {
428                detail: "worker failure".to_string(),
429                sub_status: ProtoErrorSubStatus::WorkerApiError as i32,
430                original_worker_error_detail: "invalid order".to_string(),
431                original_worker_error_type: "ApplicationError".to_string(),
432                original_worker_error_status: GrpcCode::InvalidArgument as i32,
433                original_worker_error_stack_trace: String::new(),
434            },
435        )
436    }
437
438    fn status_with_response(code: GrpcCode, response: ServiceErrorResponse) -> Status {
439        let status = GoogleRpcStatus {
440            code: code as i32,
441            message: "gRPC detail".to_string(),
442            details: vec![Any {
443                type_url: "type.googleapis.com/dex.ServiceErrorResponse".to_string(),
444                value: response.encode_to_vec(),
445            }],
446        };
447        Status::with_details(code, "gRPC detail", status.encode_to_vec().into())
448    }
449}
dex_sdk/sdk_error.rs

dex_sdk/
sdk_error.rs
