From f5286a0e67d0b40270c91be8383b80ff24516ca2 Mon Sep 17 00:00:00 2001 From: Godopu Date: Sun, 6 Sep 2026 16:15:40 +0900 Subject: [PATCH] fix(issue-5): scope the persistent publisher's CONNACK result to its own connection Found while self-reviewing phase 1. PersistentPublisher held the connect handshake result on the instance, so a refused attempt left its reason code behind for the next one. On retry the wait loop saw a non-None value straight away and never waited for the new connection's CONNACK, which is wrong in both directions: * a healthy second connection was rejected using the first attempt's rc, and * once a late callback had written a success rc, a subsequent connection that never acknowledged could be accepted on that stale value, letting a publish proceed on a socket that was not ready. Both need an asynchronous CONNACK to reproduce, which is exactly how paho delivers it; the earlier fake called the callback inline and hid it. The handshake state is now a local the callback closes over, so it can only ever describe the connection that produced it, and a client that fails the handshake is torn down instead of leaked. No production path is affected today: the CLI and shell helpers publish once per process and use OneShotPublisher. The defect mattered because the phase 2a engine is meant to build on this prototype, which is where a transient refusal would have become a stuck publisher. Also tightens PV-1 to check the bound tree against pytest's real basetemp rather than sniffing for "pytest-" in the path. Verified: the three new regressions fail against the previous implementation (2 of 3, the third covers the healthy path) and pass against this one. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd --- .agents/skills/lib_py/delegation/publisher.py | 43 +++--- tests/test_issue5_phase1_delegation.py | 128 +++++++++++++++++- 2 files changed, 151 insertions(+), 20 deletions(-) diff --git a/.agents/skills/lib_py/delegation/publisher.py b/.agents/skills/lib_py/delegation/publisher.py index ece8fc5..e07e405 100644 --- a/.agents/skills/lib_py/delegation/publisher.py +++ b/.agents/skills/lib_py/delegation/publisher.py @@ -100,7 +100,6 @@ class PersistentPublisher(_BasePublisher): self.connect_count = 0 self._client = None self._config = None - self._connected = {"rc": None} def __enter__(self): return self.open() @@ -124,8 +123,14 @@ class PersistentPublisher(_BasePublisher): client = self.client_factory("publisher", config) self.connect_count += 1 + # The handshake result belongs to THIS client, so it lives in a local the + # callback closes over. Holding it on self would let a refused attempt's + # reason code be read by the next attempt: the retry would either reject a + # healthy connection or, worse, accept one whose CONNACK never arrived. + connected = {"rc": None} + def on_connect(_c, _u, _flags, reason_code, _props): - self._connected["rc"] = reason_code + connected["rc"] = reason_code client.on_connect = on_connect client.connect(config.host, config.port, config.keepalive) @@ -133,19 +138,29 @@ class PersistentPublisher(_BasePublisher): reconnect_delay_set = getattr(client, "reconnect_delay_set", None) if reconnect_delay_set is not None: reconnect_delay_set(min_delay=1, max_delay=16) - deadline = time.monotonic() + CONNECT_ACK_TIMEOUT - while self._connected["rc"] is None and time.monotonic() < deadline: - time.sleep(0.05) - if self._connected["rc"] is None: - client.loop_stop() - raise TimeoutError("no CONNACK from broker") - if mqtt_common.reason_code_value(self._connected["rc"]) != 0: - client.loop_stop() - raise ConnectionError("broker refused connection: rc={}".format(self._connected["rc"])) + try: + deadline = time.monotonic() + CONNECT_ACK_TIMEOUT + while connected["rc"] is None and time.monotonic() < deadline: + time.sleep(0.05) + if connected["rc"] is None: + raise TimeoutError("no CONNACK from broker") + if mqtt_common.reason_code_value(connected["rc"]) != 0: + raise ConnectionError("broker refused connection: rc={}".format(connected["rc"])) + except Exception: + self._discard(client) # never leak the half-open client on the way out + raise self._client = client self._config = config return client + @staticmethod + def _discard(client): + try: + client.loop_stop() + client.disconnect() + except Exception: # pragma: no cover - best effort + pass + def publish(self, prepared): client = self._ensure_client(prepared.config) info = client.publish(prepared.topic, payload=prepared.body, @@ -159,11 +174,7 @@ class PersistentPublisher(_BasePublisher): if self._client is None: return try: - self._client.loop_stop() - self._client.disconnect() - except Exception: # pragma: no cover - best effort - pass + self._discard(self._client) finally: self._client = None self._config = None - self._connected = {"rc": None} diff --git a/tests/test_issue5_phase1_delegation.py b/tests/test_issue5_phase1_delegation.py index 2cda746..95c60e8 100644 --- a/tests/test_issue5_phase1_delegation.py +++ b/tests/test_issue5_phase1_delegation.py @@ -415,7 +415,7 @@ def test_pv5_all_four_shims_share_one_bootstrap_block(): assert "sys.path.append(_SKILLS)" in base, "standalone fallback must remain" -def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox): +def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox, tmp_path_factory): """PV-1: each alias IS the package module, and all four share one tree.""" import lib_py.delegation @@ -432,9 +432,19 @@ def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox): assert len(paths) == 1, "delegation modules span multiple trees: {}".format(paths) bound = Path(paths.pop()) real = PKG_DIR.resolve() - basetemp_ok = "pytest-" in str(bound) - assert bound == real or basetemp_ok, ( - "delegation bound to an unexpected tree: {}".format(bound) + # Provenance, not just identity: the bound tree is either the real package or a + # copy under pytest's own basetemp. Checked against the actual basetemp rather + # than sniffing for "pytest-" in the path, so an unrelated directory that merely + # looks temporary cannot satisfy it. + basetemp = Path(tmp_path_factory.getbasetemp()).resolve() + try: + bound.relative_to(basetemp) + under_basetemp = True + except ValueError: + under_basetemp = False + assert bound == real or under_basetemp, ( + "delegation bound to an unexpected tree: {} (real={}, basetemp={})".format( + bound, real, basetemp) ) # the parent package must never be dragged along with it assert Path(lib_py.__file__).resolve().parent == (SKILLS / "lib_py").resolve() @@ -880,3 +890,113 @@ def test_bench_broker_transport_requires_a_job(): with pytest.raises(ValueError): bench.run_benchmark(n=1, transport="broker") + + +# ============================================================================== +# 7. Self-review regressions — the connect handshake is per-connection +# +# Found reviewing phase 1: PersistentPublisher held the CONNACK result on the +# instance, so a refused attempt left its reason code behind for the next one. +# The retry then either rejected a healthy connection or, worse, accepted one +# whose CONNACK had never arrived. Both need an asynchronous callback to +# reproduce, which is how real paho delivers it. +# ============================================================================== + +class _AsyncAckClient(object): + """paho-shaped stub: connect() returns at once, CONNACK lands on a timer.""" + + def __init__(self, rc_sequence, calls, ack_delay=0.05): + self._rc_sequence = rc_sequence + self._calls = calls + self._ack_delay = ack_delay + self.on_connect = None + + def connect(self, host, port, keepalive=60): + import threading + idx = self._calls["n"] + self._calls["n"] += 1 + rc = self._rc_sequence[min(idx, len(self._rc_sequence) - 1)] + if rc is None: + return # models a broker that never acknowledges + timer = threading.Timer(self._ack_delay, + lambda: self.on_connect(self, None, None, rc, None)) + timer.daemon = True + timer.start() + + def loop_start(self): + return None + + def loop_stop(self): + self._calls["loop_stop"] = self._calls.get("loop_stop", 0) + 1 + + def reconnect_delay_set(self, **_kw): + return None + + def publish(self, *_a, **_k): + from lib_py.delegation.bench import FakeInfo + self._calls["publish"] = self._calls.get("publish", 0) + 1 + return FakeInfo() + + def disconnect(self): + self._calls["disconnect"] = self._calls.get("disconnect", 0) + 1 + + +def _async_factory(rc_sequence, calls): + def factory(_role, _config): + return _AsyncAckClient(rc_sequence, calls) + return factory + + +def test_persistent_publisher_recovers_on_the_attempt_after_a_refusal(): + """A refused connection must not decide the next attempt's outcome.""" + from lib_py.delegation.publisher import PersistentPublisher + + calls = {"n": 0} + publisher = PersistentPublisher(client_factory=_async_factory([5, 0], calls)) + try: + ok, err = publisher.publish_with_retry(_prepared(), attempts=3) + finally: + publisher.close() + assert ok is True, err + assert calls["n"] == 2, ( + "attempt 2 had a healthy connection and should have been the last one; " + "got {} connect() calls".format(calls["n"]) + ) + assert calls.get("disconnect", 0) >= 1, "the refused client must be torn down, not leaked" + + +def test_persistent_publisher_never_accepts_a_connection_that_did_not_ack(): + """A silent broker must time out even when an earlier attempt acked.""" + from lib_py.delegation.publisher import PersistentPublisher + import lib_py.delegation.publisher as publisher_mod + + calls = {"n": 0} + # first connection is refused (leaves a reason code behind), second never acks + publisher = PersistentPublisher(client_factory=_async_factory([5, None], calls)) + original = publisher_mod.CONNECT_ACK_TIMEOUT + publisher_mod.CONNECT_ACK_TIMEOUT = 0.3 # keep the test quick + try: + ok, err = publisher.publish_with_retry(_prepared(), attempts=2) + finally: + publisher_mod.CONNECT_ACK_TIMEOUT = original + publisher.close() + assert ok is False + assert "no CONNACK" in err, ( + "the silent second connection must time out on its own evidence, not " + "inherit the first attempt's reason code; got: {}".format(err) + ) + + +def test_persistent_publisher_reuses_a_healthy_connection_across_events(): + """The successful path still connects exactly once.""" + from lib_py.delegation.publisher import PersistentPublisher + + calls = {"n": 0} + publisher = PersistentPublisher(client_factory=_async_factory([0], calls)) + try: + for _ in range(4): + assert publisher.publish_with_retry(_prepared(), attempts=1) == (True, None) + finally: + publisher.close() + assert calls["n"] == 1, "a healthy connection must be reused" + assert calls["publish"] == 4