temporalio_common/
worker.rs

1//! Contains types that are needed by both the client and the sdk when configuring / interacting
2//! with workers.
3
4use crate::protos::temporal::api::enums::v1::VersioningBehavior as ProtoVersioningBehavior;
5use std::{
6    fs::File,
7    io::{self, BufReader, Read},
8    sync::OnceLock,
9};
10pub use temporalio_common_wasm::worker::WorkerDeploymentVersion;
11
12/// Controls how a workflow moves between worker deployment versions.
13#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)]
14#[non_exhaustive]
15pub enum VersioningBehavior {
16    /// Do not opt into worker deployment versioning.
17    #[default]
18    Unspecified,
19    /// Pin the workflow to one deployment version.
20    Pinned,
21    /// Automatically move the workflow to its target version.
22    AutoUpgrade,
23}
24
25impl From<VersioningBehavior> for ProtoVersioningBehavior {
26    fn from(value: VersioningBehavior) -> Self {
27        match value {
28            VersioningBehavior::Unspecified => Self::Unspecified,
29            VersioningBehavior::Pinned => Self::Pinned,
30            VersioningBehavior::AutoUpgrade => Self::AutoUpgrade,
31        }
32    }
33}
34
35impl From<ProtoVersioningBehavior> for VersioningBehavior {
36    fn from(value: ProtoVersioningBehavior) -> Self {
37        match value {
38            ProtoVersioningBehavior::Unspecified => Self::Unspecified,
39            ProtoVersioningBehavior::Pinned => Self::Pinned,
40            ProtoVersioningBehavior::AutoUpgrade => Self::AutoUpgrade,
41        }
42    }
43}
44
45/// Specifies which task types a worker will poll for.
46///
47/// Workers can be configured to handle any combination of workflows, activities, and nexus operations.
48#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
49pub struct WorkerTaskTypes {
50    /// Whether workflow tasks are enabled.
51    pub enable_workflows: bool,
52    /// Whether local activity tasks are enabled.
53    pub enable_local_activities: bool,
54    /// Whether remote activity tasks are enabled.
55    pub enable_remote_activities: bool,
56    /// Whether nexus tasks are enabled.
57    pub enable_nexus: bool,
58}
59
60impl WorkerTaskTypes {
61    /// Check if no task types are enabled
62    pub fn is_empty(&self) -> bool {
63        !self.enable_workflows
64            && !self.enable_local_activities
65            && !self.enable_remote_activities
66            && !self.enable_nexus
67    }
68
69    /// Create a config with all task types enabled
70    pub fn all() -> WorkerTaskTypes {
71        WorkerTaskTypes {
72            enable_workflows: true,
73            enable_local_activities: true,
74            enable_remote_activities: true,
75            enable_nexus: true,
76        }
77    }
78
79    /// Create a config with only workflow tasks enabled
80    pub fn workflow_only() -> WorkerTaskTypes {
81        WorkerTaskTypes {
82            enable_workflows: true,
83            enable_local_activities: false,
84            enable_remote_activities: false,
85            enable_nexus: false,
86        }
87    }
88
89    /// Create a config with only activity tasks enabled
90    pub fn activity_only() -> WorkerTaskTypes {
91        WorkerTaskTypes {
92            enable_workflows: false,
93            enable_local_activities: false,
94            enable_remote_activities: true,
95            enable_nexus: false,
96        }
97    }
98
99    /// Create a config with only nexus tasks enabled
100    pub fn nexus_only() -> WorkerTaskTypes {
101        WorkerTaskTypes {
102            enable_workflows: false,
103            enable_local_activities: false,
104            enable_remote_activities: false,
105            enable_nexus: true,
106        }
107    }
108
109    /// Returns true if any task type is enabled in both configs.
110    pub fn overlaps_with(&self, other: &WorkerTaskTypes) -> bool {
111        (self.enable_workflows && other.enable_workflows)
112            || (self.enable_local_activities && other.enable_local_activities)
113            || (self.enable_remote_activities && other.enable_remote_activities)
114            || (self.enable_nexus && other.enable_nexus)
115    }
116}
117
118/// Configuration for worker deployment versioning.
119#[derive(Clone, Debug, Eq, PartialEq, Hash, bon::Builder)]
120#[builder(start_fn = new)]
121#[non_exhaustive]
122pub struct WorkerDeploymentOptions {
123    /// The deployment version of this worker.
124    #[builder(start_fn)]
125    pub version: WorkerDeploymentVersion,
126    /// If set, opts in to the Worker Deployment Versioning feature, meaning this worker will only
127    /// receive tasks for workflows it claims to be compatible with.
128    #[builder(default)]
129    pub use_worker_versioning: bool,
130    /// The default versioning behavior to use for workflows that do not pass one to Core.
131    /// It is a startup-time error to specify `Some(Unspecified)` here.
132    pub default_versioning_behavior: Option<VersioningBehavior>,
133}
134
135impl WorkerDeploymentOptions {
136    /// Create deployment options from just a build ID, without opting into worker versioning.
137    pub fn from_build_id(build_id: String) -> Self {
138        Self::new(WorkerDeploymentVersion {
139            deployment_name: "".to_owned(),
140            build_id,
141        })
142        .build()
143    }
144}
145
146static CACHED_BUILD_ID: OnceLock<String> = OnceLock::new();
147
148/// Build ID derived from hashing the on-disk bytes of the current executable.
149/// Deterministic across machines for the same binary. Cached per-process.
150pub fn build_id_from_current_exe() -> &'static str {
151    CACHED_BUILD_ID
152        .get_or_init(|| compute_crc32_exe_id().unwrap_or_else(|_| "undetermined".to_owned()))
153}
154
155fn compute_crc32_exe_id() -> io::Result<String> {
156    let exe_path = std::env::current_exe()?;
157    let file = File::open(exe_path)?;
158    let mut reader = BufReader::new(file);
159
160    let mut hasher = crc32fast::Hasher::new();
161    let mut buf = [0u8; 128 * 1024];
162
163    loop {
164        let n = reader.read(&mut buf)?;
165        if n == 0 {
166            break;
167        }
168        hasher.update(&buf[..n]);
169    }
170
171    let crc = hasher.finalize();
172
173    Ok(format!("{:08x}", crc))
174}
temporalio_common/worker.rs

temporalio_common/
worker.rs
