From e6027af8fe834d3d55da95e03d8f6beec773e113 Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 2 Nov 2026 18:51:14 +0100 Subject: [PATCH] Prove the rename's stream rebuild on a real Redis: same ids, group, lag and pending entries after RENAME, ENTRIESREAD set after XSETID because Redis caps it; backend CI gets a Redis 8.2 service --- .gitea/workflows/test-backend.yml | 11 +- backend/tests/flow/test_redis_streams.py | 239 +++++++++++++++++++++++ 2 files changed, 249 insertions(+), 1 deletion(-) create mode 100644 backend/tests/flow/test_redis_streams.py diff --git a/.gitea/workflows/test-backend.yml b/.gitea/workflows/test-backend.yml index 55854a9..87e21bd 100644 --- a/.gitea/workflows/test-backend.yml +++ b/.gitea/workflows/test-backend.yml @@ -25,10 +25,18 @@ jobs: fail-fast: false matrix: python-version: ["3.12", "3.14"] + # The Redis queue and stream tests run against a real server. Same major + # version as the shipped image in docker/compose.yml. + services: + redis: + image: redis:8.2-alpine env: # uv would otherwise take a managed interpreter it already has over the # one set up below. UV_PYTHON: ${{ matrix.python-version }} + # The job runs in a container next to the service, which answers to its + # name on the default port. + FLUKSIO_TEST_REDIS_HOST: redis steps: - name: Checkout uses: actions/checkout@v4 @@ -46,9 +54,10 @@ jobs: # The extras too, as `make test-backend` does: the mail, MQTT and Influx # tests import what only the `server` extra installs. - run: uv sync --all-extras - # Nothing to start: the suite names its own SQLite file in a temp + # Nothing else to start: the suite names its own SQLite file in a temp # directory (backend/tests/__init__.py), the mail paths are patched, and # the flow engine falls back to in-memory state without a Redis host. + # Only the tests gated on FLUKSIO_TEST_REDIS_HOST use the service. - name: Run tests with coverage run: uv run bash scripts/test.sh "Coverage for ${{ github.sha }}" working-directory: backend diff --git a/backend/tests/flow/test_redis_streams.py b/backend/tests/flow/test_redis_streams.py new file mode 100644 index 0000000..31bf6b5 --- /dev/null +++ b/backend/tests/flow/test_redis_streams.py @@ -0,0 +1,239 @@ +"""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"]