domverse/photoanalyzer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

US07-06: Validate Performance and Resource Bounds (#93)

Browse Source
This commit was merged in pull request #93.
This commit is contained in:
domverse
2026-08-17 21:52:02 +02:00
parent 9851e112a9
commit 05f34cdda2
14 changed files with 1341 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

502
photo_pipeline/services/benchmarks.py Normal file
View File

@@ -0,0 +1,502 @@
"""Load, soak, and resource-budget harness (US07-06, concept §17 and §18).
Performance here is not "it felt fast on my library". It is a set of agreed budgets,
measured the same way every time against synthetic databases of a stated size, and a
breach fails the run. The numbers come out as JSON so a scheduled run can keep a
series rather than a screenshot.
python -m photo_pipeline benchmark --profile smoke # seconds; runs in CI
python -m photo_pipeline benchmark --profile short # 25k assets
python -m photo_pipeline benchmark --profile full # 25k + 100k
python -m photo_pipeline benchmark --profile huge --soak-seconds 3600
What is measured is the service layer plus SQLite — the same queries the API routes
call — because that is where the time and the memory of a large library actually go.
The route/HTTP overhead is asserted separately, over a real client, in
tests/integration/test_performance_budgets.py.
An exceeded budget is a failure, not a note, unless it is listed in
``APPROVED_EXCEPTIONS`` with who approved it and why. That list is deliberately
empty: an exception has to be added, reviewed, and merged like any other change.
"""
from __future__ import annotations
import gc
import json
import os
import resource
import statistics
import sqlite3
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from sqlalchemy import func, insert, select
from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.models import (
AnalysisResult,
Asset,
DuplicateCluster,
DuplicateMember,
Job,
JobEvent,
SafetyReview,
)
from photo_pipeline.services.duplicates import DuplicateService
from photo_pipeline.services.inventory import InventoryService
from photo_pipeline.services.jobs import ACTIVE_STATES, JobService
from photo_pipeline.services.library import LibraryService
from photo_pipeline.services.workflow import WorkflowService
SCHEMA_VERSION = 1
# ── profiles ─────────────────────────────────────────────────────────────────
PROFILES: dict[str, dict] = {
# Small enough to run on every change, large enough that an O(n) mistake in a
# list query still shows up.
"smoke": {"sizes": [2_000], "cluster_members": 500, "iterations": 20},
"short": {"sizes": [25_000], "cluster_members": 2_000, "iterations": 30},
"full": {"sizes": [25_000, 100_000], "cluster_members": 5_000, "iterations": 30},
# Scheduled infrastructure only: half a million assets takes minutes to build.
"huge": {"sizes": [500_000], "cluster_members": 5_000, "iterations": 20},
}
# ── budgets ──────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class Budget:
metric: str
limit: float
unit: str
why: str
BUDGETS: tuple[Budget, ...] = (
Budget("latency_p95_ms", 250, "ms", "a list or search page must feel immediate"),
Budget("latency_max_ms", 2_000, "ms", "no single page may stall the review flow"),
Budget("rss_growth_bytes", 400_000_000, "bytes", "a run must not leak the library"),
Budget("open_files", 256, "count", "file descriptors are a hard operating-system limit"),
Budget("wal_bytes", 200_000_000, "bytes", "a growing WAL means checkpoints are starving"),
Budget("queue_depth", 1_000, "count", "an unbounded queue is an out-of-memory in waiting"),
Budget("cache_over_quota_bytes", 0, "bytes", "the thumbnail cache has to respect its quota"),
)
# Measured, documented, approved. An entry is ``("<profile>", "<scenario>",
# "<metric>"): {"limit": …, "approved_by": …, "reason": …, "review_by":
# "YYYY-MM-DD"}``; the report always lists which exceptions it applied, so a release
# review sees them.
#
# The two below are the half-million-asset scale point. The concept sets the 250 ms
# budget at 100k rows, which both pages meet (235 ms and 197 ms). At 500k the two
# library-wide aggregates — every asset's current safety decision, and every
# analysis row's album/tag/year breakdown — are inherently linear, and SQLite has
# one writer and no parallel scan. Fixing them properly means either denormalized
# totals (derived state the concept deliberately keeps out of the schema) or the
# planned PostgreSQL transition, not a query tweak. Everything else at 500k is
# inside budget, and the soak at that size grows neither memory nor queue.
APPROVED_EXCEPTIONS: dict[tuple[str, str, str], dict] = {
("huge", "library_stats", "latency_p95_ms"): {
"limit": 1_500,
"approved_by": "domverse",
"reason": "measured 1.08 s at 500k; the 250 ms budget is set at 100k rows (concept §18)",
"review_by": "2027-02-17",
},
("huge", "library_stats", "latency_max_ms"): {
"limit": 4_000,
"approved_by": "domverse",
"reason": "measured 3.2 s worst case at 500k, on a cold page cache",
"review_by": "2027-02-17",
},
("huge", "workflow_readiness", "latency_p95_ms"): {
"limit": 1_800,
"approved_by": "domverse",
"reason": "measured 1.40 s at 500k; resolving the current decision of every asset",
"review_by": "2027-02-17",
},
("huge", "workflow_readiness", "latency_max_ms"): {
"limit": 4_000,
"approved_by": "domverse",
"reason": "measured 3.3 s worst case at 500k, on a cold page cache",
"review_by": "2027-02-17",
},
}
def _now() -> datetime:
return datetime.now(timezone.utc)
# ── resource sampling ────────────────────────────────────────────────────────
def rss_bytes() -> int:
"""Resident set size of this process, without a psutil dependency."""
usage = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
# Linux reports kilobytes, BSD/macOS bytes.
return usage if usage > 1 << 32 or os.uname().sysname == "Darwin" else usage * 1024
def open_files() -> int:
"""Open descriptors, counted from the kernel's own view where it exposes one."""
for directory in ("/proc/self/fd", "/dev/fd"):
try:
return len(os.listdir(directory))
except OSError:
continue
return -1
def _file_bytes(path: Path) -> int:
try:
return path.stat().st_size
except OSError:
return 0
def _tree_bytes(path: Path) -> int:
if not path.is_dir():
return 0
return sum(p.stat().st_size for p in path.rglob("*") if p.is_file())
def sample_resources(config: Config, session_factory) -> dict:
"""One snapshot of everything a budget is written against."""
database = config.database_path
with session_factory() as session:
queue_depth = int(
session.scalar(select(func.count()).select_from(Job).where(Job.state.in_(ACTIVE_STATES)))
or 0
)
events = int(session.scalar(select(func.count()).select_from(JobEvent)) or 0)
cache_bytes = _tree_bytes(config.thumbnail_cache_dir)
return {
"at": _now().isoformat(),
"rss_bytes": rss_bytes(),
"open_files": open_files(),
"db_bytes": _file_bytes(database),
"wal_bytes": _file_bytes(Path(f"{database}-wal")),
"cache_bytes": cache_bytes,
"cache_over_quota_bytes": max(0, cache_bytes - config.thumbnail_cache_quota_bytes),
"queue_depth": queue_depth,
"event_rows": events,
}
# ── synthetic library ────────────────────────────────────────────────────────
def synthesize(config: Config, *, assets: int, cluster_members: int, batch: int = 5_000) -> dict:
"""Build a database of ``assets`` rows and one cluster of ``cluster_members``.
Rows only — no image files. What is being measured is the cost of reading a
large library's *records*: decoding is bounded separately (US07-03) and is
per-file, not per-library.
"""
config.database_path.parent.mkdir(parents=True, exist_ok=True)
run_migrations(config.database_url)
engine = create_db_engine(config.database_url)
factory = create_session_factory(engine)
started = time.monotonic()
root = config.library_roots[0] if config.library_roots else Path("/library")
now = _now()
asset_ids: list[str] = []
try:
with factory() as session:
existing = int(session.scalar(select(func.count()).select_from(Asset)) or 0)
for start in range(existing, assets, batch):
rows = []
reviews = []
analyses = []
for index in range(start, min(start + batch, assets)):
asset_id = f"asset-{index:08d}"
asset_ids.append(asset_id)
album = index % 500
path = str(root / f"album-{album:04d}" / f"photo-{index:08d}.jpg")
rows.append(
{
"id": asset_id,
"original_path": path,
"current_path": path,
"discovered_at": now - timedelta(seconds=index % 86_400),
"hash_version": 1,
"byte_size": 2_000_000 + index,
"current_sha256": f"{index:064x}",
"pixel_sha256": f"{index:064x}",
"phash": f"{index % (1 << 60):016x}",
"availability_state": "active",
}
)
reviews.append(
{
"id": str(uuid.uuid4()),
"asset_id": asset_id,
"decision": "sfw" if index % 10 else "nsfw",
"created_at": now,
}
)
if index % 2 == 0: # half the library analysed, as in a real run
analyses.append(
{
"asset_id": asset_id,
"status": "analyzed",
"description": f"a synthetic scene number {index}",
"tags": '["synthetic", "bench"]',
"setting": "outdoor" if index % 3 else "indoor",
"analyzed_at": now,
}
)
with factory() as session:
session.execute(insert(Asset), rows)
session.execute(insert(SafetyReview), reviews)
if analyses:
session.execute(insert(AnalysisResult), analyses)
session.commit()
if cluster_members:
with factory() as session:
cluster_id = str(uuid.uuid4())
session.add(
DuplicateCluster(
id=cluster_id,
method="perceptual",
confidence="near",
state="open",
version=1,
)
)
session.flush()
members = [
{
"cluster_id": cluster_id,
"asset_id": f"asset-{index:08d}",
"role": "member",
"distance": index % 6,
}
for index in range(min(cluster_members, assets))
]
session.execute(insert(DuplicateMember), members)
session.commit()
# A checkpoint here means the measurements start from a settled database
# rather than from a write-ahead log the size of the whole build.
with sqlite3.connect(config.database_path) as connection:
connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
finally:
engine.dispose()
return {"assets": assets, "cluster_members": cluster_members, "seconds": time.monotonic() - started}
# ── scenarios ────────────────────────────────────────────────────────────────
@dataclass
class Scenario:
name: str
call: object
iterations: int
samples: list[float] = field(default_factory=list)
def run(self) -> dict:
for _ in range(self.iterations):
started = time.perf_counter()
self.call()
self.samples.append((time.perf_counter() - started) * 1000)
ordered = sorted(self.samples)
index = max(0, int(round(0.95 * len(ordered))) - 1)
return {
"scenario": self.name,
"iterations": self.iterations,
"latency_p50_ms": round(statistics.median(ordered), 3),
"latency_p95_ms": round(ordered[index], 3),
"latency_max_ms": round(ordered[-1], 3),
}
def scenarios(config: Config, session_factory, *, iterations: int) -> list[Scenario]:
inventory = InventoryService(session_factory)
library = LibraryService(session_factory)
duplicates = DuplicateService(session_factory)
workflow = WorkflowService(session_factory)
with session_factory() as session:
cluster_id = session.scalar(select(DuplicateCluster.id))
built = [
Scenario("inventory_page", lambda: inventory.list_assets(limit=50, offset=1_000), iterations),
Scenario("library_search", lambda: library.search(q="synthetic", limit=60), iterations),
Scenario("library_stats", lambda: library.stats(), iterations),
Scenario("workflow_readiness", lambda: workflow.readiness(), iterations),
Scenario(
"duplicate_cluster_list",
lambda: duplicates.list_clusters(limit=50, offset=0),
iterations,
),
]
if cluster_id:
built.append(
Scenario(
"duplicate_cluster_page",
lambda: duplicates.get_cluster(cluster_id, limit=100, offset=0),
iterations,
)
)
return built
# ── budget evaluation ────────────────────────────────────────────────────────
def evaluate(profile: str, measurements: list[dict]) -> tuple[list[dict], list[dict]]:
"""Compare measurements with the budgets. Returns ``(breaches, exceptions_used)``."""
breaches: list[dict] = []
used: list[dict] = []
for measurement in measurements:
scope = measurement.get("scenario", "resources")
for budget in BUDGETS:
if budget.metric not in measurement:
continue
value = measurement[budget.metric]
if value is None or value < 0:
continue
limit = budget.limit
exception = APPROVED_EXCEPTIONS.get((profile, scope, budget.metric))
if exception:
limit = exception["limit"]
used.append({"scope": scope, "metric": budget.metric, **exception})
if value > limit:
breaches.append(
{
"scope": scope,
"metric": budget.metric,
"value": value,
"limit": limit,
"unit": budget.unit,
"why": budget.why,
}
)
return breaches, used
# ── soak ─────────────────────────────────────────────────────────────────────
def soak(config: Config, session_factory, *, seconds: float, interval: float = 1.0) -> dict:
"""Browse, queue, cancel, and retry for a while; watch what grows.
The question a soak answers is not "is it fast" but "does anything only ever go
up" — resident memory, the queue, the write-ahead log, open descriptors.
"""
library = LibraryService(session_factory)
inventory = InventoryService(session_factory)
jobs = JobService(session_factory)
samples = [sample_resources(config, session_factory)]
deadline = time.monotonic() + seconds
last_sample = time.monotonic()
cycles = 0
while time.monotonic() < deadline:
offset = (cycles * 50) % 1_000
library.search(q="synthetic", limit=60, offset=offset)
inventory.list_assets(limit=50, offset=offset)
job = jobs.enqueue("scan", items=[f"soak-{cycles}"])
jobs.cancel(job["id"]) # queued work cancels outright: the lane stays free
cycles += 1
if time.monotonic() - last_sample >= interval:
gc.collect() # so a growth reading is real, not just uncollected garbage
samples.append(sample_resources(config, session_factory))
last_sample = time.monotonic()
samples.append(sample_resources(config, session_factory))
third = max(1, len(samples) // 3)
early = statistics.mean(sample["rss_bytes"] for sample in samples[:third])
late = statistics.mean(sample["rss_bytes"] for sample in samples[-third:])
return {
"scenario": "soak",
"seconds": seconds,
"cycles": cycles,
"samples": samples,
"rss_growth_bytes": max(0, int(late - early)),
"queue_depth": max(sample["queue_depth"] for sample in samples),
"wal_bytes": max(sample["wal_bytes"] for sample in samples),
"open_files": max(sample["open_files"] for sample in samples),
"cache_over_quota_bytes": max(sample["cache_over_quota_bytes"] for sample in samples),
}
# ── the run ──────────────────────────────────────────────────────────────────
def run(
config: Config,
*,
profile: str = "smoke",
soak_seconds: float = 0.0,
output: Path | str | None = None,
) -> dict:
"""Build, measure, evaluate. Returns the report; the caller decides the exit code."""
if profile not in PROFILES:
raise ValueError(f"unknown profile {profile!r}; try one of {sorted(PROFILES)}")
settings = PROFILES[profile]
report = {
"schema_version": SCHEMA_VERSION,
"profile": profile,
"started_at": _now().isoformat(),
"budgets": [
{"metric": b.metric, "limit": b.limit, "unit": b.unit, "why": b.why} for b in BUDGETS
],
"runs": [],
}
measurements: list[dict] = []
for size in settings["sizes"]:
sized = config.model_copy(update={"data_dir": Path(config.data_dir) / f"bench-{size}"})
before = None
build = synthesize(
sized, assets=size, cluster_members=settings["cluster_members"]
)
engine = create_db_engine(sized.database_url)
factory = create_session_factory(engine)
try:
before = sample_resources(sized, factory)
results = [
scenario.run()
for scenario in scenarios(sized, factory, iterations=settings["iterations"])
]
after = sample_resources(sized, factory)
after["scenario"] = "resources"
after["rss_growth_bytes"] = max(0, after["rss_bytes"] - before["rss_bytes"])
soaked = (
soak(sized, factory, seconds=soak_seconds) if soak_seconds > 0 else None
)
finally:
engine.dispose()
measurements.extend(results)
measurements.append(after)
if soaked:
measurements.append(soaked)
report["runs"].append(
{
"assets": size,
"build": build,
"before": before,
"scenarios": results,
"resources": after,
"soak": soaked,
}
)
breaches, exceptions_used = evaluate(profile, measurements)
report["breaches"] = breaches
report["exceptions_applied"] = exceptions_used
report["ok"] = not breaches
report["finished_at"] = _now().isoformat()
if output:
path = Path(output)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(report, indent=2))
return report