1use 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
19pub type SdkResult<T> = Result<T, SdkError>;
21
22#[derive(Clone, Copy, Debug, Eq, PartialEq)]
23#[non_exhaustive]
25pub enum ErrorSubStatus {
26 Uncategorized,
28 FlowAlreadyStarted,
30 FlowNotExists,
32 WorkerApiError,
34 LongPollTimeout,
36 ChannelMessageNotFound,
38 RequestTimeout,
40 FlowClosing,
42}
43
44#[derive(Clone, Debug)]
45pub 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 call_context: Option<String>,
55}
56
57impl ServiceError {
58 pub fn code(&self) -> GrpcCode {
60 self.code
61 }
62
63 pub fn sub_status(&self) -> ErrorSubStatus {
65 self.sub_status
66 }
67
68 pub fn detail(&self) -> &str {
70 &self.detail
71 }
72
73 pub fn operation(&self) -> &str {
75 self.operation
76 }
77
78 pub fn flow_id(&self) -> Option<&str> {
80 self.flow_id.as_deref()
81 }
82
83 pub(crate) fn local(operation: &'static str, detail: impl Into<String>) -> Self {
84 let detail = detail.into();
85 Self {
86 code: GrpcCode::Unknown,
87 sub_status: ErrorSubStatus::Uncategorized,
88 detail: detail.clone(),
89 operation,
90 flow_id: None,
91 source: Status::unknown(detail),
92 pull_registration_error: None,
93 call_context: None,
94 }
95 }
96}
97
98impl Display for ServiceError {
99 fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
100 match &self.call_context {
101 Some(call_context) => write!(formatter, "{call_context}: {}", self.detail),
102 None => formatter.write_str(&self.detail),
103 }
104 }
105}
106
107impl Error for ServiceError {
108 fn source(&self) -> Option<&(dyn Error + 'static)> {
109 match &self.pull_registration_error {
110 Some(rejection) => Some(rejection.as_ref()),
111 None => Some(&self.source),
112 }
113 }
114}
115
116#[derive(Clone, Debug, Eq, PartialEq)]
117pub struct WorkerError {
119 code: Option<GrpcCode>,
120 error_type: String,
121 detail: String,
122}
123
124impl WorkerError {
125 pub fn code(&self) -> Option<GrpcCode> {
127 self.code
128 }
129
130 pub fn error_type(&self) -> &str {
132 &self.error_type
133 }
134
135 pub fn detail(&self) -> &str {
137 &self.detail
138 }
139}
140
141#[derive(Clone, Debug)]
142#[non_exhaustive]
147pub enum SdkError {
148 Service {
150 service: ServiceError,
152 },
153 FlowAlreadyStarted {
155 service: ServiceError,
157 },
158 FlowClosing {
170 service: ServiceError,
172 },
173 FlowNotFound {
175 service: ServiceError,
177 },
178 FlowNotActiveOrNotFound {
201 service: ServiceError,
203 },
204 WorkerInvocation {
206 service: ServiceError,
208 worker: Box<WorkerError>,
210 },
211 RpcLockConflict {
213 service: ServiceError,
215 },
216 LongPollTimeout {
218 service: ServiceError,
220 },
221 ChannelMessageNotFound {
223 service: ServiceError,
225 },
226 RequestTimeout {
228 service: ServiceError,
230 },
231 FlowDefinition {
233 message: String,
235 },
236 InvalidArgument {
238 message: String,
240 },
241 ValueMapping {
243 message: String,
245 },
246 InvalidStepResult {
248 flow_type: String,
250 step_type: String,
252 detail: String,
254 },
255}
256
257impl Display for SdkError {
258 fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
259 match self {
260 Self::Service { service }
261 | Self::FlowAlreadyStarted { service }
262 | Self::FlowClosing { service }
263 | Self::FlowNotFound { service }
264 | Self::FlowNotActiveOrNotFound { service }
265 | Self::RpcLockConflict { service }
266 | Self::LongPollTimeout { service }
267 | Self::ChannelMessageNotFound { service } => Display::fmt(service, formatter),
268 Self::RequestTimeout { service } => Display::fmt(service, formatter),
269 Self::WorkerInvocation { service, .. } => Display::fmt(service, formatter),
270 Self::InvalidStepResult { detail, .. } => formatter.write_str(detail),
271 Self::FlowDefinition { message }
272 | Self::InvalidArgument { message }
273 | Self::ValueMapping { message } => formatter.write_str(message),
274 }
275 }
276}
277
278impl Error for SdkError {
279 fn source(&self) -> Option<&(dyn Error + 'static)> {
280 self.service_error()
281 .map(|error| error as &(dyn Error + 'static))
282 }
283}
284
285impl SdkError {
286 pub fn pull_registration_error(&self) -> Option<&crate::PullRegistrationError> {
292 self.service_error()
293 .and_then(|service| service.pull_registration_error.as_deref())
294 }
295
296 pub(crate) fn from_pull_registration(rejection: crate::PullRegistrationError) -> Self {
297 Self::Service {
298 service: ServiceError {
299 code: rejection.status.code(),
300 sub_status: ErrorSubStatus::Uncategorized,
301 detail: rejection.to_string(),
302 operation: "pull_worker_registration",
303 flow_id: None,
304 source: rejection.status.clone(),
305 pull_registration_error: Some(Box::new(rejection)),
306 call_context: None,
307 },
308 }
309 }
310
311 pub fn service_error(&self) -> Option<&ServiceError> {
315 match self {
316 Self::Service { service }
317 | Self::FlowAlreadyStarted { service }
318 | Self::FlowClosing { service }
319 | Self::FlowNotFound { service }
320 | Self::FlowNotActiveOrNotFound { service }
321 | Self::RpcLockConflict { service }
322 | Self::LongPollTimeout { service }
323 | Self::ChannelMessageNotFound { service } => Some(service),
324 Self::RequestTimeout { service } => Some(service),
325 Self::WorkerInvocation { service, .. } => Some(service),
326 _ => None,
327 }
328 }
329
330 pub(crate) fn from_status(
331 status: Status,
332 operation: &'static str,
333 flow_id: Option<&str>,
334 requirement: FlowTargetRequirement,
335 ) -> Self {
336 let code = status.code();
337 let decoded = decode_details(status.details());
338 let details = decoded.as_ref().ok().and_then(|details| details.as_ref());
339 let detail = details
340 .filter(|details| !details.detail.is_empty())
341 .map_or_else(
342 || {
343 if status.message().is_empty() {
344 format!("{code:?}")
345 } else {
346 status.message().to_string()
347 }
348 },
349 |details| details.detail.clone(),
350 );
351 let sub_status = details
352 .map(|details| map_sub_status(details.sub_status))
353 .unwrap_or(ErrorSubStatus::Uncategorized);
354 let service = ServiceError {
355 code,
356 sub_status,
357 detail,
358 operation,
359 flow_id: flow_id.map(str::to_string),
360 source: status,
361 pull_registration_error: None,
362 call_context: None,
363 };
364 if decoded.is_err() {
365 return Self::Service { service };
366 }
367 match sub_status {
368 ErrorSubStatus::FlowAlreadyStarted => Self::FlowAlreadyStarted { service },
369 ErrorSubStatus::FlowClosing => Self::FlowClosing { service },
370 ErrorSubStatus::FlowNotExists => match requirement {
371 FlowTargetRequirement::Active => Self::FlowNotActiveOrNotFound { service },
372 FlowTargetRequirement::Existing => Self::FlowNotFound { service },
373 FlowTargetRequirement::None => Self::Service { service },
374 },
375 ErrorSubStatus::WorkerApiError if code == GrpcCode::Aborted => {
376 Self::RpcLockConflict { service }
377 }
378 ErrorSubStatus::WorkerApiError => {
379 let details = details.expect("worker details were decoded");
380 Self::WorkerInvocation {
381 service,
382 worker: Box::new(WorkerError {
383 code: (details.original_worker_error_status != 0)
384 .then(|| GrpcCode::from_i32(details.original_worker_error_status)),
385 error_type: details.original_worker_error_type.clone(),
386 detail: details.original_worker_error_detail.clone(),
387 }),
388 }
389 }
390 ErrorSubStatus::LongPollTimeout => Self::LongPollTimeout { service },
391 ErrorSubStatus::ChannelMessageNotFound => Self::ChannelMessageNotFound { service },
392 ErrorSubStatus::RequestTimeout => Self::RequestTimeout { service },
393 ErrorSubStatus::Uncategorized => Self::Service { service },
394 }
395 }
396
397 pub(crate) fn request_timeout(operation: &'static str, flow_id: &str) -> Self {
398 let detail = "request timed out".to_string();
399 Self::RequestTimeout {
400 service: ServiceError {
401 code: GrpcCode::DeadlineExceeded,
402 sub_status: ErrorSubStatus::RequestTimeout,
403 detail: detail.clone(),
404 operation,
405 flow_id: Some(flow_id.to_string()),
406 source: Status::deadline_exceeded(detail),
407 pull_registration_error: None,
408 call_context: None,
409 },
410 }
411 }
412
413 pub(crate) fn with_flow_target_requirement(self, requirement: FlowTargetRequirement) -> Self {
416 match self {
417 Self::Service { service } if service.sub_status == ErrorSubStatus::FlowNotExists => {
418 match requirement {
419 FlowTargetRequirement::Active => Self::FlowNotActiveOrNotFound { service },
420 FlowTargetRequirement::Existing => Self::FlowNotFound { service },
421 FlowTargetRequirement::None => Self::Service { service },
422 }
423 }
424 other => other,
425 }
426 }
427
428 pub(crate) fn with_call_context(
431 self,
432 operation: &'static str,
433 flow_id: &str,
434 call_context: String,
435 ) -> Self {
436 let with_service_context = |mut service: ServiceError| {
437 service.operation = operation;
438 service.flow_id = Some(flow_id.to_string());
439 service.call_context = Some(call_context.clone());
440 service
441 };
442 let with_message_context = |message: String| format!("{call_context}: {message}");
443 match self {
444 Self::Service { service } => Self::Service {
445 service: with_service_context(service),
446 },
447 Self::FlowAlreadyStarted { service } => Self::FlowAlreadyStarted {
448 service: with_service_context(service),
449 },
450 Self::FlowClosing { service } => Self::FlowClosing {
451 service: with_service_context(service),
452 },
453 Self::FlowNotFound { service } => Self::FlowNotFound {
454 service: with_service_context(service),
455 },
456 Self::FlowNotActiveOrNotFound { service } => Self::FlowNotActiveOrNotFound {
457 service: with_service_context(service),
458 },
459 Self::WorkerInvocation { service, worker } => Self::WorkerInvocation {
460 service: with_service_context(service),
461 worker,
462 },
463 Self::RpcLockConflict { service } => Self::RpcLockConflict {
464 service: with_service_context(service),
465 },
466 Self::LongPollTimeout { service } => Self::LongPollTimeout {
467 service: with_service_context(service),
468 },
469 Self::ChannelMessageNotFound { service } => Self::ChannelMessageNotFound {
470 service: with_service_context(service),
471 },
472 Self::RequestTimeout { service } => Self::RequestTimeout {
473 service: with_service_context(service),
474 },
475 Self::FlowDefinition { message } => Self::FlowDefinition {
476 message: with_message_context(message),
477 },
478 Self::InvalidArgument { message } => Self::InvalidArgument {
479 message: with_message_context(message),
480 },
481 Self::ValueMapping { message } => Self::ValueMapping {
482 message: with_message_context(message),
483 },
484 Self::InvalidStepResult {
485 flow_type,
486 step_type,
487 detail,
488 } => Self::InvalidStepResult {
489 flow_type,
490 step_type,
491 detail: with_message_context(detail),
492 },
493 }
494 }
495}
496
497#[derive(Clone, Copy)]
498pub(crate) enum FlowTargetRequirement {
499 None,
500 Existing,
501 Active,
502}
503
504#[derive(Clone, PartialEq, Message)]
505struct GoogleRpcStatus {
506 #[prost(int32, tag = "1")]
507 code: i32,
508 #[prost(string, tag = "2")]
509 message: String,
510 #[prost(message, repeated, tag = "3")]
511 details: Vec<Any>,
512}
513
514fn decode_details(bytes: &[u8]) -> Result<Option<ServiceErrorResponse>, prost::DecodeError> {
515 if bytes.is_empty() {
516 return Ok(None);
517 }
518 match GoogleRpcStatus::decode(bytes) {
519 Ok(status) => {
520 for detail in status.details {
521 if detail.type_url.ends_with("/dex.ServiceErrorResponse") {
522 return ServiceErrorResponse::decode(detail.value.as_slice()).map(Some);
523 }
524 }
525 Ok(None)
526 }
527 Err(status_error) => ServiceErrorResponse::decode(bytes)
528 .map(Some)
529 .or(Err(status_error)),
530 }
531}
532
533fn map_sub_status(value: i32) -> ErrorSubStatus {
534 match ProtoErrorSubStatus::try_from(value).ok() {
535 Some(ProtoErrorSubStatus::FlowAlreadyStarted) => ErrorSubStatus::FlowAlreadyStarted,
536 Some(ProtoErrorSubStatus::FlowNotExists) => ErrorSubStatus::FlowNotExists,
537 Some(ProtoErrorSubStatus::WorkerApiError) => ErrorSubStatus::WorkerApiError,
538 Some(ProtoErrorSubStatus::LongPollTimeOut) => ErrorSubStatus::LongPollTimeout,
539 Some(ProtoErrorSubStatus::ChannelMessageNotFound) => ErrorSubStatus::ChannelMessageNotFound,
540 Some(ProtoErrorSubStatus::RequestTimeout) => ErrorSubStatus::RequestTimeout,
541 Some(ProtoErrorSubStatus::FlowClosing) => ErrorSubStatus::FlowClosing,
542 _ => ErrorSubStatus::Uncategorized,
543 }
544}
545
546#[cfg(test)]
547mod tests {
548 use super::*;
549
550 #[test]
551 fn missing_flow_uses_endpoint_lifecycle_requirement() {
552 let found = SdkError::from_status(
553 rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
554 "describe_flow",
555 Some("flow-id"),
556 FlowTargetRequirement::Existing,
557 );
558 assert!(matches!(found, SdkError::FlowNotFound { .. }));
559
560 let active = SdkError::from_status(
561 rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
562 "publish",
563 Some("flow-id"),
564 FlowTargetRequirement::Active,
565 );
566 assert!(matches!(active, SdkError::FlowNotActiveOrNotFound { .. }));
567 }
568
569 #[test]
570 fn call_context_names_the_call_without_changing_the_variant_or_detail() {
571 let missing = SdkError::from_status(
572 rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
573 "load_blobs",
574 None,
575 FlowTargetRequirement::None,
576 )
577 .with_call_context(
578 "get_attributes",
579 "flow-id",
580 "get_attributes Flow \"flow-id\"".to_string(),
581 );
582 assert!(matches!(missing, SdkError::Service { .. }));
583 let service = missing.service_error().expect("service metadata");
584 assert_eq!("service detail", service.detail());
585 assert_eq!("get_attributes", service.operation());
586 assert_eq!(Some("flow-id"), service.flow_id());
587 assert_eq!(
588 "get_attributes Flow \"flow-id\": service detail",
589 missing.to_string()
590 );
591
592 let mapping = SdkError::ValueMapping {
593 message: "blob cache failed".to_string(),
594 }
595 .with_call_context("get_attributes", "flow-id", "context".to_string());
596 assert!(
597 matches!(mapping, SdkError::ValueMapping { message } if message == "context: blob cache failed")
598 );
599 }
600
601 #[test]
602 fn missing_flow_during_load_blobs_maps_to_the_caller_target() {
603 let load_blobs_error = || {
604 SdkError::from_status(
605 rich_status(GrpcCode::NotFound, ProtoErrorSubStatus::FlowNotExists),
606 "load_blobs",
607 None,
608 FlowTargetRequirement::None,
609 )
610 };
611 assert!(matches!(load_blobs_error(), SdkError::Service { .. }));
612 assert!(matches!(
613 load_blobs_error().with_flow_target_requirement(FlowTargetRequirement::Existing),
614 SdkError::FlowNotFound { .. }
615 ));
616 let unrelated = SdkError::from_status(
617 Status::internal("broken"),
618 "load_blobs",
619 None,
620 FlowTargetRequirement::None,
621 )
622 .with_flow_target_requirement(FlowTargetRequirement::Existing);
623 assert!(matches!(unrelated, SdkError::Service { .. }));
624 }
625
626 #[test]
627 fn flow_closing_has_a_typed_variant() {
628 let error = SdkError::from_status(
629 rich_status(
630 GrpcCode::FailedPrecondition,
631 ProtoErrorSubStatus::FlowClosing,
632 ),
633 "wait_for_attribute_match",
634 Some("flow-id"),
635 FlowTargetRequirement::Active,
636 );
637 assert!(matches!(error, SdkError::FlowClosing { .. }));
638 let service = error
639 .service_error()
640 .expect("FlowClosing carries service metadata");
641 assert_eq!(service.sub_status(), ErrorSubStatus::FlowClosing);
642 assert_eq!(service.code(), GrpcCode::FailedPrecondition);
643 }
644
645 #[test]
646 fn worker_details_and_lock_conflicts_are_distinct() {
647 let invocation = SdkError::from_status(
648 worker_status(GrpcCode::FailedPrecondition),
649 "invoke_rpc",
650 Some("flow-id"),
651 FlowTargetRequirement::Active,
652 );
653 match invocation {
654 SdkError::WorkerInvocation { service, worker } => {
655 assert_eq!(GrpcCode::FailedPrecondition, service.code());
656 assert_eq!(Some(GrpcCode::InvalidArgument), worker.code());
657 assert_eq!("ApplicationError", worker.error_type());
658 assert_eq!("invalid order", worker.detail());
659 }
660 error => panic!("expected WorkerInvocation, got {error:?}"),
661 }
662
663 let conflict = SdkError::from_status(
664 worker_status(GrpcCode::Aborted),
665 "invoke_rpc",
666 Some("flow-id"),
667 FlowTargetRequirement::Active,
668 );
669 assert!(matches!(conflict, SdkError::RpcLockConflict { .. }));
670 }
671
672 #[test]
673 fn missing_and_malformed_details_use_generic_fallback() {
674 let missing = SdkError::from_status(
675 Status::internal("missing"),
676 "search_flows",
677 None,
678 FlowTargetRequirement::None,
679 );
680 assert!(matches!(missing, SdkError::Service { .. }));
681
682 let malformed = SdkError::from_status(
683 Status::with_details(GrpcCode::Internal, "malformed", vec![255].into()),
684 "search_flows",
685 None,
686 FlowTargetRequirement::None,
687 );
688 assert!(matches!(malformed, SdkError::Service { .. }));
689 }
690
691 fn rich_status(code: GrpcCode, sub_status: ProtoErrorSubStatus) -> Status {
692 status_with_response(
693 code,
694 ServiceErrorResponse {
695 detail: "service detail".to_string(),
696 sub_status: sub_status as i32,
697 ..ServiceErrorResponse::default()
698 },
699 )
700 }
701
702 fn worker_status(code: GrpcCode) -> Status {
703 status_with_response(
704 code,
705 ServiceErrorResponse {
706 detail: "worker failure".to_string(),
707 sub_status: ProtoErrorSubStatus::WorkerApiError as i32,
708 original_worker_error_detail: "invalid order".to_string(),
709 original_worker_error_type: "ApplicationError".to_string(),
710 original_worker_error_status: GrpcCode::InvalidArgument as i32,
711 original_worker_error_stack_trace: String::new(),
712 ..ServiceErrorResponse::default()
713 },
714 )
715 }
716
717 fn status_with_response(code: GrpcCode, response: ServiceErrorResponse) -> Status {
718 let status = GoogleRpcStatus {
719 code: code as i32,
720 message: "gRPC detail".to_string(),
721 details: vec![Any {
722 type_url: "type.googleapis.com/dex.ServiceErrorResponse".to_string(),
723 value: response.encode_to_vec(),
724 }],
725 };
726 Status::with_details(code, "gRPC detail", status.encode_to_vec().into())
727 }
728}