Fix timeout overflow and IOCP rounding; validate timer arithmetic and heap invariants

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.

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)

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()
 

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"))?
         })

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<PyAny>,
     limit: usize,
     kwargs: Option<&Bound<'_, PyDict>>,
-) -> PyResult<(TaskLocals, Py<PyAny>)> {
-    let locals = task_locals_for_loop(py, loop_obj)?;
+) -> PyResult<Py<PyAny>> {
     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<PyAny>) -> PyResult<Py<PyAny>> {
     let result = created.bind(py).cast::<PyTuple>()?;
     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.

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<Duration>) -> 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<Duration>) -> 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);

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();