Review comments

Author: Sushisource

temporalio/bridge/sdk-core

@@ -64,7 +64,7 @@ pub struct WorkerConfig {
 
 #[derive(FromPyObject)]
 pub struct PollerBehaviorSimpleMaximum {
-    pub maximum: usize,
+    pub simple_maximum: usize,
 }
 
 #[derive(FromPyObject)]
@@ -85,7 +85,7 @@ impl From<PollerBehavior> for temporal_sdk_core_api::worker::PollerBehavior {
     fn from(value: PollerBehavior) -> Self {
         match value {
             PollerBehavior::SimpleMaximum(simple) => {
-                temporal_sdk_core_api::worker::PollerBehavior::SimpleMaximum(simple.maximum)
+                temporal_sdk_core_api::worker::PollerBehavior::SimpleMaximum(simple.simple_maximum)
             }
             PollerBehavior::Autoscaling(auto) => {
                 temporal_sdk_core_api::worker::PollerBehavior::Autoscaling {

temporalio/bridge/src/worker.rs

@@ -66,7 +66,7 @@ class WorkerConfig:
 class PollerBehaviorSimpleMaximum:
     """Python representation of the Rust struct for simple poller behavior."""
 
-    maximum: int
+    simple_maximum: int
 
 
 @dataclass

temporalio/bridge/worker.py

@@ -49,7 +49,7 @@
 logger = logging.getLogger(__name__)
 
 
-@dataclass
+@dataclass(frozen=True)
 class PollerBehaviorSimpleMaximum:
     """A poller behavior that will attempt to poll as long as a slot is available, up to the
     provided maximum. Cannot be less than two for workflow tasks, or one for other tasks.
@@ -59,11 +59,11 @@ class PollerBehaviorSimpleMaximum:
 
     def _to_bridge(self) -> temporalio.bridge.worker.PollerBehavior:
         return temporalio.bridge.worker.PollerBehaviorSimpleMaximum(
-            maximum=self.maximum
+            simple_maximum=self.maximum
         )
 
 
-@dataclass
+@dataclass(frozen=True)
 class PollerBehaviorAutoscaling:
     """A poller behavior that will automatically scale the number of pollers based on feedback
     from the server. A slot must be available before beginning polling.
@@ -201,11 +201,16 @@ def __init__(
                 ``max_concurrent_workflow_tasks``, ``max_concurrent_activities``, and
                 ``max_concurrent_local_activities`` arguments.
 
+                Defaults to fixed-size 100 slots for each slot kind if unset and none of the
+                max_* arguments are provided.
+
                 WARNING: This argument is experimental
             max_concurrent_workflow_task_polls: Maximum number of concurrent
                 poll workflow task requests we will perform at a time on this
                 worker's task queue.
 
+                If set, will override any value passed to ``workflow_task_poller_behavior``.
+
                 WARNING: Deprecated, use ``workflow_task_poller_behavior`` instead
             nonsticky_to_sticky_poll_ratio: max_concurrent_workflow_task_polls *
                 this number = the number of max pollers that will be allowed for
@@ -218,6 +223,8 @@ def __init__(
                 poll activity task requests we will perform at a time on this
                 worker's task queue.
 
+                If set, will override any value passed to ``activity_task_poller_behavior``.
+
                 WARNING: Deprecated, use ``activity_task_poller_behavior`` instead
             no_remote_activities: If true, this worker will only handle workflow
                 tasks and local activities, it will not poll for activity tasks.
@@ -284,8 +291,10 @@ def __init__(
             deployment_config: Deployment config for the worker. Exclusive with `build_id` and
                 `use_worker_versioning`.
                 WARNING: This is an experimental feature and may change in the future.
-            workflow_task_poller_behavior: Specify the behavior of workflow task polling
-            activity_task_poller_behavior: Specify the behavior of activity task polling
+            workflow_task_poller_behavior: Specify the behavior of workflow task polling.
+                Defaults to a 5-poller maximum.
+            activity_task_poller_behavior: Specify the behavior of activity task polling.
+                Defaults to a 5-poller maximum.
         """
         if not activities and not workflows:
             raise ValueError("At least one activity or workflow must be specified")
@@ -448,6 +457,15 @@ def __init__(
                 build_id=build_id
             )
 
+        if max_concurrent_workflow_task_polls:
+            workflow_task_poller_behavior = PollerBehaviorSimpleMaximum(
+                maximum=max_concurrent_workflow_task_polls
+            )
+        if max_concurrent_activity_task_polls:
+            activity_task_poller_behavior = PollerBehaviorSimpleMaximum(
+                maximum=max_concurrent_activity_task_polls
+            )
+
         # Create bridge worker last. We have empirically observed that if it is
         # created before an error is raised from the activity worker
         # constructor, a deadlock/hang will occur presumably while trying to

temporalio/worker/_worker.py

@@ -6,6 +6,7 @@
 import uuid
 from datetime import timedelta
 from typing import Any, Awaitable, Callable, Optional, Sequence
+from urllib.request import urlopen
 
 import pytest
 
@@ -22,6 +23,7 @@
 )
 from temporalio.client import BuildIdOpAddNewDefault, Client, TaskReachabilityType
 from temporalio.common import RawValue, VersioningBehavior
+from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
 from temporalio.service import RPCError
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import (
@@ -44,7 +46,12 @@
     WorkflowSlotInfo,
 )
 from temporalio.workflow import VersioningIntent
-from tests.helpers import assert_eventually, new_worker, worker_versioning_enabled
+from tests.helpers import (
+    assert_eventually,
+    find_free_port,
+    new_worker,
+    worker_versioning_enabled,
+)
 
 
 def test_load_default_worker_binary_id():
@@ -920,13 +927,42 @@ async def test_workflows_can_use_default_versioning_behavior(
 async def test_can_run_autoscaling_polling_worker(
     client: Client, env: WorkflowEnvironment
 ):
+    # Create new runtime with Prom server
+    prom_addr = f"127.0.0.1:{find_free_port()}"
+    runtime = Runtime(
+        telemetry=TelemetryConfig(
+            metrics=PrometheusConfig(bind_address=prom_addr),
+        )
+    )
+    client = await Client.connect(
+        client.service_client.config.target_host,
+        namespace=client.namespace,
+        runtime=runtime,
+    )
+
     async with new_worker(
         client,
         WaitOnSignalWorkflow,
         activities=[say_hello],
-        workflow_task_poller_behavior=PollerBehaviorAutoscaling(),
-        activity_task_poller_behavior=PollerBehaviorAutoscaling(),
+        workflow_task_poller_behavior=PollerBehaviorAutoscaling(initial=2),
+        activity_task_poller_behavior=PollerBehaviorAutoscaling(initial=2),
     ) as w:
+        # Give pollers a beat to start
+        await asyncio.sleep(0.3)
+
+        with urlopen(url=f"http://{prom_addr}/metrics") as f:
+            prom_str: str = f.read().decode("utf-8")
+        prom_lines = prom_str.splitlines()
+        matches = [line for line in prom_lines if "temporal_num_pollers" in line]
+        activity_pollers = [l for l in matches if "activity_task" in l]
+        assert len(activity_pollers) == 1
+        assert activity_pollers[0].endswith("2")
+        workflow_pollers = [l for l in matches if "workflow_task" in l]
+        assert len(workflow_pollers) == 2
+        # There's sticky & non-sticky pollers, and they may have a count of 1 or 2 depending on
+        # initialization timing.
+        assert workflow_pollers[0].endswith("2") or workflow_pollers[0].endswith("1")
+        assert workflow_pollers[1].endswith("2") or workflow_pollers[1].endswith("1")
 
         async def do_workflow():
             wf = await client.start_workflow(