"""Stream rebuild checks for the rename's queue migration, enabled by 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). """ import os import time import uuid import pytest import redis GROUP = "engine" @pytest.fixture def ns(): return f"test-streams-{uuid.uuid4().hex}" @pytest.fixture def r(ns): host = os.environ.get("FLUKSIO_TEST_REDIS_HOST") if not host: pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks") client = redis.Redis( host=host, port=int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379")), decode_responses=True, ) try: yield client finally: keys = list(client.scan_iter(match=f"{ns}:*")) if keys: client.delete(*keys) client.close() def _id(entry_id: str) -> tuple[int, int]: ms, seq = entry_id.split("-") 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() def _pel(r, stream: str) -> list[dict]: return r.xpending_range(stream, GROUP, "-", "+", count=100) def _snapshot(r, stream: str) -> dict: info = r.xinfo_stream(stream) (group,) = r.xinfo_groups(stream) return { "entries": r.xrange(stream), "stream": { k: info[k] for k in ( "length", "last-generated-id", "entries-added", "max-deleted-entry-id", "recorded-first-entry-id", ) }, "group": { k: group[k] for k in ("last-delivered-id", "entries-read", "lag", "pending") }, "pel": [ (p["message_id"], p["consumer"], p["times_delivered"]) for p in _pel(r, stream) ], "consumers": sorted(c["name"] for c in r.xinfo_consumers(stream, GROUP)), } def _busy_stream(r, stream: str) -> list[str]: """Nine entries: 0 acked, 1 and 2 pending with a (1 delivered three times), 3 to 5 pending with b (3 deleted from the stream, 4 delivered twice), 6 to 8 undelivered. a's entries are idle 300 ms longer than b's.""" r.xgroup_create(stream, GROUP, id="0", mkstream=True) ids = [r.xadd(stream, {"flow": "old", "n": str(i)}) for i in range(9)] r.xreadgroup(GROUP, "a", {stream: ">"}, count=3) r.xack(stream, GROUP, ids[0]) for _ in range(2): r.xclaim(stream, GROUP, "a", 0, [ids[1]]) time.sleep(0.3) r.xreadgroup(GROUP, "b", {stream: ">"}, count=3) r.xclaim(stream, GROUP, "b", 0, [ids[4]]) r.xdel(stream, ids[3]) return ids def test_rebuild_keeps_ids_group_and_pending_entries(r, ns): stream = f"{ns}:queue" ids = _busy_stream(r, stream) before = _snapshot(r, stream) idle_before = {p["message_id"]: p["time_since_delivered"] for p in _pel(r, stream)} started = time.monotonic() rebuild(r, stream, _rename_flow) idle_after = {p["message_id"]: p["time_since_delivered"] for p in _pel(r, stream)} elapsed_ms = (time.monotonic() - started) * 1000 after = _snapshot(r, stream) assert [i for i, _ in after["entries"]] == [i for i, _ in before["entries"]] assert [f["flow"] for _, f in after["entries"]] == ["old"] + ["new"] * 7 assert after["stream"] == before["stream"] assert before["group"]["lag"] == 3 # The pending entry whose stream entry is gone is not recreated. assert after["group"] == {**before["group"], "pending": 4} assert after["pel"] == [ (ids[1], "a", 3), (ids[2], "a", 1), (ids[4], "b", 2), (ids[5], "b", 1), ] assert after["consumers"] == ["a", "b"] assert min(idle_before[ids[1]], idle_before[ids[2]]) >= 300 for entry_id, idle in idle_after.items(): assert idle_before[entry_id] <= idle <= idle_before[entry_id] + elapsed_ms + 5 new_id = r.xadd(stream, {"flow": "new", "n": "9"}) assert _id(new_id) > _id(before["stream"]["last-generated-id"]) ((_, delivered),) = r.xreadgroup(GROUP, "c", {stream: ">"}) assert [i for i, _ in delivered] == ids[6:] + [new_id] assert not r.exists(f"{stream}:rename") def test_rebuild_of_an_emptied_stream_keeps_its_id_counter(r, ns): stream = f"{ns}:due" # Ids ahead of the clock, so only XSETID keeps the next `*` id above them. ahead = int(time.time() * 1000) + 86_400_000 r.xgroup_create(stream, GROUP, id="0", mkstream=True) ids = [r.xadd(stream, {"flow": "old"}, id=f"{ahead}-{i}") for i in range(3)] r.xreadgroup(GROUP, "a", {stream: ">"}, count=2) r.xack(stream, GROUP, *ids[:2]) r.xdel(stream, *ids) before = _snapshot(r, stream) rebuild(r, stream, _rename_flow) 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): stream = f"{ns}:queue" _busy_stream(r, stream) original = r.xrange(stream) # Crash before RENAME: the stream is untouched and the temp key is left. rebuild(r, stream, _rename_flow, swap=False) assert r.xrange(stream) == original assert r.exists(f"{stream}:rename") rebuild(r, stream, _rename_flow) once = _snapshot(r, stream) # Crash after RENAME, before the journal checkpoint: the rerun changes nothing. rebuild(r, stream, _rename_flow) assert _snapshot(r, stream) == once assert not r.exists(f"{stream}:rename") @pytest.mark.parametrize("command", ["rename", "renamenx"]) def test_rename_keeps_the_ttl_of_strings_and_lists(r, ns, command): r.set(f"{ns}:string", "v", ex=600) r.rpush(f"{ns}:list", "a", "b") r.expire(f"{ns}:list", 600) for kind in ("string", "list"): assert getattr(r, command)(f"{ns}:{kind}", f"{ns}:{kind}.new") assert not r.exists(f"{ns}:{kind}") assert 590 < r.ttl(f"{ns}:{kind}.new") <= 600 assert r.lrange(f"{ns}:list.new", 0, -1) == ["a", "b"]