"""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 (`queue.rebuild_stream`, docs/private/id_refac_rename.md, 3.1). """ import os import time import uuid import pytest import redis from fluksio.flow.namespace import Rename from fluksio.flow.queue import rebuild_stream 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) OLD_NEW = Rename("old", "new") 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_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 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_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, 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. 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_stream(r, stream, OLD_NEW) once = _snapshot(r, stream) # Crash after RENAME, before the journal checkpoint: the rerun changes nothing. rebuild_stream(r, stream, OLD_NEW) 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"]