From 904477a19b02bc5f16eb732ded5cc1d9f12d044f Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 28 Jul 2026 13:15:38 +0200 Subject: [PATCH 1/2] Extract Rust-pump FD lifecycle behind seams; fault-inject it (#74) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Rust inter-stage pump hands the raw pipe descriptors to native code under several partial-failure paths — FD extraction, reader-transport pause/resume, and blocking-mode switch/restore — that had no isolated seam for fault injection. Introduce two seams in a new cuprum/_pipeline_stream_fds.py module (the FD lifecycle also lifts _pipeline_streams back under the 400-line cap): - _BlockingModeGuard: an FD-state object that switches the descriptor pair to blocking mode, capturing prior state, and restores it — with the partial-failure rollback preserved. - _paused_reader: a context manager wrapping _pause_reader_transport so the resume can't be skipped on any exit (return, exception, cancel). _run_rust_pump is refactored (via _pump_over_raw_fds) to drive these, preserving behaviour; existing backend-selection tests are repointed to the new module and still pass. Add cuprum/unittests/test_pipeline_streams_fd_lifecycle.py with Hypothesis fault-injection covering the four #74 hazards: - leaked blocking state: a round-trip property plus an injected toggle-failure asserting no descriptor is left switched; - missing resume: _paused_reader resumes once on both exits and skips resume when the transport can't pause; - wrong fallback: a blocking-toggle failure returns the Python-fallback signal and still resumes the reader; - swallowed unexpected errors: _surface_unexpected_pipe_failures raises the first non-pipe exception and suppresses broken-pipe/reset. Regenerate the maturin wheel-manifest snapshot for the two new files. Closes #74 Co-Authored-By: Claude Opus 4.8 (1M context) --- cuprum/_pipeline_stream_fds.py | 166 ++++++++++ cuprum/_pipeline_streams.py | 170 +++------- .../__snapshots__/test_maturin_build.ambr | 2 + .../test_pipeline_stream_backend_selection.py | 41 +-- .../test_pipeline_streams_fd_lifecycle.py | 297 ++++++++++++++++++ 5 files changed, 533 insertions(+), 143 deletions(-) create mode 100644 cuprum/_pipeline_stream_fds.py create mode 100644 cuprum/unittests/test_pipeline_streams_fd_lifecycle.py diff --git a/cuprum/_pipeline_stream_fds.py b/cuprum/_pipeline_stream_fds.py new file mode 100644 index 00000000..63d62832 --- /dev/null +++ b/cuprum/_pipeline_stream_fds.py @@ -0,0 +1,166 @@ +"""Raw file-descriptor lifecycle for the Rust inter-stage pump. + +When the Rust backend pumps one pipeline stage's stdout into the next stage's +stdin it takes over the raw pipe descriptors from asyncio for the duration of +the transfer. That hand-off has several partial-failure paths: extracting the +descriptor from an asyncio transport, pausing the reader transport so asyncio +does not race the Rust pump, and switching the descriptors to blocking mode +(then restoring their prior mode). This module isolates that lifecycle behind +two seams — the [`_BlockingModeGuard`][cuprum._pipeline_stream_fds._BlockingModeGuard] +FD-state object and the [`_paused_reader`][cuprum._pipeline_stream_fds._paused_reader] +context manager — so it can be fault-injected without a live pump. +""" + +from __future__ import annotations + +import contextlib +import dataclasses as dc +import os +import typing as typ + +if typ.TYPE_CHECKING: + import asyncio + import collections.abc as cabc + + +def _fd_from_transport(transport: object | None) -> int | None: + """Extract a raw FD via ``transport.get_extra_info('pipe').fileno()``.""" + get_extra = getattr(transport, "get_extra_info", None) + if get_extra is None: + return None + pipe: object | None = get_extra("pipe") + fileno = getattr(pipe, "fileno", None) if pipe is not None else None + if fileno is None: + return None + try: + return int(fileno()) + except (OSError, ValueError, TypeError, AttributeError): + return None + + +def _extract_stream_fd( + stream: asyncio.StreamReader | asyncio.StreamWriter | None, +) -> int | None: + """Extract a raw FD from an asyncio stream via its transport.""" + if stream is None: + return None + transport = getattr(stream, "transport", None) + if transport is None: + transport = getattr(stream, "_transport", None) + return _fd_from_transport(transport) + + +def _pause_reader_transport( + reader: asyncio.StreamReader, +) -> cabc.Callable[[], None] | None: + """Pause reader transport callbacks while Rust pump owns the raw FD.""" + transport = getattr(reader, "transport", None) + if transport is None: + transport = getattr(reader, "_transport", None) + pause_reading = getattr(transport, "pause_reading", None) + resume_reading = getattr(transport, "resume_reading", None) + if not callable(pause_reading) or not callable(resume_reading): + return None + try: + pause_reading() + except (RuntimeError, OSError): + return None + + def _resume() -> None: + """Resume the paused reader transport, ignoring teardown errors.""" + with contextlib.suppress(RuntimeError, OSError): + resume_reading() + + return _resume + + +def _set_stream_fds_blocking(*, reader_fd: int, writer_fd: int) -> tuple[bool, bool]: + """Switch pipe FDs to blocking mode and return their prior state.""" + reader_was_blocking = os.get_blocking(reader_fd) + writer_was_blocking = os.get_blocking(writer_fd) + reader_changed = False + try: + if not reader_was_blocking: + os.set_blocking(reader_fd, True) + reader_changed = True + if not writer_was_blocking: + os.set_blocking(writer_fd, True) + except OSError: + if reader_changed: + with contextlib.suppress(OSError, ValueError): + os.set_blocking(reader_fd, reader_was_blocking) + raise + return reader_was_blocking, writer_was_blocking + + +def _restore_stream_fd_blocking( + *, + reader_fd: int, + writer_fd: int, + reader_was_blocking: bool, + writer_was_blocking: bool, +) -> None: + """Restore pipe FD blocking mode captured before Rust pumping.""" + with contextlib.suppress(OSError, ValueError): + os.set_blocking(reader_fd, reader_was_blocking) + with contextlib.suppress(OSError, ValueError): + os.set_blocking(writer_fd, writer_was_blocking) + + +@dc.dataclass(slots=True) +class _BlockingModeGuard: + """The prior blocking mode of a reader/writer FD pair, restorable on demand. + + This is the FD-state object behind the Rust pump's blocking lifecycle: it + captures each descriptor's blocking mode when the pump takes over the raw + FDs and restores it afterwards, so no descriptor is left in the temporary + blocking mode the Rust pump requires. + """ + + reader_fd: int + writer_fd: int + reader_was_blocking: bool + writer_was_blocking: bool + + @classmethod + def engage(cls, *, reader_fd: int, writer_fd: int) -> _BlockingModeGuard: + """Switch both FDs to blocking mode, capturing their prior state. + + Raises ``OSError`` — after rolling back any partial change so no + descriptor is left switched — when either FD cannot be made blocking. + """ + reader_was_blocking, writer_was_blocking = _set_stream_fds_blocking( + reader_fd=reader_fd, + writer_fd=writer_fd, + ) + return cls( + reader_fd=reader_fd, + writer_fd=writer_fd, + reader_was_blocking=reader_was_blocking, + writer_was_blocking=writer_was_blocking, + ) + + def restore(self) -> None: + """Restore both FDs to the blocking mode captured by ``engage``.""" + _restore_stream_fd_blocking( + reader_fd=self.reader_fd, + writer_fd=self.writer_fd, + reader_was_blocking=self.reader_was_blocking, + writer_was_blocking=self.writer_was_blocking, + ) + + +@contextlib.contextmanager +def _paused_reader(reader: asyncio.StreamReader) -> cabc.Iterator[None]: + """Pause the reader transport for the block, always resuming it on exit. + + Wraps ``_pause_reader_transport`` so the resume call cannot be skipped on + any exit path — normal return, exception, or cancellation. When the + transport cannot be paused the resume is simply never attempted. + """ + resume = _pause_reader_transport(reader) + try: + yield + finally: + if resume is not None: + resume() diff --git a/cuprum/_pipeline_streams.py b/cuprum/_pipeline_streams.py index fb644ea7..a745cd5a 100644 --- a/cuprum/_pipeline_streams.py +++ b/cuprum/_pipeline_streams.py @@ -17,7 +17,6 @@ import asyncio import contextlib import dataclasses as dc -import os import typing as typ from cuprum._backend import StreamBackend, get_stream_backend @@ -30,6 +29,11 @@ from cuprum._pipeline_stage_streams import ( _create_stage_capture_tasks as _create_stage_capture_tasks, ) +from cuprum._pipeline_stream_fds import ( + _BlockingModeGuard, + _extract_stream_fd, + _paused_reader, +) from cuprum._streams import _close_stream_writer, _pump_stream if typ.TYPE_CHECKING: @@ -79,90 +83,6 @@ def reset_pump_stream_dispatch_for_testing() -> None: configure_pump_stream_dispatch_for_testing() -def _fd_from_transport(transport: object | None) -> int | None: - """Extract a raw FD via ``transport.get_extra_info('pipe').fileno()``.""" - get_extra = getattr(transport, "get_extra_info", None) - if get_extra is None: - return None - pipe: object | None = get_extra("pipe") - fileno = getattr(pipe, "fileno", None) if pipe is not None else None - if fileno is None: - return None - try: - return int(fileno()) - except (OSError, ValueError, TypeError, AttributeError): - return None - - -def _extract_stream_fd( - stream: asyncio.StreamReader | asyncio.StreamWriter | None, -) -> int | None: - """Extract a raw FD from an asyncio stream via its transport.""" - if stream is None: - return None - transport = getattr(stream, "transport", None) - if transport is None: - transport = getattr(stream, "_transport", None) - return _fd_from_transport(transport) - - -def _pause_reader_transport( - reader: asyncio.StreamReader, -) -> cabc.Callable[[], None] | None: - """Pause reader transport callbacks while Rust pump owns the raw FD.""" - transport = getattr(reader, "transport", None) - if transport is None: - transport = getattr(reader, "_transport", None) - pause_reading = getattr(transport, "pause_reading", None) - resume_reading = getattr(transport, "resume_reading", None) - if not callable(pause_reading) or not callable(resume_reading): - return None - try: - pause_reading() - except (RuntimeError, OSError): - return None - - def _resume() -> None: - """Resume the paused reader transport, ignoring teardown errors.""" - with contextlib.suppress(RuntimeError, OSError): - resume_reading() - - return _resume - - -def _set_stream_fds_blocking(*, reader_fd: int, writer_fd: int) -> tuple[bool, bool]: - """Switch pipe FDs to blocking mode and return their prior state.""" - reader_was_blocking = os.get_blocking(reader_fd) - writer_was_blocking = os.get_blocking(writer_fd) - reader_changed = False - try: - if not reader_was_blocking: - os.set_blocking(reader_fd, True) - reader_changed = True - if not writer_was_blocking: - os.set_blocking(writer_fd, True) - except OSError: - if reader_changed: - with contextlib.suppress(OSError, ValueError): - os.set_blocking(reader_fd, reader_was_blocking) - raise - return reader_was_blocking, writer_was_blocking - - -def _restore_stream_fd_blocking( - *, - reader_fd: int, - writer_fd: int, - reader_was_blocking: bool, - writer_was_blocking: bool, -) -> None: - """Restore pipe FD blocking mode captured before Rust pumping.""" - with contextlib.suppress(OSError, ValueError): - os.set_blocking(reader_fd, reader_was_blocking) - with contextlib.suppress(OSError, ValueError): - os.set_blocking(writer_fd, writer_was_blocking) - - async def _run_rust_pump( *, reader: asyncio.StreamReader, @@ -171,45 +91,14 @@ async def _run_rust_pump( writer_fd: int, ) -> bool: """Run the Rust pump for one pipe hop; return ``True`` when handled.""" - # Flush any bytes asyncio already buffered in the StreamReader - # before the Rust pump takes over the raw file descriptor. - resume_reader = _pause_reader_transport(reader) - try: - await _drain_reader_buffer(reader, writer) - except Exception: - if resume_reader is not None: - resume_reader() - raise - - try: - reader_was_blocking, writer_was_blocking = _set_stream_fds_blocking( - reader_fd=reader_fd, - writer_fd=writer_fd, - ) - except OSError: - if resume_reader is not None: - resume_reader() + handled = await _pump_over_raw_fds( + reader=reader, + writer=writer, + reader_fd=reader_fd, + writer_fd=writer_fd, + ) + if not handled: return False - - from cuprum._streams_rs import rust_pump_stream - - loop = asyncio.get_running_loop() - try: - await loop.run_in_executor( - None, - rust_pump_stream, - reader_fd, - writer_fd, - ) - finally: - _restore_stream_fd_blocking( - reader_fd=reader_fd, - writer_fd=writer_fd, - reader_was_blocking=reader_was_blocking, - writer_was_blocking=writer_was_blocking, - ) - if resume_reader is not None: - resume_reader() # The Rust pump closes the writer FD on return (drop semantics). # Suppress OSError so the asyncio transport close does not raise # EBADF when the descriptor is already gone. @@ -218,6 +107,41 @@ async def _run_rust_pump( return True +async def _pump_over_raw_fds( + *, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter | None, + reader_fd: int, + writer_fd: int, +) -> bool: + """Drive the Rust pump over the raw FDs, managing the FD lifecycle. + + Pauses the reader transport (always resuming on exit via ``_paused_reader``) + and switches the FDs to blocking mode under a ``_BlockingModeGuard`` that + restores their prior mode. Returns ``False`` — the signal to fall back to + the Python pump — when the FDs cannot be made blocking; the writer close is + left to the caller. + """ + with _paused_reader(reader): + # Flush any bytes asyncio already buffered in the StreamReader before + # the Rust pump takes over the raw file descriptor. + await _drain_reader_buffer(reader, writer) + + try: + guard = _BlockingModeGuard.engage(reader_fd=reader_fd, writer_fd=writer_fd) + except OSError: + return False + + from cuprum._streams_rs import rust_pump_stream + + loop = asyncio.get_running_loop() + try: + await loop.run_in_executor(None, rust_pump_stream, reader_fd, writer_fd) + finally: + guard.restore() + return True + + async def _drain_reader_buffer( reader: asyncio.StreamReader, writer: asyncio.StreamWriter | None, diff --git a/cuprum/unittests/__snapshots__/test_maturin_build.ambr b/cuprum/unittests/__snapshots__/test_maturin_build.ambr index d4bb3772..8587a1fe 100644 --- a/cuprum/unittests/__snapshots__/test_maturin_build.ambr +++ b/cuprum/unittests/__snapshots__/test_maturin_build.ambr @@ -14,6 +14,7 @@ 'cuprum/_pipeline_config.py', 'cuprum/_pipeline_internals.py', 'cuprum/_pipeline_stage_streams.py', + 'cuprum/_pipeline_stream_fds.py', 'cuprum/_pipeline_streams.py', 'cuprum/_pipeline_types.py', 'cuprum/_pipeline_wait.py', @@ -100,6 +101,7 @@ 'cuprum/unittests/test_pipeline.py', 'cuprum/unittests/test_pipeline_output_options.py', 'cuprum/unittests/test_pipeline_stream_backend_selection.py', + 'cuprum/unittests/test_pipeline_streams_fd_lifecycle.py', 'cuprum/unittests/test_prerequisite_docs.py', 'cuprum/unittests/test_profile_driver.py', 'cuprum/unittests/test_public_api.py', diff --git a/cuprum/unittests/test_pipeline_stream_backend_selection.py b/cuprum/unittests/test_pipeline_stream_backend_selection.py index 0f27356d..e1628d16 100644 --- a/cuprum/unittests/test_pipeline_stream_backend_selection.py +++ b/cuprum/unittests/test_pipeline_stream_backend_selection.py @@ -8,12 +8,13 @@ import asyncio import contextlib +import os import typing as typ from unittest import mock import pytest -from cuprum import _pipeline_streams +from cuprum import _pipeline_stream_fds, _pipeline_streams from cuprum._testing import ( configure_pump_stream_dispatch_for_testing, reset_pump_stream_dispatch_for_testing, @@ -30,17 +31,17 @@ @contextlib.contextmanager def _nonblocking_pipe_pair() -> cabc.Iterator[tuple[int, int, int, int]]: """Yield two pipes with active ends configured for non-blocking I/O.""" - read_fd, read_write_fd = _pipeline_streams.os.pipe() - write_read_fd, write_fd = _pipeline_streams.os.pipe() - _pipeline_streams.os.set_blocking(read_fd, False) - _pipeline_streams.os.set_blocking(write_fd, False) + read_fd, read_write_fd = os.pipe() + write_read_fd, write_fd = os.pipe() + os.set_blocking(read_fd, False) + os.set_blocking(write_fd, False) try: yield read_fd, read_write_fd, write_read_fd, write_fd finally: - _pipeline_streams.os.close(read_fd) - _pipeline_streams.os.close(read_write_fd) - _pipeline_streams.os.close(write_read_fd) - _pipeline_streams.os.close(write_fd) + os.close(read_fd) + os.close(read_write_fd) + os.close(write_read_fd) + os.close(write_fd) class _TransportWithoutPause: @@ -184,10 +185,10 @@ def _make_blocking_fd_spy( def _spy(reader_fd: int, writer_fd: int) -> int: """Assert both descriptors are blocking, then record the call.""" - assert _pipeline_streams.os.get_blocking(reader_fd), ( + assert os.get_blocking(reader_fd), ( "expected reader FD to be switched to blocking mode" ) - assert _pipeline_streams.os.get_blocking(writer_fd), ( + assert os.get_blocking(writer_fd), ( "expected writer FD to be switched to blocking mode" ) assert reader_fd == expected_reader_fd, ( @@ -260,8 +261,8 @@ async def fake_python_pump( configure_pump_stream_dispatch_for_testing(python_pump=fake_python_pump) asyncio.run(_pipeline_streams._pump_stream_dispatch(reader, writer)) - original_reader_blocking = _pipeline_streams.os.get_blocking(read_fd) - original_writer_blocking = _pipeline_streams.os.get_blocking(write_fd) + original_reader_blocking = os.get_blocking(read_fd) + original_writer_blocking = os.get_blocking(write_fd) asyncio.run(_pipeline_streams._pump_stream_dispatch(reader, None)) assert calls["rust_pump"] == 1, "expected Rust pump path to execute once" @@ -353,7 +354,7 @@ async def fake_drain_reader_buffer( call_order.append("drain") monkeypatch.setattr( - _pipeline_streams, + _pipeline_stream_fds, "_pause_reader_transport", fake_pause_reader_transport, ) @@ -363,12 +364,12 @@ async def fake_drain_reader_buffer( fake_drain_reader_buffer, ) monkeypatch.setattr( - _pipeline_streams, + _pipeline_stream_fds, "_set_stream_fds_blocking", lambda **_: (True, True), ) monkeypatch.setattr( - _pipeline_streams, + _pipeline_stream_fds, "_restore_stream_fd_blocking", lambda **_: call_order.append("restore"), ) @@ -413,7 +414,7 @@ def test_dispatch_restores_reader_blocking_when_writer_toggle_fails( lambda stream: read_fd if stream is reader else write_fd, ) - original_set_blocking = _pipeline_streams.os.set_blocking + original_set_blocking = os.set_blocking calls = {"python_pump": 0} def fake_set_blocking(fd: int, blocking: object) -> None: @@ -428,7 +429,7 @@ def fake_set_blocking(fd: int, blocking: object) -> None: raise OSError(_WRITER_TOGGLE_FAILURE) original_set_blocking(fd, bool(blocking)) - monkeypatch.setattr(_pipeline_streams.os, "set_blocking", fake_set_blocking) + monkeypatch.setattr(os, "set_blocking", fake_set_blocking) configure_pump_stream_dispatch_for_testing( python_pump=lambda reader, writer: self._fake_python_fallback( reader, @@ -442,10 +443,10 @@ def fake_set_blocking(fd: int, blocking: object) -> None: assert calls["python_pump"] == 1, ( "expected Python fallback when writer blocking toggle fails" ) - assert not _pipeline_streams.os.get_blocking(read_fd), ( + assert not os.get_blocking(read_fd), ( "expected reader FD blocking mode to be restored after fallback" ) - assert not _pipeline_streams.os.get_blocking(write_fd), ( + assert not os.get_blocking(write_fd), ( "expected writer FD to remain in its original non-blocking mode" ) diff --git a/cuprum/unittests/test_pipeline_streams_fd_lifecycle.py b/cuprum/unittests/test_pipeline_streams_fd_lifecycle.py new file mode 100644 index 00000000..e10166ab --- /dev/null +++ b/cuprum/unittests/test_pipeline_streams_fd_lifecycle.py @@ -0,0 +1,297 @@ +"""Fault-injection tests for the Rust pump's FD pause/blocking lifecycle. + +`cuprum._pipeline_streams` hands the raw pipe descriptors to the Rust pump under +two seams: `_BlockingModeGuard`, which switches the descriptors to blocking mode +and restores their prior mode, and `_paused_reader`, a context manager that +pauses the reader transport and always resumes it. These tests inject faults +into both to pin the partial-failure behaviour #74 calls out — no leaked +blocking state, no missing resume, correct fallback, and no swallowed +unexpected pipe error. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import os +import sys +import typing as typ + +import pytest +from hypothesis import given +from hypothesis import strategies as st + +from cuprum import _pipeline_stream_fds, _pipeline_streams +from cuprum._pipeline_stream_fds import _BlockingModeGuard, _paused_reader +from cuprum._pipeline_streams import _surface_unexpected_pipe_failures + +if typ.TYPE_CHECKING: + import collections.abc as cabc + +_unix_only = pytest.mark.skipif( + sys.platform == "win32", + reason="os.set_blocking on pipe FDs is a Unix contract", +) + + +@contextlib.contextmanager +def _pipe_fds() -> cabc.Iterator[tuple[int, int]]: + """Yield a fresh ``(reader_fd, writer_fd)`` pipe pair and close both.""" + reader_fd, writer_fd = os.pipe() + try: + yield reader_fd, writer_fd + finally: + for fd in (reader_fd, writer_fd): + with contextlib.suppress(OSError): + os.close(fd) + + +@_unix_only +@given(reader_blocking=st.booleans(), writer_blocking=st.booleans()) +def test_blocking_guard_round_trips_prior_mode( + *, + reader_blocking: bool, + writer_blocking: bool, +) -> None: + """``engage`` forces blocking mode; ``restore`` returns the prior mode.""" + with _pipe_fds() as (reader_fd, writer_fd): + os.set_blocking(reader_fd, reader_blocking) + os.set_blocking(writer_fd, writer_blocking) + + guard = _BlockingModeGuard.engage(reader_fd=reader_fd, writer_fd=writer_fd) + + # While the pump owns the FDs they must both be blocking, whatever + # their prior mode. + assert os.get_blocking(reader_fd) is True + assert os.get_blocking(writer_fd) is True + assert guard.reader_was_blocking == reader_blocking + assert guard.writer_was_blocking == writer_blocking + + guard.restore() + + assert os.get_blocking(reader_fd) == reader_blocking + assert os.get_blocking(writer_fd) == writer_blocking + + +@_unix_only +@given(other_blocking=st.booleans(), fault_target=st.sampled_from(["reader", "writer"])) +def test_failed_engage_leaks_no_blocking_state( + *, + other_blocking: bool, + fault_target: str, +) -> None: + """A toggle failure rolls back any change, leaking no blocking state.""" + with _pipe_fds() as (reader_fd, writer_fd): + # Force the faulted FD non-blocking so ``engage`` attempts to toggle it + # and the injected fault fires; the other FD's mode is free. + if fault_target == "reader": + target_fd = reader_fd + os.set_blocking(reader_fd, False) + os.set_blocking(writer_fd, other_blocking) + else: + target_fd = writer_fd + os.set_blocking(writer_fd, False) + os.set_blocking(reader_fd, other_blocking) + + reader_initial = os.get_blocking(reader_fd) + writer_initial = os.get_blocking(writer_fd) + + real_set_blocking = os.set_blocking + + def faulting_set_blocking(fd: int, blocking: bool) -> None: # noqa: FBT001 + """Fail when the target FD is toggled to blocking; else delegate.""" + if fd == target_fd and blocking: + msg = "injected toggle failure" + raise OSError(msg) + real_set_blocking(fd, blocking) + + os.set_blocking = faulting_set_blocking # type: ignore[assignment] + try: + with pytest.raises(OSError, match="injected toggle failure"): + _BlockingModeGuard.engage(reader_fd=reader_fd, writer_fd=writer_fd) + finally: + os.set_blocking = real_set_blocking # type: ignore[assignment] + + # No descriptor may be left in the transient blocking mode. + assert os.get_blocking(reader_fd) == reader_initial + assert os.get_blocking(writer_fd) == writer_initial + + +class _FakeTransport: + """Reader transport recording pause/resume calls for the CM tests.""" + + def __init__(self, *, pause_raises: bool = False) -> None: + """Start with no calls recorded.""" + self.pause_calls = 0 + self.resume_calls = 0 + self._pause_raises = pause_raises + + def pause_reading(self) -> None: + """Record a pause, optionally failing as a torn-down transport would.""" + self.pause_calls += 1 + if self._pause_raises: + msg = "cannot pause a closing transport" + raise RuntimeError(msg) + + def resume_reading(self) -> None: + """Record a resume.""" + self.resume_calls += 1 + + +class _FakeReader: + """Minimal stand-in exposing a ``transport`` attribute.""" + + def __init__(self, transport: object) -> None: + """Wrap ``transport`` as the reader's public transport.""" + self.transport = transport + + +def _raise_boom() -> None: + """Raise a deterministic error from inside a paused-reader block.""" + msg = "boom" + raise ValueError(msg) + + +@given(body_raises=st.booleans()) +def test_paused_reader_always_resumes_a_pausable_transport( + *, + body_raises: bool, +) -> None: + """Resume fires exactly once on both the normal and exception exits.""" + transport = _FakeTransport() + reader = typ.cast("asyncio.StreamReader", _FakeReader(transport)) + + if body_raises: + with pytest.raises(ValueError, match="boom"), _paused_reader(reader): + _raise_boom() + else: + with _paused_reader(reader): + pass + + assert transport.pause_calls == 1 + assert transport.resume_calls == 1 + + +def test_paused_reader_skips_resume_when_pause_unavailable() -> None: + """A transport without pause/resume yields without attempting a resume.""" + reader = typ.cast("asyncio.StreamReader", _FakeReader(object())) + with _paused_reader(reader): + pass + # No exception and nothing to assert on the bare object: the point is that + # exiting the context manager does not crash when resume is unavailable. + + +def test_paused_reader_skips_resume_when_pause_fails() -> None: + """When pausing raises, the resume is never attempted.""" + transport = _FakeTransport(pause_raises=True) + reader = typ.cast("asyncio.StreamReader", _FakeReader(transport)) + + with _paused_reader(reader): + pass + + assert transport.pause_calls == 1 + assert transport.resume_calls == 0 + + +def _raise_oserror(**_kwargs: object) -> tuple[bool, bool]: + """Stand in for ``_set_stream_fds_blocking`` failing to toggle a FD.""" + msg = "cannot switch descriptor to blocking mode" + raise OSError(msg) + + +def test_run_rust_pump_falls_back_and_resumes_when_blocking_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A blocking-toggle failure returns the Python-fallback signal and resumes.""" + resume_calls = {"count": 0} + + def fake_pause_reader_transport( + reader: asyncio.StreamReader, + ) -> cabc.Callable[[], None]: + """Return a resume callback that records how often it is invoked.""" + del reader + + def _resume() -> None: + """Record a resume invocation.""" + resume_calls["count"] += 1 + + return _resume + + async def fake_drain_reader_buffer( + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter | None, + ) -> None: + """Skip the real buffer flush.""" + del reader, writer + await asyncio.sleep(0) + + monkeypatch.setattr( + _pipeline_stream_fds, + "_pause_reader_transport", + fake_pause_reader_transport, + ) + monkeypatch.setattr( + _pipeline_streams, + "_drain_reader_buffer", + fake_drain_reader_buffer, + ) + monkeypatch.setattr( + _pipeline_stream_fds, + "_set_stream_fds_blocking", + _raise_oserror, + ) + + reader = typ.cast("asyncio.StreamReader", object()) + handled = asyncio.run( + _pipeline_streams._run_rust_pump( + reader=reader, + writer=None, + reader_fd=1, + writer_fd=2, + ) + ) + + assert handled is False, "a blocking-toggle failure must fall back to Python" + assert resume_calls["count"] == 1, "the reader must be resumed on fallback" + + +_SUPPRESSED_PIPE_ERRORS = (BrokenPipeError, ConnectionResetError) +_RESULT_TAGS = ["ok", "broken_pipe", "conn_reset", "value_error", "runtime_error"] + + +def _make_pipe_result(tag: str) -> object: + """Materialise a pipe-task result for the given tag as a fresh object.""" + if tag == "ok": + return object() + if tag == "broken_pipe": + return BrokenPipeError("downstream closed early") + if tag == "conn_reset": + return ConnectionResetError("peer reset") + if tag == "value_error": + return ValueError("unexpected pipe failure") + return RuntimeError("unexpected pipe failure") + + +@given(tags=st.lists(st.sampled_from(_RESULT_TAGS), max_size=8)) +def test_surface_raises_first_unexpected_and_suppresses_pipe_errors( + *, + tags: list[str], +) -> None: + """The first non-pipe exception surfaces; pipe errors and values do not.""" + results = [_make_pipe_result(tag) for tag in tags] + unexpected = [ + result + for result in results + if isinstance(result, Exception) + and not isinstance(result, _SUPPRESSED_PIPE_ERRORS) + ] + + if unexpected: + with pytest.raises((ValueError, RuntimeError)) as exc_info: + _surface_unexpected_pipe_failures(results) + assert exc_info.value is unexpected[0], ( + "the earliest unexpected exception must be the one raised" + ) + else: + # All results are either plain values or suppressed pipe errors. + _surface_unexpected_pipe_failures(results) From a3f900bb6a296a6ee55554d5f033690a019a4f6f Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 28 Jul 2026 23:07:25 +0200 Subject: [PATCH 2/2] Document the raw descriptor lifecycle seams The abstraction policy in AGENTS.md requires a new abstraction's scope and reuse policy to be recorded in the project documentation, and _pipeline_stream_fds appeared nowhere under docs/. Add a "Raw descriptor lifecycle" subsection to design-doc 13.6, which already covers the asyncio/Rust hand-off this module serves. It states the partial-failure paths the module exists to contain, what _BlockingModeGuard and _paused_reader each guarantee, the single consumer, and the reuse policy: further descriptor-lifecycle concerns for this hand-off belong here, but the seams are not a general-purpose descriptor utility. cuprum-design.md is already indexed by docs/contents.md. Also add the worked usage example AGENTS.md requires to _pump_over_raw_fds, covering both the handled and fall-back outcomes. Co-Authored-By: Claude Opus 4.8 (1M context) --- cuprum/_pipeline_streams.py | 27 +++++++++++++++++++++++++++ docs/cuprum-design.md | 26 ++++++++++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/cuprum/_pipeline_streams.py b/cuprum/_pipeline_streams.py index a745cd5a..7af8f001 100644 --- a/cuprum/_pipeline_streams.py +++ b/cuprum/_pipeline_streams.py @@ -121,6 +121,33 @@ async def _pump_over_raw_fds( restores their prior mode. Returns ``False`` — the signal to fall back to the Python pump — when the FDs cannot be made blocking; the writer close is left to the caller. + + Examples + -------- + A successful hop reports that Rust handled it, leaving the caller to close + the writer:: + + handled = await _pump_over_raw_fds( + reader=reader, + writer=writer, + reader_fd=reader_fd, + writer_fd=writer_fd, + ) + assert handled is True + + When the descriptors cannot be switched to blocking mode the call reports + ``False`` instead of raising, and the reader transport has already been + resumed, so the caller falls back to the Python pump:: + + handled = await _pump_over_raw_fds( + reader=reader, + writer=writer, + reader_fd=reader_fd, + writer_fd=writer_fd, + ) + if not handled: + await _run_python_pump(reader, writer) + """ with _paused_reader(reader): # Flush any bytes asyncio already buffered in the StreamReader before diff --git a/docs/cuprum-design.md b/docs/cuprum-design.md index ec65c4a2..e3927335 100644 --- a/docs/cuprum-design.md +++ b/docs/cuprum-design.md @@ -2089,6 +2089,32 @@ Error propagation from Rust to Python uses standard exception mechanisms. The Rust extension raises `OSError` for I/O failures, matching the behaviour of Python's built-in I/O operations. +#### Raw descriptor lifecycle + +Handing a descriptor to the Rust pump means taking it back from asyncio for the +duration of the transfer, and that hand-off has several partial-failure paths: +the descriptor may not be extractable from the transport, the reader transport +may refuse to pause, and either descriptor may refuse to switch to blocking +mode. `cuprum/_pipeline_stream_fds.py` owns that lifecycle behind two seams so +each path is testable without a live pump: + +- `_BlockingModeGuard` — the FD-state object. `engage()` switches the reader and + writer to blocking mode while capturing their prior modes, rolling back a + partial change if the second switch fails; `restore()` returns both to the + captured modes. This is what stops a descriptor being left blocking after the + transfer. +- `_paused_reader` — a context manager that pauses the reader transport and + always resumes it, on every exit path including exceptions and cancellation, + so a resume can never be skipped. + +The module's scope is deliberately narrow: descriptor extraction plus the +pause and blocking-mode lifecycle for the Rust pump hand-off. It is consumed +only by `cuprum/_pipeline_streams.py`. Its reuse policy is that new descriptor +lifecycle concerns for that hand-off belong here rather than being inlined into +the pump, but the seams are not a general-purpose descriptor utility — anything +serving a different caller should be designed against that caller's real +requirements instead of widening these. + ### 13.7 Linux splice() Optimization On Linux, the Rust extension can use the `splice()` system call for zero-copy