fix(printers): recover MQTT sessions that stopped reconnecting (#2732)

The reporter's printer lost its session to a keep-alive timeout at 02:19
    and did not come back until 11:24 — nine hours offline, with the web UI
    open throughout.

    check_staleness() was never going to catch it. Its first line is
    `if self.state.connected and self.is_stale()`, so it only ever handles the
    half-broken session that is still connected but has gone quiet. This
    client had connected=False from 02:19:42 (the offline notification fired a
    minute later), so every call returned immediately, and paho's own retry was
    the only thing left watching. When that stopped making progress nothing
    noticed.

    Adds a sweep every 60s that rebuilds a client when all four hold: it is
    disconnected, it had a working session before, it has been silent for five
    minutes, and its MQTT port still answers. The port check is what keeps this
    from becoming a nuisance — a switched-off printer is left to paho, so a
    farm powering down overnight causes no client churn and no log spam. The
    five-minute grace sits well past the 60s stale timeout and the 30s max
    reconnect backoff, so a session recovering on its own is never interrupted.

    The rebuild goes through force_reconnect_stale_session from async context,
    which takes the hard-reset path: fresh client_id and paho's QoS 1 queue
    dropped, so a project_file left unacked on the dead session cannot replay
    into the new one and trip 0500_4003 (#1136). Rate-limited per printer,
    cooldown cleared when the printer returns, and the sweep continues past a
    client that throws rather than abandoning the rest of the farm. The log
    line names how long the printer was gone and the last connect error, so a
    session that dies repeatedly leaves a trail.

    check_port gains a public alias in printer_diagnostic rather than having
    the watchdog reach for the private name.

    Also corrects the Developer Mode path added in the previous commit: the
    wiki documents it under Settings > Network, not Settings > General. The
    menu path is dropped from the translated string entirely, since it varies
    by model and firmware and the wiki carries the detail.
This commit is contained in:
maziggy
2026-08-02 09:49:52 +02:00
19 changed files with 319 additions and 27 deletions
+128
View File
@@ -6661,6 +6661,130 @@ def stop_spoolbuddy_watchdog():
logging.getLogger(__name__).info("SpoolBuddy watchdog stopped")
# Dead-MQTT-session recovery
#
# check_staleness() covers the "connected but silent" half-broken session. It
# does nothing once ``state.connected`` is False, and paho's own auto-reconnect
# is the only thing left watching at that point. When paho stops making
# progress there is no backstop at all: the #2732 bundle has a P1S drop on a
# keep-alive timeout at 02:19 and not reconnect until 11:24 — nine hours
# offline with the UI open the whole time, recovered only when something
# happened to nudge it.
#
# This loop is that backstop. It only touches printers that had a working
# session and lost it, and only when the MQTT port still answers — a printer
# that is simply switched off is left to paho, since rebuilding a client
# against an unreachable host achieves nothing and would fill the log every
# night.
_connection_watchdog_task: asyncio.Task | None = None
CONNECTION_WATCHDOG_INTERVAL = 60
# How long a printer must have been silent before we stop trusting paho.
# Comfortably above STALE_TIMEOUT (60 s) and the max reconnect backoff (30 s),
# so a session that is recovering on its own is never interrupted.
CONNECTION_WATCHDOG_OFFLINE_GRACE = 300
# Per-printer floor between rebuild attempts.
CONNECTION_WATCHDOG_RETRY_INTERVAL = 300
_connection_watchdog_last_attempt: dict[int, float] = {}
async def _recover_dead_printer_sessions() -> int:
"""Rebuild MQTT clients that have been offline too long to still be trying.
Returns the number of printers a rebuild was attempted for (for tests and
for the caller's logging). Never raises: one unreachable printer must not
stop the sweep for the rest of the farm.
"""
logger = logging.getLogger(__name__)
from backend.app.services.printer_diagnostic import PORT_MQTT, check_port
now = time.monotonic()
recovered = 0
for printer_id, client in list(printer_manager._clients.items()):
try:
if client.state.connected:
_connection_watchdog_last_attempt.pop(printer_id, None)
continue
# Time since the last inbound message is the age of the last known
# good session — no extra bookkeeping needed, and it is the same
# clock is_stale() reads. 0 means this client has never had one:
# that is the initial-connect path, where paho retrying is the
# correct and only behaviour, so leave it be.
last_msg = client._last_message_time
if not last_msg:
continue
offline_for = time.time() - last_msg
if offline_for < CONNECTION_WATCHDOG_OFFLINE_GRACE:
continue
last_attempt = _connection_watchdog_last_attempt.get(printer_id)
if last_attempt is not None and now - last_attempt < CONNECTION_WATCHDOG_RETRY_INTERVAL:
continue
if not await check_port(client.ip_address, PORT_MQTT):
# Switched off, unplugged, or off the network. Paho's retry is
# the right handler; say so at debug level and move on.
logger.debug(
"[#2732] Printer %s offline for %.0fs and its MQTT port is not answering "
"— leaving the reconnect to paho",
printer_id,
offline_for,
)
_connection_watchdog_last_attempt[printer_id] = now
continue
_connection_watchdog_last_attempt[printer_id] = now
recovered += 1
logger.warning(
"[#2732] Printer %s has been offline for %.0fs but answers on MQTT port %d — "
"rebuilding the client with a fresh session (last connect error: %s)",
printer_id,
offline_for,
PORT_MQTT,
client.last_connect_error or "none recorded",
)
# Async context, so this takes the hard-reset path: fresh client_id,
# paho's QoS 1 queue dropped. That matters — a project_file left
# unacked on the dead session would otherwise replay into the new
# one and trip 0500_4003 on the printer (#1136).
client.force_reconnect_stale_session(f"offline for {offline_for:.0f}s, port still answering")
except Exception as e:
logger.warning("[#2732] Connection watchdog failed for printer %s: %s", printer_id, e)
return recovered
async def _connection_watchdog_loop():
logger = logging.getLogger(__name__)
# Let the initial connects settle before judging anyone offline.
await asyncio.sleep(CONNECTION_WATCHDOG_OFFLINE_GRACE)
while True:
try:
await _recover_dead_printer_sessions()
except asyncio.CancelledError:
break
except Exception as e:
logger.warning("Connection watchdog sweep failed: %s", e)
await asyncio.sleep(CONNECTION_WATCHDOG_INTERVAL)
def start_connection_watchdog():
global _connection_watchdog_task
if _connection_watchdog_task is None:
_connection_watchdog_task = asyncio.create_task(_connection_watchdog_loop())
logging.getLogger(__name__).info("Printer connection watchdog started")
def stop_connection_watchdog():
global _connection_watchdog_task
if _connection_watchdog_task:
_connection_watchdog_task.cancel()
_connection_watchdog_task = None
_connection_watchdog_last_attempt.clear()
logging.getLogger(__name__).info("Printer connection watchdog stopped")
# Camera stream orphan cleanup
_camera_cleanup_task: asyncio.Task | None = None
CAMERA_CLEANUP_INTERVAL = 60
@@ -7218,6 +7342,9 @@ async def lifespan(app: FastAPI):
# Start camera stream orphan cleanup
start_camera_cleanup()
# Start the backstop for MQTT sessions paho has stopped recovering (#2732)
start_connection_watchdog()
# One-shot sweep for timelapse session directories orphaned by a crash
# or restart that happened mid-print (in-memory session tracking can't
# survive that, and nothing else reaps the leftover frames/output file)
@@ -7270,6 +7397,7 @@ async def lifespan(app: FastAPI):
stop_runtime_tracking()
stop_spoolbuddy_watchdog()
stop_camera_cleanup()
stop_connection_watchdog()
from backend.app.services.loop_watchdog import stop_loop_watchdog
stop_loop_watchdog()
@@ -57,6 +57,12 @@ async def _check_port(ip: str, port: int, timeout: float = _PORT_PROBE_TIMEOUT)
return False
# Public alias. The connection watchdog probes the MQTT port before rebuilding a
# client, so it can tell "the printer is switched off" (leave it alone, paho will
# keep retrying) from "the printer is answering but our session is dead" (#2732).
check_port = _check_port
def _auth_reason_params(reason: str | None) -> dict:
"""Map a client's CONNACK-refusal slug onto the check's `params.reason`.
@@ -0,0 +1,157 @@
"""Tests for the dead-MQTT-session watchdog (#2732).
``check_staleness()`` guards the "connected but silent" session and returns
immediately once ``state.connected`` is False — from there, paho's own
auto-reconnect is the only thing still watching. The #2732 bundle shows what
happens when that stops making progress: a P1S dropped on a keep-alive timeout
at 02:19 and did not reconnect until 11:24, nine hours offline with the UI open
throughout.
This watchdog is the backstop. The rules it has to keep are narrow on purpose —
it must not interfere with a session that is recovering on its own, and it must
not churn clients for printers that are simply switched off.
"""
import time
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.app.main import (
CONNECTION_WATCHDOG_OFFLINE_GRACE,
CONNECTION_WATCHDOG_RETRY_INTERVAL,
_connection_watchdog_last_attempt,
_recover_dead_printer_sessions,
)
def _client(*, connected: bool, last_message_age: float | None, ip: str = "192.168.1.100"):
"""Stand-in for BambuMQTTClient with only the fields the watchdog reads."""
return SimpleNamespace(
state=SimpleNamespace(connected=connected),
_last_message_time=0.0 if last_message_age is None else time.time() - last_message_age,
ip_address=ip,
last_connect_error=None,
force_reconnect_stale_session=MagicMock(),
)
async def _sweep(clients: dict, *, port_open: bool = True):
with (
patch("backend.app.main.printer_manager._clients", clients),
patch("backend.app.services.printer_diagnostic.check_port", AsyncMock(return_value=port_open)),
):
return await _recover_dead_printer_sessions()
@pytest.fixture(autouse=True)
def _clear_cooldowns():
_connection_watchdog_last_attempt.clear()
yield
_connection_watchdog_last_attempt.clear()
class TestRebuildsDeadSessions:
@pytest.mark.asyncio
async def test_rebuilds_a_long_dead_session(self):
"""The #2732 case: offline for hours, printer answering the whole time."""
client = _client(connected=False, last_message_age=32718.0)
assert await _sweep({1: client}) == 1
client.force_reconnect_stale_session.assert_called_once()
@pytest.mark.asyncio
async def test_reconnect_reason_names_the_duration(self):
client = _client(connected=False, last_message_age=32718.0)
await _sweep({1: client})
assert "32718" in client.force_reconnect_stale_session.call_args.args[0]
@pytest.mark.asyncio
async def test_sweeps_every_printer_in_the_farm(self):
clients = {i: _client(connected=False, last_message_age=9999.0) for i in range(1, 4)}
assert await _sweep(clients) == 3
class TestLeavesHealthyAndRecoveringSessionsAlone:
@pytest.mark.asyncio
async def test_connected_printer_is_untouched(self):
client = _client(connected=True, last_message_age=99999.0)
assert await _sweep({1: client}) == 0
client.force_reconnect_stale_session.assert_not_called()
@pytest.mark.asyncio
async def test_inside_the_grace_period_paho_keeps_the_job(self):
"""Below the grace window a reconnect may well be in flight; interrupting
it would turn a self-healing blip into a forced session rebuild."""
client = _client(connected=False, last_message_age=CONNECTION_WATCHDOG_OFFLINE_GRACE - 30)
assert await _sweep({1: client}) == 0
client.force_reconnect_stale_session.assert_not_called()
@pytest.mark.asyncio
async def test_a_client_that_never_had_a_session_is_left_to_paho(self):
"""No inbound message ever means this is the initial connect, where
retrying is both correct and the only thing to do."""
client = _client(connected=False, last_message_age=None)
assert await _sweep({1: client}) == 0
client.force_reconnect_stale_session.assert_not_called()
@pytest.mark.asyncio
async def test_reconnecting_clears_the_cooldown(self):
"""A printer that comes back must not carry a stale cooldown into its
next outage."""
client = _client(connected=False, last_message_age=9999.0)
await _sweep({1: client})
assert 1 in _connection_watchdog_last_attempt
client.state.connected = True
await _sweep({1: client})
assert 1 not in _connection_watchdog_last_attempt
class TestUnreachablePrinters:
@pytest.mark.asyncio
async def test_switched_off_printer_is_not_rebuilt(self):
"""Rebuilding a client against a host that isn't answering achieves
nothing and would log a warning per printer all night."""
client = _client(connected=False, last_message_age=9999.0)
assert await _sweep({1: client}, port_open=False) == 0
client.force_reconnect_stale_session.assert_not_called()
@pytest.mark.asyncio
async def test_unreachable_printer_still_takes_the_cooldown(self):
"""Otherwise every sweep re-probes the port of every dead printer."""
client = _client(connected=False, last_message_age=9999.0)
await _sweep({1: client}, port_open=False)
assert 1 in _connection_watchdog_last_attempt
class TestRetryInterval:
@pytest.mark.asyncio
async def test_does_not_rebuild_again_within_the_interval(self):
client = _client(connected=False, last_message_age=9999.0)
assert await _sweep({1: client}) == 1
assert await _sweep({1: client}) == 0
client.force_reconnect_stale_session.assert_called_once()
@pytest.mark.asyncio
async def test_retries_once_the_interval_has_passed(self):
client = _client(connected=False, last_message_age=9999.0)
await _sweep({1: client})
_connection_watchdog_last_attempt[1] -= CONNECTION_WATCHDOG_RETRY_INTERVAL + 1
assert await _sweep({1: client}) == 1
assert client.force_reconnect_stale_session.call_count == 2
class TestSweepIsFaultTolerant:
@pytest.mark.asyncio
async def test_one_broken_client_does_not_stop_the_others(self):
"""A farm sweep that aborts on the first bad client would leave every
printer after it unrecovered."""
bad = _client(connected=False, last_message_age=9999.0)
bad.force_reconnect_stale_session.side_effect = RuntimeError("boom")
good = _client(connected=False, last_message_age=9999.0)
assert await _sweep({1: bad, 2: good}) == 2
good.force_reconnect_stale_session.assert_called_once()