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.
149 lines
5.0 KiB
TypeScript
149 lines
5.0 KiB
TypeScript
import {
|
|
BaseEdge,
|
|
EdgeLabelRenderer,
|
|
type EdgeProps,
|
|
getBezierPath,
|
|
Position,
|
|
useStore,
|
|
} from "@xyflow/react"
|
|
import { memo, useEffect, useRef, useState } from "react"
|
|
|
|
import { describeArtifact, isRef } from "@/lib/media"
|
|
import { duration } from "@/lib/motion"
|
|
import { cn } from "@/lib/utils"
|
|
import type { FlowEdgeData } from "./deriveEdges"
|
|
import { useLiveValue } from "./liveStore"
|
|
|
|
/** Below this zoom the value chips would be unreadable, so they step aside. */
|
|
const CHIP_MIN_ZOOM = 0.5
|
|
|
|
/** How far a feedback edge stands off the ranks it has to get back past. */
|
|
const BOW = 60
|
|
|
|
/**
|
|
* The path a feedback edge takes.
|
|
*
|
|
* The layout ranks a flow in the direction it runs, so an edge closing a cycle
|
|
* is the one that goes the other way — a widget setting the window a node
|
|
* upstream of it reads. A curve drawn straight between those two handles is
|
|
* drawn straight through everything between them. This one leaves its handle
|
|
* the way every other edge does, swings out into a lane beside the graph, and
|
|
* comes back into its target from the far side, so what it passes it passes
|
|
* around.
|
|
*
|
|
* ponytail: how far out the lane sits is read off how far back the edge has to
|
|
* travel, not off what is actually in the way — this stands a cycle apart from
|
|
* the flow rather than guaranteeing it clears every node. The layout engine
|
|
* routes these properly and React Flow draws its own curves; taking its bend
|
|
* points is the real fix, and a bigger one.
|
|
*/
|
|
function cyclePath(
|
|
sourceX: number,
|
|
sourceY: number,
|
|
targetX: number,
|
|
targetY: number,
|
|
vertical: boolean,
|
|
): [string, number, number] {
|
|
const span = vertical
|
|
? Math.abs(targetY - sourceY)
|
|
: Math.abs(targetX - sourceX)
|
|
const lane = Math.max(BOW, span * 0.3)
|
|
const [c1x, c1y] = vertical
|
|
? [sourceX + lane, sourceY + BOW]
|
|
: [sourceX + BOW, sourceY + lane]
|
|
const [c2x, c2y] = vertical
|
|
? [targetX + lane, targetY - BOW]
|
|
: [targetX - BOW, targetY + lane]
|
|
return [
|
|
`M${sourceX},${sourceY} C${c1x},${c1y} ${c2x},${c2y} ${targetX},${targetY}`,
|
|
// Halfway along a cubic, which is where the chip rides.
|
|
(sourceX + 3 * c1x + 3 * c2x + targetX) / 8,
|
|
(sourceY + 3 * c1y + 3 * c2y + targetY) / 8,
|
|
]
|
|
}
|
|
|
|
function formatValue(value: unknown): string {
|
|
if (typeof value === "number") {
|
|
return Number.isInteger(value) ? String(value) : value.toFixed(2)
|
|
}
|
|
if (typeof value === "string") return value
|
|
// A media reference serialised whole is a line of hash: what belongs in a
|
|
// chip this size is what kind of bytes crossed the edge.
|
|
if (isRef(value)) return describeArtifact(value as Record<string, unknown>)
|
|
return JSON.stringify(value) ?? ""
|
|
}
|
|
|
|
function LiveEdgeComponent({
|
|
id,
|
|
sourceX,
|
|
sourceY,
|
|
targetX,
|
|
targetY,
|
|
sourcePosition,
|
|
targetPosition,
|
|
data,
|
|
selected,
|
|
}: EdgeProps) {
|
|
const { message, producerId, stream } = (data ?? {}) as FlowEdgeData
|
|
const live = useLiveValue(message)
|
|
const zoom = useStore((state) => state.transform[2])
|
|
const [pulsing, setPulsing] = useState(false)
|
|
const lastTs = useRef<number | null>(null)
|
|
|
|
// Every rank sits further along the flow than the one before it, so an edge
|
|
// whose target has ended up behind its source is one closing a cycle.
|
|
const vertical = sourcePosition === Position.Bottom
|
|
const backwards = vertical ? targetY < sourceY : targetX < sourceX
|
|
|
|
const [path, labelX, labelY] = backwards
|
|
? cyclePath(sourceX, sourceY, targetX, targetY, vertical)
|
|
: getBezierPath({
|
|
sourceX,
|
|
sourceY,
|
|
sourcePosition,
|
|
targetX,
|
|
targetY,
|
|
targetPosition,
|
|
})
|
|
|
|
// Restart the stroke animation whenever a newer message lands — but only
|
|
// for the producer that actually published it. A message can have several
|
|
// producers, and can also be set from a dashboard or another flow, so
|
|
// pulsing on the value alone claims things happened that did not.
|
|
useEffect(() => {
|
|
if (!live?.ts || live.ts === lastTs.current) return
|
|
lastTs.current = live.ts
|
|
const from = live.source
|
|
// No source at all is an older engine; pulse rather than go silent.
|
|
if (from && from.id !== producerId) return
|
|
setPulsing(true)
|
|
const timer = setTimeout(() => setPulsing(false), duration.pulse * 1000)
|
|
return () => clearTimeout(timer)
|
|
}, [live?.ts, live?.source, producerId])
|
|
|
|
return (
|
|
<>
|
|
<BaseEdge
|
|
id={id}
|
|
path={path}
|
|
className={cn(pulsing && "edge-live", stream && "edge-stream")}
|
|
style={selected ? { stroke: "var(--primary)" } : undefined}
|
|
/>
|
|
{live !== undefined && zoom >= CHIP_MIN_ZOOM ? (
|
|
<EdgeLabelRenderer>
|
|
<div
|
|
style={{
|
|
transform: `translate(-50%, -50%) translate(${labelX}px, ${labelY}px)`,
|
|
}}
|
|
className="pointer-events-none absolute max-w-[140px] truncate rounded-full border border-border bg-card px-2 py-0.5 font-mono text-xs text-foreground"
|
|
>
|
|
{formatValue(live.value)}
|
|
</div>
|
|
</EdgeLabelRenderer>
|
|
) : null}
|
|
</>
|
|
)
|
|
}
|
|
|
|
export const LiveEdge = memo(LiveEdgeComponent)
|