fix: refuse to start when a fixed-size pool can't hold the subscribers' connections

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`

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

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

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

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}
 

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:

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