fix(db): configurable connection pool + auth_enabled cache for large farms (#2572)

Large PostgreSQL farms exhausted the fixed pool (pool_size=10 +
max_overflow=20): with ~93 printers every connection sat idle in
transaction and unrelated requests waited out the 30s pool timeout or
failed in the auth middleware.

- Make pool sizing env-configurable (DB_POOL_SIZE / DB_MAX_OVERFLOW /
  DB_POOL_TIMEOUT / DB_POOL_RECYCLE); raise the Postgres default to
  20 + 80 with pool_pre_ping + pool_recycle=1800.
- Cache the auth_enabled probe (30s) to drop a per-request DB round-trip.
  Only enabled=True is cached, so staleness fails closed; set_auth_enabled
  invalidates immediately.
- Add GET /api/v1/system/db-pool exposing resolved config + live
  checked_out/checked_in/overflow gauges without consuming a connection.

Session-hygiene (connections held across MQTT/FTP/camera/3MF I/O) is a
separate follow-up.
This commit is contained in:
maziggy
2026-07-16 08:16:48 +02:00
parent c2b23e5e61
commit 4ad43c96de
8 changed files with 363 additions and 7 deletions
+1
View File
@@ -5,6 +5,7 @@ All notable changes to Bambuddy will be documented in this file.
## [1.2.5b2] - Unreleased
### Fixed
- **PostgreSQL connection-pool exhaustion on large printer farms (#2572, reporter @Jostxxl)** — On a ~93-printer farm the SQLAlchemy pool (hard-coded `pool_size=10` + `max_overflow=20` = 30 connections) was repeatedly saturated with all connections `idle in transaction`; unrelated API requests then waited out the 30-second pool timeout or failed in the auth middleware, and an unauthenticated `/api/v1/printers` probe took ~25s to return 401. Three things fed the pressure: the pool was fixed and not configurable; every authenticated request re-queried `auth_enabled` from the DB (the middleware alone opened a session per request just to probe it); and the pool was small for a farm. This change (a) makes pool sizing configurable via `DB_POOL_SIZE` / `DB_MAX_OVERFLOW` / `DB_POOL_TIMEOUT` / `DB_POOL_RECYCLE` env vars and raises the PostgreSQL default to `20` + `80` (100 total) with `pool_pre_ping` and a 1800s `pool_recycle`; (b) caches the `auth_enabled` probe for 30s — only the *enabled* result is ever cached, so a stale read can only ever fail closed (require auth), never open, and any toggle invalidates it immediately; and (c) adds a `GET /api/v1/system/db-pool` diagnostic exposing the resolved config plus live `checked_out` / `checked_in` / `overflow` gauges (read without checking out a connection, so it stays truthful under saturation). Note: connections being held across slow MQTT/FTP/camera/3MF work — the underlying reason transactions sit idle — is a deeper session-hygiene change tracked separately; this drop relieves and instruments the problem and makes the farm sizing configurable. See the PostgreSQL wiki page for large-farm tuning and the required `max_connections` headroom.
- **P1S camera still black on every page load, recovering only after ~20 minutes (#2521, reporter @nnimby848)** — The previous round of fixes did not take, and the reporter re-tested on two daily builds to say so. The fan-out barrier added last time — a replacement stream waits for the displaced one's socket to close before dialling, so a printer that allows a single camera connection never sees two at once — was correct, and was being **bypassed**. `shutdown_broadcaster()` *popped* the broadcaster out of the registry and only then awaited its teardown, so for the duration of the socket close the registry slot sat empty. A `/camera/stream` request landing in that window found nothing, minted a broadcaster with no predecessor to wait for, and dialled port 6000 immediately. The barrier only engages when the displaced broadcaster is still findable — and the one path that tears a stream down on purpose removed it first, disabling the barrier in exactly the case it was written for. A page reload fires `/camera/stop` and the new `/camera/stream` **concurrently**, which is why it reproduced on essentially every load. The printer then held two connections, kept feeding the orphan, and starved the live viewer: the new socket connects (the reporter's logs show `Chamber image: connected`) and then receives nothing until the printer's TCP keepalive reaps the dead one — **his 20 minutes, to the minute**. The stopped broadcaster now stays in the registry so the next viewer chains behind its socket close, which is what the barrier always intended. Pinned by a test that counts *actual* sockets through the real stop-then-restream race and fails with `2` against the old code; the existing barrier tests placed the broadcaster into the registry by hand, which is precisely why they never caught this.
- **Every camera page load attached two viewers and abandoned one (#2521)** — Found while reproducing the above, and the reason it fired on *every* load rather than occasionally. The stream-token query runs whether or not authentication is enabled, and the camera page subscribes to it: the first render produced an `<img src>` with no token, the token arrived, and the re-render **changed the src**. The browser aborts the in-flight request and issues a second one — and with auth disabled no token is required, so *both* reached the backend and attached to the fan-out. The reporter's HAR shows it exactly: two requests to the same stream URL, same cache-buster, one without `token=` and one with. His backend log shows the consequence, `subscribers=2`, on a printer that allows one connection. The src is now rendered only once the token query has settled — one URL, one request, one viewer — and an auth-disabled install whose token endpoint fails still streams, because it never needed a token.
- **A viewer that left during a black stream stayed counted for 30 seconds (#2521)** — Also found on the way. A subscriber only checked whether its client was still connected *after* it had yielded a frame, or when a 30-second idle timeout fired. So a browser that walked away while the stream was producing nothing — the exact situation above — went on being counted as an attached viewer for up to half a minute. That matters beyond tidiness: `/camera/stop` consults the subscriber count to decide whether to tear the upstream down, so a phantom viewer could make it skip the teardown entirely and leave the socket open. Disconnects are now noticed within a second even when no frames are flowing.
+6
View File
@@ -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
+18
View File
@@ -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.
+44 -3
View File
@@ -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
@@ -862,6 +863,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 +906,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:
+11
View File
@@ -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
+76 -4
View File
@@ -23,12 +23,54 @@ def _set_sqlite_pragmas(dbapi_conn, connection_record):
cursor.close()
# 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():
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:
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."""
if is_sqlite():
kwargs = {"pool_size": 20, "max_overflow": 200}
else:
kwargs = {"pool_size": 10, "max_overflow": 20}
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.
+16
View File
@@ -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."""
@@ -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()