fix(mqtt): resolve reconnect deadlock after printer disconnect (Issue #105)

The bridge could get permanently stuck after a printer went offline and
came back, even with the printer confirmed reachable via ping/nc - only
a full container restart recovered it. Two compounding bugs:

1. Two independent code paths could trigger _reconnect() concurrently
   with no coordination: the reader thread (on a failed keepalive ping)
   and publish()/publish_web() (on a failed sendall(), which happens
   constantly once the socket is dead, since the poll loop calls
   query_info() every poll_interval). Both would race into their own
   _do_connect(), each opening a competing TLS handshake against a
   printer that likely only accepts one mTLS session at a time - so
   neither converges, and the failure repeats every ~3s instead of
   backing off. Added a lock so a second _reconnect() call waits for
   the first to finish instead of starting a competing handshake.

2. publish() swallows send/reconnect failures internally and returns
   None instead of raising - so _poll_loop's `if info: ...` branch was
   silently skipped on failure, but the surrounding except-block (which
   would have triggered the existing, correct offline/reconnect
   transition) was never reached, since no exception was ever thrown.
   The poll loop had no way to tell "printer sent nothing this tick"
   apart from "the MQTT session is dead". Added client.is_connected()
   and check it explicitly when query_info() returns falsy, routing a
   dead session into the same clean offline branch already used for a
   TCP-unreachable printer.

Verified: bridge continued printing normally throughout (live print
in progress on the real printer during this fix), full test suite
green (111 tests), new tests cover the concurrent-reconnect lock and
the poll-loop offline transition on a swallowed send failure.

NIGHTLY_CHANGELOG.md

@@ -1,3 +1,4 @@
 ## Changes in this build
 
 - Feat: `server/files/metadata` now uses the printer's own `buried/report` analytics event (fires once per print start, regardless of slicer) as a fallback for `size`/`estimated_time`/`layer_count` — fixes broken `size: 1`/`estimated_time: null` placeholders for files not in the bridge's own GCode store, e.g. prints started directly from Anycubic Slicer Next (Issue #102, thanks @fmontagna). Also surfaces the printer's storage usage (`storage_total_mb`/`storage_used_mb`) in `/api/state`.
+- Fix: **the bridge could get stuck in an endless reconnect loop after a printer disconnect, even once the printer was back online and reachable** — a container restart was the only way out. Two independent reconnect paths (the MQTT reader thread and the status-poll loop) could race into competing TLS handshakes, and the poll loop never noticed a dead MQTT session on its own since a failed send silently returned no data instead of raising an error. The bridge now reconnects automatically without manual intervention (Issue #105, thanks @p2l for the precise report).

kobrax_client.py

@@ -126,6 +126,13 @@ class KobraXClient:
         # underneath it (Issue #53). Protects against recv on a stale fd.
         self._sock_gen = 0
         self._running = False
+        # Guards _reconnect() against concurrent invocation - both the reader
+        # thread (keepalive ping failure) and publish()/publish_web() (send
+        # failure) can trigger a reconnect independently. Without this, two
+        # threads could race into _do_connect() at once, each opening its own
+        # competing TLS handshake to a printer that likely only accepts one
+        # mTLS session at a time (Issue #105).
+        self._reconnect_lock = threading.Lock()
 
         # Pending requests by msgid (for response ACK)
         self._pending_msgid: dict[str, dict] = {}
@@ -231,41 +238,63 @@ class KobraXClient:
             self._sock = None
             self._sock_gen += 1
 
+    def is_connected(self) -> bool:
+        """Thread-safe check whether the MQTT socket is currently up. Used by
+        the bridge's poll loop to detect a dead session even when publish()
+        already swallowed the send failure and returned None instead of
+        raising (Issue #105) - a TCP-reachable printer alone doesn't mean the
+        MQTT/TLS session is still alive."""
+        with self._lock:
+            return self._sock is not None
+
     def _reconnect(self):
         """Persistent reconnect: keeps retrying forever until the printer is
         responds or disconnect() was called. Backoff caps at 60 s. The
         first 5 attempts log as WARNING (acute connection issue), afterwards
         only DEBUG to avoid log spam during long printer outages (e.g. switched
-        ausgeschaltet) zu vermeiden."""
-        log.warning("Connection lost - reconnecting...")
-        # Close + invalidation under the lock so no sender is mid-sendall
-        # auf den gerade geschlossenen Socket trifft (Issue #53).
-        with self._lock:
-            try:
-                if self._sock is not None:
-                    self._sock.close()
-            except Exception:
-                pass
-            self._sock = None
-            self._sock_gen += 1
-        delays = [2, 4, 8, 15, 30, 60]
-        attempt = 0
-        while self._running:
-            delay = delays[min(attempt, len(delays) - 1)]
-            try:
-                self._do_connect()
-                log.info("Reconnect successful (after %d attempts)", attempt + 1)
-                return True
-            except Exception as e:
-                attempt += 1
-                lvl = log.warning if attempt <= 5 else log.debug
-                lvl("Reconnect fehlgeschlagen (%s, Versuch %d), warte %ss…", e, attempt, delay)
-                # Split sleep so disconnect() breaks the loop faster.
-                slept = 0.0
-                while slept < delay and self._running:
-                    time.sleep(min(0.5, delay - slept))
-                    slept += 0.5
-        return False  # only when disconnect() was called
+        ausgeschaltet) zu vermeiden.
+
+        Guarded by _reconnect_lock (Issue #105): if another thread's reconnect
+        is already in flight, this call waits for it to finish instead of
+        starting a second, competing _do_connect() - the printer likely only
+        accepts one mTLS session at a time, so two parallel handshakes would
+        just interfere with each other and neither converges."""
+        if not self._reconnect_lock.acquire(blocking=False):
+            self._reconnect_lock.acquire()
+            self._reconnect_lock.release()
+            return self._sock is not None
+        try:
+            log.warning("Connection lost - reconnecting...")
+            # Close + invalidation under the lock so no sender is mid-sendall
+            # auf den gerade geschlossenen Socket trifft (Issue #53).
+            with self._lock:
+                try:
+                    if self._sock is not None:
+                        self._sock.close()
+                except Exception:
+                    pass
+                self._sock = None
+                self._sock_gen += 1
+            delays = [2, 4, 8, 15, 30, 60]
+            attempt = 0
+            while self._running:
+                delay = delays[min(attempt, len(delays) - 1)]
+                try:
+                    self._do_connect()
+                    log.info("Reconnect successful (after %d attempts)", attempt + 1)
+                    return True
+                except Exception as e:
+                    attempt += 1
+                    lvl = log.warning if attempt <= 5 else log.debug
+                    lvl("Reconnect fehlgeschlagen (%s, Versuch %d), warte %ss…", e, attempt, delay)
+                    # Split sleep so disconnect() breaks the loop faster.
+                    slept = 0.0
+                    while slept < delay and self._running:
+                        time.sleep(min(0.5, delay - slept))
+                        slept += 0.5
+            return False  # only when disconnect() was called
+        finally:
+            self._reconnect_lock.release()
 
     def _subscribe(self, topic: str):
         with self._lock:

kobrax_moonraker_bridge.py

@@ -5755,6 +5755,25 @@ class KobraXBridge:
                 info = self.client.query_info()
                 if info:
                     self._on_info(info)
+                elif not self.client.is_connected():
+                    # publish() swallows send/reconnect failures internally and
+                    # just returns None (Issue #105) - a falsy `info` alone
+                    # doesn't distinguish "printer sent nothing this tick" from
+                    # "the MQTT session itself is dead". Check is_connected()
+                    # explicitly so a dead session gets routed into the same
+                    # clean offline/reconnect path as a TCP-unreachable printer,
+                    # instead of silently retrying every poll_interval forever.
+                    log.warning("MQTT connection lost (query returned no response) - switching to offline mode")
+                    self._state["print_state"] = "error"
+                    self._state["kobra_state"] = "offline"
+                    self._state["connection_error"] = f"MQTT connection lost ({self._args.printer_ip})"
+                    try:
+                        self.client.disconnect()
+                    except Exception:
+                        pass
+                    _offline = True
+                    stop_event.wait(getattr(self._args, "poll_interval", 3))
+                    continue
                 # While printing: query print/report directly
                 if self._state["print_state"] in ("printing", "preheating",
                                                    "auto_leveling", "checking", "init"):

tests/test_mqtt_reconnect.py

@@ -0,0 +1,94 @@
+"""
+Tests für den MQTT-Reconnect-Mechanismus (Issue #105) — der Client hing nach
+einem Drucker-Reconnect fest, weil zwei unabhängige Fehlerpfade (der Reader-
+Thread-Keepalive und publish()'s eigener Reconnect-Trigger) unkoordiniert
+parallel liefen, UND weil der Poll-Loop einen None-Rückgabewert von publish()
+(statt einer Exception) nie als "Verbindung tot" erkannte.
+"""
+import threading
+import time
+
+import pytest
+
+from kobrax_client import KobraXClient
+
+
+def _client(**overrides):
+    kwargs = dict(
+        host="192.168.1.100", username="u", password="p",
+        mode_id="20030", device_id="abc123", port=9883,
+        client_id="test",
+    )
+    kwargs.update(overrides)
+    return KobraXClient(**kwargs)
+
+
+def test_is_connected_false_when_no_socket():
+    c = _client()
+    assert c.is_connected() is False
+
+
+def test_is_connected_true_when_socket_present():
+    c = _client()
+    c._sock = object()  # any truthy stand-in for a real socket
+    assert c.is_connected() is True
+
+
+def test_reconnect_concurrent_calls_only_run_do_connect_once():
+    """Two threads calling _reconnect() at the same time must not both run
+    _do_connect() - only one handshake should happen; the second caller waits
+    for the first instead of racing it (Issue #105)."""
+    c = _client()
+    c._running = True
+    do_connect_calls = []
+    call_lock = threading.Lock()
+    release_event = threading.Event()
+
+    def fake_do_connect():
+        with call_lock:
+            do_connect_calls.append(1)
+        # Simulate a slow handshake so the second _reconnect() call has time
+        # to observe the lock as already held.
+        release_event.wait(timeout=2.0)
+        c._sock = object()
+
+    c._do_connect = fake_do_connect
+
+    results = []
+
+    def run():
+        results.append(c._reconnect())
+
+    t1 = threading.Thread(target=run)
+    t2 = threading.Thread(target=run)
+    t1.start()
+    time.sleep(0.05)  # let t1 acquire the lock and enter _do_connect first
+    t2.start()
+    time.sleep(0.1)
+    release_event.set()  # let the in-flight handshake finish
+    t1.join(timeout=3)
+    t2.join(timeout=3)
+
+    assert len(do_connect_calls) == 1
+    assert results == [True, True]
+
+
+def test_reconnect_second_waiter_returns_after_first_completes():
+    c = _client()
+    c._running = True
+
+    def fake_do_connect():
+        time.sleep(0.1)
+        c._sock = object()
+
+    c._do_connect = fake_do_connect
+
+    t1 = threading.Thread(target=c._reconnect)
+    t1.start()
+    time.sleep(0.02)
+    # Second call while the first is still mid-handshake.
+    result = c._reconnect()
+    t1.join(timeout=3)
+
+    assert result is True
+    assert c._sock is not None

tests/test_poll_loop_reconnect.py

@@ -0,0 +1,96 @@
+"""
+Tests für _poll_loop's Umgang mit einer toten MQTT-Session (Issue #105).
+publish()/query_info() swallow send failures internally and return None
+instead of raising - the poll loop must treat that (combined with
+is_connected() == False) as "connection lost" and switch to the offline
+branch, instead of silently retrying forever every poll_interval.
+"""
+import argparse
+import tempfile
+import threading
+import time
+from unittest.mock import MagicMock
+
+import pytest
+
+from kobrax_moonraker_bridge import KobraXBridge
+
+
+def _bridge():
+    c = MagicMock()
+    c.callbacks = {}
+    c.connected = False
+    args = argparse.Namespace(
+        printer_ip="192.168.1.100", mqtt_port=9883, username="", password="",
+        mode_id="20030", device_id="", host="127.0.0.1", port=7125,
+        data_dir=tempfile.mkdtemp(prefix="kxpoll-"), poll_interval=0.05,
+    )
+    return KobraXBridge(c, args=args)
+
+
+def test_poll_loop_switches_to_offline_when_query_returns_none_and_disconnected():
+    b = _bridge()
+    b._state["print_state"] = "standby"
+    b._state["kobra_state"] = "free"
+    b.client.query_info.return_value = None
+    b.client.is_connected.return_value = False
+    b._printer_reachable = MagicMock(return_value=True)  # TCP still fine
+
+    stop_event = threading.Event()
+    t = threading.Thread(target=b._poll_loop, args=(stop_event,), daemon=True)
+    t.start()
+    time.sleep(0.2)
+    stop_event.set()
+    t.join(timeout=2)
+
+    assert b._state["kobra_state"] == "offline"
+    b.client.disconnect.assert_called()
+
+
+def test_poll_loop_stays_online_when_query_returns_none_but_still_connected():
+    """A single missed poll tick (info momentarily falsy) must not flip the
+    bridge offline if the MQTT session itself is still alive."""
+    b = _bridge()
+    b._state["print_state"] = "standby"
+    b._state["kobra_state"] = "free"
+    b.client.query_info.return_value = None
+    b.client.is_connected.return_value = True  # session still up
+    b.client.query_multicolor_box.return_value = None
+    b._printer_reachable = MagicMock(return_value=True)
+
+    stop_event = threading.Event()
+    t = threading.Thread(target=b._poll_loop, args=(stop_event,), daemon=True)
+    t.start()
+    time.sleep(0.2)
+    stop_event.set()
+    t.join(timeout=2)
+
+    assert b._state["kobra_state"] != "offline"
+
+
+def test_poll_loop_recovers_via_offline_branch_once_reachable_again():
+    """Once flipped offline, the existing offline branch should re-connect
+    as soon as the printer becomes reachable again (pre-existing behavior,
+    unaffected by this fix)."""
+    b = _bridge()
+    b._state["print_state"] = "standby"
+    b._state["kobra_state"] = "free"
+    b.client.query_info.return_value = None
+    b.client.is_connected.return_value = False
+    b._printer_reachable = MagicMock(return_value=True)
+
+    stop_event = threading.Event()
+    t = threading.Thread(target=b._poll_loop, args=(stop_event,), daemon=True)
+    t.start()
+    time.sleep(0.15)  # let it flip offline
+    assert b._state["kobra_state"] == "offline"
+
+    # Printer "comes back": client.connect() succeeds, subsequent query_info
+    # starts returning real data again.
+    b.client.connect.side_effect = None
+    b.client.query_info.return_value = {"data": {"state": "free"}}
+    time.sleep(0.2)
+    stop_event.set()
+    t.join(timeout=2)
+
+    b.client.connect.assert_called()