From dfa0d44589bdf5ad20c9d37b84b846d55c9fe43c Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 11:01:25 +0200 Subject: [PATCH 01/38] Back to development: 0.1.4+dev MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A tag is the only thing that is a version. Between two of them the tree is a build nobody released, and until now it went on claiming to *be* the release it came after — which is how 0.1.4 came to name both the build before the per-node digest and the build after, with the mismatch messages unable to tell them apart. `+dev` is a local version: it says "0.1.4, plus changes", it does not have to guess what the next release will be numbered, and PyPI refuses one, so a dev build cannot be published by accident. `scripts/release.sh` in the workspace root writes it after each tag; this is that commit, by hand, for the release that predates it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01A9Hdrmf2cwNABCnE5x9UJa --- backend/pyproject.toml | 2 +- uv.lock | 4 ++-- worker/pyproject.toml | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 3ea0651..a5de93d 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "fluksio" -version = "0.1.4" +version = "0.1.4+dev" description = "Node-based automation engine: flows, dashboards, batch runs" readme = "README.md" license = "AGPL-3.0-or-later" diff --git a/uv.lock b/uv.lock index d22124d..8d79503 100644 --- a/uv.lock +++ b/uv.lock @@ -869,7 +869,7 @@ wheels = [ [[package]] name = "fluksio" -version = "0.1.4" +version = "0.1.4+dev" source = { editable = "backend" } dependencies = [ { name = "aiomqtt" }, @@ -955,7 +955,7 @@ dev = [ [[package]] name = "fluksio-worker" -version = "0.1.4" +version = "0.1.4+dev" source = { editable = "worker" } dependencies = [ { name = "websockets" }, diff --git a/worker/pyproject.toml b/worker/pyproject.toml index e5f2979..2f3c0de 100644 --- a/worker/pyproject.toml +++ b/worker/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "fluksio-worker" -version = "0.1.4" +version = "0.1.4+dev" description = "Runs Fluksio nodes on a machine the engine cannot reach" readme = "README.md" license = "AGPL-3.0-or-later" From e18a669aa0cd47d5ef3404dbb4896cfbaed534c6 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 11:02:59 +0200 Subject: [PATCH 02/38] Write a run's elapsed time once, in the duration formatter's units The runs table drew when a run started and how long it took in two notations ('56m ago | 55.5 min'), and for a run still going those are the same reading twice. `dur` now steps past the minute into hours and days, the ago reading is derived from it, and a running run writes only the elapsed one. --- frontend/src/components/Runs/RunsScreen.tsx | 27 +++++++++++++++------ frontend/src/lib/utils.check.ts | 2 ++ frontend/src/lib/utils.ts | 5 ++++ 3 files changed, 26 insertions(+), 8 deletions(-) diff --git a/frontend/src/components/Runs/RunsScreen.tsx b/frontend/src/components/Runs/RunsScreen.tsx index 5156c89..a2f2110 100644 --- a/frontend/src/components/Runs/RunsScreen.tsx +++ b/frontend/src/components/Runs/RunsScreen.tsx @@ -6,7 +6,6 @@ import { useRef } from "react" import type { fluksio__api__routes__runs__RunRow as RunRow } from "@/client" import { MAX_SERIES } from "@/components/Common/UplotChart" import { ValuePreview } from "@/components/Flow/ValuePreview" -import { ago } from "@/components/Health/queries" import { Button } from "@/components/ui/button" import { Checkbox } from "@/components/ui/checkbox" import { @@ -296,6 +295,12 @@ function ParamValue({ value }: { value: unknown }) { return {paramText(value)} } +/** How long ago a moment was, in `dur`'s units so a row's two times agree. */ +function since(ts: unknown): string { + if (!ts) return "—" + return `${dur(Date.now() - Date.parse(String(ts)))} ago` +} + function RunsTable({ runs, search, @@ -456,14 +461,20 @@ function RunsTable({ only the store's, which is then the whole answer. */} {shortCommit(run.origin_commit || run.commit || "") || "—"} - + {/* When it started, and how long it then took. Two readings of - one thing, so one column rather than two. */} - {ago(String(run.created_at ?? ""))} - | - - {run.duration_ms ? dur(run.duration_ms) : "—"} - + one thing, so one column rather than two — and while a run is + still going they are the same reading, so it is written once: + how long it has been running is also how long ago it began. */} + {run.status === "running" ? ( + `${dur(run.duration_ms)} ago` + ) : ( + <> + {since(run.created_at)} + | + {run.duration_ms ? dur(run.duration_ms) : "—"} + + )} ))} diff --git a/frontend/src/lib/utils.check.ts b/frontend/src/lib/utils.check.ts index a0c7d84..5e46fc5 100644 --- a/frontend/src/lib/utils.check.ts +++ b/frontend/src/lib/utils.check.ts @@ -38,6 +38,8 @@ assert.equal(dur(0), "0 ms") assert.equal(dur(203.63), "204 ms") assert.equal(dur(1240), "1.24 s") assert.equal(dur(90_000), "1.5 min") +assert.equal(dur(12_600_000), "3.5 h") +assert.equal(dur(259_200_000), "3 d") // The plain window down to 0.01 ms, so a fast run keeps one unit across polls. assert.equal(dur(0.4), "0.4 ms") diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index acaca2a..3969708 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -44,6 +44,8 @@ export function si(value: number, digits = 3): string { /** Floor, divisor and unit per step, largest first. */ const DUR_STEPS: [number, number, string][] = [ + [86_400_000, 86_400_000, "d"], + [3_600_000, 3_600_000, "h"], [60_000, 60_000, "min"], [1_000, 1_000, "s"], [0.01, 1, "ms"], @@ -63,6 +65,9 @@ const DUR_STEPS: [number, number, string][] = [ * window transposed onto units. That floor is deliberate: without it a tile * polled every ten seconds flips between "400 µs" and "1.2 ms", and a reading * that changes unit on every poll reads as broken rather than as fast. + * + * It keeps stepping past the minute into hours and days, so that an elapsed + * time is one notation wherever it is written — "3.5 h", never "210 min". */ export function dur(ms: number, digits = 3): string { if (!Number.isFinite(ms)) return "--" From 7e8fd1182e1b6f202bf6f44e86aa47c41f83c7b7 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 11:19:30 +0200 Subject: [PATCH 03/38] Put the dashboard panels away on a click off the widgets Mirrors the flow editor's onPaneClick: clicking the canvas margin or the empty surface closes the widget settings and the dashboard panel. Widget frames, grid resize handles and menus portalled out of the canvas keep it open. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- .../components/Dashboard/DashboardEditor.tsx | 22 +++++++++++++++++++ frontend/tests/editor.spec.ts | 15 +++++++++++++ 2 files changed, 37 insertions(+) diff --git a/frontend/src/components/Dashboard/DashboardEditor.tsx b/frontend/src/components/Dashboard/DashboardEditor.tsx index f66bf97..1b6a01e 100644 --- a/frontend/src/components/Dashboard/DashboardEditor.tsx +++ b/frontend/src/components/Dashboard/DashboardEditor.tsx @@ -144,6 +144,14 @@ const AUTOSAVE_MS = 800 const INTERACTIVE = "button, a, input, select, textarea, [role='switch'], [role='combobox'], [role='slider'], .react-resizable-handle, .widget-grip" +/** + * What a click may land on without counting as a click off the widgets: the + * widget itself, and the handles the grid draws around it — resizing a tile is + * still working on it. The grid item covers both; the frame is what a stacked + * dashboard has instead. + */ +const ON_WIDGET = "[data-testid=widget-frame], .react-grid-item" + function nextId(dashboard: DashboardDef_Output, type: string): string { const taken = new Set(widgetsOf(dashboard).map((widget) => widget.id)) let candidate = type @@ -524,12 +532,26 @@ export function DashboardEditor({ return ( + {/* Clicking off the widgets is how you put a panel away, as it is on the + flow pane — the dashboard is what you went back to look at. A menu a + widget portals to `body` lands outside this element but still bubbles + through React's tree, so only a press that landed in the subtree + counts. */} + {/* biome-ignore lint/a11y/useKeyWithClickEvents: Escape closes the panel; this is a pointer shortcut, not a control. */} + {/* biome-ignore lint/a11y/noStaticElementInteractions: see above. */}
{ + const target = event.target as Element + if (!event.currentTarget.contains(target)) return + if (target.closest(ON_WIDGET)) return + setSelected(null) + setSettingsOpen(false) + }} > {body}
diff --git a/frontend/tests/editor.spec.ts b/frontend/tests/editor.spec.ts index 23ee60c..9cc1a86 100644 --- a/frontend/tests/editor.spec.ts +++ b/frontend/tests/editor.spec.ts @@ -135,6 +135,21 @@ test("the drag handle moves a widget without selecting it", async ({ await expect(settings).toBeVisible() }) +test("clicking off the widgets puts the settings panel away", async ({ + page, +}) => { + await openEditor(page) + + const settings = page.getByTestId("widget-settings") + await page.getByTestId("widget-frame").filter({ hasText: "Top" }).click() + await expect(settings).toBeVisible() + + // The margin around the canvas is not a widget, the way the flow pane is not + // a node. + await page.getByTestId("dashboard-canvas").click({ position: { x: 8, y: 8 } }) + await expect(settings).toBeHidden() +}) + test("a widget without its title can still be dragged and picked", async ({ page, }) => { From 93e45274d03598573e45915c63b0afb5ca8c7c1b Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 11:27:01 +0200 Subject: [PATCH 04/38] Let a teardown's cancellation through, and reap the workers it leaves MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Node stop paths cancelled their background task and then caught CancelledError around the await, which swallows a cancellation aimed at the caller — the trap Supervisor._cancel already documents. One shared Node._cancel_task now waits the way the supervisor does; mqtt's publisher and subscription and delay's cron call it. The api container also collected zombie python workers: orphaned when --reload replaces the process holding their handle, they reparent onto a PID 1 that reaps nothing but its own. `init: true` on the backend service. --- backend/fluksio/flow/nodes/base.py | 16 ++++++++++++ backend/fluksio/flow/nodes/delay.py | 6 +---- backend/fluksio/flow/nodes/mqtt.py | 12 ++------- backend/tests/flow/test_node_teardown.py | 31 ++++++++++++++++++++++++ docker/compose.yml | 7 ++++++ 5 files changed, 57 insertions(+), 15 deletions(-) create mode 100644 backend/tests/flow/test_node_teardown.py diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index e959070..b46a57d 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -212,6 +212,22 @@ class Node: return None return asyncio.create_task(factory()) + @staticmethod + async def _cancel_task(task: asyncio.Task[None]) -> None: + """Stop an unsupervised loop and wait for it to be gone. + + `wait` keeps whatever the task raises on its way out to itself, and + lets a cancellation aimed at *this* coroutine through — the + `except CancelledError` around `await task` it replaces swallowed that, + which left whoever asked for the teardown unkillable. The same trap + `Supervisor._cancel` documents. + """ + task.cancel() + await asyncio.wait([task]) + if not task.cancelled(): + # Retrieved so a crash on the way out is not reported at exit. + task.exception() + def report_health(self, status: str, detail: str | None = None) -> None: """Say how this node's connection is doing: ok, degraded or down.""" if self._on_health is not None: diff --git a/backend/fluksio/flow/nodes/delay.py b/backend/fluksio/flow/nodes/delay.py index ac3698a..0cc1cf3 100644 --- a/backend/fluksio/flow/nodes/delay.py +++ b/backend/fluksio/flow/nodes/delay.py @@ -211,11 +211,7 @@ class DelayNode(Node): self._stop_cron.set() if self._cron_task is not None: - self._cron_task.cancel() - try: - await self._cron_task - except asyncio.CancelledError: - pass + await self._cancel_task(self._cron_task) self._cron_task = None self._stop_cron = None diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py index f7d9ef2..f936ca8 100644 --- a/backend/fluksio/flow/nodes/mqtt.py +++ b/backend/fluksio/flow/nodes/mqtt.py @@ -461,11 +461,7 @@ class MqttNode(Node): if self._publish_queue is None: return if self._publisher_task is not None: - self._publisher_task.cancel() - try: - await self._publisher_task - except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down - pass + await self._cancel_task(self._publisher_task) self._publisher_task = None self._publish_queue = None self._loop = None @@ -515,11 +511,7 @@ class MqttNode(Node): self._stop_event.set() if self._subscription_task is not None: - self._subscription_task.cancel() - try: - await self._subscription_task - except asyncio.CancelledError: - pass + await self._cancel_task(self._subscription_task) self._subscription_task = None self._stop_event = None diff --git a/backend/tests/flow/test_node_teardown.py b/backend/tests/flow/test_node_teardown.py new file mode 100644 index 0000000..440ce9d --- /dev/null +++ b/backend/tests/flow/test_node_teardown.py @@ -0,0 +1,31 @@ +"""Tearing a node down must not swallow a cancellation meant for the caller.""" + +import asyncio + +import pytest + +from fluksio.flow.nodes import DelayNode + + +def test_stop_cron_lets_the_callers_cancellation_through(): + async def stubborn() -> None: + """A loop whose shutdown does not answer the first cancellation.""" + try: + await asyncio.sleep(3600) + except asyncio.CancelledError: + await asyncio.sleep(3600) + + async def scenario() -> None: + node = DelayNode(params={"cron": "* * * * *"}) + node._stop_cron = asyncio.Event() + node._cron_task = asyncio.create_task(stubborn()) + + stopping = asyncio.create_task(node.stop_cron()) + await asyncio.sleep(0.05) # let it reach the await on the cron task + stopping.cancel() + with pytest.raises(asyncio.CancelledError): + await stopping + + node._cron_task.cancel() + + asyncio.run(scenario()) diff --git a/docker/compose.yml b/docker/compose.yml index e554fbb..75f5e03 100644 --- a/docker/compose.yml +++ b/docker/compose.yml @@ -75,6 +75,13 @@ services: image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}' container_name: fluksio-api restart: always + # PID 1 that reaps orphans. A python node's worker processes outlive the + # process holding their handle whenever that one goes without stopping the + # pool -- which `--reload` does on every source edit -- and reparent onto + # PID 1, which is the app itself and waits for nobody else's children. The + # container filled up with zombie `python`. Declared here rather than in + # compose.dev.yml because an init closes the whole class, not just reload. + init: true security_opt: - no-new-privileges:true networks: From 192999f178bbccf37361adb67874957eeb511ee0 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 11:46:40 +0200 Subject: [PATCH 05/38] Bound MQTT broker operations with a per-node timeout Without one, aiomqtt's disconnect acknowledgement has no deadline, so a subscriber cancelled while its socket is dead never finishes unwinding and teardown abandons the task. The knob is per node because brokers differ. --- backend/fluksio/flow/nodes/mqtt.py | 16 ++++++++++++++++ backend/tests/flow/test_senders.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py index f936ca8..499ac2b 100644 --- a/backend/fluksio/flow/nodes/mqtt.py +++ b/backend/fluksio/flow/nodes/mqtt.py @@ -83,6 +83,8 @@ class MqttNode(Node): - ``qos`` (int): Quality of Service level 0, 1, or 2 (default: 0) - ``retain`` (bool): Retain flag for published messages (default: False) - ``keepalive`` (int): Keepalive interval in seconds (default: 60) + - ``timeout`` (float): Deadline for a broker operation in seconds + (default: 10.0) :type params: dict :param name: Optional name for the node. :type name: str | None @@ -151,6 +153,7 @@ class MqttNode(Node): "qos", "retain", "keepalive", + "timeout", "json_keys", "_topic_to_ports", "_wildcards", @@ -175,6 +178,15 @@ class MqttNode(Node): qos: int = 0 retain: bool = False keepalive: int = 60 + # Bounds every broker operation: subscribe, publish, and the + # disconnect acknowledgement on the way out. Without one a client + # whose socket died waits for that ack forever, and the task never + # finishes unwinding. Brokers differ, so it is per node. + timeout: float = Field( + default=10.0, + gt=0, + description="Give up on a broker operation after this many seconds.", + ) # Which key to lift out of a JSON object payload. A device that wraps # its reading — Victron's ``{"value": 5}`` — is otherwise a Python node # per port. One key for every port, or a per-port mapping. @@ -247,6 +259,7 @@ class MqttNode(Node): self.qos = cfg.qos self.retain = cfg.retain self.keepalive = cfg.keepalive + self.timeout = cfg.timeout # Runtime state self._subscription_task: asyncio.Task[None] | None = None @@ -372,6 +385,7 @@ class MqttNode(Node): password=self.password, identifier=self.client_id, keepalive=self.keepalive, + timeout=self.timeout, ) as client: self.report_health("ok") while True: @@ -393,6 +407,7 @@ class MqttNode(Node): password=self.password, identifier=self.client_id, keepalive=self.keepalive, + timeout=self.timeout, ) as client: await self._publish_with(client, data) @@ -556,6 +571,7 @@ class MqttNode(Node): password=self.password, identifier=self.client_id, keepalive=self.keepalive, + timeout=self.timeout, ) as client: # Subscribe to every unique topic for topic in self._topic_to_ports: diff --git a/backend/tests/flow/test_senders.py b/backend/tests/flow/test_senders.py index 0f1ee11..953d246 100644 --- a/backend/tests/flow/test_senders.py +++ b/backend/tests/flow/test_senders.py @@ -89,3 +89,33 @@ def test_a_string_goes_on_the_wire_bare(): asyncio.run(node._publish_with(client, {"plug": "ON", "level": 60})) assert client.published == [("actor/plug", "ON"), ("light/level", "60")] + + +def test_the_configured_timeout_reaches_the_broker_client(monkeypatch): + """Without one, a dead socket makes the disconnect ack wait forever.""" + import aiomqtt + + seen: dict = {} + + class FakeClient: + def __init__(self, **kwargs): + seen.update(kwargs) + + async def __aenter__(self): + return self + + async def __aexit__(self, *_): + return False + + async def publish(self, *_, **__): + return None + + monkeypatch.setattr(aiomqtt, "Client", FakeClient) + + node = MqttNode( + requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)], + params={"topic": {"setpoint": "heating/setpoint"}, "timeout": 2.5}, + ) + asyncio.run(node._publish_once({"setpoint": 21.0})) + + assert seen["timeout"] == 2.5 From f00045d6b6df96cb39566f5ba760152dbb53ea6e Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 11:52:43 +0200 Subject: [PATCH 06/38] Give the Influx and MQTT nodes their two missing knobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Influx client was built with no timeout, so every query and write fell through to influxdb-client's own 10 s default — invisible to a flow and unchangeable. The param is in seconds like its peers; the client counts in milliseconds, so the call sites convert. The publisher backlog was a module constant, read once at import. It is the depth at which the oldest payload is dropped and the node goes degraded, and a node that bursts wants more than one that trickles, so it moves to Params and is read where the queue is built. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- backend/fluksio/flow/nodes/influx.py | 32 +++++++++++++++++++-- backend/fluksio/flow/nodes/mqtt.py | 20 +++++++++---- backend/tests/flow/test_node_types.py | 41 +++++++++++++++++++++++++++ backend/tests/flow/test_senders.py | 32 +++++++++++++++++++++ 4 files changed, 117 insertions(+), 8 deletions(-) diff --git a/backend/fluksio/flow/nodes/influx.py b/backend/fluksio/flow/nodes/influx.py index 3b192a0..c0018f8 100644 --- a/backend/fluksio/flow/nodes/influx.py +++ b/backend/fluksio/flow/nodes/influx.py @@ -58,6 +58,7 @@ class InfluxDbNode(Node): - ``bucket`` (str): Bucket name (required) - ``write_precision`` (str): Write precision ("ns", "us", "ms", "s"), default "ms" - ``query_range`` (str): Default time range for queries, e.g., "-1h", "-24h" + - ``timeout`` (float): Deadline for a request in seconds (default: 10.0) - ``writes`` (dict): Write configurations keyed by message name, each with: - ``measurement`` (str): Measurement name to write to - ``field`` (str): Field name to write (default: "value") @@ -154,6 +155,7 @@ class InfluxDbNode(Node): "bucket", "write_precision", "query_range", + "timeout", "writes", "queries", "_write_client", @@ -169,6 +171,14 @@ class InfluxDbNode(Node): bucket: str write_precision: str = "ms" query_range: str = "-1h" + # Bounds every request to the server. Without one the client falls + # back to its own default, which no flow can see or change. Held in + # seconds like every other node; the client counts in milliseconds. + timeout: float = Field( + default=10.0, + gt=0, + description="Give up on a query or write after this many seconds.", + ) # Per-port write and query configuration. writes: dict[str, dict[str, Any]] = {} queries: dict[str, dict[str, Any]] = {} @@ -201,6 +211,7 @@ class InfluxDbNode(Node): self.bucket = cfg.bucket self.write_precision = cfg.write_precision self.query_range = cfg.query_range + self.timeout = cfg.timeout self.writes = cfg.writes self.queries = cfg.queries @@ -285,7 +296,12 @@ class InfluxDbNode(Node): echo = {key: value for key, value in request.items() if key != "flux"} logger.info("Running Flux for node '%s': %s", self.name, flux) - with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client: + with InfluxDBClient( + url=self.url, + token=self.token, + org=self.org, + timeout=int(self.timeout * 1000), + ) as client: tables = client.query_api().query(flux, org=self.org) rows = [ @@ -326,7 +342,12 @@ class InfluxDbNode(Node): from influxdb_client.client.write_api import SYNCHRONOUS try: - with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client: + with InfluxDBClient( + url=self.url, + token=self.token, + org=self.org, + timeout=int(self.timeout * 1000), + ) as client: write_api = client.write_api(write_options=SYNCHRONOUS) precision_map = { @@ -422,7 +443,12 @@ class InfluxDbNode(Node): results = {} try: - with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client: + with InfluxDBClient( + url=self.url, + token=self.token, + org=self.org, + timeout=int(self.timeout * 1000), + ) as client: query_api = client.query_api() for spec in self.output_ports: diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py index 499ac2b..f6b8e35 100644 --- a/backend/fluksio/flow/nodes/mqtt.py +++ b/backend/fluksio/flow/nodes/mqtt.py @@ -19,10 +19,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -# Deep enough to ride out a broker hiccup, shallow enough that a publisher -# which cannot keep up drops old values instead of growing without bound. -PUBLISH_QUEUE_SIZE = 256 - def topic_matches(filter_: str, topic: str) -> bool: """Does an MQTT topic filter cover this topic? @@ -85,6 +81,7 @@ class MqttNode(Node): - ``keepalive`` (int): Keepalive interval in seconds (default: 60) - ``timeout`` (float): Deadline for a broker operation in seconds (default: 10.0) + - ``publish_queue_size`` (int): Publisher backlog depth (default: 256) :type params: dict :param name: Optional name for the node. :type name: str | None @@ -154,6 +151,7 @@ class MqttNode(Node): "retain", "keepalive", "timeout", + "publish_queue_size", "json_keys", "_topic_to_ports", "_wildcards", @@ -187,6 +185,17 @@ class MqttNode(Node): gt=0, description="Give up on a broker operation after this many seconds.", ) + # Deep enough to ride out a broker hiccup, shallow enough that a + # publisher which cannot keep up drops old values instead of growing + # without bound. A node that bursts wants more than one that trickles. + publish_queue_size: int = Field( + default=256, + gt=0, + description=( + "How many payloads may wait for the broker. Past this the oldest " + "is dropped and the node reports degraded." + ), + ) # Which key to lift out of a JSON object payload. A device that wraps # its reading — Victron's ``{"value": 5}`` — is otherwise a Python node # per port. One key for every port, or a per-port mapping. @@ -260,6 +269,7 @@ class MqttNode(Node): self.retain = cfg.retain self.keepalive = cfg.keepalive self.timeout = cfg.timeout + self.publish_queue_size = cfg.publish_queue_size # Runtime state self._subscription_task: asyncio.Task[None] | None = None @@ -467,7 +477,7 @@ class MqttNode(Node): """Run the task that owns this node's connection to the broker.""" if self._publish_queue is not None: return - self._publish_queue = asyncio.Queue(maxsize=PUBLISH_QUEUE_SIZE) + self._publish_queue = asyncio.Queue(maxsize=self.publish_queue_size) self._loop = asyncio.get_running_loop() self._publisher_task = self._run_supervised("mqtt-out", self._publisher_loop) diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index 7cc1dc1..056593c 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -197,6 +197,47 @@ def test_a_flux_request_is_run_rather_than_written(monkeypatch): assert out == {"answer": {"rows": [], "range_s": 3600}} +def test_the_configured_timeout_reaches_the_influx_client(monkeypatch): + """The client counts in milliseconds; the param is seconds like its peers.""" + import influxdb_client + + from fluksio.flow.nodes import InfluxDbNode + + seen: dict = {} + + class FakeClient: + def __init__(self, **kwargs): + seen.update(kwargs) + + def __enter__(self): + return self + + def __exit__(self, *_): + return False + + def query_api(self): + return self + + def query(self, *_, **__): + return [] + + monkeypatch.setattr(influxdb_client, "InfluxDBClient", FakeClient) + + node = InfluxDbNode( + provides=[MessageSpec(name="answer", dtype=DType.JSON)], + params={ + "url": "http://influx", + "token": "t", + "org": "o", + "bucket": "b", + "timeout": 2.5, + }, + ) + node._run_flux({"flux": 'from(bucket: "b")'}) + + assert seen["timeout"] == 2500 + + def test_a_falsy_return_is_a_mistake_not_silence(): """Only None means "nothing to publish".""" from fluksio.flow.nodes import Node diff --git a/backend/tests/flow/test_senders.py b/backend/tests/flow/test_senders.py index 953d246..655066a 100644 --- a/backend/tests/flow/test_senders.py +++ b/backend/tests/flow/test_senders.py @@ -119,3 +119,35 @@ def test_the_configured_timeout_reaches_the_broker_client(monkeypatch): asyncio.run(node._publish_once({"setpoint": 21.0})) assert seen["timeout"] == 2.5 + + +def test_the_configured_backlog_reaches_the_publish_queue(monkeypatch): + """The depth is read when the queue is built, so it has to be per node.""" + import aiomqtt + + class FakeClient: + def __init__(self, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *_): + return False + + monkeypatch.setattr(aiomqtt, "Client", FakeClient) + + node = MqttNode( + requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)], + params={"topic": {"setpoint": "heating/setpoint"}, "publish_queue_size": 8}, + ) + node.assign_flow("heating", "out") + + async def scenario() -> int: + await node.start_publisher() + assert node._publish_queue is not None + size = node._publish_queue.maxsize + await node.stop_publisher() + return size + + assert asyncio.run(scenario()) == 8 From f5ea960e24fce44f03f522cd58f6bec50732af3a Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 11:56:19 +0200 Subject: [PATCH 07/38] Let a connector's teardown cancellation through too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ConnectorNode.stop cancelled its poll task and then caught CancelledError around the await — the fourth site of the trap 93e4527 closed elsewhere, swallowing a cancellation aimed at whoever asked for the teardown. It now calls the shared Node._cancel_task, which keeps retrieving whatever the loop raised on its way out, as the old `except (CancelledError, Exception)` did. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- backend/fluksio/flow/connector.py | 6 +---- backend/tests/flow/test_node_teardown.py | 33 +++++++++++++++++++----- 2 files changed, 27 insertions(+), 12 deletions(-) diff --git a/backend/fluksio/flow/connector.py b/backend/fluksio/flow/connector.py index ac49e4b..18f2e77 100644 --- a/backend/fluksio/flow/connector.py +++ b/backend/fluksio/flow/connector.py @@ -161,11 +161,7 @@ class ConnectorNode(Node): return self._stop_event.set() if self._poll_task is not None: - self._poll_task.cancel() - try: - await self._poll_task - except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down - pass + await self._cancel_task(self._poll_task) self._poll_task = None self._stop_event = None self._last_published = {} diff --git a/backend/tests/flow/test_node_teardown.py b/backend/tests/flow/test_node_teardown.py index 440ce9d..a1b284c 100644 --- a/backend/tests/flow/test_node_teardown.py +++ b/backend/tests/flow/test_node_teardown.py @@ -4,17 +4,19 @@ import asyncio import pytest +from fluksio.flow.connector import ConnectorNode from fluksio.flow.nodes import DelayNode -def test_stop_cron_lets_the_callers_cancellation_through(): - async def stubborn() -> None: - """A loop whose shutdown does not answer the first cancellation.""" - try: - await asyncio.sleep(3600) - except asyncio.CancelledError: - await asyncio.sleep(3600) +async def stubborn() -> None: + """A loop whose shutdown does not answer the first cancellation.""" + try: + await asyncio.sleep(3600) + except asyncio.CancelledError: + await asyncio.sleep(3600) + +def test_stop_cron_lets_the_callers_cancellation_through(): async def scenario() -> None: node = DelayNode(params={"cron": "* * * * *"}) node._stop_cron = asyncio.Event() @@ -29,3 +31,20 @@ def test_stop_cron_lets_the_callers_cancellation_through(): node._cron_task.cancel() asyncio.run(scenario()) + + +def test_connector_stop_lets_the_callers_cancellation_through(): + async def scenario() -> None: + node = ConnectorNode() + node._stop_event = asyncio.Event() + node._poll_task = asyncio.create_task(stubborn()) + + stopping = asyncio.create_task(node.stop()) + await asyncio.sleep(0.05) # let it reach the await on the poll task + stopping.cancel() + with pytest.raises(asyncio.CancelledError): + await stopping + + node._poll_task.cancel() + + asyncio.run(scenario()) From 70e542ec3cb53c988ecec212dfb3b5e803943e13 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 12:22:55 +0200 Subject: [PATCH 08/38] Surface a failing connector poll as node health and a flow issue The poll loop remembered what it read rather than what it published, so a value the node could not publish counted as said: the next poll skipped it, succeeded, and health went back to ok with the port still dark. Remember it only after inject returns, and report ok last. A node reporting itself down is now derived into its flow's issues on read and counted on the health summary, so the canvas marks it and Home says so. Being down does not stop the flow, and the issue clears by itself when the node reports well again. The repeating poll warning is logged once per outage rather than once per tick. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- backend/fluksio/api/routes/observability.py | 12 +++++- backend/fluksio/flow/connector.py | 23 +++++++--- backend/fluksio/flow/controller.py | 25 ++++++++++- backend/fluksio/flow/pipeline.py | 1 + .../tests/api/routes/test_observability.py | 31 ++++++++++++++ backend/tests/flow/test_connector.py | 34 +++++++++++++++ .../tests/flow/test_unhealthy_node_issue.py | 42 +++++++++++++++++++ docs/concepts/flows.md | 9 ++-- docs/interface/flow-editor.md | 7 +++- docs/reference/connector-contract.md | 11 +++-- frontend/src/client/schemas.gen.ts | 2 +- frontend/src/client/types.gen.ts | 4 +- 12 files changed, 182 insertions(+), 19 deletions(-) create mode 100644 backend/tests/flow/test_unhealthy_node_issue.py diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index f578563..54b6176 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -176,10 +176,16 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: paused = set(controller.paused_flows()) entries = list(controller.loaded.values()) errored = [e for e in entries if e.status is NodeStatus.ERROR] + unhealthy = [e for e in entries if e.health == "down"] if quarantined: problems.append(f"{len(quarantined)} flow(s) quarantined") if errored: problems.append(f"{len(errored)} node(s) failed to load") + if unhealthy: + problems.append( + f"{len(unhealthy)} node(s) down: " + f"{', '.join(sorted(e.id for e in unhealthy))}" + ) # What the canvas flags on a flow — a dependency loop, an input nothing # feeds — stops that flow running just as surely as a node that will not @@ -219,7 +225,11 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: "quarantined": len(quarantined), "invalid": len(invalid), }, - nodes={"total": len(entries), "error": len(errored)}, + nodes={ + "total": len(entries), + "error": len(errored), + "unhealthy": len(unhealthy), + }, queue=queue, loop_lag=( watchdog.snapshot() diff --git a/backend/fluksio/flow/connector.py b/backend/fluksio/flow/connector.py index 18f2e77..1cfe897 100644 --- a/backend/fluksio/flow/connector.py +++ b/backend/fluksio/flow/connector.py @@ -86,7 +86,14 @@ class ConnectorNode(Node): description="Seconds between polls; 0 polls never.", ) - __slots__ = ("config", "_poll_task", "_stop_event", "_last_published", "_artifacts") + __slots__ = ( + "config", + "_poll_task", + "_stop_event", + "_last_published", + "_artifacts", + "_down", + ) def __init__(self, **kwargs: Any) -> None: super().__init__(f=self._dispatch, **kwargs) @@ -95,6 +102,7 @@ class ConnectorNode(Node): self._stop_event: asyncio.Event | None = None self._last_published: dict[str, Any] = {} self._artifacts: ArtifactStore | None = None + self._down = False def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None: """The scheduler's entry point. Settings are already on ``self.config``.""" @@ -171,24 +179,29 @@ class ConnectorNode(Node): Only changed ports are published: a device polled every few seconds is usually saying the same thing, and every publication wakes everything - downstream of it. + downstream of it. What is remembered is what was *published*, not what + the poll returned — a publication that raised is retried next tick + rather than counting as said. """ while not (self._stop_event and self._stop_event.is_set()): try: values = await self.poll() - self.report_health("ok") changed = { port: value for port, value in (values or {}).items() if self._last_published.get(port, object()) != value } if changed: - self._last_published.update(changed) # inject runs the graph, which is blocking work. await asyncio.to_thread(self.inject, changed) + self._last_published.update(changed) + self.report_health("ok") + self._down = False except asyncio.CancelledError: break except Exception as exc: - logger.warning("Connector '%s' failed to poll: %s", self.id, exc) + if not self._down: + logger.warning("Connector '%s' failed to poll: %s", self.id, exc) + self._down = True self.report_health("down", f"{type(exc).__name__}: {exc}") await asyncio.sleep(self.config.poll_interval) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index d883416..e05d8d9 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -1224,8 +1224,31 @@ class FlowController: } ) + def _health_issues(self, flow: str | None = None) -> list[ValidationIssue]: + """Nodes that are running but not working, as issues on their flow. + + Not part of `self.issues`: that list is what a build found, and this is + what is happening now. Derived on read, so a node reporting itself well + again clears it with nothing to remember. + """ + return [ + ValidationIssue( + code="node_unhealthy", + message=( + f"Node '{entry.id.rpartition('.')[2]}' is down: " + f"{entry.health_detail or 'no detail given'}" + ), + flow=entry.flow, + node=entry.id, + ) + for entry in self.loaded.values() + if entry.health == "down" and (flow is None or entry.flow == flow) + ] + def flow_issues(self, flow: str) -> list[ValidationIssue]: - return [issue for issue in self.issues if not issue.flow or issue.flow == flow] + return [ + issue for issue in self.issues if not issue.flow or issue.flow == flow + ] + self._health_issues(flow) def preview(self, name: str) -> Preview: """Build a flow's unpublished draft without deploying it. diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index be80ea0..a2c82d8 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -59,6 +59,7 @@ class ValidationIssue(BaseModel): "unauthenticated_hook", "self_loop_needs_initial", "missing_source", + "node_unhealthy", ] message: str flow: str = "" diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index 010714f..088b7c6 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -127,6 +127,37 @@ def test_a_flow_that_cannot_run_makes_the_summary_degraded( assert not any("hooky" in problem for problem in body["problems"]) +def test_a_down_node_makes_the_summary_degraded( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A connector that cannot reach its device is not a flow that cannot run. + + It is counted on its own, so the flow keeps running and "invalid" stays + about validation. + """ + from fluksio.flow.controller import LoadedNode + + controller = client.app.state.flow_controller + before = controller.loaded + controller.loaded = { + "house.owm": LoadedNode( + id="house.owm", + flow="house", + health="down", + health_detail="ConnectionError: name resolution failed", + ) + } + try: + body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json() + finally: + controller.loaded = before + + assert body["status"] == "degraded" + assert body["nodes"]["unhealthy"] == 1 + assert any("down" in problem for problem in body["problems"]) + assert body["flows"]["invalid"] == 0 + + def test_the_history_reads_back( client: TestClient, superuser_token_headers: dict[str, str], db: Session ) -> None: diff --git a/backend/tests/flow/test_connector.py b/backend/tests/flow/test_connector.py index ed44191..4abde58 100644 --- a/backend/tests/flow/test_connector.py +++ b/backend/tests/flow/test_connector.py @@ -104,6 +104,40 @@ def test_a_failing_poll_reports_down_and_keeps_going(): assert health[-1][0] == "ok" +def test_an_undeclared_port_keeps_failing_until_the_node_declares_it(): + """A publication that raised is retried, not remembered as published. + + The loop remembers what it published. If it remembered what it read, a + value the node cannot publish would be skipped on the next poll, the poll + would succeed, and the node would go back to reporting itself healthy with + its port still dark. + """ + + class Chatty(Sensor): + """Reads a port it never declared.""" + + async def poll(self) -> dict[str, Any]: + self.polls += 1 + return {"reading": 21.5, "lat": 48.1} + + health: list[tuple[str, str | None]] = [] + node = Chatty( + provides=[MessageSpec(name="reading", dtype=DType.FLOAT)], + params={"poll_interval": 0.01}, + ) + node.assign_flow("demo", "sensor") + node._on_health = lambda _node, status, detail: health.append((status, detail)) + pipeline = Pipeline(nodes=[node]) + + run_briefly(node) + + assert pipeline.state.get("demo.reading") is None + assert health[-1][0] == "down" + assert "NodeOutputError" in (health[-1][1] or "") + # Still failing on the last poll, not just the first. + assert len([entry for entry in health if entry[0] == "down"]) > 1 + + class Actuator(ConnectorNode): """A connector that commands something instead of reading it.""" diff --git a/backend/tests/flow/test_unhealthy_node_issue.py b/backend/tests/flow/test_unhealthy_node_issue.py new file mode 100644 index 0000000..45e44fa --- /dev/null +++ b/backend/tests/flow/test_unhealthy_node_issue.py @@ -0,0 +1,42 @@ +"""A node that loaded but is not working shows up as an issue on its flow. + +Health used to go nowhere: the connector reported it, the controller stored it, +and no screen ever asked. These cover the derivation that closes that gap. +""" + +from pathlib import Path + +from fluksio.flow.controller import FlowController, LoadedNode +from fluksio.flow.store import FlowStore + + +def a_controller(tmp_path: Path) -> FlowController: + controller = FlowController(FlowStore(tmp_path / "flows")) + controller.loaded["house.owm"] = LoadedNode( + id="house.owm", + flow="house", + health="down", + health_detail="ConnectionError: name resolution failed", + ) + return controller + + +def test_a_down_node_is_an_issue_on_its_flow(tmp_path: Path) -> None: + controller = a_controller(tmp_path) + + issues = controller.flow_issues("house") + + assert [issue.code for issue in issues] == ["node_unhealthy"] + assert issues[0].node == "house.owm" + assert "name resolution failed" in issues[0].message + # Not advisory: the canvas has to mark the node. + assert not issues[0].advisory + assert controller.flow_issues("other") == [] + + +def test_the_issue_clears_when_the_node_reports_itself_well(tmp_path: Path) -> None: + controller = a_controller(tmp_path) + + controller.loaded["house.owm"].health = "ok" + + assert controller.flow_issues("house") == [] diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index a73db87..9aae75f 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -163,10 +163,13 @@ to: | `self_loop_needs_initial` | a node reads a message it also writes, with no starting value | | `node_error` | the node's code did not load: a syntax error, a missing import | | `unauthenticated_hook` | advisory — a webhook with no shared secret is open to anyone | +| `node_unhealthy` | the node loaded but is not working: a connector that cannot reach its device, or whose last publication failed | -A flow with any of these except the advisory one does not run. The health -summary on Home counts them, so "why is nothing happening?" has an answer that -does not involve reading logs. +A flow with any of these except the advisory one and `node_unhealthy` does not +run — a node reporting itself down is a live condition, not a build error, so +the rest of the flow keeps going and the issue clears by itself once the node +reports well again. The health summary on Home counts them, so "why is nothing +happening?" has an answer that does not involve reading logs. ## What happens at runtime diff --git a/docs/interface/flow-editor.md b/docs/interface/flow-editor.md index 675a091..1536341 100644 --- a/docs/interface/flow-editor.md +++ b/docs/interface/flow-editor.md @@ -146,9 +146,12 @@ The canvas validates as you edit and marks the node each issue belongs to: - a node reading a message it also writes, with nothing to start it from - code that did not load - a webhook with no shared secret (advisory — it does not stop the flow) +- a node that loaded but reports itself down, such as a connector that cannot + reach its device -A flow with any of these except the last does not run, and the health summary -on Home counts it. +A flow with any of these except the last two does not run, and the health +summary on Home counts it. The last one clears on its own once the node reports +itself well again. ## See also diff --git a/docs/reference/connector-contract.md b/docs/reference/connector-contract.md index 93cfec4..288062a 100644 --- a/docs/reference/connector-contract.md +++ b/docs/reference/connector-contract.md @@ -131,7 +131,9 @@ async def poll(self) -> dict[str, Any] | None: - Return `None` when there is nothing new. - **Only changed values are published.** A device polled every few seconds usually says the same thing, and every publication wakes everything - downstream, so the loop compares against what it last published. + downstream, so the loop compares against what it last published — what it + actually published, so a publication that failed is retried next tick rather + than counting as said. - Raising is not fatal: it is reported as a health problem and retried on the next tick. - The loop calls `inject`, which runs the graph, on a worker thread. `poll()` @@ -213,9 +215,10 @@ self.report_health("degraded", "3 of 5 registers timed out") self.report_health("down", str(exc)) ``` -Three values, `ok`, `degraded` and `down`, plus an optional detail string. The -engine forwards changes to the editor, which shows them on the node. Reporting -the same status twice is free — only changes are published. The polling loop +Three values, `ok`, `degraded` and `down`, plus an optional detail string. +Reporting the same status twice is free — only changes are published. A node +reporting `down` is named among its flow's issues and counted on the health +summary on Home; `degraded` means still working, and is not. The polling loop already reports around `poll()`; a connector managing its own connection should report when it connects and when it loses the connection. diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 93974ba..f2181a6 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -3561,7 +3561,7 @@ export const ValidationIssueSchema = { properties: { code: { type: 'string', - enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source'], + enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source', 'node_unhealthy'], title: 'Code' }, message: { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 2069fc4..05b0a4d 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -1207,7 +1207,7 @@ export type ValidationError = { * Something wrong with a flow — a fault, or merely advisory. */ export type ValidationIssue = { - code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source'; + code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy'; message: string; flow?: string; nodes?: Array<(string)>; @@ -1220,7 +1220,7 @@ export type ValidationIssue = { readonly advisory: boolean; }; -export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source'; +export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy'; export type ValidationResult = { issues?: Array; From 7600aaf7eace3a86d54fa60d083cbf9adcfae2f1 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 12:59:08 +0200 Subject: [PATCH 09/38] Guard what a dashboard save does not change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `PUT /dashboards/{name}` had no backend test: three cover what it promises — a first draft for a name nobody has used (200, not the 404 that came off in 4a2337f), an update that keeps everything the edit did not name, and a save based on a version someone moved past. The Playwright half is the same property through the editor. Pages and sections are gone since 7ff29ca, so what the editor never draws — a widget's md/sm placements and the dashboard-wide settings — is what a save has to carry, and dragging one widget is what the spec makes it carry it through. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- backend/tests/api/routes/test_dashboards.py | 88 +++++++++++++++ frontend/tests/persistence.spec.ts | 118 ++++++++++++++++++++ 2 files changed, 206 insertions(+) create mode 100644 backend/tests/api/routes/test_dashboards.py create mode 100644 frontend/tests/persistence.spec.ts diff --git a/backend/tests/api/routes/test_dashboards.py b/backend/tests/api/routes/test_dashboards.py new file mode 100644 index 0000000..609c376 --- /dev/null +++ b/backend/tests/api/routes/test_dashboards.py @@ -0,0 +1,88 @@ +"""Dashboards over HTTP: saving one, which is also how the first one is made.""" + +from fastapi.testclient import TestClient + +from fluksio.core.config import settings + +PREFIX = f"{settings.API_V1_STR}/dashboards" + + +def a_dashboard(name: str) -> dict: + return { + "name": name, + "title": "Hall", + "icon": "gauge", + "widgets": [ + { + "id": "temperature", + "type": "stat", + "title": "Temperature", + "layout": {"lg": {"x": 0, "y": 0, "w": 3, "h": 2}}, + "config": {"message": "house.temperature", "dtype": "float"}, + } + ], + "settings": {"theme": {"value": "dark", "message": "", "dtype": "str"}}, + "version": 0, + } + + +def test_a_save_creates_a_dashboard_that_does_not_exist_yet( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A first draft, not a 404: creating one *is* saving it at version 0.""" + body = a_dashboard("put_creates") + + response = client.put( + f"{PREFIX}/put_creates", headers=superuser_token_headers, json=body + ) + + assert response.status_code == 200, response.text + saved = response.json() + assert saved["version"] == 1 and saved["has_draft"] is True + # A draft alone: nothing was published, so no panel can be shown it. + assert ( + client.get(f"{PREFIX}/put_creates", headers=superuser_token_headers).status_code + == 404 + ) + draft = client.get( + f"{PREFIX}/put_creates?draft=true", headers=superuser_token_headers + ) + assert draft.json()["widgets"] == body["widgets"] + + +def test_a_save_of_an_existing_dashboard_keeps_what_it_did_not_change( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + seeded = a_dashboard("put_updates") + first = client.put( + f"{PREFIX}/put_updates", headers=superuser_token_headers, json=seeded + ).json() + + renamed = {**first, "widgets": [{**first["widgets"][0], "title": "Outside"}]} + second = client.put( + f"{PREFIX}/put_updates", headers=superuser_token_headers, json=renamed + ) + + assert second.status_code == 200, second.text + stored = second.json() + assert stored["version"] == first["version"] + 1 + assert stored["widgets"][0]["title"] == "Outside" + # Everything the edit did not name is still what was first written. + assert stored["icon"] == seeded["icon"] + assert stored["settings"] == seeded["settings"] + assert stored["widgets"][0]["layout"] == seeded["widgets"][0]["layout"] + assert stored["widgets"][0]["config"] == seeded["widgets"][0]["config"] + + +def test_a_save_based_on_a_version_someone_moved_past_is_refused( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + body = a_dashboard("put_conflicts") + client.put(f"{PREFIX}/put_conflicts", headers=superuser_token_headers, json=body) + + stale = client.put( + f"{PREFIX}/put_conflicts", headers=superuser_token_headers, json=body + ) + + assert stale.status_code == 409 + assert stale.json()["detail"]["current_version"] == 1 diff --git a/frontend/tests/persistence.spec.ts b/frontend/tests/persistence.spec.ts new file mode 100644 index 0000000..6a1a1af --- /dev/null +++ b/frontend/tests/persistence.spec.ts @@ -0,0 +1,118 @@ +import { expect, type Page, test } from "@playwright/test" +import { api, apiPage, deleteAll } from "./utils/api" + +/** + * An editor save is a rewrite of the whole document, not a patch of what moved. + * + * So everything the editor never draws rides on that one request: a widget's + * placements at the narrower breakpoints, which only a panel of that width + * reads, and the dashboard-wide settings, which live in the settings panel + * rather than on the canvas. `applyLayout` keeps them by spreading the stored + * layout under the one column set it knows; a save that forgot to would lose a + * phone's arrangement with nothing on screen to show for it. + */ + +const dashboardName = `test_persist_${Date.now().toString(36)}` + +/** The tile the editor is told to move. */ +const MOVED = { + id: "moved", + type: "clock", + title: "Moved", + layout: { lg: { x: 0, y: 0, w: 3, h: 3 } }, + config: {}, +} + +/** The tile nothing touches. Its stored form is the assertion. */ +const KEPT = { + id: "kept", + type: "stat", + title: "Kept", + layout: { + lg: { x: 8, y: 0, w: 4, h: 3 }, + md: { x: 0, y: 6, w: 5, h: 3 }, + sm: { x: 0, y: 9, w: 2, h: 2 }, + }, + config: { message: "house.kept", dtype: "float", unit: "°C" }, +} + +/** Dashboard-wide, and nowhere on the canvas the drag happens on. */ +const SETTINGS = { + theme: { value: "dark", message: "", dtype: "str" }, + touch: { value: true, message: "", dtype: "bool" }, +} + +test.use({ storageState: "playwright/.auth/user.json" }) + +test.describe.configure({ mode: "serial" }) + +test.beforeAll(async ({ browser }) => { + const page = await apiPage(browser) + // Version 0 creates: a first draft is what a save of a name nobody has used + // yet means, and the editor reads the draft. + const made = await api(page, `/dashboards/${dashboardName}`, { + method: "PUT", + data: { + name: dashboardName, + title: "Persistence", + icon: "layout-dashboard", + columns: 12, + canvas_width: 1920, + canvas_height: 1080, + version: 0, + widgets: [MOVED, KEPT], + settings: SETTINGS, + }, + }) + if (!made.ok()) + throw new Error(`dashboard PUT ${made.status()}: ${await made.text()}`) + await page.close() +}) + +test.afterAll(async ({ browser }) => { + await deleteAll(browser, [`/dashboards/${dashboardName}`]) +}) + +/** The draft as stored — what a panel would be shown once it is published. */ +async function stored(page: Page) { + const response = await api(page, `/dashboards/${dashboardName}?draft=true`) + return (await response.json()) as { + icon: string + settings: unknown + widgets: { id: string; layout: Record }[] + } +} + +test("moving one widget leaves the rest of the document alone", async ({ + page, +}) => { + await page.goto(`/dashboards/${dashboardName}?edit=true`) + const moved = page.getByTestId("widget-frame").filter({ hasText: "Moved" }) + await moved.waitFor({ state: "visible", timeout: 15000 }) + + // Drag it a tile's width to the right, which is clear of `kept` at column 8. + const box = (await moved.boundingBox())! + await moved.locator(".widget-grip").hover() + await page.mouse.down() + await page.mouse.move(box.x + box.width, box.y + box.height / 4, { + steps: 10, + }) + await page.mouse.up() + + // The save is debounced, so wait for the store rather than for the canvas. + await expect + .poll( + async () => + ( + (await stored(page)).widgets.find((w) => w.id === "moved")?.layout + .lg as { x: number } + )?.x, + { message: "the drag was never saved", timeout: 10000 }, + ) + .toBeGreaterThan(0) + + const after = await stored(page) + expect(after.widgets.find((w) => w.id === "kept")).toEqual(KEPT) + expect(after.settings).toEqual(SETTINGS) + expect(after.icon).toBe("layout-dashboard") +}) From 45348d2e0ac394415c489a00eadd9400cfab81f6 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 13:37:55 +0200 Subject: [PATCH 10/38] Check the colour disc's gesture: one publish, on release The wheel's arithmetic is checked in color.check.ts; nothing checked what a drag across it sends. A new Playwright spec drags from the centre out to three o'clock, counts the panel's publishes off the wire, and holds the disc to one message at the release rather than one per pixel. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- frontend/tests/color-drag.spec.ts | 171 ++++++++++++++++++++++++++++++ 1 file changed, 171 insertions(+) create mode 100644 frontend/tests/color-drag.spec.ts diff --git a/frontend/tests/color-drag.spec.ts b/frontend/tests/color-drag.spec.ts new file mode 100644 index 0000000..ffd395c --- /dev/null +++ b/frontend/tests/color-drag.spec.ts @@ -0,0 +1,171 @@ +import { expect, test } from "@playwright/test" + +import { api, apiPage, deleteAll } from "./utils/api" + +/** + * A colour disc sends once, at the end of the gesture. + * + * Hue is the angle and saturation the radius, so setting a colour is one drag + * across one picture — and the disc follows the pointer from the moment it + * goes down. Only the release publishes: a value per pixel would flood + * whatever is listening, and on a lamp it would strobe it. + * + * The arithmetic behind the wheel is checked in `color.check.ts`, which needs + * no browser. What needs one is the gesture: how many messages a drag sends, + * and whether the one it does send is where the pointer was let go. + */ +const flowName = `test_color_${Date.now().toString(36)}` +const dashboard = `${flowName}_d` +const target = `${flowName}.tint` + +test.use({ storageState: "playwright/.auth/user.json" }) + +test.beforeAll(async ({ browser }) => { + const page = await apiPage(browser) + // A node that only declares: nothing consumes the message, so a publish + // reaches the graph without waking an engine run. The widget is what is + // under test, not a flow. + const madeFlow = await api(page, `/flows/${flowName}`, { + method: "PUT", + data: { + name: flowName, + title: "Colour drag", + version: 1, + nodes: [ + { + id: "lamp", + type: "python", + provides: [{ name: "tint", dtype: "list", item: "float" }], + }, + ], + }, + }) + if (!madeFlow.ok()) + throw new Error(`flow PUT ${madeFlow.status()}: ${await madeFlow.text()}`) + const flow = await (await api(page, `/flows/${flowName}?draft=true`)).json() + await api(page, `/flows/${flowName}/publish`, { + method: "POST", + data: { version: flow.definition.version }, + }) + + // A leftover from a run that failed before its teardown would answer the + // create with a version conflict. + await api(page, `/dashboards/${dashboard}`, { method: "DELETE" }) + const madeBoard = await api(page, `/dashboards/${dashboard}`, { + method: "PUT", + data: { + name: dashboard, + title: "Colour drag", + version: 0, + widgets: [ + { + id: "tint", + type: "color", + title: "Tint", + layout: { lg: { x: 0, y: 0, w: 6, h: 4 } }, + config: { target, dtype: "list", format: "hsv" }, + }, + ], + }, + }) + if (!madeBoard.ok()) + throw new Error( + `dashboard PUT ${madeBoard.status()}: ${await madeBoard.text()}`, + ) + const board = await madeBoard.json() + const shown = await api(page, `/dashboards/${dashboard}/publish`, { + method: "POST", + data: { version: board.version }, + }) + if (!shown.ok()) + throw new Error(`publish ${shown.status()}: ${await shown.text()}`) + await page.close() +}) + +test.afterAll(async ({ browser }) => { + await deleteAll(browser, [`/dashboards/${dashboard}`, `/flows/${flowName}`]) +}) + +test("a dragged disc publishes once, where it was let go", async ({ page }) => { + await page.goto(`/view/${dashboard}`) + const disc = page.getByTestId("color-wheel") + await disc.waitFor({ timeout: 20000 }) + // A tile fades and scales in, so where the disc is drawn is only settled + // once that has finished — and the gesture below is aimed in pixels. + await page.waitForFunction( + () => + [...document.querySelectorAll(".widget-cell")].every( + (cell) => Number(getComputedStyle(cell).opacity) > 0.99, + ), + undefined, + { timeout: 15000 }, + ) + + // Every publish this panel makes, as the browser sends it. The setup above + // talks to the API out of band, so nothing but the widget is counted here. + const sent: [number, number, number][] = [] + page.on("request", (request) => { + if ( + request.method() === "POST" && + request.url().includes(`/messages/${target}`) + ) + sent.push(request.postDataJSON()?.value) + }) + + const box = (await disc.boundingBox())! + const middle = { x: box.x + box.width / 2, y: box.y + box.height / 2 } + + // Down in the white centre and out towards three o'clock, which the wheel's + // own frame — zero degrees at twelve, running clockwise — reads as hue 90. + await page.mouse.move(middle.x, middle.y) + await page.mouse.down() + for (let step = 1; step <= 8; step++) { + await page.mouse.move(middle.x + (box.width * 0.4 * step) / 8, middle.y, { + steps: 2, + }) + } + // Long enough that a publish made mid-drag would have been seen by now. + await page.waitForTimeout(500) + expect(sent, "the disc published while it was still being dragged").toEqual( + [], + ) + + await page.mouse.up() + await expect + .poll(() => sent.length, { + timeout: 10000, + message: "letting go of the disc published nothing", + }) + .toBe(1) + + // And it carries the colour the gesture ended on rather than the one it + // started from — otherwise a disc that ignored the drag would pass. + // Loosely: a few degrees of pixel rounding is not the point, and the + // colour it started on — white, hue 0 — is nowhere near either bound. + const [hue, saturation] = sent[0] + expect( + Math.abs(hue - 90), + `let go at three o'clock, published hue ${hue}`, + ).toBeLessThan(10) + expect( + saturation, + `let go four fifths out, published saturation ${saturation}`, + ).toBeGreaterThan(60) + + // Nothing follows the release either: one gesture, one message. + await page.waitForTimeout(1000) + expect(sent, "the release published more than once").toHaveLength(1) + + // And that one message reached the graph rather than only the wire. + await expect + .poll( + async () => { + const rows = await (await api(page, "/messages/")).json() + return (rows.data ?? rows).find( + (message: { name: string }) => message.name === target, + )?.value + }, + { timeout: 10000, message: "the release never reached the engine" }, + ) + .toEqual(sent[0]) +}) From 841209a6300560c405765b6ced30e2c4a99dfbea Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 13:40:16 +0200 Subject: [PATCH 11/38] Tell mypy what the closure already knows about placer and pool Both calls sit in a closure where a narrowing of `Placer | None` and `PythonWorkerPool | None` will not carry across the function boundary. The `remote.run_on` line below them already carried the same ignore; these two close out `make lint-backend`. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- backend/fluksio/flow/controller.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index e05d8d9..1ffce99 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -916,12 +916,12 @@ class FlowController: placer, pool = self.placer, self.workers def call(**kwargs: Any) -> Any: - with placer.claim( + with placer.claim( # type: ignore[union-attr] wanted, device=device, policy=policy, node=node_id, run=run_id ) as (target, allocation): env = derive_env(wanted, allocation) if target.worker is None: - return pool.for_env(env).run( + return pool.for_env(env).run( # type: ignore[union-attr] owner, local, code, From 42048881f7668f6a8df4fe61d1a56a2dd169f3e7 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 14:20:02 +0200 Subject: [PATCH 12/38] Hold the tile title still in edit mode, and thin crowded slider ticks The grip's `-m-1 p-1` pair cancelled on three sides only: `.dui-frame-head` sets `margin-top: -0.5rem` unlayered in core.css, which beats a Tailwind utility, so the padding above went uncancelled and the title dropped 4px the moment the editor opened. The grip's geometry moves into core.css beside the head's own half step, where one declaration cancels all four sides. Slider tick labels were placed in percent with nothing measured, so five four-character labels crowded on a tile narrower than the default four columns. A ResizeObserver on the slider reports its width and every Nth label is kept, N a divisor of the interval count so the first and last stay. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk --- .../components/Dashboard/ui/core/controls.ts | 70 +++++++++++++++---- .../src/components/Dashboard/ui/core/core.css | 12 ++++ .../Dashboard/ui/fluksio/Controls.tsx | 3 +- .../Dashboard/ui/fluksio/Surfaces.tsx | 3 +- .../Dashboard/ui/glass/Controls.tsx | 3 +- .../Dashboard/ui/glass/Surfaces.tsx | 3 +- .../Dashboard/ui/material/Controls.tsx | 3 +- .../Dashboard/ui/material/Surfaces.tsx | 3 +- 8 files changed, 78 insertions(+), 22 deletions(-) diff --git a/frontend/src/components/Dashboard/ui/core/controls.ts b/frontend/src/components/Dashboard/ui/core/controls.ts index f7efcb7..2cfa0cd 100644 --- a/frontend/src/components/Dashboard/ui/core/controls.ts +++ b/frontend/src/components/Dashboard/ui/core/controls.ts @@ -5,7 +5,7 @@ * either look: the state, the keyboard and every `aria-` live here, and a * renderer only decides what it looks like while doing it. */ -import { useCallback, useId, useRef, useState } from "react" +import { useCallback, useEffect, useId, useRef, useState } from "react" import { fractionOf } from "./config" @@ -141,6 +141,32 @@ export function tickIntervals(steps: number): number { return [4, 3, 2].find((count) => Number.isInteger(steps / count)) ?? 4 } +/** + * The labels that fit the width the scale was measured in. + * + * The marks are placed in percent, which says nothing about how wide a label + * is: five four-character labels crowd on a tile narrower than the four + * columns a slider is given by default. Keep every Nth instead, with N a + * divisor of the interval count so the first and last — the two that anchor + * the range — are always among those kept. + */ +export function fitMarks( + marks: T[], + width: number, +): T[] { + const intervals = marks.length - 1 + // Not measured yet, or nothing to thin. + if (width === 0 || intervals < 2) return marks + const widest = Math.max(...marks.map((mark) => mark.label.length)) + // A 0.75rem tabular digit runs about 7px, and neighbours need a gap. + const fits = Math.floor(width / (widest * 7 + 12)) + const stride = + [1, 2, 3, 4, 5].find( + (n) => intervals % n === 0 && intervals / n + 1 <= fits, + ) ?? intervals + return marks.filter((_, index) => index % stride === 0) +} + /** * A value set by dragging, published when the handle is let go. * @@ -175,6 +201,20 @@ export function useSliderDrag({ const [draft, setDraft] = useState(null) const current = draft ?? value + // How much room the scale under the track actually has, which is what says + // how many of its labels can be drawn without them running into each other. + const trackRef = useRef(null) + const [width, setWidth] = useState(0) + useEffect(() => { + const el = trackRef.current + if (!el) return + const observer = new ResizeObserver(([entry]) => + setWidth(entry.contentRect.width), + ) + observer.observe(el) + return () => observer.disconnect() + }, []) + const release = () => { if (draft === null) return onCommit(draft) @@ -187,8 +227,23 @@ export function useSliderDrag({ // without a precision setting of its own. const digits = (String(step).split(".")[1] ?? "").length + /** The scale under the track, drawn rather than declared: no browser + * renders `