diff --git a/docs/development.md b/docs/development.md index ea48280..7c9df01 100644 --- a/docs/development.md +++ b/docs/development.md @@ -251,6 +251,29 @@ pytest-mock's `mocker` fixture for mocks, spies, and call assertions. Use `monkeypatch.context()` when a shared function must be restored before the test finishes (for example, import machinery or warning handling). +### Stream application contracts + +Run `uv run python -m pytest tests/test_stream_usecases.py` after rebuilding the +native extension. These loopback TCP scenarios run against asyncio, rsloop and +uvloop when installed. The rsloop cases assert that both endpoints use the +native reader, so a fallback cannot silently bypass the implementation. + +| Scenario | Contract | +| --- | --- | +| JSON Lines, UTF-8 and different write sizes | `async for` yields complete records, including a final line without a newline; repeated EOF stays exhausted. | +| Length-prefixed pipeline | Empty and large bodies retain exact framing; reads larger than the watermark complete; half-close preserves the reply direction. | +| Truncated final body | `IncompleteReadError.partial` and `.expected` describe the missing body accurately. | +| Oversized line followed by another request | `readline` and iteration discard the rejected line; `readuntil` preserves it for explicit recovery. | +| Cancelled iteration | Cancellation reaches the pending read; a later iteration receives the next record on the same connection. | +| Invalid or large declared size | Negative sizes fail when awaited without consuming data; EOF preserves the full expected unsigned size. | + +Keep peer failures observable: await background tasks, close both endpoints and +the server, and check for leaked tasks and event-loop exceptions. Use events or +protocol acknowledgements to coordinate peers. Timeouts bound deadlocks; they +do not establish ordering. A TCP write boundary is not a packet or read boundary. +For a regression, first run the new test against the unchanged extension and +record the failure, then rebuild and run the affected suite after the fix. + ## Build the docs With MkDocs installed: diff --git a/python/rsloop/_stream_reader.py b/python/rsloop/_stream_reader.py new file mode 100644 index 0000000..e5d732d --- /dev/null +++ b/python/rsloop/_stream_reader.py @@ -0,0 +1,10 @@ +"""Async iteration for the native reader, using its existing line semantics.""" + + +async def next_line(reader): + # Only clean EOF ends iteration. Read errors and cancellation propagate, + # and an unterminated final line is delivered before StopAsyncIteration. + line = await reader.readline() + if not line: + raise StopAsyncIteration + return line diff --git a/src/transport/stream/fast.rs b/src/transport/stream/fast.rs index 767ac88..e1aee38 100644 --- a/src/transport/stream/fast.rs +++ b/src/transport/stream/fast.rs @@ -2109,11 +2109,46 @@ impl PyFastStreamReader { feature = "profile", hotpath::measure(impl_type = "PyFastStreamReader") )] - fn readexactly(mut slf: PyRefMut<'_, Self>, py: Python<'_>, n: usize) -> PyResult> { + fn readexactly( + mut slf: PyRefMut<'_, Self>, + py: Python<'_>, + n: &Bound<'_, PyAny>, + ) -> PyResult> { + // Keep the full unsigned size range, including expected sizes reported + // at EOF. A signed binding would reject previously accepted lengths. + let n = match n.extract::() { + Ok(n) => n, + Err(err) => { + if err.is_instance_of::(py) && n.lt(0)? { + let exc = PyValueError::new_err("readexactly size can not be less than zero"); + return slf.ready_exception_future(py, exc.into_value(py).into_any()); + } + return Err(err); + } + }; let future = slf.build_readexactly_future(py, n)?; Self::watch_waiter(slf, py, future) } + #[cfg_attr( + feature = "profile", + hotpath::measure(impl_type = "PyFastStreamReader") + )] + fn __aiter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> { + slf + } + + #[cfg_attr( + feature = "profile", + hotpath::measure(impl_type = "PyFastStreamReader") + )] + fn __anext__(slf: Py, py: Python<'_>) -> PyResult> { + Ok(py + .import("rsloop._stream_reader")? + .call_method1("next_line", (slf,))? + .unbind()) + } + #[cfg_attr( feature = "profile", hotpath::measure(impl_type = "PyFastStreamReader") diff --git a/tests/test_stream_usecases.py b/tests/test_stream_usecases.py new file mode 100644 index 0000000..31c6419 --- /dev/null +++ b/tests/test_stream_usecases.py @@ -0,0 +1,252 @@ +"""Application-level stream contracts, compared with independent event loops. + +Use loopback TCP and explicit peer acknowledgements. Write boundaries are not +assumed to be TCP packet boundaries; both fragmentation and coalescing are valid. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import struct +import sys + +import pytest +import rsloop + + +@pytest.fixture(params=["asyncio", "rsloop", "uvloop"]) +def event_loop(request): + factory = ( + pytest.importorskip("uvloop").new_event_loop + if request.param == "uvloop" + else (asyncio if request.param == "asyncio" else rsloop).new_event_loop + ) + loop = factory() + errors = [] + loop.set_exception_handler(lambda _loop, context: errors.append(context)) + try: + yield loop + finally: + pending = asyncio.all_tasks(loop) + for task in pending: + task.cancel() + if pending: + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.close() + assert not pending, "test leaked background tasks" + assert not errors, errors + + +@contextlib.asynccontextmanager +async def tcp_pair(*, limit=64): + loop = asyncio.get_running_loop() + accepted = loop.create_future() + + def accept(reader, writer): + accepted.set_result((reader, writer)) + + server = await asyncio.start_server(accept, "127.0.0.1", 0, limit=limit) + writers = [] + try: + client = await asyncio.open_connection( + "127.0.0.1", server.sockets[0].getsockname()[1], limit=limit + ) + writers.append(client[1]) + peer = await accepted + writers.append(peer[1]) + if isinstance(loop, rsloop.Loop): + assert type(client[0]).__name__ == "PyFastStreamReader" + assert type(peer[0]).__name__ == "PyFastStreamReader" + yield client, peer + finally: + for writer in writers: + writer.close() + await asyncio.gather(*(writer.wait_closed() for writer in writers)) + server.close() + await server.wait_closed() + + +@contextlib.asynccontextmanager +async def background(coro): + task = asyncio.create_task(coro) + try: + yield task + if not task.cancelled(): + await task # Peer assertions and exceptions must reach pytest. + finally: + if not task.done(): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + +@pytest.mark.parametrize("chunk_size", [1, 7, 4096]) +@pytest.mark.parametrize("final_newline", [False, True]) +def test_json_lines_async_iteration(event_loop, chunk_size, final_newline): + records = [{"message": "Привіт 🌍"}, {"count": 0}, {"finished": True}] + lines = [json.dumps(record, ensure_ascii=False).encode() for record in records] + + async def main(): + async with tcp_pair() as ((reader, writer), (peer_reader, peer_writer)): + + async def publish(): + for index, line in enumerate(lines): + if index < len(lines) - 1 or final_newline: + line += b"\n" + for offset in range(0, len(line), chunk_size): + peer_writer.write(line[offset : offset + chunk_size]) + await peer_writer.drain() + if index == len(lines) - 1: + peer_writer.write_eof() + assert await peer_reader.readexactly(1) == b"!" + + async with background(publish()): + received = [] + async for line in reader: + received.append(json.loads(line)) + writer.write(b"!") + await writer.drain() + assert received == records + assert reader.at_eof() + # Exhausted iterators must remain exhausted. + with pytest.raises(StopAsyncIteration): + await anext(reader) + + event_loop.run_until_complete(asyncio.wait_for(main(), 10)) + + +@pytest.mark.parametrize("truncated", [False, True]) +def test_pipelined_length_prefixed_messages(event_loop, truncated): + # Larger than the read watermark: a pending body must not deadlock the peer. + payloads = [b"", bytes(range(256)) * 257, b"last message"] + frames = b"".join(struct.pack("!I", len(body)) + body for body in payloads) + missing = 4 if truncated else 0 + + async def main(): + async with tcp_pair() as ((reader, writer), (peer_reader, peer_writer)): + + async def publish(): + data = frames[:-missing] if missing else frames + for offset in range(0, len(data), 1019): + peer_writer.write(data[offset : offset + 1019]) + await peer_writer.drain() + peer_writer.write_eof() + # Half-close must preserve the reverse direction. + assert await peer_reader.readexactly(2) == b"OK" + + async with background(publish()): + for index, body in enumerate(payloads): + (size,) = struct.unpack("!I", await reader.readexactly(4)) + assert size == len(body) + if truncated and index == len(payloads) - 1: + with pytest.raises(asyncio.IncompleteReadError) as caught: + await reader.readexactly(size) + assert caught.value.expected == size + assert caught.value.partial == body[:-missing] + else: + assert await reader.readexactly(size) == body + assert await reader.read() == b"" + writer.write(b"OK") + await writer.drain() + + event_loop.run_until_complete(asyncio.wait_for(main(), 10)) + + +@pytest.mark.parametrize("method", ["readline", "readuntil", "__anext__"]) +@pytest.mark.parametrize("split_before_delimiter", [False, True]) +def test_oversized_line_then_next_request(event_loop, method, split_before_delimiter): + async def main(): + async with tcp_pair(limit=8) as ((reader, _), (_, peer_writer)): + allow_suffix = asyncio.Event() + + async def publish(): + if split_before_delimiter: + peer_writer.write(b"x" * 9) + await peer_writer.drain() + # Force the overrun to occur before the newline arrives. + await allow_suffix.wait() + peer_writer.write(b"\nOK\n") + else: + peer_writer.write(b"x" * 9 + b"\nOK\n") + peer_writer.write_eof() + + async with background(publish()): + if method != "readuntil": + with pytest.raises(ValueError): + await getattr(reader, method)() + else: + with pytest.raises(asyncio.LimitOverrunError) as caught: + await reader.readuntil(b"\n") + assert caught.value.consumed == 9 + allow_suffix.set() + if method == "readuntil": + # readuntil retains the rejected record, unlike readline. + assert await reader.readexactly(10) == b"x" * 9 + b"\n" + following = await reader.readline() + if following == b"\n": + # readline may reject the prefix before seeing its delimiter, + # even when the peer supplied the entire line in one write. + assert method != "readuntil" + following = await reader.readline() + assert following == b"OK\n" + assert await reader.readline() == b"" + + event_loop.run_until_complete(asyncio.wait_for(main(), 10)) + + +@pytest.mark.parametrize("size", [-1, -64, -(2**100)]) +def test_invalid_frame_length_leaves_stream_reusable(event_loop, size): + async def main(): + async with tcp_pair() as ((reader, _), (_, peer_writer)): + peer_writer.write(b"next") + peer_writer.write_eof() + # Validate while creating the awaitable as well as awaiting it: + # asyncio reports ValueError inside the coroutine, not at creation. + pending = reader.readexactly(size) + with pytest.raises(ValueError, match="less than zero"): + await pending + assert await reader.readexactly(4) == b"next" + assert await reader.read() == b"" + + event_loop.run_until_complete(asyncio.wait_for(main(), 10)) + + +def test_cancel_iteration_then_resume_same_connection(event_loop): + async def main(): + async with tcp_pair() as ((reader, _), (_, peer_writer)): + started = asyncio.Event() + + async def next_line(): + started.set() + return await anext(reader) + + async with background(next_line()) as pending: + await started.wait() + assert not pending.done() + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending + peer_writer.write(b"after cancellation\n") + peer_writer.write_eof() + assert await anext(reader) == b"after cancellation\n" + with pytest.raises(StopAsyncIteration): + await anext(reader) + + event_loop.run_until_complete(asyncio.wait_for(main(), 10)) + + +@pytest.mark.parametrize("size", [1, sys.maxsize + 1]) +def test_eof_reports_full_declared_frame_size(event_loop, size): + async def main(): + async with tcp_pair() as ((reader, _), (_, peer_writer)): + peer_writer.write_eof() + assert await reader.read() == b"" + with pytest.raises(asyncio.IncompleteReadError) as caught: + await reader.readexactly(size) + assert caught.value.expected == size + assert caught.value.partial == b"" + + event_loop.run_until_complete(asyncio.wait_for(main(), 10))