diff --git a/Makefile b/Makefile index f81e0a1..33fcb4f 100644 --- a/Makefile +++ b/Makefile @@ -162,7 +162,7 @@ install: ## Install all dependencies (backend + frontend) cd frontend && bun install dev-backend: ## Start the FastAPI backend with hot-reload (local) - cd backend && uv run --all-extras fastapi dev fluksio/main.py + cd backend && uv run --all-extras uvicorn fluksio.main:app --reload --ws-per-message-deflate false dev-frontend: ## Start the Vite dev server (local) cd frontend && bun dev diff --git a/backend/fluksio/api/deps.py b/backend/fluksio/api/deps.py index f585e33..7b06c92 100644 --- a/backend/fluksio/api/deps.py +++ b/backend/fluksio/api/deps.py @@ -20,6 +20,7 @@ from fluksio.flow.artifacts import is_reference from fluksio.flow.controller import FlowController from fluksio.flow.dashboards import DashboardDef, DashboardStore from fluksio.flow.devices import DeviceDef, DeviceStore, DeviceUnavailable, presence +from fluksio.flow.events import event_bus from fluksio.flow.workers import PythonWorkerPool from fluksio.models import OAuthClient, TokenPayload, User, WorkerToken @@ -165,11 +166,13 @@ def _device_writable(device: DeviceDef, request: Request) -> set[str]: def _device_digests(device: DeviceDef, request: Request) -> set[str]: - """The artifacts this panel's messages are pointing at right now. + """The artifacts this panel's messages are pointing at, now or just before. A media tile fetches the bytes its message names, so the messages already bounding the panel bound this too — one step further along, through - whatever those messages currently hold. + whatever those messages currently hold. And through the last few they + held: a screen through the portal fetches a camera frame by frame, and by + the time its fetch arrives the camera may have published the next one. """ controller: FlowController | None = getattr( request.app.state, "flow_controller", None @@ -183,6 +186,8 @@ def _device_digests(device: DeviceDef, request: Request) -> set[str]: for value in controller.state.get_present(sorted(names)).values(): if is_reference(value): found.add(str(value["digest"])) + for name in names: + found.update(event_bus.digests.get(name, ())) return found @@ -215,9 +220,9 @@ def _device_may(payload: dict[str, Any], request: Request) -> None: The device is left on ``request.state`` for the route, which credits a value published on one of its ports to the node holding that port. - # ponytail: the device file and the published dashboards are re-read per - # request (the nonces are cached by the file's stat). Cache them behind - # the store's revision if this shows up in a profile. + The device file is read per request, which is what makes deleting a + device revoke it at once. The nonces are cached by the file's stat and + the published dashboards until the flow store's next commit. """ device = _device_for(payload, request.app) request.state.device = device diff --git a/backend/fluksio/api/routes/artifacts.py b/backend/fluksio/api/routes/artifacts.py index 45573af..48a1595 100644 --- a/backend/fluksio/api/routes/artifacts.py +++ b/backend/fluksio/api/routes/artifacts.py @@ -101,7 +101,9 @@ async def put_artifact( ``volatile`` puts it in the ring instead of the store — a frame a screen is watching now, which the oldest of falls out of memory rather than being - kept. A worker on another host publishing a camera sends this. + kept. A worker on another host or a device publishing a camera sends this. + It is read into memory, where the ring holds it, and never touches the + disk. """ store = _store(request) cap = settings.MAX_ARTIFACT_BYTES @@ -110,6 +112,17 @@ async def put_artifact( raise HTTPException( status_code=413, detail=f"An artifact may be at most {cap} bytes" ) + ring = store.volatile if volatile else None + if ring is not None: + body = bytearray() + async for chunk in request.stream(): + body += chunk + if cap and len(body) > cap: + raise HTTPException( + status_code=413, + detail=f"An artifact may be at most {cap} bytes", + ) + return await run_in_threadpool(ring.put, [bytes(body)], name, media_type) handle = tempfile.NamedTemporaryFile(dir=store.root, delete=False) written = 0 try: @@ -146,7 +159,9 @@ def get_artifact( """ store = _store(request) path = store.path(digest) - if path is None: + # A live frame is held in memory, not on disk. + frame = store.volatile.get(digest) if path is None and store.volatile else None + if path is None and frame is None: raise HTTPException(status_code=404, detail="No such artifact") if not _MEDIA_TYPE.match(media_type): media_type = "application/octet-stream" @@ -158,6 +173,8 @@ def get_artifact( # beside the session, for whoever follows a link to it. headers["Content-Security-Policy"] = "sandbox" headers["X-Content-Type-Options"] = "nosniff" + if path is None: + return Response(frame, media_type=media_type, headers=headers) return FileResponse(path, media_type=media_type, headers=headers) diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 8eb6f96..82a5e43 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -3,6 +3,7 @@ import asyncio import logging import os +import re import time from pathlib import Path from typing import Any @@ -32,12 +33,23 @@ from fluksio.api.deps import ( get_current_user, user_from_token, ) +from fluksio.api.routes.messages import port_source from fluksio.api.socket_auth import TICKET_TTL_S, issue_ticket, socket_token +from fluksio.core.config import settings from fluksio.core.db import engine from fluksio.flow.artifacts import is_reference from fluksio.flow.controller import NODE_TYPES, FlowController, follow_header from fluksio.flow.dashboards import DashboardStore -from fluksio.flow.devices import DeviceStore, messages_for, presence +from fluksio.flow.demand import demand +from fluksio.flow.devices import ( + DeviceDef, + DeviceStore, + DeviceUnavailable, + dashboards_for, + messages_for, + messages_of, + presence, +) from fluksio.flow.events import event_bus from fluksio.flow.messages import flow_of, qualify from fluksio.flow.pipeline import ValidationIssue @@ -1017,7 +1029,10 @@ def read_message_history( def snapshot_payload( - controller: FlowController, only: set[str] | None = None + controller: FlowController, + only: set[str] | None = None, + device: str = "", + uplink: str = "", ) -> dict[str, Any]: """Everything a client needs to catch up, sent the moment it connects. @@ -1032,12 +1047,16 @@ def snapshot_payload( gets: the values its own dashboards draw, and none of the rest — no node status, no logs, no shape of the graph. A panel renders none of that, and a screen may be hanging somewhere nobody here can see. + + A device's socket also says which of its ports have ``demand`` now, and + which ``uplink`` its frames take: ``binary`` on its own socket, ``http`` + through the portal, whose tunnel carries text only. """ if only is not None: # The same keys, emptied rather than dropped: a screen already hanging # runs whatever bundle it was paired with, and the shape of this # message is what that bundle reads. - return { + payload: dict[str, Any] = { "type": "snapshot", "values": {k: v for k, v in controller.values().items() if k in only}, "nodes": [], @@ -1047,6 +1066,10 @@ def snapshot_payload( "emits": {}, "queued": [], } + if device: + payload["demand"] = demand.ports(device) + payload["uplink"] = uplink + return payload return { "type": "snapshot", "values": controller.values(), @@ -1059,6 +1082,11 @@ def snapshot_payload( } +def demand_frame(ports: list[str]) -> str: + """What a device is told when its ports with demand change.""" + return orjson.dumps({"type": "demand", "ports": ports}).decode() + + def device_of(token: str, app: Any) -> str: """The device this credential stands for, or "" for a person or none.""" try: @@ -1067,12 +1095,30 @@ def device_of(token: str, app: Any) -> str: return "" +def revoked(token: str, app: Any) -> bool: + """Whether a device's credential has stopped standing since its handshake. + + Unpaired or deleted. One that cannot be checked right now has not: a + device closed with 1008 checks whether it is still paired. + """ + try: + decode_token(token, app=app) + except InvalidTokenError: + return True + except DeviceUnavailable: + return False + return False + + def panel_scope(token: str, app: Any) -> set[str] | None: """The messages this credential is bounded to, or None if it is a person's. The socket is the one authenticated surface the route check cannot reach, a handshake has no route to judge, so a device is bounded by what it is sent instead of by what it asks for. + + Its own camera's frames are left out unless a dashboard it shows draws + them: it sent them, and a camera's values come several a second. """ device_id = device_of(token, app) if not device_id: @@ -1081,7 +1127,18 @@ def panel_scope(token: str, app: Any) -> set[str] | None: devices: DeviceStore | None = getattr(app.state, "device_store", None) if store is None or devices is None: return set() - return messages_for(devices.find(device_id), store, devices) + device = devices.find(device_id) + found = messages_for(device, store, devices) + if device is not None: + shown = { + name for defn in dashboards_for(device, store) for name in messages_of(defn) + } + found -= { + b.message + for b in devices.bound(device.id) + if b.direction == "in" and b.dtype == "image" + } - shown + return found def event_for_panel(event: dict[str, Any], only: set[str]) -> bool: @@ -1103,9 +1160,9 @@ def event_for_panel(event: dict[str, Any], only: set[str]) -> bool: MAX_FRAME_EVENTS = 64 #: How many message names one socket may ask for bytes on. A screen draws a -#: handful of tiles; this is only here so a client cannot ask for the whole -#: namespace and be served every frame of it. -MAX_MEDIA_NAMES = 32 +#: handful of tiles, and every live one keeps a camera sending, so a phone +#: scrolled past several cameras does not pull them all. The rest are refused. +MAX_MEDIA_NAMES = 8 def wanted_names(frame: str, only: set[str] | None) -> set[str] | None: @@ -1126,7 +1183,17 @@ def wanted_names(frame: str, only: set[str] | None) -> set[str] | None: names = payload.get("names") if not isinstance(names, list): return set() - asked = {str(name) for name in names[:MAX_MEDIA_NAMES]} + asked = {str(name) for name in names} + if len(asked) > MAX_MEDIA_NAMES: + kept = list(dict.fromkeys(str(name) for name in names))[:MAX_MEDIA_NAMES] + logger.warning( + "A socket asked for live media on %d messages; only %d are served, " + "the first it named: %s", + len(asked), + MAX_MEDIA_NAMES, + ", ".join(kept), + ) + asked = set(kept) return asked if only is None else asked & only @@ -1140,10 +1207,10 @@ def media_frames( the socket that already carries the event closes that, and only for the frames a client said it was drawing. - Only what the ring holds: the durable store is what a fetch is for, and a - checkpoint has no business being pushed at anyone. The newest value per - name wins, so a client that fell behind is not handed a backlog of frames - it would only draw over. + Only what the ring holds, out of memory: the durable store is what a fetch + is for, and a checkpoint has no business being pushed at anyone. The + newest value per name wins, so a client that fell behind is not handed a + backlog of frames it would only draw over. """ if not wanted or store is None or getattr(store, "volatile", None) is None: return [] @@ -1161,12 +1228,8 @@ def media_frames( for name, event in newest.items(): value = event["value"] digest = str(value.get("digest") or "") - path = store.volatile.path(digest) - if path is None: - continue - try: - payload = path.read_bytes() - except OSError: + payload = store.volatile.get(digest) + if payload is None: continue header = orjson.dumps( { @@ -1183,6 +1246,91 @@ def media_frames( return frames +#: What a device may say a frame is. +_IMAGE_TYPE = re.compile(r"^image/[\w.+-]+$") + + +def parse_frame(data: bytes) -> tuple[dict[str, Any], bytes] | None: + """Split a binary frame into its JSON header and the bytes after it. + + The layout media frames go out in: four bytes of header length, big + endian, the header, then the bytes. None for anything else. + """ + if len(data) < 4: + return None + length = int.from_bytes(data[:4], "big") + if length == 0 or 4 + length > len(data): + return None + try: + header = orjson.loads(data[4 : 4 + length]) + except orjson.JSONDecodeError: + return None + if not isinstance(header, dict): + return None + return header, data[4 + length :] + + +def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str: + """Store a frame a device sent on its socket and publish it. + + Returns why it was refused, or "" once it is published. + + The same frame over HTTP is two requests, each authenticated: the upload, + then the publish. Here the socket was authenticated once, at its + handshake, and the frame is held and published in one step. Only a + message one of the device's own ``in`` ports is wired to, as an ``image``, + and the value is credited to the device node holding that port, as an + HTTP publish is. + """ + parsed = parse_frame(data) + if parsed is None or parsed[0].get("type") != "frame": + return "not a frame" + header, payload = parsed + name = str(header.get("name") or "") + devices: DeviceStore | None = getattr(app.state, "device_store", None) + store: Any = getattr(app.state, "artifact_store", None) + controller: FlowController | None = getattr(app.state, "flow_controller", None) + if devices is None or store is None or controller is None: + return "the engine is not running" + binding = next( + ( + b + for b in devices.bound(device.id) + if b.direction == "in" and b.message == name + ), + None, + ) + if binding is None: + return f"{name!r} is not a message one of its ports publishes" + if binding.dtype != "image": + return f"its port {binding.port!r} is wired as {binding.dtype}, not image" + media_type = str(header.get("media_type") or "") + if not _IMAGE_TYPE.match(media_type): + return f"{media_type!r} is not an image type" + cap = settings.MAX_ARTIFACT_BYTES + if cap and len(payload) > cap: + return f"a frame may be at most {cap} bytes" + reference = store.put( + [payload], name=binding.port, media_type=media_type, volatile=True + ) + try: + controller.publish_message(name, reference, port_source(devices, device, name)) + except KeyError: + return f"no flow declares {name!r}" + except TypeError as exc: + return str(exc) + return "" + + +async def _receive(websocket: WebSocket) -> str | bytes: + """The client's next frame, text or binary.""" + message = await websocket.receive() + if message["type"] == "websocket.disconnect": + raise WebSocketDisconnect(message.get("code", 1000), message.get("reason")) + data = message.get("bytes") + return data if data is not None else str(message.get("text") or "") + + async def _send(websocket: WebSocket, events: list[dict[str, Any]]) -> None: """One event, or a batch of them under `events`. @@ -1216,6 +1364,11 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - A browser brings a ticket from `POST /flows/ws-ticket`; anything that can set headers sends `Authorization: Bearer`. `?token=` is still read for one release, for clients from before either. + + A client says which media it is drawing (``{"type": "media", "names": + [...]}``) and is pushed those frames. A device also sends its camera's + frames here, as binary frames in the layout it is pushed them in, and is + told when its on-demand ports have demand. """ token = socket_token(websocket.headers, ticket, token) @@ -1228,10 +1381,10 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - only = panel_scope(token, websocket.app) if user is not None else None # A device is online for as long as this socket is open. device = device_of(token, websocket.app) if only is not None else "" - return user, only, device + return user, only, device, _device_def(websocket.app, device) try: - user, only, device = await run_in_threadpool(_authenticate) + user, only, device, device_def = await run_in_threadpool(_authenticate) except Exception as exc: # Not a refusal: the database or the device store could not say. A # device takes 1008, or a 403 before accepting, as its credential @@ -1246,6 +1399,10 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - # Nothing until a client says it is drawing something: bytes are what this # socket cannot afford to send speculatively. wanted: set[str] = set() + # This socket's entry in the registry every socket's wants are kept in. + viewer = object() + # Each reason a frame was refused for, logged once per socket. + refused: set[str] = set() await websocket.accept() @@ -1254,17 +1411,27 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - ) store = getattr(websocket.app.state, "artifact_store", None) - async def send_snapshot() -> None: - if controller is not None: - payload = await run_in_threadpool(snapshot_payload, controller, only) - await websocket.send_text(orjson.dumps(payload).decode()) + async def send_snapshot() -> list[str]: + """Send the snapshot, and return the demand it told a device of.""" + if controller is None: + return [] + payload = await run_in_threadpool( + snapshot_payload, controller, only, device, "binary" + ) + await websocket.send_text(orjson.dumps(payload).decode()) + return list(payload.get("demand", [])) - await send_snapshot() + told = await send_snapshot() async with event_bus.subscribe() as queue: - receiver = asyncio.create_task(websocket.receive_text()) + receiver = asyncio.create_task(_receive(websocket)) if device: presence.opened(device) + # Demand that moved between the snapshot and the subscription + # would otherwise wait for the next change. + ports = demand.ports(device) + if ports != told: + await websocket.send_text(demand_frame(ports)) try: while True: sender = asyncio.create_task(queue.get()) @@ -1284,14 +1451,35 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - # it every live update from then on. failure = receiver.exception() if failure is not None: - # The client is gone, and `receive_text` raises for - # good from here on. Reading it again would spin this - # loop at full tilt against a dead socket. + # The client is gone, and `receive` raises for good + # from here on. Reading it again would spin this loop + # at full tilt against a dead socket. raise failure - asked = wanted_names(receiver.result(), only) - if asked is not None: - wanted = asked - receiver = asyncio.create_task(websocket.receive_text()) + frame = receiver.result() + if isinstance(frame, bytes): + # A camera's frame. Refused ones are dropped and the + # socket stays open: one bad frame is not a reason to + # take the device's live values away. + reason = ( + await run_in_threadpool( + ingest_frame, websocket.app, device_def, frame + ) + if device_def is not None + else "only a device sends frames" + ) + if reason and reason not in refused: + refused.add(reason) + logger.warning( + "Dropped a frame from %s: %s", + device or "a person's socket", + reason, + ) + else: + asked = wanted_names(frame, only) + if asked is not None: + wanted = asked + demand.want(viewer, wanted) + receiver = asyncio.create_task(_receive(websocket)) if sender not in done: continue # `sender` is done from here: `FIRST_COMPLETED` returns with at @@ -1309,7 +1497,22 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - out: list[dict[str, Any]] = [] for event in batch: + if event.get("type") == "device_demand": + # For the device it names, and for nobody else. + if device and event.get("device") == device: + await websocket.send_text( + demand_frame(list(event.get("ports") or [])) + ) + continue if only is not None and event.get("type") == "dashboard_changed": + # An unpaired or deleted device is closed with 1008, + # which sends it to check its credential: otherwise + # its frames would still be taken here. + if device and await run_in_threadpool( + revoked, token, websocket.app + ): + await websocket.close(code=1008) + return # The scope was resolved once, at the handshake. A # panel pointed at another dashboard would otherwise # fetch the new document and then draw tiles nothing @@ -1318,7 +1521,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - # token — and that would widen this socket to # everything on the bus. only = panel_scope(token, websocket.app) or set() + device_def = _device_def(websocket.app, device) wanted &= only + demand.want(viewer, wanted) if out: await _send(websocket, out) out = [] @@ -1339,5 +1544,14 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - pass finally: receiver.cancel() + demand.drop(viewer) if device: presence.closed(device) + + +def _device_def(app: Any, device_id: str) -> DeviceDef | None: + """The device a socket stands for, as its frames are credited to it.""" + devices: DeviceStore | None = getattr(app.state, "device_store", None) + if not device_id or devices is None: + return None + return devices.find(device_id) diff --git a/backend/fluksio/api/routes/messages.py b/backend/fluksio/api/routes/messages.py index a96e1d3..93ed718 100644 --- a/backend/fluksio/api/routes/messages.py +++ b/backend/fluksio/api/routes/messages.py @@ -72,17 +72,16 @@ async def read_messages(controller: FlowControllerDep) -> Any: return MessagesPublic(data=infos, count=len(infos)) -def _port_source(request: Request, name: str) -> ValueSource | None: +def port_source( + devices: DeviceStore, device: DeviceDef, name: str +) -> ValueSource | None: """The device node a device's own publish on a bound ``in`` port is from. Said by the server rather than by the device: the node is what the canvas draws the value leaving from, and the device does not know which node in - which flow holds its port. + which flow holds its port. Also what a frame on the device's socket is + credited to. """ - device: DeviceDef | None = getattr(request.state, "device", None) - devices: DeviceStore | None = getattr(request.app.state, "device_store", None) - if device is None or devices is None: - return None for bound in devices.bound(device.id): if bound.direction == "in" and bound.message == name: return ValueSource( @@ -94,6 +93,14 @@ def _port_source(request: Request, name: str) -> ValueSource | None: return None +def _port_source(request: Request, name: str) -> ValueSource | None: + device: DeviceDef | None = getattr(request.state, "device", None) + devices: DeviceStore | None = getattr(request.app.state, "device_store", None) + if device is None or devices is None: + return None + return port_source(devices, device, name) + + @router.post("/{name}", response_model=MessageValue) async def publish_message( name: str, body: PublishRequest, controller: FlowControllerDep, request: Request diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index ac43dd1..ba0206a 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -577,6 +577,10 @@ def cmd_serve(args: argparse.Namespace) -> int: port=port, log_level=args.log_level, log_config=_log_config(args.log_level), + # Media frames are JPEGs, which deflate cannot shrink, and + # compressing each one again per open socket was a quarter of the + # engine's time with four cameras. + ws_per_message_deflate=False, ) ) try: diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index a9e96e8..b7b8ac6 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -22,7 +22,7 @@ import logging import random import time import uuid -from collections.abc import MutableMapping +from collections.abc import Callable, MutableMapping from dataclasses import replace from datetime import timedelta from typing import Any @@ -72,6 +72,8 @@ ALLOWED_PATHS = frozenset( #: Neither dict was bounded, and an id the hub reused silently dropped the #: reference to a task that was still running. MAX_IN_FLIGHT = 256 +#: How many of a client's frames are kept for a stream not listening yet. +MAX_EARLY = 8 #: How often to look for a config that appeared while the engine was up. #: Enrolment is something a person does and then waits on, so this is the #: delay they sit through — short enough not to be worth a restart. @@ -160,6 +162,10 @@ class CloudConnector: self.app_url: str | None = None #: What the hub sent each bridged hook socket, waiting to be received. self._inboxes: dict[str, asyncio.Queue[dict[str, Any]]] = {} + #: What reads a client's frames on each bridged live stream, and what + #: a client said before its stream was listening. + self._listeners: dict[str, Callable[[str], None]] = {} + self._early: dict[str, list[str]] = {} @property def revoked(self) -> bool: @@ -279,14 +285,21 @@ class CloudConnector: elif op == "ws_open": self._start_stream(socket, frame) elif op == "ws_msg": - inbox = self._inboxes.get(str(frame.get("id") or "")) + stream = str(frame.get("id") or "") + text = str(frame.get("text") or "") + inbox = self._inboxes.get(stream) + listener = self._listeners.get(stream) if inbox is not None: inbox.put_nowait( - { - "type": "websocket.receive", - "text": str(frame.get("text") or ""), - } + {"type": "websocket.receive", "text": text} ) + elif listener is not None: + listener(text) + elif stream in self._streams: + # A browser speaks as soon as the hub accepts it, + # before this end has checked its credential. + early = self._early.setdefault(stream, []) + early[:] = [*early[-(MAX_EARLY - 1) :], text] elif op == "ws_close": self._cancel_stream(str(frame.get("id") or "")) elif op == "member_removed": @@ -638,13 +651,17 @@ class CloudConnector: from fluksio.api.deps import user_from_token from fluksio.api.routes.flows import ( + demand_frame, device_of, event_for_panel, panel_scope, + revoked, snapshot_payload, + wanted_names, ) from fluksio.api.socket_auth import redeem_ticket from fluksio.core.db import engine + from fluksio.flow.demand import demand from fluksio.flow.devices import presence from fluksio.flow.events import event_bus @@ -682,24 +699,64 @@ class CloudConnector: await socket.send(_dump({"op": "ws_open_ok", "id": stream_id})) controller = getattr(self._app.state, "flow_controller", None) + told: list[str] = [] if controller is not None: + # A device here sends its frames over HTTP: the tunnel carries + # text only. + snapshot = snapshot_payload(controller, only, device, "http") + told = list(snapshot.get("demand", [])) await socket.send( - _dump( - { - "op": "ws_msg", - "id": stream_id, - "text": _dump(snapshot_payload(controller, only)), - } - ) + _dump({"op": "ws_msg", "id": stream_id, "text": _dump(snapshot)}) ) + # The media this client's tiles are drawing count as demand on the + # devices sending them, as a direct socket's do. The bytes themselves + # are fetched one by one through the portal. + viewer = ("tunnel", stream_id) + wanted: set[str] = set() + + def heard(text: str) -> None: + nonlocal wanted + asked = wanted_names(text, only) + if asked is not None: + wanted = asked + demand.want(viewer, asked) + + self._listeners[stream_id] = heard + for text in self._early.pop(stream_id, []): + heard(text) + if device: presence.opened(device) try: async with event_bus.subscribe() as queue: + if device and (ports := demand.ports(device)) != told: + await socket.send( + _dump( + { + "op": "ws_msg", + "id": stream_id, + "text": demand_frame(ports), + } + ) + ) while True: event = await queue.get() + if event.get("type") == "device_demand": + # For the device it names, and for nobody else. + if device and event.get("device") == device: + text = demand_frame(list(event.get("ports") or [])) + await socket.send( + _dump({"op": "ws_msg", "id": stream_id, "text": text}) + ) + continue if only is not None and event.get("type") == "dashboard_changed": + if device and revoked(token, self._app): + # Unpaired or deleted: 1008 sends it to check. + await socket.send( + _dump({"op": "ws_close", "id": stream_id, "code": 1008}) + ) + return # What this screen shows may have just changed under # it. Rescope, then resend the snapshot so a dashboard # it has only now been assigned draws values instead of @@ -707,8 +764,12 @@ class CloudConnector: # deleted scopes to nothing, and None would widen this # socket to everything on the bus. only = panel_scope(token, self._app) or set() + wanted &= only + demand.want(viewer, wanted) if controller is not None: - catch_up = snapshot_payload(controller, only) + catch_up = snapshot_payload( + controller, only, device, "http" + ) await socket.send( _dump( { @@ -732,6 +793,9 @@ class CloudConnector: _dump({"op": "ws_close", "id": stream_id, "code": 1011}) ) finally: + self._listeners.pop(stream_id, None) + self._early.pop(stream_id, None) + demand.drop(viewer) if device: presence.closed(device) diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 4613425..2003cc2 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -207,16 +207,19 @@ class Settings(BaseSettings): # Storing bytes and recording the reference are two steps; this is the # window between them. ARTIFACT_GC_GRACE_S: int = 3600 - #: Where frames a flow only shows live are held: memory rather than the - #: data volume, so a camera at ten frames a second is not writing to an SD - #: card. Empty works it out — a directory under `/dev/shm` named for this - #: data directory, so two instances on one host do not trim each other — - #: and falls back to the temporary directory where there is no `/dev/shm`. + #: Where a worker in this container writes the frames a flow only shows + #: live: memory rather than the data volume, so a camera at ten frames a + #: second is not writing to an SD card. The engine holds its own frames in + #: its memory. Empty works it out — a directory under `/dev/shm` named for + #: this data directory, so two instances on one host do not trim each + #: other — and falls back to the temporary directory where there is no + #: `/dev/shm`. ARTIFACT_VOLATILE_DIR: Path | None = None - #: How much the volatile ring holds before the oldest frames fall out of - #: it; 0 turns it off, and a volatile save then lands in the durable store - #: like any other. Under Docker's default 64 MB `/dev/shm`: raise - #: `shm_size` with it. + #: How many bytes of live frames the engine holds before the oldest fall + #: out, and the workers' directory too; 0 turns the ring off, and a + #: volatile save then lands in the durable store like any other. Under + #: Docker's default 64 MB `/dev/shm`: raise `shm_size` with it when + #: workers publish frames. ARTIFACT_VOLATILE_BYTES: int = 48 * 1024 * 1024 # Without a Redis host the engine keeps its state in memory. REDIS_HOST: str | None = None diff --git a/backend/fluksio/flow/artifacts.py b/backend/fluksio/flow/artifacts.py index 6515be3..be76a47 100644 --- a/backend/fluksio/flow/artifacts.py +++ b/backend/fluksio/flow/artifacts.py @@ -19,7 +19,9 @@ import hashlib import logging import os import tempfile +import threading import time +from collections import OrderedDict from collections.abc import Iterable, Iterator from pathlib import Path from typing import Any @@ -52,9 +54,10 @@ class ArtifactStore: def __init__(self, root: Path) -> None: self.root = root self.root.mkdir(parents=True, exist_ok=True) - #: A ring of frames held in memory, for media a flow only shows live. - #: Set by the engine at startup; every lookup below falls through to - #: it, which is what lets a volatile reference be an ordinary one. + #: Frames held in memory, for media a flow only shows live. Set by the + #: engine at startup unless the ring is turned off; every lookup below + #: falls through to it, which is what lets a volatile reference be an + #: ordinary one. self.volatile: VolatileStore | None = None def _path(self, digest: str) -> Path: @@ -79,7 +82,8 @@ class ArtifactStore: ``volatile`` puts it in the ring instead, where it is held in memory and falls out once newer frames need the room. The reference is the same shape either way — what differs is how long the bytes last, and a - caller that wants one kept returns it from a run. + caller that wants one kept returns it from a run. Without a ring it + lands here like anything else. """ if volatile and self.volatile is not None: return self.volatile.put(chunks, name=name, media_type=media_type) @@ -128,20 +132,26 @@ class ArtifactStore: ) def path(self, digest: str) -> Path | None: - """Where the bytes are, or None if this store does not have them. + """Where the stored bytes are, or None if this store does not have them. - The ring is looked in second, so everything that resolves a digest — - serving one over HTTP, checking a run input still exists, a panel's - scope — reaches a live frame without knowing there are two stores. + Only the durable store: a live frame has no file. ``has`` and ``read`` + are what reach both. """ if not valid_digest(digest): return None target = self._path(digest) - if target.exists(): - return target - if self.volatile is not None: - return self.volatile.path(digest) - return None + return target if target.exists() else None + + def has(self, digest: str) -> bool: + """Whether the bytes are here, stored or held in the ring. + + Everything that only asks whether a digest resolves — a run input + that has to still exist, a cached result — reaches a live frame + without knowing there are two stores. + """ + if self.path(digest) is not None: + return True + return self.volatile is not None and self.volatile.has(digest) def adopt(self, digest: str) -> bool: """Copy a volatile artifact into this store, so it outlives the ring. @@ -154,10 +164,10 @@ class ArtifactStore: return False if self._path(digest).exists(): return True - source = self.volatile.path(digest) - if source is None: + data = self.volatile.get(digest) + if data is None: return False - self.put_file(source) + self.put([data]) return True def delete(self, digest: str) -> bool: @@ -176,7 +186,11 @@ class ArtifactStore: def read(self, digest: str) -> Iterator[bytes]: target = self.path(digest) if target is None: - raise FileNotFoundError(digest) + data = self.volatile.get(digest) if self.volatile is not None else None + if data is None: + raise FileNotFoundError(digest) + yield data + return with target.open("rb") as handle: while chunk := handle.read(CHUNK): yield chunk @@ -211,44 +225,98 @@ class ArtifactStore: return removed -class VolatileStore(ArtifactStore): - """A bounded ring of artifacts, held wherever memory is cheaper than disk. +class VolatileStore: + """Live frames, held in memory and bounded by size. - A camera publishing ten frames a second is ten files a second, and on the - wall panel this is built for that disk is an SD card. So the frames go to a - memory-backed directory instead and the oldest fall out once the ring is - full: nothing sweeps it, because a frame nobody kept is not worth a pass - over the store to find. + A camera publishing five frames a second is five artifacts a second that + nobody keeps, so they never reach a disk: each one is held here in the + order it came and the oldest fall out once the bound is reached. Nothing + sweeps it, and a ``put`` costs the same however full it is. - It is an ``ArtifactStore``, digest layout and all, which is what lets a - frame be an ordinary reference — the dtype check, the panel's scope, a - node opening one with ``load_artifact`` and the widget fetching one all - work on it unchanged. + ``root`` is a memory-backed directory beside it for the frames a worker in + this container writes itself (``FLUKSIO_ARTIFACT_VOLATILE_DIR``), since a + worker is another process. Read from there once and held here after, so + a frame pushed to every open screen is read from the file once. The + engine trims that directory on a timer. + + A frame is an ordinary reference all the same: the dtype check, the + device's scope, a node opening one with ``load_artifact`` and the widget + fetching one all work on it unchanged, through ``ArtifactStore``. """ def __init__(self, root: Path, limit_bytes: int) -> None: - super().__init__(root) + self.root = root + self.root.mkdir(parents=True, exist_ok=True) self.limit_bytes = limit_bytes + self._frames: OrderedDict[str, bytes] = OrderedDict() + self._size = 0 + self._lock = threading.Lock() def put( self, chunks: Iterable[bytes], name: str = "", media_type: str = "", - volatile: bool = False, ) -> dict[str, Any]: - reference = super().put(chunks, name=name, media_type=media_type) - self.trim() - return reference + """Hold a frame and return the reference to it.""" + data = b"".join(chunks) + digest = DIGEST_PREFIX + hashlib.sha256(data).hexdigest() + self._hold(digest, data) + return { + "digest": digest, + "size": len(data), + "media_type": media_type or "application/octet-stream", + "name": name, + } + + def _hold(self, digest: str, data: bytes) -> None: + with self._lock: + if digest in self._frames: + # The same bytes again are the newest frame again. + self._frames.move_to_end(digest) + return + self._frames[digest] = data + self._size += len(data) + # The one just held stays, even past the bound: its reference is + # about to be published. + while self._size > self.limit_bytes and len(self._frames) > 1: + _, dropped = self._frames.popitem(last=False) + self._size -= len(dropped) + + def _file(self, digest: str) -> Path: + body = digest[len(DIGEST_PREFIX) :] + return self.root / body[:2] / body + + def get(self, digest: str) -> bytes | None: + """The frame's bytes, or None once it has fallen out.""" + with self._lock: + data = self._frames.get(digest) + if data is not None or not valid_digest(digest): + return data + try: + data = self._file(digest).read_bytes() + except OSError: + return None + self._hold(digest, data) + return data + + def has(self, digest: str) -> bool: + with self._lock: + if digest in self._frames: + return True + return valid_digest(digest) and self._file(digest).is_file() + + @property + def size(self) -> int: + """Bytes held in memory.""" + return self._size def trim(self) -> int: - """Drop the oldest until the ring is inside its bound. + """Drop the oldest of the workers' frames until the directory is + inside the bound. Returns how many went. - Also called on a timer, because a worker in this container writes here - itself and the engine never sees that ``put``. - - # ponytail: a scandir per trim. An in-memory index if a ring of - # thousands of frames ever shows up in a profile. + A scan of the directory, so it runs on a timer and never per frame: + only frames a worker in this container wrote are there. """ if self.limit_bytes <= 0: return 0 diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 2492a00..ee8fdb3 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -35,6 +35,7 @@ from fluksio_worker.worker_main import load_function from fluksio.core.config import settings from fluksio.flow.alerts import AlertManager +from fluksio.flow.demand import demand from fluksio.flow.events import EventBus from fluksio.flow.executor import ExecutionService from fluksio.flow.messages import MessageSpec, flow_of, qualify @@ -853,6 +854,9 @@ class FlowController: ) finally: self._lock.release() + # A device created, deleted or reporting changes whose ports send on + # demand. + demand.refresh() self._publish( { "type": "pipeline_rebuilt", @@ -868,9 +872,13 @@ class FlowController: An empty dashboard name is what a device refetches its own document on, and what its socket rescopes on. The first build only takes note: a device connecting later reads the document as it is. + + What a running flow reads may have moved with it, so demand on the + devices' ports is worked out again too. """ if self.devices is None: return + demand.refresh() wiring = self.devices.wiring() known, self._device_wiring = self._device_wiring, wiring if known is not None and known != wiring: @@ -1488,6 +1496,23 @@ class FlowController: def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]: return self.pipeline.values(flow) if self.pipeline else {} + def consumed_messages(self) -> set[str]: + """The messages a node of a running flow reads. + + What keeps a device's on-demand port sending whether or not anyone is + watching: a node that reads a camera wants every frame. A stopped or + batch flow reads nothing until it runs, and a device node is passive. + """ + pipeline = self.pipeline + if pipeline is None: + return set() + idle = self.disabled | self.batch + return { + name + for name, nodes in pipeline.consumes.items() + if any(node.flow not in idle for node in nodes) + } + def get_node(self, node_id: str) -> Node | None: entry = self.loaded.get(node_id) return entry.node if entry else None diff --git a/backend/fluksio/flow/dashboards.py b/backend/fluksio/flow/dashboards.py index 9ea244c..7286940 100644 --- a/backend/fluksio/flow/dashboards.py +++ b/backend/fluksio/flow/dashboards.py @@ -655,6 +655,10 @@ class DashboardStore: self.root = flows.root / DASHBOARD_DIR # Read-modify-write of the version counter, same as the flow store. self._lock = threading.Lock() + #: Published documents as last read, until the flow store commits. + self._published: dict[str, DashboardDef | None] = {} + self._published_at = -1 + self._published_lock = threading.Lock() def _file(self, name: str) -> Path: return self.root / checked(name) / "dashboard.json" @@ -719,6 +723,24 @@ class DashboardStore: update={"has_draft": self.has_draft(name)} ) + def published(self, name: str) -> DashboardDef | None: + """The published document, or None, as of the store's last commit. + + What a device's credential is held against on every request, a camera + fetching frame by frame included, so it is read once per commit rather + than from disk each time. Every write commits. The document is shared: + read it, never change it. + """ + with self._published_lock: + if self._published_at != self.flows.revision: + self._published, self._published_at = {}, self.flows.revision + if name not in self._published: + try: + self._published[name] = self.read(name) + except DashboardNotFound: + self._published[name] = None + return self._published[name] + def write( self, defn: DashboardDef, base_version: int | None = None ) -> DashboardDef: diff --git a/backend/fluksio/flow/demand.py b/backend/fluksio/flow/demand.py new file mode 100644 index 0000000..613045b --- /dev/null +++ b/backend/fluksio/flow/demand.py @@ -0,0 +1,114 @@ +"""Which of a device's ports someone wants right now. + +A camera is sent nothing while nobody watches it. A port the device reports as +``on_demand`` has demand while a viewer socket wants bytes for the message it +is wired to — a dashboard tile, the Devices tab's preview, a screen through the +portal — or a node of a running flow reads that message. + +Every socket says what it wants here, in one registry, and the result is +published on the bus as a ``device_demand`` event whenever a device's set +changes. Each device socket picks out its own and tells the device, so nothing +here has to know which sockets belong to which device. +""" + +from __future__ import annotations + +import threading +import time +from collections.abc import Hashable +from typing import Any + +from fluksio.flow.events import event_bus + + +class Demand: + """The media names every live socket wants, and what that asks of devices. + + One per process, as the sockets are. ``bind`` hands it the controller, + whose device store says what is wired and whose graph says what a flow + reads. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self._wanted: dict[Hashable, frozenset[str]] = {} + #: The last set published per device, so only a change is. + self._sent: dict[str, tuple[str, ...]] = {} + self._refresh_lock = threading.Lock() + self._controller: Any = None + + def bind(self, controller: Any) -> None: + self._controller = controller + self.refresh() + + def want(self, viewer: Hashable, names: set[str]) -> None: + """What one socket wants bytes for now, replacing what it said before.""" + with self._lock: + if names: + self._wanted[viewer] = frozenset(names) + else: + self._wanted.pop(viewer, None) + self.refresh() + + def drop(self, viewer: Hashable) -> None: + """A socket closed, and wants nothing any more.""" + self.want(viewer, set()) + + def viewed(self) -> set[str]: + with self._lock: + return set().union(*self._wanted.values()) + + def ports(self, device_id: str) -> list[str]: + """The device's ports that have demand, sorted.""" + return list(self._compute().get(device_id, ())) + + def refresh(self) -> None: + """Work demand out again and publish what changed. + + Called whenever its inputs move: a socket's wanted set, and a rebuild, + a started or stopped flow or a new report, which change what is wired + and what reads it. + """ + with self._refresh_lock: + found = self._compute() + for device_id in sorted(set(found) | set(self._sent)): + ports = found.get(device_id, ()) + if ports == self._sent.get(device_id, ()): + continue + if ports: + self._sent[device_id] = ports + else: + self._sent.pop(device_id, None) + event_bus.publish( + { + "type": "device_demand", + "device": device_id, + "ports": list(ports), + "ts": time.time(), + } + ) + + def _compute(self) -> dict[str, tuple[str, ...]]: + controller = self._controller + devices = getattr(controller, "devices", None) + if devices is None: + return {} + wanted = self.viewed() | controller.consumed_messages() + found: dict[str, tuple[str, ...]] = {} + for device_id in devices.wiring(): + reported = devices.reported_ports(device_id) or {} + ports = tuple( + sorted( + binding.port + for binding in devices.bound(device_id) + if binding.direction == "in" + and binding.message in wanted + and reported.get(binding.port, {}).get("on_demand") is True + ) + ) + if ports: + found[device_id] = ports + return found + + +demand = Demand() diff --git a/backend/fluksio/flow/devices.py b/backend/fluksio/flow/devices.py index 7b0a8d5..6698b45 100644 --- a/backend/fluksio/flow/devices.py +++ b/backend/fluksio/flow/devices.py @@ -45,7 +45,7 @@ from pydantic import ( model_validator, ) -from fluksio.flow.dashboards import DashboardDef, DashboardNotFound, DashboardStore +from fluksio.flow.dashboards import DashboardDef, DashboardStore from fluksio.flow.events import event_bus from fluksio.flow.messages import ( DType, @@ -985,13 +985,7 @@ def dashboards_for( """ if device is None: return [] - found: list[DashboardDef] = [] - for name in device.dashboards: - try: - found.append(store.read(name)) - except DashboardNotFound: - continue - return found + return [defn for name in device.dashboards if (defn := store.published(name))] def messages_of(defn: DashboardDef) -> set[str]: diff --git a/backend/fluksio/flow/events.py b/backend/fluksio/flow/events.py index 4acc744..98c56bc 100644 --- a/backend/fluksio/flow/events.py +++ b/backend/fluksio/flow/events.py @@ -19,6 +19,8 @@ logger = logging.getLogger(__name__) QUEUE_SIZE = 256 LOG_HISTORY = 400 +#: How many of a message's last artifact digests are remembered. +RECENT_DIGESTS = 8 class EventBus: @@ -42,6 +44,11 @@ class EventBus: # zero and draw a busy graph as idle. A session count, not a metric — # the rollups in `metrics.py` are what survives a restart. self.emits: dict[str, int] = {} + #: The last few artifacts each message pointed at, newest last. A + #: screen fetching a camera frame by frame asks for one the message + #: may already have moved past, and a device's credential reaches + #: these as well as the current one. + self.digests: dict[str, deque[str]] = {} def bind(self, loop: asyncio.AbstractEventLoop) -> None: """Attach the bus to the running event loop (called once at startup).""" @@ -59,6 +66,14 @@ class EventBus: # increment, which is a count nobody is going to miss. node = str(event.get("node") or "") self.emits[node] = self.emits.get(node, 0) + 1 + elif kind == "message_value": + value = event.get("value") + if isinstance(value, dict) and isinstance(value.get("digest"), str): + name = str(event.get("name") or "") + recent = self.digests.get(name) + if recent is None: + recent = self.digests[name] = deque(maxlen=RECENT_DIGESTS) + recent.append(value["digest"]) loop = self._loop if loop is None or not self._subscribers: return diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 5145ff8..698080e 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -306,7 +306,7 @@ def _from_run( ).all() for row in rows: if output in (row.name, row.name.rsplit(".", 1)[-1]): - if artifacts is not None and artifacts.path(row.digest) is None: + if artifacts is not None and not artifacts.has(row.digest): raise RunRejected( f"Parameter '{key}': run '{run_id}' made '{output}', but its " "bytes are gone from this instance's store" @@ -337,7 +337,7 @@ def _from_digest( f"Parameter '{key}': no run has produced '{digest}', so there is " "nothing here under that digest" ) - if artifacts is not None and artifacts.path(digest) is None: + if artifacts is not None and not artifacts.has(digest): raise RunRejected( f"Parameter '{key}': '{digest}' is known but its bytes are gone " "from this instance's store" @@ -829,10 +829,7 @@ class RunCache: def _unrestorable(self, outputs: dict[str, Any]) -> bool: return any( is_reference(value) - and ( - self.artifacts is None - or self.artifacts.path(str(value["digest"])) is None - ) + and (self.artifacts is None or not self.artifacts.has(str(value["digest"]))) for value in outputs.values() ) diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 0b1e44c..26fddf8 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -34,6 +34,7 @@ from fluksio.flow.alerts import AlertManager from fluksio.flow.artifacts import ArtifactStore, VolatileStore from fluksio.flow.controller import FlowController, RebuildBusy from fluksio.flow.dashboards import DashboardStore +from fluksio.flow.demand import demand from fluksio.flow.devices import DeviceStore, DeviceUnavailable from fluksio.flow.events import event_bus from fluksio.flow.executor import ExecutionService @@ -159,14 +160,13 @@ VOLATILE_TRIM_S = 5.0 async def _trim_volatile(ring: VolatileStore) -> None: - """Hold the volatile ring inside its bound. + """Hold the workers' frame directory inside its bound. - ``put`` trims what it wrote, but a worker in this container writes to the - ring itself and the engine never sees that one — so the bound needs - something of its own watching it. + The engine's own frames are held in memory and bounded as they arrive, but + a worker in this container writes its frames to a directory and the + engine never sees that ``put`` — so the bound needs something of its own + watching it. """ - if ring.limit_bytes <= 0: - return while True: await asyncio.sleep(VOLATILE_TRIM_S) try: @@ -190,6 +190,17 @@ def _volatile_root() -> Path: return parent / f"fluksio-volatile-{tag}" +def _volatile_ring() -> VolatileStore | None: + """The ring of live frames, or None when its bound is zero. + + Without one a volatile save is an ordinary one, and a worker writes its + frames to the durable store too. + """ + if settings.ARTIFACT_VOLATILE_BYTES <= 0: + return None + return VolatileStore(_volatile_root(), settings.ARTIFACT_VOLATILE_BYTES) + + def _mcp_sessions() -> AbstractAsyncContextManager[None]: """The MCP session manager's run scope, or nothing when MCP is off.""" if not settings.MCP_ENABLED: @@ -285,7 +296,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Frames a flow only shows live, held in memory and bounded by size. # The store falls through to it, so a volatile reference is an # ordinary one everywhere but in how long its bytes last. - volatile = VolatileStore(_volatile_root(), settings.ARTIFACT_VOLATILE_BYTES) + volatile = _volatile_ring() artifacts.volatile = volatile app.state.artifact_store = artifacts # Zero means "work it out" for cards and memory, as it always has for cores. @@ -309,7 +320,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # is given a URL instead. Node code calls the same two functions. env={ ARTIFACT_DIR_ENV: str(artifacts.root), - ARTIFACT_VOLATILE_DIR_ENV: str(volatile.root), + **({ARTIFACT_VOLATILE_DIR_ENV: str(volatile.root)} if volatile else {}), # Every slot can be busy at once, so a worker left to size its own # thread pool to the machine means as many processes as there are # slots, each believing it has the whole of it. A node that says @@ -359,6 +370,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: await run_in_threadpool(devices.move_io) app.state.device_store = devices controller.devices = devices + # What the sockets want and the flows read, mapped onto the devices' + # on-demand ports. + demand.bind(controller) # Charts need a deeper series than the default; tell the engine # before it starts recording. controller.set_history_limits(dashboards.history_requirements()) @@ -398,7 +412,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app.state.metrics_collector = MetricsCollector(event_bus) _background(app.state.metrics_collector.run(), "metrics-collector") _background(_sweep_artifacts(artifacts, controller), "artifact-gc") - _background(_trim_volatile(volatile), "artifact-ring") + if volatile is not None: + _background(_trim_volatile(volatile), "artifact-ring") await controller.start() started.append(controller.stop) run_service.start() diff --git a/backend/tests/api/routes/test_cameras.py b/backend/tests/api/routes/test_cameras.py new file mode 100644 index 0000000..8e8c543 --- /dev/null +++ b/backend/tests/api/routes/test_cameras.py @@ -0,0 +1,363 @@ +"""Cameras on a device: frames over its socket, demand, and what a screen fetches. + +openpanel ROADMAP C1. A device reports a ``camera`` port that sends only while +somebody wants it. Demand is what the viewer sockets want plus what running +flows read, mapped onto the device's on-demand ports and sent to its socket. +""" + +import json +import time +from collections.abc import Callable +from typing import Any + +import orjson +import pytest +from fastapi.testclient import TestClient + +from fluksio.api.routes import oauth +from fluksio.core.config import settings +from fluksio.flow.demand import demand +from fluksio.flow.events import RECENT_DIGESTS, event_bus +from tests.api.routes.test_devices import ( + FLOWS, + _device_node, + _pair, + _panels, + _publish_flow, +) + +PREFIX = f"{settings.API_V1_STR}/devices" +SOCKET = f"{settings.API_V1_STR}/flows/ws" +ARTIFACTS = f"{settings.API_V1_STR}/artifacts" + +#: What the NANO reports, as C1 fixes it. +REPORT = { + "schema": 1, + "platform": "esp32p4", + "board": "nano", + "firmware": "0.2.0", + "ports": [ + {"port": "camera", "direction": "in", "dtype": "image", "on_demand": True}, + {"port": "person", "direction": "in", "dtype": "bool"}, + {"port": "detections", "direction": "in", "dtype": "json"}, + {"port": "snapshot", "direction": "in", "dtype": "image"}, + ], +} + +CAMERA_NODE = _device_node( + "cam1", + node="hall", + provides=[ + {"name": "camera", "port": "camera", "dtype": "image"}, + {"name": "person", "port": "person", "dtype": "bool"}, + {"name": "snapshot", "port": "snapshot", "dtype": "image"}, + ], +) + +#: A node reading the camera's every frame, which keeps it on for good. +READER = { + "id": "watch", + "type": "debounce", + "params": {"wait": 5}, + "requires": [{"name": "camera", "dtype": "image"}], + "provides": [{"name": "moving", "dtype": "bool"}], +} + + +@pytest.fixture(autouse=True) +def _pairing_unlimited() -> None: + """Every test pairs from the test client's one address.""" + for bucket in [b for b in oauth._hits if b.startswith("pair:")]: + del oauth._hits[bucket] + + +def frame(name: str, data: bytes, media_type: str = "image/jpeg") -> bytes: + """A binary frame as a device sends it.""" + header = orjson.dumps( + {"type": "frame", "name": name, "media_type": media_type, "ts": 1234} + ) + return len(header).to_bytes(4, "big") + header + data + + +def wait_for(check: Callable[[], Any], timeout: float = 5.0) -> Any: + deadline = time.time() + timeout + while time.time() < deadline: + found = check() + if found: + return found + time.sleep(0.02) + raise AssertionError("timed out") + + +@pytest.fixture +def camera(client: TestClient, superuser_token_headers: dict[str, str]): + """A paired device with a camera, wired in flow ``cams``. Yields its header.""" + _panels(client, superuser_token_headers, {"panels": [{"id": "cam1"}]}) + device = _pair(client, superuser_token_headers, "cam1") + reported = client.put(f"{PREFIX}/cam1/report", headers=device, json=REPORT) + assert reported.status_code == 200, reported.text + _publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE]) + yield device + client.delete(f"{FLOWS}/cams", headers=superuser_token_headers) + client.delete(f"{PREFIX}/cam1", headers=superuser_token_headers) + + +def _ticket(client: TestClient, headers: dict[str, str]) -> str: + answer = client.post(f"{settings.API_V1_STR}/flows/ws-ticket", headers=headers) + return f"{SOCKET}?ticket={answer.json()['ticket']}" + + +def _next_demand(ws: Any) -> list[str]: + """The next demand message on a device's socket, skipping values.""" + while True: + message = ws.receive_json() + if message.get("type") == "demand": + return message["ports"] + + +def test_a_tile_asking_for_the_camera_turns_it_on_and_off( + client: TestClient, superuser_token_headers: dict[str, str], camera +) -> None: + """Demand follows what viewers draw, and only on a port sent on demand.""" + with client.websocket_connect(SOCKET, headers=camera) as device: + snapshot = device.receive_json() + assert snapshot["demand"] == [] + assert snapshot["uplink"] == "binary" + + with client.websocket_connect( + _ticket(client, superuser_token_headers) + ) as viewer: + viewer.receive_json() + # `person` is drawn too, but it sends all the time anyway. + viewer.send_json({"type": "media", "names": ["cams.camera", "cams.person"]}) + assert _next_demand(device) == ["camera"] + # The tile scrolled away: the page asks for nothing. + viewer.send_json({"type": "media", "names": []}) + assert _next_demand(device) == [] + viewer.send_json({"type": "media", "names": ["cams.camera"]}) + assert _next_demand(device) == ["camera"] + # The viewer closed its socket. + assert _next_demand(device) == [] + + # A snapshot taken while someone watches says so. + demand.want("someone", {"cams.camera"}) + try: + assert _next_demand(device) == ["camera"] + with client.websocket_connect(SOCKET, headers=camera) as again: + assert again.receive_json()["demand"] == ["camera"] + finally: + demand.drop("someone") + + +@pytest.mark.usefixtures("camera") +def test_a_flow_reading_the_camera_keeps_it_on( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A running flow that reads frames wants every one, watched or not.""" + assert demand.ports("cam1") == [] + _publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE, READER]) + assert demand.ports("cam1") == ["camera"] + + stopped = client.post(f"{FLOWS}/cams/stop", headers=superuser_token_headers) + assert stopped.status_code == 200, stopped.text + assert demand.ports("cam1") == [] + client.post(f"{FLOWS}/cams/start", headers=superuser_token_headers) + assert demand.ports("cam1") == ["camera"] + + # A port the device does not send on demand never has any. + _publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE]) + demand.want("someone", {"cams.snapshot", "cams.person"}) + try: + assert demand.ports("cam1") == [] + finally: + demand.drop("someone") + + +def test_a_device_sends_frames_on_its_socket( + client: TestClient, superuser_token_headers: dict[str, str], camera +) -> None: + """Held and published in one step, credited to its node, pushed to tiles.""" + with ( + client.websocket_connect(SOCKET, headers=camera) as device, + client.websocket_connect(_ticket(client, superuser_token_headers)) as viewer, + ): + device.receive_json() + viewer.receive_json() + viewer.send_json({"type": "media", "names": ["cams.camera"]}) + _next_demand(device) + + device.send_bytes(frame("cams.camera", b"\xff\xd8 a jpeg")) + + pushed = viewer.receive_bytes() + length = int.from_bytes(pushed[:4], "big") + header = orjson.loads(pushed[4 : 4 + length]) + assert header["name"] == "cams.camera" + assert header["media_type"] == "image/jpeg" + assert pushed[4 + length :] == b"\xff\xd8 a jpeg" + + value = viewer.receive_json() + assert value["type"] == "message_value" + assert value["value"]["digest"] == header["digest"] + assert value["source"]["kind"] == "node" + assert value["source"]["id"] == "cams.hall" + store = client.app.state.artifact_store + assert store.volatile.get(header["digest"]) == b"\xff\xd8 a jpeg" + # In memory, and not on the volume. + assert store.path(header["digest"]) is None + + # The device is not sent its own camera's values back, only the rest. + from fluksio.api.routes.flows import panel_scope + + scope = panel_scope(camera["Authorization"][7:], client.app) + assert scope == {"cams.person"} + + +def test_a_frame_the_device_may_not_send_is_dropped( + client: TestClient, + superuser_token_headers: dict[str, str], + camera, + caplog: pytest.LogCaptureFixture, +) -> None: + """Refused, logged and dropped, and the socket carries on.""" + controller = client.app.state.flow_controller + before = controller.values().get("cams.person") + with client.websocket_connect(SOCKET, headers=camera) as device: + device.receive_json() + for refused in ( + frame("cams.person", b"\xff\xd8"), # a bool port + frame("cams.elsewhere", b"\xff\xd8"), # not one of its ports + frame("cams.camera", b"", "text/html"), # not an image + b"\x00\x00\x00\x09not json!", + b"\x01", + ): + device.send_bytes(refused) + # Still open: a frame it may send goes through after them. + device.send_bytes(frame("cams.camera", b"\xff\xd8 after")) + wait_for(lambda: (controller.values().get("cams.camera") or {}).get("value")) + assert controller.values().get("cams.person") == before + + # A person's socket is no camera. + with client.websocket_connect(_ticket(client, superuser_token_headers)) as ws: + ws.receive_json() + ws.send_bytes(frame("cams.camera", b"\xff\xd8 forged")) + ws.send_json({"type": "media", "names": []}) + current = controller.values()["cams.camera"]["value"] + data = client.app.state.artifact_store.volatile.get(current["digest"]) + assert data == b"\xff\xd8 after" + said = [r.getMessage() for r in caplog.records if "Dropped a frame" in r.message] + assert any("'person' is wired as bool" in line for line in said) + assert any("not a message one of its ports publishes" in line for line in said) + assert any("only a device sends frames" in line for line in said) + + +def test_an_unpaired_device_is_closed_and_its_frames_stop( + client: TestClient, superuser_token_headers: dict[str, str], camera +) -> None: + from starlette.websockets import WebSocketDisconnect + + with client.websocket_connect(SOCKET, headers=camera) as device: + device.receive_json() + client.post(f"{PREFIX}/cam1/unpair", headers=superuser_token_headers) + with pytest.raises(WebSocketDisconnect) as closed: + while True: + device.receive_json() + assert closed.value.code == 1008 + + +def test_a_screen_may_fetch_the_last_few_frames(client: TestClient, camera) -> None: + """A remote screen fetching frame by frame is often one behind.""" + controller = client.app.state.flow_controller + store = client.app.state.artifact_store + refs = [] + for index in range(RECENT_DIGESTS + 2): + ref = store.put([b"frame %d" % index], media_type="image/jpeg", volatile=True) + controller.publish_message("cams.camera", ref) + refs.append(ref) + + def fetch(ref: dict[str, Any]) -> int: + return client.get(f"{ARTIFACTS}/{ref['digest']}", headers=camera).status_code + + assert [fetch(ref) for ref in refs[-RECENT_DIGESTS:]] == [200] * RECENT_DIGESTS + # Older than that, it is not what the screen is drawing. + assert fetch(refs[0]) == 403 + assert list(event_bus.digests["cams.camera"]) == [ + ref["digest"] for ref in refs[-RECENT_DIGESTS:] + ] + + +def test_a_viewer_asks_for_a_few_cameras_at_most() -> None: + from fluksio.api.routes.flows import MAX_MEDIA_NAMES, wanted_names + + names = [f"cams.c{i}" for i in range(MAX_MEDIA_NAMES + 3)] + asked = wanted_names(json.dumps({"type": "media", "names": names}), None) + assert asked == set(names[:MAX_MEDIA_NAMES]) + + +def test_viewers_through_the_portal_count_too( + client: TestClient, superuser_token_headers: dict[str, str], camera +) -> None: + """The tunnel carries text only, so a device there uploads over HTTP, and a + browser there asks for media like any other.""" + from fluksio.cloud.connector import CloudConnector + + connector = CloudConnector(client.app) + sent: dict[str, list[dict[str, Any]]] = {"dev": [], "view": []} + + class Socket: + def __init__(self, stream: str) -> None: + self.stream = stream + + async def send(self, text: str) -> None: + sent[self.stream].append(json.loads(text)) + + def texts(stream: str) -> list[dict[str, Any]]: + return [ + json.loads(frame["text"]) + for frame in sent[stream] + if frame["op"] == "ws_msg" + ] + + def demands() -> list[list[str]]: + return [t["ports"] for t in texts("dev") if t.get("type") == "demand"] + + token = camera["Authorization"][7:] + device = client.portal.start_task_soon( + connector._stream, + Socket("dev"), + {"id": "dev", "path": "/api/v1/flows/ws", "query": f"token={token}"}, + ) + try: + snapshot = wait_for(lambda: texts("dev"))[0] + assert snapshot["uplink"] == "http" + assert snapshot["demand"] == [] + + ticket = _ticket(client, superuser_token_headers).split("?", 1)[1] + viewer = client.portal.start_task_soon( + connector._stream, + Socket("view"), + {"id": "view", "path": "/api/v1/flows/ws", "query": ticket}, + ) + wait_for(lambda: texts("view")) + asking = json.dumps({"type": "media", "names": ["cams.camera"]}) + client.portal.call(_say, connector, "view", asking) + assert wait_for(demands)[-1] == ["camera"] + + viewer.cancel() + wait_for(lambda: len(demands()) == 2) + assert demands()[-1] == [] + finally: + device.cancel() + + +async def _say(connector: Any, stream: str, text: str) -> None: + """What the hub delivers when the browser behind a stream speaks.""" + connector._listeners[stream](text) + + +def test_a_zero_bound_turns_the_ring_off(monkeypatch: pytest.MonkeyPatch) -> None: + from fluksio.main import _volatile_ring + + monkeypatch.setattr(settings, "ARTIFACT_VOLATILE_BYTES", 0) + assert _volatile_ring() is None + monkeypatch.setattr(settings, "ARTIFACT_VOLATILE_BYTES", 1000) + assert _volatile_ring() is not None diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index 5283802..9250335 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/backend/tests/api/routes/test_flows.py @@ -2,7 +2,6 @@ import asyncio from datetime import UTC, datetime import pytest -from fastapi import WebSocketDisconnect from fastapi.testclient import TestClient from sqlalchemy import func from sqlmodel import Session, select @@ -538,10 +537,10 @@ async def test_the_live_socket_ends_when_its_client_goes( async def send_text(self, text: str) -> None: self.sent.append(text) - async def receive_text(self) -> str: + async def receive(self) -> dict[str, object]: if self.frames: - return self.frames.pop(0) - raise WebSocketDisconnect(1006) + return {"type": "websocket.receive", "text": self.frames.pop(0)} + return {"type": "websocket.disconnect", "code": 1006} peer = Peer() subscribed = len(event_bus._subscribers) diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index 0a4c47e..b24843a 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -790,7 +790,7 @@ def test_a_node_saves_a_frame_to_the_ring(tmp_path): assert MessageSpec(name="frame", dtype=DType.IMAGE).check(ref) is None # In the ring, and nowhere near the data volume. - assert store.volatile.path(ref["digest"]) is not None + assert store.volatile.has(ref["digest"]) assert not (store.root / ref["digest"][7:9]).exists() read = pool.run( diff --git a/backend/tests/test_artifacts.py b/backend/tests/test_artifacts.py index c23a27d..857a80f 100644 --- a/backend/tests/test_artifacts.py +++ b/backend/tests/test_artifacts.py @@ -141,25 +141,59 @@ def test_a_ring_drops_the_oldest_once_it_is_full(tmp_path): ring = VolatileStore(tmp_path / "ring", limit_bytes=3000) refs = [ring.put([bytes([i]) * 1000], name=f"{i}.bin") for i in range(5)] - held = [ref for ref in refs if ring.path(ref["digest"]) is not None] + held = [ref for ref in refs if ring.has(ref["digest"])] assert len(held) == 3 # The newest three, in order: eviction is by age, not by chance. assert held == refs[2:] + assert ring.size == 3000 + # Held in memory: nothing reached the directory. + assert not any(ring.root.iterdir()) + + +def test_the_newest_frame_stays_whatever_its_size(tmp_path): + """A frame larger than the bound is still there for its own publish.""" + ring = VolatileStore(tmp_path / "ring", limit_bytes=100) + small = ring.put([b"s" * 50]) + big = ring.put([b"b" * 500]) + + assert ring.get(big["digest"]) == b"b" * 500 + assert not ring.has(small["digest"]) + + +def test_a_workers_frame_is_read_from_its_directory_once(tmp_path): + """A worker in the container writes its frames to the ring's directory.""" + import hashlib + + ring = VolatileStore(tmp_path / "ring", limit_bytes=1_000_000) + data = b"a worker's frame" + body = hashlib.sha256(data).hexdigest() + path = ring.root / body[:2] / body + path.parent.mkdir(parents=True) + path.write_bytes(data) + + assert ring.has("sha256:" + body) + assert ring.get("sha256:" + body) == data + # Held after the first read, so every screen is served from memory. + path.unlink() + assert ring.get("sha256:" + body) == data def test_a_volatile_reference_resolves_like_any_other(tmp_path): """A frame is an ordinary reference, which is what keeps the rest honest. - Everything that resolves a digest — serving one, checking a run's input, - a panel's scope — goes through ``path``, so the ring has to answer there. + Everything that asks whether a digest resolves — checking a run's input, + a cached result — goes through ``has``, and reading one through ``read``, + so the ring has to answer there. """ store = ArtifactStore(tmp_path / "artifacts") store.volatile = VolatileStore(tmp_path / "ring", limit_bytes=1_000_000) ref = store.put([b"frame"], name="f.png", media_type="image/png", volatile=True) - assert store.path(ref["digest"]) is not None + assert store.has(ref["digest"]) + assert b"".join(store.read(ref["digest"])) == b"frame" # In the ring, and not on the volume the sweep is about. + assert store.path(ref["digest"]) is None assert not (store.root / ref["digest"][7:9]).exists() @@ -171,7 +205,7 @@ def test_a_recorded_frame_is_copied_out_of_the_ring(tmp_path): assert store.adopt(ref["digest"]) is True - store.volatile.collect(set()) + store.volatile = VolatileStore(tmp_path / "ring2", limit_bytes=1_000_000) assert store.path(ref["digest"]) is not None assert store.adopt("sha256:" + "0" * 64) is False @@ -195,7 +229,7 @@ def test_an_upload_may_ask_for_the_ring( assert answer.status_code == 200, answer.text digest = answer.json()["digest"] - assert store.volatile.path(digest) is not None + assert store.volatile.has(digest) assert not (store.root / digest[7:9]).exists() # And it is served back like anything else, which is what the widget # falls back on when the socket did not push it. diff --git a/docker/compose.dev.yml b/docker/compose.dev.yml index 4c3f486..a45d111 100644 --- a/docker/compose.dev.yml +++ b/docker/compose.dev.yml @@ -94,19 +94,24 @@ services: build: context: .. dockerfile: backend/Dockerfile - # `fastapi dev`, because `run` takes no `--reload-dir`. Watching the worker + # uvicorn itself, because `fastapi dev` cannot turn per-message deflate + # off, which `fluksio serve` does for the media frames. Watching the worker # too makes a worker edit restart the engine, which respawns the local # worker pool with the new code. command: - - fastapi - - dev + - uvicorn + - fluksio.main:app + - --reload - --host - 0.0.0.0 + - --port + - "8000" + - --ws-per-message-deflate + - "false" - --reload-dir - /app/backend - --reload-dir - /app/worker - - "fluksio/main.py" develop: watch: - path: ../backend diff --git a/docs/code/api.md b/docs/code/api.md index 163899f..56a45bd 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -173,6 +173,8 @@ curl "$FLUKSIO/artifacts/" \ The upload returns a reference with the content digest. Identical bytes are stored once. Node code uses `save_artifact` and `load_artifact` instead. +`?volatile=true` holds the bytes in the engine's memory for a live frame, where +newer frames push them out, instead of storing them. A superuser can remove stored bytes with `DELETE /artifacts/`. While a run, a current message or a flow input's initial value still references the @@ -249,6 +251,30 @@ Browsers first call `POST /flows/ws-ticket`, then connect with `?ticket=`. Tickets are single-use and expire after 30 seconds. The socket sends a snapshot followed by node, value, health, and run events. +A client asks for the bytes of the media it shows with +`{"type": "media", "names": [...]}`, which replaces what it asked for before, +for at most 8 messages. Each frame held in memory then arrives as a binary +message before the value naming it: four bytes of header length (big endian), +a JSON header `{"type": "media", "name", "digest", "media_type", "ts"}`, then +the bytes. Messages are sent without per-message compression. + +A device's socket is bounded to its own messages. Its snapshot carries +`"demand"`, the ports it reported as `on_demand` that someone wants now, and +`"uplink"`, `"binary"` on its own socket and `"http"` through the portal. When +that set changes it is sent `{"type": "demand", "ports": [...]}`. A port has +demand while a socket asks for the media of the message it is wired to, or a +node of a running flow reads that message. + +With a `binary` uplink the device sends a camera frame as a binary message in +the same layout, with the header `{"type": "frame", "name": "", +"media_type": "image/jpeg", "ts": }`. The message has to be one of +its `in` ports wired as `image`. The frame is held in memory and published +in one step, credited to its device node. A frame that does not qualify is +logged and dropped and the socket stays open. With `http` it uploads with +`PUT /artifacts?volatile=true&name=` and publishes with +`POST /messages/`. A device that is unpaired or deleted while +connected is closed with 1008. + ## Errors and client generation | Code | Meaning | diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index 950c3af..2833b50 100644 --- a/docs/interface/dashboards.md +++ b/docs/interface/dashboards.md @@ -203,6 +203,13 @@ For local streams, `volatile=True` artifacts can be pushed over the websocket with their message. Stored artifacts are fetched individually, as are frames viewed through a portal. Use occasional snapshots for remote screens. +A tile asks for frames only while it is on screen in a tab that is being looked +at, and stops when it scrolls away, the tab is hidden or the tile closes. A +clip keeps playing either way. One page is pushed frames for at most 8 +messages at once. A ninth tile fetches its frames one by one and does not turn +a camera on. A [device's camera](#a-devices-camera) sends only while a tile like +this wants it. + For continuous video, set **Live stream** to a browser-compatible camera URL. The browser must be able to reach that URL; enrollment does not expose the camera itself. See [Media payloads](../reference/payload-types.md#image-audio-video). @@ -591,6 +598,34 @@ the dashboard settings, named after the device, and the first report of a touch screen turns **Touch friendly** on. A device paired before this version keeps the switch as it was set. +### A device's camera + +A camera device (openpanel on a NANO) reports four ports: + +| Port | Type | Settings | What it sends | +|---|---|---|---| +| `camera` | `image`, on demand | `width` (640), `fps` (5), `idle_fps` (0, off), `quality` (65) | JPEG frames, held in memory and never stored | +| `person` | `bool` | `hold_s` (10) | whether someone is in view, held for `hold_s` after they leave | +| `detections` | `json` | | `{count, boxes: [{score, x0, y0, x1, y1}]}`, boxes scaled 0 to 1, on change and at most once a second | +| `snapshot` | `image` | | one stored frame each time `person` turns true | + +`camera` sends only while it has demand: while a page wants its frames (a +dashboard tile, the preview on the device's page, a screen through the +portal), or a node of a running flow reads its message. Once its flow wires +`camera`, the device's page shows a **Camera** preview. A flow that reads +`camera` keeps it sending all the time, about 300 GB a month through a cloud +instance, so the Device node warns when a node in its flow reads a port that +sends on demand. Have flows read `snapshot` or `person` instead. + +The device is told which of its ports have demand on its live socket, and sends +its frames there too, as described under [Live events](../code/api.md#live-events). +Through the portal the socket carries text only, so a device paired there +uploads each frame over HTTP instead. + +A device may fetch the last 8 artifacts each of its messages pointed at, not +only the current one, so a screen fetching a camera frame by frame through the +portal is not refused a frame the camera has just replaced. + ### Before device nodes Before device nodes a device carried its ports itself. On its first start an diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index d35c21a..22e1940 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -167,14 +167,16 @@ MIG entries are passed on as written. | `MAX_ARTIFACT_BYTES` | `2147483648` (2 GiB) | upload limit; `0` removes it | | `ARTIFACT_GC_INTERVAL_S` | `3600` | cleanup interval; `0` disables | | `ARTIFACT_GC_GRACE_S` | `3600` | minimum age before cleanup | -| `ARTIFACT_VOLATILE_DIR` | auto-detected | live-media directory under `/dev/shm`, or the temporary directory | -| `ARTIFACT_VOLATILE_BYTES` | `50331648` (48 MiB) | live-media ring size; `0` stores volatile saves durably | +| `ARTIFACT_VOLATILE_DIR` | auto-detected | where local workers write live frames, under `/dev/shm` or the temporary directory | +| `ARTIFACT_VOLATILE_BYTES` | `50331648` (48 MiB) | live-media ring size, in the engine's memory and again in that directory; `0` stores volatile saves durably | Cleanup removes durable artifacts no longer referenced by recorded runs or -current messages, and pauses while a run is active. Volatile artifacts use a -bounded ring; the oldest disappear as it fills. Recorded run outputs are -copied into durable storage. If you increase the ring size in Docker, increase -`shm_size` too. +current messages, and pauses while a run is active. Volatile artifacts live in +a bounded ring in the engine's memory, and the oldest disappear as it fills. +Recorded run outputs are copied into durable storage. A worker in the engine's +container writes its frames to `ARTIFACT_VOLATILE_DIR` instead, which the +engine trims to the same size every few seconds. If workers publish frames +under Docker, raise `shm_size` with the ring size. ## Agents diff --git a/frontend/src/components/Dashboard/MediaWidget.tsx b/frontend/src/components/Dashboard/MediaWidget.tsx index 7fd34c3..77f36cf 100644 --- a/frontend/src/components/Dashboard/MediaWidget.tsx +++ b/frontend/src/components/Dashboard/MediaWidget.tsx @@ -1,3 +1,6 @@ +import { useRef } from "react" + +import { useOnScreen } from "@/hooks/useOnScreen" import { isRef, useArtifactUrl } from "@/lib/media" import { cn } from "@/lib/utils" import { useBoundValue } from "./dataContext" @@ -15,74 +18,126 @@ import type { WidgetProps } from "./widgets" */ export function MediaWidget({ widget }: WidgetProps) { const cfg = config(widget) - const message = text(cfg.message) - const stream = text(cfg.stream_url) - const live = useBoundValue(message || undefined) - const value = live?.value - const ref = isRef(value) ? value : null - const url = useArtifactUrl(ref, message || undefined) - - const kind = (ref?.media_type ?? text(cfg.dtype)).split("/")[0] - const fit = text(cfg.fit) === "contain" ? "object-contain" : "object-cover" - const label = widget.title || ref?.name || message - - // A camera serving its own stream is played from source: the engine carries - // references at whatever rate the flow publishes them, which is not video. - if (stream && kind !== "audio") { - return ( - {label} - ) - } - - if (!message && !stream) { - return

Pick a message.

- } - if (!ref) { - return

Nothing published yet.

- } - if (!url) { - return

Loading…

- } - - if (kind === "audio") { - return ( - - ) - } - - if (kind === "video") { - return ( - - ) - } - return ( - {label} ) } + +/** + * The tile's body, for anywhere a message's media is shown. + * + * Frames are asked for only while the tile is on screen, in a tab being + * looked at: a camera is only sent while somebody watches it, and a tile + * scrolled away or behind another tab is nobody. A clip keeps playing either + * way. + */ +export function MediaView({ + message, + stream = "", + dtype = "", + fit = "cover", + autoplay = false, + label, +}: { + message: string + stream?: string + dtype?: string + fit?: "cover" | "contain" + autoplay?: boolean + label?: string +}) { + const box = useRef(null) + const shown = useOnScreen(box) + const live = useBoundValue(message || undefined) + const value = live?.value + const ref = isRef(value) ? value : null + const kind = (ref?.media_type ?? dtype).split("/")[0] + const playing = kind === "audio" || kind === "video" + const active = shown || playing + const url = useArtifactUrl( + active ? ref : null, + active ? message || undefined : undefined, + ) + + const objectFit = fit === "contain" ? "object-contain" : "object-cover" + const alt = label || ref?.name || message + + return ( +
+ {body()} +
+ ) + + function body() { + // A camera serving its own stream is played from source: the engine + // carries references at whatever rate the flow publishes them, which is + // not video. + if (stream && kind !== "audio") { + return ( + {alt} + ) + } + + if (!message && !stream) { + return

Pick a message.

+ } + if (!ref) { + return

Nothing published yet.

+ } + if (!url) { + return

Loading…

+ } + + if (kind === "audio") { + return ( + + ) + } + + if (kind === "video") { + return ( + + ) + } + + return ( + {alt} + ) + } +} diff --git a/frontend/src/components/Devices/DeviceDetail.tsx b/frontend/src/components/Devices/DeviceDetail.tsx index ba57ff8..02048f3 100644 --- a/frontend/src/components/Devices/DeviceDetail.tsx +++ b/frontend/src/components/Devices/DeviceDetail.tsx @@ -19,6 +19,7 @@ import { FlowsService, } from "@/client" import { initials, resolveIcon } from "@/components/Dashboard/icons" +import { MediaView } from "@/components/Dashboard/MediaWidget" import { dashboardsQueryOptions } from "@/components/Dashboard/queries" import { flowsQueryOptions } from "@/components/Flow/queries" import { InfoTip, PanelSection } from "@/components/Flow/SidePanel" @@ -139,6 +140,7 @@ export function DeviceDetail({ id }: { id: string }) {
+
@@ -216,6 +218,38 @@ function Basics({ ) } +/** + * What the device's camera sees, once a flow wires its `camera` port. + * + * Watching is what turns the camera on: it sends frames only while this + * preview, a tile or a flow reading them wants them. + */ +function Camera({ device }: { device: DevicePublic }) { + const bound = (device.bindings ?? []).find( + (binding) => binding.port === "camera" && binding.direction === "in", + ) + if (!bound) return null + return ( +
+ +
+ +
+ +
+ ) +} + /** Which dashboards the device shows, in the order its rail carries them. */ function Dashboards({ device, diff --git a/frontend/src/components/Flow/FlowEditor.tsx b/frontend/src/components/Flow/FlowEditor.tsx index c2e435a..c5547aa 100644 --- a/frontend/src/components/Flow/FlowEditor.tsx +++ b/frontend/src/components/Flow/FlowEditor.tsx @@ -511,6 +511,19 @@ function FlowEditorInner({ } }, [key]) + // What the flow's nodes read, for a device node to say that a camera read + // here sends all the time. + const readers = useMemo(() => { + const found = new Set() + for (const node of definitions) { + if (node.type === DEVICE_TYPE) continue + for (const spec of node.requires ?? []) { + if (spec.name) found.add(qualify(flowName, spec.name)) + } + } + return found + }, [definitions, flowName]) + /** * Where clicking an endpoint takes you: the thing it stands for. * @@ -1508,6 +1521,7 @@ function FlowEditorInner({ nodeTypes={nodeTypeInfo ?? []} origin={flowDoc.origin} suggestions={suggestions} + readers={readers} expanded={editorExpanded} onToggleExpand={() => setEditorExpanded((wide) => !wide)} onChange={updateNode} diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index 5eb533b..eb847ff 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -56,6 +56,7 @@ import useAuth from "@/hooks/useAuth" import useCustomToast from "@/hooks/useCustomToast" import { inCodeEditor, useShortcuts } from "@/lib/shortcuts" import { cn } from "@/lib/utils" +import { qualify } from "./deriveEdges" import { DtypeValue } from "./FlowBoundary" import { IDENTIFIER, @@ -508,11 +509,14 @@ function DeviceSettings({ node, flow, suggestions, + readers, onChange, }: { node: NodeDef_Input flow: string suggestions: PortSuggestions + /** Qualified messages another node of this flow reads. */ + readers: ReadonlySet onChange: (next: NodeDef_Input) => void }) { const { data } = useQuery(devicesQueryOptions()) @@ -591,6 +595,7 @@ function DeviceSettings({ flow={flow} config={config} reported={reported} + readers={readers} suggestions={suggestions.provides} onChange={ports("provides")} onConfig={setConfig} @@ -619,6 +624,7 @@ function DevicePortList({ flow, config, reported, + readers, suggestions, onChange, onConfig, @@ -632,6 +638,8 @@ function DevicePortList({ config: Record /** What the device says it has, or null when it has not said. */ reported: ReportedPort[] | null + /** Qualified messages another node of this flow reads. */ + readers?: ReadonlySet suggestions: string[] /** `renamed` is a port taking another name, or dropped (`to` of null). */ onChange: (next: MessageSpec[], renamed?: [string, string | null]) => void @@ -809,6 +817,18 @@ function DevicePortList({ The device does not report this port.

) : null} + {said?.on_demand && + spec.name && + readers?.has(qualify(flow, spec.name)) ? ( +

+ A node in this flow reads every frame, so {port} sends all the + time, also while nobody watches. Read snapshot or person + instead. +

+ ) : null} {schema ? (
expanded: boolean onChange: (next: NodeDef_Input) => void onRenameMessage: (previous: string, next: string) => void @@ -1954,6 +1976,7 @@ function PanelBody({ node={node} flow={flow} suggestions={suggestions} + readers={readers} onChange={editNode} /> ) : ( @@ -2133,6 +2156,7 @@ export function NodePanel({ nodeTypes, origin, suggestions, + readers, expanded, onChange, onRenameMessage, @@ -2148,6 +2172,8 @@ export function NodePanel({ /** Set when the flow was declared in code elsewhere; its bodies are generated. */ origin: FlowOrigin | null | undefined suggestions: PortSuggestions + /** Qualified messages the flow's nodes read, a device node's aside. */ + readers: ReadonlySet expanded: boolean onChange: (next: NodeDef_Input) => void onRenameMessage: (previous: string, next: string) => void @@ -2201,6 +2227,7 @@ export function NodePanel({ nodeType={nodeType} origin={origin} suggestions={suggestions} + readers={readers} expanded={expanded} onChange={onChange} onRenameMessage={onRenameMessage} diff --git a/frontend/src/hooks/useOnScreen.ts b/frontend/src/hooks/useOnScreen.ts new file mode 100644 index 0000000..4e81836 --- /dev/null +++ b/frontend/src/hooks/useOnScreen.ts @@ -0,0 +1,47 @@ +import { + type RefObject, + useEffect, + useState, + useSyncExternalStore, +} from "react" + +function subscribeVisibility(listener: () => void) { + document.addEventListener("visibilitychange", listener) + return () => document.removeEventListener("visibilitychange", listener) +} + +/** Whether the tab is the one being looked at. */ +export function usePageVisible(): boolean { + return useSyncExternalStore( + subscribeVisibility, + () => document.visibilityState === "visible", + () => true, + ) +} + +/** + * Whether an element is on screen: inside the viewport, in a tab that is + * being looked at. + * + * Off until the observer first says otherwise, so a tile below the fold never + * asks for anything on its way to being found out of view. + */ +export function useOnScreen(ref: RefObject): boolean { + const page = usePageVisible() + const [inView, setInView] = useState( + () => typeof IntersectionObserver === "undefined", + ) + + useEffect(() => { + const element = ref.current + if (!element || typeof IntersectionObserver === "undefined") return + const observer = new IntersectionObserver((entries) => { + const last = entries[entries.length - 1] + if (last) setInView(last.isIntersecting) + }) + observer.observe(element) + return () => observer.disconnect() + }, [ref]) + + return page && inView +}