temporalio_workflow/workflow_context/
options.rs

1use std::{collections::HashMap, time::Duration};
2
3use crate::{MemoValues, WorkflowCancellationToken, runtime::types::ContinueAsNewRequest};
4#[cfg(feature = "experimental")]
5use temporalio_common_wasm::protos::temporal::api::enums::v1::ContinueAsNewVersioningBehavior as ProtoContinueAsNewVersioningBehavior;
6use temporalio_common_wasm::{
7    ActivityCloseTimeouts, Priority, RetryPolicy,
8    data_converters::{
9        GenericPayloadConverter, PayloadConversionError, PayloadConverter, SerializationContext,
10        SerializationContextData, WorkflowSerializationContext,
11    },
12    protos::{
13        coresdk::{
14            child_workflow::{
15                ChildWorkflowCancellationType as ProtoChildWorkflowCancellationType,
16                ParentClosePolicy as ProtoParentClosePolicy,
17            },
18            common::VersioningIntent as ProtoVersioningIntent,
19            workflow_commands::{
20                ActivityCancellationType as ProtoActivityCancellationType,
21                ContinueAsNewWorkflowExecution, ScheduleActivity, ScheduleLocalActivity,
22                SignalExternalWorkflowExecution, StartChildWorkflowExecution, StartTimer,
23                WorkflowCommand, signal_external_workflow_execution, workflow_command,
24            },
25        },
26        temporal::api::{
27            common::v1::Payload,
28            enums::v1::WorkflowIdReusePolicy as ProtoWorkflowIdReusePolicy,
29            sdk::v1::{EventGroupMarker, UserMetadata},
30        },
31    },
32    search_attributes::SearchAttributes,
33};
34
35#[cfg(feature = "experimental")]
36mod continue_as_new_versioning;
37#[cfg(feature = "experimental")]
38mod nexus;
39
40#[cfg(feature = "experimental")]
41pub use continue_as_new_versioning::ContinueAsNewVersioningBehavior;
42#[cfg(feature = "experimental")]
43pub use nexus::{NexusOperationCancellationType, NexusOperationOptions};
44
45/// Controls when activity cancellation is reported back to a workflow.
46#[derive(
47    Debug, Clone, Copy, PartialEq, Eq, Hash, Default, serde::Serialize, serde::Deserialize,
48)]
49#[non_exhaustive]
50pub enum ActivityCancellationType {
51    /// Request cancellation and report it immediately.
52    #[default]
53    TryCancel,
54    /// Wait until cancellation has completed.
55    WaitCancellationCompleted,
56    /// Do not request cancellation.
57    Abandon,
58}
59
60impl From<ActivityCancellationType> for ProtoActivityCancellationType {
61    fn from(value: ActivityCancellationType) -> Self {
62        match value {
63            ActivityCancellationType::TryCancel => Self::TryCancel,
64            ActivityCancellationType::WaitCancellationCompleted => Self::WaitCancellationCompleted,
65            ActivityCancellationType::Abandon => Self::Abandon,
66        }
67    }
68}
69
70impl From<ProtoActivityCancellationType> for ActivityCancellationType {
71    fn from(value: ProtoActivityCancellationType) -> Self {
72        match value {
73            ProtoActivityCancellationType::TryCancel => Self::TryCancel,
74            ProtoActivityCancellationType::WaitCancellationCompleted => {
75                Self::WaitCancellationCompleted
76            }
77            ProtoActivityCancellationType::Abandon => Self::Abandon,
78        }
79    }
80}
81
82/// Controls when child-workflow cancellation is reported to its parent.
83#[derive(
84    Debug, Clone, Copy, PartialEq, Eq, Hash, Default, serde::Serialize, serde::Deserialize,
85)]
86#[non_exhaustive]
87pub enum ChildWorkflowCancellationType {
88    /// Do not request cancellation.
89    Abandon,
90    /// Request cancellation and report it immediately.
91    TryCancel,
92    /// Wait until cancellation has completed.
93    #[default]
94    WaitCancellationCompleted,
95    /// Wait until the cancellation request is acknowledged.
96    WaitCancellationRequested,
97}
98
99impl From<ChildWorkflowCancellationType> for ProtoChildWorkflowCancellationType {
100    fn from(value: ChildWorkflowCancellationType) -> Self {
101        match value {
102            ChildWorkflowCancellationType::Abandon => Self::Abandon,
103            ChildWorkflowCancellationType::TryCancel => Self::TryCancel,
104            ChildWorkflowCancellationType::WaitCancellationCompleted => {
105                Self::WaitCancellationCompleted
106            }
107            ChildWorkflowCancellationType::WaitCancellationRequested => {
108                Self::WaitCancellationRequested
109            }
110        }
111    }
112}
113
114impl From<ProtoChildWorkflowCancellationType> for ChildWorkflowCancellationType {
115    fn from(value: ProtoChildWorkflowCancellationType) -> Self {
116        match value {
117            ProtoChildWorkflowCancellationType::Abandon => Self::Abandon,
118            ProtoChildWorkflowCancellationType::TryCancel => Self::TryCancel,
119            ProtoChildWorkflowCancellationType::WaitCancellationCompleted => {
120                Self::WaitCancellationCompleted
121            }
122            ProtoChildWorkflowCancellationType::WaitCancellationRequested => {
123                Self::WaitCancellationRequested
124            }
125        }
126    }
127}
128
129/// Controls what happens to a child workflow when its parent closes.
130#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
131#[non_exhaustive]
132pub enum ParentClosePolicy {
133    /// Let the server choose its default.
134    #[default]
135    Unspecified,
136    /// Terminate the child workflow.
137    Terminate,
138    /// Leave the child workflow running.
139    Abandon,
140    /// Request cancellation of the child workflow.
141    RequestCancel,
142}
143
144impl From<ParentClosePolicy> for ProtoParentClosePolicy {
145    fn from(value: ParentClosePolicy) -> Self {
146        match value {
147            ParentClosePolicy::Unspecified => Self::Unspecified,
148            ParentClosePolicy::Terminate => Self::Terminate,
149            ParentClosePolicy::Abandon => Self::Abandon,
150            ParentClosePolicy::RequestCancel => Self::RequestCancel,
151        }
152    }
153}
154
155impl From<ProtoParentClosePolicy> for ParentClosePolicy {
156    fn from(value: ProtoParentClosePolicy) -> Self {
157        match value {
158            ProtoParentClosePolicy::Unspecified => Self::Unspecified,
159            ProtoParentClosePolicy::Terminate => Self::Terminate,
160            ProtoParentClosePolicy::Abandon => Self::Abandon,
161            ProtoParentClosePolicy::RequestCancel => Self::RequestCancel,
162        }
163    }
164}
165
166/// Controls whether a closed workflow ID may be reused.
167#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
168#[non_exhaustive]
169pub enum WorkflowIdReusePolicy {
170    /// Use the SDK default of allowing duplicate IDs.
171    #[default]
172    Unspecified,
173    /// Allow the workflow ID to be reused.
174    AllowDuplicate,
175    /// Allow reuse only when the previous execution failed.
176    AllowDuplicateFailedOnly,
177    /// Reject reuse of the workflow ID.
178    RejectDuplicate,
179    /// Terminate a running execution before reusing the ID.
180    TerminateIfRunning,
181}
182
183impl From<WorkflowIdReusePolicy> for ProtoWorkflowIdReusePolicy {
184    #[allow(deprecated)]
185    fn from(value: WorkflowIdReusePolicy) -> Self {
186        match value {
187            WorkflowIdReusePolicy::Unspecified => Self::Unspecified,
188            WorkflowIdReusePolicy::AllowDuplicate => Self::AllowDuplicate,
189            WorkflowIdReusePolicy::AllowDuplicateFailedOnly => Self::AllowDuplicateFailedOnly,
190            WorkflowIdReusePolicy::RejectDuplicate => Self::RejectDuplicate,
191            WorkflowIdReusePolicy::TerminateIfRunning => Self::TerminateIfRunning,
192        }
193    }
194}
195
196impl From<ProtoWorkflowIdReusePolicy> for WorkflowIdReusePolicy {
197    #[allow(deprecated)]
198    fn from(value: ProtoWorkflowIdReusePolicy) -> Self {
199        match value {
200            ProtoWorkflowIdReusePolicy::Unspecified => Self::Unspecified,
201            ProtoWorkflowIdReusePolicy::AllowDuplicate => Self::AllowDuplicate,
202            ProtoWorkflowIdReusePolicy::AllowDuplicateFailedOnly => Self::AllowDuplicateFailedOnly,
203            ProtoWorkflowIdReusePolicy::RejectDuplicate => Self::RejectDuplicate,
204            ProtoWorkflowIdReusePolicy::TerminateIfRunning => Self::TerminateIfRunning,
205        }
206    }
207}
208
209/// Selects the worker versioning behavior intended for a command.
210#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
211#[non_exhaustive]
212pub enum VersioningIntent {
213    /// Let Core choose the appropriate behavior.
214    #[default]
215    Unspecified,
216    /// Prefer a worker compatible with the current worker.
217    Compatible,
218    /// Use the target task queue's default worker version.
219    Default,
220}
221
222impl From<VersioningIntent> for ProtoVersioningIntent {
223    fn from(value: VersioningIntent) -> Self {
224        match value {
225            VersioningIntent::Unspecified => Self::Unspecified,
226            VersioningIntent::Compatible => Self::Compatible,
227            VersioningIntent::Default => Self::Default,
228        }
229    }
230}
231
232impl From<ProtoVersioningIntent> for VersioningIntent {
233    fn from(value: ProtoVersioningIntent) -> Self {
234        match value {
235            ProtoVersioningIntent::Unspecified => Self::Unspecified,
236            ProtoVersioningIntent::Compatible => Self::Compatible,
237            ProtoVersioningIntent::Default => Self::Default,
238        }
239    }
240}
241
242/// Options for scheduling an activity
243#[derive(Debug, bon::Builder, Clone)]
244#[non_exhaustive]
245#[builder(start_fn = with_close_timeouts, on(String, into), state_mod(vis = "pub"))]
246pub struct ActivityOptions {
247    /// Timeouts for activity completion.
248    ///
249    /// See [`ActivityCloseTimeouts`] for the meaning of each timeout variant.
250    #[builder(start_fn)]
251    pub close_timeouts: ActivityCloseTimeouts,
252    /// Identifier to use for tracking the activity in Workflow history.
253    /// The `activityId` can be accessed by the activity function.
254    /// Does not need to be unique.
255    ///
256    /// If `None` use the context's sequence number
257    pub activity_id: Option<String>,
258    /// Task queue to schedule the activity in
259    ///
260    /// If `None`, use the same task queue as the parent workflow.
261    pub task_queue: Option<String>,
262    /// Time that the Activity Task can stay in the Task Queue before it is picked up by a Worker.
263    /// Do not specify this timeout unless using host specific Task Queues for Activity Tasks are
264    /// being used for routing.
265    /// `schedule_to_start_timeout` is always non-retryable.
266    /// Retrying after this timeout doesn't make sense as it would just put the Activity Task back
267    /// into the same Task Queue.
268    pub schedule_to_start_timeout: Option<Duration>,
269    /// Heartbeat interval. Activity must heartbeat before this interval passes after a last
270    /// heartbeat or activity start.
271    pub heartbeat_timeout: Option<Duration>,
272    /// Determines what the SDK does when the Activity is cancelled.
273    #[builder(default, into)]
274    pub cancellation_type: ActivityCancellationType,
275    /// Cancellation token for this activity. `None` inherits workflow cancellation.
276    pub cancellation_token: Option<WorkflowCancellationToken>,
277    /// Activity retry policy
278    #[builder(into)]
279    pub retry_policy: Option<RetryPolicy>,
280    /// Summary of the activity
281    pub summary: Option<String>,
282    /// Priority for the activity
283    pub priority: Option<Priority>,
284    /// If true, disable eager execution for this activity
285    #[builder(default)]
286    pub do_not_eagerly_execute: bool,
287    /// Event group markers to attach to the resulting `ScheduleActivityTask` command.
288    ///
289    /// **Experimental:** Event Groups are not yet fully supported by the Rust SDK. This API may
290    /// change.
291    #[cfg(feature = "experimental")]
292    #[builder(default)]
293    pub event_group_markers: Vec<EventGroupMarker>,
294}
295
296impl ActivityOptions {
297    /// Returns a builder with `close_timeouts` set to [`ActivityCloseTimeouts::StartToClose`].
298    pub fn with_start_to_close_timeout(duration: Duration) -> ActivityOptionsBuilder {
299        Self::with_close_timeouts(ActivityCloseTimeouts::StartToClose(duration))
300    }
301
302    /// Returns a builder with `close_timeouts` set to [`ActivityCloseTimeouts::ScheduleToClose`].
303    pub fn with_schedule_to_close_timeout(duration: Duration) -> ActivityOptionsBuilder {
304        Self::with_close_timeouts(ActivityCloseTimeouts::ScheduleToClose(duration))
305    }
306
307    /// Creates activity options with only `start_to_close_timeout` set.
308    ///
309    /// If you need additional fields set, use [`Self::with_start_to_close_timeout`].
310    pub fn start_to_close_timeout(duration: Duration) -> Self {
311        Self::with_start_to_close_timeout(duration).build()
312    }
313
314    /// Creates activity options with only `schedule_to_close_timeout` set.
315    ///
316    /// If you need additional fields set, use [`Self::with_schedule_to_close_timeout`].
317    pub fn schedule_to_close_timeout(duration: Duration) -> Self {
318        Self::with_schedule_to_close_timeout(duration).build()
319    }
320}
321
322impl ActivityOptions {
323    pub(crate) fn into_command(
324        self,
325        seq: u32,
326        activity_type: String,
327        args: Vec<Payload>,
328        headers: HashMap<String, Payload>,
329    ) -> WorkflowCommand {
330        #[cfg(feature = "experimental")]
331        let event_group_markers = self.event_group_markers;
332        #[cfg(not(feature = "experimental"))]
333        let event_group_markers = Vec::new();
334        command_with_metadata(
335            workflow_command::Variant::ScheduleActivity(ScheduleActivity {
336                seq,
337                activity_type,
338                activity_id: self.activity_id.unwrap_or_else(|| seq.to_string()),
339                task_queue: self.task_queue.unwrap_or_default(),
340                arguments: args,
341                headers,
342                schedule_to_close_timeout: self
343                    .close_timeouts
344                    .schedule_to_close()
345                    .and_then(|duration| duration.try_into().ok()),
346                schedule_to_start_timeout: self
347                    .schedule_to_start_timeout
348                    .and_then(|duration| duration.try_into().ok()),
349                start_to_close_timeout: self
350                    .close_timeouts
351                    .start_to_close()
352                    .and_then(|duration| duration.try_into().ok()),
353                heartbeat_timeout: self
354                    .heartbeat_timeout
355                    .and_then(|duration| duration.try_into().ok()),
356                cancellation_type: ProtoActivityCancellationType::from(self.cancellation_type)
357                    .into(),
358                retry_policy: self.retry_policy.map(Into::into),
359                priority: self.priority.map(Into::into),
360                do_not_eagerly_execute: self.do_not_eagerly_execute,
361                ..Default::default()
362            }),
363            self.summary,
364            None,
365            event_group_markers,
366        )
367    }
368}
369
370/// Options for scheduling a local activity
371#[derive(Debug, Clone, bon::Builder)]
372#[non_exhaustive]
373pub struct LocalActivityOptions {
374    /// Identifier to use for tracking the activity in Workflow history.
375    /// The `activityId` can be accessed by the activity function.
376    /// Does not need to be unique.
377    ///
378    /// If `None` use the context's sequence number
379    pub activity_id: Option<String>,
380    /// Retry policy
381    #[builder(default)]
382    pub retry_policy: RetryPolicy,
383    /// Override attempt number rather than using 1.
384    /// Ideally we would not expose this in a released Rust SDK, but it's needed for test.
385    pub attempt: Option<u32>,
386    /// Override schedule time when doing timer backoff.
387    /// Ideally we would not expose this in a released Rust SDK, but it's needed for test.
388    pub original_schedule_time: Option<prost_types::Timestamp>,
389    /// Retry backoffs over this amount will use a timer rather than a local retry
390    pub timer_backoff_threshold: Option<Duration>,
391    /// How the activity will cancel
392    #[builder(default)]
393    pub cancel_type: ActivityCancellationType,
394    /// Whether to record the local activity's serialized arguments in the local activity marker.
395    ///
396    /// Enabling this makes the arguments visible in Workflow history and increases its size.
397    /// Defaults to `false`.
398    #[builder(default)]
399    pub include_arguments_in_marker: bool,
400    /// Cancellation token for this local activity. `None` inherits workflow cancellation.
401    pub cancellation_token: Option<WorkflowCancellationToken>,
402    /// Indicates how long the caller is willing to wait for local activity completion. Limits how
403    /// long retries will be attempted. When not specified defaults to the workflow execution
404    /// timeout (which may be unset).
405    pub schedule_to_close_timeout: Option<Duration>,
406    /// Limits time the local activity can idle internally before being executed. That can happen if
407    /// the worker is currently at max concurrent local activity executions. This timeout is always
408    /// non retryable as all a retry would achieve is to put it back into the same queue. Defaults
409    /// to `schedule_to_close_timeout` if not specified and that is set. Must be <=
410    /// `schedule_to_close_timeout` when set, if not, it will be clamped down.
411    pub schedule_to_start_timeout: Option<Duration>,
412    /// Maximum time the local activity is allowed to execute after the task is dispatched. This
413    /// timeout is always retryable. Either or both of `schedule_to_close_timeout` and this must be
414    /// specified. If set, this must be <= `schedule_to_close_timeout`, if not, it will be clamped
415    /// down.
416    pub start_to_close_timeout: Option<Duration>,
417    /// Single-line summary for this activity that will appear in UI/CLI.
418    pub summary: Option<String>,
419    /// Event group markers to attach to the resulting `RecordMarker` command.
420    ///
421    /// **Experimental:** Event Groups are not yet fully supported by the Rust SDK. This API may
422    /// change.
423    #[cfg(feature = "experimental")]
424    #[builder(default)]
425    pub event_group_markers: Vec<EventGroupMarker>,
426}
427
428impl Default for LocalActivityOptions {
429    fn default() -> Self {
430        Self::builder().build()
431    }
432}
433
434impl LocalActivityOptions {
435    pub(crate) fn into_command(
436        mut self,
437        seq: u32,
438        activity_type: String,
439        args: Vec<Payload>,
440        headers: HashMap<String, Payload>,
441    ) -> WorkflowCommand {
442        // Tests and some workflow code rely on the historical SDK behavior where omitted local
443        // activity timeouts are normalized before the command is emitted.
444        self.schedule_to_close_timeout
445            .get_or_insert(Duration::from_secs(100));
446        #[cfg(feature = "experimental")]
447        let event_group_markers = self.event_group_markers;
448        #[cfg(not(feature = "experimental"))]
449        let event_group_markers = Vec::new();
450        command_with_metadata(
451            workflow_command::Variant::ScheduleLocalActivity(ScheduleLocalActivity {
452                seq,
453                activity_type,
454                activity_id: self.activity_id.unwrap_or_else(|| seq.to_string()),
455                arguments: args,
456                headers,
457                retry_policy: Some(self.retry_policy.into()),
458                attempt: self.attempt.unwrap_or(1),
459                original_schedule_time: self.original_schedule_time,
460                local_retry_threshold: self
461                    .timer_backoff_threshold
462                    .and_then(|duration| duration.try_into().ok()),
463                cancellation_type: ProtoActivityCancellationType::from(self.cancel_type).into(),
464                include_arguments_in_marker: self.include_arguments_in_marker,
465                schedule_to_close_timeout: self
466                    .schedule_to_close_timeout
467                    .and_then(|duration| duration.try_into().ok()),
468                schedule_to_start_timeout: self
469                    .schedule_to_start_timeout
470                    .and_then(|duration| duration.try_into().ok()),
471                start_to_close_timeout: self
472                    .start_to_close_timeout
473                    .and_then(|duration| duration.try_into().ok()),
474            }),
475            self.summary,
476            None,
477            event_group_markers,
478        )
479    }
480}
481
482/// Options for scheduling a child workflow
483#[derive(Default, Debug, Clone, bon::Builder)]
484#[non_exhaustive]
485pub struct ChildWorkflowOptions {
486    /// Workflow ID. If unset or empty, the parent workflow generates a deterministic UUIDv4.
487    pub workflow_id: Option<String>,
488    /// Task queue to schedule the workflow in
489    ///
490    /// If `None`, use the same task queue as the parent workflow.
491    pub task_queue: Option<String>,
492    /// Cancellation strategy for the child workflow
493    #[builder(default)]
494    pub cancel_type: ChildWorkflowCancellationType,
495    /// Cancellation token for this child workflow. `None` inherits workflow cancellation.
496    pub cancellation_token: Option<WorkflowCancellationToken>,
497    /// How to respond to parent workflow ending
498    #[builder(default)]
499    pub parent_close_policy: ParentClosePolicy,
500    /// Static summary of the child workflow
501    pub static_summary: Option<String>,
502    /// Static details of the child workflow
503    pub static_details: Option<String>,
504    /// Set the policy for reusing the workflow id
505    #[builder(default)]
506    pub id_reuse_policy: WorkflowIdReusePolicy,
507    /// Optionally set the execution timeout for the workflow
508    pub execution_timeout: Option<Duration>,
509    /// Optionally indicates the default run timeout for a workflow run
510    pub run_timeout: Option<Duration>,
511    /// Optionally indicates the default task timeout for a workflow run
512    pub task_timeout: Option<Duration>,
513    /// Optionally set a cron schedule for the workflow
514    pub cron_schedule: Option<String>,
515    /// Additional search attributes to set on the child workflow.
516    pub search_attributes: Option<SearchAttributes>,
517    /// Priority for the workflow
518    pub priority: Option<Priority>,
519    /// Event group markers to attach to the resulting `StartChildWorkflowExecution` command.
520    ///
521    /// **Experimental:** Event Groups are not yet fully supported by the Rust SDK. This API may
522    /// change.
523    #[cfg(feature = "experimental")]
524    #[builder(default)]
525    pub event_group_markers: Vec<EventGroupMarker>,
526}
527
528impl ChildWorkflowOptions {
529    /// Construct a `ChildWorkflowOptions` with the specified `workflow_id`.
530    ///
531    /// Shorthand for `ChildWorkflowOptions::builder().workflow_id(Some(workflow_id)).build()`
532    pub fn workflow_id(workflow_id: String) -> Self {
533        Self::builder().workflow_id(workflow_id).build()
534    }
535
536    pub(crate) fn into_command(
537        self,
538        seq: u32,
539        workflow_type: String,
540        args: Vec<Payload>,
541        headers: HashMap<String, Payload>,
542        workflow_id: String,
543    ) -> WorkflowCommand {
544        #[cfg(feature = "experimental")]
545        let event_group_markers = self.event_group_markers;
546        #[cfg(not(feature = "experimental"))]
547        let event_group_markers = Vec::new();
548        command_with_metadata(
549            workflow_command::Variant::StartChildWorkflowExecution(StartChildWorkflowExecution {
550                seq,
551                workflow_type,
552                workflow_id,
553                task_queue: self.task_queue.unwrap_or_default(),
554                input: args,
555                headers,
556                cancellation_type: ProtoChildWorkflowCancellationType::from(self.cancel_type)
557                    .into(),
558                parent_close_policy: ProtoParentClosePolicy::from(self.parent_close_policy).into(),
559                workflow_id_reuse_policy: ProtoWorkflowIdReusePolicy::from(
560                    match self.id_reuse_policy {
561                        WorkflowIdReusePolicy::Unspecified => WorkflowIdReusePolicy::AllowDuplicate,
562                        policy => policy,
563                    },
564                )
565                .into(),
566                workflow_execution_timeout: self
567                    .execution_timeout
568                    .and_then(|duration| duration.try_into().ok()),
569                workflow_run_timeout: self
570                    .run_timeout
571                    .and_then(|duration| duration.try_into().ok()),
572                workflow_task_timeout: self
573                    .task_timeout
574                    .and_then(|duration| duration.try_into().ok()),
575                cron_schedule: self.cron_schedule.unwrap_or_default(),
576                search_attributes: self.search_attributes.map(|t| t.into_proto()),
577                priority: self.priority.map(Into::into),
578                ..Default::default()
579            }),
580            self.static_summary,
581            self.static_details,
582            event_group_markers,
583        )
584    }
585}
586
587/// Options for timer
588#[derive(Debug, Clone, bon::Builder)]
589#[non_exhaustive]
590pub struct TimerOptions {
591    /// Duration for the timer
592    #[builder(start_fn)]
593    pub duration: Duration,
594    /// Cancellation token for this timer. `None` inherits workflow cancellation.
595    pub cancellation_token: Option<WorkflowCancellationToken>,
596    /// Summary of the timer
597    pub summary: Option<String>,
598    /// Event group markers to attach to the resulting `StartTimer` command.
599    ///
600    /// **Experimental:** Event Groups are not yet fully supported by the Rust SDK. This API may
601    /// change.
602    #[cfg(feature = "experimental")]
603    #[builder(default)]
604    pub event_group_markers: Vec<EventGroupMarker>,
605}
606
607impl Default for TimerOptions {
608    fn default() -> Self {
609        Self::builder(Duration::default()).build()
610    }
611}
612
613impl From<Duration> for TimerOptions {
614    fn from(duration: Duration) -> Self {
615        TimerOptions {
616            duration,
617            ..Default::default()
618        }
619    }
620}
621
622impl TimerOptions {
623    pub(crate) fn into_command(self, seq: u32) -> WorkflowCommand {
624        #[cfg(feature = "experimental")]
625        let event_group_markers = self.event_group_markers;
626        #[cfg(not(feature = "experimental"))]
627        let event_group_markers = Vec::new();
628        command_with_metadata(
629            workflow_command::Variant::StartTimer(StartTimer {
630                seq,
631                start_to_fire_timeout: Some(
632                    self.duration
633                        .try_into()
634                        .expect("workflow timer timeout must fit into protobuf duration"),
635                ),
636            }),
637            self.summary,
638            None,
639            event_group_markers,
640        )
641    }
642}
643
644/// Options for waiting on a workflow condition.
645#[derive(Default, Debug, Clone, bon::Builder)]
646#[non_exhaustive]
647pub struct WaitConditionOptions {
648    /// Cancellation token for this wait. `None` inherits workflow cancellation.
649    pub cancellation_token: Option<WorkflowCancellationToken>,
650}
651
652/// Options for signaling a workflow from another workflow.
653#[derive(Default, Debug, Clone, bon::Builder)]
654#[non_exhaustive]
655pub struct SignalWorkflowOptions {
656    /// Cancellation token for this signal. `None` inherits workflow cancellation.
657    pub cancellation_token: Option<WorkflowCancellationToken>,
658    /// Single-line summary for this signal that will appear in UI/CLI.
659    pub summary: Option<String>,
660    /// Event group markers to attach to the resulting `SignalExternalWorkflowExecution` command.
661    ///
662    /// **Experimental:** Event Groups are not yet fully supported by the Rust SDK. This API may
663    /// change.
664    #[cfg(feature = "experimental")]
665    #[builder(default)]
666    pub event_group_markers: Vec<EventGroupMarker>,
667}
668
669impl SignalWorkflowOptions {
670    pub(crate) fn into_command(
671        self,
672        seq: u32,
673        signal_name: String,
674        args: Vec<Payload>,
675        headers: HashMap<String, Payload>,
676        target: signal_external_workflow_execution::Target,
677    ) -> WorkflowCommand {
678        #[cfg(feature = "experimental")]
679        let event_group_markers = self.event_group_markers;
680        #[cfg(not(feature = "experimental"))]
681        let event_group_markers = Vec::new();
682        command_with_metadata(
683            workflow_command::Variant::SignalExternalWorkflowExecution(
684                SignalExternalWorkflowExecution {
685                    seq,
686                    signal_name,
687                    args,
688                    target: Some(target),
689                    headers,
690                },
691            ),
692            self.summary,
693            None,
694            event_group_markers,
695        )
696    }
697}
698
699/// Options for continuing a workflow as a new execution.
700///
701/// Unset fields inherit the current workflow's values where applicable.
702#[derive(Default, Debug, bon::Builder)]
703#[non_exhaustive]
704pub struct ContinueAsNewOptions {
705    /// Override the workflow type for the new execution. If `None`, reuses the current type.
706    pub workflow_type: Option<String>,
707    /// Task queue for the new execution. If `None`, reuses the current task queue.
708    pub task_queue: Option<String>,
709    /// Timeout for a single run of the new workflow.
710    pub run_timeout: Option<Duration>,
711    /// Timeout of a single workflow task.
712    pub task_timeout: Option<Duration>,
713    /// Delay before the first workflow task of the continued run is scheduled.
714    pub backoff_start_interval: Option<Duration>,
715    /// If set, the new workflow will have these memo values. If `None`, reuses the current memo.
716    pub memo: Option<MemoValues>,
717    /// If set, the new workflow will have these search attributes. If `None`, reuses the current
718    /// search attributes.
719    pub search_attributes: Option<SearchAttributes>,
720    /// If set, the new workflow will have this retry policy. If `None`, reuses the current policy.
721    #[builder(into)]
722    pub retry_policy: Option<RetryPolicy>,
723    /// Whether the new workflow should run on a worker with a compatible build id.
724    pub versioning_intent: Option<VersioningIntent>,
725    /// Versioning behavior to use for the first workflow task of the new run.
726    ///
727    /// This experimental option is only meaningful for workers using worker deployment
728    /// versioning. `AutoUpgrade` routes the new run to the current deployment version;
729    /// `UseRampingVersion` routes it to the ramping deployment version when one is configured.
730    #[cfg(feature = "experimental")]
731    pub initial_versioning_behavior: Option<ContinueAsNewVersioningBehavior>,
732}
733
734impl ContinueAsNewOptions {
735    pub(crate) fn into_request(
736        self,
737        workflow_type: String,
738        arguments: Vec<Payload>,
739        headers: HashMap<String, Payload>,
740        payload_converter: &PayloadConverter,
741    ) -> Result<ContinueAsNewRequest, PayloadConversionError> {
742        let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
743        let context = SerializationContext::new(&context_data, payload_converter);
744        let memo = self
745            .memo
746            .map(|memo| {
747                memo.iter()
748                    .map(|(key, value)| {
749                        payload_converter
750                            .to_payload(&context, value)
751                            .map(|payload| (key.to_owned(), payload))
752                    })
753                    .collect::<Result<HashMap<_, _>, _>>()
754            })
755            .transpose()?
756            .unwrap_or_default();
757        #[cfg(feature = "experimental")]
758        let initial_versioning_behavior = ProtoContinueAsNewVersioningBehavior::from(
759            self.initial_versioning_behavior
760                .unwrap_or(ContinueAsNewVersioningBehavior::Unspecified),
761        )
762        .into();
763        #[cfg(not(feature = "experimental"))]
764        let initial_versioning_behavior = Default::default();
765        Ok(ContinueAsNewWorkflowExecution {
766            workflow_type: self.workflow_type.unwrap_or(workflow_type),
767            task_queue: self.task_queue.unwrap_or_default(),
768            arguments,
769            workflow_run_timeout: self
770                .run_timeout
771                .and_then(|duration| duration.try_into().ok()),
772            workflow_task_timeout: self
773                .task_timeout
774                .and_then(|duration| duration.try_into().ok()),
775            backoff_start_interval: self
776                .backoff_start_interval
777                .and_then(|duration| duration.try_into().ok()),
778            memo,
779            headers,
780            search_attributes: self.search_attributes.map(|t| t.into_proto()),
781            retry_policy: self.retry_policy.map(Into::into),
782            versioning_intent: ProtoVersioningIntent::from(
783                self.versioning_intent
784                    .unwrap_or(VersioningIntent::Unspecified),
785            )
786            .into(),
787            initial_versioning_behavior,
788        })
789    }
790}
791
792fn command_with_metadata(
793    variant: workflow_command::Variant,
794    summary: Option<String>,
795    details: Option<String>,
796    markers: Vec<EventGroupMarker>,
797) -> WorkflowCommand {
798    WorkflowCommand {
799        variant: Some(variant),
800        user_metadata: string_user_metadata(summary, details),
801        event_group_markers: markers,
802    }
803}
804
805fn string_user_metadata(summary: Option<String>, details: Option<String>) -> Option<UserMetadata> {
806    if summary.is_none() && details.is_none() {
807        return None;
808    }
809    let converter = PayloadConverter::default();
810    let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
811    let context = SerializationContext::new(&context_data, &converter);
812    Some(UserMetadata {
813        summary: summary.map(|value| {
814            converter
815                .to_payload(&context, &value)
816                .expect("String-to-JSON payload serialization is infallible")
817        }),
818        details: details.map(|value| {
819            converter
820                .to_payload(&context, &value)
821                .expect("String-to-JSON payload serialization is infallible")
822        }),
823    })
824}
825
826#[cfg(test)]
827mod tests {
828    use super::*;
829
830    #[test]
831    fn activity_cancellation_default_preserves_sdk_behavior() {
832        assert_eq!(
833            ActivityCancellationType::default(),
834            ActivityCancellationType::TryCancel
835        );
836    }
837
838    #[test]
839    fn child_workflow_cancellation_defaults_to_wait_for_completion() {
840        assert_eq!(
841            ChildWorkflowOptions::default().cancel_type,
842            ChildWorkflowCancellationType::WaitCancellationCompleted
843        );
844        let command = ChildWorkflowOptions::default().into_command(
845            1,
846            "child".to_string(),
847            vec![],
848            HashMap::new(),
849            "child-id".to_string(),
850        );
851        let Some(workflow_command::Variant::StartChildWorkflowExecution(command)) = command.variant
852        else {
853            panic!("expected StartChildWorkflowExecution command");
854        };
855        assert_eq!(
856            command.cancellation_type,
857            ProtoChildWorkflowCancellationType::WaitCancellationCompleted as i32
858        );
859    }
860
861    #[test]
862    fn other_policy_defaults_preserve_sdk_behavior() {
863        assert_eq!(ParentClosePolicy::default(), ParentClosePolicy::Unspecified);
864        assert_eq!(
865            WorkflowIdReusePolicy::default(),
866            WorkflowIdReusePolicy::Unspecified
867        );
868        assert_eq!(VersioningIntent::default(), VersioningIntent::Unspecified);
869    }
870
871    #[test]
872    fn continue_as_new_options_maps_backoff_start_interval_to_request() {
873        let req = ContinueAsNewOptions {
874            backoff_start_interval: Some(Duration::from_secs(7)),
875            versioning_intent: Some(VersioningIntent::Compatible),
876            ..Default::default()
877        }
878        .into_request(
879            "test-workflow".to_string(),
880            vec![],
881            HashMap::new(),
882            &PayloadConverter::default(),
883        )
884        .unwrap();
885
886        let backoff = req
887            .backoff_start_interval
888            .expect("backoff_start_interval should be set");
889        assert_eq!(backoff.seconds, 7);
890        assert_eq!(backoff.nanos, 0);
891        assert_eq!(
892            req.versioning_intent,
893            ProtoVersioningIntent::Compatible as i32
894        );
895    }
896
897    #[test]
898    fn activity_options_with_start_to_close_timeout_wrapper_supports_builder_chaining() {
899        let opts = ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5))
900            .heartbeat_timeout(Duration::from_secs(2))
901            .build();
902
903        assert_eq!(
904            opts.close_timeouts,
905            ActivityCloseTimeouts::StartToClose(Duration::from_secs(5))
906        );
907        assert_eq!(opts.heartbeat_timeout, Some(Duration::from_secs(2)));
908    }
909
910    #[test]
911    fn activity_options_with_schedule_to_close_timeout_wrapper_supports_builder_chaining() {
912        let opts = ActivityOptions::with_schedule_to_close_timeout(Duration::from_secs(5))
913            .heartbeat_timeout(Duration::from_secs(2))
914            .build();
915
916        assert_eq!(
917            opts.close_timeouts,
918            ActivityCloseTimeouts::ScheduleToClose(Duration::from_secs(5))
919        );
920        assert_eq!(opts.heartbeat_timeout, Some(Duration::from_secs(2)));
921    }
922
923    #[test]
924    fn activity_options_both_close_timeouts_map_to_command() {
925        let req =
926            ActivityOptions::with_close_timeouts(ActivityCloseTimeouts::ScheduleAndStartToClose {
927                start_to_close: Duration::from_secs(3),
928                schedule_to_close: Duration::from_secs(8),
929            })
930            .cancellation_type(ActivityCancellationType::Abandon)
931            .build()
932            .into_command(7, "test".to_string(), vec![], HashMap::new());
933        let Some(workflow_command::Variant::ScheduleActivity(req)) = req.variant else {
934            panic!("expected ScheduleActivity command");
935        };
936        assert_eq!(req.start_to_close_timeout.unwrap().seconds, 3);
937        assert_eq!(req.schedule_to_close_timeout.unwrap().seconds, 8);
938        assert_eq!(
939            req.cancellation_type,
940            ProtoActivityCancellationType::Abandon as i32
941        );
942    }
943
944    #[test]
945    fn local_activity_arguments_marker_option_maps_to_command() {
946        let default_command = LocalActivityOptions::default().into_command(
947            1,
948            "test".to_string(),
949            vec![],
950            HashMap::new(),
951        );
952        let enabled_command = LocalActivityOptions::builder()
953            .include_arguments_in_marker(true)
954            .build()
955            .into_command(1, "test".to_string(), vec![], HashMap::new());
956
957        let Some(workflow_command::Variant::ScheduleLocalActivity(default_command)) =
958            default_command.variant
959        else {
960            panic!("expected ScheduleLocalActivity command");
961        };
962        let Some(workflow_command::Variant::ScheduleLocalActivity(enabled_command)) =
963            enabled_command.variant
964        else {
965            panic!("expected ScheduleLocalActivity command");
966        };
967        assert!(!default_command.include_arguments_in_marker);
968        assert!(enabled_command.include_arguments_in_marker);
969    }
970
971    #[test]
972    fn child_workflow_run_timeout_uses_run_timeout_field() {
973        let opts = ChildWorkflowOptions {
974            workflow_id: Some("test-wf".to_string()),
975            cancel_type: ChildWorkflowCancellationType::WaitCancellationRequested,
976            parent_close_policy: ParentClosePolicy::RequestCancel,
977            id_reuse_policy: WorkflowIdReusePolicy::RejectDuplicate,
978            execution_timeout: Some(Duration::from_secs(60)),
979            run_timeout: Some(Duration::from_secs(10)),
980            ..Default::default()
981        };
982        let command = opts.into_command(
983            1,
984            "TestWorkflow".to_string(),
985            vec![],
986            HashMap::new(),
987            "test-wf".into(),
988        );
989        let Some(workflow_command::Variant::StartChildWorkflowExecution(req)) = command.variant
990        else {
991            panic!("expected StartChildWorkflowExecution command");
992        };
993        let exec_timeout = req.workflow_execution_timeout.unwrap();
994        let run_timeout = req.workflow_run_timeout.unwrap();
995        assert_eq!(exec_timeout.seconds, 60);
996        assert_eq!(run_timeout.seconds, 10);
997        assert_eq!(
998            req.cancellation_type,
999            ProtoChildWorkflowCancellationType::WaitCancellationRequested as i32
1000        );
1001        assert_eq!(
1002            req.parent_close_policy,
1003            ProtoParentClosePolicy::RequestCancel as i32
1004        );
1005        assert_eq!(
1006            req.workflow_id_reuse_policy,
1007            ProtoWorkflowIdReusePolicy::RejectDuplicate as i32
1008        );
1009    }
1010
1011    #[test]
1012    fn child_workflow_run_timeout_none_when_unset() {
1013        let opts = ChildWorkflowOptions {
1014            workflow_id: Some("test-wf".to_string()),
1015            execution_timeout: Some(Duration::from_secs(60)),
1016            ..Default::default()
1017        };
1018        let command = opts.into_command(
1019            1,
1020            "TestWorkflow".to_string(),
1021            vec![],
1022            HashMap::new(),
1023            "test-wf".into(),
1024        );
1025        let Some(workflow_command::Variant::StartChildWorkflowExecution(req)) = command.variant
1026        else {
1027            panic!("expected StartChildWorkflowExecution command");
1028        };
1029        let exec_timeout = req.workflow_execution_timeout.unwrap();
1030        assert_eq!(exec_timeout.seconds, 60);
1031        assert!(req.workflow_run_timeout.is_none());
1032    }
1033}
temporalio_workflow/workflow_context/options.rs

temporalio_workflow/workflow_context/
options.rs
