From a8dcb8c70cf7f3524d7501ccbefc11b38532de04 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Mon, 3 Aug 2026 21:58:29 -0400 Subject: [PATCH 1/2] fix(python): preserve panic settlement after loop shutdown fix Signed-off-by: Will Killian --- crates/python/src/py_api/mod.rs | 20 +++++++++++++++- .../tests/coverage/py_api_coverage_tests.rs | 24 +++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/crates/python/src/py_api/mod.rs b/crates/python/src/py_api/mod.rs index b2f055dd4..4736f0998 100644 --- a/crates/python/src/py_api/mod.rs +++ b/crates/python/src/py_api/mod.rs @@ -108,6 +108,16 @@ impl SafeFutureCompleter { } } +fn panic_message(panic: &(dyn std::any::Any + Send)) -> &str { + if let Some(message) = panic.downcast_ref::<&str>() { + message + } else if let Some(message) = panic.downcast_ref::() { + message.as_str() + } else { + "unknown error" + } +} + fn safe_future_into_py<'py, F>(py: Python<'py>, future: F) -> PyResult> where F: Future>> + Send + 'static, @@ -126,7 +136,15 @@ where let completion_future = python_future.clone_ref(py); pyo3_async_runtimes::tokio::get_runtime().spawn(async move { let result = tokio::select! { - result = pyo3_async_runtimes::tokio::scope(locals, future) => Some(result), + result = tokio::spawn(pyo3_async_runtimes::tokio::scope(locals, future)) => Some( + match result { + Ok(result) => result, + Err(error) if error.is_panic() => Err(pyo3_async_runtimes::err::RustPanic::new_err( + format!("rust future panicked: {}", panic_message(error.into_panic().as_ref())), + )), + Err(error) => Err(pyo3::exceptions::PyRuntimeError::new_err(error.to_string())), + }, + ), _ = &mut cancel_receiver => None, }; let Some(result) = result else { return }; diff --git a/crates/python/tests/coverage/py_api_coverage_tests.rs b/crates/python/tests/coverage/py_api_coverage_tests.rs index 5a449b960..d2ef89412 100644 --- a/crates/python/tests/coverage/py_api_coverage_tests.rs +++ b/crates/python/tests/coverage/py_api_coverage_tests.rs @@ -38,6 +38,30 @@ fn with_event_loop(py: Python<'_>, f: impl FnOnce(Bound<'_, PyAny>) -> T) -> result } +#[test] +fn safe_future_into_py_settles_rust_panics() { + let _python = crate::test_support::init_python_test(); + Python::attach(|py| { + with_event_loop(py, |event_loop| { + let locals = pyo3_async_runtimes::TaskLocals::new(event_loop.clone()); + let future = pyo3_async_runtimes::tokio::get_runtime() + .block_on(pyo3_async_runtimes::tokio::scope(locals, async move { + Python::attach(|py| { + safe_future_into_py(py, async move { panic!("expected test panic") }) + .map(Bound::unbind) + }) + })) + .unwrap(); + + let error = event_loop + .call_method1("run_until_complete", (future,)) + .unwrap_err(); + assert!(error.is_instance_of::(py)); + assert!(error.to_string().contains("expected test panic")); + }); + }); +} + #[test] fn py_api_helpers_and_scope_lifecycle_round_trip() { let _python = crate::test_support::init_python_test(); From ee6d71d0e302411c8db271d4fee8b630d9f2dffd Mon Sep 17 00:00:00 2001 From: Will Killian Date: Mon, 3 Aug 2026 22:18:22 -0400 Subject: [PATCH 2/2] fix(python): cancel closed-loop bridge tasks Signed-off-by: Will Killian --- crates/python/src/py_api/mod.rs | 9 +- .../tests/coverage/py_api_coverage_tests.rs | 114 ++++++++++++++++++ 2 files changed, 121 insertions(+), 2 deletions(-) diff --git a/crates/python/src/py_api/mod.rs b/crates/python/src/py_api/mod.rs index 4736f0998..d09933a0f 100644 --- a/crates/python/src/py_api/mod.rs +++ b/crates/python/src/py_api/mod.rs @@ -135,8 +135,9 @@ where )?; let completion_future = python_future.clone_ref(py); pyo3_async_runtimes::tokio::get_runtime().spawn(async move { + let mut task = tokio::spawn(pyo3_async_runtimes::tokio::scope(locals, future)); let result = tokio::select! { - result = tokio::spawn(pyo3_async_runtimes::tokio::scope(locals, future)) => Some( + result = &mut task => Some( match result { Ok(result) => result, Err(error) if error.is_panic() => Err(pyo3_async_runtimes::err::RustPanic::new_err( @@ -145,7 +146,11 @@ where Err(error) => Err(pyo3::exceptions::PyRuntimeError::new_err(error.to_string())), }, ), - _ = &mut cancel_receiver => None, + _ = &mut cancel_receiver => { + task.abort(); + let _ = task.await; + None + }, }; let Some(result) = result else { return }; Python::attach(|py| { diff --git a/crates/python/tests/coverage/py_api_coverage_tests.rs b/crates/python/tests/coverage/py_api_coverage_tests.rs index d2ef89412..cf87cbea1 100644 --- a/crates/python/tests/coverage/py_api_coverage_tests.rs +++ b/crates/python/tests/coverage/py_api_coverage_tests.rs @@ -6,6 +6,8 @@ use super::*; use std::ffi::CString; +use std::sync::mpsc; +use std::time::Duration; use pyo3::types::PyModule; use serde_json::json; @@ -38,6 +40,57 @@ fn with_event_loop(py: Python<'_>, f: impl FnOnce(Bound<'_, PyAny>) -> T) -> result } +fn test_loop(py: Python<'_>, closed: bool) -> Bound<'_, PyAny> { + let module = load_module( + py, + r#" +import threading + +class Future: + def __init__(self): + self._callbacks = [] + self._cancelled = False + + def add_done_callback(self, callback): + self._callbacks.append(callback) + + def cancelled(self): + return self._cancelled + + def cancel(self): + self._cancelled = True + for callback in self._callbacks: + callback(self) + +class Loop: + def __init__(self, closed): + self.closed = closed + self.closed_checked = threading.Event() + self.completion_scheduled = False + + def create_future(self): + return Future() + + def is_closed(self): + self.closed_checked.set() + return self.closed + + def call_soon_threadsafe(self, callback): + self.completion_scheduled = True + callback() +"#, + ); + module.getattr("Loop").unwrap().call1((closed,)).unwrap() +} + +struct CancellationSignal(mpsc::Sender<()>); + +impl Drop for CancellationSignal { + fn drop(&mut self) { + let _ = self.0.send(()); + } +} + #[test] fn safe_future_into_py_settles_rust_panics() { let _python = crate::test_support::init_python_test(); @@ -62,6 +115,67 @@ fn safe_future_into_py_settles_rust_panics() { }); } +#[test] +fn safe_future_into_py_cancels_rust_work() { + let _python = crate::test_support::init_python_test(); + Python::attach(|py| { + let event_loop = test_loop(py, false); + let locals = pyo3_async_runtimes::TaskLocals::new(event_loop.clone()); + let (started_tx, started_rx) = mpsc::sync_channel(1); + let (dropped_tx, dropped_rx) = mpsc::channel(); + let future = pyo3_async_runtimes::tokio::get_runtime() + .block_on(pyo3_async_runtimes::tokio::scope(locals, async move { + Python::attach(|py| { + safe_future_into_py(py, async move { + started_tx.send(()).unwrap(); + let _cancellation_signal = CancellationSignal(dropped_tx); + std::future::pending::>>().await + }) + .map(Bound::unbind) + }) + })) + .unwrap(); + + assert!(started_rx.recv_timeout(Duration::from_secs(1)).is_ok()); + future.bind(py).call_method0("cancel").unwrap(); + assert!(dropped_rx.recv_timeout(Duration::from_secs(1)).is_ok()); + }); +} + +#[test] +fn safe_future_into_py_skips_completion_on_closed_loop() { + let _python = crate::test_support::init_python_test(); + Python::attach(|py| { + let event_loop = test_loop(py, true); + let locals = pyo3_async_runtimes::TaskLocals::new(event_loop.clone()); + let _future = pyo3_async_runtimes::tokio::get_runtime() + .block_on(pyo3_async_runtimes::tokio::scope(locals, async move { + Python::attach(|py| { + safe_future_into_py(py, async move { Python::attach(|py| Ok(py.None())) }) + .map(Bound::unbind) + }) + })) + .unwrap(); + + assert!( + event_loop + .getattr("closed_checked") + .unwrap() + .call_method1("wait", (1.0,)) + .unwrap() + .is_truthy() + .unwrap() + ); + assert!( + !event_loop + .getattr("completion_scheduled") + .unwrap() + .is_truthy() + .unwrap() + ); + }); +} + #[test] fn py_api_helpers_and_scope_lifecycle_round_trip() { let _python = crate::test_support::init_python_test();