fix(db): configurable pool, auth_enabled cache, single-checkout auth (#2572)

Two or three concurrent UI logins exhausted the PostgreSQL pool on the
reporter's 93-printer farm: QueuePool limit of size 10 overflow 20 reached,
with all 30 sessions idle in transaction on the auth_enabled SELECT. Three
regressions had landed on dev after an earlier configurable-pool change was
reverted and never re-applied (only the route-by-route session fixes were).

- Pool sizing is env-configurable again (DB_POOL_SIZE / DB_MAX_OVERFLOW /
  DB_POOL_TIMEOUT / DB_POOL_RECYCLE); the PostgreSQL default returns to
  20 + 80 with pool_pre_ping and pool_recycle=1800, and GET
  /api/v1/system/db-pool reports resolved config + live gauges without
  checking out a connection. SQLite unchanged (20 + 200).
- is_auth_enabled caches for 30s again. Only enabled=True is ever cached, so
  a stale read can only fail closed (require auth), never open; set_auth_enabled
  invalidates immediately. An autouse test fixture resets the module cache
  between tests to keep ordering deterministic.
- Every authenticated request checked out two pooled connections: the
  permission dependency held one and the revoked-jti check opened another.
  is_jti_revoked now reuses the caller's session; the token dependencies and
  the auth-middleware gateway were restructured to open one session and pass
  it in, so each request makes a single checkout.

backend/app/api/routes/auth.py

@@ -189,9 +189,15 @@ async def set_advanced_auth_enabled(db: AsyncSession, enabled: bool) -> None:
 
 async def set_auth_enabled(db: AsyncSession, enabled: bool) -> None:
     """Set authentication enabled status."""
+    from backend.app.core.auth import invalidate_auth_enabled_cache
     from backend.app.core.db_dialect import upsert_setting
 
     await upsert_setting(db, Settings, "auth_enabled", "true" if enabled else "false")
+    # Drop the cached auth-enabled flag so the change takes effect immediately
+    # instead of after the TTL (issue #2572). Safe pre-commit: only enabled=True
+    # is ever cached, and the newly-enabled True isn't visible to other sessions
+    # until this transaction commits, so no stale value can be re-cached here.
+    invalidate_auth_enabled_cache()
     # Note: Don't commit here - let get_db handle it or commit explicitly in the route
 
 
@@ -659,7 +665,7 @@ async def get_current_user_info(
                     headers={"WWW-Authenticate": "Bearer"},
                 )
             jti: str | None = payload.get("jti")
-            if not jti or await is_jti_revoked(jti):  # B1: logout bypass fix
+            if not jti or await is_jti_revoked(jti, db):  # B1: logout bypass fix
                 raise HTTPException(
                     status_code=status.HTTP_401_UNAUTHORIZED,
                     detail="Could not validate credentials",

backend/app/api/routes/system.py

@@ -606,6 +606,24 @@ async def get_system_health(
     return await asyncio.to_thread(scan_logs, sensitive_strings=sensitive_strings)
 
 
+@router.get("/db-pool")
+async def get_db_pool(
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.SYSTEM_READ),
+):
+    """Live database connection-pool gauges for large-farm diagnostics (#2572).
+
+    Reports the resolved pool configuration plus current checked-out /
+    checked-in / overflow counts. Deliberately takes no DB session — reading
+    the pool's own counters must not itself consume a connection, so this stays
+    truthful even when the pool is saturated. On a healthy install ``checked_out``
+    sits well below ``config.pool_size + config.max_overflow``; sustained
+    saturation points at connections held across slow I/O (see #2572).
+    """
+    from backend.app.core.database import get_pool_status
+
+    return get_pool_status()
+
+
 @router.get("/appliance")
 async def get_appliance_defaults():
     """Expose appliance-set state for the SPA's bootstrap surface.

backend/app/core/auth.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import logging
 import os
 import secrets
+import time
 from datetime import datetime, timedelta, timezone
 from typing import Annotated
 
@@ -395,7 +396,7 @@ def require_energy_cost_update():
                 if username is None:
                     raise credentials_exception
                 jti: str | None = payload.get("jti")
-                if not jti or await is_jti_revoked(jti):
+                if not jti or await is_jti_revoked(jti, db):
                     raise credentials_exception
                 iat: int | float | None = payload.get("iat")
             except JWTError:
@@ -798,10 +799,18 @@ async def revoke_jti(jti: str, expires_at: datetime, username: str | None = None
             await db.rollback()  # jti already revoked — desired state, ignore
 
 
-async def is_jti_revoked(jti: str) -> bool:
-    """Return True if the given jti has been revoked."""
-    async with async_session() as db:
-        result = await db.execute(
+async def is_jti_revoked(jti: str, db: AsyncSession | None = None) -> bool:
+    """Return True if the given jti has been revoked.
+
+    Pass ``db`` to reuse the caller's session instead of opening a new one
+    (issue #2572): the permission dependencies already hold a session, and a
+    second checkout per request doubled pool pressure — a login burst then
+    exhausted the pool. With ``db`` omitted a short session is opened as before,
+    for callers that check the jti before they have a session open.
+    """
+
+    async def _query(session: AsyncSession) -> bool:
+        result = await session.execute(
             select(AuthEphemeralToken).where(
                 AuthEphemeralToken.token == jti,
                 AuthEphemeralToken.token_type == "revoked_jti",
@@ -809,6 +818,11 @@ async def is_jti_revoked(jti: str) -> bool:
         )
         return result.scalar_one_or_none() is not None
 
+    if db is not None:
+        return await _query(db)
+    async with async_session() as own_db:
+        return await _query(own_db)
+
 
 async def get_user_by_username(db: AsyncSession, username: str) -> User | None:
     """Get a user by username (case-insensitive) with groups loaded for permission checks."""
@@ -862,6 +876,33 @@ async def authenticate_user_by_email(db: AsyncSession, email: str, password: str
     return user
 
 
+# Short-lived cache for the auth-enabled flag (issue #2572). The middleware
+# and every ownership/permission dependency probe this once (or more) per
+# request; on a large farm that DB round-trip is pure overhead because the
+# value changes only when an admin toggles auth.
+#
+# SECURITY: only a ``True`` (auth-enabled) result is EVER cached. A disabled /
+# unconfigured result is never cached, so a stale cache can only ever cause a
+# request to REQUIRE auth that a moment ago wasn't required — it can never skip
+# an auth check that is now required. Staleness fails CLOSED, never open (cf.
+# GHSA-6mf4-q26m-47pv). ``set_auth_enabled`` invalidates explicitly on any
+# toggle; the TTL is only a backstop for out-of-band changes (a direct DB edit,
+# or another worker process in a multi-worker deployment).
+_AUTH_ENABLED_CACHE_TTL_SECONDS = 30.0
+_auth_enabled_cached_value: bool = False
+_auth_enabled_cached_until: float = 0.0
+
+
+def invalidate_auth_enabled_cache() -> None:
+    """Drop the cached auth-enabled flag so the next probe re-reads the DB.
+
+    Call after any write that toggles the ``auth_enabled`` setting.
+    """
+    global _auth_enabled_cached_value, _auth_enabled_cached_until
+    _auth_enabled_cached_value = False
+    _auth_enabled_cached_until = 0.0
+
+
 async def is_auth_enabled(db: AsyncSession) -> bool:
     """Check if authentication is enabled.
 
@@ -878,12 +919,25 @@ async def is_auth_enabled(db: AsyncSession) -> bool:
     no exception. Any OTHER failure (connection error, fd exhaustion,
     schema mismatch, …) propagates so the caller can deny the request
     (503 / 500). Fail-closed is the only safe default for an auth probe.
+
+    Result is cached briefly to cut per-request DB load on large farms; only
+    the enabled=True result is cached, so a stale read can only fail closed.
+    See the module-level cache comment above.
     """
+    global _auth_enabled_cached_value, _auth_enabled_cached_until
+    if _auth_enabled_cached_value and time.monotonic() < _auth_enabled_cached_until:
+        return True
+
     result = await db.execute(select(Settings).where(Settings.key == "auth_enabled"))
     setting = result.scalar_one_or_none()
-    if setting is None:
-        return False
-    return setting.value.lower() == "true"
+    enabled = setting is not None and setting.value.lower() == "true"
+    if enabled:
+        _auth_enabled_cached_value = True
+        _auth_enabled_cached_until = time.monotonic() + _AUTH_ENABLED_CACHE_TTL_SECONDS
+    else:
+        # Never cache "disabled" — keep failing closed on any future staleness.
+        _auth_enabled_cached_value = False
+    return enabled
 
 
 async def _user_from_api_key(db: AsyncSession, api_key: APIKey) -> User | None:
@@ -973,13 +1027,16 @@ async def get_current_user_optional(
         if username is None:
             raise _unauthorized
         jti: str | None = payload.get("jti")
-        if not jti or await is_jti_revoked(jti):
-            raise _unauthorized  # I6: revoked token → 401, not anonymous
         iat: int | float | None = payload.get("iat")
     except JWTError:
         raise _unauthorized
 
+    if not jti:
+        raise _unauthorized  # I6: revoked token → 401, not anonymous
+
     async with async_session() as db:
+        if await is_jti_revoked(jti, db):
+            raise _unauthorized  # I6: revoked token → 401, not anonymous
         user = await get_user_by_username(db, username)
         if user is None or not user.is_active:
             raise _unauthorized
@@ -1006,13 +1063,16 @@ async def get_current_user(
         if username is None:
             raise credentials_exception
         jti: str | None = payload.get("jti")
-        if not jti or await is_jti_revoked(jti):
-            raise credentials_exception
         iat: int | float | None = payload.get("iat")
     except JWTError:
         raise credentials_exception
 
+    if not jti:
+        raise credentials_exception
+
     async with async_session() as db:
+        if await is_jti_revoked(jti, db):
+            raise credentials_exception
         user = await get_user_by_username(db, username)
         if user is None:
             raise credentials_exception
@@ -1082,7 +1142,7 @@ async def require_auth_if_enabled(
                         headers={"WWW-Authenticate": "Bearer"},
                     )
                 jti: str | None = payload.get("jti")
-                if not jti or await is_jti_revoked(jti):
+                if not jti or await is_jti_revoked(jti, db):
                     raise HTTPException(
                         status_code=status.HTTP_401_UNAUTHORIZED,
                         detail="Could not validate credentials",
@@ -1191,7 +1251,7 @@ def require_admin_if_auth_enabled():
                         headers={"WWW-Authenticate": "Bearer"},
                     )
                 jti: str | None = payload.get("jti")
-                if not jti or await is_jti_revoked(jti):
+                if not jti or await is_jti_revoked(jti, db):
                     raise HTTPException(
                         status_code=status.HTTP_401_UNAUTHORIZED,
                         detail="Could not validate credentials",
@@ -1431,7 +1491,7 @@ def require_permission(*permissions: str | Permission):
                 if username is None:
                     raise credentials_exception
                 jti: str | None = payload.get("jti")
-                if not jti or await is_jti_revoked(jti):
+                if not jti or await is_jti_revoked(jti, db):
                     raise credentials_exception
                 iat: int | float | None = payload.get("iat")
             except JWTError:
@@ -1518,7 +1578,7 @@ def require_permission_if_auth_enabled(*permissions: str | Permission):
                             headers={"WWW-Authenticate": "Bearer"},
                         )
                     jti: str | None = payload.get("jti")
-                    if not jti or await is_jti_revoked(jti):
+                    if not jti or await is_jti_revoked(jti, db):
                         raise HTTPException(
                             status_code=status.HTTP_401_UNAUTHORIZED,
                             detail="Could not validate credentials",
@@ -1618,7 +1678,7 @@ def require_any_permission_if_auth_enabled(*permissions: str | Permission):
                             headers={"WWW-Authenticate": "Bearer"},
                         )
                     jti: str | None = payload.get("jti")
-                    if not jti or await is_jti_revoked(jti):
+                    if not jti or await is_jti_revoked(jti, db):
                         raise HTTPException(
                             status_code=status.HTTP_401_UNAUTHORIZED,
                             detail="Could not validate credentials",
@@ -1795,7 +1855,7 @@ def require_ownership_permission(
                             headers={"WWW-Authenticate": "Bearer"},
                         )
                     jti: str | None = payload.get("jti")
-                    if not jti or await is_jti_revoked(jti):
+                    if not jti or await is_jti_revoked(jti, db):
                         raise HTTPException(
                             status_code=status.HTTP_401_UNAUTHORIZED,
                             detail="Could not validate credentials",

backend/app/core/config.py

@@ -74,6 +74,17 @@ class Settings(BaseSettings):
     log_dir: Path = _log_dir
     database_url: str = _external_db_url or f"sqlite+aiosqlite:///{_db_path}"
 
+    # Database connection pool sizing. ``None`` = use the built-in, dialect-aware
+    # default (PostgreSQL: pool_size 20 + max_overflow 80; SQLite: 20 + 200).
+    # Large PostgreSQL printer farms can raise these via the DB_POOL_SIZE /
+    # DB_MAX_OVERFLOW / DB_POOL_TIMEOUT / DB_POOL_RECYCLE env vars (issue #2572).
+    # Make sure PostgreSQL ``max_connections`` comfortably exceeds
+    # (pool_size + max_overflow) x number of app worker processes.
+    db_pool_size: int | None = Field(default=None, gt=0)
+    db_max_overflow: int | None = Field(default=None, ge=0)
+    db_pool_timeout: int | None = Field(default=None, gt=0)
+    db_pool_recycle: int | None = Field(default=None, gt=0)
+
     # Logging
     log_level: str = "INFO"  # Override with LOG_LEVEL env var or DEBUG=true
     log_to_file: bool = True  # Set to false to disable file logging

backend/app/core/database.py

@@ -23,12 +23,54 @@ def _set_sqlite_pragmas(dbapi_conn, connection_record):
     cursor.close()
 
 
-def _create_engine():
-    """Create the async engine with dialect-appropriate settings."""
+# Resolved connection-pool configuration, captured at engine creation so
+# /system/db-pool can report it without re-deriving the dialect defaults.
+_pool_config: dict = {}
+
+
+def _resolve_pool_kwargs() -> dict:
+    """Build the pool kwargs for ``create_async_engine`` (issue #2572).
+
+    Dialect-aware defaults, each overridable via env (``DB_POOL_SIZE`` etc.):
+      - PostgreSQL: pool_size 20 + max_overflow 80, ``pool_pre_ping`` (recover
+        server-dropped connections instead of erroring the request) and
+        ``pool_recycle`` 1800s. The old hard-coded 10 + 20 exhausted on large
+        farms while printer callbacks held connections.
+      - SQLite: pool_size 20 + max_overflow 200 (unchanged); no pre-ping /
+        recycle — the connection is a local file, not a server socket.
+    """
     if is_sqlite():
-        kwargs = {"pool_size": 20, "max_overflow": 200}
+        pool_size = settings.db_pool_size if settings.db_pool_size is not None else 20
+        max_overflow = settings.db_max_overflow if settings.db_max_overflow is not None else 200
+        kwargs = {"pool_size": pool_size, "max_overflow": max_overflow}
     else:
-        kwargs = {"pool_size": 10, "max_overflow": 20}
+        pool_size = settings.db_pool_size if settings.db_pool_size is not None else 20
+        max_overflow = settings.db_max_overflow if settings.db_max_overflow is not None else 80
+        kwargs = {
+            "pool_size": pool_size,
+            "max_overflow": max_overflow,
+            "pool_pre_ping": True,
+            "pool_recycle": settings.db_pool_recycle if settings.db_pool_recycle is not None else 1800,
+        }
+    if settings.db_pool_timeout is not None:
+        kwargs["pool_timeout"] = settings.db_pool_timeout
+    return kwargs
+
+
+def _create_engine():
+    """Create the async engine with dialect-appropriate settings."""
+    kwargs = _resolve_pool_kwargs()
+
+    global _pool_config
+    _pool_config = {
+        "pool_size": kwargs["pool_size"],
+        "max_overflow": kwargs["max_overflow"],
+        # SQLAlchemy's own defaults when we don't pass the kwarg.
+        "pool_timeout": kwargs.get("pool_timeout", 30),
+        "pool_recycle": kwargs.get("pool_recycle", -1),
+        "pool_pre_ping": kwargs.get("pool_pre_ping", False),
+    }
+
     eng = create_async_engine(
         settings.database_url,
         echo=settings.debug,
@@ -79,6 +121,36 @@ async_session = async_sessionmaker(
 )
 
 
+def get_pool_status() -> dict:
+    """Snapshot the DB connection pool for diagnostics (issue #2572).
+
+    Returns the resolved configuration plus live gauges (checked-out /
+    checked-in / overflow). Reads the pool's own counters — it does NOT
+    check out a connection, so it stays truthful even when the pool is
+    exhausted. Gauges a given pool implementation doesn't expose come back
+    as ``None`` rather than raising.
+    """
+    pool = engine.sync_engine.pool
+    gauges: dict = {}
+    for key, method_name in (
+        ("current_size", "size"),
+        ("checked_out", "checkedout"),
+        ("checked_in", "checkedin"),
+        ("overflow", "overflow"),
+    ):
+        method = getattr(pool, method_name, None)
+        try:
+            gauges[key] = method() if callable(method) else None
+        except Exception:
+            # A gauge should never take down the diagnostics endpoint.
+            gauges[key] = None
+    return {
+        "dialect": "sqlite" if is_sqlite() else "postgresql",
+        "config": dict(_pool_config),
+        **gauges,
+    }
+
+
 async def run_with_retry(fn, *, max_attempts: int = 3, label: str = ""):
     """Run an async DB operation with retry for SQLite 'database is locked' errors.
 

backend/app/main.py

@@ -6824,16 +6824,16 @@ async def auth_middleware(request, call_next):
             raise ValueError("No jti in token")
         iat = payload.get("iat")
 
-        # Reject revoked tokens (defense-in-depth gateway check)
-        if await is_jti_revoked(jti):
-            return JSONResponse(
-                status_code=401,
-                content={"detail": "Token has been revoked"},
-                headers={"WWW-Authenticate": "Bearer"},
-            )
-
-        # Verify user exists, is active, and token is still fresh (L-R8-A)
+        # Verify user exists, is active, and token is still fresh (L-R8-A).
+        # Reject revoked tokens first (defense-in-depth gateway check), reusing
+        # this session so the gateway adds a single pooled checkout, not two (#2572).
         async with async_session() as db:
+            if await is_jti_revoked(jti, db):
+                return JSONResponse(
+                    status_code=401,
+                    content={"detail": "Token has been revoked"},
+                    headers={"WWW-Authenticate": "Bearer"},
+                )
             user = await get_user_by_username(db, username)
             if not user or not user.is_active:
                 return JSONResponse(

backend/tests/conftest.py

@@ -95,6 +95,22 @@ def reset_spoolman_location_sync_cache():
     _spoolman_location_sync_cache_clear()
 
 
+@pytest.fixture(autouse=True)
+def reset_auth_enabled_cache():
+    """Drop the module-level auth-enabled cache between tests (issue #2572).
+
+    ``is_auth_enabled`` caches an enabled=True result for a TTL. Without this
+    reset a test that enables auth would leave ``True`` cached, so a later test
+    running in auth-disabled mode (without going through ``set_auth_enabled``)
+    would wrongly see auth as enabled until the TTL expired — order-dependent
+    flakiness."""
+    from backend.app.core.auth import invalidate_auth_enabled_cache
+
+    invalidate_auth_enabled_cache()
+    yield
+    invalidate_auth_enabled_cache()
+
+
 @pytest.fixture(scope="session")
 def event_loop():
     """Create an instance of the default event loop for each test session."""

backend/tests/unit/test_db_pool_and_auth_cache.py

@@ -0,0 +1,191 @@
+"""Tests for the DB connection-pool sizing/diagnostics and the auth-enabled
+cache added for large printer farms (issue #2572)."""
+
+import asyncio
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+
+class TestPoolConfiguration:
+    """P0: env-configurable, dialect-aware pool sizing."""
+
+    def test_sqlite_defaults_when_unset(self, monkeypatch):
+        """SQLite keeps 20 + 200 when no env override is set."""
+        from backend.app.core import database
+
+        for attr in ("db_pool_size", "db_max_overflow", "db_pool_timeout", "db_pool_recycle"):
+            monkeypatch.setattr(database.settings, attr, None, raising=False)
+        monkeypatch.setattr(database, "is_sqlite", lambda: True)
+
+        kwargs = database._resolve_pool_kwargs()
+        assert kwargs["pool_size"] == 20
+        assert kwargs["max_overflow"] == 200
+        # No server-socket recycle/pre-ping for a local file.
+        assert "pool_pre_ping" not in kwargs
+        assert "pool_recycle" not in kwargs
+
+    def test_postgres_defaults_raise_the_old_limits(self, monkeypatch):
+        """Postgres default is now 20 + 80 (was 10 + 20) with pre-ping + recycle."""
+        from backend.app.core import database
+
+        for attr in ("db_pool_size", "db_max_overflow", "db_pool_timeout", "db_pool_recycle"):
+            monkeypatch.setattr(database.settings, attr, None, raising=False)
+        monkeypatch.setattr(database, "is_sqlite", lambda: False)
+
+        kwargs = database._resolve_pool_kwargs()
+        assert kwargs["pool_size"] == 20
+        assert kwargs["max_overflow"] == 80
+        assert kwargs["pool_pre_ping"] is True
+        assert kwargs["pool_recycle"] == 1800
+
+    def test_env_overrides_win_on_postgres(self, monkeypatch):
+        """DB_POOL_* overrides replace the dialect defaults."""
+        from backend.app.core import database
+
+        monkeypatch.setattr(database.settings, "db_pool_size", 100, raising=False)
+        monkeypatch.setattr(database.settings, "db_max_overflow", 200, raising=False)
+        monkeypatch.setattr(database.settings, "db_pool_timeout", 45, raising=False)
+        monkeypatch.setattr(database.settings, "db_pool_recycle", 600, raising=False)
+        monkeypatch.setattr(database, "is_sqlite", lambda: False)
+
+        kwargs = database._resolve_pool_kwargs()
+        assert kwargs["pool_size"] == 100
+        assert kwargs["max_overflow"] == 200
+        assert kwargs["pool_timeout"] == 45
+        assert kwargs["pool_recycle"] == 600
+
+    @pytest.mark.asyncio
+    async def test_concurrent_checkouts_exceed_base_pool_size(self, tmp_path):
+        """Regression (#2572): more concurrent sessions than pool_size must all
+        complete by drawing from max_overflow — not deadlock or time out.
+
+        This is the failure the farm hit: printer callbacks held every base
+        connection, so unrelated requests waited on the pool. With headroom in
+        max_overflow, concurrent checkouts beyond pool_size still succeed.
+
+        Uses a file-based SQLite URL so it gets a real queue pool — the
+        in-memory URL forces a single-connection StaticPool that ignores
+        pool_size/max_overflow entirely.
+        """
+        db_file = tmp_path / "pool_regression.db"
+        eng = create_async_engine(f"sqlite+aiosqlite:///{db_file}", pool_size=2, max_overflow=10)
+        sm = async_sessionmaker(eng)
+
+        async def _one():
+            async with sm() as s:
+                await s.execute(text("SELECT 1"))
+                # Hold the checkout briefly so the calls genuinely overlap and
+                # force the pool past its base size of 2.
+                await asyncio.sleep(0.05)
+                return (await s.execute(text("SELECT 1"))).scalar()
+
+        try:
+            results = await asyncio.gather(*[_one() for _ in range(12)])
+        finally:
+            await eng.dispose()
+
+        assert results == [1] * 12
+
+
+class TestPoolStatus:
+    """P3: diagnostics snapshot."""
+
+    def test_get_pool_status_shape(self):
+        from backend.app.core.database import get_pool_status
+
+        status = get_pool_status()
+        assert status["dialect"] in ("sqlite", "postgresql")
+        for key in ("pool_size", "max_overflow", "pool_timeout", "pool_recycle", "pool_pre_ping"):
+            assert key in status["config"]
+        # Live gauges are present (values are ints on a QueuePool).
+        for key in ("current_size", "checked_out", "checked_in", "overflow"):
+            assert key in status
+
+
+class TestAuthEnabledCache:
+    """P1: cache the auth-enabled probe, but only ever cache True."""
+
+    class _Setting:
+        def __init__(self, value):
+            self.value = value
+
+    class _Result:
+        def __init__(self, setting):
+            self._setting = setting
+
+        def scalar_one_or_none(self):
+            return self._setting
+
+    class _CountingDB:
+        def __init__(self, value):
+            self._value = value
+            self.calls = 0
+
+        async def execute(self, *args, **kwargs):
+            self.calls += 1
+            setting = None if self._value is None else TestAuthEnabledCache._Setting(self._value)
+            return TestAuthEnabledCache._Result(setting)
+
+    @pytest.mark.asyncio
+    async def test_enabled_true_is_cached(self):
+        from backend.app.core import auth as auth_mod
+
+        auth_mod.invalidate_auth_enabled_cache()
+        db = self._CountingDB("true")
+
+        assert await auth_mod.is_auth_enabled(db) is True
+        assert db.calls == 1
+        # Second probe served from cache — no new query.
+        assert await auth_mod.is_auth_enabled(db) is True
+        assert db.calls == 1
+
+        # Invalidation forces a re-read (e.g. after set_auth_enabled).
+        auth_mod.invalidate_auth_enabled_cache()
+        assert await auth_mod.is_auth_enabled(db) is True
+        assert db.calls == 2
+        auth_mod.invalidate_auth_enabled_cache()
+
+    @pytest.mark.asyncio
+    async def test_disabled_is_never_cached(self):
+        """SECURITY: a disabled result must never be cached, so staleness can
+        only ever fail closed (require auth), never open."""
+        from backend.app.core import auth as auth_mod
+
+        auth_mod.invalidate_auth_enabled_cache()
+        db = self._CountingDB("false")
+
+        assert await auth_mod.is_auth_enabled(db) is False
+        assert db.calls == 1
+        # Every probe re-reads while disabled.
+        assert await auth_mod.is_auth_enabled(db) is False
+        assert db.calls == 2
+        auth_mod.invalidate_auth_enabled_cache()
+
+    @pytest.mark.asyncio
+    async def test_unconfigured_returns_false_and_is_not_cached(self):
+        from backend.app.core import auth as auth_mod
+
+        auth_mod.invalidate_auth_enabled_cache()
+        db = self._CountingDB(None)
+
+        assert await auth_mod.is_auth_enabled(db) is False
+        assert await auth_mod.is_auth_enabled(db) is False
+        assert db.calls == 2
+        auth_mod.invalidate_auth_enabled_cache()
+
+    @pytest.mark.asyncio
+    async def test_db_error_propagates_fail_closed(self):
+        """A probe error must propagate (fail closed), not be swallowed."""
+        from backend.app.core import auth as auth_mod
+
+        auth_mod.invalidate_auth_enabled_cache()
+
+        class _RaisingDB:
+            async def execute(self, *args, **kwargs):
+                raise RuntimeError("connection lost")
+
+        with pytest.raises(RuntimeError):
+            await auth_mod.is_auth_enabled(_RaisingDB())
+        auth_mod.invalidate_auth_enabled_cache()

backend/tests/unit/test_jti_revoked_session_reuse_2572.py

@@ -0,0 +1,83 @@
+"""``is_jti_revoked`` must reuse the caller's session when given one (#2572).
+
+The permission dependencies already hold a DB session when they check whether
+a JWT's ``jti`` is revoked. The old ``is_jti_revoked`` always opened a *second*
+``async_session``, so every authenticated request checked out two pooled
+connections instead of one. On a large farm a burst of concurrent logins then
+exhausted the pool (reporter @Jostxxl). Passing the existing session collapses
+each request back to a single checkout.
+
+These tests verify both that the revocation query is still correct and that a
+provided session is genuinely reused (no second checkout), while the no-session
+call still opens its own — the behaviour older callers rely on.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.core import auth as auth_mod
+from backend.app.core.auth import is_jti_revoked
+from backend.app.models.auth_ephemeral import AuthEphemeralToken
+
+
+def _revoked_row(token: str) -> AuthEphemeralToken:
+    return AuthEphemeralToken(
+        token=token,
+        token_type="revoked_jti",
+        expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
+    )
+
+
+@pytest.mark.asyncio
+async def test_revoked_jti_detected_via_provided_session(db_session):
+    """A revoked jti is reported revoked when the caller passes its session."""
+    db_session.add(_revoked_row("jti-revoked-1"))
+    await db_session.commit()
+
+    assert await is_jti_revoked("jti-revoked-1", db_session) is True
+
+
+@pytest.mark.asyncio
+async def test_unknown_jti_not_revoked_via_provided_session(db_session):
+    assert await is_jti_revoked("jti-never-seen", db_session) is False
+
+
+@pytest.mark.asyncio
+async def test_provided_session_is_reused_not_a_second_checkout(db_session, monkeypatch):
+    """PERF regression (#2572): passing ``db`` must NOT open a new session —
+    that second checkout per request is the pool pressure we removed."""
+    opened = {"n": 0}
+    original = auth_mod.async_session
+
+    def _counting(*args, **kwargs):
+        opened["n"] += 1
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(auth_mod, "async_session", _counting)
+
+    assert await is_jti_revoked("jti-with-session", db_session) is False
+    assert opened["n"] == 0, "is_jti_revoked opened its own session despite being given one"
+
+
+@pytest.mark.asyncio
+async def test_no_session_still_opens_its_own(test_engine, monkeypatch):
+    """Callers that check the jti before they have a session (e.g. the token
+    dependencies) must keep working — omitting ``db`` opens a short one."""
+    test_async_session = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+
+    async with test_async_session() as seed:
+        seed.add(_revoked_row("jti-revoked-2"))
+        await seed.commit()
+
+    opened = {"n": 0}
+
+    def _counting(*args, **kwargs):
+        opened["n"] += 1
+        return test_async_session(*args, **kwargs)
+
+    monkeypatch.setattr(auth_mod, "async_session", _counting)
+
+    assert await is_jti_revoked("jti-revoked-2") is True
+    assert opened["n"] == 1, "is_jti_revoked should open exactly one session when none is provided"