From c2544c44c7b65db7501d6b5f2ed7df0ee47e4fd3 Mon Sep 17 00:00:00 2001 From: Media Date: Fri, 7 Aug 2026 17:13:54 -0500 Subject: [PATCH] fix: close Audio Router self-echo PUBACK/subscription-delivery race publish_router_mode registered its self-echo correlation marker before publishing but removed it in a finally block the instant the QoS 1 publish() call observed its PUBACK. A broker may legally deliver a Controller's own subscription copy of that same publish after the PUBACK, since MQTT does not order the two; when that happened, the Controller mistook its own late echo for live operator intent and relinquished automatic restoration ownership. Keep the marker claimable for a short, bounded mqtt.router_echo_grace_seconds window (default 2s) after the PUBACK instead of clearing it immediately. Genuine external/operator commands are unaffected: once the grace window elapses, a matching payload is still correctly treated as operator intent and ownership is relinquished as before. Verified with deterministic fake-clock unit tests plus a real Mosquitto broker integration test (tests/test_router_echo_mosquitto_integration.py, skipped automatically when Docker is unavailable). --- README.md | 8 +- config.yaml | 5 + controller/config.py | 6 + controller/mqtt_adapter.py | 77 +++++-- controller/service.py | 1 + tests/test_config_contracts.py | 10 + tests/test_mqtt_adapter.py | 91 ++++++-- .../test_router_echo_mosquitto_integration.py | 206 ++++++++++++++++++ 8 files changed, 367 insertions(+), 37 deletions(-) create mode 100644 tests/test_router_echo_mosquitto_integration.py diff --git a/README.md b/README.md index 5ffa728..ecb02ff 100644 --- a/README.md +++ b/README.md @@ -345,8 +345,12 @@ when all of these remain true: - No live, non-self mode command and no conflicting state change indicated an operator override. -Controller subscribes to both Router mode set and state topics. Its exact -in-flight echo is recognized; every other live mode-set publication is treated +Controller subscribes to both Router mode set and state topics. Its own +mode-set publication is correlated by exact payload and stays recognized as +a self-echo both while the publish is in flight and, because a broker may +legally deliver a subscription echo after the PUBACK that acknowledged the +same publish, for a short `mqtt.router_echo_grace_seconds` window afterward +(default 2 seconds). Every other live mode-set publication is treated conservatively as operator intent, even if it repeats `video`. A retained set message is ignored as stale intent. Controller does not publish Router state; it accepts a received live or retained state message as the Router's current diff --git a/config.yaml b/config.yaml index 6288778..e1e704f 100644 --- a/config.yaml +++ b/config.yaml @@ -10,6 +10,11 @@ mqtt: command_queue_size: 512 # QoS 1 is accepted only after the broker PUBACKs within this bound. publish_ack_timeout_seconds: 2 + # Audio Router self-echo correlation stays claimable for this long after + # a Controller-originated mode-set publish is PUBACK'd, so a subscription + # echo that is delivered after the PUBACK is still recognized as ours + # instead of being mistaken for operator intent. + router_echo_grace_seconds: 2 topics: player_command: youtube/player/cmd diff --git a/controller/config.py b/controller/config.py index 9ce1a7c..c03bbc7 100644 --- a/controller/config.py +++ b/controller/config.py @@ -33,6 +33,7 @@ class MqttConfig: reconnect_max_seconds: int = 30 command_queue_size: int = 512 publish_ack_timeout_seconds: float = 2.0 + router_echo_grace_seconds: float = 2.0 @dataclass(frozen=True) @@ -219,6 +220,11 @@ def load_config( "mqtt.publish_ack_timeout_seconds", minimum=0.1, ), + router_echo_grace_seconds=_number( + mqtt_raw.get("router_echo_grace_seconds", 2.0), + "mqtt.router_echo_grace_seconds", + minimum=0.0, + ), ) if mqtt.reconnect_max_seconds < mqtt.reconnect_min_seconds: raise ConfigError("mqtt.reconnect_max_seconds must be >= reconnect_min_seconds") diff --git a/controller/mqtt_adapter.py b/controller/mqtt_adapter.py index 8be0c6a..41bc8e5 100644 --- a/controller/mqtt_adapter.py +++ b/controller/mqtt_adapter.py @@ -6,6 +6,7 @@ import json import logging import queue import threading +import time import uuid from dataclasses import dataclass from typing import Any, Mapping @@ -58,6 +59,7 @@ class MqttAdapter: *, client: Any | None = None, events: "queue.Queue[MqttEvent] | None" = None, + monotonic=time.monotonic, ): self.config = config self.events: "queue.Queue[MqttEvent]" = events or queue.Queue( @@ -73,8 +75,16 @@ class MqttAdapter: self._started = False self._loop_started = False self._bootstrap_thread: threading.Thread | None = None + self._monotonic = monotonic self._router_echo_lock = threading.Lock() - self._pending_router_echoes: dict[str, bytes] = {} + # token -> (exact payload, grace deadline). The deadline is ``None`` + # while the correlated publish() call has not yet observed its + # PUBACK; such an entry never expires on its own, bounded instead by + # the publish ack timeout. Once PUBACK is observed, the entry keeps + # a short grace window so a subscription echo delivered just after + # the PUBACK is still recognized as ours rather than mistaken for + # operator intent. + self._pending_router_echoes: dict[str, tuple[bytes, float | None]] = {} self.last_error: str | None = None self._configure_client() @@ -466,36 +476,63 @@ class MqttAdapter: def publish_router_mode(self, mode: str) -> PublishReceipt: # A broker normally sends our publication back through the matching # subscription. Register the exact payload before publishing because - # Paho's network thread may receive that echo before publish() returns. - # The marker is removed as soon as the QoS 1 publish call completes; - # this intentionally prefers a false ownership loss over mistaking a - # later same-value operator command for a Controller echo. + # Paho's network thread may receive that echo before publish() + # returns. A PUBACK only proves broker receipt of our publish, not + # that the broker has already forwarded (or even scheduled) our own + # subscription copy of it; that copy can legally arrive after the + # PUBACK. So the marker is not removed the instant publish() + # completes: it stays claimable for a short bounded grace window + # afterward, and is removed immediately once actually claimed or + # once the grace window elapses. This closes the specific race where + # the Controller's own late echo would otherwise be mistaken for + # live operator intent and cause automatic restoration ownership to + # be relinquished. echo_token = uuid.uuid4().hex + payload = mode.encode("utf-8") with self._router_echo_lock: - self._pending_router_echoes[echo_token] = mode.encode("utf-8") - try: - return self._publish( - self.config.topics.router_mode_set, - mode, - qos=ROUTER_COMMAND_QOS, - retain=ROUTER_COMMAND_RETAIN, - require_connection=True, - invalidate_transport_on_uncertain=True, - ) - finally: - with self._router_echo_lock: - self._pending_router_echoes.pop(echo_token, None) + self._prune_router_echoes_locked() + self._pending_router_echoes[echo_token] = (payload, None) + receipt = self._publish( + self.config.topics.router_mode_set, + mode, + qos=ROUTER_COMMAND_QOS, + retain=ROUTER_COMMAND_RETAIN, + require_connection=True, + invalidate_transport_on_uncertain=True, + ) + grace_deadline = ( + self._monotonic() + self.config.mqtt.router_echo_grace_seconds + ) + with self._router_echo_lock: + if echo_token in self._pending_router_echoes: + self._pending_router_echoes[echo_token] = (payload, grace_deadline) + return receipt def _claim_router_echo(self, payload: bytes) -> bool: - """Claim one exact mode-set echo from an in-flight local publish.""" + """Claim one exact mode-set echo from an in-flight or recent local publish.""" with self._router_echo_lock: - for token, expected in tuple(self._pending_router_echoes.items()): + self._prune_router_echoes_locked() + for token, (expected, _deadline) in tuple( + self._pending_router_echoes.items() + ): if expected == payload: del self._pending_router_echoes[token] return True return False + def _prune_router_echoes_locked(self) -> None: + """Drop expired grace-window entries. Caller holds ``_router_echo_lock``.""" + + now = self._monotonic() + expired = [ + token + for token, (_, deadline) in self._pending_router_echoes.items() + if deadline is not None and deadline <= now + ] + for token in expired: + del self._pending_router_echoes[token] + def publish_legacy(self, topic: str, payload: str, *, retain: bool = True) -> PublishReceipt: return self._publish( topic, diff --git a/controller/service.py b/controller/service.py index eccf1fd..1eff7ed 100644 --- a/controller/service.py +++ b/controller/service.py @@ -81,6 +81,7 @@ class ControllerService: self.mqtt = mqtt_adapter or MqttAdapter( config, events=self.events, + monotonic=monotonic, ) self.downloads = download_manager self.playback = playback diff --git a/tests/test_config_contracts.py b/tests/test_config_contracts.py index fb9dcf1..fcaf2ba 100644 --- a/tests/test_config_contracts.py +++ b/tests/test_config_contracts.py @@ -81,6 +81,7 @@ class ConfigurationTests(unittest.TestCase): self.assertIsNone(config.router.fallback_restore_mode) self.assertEqual(config.storage.uncertain_review_after_seconds, 900.0) self.assertEqual(config.mqtt.publish_ack_timeout_seconds, 2.0) + self.assertEqual(config.mqtt.router_echo_grace_seconds, 2.0) def test_documented_environment_overrides_apply(self): env = { @@ -152,6 +153,15 @@ class ConfigurationTests(unittest.TestCase): with self.assertRaisesRegex(ConfigError, "must be finite"): load_config(path, environ={}) + def test_negative_router_echo_grace_is_rejected(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.yaml" + path.write_text( + "mqtt:\n router_echo_grace_seconds: -1\n", encoding="utf-8" + ) + with self.assertRaisesRegex(ConfigError, "router_echo_grace_seconds"): + load_config(path, environ={}) + def test_runtime_paths_must_be_absolute(self): cases = ( ("storage:\n media_base: relative/media\n", "storage.media_base"), diff --git a/tests/test_mqtt_adapter.py b/tests/test_mqtt_adapter.py index 2e8ea69..bfddecd 100644 --- a/tests/test_mqtt_adapter.py +++ b/tests/test_mqtt_adapter.py @@ -10,6 +10,14 @@ from controller.config import ControllerConfig, MqttConfig from controller.mqtt_adapter import MqttAdapter +class MutableClock: + def __init__(self, value=0.0): + self.value = value + + def __call__(self): + return self.value + + class FakeMessageInfo: def __init__(self, mid, published): self.rc = mqtt.MQTT_ERR_SUCCESS @@ -245,7 +253,7 @@ class MqttAdapterTests(unittest.TestCase): ("audio/player/mode/set", "video", 1, False), ) - def test_router_self_echo_is_marked_only_while_publish_is_in_flight(self): + def test_router_self_echo_is_marked_while_publish_is_in_flight(self): self.adapter.start() self.adapter._on_connect(self.client, None, None, 0) self.events.get_nowait() @@ -268,20 +276,73 @@ class MqttAdapterTests(unittest.TestCase): self.assertTrue(self.adapter.publish_router_mode("video").accepted) self.assertTrue(self.events.get_nowait().controller_echo) - self.client.publish_hook = None - self.adapter._on_message( - self.client, - None, - SimpleNamespace( - topic="audio/player/mode/set", - payload=b"video", - qos=1, - retain=False, - dup=False, - mid=32, - ), - ) - self.assertFalse(self.events.get_nowait().controller_echo) + def test_router_self_echo_delivered_after_puback_is_still_claimed_within_grace( + self, + ): + # Regression: a broker may legally deliver our own subscription echo + # after the PUBACK that already acknowledged the same publish. If the + # echo marker were cleared the instant publish() observes that + # PUBACK, this late echo would be mistaken for live operator intent + # and cause automatic Router restoration ownership to be relinquished. + clock = MutableClock(1000.0) + client = FakeClient() + events = queue.Queue() + adapter = MqttAdapter(self.config, client=client, events=events, monotonic=clock) + try: + adapter.start() + adapter._on_connect(client, None, None, 0) + events.get_nowait() + + self.assertTrue(adapter.publish_router_mode("video").accepted) + clock.value += self.config.mqtt.router_echo_grace_seconds - 0.5 + + adapter._on_message( + client, + None, + SimpleNamespace( + topic="audio/player/mode/set", + payload=b"video", + qos=1, + retain=False, + dup=False, + mid=32, + ), + ) + self.assertTrue(events.get_nowait().controller_echo) + finally: + adapter.stop() + + def test_router_self_echo_marker_expires_after_grace_window(self): + # A genuine external/operator command that happens to repeat the same + # value must still be recognized as operator intent once the bounded + # grace window has elapsed; the marker must not linger forever. + clock = MutableClock(1000.0) + client = FakeClient() + events = queue.Queue() + adapter = MqttAdapter(self.config, client=client, events=events, monotonic=clock) + try: + adapter.start() + adapter._on_connect(client, None, None, 0) + events.get_nowait() + + self.assertTrue(adapter.publish_router_mode("video").accepted) + clock.value += self.config.mqtt.router_echo_grace_seconds + 0.5 + + adapter._on_message( + client, + None, + SimpleNamespace( + topic="audio/player/mode/set", + payload=b"video", + qos=1, + retain=False, + dup=False, + mid=32, + ), + ) + self.assertFalse(events.get_nowait().controller_echo) + finally: + adapter.stop() def test_retained_router_set_never_consumes_controller_echo_marker(self): self.adapter.start() diff --git a/tests/test_router_echo_mosquitto_integration.py b/tests/test_router_echo_mosquitto_integration.py new file mode 100644 index 0000000..f504f20 --- /dev/null +++ b/tests/test_router_echo_mosquitto_integration.py @@ -0,0 +1,206 @@ +"""Real-broker smoke test for Audio Router self-echo correlation. + +The fake MQTT client used by ``tests/test_mqtt_adapter.py`` delivers a +publish's own echo synchronously, inside the ``publish()`` call itself, so it +cannot exercise the actual ordering risk: a real broker may legally deliver a +Controller-originated Router mode-set message back through the Controller's +own subscription either before or after the PUBACK that already acknowledged +the same publish. This test runs the real ``MqttAdapter`` against a real +Mosquitto broker in Docker to prove that ordering no longer matters. + +Skipped automatically when Docker (or a Mosquitto broker container) is not +available, so it never blocks the primary regression suite. +""" + +from __future__ import annotations + +import queue +import shutil +import socket +import subprocess +import tempfile +import time +import unittest +import uuid +from pathlib import Path + +import paho.mqtt.client as mqtt + +from controller.config import ControllerConfig, MqttConfig +from controller.mqtt_adapter import MqttAdapter + + +def _docker_available() -> bool: + return shutil.which("docker") is not None + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +@unittest.skipUnless( + _docker_available(), "docker is not available for a real Mosquitto broker" +) +class RouterSelfEchoMosquittoIntegrationTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.port = _free_port() + cls.container_name = f"controller-echo-test-{uuid.uuid4().hex[:10]}" + cls.tempdir = tempfile.TemporaryDirectory() + conf_path = Path(cls.tempdir.name) / "mosquitto.conf" + conf_path.write_text("listener 1883\nallow_anonymous true\n", encoding="utf-8") + try: + subprocess.run( + [ + "docker", + "run", + "-d", + "--rm", + "--name", + cls.container_name, + "-p", + f"127.0.0.1:{cls.port}:1883", + "-v", + f"{conf_path}:/mosquitto/config/mosquitto.conf:ro", + "eclipse-mosquitto", + ], + check=True, + capture_output=True, + text=True, + timeout=30, + ) + except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: + cls.tempdir.cleanup() + raise unittest.SkipTest( + f"could not start a real Mosquitto broker container: {exc}" + ) + try: + cls._wait_for_broker(cls.port) + except AssertionError as exc: + subprocess.run( + ["docker", "rm", "-f", cls.container_name], + capture_output=True, + timeout=15, + ) + cls.tempdir.cleanup() + raise unittest.SkipTest(str(exc)) + + @classmethod + def tearDownClass(cls) -> None: + subprocess.run( + ["docker", "rm", "-f", cls.container_name], + capture_output=True, + timeout=15, + ) + cls.tempdir.cleanup() + + @staticmethod + def _wait_for_broker(port: int, timeout: float = 15.0) -> None: + deadline = time.monotonic() + timeout + last_exc: OSError | None = None + while time.monotonic() < deadline: + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.5): + return + except OSError as exc: + last_exc = exc + time.sleep(0.2) + raise AssertionError(f"Mosquitto broker never became reachable: {last_exc}") + + def _config(self) -> ControllerConfig: + return ControllerConfig( + mqtt=MqttConfig( + host="127.0.0.1", + port=self.port, + client_id=f"controller-echo-{uuid.uuid4().hex[:10]}", + reconnect_min_seconds=1, + reconnect_max_seconds=1, + command_queue_size=32, + publish_ack_timeout_seconds=5.0, + router_echo_grace_seconds=2.0, + ) + ) + + def test_controller_recognizes_its_own_router_echo_over_a_real_broker(self) -> None: + config = self._config() + adapter = MqttAdapter(config) + adapter.start() + try: + self._wait_until( + lambda: adapter.connected, + "Controller MQTT adapter never connected to the real broker", + ) + receipt = adapter.publish_router_mode("video") + self.assertTrue(receipt.accepted, receipt.reason) + + event = self._next_message_event(adapter, timeout=5.0) + self.assertEqual(event.topic, config.topics.router_mode_set) + self.assertEqual(event.payload, b"video") + self.assertTrue( + event.controller_echo, + "Controller did not recognize its own Router mode-set echo " + "delivered by a real broker; the PUBACK-vs-subscription-echo " + "race was not closed", + ) + finally: + adapter.stop() + + def test_genuine_external_router_command_is_not_treated_as_self_echo(self) -> None: + config = self._config() + adapter = MqttAdapter(config) + adapter.start() + operator = mqtt.Client( + mqtt.CallbackAPIVersion.VERSION2, + client_id=f"operator-{uuid.uuid4().hex[:10]}", + ) + try: + self._wait_until( + lambda: adapter.connected, + "Controller MQTT adapter never connected to the real broker", + ) + operator.connect(config.mqtt.host, config.mqtt.port, keepalive=30) + operator.loop_start() + info = operator.publish( + config.topics.router_mode_set, "service", qos=1, retain=False + ) + info.wait_for_publish(timeout=5.0) + + event = self._next_message_event(adapter, timeout=5.0) + self.assertEqual(event.payload, b"service") + self.assertFalse( + event.controller_echo, + "A genuine external Router command was mistaken for a " + "Controller echo; automatic restoration ownership would be " + "incorrectly preserved instead of relinquished", + ) + finally: + operator.loop_stop() + operator.disconnect() + adapter.stop() + + @staticmethod + def _wait_until(predicate, message: str, timeout: float = 10.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.05) + raise AssertionError(message) + + @staticmethod + def _next_message_event(adapter: MqttAdapter, *, timeout: float): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + event = adapter.events.get(timeout=0.2) + except queue.Empty: + continue + if event.kind == "message": + return event + raise AssertionError("no MQTT message event was observed within the timeout") + + +if __name__ == "__main__": + unittest.main()