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.

CHANGELOG.md

@@ -29,6 +29,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **Every FTP session Bambuddy opens now records how it closed (#3009, reported by @grengojbo)** — the report traced a print completion that opened two FTP connections to the printer, deleted one file and then, as far as the log showed, did nothing else until the printer was powered off 21 minutes later, and concluded the connections were being left open. They were not: the post-print SD-card cleanup opens one connection per candidate filename and closes each in a `finally`, which a run against a real FTPS server confirms at the server end for both the delete and the 550 not-here case. The trouble is that nothing in the log could have said so. Neither the clean close nor the hard socket drop logged anything at any level, so a session closed properly and a socket genuinely abandoned produced the same output — none — and the only way to tell them apart was to read the source. Both now log one DEBUG line naming the printer, whether QUIT was acknowledged or the socket had to be dropped without it, why, and how long the session was held. Every connect in a debug log is now paired with a close, so the next person suspecting a leaked FTP connection can settle it from a support bundle rather than by inference. Nothing about the connection handling itself changed, and at default log level nothing new is printed. This does not explain the SD-card read/write error in that report or in #645; it only removes one theory from the list by making it checkable.
 
 ### Fixed
+- **A printer that had been offline for hours could stop the whole server (#3068, reported by @bazza2000)** — the reporter's A1 had been unreachable for 38 hours but still answered on its MQTT port, which is exactly the case the connection watchdog exists for: rebuild the session with a fresh client so nothing left over from the dead one can replay onto the new print. The rebuild ended in a call that waits for the old connection's network thread to finish, and that thread was stuck part-way through a TLS handshake the printer never completed, where nothing can interrupt it. The wait had no limit and it ran on the thread that serves every request, so Bambuddy stopped answering — the web UI, the API and the health check alike — while the process itself stayed up, which is why Docker's restart policy never kicked in. Retiring an old connection no longer waits for it: the replacement is built immediately and the old one is shut down in the background, cut off from the new session first so a printer that eventually answers cannot report itself connected again after being replaced. The same wait sat on three other printer paths — the queue's dispatch recovery, the staleness check behind an ordinary status poll, and editing, deleting or disconnecting a printer — and on the shutdown of the MQTT relay and smart-plug connections, where a wedged broker kept the process from exiting at all. All of them are fixed together. A connection that takes more than a few seconds to shut down now says so in the log instead of silently taking the server with it.
 - **Auto-drying skipped every composite spool (#3067, reported by @TheUltimateC0der)** — the reporter's AMS-HT would not dry PA6-CF, and drying the same spool by hand worked. The scheduler reduced a tray to a preset key by splitting on spaces only, so "PA6-CF" stayed "PA6-CF", matched none of the eight rows the preset table has, and the tray was read as holding nothing worth drying. It was never only nylon: of the 41 types a printer can report, 33 had no row under that rule, and 20 of those — every -CF, -GF and -AERO variant of PLA, PETG, ABS, ASA, PC and PA — have a base material sitting right there. The drying popover has resolved these correctly since 1.2.5, which is exactly why doing it by hand worked; the scheduler now shares that rule, including the alias map that reads PA6, PA11, PA12, PAHT, PPA and Nylon as PA. A row you have added for the exact type still wins over the base material's, and a material whose base is not in the table either is still left alone rather than dried at a temperature nobody chose. The same lookup decides per-filament humidity thresholds and the chamber preheat target, so a composite now gets the override and the preheat that were set for its material too.
 - **An archive gave up on its 3MF for good after one slow transfer (#3063, reported by @dfrysinger)** — the reporter's P1S had the file on its card and was serving it; the 19 MB transfer just did not finish inside the budget while the printer was also running its camera, its status messages and the job upload. Bambuddy wrote an empty archive and never looked again — and then downloaded that same file successfully three times in the next two minutes, throwing each copy away, because the only code that would have attached one had already run. Recovery for a fallback archive existed but was armed for exactly one give-up, the FTPS cool-off. It now also covers a transfer that ran out of time: Bambuddy comes back after one, four and ten minutes and fills the row in where an attempt lands. The two are told apart by what the printer said — a file that genuinely is not on the card is answered with 550, and that answer does not improve with waiting, so nothing is scheduled for it. Nor for a 3MF that downloaded fine but turned out to be another plate's, where retrying would put back exactly what was just discarded. The archives banner has wording for this case too, because the old text sent an owner whose card was working to switch on a setting that was already on; it names the Connection Timeout setting instead.
 - **Items Printed could not be set to 0 after a total plate failure (#3051, reported by @tdavis75)** — a jam ruined everything on the plate while the printer reported the job a success, so the honest count of usable parts was zero; the field refused to go below one. The manual quantity override is what the documentation points at for exactly this correction, and a project's completed-items count sums that column, so there was no way to tell a project that a job produced nothing. The floor was in the edit dialog alone — the API had always stored whatever it was given, which also meant a negative count was accepted and would have subtracted from the project totals. Zero is now typeable and the column is bounded at zero, and the Filament Trends widget counts a zeroed archive as no prints rather than silently reading the 0 as "unset" and charging one.

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