Merge remote-tracking branch 'origin/main' into HEAD

backend/fluksio/flow/controller.py

@@ -148,6 +148,17 @@ class RebuildBusy(RuntimeError):
     """
 
 
+class NodesNotLoaded(RuntimeError):
+    """A run's flow has nodes that could not be built, so none of it runs."""
+
+    def __init__(self, errors: dict[str, str]) -> None:
+        #: The load error of each node, by its qualified id.
+        self.errors = errors
+        node, error = next(iter(errors.items()))
+        more = f" and {len(errors) - 1} more" if len(errors) > 1 else ""
+        super().__init__(f"Node '{node}'{more} could not be loaded: {error}")
+
+
 class NodeStatus(str, Enum):
     ACTIVE = "active"
     ERROR = "error"
@@ -730,8 +741,13 @@ class FlowController:
         try:
             # Work already claimed belongs to the pipeline it was claimed
             # against; let it finish there before swapping the graph out.
-            if self.execution is not None:
-                await run_in_threadpool(self.execution.pause_intake)
+            # Work that does not finish in time keeps the old graph, which
+            # nothing has touched yet.
+            if self.execution is not None and not await run_in_threadpool(
+                self.execution.pause_intake
+            ):
+                self.execution.resume_intake()
+                raise RebuildBusy("Claimed work is still running on the pipeline")
             await self._teardown()
             # A fresh supervisor per build, so a flow quarantined by the last
             # one gets another chance once its author has changed something.
@@ -2010,10 +2026,21 @@ class FlowController:
         and webhooks would only be a second copy of what the live pipeline
         already holds. The state is the run's, which is what keeps two runs of
         one flow from overwriting each other's messages.
+
+        :raises NodesNotLoaded: When a node could not be built. The live engine
+            leaves such a node out, but a run without it would finish as if
+            the flow had done its work.
         """
-        nodes, _loaded, initial_values, _inputs = self._build_flows(
+        nodes, loaded, initial_values, _inputs = self._build_flows(
             [(flow, draft)], run=run
         )
+        errors = {
+            entry.id: entry.error or "unknown error"
+            for entry in loaded.values()
+            if entry.status is NodeStatus.ERROR
+        }
+        if errors:
+            raise NodesNotLoaded(errors)
         pipeline = Pipeline(
             nodes=nodes,
             state=state,

backend/fluksio/flow/dashboards.py

@@ -64,6 +64,10 @@ BAR_ROWS = 8
 #: client (``ui/core/counter.ts``, ``MAX_COUNTERS``).
 COUNTERS = 3
 
+#: How many windows one querying chart offers, stacked down the tile's edge.
+#: Mirrored in the client (``ChartWidget.tsx``, ``MAX_RANGES``).
+CHART_RANGES = 8
+
 #: Resolved out here on purpose: the store has a ``list`` method, which
 #: shadows the builtin for any annotation written inside the class.
 Bindings = list[dict[str, Any]]
@@ -259,10 +263,15 @@ class WidgetDef(BaseModel):
     A chart comes in two kinds. The default reads what the engine kept for a
     message. One with ``source: "query"`` asks instead, and its config is
     ``{source, request, request_dtype: "record", message, dtype: "series",
-    refresh_s, range_s}``: it publishes ``{range_s, interval_s}`` to
+    refresh_s, range_s, ranges}``: it publishes ``{range_s, interval_s}`` to
     ``request`` exactly as a slider publishes a value, and draws the ``series``
     a flow answers with on ``message``.
 
+    ``ranges`` lists the windows its rail offers, up to ``CHART_RANGES`` of
+    ``{label, range_s, interval_s}``, and ``range_s`` names the one it opens on.
+    Without them it offers 1h, 6h, 24h and 7d. ``range_s: 0`` asks for the
+    whole history, and the flow answering decides how far back that reaches.
+
     A colour widget picks what it publishes with ``format``, because fixtures
     differ and a change node per tile is not the answer:
 
@@ -448,6 +457,28 @@ class WidgetDef(BaseModel):
                     raise ValueError(
                         f"a querying chart's {key} must be '{want}', not '{bound}'"
                     )
+            ranges = self.config.get("ranges") or []
+            if not isinstance(ranges, list) or len(ranges) > CHART_RANGES:
+                raise ValueError(
+                    f"a querying chart offers a list of at most {CHART_RANGES} ranges"
+                )
+            for entry in ranges:
+                window = entry if isinstance(entry, dict) else {}
+                label = window.get("label")
+                span, step = window.get("range_s"), window.get("interval_s")
+                # `type() is int`, because a bool is an int to isinstance.
+                if not (
+                    isinstance(label, str)
+                    and label.strip()
+                    and type(span) is int
+                    and span >= 0
+                    and type(step) is int
+                    and step > 0
+                ):
+                    raise ValueError(
+                        "a querying chart's range needs a label, a whole range_s "
+                        f"of 0 or more and a whole interval_s above 0, not {entry!r}"
+                    )
             return self
 
         if self.type == "bar" and len(self.bar_rows) > BAR_ROWS:
@@ -961,21 +992,23 @@ class DashboardStore:
                         self._dump(shown.model_copy(update={"version": top + 1}))
                     )
                 draft.unlink()
-            self.flows._commit(f"Discard draft of dashboard '{name}'")
-        return self.read(name)
+            shown = self.read(name)
+            self.flows._commit(f"Discard draft of dashboard {named(shown.name, name)}")
+        return shown
 
     def delete(self, name: str) -> None:
         if not self.exists(name):
             raise DashboardNotFound(name)
         path = self._file(name)
         with self.flows._write_lock:
+            key = next((k for k, i in self.key_index().items() if i == name), name)
             path.unlink(missing_ok=True)
             self._draft_file(name).unlink(missing_ok=True)
             try:
                 path.parent.rmdir()
             except OSError:
                 pass
-            self.flows._commit(f"Delete dashboard '{name}'")
+            self.flows._commit(f"Delete dashboard {named(key, name)}")
 
     def duplicate(self, name: str) -> DashboardDef:
         """Copy the working copy into a new dashboard, as a draft only.

backend/fluksio/flow/executor.py

@@ -255,14 +255,26 @@ class ExecutionService:
         with self._inflight_lock:
             self._inflight_lock.notify_all()
 
-    def pause_intake(self) -> None:
+    def pause_intake(self) -> bool:
         """Stop claiming, and wait for what is already claimed to finish.
 
         Called around a rebuild: items claimed against the old pipeline should
-        finish there rather than half-run against the new one.
+        finish there rather than half-run against the new one. That includes
+        a claim still under way: the consumer holds its lock from the claim to
+        the dispatch, so once that lock is free here the claim is counted, and
+        every later pass finds the intake closed.
+
+        False when that took longer than `DRAIN_TIMEOUT_S`. The intake stays
+        closed either way; the caller decides whether to go on.
         """
         self._intake.clear()
+        # So a consumer waiting for capacity sees the intake closed.
+        self._capacity_changed()
         deadline = time.monotonic() + DRAIN_TIMEOUT_S
+        if not self._consumer_lock.acquire(timeout=DRAIN_TIMEOUT_S):
+            logger.warning("Rebuild did not wait out a claim")
+            return False
+        self._consumer_lock.release()
         with self._inflight_lock:
             while self._running:
                 remaining = deadline - time.monotonic()
@@ -270,8 +282,9 @@ class ExecutionService:
                     logger.warning(
                         "Rebuild did not wait out %d cascades", self._running.total()
                     )
-                    return
+                    return False
                 self._inflight_lock.wait(remaining)
+        return True
 
     def resume_intake(self) -> None:
         self._intake.set()
@@ -284,8 +297,8 @@ class ExecutionService:
         """Stop claiming, promoting, touching and reclaiming, once every
         claimed cascade has finished. For a rename, which rewrites the queue.
 
-        Unlike `pause_intake` this also waits out a claim already blocking
-        when intake clears, and a timer pass. It waits without a limit and
+        Unlike `pause_intake` this also waits out a timer pass, and keeps the
+        consumer lock until `release`. It waits without a limit and
         tells ``report`` the nodes whose cascades are still running each time
         they change. Setting ``abort`` gives up: claiming resumes and it
         returns False. True means held until `release`.

backend/fluksio/flow/pipeline.py

@@ -148,6 +148,8 @@ class NodeOutcome(BaseModel):
     artifacts: dict[str, dict[str, Any]] = {}
     #: Restored from an earlier run rather than executed.
     cached: bool = False
+    #: Not executed, because a node it waits on failed or was skipped itself.
+    skipped: bool = False
     #: Which run the restored values came from — and, because a restored row
     #: holds no series of its own, which run this node's metrics live in.
     cached_from: str = ""
@@ -1073,7 +1075,8 @@ class Pipeline:
         inputs: dict[str, Any] | None = None,
         arrived: list[str] | None = None,
     ) -> dict[str, Any] | None:
-        """Run one node and record its outputs. Never raises.
+        """Run one node and record its outputs. Never raises: None means it
+        failed, and a node that published nothing answers ``{}``.
 
         ``inputs`` is what the readiness check already read — the same keys, a
         moment earlier. Reading them again here was the second of two MGETs
@@ -1108,7 +1111,7 @@ class Pipeline:
             if key:
                 hit, restored = self._from_cache(node, key, state, entry_id)
                 if hit:
-                    return restored
+                    return restored or {}
 
             token = ARRIVED.set(arrived)
             try:
@@ -1175,7 +1178,7 @@ class Pipeline:
                         output_values=result,
                     )
                 )
-            return result
+            return result or {}
         except Exception as exc:
             # One failing node must not take the rest of the graph down.
             error = self.publish_error(node, exc, collected, entry_id)
@@ -1437,6 +1440,11 @@ class Pipeline:
 
         ready_nodes = deque(n for n, count in in_degree.items() if count == 0)
         blocked: set[Node] = set()
+        # A batch run starts from an empty state, so a node whose producer
+        # failed would be called without that input. It is skipped instead,
+        # and so is everything waiting on it. A live wave still has the last
+        # values to run on.
+        failed: set[Node] | None = set() if self.run_id else None
         gate_revision = self._gate_revision
         node_futures: dict[Future[dict[str, Any] | None], Node] = {}
         finished: SimpleQueue[Future[dict[str, Any] | None]] = SimpleQueue()
@@ -1475,10 +1483,26 @@ class Pipeline:
                 if n.passive:
                     complete(n)
                     continue
+                if failed is not None and not failed.isdisjoint(deps[n]):
+                    failed.add(n)
+                    self._observe(NodeOutcome(node=n.id, ok=False, skipped=True))
+                    complete(n)
+                    continue
                 if fresh is not None and untouched(n):
                     complete(n)
                     continue
                 ready_inputs: dict[str, Any] | None = None
+                if failed is not None:
+                    # A live wave's readiness rule: a node runs once every
+                    # triggering input has a value. A run has nothing to wait
+                    # for, so the node fails instead.
+                    ready_inputs = state.get_present(list(n.requires))
+                    missing = self._no_value(n, ready_inputs)
+                    if missing:
+                        failed.add(n)
+                        self._observe(NodeOutcome(node=n.id, ok=False, error=missing))
+                        complete(n)
+                        continue
                 if check_ready:
                     ready, ready_inputs = self._is_node_ready(n, state, entry_id)
                     if not ready:
@@ -1518,6 +1542,8 @@ class Pipeline:
                 future = finished.get()
                 n = node_futures.pop(future)
                 result = future.result()
+                if result is None and failed is not None:
+                    failed.add(n)
                 if result is not None and fresh is not None:
                     fresh.update(result)
                 # Silent, failed and throttled producers still release their
@@ -1536,6 +1562,18 @@ class Pipeline:
 
         return state
 
+    def _no_value(self, node: Node, values: dict[str, Any]) -> str:
+        """The first triggering input with nothing in ``values``, as an error,
+        or "" when there is none."""
+        for name, spec in node.requires.items():
+            if spec.trigger and name not in values:
+                producers = ", ".join(
+                    f"'{p.local_id}'" for p in self.produces.get(name, ())
+                )
+                where = f" from {producers}" if producers else ""
+                return f"No value on input '{spec.port or name}'{where}"
+        return ""
+
     def run(
         self, inputs: dict[str, Any] | None = None, nodes: set[Node] | None = None
     ) -> StateBackend:

backend/fluksio/flow/runs.py

@@ -51,7 +51,12 @@ from sqlmodel import Session, col, select
 from fluksio.core.db import engine as db_engine
 from fluksio.flow import tombstones
 from fluksio.flow.artifacts import ArtifactStore, is_reference, valid_digest
-from fluksio.flow.controller import NODE_TYPES, FlowController, RunContext
+from fluksio.flow.controller import (
+    NODE_TYPES,
+    FlowController,
+    NodesNotLoaded,
+    RunContext,
+)
 from fluksio.flow.messages import qualify, requalify
 from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, CacheHit, NodeOutcome, Pipeline
 from fluksio.flow.queue import WorkItem, WorkQueue
@@ -137,8 +142,9 @@ def batch_issues(
     history keeps every value, and the interval only thins what is published.
     Anywhere else it is a message quietly going missing.
 
-    A delay node is fine; without a queue to defer into it simply sleeps,
-    which in a run is what was asked for.
+    A delay node's ``interval`` is the same limit, put on every output it
+    has. Its ``delay`` is fine: without a queue to defer into it simply
+    sleeps, which in a run is what was asked for.
 
     The other is a node with no body. The store answers a new node's template
     when nothing was ever written for one, so such a node runs — and returns
@@ -148,8 +154,11 @@ def batch_issues(
     """
     issues: list[str] = []
     for node in flow.nodes:
-        for spec in list(node.requires) + list(node.provides):
-            if spec.interval > 0 and not spec.stream:
+        held = float(node.params.get("interval") or 0) if node.type == "delay" else 0
+        limits = [(spec, spec.interval) for spec in node.requires]
+        limits += [(spec, max(spec.interval, held)) for spec in node.provides]
+        for spec, interval in limits:
+            if interval > 0 and not spec.stream:
                 issues.append(
                     f"Node '{node.id}' rate-limits '{spec.port or spec.name}'. "
                     "A run has no timer to release what that holds back, so "
@@ -1236,8 +1245,10 @@ class RunService:
                 return True
         # The gate first, so nothing new is submitted while the running nodes
         # are being killed; a gated node is never handed to the executor, so
-        # the graph drains instead of going further.
-        pipeline.pause(self._key_of(run_id))
+        # the graph drains instead of going further. Stopped rather than
+        # paused: a pause announces itself on the engine's bus, under the key
+        # the live flow shares, so every open editor showed that one paused.
+        pipeline.set_disabled({self._key_of(run_id)})
         workers = self.controller.workers
         if workers is not None:
             # Keyed by run, so a sweep cancelling one config leaves the others
@@ -1502,7 +1513,7 @@ class RunService:
 
         def observe(outcome: NodeOutcome) -> None:
             nonlocal errors
-            if not outcome.ok:
+            if not outcome.ok and not outcome.skipped:
                 errors += 1
             self._record_node(run, key, outcome)
 
@@ -1537,19 +1548,26 @@ class RunService:
             digests = node_digests(flow)
             run.code_digest = self._restamp(run, run_digest(flow))
             state = self._state_factory(f"{RUN_NAMESPACE}:{run_id}")
-            pipeline = self.controller.build_run_pipeline(
-                flow,
-                state=state,
-                draft=run.draft,
-                observer=observe,
-                emission_observer=sink.handle,
-                run=RunContext(
-                    run_id=run_id,
-                    code_digest=run.code_digest,
-                    node_digests=digests,
-                ),
-                run_cache=None if run.no_cache else self._cache,
-            )
+            try:
+                pipeline = self.controller.build_run_pipeline(
+                    flow,
+                    state=state,
+                    draft=run.draft,
+                    observer=observe,
+                    emission_observer=sink.handle,
+                    run=RunContext(
+                        run_id=run_id,
+                        code_digest=run.code_digest,
+                        node_digests=digests,
+                    ),
+                    run_cache=None if run.no_cache else self._cache,
+                )
+            except NodesNotLoaded as exc:
+                # Nothing of it runs; the finally below records it.
+                for node, error in exc.errors.items():
+                    observe(NodeOutcome(node=node, ok=False, error=error))
+                status, reason = "error", str(exc)[:1024]
+                return
             with self._lock:
                 if run_id in self._cancelled:
                     # Cancelled while it was being built, so nothing of it
@@ -1627,10 +1645,16 @@ class RunService:
         run_id = run.id
         node = requalify(outcome.node, key, run.flow_id)[:255]
         outputs = _cacheable(outcome)
+        if outcome.cached:
+            status = "cached"
+        elif outcome.ok:
+            status = "ok"
+        else:
+            status = "skipped" if outcome.skipped else "error"
         row = RunNode(
             run_id=run_id,
             node=node,
-            status="cached" if outcome.cached else ("ok" if outcome.ok else "error"),
+            status=status,
             started_at=datetime.now(UTC),
             duration_ms=outcome.duration_ms,
             error=outcome.error[:ERROR_CAP],

backend/fluksio/flow/state.py

@@ -639,6 +639,14 @@ class RedisState(StateBackend):
         names = [k for k in keys if not k.startswith("__")]
         if names:
             pipe.sadd(self._names_key(), *names)
+            if self._ttl:
+                pipe.expire(self._names_key(), self._ttl)
+
+    def _incr(self, pipe: Any, key: str) -> None:
+        """INCR a counter in the caller's pipeline, expiring with the values."""
+        pipe.incr(self._key(key))
+        if self._ttl:
+            pipe.expire(self._key(key), self._ttl)
 
     def set(self, key: str, value: Any) -> None:
         data = self._serialize(value)
@@ -747,7 +755,9 @@ class RedisState(StateBackend):
 
     def increment(self, key: str) -> int:
         """Atomically increment a counter using Redis INCR."""
-        return cast(int, self._client.incr(self._key(key)))
+        pipe = self._client.pipeline()
+        self._incr(pipe, key)
+        return cast(int, pipe.execute()[0])
 
     def increment_multi(self, keys: list[str]) -> None:
         """Bump every counter in one round trip rather than one INCR each."""
@@ -755,7 +765,7 @@ class RedisState(StateBackend):
             return
         pipe = self._client.pipeline()
         for key in keys:
-            pipe.incr(self._key(key))
+            self._incr(pipe, key)
         pipe.execute()
 
     def get_multi(self, keys: list[str]) -> dict[str, Any]:
@@ -824,34 +834,31 @@ class RedisState(StateBackend):
 
         watch_keys = [self._key(k) for k in expected.keys()]
 
-        try:
-            # Start watching the keys for changes
-            self._client.watch(*watch_keys)
+        # WATCH has to go through the pipeline: it pins one connection, and
+        # the client-level call is a no-op in redis-py 7. Leaving the block
+        # resets the pipeline, which unwatches.
+        with self._client.pipeline() as pipe:
+            try:
+                pipe.watch(*watch_keys)  # type: ignore[no-untyped-call]
+                # A watching pipeline runs commands immediately.
+                current = cast(list[bytes | None], pipe.mget(watch_keys))
+                for raw, exp_value in zip(current, expected.values(), strict=True):
+                    if self._deserialize(raw) != exp_value:
+                        return False
 
-            # Check current values
-            current = self.get_multi(list(expected.keys()))
-
-            for key, exp_value in expected.items():
-                if current.get(key) != exp_value:
-                    self._client.unwatch()
-                    return False
-
-            # All values match - execute the transaction
-            pipe = self._client.pipeline(True)  # True = use MULTI/EXEC
-
-            for key, value in updates.items():
-                data = self._serialize(value)
-                if self._ttl:
-                    pipe.setex(self._key(key), self._ttl, data)
-                else:
-                    pipe.set(self._key(key), data)
-
-            pipe.execute()
-            return True
-
-        except redis.WatchError:
-            # Another client modified one of the watched keys
-            return False
+                pipe.multi()
+                for key, value in updates.items():
+                    data = self._serialize(value)
+                    if self._ttl:
+                        pipe.setex(self._key(key), self._ttl, data)
+                    else:
+                        pipe.set(self._key(key), data)
+                self._note(pipe, updates)
+                pipe.execute()
+                return True
+            except redis.WatchError:
+                # Another client modified one of the watched keys
+                return False
 
     def close(self) -> None:
         """Give the connection pool back."""
@@ -916,7 +923,7 @@ class RedisState(StateBackend):
             if self._ttl:
                 pipe.expire(history_key, self._ttl)
         for key in counters:
-            pipe.incr(self._key(key))
+            self._incr(pipe, key)
         self._note(pipe, values)
         pipe.execute()
 

backend/fluksio/flow/store.py

@@ -202,12 +202,14 @@ def new_id(kind: Literal["f", "d"], taken: Callable[[str], bool]) -> str:
 
 
 class FlowNotFound(KeyError):
-    def __init__(self, name: str) -> None:
+    def __init__(self, name: str, published: bool = False) -> None:
         super().__init__(name)
         self.name = name
+        self.published = published
 
     def __str__(self) -> str:
-        return f"No flow named '{self.name}'"
+        what = "published flow" if self.published else "flow"
+        return f"No {what} with the id '{self.name}'"
 
 
 class FlowExists(ValueError):
@@ -720,7 +722,7 @@ class FlowStore:
                 return self._load(path)
         path = self._flow_file(name)
         if not path.exists():
-            raise FlowNotFound(name)
+            raise FlowNotFound(name, published=not draft)
         return self._load(path)
 
     def key_index(self) -> dict[str, str]:
@@ -782,7 +784,9 @@ class FlowStore:
         and returned as a copy so a caller sorting it cannot disturb the next.
         """
         with self._read_all_lock:
-            if self._read_all_at != self.revision:
+            # Taken before scanning, so a commit landing meanwhile scans again.
+            at = self.revision
+            if self._read_all_at != at:
                 flows = []
                 for path in sorted(self.root.glob("*/flow.json")):
                     name = path.parent.name
@@ -791,7 +795,7 @@ class FlowStore:
                     except Exception:
                         logger.exception("Skipping unreadable flow '%s'", name)
                 self._read_all = flows
-                self._read_all_at = self.revision
+                self._read_all_at = at
             return list(self._read_all)
 
     def write_flow(self, flow: FlowDef) -> bool:

backend/fluksio/main.py

@@ -428,8 +428,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         app.state.device_store = devices
         controller.devices = devices
         # What the sockets want and the flows read, mapped onto the devices'
-        # on-demand ports.
+        # on-demand ports. It is process-wide, so it lets go of this
+        # controller when the engine stops.
         demand.bind(controller)
+        started.append(lambda: demand.bind(None))
         # Charts need a deeper series than the default; tell the engine
         # before it starts recording.
         controller.set_history_limits(dashboards.history_requirements())

backend/fluksio/sdk/__init__.py

@@ -877,12 +877,16 @@ def _shim(spec: NodeSpec) -> str:
     The import is aliased because ``process``'s parameters are the node's port
     names, and a port may be named after the very function it feeds — which
     would otherwise shadow the import inside the body.
+
+    Only the triggering ports are named. The engine leaves out an input that
+    has no value yet, which only a port that does not trigger can lack, so
+    those travel in ``**settings`` and the function's own default applies.
     """
     _refuse_main(spec)
     fn = spec.fn
     where = inspect.getsourcefile(fn) or fn.__module__
     repo = import_root(fn)
-    ports = [port.port for port in spec.requires]
+    ports = [port.port for port in spec.requires if port.trigger]
     signature = ", ".join(ports + ["**settings"])
     arguments = ", ".join([f"{port}={port}" for port in ports] + ["**settings"])
     call = f"_impl({arguments})"

backend/fluksio/sdk/client.py

@@ -170,6 +170,11 @@ class Conflict(ApiError):
             self.current_version = int(detail.get("current_version") or 0)
 
 
+def engine_url(url: str) -> str:
+    """An engine's address, whether it was given with the API's path or not."""
+    return url.rstrip("/").removesuffix(API).rstrip("/")
+
+
 class Client:
     """An authenticated engine, addressed over its HTTP API."""
 
@@ -182,7 +187,9 @@ class Client:
         retries: int = RETRIES,
     ) -> None:
         stored = _stored()
-        self.url = url or os.environ.get("FLUKSIO_URL") or stored.get("url") or ""
+        self.url = engine_url(
+            url or os.environ.get("FLUKSIO_URL") or stored.get("url") or ""
+        )
         self.token = (
             token or os.environ.get("FLUKSIO_TOKEN") or stored.get("token") or ""
         )
@@ -766,8 +773,9 @@ def login(
     """Exchange credentials for a token and remember the engine."""
     import httpx
 
+    url = engine_url(url)
     response = httpx.post(
-        f"{url.rstrip('/')}{API}/login/access-token",
+        f"{url}{API}/login/access-token",
         data={"username": email, "password": password},
         timeout=timeout,
     )

backend/scripts/bench_startup.py

@@ -96,11 +96,15 @@ def ensure_flow(api: Api) -> None:
     api.call(
         "PUT", f"/api/v1/flows/{BENCH_FLOW}/nodes/noop/source", {"code": NODE_SOURCE}
     )
-    api.call(
-        "POST",
-        f"/api/v1/flows/{BENCH_FLOW}/publish",
-        {"version": definition["version"]},
-    )
+    # Read again, since the saves move the version. A flow an interrupted run
+    # left published, and unchanged since, has nothing to publish.
+    current = api.call("GET", f"/api/v1/flows/{BENCH_FLOW}")
+    if current["has_draft"]:
+        api.call(
+            "POST",
+            f"/api/v1/flows/{BENCH_FLOW}/publish",
+            {"version": current["definition"]["version"]},
+        )
 
 
 def time_runs(api: Api, count: int) -> tuple[list[float], list[float]]:

backend/tests/api/routes/test_flow_ids.py

@@ -20,6 +20,7 @@ from fluksio.flow.dashboards import DashboardDef, WidgetDef
 from fluksio.flow.store import ISSUED_ID
 from fluksio.flow.tombstones import names_of
 from fluksio.models import FlowTombstone, Run
+from tests.api.routes.test_rename import _until
 
 FLOWS = f"{settings.API_V1_STR}/flows"
 DASHBOARDS = f"{settings.API_V1_STR}/dashboards"
@@ -180,8 +181,14 @@ def test_a_flow_runs_under_its_key_and_is_reached_by_its_id(
     hooked = client.post(f"/hooks/{flow_id}/tick", json={"tick": 7.0})
     assert hooked.status_code == 200, hooked.text
     assert client.post(f"/hooks/{key}/tick", json={"tick": 8.0}).status_code == 404
-    state = client.get(f"{FLOWS}/{flow_id}/state", headers=superuser_token_headers)
-    assert state.json()["values"][f"{key}.tick"]["value"] == 7.0
+
+    def values() -> dict[str, Any]:
+        state = client.get(f"{FLOWS}/{flow_id}/state", headers=superuser_token_headers)
+        return dict(state.json()["values"])
+
+    # Once the cascade it queued has run.
+    _until(lambda: f"{key}.tick" in values())
+    assert values()[f"{key}.tick"]["value"] == 7.0
 
     paused = client.post(f"{FLOWS}/{flow_id}/pause", headers=superuser_token_headers)
     assert paused.status_code == 200

backend/tests/api/routes/test_runs.py

@@ -19,7 +19,7 @@ from fluksio.core.db import engine as db_engine
 from fluksio.flow.artifacts import ArtifactStore
 from fluksio.flow.controller import FlowController
 from fluksio.flow.messages import DType, MessageSpec
-from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, NodeOutcome
+from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, NodeOutcome, Pipeline
 from fluksio.flow.runs import (
     LEASE_STALE_S,
     OUTPUT_CAP,
@@ -729,6 +729,105 @@ def test_a_draft_run_executes_the_draft(tmp_path):
             session.commit()
 
 
+def test_a_failed_stage_skips_what_waits_on_it_and_fails_the_run(tmp_path):
+    """Stages after a failure were called without their input and raised
+    TypeError, and a run whose only node could not load finished `ok`."""
+    model, score = (MessageSpec(name=n, dtype=DType.FLOAT) for n in ("model", "score"))
+    staged = FlowDef(
+        name="staged",
+        mode="batch",
+        nodes=[
+            NodeDef(id="fit", provides=[model]),
+            NodeDef(id="readout", requires=[model], provides=[score]),
+            NodeDef(id="evaluate", requires=[score]),
+        ],
+    )
+    broken = FlowDef(name="broken", mode="batch", nodes=[NodeDef(id="fit")])
+    store = FlowStore(tmp_path / "flows")
+    store.write_flow(staged)
+    store.write_flow(broken)
+    store.write_node_source("staged", "fit", "def process():\n    raise ValueError\n")
+    store.write_node_source(
+        "staged", "readout", "def process(model):\n    return {'score': model}\n"
+    )
+    store.write_node_source("staged", "evaluate", "def process(score):\n    pass\n")
+    store.write_node_source("broken", "fit", "def process(:\n")
+
+    service = RunService(controller=FlowController(store), queue=_Collect())
+    made = [service.submit(name, {}).id for name in ("staged", "broken")]
+    try:
+        for run_id in made:
+            service._drive(run_id)
+        with Session(db_engine) as session:
+            ran, failed = (session.get(Run, run_id) for run_id in made)
+            assert ran.status == failed.status == "error"
+            assert failed.status_reason.startswith("Node 'broken.fit' could not")
+            rows = session.exec(select(RunNode).where(col(RunNode.run_id).in_(made)))
+            assert {row.node: row.status for row in rows} == {
+                "staged.fit": "error",
+                "staged.readout": "skipped",
+                "staged.evaluate": "skipped",
+                "broken.fit": "error",
+            }
+    finally:
+        with Session(db_engine) as session:
+            for node in session.exec(
+                select(RunNode).where(col(RunNode.run_id).in_(made))
+            ).all():
+                session.delete(node)
+            for run in session.exec(select(Run).where(col(Run.id).in_(made))).all():
+                session.delete(run)
+            session.commit()
+
+
+def test_a_stage_with_nothing_on_an_input_fails_instead_of_being_called(tmp_path):
+    """The producer succeeded but published nothing on the port it waits on."""
+    model, score, report = (
+        MessageSpec(name=n, dtype=DType.FLOAT) for n in ("model", "score", "report")
+    )
+    flow = FlowDef(
+        name="split",
+        mode="batch",
+        nodes=[
+            NodeDef(id="fit", provides=[model, score]),
+            NodeDef(id="readout", requires=[score], provides=[report]),
+            NodeDef(id="evaluate", requires=[report]),
+        ],
+    )
+    store = FlowStore(tmp_path / "flows")
+    store.write_flow(flow)
+    store.write_node_source(
+        "split", "fit", "def process():\n    return {'model': 1.0}\n"
+    )
+    store.write_node_source(
+        "split", "readout", "def process(score):\n    return {'report': score}\n"
+    )
+    store.write_node_source("split", "evaluate", "def process(report):\n    pass\n")
+
+    service = RunService(controller=FlowController(store), queue=_Collect())
+    run_id = service.submit("split", {}).id
+    try:
+        service._drive(run_id)
+        with Session(db_engine) as session:
+            assert session.get(Run, run_id).status == "error"
+            rows = {
+                row.node: row
+                for row in session.exec(select(RunNode).where(RunNode.run_id == run_id))
+            }
+            assert {node: row.status for node, row in rows.items()} == {
+                "split.fit": "ok",
+                "split.readout": "error",
+                "split.evaluate": "skipped",
+            }
+            assert rows["split.readout"].error == "No value on input 'score' from 'fit'"
+    finally:
+        with Session(db_engine) as session:
+            for node in session.exec(select(RunNode).where(RunNode.run_id == run_id)):
+                session.delete(node)
+            session.delete(session.get(Run, run_id))
+            session.commit()
+
+
 def test_a_retry_is_a_new_run_that_names_the_one_it_repeats():
     """The way back from a run an engine restart interrupted.
 
@@ -905,6 +1004,39 @@ def test_a_run_waiting_for_a_worker_keeps_its_lease_and_can_be_cancelled(
             session.commit()
 
 
+class _Bus:
+    def __init__(self):
+        self.events = []
+
+    def publish(self, event):
+        self.events.append(event)
+
+
+def test_cancelling_a_run_does_not_pause_the_live_flow_it_shares_a_key_with():
+    """A run's pipeline publishes on the engine's bus under the flow's key, so
+    its pause marked the live flow paused in every open editor."""
+
+    class _Idle(_OneFlow):
+        workers = None
+        remote = None
+
+    service = RunService(
+        controller=_Idle(FlowDef(name="study", mode="batch")), queue=_Collect()
+    )
+    run = service.submit("study", {})
+    bus = _Bus()
+    pipeline = Pipeline(events=bus)
+    service._active[run.id] = pipeline
+    try:
+        assert service.cancel(run.id)
+        assert pipeline.is_disabled("study")
+        assert not [e for e in bus.events if e["type"] == "flow_paused"]
+    finally:
+        with Session(db_engine) as session:
+            session.delete(session.get(Run, run.id))
+            session.commit()
+
+
 def test_a_key_nobody_used_submits_normally(
     client, superuser_token_headers, monkeypatch
 ):

backend/tests/conftest.py

@@ -1,3 +1,4 @@
+import shutil
 import uuid
 from collections.abc import Generator
 from datetime import UTC, datetime
@@ -107,6 +108,19 @@ def _clean_engine_rows() -> Generator[None, None, None]:
         session.commit()
 
 
+@pytest.fixture(scope="module", autouse=True)
+def _clean_flow_store() -> Generator[None, None, None]:
+    """Start the next module on an empty flow store.
+
+    The store is session-wide like the database, so the flows, dashboards and
+    devices one module left were loaded by the next module's app at startup.
+    Removed whole, its git with it, after the module's app has stopped.
+    """
+    yield
+    shutil.rmtree(settings.FLOWS_DIR, ignore_errors=True)
+    settings.DEVICES_FILE.unlink(missing_ok=True)
+
+
 @pytest.fixture(scope="module")
 def client() -> Generator[TestClient, None, None]:
     with TestClient(app) as c:

backend/tests/flow/test_dashboards.py

@@ -5,6 +5,7 @@ from datetime import UTC, datetime
 import pytest
 
 from fluksio.flow.dashboards import (
+    CHART_RANGES,
     DashboardDef,
     DashboardNotFound,
     DashboardStore,
@@ -354,6 +355,26 @@ def test_a_querying_chart_asks_with_a_record_and_draws_a_series():
         WidgetDef(id="c", type="chart", config=query_chart(request_dtype="json"))
 
 
+def test_a_querying_chart_offers_its_own_ranges():
+    """A few labelled windows, where ``range_s: 0`` is the whole history."""
+    daily = [
+        {"label": "1M", "range_s": 30 * 86400, "interval_s": 86400},
+        {"label": "All", "range_s": 0, "interval_s": 86400},
+    ]
+    WidgetDef(id="c", type="chart", config=query_chart(ranges=daily))
+
+    for bad in (
+        [{"label": "", "range_s": 3600, "interval_s": 60}],
+        [{"label": "1h", "range_s": -1, "interval_s": 60}],
+        [{"label": "1h", "range_s": 3600, "interval_s": 0}],
+        [{"label": "1h", "range_s": 3600.5, "interval_s": 60}],
+        daily * CHART_RANGES,
+        "1M",
+    ):
+        with pytest.raises(ValueError, match="range"):
+            WidgetDef(id="c", type="chart", config=query_chart(ranges=bad))
+
+
 def test_a_colour_widget_publishes_the_shape_its_format_names():
     """The format picks the payload, so the binding is held to that one."""
     wheel = WidgetDef(

backend/tests/flow/test_engine_hold.py

@@ -5,13 +5,14 @@ service stops claiming and its timers once claimed work has finished. Both
 wait without a limit and can be aborted.
 """
 
+import asyncio
 import threading
 import time
 
 import pytest
 
 from fluksio.flow import executor
-from fluksio.flow.controller import FlowController, LoadedNode
+from fluksio.flow.controller import FlowController, LoadedNode, RebuildBusy
 from fluksio.flow.executor import ExecutionService
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.namespace import Rename
@@ -125,8 +126,54 @@ class SlowClaim(MemoryWorkQueue):
         return items
 
 
+def test_pause_intake_waits_out_a_claim_already_under_way(monkeypatch):
+    seen: list = []
+    queue = SlowClaim()
+    _, service, source = _engine(seen.append, queue)
+    wake = service._capacity_changed
+
+    def closing() -> None:
+        wake()
+        # The claim lands only once the intake has closed.
+        queue.go.set()
+
+    monkeypatch.setattr(service, "_capacity_changed", closing)
+    service.start()
+    try:
+        source.inject({"reading": 1.0})
+        assert queue.claiming.wait(5)
+        service.pause_intake()
+        assert seen == [1.0]
+        assert service.inflight == 0
+    finally:
+        queue.go.set()
+        service.stop()
+
+
+def test_a_reload_that_cannot_drain_leaves_the_pipeline_as_it_was(
+    tmp_path, monkeypatch
+):
+    monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05)
+    seen: list = []
+    running, finish, consume = _blocking(seen)
+    pipeline, service, source = _engine(consume)
+    controller = FlowController(FlowStore(tmp_path / "flows"), execution=service)
+    controller.pipeline = pipeline
+    service.start()
+    try:
+        source.inject({"reading": 1.0})
+        assert running.wait(5)
+        with pytest.raises(RebuildBusy):
+            asyncio.run(controller.reload())
+        assert controller.pipeline is pipeline
+        assert service._pipeline is pipeline
+        assert service._intake.is_set()
+    finally:
+        finish.set()
+        service.stop()
+
+
 def test_a_hold_waits_out_a_claim_already_under_way():
-    """`pause_intake` returns here and the item runs after it."""
     seen: list = []
     queue = SlowClaim()
     _, service, source = _engine(seen.append, queue)

backend/tests/flow/test_redis_state.py

@@ -0,0 +1,54 @@
+"""RedisState against a real server, enabled by FLUKSIO_TEST_REDIS_HOST (and
+optional PORT)."""
+
+import os
+import uuid
+
+import pytest
+
+from fluksio.flow.state import RedisState
+
+
+@pytest.fixture
+def state():
+    host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
+    if not host:
+        pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
+    port = int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379"))
+    state = RedisState(host, port, namespace=f"test-state-{uuid.uuid4().hex}", ttl=600)
+    try:
+        yield state
+    finally:
+        state.clear()
+        state.close()
+
+
+def test_a_ttl_expires_counters_and_the_names_set(state):
+    state.set("heat.temp", 21)
+    state.increment("__version__:heat.a")
+    state.increment_multi(["__version__:heat.b"])
+    state.record({"heat.v": 1}, {}, ["__version__:heat.c"], 0.0)
+
+    keys = state.keys()
+    assert "__names__" in keys and "__version__:heat.c" in keys
+    assert all(state._client.ttl(state._key(k)) > 0 for k in keys)
+
+
+def test_a_compare_and_swap_names_what_it_wrote(state):
+    state.set("heat.a", 1)
+    assert state.compare_and_swap_multi({"heat.a": 1}, {"heat.b": 2})
+    assert "heat.b" in state.message_names()
+
+
+def test_a_compare_and_swap_loses_to_a_write_after_its_read(state, monkeypatch):
+    state.set("heat.a", 1)
+    note = RedisState._note
+
+    def racing_note(self, pipe, keys):
+        # Another writer lands between the compare and EXEC.
+        self._client.set(self._key("heat.a"), b"2")
+        note(self, pipe, keys)
+
+    monkeypatch.setattr(RedisState, "_note", racing_note)
+    assert not state.compare_and_swap_multi({"heat.a": 1}, {"heat.a": 3})
+    assert state.get("heat.a") == 2

backend/tests/flow/test_runs.py

@@ -346,6 +346,16 @@ def test_a_rate_limited_port_cannot_be_run_as_a_batch():
     assert not batch_issues(double_flow())
 
 
+def test_a_delay_nodes_interval_cannot_be_run_as_a_batch():
+    flow = double_flow()
+    flow.nodes[0].type = "delay"
+    flow.nodes[0].params = {"interval": 5}
+    # The node puts it on every output, where it holds values back the same way.
+    assert batch_issues(flow)
+    flow.nodes[0].params = {"delay": 5}
+    assert not batch_issues(flow)
+
+
 def test_a_node_with_no_stored_code_cannot_be_run():
     """The store answers a template for one, and a template publishes nothing."""
 

backend/tests/sdk/test_build.py

@@ -240,6 +240,28 @@ def test_a_port_named_like_its_function_still_calls_the_function():
     assert namespace["process"](scale=2.0) == {"scaled": 4.0}
 
 
+@node(
+    requires=["value", Port("factor", "float", trigger=False)],
+    provides=Port("scaled", "float"),
+)
+def scale_by(value, factor=3.0):
+    return value * factor
+
+
+def test_a_port_that_does_not_trigger_may_be_missing_from_the_call():
+    """The engine leaves out an input with no value yet, so the default applies."""
+    flow = Flow(
+        "optional",
+        nodes=[scale_by],
+        inputs=[Port("value", "float", initial=1.0), Port("factor", "float")],
+    )
+    namespace: dict = {}
+    exec(compile(flow.shims()["scale_by"], "<shim>", "exec"), namespace)
+
+    assert namespace["process"](value=2.0) == {"scaled": 6.0}
+    assert namespace["process"](value=2.0, factor=0.5) == {"scaled": 1.0}
+
+
 def reads_settings(settings):
     return settings
 

backend/tests/sdk/test_client.py

@@ -33,6 +33,19 @@ def a_client(handler, monkeypatch, **kwargs):
     return Client(url="http://engine", token="t", http=http, **kwargs)
 
 
+@pytest.mark.parametrize(
+    "url",
+    [
+        "http://e:8000",
+        "http://e:8000/",
+        "http://e:8000/api/v1",
+        "http://e:8000/api/v1/",
+    ],
+)
+def test_an_engine_url_may_carry_the_api_path(url):
+    assert Client(url=url, token="t").url == "http://e:8000"
+
+
 def test_idempotent_get_is_tried_again(monkeypatch):
     calls = []
 

backend/tests/test_cloud.py

@@ -788,7 +788,9 @@ async def test_a_revoked_link_stands_down(enrolled: User) -> None:  # noqa: ARG0
         watcher = asyncio.create_task(connector.watch_enrolment(app))
         await asyncio.sleep(0.05)
         watcher.cancel()
-    assert not started
+    # The module's app runs a watcher of its own on the same patched names and
+    # may dial the enrolment above inside the window, so only this one counts.
+    assert app not in started
 
 
 def test_a_member_removed_at_the_portal_loses_their_user(db: Session) -> None:

docs/code/cli.md

@@ -151,7 +151,8 @@ Client commands use the nearest `.fluksio/client.json`, or the global
 configuration when there is no project instance. Legacy client configuration
 is a fallback.
 `FLUKSIO_URL` and `FLUKSIO_TOKEN` override saved values; command options
-`--url` and `--token` override both.
+`--url` and `--token` override both. An engine URL works with or without
+the `/api/v1` path.
 
 A client needs an engine at least as new as itself. An engine from before
 flow ids cannot create flows the way this client does, and the client says

docs/concepts/runs.md

@@ -18,9 +18,17 @@ A batch flow does not need activation and does not open subscriptions,
 schedules or webhooks. A run executes each node once after its dependencies
 finish. Missing parameters use their declared initial values.
 
+When a node fails, the nodes that wait on it are recorded as `skipped` and
+the run ends `error`. A node that cannot be loaded fails the run before
+anything executes, with the node's load error as the reason.
+A node with no value on a triggering input fails without being called, as a
+live flow would wait for it, so a canvas port with a default needs
+`trigger: false` to be optional.
+
 Non-streaming ports cannot use `interval` in a batch flow: there is no timer
-to deliver a delayed value. Streaming ports can use it to reduce display
-updates while keeping every metric point.
+to deliver a delayed value. The same goes for a delay node's `interval`
+setting. Streaming ports can use it to reduce display updates while keeping
+every metric point.
 
 ## Submitting
 

docs/interface/dashboards.md

@@ -188,6 +188,21 @@ Choose the chart's data source:
 
 Multi-line charts show a legend as well as different colours.
 
+A Query chart offers 1h, 6h, 24h and 7d down its right-hand edge, and opens on
+24h. Under **Ranges** you can give it windows of its own instead, up to eight,
+and it opens on the one its `range_s` names or else the first. Each window
+becomes the request's `range_s` and `interval_s` when picked. A `range_s` of
+`0` asks for the whole history, and the answering flow decides how far back
+that reaches.
+
+```json
+"ranges": [
+  {"label": "1M", "range_s": 2592000, "interval_s": 86400},
+  {"label": "1Y", "range_s": 31536000, "interval_s": 86400},
+  {"label": "All", "range_s": 0, "interval_s": 86400}
+]
+```
+
 !!! tip "Skip the unfinished bucket"
 
     If your query aggregates into time buckets, leave out the bucket still

frontend/src/client/schemas.gen.ts

@@ -5713,10 +5713,15 @@ by a schema per class, because the whole set is small and closed.
 A chart comes in two kinds. The default reads what the engine kept for a
 message. One with \`\`source: "query"\`\` asks instead, and its config is
 \`\`{source, request, request_dtype: "record", message, dtype: "series",
-refresh_s, range_s}\`\`: it publishes \`\`{range_s, interval_s}\`\` to
+refresh_s, range_s, ranges}\`\`: it publishes \`\`{range_s, interval_s}\`\` to
 \`\`request\`\` exactly as a slider publishes a value, and draws the \`\`series\`\`
 a flow answers with on \`\`message\`\`.
 
+\`\`ranges\`\` lists the windows its rail offers, up to \`\`CHART_RANGES\`\` of
+\`\`{label, range_s, interval_s}\`\`, and \`\`range_s\`\` names the one it opens on.
+Without them it offers 1h, 6h, 24h and 7d. \`\`range_s: 0\`\` asks for the
+whole history, and the flow answering decides how far back that reaches.
+
 A colour widget picks what it publishes with \`\`format\`\`, because fixtures
 differ and a change node per tile is not the answer:
 

frontend/src/client/types.gen.ts

@@ -1775,10 +1775,15 @@ export type WebPushKey = {
  * A chart comes in two kinds. The default reads what the engine kept for a
  * message. One with ``source: "query"`` asks instead, and its config is
  * ``{source, request, request_dtype: "record", message, dtype: "series",
- * refresh_s, range_s}``: it publishes ``{range_s, interval_s}`` to
+ * refresh_s, range_s, ranges}``: it publishes ``{range_s, interval_s}`` to
  * ``request`` exactly as a slider publishes a value, and draws the ``series``
  * a flow answers with on ``message``.
  *
+ * ``ranges`` lists the windows its rail offers, up to ``CHART_RANGES`` of
+ * ``{label, range_s, interval_s}``, and ``range_s`` names the one it opens on.
+ * Without them it offers 1h, 6h, 24h and 7d. ``range_s: 0`` asks for the
+ * whole history, and the flow answering decides how far back that reaches.
+ *
  * A colour widget picks what it publishes with ``format``, because fixtures
  * differ and a change node per tile is not the answer:
  *

frontend/src/components/Dashboard/ChartWidget.tsx

@@ -2,11 +2,7 @@ import { useQueries, useQuery } from "@tanstack/react-query"
 import { useEffect, useRef, useState } from "react"
 
 import type { HistoryPoint } from "@/client"
-import {
-  DEFAULT_RANGE,
-  RANGES,
-  type Range,
-} from "@/components/Common/RangePicker"
+import { DEFAULT_RANGE, RANGES } from "@/components/Common/RangePicker"
 import { MAX_SERIES, UplotChart } from "@/components/Common/UplotChart"
 import {
   type LiveValue,
@@ -40,9 +36,59 @@ const DEFAULT_POINTS = 300
  * quarter-hour buckets asks four times an hour instead of twice a minute, for
  * the same picture. A slower refresh than that is honoured; a faster one only
  * buys the same answer again, so it is clamped.
+ *
+ * Capped at what a timer can hold, about 24 days: `setInterval` keeps its
+ * delay in 32 bits of milliseconds, and a longer one wraps round and fires at
+ * once, which would ask again as fast as the flow answers.
  */
-export const refreshFor = (range: Range, configured: unknown) =>
-  Math.max(range.bucketS, Number(configured) || 0)
+export const refreshFor = (range: ChartRange, configured: unknown) =>
+  Math.min(Math.max(range.interval_s, Number(configured) || 0), 2_147_483)
+
+/**
+ * One window a querying chart offers, in the shape its request takes.
+ *
+ * `range_s: 0` asks for the whole history, and the flow answering decides how
+ * far back that reaches.
+ */
+export type ChartRange = { label: string; range_s: number; interval_s: number }
+
+/** How many windows one chart offers, stacked down the tile's edge. Mirrored
+ *  server-side as `CHART_RANGES` (`app/flow/dashboards.py`). */
+export const MAX_RANGES = 8
+
+/** What a chart that names no windows of its own offers. */
+const PRESETS: ChartRange[] = RANGES.map((range) => ({
+  label: range.label,
+  range_s: range.hours * 3600,
+  interval_s: range.bucketS,
+}))
+
+/**
+ * The windows a querying chart offers: its own `ranges`, else the presets.
+ *
+ * A row still being typed into in the editor is not a window yet. Drawn, it
+ * would ask for nothing and pace its timer at zero.
+ */
+export function rangesOf(cfg: Record<string, unknown>): ChartRange[] {
+  const own = ((cfg.ranges ?? []) as Partial<ChartRange>[]).filter(
+    (range): range is ChartRange =>
+      Boolean(range.label) &&
+      (range.range_s ?? -1) >= 0 &&
+      (range.interval_s ?? 0) > 0,
+  )
+  return own.length ? own : PRESETS
+}
+
+/** Which of its windows a chart opens on: the one `range_s` names, else its
+ *  first, or the presets' day. */
+export function startOf(cfg: Record<string, unknown>): number {
+  const ranges = rangesOf(cfg)
+  const named = ranges.findIndex(
+    (range) => range.range_s === Number(cfg.range_s),
+  )
+  if (named >= 0) return named
+  return ranges === PRESETS ? RANGES.indexOf(DEFAULT_RANGE) : 0
+}
 
 /** How long an unanswered request blocks an identical one. */
 const INFLIGHT_MS = 15_000
@@ -353,13 +399,14 @@ function QueryChart({ widget, dashboard }: WidgetProps) {
   const cfg = config(widget)
   const request = String(cfg.request ?? "")
   const message = String(cfg.message ?? "")
-  const [range, setRange] = useState<Range>(
-    () =>
-      RANGES.find((r) => r.hours * 3600 === Number(cfg.range_s)) ??
-      DEFAULT_RANGE,
-  )
-  const rangeS = range.hours * 3600
-  const intervalS = range.bucketS
+  const ranges = rangesOf(cfg)
+  // By position, so two windows alike are still two buttons. One the editor
+  // has since removed falls back to the first.
+  const [picked, setPicked] = useState(() => startOf(cfg))
+  const shown = picked < ranges.length ? picked : 0
+  const range = ranges[shown]
+  const rangeS = range.range_s
+  const intervalS = range.interval_s
   // Follows the window: a wider one is drawn coarser, so it is worth asking
   // about less often.
   const refreshS = refreshFor(range, cfg.refresh_s)
@@ -433,18 +480,20 @@ function QueryChart({ widget, dashboard }: WidgetProps) {
         palette={palette}
         {...presentation(cfg)}
       />
-      <RangeRail range={range} onChange={setRange} />
+      <RangeRail ranges={ranges} picked={shown} onChange={setPicked} />
     </div>
   )
 }
 
 /** The window a chart is showing, as a column beside it. */
 function RangeRail({
-  range,
+  ranges,
+  picked,
   onChange,
 }: {
-  range: Range
-  onChange: (range: Range) => void
+  ranges: ChartRange[]
+  picked: number
+  onChange: (index: number) => void
 }) {
   const { Segmented } = useUi()
   return (
@@ -452,16 +501,11 @@ function RangeRail({
       orientation="vertical"
       testId="range-picker"
       label="Time range"
-      value={String(range.hours)}
-      options={RANGES.map(
-        (entry) => [String(entry.hours), entry.label] as const,
+      value={String(picked)}
+      options={ranges.map(
+        (range, index) => [String(index), range.label] as const,
       )}
-      onChange={(hours) =>
-        onChange(
-          RANGES.find((entry) => String(entry.hours) === hours) ??
-            DEFAULT_RANGE,
-        )
-      }
+      onChange={(index) => onChange(Number(index))}
     />
   )
 }

frontend/src/components/Dashboard/panels.tsx

@@ -3,7 +3,6 @@ import { Ban, Check, ChevronDown, Plus, RotateCw, X } from "lucide-react"
 import { useState } from "react"
 
 import type { MessageInfo, SettingDef, WidgetDef } from "@/client"
-import { DEFAULT_RANGE, RANGES } from "@/components/Common/RangePicker"
 import { screensOf } from "@/components/Devices/ports"
 import { devicesQueryOptions } from "@/components/Devices/queries"
 import { RenameDialog } from "@/components/Flow/RenameDialog"
@@ -52,7 +51,14 @@ import {
 import { Switch } from "@/components/ui/switch"
 import useAuth from "@/hooks/useAuth"
 import { cn } from "@/lib/utils"
-import { MAX_SERIES, refreshFor } from "./ChartWidget"
+import {
+  type ChartRange,
+  MAX_RANGES,
+  MAX_SERIES,
+  rangesOf,
+  refreshFor,
+  startOf,
+} from "./ChartWidget"
 import { COLOR_DTYPES, COLOR_FORMATS, colorFormatOf } from "./ColorWidget"
 import {
   CANVAS_PRESETS,
@@ -569,13 +575,20 @@ export function WidgetPanel({
     })
   const querying = cfg.source === "query"
   const pinnedRuns = cfg.source === "runs"
+  // A querying chart's windows, read raw like a grid's presses: `rangesOf`
+  // drops a row until it is complete, and the rail offers what it keeps.
+  const ranges = (cfg.ranges ?? []) as Partial<ChartRange>[]
+  const setRange = (index: number, changes: Partial<ChartRange>) =>
+    set({
+      ranges: ranges.map((other, at) =>
+        at === index ? { ...other, ...changes } : other,
+      ),
+    })
+  const offered = rangesOf(cfg)
+  const start = startOf(cfg)
   // What this chart would refresh at with nothing configured. The viewer can
   // pick another window on the widget, which moves it.
-  const paced = refreshFor(
-    RANGES.find((range) => range.hours * 3600 === Number(cfg.range_s)) ??
-      DEFAULT_RANGE,
-    0,
-  )
+  const paced = refreshFor(offered[start], 0)
 
   return (
     <SidePanel
@@ -986,22 +999,105 @@ export function WidgetPanel({
                 }
               />
             </div>
+            <div className="grid gap-2">
+              <FieldLabel help="Each a label, how far back in seconds (0 for the whole history) and the bucket width in seconds. Without any, the widget offers 1h to 7d.">
+                Ranges
+              </FieldLabel>
+              {ranges.map((range, index) => (
+                <div
+                  // Position is the only identity a range row has.
+                  key={`range-${index}`}
+                  className="flex items-end gap-1.5"
+                >
+                  <Input
+                    className="min-w-0 flex-1"
+                    value={range.label ?? ""}
+                    placeholder="Label"
+                    aria-label="Range label"
+                    onChange={(event) =>
+                      setRange(index, { label: event.target.value })
+                    }
+                  />
+                  <Input
+                    type="number"
+                    className="w-24"
+                    min={0}
+                    placeholder="Seconds"
+                    aria-label="Range, seconds"
+                    value={str(range.range_s)}
+                    onChange={(event) =>
+                      setRange(index, {
+                        range_s: numberOrNone(event.target.value),
+                      })
+                    }
+                  />
+                  <Input
+                    type="number"
+                    className="w-24"
+                    min={1}
+                    placeholder="Bucket"
+                    aria-label="Bucket, seconds"
+                    value={str(range.interval_s)}
+                    onChange={(event) =>
+                      setRange(index, {
+                        interval_s: numberOrNone(event.target.value),
+                      })
+                    }
+                  />
+                  <Button
+                    variant="ghost"
+                    size="icon-sm"
+                    className="text-muted-foreground"
+                    aria-label="Remove range"
+                    onClick={() =>
+                      set({ ranges: ranges.filter((_, at) => at !== index) })
+                    }
+                  >
+                    <X />
+                  </Button>
+                </div>
+              ))}
+              {ranges.length < MAX_RANGES ? (
+                <Button
+                  variant="outline"
+                  size="sm"
+                  className="h-8 justify-self-start"
+                  // A copy of the last, so the document stays one the server
+                  // accepts while the copy is edited.
+                  onClick={() =>
+                    set({
+                      ranges: [
+                        ...ranges,
+                        { ...(ranges[ranges.length - 1] ?? offered[start]) },
+                      ],
+                    })
+                  }
+                  data-testid="add-range"
+                >
+                  <Plus />
+                  Add range
+                </Button>
+              ) : null}
+            </div>
             <div className="grid gap-1.5">
               <FieldLabel help="A viewer can pick another on the widget itself.">
                 Range shown first
               </FieldLabel>
               <Select
-                value={str(cfg.range_s ?? DEFAULT_RANGE.hours * 3600)}
-                onValueChange={(value) => set({ range_s: Number(value) })}
+                value={String(start)}
+                onValueChange={(value) =>
+                  set({ range_s: offered[Number(value)].range_s })
+                }
               >
                 <SelectTrigger>
                   <SelectValue />
                 </SelectTrigger>
                 <SelectContent>
-                  {RANGES.map((range) => (
+                  {offered.map((range, index) => (
                     <SelectItem
-                      key={range.label}
-                      value={String(range.hours * 3600)}
+                      // Position, as on the widget: two alike are still two.
+                      key={`offered-${index}`}
+                      value={String(index)}
                     >
                       {range.label}
                     </SelectItem>

frontend/tests/runs.spec.ts

@@ -23,7 +23,11 @@ const TRAIN = `def process(epochs):
 `
 
 test.afterAll(async ({ browser }) => {
-  await deleteAll(browser, [`/flows/${flowName}`])
+  // The runs after the flow, as the editor does it. Its pinned id is its key.
+  await deleteAll(browser, [
+    `/flows/${flowName}`,
+    `/runs?flow=${flowName}&history=true`,
+  ])
 })
 
 test.beforeAll(async ({ browser }) => {