From d471614e6a9c18d2a5ba739ccda8ab1656803b61 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 2 Sep 2026 10:15:14 +0200 Subject: [PATCH] Push a frame instead of storing and fetching it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The rate the media dtypes could carry was one frame every second or two: each was a file on the data volume, an event on the socket, and a request back for the bytes. This closes both halves of that, and they are one feature. `save_artifact(..., volatile=True)` writes to a `VolatileStore` — the same content-addressed store, in `/dev/shm`, bounded by size with the oldest falling out (`ARTIFACT_VOLATILE_BYTES`, 48 MB under the container's raised `shm_size`). Nothing sweeps it: a frame nobody kept is not worth walking the store to find. `ArtifactStore.path` falls through to it, which is what lets a volatile frame be an ordinary reference everywhere else — the dtype check, a panel's digest scope, `load_artifact` in a node, and the widget's own fetch all work on one unchanged. `adopt` copies one into the store when a run records it, so "returned media is kept, emitted media is not" stays true. The bytes then go down the flows websocket as a length-prefixed binary frame, sent just ahead of the `message_value` naming them, so a tile has the frame when it hears the value moved. Nothing is pushed unasked: a client names the messages it is drawing (`{"type":"media","names":[…]}`), a panel's list is intersected with the scope it already had, and only the newest frame per name in a batch is sent — a client that fell behind is not handed frames it would draw over. The tunnel relays text only, so a screen reached through a portal falls back to fetching, which is why the rate table now has two rows. Around the edges: the remote worker's fetch cache is bounded at last (`FLUKSIO_ARTIFACT_CACHE_BYTES`), since content addressing means nothing in it ever expires and a media stream fills it with chunks nothing asks for twice; a port carrying an image draws the frame in the node panel rather than only saying `image/png · frame.png · 1.79kB`; and an edge chip says that much instead of a line of hash. The media screenshot stops waiting for `networkidle` — a camera is a socket that never goes quiet, which is the point of it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YC4u66vjzW54fnHu5Juhh9 --- backend/fluksio/__init__.py | 5 +- backend/fluksio/api/routes/artifacts.py | 7 +- backend/fluksio/api/routes/flows.py | 97 ++++++++++- backend/fluksio/core/config.py | 11 ++ backend/fluksio/flow/artifacts.py | 122 +++++++++++++- backend/fluksio/flow/connector.py | 10 +- backend/fluksio/flow/runs.py | 5 + backend/fluksio/main.py | 51 +++++- backend/tests/api/routes/test_panels.py | 80 +++++++++ backend/tests/flow/test_workers.py | 81 ++++++++++ backend/tests/test_artifacts.py | 75 ++++++++- docker/compose.yml | 5 + docs/code/nodes.md | 14 +- docs/code/workers.md | 8 + docs/interface/dashboards.md | 21 +-- docs/reference/configuration.md | 12 ++ docs/reference/connector-contract.md | 16 +- docs/reference/payload-types.md | 24 ++- frontend/scripts/capture-screenshots.mjs | 13 +- .../src/components/Dashboard/MediaWidget.tsx | 79 +-------- .../src/components/Flow/EdgeInspector.tsx | 7 +- frontend/src/components/Flow/LiveEdge.tsx | 4 + .../src/components/Flow/MessageSparkline.tsx | 8 +- frontend/src/components/Flow/ValuePreview.tsx | 53 +++--- frontend/src/components/Flow/liveStore.ts | 83 ++++++++++ frontend/src/components/Flow/useFlowSocket.ts | 25 +++ frontend/src/lib/media.check.ts | 63 ++++++++ frontend/src/lib/media.ts | 152 ++++++++++++++++++ worker/fluksio_worker/worker_main.py | 117 ++++++++++++-- 29 files changed, 1101 insertions(+), 147 deletions(-) create mode 100644 frontend/src/lib/media.check.ts create mode 100644 frontend/src/lib/media.ts diff --git a/backend/fluksio/__init__.py b/backend/fluksio/__init__.py index 7a296da..2709d71 100644 --- a/backend/fluksio/__init__.py +++ b/backend/fluksio/__init__.py @@ -61,7 +61,10 @@ def emit(**ports: Any) -> None: def save_artifact( - source: Any, name: str = "", media_type: str = "application/octet-stream" + source: Any, + name: str = "", + media_type: str = "application/octet-stream", + volatile: bool = False, ) -> dict[str, Any]: """Put bytes in the artifact store and return a reference to them.""" raise RuntimeError(_OUTSIDE.format(name="save_artifact", verb="save to")) diff --git a/backend/fluksio/api/routes/artifacts.py b/backend/fluksio/api/routes/artifacts.py index 471737e..2019547 100644 --- a/backend/fluksio/api/routes/artifacts.py +++ b/backend/fluksio/api/routes/artifacts.py @@ -80,6 +80,7 @@ async def put_artifact( request: Request, name: str = Query(default=""), media_type: str = Query(default=""), + volatile: bool = Query(default=False), ) -> Any: """Store the request body and answer with the reference to it. @@ -87,6 +88,10 @@ async def put_artifact( legitimate a body here as a checkpoint, and neither should have to fit in memory twice. Capped, because nothing else here was: any account, and any worker credential, could otherwise fill the data volume. + + ``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. """ store = _store(request) cap = settings.MAX_ARTIFACT_BYTES @@ -112,7 +117,7 @@ async def put_artifact( # as long as the upload lasts. await run_in_threadpool(handle.write, chunk) return await run_in_threadpool( - store.put_file, Path(handle.name), media_type, name + store.put_file, Path(handle.name), media_type, name, volatile ) finally: Path(handle.name).unlink(missing_ok=True) diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 02d85aa..7c6992f 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -28,6 +28,7 @@ from fluksio.api.deps import ( user_from_token, ) from fluksio.core.db import engine +from fluksio.flow.artifacts import is_reference from fluksio.flow.controller import FlowController from fluksio.flow.dashboards import DashboardStore from fluksio.flow.events import event_bus @@ -931,6 +932,88 @@ def event_for_panel(event: dict[str, Any], only: set[str]) -> bool: # should not be handed the whole queue in one message. 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 + + +def wanted_names( + frame: str, only: set[str] | None, previous: set[str] +) -> set[str] | None: + """What this client is asking to be sent bytes for, or None if it said + something else. + + A client asks by name — the tiles it is drawing — and is answered only for + the names a panel credential would have been allowed anyway. Bytes are the + expensive thing on this socket, so nothing is pushed until something says + it is looking at it. + """ + try: + payload = orjson.loads(frame) + except orjson.JSONDecodeError: + return None + if not isinstance(payload, dict) or payload.get("type") != "media": + return None + names = payload.get("names") + if not isinstance(names, list): + return set() + asked = {str(name) for name in names[:MAX_MEDIA_NAMES]} + return asked if only is None else asked & only + + +def media_frames( + events: list[dict[str, Any]], wanted: set[str], store: Any +) -> list[bytes]: + """The bytes behind the media values in this batch, one frame each. + + Referenced-then-fetched costs a round trip per frame, which is what keeps + the message plane at a glance rather than a view. Pushing the bytes down + 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. + """ + if not wanted or store is None or getattr(store, "volatile", None) is None: + return [] + + newest: dict[str, dict[str, Any]] = {} + for event in events: + if event.get("type") != "message_value": + continue + name = str(event.get("name") or "") + if name not in wanted or not is_reference(event.get("value")): + continue + newest[name] = event + + frames: list[bytes] = [] + 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: + continue + header = orjson.dumps( + { + "type": "media", + "name": name, + "digest": digest, + "media_type": str(value.get("media_type") or ""), + "ts": event.get("ts"), + } + ) + # Length-prefixed, so one frame carries both halves and the reader + # never has to guess where the JSON stops. + frames.append(len(header).to_bytes(4, "big") + header + payload) + return frames + async def _send(websocket: WebSocket, events: list[dict[str, Any]]) -> None: """One event, or a batch of them under `events`. @@ -962,12 +1045,16 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None: await websocket.close(code=1008) return only = panel_scope(token, websocket.app) + # Nothing until a client says it is drawing something: bytes are what this + # socket cannot afford to send speculatively. + wanted: set[str] = set() await websocket.accept() controller: FlowController | None = getattr( websocket.app.state, "flow_controller", None ) + store = getattr(websocket.app.state, "artifact_store", None) async def send_snapshot() -> None: if controller is not None: @@ -989,7 +1076,10 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None: # stream — a keepalive, or anything else it decides to # say, used to be read as the client going away and cost # it every live update from then on. - receiver.exception() + if receiver.exception() is None: + asked = wanted_names(receiver.result(), only, wanted) + if asked is not None: + wanted = asked receiver = asyncio.create_task(websocket.receive_text()) if sender not in done: continue @@ -1017,6 +1107,7 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None: # token — and that would widen this socket to # everything on the bus. only = panel_scope(token, websocket.app) or set() + wanted &= only if out: await _send(websocket, out) out = [] @@ -1025,6 +1116,10 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None: continue out.append(event) if out: + # The bytes first: a tile that has the frame when the + # value lands draws it in one pass rather than two. + for frame in media_frames(out, wanted, store): + await websocket.send_bytes(frame) await _send(websocket, out) except (WebSocketDisconnect, RuntimeError): # A peer that goes away mid-send takes the RuntimeError route diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 1a2710b..c1f344e 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -150,6 +150,17 @@ 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`. + 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. + ARTIFACT_VOLATILE_BYTES: int = 48 * 1024 * 1024 # Without a Redis host the engine keeps its state in memory. REDIS_HOST: str | None = None REDIS_PORT: int = 6379 diff --git a/backend/fluksio/flow/artifacts.py b/backend/fluksio/flow/artifacts.py index 616e1ec..a1d5ad5 100644 --- a/backend/fluksio/flow/artifacts.py +++ b/backend/fluksio/flow/artifacts.py @@ -52,6 +52,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. + self.volatile: VolatileStore | None = None def _path(self, digest: str) -> Path: body = digest[len(DIGEST_PREFIX) :] @@ -60,14 +64,25 @@ class ArtifactStore: return self.root / body[:2] / body def put( - self, chunks: Iterable[bytes], name: str = "", media_type: str = "" + self, + chunks: Iterable[bytes], + name: str = "", + media_type: str = "", + volatile: bool = False, ) -> dict[str, Any]: """Store a stream and return the reference to it. Written to a temporary file first and moved into place once the digest is known, so a half-written artifact never has a name anyone can find. A file already there is left alone: identical content is identical. + + ``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. """ + if volatile and self.volatile is not None: + return self.volatile.put(chunks, name=name, media_type=media_type) digester = hashlib.sha256() size = 0 handle = tempfile.NamedTemporaryFile(dir=self.root, delete=False) @@ -98,21 +113,52 @@ class ArtifactStore: } def put_file( - self, path: Path, media_type: str = "", name: str = "" + self, + path: Path, + media_type: str = "", + name: str = "", + volatile: bool = False, ) -> dict[str, Any]: with path.open("rb") as handle: return self.put( iter(lambda: handle.read(CHUNK), b""), name=name or path.name, media_type=media_type, + volatile=volatile, ) def path(self, digest: str) -> Path | None: - """Where the bytes are, or None if this store does not have them.""" + """Where the 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. + """ if not valid_digest(digest): return None target = self._path(digest) - return target if target.exists() else None + if target.exists(): + return target + if self.volatile is not None: + return self.volatile.path(digest) + return None + + def adopt(self, digest: str) -> bool: + """Copy a volatile artifact into this store, so it outlives the ring. + + What makes "emitted media is not kept, returned media is" true: a run + recording a reference calls this, and the frame stops being one the + next few seconds can evict. + """ + if self.volatile is None or not valid_digest(digest): + return False + if self._path(digest).exists(): + return True + source = self.volatile.path(digest) + if source is None: + return False + self.put_file(source) + return True def read(self, digest: str) -> Iterator[bytes]: target = self.path(digest) @@ -150,3 +196,71 @@ class ArtifactStore: except OSError: logger.warning("Could not remove artifact %s", entry.name) return removed + + +class VolatileStore(ArtifactStore): + """A bounded ring of artifacts, held wherever memory is cheaper than disk. + + 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. + + 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. + """ + + def __init__(self, root: Path, limit_bytes: int) -> None: + super().__init__(root) + self.limit_bytes = limit_bytes + + 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 + + def trim(self) -> int: + """Drop the oldest until the ring is inside its bound. + + 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. + """ + if self.limit_bytes <= 0: + return 0 + entries: list[tuple[float, int, Path]] = [] + total = 0 + for entry in self.root.glob("*/*"): + try: + stat = entry.stat() + except OSError: + continue + if not entry.is_file(): + continue + entries.append((stat.st_mtime, stat.st_size, entry)) + total += stat.st_size + if total <= self.limit_bytes: + return 0 + + removed = 0 + for _mtime, size, entry in sorted(entries): + if total <= self.limit_bytes: + break + try: + entry.unlink() + except OSError: + continue + total -= size + removed += 1 + return removed diff --git a/backend/fluksio/flow/connector.py b/backend/fluksio/flow/connector.py index 1cfe897..de7a378 100644 --- a/backend/fluksio/flow/connector.py +++ b/backend/fluksio/flow/connector.py @@ -141,6 +141,7 @@ class ConnectorNode(Node): data: bytes, name: str = "", media_type: str = "application/octet-stream", + volatile: bool = False, ) -> dict[str, Any]: """Store bytes and return the reference to publish on a media port. @@ -149,6 +150,11 @@ class ConnectorNode(Node): store and the reference names them, which is what an ``image``, ``audio`` or ``video`` port carries. + ``volatile`` is what a camera publishes with: the bytes go to a ring + held in memory rather than to the data volume, are pushed to whatever + screen is watching, and last seconds. Use it for a frame; leave it off + for a recording somebody asked to keep. + Only available once the node has started — the store belongs to the engine, and is handed over then. """ @@ -156,7 +162,9 @@ class ConnectorNode(Node): raise RuntimeError( "no artifact store: a connector can only save bytes once it has started" ) - return self._artifacts.put([data], name=name, media_type=media_type) + return self._artifacts.put( + [data], name=name, media_type=media_type, volatile=volatile + ) async def start(self, app: FastAPI | None = None) -> None: self._artifacts = getattr(app.state, "artifact_store", None) if app else None diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 9aa6e36..867bc83 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -1313,6 +1313,11 @@ class RunService: with Session(db_engine) as session: session.merge(row) for message, ref in outcome.artifacts.items(): + # A run recording a reference is what "returned media is + # kept" means: copied out of the volatile ring, or the row + # would outlive the bytes it names by a few seconds. + if self._artifacts is not None: + self._artifacts.adopt(str(ref.get("digest") or "")) session.merge( RunArtifact( run_id=run_id, diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 4d408a6..c6c2020 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -1,16 +1,19 @@ import asyncio import contextlib +import hashlib import inspect import logging +import tempfile from collections.abc import AsyncIterator, Callable, Coroutine from contextlib import AbstractAsyncContextManager, asynccontextmanager +from pathlib import Path from typing import Any from fastapi import FastAPI, Request from fastapi.concurrency import run_in_threadpool from fastapi.responses import JSONResponse from fastapi.routing import APIRoute -from fluksio_worker.worker_main import ARTIFACT_DIR_ENV +from fluksio_worker.worker_main import ARTIFACT_DIR_ENV, ARTIFACT_VOLATILE_DIR_ENV from starlette.middleware.cors import CORSMiddleware from fluksio import __version__ @@ -23,7 +26,7 @@ from fluksio.core.db import engine as db_engine from fluksio.core.db import prepare from fluksio.flow import logs, modules from fluksio.flow.alerts import AlertManager -from fluksio.flow.artifacts import ArtifactStore +from fluksio.flow.artifacts import ArtifactStore, VolatileStore from fluksio.flow.controller import FlowController, RebuildBusy from fluksio.flow.dashboards import DashboardStore from fluksio.flow.events import event_bus @@ -113,6 +116,43 @@ async def _sweep_artifacts(store: ArtifactStore, controller: FlowController) -> logger.exception("Artifact sweep failed") +#: How often the ring is measured. Seconds rather than the sweep's hour: it is +#: bounded by size and the frames arriving are what push the old ones out. +VOLATILE_TRIM_S = 5.0 + + +async def _trim_volatile(ring: VolatileStore) -> None: + """Hold the volatile ring 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. + """ + if ring.limit_bytes <= 0: + return + while True: + await asyncio.sleep(VOLATILE_TRIM_S) + try: + await run_in_threadpool(ring.trim) + except Exception: + logger.exception("Could not trim the volatile artifact ring") + + +def _volatile_root() -> Path: + """Where the ring goes: memory if this machine has some to lend. + + Named for the data directory rather than fixed, so two engines on one host + have a ring each instead of quietly evicting each other's frames. + """ + configured = settings.ARTIFACT_VOLATILE_DIR + if configured is not None: + return configured + tag = hashlib.sha256(str(settings.DATA_DIR.resolve()).encode()).hexdigest()[:8] + shm = Path("/dev/shm") + parent = shm if shm.is_dir() else Path(tempfile.gettempdir()) + return parent / f"fluksio-volatile-{tag}" + + def _mcp_sessions() -> AbstractAsyncContextManager[None]: """The MCP session manager's run scope, or nothing when MCP is off.""" if not settings.MCP_ENABLED: @@ -205,6 +245,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Beside the flows rather than in them: an artifact is what a run produced, # not something anyone wrote, so it has no business in the git repository. artifacts = ArtifactStore(settings.FLOWS_DIR.parent / "artifacts") + # 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) + artifacts.volatile = volatile app.state.artifact_store = artifacts accountant = ResourceAccountant( cpus=settings.FLOW_CPUS, gpus=settings.FLOW_GPUS @@ -223,6 +268,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), # 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 @@ -292,6 +338,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: _background(alerts.run(), "alert-manager") _background(MetricsCollector(event_bus).run(), "metrics-collector") _background(_sweep_artifacts(artifacts, controller), "artifact-gc") + _background(_trim_volatile(volatile), "artifact-ring") await controller.start() started.append(controller.stop) run_service.start() diff --git a/backend/tests/api/routes/test_panels.py b/backend/tests/api/routes/test_panels.py index 331f534..0efdda0 100644 --- a/backend/tests/api/routes/test_panels.py +++ b/backend/tests/api/routes/test_panels.py @@ -838,3 +838,83 @@ def test_a_panel_fetches_only_the_media_its_tiles_are_showing( ).status_code == 403 ) + + +def test_a_socket_pushes_only_the_frames_it_was_asked_for(tmp_path) -> None: + """Bytes are the one thing this socket cannot send speculatively. + + A frame a second per subscriber is affordable; every frame to every open + editor tab is not. So nothing goes until a client names what it is drawing, + and a panel credential can only name what it was already allowed to see. + """ + import orjson + + from fluksio.api.routes.flows import media_frames, wanted_names + from fluksio.flow.artifacts import ArtifactStore, VolatileStore + + store = ArtifactStore(tmp_path / "artifacts") + store.volatile = VolatileStore(tmp_path / "ring", limit_bytes=1_000_000) + frame = store.put([b"\x89PNG..."], name="f.png", media_type="image/png", + volatile=True) + kept = store.put([b"checkpoint"], name="w.pt") + + asking = orjson.dumps({"type": "media", "names": ["cam.frame", "other.frame"]}) + # A person's socket gets what it asked for; a panel's is intersected with + # what it draws, so naming a message is not a way around the scope. + assert wanted_names(asking.decode(), None, set()) == {"cam.frame", "other.frame"} + assert wanted_names(asking.decode(), {"cam.frame"}, set()) == {"cam.frame"} + # Anything else on this socket leaves the set alone. + assert wanted_names('{"type":"ping"}', None, set()) is None + assert wanted_names("not json", None, set()) is None + + events = [ + {"type": "message_value", "name": "cam.frame", "value": frame, "ts": 1.0}, + {"type": "message_value", "name": "other.frame", "value": frame, "ts": 1.0}, + {"type": "message_value", "name": "cam.model", "value": kept, "ts": 1.0}, + {"type": "node_log", "name": "cam.frame", "value": frame}, + ] + frames = media_frames(events, {"cam.frame", "cam.model"}, store) + + # One frame: the ring's, for the name that asked. The durable checkpoint is + # what a fetch is for, and a log is not a value. + assert len(frames) == 1 + length = int.from_bytes(frames[0][:4], "big") + header = orjson.loads(frames[0][4 : 4 + length]) + assert header == { + "type": "media", + "name": "cam.frame", + "digest": frame["digest"], + "media_type": "image/png", + "ts": 1.0, + } + assert frames[0][4 + length :] == b"\x89PNG..." + + # Nothing asked for, nothing sent. + assert media_frames(events, set(), store) == [] + + +def test_only_the_newest_frame_of_a_batch_is_pushed(tmp_path) -> None: + """A client that fell behind is not handed frames it would only draw over.""" + import orjson + + from fluksio.api.routes.flows import media_frames + from fluksio.flow.artifacts import ArtifactStore, VolatileStore + + store = ArtifactStore(tmp_path / "artifacts") + store.volatile = VolatileStore(tmp_path / "ring", limit_bytes=1_000_000) + first = store.put([b"one"], media_type="image/png", volatile=True) + second = store.put([b"two"], media_type="image/png", volatile=True) + + frames = media_frames( + [ + {"type": "message_value", "name": "cam.frame", "value": first, "ts": 1.0}, + {"type": "message_value", "name": "cam.frame", "value": second, "ts": 2.0}, + ], + {"cam.frame"}, + store, + ) + + assert len(frames) == 1 + length = int.from_bytes(frames[0][:4], "big") + assert orjson.loads(frames[0][4 : 4 + length])["digest"] == second["digest"] + assert frames[0][4 + length :] == b"two" diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index 335c708..1e9798b 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -730,3 +730,84 @@ def test_cancelling_reaches_a_node_running_in_a_child(pool): assert pool.cancel("demo.slow"), "the parent must find a child's node" thread.join(timeout=10) assert failed + + +def test_a_node_saves_a_frame_to_the_ring(tmp_path): + """A camera writes to memory, and the next node reads it from there. + + The point of the ring being an ordinary store: node code says one word + more, and everything downstream — the dtype, ``load_artifact``, the widget + — carries on not knowing there are two of them. + """ + from fluksio_worker.worker_main import ARTIFACT_VOLATILE_DIR_ENV + + from fluksio.flow.artifacts import VolatileStore + + store = ArtifactStore(tmp_path / "artifacts") + store.volatile = VolatileStore(tmp_path / "ring", limit_bytes=1_000_000) + pool = PythonWorkerPool( + python=sys.executable, + size=1, + env={ + ARTIFACT_DIR_ENV: str(store.root), + ARTIFACT_VOLATILE_DIR_ENV: str(store.volatile.root), + }, + ) + pool.start() + try: + ref = pool.run( + "demo", + "frame", + "import fluksio\n" + "def process():\n" + " return {'frame': fluksio.save_artifact(\n" + " b'\\x89PNG' + b'p' * 64, 'f.png',\n" + " media_type='image/png', volatile=True)}\n", + {}, + "demo.frame", + timeout=10, + )["frame"] + + 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 not (store.root / ref["digest"][7:9]).exists() + + read = pool.run( + "demo", + "read", + "import fluksio\n" + "def process(frame):\n" + " with open(fluksio.load_artifact(frame), 'rb') as f:\n" + " return {'size': len(f.read())}\n", + {"frame": ref}, + "demo.read", + timeout=10, + ) + assert read == {"size": 68} + finally: + pool.stop() + + +def test_the_fetch_cache_is_bounded(tmp_path): + """A worker downloading a media stream fills its cache with chunks nothing + will ask for twice, and content addressing means nothing ever expires.""" + import os + + from fluksio_worker.worker_main import ARTIFACT_CACHE_BYTES_ENV, _trim_cache + + cache = tmp_path / "cache" + cache.mkdir() + for index in range(5): + entry = cache / f"chunk{index}" + entry.write_bytes(b"c" * 1000) + os.utime(entry, (index, index)) + + os.environ[ARTIFACT_CACHE_BYTES_ENV] = "3000" + try: + _trim_cache(str(cache)) + finally: + del os.environ[ARTIFACT_CACHE_BYTES_ENV] + + left = sorted(entry.name for entry in cache.iterdir()) + assert left == ["chunk2", "chunk3", "chunk4"] diff --git a/backend/tests/test_artifacts.py b/backend/tests/test_artifacts.py index 2b8a3e4..3364899 100644 --- a/backend/tests/test_artifacts.py +++ b/backend/tests/test_artifacts.py @@ -3,10 +3,13 @@ import time from datetime import UTC, datetime import pytest +from fastapi.testclient import TestClient from sqlmodel import Session, col, select +from fluksio.core.config import settings + from fluksio.core.db import engine as db_engine -from fluksio.flow.artifacts import ArtifactStore +from fluksio.flow.artifacts import ArtifactStore, VolatileStore from fluksio.flow.runs import new_run_id, sweep_artifacts from fluksio.flow.state import MemoryState from fluksio.models import Run, RunArtifact @@ -108,3 +111,73 @@ def test_a_streamed_chunk_falls_out_once_the_message_moves_on(store): assert sweep_artifacts(store, state, grace_s=5) == 1 assert store.path(first["digest"]) is None assert store.path(second["digest"]) is not None + + +def test_a_ring_drops_the_oldest_once_it_is_full(tmp_path): + """What makes a camera affordable: the room is fixed, not the history.""" + 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] + assert len(held) == 3 + # The newest three, in order: eviction is by age, not by chance. + assert held == refs[2:] + + +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. + """ + 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 + # In the ring, and not on the volume the sweep is about. + assert not (store.root / ref["digest"][7:9]).exists() + + +def test_a_recorded_frame_is_copied_out_of_the_ring(tmp_path): + """Emitted media is not kept; returned media is.""" + store = ArtifactStore(tmp_path / "artifacts") + store.volatile = VolatileStore(tmp_path / "ring", limit_bytes=1_000_000) + ref = store.put([b"kept"], name="f.png", volatile=True) + + assert store.adopt(ref["digest"]) is True + + store.volatile.collect(set()) + assert store.path(ref["digest"]) is not None + assert store.adopt("sha256:" + "0" * 64) is False + + +def test_an_upload_may_ask_for_the_ring( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A camera on another host publishes over HTTP and wants the ring too.""" + store = client.app.state.artifact_store + previous = store.volatile + store.volatile = VolatileStore( + store.root.parent / "test-ring", limit_bytes=1_000_000 + ) + try: + url = f"{settings.API_V1_STR}/artifacts" + answer = client.put( + f"{url}?name=f.png&media_type=image%2Fpng&volatile=1", + headers=superuser_token_headers, + content=b"\x89PNG-frame", + ) + assert answer.status_code == 200, answer.text + digest = answer.json()["digest"] + + assert store.volatile.path(digest) is not None + 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. + served = client.get(f"{url}/{digest}", headers=superuser_token_headers) + assert served.status_code == 200 + assert served.content == b"\x89PNG-frame" + finally: + store.volatile = previous diff --git a/docker/compose.yml b/docker/compose.yml index d8d359a..160e594 100644 --- a/docker/compose.yml +++ b/docker/compose.yml @@ -141,6 +141,11 @@ services: volumes: - app-flow-data:/data + # `/dev/shm`, where frames a flow only shows live are held. Docker's + # default is 64 MB, which the ring's own bound sits just under; raise both + # together (ARTIFACT_VOLATILE_BYTES) for more cameras or bigger frames. + shm_size: 128mb + # Deep health: fails when the event loop is wedged or Redis is gone, not # just when the process is dead. Autoheal restarts on unhealthy. healthcheck: diff --git a/docs/code/nodes.md b/docs/code/nodes.md index f399297..3ef2a18 100644 --- a/docs/code/nodes.md +++ b/docs/code/nodes.md @@ -137,7 +137,8 @@ def process(speech): # an `audio` port def process(camera_url): for index, jpeg in enumerate(grab(camera_url)): # a generator frame = fluksio.save_artifact( - jpeg, f"frame-{index:05d}.jpg", media_type="image/jpeg" + jpeg, f"frame-{index:05d}.jpg", + media_type="image/jpeg", volatile=True, ) frame["meta"] = {"seq": index} yield {"frame": frame} # an `image` stream port @@ -148,12 +149,21 @@ has to match, so a node declaring `audio` never receives a video by accident. See [Payload types](../reference/payload-types.md#image-audio-video) for what each carries and what rates are realistic. +`volatile=True` is for a frame rather than a result. The bytes go to a ring in +memory instead of the data volume, and the engine pushes them down the +websocket to whichever screens are drawing that message — so a camera runs at +ten frames a second without writing anything to disk. They last as long as it +takes newer frames to need the room. Leave it off for a clip somebody asked to +keep. + !!! warning "Emitted media is not kept; returned media is" Only what a node *returns* is recorded against its run. Frames yielded along the way are replaced in state by the next one, and the artifact sweep removes bytes nothing refers to any more, which is what stops a camera - filling the disk. If a particular frame matters, return it. + filling the disk. If a particular frame matters, return it: a volatile one + is copied out of the ring when the run records it, so returning it is also + what makes it outlive the next few seconds. ## Printing diff --git a/docs/code/workers.md b/docs/code/workers.md index 24b6bae..2655f37 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -141,6 +141,14 @@ filesystem writes to it directly; one that does not fetches and uploads over HTTP, using the artifact endpoint beside the socket it already has. Either way your node code is the same two calls. +A fetch is cached on the worker by digest, since content addressing means an +entry is never stale. Nothing expires on its own, so the cache is bounded by +size and the oldest fall out: `FLUKSIO_ARTIFACT_CACHE` says where it lives +(default a directory in the temporary directory) and +`FLUKSIO_ARTIFACT_CACHE_BYTES` how much it holds (default 1 GiB). Worth raising +where a worker reads the same large inputs repeatedly, and worth leaving alone +where it reads a media stream — those are chunks nothing asks for twice. + ## Seeing what is attached ```sh diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index daaab5f..3c3a2fd 100644 --- a/docs/interface/dashboards.md +++ b/docs/interface/dashboards.md @@ -105,21 +105,24 @@ so an answer to a different question is ignored. A media widget draws what its message points at: a picture, a clip with controls, a video. Media does not travel as a message; a reference to it does. -The tile fetches the bytes behind whichever reference the message holds, -and redraws when a new one arrives. +The tile takes the bytes behind whichever reference the message holds, and +redraws when a new one arrives. **Crop or fit** decides how a picture fills the tile. **Play as it arrives** starts a clip by itself, though a browser only plays sound once somebody has touched the page, so a screen nobody has tapped stays silent. -Rate is the thing to get right. A frame every second or two is a glance at a -door, and works; through the portal, make that every few seconds. Live video is -not something to push through the message plane at all. Put the camera's own -address in **Live stream** and the browser plays it from source, leaving the -messages to carry the occasional still that a flow can actually react to. +Rate is the thing to get right, and it is decided by the node publishing rather +than by the tile. A frame the engine is holding in memory is pushed down the +same websocket that carries the value, so ten a second is a real view on the +local network; a stored one is fetched, a round trip each, which suits a glance +at a door every second or two. Through the portal everything is fetched, so +make it every few seconds there. Above that, put the camera's own address in +**Live stream** and the browser plays it from source, leaving the messages to +carry the occasional still that a flow can actually react to. -Panels see media the same way, and only their own: a screen may fetch the bytes -its own tiles are showing and nothing else. +Panels see media the same way, and only their own: a screen is sent, and may +fetch, the bytes its own tiles are showing and nothing else. ## Player tiles diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 50f41d8..b40d595 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -131,6 +131,9 @@ warning into a refusal to start. | `OBS_RETENTION_DAYS` | `30` | how long metrics, events and run records are kept | | `ARTIFACT_GC_INTERVAL_S` | `3600` | how often artifact bytes nothing refers to are swept away; 0 never sweeps | | `ARTIFACT_GC_GRACE_S` | `3600` | how long a freshly written artifact is spared, whatever refers to it | +| `MAX_ARTIFACT_BYTES` | `2147483648` | the largest body `PUT /artifacts` will take; 0 removes the limit | +| `ARTIFACT_VOLATILE_DIR` | worked out | where frames a flow only shows live are held; empty picks a directory under `/dev/shm` named for the data directory, and falls back to the temporary directory | +| `ARTIFACT_VOLATILE_BYTES` | `50331648` | how much that ring holds before the oldest frames fall out; 0 turns it off and volatile saves land in the store | The three concurrency limits are also flags on `fluksio serve` (`--max-workers`, `--max-cascades`, `--max-runs`), as is the card count, @@ -146,6 +149,15 @@ sweep is what keeps a flow streaming media from filling the disk. It stands aside entirely while a run is in flight, since a node may store a checkpoint long before it returns the reference to it. +Frames saved with `volatile=True` skip all of that. They go to a ring in memory +instead of the volume, the oldest falling out once the newest need the room, +and the engine pushes them down the websocket to whichever screens are drawing +them — which is what a camera at ten frames a second needs and the store cannot +give it. A frame a run *records* is copied into the store on the way, so +returned media is kept and emitted media is not. Under Docker the ring lives in +the container's `/dev/shm`, whose default is 64 MB: raise `shm_size` alongside +`ARTIFACT_VOLATILE_BYTES`. + A node that declares nothing is not accounted against `FLOW_CPUS`; it runs on the shared pool and is given `FLOW_CPUS / FLOW_MAX_WORKERS` as a thread cap, so several at once cannot each size themselves to the whole machine. Setting diff --git a/docs/reference/connector-contract.md b/docs/reference/connector-contract.md index 0033505..9f897b8 100644 --- a/docs/reference/connector-contract.md +++ b/docs/reference/connector-contract.md @@ -177,7 +177,9 @@ connector publishes a reference to it instead: async def poll(self) -> dict[str, Any] | None: jpeg = await asyncio.to_thread(self._grab) return { - "frame": self.save_artifact(jpeg, "frame.jpg", media_type="image/jpeg") + "frame": self.save_artifact( + jpeg, "frame.jpg", media_type="image/jpeg", volatile=True + ) } ``` @@ -186,10 +188,18 @@ async def poll(self) -> dict[str, Any] | None: works once the node has started, since the store is the engine's and is handed over then. +`volatile=True` is what a camera publishes with. The frame goes to a ring in +memory rather than the data volume and is pushed down the websocket to +whichever screens are drawing it, so a wall panel sees ten frames a second and +the SD card under it is never written to. Frames last until newer ones need the +room; leave the flag off for a reading somebody asked to keep. + Each reading is a new artifact, which the poll loop publishes because its digest differs from the last. Set `poll_interval` to what somebody actually -wants to look at: a frame every second or two is a glance, and live video -belongs on the camera's own stream rather than in the graph. +wants to look at. Through a portal the bytes are fetched rather than pushed, so +a remote panel wants a frame every second or two; higher rates than that are +for the local network, and full-rate video still belongs on the camera's own +stream. ## Lifecycle diff --git a/docs/reference/payload-types.md b/docs/reference/payload-types.md index 82f9a4d..12feca7 100644 --- a/docs/reference/payload-types.md +++ b/docs/reference/payload-types.md @@ -117,19 +117,27 @@ sequence numbers are for whoever consumes the media. Bytes still never travel as a message. A camera publishes one reference per frame and a microphone one per chunk, which makes a media stream an ordinary -[streaming port](../concepts/flows.md#streaming-ports), and each frame an -artifact. What that costs is worth knowing before pointing a camera at it: +[streaming port](../concepts/flows.md#streaming-ports). What that costs +depends on how the bytes reach the screen: | Rate | Where it works | |---|---| | A clip a second (speech) | anywhere, including through the portal | -| A frame every second or two (a glance at a door) | locally; through the portal, every few seconds | -| Live video, 15–30 fps | not here — see below | +| Ten frames a second (a camera worth watching) | on the local network, with `volatile=True` | +| A frame every second or two (a glance at a door) | anywhere, including through the portal | +| Full-rate video, 30 fps and up | not here — see below | -Real-time video is not a message-plane problem: every frame would be an -artifact, an event and a fetch. Point a media widget's **stream URL** at -whatever the camera already serves and the browser plays it from source; the -messages then carry the occasional still, and the flow reacts to those. +The difference is one flag. A frame saved with +[`volatile=True`](../code/nodes.md#media) is held in memory and pushed down the +websocket in front of the value naming it, so a screen draws it without asking +for anything. A stored one is fetched instead: a round trip per frame, which is +a glance rather than a view. Through a portal every frame is fetched, so a +remote panel is in the second row whatever the flag says. + +Above that, video is not a message-plane problem — every frame would still be +an event. Point a media widget's **stream URL** at whatever the camera already +serves and the browser plays it from source; the messages then carry the +occasional still, and the flow reacts to those. ### `json` diff --git a/frontend/scripts/capture-screenshots.mjs b/frontend/scripts/capture-screenshots.mjs index d8d55de..84735a4 100644 --- a/frontend/scripts/capture-screenshots.mjs +++ b/frontend/scripts/capture-screenshots.mjs @@ -199,13 +199,18 @@ async function captureDashboards(page, dir) { * * Skipped unless the media example is seeded (root `make seed-example-media`), * since it is the one shot that needs a source of frames. The bytes arrive as - * a blob — the tile fetches them with the session's credential, which no `img` - * could carry on its own — so a `blob:` source is the proof the whole path ran - * rather than that a picture is merely present. + * a blob — pushed down the socket, or fetched with the session's credential, + * which no `img` could carry on its own — so a `blob:` source is the proof the + * whole path ran rather than that a picture is merely present. + * + * The one page that cannot wait for `networkidle`: a camera publishing several + * frames a second is a socket that never goes quiet, which is the point of it. + * The blob source below is a stronger wait anyway — it says a frame arrived, + * where idleness only ever said the page stopped asking. */ async function captureMedia(page, dir) { const answer = await page.goto(`${APP_URL}/view/camera`, { - waitUntil: "networkidle", + waitUntil: "domcontentloaded", }) if (!answer?.ok()) return diff --git a/frontend/src/components/Dashboard/MediaWidget.tsx b/frontend/src/components/Dashboard/MediaWidget.tsx index 97976c6..7fd34c3 100644 --- a/frontend/src/components/Dashboard/MediaWidget.tsx +++ b/frontend/src/components/Dashboard/MediaWidget.tsx @@ -1,82 +1,17 @@ -import { useEffect, useState } from "react" - -import { OpenAPI } from "@/client" -import { apiToken } from "@/lib/portal" +import { isRef, useArtifactUrl } from "@/lib/media" import { cn } from "@/lib/utils" import { useBoundValue } from "./dataContext" import { config, text } from "./ui/core/config" import type { WidgetProps } from "./widgets" -/** An artifact reference, as a message carries one. */ -type MediaRef = { - digest?: string - media_type?: string - name?: string - size?: number -} - -const isRef = (value: unknown): value is MediaRef => - typeof value === "object" && - value !== null && - typeof (value as MediaRef).digest === "string" && - (value as MediaRef).digest!.startsWith("sha256:") - -/** - * A local URL for an artifact's bytes, refreshed whenever the digest changes. - * - * Not the endpoint itself: `/artifacts/{digest}` takes a bearer token, and no - * `` or `