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