Test real stream use cases and fix native reader iteration and size validation

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:

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

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<Py<PyAny>> {
+    fn readexactly(
+        mut slf: PyRefMut<'_, Self>,
+        py: Python<'_>,
+        n: &Bound<'_, PyAny>,
+    ) -> PyResult<Py<PyAny>> {
+        // 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::<usize>() {
+            Ok(n) => n,
+            Err(err) => {
+                if err.is_instance_of::<pyo3::exceptions::PyOverflowError>(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<Self>, py: Python<'_>) -> PyResult<Py<PyAny>> {
+        Ok(py
+            .import("rsloop._stream_reader")?
+            .call_method1("next_line", (slf,))?
+            .unbind())
+    }
+
     #[cfg_attr(
         feature = "profile",
         hotpath::measure(impl_type = "PyFastStreamReader")

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