temporalio_workflow/
workflow_interceptors.rs

1//! Intercept inbound and outbound calls made during workflow execution.
2//!
3//! Workflow interceptors allow observing, transforming, or short-circuit workflow
4//! operations without putting that behavior in each workflow implementation.
5//!
6//! [`WorkflowInterceptor`] has two groups of methods:
7//!
8//! - Inbound methods wrap calls into workflow code, such as executing the workflow or handling
9//!   a signal, query, or update.
10//! - Outbound methods wrap commands issued by workflow code, such as scheduling an activity,
11//!   starting a timer, or signaling another workflow.
12//!
13//! Each method receives a [`WorkflowNext`] continuation. An interceptor can change the input
14//! before calling [`WorkflowNext::run`], inspect or change the returned value, or deliberately not
15//! call `next` to short-circuit the operation. Most interceptors should call `next` exactly once.
16//!
17//! Async operation interceptors return [`WorkflowInterceptorFuture`]. Wrap an `async` block with
18//! [`WorkflowInterceptorFuture::new`] when work must happen after the next interceptor completes.
19//! Synchronous methods, including queries and update validators, cannot await workflow operations.
20//!
21//! Workers register interceptors with `register_workflow_interceptors` on their
22//! worker options. Interceptors are entered in insertion order for inbound calls and in reverse
23//! insertion order for outbound calls.
24//!
25//! # Determinism
26//!
27//! Interceptors execute as part of the workflow and are replayed with it. They must follow the same
28//! determinism rules as workflow code: do not read wall-clock time, perform network or filesystem
29//! I/O, use nondeterministic randomness, or await arbitrary futures. Use values from the
30//! interceptor context and SDK-provided workflow futures instead. [`WorkflowInterceptorFuture`]
31//! identifies a future for the workflow scheduler; it does not make an arbitrary future
32//! deterministic.
33//!
34//! [`WorkflowInterceptorContext::is_replaying`] and
35//! [`WorkflowInterceptorContext::is_replaying_history_events`] can be used to suppress duplicate external
36//! observability during replay, but replay state must not change commands or results that affect
37//! workflow behavior.
38//!
39//! # Example
40//!
41//! This interceptor wraps workflow execution and transforms string outputs after the workflow has
42//! completed. The constructor is passed to the worker during worker setup.
43//!
44//! ```
45//! # use temporalio_workflow::{
46//! #     WorkflowContextView,
47//! #     workflow_interceptors::{
48//! #         ExecuteWorkflowInput, ExecuteWorkflowResult, WorkflowInterceptor,
49//! #         WorkflowInterceptorConstructor, WorkflowInterceptorContext, WorkflowInterceptorFuture,
50//! #         WorkflowNext, WorkflowOutputValue,
51//! #     },
52//! # };
53//!
54//! struct UppercaseStringOutput;
55//!
56//! impl WorkflowInterceptor for UppercaseStringOutput {
57//!     fn execute<'a>(
58//!         &'a self,
59//!         _ctx: WorkflowInterceptorContext,
60//!         input: ExecuteWorkflowInput,
61//!         next: WorkflowNext<
62//!             'a,
63//!             ExecuteWorkflowInput,
64//!             WorkflowInterceptorFuture<'a, ExecuteWorkflowResult>,
65//!         >,
66//!     ) -> WorkflowInterceptorFuture<'a, ExecuteWorkflowResult> {
67//!         WorkflowInterceptorFuture::new(async move {
68//!             let output = next.run(input).await?;
69//!             if let Some(value) = output.downcast_ref::<String>() {
70//!                 return Ok(Box::new(value.to_uppercase()) as Box<dyn WorkflowOutputValue>);
71//!             }
72//!             Ok(output)
73//!         })
74//!     }
75//! }
76//!
77//! fn interceptor_constructor() -> WorkflowInterceptorConstructor {
78//!     WorkflowInterceptorConstructor::new(|_ctx: &WorkflowContextView| UppercaseStringOutput)
79//! }
80//!
81//! # let _ = interceptor_constructor();
82//! ```
83
84use crate::{
85    ActivityOptions, BaseWorkflowContext, CancellableFuture, CancellableFutureWithReason,
86    ChildWorkflowOptions, ContinueAsNewOptions, ExternalWorkflowHandle, LocalActivityOptions,
87    SignalWorkflowOptions, StartChildWorkflowOutput, StartedChildWorkflow, TimerOptions,
88    WorkflowCancellationToken, WorkflowContextView, WorkflowRandomStream,
89    cancellation::WorkflowCancellationRegistration,
90    runtime::{
91        entry::WorkflowError,
92        model::{CancelExternalWfResult, TimerResult, WorkflowResult, WorkflowTermination},
93    },
94};
95use futures_util::{
96    FutureExt,
97    future::{Fuse, FusedFuture, LocalBoxFuture},
98};
99use std::{
100    any::Any,
101    collections::HashMap,
102    convert::Infallible,
103    future::Future,
104    pin::Pin,
105    rc::Rc,
106    sync::Arc,
107    task::{Context, Poll},
108    time::SystemTime,
109};
110use temporalio_common_wasm::{
111    ActivityDefinition, WorkflowDefinition,
112    data_converters::{
113        GenericPayloadConverter, PayloadConversionError, PayloadConverter, SerializationContext,
114        SerializationContextData, TemporalDeserializable, TemporalSerializable,
115        WorkflowSerializationContext,
116    },
117    error::{
118        ActivityExecutionError, ChildWorkflowExecutionError, ChildWorkflowStartError,
119        WorkflowSignalError,
120    },
121    protos::temporal::api::common::v1::Payload,
122    search_attributes::SearchAttributes,
123};
124
125#[cfg(feature = "experimental")]
126pub(crate) use nexus::call_start_nexus_operation;
127#[cfg(feature = "experimental")]
128pub use nexus::{StartNexusOperationInput, StartNexusOperationResult};
129
130mod workflow_output_value {
131    use super::*;
132
133    pub trait Sealed {
134        fn to_workflow_payload(
135            &self,
136            context: &SerializationContext<'_>,
137        ) -> Result<Payload, PayloadConversionError>;
138
139        fn to_workflow_payloads(
140            &self,
141            context: &SerializationContext<'_>,
142        ) -> Result<Vec<Payload>, PayloadConversionError>;
143    }
144
145    impl<T> Sealed for T
146    where
147        T: Any + TemporalSerializable,
148    {
149        fn to_workflow_payload(
150            &self,
151            context: &SerializationContext<'_>,
152        ) -> Result<Payload, PayloadConversionError> {
153            context.converter.to_payload(context, self)
154        }
155
156        fn to_workflow_payloads(
157            &self,
158            context: &SerializationContext<'_>,
159        ) -> Result<Vec<Payload>, PayloadConversionError> {
160            context.converter.to_payloads(context, self)
161        }
162    }
163}
164
165/// Type-erased workflow output carried through the workflow interceptor chain.
166pub trait WorkflowOutputValue: Any + TemporalSerializable + workflow_output_value::Sealed {
167    /// Access this value as [`Any`] for type-specific inspection.
168    fn as_any(&self) -> &dyn Any;
169}
170
171impl<T> WorkflowOutputValue for T
172where
173    T: Any + TemporalSerializable,
174{
175    fn as_any(&self) -> &dyn Any {
176        self
177    }
178}
179
180impl dyn WorkflowOutputValue {
181    /// Attempt to access the workflow output as a concrete type.
182    pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
183        self.as_any().downcast_ref()
184    }
185
186    pub(crate) fn serialize_payload(
187        &self,
188        context: &SerializationContext<'_>,
189    ) -> Result<Payload, PayloadConversionError> {
190        self.to_workflow_payload(context)
191    }
192
193    pub(crate) fn serialize_payloads(
194        &self,
195        context: &SerializationContext<'_>,
196    ) -> Result<Vec<Payload>, PayloadConversionError> {
197        self.to_workflow_payloads(context)
198    }
199}
200
201pub(crate) fn serialize_workflow_output(
202    output: &dyn WorkflowOutputValue,
203    converter: &PayloadConverter,
204) -> Result<Payload, PayloadConversionError> {
205    let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
206    let ctx = SerializationContext::new(&context_data, converter);
207    output.serialize_payload(&ctx)
208}
209
210/// Result of an intercepted workflow execution.
211pub type ExecuteWorkflowResult = WorkflowResult<Box<dyn WorkflowOutputValue>>;
212
213/// Result of an intercepted signal handler.
214pub type HandleSignalResult = Result<(), WorkflowError>;
215
216/// Result of an intercepted update handler.
217pub type HandleUpdateResult = Result<Box<dyn WorkflowOutputValue>, WorkflowError>;
218
219/// Result of an intercepted query handler.
220pub type HandleQueryResult = Result<Box<dyn WorkflowOutputValue>, WorkflowError>;
221
222/// Result of an intercepted update validator.
223pub type ValidateUpdateResult = Result<(), WorkflowError>;
224
225/// Future produced by workflow interceptors.
226///
227/// The SDK polls a newly created interceptor future once while processing its activation. This
228/// runs synchronous interceptor and synchronous handler work through the first genuine
229/// [`Poll::Pending`]. Async workflow and handler bodies are not entered
230/// until normal routine polling. Awaiting a pending workflow future before calling
231/// [`WorkflowNext::run`] intentionally delays the underlying handler.
232///
233/// This type identifies futures that are polled inside workflow execution. It does not make
234/// arbitrary Rust futures deterministic. Interceptor implementations must only await workflow
235/// scheduler primitives or SDK-provided workflow futures.
236pub struct WorkflowInterceptorFuture<'a, T>(LocalBoxFuture<'a, T>);
237
238impl<'a, T> WorkflowInterceptorFuture<'a, T> {
239    /// Create a workflow interceptor future from a local future.
240    pub fn new(fut: impl Future<Output = T> + 'a) -> Self {
241        Self(fut.boxed_local())
242    }
243}
244
245impl<'a, T> Unpin for WorkflowInterceptorFuture<'a, T> {}
246
247impl<T> Future for WorkflowInterceptorFuture<'_, T> {
248    type Output = T;
249
250    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
251        self.0.as_mut().poll(cx)
252    }
253}
254
255/// Continuation for a workflow interceptor operation.
256pub struct WorkflowNext<'a, I, O> {
257    inner: Box<dyn FnOnce(I) -> O + 'a>,
258}
259
260impl<'a, I, O> WorkflowNext<'a, I, O> {
261    pub(crate) fn new(f: impl FnOnce(I) -> O + 'a) -> Self {
262        Self { inner: Box::new(f) }
263    }
264
265    /// Continue the call chain with the provided input.
266    pub fn run(self, input: I) -> O {
267        (self.inner)(input)
268    }
269}
270
271/// Workflow execution context available to async-capable inbound interceptors.
272#[derive(Clone)]
273pub struct WorkflowInterceptorContext {
274    base: BaseWorkflowContext,
275}
276
277impl WorkflowInterceptorContext {
278    pub(crate) fn new(base: BaseWorkflowContext) -> Self {
279        Self { base }
280    }
281
282    /// Return the workflow's unique identifier.
283    pub fn workflow_id(&self) -> &str {
284        self.base.workflow_id()
285    }
286
287    /// Return the run id of this workflow execution.
288    pub fn run_id(&self) -> &str {
289        self.base.run_id()
290    }
291
292    /// Return the namespace the workflow is executing in.
293    pub fn namespace(&self) -> &str {
294        self.base.namespace()
295    }
296
297    /// Return the task queue the workflow is executing in.
298    pub fn task_queue(&self) -> &str {
299        self.base.task_queue()
300    }
301
302    /// Return the workflow type name.
303    pub fn workflow_type(&self) -> &str {
304        self.base.workflow_type()
305    }
306
307    /// Return the current time according to the workflow.
308    pub fn workflow_time(&self) -> Option<SystemTime> {
309        self.base.workflow_time()
310    }
311
312    /// Return the length of history so far at this point in the workflow.
313    pub fn history_length(&self) -> u32 {
314        self.base.history_length()
315    }
316
317    /// Return current values for workflow search attributes.
318    pub fn search_attributes(&self) -> SearchAttributes {
319        self.base.search_attributes()
320    }
321
322    /// Returns true if the workflow is replaying (including during queries and update validators), false otherwise.
323    pub fn is_replaying(&self) -> bool {
324        self.base.is_replaying()
325    }
326
327    /// Return true if the workflow is replaying history events (excluding queries and update validators), false otherwise.
328    pub fn is_replaying_history_events(&self) -> bool {
329        self.base.is_replaying_history_events()
330    }
331
332    /// Returns the payload converter used by the worker running this workflow.
333    pub fn payload_converter(&self) -> &PayloadConverter {
334        self.base.payload_converter()
335    }
336
337    /// Return the workflow's root cancellation token.
338    pub fn cancellation_token(&self) -> WorkflowCancellationToken {
339        self.base.cancellation_token()
340    }
341
342    /// Returns the deterministic pseudo-random stream associated with `name`.
343    ///
344    /// Named streams let interceptors consume replay-safe randomness without changing the
345    /// workflow's default random sequence or another interceptor's named sequence. Query and
346    /// update-validator interceptors receive [`SyncWorkflowInterceptorContext`], which does not
347    /// expose random streams because those handlers are read-only.
348    pub fn random_stream(&self, name: impl Into<String>) -> WorkflowRandomStream {
349        self.base.random_stream(name)
350    }
351
352    /// Request to create a timer through the workflow outbound interceptor chain.
353    pub fn timer<T: Into<TimerOptions>>(
354        &self,
355        opts: T,
356    ) -> impl CancellableFuture<Output = TimerResult> + use<T> {
357        self.base.timer(opts)
358    }
359
360    /// Request to run an activity through the workflow outbound interceptor chain.
361    pub fn execute_activity<AD: ActivityDefinition>(
362        &self,
363        activity: AD,
364        input: impl Into<AD::Input>,
365        opts: ActivityOptions,
366    ) -> impl CancellableFuture<Output = Result<AD::Output, ActivityExecutionError>>
367    where
368        AD::Output: TemporalDeserializable,
369    {
370        self.base.execute_activity(activity, input, opts)
371    }
372
373    /// Request to run a local activity through the workflow outbound interceptor chain.
374    pub fn execute_local_activity<AD: ActivityDefinition>(
375        &self,
376        activity: AD,
377        input: impl Into<AD::Input>,
378        opts: LocalActivityOptions,
379    ) -> impl CancellableFuture<Output = Result<AD::Output, ActivityExecutionError>>
380    where
381        AD::Output: TemporalDeserializable,
382    {
383        self.base.execute_local_activity(activity, input, opts)
384    }
385
386    /// Start a child workflow through the workflow outbound interceptor chain.
387    pub fn start_child_workflow<WD: WorkflowDefinition + 'static>(
388        &self,
389        workflow: WD,
390        input: impl Into<WD::Input>,
391        opts: ChildWorkflowOptions,
392    ) -> impl CancellableFutureWithReason<
393        Output = Result<StartedChildWorkflow<WD>, ChildWorkflowStartError>,
394    >
395    where
396        WD::Output: TemporalDeserializable,
397    {
398        self.base.start_child_workflow(workflow, input, opts)
399    }
400
401    /// Get a handle to an external workflow for signaling or requesting cancellation.
402    pub fn external_workflow(
403        &self,
404        workflow_id: impl Into<String>,
405        run_id: Option<String>,
406    ) -> ExternalWorkflowHandle {
407        self.base.external_workflow(workflow_id, run_id)
408    }
409}
410
411/// Workflow execution context available to sync-only inbound interceptors.
412#[derive(Clone)]
413pub struct SyncWorkflowInterceptorContext {
414    base: BaseWorkflowContext,
415}
416
417impl SyncWorkflowInterceptorContext {
418    pub(crate) fn new(base: BaseWorkflowContext) -> Self {
419        Self { base }
420    }
421
422    /// Return the workflow's unique identifier.
423    pub fn workflow_id(&self) -> &str {
424        self.base.workflow_id()
425    }
426
427    /// Return the run id of this workflow execution.
428    pub fn run_id(&self) -> &str {
429        self.base.run_id()
430    }
431
432    /// Return the namespace the workflow is executing in.
433    pub fn namespace(&self) -> &str {
434        self.base.namespace()
435    }
436
437    /// Return the task queue the workflow is executing in.
438    pub fn task_queue(&self) -> &str {
439        self.base.task_queue()
440    }
441
442    /// Return the workflow type name.
443    pub fn workflow_type(&self) -> &str {
444        self.base.workflow_type()
445    }
446
447    /// Return the workflow's root cancellation token.
448    pub fn cancellation_token(&self) -> WorkflowCancellationToken {
449        self.base.cancellation_token()
450    }
451
452    /// Return the current time according to the workflow.
453    pub fn workflow_time(&self) -> Option<SystemTime> {
454        self.base.workflow_time()
455    }
456
457    /// Return the length of history so far at this point in the workflow.
458    pub fn history_length(&self) -> u32 {
459        self.base.history_length()
460    }
461
462    /// Return current values for workflow search attributes.
463    pub fn search_attributes(&self) -> SearchAttributes {
464        self.base.search_attributes()
465    }
466
467    /// Returns true if the current workflow task is happening under replay.
468    pub fn is_replaying(&self) -> bool {
469        self.base.is_replaying()
470    }
471
472    /// Returns true if the current work is replaying history events.
473    pub fn is_replaying_history_events(&self) -> bool {
474        self.base.is_replaying_history_events()
475    }
476
477    /// Returns the payload converter used by the worker running this workflow.
478    pub fn payload_converter(&self) -> &PayloadConverter {
479        self.base.payload_converter()
480    }
481}
482
483struct DecodedInput {
484    value: Option<Box<dyn Any>>,
485    headers: HashMap<String, Payload>,
486}
487
488impl DecodedInput {
489    fn new(value: Option<Box<dyn Any>>, headers: HashMap<String, Payload>) -> Self {
490        Self { value, headers }
491    }
492
493    fn input_ref<T: Any>(&self) -> Option<&T> {
494        self.value.as_ref()?.downcast_ref()
495    }
496
497    fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
498        self.value.as_mut()?.downcast_mut()
499    }
500
501    fn headers(&self) -> &HashMap<String, Payload> {
502        &self.headers
503    }
504
505    fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
506        &mut self.headers
507    }
508
509    fn into_parts(self) -> (Option<Box<dyn Any>>, HashMap<String, Payload>) {
510        (self.value, self.headers)
511    }
512}
513
514/// Input passed to [`WorkflowInterceptor::initialize_workflow`].
515///
516/// The decoded input provided to workflow's `#[init]` method.
517/// If a workflow has no `#[init]`, inputs are instead passed to [`WorkflowInterceptor::execute`].
518#[non_exhaustive]
519pub struct InitializeWorkflowInput {
520    decoded: DecodedInput,
521}
522
523impl InitializeWorkflowInput {
524    pub(crate) fn new(value: Option<Box<dyn Any>>, headers: HashMap<String, Payload>) -> Self {
525        Self {
526            decoded: DecodedInput::new(value, headers),
527        }
528    }
529
530    pub(crate) fn into_parts(self) -> (Option<Box<dyn Any>>, HashMap<String, Payload>) {
531        self.decoded.into_parts()
532    }
533
534    /// Attempt to access the decoded workflow input as a concrete type.
535    pub fn input_ref<T: Any>(&self) -> Option<&T> {
536        self.decoded.input_ref()
537    }
538
539    /// Attempt to mutably access the decoded workflow input as a concrete type.
540    pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
541        self.decoded.input_mut()
542    }
543
544    /// Headers attached to the workflow execution.
545    pub fn headers(&self) -> &HashMap<String, Payload> {
546        self.decoded.headers()
547    }
548
549    /// Mutably access headers attached to the workflow execution.
550    pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
551        self.decoded.headers_mut()
552    }
553}
554
555/// Result of workflow initialization.
556pub struct InitializeWorkflowOutput {
557    _private: (),
558}
559
560impl InitializeWorkflowOutput {
561    pub(crate) fn new() -> Self {
562        Self { _private: () }
563    }
564}
565
566/// Input passed to [`WorkflowInterceptor::execute`].
567///
568/// The decoded input provided to workflow's `#[run]` method.
569/// Inputs consumed by `#[init]` are instead passed to [`WorkflowInterceptor::initialize_workflow`].
570#[non_exhaustive]
571pub struct ExecuteWorkflowInput {
572    decoded: DecodedInput,
573}
574
575impl ExecuteWorkflowInput {
576    pub(crate) fn new(value: Option<Box<dyn Any>>, headers: HashMap<String, Payload>) -> Self {
577        Self {
578            decoded: DecodedInput::new(value, headers),
579        }
580    }
581
582    pub(crate) fn into_parts(self) -> (Option<Box<dyn Any>>, HashMap<String, Payload>) {
583        self.decoded.into_parts()
584    }
585
586    /// Attempt to access the decoded workflow input as a concrete type.
587    pub fn input_ref<T: Any>(&self) -> Option<&T> {
588        self.decoded.input_ref()
589    }
590
591    /// Attempt to mutably access the decoded workflow input as a concrete type.
592    pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
593        self.decoded.input_mut()
594    }
595
596    /// Headers attached to the workflow execution.
597    pub fn headers(&self) -> &HashMap<String, Payload> {
598        self.decoded.headers()
599    }
600
601    /// Mutably access headers attached to the workflow execution.
602    pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
603        self.decoded.headers_mut()
604    }
605}
606
607macro_rules! handler_input {
608    ($name:ident, $doc:literal, $field:ident, $field_doc:literal $(, $id_field:ident, $id_doc:literal)?) => {
609        #[doc = $doc]
610        #[non_exhaustive]
611        pub struct $name {
612            $($id_field: String,)?
613            $field: String,
614            decoded: DecodedInput,
615        }
616
617        impl $name {
618            pub(crate) fn new(
619                $($id_field: String,)?
620                $field: String,
621                value: Box<dyn Any>,
622                headers: HashMap<String, Payload>,
623            ) -> Self {
624                Self {
625                    $($id_field,)?
626                    $field,
627                    decoded: DecodedInput::new(Some(value), headers),
628                }
629            }
630
631            pub(crate) fn into_parts(self) -> (String, Box<dyn Any>, HashMap<String, Payload>) {
632                let (value, headers) = self.decoded.into_parts();
633                (
634                    self.$field,
635                    value.expect("handler input must exist after typed decode"),
636                    headers,
637                )
638            }
639
640            #[doc = $field_doc]
641            pub fn name(&self) -> &str {
642                &self.$field
643            }
644
645            $(
646                #[doc = $id_doc]
647                pub fn id(&self) -> &str {
648                    &self.$id_field
649                }
650            )?
651
652            /// Attempt to access the decoded input as a concrete type.
653            pub fn input_ref<T: Any>(&self) -> Option<&T> {
654                self.decoded.input_ref()
655            }
656
657            /// Attempt to mutably access the decoded input as a concrete type.
658            pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
659                self.decoded.input_mut()
660            }
661
662            /// Headers attached to this handler invocation.
663            pub fn headers(&self) -> &HashMap<String, Payload> {
664                self.decoded.headers()
665            }
666
667            /// Mutably access headers attached to this handler invocation.
668            pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
669                self.decoded.headers_mut()
670            }
671        }
672    };
673}
674
675handler_input!(
676    HandleSignalInput,
677    "Input passed to [`WorkflowInterceptor::handle_signal`].",
678    signal_name,
679    "Return the signal name."
680);
681
682handler_input!(
683    HandleUpdateInput,
684    "Input passed to [`WorkflowInterceptor::handle_update`].",
685    update_name,
686    "Return the update name.",
687    update_id,
688    "Return the update ID."
689);
690
691handler_input!(
692    HandleQueryInput,
693    "Input passed to [`WorkflowInterceptor::handle_query`].",
694    query_name,
695    "Return the query name.",
696    query_id,
697    "Return the query ID."
698);
699
700/// Input passed to [`WorkflowInterceptor::validate_update`].
701#[non_exhaustive]
702pub struct ValidateUpdateInput {
703    update_id: String,
704    update_name: String,
705    decoded: DecodedInput,
706}
707
708impl ValidateUpdateInput {
709    pub(crate) fn new(
710        update_id: String,
711        update_name: String,
712        value: Box<dyn Any>,
713        headers: HashMap<String, Payload>,
714    ) -> Self {
715        Self {
716            update_id,
717            update_name,
718            decoded: DecodedInput::new(Some(value), headers),
719        }
720    }
721
722    pub(crate) fn into_parts(self) -> (String, Box<dyn Any>, HashMap<String, Payload>) {
723        let (value, headers) = self.decoded.into_parts();
724        (
725            self.update_name,
726            value.expect("update validation input must exist after typed decode"),
727            headers,
728        )
729    }
730
731    /// Return the update name.
732    pub fn name(&self) -> &str {
733        &self.update_name
734    }
735
736    /// Return the update ID.
737    pub fn id(&self) -> &str {
738        &self.update_id
739    }
740
741    /// Attempt to access the decoded input as a concrete type.
742    pub fn input_ref<T: Any>(&self) -> Option<&T> {
743        self.decoded.input_ref()
744    }
745
746    /// Attempt to mutably access the decoded input as a concrete type.
747    pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
748        self.decoded.input_mut()
749    }
750
751    /// Headers attached to this update invocation.
752    pub fn headers(&self) -> &HashMap<String, Payload> {
753        self.decoded.headers()
754    }
755
756    /// Mutably access headers attached to this update invocation.
757    pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
758        self.decoded.headers_mut()
759    }
760}
761
762/// Type-erased output returned by an intercepted outbound workflow call.
763pub trait WorkflowOutboundValue: Any {
764    /// Access the concrete value through [`Any`].
765    fn as_any(&self) -> &dyn Any;
766
767    /// Convert this value into [`Any`] for a consuming downcast.
768    fn into_any(self: Box<Self>) -> Box<dyn Any>;
769}
770
771impl<T: Any> WorkflowOutboundValue for T {
772    fn as_any(&self) -> &dyn Any {
773        self
774    }
775
776    fn into_any(self: Box<Self>) -> Box<dyn Any> {
777        self
778    }
779}
780
781impl dyn WorkflowOutboundValue {
782    /// Attempt to access the output as a concrete type.
783    pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
784        self.as_any().downcast_ref()
785    }
786
787    /// Attempt to convert the output into a concrete type.
788    pub fn downcast<T: Any>(self: Box<Self>) -> Result<Box<T>, Box<dyn Any>> {
789        self.into_any().downcast()
790    }
791}
792
793/// Future returned by a non-cancellable outbound interceptor operation.
794pub struct WorkflowOutboundFuture<T> {
795    state: WorkflowOutboundFutureState<T>,
796}
797
798enum WorkflowOutboundFutureState<T> {
799    Running(Fuse<LocalBoxFuture<'static, T>>),
800    Prefetched(Option<T>),
801    Terminated,
802}
803
804impl<T> WorkflowOutboundFuture<T> {
805    /// Create an outbound future.
806    pub fn new(future: impl Future<Output = T> + 'static) -> Self {
807        Self {
808            state: WorkflowOutboundFutureState::Running(future.boxed_local().fuse()),
809        }
810    }
811
812    /// Create an immediately ready outbound future.
813    pub fn ready(value: T) -> Self
814    where
815        T: 'static,
816    {
817        Self::new(async move { value })
818    }
819
820    /// Transform the result of this future.
821    pub fn map<U>(self, map: impl FnOnce(T) -> U + 'static) -> WorkflowOutboundFuture<U>
822    where
823        T: 'static,
824        U: 'static,
825    {
826        WorkflowOutboundFuture::new(async move { map(self.await) })
827    }
828
829    pub(crate) fn poll_for_construction(&mut self, cx: &mut Context<'_>) {
830        let WorkflowOutboundFutureState::Running(future) = &mut self.state else {
831            return;
832        };
833        if let Poll::Ready(value) = future.poll_unpin(cx) {
834            self.state = WorkflowOutboundFutureState::Prefetched(Some(value));
835        }
836    }
837}
838
839impl<T> Unpin for WorkflowOutboundFuture<T> {}
840
841impl<T> Future for WorkflowOutboundFuture<T> {
842    type Output = T;
843
844    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
845        match &mut self.state {
846            WorkflowOutboundFutureState::Running(future) => {
847                let result = future.poll_unpin(cx);
848                if result.is_ready() {
849                    self.state = WorkflowOutboundFutureState::Terminated;
850                }
851                result
852            }
853            WorkflowOutboundFutureState::Prefetched(value) => {
854                let value = value
855                    .take()
856                    .expect("outbound future polled after completion");
857                self.state = WorkflowOutboundFutureState::Terminated;
858                Poll::Ready(value)
859            }
860            WorkflowOutboundFutureState::Terminated => {
861                panic!("outbound future polled after completion")
862            }
863        }
864    }
865}
866
867impl<T> FusedFuture for WorkflowOutboundFuture<T> {
868    fn is_terminated(&self) -> bool {
869        matches!(self.state, WorkflowOutboundFutureState::Terminated)
870    }
871}
872
873/// Cancellation callback retained when an interceptor wraps an operation future.
874#[derive(Clone)]
875pub struct WorkflowCancellationHandle {
876    cancel: Rc<dyn Fn(Option<String>)>,
877}
878
879impl WorkflowCancellationHandle {
880    /// Create a cancellation handle.
881    pub fn new(cancel: impl Fn(Option<String>) + 'static) -> Self {
882        Self {
883            cancel: Rc::new(cancel),
884        }
885    }
886
887    pub(crate) fn noop() -> Self {
888        Self::new(|_| {})
889    }
890
891    /// Cancel with an optional reason.
892    pub fn cancel(&self, reason: Option<String>) {
893        (self.cancel)(reason);
894    }
895}
896
897/// Future returned by a cancellable outbound interceptor operation.
898pub struct CancellableWorkflowOutboundFuture<T> {
899    inner: WorkflowOutboundFuture<T>,
900    cancellation: WorkflowCancellationHandle,
901    cancellation_registration: Option<WorkflowCancellationRegistration>,
902}
903
904impl<T> CancellableWorkflowOutboundFuture<T> {
905    /// Create a cancellable outbound future.
906    pub fn new(
907        future: impl Future<Output = T> + 'static,
908        cancellation: WorkflowCancellationHandle,
909    ) -> Self {
910        Self {
911            inner: WorkflowOutboundFuture::new(future),
912            cancellation,
913            cancellation_registration: None,
914        }
915    }
916
917    pub(crate) fn with_cancellation_token(mut self, token: WorkflowCancellationToken) -> Self {
918        let cancellation = self.cancellation.clone();
919        self.cancellation_registration = Some(token.register(move |reason| {
920            cancellation.cancel(reason);
921        }));
922        self
923    }
924
925    pub(crate) fn unregister_cancellation(&mut self) {
926        if let Some(registration) = &mut self.cancellation_registration {
927            registration.unregister();
928        }
929    }
930
931    /// Return the operation's cancellation handle.
932    pub fn cancellation_handle(&self) -> WorkflowCancellationHandle {
933        self.cancellation.clone()
934    }
935
936    /// Transform the result while retaining cancellation behavior.
937    pub fn map<U>(self, map: impl FnOnce(T) -> U + 'static) -> CancellableWorkflowOutboundFuture<U>
938    where
939        T: 'static,
940        U: 'static,
941    {
942        let cancellation = self.cancellation.clone();
943        CancellableWorkflowOutboundFuture::new(async move { map(self.await) }, cancellation)
944    }
945
946    pub(crate) fn poll_for_construction(&mut self, cx: &mut Context<'_>) {
947        self.inner.poll_for_construction(cx);
948    }
949}
950
951impl<T> Unpin for CancellableWorkflowOutboundFuture<T> {}
952
953impl<T> Future for CancellableWorkflowOutboundFuture<T> {
954    type Output = T;
955
956    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
957        let result = Pin::new(&mut self.inner).poll(cx);
958        if result.is_ready() {
959            self.unregister_cancellation();
960        }
961        result
962    }
963}
964
965impl<T> FusedFuture for CancellableWorkflowOutboundFuture<T> {
966    fn is_terminated(&self) -> bool {
967        self.inner.is_terminated()
968    }
969}
970
971impl<T> CancellableFuture for CancellableWorkflowOutboundFuture<T> {
972    fn cancel(&self) {
973        if !self.inner.is_terminated() {
974            self.cancellation.cancel(None);
975        }
976    }
977}
978
979impl<T> CancellableFutureWithReason for CancellableWorkflowOutboundFuture<T> {
980    fn cancel_with_reason(&self, reason: String) {
981        if !self.inner.is_terminated() {
982            self.cancellation.cancel(Some(reason));
983        }
984    }
985}
986
987macro_rules! typed_outbound_input {
988    ($name:ident) => {
989        impl $name {
990            /// Attempt to access the decoded input as a concrete type.
991            pub fn input_ref<T: Any>(&self) -> Option<&T> {
992                self.decoded.input_ref()
993            }
994
995            /// Attempt to mutably access the decoded input as a concrete type.
996            pub fn input_mut<T: Any>(&mut self) -> Option<&mut T> {
997                self.decoded.input_mut()
998            }
999
1000            /// Headers attached to this outbound call.
1001            pub fn headers(&self) -> &HashMap<String, Payload> {
1002                self.decoded.headers()
1003            }
1004
1005            /// Mutably access headers attached to this outbound call.
1006            pub fn headers_mut(&mut self) -> &mut HashMap<String, Payload> {
1007                self.decoded.headers_mut()
1008            }
1009        }
1010    };
1011}
1012
1013/// Input passed to [`WorkflowInterceptor::start_timer`].
1014#[non_exhaustive]
1015pub struct StartTimerInput {
1016    options: TimerOptions,
1017}
1018
1019impl StartTimerInput {
1020    pub(crate) fn new(options: TimerOptions) -> Self {
1021        Self { options }
1022    }
1023
1024    pub(crate) fn into_options(self) -> TimerOptions {
1025        self.options
1026    }
1027
1028    /// Timer options.
1029    pub fn options(&self) -> &TimerOptions {
1030        &self.options
1031    }
1032
1033    /// Mutably access timer options.
1034    pub fn options_mut(&mut self) -> &mut TimerOptions {
1035        &mut self.options
1036    }
1037}
1038
1039/// Input passed to [`WorkflowInterceptor::schedule_activity`].
1040#[non_exhaustive]
1041pub struct ScheduleActivityInput {
1042    activity_type: String,
1043    decoded: DecodedInput,
1044    options: ActivityOptions,
1045}
1046
1047impl ScheduleActivityInput {
1048    pub(crate) fn new(
1049        activity_type: String,
1050        input: Box<dyn Any>,
1051        options: ActivityOptions,
1052    ) -> Self {
1053        Self {
1054            activity_type,
1055            decoded: DecodedInput::new(Some(input), HashMap::new()),
1056            options,
1057        }
1058    }
1059
1060    pub(crate) fn into_parts(
1061        self,
1062    ) -> (
1063        String,
1064        Box<dyn Any>,
1065        HashMap<String, Payload>,
1066        ActivityOptions,
1067    ) {
1068        let (input, headers) = self.decoded.into_parts();
1069        (
1070            self.activity_type,
1071            input.expect("activity input must exist"),
1072            headers,
1073            self.options,
1074        )
1075    }
1076
1077    /// Activity type.
1078    pub fn activity_type(&self) -> &str {
1079        &self.activity_type
1080    }
1081
1082    /// Mutably access the activity type.
1083    pub fn activity_type_mut(&mut self) -> &mut String {
1084        &mut self.activity_type
1085    }
1086
1087    /// Activity options.
1088    pub fn options(&self) -> &ActivityOptions {
1089        &self.options
1090    }
1091
1092    /// Mutably access activity options.
1093    pub fn options_mut(&mut self) -> &mut ActivityOptions {
1094        &mut self.options
1095    }
1096}
1097
1098typed_outbound_input!(ScheduleActivityInput);
1099
1100/// Input passed to [`WorkflowInterceptor::schedule_local_activity`].
1101#[non_exhaustive]
1102pub struct ScheduleLocalActivityInput {
1103    activity_type: String,
1104    decoded: DecodedInput,
1105    options: LocalActivityOptions,
1106}
1107
1108impl ScheduleLocalActivityInput {
1109    pub(crate) fn new(
1110        activity_type: String,
1111        input: Box<dyn Any>,
1112        options: LocalActivityOptions,
1113    ) -> Self {
1114        Self {
1115            activity_type,
1116            decoded: DecodedInput::new(Some(input), HashMap::new()),
1117            options,
1118        }
1119    }
1120
1121    pub(crate) fn into_parts(
1122        self,
1123    ) -> (
1124        String,
1125        Box<dyn Any>,
1126        HashMap<String, Payload>,
1127        LocalActivityOptions,
1128    ) {
1129        let (input, headers) = self.decoded.into_parts();
1130        (
1131            self.activity_type,
1132            input.expect("local activity input must exist"),
1133            headers,
1134            self.options,
1135        )
1136    }
1137
1138    /// Activity type.
1139    pub fn activity_type(&self) -> &str {
1140        &self.activity_type
1141    }
1142
1143    /// Mutably access the activity type.
1144    pub fn activity_type_mut(&mut self) -> &mut String {
1145        &mut self.activity_type
1146    }
1147
1148    /// Local activity options.
1149    pub fn options(&self) -> &LocalActivityOptions {
1150        &self.options
1151    }
1152
1153    /// Mutably access local activity options.
1154    pub fn options_mut(&mut self) -> &mut LocalActivityOptions {
1155        &mut self.options
1156    }
1157}
1158
1159typed_outbound_input!(ScheduleLocalActivityInput);
1160
1161/// Input passed to [`WorkflowInterceptor::start_child_workflow`].
1162#[non_exhaustive]
1163pub struct StartChildWorkflowInput {
1164    workflow_type: String,
1165    decoded: DecodedInput,
1166    options: ChildWorkflowOptions,
1167}
1168
1169impl StartChildWorkflowInput {
1170    pub(crate) fn new(
1171        workflow_type: String,
1172        input: Box<dyn Any>,
1173        options: ChildWorkflowOptions,
1174    ) -> Self {
1175        Self {
1176            workflow_type,
1177            decoded: DecodedInput::new(Some(input), HashMap::new()),
1178            options,
1179        }
1180    }
1181
1182    pub(crate) fn into_parts(
1183        self,
1184    ) -> (
1185        String,
1186        Box<dyn Any>,
1187        HashMap<String, Payload>,
1188        ChildWorkflowOptions,
1189    ) {
1190        let (input, headers) = self.decoded.into_parts();
1191        (
1192            self.workflow_type,
1193            input.expect("child workflow input must exist"),
1194            headers,
1195            self.options,
1196        )
1197    }
1198
1199    /// Workflow type.
1200    pub fn workflow_type(&self) -> &str {
1201        &self.workflow_type
1202    }
1203
1204    /// Mutably access the workflow type.
1205    pub fn workflow_type_mut(&mut self) -> &mut String {
1206        &mut self.workflow_type
1207    }
1208
1209    /// Child workflow options.
1210    pub fn options(&self) -> &ChildWorkflowOptions {
1211        &self.options
1212    }
1213
1214    /// Mutably access child workflow options.
1215    pub fn options_mut(&mut self) -> &mut ChildWorkflowOptions {
1216        &mut self.options
1217    }
1218}
1219
1220typed_outbound_input!(StartChildWorkflowInput);
1221
1222/// Workflow targeted by an outbound signal.
1223#[derive(Clone, Debug, PartialEq, Eq)]
1224#[non_exhaustive]
1225pub enum SignalWorkflowTarget {
1226    /// A child workflow identified by workflow ID.
1227    Child {
1228        /// Child workflow ID.
1229        workflow_id: String,
1230    },
1231    /// An external workflow execution.
1232    External {
1233        /// Target namespace.
1234        namespace: String,
1235        /// Target workflow ID.
1236        workflow_id: String,
1237        /// Target run ID, or the latest run when absent.
1238        run_id: Option<String>,
1239    },
1240}
1241
1242/// Input passed to [`WorkflowInterceptor::signal_workflow`].
1243#[non_exhaustive]
1244pub struct SignalWorkflowInput {
1245    signal_name: String,
1246    target: SignalWorkflowTarget,
1247    decoded: DecodedInput,
1248    options: SignalWorkflowOptions,
1249}
1250
1251impl SignalWorkflowInput {
1252    pub(crate) fn new(
1253        signal_name: String,
1254        target: SignalWorkflowTarget,
1255        input: Box<dyn Any>,
1256        options: SignalWorkflowOptions,
1257    ) -> Self {
1258        Self {
1259            signal_name,
1260            target,
1261            decoded: DecodedInput::new(Some(input), HashMap::new()),
1262            options,
1263        }
1264    }
1265
1266    #[allow(clippy::type_complexity)]
1267    pub(crate) fn into_parts(
1268        self,
1269    ) -> (
1270        String,
1271        SignalWorkflowTarget,
1272        Box<dyn Any>,
1273        HashMap<String, Payload>,
1274        SignalWorkflowOptions,
1275    ) {
1276        let (input, headers) = self.decoded.into_parts();
1277        (
1278            self.signal_name,
1279            self.target,
1280            input.expect("signal input must exist"),
1281            headers,
1282            self.options,
1283        )
1284    }
1285
1286    /// Signal name.
1287    pub fn signal_name(&self) -> &str {
1288        &self.signal_name
1289    }
1290
1291    /// Mutably access the signal name.
1292    pub fn signal_name_mut(&mut self) -> &mut String {
1293        &mut self.signal_name
1294    }
1295
1296    /// Signal target.
1297    pub fn target(&self) -> &SignalWorkflowTarget {
1298        &self.target
1299    }
1300
1301    /// Mutably access the signal target.
1302    pub fn target_mut(&mut self) -> &mut SignalWorkflowTarget {
1303        &mut self.target
1304    }
1305
1306    /// Signal options.
1307    pub fn options(&self) -> &SignalWorkflowOptions {
1308        &self.options
1309    }
1310
1311    /// Mutably access signal options.
1312    pub fn options_mut(&mut self) -> &mut SignalWorkflowOptions {
1313        &mut self.options
1314    }
1315}
1316
1317typed_outbound_input!(SignalWorkflowInput);
1318
1319/// Input passed to [`WorkflowInterceptor::cancel_external_workflow`].
1320#[derive(Clone, Debug)]
1321#[non_exhaustive]
1322pub struct CancelExternalWorkflowInput {
1323    /// Target workflow ID.
1324    pub workflow_id: String,
1325    /// Target run ID, or the latest run when absent.
1326    pub run_id: Option<String>,
1327    /// Cancellation reason.
1328    pub reason: Option<String>,
1329}
1330
1331/// Input passed to [`WorkflowInterceptor::continue_as_new`].
1332#[non_exhaustive]
1333pub struct ContinueAsNewInput {
1334    decoded: DecodedInput,
1335    options: ContinueAsNewOptions,
1336}
1337
1338impl ContinueAsNewInput {
1339    pub(crate) fn new(input: Box<dyn Any>, options: ContinueAsNewOptions) -> Self {
1340        Self {
1341            decoded: DecodedInput::new(Some(input), HashMap::new()),
1342            options,
1343        }
1344    }
1345
1346    pub(crate) fn into_parts(
1347        self,
1348    ) -> (Box<dyn Any>, HashMap<String, Payload>, ContinueAsNewOptions) {
1349        let (input, headers) = self.decoded.into_parts();
1350        (
1351            input.expect("continue-as-new input must exist"),
1352            headers,
1353            self.options,
1354        )
1355    }
1356
1357    /// Continue-as-new options.
1358    pub fn options(&self) -> &ContinueAsNewOptions {
1359        &self.options
1360    }
1361
1362    /// Mutably access continue-as-new options.
1363    pub fn options_mut(&mut self) -> &mut ContinueAsNewOptions {
1364        &mut self.options
1365    }
1366}
1367
1368typed_outbound_input!(ContinueAsNewInput);
1369
1370/// Result of an intercepted activity call.
1371pub type ScheduleActivityResult = Result<Box<dyn WorkflowOutboundValue>, ActivityExecutionError>;
1372
1373/// Result of an intercepted child workflow completion.
1374pub type ChildWorkflowOutboundResult =
1375    Result<Box<dyn WorkflowOutboundValue>, ChildWorkflowExecutionError>;
1376
1377/// Result of an intercepted signal call.
1378pub type SignalWorkflowResult = Result<(), WorkflowSignalError>;
1379
1380/// Result of an intercepted child workflow start.
1381pub type StartChildWorkflowResult = Result<StartChildWorkflowOutput, ChildWorkflowStartError>;
1382
1383/// Result of an intercepted continue-as-new call.
1384pub type ContinueAsNewResult = Result<Infallible, WorkflowTermination>;
1385
1386/// Interceptor for calls into workflow code and commands issued by workflow code.
1387///
1388/// Implement this trait for behavior that should wrap workflow operations. Inbound
1389/// methods intercept operations such as workflow execution and handler dispatch;
1390/// outbound methods intercept timers, activities, child workflows, external workflow calls,
1391/// continue-as-new, and Nexus operations.
1392///
1393/// Implementations normally calls [`WorkflowNext::run`] exactly once. It may transform the input first,
1394/// then inspect or transform the result. Not calling `next` short-circuits the operation, so it
1395/// should only be done if intentionally skipping the operation.
1396///
1397/// The async inbound methods return [`WorkflowInterceptorFuture`]. Use
1398/// [`WorkflowInterceptorFuture::new`] to wrap an `async` block around the downstream future:
1399/// call `next.run(input).await` to call the next interceptor. See the [module-level guide](self)
1400/// for a complete example and the determinism requirements.
1401///
1402/// Interceptors run as workflow code and are recreated when an evicted workflow is rebuilt. Their
1403/// behavior and any instance-local state must remain deterministic under replay. Async methods may
1404/// await only workflow scheduler primitives or SDK-provided workflow futures.
1405///
1406/// A [`WorkflowInterceptorConstructor`] creates one interceptor for each in-memory workflow
1407/// instance. The same interceptor object handles inbound and outbound calls for that instance and
1408/// is recreated if the workflow is evicted and rebuilt.
1409/// Inbound interceptors are called in regsitration order and outbound interceptors are called in reverse order.
1410pub trait WorkflowInterceptor: 'static {
1411    /// Called to invoke the workflow's `#[init]` method.
1412    ///
1413    /// It is only called for workflows that define `#[init]`, before the workflow instance exists.
1414    fn initialize_workflow(
1415        &self,
1416        _ctx: WorkflowContextView,
1417        input: InitializeWorkflowInput,
1418        next: WorkflowNext<'_, InitializeWorkflowInput, InitializeWorkflowOutput>,
1419    ) -> InitializeWorkflowOutput {
1420        next.run(input)
1421    }
1422
1423    /// Called to invoke the workflow run method.
1424    ///
1425    /// Inputs consumed by `#[init]` are instead passed to
1426    /// [`WorkflowInterceptor::initialize_workflow`].
1427    fn execute<'a>(
1428        &'a self,
1429        _ctx: WorkflowInterceptorContext,
1430        input: ExecuteWorkflowInput,
1431        next: WorkflowNext<
1432            'a,
1433            ExecuteWorkflowInput,
1434            WorkflowInterceptorFuture<'a, ExecuteWorkflowResult>,
1435        >,
1436    ) -> WorkflowInterceptorFuture<'a, ExecuteWorkflowResult> {
1437        next.run(input)
1438    }
1439
1440    /// Called to invoke a signal handler.
1441    fn handle_signal<'a>(
1442        &'a self,
1443        _ctx: WorkflowInterceptorContext,
1444        input: HandleSignalInput,
1445        next: WorkflowNext<
1446            'a,
1447            HandleSignalInput,
1448            WorkflowInterceptorFuture<'a, HandleSignalResult>,
1449        >,
1450    ) -> WorkflowInterceptorFuture<'a, HandleSignalResult> {
1451        next.run(input)
1452    }
1453
1454    /// Called to invoke an update handler.
1455    fn handle_update<'a>(
1456        &'a self,
1457        _ctx: WorkflowInterceptorContext,
1458        input: HandleUpdateInput,
1459        next: WorkflowNext<
1460            'a,
1461            HandleUpdateInput,
1462            WorkflowInterceptorFuture<'a, HandleUpdateResult>,
1463        >,
1464    ) -> WorkflowInterceptorFuture<'a, HandleUpdateResult> {
1465        next.run(input)
1466    }
1467
1468    /// Called to invoke a query handler.
1469    fn handle_query(
1470        &self,
1471        _ctx: SyncWorkflowInterceptorContext,
1472        input: HandleQueryInput,
1473        next: WorkflowNext<'_, HandleQueryInput, HandleQueryResult>,
1474    ) -> HandleQueryResult {
1475        next.run(input)
1476    }
1477
1478    /// Called to validate an update.
1479    fn validate_update(
1480        &self,
1481        _ctx: SyncWorkflowInterceptorContext,
1482        input: ValidateUpdateInput,
1483        next: WorkflowNext<'_, ValidateUpdateInput, ValidateUpdateResult>,
1484    ) -> ValidateUpdateResult {
1485        next.run(input)
1486    }
1487
1488    /// Called when the workflow starts a timer.
1489    fn start_timer(
1490        &self,
1491        _ctx: WorkflowInterceptorContext,
1492        input: StartTimerInput,
1493        next: WorkflowNext<
1494            'static,
1495            StartTimerInput,
1496            CancellableWorkflowOutboundFuture<TimerResult>,
1497        >,
1498    ) -> CancellableWorkflowOutboundFuture<TimerResult> {
1499        next.run(input)
1500    }
1501
1502    /// Called when the workflow schedules an activity.
1503    fn schedule_activity(
1504        &self,
1505        _ctx: WorkflowInterceptorContext,
1506        input: ScheduleActivityInput,
1507        next: WorkflowNext<
1508            'static,
1509            ScheduleActivityInput,
1510            CancellableWorkflowOutboundFuture<ScheduleActivityResult>,
1511        >,
1512    ) -> CancellableWorkflowOutboundFuture<ScheduleActivityResult> {
1513        next.run(input)
1514    }
1515
1516    /// Called when the workflow schedules a local activity.
1517    fn schedule_local_activity(
1518        &self,
1519        _ctx: WorkflowInterceptorContext,
1520        input: ScheduleLocalActivityInput,
1521        next: WorkflowNext<
1522            'static,
1523            ScheduleLocalActivityInput,
1524            CancellableWorkflowOutboundFuture<ScheduleActivityResult>,
1525        >,
1526    ) -> CancellableWorkflowOutboundFuture<ScheduleActivityResult> {
1527        next.run(input)
1528    }
1529
1530    /// Called when the workflow starts a child workflow.
1531    fn start_child_workflow(
1532        &self,
1533        _ctx: WorkflowInterceptorContext,
1534        input: StartChildWorkflowInput,
1535        next: WorkflowNext<
1536            'static,
1537            StartChildWorkflowInput,
1538            CancellableWorkflowOutboundFuture<StartChildWorkflowResult>,
1539        >,
1540    ) -> CancellableWorkflowOutboundFuture<StartChildWorkflowResult> {
1541        next.run(input)
1542    }
1543
1544    /// Called when the workflow signals a child or external workflow.
1545    fn signal_workflow(
1546        &self,
1547        _ctx: WorkflowInterceptorContext,
1548        input: SignalWorkflowInput,
1549        next: WorkflowNext<
1550            'static,
1551            SignalWorkflowInput,
1552            CancellableWorkflowOutboundFuture<SignalWorkflowResult>,
1553        >,
1554    ) -> CancellableWorkflowOutboundFuture<SignalWorkflowResult> {
1555        next.run(input)
1556    }
1557
1558    /// Called when the workflow requests cancellation of an external workflow.
1559    fn cancel_external_workflow(
1560        &self,
1561        _ctx: WorkflowInterceptorContext,
1562        input: CancelExternalWorkflowInput,
1563        next: WorkflowNext<
1564            'static,
1565            CancelExternalWorkflowInput,
1566            WorkflowOutboundFuture<CancelExternalWfResult>,
1567        >,
1568    ) -> WorkflowOutboundFuture<CancelExternalWfResult> {
1569        next.run(input)
1570    }
1571
1572    /// Called when the workflow continues as new.
1573    fn continue_as_new(
1574        &self,
1575        _ctx: SyncWorkflowInterceptorContext,
1576        input: ContinueAsNewInput,
1577        next: WorkflowNext<'static, ContinueAsNewInput, ContinueAsNewResult>,
1578    ) -> ContinueAsNewResult {
1579        next.run(input)
1580    }
1581
1582    /// Called when the workflow starts a Nexus operation.
1583    #[cfg(feature = "experimental")]
1584    fn start_nexus_operation(
1585        &self,
1586        _ctx: WorkflowInterceptorContext,
1587        input: StartNexusOperationInput,
1588        next: WorkflowNext<
1589            'static,
1590            StartNexusOperationInput,
1591            CancellableWorkflowOutboundFuture<StartNexusOperationResult>,
1592        >,
1593    ) -> CancellableWorkflowOutboundFuture<StartNexusOperationResult> {
1594        next.run(input)
1595    }
1596}
1597
1598macro_rules! outbound_chain {
1599    ($fn_name:ident, $method:ident, $context:ty, $input:ty, $output:ty) => {
1600        pub(crate) fn $fn_name(
1601            interceptors: Rc<[Arc<dyn WorkflowInterceptor>]>,
1602            ctx: $context,
1603            input: $input,
1604            next: WorkflowNext<'static, $input, $output>,
1605        ) -> $output {
1606            fn call(
1607                interceptors: Rc<[Arc<dyn WorkflowInterceptor>]>,
1608                interceptor_count: usize,
1609                ctx: $context,
1610                input: $input,
1611                next: WorkflowNext<'static, $input, $output>,
1612            ) -> $output {
1613                if let Some(interceptor_index) = interceptor_count.checked_sub(1) {
1614                    let interceptor = interceptors[interceptor_index].clone();
1615                    let next_ctx = ctx.clone();
1616                    let downstream = WorkflowNext::new(move |input| {
1617                        call(interceptors, interceptor_index, next_ctx, input, next)
1618                    });
1619                    interceptor.$method(ctx, input, downstream)
1620                } else {
1621                    next.run(input)
1622                }
1623            }
1624
1625            let interceptor_count = interceptors.len();
1626            call(interceptors, interceptor_count, ctx, input, next)
1627        }
1628    };
1629}
1630
1631#[cfg(feature = "experimental")]
1632mod nexus;
1633
1634outbound_chain!(
1635    call_start_timer,
1636    start_timer,
1637    WorkflowInterceptorContext,
1638    StartTimerInput,
1639    CancellableWorkflowOutboundFuture<TimerResult>
1640);
1641outbound_chain!(
1642    call_schedule_activity,
1643    schedule_activity,
1644    WorkflowInterceptorContext,
1645    ScheduleActivityInput,
1646    CancellableWorkflowOutboundFuture<ScheduleActivityResult>
1647);
1648outbound_chain!(
1649    call_schedule_local_activity,
1650    schedule_local_activity,
1651    WorkflowInterceptorContext,
1652    ScheduleLocalActivityInput,
1653    CancellableWorkflowOutboundFuture<ScheduleActivityResult>
1654);
1655outbound_chain!(
1656    call_start_child_workflow,
1657    start_child_workflow,
1658    WorkflowInterceptorContext,
1659    StartChildWorkflowInput,
1660    CancellableWorkflowOutboundFuture<StartChildWorkflowResult>
1661);
1662outbound_chain!(
1663    call_signal_workflow,
1664    signal_workflow,
1665    WorkflowInterceptorContext,
1666    SignalWorkflowInput,
1667    CancellableWorkflowOutboundFuture<SignalWorkflowResult>
1668);
1669outbound_chain!(
1670    call_cancel_external_workflow,
1671    cancel_external_workflow,
1672    WorkflowInterceptorContext,
1673    CancelExternalWorkflowInput,
1674    WorkflowOutboundFuture<CancelExternalWfResult>
1675);
1676outbound_chain!(
1677    call_continue_as_new,
1678    continue_as_new,
1679    SyncWorkflowInterceptorContext,
1680    ContinueAsNewInput,
1681    ContinueAsNewResult
1682);
1683type WorkflowInterceptorConstructorFn =
1684    dyn Fn(&WorkflowContextView) -> Arc<dyn WorkflowInterceptor> + Send + Sync + 'static;
1685
1686/// Creates one interceptor for each in-memory workflow instance.
1687///
1688/// The constructor receives a read-only view of the workflow's initialization context. It may use
1689/// that context to initialize interceptor state but must remain deterministic because it runs again
1690/// when an evicted workflow is rebuilt.
1691#[derive(Clone)]
1692pub struct WorkflowInterceptorConstructor {
1693    constructor: Arc<WorkflowInterceptorConstructorFn>,
1694}
1695
1696impl WorkflowInterceptorConstructor {
1697    /// Create a workflow interceptor constructor.
1698    pub fn new<F, I>(constructor: F) -> Self
1699    where
1700        F: Fn(&WorkflowContextView) -> I + Send + Sync + 'static,
1701        I: WorkflowInterceptor,
1702    {
1703        Self {
1704            constructor: Arc::new(move |ctx| Arc::new(constructor(ctx))),
1705        }
1706    }
1707
1708    pub(crate) fn construct(&self, ctx: &WorkflowContextView) -> Arc<dyn WorkflowInterceptor> {
1709        (self.constructor)(ctx)
1710    }
1711}
1712
1713pub(crate) fn wrong_workflow_input_type(type_name: &'static str) -> WorkflowTermination {
1714    WorkflowTermination::failed_application(temporalio_common_wasm::error::ApplicationFailure::new(
1715        anyhow::anyhow!(
1716            "Workflow inbound interceptor returned arguments with wrong concrete type for workflow {type_name}"
1717        ),
1718    ))
1719}
1720
1721#[cfg(test)]
1722mod tests {
1723    use super::*;
1724    use std::cell::Cell;
1725
1726    fn cancellable_future<T: 'static>(
1727        future: impl Future<Output = T> + 'static,
1728        token: &WorkflowCancellationToken,
1729        cancellation_count: &Rc<Cell<usize>>,
1730    ) -> CancellableWorkflowOutboundFuture<T> {
1731        let cancellation_count = cancellation_count.clone();
1732        CancellableWorkflowOutboundFuture::new(
1733            future,
1734            WorkflowCancellationHandle::new(move |_| {
1735                cancellation_count.set(cancellation_count.get() + 1);
1736            }),
1737        )
1738        .with_cancellation_token(token.clone())
1739    }
1740
1741    #[test]
1742    fn pending_cancellable_future_observes_token_cancellation() {
1743        let token = WorkflowCancellationToken::new();
1744        let cancellation_count = Rc::new(Cell::new(0));
1745        let _future = cancellable_future(std::future::pending::<()>(), &token, &cancellation_count);
1746
1747        token.cancel();
1748
1749        assert_eq!(cancellation_count.get(), 1);
1750    }
1751
1752    #[test]
1753    fn completed_cancellable_future_unregisters_from_token() {
1754        let token = WorkflowCancellationToken::new();
1755        let cancellation_count = Rc::new(Cell::new(0));
1756        let future = cancellable_future(std::future::ready(()), &token, &cancellation_count);
1757
1758        assert_eq!(future.now_or_never(), Some(()));
1759        token.cancel();
1760
1761        assert_eq!(cancellation_count.get(), 0);
1762    }
1763
1764    #[test]
1765    fn mapped_cancellable_future_unregisters_from_token() {
1766        let token = WorkflowCancellationToken::new();
1767        let cancellation_count = Rc::new(Cell::new(0));
1768        let future = cancellable_future(std::future::ready(1), &token, &cancellation_count)
1769            .map(|value| value + 1);
1770
1771        assert_eq!(future.now_or_never(), Some(2));
1772        token.cancel();
1773
1774        assert_eq!(cancellation_count.get(), 0);
1775    }
1776
1777    #[test]
1778    fn shared_cancellable_future_unregisters_from_token() {
1779        let token = WorkflowCancellationToken::new();
1780        let cancellation_count = Rc::new(Cell::new(0));
1781        let future =
1782            cancellable_future(std::future::ready(1), &token, &cancellation_count).shared();
1783
1784        assert_eq!(future.clone().now_or_never(), Some(1));
1785        token.cancel();
1786
1787        assert_eq!(cancellation_count.get(), 0);
1788    }
1789}
temporalio_workflow/workflow_interceptors.rs

temporalio_workflow/
workflow_interceptors.rs
