Rename mapper: flow/namespace.py maps state keys, work items and alert dedup keys from an old flow key to a new one; the state key families and WorkItem fields are registered there, the key builders take their prefixes from it, and done markers read back with queue.done_marker

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 ``<key>.<name>``, 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, ``<flow key>.<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 ``<prefix>:<part>:<part>``. 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 ``<event type>:<node or flow>`` (`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()}

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."""

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

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 ``<namespace>:__done__:<entry id>:<node>``.
+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(

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

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,
+    }