From 346b113e3387bf5e49f6fe215a97c5ea85039a99 Mon Sep 17 00:00:00 2001 From: solidiquis Date: Tue, 29 Sep 2026 11:52:50 -0700 Subject: [PATCH 1/4] rust(bug): fix python bindings segfault --- python/CHANGELOG.md | 4 + python/pyproject.toml | 14 +- rust/CHANGELOG.md | 13 ++ rust/crates/sift_stream_bindings/Cargo.toml | 2 +- rust/crates/sift_stream_bindings/src/lib.rs | 46 +++++ .../sift_stream_bindings/src/runtime.rs | 169 +++++++++++++++++ .../src/stream/auto_register.rs | 2 +- .../src/stream/builder.rs | 8 +- .../sift_stream_bindings/src/stream/mod.rs | 2 +- .../tests/test_runtime_shutdown.py | 179 ++++++++++++++++++ 10 files changed, 425 insertions(+), 14 deletions(-) create mode 100644 rust/crates/sift_stream_bindings/src/runtime.rs create mode 100644 rust/crates/sift_stream_bindings/tests/test_runtime_shutdown.py diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index 1e016f978d..7b544ba3da 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -3,6 +3,10 @@ All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/). +## [v0.22.1] - September 29, 2026 + +- Fix streaming ingestion processes crashing with SIGSEGV or SIGABRT at interpreter exit. The `sift-stream-bindings` tokio runtime was never shut down, so a runtime thread that finished work while Python was finalizing re-entered the interpreter and killed the process after the program had completed. The bindings now stop the runtime from an `atexit` hook, and expose `shutdown(timeout=5.0)` for callers that exit with `os._exit()`. Requires `sift-stream-bindings` 0.5.2. + ## [v0.22.0] - September 25, 2026 - Added `client.user_defined_functions` for reusable expressions. diff --git a/python/pyproject.toml b/python/pyproject.toml index 6a2fdfb484..4bd414520e 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "sift_stack_py" -version = "0.22.0" +version = "0.22.1" description = "Python client library for the Sift API" requires-python = ">=3.8" readme = { file = "README.md", content-type = "text/markdown" } @@ -71,7 +71,7 @@ all = [ 'pyarrow>=17.0.0', 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", - 'sift-stream-bindings==0.5.1', + 'sift-stream-bindings==0.5.2', 'types-pyOpenSSL<24.0.0', ] build = [ @@ -119,7 +119,7 @@ dev-all = [ 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", 'ruff~=0.12.10', - 'sift-stream-bindings==0.5.1', + 'sift-stream-bindings==0.5.2', 'tomlkit~=0.13.3', 'types-pyOpenSSL<24.0.0', ] @@ -179,7 +179,7 @@ docs-build = [ 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", 'ruff~=0.12.10', - 'sift-stream-bindings==0.5.1', + 'sift-stream-bindings==0.5.2', 'tomlkit~=0.13.3', 'types-pyOpenSSL<24.0.0', ] @@ -209,10 +209,10 @@ rosbags = [ "rosbags~=0.0 ; python_full_version >= '3.8.2'", ] sift-stream = [ - 'sift-stream-bindings==0.5.1', + 'sift-stream-bindings==0.5.2', ] sift-stream-bindings = [ - 'sift-stream-bindings==0.5.1', + 'sift-stream-bindings==0.5.2', ] tdms = [ 'npTDMS~=1.9', @@ -260,7 +260,7 @@ docs = [ openssl = ["pyOpenSSL<24.0.0", "types-pyOpenSSL<24.0.0", "cffi~=1.14"] tdms = ["npTDMS~=1.9"] rosbags = ["rosbags~=0.0 ; python_full_version >= '3.8.2'"] -sift-stream = ["sift-stream-bindings==0.5.1"] +sift-stream = ["sift-stream-bindings==0.5.2"] hdf5 = ["h5py~=3.11", "polars~=1.8"] # polars is only used by sift_py; remove once sift_py is fully deprecated ulog = ["pyulog~=1.2.2"] mcap = ["mcap~=1.4", "mcap-ros2-support~=0.5.7"] diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index 622bcf318f..f5f4ace557 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -16,6 +16,19 @@ the word "link". ### Bug Fixes +#### Runtime threads no longer crash the Python process at interpreter exit + +##### `sift-stream-bindings` 0.5.2 + +The bindings ran every async call on a process-global tokio runtime that was never shut down. +A worker that finished work while CPython was finalizing re-entered Python and killed the +process with SIGSEGV or SIGABRT after the program had already completed. The crate now owns its +runtime and registers a new `shutdown(timeout=5.0)` function with `atexit`, so the runtime is +stopped and its threads joined while the interpreter is still intact. `shutdown()` can also be +called directly, for example before `os._exit()`, and `is_shut_down()` reports whether it has +run. Calls that need the runtime raise `RuntimeError` afterwards. Runtime threads are now named +`sift-stream-worker`. + #### Asset-name check is now case-insensitive `SiftStream` no longer rejects an ingestion config whose asset name differs from the one in Sift diff --git a/rust/crates/sift_stream_bindings/Cargo.toml b/rust/crates/sift_stream_bindings/Cargo.toml index cf09e4655c..0415506bc2 100644 --- a/rust/crates/sift_stream_bindings/Cargo.toml +++ b/rust/crates/sift_stream_bindings/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "sift-stream-bindings" -version = "0.5.1" +version = "0.5.2" # Released to PyPI as a wheel, never to crates.io. publish = false edition = { workspace = true } diff --git a/rust/crates/sift_stream_bindings/src/lib.rs b/rust/crates/sift_stream_bindings/src/lib.rs index dc87d67caa..4b0149f137 100644 --- a/rust/crates/sift_stream_bindings/src/lib.rs +++ b/rust/crates/sift_stream_bindings/src/lib.rs @@ -1,11 +1,14 @@ +use pyo3::exceptions::PyValueError; use pyo3::prelude::*; use pyo3_stub_gen::define_stub_info_gatherer; use std::sync::{Mutex, Once}; +use std::time::Duration; use tracing::Level; use tracing_subscriber::{Layer, filter, layer::SubscriberExt}; mod error; mod metrics; +mod runtime; mod sift; mod stream; @@ -135,6 +138,40 @@ fn is_tracing_initialized() -> bool { INIT_TRACING.is_completed() } +/// Stop the background runtime that runs this library's async work and wait up to +/// `timeout` seconds for its threads to finish. +/// +/// The module registers this with `atexit` when it is imported, so a normal interpreter +/// shutdown calls it for you. Without it, runtime threads that complete work while the +/// interpreter is finalizing re-enter Python and crash the process (SIGSEGV or SIGABRT). +/// `atexit` handlers registered after this module was imported run before it, so they can +/// still finish a stream. Call it yourself to stop the runtime earlier, for example right +/// before `os._exit()`. +/// +/// Finish your streams first. In-flight work is cancelled, awaitables that have not +/// completed never resolve, and any later call that needs the runtime raises `RuntimeError`. +/// Calling this more than once is harmless. +/// +/// Args: +/// timeout: Seconds to wait for in-flight work before giving up (default: 5.0) +#[pyfunction] +#[pyo3(signature = (timeout = runtime::DEFAULT_SHUTDOWN_TIMEOUT_SECS))] +fn shutdown(py: Python<'_>, timeout: f64) -> PyResult<()> { + if !timeout.is_finite() || timeout < 0.0 { + return Err(PyValueError::new_err( + "timeout must be a non-negative number of seconds", + )); + } + runtime::shutdown(py, Duration::from_secs_f64(timeout)); + Ok(()) +} + +/// Whether `shutdown()` has run. Calls that need the runtime raise `RuntimeError` after it. +#[pyfunction] +fn is_shut_down() -> bool { + runtime::is_shut_down() +} + // Cannot organize into submodules right now // See below for issues with submodules using pyo3 // https://github.com/PyO3/pyo3/issues/759 @@ -180,5 +217,14 @@ fn sift_stream_bindings(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_function(wrap_pyfunction!(init_tracing, m)?)?; m.add_function(wrap_pyfunction!(init_tracing_with_file, m)?)?; m.add_function(wrap_pyfunction!(is_tracing_initialized, m)?)?; + m.add_function(wrap_pyfunction!(is_shut_down, m)?)?; + + // Stop the runtime while the interpreter is still intact. `atexit` runs before CPython + // marks itself finalizing, so runtime threads never re-enter Python after that point. + let shutdown_fn = wrap_pyfunction!(shutdown, m)?; + m.add_function(shutdown_fn.clone())?; + m.py() + .import("atexit")? + .call_method1("register", (shutdown_fn,))?; Ok(()) } diff --git a/rust/crates/sift_stream_bindings/src/runtime.rs b/rust/crates/sift_stream_bindings/src/runtime.rs new file mode 100644 index 0000000000..8f36412dd3 --- /dev/null +++ b/rust/crates/sift_stream_bindings/src/runtime.rs @@ -0,0 +1,169 @@ +//! The tokio runtime that backs every awaitable this crate hands to Python. +//! +//! `pyo3_async_runtimes::tokio` keeps its runtime in a `OnceLock` with no way to +//! shut it down, so its worker threads stay alive through interpreter +//! finalization. A worker that completes a future after CPython has torn down +//! its thread-state bookkeeping re-enters Python to resolve the asyncio future +//! and crashes the process: SIGSEGV once `PyGILState_Ensure` dereferences the +//! freed interpreter state, or SIGABRT when CPython calls `pthread_exit` on a +//! thread that has Rust frames on its stack. +//! +//! This module owns the runtime instead. [`shutdown`] stops it and joins its +//! threads while the interpreter is still intact, and the module init registers +//! it with `atexit` so normal interpreter shutdown always runs it. Everything +//! else (task locals, resolving the asyncio future, cancellation) still comes +//! from `pyo3_async_runtimes::generic`, driven through [`SiftRuntime`]. + +use pyo3::exceptions::PyRuntimeError; +use pyo3::prelude::*; +use pyo3_async_runtimes::TaskLocals; +use pyo3_async_runtimes::generic::{self, ContextExt, Runtime as GenericRuntime}; +use std::cell::OnceCell; +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Mutex, OnceLock, PoisonError}; +use std::time::Duration; +use tokio::runtime::{Builder, Handle, Runtime}; + +/// How long [`shutdown`] waits for in-flight work when the caller gives no timeout. +pub const DEFAULT_SHUTDOWN_TIMEOUT_SECS: f64 = 5.0; + +/// The runtime itself. `None` before first use and again after [`shutdown`]. +static RUNTIME: Mutex> = Mutex::new(None); +/// Handle used to spawn work. Set once, when the runtime is first built. +static HANDLE: OnceLock = OnceLock::new(); +/// Set by [`shutdown`] before it drains `RUNTIME`; never cleared. +static SHUT_DOWN: AtomicBool = AtomicBool::new(false); + +tokio::task_local! { + static TASK_LOCALS: OnceCell; +} + +/// Marker type that plugs this crate's runtime into `pyo3_async_runtimes::generic`. +pub(crate) struct SiftRuntime; + +impl GenericRuntime for SiftRuntime { + type JoinError = tokio::task::JoinError; + type JoinHandle = tokio::task::JoinHandle<()>; + + fn spawn(fut: F) -> Self::JoinHandle + where + F: Future + Send + 'static, + { + handle().spawn(fut) + } + + fn spawn_blocking(f: F) -> Self::JoinHandle + where + F: FnOnce() + Send + 'static, + { + handle().spawn_blocking(f) + } +} + +impl ContextExt for SiftRuntime { + fn scope(locals: TaskLocals, fut: F) -> Pin + Send>> + where + F: Future + Send + 'static, + { + let cell = OnceCell::new(); + cell.set(locals) + .unwrap_or_else(|_| unreachable!("fresh OnceCell is empty")); + Box::pin(TASK_LOCALS.scope(cell, fut)) + } + + fn get_task_locals() -> Option { + TASK_LOCALS + .try_with(|c| c.get().cloned()) + .unwrap_or_default() + } +} + +/// Converts a Rust future into a Python awaitable that runs on this crate's runtime. +/// +/// Drop-in replacement for `pyo3_async_runtimes::tokio::future_into_py`. Raises +/// `RuntimeError` once [`shutdown`] has run, since there is no runtime left to +/// drive the future. +pub(crate) fn future_into_py<'py, F, T>(py: Python<'py>, fut: F) -> PyResult> +where + F: Future> + Send + 'static, + T: for<'a> IntoPyObject<'a> + Send + 'static, +{ + ensure_started()?; + let locals = generic::get_current_locals::(py)?; + generic::future_into_py_with_locals::(py, locals, fut) +} + +/// Whether [`shutdown`] has run. +pub(crate) fn is_shut_down() -> bool { + SHUT_DOWN.load(Ordering::SeqCst) +} + +/// Stops the runtime and waits up to `timeout` for its threads to finish. +/// +/// Work still in flight is cancelled at its next await point, awaitables that +/// have not completed never resolve, and every later call that needs the +/// runtime raises `RuntimeError`. The GIL is released while waiting so that a +/// task which is already resolving a Python future can finish. Calling this more +/// than once is harmless. +pub(crate) fn shutdown(py: Python<'_>, timeout: Duration) { + SHUT_DOWN.store(true, Ordering::SeqCst); + let runtime = RUNTIME + .lock() + .unwrap_or_else(PoisonError::into_inner) + .take(); + if let Some(runtime) = runtime { + py.detach(|| runtime.shutdown_timeout(timeout)); + } +} + +fn shut_down_err() -> PyErr { + PyRuntimeError::new_err( + "the sift_stream_bindings runtime has been shut down; it stops at interpreter exit \ + or when sift_stream_bindings.shutdown() is called", + ) +} + +/// Builds the runtime on first use. Fails with `RuntimeError` after [`shutdown`]. +fn ensure_started() -> PyResult<()> { + if is_shut_down() { + return Err(shut_down_err()); + } + if HANDLE.get().is_some() { + return Ok(()); + } + + let mut slot = RUNTIME.lock().unwrap_or_else(PoisonError::into_inner); + if HANDLE.get().is_some() { + // Another thread built it while we waited for the lock. + return Ok(()); + } + let runtime = Builder::new_multi_thread() + .enable_all() + .thread_name("sift-stream-worker") + .build() + .map_err(|e| { + PyRuntimeError::new_err(format!( + "failed to start the sift_stream_bindings runtime: {e}" + )) + })?; + let _ = HANDLE.set(runtime.handle().clone()); + // `shutdown` sets the flag before draining the slot, so if it ran while we + // were building, drop this runtime now rather than leave its threads alive. + if is_shut_down() { + drop(slot); + drop(runtime); + return Err(shut_down_err()); + } + *slot = Some(runtime); + Ok(()) +} + +/// The spawn handle. Every caller goes through [`future_into_py`], which runs +/// [`ensure_started`] first, so the handle exists by the time this is reached. +fn handle() -> &'static Handle { + HANDLE + .get() + .expect("sift_stream_bindings runtime is started before work is spawned on it") +} diff --git a/rust/crates/sift_stream_bindings/src/stream/auto_register.rs b/rust/crates/sift_stream_bindings/src/stream/auto_register.rs index ed139fa04e..e54866be10 100644 --- a/rust/crates/sift_stream_bindings/src/stream/auto_register.rs +++ b/rust/crates/sift_stream_bindings/src/stream/auto_register.rs @@ -1,9 +1,9 @@ use super::{FlowPy, SiftStreamInner, SiftStreamPy, stream_consumed_err}; use crate::error::SiftErrorWrapper; use crate::metrics::SiftStreamMetricsSnapshotPy; +use crate::runtime::future_into_py; use crate::stream::config::{FlowConfigPy, FlowDescriptorPy, RunSelectorPy}; use pyo3::prelude::*; -use pyo3_async_runtimes::tokio::future_into_py; use pyo3_stub_gen::derive::*; use sift_stream::{ AutoRegisterStream, FileBackup, Flow, LiveStreamingOnly, LiveStreamingWithBackups, diff --git a/rust/crates/sift_stream_bindings/src/stream/builder.rs b/rust/crates/sift_stream_bindings/src/stream/builder.rs index 0b84b2f397..c06fe62cb8 100644 --- a/rust/crates/sift_stream_bindings/src/stream/builder.rs +++ b/rust/crates/sift_stream_bindings/src/stream/builder.rs @@ -128,7 +128,7 @@ impl SiftStreamBuilderPy { let mode_builder = config_builder.live_only(); - let awaitable = pyo3_async_runtimes::tokio::future_into_py(py, async move { + let awaitable = crate::runtime::future_into_py(py, async move { match mode_builder.build().await { Ok(stream) => Ok(SiftStreamPy::from(stream)), Err(e) => Err(crate::error::SiftErrorWrapper(e).into()), @@ -321,7 +321,7 @@ impl LiveOnlyBuilderPy { let ingestion_cap = self.ingestion_data_channel_capacity; let control_cap = self.control_channel_capacity; - let awaitable = pyo3_async_runtimes::tokio::future_into_py(py, async move { + let awaitable = crate::runtime::future_into_py(py, async move { let config_builder = make_stream_config_builder(base)?; let mode_builder = config_builder @@ -408,7 +408,7 @@ impl LiveWithBackupsBuilderPy { let backup_cap = self.backup_data_channel_capacity; let control_cap = self.control_channel_capacity; - let awaitable = pyo3_async_runtimes::tokio::future_into_py(py, async move { + let awaitable = crate::runtime::future_into_py(py, async move { let config_builder = make_stream_config_builder(base)?; let mode_builder = config_builder @@ -481,7 +481,7 @@ impl FileBackupBuilderPy { .metrics_streaming_interval .map(std::time::Duration::from); - let awaitable = pyo3_async_runtimes::tokio::future_into_py(py, async move { + let awaitable = crate::runtime::future_into_py(py, async move { let config_builder = make_stream_config_builder(base)?; let mode_builder = config_builder diff --git a/rust/crates/sift_stream_bindings/src/stream/mod.rs b/rust/crates/sift_stream_bindings/src/stream/mod.rs index 8a5cfc96d9..28717288c3 100644 --- a/rust/crates/sift_stream_bindings/src/stream/mod.rs +++ b/rust/crates/sift_stream_bindings/src/stream/mod.rs @@ -8,12 +8,12 @@ pub mod time; use crate::error::SiftErrorWrapper; use crate::metrics::SiftStreamMetricsSnapshotPy; +use crate::runtime::future_into_py; use crate::stream::channel::ChannelValuePy; use crate::stream::config::{FlowConfigPy, FlowDescriptorPy, RunSelectorPy}; use crate::stream::request::IngestWithConfigDataStreamRequestWrapperPy; use crate::stream::time::TimeValuePy; use pyo3::{prelude::*, types::PyIterator}; -use pyo3_async_runtimes::tokio::future_into_py; use pyo3_stub_gen::derive::*; use sift_rs::ingest::v1::IngestWithConfigDataStreamRequest; use sift_stream::{ diff --git a/rust/crates/sift_stream_bindings/tests/test_runtime_shutdown.py b/rust/crates/sift_stream_bindings/tests/test_runtime_shutdown.py new file mode 100644 index 0000000000..16970dfa0f --- /dev/null +++ b/rust/crates/sift_stream_bindings/tests/test_runtime_shutdown.py @@ -0,0 +1,179 @@ +"""Regression tests for the crate-owned tokio runtime and its shutdown. + +Runtime threads that finish work while the interpreter is finalizing used to +re-enter Python and crash the process (SIGSEGV/SIGABRT on Linux). The module +now registers ``shutdown()`` with ``atexit`` so the runtime is stopped and its +threads joined before finalization starts. + +Every scenario runs in a subprocess: ``shutdown()`` is process-wide and cannot +be undone, and the crash under test happens at interpreter exit. +""" + +import os +import socket +import struct +import subprocess +import sys +import textwrap +import threading + +import pytest +from sift_stream_bindings import is_shut_down, shutdown + +CHILD_TIMEOUT_S = 60 + + +class ResettingListener: + """Accepts and holds connections; resets them all when told to over a control socket. + + Each ``build()`` against ``uri`` connects and then waits for a gRPC reply that never + comes. Resetting the connection fails the build a few milliseconds later, which makes a + runtime thread re-enter Python to reject the awaitable. The control socket lets the child + choose the exact moment, including from inside interpreter finalization. + """ + + def __init__(self) -> None: + self._data = socket.create_server(("127.0.0.1", 0), backlog=128) + self._ctrl = socket.create_server(("127.0.0.1", 0), backlog=16) + self.uri = f"http://127.0.0.1:{self._data.getsockname()[1]}" + self.ctrl_port = self._ctrl.getsockname()[1] + self._lock = threading.Lock() + self._conns: list[socket.socket] = [] + threading.Thread(target=self._accept_data, daemon=True).start() + threading.Thread(target=self._accept_ctrl, daemon=True).start() + + def _accept_data(self) -> None: + while True: + conn, _ = self._data.accept() + with self._lock: + self._conns.append(conn) + + def _accept_ctrl(self) -> None: + while True: + conn, _ = self._ctrl.accept() + threading.Thread(target=self._reset_on_signal, args=(conn,), daemon=True).start() + + def _reset_on_signal(self, ctrl: socket.socket) -> None: + try: + ctrl.recv(2) + finally: + with self._lock: + conns, self._conns = self._conns, [] + for conn in conns: + # SO_LINGER 0 turns close() into a TCP RST. + conn.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0)) + conn.close() + + +def run_child(source: str, *args: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, "-c", textwrap.dedent(source), *args], + capture_output=True, + text=True, + timeout=CHILD_TIMEOUT_S, + env={**os.environ, "RUST_BACKTRACE": "0"}, + ) + + +def assert_exited_cleanly(child: subprocess.CompletedProcess) -> None: + assert child.returncode == 0, f"child exited {child.returncode}\n{child.stderr}" + assert "panicked" not in child.stderr, child.stderr + assert "Fatal Python error" not in child.stderr, child.stderr + + +# Starts a few builds that stay pending, then arranges for their connections to be reset +# from a __del__ that runs during module teardown, after CPython has marked itself +# finalizing. Without the atexit shutdown the runtime threads re-enter Python at that +# point: pyo3 panics on every worker, and on Linux the process dies by signal. +PENDING_BUILDS_THEN_EXIT = """ + import asyncio, os, socket, sys, time + from sift_stream_bindings import IngestionConfigFormPy, SiftStreamBuilderPy + + uri, ctrl_port, mode = sys.argv[1], int(sys.argv[2]), sys.argv[3] + ctrl = socket.create_connection(("127.0.0.1", ctrl_port)) + ctrl_fd = os.dup(ctrl.fileno()) # a raw fd survives module teardown; the socket object may not + + async def start_builds(): + for _ in range(5): + builder = SiftStreamBuilderPy(uri=uri, apikey="not-a-real-key") + builder.enable_tls = False + builder.ingestion_config_form = IngestionConfigFormPy( + asset_name="repro", client_key="repro", flows=[] + ) + builder.build() + await asyncio.sleep(0.6) # let every build connect and block on its first request + + asyncio.run(start_builds()) + + class ResetDuringFinalization: + def __init__(self, fd, write, sleep): + self.fd, self.write, self.sleep = fd, write, sleep + + def __del__(self): + self.write(self.fd, b"go") + self.sleep(0.5) # releases the GIL so runtime threads can try to re-enter + + _keep = ResetDuringFinalization(ctrl_fd, os.write, time.sleep) + if mode == "explicit-shutdown": + import sift_stream_bindings + sift_stream_bindings.shutdown() + assert sift_stream_bindings.is_shut_down() +""" + + +@pytest.mark.parametrize("mode", ["atexit", "explicit-shutdown"]) +def test_runtime_threads_do_not_reenter_python_during_finalization(mode: str) -> None: + listener = ResettingListener() + child = run_child(PENDING_BUILDS_THEN_EXIT, listener.uri, str(listener.ctrl_port), mode) + assert_exited_cleanly(child) + + +def test_shutdown_is_idempotent_and_blocks_new_work() -> None: + child = run_child( + """ + import asyncio + import sift_stream_bindings + from sift_stream_bindings import IngestionConfigFormPy, SiftStreamBuilderPy + + assert not sift_stream_bindings.is_shut_down() + sift_stream_bindings.shutdown(timeout=1.0) + sift_stream_bindings.shutdown() # second call is a no-op + assert sift_stream_bindings.is_shut_down() + + async def main(): + builder = SiftStreamBuilderPy(uri="http://127.0.0.1:1", apikey="k") + builder.ingestion_config_form = IngestionConfigFormPy( + asset_name="a", client_key="a", flows=[] + ) + try: + builder.build() + except RuntimeError as e: + assert "shut down" in str(e), e + else: + raise AssertionError("build() should fail after shutdown()") + + asyncio.run(main()) + """ + ) + assert_exited_cleanly(child) + + +def test_shutdown_before_any_use_is_safe() -> None: + child = run_child( + """ + import sift_stream_bindings + sift_stream_bindings.shutdown() + assert sift_stream_bindings.is_shut_down() + """ + ) + assert_exited_cleanly(child) + + +def test_shutdown_rejects_bad_timeout() -> None: + # Runs in-process: an invalid timeout is rejected before the runtime is touched. + assert not is_shut_down() + with pytest.raises(ValueError): + shutdown(timeout=-1.0) + with pytest.raises(ValueError): + shutdown(timeout=float("nan")) + assert not is_shut_down() From 7262b5dcc1674aae5f321ec8523db5be705f626b Mon Sep 17 00:00:00 2001 From: solidiquis Date: Tue, 29 Sep 2026 12:00:42 -0700 Subject: [PATCH 2/4] fix version --- PR_DESC.md | 40 ++++++++++++++++++++++++++++++++++++++++ python/CHANGELOG.md | 4 ---- python/pyproject.toml | 14 +++++++------- 3 files changed, 47 insertions(+), 11 deletions(-) create mode 100644 PR_DESC.md diff --git a/PR_DESC.md b/PR_DESC.md new file mode 100644 index 0000000000..99838b3799 --- /dev/null +++ b/PR_DESC.md @@ -0,0 +1,40 @@ +## rust(fix): stop the sift-stream-bindings tokio runtime before interpreter exit + +### Problem + +A customer on `sift-stream-bindings` 0.3.0 reported Python processes dying with SIGSEGV, and sometimes SIGABRT, after their program had finished. The faulting thread was always a `tokio-rt-worker` inside `libpython`. Their production hit rate was 2 of 34 processes. Every stream was awaited and `finish()`ed before exit, so the crash turned successful runs into non-zero exit codes for their supervisor. + +The current release (0.5.1) has the same bug. Both versions use pyo3-async-runtimes 0.28, which builds its tokio runtime in a process-global `OnceLock` with no shutdown path. Every awaitable the bindings hand to Python is resolved by a runtime thread that re-enters the interpreter to call `set_result` or `set_exception`. CPython runs `atexit`, then marks itself finalizing and tears down its thread-state machinery. A runtime thread that finishes after that point either dereferences freed interpreter state in `PyGILState_Ensure` (SIGSEGV) or gets `pthread_exit` called on it with Rust frames on the stack, which glibc unwinds and Rust aborts (SIGABRT). pyo3's own finalizing guard only applies on Python 3.13 and up and is racy. + +### Fix + +The bindings now own their tokio runtime (`src/runtime.rs`) and plug it into `pyo3_async_runtimes::generic`, so task locals, cancellation, and future resolution are unchanged. All sixteen `future_into_py` call sites route through it. + +- New `shutdown(timeout=5.0)` stops the runtime and joins its threads with the GIL released. The module registers it with `atexit` at import, so a normal interpreter exit always runs it while the interpreter is still intact. It can also be called directly, for example before `os._exit()`. +- New `is_shut_down()` reports whether it has run. Calls that need the runtime afterwards raise `RuntimeError` instead of crashing. +- Runtime threads are named `sift-stream-worker` so future crash logs identify them. + +### Verification + +Linux container (Debian 13, aarch64, CPython 3.11.16, glibc 2.41), running the customer's repro script unchanged: + +| Build | Customer repro, 150 trials | Deterministic repro, 20 trials | +|---|---|---| +| main (0.5.1) | 134 clean, 10 SIGABRT, 6 SIGSEGV | 20 clean, pyo3 panics on every worker | +| this branch | 150 clean | 20 clean, no stderr | + +macOS does not unwind Rust frames on `pthread_exit`, so the customer's repro passes there even on main. A deterministic variant that fires connection resets from a `__del__` during module teardown shows the same re-entry on macOS on Python 3.11 and 3.13, and is clean with the fix. + +- Bindings test suite: 63 passed on Linux 3.11, macOS 3.11, and macOS 3.13. Four new subprocess tests in `tests/test_runtime_shutdown.py` cover the finalization scenario, explicit `shutdown()`, idempotence, shutdown before first use, and bad timeouts. +- `cargo clippy --all-targets -D warnings` clean. Generated `.pyi` stub unchanged. + +### Releases + +- `sift-stream-bindings` 0.5.1 to 0.5.2, changelog entry under Unreleased in `rust/CHANGELOG.md`. +- The Python package is not touched here. Once the 0.5.2 wheel is on PyPI, a separate PR bumps `sift-stack-py`, moves its pins to `sift-stream-bindings==0.5.2`, and regenerates `uv.lock`. + +### Customer follow-up + +Their diagnosis was correct on all three points: a `shutdown()` API now exists, the `atexit` hook lives inside the extension, and runtime ownership did not change between 0.3.0 and 0.5.1. They need the 0.5.2 release, not an upgrade to any existing version. + +🤖 Generated with [Claude Code](https://claude.com/claude-code) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index 7b544ba3da..1e016f978d 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -3,10 +3,6 @@ All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/). -## [v0.22.1] - September 29, 2026 - -- Fix streaming ingestion processes crashing with SIGSEGV or SIGABRT at interpreter exit. The `sift-stream-bindings` tokio runtime was never shut down, so a runtime thread that finished work while Python was finalizing re-entered the interpreter and killed the process after the program had completed. The bindings now stop the runtime from an `atexit` hook, and expose `shutdown(timeout=5.0)` for callers that exit with `os._exit()`. Requires `sift-stream-bindings` 0.5.2. - ## [v0.22.0] - September 25, 2026 - Added `client.user_defined_functions` for reusable expressions. diff --git a/python/pyproject.toml b/python/pyproject.toml index 4bd414520e..6a2fdfb484 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "sift_stack_py" -version = "0.22.1" +version = "0.22.0" description = "Python client library for the Sift API" requires-python = ">=3.8" readme = { file = "README.md", content-type = "text/markdown" } @@ -71,7 +71,7 @@ all = [ 'pyarrow>=17.0.0', 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", - 'sift-stream-bindings==0.5.2', + 'sift-stream-bindings==0.5.1', 'types-pyOpenSSL<24.0.0', ] build = [ @@ -119,7 +119,7 @@ dev-all = [ 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", 'ruff~=0.12.10', - 'sift-stream-bindings==0.5.2', + 'sift-stream-bindings==0.5.1', 'tomlkit~=0.13.3', 'types-pyOpenSSL<24.0.0', ] @@ -179,7 +179,7 @@ docs-build = [ 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", 'ruff~=0.12.10', - 'sift-stream-bindings==0.5.2', + 'sift-stream-bindings==0.5.1', 'tomlkit~=0.13.3', 'types-pyOpenSSL<24.0.0', ] @@ -209,10 +209,10 @@ rosbags = [ "rosbags~=0.0 ; python_full_version >= '3.8.2'", ] sift-stream = [ - 'sift-stream-bindings==0.5.2', + 'sift-stream-bindings==0.5.1', ] sift-stream-bindings = [ - 'sift-stream-bindings==0.5.2', + 'sift-stream-bindings==0.5.1', ] tdms = [ 'npTDMS~=1.9', @@ -260,7 +260,7 @@ docs = [ openssl = ["pyOpenSSL<24.0.0", "types-pyOpenSSL<24.0.0", "cffi~=1.14"] tdms = ["npTDMS~=1.9"] rosbags = ["rosbags~=0.0 ; python_full_version >= '3.8.2'"] -sift-stream = ["sift-stream-bindings==0.5.2"] +sift-stream = ["sift-stream-bindings==0.5.1"] hdf5 = ["h5py~=3.11", "polars~=1.8"] # polars is only used by sift_py; remove once sift_py is fully deprecated ulog = ["pyulog~=1.2.2"] mcap = ["mcap~=1.4", "mcap-ros2-support~=0.5.7"] From 36c984e312eff0dbb49a9a137590835f173af1dc Mon Sep 17 00:00:00 2001 From: solidiquis Date: Tue, 29 Sep 2026 12:34:01 -0700 Subject: [PATCH 3/4] fix --- PR_DESC.md | 40 ---------------------------------------- 1 file changed, 40 deletions(-) delete mode 100644 PR_DESC.md diff --git a/PR_DESC.md b/PR_DESC.md deleted file mode 100644 index 99838b3799..0000000000 --- a/PR_DESC.md +++ /dev/null @@ -1,40 +0,0 @@ -## rust(fix): stop the sift-stream-bindings tokio runtime before interpreter exit - -### Problem - -A customer on `sift-stream-bindings` 0.3.0 reported Python processes dying with SIGSEGV, and sometimes SIGABRT, after their program had finished. The faulting thread was always a `tokio-rt-worker` inside `libpython`. Their production hit rate was 2 of 34 processes. Every stream was awaited and `finish()`ed before exit, so the crash turned successful runs into non-zero exit codes for their supervisor. - -The current release (0.5.1) has the same bug. Both versions use pyo3-async-runtimes 0.28, which builds its tokio runtime in a process-global `OnceLock` with no shutdown path. Every awaitable the bindings hand to Python is resolved by a runtime thread that re-enters the interpreter to call `set_result` or `set_exception`. CPython runs `atexit`, then marks itself finalizing and tears down its thread-state machinery. A runtime thread that finishes after that point either dereferences freed interpreter state in `PyGILState_Ensure` (SIGSEGV) or gets `pthread_exit` called on it with Rust frames on the stack, which glibc unwinds and Rust aborts (SIGABRT). pyo3's own finalizing guard only applies on Python 3.13 and up and is racy. - -### Fix - -The bindings now own their tokio runtime (`src/runtime.rs`) and plug it into `pyo3_async_runtimes::generic`, so task locals, cancellation, and future resolution are unchanged. All sixteen `future_into_py` call sites route through it. - -- New `shutdown(timeout=5.0)` stops the runtime and joins its threads with the GIL released. The module registers it with `atexit` at import, so a normal interpreter exit always runs it while the interpreter is still intact. It can also be called directly, for example before `os._exit()`. -- New `is_shut_down()` reports whether it has run. Calls that need the runtime afterwards raise `RuntimeError` instead of crashing. -- Runtime threads are named `sift-stream-worker` so future crash logs identify them. - -### Verification - -Linux container (Debian 13, aarch64, CPython 3.11.16, glibc 2.41), running the customer's repro script unchanged: - -| Build | Customer repro, 150 trials | Deterministic repro, 20 trials | -|---|---|---| -| main (0.5.1) | 134 clean, 10 SIGABRT, 6 SIGSEGV | 20 clean, pyo3 panics on every worker | -| this branch | 150 clean | 20 clean, no stderr | - -macOS does not unwind Rust frames on `pthread_exit`, so the customer's repro passes there even on main. A deterministic variant that fires connection resets from a `__del__` during module teardown shows the same re-entry on macOS on Python 3.11 and 3.13, and is clean with the fix. - -- Bindings test suite: 63 passed on Linux 3.11, macOS 3.11, and macOS 3.13. Four new subprocess tests in `tests/test_runtime_shutdown.py` cover the finalization scenario, explicit `shutdown()`, idempotence, shutdown before first use, and bad timeouts. -- `cargo clippy --all-targets -D warnings` clean. Generated `.pyi` stub unchanged. - -### Releases - -- `sift-stream-bindings` 0.5.1 to 0.5.2, changelog entry under Unreleased in `rust/CHANGELOG.md`. -- The Python package is not touched here. Once the 0.5.2 wheel is on PyPI, a separate PR bumps `sift-stack-py`, moves its pins to `sift-stream-bindings==0.5.2`, and regenerates `uv.lock`. - -### Customer follow-up - -Their diagnosis was correct on all three points: a `shutdown()` API now exists, the `atexit` hook lives inside the extension, and runtime ownership did not change between 0.3.0 and 0.5.1. They need the 0.5.2 release, not an upgrade to any existing version. - -🤖 Generated with [Claude Code](https://claude.com/claude-code) From 8aea1858c775de3ac3064e236f7d99070d90b4a5 Mon Sep 17 00:00:00 2001 From: solidiquis Date: Tue, 29 Sep 2026 14:09:31 -0700 Subject: [PATCH 4/4] PR feedback --- rust/CHANGELOG.md | 5 +-- rust/crates/sift_stream_bindings/src/lib.rs | 19 +++++++++-- .../sift_stream_bindings/src/runtime.rs | 24 +++++++++++--- .../tests/test_runtime_shutdown.py | 32 +++++++++++++++++++ 4 files changed, 72 insertions(+), 8 deletions(-) diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index f5f4ace557..53ec9e902d 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -26,8 +26,9 @@ process with SIGSEGV or SIGABRT after the program had already completed. The cra runtime and registers a new `shutdown(timeout=5.0)` function with `atexit`, so the runtime is stopped and its threads joined while the interpreter is still intact. `shutdown()` can also be called directly, for example before `os._exit()`, and `is_shut_down()` reports whether it has -run. Calls that need the runtime raise `RuntimeError` afterwards. Runtime threads are now named -`sift-stream-worker`. +run. Calls that need the runtime raise `RuntimeError` afterwards. If threads are still alive when +the timeout expires, `shutdown()` logs a warning and issues a `RuntimeWarning`, since those threads +can still crash the process. Runtime threads are now named `sift-stream-worker`. #### Asset-name check is now case-insensitive diff --git a/rust/crates/sift_stream_bindings/src/lib.rs b/rust/crates/sift_stream_bindings/src/lib.rs index 4b0149f137..8d94d6f900 100644 --- a/rust/crates/sift_stream_bindings/src/lib.rs +++ b/rust/crates/sift_stream_bindings/src/lib.rs @@ -1,6 +1,7 @@ -use pyo3::exceptions::PyValueError; +use pyo3::exceptions::{PyRuntimeWarning, PyValueError}; use pyo3::prelude::*; use pyo3_stub_gen::define_stub_info_gatherer; +use std::ffi::CString; use std::sync::{Mutex, Once}; use std::time::Duration; use tracing::Level; @@ -152,6 +153,10 @@ fn is_tracing_initialized() -> bool { /// completed never resolve, and any later call that needs the runtime raises `RuntimeError`. /// Calling this more than once is harmless. /// +/// If runtime threads are still alive when the timeout expires, this logs a warning through +/// `tracing` and issues a Python `RuntimeWarning`, because those threads can still re-enter +/// Python during finalization. Raise the timeout or finish streams earlier if you see it. +/// /// Args: /// timeout: Seconds to wait for in-flight work before giving up (default: 5.0) #[pyfunction] @@ -162,7 +167,17 @@ fn shutdown(py: Python<'_>, timeout: f64) -> PyResult<()> { "timeout must be a non-negative number of seconds", )); } - runtime::shutdown(py, Duration::from_secs_f64(timeout)); + let leftover = runtime::shutdown(py, Duration::from_secs_f64(timeout)); + if leftover > 0 { + let message = format!( + "sift_stream_bindings.shutdown() timed out after {timeout}s with {leftover} runtime \ + thread(s) still running; they may re-enter Python during interpreter finalization \ + and crash the process. Finish streams before exit or raise the timeout." + ); + tracing::warn!("{message}"); + let message = CString::new(message)?; + PyErr::warn(py, py.get_type::().as_any(), &message, 1)?; + } Ok(()) } diff --git a/rust/crates/sift_stream_bindings/src/runtime.rs b/rust/crates/sift_stream_bindings/src/runtime.rs index 8f36412dd3..8b1dca5f0c 100644 --- a/rust/crates/sift_stream_bindings/src/runtime.rs +++ b/rust/crates/sift_stream_bindings/src/runtime.rs @@ -21,7 +21,7 @@ use pyo3_async_runtimes::generic::{self, ContextExt, Runtime as GenericRuntime}; use std::cell::OnceCell; use std::future::Future; use std::pin::Pin; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Mutex, OnceLock, PoisonError}; use std::time::Duration; use tokio::runtime::{Builder, Handle, Runtime}; @@ -35,6 +35,8 @@ static RUNTIME: Mutex> = Mutex::new(None); static HANDLE: OnceLock = OnceLock::new(); /// Set by [`shutdown`] before it drains `RUNTIME`; never cleared. static SHUT_DOWN: AtomicBool = AtomicBool::new(false); +/// Runtime threads (workers and blocking pool) currently alive, kept by tokio's thread hooks. +static LIVE_THREADS: AtomicUsize = AtomicUsize::new(0); tokio::task_local! { static TASK_LOCALS: OnceCell; @@ -107,14 +109,22 @@ pub(crate) fn is_shut_down() -> bool { /// runtime raises `RuntimeError`. The GIL is released while waiting so that a /// task which is already resolving a Python future can finish. Calling this more /// than once is harmless. -pub(crate) fn shutdown(py: Python<'_>, timeout: Duration) { +/// +/// Returns the number of runtime threads still alive when the wait ended. Tokio +/// gives no signal when `shutdown_timeout` gives up, so this is how a caller +/// learns that threads outlived the timeout and could still re-enter Python. +pub(crate) fn shutdown(py: Python<'_>, timeout: Duration) -> usize { SHUT_DOWN.store(true, Ordering::SeqCst); let runtime = RUNTIME .lock() .unwrap_or_else(PoisonError::into_inner) .take(); - if let Some(runtime) = runtime { - py.detach(|| runtime.shutdown_timeout(timeout)); + match runtime { + Some(runtime) => py.detach(|| { + runtime.shutdown_timeout(timeout); + LIVE_THREADS.load(Ordering::SeqCst) + }), + None => 0, } } @@ -142,6 +152,12 @@ fn ensure_started() -> PyResult<()> { let runtime = Builder::new_multi_thread() .enable_all() .thread_name("sift-stream-worker") + .on_thread_start(|| { + LIVE_THREADS.fetch_add(1, Ordering::SeqCst); + }) + .on_thread_stop(|| { + LIVE_THREADS.fetch_sub(1, Ordering::SeqCst); + }) .build() .map_err(|e| { PyRuntimeError::new_err(format!( diff --git a/rust/crates/sift_stream_bindings/tests/test_runtime_shutdown.py b/rust/crates/sift_stream_bindings/tests/test_runtime_shutdown.py index 16970dfa0f..82ef5109e3 100644 --- a/rust/crates/sift_stream_bindings/tests/test_runtime_shutdown.py +++ b/rust/crates/sift_stream_bindings/tests/test_runtime_shutdown.py @@ -169,6 +169,38 @@ def test_shutdown_before_any_use_is_safe() -> None: assert_exited_cleanly(child) +def test_shutdown_warns_when_threads_outlive_timeout() -> None: + # Builds that are blocked on a server that never replies keep runtime threads busy. A zero + # timeout cannot wait for them, so shutdown() must say that threads are still alive. + listener = ResettingListener() + child = run_child( + """ + import asyncio, sys, warnings + import sift_stream_bindings + from sift_stream_bindings import IngestionConfigFormPy, SiftStreamBuilderPy + + async def main(): + for _ in range(3): + builder = SiftStreamBuilderPy(uri=sys.argv[1], apikey="not-a-real-key") + builder.enable_tls = False + builder.ingestion_config_form = IngestionConfigFormPy( + asset_name="repro", client_key="repro", flows=[] + ) + builder.build() + await asyncio.sleep(0.3) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + sift_stream_bindings.shutdown(timeout=0.0) + messages = [str(w.message) for w in caught if w.category is RuntimeWarning] + assert any("timed out" in m and "still running" in m for m in messages), caught + + asyncio.run(main()) + """, + listener.uri, + ) + assert_exited_cleanly(child) + + def test_shutdown_rejects_bad_timeout() -> None: # Runs in-process: an invalid timeout is rejected before the runtime is touched. assert not is_shut_down()