test(issue-26): C0 reads broker state only after the matching ack; the local broker acks a PUBLISH after routing it

Review (fa45b0e6): paho's connect() does not wait for the CONNACK, so C0
could read protocol_levels before the broker had handled the second
CONNECT, and the `before` slice could take in a previous test's connection.

- _Client waits for CONNACK (and UNSUBACK, instead of sleeping 0.3 s);
  C0 checks the levels of its own two client ids (stats() levels_by_client).
- The local broker now routes and stores a PUBLISH before its PUBACK, so a
  publisher holding the ack may rely on the retained state it asked for
  (late subscriber replay, retained delete).
- test_c0_holds_when_the_broker_is_slow runs C0 with CONNECT/UNSUBSCRIBE
  delayed and PUBLISH routed late; the old C0 fails it ([] == [4, 4]), as do
  an ack-before-route broker and a fixed-sleep unsubscribe.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/_local_mqtt.py

@@ -125,6 +125,7 @@ class LocalBroker:
     def stats(self) -> dict:
         with self._lock:
             return {"connections": self.connections, "protocol_levels": list(self.protocol_levels),
+                    "levels_by_client": dict(zip(self.client_ids, self.protocol_levels)),
                     "packets": dict(self.packets), "open": len(self._conns)}
 
     # -- network -------------------------------------------------------------
@@ -188,11 +189,15 @@ class LocalBroker:
             topic, i = _str(body, 0)
             if qos == 2:              # the repo publishes QoS 0/1 only
                 return False
+            pid = b""
             if qos == 1:
                 pid = body[i:i + 2]
                 i += 2
-                conn.send(_packet(PUBACK << 4, pid))
+            # Route (and store a retained message) BEFORE the PUBACK: a publisher
+            # that has its ack may rely on the broker state it asked for.
             self._route(topic.decode("utf-8"), body[i:], qos, retain)
+            if qos == 1:
+                conn.send(_packet(PUBACK << 4, pid))
             return True
         if ptype == SUBSCRIBE:
             pid, i, granted, new = body[:2], 2, bytearray(), []

tests/test_local_mqtt.py

@@ -51,22 +51,37 @@ def test_c3_the_product_config_resolves_to_the_local_broker():
 # ---------------------------------------------------------------------------
 
 class _Client:
-    """A real mqtt_common.make_client client, connected and looping."""
+    """A real mqtt_common.make_client client, connected and looping.
+
+    Every step that changes broker state waits for the broker's ack (CONNACK,
+    SUBACK, UNSUBACK, PUBACK), and the broker records/applies the change
+    before it acks - so a test may read broker state right after a step.
+    paho's connect() only sends CONNECT; it does not wait for the CONNACK."""
 
     def __init__(self, role):
         self.msgs = queue.Queue()
         self.subacks = queue.Queue()
+        self.unsubacks = queue.Queue()
+        self.connacks = queue.Queue()
         self.c = mqtt_common.make_client(role, mqtt_common.broker_config_from_env())
+        self.client_id = self.c._client_id.decode()
+        self.c.on_connect = lambda _c, _u, _f, rc, _p: self.connacks.put(mqtt_common.reason_code_value(rc))
         self.c.on_message = lambda _c, _u, m: self.msgs.put((m.topic, m.payload, m.qos, m.retain))
         self.c.on_subscribe = lambda _c, _u, mid, rcs, _p: self.subacks.put([mqtt_common.reason_code_value(r) for r in rcs])
+        self.c.on_unsubscribe = lambda *_a: self.unsubacks.put(True)
         cfg = mqtt_common.broker_config_from_env()
         self.c.connect(cfg.host, cfg.port, cfg.keepalive)
         self.c.loop_start()
+        assert self.connacks.get(timeout=5) == 0, "CONNACK refused"
 
     def sub(self, filt, qos=1):
         self.c.subscribe(filt, qos=qos)
         return self.subacks.get(timeout=5)
 
+    def unsub(self, filt):
+        self.c.unsubscribe(filt)
+        return self.unsubacks.get(timeout=5)
+
     def get(self, timeout=5):
         return self.msgs.get(timeout=timeout)
 
@@ -82,21 +97,19 @@ class _Client:
         self.c.disconnect()
 
 
-def test_c0_make_client_speaks_3_1_1_wildcards_qos1_and_retained():
-    broker = _local_mqtt.current_broker()
-    before = broker.stats()["connections"]
+def _c0_conformance(broker):
     root = "mam/t26-" + uuid.uuid4().hex[:8]
     a, b = _Client("c0-pub"), _Client("c0-sub")
     try:
-        assert broker.stats()["protocol_levels"][before:] == [4, 4], "paho's default MQTT 3.1.1"
+        levels = broker.stats()["levels_by_client"]
+        assert (levels.get(a.client_id), levels.get(b.client_id)) == (4, 4), "paho's default MQTT 3.1.1"
         assert b.sub(f"{root}/+/events", qos=1) == [1]
         a.c.publish(f"{root}/j1/events", b"one", qos=1).wait_for_publish(5)
         assert b.get() == (f"{root}/j1/events", b"one", 1, False)
         assert b.sub(f"{root}/#", qos=0) == [0]
         a.c.publish(f"{root}/j1/deep/x", b"two", qos=1).wait_for_publish(5)
         assert b.get() == (f"{root}/j1/deep/x", b"two", 0, False), "min(pub qos, sub qos)"
-        b.c.unsubscribe(f"{root}/#")
-        time.sleep(0.3)
+        assert b.unsub(f"{root}/#")
         a.c.publish(f"{root}/j1/deep/y", b"three", qos=1).wait_for_publish(5)
         assert b.none(), "unsubscribed"
         # retained: replayed to a late subscriber, deleted by an empty payload
@@ -120,6 +133,33 @@ def test_c0_make_client_speaks_3_1_1_wildcards_qos1_and_retained():
         b.close()
 
 
+def test_c0_make_client_speaks_3_1_1_wildcards_qos1_and_retained():
+    _c0_conformance(_local_mqtt.current_broker())
+
+
+def test_c0_holds_when_the_broker_is_slow(monkeypatch):
+    """The C0 races, made deterministic. The broker takes its time on CONNECT
+    and UNSUBSCRIBE, and routes each PUBLISH later still (longer than a new
+    client takes to connect and subscribe). A test that reads broker state
+    before the matching ack, or sleeps a fixed time, fails here; so does a
+    broker that acks a PUBLISH before routing it."""
+    broker = _local_mqtt.current_broker()
+    handle, route = broker._handle, broker._route
+    delay = {_local_mqtt.CONNECT: 0.3, _local_mqtt.UNSUBSCRIBE: 1.5}
+
+    def slow_handle(conn, ptype, flags, body):
+        time.sleep(delay.get(ptype, 0))
+        return handle(conn, ptype, flags, body)
+
+    def slow_route(*args):
+        time.sleep(0.8)
+        return route(*args)
+
+    monkeypatch.setattr(broker, "_handle", slow_handle)
+    monkeypatch.setattr(broker, "_route", slow_route)
+    _c0_conformance(broker)
+
+
 def _env(tmp_path: Path) -> dict:
     env = dict(os.environ)
     env.update(local_broker_env())