temporalio_workflow/workflow_context/
options.rs

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

temporalio_workflow/workflow_context/
options.rs
