From 46b57de4ec770637874e84c91228e2dc53c6cdc9 Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 2 Nov 2026 19:50:54 +0100 Subject: [PATCH] Rename migrations for live state and the work queue: state keys move with RENAMENX in batches (type, TTL and history capacity kept), the queue rebuilds its streams with the same ids, group and pending entries, rewrites timers, parked and held items and done markers, and can hold new work in a list released in order; nothing calls them yet --- backend/fluksio/flow/queue.py | 387 +++++++++++++++++- backend/fluksio/flow/state.py | 115 +++++- backend/tests/flow/test_namespace.py | 19 +- backend/tests/flow/test_redis_streams.py | 85 +--- backend/tests/flow/test_rename_runtime.py | 458 ++++++++++++++++++++++ 5 files changed, 981 insertions(+), 83 deletions(-) create mode 100644 backend/tests/flow/test_rename_runtime.py diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index 44e31ec..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__) @@ -54,6 +55,23 @@ STALE_CONSUMER_IDLE_MS = 3_600_000 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 @@ -72,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: @@ -172,6 +208,94 @@ def done_marker(key: str) -> tuple[str, str]: 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.""" @@ -309,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.""" @@ -328,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, @@ -344,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() @@ -439,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: @@ -473,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. @@ -515,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() @@ -561,12 +807,16 @@ class RedisWorkQueue(WorkQueue): 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 @@ -795,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]: @@ -896,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 f313ea3..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,7 +17,7 @@ from typing import Any, cast import orjson import redis -from fluksio.flow.namespace import HISTORY, NAMES +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 @@ -36,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: @@ -50,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. @@ -289,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) @@ -459,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): """ @@ -843,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 index 2ccb45c..acea183 100644 --- a/backend/tests/flow/test_namespace.py +++ b/backend/tests/flow/test_namespace.py @@ -9,6 +9,7 @@ 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 @@ -144,10 +145,18 @@ FAMILY_LITERAL = re.compile(r"(__[a-z][a-z_]*__)(?::|$)") 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.""" +@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))) @@ -155,7 +164,7 @@ def test_every_family_in_the_key_builders_is_registered(module): for family in FAMILY_LITERAL.findall(node.value) } - PYTHON - assert found <= set(FAMILIES), found - set(FAMILIES) + assert found <= set(registry), found - set(registry) def test_every_work_item_field_is_classified(): 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