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

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

@@ -36,6 +36,8 @@ Python is 3.14, dependencies managed by `uv`. The API is exposed on `:8000`.
 
 `UseCases` group (request-scoped): one provider per use case in `app/use_cases/`. modern-di fills each dataclass field by its type, so a use case needs no `kwargs`.
 
+**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 answers anything not caused by a pool timeout the way Litestar's default handler would: the debug page when `SERVICE_DEBUG` is on, a plain 500 otherwise. `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:
 
 ```python
@@ -65,15 +67,15 @@ class ListDecksUseCase:
 
 **Migrations**: `migrations/env.py` reads `app.models.METADATA` and rewrites the DSN driver from `postgresql+asyncpg` → `postgresql+psycopg` (Alembic runs on sync psycopg 3). Always run autogen against an upgraded DB — `just migration` enforces this.
 
-**Settings** (`app/settings.py`): `pydantic_settings.BaseSettings` reads from env vars (see `docker-compose.yml` for `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `DB_DSN`; `DB_REPLICA_DSN` is optional). `api_bootstrapper_config` builds the `LitestarConfig` consumed by `lite-bootstrap`.
+**Settings** (`app/settings.py`): `pydantic_settings.BaseSettings` reads from env vars (see `docker-compose.yml` for `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `DB_DSN`; `DB_REPLICA_DSN` is optional). Pool and server sizing uses `DB_POOL_SIZE`, `DB_MAX_OVERFLOW`, `DB_POOL_TIMEOUT`, `DB_POOL_RECYCLE` and `APP_WORKERS` (see **Pool and server sizing**). `api_bootstrapper_config` builds the `LitestarConfig` consumed by `lite-bootstrap`.
 
 ## Conventions
 
 - Routes live in `app/api/`, one module per resource (`decks.py`, `cards.py`), each exposing its own `ROUTER` (`litestar.Router`, prefix `/api`). `application.build_app` registers them all via `route_handlers=[decks.ROUTER, cards.ROUTER]`. Add a new resource by creating `app/api/<name>.py`, defining handlers + a `ROUTER`, and adding it to that list. Its database work goes in `app/use_cases/` with a provider in `ioc.UseCases`.
 - Pydantic schemas in `app/schemas.py` use `from_attributes=True` (via `Base`) so they validate directly from ORM instances (`schemas.X.model_validate(orm_instance)`). Collection responses go through `Collection[T].from_models(...)` (e.g. `schemas.Cards`).
 - List endpoints page by id (keyset). They take `limit` (1 to `schemas.MAX_PAGE_LIMIT`, default `schemas.DEFAULT_PAGE_LIMIT`) and `after`, the last id the client has seen, and return a `Page[T]` subclass (`schemas.DecksPage`, `schemas.CardsPage`) with `items` and `next_after`. The repository filters `id > after` and fetches `limit + 1` rows ordered by id. `Page.from_models(rows, limit=limit)` drops the extra row and sets `next_after` to the last returned id when that row existed, or `null` on the last page. Don't page with OFFSET: Postgres reads and discards every skipped row.
 - Deck responses are deliberately two-shaped: `list_decks`/`create_deck`/`update_deck` return the light `schemas.Deck` (no `cards`), while `get_deck` returns `schemas.DeckWithCards`. The split mirrors loading — lists use `noload` (no cards query), detail uses `joinedload` via `fetch_with_cards` — so the type states exactly what each endpoint loads.
-- Domain exceptions: register handlers in `application.build_app`'s `exception_handlers` dict (`DuplicateKeyError` → `exceptions.duplicate_key_error_handler` returns 400, `NotFoundError` → `exceptions.not_found_error_handler` returns 404). Handlers don't raise HTTP errors themselves: a missing row surfaces as advanced-alchemy's `NotFoundError`.
+- Domain exceptions: register handlers in `application.build_app`'s `exception_handlers` dict (`DuplicateKeyError` → `exceptions.duplicate_key_error_handler` returns 400, `NotFoundError` and `ForeignKeyError` → `exceptions.not_found_error_handler` returns 404, `sqlalchemy.exc.TimeoutError` and `RepositoryError` → `exceptions.database_unavailable_handler` returns 503 for a pool timeout). Handlers don't raise HTTP errors themselves: a missing row surfaces as advanced-alchemy's `NotFoundError`.
 - `ruff` is configured with `select = ["ALL"]` and a line length of 120 — expect strict lint. `app/resources/` is exempt from the `TC` rules because modern-di resolves DI creator annotations at runtime. Type-check with `ty`; suppressions already exist for `invalid-argument-type` around `LifespanManager` / `ASGITransport` in `tests/conftest.py`.
 
 ## Agent skills

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,10 +3,11 @@
 
 import modern_di
 import modern_di_litestar
-from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError
+from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError, RepositoryError
 from lite_bootstrap import LitestarBootstrapper
 from litestar.config.app import AppConfig
 from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
+from sqlalchemy import exc as sa_exc
 
 from app import exceptions, ioc
 from app.api import cards, decks
@@ -26,6 +27,8 @@ def build_app() -> litestar.Litestar:
                 DuplicateKeyError: exceptions.duplicate_key_error_handler,
                 NotFoundError: exceptions.not_found_error_handler,
                 ForeignKeyError: exceptions.not_found_error_handler,
+                sa_exc.TimeoutError: exceptions.database_unavailable_handler,
+                RepositoryError: exceptions.database_unavailable_handler,
             },
             route_handlers=[decks.ROUTER, cards.ROUTER],
             plugins=[modern_di_litestar.ModernDIPlugin(di_container, autowired_groups=[ioc.UseCases])],

app/exceptions.py

@@ -2,6 +2,8 @@
 
 import litestar
 from litestar import status_codes
+from litestar.exceptions.responses import create_debug_response, create_exception_response
+from sqlalchemy import exc as sa_exc
 
 
 if typing.TYPE_CHECKING:
@@ -22,3 +24,14 @@ def not_found_error_handler(_: object, __: NotFoundError | ForeignKeyError) -> l
         content={"detail": "Not found"},
         status_code=status_codes.HTTP_404_NOT_FOUND,
     )
+
+
+def database_unavailable_handler(request: litestar.Request, exc: Exception) -> litestar.Response[typing.Any]:
+    if not isinstance(exc, sa_exc.TimeoutError) and not isinstance(exc.__cause__, sa_exc.TimeoutError):
+        fallback: typing.Final = create_debug_response if request.app.debug else create_exception_response
+        return fallback(request=request, exc=exc)
+    return litestar.Response(
+        media_type=litestar.MediaType.JSON,
+        content={"detail": "Database unavailable"},
+        status_code=status_codes.HTTP_503_SERVICE_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,51 @@
+import typing
+import uuid
+
+import pytest
+from litestar import status_codes as 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()
+
+    response: typing.Final = await client.post(f"/api/decks/{deck.id}/cards/", json=[{"front": "nul \x00 byte"}])
+    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR

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)