From d8b718b28c764679d76ce45293536ab74dbf78d8 Mon Sep 17 00:00:00 2001 From: parbhatkapila4 Date: Wed, 21 Oct 2026 14:10:50 +0530 Subject: [PATCH 1/3] Keep log forwarding running when one client or worker on a runtime is dropped --- changelog/fixed/talkative-tortoise.md | 1 + temporalio/bridge/src/runtime.rs | 5 ++-- tests/test_runtime.py | 34 +++++++++++++++++++++++++++ 3 files changed, 38 insertions(+), 2 deletions(-) create mode 100644 changelog/fixed/talkative-tortoise.md diff --git a/changelog/fixed/talkative-tortoise.md b/changelog/fixed/talkative-tortoise.md new file mode 100644 index 0000000000..3c7919cc3c --- /dev/null +++ b/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`. diff --git a/temporalio/bridge/src/runtime.rs b/temporalio/bridge/src/runtime.rs index c8fca51e5c..5abb51eeca 100644 --- a/temporalio/bridge/src/runtime.rs +++ b/temporalio/bridge/src/runtime.rs @@ -239,8 +239,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(); } } diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 0eca3edb3a..30579121c2 100644 --- a/tests/test_runtime.py +++ b/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" From 803d7d70dcf8676cb7309bfc8ee7f3e40c5274b5 Mon Sep 17 00:00:00 2001 From: parbhatkapila4 Date: Thu, 22 Oct 2026 00:11:55 +0530 Subject: [PATCH 2/3] Release the GIL while the last runtime clone shuts down Tokio --- temporalio/bridge/src/runtime.rs | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/temporalio/bridge/src/runtime.rs b/temporalio/bridge/src/runtime.rs index 5abb51eeca..2de3d4c988 100644 --- a/temporalio/bridge/src/runtime.rs +++ b/temporalio/bridge/src/runtime.rs @@ -4,6 +4,7 @@ use pyo3::prelude::*; use pythonize::pythonize; use std::collections::HashMap; use std::future::Future; +use std::mem::ManuallyDrop; use std::net::SocketAddr; use std::pin::Pin; use std::str::FromStr; @@ -33,7 +34,7 @@ pub struct RuntimeRef { #[derive(Clone)] pub(crate) struct Runtime { pub(crate) pid: u32, - pub(crate) core: Arc, + pub(crate) core: ManuallyDrop>, metrics_call_buffer: Option>>, log_forwarder_handle: Option>>, } @@ -199,7 +200,7 @@ pub fn init_runtime(options: RuntimeOptions) -> PyResult { Ok(RuntimeRef { runtime: Runtime { pid: std::process::id(), - core: Arc::new(core), + core: ManuallyDrop::new(Arc::new(core)), metrics_call_buffer, log_forwarder_handle, }, @@ -244,6 +245,16 @@ impl Drop for Runtime { if let Some(handle) = self.log_forwarder_handle.take().and_then(Arc::into_inner) { handle.abort(); } + + // SAFETY: `core` is not used again after this + let core = unsafe { ManuallyDrop::take(&mut self.core) }; + if let Some(core) = Arc::into_inner(core) { + // Dropping the last clone shuts down Tokio, which waits for running tasks. + // The log forwarder may be one of them, waiting for the GIL, so release it. + let mut core = Some(core); + let _ = Python::try_attach(|py| py.detach(|| drop(core.take()))); + drop(core); + } } } From 27221b1928fdd26f44f5e2664c3dd95aa0e04ca9 Mon Sep 17 00:00:00 2001 From: parbhatkapila4 Date: Thu, 22 Oct 2026 01:21:16 +0530 Subject: [PATCH 3/3] Release the GIL in a Core runtime wrapper instead of using unsafe --- temporalio/bridge/src/runtime.rs | 34 ++++++++++++++++++++------------ 1 file changed, 21 insertions(+), 13 deletions(-) diff --git a/temporalio/bridge/src/runtime.rs b/temporalio/bridge/src/runtime.rs index 2de3d4c988..b5d70df08b 100644 --- a/temporalio/bridge/src/runtime.rs +++ b/temporalio/bridge/src/runtime.rs @@ -4,7 +4,6 @@ use pyo3::prelude::*; use pythonize::pythonize; use std::collections::HashMap; use std::future::Future; -use std::mem::ManuallyDrop; use std::net::SocketAddr; use std::pin::Pin; use std::str::FromStr; @@ -34,11 +33,30 @@ pub struct RuntimeRef { #[derive(Clone)] pub(crate) struct Runtime { pub(crate) pid: u32, - pub(crate) core: ManuallyDrop>, + pub(crate) core: Arc, metrics_call_buffer: Option>>, log_forwarder_handle: Option>>, } +/// 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); + +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, @@ -200,7 +218,7 @@ pub fn init_runtime(options: RuntimeOptions) -> PyResult { Ok(RuntimeRef { runtime: Runtime { pid: std::process::id(), - core: ManuallyDrop::new(Arc::new(core)), + core: Arc::new(CoreRuntimeHandle(Some(core))), metrics_call_buffer, log_forwarder_handle, }, @@ -245,16 +263,6 @@ impl Drop for Runtime { if let Some(handle) = self.log_forwarder_handle.take().and_then(Arc::into_inner) { handle.abort(); } - - // SAFETY: `core` is not used again after this - let core = unsafe { ManuallyDrop::take(&mut self.core) }; - if let Some(core) = Arc::into_inner(core) { - // Dropping the last clone shuts down Tokio, which waits for running tasks. - // The log forwarder may be one of them, waiting for the GIL, so release it. - let mut core = Some(core); - let _ = Python::try_attach(|py| py.detach(|| drop(core.take()))); - drop(core); - } } }