From 2d65118d823e903e8f1b36bfc521351d1191909c Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 2 Nov 2026 21:30:01 +0100 Subject: [PATCH] Rename journal, steps and recovery: rename.json (RENAME_JOURNAL_FILE) records each step, documents are written whole and committed alone (FlowStore.replace_documents, identity.write_documents) and rolled back before the journal goes forward; history per run scoped by flow id, alert ledger, alerts.json, state and queue steps idempotent and checkpointed; audit row once per op; recover_documents before assign_ids and recover_runtime before the engine is built, retried until done; repeated requests get the stored result, stale ones 409; plan refuses rename_in_progress; alerts.json written atomically; completeness scanner in the test utilities --- backend/fluksio/api/routes/alerts.py | 6 +- backend/fluksio/core/config.py | 6 + backend/fluksio/flow/identity.py | 46 +- backend/fluksio/flow/rename.py | 568 ++++++++++++++++++++++++- backend/fluksio/flow/store.py | 19 + backend/fluksio/main.py | 39 +- backend/tests/test_rename_journal.py | 600 +++++++++++++++++++++++++++ backend/tests/utils/rename.py | 160 +++++++ docs/reference/configuration.md | 1 + 9 files changed, 1420 insertions(+), 25 deletions(-) create mode 100644 backend/tests/test_rename_journal.py create mode 100644 backend/tests/utils/rename.py diff --git a/backend/fluksio/api/routes/alerts.py b/backend/fluksio/api/routes/alerts.py index 256ad00..aee3647 100644 --- a/backend/fluksio/api/routes/alerts.py +++ b/backend/fluksio/api/routes/alerts.py @@ -15,6 +15,7 @@ from fluksio.api.deps import ( from fluksio.core.config import settings from fluksio.flow import telegram, webpush from fluksio.flow.alerts import Alert, AlertsConfig +from fluksio.flow.identity import _replace from fluksio.flow.secrets import SECRET_REF, SecretNotFound, resolve_params from fluksio.models import Message @@ -48,10 +49,11 @@ def read_config() -> AlertsConfig: def write_config(config: AlertsConfig) -> None: - """Blocking.""" + """Blocking. Whole or not at all: a file cut short would read as no + configuration, which is alerting turned off.""" path = _path() path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(config.model_dump_json(indent=2)) + _replace(path, config.model_dump_json(indent=2)) @router.get("/config", response_model=AlertsConfig, dependencies=SUPERUSER) diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 6164cc9..2ddaaea 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -23,6 +23,7 @@ DERIVED_PATHS = { "FLOWS_DIR": "flows", "SECRETS_FILE": "secrets.enc", "ALERTS_FILE": "alerts.json", + "RENAME_JOURNAL_FILE": "rename.json", "WEBPUSH_FILE": "webpush.json", "PANELS_FILE": "panels.json", "DEVICES_FILE": "devices.json", @@ -85,6 +86,11 @@ class Settings(BaseSettings): # Which failures reach which channel. Beside the flows, not in them: # alerting is the deployment's concern, not any one flow's. ALERTS_FILE: Path = Path("flow-data/alerts.json") + # The last key rename, step by step, so one a crash interrupted is + # finished or undone on the next start. Outside the flows, whose git + # history it must not enter, and kept once done so a repeated request is + # recognised. + RENAME_JOURNAL_FILE: Path = Path("flow-data/rename.json") # This instance's web push keypair and the browsers subscribed to it. # Beside the alerts it serves; deleting it makes every device subscribe # again. diff --git a/backend/fluksio/flow/identity.py b/backend/fluksio/flow/identity.py index 7e551eb..496ae2c 100644 --- a/backend/fluksio/flow/identity.py +++ b/backend/fluksio/flow/identity.py @@ -210,7 +210,7 @@ def assign_ids(root: Path) -> list[str]: _replace(path, json.dumps(stamped, indent=2, ensure_ascii=False) + "\n") paths = sorted(changed) - committed = _commit(root, paths) + committed = _commit(root, paths, COMMIT_MESSAGE) logger.warning( "Gave each flow and dashboard without an id its key as one, in %s", ", ".join(paths), @@ -220,18 +220,42 @@ def assign_ids(root: Path) -> list[str]: return paths +def write_documents(root: Path, texts: dict[str, str], message: str) -> bool: + """Write each of *texts*, by path relative to *root*, whose file holds + something else, then commit those paths alone with *message*. + + What a key rename writes and undoes (`fluksio.flow.rename`). Whatever else + is uncommitted stays so. False when the commit failed (`_commit`). + """ + for rel, text in texts.items(): + path = root / rel + if not path.exists() or path.read_text() != text: + _replace(path, text) + return _commit(root, sorted(texts), message) + + def _replace(path: Path, text: str) -> None: - """Write *path* whole or not at all: a crash leaves the old file or the new.""" + """Write *path* whole or not at all: a crash leaves the old file or the new. + + The directory is synced too, without which the replace itself may not + survive a power cut. + """ temp = path.with_name(f".{path.name}.tmp") with temp.open("w") as handle: handle.write(text) handle.flush() os.fsync(handle.fileno()) os.replace(temp, path) + directory = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) -def _commit(root: Path, paths: list[str]) -> bool: - """Commit *paths* alone, leaving whatever else is uncommitted as it was. +def _commit(root: Path, paths: list[str], message: str) -> bool: + """Commit *paths* alone with *message*, leaving whatever else is + uncommitted as it was. True once nothing is left to commit: done now, done by the run this one finishes, or no git to do it with, in which case the store is just files, @@ -262,7 +286,7 @@ def _commit(root: Path, paths: list[str]) -> bool: "-q", "--only", "-m", - COMMIT_MESSAGE, + message, "--", *paths, ) @@ -270,11 +294,19 @@ def _commit(root: Path, paths: list[str]) -> bool: return True if result.returncode != 0: logger.warning( - "Could not commit the ids, so the next start tries again: %s", + "Could not commit '%s': %s", + message, (result.stderr or result.stdout).strip(), ) return False return True -__all__ = ["COMMIT_MESSAGE", "IdPlan", "IdsRefused", "assign_ids", "plan_ids"] +__all__ = [ + "COMMIT_MESSAGE", + "IdPlan", + "IdsRefused", + "assign_ids", + "plan_ids", + "write_documents", +] diff --git a/backend/fluksio/flow/rename.py b/backend/fluksio/flow/rename.py index 9b4a4bf..63f3ce8 100644 --- a/backend/fluksio/flow/rename.py +++ b/backend/fluksio/flow/rename.py @@ -4,24 +4,31 @@ 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. ``docs/private/id_refac_rename.md`` has the whole protocol. +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. """ 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 Callable, Iterable, Iterator from dataclasses import dataclass +from datetime import UTC, datetime from pathlib import Path from typing import Any, Literal, cast import orjson -from pydantic import BaseModel -from sqlalchemy import String +from pydantic import BaseModel, Field +from sqlalchemy import String, literal, update from sqlmodel import Session, col, func, select from starlette.types import Scope @@ -36,13 +43,25 @@ from fluksio.flow.dashboards import ( DashboardStore, results_name, ) +from fluksio.flow.events import EventBus from fluksio.flow.executor import wait_out +from fluksio.flow.identity import _replace, write_documents from fluksio.flow.messages import flow_of from fluksio.flow.namespace import FLOW_FIELDS, NAME, OUTPUTS, Rename, names, rebind +from fluksio.flow.queue import WorkQueue from fluksio.flow.schemas import FlowDef, _validate_name -from fluksio.flow.state import MemoryState +from fluksio.flow.state import MemoryState, StateBackend from fluksio.flow.store import ISSUED_ID, FlowStore, InvalidName, ReservedKey -from fluksio.models import AlertSent, Run, RunArtifact, RunMetric, RunNode +from fluksio.models import ( + AlertSent, + EngineEvent, + Run, + RunArtifact, + RunMetric, + RunNode, +) + +logger = logging.getLogger(__name__) Kind = Literal["flow", "dashboard"] @@ -452,6 +471,12 @@ def plan_rename( 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) @@ -689,3 +714,536 @@ def _fingerprint(plan: RenamePlan) -> str: ) 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:`` 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.""" + + def __init__(self, status: int, code: str, detail: str) -> None: + super().__init__(detail) + self.status = status + self.code = code + self.detail = detail + + +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) diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index 0cc3fb6..9f9b58c 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -943,6 +943,25 @@ class FlowStore: self._commit(f"Delete flow {named(key, flow_id)}") return key + def replace_documents(self, texts: dict[str, str], message: str) -> bool: + """Write flow and dashboard documents whole, *texts* by path relative + to the store, and commit those paths alone with *message*. + + A key rename's (`fluksio.flow.rename`): each text carries its version + already, and whatever else is uncommitted stays so. False when the + commit failed. A store without git counts as committed. + """ + # Here rather than at the top: it reads where dashboards live, and + # the dashboards module imports this one. + from fluksio.flow.identity import write_documents + + with self._write_lock: + try: + return write_documents(self.root, texts, message) + finally: + # Whatever got written, the caches are to read it again. + self.revision += 1 + def duplicate_flow( self, flow_id: str, busy: Callable[[str], bool] | None = None ) -> FlowDef: diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index d0d0395..1675883 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -49,7 +49,12 @@ from fluksio.flow.plugins import load_plugins 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 RenameGate +from fluksio.flow.rename import ( + RenameGate, + recover_documents, + recover_runtime, + retry_until_done, +) from fluksio.flow.resources import ( ResourceAccountant, fair_share_env, @@ -295,15 +300,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Connectors register their node types before any flow is built with them. load_plugins() - alerts = AlertManager( - event_bus, config=read_alerts_config(), ledger=AlertLedger() - ) - execution = ExecutionService( - queue=_work_queue(), - max_workers=settings.FLOW_MAX_WORKERS, - events=event_bus, - max_cascades=settings.FLOW_MAX_CASCADES, - ) + # A rename the last process left half done settles its documents + # first: a published document and its draft written apart would + # have different keys, which assigning ids refuses. + await run_in_threadpool(recover_documents, settings.FLOWS_DIR) # Every flow and dashboard has an id before anything reads one: the # store's own `_migrate`, the device migrations below and every cache # after them write documents back with whatever id they read. @@ -313,6 +313,23 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # no flow holds any more is passed on as it is, which is the id of a # flow from before ids. event_bus.flow_id_of = lambda key: store.key_index().get(key, key) + state = _state_backend() + queue = _work_queue() + # Then the rest of it, before anything reads alerts, devices, modules, + # metrics, flows or runs. Until it is finished nothing is served. + await retry_until_done(lambda: recover_runtime(state, queue), event_bus) + # Work that arrived while a rename held the queue, which a crash + # left behind. + await run_in_threadpool(queue.release_held) + alerts = AlertManager( + event_bus, config=read_alerts_config(), ledger=AlertLedger() + ) + execution = ExecutionService( + queue=queue, + max_workers=settings.FLOW_MAX_WORKERS, + events=event_bus, + max_cascades=settings.FLOW_MAX_CASCADES, + ) # The packages node code imports. In the background, because a first # install of anything substantial takes minutes and `uv` is given five # of them twice over — held here, the container never answers its @@ -376,7 +393,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app.state.worker_hub = worker_hub controller = FlowController( store=store, - state=_state_backend(), + state=state, events=event_bus, max_workers=settings.FLOW_MAX_WORKERS, fastapi_app=app, diff --git a/backend/tests/test_rename_journal.py b/backend/tests/test_rename_journal.py new file mode 100644 index 0000000..db2643d --- /dev/null +++ b/backend/tests/test_rename_journal.py @@ -0,0 +1,600 @@ +"""A key rename's journal, its steps and what the next start does with one a +crash interrupted (section 4 of docs/private/id_refac_rename.md). Memory +always, Redis with FLUKSIO_TEST_REDIS_HOST (and optional PORT).""" + +import os +import shutil +import subprocess +import uuid +from collections import Counter +from collections.abc import Iterator +from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace + +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 identity +from fluksio.flow import rename as renaming +from fluksio.flow import state as state_module +from fluksio.flow.alerts import AlertsConfig, Channel +from fluksio.flow.controller import FlowController +from fluksio.flow.dashboards import DashboardDef, DashboardStore, WidgetDef +from fluksio.flow.executor import ExecutionService +from fluksio.flow.identity import plan_ids +from fluksio.flow.messages import MessageSpec +from fluksio.flow.namespace import Rename +from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkItem +from fluksio.flow.rename import ( + RenameConflict, + RenameResult, + apply_documents, + begin, + finish, + plan_rename, + read_journal, + recover_documents, + recover_runtime, + repeated, + rewrite_documents, + roll_back, + run_steps, +) +from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef +from fluksio.flow.state import MemoryState, RedisState +from fluksio.flow.store import FlowStore +from fluksio.models import AlertSent, EngineEvent, Run, RunArtifact, RunMetric, RunNode +from tests.utils.rename import leftovers, stability +from tests.utils.stores import created + +HEAT = Rename("heating", "warmth") +BACK = Rename("warmth", "heating") +#: The flows whose documents name `heating`: itself, `lights` and `notes`. +AFFECTED = ["heating", "lights", "notes"] +#: Document variants the rename rewrites: two of `heating`, one each of +#: `lights` and `notes`, two of the dashboard `home`. +DOCUMENTS = 6 +#: Outputs a node returned, as the stage cache keeps them. +CACHED = '{"heating.temp":20.5,"lights.level":2}' +NOW = datetime.now(UTC) + + +class Crash(Exception): + """The process stopping where it stands.""" + + +def git(root: Path, *args: str) -> str: + return subprocess.run( + ["git", "-C", str(root), *args], capture_output=True, text=True, check=True + ).stdout + + +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 _reads(node_id: str, *messages: str) -> NodeDef: + return NodeDef(id=node_id, requires=[MessageSpec(name=m) for m in messages]) + + +@pytest.fixture(scope="module") +def template(tmp_path_factory: pytest.TempPathFactory) -> Path: + """The store each test copies: `heating`, published with a draft over it; + `lights` reading it; `notes`, a draft reading it; `heating_extra`, which + only looks alike; and `home`, a dashboard bound to it with a draft over it. + Every flow has its key as its id, as one from before ids does.""" + root = tmp_path_factory.mktemp("template") / "flows" + store = FlowStore(root) + dashboards = DashboardStore(store) + store.write_flow( + FlowDef( + name="heating", + nodes=[ + NodeDef( + id="sensor", + provides=[ + MessageSpec(name="temp"), + MessageSpec(name="heating.set"), + ], + ) + ], + inputs=[FlowInput(spec=MessageSpec(name="heating.target"))], + ) + ) + draft = store.read_flow("heating", draft=True) + draft.nodes.append(_reads("valve", "heating.temp")) + store.write_draft(draft, draft.version) + store.write_flow(FlowDef(name="lights", nodes=[_reads("lamp", "heating.temp")])) + created(store, FlowDef(name="notes", nodes=[_reads("note", "heating.mode")])) + store.write_flow(FlowDef(name="heating_extra", nodes=[_reads("x", "lights.x")])) + dashboards.write( + DashboardDef( + name="home", + widgets=[ + WidgetDef(id="t", type="stat", config={"message": "heating.temp"}) + ], + ) + ) + home = dashboards.read("home", draft=True) + dashboards.write_draft(home.model_copy(update={"title": "Home"}), home.version) + return root + + +@pytest.fixture +def world( + template: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> Iterator[SimpleNamespace]: + """The template's store, alert channels and history: runs of `heating`, + of `lights` reading it, of `heating_extra` and of a deleted flow that + once had the key `heating`.""" + root = tmp_path / "flows" + shutil.copytree(template, root) + monkeypatch.setattr(settings, "RENAME_JOURNAL_FILE", tmp_path / "rename.json") + monkeypatch.setattr(settings, "ALERTS_FILE", tmp_path / "alerts.json") + channels = [ + Channel(name="board", kind="dashboard", config={"message": "heating.alert"}), + Channel(name="phone", kind="ntfy", config={"topic": "heating.alert"}), + ] + settings.ALERTS_FILE.write_text( + AlertsConfig(channels=channels).model_dump_json(indent=2) + ) + with Session(engine) as session: + for run_id, flow_id in ( + ("r1", "heating"), + ("r2", "heating"), + ("r3", "lights"), + ("r4", "heating_extra"), + ("r9", "f_gone2xa"), + ): + session.add( + Run( + id=run_id, + flow_id=flow_id, + created_at=NOW, + status="ok", + params={"note": "heating.temp"}, + ) + ) + session.commit() + session.add_all( + [ + RunMetric(run_id="r1", name="heating.loss", step=0, node="heating.s"), + RunMetric(run_id="r1", name="heating.loss", step=1, node="heating.s"), + RunMetric(run_id="r1", name="heating_extra.x", step=0), + RunMetric(run_id="r3", name="heating.temp", step=0, node="lights.lamp"), + RunMetric(run_id="r4", name="heating_extra.x", step=0), + RunMetric(run_id="r9", name="heating.loss", step=0), + RunArtifact(run_id="r2", name="heating.img", node="heating.s"), + RunArtifact(run_id="r9", name="heating.img"), + RunNode(run_id="r1", node="heating.s", cache_key="k1", outputs=CACHED), + RunNode( + run_id="r2", + node="heating.s", + cache_key="k2", + cached_from="r1", + outputs=CACHED, + ), + RunNode( + run_id="r3", + node="lights.lamp", + cache_key="k3", + outputs='{"lights.level":2}', + ), + RunNode( + run_id="r9", node="f_gone2xa.s", cache_key="k9", outputs=CACHED + ), + AlertSent(key="node_error:heating.s", ts=1.0), + AlertSent(key="flow_quarantined:heating", ts=1.0), + AlertSent(key="node_error:heating_extra.x", ts=1.0), + EngineEvent( + ts=NOW, + type="node_error", + flow_id="heating", + node="heating.s", + detail="heating.temp went missing", + ), + ] + ) + session.commit() + store = FlowStore(root) + yield SimpleNamespace(root=root, store=store, dashboards=DashboardStore(store)) + with Session(engine) as session: + for model in (RunMetric, RunArtifact, RunNode, Run, AlertSent, EngineEvent): + session.execute(delete(model)) + session.commit() + + +def _redis_server() -> tuple[str, int]: + host = os.environ.get("FLUKSIO_TEST_REDIS_HOST") + if not host: + pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks") + return host, int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379")) + + +@pytest.fixture +def backends(request: pytest.FixtureRequest) -> Iterator[SimpleNamespace]: + """State and queue, and how the next start gets them. Memory ones stand + in for durable backends here, so a restart hands back the same objects.""" + if request.param == "memory": + state, queue = MemoryState(), MemoryWorkQueue() + yield SimpleNamespace(open=lambda: (state, queue)) + return + host, port = _redis_server() + namespace = f"test-journal-{uuid.uuid4().hex}" + opened: list = [] + + def open_() -> tuple[RedisState, RedisWorkQueue]: + state = RedisState(host, port, namespace=namespace) + queue = RedisWorkQueue( + host, port, namespace=f"{namespace}-queue", consumer="engine-a" + ) + opened.extend((state, queue)) + return state, queue + + try: + yield SimpleNamespace(open=open_) + finally: + client = opened[0]._client + stale = list(client.scan_iter(match=f"{namespace}*")) + if stale: + client.delete(*stale) + for one in opened: + one.close() + + +def _item(node: str, kind: str = "cascade", **fields) -> WorkItem: + return WorkItem(kind=kind, node=node, flow=node.split(".")[0], **fields) + + +def _fill(state, queue) -> None: + """Live state and queued work, naming `heating` or not.""" + for key, value in ( + ("heating.temp", 20.5), + ("__version__:heating.temp", 3), + ("__ts__:heating.temp", 1.0), + ("__last_seen__:lights.lamp:heating.temp", 3), + ("__node__:heating.valve:last:x", 1), + ("heating_extra.x", 1), + ("lights.level", 2), + ): + state.set(key, value) + queue.add(_item("heating.acked")) + queue.ack(*queue.claim(1, 10)) + queue.add(_item("heating.valve", outputs={"heating.temp": 21})) + queue.add(_item("lights.lamp", outputs={"heating.temp": 21})) + queue.add(_item("heating_extra.x")) + queue.add_delayed(_item("heating.pump", kind="flush"), 100.0) + queue.move_due(150.0) + (claimed,) = queue.claim(1, 10, due_only=True) + queue.mark_done(claimed.entry_id, "heating.pump") + queue.add_delayed(_item("heating.limit"), 9e9) + queue.park("heating", _item("heating.p")) + + +def _rename( + world, state, queue, op: str = "op-1", durable: bool = True +) -> RenameResult: + """What the rename route does once the engine is held (S8): the journal, + the documents, the steps and the end.""" + rewrites = rewrite_documents(world.store, world.dashboards, "flow", "heating", HEAT) + journal = begin( + op, + "flow", + "heating", + HEAT, + rewrites, + actor="admin@example.com", + durable=durable, + ) + apply_documents(journal, world.store) + run_steps(journal, state, queue) + return finish(journal) + + +def _recover(world, state, queue) -> None: + """What the next start does before it serves (`main.lifespan`).""" + recover_documents(world.root) + # What assigning ids checks first, and refuses to start on. + assert plan_ids(world.root).problems == [] + recover_runtime(state, queue) + queue.release_held() + + +def _crash_at(patch: pytest.MonkeyPatch, point: str, nth: int) -> None: + seen: Counter[str] = Counter() + + def reached(name: str) -> None: + seen[name] += 1 + if seen[name] == nth: + raise Crash(f"{name} #{nth}") + + if point == "document": + write = identity._replace + + def replace(path: Path, text: str) -> None: + write(path, text) + reached("document") + + patch.setattr(identity, "_replace", replace) + return + + def fault(name: str) -> None: + if name == point: + reached(name) + + patch.setattr(renaming, "fault", fault) + + +def _audits() -> list[tuple[str, str, str]]: + with Session(engine) as session: + return [ + (row.flow_id, row.actor, row.detail) + for row in session.exec( + select(EngineEvent).where(col(EngineEvent.type) == "audit") + ) + ] + + +#: Every point a crash can stop a rename at, by backend: after each document, +#: the commit, each run's history, each state batch and queue structure, each +#: step, and after the journal is done but before anyone was told. +_EVERYWHERE = [ + *(("document", n) for n in range(1, DOCUMENTS + 1)), + ("commit", 1), + *(("sql:run", n) for n in (1, 2, 3)), + ("alert_ledger", 1), + ("alerts_file", 1), + ("state:keys", 1), + ("state", 1), + ("queue", 1), + ("audit", 1), + ("done", 1), +] +_QUEUE_STRUCTURES = ("due", "backlog", "delayed", "lists", "done markers") +CRASHES = [ + *(("memory", point, n) for point, n in [*_EVERYWHERE, ("queue:queue", 1)]), + *( + ("redis", point, n) + for point, n in [ + *_EVERYWHERE, + ("state:keys", 2), + *((f"queue:{one}", 1) for one in _QUEUE_STRUCTURES), + ] + ), +] + + +@pytest.mark.parametrize( + ("backends", "point", "nth"), + CRASHES, + indirect=["backends"], + ids=[f"{kind}-{point}-{n}" for kind, point, n in CRASHES], +) +def test_a_crash_anywhere_is_undone_or_finished_by_the_next_start( + world, backends, point: str, nth: int, monkeypatch: pytest.MonkeyPatch +): + # Small batches, so a crash can fall between two of them. + monkeypatch.setattr(state_module, "RENAME_BATCH", 2) + state, queue = backends.open() + _fill(state, queue) + counts = state.count(HEAT), queue.count(HEAT) + files = _files(world.root) + stable = stability(world.root) + found = leftovers("heating", world.root, AFFECTED, state, queue) + assert found + + with monkeypatch.context() as patch: + _crash_at(patch, point, nth) + with pytest.raises(Crash): + _rename(world, state, queue) + if (point, nth) == ("document", 1): + # A published document and its draft with different keys, which + # assigning ids would refuse to start on. + assert plan_ids(world.root).problems + + state, queue = backends.open() + _recover(world, state, queue) + + journal = read_journal() + assert journal is not None + if point in ("document", "commit"): + assert journal.phase == "rolled_back" + assert _files(world.root) == files + assert leftovers("heating", world.root, AFFECTED, state, queue) == found + assert (state.count(HEAT), queue.count(HEAT)) == counts + assert _audits() == [] + else: + assert journal.phase == "done" + assert journal.result is not None + assert leftovers("heating", world.root, AFFECTED, state, queue) == [] + # Moved, none lost and none twice. + assert (state.count(BACK), queue.count(BACK)) == counts + assert stability(world.root) == stable + assert len(_audits()) == 1 + assert git(world.root, "status", "--porcelain") == "" + + # Settled is settled: another start changes nothing. + settled = _files(world.root), journal + _recover(world, state, queue) + assert (_files(world.root), read_journal()) == settled + + +def test_history_moves_for_the_affected_flows_only(world): + """By flow id, so a deleted flow that had the key keeps its history, and + payloads and log text keep whatever they say.""" + _rename(world, MemoryState(), MemoryWorkQueue()) + + with Session(engine) as session: + metrics = session.exec(select(RunMetric.run_id, RunMetric.name)).all() + artifacts = session.exec(select(RunArtifact.run_id, RunArtifact.name)).all() + nodes = { + (row.run_id, row.node): (row.cache_key, row.cached_from, row.outputs) + for row in session.exec(select(RunNode)) + } + ledger = session.exec(select(AlertSent.key)).all() + params = session.exec(select(Run.params)).all() + details = session.exec( + select(EngineEvent.detail).where(col(EngineEvent.type) != "audit") + ).all() + assert sorted(metrics) == [ + ("r1", "heating_extra.x"), + ("r1", "warmth.loss"), + ("r1", "warmth.loss"), + ("r3", "warmth.temp"), + ("r4", "heating_extra.x"), + ("r9", "heating.loss"), + ] + assert sorted(artifacts) == [("r2", "warmth.img"), ("r9", "heating.img")] + moved = '{"lights.level":2,"warmth.temp":20.5}' + # A renamed output is no longer what its cache key promised, and where + # it was restored from still holds. + assert nodes == { + ("r1", "heating.s"): ("", "", moved), + ("r2", "heating.s"): ("", "r1", moved), + ("r3", "lights.lamp"): ("k3", "", '{"lights.level":2}'), + ("r9", "f_gone2xa.s"): ("k9", "", CACHED), + } + assert sorted(ledger) == [ + "flow_quarantined:warmth", + "node_error:heating_extra.x", + "node_error:warmth.s", + ] + assert params == [{"note": "heating.temp"}] * 5 + assert details == ["heating.temp went missing"] + assert _audits() == [ + ("heating", "admin@example.com", "renamed 'heating' to 'warmth' [op op-1]") + ] + alerts = AlertsConfig.model_validate_json(settings.ALERTS_FILE.read_text()) + assert [c.config for c in alerts.channels] == [ + {"message": "warmth.alert"}, + {"topic": "heating.alert"}, + ] + + +def test_a_rollback_restores_its_own_files_and_leaves_other_changes(world): + """Whatever else in the store is uncommitted stays so, and out of the + revert.""" + extra = world.root / "heating_extra" / "flow.json" + extra.write_text(extra.read_text().replace('"lights.x"', '"lights.y"')) + (world.root / "scratch.txt").write_text("mine") + status = git(world.root, "status", "--porcelain") + files = _files(world.root) + + state, queue = MemoryState(), MemoryWorkQueue() + with pytest.MonkeyPatch.context() as patch: + _crash_at(patch, "commit", 1) + with pytest.raises(Crash): + _rename(world, state, queue) + recover_documents(world.root) + + journal = read_journal() + assert journal is not None and journal.phase == "rolled_back" + assert _files(world.root) == files + assert git(world.root, "status", "--porcelain") == status + assert git(world.root, "log", "-2", "--format=%s").splitlines() == [ + "Revert rename of flow 'heating' to 'warmth'", + "Rename flow 'heating' to 'warmth' (heating)", + ] + assert sorted( + git(world.root, "show", "--name-only", "--format=", "HEAD").split() + ) == (sorted(doc.path for doc in journal.files)) + + +def test_a_commit_that_fails_is_undone_in_process(world, monkeypatch): + files, head = _files(world.root), git(world.root, "rev-parse", "HEAD") + rewrites = rewrite_documents(world.store, world.dashboards, "flow", "heating", HEAT) + journal = begin("op-1", "flow", "heating", HEAT, rewrites) + with monkeypatch.context() as patch: + patch.setattr(identity, "_commit", lambda *_: False) + with pytest.raises(RuntimeError): + apply_documents(journal, world.store) + assert world.store.read_flow("heating").name == "warmth" + + roll_back(journal, world.store) + + assert read_journal().phase == "rolled_back" + assert _files(world.root) == files + assert git(world.root, "rev-parse", "HEAD") == head + # The caches read the files again. + assert world.store.read_flow("heating").name == "heating" + assert world.store.key_index()["heating"] == "heating" + + +def test_a_store_without_git_renames_and_recovers(world): + shutil.rmtree(world.root / ".git") + files = _files(world.root) + state, queue = MemoryState(), MemoryWorkQueue() + with pytest.MonkeyPatch.context() as patch: + _crash_at(patch, "commit", 1) + with pytest.raises(Crash): + _rename(world, state, queue) + _recover(world, state, queue) + assert _files(world.root) == files + + _rename(world, state, queue, op="op-2") + assert leftovers("heating", world.root, AFFECTED, state, queue) == [] + + +def test_a_restart_on_memory_backends_finishes_the_rest(world): + """What was in memory went with the process, so there is nothing of it + left to move, and the rest is finished as on Redis.""" + state, queue = MemoryState(), MemoryWorkQueue() + _fill(state, queue) + with pytest.MonkeyPatch.context() as patch: + _crash_at(patch, "sql:run", 2) + with pytest.raises(Crash): + _rename(world, state, queue, durable=False) + + fresh = MemoryState(), MemoryWorkQueue() + _recover(world, *fresh) + + assert read_journal().phase == "done" + assert leftovers("heating", world.root, AFFECTED, *fresh) == [] + assert fresh[0].keys() == [] + + +def test_a_repeated_request_gets_the_stored_result_and_a_stale_one_409(world): + version = world.store.read_flow("heating", draft=True).version + assert repeated("op-1", "heating", version, "heating", version) is None + + result = _rename(world, MemoryState(), MemoryWorkQueue()) + + assert [(d.kind, d.id, d.version, d.draft_version) for d in result.documents] == [ + ("flow", "heating", version + 1, version + 1), + ("flow", "lights", 2, None), + ("flow", "notes", 2, 2), + ("dashboard", "home", 3, 3), + ] + now = world.store.read_flow("heating", draft=True) + assert (now.name, now.version) == ("warmth", version + 1) + assert repeated("op-1", now.name, now.version, "heating", version) == result + # The same rename asked for again under another id: its key matches, + # which proves nothing. + with pytest.raises(RenameConflict) as stale: + repeated("op-2", now.name, now.version, "heating", version) + assert (stale.value.status, stale.value.code) == (409, "stale") + + # One still finishing turns every other away, and the planner says so. + journal = read_journal() + journal.phase = "forward" + journal.save() + for request, code in (("op-1", "completing"), ("op-3", "rename_in_progress")): + with pytest.raises(RenameConflict) as busy: + repeated(request, now.name, now.version, now.name, now.version) + assert (busy.value.status, busy.value.code) == (503, code) + controller = FlowController( + world.store, + state=MemoryState(), + execution=ExecutionService(MemoryWorkQueue()), + ) + plan = plan_rename( + controller, world.dashboards, "flow", "lights", "lamps", AlertsConfig() + ) + assert [r.code for r in plan.refusals] == ["rename_in_progress"] diff --git a/backend/tests/utils/rename.py b/backend/tests/utils/rename.py new file mode 100644 index 0000000..616802f --- /dev/null +++ b/backend/tests/utils/rename.py @@ -0,0 +1,160 @@ +"""What a key rename has to move, read where each kind of thing keeps names. + +Only the fields classified as holding message names, never a text search: a +flow from before ids keeps its key as its id, in its directory and its history +rows; a deleted flow that had the same key keeps its history; and payloads, +logs and git history keep whatever they say. +""" + +import json +from collections.abc import Iterable +from pathlib import Path +from typing import Any + +from sqlmodel import Session, col, select + +from fluksio.core.config import settings +from fluksio.core.db import engine +from fluksio.flow.dashboards import DASHBOARD_BINDINGS, DASHBOARD_DIR +from fluksio.flow.messages import flow_of +from fluksio.flow.namespace import FLOW_FIELDS, NAME, OUTPUTS, Rename, names, touches +from fluksio.flow.queue import WorkQueue +from fluksio.flow.rename import DEDUP, SQL_NAMES +from fluksio.flow.state import StateBackend +from fluksio.models import ( + AlertSent, + EngineEvent, + FlowRun, + Run, + RunArtifact, + RunMetric, + RunNode, +) + +#: The history tables by name, as `SQL_NAMES` has them. +_TABLES = {model.__tablename__: model for model in (RunMetric, RunArtifact, RunNode)} + + +def _documents(root: Path) -> Iterable[tuple[Path, tuple[tuple[str, ...], ...]]]: + for path in sorted(root.glob("*/flow*.json")): + # `_devices/flow.json` is a device, not a flow. + if not path.parent.name.startswith(("_", ".")): + yield path, FLOW_FIELDS + for path in sorted((root / DASHBOARD_DIR).glob("*/dashboard*.json")): + yield path, DASHBOARD_BINDINGS + + +def leftovers( + key: str, + root: Path, + flow_ids: Iterable[str], + state: StateBackend | None = None, + queue: WorkQueue | None = None, +) -> list[str]: + """Every classified field that still names something of the flow key + *key*: document keys and bindings under *root*, the history of the flows + *flow_ids*, the alert ledger and channels, live state and queued work.""" + + def named(name: str) -> bool: + return "." in name and flow_of(name) == key + + found = [] + for path, paths in _documents(root): + rel = path.relative_to(root) + data = json.loads(path.read_text()) + if paths is FLOW_FIELDS and data.get("name") == key: + found.append(f"{rel}: its key") + found += [f"{rel}: {name}" for name in names(data, paths) if named(name)] + + runs = select(Run.id).where(col(Run.flow_id).in_([*flow_ids])) + with Session(engine) as session: + for (table, column), holds in SQL_NAMES.items(): + if holds == DEDUP: + # The tail is a node, or a flow key on its own. + found += [ + f"{table}.{column}: {one}" + for one in session.exec(select(AlertSent.key)).all() + if flow_of(one.partition(":")[2]) == key + ] + continue + model = _TABLES[table] + values = session.exec( + select(getattr(model, column)).where(col(model.run_id).in_(runs)) + ).all() + if holds == NAME: + found += [f"{table}.{column}: {v}" for v in values if named(v)] + elif holds == OUTPUTS: + found += [ + f"{table}.{column}: {name}" + for text in values + for name in json.loads(text or "{}") + if named(name) + ] + + try: + alerts = json.loads(settings.ALERTS_FILE.read_text()) + except FileNotFoundError: + alerts = {} + found += [ + f"alerts.json: {channel['config']['message']}" + for channel in alerts.get("channels", []) + if channel["kind"] == "dashboard" and named(channel["config"]["message"]) + ] + + if state is not None: + found += [one for one in state.keys() if touches(key, one)] + found += [f"names: {one}" for one in state.message_names() if named(one)] + if queue is not None: + probe = Rename(key, f"{key}_probe") + found += [ + f"queued: {one.node}" for one in queue.live_items() if probe.item(one) + ] + markers = queue.count(probe)["done_markers"] + found += [f"done markers: {markers}"] if markers else [] + return found + + +#: Columns that hold ids, which a rename never changes. +_IDENTITIES: tuple[Any, ...] = ( + Run.id, + Run.flow_id, + Run.for_node, + RunNode.node, + RunMetric.node, + RunArtifact.node, + FlowRun.id, + FlowRun.flow_id, +) + + +def stability(root: Path) -> dict[str, Any]: + """What a rename leaves exactly as it was: the store's directories, + every file in it but the documents, and the id columns of the history + (`EngineEvent` without the audit rows a rename adds).""" + documents = {path for path, _ in _documents(root)} + files = { + str(path.relative_to(root)): path.read_bytes() + for path in sorted(root.rglob("*")) + if path.is_file() and ".git" not in path.parts and path not in documents + } + directories = sorted( + str(path.relative_to(root)) + for path in root.rglob("*") + if path.is_dir() and ".git" not in path.parts + ) + with Session(engine) as session: + ids = { + str(column): sorted(map(str, session.exec(select(column)).all())) + for column in _IDENTITIES + } + ids["engine_event"] = sorted( + map( + str, + session.exec( + select(EngineEvent.flow_id, EngineEvent.node).where( + col(EngineEvent.type) != "audit" + ) + ).all(), + ) + ) + return {"files": files, "directories": directories, "ids": ids} diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 1cdd9a2..ef49207 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -21,6 +21,7 @@ For CLI installations, select storage with `--data-dir`, `FLUKSIO_HOME` or | `FLOWS_DIR` | `$DATA_DIR/flows` | flow git repository | | `SECRETS_FILE` | `$DATA_DIR/secrets.enc` | encrypted credentials | | `ALERTS_FILE` | `$DATA_DIR/alerts.json` | alert channels and rules | +| `RENAME_JOURNAL_FILE` | `$DATA_DIR/rename.json` | the last key rename, step by step, so a restart finishes or undoes one a crash interrupted | | `WEBPUSH_FILE` | `$DATA_DIR/webpush.json` | push keys and browser subscriptions | | `PROVISIONERS_FILE` | `$DATA_DIR/provisioners.json` | optional worker provisioning configuration | | `DEVICES_FILE` | `$DATA_DIR/devices.json` | device credential nonces, outside the flow repository |