temporalio_workflow/runtime/
model.rs

1//! Runtime protocol and execution model types shared by workflow code and native hosts.
2
3use crate::{
4    WorkflowCancellationError,
5    runtime::types::ContinueAsNewRequest,
6    workflow_context::{
7        ChildWfCommon, NexusUnblockData, PendingChildWorkflow, StartedNexusOperation,
8    },
9};
10use temporalio_common_wasm::{
11    WorkflowDefinition,
12    data_converters::PayloadConversionError,
13    error::{
14        ActivityExecutionError, ApplicationFailure, ChildWorkflowExecutionError,
15        ChildWorkflowStartError, WorkflowSignalError,
16    },
17    protos::{
18        coresdk::{
19            activity_result::ActivityResolution,
20            child_workflow::ChildWorkflowResult,
21            nexus::NexusOperationResult,
22            workflow_activation::{
23                resolve_child_workflow_execution_start::Status as ChildWorkflowStartStatus,
24                resolve_nexus_operation_start,
25            },
26        },
27        temporal::api::failure::v1::Failure,
28    },
29};
30
31#[derive(Debug)]
32pub enum UnblockEvent {
33    Timer(u32, TimerResult),
34    Activity(u32, Box<ActivityResolution>),
35    WorkflowStart(u32, Box<ChildWorkflowStartStatus>),
36    WorkflowComplete(u32, Box<ChildWorkflowResult>),
37    SignalExternal(u32, Option<Failure>),
38    CancelExternal(u32, Option<Failure>),
39    NexusOperationStart(u32, Box<resolve_nexus_operation_start::Status>),
40    NexusOperationComplete(u32, Box<NexusOperationResult>),
41}
42
43/// Result of awaiting on a timer
44#[derive(Debug, Copy, Clone, PartialEq, Eq)]
45pub enum TimerResult {
46    /// The timer was cancelled
47    Cancelled,
48    /// The timer elapsed and fired
49    Fired,
50}
51
52/// Successful result of sending a signal to an external workflow
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub struct SignalExternalOk;
55/// Result of awaiting on sending a signal to an external workflow
56pub type SignalExternalWfResult = Result<SignalExternalOk, Failure>;
57
58/// Successful result of sending a cancel request to an external workflow
59#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60pub struct CancelExternalOk;
61/// Result of awaiting on sending a cancel request to an external workflow
62pub type CancelExternalWfResult = Result<CancelExternalOk, Failure>;
63
64pub(crate) trait Unblockable {
65    type OtherDat;
66
67    fn unblock(ue: UnblockEvent, od: Self::OtherDat) -> Self;
68}
69
70impl Unblockable for TimerResult {
71    type OtherDat = ();
72
73    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
74        match ue {
75            UnblockEvent::Timer(_, result) => result,
76            _ => panic!("Invalid unblock event for timer"),
77        }
78    }
79}
80
81impl Unblockable for ActivityResolution {
82    type OtherDat = ();
83
84    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
85        match ue {
86            UnblockEvent::Activity(_, result) => *result,
87            _ => panic!("Invalid unblock event for activity"),
88        }
89    }
90}
91
92impl<WD: WorkflowDefinition> Unblockable for PendingChildWorkflow<WD> {
93    type OtherDat = ChildWfCommon;
94
95    fn unblock(ue: UnblockEvent, od: Self::OtherDat) -> Self {
96        match ue {
97            UnblockEvent::WorkflowStart(_, result) => Self {
98                status: *result,
99                common: od,
100                _phantom: std::marker::PhantomData,
101            },
102            _ => panic!("Invalid unblock event for child workflow start"),
103        }
104    }
105}
106
107impl Unblockable for ChildWorkflowResult {
108    type OtherDat = ();
109
110    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
111        match ue {
112            UnblockEvent::WorkflowComplete(_, result) => *result,
113            _ => panic!("Invalid unblock event for child workflow complete"),
114        }
115    }
116}
117
118impl Unblockable for SignalExternalWfResult {
119    type OtherDat = ();
120
121    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
122        match ue {
123            UnblockEvent::SignalExternal(_, maybefail) => {
124                maybefail.map_or(Ok(SignalExternalOk), Err)
125            }
126            _ => panic!("Invalid unblock event for signal external workflow result"),
127        }
128    }
129}
130
131impl Unblockable for CancelExternalWfResult {
132    type OtherDat = ();
133
134    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
135        match ue {
136            UnblockEvent::CancelExternal(_, maybefail) => {
137                maybefail.map_or(Ok(CancelExternalOk), Err)
138            }
139            _ => panic!("Invalid unblock event for cancel external workflow result"),
140        }
141    }
142}
143
144pub(crate) type NexusStartResult = Result<StartedNexusOperation, Failure>;
145
146impl Unblockable for NexusStartResult {
147    type OtherDat = NexusUnblockData;
148
149    fn unblock(ue: UnblockEvent, od: Self::OtherDat) -> Self {
150        let NexusUnblockData {
151            result_future,
152            schedule_seq,
153            base_ctx,
154        } = od;
155        match ue {
156            UnblockEvent::NexusOperationStart(_, result) => match *result {
157                resolve_nexus_operation_start::Status::OperationToken(op_token) => {
158                    Ok(StartedNexusOperation {
159                        operation_token: Some(op_token),
160                        result_future,
161                        schedule_seq,
162                        base_ctx,
163                    })
164                }
165                resolve_nexus_operation_start::Status::StartedSync(_) => {
166                    Ok(StartedNexusOperation {
167                        operation_token: None,
168                        result_future,
169                        schedule_seq,
170                        base_ctx,
171                    })
172                }
173                resolve_nexus_operation_start::Status::Failed(f) => Err(f),
174            },
175            _ => panic!("Invalid unblock event for nexus operation"),
176        }
177    }
178}
179
180impl Unblockable for NexusOperationResult {
181    type OtherDat = ();
182
183    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
184        match ue {
185            UnblockEvent::NexusOperationComplete(_, result) => *result,
186            _ => panic!("Invalid unblock event for nexus operation complete"),
187        }
188    }
189}
190
191#[derive(Debug, Clone)]
192pub enum CancellableID {
193    Timer(u32),
194    Activity(u32),
195    LocalActivity(u32),
196    ChildWorkflow { seqnum: u32, reason: String },
197    SignalExternalWorkflow(u32),
198    NexusOp(u32),
199}
200
201impl CancellableID {
202    pub(crate) fn with_reason(self, reason: String) -> Self {
203        match self {
204            CancellableID::ChildWorkflow { seqnum, .. } => {
205                CancellableID::ChildWorkflow { seqnum, reason }
206            }
207            other => other,
208        }
209    }
210}
211
212/// The result of running a workflow.
213pub type WorkflowResult<T> = Result<T, WorkflowTermination>;
214
215/// Represents ways a workflow can terminate without producing a normal result.
216///
217/// Payload conversion errors returned by workflow operations propagated directly into `WorkflowTermination`, such as with `?`, will fail
218/// the current Workflow Task so it can be retried.
219///
220/// Wrap an error in an [`ApplicationFailure`] to explicitly fail the Workflow Execution.
221#[derive(Debug, thiserror::Error)]
222pub enum WorkflowTermination {
223    #[error("Workflow cancelled")]
224    Cancelled,
225    #[error("Workflow evicted from cache")]
226    Evicted,
227    #[error("Continue as new")]
228    ContinueAsNew(Box<ContinueAsNewRequest>),
229    #[error("Workflow failed: {0}")]
230    Failed(#[source] temporalio_common_wasm::error::OutgoingWorkflowError),
231}
232
233impl WorkflowTermination {
234    pub fn continue_as_new(can: ContinueAsNewRequest) -> Self {
235        Self::ContinueAsNew(Box::new(can))
236    }
237
238    /// Construct a [`WorkflowTermination::Failed`] from an [`ApplicationFailure`].
239    pub fn failed_application(err: ApplicationFailure) -> Self {
240        Self::Failed(err.into())
241    }
242}
243
244impl From<WorkflowCancellationError> for WorkflowTermination {
245    fn from(_value: WorkflowCancellationError) -> Self {
246        Self::Cancelled
247    }
248}
249
250impl From<ApplicationFailure> for WorkflowTermination {
251    fn from(value: ApplicationFailure) -> Self {
252        Self::Failed(value.into())
253    }
254}
255
256impl From<PayloadConversionError> for WorkflowTermination {
257    fn from(value: PayloadConversionError) -> Self {
258        Self::Failed(value.into())
259    }
260}
261
262impl From<ActivityExecutionError> for WorkflowTermination {
263    fn from(value: ActivityExecutionError) -> Self {
264        Self::Failed(value.into())
265    }
266}
267
268impl From<ChildWorkflowExecutionError> for WorkflowTermination {
269    fn from(value: ChildWorkflowExecutionError) -> Self {
270        Self::Failed(value.into())
271    }
272}
273
274impl From<WorkflowSignalError> for WorkflowTermination {
275    fn from(value: WorkflowSignalError) -> Self {
276        Self::Failed(value.into())
277    }
278}
279
280impl From<ChildWorkflowStartError> for WorkflowTermination {
281    fn from(value: ChildWorkflowStartError) -> Self {
282        Self::Failed(value.into())
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use super::*;
289    use rstest::rstest;
290    use temporalio_common_wasm::error::OutgoingWorkflowError;
291
292    fn conversion_error() -> PayloadConversionError {
293        PayloadConversionError::EncodingError(std::io::Error::other("test conversion error").into())
294    }
295
296    #[rstest]
297    #[case::payload(conversion_error())]
298    #[case::activity(ActivityExecutionError::Serialization(conversion_error()))]
299    #[case::child_start(ChildWorkflowStartError::Serialization(conversion_error()))]
300    #[case::child_execution(ChildWorkflowExecutionError::Serialization(conversion_error()))]
301    #[case::signal(WorkflowSignalError::Serialization(conversion_error()))]
302    fn conversion_error_is_preserved_in_workflow_termination<T: Into<WorkflowTermination>>(
303        #[case] error: T,
304    ) {
305        let termination = error.into();
306        let WorkflowTermination::Failed(OutgoingWorkflowError::PayloadConversion(err)) =
307            termination
308        else {
309            panic!("expected a payload conversion failure");
310        };
311        assert_eq!(err.to_string(), "Encoding error: test conversion error");
312    }
313
314    #[test]
315    fn explicitly_wrapped_conversion_error_remains_an_application_failure() {
316        let termination = WorkflowTermination::from(ApplicationFailure::new(conversion_error()));
317
318        assert!(matches!(
319            termination,
320            WorkflowTermination::Failed(OutgoingWorkflowError::Application(_))
321        ));
322    }
323}
temporalio_workflow/runtime/model.rs

temporalio_workflow/runtime/
model.rs
