diff --git a/NOTEPAD.md b/NOTEPAD.md index a70d0de..760ae43 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -12,6 +12,29 @@ Deferring because out of scope is fine, but don't mention deferring than. ### To be sorted +- BUG/FLOW: **a cancelled request can leave `FlowController._lock` held forever.** + Seeding nineteen flows over a client that timed out mid-request left the next + `POST /flows/{name}/start` waiting on the lock indefinitely — ten minutes, until + the container was restarted. A py-spy dump showed *no* thread in the reload path, + so the coroutine that holds it is suspended at an `await` inside `reload()`, most + likely in `_teardown()` awaiting a supervised task's cancellation. Everything else + kept working — health, reads, MQTT — so the engine looked fine and only anything + needing a rebuild hung. Two things worth doing: release the lock on cancellation + (`asyncio.timeout` around the teardown, or a `finally` that cannot be skipped), and + fail a `start` that waits more than a few seconds for the lock rather than hanging. +- BUG/INFRA: **475 zombie `git` processes** in the API container after a seeding + session. `FlowStore._git` uses `subprocess.run`, which reaps its own child — these + are the `git gc --auto` daemons `git commit` spawns, reparented to PID 1 when their + parent exits. PID 1 is the FastAPI process, which never reaps orphans. Harmless at + 484 processes against a 34719 limit, but it grows with every save, and a container + that commits per keystroke will get there. Fix: `git -c gc.auto=0 commit`, since a + store that never packs is a separate (and real) concern — 1898 commits had left + 5863 loose objects and no packs. +- PERF/API: seeding nineteen flows takes two reloads each — one to publish, one to + stop — so a full rebuild of every flow in the installation runs about forty times + for one seed. It is the slowest thing about standing an installation up, and a + `PUT` that could say "published, stopped" in one call would halve it. + - FEAT/SEC: `locked` is a read-only surface, not a permission — the server accepts a publish from a panel whose dashboard says locked. Making it real means carrying the flag into `_panel_may`. diff --git a/scripts/tinyhouse/__main__.py b/scripts/tinyhouse/__main__.py index 1f4205d..c88ed71 100644 --- a/scripts/tinyhouse/__main__.py +++ b/scripts/tinyhouse/__main__.py @@ -176,7 +176,15 @@ def _check_sources(flows: list[Flow]) -> list[str]: ports = {} for spec in node.get("requires", []): port = spec.get("port") or spec["name"] - ports[port] = SHAPES.get(port, SAMPLE.get(spec["dtype"])) + sample = SHAPES.get(port, SAMPLE.get(spec["dtype"])) + # The shape a node is written against has to be one its port + # would actually accept. A payload with a list inside it is + # `json`, not `record` — and the difference only shows when a + # real answer arrives, which is far too late. + bad = _rejects(spec, sample) + if bad: + problems.append(f"{where} port '{port}': {bad}") + ports[port] = sample try: result = process(**ports, **node.get("params", {})) except Exception as exc: # noqa: BLE001 - reported, not raised @@ -283,6 +291,23 @@ def _check_widgets(known: set[str]) -> list[str]: # ── pushing it ─────────────────────────────────────────────────────────── +def _rejects(spec: dict[str, Any], sample: Any) -> str: + """Why this port would refuse the shape it is being written against.""" + if sample is None: + return "" + try: + from fluksio.flow.messages import MessageSpec + except ImportError: + return "" + try: + MessageSpec(**{k: v for k, v in spec.items() if k != "port"}).check(sample) + except TypeError as exc: + return str(exc) + except Exception: # noqa: BLE001 - a spec we cannot build says nothing + return "" + return "" + + def _check_schemas(flows: list[Flow]) -> list[str]: """Hand every node to the engine's own models before the API sees them. diff --git a/scripts/tinyhouse/screens.py b/scripts/tinyhouse/screens.py index 0abf30d..33ef73f 100644 --- a/scripts/tinyhouse/screens.py +++ b/scripts/tinyhouse/screens.py @@ -18,7 +18,13 @@ from typing import Any from .api import Flow from .sensing import _broker -BUILD = '''"""A chart's window into Flux. The database-specific half, and the only one.""" +BUILD = '''"""A chart's window into Flux. The database-specific half, and the only one. + +One script, not one per line: several `from()` statements in a single script +each produce a result called `_result`, and InfluxDB refuses a script that +names two the same. The measurements go into one filter instead, and the rows +come back tagged with which one they are. +""" BUCKET = "{bucket}" @@ -26,23 +32,19 @@ BUCKET = "{bucket}" def process(chart_request, series=()): span = int(chart_request["range_s"]) every = max(1, int(chart_request["interval_s"])) - parts = [] - for measurement in series: - parts.append( - "\\n".join( - [ - 'from(bucket: "%s")' % BUCKET, - " |> range(start: -%ds)" % span, - ' |> filter(fn: (r) => r["_measurement"] == "%s")' % measurement, - ' |> filter(fn: (r) => r["_field"] == "value")', - " |> aggregateWindow(every: %ds, fn: mean, createEmpty: false)" - % every, - ] - ) - ) + wanted = " or ".join('r["_measurement"] == "%s"' % m for m in series) + flux = "\\n".join( + [ + 'from(bucket: "%s")' % BUCKET, + " |> range(start: -%ds)" % span, + " |> filter(fn: (r) => %s)" % (wanted or "false"), + ' |> filter(fn: (r) => r["_field"] == "value")', + " |> aggregateWindow(every: %ds, fn: mean, createEmpty: false)" % every, + ] + ) return { "query": { - "flux": "\\n".join(parts), + "flux": flux, "range_s": chart_request["range_s"], "interval_s": chart_request["interval_s"], } @@ -135,9 +137,9 @@ def history(h: dict[str, Any]) -> Flow: "requires": [ {"name": f"{name}_query", "port": "query", "dtype": "record"} ], - "provides": [ - {"name": f"{name}_rows", "port": "rows", "dtype": "record"} - ], + # The answer holds a list of rows, and a record may only hold + # flat scalars. + "provides": [{"name": f"{name}_rows", "port": "rows", "dtype": "json"}], } ) flow.add( @@ -146,9 +148,7 @@ def history(h: dict[str, Any]) -> Flow: "type": "python", "title": f"{name.title()}: rows to a series", "params": {"labels": labels}, - "requires": [ - {"name": f"{name}_rows", "port": "rows", "dtype": "record"} - ], + "requires": [{"name": f"{name}_rows", "port": "rows", "dtype": "json"}], "provides": [ {"name": f"{name}_series", "port": "series", "dtype": "series"} ], @@ -296,7 +296,12 @@ def kiosk(h: dict[str, Any]) -> Flow: "dtype": "float", "trigger": False, }, - {"name": "shutters.bed_down", "port": "bed_down", "dtype": "bool"}, + { + "name": "shutters.bed_down", + "port": "bed_down", + "dtype": "bool", + "trigger": False, + }, ], "provides": [{"name": "brightness", "dtype": "int"}], }, diff --git a/scripts/tinyhouse/sensing.py b/scripts/tinyhouse/sensing.py index 1347055..8bb9e79 100644 --- a/scripts/tinyhouse/sensing.py +++ b/scripts/tinyhouse/sensing.py @@ -286,16 +286,33 @@ def power(h: dict[str, Any]) -> Flow: "id": "say", "type": "python", "title": "Worth a push?", - "requires": [{"name": "alert_changed", "port": "alert", "dtype": "record"}], - "provides": [{"name": "push", "dtype": "str"}], + "requires": [ + {"name": "alert_changed", "port": "alert", "dtype": "record"}, + { + "name": "seen_trouble", + "port": "seen", + "dtype": "bool", + "trigger": False, + }, + ], + "provides": [ + {"name": "push", "dtype": "str"}, + {"name": "seen_trouble", "port": "seen_trouble", "dtype": "bool"}, + ], }, - '''"""A level going back to normal is worth knowing; an info line is not.""" + '''"""A level going back to normal is worth knowing; the first one is not. + +Recovery only reads as recovery if something went wrong first. An engine that +has just started has nothing to recover from, so it says nothing — otherwise +every restart pushes "power levels are normal" at somebody's phone. +""" -def process(alert): - if alert.get("severity") == "info" and "normal" not in alert.get("body", ""): - return None - return {"push": alert.get("body", "")} +def process(alert, seen=False): + trouble = alert.get("severity") != "info" + if not trouble and not seen: + return {"seen_trouble": False} + return {"push": alert.get("body", ""), "seen_trouble": trouble} ''', ) flow.add( @@ -312,6 +329,7 @@ def process(alert): "requires": [{"name": "push", "dtype": "str"}], } ) + flow.input("seen_trouble", "bool", False) flow.add( { "id": "history", @@ -1023,7 +1041,16 @@ def presence(h: dict[str, Any]) -> Flow: "requires": [ {"name": "anyone_home", "dtype": "bool"}, {"name": "at_desk", "dtype": "bool"}, - {"name": "shutters.bed_down", "port": "bed_down", "dtype": "bool"}, + # Read, never waited on: this flow has to run while the + # shutters are still Node-RED's, which is most of the + # changeover. Someone going to bed is noticed on the next + # minute rather than the same second, which is soon enough. + { + "name": "shutters.bed_down", + "port": "bed_down", + "dtype": "bool", + "trigger": False, + }, {"name": "clock.hour", "port": "hour", "dtype": "int"}, { "name": "clock.is_weekend",