Merge pull request #755 from christianlouis/copilot/fix-timeout-error-notifications-api

fix: resolve merge conflict in database.py and remove duplicate entries from merge
This commit is contained in:
Christian Krakau-Louis
2026-03-19 12:29:39 +01:00
committed by GitHub
13 changed files with 787 additions and 100 deletions
+10
View File
@@ -7,6 +7,16 @@ GOTENBERG_URL=http://gotenberg:3000
ALLOW_FILE_DELETE=true # Allow deletion of file records
COMPLIANCE_ENABLED=true # Enable compliance templates dashboard (GDPR, HIPAA, SOC 2)
# **Database Connection Pool** (PostgreSQL / MySQL only; ignored for SQLite)
# DB_POOL_SIZE=10 # Persistent connections per worker (default: 10)
# DB_MAX_OVERFLOW=20 # Extra connections under burst (default: 20)
# DB_POOL_TIMEOUT=30 # Seconds to wait for a pool connection (default: 30)
# DB_POOL_RECYCLE=1800 # Recycle connections after N seconds (default: 1800)
# **Per-User Upload Rate Limiting** (health-aware, Redis-backed)
# UPLOAD_RATE_LIMIT_PER_USER=20 # Max uploads per user per window (default: 20)
# UPLOAD_RATE_LIMIT_WINDOW=60 # Sliding window in seconds (default: 60)
# **System Reset / Factory Reset**
# FACTORY_RESET_ON_STARTUP=false # Wipe all user data on every startup (demo/testing only)
# ENABLE_FACTORY_RESET=false # Show the System Reset page in admin UI
+7 -1
View File
@@ -20,6 +20,7 @@ from sqlalchemy.orm import Session
from app.auth import require_login
from app.config import settings
from app.database import get_db
from app.middleware.upload_rate_limit import require_upload_rate_limit
from app.models import FileProcessingStep, FileRecord, ProcessingLog
from app.tasks.convert_to_pdf import convert_to_pdf
from app.tasks.process_document import process_document
@@ -1298,7 +1299,12 @@ def _check_for_exact_duplicate(db: DbSession, target_path: str, safe_filename: s
@router.post("/ui-upload")
@require_login
async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...)):
async def ui_upload(
request: Request,
db: DbSession,
file: UploadFile = File(...),
_rate_ok: None = Depends(require_upload_rate_limit),
):
"""Endpoint to accept a user-uploaded file and enqueue it for processing."""
workdir = settings.workdir
+7 -2
View File
@@ -11,11 +11,12 @@ from typing import Optional
import aiofiles
import httpx
from fastapi import APIRouter, HTTPException, Request
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, HttpUrl, field_validator
from app.auth import require_login
from app.config import settings
from app.middleware.upload_rate_limit import require_upload_rate_limit
from app.tasks.process_document import process_document
from app.utils.allowed_types import ALLOWED_MIME_TYPES
from app.utils.filename_utils import sanitize_filename
@@ -107,7 +108,11 @@ def validate_file_type(content_type: str, filename: str) -> bool:
@router.post("/process-url")
@require_login
async def process_url(request: Request, url_request: URLUploadRequest):
async def process_url(
request: Request,
url_request: URLUploadRequest,
_rate_ok: None = Depends(require_upload_rate_limit),
):
"""
Download a file from a URL and enqueue it for processing.
+23 -30
View File
@@ -13,6 +13,24 @@ class Settings(BaseSettings):
database_url: str
redis_url: str
# Database connection-pool tuning (ignored for SQLite, which uses NullPool).
db_pool_size: int = Field(
default=10,
description="Number of persistent connections kept in the pool per worker process.",
)
db_max_overflow: int = Field(
default=20,
description="Additional connections allowed beyond db_pool_size under burst load.",
)
db_pool_timeout: int = Field(
default=30,
description="Seconds to wait for a connection from the pool before raising a TimeoutError.",
)
db_pool_recycle: int = Field(
default=1800,
description="Recycle (close and reopen) connections after this many seconds to avoid stale connections.",
)
openai_api_key: str
openai_base_url: str = "https://api.openai.com/v1" # Default to OpenAI's endpoint
openai_model: str = "gpt-4o-mini" # Default model
@@ -1115,43 +1133,18 @@ class Settings(BaseSettings):
),
)
# Database Connection Pool Configuration
# Controls SQLAlchemy QueuePool behaviour for PostgreSQL/MySQL.
# SQLite uses NullPool and ignores these settings.
db_pool_size: int = Field(
default=5,
description="Number of persistent connections kept in the pool. Ignored for SQLite.",
)
db_max_overflow: int = Field(
default=10,
description=("Maximum number of connections that can be opened beyond db_pool_size. Ignored for SQLite."),
)
db_pool_timeout: int = Field(
default=30,
description="Seconds to wait for a connection from the pool before raising an error. Ignored for SQLite.",
)
db_pool_recycle: int = Field(
default=1800,
description=(
"Seconds after which a connection is recycled to prevent stale connections. "
"Ignored for SQLite. Default: 1800 (30 minutes)."
),
)
# Per-user upload rate limiting (health-aware limiter)
# Controls how many uploads a single user may submit within a sliding window.
# Per-user upload rate limiting (health-aware, Redis-backed sliding window)
upload_rate_limit_per_user: int = Field(
default=20,
description=(
"Maximum number of uploads allowed per user within the upload_rate_limit_window. "
"The limiter may dynamically reduce this value when Redis queue depth or CPU load is high."
"Maximum number of file uploads allowed per user within the sliding window. "
"The effective limit may be reduced dynamically when the system is under heavy load "
"(high queue depth or CPU usage). Set to 0 to disable per-user upload rate limiting."
),
)
upload_rate_limit_window: int = Field(
default=60,
description=(
"Sliding window in seconds over which upload_rate_limit_per_user is enforced. Default: 60 seconds."
),
description="Sliding window size in seconds for per-user upload rate limiting (default: 60).",
)
# Rate Limiting Configuration (see SECURITY_AUDIT.md and docs/API.md)
+28 -13
View File
@@ -18,22 +18,37 @@ logger = logging.getLogger(__name__)
Base = declarative_base()
# Parse the DATABASE_URL
# ---------------------------------------------------------------------------
# Engine construction
# ---------------------------------------------------------------------------
DB_URL = settings.database_url
_db_url = make_url(DB_URL)
if _db_url.get_backend_name() == "sqlite":
# SQLite does not benefit from connection pooling; NullPool avoids contention.
engine = create_engine(DB_URL, connect_args={"check_same_thread": False}, poolclass=NullPool)
_parsed_url = make_url(DB_URL)
_connect_args: dict[str, Any] = {}
_engine_kwargs: dict[str, Any] = {
"pool_pre_ping": True, # detect stale / dropped connections before use
}
if _parsed_url.get_backend_name() == "sqlite":
# SQLite does not benefit from connection pooling and is prone to
# QueuePool exhaustion under concurrent access. NullPool opens a fresh
# connection for each request and closes it immediately afterwards,
# completely avoiding the "QueuePool limit reached" TimeoutError.
_connect_args["check_same_thread"] = False
_engine_kwargs["poolclass"] = NullPool
else:
# PostgreSQL / MySQL / other: use a configurable QueuePool.
engine = create_engine(
DB_URL,
poolclass=QueuePool,
pool_size=settings.db_pool_size,
max_overflow=settings.db_max_overflow,
pool_timeout=settings.db_pool_timeout,
pool_recycle=settings.db_pool_recycle,
# PostgreSQL / MySQL — use a bounded QueuePool with configurable limits.
_engine_kwargs["poolclass"] = QueuePool
_engine_kwargs.update(
{
"pool_size": settings.db_pool_size,
"max_overflow": settings.db_max_overflow,
"pool_timeout": settings.db_pool_timeout,
"pool_recycle": settings.db_pool_recycle,
}
)
engine = create_engine(DB_URL, connect_args=_connect_args, **_engine_kwargs)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+290
View File
@@ -0,0 +1,290 @@
"""Per-user, health-aware upload rate limiter for DocuElevate.
This module provides a FastAPI dependency that enforces per-user upload rate
limits using a Redis-backed sliding window counter. The effective limit is
dynamically reduced when the system is under heavy load (high Celery queue
depth or elevated CPU load average), ensuring the server remains responsive
to all users even during bulk-upload scenarios.
Usage in an endpoint::
from app.middleware.upload_rate_limit import require_upload_rate_limit
@router.post("/ui-upload")
@require_login
async def ui_upload(
request: Request,
_rate_ok: None = Depends(require_upload_rate_limit),
...
):
...
See ``docs/ConfigurationGuide.md`` for the configuration options
(``UPLOAD_RATE_LIMIT_PER_USER``, ``UPLOAD_RATE_LIMIT_WINDOW``).
"""
from __future__ import annotations
import logging
import os
import time
from typing import Any
import redis
from fastapi import HTTPException, Request, status
from app.config import settings
from app.utils.user_scope import get_current_owner_id
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Redis key prefix
# ---------------------------------------------------------------------------
_KEY_PREFIX = "docuelevate:upload_rate"
# ---------------------------------------------------------------------------
# Health-check queue names (Celery defaults used by DocuElevate)
# ---------------------------------------------------------------------------
_CELERY_QUEUES = ("document_processor", "default", "celery")
# ---------------------------------------------------------------------------
# Singleton Redis client (lazy-initialised; fail-open when unavailable)
# ---------------------------------------------------------------------------
_redis_client: redis.Redis | None = None
def _get_redis() -> redis.Redis | None:
"""Return a shared Redis client, or *None* when Redis is unavailable."""
global _redis_client
if _redis_client is not None:
return _redis_client
try:
_redis_client = redis.Redis.from_url(
settings.redis_url,
decode_responses=True,
socket_connect_timeout=2,
socket_timeout=2,
)
# Quick connectivity check raises on failure.
_redis_client.ping()
return _redis_client
except Exception: # noqa: BLE001
logger.debug("Redis unavailable for upload rate limiter falling back to allow-all", exc_info=True)
_redis_client = None
return None
# ---------------------------------------------------------------------------
# Health metrics helpers
# ---------------------------------------------------------------------------
def _get_queue_depth(r: redis.Redis) -> int:
"""Return the total number of pending tasks across all Celery queues."""
total = 0
for queue_name in _CELERY_QUEUES:
try:
total += r.llen(queue_name)
except Exception: # noqa: BLE001, S110
logger.debug("Could not read queue length for %r", queue_name, exc_info=True)
return total
def _get_cpu_load_ratio() -> float:
"""Return the 1-minute load average divided by the number of CPU cores.
Returns ``0.0`` on platforms that do not support :func:`os.getloadavg`
(e.g. Windows) so that the limiter never penalises on those systems.
"""
try:
load_1m = os.getloadavg()[0]
cpu_count = os.cpu_count() or 1
return load_1m / cpu_count
except (OSError, AttributeError):
return 0.0
def compute_effective_limit(
base_limit: int,
queue_depth: int = 0,
cpu_load_ratio: float = 0.0,
) -> tuple[int, float, str]:
"""Compute the effective upload rate limit based on system health.
The function applies a *reduction factor* (``0.0 < factor ≤ 1.0``) to the
configured base limit. Both queue depth and CPU load contribute
independently; the lowest factor wins.
Args:
base_limit: The configured maximum uploads per window.
queue_depth: Total pending tasks in Celery queues.
cpu_load_ratio: 1-minute load average divided by CPU count.
Returns:
A 3-tuple of ``(effective_limit, factor, reason)`` where *reason*
is a human-readable tag for logging.
"""
factor = 1.0
reason = "normal"
# --- Queue-depth thresholds ---
if queue_depth > 200:
factor, reason = min(factor, 0.10), f"critical_queue({queue_depth})"
elif queue_depth > 100:
factor, reason = min(factor, 0.25), f"high_queue({queue_depth})"
elif queue_depth > 50:
factor, reason = min(factor, 0.50), f"moderate_queue({queue_depth})"
# --- CPU-load thresholds ---
if cpu_load_ratio > 3.0:
new_factor = 0.10
if new_factor < factor:
factor, reason = new_factor, f"critical_cpu({cpu_load_ratio:.1f})"
elif cpu_load_ratio > 2.0:
new_factor = 0.25
if new_factor < factor:
factor, reason = new_factor, f"high_cpu({cpu_load_ratio:.1f})"
elif cpu_load_ratio > 1.5:
new_factor = 0.50
if new_factor < factor:
factor, reason = new_factor, f"moderate_cpu({cpu_load_ratio:.1f})"
effective = max(1, int(base_limit * factor))
return effective, factor, reason
# ---------------------------------------------------------------------------
# Core sliding-window check (Redis sorted set)
# ---------------------------------------------------------------------------
def _check_and_record(
r: redis.Redis,
user_id: str,
window: int,
effective_limit: int,
) -> dict[str, Any] | None:
"""Atomically check the user's upload count and record the new upload.
Uses a Redis sorted set where each member is a unique timestamp-based ID
and the score is the Unix timestamp. Entries older than *window* seconds
are pruned on every call so the set never grows unbounded.
Returns:
``None`` if the request is allowed, or a ``dict`` with ``count``,
``limit``, and ``retry_after`` if the limit is exceeded.
"""
key = f"{_KEY_PREFIX}:{user_id}"
now = time.time()
window_start = now - window
pipe = r.pipeline(transaction=True)
# 1. Remove entries outside the window
pipe.zremrangebyscore(key, "-inf", window_start)
# 2. Count current entries
pipe.zcard(key)
# 3. Retrieve the oldest entry's score (to compute retry_after)
pipe.zrange(key, 0, 0, withscores=True)
results = pipe.execute()
current_count: int = results[1]
oldest_entries: list = results[2]
if current_count >= effective_limit:
# Compute how long until the oldest entry expires from the window.
if oldest_entries:
oldest_score = oldest_entries[0][1]
retry_after = max(1, int((oldest_score + window) - now))
else:
retry_after = max(1, window // 2)
return {
"count": current_count,
"limit": effective_limit,
"retry_after": retry_after,
}
# 4. Record this upload (unique member = timestamp with random suffix)
member = f"{now}:{os.urandom(4).hex()}"
pipe2 = r.pipeline(transaction=True)
pipe2.zadd(key, {member: now})
pipe2.expire(key, window + 60) # TTL slightly longer than window
pipe2.execute()
return None
# ---------------------------------------------------------------------------
# FastAPI dependency
# ---------------------------------------------------------------------------
async def require_upload_rate_limit(request: Request) -> None:
"""FastAPI dependency that enforces per-user upload rate limits.
The dependency is designed to **fail open**: if Redis is unavailable the
request is allowed through so that uploads are never blocked by a
monitoring outage.
Raises:
HTTPException: 429 Too Many Requests when the per-user upload limit
is exceeded. The ``Retry-After`` header indicates how many
seconds the client should wait before retrying.
"""
r = _get_redis()
if r is None:
# Redis unavailable fail open.
return
# Identify the user (owner_id for multi-user, IP fallback).
user_id = get_current_owner_id(request)
if not user_id:
user_id = f"ip:{request.client.host}" if request.client else "ip:unknown"
base_limit: int = settings.upload_rate_limit_per_user
window: int = settings.upload_rate_limit_window
# Gather health metrics and compute effective limit.
try:
queue_depth = _get_queue_depth(r)
except Exception: # noqa: BLE001
queue_depth = 0
cpu_load_ratio = _get_cpu_load_ratio()
effective_limit, factor, health_reason = compute_effective_limit(base_limit, queue_depth, cpu_load_ratio)
# Sliding-window check.
try:
rejection = _check_and_record(r, user_id, window, effective_limit)
except Exception as exc: # noqa: BLE001
logger.warning("Upload rate-limit check failed (allowing request): %s", exc)
return
if rejection is not None:
retry_after = rejection["retry_after"]
logger.warning(
"Upload rate limit exceeded: user=%s count=%d/%d window=%ds health=%s retry_after=%ds",
user_id,
rejection["count"],
rejection["limit"],
window,
health_reason,
retry_after,
)
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=(
f"Upload rate limit exceeded ({rejection['count']}/{rejection['limit']} "
f"in {window}s). Retry after {retry_after}s."
),
headers={"Retry-After": str(retry_after)},
)
if factor < 1.0:
logger.info(
"Upload allowed with reduced limit: user=%s effective=%d/%d health=%s",
user_id,
effective_limit,
base_limit,
health_reason,
)
+44 -45
View File
@@ -39,6 +39,50 @@ SETTING_METADATA = {
"required": True,
"restart_required": True,
},
"db_pool_size": {
"category": "Core",
"description": (
"Number of persistent database connections kept in the pool per worker process. "
"Ignored for SQLite (which uses NullPool). Default: 10."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_max_overflow": {
"category": "Core",
"description": (
"Additional database connections allowed beyond db_pool_size under burst load. "
"Ignored for SQLite. Default: 20."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_pool_timeout": {
"category": "Core",
"description": (
"Seconds to wait for a database connection from the pool before raising a TimeoutError. "
"Ignored for SQLite. Default: 30."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_pool_recycle": {
"category": "Core",
"description": (
"Recycle (close and reopen) database connections after this many seconds "
"to avoid stale connections. Ignored for SQLite. Default: 1800."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"workdir": {
"category": "Core",
"description": "Working directory for file storage and processing",
@@ -2557,51 +2601,6 @@ SETTING_METADATA = {
"required": False,
"restart_required": False,
},
# Database Connection Pool
"db_pool_size": {
"category": "Core",
"description": (
"Number of persistent connections kept in the SQLAlchemy QueuePool. "
"Has no effect for SQLite databases. Default: 5."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_max_overflow": {
"category": "Core",
"description": (
"Maximum extra connections that can be opened beyond db_pool_size. "
"Has no effect for SQLite databases. Default: 10."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_pool_timeout": {
"category": "Core",
"description": (
"Seconds to wait for a connection from the pool before raising an error. "
"Has no effect for SQLite databases. Default: 30."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_pool_recycle": {
"category": "Core",
"description": (
"Seconds after which idle connections are recycled to prevent stale connections. "
"Has no effect for SQLite databases. Default: 1800 (30 minutes)."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
# Per-user upload rate limiting
"upload_rate_limit_per_user": {
"category": "Security",
+7 -1
View File
@@ -27,9 +27,11 @@ DocuElevate implements rate limiting to protect against abuse and DoS attacks. R
### Default Limits
- **Default endpoints**: 100 requests per minute
- **File upload**: 600 requests per minute
- **File upload**: 600 requests per minute (global) + 20 per user per 60 s (per-user, health-aware)
- **Authentication**: 10 requests per minute
**Per-user upload rate limiting**: Upload endpoints (`/api/ui-upload`, `/api/process-url`) enforce a per-user sliding-window limit that adapts to system load. Under heavy queue depth or high CPU usage, the effective limit is reduced automatically. See the [Configuration Guide](ConfigurationGuide.md#per-user-upload-rate-limiting) for details.
**Note**: Document processing endpoints (OCR, metadata extraction) use built-in queue throttling to control processing rates and prevent upstream API overloads. No additional API-level rate limit is applied to processing endpoints.
### Rate Limit Headers
@@ -53,6 +55,10 @@ RATE_LIMITING_ENABLED=true
RATE_LIMIT_DEFAULT=100/minute
RATE_LIMIT_UPLOAD=600/minute
RATE_LIMIT_AUTH=10/minute
# Per-user upload rate limiting (health-aware)
UPLOAD_RATE_LIMIT_PER_USER=20 # Max uploads per user per window
UPLOAD_RATE_LIMIT_WINDOW=60 # Sliding window in seconds
```
See [Configuration Guide](ConfigurationGuide.md) for more details.
+26
View File
@@ -11,6 +11,10 @@ Configuration is primarily done through environment variables specified in a `.e
| **Variable** | **Description** | **Example** |
|------------------------|----------------------------------------------------------|--------------------------------|
| `DATABASE_URL` | Path/URL to the SQLite database (or other SQL backend). Use the [Database Wizard](/database-wizard) for guided setup. See [Database Configuration](DatabaseConfiguration.md). | `sqlite:///./app/database.db` |
| `DB_POOL_SIZE` | Number of persistent connections in the pool per worker (PostgreSQL/MySQL only; ignored for SQLite). | `10` |
| `DB_MAX_OVERFLOW` | Additional connections beyond `DB_POOL_SIZE` under burst load (PostgreSQL/MySQL only). | `20` |
| `DB_POOL_TIMEOUT` | Seconds to wait for a pool connection before raising `TimeoutError` (PostgreSQL/MySQL only). | `30` |
| `DB_POOL_RECYCLE` | Recycle connections after this many seconds to avoid stale connections (PostgreSQL/MySQL only). | `1800` |
| `REDIS_URL` | URL for Redis, used by Celery for broker & result store. | `redis://redis:6379/0` |
| `WORKDIR` | Working directory for the application. | `/workdir` |
| `GOTENBERG_URL` | Gotenberg PDF processing URL. | `http://gotenberg:3000` |
@@ -81,6 +85,28 @@ Control how the web UI queues and paces file uploads to avoid overwhelming the b
**Example**: With `UPLOAD_CONCURRENCY=3` and `UPLOAD_QUEUE_DELAY_MS=500`, a directory of 5,000 files is uploaded ≈ 3 at a time with 500 ms pacing the backend processes files at its own rate while the queue drains in the background without triggering API rate limits.
### Per-User Upload Rate Limiting
Server-side rate limiting that prevents any single user from overwhelming the system with bulk uploads. The limiter uses a Redis-backed sliding window and dynamically adjusts limits based on system health.
| **Variable** | **Description** | **Default** |
|--------------------------------|------------------------------------------------------------------------------------------------------------------------------|-------------|
| `UPLOAD_RATE_LIMIT_PER_USER` | Maximum uploads allowed per user within the sliding window. Effective limit may be reduced under load. | `20` |
| `UPLOAD_RATE_LIMIT_WINDOW` | Sliding window size in seconds. | `60` |
**Health-aware dynamic limiting**: The effective per-user limit is automatically reduced when the system is under heavy load:
| **System condition** | **Effective limit** | **Trigger** |
|--------------------------------|---------------------|--------------------------------|
| Normal | 100 % of base | Queue < 50, CPU load normal |
| Moderate load | 50 % of base | Queue 50100 or CPU > 1.5× |
| High load | 25 % of base | Queue 100200 or CPU > 2× |
| Critical load | 10 % of base | Queue > 200 or CPU > 3× |
When a user exceeds the limit, the server returns **HTTP 429 Too Many Requests** with a `Retry-After` header. The browser client (see *Client-Side Upload Throttling* above) automatically pauses and retries.
> **Note**: The limiter fails open — if Redis is unavailable, all uploads are allowed through so that a monitoring outage never blocks document processing.
### File Upload Size Limits
**Security Feature**: Control file upload sizes to prevent resource exhaustion attacks. See [SECURITY_AUDIT.md](../SECURITY_AUDIT.md#5-file-upload-size-limits) for security details.
+25 -8
View File
@@ -337,18 +337,26 @@ The Helm chart includes a pre-install and pre-upgrade Job hook that runs `alembi
## Connection Pooling
SQLAlchemy manages a connection pool automatically. The defaults are suitable for most deployments. For high-concurrency or Kubernetes deployments you may want to tune:
SQLAlchemy manages a connection pool automatically. DocuElevate selects the pool
strategy based on the database backend:
- **SQLite** — uses `NullPool` (a fresh connection per request, closed immediately).
This avoids the `QueuePool limit reached` `TimeoutError` that can occur under
concurrent load because SQLite does not benefit from persistent connection pooling.
- **PostgreSQL / MySQL** — uses a bounded `QueuePool` whose size is configurable
via environment variables.
```bash
# Optional — these are set via environment variables if you extend app/database.py
# Typical production values:
DB_POOL_SIZE=10 # Number of persistent connections per worker
DB_MAX_OVERFLOW=20 # Additional connections allowed beyond pool_size
DB_POOL_TIMEOUT=30 # Seconds to wait for a connection from the pool
DB_POOL_RECYCLE=1800 # Recycle connections after 30 minutes (avoids stale connections)
# Tune these for PostgreSQL / MySQL (ignored when using SQLite):
DB_POOL_SIZE=10 # Number of persistent connections per worker (default: 10)
DB_MAX_OVERFLOW=20 # Additional connections allowed beyond pool_size (default: 20)
DB_POOL_TIMEOUT=30 # Seconds to wait for a connection from the pool (default: 30)
DB_POOL_RECYCLE=1800 # Recycle connections after 30 minutes (default: 1800)
```
> **Note:** These environment variables are not exposed in the default `app/config.py`. If you need to tune them, extend the database engine creation in `app/database.py`.
All backends also enable `pool_pre_ping`, which sends a lightweight health-check
before each connection is handed out. This detects stale or dropped connections
and transparently reconnects.
For **PgBouncer** (external connection pooling), point `DATABASE_URL` at your PgBouncer instance and use transaction-mode pooling:
@@ -512,4 +520,13 @@ Then retry `alembic upgrade head`.
Either increase `max_connections` in `postgresql.conf` or add PgBouncer in front of PostgreSQL. The default PostgreSQL `max_connections` is `100`; reduce `DB_POOL_SIZE` per worker to stay within this limit.
### "QueuePool limit reached" TimeoutError (SQLite)
If you see `TimeoutError: QueuePool limit of size 5 overflow 10 reached`, your
deployment is still running an older version of DocuElevate that used a bounded
connection pool for SQLite. Upgrade to the latest release — SQLite now uses
`NullPool`, which eliminates this error entirely. If you are already on the
latest version and are still seeing pool exhaustion, ensure you are not
overriding the engine creation manually.
For more help, see the [Troubleshooting Guide](Troubleshooting.md).
+9
View File
@@ -115,6 +115,7 @@ def client(db_session) -> TestClient:
# Import the canonical get_db function
from app.database import get_db
from app.middleware.upload_rate_limit import require_upload_rate_limit
# Override the get_db dependency to use our test database
def override_get_db():
@@ -126,6 +127,14 @@ def client(db_session) -> TestClient:
# Override the single canonical get_db dependency
fastapi_app.dependency_overrides[get_db] = override_get_db
# Disable per-user upload rate limiting in tests so that upload-heavy
# test suites are not rejected with 429 Too Many Requests.
async def _no_rate_limit() -> None:
"""No-op override: skip upload rate limiting during tests."""
return None
fastapi_app.dependency_overrides[require_upload_rate_limit] = _no_rate_limit
# Use base_url to satisfy TrustedHostMiddleware
with TestClient(fastapi_app, base_url="http://localhost") as test_client:
yield test_client
+46
View File
@@ -998,3 +998,49 @@ class TestAlembicUpgrade:
# Verify head is reachable
heads = script.get_heads()
assert len(heads) == 1 # Should be a single linear chain
@pytest.mark.unit
class TestEnginePoolConfiguration:
"""Tests for database engine pool configuration (pool class and options)."""
def test_sqlite_engine_uses_null_pool(self):
"""SQLite engines must use NullPool to prevent QueuePool exhaustion."""
from sqlalchemy.pool import NullPool
from app.database import engine
# The test environment uses SQLite, so NullPool should be in effect.
assert isinstance(engine.pool, NullPool)
def test_create_engine_sqlite_null_pool(self):
"""Explicitly create a SQLite engine to confirm NullPool is applied."""
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool
test_engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=NullPool,
)
assert isinstance(test_engine.pool, NullPool)
test_engine.dispose()
def test_pool_settings_exist_in_config(self):
"""Verify that pool tuning settings are exposed through config."""
from app.config import settings
assert hasattr(settings, "db_pool_size")
assert hasattr(settings, "db_max_overflow")
assert hasattr(settings, "db_pool_timeout")
assert hasattr(settings, "db_pool_recycle")
def test_pool_settings_have_sensible_defaults(self):
"""Default pool settings should be larger than SQLAlchemy's built-in defaults."""
from app.config import settings
# SQLAlchemy defaults: pool_size=5, max_overflow=10
assert settings.db_pool_size >= 10
assert settings.db_max_overflow >= 20
assert settings.db_pool_timeout >= 30
assert settings.db_pool_recycle >= 1800
+265
View File
@@ -0,0 +1,265 @@
"""Tests for per-user health-aware upload rate limiting (app/middleware/upload_rate_limit.py)."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from app.middleware.upload_rate_limit import compute_effective_limit
# ---------------------------------------------------------------------------
# Tests for compute_effective_limit (pure function, no Redis needed)
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestComputeEffectiveLimit:
"""Tests for the health-aware effective-limit calculation."""
def test_normal_conditions_return_base_limit(self):
"""Under normal conditions the full base limit should be returned."""
effective, factor, reason = compute_effective_limit(20, queue_depth=0, cpu_load_ratio=0.0)
assert effective == 20
assert factor == 1.0
assert reason == "normal"
def test_moderate_queue_halves_limit(self):
"""Queue depth > 50 should halve the base limit."""
effective, factor, reason = compute_effective_limit(20, queue_depth=60, cpu_load_ratio=0.0)
assert effective == 10
assert factor == 0.5
assert "moderate_queue" in reason
def test_high_queue_quarters_limit(self):
"""Queue depth > 100 should quarter the base limit."""
effective, factor, reason = compute_effective_limit(20, queue_depth=120, cpu_load_ratio=0.0)
assert effective == 5
assert factor == 0.25
assert "high_queue" in reason
def test_critical_queue_drops_to_ten_percent(self):
"""Queue depth > 200 should drop to 10% of base limit."""
effective, factor, reason = compute_effective_limit(20, queue_depth=250, cpu_load_ratio=0.0)
assert effective == 2
assert factor == 0.10
assert "critical_queue" in reason
def test_moderate_cpu_halves_limit(self):
"""CPU load ratio > 1.5 should halve the base limit."""
effective, factor, reason = compute_effective_limit(20, queue_depth=0, cpu_load_ratio=1.8)
assert effective == 10
assert factor == 0.5
assert "moderate_cpu" in reason
def test_high_cpu_quarters_limit(self):
"""CPU load ratio > 2.0 should quarter the base limit."""
effective, factor, reason = compute_effective_limit(20, queue_depth=0, cpu_load_ratio=2.5)
assert effective == 5
assert factor == 0.25
assert "high_cpu" in reason
def test_critical_cpu_drops_to_ten_percent(self):
"""CPU load ratio > 3.0 should drop to 10% of base limit."""
effective, factor, reason = compute_effective_limit(20, queue_depth=0, cpu_load_ratio=4.0)
assert effective == 2
assert factor == 0.10
assert "critical_cpu" in reason
def test_worst_metric_wins(self):
"""The lowest factor from queue and CPU should be applied."""
# Queue says 0.5, CPU says 0.25 → 0.25 wins
effective, factor, reason = compute_effective_limit(20, queue_depth=60, cpu_load_ratio=2.5)
assert effective == 5
assert factor == 0.25
def test_minimum_effective_limit_is_one(self):
"""Even under extreme load the effective limit must be ≥ 1."""
effective, _factor, _reason = compute_effective_limit(1, queue_depth=999, cpu_load_ratio=10.0)
assert effective >= 1
def test_zero_base_limit_returns_zero(self):
"""A base limit of 0 (disabled) should clamp to at least 1."""
effective, _factor, _reason = compute_effective_limit(0, queue_depth=0, cpu_load_ratio=0.0)
# max(1, int(0 * 1.0)) = max(1, 0) = 1
# A base_limit of 0 means "disabled" and is handled upstream
# (the dependency skips the check entirely), but the pure function
# still clamps to 1 as a safety net.
assert effective == 1
# ---------------------------------------------------------------------------
# Tests for the FastAPI dependency (mocked Redis)
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestRequireUploadRateLimit:
"""Tests for the require_upload_rate_limit FastAPI dependency."""
@pytest.mark.asyncio
async def test_allows_request_when_redis_unavailable(self):
"""When Redis is down the dependency should fail open (allow the request)."""
from app.middleware.upload_rate_limit import require_upload_rate_limit
mock_request = MagicMock()
mock_request.session = {}
mock_request.client = MagicMock()
mock_request.client.host = "127.0.0.1"
with patch("app.middleware.upload_rate_limit._get_redis", return_value=None):
# Should NOT raise
result = await require_upload_rate_limit(mock_request)
assert result is None
@pytest.mark.asyncio
async def test_allows_request_under_limit(self):
"""A user below the rate limit should be allowed through."""
from app.middleware.upload_rate_limit import require_upload_rate_limit
mock_request = MagicMock()
mock_request.session = {"user": {"username": "testuser"}}
mock_request.client = MagicMock()
mock_request.client.host = "10.0.0.1"
mock_redis = MagicMock()
mock_pipe = MagicMock()
mock_pipe.execute.return_value = [
0, # zremrangebyscore result
5, # zcard — current count (under limit of 20)
[], # zrange oldest
]
mock_redis.pipeline.return_value = mock_pipe
mock_redis.llen.return_value = 0 # empty queues
mock_pipe2 = MagicMock()
mock_pipe2.execute.return_value = [True, True]
# The second pipeline call (record upload)
mock_redis.pipeline.side_effect = [mock_pipe, mock_pipe2]
with (
patch("app.middleware.upload_rate_limit._get_redis", return_value=mock_redis),
patch("app.middleware.upload_rate_limit.get_current_owner_id", return_value="testuser"),
patch("app.middleware.upload_rate_limit._get_cpu_load_ratio", return_value=0.1),
):
result = await require_upload_rate_limit(mock_request)
assert result is None
@pytest.mark.asyncio
async def test_rejects_request_over_limit(self):
"""A user at or over the rate limit should receive a 429."""
from fastapi import HTTPException
from app.middleware.upload_rate_limit import require_upload_rate_limit
mock_request = MagicMock()
mock_request.session = {"user": {"username": "spammer"}}
mock_request.client = MagicMock()
mock_request.client.host = "10.0.0.2"
mock_redis = MagicMock()
mock_pipe = MagicMock()
mock_pipe.execute.return_value = [
0, # zremrangebyscore
20, # zcard — at limit
[("oldest_entry", 1000000.0)], # oldest entry for retry_after
]
mock_redis.pipeline.return_value = mock_pipe
mock_redis.llen.return_value = 0
with (
patch("app.middleware.upload_rate_limit._get_redis", return_value=mock_redis),
patch("app.middleware.upload_rate_limit.get_current_owner_id", return_value="spammer"),
patch("app.middleware.upload_rate_limit._get_cpu_load_ratio", return_value=0.0),
):
with pytest.raises(HTTPException) as exc_info:
await require_upload_rate_limit(mock_request)
assert exc_info.value.status_code == 429
assert "Retry-After" in exc_info.value.headers
@pytest.mark.asyncio
async def test_health_reduces_effective_limit(self):
"""When queues are deep, the effective limit should drop, causing a 429 sooner."""
from fastapi import HTTPException
from app.middleware.upload_rate_limit import require_upload_rate_limit
mock_request = MagicMock()
mock_request.session = {"user": {"username": "normaluser"}}
mock_request.client = MagicMock()
mock_request.client.host = "10.0.0.3"
mock_redis = MagicMock()
mock_pipe = MagicMock()
# 12 uploads already — under normal limit of 20 but over health-reduced limit
mock_pipe.execute.return_value = [
0, # zremrangebyscore
12, # zcard — 12 uploads in window
[("oldest", 1000000.0)],
]
mock_redis.pipeline.return_value = mock_pipe
# Simulate deep queue (>100) → effective limit = 25% of 20 = 5
mock_redis.llen.return_value = 40 # 40 per queue * 3 = 120 total
with (
patch("app.middleware.upload_rate_limit._get_redis", return_value=mock_redis),
patch("app.middleware.upload_rate_limit.get_current_owner_id", return_value="normaluser"),
patch("app.middleware.upload_rate_limit._get_cpu_load_ratio", return_value=0.0),
):
with pytest.raises(HTTPException) as exc_info:
await require_upload_rate_limit(mock_request)
assert exc_info.value.status_code == 429
@pytest.mark.asyncio
async def test_falls_back_to_ip_when_no_user(self):
"""Unauthenticated requests should use IP-based rate limiting."""
from app.middleware.upload_rate_limit import require_upload_rate_limit
mock_request = MagicMock()
mock_request.session = {}
mock_request.client = MagicMock()
mock_request.client.host = "192.168.1.100"
mock_redis = MagicMock()
mock_pipe = MagicMock()
mock_pipe.execute.return_value = [0, 0, []]
mock_redis.pipeline.return_value = mock_pipe
mock_redis.llen.return_value = 0
mock_pipe2 = MagicMock()
mock_pipe2.execute.return_value = [True, True]
mock_redis.pipeline.side_effect = [mock_pipe, mock_pipe2]
with (
patch("app.middleware.upload_rate_limit._get_redis", return_value=mock_redis),
patch("app.middleware.upload_rate_limit.get_current_owner_id", return_value=None),
patch("app.middleware.upload_rate_limit._get_cpu_load_ratio", return_value=0.0),
):
result = await require_upload_rate_limit(mock_request)
assert result is None
# ---------------------------------------------------------------------------
# Tests for configuration
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestUploadRateLimitConfig:
"""Tests for upload rate limit configuration settings."""
def test_settings_exist(self):
"""Verify per-user upload rate limit settings are exposed in config."""
from app.config import settings
assert hasattr(settings, "upload_rate_limit_per_user")
assert hasattr(settings, "upload_rate_limit_window")
def test_sensible_defaults(self):
"""Default values should be reasonable for a multi-user system."""
from app.config import settings
assert settings.upload_rate_limit_per_user >= 10
assert settings.upload_rate_limit_per_user <= 100
assert settings.upload_rate_limit_window >= 30
assert settings.upload_rate_limit_window <= 300