From 312dc21afe0c16c7212a8ef852675246191f8ae4 Mon Sep 17 00:00:00 2001 From: Hector Date: Thu, 5 Nov 2026 13:40:20 +0000 Subject: [PATCH 1/2] fix(client): Use mget_nonatomic for RunStorage.read_many on RedisCluster On redis>=4.1, RedisCluster.mget raises "MGET - all keys must map to the same key slot" unless every key shares a hash slot. RunStorage.read_many mgets one key per schedule, so the scheduler crashes on startup with more than one schedule. The crash comes from moving from redis-py-cluster to redis-py>=4.1, not from a redis-py major version bump. Route those mgets through a helper. It calls mget_nonatomic when the client has one (redis-py>=4.1 RedisCluster) and plain mget otherwise. That covers StrictRedis on every version, where there are no slots, and redis-py-cluster, whose mget already splits the keys. None of the existing code switches between redis-py-cluster and redis-py at runtime; the only switch is the TYPE_CHECKING import fallback. So this detects the method instead of importing a class. Add RedisCluster tests that route every slot to the single CI redis. They live in their own module and skip when redis.cluster can't be imported, which is the case on the locked dev version (redis 3.5.3). They pass locally on redis 5.0.1. ## mget across redis-py versions Non-cluster (redis.Redis / StrictRedis): mget sends a single MGET on every version from 3.x to 8.x. mget_nonatomic has never existed there. | Client / version | mget across slots | mget_nonatomic | |---|---|---| | redis-py-cluster 1.x (on redis-py 2.x) | Doesn't raise: one GET per key, not atomic ([1.3.6 client.py:675](https://github.com/Grokzen/redis-py-cluster/blob/1.3.6/rediscluster/client.py#L675)) | Doesn't exist | | redis-py-cluster 2.x (on redis-py 3.x, last release 2.1.3) | Same: one GET per key ([2.1.3 client.py:1003](https://github.com/Grokzen/redis-py-cluster/blob/2.1.3/rediscluster/client.py#L1003)) | Doesn't exist | | redis-py 4.0 | No cluster client | - | | redis-py 4.1.0 (cluster support added in [#1660](https://github.com/redis/redis-py/pull/1660)) | Raises RedisClusterException ([cluster.py:803](https://github.com/redis/redis-py/blob/v4.1.0/redis/cluster.py#L803)) | Added: one MGET per slot, one after another ([commands/cluster.py:26](https://github.com/redis/redis-py/blob/v4.1.0/redis/commands/cluster.py#L26)) | | redis-py 4.3.0 | Raises, async RedisCluster too ([#2099](https://github.com/redis/redis-py/pull/2099)) | Async version added ([commands/cluster.py:205](https://github.com/redis/redis-py/blob/v4.3.0/redis/commands/cluster.py#L205)) | | redis-py 4.3.4 | Raises | Per-slot MGETs run as one pipeline ([#2230](https://github.com/redis/redis-py/pull/2230)) | | redis-py 5.0.0 | Raises ([cluster.py:1002](https://github.com/redis/redis-py/blob/v5.0.0/redis/cluster.py#L1002)) | Unchanged ([commands/cluster.py:156](https://github.com/redis/redis-py/blob/v5.0.0/redis/commands/cluster.py#L156)) | | redis-py 6.x / 7.x | Raises | Unchanged | | redis-py 8.x (8.1.0) | Raises ([cluster.py:1494](https://github.com/redis/redis-py/blob/v8.1.0/redis/cluster.py#L1494)) | Unchanged ([commands/cluster.py:234](https://github.com/redis/redis-py/blob/v8.1.0/redis/commands/cluster.py#L234)) | ## redis-py-cluster mget vs redis-py mget_nonatomic Both return values in input-key order, with None for missing keys, and neither is atomic across slots. | | redis-py-cluster mget | redis-py mget_nonatomic | |---|---|---| | Commands sent | One GET per key | One MGET per hash slot | | Round trips | N, one after another | 4.1.0-4.3.3: one per slot, one after another. 4.3.4+: a single pipeline | | Atomicity | None, even within a slot | Atomic within each slot, not across slots | | Replica reads | - | Can read from replicas when read_from_replicas is set ([v8.1.0 commands/cluster.py:207](https://github.com/redis/redis-py/blob/v8.1.0/redis/commands/cluster.py#L207)) | Co-Authored-By: Claude Opus 5.5 --- .../taskbroker_client/scheduler/storage.py | 14 ++++-- .../tests/scheduler/test_storage_cluster.py | 45 +++++++++++++++++++ 2 files changed, 56 insertions(+), 3 deletions(-) create mode 100644 clients/python/tests/scheduler/test_storage_cluster.py diff --git a/clients/python/src/taskbroker_client/scheduler/storage.py b/clients/python/src/taskbroker_client/scheduler/storage.py index 92fca58a..f223b8af 100644 --- a/clients/python/src/taskbroker_client/scheduler/storage.py +++ b/clients/python/src/taskbroker_client/scheduler/storage.py @@ -118,6 +118,14 @@ def __init__(self, metrics: MetricsBackend, redis: RedisCluster | StrictRedis) - def _make_key(self, key: str) -> str: return f"tw:scheduler:{key}" + def _mget_nonatomic(self, keys: list[str]) -> list[str | None]: + # We want mget_nonatomic (spelled mget in redis-py-cluster) when dealing + # with a cluster. For non clusters we want plain mget. + mget_nonatomic = getattr(self._redis, "mget_nonatomic", None) + if mget_nonatomic is not None: + return mget_nonatomic(keys) + return self._redis.mget(keys) + def set(self, key: str, next_runtime: datetime) -> bool: """ Record a spawn time for a task. @@ -168,9 +176,9 @@ def read_many( compat_keys = [sk.split(":", 1)[1] for sk in storage_keys] legacy_keys = [ck.rsplit(":", 1)[0] for ck in compat_keys] - new_values = self._redis.mget([self._make_key(sk) for sk in storage_keys]) - compat_values = self._redis.mget([self._make_key(ck) for ck in compat_keys]) - legacy_values = self._redis.mget([self._make_key(lk) for lk in legacy_keys]) + new_values = self._mget_nonatomic([self._make_key(sk) for sk in storage_keys]) + compat_values = self._mget_nonatomic([self._make_key(ck) for ck in compat_keys]) + legacy_values = self._mget_nonatomic([self._make_key(lk) for lk in legacy_keys]) run_times: dict[str, datetime | None] = {} for storage_key, new_val, compat_val, legacy_val in zip( diff --git a/clients/python/tests/scheduler/test_storage_cluster.py b/clients/python/tests/scheduler/test_storage_cluster.py new file mode 100644 index 00000000..e7f911f5 --- /dev/null +++ b/clients/python/tests/scheduler/test_storage_cluster.py @@ -0,0 +1,45 @@ +from collections.abc import Iterator +from datetime import UTC, datetime +from unittest import mock + +import pytest + +# redis-py only ships RedisCluster from 4.1; the locked dev version may be older. +pytest.importorskip("redis.cluster") + +from redis.cluster import PRIMARY, ClusterNode, NodesManager, RedisCluster # noqa: E402 +from redis.crc import REDIS_CLUSTER_HASH_SLOTS # noqa: E402 + +from taskbroker_client.metrics import NoOpMetricsBackend # noqa: E402 +from taskbroker_client.scheduler.storage import RunStorage # noqa: E402 + + +def _single_server_slots(self: NodesManager) -> None: + # CI's redis isn't a cluster, so route every slot to it. The slot check that + # fails is client-side, so this reproduces what a real cluster does. + node = ClusterNode("localhost", 6379, server_type=PRIMARY) + node.redis_connection = self.create_redis_node("localhost", 6379, **self.connection_kwargs) + self.nodes_cache = {node.name: node} + self.slots_cache = {slot: [node] for slot in range(REDIS_CLUSTER_HASH_SLOTS)} + self.default_node = node + + +@pytest.fixture +def cluster_storage() -> Iterator[RunStorage]: + with mock.patch.object(NodesManager, "initialize", _single_server_slots): + client = RedisCluster(host="localhost", port=6379, decode_responses=True) + storage = RunStorage(metrics=NoOpMetricsBackend(), redis=client) + keys = ["first:test:valid:300", "second:test:valid:300"] + yield storage + for key in keys: + storage.delete(key) + + +def test_run_storage_read_many_on_redis_cluster(cluster_storage: RunStorage) -> None: + cluster_storage.set("first:test:valid:300", datetime(2099, 1, 1, tzinfo=UTC)) + first_run = cluster_storage.read("first:test:valid:300") + + result = cluster_storage.read_many(["first:test:valid:300", "second:test:valid:300"]) + + assert first_run is not None + assert result == {"first:test:valid:300": first_run, "second:test:valid:300": None} From 25343fa4bc6675640dc5f5617739b3d1c4afa555 Mon Sep 17 00:00:00 2001 From: Hector Date: Thu, 5 Nov 2026 16:26:33 +0000 Subject: [PATCH 2/2] Address comments Co-Authored-By: Claude Opus 5.5 --- clients/python/tests/scheduler/test_storage_cluster.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/clients/python/tests/scheduler/test_storage_cluster.py b/clients/python/tests/scheduler/test_storage_cluster.py index e7f911f5..267635ad 100644 --- a/clients/python/tests/scheduler/test_storage_cluster.py +++ b/clients/python/tests/scheduler/test_storage_cluster.py @@ -28,11 +28,8 @@ def _single_server_slots(self: NodesManager) -> None: def cluster_storage() -> Iterator[RunStorage]: with mock.patch.object(NodesManager, "initialize", _single_server_slots): client = RedisCluster(host="localhost", port=6379, decode_responses=True) - storage = RunStorage(metrics=NoOpMetricsBackend(), redis=client) - keys = ["first:test:valid:300", "second:test:valid:300"] - yield storage - for key in keys: - storage.delete(key) + client.flushdb() + yield RunStorage(metrics=NoOpMetricsBackend(), redis=client) def test_run_storage_read_many_on_redis_cluster(cluster_storage: RunStorage) -> None: