💥 Add explicit Worker configuration for type (workflow/activity/nexus) (#1059)

Author: yuandrew

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

@@ -129,8 +129,8 @@ pub trait Worker: Send + Sync {
     /// [Worker::complete_workflow_activation] and [Worker::complete_activity_task] for those
     /// 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::no_remote_activities], you may
-    /// skip calling [Worker::poll_activity_task].
+    /// connection and free resources. If you have set [WorkerConfig::task_types] to exclude
+    /// [WorkerTaskTypes::activity_only()], 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

@@ -16,6 +16,59 @@ use std::{
     time::Duration,
 };
 
+/// Specifies which task types a worker will poll for.
+///
+/// Workers can be configured to handle any combination of workflows, activities, and nexus operations.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub struct WorkerTaskTypes {
+    pub enable_workflows: bool,
+    pub enable_activities: bool,
+    pub enable_nexus: bool,
+}
+
+impl WorkerTaskTypes {
+    /// Check if no task types are enabled
+    pub fn is_empty(&self) -> bool {
+        !self.enable_workflows && !self.enable_activities && !self.enable_nexus
+    }
+
+    /// Create a config with all task types enabled
+    pub fn all() -> WorkerTaskTypes {
+        WorkerTaskTypes {
+            enable_workflows: true,
+            enable_activities: true,
+            enable_nexus: true,
+        }
+    }
+
+    /// Create a config with only workflow tasks enabled
+    pub fn workflow_only() -> WorkerTaskTypes {
+        WorkerTaskTypes {
+            enable_workflows: true,
+            enable_activities: false,
+            enable_nexus: false,
+        }
+    }
+
+    /// Create a config with only activity tasks enabled
+    pub fn activity_only() -> WorkerTaskTypes {
+        WorkerTaskTypes {
+            enable_workflows: false,
+            enable_activities: true,
+            enable_nexus: false,
+        }
+    }
+
+    /// Create a config with only nexus tasks enabled
+    pub fn nexus_only() -> WorkerTaskTypes {
+        WorkerTaskTypes {
+            enable_workflows: false,
+            enable_activities: false,
+            enable_nexus: true,
+        }
+    }
+}
+
 /// Defines per-worker configuration options
 #[derive(Clone, derive_builder::Builder)]
 #[builder(setter(into), build_fn(validate = "Self::validate"))]
@@ -64,10 +117,10 @@ pub struct WorkerConfig {
     /// worker's task queue
     #[builder(default = "PollerBehavior::SimpleMaximum(5)")]
     pub nexus_task_poller_behavior: PollerBehavior,
-    /// If set to true this worker will only handle workflow tasks and local activities, it will not
-    /// poll for activity tasks.
-    #[builder(default = "false")]
-    pub no_remote_activities: bool,
+    /// Specifies which task types this worker will poll for.
+    ///
+    /// Note: At least one task type must be specified or the worker will fail validation.
+    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.
     #[builder(default = "Duration::from_secs(10)")]
@@ -218,6 +271,15 @@ impl WorkerConfigBuilder {
     }
 
     fn validate(&self) -> Result<(), String> {
+        let task_types = self
+            .task_types
+            .as_ref()
+            .cloned()
+            .unwrap_or_else(WorkerTaskTypes::all);
+        if task_types.is_empty() {
+            return Err("At least one task type must be enabled in `task_types`".to_owned());
+        }
+
         if let Some(b) = self.workflow_task_poller_behavior.as_ref() {
             b.validate()?
         }
@@ -249,6 +311,17 @@ impl WorkerConfigBuilder {
             return Err("`max_outstanding_nexus_tasks` must be > 0".to_owned());
         }
 
+        // Validate workflow cache is consistent with task_types
+        if !task_types.enable_workflows
+            && let Some(cache) = self.max_cached_workflows.as_ref()
+            && *cache > 0
+        {
+            return Err(
+                    "Cannot have `max_cached_workflows` > 0 when workflows are not enabled in `task_types`"
+                        .to_owned(),
+                );
+        }
+
         if let Some(cache) = self.max_cached_workflows.as_ref()
             && *cache > 0
         {

crates/common/tests/worker_task_types_test.rs

@@ -0,0 +1,121 @@
+use temporalio_common::worker::{WorkerConfigBuilder, WorkerTaskTypes, WorkerVersioningStrategy};
+
+fn default_versioning_strategy() -> WorkerVersioningStrategy {
+    WorkerVersioningStrategy::None {
+        build_id: String::new(),
+    }
+}
+
+#[test]
+fn test_default_configuration_polls_all_types() {
+    let config = WorkerConfigBuilder::default()
+        .namespace("default")
+        .task_queue("test-queue")
+        .versioning_strategy(default_versioning_strategy())
+        .task_types(WorkerTaskTypes::all())
+        .build()
+        .expect("Failed to build default config");
+
+    let effective = &config.task_types;
+    assert!(
+        effective.enable_workflows,
+        "Should poll workflows by default"
+    );
+    assert!(
+        effective.enable_activities,
+        "Should poll activities by default"
+    );
+    assert!(effective.enable_nexus, "Should poll nexus by default");
+}
+
+#[test]
+fn test_empty_task_types_fails_validation() {
+    let result = WorkerConfigBuilder::default()
+        .namespace("default")
+        .task_queue("test-queue")
+        .versioning_strategy(default_versioning_strategy())
+        .task_types(WorkerTaskTypes {
+            enable_workflows: false,
+            enable_activities: false,
+            enable_nexus: false,
+        })
+        .build();
+
+    assert!(result.is_err(), "Empty task_types should fail validation");
+    let err = result.err().unwrap().to_string();
+    assert!(
+        err.contains("At least one task type"),
+        "Error should mention task types: {err}",
+    );
+}
+
+#[test]
+fn test_workflow_cache_without_workflows_fails() {
+    let result = WorkerConfigBuilder::default()
+        .namespace("default")
+        .task_queue("test-queue")
+        .versioning_strategy(default_versioning_strategy())
+        .task_types(WorkerTaskTypes::activity_only())
+        .max_cached_workflows(10usize)
+        .build();
+
+    assert!(
+        result.is_err(),
+        "Workflow cache > 0 without workflows should fail"
+    );
+    let err = result.err().unwrap().to_string();
+    assert!(
+        err.contains("max_cached_workflows"),
+        "Error should mention max_cached_workflows: {err}",
+    );
+}
+
+#[test]
+fn test_all_combinations() {
+    let combinations = [
+        (WorkerTaskTypes::workflow_only(), "workflows only"),
+        (WorkerTaskTypes::activity_only(), "activities only"),
+        (WorkerTaskTypes::nexus_only(), "nexus only"),
+        (
+            WorkerTaskTypes {
+                enable_workflows: true,
+                enable_activities: true,
+                enable_nexus: false,
+            },
+            "workflows + activities",
+        ),
+        (
+            WorkerTaskTypes {
+                enable_workflows: true,
+                enable_activities: false,
+                enable_nexus: true,
+            },
+            "workflows + nexus",
+        ),
+        (
+            WorkerTaskTypes {
+                enable_workflows: false,
+                enable_activities: true,
+                enable_nexus: true,
+            },
+            "activities + nexus",
+        ),
+        (WorkerTaskTypes::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)
+            .build()
+            .unwrap_or_else(|e| panic!("Failed to build config for {description}: {e:?}"));
+
+        let effective = config.task_types;
+        assert_eq!(
+            effective, task_types,
+            "Effective types should match for {description}",
+        );
+    }
+}

crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h

@@ -738,6 +738,12 @@ typedef struct TemporalCoreTunerHolder {
   struct TemporalCoreSlotSupplier nexus_task_slot_supplier;
 } TemporalCoreTunerHolder;
 
+typedef struct TemporalCoreWorkerTaskTypes {
+  bool enable_workflows;
+  bool enable_activities;
+  bool enable_nexus;
+} TemporalCoreWorkerTaskTypes;
+
 typedef struct TemporalCorePollerBehaviorSimpleMaximum {
   uintptr_t simple_maximum;
 } TemporalCorePollerBehaviorSimpleMaximum;
@@ -765,7 +771,7 @@ typedef struct TemporalCoreWorkerOptions {
   struct TemporalCoreByteArrayRef identity_override;
   uint32_t max_cached_workflows;
   struct TemporalCoreTunerHolder tuner;
-  bool no_remote_activities;
+  struct TemporalCoreWorkerTaskTypes task_types;
   uint64_t sticky_queue_schedule_to_start_timeout_millis;
   uint64_t max_heartbeat_throttle_interval_millis;
   uint64_t default_heartbeat_throttle_interval_millis;

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

@@ -43,7 +43,7 @@ pub struct WorkerOptions {
     pub identity_override: ByteArrayRef,
     pub max_cached_workflows: u32,
     pub tuner: TunerHolder,
-    pub no_remote_activities: bool,
+    pub task_types: WorkerTaskTypes,
     pub sticky_queue_schedule_to_start_timeout_millis: u64,
     pub max_heartbeat_throttle_interval_millis: u64,
     pub default_heartbeat_throttle_interval_millis: u64,
@@ -58,6 +58,23 @@ pub struct WorkerOptions {
     pub nondeterminism_as_workflow_fail_for_types: ByteArrayRefArray,
 }
 
+#[repr(C)]
+pub struct WorkerTaskTypes {
+    pub enable_workflows: bool,
+    pub enable_activities: bool,
+    pub enable_nexus: bool,
+}
+
+impl From<&WorkerTaskTypes> for temporalio_common::worker::WorkerTaskTypes {
+    fn from(t: &WorkerTaskTypes) -> Self {
+        Self {
+            enable_workflows: t.enable_workflows,
+            enable_activities: t.enable_activities,
+            enable_nexus: t.enable_nexus,
+        }
+    }
+}
+
 #[repr(C)]
 pub struct PollerBehaviorSimpleMaximum {
     pub simple_maximum: usize,
@@ -1183,7 +1200,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))
-            .no_remote_activities(opt.no_remote_activities)
+            .task_types(temporalio_common::worker::WorkerTaskTypes::from(
+                &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,
+    worker::{PollerBehavior, WorkerTaskTypes},
 };
 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.no_remote_activities = true;
+            wc.task_types = WorkerTaskTypes::workflow_only();
         } else {
             wc.max_task_queue_activities_per_second = Some(1.0);
         }