diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index ee0c0e9..eaffc4f 100644 --- a/backend/fluksio/flow/controller.py +++ b/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, diff --git a/backend/fluksio/flow/dashboards.py b/backend/fluksio/flow/dashboards.py index 26c7efb..85510d2 100644 --- a/backend/fluksio/flow/dashboards.py +++ b/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. diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index 89f15d0..6b98c78 100644 --- a/backend/fluksio/flow/executor.py +++ b/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`. diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index b919678..6d397a5 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/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: diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 555d4a8..65aa1df 100644 --- a/backend/fluksio/flow/runs.py +++ b/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], diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py index 74c8d8f..d6fc341 100644 --- a/backend/fluksio/flow/state.py +++ b/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() diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index 9f9b58c..f3f117b 100644 --- a/backend/fluksio/flow/store.py +++ b/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: diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 842bfe2..595ab26 100644 --- a/backend/fluksio/main.py +++ b/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()) diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index c41fb5e..d841631 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/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})" diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 83589e9..1a5c3b0 100644 --- a/backend/fluksio/sdk/client.py +++ b/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, ) diff --git a/backend/scripts/bench_startup.py b/backend/scripts/bench_startup.py index 3610ee8..6bb3a46 100644 --- a/backend/scripts/bench_startup.py +++ b/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]]: diff --git a/backend/tests/api/routes/test_flow_ids.py b/backend/tests/api/routes/test_flow_ids.py index a3b3050..a7550f9 100644 --- a/backend/tests/api/routes/test_flow_ids.py +++ b/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 diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 3dd4261..1061161 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/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 ): diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index ef3f4d7..3237d2d 100644 --- a/backend/tests/conftest.py +++ b/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: diff --git a/backend/tests/flow/test_dashboards.py b/backend/tests/flow/test_dashboards.py index 9495276..afe6ca1 100644 --- a/backend/tests/flow/test_dashboards.py +++ b/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( diff --git a/backend/tests/flow/test_engine_hold.py b/backend/tests/flow/test_engine_hold.py index c796d4c..6a5704e 100644 --- a/backend/tests/flow/test_engine_hold.py +++ b/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) diff --git a/backend/tests/flow/test_redis_state.py b/backend/tests/flow/test_redis_state.py new file mode 100644 index 0000000..ae84d53 --- /dev/null +++ b/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 diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py index f1abf65..b1f56db 100644 --- a/backend/tests/flow/test_runs.py +++ b/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.""" diff --git a/backend/tests/sdk/test_build.py b/backend/tests/sdk/test_build.py index 380540d..4bb4b6c 100644 --- a/backend/tests/sdk/test_build.py +++ b/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"], "", "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 diff --git a/backend/tests/sdk/test_client.py b/backend/tests/sdk/test_client.py index aaf7c1a..3a1b0f5 100644 --- a/backend/tests/sdk/test_client.py +++ b/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 = [] diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py index 84df2be..55d7a51 100644 --- a/backend/tests/test_cloud.py +++ b/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: diff --git a/docs/code/cli.md b/docs/code/cli.md index 89ad14e..1cbd006 100644 --- a/docs/code/cli.md +++ b/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 diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 351e814..333b9a8 100644 --- a/docs/concepts/runs.md +++ b/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 diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index 1b00597..f9c7aba 100644 --- a/docs/interface/dashboards.md +++ b/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 diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index cadd163..8ee7b12 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/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: diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index afcfa0b..0f17b31 100644 --- a/frontend/src/client/types.gen.ts +++ b/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: * diff --git a/frontend/src/components/Dashboard/ChartWidget.tsx b/frontend/src/components/Dashboard/ChartWidget.tsx index a8ee6ca..b8fb6e6 100644 --- a/frontend/src/components/Dashboard/ChartWidget.tsx +++ b/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): ChartRange[] { + const own = ((cfg.ranges ?? []) as Partial[]).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): 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( - () => - 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)} /> - + ) } /** 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))} /> ) } diff --git a/frontend/src/components/Dashboard/panels.tsx b/frontend/src/components/Dashboard/panels.tsx index b93b451..d47b04a 100644 --- a/frontend/src/components/Dashboard/panels.tsx +++ b/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[] + const setRange = (index: number, changes: Partial) => + 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 ( +
+ + Ranges + + {ranges.map((range, index) => ( +
+ + setRange(index, { label: event.target.value }) + } + /> + + setRange(index, { + range_s: numberOrNone(event.target.value), + }) + } + /> + + setRange(index, { + interval_s: numberOrNone(event.target.value), + }) + } + /> + +
+ ))} + {ranges.length < MAX_RANGES ? ( + + ) : null} +
Range shown first