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