From f57f6873488de8c3d44c16329a89a37cfcd52317 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 27 Oct 2026 22:15:29 +0300 Subject: [PATCH 1/2] Bound subscriber statements so loops reconnect after a silent primary loss A fetch or terminal write on a connection whose primary vanished without closing it (power loss, network partition) hung until the kernel's TCP timeout, so the subscriber stopped consuming after a failover. Each fetch, delete, batch delete and retry update is now capped at the new query_timeout_seconds subscriber option (default: lease_ttl_seconds); on timeout the connection is invalidated and the loop reconnects. Adds failover drills (just test-dr): a primary/streaming-replica compose cluster where the primary is either killed or cut off from the network and the replica is promoted. Not part of CI. --- AGENTS.md | 3 + docker-compose.dr.yml | 52 ++++++++ docs/operations/checklist.md | 5 + docs/usage/subscriber.md | 34 ++++++ faststream_outbox/fastapi/router.py | 2 + faststream_outbox/registrator.py | 2 + faststream_outbox/router.py | 2 + faststream_outbox/subscriber/config.py | 25 +++- faststream_outbox/subscriber/factory.py | 4 +- faststream_outbox/subscriber/usecase.py | 52 +++++--- justfile | 11 ++ pyproject.toml | 4 +- tests/test_fastapi.py | 12 ++ tests/test_integration.py | 118 +++++++++++++++++- tests/test_unit.py | 31 ++++- tests_dr/__init__.py | 0 tests_dr/primary-init.sh | 2 + tests_dr/replica-entrypoint.sh | 7 ++ tests_dr/test_failover.py | 152 ++++++++++++++++++++++++ 19 files changed, 494 insertions(+), 24 deletions(-) create mode 100644 docker-compose.dr.yml create mode 100644 tests_dr/__init__.py create mode 100644 tests_dr/primary-init.sh create mode 100644 tests_dr/replica-entrypoint.sh create mode 100644 tests_dr/test_failover.py diff --git a/AGENTS.md b/AGENTS.md index 93ec82e..ec22ca8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -19,6 +19,9 @@ the vocabulary. Read it before naming a concept in code, a test name, or an issu pytest`; `tests/test_integration.py` skips if `POSTGRES_DSN` is unreachable. The compose Postgres 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 cut off from the + network), each on a fresh primary/replica cluster from `docker-compose.dr.yml`. It needs the Docker + socket, 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. diff --git a/docker-compose.dr.yml b/docker-compose.dr.yml new file mode 100644 index 0000000..9596931 --- /dev/null +++ b/docker-compose.dr.yml @@ -0,0 +1,52 @@ +name: outbox-dr + +services: + dr: + build: + context: . + dockerfile: ./Dockerfile + user: root + volumes: + - .:/code + - /code/.venv + - /var/run/docker.sock:/var/run/docker.sock + depends_on: + pg-replica: + condition: service_healthy + environment: + - POSTGRES_DSN=postgresql+asyncpg://outbox:outbox@/outbox?host=pg-primary:5432&host=pg-replica:5432&target_session_attrs=primary + - PRIMARY_DSN=postgresql://outbox:outbox@pg-primary:5432/outbox + - REPLICA_DSN=postgresql://outbox:outbox@pg-replica:5432/outbox + + pg-primary: + image: postgres:17 + container_name: outbox-dr-primary + environment: + - POSTGRES_USER=outbox + - POSTGRES_PASSWORD=outbox + - POSTGRES_DB=outbox + volumes: + - ./tests_dr/primary-init.sh:/docker-entrypoint-initdb.d/primary-init.sh:ro + healthcheck: + test: ["CMD-SHELL", "pg_isready -U outbox -d outbox"] + interval: 1s + timeout: 3s + retries: 30 + + pg-replica: + image: postgres:17 + container_name: outbox-dr-replica + 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 -U outbox -d outbox"] + interval: 1s + timeout: 3s + retries: 60 diff --git a/docs/operations/checklist.md b/docs/operations/checklist.md index 9e1d4b7..0a07039 100644 --- a/docs/operations/checklist.md +++ b/docs/operations/checklist.md @@ -25,6 +25,11 @@ story. cutoff is server-side `make_interval(...)`, immune to clock skew. Tuning: [Subscriber § Slow handlers: dedicated queue](../usage/subscriber.md#slow-handlers-dedicated-queue). +- [ ] Engine URL follows the primary across a failover (a proxy, or a + multi-host URL with `target_session_attrs=primary`). Subscribers with a + long lease set a shorter `query_timeout_seconds`; recovery from a primary + that went silent takes about twice that value. See [Subscriber § Database + failover](../usage/subscriber.md#database-failover). - [ ] Slow handlers segregated onto their own subscriber with a taller `lease_ttl_seconds`. Don't raise it globally; that delays reclaim of *actually* stuck rows everywhere. diff --git a/docs/usage/subscriber.md b/docs/usage/subscriber.md index 6f4bf74..761ddad 100644 --- a/docs/usage/subscriber.md +++ b/docs/usage/subscriber.md @@ -90,6 +90,7 @@ Per-subscriber knobs, passed to `@broker.subscriber("…", …)`: | `min_fetch_interval` | `1.0` s | Base for the adaptive idle backoff (jittered ±50%, so an actual wait can land below it); no sleep at all while fetches keep returning rows | | `max_fetch_interval` | `10.0` s | Ceiling for the adaptive idle backoff (with jitter) | | `lease_ttl_seconds` | `60.0` s | How long a claim is valid before another fetch may reclaim it. Must exceed your handler's P99 with margin. | +| `query_timeout_seconds` | `None` (uses `lease_ttl_seconds`) | Client-side cap on each fetch and terminal write. A statement still running when it expires is abandoned, its connection is closed, and the loop reconnects. Set it below a long lease so a lost primary is noticed sooner. Must be `> 0` and at most `lease_ttl_seconds`. See [Database failover](#database-failover). | | `max_deliveries` | `None` (unbounded) | Total claims (including lease-expiry re-claims) after which the row is dropped without invoking the handler. Defends against handlers that consistently wedge. Must be at least `1`; use `None`, not `0`, for no cap. | | `terminal_flush_batch_size` | `1` (off) | Coalesce completed terminal `DELETE`s into one `DELETE … RETURNING` per N rows. `1` is one round-trip per message (unchanged). Higher trades a wider crash-redelivery window for far fewer round-trips. See [Batching terminal deletes](#batching-terminal-deletes). | | `ack_policy` | `AckPolicy.NACK_ON_ERROR` | See [Ack policy](#ack-policy) | @@ -360,6 +361,39 @@ startup with `FATAL: too many connections`. *Operator-side: [Production checklist § Sizing](../operations/checklist.md#sizing).* +## Database failover + +Point the engine at whichever node is currently primary, either through a proxy +or virtual IP, or with a multi-host URL that asks asyncpg for the writable node: + +```python +engine = create_async_engine( + "postgresql+asyncpg://user:pass@/app?host=pg-1:5432&host=pg-2:5432&target_session_attrs=primary", +) +``` + +Without `target_session_attrs=primary`, a reconnect can land on a standby, where +the fetch fails with a read-only error. + +How fast the subscriber recovers depends on how the old primary went away: + +- **It closed its connections** (the process crashed or was killed). The next + statement fails at once, and the fetch and worker loops back off and + reconnect within seconds. +- **It went silent** (the host lost power, or the network was partitioned). + Statements hang because no error ever arrives. Each fetch and terminal write + is capped at `query_timeout_seconds`, after which the connection is dropped and + the loop reconnects. The fetch loop hits that cap first; the worker hits it + again on its own stale connection once it gets a row. Recovery takes about + `2 × query_timeout_seconds`, which is two minutes with the default 60 s lease. + A subscriber with a long lease for slow handlers should set a shorter + `query_timeout_seconds`. + +With asynchronous replication the new primary can miss the last commits on the +old one. A lost insert disappears together with the business rows from the same +transaction. A lost terminal `DELETE` brings a handled row back, so it is +delivered again; handlers are idempotent for that reason anyway. + ## Read-only inspection `subscriber.get_one()` and `async for msg in subscriber:` are not diff --git a/faststream_outbox/fastapi/router.py b/faststream_outbox/fastapi/router.py index 9b71015..4da284e 100644 --- a/faststream_outbox/fastapi/router.py +++ b/faststream_outbox/fastapi/router.py @@ -167,6 +167,7 @@ def subscriber( # ty: ignore[invalid-method-override] # noqa: PLR0913 min_fetch_interval: float = 1.0, max_fetch_interval: float = 10.0, lease_ttl_seconds: float = 60.0, + query_timeout_seconds: float | None = None, max_deliveries: int | None = None, terminal_flush_batch_size: int = 1, ack_policy: AckPolicy | None = None, @@ -200,6 +201,7 @@ def subscriber( # ty: ignore[invalid-method-override] # noqa: PLR0913 min_fetch_interval=min_fetch_interval, max_fetch_interval=max_fetch_interval, lease_ttl_seconds=lease_ttl_seconds, + query_timeout_seconds=query_timeout_seconds, max_deliveries=max_deliveries, terminal_flush_batch_size=terminal_flush_batch_size, ack_policy=ack_policy, diff --git a/faststream_outbox/registrator.py b/faststream_outbox/registrator.py index 28acd54..e988026 100644 --- a/faststream_outbox/registrator.py +++ b/faststream_outbox/registrator.py @@ -33,6 +33,7 @@ def subscriber( # ty: ignore[invalid-method-override] min_fetch_interval: float = 1.0, max_fetch_interval: float = 10.0, lease_ttl_seconds: float = 60.0, + query_timeout_seconds: float | None = None, max_deliveries: int | None = None, terminal_flush_batch_size: int = 1, ack_policy: AckPolicy | None = None, @@ -56,6 +57,7 @@ def subscriber( # ty: ignore[invalid-method-override] min_fetch_interval=min_fetch_interval, max_fetch_interval=max_fetch_interval, lease_ttl_seconds=lease_ttl_seconds, + query_timeout_seconds=query_timeout_seconds, max_deliveries=max_deliveries, terminal_flush_batch_size=terminal_flush_batch_size, ack_policy=ack_policy, diff --git a/faststream_outbox/router.py b/faststream_outbox/router.py index a919e1a..82895cf 100644 --- a/faststream_outbox/router.py +++ b/faststream_outbox/router.py @@ -31,6 +31,7 @@ def __init__( # noqa: PLR0913 min_fetch_interval: float = 1.0, max_fetch_interval: float = 10.0, lease_ttl_seconds: float = 60.0, + query_timeout_seconds: float | None = None, max_deliveries: int | None = None, terminal_flush_batch_size: int = 1, ack_policy: AckPolicy | None = None, @@ -51,6 +52,7 @@ def __init__( # noqa: PLR0913 min_fetch_interval=min_fetch_interval, max_fetch_interval=max_fetch_interval, lease_ttl_seconds=lease_ttl_seconds, + query_timeout_seconds=query_timeout_seconds, max_deliveries=max_deliveries, terminal_flush_batch_size=terminal_flush_batch_size, ack_policy=ack_policy, diff --git a/faststream_outbox/subscriber/config.py b/faststream_outbox/subscriber/config.py index a3c3161..e54ed12 100644 --- a/faststream_outbox/subscriber/config.py +++ b/faststream_outbox/subscriber/config.py @@ -74,6 +74,7 @@ class OutboxSubscriberConfig(SubscriberUsecaseConfig): min_fetch_interval: float max_fetch_interval: float lease_ttl_seconds: float + query_timeout_seconds: float | None = None max_deliveries: int | None terminal_flush_batch_size: int propagate_inbound_headers: bool @@ -127,6 +128,7 @@ def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob if self.lease_ttl_seconds <= 0: msg = f"lease_ttl_seconds must be > 0, got {self.lease_ttl_seconds}" raise ValueError(msg) + self._validate_query_timeout() if self.min_fetch_interval > self.max_fetch_interval: msg = ( f"min_fetch_interval ({self.min_fetch_interval}) must be <= max_fetch_interval " @@ -169,6 +171,20 @@ def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob f"2 * max_fetch_interval + P99(handler).", ) + def _validate_query_timeout(self) -> None: + if self.query_timeout_seconds is None: + return + if self.query_timeout_seconds <= 0: + msg = f"query_timeout_seconds must be > 0, got {self.query_timeout_seconds}" + raise ValueError(msg) + if self.query_timeout_seconds > self.lease_ttl_seconds: + msg = ( + f"query_timeout_seconds ({self.query_timeout_seconds}) must be <= lease_ttl_seconds " + f"({self.lease_ttl_seconds}): a fetch or terminal write still running past the lease " + f"has already lost the row to reclaim." + ) + raise ValueError(msg) + def _validate_numbers(self) -> None: """Reject bools passed as counts, NaN or infinite intervals, and a ``max_deliveries`` below 1.""" for name in ("max_workers", "fetch_batch_size", "terminal_flush_batch_size", "max_deliveries"): @@ -178,14 +194,19 @@ def _validate_numbers(self) -> None: if isinstance(value, bool) or not isinstance(value, int): msg = f"{name} must be an int, got {value!r}" raise TypeError(msg) - for name in ("min_fetch_interval", "max_fetch_interval", "lease_ttl_seconds"): - if not math.isfinite(getattr(self, name)): + for name in ("min_fetch_interval", "max_fetch_interval", "lease_ttl_seconds", "query_timeout_seconds"): + if getattr(self, name) is not None and not math.isfinite(getattr(self, name)): msg = f"{name} must be a finite number, got {getattr(self, name)}" raise ValueError(msg) if self.max_deliveries is not None and self.max_deliveries < 1: msg = f"max_deliveries must be >= 1 or None (unlimited), got {self.max_deliveries}" raise ValueError(msg) + @property + def query_timeout(self) -> float: + """Client-side cap on each fetch and terminal write; ``lease_ttl_seconds`` unless set.""" + return self.lease_ttl_seconds if self.query_timeout_seconds is None else self.query_timeout_seconds + @property def ack_policy(self) -> AckPolicy: if self._ack_policy is EMPTY: diff --git a/faststream_outbox/subscriber/factory.py b/faststream_outbox/subscriber/factory.py index 8aa8b8d..ce8f81a 100644 --- a/faststream_outbox/subscriber/factory.py +++ b/faststream_outbox/subscriber/factory.py @@ -13,7 +13,7 @@ from faststream_outbox.retry import RetryStrategyProto -def create_subscriber( +def create_subscriber( # noqa: PLR0913 *, queues: list[str], max_workers: int, @@ -25,6 +25,7 @@ def create_subscriber( max_deliveries: int | None, terminal_flush_batch_size: int, config: "OutboxBrokerConfig", + query_timeout_seconds: float | None = None, ack_policy: AckPolicy | None = None, propagate_inbound_headers: bool = False, title_: str | None = None, @@ -43,6 +44,7 @@ def create_subscriber( min_fetch_interval=min_fetch_interval, max_fetch_interval=max_fetch_interval, lease_ttl_seconds=lease_ttl_seconds, + query_timeout_seconds=query_timeout_seconds, max_deliveries=max_deliveries, terminal_flush_batch_size=terminal_flush_batch_size, propagate_inbound_headers=propagate_inbound_headers, diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 5e358d2..adaa00a 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -103,6 +103,8 @@ class _PendingFlush: # the kernel keepalive; cap it and fall back to an immediate terminate. _LISTEN_CLOSE_TIMEOUT = 5.0 +_T = typing.TypeVar("_T") + def _compute_backoff(attempt: int, ceiling: float, *, base: float = 1.0) -> float: """Exponential backoff with ±50% jitter, capped at *ceiling*. @@ -427,11 +429,14 @@ async def _fetch_inner( # fetch_conn is None only in the test-broker path against FakeOutboxClient # (which ignores conn); the real OutboxClient raises if conn is None. The # AbstractOutboxClient surface admits None so both implementations type-check. - rows = await self._client.fetch( + rows = await self._bounded( fetch_conn, - self._queues, - limit=free, - lease_ttl_seconds=self._config.lease_ttl_seconds, + self._client.fetch( + fetch_conn, + self._queues, + limit=free, + lease_ttl_seconds=self._config.lease_ttl_seconds, + ), ) for row in rows: self._inflight.put_nowait(row) @@ -528,6 +533,20 @@ def _on_notify(self, *args: object) -> None: if payload is None or payload in self._served_queues: self._notify_event.set() + async def _bounded(self, conn: "AsyncConnection | None", statement: Awaitable[_T]) -> _T: + """Cap a fetch/terminal write at ``query_timeout`` so a silently dropped socket raises. + + On timeout the connection is invalidated (closed without a round-trip on the dead + socket) and the error propagates, so the outer loop reconnects. + """ + try: + async with asyncio.timeout(self._config.query_timeout): + return await statement + except TimeoutError: + if conn is not None: + await conn.invalidate() + raise + async def _worker_loop(self) -> None: """Thin wrapper around :meth:`_run_with_reconnect` for the worker path.""" await self._run_with_reconnect( @@ -862,7 +881,7 @@ async def _flush_buffer( return pairs = [(p.row.id, p.row.acquired_token) for p in buffer if p.row.acquired_token is not None] try: - deleted = await self._client.delete_batch_with_lease(writer_conn, pairs) + deleted = await self._bounded(writer_conn, self._client.delete_batch_with_lease(writer_conn, pairs)) except BaseException: # incl. CancelledError from the exit-flush timeout: balance task_done before re-raising for pending in buffer: self._emit_dispatch_abandoned(pending.row) @@ -944,11 +963,9 @@ async def _flush_terminal( self._outer_config.last_exception_renderer, ), } - deleted = await self._client.delete_with_lease( + deleted = await self._bounded( writer_conn, - row.id, - row.acquired_token, - dlq_payload=dlq_payload, + self._client.delete_with_lease(writer_conn, row.id, row.acquired_token, dlq_payload=dlq_payload), ) if not deleted: self._emit_lease_lost(row, phase="terminal") @@ -977,14 +994,17 @@ async def _flush_retry( # P19: first_attempt_at / last_attempt_at are the worker's clock; next_attempt_at # is computed server-side (now() + delay) inside mark_pending_with_lease, so the # reschedule time stays clock-skew-immune even though these audit timestamps don't. - updated = await self._client.mark_pending_with_lease( + updated = await self._bounded( writer_conn, - row.id, - row.acquired_token, - delay_seconds=row.pending_delay_seconds, - attempts_count=row.attempts_count, - first_attempt_at=row.first_attempt_at, # ty: ignore[invalid-argument-type] - last_attempt_at=row.last_attempt_at, # ty: ignore[invalid-argument-type] + self._client.mark_pending_with_lease( + writer_conn, + row.id, + row.acquired_token, + delay_seconds=row.pending_delay_seconds, + attempts_count=row.attempts_count, + first_attempt_at=row.first_attempt_at, # ty: ignore[invalid-argument-type] + last_attempt_at=row.last_attempt_at, # ty: ignore[invalid-argument-type] + ), ) if not updated: self._emit_lease_lost(row, phase="retry") diff --git a/justfile b/justfile index a8c09b2..4b2f0a8 100644 --- a/justfile +++ b/justfile @@ -9,6 +9,17 @@ sh: test *args: down && down docker compose run application uv run pytest {{ args }} +# Primary/replica failover drills (kill, network blackhole), 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 blackhole; 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 bf9bec4..56c93c0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -111,7 +111,7 @@ isort.no-lines-before = ["standard-library", "local-folder"] max-args = 15 [tool.ruff.lint.extend-per-file-ignores] -"tests/**" = [ +"{tests,tests_dr}/**" = [ "S101", # allow asserts "PLR2004", # allow magic values ] @@ -132,7 +132,7 @@ Issues = "https://github.com/modern-python/faststream-outbox/issues" Changelog = "https://github.com/modern-python/faststream-outbox/releases" [tool.coverage.run] -omit = ["benchmarks/*"] +omit = ["benchmarks/*", "tests_dr/*"] # SQLAlchemy's async layer runs statements in greenlets; untracked switches drop executed lines. concurrency = ["greenlet", "thread"] diff --git a/tests/test_fastapi.py b/tests/test_fastapi.py index 76d85fd..5041e22 100644 --- a/tests/test_fastapi.py +++ b/tests/test_fastapi.py @@ -37,6 +37,7 @@ from faststream_outbox.fastapi import ( OutboxRouter, ) +from faststream_outbox.subscriber.usecase import OutboxSubscriber from faststream_outbox.testing import TestOutboxBroker, fake_session @@ -92,6 +93,17 @@ async def handle(body: dict) -> None: ... assert record[0].filename == __file__ # attributed to this test (the user), not a package frame +def test_fastapi_router_forwards_query_timeout() -> None: + router = OutboxRouter(outbox_table=_make_outbox_table()) + + @router.subscriber("orders", query_timeout_seconds=5.0) + async def handle(body: dict) -> None: ... + + sub = router.broker.subscribers[0] + assert isinstance(sub, OutboxSubscriber) + assert sub._config.query_timeout == 5.0 # noqa: SLF001 + + async def test_subscriber_receives_fastapi_depends_session() -> None: """INVARIANT: a ``Depends`` inside an outbox handler resolves the same session an HTTP route would. diff --git a/tests/test_integration.py b/tests/test_integration.py index b5d174a..458dbc0 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -6,7 +6,7 @@ import json import logging import uuid -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from typing import Any from unittest import mock @@ -16,7 +16,7 @@ from faststream.kafka import KafkaBroker, TestKafkaBroker from sqlalchemy import MetaData, Table, case, event, func, insert, literal, select, text from sqlalchemy.exc import IntegrityError -from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine from faststream_outbox import ( ConstantRetry, @@ -29,6 +29,7 @@ ) from faststream_outbox.client import OutboxClient from faststream_outbox.envelope import _encode_payload as encode_payload +from faststream_outbox.message import OutboxInnerMessage from faststream_outbox.publisher.fake import OutboxFakePublisher @@ -815,6 +816,119 @@ async def handle(body: dict) -> None: assert len(set(fetch_pids)) == 1, f"persistent connection should hold one pid, saw {set(fetch_pids)}" +async def test_hung_fetch_times_out_and_fetch_loop_reconnects(pg_engine, outbox_table) -> None: + """A fetch stuck on a silently dropped socket must not stall the subscriber forever.""" + received: list[dict] = [] + broker = OutboxBroker(pg_engine, outbox_table=outbox_table, graceful_timeout=2.0) + original_fetch = OutboxClient.fetch + hung: list[bool] = [] + + async def hang_once( + self: OutboxClient, + conn: AsyncConnection | None, + queues: Sequence[str], + *, + limit: int, + lease_ttl_seconds: float, + ) -> list[OutboxInnerMessage]: + if not hung: + hung.append(True) + await asyncio.Event().wait() + return await original_fetch(self, conn, queues, limit=limit, lease_ttl_seconds=lease_ttl_seconds) + + @broker.subscriber("orders", min_fetch_interval=0.05, max_fetch_interval=0.2, lease_ttl_seconds=1.0) + async def handle(body: dict) -> None: + received.append(body) + + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + with mock.patch.object(OutboxClient, "fetch", hang_once): + async with broker: + async with session_factory() as session, session.begin(): + await broker.publish({"order_id": 1}, queue="orders", session=session) + await _wait_until(lambda: received, timeout=10.0) + + assert received == [{"order_id": 1}] + + +async def test_query_timeout_bounds_hung_fetch_below_lease_ttl(pg_engine, outbox_table) -> None: + """A long lease (slow handlers) must not delay detecting a silently dropped socket.""" + received: list[dict] = [] + broker = OutboxBroker(pg_engine, outbox_table=outbox_table, graceful_timeout=2.0) + original_fetch = OutboxClient.fetch + hung: list[bool] = [] + + async def hang_once( + self: OutboxClient, + conn: AsyncConnection | None, + queues: Sequence[str], + *, + limit: int, + lease_ttl_seconds: float, + ) -> list[OutboxInnerMessage]: + if not hung: + hung.append(True) + await asyncio.Event().wait() + return await original_fetch(self, conn, queues, limit=limit, lease_ttl_seconds=lease_ttl_seconds) + + @broker.subscriber( + "orders", + min_fetch_interval=0.05, + max_fetch_interval=0.2, + lease_ttl_seconds=600.0, + query_timeout_seconds=0.5, + ) + async def handle(body: dict) -> None: + received.append(body) + + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + with mock.patch.object(OutboxClient, "fetch", hang_once): + async with broker: + async with session_factory() as session, session.begin(): + await broker.publish({"order_id": 1}, queue="orders", session=session) + await _wait_until(lambda: received, timeout=10.0) + + assert received == [{"order_id": 1}] + + +async def test_hung_terminal_write_times_out_and_worker_reconnects(pg_engine, outbox_table) -> None: + """A terminal DELETE stuck on a silently dropped socket must not wedge the worker forever.""" + received: list[dict] = [] + broker = OutboxBroker(pg_engine, outbox_table=outbox_table, graceful_timeout=2.0) + original_delete = OutboxClient.delete_with_lease + hung: list[bool] = [] + deleted: list[bool] = [] + + async def hang_once( + self: OutboxClient, + conn: AsyncConnection | None, + row_id: int, + token: uuid.UUID, + *, + dlq_payload: Mapping[str, Any] | None = None, + ) -> bool: + if not hung: + hung.append(True) + await asyncio.Event().wait() + result = await original_delete(self, conn, row_id, token, dlq_payload=dlq_payload) + deleted.append(result) + return result + + @broker.subscriber("orders", min_fetch_interval=0.05, max_fetch_interval=0.2, lease_ttl_seconds=1.0) + async def handle(body: dict) -> None: + received.append(body) + + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + with mock.patch.object(OutboxClient, "delete_with_lease", hang_once): + async with broker: + async with session_factory() as session, session.begin(): + await broker.publish({"order_id": 1}, queue="orders", session=session) + await _wait_until(lambda: any(deleted), timeout=10.0) + + assert len(received) >= 2 + assert all(body == {"order_id": 1} for body in received) + assert await _row_count(pg_engine, outbox_table) == 0 + + async def test_publish_with_activate_in_delays_delivery(pg_engine, outbox_table) -> None: """Handler must not see the message until activate_in elapses.""" received: list[dict] = [] diff --git a/tests/test_unit.py b/tests/test_unit.py index 7f0fb25..361751b 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -3217,6 +3217,33 @@ def test_subscriber_rejects_non_positive_lease_ttl() -> None: _register_subscriber(broker, lease_ttl_seconds=0.0) +@pytest.mark.parametrize("value", [0.0, -1.0]) +def test_subscriber_rejects_non_positive_query_timeout(value: float) -> None: + with pytest.raises(ValueError, match="query_timeout_seconds must be > 0"): + _register_subscriber(_make_broker(), query_timeout_seconds=value) + + +def test_subscriber_rejects_query_timeout_above_lease_ttl() -> None: + """A fetch or terminal write that outlives the lease has already lost the row.""" + with pytest.raises(ValueError, match=r"query_timeout_seconds \(61.0\) must be <= lease_ttl_seconds \(60.0\)"): + _register_subscriber(_make_broker(), query_timeout_seconds=61.0, lease_ttl_seconds=60.0) + + +def test_subscriber_query_timeout_defaults_to_lease_ttl() -> None: + sub = _make_broker().subscriber("q", lease_ttl_seconds=42.0) + assert sub._config.query_timeout == 42.0 # noqa: SLF001 + + +def test_outbox_route_forwards_query_timeout() -> None: + async def handle(body: str) -> None: ... + + broker = _make_broker() + broker.include_router(OutboxRouter(handlers=(OutboxRoute(handle, "orders", query_timeout_seconds=5.0),))) + sub = broker.subscribers[0] + assert isinstance(sub, OutboxSubscriber) + assert sub._config.query_timeout == 5.0 # noqa: SLF001 + + def test_subscriber_rejects_ack_first() -> None: # ACK_FIRST has no legitimate outbox use — deletes before the handler runs, so a # handler crash silently drops the row. Better to refuse than warn-and-ship. @@ -3300,7 +3327,9 @@ def test_subscriber_rejects_max_deliveries_below_one(max_deliveries: int) -> Non _register_subscriber(_make_broker(), max_deliveries=max_deliveries) -@pytest.mark.parametrize("field", ["min_fetch_interval", "max_fetch_interval", "lease_ttl_seconds"]) +@pytest.mark.parametrize( + "field", ["min_fetch_interval", "max_fetch_interval", "lease_ttl_seconds", "query_timeout_seconds"] +) @pytest.mark.parametrize("value", [math.nan, math.inf]) def test_subscriber_rejects_non_finite_intervals(field: str, value: float) -> None: with pytest.raises(ValueError, match=f"{field} must be a finite number"): diff --git a/tests_dr/__init__.py b/tests_dr/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests_dr/primary-init.sh b/tests_dr/primary-init.sh new file mode 100644 index 0000000..c7de9fe --- /dev/null +++ b/tests_dr/primary-init.sh @@ -0,0 +1,2 @@ +#!/bin/sh +echo "host replication all all scram-sha-256" >> "$PGDATA/pg_hba.conf" diff --git a/tests_dr/replica-entrypoint.sh b/tests_dr/replica-entrypoint.sh new file mode 100644 index 0000000..65b10d0 --- /dev/null +++ b/tests_dr/replica-entrypoint.sh @@ -0,0 +1,7 @@ +#!/bin/bash +set -eu +if [ ! -s "$PGDATA/PG_VERSION" ]; then + pg_basebackup -h pg-primary -U outbox -D "$PGDATA" -R -X stream + chmod 0700 "$PGDATA" +fi +exec postgres diff --git a/tests_dr/test_failover.py b/tests_dr/test_failover.py new file mode 100644 index 0000000..ba7cd00 --- /dev/null +++ b/tests_dr/test_failover.py @@ -0,0 +1,152 @@ +"""Primary/replica failover drills. Run via ``just test-dr``, one fresh cluster per scenario.""" + +import asyncio +import json +import os +import time +from collections.abc import Awaitable, Callable + +import asyncpg +import pytest +from sqlalchemy import MetaData +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from faststream_outbox import OutboxBroker, make_outbox_table + + +CLUSTER_DSN = os.environ.get("POSTGRES_DSN", "") +PRIMARY_DSN = os.environ.get("PRIMARY_DSN", "") +REPLICA_DSN = os.environ.get("REPLICA_DSN", "") +DOCKER_SOCKET = "/var/run/docker.sock" +PRIMARY_CONTAINER = "outbox-dr-primary" +NETWORK = "outbox-dr_default" +QUEUE = "dr" +BEFORE = 50 +AFTER = 50 +QUERY_TIMEOUT_SECONDS = 10.0 +RECOVERY_BUDGET_SECONDS = 90.0 +PUBLISH_ATTEMPT_TIMEOUT_SECONDS = 5.0 +TEST_BUDGET_SECONDS = 240.0 + +pytestmark = pytest.mark.skipif(not CLUSTER_DSN, reason="needs the docker-compose.dr.yml cluster") + + +async def _docker(method: str, path: str, body: dict[str, object] | None = None) -> None: + payload = json.dumps(body).encode() if body is not None else b"" + reader, writer = await asyncio.open_unix_connection(DOCKER_SOCKET) + head = ( + f"{method} {path} HTTP/1.1\r\nHost: docker\r\nConnection: close\r\n" + f"Content-Type: application/json\r\nContent-Length: {len(payload)}\r\n\r\n" + ) + writer.write(head.encode() + payload) + 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 _blackhole_primary() -> None: + await _docker("POST", f"/networks/{NETWORK}/disconnect", {"Container": PRIMARY_CONTAINER, "Force": True}) + + +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_replica_caught_up() -> None: + primary = await asyncpg.connect(PRIMARY_DSN) + replica = await asyncpg.connect(REPLICA_DSN) + try: + target = await primary.fetchval("SELECT pg_current_wal_lsn()") + deadline = time.monotonic() + 30 + while await replica.fetchval("SELECT pg_last_wal_replay_lsn() < $1::pg_lsn", target): + assert time.monotonic() < deadline, "replica did not catch up" + await asyncio.sleep(0.1) + finally: + await primary.close() + await replica.close() + + +async def _promote_replica() -> None: + replica = await asyncpg.connect(REPLICA_DSN) + try: + assert await replica.fetchval("SELECT pg_promote()") + finally: + await replica.close() + + +@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(CLUSTER_DSN) + metadata = MetaData() + table = make_outbox_table(metadata, table_name="dr_outbox") + async with engine.begin() as conn: + await conn.run_sync(metadata.create_all) + broker = OutboxBroker(engine, outbox_table=table) + sessions = async_sessionmaker(engine, expire_on_commit=False) + handled_at: dict[int, float] = {} + deliveries: list[int] = [] + + @broker.subscriber(QUEUE, query_timeout_seconds=QUERY_TIMEOUT_SECONDS) + async def handle(body: dict[str, int]) -> None: + deliveries.append(body["n"]) + handled_at.setdefault(body["n"], time.monotonic()) + + async def publish(n: int) -> None: + async with sessions() as session, session.begin(): + await broker.publish({"n": n}, queue=QUEUE, session=session) + + async def publish_until_accepted(n: int, deadline: float) -> None: + while True: + try: + async with asyncio.timeout(PUBLISH_ATTEMPT_TIMEOUT_SECONDS): + await publish(n) + except Exception: + if time.monotonic() > deadline: + raise + await asyncio.sleep(0.5) + else: + return + + async with asyncio.timeout(TEST_BUDGET_SECONDS), broker: + for n in range(BEFORE): + await publish(n) + await _wait_until( + lambda: len(handled_at) == BEFORE, + deadline=time.monotonic() + 30, + what="pre-failover messages", + ) + await _wait_replica_caught_up() + + failed_at = time.monotonic() + await fail_primary() + await _promote_replica() + deadline = failed_at + RECOVERY_BUDGET_SECONDS + for n in range(BEFORE, BEFORE + AFTER): + await publish_until_accepted(n, deadline) + publish_recovered = time.monotonic() - failed_at + await _wait_until( + lambda: all(n in handled_at for n in range(BEFORE, BEFORE + AFTER)), + deadline=deadline, + what="post-failover messages", + ) + subscriber_recovered = max(handled_at[n] for n in range(BEFORE, BEFORE + AFTER)) - failed_at + + # Idle past max_fetch_interval so only a live LISTEN connection can deliver within 3s. + await asyncio.sleep(12) + sent_at = time.monotonic() + await publish(BEFORE + AFTER) + await _wait_until(lambda: BEFORE + AFTER in handled_at, deadline=sent_at + 3, what="NOTIFY wakeup") + + print( # noqa: T201 + f"\npublish recovered {publish_recovered:.1f}s, subscriber recovered {subscriber_recovered:.1f}s, " + f"redeliveries {len(deliveries) - len(handled_at)}", + ) From a75bb58956edc2163b55e71f32f551ee99335d3d Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 27 Oct 2026 22:21:32 +0300 Subject: [PATCH 2/2] Skip failover drills unless DR_CLUSTER_DSN is set The floors job runs pytest . and sets POSTGRES_DSN, so the drills were collected and tried to connect to a cluster that does not exist. --- docker-compose.dr.yml | 2 +- tests_dr/test_failover.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docker-compose.dr.yml b/docker-compose.dr.yml index 9596931..176c642 100644 --- a/docker-compose.dr.yml +++ b/docker-compose.dr.yml @@ -14,7 +14,7 @@ services: pg-replica: condition: service_healthy environment: - - POSTGRES_DSN=postgresql+asyncpg://outbox:outbox@/outbox?host=pg-primary:5432&host=pg-replica:5432&target_session_attrs=primary + - DR_CLUSTER_DSN=postgresql+asyncpg://outbox:outbox@/outbox?host=pg-primary:5432&host=pg-replica:5432&target_session_attrs=primary - PRIMARY_DSN=postgresql://outbox:outbox@pg-primary:5432/outbox - REPLICA_DSN=postgresql://outbox:outbox@pg-replica:5432/outbox diff --git a/tests_dr/test_failover.py b/tests_dr/test_failover.py index ba7cd00..fd6d70d 100644 --- a/tests_dr/test_failover.py +++ b/tests_dr/test_failover.py @@ -14,7 +14,7 @@ from faststream_outbox import OutboxBroker, make_outbox_table -CLUSTER_DSN = os.environ.get("POSTGRES_DSN", "") +CLUSTER_DSN = os.environ.get("DR_CLUSTER_DSN", "") PRIMARY_DSN = os.environ.get("PRIMARY_DSN", "") REPLICA_DSN = os.environ.get("REPLICA_DSN", "") DOCKER_SOCKET = "/var/run/docker.sock"