diff --git a/backend/fluksio/api/deps.py b/backend/fluksio/api/deps.py index 6110016..cf57a94 100644 --- a/backend/fluksio/api/deps.py +++ b/backend/fluksio/api/deps.py @@ -21,6 +21,7 @@ from fluksio.flow.controller import FlowController from fluksio.flow.dashboards import DashboardDef, DashboardStore from fluksio.flow.devices import DeviceDef, DeviceStore, DeviceUnavailable, presence from fluksio.flow.events import event_bus +from fluksio.flow.rename import Renamer from fluksio.flow.workers import PythonWorkerPool from fluksio.models import OAuthClient, TokenPayload, User, WorkerToken @@ -583,6 +584,16 @@ def get_device_store(request: Request) -> DeviceStore: DeviceStoreDep = Annotated[DeviceStore, Depends(get_device_store)] +def get_renamer(request: Request) -> Renamer: + renamer: Renamer | None = getattr(request.app.state, "renamer", None) + if renamer is None: + raise HTTPException(status_code=503, detail="The flow engine is not running") + return renamer + + +RenamerDep = Annotated[Renamer, Depends(get_renamer)] + + def get_worker_pool(request: Request) -> PythonWorkerPool: pool: PythonWorkerPool | None = getattr(request.app.state, "worker_pool", None) if pool is None: diff --git a/backend/fluksio/api/routes/dashboards.py b/backend/fluksio/api/routes/dashboards.py index 05e8bea..4c6ae75 100644 --- a/backend/fluksio/api/routes/dashboards.py +++ b/backend/fluksio/api/routes/dashboards.py @@ -7,7 +7,7 @@ label unique among dashboards and what ``GET /dashboards?name=`` looks up. import time from typing import Any -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.concurrency import run_in_threadpool from pydantic import BaseModel @@ -15,6 +15,7 @@ from fluksio.api.deps import ( CurrentUser, DashboardStoreDep, FlowControllerDep, + RenamerDep, check_pin, get_current_active_superuser, get_current_user, @@ -30,7 +31,13 @@ from fluksio.flow.dashboards import ( results_dashboard, ) from fluksio.flow.events import event_bus -from fluksio.flow.rename import RenamePlan, RenamePlanRequest, plan_rename +from fluksio.flow.rename import ( + RenamePlan, + RenamePlanRequest, + RenameRequest, + RenameResult, + plan_rename, +) from fluksio.flow.store import FlowNotFound, IdTaken, StaleVersion from fluksio.models import Message @@ -279,3 +286,29 @@ async def plan_dashboard_rename( ) except DashboardNotFound: raise _missing(dashboard_id) + + +@router.post( + "/{dashboard_id}/rename", + response_model=RenameResult, + dependencies=[Depends(get_current_active_superuser)], +) +async def rename_dashboard( + dashboard_id: str, + body: RenameRequest, + request: Request, + renamer: RenamerDep, + user: CurrentUser, +) -> Any: + """Rename this dashboard's key, as its plan described it. + + Its two documents and nothing else: devices list it by id, and nothing + binds to it by key. The same ``request_id`` again gets the same answer; + refusals are as for a flow. + """ + try: + return await renamer.rename( + "dashboard", dashboard_id, body, user.email, request.is_disconnected + ) + except DashboardNotFound: + raise _missing(dashboard_id) diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index cf20e7d..a7fb662 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -35,6 +35,7 @@ from fluksio.api.deps import ( DashboardStoreDep, FlowControllerDep, HiddenDep, + RenamerDep, SessionDep, TokenDep, check_pin, @@ -69,7 +70,13 @@ from fluksio.flow.events import event_bus from fluksio.flow.messages import MEDIA_FAMILIES, DType, flow_of, qualify from fluksio.flow.nodes.wren import TRAIN_FLOW, WrenNode, pack_meta from fluksio.flow.pipeline import EngineHeld, ValidationIssue -from fluksio.flow.rename import RenamePlan, RenamePlanRequest, plan_rename +from fluksio.flow.rename import ( + RenamePlan, + RenamePlanRequest, + RenameRequest, + RenameResult, + plan_rename, +) from fluksio.flow.runs import RunRejected from fluksio.flow.schemas import ( NAME_PATTERN, @@ -776,6 +783,60 @@ async def plan_flow_rename( raise _missing(flow_id) +@router.post( + "/{flow_id}/rename", + response_model=RenameResult, + dependencies=[Depends(get_current_active_superuser)], +) +async def rename_flow( + flow_id: str, + body: RenameRequest, + request: Request, + renamer: RenamerDep, + user: CurrentUser, +) -> Any: + """Rename this flow's key, as its plan described it. + + Everything named under the key moves: every flow and dashboard binding + one of its messages, its history, live state and queued work. The engine + waits meanwhile, for as long as the work running when it was asked takes + to finish. Until then it can be called off, with ``DELETE`` on the + request or by closing the connection, and nothing changes (409 + ``aborted``); ``GET /observability/summary`` says what it waits for. + + The same ``request_id`` again gets the same answer. A rename that + could not finish yet answers 503 ``completing`` and is finished in the + background. Refusals are the plan's codes (409, 422 for a key that is no + key), ``stale`` and ``plan_changed`` (409). + """ + try: + return await renamer.rename( + "flow", flow_id, body, user.email, request.is_disconnected + ) + except FlowNotFound: + raise _missing(flow_id) + + +@router.delete( + "/{flow_id}/rename/{request_id}", + response_model=Message, + dependencies=[Depends(get_current_active_superuser)], +) +async def cancel_flow_rename(flow_id: str, request_id: str, renamer: RenamerDep) -> Any: + """Call off a rename still waiting for the engine. It answers 409 + ``aborted`` once everything is as it was, and one that has begun to + change things refuses this with 409 ``too_late``.""" + # Async, so on the event loop, where the rename decides when it is past + # calling off. + try: + renamer.cancel(flow_id, request_id) + except KeyError: + raise HTTPException( + status_code=404, detail=f"No rename '{request_id}' of '{flow_id}' waits" + ) + return Message(message=f"Calling off rename '{request_id}'") + + # ----------------------------------------------------------------------------- # Node source # ----------------------------------------------------------------------------- diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 4c1e913..9c6bf8f 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -28,6 +28,7 @@ from fluksio.api.deps import ( from fluksio.api.routes.runs import elapsed_ms from fluksio.core.config import settings from fluksio.flow.controller import NodeStatus +from fluksio.flow.rename import RenameStatus from fluksio.flow.tombstones import names_of, shown_nodes, stored_node from fluksio.models import EngineEvent, FlowRun, Message, MetricBucket @@ -65,6 +66,9 @@ class HealthSummary(BaseModel): #: route the client knows and the engine does not answers a flat 404. This #: is what turns that into a sentence. Absent means older than this field. version: str = "" + #: A key rename under way: what it waits for while it can still be + #: called off, and why a step failed while it is retried. + rename: RenameStatus | None = None class SeriesPoint(BaseModel): @@ -285,6 +289,7 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: f"{', '.join(sorted(e.id for e in failed_once))}" ) + renamer = getattr(request.app.state, "renamer", None) return HealthSummary( status="degraded" if problems else "ok", problems=problems, @@ -322,6 +327,7 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: else {"ewma": 0.0, "max_60s": 0.0} ), version=__version__, + rename=renamer.pending if renamer is not None else None, ) diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index 90145a1..a8ed12d 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -99,18 +99,24 @@ def _mention_undeclared_cards() -> None: if settings.FLOW_GPUS or machine_gpus(): return from fluksio.flow.runs import required_resources - from fluksio.flow.store import FlowStore + from fluksio.flow.schemas import FlowDef - store = FlowStore(settings.FLOWS_DIR) + # Read as files, not through a `FlowStore`: building one can commit the + # store, and this runs before the engine has settled an interrupted + # rename and given every document its id. asking = [] - for name in store.list_flows(): + for path in sorted(settings.FLOWS_DIR.glob("*/flow.json")): + # `_devices/flow.json` is a device, not a flow. + if path.parent.name.startswith("_"): + continue try: - needs = required_resources(store.read_flow(name)) + flow = FlowDef.model_validate_json(path.read_text()) + needs = required_resources(flow) except Exception: # A flow that will not parse is the engine's to complain about. continue if (needs or {}).get("gpus"): - asking.append(name) + asking.append(flow.name) if not asking: return named = ", ".join(sorted(asking)[:3]) + (" …" if len(asking) > 3 else "") diff --git a/backend/fluksio/flow/alerts.py b/backend/fluksio/flow/alerts.py index da1b8fc..d6ef6a7 100644 --- a/backend/fluksio/flow/alerts.py +++ b/backend/fluksio/flow/alerts.py @@ -17,7 +17,7 @@ import logging import time from collections import deque from collections.abc import Callable -from typing import Any, Literal, Protocol +from typing import TYPE_CHECKING, Any, Literal, Protocol import httpx from pydantic import BaseModel, Field @@ -25,6 +25,9 @@ from pydantic import BaseModel, Field from fluksio.flow.events import EventBus from fluksio.flow.secrets import resolve_params +if TYPE_CHECKING: + from fluksio.flow.namespace import Rename + logger = logging.getLogger(__name__) # What is worth waking someone for. Everything else on the bus is traffic. @@ -271,6 +274,25 @@ class AlertManager: return await self._deliver(event) + def rename(self, rename: Rename) -> None: + """Follow a key rename: dashboard channels publish to the moved + message, as `alerts.json` now says, and what was sent, held back or + muted is kept under the new names with its timestamps, so nothing is + sent again. A connection waiting out its grace under an old name is + let go: its node reports again once it runs under the new one.""" + config = self.config.model_copy(deep=True) + for channel in config.channels: + message = channel.config.get("message") + if channel.kind == "dashboard" and isinstance(message, str): + channel.config["message"] = rename.binding(message) + self.config = config + self._last_sent = {rename.dedup_key(k): v for k, v in self._last_sent.items()} + self._suppressed = {rename.dedup_key(k): v for k, v in self._suppressed.items()} + self._health_flips = {rename.name(k): v for k, v in self._health_flips.items()} + self._flapping = {rename.name(k): v for k, v in self._flapping.items()} + for node in [node for node in self._down if rename.name(node) != node]: + self._cancel_down(node) + def _cancel_down(self, node: str) -> None: task = self._down.pop(node, None) if task is not None: diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 981f9c5..ee0c0e9 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -39,7 +39,7 @@ from fluksio.flow.demand import demand from fluksio.flow.events import EventBus from fluksio.flow.executor import ExecutionService from fluksio.flow.messages import MessageSpec, flow_of, qualify -from fluksio.flow.namespace import touches +from fluksio.flow.namespace import Rename, touches from fluksio.flow.nodes import ( RESERVED_SETTINGS, AssistantNode, @@ -913,6 +913,75 @@ class FlowController: ) self._rewired() + async def splice_renamed(self, rename: Rename, rebuilt: list[str]) -> None: + """Put the flows a key rename rewrote back into the graph, built from + their new documents and named as they are now. + + Under the controller lock, which the rename holds, with the engine + held and the flows *rebuilt* (ids) torn down before their documents + changed. Not `reload_flow`, which would take the lock again, nor + `reload`, which clears every pause. Stopped flows stay stopped, which + is stored by id, pauses carry over under the new keys and parked work + stays parked. A rebuilt flow's quarantine is gone with its teardown, + as on a publish. What the documents feed is refreshed before anything + starts: history limits, device wiring and what devices are asked to + send. + """ + pipeline = self.pipeline + if pipeline is None: + return + paused = pipeline.paused_flows() + builds = await run_in_threadpool( + lambda: [(flow_id, self._build_one(flow_id)) for flow_id in rebuilt] + ) + # By the name each node has now, so a failure carries over. + previous = {rename.name(node_id): e for node_id, e in self.loaded.items()} + loaded = dict(self.loaded) + keys = [] + for flow_id, built in builds: + if built is None: + continue + flow, nodes, entries, initial_values, enabled = built + old, key = self._flow_keys.get(flow_id, flow.name), flow.name + if old != key: + pipeline.replace_flow(old, []) + pipeline.replace_flow(key, nodes, initial_values) + self._flow_keys[flow_id] = key + for node_id, entry in entries.items(): + before = previous.get(node_id) + if before is not None and before.last_error: + entry.inherit_failure(before) + loaded = { + node_id: entry for node_id, entry in loaded.items() if entry.flow != old + } | entries + self._flow_inputs.pop(old, None) + self._flow_inputs[key] = _declared_inputs(flow)[0] + self.disabled = (self.disabled - {old}) | (set() if enabled else {key}) + self.batch = (self.batch - {old}) | ( + {key} if flow.mode == "batch" else set() + ) + keys.append(key) + self.loaded = loaded + pipeline.set_disabled(self.disabled) + # The replace cleared the pause of every flow it touched. + for was in map(rename.flow, paused): + if not pipeline.is_paused(was): + pipeline.pause(was) + self.issues = _collect_issues( + self.loaded, pipeline, self._all_flow_inputs(), self.devices + ) + if self.dashboards is not None: + self.set_history_limits( + await run_in_threadpool(self.dashboards.history_requirements) + ) + if self.devices is not None: + # Taken note of without telling the devices: the rename does, + # once it has let go of the engine. + await run_in_threadpool(demand.refresh) + self._device_wiring = await run_in_threadpool(self.devices.wiring) + for key in keys: + await self._activate(key) + async def recheck(self, wait: float | None = FLOW_REBUILD_WAIT) -> None: """Validate the built graph again without rebuilding anything. diff --git a/backend/fluksio/flow/events.py b/backend/fluksio/flow/events.py index 337d4e0..82a6d7e 100644 --- a/backend/fluksio/flow/events.py +++ b/backend/fluksio/flow/events.py @@ -15,7 +15,10 @@ import threading from collections import deque from collections.abc import AsyncIterator, Callable from contextlib import asynccontextmanager -from typing import Any +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from fluksio.flow.namespace import Rename logger = logging.getLogger(__name__) @@ -24,6 +27,8 @@ DISPATCH_BUDGET = 1024 LOG_HISTORY = 400 #: How many of a message's last artifact digests are remembered. RECENT_DIGESTS = 8 +#: How many turns of the loop `settle` gives the dispatch at most. +SETTLE_TURNS = 100 def _critical(event: dict[str, Any]) -> bool: @@ -211,6 +216,29 @@ class EventBus: if self._loop is not None: self._loop.call_soon(self._dispatch) + async def settle(self) -> None: + """Wait until what was published so far has reached every subscriber. + + For a rename, before any key changes: an event a node published under + the old key is then recorded under it. Bounded, since a busy bus may + never be empty. + """ + for _ in range(SETTLE_TURNS): + with self._pending_lock: + if not self._flush_scheduled: + return + await asyncio.sleep(0) + + def rename(self, rename: Rename) -> None: + """Keep emit counts and recent digests under the names a key rename + gave their nodes and messages, so a client reconnecting or a device + fetching a recent frame finds them there. Rebound, not changed in + place: publishers on other threads write to them.""" + self.emits = {rename.name(node): n for node, n in list(self.emits.items())} + self.digests = { + rename.name(name): recent for name, recent in list(self.digests.items()) + } + def dropped(self, queue: asyncio.Queue[dict[str, Any]]) -> bool: """Whether this subscriber lost events since it last asked. diff --git a/backend/fluksio/flow/rename.py b/backend/fluksio/flow/rename.py index 63f3ce8..257589b 100644 --- a/backend/fluksio/flow/rename.py +++ b/backend/fluksio/flow/rename.py @@ -6,7 +6,9 @@ live state and queued work, so nothing may write through the API meanwhile. a rename would change and what refuses it, and `rewrite_documents` is its document part. The `Journal` records each step before the next, so a rename a crash interrupted is undone or finished on the next start (`recover_documents`, -`recover_runtime`). ``docs/private/id_refac_rename.md`` has the whole protocol. +`recover_runtime`). `Renamer` runs one: the gate, the engine holds, the +journal and its steps, and the engine put back together under the new names. +``docs/private/id_refac_rename.md`` has the whole protocol. """ from __future__ import annotations @@ -20,8 +22,8 @@ import re import threading import time from collections import Counter -from collections.abc import Callable, Iterable, Iterator -from dataclasses import dataclass +from collections.abc import Awaitable, Callable, Iterable, Iterator +from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path from typing import Any, Literal, cast @@ -34,8 +36,13 @@ from starlette.types import Scope from fluksio.core.config import settings from fluksio.core.db import engine as db_engine -from fluksio.flow.alerts import AlertsConfig -from fluksio.flow.controller import HOOK_PREFIX, FlowController +from fluksio.flow.alerts import AlertManager, AlertsConfig +from fluksio.flow.controller import ( + HOOK_PREFIX, + REBUILD_WAIT, + FlowController, + RebuildBusy, +) from fluksio.flow.dashboards import ( DASHBOARD_BINDINGS, DASHBOARD_DIR, @@ -44,10 +51,12 @@ from fluksio.flow.dashboards import ( results_name, ) from fluksio.flow.events import EventBus -from fluksio.flow.executor import wait_out +from fluksio.flow.executor import HOLD_POLL_S, ExecutionService, wait_out from fluksio.flow.identity import _replace, write_documents from fluksio.flow.messages import flow_of +from fluksio.flow.metrics import MetricsCollector from fluksio.flow.namespace import FLOW_FIELDS, NAME, OUTPUTS, Rename, names, rebind +from fluksio.flow.pipeline import Pipeline from fluksio.flow.queue import WorkQueue from fluksio.flow.schemas import FlowDef, _validate_name from fluksio.flow.state import MemoryState, StateBackend @@ -73,13 +82,13 @@ _READS = frozenset({"GET", "HEAD", "OPTIONS"}) #: and are held there), a browser's push subscription, the portal, workers #: attaching, artifact bytes (which are named by digest, and which a node on a #: remote worker writes while the rename waits for it), the socket ticket and -#: the rename routes, which keep their own order. +#: the rename routes (plan, rename and its cancel), which keep their own order. _EXEMPT = re.compile( r"/(?:login|password-recovery|password-recovery-html-content|reset-password" r"|oauth|users|ai|cloud|workers|artifacts)(?:/|$)" r"|/alerts/webpush(?:/|$)" r"|/flows/ws-ticket$" - r"|/(?:flows|dashboards)/[^/]+/rename(?:/plan)?$" + r"|/(?:flows|dashboards)/[^/]+/rename(?:/[^/]+)?$" ) @@ -353,6 +362,38 @@ class RenamePlanRequest(BaseModel): new_name: str +class RenameRequest(BaseModel): + """A rename as its plan described it. Refused when the object or the + plan moved since (409 ``stale`` or ``plan_changed``).""" + + #: The client's own id for this request, which a retry repeats. + request_id: str = Field(min_length=1, max_length=64) + expected_name: str + #: The working copy's version the plan was made against. + expected_version: int + new_name: str + fingerprint: str + + +class RenameStatus(BaseModel): + """A rename under way, for whoever is waiting on it or was turned away.""" + + op: str + kind: Kind + id: str + old: str + new: str + #: ``draining`` while it waits for work to finish, when it can still be + #: called off, then the journal's phase: ``prepared`` while the + #: documents are written and ``forward`` while the rest moves. + phase: str + #: What it is waiting for: API requests (``POST /api/v1/...``), nodes + #: and messages running outside the queue, and nodes whose cascades run. + busy: list[str] = [] + #: Why the last attempt at a step failed, while it is retried. + error: str = "" + + class RenameRefusal(BaseModel): """Why a rename cannot go ahead, with the ids of what is in the way.""" @@ -856,13 +897,15 @@ def read_journal() -> Journal | None: class RenameConflict(Exception): """A rename request that cannot go ahead as asked, with the status and - code it is answered with.""" + code it is answered with, and what else the answer carries: the ``op`` + in the way, or the fresh ``plan`` a refusal comes from.""" - def __init__(self, status: int, code: str, detail: str) -> None: + def __init__(self, status: int, code: str, detail: str, **extra: Any) -> None: super().__init__(detail) self.status = status self.code = code self.detail = detail + self.extra = extra def repeated( @@ -1247,3 +1290,533 @@ async def retry_until_done(attempt: Callable[[], None], events: EventBus) -> Non ) await asyncio.sleep(wait) wait = min(wait * 2, longest) + + +# ----------------------------------------------------------------------------- +# Running one +# ----------------------------------------------------------------------------- + +#: Refusals of a key that is no key at all, answered 422 rather than 409. +_MALFORMED = frozenset({"invalid_name", "reserved_shape"}) + + +def _aborted() -> RenameConflict: + return RenameConflict(409, "aborted", "The rename was called off; nothing changed") + + +@dataclass +class _Held: + """What a rename holds so far, so whatever stops it lets go of exactly + that, last taken first.""" + + gate: bool = False + lock: bool = False + pipeline: Pipeline | None = None + execution: ExecutionService | None = None + queue: WorkQueue | None = None + #: Keys of the flows taken down, which start again as they were if the + #: rename does not go ahead. + torn_down: list[str] = field(default_factory=list) + + +class Renamer: + """Runs renames, one at a time, in the order section 2.3 of the protocol + gives. + + A flow's rename closes the gate, takes the controller lock and holds the + engine. Each of those waits for the work in front of it without a limit, + says what it waits for (`pending`, and a ``rename_waiting`` event) and + can be called off meanwhile, by `cancel` or by the caller going away. + From the journal's first write on it only goes forward: a step that fails + keeps everything held and is tried again in the background. A dashboard's + rename is its documents and the audit, with nothing held. + """ + + def __init__( + self, + controller: FlowController, + dashboards: DashboardStore, + gate: RenameGate, + events: EventBus, + alerts: AlertManager | None = None, + metrics: MetricsCollector | None = None, + ) -> None: + self.controller = controller + self.dashboards = dashboards + self.gate = gate + self.events = events + self.alerts = alerts + self.metrics = metrics + #: The rename under way, None when there is none. + self.pending: RenameStatus | None = None + self._abort = threading.Event() + #: A rename being finished in the background once a step failed. + self._resume: asyncio.Task[None] | None = None + + async def rename( + self, + kind: Kind, + obj_id: str, + request: RenameRequest, + actor: str = "", + disconnected: Callable[[], Awaitable[bool]] | None = None, + ) -> RenameResult: + """Rename the flow or dashboard *obj_id* as *request* asks, or answer + it with the result it got before. + + *disconnected* says when the caller has gone, which calls the rename + off as `cancel` does, for as long as that is still possible. + + :raises RenameConflict: For every answer but the result. + :raises FlowNotFound: For a flow that does not exist. + :raises DashboardNotFound: For a dashboard that does not exist. + """ + if self.pending is not None: + mine = self.pending.op == request.request_id + raise RenameConflict( + 503, + "completing" if mine else "rename_in_progress", + f"The rename of '{self.pending.old}' to '{self.pending.new}' " + "is under way", + op=self.pending.op, + ) + status = self.pending = RenameStatus( + op=request.request_id, + kind=kind, + id=obj_id, + old=request.expected_name, + new=request.new_name, + phase="draining", + ) + abort = self._abort = threading.Event() + task = asyncio.create_task(self._run(status, request, actor, abort)) + # Read here, or by nobody once the caller has gone. + task.add_done_callback(lambda done: done.cancelled() or done.exception()) + watcher = ( + asyncio.create_task(self._watch(disconnected, abort)) + if disconnected is not None + else None + ) + try: + # Shielded: a caller going away calls the rename off while it + # still can be, and never stops it halfway after. + return await asyncio.shield(task) + except asyncio.CancelledError: + abort.set() + raise + finally: + if watcher is not None: + watcher.cancel() + + def cancel(self, obj_id: str, op: str) -> None: + """Call off the rename *op* of *obj_id* while it still waits for the + engine. It answers 409 ``aborted`` once everything is as it was. + + :raises KeyError: When no such rename is under way. + :raises RenameConflict: 409 ``too_late`` once it changes things. + """ + status = self.pending + if status is None or (status.id, status.op) != (obj_id, op): + raise KeyError(op) + if status.phase != "draining": + raise RenameConflict( + 409, "too_late", "The rename has begun changing things and finishes" + ) + self._abort.set() + + @staticmethod + async def _watch( + disconnected: Callable[[], Awaitable[bool]], abort: threading.Event + ) -> None: + while not abort.is_set(): + if await disconnected(): + abort.set() + return + await asyncio.sleep(HOLD_POLL_S) + + def _report(self, busy: list[str]) -> None: + """What the rename waits for, each time that changes. From a thread.""" + status = self.pending + if status is None: + return + self.events.publish( + { + "type": "rename_waiting", + "op": status.op, + "kind": status.kind, + "id": status.id, + "busy": busy, + "ts": time.time(), + } + ) + status.busy = busy + + async def _run( + self, + status: RenameStatus, + request: RenameRequest, + actor: str, + abort: threading.Event, + ) -> RenameResult: + try: + if status.kind == "dashboard": + return await self._rename_dashboard(status, request, actor) + checked = await asyncio.to_thread(self._checked, "flow", status.id, request) + if isinstance(checked, RenameResult): + return checked + return await self._rename_flow(status, checked, request, actor, abort) + finally: + # Kept while its journal is not final: it is being finished, or + # a restart has to. + if self.pending is status and status.phase in ("draining", "rolled_back"): + self.pending = None + + def _checked( + self, kind: Kind, obj_id: str, request: RenameRequest + ) -> RenamePlan | RenameResult: + """The result a request asked again got, or the plan it goes ahead + on, once that is what the request was planned against. Blocking. + + :raises RenameConflict: 409 ``stale`` or ``plan_changed``, or the + plan's first refusal: 422 for a key that is no key, else 409. + """ + current: FlowDef | DashboardDef = ( + self.controller.store.read_flow(obj_id, draft=True) + if kind == "flow" + else self.dashboards.read(obj_id, draft=True) + ) + stored = repeated( + request.request_id, + current.name, + current.version, + request.expected_name, + request.expected_version, + ) + if stored is not None: + return stored + plan = plan_rename( + self.controller, + self.dashboards, + kind, + obj_id, + request.new_name, + self.alerts.config if self.alerts is not None else AlertsConfig(), + ) + if plan.refusals: + first = plan.refusals[0] + raise RenameConflict( + 422 if first.code in _MALFORMED else 409, + first.code, + first.detail, + plan=plan.model_dump(mode="json"), + ) + if plan.fingerprint != request.fingerprint: + raise RenameConflict( + 409, + "plan_changed", + "What the rename changes has moved since it was planned", + plan=plan.model_dump(mode="json"), + ) + return plan + + async def _rename_flow( + self, + status: RenameStatus, + plan: RenamePlan, + request: RenameRequest, + actor: str, + abort: threading.Event, + ) -> RenameResult: + controller, op = self.controller, request.request_id + rename = Rename(plan.name, plan.new_name) + held = _Held() + try: + await self._hold(held, op, plan, abort) + # Again, now that nothing can move underneath it. + plan = cast( + RenamePlan, + await asyncio.to_thread(self._checked, "flow", plan.id, request), + ) + if abort.is_set(): + raise _aborted() + # Built ones only: a flow that is only a draft runs nothing. + rebuilt = [ + doc.id + for doc in plan.documents + if doc.kind == "flow" and doc.id in controller._flow_keys + ] + for flow_id in rebuilt: + key = controller._flow_keys[flow_id] + held.torn_down.append(key) + await controller._teardown(key) + if controller.execution is not None: + await asyncio.to_thread(controller.execution.queue.hold) + held.queue = controller.execution.queue + # Whatever the old names published is recorded under them. + await self.events.settle() + if self.metrics is not None: + await self.metrics.flush_now() + rewrites = await asyncio.to_thread( + rewrite_documents, + controller.store, + self.dashboards, + "flow", + plan.id, + rename, + ) + if abort.is_set(): + raise _aborted() + # Synchronously after the last look at abort: `cancel` is refused + # from here, and the rename is finished rather than called off. + status.phase = "prepared" + try: + journal = await asyncio.to_thread( + lambda: begin( + op, + "flow", + plan.id, + rename, + rewrites, + actor=actor, + fingerprint=plan.fingerprint, + durable=not isinstance(controller.state, MemoryState), + rebuilt=rebuilt, + ) + ) + except BaseException: + status.phase = "draining" + raise + except BaseException: + await self._release(held) + raise + + try: + await asyncio.to_thread(apply_documents, journal, controller.store) + except Exception as exc: + logger.exception("Could not write the renamed documents; undoing them") + try: + await asyncio.to_thread(roll_back, journal, controller.store) + status.phase = "rolled_back" + finally: + await self._release(held) + raise RenameConflict( + 500, "failed", f"The documents could not be written: {exc}", op=op + ) from exc + status.phase = "forward" + # Never started again as they were: the splice builds them anew. + held.torn_down = [] + try: + await asyncio.to_thread(self._finish_steps, journal, status) + except Exception as exc: + self.events.publish( + { + "type": "engine_degraded", + "detail": f"A rename could not finish: {exc}", + } + ) + self._resume = asyncio.create_task( + self._finish_later(journal, held, status) + ) + raise RenameConflict( + 503, + "completing", + f"The rename could not finish yet and is being retried: {exc}", + op=op, + ) from None + return await self._complete(journal, held, status) + + async def _hold( + self, held: _Held, op: str, plan: RenamePlan, abort: threading.Event + ) -> None: + """The gate, the controller lock, the pipeline and the execution + service, each once what runs in front of it has finished. What it got + is in *held*. + + :raises RenameConflict: 409 ``aborted`` once called off, 503 ``busy`` + when a rebuild keeps the controller for longer than one may. + """ + controller = self.controller + affected = [doc.id for doc in plan.documents if doc.kind == "flow"] + held.gate = await asyncio.to_thread( + self.gate.close, op, affected, abort, self._report + ) + if not held.gate: + raise _aborted() + try: + await controller._acquire(REBUILD_WAIT) + except RebuildBusy as exc: + raise RenameConflict(503, "busy", str(exc), op=op) from None + held.lock = True + pipeline, execution = controller.pipeline, controller.execution + if pipeline is not None: + if not await asyncio.to_thread(pipeline.hold, abort, self._report): + raise _aborted() + held.pipeline = pipeline + if execution is not None: + if not await asyncio.to_thread(execution.hold, abort, self._report): + raise _aborted() + held.execution = execution + if self.pending is not None: + self.pending.busy = [] + + def _finish_steps(self, journal: Journal, status: RenameStatus) -> None: + """Every step after the documents, the audit and the journal's end. + Blocking, and tried again until it succeeds.""" + execution = self.controller.execution + try: + run_steps( + journal, + self.controller.state, + execution.queue if execution is not None else None, + ) + finish(journal) + except Exception as exc: + status.error = journal.error or f"{type(exc).__name__}: {exc}" + raise + status.error = "" + + async def _finish_later( + self, journal: Journal, held: _Held, status: RenameStatus + ) -> None: + """Decision 5: held, and tried again however long it takes.""" + await retry_until_done(lambda: self._finish_steps(journal, status), self.events) + await self._complete(journal, held, status) + + async def _complete( + self, journal: Journal, held: _Held, status: RenameStatus + ) -> RenameResult: + """The engine under the new names, then everything let go of and + everyone told. Its journal is done already: what is left here lives + in memory, which a restart builds afresh anyway.""" + rename = journal.rename + if held.execution is not None: + held.execution.rename_reclaimed(rename) + self.events.rename(rename) + if self.alerts is not None: + self.alerts.rename(rename) + try: + await self.controller.splice_renamed(rename, journal.affected.rebuilt) + except Exception as exc: + # What a failed `reload_flow` leaves: those flows stopped, until + # they are published again. + logger.exception("Could not rebuild the renamed flows") + self.events.publish( + { + "type": "engine_degraded", + "detail": f"The renamed flows could not be rebuilt: {exc}", + } + ) + await self._release(held) + if self.pending is status: + self.pending = None + self._announce(journal) + return cast(RenameResult, journal.result) + + async def _rename_dashboard( + self, status: RenameStatus, request: RenameRequest, actor: str + ) -> RenameResult: + done = await asyncio.to_thread( + self._dashboard_documents, status, request, actor + ) + if isinstance(done, RenameResult): + return done + await retry_until_done(lambda: self._finish_steps(done, status), self.events) + if self.pending is status: + self.pending = None + self._announce(done) + return cast(RenameResult, done.result) + + def _dashboard_documents( + self, status: RenameStatus, request: RenameRequest, actor: str + ) -> Journal | RenameResult: + """A dashboard's documents, renamed and committed. Blocking. + + Under the dashboard store's lock, which every dashboard save takes + before the flow store's, so none lands between the recheck and the + write. Nothing else names a dashboard's key, so nothing is held. + """ + store = self.controller.store + with self.dashboards._lock: + checked = self._checked("dashboard", status.id, request) + if isinstance(checked, RenameResult): + return checked + rename = Rename(checked.name, checked.new_name) + rewrites = rewrite_documents( + store, self.dashboards, "dashboard", status.id, rename + ) + journal = begin( + request.request_id, + "dashboard", + status.id, + rename, + rewrites, + actor=actor, + fingerprint=checked.fingerprint, + ) + status.phase = "prepared" + try: + apply_documents(journal, store) + except Exception as exc: + roll_back(journal, store) + status.phase = "rolled_back" + raise RenameConflict( + 500, + "failed", + f"The documents could not be written: {exc}", + op=request.request_id, + ) from exc + status.phase = "forward" + return journal + + async def _release(self, held: _Held) -> None: + """Let go of what *held* says, last taken first. Flows taken down for + a rename that does not go ahead start again as they were.""" + if held.queue is not None: + try: + await asyncio.to_thread(held.queue.release) + except Exception: + # Left in the held list, which the next start releases. + logger.exception("Could not release the work held for a rename") + for key in held.torn_down: + await self.controller._activate(key) + if held.execution is not None: + held.execution.release() + if held.pipeline is not None: + held.pipeline.release() + if held.lock: + self.controller._lock.release() + if held.gate: + self.gate.open() + + def _announce(self, journal: Journal) -> None: + """Tell the clients, once everything is let go of: editors and panels + refetch, and the canvas takes the pipeline as it is now.""" + now = time.time() + self.events.publish( + { + "type": "renamed", + "kind": journal.kind, + "id": journal.id, + "old": journal.old, + "new": journal.new, + "op": journal.op, + "ts": now, + } + ) + for dash_id in journal.affected.dashboards: + self.events.publish( + {"type": "dashboard_changed", "dashboard": dash_id, "ts": now} + ) + if journal.kind != "flow": + return + # What the devices' ports are wired to moved with the flow. + self.events.publish({"type": "dashboard_changed", "dashboard": "", "ts": now}) + controller = self.controller + self.events.publish( + { + "type": "pipeline_rebuilt", + "issues": [issue.model_dump() for issue in controller.issues], + "nodes": [status.model_dump() for status in controller.node_statuses()], + "paused": controller.paused_flows(), + } + ) diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 1675883..bbf1012 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -50,7 +50,9 @@ from fluksio.flow.provision import load_provisioners from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue from fluksio.flow.remote import RemoteWorkerHub from fluksio.flow.rename import ( + RenameConflict, RenameGate, + Renamer, recover_documents, recover_runtime, retry_until_done, @@ -464,6 +466,14 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # On the app so a superuser can ask it to flush now. app.state.metrics_collector = MetricsCollector(event_bus) _background(app.state.metrics_collector.run(), "metrics-collector") + app.state.renamer = Renamer( + controller, + dashboards, + rename_gate, + event_bus, + alerts=alerts, + metrics=app.state.metrics_collector, + ) _background(_sweep_artifacts(artifacts, controller), "artifact-gc") if volatile is not None: _background(_trim_volatile(volatile), "artifact-ring") @@ -688,6 +698,18 @@ async def engine_held(request: Request, exc: EngineHeld) -> JSONResponse: # noq ) +@app.exception_handler(RenameConflict) +async def rename_conflict(request: Request, exc: RenameConflict) -> JSONResponse: # noqa: ARG001 + """A rename answered with a code a client acts on, as the gate's 503 is: + a refusal or a stale request (409), a key that is no key (422), or one + finishing (503, which is worth retrying).""" + return JSONResponse( + status_code=exc.status, + content={"detail": exc.detail, "code": exc.code, **exc.extra}, + headers={"Retry-After": "5"} if exc.status == 503 else None, + ) + + # Tagged because the operation-id builder reads the first tag; the route # itself stays out of the schema. @app.get( diff --git a/backend/tests/api/routes/test_rename.py b/backend/tests/api/routes/test_rename.py new file mode 100644 index 0000000..4ac5ea4 --- /dev/null +++ b/backend/tests/api/routes/test_rename.py @@ -0,0 +1,422 @@ +"""Renaming a flow's or a dashboard's key over HTTP (sections 5.2 and 5.3 of +docs/private/id_refac_rename.md): who may, what a request is answered with, +what waits for it, and what keeps working by id afterwards.""" + +import threading +import time +import uuid +from pathlib import Path +from typing import Any + +import httpx +import pytest +from fastapi.testclient import TestClient + +from fluksio.api.routes import oauth +from fluksio.core.config import settings +from fluksio.flow import rename as renaming +from fluksio.main import app + +API = settings.API_V1_STR +FLOWS = f"{API}/flows" +DASHBOARDS = f"{API}/dashboards" +DEVICES = f"{API}/devices" +MESSAGES = f"{API}/messages" + +TWICE = "def process(tick):\n return {'twice': tick * 2}\n" + + +@pytest.fixture(autouse=True) +def _journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A journal of the test's own, and pairing unlimited, as in the device + tests.""" + monkeypatch.setattr(settings, "RENAME_JOURNAL_FILE", tmp_path / "rename.json") + for bucket in [b for b in oauth._hits if b.startswith("pair:")]: + del oauth._hits[bucket] + + +def _key() -> str: + return f"rn_{uuid.uuid4().hex[:8]}" + + +def _flow( + client: TestClient, + headers: dict[str, str], + key: str, + codes: dict[str, str] | None = None, + **document: Any, +) -> str: + """A flow under an issued id, its python nodes' *codes* written, and + published; its id.""" + made = client.post(f"{FLOWS}/", headers=headers, json={"name": key, **document}) + assert made.status_code == 201, made.text + definition = made.json()["definition"] + for node_id, code in (codes or {}).items(): + saved = client.put( + f"{FLOWS}/{definition['id']}/nodes/{node_id}/source", + headers=headers, + json={"code": code}, + ) + assert saved.status_code == 200, saved.text + version = client.get(f"{FLOWS}/{definition['id']}", headers=headers).json()[ + "definition" + ]["version"] + published = client.post( + f"{FLOWS}/{definition['id']}/publish", + headers=headers, + json={"version": version}, + ) + assert published.status_code == 200, published.text + return str(definition["id"]) + + +def _hooked(code_node: dict | None = None) -> dict[str, Any]: + """A webhook putting ``tick`` in, and what reads it.""" + nodes: list[dict[str, Any]] = [ + { + "id": "hook", + "type": "http", + "params": {"url": "/tick", "method": "POST"}, + "provides": [{"name": "tick", "dtype": "float"}], + } + ] + if code_node is not None: + nodes.append(code_node) + return {"nodes": nodes} + + +def _request( + client: TestClient, + headers: dict[str, str], + obj_id: str, + new: str, + kind: str = FLOWS, + **override: Any, +) -> dict[str, Any]: + plan = client.post( + f"{kind}/{obj_id}/rename/plan", headers=headers, json={"new_name": new} + ) + assert plan.status_code == 200, plan.text + body = plan.json() + return { + "request_id": uuid.uuid4().hex, + "expected_name": body["name"], + "expected_version": body["version"], + "new_name": new, + "fingerprint": body["fingerprint"], + **override, + } + + +def _in_thread(call) -> tuple[threading.Thread, list[httpx.Response]]: + answered: list[httpx.Response] = [] + thread = threading.Thread(target=lambda: answered.append(call()), daemon=True) + thread.start() + return thread, answered + + +def _until(check, timeout: float = 20.0) -> None: + deadline = time.monotonic() + timeout + while not check(): + assert time.monotonic() < deadline, "timed out" + time.sleep(0.05) + + +def test_a_rename_is_a_superusers_and_answered_once( + client: TestClient, + superuser_token_headers: dict[str, str], + normal_user_token_headers: dict[str, str], +) -> None: + """Every refusal by its code, then the rename, the same answer for the + same request, and its webhook where it was, by id.""" + su = superuser_token_headers + key, new = _key(), _key() + flow_id = _flow(client, su, key, **_hooked()) + try: + body = _request(client, su, flow_id, new) + url = f"{FLOWS}/{flow_id}/rename" + assert ( + client.post(url, headers=normal_user_token_headers, json=body).status_code + == 403 + ) + assert ( + client.post(f"{FLOWS}/f_nothere0/rename", headers=su, json=body).status_code + == 404 + ) + for override, status, code in ( + ({"new_name": "Not A Key"}, 422, "invalid_name"), + ({"expected_version": body["expected_version"] - 1}, 409, "stale"), + ({"fingerprint": "sha256:moved"}, 409, "plan_changed"), + ): + refused = client.post(url, headers=su, json={**body, **override}) + assert (refused.status_code, refused.json()["code"]) == (status, code) + assert "refusals" in refused.json()["plan"] + + renamed = client.post(url, headers=su, json=body) + assert renamed.status_code == 200, renamed.text + result = renamed.json() + assert ( + result["kind"], + result["id"], + result["old_name"], + result["new_name"], + ) == ( + "flow", + flow_id, + key, + new, + ) + assert result["op"] == body["request_id"] + assert client.post(url, headers=su, json=body).json() == result + again = client.post(url, headers=su, json={**body, "request_id": "other"}) + assert (again.status_code, again.json()["code"]) == (409, "stale") + + detail = client.get(f"{FLOWS}/{flow_id}", headers=su).json() + assert detail["definition"]["name"] == new + # The webhook is under the id, which the rename left alone. + hooked = client.post(f"/hooks/{flow_id}/tick", json={"tick": 7.0}) + assert hooked.status_code == 200, hooked.text + + def values() -> dict[str, Any]: + found = client.get(f"{FLOWS}/{flow_id}/state", headers=su).json() + return dict(found["values"]) + + # Once the cascade it queued has run. + _until(lambda: f"{new}.tick" in values()) + assert values()[f"{new}.tick"]["value"] == 7.0 + assert f"{key}.tick" not in values() + summary = client.get(f"{API}/observability/summary", headers=su).json() + assert summary["rename"] is None + finally: + client.delete(f"{FLOWS}/{flow_id}", headers=su) + + +def test_a_devices_credential_follows_the_renamed_flow( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A message it may send moves with the flow, and renaming its dashboard + leaves the device's list as it was.""" + su = superuser_token_headers + key, new, board, device = _key(), _key(), _key(), _key() + flow_id = _flow( + client, su, key, inputs=[{"spec": {"name": "button", "dtype": "float"}}] + ) + made = client.post(f"{DASHBOARDS}/", headers=su, json={"name": board}).json() + dash_id = made["id"] + made["widgets"] = [ + {"id": "s", "type": "slider", "config": {"target": f"{key}.button"}} + ] + saved = client.put(f"{DASHBOARDS}/{dash_id}", headers=su, json=made) + assert saved.status_code == 200, saved.text + client.post( + f"{DASHBOARDS}/{dash_id}/publish", + headers=su, + json={"version": saved.json()["version"]}, + ) + created = client.post(f"{DEVICES}/{device}", headers=su).json() + client.put( + f"{DEVICES}/{device}", + headers=su, + json={**created, "dashboards": [dash_id]}, + ) + started = client.post(f"{DEVICES}/pair").json() + client.post(f"{DEVICES}/{device}/pair", headers=su, json={"code": started["code"]}) + token = client.get( + f"{DEVICES}/pair/{started['code']}", params={"secret": started["secret"]} + ).json()["access_token"] + panel = {"Authorization": f"Bearer {token}"} + + def send(name: str) -> int: + return client.post( + f"{MESSAGES}/{name}", headers=panel, json={"value": 1.0} + ).status_code + + try: + assert send(f"{key}.button") == 200 + renamed = client.post( + f"{FLOWS}/{flow_id}/rename", + headers=su, + json=_request(client, su, flow_id, new), + ) + assert renamed.status_code == 200, renamed.text + assert renamed.json()["documents"][-1]["id"] == dash_id + assert send(f"{key}.button") == 403 + assert send(f"{new}.button") == 200 + + renamed = client.post( + f"{DASHBOARDS}/{dash_id}/rename", + headers=su, + json=_request(client, su, dash_id, f"{board}_b", kind=DASHBOARDS), + ) + assert renamed.status_code == 200, renamed.text + assert client.get(f"{DASHBOARDS}/{dash_id}", headers=su).json()["name"] == ( + f"{board}_b" + ) + listed = client.get(f"{DEVICES}/{device}", headers=su).json() + assert listed["dashboards"] == [dash_id] + assert send(f"{new}.button") == 200 + finally: + client.delete(f"{DEVICES}/{device}", headers=su) + client.delete(f"{DASHBOARDS}/{dash_id}", headers=su) + client.delete(f"{FLOWS}/{flow_id}", headers=su) + + +def test_what_arrives_during_a_rename_waits_or_is_turned_away( + client: TestClient, + superuser_token_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Saves, publishes, creates, deletes, runs, imports, sync and messages + get 503 while the rename holds the engine, its own flow's webhook too. + Another flow's webhook is taken, and what it sets off runs once the + rename lets go. Reads carry on.""" + su = superuser_token_headers + key, new, other_key = _key(), _key(), _key() + flow_id = _flow(client, su, key, **_hooked()) + other = _flow( + client, + su, + other_key, + {"double": TWICE}, + **_hooked( + { + "id": "double", + "type": "python", + "requires": [{"name": "tick", "dtype": "float"}], + "provides": [{"name": "twice", "dtype": "float"}], + } + ), + ) + stale = client.get(f"{FLOWS}/{flow_id}", headers=su).json()["definition"] + holding, go = threading.Event(), threading.Event() + begin = renaming.begin + + def held_begin(*args: Any, **kwargs: Any) -> Any: + holding.set() + assert go.wait(20) + return begin(*args, **kwargs) + + monkeypatch.setattr(renaming, "begin", held_begin) + body = _request(client, su, flow_id, new) + thread, answered = _in_thread( + lambda: client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body) + ) + try: + assert holding.wait(20) + for method, path, payload in ( + ("put", f"{FLOWS}/{flow_id}", stale), + ("post", f"{FLOWS}/{flow_id}/publish", {"version": stale["version"]}), + ("post", f"{FLOWS}/", {"name": _key()}), + ("delete", f"{FLOWS}/{other}", None), + ("post", f"{API}/runs/flows/{other}", {"inputs": {}}), + ("post", f"{API}/runs/import", []), + ("post", f"{MESSAGES}/{other_key}.tick", {"value": 1.0}), + ): + refused = client.request(method, path, headers=su, json=payload) + assert refused.status_code == 503, (path, refused.text) + assert refused.json()["code"] == "renaming" + assert refused.headers["retry-after"] == "5" + assert ( + client.post(f"/hooks/{flow_id}/tick", json={"tick": 1.0}).status_code == 503 + ) + assert ( + client.post(f"/hooks/{other}/tick", json={"tick": 3.0}).status_code == 200 + ) + assert client.get(f"{FLOWS}/{flow_id}", headers=su).status_code == 200 + summary = client.get(f"{API}/observability/summary", headers=su).json() + assert summary["rename"]["phase"] == "prepared" + + def doubled() -> Any: + values = client.get(f"{FLOWS}/{other}/state", headers=su).json()["values"] + return values.get(f"{other_key}.twice", {}).get("value") + + time.sleep(0.3) + assert doubled() is None + finally: + go.set() + thread.join(30) + assert answered[0].status_code == 200, answered[0].text + try: + _until(lambda: doubled() == 6.0) + # An editor still holding the old document is refused, not saved back. + assert ( + client.put(f"{FLOWS}/{flow_id}", headers=su, json=stale).status_code == 409 + ) + finally: + client.delete(f"{FLOWS}/{other}", headers=su) + client.delete(f"{FLOWS}/{flow_id}", headers=su) + + +def test_a_rename_waits_for_running_work_and_can_be_called_off( + client: TestClient, superuser_token_headers: dict[str, str], tmp_path: Path +) -> None: + """It says what it waits for, and called off it changes nothing.""" + su = superuser_token_headers + key, new = _key(), _key() + go = tmp_path / "go" + slow = ( + "import os, time\n" + "def process():\n" + f" while not os.path.exists({str(go)!r}):\n" + " time.sleep(0.05)\n" + " return {'reading': 1.0}\n" + ) + flow_id = _flow( + client, + su, + key, + {"sensor": slow}, + nodes=[ + { + "id": "sensor", + "type": "python", + "provides": [{"name": "reading", "dtype": "float"}], + } + ], + ) + trigger = f"{FLOWS}/{flow_id}/nodes/sensor/trigger" + triggering, triggered = _in_thread( + lambda: client.post(trigger, headers=su, json={"values": {}}) + ) + body = _request(client, su, flow_id, new) + try: + # In, and counted, before the gate closes. + _until(lambda: f"POST {trigger}" in app.state.rename_gate._admitted) + thread, answered = _in_thread( + lambda: client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body) + ) + + def waiting() -> Any: + summary = client.get(f"{API}/observability/summary", headers=su).json() + return summary["rename"] + + _until( + lambda: waiting() is not None and waiting()["busy"] == [f"POST {trigger}"] + ) + assert waiting()["phase"] == "draining" + assert client.put(f"{FLOWS}/{flow_id}", headers=su, json={}).status_code == 503 + + cancel = f"{FLOWS}/{flow_id}/rename/{body['request_id']}" + assert ( + client.delete(f"{FLOWS}/{flow_id}/rename/other", headers=su).status_code + == 404 + ) + assert client.delete(cancel, headers=su).status_code == 200 + thread.join(30) + assert (answered[0].status_code, answered[0].json()["code"]) == (409, "aborted") + assert waiting() is None + finally: + go.touch() + triggering.join(60) + assert triggered[0].status_code == 200, triggered[0].text + try: + definition = client.get(f"{FLOWS}/{flow_id}", headers=su).json()["definition"] + assert (definition["name"], definition["version"]) == ( + key, + body["expected_version"], + ) + # The gate is open again, and the same rename now goes through. + renamed = client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body) + assert renamed.status_code == 200, renamed.text + finally: + client.delete(f"{FLOWS}/{flow_id}", headers=su) diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index e0e4861..33783f0 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -1,5 +1,6 @@ """The command line, and the one ordering it depends on.""" +import shutil import subprocess import sys from pathlib import Path @@ -842,12 +843,15 @@ def test_serve_says_when_a_flow_wants_a_card_nobody_declared( ) ) monkeypatch.setattr(settings, "FLOWS_DIR", tmp_path / "flows") + # It only reads: the engine has not settled the store yet. + shutil.rmtree(tmp_path / "flows" / ".git") monkeypatch.setattr(settings, "FLOW_GPUS", 0) cli._mention_undeclared_cards() said = capsys.readouterr().out assert "finetune asks for one" in said assert "--gpus" in said + assert not (tmp_path / "flows" / ".git").exists() # Told how many there are, it has nothing to say. monkeypatch.setattr(settings, "FLOW_GPUS", 1) diff --git a/backend/tests/test_rename_run.py b/backend/tests/test_rename_run.py new file mode 100644 index 0000000..d760043 --- /dev/null +++ b/backend/tests/test_rename_run.py @@ -0,0 +1,577 @@ +"""Running a key rename against a live engine (sections 2.3 and 2.4 of +docs/private/id_refac_rename.md): the barrier, calling it off, the splice and +the runtime controls it keeps. Memory always, the end-to-end queue check with +FLUKSIO_TEST_REDIS_HOST (and optional PORT).""" + +import asyncio +import json +import os +import threading +import time +import uuid +from collections.abc import Iterator +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest +from sqlmodel import Session, col, delete, select + +from fluksio.core.config import settings +from fluksio.core.db import engine +from fluksio.flow import executor +from fluksio.flow import rename as renaming +from fluksio.flow.alerts import AlertManager, AlertsConfig, Channel +from fluksio.flow.controller import NODE_TYPES, FlowController, NodeType +from fluksio.flow.dashboards import DashboardDef, DashboardStore, WidgetDef +from fluksio.flow.events import EventBus +from fluksio.flow.executor import ExecutionService +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.nodes import Node +from fluksio.flow.pipeline import EngineHeld +from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkItem +from fluksio.flow.rename import ( + RenameConflict, + RenameGate, + Renamer, + RenameRequest, + plan_rename, + read_journal, +) +from fluksio.flow.schemas import FlowDef, FlowOrigin, NodeDef +from fluksio.flow.state import MemoryState, RedisState +from fluksio.flow.store import FlowStore +from fluksio.models import EngineEvent +from tests.utils.rename import leftovers, stability + +#: What the probes did, by node id, in order. +EFFECTS: list[str] = [] +#: A probe set to wait says so here, then waits for GO. +RUNNING = threading.Event() +GO = threading.Event() + + +class Probe(Node): + """A node reaching outside, so doing it twice would be a second command. + What it did is written down, and with ``wait`` it takes as long as the + test says.""" + + idempotent = False + + def __init__(self, requires=(), provides=(), params=None, name=None) -> None: + super().__init__( + f=self._act, requires=requires, provides=provides, params=params, name=name + ) + + def _act(self, params: dict[str, Any], **inputs: Any) -> None: + if params.get("wait"): + RUNNING.set() + assert GO.wait(10) + EFFECTS.append(self.id) + + +def _probe(node_id: str, requires=(), provides=(), **params: Any) -> NodeDef: + def specs(names) -> list[MessageSpec]: + return [MessageSpec(name=name, dtype=DType.FLOAT) for name in names] + + return NodeDef( + id=node_id, + type="probe", + requires=specs(requires), + provides=specs(provides), + params=params, + ) + + +@pytest.fixture +def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[SimpleNamespace]: + """`heating`, a sensor and a slow actor on it; `lights` reading it; + `heating_extra`, which only looks alike; `home`, a dashboard charting + `heating.temp`. Ids are keys, as for a flow from before ids.""" + monkeypatch.setattr(settings, "RENAME_JOURNAL_FILE", tmp_path / "rename.json") + monkeypatch.setattr(settings, "ALERTS_FILE", tmp_path / "alerts.json") + monkeypatch.setitem( + NODE_TYPES, "probe", NodeType(title="Probe", description="", cls=Probe) + ) + EFFECTS.clear() + RUNNING.clear() + GO.clear() + store = FlowStore(tmp_path / "flows") + store.write_flow( + FlowDef( + name="heating", + nodes=[ + _probe("sensor", provides=["temp"]), + _probe("act", requires=["temp"], wait=True), + ], + ) + ) + store.write_flow( + FlowDef(name="lights", nodes=[_probe("lamp", requires=["heating.temp"])]) + ) + store.write_flow(FlowDef(name="heating_extra", nodes=[_probe("x", provides=["x"])])) + dashboards = DashboardStore(store) + dashboards.write( + DashboardDef( + name="home", + widgets=[ + WidgetDef( + id="c", + type="chart", + config={ + "series": [{"message": "heating.temp"}], + "history": {"points": 500}, + }, + ) + ], + ) + ) + yield SimpleNamespace(root=store.root, store=store, dashboards=dashboards) + # Let a probe still waiting go, so its thread ends with the test. + GO.set() + with Session(engine) as session: + session.execute(delete(EngineEvent)) + session.commit() + + +async def _engine(world, state=None, queue=None, consume: bool = True): + """The engine over *world*'s store, built and, with *consume*, claiming.""" + events = EventBus() + events.bind(asyncio.get_running_loop()) + queue = queue if queue is not None else MemoryWorkQueue() + execution = ExecutionService(queue, max_cascades=2) + alerts = AlertManager( + events, + config=AlertsConfig( + channels=[ + Channel( + name="board", kind="dashboard", config={"message": "heating.alert"} + ) + ] + ), + ) + controller = FlowController( + world.store, + state=state if state is not None else MemoryState(), + events=events, + execution=execution, + alerts=alerts, + ) + controller.dashboards = world.dashboards + controller.set_history_limits(world.dashboards.history_requirements()) + if consume: + await controller.start() + else: + await controller.reload(wait=None) + renamer = Renamer(controller, world.dashboards, RenameGate(), events, alerts=alerts) + # What the build said is not for whoever listens next. + await events.settle() + return SimpleNamespace( + controller=controller, + renamer=renamer, + events=events, + queue=queue, + alerts=alerts, + state=controller.state, + ) + + +def _request(eng, new: str = "warmth", obj: str = "heating", op: str = "op-1"): + kind = "flow" if obj in eng.controller.store.list_flows() else "dashboard" + plan = plan_rename( + eng.controller, eng.renamer.dashboards, kind, obj, new, AlertsConfig() + ) + return RenameRequest( + request_id=op, + expected_name=plan.name, + expected_version=plan.version, + new_name=new, + fingerprint=plan.fingerprint, + ) + + +async def _until(check, timeout: float = 10.0) -> None: + deadline = time.monotonic() + timeout + while not check(): + assert time.monotonic() < deadline, "timed out" + await asyncio.sleep(0.02) + + +async def _heard(events: EventBus, queue: asyncio.Queue) -> list[dict]: + await events.settle() + found = [] + while not queue.empty(): + found.append(queue.get_nowait()) + return found + + +def _files(root: Path) -> dict[str, bytes]: + return { + str(path.relative_to(root)): path.read_bytes() + for path in sorted(root.rglob("*")) + if path.is_file() and ".git" not in path.parts + } + + +def test_a_rename_moves_the_engine_and_keeps_its_controls(world): + """Stopped and paused are kept everywhere, quarantine only off the flows + it rebuilds (decision 6), and everything the engine keeps by name + follows.""" + world.store.write_enabled("lights", False) + + async def scenario() -> None: + eng = await _engine(world) + controller, pipeline = eng.controller, eng.controller.pipeline + try: + eng.state.set("heating.temp", 20.5) + controller.pause_flow("heating") + controller.pause_flow("heating_extra") + eng.queue.park("heating", WorkItem("cascade", "heating.sensor", "heating")) + for key in ("heating.sensor:loop", "heating_extra.x:loop"): + controller.supervisor._flows[key] = key.split(".")[0] + controller.supervisor._resting[key] = key.split(".")[0] + controller.loaded["heating.act"].last_error = "boom" + eng.events.emits["heating.act"] = 3 + eng.events.digests["heating.cam"] = ["d1"] + eng.alerts._last_sent["node_error:heating.act"] = 7.0 + request = _request(eng) + + async with eng.events.subscribe() as heard: + result = await eng.renamer.rename("flow", "heating", request, "me") + told = await _heard(eng.events, heard) + + assert (result.old_name, result.new_name) == ("heating", "warmth") + assert sorted(controller.loaded) == [ + "heating_extra.x", + "lights.lamp", + "warmth.act", + "warmth.sensor", + ] + assert controller.loaded["warmth.act"].last_error == "boom" + assert controller.disabled == {"lights"} + assert controller.paused_flows() == ["heating_extra", "warmth"] + assert controller.quarantined == {"heating_extra"} + assert controller.pipeline is pipeline + lamp = pipeline.get_node_by_id("lights.lamp") + assert [spec.name for spec in lamp.input_ports] == ["warmth.temp"] + assert eng.state.get("warmth.temp") == 20.5 + assert "heating.temp" not in eng.state + assert controller.history_limits == {"warmth.temp": 500} + assert [item.node for item in eng.queue.unpark("warmth")] == [ + "warmth.sensor" + ] + assert eng.events.emits == {"warmth.act": 3} + assert eng.events.digests == {"warmth.cam": ["d1"]} + assert eng.alerts._last_sent == {"node_error:warmth.act": 7.0} + assert eng.alerts.config.channels[0].config == {"message": "warmth.alert"} + + renamed = [e for e in told if e["type"] == "renamed"] + assert [ + {k: e[k] for k in ("kind", "id", "old", "new", "op")} for e in renamed + ] == [ + { + "kind": "flow", + "id": "heating", + "old": "heating", + "new": "warmth", + "op": "op-1", + } + ] + changed = [e["dashboard"] for e in told if e["type"] == "dashboard_changed"] + assert changed == ["home", ""] + (rebuilt,) = [e for e in told if e["type"] == "pipeline_rebuilt"] + assert rebuilt["paused"] == ["heating_extra", "warmth"] + + assert not eng.renamer.gate.closed and eng.renamer.pending is None + assert not controller._lock.locked() + assert read_journal().phase == "done" + # Nothing is held: a value goes in again. + controller.publish_message("heating_extra.x", 1.0) + finally: + await eng.controller.stop() + + asyncio.run(scenario()) + + +@pytest.mark.parametrize("how", ["cancel", "disconnect"]) +def test_a_rename_called_off_while_draining_changes_nothing(world, how): + """It waits for the cascade running as long as that takes, says so, and + once called off leaves everything as it was, the engine running.""" + files = _files(world.root) + gone = threading.Event() + + async def disconnected() -> bool: + return gone.is_set() + + async def scenario() -> None: + eng = await _engine(world) + controller = eng.controller + try: + controller.get_node("heating.sensor").inject({"temp": 21.0}) + assert await asyncio.to_thread(RUNNING.wait, 5) + async with eng.events.subscribe() as heard: + renaming_ = asyncio.create_task( + eng.renamer.rename( + "flow", "heating", _request(eng), "me", disconnected + ) + ) + await _until( + lambda: eng.renamer.pending is not None + and eng.renamer.pending.busy == ["heating.sensor"] + ) + waiting = [ + e["busy"] + for e in await _heard(eng.events, heard) + if e["type"] == "rename_waiting" + ] + assert waiting[-1] == ["heating.sensor"] + assert eng.renamer.gate.closed + with pytest.raises(EngineHeld): + controller.publish_message("heating_extra.x", 1.0) + + if how == "cancel": + with pytest.raises(KeyError): + eng.renamer.cancel("heating", "op-2") + eng.renamer.cancel("heating", "op-1") + else: + gone.set() + with pytest.raises(RenameConflict) as refused: + await renaming_ + assert (refused.value.status, refused.value.code) == (409, "aborted") + + assert _files(world.root) == files + assert read_journal() is None + assert not eng.renamer.gate.closed and eng.renamer.pending is None + assert not controller._lock.locked() + assert sorted(controller.loaded) == [ + "heating.act", + "heating.sensor", + "heating_extra.x", + "lights.lamp", + ] + # The cascade it waited for finishes, and the next one runs. + GO.set() + await _until(lambda: sorted(EFFECTS) == ["heating.act", "lights.lamp"]) + controller.get_node("heating.sensor").inject({"temp": 22.0}) + await _until(lambda: len(EFFECTS) == 4) + finally: + GO.set() + await controller.stop() + + asyncio.run(scenario()) + + +def test_a_plan_that_moved_while_draining_is_refused_under_the_barrier(world): + """A code-defined dependent that appeared while the rename waited.""" + + async def scenario() -> None: + eng = await _engine(world) + try: + eng.controller.get_node("heating.sensor").inject({"temp": 21.0}) + assert await asyncio.to_thread(RUNNING.wait, 5) + renaming_ = asyncio.create_task( + eng.renamer.rename("flow", "heating", _request(eng), "me") + ) + await _until(lambda: eng.renamer.gate.closed) + lights = world.store.read_flow("lights") + world.store.write_flow(lights.model_copy(update={"origin": FlowOrigin()})) + GO.set() + with pytest.raises(RenameConflict) as refused: + await renaming_ + assert (refused.value.status, refused.value.code) == ( + 409, + "code_defined_dependent", + ) + assert refused.value.extra["plan"]["refusals"][0]["ids"] == ["lights"] + assert world.store.read_flow("heating").name == "heating" + assert not eng.renamer.gate.closed + assert not eng.controller._lock.locked() + finally: + await eng.controller.stop() + + asyncio.run(scenario()) + + +def test_a_failing_step_is_retried_until_the_rename_finishes(world, monkeypatch): + """Decision 5: held, said, and finished in the background.""" + monkeypatch.setattr(renaming, "RETRY_S", (0.05, 0.1)) + failures = [] + + def fault(point: str) -> None: + if point == "state" and len(failures) < 2: + failures.append(point) + raise ConnectionError("state unreachable") + + monkeypatch.setattr(renaming, "fault", fault) + + async def scenario() -> None: + eng = await _engine(world) + try: + request = _request(eng) + async with eng.events.subscribe() as heard: + with pytest.raises(RenameConflict) as busy: + await eng.renamer.rename("flow", "heating", request, "me") + assert (busy.value.status, busy.value.code) == (503, "completing") + pending = eng.renamer.pending + assert pending.phase == "forward" + assert "state unreachable" in pending.error + assert eng.renamer.gate.closed + for op, code in ( + ("op-1", "completing"), + ("op-2", "rename_in_progress"), + ): + other = request.model_copy(update={"request_id": op}) + with pytest.raises(RenameConflict) as again: + await eng.renamer.rename("flow", "heating", other, "me") + assert (again.value.status, again.value.code) == (503, code) + + await asyncio.wait_for(eng.renamer._resume, 10) + told = [e["type"] for e in await _heard(eng.events, heard)] + assert told.count("engine_degraded") == 2 + assert "renamed" in told + assert eng.renamer.pending is None and not eng.renamer.gate.closed + assert read_journal().phase == "done" + assert sorted(eng.controller.loaded)[-2:] == ["warmth.act", "warmth.sensor"] + # Asked again, the same answer. + again = await eng.renamer.rename("flow", "heating", request, "me") + assert again == read_journal().result + finally: + await eng.controller.stop() + + asyncio.run(scenario()) + + +def test_renaming_there_and_back_restores_the_namespace(world): + """With the producers stopped, `heating` to `warmth` and back leaves + documents, state and queued work as they were, but for versions.""" + + def documents() -> dict[str, Any]: + found = {} + for path in sorted(world.root.rglob("*.json")): + if ".git" not in path.parts: + data = json.loads(path.read_text()) + if isinstance(data, dict): + data.pop("version", None) + found[str(path.relative_to(world.root))] = data + return found + + async def scenario() -> None: + eng = await _engine(world, consume=False) + try: + eng.state.set("heating.temp", 20.5) + eng.state.set("__node__:heating.act:last:x", 1) + eng.queue.add( + WorkItem("cascade", "heating.sensor", "heating", {"heating.temp": 1}) + ) + eng.queue.add_delayed(WorkItem("flush", "lights.lamp", "lights"), 9e9) + before = documents(), stability(world.root) + state = {key: eng.state.get(key) for key in eng.state.keys()} + queued = [(i.node, i.outputs) for i in eng.queue.live_items()] + + await eng.renamer.rename("flow", "heating", _request(eng), "me") + assert ( + leftovers( + "heating", world.root, ["heating", "lights"], eng.state, eng.queue + ) + == [] + ) + await eng.renamer.rename( + "flow", "heating", _request(eng, "heating", op="op-2"), "me" + ) + + assert (documents(), stability(world.root)) == before + assert {key: eng.state.get(key) for key in eng.state.keys()} == state + assert [(i.node, i.outputs) for i in eng.queue.live_items()] == queued + assert sorted(eng.controller.loaded)[0] == "heating.act" + finally: + await eng.controller.stop() + + asyncio.run(scenario()) + + +def test_a_dashboard_rename_is_its_documents_and_the_audit_alone(world): + async def scenario() -> None: + eng = await _engine(world) + try: + closings = [] + eng.renamer.gate.close = lambda *args: closings.append(args) + request = _request(eng, "house", obj="home") + async with eng.events.subscribe() as heard: + result = await eng.renamer.rename("dashboard", "home", request, "me") + told = await _heard(eng.events, heard) + + assert [(d.kind, d.id) for d in result.documents] == [("dashboard", "home")] + assert world.dashboards.read("home").name == "house" + assert closings == [] + assert [(e["type"], e.get("dashboard", e.get("new"))) for e in told] == [ + ("renamed", "house"), + ("dashboard_changed", "home"), + ] + with Session(engine) as session: + audit = session.exec( + select(EngineEvent.flow_id, EngineEvent.detail).where( + col(EngineEvent.type) == "audit" + ) + ).all() + assert audit == [("", "renamed dashboard 'home' to 'house' [op op-1]")] + again = await eng.renamer.rename("dashboard", "home", request, "me") + assert again == result + finally: + await eng.controller.stop() + + asyncio.run(scenario()) + + +def test_a_pending_side_effect_runs_exactly_once_across_a_redis_rename( + world, monkeypatch +): + """Claimed by an engine that went away: one item whose side effect was + recorded as done, which must not run again, and one whose was not, which + runs once, under the new name.""" + host = os.environ.get("FLUKSIO_TEST_REDIS_HOST") + if not host: + pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks") + port = int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379")) + monkeypatch.setattr(executor, "RECLAIM_IDLE_MS", 50) + monkeypatch.setattr(executor, "RECLAIM_INTERVAL_S", 0.05) + namespace = f"test-rename-run-{uuid.uuid4().hex}" + state = RedisState(host, port, namespace=namespace) + queue = RedisWorkQueue( + host, port, namespace=f"{namespace}-queue", consumer="engine-a" + ) + # The act waits for nobody here. + heating = world.store.read_flow("heating") + heating.nodes[1].params = {} + world.store.write_flow(heating) + + async def scenario() -> None: + eng = await _engine(world, state=state, queue=queue, consume=False) + try: + for value in (1.0, 2.0): + queue.add( + WorkItem( + "cascade", "heating.sensor", "heating", {"heating.temp": value} + ) + ) + done, pending = queue.claim(2, 10) + queue.mark_done(done.entry_id, "heating.act") + await eng.renamer.rename("flow", "heating", _request(eng), "me") + assert [i.node for i in queue.live_items()] == ["warmth.sensor"] * 2 + + eng.controller.execution.start() + await _until(lambda: EFFECTS.count("lights.lamp") == 2) + await asyncio.sleep(0.5) + assert sorted(EFFECTS) == ["lights.lamp", "lights.lamp", "warmth.act"] + assert queue.live_items() == [] + assert "heating.temp" not in state and "warmth.temp" in state + finally: + await eng.controller.stop() + + try: + asyncio.run(scenario()) + finally: + client = state._client + stale = list(client.scan_iter(match=f"{namespace}*")) + if stale: + client.delete(*stale) + state.close() diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 38c0d00..cadd163 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -2192,6 +2192,16 @@ export const HealthSummarySchema = { type: 'string', title: 'Version', default: '' + }, + rename: { + anyOf: [ + { + '$ref': '#/components/schemas/RenameStatus' + }, + { + type: 'null' + } + ] } }, type: 'object', @@ -4019,6 +4029,122 @@ export const RenameRefusalSchema = { description: 'Why a rename cannot go ahead, with the ids of what is in the way.' } as const; +export const RenameRequestSchema = { + properties: { + request_id: { + type: 'string', + maxLength: 64, + minLength: 1, + title: 'Request Id' + }, + expected_name: { + type: 'string', + title: 'Expected Name' + }, + expected_version: { + type: 'integer', + title: 'Expected Version' + }, + new_name: { + type: 'string', + title: 'New Name' + }, + fingerprint: { + type: 'string', + title: 'Fingerprint' + } + }, + type: 'object', + required: ['request_id', 'expected_name', 'expected_version', 'new_name', 'fingerprint'], + title: 'RenameRequest', + description: `A rename as its plan described it. Refused when the object or the +plan moved since (409 \`\`stale\`\` or \`\`plan_changed\`\`).` +} as const; + +export const RenameResultSchema = { + properties: { + op: { + type: 'string', + title: 'Op' + }, + kind: { + type: 'string', + enum: ['flow', 'dashboard'], + title: 'Kind' + }, + id: { + type: 'string', + title: 'Id' + }, + old_name: { + type: 'string', + title: 'Old Name' + }, + new_name: { + type: 'string', + title: 'New Name' + }, + documents: { + items: { + '$ref': '#/components/schemas/RenamedDocument' + }, + type: 'array', + title: 'Documents' + } + }, + type: 'object', + required: ['op', 'kind', 'id', 'old_name', 'new_name', 'documents'], + title: 'RenameResult', + description: 'What a finished rename answers, again for a repeated request.' +} as const; + +export const RenameStatusSchema = { + properties: { + op: { + type: 'string', + title: 'Op' + }, + kind: { + type: 'string', + enum: ['flow', 'dashboard'], + title: 'Kind' + }, + id: { + type: 'string', + title: 'Id' + }, + old: { + type: 'string', + title: 'Old' + }, + new: { + type: 'string', + title: 'New' + }, + phase: { + type: 'string', + title: 'Phase' + }, + busy: { + items: { + type: 'string' + }, + type: 'array', + title: 'Busy', + default: [] + }, + error: { + type: 'string', + title: 'Error', + default: '' + } + }, + type: 'object', + required: ['op', 'kind', 'id', 'old', 'new', 'phase'], + title: 'RenameStatus', + description: 'A rename under way, for whoever is waiting on it or was turned away.' +} as const; + export const RenameWarningSchema = { properties: { code: { @@ -4036,6 +4162,39 @@ export const RenameWarningSchema = { description: 'Something a rename changes, or leaves, that its caller should know.' } as const; +export const RenamedDocumentSchema = { + properties: { + kind: { + type: 'string', + enum: ['flow', 'dashboard'], + title: 'Kind' + }, + id: { + type: 'string', + title: 'Id' + }, + version: { + type: 'integer', + title: 'Version' + }, + draft_version: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Draft Version' + } + }, + type: 'object', + required: ['kind', 'id', 'version'], + title: 'RenamedDocument', + description: "A document's versions after a rename. Both variants have the same." +} as const; + export const ResourceLevelSchema = { properties: { total: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index cb4e815..fcfa1e9 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsStartTelegramLinkData, AlertsStartTelegramLinkResponse, AlertsReadTelegramLinkData, AlertsReadTelegramLinkResponse, AlertsCancelTelegramLinkData, AlertsCancelTelegramLinkResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsData, DashboardsReadDashboardsResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsPlanDashboardRenameData, DashboardsPlanDashboardRenameResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsData, FlowsReadFlowsResponse, FlowsCreateFlowData, FlowsCreateFlowResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsPlanFlowRenameData, FlowsPlanFlowRenameResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsTrainNodeData, FlowsTrainNodeResponse, FlowsReadTrainingData, FlowsReadTrainingResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsStartTelegramLinkData, AlertsStartTelegramLinkResponse, AlertsReadTelegramLinkData, AlertsReadTelegramLinkResponse, AlertsCancelTelegramLinkData, AlertsCancelTelegramLinkResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsData, DashboardsReadDashboardsResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsPlanDashboardRenameData, DashboardsPlanDashboardRenameResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsData, FlowsReadFlowsResponse, FlowsCreateFlowData, FlowsCreateFlowResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsPlanFlowRenameData, FlowsPlanFlowRenameResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsCancelFlowRenameData, FlowsCancelFlowRenameResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsTrainNodeData, FlowsTrainNodeResponse, FlowsReadTrainingData, FlowsReadTrainingResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AiService { /** @@ -892,6 +892,34 @@ export class DashboardsService { } }); } + + /** + * Rename Dashboard + * Rename this dashboard's key, as its plan described it. + * + * Its two documents and nothing else: devices list it by id, and nothing + * binds to it by key. The same ``request_id`` again gets the same answer; + * refusals are as for a flow. + * @param data The data for the request. + * @param data.dashboardId + * @param data.requestBody + * @returns RenameResult Successful Response + * @throws ApiError + */ + public static renameDashboard(data: DashboardsRenameDashboardData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/dashboards/{dashboard_id}/rename', + path: { + dashboard_id: data.dashboardId + }, + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } } export class DevicesService { @@ -1573,6 +1601,67 @@ export class FlowsService { }); } + /** + * Rename Flow + * Rename this flow's key, as its plan described it. + * + * Everything named under the key moves: every flow and dashboard binding + * one of its messages, its history, live state and queued work. The engine + * waits meanwhile, for as long as the work running when it was asked takes + * to finish. Until then it can be called off, with ``DELETE`` on the + * request or by closing the connection, and nothing changes (409 + * ``aborted``); ``GET /observability/summary`` says what it waits for. + * + * The same ``request_id`` again gets the same answer. A rename that + * could not finish yet answers 503 ``completing`` and is finished in the + * background. Refusals are the plan's codes (409, 422 for a key that is no + * key), ``stale`` and ``plan_changed`` (409). + * @param data The data for the request. + * @param data.flowId + * @param data.requestBody + * @returns RenameResult Successful Response + * @throws ApiError + */ + public static renameFlow(data: FlowsRenameFlowData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{flow_id}/rename', + path: { + flow_id: data.flowId + }, + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Cancel Flow Rename + * Call off a rename still waiting for the engine. It answers 409 + * ``aborted`` once everything is as it was, and one that has begun to + * change things refuses this with 409 ``too_late``. + * @param data The data for the request. + * @param data.flowId + * @param data.requestId + * @returns Message Successful Response + * @throws ApiError + */ + public static cancelFlowRename(data: FlowsCancelFlowRenameData): CancelablePromise { + return __request(OpenAPI, { + method: 'DELETE', + url: '/api/v1/flows/{flow_id}/rename/{request_id}', + path: { + flow_id: data.flowId, + request_id: data.requestId + }, + errors: { + 422: 'Validation Error' + } + }); + } + /** * Read Node Source * Read a node's Python source, including unpublished edits. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 15b4019..afcfa0b 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -722,6 +722,7 @@ export type HealthSummary = { [key: string]: (number); }; version?: string; + rename?: (RenameStatus | null); }; export type HistoryCounts = { @@ -1258,6 +1259,16 @@ export type RemoteUserBody = { code: string; }; +/** + * A document's versions after a rename. Both variants have the same. + */ +export type RenamedDocument = { + kind: 'flow' | 'dashboard'; + id: string; + version: number; + draft_version?: (number | null); +}; + /** * What renaming a flow's or dashboard's key would change, and whether it * may. Read-only; the rename request repeats ``version`` and @@ -1295,6 +1306,44 @@ export type RenameRefusal = { ids?: Array<(string)>; }; +/** + * A rename as its plan described it. Refused when the object or the + * plan moved since (409 ``stale`` or ``plan_changed``). + */ +export type RenameRequest = { + request_id: string; + expected_name: string; + expected_version: number; + new_name: string; + fingerprint: string; +}; + +/** + * What a finished rename answers, again for a repeated request. + */ +export type RenameResult = { + op: string; + kind: 'flow' | 'dashboard'; + id: string; + old_name: string; + new_name: string; + documents: Array; +}; + +/** + * A rename under way, for whoever is waiting on it or was turned away. + */ +export type RenameStatus = { + op: string; + kind: 'flow' | 'dashboard'; + id: string; + old: string; + new: string; + phase: string; + busy?: Array<(string)>; + error?: string; +}; + /** * Something a rename changes, or leaves, that its caller should know. */ @@ -2024,6 +2073,13 @@ export type DashboardsPlanDashboardRenameData = { export type DashboardsPlanDashboardRenameResponse = (RenamePlan); +export type DashboardsRenameDashboardData = { + dashboardId: string; + requestBody: RenameRequest; +}; + +export type DashboardsRenameDashboardResponse = (RenameResult); + export type DevicesStartPairingData = { requestBody?: (PairHello | null); }; @@ -2189,6 +2245,20 @@ export type FlowsPlanFlowRenameData = { export type FlowsPlanFlowRenameResponse = (RenamePlan); +export type FlowsRenameFlowData = { + flowId: string; + requestBody: RenameRequest; +}; + +export type FlowsRenameFlowResponse = (RenameResult); + +export type FlowsCancelFlowRenameData = { + flowId: string; + requestId: string; +}; + +export type FlowsCancelFlowRenameResponse = (Message); + export type FlowsReadNodeSourceData = { flowId: string; nodeId: string;