temporalio_workflow/runtime/
model.rs

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

temporalio_workflow/runtime/
model.rs
