Melvin Strobl
ecc4bee4bb
- Volatile frames are held in memory with a byte-bounded queue instead of the /dev/shm ring scanned on every put; workers keep writing to the directory, which is trimmed on a timer. A volatile upload no longer spools to disk. ARTIFACT_VOLATILE_BYTES=0 turns the ring off. - A device sends camera frames as binary messages on its socket, held and published in one step and credited to its device node; refused frames are logged and dropped. Its snapshot carries demand and the uplink (binary, or http through the portal). - One demand registry for every socket's wanted media, the tunnel's included, plus what running flows read, mapped onto on-demand ports and published as device_demand; at most 8 media names per socket. - Per-message deflate off in serve and dev; a device may fetch the last 8 digests of a message; published dashboards cached per store revision; a revoked device socket is closed with 1008. - Media tiles want bytes only while on screen; the Devices tab previews a wired camera; the device node warns when a flow reads an on-demand port.
1558 lines
58 KiB
Python
1558 lines
58 KiB
Python
"""The flow API. Everything the editor can do is available here first."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import re
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import orjson
|
|
from fastapi import (
|
|
APIRouter,
|
|
Depends,
|
|
HTTPException,
|
|
Request,
|
|
WebSocket,
|
|
WebSocketDisconnect,
|
|
)
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from jwt.exceptions import InvalidTokenError
|
|
from pydantic import BaseModel
|
|
from sqlmodel import Session, col, select
|
|
|
|
from fluksio.api.deps import (
|
|
CODE_IS_SUPERUSERS,
|
|
CurrentUser,
|
|
FlowControllerDep,
|
|
SessionDep,
|
|
TokenDep,
|
|
decode_token,
|
|
get_code_author,
|
|
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.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
|
|
from fluksio.flow.runs import RunRejected
|
|
from fluksio.flow.schemas import (
|
|
NAME_PATTERN,
|
|
BrainGraph,
|
|
FlowDef,
|
|
FlowsPublic,
|
|
FlowStatePublic,
|
|
FlowSummary,
|
|
HistoryPoint,
|
|
LibraryNode,
|
|
MessageHistory,
|
|
MessageValue,
|
|
NodeDef,
|
|
NodeSource,
|
|
NodeSourceSaved,
|
|
NodeStatusPublic,
|
|
NodeTypeInfo,
|
|
)
|
|
from fluksio.flow.state import as_number
|
|
from fluksio.flow.store import (
|
|
FlowExists,
|
|
FlowNotFound,
|
|
LibExists,
|
|
LibNotFound,
|
|
StaleVersion,
|
|
)
|
|
from fluksio.models import Flavor, Message, Run
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)]
|
|
)
|
|
# The websocket authenticates from its handshake, a header or a ticket, so it
|
|
# stays off the router that enforces the HTTP bearer scheme.
|
|
ws_router = APIRouter(prefix="/flows", tags=["flows"])
|
|
|
|
|
|
class Endpoint(BaseModel):
|
|
"""Something wired into this flow that is not a node in it.
|
|
|
|
A dashboard control setting one of its messages, a tile showing one, or a
|
|
node in another flow on the far side of a dotted name. The canvas draws these so a value never appears to come
|
|
from nowhere — or worse, appears to come from whichever node happens to be
|
|
drawn as a producer.
|
|
"""
|
|
|
|
#: dashboard or flow.
|
|
kind: str
|
|
#: Stable within its kind, and used as the canvas node id.
|
|
id: str
|
|
label: str
|
|
#: What sort of widget, or the node type in the other flow.
|
|
detail: str = ""
|
|
#: Messages of this flow it publishes, and ones it reads.
|
|
provides: list[str] = []
|
|
requires: list[str] = []
|
|
|
|
|
|
class FlowDetail(BaseModel):
|
|
"""A flow plus how it is currently doing.
|
|
|
|
``definition`` is the working copy — the unpublished draft when there is
|
|
one — because that is what the editor shows. ``nodes`` reports the
|
|
published flow, which is what is actually running.
|
|
"""
|
|
|
|
definition: FlowDef
|
|
nodes: list[NodeStatusPublic] = []
|
|
issues: list[ValidationIssue] = []
|
|
has_draft: bool = False
|
|
enabled: bool = True
|
|
paused: bool = False
|
|
#: Dashboards and other flows wired into this one.
|
|
endpoints: list[Endpoint] = []
|
|
|
|
|
|
class ValidationResult(BaseModel):
|
|
issues: list[ValidationIssue] = []
|
|
|
|
|
|
class RenameRequest(BaseModel):
|
|
new_name: str
|
|
|
|
|
|
class PublishRequest(BaseModel):
|
|
version: int
|
|
|
|
|
|
class ShareRequest(BaseModel):
|
|
lib_name: str
|
|
|
|
|
|
class RunRequest(BaseModel):
|
|
inputs: dict[str, Any] = {}
|
|
|
|
|
|
class TriggerRequest(BaseModel):
|
|
values: dict[str, Any] = {}
|
|
|
|
|
|
def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint]:
|
|
"""Everything wired into this flow from outside it."""
|
|
flow = definition.name
|
|
found: dict[str, Endpoint] = {}
|
|
|
|
dashboards: DashboardStore | None = getattr(controller, "dashboards", None)
|
|
if dashboards is not None:
|
|
for binding in dashboards.bindings_for(flow):
|
|
found[f"dashboard:{binding['dashboard']}:{binding['widget']}"] = Endpoint(
|
|
kind="dashboard",
|
|
id=f"dashboard:{binding['dashboard']}:{binding['widget']}",
|
|
label=binding["title"],
|
|
detail=binding["type"],
|
|
provides=[binding["provides"]] if binding["provides"] else [],
|
|
requires=binding["requires"],
|
|
)
|
|
|
|
def _flow_endpoint(key: str, detail: str) -> Endpoint:
|
|
"""One label per node on the far side, however many names reach it."""
|
|
return found.setdefault(
|
|
f"flow:{key}",
|
|
Endpoint(kind="flow", id=f"flow:{key}", label=key, detail=detail),
|
|
)
|
|
|
|
# An outsider reaching into this flow.
|
|
for other, node_id, provides, requires in controller.cross_flow_nodes(flow):
|
|
endpoint = _flow_endpoint(f"{other}.{node_id}", "flow")
|
|
endpoint.provides += provides
|
|
endpoint.requires += requires
|
|
|
|
# And this flow reaching out: its own ports bound to a message of another
|
|
# flow. Read from the working document rather than from the store, so a
|
|
# name just typed is drawn before it has been published — the same reason
|
|
# the canvas draws its own boundary from the document.
|
|
for node in definition.nodes:
|
|
for spec, ours_publishes in [
|
|
*((spec, False) for spec in node.requires),
|
|
*((spec, True) for spec in node.provides),
|
|
]:
|
|
message = qualify(flow, spec.name or "")
|
|
if not message or flow_of(message) == flow:
|
|
continue
|
|
# The node at the other end, so the label reads like a dashboard's:
|
|
# what it is on the first line, what sort of thing it is on the
|
|
# second. A message no published flow declares yet has no other end
|
|
# to name, so it is drawn as the message it is.
|
|
far = controller.message_node(
|
|
message, published=not ours_publishes, exclude=flow
|
|
)
|
|
endpoint = _flow_endpoint(
|
|
f"{far.flow}.{far.node}" if far else message,
|
|
far.type if far else "flow",
|
|
)
|
|
# An endpoint publishing into this flow is what this flow reads.
|
|
side = endpoint.requires if ours_publishes else endpoint.provides
|
|
if message not in side:
|
|
side.append(message)
|
|
|
|
return list(found.values())
|
|
|
|
|
|
def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail:
|
|
name = definition.name
|
|
running = {
|
|
"enabled": controller.is_enabled(name),
|
|
"paused": controller.is_paused(name),
|
|
}
|
|
endpoints = _endpoints(controller, definition)
|
|
if controller.store.has_draft(name):
|
|
# Report the draft the editor is showing, not the version running
|
|
# underneath it — otherwise a node the author just broke looks fine.
|
|
preview = controller.preview(name)
|
|
return FlowDetail(
|
|
definition=definition,
|
|
nodes=preview.nodes,
|
|
issues=preview.issues,
|
|
has_draft=True,
|
|
endpoints=endpoints,
|
|
**running,
|
|
)
|
|
return FlowDetail(
|
|
definition=definition,
|
|
nodes=controller.node_statuses(name),
|
|
issues=controller.flow_issues(name),
|
|
endpoints=endpoints,
|
|
**running,
|
|
)
|
|
|
|
|
|
def _read_flow(controller: FlowController, name: str) -> FlowDef:
|
|
"""The working copy: the draft when there is one, else what is published."""
|
|
try:
|
|
return controller.store.read_flow(name, draft=True)
|
|
except FlowNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
|
|
|
|
|
|
def _audit(action: str, flow: str, user: CurrentUser) -> None:
|
|
"""Record who changed what. The collector writes it down; the bus carries it."""
|
|
event_bus.publish(
|
|
{
|
|
"type": "audit",
|
|
"action": action,
|
|
"flow": flow,
|
|
"user": user.email,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
|
|
|
|
def _source_ref(definition: FlowDef, node_id: str) -> str | None:
|
|
"""The library source this node runs, if it is a shared one."""
|
|
node = next((n for n in definition.nodes if n.id == node_id), None)
|
|
return node.source_ref if node else None
|
|
|
|
|
|
def _require_enabled(controller: FlowController, name: str) -> None:
|
|
if not controller.is_enabled(name):
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"Flow '{name}' is stopped — start it before running it",
|
|
)
|
|
|
|
|
|
def _flow_state(controller: FlowController, name: str) -> FlowStatePublic:
|
|
return FlowStatePublic(
|
|
values={
|
|
key: MessageValue(**value) for key, value in controller.values(name).items()
|
|
},
|
|
nodes=controller.node_statuses(name),
|
|
)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Flows
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/", response_model=FlowsPublic)
|
|
def read_flows(controller: FlowControllerDep) -> Any:
|
|
"""List every flow."""
|
|
summaries = []
|
|
for name in controller.store.list_flows():
|
|
try:
|
|
definition = controller.store.read_flow(name, draft=True)
|
|
except FlowNotFound:
|
|
continue
|
|
statuses = controller.node_statuses(name)
|
|
summaries.append(
|
|
FlowSummary(
|
|
name=definition.name,
|
|
title=definition.title,
|
|
mode=definition.mode,
|
|
node_count=len(definition.nodes),
|
|
error_count=sum(1 for s in statuses if s.status == "error"),
|
|
has_draft=controller.store.has_draft(name),
|
|
enabled=controller.is_enabled(name),
|
|
paused=controller.is_paused(name),
|
|
quarantined=controller.is_quarantined(name),
|
|
version=definition.version,
|
|
updated_at=controller.store.updated_at(name),
|
|
)
|
|
)
|
|
return FlowsPublic(data=summaries, count=len(summaries))
|
|
|
|
|
|
@router.get("/node-types", response_model=list[NodeTypeInfo])
|
|
def read_node_types() -> Any:
|
|
"""The node types that can be placed on a canvas."""
|
|
from fluksio.flow.controller import node_type_info
|
|
|
|
return node_type_info()
|
|
|
|
|
|
# Above "/{name}" for the same reason "library" is: a flow called "graph" would
|
|
# otherwise be unreachable.
|
|
@router.get("/graph", response_model=BrainGraph)
|
|
def read_graph(controller: FlowControllerDep) -> Any:
|
|
"""Every flow as one graph, with nodes talking to the same thing merged."""
|
|
return controller.brain_graph()
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Shared nodes
|
|
#
|
|
# Declared above the "/{name}" routes: "library" would otherwise be read as a
|
|
# flow name.
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/library", response_model=list[LibraryNode])
|
|
def read_library(controller: FlowControllerDep) -> Any:
|
|
"""The node sources shared across flows, and which nodes use each."""
|
|
return [
|
|
LibraryNode(name=name, used_by=controller.store.usages(name))
|
|
for name in controller.store.list_lib()
|
|
]
|
|
|
|
|
|
@router.delete("/library/{lib_name}", response_model=Message)
|
|
async def delete_shared_node(
|
|
lib_name: str, controller: FlowControllerDep, detach: bool = False
|
|
) -> Any:
|
|
"""Remove a shared source, as long as no flow still runs it.
|
|
|
|
With ``detach``, every node using it keeps a copy of the code as its own
|
|
first, so the source can be retired while flows still use it.
|
|
"""
|
|
if detach:
|
|
running = await run_in_threadpool(controller.store.usages, lib_name, False)
|
|
try:
|
|
await run_in_threadpool(controller.store.detach_lib, lib_name)
|
|
except LibNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No shared node '{lib_name}'")
|
|
for flow in sorted({usage.split(".", 1)[0] for usage in running}):
|
|
await controller.reload_flow(flow)
|
|
return Message(message=f"Deleted shared node '{lib_name}'")
|
|
# The published flows too: a draft that unshared a node still runs it.
|
|
used_by = sorted(
|
|
{*controller.store.usages(lib_name), *controller.store.usages(lib_name, False)}
|
|
)
|
|
if used_by:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"'{lib_name}' is still used by {', '.join(used_by)}",
|
|
)
|
|
try:
|
|
await run_in_threadpool(controller.store.delete_lib_source, lib_name)
|
|
except LibNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No shared node '{lib_name}'")
|
|
return Message(message=f"Deleted shared node '{lib_name}'")
|
|
|
|
|
|
@router.get("/{name}", response_model=FlowDetail)
|
|
def read_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Read one flow, with the state of its nodes."""
|
|
return _detail(controller, _read_flow(controller, name))
|
|
|
|
|
|
def _check_flavors(definition: FlowDef) -> None:
|
|
"""Refuse a size nobody stored, here rather than when the node is built.
|
|
|
|
Catching it at the save covers the canvas and ``fluksio sync`` in one
|
|
place; a node that only finds out at build time is a red node somebody has
|
|
to go and look at.
|
|
"""
|
|
named = {
|
|
node.resources.flavor
|
|
for node in definition.nodes
|
|
if node.resources is not None and node.resources.flavor
|
|
}
|
|
if not named:
|
|
return
|
|
with Session(engine) as session:
|
|
known = set(session.exec(select(Flavor.name)).all())
|
|
unknown = sorted(named - known)
|
|
if unknown:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"No flavor named '{unknown[0]}' "
|
|
f"(known: {', '.join(sorted(known)) or 'none'})",
|
|
)
|
|
|
|
|
|
def _check_origin(definition: FlowDef) -> None:
|
|
"""Refuse an ``origin.repo`` that is not a project's own directory.
|
|
|
|
Every run of a code-defined flow hashes the python files below it, so
|
|
``/`` or a home directory makes each run walk far more than the project,
|
|
and a relative path names nothing the engine can find.
|
|
"""
|
|
repo = definition.origin.repo if definition.origin is not None else ""
|
|
if not repo:
|
|
return
|
|
if not os.path.isabs(repo):
|
|
problem = "is a relative path"
|
|
elif os.path.realpath(repo) == "/":
|
|
problem = "is the filesystem root"
|
|
elif os.path.realpath(repo) == os.path.realpath(Path.home()):
|
|
problem = "is the home directory"
|
|
else:
|
|
return
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=(
|
|
f"origin.repo '{repo}' {problem}. It has to be the absolute path "
|
|
"of the repository the flow is declared in, since every run hashes "
|
|
"the python files below it. Sync from inside that repository."
|
|
),
|
|
)
|
|
|
|
|
|
def _worker_env(node: NodeDef | None) -> dict[str, str]:
|
|
return node.resources.env if node is not None and node.resources else {}
|
|
|
|
|
|
def _changes_code(node: NodeDef, old: NodeDef | None) -> bool:
|
|
"""Whether saving *node* over *old* changes code the engine would run.
|
|
|
|
A function node's body has its own route, gated there. What the document
|
|
carries is which nodes run code at all, which shared source each runs, the
|
|
command in an exec node's settings, and the environment a node's worker
|
|
starts with — where `PYTHONPATH` alone is enough to import a file of
|
|
somebody's choosing.
|
|
"""
|
|
if _worker_env(old) != _worker_env(node):
|
|
return True
|
|
spec = NODE_TYPES.get(node.type)
|
|
if spec is None or not spec.runs_code:
|
|
return False
|
|
return (
|
|
old is None
|
|
or old.type != node.type
|
|
or old.source_ref != node.source_ref
|
|
or (not spec.has_source and old.params != node.params)
|
|
)
|
|
|
|
|
|
def _refuse_code_changes(definition: FlowDef, before: FlowDef | None) -> None:
|
|
"""Refuse a non-superuser a change to code this flow would run.
|
|
|
|
*before* is the working copy being replaced, None for a new flow. Wiring,
|
|
titles, a function node's settings and every other node stay anyone's.
|
|
"""
|
|
previous = {node.id: node for node in before.nodes} if before else {}
|
|
for node in definition.nodes:
|
|
if _changes_code(node, previous.get(node.id)):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"{CODE_IS_SUPERUSERS} (node '{node.id}')",
|
|
)
|
|
|
|
|
|
@router.put("/{name}", response_model=FlowDetail)
|
|
async def save_flow(
|
|
name: str,
|
|
definition: FlowDef,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
) -> Any:
|
|
"""Save unpublished changes to a flow.
|
|
|
|
This writes a draft: the running pipeline keeps the published version until
|
|
someone publishes. ``version`` is the one the editor last saw — a mismatch
|
|
means another client saved in between and answers 409 rather than throwing
|
|
their work away.
|
|
"""
|
|
if definition.name != name:
|
|
raise HTTPException(
|
|
status_code=400, detail="The flow name in the body must match the URL"
|
|
)
|
|
|
|
duplicates = {n.id for n in definition.nodes}
|
|
if len(duplicates) != len(definition.nodes):
|
|
raise HTTPException(status_code=400, detail="Node names must be unique")
|
|
|
|
_check_flavors(definition)
|
|
_check_origin(definition)
|
|
if not user.is_superuser:
|
|
before = _read_flow(controller, name) if controller.store.exists(name) else None
|
|
_refuse_code_changes(definition, before)
|
|
|
|
try:
|
|
stored = await run_in_threadpool(
|
|
controller.store.write_draft, definition, definition.version
|
|
)
|
|
except StaleVersion as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={"message": str(exc), "current_version": exc.current},
|
|
)
|
|
return _detail(controller, stored)
|
|
|
|
|
|
@router.post("/{name}/publish", response_model=FlowDetail)
|
|
async def publish_flow(
|
|
name: str,
|
|
body: PublishRequest,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
) -> Any:
|
|
"""Deploy the unpublished changes: the engine picks them up from here."""
|
|
_read_flow(controller, name)
|
|
if not controller.store.has_draft(name):
|
|
raise HTTPException(
|
|
status_code=400, detail=f"Flow '{name}' has no unpublished changes"
|
|
)
|
|
try:
|
|
published = await run_in_threadpool(
|
|
controller.store.publish_flow, name, body.version
|
|
)
|
|
except StaleVersion as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={"message": str(exc), "current_version": exc.current},
|
|
)
|
|
_audit("published", name, user)
|
|
await controller.reload_flow(name)
|
|
return _detail(controller, published)
|
|
|
|
|
|
@router.post("/{name}/discard-draft", response_model=FlowDetail)
|
|
async def discard_draft(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Throw the unpublished changes away and go back to what is running."""
|
|
_read_flow(controller, name)
|
|
if not controller.store.has_draft(name):
|
|
raise HTTPException(
|
|
status_code=400, detail=f"Flow '{name}' has no unpublished changes"
|
|
)
|
|
if not controller.store.is_published(name):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
f"Flow '{name}' has never been published — delete it instead of "
|
|
"discarding it"
|
|
),
|
|
)
|
|
published = await run_in_threadpool(controller.store.discard_draft, name)
|
|
return _detail(controller, published)
|
|
|
|
|
|
@router.delete("/{name}", response_model=Message)
|
|
async def delete_flow(
|
|
name: str, controller: FlowControllerDep, user: CurrentUser, session: SessionDep
|
|
) -> Any:
|
|
"""Delete a flow and everything in it, except the record of what it ran."""
|
|
live = session.exec(
|
|
select(Run.id).where(
|
|
col(Run.flow) == name, col(Run.status).in_(("running", "queued"))
|
|
)
|
|
).first()
|
|
if live is not None:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=(
|
|
f"Flow '{name}' has a run in progress ({live}). Cancel it, or "
|
|
"wait for it to finish, before deleting the flow."
|
|
),
|
|
)
|
|
try:
|
|
await run_in_threadpool(controller.store.delete_flow, name)
|
|
except FlowNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
|
|
_audit("deleted", name, user)
|
|
# Its files are gone; its values and queued work would otherwise linger.
|
|
# Its runs stay: a run record is a record of what ran, and surviving the
|
|
# flow it belonged to is the point of keeping one. `DELETE /runs?flow=`
|
|
# is what clears them, and with `history=true` its cascades too.
|
|
await run_in_threadpool(controller.forget_flow, name)
|
|
await controller.reload_flow(name)
|
|
return Message(message=f"Deleted flow '{name}'")
|
|
|
|
|
|
@router.post("/{name}/rename", response_model=FlowDetail)
|
|
async def rename_flow(
|
|
name: str,
|
|
body: RenameRequest,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Rename a flow, along with every reference to its messages."""
|
|
if not NAME_PATTERN.match(body.new_name):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
"Use lowercase letters, digits and underscores, starting with a letter"
|
|
),
|
|
)
|
|
try:
|
|
renamed = await run_in_threadpool(
|
|
controller.store.rename_flow, name, body.new_name
|
|
)
|
|
except FlowNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
|
|
except FlowExists as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc))
|
|
|
|
# The old name is nobody's namespace now; its values would sit there under a
|
|
# flow that no longer exists. They repopulate under the new name on the next
|
|
# run, so this is cleanup rather than a migration.
|
|
await run_in_threadpool(controller.forget_flow, name)
|
|
await controller.reload()
|
|
return _detail(controller, renamed)
|
|
|
|
|
|
@router.post("/{name}/duplicate", response_model=FlowDetail)
|
|
async def duplicate_flow(
|
|
name: str, controller: FlowControllerDep, user: CurrentUser
|
|
) -> Any:
|
|
"""Copy a flow into a new one that is only a draft, named after it.
|
|
|
|
Nothing to reload: a draft does not run. Its code nodes are new code in
|
|
the copy, so copying them is a superuser's, like writing them.
|
|
"""
|
|
if not user.is_superuser:
|
|
_refuse_code_changes(_read_flow(controller, name), None)
|
|
try:
|
|
copy = await run_in_threadpool(controller.store.duplicate_flow, name)
|
|
except FlowNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
|
|
return _detail(controller, copy)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Node source
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/{name}/nodes/{node_id}/source", response_model=NodeSource)
|
|
def read_node_source(
|
|
name: str,
|
|
node_id: str,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Read a node's Python source, including unpublished edits."""
|
|
definition = _read_flow(controller, name)
|
|
ref = _source_ref(definition, node_id)
|
|
if ref:
|
|
try:
|
|
return NodeSource(code=controller.store.read_lib_source(ref))
|
|
except LibNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No shared node '{ref}'")
|
|
return NodeSource(
|
|
code=controller.store.read_node_source(name, node_id, draft=True),
|
|
missing=not controller.store.has_node_source(name, node_id, draft=True),
|
|
)
|
|
|
|
|
|
def _follow_header(
|
|
controller: FlowController, name: str, node_id: str, code: str, drop: bool
|
|
) -> FlowDef | None:
|
|
"""Bring a function node's ports and settings in line with its header.
|
|
|
|
The draft is what the editor is on, so that is what moves, and only for a
|
|
node with its own code: a shared one serves other flows with other ports,
|
|
and a synced flow's ports are declared in its repository. Returns the
|
|
draft when it changed.
|
|
"""
|
|
definition = controller.store.read_flow(name, draft=True)
|
|
node = next((n for n in definition.nodes if n.id == node_id), None)
|
|
spec = NODE_TYPES.get(node.type) if node is not None else None
|
|
if (
|
|
node is None
|
|
or spec is None
|
|
or not spec.free_params
|
|
or node.source_ref
|
|
or definition.origin
|
|
):
|
|
return None
|
|
followed = follow_header(node, code, drop)
|
|
if followed == node:
|
|
return None
|
|
definition.nodes = [followed if n.id == node_id else n for n in definition.nodes]
|
|
try:
|
|
return controller.store.write_draft(definition, definition.version)
|
|
except StaleVersion:
|
|
# Someone saved in between; the next save of this code tries again.
|
|
return None
|
|
|
|
|
|
@router.put(
|
|
"/{name}/nodes/{node_id}/source",
|
|
response_model=NodeSourceSaved,
|
|
dependencies=[Depends(get_code_author)],
|
|
)
|
|
async def save_node_source(
|
|
name: str,
|
|
node_id: str,
|
|
source: NodeSource,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
drop: bool = True,
|
|
) -> Any:
|
|
"""Save a node's source as an unpublished edit and report whether it loads.
|
|
|
|
The answer comes from compiling the code rather than from the running
|
|
pipeline: a draft is not deployed, and compiling is both faster and more
|
|
precise about what the author just typed.
|
|
|
|
A shared node writes to the library, so the fix reaches every flow using
|
|
it — and that one is live immediately rather than waiting for a publish,
|
|
because the copy is not any single flow's to hold back. Being live, it is
|
|
audited like a publish.
|
|
|
|
A node with its own code takes its input ports and settings from the
|
|
header of ``process`` (see `follow_header`), and the answer carries the
|
|
node when that changed it. ``drop=false`` is the editor's autosave, which
|
|
adds and renames from the header but keeps what it no longer names until
|
|
the editor is left.
|
|
"""
|
|
definition = _read_flow(controller, name)
|
|
ref = _source_ref(definition, node_id)
|
|
if ref:
|
|
changed = await run_in_threadpool(
|
|
controller.store.write_lib_source, ref, source.code
|
|
)
|
|
if changed:
|
|
_audit(f"edited shared node '{ref}'", name, user)
|
|
await controller.reload_lib_users(ref)
|
|
else:
|
|
await run_in_threadpool(
|
|
controller.store.write_node_source, name, node_id, source.code, True
|
|
)
|
|
followed = await run_in_threadpool(
|
|
_follow_header, controller, name, node_id, source.code, drop
|
|
)
|
|
node_def = next((n for n in definition.nodes if n.id == node_id), None)
|
|
worker = (
|
|
node_def.worker
|
|
if node_def is not None and node_def.worker_policy == "require"
|
|
else None
|
|
)
|
|
error = await run_in_threadpool(
|
|
controller.compile_check, name, node_id, source.code, worker
|
|
)
|
|
return NodeSourceSaved(
|
|
id=f"{name}.{node_id}",
|
|
status="error" if error else "active",
|
|
error=error,
|
|
node=next((n for n in followed.nodes if n.id == node_id), None)
|
|
if followed
|
|
else None,
|
|
version=followed.version if followed else None,
|
|
)
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/share", response_model=FlowDetail)
|
|
async def share_node(
|
|
name: str,
|
|
node_id: str,
|
|
body: ShareRequest,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Move this node's code into the library so other flows can run it too."""
|
|
_read_flow(controller, name)
|
|
if not NAME_PATTERN.match(body.lib_name):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
"Use lowercase letters, digits and underscores, starting with a letter"
|
|
),
|
|
)
|
|
try:
|
|
await run_in_threadpool(
|
|
controller.store.share_node, name, node_id, body.lib_name
|
|
)
|
|
except LibExists as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc))
|
|
return _detail(controller, _read_flow(controller, name))
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/unshare", response_model=FlowDetail)
|
|
async def unshare_node(
|
|
name: str,
|
|
node_id: str,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Take a private copy of the shared code back into this flow."""
|
|
_read_flow(controller, name)
|
|
try:
|
|
await run_in_threadpool(controller.store.unshare_node, name, node_id)
|
|
except LibNotFound as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
return _detail(controller, _read_flow(controller, name))
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Running, stopped, paused
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/{name}/start", response_model=FlowDetail)
|
|
async def start_flow(
|
|
name: str, controller: FlowControllerDep, user: CurrentUser
|
|
) -> Any:
|
|
"""Let the engine run this flow again."""
|
|
_read_flow(controller, name)
|
|
await controller.set_enabled(name, True)
|
|
_audit("started", name, user)
|
|
return _detail(controller, _read_flow(controller, name))
|
|
|
|
|
|
@router.post("/{name}/stop", response_model=FlowDetail)
|
|
async def stop_flow(name: str, controller: FlowControllerDep, user: CurrentUser) -> Any:
|
|
"""Take this flow off the engine: no subscriptions, schedules or webhooks."""
|
|
_read_flow(controller, name)
|
|
await controller.set_enabled(name, False)
|
|
_audit("stopped", name, user)
|
|
return _detail(controller, _read_flow(controller, name))
|
|
|
|
|
|
@router.post("/{name}/pause", response_model=Message)
|
|
def pause_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Hold the flow's nodes so its messages can be stepped through."""
|
|
_read_flow(controller, name)
|
|
controller.pause_flow(name)
|
|
return Message(message=f"Paused flow '{name}'")
|
|
|
|
|
|
@router.post("/{name}/resume", response_model=Message)
|
|
async def resume_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Let the flow carry on, running whatever was held back."""
|
|
_read_flow(controller, name)
|
|
await run_in_threadpool(controller.resume_flow, name)
|
|
return Message(message=f"Resumed flow '{name}'")
|
|
|
|
|
|
@router.post("/{name}/step", response_model=Message)
|
|
async def step_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Run one message a pause is holding back, leaving the flow paused."""
|
|
_read_flow(controller, name)
|
|
node = await run_in_threadpool(controller.step_flow, name)
|
|
if node is None:
|
|
return Message(message=f"Nothing held back in flow '{name}'")
|
|
return Message(message=f"Stepped '{node}'")
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Validation and execution
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/{name}/validate", response_model=ValidationResult)
|
|
def validate_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Report what would keep this flow from running."""
|
|
_read_flow(controller, name)
|
|
if controller.store.has_draft(name):
|
|
return ValidationResult(issues=controller.preview(name).issues)
|
|
return ValidationResult(issues=controller.flow_issues(name))
|
|
|
|
|
|
@router.post("/{name}/run", response_model=FlowStatePublic)
|
|
async def run_flow(
|
|
name: str,
|
|
body: RunRequest,
|
|
request: Request,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
) -> Any:
|
|
"""Run every node of a flow once.
|
|
|
|
With unpublished changes this runs the draft, so the button matches what is
|
|
on the canvas. Nothing is deployed by running it.
|
|
|
|
A *batch* flow is submitted as a run instead, because that is what running
|
|
one means: it is what keeps the parameters, the series and the result, and
|
|
a button that quietly did something else would be a trap.
|
|
"""
|
|
definition = _read_flow(controller, name)
|
|
_require_enabled(controller, name)
|
|
if definition.mode == "batch":
|
|
service = getattr(request.app.state, "run_service", None)
|
|
if service is None:
|
|
raise HTTPException(status_code=503, detail="Runs are not available")
|
|
try:
|
|
await run_in_threadpool(
|
|
service.submit,
|
|
name,
|
|
params=body.inputs,
|
|
cause="api",
|
|
actor=user.email,
|
|
draft=controller.store.has_draft(name),
|
|
)
|
|
except RunRejected as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
return _flow_state(controller, name)
|
|
|
|
inputs = {qualify(name, key): value for key, value in body.inputs.items()}
|
|
if controller.store.has_draft(name):
|
|
await run_in_threadpool(controller.run_preview, name, inputs)
|
|
else:
|
|
await run_in_threadpool(controller.run_flow, name, inputs)
|
|
return _flow_state(controller, name)
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/trigger", response_model=FlowStatePublic)
|
|
async def trigger_node(
|
|
name: str,
|
|
node_id: str,
|
|
body: TriggerRequest,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Feed values into a single node."""
|
|
_require_enabled(controller, name)
|
|
try:
|
|
error = await run_in_threadpool(
|
|
controller.trigger_node, f"{name}.{node_id}", body.values
|
|
)
|
|
except KeyError:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
|
|
)
|
|
if error:
|
|
# The canvas already has the failure from the bus; the person who
|
|
# clicked gets to read it too, instead of a stack trace in the log.
|
|
raise HTTPException(status_code=400, detail=error)
|
|
return _flow_state(controller, name)
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/cancel", response_model=Message)
|
|
def cancel_node(name: str, node_id: str, controller: FlowControllerDep) -> Any:
|
|
"""Stop a node that is running right now, by killing the worker running it.
|
|
|
|
Idempotent on purpose: by the time a click reaches here the node may well
|
|
have finished, and that is the outcome that was asked for.
|
|
"""
|
|
pool = controller.workers
|
|
if pool is not None and pool.cancel(f"{name}.{node_id}"):
|
|
return Message(message=f"Stopped '{node_id}'")
|
|
return Message(message=f"'{node_id}' was not running")
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/acknowledge", response_model=Message)
|
|
def acknowledge_node_error(
|
|
name: str, node_id: str, controller: FlowControllerDep
|
|
) -> Any:
|
|
"""Dismiss what a node last failed with, so the canvas stops marking it.
|
|
|
|
A failure outlives the next good run on purpose — otherwise one that fired
|
|
an alert leaves no trace by the time anyone looks. Reading the traceback is
|
|
what says it has been seen.
|
|
"""
|
|
try:
|
|
controller.acknowledge_error(f"{name}.{node_id}")
|
|
except KeyError:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
|
|
) from None
|
|
# Another browser holds the same marker and has no reason to refetch, so it
|
|
# would keep showing a failure that is gone until its next snapshot.
|
|
event_bus.publish(
|
|
{
|
|
"type": "node_error_acknowledged",
|
|
"node": f"{name}.{node_id}",
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
return Message(message=f"Cleared the failure on '{node_id}'")
|
|
|
|
|
|
@router.get("/{name}/state", response_model=FlowStatePublic)
|
|
def read_flow_state(name: str, controller: FlowControllerDep) -> Any:
|
|
"""The last value seen on every message of this flow."""
|
|
_read_flow(controller, name)
|
|
return _flow_state(controller, name)
|
|
|
|
|
|
@router.get("/{name}/history/{message}", response_model=MessageHistory)
|
|
def read_message_history(
|
|
name: str,
|
|
message: str,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""The recent values of one message, for plotting.
|
|
|
|
``message`` may be given bare or qualified; a message that never carried a
|
|
number or a flag comes back with an empty series.
|
|
"""
|
|
_read_flow(controller, name)
|
|
key = qualify(name, message)
|
|
return MessageHistory(
|
|
message=key,
|
|
numeric=as_number(controller.state.get(key)) is not None,
|
|
points=[
|
|
HistoryPoint(ts=ts, value=value)
|
|
for ts, value in controller.state.history(key)
|
|
],
|
|
)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Live updates
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def snapshot_payload(
|
|
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.
|
|
|
|
Also sent by the tunnel connector, which serves this websocket inline: the
|
|
two have to agree, so they build the message here rather than each their
|
|
own. ``emits`` is what the bus counted while nobody was listening — a
|
|
client that reconnects between two pages would otherwise start from zero.
|
|
``queued`` is the nodes waiting for a machine, for the same reason: the bus
|
|
announces a wait once, and a client that missed it would draw the node idle.
|
|
|
|
``only`` bounds it to a set of message names, which is what a wall panel
|
|
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.
|
|
payload: dict[str, Any] = {
|
|
"type": "snapshot",
|
|
"values": {k: v for k, v in controller.values().items() if k in only},
|
|
"nodes": [],
|
|
"issues": [],
|
|
"paused": [],
|
|
"logs": [],
|
|
"emits": {},
|
|
"queued": [],
|
|
}
|
|
if device:
|
|
payload["demand"] = demand.ports(device)
|
|
payload["uplink"] = uplink
|
|
return payload
|
|
return {
|
|
"type": "snapshot",
|
|
"values": controller.values(),
|
|
"nodes": [s.model_dump() for s in controller.node_statuses()],
|
|
"issues": [i.model_dump() for i in controller.issues],
|
|
"paused": controller.paused_flows(),
|
|
"logs": list(event_bus.recent_logs),
|
|
"emits": dict(event_bus.emits),
|
|
"queued": controller.placer.snapshot()["waiting"] if controller.placer else [],
|
|
}
|
|
|
|
|
|
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:
|
|
return str(decode_token(token, app=app).get("panel") or "")
|
|
except InvalidTokenError:
|
|
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:
|
|
return None
|
|
store: DashboardStore | None = getattr(app.state, "dashboard_store", None)
|
|
devices: DeviceStore | None = getattr(app.state, "device_store", None)
|
|
if store is None or devices is None:
|
|
return set()
|
|
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:
|
|
"""Whether a panel's socket should carry this event.
|
|
|
|
The same bound as the snapshot above, applied to the stream that follows
|
|
it: a value the panel draws, and nothing else on the bus — save for a
|
|
dashboard being published, which is how a screen hears that the document
|
|
it is drawing, or the set of them it was given, has moved.
|
|
"""
|
|
if event.get("type") == "dashboard_changed":
|
|
return True
|
|
return event.get("type") == "message_value" and str(event.get("name") or "") in only
|
|
|
|
|
|
# How many events one frame may carry. A ceiling rather than a target: the
|
|
# batch is whatever the bus happens to hold, and a client that has been away
|
|
# 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, 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:
|
|
"""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}
|
|
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
|
|
|
|
|
|
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, 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 []
|
|
|
|
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 "")
|
|
payload = store.volatile.get(digest)
|
|
if payload is None:
|
|
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
|
|
|
|
|
|
#: 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`.
|
|
|
|
Serialised once with orjson rather than per client with the stdlib, which
|
|
is what `send_json` does.
|
|
"""
|
|
payload = events[0] if len(events) == 1 else {"type": "batch", "events": events}
|
|
await websocket.send_text(orjson.dumps(payload).decode())
|
|
|
|
|
|
class SocketTicket(BaseModel):
|
|
ticket: str
|
|
expires_in: float
|
|
|
|
|
|
@router.post("/ws-ticket", response_model=SocketTicket)
|
|
def issue_socket_ticket(token: TokenDep) -> Any:
|
|
"""A one-time pass for opening `/flows/ws`, good for thirty seconds.
|
|
|
|
A browser cannot set a header on a websocket handshake, and a token in the
|
|
URL ends up in every access log on the way. The ticket goes there instead
|
|
and stands for the bearer token of this call, panel credentials included.
|
|
"""
|
|
return SocketTicket(ticket=issue_ticket(token), expires_in=TICKET_TTL_S)
|
|
|
|
|
|
@ws_router.websocket("/ws")
|
|
async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -> None:
|
|
"""Stream values, node status and execution events as they happen.
|
|
|
|
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)
|
|
|
|
# On a thread: authenticating is a database round trip, and the snapshot
|
|
# below reads the whole of state. Neither belongs on the event loop, which
|
|
# every other socket and every request is sharing.
|
|
def _authenticate() -> Any:
|
|
with Session(engine) as session:
|
|
user = user_from_token(session, token, app=websocket.app)
|
|
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, _device_def(websocket.app, device)
|
|
|
|
try:
|
|
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
|
|
# being gone and checks it, so this one is told to try again later.
|
|
logger.warning("Could not check a socket's credential: %s", exc)
|
|
await websocket.accept()
|
|
await websocket.close(code=1013)
|
|
return
|
|
if user is None:
|
|
await websocket.close(code=1008)
|
|
return
|
|
# 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()
|
|
|
|
controller: FlowController | None = getattr(
|
|
websocket.app.state, "flow_controller", None
|
|
)
|
|
store = getattr(websocket.app.state, "artifact_store", None)
|
|
|
|
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", []))
|
|
|
|
told = await send_snapshot()
|
|
|
|
async with event_bus.subscribe() as queue:
|
|
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())
|
|
done, _ = await asyncio.wait(
|
|
{sender, receiver}, return_when=asyncio.FIRST_COMPLETED
|
|
)
|
|
if receiver in done:
|
|
if sender not in done:
|
|
# Nobody will ever read this one: the next turn makes
|
|
# its own. Left behind it stays parked on the queue's
|
|
# waiter list, takes an event the live reader wanted,
|
|
# and is reported as a task destroyed while pending.
|
|
sender.cancel()
|
|
# A frame from the client. Only a disconnect ends the
|
|
# 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.
|
|
failure = receiver.exception()
|
|
if failure is not None:
|
|
# 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
|
|
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
|
|
# least one of the two, so `receiver` missing means this one is.
|
|
|
|
# Everything the bus has right now, not just the one event
|
|
# that woke this: a cascade puts a dozen in at once, and one
|
|
# frame carrying them costs one wakeup rather than a dozen.
|
|
batch = [sender.result()]
|
|
while len(batch) < MAX_FRAME_EVENTS:
|
|
try:
|
|
batch.append(queue.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
|
|
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
|
|
# ever updates. ``or set()`` because a panel that was
|
|
# deleted resolves to None, the same as a person's
|
|
# 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 = []
|
|
await send_snapshot()
|
|
if only is not None and not event_for_panel(event, only):
|
|
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
|
|
# ("websocket.send after websocket.close") rather than the clean
|
|
# disconnect. Either way the socket is gone and the loop is over.
|
|
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)
|