diff --git a/backend/fluksio/flow/namespace.py b/backend/fluksio/flow/namespace.py index 3110ce7..91542ab 100644 --- a/backend/fluksio/flow/namespace.py +++ b/backend/fluksio/flow/namespace.py @@ -48,6 +48,11 @@ HELD = "__held__" FLUSH = "__flush__" #: What an engine-shipped node type remembers (`Node.remember`). NODE_STATE = "__node__" +#: Not state: the work queue's stream from before the queue had a root of its +#: own. Startup deletes it unless it holds work nobody delivered +#: (`RedisWorkQueue.drop_legacy`). Kept, it names no flow's state, and the +#: queue counts that work as queued instead. +LEGACY_QUEUE = "__queue__" #: A qualified message or node name, ``.``. NAME = "name" @@ -72,6 +77,7 @@ FAMILIES: dict[str, tuple[str, ...]] = { HELD: (NAME,), FLUSH: (NAME,), NODE_STATE: (NAME, PRIVATE), + LEGACY_QUEUE: (), } #: What each `WorkItem` field holds that a rename changes, or None. A guard is diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index 2168f08..29d2dd6 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -22,7 +22,7 @@ import secrets import threading import time from abc import ABC, abstractmethod -from collections import OrderedDict, deque +from collections import Counter, OrderedDict, deque from collections.abc import Callable, Iterable from dataclasses import dataclass, field from typing import Any, cast @@ -72,6 +72,9 @@ STRUCTURES = frozenset( "__dead__", } ) +#: Where the Redis queue lived until it got a root of its own (2026-08): the +#: state backend's. Nothing has read what it left there since (`drop_legacy`). +LEGACY_ROOT = "pipeline" # 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 @@ -200,6 +203,12 @@ def owned_by(item: WorkItem, flow: str) -> bool: ) +def _described(items: Iterable[WorkItem]) -> str: + """What *items* are, for a log line: ``2 flush home.house, 1 cascade ...``.""" + counts = Counter((item.kind, item.node) for item in items) + return ", ".join(f"{n} {kind} {node}" for (kind, node), n in counts.most_common()) + + 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 @@ -468,6 +477,12 @@ class WorkQueue(ABC): another engine on the same server, which a rename cannot hold.""" return [] + def drop_legacy(self, root: str = LEGACY_ROOT) -> None: + """Delete what the queue kept under *root* before it moved. For + startup, before anything lists state keys. A queue in this process + has nothing to delete.""" + return None + @abstractmethod def rename( self, rename: Rename, progress: Callable[[str], None] | None = None @@ -776,6 +791,8 @@ class RedisWorkQueue(WorkQueue): self._hold_lock = threading.Condition() self._holding = False self._adding = 0 + # The old layout's stream, while it holds work nobody delivered. + self._legacy = "" self._ensure_group() self._drop_stale_consumers() @@ -814,6 +831,64 @@ class RedisWorkQueue(WorkQueue): except redis.ResponseError: pass + def drop_legacy(self, root: str = LEGACY_ROOT) -> None: + """Delete the keys the queue wrote under *root* before it had its own. + + They sit among the state backend's keys, as families `namespace` does + not know, so a key check, a delete or a rename would trip over them. + The stream goes once its group has delivered and acknowledged every + entry. One still holding work stays and is logged as an error on + every start. Nothing runs that work, but deleting it is the + operator's call, and until then `live_items` counts it, so a key it + names is not given to a new flow. Timers, parked work, dead letters + and done markers go regardless, each logged with what it held. Run + again, it finds nothing and says nothing. + """ + stream = f"{root}:__queue__" + if live := self._live_entries(stream): + self._legacy = stream + logger.error( + "Kept the old queue stream '%s': %d entries were never delivered " + "or acknowledged (%s). Nothing runs them, and a new flow cannot " + "take a key they name. Read them with XRANGE, then delete the " + "key by hand (DEPLOY.md)", + stream, + len(live), + _described(WorkItem.from_fields(f, i) for i, f in live), + ) + for key in (stream, f"{root}:__dead__"): + if key != self._legacy and self._redis.exists(key): + entries = cast(int, self._redis.xlen(key)) + self._redis.delete(key) + logger.warning( + "Deleted the old queue key '%s' (%d entries)", key, entries + ) + timers = f"{root}:__delayed__" + for key in [timers, *self._redis.scan_iter(f"{root}:__parked__:*")]: + raw = cast( + list[str], + self._redis.zrange(key, 0, -1) + if key == timers + else self._redis.lrange(key, 0, -1), + ) + if raw: + items = [WorkItem.from_fields(orjson.loads(r), "") for r in raw] + self._redis.delete(key) + logger.warning( + "Deleted the old queue key '%s' (%d items: %s)", + key, + len(items), + _described(items), + ) + if markers := list(self._redis.scan_iter(f"{root}:{DONE}:*")): + self._redis.delete(*markers) + logger.warning( + "Deleted the old done markers '%s:%s:*' (%d keys)", + root, + DONE, + len(markers), + ) + def _parked_key(self, flow: str) -> str: return f"{self._ns}:__parked__:{flow}" @@ -1069,6 +1144,13 @@ class RedisWorkQueue(WorkQueue): WorkItem.from_fields(orjson.loads(raw), "") for raw in cast(list[str], self._redis.lrange(self._held_key, 0, -1)) ] + if self._legacy: + # Nothing runs it unless someone moves it back by hand, and then + # a new flow under a key it names would inherit it (`drop_legacy`). + items += [ + WorkItem.from_fields(fields, "") + for _, fields in self._live_entries(self._legacy) + ] return items def _group_info(self, stream: str = "") -> dict[str, Any]: diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index bbf1012..cf1cde5 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -317,7 +317,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: event_bus.flow_id_of = lambda key: store.key_index().get(key, key) state = _state_backend() queue = _work_queue() - # Then the rest of it, before anything reads alerts, devices, modules, + # What the queue left among the state keys before it had a root of + # its own, which a rename's state step would refuse. + await run_in_threadpool(queue.drop_legacy) + # Then the rest of the rename, before anything reads alerts, devices, modules, # metrics, flows or runs. Until it is finished nothing is served. await retry_until_done(lambda: recover_runtime(state, queue), event_bus) # Work that arrived while a rename held the queue, which a crash diff --git a/backend/tests/flow/test_redis_queue.py b/backend/tests/flow/test_redis_queue.py index 13d4847..7404398 100644 --- a/backend/tests/flow/test_redis_queue.py +++ b/backend/tests/flow/test_redis_queue.py @@ -1,12 +1,15 @@ """Redis delivery checks, enabled by FLUKSIO_TEST_REDIS_HOST (and optional PORT).""" +import logging import os import uuid from concurrent.futures import ThreadPoolExecutor +import orjson import pytest import redis +from fluksio.flow.namespace import touches from fluksio.flow.queue import RedisWorkQueue, WorkItem @@ -93,3 +96,60 @@ def test_clearing_a_flow_drops_its_queued_claimed_and_held_work(queue): assert queue.reclaim_stale(0) == [] queue.ack(claimed) assert [i.node for i in queue.claim(10, 10)] == ["kept.a", ""] + + +def _old_layout(queue, root, *, acked): + """What the queue wrote under the state backend's root before it moved.""" + r = queue._redis + item = WorkItem(kind="cascade", node="home.tick", flow="home").to_fields() + r.xgroup_create(f"{root}:__queue__", "engine", id="0", mkstream=True) + r.xadd(f"{root}:__queue__", item) + r.xadd(f"{root}:__queue__", item) + # Both delivered and acknowledged, or one claimed and one never handed out. + read = r.xreadgroup( + "engine", "old", {f"{root}:__queue__": ">"}, None if acked else 1 + ) + if acked: + r.xack(f"{root}:__queue__", "engine", *[i for i, _ in read[0][1]]) + flush = WorkItem(kind="flush", node="home.house", flow="home").to_fields() + r.zadd(f"{root}:__delayed__", {orjson.dumps(flush): 1787297711.0}) + r.xadd(f"{root}:__dead__", {**item, "reason": "failed"}) + r.rpush(f"{root}:__parked__:home", orjson.dumps(item)) + r.set(f"{root}:__done__:1-0:home.tick", "1") + + +def test_the_old_layout_is_deleted_once_and_the_queue_kept(queue, caplog): + root = f"{queue._ns}-old" + queue.add(WorkItem(kind="cascade", node="f.a", flow="f")) + queue.add_delayed(WorkItem(kind="flush", node="f.b", flow="f"), 9e9) + queue.park("f", WorkItem(kind="cascade", node="f.c", flow="f")) + current = set(queue._redis.scan_iter(match=f"{queue._ns}:*")) + _old_layout(queue, root, acked=True) + + with caplog.at_level(logging.WARNING, logger="fluksio.flow.queue"): + queue.drop_legacy(root) + + assert list(queue._redis.scan_iter(match=f"{root}:*")) == [] + assert set(queue._redis.scan_iter(match=f"{queue._ns}:*")) == current + assert len(queue.live_items()) == 3 + logged = caplog.text + assert "1 flush home.house" in logged and "__parked__:home" in logged + assert "__dead__" in logged and "__done__:*' (1 keys)" in logged + caplog.clear() + queue.drop_legacy(root) + assert caplog.records == [] + + +def test_undelivered_old_work_is_kept_and_counted(queue, caplog): + root = f"{queue._ns}-old" + _old_layout(queue, root, acked=False) + + with caplog.at_level(logging.WARNING, logger="fluksio.flow.queue"): + queue.drop_legacy(root) + + # Claimed and never acknowledged, and never handed out at all. + assert set(queue._redis.scan_iter(match=f"{root}:*")) == {f"{root}:__queue__"} + assert "2 cascade home.tick" in caplog.text + assert [i.node for i in queue.live_items()] == ["home.tick", "home.tick"] + assert not touches("home", "__queue__") + queue._redis.delete(f"{root}:__queue__")