diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 853585a..d624aa8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -46,6 +46,7 @@ jobs: env: DATABASE_URL: postgresql+psycopg://fastapi_user:fastapi_password@localhost:5432/fastapi_test + BENCHMARK_DATABASE_URL: postgresql+psycopg://fastapi_user:fastapi_password@localhost:5432/fastapi_test SECRET_KEY: ci-only-secret-key-with-at-least-32-bytes JWT_AUDIENCE: fastapi-client JWT_ISSUER: fastapi-production-api @@ -81,6 +82,32 @@ jobs: - name: Run tests run: uv run pytest --cov-report=xml + - name: Run database benchmark smoke test + run: | + uv run python -m benchmarks.db prepare --records 25 + uv run python -m benchmarks.db run \ + --mode both \ + --scenarios authenticated_read refresh_rotation \ + --iterations 8 \ + --warmup 2 \ + --concurrency 2 \ + --pool-size 2 \ + --output benchmark-smoke.json + uv run python -m benchmarks.db verify-async-rollback + + - name: Clean up database benchmark fixtures + if: always() + run: uv run python -m benchmarks.db cleanup + + - name: Upload database benchmark smoke result + if: always() + uses: actions/upload-artifact@v7 + with: + name: database-benchmark-smoke + path: benchmark-smoke.json + if-no-files-found: warn + retention-days: 14 + - name: Upload coverage report uses: actions/upload-artifact@v7 with: diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 5555ec1..8f20fb8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -42,6 +42,7 @@ when distributed rate limiting is enabled. | Error boundary | `src/app/exceptions/` | Stable client errors and safe unexpected-error responses | | Schema evolution | `alembic/` | Ordered PostgreSQL migrations | | Verification | `tests/` | Endpoint, security, failure-path, and operational tests | +| Database evaluation | `benchmarks/` | Opt-in sync baseline and isolated async prototype; never imported by production startup | Routers should remain thin: validate input, compose dependencies and services, and translate results into HTTP responses. Reusable authentication or database @@ -146,6 +147,13 @@ for schema changes. Production releases should run migrations once before starting new application workers. Do not let every worker race to apply schema changes. +The production persistence path remains synchronous for v1.3.0. The asyncpg +engine under `benchmarks/` exists only for controlled comparison and preserves +equivalent SQL and transaction boundaries. It is not an alternate application +dependency. See [DATABASE_BENCHMARKS.md](DATABASE_BENCHMARKS.md) and +[ADR 0001](docs/decisions/0001-keep-sync-sqlalchemy.md) for the evidence gate +and decision rationale. + In transactional delivery mode, the lifecycle token and encrypted outbox row commit together. Workers claim due rows in short `FOR UPDATE SKIP LOCKED` transactions, commit leases before SMTP I/O, and condition finalization on the diff --git a/CHANGELOG.md b/CHANGELOG.md index e37c28b..b6ade9f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,11 @@ project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - W3C Trace Context propagation through bounded outbox metadata, JSON log `trace_id`/`span_id` correlation, OTLP/HTTP export, sampling controls, and secret-redaction safeguards +- Reproducible PostgreSQL benchmark harness with deterministic isolated fixtures, + sync/async query parity, latency percentiles, pool telemetry, and JSON output +- CI correctness smoke coverage for the asyncpg prototype and rollback semantics +- Architecture decision retaining synchronous SQLAlchemy for v1.3.0 until + representative measurements justify a separately scoped migration ## [1.2.0] - 2026-08-09 diff --git a/DATABASE_BENCHMARKS.md b/DATABASE_BENCHMARKS.md new file mode 100644 index 0000000..77f2b5c --- /dev/null +++ b/DATABASE_BENCHMARKS.md @@ -0,0 +1,144 @@ +# Database benchmark and async evaluation + +This guide defines the reproducible database evaluation used for the v1.3 +architecture decision. The production application remains synchronous. The +async implementation under `benchmarks/` is an isolated experiment and is not +imported by application startup or included as a console entry point. + +## Decision + +Keep synchronous SQLAlchemy for v1.3.0. The repository does not yet have +representative production measurements showing that an async migration would +justify its transaction, testing, debugging, and operational cost. The new +comparison harness makes that decision repeatable; a future dedicated migration +issue requires sustained evidence against a real workload before changing the +production path. + +CI smoke timings are deliberately not treated as performance evidence. Shared +runners are noisy and the smoke workload is too short. The decision can be +revisited when five extended runs show a material, repeatable improvement at +the expected concurrency without increasing error rate or weakening existing +transaction semantics. + +## What is measured + +Both modes execute the same SQL and transaction boundaries against the same +fixture data and PostgreSQL instance. + +| Scenario | Database behavior represented | +| --- | --- | +| `authenticated_read` | Indexed current-user lookup after token validation | +| `session_list` | Bounded device-session list through a compound index | +| `refresh_rotation` | Locked read and update in one rotation transaction | +| `write` | Representative bounded update transaction | +| `mixed` | Equal deterministic mix of the four database-bound scenarios | + +The harness reports requests per second, median/p95/p99/min/max latency, error +rate and bounded error classes, connection acquisition time, checkout/checkin +counts, and peak checked-out connections. It does not record credentials, +queries containing user input, application tokens, or row contents. + +The database-focused scenarios intentionally exclude bcrypt, JWT signing, +HTTP parsing, Redis latency, SMTP, and tracing export. Those costs should be +measured separately by the planned end-to-end load-testing work. This separation +prevents external dependency latency from being mistaken for a database-driver +effect. + +## Safety model + +The tool never implicitly uses `DATABASE_URL`. Set `BENCHMARK_DATABASE_URL` or +pass `--database-url`. It accepts only PostgreSQL on `localhost`, `127.0.0.1`, +`::1`, or the local Compose service name `postgres`, and the database name must +contain `test`, `bench`, `local`, or `dev`. + +Fixtures live only in the `fastapi_benchmark` schema. `prepare` resets that +schema's two tables; `cleanup` drops only that schema. Never point the tool at a +shared or production database. + +## Short local smoke run + +Start and migrate the local services, then use the test database or create a +dedicated database whose name contains `benchmark`: + +```bash +export BENCHMARK_DATABASE_URL='postgresql+psycopg://fastapi_user:fastapi_password@localhost:5432/fastapi_test' +uv run python -m benchmarks.db prepare --records 100 +uv run python -m benchmarks.db run --mode both --scenarios authenticated_read refresh_rotation --iterations 20 --warmup 5 --concurrency 2 --pool-size 2 --output benchmark-smoke.json +uv run python -m benchmarks.db verify-async-rollback +uv run python -m benchmarks.db cleanup +``` + +PowerShell uses the same commands after setting: + +```powershell +$env:BENCHMARK_DATABASE_URL='postgresql+psycopg://fastapi_user:fastapi_password@localhost:5432/fastapi_test' +``` + +CI runs this bounded smoke path and uploads `benchmark-smoke.json`. It verifies +imports, fixtures, both drivers, stable result structure, and async rollback; +it enforces no timing threshold. + +## Extended comparison methodology + +Record the following alongside every retained result artifact: + +- commit SHA and unchanged `uv.lock`; +- Python 3.13 patch version, operating system, CPU model/count, and memory; +- PostgreSQL 17 patch version, configuration, and whether it is local; +- one application/benchmark process unless a worker comparison is explicit; +- identical `--pool-size` and `--max-overflow` for both modes; +- dataset size, scenario, concurrency, warm-up, iterations, and repetition; +- tracing disabled, log level, metrics mode, and Redis excluded from DB-only runs; +- CPU and resident-memory observations from the OS or container runtime; +- cold-start observations kept separate from steady-state results. + +Recommended controlled run: + +1. Use a dedicated local PostgreSQL 17 database on loopback with no competing + workload and prepare at least 10,000 principals. +2. Keep pool size at 5 and overflow at 0 for both drivers. +3. For each concurrency in 1, 5, 10, 25, and 50, run all scenarios with 500 + warm-up operations and at least 10,000 measured operations. +4. Alternate sync-first and async-first ordering to reduce thermal/order bias. +5. Repeat the complete matrix five times. Retain raw JSON rather than only + copied summary values. +6. Report the median of run-level throughput and latency percentiles plus the + observed range. Investigate all errors and pool waits before comparing speed. + +Example extended invocation for one matrix point: + +```bash +uv run python -m benchmarks.db prepare --records 10000 +uv run python -m benchmarks.db run --mode both --iterations 10000 --warmup 500 --concurrency 25 --pool-size 5 --max-overflow 0 --output results/c25-run1.json +uv run python -m benchmarks.db cleanup +``` + +## Interpretation and adoption gate + +Throughput alone is insufficient. Compare p95/p99 latency, acquisition wait, +peak pool use, error rate, CPU, memory, and transaction correctness. Direct +equivalence ends above the driver/session layer: the synchronous production +stack can use multiple worker processes, while one async process multiplexes +database waits on an event loop. Any worker-level comparison must document that +difference rather than presenting it as a driver-only result. + +Open a dedicated async-migration issue only if representative extended runs +show a repeatable improvement large enough to matter to the service SLO (a 20% +change is a useful review trigger, not a universal threshold), with equal error +rate, bounded connection pressure, passing rollback/security tests, and a clear +operational plan. Otherwise retain the simpler synchronous path or consider +async only for a narrowly isolated I/O-heavy component. + +## Current comparison record + +| Evidence | Sync | Async prototype | Architectural weight | +| --- | --- | --- | --- | +| CI PostgreSQL smoke | Required and artifacted | Required and artifacted | Correctness only | +| Transaction rollback | Existing production tests | Explicit rollback probe | Must pass | +| Production integration | Current maintained path | Not integrated | Favors sync | +| Representative extended measurements | Not yet supplied | Not yet supplied | No migration evidence | + +The absence of representative measurements is recorded explicitly rather than +replaced with fabricated numbers. Under the issue's decision guardrails, that +evidence supports keeping sync for v1.3.0 and deferring any full migration. + diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 6e777ff..1821aed 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -100,6 +100,15 @@ The test database is flushed before and after these tests. Never point Worker concurrency tests require PostgreSQL because SQLite does not implement `FOR UPDATE SKIP LOCKED`. The standard CI job runs these tests on PostgreSQL. +## Database performance evaluation + +The opt-in benchmark compares the maintained synchronous engine with a separate +asyncpg prototype without changing application code. It requires an explicitly +named local/test database and owns only the `fastapi_benchmark` schema. Run the +short smoke workflow or the controlled extended matrix in +[DATABASE_BENCHMARKS.md](DATABASE_BENCHMARKS.md). Never use production +credentials or infer performance regressions from CI runner timings. + ## Typical contribution workflow ```bash @@ -233,4 +242,4 @@ issuer in the key. Redis-backed integration tests require `REDIS_TEST_URL`. CI supplies a dedicated Redis database so multi-instance sharing, TTL behavior, invalidation, outage fallback, and refresh-lock behavior can be exercised without touching -unrelated Redis data. \ No newline at end of file +unrelated Redis data. diff --git a/README.md b/README.md index 56ee49d..0021df2 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,7 @@ contributor workflows, and local troubleshooting. | --- | --- | | [API examples](API_EXAMPLES.md) | Register, authenticate, rotate tokens, call admin routes, and inspect operations endpoints | | [Architecture](ARCHITECTURE.md) | Understand module boundaries, request flow, authentication, transactions, and extension points | +| [Database benchmarks](DATABASE_BENCHMARKS.md) | Reproduce sync/async PostgreSQL comparisons and understand the v1.3 architecture decision | | [Local development](DEVELOPMENT.md) | Set up a checkout, run common commands, contribute, or troubleshoot locally | | [Deployment](DEPLOYMENT.md) | Configure a production host, release safely, terminate TLS, and operate the service | | [Monitoring](MONITORING.md) | Configure probes, Prometheus, multi-worker metrics, alerts, logs, and incident diagnosis | @@ -244,6 +245,11 @@ Pytest measures statement and branch coverage for the application packages and fails below 90%. CI also publishes `coverage.xml` as a workflow artifact for review and downstream reporting. +CI also runs a bounded PostgreSQL correctness smoke test for the isolated sync +and async database benchmark and uploads its machine-readable JSON artifact. +See [DATABASE_BENCHMARKS.md](DATABASE_BENCHMARKS.md); shared-runner timings are +not used as performance gates. + CI runs against PostgreSQL 17 rather than silently substituting SQLite. It also verifies that the built wheel contains and can import the application. diff --git a/ROADMAP.md b/ROADMAP.md index aeb153b..612a568 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -58,7 +58,7 @@ Status: in progress - Redis-backed rate limiting (completed) - Transactional outbox and background worker processing (completed) - Safe OIDC discovery/JWKS caching and invalidation guidance (completed) -- Async database evaluation and performance benchmarks +- Async database evaluation and performance benchmarks (completed; sync retained) - Load-testing examples - OpenTelemetry tracing example (completed) diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..db5f0f8 --- /dev/null +++ b/benchmarks/__init__.py @@ -0,0 +1 @@ +"""Opt-in database benchmarks that are not part of the production package.""" diff --git a/benchmarks/db.py b/benchmarks/db.py new file mode 100644 index 0000000..ec2bd7b --- /dev/null +++ b/benchmarks/db.py @@ -0,0 +1,611 @@ +"""Controlled PostgreSQL benchmark for the sync path and an async prototype. + +The benchmark owns only the ``fastapi_benchmark`` schema. It deliberately does +not import the FastAPI application or replace its synchronous database path. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import platform +import statistics +import sys +import threading +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from pathlib import Path +from time import perf_counter +from typing import Any, Literal + +from sqlalchemy import create_engine, event, make_url, text +from sqlalchemy.engine import URL, Engine +from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine + +SCHEMA = "fastapi_benchmark" +SCHEMA_VERSION = 1 +SCENARIOS = ( + "authenticated_read", + "session_list", + "refresh_rotation", + "write", + "mixed", +) +Mode = Literal["sync", "async"] + + +class BenchmarkSafetyError(ValueError): + """Raised when a URL is not clearly an isolated local/test database.""" + + +class RollbackProbe(RuntimeError): + """Internal sentinel used to force an async transaction rollback.""" + + +@dataclass(frozen=True) +class RunConfig: + iterations: int + warmup: int + concurrency: int + pool_size: int + max_overflow: int + + +@dataclass(frozen=True) +class Sample: + latency_ms: float + pool_wait_ms: float + error: str | None = None + + +class PoolObserver: + """Collect bounded pool activity without logging connection details.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self.checkouts = 0 + self.checkins = 0 + self.in_use = 0 + self.peak_in_use = 0 + + def checkout(self, *_: Any) -> None: + with self._lock: + self.checkouts += 1 + self.in_use += 1 + self.peak_in_use = max(self.peak_in_use, self.in_use) + + def checkin(self, *_: Any) -> None: + with self._lock: + self.checkins += 1 + self.in_use = max(0, self.in_use - 1) + + def snapshot(self) -> dict[str, int]: + with self._lock: + return { + "checkouts": self.checkouts, + "checkins": self.checkins, + "peak_in_use": self.peak_in_use, + } + + def reset(self) -> None: + with self._lock: + self.checkouts = 0 + self.checkins = 0 + self.in_use = 0 + self.peak_in_use = 0 + + +def validate_benchmark_url(value: str) -> URL: + """Fail closed unless the target is an obvious local PostgreSQL test DB.""" + url = make_url(value) + if url.get_backend_name() != "postgresql": + raise BenchmarkSafetyError("Benchmarking requires PostgreSQL.") + + host = (url.host or "").lower() + if host not in {"localhost", "127.0.0.1", "::1", "postgres"}: + raise BenchmarkSafetyError( + "Refusing a non-local database host; use an isolated local/CI database." + ) + + database = (url.database or "").lower() + if not any(marker in database for marker in ("test", "bench", "local", "dev")): + raise BenchmarkSafetyError( + "Database name must contain test, bench, local, or dev." + ) + return url + + +def sync_url(value: str) -> URL: + return validate_benchmark_url(value).set(drivername="postgresql+psycopg") + + +def async_url(value: str) -> URL: + return validate_benchmark_url(value).set(drivername="postgresql+asyncpg") + + +def _engine(value: str, *, pool_size: int = 5, max_overflow: int = 0) -> Engine: + return create_engine( + sync_url(value), + pool_pre_ping=True, + pool_size=pool_size, + max_overflow=max_overflow, + pool_timeout=10, + ) + + +def _async_engine( + value: str, *, pool_size: int = 5, max_overflow: int = 0 +) -> AsyncEngine: + return create_async_engine( + async_url(value), + pool_pre_ping=True, + pool_size=pool_size, + max_overflow=max_overflow, + pool_timeout=10, + ) + + +def prepare_fixture(database_url: str, records: int) -> None: + if records < 10: + raise ValueError("records must be at least 10") + engine = _engine(database_url) + try: + with engine.begin() as connection: + connection.execute(text(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}")) + connection.execute( + text( + f""" + CREATE TABLE IF NOT EXISTS {SCHEMA}.principals ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + username VARCHAR(80) NOT NULL UNIQUE, + role VARCHAR(20) NOT NULL, + rotation_counter BIGINT NOT NULL DEFAULT 0, + payload TEXT NOT NULL + ) + """ + ) + ) + connection.execute( + text( + f""" + CREATE TABLE IF NOT EXISTS {SCHEMA}.sessions ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + principal_id BIGINT NOT NULL REFERENCES + {SCHEMA}.principals(id) ON DELETE CASCADE, + created_at TIMESTAMPTZ NOT NULL, + revoked_at TIMESTAMPTZ NULL + ) + """ + ) + ) + connection.execute( + text( + f"CREATE INDEX IF NOT EXISTS ix_benchmark_sessions_principal " + f"ON {SCHEMA}.sessions (principal_id, created_at DESC)" + ) + ) + connection.execute( + text( + f"TRUNCATE {SCHEMA}.sessions, {SCHEMA}.principals RESTART IDENTITY" + ) + ) + principals = [ + { + "username": f"benchmark-user-{index:06d}", + "role": "admin" if index % 20 == 0 else "user", + "payload": "x" * 256, + } + for index in range(records) + ] + connection.execute( + text( + f"INSERT INTO {SCHEMA}.principals (username, role, payload) " + "VALUES (:username, :role, :payload)" + ), + principals, + ) + connection.execute( + text( + f""" + INSERT INTO {SCHEMA}.sessions (principal_id, created_at) + SELECT principal.id, NOW() - (n * INTERVAL '1 minute') + FROM {SCHEMA}.principals AS principal + CROSS JOIN generate_series(1, 3) AS n + """ + ) + ) + connection.execute(text(f"ANALYZE {SCHEMA}.principals")) + connection.execute(text(f"ANALYZE {SCHEMA}.sessions")) + finally: + engine.dispose() + + +def cleanup_fixture(database_url: str) -> None: + engine = _engine(database_url) + try: + with engine.begin() as connection: + connection.execute(text(f"DROP SCHEMA IF EXISTS {SCHEMA} CASCADE")) + finally: + engine.dispose() + + +def _scenario_for(scenario: str, index: int) -> str: + if scenario != "mixed": + return scenario + return ("authenticated_read", "session_list", "refresh_rotation", "write")[ + index % 4 + ] + + +def _statements(scenario: str, index: int) -> list[tuple[str, dict[str, Any]]]: + scenario = _scenario_for(scenario, index) + record_id = (index % 10) + 1 + username = f"benchmark-user-{index % 10:06d}" + if scenario == "authenticated_read": + return [ + ( + f"SELECT id, role, payload FROM {SCHEMA}.principals " + "WHERE username = :username", + {"username": username}, + ) + ] + if scenario == "session_list": + return [ + ( + f"SELECT id, created_at FROM {SCHEMA}.sessions " + "WHERE principal_id = :record_id AND revoked_at IS NULL " + "ORDER BY created_at DESC LIMIT 20", + {"record_id": record_id}, + ) + ] + if scenario == "refresh_rotation": + return [ + ( + f"SELECT rotation_counter FROM {SCHEMA}.principals " + "WHERE id = :record_id FOR UPDATE", + {"record_id": record_id}, + ), + ( + f"UPDATE {SCHEMA}.principals SET rotation_counter = " + "rotation_counter + 1 WHERE id = :record_id", + {"record_id": record_id}, + ), + ] + if scenario == "write": + return [ + ( + f"UPDATE {SCHEMA}.principals SET payload = payload " + "WHERE id = :record_id", + {"record_id": record_id}, + ) + ] + raise ValueError(f"Unknown scenario: {scenario}") + + +def _attach_observer(engine: Engine, observer: PoolObserver) -> None: + event.listen(engine, "checkout", observer.checkout) + event.listen(engine, "checkin", observer.checkin) + + +def _sync_sample(engine: Engine, scenario: str, index: int) -> Sample: + started = perf_counter() + try: + acquire_started = perf_counter() + with engine.begin() as connection: + acquired = perf_counter() + for statement, params in _statements(scenario, index): + connection.execute(text(statement), params) + finished = perf_counter() + return Sample( + latency_ms=(finished - started) * 1000, + pool_wait_ms=(acquired - acquire_started) * 1000, + ) + except Exception as error: # benchmark errors belong in the result artifact + return Sample( + latency_ms=(perf_counter() - started) * 1000, + pool_wait_ms=0, + error=type(error).__name__, + ) + + +async def _async_sample(engine: AsyncEngine, scenario: str, index: int) -> Sample: + started = perf_counter() + try: + acquire_started = perf_counter() + async with engine.begin() as connection: + acquired = perf_counter() + for statement, params in _statements(scenario, index): + await connection.execute(text(statement), params) + finished = perf_counter() + return Sample( + latency_ms=(finished - started) * 1000, + pool_wait_ms=(acquired - acquire_started) * 1000, + ) + except Exception as error: # benchmark errors belong in the result artifact + return Sample( + latency_ms=(perf_counter() - started) * 1000, + pool_wait_ms=0, + error=type(error).__name__, + ) + + +def _percentile(values: list[float], percentile: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + position = (len(ordered) - 1) * percentile + lower = int(position) + upper = min(lower + 1, len(ordered) - 1) + weight = position - lower + return ordered[lower] * (1 - weight) + ordered[upper] * weight + + +def summarize( + *, + mode: Mode, + scenario: str, + samples: list[Sample], + duration: float, + observer: PoolObserver, +) -> dict[str, Any]: + successful = [sample for sample in samples if sample.error is None] + latencies = [sample.latency_ms for sample in successful] + waits = [sample.pool_wait_ms for sample in successful] + error_types: dict[str, int] = {} + for sample in samples: + if sample.error: + error_types[sample.error] = error_types.get(sample.error, 0) + 1 + + def distribution(values: list[float]) -> dict[str, float]: + return { + "minimum": round(min(values, default=0), 3), + "median": round(statistics.median(values) if values else 0, 3), + "p95": round(_percentile(values, 0.95), 3), + "p99": round(_percentile(values, 0.99), 3), + "maximum": round(max(values, default=0), 3), + } + + return { + "mode": mode, + "scenario": scenario, + "attempts": len(samples), + "successes": len(successful), + "errors": len(samples) - len(successful), + "error_rate": round((len(samples) - len(successful)) / len(samples), 6) + if samples + else 0, + "error_types": error_types, + "duration_seconds": round(duration, 6), + "throughput_rps": round(len(successful) / duration, 3) if duration else 0, + "latency_ms": distribution(latencies), + "pool_wait_ms": distribution(waits), + "pool": observer.snapshot(), + } + + +def run_sync(database_url: str, scenario: str, config: RunConfig) -> dict[str, Any]: + engine = _engine( + database_url, + pool_size=config.pool_size, + max_overflow=config.max_overflow, + ) + observer = PoolObserver() + _attach_observer(engine, observer) + try: + for index in range(config.warmup): + _sync_sample(engine, scenario, index) + observer.reset() + started = perf_counter() + with ThreadPoolExecutor(max_workers=config.concurrency) as executor: + futures = [ + executor.submit(_sync_sample, engine, scenario, index) + for index in range(config.iterations) + ] + samples = [future.result() for future in as_completed(futures)] + duration = perf_counter() - started + return summarize( + mode="sync", + scenario=scenario, + samples=samples, + duration=duration, + observer=observer, + ) + finally: + engine.dispose() + + +async def run_async( + database_url: str, scenario: str, config: RunConfig +) -> dict[str, Any]: + engine = _async_engine( + database_url, + pool_size=config.pool_size, + max_overflow=config.max_overflow, + ) + observer = PoolObserver() + _attach_observer(engine.sync_engine, observer) + semaphore = asyncio.Semaphore(config.concurrency) + + async def bounded(index: int) -> Sample: + async with semaphore: + return await _async_sample(engine, scenario, index) + + try: + for index in range(config.warmup): + await bounded(index) + observer.reset() + started = perf_counter() + samples = await asyncio.gather( + *(bounded(index) for index in range(config.iterations)) + ) + duration = perf_counter() - started + return summarize( + mode="async", + scenario=scenario, + samples=samples, + duration=duration, + observer=observer, + ) + finally: + await engine.dispose() + + +async def verify_async_rollback(database_url: str) -> bool: + """Verify that the isolated async prototype preserves rollback semantics.""" + engine = _async_engine(database_url, pool_size=1) + try: + async with engine.connect() as connection: + before = await connection.scalar( + text(f"SELECT rotation_counter FROM {SCHEMA}.principals WHERE id = 1") + ) + try: + async with engine.begin() as connection: + await connection.execute( + text( + f"UPDATE {SCHEMA}.principals SET rotation_counter = " + "rotation_counter + 1 WHERE id = 1" + ) + ) + raise RollbackProbe("force rollback") + except RollbackProbe: + pass + async with engine.connect() as connection: + after = await connection.scalar( + text(f"SELECT rotation_counter FROM {SCHEMA}.principals WHERE id = 1") + ) + return before == after + finally: + await engine.dispose() + + +def build_report( + database_url: str, + modes: list[Mode], + scenarios: list[str], + config: RunConfig, +) -> dict[str, Any]: + validated = validate_benchmark_url(database_url) + results = [] + for scenario in scenarios: + if "sync" in modes: + results.append(run_sync(database_url, scenario, config)) + if "async" in modes: + results.append(asyncio.run(run_async(database_url, scenario, config))) + return { + "schema_version": SCHEMA_VERSION, + "generated_at": datetime.now(UTC).isoformat(), + "environment": { + "python": platform.python_version(), + "platform": platform.platform(), + "database_host": validated.host, + "database_name": validated.database, + }, + "config": asdict(config), + "results": results, + } + + +def _database_url(args: argparse.Namespace) -> str: + value = args.database_url or os.environ.get("BENCHMARK_DATABASE_URL") + if not value: + raise BenchmarkSafetyError( + "Set BENCHMARK_DATABASE_URL or pass --database-url; DATABASE_URL is " + "intentionally not used implicitly." + ) + validate_benchmark_url(value) + return value + + +def _add_database_argument(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--database-url", + help="Isolated local/CI PostgreSQL URL (or set BENCHMARK_DATABASE_URL).", + ) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + prepare = subparsers.add_parser("prepare", help="Create deterministic fixtures.") + _add_database_argument(prepare) + prepare.add_argument("--records", type=int, default=1_000) + + cleanup = subparsers.add_parser("cleanup", help="Drop only the benchmark schema.") + _add_database_argument(cleanup) + + rollback = subparsers.add_parser( + "verify-async-rollback", help="Check async transaction rollback semantics." + ) + _add_database_argument(rollback) + + run = subparsers.add_parser("run", help="Run controlled sync/async comparisons.") + _add_database_argument(run) + run.add_argument("--mode", choices=("sync", "async", "both"), default="both") + run.add_argument( + "--scenarios", nargs="+", choices=SCENARIOS, default=list(SCENARIOS) + ) + run.add_argument("--iterations", type=int, default=1_000) + run.add_argument("--warmup", type=int, default=100) + run.add_argument("--concurrency", type=int, default=10) + run.add_argument("--pool-size", type=int, default=5) + run.add_argument("--max-overflow", type=int, default=0) + run.add_argument("--output", type=Path, default=Path("benchmark-results.json")) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + database_url = _database_url(args) + if args.command == "prepare": + prepare_fixture(database_url, args.records) + return 0 + if args.command == "cleanup": + cleanup_fixture(database_url) + return 0 + if args.command == "verify-async-rollback": + if not asyncio.run(verify_async_rollback(database_url)): + print("Async rollback probe failed.", file=sys.stderr) + return 1 + print("Async rollback probe passed.") + return 0 + + values = ( + args.iterations, + args.concurrency, + args.pool_size, + ) + if ( + any(value < 1 for value in values) + or args.warmup < 0 + or args.max_overflow < 0 + ): + raise ValueError("iterations/concurrency/pool-size must be positive") + modes: list[Mode] = ["sync", "async"] if args.mode == "both" else [args.mode] + report = build_report( + database_url, + modes, + args.scenarios, + RunConfig( + iterations=args.iterations, + warmup=args.warmup, + concurrency=args.concurrency, + pool_size=args.pool_size, + max_overflow=args.max_overflow, + ), + ) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + print(json.dumps(report, indent=2)) + return 1 if any(result["errors"] for result in report["results"]) else 0 + except (BenchmarkSafetyError, ValueError) as error: + print(f"Error: {error}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/decisions/0001-keep-sync-sqlalchemy.md b/docs/decisions/0001-keep-sync-sqlalchemy.md new file mode 100644 index 0000000..1e89efc --- /dev/null +++ b/docs/decisions/0001-keep-sync-sqlalchemy.md @@ -0,0 +1,36 @@ +# ADR 0001: Keep synchronous SQLAlchemy for v1.3.0 + +- Status: accepted +- Date: 2026-08-11 +- Decision owners: maintainers + +## Context + +The application uses synchronous SQLAlchemy and psycopg with explicit request +sessions and transaction rollback. Async SQLAlchemy may improve concurrency for +some database-wait-heavy workloads, but a full conversion would affect routers, +services, repositories, tests, tracing, workers, and operational debugging. + +## Decision + +Keep the production path synchronous for v1.3.0. Maintain the asyncpg prototype +only in the opt-in benchmark harness. Do not expose an async session dependency +to application code and do not change API, authentication, outbox, or migration +semantics. + +## Rationale + +The repository has no representative extended measurement demonstrating that a +migration would produce a meaningful SLO improvement. CI smoke timing is noisy +and is used only for correctness. The simplest architecture satisfying current +evidence is therefore the existing synchronous path. + +## Consequences + +- Production behavior and operational knowledge remain stable. +- Sync and async can be compared with equivalent SQL, fixtures, pools, and + transaction boundaries using `DATABASE_BENCHMARKS.md`. +- Teams must run and retain representative extended results before reopening + the decision. +- A full async migration, if justified later, requires a separate scoped issue. + diff --git a/pyproject.toml b/pyproject.toml index 77f9e8b..0a8d2a3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,6 +74,7 @@ module-name = ["app", "fastapi_production_api"] [dependency-groups] dev = [ + "asyncpg>=0.31.0,<1", "httpx2>=2.7.0", "pip-audit>=2.9.0", "pytest>=9.1.1", @@ -83,6 +84,7 @@ dev = [ [tool.pytest.ini_options] testpaths = ["tests"] +pythonpath = ["."] addopts = "-ra --cov --cov-report=term-missing --cov-fail-under=90" [tool.coverage.run] diff --git a/tests/test_database_benchmark.py b/tests/test_database_benchmark.py new file mode 100644 index 0000000..06f12bf --- /dev/null +++ b/tests/test_database_benchmark.py @@ -0,0 +1,107 @@ +import asyncio + +import pytest + +from app.core.config import settings +from benchmarks.db import ( + BenchmarkSafetyError, + PoolObserver, + RunConfig, + Sample, + async_url, + cleanup_fixture, + prepare_fixture, + run_async, + run_sync, + summarize, + sync_url, + validate_benchmark_url, + verify_async_rollback, +) + + +def test_benchmark_url_is_fail_closed(): + with pytest.raises(BenchmarkSafetyError, match="PostgreSQL"): + validate_benchmark_url("sqlite:///benchmark.db") + with pytest.raises(BenchmarkSafetyError, match="non-local"): + validate_benchmark_url( + "postgresql://user:password@database.example.com/app_test" + ) + with pytest.raises(BenchmarkSafetyError, match="Database name"): + validate_benchmark_url("postgresql://user:password@localhost/production") + + +def test_benchmark_url_selects_explicit_drivers_without_exposing_credentials(): + value = "postgresql://user:password@localhost/app_test" + + assert sync_url(value).drivername == "postgresql+psycopg" + assert async_url(value).drivername == "postgresql+asyncpg" + + +def test_summary_schema_is_stable_and_counts_errors(): + observer = PoolObserver() + observer.checkout() + observer.checkin() + + result = summarize( + mode="sync", + scenario="authenticated_read", + samples=[ + Sample(latency_ms=10, pool_wait_ms=1), + Sample(latency_ms=20, pool_wait_ms=2), + Sample(latency_ms=30, pool_wait_ms=0, error="DatabaseError"), + ], + duration=1, + observer=observer, + ) + + assert result == { + "mode": "sync", + "scenario": "authenticated_read", + "attempts": 3, + "successes": 2, + "errors": 1, + "error_rate": 0.333333, + "error_types": {"DatabaseError": 1}, + "duration_seconds": 1, + "throughput_rps": 2.0, + "latency_ms": { + "minimum": 10, + "median": 15.0, + "p95": 19.5, + "p99": 19.9, + "maximum": 20, + }, + "pool_wait_ms": { + "minimum": 1, + "median": 1.5, + "p95": 1.95, + "p99": 1.99, + "maximum": 2, + }, + "pool": {"checkouts": 1, "checkins": 1, "peak_in_use": 1}, + } + + +def test_sync_and_async_prototypes_run_against_isolated_postgres(): + database_url = settings.DATABASE_URL + config = RunConfig( + iterations=8, + warmup=2, + concurrency=2, + pool_size=2, + max_overflow=0, + ) + prepare_fixture(database_url, records=20) + try: + sync_result = run_sync(database_url, "authenticated_read", config) + async_result = asyncio.run( + run_async(database_url, "authenticated_read", config) + ) + + assert sync_result["successes"] == config.iterations + assert async_result["successes"] == config.iterations + assert sync_result["errors"] == async_result["errors"] == 0 + assert asyncio.run(verify_async_rollback(database_url)) is True + finally: + cleanup_fixture(database_url) diff --git a/tests/test_documentation.py b/tests/test_documentation.py index e3af8f6..3ba23ae 100644 --- a/tests/test_documentation.py +++ b/tests/test_documentation.py @@ -15,6 +15,7 @@ "CHANGELOG.md", "CODE_OF_CONDUCT.md", "CONTRIBUTING.md", + "DATABASE_BENCHMARKS.md", "DEPLOYMENT.md", "DEVELOPMENT.md", "MONITORING.md", @@ -24,6 +25,7 @@ "SECURITY.md", ) ] +DOCUMENTS.append(PROJECT_ROOT / "docs" / "decisions" / "0001-keep-sync-sqlalchemy.md") MARKDOWN_LINK = re.compile(r"(?=0.31.0,<1" }, { name = "httpx2", specifier = ">=2.7.0" }, { name = "pip-audit", specifier = ">=2.9.0" }, { name = "pytest", specifier = ">=9.1.1" },