From 61eb3804ded56fd5dd2e8701f9c9b3113c707972 Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sun, 2 Aug 2026 16:21:00 +0200 Subject: [PATCH 1/3] fix(agent): stop reconnects and restarts erasing device truth from the status topic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The retained status carried nulls for serial, firmware, battery_pct, voltage_v and media_ok almost all of the time, so InvenTree's Machines page read "CONNECTED — media unreported" for a printer the agent had fully identified on its last job. Two causes, and the second is the one that actually bites. _on_connect published a *hardcoded* bare PrinterStatus, retained. That fires on every reconnect, not only the first connect, and paho reconnects silently -- the broker's Istio route caps at a 24h timeout and flaps besides -- so it ran several times a day. It was not merely uninformative, it was lossy: it overwrote a retained message that had been correct a moment earlier, and the printer is asleep between jobs and cannot be asked again. Evidence was an agent 25h up with NRestarts=0 and every device field null. The rarer cause is the one the issue was filed about: a fresh process knew nothing at all, because the device fields lived in six attributes on PrintWorker that were only ever filled inside a send. Both are the same missing idea -- nowhere to keep what the printer said. DeviceSnapshot is that place, persisted to the spool DB after every capture and read back at startup. The MQTT source now remembers the last retained status it published, seeded from the spool, and republishes *that* on connect. The will and the shutdown notice carry it too: losing the link is news about reachability, not grounds for forgetting the serial. Remembered truth published as though it were live is its own kind of lie, so PrinterStatus gains device_seen_at. An absolute UTC instant rather than an age, because the message is retained -- an age is computed once and then sits on the broker getting wronger. It only advances when the printer actually reported something, so a connection that answered nothing cannot launder three-day-old media state into looking current. device.probe_on_start (on by default) surveys the printer once at startup, so an agent that comes up next to an awake printer publishes live truth immediately instead of waiting for the next job. It cannot wake a sleeping unit -- AUTO_POWER_TIME really does power the radio down and only the button brings it back -- so a miss costs one connect timeout and the stored snapshot is published unchanged. It runs on the print loop and only at startup, never from on_connect: that callback is paho's network thread, and probing from it would touch the printer concurrently with a print, which is exactly what the single-threaded loop exists to prevent. Collecting the snapshot also removed the duplicated PrinterStatus construction between worker.py and source_mqtt.py, which is why the two could disagree about what the topic should say in the first place. tests/test_source_mqtt.py is new -- that module had no coverage, which is how a hardcoded blank status survived in the one callback that runs most often. The README still claimed "there is no read channel" and that low battery was undetectable. That was falsified by #9; corrected here. Closes #12 --- README.md | 17 ++- deploy/agent.toml.example | 4 + src/labelfab/agent/__init__.py | 2 + src/labelfab/agent/__main__.py | 5 + src/labelfab/agent/config.py | 7 + src/labelfab/agent/device_state.py | 123 ++++++++++++++++ src/labelfab/agent/source_mqtt.py | 56 ++++++-- src/labelfab/agent/spool.py | 50 +++++++ src/labelfab/agent/worker.py | 106 ++++++++------ src/labelfab/contract/models.py | 11 ++ tests/test_agent_config.py | 10 ++ tests/test_device_state.py | 117 +++++++++++++++ tests/test_source_mqtt.py | 219 +++++++++++++++++++++++++++++ tests/test_spool.py | 31 +++- tests/test_worker.py | 75 ++++++++++ 15 files changed, 772 insertions(+), 61 deletions(-) create mode 100644 src/labelfab/agent/device_state.py create mode 100644 tests/test_device_state.py create mode 100644 tests/test_source_mqtt.py diff --git a/README.md b/README.md index 307bd8c..4accd47 100644 --- a/README.md +++ b/README.md @@ -82,9 +82,20 @@ Protocol constants, tape geometry and the day-one bring-up checklist live in [`HARDWARE-NOTES.md`](HARDWARE-NOTES.md). Read it before changing anything in `labelfab.device`. -**There is no read channel.** The D30 never reports back, so "printed" means "the bytes -were accepted and we waited the physical print duration". Out-of-tape, jams and low -battery are undetectable. +**There is a read channel**, contrary to the received wisdom: the D30 answers queries on +both transports and pushes media-state changes unsolicited. The retained status topic +carries what it said — serial, firmware, battery percentage and terminal voltage, and +whether media is loaded — not just whether the agent is up. + +It says so *and when it said so*. The printer is only reachable while it is being printed +to (it auto-powers-off, and a held-open socket just relocates that), so the agent +remembers the last thing it heard and republishes it, stamped with `device_seen_at`. +A consumer can then tell live truth from remembered truth instead of guessing, and no +printer is woken to keep a status page tidy. `device.probe_on_start` surveys the printer +once at startup when it happens to be awake. + +Jams remain undetectable — the media bit distinguishes loaded from not, and nothing +observed so far distinguishes a jam from either. ## License diff --git a/deploy/agent.toml.example b/deploy/agent.toml.example index 92cde54..0acedb4 100644 --- a/deploy/agent.toml.example +++ b/deploy/agent.toml.example @@ -42,6 +42,10 @@ raster_width_px = 96 # 96 = 12mm head (verified); 120 = 15mm (day-1 hypo pace_factor = 1.2 # lower until a long strip garbles, then +50% density = 1 # 1 light | 2 medium | 4 heavy; raise if codes misscan wake_dummy_feed = false +# Survey the printer once at startup so the status topic carries live truth instead of +# what was remembered from the last print. Cannot wake a sleeping unit, so a miss costs +# one connect timeout and the stored status is published unchanged. +probe_on_start = true [tape] width_mm = 15.0 diff --git a/src/labelfab/agent/__init__.py b/src/labelfab/agent/__init__.py index 041c5dd..339f205 100644 --- a/src/labelfab/agent/__init__.py +++ b/src/labelfab/agent/__init__.py @@ -11,6 +11,7 @@ from labelfab.agent.coalescer import Batch, Coalescer, PendingLabel from labelfab.agent.config import Config, load +from labelfab.agent.device_state import DeviceSnapshot from labelfab.agent.publisher import NullPublisher, Publisher, RecordingPublisher from labelfab.agent.spool import InsertResult, Outcome, Spool from labelfab.agent.worker import PrintWorker @@ -19,6 +20,7 @@ "Batch", "Coalescer", "Config", + "DeviceSnapshot", "InsertResult", "NullPublisher", "Outcome", diff --git a/src/labelfab/agent/__main__.py b/src/labelfab/agent/__main__.py index 0fcde99..be69b70 100644 --- a/src/labelfab/agent/__main__.py +++ b/src/labelfab/agent/__main__.py @@ -93,6 +93,11 @@ def run(self) -> int: signal.signal(signal.SIGINT, self.stop) signal.signal(signal.SIGTERM, self.stop) + # Before the source starts, so whatever these two learn is what the first + # retained status carries: the source republishes the last thing published, + # and nothing has been published yet. + if self.config.device.probe_on_start: + self.worker.probe_device() self.worker.recover() if self.dir_source: self.dir_source.poll() diff --git a/src/labelfab/agent/config.py b/src/labelfab/agent/config.py index c10dfa8..6903999 100644 --- a/src/labelfab/agent/config.py +++ b/src/labelfab/agent/config.py @@ -95,6 +95,13 @@ class DeviceSection(BaseModel): #: Drop the socket between batches: the D30 auto-sleeps, so a held-open socket #: just relocates the failure. Reconnect-per-batch is cheaper to reason about. idle_disconnect: bool = True + #: Ask the printer what it is once at startup, so the status topic carries live + #: truth rather than remembered truth from the moment the agent comes up. This + #: cannot wake a sleeping unit -- it has powered its radio off and only the button + #: brings it back -- so the cost of a miss is one connect timeout, after which the + #: stored snapshot is published unchanged. On by default; turn it off if that + #: timeout is in the way of a fast start. Never runs on a broker reconnect. + probe_on_start: bool = True @field_validator("density") @classmethod diff --git a/src/labelfab/agent/device_state.py b/src/labelfab/agent/device_state.py new file mode 100644 index 0000000..5ce7bd1 --- /dev/null +++ b/src/labelfab/agent/device_state.py @@ -0,0 +1,123 @@ +"""What the printer last said about itself, and when it said it. + +The D30 only talks while it is being printed to. It auto-powers-off, and +``device.idle_disconnect`` defaults to true precisely because a held-open socket just +relocates the failure -- so between jobs there is nobody to ask. Every consumer of the +status topic therefore reads *remembered* device truth most of the time, and the honest +way to serve that is to remember it deliberately and say how old it is, rather than +either forgetting it or waking the printer to refresh it. + +``seen_at`` is the whole point. Without it a consumer cannot distinguish "media is +loaded" from "media was loaded at some unknown time in the past", and rendering the +second as the first is exactly the failure the tri-state ``media_ok`` was introduced to +avoid, one layer along. + +This also collects the six loose ``_device_*`` fields the worker used to carry and the +two places that built a ``PrinterStatus`` out of them. That duplication is why the +worker and the MQTT source disagreed: one published everything the printer had said, +the other published a hardcoded blank. +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from pydantic import BaseModel, ConfigDict + +from labelfab.contract import PrinterStatus +from labelfab.device.d30 import MODEL +from labelfab.device.feedback import DeviceFeedback + + +class DeviceSnapshot(BaseModel): + """Last-known device truth, persisted in the spool and republished on connect.""" + + # ``extra="ignore"`` is load-bearing rather than a default: this is stored in the + # spool DB, which outlives package upgrades, so a row written by a different agent + # version has to degrade to the fields it has in common. Failing to decode it would + # take the agent down at boot, and a status field is not worth an outage. + model_config = ConfigDict(extra="ignore", frozen=True) + + serial: str | None = None + firmware: str | None = None + battery_pct: int | None = None + voltage_v: float | None = None + media_ok: bool | None = None + #: What the printer was complaining about the last time it was asked. Never latched + #: -- see ``merge``. + fault: str | None = None + #: Epoch seconds when the printer last reported anything, or ``None`` if it never + #: has. Stored as a float to match the rest of the spool; converted to a UTC + #: instant at the wire boundary. + seen_at: float | None = None + + # -- accumulation ------------------------------------------------------- # + + def merge(self, fb: DeviceFeedback, *, now: float) -> DeviceSnapshot: + """Fold one connection's feedback in, keeping whatever it did not report. + + A connection that answered nothing must not blank out a serial we already + know. ``fault`` is the deliberate exception: it is recomputed from scratch + every connection, so a cleared media error clears the status instead of + latching an error forever. + + ``seen_at`` only moves when the printer actually said something. Advancing it + on a silent connection would claim a freshness we do not have, and this + timestamp is the only thing standing between a consumer and treating + three-day-old truth as live. + """ + reported = (fb.serial, fb.firmware, fb.battery_pct, fb.voltage_v, fb.paper_ok) + return DeviceSnapshot( + serial=fb.serial or self.serial, + firmware=fb.firmware or self.firmware, + battery_pct=self.battery_pct if fb.battery_pct is None else fb.battery_pct, + voltage_v=self.voltage_v if fb.voltage_v is None else fb.voltage_v, + media_ok=self.media_ok if fb.paper_ok is None else fb.paper_ok, + fault=fb.fault(), + seen_at=now if any(v is not None for v in reported) else self.seen_at, + ) + + # -- publishing --------------------------------------------------------- # + + def settled_state(self) -> str: + """What to publish once nothing is in flight. + + "idle" unless the printer is complaining. Kept apart from the status builder so + that "printing" and "disconnected" -- facts about the link, not about the media + -- are never silently rewritten into it. + """ + return "error" if self.fault else "idle" + + def to_status( + self, + printer_id: str, + *, + state: str, + tape_width_mm: float | None = None, + pending_labels: int = 0, + ) -> PrinterStatus: + """Build the retained status message from this snapshot.""" + return PrinterStatus( + printer_id=printer_id, + state=state, # type: ignore[arg-type] + model=MODEL, + serial=self.serial, + firmware=self.firmware, + battery_pct=self.battery_pct, + voltage_v=self.voltage_v, + media_ok=self.media_ok, + tape_width_mm=tape_width_mm, + pending_labels=pending_labels, + error=self.fault, + device_seen_at=self.seen_at_utc(), + ) + + def seen_at_utc(self) -> datetime | None: + """``seen_at`` as a UTC instant, whole seconds. + + Truncated because sub-second precision on "when did the printer last speak" is + noise in a payload whose primary reader is a human running ``mosquitto_sub -v``. + """ + if self.seen_at is None: + return None + return datetime.fromtimestamp(self.seen_at, tz=timezone.utc).replace(microsecond=0) diff --git a/src/labelfab/agent/source_mqtt.py b/src/labelfab/agent/source_mqtt.py index 7821552..5c9954d 100644 --- a/src/labelfab/agent/source_mqtt.py +++ b/src/labelfab/agent/source_mqtt.py @@ -55,10 +55,16 @@ def __init__(self, config: Config, spool: Spool, enqueue: Callable[[str], None]) if config.mqtt.username: self.client.username_pw_set(config.mqtt.username, config.mqtt.password) - # Last will: if the agent drops, the retained status flips to disconnected so - # a producer sees the printer is unreachable before it ever clicks print. - will = PrinterStatus(printer_id=config.agent.printer_id, state="disconnected") - self.client.will_set(config.topic("status"), will.model_dump_json(), qos=1, retain=True) + # What the retained topic currently says. Seeded from the spool so a fresh + # process starts out holding the previous one's device truth, then updated by + # every retained publish -- which is what lets a reconnect restore the topic + # instead of flattening it. + snapshot = spool.device_snapshot() + self._last_status = snapshot.to_status( + config.agent.printer_id, + state=snapshot.settled_state(), + tape_width_mm=config.tape.width_mm, + ) self.client.on_connect = self._on_connect self.client.on_message = self._on_message @@ -66,24 +72,45 @@ def __init__(self, config: Config, spool: Spool, enqueue: Callable[[str], None]) # -- lifecycle ---------------------------------------------------------- # def start(self) -> None: + # The will is set here rather than in __init__ because paho cannot change it on + # a live connection, so this is the last moment it can be made current -- after + # the startup probe and crash recovery have had their say. + self.client.will_set( + self.config.topic("status"), self._disconnected().model_dump_json(), qos=1, retain=True + ) self.client.connect(self.config.mqtt.host, self.config.mqtt.port, self.config.mqtt.keepalive_s) self.client.loop_start() def stop(self) -> None: try: - self.publish_status( - PrinterStatus(printer_id=self.config.agent.printer_id, state="disconnected") - ) + self.publish_status(self._disconnected()) finally: self.client.loop_stop() self.client.disconnect() + def _disconnected(self) -> PrinterStatus: + """Last-known truth with the link marked down. + + Backs both the will and the shutdown notice: if the agent drops, a producer + must see that the printer is unreachable before it ever clicks print. Losing + the link is news about reachability, though, not grounds for forgetting the + serial, the firmware and the media state -- ``device_seen_at`` rides along to + say how old they are. + """ + return self._last_status.model_copy(update={"state": "disconnected", "pending_labels": 0}) + # -- Publisher protocol ------------------------------------------------- # def publish_result(self, result: JobResult) -> None: self.client.publish(self.config.topic("results"), result.model_dump_json(), qos=1) def publish_status(self, status: PrinterStatus, *, retain: bool = True) -> None: + if retain: + # Recorded before the publish and regardless of whether it succeeds: the + # startup probe and crash recovery both publish before the client is + # connected, so those go nowhere and the connect below is what lands them. + # A non-retained publish is not what the topic holds, so it is not recorded. + self._last_status = status self.client.publish(self.config.topic("status"), status.model_dump_json(), qos=1, retain=retain) def publish_progress(self, job_id: str, printed: int, total: int) -> None: @@ -98,13 +125,14 @@ def _on_connect(self, client, userdata, flags, reason_code, properties=None) -> return client.subscribe(self.config.topic("jobs"), qos=self.config.mqtt.qos) client.subscribe(self.config.topic("cmd"), qos=1) - self.publish_status( - PrinterStatus( - printer_id=self.config.agent.printer_id, - state="idle", - tape_width_mm=self.config.tape.width_mm, - ) - ) + # Republish what the topic already said, rather than a freshly synthesised + # blank. This fires on every reconnect, not just the first connect, and paho + # reconnects silently -- the Istio route for the broker caps at a 24h timeout + # and flaps besides, so it runs several times a day. Publishing a hardcoded + # bare status here was not merely uninformative, it was lossy: it overwrote a + # retained message that was correct a moment earlier, and the printer is asleep + # and cannot be asked again. + self.publish_status(self._last_status) def _on_message(self, client, userdata, msg) -> None: if msg.topic.endswith("/cmd"): diff --git a/src/labelfab/agent/spool.py b/src/labelfab/agent/spool.py index 5b8bffb..c3dc028 100644 --- a/src/labelfab/agent/spool.py +++ b/src/labelfab/agent/spool.py @@ -14,10 +14,16 @@ its cached result and prints nothing. * ``dedupe_key`` deduplicates individual *labels* across different jobs, so a "print bin A4" that fires twice an hour apart does not waste tape. + +One thing here is not about jobs at all: the last :class:`DeviceSnapshot`. It is stored +alongside them because this file is already the agent's only durable place, and because +the printer is only reachable during a print -- so the alternative to remembering what +it said is waking it up to ask again. """ from __future__ import annotations +import logging import sqlite3 import time from collections.abc import Callable @@ -25,8 +31,13 @@ from enum import Enum from pathlib import Path +from pydantic import ValidationError + +from labelfab.agent.device_state import DeviceSnapshot from labelfab.contract import JobResult, PrintJob +log = logging.getLogger("labelfab.spool") + SCHEMA = """ CREATE TABLE IF NOT EXISTS jobs ( job_id TEXT PRIMARY KEY, @@ -51,6 +62,14 @@ printed_at REAL NOT NULL, PRIMARY KEY (job_id, label_index) ); + +-- One row, holding the last DeviceSnapshot. The printer only talks while it is being +-- printed to, so without this the agent starts every process knowing nothing about the +-- hardware it drives until the next job happens to come along. +CREATE TABLE IF NOT EXISTS device_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + payload TEXT NOT NULL +); """ @@ -186,3 +205,34 @@ def printed_labels(self, job_id: str) -> set[int]: "SELECT label_index FROM printed_labels WHERE job_id = ?", (job_id,) ).fetchall() return {r["label_index"] for r in rows} + + # -- last-known device truth -------------------------------------------- # + # + # Written after every print, read once at startup. This is the only state here that + # is not about jobs, and it lives here because the alternative -- asking the printer + # on boot -- means waking a device whose whole operating model is "asleep until + # needed", every time the agent restarts. + + def save_device_snapshot(self, snapshot: DeviceSnapshot) -> None: + self._db.execute( + "INSERT INTO device_state (id, payload) VALUES (1, ?) " + "ON CONFLICT(id) DO UPDATE SET payload = excluded.payload", + (snapshot.model_dump_json(),), + ) + + def device_snapshot(self) -> DeviceSnapshot: + """The last snapshot written, or an empty one. + + Empty covers both "this agent has never printed" and "the stored row is no + longer readable". The second degrades rather than raising on purpose: an + unusable status field is a cosmetic problem, and taking the print agent down at + boot over one would be a much larger one. + """ + row = self._db.execute("SELECT payload FROM device_state WHERE id = 1").fetchone() + if row is None: + return DeviceSnapshot() + try: + return DeviceSnapshot.model_validate_json(row["payload"]) + except ValidationError: + log.warning("stored device snapshot is unreadable; starting without one") + return DeviceSnapshot() diff --git a/src/labelfab/agent/worker.py b/src/labelfab/agent/worker.py index da32ed4..89895cc 100644 --- a/src/labelfab/agent/worker.py +++ b/src/labelfab/agent/worker.py @@ -30,7 +30,6 @@ PX_PER_MM, JobResult, LabelResult, - PrinterStatus, PrintJob, TapeSpec, ) @@ -44,6 +43,10 @@ #: Backoff between reconnect attempts within one send, in seconds. _BACKOFF_S = (1.0, 3.0, 9.0) +#: How long the startup probe waits for the printer's replies to arrive on the +#: transport's reader thread before reading them. +_PROBE_SETTLE_S = 1.0 + @dataclass class _JobAcc: @@ -95,14 +98,10 @@ def __init__( self.retry_interval_s = retry_interval_s #: job_id -> last stall time, so a napping printer is retried, not busy-looped. self._stalled: dict[str, float] = {} - #: Latest device serial reported by the printer, on either transport. - self._device_serial: str | None = None - self._device_firmware: str | None = None - self._device_battery_pct: int | None = None - self._device_voltage_v: float | None = None - self._device_media_ok: bool | None = None - #: Last fault the printer reported, or None. Drives the "error" status state. - self._device_fault: str | None = None + #: What the printer last said about itself. Seeded from the spool, so a restart + #: starts out knowing what the previous process learned instead of publishing a + #: row of nulls until something happens to print. + self._device = spool.device_snapshot() self.coalescer = Coalescer( max_wait_s=config.strip.max_wait_s, max_length_mm=config.strip.max_length_mm, @@ -252,6 +251,48 @@ def recover(self) -> None: except KeyError: pass + def probe_device(self) -> bool: + """Ask the printer what it is, if it happens to be awake. Returns whether it answered. + + This is not a wake attempt and cannot be one: a slept D30 has powered its radio + down (``AUTO_POWER_TIME``, HARDWARE-NOTES §7) and only the button brings it + back, so this either finds it already awake and learns the truth, or fails + inside the transport's connect timeout and leaves the remembered snapshot + exactly as it was. + + Deliberately called from the run loop at startup, and only there. Not from the + MQTT ``on_connect`` callback: that is paho's network thread, so probing from it + would touch the printer concurrently with a print -- the one thing the + single-threaded loop exists to make impossible -- and it fires on every broker + reconnect, which happens several times a day and says nothing about the printer. + + One attempt, no backoff. The retry ladder in ``_send`` exists to get a *job* + onto tape; there is no job here and nothing is lost by giving up immediately. + """ + printer = self.printer_factory() + try: + printer.connect() + printer.refresh_telemetry() + except D30Error as exc: + printer.close() + log.info("startup probe: printer did not answer (%s); keeping the stored status", exc) + return False + except Exception: # a status nicety must never be what stops the agent starting + printer.close() + log.exception("startup probe failed; keeping the stored status") + return False + # Replies land asynchronously on the transport's reader thread, and unlike the + # print path there is no print duration to cover the wait. Settling short is + # safe by construction: an empty merge keeps every stored field and does not + # advance seen_at, so the worst case is learning nothing, never losing anything. + self.sleep(_PROBE_SETTLE_S) + self._capture_feedback(printer) + printer.close() + # Publish it: the MQTT source has not connected yet, so this does not reach the + # broker -- it updates what the source will republish the moment it does. + self._publish_status(self._settled_state()) + return True + # -- flushing and printing ---------------------------------------------- # def _flush(self) -> None: @@ -375,26 +416,17 @@ def _capture_feedback(self, printer: PhomemoD30) -> None: changes just as BLE does -- so this is unconditional. A transport that somehow has no ``feedback`` is a programming error worth surfacing, not a case to silently skip; the previous ``None`` guard would have hidden exactly that. + + Persisted immediately. The printer is only reachable while it is being printed + to, so this instant is the *only* chance to learn any of it, and dropping it on + process exit is what left the status topic full of nulls. """ fb = printer.feedback - # Only overwrite what it actually reported this time: a connection that - # answered nothing must not blank out a serial we already know. The fault is - # the exception -- it is recomputed every time, so a cleared media error - # clears the status rather than latching an error forever. - if fb.serial: - self._device_serial = fb.serial - if fb.firmware: - self._device_firmware = fb.firmware - if fb.battery_pct is not None: - self._device_battery_pct = fb.battery_pct - if fb.voltage_v is not None: - self._device_voltage_v = fb.voltage_v - if fb.paper_ok is not None: - self._device_media_ok = fb.paper_ok - self._device_fault = fb.fault() + self._device = self._device.merge(fb, now=self.clock()) + self.spool.save_device_snapshot(self._device) log.info("device feedback: %s", fb.summary()) - if self._device_fault: - log.warning("printer reports a fault: %s", self._device_fault) + if self._device.fault: + log.warning("printer reports a fault: %s", self._device.fault) # -- result accounting -------------------------------------------------- # @@ -470,27 +502,15 @@ def _reject(self, job: PrintJob, reason: str) -> None: # -- status ------------------------------------------------------------- # def _settled_state(self) -> str: - """What to publish once a batch is done. - - "idle" unless the printer is complaining. Kept explicit rather than folded into - ``_publish_status`` so "printing" and "disconnected" -- which are facts about - the link, not the media -- are never silently rewritten. - """ - return "error" if self._device_fault else "idle" + """What to publish once a batch is done. See ``DeviceSnapshot.settled_state``.""" + return self._device.settled_state() def _publish_status(self, state: str, *, pending: int = 0) -> None: self.publisher.publish_status( - PrinterStatus( - printer_id=self.config.agent.printer_id, - state=state, # type: ignore[arg-type] - model=MODEL, - serial=self._device_serial, - firmware=self._device_firmware, - battery_pct=self._device_battery_pct, - voltage_v=self._device_voltage_v, - media_ok=self._device_media_ok, + self._device.to_status( + self.config.agent.printer_id, + state=state, tape_width_mm=self.config.tape.width_mm, pending_labels=pending, - error=self._device_fault, ) ) diff --git a/src/labelfab/contract/models.py b/src/labelfab/contract/models.py index 8b5b0ee..69104bb 100644 --- a/src/labelfab/contract/models.py +++ b/src/labelfab/contract/models.py @@ -7,6 +7,7 @@ from __future__ import annotations +from datetime import datetime from typing import Annotated, Any, Literal from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -260,6 +261,16 @@ class PrinterStatus(Base): tape_width_mm: float | None = None pending_labels: int = 0 error: str | None = None + #: When the device fields above were last heard from the printer; ``None`` if it + #: has never said anything. The D30 is only reachable while it is being printed to, + #: so most reads of this topic are reads of *remembered* truth: the agent stores + #: what it last heard and republishes it across restarts and broker reconnects + #: rather than going quiet. This is how a consumer tells the two apart. + #: + #: An absolute instant rather than an age because this message is *retained*: an + #: age is computed at publish time and then sits on the broker getting wronger, + #: while a timestamp stays true no matter how long nobody republishes it. + device_seen_at: datetime | None = None def job_schema() -> dict[str, Any]: diff --git a/tests/test_agent_config.py b/tests/test_agent_config.py index 640e6dc..8195462 100644 --- a/tests/test_agent_config.py +++ b/tests/test_agent_config.py @@ -46,6 +46,16 @@ def test_density_defaults_to_light_and_reaches_the_driver(): assert printer.config.density == 1 +def test_the_startup_probe_is_on_and_can_be_turned_off(tmp_path): + """On by default because it cannot wake a sleeping printer, so the only cost of a + miss is one connect timeout. The knob exists for hosts where that is in the way.""" + assert Config().device.probe_on_start is True + + toml = tmp_path / "agent.toml" + toml.write_text("[device]\nprobe_on_start = false\n") + assert load(toml).device.probe_on_start is False + + def test_an_unknown_density_is_rejected_at_load(): import pytest from pydantic import ValidationError diff --git a/tests/test_device_state.py b/tests/test_device_state.py new file mode 100644 index 0000000..bc07660 --- /dev/null +++ b/tests/test_device_state.py @@ -0,0 +1,117 @@ +"""Last-known device truth: what is remembered, and how old it is allowed to look. + +The rules here all serve one property. The printer is only reachable while it is being +printed to, so almost every read of the status topic is a read of *remembered* truth. +That is fine as long as nobody can mistake it for live truth — which is what ``seen_at`` +is for, and why it is not allowed to drift forward on its own. +""" + +from __future__ import annotations + +from datetime import timezone + +from labelfab.agent import DeviceSnapshot +from labelfab.device import DeviceFeedback + +SERIAL = "Q223P4C31420105" + +#: Frames as the printer actually sends them; see tests/test_feedback.py. +FULL = ("1a08" + SERIAL.encode().hex(), "1a07020102", "1a0464", "1a2f01a1", "1a0689") +MEDIA_BAD = "1a0688" + + +def _feedback(*frames_hex: str) -> DeviceFeedback: + fb = DeviceFeedback() + for frame in frames_hex: + fb.ingest(bytes.fromhex(frame)) + return fb + + +def test_merge_decodes_a_whole_connection(): + snap = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + assert snap.serial == SERIAL + assert snap.firmware == "2.1.2" + assert snap.battery_pct == 100 + assert snap.voltage_v == 4.17 + assert snap.media_ok is True + assert snap.fault is None + assert snap.seen_at == 1000.0 + + +def test_a_quiet_connection_keeps_what_we_already_knew(): + """Otherwise one uncommunicative connect erases a serial learned an hour ago.""" + known = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + still = known.merge(_feedback(), now=2000.0) + assert still.serial == SERIAL + assert still.firmware == "2.1.2" + assert still.battery_pct == 100 + assert still.media_ok is True + + +def test_a_quiet_connection_does_not_advance_seen_at(): + """The timestamp is the only thing separating remembered truth from live truth. + + Moving it because we merely *connected* would launder three-day-old media state + into something a consumer renders as current. + """ + known = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + assert known.merge(_feedback(), now=9999.0).seen_at == 1000.0 + + +def test_a_partial_report_advances_seen_at_and_leaves_the_rest(): + later = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0).merge(_feedback("1a0458"), now=2000.0) + assert later.battery_pct == 88 # the one thing it reported + assert later.serial == SERIAL # everything it did not + assert later.seen_at == 2000.0 + + +def test_a_fault_does_not_latch(): + """A media error that has cleared must stop being reported, or the printer looks + broken until someone restarts the agent.""" + faulted = DeviceSnapshot().merge(_feedback(MEDIA_BAD), now=1000.0) + assert faulted.fault and faulted.settled_state() == "error" + + cleared = faulted.merge(_feedback(*FULL), now=2000.0) + assert cleared.fault is None + assert cleared.settled_state() == "idle" + assert cleared.media_ok is True + + +def test_status_carries_the_snapshot_and_its_age(): + snap = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + status = snap.to_status("d30-workshop", state="idle", tape_width_mm=15.0, pending_labels=3) + + assert status.serial == SERIAL + assert status.media_ok is True + assert status.pending_labels == 3 + assert status.device_seen_at is not None + assert status.device_seen_at.tzinfo is not None # a bare instant is unusable + assert status.device_seen_at.timestamp() == 1000.0 + + +def test_never_observed_publishes_null_not_a_guess(): + status = DeviceSnapshot().to_status("d30-workshop", state="idle") + assert status.device_seen_at is None + assert status.media_ok is None # silence is still not health + + +def test_seen_at_is_whole_seconds_utc(): + """Sub-second precision on "when did the printer last speak" is noise in a payload + whose main reader is a human running mosquitto_sub.""" + seen = DeviceSnapshot(seen_at=1000.75).seen_at_utc() + assert seen is not None + assert seen.microsecond == 0 + assert seen.tzinfo is timezone.utc + + +def test_a_snapshot_survives_a_json_round_trip(): + snap = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + assert DeviceSnapshot.model_validate_json(snap.model_dump_json()) == snap + + +def test_a_snapshot_from_another_version_degrades_rather_than_raising(): + """This row outlives package upgrades. Refusing to decode one written by a different + agent version would turn a cosmetic status field into a boot failure.""" + snap = DeviceSnapshot.model_validate_json(f'{{"serial": "{SERIAL}", "future_field": 42}}') + assert snap.serial == SERIAL + assert snap.seen_at is None diff --git a/tests/test_source_mqtt.py b/tests/test_source_mqtt.py new file mode 100644 index 0000000..691f7b5 --- /dev/null +++ b/tests/test_source_mqtt.py @@ -0,0 +1,219 @@ +"""What the retained status topic holds, across connects and reconnects. + +Driven through a fake paho client rather than a broker: everything worth asserting +here is about *which payload* is published on which callback, and none of it needs a +network. The one thing a broker would add — that paho reconnects on its own, silently, +several times a day — is the premise, not something under test. +""" + +from __future__ import annotations + +import json + +import pytest +from conftest import make_config + +from labelfab.agent import DeviceSnapshot, Spool +from labelfab.agent.source_mqtt import FLUSH_COMMAND, MqttSource +from labelfab.contract import PrinterStatus + +SERIAL = "Q223P4C31420105" + + +class FakeClient: + """Stands in for ``paho.mqtt.client.Client``, recording publishes and subscriptions.""" + + def __init__(self, *_args, **_kwargs) -> None: + self.published: list[tuple[str, str, bool]] = [] + self.subscribed: list[str] = [] + self.will: tuple[str, str] | None = None + self.on_connect = None + self.on_message = None + self.acked: list[int] = [] + + # -- setup the source performs ------------------------------------------ # + def ws_set_options(self, **_kw) -> None: ... + def tls_set(self, *_a, **_kw) -> None: ... + def username_pw_set(self, *_a, **_kw) -> None: ... + def connect(self, *_a, **_kw) -> None: ... + def loop_start(self) -> None: ... + def loop_stop(self) -> None: ... + def disconnect(self) -> None: ... + + def will_set(self, topic, payload, qos=0, retain=False) -> None: + self.will = (topic, payload) + + def subscribe(self, topic, qos=0) -> None: + self.subscribed.append(topic) + + def publish(self, topic, payload, qos=0, retain=False) -> None: + self.published.append((topic, payload, retain)) + + def ack(self, mid, qos) -> None: + self.acked.append(mid) + + # -- what tests read ---------------------------------------------------- # + @property + def statuses(self) -> list[dict]: + return [json.loads(p) for t, p, _ in self.published if t.endswith("/status")] + + +@pytest.fixture +def source(tmp_path, monkeypatch): + """Build an MqttSource over a fake client, on a spool the test can pre-load.""" + import paho.mqtt.client as mqtt + + monkeypatch.setattr(mqtt, "Client", FakeClient) + + def build(snapshot: DeviceSnapshot | None = None) -> MqttSource: + spool = Spool(tmp_path / "spool.db") + if snapshot is not None: + spool.save_device_snapshot(snapshot) + config = make_config() + config.mqtt.host = "broker.invalid" + return MqttSource(config, spool, lambda _job_id: None) + + return build + + +def _connect(src: MqttSource) -> None: + """Fire the callback paho fires on connect — and on every silent reconnect.""" + src._on_connect(src.client, None, {}, 0) + + +KNOWN = DeviceSnapshot( + serial=SERIAL, firmware="2.1.2", battery_pct=100, voltage_v=4.18, media_ok=True, seen_at=1000.0 +) + + +def test_a_reconnect_republishes_device_truth_instead_of_erasing_it(source): + """The reported defect. A reconnect used to publish a hardcoded bare status, + retained, over a message that was correct a moment earlier — and the printer is + asleep, so nothing could put it back until the next print.""" + src = source() + src.publish_status( + PrinterStatus(printer_id="d30-workshop", state="idle", serial=SERIAL, media_ok=True) + ) + + _connect(src) # paho reconnects on its own, several times a day + + restored = src.client.statuses[-1] + assert restored["serial"] == SERIAL + assert restored["media_ok"] is True + + +def test_a_fresh_process_republishes_what_the_last_one_learned(source): + """The original issue: a restarted agent had nothing to say about the printer until + something happened to print.""" + src = source(KNOWN) + _connect(src) + + published = src.client.statuses[-1] + assert published["serial"] == SERIAL + assert published["firmware"] == "2.1.2" + assert published["media_ok"] is True + assert published["device_seen_at"] is not None + assert published["state"] == "idle" + + +def test_a_remembered_fault_is_not_downgraded_to_healthy_on_restart(source): + src = source(KNOWN.model_copy(update={"media_ok": False, "fault": "media not ready"})) + _connect(src) + + published = src.client.statuses[-1] + assert published["state"] == "error" + assert published["media_ok"] is False + assert published["error"] == "media not ready" + + +def test_an_agent_that_has_never_printed_claims_nothing(source): + src = source() + _connect(src) + + published = src.client.statuses[-1] + assert published["state"] == "idle" + assert published["serial"] is None + assert published["media_ok"] is None # not "fine", just unsaid + assert published["device_seen_at"] is None + + +def test_the_status_is_retained_and_the_topics_are_subscribed(source): + src = source(KNOWN) + _connect(src) + + assert all(retain for topic, _, retain in src.client.published if topic.endswith("/status")) + assert src.client.subscribed == [ + "se/v1/print/d30-workshop/jobs", + "se/v1/print/d30-workshop/cmd", + ] + + +def test_a_failed_connect_publishes_nothing(source): + src = source(KNOWN) + src._on_connect(src.client, None, {}, 5) # not authorised + assert src.client.published == [] + + +def test_the_will_carries_last_known_truth(source): + """An ungraceful drop should say the printer is unreachable, not forget what it is.""" + src = source(KNOWN) + src.start() + + assert src.client.will is not None + will = json.loads(src.client.will[1]) + assert will["state"] == "disconnected" + assert will["serial"] == SERIAL + assert will["device_seen_at"] is not None + + +def test_shutdown_says_disconnected_without_forgetting(source): + src = source(KNOWN) + src.stop() + + final = src.client.statuses[-1] + assert final["state"] == "disconnected" + assert final["pending_labels"] == 0 + assert final["serial"] == SERIAL + + +def test_a_reconnect_mid_print_republishes_printing_not_idle(source): + """`state` is a fact about the link and the batch. Synthesising "idle" here would + tell a producer the strip it is waiting on had finished.""" + src = source(KNOWN) + src.publish_status(KNOWN.to_status("d30-workshop", state="printing", pending_labels=12)) + + _connect(src) + + republished = src.client.statuses[-1] + assert republished["state"] == "printing" + assert republished["pending_labels"] == 12 + + +def test_a_non_retained_status_does_not_become_what_the_topic_holds(source): + src = source(KNOWN) + src.publish_status(PrinterStatus(printer_id="d30-workshop", state="printing"), retain=False) + + _connect(src) + + assert src.client.statuses[-1]["serial"] == SERIAL + + +def test_a_flush_command_reaches_the_print_loop(tmp_path, monkeypatch): + """The cmd topic is handled on paho's thread, so it must only ever enqueue.""" + import paho.mqtt.client as mqtt + + monkeypatch.setattr(mqtt, "Client", FakeClient) + enqueued: list[str] = [] + config = make_config() + config.mqtt.host = "broker.invalid" + src = MqttSource(config, Spool(tmp_path / "spool.db"), enqueued.append) + + class _Msg: + topic = "se/v1/print/d30-workshop/cmd" + payload = b"flush" + qos = 1 + mid = 7 + + src._on_message(src.client, None, _Msg()) + assert enqueued == [FLUSH_COMMAND] + assert src.client.acked == [7] diff --git a/tests/test_spool.py b/tests/test_spool.py index 6f02d46..2b5961f 100644 --- a/tests/test_spool.py +++ b/tests/test_spool.py @@ -4,7 +4,7 @@ from conftest import Clock, make_job -from labelfab.agent import Outcome, Spool +from labelfab.agent import DeviceSnapshot, Outcome, Spool from labelfab.contract import JobResult, LabelResult @@ -45,6 +45,35 @@ def test_label_printed_survives_and_dedups(tmp_path): assert spool.printed_labels("j") == {0, 2} +def test_device_snapshot_survives_the_process(tmp_path): + """The printer only talks during a print, so this is the only way a restarted agent + knows anything about the hardware it drives.""" + path = tmp_path / "s.db" + snap = DeviceSnapshot(serial="Q223P4C31420105", firmware="2.1.2", media_ok=True, seen_at=1000.0) + + first = Spool(path) + assert first.device_snapshot() == DeviceSnapshot() # nothing learned yet + first.save_device_snapshot(snap) + first.close() + + assert Spool(path).device_snapshot() == snap + + +def test_the_latest_device_snapshot_replaces_the_last(tmp_path): + spool = Spool(tmp_path / "s.db") + spool.save_device_snapshot(DeviceSnapshot(battery_pct=100, seen_at=1000.0)) + spool.save_device_snapshot(DeviceSnapshot(battery_pct=88, seen_at=2000.0)) + assert spool.device_snapshot().battery_pct == 88 + + +def test_an_unreadable_device_snapshot_does_not_take_the_agent_down(tmp_path): + """A status field is cosmetic; refusing to boot over one would not be.""" + spool = Spool(tmp_path / "s.db") + spool.save_device_snapshot(DeviceSnapshot(serial="X")) + spool._db.execute("UPDATE device_state SET payload = 'not json' WHERE id = 1") + assert spool.device_snapshot() == DeviceSnapshot() + + def test_dedupe_recorded_and_purged(tmp_path): clock = Clock() spool = Spool(tmp_path / "s.db", clock=clock) diff --git a/tests/test_worker.py b/tests/test_worker.py index 2bb4047..8765035 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -228,6 +228,81 @@ def test_a_fault_clears_on_the_next_good_batch(harness): assert settled.media_ok is True +def test_status_says_when_the_printer_was_last_heard_from(harness, clock): + h = harness() + _reporting_factory(h) + h.submit(make_job("j", n_labels=1, flush=True)) + + seen = h.publisher.statuses[-1].device_seen_at + assert seen is not None and seen.timestamp() == clock.now + + +def test_device_truth_is_persisted_for_the_next_process(harness): + """Everything the printer says is said during a print. Not writing it down here is + what left the status topic full of nulls after a restart.""" + h = harness() + _reporting_factory(h) + h.submit(make_job("j", n_labels=1, flush=True)) + + stored = h.spool.device_snapshot() + assert stored.serial == "Q223P4C31420105" + assert stored.firmware == "2.1.2" + assert stored.media_ok is True + assert stored.seen_at is not None + + +def test_a_restarted_worker_does_not_start_blind(tmp_path, clock): + """A fresh process on the same spool publishes what the previous one learned — + including on the disconnected status it emits when the printer is asleep.""" + from conftest import WorkerHarness + + first = WorkerHarness(tmp_path, clock) + _reporting_factory(first) + first.submit(make_job("j", n_labels=1, flush=True)) + first.spool.close() + + restarted = WorkerHarness(tmp_path, clock) # same spool.db, new worker + restarted.offline = True # printer has since gone to sleep + restarted.submit(make_job("k", n_labels=1, flush=True)) + + status = restarted.publisher.statuses[-1] + assert status.state == "disconnected" + assert status.serial == "Q223P4C31420105" + assert status.media_ok is True + assert status.device_seen_at is not None + + +def test_the_startup_probe_surveys_a_printer_that_is_awake(harness, clock): + h = harness() + _reporting_factory(h) + + assert h.worker.probe_device() is True + + status = h.publisher.statuses[-1] + assert status.state == "idle" + assert status.serial == "Q223P4C31420105" + assert status.device_seen_at is not None and status.device_seen_at.timestamp() == clock.now + assert h.frames == 0 # surveyed, not printed + + +def test_a_sleeping_printer_leaves_the_stored_status_alone(tmp_path, clock): + """A slept D30 has its radio off and cannot be woken over Bluetooth, so the probe + is expected to miss. Missing must cost nothing beyond a connect timeout.""" + from conftest import WorkerHarness + + first = WorkerHarness(tmp_path, clock) + _reporting_factory(first) + first.submit(make_job("j", n_labels=1, flush=True)) + first.spool.close() + + restarted = WorkerHarness(tmp_path, clock) + restarted.offline = True + + assert restarted.worker.probe_device() is False + assert restarted.spool.device_snapshot().serial == "Q223P4C31420105" + assert restarted.publisher.statuses == [] # nothing learned, nothing to say + + def test_a_fault_is_captured_even_when_the_print_fails(harness): """The fault is usually *why* it failed, so closing without reading it loses it.""" h = harness() From 09c6487083c204d79383b2809e8a610cb9c206cd Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sun, 2 Aug 2026 16:28:03 +0200 Subject: [PATCH 2/3] fix(agent): re-arm the will on connect so it stops describing startup forever Review flagged that the will goes stale after start(): paho bakes it into the CONNECT packet, so the payload the broker holds is fixed for the life of a connection and an ungraceful drop weeks in would have published boot-time device truth. Half of that is unavoidable and stays true -- nothing can change what fires if *this* connection dies. The other half is not: paho rebuilds CONNECT from the same _will_* fields on every automatic reconnect (verified in 2.1.0 -- will_set has no connected-state guard, and _send_connect reads _will_payload at packet-build time). Re-arming from _on_connect therefore bounds the staleness by the reconnect interval, a few hours here, instead of by the process lifetime. Armed from paho's network thread, which is also the thread that builds the packet, so the topic and payload can never come from different arming passes. --- src/labelfab/agent/source_mqtt.py | 24 +++++++++++++++++++----- tests/test_source_mqtt.py | 19 +++++++++++++++++++ 2 files changed, 38 insertions(+), 5 deletions(-) diff --git a/src/labelfab/agent/source_mqtt.py b/src/labelfab/agent/source_mqtt.py index 5c9954d..3c6f40b 100644 --- a/src/labelfab/agent/source_mqtt.py +++ b/src/labelfab/agent/source_mqtt.py @@ -72,14 +72,27 @@ def __init__(self, config: Config, spool: Spool, enqueue: Callable[[str], None]) # -- lifecycle ---------------------------------------------------------- # def start(self) -> None: - # The will is set here rather than in __init__ because paho cannot change it on - # a live connection, so this is the last moment it can be made current -- after - # the startup probe and crash recovery have had their say. + # Armed here rather than in __init__ so it reflects whatever the startup probe + # and crash recovery learned, which happen after this object is built. + self._arm_will() + self.client.connect(self.config.mqtt.host, self.config.mqtt.port, self.config.mqtt.keepalive_s) + self.client.loop_start() + + def _arm_will(self) -> None: + """Point the will at the current last-known truth. + + The will lives in the CONNECT packet, so the broker holds a fixed payload for + the whole session and no amount of calling this changes what fires if *this* + connection drops. paho rebuilds CONNECT from these fields on every reconnect + though, so re-arming still moves the staleness window from "whenever the + process started" -- potentially weeks -- down to "the last reconnect", which + here is a few hours at most. Called from ``_on_connect``, i.e. from paho's own + network thread, which is also the thread that builds the packet, so there is + no window where the payload and the topic disagree. + """ self.client.will_set( self.config.topic("status"), self._disconnected().model_dump_json(), qos=1, retain=True ) - self.client.connect(self.config.mqtt.host, self.config.mqtt.port, self.config.mqtt.keepalive_s) - self.client.loop_start() def stop(self) -> None: try: @@ -133,6 +146,7 @@ def _on_connect(self, client, userdata, flags, reason_code, properties=None) -> # retained message that was correct a moment earlier, and the printer is asleep # and cannot be asked again. self.publish_status(self._last_status) + self._arm_will() # for the connect after this one; see _arm_will def _on_message(self, client, userdata, msg) -> None: if msg.topic.endswith("/cmd"): diff --git a/tests/test_source_mqtt.py b/tests/test_source_mqtt.py index 691f7b5..6eb527b 100644 --- a/tests/test_source_mqtt.py +++ b/tests/test_source_mqtt.py @@ -166,6 +166,25 @@ def test_the_will_carries_last_known_truth(source): assert will["device_seen_at"] is not None +def test_the_will_is_re_armed_so_it_does_not_describe_startup_forever(source): + """The will is fixed for the life of a connection -- it rides in the CONNECT packet + and the broker holds it. paho rebuilds that packet on every reconnect, though, so + re-arming bounds how stale the will can be by the reconnect interval rather than by + the process lifetime.""" + src = source() + src.start() + assert json.loads(src.client.will[1])["serial"] is None # nothing known at boot + + _connect(src) + src.publish_status(KNOWN.to_status("d30-workshop", state="idle")) + _connect(src) # a later reconnect: this is where the newer will gets armed + + will = json.loads(src.client.will[1]) + assert will["state"] == "disconnected" + assert will["serial"] == SERIAL + assert will["device_seen_at"] is not None + + def test_shutdown_says_disconnected_without_forgetting(source): src = source(KNOWN) src.stop() From a3f0556c240a76f04a8f237f4e2f3fe9f99680fd Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sun, 2 Aug 2026 16:33:34 +0200 Subject: [PATCH 3/3] test(agent): pin that a restart corrects the stale will it followed Review on the consumer PR pointed out that a fault learned mid-session cannot reach the will -- the broker holds it from CONNECT -- so an agent killed after a media fault publishes the older, healthier reading over the newer one. Nothing prevents that, and neither suggested remedy helps: on_disconnect fires with no connection to publish on, and a SIGKILL runs no callback at all. What bounds it is recovery. The unit is Restart=always with RestartSec=5, and a restarted agent seeds _last_status from the spool -- which does hold the fault -- and republishes on connect, so the stale row is seconds wide rather than indefinite. That property was load-bearing in the argument and untested, which is a bad combination. Now asserted end to end: arm a healthy will, learn a fault mid-session, restart on the same spool, and check the republished status carries the fault and a device_seen_at strictly newer than the will's. --- tests/test_source_mqtt.py | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/tests/test_source_mqtt.py b/tests/test_source_mqtt.py index 6eb527b..39ffede 100644 --- a/tests/test_source_mqtt.py +++ b/tests/test_source_mqtt.py @@ -185,6 +185,41 @@ def test_the_will_is_re_armed_so_it_does_not_describe_startup_forever(source): assert will["device_seen_at"] is not None +def test_a_stale_will_is_corrected_by_the_restart_that_follows_it(tmp_path, monkeypatch): + """The will is frozen at connect, so a fault learned mid-session cannot reach it: if + the agent is killed after that fault, the broker publishes the older, healthier + reading over the newer one. Nothing can change what the broker holds for a live + session -- what stops it mattering is that the next process seeds from the spool, + which does have the fault, and republishes on connect. The unit is Restart=always + with RestartSec=5, so that correction is seconds behind the will.""" + import paho.mqtt.client as mqtt + + monkeypatch.setattr(mqtt, "Client", FakeClient) + config = make_config() + config.mqtt.host = "broker.invalid" + + spool = Spool(tmp_path / "spool.db") + spool.save_device_snapshot(KNOWN) + healthy = MqttSource(config, spool, lambda _j: None) + healthy.start() + assert json.loads(healthy.client.will[1])["media_ok"] is True # what a crash will say + + # Mid-session: a print finds the tape gone. The topic is correct; the will is not. + faulted = KNOWN.model_copy(update={"media_ok": False, "fault": "media not ready", "seen_at": 5000.0}) + spool.save_device_snapshot(faulted) + healthy.publish_status(faulted.to_status("d30-workshop", state="error")) + spool.close() + + # SIGKILL: the broker publishes the stale will, then systemd brings the agent back. + restarted = MqttSource(config, Spool(tmp_path / "spool.db"), lambda _j: None) + _connect(restarted) + + corrected = restarted.client.statuses[-1] + assert corrected["media_ok"] is False + assert corrected["error"] == "media not ready" + assert corrected["device_seen_at"] > json.loads(healthy.client.will[1])["device_seen_at"] + + def test_shutdown_says_disconnected_without_forgetting(source): src = source(KNOWN) src.stop()