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)]
120pub struct WorkerDeploymentOptions {
121    /// The deployment version of this worker.
122    pub version: WorkerDeploymentVersion,
123    /// If set, opts in to the Worker Deployment Versioning feature, meaning this worker will only
124    /// receive tasks for workflows it claims to be compatible with.
125    pub use_worker_versioning: bool,
126    /// The default versioning behavior to use for workflows that do not pass one to Core.
127    /// It is a startup-time error to specify `Some(Unspecified)` here.
128    pub default_versioning_behavior: Option<VersioningBehavior>,
129}
130
131impl WorkerDeploymentOptions {
132    /// Create deployment options from just a build ID, without opting into worker versioning.
133    pub fn from_build_id(build_id: String) -> Self {
134        Self {
135            version: WorkerDeploymentVersion {
136                deployment_name: "".to_owned(),
137                build_id,
138            },
139            use_worker_versioning: false,
140            default_versioning_behavior: None,
141        }
142    }
143}
144
145static CACHED_BUILD_ID: OnceLock<String> = OnceLock::new();
146
147/// Build ID derived from hashing the on-disk bytes of the current executable.
148/// Deterministic across machines for the same binary. Cached per-process.
149pub fn build_id_from_current_exe() -> &'static str {
150    CACHED_BUILD_ID
151        .get_or_init(|| compute_crc32_exe_id().unwrap_or_else(|_| "undetermined".to_owned()))
152}
153
154fn compute_crc32_exe_id() -> io::Result<String> {
155    let exe_path = std::env::current_exe()?;
156    let file = File::open(exe_path)?;
157    let mut reader = BufReader::new(file);
158
159    let mut hasher = crc32fast::Hasher::new();
160    let mut buf = [0u8; 128 * 1024];
161
162    loop {
163        let n = reader.read(&mut buf)?;
164        if n == 0 {
165            break;
166        }
167        hasher.update(&buf[..n]);
168    }
169
170    let crc = hasher.finalize();
171
172    Ok(format!("{:08x}", crc))
173}
temporalio_common/worker.rs

temporalio_common/
worker.rs
