Pass unset poller behavior to core so workers can auto-enroll in poller autoscaling

changelog/added/eager-poller-pelican.md

@@ -0,0 +1 @@
+Workers that don't configure poller behavior now auto-enroll in poller autoscaling when the server enables it (`frontend.pollerAutoscalingAutoEnroll`); otherwise they keep using a fixed 5-poller maximum.

temporalio/ext/src/worker.rs

@@ -511,21 +511,21 @@ fn build_config(options: Struct, runtime_handle: &RuntimeHandle) -> Result<Worke
         })
         .maybe_client_identity_override(options.member::<Option<String>>(id!("identity_override"))?)
         .max_cached_workflows(options.member::<usize>(id!("max_cached_workflows"))?)
-        .workflow_task_poller_behavior({
-            let poller_behavior = options
+        .maybe_workflow_task_poller_behavior(
+            options
                 .child(id!("workflow_task_poller_behavior"))?
-                .ok_or_else(|| error!("Worker options must have workflow_task_poller_behavior"))?;
-            extract_poller_behavior(poller_behavior)?
-        })
+                .map(extract_poller_behavior)
+                .transpose()?,
+        )
         .nonsticky_to_sticky_poll_ratio(
             options.member::<f32>(id!("nonsticky_to_sticky_poll_ratio"))?,
         )
-        .activity_task_poller_behavior({
-            let poller_behavior = options
+        .maybe_activity_task_poller_behavior(
+            options
                 .child(id!("activity_task_poller_behavior"))?
-                .ok_or_else(|| error!("Worker options must have activity_task_poller_behavior"))?;
-            extract_poller_behavior(poller_behavior)?
-        })
+                .map(extract_poller_behavior)
+                .transpose()?,
+        )
         .task_types(WorkerTaskTypes {
             enable_workflows: options.member(id!("enable_workflows"))?,
             enable_local_activities: options.member(id!("enable_local_activities"))?,

temporalio/lib/temporalio/worker.rb

@@ -395,15 +395,15 @@ def self._validate_plugins!(plugins)
     # @param logger [Logger] Logger to override client logger with. Default is the client logger.
     # @param max_cached_workflows [Integer] Number of workflows held in cache for use by sticky task queue. If set to 0,
     #   workflow caching and sticky queuing are disabled.
-    # @param max_concurrent_workflow_task_polls [Integer] Maximum number of concurrent poll workflow task requests we
-    #   will perform at a time on this worker's task queue.
+    # @param max_concurrent_workflow_task_polls [Integer, nil] Maximum number of concurrent poll workflow task requests
+    #   we will perform at a time on this worker's task queue.
     # @param nonsticky_to_sticky_poll_ratio [Float] `max_concurrent_workflow_task_polls` * this number = the number of
     #   max pollers that will be allowed for the nonsticky queue when sticky tasks are enabled. If both defaults are
     #   used, the sticky queue will allow 4 max pollers while the nonsticky queue will allow one. The minimum for either
     #   poller is 1, so if `max_concurrent_workflow_task_polls` is 1 and sticky queues are enabled, there will be 2
     #   concurrent polls.
-    # @param max_concurrent_activity_task_polls [Integer] Maximum number of concurrent poll activity task requests we
-    #   will perform at a time on this worker's task queue.
+    # @param max_concurrent_activity_task_polls [Integer, nil] Maximum number of concurrent poll activity task requests
+    #   we will perform at a time on this worker's task queue.
     # @param no_remote_activities [Boolean] If true, this worker will only handle workflow tasks and local activities,
     #   it will not poll for activity tasks.
     # @param sticky_queue_schedule_to_start_timeout [Float] How long a workflow task is allowed to sit on the sticky
@@ -450,9 +450,9 @@ def self._validate_plugins!(plugins)
     # @param patch_activation_callback [Proc, nil] Experimental callback to decide whether the first non-replay call to
     #   {Workflow.patched} for a patch ID should activate that patch. The callback receives a {PatchActivationInput} and
     #   must return `true` to activate the patch or `false` to leave it inactive.
-    # @param workflow_task_poller_behavior [PollerBehavior] Specify the behavior of workflow task
+    # @param workflow_task_poller_behavior [PollerBehavior, nil] Specify the behavior of workflow task
     #   polling. Defaults to a 5-poller maximum.
-    # @param activity_task_poller_behavior [PollerBehavior] Specify the behavior of activity task
+    # @param activity_task_poller_behavior [PollerBehavior, nil] Specify the behavior of activity task
     #   polling. Defaults to a 5-poller maximum.
     # @param debug_mode [Boolean] If true, deadlock detection is disabled. Deadlock detection will fail workflow tasks
     #   if they block the thread for too long. This defaults to true if the `TEMPORAL_DEBUG` environment variable is
@@ -473,9 +473,9 @@ def initialize(
       identity: nil,
       logger: client.options.logger,
       max_cached_workflows: 1000,
-      max_concurrent_workflow_task_polls: 5,
+      max_concurrent_workflow_task_polls: nil,
       nonsticky_to_sticky_poll_ratio: 0.2,
-      max_concurrent_activity_task_polls: 5,
+      max_concurrent_activity_task_polls: nil,
       no_remote_activities: false,
       sticky_queue_schedule_to_start_timeout: 10,
       max_heartbeat_throttle_interval: 60,
@@ -491,13 +491,20 @@ def initialize(
       unsafe_workflow_io_enabled: false,
       deployment_options: Worker.default_deployment_options,
       patch_activation_callback: nil,
-      workflow_task_poller_behavior: PollerBehavior::SimpleMaximum.new(max_concurrent_workflow_task_polls),
-      activity_task_poller_behavior: PollerBehavior::SimpleMaximum.new(max_concurrent_activity_task_polls),
+      workflow_task_poller_behavior: nil,
+      activity_task_poller_behavior: nil,
       debug_mode: %w[true 1].include?(ENV['TEMPORAL_DEBUG'].to_s.downcase),
       disable_payload_error_limit: false
     )
       Internal::ProtoUtils.assert_non_reserved_name(task_queue)
 
+      if max_concurrent_workflow_task_polls
+        workflow_task_poller_behavior ||= PollerBehavior::SimpleMaximum.new(max_concurrent_workflow_task_polls)
+      end
+      if max_concurrent_activity_task_polls
+        activity_task_poller_behavior ||= PollerBehavior::SimpleMaximum.new(max_concurrent_activity_task_polls)
+      end
+
       @options = Options.new(
         client:,
         task_queue:,
@@ -582,9 +589,9 @@ def _initialize_from_options
           tuner: @options.tuner._to_bridge_options,
           identity_override: @options.identity,
           max_cached_workflows: @options.max_cached_workflows,
-          workflow_task_poller_behavior: @options.workflow_task_poller_behavior._to_bridge_options,
+          workflow_task_poller_behavior: @options.workflow_task_poller_behavior&._to_bridge_options,
           nonsticky_to_sticky_poll_ratio: @options.nonsticky_to_sticky_poll_ratio,
-          activity_task_poller_behavior: @options.activity_task_poller_behavior._to_bridge_options,
+          activity_task_poller_behavior: @options.activity_task_poller_behavior&._to_bridge_options,
           enable_workflows: !@options.workflows.empty?,
           enable_local_activities: !@options.workflows.empty? && !@options.activities.empty?,
           enable_remote_activities: !@options.activities.empty? && !@options.no_remote_activities,

temporalio/rbi/temporalio/internal/bridge/worker.rbi

@@ -58,15 +58,15 @@ class Temporalio::Internal::Bridge::Worker::Options < ::Struct
       tuner: Temporalio::Internal::Bridge::Worker::TunerOptions,
       identity_override: T.nilable(String),
       max_cached_workflows: Integer,
-      workflow_task_poller_behavior: T.any(
+      workflow_task_poller_behavior: T.nilable(T.any(
         Temporalio::Internal::Bridge::Worker::PollerBehaviorSimpleMaximum,
         Temporalio::Internal::Bridge::Worker::PollerBehaviorAutoscaling
-      ),
+      )),
       nonsticky_to_sticky_poll_ratio: T.any(Integer, Float),
-      activity_task_poller_behavior: T.any(
+      activity_task_poller_behavior: T.nilable(T.any(
         Temporalio::Internal::Bridge::Worker::PollerBehaviorSimpleMaximum,
         Temporalio::Internal::Bridge::Worker::PollerBehaviorAutoscaling
-      ),
+      )),
       enable_workflows: T::Boolean,
       enable_local_activities: T::Boolean,
       enable_remote_activities: T::Boolean,

temporalio/rbi/temporalio/worker.rbi

@@ -20,9 +20,9 @@ class Temporalio::Worker
       identity: T.nilable(String),
       logger: Logger,
       max_cached_workflows: Integer,
-      max_concurrent_workflow_task_polls: Integer,
+      max_concurrent_workflow_task_polls: T.nilable(Integer),
       nonsticky_to_sticky_poll_ratio: Numeric,
-      max_concurrent_activity_task_polls: Integer,
+      max_concurrent_activity_task_polls: T.nilable(Integer),
       no_remote_activities: T::Boolean,
       sticky_queue_schedule_to_start_timeout: Numeric,
       max_heartbeat_throttle_interval: Numeric,
@@ -38,8 +38,8 @@ class Temporalio::Worker
       unsafe_workflow_io_enabled: T::Boolean,
       deployment_options: Temporalio::Worker::DeploymentOptions,
       patch_activation_callback: T.nilable(T.proc.params(input: Temporalio::Worker::PatchActivationInput).returns(T::Boolean)),
-      workflow_task_poller_behavior: Temporalio::Worker::PollerBehavior,
-      activity_task_poller_behavior: Temporalio::Worker::PollerBehavior,
+      workflow_task_poller_behavior: T.nilable(Temporalio::Worker::PollerBehavior),
+      activity_task_poller_behavior: T.nilable(Temporalio::Worker::PollerBehavior),
       debug_mode: T::Boolean,
       disable_payload_error_limit: T::Boolean
     ).void
@@ -197,13 +197,13 @@ class Temporalio::Worker::Options < ::Data
   sig { returns(Integer) }
   def max_cached_workflows; end
 
-  sig { returns(Integer) }
+  sig { returns(T.nilable(Integer)) }
   def max_concurrent_workflow_task_polls; end
 
   sig { returns(Numeric) }
   def nonsticky_to_sticky_poll_ratio; end
 
-  sig { returns(Integer) }
+  sig { returns(T.nilable(Integer)) }
   def max_concurrent_activity_task_polls; end
 
   sig { returns(T::Boolean) }
@@ -248,10 +248,10 @@ class Temporalio::Worker::Options < ::Data
   sig { returns(T.nilable(T.proc.params(input: Temporalio::Worker::PatchActivationInput).returns(T::Boolean))) }
   def patch_activation_callback; end
 
-  sig { returns(Temporalio::Worker::PollerBehavior) }
+  sig { returns(T.nilable(Temporalio::Worker::PollerBehavior)) }
   def workflow_task_poller_behavior; end
 
-  sig { returns(Temporalio::Worker::PollerBehavior) }
+  sig { returns(T.nilable(Temporalio::Worker::PollerBehavior)) }
   def activity_task_poller_behavior; end
 
   sig { returns(Temporalio::Worker::DeploymentOptions) }

temporalio/sig/temporalio/internal/bridge/worker.rbs

@@ -8,9 +8,9 @@ module Temporalio
           attr_accessor tuner: TunerOptions
           attr_accessor identity_override: String?
           attr_accessor max_cached_workflows: Integer
-          attr_accessor workflow_task_poller_behavior: PollerBehaviorSimpleMaximum | PollerBehaviorAutoscaling
+          attr_accessor workflow_task_poller_behavior: (PollerBehaviorSimpleMaximum | PollerBehaviorAutoscaling)?
           attr_accessor nonsticky_to_sticky_poll_ratio: Float
-          attr_accessor activity_task_poller_behavior: PollerBehaviorSimpleMaximum | PollerBehaviorAutoscaling
+          attr_accessor activity_task_poller_behavior: (PollerBehaviorSimpleMaximum | PollerBehaviorAutoscaling)?
           attr_accessor enable_workflows: bool
           attr_accessor enable_local_activities: bool
           attr_accessor enable_remote_activities: bool
@@ -34,9 +34,9 @@ module Temporalio
             tuner: TunerOptions,
             identity_override: String?,
             max_cached_workflows: Integer,
-            workflow_task_poller_behavior: PollerBehaviorSimpleMaximum | PollerBehaviorAutoscaling,
+            workflow_task_poller_behavior: (PollerBehaviorSimpleMaximum | PollerBehaviorAutoscaling)?,
             nonsticky_to_sticky_poll_ratio: Float,
-            activity_task_poller_behavior: PollerBehaviorSimpleMaximum | PollerBehaviorAutoscaling,
+            activity_task_poller_behavior: (PollerBehaviorSimpleMaximum | PollerBehaviorAutoscaling)?,
             enable_workflows: bool,
             enable_local_activities: bool,
             enable_remote_activities: bool,

temporalio/sig/temporalio/worker.rbs

@@ -21,9 +21,9 @@ module Temporalio
       attr_reader identity: String
       attr_reader logger: Logger
       attr_reader max_cached_workflows: Integer
-      attr_reader max_concurrent_workflow_task_polls: Integer
+      attr_reader max_concurrent_workflow_task_polls: Integer?
       attr_reader nonsticky_to_sticky_poll_ratio: Float
-      attr_reader max_concurrent_activity_task_polls: Integer
+      attr_reader max_concurrent_activity_task_polls: Integer?
       attr_reader no_remote_activities: bool
       attr_reader sticky_queue_schedule_to_start_timeout: Float
       attr_reader max_heartbeat_throttle_interval: Float
@@ -37,8 +37,8 @@ module Temporalio
       attr_reader workflow_failure_exception_types: Array[singleton(Exception)]
       attr_reader workflow_payload_codec_thread_pool: ThreadPool?
       attr_reader unsafe_workflow_io_enabled: bool
-      attr_reader workflow_task_poller_behavior: PollerBehavior
-      attr_reader activity_task_poller_behavior: PollerBehavior
+      attr_reader workflow_task_poller_behavior: PollerBehavior?
+      attr_reader activity_task_poller_behavior: PollerBehavior?
       attr_reader deployment_options: Worker::DeploymentOptions
       attr_reader patch_activation_callback: Proc?
       attr_reader debug_mode: bool
@@ -57,9 +57,9 @@ module Temporalio
         identity: String?,
         logger: Logger,
         max_cached_workflows: Integer,
-        max_concurrent_workflow_task_polls: Integer,
+        max_concurrent_workflow_task_polls: Integer?,
         nonsticky_to_sticky_poll_ratio: Float,
-        max_concurrent_activity_task_polls: Integer,
+        max_concurrent_activity_task_polls: Integer?,
         no_remote_activities: bool,
         sticky_queue_schedule_to_start_timeout: Float,
         max_heartbeat_throttle_interval: Float,
@@ -73,8 +73,8 @@ module Temporalio
         workflow_failure_exception_types: Array[singleton(Exception)],
         workflow_payload_codec_thread_pool: ThreadPool?,
         unsafe_workflow_io_enabled: bool,
-        workflow_task_poller_behavior: PollerBehavior,
-        activity_task_poller_behavior: PollerBehavior,
+        workflow_task_poller_behavior: PollerBehavior?,
+        activity_task_poller_behavior: PollerBehavior?,
         deployment_options: Worker::DeploymentOptions,
         patch_activation_callback: Proc?,
         debug_mode: bool,
@@ -123,9 +123,9 @@ module Temporalio
       ?identity: String?,
       ?logger: Logger,
       ?max_cached_workflows: Integer,
-      ?max_concurrent_workflow_task_polls: Integer,
+      ?max_concurrent_workflow_task_polls: Integer?,
       ?nonsticky_to_sticky_poll_ratio: Float,
-      ?max_concurrent_activity_task_polls: Integer,
+      ?max_concurrent_activity_task_polls: Integer?,
       ?no_remote_activities: bool,
       ?sticky_queue_schedule_to_start_timeout: Float,
       ?max_heartbeat_throttle_interval: Float,
@@ -139,8 +139,8 @@ module Temporalio
       ?workflow_failure_exception_types: Array[singleton(Exception)],
       ?workflow_payload_codec_thread_pool: ThreadPool?,
       ?unsafe_workflow_io_enabled: bool,
-      ?workflow_task_poller_behavior: PollerBehavior,
-      ?activity_task_poller_behavior: PollerBehavior,
+      ?workflow_task_poller_behavior: PollerBehavior?,
+      ?activity_task_poller_behavior: PollerBehavior?,
       ?deployment_options: Worker::DeploymentOptions,
       ?patch_activation_callback: Proc?,
       ?debug_mode: bool,