More rusty, clarify naming, simplify Runtime args, add Client plugins to list

Author: yuandrew

temporalio/client.py

@@ -349,6 +349,11 @@ def api_key(self, value: Optional[str]) -> None:
         self.service_client.config.api_key = value
         self.service_client.update_api_key(value)
 
+    @property
+    def plugins(self) -> Sequence[Plugin]:
+        """Plugins used by this client."""
+        return self._config["plugins"]
+
     # Overload for no-param workflow
     @overload
     async def start_workflow(

temporalio/runtime.py

@@ -73,8 +73,8 @@ def set_default(runtime: Runtime, *, error_if_already_set: bool = True) -> None:
     def __init__(
         self,
         *,
-        telemetry: Optional[TelemetryConfig] = None,
-        runtime_options: Optional["RuntimeOptions"] = None,
+        telemetry: TelemetryConfig,
+        worker_heartbeat_interval: Optional[timedelta] = timedelta(seconds=30),
     ) -> None:
         """Create a runtime with the provided configuration.
 
@@ -83,25 +83,29 @@ def __init__(
         Args:
             telemetry: Telemetry configuration when not supplying
                 ``runtime_options``.
-            runtime_options: Full runtime configuration including telemetry and
-                worker heartbeating options.
+            worker_heartbeat_interval: Interval for worker heartbeats. ``None``
+                disables heartbeating.
 
         Raises:
-            ValueError: If both ``telemetry`` and ``runtime_options`` are
-                provided.
+            ValueError: If both ```runtime_options`` is a negative value.
         """
-        if runtime_options and telemetry:
-            raise ValueError("Cannot supply both telemetry and runtime_options")
-
-        if runtime_options is None:
-            telemetry = telemetry or TelemetryConfig()
-            runtime_options = RuntimeOptions(telemetry=telemetry)
+        if worker_heartbeat_interval is None:
+            heartbeat_millis = None
         else:
-            telemetry = runtime_options.telemetry
+            if worker_heartbeat_interval <= timedelta(0):
+                raise ValueError("worker_heartbeat_interval must be positive")
+            heartbeat_millis = int(worker_heartbeat_interval.total_seconds() * 1000)
+            if heartbeat_millis == 0:
+                heartbeat_millis = 1
+
+        self._heartbeat_millis = heartbeat_millis
 
-        self._core_runtime = temporalio.bridge.runtime.Runtime(
-            options=runtime_options._to_bridge_config()
+        runtime_options = temporalio.bridge.runtime.RuntimeOptions(
+            telemetry=telemetry._to_bridge_config(),
+            worker_heartbeat_interval_millis=heartbeat_millis,
         )
+
+        self._core_runtime = temporalio.bridge.runtime.Runtime(options=runtime_options)
         if isinstance(telemetry.metrics, MetricBuffer):
             telemetry.metrics._runtime = self
         core_meter = temporalio.bridge.metric.MetricMeter.create(self._core_runtime)
@@ -415,34 +419,6 @@ def _to_bridge_config(self) -> temporalio.bridge.runtime.TelemetryConfig:
         )
 
 
-@dataclass(frozen=True)
-class RuntimeOptions:
-    """Configuration for runtime initialization."""
-
-    telemetry: TelemetryConfig = field(default_factory=TelemetryConfig)
-    """Telemetry configuration applied to the runtime."""
-
-    worker_heartbeat_interval: Optional[timedelta] = timedelta(seconds=30)
-    """Interval for worker heartbeats. ``None`` disables heartbeating."""
-
-    def _to_bridge_config(self) -> temporalio.bridge.runtime.RuntimeOptions:
-        heartbeat_millis: Optional[int]
-        if self.worker_heartbeat_interval is None:
-            heartbeat_millis = None
-        else:
-            if self.worker_heartbeat_interval <= timedelta(0):
-                raise ValueError("worker_heartbeat_interval must be positive")
-            heartbeat_millis = int(
-                self.worker_heartbeat_interval.total_seconds() * 1000
-            )
-            if heartbeat_millis == 0:
-                heartbeat_millis = 1
-        return temporalio.bridge.runtime.RuntimeOptions(
-            telemetry=self.telemetry._to_bridge_config(),
-            worker_heartbeat_interval_millis=heartbeat_millis,
-        )
-
-
 BufferedMetricKind = NewType("BufferedMetricKind", int)
 """Representation of a buffered metric kind."""
 

temporalio/worker/_worker.py

@@ -559,6 +559,10 @@ def check_activity(activity):
                 maximum=config["max_concurrent_activity_task_polls"]
             )
 
+        worker_plugins = [plugin.name() for plugin in config.get("plugins", [])]
+        client_plugins = [plugin.name() for plugin in config["client"].plugins]
+        plugins = list(set(worker_plugins + client_plugins))
+
         # 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
@@ -613,7 +617,7 @@ def check_activity(activity):
                 nexus_task_poller_behavior=config[
                     "nexus_task_poller_behavior"
                 ]._to_bridge(),
-                plugins=[plugin.name() for plugin in config.get("plugins", [])],
+                plugins=plugins,
                 skip_client_worker_set_check=config["skip_client_worker_set_check"],
             ),
         )

tests/test_runtime.py

@@ -16,7 +16,6 @@
     LoggingConfig,
     PrometheusConfig,
     Runtime,
-    RuntimeOptions,
     TelemetryConfig,
     TelemetryFilter,
 )
@@ -265,20 +264,24 @@ async def check_metrics() -> None:
 
 
 def test_runtime_options_to_bridge_config() -> None:
-    assert (
-        RuntimeOptions()._to_bridge_config().worker_heartbeat_interval_millis == 30_000
+    runtime = Runtime(telemetry=TelemetryConfig())
+    assert runtime._heartbeat_millis == 30_000
+
+    runtime = Runtime(
+        telemetry=TelemetryConfig(),
+        worker_heartbeat_interval=timedelta(seconds=60),
     )
-    bridge_config = RuntimeOptions(
-        worker_heartbeat_interval=timedelta(seconds=60)
-    )._to_bridge_config()
-    assert bridge_config.worker_heartbeat_interval_millis == 60_000
+    assert runtime._heartbeat_millis == 60_000
 
-    bridge_config1 = RuntimeOptions(worker_heartbeat_interval=None)._to_bridge_config()
-    assert bridge_config1.worker_heartbeat_interval_millis is None
+    runtime = Runtime(
+        telemetry=TelemetryConfig(),
+        worker_heartbeat_interval=None,
+    )
+    assert runtime._heartbeat_millis is None
 
 
 def test_runtime_options_invalid_heartbeat() -> None:
     with pytest.raises(ValueError):
-        RuntimeOptions(
-            worker_heartbeat_interval=timedelta(seconds=-5)
-        )._to_bridge_config()
+        Runtime(
+            telemetry=TelemetryConfig(), worker_heartbeat_interval=timedelta(seconds=-5)
+        )

tests/worker/test_worker.py

@@ -32,7 +32,6 @@
 from temporalio.runtime import (
     PrometheusConfig,
     Runtime,
-    RuntimeOptions,
     TelemetryConfig,
 )
 from temporalio.service import RPCError

tests/worker/test_workflow.py

@@ -5302,7 +5302,7 @@ async def run(self) -> None:
 
 
 async def test_workflow_replace_worker_client(client: Client):
-    other_runtime = Runtime()
+    other_runtime = Runtime(telemetry=TelemetryConfig())
     other_client = await Client.connect(
         client.service_client.config.target_host,
         namespace=client.namespace,