feat(sdk): auto-enroll default pollers into autoscaling

CHANGELOG.md

@@ -20,6 +20,10 @@ to docs, or any other relevant information.
 ## [0.5.0]
 
 ### Added
+* Workers are now automatically enrolled into poller autoscaling when the namespace advertises the
+  `poller_autoscaling_auto_enroll` capability. This only applies to poller types left at their
+  default (the worker set neither a fixed poller count nor a poller behavior); explicitly
+  configured pollers are left unchanged.
 * `client()` and `workflow_handle()` helpers to `ActivityContext` for easily obtaining a Temporal client
 * Exposed `backoff_start_interval` when continuing as new, which will delay the first task of the
   continued workflow by the configured interval.

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

@@ -94,24 +94,27 @@ pub struct PollerBehavior {
     pub autoscaling: *const PollerBehaviorAutoscaling,
 }
 
-impl TryFrom<&PollerBehavior> for temporalio_sdk_core::PollerBehavior {
-    type Error = anyhow::Error;
-    fn try_from(value: &PollerBehavior) -> Result<Self, Self::Error> {
-        if !value.simple_maximum.is_null() && !value.autoscaling.is_null() {
+impl PollerBehavior {
+    /// Converts an FFI poller behavior into an optional core poller behavior. Both fields null
+    /// means the poller was left unset by lang: core applies the default behavior and treats the
+    /// poller as eligible for automatic enrollment into poller autoscaling.
+    fn to_core(&self) -> anyhow::Result<Option<temporalio_sdk_core::PollerBehavior>> {
+        if !self.simple_maximum.is_null() && !self.autoscaling.is_null() {
             bail!("simple_maximum and autoscaling cannot both be non-null values");
         }
-        if let Some(value) = unsafe { value.simple_maximum.as_ref() } {
-            return Ok(temporalio_sdk_core::PollerBehavior::SimpleMaximum(
-                value.simple_maximum,
-            ));
-        } else if let Some(value) = unsafe { value.autoscaling.as_ref() } {
-            return Ok(temporalio_sdk_core::PollerBehavior::Autoscaling {
-                minimum: value.minimum,
-                maximum: value.maximum,
-                initial: value.initial,
-            });
+        if let Some(sm) = unsafe { self.simple_maximum.as_ref() } {
+            Ok(Some(temporalio_sdk_core::PollerBehavior::SimpleMaximum(
+                sm.simple_maximum,
+            )))
+        } else if let Some(a) = unsafe { self.autoscaling.as_ref() } {
+            Ok(Some(temporalio_sdk_core::PollerBehavior::Autoscaling {
+                minimum: a.minimum,
+                maximum: a.maximum,
+                initial: a.initial,
+            }))
+        } else {
+            Ok(None)
         }
-        bail!("simple_maximum and autoscaling cannot both be null values");
     }
 }
 
@@ -1235,16 +1238,10 @@ impl TryFrom<&WorkerOptions> for temporalio_sdk_core::WorkerConfig {
             // auto-cancel-activity behavior or shutdown will not occur, so we
             // always set it even if 0.
             .graceful_shutdown_period(Duration::from_millis(opt.graceful_shutdown_period_millis))
-            .workflow_task_poller_behavior(temporalio_sdk_core::PollerBehavior::try_from(
-                &opt.workflow_task_poller_behavior,
-            )?)
+            .maybe_workflow_task_poller_behavior(opt.workflow_task_poller_behavior.to_core()?)
             .nonsticky_to_sticky_poll_ratio(opt.nonsticky_to_sticky_poll_ratio)
-            .activity_task_poller_behavior(temporalio_sdk_core::PollerBehavior::try_from(
-                &opt.activity_task_poller_behavior,
-            )?)
-            .nexus_task_poller_behavior(temporalio_sdk_core::PollerBehavior::try_from(
-                &opt.nexus_task_poller_behavior,
-            )?)
+            .maybe_activity_task_poller_behavior(opt.activity_task_poller_behavior.to_core()?)
+            .maybe_nexus_task_poller_behavior(opt.nexus_task_poller_behavior.to_core()?)
             .workflow_failure_errors(if opt.nondeterminism_as_workflow_fail {
                 HashSet::from([WorkflowErrorType::Nondeterminism])
             } else {
@@ -1405,6 +1402,55 @@ mod tests {
         }
     }
 
+    #[test]
+    fn ffi_poller_behavior_opt_maps_unset_to_none() {
+        let unset = PollerBehavior {
+            simple_maximum: std::ptr::null(),
+            autoscaling: std::ptr::null(),
+        };
+        assert!(unset.to_core().unwrap().is_none());
+    }
+
+    #[test]
+    fn ffi_poller_behavior_opt_maps_configured_behaviors() {
+        assert_eq!(
+            simple_poller_behavior(3).to_core().unwrap(),
+            Some(temporalio_sdk_core::PollerBehavior::SimpleMaximum(3)),
+        );
+        let autoscaling = Box::leak(Box::new(PollerBehaviorAutoscaling {
+            minimum: 2,
+            maximum: 20,
+            initial: 4,
+        }));
+        let pb = PollerBehavior {
+            simple_maximum: std::ptr::null(),
+            autoscaling: autoscaling as *const _,
+        };
+        assert_eq!(
+            pb.to_core().unwrap(),
+            Some(temporalio_sdk_core::PollerBehavior::Autoscaling {
+                minimum: 2,
+                maximum: 20,
+                initial: 4,
+            }),
+        );
+    }
+
+    #[test]
+    fn ffi_poller_behavior_opt_rejects_both_set() {
+        let simple = Box::leak(Box::new(PollerBehaviorSimpleMaximum { simple_maximum: 1 }));
+        let autoscaling = Box::leak(Box::new(PollerBehaviorAutoscaling {
+            minimum: 1,
+            maximum: 2,
+            initial: 1,
+        }));
+        let pb = PollerBehavior {
+            simple_maximum: simple as *const _,
+            autoscaling: autoscaling as *const _,
+        };
+        assert!(pb.to_core().is_err());
+    }
+
     fn base_worker_options(
         namespace: &str,
         task_queue: &str,

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

@@ -89,7 +89,7 @@ async fn shutdown_interrupts_both_polls() {
                 .activity_task_poller_behavior(PollerBehavior::SimpleMaximum(1_usize))
                 .build()
                 .unwrap();
-            cfg.workflow_task_poller_behavior = PollerBehavior::SimpleMaximum(1_usize);
+            cfg.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(1_usize));
             cfg
         },
         mock_client,

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

@@ -1326,3 +1326,61 @@ async fn graceful_shutdown_sends_shutdown_worker_rpc_during_initiate() {
 
     worker.finalize_shutdown().await;
 }
+
+#[tokio::test]
+async fn validate_enables_auto_enroll_capability() {
+    let mut mock = mock_worker_client();
+    mock.expect_describe_namespace().returning(|| {
+        Ok(DescribeNamespaceResponse {
+            namespace_info: Some(NamespaceInfo {
+                capabilities: Some(Capabilities {
+                    poller_autoscaling_auto_enroll: true,
+                    ..Capabilities::default()
+                }),
+                ..NamespaceInfo::default()
+            }),
+            ..DescribeNamespaceResponse::default()
+        })
+    });
+    let t = canned_histories::single_timer("1");
+    let mut mh = MockPollCfg::from_resp_batches("fakeid", t, [1], mock);
+    mh.enforce_correct_number_of_polls = false;
+    let worker = mock_worker(build_mock_pollers(mh));
+
+    worker.validate().await.unwrap();
+    let caps = worker.get_namespace_capabilities();
+    assert!(caps.poller_autoscaling_auto_enroll());
+    // The two capabilities are independent: advertising auto-enroll alone does not imply the
+    // `poller_autoscaling` capability (the server advertises each on its own).
+    assert!(!caps.poller_autoscaling());
+
+    worker.drain_pollers_and_shutdown().await;
+}
+
+#[tokio::test]
+async fn validate_without_auto_enroll_leaves_capabilities_off() {
+    let mut mock = mock_worker_client();
+    mock.expect_describe_namespace().returning(|| {
+        Ok(DescribeNamespaceResponse {
+            namespace_info: Some(NamespaceInfo {
+                capabilities: Some(Capabilities {
+                    poller_autoscaling_auto_enroll: false,
+                    ..Capabilities::default()
+                }),
+                ..NamespaceInfo::default()
+            }),
+            ..DescribeNamespaceResponse::default()
+        })
+    });
+    let t = canned_histories::single_timer("1");
+    let mut mh = MockPollCfg::from_resp_batches("fakeid", t, [1], mock);
+    mh.enforce_correct_number_of_polls = false;
+    let worker = mock_worker(build_mock_pollers(mh));
+
+    worker.validate().await.unwrap();
+    let caps = worker.get_namespace_capabilities();
+    assert!(!caps.poller_autoscaling_auto_enroll());
+    assert!(!caps.poller_autoscaling());
+
+    worker.drain_pollers_and_shutdown().await;
+}

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

@@ -2805,7 +2805,7 @@ async fn poller_wont_run_ahead_of_task_slots() {
             let mut cfg = test_worker_cfg().build().unwrap();
             cfg.max_cached_workflows = 10_usize;
             cfg.max_outstanding_workflow_tasks = Some(10_usize);
-            cfg.workflow_task_poller_behavior = PollerBehavior::SimpleMaximum(10_usize);
+            cfg.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(10_usize));
             cfg.task_types = WorkerTaskTypes::workflow_only();
             cfg
         },
@@ -3012,7 +3012,7 @@ async fn slot_provider_cant_hand_out_more_permits_than_cache_size() {
                     .workflow_slot_supplier(Arc::new(EndlessSupplier {}))
                     .build(),
             ));
-            cfg.workflow_task_poller_behavior = PollerBehavior::SimpleMaximum(10_usize);
+            cfg.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(10_usize));
             cfg.task_types = WorkerTaskTypes::workflow_only();
             cfg
         },
@@ -3160,7 +3160,7 @@ async fn both_normal_and_sticky_pollers_poll_concurrently() {
             let mut cfg = test_worker_cfg().build().unwrap();
             cfg.max_cached_workflows = 500_usize; // We need cache, but don't want to deal with evictions
             cfg.max_outstanding_workflow_tasks = Some(2_usize);
-            cfg.workflow_task_poller_behavior = PollerBehavior::SimpleMaximum(2_usize);
+            cfg.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(2_usize));
             cfg.nonsticky_to_sticky_poll_ratio = 0.2;
             cfg.task_types = WorkerTaskTypes::workflow_only();
             cfg

crates/sdk-core/src/replay/mod.rs

@@ -75,7 +75,7 @@ where
 
     pub(crate) fn into_core_worker(mut self) -> Result<Worker, anyhow::Error> {
         self.config.max_cached_workflows = 1;
-        self.config.workflow_task_poller_behavior = PollerBehavior::SimpleMaximum(1);
+        self.config.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(1));
         self.config.task_types = WorkerTaskTypes::workflow_only();
         self.config.skip_client_worker_set_check = true;
         let historator = Historator::new(self.history_stream);