fix(moonraker): deliver ws callbacks on the GUI thread, fail them on disconnect, fetch spools via the Spoolman proxy

BlocksScreen/devices/amu/manager.py

@@ -165,7 +165,8 @@ def fetch_spool(self, gate: int, spool_id: int) -> None:
             return
 
         def _on_result(result: dict | None) -> None:
-            if result is not None:
+            # An error reply parses as an empty spool and would blank the gate.
+            if result and "error" not in result:
                 self.spool_fetched.emit(gate, result)
 
         self._ws.api.get_spool(spool_id, _on_result)

BlocksScreen/lib/files.py

@@ -458,21 +458,25 @@ def request_print_duration(self, filename: str) -> None:
         job_id = str(metadata.job_id)
         if job_id in self._job_durations:
             return
-        # Mark before sending: error replies skip the callback.
+        # Mark before sending so an in-flight lookup is not re-requested.
         self._job_durations[job_id] = None
-        self.ws.api.history_get_job(
+        sent = self.ws.api.history_get_job(
             job_id,
             lambda result, name=filename, uid=job_id: self._history_job.emit(
                 name, uid, result or {}
             ),
         )
+        if not sent:
+            self._job_durations.pop(job_id, None)
 
     @QtCore.pyqtSlot(str, str, dict, name="on_history_job")
     def _on_history_job(self, filename: str, job_id: str, result: dict) -> None:
         """Cache a completed job's duration and re-emit fileinfo."""
         job = result.get("job") or {}
         status = job.get("status")
-        if status == "in_progress":
+        lost = (result.get("error") or {}).get("code") == -1
+        # Re-ask after a disconnect or a running job; a deleted job stays final.
+        if lost or status == "in_progress":
             self._job_durations.pop(job_id, None)
             return
         duration = job.get("print_duration")

BlocksScreen/lib/moonrakerComm.py

@@ -42,6 +42,7 @@ class MoonWebSocket(QtCore.QObject, threading.Thread):
     klippy_state_signal = QtCore.pyqtSignal(str, name="klippy_state")
     query_server_info_signal = QtCore.pyqtSignal(name="query_server_information")
     server_components_signal = QtCore.pyqtSignal(list, name="server_components")
+    request_reply_signal = QtCore.pyqtSignal(object, object, name="request_reply")
 
     _KLIPPY_NOTIFY_METHODS: typing.ClassVar[frozenset[str]] = frozenset(
         {
@@ -84,6 +85,7 @@ def __init__(self, parent: QtCore.QObject) -> None:
         )
 
         self.klippy_state_signal.connect(self.api.request_printer_info)
+        self.request_reply_signal.connect(self._run_callback)
         logger.info("Websocket object initialized")
 
     @property
@@ -201,6 +203,13 @@ def on_close(self, *args) -> None:
         _close_message = args[2] if len(args) == 3 else None
         with self._state_lock:
             self.connected = False
+        with self._request_lock:
+            callbacks = [e[2] for e in self.request_table.values() if e[2] is not None]
+            self.request_table.clear()
+        for callback in callbacks:
+            # Same shape as a Moonraker JSON-RPC error; -1 marks it client-side.
+            error = {"code": -1, "message": "disconnected"}
+            self.request_reply_signal.emit(callback, {"error": error})
         self.ws.keep_running = False
         self.connection_lost[str].emit(
             f"code: {_close_status_code} | message {_close_message}"
@@ -302,12 +311,14 @@ def on_message(self, *args) -> None:
             else:
                 _callback = _entry[2] if len(_entry) > 2 else None
                 if _callback is not None:
-                    if "error" not in response:
-                        _callback(response.get("result"))
-                    else:
+                    if "error" in response:
                         logger.error(
-                            "WS request %s error: %s", _entry[0], response.get("error")
+                            "WS request %s error: %s", _entry[0], response["error"]
                         )
+                        _reply = {"error": response["error"]}
+                    else:
+                        _reply = response.get("result")
+                    self.request_reply_signal.emit(_callback, _reply)
                     return
                 if "error" in response:
                     message_event = WebSocketMessageReceived(
@@ -349,6 +360,14 @@ def on_message(self, *args) -> None:
                 "Unexpected error while creating websocket message event: %s", e
             )
 
+    @QtCore.pyqtSlot(object, object, name="run_callback")
+    def _run_callback(self, callback, reply) -> None:
+        """Run a request callback on the GUI thread; a raise must not kill the app."""
+        try:
+            callback(reply)
+        except Exception:
+            logger.exception("WS request callback %r failed", callback)
+
     def send_request(
         self, method: str, params: dict | None = None, callback=None
     ) -> bool:
@@ -357,11 +376,14 @@ def send_request(
         Args:
             method (str): Websocket method name
             params (dict, optional): parameters for the websocket method. Defaults to None.
-            callback (callable, optional): Called with ``response["result"]`` when the
-            response arrives. If None, the response is routed as a
+            callback (callable, optional): Called once on the GUI thread with
+            ``response["result"]``, or with ``{"error": {"code": int, "message": str}}``
+            on a JSON-RPC error or a disconnect (code -1); check for ``"error"``
+            before reading the result. If None, the response is routed as a
             ``WebSocketMessageReceived`` event to the parent widget. Defaults to None.
         Returns:
-            bool: Whether the method finished and a request was sent
+            bool: True if sent; the callback then runs exactly once. False if not
+            sent; the callback never runs, so the caller handles the failure.
         """
         if params is None:
             params = {}
@@ -382,7 +404,16 @@ def send_request(
             "params": params,
             "id": _rid,
         }
-        _ws.send(json.dumps(packet))
+        try:
+            _ws.send(json.dumps(packet))
+        except (websocket.WebSocketException, OSError) as e:
+            logger.warning("WS request %s not sent: %s", method, e)
+            with self._request_lock:
+                queued = self.request_table.pop(_rid, None) is not None
+            if queued or callback is None:
+                return False
+            # on_close already failed it; False would make the caller answer it twice.
+            return True
         return True
 
 
@@ -923,10 +954,11 @@ def set_spool_id(self, spool_id: int | None, callback=None) -> bool:
         )
 
     def get_spool(self, spool_id: int, callback) -> bool:
-        """Request spool data from Moonraker's Spoolman proxy"""
-        return self._ws.send_request(
-            method="server.spoolman.get_spool",
-            params={"spool_id": spool_id},
+        """Request one spool through the Spoolman proxy; there is no get_spool RPC."""
+        return self.spoolman_proxy(
+            request_method="GET",
+            path=f"/v1/spool/{spool_id}",
+            use_v2_response=False,
             callback=callback,
         )
 

BlocksScreen/lib/moonrest.py

@@ -77,19 +77,19 @@ def get_server_info(self):
         return self.get_request(method="server/info")
 
     def get_spool(self, spool_id: int) -> dict | None:
-        """Spoolman spool via Moonraker, None on error."""
-        response = self.get_request(f"server/spoolman/spool/{spool_id}")
+        """Spoolman spool via the Moonraker proxy, None on error."""
+        response = self.post_request(
+            "server/spoolman/proxy",
+            json={
+                "request_method": "GET",
+                "path": f"/v1/spool/{spool_id}",
+                "use_v2_response": False,
+            },
+        )
         if not isinstance(response, dict):
             return None
         return response.get("result")
 
-    def set_spool_used_weight(self, spool_id: int, weight: float) -> bool:
-        """Set a spool's used_weight; True on success."""
-        response = self.post_request(
-            f"server/spoolman/spool/{spool_id}", json={"used_weight": weight}
-        )
-        return response is not None
-
     def firmware_restart(self):
         """POST firmware_restart to Moonraker."""
         return self.post_request(method="printer/firmware_restart")

tests/amu/test_manager_unit.py

@@ -381,6 +381,24 @@ def test_no_emit_on_ws_failure(self, manager, qtbot) -> None:
         with qtbot.assertNotEmitted(manager.spool_fetched):
             callback(None)
 
+    @pytest.mark.parametrize(
+        "error",
+        [
+            {"code": 404, "message": "spool not found"},
+            {"code": -1, "message": "disconnected"},
+        ],
+    )
+    def test_error_reply_leaves_the_gate_alone(self, manager, qtbot, error) -> None:
+        """An error read as spool data would send MATERIAL= COLOR= SPOOLID=-1."""
+        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
+        manager.fetch_spool(0, 42)
+        callback = manager._ws.api.get_spool.call_args.args[1]
+        with (
+            qtbot.assertNotEmitted(manager.spool_fetched),
+            qtbot.assertNotEmitted(manager.run_gcode_signal),
+        ):
+            callback({"error": error})
+
 
 class TestLoadCellUpdate:
     def test_noop_when_mmu_state_none(self, manager) -> None:

tests/amu/test_moonrest_unit.py

@@ -19,35 +19,26 @@ def test_return_spool_dict_on_sucess(self, rest) -> None:
             "used_weight": 50.0,
             "remaining_weight": 200.0,
         }
-        with patch.object(rest, "get_request", return_value={"result": spool_data}):
+        with patch.object(rest, "post_request", return_value={"result": spool_data}):
             assert rest.get_spool(42) == spool_data
 
     def test_returns_non_on_http_error(self, rest) -> None:
-        with patch.object(rest, "get_request", return_value=None):
+        with patch.object(rest, "post_request", return_value=None):
             assert rest.get_spool(42) is None
 
     def test_returns_none_on_missing_result_key(self, rest):
-        with patch.object(rest, "get_request", return_value={"something": "else"}):
+        with patch.object(rest, "post_request", return_value={"something": "else"}):
             assert rest.get_spool(42) is None
 
-    def test_calls_correct_endpoint(self, rest) -> None:
-        with patch.object(rest, "get_request", return_value=None) as mock_get:
-            rest.get_spool(7)
-        mock_get.assert_called_once_with("server/spoolman/spool/7")
-
-
-class TestSetSpoolUsedWeight:
-    def test_returns_true_on_sucess(self, rest) -> None:
-        with patch.object(rest, "post_request", return_value={"result": "ok"}):
-            assert rest.set_spool_used_weight(42, 75.5)
-
-    def test_returns_false_on_error(self, rest) -> None:
-        with patch.object(rest, "post_request", return_value={"result": "ok"}):
-            assert rest.set_spool_used_weight(42, 75.5)
-
-    def test_calls_correct_endpoint(self, rest) -> None:
+    # /server/spoolman/spool/{id} is a 404; only the proxy route exists.
+    def test_goes_through_the_proxy(self, rest) -> None:
         with patch.object(rest, "post_request", return_value=None) as mock_post:
-            rest.set_spool_used_weight(5, 100.0)
+            rest.get_spool(7)
         mock_post.assert_called_once_with(
-            "server/spoolman/spool/5", json={"used_weight": 100.0}
+            "server/spoolman/proxy",
+            json={
+                "request_method": "GET",
+                "path": "/v1/spool/7",
+                "use_v2_response": False,
+            },
         )

tests/lib/test_moonraker_comm_unit.py

@@ -1,14 +1,15 @@
 """Unit tests for moonrakerComm."""
 
 import json
+import threading
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import pytest
 from PyQt6 import QtCore
 
 from BlocksScreen.lib import moonrakerComm
-from BlocksScreen.lib.moonrakerComm import OneShotTokenError
+from BlocksScreen.lib.moonrakerComm import MoonAPI, OneShotTokenError
 
 
 class TestOneShotTokenError:
@@ -120,6 +121,42 @@ def test_reconnect_reemits_unchanged_state(self, ws):
         assert states == ["ready", "ready"]
 
 
+class TestRequestCallbacks:
+    def _reply(self, sock, callback, payload: dict) -> None:
+        sock.request_table[9] = ["server.spoolman.proxy", {}, callback]
+        sock.on_message(None, json.dumps({"jsonrpc": "2.0", "id": 9, **payload}))
+
+    def test_result_reaches_callback(self, ws):
+        got = []
+        self._reply(ws, got.append, {"result": {"response": [1], "error": None}})
+        assert got == [{"response": [1], "error": None}]
+
+    def test_jsonrpc_error_reaches_callback(self, ws):
+        got = []
+        err = {"code": 503, "message": "Spoolman server not available"}
+        self._reply(ws, got.append, {"error": err})
+        assert got == [{"error": err}]
+
+    def test_raising_callback_is_logged_not_raised(self, ws, caplog):
+        def boom(_result):
+            raise RuntimeError("boom")
+
+        self._reply(ws, boom, {"result": {}})
+        assert "callback" in caplog.text
+
+    def test_callback_runs_on_gui_thread(self, ws, qtbot):
+        threads = []
+
+        def record(_result):
+            threads.append(threading.current_thread())
+
+        worker = threading.Thread(target=self._reply, args=(ws, record, {"result": {}}))
+        worker.start()
+        worker.join()
+        qtbot.waitUntil(lambda: bool(threads))
+        assert threads == [threading.main_thread()]
+
+
 class TestReconnectCycle:
     def test_retries_forever_until_connected(self, ws, monkeypatch):
         connect = MagicMock(return_value=False)
@@ -192,3 +229,73 @@ def test_close_leaves_reconnect_to_watchdog(self, ws, monkeypatch, qtbot):
         ws.on_close(None, 1006, "gone")
         qtbot.wait(50)
         connect.assert_not_called()
+
+    def test_close_fails_pending_requests(self, ws, qtbot):
+        ws.ws = MagicMock()
+        got = []
+        ws.request_table[3] = ["server.info", {}, None]
+        ws.request_table[4] = ["server.spoolman.proxy", {}, got.append]
+        ws.on_close(None, 1006, "gone")
+        assert ws.request_table == {}
+        qtbot.waitUntil(lambda: bool(got))
+        assert got == [{"error": {"code": -1, "message": "disconnected"}}]
+
+    def test_failed_send_returns_false_and_forgets_request(self, ws, monkeypatch):
+        # conftest stubs websocket; an except clause needs a real class
+        monkeypatch.setattr(moonrakerComm.websocket, "WebSocketException", OSError)
+        ws.connected = True
+        ws.ws = MagicMock()
+        ws.ws.send.side_effect = OSError("broken pipe")
+        assert ws.send_request("server.info", callback=MagicMock()) is False
+        assert ws.request_table == {}
+
+    def test_send_racing_close_answers_the_callback_once(self, ws, qtbot, monkeypatch):
+        """on_close already failed the callback, so True keeps the caller from a 2nd."""
+        monkeypatch.setattr(moonrakerComm.websocket, "WebSocketException", OSError)
+        ws.connected = True
+        ws.ws = MagicMock()
+
+        def close_then_fail(_packet):
+            ws.on_close(None, 1006, "gone")
+            raise OSError("broken pipe")
+
+        ws.ws.send.side_effect = close_then_fail
+        got = []
+        assert ws.send_request("server.info", callback=got.append) is True
+        qtbot.waitUntil(lambda: bool(got))
+        qtbot.wait(20)
+        assert got == [{"error": {"code": -1, "message": "disconnected"}}]
+
+    def test_send_while_disconnected_never_calls_back(self, ws, qtbot):
+        callback = MagicMock()
+        assert ws.send_request("server.info", callback=callback) is False
+        assert ws.request_table == {}
+        qtbot.wait(20)
+        callback.assert_not_called()
+
+
+@pytest.fixture
+def api():
+    """MoonAPI over a stub socket; it parents itself to ws, so ws must be a QObject."""
+    sock = QtCore.QObject()
+    sock.send_request = MagicMock(return_value=True)
+    return MoonAPI(sock)
+
+
+class TestGetSpool:
+    def test_goes_through_the_proxy(self, api) -> None:
+        api.get_spool(42, None)
+        params = api._ws.send_request.call_args.kwargs
+        assert params["method"] == "server.spoolman.proxy"
+        assert params["params"] == {
+            "use_v2_response": False,
+            "request_method": "GET",
+            "path": "/v1/spool/42",
+        }
+
+    def test_passes_the_callback_through(self, api) -> None:
+        def cb(_result):
+            return None
+
+        api.get_spool(1, cb)
+        assert api._ws.send_request.call_args.kwargs["callback"] is cb