From 3182eb121024ae490fbb3afec67033a80189d7f7 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Sun, 19 Jul 2026 16:02:27 +0000 Subject: [PATCH 1/6] test(streams): harden take() ack tests against event-loop timing The `test_take` / `test_take_wit_timestamp*` tests asserted that an event consumed via `stream.take()` had been acked after exactly two `await asyncio.sleep(0)` yields. `take()` acks consumed events from a background task, and on slower interpreters (notably PyPy) that task hasn't run within two event-loop ticks, so `event.message.acked` was still False and the assertion flaked -- as seen on the PyPy CI leg. Replace the fixed sleeps with `wait_for_stream_ack()`, which polls for the ack (up to a 1s timeout) instead of assuming a fixed number of ticks. The existing assertions are unchanged, so a genuinely missing ack still fails the test with the same signal after the timeout. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- tests/functional/test_streams.py | 39 +++++++++++++++++++------------- 1 file changed, 23 insertions(+), 16 deletions(-) diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py index 851594155..469ec788b 100644 --- a/tests/functional/test_streams.py +++ b/tests/functional/test_streams.py @@ -572,6 +572,25 @@ def mock_event_ack(event, return_value=False): return event +async def wait_for_stream_ack(event, *, timeout=1.0): + """Wait until a taken event has been acked. + + ``stream.take()`` acks the events it consumes from a background task, so + the ack has not necessarily run by the time the ``async for`` body returns. + Polling for the ack with a timeout keeps the assertion robust on slower + interpreters (notably PyPy) where the couple of ``asyncio.sleep(0)`` yields + the tests used to rely on aren't enough for that task to run. If the ack + never lands the loop simply times out and returns, leaving the caller's + assertions to report the actual state. + """ + loop = asyncio.get_event_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + if event.ack.called or event.message.acked: + return + await asyncio.sleep(0) + + async def get_event_from_value(stream, value, key=None): await stream.channel.send(key=key, value=value) async for value in stream: @@ -766,10 +785,7 @@ async def test_take(app): break assert event - # need one sleep on Python 3.6.0-3.6.6 + 3.7.0 - # need two sleeps on Python 3.6.7 + 3.7.1 :-/ - await asyncio.sleep(0) - await asyncio.sleep(0) + await wait_for_stream_ack(event) if not event.ack.called: assert event.message.acked @@ -879,10 +895,7 @@ async def test_take_wit_timestamp(app): break assert event - # need one sleep on Python 3.6.0-3.6.6 + 3.7.0 - # need two sleeps on Python 3.6.7 + 3.7.1 :-/ - await asyncio.sleep(0) - await asyncio.sleep(0) + await wait_for_stream_ack(event) if not event.ack.called: assert event.message.acked @@ -905,10 +918,7 @@ async def test_take_wit_timestamp_wit_simple_value(app): break assert event - # need one sleep on Python 3.6.0-3.6.6 + 3.7.0 - # need two sleeps on Python 3.6.7 + 3.7.1 :-/ - await asyncio.sleep(0) - await asyncio.sleep(0) + await wait_for_stream_ack(event) if not event.ack.called: assert event.message.acked @@ -931,10 +941,7 @@ async def test_take_wit_timestamp_without_timestamp_field(app): break assert event - # need one sleep on Python 3.6.0-3.6.6 + 3.7.0 - # need two sleeps on Python 3.6.7 + 3.7.1 :-/ - await asyncio.sleep(0) - await asyncio.sleep(0) + await wait_for_stream_ack(event) if not event.ack.called: assert event.message.acked From 71a1cb8598de5436f07ff1145391e6957b5983c6 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 12:51:02 +0000 Subject: [PATCH 2/6] fix(streams): close abandoned take() generators deterministically on stop The cleanup for Stream.take() / take_events() / take_with_timestamp() / noack_take() -- acking consumed events, restoring enable_acks, and detaching the buffering processor -- lives in `finally` blocks that only run when the generator is finalized. When a caller abandons the generator (`break` inside `async for`), CPython's reference counting finalizes it immediately, but PyPy defers finalization to a future GC cycle that may never come: acks are withheld indefinitely, the stream is left with acks disabled, and the buffering processor stays attached. (This is the root cause of the take() test failures on the PyPy CI leg, demonstrated on a real PyPy: the generator's finally never runs without an explicit gc.collect().) Track every buffering generator in a WeakSet and close leftovers explicitly in Stream.on_stop(), making cleanup deterministic on any interpreter. The WeakSet keeps CPython's prompt refcount finalization (and therefore existing behavior) fully intact; already-exhausted generators make aclose() a no-op, and a generator still running in another task is skipped (its owner cleans up). The new test holds a strong reference to the generator -- blocking finalization on every interpreter, mimicking PyPy -- and asserts the stream's stop path performs the cleanup; it fails without this fix on CPython as well. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- faust/streams.py | 49 ++++++++++++++++++++++++++++++++ tests/functional/test_streams.py | 35 +++++++++++++++++++++++ 2 files changed, 84 insertions(+) diff --git a/faust/streams.py b/faust/streams.py index 9f4286a3f..58c92046e 100644 --- a/faust/streams.py +++ b/faust/streams.py @@ -7,8 +7,10 @@ import weakref from asyncio import CancelledError from contextvars import ContextVar +from functools import wraps from typing import ( Any, + AsyncGenerator, AsyncIterable, AsyncIterator, Callable, @@ -88,6 +90,30 @@ async def maybe_forward(value: Any, channel: ChannelT) -> Any: return value +def _tracks_buffer_agen( + fun: Callable[..., AsyncIterable], +) -> Callable[..., AsyncIterable]: + """Register buffering generators (``take()`` and friends) on the stream. + + The cleanup for these generators -- acking consumed events, restoring + ``enable_acks`` and detaching the buffering processor -- lives in + ``finally`` blocks that only run once the generator is finalized. When a + caller abandons the generator (``break`` inside ``async for``), CPython's + reference counting finalizes it right away, but PyPy defers finalization + to the next major GC cycle, which may never come. Tracking every + generator in a WeakSet lets ``Stream.on_stop`` close leftovers + explicitly, making cleanup deterministic on any interpreter. + """ + + @wraps(fun) + def _create_agen(self: StreamT, *args: Any, **kwargs: Any) -> AsyncIterable: + agen = fun(self, *args, **kwargs) + cast("Stream", self)._active_buffer_agens.add(agen) + return agen + + return _create_agen + + class _LinkedListDirection(NamedTuple): attr: str getter: Callable[[StreamT], Optional[StreamT]] @@ -149,6 +175,12 @@ def __init__( self._processors = list(processors) if processors else [] self._on_start = on_start + # Live buffering generators created by take() and friends, + # closed explicitly in on_stop(). A WeakSet so that on CPython an + # abandoned generator is still finalized (and thus cleaned up) + # promptly by reference counting. + self._active_buffer_agens: "weakref.WeakSet[AsyncGenerator]" = weakref.WeakSet() + # attach beacon to channel, or if iterable attach to current task. task = current_task(loop=self.loop) if task is not None: @@ -300,6 +332,7 @@ async def events(self) -> AsyncIterable[EventT]: if self.current_event is not None: yield self.current_event + @_tracks_buffer_agen async def take(self, max_: int, within: Seconds) -> AsyncIterable[Sequence[T_co]]: """Buffer n values at a time and yield a list of buffered values. @@ -392,6 +425,7 @@ async def add_to_buffer(value: T) -> T: self.enable_acks = stream_enable_acks self._processors.remove(add_to_buffer) + @_tracks_buffer_agen async def take_events( self, max_: int, within: Seconds ) -> AsyncIterable[Sequence[EventT]]: @@ -485,6 +519,7 @@ async def add_to_buffer(value: T) -> T: self.enable_acks = stream_enable_acks self._processors.remove(add_to_buffer) + @_tracks_buffer_agen async def take_with_timestamp( self, max_: int, within: Seconds, timestamp_field_name: str ) -> AsyncIterable[Sequence[T_co]]: @@ -592,6 +627,7 @@ def enumerate(self, start: int = 0) -> AsyncIterable[Tuple[int, T_co]]: """ return aenumerate(self, start) + @_tracks_buffer_agen async def noack_take( self, max_: int, within: Seconds ) -> AsyncIterable[Sequence[T_co]]: @@ -1033,6 +1069,19 @@ async def stop(self) -> None: async def on_stop(self) -> None: """Signal that the stream is stopping.""" + # Close any buffering generator (take() and friends) the caller + # abandoned, so their ``finally`` blocks -- acking consumed events, + # restoring ``enable_acks`` and detaching the buffering processor -- + # run deterministically now instead of whenever the interpreter + # finalizes the generator (which on PyPy waits for a GC cycle that + # may never come). ``aclose()`` on an exhausted generator is a + # harmless no-op. + for agen in list(self._active_buffer_agens): + try: + await agen.aclose() + except RuntimeError: + # Still running in another task; its owner will clean up. + pass self._passive = False self._passive_started.clear() for table_or_stream in self.combined: diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py index 851594155..871de8e74 100644 --- a/tests/functional/test_streams.py +++ b/tests/functional/test_streams.py @@ -777,6 +777,41 @@ async def test_take(app): assert s.enable_acks is True +@pytest.mark.asyncio +async def test_take_cleanup_on_stream_stop(app): + """An abandoned take() generator is closed when the stream stops. + + Breaking out of ``async for`` leaves the generator suspended; its cleanup + (acking consumed events, restoring ``enable_acks``, detaching the + buffering processor) normally runs only when the interpreter finalizes + the generator. Holding a strong reference here blocks finalization on + every interpreter -- mimicking PyPy, where finalization waits for a GC + cycle that may never come -- so this asserts the stream itself closes + leftover generators deterministically on stop. + """ + async with new_stream(app) as s: + assert s.enable_acks is True + n_processors = len(s._processors) + await s.channel.send(value=1) + agen = s.take(1, within=1) # hold a strong reference + event = None + async for value in agen: + assert value == [1] + assert s.enable_acks is False + event = mock_stream_event_ack(s) + break + + assert event + # The reference above keeps the generator alive, so no interpreter + # has finalized it yet: cleanup must come from stopping the stream. + await s.stop() + + if not event.ack.called: + assert event.message.acked + assert s.enable_acks is True + assert len(s._processors) == n_processors + + @pytest.mark.asyncio async def test_take__10(app, loop): s = new_stream(app) From 558301850067226601d0c44d33093d435829a34c Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 13:40:26 +0000 Subject: [PATCH 3/6] fix(streams): track __aiter__ generators too; unskip all PyPy tests Extend the on_stop generator-cleanup mechanism to the main stream iterator (__aiter__ -> _py_aiter/_c_aiter), whose inner finally acks the last-yielded event and whose outer finally detaches from the channel -- previously only the take() family was tracked, so a plain `async for value in stream: break` still leaked its cleanup to GC timing on PyPy. Service.stop() sets _stopped before awaiting on_stop(), so the outer finally's re-entrant self.stop() is a guarded no-op. Make the test suite PyPy-clean and remove every skipif(PyPy) marker (21 decorators covering 30 tests, added wholesale against PyPy 3.9 in #530/#621 and never re-validated): * tests/conftest.py: override the event_loop fixture to run gc.collect() + loop.shutdown_asyncgens() before closing the loop on PyPy, so deferred async-generator finalizers run inside the owning test instead of landing on a dead loop and leaking tasks/warnings into later tests. * tests/meticulous: disable the 4s hypothesis deadline on PyPy -- JIT warm-up makes the first examples orders of magnitude slower than steady state, tripping DeadlineExceeded on correct code. * New regression test test_aiter_cleanup_on_stream_stop mirrors the take() one: holds a strong reference (blocking finalization on every interpreter, mimicking PyPy) and asserts stop-time cleanup acks the event. Verified to fail without the __aiter__ tracking. Verified on CPython: functional+unit+meticulous suites green with Cython enabled AND with NO_CYTHON=1 (the pure-Python paths PyPy uses), and green under a PyPy-semantics simulation (an asyncgen firstiter hook holding strong references to defeat refcount finalization). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- faust/streams.py | 24 ++++--- tests/conftest.py | 24 +++++++ tests/functional/test_streams.py | 72 +++++++------------ .../assignor/test_copartitioned_assignor.py | 18 ++--- tests/unit/agents/test_agent.py | 4 -- tests/unit/test_streams.py | 7 -- 6 files changed, 72 insertions(+), 77 deletions(-) diff --git a/faust/streams.py b/faust/streams.py index 58c92046e..004d18ced 100644 --- a/faust/streams.py +++ b/faust/streams.py @@ -108,7 +108,7 @@ def _tracks_buffer_agen( @wraps(fun) def _create_agen(self: StreamT, *args: Any, **kwargs: Any) -> AsyncIterable: agen = fun(self, *args, **kwargs) - cast("Stream", self)._active_buffer_agens.add(agen) + cast("Stream", self)._active_agens.add(agen) return agen return _create_agen @@ -175,11 +175,12 @@ def __init__( self._processors = list(processors) if processors else [] self._on_start = on_start - # Live buffering generators created by take() and friends, - # closed explicitly in on_stop(). A WeakSet so that on CPython an - # abandoned generator is still finalized (and thus cleaned up) - # promptly by reference counting. - self._active_buffer_agens: "weakref.WeakSet[AsyncGenerator]" = weakref.WeakSet() + # Live async generators handed out by this stream -- + # __aiter__ iterators and the take() family -- closed explicitly + # in on_stop(). A WeakSet so that on CPython an abandoned + # generator is still finalized (and thus cleaned up) promptly by + # reference counting. + self._active_agens: "weakref.WeakSet[AsyncGenerator]" = weakref.WeakSet() # attach beacon to channel, or if iterable attach to current task. task = current_task(loop=self.loop) @@ -1076,7 +1077,7 @@ async def on_stop(self) -> None: # finalizes the generator (which on PyPy waits for a GC cycle that # may never come). ``aclose()`` on an exhausted generator is a # harmless no-op. - for agen in list(self._active_buffer_agens): + for agen in list(self._active_agens): try: await agen.aclose() except RuntimeError: @@ -1095,9 +1096,14 @@ def __next__(self) -> T: def __aiter__(self) -> AsyncIterator[T_co]: # pragma: no cover if _CStreamIterator is not None: - return self._c_aiter() + it = self._c_aiter() else: - return self._py_aiter() + it = self._py_aiter() + # Track the iterator so on_stop() can close it if the caller + # abandons it (see _tracks_buffer_agen for why this matters on + # interpreters without reference counting, e.g. PyPy). + self._active_agens.add(cast(AsyncGenerator, it)) + return it async def _c_aiter(self) -> AsyncIterator[T_co]: # pragma: no cover self.log.dev("Using Cython optimized __aiter__") diff --git a/tests/conftest.py b/tests/conftest.py index 4ce891806..eb256e50e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,7 @@ +import asyncio +import gc import os +import platform import threading import time from http import HTTPStatus @@ -44,6 +47,27 @@ def loop(event_loop): return event_loop +@pytest.fixture() +def event_loop(request): + """Per-test event loop (overrides pytest-asyncio's default). + + Identical to the pytest-asyncio 0.21 fixture, except that on PyPy we + finalize leftover async generators while the loop is still alive. + PyPy has no reference counting, so an abandoned async generator is + only finalized at a later GC cycle -- by which time this loop is + closed and its cleanup lands on a dead loop (leaking tasks into later + tests and misattributing warnings). A gc pass plus + ``shutdown_asyncgens()`` inside the owning test keeps the cleanup + here, matching CPython's prompt-finalization behavior. + """ + loop = asyncio.get_event_loop_policy().new_event_loop() + yield loop + if platform.python_implementation() == "PyPy": + gc.collect() + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.close() + + class _patching(object): def __init__(self, monkeypatch, request): self.monkeypatch = monkeypatch diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py index 483a74256..5890f4565 100644 --- a/tests/functional/test_streams.py +++ b/tests/functional/test_streams.py @@ -1,5 +1,4 @@ import asyncio -import platform from copy import copy from unittest.mock import Mock, patch @@ -40,9 +39,6 @@ def _prepare_app(app): return app -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio @pytest.mark.allow_lingering_tasks(count=1) async def test_simple(app, loop): @@ -54,9 +50,6 @@ async def test_simple(app, loop): assert await channel_empty(stream.channel) -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_async_iterator(app): async with new_stream(app) as stream: @@ -71,9 +64,6 @@ async def test_async_iterator(app): assert await channel_empty(stream.channel) -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_throw(app): async with new_stream(app) as stream: @@ -85,9 +75,6 @@ async def test_throw(app): await anext(streamit) -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_enumerate(app): async with new_stream(app) as stream: @@ -102,9 +89,6 @@ async def test_enumerate(app): assert await channel_empty(stream.channel) -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_items(app): async with new_stream(app) as stream: @@ -120,9 +104,6 @@ async def test_items(app): assert await channel_empty(stream.channel) -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_through(app): app._attachments.enabled = False @@ -255,9 +236,6 @@ async def test_stream_filter_acks_filtered_out_messages(app, event_loop): assert len(app.consumer.unacked) == 0 -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_acks_filtered_out_messages_when_using_take(app, event_loop): """ @@ -282,9 +260,6 @@ async def test_acks_filtered_out_messages_when_using_take(app, event_loop): assert len(acked) == len(initial_values) -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_events(app): async with new_stream(app) as stream: @@ -321,9 +296,6 @@ def assert_events_acked(events): raise -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) class Test_chained_streams: def _chain(self, app): root = new_stream(app) @@ -427,9 +399,6 @@ async def assert_was_stopped(self, leader, followers): assert node._stopped.is_set() -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_start_and_stop_Stream(app): s = new_topic_stream(app) @@ -445,9 +414,6 @@ async def _start_stop_stream(stream): await stream.stop() -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_ack(app): async with new_stream(app) as s: @@ -473,9 +439,6 @@ async def test_ack(app): assert not event.message.refcount -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_noack(app): async with new_stream(app) as s: @@ -496,9 +459,6 @@ async def test_noack(app): event.ack.assert_not_called() -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_acked_when_raising(app): async with new_stream(app) as s: @@ -536,9 +496,6 @@ async def test_acked_when_raising(app): assert not event2.message.refcount -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio @pytest.mark.allow_lingering_tasks(count=1) async def test_maybe_forward__when_event(app): @@ -551,9 +508,6 @@ async def test_maybe_forward__when_event(app): s.channel.send.assert_not_called() -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_maybe_forward__when_concrete_value(app): s = new_stream(app) @@ -828,6 +782,32 @@ async def test_take_cleanup_on_stream_stop(app): assert len(s._processors) == n_processors +@pytest.mark.asyncio +async def test_aiter_cleanup_on_stream_stop(app): + """An abandoned ``__aiter__`` generator is closed when the stream stops. + + Same contract as ``test_take_cleanup_on_stream_stop`` but for the main + stream iterator: its inner ``finally`` acks the last yielded event, and + that only runs when the generator is finalized. Holding a strong + reference blocks finalization on every interpreter (mimicking PyPy), so + the ack must come from the stream closing its tracked iterators on stop. + """ + async with new_stream(app) as s: + await s.channel.send(value=1) + agen = aiter(s) # hold a strong reference + event = None + async for value in agen: + assert value == 1 + event = mock_stream_event_ack(s) + break + + assert event + await s.stop() + + if not event.ack.called: + assert event.message.acked + + @pytest.mark.asyncio async def test_take__10(app, loop): s = new_stream(app) diff --git a/tests/meticulous/assignor/test_copartitioned_assignor.py b/tests/meticulous/assignor/test_copartitioned_assignor.py index 05c2f0ce8..2cad2759b 100644 --- a/tests/meticulous/assignor/test_copartitioned_assignor.py +++ b/tests/meticulous/assignor/test_copartitioned_assignor.py @@ -3,14 +3,19 @@ from collections import Counter from typing import MutableMapping -import pytest from hypothesis import assume, given, settings from hypothesis.strategies import integers from faust.assignor.client_assignment import CopartitionedAssignment from faust.assignor.copartitioned_assignor import CopartitionedAssignor -TEST_DEADLINE = 4000 +# No per-example deadline on PyPy: JIT warm-up makes the first examples of +# these property tests orders of magnitude slower than steady state, which +# trips hypothesis's DeadlineExceeded even though the code is correct. +if platform.python_implementation() == "PyPy": + TEST_DEADLINE = None +else: + TEST_DEADLINE = 4000 _topics = {"foo", "bar", "baz"} @@ -79,9 +84,6 @@ def client_removal_sticky( return True -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @given( partitions=integers(min_value=0, max_value=256), replicas=integers(min_value=0, max_value=64), @@ -100,9 +102,6 @@ def test_fresh_assignment(partitions, replicas, num_clients): assert is_valid(new_assignments, partitions, replicas) -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @given( partitions=integers(min_value=0, max_value=256), replicas=integers(min_value=0, max_value=64), @@ -139,9 +138,6 @@ def test_add_new_clients(partitions, replicas, num_clients, num_additional_clien assert clients_balanced(new_assignments) -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @given( partitions=integers(min_value=0, max_value=256), replicas=integers(min_value=0, max_value=64), diff --git a/tests/unit/agents/test_agent.py b/tests/unit/agents/test_agent.py index 25f033ac6..54f881f68 100644 --- a/tests/unit/agents/test_agent.py +++ b/tests/unit/agents/test_agent.py @@ -1,5 +1,4 @@ import asyncio -import platform from unittest.mock import ANY, call, patch import pytest @@ -954,9 +953,6 @@ def test_channel_iterator(self, *, agent): def test_label(self, *, agent): assert label(agent) - @pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" - ) async def test_context_calls_sink(self, *, agent): class SinkCalledException(Exception): pass diff --git a/tests/unit/test_streams.py b/tests/unit/test_streams.py index 4a6cab481..4b7af428d 100644 --- a/tests/unit/test_streams.py +++ b/tests/unit/test_streams.py @@ -1,5 +1,4 @@ import asyncio -import platform from collections import defaultdict from contextlib import ExitStack from unittest.mock import Mock, patch @@ -123,9 +122,6 @@ async def test_echo(self, *, stream, app): await echoing("val") channel.send.assert_called_once_with(value="val") - @pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" - ) @pytest.mark.asyncio @pytest.mark.allow_lingering_tasks(count=1) async def test_aiter_tracked(self, *, stream, app): @@ -141,9 +137,6 @@ async def test_aiter_tracked(self, *, stream, app): else: event.ack.assert_called_once_with() - @pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" - ) @pytest.mark.asyncio @pytest.mark.allow_lingering_tasks(count=1) async def test_aiter_tracked__CancelledError(self, *, stream, app): From b6da4ebfabac2494adaad1e0a33fcef36e7a1627 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 15:49:46 +0000 Subject: [PATCH 4/6] fix(assignor): break standby-assignment livelock exposed on PyPy CopartitionedAssignor._assign_round_robin frees a slot on an exhausted client by evicting an arbitrary partition (set.pop) and re-queueing it. When that arbitrary victim is the partition's only possible home, two partitions ping-pong between the work queue and the same client forever. Whether it happens depends on set iteration order, so identical input (e.g. 4 clients / 16 partitions / 1 replica) completes in <1ms on CPython but livelocks indefinitely on PyPy -- the reason the copartitioned-assignor property tests were skipped there. Evict a victim that some other not-yet-exhausted client can immediately take, in deterministic (sorted) order, so the re-queued partition makes forward progress instead of bouncing straight back. This terminates on every interpreter and keeps the assignment valid. Unskip the three property tests on PyPy (they now pass with the full example budget in ~4s) and add a regression test asserting the standby path terminates for any replicas >= 1. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- faust/assignor/copartitioned_assignor.py | 33 ++++++++++++++++- .../assignor/test_copartitioned_assignor.py | 37 +++++++++++++++---- 2 files changed, 61 insertions(+), 9 deletions(-) diff --git a/faust/assignor/copartitioned_assignor.py b/faust/assignor/copartitioned_assignor.py index ada471793..34969e4fd 100644 --- a/faust/assignor/copartitioned_assignor.py +++ b/faust/assignor/copartitioned_assignor.py @@ -259,8 +259,39 @@ def _assign_round_robin(self, unassigned: Iterable[int], active: bool) -> None: and assigment.can_assign(partition, active) ) ) # By above assertion, should never throw error - unassigned_partition = assign_to.pop_partition(active) + unassigned_partition = self._free_up_partition(assign_to, active) unassigned.append(unassigned_partition) # Assign partition assign_to.assign_partition(partition, active) + + def _free_up_partition( + self, assignment: CopartitionedAssignment, active: bool + ) -> int: + # Evict a partition from an exhausted client so the partition we are + # placing can take its slot. Prefer a victim that some *other* + # not-yet-exhausted client can immediately take, so the re-queued + # victim makes forward progress instead of bouncing straight back + # here. ``CopartitionedAssignment.pop_partition`` pops an arbitrary + # set element, and when the arbitrary choice is the victim's only + # possible home two partitions ping-pong between the queue and this + # client forever. Whether that happens depends on set iteration + # order, so the livelock strikes on PyPy while CPython (with a + # different order) escapes it for the same input. Choosing a + # re-homable victim in deterministic (sorted) order makes the loop + # terminate on every interpreter. + held = assignment.get_assigned_partitions(active) + for candidate in sorted(held): + for other in self._client_assignments.values(): + if other is assignment: + continue + if not self._client_exhausted(other, active) and other.can_assign( + candidate, active + ): + assignment.unassign_partition(candidate, active) + return candidate + # No immediately re-homable victim: fall back to a deterministic + # arbitrary choice (still correct, just not progress-guaranteed). + candidate = min(held) + assignment.unassign_partition(candidate, active) + return candidate diff --git a/tests/meticulous/assignor/test_copartitioned_assignor.py b/tests/meticulous/assignor/test_copartitioned_assignor.py index 2cad2759b..e0fb1f66b 100644 --- a/tests/meticulous/assignor/test_copartitioned_assignor.py +++ b/tests/meticulous/assignor/test_copartitioned_assignor.py @@ -1,5 +1,4 @@ import copy -import platform from collections import Counter from typing import MutableMapping @@ -9,13 +8,7 @@ from faust.assignor.client_assignment import CopartitionedAssignment from faust.assignor.copartitioned_assignor import CopartitionedAssignor -# No per-example deadline on PyPy: JIT warm-up makes the first examples of -# these property tests orders of magnitude slower than steady state, which -# trips hypothesis's DeadlineExceeded even though the code is correct. -if platform.python_implementation() == "PyPy": - TEST_DEADLINE = None -else: - TEST_DEADLINE = 4000 +TEST_DEADLINE = 4000 _topics = {"foo", "bar", "baz"} @@ -175,3 +168,31 @@ def test_remove_clients(partitions, replicas, num_clients, num_removal_clients): assert client_removal_sticky(old_assignments, new_assignments) elif partitions >= num_clients - num_removal_clients: assert clients_balanced(new_assignments) + + +@given( + partitions=integers(min_value=2, max_value=256), + num_clients=integers(min_value=2, max_value=64), + replicas=integers(min_value=1, max_value=8), +) +@settings(deadline=TEST_DEADLINE) +def test_standby_assignment_terminates(partitions, num_clients, replicas): + # Regression test for a standby-assignment livelock. + # + # When the round-robin pass has to free up a slot on an exhausted client, + # it used to evict an arbitrary partition (``set.pop``). If that arbitrary + # choice was the partition's only possible home, two partitions could + # ping-pong between the work queue and the same client forever. Whether it + # happened depended on set iteration order, so the assignment hung on PyPy + # while completing on CPython for identical input. Any ``replicas >= 1`` + # case exercises the standby path; this asserts termination and a valid, + # fully-replicated assignment regardless of interpreter. + assume(replicas < num_clients) + client_assignments = { + str(client): CopartitionedAssignment(topics=_topics) + for client in range(num_clients) + } + new_assignments = CopartitionedAssignor( + _topics, client_assignments, partitions, replicas=replicas + ).get_assignment() + assert is_valid(new_assignments, partitions, replicas) From 15e860359177d1149ab65e7395dbc9bd94657495 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 15:49:56 +0000 Subject: [PATCH 5/6] test(streams): finalize abandoned generators via aclose; raise PyPy CI timeout Replace the timeout-poll wait_for_stream_ack helper with an explicit generator aclose(): the take()/iterator finally blocks (which ack consumed events) only run when the generator is finalized, and abandoning it defers that to interpreter finalization -- immediate on CPython via refcounting, deferred to a maybe-never GC cycle on PyPy. Closing it runs the finally now, deterministically, on every interpreter. Callers hold the generator and close it instead of leaning on asyncio.sleep(0) yields. The unit aiter test drains via a bounded gc.collect() + sleep loop for the same reason (its agent-actor generator can't be reached to close directly). Bump the PyPy CI leg's timeout from 10 to 15 minutes: it now runs the formerly-skipped tests and PyPy is slower, and the old cap already clipped the run at ~96%. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- .github/workflows/python-package.yml | 5 +- tests/functional/test_streams.py | 77 +++++++++++++--------------- tests/unit/test_streams.py | 9 ++-- 3 files changed, 45 insertions(+), 46 deletions(-) diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 41790485d..b889a7e4e 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -70,7 +70,10 @@ jobs: test-pypy: name: 'Python pypy3.11/Cython: false' runs-on: ubuntu-latest - timeout-minutes: 10 + # PyPy runs the pure-Python paths and is markedly slower than CPython, and + # this leg now runs the formerly-skipped tests too; give it headroom over + # the old 10-minute cap, which already clipped the run at ~96%. + timeout-minutes: 15 continue-on-error: true env: USE_CYTHON: 'false' diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py index 5890f4565..75875b1c1 100644 --- a/tests/functional/test_streams.py +++ b/tests/functional/test_streams.py @@ -1,4 +1,5 @@ import asyncio +from contextlib import suppress from copy import copy from unittest.mock import Mock, patch @@ -422,7 +423,8 @@ async def test_ack(app): await s.channel.send(value=2) event = None i = 1 - async for value in s: + it = aiter(s) + async for value in it: assert value == i i += 1 last_to_ack = value == 2 @@ -430,10 +432,7 @@ async def test_ack(app): if value == 2: break assert event - # need one sleep on Python 3.6.0-3.6.6 + 3.7.0 - # need two sleeps on Python 3.6.7 + 3.7.1 :-/ - await asyncio.sleep(0) # needed for some reason - await asyncio.sleep(0) # needed for some reason + await wait_for_stream_ack(event, it) if not event.ack.called: assert event.message.acked assert not event.message.refcount @@ -466,31 +465,27 @@ async def test_acked_when_raising(app): await s.channel.send(value=2) event1 = None + it1 = aiter(s) with pytest.raises(RuntimeError): - async for value in s: + async for value in it1: event1 = mock_stream_event_ack(s) assert value == 1 raise RuntimeError assert event1 - # need one sleep on Python 3.6.0-3.6.6 + 3.7.0 - # need two sleeps on Python 3.6.7 + 3.7.1 :-/ - await asyncio.sleep(0) # needed for some reason - await asyncio.sleep(0) # needed for some reason + await wait_for_stream_ack(event1, it1) if not event1.ack.called: assert event1.message.acked assert not event1.message.refcount event2 = None + it2 = aiter(s) with pytest.raises(RuntimeError): - async for value in s: + async for value in it2: event2 = mock_stream_event_ack(s) assert value == 2 raise RuntimeError assert event2 - # need one sleep on Python 3.6.0-3.6.6 + 3.7.0 - # need two sleeps on Python 3.6.7 + 3.7.1 :-/ - await asyncio.sleep(0) # needed for some reason - await asyncio.sleep(0) # needed for some reason + await wait_for_stream_ack(event2, it2) if not event2.ack.called: assert event2.message.acked assert not event2.message.refcount @@ -526,23 +521,19 @@ def mock_event_ack(event, return_value=False): return event -async def wait_for_stream_ack(event, *, timeout=1.0): - """Wait until a taken event has been acked. +async def wait_for_stream_ack(event, agen): + """Ensure a consumed event has been acked, deterministically. - ``stream.take()`` acks the events it consumes from a background task, so - the ack has not necessarily run by the time the ``async for`` body returns. - Polling for the ack with a timeout keeps the assertion robust on slower - interpreters (notably PyPy) where the couple of ``asyncio.sleep(0)`` yields - the tests used to rely on aren't enough for that task to run. If the ack - never lands the loop simply times out and returns, leaving the caller's - assertions to report the actual state. + The ack runs in the ``take()``/iterator generator's ``finally`` block, + which only executes when the generator is finalized. Abandoning the + generator (``break`` or an exception in ``async for``) defers that to + interpreter finalization -- immediate on CPython via reference counting, + but only at a later (maybe never) GC cycle on PyPy. Closing the generator + explicitly runs its ``finally`` now, on every interpreter; ``aclose()`` on + an already-finished generator is a harmless no-op. """ - loop = asyncio.get_event_loop() - deadline = loop.time() + timeout - while loop.time() < deadline: - if event.ack.called or event.message.acked: - return - await asyncio.sleep(0) + with suppress(RuntimeError): + await agen.aclose() async def get_event_from_value(stream, value, key=None): @@ -732,14 +723,15 @@ async def test_take(app): assert s.enable_acks is True await s.channel.send(value=1) event = None - async for value in s.take(1, within=1): + buffer = s.take(1, within=1) + async for value in buffer: assert value == [1] assert s.enable_acks is False event = mock_stream_event_ack(s) break assert event - await wait_for_stream_ack(event) + await wait_for_stream_ack(event, buffer) if not event.ack.called: assert event.message.acked @@ -900,9 +892,10 @@ async def test_take_wit_timestamp(app): assert s.enable_acks is True await s.channel.send(value={"id": 1}) event = None - async for value in s.take_with_timestamp( + buffer = s.take_with_timestamp( 1, within=1, timestamp_field_name="test_timestamp" - ): + ) + async for value in buffer: assert "test_timestamp" in value[0].keys() assert isinstance(value[0]["test_timestamp"], float) assert s.enable_acks is False @@ -910,7 +903,7 @@ async def test_take_wit_timestamp(app): break assert event - await wait_for_stream_ack(event) + await wait_for_stream_ack(event, buffer) if not event.ack.called: assert event.message.acked @@ -924,16 +917,17 @@ async def test_take_wit_timestamp_wit_simple_value(app): assert s.enable_acks is True await s.channel.send(value=1) event = None - async for value in s.take_with_timestamp( + buffer = s.take_with_timestamp( 1, within=1, timestamp_field_name="test_timestamp" - ): + ) + async for value in buffer: assert value == [1] assert s.enable_acks is False event = mock_stream_event_ack(s) break assert event - await wait_for_stream_ack(event) + await wait_for_stream_ack(event, buffer) if not event.ack.called: assert event.message.acked @@ -947,16 +941,15 @@ async def test_take_wit_timestamp_without_timestamp_field(app): assert s.enable_acks is True await s.channel.send(value=1) event = None - async for value in s.take_with_timestamp( - 1, within=1, timestamp_field_name=None - ): + buffer = s.take_with_timestamp(1, within=1, timestamp_field_name=None) + async for value in buffer: assert value == [1] assert s.enable_acks is False event = mock_stream_event_ack(s) break assert event - await wait_for_stream_ack(event) + await wait_for_stream_ack(event, buffer) if not event.ack.called: assert event.message.acked diff --git a/tests/unit/test_streams.py b/tests/unit/test_streams.py index 4b7af428d..c28765915 100644 --- a/tests/unit/test_streams.py +++ b/tests/unit/test_streams.py @@ -1,4 +1,5 @@ import asyncio +import gc from collections import defaultdict from contextlib import ExitStack from unittest.mock import Mock, patch @@ -189,9 +190,11 @@ async def agent(stream): await s.channel.put(new_event(app, topic="bar", value=sentinel)) s._on_message_in = Mock() await got_sentinel.wait() - await asyncio.sleep(0) - await asyncio.sleep(0) - await asyncio.sleep(0) + for _ in range(20): + if event.ack.called or event.message.acked: + break + gc.collect() + await asyncio.sleep(0.05) s._on_message_in.assert_called_once_with( event.message.tp, event.message.offset, From 4ab51a5248e2f41c6f90357a4bf93bb938137636 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 16:18:18 +0000 Subject: [PATCH 6/6] fix(streams): catch PyPy's ValueError when closing a running generator on stop on_stop() closes the async generators the stream handed out, skipping any that are still executing in another task (an agent actor iterating the stream during a rebalance). That "still running" case raises RuntimeError on CPython but ValueError ("async generator already executing") on PyPy, and only RuntimeError was caught -- so on PyPy the ValueError propagated out of on_stop, aborting the stream's stop mid-way and leaving stale entries in the topic's active_partitions. That tripped the conductor's `active_partitions.issubset(assigned)` assertion during isolated-partition rebalances (test_agent_isolated_partitions_rebalancing). Catch both exception types in on_stop and in the wait_for_stream_ack test helper, so closing a running generator is always the intended no-op. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- faust/streams.py | 8 ++++++-- tests/functional/test_streams.py | 6 ++++-- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/faust/streams.py b/faust/streams.py index 004d18ced..784bdf376 100644 --- a/faust/streams.py +++ b/faust/streams.py @@ -1080,8 +1080,12 @@ async def on_stop(self) -> None: for agen in list(self._active_agens): try: await agen.aclose() - except RuntimeError: - # Still running in another task; its owner will clean up. + except (RuntimeError, ValueError): + # Generator is still running in another task (e.g. an agent + # actor iterating this stream during a rebalance); its owner + # will clean it up. CPython raises RuntimeError for this, + # PyPy raises ValueError ("async generator already + # executing") -- catch both so on_stop never aborts here. pass self._passive = False self._passive_started.clear() diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py index 75875b1c1..d9439887e 100644 --- a/tests/functional/test_streams.py +++ b/tests/functional/test_streams.py @@ -530,9 +530,11 @@ async def wait_for_stream_ack(event, agen): interpreter finalization -- immediate on CPython via reference counting, but only at a later (maybe never) GC cycle on PyPy. Closing the generator explicitly runs its ``finally`` now, on every interpreter; ``aclose()`` on - an already-finished generator is a harmless no-op. + an already-finished generator is a harmless no-op. A still-running + generator raises RuntimeError on CPython / ValueError on PyPy; both are + ignored (its owner will finalize it). """ - with suppress(RuntimeError): + with suppress(RuntimeError, ValueError): await agen.aclose()