578 lines
22 KiB
Python
578 lines
22 KiB
Python
"""Running a key rename against a live engine (sections 2.3 and 2.4 of
|
|
docs/private/id_refac_rename.md): the barrier, calling it off, the splice and
|
|
the runtime controls it keeps. Memory always, the end-to-end queue check with
|
|
FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from sqlmodel import Session, col, delete, select
|
|
|
|
from fluksio.core.config import settings
|
|
from fluksio.core.db import engine
|
|
from fluksio.flow import executor
|
|
from fluksio.flow import rename as renaming
|
|
from fluksio.flow.alerts import AlertManager, AlertsConfig, Channel
|
|
from fluksio.flow.controller import NODE_TYPES, FlowController, NodeType
|
|
from fluksio.flow.dashboards import DashboardDef, DashboardStore, WidgetDef
|
|
from fluksio.flow.events import EventBus
|
|
from fluksio.flow.executor import ExecutionService
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.nodes import Node
|
|
from fluksio.flow.pipeline import EngineHeld
|
|
from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkItem
|
|
from fluksio.flow.rename import (
|
|
RenameConflict,
|
|
RenameGate,
|
|
Renamer,
|
|
RenameRequest,
|
|
plan_rename,
|
|
read_journal,
|
|
)
|
|
from fluksio.flow.schemas import FlowDef, FlowOrigin, NodeDef
|
|
from fluksio.flow.state import MemoryState, RedisState
|
|
from fluksio.flow.store import FlowStore
|
|
from fluksio.models import EngineEvent
|
|
from tests.utils.rename import leftovers, stability
|
|
|
|
#: What the probes did, by node id, in order.
|
|
EFFECTS: list[str] = []
|
|
#: A probe set to wait says so here, then waits for GO.
|
|
RUNNING = threading.Event()
|
|
GO = threading.Event()
|
|
|
|
|
|
class Probe(Node):
|
|
"""A node reaching outside, so doing it twice would be a second command.
|
|
What it did is written down, and with ``wait`` it takes as long as the
|
|
test says."""
|
|
|
|
idempotent = False
|
|
|
|
def __init__(self, requires=(), provides=(), params=None, name=None) -> None:
|
|
super().__init__(
|
|
f=self._act, requires=requires, provides=provides, params=params, name=name
|
|
)
|
|
|
|
def _act(self, params: dict[str, Any], **inputs: Any) -> None:
|
|
if params.get("wait"):
|
|
RUNNING.set()
|
|
assert GO.wait(10)
|
|
EFFECTS.append(self.id)
|
|
|
|
|
|
def _probe(node_id: str, requires=(), provides=(), **params: Any) -> NodeDef:
|
|
def specs(names) -> list[MessageSpec]:
|
|
return [MessageSpec(name=name, dtype=DType.FLOAT) for name in names]
|
|
|
|
return NodeDef(
|
|
id=node_id,
|
|
type="probe",
|
|
requires=specs(requires),
|
|
provides=specs(provides),
|
|
params=params,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[SimpleNamespace]:
|
|
"""`heating`, a sensor and a slow actor on it; `lights` reading it;
|
|
`heating_extra`, which only looks alike; `home`, a dashboard charting
|
|
`heating.temp`. Ids are keys, as for a flow from before ids."""
|
|
monkeypatch.setattr(settings, "RENAME_JOURNAL_FILE", tmp_path / "rename.json")
|
|
monkeypatch.setattr(settings, "ALERTS_FILE", tmp_path / "alerts.json")
|
|
monkeypatch.setitem(
|
|
NODE_TYPES, "probe", NodeType(title="Probe", description="", cls=Probe)
|
|
)
|
|
EFFECTS.clear()
|
|
RUNNING.clear()
|
|
GO.clear()
|
|
store = FlowStore(tmp_path / "flows")
|
|
store.write_flow(
|
|
FlowDef(
|
|
name="heating",
|
|
nodes=[
|
|
_probe("sensor", provides=["temp"]),
|
|
_probe("act", requires=["temp"], wait=True),
|
|
],
|
|
)
|
|
)
|
|
store.write_flow(
|
|
FlowDef(name="lights", nodes=[_probe("lamp", requires=["heating.temp"])])
|
|
)
|
|
store.write_flow(FlowDef(name="heating_extra", nodes=[_probe("x", provides=["x"])]))
|
|
dashboards = DashboardStore(store)
|
|
dashboards.write(
|
|
DashboardDef(
|
|
name="home",
|
|
widgets=[
|
|
WidgetDef(
|
|
id="c",
|
|
type="chart",
|
|
config={
|
|
"series": [{"message": "heating.temp"}],
|
|
"history": {"points": 500},
|
|
},
|
|
)
|
|
],
|
|
)
|
|
)
|
|
yield SimpleNamespace(root=store.root, store=store, dashboards=dashboards)
|
|
# Let a probe still waiting go, so its thread ends with the test.
|
|
GO.set()
|
|
with Session(engine) as session:
|
|
session.execute(delete(EngineEvent))
|
|
session.commit()
|
|
|
|
|
|
async def _engine(world, state=None, queue=None, consume: bool = True):
|
|
"""The engine over *world*'s store, built and, with *consume*, claiming."""
|
|
events = EventBus()
|
|
events.bind(asyncio.get_running_loop())
|
|
queue = queue if queue is not None else MemoryWorkQueue()
|
|
execution = ExecutionService(queue, max_cascades=2)
|
|
alerts = AlertManager(
|
|
events,
|
|
config=AlertsConfig(
|
|
channels=[
|
|
Channel(
|
|
name="board", kind="dashboard", config={"message": "heating.alert"}
|
|
)
|
|
]
|
|
),
|
|
)
|
|
controller = FlowController(
|
|
world.store,
|
|
state=state if state is not None else MemoryState(),
|
|
events=events,
|
|
execution=execution,
|
|
alerts=alerts,
|
|
)
|
|
controller.dashboards = world.dashboards
|
|
controller.set_history_limits(world.dashboards.history_requirements())
|
|
if consume:
|
|
await controller.start()
|
|
else:
|
|
await controller.reload(wait=None)
|
|
renamer = Renamer(controller, world.dashboards, RenameGate(), events, alerts=alerts)
|
|
# What the build said is not for whoever listens next.
|
|
await events.settle()
|
|
return SimpleNamespace(
|
|
controller=controller,
|
|
renamer=renamer,
|
|
events=events,
|
|
queue=queue,
|
|
alerts=alerts,
|
|
state=controller.state,
|
|
)
|
|
|
|
|
|
def _request(eng, new: str = "warmth", obj: str = "heating", op: str = "op-1"):
|
|
kind = "flow" if obj in eng.controller.store.list_flows() else "dashboard"
|
|
plan = plan_rename(
|
|
eng.controller, eng.renamer.dashboards, kind, obj, new, AlertsConfig()
|
|
)
|
|
return RenameRequest(
|
|
request_id=op,
|
|
expected_name=plan.name,
|
|
expected_version=plan.version,
|
|
new_name=new,
|
|
fingerprint=plan.fingerprint,
|
|
)
|
|
|
|
|
|
async def _until(check, timeout: float = 10.0) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while not check():
|
|
assert time.monotonic() < deadline, "timed out"
|
|
await asyncio.sleep(0.02)
|
|
|
|
|
|
async def _heard(events: EventBus, queue: asyncio.Queue) -> list[dict]:
|
|
await events.settle()
|
|
found = []
|
|
while not queue.empty():
|
|
found.append(queue.get_nowait())
|
|
return found
|
|
|
|
|
|
def _files(root: Path) -> dict[str, bytes]:
|
|
return {
|
|
str(path.relative_to(root)): path.read_bytes()
|
|
for path in sorted(root.rglob("*"))
|
|
if path.is_file() and ".git" not in path.parts
|
|
}
|
|
|
|
|
|
def test_a_rename_moves_the_engine_and_keeps_its_controls(world):
|
|
"""Stopped and paused are kept everywhere, quarantine only off the flows
|
|
it rebuilds (decision 6), and everything the engine keeps by name
|
|
follows."""
|
|
world.store.write_enabled("lights", False)
|
|
|
|
async def scenario() -> None:
|
|
eng = await _engine(world)
|
|
controller, pipeline = eng.controller, eng.controller.pipeline
|
|
try:
|
|
eng.state.set("heating.temp", 20.5)
|
|
controller.pause_flow("heating")
|
|
controller.pause_flow("heating_extra")
|
|
eng.queue.park("heating", WorkItem("cascade", "heating.sensor", "heating"))
|
|
for key in ("heating.sensor:loop", "heating_extra.x:loop"):
|
|
controller.supervisor._flows[key] = key.split(".")[0]
|
|
controller.supervisor._resting[key] = key.split(".")[0]
|
|
controller.loaded["heating.act"].last_error = "boom"
|
|
eng.events.emits["heating.act"] = 3
|
|
eng.events.digests["heating.cam"] = ["d1"]
|
|
eng.alerts._last_sent["node_error:heating.act"] = 7.0
|
|
request = _request(eng)
|
|
|
|
async with eng.events.subscribe() as heard:
|
|
result = await eng.renamer.rename("flow", "heating", request, "me")
|
|
told = await _heard(eng.events, heard)
|
|
|
|
assert (result.old_name, result.new_name) == ("heating", "warmth")
|
|
assert sorted(controller.loaded) == [
|
|
"heating_extra.x",
|
|
"lights.lamp",
|
|
"warmth.act",
|
|
"warmth.sensor",
|
|
]
|
|
assert controller.loaded["warmth.act"].last_error == "boom"
|
|
assert controller.disabled == {"lights"}
|
|
assert controller.paused_flows() == ["heating_extra", "warmth"]
|
|
assert controller.quarantined == {"heating_extra"}
|
|
assert controller.pipeline is pipeline
|
|
lamp = pipeline.get_node_by_id("lights.lamp")
|
|
assert [spec.name for spec in lamp.input_ports] == ["warmth.temp"]
|
|
assert eng.state.get("warmth.temp") == 20.5
|
|
assert "heating.temp" not in eng.state
|
|
assert controller.history_limits == {"warmth.temp": 500}
|
|
assert [item.node for item in eng.queue.unpark("warmth")] == [
|
|
"warmth.sensor"
|
|
]
|
|
assert eng.events.emits == {"warmth.act": 3}
|
|
assert eng.events.digests == {"warmth.cam": ["d1"]}
|
|
assert eng.alerts._last_sent == {"node_error:warmth.act": 7.0}
|
|
assert eng.alerts.config.channels[0].config == {"message": "warmth.alert"}
|
|
|
|
renamed = [e for e in told if e["type"] == "renamed"]
|
|
assert [
|
|
{k: e[k] for k in ("kind", "id", "old", "new", "op")} for e in renamed
|
|
] == [
|
|
{
|
|
"kind": "flow",
|
|
"id": "heating",
|
|
"old": "heating",
|
|
"new": "warmth",
|
|
"op": "op-1",
|
|
}
|
|
]
|
|
changed = [e["dashboard"] for e in told if e["type"] == "dashboard_changed"]
|
|
assert changed == ["home", ""]
|
|
(rebuilt,) = [e for e in told if e["type"] == "pipeline_rebuilt"]
|
|
assert rebuilt["paused"] == ["heating_extra", "warmth"]
|
|
|
|
assert not eng.renamer.gate.closed and eng.renamer.pending is None
|
|
assert not controller._lock.locked()
|
|
assert read_journal().phase == "done"
|
|
# Nothing is held: a value goes in again.
|
|
controller.publish_message("heating_extra.x", 1.0)
|
|
finally:
|
|
await eng.controller.stop()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("how", ["cancel", "disconnect"])
|
|
def test_a_rename_called_off_while_draining_changes_nothing(world, how):
|
|
"""It waits for the cascade running as long as that takes, says so, and
|
|
once called off leaves everything as it was, the engine running."""
|
|
files = _files(world.root)
|
|
gone = threading.Event()
|
|
|
|
async def disconnected() -> bool:
|
|
return gone.is_set()
|
|
|
|
async def scenario() -> None:
|
|
eng = await _engine(world)
|
|
controller = eng.controller
|
|
try:
|
|
controller.get_node("heating.sensor").inject({"temp": 21.0})
|
|
assert await asyncio.to_thread(RUNNING.wait, 5)
|
|
async with eng.events.subscribe() as heard:
|
|
renaming_ = asyncio.create_task(
|
|
eng.renamer.rename(
|
|
"flow", "heating", _request(eng), "me", disconnected
|
|
)
|
|
)
|
|
await _until(
|
|
lambda: eng.renamer.pending is not None
|
|
and eng.renamer.pending.busy == ["heating.sensor"]
|
|
)
|
|
waiting = [
|
|
e["busy"]
|
|
for e in await _heard(eng.events, heard)
|
|
if e["type"] == "rename_waiting"
|
|
]
|
|
assert waiting[-1] == ["heating.sensor"]
|
|
assert eng.renamer.gate.closed
|
|
with pytest.raises(EngineHeld):
|
|
controller.publish_message("heating_extra.x", 1.0)
|
|
|
|
if how == "cancel":
|
|
with pytest.raises(KeyError):
|
|
eng.renamer.cancel("heating", "op-2")
|
|
eng.renamer.cancel("heating", "op-1")
|
|
else:
|
|
gone.set()
|
|
with pytest.raises(RenameConflict) as refused:
|
|
await renaming_
|
|
assert (refused.value.status, refused.value.code) == (409, "aborted")
|
|
|
|
assert _files(world.root) == files
|
|
assert read_journal() is None
|
|
assert not eng.renamer.gate.closed and eng.renamer.pending is None
|
|
assert not controller._lock.locked()
|
|
assert sorted(controller.loaded) == [
|
|
"heating.act",
|
|
"heating.sensor",
|
|
"heating_extra.x",
|
|
"lights.lamp",
|
|
]
|
|
# The cascade it waited for finishes, and the next one runs.
|
|
GO.set()
|
|
await _until(lambda: sorted(EFFECTS) == ["heating.act", "lights.lamp"])
|
|
controller.get_node("heating.sensor").inject({"temp": 22.0})
|
|
await _until(lambda: len(EFFECTS) == 4)
|
|
finally:
|
|
GO.set()
|
|
await controller.stop()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_a_plan_that_moved_while_draining_is_refused_under_the_barrier(world):
|
|
"""A code-defined dependent that appeared while the rename waited."""
|
|
|
|
async def scenario() -> None:
|
|
eng = await _engine(world)
|
|
try:
|
|
eng.controller.get_node("heating.sensor").inject({"temp": 21.0})
|
|
assert await asyncio.to_thread(RUNNING.wait, 5)
|
|
renaming_ = asyncio.create_task(
|
|
eng.renamer.rename("flow", "heating", _request(eng), "me")
|
|
)
|
|
await _until(lambda: eng.renamer.gate.closed)
|
|
lights = world.store.read_flow("lights")
|
|
world.store.write_flow(lights.model_copy(update={"origin": FlowOrigin()}))
|
|
GO.set()
|
|
with pytest.raises(RenameConflict) as refused:
|
|
await renaming_
|
|
assert (refused.value.status, refused.value.code) == (
|
|
409,
|
|
"code_defined_dependent",
|
|
)
|
|
assert refused.value.extra["plan"]["refusals"][0]["ids"] == ["lights"]
|
|
assert world.store.read_flow("heating").name == "heating"
|
|
assert not eng.renamer.gate.closed
|
|
assert not eng.controller._lock.locked()
|
|
finally:
|
|
await eng.controller.stop()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_a_failing_step_is_retried_until_the_rename_finishes(world, monkeypatch):
|
|
"""Decision 5: held, said, and finished in the background."""
|
|
monkeypatch.setattr(renaming, "RETRY_S", (0.05, 0.1))
|
|
failures = []
|
|
|
|
def fault(point: str) -> None:
|
|
if point == "state" and len(failures) < 2:
|
|
failures.append(point)
|
|
raise ConnectionError("state unreachable")
|
|
|
|
monkeypatch.setattr(renaming, "fault", fault)
|
|
|
|
async def scenario() -> None:
|
|
eng = await _engine(world)
|
|
try:
|
|
request = _request(eng)
|
|
async with eng.events.subscribe() as heard:
|
|
with pytest.raises(RenameConflict) as busy:
|
|
await eng.renamer.rename("flow", "heating", request, "me")
|
|
assert (busy.value.status, busy.value.code) == (503, "completing")
|
|
pending = eng.renamer.pending
|
|
assert pending.phase == "forward"
|
|
assert "state unreachable" in pending.error
|
|
assert eng.renamer.gate.closed
|
|
for op, code in (
|
|
("op-1", "completing"),
|
|
("op-2", "rename_in_progress"),
|
|
):
|
|
other = request.model_copy(update={"request_id": op})
|
|
with pytest.raises(RenameConflict) as again:
|
|
await eng.renamer.rename("flow", "heating", other, "me")
|
|
assert (again.value.status, again.value.code) == (503, code)
|
|
|
|
await asyncio.wait_for(eng.renamer._resume, 10)
|
|
told = [e["type"] for e in await _heard(eng.events, heard)]
|
|
assert told.count("engine_degraded") == 2
|
|
assert "renamed" in told
|
|
assert eng.renamer.pending is None and not eng.renamer.gate.closed
|
|
assert read_journal().phase == "done"
|
|
assert sorted(eng.controller.loaded)[-2:] == ["warmth.act", "warmth.sensor"]
|
|
# Asked again, the same answer.
|
|
again = await eng.renamer.rename("flow", "heating", request, "me")
|
|
assert again == read_journal().result
|
|
finally:
|
|
await eng.controller.stop()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_renaming_there_and_back_restores_the_namespace(world):
|
|
"""With the producers stopped, `heating` to `warmth` and back leaves
|
|
documents, state and queued work as they were, but for versions."""
|
|
|
|
def documents() -> dict[str, Any]:
|
|
found = {}
|
|
for path in sorted(world.root.rglob("*.json")):
|
|
if ".git" not in path.parts:
|
|
data = json.loads(path.read_text())
|
|
if isinstance(data, dict):
|
|
data.pop("version", None)
|
|
found[str(path.relative_to(world.root))] = data
|
|
return found
|
|
|
|
async def scenario() -> None:
|
|
eng = await _engine(world, consume=False)
|
|
try:
|
|
eng.state.set("heating.temp", 20.5)
|
|
eng.state.set("__node__:heating.act:last:x", 1)
|
|
eng.queue.add(
|
|
WorkItem("cascade", "heating.sensor", "heating", {"heating.temp": 1})
|
|
)
|
|
eng.queue.add_delayed(WorkItem("flush", "lights.lamp", "lights"), 9e9)
|
|
before = documents(), stability(world.root)
|
|
state = {key: eng.state.get(key) for key in eng.state.keys()}
|
|
queued = [(i.node, i.outputs) for i in eng.queue.live_items()]
|
|
|
|
await eng.renamer.rename("flow", "heating", _request(eng), "me")
|
|
assert (
|
|
leftovers(
|
|
"heating", world.root, ["heating", "lights"], eng.state, eng.queue
|
|
)
|
|
== []
|
|
)
|
|
await eng.renamer.rename(
|
|
"flow", "heating", _request(eng, "heating", op="op-2"), "me"
|
|
)
|
|
|
|
assert (documents(), stability(world.root)) == before
|
|
assert {key: eng.state.get(key) for key in eng.state.keys()} == state
|
|
assert [(i.node, i.outputs) for i in eng.queue.live_items()] == queued
|
|
assert sorted(eng.controller.loaded)[0] == "heating.act"
|
|
finally:
|
|
await eng.controller.stop()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_a_dashboard_rename_is_its_documents_and_the_audit_alone(world):
|
|
async def scenario() -> None:
|
|
eng = await _engine(world)
|
|
try:
|
|
closings = []
|
|
eng.renamer.gate.close = lambda *args: closings.append(args)
|
|
request = _request(eng, "house", obj="home")
|
|
async with eng.events.subscribe() as heard:
|
|
result = await eng.renamer.rename("dashboard", "home", request, "me")
|
|
told = await _heard(eng.events, heard)
|
|
|
|
assert [(d.kind, d.id) for d in result.documents] == [("dashboard", "home")]
|
|
assert world.dashboards.read("home").name == "house"
|
|
assert closings == []
|
|
assert [(e["type"], e.get("dashboard", e.get("new"))) for e in told] == [
|
|
("renamed", "house"),
|
|
("dashboard_changed", "home"),
|
|
]
|
|
with Session(engine) as session:
|
|
audit = session.exec(
|
|
select(EngineEvent.flow_id, EngineEvent.detail).where(
|
|
col(EngineEvent.type) == "audit"
|
|
)
|
|
).all()
|
|
assert audit == [("", "renamed dashboard 'home' to 'house' [op op-1]")]
|
|
again = await eng.renamer.rename("dashboard", "home", request, "me")
|
|
assert again == result
|
|
finally:
|
|
await eng.controller.stop()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_a_pending_side_effect_runs_exactly_once_across_a_redis_rename(
|
|
world, monkeypatch
|
|
):
|
|
"""Claimed by an engine that went away: one item whose side effect was
|
|
recorded as done, which must not run again, and one whose was not, which
|
|
runs once, under the new name."""
|
|
host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
|
|
if not host:
|
|
pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
|
|
port = int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379"))
|
|
monkeypatch.setattr(executor, "RECLAIM_IDLE_MS", 50)
|
|
monkeypatch.setattr(executor, "RECLAIM_INTERVAL_S", 0.05)
|
|
namespace = f"test-rename-run-{uuid.uuid4().hex}"
|
|
state = RedisState(host, port, namespace=namespace)
|
|
queue = RedisWorkQueue(
|
|
host, port, namespace=f"{namespace}-queue", consumer="engine-a"
|
|
)
|
|
# The act waits for nobody here.
|
|
heating = world.store.read_flow("heating")
|
|
heating.nodes[1].params = {}
|
|
world.store.write_flow(heating)
|
|
|
|
async def scenario() -> None:
|
|
eng = await _engine(world, state=state, queue=queue, consume=False)
|
|
try:
|
|
for value in (1.0, 2.0):
|
|
queue.add(
|
|
WorkItem(
|
|
"cascade", "heating.sensor", "heating", {"heating.temp": value}
|
|
)
|
|
)
|
|
done, pending = queue.claim(2, 10)
|
|
queue.mark_done(done.entry_id, "heating.act")
|
|
await eng.renamer.rename("flow", "heating", _request(eng), "me")
|
|
assert [i.node for i in queue.live_items()] == ["warmth.sensor"] * 2
|
|
|
|
eng.controller.execution.start()
|
|
await _until(lambda: EFFECTS.count("lights.lamp") == 2)
|
|
await asyncio.sleep(0.5)
|
|
assert sorted(EFFECTS) == ["lights.lamp", "lights.lamp", "warmth.act"]
|
|
assert queue.live_items() == []
|
|
assert "heating.temp" not in state and "warmth.temp" in state
|
|
finally:
|
|
await eng.controller.stop()
|
|
|
|
try:
|
|
asyncio.run(scenario())
|
|
finally:
|
|
client = state._client
|
|
stale = list(client.scan_iter(match=f"{namespace}*"))
|
|
if stale:
|
|
client.delete(*stale)
|
|
state.close()
|