Key topic_buffer_full by TP in both conductors

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*:

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)
 

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))

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.
 

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