diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2c0a0e2..e701866 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -124,7 +124,7 @@ jobs: python-version: "3.13" - run: docker exec ${{ job.services.redis_cluster.id }} redis-cli -p 7000 cluster info - run: uv sync --no-default-groups --group test - - run: uv run --no-sync pytest tests/test_integration_cluster.py -v + - run: uv run --no-sync pytest tests/test_integration_cluster.py tests/test_integration_cluster_resharding.py -v real-redis-wrongtype: runs-on: ubuntu-latest diff --git a/docs/production-readiness.md b/docs/production-readiness.md index 6552813..04cbafd 100644 --- a/docs/production-readiness.md +++ b/docs/production-readiness.md @@ -166,7 +166,7 @@ These are not bugs, but areas without dedicated test coverage: - VT reclaim LREM is O(expired x processing_queue_depth) — benchmarked at 73ms for 50K messages with 100 expired - Double external `CancelledError` during async `process_message` finally block can replace original exception - Async deduplication callable — nested coroutines and event-loop-cross interaction remain untested -- Redis Cluster runtime coverage is partial, not absent. Construction-time hash-tag validation is tested, and `test_integration_cluster.py` exercises hash-tagged multi-key Lua evaluation (publish/ack/reclaim, sync and async) plus cross-slot `pcall`-rejection paths against a real `RedisCluster` — but only when `REDIS_CLUSTER_URL` is set (its dedicated `real-redis-cluster` CI job), skipping on local runs without a cluster. `test_cluster_moved_redirect.py` pins MOVED/ASK redirect handling by *simulating* a redirect at the gateway's script-eval boundary (an injected `MovedError`/`AskError` on single-node Redis): a redirect that escapes redis-py surfaces as a catchable, non-retryable `ResponseError`, is not rewrapped as `LuaScriptError`, and leaves the lease/claim bookkeeping intact so a settled-migration retry succeeds. What still needs a real multi-node resharding fixture: redis-py's transparent redirect-following *below* `client.eval` during a live slot migration — the simulation patches the eval boundary and cannot reach the path where redis-py refreshes its slot map and re-issues the command, nor the degraded-cluster case where a bare redirect reaches the gateway only after redis-py exhausts its own redirect retries. +- Redis Cluster runtime coverage is partial, not absent. Construction-time hash-tag validation is tested, and `test_integration_cluster.py` exercises hash-tagged multi-key Lua evaluation (publish/ack/reclaim, sync and async) plus cross-slot `pcall`-rejection paths against a real `RedisCluster` — but only when `REDIS_CLUSTER_URL` is set (its dedicated `real-redis-cluster` CI job), skipping on local runs without a cluster. `test_cluster_moved_redirect.py` pins MOVED/ASK redirect handling by *simulating* a redirect at the gateway's script-eval boundary (an injected `MovedError`/`AskError` on single-node Redis): a redirect that escapes redis-py surfaces as a catchable, non-retryable `ResponseError`, is not rewrapped as `LuaScriptError`, and leaves the lease/claim bookkeeping intact so a settled-migration retry succeeds. `test_integration_cluster_resharding.py` (same CI job) now exercises a *live* slot migration on the real multi-node cluster: with a lease outstanding it runs the full `CLUSTER SETSLOT` + `MIGRATE` resharding handshake, then drives lease renewal, ack, and a fresh publish/claim through the gateway — pinning that redis-py transparently refreshes its slot map and re-routes *below* `client.eval` with no error reaching the gateway. What still lacks a real-cluster fixture: redirects during *partial or failed* migrations (ASK transitional states while a slot is mid-handshake with keys split across nodes), multi-slot resharding storms, and the degraded case where a bare redirect reaches the gateway only after redis-py exhausts its own redirect retries — that last path remains covered only by the single-node simulation. - Heartbeat lease renewal timing against real Redis — lifecycle tests use fakeredis; renewal latency, thread scheduling jitter, and real network delays are not exercised. - Model-based randomized testing covers sync gateway only — async gateway and non-VT claim path at integration level remain unexercised by the model harness. diff --git a/tests/test_integration_cluster_resharding.py b/tests/test_integration_cluster_resharding.py new file mode 100644 index 0000000..9d2101f --- /dev/null +++ b/tests/test_integration_cluster_resharding.py @@ -0,0 +1,219 @@ +"""Live slot-migration (resharding) coverage against a real multi-node cluster. + +``test_cluster_moved_redirect.py`` pins MOVED/ASK handling by *simulating* a +redirect at the gateway's script-eval boundary. This file closes the remaining +gap: it performs a real ``CLUSTER SETSLOT`` + ``MIGRATE`` handshake on a live +multi-node cluster while a lease is outstanding, then drives lease renewal, +ack, and a fresh claim through the gateway. redis-py's ``RedisCluster`` is +expected to observe the ``MOVED`` reply below ``client.eval``, refresh its +slot map, and re-route transparently — none of it should surface to the +gateway. + +Requires ``REDIS_CLUSTER_URL`` (the dedicated ``real-redis-cluster`` CI job); +skips locally without a cluster. +""" + +import os +import time + +import pytest +import redis +from redis.cluster import key_slot + +from redis_message_queue import ClaimedMessage +from redis_message_queue.redis_message_queue import RedisMessageQueue + +pytestmark = pytest.mark.integration + +# Dedicated hash tag so a failed migration can never poison sibling tests' +# slots ({cluster-test}, {cluster-test-b1}); the fixture still migrates the +# slot back to its original owner afterwards to leave topology untouched. +RESHARDING_TAG = "{resharding-test}" + +# Migration steps are individually fast (single container, loopback); this +# bounds the polling loops that wait for cluster-wide topology agreement. +_SETTLE_TIMEOUT_SECONDS = 10.0 +_SETTLE_POLL_INTERVAL_SECONDS = 0.05 + + +@pytest.fixture(scope="session") +def real_redis_cluster_url(): + url = os.environ.get("REDIS_CLUSTER_URL") + if not url: + pytest.skip("Redis Cluster not available") + return url + + +@pytest.fixture() +def real_redis_cluster_client(real_redis_cluster_url): + client = redis.RedisCluster.from_url(real_redis_cluster_url, socket_connect_timeout=2) + try: + client.ping() + except redis.RedisError as exc: + client.close() + pytest.skip(f"Redis Cluster not available: {exc}") + _delete_resharding_test_keys(client) + yield client + _delete_resharding_test_keys(client) + client.close() + + +def _delete_resharding_test_keys(client) -> None: + for key in list(client.scan_iter(match=f"{RESHARDING_TAG}*")): + client.delete(key) + + +class _MasterNode: + """One cluster master reached over a direct (non-cluster) connection. + + The migration handshake must address individual nodes; the routing + ``RedisCluster`` client would re-route or fan out the ``CLUSTER SETSLOT`` + admin commands, so each master gets a plain ``redis.Redis`` connection. + """ + + def __init__(self, host: str, port: int) -> None: + self.host = host + self.port = port + self.conn = redis.Redis(host=host, port=port, socket_connect_timeout=2) + self.node_id: str = self.conn.execute_command("CLUSTER MYID").decode() + + @property + def name(self) -> str: + return f"{self.host}:{self.port}" + + def close(self) -> None: + self.conn.close() + + +@pytest.fixture() +def cluster_masters(real_redis_cluster_client): + masters = [_MasterNode(node.host, node.port) for node in real_redis_cluster_client.get_primaries()] + if len(masters) < 2: + pytest.skip("slot migration requires at least two masters") + yield masters + for master in masters: + master.close() + + +def _master_by_name(masters: list[_MasterNode], name: str) -> _MasterNode: + for master in masters: + if master.name == name: + return master + raise AssertionError(f"no direct connection for cluster node {name!r}; masters: {[m.name for m in masters]}") + + +def _slot_owner_name(master: _MasterNode, probe_key: str) -> str: + """Return ``host:port`` of the node this master believes owns the probe key's slot. + + Sends a read for the key directly to the master: the owner answers, any + other node replies ``MOVED `` (raised as a typed + ``MovedError`` even on a plain client — the parser maps the prefix). This + avoids parsing ``CLUSTER SLOTS``/``CLUSTER SHARDS`` topology payloads. + """ + try: + master.conn.exists(probe_key) + except redis.exceptions.MovedError as exc: + return f"{exc.host}:{exc.port}" + return master.name + + +def _wait_for_slot_owner(masters: list[_MasterNode], probe_key: str, expected_owner: _MasterNode) -> None: + """Poll until every master agrees the probe key's slot is owned by ``expected_owner``.""" + deadline = time.monotonic() + _SETTLE_TIMEOUT_SECONDS + while True: + owners = {master.name: _slot_owner_name(master, probe_key) for master in masters} + if all(owner == expected_owner.name for owner in owners.values()): + return + if time.monotonic() >= deadline: + raise AssertionError(f"slot ownership did not settle on {expected_owner.name}; current view: {owners}") + time.sleep(_SETTLE_POLL_INTERVAL_SECONDS) + + +def _migrate_slot( + masters: list[_MasterNode], + slot: int, + source: _MasterNode, + target: _MasterNode, + probe_key: str, +) -> None: + """Live-migrate ``slot`` from ``source`` to ``target`` (real resharding handshake).""" + target.conn.execute_command("CLUSTER SETSLOT", slot, "IMPORTING", source.node_id) + source.conn.execute_command("CLUSTER SETSLOT", slot, "MIGRATING", target.node_id) + + while True: + keys = source.conn.execute_command("CLUSTER GETKEYSINSLOT", slot, 100) + if not keys: + break + source.conn.execute_command("MIGRATE", target.host, target.port, "", 0, 5000, "KEYS", *keys) + + # Finalize on the target first (per the Redis resharding protocol), then + # the source, then broadcast so every master converges without waiting for + # gossip. + target.conn.execute_command("CLUSTER SETSLOT", slot, "NODE", target.node_id) + source.conn.execute_command("CLUSTER SETSLOT", slot, "NODE", target.node_id) + for master in masters: + if master.name not in (source.name, target.name): + master.conn.execute_command("CLUSTER SETSLOT", slot, "NODE", target.node_id) + + _wait_for_slot_owner(masters, probe_key, target) + + +def test_live_slot_migration_is_transparent_to_gateway_and_queue( + real_redis_cluster_client, + cluster_masters, +): + """Reshard the queue's slot mid-lease; renew, ack, and a fresh claim must all succeed. + + Pins redis-py's transparent redirect-following below ``client.eval``: after + the migration the routing client's slot map is stale, so the first script + eval hits the old owner, gets a real ``MOVED``, and redis-py must refresh + and re-route without the gateway ever seeing an error. + """ + queue = RedisMessageQueue( + RESHARDING_TAG, + client=real_redis_cluster_client, + deduplication=False, + visibility_timeout_seconds=60, + max_delivery_count=None, + ) + + # Publish + claim so a lease is outstanding across the migration. + queue.publish("survive-resharding") + claimed = queue._redis.wait_for_message_and_move(queue.key.pending, queue.key.processing) + assert isinstance(claimed, ClaimedMessage) + + slot = key_slot(RESHARDING_TAG.encode("utf-8")) + source = _master_by_name(cluster_masters, real_redis_cluster_client.get_node_from_key(queue.key.pending).name) + target = next(master for master in cluster_masters if master.name != source.name) + + _migrate_slot(cluster_masters, slot, source, target, probe_key=queue.key.pending) + try: + # The data physically moved: the queue state now lives on the target. + assert source.conn.execute_command("CLUSTER COUNTKEYSINSLOT", slot) == 0 + assert target.conn.execute_command("CLUSTER COUNTKEYSINSLOT", slot) > 0 + + # Lease renewal through the gateway: the routing client's slot map is + # stale, so this eval takes the real MOVED → refresh → re-route path. + assert ( + queue._redis.renew_message_lease(queue.key.processing, claimed.stored_message, claimed.lease_token) is True + ) + + # Ack through the gateway completes the in-flight message. + assert ( + queue._redis.remove_message(queue.key.processing, claimed.stored_message, lease_token=claimed.lease_token) + is True + ) + assert real_redis_cluster_client.llen(queue.key.pending) == 0 + assert real_redis_cluster_client.llen(queue.key.processing) == 0 + + # Queue-level end-to-end after the migration: a fresh publish + claim + # cycle completes normally against the new slot owner. + queue.publish("post-migration") + with queue.process_message() as message: + assert message == b"post-migration" + assert real_redis_cluster_client.llen(queue.key.pending) == 0 + assert real_redis_cluster_client.llen(queue.key.processing) == 0 + finally: + # Migrate the slot back so the cluster topology is exactly as found + # and sibling tests (and re-runs) see a clean state. + _migrate_slot(cluster_masters, slot, target, source, probe_key=queue.key.pending)