From f5c9fd435257b2fa5220d153265963bb673e2e84 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 28 Oct 2026 18:33:24 +0300 Subject: [PATCH] test: Sentinel failover drills (just test-dr) --- AGENTS.md | 3 + docker-compose.dr.yml | 62 +++++++++++++ justfile | 11 +++ pyproject.toml | 5 +- tests_dr/Dockerfile | 14 +++ tests_dr/__init__.py | 0 tests_dr/sentinel-entrypoint.sh | 11 +++ tests_dr/test_failover.py | 155 ++++++++++++++++++++++++++++++++ 8 files changed, 260 insertions(+), 1 deletion(-) create mode 100644 docker-compose.dr.yml create mode 100644 tests_dr/Dockerfile create mode 100644 tests_dr/__init__.py create mode 100644 tests_dr/sentinel-entrypoint.sh create mode 100644 tests_dr/test_failover.py diff --git a/AGENTS.md b/AGENTS.md index b9f42c3..6d6fe5a 100644 --- a/AGENTS.md +++ b/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 diff --git a/docker-compose.dr.yml b/docker-compose.dr.yml new file mode 100644 index 0000000..8afa1fb --- /dev/null +++ b/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 diff --git a/justfile b/justfile index dfc9d75..22c39ea 100644 --- a/justfile +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index d5e8767..555e3a3 100644 --- a/pyproject.toml +++ b/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:"] diff --git a/tests_dr/Dockerfile b/tests_dr/Dockerfile new file mode 100644 index 0000000..26bd746 --- /dev/null +++ b/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 diff --git a/tests_dr/__init__.py b/tests_dr/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests_dr/sentinel-entrypoint.sh b/tests_dr/sentinel-entrypoint.sh new file mode 100644 index 0000000..ef8196f --- /dev/null +++ b/tests_dr/sentinel-entrypoint.sh @@ -0,0 +1,11 @@ +#!/bin/sh +set -eu +cat > /tmp/sentinel.conf < 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]