feat: add an opt-in write probe to the health checks

README.md

@@ -251,6 +251,11 @@ from redis_client_kit import check_async_redis_health
 is_healthy = await check_async_redis_health(client)
 if is_healthy:
     print("Redis is ready!")
+
+# PING alone says healthy for a read-only replica and for a primary that is out of memory
+# under noeviction. write_key adds a SET <key> 1 EX 60 after the ping, for a service that
+# needs Redis for more than reads.
+is_healthy = await check_async_redis_health(client, write_key="myapp:health")
 ```
 
 ### Graceful Shutdown

docs/agents.md

@@ -60,8 +60,9 @@ Four nouns and one direction of travel.
 * **Metrics** — anything satisfying `RedisMetricsProtocol`
   (`record_command`, `record_error`, `record_pool_stats`). `RedisMetrics` from the
   `metrics` extra is a Prometheus implementation of it.
-* **Lifecycle** — `check_*_redis_health(client)` pings and returns a bool;
-  `close_*_redis_client(client)` closes and swallows. Neither raises.
+* **Lifecycle** — `check_*_redis_health(client)` pings and returns a bool, and with
+  `write_key=` writes that key as well; `close_*_redis_client(client)` closes and
+  swallows. Neither raises.
 
 Nothing here holds state of its own. The client is the state, and it is `redis-py`'s.
 
@@ -201,8 +202,8 @@ Everything in this table is importable from `redis_client_kit` itself.
 |---|---|---|
 | `create_async_redis_client` | `(settings, metrics=None)` | `Redis | RedisCluster`, instrumented when `metrics` is given |
 | `create_redis_client` | `(settings, metrics=None)` | the sync equivalent |
-| `check_async_redis_health` | `await (client)` | `bool` — never raises |
-| `check_redis_health` | `(client)` | `bool` — never raises |
+| `check_async_redis_health` | `await (client, write_key=None)` | `bool` — never raises; `write_key` adds `SET <write_key> 1 EX 60` after the ping |
+| `check_redis_health` | `(client, write_key=None)` | the sync equivalent |
 | `close_async_redis_client` | `await (client)` | `None` — shielded, 10 s timeout, never raises |
 | `close_redis_client` | `(client)` | `None` — never raises |
 | `build_base_redis_kwargs` | `(settings, asyncio=False)` | `dict[str, object]` of `redis-py` keyword arguments; `asyncio=True` for a `redis.asyncio` client |
@@ -222,7 +223,7 @@ The rest lives one import deeper.
 | `redis_client_kit.sync` | — | `SyncRedisClient`, `InstrumentedRedis`, `InstrumentedRedisCluster`, `create_redis_client`, `check_redis_health`, `close_redis_client` |
 | `redis_client_kit.config` | — | `RedisSettingsProtocol` and its parts: `RedisConnectionProtocol`, `RedisClusterProtocol`, `RedisPoolProtocol`, `RedisRetryProtocol`, `RedisSSLProtocol`, `RedisResponseProtocol` |
 | `redis_client_kit.protocols` | — | `RedisMetricsProtocol` |
-| `redis_client_kit.utils` | — | the three exported helpers, plus `mask_redis_kwargs(kwargs)` for logging |
+| `redis_client_kit.utils` | — | the three exported helpers, plus `mask_redis_kwargs(kwargs)` for logging and `WRITE_PROBE_TTL_S`, the write probe's expiry in seconds |
 | `redis_client_kit.settings` | `settings` | `BaseRedisSettings`, `RedisConnectionSettings`, `RedisClusterSettings`, `RedisPoolSettings`, `RedisRetrySettings`, `RedisSSLSettings`, `RedisResponseSettings` |
 | `redis_client_kit.metrics` | `metrics` | `RedisMetrics`, `REDIS_COMMAND_DURATION_BUCKETS` |
 | `redis_client_kit.providers` | `providers` | `AsyncRedisProvider(check_health_on_startup=True, provide_default_metrics=True)` |
@@ -328,9 +329,18 @@ See rules 15 to 17.
    `connection=RedisConnectionSettings(host="…")`.
 9. **`health_check_interval` is `redis-py`'s per-connection ping interval**, not the
    `check_*_redis_health` function. `0` and `None` both disable it.
-10. **The health check never raises and never says "maybe".** It returns `False` on any
-    error and logs it. A cluster `ping()` returns one entry per node, and the check is
-    `True` only when every node answered truthily.
+10. **The health check never raises and never says "maybe", and by default it only
+    pings.** It returns `False` on any error and logs it. A cluster `ping()` returns one
+    entry per node, and the check is `True` only when every node answered truthily. But
+    `PING` is a liveness answer, not a readiness one: a read-only replica and a server at
+    `maxmemory` under `noeviction` both reply `PONG` and refuse every write. Pass
+    `write_key="myapp:health"` and the check runs `SET <write_key> 1 EX 60`
+    (`WRITE_PROBE_TTL_S`) after the ping, returning `False` on `ReadOnlyError`,
+    `OutOfMemoryError` or anything else the write raises. The key is yours to name and
+    yours to prefix — `key_prefix` is applied to nothing (rule 7) — and it is left to
+    expire rather than deleted. It costs one more round trip under the same
+    `socket_timeout`, and on a cluster the write reaches only the node owning that key's
+    slot, so `True` there means every node answered and one of them took a write.
 11. **Closing never raises either.** `close_async_redis_client` shields `aclose()` and
     gives it 10 seconds (`ACLOSE_TIMEOUT_S`); a timeout or a broken close is a warning in
     the log, not an exception. Only `CancelledError` and `KeyboardInterrupt` propagate.
@@ -436,6 +446,16 @@ if not await check_async_redis_health(client):
     raise RuntimeError("Redis is not answering")
 ```
 
+```python
+# WRONG — readiness for a service that writes, from a check a read-only replica passes
+if not await check_async_redis_health(client):
+    return Response("unready", status_code=503)
+
+# RIGHT — ask for the write you are going to need
+if not await check_async_redis_health(client, write_key="myapp:health"):
+    return Response("unready", status_code=503)
+```
+
 ```python
 # WRONG — a per-request client, and a close that can take the request down with it
 async def handler(settings):
@@ -467,7 +487,9 @@ Python, by Pydantic or by `redis-py`.
 | `ImportError` from an optional submodule | the extra is not installed; the message names it |
 
 `check_*_redis_health` and `close_*_redis_client` convert `redis-py`'s errors into a
-`False` and a log line respectively — they are the two places that swallow.
+`False` and a log line respectively — they are the two places that swallow. A refused
+write probe goes the same way: `ReadOnlyError` and `OutOfMemoryError` come back as `False`
+with a warning in the log, not as an exception.
 
 ## Documentation map
 

docs/guide/advanced.md

@@ -240,6 +240,16 @@ is_healthy = await check_async_redis_health(client)
 # Returns True only if all nodes respond
 ```
 
+The [write probe](quickstart.md#health-checks) works here too, with one caveat: the ping
+still reaches every node, but `SET` reaches only the node that owns that key's slot.
+
+```python
+is_healthy = await check_async_redis_health(client, write_key="myapp:health")
+```
+
+`True` means every node answered the ping and one of them accepted a write. To cover the
+writability of every slot you would need a key per slot, which this check does not do.
+
 ### Read from Replicas
 
 Enable reading from replicas for read-heavy workloads:

docs/guide/quickstart.md

@@ -95,6 +95,24 @@ else:
     print("Redis is not available")
 ```
 
+That is a `PING`, and `PING` answers `PONG` from a server that cannot take a write: a
+read-only replica that a failover or a DNS mistake pointed you at, or a primary at
+`maxmemory` under `noeviction`. Both are healthy for a reader and down for everyone else.
+If your service needs Redis for more than reads, hand the check a key and it writes too:
+
+```python
+is_healthy = await check_async_redis_health(client, write_key="myapp:health")
+```
+
+The write is `SET <write_key> 1 EX 60` (`WRITE_PROBE_TTL_S` in `redis_client_kit.utils`),
+run after the ping and only if the ping answered. The key is left to expire on its own.
+Name it yourself, prefix included — this library applies `key_prefix` to nothing. The
+default is unchanged: no `write_key`, no write. `check_redis_health` takes the same
+argument.
+
+The cost is one more round trip, under the same `socket_timeout`, and on a cluster the
+write reaches only the node that owns that key's slot — the ping still covers every node.
+
 ## Graceful Shutdown
 
 Always close the client properly:

redis_client_kit/aio/lifecycle.py

@@ -8,6 +8,7 @@
 
 from redis.exceptions import RedisClusterException, RedisError
 
+from ..utils import WRITE_PROBE_TTL_S
 from .types import AsyncRedisClient
 
 logger = logging.getLogger(__name__)
@@ -36,11 +37,19 @@ async def close_async_redis_client(client: AsyncRedisClient) -> None:
         logger.info("Closed Redis client")
 
 
-async def check_async_redis_health(client: AsyncRedisClient) -> bool:
+async def check_async_redis_health(client: AsyncRedisClient, *, write_key: str | None = None) -> bool:
     """Check async Redis connection health.
 
+    Pings the server, and with ``write_key`` writes that key as well. A ping alone is a
+    liveness answer: a read-only replica and a server at ``maxmemory`` under ``noeviction``
+    both reply PONG and refuse every write. The write probe turns that into a readiness
+    answer for a service that needs Redis for more than reads.
+
     Args:
         client: Redis client (single or cluster) to check
+        write_key: Key to ``SET`` after the ping, expiring after ``WRITE_PROBE_TTL_S``
+            seconds. None, the default, pings only. Prefix it yourself; this package
+            applies no key prefix.
 
     Returns:
         True if Redis is healthy, False otherwise
@@ -50,9 +59,12 @@ async def check_async_redis_health(client: AsyncRedisClient) -> bool:
 
         # Redis Cluster ping returns dict[str, bool] (node_id -> success); single node returns bool.
         if isinstance(result, dict) and result and all(isinstance(k, str) for k in result):
-            return all(bool(v) for v in result.values())
+            healthy = all(bool(v) for v in result.values())
+        else:
+            healthy = bool(result)
 
-        return bool(result)
+        if healthy and write_key is not None:
+            healthy = bool(await client.set(write_key, "1", ex=WRITE_PROBE_TTL_S))
     except (RedisError, RedisClusterException, OSError, ConnectionError, TimeoutError) as e:
         logger.warning(
             "Async Redis health check failed",
@@ -66,6 +78,8 @@ async def check_async_redis_health(client: AsyncRedisClient) -> bool:
         # We return False to indicate unhealthy state but log the full exception for debugging.
         logger.exception("Unexpected error during Redis health check")
         return False
+    else:
+        return healthy
 
 
 __all__ = ["check_async_redis_health", "close_async_redis_client"]

redis_client_kit/sync/lifecycle.py

@@ -7,6 +7,7 @@
 
 from redis.exceptions import RedisClusterException, RedisError
 
+from ..utils import WRITE_PROBE_TTL_S
 from .types import SyncRedisClient
 
 logger = logging.getLogger(__name__)
@@ -28,11 +29,19 @@ def close_redis_client(client: SyncRedisClient) -> None:
         logger.info("Closed Redis client")
 
 
-def check_redis_health(client: SyncRedisClient) -> bool:
+def check_redis_health(client: SyncRedisClient, *, write_key: str | None = None) -> bool:
     """Check sync Redis connection health.
 
+    Pings the server, and with ``write_key`` writes that key as well. A ping alone is a
+    liveness answer: a read-only replica and a server at ``maxmemory`` under ``noeviction``
+    both reply PONG and refuse every write. The write probe turns that into a readiness
+    answer for a service that needs Redis for more than reads.
+
     Args:
         client: Redis client (single or cluster) to check
+        write_key: Key to ``SET`` after the ping, expiring after ``WRITE_PROBE_TTL_S``
+            seconds. None, the default, pings only. Prefix it yourself; this package
+            applies no key prefix.
 
     Returns:
         True if Redis is healthy, False otherwise
@@ -42,9 +51,12 @@ def check_redis_health(client: SyncRedisClient) -> bool:
 
         # Redis Cluster ping returns dict[str, bool] (node_id -> success); single node returns bool.
         if isinstance(result, dict) and result and all(isinstance(k, str) for k in result):
-            return all(bool(v) for v in result.values())
+            healthy = all(bool(v) for v in result.values())
+        else:
+            healthy = bool(result)
 
-        return bool(result)
+        if healthy and write_key is not None:
+            healthy = bool(client.set(write_key, "1", ex=WRITE_PROBE_TTL_S))
     except (RedisError, RedisClusterException, OSError, ConnectionError, TimeoutError) as e:
         logger.warning(
             "Redis health check failed",
@@ -58,6 +70,8 @@ def check_redis_health(client: SyncRedisClient) -> bool:
         # We return False to indicate unhealthy state but log the full exception for debugging.
         logger.exception("Unexpected error during Redis health check")
         return False
+    else:
+        return healthy
 
 
 __all__ = ["check_redis_health", "close_redis_client"]

redis_client_kit/utils.py

@@ -12,6 +12,8 @@
 
 from .config import RedisSettingsProtocol
 
+WRITE_PROBE_TTL_S = 60
+
 
 def parse_redis_url_node(node: str) -> tuple[str, int]:
     """Parse Redis node from host:port, [ipv6]:port or redis:// URL string."""
@@ -143,6 +145,7 @@ def mask_redis_kwargs(kwargs: dict[str, object]) -> dict[str, object]:
 
 
 __all__ = [
+    "WRITE_PROBE_TTL_S",
     "build_base_redis_kwargs",
     "build_redis_retry",
     "mask_redis_kwargs",

tests/integration/conftest.py

@@ -3,6 +3,8 @@
 Automatically applies pytest.mark.integration to all tests in integration/ directory.
 """
 
+from collections.abc import Iterator
+
 try:
     import docker
     from docker.errors import DockerException
@@ -11,10 +13,18 @@
     DockerException = Exception  # type: ignore[assignment, misc]
 
 import pytest
+from redis import Redis
+from redis.exceptions import OutOfMemoryError
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.network import Network
+from testcontainers.core.wait_strategies import LogMessageWaitStrategy
 from testcontainers.redis import RedisContainer
 
 from redis_client_kit import RedisSettingsProtocol
 
+REDIS_IMAGE = "redis:7-alpine"
+REDIS_PORT = 6379
+
 
 def is_docker_available() -> bool:
     """Check if docker is available to run integration tests."""
@@ -117,11 +127,60 @@ def __init__(self) -> None:
 @pytest.fixture(scope="module")
 def redis_container() -> RedisContainer:
     """Provide a Redis container for integration tests."""
-    with RedisContainer("redis:7-alpine") as redis:
+    with RedisContainer(REDIS_IMAGE) as redis:
         yield redis
 
 
 @pytest.fixture
 def fake_redis_settings() -> FakeRedisSettings:
     """Provide fake Redis settings for testing."""
     return FakeRedisSettings()
+
+
+# Servers that answer PING and refuse writes
+
+
+@pytest.fixture(scope="module")
+def full_noeviction_redis() -> Iterator[DockerContainer]:
+    """A primary capped at 1 MB under ``noeviction``, filled until it refuses the next write."""
+    container = (
+        DockerContainer(REDIS_IMAGE)
+        .with_exposed_ports(REDIS_PORT)
+        .with_command("redis-server --maxmemory 1mb --maxmemory-policy noeviction")
+        .waiting_for(LogMessageWaitStrategy("Ready to accept connections"))
+    )
+    with container:
+        fill_until_full(container)
+        yield container
+
+
+@pytest.fixture(scope="module")
+def read_only_replica() -> Iterator[DockerContainer]:
+    """A replica started as ``redis-server --replicaof primary 6379``, synced with a live primary."""
+    with Network() as network:
+        primary = (
+            DockerContainer(REDIS_IMAGE)
+            .with_exposed_ports(REDIS_PORT)
+            .with_network(network)
+            .with_network_aliases("primary")
+        )
+        replica = (
+            DockerContainer(REDIS_IMAGE)
+            .with_exposed_ports(REDIS_PORT)
+            .with_network(network)
+            .with_command(f"redis-server --replicaof primary {REDIS_PORT}")
+            .waiting_for(LogMessageWaitStrategy("MASTER <-> REPLICA sync: Finished with success"))
+        )
+        with primary, replica:
+            yield replica
+
+
+def fill_until_full(container: DockerContainer) -> None:
+    """Write 8 KB values until one is refused, so a test meets a server that is already full."""
+    with Redis(host=container.get_container_host_ip(), port=int(container.get_exposed_port(REDIS_PORT))) as client:
+        for i in range(512):
+            try:
+                client.set(f"fill:{i}", "x" * 8192)
+            except OutOfMemoryError:
+                return
+    raise RuntimeError("Redis took 4 MB of writes under a 1 mb maxmemory cap")