From 9ed8f4b3acc7afab603581cbf9cf7586d7b3a439 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 21:31:39 +0000 Subject: [PATCH] Key topic_buffer_full by TP in both conductors `Monitor.topic_buffer_full` is a `Counter[TP]`, and two paths report into it: the pressure-high callback, which passes a TP, and the full-queue path, which passed the *channel*. The same partition therefore accumulated under two different keys depending on which path noticed the buffer was full -- splitting its count, and adding a second `/stats` entry labelled by channel for a partition already listed by TP. Both implementations had it, which is why it went unfixed for so long: the comment in faust/transport/conductor.py recorded the defect and explicitly declined to fix it, because correcting one twin alone would have made the two disagree. With the parity suite in place that objection is gone -- both are corrected here, together, and the suite holds them level. The `# type: ignore[arg-type]` on the call goes away with it; `mypy -p faust` is clean without it, which is the type checker confirming the argument is now the one the sensor declares. ## Note on what parity testing does not do The conductor parity tests were green throughout, before and after. Both implementations passed the channel, so they agreed with each other perfectly while both were wrong. A differential test only finds *divergence*; a shared mistake is invisible to it. So the coverage added here is deliberately not another comparison: * the full-queue parity test now records the sensor's *argument* rather than a call count, and asserts it equals the TP; * a new test drives a real `Monitor` through the full-queue path and asserts every key of `topic_buffer_full` is a TP. It is parametrised over both implementations rather than comparing them, and runs against the pure-Python conductor even when the extension is absent, since the defect was in both. Verified by reverting both twins and confirming each new assertion fails: `Got: []` and `keyed it by ['Topic']`. Suite green in every configuration: extensions built (2272 passed), absent (2208 passed), free-threaded 3.14t under PYTHON_GIL=0 (2276 passed), and `mypy -p faust` clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- docs/developerguide/cython.rst | 24 ++++-- faust/transport/_cython/conductor.pyx | 7 +- faust/transport/conductor.py | 19 ++--- tests/unit/test_cython_parity.py | 6 +- tests/unit/transport/test_conductor_parity.py | 84 ++++++++++++++++--- 5 files changed, 111 insertions(+), 29 deletions(-) diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst index 0828a3368..4548fa4b0 100644 --- a/docs/developerguide/cython.rst +++ b/docs/developerguide/cython.rst @@ -61,11 +61,25 @@ repeatedly: * **#608**, *"Fix cython stream_event_in to match python impl"* -- shipped, and fixed only after the fact. -* ``Conductor.on_topic_buffer_full`` passes a channel where a ``TP`` is - expected, so ``Monitor``'s per-TP counts are wrong. The comment in - ``faust/transport/conductor.py`` records that the defect is **deliberately - left unfixed**, because fixing one twin alone would make the two disagree. - The duplication turned a small bug into one nobody wants to touch. +* ``Conductor``'s full-queue path passed a channel to + ``on_topic_buffer_full`` where a ``TP`` was expected, so + ``Monitor.topic_buffer_full`` -- a ``Counter[TP]`` -- was keyed by channel + from that path and by ``TP`` from the pressure-high path. The same + partition accumulated under two keys, splitting its count and adding a + second ``/stats`` entry for it. + + Both twins had it, so for a long time the comment in + ``faust/transport/conductor.py`` recorded the defect as **deliberately left + unfixed**: correcting one alone would have made them disagree. The + duplication turned a one-line bug into one nobody wanted to touch. It is + fixed now -- in both, together, which is what the parity suites make safe. + + Worth noting what did *not* catch it: the parity tests were green + throughout, because both implementations were wrong in the same way. A + differential test only finds divergence. Shared mistakes need an assertion + about the behaviour itself, which is why the conductor suite now checks that + the sensor is handed a ``TP`` rather than only that both sides hand it the + same thing. * ``StreamIterator._try_get_quick_value`` carried two bugs that concealed each other. ``chan_queue_empty`` holds the bound ``queue.empty`` *method*: diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx index 0e0378b08..302100c48 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/faust/transport/_cython/conductor.pyx @@ -114,7 +114,12 @@ cdef class ConductorHandler: delivered.add(channel) async def _handle_full(self, event, chan, delivered): - self.on_topic_buffer_full(chan) + # ``self.tp``, not the channel: the sensor takes a ``TP`` (as + # ``on_pressure_high`` below passes), and ``Monitor.topic_buffer_full`` + # is a ``Counter[TP]``. Passing the channel here keyed part of that + # counter by channel instead, so the same partition was counted under + # two different keys depending on which path reported it. + self.on_topic_buffer_full(self.tp) await chan.put(event) delivered.add(chan) diff --git a/faust/transport/conductor.py b/faust/transport/conductor.py index 853e38428..1cc99f83c 100644 --- a/faust/transport/conductor.py +++ b/faust/transport/conductor.py @@ -164,16 +164,15 @@ async def on_message(message: Message) -> None: ) delivered.add(chan) if full: - for _, dest_chan in full: - # XXX wrong argument: ``SensorT.on_topic_buffer_full`` - # takes a ``TP`` (as ``on_pressure_high`` above is - # passed), but a channel is passed here, so - # ``Monitor.topic_buffer_full`` is keyed by channel - # and its per-TP counts are wrong. The Cython twin - # in ``_cython/conductor.pyx`` has the same bug; - # fixing either alone would make them disagree, so - # the defect is only recorded here, not fixed. - on_topic_buffer_full(dest_chan) # type: ignore[arg-type] + for _ in full: + # ``tp``, not the channel: the sensor takes a + # ``TP`` (as ``on_pressure_high`` above passes), + # and ``Monitor.topic_buffer_full`` is a + # ``Counter[TP]``. Passing the channel here keyed + # part of that counter by channel instead, so the + # same partition was counted under two different + # keys depending on which path reported it. + on_topic_buffer_full(tp) await asyncio.wait( [ asyncio.ensure_future(dest_chan.put(dest_event)) diff --git a/tests/unit/test_cython_parity.py b/tests/unit/test_cython_parity.py index f64cac810..c505885c2 100644 --- a/tests/unit/test_cython_parity.py +++ b/tests/unit/test_cython_parity.py @@ -6,9 +6,9 @@ been enforcing that -- and the duplication has already cost real bugs: * #608, "Fix cython stream_event_in to match python impl"; -* the ``on_topic_buffer_full`` defect recorded but deliberately left unfixed - in ``faust/transport/conductor.py``, because fixing one twin alone would - make them disagree; +* the ``on_topic_buffer_full`` defect that sat recorded but unfixed in + ``faust/transport/conductor.py`` for as long as it did precisely because + fixing one twin alone would have made them disagree; * the ``_try_get_quick_value`` pair fixed alongside this file, where the extension's queue fast path was both unreachable and, had it run, wrong. diff --git a/tests/unit/transport/test_conductor_parity.py b/tests/unit/transport/test_conductor_parity.py index 92b3583ff..4f3a8b928 100644 --- a/tests/unit/transport/test_conductor_parity.py +++ b/tests/unit/transport/test_conductor_parity.py @@ -16,9 +16,14 @@ The existing conductor tests replace the handler with an ``AsyncMock`` and assert it was called, so none of that logic was covered on either side. The duplication has already produced bugs that only differential testing catches -- -see ``docs/developerguide/cython.rst`` -- including one recorded in -``faust/transport/conductor.py`` as deliberately unfixed, because correcting one -twin alone would make the two disagree. +see ``docs/developerguide/cython.rst``. + +Note the converse, though: a differential test only finds *divergence*. The +``on_topic_buffer_full`` defect (both implementations passed a channel where the +sensor wanted a ``TP``) kept these comparisons green the whole time it was +present, because both sides were wrong identically. Shared mistakes need an +assertion about the behaviour itself, so a few tests below check what a value +*is* and not only that both implementations produce the same one. Both implementations are driven against **the same** conductor and the same ``channels`` set, one after the other, rather than against two separately-built @@ -40,6 +45,7 @@ import pytest from faust.exceptions import KeyDecodeError, ValueDecodeError +from faust.sensors import Monitor from faust.transport.conductor import Conductor, ConductorHandler from faust.types import TP, Message from tests.helpers import AsyncMock @@ -189,7 +195,9 @@ def observations(self, message: Optional[Message] = None) -> Dict[str, Any]: "delivered": delivered, "n_delivered_total": sum(len(v) for v in delivered.values()), "n_decodes": len(self.decodes), - "buffer_full_sensor": len(self.buffer_full_sensor), + # The arguments, not just the count: what gets passed to + # `on_topic_buffer_full` is the metric's key. + "buffer_full_sensor": list(self.buffer_full_sensor), "consumer_buffer_full": len(self.consumer_buffer_full), "consumer_buffer_drop": len(self.consumer_buffer_drop), "key_decode_errors": sorted(self.key_decode_errors), @@ -398,6 +406,11 @@ async def test_parity__queue_full_path(harness) -> None: Filling the queue first drives ``_handle_full``, a separate branch in both implementations that also fires the ``on_topic_buffer_full`` sensor. + + The sensor argument is checked explicitly, not just for parity. Both + implementations used to pass the *channel* here, so they agreed with each + other and this comparison stayed green while both were wrong -- a shared + mistake is exactly what a differential test cannot see. """ async def scenario(handler, h): @@ -414,18 +427,69 @@ async def scenario(handler, h): chan.queue.get_nowait() await asyncio.wait_for(pending, timeout=5) return { - "buffer_full_sensor": len(h.buffer_full_sensor), - "consumer_buffer_full": len(h.consumer_buffer_full), - "consumer_buffer_drop": len(h.consumer_buffer_drop), + "buffer_full_sensor": list(h.buffer_full_sensor), + "consumer_buffer_full": list(h.consumer_buffer_full), + "consumer_buffer_drop": list(h.consumer_buffer_drop), "qsize": chan.queue.qsize(), "refcount": message.refcount, } results = await run_both(harness, scenario) assert_parity(results) - assert results["cython"][ - "buffer_full_sensor" - ], "the full-queue path did not fire the on_topic_buffer_full sensor" + reported = results["cython"]["buffer_full_sensor"] + assert reported, "the full-queue path did not fire the on_topic_buffer_full sensor" + assert all(arg == TP1 for arg in reported), ( + f"on_topic_buffer_full must be given the TP -- it is the key of " + f"Monitor.topic_buffer_full, a Counter[TP], and the pressure-high path " + f"already passes one. Got: {reported}" + ) + + +@pytest.mark.asyncio +@pytest.mark.conf(stream_buffer_maxsize=2) +@pytest.mark.parametrize("impl", IMPLS) +async def test_monitor_counts_buffer_full_by_tp(app, impl) -> None: + """``Monitor.topic_buffer_full`` must be keyed by TP, from either path. + + The counter is a ``Counter[TP]``, and two code paths report into it: the + pressure-high callback (which always passed a TP) and the full-queue path + (which passed the channel). The same partition therefore accumulated under + two different keys, so per-TP counts were split and ``/stats`` grew a second + entry labelled by channel for the same partition. + + Unlike the parity tests, this asserts the behaviour rather than agreement: + both implementations made the same mistake, so they agreed with each other + throughout. Runs against the pure-Python conductor too, since the defect + was in both. + """ + if impl == "cython" and ConductorHandler is None: + pytest.skip("conductor extension not built in place") + + monitor = Monitor() + app.sensors.add(monitor) + + h = Harness(app, n_channels=1) + # Undo the harness's sensor stub: the real delegate is what is under test. + app.sensors.on_topic_buffer_full = monitor.on_topic_buffer_full + + handler = h.build(impl) + chan = h.channels[0] + for i in range(2): # stream_buffer_maxsize -> forces the full-queue path + chan.queue.put_nowait(f"filler{i}") + + pending = asyncio.ensure_future(handler(h.message())) + await asyncio.sleep(0) + chan.queue.get_nowait() + chan.queue.get_nowait() + await asyncio.wait_for(pending, timeout=5) + + assert monitor.topic_buffer_full, "the full-queue path reported nothing" + bad = [key for key in monitor.topic_buffer_full if not isinstance(key, TP)] + assert not bad, ( + f"Monitor.topic_buffer_full is a Counter[TP], but the {impl} conductor " + f"keyed it by {[type(k).__name__ for k in bad]}: {bad}" + ) + assert monitor.topic_buffer_full[TP1] > 0 @requires_cython_conductor