temporalio_workflow/runtime/
entry.rs

1//! Runtime entry traits implemented by workflow definitions and message handlers.
2
3use crate::{
4    SyncWorkflowContext, WorkflowContext, WorkflowContextView,
5    runtime::{
6        mark_intercepted_handler_ready, model::WorkflowTermination,
7        types::WorkflowDefinitionDescriptor,
8    },
9    workflow_interceptors::WorkflowOutputValue,
10};
11use futures_util::future::{FutureExt, LocalBoxFuture};
12use std::any::Any;
13use temporalio_common_wasm::{
14    QueryDefinition, SignalDefinition, UpdateDefinition, WorkflowDefinition,
15    data_converters::{
16        GenericPayloadConverter, PayloadConversionError, PayloadConverter, SerializationContext,
17        SerializationContextData, TemporalSerializable,
18    },
19    protos::temporal::api::{
20        common::v1::{Payload, Payloads},
21        failure::v1::Failure,
22    },
23};
24
25/// Error type for workflow operations
26#[derive(Debug, thiserror::Error)]
27pub enum WorkflowError {
28    /// Error during payload conversion
29    #[error("Payload conversion error: {0}")]
30    PayloadConversion(#[from] PayloadConversionError),
31
32    /// Workflow execution error
33    #[error("Workflow execution error: {0}")]
34    Execution(#[from] Box<dyn std::error::Error + Send + Sync>),
35}
36
37impl From<WorkflowError> for Failure {
38    fn from(err: WorkflowError) -> Self {
39        Failure {
40            message: err.to_string(),
41            ..Default::default()
42        }
43    }
44}
45
46fn downcast_handler_input<T: Any>(input: Box<dyn Any>, handler_kind: &'static str) -> T {
47    *input.downcast::<T>().unwrap_or_else(|_| {
48        panic!("typed {handler_kind} dispatch received input with wrong concrete type")
49    })
50}
51
52/// Trait implemented by workflow structs to enable execution by the worker.
53///
54/// This trait is typically generated by the `#[workflow_methods]` macro and should not
55/// be implemented manually in most cases.
56pub trait WorkflowImplementation: Sized + 'static {
57    /// The marker struct for the run method that implements `WorkflowDefinition`
58    type Run: WorkflowDefinition;
59
60    /// Whether this workflow has a user-defined `#[init]` method.
61    /// Set to `true` by the macro when `#[init]` is present, `false` otherwise.
62    const HAS_INIT: bool;
63
64    /// Whether the init method accepts the workflow input.
65    /// If true, input goes to init. If false, input goes to run.
66    const INIT_TAKES_INPUT: bool;
67
68    /// Returns the workflow type name.
69    fn name() -> &'static str;
70
71    /// Returns the exported workflow definition metadata for this workflow.
72    fn definition() -> WorkflowDefinitionDescriptor;
73
74    /// Initialize the workflow instance.
75    fn init(
76        ctx: WorkflowContextView,
77        input: Option<<Self::Run as WorkflowDefinition>::Input>,
78    ) -> Self;
79
80    /// Execute the workflow's main run function.
81    fn run(
82        ctx: WorkflowContext<Self>,
83        input: Option<<Self::Run as WorkflowDefinition>::Input>,
84    ) -> LocalBoxFuture<'static, Result<Box<dyn WorkflowOutputValue>, WorkflowTermination>>;
85
86    /// Decode a signal's payloads into that signal handler's concrete input type.
87    fn decode_signal_input(
88        _name: &str,
89        _payloads: Payloads,
90        _converter: &PayloadConverter,
91    ) -> Result<Option<Box<dyn Any>>, WorkflowError> {
92        Ok(None)
93    }
94
95    /// Dispatch a signal using an already decoded input value.
96    fn dispatch_signal(
97        ctx: WorkflowContext<Self>,
98        name: &str,
99        input: Box<dyn Any>,
100    ) -> LocalBoxFuture<'static, Result<(), WorkflowError>>;
101
102    /// Decode a query's payloads into that query handler's concrete input type.
103    fn decode_query_input(
104        _name: &str,
105        _payloads: &Payloads,
106        _converter: &PayloadConverter,
107    ) -> Result<Option<Box<dyn Any>>, WorkflowError> {
108        Ok(None)
109    }
110
111    /// Dispatch a query using an already decoded input value.
112    fn dispatch_query(
113        &self,
114        ctx: WorkflowContextView,
115        name: &str,
116        input: Box<dyn Any>,
117    ) -> Result<Box<dyn WorkflowOutputValue>, WorkflowError>;
118
119    /// Decode an update's payloads into that update handler's concrete input type.
120    fn decode_update_input(
121        _name: &str,
122        _payloads: Payloads,
123        _converter: &PayloadConverter,
124    ) -> Result<Option<Box<dyn Any>>, WorkflowError> {
125        Ok(None)
126    }
127
128    /// Dispatch an update using an already decoded input value.
129    fn dispatch_update(
130        ctx: WorkflowContext<Self>,
131        name: &str,
132        input: Box<dyn Any>,
133    ) -> LocalBoxFuture<'static, Result<Box<dyn WorkflowOutputValue>, WorkflowError>>;
134
135    /// Validate an update using an already decoded input value.
136    fn validate_update(
137        &self,
138        ctx: WorkflowContextView,
139        name: &str,
140        input: Box<dyn Any>,
141    ) -> Result<(), WorkflowError>;
142}
143
144/// Trait for executing synchronous signal handlers on a workflow.
145pub trait ExecutableSyncSignal<S: SignalDefinition>: WorkflowImplementation {
146    /// Handle an incoming signal with the given input.
147    fn handle(&mut self, ctx: &mut SyncWorkflowContext<Self>, input: S::Input);
148
149    /// Dispatch the signal with an already decoded input.
150    fn dispatch(
151        ctx: WorkflowContext<Self>,
152        input: Box<dyn Any>,
153    ) -> LocalBoxFuture<'static, Result<(), WorkflowError>> {
154        let input = downcast_handler_input::<S::Input>(input, "signal");
155        let mut sync_ctx = ctx.sync_context();
156        ctx.state_mut(|wf| Self::handle(wf, &mut sync_ctx, input));
157        mark_intercepted_handler_ready();
158        std::future::ready(Ok(())).boxed_local()
159    }
160}
161
162/// Trait for executing asynchronous signal handlers on a workflow.
163pub trait ExecutableAsyncSignal<S: SignalDefinition>: WorkflowImplementation {
164    /// Handle an incoming signal with the given input.
165    fn handle(ctx: WorkflowContext<Self>, input: S::Input) -> LocalBoxFuture<'static, ()>;
166
167    /// Dispatch the signal with an already decoded input.
168    fn dispatch(
169        ctx: WorkflowContext<Self>,
170        input: Box<dyn Any>,
171    ) -> LocalBoxFuture<'static, Result<(), WorkflowError>> {
172        let input = downcast_handler_input::<S::Input>(input, "signal");
173        Self::handle(ctx, input).map(|()| Ok(())).boxed_local()
174    }
175}
176
177/// Trait for executing query handlers on a workflow.
178pub trait ExecutableQuery<Q: QueryDefinition>: WorkflowImplementation {
179    /// Handle a query with the given input and return the result.
180    fn handle(
181        &self,
182        ctx: &WorkflowContextView,
183        input: Q::Input,
184    ) -> Result<Q::Output, Box<dyn std::error::Error + Send + Sync>>;
185
186    /// Dispatch the query with an already decoded input.
187    fn dispatch(
188        &self,
189        ctx: &WorkflowContextView,
190        input: Box<dyn Any>,
191    ) -> Result<Box<dyn WorkflowOutputValue>, WorkflowError> {
192        let input = downcast_handler_input::<Q::Input>(input, "query");
193        let output = self.handle(ctx, input).map_err(WorkflowError::Execution)?;
194        Ok(Box::new(output))
195    }
196}
197
198/// Trait for executing synchronous update handlers on a workflow.
199pub trait ExecutableSyncUpdate<U: UpdateDefinition>: WorkflowImplementation {
200    /// Handle an update with the given input and return the result.
201    fn handle(
202        &mut self,
203        ctx: &mut SyncWorkflowContext<Self>,
204        input: U::Input,
205    ) -> Result<U::Output, Box<dyn std::error::Error + Send + Sync>>;
206
207    /// Validate an update before it is applied.
208    fn validate(
209        &self,
210        _ctx: &WorkflowContextView,
211        _input: &U::Input,
212    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
213        Ok(())
214    }
215
216    /// Dispatch the update with an already decoded input.
217    fn dispatch(
218        ctx: WorkflowContext<Self>,
219        input: Box<dyn Any>,
220    ) -> LocalBoxFuture<'static, Result<Box<dyn WorkflowOutputValue>, WorkflowError>> {
221        let input = downcast_handler_input::<U::Input>(input, "update");
222        let mut sync_ctx = ctx.sync_context();
223        let result = ctx.state_mut(|wf| Self::handle(wf, &mut sync_ctx, input));
224        mark_intercepted_handler_ready();
225        match result {
226            Ok(output) => std::future::ready(Ok(Box::new(output) as Box<dyn WorkflowOutputValue>))
227                .boxed_local(),
228            Err(e) => std::future::ready(Err(WorkflowError::Execution(e))).boxed_local(),
229        }
230    }
231
232    /// Dispatch validation with an already decoded input.
233    fn dispatch_validate(
234        &self,
235        ctx: &WorkflowContextView,
236        input: Box<dyn Any>,
237    ) -> Result<(), WorkflowError> {
238        let input = downcast_handler_input::<U::Input>(input, "update validation");
239        self.validate(ctx, &input).map_err(WorkflowError::Execution)
240    }
241}
242
243/// Trait for executing asynchronous update handlers on a workflow.
244pub trait ExecutableAsyncUpdate<U: UpdateDefinition>: WorkflowImplementation {
245    /// Handle an update with the given input and return the result.
246    fn handle(
247        ctx: WorkflowContext<Self>,
248        input: U::Input,
249    ) -> LocalBoxFuture<'static, Result<U::Output, Box<dyn std::error::Error + Send + Sync>>>;
250
251    /// Validate an update before it is applied.
252    fn validate(
253        &self,
254        _ctx: &WorkflowContextView,
255        _input: &U::Input,
256    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
257        Ok(())
258    }
259
260    /// Dispatch the update with an already decoded input.
261    fn dispatch(
262        ctx: WorkflowContext<Self>,
263        input: Box<dyn Any>,
264    ) -> LocalBoxFuture<'static, Result<Box<dyn WorkflowOutputValue>, WorkflowError>> {
265        let input = downcast_handler_input::<U::Input>(input, "update");
266        let future = async move {
267            let output = Self::handle(ctx, input)
268                .await
269                .map_err(WorkflowError::Execution)?;
270            Ok(Box::new(output) as Box<dyn WorkflowOutputValue>)
271        };
272        future.boxed_local()
273    }
274
275    /// Dispatch validation with an already decoded input.
276    fn dispatch_validate(
277        &self,
278        ctx: &WorkflowContextView,
279        input: Box<dyn Any>,
280    ) -> Result<(), WorkflowError> {
281        let input = downcast_handler_input::<U::Input>(input, "update validation");
282        self.validate(ctx, &input).map_err(WorkflowError::Execution)
283    }
284}
285
286/// Serialize handler output to a payload.
287pub(crate) fn serialize_output<O: TemporalSerializable + 'static>(
288    output: &O,
289    converter: &PayloadConverter,
290) -> Result<Payload, WorkflowError> {
291    let ctx = SerializationContext {
292        data: &SerializationContextData::Workflow,
293        converter,
294    };
295    converter.to_payload(&ctx, output).map_err(Into::into)
296}
297
298/// Serialize a workflow result value to a payload.
299pub fn serialize_result<T: TemporalSerializable + 'static>(
300    result: T,
301    converter: &PayloadConverter,
302) -> Result<Payload, WorkflowError> {
303    serialize_output(&result, converter)
304}
temporalio_workflow/runtime/entry.rs

temporalio_workflow/runtime/
entry.rs
