diff --git a/backend/fluksio/flow/namespace.py b/backend/fluksio/flow/namespace.py new file mode 100644 index 0000000..e740232 --- /dev/null +++ b/backend/fluksio/flow/namespace.py @@ -0,0 +1,187 @@ +"""The names the engine keeps live state and queued work under, and how a +rename of a flow key maps them. + +A flow's messages and nodes are named ``.``, and the engine's own +bookkeeping on them sits in state beside the values, under a reserved prefix +and the names it is about: ``__version__:heat.temp``, +``__last_seen__:heat.valve:hall.temp``. Renaming a flow key moves all of it to +the new name, so every family is registered here, beside the builders that +take their prefix from it, and a key of a family that is not is refused rather +than left behind. + +Nothing here does I/O. It maps names; moving the keys and items is the state +and queue backends' work. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from typing import TYPE_CHECKING, Any + +import orjson + +from fluksio.flow.messages import flow_of + +if TYPE_CHECKING: + from fluksio.flow.queue import WorkItem + +#: The message names in state, as a set (`RedisState` only). +NAMES = "__names__" +#: A message's recorded points (`RedisState`; memory keeps them apart). +HISTORY = "__history__" +#: How often a message was published. +VERSION = "__version__" +#: When a message was last published. +TIMESTAMP = "__ts__" +#: The version of a message a node last ran on. +LAST_SEEN = "__last_seen__" +#: When a rate-limited input last woke a node. +DELIVERED = "__in_ts__" +#: The value a rate-limited output kept back. +HELD = "__held__" +#: When a node's rate-limit window is due to be let through. +FLUSH = "__flush__" +#: What an engine-shipped node type remembers (`Node.remember`). +NODE_STATE = "__node__" + +#: A qualified message or node name, ``.``. +NAME = "name" +#: A node's own key, user data that may contain colons. +PRIVATE = "private" +#: A flow key. +FLOW = "flow" +#: A map keyed by qualified message names. +OUTPUTS = "outputs" + +#: Every bookkeeping family in live state, by prefix, with the parts that +#: follow it. A key is ``::``. Names never contain a colon, +#: so the key splits at its first ones and the last part takes the rest, which +#: is how a private key keeps its own. +FAMILIES: dict[str, tuple[str, ...]] = { + NAMES: (), + HISTORY: (NAME,), + VERSION: (NAME,), + TIMESTAMP: (NAME,), + LAST_SEEN: (NAME, NAME), + DELIVERED: (NAME, NAME), + HELD: (NAME,), + FLUSH: (NAME,), + NODE_STATE: (NAME, PRIVATE), +} + +#: What each `WorkItem` field holds that a rename changes, or None. A guard is +#: the node's own key, and the state it guards moves with the node. +ITEM_FIELDS: dict[str, str | None] = { + "kind": None, + "node": NAME, + "flow": FLOW, + "outputs": OUTPUTS, + "cause": None, + "not_before": None, + "guard_key": None, + "guard_value": None, + "entry_id": None, + "deliveries": None, + "enqueued_at": None, + "run_id": None, + "parent": None, +} + + +def _parts(key: str) -> tuple[str, list[tuple[str, str]]]: + """A state key's family prefix and its parts, each with what it holds. + + A key without a prefix is a message value, and its family is ``""``. + """ + if not key.startswith("__"): + return "", [(key, NAME)] + family, sep, rest = key.partition(":") + shape = FAMILIES.get(family) + parts = rest.split(":", len(shape) - 1) if shape is not None and sep else [] + if shape is None or len(parts) != len(shape): + raise ValueError(f"State key '{key}' is in no registered family") + return family, list(zip(parts, shape, strict=True)) + + +def touches(flow: str, key: str) -> bool: + """Whether the state key *key* names anything of the flow key *flow*.""" + _, parts = _parts(key) + return any(kind == NAME and flow_of(part) == flow for part, kind in parts) + + +@dataclass(frozen=True) +class Rename: + """The flow key *old* becoming *new*, applied to the names that embed it.""" + + old: str + new: str + + def name(self, qualified: str) -> str: + """A qualified message or node name in the new namespace. + + Only the part before the first dot is the flow, so renaming ``heat`` + leaves ``heating.temp`` alone. A bare key is its own namespace. + """ + if flow_of(qualified) != self.old: + return qualified + return self.new + qualified[len(self.old) :] + + def flow(self, key: str) -> str: + return self.new if key == self.old else key + + def state_key(self, key: str) -> str | None: + """Where the state key *key* moves to, or None if it stays. + + :raises ValueError: For a bookkeeping key of an unregistered family. + """ + family, parts = _parts(key) + mapped = [self.name(part) if kind == NAME else part for part, kind in parts] + renamed = ":".join([family, *mapped]) if family else mapped[0] + return renamed if renamed != key else None + + def item(self, item: WorkItem) -> WorkItem | None: + """*item* with its names in the new namespace, or None if it has none. + + A ``run`` item names its flow by id, which a rename leaves as it is. + """ + if item.kind == "run": + return None + changes = {} + for field, holds in ITEM_FIELDS.items(): + if holds is not None: + value = getattr(item, field) + mapped = self._mapped(holds, value) + if mapped != value: + changes[field] = mapped + return replace(item, **changes) if changes else None + + def fields(self, fields: dict[str, str]) -> dict[str, str] | None: + """A serialized item (`WorkItem.to_fields`) in the new namespace, or + None if it names nothing renamed. Fields a rename does not touch keep + their exact text, so an unchanged item serializes as it did.""" + if fields.get("kind") == "run": + return None + renamed = dict(fields) + for field, holds in ITEM_FIELDS.items(): + if holds is None or field not in fields: + continue + if holds == OUTPUTS: + outputs = orjson.loads(fields[field] or "{}") + mapped = self._mapped(holds, outputs) + if mapped != outputs: + renamed[field] = orjson.dumps(mapped).decode() + else: + renamed[field] = self._mapped(holds, fields[field]) + return renamed if renamed != fields else None + + def dedup_key(self, key: str) -> str: + """An alert's ``:`` (`alerts.dedup_key`).""" + event, sep, about = key.partition(":") + return f"{event}{sep}{self.name(about)}" + + def _mapped(self, holds: str, value: Any) -> Any: + if holds == NAME: + return self.name(value) + if holds == FLOW: + return self.flow(value) + return {self.name(name): one for name, one in value.items()} diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index 12ed281..81fef88 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Any from fluksio.flow import logs from fluksio.flow.messages import MessageSpec, qualify +from fluksio.flow.namespace import NODE_STATE from fluksio.flow.supervision import Supervisor if TYPE_CHECKING: @@ -283,7 +284,7 @@ class Node: # ------------------------------------------------------------------------- def _state_key(self, key: str) -> str: - return f"__node__:{self.id}:{key}" + return f"{NODE_STATE}:{self.id}:{key}" def remember(self, key: str, value: Any) -> None: """Keep a value for the next run of this node.""" diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index ab20a36..8eb2bfa 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -34,6 +34,14 @@ from fluksio.flow import logs from fluksio.flow.artifacts import is_reference from fluksio.flow.events import EventBus from fluksio.flow.messages import flow_of, requalify +from fluksio.flow.namespace import ( + DELIVERED, + FLUSH, + HELD, + LAST_SEEN, + TIMESTAMP, + VERSION, +) from fluksio.flow.nodes import Node from fluksio.flow.queue import WorkQueue from fluksio.flow.state import MemoryState, StateBackend @@ -633,25 +641,25 @@ class Pipeline: # ------------------------------------------------------------------------- def _version_key(self, msg_name: str) -> str: - return f"__version__:{msg_name}" + return f"{VERSION}:{msg_name}" def _last_seen_key(self, node_name: str, msg_name: str) -> str: - return f"__last_seen__:{node_name}:{msg_name}" + return f"{LAST_SEEN}:{node_name}:{msg_name}" def _timestamp_key(self, msg_name: str) -> str: - return f"__ts__:{msg_name}" + return f"{TIMESTAMP}:{msg_name}" def _delivered_key(self, node_name: str, msg_name: str) -> str: """When a rate-limited input last woke this node.""" - return f"__in_ts__:{node_name}:{msg_name}" + return f"{DELIVERED}:{node_name}:{msg_name}" def _held_key(self, msg_name: str) -> str: """The value a rate-limited output kept back, waiting for its window.""" - return f"__held__:{msg_name}" + return f"{HELD}:{msg_name}" def _flush_key(self, node_name: str) -> str: """When a rate-limit window of this node is due to be let through.""" - return f"__flush__:{node_name}" + return f"{FLUSH}:{node_name}" def _window_split( self, node: Node, result: dict[str, Any], now: float diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index cb68b82..44e31ec 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -52,6 +52,8 @@ STALE_CONSUMER_IDLE_MS = 3_600_000 # The Redis one expires each after an hour; this is the same idea sized by # count, since redelivery happens seconds after the claim and never later. DONE_MARKERS = 10_000 +# The Redis queue keeps them as ``:__done__::``. +DONE = "__done__" # A promotion is one Redis operation, including when the caller loses its # connection before hearing the result. Write first: Lua does not roll back a @@ -162,6 +164,14 @@ def owned_by(item: WorkItem, flow: str) -> bool: ) +def done_marker(key: str) -> tuple[str, str]: + """The entry id and node a done marker is for: `RedisWorkQueue._done_key` + read back. Split at the last colon, since an entry id from the due lane + has one of its own and a node id never does.""" + entry_id, node = key.split(f":{DONE}:", 1)[1].rsplit(":", 1) + return entry_id, node + + class WorkQueue(ABC): """What the execution service pulls from.""" @@ -548,7 +558,7 @@ class RedisWorkQueue(WorkQueue): return f"{self._ns}:__parked__:{flow}" def _done_key(self, entry_id: str, node: str) -> str: - return f"{self._ns}:__done__:{entry_id}:{node}" + return f"{self._ns}:{DONE}:{entry_id}:{node}" def add(self, item: WorkItem) -> None: self._redis.xadd( diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py index 23ed33a..f313ea3 100644 --- a/backend/fluksio/flow/state.py +++ b/backend/fluksio/flow/state.py @@ -17,6 +17,8 @@ from typing import Any, cast import orjson import redis +from fluksio.flow.namespace import HISTORY, NAMES + #: A node may return a dict keyed by something other than a string, and the #: stdlib encoder this replaced turned those keys into strings rather than #: refusing them. Two differences remain and are both improvements: a @@ -533,7 +535,7 @@ class RedisState(StateBackend): return self._deserialize(data) if data is not None else default def _names_key(self) -> str: - return self._key("__names__") + return self._key(NAMES) def message_names(self) -> list[str]: """The message names, from the set kept beside them. @@ -770,7 +772,7 @@ class RedisState(StateBackend): self._client.close() def _history_key(self, key: str) -> str: - return self._key(f"__history__:{key}") + return self._key(f"{HISTORY}:{key}") def append_history( self, values: dict[str, Any], ts: float, limits: dict[str, int] | None = None diff --git a/backend/tests/flow/test_namespace.py b/backend/tests/flow/test_namespace.py new file mode 100644 index 0000000..2ccb45c --- /dev/null +++ b/backend/tests/flow/test_namespace.py @@ -0,0 +1,171 @@ +"""The rename mapper, and the registries that keep it complete.""" + +import ast +import dataclasses +import inspect +import re +from types import SimpleNamespace + +import pytest + +from fluksio.flow import namespace, pipeline, state +from fluksio.flow.alerts import dedup_key +from fluksio.flow.namespace import FAMILIES, ITEM_FIELDS, Rename, touches +from fluksio.flow.nodes import Node, base +from fluksio.flow.pipeline import Pipeline +from fluksio.flow.queue import RedisWorkQueue, WorkItem, done_marker +from fluksio.flow.state import MemoryState, RedisState + +HEAT = Rename("heat", "warm") + + +@pytest.mark.parametrize( + ("key", "moved"), + [ + ("heat.temp", "warm.temp"), + ("heat.room.temp", "warm.room.temp"), + ("heating.temp", None), + ("__version__:heat.temp", "__version__:warm.temp"), + ("__version__:heating.temp", None), + ("__history__:heat.room.temp", "__history__:warm.room.temp"), + ("__flush__:heat.valve", "__flush__:warm.valve"), + ("__last_seen__:heat.valve:hall.temp", "__last_seen__:warm.valve:hall.temp"), + ("__last_seen__:hall.lamp:heat.temp", "__last_seen__:hall.lamp:warm.temp"), + ("__last_seen__:heat.valve:heat.temp", "__last_seen__:warm.valve:warm.temp"), + ("__in_ts__:heating.valve:heat.temp", "__in_ts__:heating.valve:warm.temp"), + ("__in_ts__:heating.valve:heating.temp", None), + # A node's own key is its data: only the node part moves. + ("__node__:heat.limit:wait:heat.temp", "__node__:warm.limit:wait:heat.temp"), + ("__node__:hall.limit:wait:heat.temp", None), + ("__names__", None), + # Already under the new key. + ("warm.temp", None), + ("__last_seen__:warm.valve:warm.temp", None), + ], +) +def test_state_keys_move_by_the_names_they_embed(key, moved): + assert HEAT.state_key(key) == moved + + +@pytest.mark.parametrize( + "key", + ["__bogus__:heat.temp", "__version__", "__names__:heat.temp", "__in_ts__:heat.v"], +) +def test_a_key_of_no_registered_family_is_refused(key): + with pytest.raises(ValueError, match="no registered family"): + HEAT.state_key(key) + + +def test_every_builder_maps_onto_itself_under_the_new_key(): + """The families agree with the builders that write them.""" + pipe = Pipeline(nodes=[], state=MemoryState()) + redis_state = RedisState(namespace="pipeline") + + def node_key(flow, key): + node = Node(f=lambda params: None, name="limit") + node.assign_flow(flow, "limit") + return node._state_key(key) + + builders = [ + lambda f: pipe._version_key(f"{f}.temp"), + lambda f: pipe._timestamp_key(f"{f}.temp"), + lambda f: pipe._held_key(f"{f}.temp"), + lambda f: pipe._flush_key(f"{f}.valve"), + lambda f: pipe._last_seen_key(f"{f}.valve", f"{f}.temp"), + lambda f: pipe._delivered_key("hall.lamp", f"{f}.temp"), + lambda f: node_key(f, "wait:heat.temp"), + lambda f: redis_state._history_key(f"{f}.temp").removeprefix("pipeline:"), + ] + + for build in builders: + assert HEAT.state_key(build("heat")) == build("warm") + redis_state.close() + + +def test_touches_reads_the_same_parts(): + assert touches("heat", "__last_seen__:hall.lamp:heat.temp") + assert not touches("heat", "__node__:hall.limit:wait:heat.temp") + assert not touches("heat", "heating.temp") + assert not touches("heat", "__names__") + + +def test_work_items_move_their_node_flow_and_output_names(): + own = WorkItem( + kind="flush", node="heat.valve", flow="heat", guard_key="wait:heat.temp" + ) + reaching = WorkItem( + kind="cascade", + node="hall.lamp", + flow="hall", + outputs={"heat.target": 21, "hall.on": True}, + ) + + assert HEAT.item(own) == dataclasses.replace(own, node="warm.valve", flow="warm") + assert HEAT.item(reaching) == dataclasses.replace( + reaching, outputs={"warm.target": 21, "hall.on": True} + ) + for item in (own, reaching): + assert HEAT.fields(item.to_fields()) == HEAT.item(item).to_fields() + + +def test_work_items_without_the_old_names_stay(): + # A run item names its flow by id, and the first ids were the keys. + run = WorkItem(kind="run", node="", flow="heat", run_id="r1") + other = WorkItem(kind="cascade", node="heating.valve", flow="heating") + fields = {**other.to_fields(), "not_before": "0"} + + assert HEAT.item(run) is None + assert HEAT.fields(run.to_fields()) is None + assert HEAT.item(other) is None + assert HEAT.fields(fields) is None + + +def test_alert_dedup_keys_follow_the_node_or_flow(): + def key(**event): + return HEAT.dedup_key(dedup_key(event)) + + assert key(type="node_error", node="heat.valve", flow="heat") == ( + "node_error:warm.valve" + ) + assert key(type="flow_quarantined", flow="heat") == "flow_quarantined:warm" + assert key(type="flow_quarantined", flow="heating") == "flow_quarantined:heating" + assert key(type="engine_degraded") == "engine_degraded:" + + +def test_done_markers_read_back(): + queue = SimpleNamespace(_ns="queue") + for entry_id in ("1700000000000-0", "due:1700000000000-0"): + key = RedisWorkQueue._done_key(queue, entry_id, "heat.valve") + assert done_marker(key) == (entry_id, "heat.valve") + + +FAMILY_LITERAL = re.compile(r"(__[a-z][a-z_]*__)(?::|$)") +#: What Python itself names this way, such as ``getattr(f, "__name__")``. +PYTHON = set(dir(type)) | set(dir(lambda: None)) + + +@pytest.mark.parametrize("module", [pipeline, base, state]) +def test_every_family_in_the_key_builders_is_registered(module): + """A key family written as a literal here, not registered in `namespace`, + would be left behind by a rename.""" + found = { + family + for node in ast.walk(ast.parse(inspect.getsource(module))) + if isinstance(node, ast.Constant) and isinstance(node.value, str) + for family in FAMILY_LITERAL.findall(node.value) + } - PYTHON + + assert found <= set(FAMILIES), found - set(FAMILIES) + + +def test_every_work_item_field_is_classified(): + fields = {field.name for field in dataclasses.fields(WorkItem)} + + assert fields == set(ITEM_FIELDS) + assert set(WorkItem(kind="cascade", node="a.b", flow="a").to_fields()) <= fields + assert set(ITEM_FIELDS.values()) <= { + None, + namespace.NAME, + namespace.FLOW, + namespace.OUTPUTS, + }