From 8f3b7b4a10105262a56297a8b3e7d269a1aefa45 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 15:44:17 +0300 Subject: [PATCH 1/2] fix: refuse to start when a fixed-size pool can't hold the subscribers' connections --- docs/operations/checklist.md | 4 +- docs/operations/troubleshooting.md | 38 ++++++++------- docs/usage/subscriber.md | 15 +++--- faststream_outbox/broker.py | 27 ++++++++++ faststream_outbox/subscriber/usecase.py | 5 ++ tests/test_unit.py | 65 +++++++++++++++++++++++++ 6 files changed, 130 insertions(+), 24 deletions(-) diff --git a/docs/operations/checklist.md b/docs/operations/checklist.md index c3bc215..9e1d4b7 100644 --- a/docs/operations/checklist.md +++ b/docs/operations/checklist.md @@ -9,7 +9,9 @@ story. - [ ] Engine pool ≥ `Σ subs × (max_workers + 1)`. Every subscriber holds `max_workers + 1` SQLAlchemy pool connections (one writer per worker + one fetch) plus one raw asyncpg connection for - `LISTEN`. Sub-budget formula in [Subscriber § Connection + `LISTEN`. With a fixed-size pool, `broker.start()` raises below this + number; budget extra connections for the application's own sessions on + top. Sub-budget formula in [Subscriber § Connection budget](../usage/subscriber.md#connection-budget). - [ ] Postgres `max_connections` ≥ `replicas × Σ subs × (max_workers + 2)`. That covers `max_workers + 1` pool connections plus the raw asyncpg `LISTEN` diff --git a/docs/operations/troubleshooting.md b/docs/operations/troubleshooting.md index bd3405a..158c6a7 100644 --- a/docs/operations/troubleshooting.md +++ b/docs/operations/troubleshooting.md @@ -10,7 +10,8 @@ owns the underlying design. | [Outbox row count grows + `lease_lost` spike](#outbox-row-count-grows-lease_lost-spike) | DLQ CTE failing (DLQ schema drift) | | [Outbox row count grows, no `lease_lost`](#outbox-row-count-grows-no-lease_lost) | Fetch loop not running, or rows future-dated | | [Idle dispatch latency > `max_fetch_interval`](#idle-dispatch-latency-max_fetch_interval) | LISTEN setup failed, so the subscriber falls back to polling | -| [Subscriber dispatch never starts; rows pile up](#subscriber-blocks-at-brokerstart) | Engine pool exhausted on writer-connection checkout | +| [`broker.start()` raises "Outbox subscribers need N pooled connections"](#subscriber-blocks-at-brokerstart) | Engine pool smaller than the subscribers hold | +| [Subscriber dispatch never starts; rows pile up](#subscriber-blocks-at-brokerstart) | Pool fits the subscribers, but the application's sessions take the rest | | [Duplicate handler invocations](#duplicate-handler-invocations) | Lease expired before handler returned, or handler not idempotent | | [Rolling deploy leaks rows](#rolling-deploy-leaks-rows) | `graceful_timeout` < handler P99, or k8s grace too short | | [`activate_in` / `activate_at` fires immediately in tests](#activate_in-activate_at-fires-immediately-in-tests) | `TestOutboxBroker(run_loops=False)` ignores scheduling | @@ -129,22 +130,25 @@ Fetch loop](../introduction/how-it-works.md#subscriber-two-async-loops). ## Subscriber dispatch never starts; rows pile up { #subscriber-blocks-at-brokerstart } -Rows are published but never dispatched (the table grows), and the -subscriber's loops emit repeating reconnect ERROR logs. `broker.start()` -(or the FastAPI `include_router` lifespan) returns normally because it -only schedules the loop tasks, so the failure shows up *after* startup -and does not look like a hang. - -The likely cause is an exhausted SQLAlchemy pool on the per-worker writer -connection checkout. The fetch and worker loops can't acquire their -connections, so each cycle errors and backs off. Each subscriber needs -`max_workers + 1` pool connections, and the default pool is `pool_size=5, -max_overflow=10`. A handful of single-worker subscribers fits; a fleet -of high-`max_workers` subscribers does not. - -To confirm, inspect the engine pool. Compute `Σ subs × (max_workers -+ 1)` from your subscriber registrations and compare it to -`pool_size + max_overflow`. +Each subscriber holds `max_workers + 1` pool connections for as long as it +runs, and the default pool is `pool_size=5, max_overflow=10`. A handful of +single-worker subscribers fits; a fleet of high-`max_workers` subscribers +does not. + +With a fixed-size pool, `broker.start()` (or the FastAPI `include_router` +lifespan) checks this first and raises `RuntimeError: Outbox subscribers +need N pooled connections ...`, naming the number required and what the +pool allows. + +If startup succeeds but rows are published and never dispatched, and the +subscriber's loops emit repeating reconnect ERROR logs, the pool fits the +subscribers but the application's own sessions are taking the remaining +connections. The fetch and worker loops can't check out a connection, so +each cycle errors and backs off. + +To confirm, compute `Σ subs × (max_workers + 1)` from your subscriber +registrations, add the connections your application's sessions use at +peak, and compare it to `pool_size + max_overflow`. To fix it, raise `pool_size` / `max_overflow` on the engine, or lower `max_workers` per subscriber. Also confirm that Postgres has diff --git a/docs/usage/subscriber.md b/docs/usage/subscriber.md index caf53ab..6f4bf74 100644 --- a/docs/usage/subscriber.md +++ b/docs/usage/subscriber.md @@ -340,12 +340,15 @@ The base strategy also enforces `max_attempts` and Each subscriber holds `max_workers + 1` long-lived SQLAlchemy pool connections (one writer per worker + one fetch), plus one raw asyncpg connection for `LISTEN` when available. Size your engine pool for -`Σ subscribers × (max_workers + 1)`. An undersized pool does not block -`broker.start()`, because `start()` only schedules the loop tasks and returns. -Instead the fetch/worker loops stall on pool checkout and surface as -repeating reconnect ERROR logs with dispatch silently starved. SQLAlchemy's -default `pool_size=5, max_overflow=10` covers a handful of single-worker -subscribers; raise it for larger fleets. +`Σ subscribers × (max_workers + 1)`. With a fixed-size pool (SQLAlchemy's +default `QueuePool`), `broker.start()` raises `RuntimeError` when +`pool_size + max_overflow` is below that sum. `NullPool` and an unlimited +`max_overflow=-1` aren't checked. The check counts only the subscribers, so +leave room for the application's own sessions: if those take the remaining +connections, the fetch and worker loops stall on pool checkout and show up as +repeating reconnect ERROR logs. SQLAlchemy's default `pool_size=5, +max_overflow=10` covers a handful of single-worker subscribers; raise it for +larger fleets. Server-side, the footprint is one larger: the raw asyncpg `LISTEN` connection lives outside the pool, so each subscriber consumes diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index 2c01e57..ef50347 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -27,6 +27,7 @@ from faststream.specification.schema import BrokerSpec from faststream.specification.schema.extra import Tag, TagDict from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.pool import QueuePool from typing_extensions import override from faststream_outbox.client import AbstractOutboxClient, OutboxClient @@ -228,11 +229,37 @@ async def __aenter__(self) -> typing.Self: @override async def start(self) -> None: + self._check_pool_capacity() await self.connect() await super().start() self._warn_on_unstarted_foreign_publishers() self._warn_on_duplicate_queues() + def _check_pool_capacity(self) -> None: + """Raise when a fixed-size pool can't hold every subscriber's connections at once. + + ``max_overflow`` has no public accessor; a missing ``_max_overflow`` skips the check + rather than failing startup. + """ + engine = self.config.broker_config.engine + if engine is None: + return + pool = engine.pool + max_overflow = getattr(pool, "_max_overflow", None) + if not isinstance(pool, QueuePool) or not isinstance(max_overflow, int) or max_overflow < 0: + return + needed = sum(getattr(sub, "_pooled_connections", 0) for sub in self.subscribers) + capacity = pool.size() + max_overflow + if needed > capacity: + msg = ( + f"Outbox subscribers need {needed} pooled connections for as long as they run (one fetch " + f"connection plus one per worker, for each subscriber), but the engine's pool allows " + f"{capacity} (pool_size={pool.size()} + max_overflow={max_overflow}). Raise pool_size or " + f"max_overflow on the engine, or lower max_workers, and leave room for the application's " + f"own sessions." + ) + raise RuntimeError(msg) + def _warn_on_duplicate_queues(self) -> None: # P22: the registration-time overlap warning only sees the broker's own # _subscribers, so duplicates introduced via include_router slip through. Re-check diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 5feedae..5e358d2 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -236,6 +236,11 @@ def _call_name(self) -> str: def _fetch_tags(self) -> dict[str, typing.Any]: return self._base_tags(self._queues[0] if self._queues else "") + @property + def _pooled_connections(self) -> int: + """Engine-pool connections held while running: the fetch connection plus one writer per worker.""" + return 1 + self._config.max_workers + def _base_tags(self, queue: str) -> dict[str, typing.Any]: return {"queue": queue, "subscriber": self._call_name} diff --git a/tests/test_unit.py b/tests/test_unit.py index aa9d423..8d58b47 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -26,6 +26,7 @@ from sqlalchemy import MetaData, Table from sqlalchemy.dialects import postgresql from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlalchemy.pool import NullPool, QueuePool from typing_extensions import get_protocol_members from faststream_outbox import ( @@ -4505,3 +4506,67 @@ async def test_every_buffered_row_is_closed_by_one_event_with_a_duration(delete_ await sub._flush_buffer(buffer, writer_conn=None) # noqa: SLF001 _assert_dispatch_closed_once(events) + + +def _broker_on_pool(**engine_kwargs: typing.Any) -> OutboxBroker: + return _make_broker(create_async_engine("postgresql+asyncpg://u:p@localhost/db", **engine_kwargs)) + + +def _subscribe(registrator: OutboxBroker | OutboxRouter, queue: str, max_workers: int) -> None: + @registrator.subscriber(queue, max_workers=max_workers) + async def handle(body: dict) -> None: ... + + +async def test_start_rejects_a_pool_smaller_than_the_subscribers_hold() -> None: + """INVARIANT: ``start()`` refuses a fixed-size pool with fewer connections than the subscribers hold. + + Each subscriber keeps ``1 + max_workers`` pooled connections for its whole life: the fetch connection + and one writer per worker. Below that total the loops wait on each other for connections, every + checkout times out after ``pool_timeout``, and the subscribers retry forever without consuming. + """ + broker = _broker_on_pool(pool_size=2, max_overflow=1) + _subscribe(broker, "orders", max_workers=3) + try: + with pytest.raises(RuntimeError, match=r"need 4 pooled connections .* allows 3"): + await broker.start() + finally: + await broker.stop() + await broker.config.broker_config.engine.dispose() + + +async def test_pool_check_counts_router_subscribers() -> None: + broker = _broker_on_pool(pool_size=2, max_overflow=0) + router = OutboxRouter() + _subscribe(router, "orders", max_workers=2) + broker.include_router(router) + with pytest.raises(RuntimeError, match="need 3 pooled connections"): + broker._check_pool_capacity() # noqa: SLF001 + + +@pytest.mark.parametrize( + "engine_kwargs", + [ + {"pool_size": 2, "max_overflow": 2}, + {"poolclass": NullPool}, + {"pool_size": 1, "max_overflow": -1}, + {"pool_size": 0}, + ], + ids=["exactly-enough", "null-pool", "unlimited-overflow", "unlimited-size"], +) +def test_pool_check_accepts_pools_that_fit_or_have_no_fixed_size(engine_kwargs: dict[str, typing.Any]) -> None: + broker = _broker_on_pool(**engine_kwargs) + _subscribe(broker, "orders", max_workers=3) + broker._check_pool_capacity() # noqa: SLF001 + + +def test_pool_check_skips_a_broker_without_an_engine() -> None: + broker = _make_broker() + _subscribe(broker, "orders", max_workers=3) + broker._check_pool_capacity() # noqa: SLF001 + + +def test_queue_pool_exposes_the_overflow_limit_the_pool_check_reads() -> None: + """``_max_overflow`` is private; this fails here, not at a user's startup, if SQLAlchemy renames it.""" + pool = create_async_engine("postgresql+asyncpg://u:p@localhost/db", pool_size=3, max_overflow=4).pool + assert isinstance(pool, QueuePool) + assert pool._max_overflow == 4 # noqa: SLF001 From 4ee62fd281e463cd0408dea43bfadc871d770e49 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 15:44:17 +0300 Subject: [PATCH 2/2] test: widen the lease margins of the buffered-delete age-bound test --- tests/test_fake.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_fake.py b/tests/test_fake.py index abf2168..800aeaf 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -2333,7 +2333,7 @@ async def test_loop_mode_buffered_deletes_flush_before_the_lease_expires() -> No @broker.subscriber( "orders", terminal_flush_batch_size=100, - lease_ttl_seconds=0.4, + lease_ttl_seconds=1.0, min_fetch_interval=0.01, max_fetch_interval=0.05, ) @@ -2343,12 +2343,12 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker, run_loops=True) async with test_broker: - for i in range(30): + for i in range(60): payload, hdrs = encode_payload(f"m{i}") test_broker.feed("orders", payload, headers=hdrs) - await _wait_until(lambda: not test_broker.fake_client.rows, timeout=5.0) + await _wait_until(lambda: not test_broker.fake_client.rows, timeout=10.0) - assert sorted(handled) == sorted(f"m{i}" for i in range(30)) + assert sorted(handled) == sorted(f"m{i}" for i in range(60)) async def test_handler_returning_outbox_response_takes_correlation_id_from_headers() -> None: