Merge pull request #30 from listlessbird/perf-stuff

Perf stuff

Author: listlessbird

backend-api/.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "workbench.colorCustomizations": {
+        "statusBar.background": "#335c99",
+        "statusBar.foreground": "#ffffff",
+        "sash.hoverBorder": "#335c99"
+    }
+}

backend-api/Dockerfile

@@ -54,7 +54,7 @@ USER appuser
 EXPOSE 8000
 
 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
-    CMD curl -f http://localhost:8000/live || exit 1
+    CMD curl -f http://localhost:8000/ready || exit 1
 
 CMD ["sh", "-c", "alembic upgrade head && uvicorn api.main:app --host 0.0.0.0 --port 8000"]
 

backend-api/README.md

@@ -96,7 +96,7 @@ The worker must be running for ingest and rebuild jobs to execute.
 
 - API: `http://localhost:8000`
 - API docs: `http://localhost:8000/docs`
-- Health check: `http://localhost:8000/live`
+- Health check: `http://localhost:8000/ready`
 - Temporal UI: `http://localhost:8233`
 - MinIO console: `http://localhost:9001`
 
@@ -111,7 +111,7 @@ minioadmin / minioadmin
 Check service health:
 
 ```bash
-curl http://localhost:8000/live
+curl http://localhost:8000/ready
 ```
 
 Queue one image for ingestion:

backend-api/docker-compose.yml

@@ -110,7 +110,7 @@ services:
       minio:
         condition: service_healthy
     healthcheck:
-      test: [ "CMD", "curl", "-f", "http://localhost:8000/live" ]
+      test: [ "CMD", "curl", "-f", "http://localhost:8000/ready" ]
       interval: 30s
       timeout: 10s
       retries: 3

backend-api/docker.compose.prod.yml

@@ -31,7 +31,7 @@ services:
           memory: 2048M
     restart: unless-stopped
     healthcheck:
-      test: [ "CMD", "curl", "-f", "http://localhost:8000/live" ]
+      test: [ "CMD", "curl", "-f", "http://localhost:8000/ready" ]
       interval: 30s
       timeout: 10s
       retries: 3

backend-api/pyproject.toml

@@ -24,6 +24,8 @@ dependencies = [
     "modal==1.3.4",
     "slowapi==0.1.9",
     "axiom-py==0.10.0",
+    "asyncpg>=0.31.0",
+    "aiobotocore>=3.7.0",
 ]
 
 # Extras only needed to run models locally (GPU_BACKEND=local) or to import
@@ -68,7 +70,11 @@ dev = [
     { include-group = "hooks" },
 ]
 lint = ["ruff==0.15.2"]
-typecheck = ["pyright"]
+typecheck = [
+    "pyright",
+    "types-aiobotocore>=3.7.0",
+    "types-aiobotocore-s3>=3.7.0",
+]
 test = [
     "pytest==9.0.2",
     "pytest-asyncio==1.3.0",
@@ -131,6 +137,17 @@ executionEnvironments = [
     ] },
 ]
 typeCheckingMode = "basic"
+strict = [
+    "src/shared/db.py",
+    "src/shared/services/api_storage.py",
+    "src/domain/job_rules.py",
+    "src/domain/job_store.py",
+    "src/domain/image_catalog.py",
+    "src/domain/ingestion_browse.py",
+    "src/domain/source_item_browse.py",
+    "src/domain/source_registry.py",
+    "src/domain/source_retry.py",
+]
 useLibraryCodeForTypes = true
 reportMissingTypeStubs = false
 reportPrivateImportUsage = false

backend-api/scripts/loadprobe.sh

@@ -0,0 +1,116 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+DB_URL="${DB_URL:-postgresql://postgres:postgres@localhost:5432/mimeme}"
+PG_CONTAINER="${PG_CONTAINER:-mimeme-postgres}"
+API_PORT="${API_PORT:-8000}"
+DURATION="${DURATION:-30s}"
+CONCURRENCY="${CONCURRENCY:-200}"
+OVERLOAD_CONCURRENCY="${OVERLOAD_CONCURRENCY:-400}"
+SEED_COUNT="${SEED_COUNT:-5000}"
+OHA_BIN="${OHA_BIN:-oha}"
+OUT_DIR="${OUT_DIR:-data/loadprobe}"
+OVERLOAD_CPUS="${OVERLOAD_CPUS:-0.1}"
+
+TARGET="http://127.0.0.1:${API_PORT}/images?limit=20"
+API_LOG="${OUT_DIR}/api.log"
+
+mkdir -p "$OUT_DIR"
+
+if ! docker ps --format '{{.Names}}' | grep -qx "$PG_CONTAINER"; then
+  docker compose up -d postgres
+  sleep 3
+fi
+
+export DB_URL
+SEED_COUNT="$SEED_COUNT" uv run python - <<'PY'
+import os
+
+from sqlalchemy import create_engine, func, select
+from sqlalchemy.orm import Session
+
+from shared.config import settings
+from shared.models.orm import Base, Image
+
+seed_count = int(os.environ["SEED_COUNT"])
+engine = create_engine(settings.db_url_str)
+Base.metadata.create_all(engine)
+with Session(engine) as session:
+    existing = session.scalar(select(func.count()).select_from(Image)) or 0
+    for i in range(existing, seed_count):
+        session.add(
+            Image(
+                sha256=f"loadprobe-{i:056d}",
+                dataset="loadprobe",
+                s3_key=f"images/loadprobe/{i}.jpg",
+                width=640,
+                height=480,
+                format="jpeg",
+                file_size=12345,
+            )
+        )
+    session.commit()
+    total = session.scalar(select(func.count()).select_from(Image)) or 0
+print(f"images in db: {total}")
+PY
+
+APP_ENV=development DEBUG=false LOG_LEVEL=INFO \
+RATE_LIMIT_ENABLED=false PRELOAD_TEXT_ENCODER_ON_STARTUP=false GPU_BACKEND=local \
+AXIOM_API_TOKEN= AXIOM_DATASET= S3_ENDPOINT_URL=http://127.0.0.1:9 \
+uv run uvicorn api.main:app --host 127.0.0.1 --port "$API_PORT" --log-level warning \
+  >"$API_LOG" 2>&1 &
+API_PID=$!
+trap 'kill "$API_PID" 2>/dev/null || true; docker update --cpus 0 "$PG_CONTAINER" >/dev/null 2>&1 || true' EXIT
+
+for _ in $(seq 1 60); do
+  curl -fsS -o /dev/null "$TARGET" 2>/dev/null && break
+  sleep 1
+done
+curl -fsS -o /dev/null "$TARGET"
+
+echo "== warmup, c=50 for 5s"
+"$OHA_BIN" -z 5s -c 50 --no-tui --output-format quiet "$TARGET" >/dev/null
+LAG_AFTER_WARMUP=$(grep -c event_loop_lag "$API_LOG" || true)
+
+echo "== phase (a): normal load, c=${CONCURRENCY} for ${DURATION}"
+"$OHA_BIN" -z "$DURATION" -c "$CONCURRENCY" --no-tui --output-format json "$TARGET" >"$OUT_DIR/normal.json"
+LAG_AFTER_NORMAL=$(grep -c event_loop_lag "$API_LOG" || true)
+
+echo "== phase (b): overload, c=${OVERLOAD_CONCURRENCY}, postgres constrained to ${OVERLOAD_CPUS} cpus"
+docker update --cpus "$OVERLOAD_CPUS" "$PG_CONTAINER" >/dev/null
+"$OHA_BIN" -z "$DURATION" -c "$OVERLOAD_CONCURRENCY" --no-tui --output-format json "$TARGET" >"$OUT_DIR/overload.json"
+docker update --cpus 0 "$PG_CONTAINER" >/dev/null
+
+kill "$API_PID"
+wait "$API_PID" 2>/dev/null || true
+
+LAG_TOTAL=$(grep -c event_loop_lag "$API_LOG" || true)
+TRACEBACKS=$(grep -c Traceback "$API_LOG" || true)
+
+LAG_AFTER_WARMUP="$LAG_AFTER_WARMUP" \
+LAG_AFTER_NORMAL="$LAG_AFTER_NORMAL" LAG_TOTAL="$LAG_TOTAL" TRACEBACKS="$TRACEBACKS" \
+OUT_DIR="$OUT_DIR" uv run python - <<'PY'
+import json
+import os
+
+out_dir = os.environ["OUT_DIR"]
+for phase in ("normal", "overload"):
+    with open(f"{out_dir}/{phase}.json") as f:
+        report = json.load(f)
+    summary = report["summary"]
+    percentiles = report["latencyPercentiles"]
+    print(f"-- {phase}")
+    print(f"   rps={summary['requestsPerSec']:.1f} total={report['statusCodeDistribution']}")
+    for key in ("p50", "p90", "p99"):
+        print(f"   {key}={percentiles[key] * 1000:.1f}ms")
+    print(f"   errors={report['errorDistribution']}")
+warmup = int(os.environ["LAG_AFTER_WARMUP"])
+after_normal = int(os.environ["LAG_AFTER_NORMAL"])
+total = int(os.environ["LAG_TOTAL"])
+print(f"-- loop lag events: warmup={warmup} normal={after_normal - warmup} overload={total - after_normal}")
+print(f"-- tracebacks in api log: {os.environ['TRACEBACKS']}")
+PY
+
+echo "raw reports in ${OUT_DIR}/, api log in ${API_LOG}"

backend-api/src/activities/indexing/activities.py

@@ -7,6 +7,7 @@
 import numpy as np
 import structlog
 from botocore.exceptions import ClientError
+from sqlalchemy import select
 from temporalio import activity
 
 from activities.indexing.faiss_manager import FaissIndexManager
@@ -40,14 +41,12 @@ def build_index_activity(input: BuildIndexInput) -> BuildIndexOutput:
     index_manager = FaissIndexManager.get_instance()
     try:
         with session_scope() as session:
-            done_procs = (
-                session.query(Processing.image_id, Processing.embed_s3_key)
-                .filter(
+            done_procs = session.execute(
+                select(Processing.image_id, Processing.embed_s3_key).where(
                     Processing.embed_status == ProcessingStatus.DONE,
                     Processing.embed_s3_key.isnot(None),
                 )
-                .all()
-            )
+            ).all()
 
         total_candidates = len(done_procs)
         candidates = [

backend-api/src/activities/indexing/index_catalog.py

@@ -3,6 +3,7 @@
 from collections.abc import Sequence
 from typing import Any, Protocol
 
+from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 
 from shared.models import IndexBuild
@@ -20,10 +21,10 @@ def delete_cached_version(self, version: str) -> None: ...
 
 class ActiveIndexCatalog:
     def active_build(self, db: Session) -> IndexBuild | None:
-        return db.query(IndexBuild).filter(IndexBuild.is_active).first()
+        return db.scalars(select(IndexBuild).where(IndexBuild.is_active.is_(True))).first()
 
     def list_builds_newest_first(self, db: Session) -> Sequence[IndexBuild]:
-        return db.query(IndexBuild).order_by(IndexBuild.created_at.desc()).all()
+        return db.scalars(select(IndexBuild).order_by(IndexBuild.created_at.desc())).all()
 
     def add_inactive_build(
         self,
@@ -57,12 +58,15 @@ def mark_latest_available(
         metadata: dict[str, Any],
         artifacts: StoredIndexArtifacts,
     ) -> None:
-        db.query(IndexBuild).filter(
-            IndexBuild.is_active,
-            IndexBuild.version != version,
-        ).update({"is_active": False})
 
-        build = db.query(IndexBuild).filter(IndexBuild.version == version).first()
+        db.execute(
+            update(IndexBuild)
+            .where(IndexBuild.is_active.is_(True), IndexBuild.version != version)
+            .values(is_active=False)
+        )
+
+        build = db.scalars(select(IndexBuild).where(IndexBuild.version == version)).first()
+
         if build is None:
             db.add(
                 IndexBuild(
@@ -91,8 +95,9 @@ def mark_latest_available(
         db.commit()
 
     def swap_to_version(self, version: str, db: Session) -> None:
-        db.query(IndexBuild).filter(IndexBuild.is_active).update({"is_active": False})
-        db.query(IndexBuild).filter(IndexBuild.version == version).update({"is_active": True})
+        db.execute(update(IndexBuild).where(IndexBuild.is_active.is_(True)).values(is_active=False))
+        db.execute(update(IndexBuild).where(IndexBuild.version == version).values(is_active=True))
+
         db.commit()
 
     def garbage_collect(