fix(mqtt): never wait for a wedged paho network thread (issue #3068)

    The reporter's A1 had been offline 38 hours and still answered on 8883, so
    the connection watchdog did exactly what it exists for: rebuild the session
    with a fresh client, since anything left in the old one's QoS 1 queue would
    otherwise replay onto the next print (#1136). The rebuild ended in paho's
    loop_stop(), which sets a terminate flag and then joins 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 keepalive as its socket
    timeout -- 30s here -- and a socket timeout is per operation, renewed by
    every byte the peer sends. A printer that answers TCP and then trickles
    holds the join open for as long as it likes.

    The join ran on the asyncio thread. Bambuddy stopped answering anything --
    UI, API, /health -- while the process stayed up, which is why a
    restart: unless-stopped container never restarted.

    Retiring a client no longer waits for it. The replacement is built at once
    and the old one is shut down on a thread of its own that nobody joins. Its
    callbacks are detached first, inline: 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, and with the teardown detached a zombie that finishes its
    handshake would otherwise auto-reconnect and report itself connected behind
    its replacement's back. disconnect() still goes out, still promptly, because
    that is what stops paho's auto-reconnect and the replay with it.

    The reported watchdog is one of six callers. The queue's dispatch recovery
    and check_staleness -- reached from an ordinary status poll -- share
    _hard_reset_client; editing, deleting and hand-disconnecting a printer share
    disconnect(); the relay and smart-plug services had the same join on their
    shutdown path, where a wedged broker stopped the process from exiting at
    all. #1445 was this join too, from the add-printer probe, and its off-loop
    teardown stays as it is.

    disconnect() stays quiet on the way out, as it always effectively did.
    paho's callback used to land during the join, but it suppresses itself for a
    clean disconnect of a printer that reported in the last ten seconds, so a
    healthy printer disconnected by hand never announced itself offline.
    Announcing it now would tell the user their printer had gone offline a
    minute after they disconnected it on purpose (#1752).

    A retirement that takes more than five seconds logs which printer it was.
    The whole point is that the next one of these should not have to be
    diagnosed from a thread dump.

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."""

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:

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:

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

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