diff --git a/backend/app/services/bambu_mqtt.py b/backend/app/services/bambu_mqtt.py index f39a4892d..a0fd0e9ca 100644 --- a/backend/app/services/bambu_mqtt.py +++ b/backend/app/services/bambu_mqtt.py @@ -24,6 +24,7 @@ import paho.mqtt.client as mqtt from backend.app.services.hms_actions import HMSAction, get_actions_for_error_code from backend.app.services.hms_errors import describe_fault from backend.app.utils.ams_drying import ACTIVE_DRY_STATUSES +from backend.app.utils.paho_teardown import retire_paho_client logger = logging.getLogger(__name__) @@ -1610,15 +1611,14 @@ class BambuMQTTClient: # reconnect, mixing stale commands into the next dispatch and # triggering 0500_4003 SD R/W on the printer. # - # Paho-network-thread callers (line ~2604/~2623 — dev-mode probe and - # ams_filament_setting zombie detection inside `_update_state`) - # → socket-close fallback. Calling `loop_stop()` from inside the - # network thread would self-join and deadlock; the safe pattern is - # to close the socket and let paho's own loop detect the broken - # connection and auto-reconnect (same instance, same client_id — - # queue replay is theoretically possible here but those paths have - # always done socket-close and #1136 was specifically triggered - # from the dispatch path). + # Paho-network-thread callers (dev-mode probe and ams_filament_setting + # zombie detection, both inside `_update_state`) + # → socket-close fallback. There is no running loop on that thread to + # hand the rebuilt client, so close the socket and let paho's own + # loop detect the broken connection and auto-reconnect (same + # instance, same client_id — queue replay is theoretically possible + # here but those paths have always done socket-close and #1136 was + # specifically triggered from the dispatch path). logger.warning("[%s] Forcing MQTT reconnect: %s", self.serial_number, reason) self._stale_reconnecting = True self.state.connected = False @@ -1629,11 +1629,11 @@ class BambuMQTTClient: def _reset_client_for_reconnect(self) -> None: """Route between hard-reset and socket-close based on caller thread. - Hard-reset (preferred) requires we're not running on paho's network - thread, since `loop_stop()` on the same thread deadlocks. Detect via - ``asyncio.get_running_loop()`` — paho's callback thread has no loop; - every legitimate hard-reset caller (FastAPI handlers, background - async tasks) does.""" + Hard-reset (preferred) rebuilds the client, and the rebuild needs a + running loop to hand to ``connect()``. ``asyncio.get_running_loop()`` + answers that and identifies the caller in one go — paho's callback + thread has no loop; every legitimate hard-reset caller (FastAPI + handlers, background async tasks) does.""" try: loop = asyncio.get_running_loop() except RuntimeError: @@ -1650,18 +1650,15 @@ class BambuMQTTClient: client_id, so the broker drops the old session and paho's local QoS 1 queue is gone. Must NOT be called from paho's network thread. Caller is responsible for setting ``_stale_reconnecting`` and - broadcasting the disconnected state.""" + broadcasting the disconnected state. + + Returns as fast as it can build a client: the old one's teardown is + handed off rather than waited on, because waiting on it is what + stopped the event loop in #3068. See ``retire_paho_client``.""" old_client = self._client self._client = None if old_client is not None: - try: - old_client.disconnect() # MQTT DISCONNECT — broker drops session - except Exception: - pass - try: - old_client.loop_stop() # blocks briefly until the network thread exits - except Exception: - pass + retire_paho_client(old_client, self.serial_number) # Skip reconnect if no asyncio loop is available (test environment or # pre-init). The next initial connect() call from PrinterManager will # set up the client fresh. @@ -6384,14 +6381,32 @@ class BambuMQTTClient: return True def disconnect(self, timeout: float = 0): - """Disconnect from the printer.""" + """Disconnect from the printer. + + Waits up to *timeout* for paho to report the disconnect, then lets the + client go without joining its network thread — the callers are route + handlers (printer edited, deleted, disconnected by hand) running on the + asyncio thread, and that join has no bound (#3068).""" if self._client: + old_client = self._client self._disconnection_event = threading.Event() - self._client.disconnect() + old_client.disconnect() + # The callback that sets this fires on paho's thread, so it has to + # be given its window before retire_paho_client detaches it. self._disconnection_event.wait(timeout=timeout) - self._client.loop_stop() self._client = None + retire_paho_client(old_client, self.serial_number) self.state.connected = False + # Deliberately no on_state_change here. paho's disconnect callback + # used to land during the join, but `_on_disconnect` suppresses + # itself for a clean disconnect of a printer that reported within + # the last 10s -- which is every healthy printer -- so a + # hand-disconnected printer never broadcast one. Announcing it now + # would fire the connected→disconnected edge in + # `on_printer_status_change` and notify the user their printer went + # offline a minute after they disconnected it on purpose (#1752). + # The callers drop the client from the manager anyway, so the next + # status read already shows it gone. def send_command(self, command: dict): """Send a command to the printer.""" diff --git a/backend/app/services/mqtt_relay.py b/backend/app/services/mqtt_relay.py index 8f04d48cb..bb839aff0 100644 --- a/backend/app/services/mqtt_relay.py +++ b/backend/app/services/mqtt_relay.py @@ -15,6 +15,8 @@ from typing import Any import paho.mqtt.client as mqtt +from backend.app.utils.paho_teardown import retire_paho_client + logger = logging.getLogger(__name__) @@ -200,7 +202,7 @@ class MQTTRelayService: self._disconnection_event = threading.Event() self.client.disconnect() await asyncio.to_thread(self._disconnection_event.wait, timeout=timeout) - self.client.loop_stop() + retire_paho_client(self.client, "relay") except Exception as e: logger.debug("MQTT disconnect error (ignored): %s", e) finally: diff --git a/backend/app/services/mqtt_smart_plug.py b/backend/app/services/mqtt_smart_plug.py index 4d3855257..f778807f7 100644 --- a/backend/app/services/mqtt_smart_plug.py +++ b/backend/app/services/mqtt_smart_plug.py @@ -13,6 +13,8 @@ from typing import Any import paho.mqtt.client as mqtt +from backend.app.utils.paho_teardown import retire_paho_client + logger = logging.getLogger(__name__) @@ -482,7 +484,7 @@ class MQTTSmartPlugService: self._disconnection_event = threading.Event() self.client.disconnect() await asyncio.to_thread(self._disconnection_event.wait, timeout=timeout) - self.client.loop_stop() + retire_paho_client(self.client, "smart-plugs") except Exception as e: logger.debug("MQTT smart plug disconnect error (ignored): %s", e) finally: diff --git a/backend/app/utils/paho_teardown.py b/backend/app/utils/paho_teardown.py new file mode 100644 index 000000000..815bb921e --- /dev/null +++ b/backend/app/utils/paho_teardown.py @@ -0,0 +1,96 @@ +"""Letting go of a paho MQTT client without waiting for its network thread. + +`Client.loop_stop()` is two statements: set `_thread_terminate`, then `join()` +the network thread with no timeout. That thread only reads the flag between +iterations of `loop_forever`, so it cannot read it while parked inside +`reconnect()` -> `_ssl_wrap_socket()` -> `do_handshake()`. paho gives that +handshake the connection's keepalive as its socket timeout -- 30s for a printer +-- and a socket timeout is per operation, renewed by every byte the peer sends. +A broker that still answers on its port but never finishes the handshake +therefore holds the join open for as long as it keeps trickling; a silent one +still holds it 30s. + +Whoever called `loop_stop()` waits that out, and in Bambuddy that caller is the +asyncio thread. #3068: a printer 38 hours offline, still answering on 8883, was +picked up by the connection watchdog exactly as intended; the rebuild ended in +that join and the process stopped serving HTTP -- UI, API and health check -- +while staying alive, so the container's `restart: unless-stopped` never fired. +#1445 was the same join reached from the add-printer probe. +""" + +import logging +import threading +import time + +logger = logging.getLogger(__name__) + +# How long a retirement may take before it is worth a line in the support +# bundle. A healthy paho thread exits in well under a second. +_RETIRE_SLOW_SECONDS = 5.0 + +# The callbacks Bambuddy's three MQTT services set between them. Anything else +# paho offers is already None because nobody here assigns it. +_CALLBACKS = ("on_connect", "on_disconnect", "on_subscribe", "on_message") + + +def retire_paho_client(client, label: str) -> None: + """Shut *client* down on a thread of its own and return immediately. + + *label* names the connection in logs and in the retirement thread's name, + which is where a thread dump from the next stuck one will be read. + + Two things happen inline rather than on that thread: + + - The callbacks are cleared here. Blocking until the network thread was + gone is what used to guarantee a client we had let go of could no longer + touch our state; with the teardown detached, a zombie that finishes its + handshake would auto-reconnect and report itself connected behind its + replacement's back. + - Nothing else -- not even `disconnect()`, which is cheap enough to run + here (it queues a packet and returns) but would be one more thing + between the caller and its return for no gain, since the thread starts + within microseconds. It still happens and still matters: it is what + stops paho's auto-reconnect, and with it the chance of an unacked + `project_file` replaying onto a revived session (#1136). + """ + for attr in _CALLBACKS: + try: + setattr(client, attr, None) + except Exception: # pragma: no cover - paho always allows this + pass + + def _teardown() -> None: + started = time.monotonic() + try: + client.disconnect() + except Exception: + pass + try: + client.loop_stop() + except Exception: + pass + waited = time.monotonic() - started + if waited >= _RETIRE_SLOW_SECONDS: + # The stall that used to be the event loop's. Worth saying out + # loud: it means this connection is wedged somewhere paho cannot + # interrupt, and the next report of it should not have to be + # diagnosed from a thread dump again. + logger.warning( + "[%s] Retiring the old MQTT client took %.0fs (paho's network thread would " + "not stop). The connection was replaced anyway.", + label, + waited, + ) + + try: + threading.Thread(target=_teardown, name=f"mqtt-retire-{label}", daemon=True).start() + except RuntimeError as e: + # Out of threads entirely, which means the process has larger problems. + # Send the DISCONNECT inline anyway -- it is what keeps the abandoned + # session from reconnecting and replaying (#1136) -- and leave the + # network thread to paho. + logger.error("[%s] Could not start the MQTT teardown thread: %s", label, e) + try: + client.disconnect() + except Exception: + pass diff --git a/backend/tests/unit/test_mqtt_client_retirement_3068.py b/backend/tests/unit/test_mqtt_client_retirement_3068.py new file mode 100644 index 000000000..ab942a86f --- /dev/null +++ b/backend/tests/unit/test_mqtt_client_retirement_3068.py @@ -0,0 +1,331 @@ +"""Letting go of a paho client must never block the thread that let it go. + +#3068: a printer that had been offline 38 hours still answered on 8883. The +connection watchdog rebuilt its session, which ended in paho's `loop_stop()` +-- set a terminate flag, then `join()` the network thread with no timeout. The +network thread was parked in `reconnect()`'s TLS handshake, where it cannot +read that flag, so the join never returned. The join was running on the asyncio +thread: the process stayed up, `/health` stopped being answered, and Docker's +`restart: unless-stopped` never fired because nothing had exited. + +Four call paths reach that join from the event loop -- the connection watchdog, +the queue dispatch deadline, `check_staleness()` on an ordinary status poll, +and `disconnect()` from the printer routes. They all funnel through the two +places tested here. The relay and smart-plug services had the same join on +their shutdown path and now share the same teardown. +""" + +import asyncio +import threading +import time +from unittest.mock import MagicMock, patch + +import pytest + +from backend.app.services.bambu_mqtt import BambuMQTTClient +from backend.app.utils.paho_teardown import retire_paho_client + + +class WedgedPahoClient: + """A paho client whose network thread will not stop. + + `loop_stop()` blocks until `release()` is called, which is what a real one + does while its thread sits in `do_handshake()` against a printer that + answers TCP and then goes quiet. + """ + + def __init__(self): + self.released = threading.Event() + self.disconnect_called = threading.Event() + self.loop_stop_returned = threading.Event() + self.on_connect = "sentinel" + self.on_disconnect = "sentinel" + self.on_subscribe = "sentinel" + self.on_message = "sentinel" + + def disconnect(self): + self.disconnect_called.set() + + def loop_stop(self): + self.released.wait(timeout=10) + self.loop_stop_returned.set() + + def release(self): + self.released.set() + + +@pytest.fixture +def client(): + return BambuMQTTClient( + ip_address="192.168.1.100", + serial_number="00M09A123456789", + access_code="12345678", + ) + + +class TestRetiringAClient: + def test_it_returns_while_the_old_client_is_still_stopping(self): + wedged = WedgedPahoClient() + try: + started = time.monotonic() + retire_paho_client(wedged, "00M09A123456789") + elapsed = time.monotonic() - started + + assert elapsed < 2.0, f"retirement blocked the caller for {elapsed:.2f}s (#3068)" + assert not wedged.loop_stop_returned.is_set(), "loop_stop was joined, not handed off" + finally: + wedged.release() + + def test_the_callbacks_are_detached_before_the_caller_moves_on(self): + # Blocking until the network thread was gone is what used to guarantee + # a client we had let go of could no longer touch our state. With the + # teardown detached, a zombie that finishes its handshake would + # auto-reconnect and set connected=True behind its replacement's back, + # so the detach has to happen inline. + wedged = WedgedPahoClient() + try: + retire_paho_client(wedged, "00M09A123456789") + assert wedged.on_connect is None + assert wedged.on_disconnect is None + assert wedged.on_subscribe is None + assert wedged.on_message is None + finally: + wedged.release() + + def test_the_old_session_is_still_disconnected_and_stopped(self): + # disconnect() is what stops paho's auto-reconnect, and with it the + # chance of an unacked project_file replaying onto a revived session + # (#1136). Handing it off must not mean skipping it. + wedged = WedgedPahoClient() + assert wedged.disconnect_called.wait(timeout=0) is False + retire_paho_client(wedged, "00M09A123456789") + assert wedged.disconnect_called.wait(timeout=5), "the old client was never disconnected" + wedged.release() + assert wedged.loop_stop_returned.wait(timeout=5), "the old client's loop was never stopped" + + def test_a_client_that_raises_on_teardown_is_still_let_go(self): + exploding = MagicMock() + exploding.disconnect.side_effect = RuntimeError("socket already gone") + exploding.loop_stop.side_effect = RuntimeError("no thread") + + retire_paho_client(exploding, "00M09A123456789") + + deadline = time.monotonic() + 5 + while time.monotonic() < deadline and not exploding.loop_stop.called: + time.sleep(0.01) + assert exploding.loop_stop.called + + def test_the_retirement_thread_is_named_for_the_printer(self): + # A support bundle's thread dump is how the next one of these gets + # recognised; an anonymous Thread-7 says nothing. + wedged = WedgedPahoClient() + try: + retire_paho_client(wedged, "00M09A123456789") + names = [t.name for t in threading.enumerate()] + assert "mqtt-retire-00M09A123456789" in names + finally: + wedged.release() + + +class TestHardReset: + def test_it_does_not_wait_for_the_old_network_thread(self, client): + wedged = WedgedPahoClient() + client._client = wedged + client._loop = None # no rebuild, so only the teardown is measured + + try: + started = time.monotonic() + client._hard_reset_client() + elapsed = time.monotonic() - started + + assert elapsed < 2.0, f"_hard_reset_client blocked for {elapsed:.2f}s (#3068)" + assert client._client is None + finally: + wedged.release() + + def test_the_replacement_gets_a_fresh_client_id(self, client): + # The #1136 property: the new session must not inherit paho's QoS 1 + # queue, which is what a new client_id buys. + wedged = WedgedPahoClient() + client._client = wedged + client._loop = MagicMock() + + with patch("backend.app.services.bambu_mqtt.mqtt.Client") as MockClient: + MockClient.return_value = MagicMock() + try: + client._hard_reset_client() + finally: + wedged.release() + + assert MockClient.call_count == 1 + new_id = MockClient.call_args.kwargs["client_id"] + assert client.serial_number in new_id + assert client._client is MockClient.return_value + + @pytest.mark.asyncio + async def test_a_wedged_printer_does_not_stall_the_event_loop(self, client): + # The reported failure, end to end: force_reconnect_stale_session is + # what the connection watchdog and the queue dispatch deadline both + # call, from a coroutine. A heartbeat has to keep ticking through it. + wedged = WedgedPahoClient() + client._client = wedged + + ticks = 0 + + async def heartbeat(): + nonlocal ticks + while True: + await asyncio.sleep(0.02) + ticks += 1 + + beat = asyncio.create_task(heartbeat()) + try: + with patch("backend.app.services.bambu_mqtt.mqtt.Client") as MockClient: + MockClient.return_value = MagicMock() + started = time.monotonic() + client.force_reconnect_stale_session("offline for 900s, port still answering") + elapsed = time.monotonic() - started + await asyncio.sleep(0.1) + finally: + beat.cancel() + wedged.release() + try: + await beat + except asyncio.CancelledError: + pass + + assert elapsed < 2.0, f"the forced reconnect held the event loop for {elapsed:.2f}s (#3068)" + assert ticks > 0, "the event loop made no progress while the old client was stopping" + assert client.state.connected is False + + +class TestDisconnect: + def test_it_does_not_wait_for_the_old_network_thread(self, client): + # Reached from PUT/DELETE /printers/{id} and POST + # /printers/{id}/disconnect, all on the asyncio thread. + wedged = WedgedPahoClient() + client._client = wedged + client.state.connected = True + + try: + started = time.monotonic() + client.disconnect() + elapsed = time.monotonic() - started + + assert elapsed < 2.0, f"disconnect() blocked for {elapsed:.2f}s (#3068)" + assert client._client is None + assert client.state.connected is False + finally: + wedged.release() + + def test_the_disconnect_callback_still_gets_its_window(self, client): + # The callback that releases the timeout fires on paho's thread, so it + # has to run before the retirement detaches it -- otherwise every + # caller with a non-zero timeout waits the timeout out in full. + class AnsweringClient(WedgedPahoClient): + def disconnect(self): + super().disconnect() + if self.on_disconnect is not None: + self.on_disconnect(self, None) + + answering = AnsweringClient() + answering.on_disconnect = client._on_disconnect # as connect() wires it + client._client = answering + + try: + started = time.monotonic() + client.disconnect(timeout=5) + elapsed = time.monotonic() - started + + assert elapsed < 2.0, ( + f"disconnect(timeout=5) took {elapsed:.2f}s — the callback was detached " + "before it could report the disconnect" + ) + assert client._disconnection_event.is_set() + finally: + answering.release() + + def test_disconnecting_twice_is_harmless(self, client): + wedged = WedgedPahoClient() + client._client = wedged + try: + client.disconnect() + client.disconnect() + assert client._client is None + finally: + wedged.release() + + +class TestTheOtherMqttServices: + """The relay and the smart-plug service tear their brokers down the same + way, at shutdown. A wedged broker there does not stop request serving -- + nothing is being served by then -- but it does stop the process exiting, + which leaves the container to be killed rather than stopped.""" + + @pytest.mark.asyncio + async def test_the_relay_does_not_wait_for_its_network_thread(self): + from backend.app.services.mqtt_relay import MQTTRelayService + + wedged = WedgedPahoClient() + service = MQTTRelayService() + service.client = wedged + service.connected = True + + try: + started = time.monotonic() + await service.disconnect() + elapsed = time.monotonic() - started + + assert elapsed < 2.0, f"relay shutdown blocked for {elapsed:.2f}s (#3068)" + assert service.client is None + # Only the retirement detaches callbacks, so this proves it ran + # rather than the service's except-block swallowing it. + assert wedged.on_disconnect is None + finally: + wedged.release() + + @pytest.mark.asyncio + async def test_the_smart_plug_service_does_not_wait_for_its_network_thread(self): + from backend.app.services.mqtt_smart_plug import MQTTSmartPlugService + + wedged = WedgedPahoClient() + service = MQTTSmartPlugService() + service.client = wedged + service.connected = True + + try: + started = time.monotonic() + await service.disconnect() + elapsed = time.monotonic() - started + + assert elapsed < 2.0, f"smart-plug shutdown blocked for {elapsed:.2f}s (#3068)" + assert service.client is None + # Only the retirement detaches callbacks, so this proves it ran + # rather than the service's except-block swallowing it. + assert wedged.on_disconnect is None + finally: + wedged.release() + + +class TestDisconnectStaysQuiet: + def test_a_hand_disconnected_printer_is_not_announced_as_offline(self, client): + # paho's disconnect callback used to land during the join, but + # `_on_disconnect` suppresses itself for a clean disconnect of a + # printer that reported in the last 10s, so a healthy printer + # disconnected on purpose never broadcast one. Announcing it here + # instead would reach the connected→disconnected edge and notify the + # user their printer went offline a minute later (#1752). + seen = [] + client.on_state_change = seen.append + client._last_message_time = time.time() + wedged = WedgedPahoClient() + client._client = wedged + client.state.connected = True + + try: + client.disconnect() + finally: + wedged.release() + + assert seen == [], "disconnecting a printer by hand announced it as offline" + assert client.state.connected is False