test: Sentinel failover drills (just test-dr)

AGENTS.md

@@ -15,6 +15,9 @@ a concept in code, a test name, or an issue title.
   interpreter (the CI matrix, or `just test` in docker), not just the no-Redis subset. An
   integration-only failure, such as a broker-shutdown hang that only bites under a specific
   interpreter's async timing, won't surface in the no-Redis subset alone.
+- `just test-dr` runs the Sentinel failover drills in `tests_dr/` (primary killed, primary
+  paused), each on a fresh cluster from `docker-compose.dr.yml`: a primary, a replica that Sentinel
+  promotes, and three sentinels. It needs the Docker socket, takes about a minute, and is not in CI.
 
 ## Workflow
 

docker-compose.dr.yml

@@ -0,0 +1,62 @@
+name: timers-dr
+
+x-sentinel: &sentinel
+  image: redis:8
+  entrypoint: ["sh", "/sentinel-entrypoint.sh"]
+  volumes:
+    - ./tests_dr/sentinel-entrypoint.sh:/sentinel-entrypoint.sh:ro
+  depends_on:
+    redis-replica:
+      condition: service_healthy
+  healthcheck:
+    test: ["CMD-SHELL", "redis-cli -p 26379 sentinel ckquorum mymaster | grep -q OK"]
+    interval: 1s
+    timeout: 3s
+    retries: 60
+
+services:
+  dr:
+    build:
+      context: .
+      dockerfile: ./tests_dr/Dockerfile
+    user: root
+    volumes:
+      - .:/code
+      - /code/.venv
+      - /var/run/docker.sock:/var/run/docker.sock
+    depends_on:
+      sentinel-1:
+        condition: service_healthy
+      sentinel-2:
+        condition: service_healthy
+      sentinel-3:
+        condition: service_healthy
+    environment:
+      - SENTINELS=sentinel-1:26379,sentinel-2:26379,sentinel-3:26379
+
+  redis-primary:
+    image: redis:8
+    container_name: timers-dr-primary
+    command: ["redis-server", "--replica-announce-ip", "redis-primary"]
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 1s
+      timeout: 3s
+      retries: 30
+
+  redis-replica:
+    image: redis:8
+    container_name: timers-dr-replica
+    command: ["redis-server", "--replicaof", "redis-primary", "6379", "--replica-announce-ip", "redis-replica"]
+    depends_on:
+      redis-primary:
+        condition: service_healthy
+    healthcheck:
+      test: ["CMD-SHELL", "redis-cli info replication | grep -q master_link_status:up"]
+      interval: 1s
+      timeout: 3s
+      retries: 60
+
+  sentinel-1: *sentinel
+  sentinel-2: *sentinel
+  sentinel-3: *sentinel

justfile

@@ -9,6 +9,17 @@ sh:
 test *args: down && down
     docker compose run application uv run pytest {{ args }}
 
+# Sentinel failover drills (kill, pause the primary), each on a fresh cluster. Not run in CI.
+test-dr:
+    #!/usr/bin/env sh
+    set -eu
+    compose="docker compose -f docker-compose.dr.yml"
+    trap '$compose down --remove-orphans --volumes' EXIT
+    for scenario in kill pause; do
+        $compose down --remove-orphans --volumes
+        $compose run --rm --build dr uv run pytest tests_dr -s -k "$scenario"
+    done
+
 test-ci:
     uv run --no-sync pytest --cov=. --cov-report term-missing --cov-report xml
 

pyproject.toml

@@ -70,7 +70,7 @@ isort.lines-after-imports = 2
 isort.no-lines-before = ["standard-library", "local-folder"]
 
 [tool.ruff.lint.extend-per-file-ignores]
-"tests/**" = [
+"{tests,tests_dr}/**" = [
     "S101", # allow asserts
     "PLR2004", # allow magic values
 ]
@@ -88,6 +88,9 @@ Repository = "https://github.com/modern-python/faststream-redis-timers"
 Issues = "https://github.com/modern-python/faststream-redis-timers/issues"
 Changelog = "https://github.com/modern-python/faststream-redis-timers/releases"
 
+[tool.coverage.run]
+omit = ["tests_dr/*"]
+
 [tool.coverage.report]
 fail_under = 100
 exclude_also = ["if typing.TYPE_CHECKING:"]

tests_dr/Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.13-slim
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
+
+ENV UV_PROJECT_ENVIRONMENT=/code/.venv \
+    UV_NO_MANAGED_PYTHON=1 \
+    UV_NO_CACHE=true \
+    UV_LINK_MODE=copy
+
+WORKDIR /code
+
+COPY pyproject.toml uv.lock ./
+
+RUN uv sync --all-extras --frozen --no-install-project

tests_dr/sentinel-entrypoint.sh

@@ -0,0 +1,11 @@
+#!/bin/sh
+set -eu
+cat > /tmp/sentinel.conf <<CONF
+port 26379
+sentinel resolve-hostnames yes
+sentinel announce-hostnames yes
+sentinel monitor mymaster redis-primary 6379 2
+sentinel down-after-milliseconds mymaster 2000
+sentinel failover-timeout mymaster 10000
+CONF
+exec redis-sentinel /tmp/sentinel.conf

tests_dr/test_failover.py

@@ -0,0 +1,155 @@
+"""Sentinel failover drills. Run via ``just test-dr``, one fresh cluster per scenario."""
+
+import asyncio
+import os
+import time
+from collections.abc import Awaitable, Callable
+from datetime import timedelta
+
+import pytest
+from redis.asyncio import Redis
+from redis.asyncio.sentinel import Sentinel
+
+from faststream_redis_timers import TimersBroker
+
+
+SENTINELS = [
+    (host, int(port)) for host, port in (s.split(":") for s in os.environ.get("SENTINELS", "").split(",") if s)
+]
+SERVICE = "mymaster"
+DOCKER_SOCKET = "/var/run/docker.sock"
+PRIMARY_CONTAINER = "timers-dr-primary"
+PRIMARY_HOST = "redis-primary"
+TOPIC = "dr"
+BEFORE = 50
+DURING = 50
+AFTER = 50
+DURING_SPREAD_SECONDS = 10.0
+LEASE_TTL_SECONDS = 5
+RECOVERY_BUDGET_SECONDS = 90.0
+PUBLISH_ATTEMPT_TIMEOUT_SECONDS = 5.0
+TEST_BUDGET_SECONDS = 240.0
+
+pytestmark = pytest.mark.skipif(not SENTINELS, reason="needs the docker-compose.dr.yml cluster")
+
+
+async def _docker(method: str, path: str) -> None:
+    reader, writer = await asyncio.open_unix_connection(DOCKER_SOCKET)
+    writer.write(f"{method} {path} HTTP/1.1\r\nHost: docker\r\nConnection: close\r\nContent-Length: 0\r\n\r\n".encode())
+    await writer.drain()
+    status_line = await reader.readline()
+    writer.close()
+    assert int(status_line.split(b" ", 2)[1]) < 300, status_line.decode(errors="replace")
+
+
+async def _kill_primary() -> None:
+    await _docker("POST", f"/containers/{PRIMARY_CONTAINER}/kill")
+
+
+async def _pause_primary() -> None:
+    """Freeze the primary: its name resolves and the kernel accepts connects, but nothing ever answers."""
+    await _docker("POST", f"/containers/{PRIMARY_CONTAINER}/pause")
+
+
+async def _wait_until(predicate: Callable[[], bool], *, deadline: float, what: str) -> None:
+    while not predicate():
+        if time.monotonic() > deadline:
+            msg = f"timed out waiting for {what}"
+            raise AssertionError(msg)
+        await asyncio.sleep(0.1)
+
+
+async def _wait_sentinels_see_replica() -> None:
+    deadline = time.monotonic() + 30
+    for host, port in SENTINELS:
+        sentinel = Redis(host=host, port=port, decode_responses=True)
+        try:
+            while not await sentinel.sentinel_slaves(SERVICE):
+                assert time.monotonic() < deadline, f"{host} never discovered the replica"
+                await asyncio.sleep(0.5)
+        finally:
+            await sentinel.aclose()  # ty: ignore[unresolved-attribute]
+
+
+async def _wait_failed_over(deadline: float) -> None:
+    while True:
+        sentinel_client = Redis(host=SENTINELS[0][0], port=SENTINELS[0][1], decode_responses=True)
+        try:
+            address = await sentinel_client.sentinel_get_master_addr_by_name(SERVICE)
+        finally:
+            await sentinel_client.aclose()  # ty: ignore[unresolved-attribute]
+        if address and address[0] != PRIMARY_HOST:
+            return
+        assert time.monotonic() < deadline, "sentinels never promoted the replica"
+        await asyncio.sleep(0.2)
+
+
+async def _publish(broker: TimersBroker, n: int, activate_in: timedelta = timedelta(0)) -> None:
+    await broker.publish({"n": n}, topic=TOPIC, timer_id=f"t-{n}", activate_in=activate_in)
+
+
+async def _publish_until_accepted(broker: TimersBroker, n: int, deadline: float) -> None:
+    while True:
+        try:
+            async with asyncio.timeout(PUBLISH_ATTEMPT_TIMEOUT_SECONDS):
+                await _publish(broker, n)
+        except Exception:
+            if time.monotonic() > deadline:
+                raise
+            await asyncio.sleep(0.5)
+        else:
+            return
+
+
+@pytest.mark.parametrize("fail_primary", [_kill_primary, _pause_primary], ids=["kill", "pause"])
+async def test_timers_survive_failover(fail_primary: Callable[[], Awaitable[None]]) -> None:
+    sentinel = Sentinel(SENTINELS)
+    client = sentinel.master_for(SERVICE)
+    broker = TimersBroker(client)
+    handled_at: dict[int, float] = {}
+    deliveries: list[int] = []
+
+    @broker.subscriber(TOPIC, lease_ttl=LEASE_TTL_SECONDS)
+    async def handle(body: dict[str, int]) -> None:
+        deliveries.append(body["n"])
+        handled_at.setdefault(body["n"], time.monotonic())
+
+    await _wait_sentinels_see_replica()
+    async with asyncio.timeout(TEST_BUDGET_SECONDS), broker:
+        for n in range(BEFORE):
+            await _publish(broker, n)
+        await _wait_until(
+            lambda: len(handled_at) == BEFORE,
+            deadline=time.monotonic() + 30,
+            what="pre-failover timers",
+        )
+        step = DURING_SPREAD_SECONDS / DURING
+        for i, n in enumerate(range(BEFORE, BEFORE + DURING)):
+            await _publish(broker, n, activate_in=timedelta(seconds=1 + i * step))
+        assert await client.execute_command("WAIT", 1, 5000) == 1, "replica did not acknowledge the writes"
+
+        failed_at = time.monotonic()
+        await fail_primary()
+        deadline = failed_at + RECOVERY_BUDGET_SECONDS
+        await _wait_failed_over(deadline)
+        promoted = time.monotonic() - failed_at
+        total = BEFORE + DURING + AFTER
+        for n in range(BEFORE + DURING, total):
+            await _publish_until_accepted(broker, n, deadline)
+        publish_recovered = time.monotonic() - failed_at
+        await _wait_until(
+            lambda: len(handled_at) == total,
+            deadline=deadline,
+            what="post-failover timers",
+        )
+        subscriber_recovered = max(handled_at[n] for n in range(BEFORE, total)) - failed_at
+
+        while leftover := await broker.get_pending_timers(TOPIC):
+            assert time.monotonic() < deadline, f"timers never acked: {leftover}"
+            await asyncio.sleep(0.5)
+
+    print(  # noqa: T201
+        f"\npromoted {promoted:.1f}s, publish recovered {publish_recovered:.1f}s, "
+        f"subscriber recovered {subscriber_recovered:.1f}s, redeliveries {len(deliveries) - len(handled_at)}",
+    )
+    await client.aclose()  # ty: ignore[unresolved-attribute]