From 3182eb121024ae490fbb3afec67033a80189d7f7 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Sun, 19 Jul 2026 16:02:27 +0000 Subject: [PATCH 1/2] 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 03e5fde6d04165513f1f44daa3867f419bfa020f Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 20 Jul 2026 12:41:32 +0000 Subject: [PATCH 2/2] test(streams): force gc in wait_for_stream_ack so take() acks land on PyPy The polling helper alone still left the take() tests failing on the PyPy CI leg. Root cause: breaking out of `async for value in s.take(...)` leaves the take() async generator suspended, and the `finally:` block that acks the buffered events only runs when the generator is finalized. CPython's refcounting finalizes it immediately, so polling a few event-loop ticks was enough -- but PyPy has no refcounting, so without an explicit gc.collect() the finalizer (and therefore the ack) may never run at all, no matter how long the helper waited. Nudge the collector on each poll round (the same trick faust's own Consumer.wait_empty() uses) and yield with a real sleep so the scheduled aclose() gets to run. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- tests/functional/test_streams.py | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py index 469ec788b..521c99378 100644 --- a/tests/functional/test_streams.py +++ b/tests/functional/test_streams.py @@ -1,4 +1,5 @@ import asyncio +import gc import platform from copy import copy from unittest.mock import Mock, patch @@ -575,20 +576,25 @@ def mock_event_ack(event, return_value=False): 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. + Breaking out of ``async for value in s.take(...)`` leaves the ``take()`` + async generator suspended; the ``finally:`` block that acks the buffered + events only runs once the generator is finalized. CPython's refcounting + finalizes it as soon as the loop exits (so the ack lands within a couple + of event-loop ticks), but PyPy has no refcounting -- without an explicit + ``gc.collect()`` the finalizer (and thus the ack) may never run at all, + no matter how long we sleep. So poll for the ack, nudging the collector + each round to trigger generator finalization, exactly like + ``Consumer.wait_empty()`` does in faust itself. If the ack never lands + the loop 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) + gc.collect() + await asyncio.sleep(0.05) async def get_event_from_value(stream, value, key=None):