Files
app/frontend/src/lib/media.ts
T
stroblme 40864d5ca5 Push a frame instead of storing and fetching it
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.
2026-09-02 10:15:14 +02:00

153 lines
4.9 KiB
TypeScript

/**
* Media references, and the bytes behind them.
*
* A frame never travels as a message — the reference does. What the bytes cost
* to get is what limits the rate: fetched, it is a round trip per frame and a
* camera is a glance; pushed down the socket that already carries the value,
* it is a view. Both paths are here, and a caller does not choose between
* them: the hook takes whichever arrived.
*/
import { useEffect, useState } from "react"
import { OpenAPI } from "@/client"
import { liveStore, useLiveBytes } from "@/components/Flow/liveStore"
import { apiToken } from "@/lib/portal"
import { si } from "@/lib/utils"
/** An artifact reference, as a message carries one. */
export type MediaRef = {
digest?: string
media_type?: string
name?: string
size?: number
}
export const isRef = (value: unknown): value is MediaRef =>
typeof value === "object" &&
value !== null &&
typeof (value as MediaRef).digest === "string" &&
(value as MediaRef).digest!.startsWith("sha256:")
/**
* What a reference *is*, in the space a value would have taken: what kind of
* bytes, what they were called, how many of them.
*/
export function describeArtifact(record: Record<string, unknown>): string {
const name = typeof record.name === "string" && record.name ? record.name : ""
const size = typeof record.size === "number" ? `${si(record.size)}B` : ""
// The media type when there is one: what kind of bytes these are is the
// first thing worth knowing about a frame or a clip, and it is what the
// port's own type was declared against.
const media =
typeof record.media_type === "string" &&
record.media_type &&
record.media_type !== "application/octet-stream"
? record.media_type
: "artifact"
return [media, name, size].filter(Boolean).join(" · ")
}
/** The header a pushed frame carries in front of its bytes. */
export type MediaFrame = {
name: string
digest: string
media_type: string
ts?: number
}
/**
* Split a binary socket frame into what it says and what it carries.
*
* Four bytes of header length, the header as JSON, then the bytes. Returns
* null for anything that does not parse, since a frame the client cannot read
* is one to drop rather than one to crash on.
*/
export function parseMediaFrame(
buffer: ArrayBuffer,
): { header: MediaFrame; bytes: Blob } | null {
if (buffer.byteLength < 4) return null
const length = new DataView(buffer).getUint32(0)
if (length <= 0 || 4 + length > buffer.byteLength) return null
try {
const header = JSON.parse(
new TextDecoder().decode(new Uint8Array(buffer, 4, length)),
) as MediaFrame
if (
typeof header?.digest !== "string" ||
typeof header?.name !== "string"
) {
return null
}
return {
header,
bytes: new Blob([new Uint8Array(buffer, 4 + length)], {
type: header.media_type || "application/octet-stream",
}),
}
} catch {
return null
}
}
/**
* A local URL for an artifact's bytes, refreshed whenever the digest changes.
*
* Pushed bytes are used where the socket sent them, which is the fast path and
* costs no request at all. Otherwise they are fetched: `/artifacts/{digest}`
* takes a bearer token and no `<img>` or `<audio>` can carry a header, so the
* bytes come through fetch and are handed to the element as an object URL.
*
* Passing `name` registers interest in that message, which is what tells the
* engine to push its frames at all — nothing is sent to a screen that is not
* drawing it.
*/
export function useArtifactUrl(ref: MediaRef | null, name?: string): string {
const digest = ref?.digest ?? ""
const mediaType = ref?.media_type ?? ""
const pushed = useLiveBytes(digest)
const [fetched, setFetched] = useState("")
useEffect(() => {
if (!name) return
return liveStore.wantBytes(name)
}, [name])
useEffect(() => {
// Already here: the socket carried the bytes in front of the value.
if (!digest || pushed) {
setFetched("")
return
}
let live = true
let made = ""
const token = apiToken()
const query = mediaType
? `?media_type=${encodeURIComponent(mediaType)}`
: ""
const request = new AbortController()
fetch(`${OpenAPI.BASE}/api/v1/artifacts/${digest}${query}`, {
headers: token ? { Authorization: `Bearer ${token}` } : {},
signal: request.signal,
})
.then((answer) => (answer.ok ? answer.blob() : Promise.reject(answer)))
.then((blob) => {
if (!live) return
made = URL.createObjectURL(blob)
setFetched(made)
})
.catch(() => {
if (live) setFetched("")
})
return () => {
live = false
// Abandoned rather than merely ignored: a tile showing a camera starts a
// request per frame, and the ones it no longer wants should not still be
// arriving.
request.abort()
if (made) URL.revokeObjectURL(made)
}
}, [digest, mediaType, pushed])
return pushed || fetched
}