Remove use of bitflags

Author: yuandrew

crates/common/Cargo.toml

@@ -23,7 +23,6 @@ test-utilities = ["history_builders"]
 anyhow = "1.0"
 async-trait = "0.1"
 base64 = "0.22"
-bitflags = "2.6"
 dirs = { version = "6.0", optional = true }
 derive_builder = { workspace = true }
 derive_more = { workspace = true }

crates/common/src/errors.rs

@@ -40,10 +40,14 @@ pub enum CompleteWfError {
         /// The run associated with the completion
         run_id: String,
     },
+    /// Workflows have not been enabled on this worker.
+    #[error("Workflows are not enabled on this worker")]
+    WorkflowNotEnabled,
 }
 
 /// Errors thrown by [crate::Worker::complete_activity_task]
 #[derive(thiserror::Error, Debug)]
+#[allow(clippy::large_enum_variant)]
 pub enum CompleteActivityError {
     /// Lang SDK sent us a malformed activity completion. This likely means a bug in the lang sdk.
     #[error("Lang SDK sent us a malformed activity completion ({reason}): {completion:?}")]
@@ -53,6 +57,9 @@ pub enum CompleteActivityError {
         /// The completion, which may not be included to avoid unnecessary copies.
         completion: Option<ActivityExecutionResult>,
     },
+    /// Activities have not been enabled on this worker.
+    #[error("Activities are not enabled on this worker")]
+    ActivityNotEnabled,
 }
 
 /// Errors thrown by [crate::Worker::complete_nexus_task]

crates/common/src/lib.rs

@@ -130,7 +130,7 @@ pub trait Worker: Send + Sync {
     /// workflows & activities until they are done. At that point, the lang SDK can end the process,
     /// or drop the [Worker] instance via [Worker::finalize_shutdown], which will close the
     /// connection and free resources. If you have set [WorkerConfig::task_types] to exclude
-    /// [WorkerTaskTypes::ACTIVITIES], you may skip calling [Worker::poll_activity_task].
+    /// [worker_task_types::activities()], you may skip calling [Worker::poll_activity_task].
     ///
     /// Lang implementations should use [Worker::initiate_shutdown] followed by
     /// [Worker::finalize_shutdown].

crates/common/src/worker.rs

@@ -8,7 +8,6 @@ use crate::{
     },
     telemetry::metrics::TemporalMeter,
 };
-use bitflags::bitflags;
 use std::{
     any::Any,
     collections::{HashMap, HashSet},
@@ -17,36 +16,70 @@ use std::{
     time::Duration,
 };
 
-bitflags! {
-    /// Specifies which task types a worker will poll for.
-    ///
-    /// This allows fine-grained control over what kinds of work this worker handles.
-    /// Workers can be configured to handle any combination of workflows, activities, and nexus operations.
-    #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-    pub struct WorkerTaskTypes: u8 {
-        /// Poll for workflow tasks
-        const WORKFLOWS = 0b001;
-        /// Poll for activity tasks (remote activities)
-        const ACTIVITIES = 0b010;
-        /// Poll for nexus tasks
-        const NEXUS = 0b100;
-    }
+/// Represents a single task type that a worker can poll for
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum WorkerTaskType {
+    /// Poll for workflow tasks
+    Workflows,
+    /// Poll for activity tasks (remote activities)
+    Activities,
+    /// Poll for nexus tasks
+    Nexus,
 }
 
-impl WorkerTaskTypes {
-    /// Returns true if this worker should poll for workflow tasks
-    pub fn polls_workflows(&self) -> bool {
-        self.contains(Self::WORKFLOWS)
+/// Specifies which task types a worker will poll for.
+///
+/// This is a set of [WorkerTaskType] values. Workers can be configured to handle
+/// any combination of workflows, activities, and nexus operations.
+pub type WorkerTaskTypes = HashSet<WorkerTaskType>;
+
+/// Helper functions for working with WorkerTaskTypes
+pub mod worker_task_types {
+    use super::{WorkerTaskType, WorkerTaskTypes};
+
+    /// Create a set with all task types enabled
+    pub fn all() -> WorkerTaskTypes {
+        [
+            WorkerTaskType::Workflows,
+            WorkerTaskType::Activities,
+            WorkerTaskType::Nexus,
+        ]
+        .into_iter()
+        .collect()
+    }
+
+    /// Create a set with only workflow tasks enabled
+    pub fn workflows() -> WorkerTaskTypes {
+        [WorkerTaskType::Workflows].into_iter().collect()
+    }
+
+    /// Create a set with only activity tasks enabled
+    pub fn activities() -> WorkerTaskTypes {
+        [WorkerTaskType::Activities].into_iter().collect()
     }
 
-    /// Returns true if this worker should poll for activity tasks
-    pub fn polls_activities(&self) -> bool {
-        self.contains(Self::ACTIVITIES)
+    /// Create a set with only nexus tasks enabled
+    pub fn nexus() -> WorkerTaskTypes {
+        [WorkerTaskType::Nexus].into_iter().collect()
     }
 
-    /// Returns true if this worker should poll for nexus tasks
-    pub fn polls_nexus(&self) -> bool {
-        self.contains(Self::NEXUS)
+    /// Create a set from a bitmask (for C FFI compatibility)
+    /// 0x01 = Workflows, 0x02 = Activities, 0x04 = Nexus
+    pub fn from_bits(bits: u8) -> WorkerTaskTypes {
+        if bits == 0 {
+            return all();
+        }
+        let mut set = WorkerTaskTypes::new();
+        if bits & 0x01 != 0 {
+            set.insert(WorkerTaskType::Workflows);
+        }
+        if bits & 0x02 != 0 {
+            set.insert(WorkerTaskType::Activities);
+        }
+        if bits & 0x04 != 0 {
+            set.insert(WorkerTaskType::Nexus);
+        }
+        set
     }
 }
 
@@ -101,12 +134,10 @@ pub struct WorkerConfig {
     /// Specifies which task types this worker will poll for.
     ///
     /// By default, workers poll for all task types (workflows, activities, and nexus).
-    /// You can restrict this to any combination. For example, a workflow-only worker would use
-    /// `WorkerTaskTypes::WORKFLOWS`, while a worker handling activities and nexus but not workflows
-    /// would use `WorkerTaskTypes::ACTIVITIES | WorkerTaskTypes::NEXUS`.
+    /// You can restrict this to any combination.
     ///
     /// Note: At least one task type must be specified or the worker will fail validation.
-    #[builder(default = "WorkerTaskTypes::all()")]
+    #[builder(default = "worker_task_types::all()")]
     pub task_types: WorkerTaskTypes,
     /// How long a workflow task is allowed to sit on the sticky queue before it is timed out
     /// and moved to the non-sticky queue where it may be picked up by any worker.
@@ -258,7 +289,11 @@ impl WorkerConfigBuilder {
     }
 
     fn validate(&self) -> Result<(), String> {
-        let task_types = self.task_types.unwrap_or_else(WorkerTaskTypes::all);
+        let task_types = self
+            .task_types
+            .as_ref()
+            .cloned()
+            .unwrap_or_else(worker_task_types::all);
         if task_types.is_empty() {
             return Err("At least one task type must be enabled in `task_types`".to_owned());
         }
@@ -295,7 +330,7 @@ impl WorkerConfigBuilder {
         }
 
         // Validate workflow cache is consistent with task_types
-        if !task_types.contains(WorkerTaskTypes::WORKFLOWS)
+        if !task_types.contains(&WorkerTaskType::Workflows)
             && let Some(cache) = self.max_cached_workflows.as_ref()
             && *cache > 0
         {

crates/common/tests/worker_task_types_test.rs

@@ -1,4 +1,8 @@
-use temporalio_common::worker::{WorkerConfigBuilder, WorkerTaskTypes, WorkerVersioningStrategy};
+use std::collections::HashSet;
+use temporalio_common::worker::{
+    WorkerConfigBuilder, WorkerTaskType, WorkerTaskTypes, WorkerVersioningStrategy,
+    worker_task_types,
+};
 
 fn default_versioning_strategy() -> WorkerVersioningStrategy {
     WorkerVersioningStrategy::None {
@@ -15,16 +19,19 @@ fn test_default_configuration_polls_all_types() {
         .build()
         .expect("Failed to build default config");
 
-    let effective = config.task_types;
+    let effective = &config.task_types;
     assert!(
-        effective.polls_workflows(),
+        effective.contains(&WorkerTaskType::Workflows),
         "Should poll workflows by default"
     );
     assert!(
-        effective.polls_activities(),
+        effective.contains(&WorkerTaskType::Activities),
         "Should poll activities by default"
     );
-    assert!(effective.polls_nexus(), "Should poll nexus by default");
+    assert!(
+        effective.contains(&WorkerTaskType::Nexus),
+        "Should poll nexus by default"
+    );
 }
 
 #[test]
@@ -33,32 +40,53 @@ fn test_workflow_only_worker() {
         .namespace("default")
         .task_queue("test-queue")
         .versioning_strategy(default_versioning_strategy())
-        .task_types(WorkerTaskTypes::WORKFLOWS)
+        .task_types(HashSet::from([WorkerTaskType::Workflows]))
         .max_cached_workflows(0usize)
         .build()
         .expect("Failed to build workflow-only config");
 
-    let effective = config.task_types;
-    assert!(effective.polls_workflows(), "Should poll workflows");
-    assert!(!effective.polls_activities(), "Should NOT poll activities");
-    assert!(!effective.polls_nexus(), "Should NOT poll nexus");
+    let effective = &config.task_types;
+    assert!(
+        effective.contains(&WorkerTaskType::Workflows),
+        "Should poll workflows"
+    );
+    assert!(
+        !effective.contains(&WorkerTaskType::Activities),
+        "Should NOT poll activities"
+    );
+    assert!(
+        !effective.contains(&WorkerTaskType::Nexus),
+        "Should NOT poll nexus"
+    );
 }
 
 #[test]
 fn test_activity_and_nexus_worker() {
+    let types: WorkerTaskTypes = [WorkerTaskType::Activities, WorkerTaskType::Nexus]
+        .into_iter()
+        .collect();
     let config = WorkerConfigBuilder::default()
         .namespace("default")
         .task_queue("test-queue")
         .versioning_strategy(default_versioning_strategy())
-        .task_types(WorkerTaskTypes::ACTIVITIES | WorkerTaskTypes::NEXUS)
+        .task_types(types)
         .max_cached_workflows(0usize)
         .build()
         .expect("Failed to build activity+nexus config");
 
-    let effective = config.task_types;
-    assert!(!effective.polls_workflows(), "Should NOT poll workflows");
-    assert!(effective.polls_activities(), "Should poll activities");
-    assert!(effective.polls_nexus(), "Should poll nexus");
+    let effective = &config.task_types;
+    assert!(
+        !effective.contains(&WorkerTaskType::Workflows),
+        "Should NOT poll workflows"
+    );
+    assert!(
+        effective.contains(&WorkerTaskType::Activities),
+        "Should poll activities"
+    );
+    assert!(
+        effective.contains(&WorkerTaskType::Nexus),
+        "Should poll nexus"
+    );
 }
 
 #[test]
@@ -67,7 +95,7 @@ fn test_empty_task_types_fails_validation() {
         .namespace("default")
         .task_queue("test-queue")
         .versioning_strategy(default_versioning_strategy())
-        .task_types(WorkerTaskTypes::empty())
+        .task_types(WorkerTaskTypes::new())
         .build();
 
     assert!(result.is_err(), "Empty task_types should fail validation");
@@ -84,7 +112,7 @@ fn test_workflow_cache_without_workflows_fails() {
         .namespace("default")
         .task_queue("test-queue")
         .versioning_strategy(default_versioning_strategy())
-        .task_types(WorkerTaskTypes::ACTIVITIES)
+        .task_types(worker_task_types::activities())
         .max_cached_workflows(10usize)
         .build();
 
@@ -102,30 +130,36 @@ fn test_workflow_cache_without_workflows_fails() {
 #[test]
 fn test_all_combinations() {
     let combinations = [
-        (WorkerTaskTypes::WORKFLOWS, "workflows only"),
-        (WorkerTaskTypes::ACTIVITIES, "activities only"),
-        (WorkerTaskTypes::NEXUS, "nexus only"),
+        (worker_task_types::workflows(), "workflows only"),
+        (worker_task_types::activities(), "activities only"),
+        (worker_task_types::nexus(), "nexus only"),
         (
-            WorkerTaskTypes::WORKFLOWS | WorkerTaskTypes::ACTIVITIES,
+            [WorkerTaskType::Workflows, WorkerTaskType::Activities]
+                .into_iter()
+                .collect(),
             "workflows + activities",
         ),
         (
-            WorkerTaskTypes::WORKFLOWS | WorkerTaskTypes::NEXUS,
+            [WorkerTaskType::Workflows, WorkerTaskType::Nexus]
+                .into_iter()
+                .collect(),
             "workflows + nexus",
         ),
         (
-            WorkerTaskTypes::ACTIVITIES | WorkerTaskTypes::NEXUS,
+            [WorkerTaskType::Activities, WorkerTaskType::Nexus]
+                .into_iter()
+                .collect(),
             "activities + nexus",
         ),
-        (WorkerTaskTypes::all(), "all types"),
+        (worker_task_types::all(), "all types"),
     ];
 
     for (task_types, description) in combinations {
         let config = WorkerConfigBuilder::default()
             .namespace("default")
             .task_queue("test-queue")
             .versioning_strategy(default_versioning_strategy())
-            .task_types(task_types)
+            .task_types(task_types.clone())
             .build()
             .unwrap_or_else(|e| panic!("Failed to build config for {description}: {e:?}"));
 

crates/sdk-core-c-bridge/src/worker.rs

@@ -1183,13 +1183,9 @@ impl TryFrom<&WorkerOptions> for temporalio_sdk_core::WorkerConfig {
             .client_identity_override(opt.identity_override.to_option_string())
             .max_cached_workflows(opt.max_cached_workflows as usize)
             .tuner(Arc::new(converted_tuner))
-            .task_types({
-                if opt.task_types == 0 {
-                    temporalio_common::worker::WorkerTaskTypes::all()
-                } else {
-                    temporalio_common::worker::WorkerTaskTypes::from_bits_truncate(opt.task_types)
-                }
-            })
+            .task_types(temporalio_common::worker::worker_task_types::from_bits(
+                opt.task_types,
+            ))
             .sticky_queue_schedule_to_start_timeout(Duration::from_millis(
                 opt.sticky_queue_schedule_to_start_timeout_millis,
             ))

crates/sdk-core/src/core_tests/activity_tasks.rs

@@ -53,7 +53,7 @@ use temporalio_common::{
         },
         test_utils::start_timer_cmd,
     },
-    worker::{PollerBehavior, WorkerTaskTypes},
+    worker::{PollerBehavior, worker_task_types},
 };
 use tokio::{join, time::sleep};
 use tokio_util::sync::CancellationToken;
@@ -725,7 +725,7 @@ async fn no_eager_activities_requested_when_worker_options_disable_it(
     mock.worker_cfg(|wc| {
         wc.max_cached_workflows = 2;
         if reason == "no_remote" {
-            wc.task_types = WorkerTaskTypes::WORKFLOWS;
+            wc.task_types = worker_task_types::workflows();
         } else {
             wc.max_task_queue_activities_per_second = Some(1.0);
         }