temporalio_workflow/runtime/
model.rs

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

temporalio_workflow/runtime/
model.rs
