diff --git a/NIGHTLY_CHANGELOG.md b/NIGHTLY_CHANGELOG.md index 87cefde..caf633d 100644 --- a/NIGHTLY_CHANGELOG.md +++ b/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). diff --git a/kobrax_client.py b/kobrax_client.py index 7d03980..a46a0fd 100644 --- a/kobrax_client.py +++ b/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: diff --git a/kobrax_moonraker_bridge.py b/kobrax_moonraker_bridge.py index 5792063..b7f3c13 100644 --- a/kobrax_moonraker_bridge.py +++ b/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"): diff --git a/tests/test_mqtt_reconnect.py b/tests/test_mqtt_reconnect.py new file mode 100644 index 0000000..e7ab5f5 --- /dev/null +++ b/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 diff --git a/tests/test_poll_loop_reconnect.py b/tests/test_poll_loop_reconnect.py new file mode 100644 index 0000000..89becff --- /dev/null +++ b/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()