1823 lines
64 KiB
Python
1823 lines
64 KiB
Python
"""Renaming a flow's or a dashboard's key.
|
|
|
|
A rename moves everything named under a flow's key, in documents, history,
|
|
live state and queued work, so nothing may write through the API meanwhile.
|
|
`RenameGate` is the door that keeps it out. `plan_rename` says beforehand what
|
|
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`). `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
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import re
|
|
import threading
|
|
import time
|
|
from collections import Counter
|
|
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
|
|
|
|
import orjson
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import String, literal, update
|
|
from sqlmodel import Session, col, func, select
|
|
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 AlertManager, AlertsConfig
|
|
from fluksio.flow.controller import (
|
|
HOOK_PREFIX,
|
|
REBUILD_WAIT,
|
|
FlowController,
|
|
RebuildBusy,
|
|
)
|
|
from fluksio.flow.dashboards import (
|
|
DASHBOARD_BINDINGS,
|
|
DASHBOARD_DIR,
|
|
DashboardDef,
|
|
DashboardStore,
|
|
results_name,
|
|
)
|
|
from fluksio.flow.events import EventBus
|
|
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
|
|
from fluksio.flow.store import ISSUED_ID, FlowStore, InvalidName, ReservedKey
|
|
from fluksio.models import (
|
|
AlertSent,
|
|
EngineEvent,
|
|
Run,
|
|
RunArtifact,
|
|
RunMetric,
|
|
RunNode,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
Kind = Literal["flow", "dashboard"]
|
|
|
|
#: What reading looks like. Everything else under the API changes something.
|
|
_READS = frozenset({"GET", "HEAD", "OPTIONS"})
|
|
|
|
#: Mutations that cannot reach a flow's namespace, under the API prefix: signing
|
|
#: in and accounts, the assistant (whose tool calls come back through the API
|
|
#: 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 (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(?:/[^/]+)?$"
|
|
)
|
|
|
|
|
|
class RenameGate:
|
|
"""The door every request passes, which a rename closes.
|
|
|
|
Open, it counts the mutating API requests it lets in, for as long as each
|
|
takes. Closed, it turns new ones away and the rename waits for those
|
|
already in, so it starts from a namespace nothing is writing to through
|
|
the API. Hooks of the flows being renamed are turned away too, since their
|
|
nodes are torn down meanwhile. Reads, websockets and the exempt routes
|
|
pass either way, uncounted.
|
|
|
|
Thread-safe: requests pass on the event loop and `close` waits on a
|
|
thread, as the engine holds do.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Condition()
|
|
self._admitted: Counter[str] = Counter()
|
|
self._closed = False
|
|
self._affected: frozenset[str] = frozenset()
|
|
#: The rename that closed it, for whoever is turned away.
|
|
self.op = ""
|
|
|
|
@property
|
|
def closed(self) -> bool:
|
|
with self._lock:
|
|
return self._closed
|
|
|
|
def admit(self, scope: Scope) -> tuple[bool, str]:
|
|
"""Whether *scope* may pass, and the name it is counted under until
|
|
`leave`, empty when it is not counted."""
|
|
path: str = scope["path"]
|
|
with self._lock:
|
|
if path.startswith(f"{HOOK_PREFIX}/"):
|
|
flow_id = path[len(HOOK_PREFIX) + 1 :].split("/", 1)[0]
|
|
return not (self._closed and flow_id in self._affected), ""
|
|
api = settings.API_V1_STR
|
|
if (
|
|
scope["type"] != "http"
|
|
or scope["method"] in _READS
|
|
or not path.startswith(f"{api}/")
|
|
or _EXEMPT.match(path, len(api))
|
|
):
|
|
return True, ""
|
|
if self._closed:
|
|
return False, ""
|
|
name = f"{scope['method']} {path}"
|
|
self._admitted[name] += 1
|
|
return True, name
|
|
|
|
def leave(self, name: str) -> None:
|
|
"""A request counted under *name* is done."""
|
|
with self._lock:
|
|
self._admitted -= Counter([name])
|
|
self._lock.notify_all()
|
|
|
|
def close(
|
|
self,
|
|
op: str,
|
|
affected: Iterable[str],
|
|
abort: threading.Event,
|
|
report: Callable[[list[str]], None] | None = None,
|
|
) -> bool:
|
|
"""Turn mutations and the hooks of the *affected* flow ids away, then
|
|
wait for the requests already in to finish. Blocking.
|
|
|
|
Without a limit: ``report`` hears the requests still running each time
|
|
they change, and setting ``abort`` gives up, opens the gate again and
|
|
answers False. True means closed until `open`.
|
|
"""
|
|
with self._lock:
|
|
self._closed = True
|
|
self._affected = frozenset(affected)
|
|
self.op = op
|
|
if wait_out(self._lock, lambda: sorted(self._admitted), abort, report):
|
|
return True
|
|
self.open()
|
|
return False
|
|
|
|
def open(self) -> None:
|
|
with self._lock:
|
|
self._closed = False
|
|
self._affected = frozenset()
|
|
self.op = ""
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Documents
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Rewrite:
|
|
"""A stored document a rename changes, as texts before and after."""
|
|
|
|
kind: Kind
|
|
id: str
|
|
name: str
|
|
published_version: int | None
|
|
draft_version: int | None
|
|
#: Bindings moved, in whichever variant moves more of them.
|
|
fields: int
|
|
#: Bindings moved in the draft, which stays unpublished.
|
|
draft_fields: int
|
|
#: Declared in code (`FlowDef.origin`), so a sync puts the old names back.
|
|
code_defined: bool
|
|
#: ``(path, before, after)`` per variant, the path relative to the store.
|
|
files: list[tuple[str, str, str]]
|
|
|
|
|
|
_VARIANTS = (("published", ".json"), ("draft", ".draft.json"))
|
|
|
|
|
|
def _documents(
|
|
store: FlowStore, dashboards: DashboardStore
|
|
) -> Iterator[tuple[Kind, str, str, tuple[tuple[str, ...], ...]]]:
|
|
"""Every stored flow and dashboard: kind, id, path without its variant's
|
|
suffix, and where it names messages."""
|
|
for flow_id in store.list_flows():
|
|
yield "flow", flow_id, f"{flow_id}/flow", FLOW_FIELDS
|
|
found = {path.parent.name for path in dashboards.root.glob("*/dashboard*.json")}
|
|
for dash_id in sorted(found):
|
|
yield (
|
|
"dashboard",
|
|
dash_id,
|
|
f"{DASHBOARD_DIR}/{dash_id}/dashboard",
|
|
DASHBOARD_BINDINGS,
|
|
)
|
|
|
|
|
|
def _variants(root: Path, stem: str) -> dict[str, tuple[str, dict[str, Any]]]:
|
|
"""A document's published and draft variants, as text and as JSON.
|
|
|
|
Plain JSON rather than the model, so a field this version does not model,
|
|
or a dashboard still written as pages, comes through as it was. Empty for
|
|
one that does not parse, which the engine cannot read either.
|
|
"""
|
|
found: dict[str, tuple[str, dict[str, Any]]] = {}
|
|
for variant, suffix in _VARIANTS:
|
|
path = root / f"{stem}{suffix}"
|
|
if not path.exists():
|
|
continue
|
|
text = path.read_text()
|
|
try:
|
|
data = orjson.loads(text)
|
|
except orjson.JSONDecodeError:
|
|
return {}
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
found[variant] = (text, data)
|
|
return found
|
|
|
|
|
|
def rewrite_documents(
|
|
store: FlowStore,
|
|
dashboards: DashboardStore,
|
|
kind: Kind,
|
|
obj_id: str,
|
|
rename: Rename,
|
|
) -> list[Rewrite]:
|
|
"""Every document the rename of *obj_id* changes, published and draft.
|
|
Reads only; writing the texts is the rename's.
|
|
|
|
The object itself takes the new key. Renaming a flow also moves every
|
|
qualified name of its namespace that a flow names (`FLOW_FIELDS`, its own
|
|
included) or a dashboard binds (`DASHBOARD_BINDINGS`). Bare names are a
|
|
flow's own and follow its key by themselves, and no other field changes.
|
|
Both variants of a document that changes go to one version above either,
|
|
so an editor holding the old text is refused rather than saving it back,
|
|
and a draft discarded later cannot bring an older version back.
|
|
"""
|
|
rewrites = []
|
|
for doc_kind, doc_id, stem, paths in _documents(store, dashboards):
|
|
own = doc_kind == kind and doc_id == obj_id
|
|
if kind == "dashboard" and not own:
|
|
continue
|
|
variants = _variants(store.root, stem)
|
|
if not variants:
|
|
continue
|
|
moved = {
|
|
variant: rebind(data, paths, rename.binding) if kind == "flow" else 0
|
|
for variant, (_, data) in variants.items()
|
|
}
|
|
if not own and not any(moved.values()):
|
|
continue
|
|
versions = {
|
|
variant: int(data.get("version", 1))
|
|
for variant, (_, data) in variants.items()
|
|
}
|
|
working = variants.get("draft") or variants["published"]
|
|
rewrite = Rewrite(
|
|
kind=doc_kind,
|
|
id=doc_id,
|
|
name=str(working[1].get("name", doc_id)),
|
|
published_version=versions.get("published"),
|
|
draft_version=versions.get("draft"),
|
|
fields=max(moved.values()),
|
|
draft_fields=moved.get("draft", 0),
|
|
code_defined=any(data.get("origin") for _, data in variants.values()),
|
|
files=[],
|
|
)
|
|
top = max(versions.values()) + 1
|
|
for variant, suffix in _VARIANTS:
|
|
if variant not in variants:
|
|
continue
|
|
text, data = variants[variant]
|
|
if own:
|
|
data["name"] = rename.new
|
|
data["version"] = top
|
|
# The stores write what pydantic writes, which this matches.
|
|
after = orjson.dumps(data, option=orjson.OPT_INDENT_2).decode()
|
|
if text.endswith("\n"):
|
|
after += "\n"
|
|
rewrite.files.append((f"{stem}{suffix}", text, after))
|
|
rewrites.append(rewrite)
|
|
return rewrites
|
|
|
|
|
|
def references(store: FlowStore, dashboards: DashboardStore, key: str) -> list[str]:
|
|
"""The flows and dashboards that name a message of the flow key *key*,
|
|
published or draft, by id.
|
|
|
|
A rename onto *key* would hand those bindings to the renamed flow, and a
|
|
rename back would take them along. Creating a flow under *key* is another
|
|
matter: that is how bindings declared ahead of their flow are met.
|
|
"""
|
|
found = []
|
|
for _, doc_id, stem, paths in _documents(store, dashboards):
|
|
bound = (
|
|
name
|
|
for _, data in _variants(store.root, stem).values()
|
|
for name in names(data, paths)
|
|
)
|
|
if any("." in name and flow_of(name) == key for name in bound):
|
|
found.append(doc_id)
|
|
return found
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Plan
|
|
# -----------------------------------------------------------------------------
|
|
|
|
#: An alert's dedup key (`alerts.dedup_key`), whose tail is a node or a flow.
|
|
DEDUP = "dedup"
|
|
|
|
#: Where the history names messages, by table and column: a qualified name
|
|
#: (`namespace.NAME`), a JSON object keyed by them (`namespace.OUTPUTS`) or an
|
|
#: alert's dedup key. A rename rewrites exactly these. Every other column
|
|
#: holds ids, which never change, or payload and free text, which are history,
|
|
#: and a registry test fails when a column is added without saying which.
|
|
SQL_NAMES: dict[tuple[str, str], str] = {
|
|
("run_metric", "name"): NAME,
|
|
("run_artifact", "name"): NAME,
|
|
("run_node", "outputs"): OUTPUTS,
|
|
("alert_sent", "key"): DEDUP,
|
|
}
|
|
|
|
#: How recently another consumer must have read the queue to count as an
|
|
#: engine still running: it blocks a second at a time.
|
|
OTHER_ENGINE_MS = 10_000
|
|
|
|
#: Run states a rename refuses: the run would read names mid-move.
|
|
_ACTIVE_RUNS = ("queued", "running")
|
|
|
|
|
|
class RenamePlanRequest(BaseModel):
|
|
"""The key a flow or dashboard would be renamed to."""
|
|
|
|
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."""
|
|
|
|
code: str
|
|
detail: str
|
|
ids: list[str] = []
|
|
|
|
|
|
class RenameWarning(BaseModel):
|
|
"""Something a rename changes, or leaves, that its caller should know."""
|
|
|
|
code: str
|
|
detail: str
|
|
|
|
|
|
class PlannedDocument(BaseModel):
|
|
"""A document a rename rewrites, and the versions it was planned against."""
|
|
|
|
kind: Kind
|
|
id: str
|
|
name: str
|
|
published_version: int | None
|
|
draft_version: int | None
|
|
#: Bindings moved, in whichever variant moves more of them.
|
|
fields: int
|
|
|
|
|
|
class AlertCounts(BaseModel):
|
|
channels: int
|
|
ledger_rows: int
|
|
|
|
|
|
class HistoryCounts(BaseModel):
|
|
runs: int
|
|
metric_rows: int
|
|
artifact_rows: int
|
|
output_rows: int
|
|
|
|
|
|
class RuntimeCounts(BaseModel):
|
|
#: False with memory backends, whose state and queue a restart loses anyway.
|
|
durable: bool
|
|
state_keys: int
|
|
queue_items: int
|
|
delayed: int
|
|
parked: int
|
|
done_markers: int
|
|
|
|
|
|
class RenamePlan(BaseModel):
|
|
"""What renaming a flow's or dashboard's key would change, and whether it
|
|
may. Read-only; the rename request repeats ``version`` and
|
|
``fingerprint``, and is refused when either moved since."""
|
|
|
|
kind: Kind
|
|
id: str
|
|
name: str
|
|
new_name: str
|
|
#: The working copy's version, the draft's when there is one.
|
|
version: int
|
|
#: Over the kind, id and versions of every listed document and both
|
|
#: keys, so a document changing between plan and rename is noticed.
|
|
fingerprint: str = ""
|
|
allowed: bool = False
|
|
refusals: list[RenameRefusal] = []
|
|
documents: list[PlannedDocument] = []
|
|
#: Dry-run counts, a flow's only, which inform and are not checked again.
|
|
alerts: AlertCounts | None = None
|
|
history: HistoryCounts | None = None
|
|
runtime: RuntimeCounts | None = None
|
|
warnings: list[RenameWarning] = []
|
|
|
|
|
|
def plan_rename(
|
|
controller: FlowController,
|
|
dashboards: DashboardStore,
|
|
kind: Kind,
|
|
obj_id: str,
|
|
new_name: str,
|
|
alerts: AlertsConfig,
|
|
) -> RenamePlan:
|
|
"""What renaming the flow or dashboard *obj_id* to *new_name* would change,
|
|
and what refuses it. Blocking, and changes nothing.
|
|
|
|
:raises FlowNotFound: For a flow that does not exist.
|
|
:raises DashboardNotFound: For a dashboard that does not exist.
|
|
"""
|
|
store = controller.store
|
|
current: FlowDef | DashboardDef
|
|
if kind == "flow":
|
|
current = store.read_flow(obj_id, draft=True)
|
|
keys, ids = store.key_index(), set(store.list_flows())
|
|
else:
|
|
current = dashboards.read(obj_id, draft=True)
|
|
keys = dashboards.key_index()
|
|
ids = {path.parent.name for path in dashboards.root.glob("*/dashboard*.json")}
|
|
old = current.name
|
|
plan = RenamePlan(
|
|
kind=kind, id=obj_id, name=old, new_name=new_name, version=current.version
|
|
)
|
|
|
|
def refuse(code: str, detail: str, refused: Iterable[str] = ()) -> None:
|
|
plan.refusals.append(RenameRefusal(code=code, detail=detail, ids=[*refused]))
|
|
|
|
def warn(code: str, detail: str) -> None:
|
|
plan.warnings.append(RenameWarning(code=code, detail=detail))
|
|
|
|
try:
|
|
_validate_name(new_name)
|
|
except ValueError:
|
|
refuse("invalid_name", str(InvalidName(new_name)))
|
|
return plan
|
|
if ISSUED_ID.fullmatch(new_name):
|
|
refuse("reserved_shape", str(ReservedKey(new_name)))
|
|
return plan
|
|
if new_name == old:
|
|
refuse("same_name", f"The {kind} is already called '{old}'")
|
|
return plan
|
|
journal = read_journal()
|
|
if journal is not None and not journal.final:
|
|
refuse(
|
|
"rename_in_progress",
|
|
f"The rename of '{journal.old}' to '{journal.new}' is finishing",
|
|
)
|
|
|
|
rename = Rename(old, new_name)
|
|
rewrites = rewrite_documents(store, dashboards, kind, obj_id, rename)
|
|
plan.documents = [
|
|
PlannedDocument(
|
|
kind=r.kind,
|
|
id=r.id,
|
|
name=r.name,
|
|
published_version=r.published_version,
|
|
draft_version=r.draft_version,
|
|
fields=r.fields,
|
|
)
|
|
for r in rewrites
|
|
]
|
|
plan.fingerprint = _fingerprint(plan)
|
|
if new_name in keys:
|
|
refuse("name_taken", f"Another {kind} has the key '{new_name}'")
|
|
if new_name in ids - {obj_id}:
|
|
warn(
|
|
"new_name_is_id",
|
|
f"'{new_name}' is also the id of another {kind}, so a client naming "
|
|
"it by either is asked which one it means",
|
|
)
|
|
if kind == "flow":
|
|
_plan_flow(plan, controller, dashboards, rename, rewrites, alerts, refuse, warn)
|
|
plan.allowed = not plan.refusals
|
|
return plan
|
|
|
|
|
|
def _plan_flow(
|
|
plan: RenamePlan,
|
|
controller: FlowController,
|
|
dashboards: DashboardStore,
|
|
rename: Rename,
|
|
rewrites: list[Rewrite],
|
|
alerts: AlertsConfig,
|
|
refuse: Callable[..., None],
|
|
warn: Callable[[str, str], None],
|
|
) -> None:
|
|
"""The part of a plan only a flow has: what holds its namespace, and
|
|
what moves outside the documents."""
|
|
old, new = rename.old, rename.new
|
|
own = next(r for r in rewrites if r.kind == "flow" and r.id == plan.id)
|
|
if own.code_defined:
|
|
refuse(
|
|
"code_defined",
|
|
f"'{old}' is declared in code: rename it there and sync it again",
|
|
)
|
|
coded = [r.id for r in rewrites if r is not own and r.code_defined]
|
|
if coded:
|
|
refuse(
|
|
"code_defined_dependent",
|
|
f"{len(coded)} flows declared in code name messages of '{old}', and "
|
|
"a sync would put the old names back: rename them in code first",
|
|
coded,
|
|
)
|
|
flow_ids = [r.id for r in rewrites if r.kind == "flow"]
|
|
with Session(db_engine) as session:
|
|
active = session.exec(
|
|
select(Run.id, Run.flow_id, Run.for_node).where(
|
|
col(Run.status).in_(_ACTIVE_RUNS)
|
|
)
|
|
).all()
|
|
running = [
|
|
run_id
|
|
for run_id, flow_id, for_node in active
|
|
if flow_id in flow_ids or flow_of(for_node or "") in flow_ids
|
|
]
|
|
if running:
|
|
refuse(
|
|
"run_active",
|
|
f"{len(running)} runs of affected flows are queued or running",
|
|
running,
|
|
)
|
|
history, collisions, too_long = _history(session, rename, flow_ids)
|
|
ledger = [
|
|
rename.dedup_key(key)
|
|
for key in session.exec(select(AlertSent.key)).all()
|
|
if rename.dedup_key(key) != key
|
|
]
|
|
if collisions:
|
|
refuse(
|
|
"history_collision",
|
|
f"{collisions} history rows of the affected flows are already "
|
|
f"named '{new}.*'",
|
|
)
|
|
too_long += [key for key in ledger if len(key) > _length(col(AlertSent.key))]
|
|
if too_long:
|
|
refuse(
|
|
"name_too_long",
|
|
f"'{too_long[0]}' would be longer than its column holds",
|
|
)
|
|
|
|
residue = controller.key_residue(new)
|
|
referencing = references(controller.store, dashboards, new)
|
|
if residue or referencing:
|
|
named = [*residue, *(f"bindings in '{one}'" for one in referencing)]
|
|
refuse(
|
|
"target_in_use",
|
|
f"'{new}' still has {', '.join(named[:3])} behind it",
|
|
referencing,
|
|
)
|
|
|
|
state = controller.state
|
|
queue = controller.execution.queue if controller.execution else None
|
|
counts = queue.count(rename) if queue else {}
|
|
plan.runtime = RuntimeCounts(
|
|
durable=not isinstance(state, MemoryState),
|
|
state_keys=state.count(rename),
|
|
queue_items=counts.get("queue_items", 0),
|
|
delayed=counts.get("delayed", 0),
|
|
parked=counts.get("parked", 0),
|
|
done_markers=counts.get("done_markers", 0),
|
|
)
|
|
if queue is not None:
|
|
others = queue.other_consumers(OTHER_ENGINE_MS)
|
|
if others:
|
|
refuse(
|
|
"other_engine",
|
|
f"Another engine reads the queue ({', '.join(others)}), which a "
|
|
"rename cannot hold",
|
|
)
|
|
version = queue.server_version()
|
|
if version is not None and version < (7, 0):
|
|
refuse(
|
|
"redis_too_old",
|
|
f"Redis {version[0]}.{version[1]} is older than 7.0, which "
|
|
"a rename needs",
|
|
)
|
|
|
|
plan.history = history
|
|
plan.alerts = AlertCounts(
|
|
channels=sum(
|
|
channel.kind == "dashboard"
|
|
and rename.binding(str(channel.config.get("message") or ""))
|
|
!= channel.config.get("message")
|
|
for channel in alerts.channels
|
|
),
|
|
ledger_rows=len(ledger),
|
|
)
|
|
|
|
warn(
|
|
"external_references",
|
|
f"REST /messages/{old}.*, scripts, CLI, SDK and MCP key arguments and "
|
|
"hand-written bindings keep the old name",
|
|
)
|
|
if results_name(old) in dashboards.key_index():
|
|
warn(
|
|
"results_dashboard",
|
|
f"The dashboard '{results_name(old)}' keeps its name",
|
|
)
|
|
drafts = [r.id for r in rewrites if r.draft_fields]
|
|
if drafts:
|
|
warn(
|
|
"drafts_rewritten",
|
|
f"{len(drafts)} unpublished drafts change and stay unpublished",
|
|
)
|
|
pipeline = controller.pipeline
|
|
if (pipeline is not None and pipeline.is_paused(old)) or plan.runtime.parked:
|
|
warn("paused", f"'{old}' is paused, and its parked work moves with it")
|
|
if not plan.runtime.durable:
|
|
warn(
|
|
"not_durable",
|
|
"State and queued work are kept in memory, so a restart during the "
|
|
"rename loses them, as any restart does",
|
|
)
|
|
rows = history.metric_rows + history.artifact_rows + history.output_rows
|
|
warn(
|
|
"engine_pause",
|
|
f"Automations wait while {rows} history rows and "
|
|
f"{plan.runtime.state_keys} state keys move",
|
|
)
|
|
|
|
|
|
def _history(
|
|
session: Session, rename: Rename, flow_ids: list[str]
|
|
) -> tuple[HistoryCounts, int, list[str]]:
|
|
"""What moves in the history of the flows *flow_ids*, how many of its
|
|
names are in the new namespace already, and the names that would no
|
|
longer fit their column."""
|
|
old, new = f"{rename.old}.", f"{rename.new}."
|
|
grow = len(rename.new) - len(rename.old)
|
|
runs = select(Run.id).where(col(Run.flow_id).in_(flow_ids))
|
|
counts = {}
|
|
collisions = 0
|
|
too_long = []
|
|
for model in (RunMetric, RunArtifact):
|
|
name = col(model.name)
|
|
scope = col(model.run_id).in_(runs)
|
|
moving = func.substr(name, 1, len(old)) == old
|
|
counts[model] = session.exec(
|
|
select(func.count()).select_from(model).where(scope, moving)
|
|
).one()
|
|
collisions += session.exec(
|
|
select(func.count())
|
|
.select_from(model)
|
|
.where(scope, func.substr(name, 1, len(new)) == new)
|
|
).one()
|
|
longest = session.exec(
|
|
select(name).where(scope, moving).order_by(func.length(name).desc())
|
|
).first()
|
|
if longest is not None and len(longest) + grow > _length(name):
|
|
too_long.append(rename.name(longest))
|
|
outputs = session.exec(
|
|
select(RunNode.outputs).where(
|
|
col(RunNode.run_id).in_(runs),
|
|
col(RunNode.outputs).contains(f'"{old}', autoescape=True),
|
|
)
|
|
).all()
|
|
moved_outputs = sum(
|
|
any(rename.binding(key) != key for key in json.loads(text or "null") or {})
|
|
for text in outputs
|
|
)
|
|
total = session.exec(
|
|
select(func.count()).select_from(Run).where(col(Run.flow_id).in_(flow_ids))
|
|
).one()
|
|
history = HistoryCounts(
|
|
runs=total,
|
|
metric_rows=counts[RunMetric],
|
|
artifact_rows=counts[RunArtifact],
|
|
output_rows=moved_outputs,
|
|
)
|
|
return history, collisions, too_long
|
|
|
|
|
|
def _length(column: Any) -> int:
|
|
"""How many characters a text column holds."""
|
|
return cast(int, cast(String, column.type).length)
|
|
|
|
|
|
def _fingerprint(plan: RenamePlan) -> str:
|
|
seen = sorted(
|
|
(d.kind, d.id, d.published_version or 0, d.draft_version or 0)
|
|
for d in plan.documents
|
|
)
|
|
body = json.dumps([seen, plan.name, plan.new_name]).encode()
|
|
return f"sha256:{hashlib.sha256(body).hexdigest()}"
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Journal
|
|
# -----------------------------------------------------------------------------
|
|
|
|
#: Where a rename stands. ``prepared`` holds the documents' texts before any
|
|
#: is written, and a crash there is undone. ``forward`` follows the scoped
|
|
#: commit, and from there on the rename is only finished, never undone.
|
|
Phase = Literal["prepared", "forward", "done", "rolled_back"]
|
|
|
|
#: Waits between attempts at a rename's remaining steps, in seconds: the
|
|
#: first, doubled after each failure up to the last.
|
|
RETRY_S = (2.0, 60.0)
|
|
|
|
|
|
def fault(point: str) -> None:
|
|
"""Called at each named point of a rename's steps, and does nothing.
|
|
|
|
A test replaces it to stop a rename there, as a crash would. Each point
|
|
follows the work it names and precedes the journal saying so:
|
|
``commit`` once the documents are committed, ``sql:run`` after each
|
|
run's history, ``state:keys`` after each batch of state keys,
|
|
``queue:<structure>`` after each queue structure (`WorkQueue.rename`),
|
|
and ``alert_ledger``, ``alerts_file``, ``state``, ``queue`` and
|
|
``audit`` after each step. ``done`` follows the journal saying so. A
|
|
crash between two document writes is `identity._replace`'s to fake.
|
|
"""
|
|
|
|
|
|
class Affected(BaseModel):
|
|
"""What a rename changes, by id."""
|
|
|
|
#: Flows whose documents change: the renamed one and its dependents.
|
|
flows: list[str] = []
|
|
#: Of those, the ones the controller built, which it builds again.
|
|
rebuilt: list[str] = []
|
|
dashboards: list[str] = []
|
|
|
|
|
|
class JournaledDocument(BaseModel):
|
|
"""One document variant's whole text before and after the rename."""
|
|
|
|
kind: Kind
|
|
id: str
|
|
#: Relative to the store.
|
|
path: str
|
|
before: str
|
|
after: str
|
|
|
|
|
|
class SqlProgress(BaseModel):
|
|
done: bool = False
|
|
#: The last run whose history moved, in id order. Saved at most once a
|
|
#: second; a run moved twice is moved once, since only old names match.
|
|
after_run: str = ""
|
|
|
|
|
|
class Steps(BaseModel):
|
|
"""The steps after the documents, each set once it is done."""
|
|
|
|
sql: SqlProgress = Field(default_factory=SqlProgress)
|
|
alert_ledger: bool = False
|
|
alerts_file: bool = False
|
|
state: bool = False
|
|
queue: bool = False
|
|
audit: bool = False
|
|
|
|
|
|
class RenamedDocument(BaseModel):
|
|
"""A document's versions after a rename. Both variants have the same."""
|
|
|
|
kind: Kind
|
|
id: str
|
|
version: int
|
|
#: None for a document without a draft.
|
|
draft_version: int | None = None
|
|
|
|
|
|
class RenameResult(BaseModel):
|
|
"""What a finished rename answers, again for a repeated request."""
|
|
|
|
op: str
|
|
kind: Kind
|
|
id: str
|
|
old_name: str
|
|
new_name: str
|
|
documents: list[RenamedDocument]
|
|
|
|
|
|
class Journal(BaseModel):
|
|
"""A rename's record on disk (`settings.RENAME_JOURNAL_FILE`), written
|
|
before each step, so whatever stops it is finished or undone from here.
|
|
The last one stays, which is how a repeated request is recognised."""
|
|
|
|
format: int = 1
|
|
#: The request's id.
|
|
op: str
|
|
kind: Kind
|
|
id: str
|
|
old: str
|
|
new: str
|
|
actor: str = ""
|
|
started_at: float
|
|
phase: Phase = "prepared"
|
|
fingerprint: str = ""
|
|
#: False with memory backends: a restart then has no state or queued
|
|
#: work to move, since it lost them with the process.
|
|
durable: bool = True
|
|
affected: Affected = Field(default_factory=Affected)
|
|
files: list[JournaledDocument] = []
|
|
steps: Steps = Field(default_factory=Steps)
|
|
result: RenameResult | None = None
|
|
#: Why the last attempt at the steps failed.
|
|
error: str = ""
|
|
|
|
@property
|
|
def final(self) -> bool:
|
|
return self.phase in ("done", "rolled_back")
|
|
|
|
@property
|
|
def rename(self) -> Rename:
|
|
return Rename(self.old, self.new)
|
|
|
|
def save(self) -> None:
|
|
"""Whole or not at all, and on disk before this returns."""
|
|
path = settings.RENAME_JOURNAL_FILE
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
_replace(path, self.model_dump_json(indent=2) + "\n")
|
|
|
|
|
|
def read_journal() -> Journal | None:
|
|
"""The last rename's journal, None when there never was one."""
|
|
try:
|
|
text = settings.RENAME_JOURNAL_FILE.read_text()
|
|
except FileNotFoundError:
|
|
return None
|
|
return Journal.model_validate_json(text)
|
|
|
|
|
|
class RenameConflict(Exception):
|
|
"""A rename request that cannot go ahead as asked, with the status and
|
|
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, **extra: Any) -> None:
|
|
super().__init__(detail)
|
|
self.status = status
|
|
self.code = code
|
|
self.detail = detail
|
|
self.extra = extra
|
|
|
|
|
|
def repeated(
|
|
request_id: str,
|
|
name: str,
|
|
version: int,
|
|
expected_name: str,
|
|
expected_version: int,
|
|
) -> RenameResult | None:
|
|
"""The stored result of a rename asked for again, or None for a request
|
|
that may go ahead. *name* and *version* are the object's now, the
|
|
expected ones what the request was planned against.
|
|
|
|
A matching key is no proof of the same operation: a client that retries
|
|
a finished rename under a new request id expected the old key, and is
|
|
told its request is stale.
|
|
|
|
:raises RenameConflict: 503 ``completing`` for this request still
|
|
running, 503 ``rename_in_progress`` for another one, 409 ``stale``
|
|
for a request planned against another key or version.
|
|
"""
|
|
journal = read_journal()
|
|
if journal is not None and journal.op == request_id:
|
|
if journal.phase == "done" and journal.result is not None:
|
|
return journal.result
|
|
if not journal.final:
|
|
raise RenameConflict(503, "completing", "This rename is finishing")
|
|
if journal is not None and not journal.final:
|
|
raise RenameConflict(
|
|
503,
|
|
"rename_in_progress",
|
|
f"The rename of '{journal.old}' to '{journal.new}' is finishing",
|
|
)
|
|
if (expected_name, expected_version) != (name, version):
|
|
raise RenameConflict(
|
|
409,
|
|
"stale",
|
|
f"It is '{name}' at version {version} now, not '{expected_name}' "
|
|
f"at version {expected_version}",
|
|
)
|
|
return None
|
|
|
|
|
|
def begin(
|
|
op: str,
|
|
kind: Kind,
|
|
obj_id: str,
|
|
rename: Rename,
|
|
rewrites: list[Rewrite],
|
|
*,
|
|
actor: str = "",
|
|
fingerprint: str = "",
|
|
durable: bool = True,
|
|
rebuilt: Iterable[str] = (),
|
|
) -> Journal:
|
|
"""Write the journal of a rename about to change *rewrites*: its first
|
|
write, after which it is finished or undone but no longer called off."""
|
|
journal = Journal(
|
|
op=op,
|
|
kind=kind,
|
|
id=obj_id,
|
|
old=rename.old,
|
|
new=rename.new,
|
|
actor=actor,
|
|
started_at=time.time(),
|
|
fingerprint=fingerprint,
|
|
durable=durable,
|
|
affected=Affected(
|
|
flows=[r.id for r in rewrites if r.kind == "flow"],
|
|
rebuilt=[*rebuilt],
|
|
dashboards=[r.id for r in rewrites if r.kind == "dashboard"],
|
|
),
|
|
files=[
|
|
JournaledDocument(
|
|
kind=r.kind, id=r.id, path=path, before=before, after=after
|
|
)
|
|
for r in rewrites
|
|
for path, before, after in r.files
|
|
],
|
|
)
|
|
journal.save()
|
|
return journal
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Steps
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def _message(journal: Journal, revert: bool = False) -> str:
|
|
what = f"{journal.kind} '{journal.old}' to '{journal.new}'"
|
|
if revert:
|
|
return f"Revert rename of {what}"
|
|
return f"Rename {what} ({journal.id})"
|
|
|
|
|
|
def apply_documents(journal: Journal, store: FlowStore) -> None:
|
|
"""Replace every document with its new text and commit them alone, then
|
|
go forward. Raises when that fails, and `roll_back` undoes it."""
|
|
texts = {doc.path: doc.after for doc in journal.files}
|
|
if not store.replace_documents(texts, _message(journal)):
|
|
raise RuntimeError("Could not commit the renamed documents")
|
|
fault("commit")
|
|
journal.phase = "forward"
|
|
journal.save()
|
|
|
|
|
|
def roll_back(journal: Journal, store: FlowStore) -> None:
|
|
"""Put back the documents' texts from before, and commit that. Only
|
|
before the journal went forward; nothing else was written by then."""
|
|
store.replace_documents(
|
|
{doc.path: doc.before for doc in journal.files}, _message(journal, True)
|
|
)
|
|
journal.phase = "rolled_back"
|
|
journal.save()
|
|
|
|
|
|
def run_steps(
|
|
journal: Journal, state: StateBackend | None, queue: WorkQueue | None
|
|
) -> None:
|
|
"""Every step after the documents that is not done yet, in order: the
|
|
history, the alert ledger and channels, live state and queued work.
|
|
|
|
Each can be run again, since only what still has an old name moves, so
|
|
the journal only says which are finished. Without *state* or *queue*
|
|
their steps are passed over: a memory backend's restart has nothing left
|
|
to move. The engine must be held meanwhile.
|
|
"""
|
|
if journal.kind != "flow":
|
|
return
|
|
rename, steps = journal.rename, journal.steps
|
|
|
|
def finished(step: str) -> None:
|
|
fault(step)
|
|
setattr(steps, step, True)
|
|
journal.save()
|
|
|
|
try:
|
|
if not steps.sql.done:
|
|
_rename_history(journal)
|
|
if not steps.alert_ledger:
|
|
_rename_ledger(rename)
|
|
finished("alert_ledger")
|
|
if not steps.alerts_file:
|
|
_rename_channels(rename)
|
|
finished("alerts_file")
|
|
if not steps.state:
|
|
if state is not None:
|
|
state.rename_keys(rename, lambda what: fault(f"state:{what}"))
|
|
finished("state")
|
|
if not steps.queue:
|
|
if queue is not None:
|
|
queue.rename(rename, lambda what: fault(f"queue:{what}"))
|
|
finished("queue")
|
|
except Exception as exc:
|
|
journal.error = f"{type(exc).__name__}: {exc}"
|
|
with contextlib.suppress(OSError):
|
|
journal.save()
|
|
raise
|
|
|
|
|
|
def finish(journal: Journal) -> RenameResult:
|
|
"""Write the audit row, once, and mark the journal done with the result
|
|
a repeated request is answered with."""
|
|
if not journal.steps.audit:
|
|
_audit(journal)
|
|
fault("audit")
|
|
journal.steps.audit = True
|
|
journal.save()
|
|
journal.result = _result(journal)
|
|
journal.phase = "done"
|
|
journal.error = ""
|
|
journal.save()
|
|
fault("done")
|
|
return journal.result
|
|
|
|
|
|
def _rename_history(journal: Journal) -> None:
|
|
"""Message names in the history of the affected flows, a run at a time.
|
|
|
|
Scoped by the runs' flow ids, so a deleted flow that once had the same
|
|
key keeps its history as it was. Renamed outputs are no longer what
|
|
their cache key promised, so the key goes, and the outputs and where
|
|
they were restored from stay.
|
|
"""
|
|
rename, progress = journal.rename, journal.steps.sql
|
|
old, new = f"{rename.old}.", f"{rename.new}."
|
|
with Session(db_engine) as session:
|
|
runs = session.exec(
|
|
select(Run.id)
|
|
.where(
|
|
col(Run.flow_id).in_(journal.affected.flows),
|
|
col(Run.id) > progress.after_run,
|
|
)
|
|
.order_by(col(Run.id))
|
|
).all()
|
|
saved = time.monotonic()
|
|
for run_id in runs:
|
|
with Session(db_engine) as session:
|
|
for model in (RunMetric, RunArtifact):
|
|
name = col(model.name)
|
|
session.exec(
|
|
update(model)
|
|
.where(
|
|
col(model.run_id) == run_id,
|
|
func.substr(name, 1, len(old)) == old,
|
|
)
|
|
.values(name=literal(new).concat(func.substr(name, len(old) + 1)))
|
|
)
|
|
for row in session.exec(
|
|
select(RunNode).where(
|
|
col(RunNode.run_id) == run_id,
|
|
col(RunNode.outputs).contains(f'"{old}', autoescape=True),
|
|
)
|
|
).all():
|
|
outputs = json.loads(row.outputs or "{}")
|
|
moved = {rename.binding(key): value for key, value in outputs.items()}
|
|
if moved != outputs:
|
|
# Canonical, as `runs._cacheable` writes it.
|
|
row.outputs = json.dumps(
|
|
moved, sort_keys=True, separators=(",", ":")
|
|
)
|
|
row.cache_key = ""
|
|
session.add(row)
|
|
session.commit()
|
|
fault("sql:run")
|
|
if time.monotonic() - saved >= 1.0:
|
|
progress.after_run = run_id
|
|
journal.save()
|
|
saved = time.monotonic()
|
|
progress.done = True
|
|
journal.save()
|
|
|
|
|
|
def _rename_ledger(rename: Rename) -> None:
|
|
"""The keys alerts were sent under, so a cooldown carries over."""
|
|
with Session(db_engine) as session:
|
|
for row in session.exec(select(AlertSent)).all():
|
|
key = rename.dedup_key(row.key)
|
|
if key != row.key:
|
|
row.key = key
|
|
session.add(row)
|
|
session.commit()
|
|
|
|
|
|
def _rename_channels(rename: Rename) -> None:
|
|
"""The message a dashboard alert channel publishes to, in `alerts.json`."""
|
|
path = settings.ALERTS_FILE
|
|
try:
|
|
config = AlertsConfig.model_validate_json(path.read_text())
|
|
except (OSError, ValueError):
|
|
# None, or one the engine reads as empty: no channel names anything.
|
|
return
|
|
moved = 0
|
|
for channel in config.channels:
|
|
message = channel.config.get("message")
|
|
if channel.kind == "dashboard" and isinstance(message, str):
|
|
channel.config["message"] = rename.binding(message)
|
|
moved += channel.config["message"] != message
|
|
if moved:
|
|
_replace(path, config.model_dump_json(indent=2))
|
|
|
|
|
|
def _audit(journal: Journal) -> None:
|
|
"""Who renamed what, straight into the database rather than through the
|
|
bus, and only if it is not there already: a rename finished twice, by a
|
|
crash and the recovery, is recorded once."""
|
|
what = "" if journal.kind == "flow" else "dashboard "
|
|
detail = f"renamed {what}'{journal.old}' to '{journal.new}' [op {journal.op}]"
|
|
with Session(db_engine) as session:
|
|
seen = session.exec(
|
|
select(EngineEvent.id).where(
|
|
col(EngineEvent.type) == "audit", col(EngineEvent.detail) == detail
|
|
)
|
|
).first()
|
|
if seen is None:
|
|
session.add(
|
|
EngineEvent(
|
|
ts=datetime.now(UTC),
|
|
type="audit",
|
|
flow_id=journal.id if journal.kind == "flow" else "",
|
|
detail=detail,
|
|
actor=journal.actor,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
|
|
def _result(journal: Journal) -> RenameResult:
|
|
documents: dict[tuple[str, str], RenamedDocument] = {}
|
|
for doc in journal.files:
|
|
version = int(orjson.loads(doc.after)["version"])
|
|
entry = documents.setdefault(
|
|
(doc.kind, doc.id),
|
|
RenamedDocument(kind=doc.kind, id=doc.id, version=version),
|
|
)
|
|
if doc.path.endswith(".draft.json"):
|
|
entry.draft_version = version
|
|
return RenameResult(
|
|
op=journal.op,
|
|
kind=journal.kind,
|
|
id=journal.id,
|
|
old_name=journal.old,
|
|
new_name=journal.new,
|
|
documents=[*documents.values()],
|
|
)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Recovery
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def recover_documents(root: Path) -> None:
|
|
"""Settle the documents of a rename a crash interrupted. For startup,
|
|
before anything reads the store at *root*, ids being assigned included.
|
|
|
|
Before the journal went forward the old texts go back, otherwise the new
|
|
ones are made sure of, and only those paths are committed: a published
|
|
document and its draft left with different keys would refuse to start.
|
|
Plain files, no `FlowStore`.
|
|
"""
|
|
journal = read_journal()
|
|
if journal is None or journal.final:
|
|
return
|
|
if journal.phase == "prepared":
|
|
logger.warning(
|
|
"Undoing the rename of %s '%s' to '%s', which was interrupted "
|
|
"before it changed anything but its documents",
|
|
journal.kind,
|
|
journal.old,
|
|
journal.new,
|
|
)
|
|
texts = {doc.path: doc.before for doc in journal.files}
|
|
write_documents(root, texts, _message(journal, True))
|
|
journal.phase = "rolled_back"
|
|
journal.save()
|
|
return
|
|
write_documents(
|
|
root, {doc.path: doc.after for doc in journal.files}, _message(journal)
|
|
)
|
|
|
|
|
|
def recover_runtime(state: StateBackend, queue: WorkQueue) -> None:
|
|
"""Finish a rename a crash interrupted once its documents are settled
|
|
(`recover_documents`): the steps left, the audit row and the journal's
|
|
end. For startup, before anything reads alerts, devices, modules,
|
|
metrics, the controller or runs. Raises when a step fails, to be tried
|
|
again (`retry_until_done`)."""
|
|
journal = read_journal()
|
|
if journal is None or journal.phase != "forward":
|
|
return
|
|
logger.warning(
|
|
"Finishing the rename of %s '%s' to '%s'",
|
|
journal.kind,
|
|
journal.old,
|
|
journal.new,
|
|
)
|
|
if journal.durable:
|
|
run_steps(journal, state, queue)
|
|
else:
|
|
run_steps(journal, None, None)
|
|
finish(journal)
|
|
|
|
|
|
async def retry_until_done(attempt: Callable[[], None], events: EventBus) -> None:
|
|
"""Run *attempt* on a thread until it succeeds, however long that takes:
|
|
a rename half done is not to be served. Each failure is logged and
|
|
published as ``engine_degraded``, and the wait before the next doubles."""
|
|
wait, longest = RETRY_S
|
|
while True:
|
|
try:
|
|
await asyncio.to_thread(attempt)
|
|
return
|
|
except Exception as exc:
|
|
logger.exception("A rename could not finish; trying again in %.0f s", wait)
|
|
events.publish(
|
|
{
|
|
"type": "engine_degraded",
|
|
"detail": f"A rename could not finish: {exc}",
|
|
}
|
|
)
|
|
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(),
|
|
}
|
|
)
|