Media
c2544c44c7
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).
560 lines
21 KiB
Python
560 lines
21 KiB
Python
"""Resilient MQTT transport with explicit Player command guarantees."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import queue
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from typing import Any, Mapping
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
from .config import ControllerConfig
|
|
from .contracts import (
|
|
PLAYER_COMMAND_QOS,
|
|
PLAYER_COMMAND_RETAIN,
|
|
PLAYER_STATE_QOS,
|
|
PLAYER_STATUS_QOS,
|
|
PLAYER_TELEMETRY_QOS,
|
|
ROUTER_COMMAND_QOS,
|
|
ROUTER_COMMAND_RETAIN,
|
|
)
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MqttEvent:
|
|
kind: str
|
|
topic: str | None = None
|
|
payload: bytes = b""
|
|
qos: int = 0
|
|
retain: bool = False
|
|
duplicate: bool = False
|
|
message_id: int | None = None
|
|
connection_generation: int | None = None
|
|
reason: str | None = None
|
|
controller_echo: bool = False
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PublishReceipt:
|
|
accepted: bool
|
|
message_id: int | None = None
|
|
reason: str | None = None
|
|
uncertain: bool = False
|
|
|
|
|
|
class MqttAdapter:
|
|
"""Own Paho lifecycle while keeping all domain work off its callbacks."""
|
|
|
|
def __init__(
|
|
self,
|
|
config: ControllerConfig,
|
|
*,
|
|
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(
|
|
maxsize=config.mqtt.command_queue_size
|
|
)
|
|
self._injected_client = client is not None
|
|
self.client = client or self._new_client()
|
|
self._generation_lock = threading.Lock()
|
|
self._connection_generation = 0
|
|
self._connected = threading.Event()
|
|
self._stop_event = threading.Event()
|
|
self._lifecycle_lock = threading.Lock()
|
|
self._started = False
|
|
self._loop_started = False
|
|
self._bootstrap_thread: threading.Thread | None = None
|
|
self._monotonic = monotonic
|
|
self._router_echo_lock = threading.Lock()
|
|
# 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()
|
|
|
|
def _new_client(self):
|
|
return mqtt.Client(
|
|
mqtt.CallbackAPIVersion.VERSION2,
|
|
client_id=self.config.mqtt.client_id,
|
|
clean_session=True,
|
|
)
|
|
|
|
def _configure_client(self) -> None:
|
|
if self.config.mqtt.username is not None:
|
|
self.client.username_pw_set(
|
|
self.config.mqtt.username,
|
|
self.config.mqtt.password,
|
|
)
|
|
self.client.on_connect = self._on_connect
|
|
self.client.on_connect_fail = self._on_connect_fail
|
|
self.client.on_disconnect = self._on_disconnect
|
|
self.client.on_message = self._on_message
|
|
self.client.suppress_exceptions = True
|
|
self.client.manual_ack_set(True)
|
|
self.client.max_queued_messages_set(self.config.mqtt.command_queue_size)
|
|
self.client.reconnect_delay_set(
|
|
min_delay=self.config.mqtt.reconnect_min_seconds,
|
|
max_delay=self.config.mqtt.reconnect_max_seconds,
|
|
)
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
return self._connected.is_set()
|
|
|
|
@property
|
|
def connection_generation(self) -> int:
|
|
with self._generation_lock:
|
|
return self._connection_generation
|
|
|
|
@property
|
|
def started(self) -> bool:
|
|
with self._lifecycle_lock:
|
|
return self._started
|
|
|
|
def start(self) -> None:
|
|
"""Start without requiring the broker or network to be reachable."""
|
|
with self._lifecycle_lock:
|
|
if self._started:
|
|
return
|
|
self._started = True
|
|
self._stop_event.clear()
|
|
try:
|
|
self._begin_async_connection()
|
|
except (OSError, mqtt.WebsocketConnectionError, RuntimeError) as exc:
|
|
self.last_error = f"initial MQTT setup failed: {exc}"
|
|
LOG.warning("%s; retrying in background", self.last_error)
|
|
self._start_bootstrap_retry()
|
|
|
|
def _begin_async_connection(self) -> None:
|
|
with self._lifecycle_lock:
|
|
if not self._started or self._stop_event.is_set():
|
|
return
|
|
client = self.client
|
|
result = client.connect_async(
|
|
self.config.mqtt.host,
|
|
self.config.mqtt.port,
|
|
keepalive=self.config.mqtt.keepalive_seconds,
|
|
)
|
|
# Paho 2.x returns None after configuring an asynchronous connect;
|
|
# older/test clients may return MQTT_ERR_SUCCESS.
|
|
if result not in (None, mqtt.MQTT_ERR_SUCCESS):
|
|
raise RuntimeError(f"failed to configure MQTT connection: {result}")
|
|
with self._lifecycle_lock:
|
|
if (
|
|
not self._started
|
|
or self._stop_event.is_set()
|
|
or client is not self.client
|
|
):
|
|
return
|
|
if not self._loop_started:
|
|
result = client.loop_start()
|
|
if result != mqtt.MQTT_ERR_SUCCESS:
|
|
raise RuntimeError(f"failed to start MQTT network loop: {result}")
|
|
self._loop_started = True
|
|
|
|
def _start_bootstrap_retry(self) -> None:
|
|
with self._lifecycle_lock:
|
|
if self._bootstrap_thread is not None and self._bootstrap_thread.is_alive():
|
|
return
|
|
self._bootstrap_thread = threading.Thread(
|
|
target=self._bootstrap_retry,
|
|
name="mqtt-bootstrap-retry",
|
|
daemon=True,
|
|
)
|
|
self._bootstrap_thread.start()
|
|
|
|
def _bootstrap_retry(self) -> None:
|
|
delay = float(self.config.mqtt.reconnect_min_seconds)
|
|
maximum = float(self.config.mqtt.reconnect_max_seconds)
|
|
while not self._stop_event.wait(delay):
|
|
try:
|
|
self._begin_async_connection()
|
|
self.last_error = None
|
|
return
|
|
except (OSError, mqtt.WebsocketConnectionError) as exc:
|
|
self.last_error = f"MQTT setup retry failed: {exc}"
|
|
LOG.warning("%s", self.last_error)
|
|
delay = min(maximum, max(delay * 2, 1.0))
|
|
except RuntimeError as exc:
|
|
self.last_error = str(exc)
|
|
LOG.warning("%s", self.last_error)
|
|
delay = min(maximum, max(delay * 2, 1.0))
|
|
|
|
def stop(self) -> None:
|
|
# Signal callbacks/retry work before waiting for the lifecycle lock.
|
|
# A callback already inside subscribe/connect then observes shutdown
|
|
# and cannot publish a post-stop connected state.
|
|
self._stop_event.set()
|
|
with self._lifecycle_lock:
|
|
if not self._started:
|
|
self._connected.clear()
|
|
return
|
|
self._started = False
|
|
loop_started = self._loop_started
|
|
self._loop_started = False
|
|
self._connected.clear()
|
|
if loop_started:
|
|
try:
|
|
self.client.disconnect()
|
|
except Exception:
|
|
LOG.exception("MQTT disconnect failed")
|
|
self.client.loop_stop()
|
|
thread = self._bootstrap_thread
|
|
if thread is not None and thread is not threading.current_thread():
|
|
thread.join(timeout=1.0)
|
|
|
|
@staticmethod
|
|
def _reason_success(reason_code: Any) -> bool:
|
|
try:
|
|
return int(reason_code) == 0
|
|
except (TypeError, ValueError):
|
|
return reason_code == 0
|
|
|
|
def _on_connect(self, client, userdata, flags, reason_code, properties=None) -> None:
|
|
try:
|
|
with self._lifecycle_lock:
|
|
if (
|
|
client is not self.client
|
|
or not self._started
|
|
or self._stop_event.is_set()
|
|
):
|
|
return
|
|
if not self._reason_success(reason_code):
|
|
self.last_error = f"MQTT connect failed: {reason_code}"
|
|
LOG.warning("%s", self.last_error)
|
|
return
|
|
if self._stop_event.is_set():
|
|
return
|
|
with self._generation_lock:
|
|
self._connection_generation += 1
|
|
generation = self._connection_generation
|
|
self._connected.set()
|
|
self.last_error = None
|
|
subscriptions = (
|
|
(self.config.topics.player_status, PLAYER_STATUS_QOS),
|
|
(self.config.topics.player_telemetry, PLAYER_TELEMETRY_QOS),
|
|
(self.config.topics.player_state, PLAYER_STATE_QOS),
|
|
(self.config.topics.controller_command, 1),
|
|
(self.config.topics.router_mode_set, ROUTER_COMMAND_QOS),
|
|
(self.config.topics.router_mode_state, 1),
|
|
)
|
|
for topic, qos in subscriptions:
|
|
if self._stop_event.is_set():
|
|
self._connected.clear()
|
|
return
|
|
client.subscribe(topic, qos=qos)
|
|
if self._stop_event.is_set():
|
|
self._connected.clear()
|
|
return
|
|
if self._stop_event.is_set():
|
|
self._connected.clear()
|
|
return
|
|
self._enqueue(
|
|
MqttEvent(
|
|
kind="connected",
|
|
connection_generation=generation,
|
|
)
|
|
)
|
|
except BaseException:
|
|
LOG.exception("unexpected MQTT on_connect failure")
|
|
|
|
def _on_connect_fail(self, client, userdata) -> None:
|
|
self.last_error = (
|
|
f"MQTT connection to {self.config.mqtt.host}:{self.config.mqtt.port} failed"
|
|
)
|
|
LOG.warning("%s; Paho will retry", self.last_error)
|
|
|
|
def _on_disconnect(
|
|
self,
|
|
client,
|
|
userdata,
|
|
disconnect_flags,
|
|
reason_code,
|
|
properties=None,
|
|
) -> None:
|
|
if client is not self.client:
|
|
return
|
|
self._connected.clear()
|
|
try:
|
|
self._enqueue(
|
|
MqttEvent(
|
|
kind="disconnected",
|
|
connection_generation=self.connection_generation,
|
|
reason=str(reason_code),
|
|
)
|
|
)
|
|
except BaseException:
|
|
LOG.exception("unexpected MQTT on_disconnect failure")
|
|
|
|
def _on_message(self, client, userdata, message) -> None:
|
|
try:
|
|
# Capture client identity and generation atomically with session
|
|
# replacement. A callback that was already running when an
|
|
# uncertain command discards the transport must be able to leave a
|
|
# full queue so loop_stop() cannot deadlock waiting for it.
|
|
with self._lifecycle_lock:
|
|
if client is not self.client:
|
|
return
|
|
with self._generation_lock:
|
|
generation = self._connection_generation
|
|
topic = str(message.topic)
|
|
payload = bytes(message.payload)
|
|
retained = bool(getattr(message, "retain", False))
|
|
self._enqueue(
|
|
MqttEvent(
|
|
kind="message",
|
|
topic=topic,
|
|
payload=payload,
|
|
qos=int(getattr(message, "qos", 0)),
|
|
retain=retained,
|
|
duplicate=bool(getattr(message, "dup", False)),
|
|
message_id=getattr(message, "mid", None),
|
|
connection_generation=generation,
|
|
controller_echo=(
|
|
topic == self.config.topics.router_mode_set
|
|
and not retained
|
|
and self._claim_router_echo(payload)
|
|
),
|
|
)
|
|
)
|
|
except BaseException:
|
|
LOG.exception("unexpected MQTT on_message failure")
|
|
|
|
def _enqueue(self, event: MqttEvent) -> bool:
|
|
while not self._stop_event.is_set():
|
|
if (
|
|
event.connection_generation is not None
|
|
and event.connection_generation != self.connection_generation
|
|
):
|
|
return False
|
|
try:
|
|
self.events.put(event, timeout=0.2)
|
|
return True
|
|
except queue.Full:
|
|
continue
|
|
return False
|
|
|
|
def acknowledge(self, event: MqttEvent) -> bool:
|
|
if event.message_id is None or event.qos <= 0:
|
|
return True
|
|
if event.connection_generation != self.connection_generation:
|
|
LOG.warning(
|
|
"skipping stale MQTT ack mid=%s from generation=%s",
|
|
event.message_id,
|
|
event.connection_generation,
|
|
)
|
|
return False
|
|
try:
|
|
result = self.client.ack(event.message_id, event.qos)
|
|
except BaseException:
|
|
LOG.exception("MQTT ack failed mid=%s", event.message_id)
|
|
return False
|
|
return result == mqtt.MQTT_ERR_SUCCESS
|
|
|
|
def _publish(
|
|
self,
|
|
topic: str,
|
|
payload: str,
|
|
*,
|
|
qos: int,
|
|
retain: bool,
|
|
require_connection: bool,
|
|
invalidate_transport_on_uncertain: bool = False,
|
|
) -> PublishReceipt:
|
|
if require_connection and (not self.started or not self.connected):
|
|
return PublishReceipt(False, reason="MQTT is disconnected")
|
|
try:
|
|
info = self.client.publish(topic, payload, qos=qos, retain=retain)
|
|
except BaseException as exc:
|
|
LOG.exception("MQTT publish failed topic=%s", topic)
|
|
return PublishReceipt(False, reason=str(exc))
|
|
if info.rc != mqtt.MQTT_ERR_SUCCESS:
|
|
return PublishReceipt(False, message_id=getattr(info, "mid", None), reason=f"MQTT rc={info.rc}")
|
|
if qos > 0 and hasattr(info, "wait_for_publish"):
|
|
try:
|
|
info.wait_for_publish(
|
|
timeout=self.config.mqtt.publish_ack_timeout_seconds
|
|
)
|
|
acknowledged = bool(info.is_published())
|
|
except BaseException as exc:
|
|
acknowledged = False
|
|
detail = str(exc) or exc.__class__.__name__
|
|
else:
|
|
detail = (
|
|
f"no broker PUBACK within "
|
|
f"{self.config.mqtt.publish_ack_timeout_seconds:g} seconds"
|
|
)
|
|
if not acknowledged:
|
|
reason = f"MQTT QoS 1 publication is uncertain: {detail}"
|
|
if invalidate_transport_on_uncertain:
|
|
self._discard_unacknowledged_transport(reason)
|
|
return PublishReceipt(
|
|
False,
|
|
message_id=getattr(info, "mid", None),
|
|
reason=reason,
|
|
uncertain=True,
|
|
)
|
|
return PublishReceipt(True, message_id=getattr(info, "mid", None))
|
|
|
|
def _discard_unacknowledged_transport(self, reason: str) -> None:
|
|
"""Drop the Paho session so stale QoS 1 commands cannot replay."""
|
|
|
|
if self._injected_client:
|
|
# Test transports do not own a recreatable Paho session. Production
|
|
# always follows the replacement path below.
|
|
with self._generation_lock:
|
|
self._connection_generation += 1
|
|
self._connected.clear()
|
|
return
|
|
with self._lifecycle_lock:
|
|
old_client = self.client
|
|
old_loop_started = self._loop_started
|
|
self._loop_started = False
|
|
should_restart = self._started and not self._stop_event.is_set()
|
|
self._connected.clear()
|
|
with self._generation_lock:
|
|
previous_generation = self._connection_generation
|
|
self._connection_generation += 1
|
|
self.client = self._new_client()
|
|
self._configure_client()
|
|
try:
|
|
old_client.disconnect()
|
|
except BaseException:
|
|
LOG.exception("failed to disconnect uncertain MQTT transport")
|
|
if old_loop_started:
|
|
try:
|
|
old_client.loop_stop()
|
|
except BaseException:
|
|
LOG.exception("failed to stop uncertain MQTT transport")
|
|
self.last_error = reason
|
|
try:
|
|
self.events.put_nowait(
|
|
MqttEvent(
|
|
kind="disconnected",
|
|
connection_generation=previous_generation,
|
|
reason=reason,
|
|
)
|
|
)
|
|
except queue.Full:
|
|
# This method can run on the sole coordinator thread; blocking on
|
|
# its own queue would deadlock. The uncertain PublishReceipt still
|
|
# drives the operation-specific conservative transition.
|
|
LOG.error("MQTT event queue was full while discarding uncertain transport")
|
|
if should_restart:
|
|
self._start_bootstrap_retry()
|
|
|
|
@staticmethod
|
|
def _json_payload(payload: Mapping[str, Any]) -> str:
|
|
return json.dumps(payload, allow_nan=False, separators=(",", ":"))
|
|
|
|
def publish_player_command(self, payload: Mapping[str, Any]) -> PublishReceipt:
|
|
return self._publish(
|
|
self.config.topics.player_command,
|
|
self._json_payload(payload),
|
|
qos=PLAYER_COMMAND_QOS,
|
|
retain=PLAYER_COMMAND_RETAIN,
|
|
require_connection=True,
|
|
invalidate_transport_on_uncertain=True,
|
|
)
|
|
|
|
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. 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._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 or recent local publish."""
|
|
|
|
with self._router_echo_lock:
|
|
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,
|
|
payload,
|
|
qos=0,
|
|
retain=retain,
|
|
require_connection=True,
|
|
)
|
|
|
|
def publish_json(
|
|
self,
|
|
topic: str,
|
|
payload: Mapping[str, Any],
|
|
*,
|
|
qos: int = 0,
|
|
retain: bool = False,
|
|
) -> PublishReceipt:
|
|
return self._publish(
|
|
topic,
|
|
self._json_payload(payload),
|
|
qos=qos,
|
|
retain=retain,
|
|
require_connection=True,
|
|
)
|