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..19c02f4 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -23,14 +23,15 @@ import threading import time from abc import ABC, abstractmethod from collections import OrderedDict, deque -from collections.abc import Callable +from collections.abc import Callable, Iterable from dataclasses import dataclass, field from typing import Any, cast import orjson import redis -from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S +from fluksio.flow.namespace import Rename +from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S, move_keys logger = logging.getLogger(__name__) @@ -52,6 +53,25 @@ 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__" +# How many entries a rename reads, or writes, per round trip. +RENAME_PAGE = 500 +#: Every structure under the Redis queue's namespace. `rename` rewrites the +#: names in all of them but the dead letters, which are history. One added +#: here holds names too, or has to say why it does not. +STRUCTURES = frozenset( + { + "__queue__", + "__due__", + "__delayed__", + "__parked__", + "__parked_flows__", + "__held__", + DONE, + "__dead__", + } +) # 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 @@ -70,6 +90,24 @@ end return #due """ +# Work held during a rename goes into the backlog oldest first, behind +# everything already there. As in `_MOVE_DUE`, an item leaves the held list +# only once its XADD went through. 100 per call, so Redis is never blocked +# for long. +_RELEASE = """ +local held = redis.call('LRANGE', KEYS[1], 0, 99) +for _, raw in ipairs(held) do + local fields = {} + for key, value in pairs(cjson.decode(raw)) do + fields[#fields + 1] = key + fields[#fields + 1] = value + end + redis.call('XADD', KEYS[2], 'MAXLEN', '~', ARGV[1], '*', unpack(fields)) + redis.call('LPOP', KEYS[1]) +end +return #held +""" + @dataclass class WorkItem: @@ -162,6 +200,102 @@ 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 + + +def _stream_id(entry_id: str) -> tuple[int, int]: + ms, _, seq = entry_id.partition("-") + return int(ms), int(seq) + + +def rebuild_stream(client: redis.Redis, stream: str, rename: Rename) -> None: + """Rewrite the names in the entries of *stream* still to run, keeping + every entry id, the group and its consumers, each pending entry's owner, + delivery count and idle time, and the stream's id counters. + + An entry cannot be changed in place, and one added again gets a new id, + which is what runs, events and done markers are filed under, and loses its + delivery count. So the stream is copied into a temporary key with its ids + and swapped in with one RENAME, the only commit point: a rerun after a + crash on either side of it converges. Acknowledged entries are history and + copied as they are. Nothing may write to the stream meanwhile. Needs Redis + 7.0 (docs/private/id_refac_rename.md, 3.1). + """ + tmp = f"{stream}:rename" + client.delete(tmp) + info = cast(dict[str, Any], client.xinfo_stream(stream)) + groups = cast(list[dict[str, Any]], client.xinfo_groups(stream)) + group = next(g for g in groups if g["name"] == GROUP) + last = str(group["last-delivered-id"]) + pel: list[dict[str, Any]] = [] + start = "-" + while rows := cast( + list[dict[str, Any]], + client.xpending_range(stream, GROUP, min=start, max="+", count=RENAME_PAGE), + ): + pel += rows + start = f"({rows[-1]['message_id']}" + pending = {row["message_id"] for row in pel} + + pipe = client.pipeline(transaction=False) + # MKSTREAM, so XSETID finds the key when there is nothing to copy. + pipe.xgroup_create(tmp, GROUP, id=last, mkstream=True) + for consumer in cast(list[dict[str, Any]], client.xinfo_consumers(stream, GROUP)): + pipe.xgroup_createconsumer(tmp, GROUP, consumer["name"]) + copied: set[str] = set() + start = "-" + while entries := cast( + list[tuple[str, dict[str, str]]], + client.xrange(stream, min=start, max="+", count=RENAME_PAGE), + ): + for entry_id, fields in entries: + live = entry_id in pending or _stream_id(entry_id) > _stream_id(last) + renamed = rename.fields(fields) if live else None + pipe.xadd(tmp, cast(Any, renamed or fields), id=entry_id) + if entry_id in pending: + copied.add(entry_id) + pipe.execute() + start = f"({entries[-1][0]}" + # redis-py has no XSETID of its own, and leaves the generic call untyped. + cast(Any, pipe).execute_command( + "XSETID", + tmp, + info["last-generated-id"], + "ENTRIESADDED", + info["entries-added"], + "MAXDELETEDID", + info["max-deleted-entry-id"], + ) + # Redis caps ENTRIESREAD at the stream's entries-added counter, which only + # XSETID has set. Unknown stays unknown. + if group["entries-read"] is not None: + pipe.xgroup_setid(tmp, GROUP, last, entries_read=group["entries-read"]) + # A pending entry whose stream entry was trimmed is not recreated, as + # `_reclaim` drops it too. XCLAIM would still create its consumer. + for row in pel: + if row["message_id"] in copied: + pipe.xclaim( + tmp, + GROUP, + row["consumer"], + 0, + [row["message_id"]], + idle=row["time_since_delivered"], + retrycount=row["times_delivered"], + force=True, + justid=True, + ) + pipe.execute() + # On its own: a pipeline without a transaction runs every command even + # after one of them failed. + client.rename(tmp, stream) + + class WorkQueue(ABC): """What the execution service pulls from.""" @@ -299,6 +433,52 @@ class WorkQueue(ABC): def close(self) -> None: """Release whatever the queue holds.""" + # ------------------------------------------------------------------------- + # Renaming a flow key + # ------------------------------------------------------------------------- + + @abstractmethod + def hold(self) -> None: + """Divert `add` into a held list until `release`, so a rename can + rewrite the queue while listeners keep journaling work.""" + + @abstractmethod + def release(self) -> None: + """Move the held work into the backlog, oldest first and behind + everything already there, and stop diverting.""" + + @property + @abstractmethod + def held(self) -> bool: + """Whether `add` is diverted.""" + + @abstractmethod + def release_held(self) -> int: + """Release a held list left by a process that stopped mid-rename. For + startup, before anything claims. Returns how many items moved.""" + + @abstractmethod + def rename( + self, rename: Rename, progress: Callable[[str], None] | None = None + ) -> None: + """Move the names in the work still to run from ``rename.old`` to + ``rename.new`` (`Rename.fields`). + + Queued, claimed, delayed, parked and held items, and the done markers + that keep a replayed side effect from repeating. Entry ids, order, + deadlines, delivery counts, owners and guards stay as they are. Dead + letters and acknowledged entries are history and stay too. A rerun + after a crash completes the job. Nothing may claim, promote, reclaim, + park or unpark meanwhile. + + :param progress: Called with the name of each structure once it is done. + """ + + @abstractmethod + def count(self, rename: Rename) -> dict[str, int]: + """How many entries `rename` would rewrite: ``queue_items`` (queued, + claimed or held), ``delayed``, ``parked`` and ``done_markers``.""" + class MemoryWorkQueue(WorkQueue): """In-process queue. No durability: a crash loses whatever is in flight.""" @@ -318,6 +498,9 @@ class MemoryWorkQueue(WorkQueue): self._done: OrderedDict[tuple[str, str], None] = OrderedDict() # Claimed and not yet acknowledged, which is what Redis's `pending` is. self._in_flight = 0 + # What was added while a rename held the queue (`hold`). + self._held: deque[WorkItem] = deque() + self._holding = False self._counter = 0 self._seq = 0 # The ids become `FlowRun.id`, a primary key. Counting from zero alone, @@ -334,6 +517,9 @@ class MemoryWorkQueue(WorkQueue): with self._wake: if not item.entry_id: item.entry_id = self._next_id() + if self._holding: + self._held.append(item) + return self._items.append(item) self._wake.notify() @@ -429,6 +615,7 @@ class MemoryWorkQueue(WorkQueue): *self._due, *(entry[2] for entry in self._delayed), *(item for held in self._parked.values() for item in held), + *self._held, ] def backlog(self) -> int: @@ -463,6 +650,66 @@ class MemoryWorkQueue(WorkQueue): def close(self) -> None: """Nothing is held outside the process.""" + def hold(self) -> None: + with self._lock: + self._holding = True + + def release(self) -> None: + with self._wake: + self._holding = False + self._items.extend(self._held) + self._held.clear() + self._wake.notify() + + @property + def held(self) -> bool: + return self._holding + + def release_held(self) -> int: + """Nothing held here outlives the process.""" + return 0 + + def rename( + self, rename: Rename, progress: Callable[[str], None] | None = None + ) -> None: + """One pass under the lock. Timers keep their place in the heap, + which orders them by deadline and arrival only.""" + + def moved(item: WorkItem) -> WorkItem: + return rename.item(item) or item + + with self._lock: + if rename.old in self._parked and rename.new in self._parked: + raise RuntimeError(f"Work is parked under '{rename.new}' already") + for lane in (self._items, self._due, self._held): + items = [moved(item) for item in lane] + lane.clear() + lane.extend(items) + self._delayed = [(at, n, moved(item)) for at, n, item in self._delayed] + self._parked = { + rename.flow(flow): [moved(item) for item in items] + for flow, items in self._parked.items() + } + self._done = OrderedDict( + ((entry_id, rename.name(node)), None) for entry_id, node in self._done + ) + if progress is not None: + progress("queue") + + def count(self, rename: Rename) -> dict[str, int]: + """What is claimed is gone from here, so it is not counted.""" + + def changed(items: Iterable[WorkItem]) -> int: + return sum(rename.item(item) is not None for item in items) + + with self._lock: + return { + "queue_items": changed([*self._items, *self._due, *self._held]), + "delayed": changed(entry[2] for entry in self._delayed), + "parked": changed(i for held in self._parked.values() for i in held), + "done_markers": sum(rename.name(n) != n for _, n in self._done), + } + class RedisWorkQueue(WorkQueue): """Redis Streams queue: the one that survives the process. @@ -505,7 +752,16 @@ class RedisWorkQueue(WorkQueue): # finding them with a keyspace scan walked every state and idempotency # key in the database, twice per health poll. self._parked_flows_key = f"{namespace}:__parked_flows__" + # What was added while a rename held the queue (`hold`). + self._held_key = f"{namespace}:__held__" self._move_due = self._redis.register_script(_MOVE_DUE) + self._release = self._redis.register_script(_RELEASE) + # `add` reads the flag and writes under this, and `release` empties the + # held list and clears the flag under it. So nothing added after `hold` + # lands in a stream a rename is rebuilding, and nothing added while + # held overtakes held work. + self._hold_lock = threading.Lock() + self._holding = False self._ensure_group() self._drop_stale_consumers() @@ -548,15 +804,19 @@ 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( - self._stream, - cast(Any, item.to_fields()), - maxlen=STREAM_MAXLEN, - approximate=True, - ) + with self._hold_lock: + if self._holding: + self._redis.rpush(self._held_key, orjson.dumps(item.to_fields())) + return + self._redis.xadd( + self._stream, + cast(Any, item.to_fields()), + maxlen=STREAM_MAXLEN, + approximate=True, + ) def add_delayed(self, item: WorkItem, not_before: float) -> None: item.not_before = not_before @@ -785,6 +1045,10 @@ class RedisWorkQueue(WorkQueue): list[str], self._redis.lrange(self._parked_key(flow), 0, -1) ) ] + items += [ + WorkItem.from_fields(orjson.loads(raw), "") + for raw in cast(list[str], self._redis.lrange(self._held_key, 0, -1)) + ] return items def _group_info(self, stream: str = "") -> dict[str, Any]: @@ -886,3 +1150,120 @@ class RedisWorkQueue(WorkQueue): def close(self) -> None: self._redis.close() + + def hold(self) -> None: + with self._hold_lock: + self._holding = True + + def release(self) -> None: + with self._hold_lock: + self.release_held() + self._holding = False + + @property + def held(self) -> bool: + return self._holding + + def release_held(self) -> int: + released = 0 + while moved := cast( + int, + self._release(keys=[self._held_key, self._stream], args=[STREAM_MAXLEN]), + ): + released += moved + return released + + def rename( + self, rename: Rename, progress: Callable[[str], None] | None = None + ) -> None: + """Each structure as section 3 of docs/private/id_refac_rename.md has + it: the streams rebuilt, then timers, lists and done markers, each + rewritten one member, slot or key at a time.""" + + def done(structure: str) -> None: + if progress is not None: + progress(structure) + + for stream, lane in ((self._due_stream, "due"), (self._stream, "backlog")): + rebuild_stream(self._redis, stream, rename) + done(lane) + + # One transaction, so each timer moves with its deadline. Promotion is + # held, and a timer added meanwhile is not touched. + pipe = self._redis.pipeline() + for raw, at in cast( + list[tuple[str, float]], + self._redis.zrange(self._delayed_key, 0, -1, withscores=True), + ): + fields = rename.fields(orjson.loads(raw)) + if fields is not None: + pipe.zrem(self._delayed_key, raw) + pipe.zadd(self._delayed_key, {orjson.dumps(fields): at}) + pipe.execute() + done("delayed") + + # Slot by slot: nothing pops while held, and `add` only appends, so + # the indices read hold. + for key in [*self._parked_keys(rename), self._held_key]: + pipe = self._redis.pipeline(transaction=False) + for i, raw in enumerate(cast(list[str], self._redis.lrange(key, 0, -1))): + fields = rename.fields(orjson.loads(raw)) + if fields is not None: + pipe.lset(key, i, orjson.dumps(fields).decode()) + pipe.execute() + old, new = self._parked_key(rename.old), self._parked_key(rename.new) + if self._redis.exists(old): + # Checked first: inside the transaction a refused RENAMENX would + # not stop the index from moving. + if self._redis.exists(new): + raise RuntimeError(f"Work is parked under '{rename.new}' already") + pipe = self._redis.pipeline() + pipe.renamenx(old, new) + pipe.srem(self._parked_flows_key, rename.old) + pipe.sadd(self._parked_flows_key, rename.new) + pipe.execute() + done("lists") + + move_keys(self._redis, self._marker_moves(rename)) + done("done markers") + + def _parked_keys(self, rename: Rename) -> list[str]: + """Every parked list, the renamed flow's under either name included.""" + flows = cast(set[str], self._redis.smembers(self._parked_flows_key)) + return list( + dict.fromkeys( + map(self._parked_key, [*sorted(flows), rename.old, rename.new]) + ) + ) + + def _marker_moves(self, rename: Rename) -> list[tuple[str, str]]: + moves = [] + for key in self._redis.scan_iter( + match=f"{self._ns}:{DONE}:*", count=RENAME_PAGE + ): + entry_id, node = done_marker(key) + if rename.name(node) != node: + moves.append((key, self._done_key(entry_id, rename.name(node)))) + return moves + + def count(self, rename: Rename) -> dict[str, int]: + def changed(raws: Iterable[str]) -> int: + return sum(rename.fields(orjson.loads(raw)) is not None for raw in raws) + + def listed(key: str) -> list[str]: + return cast(list[str], self._redis.lrange(key, 0, -1)) + + live = [ + fields + for stream in (self._due_stream, self._stream) + for _, fields in self._live_entries(stream) + ] + return { + "queue_items": sum(rename.fields(f) is not None for f in live) + + changed(listed(self._held_key)), + "delayed": changed( + cast(list[str], self._redis.zrange(self._delayed_key, 0, -1)) + ), + "parked": sum(changed(listed(key)) for key in self._parked_keys(rename)), + "done_markers": len(self._marker_moves(rename)), + } diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py index 23ed33a..74c8d8f 100644 --- a/backend/fluksio/flow/state.py +++ b/backend/fluksio/flow/state.py @@ -9,7 +9,7 @@ from __future__ import annotations from abc import ABC, abstractmethod from collections import deque -from collections.abc import Iterable, Iterator +from collections.abc import Callable, Iterable, Iterator from contextlib import contextmanager from threading import RLock from typing import Any, cast @@ -17,6 +17,8 @@ from typing import Any, cast import orjson import redis +from fluksio.flow.namespace import HISTORY, NAMES, Rename + #: 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 @@ -34,6 +36,9 @@ SOCKET_TIMEOUT_S = 5.0 CONNECT_TIMEOUT_S = 2.0 HISTORY_LIMIT = 120 +#: How many keys a rename moves per round trip, and between two calls of its +#: ``progress``, which is where the rename journal can stop. +RENAME_BATCH = 500 def as_number(value: Any) -> float | None: @@ -48,6 +53,30 @@ def as_number(value: Any) -> float | None: return float(value) +def move_keys(client: redis.Redis, moves: list[tuple[str, str]]) -> int: + """``RENAMENX`` each ``(key, target)`` in one round trip, which keeps the + Redis type and the remaining TTL. Returns how many moved. + + A key gone since it was listed has expired and is skipped. A target that + exists is refused rather than overwritten: a rename checks that nothing + lives under the new name before it starts, so one there now is a fault. + """ + pipe = client.pipeline(transaction=False) + for key, target in moves: + pipe.renamenx(key, target) + replies = pipe.execute(raise_on_error=False) + moved = 0 + for (key, target), reply in zip(moves, replies, strict=True): + if isinstance(reply, redis.ResponseError) and "no such key" in str(reply): + continue + if isinstance(reply, Exception): + raise reply + if not reply: + raise RuntimeError(f"Will not move '{key}' onto '{target}', which exists") + moved += 1 + return moved + + class StateBackend(ABC): """ Abstract base class for state storage backends. @@ -287,6 +316,32 @@ class StateBackend(ABC): """ ... + # ------------------------------------------------------------------------- + # Renaming a flow key + # ------------------------------------------------------------------------- + + @abstractmethod + def rename_keys( + self, rename: Rename, progress: Callable[[str], None] | None = None + ) -> int: + """Move every key that names something of ``rename.old`` to the new + name (`Rename.state_key`), with its history. + + Keys move, values do not: a message's payload and a node's private + state are its own, whatever names they mention. A key already under + the new name stays where it is, so a rerun after a crash finishes the + job. A target that exists is refused rather than overwritten. + + :param progress: Called after each batch of keys. + :returns: How many keys moved. + :raises ValueError: For a bookkeeping key of an unregistered family. + """ + ... + + def count(self, rename: Rename) -> int: + """How many keys `rename_keys` would move.""" + return sum(rename.state_key(key) is not None for key in self.keys()) + def __contains__(self, key: str) -> bool: return self.exists(key) @@ -457,6 +512,39 @@ class MemoryState(StateBackend): with self._lock: return list(self._history.get(key, ())) + def _moves(self, rename: Rename) -> tuple[dict[str, str], dict[str, str]]: + """Where the keys of values and of histories go. Histories are kept + by bare message name here.""" + data = {k: t for k in self._data if (t := rename.state_key(k)) is not None} + history = {k: rename.name(k) for k in self._history if rename.name(k) != k} + return data, history + + def rename_keys( + self, rename: Rename, progress: Callable[[str], None] | None = None + ) -> int: + """One pass under the lock. A history moves as the deque it is, so it + keeps its capacity.""" + with self._lock: + data, history = self._moves(rename) + for store, moves in ((self._data, data), (self._history, history)): + clash = next((t for t in moves.values() if t in store), None) + if clash is not None: + raise RuntimeError( + f"Will not move a key onto '{clash}', which exists" + ) + for key, target in data.items(): + self._data[target] = self._data.pop(key) + for key, target in history.items(): + self._history[target] = self._history.pop(key) + if progress is not None: + progress("keys") + return len(data) + len(history) + + def count(self, rename: Rename) -> int: + with self._lock: + data, history = self._moves(rename) + return len(data) + len(history) + class RedisState(StateBackend): """ @@ -533,7 +621,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 +858,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 @@ -841,3 +929,28 @@ class RedisState(StateBackend): ts, value = orjson.loads(entry) points.append((float(ts), float(value))) return points + + def rename_keys( + self, rename: Rename, progress: Callable[[str], None] | None = None + ) -> int: + """``RENAMENX`` in batches, then the names set in one transaction. + + The keys are listed before any moves, since a SCAN running over keys + that are being renamed can return one twice. + """ + moves = [(k, t) for k in self.keys() if (t := rename.state_key(k)) is not None] + moved = 0 + for start in range(0, len(moves), RENAME_BATCH): + batch = moves[start : start + RENAME_BATCH] + moved += move_keys( + self._client, [(self._key(k), self._key(t)) for k, t in batch] + ) + if progress is not None: + progress("keys") + names = [name for name in self.message_names() if rename.name(name) != name] + if names: + pipe = self._client.pipeline() + pipe.srem(self._names_key(), *names) + pipe.sadd(self._names_key(), *map(rename.name, names)) + pipe.execute() + return moved diff --git a/backend/tests/flow/test_namespace.py b/backend/tests/flow/test_namespace.py new file mode 100644 index 0000000..acea183 --- /dev/null +++ b/backend/tests/flow/test_namespace.py @@ -0,0 +1,180 @@ +"""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 import queue as queue_module +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", "registry"), + [ + (pipeline, FAMILIES), + (base, FAMILIES), + (state, FAMILIES), + (queue_module, queue_module.STRUCTURES), + ], +) +def test_every_family_in_the_key_builders_is_registered(module, registry): + """A key family written as a literal here, not registered in `namespace` + (the queue's in `queue.STRUCTURES`), 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(registry), found - set(registry) + + +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, + } diff --git a/backend/tests/flow/test_redis_streams.py b/backend/tests/flow/test_redis_streams.py index 31bf6b5..515a5e6 100644 --- a/backend/tests/flow/test_redis_streams.py +++ b/backend/tests/flow/test_redis_streams.py @@ -3,7 +3,7 @@ FLUKSIO_TEST_REDIS_HOST (and optional PORT). They prove on a real server that a stream copied into a temp key with its ids, group and pending entries and swapped in with RENAME looks like the original to -the engine (docs/private/id_refac_rename.md, 3.1). +the engine (`queue.rebuild_stream`, docs/private/id_refac_rename.md, 3.1). """ import os @@ -13,6 +13,9 @@ import uuid import pytest import redis +from fluksio.flow.namespace import Rename +from fluksio.flow.queue import rebuild_stream + GROUP = "engine" @@ -45,67 +48,7 @@ def _id(entry_id: str) -> tuple[int, int]: return int(ms), int(seq) -def _rename_flow(fields: dict[str, str]) -> dict[str, str]: - return {**fields, "flow": "new"} if fields.get("flow") == "old" else fields - - -# ponytail: unpaged. S3 moves it into RedisWorkQueue and pages XRANGE and XPENDING. -def rebuild(r, stream: str, rewrite, *, swap: bool = True) -> None: - """Copy *stream* into a temp key with the same ids, group and pending - entries, rewriting the entries still to run, and swap it in.""" - tmp = f"{stream}:rename" - r.delete(tmp) - info = r.xinfo_stream(stream) - (group,) = [g for g in r.xinfo_groups(stream) if g["name"] == GROUP] - last = group["last-delivered-id"] - pel = ( - r.xpending_range(stream, GROUP, "-", "+", count=group["pending"]) - if group["pending"] - else [] - ) - pending = {p["message_id"] for p in pel} - - pipe = r.pipeline(transaction=False) - # MKSTREAM so XSETID finds the key even when nothing is copied. - pipe.xgroup_create(tmp, GROUP, id=last, mkstream=True) - for consumer in r.xinfo_consumers(stream, GROUP): - pipe.xgroup_createconsumer(tmp, GROUP, consumer["name"]) - copied = set() - for entry_id, fields in r.xrange(stream): - live = entry_id in pending or _id(entry_id) > _id(last) - pipe.xadd(tmp, rewrite(fields) if live else fields, id=entry_id) - copied.add(entry_id) - pipe.execute_command( - "XSETID", - tmp, - info["last-generated-id"], - "ENTRIESADDED", - info["entries-added"], - "MAXDELETEDID", - info["max-deleted-entry-id"], - ) - # Redis caps ENTRIESREAD at the stream's entries-added counter, which is - # still the copied count before XSETID, so it is set only now. - if group["entries-read"] is not None: - pipe.execute_command( - "XGROUP", "SETID", tmp, GROUP, last, "ENTRIESREAD", group["entries-read"] - ) - for p in pel: - if p["message_id"] in copied: - pipe.xclaim( - tmp, - GROUP, - p["consumer"], - 0, - [p["message_id"]], - idle=p["time_since_delivered"], - retrycount=p["times_delivered"], - force=True, - justid=True, - ) - if swap: - pipe.rename(tmp, stream) - pipe.execute() +OLD_NEW = Rename("old", "new") def _pel(r, stream: str) -> list[dict]: @@ -162,7 +105,7 @@ def test_rebuild_keeps_ids_group_and_pending_entries(r, ns): idle_before = {p["message_id"]: p["time_since_delivered"] for p in _pel(r, stream)} started = time.monotonic() - rebuild(r, stream, _rename_flow) + rebuild_stream(r, stream, OLD_NEW) idle_after = {p["message_id"]: p["time_since_delivered"] for p in _pel(r, stream)} elapsed_ms = (time.monotonic() - started) * 1000 @@ -202,26 +145,32 @@ def test_rebuild_of_an_emptied_stream_keeps_its_id_counter(r, ns): r.xdel(stream, *ids) before = _snapshot(r, stream) - rebuild(r, stream, _rename_flow) + rebuild_stream(r, stream, OLD_NEW) assert _snapshot(r, stream) == before assert _id(r.xadd(stream, {"flow": "new"})) > _id(ids[-1]) -def test_a_rerun_after_a_crash_on_either_side_of_rename_converges(r, ns): +def test_a_rerun_after_a_crash_on_either_side_of_rename_converges(r, ns, monkeypatch): stream = f"{ns}:queue" _busy_stream(r, stream) original = r.xrange(stream) + def crash(*args): + raise redis.ConnectionError + # Crash before RENAME: the stream is untouched and the temp key is left. - rebuild(r, stream, _rename_flow, swap=False) + with monkeypatch.context() as patch: + patch.setattr(r, "rename", crash) + with pytest.raises(redis.ConnectionError): + rebuild_stream(r, stream, OLD_NEW) assert r.xrange(stream) == original assert r.exists(f"{stream}:rename") - rebuild(r, stream, _rename_flow) + rebuild_stream(r, stream, OLD_NEW) once = _snapshot(r, stream) # Crash after RENAME, before the journal checkpoint: the rerun changes nothing. - rebuild(r, stream, _rename_flow) + rebuild_stream(r, stream, OLD_NEW) assert _snapshot(r, stream) == once assert not r.exists(f"{stream}:rename") diff --git a/backend/tests/flow/test_rename_runtime.py b/backend/tests/flow/test_rename_runtime.py new file mode 100644 index 0000000..6ad5b87 --- /dev/null +++ b/backend/tests/flow/test_rename_runtime.py @@ -0,0 +1,458 @@ +"""Live state and queued work following a renamed flow key (sections 1.2, 1.3 +and 3 of docs/private/id_refac_rename.md). Memory always, Redis with +FLUKSIO_TEST_REDIS_HOST (and optional PORT).""" + +import os +import uuid + +import pytest +import redis + +from fluksio.flow import state as state_module +from fluksio.flow.namespace import HISTORY, NAMES, Rename +from fluksio.flow.queue import GROUP, MemoryWorkQueue, RedisWorkQueue, WorkItem +from fluksio.flow.state import MemoryState, RedisState + +HEAT = Rename("heat", "warm") +BACK = Rename("warm", "heat") +NOTHING = dict.fromkeys(("queue_items", "delayed", "parked", "done_markers"), 0) + + +def _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")) + + +def _drop(client, ns: str) -> None: + keys = list(client.scan_iter(match=f"{ns}:*")) + if keys: + client.delete(*keys) + + +def _redis_queue(ns: str, consumer: str) -> RedisWorkQueue: + host, port = _server() + return RedisWorkQueue(host, port, namespace=ns, consumer=consumer) + + +@pytest.fixture +def redis_queue(): + queue = _redis_queue(f"test-rename-{uuid.uuid4().hex}", "engine-a") + try: + yield queue + finally: + _drop(queue._redis, queue._ns) + queue.close() + + +@pytest.fixture(params=["memory", "redis"]) +def queue(request): + if request.param == "memory": + yield MemoryWorkQueue() + else: + yield request.getfixturevalue("redis_queue") + + +@pytest.fixture(params=["memory", "redis"]) +def state(request): + if request.param == "memory": + yield MemoryState() + return + host, port = _server() + state = RedisState(host, port, namespace=f"test-rename-{uuid.uuid4().hex}", ttl=600) + try: + yield state + finally: + _drop(state._client, state._namespace) + state.close() + + +def item(node: str, kind: str = "cascade", **fields) -> WorkItem: + return WorkItem(kind=kind, node=node, flow=node.split(".")[0], **fields) + + +def _fill(queue) -> WorkItem: + """Work in every place a queue keeps it, naming heat or not. Returns the + item left claimed.""" + queue.add(item("heat.acked")) + queue.ack(*queue.claim(1, 10)) + queue.add(item("heat.valve", outputs={"heat.temp": 21})) + queue.add(item("hall.lamp", outputs={"heat.target": 21, "hall.on": True})) + queue.add(item("heating.valve")) + queue.add(WorkItem(kind="run", node="", flow="heat", run_id="r1")) + queue.add_delayed(item("heat.pump", kind="flush"), 100.0) + queue.move_due(150.0) + (claimed,) = queue.claim(1, 10, due_only=True) + queue.mark_done(claimed.entry_id, "heat.pump") + queue.mark_done("other", "heating.valve") + queue.add_delayed(item("heat.limit", guard_key="generation", guard_value="3"), 9e9) + queue.add_delayed(item("heating.limit"), 9e9 + 1) + queue.park("heat", item("heat.p")) + queue.park("hall", item("hall.q", outputs={"heat.x": 1})) + queue.dead_letter(item("heat.dead"), "3 deliveries") + return claimed + + +def _dump(queue) -> object: + """Everything a queue holds, apart from how long it has been idle.""" + if isinstance(queue, MemoryWorkQueue): + return ( + list(queue._items), + list(queue._due), + list(queue._delayed), + dict(queue._parked), + list(queue._done), + list(queue._held), + ) + r = queue._redis + dump = {} + for key in sorted(r.scan_iter(match=f"{queue._ns}:*")): + kind = r.type(key) + if kind == "stream": + info = r.xinfo_stream(key) + groups = r.xinfo_groups(key) + dump[key] = ( + r.xrange(key), + [info[k] for k in ("last-generated-id", "entries-added")], + [ + {k: g[k] for k in ("last-delivered-id", "entries-read", "lag")} + for g in groups + ], + [ + (p["message_id"], p["consumer"], p["times_delivered"]) + for p in r.xpending_range(key, GROUP, "-", "+", 100) + ] + if groups + else [], + ) + elif kind == "zset": + dump[key] = r.zrange(key, 0, -1, withscores=True) + elif kind == "list": + dump[key] = r.lrange(key, 0, -1) + elif kind == "set": + dump[key] = sorted(r.smembers(key)) + else: + dump[key] = (r.get(key), r.ttl(key) > 3500) + return dump + + +def test_queued_work_moves_to_the_new_names(queue): + claimed = _fill(queue) + before = queue.live_items() + # Memory hands a claimed item out and forgets it. + in_stream = 3 if isinstance(queue, RedisWorkQueue) else 2 + assert queue.count(HEAT) == { + "queue_items": in_stream, + "delayed": 1, + "parked": 2, + "done_markers": 1, + } + + queue.rename(HEAT) + + def order(one): + return one.entry_id, one.node, one.enqueued_at + + # Ids, deadlines, guards and everything else but the names stay. + assert sorted(queue.live_items(), key=order) == sorted( + [HEAT.item(one) or one for one in before], key=order + ) + assert queue.count(HEAT) == NOTHING + assert queue.was_done(claimed.entry_id, "warm.pump") + assert not queue.was_done(claimed.entry_id, "heat.pump") + assert queue.was_done("other", "heating.valve") + assert queue.next_due() == 9e9 + assert queue.move_due(9e9) == 1 + assert [one.node for one in queue.claim(10, 10)] == [ + "warm.limit", + "warm.valve", + "hall.lamp", + "heating.valve", + "", + ] + assert queue.unpark("heat") == [] + assert [one.node for one in queue.unpark("warm")] == ["warm.p"] + assert [one.outputs for one in queue.unpark("hall")] == [{"warm.x": 1}] + if isinstance(queue, RedisWorkQueue): + # History stays as it was. + assert [d["node"] for d in queue.dead_letters()] == ["heat.dead"] + assert queue._redis.xrange(queue._stream)[0][1]["node"] == "heat.acked" + + +def test_a_second_run_changes_nothing_and_renaming_back_restores_the_queue(queue): + _fill(queue) + queue.hold() + queue.add(item("heat.held")) + before = _dump(queue) + + queue.rename(HEAT) + once = _dump(queue) + queue.rename(HEAT) + + assert once != before + assert _dump(queue) == once + queue.rename(BACK) + assert _dump(queue) == before + + +def test_held_work_is_released_in_order_behind_older_work(queue): + queue.add(item("hall.a")) + queue.hold() + queue.add(item("heat.b")) + queue.add(item("hall.c")) + + assert queue.held + assert queue.count(HEAT)["queue_items"] == 1 + assert {one.node for one in queue.live_items()} == {"hall.a", "heat.b", "hall.c"} + queue.rename(HEAT) + queue.release() + queue.add(item("hall.d")) + + assert not queue.held + assert [one.node for one in queue.claim(10, 10)] == [ + "hall.a", + "warm.b", + "hall.c", + "hall.d", + ] + + +def test_a_guarded_timer_still_finds_its_guard(queue): + state = MemoryState() + state["__node__:heat.limit:generation"] = 3 + queue.add_delayed(item("heat.limit", guard_key="generation", guard_value="3"), 1.0) + + queue.rename(HEAT) + state.rename_keys(HEAT) + queue.move_due(1.0) + + (due,) = queue.claim(1, 10) + assert (due.node, due.guard_key, due.guard_value) == ( + "warm.limit", + "generation", + "3", + ) + # What the execution service compares it with (`Node.recall`). + assert str(state.get(f"__node__:{due.node}:{due.guard_key}")) == due.guard_value + + +def test_a_pending_item_keeps_its_id_owner_and_count_and_is_not_repeated( + redis_queue, +): + redis_queue.add(item("heat.valve")) + (claimed,) = redis_queue.claim(1, 10) + # A second delivery, as a reclaim after a crash would make it. + redis_queue._redis.xclaim( + redis_queue._stream, GROUP, "engine-a", 0, [claimed.entry_id] + ) + redis_queue.mark_done(claimed.entry_id, "heat.valve") + + redis_queue.rename(HEAT) + + (row,) = redis_queue._redis.xpending_range(redis_queue._stream, GROUP, "-", "+", 10) + assert (row["message_id"], row["consumer"], row["times_delivered"]) == ( + claimed.entry_id, + "engine-a", + 2, + ) + other = _redis_queue(redis_queue._ns, "engine-b") + try: + (back,) = other.reclaim_stale(0) + # What the pipeline asks before it runs a side effect again. + assert other.was_done(back.entry_id, back.node) + finally: + other.close() + assert (back.entry_id, back.node, back.deliveries) == ( + claimed.entry_id, + "warm.valve", + 3, + ) + marker = redis_queue._done_key(claimed.entry_id, "warm.valve") + assert 3500 < redis_queue._redis.ttl(marker) <= 3600 + + +def test_a_crash_before_a_stream_is_swapped_in_and_a_rerun_converge( + redis_queue, monkeypatch +): + _fill(redis_queue) + before = _dump(redis_queue) + + def crash(*args): + raise redis.ConnectionError + + with monkeypatch.context() as patch: + patch.setattr(redis_queue._redis, "rename", crash) + with pytest.raises(redis.ConnectionError): + redis_queue.rename(HEAT) + after = _dump(redis_queue) + assert after.pop(f"{redis_queue._due_stream}:rename") + assert after == before + + redis_queue.rename(HEAT) + assert redis_queue.count(HEAT) == NOTHING + redis_queue.rename(BACK) + assert _dump(redis_queue) == before + + +def test_work_held_while_the_held_list_is_rewritten_survives(redis_queue, monkeypatch): + redis_queue.hold() + redis_queue.add(item("heat.a")) + redis_queue.add(item("hall.b")) + lrange = redis_queue._redis.lrange + + def racing(key, *args): + raw = lrange(key, *args) + if key == redis_queue._held_key and len(raw) == 2: + redis_queue.add(item("heat.c")) + return raw + + with monkeypatch.context() as patch: + patch.setattr(redis_queue._redis, "lrange", racing) + redis_queue.rename(HEAT) + redis_queue.release() + + assert [one.node for one in redis_queue.claim(10, 10)] == [ + "warm.a", + "hall.b", + "heat.c", + ] + + +def test_a_held_list_left_by_a_stopped_process_is_released_at_startup( + redis_queue, +): + redis_queue.hold() + for i in range(150): + redis_queue.add(item(f"hall.n{i}")) + + fresh = _redis_queue(redis_queue._ns, "engine-b") + try: + assert not fresh.held + assert fresh.release_held() == 150 + assert fresh.release_held() == 0 + claimed = fresh.claim(200, 10) + finally: + fresh.close() + assert [one.node for one in claimed] == [f"hall.n{i}" for i in range(150)] + + +def _fill_state(state) -> None: + state.record( + {"heat.temp": 21.5, "hall.temp": 19.0, "heating.temp": 30.0}, + {"__ts__:heat.temp": 1.0, "__ts__:hall.temp": 1.0}, + ["__version__:heat.temp", "__version__:hall.temp"], + ts=1.0, + limits={"heat.temp": 5}, + ) + for i in range(2, 8): + state.append_history({"heat.temp": float(i)}, float(i), {"heat.temp": 5}) + state.update( + { + "__last_seen__:heat.valve:hall.temp": 1, + "__last_seen__:hall.lamp:heat.temp": 2, + "__in_ts__:heat.valve:heat.temp": 3.0, + "__held__:heat.out": {"heat.temp": 1}, + "__flush__:heat.valve": 4.0, + # A node's own state may mention any name: only its key moves. + "__node__:heat.limit:wait:heat.temp": {"last": "heat.temp"}, + "__node__:hall.limit:wait:heat.temp": 5, + } + ) + + +def _state_dump(state) -> object: + keys = sorted(state.keys()) + names = sorted(state.message_names()) + return ( + keys, + {k: state.get(k) for k in keys if not k.startswith((HISTORY, NAMES))}, + names, + {name: state.history(name) for name in names}, + ) + + +def test_state_moves_to_the_new_names_with_its_type_ttl_and_capacity(state): + _fill_state(state) + + assert state.count(HEAT) == 10 + assert state.rename_keys(HEAT) == 10 + + assert state.count(HEAT) == 0 + assert sorted(state.message_names()) == ["hall.temp", "heating.temp", "warm.temp"] + assert not state.exists("heat.temp") + assert state.get_multi( + [ + "warm.temp", + "__version__:warm.temp", + "__last_seen__:warm.valve:hall.temp", + "__last_seen__:hall.lamp:warm.temp", + "__in_ts__:warm.valve:warm.temp", + "__held__:warm.out", + "__flush__:warm.valve", + "__node__:warm.limit:wait:heat.temp", + "__node__:hall.limit:wait:heat.temp", + ] + ) == { + "warm.temp": 21.5, + "__version__:warm.temp": 1, + "__last_seen__:warm.valve:hall.temp": 1, + "__last_seen__:hall.lamp:warm.temp": 2, + "__in_ts__:warm.valve:warm.temp": 3.0, + "__held__:warm.out": {"heat.temp": 1}, + "__flush__:warm.valve": 4.0, + "__node__:warm.limit:wait:heat.temp": {"last": "heat.temp"}, + "__node__:hall.limit:wait:heat.temp": 5, + } + assert state.history("warm.temp") == [(float(i), float(i)) for i in range(3, 8)] + if isinstance(state, MemoryState): + assert state._history["warm.temp"].maxlen == 5 + else: + history = state._key(f"{HISTORY}:warm.temp") + assert state._client.type(history) == b"list" + for key in ( + history, + state._key("warm.temp"), + state._key("__flush__:warm.valve"), + ): + assert 590 < state._client.ttl(key) <= 600 + + +def test_renaming_state_twice_changes_nothing_and_back_restores_it(state): + _fill_state(state) + before = _state_dump(state) + + state.rename_keys(HEAT) + once = _state_dump(state) + + assert state.rename_keys(HEAT) == 0 + assert _state_dump(state) == once + state.rename_keys(BACK) + assert _state_dump(state) == before + + +def test_a_state_rename_stopped_after_a_batch_completes_on_a_rerun(state, monkeypatch): + monkeypatch.setattr(state_module, "RENAME_BATCH", 3) + _fill_state(state) + before = _state_dump(state) + + def stop(_step): + raise ConnectionError + + with pytest.raises(ConnectionError): + state.rename_keys(HEAT, progress=stop) + state.rename_keys(HEAT) + + assert state.count(HEAT) == 0 + assert sorted(state.message_names()) == ["hall.temp", "heating.temp", "warm.temp"] + state.rename_keys(BACK) + assert _state_dump(state) == before + + +def test_state_is_never_moved_onto_a_key_that_exists(state): + state["heat.temp"] = 1 + state["warm.temp"] = 2 + + with pytest.raises(RuntimeError, match="exists"): + state.rename_keys(HEAT) + assert state.get("warm.temp") == 2