feat: size Granian to the pool and return 503 on pool timeouts (#100)

Add APP_WORKERS, DB_POOL_TIMEOUT and DB_POOL_RECYCLE. Set Granian backpressure to pool size plus overflow so excess requests wait in the socket backlog instead of on the pool, and require a positive pool size and non-negative overflow so that sum stays meaningful. A pool checkout timeout, raw or wrapped by advanced-alchemy in RepositoryError, now returns 503 instead of 500.

Author: lesnik512

AGENTS.md

@@ -35,7 +35,7 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
 2. Builds a `FastAPIBootstrapper` from `settings.api_bootstrapper_config`, injecting the SQLAlchemy OpenTelemetry instrumentor.
 3. `modern_di_fastapi.setup_di(app, container)` wires DI scopes onto the FastAPI app.
 4. Includes one `ROUTER` per resource (`app.api.decks.ROUTER`, `app.api.cards.ROUTER`) under `/api` via `include_routers`. Add a new resource by creating `app/api/<name>.py` with its own `APIRouter`-based `ROUTER` and registering it there.
-5. Registers `DuplicateKeyError` → 422 handler from `app/exceptions.py`.
+5. Registers handlers from `app/exceptions.py`: `DuplicateKeyError` → 422, `NotFoundError` and `ForeignKeyError` → 404, and `sqlalchemy.exc.TimeoutError` and `RepositoryError` → `database_unavailable_handler`, which returns 503 for a pool timeout and re-raises anything else.
 
 ### DI scopes (modern-di)
 
@@ -55,6 +55,10 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
 
 Endpoints inject a use case with `FromDI(UseCase)` from `modern_di_fastapi`, call it, and convert the result to a schema. They never touch repositories or the session. Add new providers to the matching group rather than constructing services manually in routes.
 
+### Pool and server sizing
+
+`DB_POOL_SIZE` (positive) and `DB_MAX_OVERFLOW` (non-negative) size the pool, and `app/__main__.py` sets Granian's `backpressure` to their sum, so each worker accepts only as many requests at once as it has connections. Requests beyond that wait in Granian's socket backlog instead of queueing on the pool. `APP_WORKERS` sets the Granian worker count, and each worker has its own pool. `DB_POOL_TIMEOUT` (default 30 s) bounds the wait for a connection, and a timeout returns 503 through `exceptions.database_unavailable_handler`. advanced-alchemy wraps the timeout in a `RepositoryError`, so the handler is also registered for `RepositoryError` and re-raises anything not caused by a pool timeout. `DB_POOL_RECYCLE` defaults to SQLAlchemy's `-1` (never). Set it below the idle timeout of any proxy or load balancer in front of Postgres, so a connection it dropped is not handed out and retried after a 1-2 s backoff.
+
 ### Use cases and lost connections
 
 `pool_pre_ping` is off by default (`DB_POOL_PRE_PING`). A pooled connection that died while idle is handled by `db_retry.postgres_retry` instead, which saves a round trip on every checkout. That only works if the retried function owns its whole transaction, so every use case has this shape:
@@ -91,7 +95,7 @@ class ListDecksUseCase:
 
 ### Settings
 
-`app/settings.py` — `pydantic_settings.BaseSettings`. Env vars are unprefixed (`DB_DSN`, `DB_REPLICA_DSN`, `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `LOG_LEVEL`, `APP_HOST`, `APP_PORT`, `OPENTELEMETRY_ENDPOINT`, `SENTRY_DSN`, `CORS_ALLOWED_ORIGINS`, ...). `api_bootstrapper_config` produces a `FastAPIConfig` for `lite-bootstrap`. `LOG_LEVEL` is case-insensitive and sets both Granian's log level and the app's root logger.
+`app/settings.py` — `pydantic_settings.BaseSettings`. Env vars are unprefixed (`DB_DSN`, `DB_REPLICA_DSN`, `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `LOG_LEVEL`, `APP_HOST`, `APP_PORT`, `OPENTELEMETRY_ENDPOINT`, `SENTRY_DSN`, `CORS_ALLOWED_ORIGINS`, `APP_WORKERS`, `DB_POOL_TIMEOUT`, `DB_POOL_RECYCLE`, ...). `api_bootstrapper_config` produces a `FastAPIConfig` for `lite-bootstrap`. `LOG_LEVEL` is case-insensitive and sets both Granian's log level and the app's root logger.
 
 ### Tests
 

app/__main__.py

@@ -10,6 +10,8 @@
         factory=True,
         address=settings.app_host,
         port=settings.app_port,
+        workers=settings.app_workers,
+        backpressure=settings.db_pool_size + settings.db_max_overflow,
         interface=Interfaces.ASGI,
         log_level=settings.log_level,
         loop=Loops.uvloop,

app/application.py

@@ -3,9 +3,10 @@
 
 import modern_di
 import modern_di_fastapi
-from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError
+from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError, RepositoryError
 from lite_bootstrap import FastAPIBootstrapper
 from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
+from sqlalchemy import exc as sa_exc
 
 from app import exceptions, ioc
 from app.api import cards, decks
@@ -43,4 +44,6 @@ def build_app() -> fastapi.FastAPI:
         ForeignKeyError,
         exceptions.not_found_error_handler,  # ty: ignore[invalid-argument-type]
     )
+    app.add_exception_handler(sa_exc.TimeoutError, exceptions.database_unavailable_handler)
+    app.add_exception_handler(RepositoryError, exceptions.database_unavailable_handler)
     return app

app/exceptions.py

@@ -1,6 +1,7 @@
 from typing import TYPE_CHECKING
 
 from fastapi.responses import JSONResponse
+from sqlalchemy import exc as sa_exc
 from starlette import status
 
 
@@ -21,3 +22,12 @@ async def not_found_error_handler(_: Request, __: NotFoundError | ForeignKeyErro
         status_code=status.HTTP_404_NOT_FOUND,
         content={"detail": "Not found"},
     )
+
+
+async def database_unavailable_handler(_: Request, exc: Exception) -> JSONResponse:
+    if not isinstance(exc, sa_exc.TimeoutError) and not isinstance(exc.__cause__, sa_exc.TimeoutError):
+        raise exc
+    return JSONResponse(
+        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        content={"detail": "Database unavailable"},
+    )

app/resources/db.py

@@ -24,6 +24,8 @@ def create_sa_engine(url: URL, *, read_only: bool = False) -> sa.AsyncEngine:
         pool_size=settings.db_pool_size,
         pool_pre_ping=settings.db_pool_pre_ping,
         max_overflow=settings.db_max_overflow,
+        pool_timeout=settings.db_pool_timeout,
+        pool_recycle=settings.db_pool_recycle,
         isolation_level="AUTOCOMMIT" if read_only else None,
         connect_args={"server_settings": {"default_transaction_read_only": "on"}} if read_only else {},
     )

app/settings.py

@@ -16,12 +16,15 @@ class Settings(pydantic_settings.BaseSettings):
 
     db_dsn: str = "postgresql+asyncpg://postgres:password@db/postgres"
     db_replica_dsn: str = ""
-    db_pool_size: int = 5
-    db_max_overflow: int = 0
+    db_pool_size: pydantic.PositiveInt = 5
+    db_max_overflow: pydantic.NonNegativeInt = 0
     db_pool_pre_ping: bool = False
+    db_pool_timeout: float = 30
+    db_pool_recycle: int = -1
 
     app_host: str = "0.0.0.0"  # noqa: S104
     app_port: int = 8000
+    app_workers: int = 1
 
     opentelemetry_endpoint: str = ""
     sentry_dsn: str = ""

tests/test_main.py

@@ -6,16 +6,25 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app import ioc
+from app.settings import settings
 
 
 if TYPE_CHECKING:
     import pytest
 
 
 def test_main(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr("granian.Granian", mock.Mock())
+    granian = mock.Mock()
+    monkeypatch.setattr("granian.Granian", granian)
+    monkeypatch.setattr(settings, "app_workers", 4)
+    monkeypatch.setattr(settings, "db_pool_size", 5)
+    monkeypatch.setattr(settings, "db_max_overflow", 2)
+
     runpy.run_module("app.__main__", run_name="__main__")
 
+    assert granian.call_args.kwargs["workers"] == 4  # noqa: PLR2004
+    assert granian.call_args.kwargs["backpressure"] == 7  # noqa: PLR2004
+
 
 async def test_session() -> None:
     container = modern_di.Container(groups=ioc.ALL_GROUPS)

tests/test_pool.py

@@ -0,0 +1,52 @@
+import typing
+import uuid
+
+import pytest
+from advanced_alchemy.exceptions import RepositoryError
+from fastapi import status
+
+from app import ioc
+from app.resources.db import create_primary_sa_engine
+from app.settings import settings
+from tests import factories
+
+
+if typing.TYPE_CHECKING:
+    import modern_di
+    from httpx import AsyncClient
+
+
+@pytest.mark.parametrize(
+    "request_kwargs",
+    [
+        {"method": "get", "url": "/api/decks/1/"},
+        {"method": "post", "url": "/api/decks/", "json": {"name": "pool", "idempotency_key": str(uuid.uuid4())}},
+    ],
+)
+async def test_exhausted_pool_returns_503(
+    client: AsyncClient,
+    di_container: modern_di.Container,
+    monkeypatch: pytest.MonkeyPatch,
+    request_kwargs: dict[str, typing.Any],
+) -> None:
+    monkeypatch.setattr(settings, "db_pool_size", 1)
+    monkeypatch.setattr(settings, "db_max_overflow", 0)
+    monkeypatch.setattr(settings, "db_pool_timeout", 0.1)
+    engine: typing.Final = create_primary_sa_engine()
+    di_container.override(ioc.Database.dynamic_engine, engine)
+    try:
+        async with engine.connect():
+            response = await client.request(**request_kwargs)
+    finally:
+        di_container.reset_override(ioc.Database.dynamic_engine)
+        await engine.dispose()
+
+    assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
+
+
+@pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory")
+async def test_other_repository_errors_are_not_turned_into_503(client: AsyncClient) -> None:
+    deck: typing.Final = await factories.DeckModelFactory.create_async()
+
+    with pytest.raises(RepositoryError):
+        await client.post(f"/api/decks/{deck.id}/cards/", json=[{"front": "nul \x00 byte"}])

tests/test_settings.py

@@ -1,5 +1,7 @@
 import logging
 
+import pydantic
+import pytest
 from granian.log import LogLevels
 
 from app.settings import Settings
@@ -23,3 +25,9 @@ def test_log_level_is_case_insensitive_and_reaches_app_logging() -> None:
 
     assert custom.log_level == LogLevels.warning
     assert custom.api_bootstrapper_config.logging_log_level == logging.WARNING
+
+
+@pytest.mark.parametrize("pool_settings", [{"db_pool_size": 0}, {"db_max_overflow": -1}])
+def test_pool_settings_reject_unbounded_pools(pool_settings: dict[str, int]) -> None:
+    with pytest.raises(pydantic.ValidationError):
+        Settings.model_validate(pool_settings)