Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

history: one Flux script per chart, and rows are json

Browse Source 
Two things the installation found that the checks did not.

A script with three `from()` statements in it produces three results all
called `_result`, and InfluxDB refuses that outright — so the measurements go
into one filter and the rows come back tagged with which one they are.

And the answer a database node hands back holds a *list* of rows, which a
record may not: a record is flat scalars. It was declared one, so every chart
failed on the type check the moment a real answer arrived.

The second one is now caught before anything is pushed: the preflight runs
each sample shape past the port that would receive it, which is what turns
"expected record, got dict" from a runtime surprise into a line of output.

Also records the two engine faults this seeding session surfaced.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
stroblme
2026-08-22 16:24:38 +02:00
co-authored by Claude Opus 5
parent 8734e51ef1
commit 1b455b5e55
4 changed files with 112 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
NOTEPAD.md
+23
View File
@@ -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`.
scripts/tinyhouse/__main__.py
+26 -1
View File
@@ -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.
scripts/tinyhouse/screens.py
+28 -23
View File
@@ -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"}],
},
scripts/tinyhouse/sensing.py
+35 -8
View File
@@ -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",