From 8224172a990bc734e74664d1886dbf7b255cf7c0 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Sun, 19 Jul 2026 10:46:17 +0000 Subject: [PATCH 1/2] Fix #319: record event runtime for events acked via stream.take() Stream.take() disables acks and acks its buffered events manually after the main iteration loop has already yielded. The main loop only calls on_stream_event_out when do_ack is set, so for take() the out-event was delivered by Stream.ack() -- but without the sensor state returned by on_stream_event_in. on_stream_event_out needs that state to compute the runtime, so events_runtime was never populated and metrics such as events_runtime_ms_sum stayed at zero. Capture the per-sensor state on the event in SensorDelegate.on_stream_event_in (covers both the Python and Cython stream iterators, which share that delegate) and forward it from Stream.ack(), clearing it afterwards so a second ack cannot double-count. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- faust/events.py | 7 ++++++- faust/sensors/base.py | 8 +++++++- faust/streams.py | 8 +++++++- faust/types/events.py | 4 +++- tests/functional/test_streams.py | 25 +++++++++++++++++++++++++ tests/unit/test_streams.py | 6 ++++++ 6 files changed, 54 insertions(+), 4 deletions(-) diff --git a/faust/events.py b/faust/events.py index e45dd4fe1..5c64c352f 100644 --- a/faust/events.py +++ b/faust/events.py @@ -2,7 +2,7 @@ import typing from types import TracebackType -from typing import Any, Awaitable, Optional, Type, Union, cast +from typing import Any, Awaitable, Dict, Optional, Type, Union, cast from faust.types import ( AppT, @@ -128,6 +128,11 @@ def __init__( self.headers = {} self.acked: bool = False + #: Per-sensor state captured by ``Sensor.on_stream_event_in`` and + #: handed back to ``on_stream_event_out`` when the event is acked. + #: Stored here so deferred acks (e.g. ``Stream.take``) can still + #: report the event runtime. See issue #319. + self.sensor_state: Optional[Dict] = None async def send( self, diff --git a/faust/sensors/base.py b/faust/sensors/base.py index 46f679e67..6f1c350b8 100644 --- a/faust/sensors/base.py +++ b/faust/sensors/base.py @@ -182,10 +182,16 @@ def on_stream_event_in( self, tp: TP, offset: int, stream: StreamT, event: EventT ) -> Optional[Dict]: """Call when stream starts processing an event.""" - return { + state = { sensor: sensor.on_stream_event_in(tp, offset, stream, event) for sensor in self._sensors } + # Remember the per-sensor state on the event itself so that a + # deferred ack (e.g. Stream.take(), which acks manually long after + # the main loop yielded) can still deliver it to on_stream_event_out + # and record the true event runtime. See issue #319. + event.sensor_state = state + return state def on_stream_event_out( self, tp: TP, offset: int, stream: StreamT, event: EventT, state: Dict = None diff --git a/faust/streams.py b/faust/streams.py index 9f4286a3f..9a775eb86 100644 --- a/faust/streams.py +++ b/faust/streams.py @@ -1268,7 +1268,13 @@ async def ack(self, event: EventT) -> bool: message = event.message tp = message.tp offset = message.offset - self._on_stream_event_out(tp, offset, self, event) + # Forward the sensor state captured in on_stream_event_in so the + # event runtime is recorded even when acking is deferred (e.g. the + # buffered events produced by Stream.take()). Clear it afterwards so + # a second ack of the same event cannot double-count. See issue #319. + sensor_state = getattr(event, "sensor_state", None) + self._on_stream_event_out(tp, offset, self, event, sensor_state) + event.sensor_state = None if last_stream_to_ack: self._on_message_out(tp, offset, message) return last_stream_to_ack diff --git a/faust/types/events.py b/faust/types/events.py index 9b1498609..33d4d807e 100644 --- a/faust/types/events.py +++ b/faust/types/events.py @@ -4,6 +4,7 @@ Any, AsyncContextManager, Awaitable, + Dict, Generic, Mapping, Optional, @@ -38,8 +39,9 @@ class EventT(Generic[T], AsyncContextManager): headers: Mapping message: Message acked: bool + sensor_state: Optional[Dict] - __slots__ = ("app", "key", "value", "headers", "message", "acked") + __slots__ = ("app", "key", "value", "headers", "message", "acked", "sensor_state") @abc.abstractmethod def __init__( diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py index 851594155..d38e7a3de 100644 --- a/tests/functional/test_streams.py +++ b/tests/functional/test_streams.py @@ -9,6 +9,7 @@ import faust from faust.exceptions import ImproperlyConfigured +from faust.sensors import Monitor from faust.streams import maybe_forward from tests.helpers import AsyncMock @@ -777,6 +778,30 @@ async def test_take(app): assert s.enable_acks is True +@pytest.mark.skipif( + platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" +) +@pytest.mark.asyncio +async def test_take__records_event_runtime(app): + # Regression test for #319: events buffered by take() are acked manually + # after the main loop has already yielded, so on_stream_event_out used to + # be called without the sensor state and no event runtime was recorded. + monitor = Monitor() + app.sensors.add(monitor) + async with new_stream(app) as s: + await s.channel.send(value=1) + async for value in s.take(1, within=1): + assert value == [1] + break + # let take()'s finally ack the buffered event + await asyncio.sleep(0) + await asyncio.sleep(0) + + # on_stream_event_out ran with the captured sensor state, so a runtime + # was appended (would be empty before the fix). + assert monitor.events_runtime + + @pytest.mark.asyncio async def test_take__10(app, loop): s = new_stream(app) diff --git a/tests/unit/test_streams.py b/tests/unit/test_streams.py index 4a6cab481..83149eb4a 100644 --- a/tests/unit/test_streams.py +++ b/tests/unit/test_streams.py @@ -210,15 +210,21 @@ async def agent(stream): async def test_ack(self, *, stream): event = Mock() event.ack.return_value = True + event.sensor_state = {"sensor": "state"} stream._on_stream_event_out = Mock() stream._on_message_out = Mock() assert await stream.ack(event) + # the sensor state captured in on_stream_event_in is forwarded so + # deferred acks still record the event runtime (issue #319) stream._on_stream_event_out.assert_called_once_with( event.message.tp, event.message.offset, stream, event, + {"sensor": "state"}, ) + # ...and cleared afterwards so a second ack can't double-count + assert event.sensor_state is None stream._on_message_out.assert_called_once_with( event.message.tp, event.message.offset, From 630d7ff1fdfd72845fecdf9289fa418716cddfbb Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 10:28:23 +0000 Subject: [PATCH 2/2] Drop obsolete PyPy skip from the take() runtime regression test master's #716 ("full PyPy support -- all PyPy skips removed") deleted every PyPy skip in tests/functional/test_streams.py together with the now-unused `import platform`. This branch's new test_take__records_event_runtime still carried a @pytest.mark.skipif(platform.python_implementation() == "PyPy", ...) guard. Because the two sides touch different lines, merging master would have applied cleanly and left a use of `platform` with no import -- the decorator is evaluated at import time, so pytest would abort during collection with NameError and take the whole suite down on every aiokafka leg, plus F821 in the lint job. (PR #699 hit exactly this after its master merge.) Remove the decorator rather than keeping the import: #716 made PyPy a fully supported target, so this test should run there like the rest. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017K8xAH8Z3xWKHhNRCG2mg1 --- tests/functional/test_streams.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/tests/functional/test_streams.py b/tests/functional/test_streams.py index d38e7a3de..4018ea608 100644 --- a/tests/functional/test_streams.py +++ b/tests/functional/test_streams.py @@ -778,9 +778,6 @@ async def test_take(app): assert s.enable_acks is True -@pytest.mark.skipif( - platform.python_implementation() == "PyPy", reason="Not yet supported on PyPy" -) @pytest.mark.asyncio async def test_take__records_event_runtime(app): # Regression test for #319: events buffered by take() are acked manually