diff --git a/docs/math-validation.md b/docs/math-validation.md new file mode 100644 index 0000000..1cbf06b --- /dev/null +++ b/docs/math-validation.md @@ -0,0 +1,121 @@ +# Time arithmetic and timer invariants + +This review uses master `5f5027ccf44b47389290138102f53d572909ed5c` +(after PR #109) as its baseline. Its scope is duration conversion, deadline +saturation, interval phase arithmetic, timer ordering/cancellation and stream +watermark arithmetic. The changes concern executor shutdown and IOCP rounding; +the additional model tests exercise the existing interval and heap algorithms. + +## Requirements and independent checks + +| Requirement | Oracle / test | Outcome | +| --- | --- | --- | +| A positive finite executor timeout must not panic during conversion | Public Python API at the binary64 predecessor of `2**64`, `2**64` and `sys.float_info.max` | Baseline: one pass, two `RustPanic` failures. Fixed: all pass. | +| An oversized timeout must remain cancellable | Block the shutdown worker with an event, cancel its awaitable, then release and observe worker completion | Pass; one waiting shutdown call, no spurious nonwaiting fallback. | +| Positive finite IOCP waits must not become zero-timeout polls | Integer bounds on every nanosecond from 0 through 2,000,000, plus cap/overflow boundaries | Baseline fails at 1 ns; fixed passes. | +| Interval remainder preserves phase across the full Duration representation | Euclidean bounds and divisibility on 56 boundary pairs; 8,256 whole-period translation cases | Pass, including `Duration::MAX`; zero divisor is excluded by the caller. | +| Indexed heap preserves minimum, tie order and registration identity | Ordered-map oracle on 64 fixed xorshift seeds, 512 operations each | Pass; 32,768 operations with insert, cancellation, stale handles and expiry. | +| Heap/slab indices remain a bijection and parent keys never exceed child keys | Check both directions, size and order after every model operation | Pass. | + +The heap oracle uses independent insertion IDs and a `BTreeMap`, rather than +another implementation of the four-ary heap. Each trace ends by draining every +remaining timer and comparing the exact wake order. Seeds are integers 1–64; +deadlines are offsets 0–31 ns, deliberately producing ties. Clock progress does +not determine expected results. + +## Numerical risk ledger + +**Executor conversion (fixed).** Binary64 can represent finite seconds far +beyond `Duration`'s range. The positive-finite branch now uses fallible +conversion and saturates overflow to `Duration::MAX`. The locked async-io 2 +implementation uses `Instant::checked_add` and a never-firing timer when the +deadline cannot be represented. Worker completion and cancellation still win; +this does not stop a running worker thread. Ordinary rounding and existing +None/infinity/nonpositive/NaN policies are unchanged. + +**IOCP quantization (fixed).** Let `n` be integer nanoseconds, `q = 1,000,000` +and `M = 2**32 - 2`. The encoded finite timeout is +`m = min(ceil(n / q), M)` milliseconds. For `n <= M*q`, +`n <= m*q < n + q`; positive inputs therefore stay positive. Above the cap, +the wait is deliberately shortened to `M` ms and the scheduler must recheck +the actual deadline. `None` alone uses `2**32 - 1` (INFINITE). +`Duration::MAX.as_nanos() < 2**94`, so the calculation fits `u128`. +The conversion remains allocation-free and constant-time. + +[GetQueuedCompletionStatusEx documentation](https://learn.microsoft.com/en-us/windows/win32/api/ioapiset/nf-ioapiset-getqueuedcompletionstatusex) +defines zero as an immediate timeout and INFINITE as an unbounded wait. Rounding +up avoids converting a requested short sleep into polling. The less-than-1-ms +bound describes conversion error, **not** Windows wake latency: OS scheduling +can add delay. No Windows CPU or latency benchmark was run for this review. + +**Interval arithmetic (verified cases, no production change).** A remainder +`r` is uniquely characterized by `0 <= r < p`, `r <= d` and divisibility of +`d-r` by positive `p`. Adding whole periods preserves phase. Tests use exact +integer arithmetic; no floating-point epsilon is needed. + +**Heap identity (bounded testing).** The model exercises slab reuse and stale +cancellations, not all possible operation traces or `u64` generation wraparound. +Passing these checks is not a general proof of concurrency or memory safety. + +## Executed validation + +Linux x86_64, CPython 3.14.7, Rust nightly-2026-09-25: + +- Release extension build, root and runtime-harness Clippy with all targets/all + features and warnings denied, Rust formatting and changed Python lint/format: + passed. +- `python scripts/run_rust_tests.py`: 331 passed, 15 failed. +- `python scripts/run_rust_tests.py --all-features`: 427 passed, 19 failed. + All failures report EPERM from unavailable io_uring/Unix-socket operations. +- Release Python suite with a local Redis server: 477 passed, 2 Unix-socket + EPERM failures, 3 skipped. Redis/hiredis tests ran on asyncio, uvloop and rsloop. +- Tooling suite: 98 passed. Focused Pyright on `python/rsloop`, `tests/typing` + and `tests/test_executor_timeout_bounds.py`: no errors. +- Hotpath audit: 2079 definitions, zero missing hooks; its three tests passed. + +Full Rust/Python suites are not green in this restricted environment. The +Python skips require Winsock, free-threaded Python and a profile extension. +Windows, macOS, free-threaded Python and a working io_uring backend were not +executed locally. The IOCP arithmetic itself is compiled and tested on Linux +through its existing `cfg(any(windows, test))` configuration. + +## CI follow-up: free-threaded stream connection handoff + +The initial PR CI failed on Linux / Python 3.15t in +`test_close_flushes_coalesced_server_writes`. The smol worker assembled the +`open_connection` result by borrowing its protocol while the event loop could +hold a mutable borrow during data/EOF delivery (`PyBorrowError` in +`fast_open_connection_result`). This was reproduced locally on CPython +3.15.0rc3 free-threaded, with `PYTHON_GIL=0`. + +Result assembly now follows a direct Python await on the calling loop. It +cannot run concurrently with that loop's protocol callbacks, and cancellation +propagates directly to `create_connection`. A deterministic event-based test +checks cancellation reaches the inner coroutine; another checks thread, loop +and context identity after completion from a worker thread. A socket stress +test checks immediate peer data/close during the handoff. + +On the fixed free-threaded debug extension, 100 stress iterations (12,800 +connections) passed. The local full suite had 416 passes, 2 Unix-socket EPERM +failures and 68 skips (optional uvloop/Redis dependencies, Winsock and profile +build). These checks supplement the earlier GIL-enabled validation above; +stress success does not constitute a proof for arbitrary interleavings. + + +## CI follow-up: STARTTLS comparison fixture + +Windows CPython 3.11.9 intermittently timed out in the **stdlib Proactor** +parameter of `test_stream_writer_start_tls_round_trip`, after the client +handshake returned while the server handshake remained pending. Native rsloop +parameters passed. Explicit start events and pausing baseline plaintext reads +before READY prevent early ClientHello from being stranded in the plaintext +reader, but did not eliminate the Proactor failure. A diagnostic run with a +longer deadline passed in under 0.19 s; this does not establish that the old +3-second deadline was too short. + +The comparison now explicitly uses `asyncio.SelectorEventLoop` on all platforms, +alongside uvloop where available and rsloop. This keeps the stdlib reference +consistent across OSes; it does not repair or verify Windows Proactor STARTTLS. +The original deadlines are retained. Native rsloop/IOCP STARTTLS remains in the +Windows matrix, including both client-first and server-first upgrade order. +The test runner reports the ten slowest tests to aid future CI diagnosis. diff --git a/python/rsloop/_stream_connection.py b/python/rsloop/_stream_connection.py new file mode 100644 index 0000000..706e293 --- /dev/null +++ b/python/rsloop/_stream_connection.py @@ -0,0 +1,8 @@ +"""Keep native stream result assembly on the caller's event loop thread.""" + + +async def open_connection(awaitable, finish): + # A direct await propagates cancellation into the connection operation. + # There is no suspension between receiving the transport and owning it in + # a writer, nor a worker-thread protocol borrow racing data/EOF callbacks. + return finish(await awaitable) diff --git a/scripts/run_python_tests.py b/scripts/run_python_tests.py index 51a8147..27da69d 100644 --- a/scripts/run_python_tests.py +++ b/scripts/run_python_tests.py @@ -35,7 +35,9 @@ def main() -> int: try: # Own the recurring watchdog: pytest's faulthandler plugin otherwise # cancels it after each test. Forward selectors and other pytest flags. - return int(pytest.main(["-p", "no:faulthandler", *sys.argv[1:]])) + return int( + pytest.main(["-p", "no:faulthandler", "--durations=10", *sys.argv[1:]]) + ) finally: faulthandler.cancel_dump_traceback_later() diff --git a/src/bindings/loop_api/executor.rs b/src/bindings/loop_api/executor.rs index df2ce36..55939d7 100644 --- a/src/bindings/loop_api/executor.rs +++ b/src/bindings/loop_api/executor.rs @@ -278,7 +278,12 @@ async fn shutdown_executor_with_timeout( .map_err(|err| PyRuntimeError::new_err(err.to_string()))?; let timed_out = if timeout.is_finite() && timeout > 0.0 { - match crate::rust_async::timeout(Duration::from_secs_f64(timeout), async move { + // Positive finite inputs can exceed Duration's range. Saturate rather + // than panic after the shutdown worker has already been started. + // async-io treats an unrepresentable Instant deadline as never firing; + // worker completion and cancellation still resolve the outer future. + let duration = Duration::try_from_secs_f64(timeout).unwrap_or(Duration::MAX); + match crate::rust_async::timeout(duration, async move { rx.await .map_err(|_| PyRuntimeError::new_err("default executor shutdown worker dropped"))? }) diff --git a/src/transport/stream/fast.rs b/src/transport/stream/fast.rs index 3d64aeb..767ac88 100644 --- a/src/transport/stream/fast.rs +++ b/src/transport/stream/fast.rs @@ -7,7 +7,6 @@ use pyo3::{ sync::PyOnceLock, types::{PyByteArray, PyBytes, PyDict, PyTuple, PyWeakrefMethods, PyWeakrefReference}, }; -use pyo3_async_runtimes::TaskLocals; use super::{ PyStreamTransport, @@ -2861,8 +2860,7 @@ fn fast_open_connection_awaitable( port_obj: Py, limit: usize, kwargs: Option<&Bound<'_, PyDict>>, -) -> PyResult<(TaskLocals, Py)> { - let locals = task_locals_for_loop(py, loop_obj)?; +) -> PyResult> { let factory = Py::new( py, PyFastProtocolFactory { @@ -2873,11 +2871,11 @@ fn fast_open_connection_awaitable( )?; let kwargs = copy_kwargs(py, kwargs)?; let create_args = PyTuple::new(py, [factory.into_any(), host_obj, port_obj])?; - let awaitable = loop_obj.call_method(py, "create_connection", &create_args, kwargs.as_ref())?; - Ok((locals, awaitable)) + loop_obj.call_method(py, "create_connection", &create_args, kwargs.as_ref()) } #[cfg_attr(feature = "profile", hotpath::measure)] +#[pyfunction] fn fast_open_connection_result(py: Python<'_>, created: Py) -> PyResult> { let result = created.bind(py).cast::()?; let transport = result.get_item(0)?.unbind(); @@ -2915,20 +2913,21 @@ pub fn open_connection( let (host_obj, port_obj) = host_port_objects(py, host, port); let loop_obj = native_stream_loop(py)?; - let (locals, awaitable) = + let awaitable = fast_open_connection_awaitable(py, &loop_obj, host_obj, port_obj, limit, kwargs)?; - Ok( - crate::rust_async::future_into_py_with_locals(py, locals.clone(), async move { - let created = Python::attach(|py| { - pyo3_async_runtimes::into_future_with_locals(&locals, awaitable.bind(py).clone()) - })? - .await?; - - Python::attach(|py| fast_open_connection_result(py, created)) - })? - .unbind(), - ) + // Result assembly borrows the protocol to obtain its reader. Keep it on + // the loop thread with data_received/connection_lost; a Send bridge runs + // on smol workers and races those mutable borrows without the GIL. + // Direct Python await also forwards cancellation to create_connection. + Ok(py + .import("rsloop._stream_connection")? + .getattr("open_connection")? + .call1(( + awaitable, + wrap_pyfunction!(fast_open_connection_result, py)?, + ))? + .unbind()) } /// Returns an awaitable that starts a stream server. diff --git a/src/vibeio/driver/mod.rs b/src/vibeio/driver/mod.rs index 8d92210..f532be4 100644 --- a/src/vibeio/driver/mod.rs +++ b/src/vibeio/driver/mod.rs @@ -238,12 +238,18 @@ mod completion_encoding_tests { } } -/// Reserve Windows INFINITE for an explicitly unbounded wait. +/// Round positive waits up to milliseconds; reserve Windows INFINITE for None. #[cfg_attr(feature = "profile", hotpath::measure)] #[cfg(any(windows, test))] fn iocp_timeout_ms(timeout: Option) -> u32 { match timeout { - Some(timeout) => timeout.as_millis().min((u32::MAX - 1) as u128) as u32, + // Truncation turns sub-millisecond waits into zero-timeout polls and + // busy-spins until the deadline. Round first, then cap below INFINITE. + // Duration::MAX in nanoseconds fits u128, including div_ceil's result. + Some(timeout) => timeout + .as_nanos() + .div_ceil(1_000_000) + .min((u32::MAX - 1) as u128) as u32, None => u32::MAX, } } @@ -252,6 +258,30 @@ fn iocp_timeout_ms(timeout: Option) -> u32 { mod iocp_timeout_tests { use super::*; + #[test] + fn finite_wait_rounding_is_monotone_and_has_less_than_one_ms_error() { + // Integer inequalities are an independent oracle for ceil(ns / 1e6). + // No floating-point tolerance or wall-clock scheduling is involved. + let cap = u128::from(u32::MAX - 1) * 1_000_000; + let mut previous = 0; + for nanos in + (0..2_000_001_u64).chain([cap as u64 - 1, cap as u64, cap as u64 + 1, u64::MAX]) + { + let actual = iocp_timeout_ms(Some(Duration::from_nanos(nanos))); + assert!(actual >= previous); + assert!(actual < u32::MAX); + let represented = u128::from(actual) * 1_000_000; + if u128::from(nanos) <= cap { + assert!(represented >= u128::from(nanos), "rounded down {nanos} ns"); + assert!(represented - u128::from(nanos) < 1_000_000); + } else { + assert_eq!(represented, cap); + } + assert_eq!(actual == 0, nanos == 0); + previous = actual; + } + } + #[test] fn finite_timeouts_never_select_infinite_wait() { assert_eq!(iocp_timeout_ms(None), u32::MAX); diff --git a/src/vibeio/time/interval.rs b/src/vibeio/time/interval.rs index e4bd3ec..c1513d6 100644 --- a/src/vibeio/time/interval.rs +++ b/src/vibeio/time/interval.rs @@ -162,6 +162,44 @@ fn plan_tick( mod tests { use super::*; + #[test] + fn duration_remainders_satisfy_euclidean_invariants_at_numeric_limits() { + let values = [ + Duration::ZERO, + Duration::from_nanos(1), + Duration::new(0, 999_999_999), + Duration::from_secs(1), + Duration::new(1, 1), + Duration::new(u64::MAX / 2, 999_999_999), + Duration::from_secs(u64::MAX), + Duration::MAX, + ]; + for dividend in values { + for divisor in values.into_iter().filter(|value| !value.is_zero()) { + let remainder = duration_remainder(dividend, divisor); + // These two conditions uniquely determine Euclidean remainder; + // check the reconstructed Duration, not a duplicate conversion. + assert!(remainder < divisor); + assert!(remainder <= dividend); + assert_eq!((dividend - remainder).as_nanos() % divisor.as_nanos(), 0); + if dividend < divisor { + assert_eq!(remainder, dividend); + } + } + } + // Translation by any whole number of periods preserves phase. + for period in 1..=64 { + for nanos in 0..=128 { + let divisor = Duration::from_nanos(period); + let dividend = Duration::from_nanos(nanos); + assert_eq!( + duration_remainder(dividend + divisor * 17, divisor), + duration_remainder(dividend, divisor) + ); + } + } + } + #[test] fn tick_plan_preserves_cadence_at_and_between_boundaries() { let base = Instant::now(); diff --git a/src/vibeio/timer.rs b/src/vibeio/timer.rs index 1f549e0..6d956e7 100644 --- a/src/vibeio/timer.rs +++ b/src/vibeio/timer.rs @@ -301,6 +301,99 @@ mod tests { timer } + #[test] + fn indexed_heap_matches_ordered_model_after_cancellation_and_expiry() { + use std::{collections::BTreeMap, sync::Mutex}; + + struct RecordWake(usize, Arc>>); + impl std::task::Wake for RecordWake { + fn wake(self: Arc) { + self.1.lock().unwrap().push(self.0); + } + } + + // Fixed seeds make every mixed-operation trace reproducible. The + // oracle is an ordered map with independent insertion IDs, not a heap. + for seed in 1..=64_u64 { + let mut rng = seed; + let base = Instant::now(); + let mut heap = DeadlineHeap::new(); + let mut model = BTreeMap::new(); + let mut history = Vec::new(); + let observed = Arc::new(Mutex::new(Vec::new())); + for id in 0..512 { + rng ^= rng << 13; + rng ^= rng >> 7; + rng ^= rng << 17; + let deadline = base + Duration::from_nanos((rng >> 8) % 32); + match rng % 4 { + 0 | 1 => { + let handle = heap.insert( + deadline, + Waker::from(Arc::new(RecordWake(id, observed.clone()))), + ); + model.insert((deadline, id), handle); + history.push(((deadline, id), handle)); + } + 2 if !history.is_empty() => { + // Include already cancelled/expired handles after slab + // reuse: they must not remove another registration. + let (key, handle) = history[(rng >> 16) as usize % history.len()]; + assert_eq!(heap.remove(handle).is_some(), model.remove(&key).is_some()); + } + _ => { + let expected: Vec<_> = model + .keys() + .take_while(|(when, _)| *when <= deadline) + .copied() + .collect(); + let mut expired = Vec::new(); + heap.pop_expired(deadline, &mut expired); + for waker in expired { + waker.wake(); + } + let actual = std::mem::take(&mut *observed.lock().unwrap()); + assert_eq!( + actual, + expected.iter().map(|(_, id)| *id).collect::>() + ); + for key in expected { + model.remove(&key); + } + } + } + assert_eq!(heap.entries.len(), model.len(), "seed={seed}, step={id}"); + assert_eq!(heap.heap.len(), model.len()); + assert_eq!( + heap.deadline(), + model.first_key_value().map(|((when, _), _)| *when) + ); + for (position, slab_index) in heap.heap.iter().enumerate() { + let entry = &heap.entries[*slab_index]; + assert_eq!(entry.heap_index, position); + if position > 0 { + let parent = &heap.entries[heap.heap[(position - 1) / 4]]; + assert!( + (parent.deadline, parent.generation) + <= (entry.deadline, entry.generation) + ); + } + } + } + let mut expired = Vec::new(); + heap.pop_expired(base + Duration::from_secs(1), &mut expired); + for waker in expired { + waker.wake(); + } + assert_eq!( + *observed.lock().unwrap(), + model.keys().map(|(_, id)| *id).collect::>() + ); + assert!(heap.heap.is_empty()); + assert!(heap.entries.is_empty()); + } + } + thread_local! { static REENTER: RefCell>> = RefCell::new(None); } diff --git a/tests/test_executor_timeout_bounds.py b/tests/test_executor_timeout_bounds.py new file mode 100644 index 0000000..0a1b0e5 --- /dev/null +++ b/tests/test_executor_timeout_bounds.py @@ -0,0 +1,67 @@ +"""Extreme finite timeouts must not interrupt an otherwise successful shutdown.""" + +import asyncio +import math +import sys +import threading +from typing import Any, cast + +import pytest +import rsloop + + +@pytest.mark.parametrize( + "timeout", + [math.nextafter(float(2**64), 0.0), float(2**64), sys.float_info.max], +) +def test_large_finite_executor_timeout_completes_without_panic(timeout): + calls = [] + + class Executor: + def shutdown(self, wait): + calls.append(wait) + + async def main(): + loop = asyncio.get_running_loop() + loop.set_default_executor(cast(Any, Executor())) + await asyncio.wait_for( + cast(Any, loop.shutdown_default_executor)(timeout=timeout), 2 + ) + with pytest.raises(RuntimeError, match="Executor shutdown has been called"): + await loop.run_in_executor(None, lambda: None) + + rsloop.run(main()) + assert calls == [True] + + +def test_oversized_executor_timeout_remains_cancellable(): + release = threading.Event() + calls = [] + + async def main(): + loop = asyncio.get_running_loop() + started = asyncio.Event() + finished = asyncio.Event() + + class Executor: + def shutdown(self, wait): + calls.append(wait) + loop.call_soon_threadsafe(started.set) + release.wait(2) + loop.call_soon_threadsafe(finished.set) + + loop.set_default_executor(cast(Any, Executor())) + pending = asyncio.ensure_future( + cast(Any, loop.shutdown_default_executor)(timeout=sys.float_info.max) + ) + try: + await asyncio.wait_for(started.wait(), 2) + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending + finally: + release.set() + await asyncio.wait_for(finished.wait(), 2) + + rsloop.run(main()) + assert calls == [True] diff --git a/tests/test_free_threading.py b/tests/test_free_threading.py index 6f2df35..ae7fcdc 100644 --- a/tests/test_free_threading.py +++ b/tests/test_free_threading.py @@ -129,6 +129,41 @@ def test_extension_does_not_force_the_gil_back_on(self) -> None: def test_parallel_loops_echo_over_fast_streams(self) -> None: self._run_parallel_echo(asyncio.start_server, asyncio.open_connection) + def test_immediate_peer_close_during_native_connection_handoff(self) -> None: + # The peer can deliver data/EOF before open_connection has constructed + # the client writer. Exercise that handoff repeatedly, including with + # no GIL to serialize the bridge worker and protocol callbacks. + payload = b"a" * 128 + b"b" * 128 + + async def exercise() -> None: + accepted = [] + + def connected(_reader, writer): + accepted.append(writer) + writer.write(payload[:128]) + writer.write(payload[128:]) + writer.close() + + server = await asyncio.start_server(connected, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + try: + for _ in range(128): + reader, writer = await asyncio.open_connection("127.0.0.1", port) + try: + assert await reader.read() == payload + assert reader.at_eof() + finally: + writer.close() + await writer.wait_closed() + assert len(accepted) == 128 + finally: + server.close() + await server.wait_closed() + for writer in accepted: + await writer.wait_closed() + + run_in_own_loop(lambda: asyncio.wait_for(exercise(), 30)) + def test_parallel_loops_echo_over_stdlib_stream_reader_protocol(self) -> None: # The generic fast path writes into `StreamReader._buffer` through a raw # pointer; running it on several loops at once is the regression test diff --git a/tests/test_stream_connection.py b/tests/test_stream_connection.py new file mode 100644 index 0000000..bdbbf8b --- /dev/null +++ b/tests/test_stream_connection.py @@ -0,0 +1,39 @@ +"""Result assembly must retain loop affinity after cross-thread completion.""" + +import asyncio +import contextvars +import threading + +from rsloop._stream_connection import open_connection + + +def test_connection_result_assembly_preserves_loop_thread_and_context(): + marker = contextvars.ContextVar("connection_marker") + + async def main(): + loop = asyncio.get_running_loop() + owner = threading.get_ident() + result = object() + connected = loop.create_future() + marker.set(result) + + def finish(created): + assert threading.get_ident() == owner + assert asyncio.get_running_loop() is loop + assert marker.get() is result + assert created is result + return result + + worker = threading.Thread( + target=loop.call_soon_threadsafe, args=(connected.set_result, result) + ) + worker.start() + try: + assert ( + await asyncio.wait_for(open_connection(connected, finish), 5) is result + ) + finally: + worker.join(timeout=5) + assert not worker.is_alive() + + asyncio.run(main()) diff --git a/tests/test_stream_routing.py b/tests/test_stream_routing.py index 1947b3f..838ef84 100644 --- a/tests/test_stream_routing.py +++ b/tests/test_stream_routing.py @@ -38,7 +38,9 @@ async def echo(reader, writer): startup = asyncio.start_server(echo, "127.0.0.1", 0) try: server = loop.run_until_complete(startup) - pending = asyncio.open_connection("127.0.0.1", server.sockets[0].getsockname()[1]) + pending = asyncio.open_connection( + "127.0.0.1", server.sockets[0].getsockname()[1] + ) async def exchange(): reader, writer = await asyncio.create_task(pending) @@ -100,6 +102,40 @@ async def main(): asyncio.run(main()) +@pytest.mark.parametrize("cancel", [False, True]) +def test_native_connection_handoff_preserves_error_and_cancellation(cancel): + started = asyncio.Event() + finished = asyncio.Event() + + class Loop(rsloop.Loop): + async def create_connection(self, *args, **kwargs): + started.set() + try: + if cancel: + await asyncio.get_running_loop().create_future() + raise ConnectionRefusedError("connection probe") + finally: + finished.set() + + async def main(): + pending = asyncio.create_task(asyncio.open_connection("127.0.0.1", 1)) + await asyncio.wait_for(started.wait(), 5) + if cancel: + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending + else: + with pytest.raises(ConnectionRefusedError, match="connection probe"): + await pending + await asyncio.wait_for(finished.wait(), 5) + + loop = Loop() + try: + loop.run_until_complete(asyncio.wait_for(main(), 10)) + finally: + loop.close() + + @pytest.mark.parametrize("loop_name", ["asyncio", "uvloop"]) def test_other_loops_keep_their_stdlib_streams(loop_name): if loop_name == "uvloop": diff --git a/tests/test_tls.py b/tests/test_tls.py index 6089a98..76f125b 100644 --- a/tests/test_tls.py +++ b/tests/test_tls.py @@ -582,16 +582,24 @@ def test_stream_writer_start_tls_round_trip(tmp_path, backend, client_first): in inspect.signature(asyncio.StreamWriter.start_tls).parameters ): timeouts["ssl_shutdown_timeout"] = 1 + # Use the same selector-based stdlib reference on every platform. + # Windows CPython 3.11 ProactorEventLoop intermittently stalls this + # STARTTLS exchange after the client completes its handshake. Native + # rsloop/IOCP remains covered by the rsloop parameter on Windows. factory = ( pytest.importorskip("uvloop").new_event_loop if backend == "uvloop" - else (asyncio if backend == "asyncio" else rsloop).new_event_loop + else ( + asyncio.SelectorEventLoop if backend == "asyncio" else rsloop.new_event_loop + ) ) server_ctx, client_ctx = make_ssl_contexts(str(tmp_path)) async def main(): loop = asyncio.get_running_loop() server_go = asyncio.Event() + server_started = asyncio.Event() + client_started = asyncio.Event() done = loop.create_future() callbacks = 0 @@ -600,10 +608,17 @@ async def serve(reader, writer): callbacks += 1 try: assert await reader.readline() == b"STARTTLS\n" + if backend != "rsloop": + # Baseline loops can lose ClientHello bytes delivered to + # the plaintext reader before TLS takes over (e.g. CPython + # #142352). Pause before advertising readiness. + # Keep rsloop's early-ClientHello path exercised unchanged. + writer.transport.pause_reading() writer.write(b"READY\n") await writer.drain() await server_go.wait() old_transport = writer.transport + server_started.set() assert await writer.start_tls(server_ctx, **timeouts) is None assert writer.transport is not old_transport # Older asyncio versions retain the reader's private transport. @@ -635,16 +650,19 @@ async def serve(reader, writer): old_transport = writer.transport if not client_first: server_go.set() - await asyncio.sleep(0) - upgrade = asyncio.create_task( - writer.start_tls( + await server_started.wait() + + async def upgrade_client(): + client_started.set() + return await writer.start_tls( client_ctx, server_hostname="localhost", **timeouts, ) - ) + + upgrade = asyncio.create_task(upgrade_client()) if client_first: - await asyncio.sleep(0) + await client_started.wait() server_go.set() assert await upgrade is None assert writer.transport is not old_transport diff --git a/tests/tooling/test_test_runner.py b/tests/tooling/test_test_runner.py index 692cb27..f3bae57 100644 --- a/tests/tooling/test_test_runner.py +++ b/tests/tooling/test_test_runner.py @@ -70,6 +70,7 @@ def test_wrapper_forwards_arguments_and_exit_code(wrapper_calls, exit_code, mock [ "-p", "no:faulthandler", + "--durations=10", "tests/test_run.py", "-k", "stats",