diff --git a/changelog/fixed/talkative-tortoise.md b/changelog/fixed/talkative-tortoise.md new file mode 100644 index 000000000..3c7919cc3 --- /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 c8fca51e5..b5d70df08 100644 --- a/temporalio/bridge/src/runtime.rs +++ b/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, + 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, @@ -199,7 +218,7 @@ pub fn init_runtime(options: RuntimeOptions) -> PyResult { 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(); } } diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 0eca3edb3..30579121c 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"