temporalio_workflow/workflow_context/
options.rs

1use std::{collections::HashMap, time::Duration};
2
3use crate::{MemoValues, runtime::types::ContinueAsNewRequest};
4use temporalio_common_wasm::{
5    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_activation::SignalWorkflow,
19            workflow_commands::{
20                ActivityCancellationType as ProtoActivityCancellationType,
21                ContinueAsNewWorkflowExecution, ScheduleActivity, ScheduleLocalActivity,
22                ScheduleNexusOperation, StartChildWorkflowExecution, StartTimer, WorkflowCommand,
23                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::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    /// Activity retry policy
315    #[builder(into)]
316    pub retry_policy: Option<RetryPolicy>,
317    /// Summary of the activity
318    pub summary: Option<String>,
319    /// Priority for the activity
320    pub priority: Option<Priority>,
321    /// If true, disable eager execution for this activity
322    #[builder(default)]
323    pub do_not_eagerly_execute: bool,
324}
325
326impl ActivityOptions {
327    /// Returns a builder with `close_timeout` set to [`ActivityCloseTimeouts::StartToClose`].
328    pub fn with_start_to_close_timeout(duration: Duration) -> ActivityOptionsBuilder {
329        Self::with_close_timeouts(ActivityCloseTimeouts::StartToClose(duration))
330    }
331
332    /// Returns a builder with `close_timeout` set to [`ActivityCloseTimeouts::ScheduleToClose`].
333    pub fn with_schedule_to_close_timeout(duration: Duration) -> ActivityOptionsBuilder {
334        Self::with_close_timeouts(ActivityCloseTimeouts::ScheduleToClose(duration))
335    }
336
337    /// Creates activity options with only `start_to_close_timeout` set.
338    ///
339    /// If you need additional fields set, use [`Self::with_start_to_close_timeout`].
340    pub fn start_to_close_timeout(duration: Duration) -> Self {
341        Self::with_start_to_close_timeout(duration).build()
342    }
343
344    /// Creates activity options with only `schedule_to_close_timeout` set.
345    ///
346    /// If you need additional fields set, use [`Self::with_schedule_to_close_timeout`].
347    pub fn schedule_to_close_timeout(duration: Duration) -> Self {
348        Self::with_schedule_to_close_timeout(duration).build()
349    }
350}
351
352/// The timeouts applied to an activity's completion.
353#[derive(Debug, Clone, Copy, PartialEq, Eq)]
354pub enum ActivityCloseTimeouts {
355    /// Total time that a workflow is willing to wait for Activity to complete.
356    /// `ActivityCloseTimeouts::ScheduleToClose` limits the total time of an Activity's execution
357    /// including retries (use `ActivityCloseTimeouts::StartToClose` to limit the time of a single
358    /// attempt).
359    ScheduleToClose(Duration),
360    /// Maximum time of a single Activity execution attempt. Note that the Temporal Server doesn't
361    /// detect Worker process failures directly. It relies on this timeout to detect that an
362    /// Activity that didn't complete on time. So this timeout should be as short as the longest
363    /// possible execution of the Activity body. Potentially long running Activities must specify
364    /// `ActivityOptions::heartbeat_timeout` and heartbeat from the activity periodically for timely
365    /// failure detection.
366    StartToClose(Duration),
367    /// Applies both execution-attempt and overall-completion bounds.
368    Both {
369        /// Maximum time of a single Activity execution attempt.
370        start_to_close: Duration,
371        /// Total time that a workflow is willing to wait for Activity to complete.
372        schedule_to_close: Duration,
373    },
374}
375
376impl ActivityCloseTimeouts {
377    fn into_durations(self) -> (Option<Duration>, Option<Duration>) {
378        match self {
379            Self::ScheduleToClose(schedule_to_close) => (None, Some(schedule_to_close)),
380            Self::StartToClose(start_to_close) => (Some(start_to_close), None),
381            Self::Both {
382                start_to_close,
383                schedule_to_close,
384            } => (Some(start_to_close), Some(schedule_to_close)),
385        }
386    }
387}
388
389impl ActivityOptions {
390    pub(crate) fn into_command(
391        self,
392        seq: u32,
393        activity_type: String,
394        args: Vec<Payload>,
395        headers: HashMap<String, Payload>,
396    ) -> WorkflowCommand {
397        let (start_to_close_timeout, schedule_to_close_timeout) =
398            self.close_timeouts.into_durations();
399        command_with_metadata(
400            workflow_command::Variant::ScheduleActivity(ScheduleActivity {
401                seq,
402                activity_type,
403                activity_id: self.activity_id.unwrap_or_else(|| seq.to_string()),
404                task_queue: self.task_queue.unwrap_or_default(),
405                arguments: args,
406                headers,
407                schedule_to_close_timeout: schedule_to_close_timeout
408                    .and_then(|duration| duration.try_into().ok()),
409                schedule_to_start_timeout: self
410                    .schedule_to_start_timeout
411                    .and_then(|duration| duration.try_into().ok()),
412                start_to_close_timeout: start_to_close_timeout
413                    .and_then(|duration| duration.try_into().ok()),
414                heartbeat_timeout: self
415                    .heartbeat_timeout
416                    .and_then(|duration| duration.try_into().ok()),
417                cancellation_type: ProtoActivityCancellationType::from(self.cancellation_type)
418                    .into(),
419                retry_policy: self.retry_policy.map(Into::into),
420                priority: self.priority.map(Into::into),
421                do_not_eagerly_execute: self.do_not_eagerly_execute,
422                ..Default::default()
423            }),
424            self.summary,
425            None,
426        )
427    }
428}
429
430/// Options for scheduling a local activity
431#[derive(Default, Debug, Clone)]
432pub struct LocalActivityOptions {
433    /// Identifier to use for tracking the activity in Workflow history.
434    /// The `activityId` can be accessed by the activity function.
435    /// Does not need to be unique.
436    ///
437    /// If `None` use the context's sequence number
438    pub activity_id: Option<String>,
439    /// Retry policy
440    pub retry_policy: RetryPolicy,
441    /// Override attempt number rather than using 1.
442    /// Ideally we would not expose this in a released Rust SDK, but it's needed for test.
443    pub attempt: Option<u32>,
444    /// Override schedule time when doing timer backoff.
445    /// Ideally we would not expose this in a released Rust SDK, but it's needed for test.
446    pub original_schedule_time: Option<prost_types::Timestamp>,
447    /// Retry backoffs over this amount will use a timer rather than a local retry
448    pub timer_backoff_threshold: Option<Duration>,
449    /// How the activity will cancel
450    pub cancel_type: ActivityCancellationType,
451    /// Indicates how long the caller is willing to wait for local activity completion. Limits how
452    /// long retries will be attempted. When not specified defaults to the workflow execution
453    /// timeout (which may be unset).
454    pub schedule_to_close_timeout: Option<Duration>,
455    /// Limits time the local activity can idle internally before being executed. That can happen if
456    /// the worker is currently at max concurrent local activity executions. This timeout is always
457    /// non retryable as all a retry would achieve is to put it back into the same queue. Defaults
458    /// to `schedule_to_close_timeout` if not specified and that is set. Must be <=
459    /// `schedule_to_close_timeout` when set, if not, it will be clamped down.
460    pub schedule_to_start_timeout: Option<Duration>,
461    /// Maximum time the local activity is allowed to execute after the task is dispatched. This
462    /// timeout is always retryable. Either or both of `schedule_to_close_timeout` and this must be
463    /// specified. If set, this must be <= `schedule_to_close_timeout`, if not, it will be clamped
464    /// down.
465    pub start_to_close_timeout: Option<Duration>,
466    /// Single-line summary for this activity that will appear in UI/CLI.
467    pub summary: Option<String>,
468}
469
470impl LocalActivityOptions {
471    pub(crate) fn into_command(
472        mut self,
473        seq: u32,
474        activity_type: String,
475        args: Vec<Payload>,
476        headers: HashMap<String, Payload>,
477    ) -> WorkflowCommand {
478        // Tests and some workflow code rely on the historical SDK behavior where omitted local
479        // activity timeouts are normalized before the command is emitted.
480        self.schedule_to_close_timeout
481            .get_or_insert(Duration::from_secs(100));
482        command_with_metadata(
483            workflow_command::Variant::ScheduleLocalActivity(ScheduleLocalActivity {
484                seq,
485                activity_type,
486                activity_id: self.activity_id.unwrap_or_else(|| seq.to_string()),
487                arguments: args,
488                headers,
489                retry_policy: Some(self.retry_policy.into()),
490                attempt: self.attempt.unwrap_or(1),
491                original_schedule_time: self.original_schedule_time,
492                local_retry_threshold: self
493                    .timer_backoff_threshold
494                    .and_then(|duration| duration.try_into().ok()),
495                cancellation_type: ProtoActivityCancellationType::from(self.cancel_type).into(),
496                schedule_to_close_timeout: self
497                    .schedule_to_close_timeout
498                    .and_then(|duration| duration.try_into().ok()),
499                schedule_to_start_timeout: self
500                    .schedule_to_start_timeout
501                    .and_then(|duration| duration.try_into().ok()),
502                start_to_close_timeout: self
503                    .start_to_close_timeout
504                    .and_then(|duration| duration.try_into().ok()),
505            }),
506            self.summary,
507            None,
508        )
509    }
510}
511
512/// Options for scheduling a child workflow
513#[derive(Default, Debug, Clone, bon::Builder)]
514#[non_exhaustive]
515pub struct ChildWorkflowOptions {
516    /// Workflow ID. If unset or empty, the parent workflow generates a deterministic UUIDv4.
517    pub workflow_id: Option<String>,
518    /// Task queue to schedule the workflow in
519    ///
520    /// If `None`, use the same task queue as the parent workflow.
521    pub task_queue: Option<String>,
522    /// Cancellation strategy for the child workflow
523    #[builder(default)]
524    pub cancel_type: ChildWorkflowCancellationType,
525    /// How to respond to parent workflow ending
526    #[builder(default)]
527    pub parent_close_policy: ParentClosePolicy,
528    /// Static summary of the child workflow
529    pub static_summary: Option<String>,
530    /// Static details of the child workflow
531    pub static_details: Option<String>,
532    /// Set the policy for reusing the workflow id
533    #[builder(default)]
534    pub id_reuse_policy: WorkflowIdReusePolicy,
535    /// Optionally set the execution timeout for the workflow
536    pub execution_timeout: Option<Duration>,
537    /// Optionally indicates the default run timeout for a workflow run
538    pub run_timeout: Option<Duration>,
539    /// Optionally indicates the default task timeout for a workflow run
540    pub task_timeout: Option<Duration>,
541    /// Optionally set a cron schedule for the workflow
542    pub cron_schedule: Option<String>,
543    /// Additional search attributes to set on the child workflow.
544    pub search_attributes: Option<SearchAttributes>,
545    /// Priority for the workflow
546    pub priority: Option<Priority>,
547}
548
549impl ChildWorkflowOptions {
550    /// Construct a `ChildWorkflowOptions` with the specified `workflow_id`.
551    ///
552    /// Shorthand for `ChildWorkflowOptions::builder().workflow_id(Some(workflow_id)).build()`
553    pub fn workflow_id(workflow_id: String) -> Self {
554        Self::builder().workflow_id(workflow_id).build()
555    }
556
557    pub(crate) fn into_command(
558        self,
559        seq: u32,
560        workflow_type: String,
561        args: Vec<Payload>,
562        headers: HashMap<String, Payload>,
563        workflow_id: String,
564    ) -> WorkflowCommand {
565        command_with_metadata(
566            workflow_command::Variant::StartChildWorkflowExecution(StartChildWorkflowExecution {
567                seq,
568                workflow_type,
569                workflow_id,
570                task_queue: self.task_queue.unwrap_or_default(),
571                input: args,
572                headers,
573                cancellation_type: ProtoChildWorkflowCancellationType::from(self.cancel_type)
574                    .into(),
575                parent_close_policy: ProtoParentClosePolicy::from(self.parent_close_policy).into(),
576                workflow_id_reuse_policy: ProtoWorkflowIdReusePolicy::from(
577                    match self.id_reuse_policy {
578                        WorkflowIdReusePolicy::Unspecified => WorkflowIdReusePolicy::AllowDuplicate,
579                        policy => policy,
580                    },
581                )
582                .into(),
583                workflow_execution_timeout: self
584                    .execution_timeout
585                    .and_then(|duration| duration.try_into().ok()),
586                workflow_run_timeout: self
587                    .run_timeout
588                    .and_then(|duration| duration.try_into().ok()),
589                workflow_task_timeout: self
590                    .task_timeout
591                    .and_then(|duration| duration.try_into().ok()),
592                cron_schedule: self.cron_schedule.unwrap_or_default(),
593                search_attributes: self.search_attributes.map(|t| t.into_proto()),
594                priority: self.priority.map(Into::into),
595                ..Default::default()
596            }),
597            self.static_summary,
598            self.static_details,
599        )
600    }
601}
602
603/// Information needed to send a specific signal
604#[derive(Debug)]
605pub struct Signal {
606    /// The signal name
607    pub signal_name: String,
608    /// The data the signal carries
609    pub data: SignalData,
610}
611
612impl Signal {
613    /// Create a new signal
614    pub fn new(
615        name: impl Into<String>,
616        input: impl IntoIterator<Item = impl Into<Payload>>,
617    ) -> Self {
618        Self {
619            signal_name: name.into(),
620            data: SignalData::new(input),
621        }
622    }
623
624    pub(crate) fn into_invocation(self) -> SignalWorkflow {
625        SignalWorkflow {
626            signal_name: self.signal_name,
627            input: self.data.input,
628            identity: String::new(),
629            headers: self.data.headers,
630        }
631    }
632}
633
634/// Data contained within a signal
635#[derive(Default, Debug)]
636pub struct SignalData {
637    /// The arguments the signal will receive
638    pub input: Vec<Payload>,
639    /// Metadata attached to the signal
640    pub headers: HashMap<String, Payload>,
641}
642
643impl SignalData {
644    /// Create data for a signal
645    pub fn new(input: impl IntoIterator<Item = impl Into<Payload>>) -> Self {
646        Self {
647            input: input.into_iter().map(Into::into).collect(),
648            headers: HashMap::new(),
649        }
650    }
651
652    /// Set a header k/v pair attached to the signal
653    pub fn with_header(
654        &mut self,
655        key: impl Into<String>,
656        payload: impl Into<Payload>,
657    ) -> &mut Self {
658        self.headers.insert(key.into(), payload.into());
659        self
660    }
661}
662
663/// Options for timer
664#[derive(Default, Debug, Clone)]
665pub struct TimerOptions {
666    /// Duration for the timer
667    pub duration: Duration,
668    /// Summary of the timer
669    pub summary: Option<String>,
670}
671
672impl From<Duration> for TimerOptions {
673    fn from(duration: Duration) -> Self {
674        TimerOptions {
675            duration,
676            ..Default::default()
677        }
678    }
679}
680
681impl TimerOptions {
682    pub(crate) fn into_command(self, seq: u32) -> WorkflowCommand {
683        command_with_metadata(
684            workflow_command::Variant::StartTimer(StartTimer {
685                seq,
686                start_to_fire_timeout: Some(
687                    self.duration
688                        .try_into()
689                        .expect("workflow timer timeout must fit into protobuf duration"),
690                ),
691            }),
692            self.summary,
693            None,
694        )
695    }
696}
697
698/// Options for Nexus Operations
699#[derive(Default, Debug, Clone)]
700pub struct NexusOperationOptions {
701    /// Endpoint name, must exist in the endpoint registry or this command will fail.
702    pub endpoint: String,
703    /// Service name.
704    pub service: String,
705    /// Operation name.
706    pub operation: String,
707    /// Input for the operation. The server converts this into Nexus request content and the
708    /// appropriate content headers internally when sending the StartOperation request. On the
709    /// handler side, if it is also backed by Temporal, the content is transformed back to the
710    /// original Payload sent in this command.
711    pub input: Option<Payload>,
712    /// Schedule-to-close timeout for this operation.
713    /// Indicates how long the caller is willing to wait for operation completion.
714    /// Calls are retried internally by the server.
715    pub schedule_to_close_timeout: Option<Duration>,
716    /// Header to attach to the Nexus request.
717    /// Users are responsible for encrypting sensitive data in this header as it is stored in
718    /// workflow history and transmitted to external services as-is. This is useful for propagating
719    /// tracing information. Note these headers are not the same as Temporal headers on internal
720    /// activities and child workflows, these are transmitted to Nexus operations that may be
721    /// external and are not traditional payloads.
722    pub nexus_header: HashMap<String, String>,
723    /// Cancellation type for the operation
724    pub cancellation_type: Option<NexusOperationCancellationType>,
725    /// Schedule-to-start timeout for this operation.
726    /// Indicates how long the caller is willing to wait for the operation to be started (or completed if synchronous)
727    /// by the handler. If the operation is not started within this timeout, it will fail with
728    /// TIMEOUT_TYPE_SCHEDULE_TO_START.
729    /// If not set or zero, no schedule-to-start timeout is enforced.
730    pub schedule_to_start_timeout: Option<Duration>,
731    /// Start-to-close timeout for this operation.
732    /// Indicates how long the caller is willing to wait for an asynchronous operation to complete after it has been
733    /// started. If the operation does not complete within this timeout after starting, it will fail with
734    /// TIMEOUT_TYPE_START_TO_CLOSE.
735    /// Only applies to asynchronous operations. Synchronous operations ignore this timeout.
736    /// If not set or zero, no start-to-close timeout is enforced.
737    pub start_to_close_timeout: Option<Duration>,
738}
739
740impl NexusOperationOptions {
741    pub(crate) fn into_command(self, seq: u32) -> WorkflowCommand {
742        workflow_command::Variant::ScheduleNexusOperation(ScheduleNexusOperation {
743            seq,
744            endpoint: self.endpoint,
745            service: self.service,
746            operation: self.operation,
747            input: self.input,
748            schedule_to_close_timeout: self
749                .schedule_to_close_timeout
750                .and_then(|duration| duration.try_into().ok()),
751            schedule_to_start_timeout: self
752                .schedule_to_start_timeout
753                .and_then(|duration| duration.try_into().ok()),
754            start_to_close_timeout: self
755                .start_to_close_timeout
756                .and_then(|duration| duration.try_into().ok()),
757            nexus_header: self.nexus_header,
758            cancellation_type: ProtoNexusOperationCancellationType::from(
759                self.cancellation_type
760                    .unwrap_or(NexusOperationCancellationType::WaitCancellationCompleted),
761            )
762            .into(),
763        })
764        .into()
765    }
766}
767
768/// Versioning behavior to use for the first workflow task of a new continue-as-new run.
769#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
770#[non_exhaustive]
771pub enum ContinueAsNewVersioningBehavior {
772    /// No initial versioning behavior was specified.
773    #[default]
774    Unspecified,
775    /// Start the new run with AutoUpgrade behavior.
776    AutoUpgrade,
777    /// Start the new run on the task queue's ramping deployment version.
778    UseRampingVersion,
779}
780
781impl From<ContinueAsNewVersioningBehavior> for ProtoContinueAsNewVersioningBehavior {
782    fn from(value: ContinueAsNewVersioningBehavior) -> Self {
783        match value {
784            ContinueAsNewVersioningBehavior::Unspecified => {
785                ProtoContinueAsNewVersioningBehavior::Unspecified
786            }
787            ContinueAsNewVersioningBehavior::AutoUpgrade => {
788                ProtoContinueAsNewVersioningBehavior::AutoUpgrade
789            }
790            ContinueAsNewVersioningBehavior::UseRampingVersion => {
791                ProtoContinueAsNewVersioningBehavior::UseRampingVersion
792            }
793        }
794    }
795}
796
797impl From<ProtoContinueAsNewVersioningBehavior> for ContinueAsNewVersioningBehavior {
798    fn from(value: ProtoContinueAsNewVersioningBehavior) -> Self {
799        match value {
800            ProtoContinueAsNewVersioningBehavior::Unspecified => {
801                ContinueAsNewVersioningBehavior::Unspecified
802            }
803            ProtoContinueAsNewVersioningBehavior::AutoUpgrade => {
804                ContinueAsNewVersioningBehavior::AutoUpgrade
805            }
806            ProtoContinueAsNewVersioningBehavior::UseRampingVersion => {
807                ContinueAsNewVersioningBehavior::UseRampingVersion
808            }
809        }
810    }
811}
812
813/// Options for continuing a workflow as a new execution.
814///
815/// Unset fields inherit the current workflow's values where applicable.
816#[derive(Default, Debug, bon::Builder)]
817#[non_exhaustive]
818pub struct ContinueAsNewOptions {
819    /// Override the workflow type for the new execution. If `None`, reuses the current type.
820    pub workflow_type: Option<String>,
821    /// Task queue for the new execution. If `None`, reuses the current task queue.
822    pub task_queue: Option<String>,
823    /// Timeout for a single run of the new workflow.
824    pub run_timeout: Option<Duration>,
825    /// Timeout of a single workflow task.
826    pub task_timeout: Option<Duration>,
827    /// Delay before the first workflow task of the continued run is scheduled.
828    pub backoff_start_interval: Option<Duration>,
829    /// If set, the new workflow will have these memo values. If `None`, reuses the current memo.
830    pub memo: Option<MemoValues>,
831    /// If set, the new workflow will have these search attributes. If `None`, reuses the current
832    /// search attributes.
833    pub search_attributes: Option<SearchAttributes>,
834    /// If set, the new workflow will have this retry policy. If `None`, reuses the current policy.
835    #[builder(into)]
836    pub retry_policy: Option<RetryPolicy>,
837    /// Whether the new workflow should run on a worker with a compatible build id.
838    pub versioning_intent: Option<VersioningIntent>,
839    /// Versioning behavior to use for the first workflow task of the new run.
840    ///
841    /// This experimental option is only meaningful for workers using worker deployment
842    /// versioning. `AutoUpgrade` routes the new run to the current deployment version;
843    /// `UseRampingVersion` routes it to the ramping deployment version when one is configured.
844    pub initial_versioning_behavior: Option<ContinueAsNewVersioningBehavior>,
845}
846
847impl ContinueAsNewOptions {
848    pub(crate) fn into_request(
849        self,
850        workflow_type: String,
851        arguments: Vec<Payload>,
852        headers: HashMap<String, Payload>,
853        payload_converter: &PayloadConverter,
854    ) -> Result<ContinueAsNewRequest, PayloadConversionError> {
855        let memo = self
856            .memo
857            .map(|memo| memo.encode(payload_converter))
858            .transpose()?
859            .unwrap_or_default();
860        Ok(ContinueAsNewWorkflowExecution {
861            workflow_type: self.workflow_type.unwrap_or(workflow_type),
862            task_queue: self.task_queue.unwrap_or_default(),
863            arguments,
864            workflow_run_timeout: self
865                .run_timeout
866                .and_then(|duration| duration.try_into().ok()),
867            workflow_task_timeout: self
868                .task_timeout
869                .and_then(|duration| duration.try_into().ok()),
870            backoff_start_interval: self
871                .backoff_start_interval
872                .and_then(|duration| duration.try_into().ok()),
873            memo,
874            headers,
875            search_attributes: self.search_attributes.map(|t| t.into_proto()),
876            retry_policy: self.retry_policy.map(Into::into),
877            versioning_intent: ProtoVersioningIntent::from(
878                self.versioning_intent
879                    .unwrap_or(VersioningIntent::Unspecified),
880            )
881            .into(),
882            initial_versioning_behavior: ProtoContinueAsNewVersioningBehavior::from(
883                self.initial_versioning_behavior
884                    .unwrap_or(ContinueAsNewVersioningBehavior::Unspecified),
885            )
886            .into(),
887        })
888    }
889}
890
891fn command_with_metadata(
892    variant: workflow_command::Variant,
893    summary: Option<String>,
894    details: Option<String>,
895) -> WorkflowCommand {
896    WorkflowCommand {
897        variant: Some(variant),
898        user_metadata: string_user_metadata(summary, details),
899    }
900}
901
902fn string_user_metadata(summary: Option<String>, details: Option<String>) -> Option<UserMetadata> {
903    if summary.is_none() && details.is_none() {
904        return None;
905    }
906    let converter = PayloadConverter::default();
907    let context = SerializationContext {
908        data: &SerializationContextData::Workflow,
909        converter: &converter,
910    };
911    Some(UserMetadata {
912        summary: summary.map(|value| {
913            converter
914                .to_payload(&context, &value)
915                .expect("String-to-JSON payload serialization is infallible")
916        }),
917        details: details.map(|value| {
918            converter
919                .to_payload(&context, &value)
920                .expect("String-to-JSON payload serialization is infallible")
921        }),
922    })
923}
924
925#[cfg(test)]
926mod tests {
927    use super::*;
928
929    #[test]
930    fn activity_cancellation_default_preserves_sdk_behavior() {
931        assert_eq!(
932            ActivityCancellationType::default(),
933            ActivityCancellationType::TryCancel
934        );
935    }
936
937    #[test]
938    fn child_workflow_cancellation_defaults_to_wait_for_completion() {
939        assert_eq!(
940            ChildWorkflowOptions::default().cancel_type,
941            ChildWorkflowCancellationType::WaitCancellationCompleted
942        );
943        let command = ChildWorkflowOptions::default().into_command(
944            1,
945            "child".to_string(),
946            vec![],
947            HashMap::new(),
948            "child-id".to_string(),
949        );
950        let Some(workflow_command::Variant::StartChildWorkflowExecution(command)) = command.variant
951        else {
952            panic!("expected StartChildWorkflowExecution command");
953        };
954        assert_eq!(
955            command.cancellation_type,
956            ProtoChildWorkflowCancellationType::WaitCancellationCompleted as i32
957        );
958    }
959
960    #[test]
961    fn other_policy_defaults_preserve_sdk_behavior() {
962        assert_eq!(ParentClosePolicy::default(), ParentClosePolicy::Unspecified);
963        assert_eq!(
964            WorkflowIdReusePolicy::default(),
965            WorkflowIdReusePolicy::Unspecified
966        );
967        assert_eq!(VersioningIntent::default(), VersioningIntent::Unspecified);
968        assert_eq!(
969            NexusOperationCancellationType::default(),
970            NexusOperationCancellationType::WaitCancellationCompleted
971        );
972    }
973
974    #[test]
975    fn continue_as_new_options_maps_backoff_start_interval_to_request() {
976        let req = ContinueAsNewOptions {
977            backoff_start_interval: Some(Duration::from_secs(7)),
978            versioning_intent: Some(VersioningIntent::Compatible),
979            ..Default::default()
980        }
981        .into_request(
982            "test-workflow".to_string(),
983            vec![],
984            HashMap::new(),
985            &PayloadConverter::default(),
986        )
987        .unwrap();
988
989        let backoff = req
990            .backoff_start_interval
991            .expect("backoff_start_interval should be set");
992        assert_eq!(backoff.seconds, 7);
993        assert_eq!(backoff.nanos, 0);
994        assert_eq!(
995            req.versioning_intent,
996            ProtoVersioningIntent::Compatible as i32
997        );
998    }
999
1000    #[test]
1001    fn activity_options_with_start_to_close_timeout_wrapper_supports_builder_chaining() {
1002        let opts = ActivityOptions::with_start_to_close_timeout(Duration::from_secs(5))
1003            .heartbeat_timeout(Duration::from_secs(2))
1004            .build();
1005
1006        assert_eq!(
1007            opts.close_timeouts,
1008            ActivityCloseTimeouts::StartToClose(Duration::from_secs(5))
1009        );
1010        assert_eq!(opts.heartbeat_timeout, Some(Duration::from_secs(2)));
1011    }
1012
1013    #[test]
1014    fn activity_options_with_schedule_to_close_timeout_wrapper_supports_builder_chaining() {
1015        let opts = ActivityOptions::with_schedule_to_close_timeout(Duration::from_secs(5))
1016            .heartbeat_timeout(Duration::from_secs(2))
1017            .build();
1018
1019        assert_eq!(
1020            opts.close_timeouts,
1021            ActivityCloseTimeouts::ScheduleToClose(Duration::from_secs(5))
1022        );
1023        assert_eq!(opts.heartbeat_timeout, Some(Duration::from_secs(2)));
1024    }
1025
1026    #[test]
1027    fn activity_options_both_close_timeouts_map_to_command() {
1028        let req = ActivityOptions::with_close_timeouts(ActivityCloseTimeouts::Both {
1029            start_to_close: Duration::from_secs(3),
1030            schedule_to_close: Duration::from_secs(8),
1031        })
1032        .cancellation_type(ActivityCancellationType::Abandon)
1033        .build()
1034        .into_command(7, "test".to_string(), vec![], HashMap::new());
1035        let Some(workflow_command::Variant::ScheduleActivity(req)) = req.variant else {
1036            panic!("expected ScheduleActivity command");
1037        };
1038        assert_eq!(req.start_to_close_timeout.unwrap().seconds, 3);
1039        assert_eq!(req.schedule_to_close_timeout.unwrap().seconds, 8);
1040        assert_eq!(
1041            req.cancellation_type,
1042            ProtoActivityCancellationType::Abandon as i32
1043        );
1044    }
1045
1046    #[test]
1047    fn child_workflow_run_timeout_uses_run_timeout_field() {
1048        let opts = ChildWorkflowOptions {
1049            workflow_id: Some("test-wf".to_string()),
1050            cancel_type: ChildWorkflowCancellationType::WaitCancellationRequested,
1051            parent_close_policy: ParentClosePolicy::RequestCancel,
1052            id_reuse_policy: WorkflowIdReusePolicy::RejectDuplicate,
1053            execution_timeout: Some(Duration::from_secs(60)),
1054            run_timeout: Some(Duration::from_secs(10)),
1055            ..Default::default()
1056        };
1057        let command = opts.into_command(
1058            1,
1059            "TestWorkflow".to_string(),
1060            vec![],
1061            HashMap::new(),
1062            "test-wf".into(),
1063        );
1064        let Some(workflow_command::Variant::StartChildWorkflowExecution(req)) = command.variant
1065        else {
1066            panic!("expected StartChildWorkflowExecution command");
1067        };
1068        let exec_timeout = req.workflow_execution_timeout.unwrap();
1069        let run_timeout = req.workflow_run_timeout.unwrap();
1070        assert_eq!(exec_timeout.seconds, 60);
1071        assert_eq!(run_timeout.seconds, 10);
1072        assert_eq!(
1073            req.cancellation_type,
1074            ProtoChildWorkflowCancellationType::WaitCancellationRequested as i32
1075        );
1076        assert_eq!(
1077            req.parent_close_policy,
1078            ProtoParentClosePolicy::RequestCancel as i32
1079        );
1080        assert_eq!(
1081            req.workflow_id_reuse_policy,
1082            ProtoWorkflowIdReusePolicy::RejectDuplicate as i32
1083        );
1084    }
1085
1086    #[test]
1087    fn child_workflow_run_timeout_none_when_unset() {
1088        let opts = ChildWorkflowOptions {
1089            workflow_id: Some("test-wf".to_string()),
1090            execution_timeout: Some(Duration::from_secs(60)),
1091            ..Default::default()
1092        };
1093        let command = opts.into_command(
1094            1,
1095            "TestWorkflow".to_string(),
1096            vec![],
1097            HashMap::new(),
1098            "test-wf".into(),
1099        );
1100        let Some(workflow_command::Variant::StartChildWorkflowExecution(req)) = command.variant
1101        else {
1102            panic!("expected StartChildWorkflowExecution command");
1103        };
1104        let exec_timeout = req.workflow_execution_timeout.unwrap();
1105        assert_eq!(exec_timeout.seconds, 60);
1106        assert!(req.workflow_run_timeout.is_none());
1107    }
1108}
temporalio_workflow/workflow_context/options.rs

temporalio_workflow/workflow_context/
options.rs
