From 59e1bec124238a28c8c12bdeaada88d898aa0f41 Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sun, 2 Aug 2026 21:46:50 +0200 Subject: [PATCH 1/4] feat(status): make Restart Machine ask the printer instead of re-reading memory The retained status is what the agent last *heard*, so re-reading it on a manual refresh returned the same thing the page was already showing. Pressing the button could not tell you anything you did not already know. It now publishes `probe` on the cmd topic and waits for the agent to go and look. That topic already existed for `flush`, the agent already subscribes to it, and the broker ACL already grants inventree write on it -- so no new topic, no ACL change, no new credential. "Restart Machine" is the refresh button whether or not it is called that: InvenTree's machine framework has no generic custom-action mechanism. restart_machine is the one hardcoded action, with its own endpoint and a fixed place in the menu, so it is the only lever a driver gets and it was being wasted. probe_status subscribes before publishing, because the answer is a retained message and a subscription set up afterwards would race the agent. The message already on the topic arrives immediately and is kept as the baseline, so a probe the printer cannot answer -- the normal case for a D30 that has powered itself down -- degrades to exactly what read_status would have returned. The page keeps the old reading and its age instead of going blank. The automatic paths keep reading rather than probing. Neither startup nor the periodic ping should dial a sleeping printer on a schedule nobody asked for. --- README.md | 23 +++++-- inventree_label_dispatch/dispatch.py | 54 +++++++++++++++++ inventree_label_dispatch/driver.py | 36 +++++++++-- tests/test_dispatch.py | 91 ++++++++++++++++++++++++++++ 4 files changed, 194 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index c9103ec..9fbf5df 100644 --- a/README.md +++ b/README.md @@ -33,11 +33,24 @@ InvenTree imports so it is unit-tested here: The status is read from the retained topic in three places, which between them keep the page honest without polling the printer itself: -| when | hook | -|---|---| -| machine initialisation | `init_machine` | -| a manual restart from the Admin Center | `restart_machine` | -| periodically | `ping_machines`, driven by InvenTree's `MACHINE_PING_ENABLED` (on by default) | +| when | hook | asks the printer? | +|---|---|---| +| machine initialisation | `init_machine` | no — reads the retained topic | +| **Restart Machine** in the Admin Center | `restart_machine` | **yes** — publishes `probe` on the `cmd` topic | +| periodically | `ping_machines`, driven by InvenTree's `MACHINE_PING_ENABLED` (on by default) | no | + +**"Restart Machine" is the refresh button.** InvenTree's machine framework has no generic +custom-action mechanism — `restart_machine` is the one hardcoded action, with its own +endpoint and a fixed place in the menu — so that item is the only lever a driver gets. +It now sends `probe` and the agent goes and looks at the printer, rather than re-reading +the retained topic the page was already showing. The `cmd` topic already existed for +`flush`, and the broker ACL already grants `inventree` write on it, so this needed no +new topic and no credential change. + +The automatic paths deliberately do *not* probe: neither startup nor the periodic ping +should dial a sleeping printer on a schedule nobody asked for. And a probe the printer +cannot answer falls back to remembered truth, so the page keeps the old reading and its +age rather than going blank. The periodic one is not a nicety. Machine status lives in the Django cache, so clearing that cache drops every `machine:*` key at once — and this deployment runs Redis without diff --git a/inventree_label_dispatch/dispatch.py b/inventree_label_dispatch/dispatch.py index 4e0c236..aa59dd1 100644 --- a/inventree_label_dispatch/dispatch.py +++ b/inventree_label_dispatch/dispatch.py @@ -12,6 +12,7 @@ from __future__ import annotations import json +import time import uuid from dataclasses import dataclass @@ -147,6 +148,59 @@ def _on_message(_c, _u, msg): client.disconnect() +def probe_status(printer_id: str, conn: Connection, wait_s: float = 8.0) -> dict | None: + """Ask the agent to go and look at the printer, and return what it then publishes. + + ``read_status`` returns what the agent *remembers*, which is the right default: + the D30 is only reachable while it is being printed to, so remembered truth is + almost always the best truth available and costs nothing. This is the other case -- + a human pressed refresh and wants the printer asked, not the memory read. + + Subscribes *before* publishing the command, because the answer is a retained + message and a subscription set up afterwards would race the agent. The retained + message already on the topic arrives immediately and is kept as the baseline, so a + probe the agent cannot fulfil (the printer is asleep, which is normal) degrades to + exactly what ``read_status`` would have returned rather than to nothing. + + ``wait_s`` covers the agent's whole round trip -- connect, session setup, telemetry, + publish -- which is a few seconds when the printer answers and one connect timeout + when it does not. This blocks the request that triggered it, so it is bounded + rather than generous. + """ + import queue + + received: queue.Queue = queue.Queue() + client = _new_client(conn, "probe") + client.on_message = lambda _c, _u, msg: received.put_nowait(msg.payload) + client.connect(conn.host, conn.port, conn.keepalive_s) + client.subscribe(conn.topic(printer_id, "status"), qos=1) + client.loop_start() + try: + latest = None + try: # the retained message: what we already knew, before asking + latest = json.loads(received.get(timeout=conn.timeout_s)) + except (queue.Empty, ValueError): + pass + + info = client.publish(conn.topic(printer_id, "cmd"), "probe", qos=1) + info.wait_for_publish(conn.timeout_s) + + deadline = time.monotonic() + wait_s + while time.monotonic() < deadline: + try: + payload = received.get(timeout=max(0.1, deadline - time.monotonic())) + except queue.Empty: + break + try: + return json.loads(payload) + except ValueError: + continue + return latest + finally: + client.loop_stop() + client.disconnect() + + def read_status(printer_id: str, conn: Connection) -> dict | None: """One-shot read of the retained status topic. ``None`` if nothing is retained. diff --git a/inventree_label_dispatch/driver.py b/inventree_label_dispatch/driver.py index 00305f5..ea99504 100644 --- a/inventree_label_dispatch/driver.py +++ b/inventree_label_dispatch/driver.py @@ -30,7 +30,14 @@ class only reads settings, turns items into label specs, and publishes. from rest_framework import serializers from .config import connection_from_env -from .dispatch import build_job, build_label, publish, publish_awaiting_result, read_status +from .dispatch import ( + build_job, + build_label, + probe_status, + publish, + publish_awaiting_result, + read_status, +) from .extractors import extract from .status import classify_result, classify_status @@ -111,15 +118,27 @@ def _apply(self, machine, member: str, text: str) -> None: machine.set_status_text(text) def restart_machine(self, machine) -> None: - """Re-read the printer's state after a manual restart. + """Ask the printer what it is, and show the answer. The page's refresh button. A restart clears the machine's shared state, and without this nothing puts it back: the page then shows UNKNOWN with an empty status text -- the defaults `get_shared_state` returns for missing keys -- while the machine itself reports `initialized`, no errors and a driver available. An empty status text is diagnostic on its own, since no path in this driver produces one. + + This is also the only lever there is. InvenTree's machine framework has no + generic custom-action mechanism -- `restart_machine` is the one hardcoded + action, with its own endpoint and a fixed place in the Machines menu -- so + "Restart Machine" *is* the refresh button whether or not it is called that. + Making it re-read the retained topic and nothing else wasted it: the topic + holds what the agent last heard, which is the same thing the page was already + showing. Now it sends `probe` and the agent goes and looks. + + Falls back to remembered truth when the printer does not answer, which is the + normal case for a D30 that has powered itself down: the reading keeps its old + `device_seen_at` and the page says how old it is, instead of going blank. """ - self._refresh_status(machine) + self._refresh_status(machine, probe=True) def ping_machines(self) -> None: """Periodic refresh, driven by InvenTree's MACHINE_PING_ENABLED (default on). @@ -141,10 +160,17 @@ def ping_machines(self) -> None: except Exception as exc: # noqa: BLE001 - one machine must not break the loop machine.handle_error(exc) - def _refresh_status(self, machine) -> None: + def _refresh_status(self, machine, *, probe: bool = False) -> None: + """Read the retained status, or ask the printer for a fresh one. + + ``probe`` is off for the automatic paths -- init and the periodic ping -- so + neither of them dials a sleeping printer on a schedule nobody asked for. It is + on only where a human pressed the button. + """ printer_id = machine.get_setting("PRINTER_ID", "D") try: - status = read_status(printer_id, connection_from_env()) + conn = connection_from_env() + status = probe_status(printer_id, conn) if probe else read_status(printer_id, conn) except Exception as exc: machine.set_status(LabelPrinterStatus.UNKNOWN) machine.set_status_text(_("could not reach the broker: %(err)s") % {"err": str(exc)[:80]}) diff --git a/tests/test_dispatch.py b/tests/test_dispatch.py index 1cbfc28..add8cf2 100644 --- a/tests/test_dispatch.py +++ b/tests/test_dispatch.py @@ -155,3 +155,94 @@ def publish_with_garbage(topic, payload, qos=0): fake.publish = publish_with_garbage result = dispatch.publish_awaiting_result(job, Connection(host="broker"), timeout_s=1.0) assert result["state"] == "completed" + + +# -- refresh: ask the printer, not the memory ------------------------------- # +# +# read_status returns what the agent remembers, which is the right default. probe_status +# is the other case: a human pressed refresh and wants the printer actually asked. + + +class _ProbeBroker(_FakeClient): + """Retains one status, and optionally publishes a fresh one when probed.""" + + def __init__(self, retained, fresh=None): + super().__init__() + self.retained, self.fresh = retained, fresh + self.subscribed = [] + self.on_message = None + self.order = [] + + def subscribe(self, topic, qos=0): + self.subscribed.append(topic) + self.order.append("subscribe") + if self.retained is not None: # retained messages arrive on subscribe + self._deliver(self.retained) + + def publish(self, topic, payload, qos=0): + self.order.append("publish") + info = super().publish(topic, payload, qos) + if self.fresh is not None: + self._deliver(self.fresh) + return info + + def _deliver(self, obj): + payload = obj if isinstance(obj, bytes) else json.dumps(obj).encode() + self.on_message(self, None, type("Msg", (), {"payload": payload})()) + + +def test_probe_asks_the_agent_and_returns_the_fresh_answer(monkeypatch): + fake = _ProbeBroker( + retained={"state": "idle", "media_ok": True, "device_seen_at": "2026-08-01T00:00:00Z"}, + fresh={"state": "idle", "media_ok": True, "device_seen_at": "2026-08-02T12:00:00Z"}, + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + + assert got["device_seen_at"] == "2026-08-02T12:00:00Z" + topic, payload, qos = fake.published[0] + assert topic == "se/v1/print/d30-workshop/cmd" + assert payload == "probe" + assert qos == 1 + + +def test_probe_subscribes_before_asking(monkeypatch): + """The answer is a retained publish; subscribing afterwards races the agent.""" + fake = _ProbeBroker(retained={"state": "idle"}, fresh={"state": "idle"}) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + + assert fake.order[:2] == ["subscribe", "publish"] + + +def test_a_printer_that_does_not_answer_falls_back_to_what_we_knew(monkeypatch): + """The normal case for a D30 that has powered itself down. The reading keeps its + old timestamp and the page says how old it is, rather than going blank.""" + fake = _ProbeBroker( + retained={"state": "idle", "media_ok": True, "device_seen_at": "2026-08-01T00:00:00Z"}, + fresh=None, # agent probed, printer was asleep, so it published nothing + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=0.3) + + assert got["device_seen_at"] == "2026-08-01T00:00:00Z" + + +def test_probe_with_nothing_retained_and_no_answer_is_none(monkeypatch): + fake = _ProbeBroker(retained=None, fresh=None) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + conn = Connection(host="broker", timeout_s=0.3) + + assert dispatch.probe_status("d30-workshop", conn, wait_s=0.3) is None + + +def test_probe_survives_a_malformed_retained_payload(monkeypatch): + fake = _ProbeBroker(retained=b"{not json", fresh={"state": "idle", "media_ok": True}) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + conn = Connection(host="broker", timeout_s=0.3) + + got = dispatch.probe_status("d30-workshop", conn, wait_s=1) + assert got == {"state": "idle", "media_ok": True} From e77e9790b48c6a0c15e9b95b2bb43dc278e393aa Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sun, 2 Aug 2026 21:51:42 +0200 Subject: [PATCH 2/4] docs(status): spell out which schedule 'Enable Machine Ping' actually governs It governs everything in this driver: InvenTree's task short-circuits on the setting before any driver is called, so switching it off stops ping_machines entirely. There is nothing to add here for that. What it does not govern is the print agent. device.probe_interval_s is the agent's own timer and it is the one that contacts the printer, so turning off Machine Ping stops this page reading without stopping the agent asking. That is the right layering -- the retained topic is a shared bus and InvenTree is one consumer of it, so a setting here reaching across to reconfigure the device agent would be the more surprising behaviour -- but it is not guessable from either side, hence writing it down on both. Also notes the 5-minute ping cadence, which is the sensible floor for probe_interval_s. --- README.md | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 9fbf5df..11057ce 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,20 @@ the retained topic the page was already showing. The `cmd` topic already existed new topic and no credential change. The automatic paths deliberately do *not* probe: neither startup nor the periodic ping -should dial a sleeping printer on a schedule nobody asked for. And a probe the printer +should dial a sleeping printer on a schedule nobody asked for. + +**Two schedules, and only one of them is yours.** `MACHINE_PING_ENABLED` ("Enable Machine +Ping", on by default, every 5 minutes) governs everything here — InvenTree's task +short-circuits before any driver is called, so switching it off stops `ping_machines` +entirely. It does *not* reach the print agent: `device.probe_interval_s` on the agent is +its own timer, and that is the one that contacts the printer. Turning off Machine Ping +stops this page **reading**; it does not stop the agent **asking**. Deliberate — the +retained topic is a shared bus and InvenTree is one consumer of it, so a setting here +silently reconfiguring the device agent would be the more surprising behaviour. + +Nothing is lost by leaving the ping on: it reads a retained message and never touches +Bluetooth. And there is little point setting `probe_interval_s` below 5 minutes, since +that is as often as this page re-reads. And a probe the printer cannot answer falls back to remembered truth, so the page keeps the old reading and its age rather than going blank. From e43542c53af607851e6b67ef293f1a9c36222bd6 Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sun, 2 Aug 2026 22:05:00 +0200 Subject: [PATCH 3/4] fix(status): return the probe's answer, not whatever status arrives first probe_status took the first status published after sending the command. The agent publishes on its own events too -- a print starting and finishing, a broker reconnect, another operator pressing the same button -- so any of those landing first was returned as the answer while the real one was read and thrown away. The realistic trigger is a print in flight when someone clicks refresh: the page shows `state=printing` on a status it just asked to have refreshed, until the next ping corrects it. Mild and self-correcting, which is exactly why nobody would report it -- `printing` on a printer that is printing looks right. It now waits for a status whose device_seen_at is later than the one already retained. A print that actually reached the printer also carries a newer observation and is accepted, correctly: that is the same freshness a probe would have produced, obtained the same way. An identical republish, or a state change with an unchanged observation, is not. The fallback improves as a side effect: an unanswered probe now reports the newest status seen while waiting rather than the one that happened to be retained when we subscribed. Comparing parsed instants rather than raw strings, and parse_seen_at moves out of _ago so both callers share it. The strings sort chronologically only while every producer emits whole seconds -- "...:42.5Z" sorts before "...:42Z" because "." precedes "Z" -- so a sub-second stamp would have been discarded as older. Pinned by a test. The docstring promised "returns what it then publishes" while the code guaranteed "first status after the command". Now they agree. --- inventree_label_dispatch/dispatch.py | 34 ++++++++++- inventree_label_dispatch/status.py | 30 +++++++--- tests/test_dispatch.py | 86 ++++++++++++++++++++++++++-- 3 files changed, 135 insertions(+), 15 deletions(-) diff --git a/inventree_label_dispatch/dispatch.py b/inventree_label_dispatch/dispatch.py index aa59dd1..e79a676 100644 --- a/inventree_label_dispatch/dispatch.py +++ b/inventree_label_dispatch/dispatch.py @@ -16,6 +16,8 @@ import uuid from dataclasses import dataclass +from .status import parse_seen_at + #: Mirrors labelfab's contract v1. Kept as a plain dict builder rather than importing #: labelfab so the InvenTree image does not pull in Pillow/qrcode a second time. CONTRACT_VERSION = 1 @@ -149,18 +151,25 @@ def _on_message(_c, _u, msg): def probe_status(printer_id: str, conn: Connection, wait_s: float = 8.0) -> dict | None: - """Ask the agent to go and look at the printer, and return what it then publishes. + """Ask the agent to look at the printer, and return the first *newer* reading. ``read_status`` returns what the agent *remembers*, which is the right default: the D30 is only reachable while it is being printed to, so remembered truth is almost always the best truth available and costs nothing. This is the other case -- a human pressed refresh and wants the printer asked, not the memory read. + "Newer" is load-bearing and means a ``device_seen_at`` later than the one already + retained, not merely the next message to arrive. The agent publishes on its own + events too, so without that test a print starting, a broker reconnect or somebody + else pressing the same button could be returned as the answer while the real one + was thrown away. + Subscribes *before* publishing the command, because the answer is a retained message and a subscription set up afterwards would race the agent. The retained message already on the topic arrives immediately and is kept as the baseline, so a probe the agent cannot fulfil (the printer is asleep, which is normal) degrades to - exactly what ``read_status`` would have returned rather than to nothing. + exactly what ``read_status`` would have returned rather than to nothing -- or to + something newer, if one of those other events arrived while we waited. ``wait_s`` covers the agent's whole round trip -- connect, session setup, telemetry, publish -- which is a few seconds when the printer answers and one connect timeout @@ -185,6 +194,18 @@ def probe_status(printer_id: str, conn: Connection, wait_s: float = 8.0) -> dict info = client.publish(conn.topic(printer_id, "cmd"), "probe", qos=1) info.wait_for_publish(conn.timeout_s) + # The answer is the first status carrying an observation *newer* than the one we + # already had -- not simply the first status to arrive. The agent publishes on + # its own events too (a print starting and finishing, a broker reconnect, + # somebody else pressing this button), and any of those landing first would + # otherwise be mistaken for the answer while the real one was discarded. The + # symptom is mild and self-correcting, which is exactly why it would have been + # hard to spot: `state=printing` on a page you just refreshed looks plausible. + # + # A status published by a print that *did* reach the printer also carries a + # newer observation, and returning that is right -- it is the same freshness a + # probe would have produced, obtained the same way. + baseline = parse_seen_at((latest or {}).get("device_seen_at")) deadline = time.monotonic() + wait_s while time.monotonic() < deadline: try: @@ -192,9 +213,16 @@ def probe_status(printer_id: str, conn: Connection, wait_s: float = 8.0) -> dict except queue.Empty: break try: - return json.loads(payload) + message = json.loads(payload) except ValueError: continue + seen = parse_seen_at(message.get("device_seen_at")) + if seen is not None and (baseline is None or seen > baseline): + return message + # Not the answer, but still more current than what we started with: keep it + # as the fallback, so a probe that goes unanswered reports the newest state + # rather than the one that happened to be retained when we subscribed. + latest = message return latest finally: client.loop_stop() diff --git a/inventree_label_dispatch/status.py b/inventree_label_dispatch/status.py index e3d6510..826538e 100644 --- a/inventree_label_dispatch/status.py +++ b/inventree_label_dispatch/status.py @@ -29,31 +29,45 @@ _GOOD_JOB_STATES = frozenset({"completed"}) -def _ago(seen: object, now: datetime) -> str: - """How long ago the device fields were true, as a short suffix. ``""`` if unknown. +def parse_seen_at(seen: object) -> datetime | None: + """``device_seen_at`` as an instant, or ``None`` if it is missing or unusable. ``seen`` is typed as ``object`` rather than ``str | None`` on purpose: it comes straight out of ``json.loads`` on a payload this process did not produce, so it can be any JSON type at all, and pretending otherwise would just move the lie into the annotation. - Empty rather than a guess in three cases, all of which mean the same thing: nobody - said. An agent older than the field simply omits it, and the page should look - exactly as it did before rather than sprouting a fake age. + ``None`` in three cases, all of which mean the same thing: nobody said. An agent + older than the field simply omits it. Nothing in here may raise. It runs inside a Django worker on a payload from another process, so a surprise is a 500 on the Machines page, not a test failure -- which is also why the ``Z`` is handled by hand: pydantic emits it and ``fromisoformat`` did not accept it until Python 3.11, while this package supports 3.9. + + Shared with :mod:`dispatch`, which compares two of these rather than rendering one. + Comparing the parsed instants and not the raw strings matters: they sort the same + way only while every producer emits whole seconds, and ``...:42.5Z`` sorts *before* + ``...:42Z`` because ``.`` precedes ``Z``. """ if not seen: - return "" + return None try: stamp = datetime.fromisoformat(str(seen).replace("Z", "+00:00")) except (TypeError, ValueError): + return None + return stamp if stamp.tzinfo is not None else stamp.replace(tzinfo=timezone.utc) + + +def _ago(seen: object, now: datetime) -> str: + """How long ago the device fields were true, as a short suffix. ``""`` if unknown. + + Empty rather than a guess when the agent did not say, so a page served by an agent + older than the field looks exactly as it did before rather than sprouting a fake age. + """ + stamp = parse_seen_at(seen) + if stamp is None: return "" - if stamp.tzinfo is None: - stamp = stamp.replace(tzinfo=timezone.utc) # Clamped at zero: the agent's clock and this one are not the same clock, and a # small skew must read as "just now", never as a negative age. delta = max(0, int((now - stamp).total_seconds())) diff --git a/tests/test_dispatch.py b/tests/test_dispatch.py index add8cf2..cce9b5e 100644 --- a/tests/test_dispatch.py +++ b/tests/test_dispatch.py @@ -164,11 +164,17 @@ def publish_with_garbage(topic, payload, qos=0): class _ProbeBroker(_FakeClient): - """Retains one status, and optionally publishes a fresh one when probed.""" + """Retains one status, and optionally publishes more when probed. + + ``fresh`` may be a single status or a list, so a test can reproduce the agent + publishing on its own events -- a print starting, a reconnect -- while a probe is + still in flight. + """ def __init__(self, retained, fresh=None): super().__init__() - self.retained, self.fresh = retained, fresh + self.retained = retained + self.fresh = [] if fresh is None else (fresh if isinstance(fresh, list) else [fresh]) self.subscribed = [] self.on_message = None self.order = [] @@ -182,8 +188,8 @@ def subscribe(self, topic, qos=0): def publish(self, topic, payload, qos=0): self.order.append("publish") info = super().publish(topic, payload, qos) - if self.fresh is not None: - self._deliver(self.fresh) + for message in self.fresh: + self._deliver(message) return info def _deliver(self, obj): @@ -246,3 +252,75 @@ def test_probe_survives_a_malformed_retained_payload(monkeypatch): got = dispatch.probe_status("d30-workshop", conn, wait_s=1) assert got == {"state": "idle", "media_ok": True} + + +def test_a_print_starting_mid_probe_does_not_win(monkeypatch): + """The agent publishes on its own events -- a print starting, a broker reconnect, + another click of the same button. Any of those landing first would otherwise be + returned while the real answer was discarded, and `state=printing` on a page you + just refreshed looks plausible enough that nobody would report it.""" + old = "2026-08-01T00:00:00Z" + fake = _ProbeBroker( + retained={"state": "idle", "media_ok": True, "device_seen_at": old}, + fresh=[ + # the in-flight print announcing itself: same observation, new state + {"state": "printing", "pending_labels": 3, "device_seen_at": old}, + # the probe's actual answer + {"state": "idle", "media_ok": True, "device_seen_at": "2026-08-02T12:00:00Z"}, + ], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + + assert got["state"] == "idle" + assert got["device_seen_at"] == "2026-08-02T12:00:00Z" + + +def test_a_print_that_reached_the_printer_is_a_valid_answer(monkeypatch): + """It carries a newer observation obtained the same way a probe would have.""" + fake = _ProbeBroker( + retained={"state": "idle", "media_ok": True, "device_seen_at": "2026-08-01T00:00:00Z"}, + fresh=[{"state": "idle", "media_ok": False, "device_seen_at": "2026-08-02T09:00:00Z"}], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + assert got["media_ok"] is False + + +def test_an_unanswered_probe_reports_the_newest_state_not_the_stale_one(monkeypatch): + """If the printer never answers but the agent said something else meanwhile, that + is still more current than what happened to be retained when we subscribed.""" + old = "2026-08-01T00:00:00Z" + fake = _ProbeBroker( + retained={"state": "idle", "device_seen_at": old}, + fresh=[{"state": "printing", "pending_labels": 2, "device_seen_at": old}], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=0.3) + assert got["state"] == "printing" + + +def test_an_identical_republish_is_not_an_answer(monkeypatch): + """A broker reconnect republishes the same retained status verbatim.""" + same = {"state": "idle", "media_ok": True, "device_seen_at": "2026-08-01T00:00:00Z"} + fake = _ProbeBroker(retained=same, fresh=[dict(same)]) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=0.3) + assert got == same # fell through to the fallback rather than claiming a fresh look + + +def test_sub_second_stamps_compare_chronologically_not_lexically(monkeypatch): + """'...:42.5Z' sorts before '...:42Z' as a string, because '.' precedes 'Z'. Parsing + is what keeps a newer reading from being discarded as older.""" + fake = _ProbeBroker( + retained={"state": "idle", "device_seen_at": "2026-08-02T12:00:42Z"}, + fresh=[{"state": "idle", "media_ok": True, "device_seen_at": "2026-08-02T12:00:42.5Z"}], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + assert got["media_ok"] is True From db6c925ac62ba6e3840bd7409df12a098cec4462 Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sun, 2 Aug 2026 22:41:09 +0200 Subject: [PATCH 4/4] fix(status): base the probe's answer on what we knew, not on the agent's emission rules Review pointed out that when nothing has ever been observed the baseline is None, and the newness filter weakens to "the first message carrying any timestamp at all". That is safe today only because a pre-observation agent publishes null everywhere -- an invariant that lives in the other repo and could be relaxed there without anyone here noticing. Everything queued before the command goes out is now folded into the baseline: the retained message plus whatever the agent happened to publish while we were connecting. "Newer than what we knew" becomes a property of this function rather than a bet on how labelfab emits. Drained before publishing rather than after, which matters more than it looks. Draining afterwards would let a quick answer be swallowed into the baseline it is supposed to beat -- still returned, via the fallback, but only after blocking for the full wait. The test suite noticed: it ran 5s slower with the drain in the wrong place. Also corrects the wait bound in the docstring. wait_s covers the agent's round trip, but reading the retained message and waiting for the command's PUBACK can each cost conn.timeout_s first, so anyone sizing a request timeout wants 2*conn.timeout_s + wait_s. Sub-millisecond against a healthy broker, and a broker slow enough for it to matter has already broken read_status everywhere else. --- inventree_label_dispatch/dispatch.py | 32 +++++++++++++++++++++++--- tests/test_dispatch.py | 34 ++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/inventree_label_dispatch/dispatch.py b/inventree_label_dispatch/dispatch.py index e79a676..217fe7c 100644 --- a/inventree_label_dispatch/dispatch.py +++ b/inventree_label_dispatch/dispatch.py @@ -171,10 +171,16 @@ def probe_status(printer_id: str, conn: Connection, wait_s: float = 8.0) -> dict exactly what ``read_status`` would have returned rather than to nothing -- or to something newer, if one of those other events arrived while we waited. - ``wait_s`` covers the agent's whole round trip -- connect, session setup, telemetry, + ``wait_s`` covers the agent's round trip -- connect, session setup, telemetry, publish -- which is a few seconds when the printer answers and one connect timeout - when it does not. This blocks the request that triggered it, so it is bounded - rather than generous. + when it does not. + + It is not the whole bound, though, and anyone sizing a request timeout around this + should use ``2 * conn.timeout_s + wait_s``: reading the retained message and waiting + for the command's PUBACK can each cost ``conn.timeout_s`` first. Both are sub- + millisecond against a healthy broker, and a broker slow enough for them to matter + has already broken ``read_status`` on every other path. This blocks the request that + triggered it either way, so all three are bounded rather than generous. """ import queue @@ -191,6 +197,26 @@ def probe_status(printer_id: str, conn: Connection, wait_s: float = 8.0) -> dict except (queue.Empty, ValueError): pass + # Everything queued before the command goes out is context, not answer: the + # retained message plus anything the agent happened to publish while we were + # connecting. Folding it into the baseline is what makes "newer than what we + # knew" a property of this function rather than a bet on how the agent emits. + # Otherwise, when nothing has ever been observed and the baseline is None, the + # filter weakens to "the first message carrying any timestamp at all", and that + # is only safe because a pre-observation agent publishes null everywhere -- an + # invariant that lives in the other repo and could be relaxed there without + # anyone here noticing. + # + # Drained before publishing rather than after, so a quick answer cannot be + # swallowed into the baseline it is supposed to beat. + while True: + try: + latest = json.loads(received.get_nowait()) + except queue.Empty: + break + except ValueError: + continue + info = client.publish(conn.topic(printer_id, "cmd"), "probe", qos=1) info.wait_for_publish(conn.timeout_s) diff --git a/tests/test_dispatch.py b/tests/test_dispatch.py index cce9b5e..b6da0e7 100644 --- a/tests/test_dispatch.py +++ b/tests/test_dispatch.py @@ -324,3 +324,37 @@ def test_sub_second_stamps_compare_chronologically_not_lexically(monkeypatch): got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) assert got["media_ok"] is True + + +def test_a_status_published_before_the_command_is_context_not_answer(monkeypatch): + """Whatever the agent published while we were connecting is part of what we already + knew. Folding it into the baseline is what stops `probe_status` depending on the + agent's emission contract: without it, a never-observed printer weakens the filter + to "the first message carrying any timestamp at all".""" + + class _EarlyPublisher(_ProbeBroker): + def subscribe(self, topic, qos=0): + super().subscribe(topic, qos) + # arrives after the retained one but before we send the command + self._deliver({"state": "idle", "media_ok": True, "device_seen_at": "2026-08-02T10:00:00Z"}) + + fake = _EarlyPublisher( + retained={"state": "idle", "device_seen_at": None}, + fresh=[{"state": "idle", "media_ok": False, "device_seen_at": "2026-08-02T12:00:00Z"}], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + + # the 10:00 reading was already on the wire before we asked, so the answer is 12:00 + assert got["device_seen_at"] == "2026-08-02T12:00:00Z" + assert got["media_ok"] is False + + +def test_a_never_observed_printer_still_falls_back_rather_than_inventing(monkeypatch): + """Nothing retained, nothing observed, printer asleep: there is no answer to give.""" + fake = _ProbeBroker(retained={"state": "idle", "device_seen_at": None}, fresh=None) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=0.3) + assert got == {"state": "idle", "device_seen_at": None}