Add end-to-end conductor parity tests, and fix the divergence they found

docs/developerguide/cython.rst

@@ -85,9 +85,33 @@ repeatedly:
   So the extension quietly did *more* work than the pure-Python code it was
   meant to accelerate, for as long as it has existed.
 
+* ``ConductorHandler`` had **the same shape of fault, independently**.  The
+  conductor deserializes a message once and reuses the event for every channel
+  whose ``(key_type, value_type)`` pair matches.  In the extension,
+  ``event_keyid`` was only ever assigned from ``_decode()``, which returned it
+  *unchanged* on the first pass -- so it stayed ``None`` forever and the reuse
+  branch was dead.  Every subscribed channel re-deserialized the payload.
+
+  That masked a second fault, again: had the keyid ever been set, a mismatched
+  pair fell off the end of ``_decode`` and returned a bare ``None``, which
+  unpacking into two names raises :exc:`TypeError` on.  Fixing the reuse alone
+  would have converted a silent inefficiency into a crash on any topic whose
+  subscribers declare different key or value types.
+
+  It was not only a performance difference.  A channel whose event is reused
+  never calls ``decode`` at all, so a channel that *would* have failed to
+  deserialize raised no error under the pure-Python conductor and raised one
+  under the extension -- changing which channels got the message, and how many
+  acks the message received.
+
 None of these were caught by a test, because until recently no test ever
 imported the compiled modules.
 
+The parity suites are :file:`tests/unit/test_cython_parity.py` (windows, the
+stream iterator's queue fast path) and
+:file:`tests/unit/transport/test_conductor_parity.py` (the conductor's
+per-message fan-out, driven end to end through both implementations).
+
 :file:`tests/unit/test_cython_parity.py` covers both halves: it asserts the
 accelerators are loaded when they are required, and compares the two
 implementations where they can be driven directly.

faust/transport/_cython/conductor.pyx

@@ -69,10 +69,31 @@ cdef class ConductorHandler:
             full = []
             try:
                 for chan in channels:
-                    event, event_keyid = self._decode(event, chan, event_keyid)
+                    # Deserialize once and reuse the event for every channel
+                    # whose key/value types match, exactly as conductor.py
+                    # does.  `event`/`event_keyid` stay pinned to the first
+                    # channel; a channel with a different type pair gets its
+                    # own event without displacing the pinned one.
+                    #
+                    # This used to go through `_decode()`, which never worked:
+                    # `event_keyid` was only ever assigned from that helper's
+                    # return value, and the helper returned it *unchanged* on
+                    # the first pass, so it stayed None forever and the reuse
+                    # branch was dead -- every channel re-deserialized the
+                    # payload.  That in turn masked a second fault: had the
+                    # keyid ever been set, a mismatch fell off the end of
+                    # `_decode` returning a bare None, and unpacking it into
+                    # two names would have raised TypeError.
+                    keyid = (chan.key_type, chan.value_type)
                     if event is None:
                         event = await chan.decode(message, propagate=True)
-                    if not self._put(event, chan, full):
+                        event_keyid = keyid
+                        dest_event = event
+                    elif keyid == event_keyid:
+                        dest_event = event
+                    else:
+                        dest_event = await chan.decode(message, propagate=True)
+                    if not self._put(dest_event, chan, full):
                         continue
                     delivered.add(chan)
                 if full:
@@ -109,13 +130,6 @@ cdef class ConductorHandler:
     def on_pressure_drop(self) -> None:
         self.consumer_on_buffer_drop(self.tp)
 
-    cdef object _decode(self, object event, object channel, object event_keyid):
-        keyid = channel.key_type, channel.value_type
-        if event_keyid is None or event is None:
-            return None, event_keyid
-        if keyid == event_keyid:
-            return event, keyid
-
     cdef bint _put(self,
                    object event,
                    object channel,