fix!: request Redis | RedisCluster in the Dishka provider and read per slot on a cluster

docs/agents.md

@@ -213,11 +213,13 @@ result = await coordinator.coordinate(
 | `delete_many(operation, idempotency_keys)` | `int` | as `delete` |
 
 `RedisAsyncIdempotencyRepository(redis, *, key_prefix="idempotency:", metrics=None)` is the
-implementation: `SET key value EX ceil(record.ttl_seconds) NX` for `save`, the same without
-`NX` for `replace`, `GET` for `get`, `MGET` for `get_many`, and a non-transactional pipeline
-for `save_many` so it works on Redis Cluster. It deletes any record it reads back expired and
-reports that as a miss. A repository of your own needs `replace` too: the coordinator raises
-`TypeError` at construction without it, unless `in_flight="run"`.
+implementation, and `redis` is a `redis.asyncio.Redis` or a `RedisCluster`: `SET key value EX
+ceil(record.ttl_seconds) NX` for `save`, the same without `NX` for `replace`, `GET` for `get`,
+`MGET` for `get_many` — one per hash slot, via redis-py's `mget_nonatomic`, on a cluster —
+and a non-transactional pipeline for `save_many`, so all of it works on Redis Cluster. It
+deletes any record it reads back expired and reports that as a miss. A repository of your own
+needs `replace` too: the coordinator raises `TypeError` at construction without it, unless
+`in_flight="run"`.
 
 ### Metrics
 
@@ -259,14 +261,19 @@ from idempotency_kit.dishka import (
 )
 
 container = make_async_container(
-    MyRedisProvider(),          # provides redis.asyncio.Redis
+    MyRedisProvider(),          # provides redis.asyncio.Redis | RedisCluster
     MySettingsProvider(),       # provides IdempotencySettingsProtocol
     IdempotencyProvider(),      # IdempotencyDomainService + IdempotencyMetricsProtocol
     AsyncRedisIdempotencyProvider(),
     AsyncIdempotencyCoordinatorProvider(),
 )
 ```
 
+`AsyncRedisIdempotencyProvider` asks for `redis.asyncio.Redis | RedisCluster`, exactly — the
+key `redis_client_kit.AsyncRedisClient` names, so `redis_client_kit.providers.AsyncRedisProvider`
+takes the place of `MyRedisProvider` as is. Dishka matches keys, not subclasses: a provider of
+your own annotated `-> Redis` is a different key, and the container fails to build.
+
 All three are `Scope.APP`. `IdempotencyProvider` gives one metrics collector to the whole
 process — `PrometheusIdempotencyMetrics` when `settings.metrics_enabled`, a no-op otherwise.
 Another backend goes in with `@provide(override=True)` in a provider listed after it.
@@ -391,6 +398,11 @@ fields existed is read as enabled, `"wait"` and 30 seconds.
     `in_flight="run"` mode the fingerprint is checked on the read before the action; a
     mismatch discovered on the collision after the action is logged and counted as
     `key_reuse`, and the caller keeps its own result, because the side effect has happened.
+25. **The Dishka key for the client is `redis.asyncio.Redis | RedisCluster`, exactly.**
+    `AsyncRedisIdempotencyProvider` requests that union — the key
+    `redis_client_kit.AsyncRedisClient` names — and Dishka matches keys, not subclasses. A
+    provider annotated `-> Redis` fails the container at construction with
+    `GraphMissingFactoryError`; change the annotation, not the client.
 
 ## Common mistakes
 

docs/api_reference.md

@@ -105,10 +105,10 @@ Interface for metrics collection.
 Redis implementation of the repository protocol.
 
 - **Constructor**:
-  - `redis` (`redis.asyncio.Redis`): Any async Redis client, including a subclass such as an instrumented or fake one.
+  - `redis` (`redis.asyncio.Redis | RedisCluster`): Any async Redis client, single-node or cluster, including a subclass such as an instrumented or fake one.
   - `key_prefix` (str, default: "idempotency:"): Prefix for all Redis keys. (**keyword-only**)
   - `metrics` (IdempotencyMetricsProtocol, optional): Metrics collector. (**keyword-only**)
-- **Storage**: `SET key value EX ttl NX` for `save`, the same without `NX` for `replace`, `GET` for `get`.
+- **Storage**: `SET key value EX ttl NX` for `save`, the same without `NX` for `replace`, `GET` for `get`, `MGET` for `get_many` — one per hash slot, via `mget_nonatomic`, on a `RedisCluster`.
 
 ## Exceptions
 

docs/architecture.md

@@ -147,6 +147,8 @@ This distinction allows developers to decide whether to fail the request or proc
 
 Bulk operations (`get_many`, `save_many`, `delete_many`) are designed to be efficient by using Redis `MGET` and non-transactional pipelines.
 
+**`MGET` per hash slot**: on a cluster a single `MGET` may not span hash slots — redis-py raises `RedisClusterException` before sending — so with a `RedisCluster` client `get_many` calls `mget_nonatomic`, which issues one `MGET` per slot and returns the values in input order. `delete_many` and the expired-key cleanup rely on the cluster client's own per-slot split of `DEL`.
+
 **Non-transactional pipelines** (`transaction=False`) are used for `save_many` to ensure compatibility with Redis Cluster. In a cluster environment, different keys can map to different hash slots, making standard `MULTI/EXEC` transactions impossible for arbitrary keys. By using a non-transactional pipeline, we send all commands in a single network round-trip while allowing them to be processed independently across different cluster nodes.
 
 ## Metrics and Observability

docs/user_guide.md

@@ -327,7 +327,7 @@ except IdempotencyValidationError as e:
 
 ## Redis Cluster Compatibility
 
-The `RedisAsyncIdempotencyRepository` is fully compatible with Redis Cluster. It uses non-transactional pipelines (`transaction=False`) for bulk operations, allowing keys to be distributed across different hash slots.
+The `RedisAsyncIdempotencyRepository` takes a `redis.asyncio.RedisCluster` as well as a `Redis` and is fully compatible with Redis Cluster. Every write is single-key. `get_many` sends one `MGET` per hash slot on a cluster (redis-py's `mget_nonatomic`), because a single `MGET` cannot span slots. `save_many` uses a non-transactional pipeline (`transaction=False`), allowing keys to be distributed across different hash slots.
 
 **Note on Atomicity**: Since non-transactional pipelines are used, `save_many` is **not atomic** by default. If an error occurs, some records might remain in Redis. Use `rollback_on_error=True` if you need to ensure that either all records are saved or none (the library will manually delete successfully saved records if a failure occurs).
 
@@ -742,7 +742,7 @@ def create_api_container(settings: Settings) -> AsyncContainer:
     )
 ```
 
-Your own providers supply the `Redis` client and the settings object; the settings object must satisfy `IdempotencySettingsProtocol`, which `BaseIdempotencySettings` does. `IdempotencyProvider` also provides the metrics collector: `PrometheusIdempotencyMetrics` when `metrics_enabled` is true (install the `prometheus` extra), a no-op collector otherwise. To use another metrics backend, provide `IdempotencyMetricsProtocol` yourself with `@provide(override=True)` in a provider listed after `IdempotencyProvider()`.
+Your own providers supply the Redis client and the settings object. The client is requested as `redis.asyncio.Redis | RedisCluster` — the key `redis_client_kit.AsyncRedisClient` names, so `redis_client_kit.providers.AsyncRedisProvider` stands in for `RedisProvider()` above with no adapter in between. A provider of your own must use that exact annotation: Dishka matches keys exactly, and `Redis` alone is a different one. The settings object must satisfy `IdempotencySettingsProtocol`, which `BaseIdempotencySettings` does. `IdempotencyProvider` also provides the metrics collector: `PrometheusIdempotencyMetrics` when `metrics_enabled` is true (install the `prometheus` extra), a no-op collector otherwise. To use another metrics backend, provide `IdempotencyMetricsProtocol` yourself with `@provide(override=True)` in a provider listed after `IdempotencyProvider()`.
 
 `enabled` on the settings object is the kill switch: with `enabled=False` the coordinator these providers build runs the action and nothing else — no read, no write, no metric. `in_flight` and `in_flight_lease_seconds` reach the coordinator the same way. A settings object that predates these fields is read as enabled, `"wait"` and 30 seconds.
 

idempotency_kit/dishka/aio/redis.py

@@ -1,7 +1,7 @@
 """Dishka provider for Redis-backed async idempotency repository."""
 
 from dishka import Provider, Scope, provide
-from redis.asyncio import Redis as AsyncRedisClient
+from redis.asyncio import Redis, RedisCluster
 
 from idempotency_kit import AsyncIdempotencyRepository, IdempotencyMetricsProtocol
 from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository
@@ -10,14 +10,20 @@
 
 
 class AsyncRedisIdempotencyProvider(Provider):
-    """Provider for async Redis-backed idempotency repository."""
+    """Provider for async Redis-backed idempotency repository.
+
+    The client is requested as ``Redis | RedisCluster`` -- the key
+    ``redis_client_kit.AsyncRedisClient`` names, so redis-client-kit's
+    ``AsyncRedisProvider`` satisfies it as is. Dishka matches keys exactly:
+    a provider of your own has to use that annotation, not ``Redis`` alone.
+    """
 
     scope = Scope.APP
 
     @provide
     def get_repository(
         self,
-        redis: AsyncRedisClient,
+        redis: Redis | RedisCluster,
         settings: IdempotencySettingsProtocol,
         metrics: IdempotencyMetricsProtocol,
     ) -> AsyncIdempotencyRepository:

idempotency_kit/infra/storage/redis/aio/repository.py

@@ -17,9 +17,10 @@
     _HAS_ORJSON = False
 
 try:
-    from redis.asyncio import Redis as AsyncRedisClient
+    from redis.asyncio import Redis, RedisCluster
 
     _HAS_REDIS = True
+    AsyncRedisClient = Redis | RedisCluster
 except ImportError:
     _HAS_REDIS = False
     AsyncRedisClient = Any  # type: ignore[assignment,misc,valid-type]
@@ -61,7 +62,7 @@ def __init__(
         """Initialize repository.
 
         Args:
-            redis: Redis client instance
+            redis: Redis client instance, single-node or cluster
             key_prefix: Prefix for all Redis keys (default: "idempotency:")
             metrics: Metrics collector instance (default: NoOp)
         """
@@ -316,7 +317,7 @@ async def delete(self, operation: str, idempotency_key: str) -> bool:
             self._metrics.record_latency(operation, "delete", time.perf_counter() - start)
 
     async def get_many(self, operation: str, idempotency_keys: list[str]) -> dict[str, IdempotencyRecord]:
-        """Retrieve multiple records from Redis using MGET.
+        """Retrieve multiple records from Redis using MGET, one per hash slot on a cluster.
 
         Args:
             operation: Name of the operation
@@ -340,7 +341,11 @@ async def get_many(self, operation: str, idempotency_keys: list[str]) -> dict[st
 
             keys = [self._make_key(operation, key) for key in idempotency_keys]
             try:
-                values = await self._redis.mget(keys)
+                # One MGET cannot span hash slots on a cluster; mget_nonatomic sends one per slot.
+                if isinstance(self._redis, RedisCluster):
+                    values = await self._redis.mget_nonatomic(keys)
+                else:
+                    values = await self._redis.mget(keys)
             except Exception as e:
                 self._metrics.record_error(operation, type(e).__name__)
                 logger.exception(

pyproject.toml

@@ -52,6 +52,7 @@ test = [
     "orjson>=3.11.7,<4.0.0",
     "dishka>=1.0.0",
     "prometheus-client>=0.15.0",
+    "redis-client-kit[providers,settings]>=0.2.0",
 ]
 docs = [
     "zensical>=0.0.37",

tests/unit/dishka/test_providers.py

@@ -4,7 +4,10 @@
 from dishka import Provider, Scope, make_async_container, provide
 from fakeredis.aioredis import FakeRedis
 from prometheus_client import REGISTRY
-from redis.asyncio import Redis as AsyncRedisClient
+from redis.asyncio import Redis, RedisCluster
+from redis_client_kit.config import RedisSettingsProtocol
+from redis_client_kit.providers import AsyncRedisProvider
+from redis_client_kit.settings import BaseRedisSettings
 
 from idempotency_kit import (
     AsyncIdempotencyCoordinator,
@@ -21,6 +24,7 @@
     IdempotencySettingsProtocol,
 )
 from idempotency_kit.infra.metrics.prometheus import PrometheusIdempotencyMetrics
+from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository
 from idempotency_kit.settings import BaseIdempotencySettings
 
 
@@ -55,11 +59,27 @@ def settings(self) -> IdempotencySettingsProtocol:
         )
 
     @provide
-    def redis(self) -> AsyncRedisClient:
-        """Provide a fake Redis client (FakeRedis subclasses redis.asyncio.Redis)."""
+    def redis(self) -> Redis | RedisCluster:
+        """Provide a fake Redis client (FakeRedis subclasses redis.asyncio.Redis) under the shipped provider's key."""
         return FakeRedis()
 
 
+class _RedisClientKitAppProvider(Provider):
+    """Application-side provider for the redis-client-kit wiring: its settings object and ours."""
+
+    scope = Scope.APP
+
+    @provide
+    def redis_settings(self) -> RedisSettingsProtocol:
+        """Provide redis-client-kit settings; nothing connects while the startup health check is off."""
+        return BaseRedisSettings(key_prefix="probe")
+
+    @provide
+    def settings(self) -> IdempotencySettingsProtocol:
+        """Provide idempotency settings."""
+        return BaseIdempotencySettings(key_prefix="probe:", metrics_enabled=False)
+
+
 @pytest.mark.asyncio
 async def test__shipped_providers__metrics_disabled__container_resolves_coordinator() -> None:
     """Test that the shipped providers alone can build a container and resolve the coordinator."""
@@ -221,3 +241,26 @@ async def test__shipped_providers__in_flight_settings__reach_the_coordinator() -
         assert coordinator._in_flight_lease_seconds == 5
     finally:
         await container.close()
+
+
+@pytest.mark.asyncio
+async def test__shipped_providers__redis_client_kit_client__container_builds_and_resolves_repository() -> None:
+    """The client redis-client-kit provides is the one the Redis provider asks for: no adapter in between (#32)."""
+    # Arrange
+    container = make_async_container(
+        AsyncRedisProvider(check_health_on_startup=False),
+        _RedisClientKitAppProvider(),
+        IdempotencyProvider(),
+        AsyncRedisIdempotencyProvider(),
+        AsyncIdempotencyCoordinatorProvider(),
+    )
+
+    # Act
+    try:
+        repository = await container.get(AsyncIdempotencyRepository)
+
+        # Assert
+        assert isinstance(repository, RedisAsyncIdempotencyRepository)
+        assert isinstance(repository._redis, Redis)
+    finally:
+        await container.close()

tests/unit/storage/redis/aio/test_bulk.py

@@ -6,6 +6,7 @@
 import orjson
 import pytest
 from fakeredis import FakeAsyncRedis as AsyncRedisClient
+from redis.asyncio import RedisCluster
 
 from idempotency_kit import (
     IdempotencyDomainService,
@@ -364,3 +365,22 @@ async def test_save_many_mixed_validation_errors(fake_redis: AsyncRedisClient) -
 
     with pytest.raises(IdempotencyValidationError, match="Bulk validation failed"):
         await repo.save_many([invalid_op, invalid_key])
+
+
+@pytest.mark.asyncio
+async def test__get_many__cluster_client__reads_with_one_mget_per_slot() -> None:
+    """A single MGET cannot span hash slots on a cluster, so the repository asks redis-py for one per slot."""
+    # Arrange
+    record = IdempotencyDomainService().create_record(operation="bulk", idempotency_key="key1", result={"n": 1})
+    cluster = AsyncMock(spec=RedisCluster)
+    cluster.mget_nonatomic.return_value = [orjson.dumps(record.model_dump(mode="json")), None]
+    repo = RedisAsyncIdempotencyRepository(cluster)
+
+    # Act
+    results = await repo.get_many("bulk", ["key1", "key2"])
+
+    # Assert
+    assert set(results) == {"key1"}
+    assert results["key1"].result == {"n": 1}
+    cluster.mget_nonatomic.assert_awaited_once_with(["idempotency:bulk:key1", "idempotency:bulk:key2"])
+    cluster.mget.assert_not_called()