1use crate::{
4 WorkflowExecution,
5 data_converters::{
6 DecodablePayloads, GenericPayloadConverter, PayloadConversionError, PayloadConverter,
7 RawValue, SerializationContext, SerializationContextData, TemporalDeserializable,
8 TemporalSerializable,
9 },
10 protos::temporal::api::{
11 common::v1::{Payload, Payloads},
12 enums::v1::{
13 ApplicationErrorCategory as ProtoApplicationErrorCategory,
14 RetryState as ProtoRetryState, TimeoutType as ProtoTimeoutType,
15 },
16 failure::v1::Failure,
17 },
18};
19use std::time::Duration;
20
21#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
23#[non_exhaustive]
24pub enum StartChildWorkflowExecutionFailedCause {
25 Unspecified,
27 WorkflowAlreadyExists,
29 Unknown,
31}
32
33#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
35#[non_exhaustive]
36pub enum RetryState {
37 Unspecified,
39 InProgress,
41 NonRetryableFailure,
43 Timeout,
45 MaximumAttemptsReached,
47 RetryPolicyNotSet,
49 InternalServerError,
51 CancelRequested,
53 Unknown,
55}
56
57impl RetryState {
58 fn from_raw(value: i32) -> Self {
59 match ProtoRetryState::try_from(value) {
60 Ok(ProtoRetryState::Unspecified) => Self::Unspecified,
61 Ok(ProtoRetryState::InProgress) => Self::InProgress,
62 Ok(ProtoRetryState::NonRetryableFailure) => Self::NonRetryableFailure,
63 Ok(ProtoRetryState::Timeout) => Self::Timeout,
64 Ok(ProtoRetryState::MaximumAttemptsReached) => Self::MaximumAttemptsReached,
65 Ok(ProtoRetryState::RetryPolicyNotSet) => Self::RetryPolicyNotSet,
66 Ok(ProtoRetryState::InternalServerError) => Self::InternalServerError,
67 Ok(ProtoRetryState::CancelRequested) => Self::CancelRequested,
68 Err(_) => Self::Unknown,
69 }
70 }
71}
72
73impl From<ProtoRetryState> for RetryState {
74 fn from(value: ProtoRetryState) -> Self {
75 Self::from_raw(value as i32)
76 }
77}
78
79#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
81#[non_exhaustive]
82pub enum TimeoutType {
83 Unspecified,
85 StartToClose,
87 ScheduleToStart,
89 ScheduleToClose,
91 Heartbeat,
93 Unknown,
95}
96
97impl TimeoutType {
98 fn from_raw(value: i32) -> Self {
99 match ProtoTimeoutType::try_from(value) {
100 Ok(ProtoTimeoutType::Unspecified) => Self::Unspecified,
101 Ok(ProtoTimeoutType::StartToClose) => Self::StartToClose,
102 Ok(ProtoTimeoutType::ScheduleToStart) => Self::ScheduleToStart,
103 Ok(ProtoTimeoutType::ScheduleToClose) => Self::ScheduleToClose,
104 Ok(ProtoTimeoutType::Heartbeat) => Self::Heartbeat,
105 Err(_) => Self::Unknown,
106 }
107 }
108}
109
110impl From<ProtoTimeoutType> for TimeoutType {
111 fn from(value: ProtoTimeoutType) -> Self {
112 Self::from_raw(value as i32)
113 }
114}
115
116trait SerializableFailurePayload: Send + Sync {
119 fn to_payloads(
120 &self,
121 payload_converter: &PayloadConverter,
122 context: &SerializationContextData,
123 ) -> Result<Vec<Payload>, PayloadConversionError>;
124}
125
126impl<T> SerializableFailurePayload for T
127where
128 T: TemporalSerializable + Send + Sync + 'static,
129{
130 fn to_payloads(
131 &self,
132 payload_converter: &PayloadConverter,
133 context: &SerializationContextData,
134 ) -> Result<Vec<Payload>, PayloadConversionError> {
135 payload_converter.to_payloads(&SerializationContext::new(context, payload_converter), self)
136 }
137}
138
139#[derive(derive_more::Debug)]
141pub struct FailurePayloads {
142 repr: FailurePayloadsRepr,
143}
144
145#[derive(derive_more::Debug)]
146enum FailurePayloadsRepr {
147 #[debug("Serializable(...)")]
148 Serializable(#[debug(skip)] Box<dyn SerializableFailurePayload>),
149 Decoded(DecodablePayloads),
150}
151
152impl FailurePayloads {
153 pub(crate) fn encode(
154 &self,
155 payload_converter: &PayloadConverter,
156 context: &SerializationContextData,
157 ) -> Result<Payloads, PayloadConversionError> {
158 let payloads = match &self.repr {
159 FailurePayloadsRepr::Serializable(value) => {
160 value.to_payloads(payload_converter, context)?
161 }
162 FailurePayloadsRepr::Decoded(value) => value.raw().to_vec(),
163 };
164 Ok(Payloads { payloads })
165 }
166
167 pub fn deserialize<T: TemporalDeserializable + 'static>(
169 &self,
170 ) -> Result<T, PayloadConversionError> {
171 match &self.repr {
172 FailurePayloadsRepr::Decoded(value) => value.deserialize(),
173 FailurePayloadsRepr::Serializable(_) => Err(PayloadConversionError::WrongEncoding),
174 }
175 }
176
177 pub fn raw(&self) -> Option<&[Payload]> {
179 match &self.repr {
180 FailurePayloadsRepr::Decoded(value) => Some(value.raw()),
181 FailurePayloadsRepr::Serializable(_) => None,
182 }
183 }
184
185 pub fn into_raw(self) -> Option<RawValue> {
187 match self.repr {
188 FailurePayloadsRepr::Decoded(value) => Some(value.into_raw()),
189 FailurePayloadsRepr::Serializable(_) => None,
190 }
191 }
192}
193
194impl From<DecodablePayloads> for FailurePayloads {
195 fn from(value: DecodablePayloads) -> Self {
196 Self {
197 repr: FailurePayloadsRepr::Decoded(value),
198 }
199 }
200}
201
202impl<T> From<T> for FailurePayloads
203where
204 T: TemporalSerializable + Send + Sync + 'static,
205{
206 fn from(value: T) -> Self {
207 Self {
208 repr: FailurePayloadsRepr::Serializable(Box::new(value)),
209 }
210 }
211}
212
213#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
215#[non_exhaustive]
216pub enum ApplicationErrorCategory {
217 #[default]
219 Unspecified,
220 Benign,
223}
224
225impl From<ApplicationErrorCategory> for ProtoApplicationErrorCategory {
226 fn from(value: ApplicationErrorCategory) -> Self {
227 match value {
228 ApplicationErrorCategory::Unspecified => ProtoApplicationErrorCategory::Unspecified,
229 ApplicationErrorCategory::Benign => ProtoApplicationErrorCategory::Benign,
230 }
231 }
232}
233
234impl From<ProtoApplicationErrorCategory> for ApplicationErrorCategory {
235 fn from(value: ProtoApplicationErrorCategory) -> Self {
236 match value {
237 ProtoApplicationErrorCategory::Unspecified => ApplicationErrorCategory::Unspecified,
238 ProtoApplicationErrorCategory::Benign => ApplicationErrorCategory::Benign,
239 }
240 }
241}
242
243#[derive(Debug, bon::Builder)]
245#[builder(start_fn = builder, state_mod(vis = "pub"))]
246pub struct ApplicationFailure {
247 #[builder(start_fn, into)]
248 source: Box<dyn std::error::Error + Send + Sync>,
249 type_name: Option<String>,
250 #[builder(default)]
251 non_retryable: bool,
252 next_retry_delay: Option<Duration>,
253 #[builder(default = ApplicationErrorCategory::Unspecified)]
254 category: ApplicationErrorCategory,
255 #[builder(into)]
256 details: Option<FailurePayloads>,
257 failure: Option<Failure>,
258 cause: Option<Box<IncomingError>>,
259}
260
261impl ApplicationFailure {
262 pub fn new(source: impl Into<Box<dyn std::error::Error + Send + Sync>>) -> Self {
264 Self {
265 source: source.into(),
266 type_name: None,
267 non_retryable: false,
268 next_retry_delay: None,
269 category: ApplicationErrorCategory::Unspecified,
270 details: None,
271 failure: None,
272 cause: None,
273 }
274 }
275
276 pub fn non_retryable(source: impl Into<Box<dyn std::error::Error + Send + Sync>>) -> Self {
278 Self {
279 non_retryable: true,
280 ..Self::new(source)
281 }
282 }
283
284 pub fn source_error(&self) -> &(dyn std::error::Error + Send + Sync + 'static) {
286 &*self.source as &(dyn std::error::Error + Send + Sync + 'static)
287 }
288
289 pub fn type_name(&self) -> Option<&str> {
291 self.type_name.as_deref()
292 }
293
294 pub fn is_non_retryable(&self) -> bool {
296 self.non_retryable
297 }
298
299 pub fn next_retry_delay(&self) -> Option<Duration> {
301 self.next_retry_delay
302 }
303
304 pub fn category(&self) -> ApplicationErrorCategory {
306 self.category
307 }
308
309 pub fn details<T: TemporalDeserializable + 'static>(
311 &self,
312 ) -> Result<Option<T>, PayloadConversionError> {
313 self.details
314 .as_ref()
315 .map(FailurePayloads::deserialize)
316 .transpose()
317 }
318
319 pub fn raw_details(&self) -> Option<&[Payload]> {
321 self.details.as_ref().and_then(FailurePayloads::raw)
322 }
323
324 pub(crate) fn failure_payloads(&self) -> Option<&FailurePayloads> {
325 self.details.as_ref()
326 }
327
328 pub fn failure(&self) -> Option<&Failure> {
330 self.failure.as_ref()
331 }
332
333 pub fn into_failure(self) -> Option<Failure> {
335 self.failure
336 }
337
338 pub fn cause(&self) -> Option<&IncomingError> {
340 self.cause.as_deref()
341 }
342
343 pub fn as_timeout(&self) -> Option<&TimeoutError> {
346 self.cause().and_then(IncomingError::as_timeout)
347 }
348
349 pub fn as_cancelled(&self) -> Option<&CancelledError> {
352 self.cause().and_then(IncomingError::as_cancelled)
353 }
354
355 pub(crate) fn from_failure(
356 failure: Failure,
357 cause: Option<IncomingError>,
358 payload_converter: &PayloadConverter,
359 context: &SerializationContextData,
360 ) -> Self {
361 let app_info = failure
362 .maybe_application_failure()
363 .cloned()
364 .unwrap_or_default();
365 let type_name = (!app_info.r#type.is_empty()).then_some(app_info.r#type.clone());
366 Self {
367 source: failure.message.clone().into(),
368 type_name,
369 non_retryable: app_info.non_retryable,
370 next_retry_delay: app_info.next_retry_delay.and_then(|d| d.try_into().ok()),
371 category: app_info.category().into(),
372 details: app_info.details.map(|details| {
373 FailurePayloads::from(DecodablePayloads::new(
374 details.payloads,
375 payload_converter.clone(),
376 context.clone(),
377 ))
378 }),
379 failure: Some(failure),
380 cause: cause.map(Box::new),
381 }
382 }
383}
384
385impl std::fmt::Display for ApplicationFailure {
386 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
387 write!(f, "{}", self.source)
388 }
389}
390
391impl std::error::Error for ApplicationFailure {
392 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
393 self.cause
394 .as_deref()
395 .map(|cause| cause as &(dyn std::error::Error + 'static))
396 .or_else(|| Some(self.source.as_ref()))
397 }
398}
399
400impl From<anyhow::Error> for ApplicationFailure {
401 fn from(value: anyhow::Error) -> Self {
402 Self::new(value)
403 }
404}
405
406impl From<PayloadConversionError> for ApplicationFailure {
407 fn from(value: PayloadConversionError) -> Self {
408 Self::new(value)
409 }
410}
411
412#[derive(Debug, thiserror::Error)]
414#[non_exhaustive]
415pub enum OutgoingError {
416 #[error(transparent)]
418 Activity(#[from] OutgoingActivityError),
419 #[error(transparent)]
421 Workflow(#[from] OutgoingWorkflowError),
422}
423
424#[derive(Debug, thiserror::Error)]
426#[non_exhaustive]
427pub enum OutgoingActivityError {
428 #[error(transparent)]
430 Application(#[from] Box<ApplicationFailure>),
431 #[error("Activity cancelled")]
433 Cancelled {
434 details: Option<FailurePayloads>,
436 },
437}
438
439#[derive(Debug, thiserror::Error)]
441#[non_exhaustive]
442pub enum OutgoingWorkflowError {
443 #[error(transparent)]
445 Application(#[from] Box<ApplicationFailure>),
446 #[error(transparent)]
448 PayloadConversion(#[from] PayloadConversionError),
449 #[error(transparent)]
451 ActivityExecution(#[from] Box<ActivityExecutionError>),
452 #[error(transparent)]
454 ChildWorkflowExecution(#[from] Box<ChildWorkflowExecutionError>),
455 #[error(transparent)]
457 ChildWorkflowStart(#[from] Box<ChildWorkflowStartError>),
458 #[error(transparent)]
460 WorkflowSignal(#[from] Box<WorkflowSignalError>),
461 #[error(transparent)]
463 CancelExternalWorkflow(#[from] Box<CancelExternalWorkflowError>),
464}
465
466impl OutgoingWorkflowError {
467 pub fn as_cancelled(&self) -> Option<&CancelledError> {
470 match self {
471 Self::Application(err) => err.as_cancelled(),
472 Self::PayloadConversion(_) => None,
473 Self::ActivityExecution(err) => err.as_cancelled(),
474 Self::ChildWorkflowExecution(err) => err.as_cancelled(),
475 Self::ChildWorkflowStart(err) => err.as_cancelled(),
476 Self::WorkflowSignal(err) => err.as_cancelled(),
477 Self::CancelExternalWorkflow(err) => err.as_cancelled(),
478 }
479 }
480}
481
482impl From<anyhow::Error> for OutgoingWorkflowError {
483 fn from(value: anyhow::Error) -> Self {
484 Self::Application(Box::new(ApplicationFailure::new(value)))
485 }
486}
487
488impl From<ApplicationFailure> for OutgoingWorkflowError {
489 fn from(value: ApplicationFailure) -> Self {
490 Self::Application(Box::new(value))
491 }
492}
493
494impl From<ActivityExecutionError> for OutgoingWorkflowError {
495 fn from(value: ActivityExecutionError) -> Self {
496 match value {
497 ActivityExecutionError::Serialization(err) => Self::PayloadConversion(err),
498 other => Self::ActivityExecution(Box::new(other)),
499 }
500 }
501}
502
503impl From<ChildWorkflowExecutionError> for OutgoingWorkflowError {
504 fn from(value: ChildWorkflowExecutionError) -> Self {
505 match value {
506 ChildWorkflowExecutionError::Serialization(err) => Self::PayloadConversion(err),
507 other => Self::ChildWorkflowExecution(Box::new(other)),
508 }
509 }
510}
511
512impl From<ChildWorkflowStartError> for OutgoingWorkflowError {
513 fn from(value: ChildWorkflowStartError) -> Self {
514 match value {
515 ChildWorkflowStartError::Serialization(err) => Self::PayloadConversion(err),
516 other => Self::ChildWorkflowStart(Box::new(other)),
517 }
518 }
519}
520
521impl From<WorkflowSignalError> for OutgoingWorkflowError {
522 fn from(value: WorkflowSignalError) -> Self {
523 match value {
524 WorkflowSignalError::Serialization(err) => Self::PayloadConversion(err),
525 other => Self::WorkflowSignal(Box::new(other)),
526 }
527 }
528}
529
530impl From<CancelExternalWorkflowError> for OutgoingWorkflowError {
531 fn from(value: CancelExternalWorkflowError) -> Self {
532 match value {
533 CancelExternalWorkflowError::Serialization(err) => Self::PayloadConversion(err),
534 other => Self::CancelExternalWorkflow(Box::new(other)),
535 }
536 }
537}
538
539#[derive(Debug)]
541#[non_exhaustive]
542pub enum IncomingError {
543 Application(ApplicationFailure),
545 Timeout(TimeoutError),
547 Cancelled(CancelledError),
549 Terminated(TerminatedError),
551 Server(ServerError),
553 ResetWorkflow(ResetWorkflowError),
555 Activity(ActivityFailureError),
557 ChildWorkflowExecution(ChildWorkflowFailureError),
559 NexusOperationExecution(IncomingNexusOperationExecutionError),
561 NexusHandler(IncomingNexusHandlerError),
563}
564
565impl IncomingError {
566 pub fn failure(&self) -> &Failure {
568 match self {
569 IncomingError::Application(err) => err
570 .failure()
571 .expect("decoded application failures retain their original proto"),
572 IncomingError::Timeout(err) => err.failure(),
573 IncomingError::Cancelled(err) => err.failure(),
574 IncomingError::Terminated(err) => err.failure(),
575 IncomingError::Server(err) => err.failure(),
576 IncomingError::ResetWorkflow(err) => err.failure(),
577 IncomingError::Activity(err) => err.failure(),
578 IncomingError::ChildWorkflowExecution(err) => err.failure(),
579 IncomingError::NexusOperationExecution(err) => err.failure(),
580 IncomingError::NexusHandler(err) => err.failure(),
581 }
582 }
583
584 pub fn cause(&self) -> Option<&IncomingError> {
586 match self {
587 IncomingError::Application(err) => err.cause(),
588 IncomingError::Timeout(err) => err.cause(),
589 IncomingError::Cancelled(err) => err.cause(),
590 IncomingError::Terminated(err) => err.cause(),
591 IncomingError::Server(err) => err.cause(),
592 IncomingError::ResetWorkflow(err) => err.cause(),
593 IncomingError::Activity(err) => err.cause(),
594 IncomingError::ChildWorkflowExecution(err) => err.cause(),
595 IncomingError::NexusOperationExecution(err) => err.cause(),
596 IncomingError::NexusHandler(err) => err.cause(),
597 }
598 }
599
600 pub fn into_failure(self) -> Failure {
602 match self {
603 IncomingError::Application(err) => err
604 .into_failure()
605 .expect("decoded application failures retain their original proto"),
606 IncomingError::Timeout(err) => err.into_failure(),
607 IncomingError::Cancelled(err) => err.into_failure(),
608 IncomingError::Terminated(err) => err.into_failure(),
609 IncomingError::Server(err) => err.into_failure(),
610 IncomingError::ResetWorkflow(err) => err.into_failure(),
611 IncomingError::Activity(err) => err.into_failure(),
612 IncomingError::ChildWorkflowExecution(err) => err.into_failure(),
613 IncomingError::NexusOperationExecution(err) => err.into_failure(),
614 IncomingError::NexusHandler(err) => err.into_failure(),
615 }
616 }
617
618 pub fn as_timeout(&self) -> Option<&TimeoutError> {
620 match self {
621 IncomingError::Timeout(err) => Some(err),
622 _ => None,
623 }
624 }
625
626 pub fn as_cancelled(&self) -> Option<&CancelledError> {
628 match self {
629 IncomingError::Cancelled(err) => Some(err),
630 _ => None,
631 }
632 }
633}
634
635impl std::fmt::Display for IncomingError {
636 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
637 match self {
638 IncomingError::Application(err) => err.fmt(f),
639 IncomingError::Timeout(err) => err.fmt(f),
640 IncomingError::Cancelled(err) => err.fmt(f),
641 IncomingError::Terminated(err) => err.fmt(f),
642 IncomingError::Server(err) => err.fmt(f),
643 IncomingError::ResetWorkflow(err) => err.fmt(f),
644 IncomingError::Activity(err) => err.fmt(f),
645 IncomingError::ChildWorkflowExecution(err) => err.fmt(f),
646 IncomingError::NexusOperationExecution(err) => err.fmt(f),
647 IncomingError::NexusHandler(err) => err.fmt(f),
648 }
649 }
650}
651
652impl std::error::Error for IncomingError {
653 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
654 match self {
655 IncomingError::Application(err) => Some(err),
656 IncomingError::Timeout(err) => Some(err),
657 IncomingError::Cancelled(err) => Some(err),
658 IncomingError::Terminated(err) => Some(err),
659 IncomingError::Server(err) => Some(err),
660 IncomingError::ResetWorkflow(err) => Some(err),
661 IncomingError::Activity(err) => Some(err),
662 IncomingError::ChildWorkflowExecution(err) => Some(err),
663 IncomingError::NexusOperationExecution(err) => Some(err),
664 IncomingError::NexusHandler(err) => Some(err),
665 }
666 }
667}
668
669macro_rules! impl_incoming_failure_wrapper {
670 ($name:ident) => {
671 impl $name {
672 pub fn failure(&self) -> &Failure {
674 &self.failure
675 }
676
677 pub fn cause(&self) -> Option<&IncomingError> {
679 self.cause.as_deref()
680 }
681
682 pub fn into_failure(self) -> Failure {
684 self.failure
685 }
686
687 pub fn into_parts(self) -> (Failure, Option<IncomingError>) {
689 (self.failure, self.cause.map(|cause| *cause))
690 }
691 }
692
693 impl std::fmt::Display for $name {
694 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
695 self.failure.fmt(f)
696 }
697 }
698
699 impl std::error::Error for $name {
700 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
701 self.cause
702 .as_deref()
703 .map(|cause| cause as &(dyn std::error::Error + 'static))
704 }
705 }
706 };
707}
708
709macro_rules! incoming_failure_wrapper {
710 ($name:ident, $doc:literal) => {
711 #[doc = $doc]
712 #[derive(Debug)]
713 pub struct $name {
714 failure: Failure,
715 cause: Option<Box<IncomingError>>,
716 }
717
718 impl $name {
719 pub(crate) fn new(failure: Failure, cause: Option<IncomingError>) -> Self {
721 Self {
722 failure,
723 cause: cause.map(Box::new),
724 }
725 }
726 }
727
728 impl_incoming_failure_wrapper!($name);
729 };
730}
731
732#[derive(Debug)]
734pub struct TimeoutError {
735 failure: Failure,
736 cause: Option<Box<IncomingError>>,
737 timeout_type: TimeoutType,
738 last_heartbeat_details: Option<DecodablePayloads>,
739}
740
741impl TimeoutError {
742 pub(crate) fn new(
744 failure: Failure,
745 failure_info: crate::protos::temporal::api::failure::v1::TimeoutFailureInfo,
746 cause: Option<IncomingError>,
747 payload_converter: &PayloadConverter,
748 context: &SerializationContextData,
749 ) -> Self {
750 Self {
751 failure,
752 cause: cause.map(Box::new),
753 timeout_type: TimeoutType::from_raw(failure_info.timeout_type),
754 last_heartbeat_details: failure_info.last_heartbeat_details.map(|details| {
755 DecodablePayloads::new(details.payloads, payload_converter.clone(), context.clone())
756 }),
757 }
758 }
759
760 pub fn timeout_type(&self) -> TimeoutType {
762 self.timeout_type
763 }
764
765 pub fn last_heartbeat_details<T: TemporalDeserializable + 'static>(
767 &self,
768 ) -> Result<Option<T>, PayloadConversionError> {
769 self.last_heartbeat_details
770 .as_ref()
771 .map(DecodablePayloads::deserialize)
772 .transpose()
773 }
774
775 pub fn raw_last_heartbeat_details(&self) -> Option<&[Payload]> {
777 self.last_heartbeat_details
778 .as_ref()
779 .map(DecodablePayloads::raw)
780 }
781}
782
783impl_incoming_failure_wrapper!(TimeoutError);
784
785#[derive(Debug)]
787pub struct CancelledError {
788 failure: Failure,
789 cause: Option<Box<IncomingError>>,
790 details: Option<DecodablePayloads>,
791}
792
793impl CancelledError {
794 pub(crate) fn new(
796 failure: Failure,
797 failure_info: crate::protos::temporal::api::failure::v1::CanceledFailureInfo,
798 cause: Option<IncomingError>,
799 payload_converter: &PayloadConverter,
800 context: &SerializationContextData,
801 ) -> Self {
802 Self {
803 failure,
804 cause: cause.map(Box::new),
805 details: failure_info.details.map(|details| {
806 DecodablePayloads::new(details.payloads, payload_converter.clone(), context.clone())
807 }),
808 }
809 }
810
811 pub fn details<T: TemporalDeserializable + 'static>(
814 &self,
815 ) -> Result<Option<T>, PayloadConversionError> {
816 self.details
817 .as_ref()
818 .map(DecodablePayloads::deserialize)
819 .transpose()
820 }
821
822 pub fn raw_details(&self) -> Option<&[Payload]> {
824 self.details.as_ref().map(DecodablePayloads::raw)
825 }
826}
827
828impl_incoming_failure_wrapper!(CancelledError);
829incoming_failure_wrapper!(TerminatedError, "A normalized terminated failure.");
830incoming_failure_wrapper!(ServerError, "A normalized server failure.");
831incoming_failure_wrapper!(ResetWorkflowError, "A normalized reset-workflow failure.");
832
833#[derive(Debug)]
835pub struct ActivityFailureError {
836 failure: Failure,
837 cause: Option<Box<IncomingError>>,
838 activity_id: String,
839 activity_type: Option<String>,
840 scheduled_event_id: i64,
841 started_event_id: i64,
842 identity: String,
843 retry_state: RetryState,
844}
845
846impl ActivityFailureError {
847 pub(crate) fn new(
849 failure: Failure,
850 failure_info: crate::protos::temporal::api::failure::v1::ActivityFailureInfo,
851 cause: Option<IncomingError>,
852 ) -> Self {
853 let retry_state = RetryState::from_raw(failure_info.retry_state);
854 Self {
855 failure,
856 cause: cause.map(Box::new),
857 activity_id: failure_info.activity_id,
858 activity_type: failure_info
859 .activity_type
860 .map(|activity_type| activity_type.name),
861 scheduled_event_id: failure_info.scheduled_event_id,
862 started_event_id: failure_info.started_event_id,
863 identity: failure_info.identity,
864 retry_state,
865 }
866 }
867
868 pub fn activity_id(&self) -> &str {
870 &self.activity_id
871 }
872
873 pub fn activity_type(&self) -> Option<&str> {
875 self.activity_type.as_deref()
876 }
877
878 pub fn scheduled_event_id(&self) -> i64 {
880 self.scheduled_event_id
881 }
882
883 pub fn started_event_id(&self) -> i64 {
885 self.started_event_id
886 }
887
888 pub fn identity(&self) -> &str {
890 &self.identity
891 }
892
893 pub fn retry_state(&self) -> RetryState {
895 self.retry_state
896 }
897
898 pub fn as_timeout(&self) -> Option<&TimeoutError> {
901 self.cause().and_then(IncomingError::as_timeout)
902 }
903
904 pub fn as_cancelled(&self) -> Option<&CancelledError> {
907 self.cause().and_then(IncomingError::as_cancelled)
908 }
909}
910
911impl_incoming_failure_wrapper!(ActivityFailureError);
912#[derive(Debug)]
914pub struct ChildWorkflowFailureError {
915 failure: Failure,
916 cause: Option<Box<IncomingError>>,
917 namespace: String,
918 workflow_execution: Option<WorkflowExecution>,
919 workflow_type: Option<String>,
920 initiated_event_id: i64,
921 started_event_id: i64,
922 retry_state: RetryState,
923}
924
925impl ChildWorkflowFailureError {
926 pub(crate) fn new(
928 failure: Failure,
929 failure_info: crate::protos::temporal::api::failure::v1::ChildWorkflowExecutionFailureInfo,
930 cause: Option<IncomingError>,
931 ) -> Self {
932 let retry_state = RetryState::from_raw(failure_info.retry_state);
933 Self {
934 failure,
935 cause: cause.map(Box::new),
936 namespace: failure_info.namespace,
937 workflow_execution: failure_info.workflow_execution.map(Into::into),
938 workflow_type: failure_info
939 .workflow_type
940 .map(|workflow_type| workflow_type.name),
941 initiated_event_id: failure_info.initiated_event_id,
942 started_event_id: failure_info.started_event_id,
943 retry_state,
944 }
945 }
946
947 pub fn namespace(&self) -> &str {
949 &self.namespace
950 }
951
952 pub fn workflow_execution(&self) -> Option<&WorkflowExecution> {
954 self.workflow_execution.as_ref()
955 }
956
957 pub fn workflow_type(&self) -> Option<&str> {
959 self.workflow_type.as_deref()
960 }
961
962 pub fn initiated_event_id(&self) -> i64 {
964 self.initiated_event_id
965 }
966
967 pub fn started_event_id(&self) -> i64 {
969 self.started_event_id
970 }
971
972 pub fn retry_state(&self) -> RetryState {
974 self.retry_state
975 }
976
977 pub fn as_timeout(&self) -> Option<&TimeoutError> {
980 self.cause().and_then(IncomingError::as_timeout)
981 }
982
983 pub fn as_cancelled(&self) -> Option<&CancelledError> {
986 self.cause().and_then(IncomingError::as_cancelled)
987 }
988}
989
990impl_incoming_failure_wrapper!(ChildWorkflowFailureError);
991incoming_failure_wrapper!(
992 IncomingNexusOperationExecutionError,
993 "A normalized nexus operation failure wrapper."
994);
995incoming_failure_wrapper!(
996 IncomingNexusHandlerError,
997 "A normalized nexus handler failure wrapper."
998);
999
1000#[derive(Debug, thiserror::Error)]
1002#[non_exhaustive]
1003pub enum ActivityExecutionError {
1004 #[error("Activity failed: {}", .0.failure().message)]
1006 Failed(#[source] ActivityFailureError),
1007 #[error("Activity cancelled: {}", .0.failure().message)]
1009 Cancelled(#[source] CancelledError),
1010 #[error("Payload conversion failed: {0}")]
1012 Serialization(#[from] PayloadConversionError),
1013}
1014
1015impl ActivityExecutionError {
1016 pub fn failure(&self) -> Option<&Failure> {
1018 match self {
1019 ActivityExecutionError::Failed(err) => Some(err.failure()),
1020 ActivityExecutionError::Cancelled(err) => Some(err.failure()),
1021 ActivityExecutionError::Serialization(_) => None,
1022 }
1023 }
1024
1025 pub fn cause(&self) -> Option<&IncomingError> {
1027 match self {
1028 ActivityExecutionError::Failed(err) => err.cause(),
1029 ActivityExecutionError::Cancelled(err) => err.cause(),
1030 ActivityExecutionError::Serialization(_) => None,
1031 }
1032 }
1033
1034 pub fn reason(&self) -> Option<&IncomingError> {
1036 match self {
1037 ActivityExecutionError::Failed(err) => err.cause(),
1038 ActivityExecutionError::Cancelled(_) | ActivityExecutionError::Serialization(_) => None,
1039 }
1040 }
1041
1042 pub fn as_timeout(&self) -> Option<&TimeoutError> {
1045 match self {
1046 ActivityExecutionError::Failed(err) => err.as_timeout(),
1047 ActivityExecutionError::Serialization(_) | ActivityExecutionError::Cancelled(_) => None,
1048 }
1049 }
1050
1051 pub fn as_cancelled(&self) -> Option<&CancelledError> {
1054 match self {
1055 ActivityExecutionError::Failed(err) => err.as_cancelled(),
1056 ActivityExecutionError::Cancelled(err) => Some(err),
1057 ActivityExecutionError::Serialization(_) => None,
1058 }
1059 }
1060}
1061
1062#[derive(Debug, thiserror::Error)]
1064#[non_exhaustive]
1065pub enum ChildWorkflowStartError {
1066 #[error("Child workflow start cancelled: {}", .0.failure().message)]
1068 Cancelled(#[source] Box<CancelledError>),
1069 #[error(
1071 "Child workflow start failed: workflow_id={workflow_id}, workflow_type={workflow_type}, cause={cause:?}"
1072 )]
1073 StartFailed {
1074 workflow_id: String,
1076 workflow_type: String,
1078 cause: StartChildWorkflowExecutionFailedCause,
1080 },
1081 #[error("Payload conversion failed: {0}")]
1083 Serialization(#[from] PayloadConversionError),
1084}
1085
1086impl ChildWorkflowStartError {
1087 pub fn failure(&self) -> Option<&Failure> {
1089 match self {
1090 ChildWorkflowStartError::Cancelled(err) => Some(err.failure()),
1091 ChildWorkflowStartError::StartFailed { .. }
1092 | ChildWorkflowStartError::Serialization(_) => None,
1093 }
1094 }
1095
1096 pub fn cause(&self) -> Option<&IncomingError> {
1098 match self {
1099 ChildWorkflowStartError::Cancelled(err) => err.cause(),
1100 ChildWorkflowStartError::StartFailed { .. }
1101 | ChildWorkflowStartError::Serialization(_) => None,
1102 }
1103 }
1104
1105 pub fn as_cancelled(&self) -> Option<&CancelledError> {
1108 match self {
1109 ChildWorkflowStartError::Cancelled(err) => Some(err),
1110 ChildWorkflowStartError::StartFailed { .. }
1111 | ChildWorkflowStartError::Serialization(_) => None,
1112 }
1113 }
1114}
1115
1116#[derive(Debug, thiserror::Error)]
1118#[non_exhaustive]
1119pub enum ChildWorkflowExecutionError {
1120 #[error("Child workflow failed: {}", .0.failure().message)]
1122 Failed(#[source] Box<ChildWorkflowFailureError>),
1123 #[error("Payload conversion failed: {0}")]
1125 Serialization(#[from] PayloadConversionError),
1126}
1127
1128impl ChildWorkflowExecutionError {
1129 pub fn failure(&self) -> Option<&Failure> {
1131 match self {
1132 ChildWorkflowExecutionError::Failed(err) => Some(err.failure()),
1133 ChildWorkflowExecutionError::Serialization(_) => None,
1134 }
1135 }
1136
1137 pub fn cause(&self) -> Option<&IncomingError> {
1139 match self {
1140 ChildWorkflowExecutionError::Failed(err) => err.cause(),
1141 ChildWorkflowExecutionError::Serialization(_) => None,
1142 }
1143 }
1144
1145 pub fn reason(&self) -> Option<&IncomingError> {
1147 match self {
1148 ChildWorkflowExecutionError::Failed(err) => err.cause(),
1149 ChildWorkflowExecutionError::Serialization(_) => None,
1150 }
1151 }
1152
1153 pub fn as_timeout(&self) -> Option<&TimeoutError> {
1156 match self {
1157 ChildWorkflowExecutionError::Failed(err) => err.as_timeout(),
1158 ChildWorkflowExecutionError::Serialization(_) => None,
1159 }
1160 }
1161
1162 pub fn as_cancelled(&self) -> Option<&CancelledError> {
1165 match self {
1166 ChildWorkflowExecutionError::Failed(err) => err.as_cancelled(),
1167 ChildWorkflowExecutionError::Serialization(_) => None,
1168 }
1169 }
1170}
1171
1172#[derive(Debug, thiserror::Error)]
1174#[non_exhaustive]
1175pub enum WorkflowSignalError {
1176 #[error("Workflow not found: {}", .0.failure().message)]
1178 NotFound(#[source] Box<WorkflowSignalFailureError>),
1179 #[error("Child workflow signal failed: {}", .0.failure().message)]
1181 Failed(#[source] Box<WorkflowSignalFailureError>),
1182 #[error("Signal payload conversion failed: {0}")]
1184 Serialization(#[from] PayloadConversionError),
1185}
1186
1187#[derive(Debug, thiserror::Error)]
1189pub enum CancelExternalWorkflowError {
1190 #[error("Workflow not found: {}", .0.failure().message)]
1192 NotFound(#[source] Box<WorkflowCancelFailureError>),
1193 #[error("External workflow cancellation request failed: {}", .0.failure().message)]
1195 Failed(#[source] Box<WorkflowCancelFailureError>),
1196 #[error("External workflow cancellation failure conversion failed: {0}")]
1198 Serialization(#[from] PayloadConversionError),
1199}
1200
1201impl CancelExternalWorkflowError {
1202 pub fn failure(&self) -> Option<&Failure> {
1204 match self {
1205 Self::NotFound(err) | Self::Failed(err) => Some(err.failure()),
1206 Self::Serialization(_) => None,
1207 }
1208 }
1209
1210 pub fn cause(&self) -> Option<&IncomingError> {
1212 match self {
1213 Self::NotFound(err) | Self::Failed(err) => err.cause(),
1214 Self::Serialization(_) => None,
1215 }
1216 }
1217
1218 pub fn reason(&self) -> Option<&IncomingError> {
1220 match self {
1221 Self::NotFound(err) | Self::Failed(err) => Some(err.error()),
1222 Self::Serialization(_) => None,
1223 }
1224 }
1225
1226 pub fn as_cancelled(&self) -> Option<&CancelledError> {
1228 self.reason()?.as_cancelled()
1229 }
1230}
1231
1232impl WorkflowSignalError {
1233 pub fn failure(&self) -> Option<&Failure> {
1235 match self {
1236 WorkflowSignalError::NotFound(err) | WorkflowSignalError::Failed(err) => {
1237 Some(err.failure())
1238 }
1239 WorkflowSignalError::Serialization(_) => None,
1240 }
1241 }
1242
1243 pub fn cause(&self) -> Option<&IncomingError> {
1245 match self {
1246 WorkflowSignalError::NotFound(err) | WorkflowSignalError::Failed(err) => err.cause(),
1247 WorkflowSignalError::Serialization(_) => None,
1248 }
1249 }
1250
1251 pub fn reason(&self) -> Option<&IncomingError> {
1253 match self {
1254 WorkflowSignalError::NotFound(err) | WorkflowSignalError::Failed(err) => {
1255 Some(err.error())
1256 }
1257 WorkflowSignalError::Serialization(_) => None,
1258 }
1259 }
1260
1261 pub fn as_cancelled(&self) -> Option<&CancelledError> {
1264 self.reason()?.as_cancelled()
1265 }
1266}
1267
1268#[derive(Debug)]
1270pub struct WorkflowSignalFailureError {
1271 failure: Failure,
1272 error: Box<IncomingError>,
1273}
1274
1275impl WorkflowSignalFailureError {
1276 pub(crate) fn new(failure: Failure, error: IncomingError) -> Self {
1278 Self {
1279 failure,
1280 error: Box::new(error),
1281 }
1282 }
1283
1284 pub fn failure(&self) -> &Failure {
1286 &self.failure
1287 }
1288
1289 pub fn cause(&self) -> Option<&IncomingError> {
1291 self.error.cause()
1292 }
1293
1294 pub fn error(&self) -> &IncomingError {
1296 &self.error
1297 }
1298}
1299
1300impl std::fmt::Display for WorkflowSignalFailureError {
1301 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1302 self.failure.fmt(f)
1303 }
1304}
1305
1306impl std::error::Error for WorkflowSignalFailureError {
1307 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1308 self.cause()
1309 .map(|cause| cause as &(dyn std::error::Error + 'static))
1310 }
1311}
1312
1313#[derive(Debug)]
1315pub struct WorkflowCancelFailureError {
1316 failure: Failure,
1317 error: Box<IncomingError>,
1318}
1319
1320impl WorkflowCancelFailureError {
1321 pub(crate) fn new(failure: Failure, error: IncomingError) -> Self {
1323 Self {
1324 failure,
1325 error: Box::new(error),
1326 }
1327 }
1328
1329 pub fn failure(&self) -> &Failure {
1331 &self.failure
1332 }
1333
1334 pub fn cause(&self) -> Option<&IncomingError> {
1336 self.error.cause()
1337 }
1338
1339 pub fn error(&self) -> &IncomingError {
1341 &self.error
1342 }
1343}
1344
1345impl std::fmt::Display for WorkflowCancelFailureError {
1346 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1347 self.failure.fmt(f)
1348 }
1349}
1350
1351impl std::error::Error for WorkflowCancelFailureError {
1352 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1353 self.cause()
1354 .map(|cause| cause as &(dyn std::error::Error + 'static))
1355 }
1356}
1357
1358#[cfg(test)]
1359mod tests {
1360 use super::*;
1361 use crate::{
1362 data_converters::{
1363 DefaultFailureConverter, FailureConverter, GenericPayloadConverter, PayloadConverter,
1364 SerializationContext, SerializationContextData, WorkflowSerializationContext,
1365 },
1366 protos::temporal::api::{
1367 common::v1::Payload,
1368 failure::v1::{ActivityFailureInfo, TimeoutFailureInfo, failure::FailureInfo},
1369 },
1370 };
1371
1372 struct AlwaysFailsSerialize;
1373
1374 impl serde::Serialize for AlwaysFailsSerialize {
1375 fn serialize<S: serde::Serializer>(&self, _serializer: S) -> Result<S::Ok, S::Error> {
1376 Err(serde::ser::Error::custom("serialize boom"))
1377 }
1378 }
1379
1380 #[test]
1381 fn decoded_failures_hide_unknown_state_values() {
1382 let failure = Failure {
1383 cause: Some(Box::new(Failure {
1384 failure_info: Some(FailureInfo::TimeoutFailureInfo(TimeoutFailureInfo {
1385 timeout_type: 654_321,
1386 ..Default::default()
1387 })),
1388 ..Default::default()
1389 })),
1390 failure_info: Some(FailureInfo::ActivityFailureInfo(ActivityFailureInfo {
1391 retry_state: 123_456,
1392 ..Default::default()
1393 })),
1394 ..Default::default()
1395 };
1396
1397 let decoded = DefaultFailureConverter::default()
1398 .to_error(
1399 failure,
1400 &PayloadConverter::default(),
1401 &SerializationContextData::Workflow(WorkflowSerializationContext::new()),
1402 )
1403 .unwrap();
1404 let IncomingError::Activity(activity) = decoded else {
1405 panic!("expected activity failure");
1406 };
1407 assert_eq!(activity.retry_state(), RetryState::Unknown);
1408 let Some(FailureInfo::ActivityFailureInfo(raw_activity)) =
1409 activity.failure().failure_info.as_ref()
1410 else {
1411 panic!("expected raw activity failure info");
1412 };
1413 assert_eq!(raw_activity.retry_state, 123_456);
1414 let Some(IncomingError::Timeout(timeout)) = activity.cause() else {
1415 panic!("expected timeout cause");
1416 };
1417 assert_eq!(timeout.timeout_type(), TimeoutType::Unknown);
1418 let Some(FailureInfo::TimeoutFailureInfo(raw_timeout)) =
1419 timeout.failure().failure_info.as_ref()
1420 else {
1421 panic!("expected raw timeout failure info");
1422 };
1423 assert_eq!(raw_timeout.timeout_type, 654_321);
1424 }
1425
1426 #[test]
1427 fn constructors_set_retryability_defaults() {
1428 assert!(!ApplicationFailure::new(anyhow::anyhow!("retryable")).is_non_retryable());
1429 assert!(
1430 ApplicationFailure::non_retryable(anyhow::anyhow!("non-retryable")).is_non_retryable()
1431 );
1432 }
1433
1434 #[test]
1435 fn conversion_preserves_application_metadata() {
1436 let payloads = Payloads {
1437 payloads: vec![Payload {
1438 data: b"details".to_vec(),
1439 ..Default::default()
1440 }],
1441 };
1442 let failure = DefaultFailureConverter::default().to_failure(
1443 OutgoingError::Workflow(OutgoingWorkflowError::Application(Box::new(
1444 ApplicationFailure::builder(anyhow::anyhow!("oops"))
1445 .type_name("MyType".to_owned())
1446 .non_retryable(true)
1447 .next_retry_delay(Duration::from_secs(3))
1448 .category(ApplicationErrorCategory::Benign)
1449 .details(RawValue::new(payloads.payloads.clone()))
1450 .build(),
1451 ))),
1452 &PayloadConverter::default(),
1453 &SerializationContextData::None,
1454 );
1455 let Some(FailureInfo::ApplicationFailureInfo(info)) = failure.failure_info else {
1456 panic!("expected application failure info");
1457 };
1458 assert_eq!(failure.message, "oops");
1459 assert_eq!(info.r#type, "MyType");
1460 assert!(info.non_retryable);
1461 assert_eq!(info.details, Some(payloads));
1462 assert_eq!(info.category(), ProtoApplicationErrorCategory::Benign);
1463 assert_eq!(info.next_retry_delay.unwrap().seconds, 3);
1464 }
1465
1466 #[test]
1467 fn builder_accepts_raw_payload_details() {
1468 let payload = Payload {
1469 data: b"details".to_vec(),
1470 ..Default::default()
1471 };
1472 let failure = DefaultFailureConverter::default().to_failure(
1473 OutgoingError::Workflow(OutgoingWorkflowError::Application(Box::new(
1474 ApplicationFailure::builder(anyhow::anyhow!("oops"))
1475 .details(RawValue::new(vec![payload.clone()]))
1476 .build(),
1477 ))),
1478 &PayloadConverter::default(),
1479 &SerializationContextData::None,
1480 );
1481
1482 let Some(FailureInfo::ApplicationFailureInfo(info)) = failure.failure_info else {
1483 panic!("expected application failure info");
1484 };
1485 assert_eq!(info.details.unwrap().payloads, vec![payload]);
1486 }
1487
1488 #[test]
1489 fn builder_accepts_serializable_details() {
1490 let failure = DefaultFailureConverter::default().to_failure(
1491 OutgoingError::Workflow(OutgoingWorkflowError::Application(Box::new(
1492 ApplicationFailure::builder(anyhow::anyhow!("oops"))
1493 .details("details".to_string())
1494 .build(),
1495 ))),
1496 &PayloadConverter::default(),
1497 &SerializationContextData::None,
1498 );
1499
1500 let Some(FailureInfo::ApplicationFailureInfo(info)) = failure.failure_info else {
1501 panic!("expected application failure info");
1502 };
1503 let payloads = info.details.expect("expected details").payloads;
1504 let converter = PayloadConverter::default();
1505 let details: String = converter
1506 .from_payloads(
1507 &SerializationContext::new(&SerializationContextData::None, &converter),
1508 payloads,
1509 )
1510 .unwrap();
1511 assert_eq!(details, "details");
1512 }
1513
1514 #[test]
1515 fn application_failure_encoding_surfaces_detail_encoding_errors() {
1516 let failure = DefaultFailureConverter::default().to_failure(
1517 OutgoingError::Workflow(OutgoingWorkflowError::Application(Box::new(
1518 ApplicationFailure::builder(anyhow::anyhow!("oops"))
1519 .details(AlwaysFailsSerialize)
1520 .build(),
1521 ))),
1522 &PayloadConverter::default(),
1523 &SerializationContextData::None,
1524 );
1525
1526 assert_eq!(
1527 failure.message,
1528 "Failed converting error to failure: Encoding error: serialize boom, original error message: oops"
1529 );
1530 assert!(matches!(
1531 failure.failure_info,
1532 Some(FailureInfo::ApplicationFailureInfo(_))
1533 ));
1534 }
1535
1536 #[test]
1537 fn anyhow_workflow_errors_default_to_application_outgoing_errors() {
1538 let outgoing: OutgoingWorkflowError = anyhow::anyhow!("workflow boom").into();
1539
1540 let OutgoingWorkflowError::Application(app) = outgoing else {
1541 panic!("plain workflow errors should default to application failures");
1542 };
1543 assert_eq!(app.to_string(), "workflow boom");
1544 }
1545
1546 #[test]
1547 fn payload_conversion_errors_use_dedicated_outgoing_variant() {
1548 let outgoing: OutgoingWorkflowError =
1549 PayloadConversionError::EncodingError(anyhow::anyhow!("encode boom").into()).into();
1550
1551 let OutgoingWorkflowError::PayloadConversion(err) = outgoing else {
1552 panic!("expected a payload conversion failure");
1553 };
1554 assert_eq!(err.to_string(), "Encoding error: encode boom");
1555 }
1556}