Fix the DR drill's replica startup race; add a never-promoted standby (#266)

* Health-check the DR drill's Postgres over TCP so the replica waits for the real server

* Add a never-promoted standby to the failover drills

Author: lesnik512

AGENTS.md

@@ -20,8 +20,9 @@ the vocabulary. Read it before naming a concept in code, a test name, or an issu
   publishes no host port, so run the DB tests through `just test`. The coverage gate is on by
   default, so a partial run trips it; pass `--no-cov` while iterating.
 - `just test-dr` runs the failover drills in `tests_dr/` (primary killed, primary's packets
-  dropped), each on a fresh primary/replica cluster from `docker-compose.dr.yml`. It needs the
-  Docker socket and `NET_ADMIN`, takes a few minutes, and is not in CI.
+  dropped), each on a fresh cluster from `docker-compose.dr.yml`: a primary, a replica that gets
+  promoted, and a standby that stays read-only. It needs the Docker socket and `NET_ADMIN`, takes
+  a few minutes, and is not in CI.
 - Nothing validates Markdown links outside `docs/`. `just docs-build` runs `mkdocs --strict` over
   the site only; root Markdown, `.github/`, and `docs/agents/` are unchecked.
 

docker-compose.dr.yml

@@ -15,8 +15,11 @@ services:
     depends_on:
       pg-replica:
         condition: service_healthy
+      pg-standby:
+        condition: service_healthy
     environment:
       - PRIMARY_DSN=postgresql://outbox:outbox@pg-primary:5432/outbox
+      - STANDBY_DSN=postgresql://outbox:outbox@pg-standby:5432/outbox
       - REPLICA_DSN=postgresql://outbox:outbox@pg-replica:5432/outbox
 
   pg-primary:
@@ -29,7 +32,7 @@ services:
     volumes:
       - ./tests_dr/primary-init.sh:/docker-entrypoint-initdb.d/primary-init.sh:ro
     healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U outbox -d outbox"]
+      test: ["CMD-SHELL", "pg_isready -h 127.0.0.1 -U outbox -d outbox"]
       interval: 1s
       timeout: 3s
       retries: 30
@@ -47,7 +50,25 @@ services:
       pg-primary:
         condition: service_healthy
     healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U outbox -d outbox"]
+      test: ["CMD-SHELL", "pg_isready -h 127.0.0.1 -U outbox -d outbox"]
+      interval: 1s
+      timeout: 3s
+      retries: 60
+
+  pg-standby:
+    image: postgres:17
+    container_name: outbox-dr-standby
+    user: postgres
+    entrypoint: ["bash", "/replica-entrypoint.sh"]
+    environment:
+      - PGPASSWORD=outbox
+    volumes:
+      - ./tests_dr/replica-entrypoint.sh:/replica-entrypoint.sh:ro
+    depends_on:
+      pg-primary:
+        condition: service_healthy
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -h 127.0.0.1 -U outbox -d outbox"]
       interval: 1s
       timeout: 3s
       retries: 60

tests_dr/test_failover.py

@@ -16,6 +16,7 @@
 
 
 PRIMARY_DSN = os.environ.get("PRIMARY_DSN", "")
+STANDBY_DSN = os.environ.get("STANDBY_DSN", "")
 REPLICA_DSN = os.environ.get("REPLICA_DSN", "")
 DOCKER_SOCKET = "/var/run/docker.sock"
 PRIMARY_CONTAINER = "outbox-dr-primary"
@@ -33,19 +34,31 @@
 pytestmark = pytest.mark.skipif(not PRIMARY_DSN, reason="needs the docker-compose.dr.yml cluster")
 
 
-async def _connect_to_primary() -> asyncpg.Connection:
-    """Try each host under its own timeout; asyncpg's ``timeout`` covers all hosts at once."""
-    errors: list[Exception] = []
-    for dsn in (PRIMARY_DSN, REPLICA_DSN):
-        try:
-            async with asyncio.timeout(CONNECT_TIMEOUT_SECONDS):
-                return await asyncpg.connect(
-                    dsn, target_session_attrs="primary", command_timeout=COMMAND_TIMEOUT_SECONDS
-                )
-        except (OSError, asyncpg.TargetServerAttributeNotMatched) as e:
-            errors.append(e)
-    msg = "no primary reachable"
-    raise ExceptionGroup(msg, errors)
+def _primary_connector(rejected_standbys: list[str]) -> Callable[[], Awaitable[asyncpg.Connection]]:
+    """Try each host under its own timeout; asyncpg's ``timeout`` covers all hosts at once.
+
+    The standby sits between the old primary and the promoted replica, so after failover every
+    connect has to reject it as read-only before it reaches the new primary.
+    """
+
+    async def connect() -> asyncpg.Connection:
+        errors: list[Exception] = []
+        for dsn in (PRIMARY_DSN, STANDBY_DSN, REPLICA_DSN):
+            try:
+                async with asyncio.timeout(CONNECT_TIMEOUT_SECONDS):
+                    return await asyncpg.connect(
+                        dsn, target_session_attrs="primary", command_timeout=COMMAND_TIMEOUT_SECONDS
+                    )
+            except asyncpg.TargetServerAttributeNotMatched as e:
+                if dsn == STANDBY_DSN:
+                    rejected_standbys.append(dsn)
+                errors.append(e)
+            except OSError as e:
+                errors.append(e)
+        msg = "no primary reachable"
+        raise ExceptionGroup(msg, errors)
+
+    return connect
 
 
 async def _docker(method: str, path: str, body: dict[str, object] | None = None) -> None:
@@ -99,6 +112,14 @@ async def _wait_replica_caught_up() -> None:
         await replica.close()
 
 
+async def _standby_in_recovery() -> bool:
+    standby = await asyncpg.connect(STANDBY_DSN)
+    try:
+        return bool(await standby.fetchval("SELECT pg_is_in_recovery()"))
+    finally:
+        await standby.close()
+
+
 async def _promote_replica() -> None:
     replica = await asyncpg.connect(REPLICA_DSN)
     try:
@@ -109,7 +130,8 @@ async def _promote_replica() -> None:
 
 @pytest.mark.parametrize("fail_primary", [_kill_primary, _blackhole_primary], ids=["kill", "blackhole"])
 async def test_subscriber_recovers_after_failover(fail_primary: Callable[[], Awaitable[None]]) -> None:
-    engine = create_async_engine("postgresql+asyncpg://", async_creator=_connect_to_primary)
+    rejected_standbys: list[str] = []
+    engine = create_async_engine("postgresql+asyncpg://", async_creator=_primary_connector(rejected_standbys))
     metadata = MetaData()
     table = make_outbox_table(metadata, table_name="dr_outbox")
     async with engine.begin() as conn:
@@ -169,8 +191,10 @@ async def publish_until_accepted(n: int, deadline: float) -> None:
         await publish(BEFORE + AFTER)
         published_at = time.monotonic()
         await _wait_until(lambda: BEFORE + AFTER in handled_at, deadline=published_at + 3, what="NOTIFY wakeup")
+        assert rejected_standbys, "no connect attempt reached the standby"
+        assert await _standby_in_recovery()
 
         print(  # noqa: T201
             f"\npublish recovered {publish_recovered:.1f}s, subscriber recovered {subscriber_recovered:.1f}s, "
-            f"redeliveries {len(deliveries) - len(handled_at)}",
+            f"redeliveries {len(deliveries) - len(handled_at)}, standby rejected {len(rejected_standbys)}x",
         )