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
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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:<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."""
|
||||
|
||||
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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
+28
-11
@@ -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,
|
||||
|
||||
@@ -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"]
|
||||
@@ -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}
|
||||
@@ -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 |
|
||||
|
||||
Reference in New Issue
Block a user