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    /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists
290    /// only for internal test purposes. This API *will* change.
291    #[cfg(feature = "experimental")]
292    #[doc(hidden)]
293    #[builder(default)]
294    pub event_group_markers: Vec<EventGroupMarker>,
295}
296
297impl ActivityOptions {
298    /// Returns a builder with `close_timeouts` set to [`ActivityCloseTimeouts::StartToClose`].
299    pub fn with_start_to_close_timeout(duration: Duration) -> ActivityOptionsBuilder {
300        Self::with_close_timeouts(ActivityCloseTimeouts::StartToClose(duration))
301    }
302
303    /// Returns a builder with `close_timeouts` set to [`ActivityCloseTimeouts::ScheduleToClose`].
304    pub fn with_schedule_to_close_timeout(duration: Duration) -> ActivityOptionsBuilder {
305        Self::with_close_timeouts(ActivityCloseTimeouts::ScheduleToClose(duration))
306    }
307
308    /// Creates activity options with only `start_to_close_timeout` set.
309    ///
310    /// If you need additional fields set, use [`Self::with_start_to_close_timeout`].
311    pub fn start_to_close_timeout(duration: Duration) -> Self {
312        Self::with_start_to_close_timeout(duration).build()
313    }
314
315    /// Creates activity options with only `schedule_to_close_timeout` set.
316    ///
317    /// If you need additional fields set, use [`Self::with_schedule_to_close_timeout`].
318    pub fn schedule_to_close_timeout(duration: Duration) -> Self {
319        Self::with_schedule_to_close_timeout(duration).build()
320    }
321}
322
323impl ActivityOptions {
324    pub(crate) fn into_command(
325        self,
326        seq: u32,
327        activity_type: String,
328        args: Vec<Payload>,
329        headers: HashMap<String, Payload>,
330    ) -> WorkflowCommand {
331        #[cfg(feature = "experimental")]
332        let event_group_markers = self.event_group_markers;
333        #[cfg(not(feature = "experimental"))]
334        let event_group_markers = Vec::new();
335        command_with_metadata(
336            workflow_command::Variant::ScheduleActivity(ScheduleActivity {
337                seq,
338                activity_type,
339                activity_id: self.activity_id.unwrap_or_else(|| seq.to_string()),
340                task_queue: self.task_queue.unwrap_or_default(),
341                arguments: args,
342                headers,
343                schedule_to_close_timeout: self
344                    .close_timeouts
345                    .schedule_to_close()
346                    .and_then(|duration| duration.try_into().ok()),
347                schedule_to_start_timeout: self
348                    .schedule_to_start_timeout
349                    .and_then(|duration| duration.try_into().ok()),
350                start_to_close_timeout: self
351                    .close_timeouts
352                    .start_to_close()
353                    .and_then(|duration| duration.try_into().ok()),
354                heartbeat_timeout: self
355                    .heartbeat_timeout
356                    .and_then(|duration| duration.try_into().ok()),
357                cancellation_type: ProtoActivityCancellationType::from(self.cancellation_type)
358                    .into(),
359                retry_policy: self.retry_policy.map(Into::into),
360                priority: self.priority.map(Into::into),
361                do_not_eagerly_execute: self.do_not_eagerly_execute,
362                ..Default::default()
363            }),
364            self.summary,
365            None,
366            event_group_markers,
367        )
368    }
369}
370
371/// Options for scheduling a local activity
372#[derive(Debug, Clone, bon::Builder)]
373#[non_exhaustive]
374pub struct LocalActivityOptions {
375    /// Identifier to use for tracking the activity in Workflow history.
376    /// The `activityId` can be accessed by the activity function.
377    /// Does not need to be unique.
378    ///
379    /// If `None` use the context's sequence number
380    pub activity_id: Option<String>,
381    /// Retry policy
382    #[builder(default)]
383    pub retry_policy: RetryPolicy,
384    /// Override attempt number rather than using 1.
385    /// Ideally we would not expose this in a released Rust SDK, but it's needed for test.
386    pub attempt: Option<u32>,
387    /// Override schedule time when doing timer backoff.
388    /// Ideally we would not expose this in a released Rust SDK, but it's needed for test.
389    pub original_schedule_time: Option<prost_types::Timestamp>,
390    /// Retry backoffs over this amount will use a timer rather than a local retry
391    pub timer_backoff_threshold: Option<Duration>,
392    /// How the activity will cancel
393    #[builder(default)]
394    pub cancel_type: ActivityCancellationType,
395    /// Whether to record the local activity's serialized arguments in the local activity marker.
396    ///
397    /// Enabling this makes the arguments visible in Workflow history and increases its size.
398    /// Defaults to `false`.
399    #[builder(default)]
400    pub include_arguments_in_marker: bool,
401    /// Cancellation token for this local activity. `None` inherits workflow cancellation.
402    pub cancellation_token: Option<WorkflowCancellationToken>,
403    /// Indicates how long the caller is willing to wait for local activity completion. Limits how
404    /// long retries will be attempted. When not specified defaults to the workflow execution
405    /// timeout (which may be unset).
406    pub schedule_to_close_timeout: Option<Duration>,
407    /// Limits time the local activity can idle internally before being executed. That can happen if
408    /// the worker is currently at max concurrent local activity executions. This timeout is always
409    /// non retryable as all a retry would achieve is to put it back into the same queue. Defaults
410    /// to `schedule_to_close_timeout` if not specified and that is set. Must be <=
411    /// `schedule_to_close_timeout` when set, if not, it will be clamped down.
412    pub schedule_to_start_timeout: Option<Duration>,
413    /// Maximum time the local activity is allowed to execute after the task is dispatched. This
414    /// timeout is always retryable. Either or both of `schedule_to_close_timeout` and this must be
415    /// specified. If set, this must be <= `schedule_to_close_timeout`, if not, it will be clamped
416    /// down.
417    pub start_to_close_timeout: Option<Duration>,
418    /// Single-line summary for this activity that will appear in UI/CLI.
419    pub summary: Option<String>,
420    /// Event group markers to attach to the resulting `RecordMarker` command.
421    ///
422    /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists
423    /// only for internal test purposes. This API *will* change.
424    #[cfg(feature = "experimental")]
425    #[doc(hidden)]
426    #[builder(default)]
427    pub event_group_markers: Vec<EventGroupMarker>,
428}
429
430impl Default for LocalActivityOptions {
431    fn default() -> Self {
432        Self::builder().build()
433    }
434}
435
436impl LocalActivityOptions {
437    pub(crate) fn into_command(
438        mut self,
439        seq: u32,
440        activity_type: String,
441        args: Vec<Payload>,
442        headers: HashMap<String, Payload>,
443    ) -> WorkflowCommand {
444        // Tests and some workflow code rely on the historical SDK behavior where omitted local
445        // activity timeouts are normalized before the command is emitted.
446        self.schedule_to_close_timeout
447            .get_or_insert(Duration::from_secs(100));
448        #[cfg(feature = "experimental")]
449        let event_group_markers = self.event_group_markers;
450        #[cfg(not(feature = "experimental"))]
451        let event_group_markers = Vec::new();
452        command_with_metadata(
453            workflow_command::Variant::ScheduleLocalActivity(ScheduleLocalActivity {
454                seq,
455                activity_type,
456                activity_id: self.activity_id.unwrap_or_else(|| seq.to_string()),
457                arguments: args,
458                headers,
459                retry_policy: Some(self.retry_policy.into()),
460                attempt: self.attempt.unwrap_or(1),
461                original_schedule_time: self.original_schedule_time,
462                local_retry_threshold: self
463                    .timer_backoff_threshold
464                    .and_then(|duration| duration.try_into().ok()),
465                cancellation_type: ProtoActivityCancellationType::from(self.cancel_type).into(),
466                include_arguments_in_marker: self.include_arguments_in_marker,
467                schedule_to_close_timeout: self
468                    .schedule_to_close_timeout
469                    .and_then(|duration| duration.try_into().ok()),
470                schedule_to_start_timeout: self
471                    .schedule_to_start_timeout
472                    .and_then(|duration| duration.try_into().ok()),
473                start_to_close_timeout: self
474                    .start_to_close_timeout
475                    .and_then(|duration| duration.try_into().ok()),
476            }),
477            self.summary,
478            None,
479            event_group_markers,
480        )
481    }
482}
483
484/// Options for scheduling a child workflow
485#[derive(Default, Debug, Clone, bon::Builder)]
486#[non_exhaustive]
487pub struct ChildWorkflowOptions {
488    /// Workflow ID. If unset or empty, the parent workflow generates a deterministic UUIDv4.
489    pub workflow_id: Option<String>,
490    /// Task queue to schedule the workflow in
491    ///
492    /// If `None`, use the same task queue as the parent workflow.
493    pub task_queue: Option<String>,
494    /// Cancellation strategy for the child workflow
495    #[builder(default)]
496    pub cancel_type: ChildWorkflowCancellationType,
497    /// Cancellation token for this child workflow. `None` inherits workflow cancellation.
498    pub cancellation_token: Option<WorkflowCancellationToken>,
499    /// How to respond to parent workflow ending
500    #[builder(default)]
501    pub parent_close_policy: ParentClosePolicy,
502    /// Static summary of the child workflow
503    pub static_summary: Option<String>,
504    /// Static details of the child workflow
505    pub static_details: Option<String>,
506    /// Set the policy for reusing the workflow id
507    #[builder(default)]
508    pub id_reuse_policy: WorkflowIdReusePolicy,
509    /// Optionally set the execution timeout for the workflow
510    pub execution_timeout: Option<Duration>,
511    /// Optionally indicates the default run timeout for a workflow run
512    pub run_timeout: Option<Duration>,
513    /// Optionally indicates the default task timeout for a workflow run
514    pub task_timeout: Option<Duration>,
515    /// Optionally set a cron schedule for the workflow
516    pub cron_schedule: Option<String>,
517    /// Additional search attributes to set on the child workflow.
518    pub search_attributes: Option<SearchAttributes>,
519    /// Priority for the workflow
520    pub priority: Option<Priority>,
521    /// Event group markers to attach to the resulting `StartChildWorkflowExecution` command.
522    ///
523    /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists
524    /// only for internal test purposes. This API *will* change.
525    #[cfg(feature = "experimental")]
526    #[doc(hidden)]
527    #[builder(default)]
528    pub event_group_markers: Vec<EventGroupMarker>,
529}
530
531impl ChildWorkflowOptions {
532    /// Construct a `ChildWorkflowOptions` with the specified `workflow_id`.
533    ///
534    /// Shorthand for `ChildWorkflowOptions::builder().workflow_id(Some(workflow_id)).build()`
535    pub fn workflow_id(workflow_id: String) -> Self {
536        Self::builder().workflow_id(workflow_id).build()
537    }
538
539    pub(crate) fn into_command(
540        self,
541        seq: u32,
542        workflow_type: String,
543        args: Vec<Payload>,
544        headers: HashMap<String, Payload>,
545        workflow_id: String,
546    ) -> WorkflowCommand {
547        #[cfg(feature = "experimental")]
548        let event_group_markers = self.event_group_markers;
549        #[cfg(not(feature = "experimental"))]
550        let event_group_markers = Vec::new();
551        command_with_metadata(
552            workflow_command::Variant::StartChildWorkflowExecution(StartChildWorkflowExecution {
553                seq,
554                workflow_type,
555                workflow_id,
556                task_queue: self.task_queue.unwrap_or_default(),
557                input: args,
558                headers,
559                cancellation_type: ProtoChildWorkflowCancellationType::from(self.cancel_type)
560                    .into(),
561                parent_close_policy: ProtoParentClosePolicy::from(self.parent_close_policy).into(),
562                workflow_id_reuse_policy: ProtoWorkflowIdReusePolicy::from(
563                    match self.id_reuse_policy {
564                        WorkflowIdReusePolicy::Unspecified => WorkflowIdReusePolicy::AllowDuplicate,
565                        policy => policy,
566                    },
567                )
568                .into(),
569                workflow_execution_timeout: self
570                    .execution_timeout
571                    .and_then(|duration| duration.try_into().ok()),
572                workflow_run_timeout: self
573                    .run_timeout
574                    .and_then(|duration| duration.try_into().ok()),
575                workflow_task_timeout: self
576                    .task_timeout
577                    .and_then(|duration| duration.try_into().ok()),
578                cron_schedule: self.cron_schedule.unwrap_or_default(),
579                search_attributes: self.search_attributes.map(|t| t.into_proto()),
580                priority: self.priority.map(Into::into),
581                ..Default::default()
582            }),
583            self.static_summary,
584            self.static_details,
585            event_group_markers,
586        )
587    }
588}
589
590/// Options for timer
591#[derive(Debug, Clone, bon::Builder)]
592#[non_exhaustive]
593pub struct TimerOptions {
594    /// Duration for the timer
595    #[builder(start_fn)]
596    pub duration: Duration,
597    /// Cancellation token for this timer. `None` inherits workflow cancellation.
598    pub cancellation_token: Option<WorkflowCancellationToken>,
599    /// Summary of the timer
600    pub summary: Option<String>,
601    /// Event group markers to attach to the resulting `StartTimer` command.
602    ///
603    /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists
604    /// only for internal test purposes. This API *will* change.
605    #[cfg(feature = "experimental")]
606    #[doc(hidden)]
607    #[builder(default)]
608    pub event_group_markers: Vec<EventGroupMarker>,
609}
610
611impl Default for TimerOptions {
612    fn default() -> Self {
613        Self::builder(Duration::default()).build()
614    }
615}
616
617impl From<Duration> for TimerOptions {
618    fn from(duration: Duration) -> Self {
619        TimerOptions {
620            duration,
621            ..Default::default()
622        }
623    }
624}
625
626impl TimerOptions {
627    pub(crate) fn into_command(self, seq: u32) -> WorkflowCommand {
628        #[cfg(feature = "experimental")]
629        let event_group_markers = self.event_group_markers;
630        #[cfg(not(feature = "experimental"))]
631        let event_group_markers = Vec::new();
632        command_with_metadata(
633            workflow_command::Variant::StartTimer(StartTimer {
634                seq,
635                start_to_fire_timeout: Some(
636                    self.duration
637                        .try_into()
638                        .expect("workflow timer timeout must fit into protobuf duration"),
639                ),
640            }),
641            self.summary,
642            None,
643            event_group_markers,
644        )
645    }
646}
647
648/// Options for waiting on a workflow condition.
649#[derive(Default, Debug, Clone, bon::Builder)]
650#[non_exhaustive]
651pub struct WaitConditionOptions {
652    /// Cancellation token for this wait. `None` inherits workflow cancellation.
653    pub cancellation_token: Option<WorkflowCancellationToken>,
654}
655
656/// Options for signaling a workflow from another workflow.
657#[derive(Default, Debug, Clone, bon::Builder)]
658#[non_exhaustive]
659pub struct SignalWorkflowOptions {
660    /// Cancellation token for this signal. `None` inherits workflow cancellation.
661    pub cancellation_token: Option<WorkflowCancellationToken>,
662    /// Single-line summary for this signal that will appear in UI/CLI.
663    pub summary: Option<String>,
664    /// Event group markers to attach to the resulting `SignalExternalWorkflowExecution` command.
665    ///
666    /// **Unstable:** Event Groups are not yet implemented in the Rust SDK; this field exists
667    /// only for internal test purposes. This API *will* change.
668    #[cfg(feature = "experimental")]
669    #[doc(hidden)]
670    #[builder(default)]
671    pub event_group_markers: Vec<EventGroupMarker>,
672}
673
674impl SignalWorkflowOptions {
675    pub(crate) fn into_command(
676        self,
677        seq: u32,
678        signal_name: String,
679        args: Vec<Payload>,
680        headers: HashMap<String, Payload>,
681        target: signal_external_workflow_execution::Target,
682    ) -> WorkflowCommand {
683        #[cfg(feature = "experimental")]
684        let event_group_markers = self.event_group_markers;
685        #[cfg(not(feature = "experimental"))]
686        let event_group_markers = Vec::new();
687        command_with_metadata(
688            workflow_command::Variant::SignalExternalWorkflowExecution(
689                SignalExternalWorkflowExecution {
690                    seq,
691                    signal_name,
692                    args,
693                    target: Some(target),
694                    headers,
695                },
696            ),
697            self.summary,
698            None,
699            event_group_markers,
700        )
701    }
702}
703
704/// Options for continuing a workflow as a new execution.
705///
706/// Unset fields inherit the current workflow's values where applicable.
707#[derive(Default, Debug, bon::Builder)]
708#[non_exhaustive]
709pub struct ContinueAsNewOptions {
710    /// Override the workflow type for the new execution. If `None`, reuses the current type.
711    pub workflow_type: Option<String>,
712    /// Task queue for the new execution. If `None`, reuses the current task queue.
713    pub task_queue: Option<String>,
714    /// Timeout for a single run of the new workflow.
715    pub run_timeout: Option<Duration>,
716    /// Timeout of a single workflow task.
717    pub task_timeout: Option<Duration>,
718    /// Delay before the first workflow task of the continued run is scheduled.
719    pub backoff_start_interval: Option<Duration>,
720    /// If set, the new workflow will have these memo values. If `None`, reuses the current memo.
721    pub memo: Option<MemoValues>,
722    /// If set, the new workflow will have these search attributes. If `None`, reuses the current
723    /// search attributes.
724    pub search_attributes: Option<SearchAttributes>,
725    /// If set, the new workflow will have this retry policy. If `None`, reuses the current policy.
726    #[builder(into)]
727    pub retry_policy: Option<RetryPolicy>,
728    /// Whether the new workflow should run on a worker with a compatible build id.
729    pub versioning_intent: Option<VersioningIntent>,
730    /// Versioning behavior to use for the first workflow task of the new run.
731    ///
732    /// This experimental option is only meaningful for workers using worker deployment
733    /// versioning. `AutoUpgrade` routes the new run to the current deployment version;
734    /// `UseRampingVersion` routes it to the ramping deployment version when one is configured.
735    #[cfg(feature = "experimental")]
736    pub initial_versioning_behavior: Option<ContinueAsNewVersioningBehavior>,
737}
738
739impl ContinueAsNewOptions {
740    pub(crate) fn into_request(
741        self,
742        workflow_type: String,
743        arguments: Vec<Payload>,
744        headers: HashMap<String, Payload>,
745        payload_converter: &PayloadConverter,
746    ) -> Result<ContinueAsNewRequest, PayloadConversionError> {
747        let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
748        let context = SerializationContext::new(&context_data, payload_converter);
749        let memo = self
750            .memo
751            .map(|memo| {
752                memo.iter()
753                    .map(|(key, value)| {
754                        payload_converter
755                            .to_payload(&context, value)
756                            .map(|payload| (key.to_owned(), payload))
757                    })
758                    .collect::<Result<HashMap<_, _>, _>>()
759            })
760            .transpose()?
761            .unwrap_or_default();
762        #[cfg(feature = "experimental")]
763        let initial_versioning_behavior = ProtoContinueAsNewVersioningBehavior::from(
764            self.initial_versioning_behavior
765                .unwrap_or(ContinueAsNewVersioningBehavior::Unspecified),
766        )
767        .into();
768        #[cfg(not(feature = "experimental"))]
769        let initial_versioning_behavior = Default::default();
770        Ok(ContinueAsNewWorkflowExecution {
771            workflow_type: self.workflow_type.unwrap_or(workflow_type),
772            task_queue: self.task_queue.unwrap_or_default(),
773            arguments,
774            workflow_run_timeout: self
775                .run_timeout
776                .and_then(|duration| duration.try_into().ok()),
777            workflow_task_timeout: self
778                .task_timeout
779                .and_then(|duration| duration.try_into().ok()),
780            backoff_start_interval: self
781                .backoff_start_interval
782                .and_then(|duration| duration.try_into().ok()),
783            memo,
784            headers,
785            search_attributes: self.search_attributes.map(|t| t.into_proto()),
786            retry_policy: self.retry_policy.map(Into::into),
787            versioning_intent: ProtoVersioningIntent::from(
788                self.versioning_intent
789                    .unwrap_or(VersioningIntent::Unspecified),
790            )
791            .into(),
792            initial_versioning_behavior,
793        })
794    }
795}
796
797fn command_with_metadata(
798    variant: workflow_command::Variant,
799    summary: Option<String>,
800    details: Option<String>,
801    markers: Vec<EventGroupMarker>,
802) -> WorkflowCommand {
803    WorkflowCommand {
804        variant: Some(variant),
805        user_metadata: string_user_metadata(summary, details),
806        event_group_markers: markers,
807    }
808}
809
810fn string_user_metadata(summary: Option<String>, details: Option<String>) -> Option<UserMetadata> {
811    if summary.is_none() && details.is_none() {
812        return None;
813    }
814    let converter = PayloadConverter::default();
815    let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
816    let context = SerializationContext::new(&context_data, &converter);
817    Some(UserMetadata {
818        summary: summary.map(|value| {
819            converter
820                .to_payload(&context, &value)
821                .expect("String-to-JSON payload serialization is infallible")
822        }),
823        details: details.map(|value| {
824            converter
825                .to_payload(&context, &value)
826                .expect("String-to-JSON payload serialization is infallible")
827        }),
828    })
829}
830
831#[cfg(test)]
832mod tests {
833    use super::*;
834
835    #[test]
836    fn activity_cancellation_default_preserves_sdk_behavior() {
837        assert_eq!(
838            ActivityCancellationType::default(),
839            ActivityCancellationType::TryCancel
840        );
841    }
842
843    #[test]
844    fn child_workflow_cancellation_defaults_to_wait_for_completion() {
845        assert_eq!(
846            ChildWorkflowOptions::default().cancel_type,
847            ChildWorkflowCancellationType::WaitCancellationCompleted
848        );
849        let command = ChildWorkflowOptions::default().into_command(
850            1,
851            "child".to_string(),
852            vec![],
853            HashMap::new(),
854            "child-id".to_string(),
855        );
856        let Some(workflow_command::Variant::StartChildWorkflowExecution(command)) = command.variant
857        else {
858            panic!("expected StartChildWorkflowExecution command");
859        };
860        assert_eq!(
861            command.cancellation_type,
862            ProtoChildWorkflowCancellationType::WaitCancellationCompleted as i32
863        );
864    }
865
866    #[test]
867    fn other_policy_defaults_preserve_sdk_behavior() {
868        assert_eq!(ParentClosePolicy::default(), ParentClosePolicy::Unspecified);
869        assert_eq!(
870            WorkflowIdReusePolicy::default(),
871            WorkflowIdReusePolicy::Unspecified
872        );
873        assert_eq!(VersioningIntent::default(), VersioningIntent::Unspecified);
874    }
875
876    #[test]
877    fn continue_as_new_options_maps_backoff_start_interval_to_request() {
878        let req = ContinueAsNewOptions {
879            backoff_start_interval: Some(Duration::from_secs(7)),
880            versioning_intent: Some(VersioningIntent::Compatible),
881            ..Default::default()
882        }
883        .into_request(
884            "test-workflow".to_string(),
885            vec![],
886            HashMap::new(),
887            &PayloadConverter::default(),
888        )
889        .unwrap();
890
891        let backoff = req
892            .backoff_start_interval
893            .expect("backoff_start_interval should be set");
894        assert_eq!(backoff.seconds, 7);
895        assert_eq!(backoff.nanos, 0);
896        assert_eq!(
897            req.versioning_intent,
898            ProtoVersioningIntent::Compatible as i32
899        );
900    }
901
902    #[test]
903    fn activity_options_with_start_to_close_timeout_wrapper_supports_builder_chaining() {
904        let opts = ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5))
905            .heartbeat_timeout(Duration::from_secs(2))
906            .build();
907
908        assert_eq!(
909            opts.close_timeouts,
910            ActivityCloseTimeouts::StartToClose(Duration::from_secs(5))
911        );
912        assert_eq!(opts.heartbeat_timeout, Some(Duration::from_secs(2)));
913    }
914
915    #[test]
916    fn activity_options_with_schedule_to_close_timeout_wrapper_supports_builder_chaining() {
917        let opts = ActivityOptions::with_schedule_to_close_timeout(Duration::from_secs(5))
918            .heartbeat_timeout(Duration::from_secs(2))
919            .build();
920
921        assert_eq!(
922            opts.close_timeouts,
923            ActivityCloseTimeouts::ScheduleToClose(Duration::from_secs(5))
924        );
925        assert_eq!(opts.heartbeat_timeout, Some(Duration::from_secs(2)));
926    }
927
928    #[test]
929    fn activity_options_both_close_timeouts_map_to_command() {
930        let req =
931            ActivityOptions::with_close_timeouts(ActivityCloseTimeouts::ScheduleAndStartToClose {
932                start_to_close: Duration::from_secs(3),
933                schedule_to_close: Duration::from_secs(8),
934            })
935            .cancellation_type(ActivityCancellationType::Abandon)
936            .build()
937            .into_command(7, "test".to_string(), vec![], HashMap::new());
938        let Some(workflow_command::Variant::ScheduleActivity(req)) = req.variant else {
939            panic!("expected ScheduleActivity command");
940        };
941        assert_eq!(req.start_to_close_timeout.unwrap().seconds, 3);
942        assert_eq!(req.schedule_to_close_timeout.unwrap().seconds, 8);
943        assert_eq!(
944            req.cancellation_type,
945            ProtoActivityCancellationType::Abandon as i32
946        );
947    }
948
949    #[test]
950    fn local_activity_arguments_marker_option_maps_to_command() {
951        let default_command = LocalActivityOptions::default().into_command(
952            1,
953            "test".to_string(),
954            vec![],
955            HashMap::new(),
956        );
957        let enabled_command = LocalActivityOptions::builder()
958            .include_arguments_in_marker(true)
959            .build()
960            .into_command(1, "test".to_string(), vec![], HashMap::new());
961
962        let Some(workflow_command::Variant::ScheduleLocalActivity(default_command)) =
963            default_command.variant
964        else {
965            panic!("expected ScheduleLocalActivity command");
966        };
967        let Some(workflow_command::Variant::ScheduleLocalActivity(enabled_command)) =
968            enabled_command.variant
969        else {
970            panic!("expected ScheduleLocalActivity command");
971        };
972        assert!(!default_command.include_arguments_in_marker);
973        assert!(enabled_command.include_arguments_in_marker);
974    }
975
976    #[test]
977    fn child_workflow_run_timeout_uses_run_timeout_field() {
978        let opts = ChildWorkflowOptions {
979            workflow_id: Some("test-wf".to_string()),
980            cancel_type: ChildWorkflowCancellationType::WaitCancellationRequested,
981            parent_close_policy: ParentClosePolicy::RequestCancel,
982            id_reuse_policy: WorkflowIdReusePolicy::RejectDuplicate,
983            execution_timeout: Some(Duration::from_secs(60)),
984            run_timeout: Some(Duration::from_secs(10)),
985            ..Default::default()
986        };
987        let command = opts.into_command(
988            1,
989            "TestWorkflow".to_string(),
990            vec![],
991            HashMap::new(),
992            "test-wf".into(),
993        );
994        let Some(workflow_command::Variant::StartChildWorkflowExecution(req)) = command.variant
995        else {
996            panic!("expected StartChildWorkflowExecution command");
997        };
998        let exec_timeout = req.workflow_execution_timeout.unwrap();
999        let run_timeout = req.workflow_run_timeout.unwrap();
1000        assert_eq!(exec_timeout.seconds, 60);
1001        assert_eq!(run_timeout.seconds, 10);
1002        assert_eq!(
1003            req.cancellation_type,
1004            ProtoChildWorkflowCancellationType::WaitCancellationRequested as i32
1005        );
1006        assert_eq!(
1007            req.parent_close_policy,
1008            ProtoParentClosePolicy::RequestCancel as i32
1009        );
1010        assert_eq!(
1011            req.workflow_id_reuse_policy,
1012            ProtoWorkflowIdReusePolicy::RejectDuplicate as i32
1013        );
1014    }
1015
1016    #[test]
1017    fn child_workflow_run_timeout_none_when_unset() {
1018        let opts = ChildWorkflowOptions {
1019            workflow_id: Some("test-wf".to_string()),
1020            execution_timeout: Some(Duration::from_secs(60)),
1021            ..Default::default()
1022        };
1023        let command = opts.into_command(
1024            1,
1025            "TestWorkflow".to_string(),
1026            vec![],
1027            HashMap::new(),
1028            "test-wf".into(),
1029        );
1030        let Some(workflow_command::Variant::StartChildWorkflowExecution(req)) = command.variant
1031        else {
1032            panic!("expected StartChildWorkflowExecution command");
1033        };
1034        let exec_timeout = req.workflow_execution_timeout.unwrap();
1035        assert_eq!(exec_timeout.seconds, 60);
1036        assert!(req.workflow_run_timeout.is_none());
1037    }
1038}
temporalio_workflow/workflow_context/options.rs

temporalio_workflow/workflow_context/
options.rs
