Startup deletes the work queue's keys from before it moved from pipeline: to queue: (stream, timers, dead letters, parked lists, done markers), logged with what they held; an old stream still holding undelivered work stays, is logged as an error, and counts as queued work for the keys it names

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, ``<flow key>.<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

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]:

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

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__")