Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
stroblme
2026-11-02 19:50:54 +01:00
parent ebcc480a8c
commit 46b57de4ec
5 changed files with 981 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/flow/queue.py
+379 -8
View File
@@ -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 ``<namespace>:__done__:<entry id>:<node>``.
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)),
}
backend/fluksio/flow/state.py
+113 -2
View File
@@ -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
backend/tests/flow/test_namespace.py
+14 -5
View File
@@ -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():
backend/tests/flow/test_redis_streams.py
+17 -68
View File
@@ -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")
backend/tests/flow/test_rename_runtime.py
+458
View File
@@ -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