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, CancelExternalWorkflowError,
19        ChildWorkflowExecutionError, 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::{
32            enums::v1::{
33                CancelExternalWorkflowExecutionFailedCause,
34                SignalExternalWorkflowExecutionFailedCause,
35            },
36            failure::v1::Failure,
37        },
38    },
39};
40
41#[cfg_attr(not(feature = "experimental"), allow(dead_code))]
42#[derive(Debug)]
43pub(crate) enum UnblockEvent {
44    Timer(u32, TimerResult),
45    Activity(u32, Box<ActivityResolution>),
46    WorkflowStart(u32, Box<ChildWorkflowStartStatus>),
47    WorkflowComplete(u32, Box<ChildWorkflowResult>),
48    SignalExternal(u32, Option<SignalExternalWfFailure>),
49    CancelExternal(u32, Option<CancelExternalWfFailure>),
50    NexusOperationStart(u32, Box<resolve_nexus_operation_start::Status>),
51    NexusOperationComplete(u32, Box<NexusOperationResult>),
52}
53
54/// Result of awaiting on a timer
55#[derive(Debug, Copy, Clone, PartialEq, Eq)]
56pub enum TimerResult {
57    /// The timer was cancelled
58    Cancelled,
59    /// The timer elapsed and fired
60    Fired,
61}
62
63/// Successful result of sending a signal to an external workflow
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub(crate) struct SignalExternalOk;
66#[derive(Debug)]
67pub(crate) struct SignalExternalWfFailure {
68    pub(crate) failure: Failure,
69    pub(crate) cause: SignalExternalWorkflowExecutionFailedCause,
70}
71/// Result of awaiting on sending a signal to an external workflow
72pub(crate) type SignalExternalWfResult = Result<SignalExternalOk, SignalExternalWfFailure>;
73
74/// Distinguishes external cancellation resolutions from other command results.
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
76pub(crate) struct CancelExternalOk;
77#[derive(Debug)]
78pub(crate) struct CancelExternalWfFailure {
79    pub(crate) failure: Failure,
80    pub(crate) cause: CancelExternalWorkflowExecutionFailedCause,
81}
82/// Internal result delivered when an external cancellation command resolves.
83pub(crate) type CancelExternalWfResult = Result<CancelExternalOk, CancelExternalWfFailure>;
84
85pub(crate) trait Unblockable {
86    type OtherDat;
87
88    fn unblock(ue: UnblockEvent, od: Self::OtherDat) -> Self;
89}
90
91impl Unblockable for TimerResult {
92    type OtherDat = ();
93
94    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
95        match ue {
96            UnblockEvent::Timer(_, result) => result,
97            _ => panic!("Invalid unblock event for timer"),
98        }
99    }
100}
101
102impl Unblockable for ActivityResolution {
103    type OtherDat = ();
104
105    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
106        match ue {
107            UnblockEvent::Activity(_, result) => *result,
108            _ => panic!("Invalid unblock event for activity"),
109        }
110    }
111}
112
113impl<WD: WorkflowDefinition> Unblockable for PendingChildWorkflow<WD> {
114    type OtherDat = ChildWfCommon;
115
116    fn unblock(ue: UnblockEvent, od: Self::OtherDat) -> Self {
117        match ue {
118            UnblockEvent::WorkflowStart(_, result) => Self {
119                status: *result,
120                common: od,
121                _phantom: std::marker::PhantomData,
122            },
123            _ => panic!("Invalid unblock event for child workflow start"),
124        }
125    }
126}
127
128impl Unblockable for ChildWorkflowResult {
129    type OtherDat = ();
130
131    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
132        match ue {
133            UnblockEvent::WorkflowComplete(_, result) => *result,
134            _ => panic!("Invalid unblock event for child workflow complete"),
135        }
136    }
137}
138
139impl Unblockable for SignalExternalWfResult {
140    type OtherDat = ();
141
142    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
143        match ue {
144            UnblockEvent::SignalExternal(_, maybefail) => {
145                maybefail.map_or(Ok(SignalExternalOk), Err)
146            }
147            _ => panic!("Invalid unblock event for signal external workflow result"),
148        }
149    }
150}
151
152impl Unblockable for CancelExternalWfResult {
153    type OtherDat = ();
154
155    fn unblock(ue: UnblockEvent, _: Self::OtherDat) -> Self {
156        match ue {
157            UnblockEvent::CancelExternal(_, maybefail) => {
158                maybefail.map_or(Ok(CancelExternalOk), Err)
159            }
160            _ => panic!("Invalid unblock event for cancel external workflow result"),
161        }
162    }
163}
164
165#[cfg_attr(not(feature = "experimental"), allow(dead_code))]
166#[derive(Debug, Clone)]
167pub(crate) enum CancellableID {
168    Timer(u32),
169    Activity(u32),
170    LocalActivity(u32),
171    ChildWorkflow { seqnum: u32, reason: String },
172    SignalExternalWorkflow(u32),
173    NexusOp(u32),
174}
175
176impl CancellableID {
177    pub(crate) fn with_reason(self, reason: String) -> Self {
178        match self {
179            CancellableID::ChildWorkflow { seqnum, .. } => {
180                CancellableID::ChildWorkflow { seqnum, reason }
181            }
182            other => other,
183        }
184    }
185}
186
187/// The result of running a workflow.
188pub type WorkflowResult<T> = Result<T, WorkflowTermination>;
189
190/// Represents ways a workflow can terminate without producing a normal result.
191///
192/// Payload conversion errors returned by workflow operations propagated directly into `WorkflowTermination`, such as with `?`, will fail
193/// the current Workflow Task so it can be retried.
194///
195/// Wrap an error in an [`ApplicationFailure`] to explicitly fail the Workflow Execution.
196#[derive(derive_more::Debug, thiserror::Error)]
197pub enum WorkflowTermination {
198    /// The Workflow Execution was cancelled, optionally with user-supplied details.
199    #[error("Workflow cancelled")]
200    Cancelled {
201        /// Optional cancellation details.
202        #[debug(skip)]
203        details: Option<Box<dyn WorkflowOutputValue + Send + Sync>>,
204    },
205    /// The workflow was evicted and must stop without producing a completion command.
206    #[error("Workflow evicted from cache")]
207    Evicted,
208    /// The workflow requested a new run with the supplied command attributes.
209    #[error("Continue as new")]
210    ContinueAsNew(Box<ContinueAsNewRequest>),
211    /// The Workflow Execution failed with an error already converted for outbound handling.
212    #[error("Workflow failed: {0}")]
213    Failed(#[source] temporalio_common_wasm::error::OutgoingWorkflowError),
214}
215
216impl WorkflowTermination {
217    /// Construct a cancelled workflow termination without details.
218    pub fn cancelled() -> Self {
219        Self::Cancelled { details: None }
220    }
221
222    /// Construct a cancelled workflow termination with details that will be converted using the
223    /// active payload converter.
224    pub fn cancelled_with_details<T>(details: T) -> Self
225    where
226        T: TemporalSerializable + Send + Sync + 'static,
227    {
228        Self::Cancelled {
229            details: Some(Box::new(details)),
230        }
231    }
232
233    /// Constructs a termination that asks the worker to continue the workflow as a new run.
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<CancelExternalWorkflowError> for WorkflowTermination {
281    fn from(value: CancelExternalWorkflowError) -> Self {
282        Self::Failed(value.into())
283    }
284}
285
286impl From<ChildWorkflowStartError> for WorkflowTermination {
287    fn from(value: ChildWorkflowStartError) -> Self {
288        Self::Failed(value.into())
289    }
290}
291
292#[cfg(test)]
293mod tests {
294    use super::*;
295    use rstest::rstest;
296    use temporalio_common_wasm::error::OutgoingWorkflowError;
297
298    fn conversion_error() -> PayloadConversionError {
299        PayloadConversionError::EncodingError(std::io::Error::other("test conversion error").into())
300    }
301
302    #[rstest]
303    #[case::payload(conversion_error())]
304    #[case::activity(ActivityExecutionError::Serialization(conversion_error()))]
305    #[case::child_start(ChildWorkflowStartError::Serialization(conversion_error()))]
306    #[case::child_execution(ChildWorkflowExecutionError::Serialization(conversion_error()))]
307    #[case::signal(WorkflowSignalError::Serialization(conversion_error()))]
308    #[case::cancel_external(CancelExternalWorkflowError::Serialization(conversion_error()))]
309    fn conversion_error_is_preserved_in_workflow_termination<T: Into<WorkflowTermination>>(
310        #[case] error: T,
311    ) {
312        let termination = error.into();
313        let WorkflowTermination::Failed(OutgoingWorkflowError::PayloadConversion(err)) =
314            termination
315        else {
316            panic!("expected a payload conversion failure");
317        };
318        assert_eq!(err.to_string(), "Encoding error: test conversion error");
319    }
320
321    #[test]
322    fn explicitly_wrapped_conversion_error_remains_an_application_failure() {
323        let termination = WorkflowTermination::from(ApplicationFailure::new(conversion_error()));
324
325        assert!(matches!(
326            termination,
327            WorkflowTermination::Failed(OutgoingWorkflowError::Application(_))
328        ));
329    }
330}
temporalio_workflow/runtime/model.rs

temporalio_workflow/runtime/
model.rs
