Fix and simplify plumbing plugins from WorkerConfig to core, impose runtime client identity requirement

yuandrew · yuandrew · commit 2484ef7b5477 · 2025-10-20T17:55:31.000-07:00
diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock
diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml
@@ -19,7 +19,7 @@ crate-type = ["cdylib"]
 anyhow = "1.0"
 async-trait = "0.1"
 futures = "0.3"
-prost = "0.13"
+prost = "0.14"
 pyo3 = { version = "0.25", features = [
     "extension-module",
     "abi3-py39",
@@ -38,7 +38,7 @@ temporal-sdk-core-api = { version = "0.1.0", path = "./sdk-core/core-api", featu
 temporal-sdk-core-protos = { version = "0.1.0", path = "./sdk-core/sdk-core-protos" }
 tokio = "1.26"
 tokio-stream = "0.1"
-tonic = "0.13"
+tonic = "0.14"
 tracing = "0.1"
 url = "2.2"
 
diff --git a/temporalio/bridge/runtime.py b/temporalio/bridge/runtime.py
@@ -96,7 +96,7 @@ class RuntimeOptions:
     """Python representation of the Rust struct for runtime options."""
 
     telemetry: TelemetryConfig
-    worker_heartbeat_interval_millis: Optional[int] = 30000 # 30s
+    worker_heartbeat_interval_millis: Optional[int] = 30_000  # 30s
 
 
 # WARNING: This must match Rust runtime::BufferedLogEntry
diff --git a/temporalio/bridge/sdk-core b/temporalio/bridge/sdk-core
@@ -1 +1 @@
-Subproject commit 401a38506b8b5780c8c6dd87016c825b05c2a70e
+Subproject commit 9e9a46191656fc9ccd95589dac3552410561d620
diff --git a/temporalio/bridge/src/client.rs b/temporalio/bridge/src/client.rs
@@ -6,18 +6,19 @@ use std::time::Duration;
 use temporal_client::{
     ClientKeepAliveConfig as CoreClientKeepAliveConfig, ClientOptions, ClientOptionsBuilder,
     ConfiguredClient, HttpConnectProxyOptions, RetryClient, RetryConfig,
-    TemporalServiceClientWithMetrics, TlsConfig,
+    TlsConfig, TemporalServiceClient,
 };
-use tonic::metadata::{
-    AsciiMetadataKey, AsciiMetadataValue, BinaryMetadataKey, BinaryMetadataValue,
+use temporal_client::tonic::{
+    self,
+    metadata::{AsciiMetadataKey, AsciiMetadataValue, BinaryMetadataKey, BinaryMetadataValue},
 };
 use url::Url;
 
 use crate::runtime;
 
 pyo3::create_exception!(temporal_sdk_bridge, RPCError, PyException);
 
-type Client = RetryClient<ConfiguredClient<TemporalServiceClientWithMetrics>>;
+type Client = RetryClient<ConfiguredClient<TemporalServiceClient>>;
 
 #[pyclass]
 pub struct ClientRef {
diff --git a/temporalio/worker/_replayer.py b/temporalio/worker/_replayer.py
@@ -80,7 +80,6 @@ def __init__(
             runtime=runtime,
             disable_safe_workflow_eviction=disable_safe_workflow_eviction,
             header_codec_behavior=header_codec_behavior,
-            plugins=[plugin.name() for plugin in plugins],
         )
 
         # Apply plugin configuration
@@ -294,7 +293,7 @@ def on_eviction_hook(
                     nexus_task_poller_behavior=temporalio.bridge.worker.PollerBehaviorSimpleMaximum(
                         1
                     ),
-                    plugins=self._config.get("plugins", []),
+                    plugins=[plugin.name() for plugin in self.plugins],
                     skip_client_worker_set_check=True,
                 ),
             )
@@ -371,7 +370,6 @@ class ReplayerConfig(TypedDict, total=False):
     runtime: Optional[temporalio.runtime.Runtime]
     disable_safe_workflow_eviction: bool
     header_codec_behavior: HeaderCodecBehavior
-    plugins: Sequence[str]
 
 
 @dataclass(frozen=True)
diff --git a/temporalio/worker/_worker.py b/temporalio/worker/_worker.py
@@ -381,7 +381,7 @@ def __init__(
                     f"The same plugin type {type(client_plugin)} is present from both client and worker. It may run twice and may not be the intended behavior."
                 )
         plugins = plugins_from_client + list(plugins)
-        config["plugins"] = [plugin.name() for plugin in plugins]
+        config["plugins"] = plugins
 
         self.plugins = plugins
         for plugin in plugins:
@@ -615,7 +615,7 @@ def check_activity(activity):
                 nexus_task_poller_behavior=config[
                     "nexus_task_poller_behavior"
                 ]._to_bridge(),
-                plugins=config.get("plugins", []),
+                plugins=[plugin.name() for plugin in config.get("plugins", [])],
                 skip_client_worker_set_check=config["skip_client_worker_set_check"],
             ),
         )
@@ -629,7 +629,6 @@ def config(self) -> WorkerConfig:
         config = self._config.copy()
         config["activities"] = list(config.get("activities", []))
         config["workflows"] = list(config.get("workflows", []))
-        config["plugins"] = list(config.get("plugins", []))
         return config
 
     @property
@@ -911,7 +910,7 @@ class WorkerConfig(TypedDict, total=False):
     workflow_task_poller_behavior: PollerBehavior
     activity_task_poller_behavior: PollerBehavior
     nexus_task_poller_behavior: PollerBehavior
-    plugins: Sequence[str]
+    plugins: Sequence[Plugin]
     skip_client_worker_set_check: bool
 
 
diff --git a/tests/test_plugins.py b/tests/test_plugins.py
@@ -150,15 +150,15 @@ async def test_worker_plugin_basic_config(client: Client) -> None:
         plugins=[MyWorkerPlugin()],
     )
     assert worker.config().get("task_queue") == "replaced_queue"
-    assert worker.config().get("plugins") == [MyWorkerPlugin().name()]
+    assert [p.name() for p in worker.config().get("plugins")] == [MyWorkerPlugin().name()]
 
     # Test client plugin propagation to worker plugins
     new_config = client.config()
     new_config["plugins"] = [MyCombinedPlugin()]
     client = Client(**new_config)
     worker = Worker(client, task_queue="queue", activities=[never_run_activity])
     assert worker.config().get("task_queue") == "combined"
-    assert worker.config().get("plugins") == [MyCombinedPlugin().name()]
+    assert [p.name() for p in worker.config().get("plugins")] == [MyCombinedPlugin().name()]
 
     # Test both. Client propagated plugins are called first, so the worker plugin overrides in this case
     worker = Worker(
@@ -168,7 +168,7 @@ async def test_worker_plugin_basic_config(client: Client) -> None:
         plugins=[MyWorkerPlugin()],
     )
     assert worker.config().get("task_queue") == "replaced_queue"
-    assert worker.config().get("plugins") == [
+    assert [p.name() for p in worker.config().get("plugins")] == [
         MyCombinedPlugin().name(),
         MyWorkerPlugin().name(),
     ]
@@ -307,7 +307,6 @@ async def test_replay(client: Client) -> None:
     replayer = Replayer(workflows=[], plugins=[plugin])
     assert len(replayer.config().get("workflows") or []) == 1
     assert replayer.config().get("data_converter") == pydantic_data_converter
-    assert replayer.config().get("plugins") == [plugin.name()]
 
     await replayer.replay_workflow(await handle.fetch_history())
 
diff --git a/tests/test_runtime.py b/tests/test_runtime.py
@@ -21,7 +21,7 @@
     TelemetryFilter,
 )
 from temporalio.worker import Worker
-from tests.helpers import assert_eq_eventually, assert_eventually, find_free_port
+from tests.helpers import assert_eq_eventually, assert_eventually, find_free_port, worker_versioning_enabled
 
 
 @workflow.defn
@@ -260,10 +260,12 @@ async def check_metrics() -> None:
 
 
 def test_runtime_options_to_bridge_config() -> None:
-    assert RuntimeOptions()._to_bridge_config().worker_heartbeat_interval_millis is None
-    options = RuntimeOptions(worker_heartbeat_interval=timedelta(seconds=30))
-    bridge_config = options._to_bridge_config()
-    assert bridge_config.worker_heartbeat_interval_millis == 30_000
+    assert RuntimeOptions()._to_bridge_config().worker_heartbeat_interval_millis == 30_000
+    bridge_config = RuntimeOptions(worker_heartbeat_interval=timedelta(seconds=60))._to_bridge_config()
+    assert bridge_config.worker_heartbeat_interval_millis == 60_000
+
+    bridge_config1 = RuntimeOptions(worker_heartbeat_interval=None)._to_bridge_config()
+    assert bridge_config1.worker_heartbeat_interval_millis is None
 
 
 def test_runtime_options_invalid_heartbeat() -> None:
diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py
@@ -1143,7 +1143,7 @@ async def test_can_run_autoscaling_polling_worker(
         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]
+        workflow_pollers = [l for l in matches if "workflow_task" in l and w.task_queue 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.
diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py
@@ -5302,57 +5302,19 @@ async def run(self) -> None:
             await asyncio.sleep(0.1)
 
 
-async def test_workflow_replace_worker_client(client: Client, env: WorkflowEnvironment):
-    if env.supports_time_skipping:
-        pytest.skip("Only testing against two real servers")
-    # We are going to start a second ephemeral server and then replace the
-    # client. So we will start a no-cache ticking workflow with the current
-    # client and confirm it has accomplished at least one task. Then we will
-    # start another on the other client, and confirm it gets started too. Then
-    # we will terminate both. We have to use a ticking workflow with only one
-    # poller to force a quick re-poll to recognize our client change quickly (as
-    # opposed to just waiting the minute for poll timeout).
-    async with await WorkflowEnvironment.start_local(
-        dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION
-    ) as other_env:
-        # Start both workflows on different servers
-        task_queue = f"tq-{uuid.uuid4()}"
-        handle1 = await client.start_workflow(
-            TickingWorkflow.run, id=f"wf-{uuid.uuid4()}", task_queue=task_queue
-        )
-        handle2 = await other_env.client.start_workflow(
-            TickingWorkflow.run, id=f"wf-{uuid.uuid4()}", task_queue=task_queue
-        )
-
-        async def any_task_completed(handle: WorkflowHandle) -> bool:
-            async for e in handle.fetch_history_events():
-                if e.HasField("workflow_task_completed_event_attributes"):
-                    return True
-            return False
-
-        # Now start the worker on the first env
-        async with Worker(
-            client,
-            task_queue=task_queue,
-            workflows=[TickingWorkflow],
-            max_cached_workflows=0,
-            max_concurrent_workflow_task_polls=1,
-        ) as worker:
-            # Confirm the first ticking workflow has completed a task but not
-            # the second
-            await assert_eq_eventually(True, lambda: any_task_completed(handle1))
-            assert not await any_task_completed(handle2)
-
-            # Now replace the client, which should be used fairly quickly
-            # because we should have timer-done poll completions every 100ms
-            worker.client = other_env.client
-
-            # Now confirm the other workflow has started
-            await assert_eq_eventually(True, lambda: any_task_completed(handle2))
-
-            # Terminate both
-            await handle1.terminate()
-            await handle2.terminate()
+async def test_workflow_replace_worker_client(client: Client):
+    other_runtime = Runtime()
+    other_client = await Client.connect(
+        client.service_client.config.target_host,
+        namespace=client.namespace,
+        runtime=other_runtime,
+    )
+    async with new_worker(client, HelloWorkflow) as worker:
+        with pytest.raises(
+            ValueError,
+            match="New client is not on the same runtime as the existing client",
+        ):
+            worker.client = other_client
 
 
 @activity.defn(dynamic=True)

Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,6 @@ def __init__(`
`80`	`80`	`runtime=runtime,`
`81`	`81`	`disable_safe_workflow_eviction=disable_safe_workflow_eviction,`
`82`	`82`	`header_codec_behavior=header_codec_behavior,`
`83`		`- plugins=[plugin.name() for plugin in plugins],`
`84`	`83`	`)`
`85`	`84`
`86`	`85`	`# Apply plugin configuration`
`@@ -294,7 +293,7 @@ def on_eviction_hook(`
`294`	`293`	`nexus_task_poller_behavior=temporalio.bridge.worker.PollerBehaviorSimpleMaximum(`
`295`	`294`	`1`
`296`	`295`	`),`
`297`		`- plugins=self._config.get("plugins", []),`
	`296`	`+ plugins=[plugin.name() for plugin in self.plugins],`
`298`	`297`	`skip_client_worker_set_check=True,`
`299`	`298`	`),`
`300`	`299`	`)`
`@@ -371,7 +370,6 @@ class ReplayerConfig(TypedDict, total=False):`
`371`	`370`	`runtime: Optional[temporalio.runtime.Runtime]`
`372`	`371`	`disable_safe_workflow_eviction: bool`
`373`	`372`	`header_codec_behavior: HeaderCodecBehavior`
`374`		`- plugins: Sequence[str]`
`375`	`373`
`376`	`374`
`377`	`375`	`@dataclass(frozen=True)`