Keep log forwarding running when one client or worker on a runtime is dropped

changelog/fixed/talkative-tortoise.md

@@ -0,0 +1 @@
+Dropping a client or worker no longer stops log forwarding for the other clients and workers that share its `Runtime`.

temporalio/bridge/src/runtime.rs

@@ -33,11 +33,30 @@ pub struct RuntimeRef {
 #[derive(Clone)]
 pub(crate) struct Runtime {
     pub(crate) pid: u32,
-    pub(crate) core: Arc<CoreRuntime>,
+    pub(crate) core: Arc<CoreRuntimeHandle>,
     metrics_call_buffer: Option<Arc<MetricsCallBuffer<BufferedMetricRef>>>,
     log_forwarder_handle: Option<Arc<JoinHandle<()>>>,
 }
 
+/// Owns the Core runtime so the last runtime clone can shut it down with the GIL released.
+/// Shutting down waits for running Tokio tasks, and the log forwarder may be one of them,
+/// waiting for the GIL.
+pub(crate) struct CoreRuntimeHandle(Option<CoreRuntime>);
+
+impl std::ops::Deref for CoreRuntimeHandle {
+    type Target = CoreRuntime;
+
+    fn deref(&self) -> &CoreRuntime {
+        self.0.as_ref().expect("Core runtime used after drop")
+    }
+}
+
+impl Drop for CoreRuntimeHandle {
+    fn drop(&mut self) {
+        let _ = Python::try_attach(|py| py.detach(|| drop(self.0.take())));
+    }
+}
+
 #[derive(FromPyObject)]
 pub struct TelemetryConfig {
     logging: Option<LoggingConfig>,
@@ -199,7 +218,7 @@ pub fn init_runtime(options: RuntimeOptions) -> PyResult<RuntimeRef> {
     Ok(RuntimeRef {
         runtime: Runtime {
             pid: std::process::id(),
-            core: Arc::new(core),
+            core: Arc::new(CoreRuntimeHandle(Some(core))),
             metrics_call_buffer,
             log_forwarder_handle,
         },
@@ -239,8 +258,9 @@ impl Runtime {
 
 impl Drop for Runtime {
     fn drop(&mut self) {
-        // Stop the log forwarder
-        if let Some(handle) = self.log_forwarder_handle.as_ref() {
+        // Clients and workers each hold a clone of the runtime, so only stop the
+        // log forwarder when the last clone is dropped
+        if let Some(handle) = self.log_forwarder_handle.take().and_then(Arc::into_inner) {
             handle.abort();
         }
     }

tests/test_runtime.py

@@ -1,3 +1,4 @@
+import gc
 import logging
 import logging.handlers
 import queue
@@ -144,6 +145,39 @@ async def log_queue_len() -> int:
         )
 
 
+async def test_runtime_log_forwarding_after_client_dropped(env: WorkflowEnvironment):
+    log_queue: queue.Queue[logging.LogRecord] = queue.Queue()
+    log_queue_list = cast(list[logging.LogRecord], log_queue.queue)
+    handler = logging.handlers.QueueHandler(log_queue)
+    logger = logging.getLogger(f"log-{uuid.uuid4()}")
+    logger.setLevel(logging.INFO)
+
+    async def log_queue_len() -> int:
+        return len(log_queue_list)
+
+    runtime = Runtime(
+        telemetry=TelemetryConfig(
+            logging=LoggingConfig(
+                filter=TelemetryFilter(core_level="INFO", other_level="ERROR"),
+                forwarding=LogForwardingConfig(logger=logger),
+            )
+        )
+    )
+
+    # Every client holds its own handle to the runtime, so dropping one client
+    # must not stop log forwarding for everything else on the runtime
+    client = await env.connect_client(runtime=runtime)
+    del client
+    gc.collect()
+
+    with LogHandler.apply(logger, handler):
+        runtime._core_runtime.write_test_info_log("info1", "extra1")
+        await assert_eq_eventually(1, log_queue_len)
+        assert log_queue_list[0].message.startswith(
+            "[sdk_core::temporal_sdk_bridge::runtime] info1"
+        )
+
+
 def test_runtime_console_logging_format():
     config = LoggingConfig(filter="INFO", format=LoggingFormat.JSON)._to_bridge_config()
     assert config.format == "json"