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

Merge branch 'rename-s2'

Browse Source
This commit is contained in:
stroblme
2026-11-02 19:58:45 +01:00
parent aa95245456 46b57de4ec
commit be18a94089
8 changed files with 1364 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/flow/namespace.py
+187
View File
@@ -0,0 +1,187 @@
"""The names the engine keeps live state and queued work under, and how a
rename of a flow key maps them.
A flow's messages and nodes are named ``<key>.<name>``, and the engine's own
bookkeeping on them sits in state beside the values, under a reserved prefix
and the names it is about: ``__version__:heat.temp``,
``__last_seen__:heat.valve:hall.temp``. Renaming a flow key moves all of it to
the new name, so every family is registered here, beside the builders that
take their prefix from it, and a key of a family that is not is refused rather
than left behind.
Nothing here does I/O. It maps names; moving the keys and items is the state
and queue backends' work.
"""
from __future__ import annotations
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, Any
import orjson
from fluksio.flow.messages import flow_of
if TYPE_CHECKING:
from fluksio.flow.queue import WorkItem
#: The message names in state, as a set (`RedisState` only).
NAMES = "__names__"
#: A message's recorded points (`RedisState`; memory keeps them apart).
HISTORY = "__history__"
#: How often a message was published.
VERSION = "__version__"
#: When a message was last published.
TIMESTAMP = "__ts__"
#: The version of a message a node last ran on.
LAST_SEEN = "__last_seen__"
#: When a rate-limited input last woke a node.
DELIVERED = "__in_ts__"
#: The value a rate-limited output kept back.
HELD = "__held__"
#: When a node's rate-limit window is due to be let through.
FLUSH = "__flush__"
#: What an engine-shipped node type remembers (`Node.remember`).
NODE_STATE = "__node__"
#: A qualified message or node name, ``<flow key>.<name>``.
NAME = "name"
#: A node's own key, user data that may contain colons.
PRIVATE = "private"
#: A flow key.
FLOW = "flow"
#: A map keyed by qualified message names.
OUTPUTS = "outputs"
#: Every bookkeeping family in live state, by prefix, with the parts that
#: follow it. A key is ``<prefix>:<part>:<part>``. Names never contain a colon,
#: so the key splits at its first ones and the last part takes the rest, which
#: is how a private key keeps its own.
FAMILIES: dict[str, tuple[str, ...]] = {
NAMES: (),
HISTORY: (NAME,),
VERSION: (NAME,),
TIMESTAMP: (NAME,),
LAST_SEEN: (NAME, NAME),
DELIVERED: (NAME, NAME),
HELD: (NAME,),
FLUSH: (NAME,),
NODE_STATE: (NAME, PRIVATE),
}
#: What each `WorkItem` field holds that a rename changes, or None. A guard is
#: the node's own key, and the state it guards moves with the node.
ITEM_FIELDS: dict[str, str | None] = {
"kind": None,
"node": NAME,
"flow": FLOW,
"outputs": OUTPUTS,
"cause": None,
"not_before": None,
"guard_key": None,
"guard_value": None,
"entry_id": None,
"deliveries": None,
"enqueued_at": None,
"run_id": None,
"parent": None,
}
def _parts(key: str) -> tuple[str, list[tuple[str, str]]]:
"""A state key's family prefix and its parts, each with what it holds.
A key without a prefix is a message value, and its family is ``""``.
"""
if not key.startswith("__"):
return "", [(key, NAME)]
family, sep, rest = key.partition(":")
shape = FAMILIES.get(family)
parts = rest.split(":", len(shape) - 1) if shape is not None and sep else []
if shape is None or len(parts) != len(shape):
raise ValueError(f"State key '{key}' is in no registered family")
return family, list(zip(parts, shape, strict=True))
def touches(flow: str, key: str) -> bool:
"""Whether the state key *key* names anything of the flow key *flow*."""
_, parts = _parts(key)
return any(kind == NAME and flow_of(part) == flow for part, kind in parts)
@dataclass(frozen=True)
class Rename:
"""The flow key *old* becoming *new*, applied to the names that embed it."""
old: str
new: str
def name(self, qualified: str) -> str:
"""A qualified message or node name in the new namespace.
Only the part before the first dot is the flow, so renaming ``heat``
leaves ``heating.temp`` alone. A bare key is its own namespace.
"""
if flow_of(qualified) != self.old:
return qualified
return self.new + qualified[len(self.old) :]
def flow(self, key: str) -> str:
return self.new if key == self.old else key
def state_key(self, key: str) -> str | None:
"""Where the state key *key* moves to, or None if it stays.
:raises ValueError: For a bookkeeping key of an unregistered family.
"""
family, parts = _parts(key)
mapped = [self.name(part) if kind == NAME else part for part, kind in parts]
renamed = ":".join([family, *mapped]) if family else mapped[0]
return renamed if renamed != key else None
def item(self, item: WorkItem) -> WorkItem | None:
"""*item* with its names in the new namespace, or None if it has none.
A ``run`` item names its flow by id, which a rename leaves as it is.
"""
if item.kind == "run":
return None
changes = {}
for field, holds in ITEM_FIELDS.items():
if holds is not None:
value = getattr(item, field)
mapped = self._mapped(holds, value)
if mapped != value:
changes[field] = mapped
return replace(item, **changes) if changes else None
def fields(self, fields: dict[str, str]) -> dict[str, str] | None:
"""A serialized item (`WorkItem.to_fields`) in the new namespace, or
None if it names nothing renamed. Fields a rename does not touch keep
their exact text, so an unchanged item serializes as it did."""
if fields.get("kind") == "run":
return None
renamed = dict(fields)
for field, holds in ITEM_FIELDS.items():
if holds is None or field not in fields:
continue
if holds == OUTPUTS:
outputs = orjson.loads(fields[field] or "{}")
mapped = self._mapped(holds, outputs)
if mapped != outputs:
renamed[field] = orjson.dumps(mapped).decode()
else:
renamed[field] = self._mapped(holds, fields[field])
return renamed if renamed != fields else None
def dedup_key(self, key: str) -> str:
"""An alert's ``<event type>:<node or flow>`` (`alerts.dedup_key`)."""
event, sep, about = key.partition(":")
return f"{event}{sep}{self.name(about)}"
def _mapped(self, holds: str, value: Any) -> Any:
if holds == NAME:
return self.name(value)
if holds == FLOW:
return self.flow(value)
return {self.name(name): one for name, one in value.items()}
backend/fluksio/flow/nodes/base.py
+2 -1
View File
@@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Any
from fluksio.flow import logs
from fluksio.flow.messages import MessageSpec, qualify
from fluksio.flow.namespace import NODE_STATE
from fluksio.flow.supervision import Supervisor
if TYPE_CHECKING:
@@ -283,7 +284,7 @@ class Node:
# -------------------------------------------------------------------------
def _state_key(self, key: str) -> str:
return f"__node__:{self.id}:{key}"
return f"{NODE_STATE}:{self.id}:{key}"
def remember(self, key: str, value: Any) -> None:
"""Keep a value for the next run of this node."""
backend/fluksio/flow/pipeline.py
+14 -6
View File
@@ -34,6 +34,14 @@ from fluksio.flow import logs
from fluksio.flow.artifacts import is_reference
from fluksio.flow.events import EventBus
from fluksio.flow.messages import flow_of, requalify
from fluksio.flow.namespace import (
DELIVERED,
FLUSH,
HELD,
LAST_SEEN,
TIMESTAMP,
VERSION,
)
from fluksio.flow.nodes import Node
from fluksio.flow.queue import WorkQueue
from fluksio.flow.state import MemoryState, StateBackend
@@ -633,25 +641,25 @@ class Pipeline:
# -------------------------------------------------------------------------
def _version_key(self, msg_name: str) -> str:
return f"__version__:{msg_name}"
return f"{VERSION}:{msg_name}"
def _last_seen_key(self, node_name: str, msg_name: str) -> str:
return f"__last_seen__:{node_name}:{msg_name}"
return f"{LAST_SEEN}:{node_name}:{msg_name}"
def _timestamp_key(self, msg_name: str) -> str:
return f"__ts__:{msg_name}"
return f"{TIMESTAMP}:{msg_name}"
def _delivered_key(self, node_name: str, msg_name: str) -> str:
"""When a rate-limited input last woke this node."""
return f"__in_ts__:{node_name}:{msg_name}"
return f"{DELIVERED}:{node_name}:{msg_name}"
def _held_key(self, msg_name: str) -> str:
"""The value a rate-limited output kept back, waiting for its window."""
return f"__held__:{msg_name}"
return f"{HELD}:{msg_name}"
def _flush_key(self, node_name: str) -> str:
"""When a rate-limit window of this node is due to be let through."""
return f"__flush__:{node_name}"
return f"{FLUSH}:{node_name}"
def _window_split(
self, node: Node, result: dict[str, Any], now: float
backend/fluksio/flow/queue.py
+390 -9
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__)
@@ -52,6 +53,25 @@ STALE_CONSUMER_IDLE_MS = 3_600_000
# The Redis one expires each after an hour; this is the same idea sized by
# count, since redelivery happens seconds after the claim and never later.
DONE_MARKERS = 10_000
# The Redis queue keeps them as ``<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
@@ -70,6 +90,24 @@ end
return #due
"""
# Work held during a rename goes into the backlog oldest first, behind
# everything already there. As in `_MOVE_DUE`, an item leaves the held list
# only once its XADD went through. 100 per call, so Redis is never blocked
# for long.
_RELEASE = """
local held = redis.call('LRANGE', KEYS[1], 0, 99)
for _, raw in ipairs(held) do
local fields = {}
for key, value in pairs(cjson.decode(raw)) do
fields[#fields + 1] = key
fields[#fields + 1] = value
end
redis.call('XADD', KEYS[2], 'MAXLEN', '~', ARGV[1], '*', unpack(fields))
redis.call('LPOP', KEYS[1])
end
return #held
"""
@dataclass
class WorkItem:
@@ -162,6 +200,102 @@ def owned_by(item: WorkItem, flow: str) -> bool:
)
def done_marker(key: str) -> tuple[str, str]:
"""The entry id and node a done marker is for: `RedisWorkQueue._done_key`
read back. Split at the last colon, since an entry id from the due lane
has one of its own and a node id never does."""
entry_id, node = key.split(f":{DONE}:", 1)[1].rsplit(":", 1)
return entry_id, node
def _stream_id(entry_id: str) -> tuple[int, int]:
ms, _, seq = entry_id.partition("-")
return int(ms), int(seq)
def rebuild_stream(client: redis.Redis, stream: str, rename: Rename) -> None:
"""Rewrite the names in the entries of *stream* still to run, keeping
every entry id, the group and its consumers, each pending entry's owner,
delivery count and idle time, and the stream's id counters.
An entry cannot be changed in place, and one added again gets a new id,
which is what runs, events and done markers are filed under, and loses its
delivery count. So the stream is copied into a temporary key with its ids
and swapped in with one RENAME, the only commit point: a rerun after a
crash on either side of it converges. Acknowledged entries are history and
copied as they are. Nothing may write to the stream meanwhile. Needs Redis
7.0 (docs/private/id_refac_rename.md, 3.1).
"""
tmp = f"{stream}:rename"
client.delete(tmp)
info = cast(dict[str, Any], client.xinfo_stream(stream))
groups = cast(list[dict[str, Any]], client.xinfo_groups(stream))
group = next(g for g in groups if g["name"] == GROUP)
last = str(group["last-delivered-id"])
pel: list[dict[str, Any]] = []
start = "-"
while rows := cast(
list[dict[str, Any]],
client.xpending_range(stream, GROUP, min=start, max="+", count=RENAME_PAGE),
):
pel += rows
start = f"({rows[-1]['message_id']}"
pending = {row["message_id"] for row in pel}
pipe = client.pipeline(transaction=False)
# MKSTREAM, so XSETID finds the key when there is nothing to copy.
pipe.xgroup_create(tmp, GROUP, id=last, mkstream=True)
for consumer in cast(list[dict[str, Any]], client.xinfo_consumers(stream, GROUP)):
pipe.xgroup_createconsumer(tmp, GROUP, consumer["name"])
copied: set[str] = set()
start = "-"
while entries := cast(
list[tuple[str, dict[str, str]]],
client.xrange(stream, min=start, max="+", count=RENAME_PAGE),
):
for entry_id, fields in entries:
live = entry_id in pending or _stream_id(entry_id) > _stream_id(last)
renamed = rename.fields(fields) if live else None
pipe.xadd(tmp, cast(Any, renamed or fields), id=entry_id)
if entry_id in pending:
copied.add(entry_id)
pipe.execute()
start = f"({entries[-1][0]}"
# redis-py has no XSETID of its own, and leaves the generic call untyped.
cast(Any, pipe).execute_command(
"XSETID",
tmp,
info["last-generated-id"],
"ENTRIESADDED",
info["entries-added"],
"MAXDELETEDID",
info["max-deleted-entry-id"],
)
# Redis caps ENTRIESREAD at the stream's entries-added counter, which only
# XSETID has set. Unknown stays unknown.
if group["entries-read"] is not None:
pipe.xgroup_setid(tmp, GROUP, last, entries_read=group["entries-read"])
# A pending entry whose stream entry was trimmed is not recreated, as
# `_reclaim` drops it too. XCLAIM would still create its consumer.
for row in pel:
if row["message_id"] in copied:
pipe.xclaim(
tmp,
GROUP,
row["consumer"],
0,
[row["message_id"]],
idle=row["time_since_delivered"],
retrycount=row["times_delivered"],
force=True,
justid=True,
)
pipe.execute()
# On its own: a pipeline without a transaction runs every command even
# after one of them failed.
client.rename(tmp, stream)
class WorkQueue(ABC):
"""What the execution service pulls from."""
@@ -299,6 +433,52 @@ class WorkQueue(ABC):
def close(self) -> None:
"""Release whatever the queue holds."""
# -------------------------------------------------------------------------
# Renaming a flow key
# -------------------------------------------------------------------------
@abstractmethod
def hold(self) -> None:
"""Divert `add` into a held list until `release`, so a rename can
rewrite the queue while listeners keep journaling work."""
@abstractmethod
def release(self) -> None:
"""Move the held work into the backlog, oldest first and behind
everything already there, and stop diverting."""
@property
@abstractmethod
def held(self) -> bool:
"""Whether `add` is diverted."""
@abstractmethod
def release_held(self) -> int:
"""Release a held list left by a process that stopped mid-rename. For
startup, before anything claims. Returns how many items moved."""
@abstractmethod
def rename(
self, rename: Rename, progress: Callable[[str], None] | None = None
) -> None:
"""Move the names in the work still to run from ``rename.old`` to
``rename.new`` (`Rename.fields`).
Queued, claimed, delayed, parked and held items, and the done markers
that keep a replayed side effect from repeating. Entry ids, order,
deadlines, delivery counts, owners and guards stay as they are. Dead
letters and acknowledged entries are history and stay too. A rerun
after a crash completes the job. Nothing may claim, promote, reclaim,
park or unpark meanwhile.
:param progress: Called with the name of each structure once it is done.
"""
@abstractmethod
def count(self, rename: Rename) -> dict[str, int]:
"""How many entries `rename` would rewrite: ``queue_items`` (queued,
claimed or held), ``delayed``, ``parked`` and ``done_markers``."""
class MemoryWorkQueue(WorkQueue):
"""In-process queue. No durability: a crash loses whatever is in flight."""
@@ -318,6 +498,9 @@ class MemoryWorkQueue(WorkQueue):
self._done: OrderedDict[tuple[str, str], None] = OrderedDict()
# Claimed and not yet acknowledged, which is what Redis's `pending` is.
self._in_flight = 0
# What was added while a rename held the queue (`hold`).
self._held: deque[WorkItem] = deque()
self._holding = False
self._counter = 0
self._seq = 0
# The ids become `FlowRun.id`, a primary key. Counting from zero alone,
@@ -334,6 +517,9 @@ class MemoryWorkQueue(WorkQueue):
with self._wake:
if not item.entry_id:
item.entry_id = self._next_id()
if self._holding:
self._held.append(item)
return
self._items.append(item)
self._wake.notify()
@@ -429,6 +615,7 @@ class MemoryWorkQueue(WorkQueue):
*self._due,
*(entry[2] for entry in self._delayed),
*(item for held in self._parked.values() for item in held),
*self._held,
]
def backlog(self) -> int:
@@ -463,6 +650,66 @@ class MemoryWorkQueue(WorkQueue):
def close(self) -> None:
"""Nothing is held outside the process."""
def hold(self) -> None:
with self._lock:
self._holding = True
def release(self) -> None:
with self._wake:
self._holding = False
self._items.extend(self._held)
self._held.clear()
self._wake.notify()
@property
def held(self) -> bool:
return self._holding
def release_held(self) -> int:
"""Nothing held here outlives the process."""
return 0
def rename(
self, rename: Rename, progress: Callable[[str], None] | None = None
) -> None:
"""One pass under the lock. Timers keep their place in the heap,
which orders them by deadline and arrival only."""
def moved(item: WorkItem) -> WorkItem:
return rename.item(item) or item
with self._lock:
if rename.old in self._parked and rename.new in self._parked:
raise RuntimeError(f"Work is parked under '{rename.new}' already")
for lane in (self._items, self._due, self._held):
items = [moved(item) for item in lane]
lane.clear()
lane.extend(items)
self._delayed = [(at, n, moved(item)) for at, n, item in self._delayed]
self._parked = {
rename.flow(flow): [moved(item) for item in items]
for flow, items in self._parked.items()
}
self._done = OrderedDict(
((entry_id, rename.name(node)), None) for entry_id, node in self._done
)
if progress is not None:
progress("queue")
def count(self, rename: Rename) -> dict[str, int]:
"""What is claimed is gone from here, so it is not counted."""
def changed(items: Iterable[WorkItem]) -> int:
return sum(rename.item(item) is not None for item in items)
with self._lock:
return {
"queue_items": changed([*self._items, *self._due, *self._held]),
"delayed": changed(entry[2] for entry in self._delayed),
"parked": changed(i for held in self._parked.values() for i in held),
"done_markers": sum(rename.name(n) != n for _, n in self._done),
}
class RedisWorkQueue(WorkQueue):
"""Redis Streams queue: the one that survives the process.
@@ -505,7 +752,16 @@ class RedisWorkQueue(WorkQueue):
# finding them with a keyspace scan walked every state and idempotency
# key in the database, twice per health poll.
self._parked_flows_key = f"{namespace}:__parked_flows__"
# What was added while a rename held the queue (`hold`).
self._held_key = f"{namespace}:__held__"
self._move_due = self._redis.register_script(_MOVE_DUE)
self._release = self._redis.register_script(_RELEASE)
# `add` reads the flag and writes under this, and `release` empties the
# held list and clears the flag under it. So nothing added after `hold`
# lands in a stream a rename is rebuilding, and nothing added while
# held overtakes held work.
self._hold_lock = threading.Lock()
self._holding = False
self._ensure_group()
self._drop_stale_consumers()
@@ -548,15 +804,19 @@ class RedisWorkQueue(WorkQueue):
return f"{self._ns}:__parked__:{flow}"
def _done_key(self, entry_id: str, node: str) -> str:
return f"{self._ns}:__done__:{entry_id}:{node}"
return f"{self._ns}:{DONE}:{entry_id}:{node}"
def add(self, item: WorkItem) -> None:
self._redis.xadd(
self._stream,
cast(Any, item.to_fields()),
maxlen=STREAM_MAXLEN,
approximate=True,
)
with self._hold_lock:
if self._holding:
self._redis.rpush(self._held_key, orjson.dumps(item.to_fields()))
return
self._redis.xadd(
self._stream,
cast(Any, item.to_fields()),
maxlen=STREAM_MAXLEN,
approximate=True,
)
def add_delayed(self, item: WorkItem, not_before: float) -> None:
item.not_before = not_before
@@ -785,6 +1045,10 @@ class RedisWorkQueue(WorkQueue):
list[str], self._redis.lrange(self._parked_key(flow), 0, -1)
)
]
items += [
WorkItem.from_fields(orjson.loads(raw), "")
for raw in cast(list[str], self._redis.lrange(self._held_key, 0, -1))
]
return items
def _group_info(self, stream: str = "") -> dict[str, Any]:
@@ -886,3 +1150,120 @@ class RedisWorkQueue(WorkQueue):
def close(self) -> None:
self._redis.close()
def hold(self) -> None:
with self._hold_lock:
self._holding = True
def release(self) -> None:
with self._hold_lock:
self.release_held()
self._holding = False
@property
def held(self) -> bool:
return self._holding
def release_held(self) -> int:
released = 0
while moved := cast(
int,
self._release(keys=[self._held_key, self._stream], args=[STREAM_MAXLEN]),
):
released += moved
return released
def rename(
self, rename: Rename, progress: Callable[[str], None] | None = None
) -> None:
"""Each structure as section 3 of docs/private/id_refac_rename.md has
it: the streams rebuilt, then timers, lists and done markers, each
rewritten one member, slot or key at a time."""
def done(structure: str) -> None:
if progress is not None:
progress(structure)
for stream, lane in ((self._due_stream, "due"), (self._stream, "backlog")):
rebuild_stream(self._redis, stream, rename)
done(lane)
# One transaction, so each timer moves with its deadline. Promotion is
# held, and a timer added meanwhile is not touched.
pipe = self._redis.pipeline()
for raw, at in cast(
list[tuple[str, float]],
self._redis.zrange(self._delayed_key, 0, -1, withscores=True),
):
fields = rename.fields(orjson.loads(raw))
if fields is not None:
pipe.zrem(self._delayed_key, raw)
pipe.zadd(self._delayed_key, {orjson.dumps(fields): at})
pipe.execute()
done("delayed")
# Slot by slot: nothing pops while held, and `add` only appends, so
# the indices read hold.
for key in [*self._parked_keys(rename), self._held_key]:
pipe = self._redis.pipeline(transaction=False)
for i, raw in enumerate(cast(list[str], self._redis.lrange(key, 0, -1))):
fields = rename.fields(orjson.loads(raw))
if fields is not None:
pipe.lset(key, i, orjson.dumps(fields).decode())
pipe.execute()
old, new = self._parked_key(rename.old), self._parked_key(rename.new)
if self._redis.exists(old):
# Checked first: inside the transaction a refused RENAMENX would
# not stop the index from moving.
if self._redis.exists(new):
raise RuntimeError(f"Work is parked under '{rename.new}' already")
pipe = self._redis.pipeline()
pipe.renamenx(old, new)
pipe.srem(self._parked_flows_key, rename.old)
pipe.sadd(self._parked_flows_key, rename.new)
pipe.execute()
done("lists")
move_keys(self._redis, self._marker_moves(rename))
done("done markers")
def _parked_keys(self, rename: Rename) -> list[str]:
"""Every parked list, the renamed flow's under either name included."""
flows = cast(set[str], self._redis.smembers(self._parked_flows_key))
return list(
dict.fromkeys(
map(self._parked_key, [*sorted(flows), rename.old, rename.new])
)
)
def _marker_moves(self, rename: Rename) -> list[tuple[str, str]]:
moves = []
for key in self._redis.scan_iter(
match=f"{self._ns}:{DONE}:*", count=RENAME_PAGE
):
entry_id, node = done_marker(key)
if rename.name(node) != node:
moves.append((key, self._done_key(entry_id, rename.name(node))))
return moves
def count(self, rename: Rename) -> dict[str, int]:
def changed(raws: Iterable[str]) -> int:
return sum(rename.fields(orjson.loads(raw)) is not None for raw in raws)
def listed(key: str) -> list[str]:
return cast(list[str], self._redis.lrange(key, 0, -1))
live = [
fields
for stream in (self._due_stream, self._stream)
for _, fields in self._live_entries(stream)
]
return {
"queue_items": sum(rename.fields(f) is not None for f in live)
+ changed(listed(self._held_key)),
"delayed": changed(
cast(list[str], self._redis.zrange(self._delayed_key, 0, -1))
),
"parked": sum(changed(listed(key)) for key in self._parked_keys(rename)),
"done_markers": len(self._marker_moves(rename)),
}
backend/fluksio/flow/state.py
+116 -3
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,6 +17,8 @@ from typing import Any, cast
import orjson
import redis
from fluksio.flow.namespace import HISTORY, NAMES, Rename
#: A node may return a dict keyed by something other than a string, and the
#: stdlib encoder this replaced turned those keys into strings rather than
#: refusing them. Two differences remain and are both improvements: a
@@ -34,6 +36,9 @@ SOCKET_TIMEOUT_S = 5.0
CONNECT_TIMEOUT_S = 2.0
HISTORY_LIMIT = 120
#: How many keys a rename moves per round trip, and between two calls of its
#: ``progress``, which is where the rename journal can stop.
RENAME_BATCH = 500
def as_number(value: Any) -> float | None:
@@ -48,6 +53,30 @@ def as_number(value: Any) -> float | None:
return float(value)
def move_keys(client: redis.Redis, moves: list[tuple[str, str]]) -> int:
"""``RENAMENX`` each ``(key, target)`` in one round trip, which keeps the
Redis type and the remaining TTL. Returns how many moved.
A key gone since it was listed has expired and is skipped. A target that
exists is refused rather than overwritten: a rename checks that nothing
lives under the new name before it starts, so one there now is a fault.
"""
pipe = client.pipeline(transaction=False)
for key, target in moves:
pipe.renamenx(key, target)
replies = pipe.execute(raise_on_error=False)
moved = 0
for (key, target), reply in zip(moves, replies, strict=True):
if isinstance(reply, redis.ResponseError) and "no such key" in str(reply):
continue
if isinstance(reply, Exception):
raise reply
if not reply:
raise RuntimeError(f"Will not move '{key}' onto '{target}', which exists")
moved += 1
return moved
class StateBackend(ABC):
"""
Abstract base class for state storage backends.
@@ -287,6 +316,32 @@ class StateBackend(ABC):
"""
...
# -------------------------------------------------------------------------
# Renaming a flow key
# -------------------------------------------------------------------------
@abstractmethod
def rename_keys(
self, rename: Rename, progress: Callable[[str], None] | None = None
) -> int:
"""Move every key that names something of ``rename.old`` to the new
name (`Rename.state_key`), with its history.
Keys move, values do not: a message's payload and a node's private
state are its own, whatever names they mention. A key already under
the new name stays where it is, so a rerun after a crash finishes the
job. A target that exists is refused rather than overwritten.
:param progress: Called after each batch of keys.
:returns: How many keys moved.
:raises ValueError: For a bookkeeping key of an unregistered family.
"""
...
def count(self, rename: Rename) -> int:
"""How many keys `rename_keys` would move."""
return sum(rename.state_key(key) is not None for key in self.keys())
def __contains__(self, key: str) -> bool:
return self.exists(key)
@@ -457,6 +512,39 @@ class MemoryState(StateBackend):
with self._lock:
return list(self._history.get(key, ()))
def _moves(self, rename: Rename) -> tuple[dict[str, str], dict[str, str]]:
"""Where the keys of values and of histories go. Histories are kept
by bare message name here."""
data = {k: t for k in self._data if (t := rename.state_key(k)) is not None}
history = {k: rename.name(k) for k in self._history if rename.name(k) != k}
return data, history
def rename_keys(
self, rename: Rename, progress: Callable[[str], None] | None = None
) -> int:
"""One pass under the lock. A history moves as the deque it is, so it
keeps its capacity."""
with self._lock:
data, history = self._moves(rename)
for store, moves in ((self._data, data), (self._history, history)):
clash = next((t for t in moves.values() if t in store), None)
if clash is not None:
raise RuntimeError(
f"Will not move a key onto '{clash}', which exists"
)
for key, target in data.items():
self._data[target] = self._data.pop(key)
for key, target in history.items():
self._history[target] = self._history.pop(key)
if progress is not None:
progress("keys")
return len(data) + len(history)
def count(self, rename: Rename) -> int:
with self._lock:
data, history = self._moves(rename)
return len(data) + len(history)
class RedisState(StateBackend):
"""
@@ -533,7 +621,7 @@ class RedisState(StateBackend):
return self._deserialize(data) if data is not None else default
def _names_key(self) -> str:
return self._key("__names__")
return self._key(NAMES)
def message_names(self) -> list[str]:
"""The message names, from the set kept beside them.
@@ -770,7 +858,7 @@ class RedisState(StateBackend):
self._client.close()
def _history_key(self, key: str) -> str:
return self._key(f"__history__:{key}")
return self._key(f"{HISTORY}:{key}")
def append_history(
self, values: dict[str, Any], ts: float, limits: dict[str, int] | None = None
@@ -841,3 +929,28 @@ class RedisState(StateBackend):
ts, value = orjson.loads(entry)
points.append((float(ts), float(value)))
return points
def rename_keys(
self, rename: Rename, progress: Callable[[str], None] | None = None
) -> int:
"""``RENAMENX`` in batches, then the names set in one transaction.
The keys are listed before any moves, since a SCAN running over keys
that are being renamed can return one twice.
"""
moves = [(k, t) for k in self.keys() if (t := rename.state_key(k)) is not None]
moved = 0
for start in range(0, len(moves), RENAME_BATCH):
batch = moves[start : start + RENAME_BATCH]
moved += move_keys(
self._client, [(self._key(k), self._key(t)) for k, t in batch]
)
if progress is not None:
progress("keys")
names = [name for name in self.message_names() if rename.name(name) != name]
if names:
pipe = self._client.pipeline()
pipe.srem(self._names_key(), *names)
pipe.sadd(self._names_key(), *map(rename.name, names))
pipe.execute()
return moved
backend/tests/flow/test_namespace.py
+180
View File
@@ -0,0 +1,180 @@
"""The rename mapper, and the registries that keep it complete."""
import ast
import dataclasses
import inspect
import re
from types import SimpleNamespace
import pytest
from fluksio.flow import namespace, pipeline, state
from fluksio.flow import queue as queue_module
from fluksio.flow.alerts import dedup_key
from fluksio.flow.namespace import FAMILIES, ITEM_FIELDS, Rename, touches
from fluksio.flow.nodes import Node, base
from fluksio.flow.pipeline import Pipeline
from fluksio.flow.queue import RedisWorkQueue, WorkItem, done_marker
from fluksio.flow.state import MemoryState, RedisState
HEAT = Rename("heat", "warm")
@pytest.mark.parametrize(
("key", "moved"),
[
("heat.temp", "warm.temp"),
("heat.room.temp", "warm.room.temp"),
("heating.temp", None),
("__version__:heat.temp", "__version__:warm.temp"),
("__version__:heating.temp", None),
("__history__:heat.room.temp", "__history__:warm.room.temp"),
("__flush__:heat.valve", "__flush__:warm.valve"),
("__last_seen__:heat.valve:hall.temp", "__last_seen__:warm.valve:hall.temp"),
("__last_seen__:hall.lamp:heat.temp", "__last_seen__:hall.lamp:warm.temp"),
("__last_seen__:heat.valve:heat.temp", "__last_seen__:warm.valve:warm.temp"),
("__in_ts__:heating.valve:heat.temp", "__in_ts__:heating.valve:warm.temp"),
("__in_ts__:heating.valve:heating.temp", None),
# A node's own key is its data: only the node part moves.
("__node__:heat.limit:wait:heat.temp", "__node__:warm.limit:wait:heat.temp"),
("__node__:hall.limit:wait:heat.temp", None),
("__names__", None),
# Already under the new key.
("warm.temp", None),
("__last_seen__:warm.valve:warm.temp", None),
],
)
def test_state_keys_move_by_the_names_they_embed(key, moved):
assert HEAT.state_key(key) == moved
@pytest.mark.parametrize(
"key",
["__bogus__:heat.temp", "__version__", "__names__:heat.temp", "__in_ts__:heat.v"],
)
def test_a_key_of_no_registered_family_is_refused(key):
with pytest.raises(ValueError, match="no registered family"):
HEAT.state_key(key)
def test_every_builder_maps_onto_itself_under_the_new_key():
"""The families agree with the builders that write them."""
pipe = Pipeline(nodes=[], state=MemoryState())
redis_state = RedisState(namespace="pipeline")
def node_key(flow, key):
node = Node(f=lambda params: None, name="limit")
node.assign_flow(flow, "limit")
return node._state_key(key)
builders = [
lambda f: pipe._version_key(f"{f}.temp"),
lambda f: pipe._timestamp_key(f"{f}.temp"),
lambda f: pipe._held_key(f"{f}.temp"),
lambda f: pipe._flush_key(f"{f}.valve"),
lambda f: pipe._last_seen_key(f"{f}.valve", f"{f}.temp"),
lambda f: pipe._delivered_key("hall.lamp", f"{f}.temp"),
lambda f: node_key(f, "wait:heat.temp"),
lambda f: redis_state._history_key(f"{f}.temp").removeprefix("pipeline:"),
]
for build in builders:
assert HEAT.state_key(build("heat")) == build("warm")
redis_state.close()
def test_touches_reads_the_same_parts():
assert touches("heat", "__last_seen__:hall.lamp:heat.temp")
assert not touches("heat", "__node__:hall.limit:wait:heat.temp")
assert not touches("heat", "heating.temp")
assert not touches("heat", "__names__")
def test_work_items_move_their_node_flow_and_output_names():
own = WorkItem(
kind="flush", node="heat.valve", flow="heat", guard_key="wait:heat.temp"
)
reaching = WorkItem(
kind="cascade",
node="hall.lamp",
flow="hall",
outputs={"heat.target": 21, "hall.on": True},
)
assert HEAT.item(own) == dataclasses.replace(own, node="warm.valve", flow="warm")
assert HEAT.item(reaching) == dataclasses.replace(
reaching, outputs={"warm.target": 21, "hall.on": True}
)
for item in (own, reaching):
assert HEAT.fields(item.to_fields()) == HEAT.item(item).to_fields()
def test_work_items_without_the_old_names_stay():
# A run item names its flow by id, and the first ids were the keys.
run = WorkItem(kind="run", node="", flow="heat", run_id="r1")
other = WorkItem(kind="cascade", node="heating.valve", flow="heating")
fields = {**other.to_fields(), "not_before": "0"}
assert HEAT.item(run) is None
assert HEAT.fields(run.to_fields()) is None
assert HEAT.item(other) is None
assert HEAT.fields(fields) is None
def test_alert_dedup_keys_follow_the_node_or_flow():
def key(**event):
return HEAT.dedup_key(dedup_key(event))
assert key(type="node_error", node="heat.valve", flow="heat") == (
"node_error:warm.valve"
)
assert key(type="flow_quarantined", flow="heat") == "flow_quarantined:warm"
assert key(type="flow_quarantined", flow="heating") == "flow_quarantined:heating"
assert key(type="engine_degraded") == "engine_degraded:"
def test_done_markers_read_back():
queue = SimpleNamespace(_ns="queue")
for entry_id in ("1700000000000-0", "due:1700000000000-0"):
key = RedisWorkQueue._done_key(queue, entry_id, "heat.valve")
assert done_marker(key) == (entry_id, "heat.valve")
FAMILY_LITERAL = re.compile(r"(__[a-z][a-z_]*__)(?::|$)")
#: What Python itself names this way, such as ``getattr(f, "__name__")``.
PYTHON = set(dir(type)) | set(dir(lambda: None))
@pytest.mark.parametrize(
("module", "registry"),
[
(pipeline, FAMILIES),
(base, FAMILIES),
(state, FAMILIES),
(queue_module, queue_module.STRUCTURES),
],
)
def test_every_family_in_the_key_builders_is_registered(module, registry):
"""A key family written as a literal here, not registered in `namespace`
(the queue's in `queue.STRUCTURES`), would be left behind by a rename."""
found = {
family
for node in ast.walk(ast.parse(inspect.getsource(module)))
if isinstance(node, ast.Constant) and isinstance(node.value, str)
for family in FAMILY_LITERAL.findall(node.value)
} - PYTHON
assert found <= set(registry), found - set(registry)
def test_every_work_item_field_is_classified():
fields = {field.name for field in dataclasses.fields(WorkItem)}
assert fields == set(ITEM_FIELDS)
assert set(WorkItem(kind="cascade", node="a.b", flow="a").to_fields()) <= fields
assert set(ITEM_FIELDS.values()) <= {
None,
namespace.NAME,
namespace.FLOW,
namespace.OUTPUTS,
}
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