dex_sdk/
sdk_error.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Sustainable Use 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-Sustainable-Use-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.
24#[non_exhaustive]
25pub enum ErrorSubStatus {
26    /// Dex returned no specific status.
27    Uncategorized,
28    /// A start request targeted an already-existing Flow ID.
29    FlowAlreadyStarted,
30    /// The requested Flow does not exist or is no longer active.
31    FlowNotExists,
32    /// A Worker Step or RPC invocation failed.
33    WorkerApiError,
34    /// A long-poll deadline elapsed while the Flow remained active.
35    LongPollTimeout,
36    /// A pending Channel message ID no longer exists.
37    ChannelMessageNotFound,
38    /// A caller-defined request budget expired.
39    RequestTimeout,
40    /// Dex rejected an update-class request during Flow close cleanup.
41    FlowClosing,
42}
43
44#[derive(Clone, Debug)]
45/// Preserves one failed Dex service operation and its transport source.
46pub struct ServiceError {
47    code: GrpcCode,
48    sub_status: ErrorSubStatus,
49    detail: String,
50    operation: &'static str,
51    flow_id: Option<String>,
52    source: Status,
53    pull_registration_error: Option<Box<crate::PullRegistrationError>>,
54}
55
56impl ServiceError {
57    /// Returns the outer gRPC status code.
58    pub fn code(&self) -> GrpcCode {
59        self.code
60    }
61
62    /// Returns the decoded Dex-specific status.
63    pub fn sub_status(&self) -> ErrorSubStatus {
64        self.sub_status
65    }
66
67    /// Returns the most specific server or transport detail available.
68    pub fn detail(&self) -> &str {
69        &self.detail
70    }
71
72    /// Returns the SDK operation name that failed.
73    pub fn operation(&self) -> &str {
74        self.operation
75    }
76
77    /// Returns the targeted Flow ID when the operation had one.
78    pub fn flow_id(&self) -> Option<&str> {
79        self.flow_id.as_deref()
80    }
81
82    pub(crate) fn local(operation: &'static str, detail: impl Into<String>) -> Self {
83        let detail = detail.into();
84        Self {
85            code: GrpcCode::Unknown,
86            sub_status: ErrorSubStatus::Uncategorized,
87            detail: detail.clone(),
88            operation,
89            flow_id: None,
90            source: Status::unknown(detail),
91            pull_registration_error: None,
92        }
93    }
94}
95
96impl Display for ServiceError {
97    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
98        formatter.write_str(&self.detail)
99    }
100}
101
102impl Error for ServiceError {
103    fn source(&self) -> Option<&(dyn Error + 'static)> {
104        match &self.pull_registration_error {
105            Some(rejection) => Some(rejection.as_ref()),
106            None => Some(&self.source),
107        }
108    }
109}
110
111#[derive(Clone, Debug, Eq, PartialEq)]
112/// Preserves the original Worker-side error carried by a service failure.
113pub struct WorkerError {
114    code: Option<GrpcCode>,
115    error_type: String,
116    detail: String,
117}
118
119impl WorkerError {
120    /// Returns the Worker gRPC status when available.
121    pub fn code(&self) -> Option<GrpcCode> {
122        self.code
123    }
124
125    /// Returns the Worker exception or error type name.
126    pub fn error_type(&self) -> &str {
127        &self.error_type
128    }
129
130    /// Returns the Worker-provided failure detail.
131    pub fn detail(&self) -> &str {
132        &self.detail
133    }
134}
135
136#[derive(Clone, Debug)]
137/// Reports service, definition, value, handler, and Flow-completion failures.
138///
139/// Match specific variants when recovery differs. [`Self::service_error`] extracts common gRPC
140/// metadata from every service-backed variant.
141#[non_exhaustive]
142pub enum SdkError {
143    /// A general Dex service or transport call failed.
144    Service {
145        /// Structured service failure.
146        service: ServiceError,
147    },
148    /// Starting a Flow conflicted with an existing Flow ID.
149    FlowAlreadyStarted {
150        /// Structured start-operation failure.
151        service: ServiceError,
152    },
153    /// An update-class request arrived while Dex was closing the Flow.
154    ///
155    /// Re-read the retained terminal Flow state before deciding what to do next.
156    /// Match this variant instead of comparing service detail text:
157    ///
158    /// ```no_run
159    /// # use dex_sdk::{SdkError, SdkResult};
160    /// fn should_reread_terminal_state(result: &SdkResult<()>) -> bool {
161    ///     matches!(result, Err(SdkError::FlowClosing { .. }))
162    /// }
163    /// ```
164    FlowClosing {
165        /// Structured rejected-operation failure, including the original transport cause.
166        service: ServiceError,
167    },
168    /// An operation requiring an existing Flow targeted an unknown ID.
169    FlowNotFound {
170        /// Structured lookup failure.
171        service: ServiceError,
172    },
173    /// A Flow was not found or the operation could not use its closed execution.
174    ///
175    /// RPC calls also use this variant for missing query-only targets. A query-only RPC without
176    /// locks, transactionality, durable effects, or Server-forced Update routing can read a retained
177    /// terminal execution. At that confirmed read boundary, treat this variant as not found without
178    /// a lifecycle probe. Preserve other service and Worker failures.
179    ///
180    /// For mutations and active-only RPCs, this variant does not distinguish missing from closed
181    /// targets or prove that the requested action succeeded. Match it as follows:
182    ///
183    /// ```no_run
184    /// # use dex_sdk::{SdkError, SdkResult};
185    /// fn report_unavailable(result: SdkResult<()>) -> SdkResult<()> {
186    ///     match result {
187    ///         Err(SdkError::FlowNotActiveOrNotFound { service }) => {
188    ///             eprintln!("{} has no usable target", service.operation());
189    ///             Err(SdkError::FlowNotActiveOrNotFound { service })
190    ///         }
191    ///         other => other,
192    ///     }
193    /// }
194    /// ```
195    FlowNotActiveOrNotFound {
196        /// Structured target failure, including the original transport cause.
197        service: ServiceError,
198    },
199    /// A Worker Step or RPC handler failed.
200    WorkerInvocation {
201        /// Outer Dex service failure.
202        service: ServiceError,
203        /// Original Worker-side failure metadata.
204        worker: Box<WorkerError>,
205    },
206    /// A locking RPC could not acquire all requested Attribute locks.
207    RpcLockConflict {
208        /// Structured aborted-operation failure.
209        service: ServiceError,
210    },
211    /// A long-poll timeout elapsed without indicating a Flow failure.
212    LongPollTimeout {
213        /// Structured timeout failure.
214        service: ServiceError,
215    },
216    /// A requested Channel message is no longer pending.
217    ChannelMessageNotFound {
218        /// Structured deletion failure.
219        service: ServiceError,
220    },
221    /// A caller-defined request budget expired.
222    RequestTimeout {
223        /// Structured timeout failure.
224        service: ServiceError,
225    },
226    /// A registered Flow, Step, RPC, or persistence definition is invalid.
227    FlowDefinition {
228        /// Developer-actionable contract violation.
229        message: String,
230    },
231    /// An SDK method received an invalid argument.
232    InvalidArgument {
233        /// Developer-actionable argument violation.
234        message: String,
235    },
236    /// An application value could not be serialized or decoded.
237    ValueMapping {
238        /// Developer-actionable mapping failure.
239        message: String,
240    },
241    /// A Step returned a value that violates its handler contract.
242    InvalidStepResult {
243        /// Flow type containing the Step.
244        flow_type: String,
245        /// Step type that returned the invalid value.
246        step_type: String,
247        /// Developer-actionable result violation.
248        detail: String,
249    },
250}
251
252impl Display for SdkError {
253    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
254        match self {
255            Self::Service { service }
256            | Self::FlowAlreadyStarted { service }
257            | Self::FlowClosing { service }
258            | Self::FlowNotFound { service }
259            | Self::FlowNotActiveOrNotFound { service }
260            | Self::RpcLockConflict { service }
261            | Self::LongPollTimeout { service }
262            | Self::ChannelMessageNotFound { service } => Display::fmt(service, formatter),
263            Self::RequestTimeout { service } => Display::fmt(service, formatter),
264            Self::WorkerInvocation { service, .. } => Display::fmt(service, formatter),
265            Self::InvalidStepResult { detail, .. } => formatter.write_str(detail),
266            Self::FlowDefinition { message }
267            | Self::InvalidArgument { message }
268            | Self::ValueMapping { message } => formatter.write_str(message),
269        }
270    }
271}
272
273impl Error for SdkError {
274    fn source(&self) -> Option<&(dyn Error + 'static)> {
275        self.service_error()
276            .map(|error| error as &(dyn Error + 'static))
277    }
278}
279
280impl SdkError {
281    /// Returns the typed pull queue rejection when a permanent error leaves no registered stream.
282    ///
283    /// Registration waiters and opt-in Worker shutdown return the existing [`Self::Service`]
284    /// variant. This accessor preserves its queue name and original gRPC status. Other errors
285    /// return `None`. See [`crate::PullRegistrationError`] for a complete inspection example.
286    pub fn pull_registration_error(&self) -> Option<&crate::PullRegistrationError> {
287        self.service_error()
288            .and_then(|service| service.pull_registration_error.as_deref())
289    }
290
291    pub(crate) fn from_pull_registration(rejection: crate::PullRegistrationError) -> Self {
292        Self::Service {
293            service: ServiceError {
294                code: rejection.status.code(),
295                sub_status: ErrorSubStatus::Uncategorized,
296                detail: rejection.to_string(),
297                operation: "pull_worker_registration",
298                flow_id: None,
299                source: rejection.status.clone(),
300                pull_registration_error: Some(Box::new(rejection)),
301            },
302        }
303    }
304
305    /// Returns shared service metadata for service-backed variants.
306    ///
307    /// Local definition, argument, mapping, and invalid-result errors return `None`.
308    pub fn service_error(&self) -> Option<&ServiceError> {
309        match self {
310            Self::Service { service }
311            | Self::FlowAlreadyStarted { service }
312            | Self::FlowClosing { service }
313            | Self::FlowNotFound { service }
314            | Self::FlowNotActiveOrNotFound { service }
315            | Self::RpcLockConflict { service }
316            | Self::LongPollTimeout { service }
317            | Self::ChannelMessageNotFound { service } => Some(service),
318            Self::RequestTimeout { service } => Some(service),
319            Self::WorkerInvocation { service, .. } => Some(service),
320            _ => None,
321        }
322    }
323
324    pub(crate) fn from_status(
325        status: Status,
326        operation: &'static str,
327        flow_id: Option<&str>,
328        requirement: FlowTargetRequirement,
329    ) -> Self {
330        let code = status.code();
331        let decoded = decode_details(status.details());
332        let details = decoded.as_ref().ok().and_then(|details| details.as_ref());
333        let detail = details
334            .filter(|details| !details.detail.is_empty())
335            .map_or_else(
336                || {
337                    if status.message().is_empty() {
338                        format!("{code:?}")
339                    } else {
340                        status.message().to_string()
341                    }
342                },
343                |details| details.detail.clone(),
344            );
345        let sub_status = details
346            .map(|details| map_sub_status(details.sub_status))
347            .unwrap_or(ErrorSubStatus::Uncategorized);
348        let service = ServiceError {
349            code,
350            sub_status,
351            detail,
352            operation,
353            flow_id: flow_id.map(str::to_string),
354            source: status,
355            pull_registration_error: None,
356        };
357        if decoded.is_err() {
358            return Self::Service { service };
359        }
360        match sub_status {
361            ErrorSubStatus::FlowAlreadyStarted => Self::FlowAlreadyStarted { service },
362            ErrorSubStatus::FlowClosing => Self::FlowClosing { service },
363            ErrorSubStatus::FlowNotExists => match requirement {
364                FlowTargetRequirement::Active => Self::FlowNotActiveOrNotFound { service },
365                FlowTargetRequirement::Existing => Self::FlowNotFound { service },
366                FlowTargetRequirement::None => Self::Service { service },
367            },
368            ErrorSubStatus::WorkerApiError if code == GrpcCode::Aborted => {
369                Self::RpcLockConflict { service }
370            }
371            ErrorSubStatus::WorkerApiError => {
372                let details = details.expect("worker details were decoded");
373                Self::WorkerInvocation {
374                    service,
375                    worker: Box::new(WorkerError {
376                        code: (details.original_worker_error_status != 0)
377                            .then(|| GrpcCode::from_i32(details.original_worker_error_status)),
378                        error_type: details.original_worker_error_type.clone(),
379                        detail: details.original_worker_error_detail.clone(),
380                    }),
381                }
382            }
383            ErrorSubStatus::LongPollTimeout => Self::LongPollTimeout { service },
384            ErrorSubStatus::ChannelMessageNotFound => Self::ChannelMessageNotFound { service },
385            ErrorSubStatus::RequestTimeout => Self::RequestTimeout { service },
386            ErrorSubStatus::Uncategorized => Self::Service { service },
387        }
388    }
389
390    pub(crate) fn request_timeout(operation: &'static str, flow_id: &str) -> Self {
391        let detail = "request timed out".to_string();
392        Self::RequestTimeout {
393            service: ServiceError {
394                code: GrpcCode::DeadlineExceeded,
395                sub_status: ErrorSubStatus::RequestTimeout,
396                detail: detail.clone(),
397                operation,
398                flow_id: Some(flow_id.to_string()),
399                source: Status::deadline_exceeded(detail),
400                pull_registration_error: None,
401            },
402        }
403    }
404}
405
406#[derive(Clone, Copy)]
407pub(crate) enum FlowTargetRequirement {
408    None,
409    Existing,
410    Active,
411}
412
413#[derive(Clone, PartialEq, Message)]
414struct GoogleRpcStatus {
415    #[prost(int32, tag = "1")]
416    code: i32,
417    #[prost(string, tag = "2")]
418    message: String,
419    #[prost(message, repeated, tag = "3")]
420    details: Vec<Any>,
421}
422
423fn decode_details(bytes: &[u8]) -> Result<Option<ServiceErrorResponse>, prost::DecodeError> {
424    if bytes.is_empty() {
425        return Ok(None);
426    }
427    match GoogleRpcStatus::decode(bytes) {
428        Ok(status) => {
429            for detail in status.details {
430                if detail.type_url.ends_with("/dex.ServiceErrorResponse") {
431                    return ServiceErrorResponse::decode(detail.value.as_slice()).map(Some);
432                }
433            }
434            Ok(None)
435        }
436        Err(status_error) => ServiceErrorResponse::decode(bytes)
437            .map(Some)
438            .or(Err(status_error)),
439    }
440}
441
442fn map_sub_status(value: i32) -> ErrorSubStatus {
443    match ProtoErrorSubStatus::try_from(value).ok() {
444        Some(ProtoErrorSubStatus::FlowAlreadyStarted) => ErrorSubStatus::FlowAlreadyStarted,
445        Some(ProtoErrorSubStatus::FlowNotExists) => ErrorSubStatus::FlowNotExists,
446        Some(ProtoErrorSubStatus::WorkerApiError) => ErrorSubStatus::WorkerApiError,
447        Some(ProtoErrorSubStatus::LongPollTimeOut) => ErrorSubStatus::LongPollTimeout,
448        Some(ProtoErrorSubStatus::ChannelMessageNotFound) => ErrorSubStatus::ChannelMessageNotFound,
449        Some(ProtoErrorSubStatus::RequestTimeout) => ErrorSubStatus::RequestTimeout,
450        Some(ProtoErrorSubStatus::FlowClosing) => ErrorSubStatus::FlowClosing,
451        _ => ErrorSubStatus::Uncategorized,
452    }
453}
454
455#[cfg(test)]
456mod tests {
457    use super::*;
458
459    #[test]
460    fn missing_flow_uses_endpoint_lifecycle_requirement() {
461        let found = SdkError::from_status(
462            rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
463            "describe_flow",
464            Some("flow-id"),
465            FlowTargetRequirement::Existing,
466        );
467        assert!(matches!(found, SdkError::FlowNotFound { .. }));
468
469        let active = SdkError::from_status(
470            rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
471            "publish",
472            Some("flow-id"),
473            FlowTargetRequirement::Active,
474        );
475        assert!(matches!(active, SdkError::FlowNotActiveOrNotFound { .. }));
476    }
477
478    #[test]
479    fn flow_closing_has_a_typed_variant() {
480        let error = SdkError::from_status(
481            rich_status(
482                GrpcCode::FailedPrecondition,
483                ProtoErrorSubStatus::FlowClosing,
484            ),
485            "wait_for_attribute_match",
486            Some("flow-id"),
487            FlowTargetRequirement::Active,
488        );
489        assert!(matches!(error, SdkError::FlowClosing { .. }));
490        let service = error
491            .service_error()
492            .expect("FlowClosing carries service metadata");
493        assert_eq!(service.sub_status(), ErrorSubStatus::FlowClosing);
494        assert_eq!(service.code(), GrpcCode::FailedPrecondition);
495    }
496
497    #[test]
498    fn worker_details_and_lock_conflicts_are_distinct() {
499        let invocation = SdkError::from_status(
500            worker_status(GrpcCode::FailedPrecondition),
501            "invoke_rpc",
502            Some("flow-id"),
503            FlowTargetRequirement::Active,
504        );
505        match invocation {
506            SdkError::WorkerInvocation { service, worker } => {
507                assert_eq!(GrpcCode::FailedPrecondition, service.code());
508                assert_eq!(Some(GrpcCode::InvalidArgument), worker.code());
509                assert_eq!("ApplicationError", worker.error_type());
510                assert_eq!("invalid order", worker.detail());
511            }
512            error => panic!("expected WorkerInvocation, got {error:?}"),
513        }
514
515        let conflict = SdkError::from_status(
516            worker_status(GrpcCode::Aborted),
517            "invoke_rpc",
518            Some("flow-id"),
519            FlowTargetRequirement::Active,
520        );
521        assert!(matches!(conflict, SdkError::RpcLockConflict { .. }));
522    }
523
524    #[test]
525    fn missing_and_malformed_details_use_generic_fallback() {
526        let missing = SdkError::from_status(
527            Status::internal("missing"),
528            "search_flows",
529            None,
530            FlowTargetRequirement::None,
531        );
532        assert!(matches!(missing, SdkError::Service { .. }));
533
534        let malformed = SdkError::from_status(
535            Status::with_details(GrpcCode::Internal, "malformed", vec![255].into()),
536            "search_flows",
537            None,
538            FlowTargetRequirement::None,
539        );
540        assert!(matches!(malformed, SdkError::Service { .. }));
541    }
542
543    fn rich_status(code: GrpcCode, sub_status: ProtoErrorSubStatus) -> Status {
544        status_with_response(
545            code,
546            ServiceErrorResponse {
547                detail: "service detail".to_string(),
548                sub_status: sub_status as i32,
549                ..ServiceErrorResponse::default()
550            },
551        )
552    }
553
554    fn worker_status(code: GrpcCode) -> Status {
555        status_with_response(
556            code,
557            ServiceErrorResponse {
558                detail: "worker failure".to_string(),
559                sub_status: ProtoErrorSubStatus::WorkerApiError as i32,
560                original_worker_error_detail: "invalid order".to_string(),
561                original_worker_error_type: "ApplicationError".to_string(),
562                original_worker_error_status: GrpcCode::InvalidArgument as i32,
563                original_worker_error_stack_trace: String::new(),
564                ..ServiceErrorResponse::default()
565            },
566        )
567    }
568
569    fn status_with_response(code: GrpcCode, response: ServiceErrorResponse) -> Status {
570        let status = GoogleRpcStatus {
571            code: code as i32,
572            message: "gRPC detail".to_string(),
573            details: vec![Any {
574                type_url: "type.googleapis.com/dex.ServiceErrorResponse".to_string(),
575                value: response.encode_to_vec(),
576            }],
577        };
578        Status::with_details(code, "gRPC detail", status.encode_to_vec().into())
579    }
580}
dex_sdk/sdk_error.rs

dex_sdk/
sdk_error.rs
