perf(api): CUST-WP-0041 — DB indexes, TTL caches, noload on list endpoints
- Migration t7o8p9q0r1s2: indexes on tasks.status, tasks(workstream_id,status), workstreams.status, sbom_snapshots(repo_id,snapshot_at) - workplan-index: 30 s TTL cache + ?refresh param (4171 ms → 16 ms on hit) - /state/summary: 15 s TTL cache, bypassed on Cache-Control: no-cache - /topics/: noload(workstreams, decisions, progress_events) (2382 ms → 115 ms) - /domains/: noload(topics, repos, goals) (2252 ms → 39 ms) - /repos/: noload(goals) (2222 ms → 599 ms first / fast on repeat) - conftest: reset TTL caches between tests to prevent bleed-through Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -2,6 +2,7 @@ import uuid
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import noload
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from api.database import get_session
|
||||
@@ -23,7 +24,11 @@ async def list_domains(
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> list[Domain]:
|
||||
response.headers["Cache-Control"] = "max-age=60, stale-while-revalidate=30"
|
||||
q = select(Domain).order_by(Domain.name)
|
||||
q = select(Domain).options(
|
||||
noload(Domain.topics),
|
||||
noload(Domain.repos),
|
||||
noload(Domain.goals),
|
||||
).order_by(Domain.name)
|
||||
if status and status != "all":
|
||||
q = q.where(Domain.status == status)
|
||||
elif status is None:
|
||||
|
||||
@@ -11,6 +11,7 @@ from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response, status
|
||||
from sqlalchemy import case, func, select
|
||||
from sqlalchemy.orm import noload
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from api.config import settings
|
||||
@@ -55,7 +56,7 @@ async def list_repos(
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> list[ManagedRepo]:
|
||||
response.headers["Cache-Control"] = "max-age=60, stale-while-revalidate=30"
|
||||
q = select(ManagedRepo).order_by(ManagedRepo.name)
|
||||
q = select(ManagedRepo).options(noload(ManagedRepo.goals)).order_by(ManagedRepo.name)
|
||||
if domain:
|
||||
domain_row = await session.execute(select(Domain).where(Domain.slug == domain))
|
||||
domain_obj = domain_row.scalar_one_or_none()
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from fastapi import APIRouter, Depends, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
from sqlalchemy import func, select, text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
@@ -41,9 +42,20 @@ from task_flow_engine import FlowEngine
|
||||
|
||||
router = APIRouter(prefix="/state", tags=["state"])
|
||||
|
||||
_SUMMARY_CACHE: StateSummary | None = None
|
||||
_SUMMARY_CACHE_AT: float = 0.0
|
||||
_SUMMARY_TTL = 15.0
|
||||
|
||||
|
||||
@router.get("/summary", response_model=StateSummary)
|
||||
async def get_summary(session: AsyncSession = Depends(get_session)) -> StateSummary:
|
||||
async def get_summary(
|
||||
request: Request,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> StateSummary:
|
||||
global _SUMMARY_CACHE, _SUMMARY_CACHE_AT
|
||||
no_cache = "no-cache" in request.headers.get("cache-control", "")
|
||||
if not no_cache and _SUMMARY_CACHE is not None and (time.monotonic() - _SUMMARY_CACHE_AT) < _SUMMARY_TTL:
|
||||
return _SUMMARY_CACHE
|
||||
# Run all queries sequentially on one session.
|
||||
# AsyncSession does not support concurrent operations (no gather on same session).
|
||||
|
||||
@@ -294,7 +306,7 @@ async def get_summary(session: AsyncSession = Depends(get_session)) -> StateSumm
|
||||
)
|
||||
)).scalar() or 0
|
||||
|
||||
return StateSummary(
|
||||
result = StateSummary(
|
||||
generated_at=datetime.now(tz=timezone.utc),
|
||||
totals=totals,
|
||||
topics=[
|
||||
@@ -330,6 +342,9 @@ async def get_summary(session: AsyncSession = Depends(get_session)) -> StateSumm
|
||||
for w in open_ws
|
||||
],
|
||||
)
|
||||
_SUMMARY_CACHE = result
|
||||
_SUMMARY_CACHE_AT = time.monotonic()
|
||||
return result
|
||||
|
||||
|
||||
async def _build_domain_summaries(session: AsyncSession) -> list[DomainSummary]:
|
||||
|
||||
@@ -2,6 +2,7 @@ import uuid
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response, status
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import noload
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from api.database import get_session
|
||||
@@ -28,7 +29,11 @@ async def list_topics(
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> list[Topic]:
|
||||
response.headers["Cache-Control"] = "max-age=60, stale-while-revalidate=30"
|
||||
q = select(Topic)
|
||||
q = select(Topic).options(
|
||||
noload(Topic.workstreams),
|
||||
noload(Topic.decisions),
|
||||
noload(Topic.progress_events),
|
||||
)
|
||||
if status:
|
||||
q = q.where(Topic.status == status)
|
||||
q = q.order_by(Topic.created_at)
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import uuid
|
||||
import socket
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
@@ -19,6 +20,10 @@ from api.schemas.workstream import (
|
||||
|
||||
router = APIRouter(prefix="/workstreams", tags=["workstreams"])
|
||||
|
||||
_INDEX_CACHE: dict[str, Any] | None = None
|
||||
_INDEX_CACHE_AT: float = 0.0
|
||||
_INDEX_TTL = 30.0
|
||||
|
||||
|
||||
def _repo_path(repo: ManagedRepo) -> Path | None:
|
||||
hostname = socket.gethostname()
|
||||
@@ -92,8 +97,15 @@ async def list_workstreams(
|
||||
|
||||
|
||||
@router.get("/workplan-index")
|
||||
async def workplan_index(session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
async def workplan_index(
|
||||
refresh: bool = Query(False, description="Force cache invalidation"),
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> dict[str, Any]:
|
||||
"""Map file-backed workstream ids to their local workplan filenames."""
|
||||
global _INDEX_CACHE, _INDEX_CACHE_AT
|
||||
if not refresh and _INDEX_CACHE is not None and (time.monotonic() - _INDEX_CACHE_AT) < _INDEX_TTL:
|
||||
return _INDEX_CACHE
|
||||
|
||||
result = await session.execute(
|
||||
select(ManagedRepo).where(ManagedRepo.status == "active").order_by(ManagedRepo.slug)
|
||||
)
|
||||
@@ -119,7 +131,9 @@ async def workplan_index(session: AsyncSession = Depends(get_session)) -> dict[s
|
||||
"repo_slug": repo.slug,
|
||||
"archived": archived,
|
||||
}
|
||||
return {"workstreams": index}
|
||||
_INDEX_CACHE = {"workstreams": index}
|
||||
_INDEX_CACHE_AT = time.monotonic()
|
||||
return _INDEX_CACHE
|
||||
|
||||
|
||||
@router.post("/", response_model=WorkstreamRead, status_code=status.HTTP_201_CREATED)
|
||||
|
||||
27
state-hub/migrations/versions/t7o8p9q0r1s2_perf_indexes.py
Normal file
27
state-hub/migrations/versions/t7o8p9q0r1s2_perf_indexes.py
Normal file
@@ -0,0 +1,27 @@
|
||||
"""perf: add missing status and composite indexes
|
||||
|
||||
Revision ID: t7o8p9q0r1s2
|
||||
Revises: s6n7o8p9q0r1
|
||||
Create Date: 2026-05-15
|
||||
|
||||
"""
|
||||
from alembic import op
|
||||
|
||||
revision = "t7o8p9q0r1s2"
|
||||
down_revision = "s6n7o8p9q0r1"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_index("ix_tasks_status", "tasks", ["status"])
|
||||
op.create_index("ix_tasks_workstream_status", "tasks", ["workstream_id", "status"])
|
||||
op.create_index("ix_workstreams_status", "workstreams", ["status"])
|
||||
op.create_index("ix_sbom_snapshots_repo_at", "sbom_snapshots", ["repo_id", "snapshot_at"])
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index("ix_sbom_snapshots_repo_at", table_name="sbom_snapshots")
|
||||
op.drop_index("ix_workstreams_status", table_name="workstreams")
|
||||
op.drop_index("ix_tasks_workstream_status", table_name="tasks")
|
||||
op.drop_index("ix_tasks_status", table_name="tasks")
|
||||
@@ -50,6 +50,14 @@ def _schema():
|
||||
def _truncate(_schema):
|
||||
"""Truncate all tables after each test for isolation."""
|
||||
from api.models import Base
|
||||
import api.routers.state as _state_router
|
||||
import api.routers.workstreams as _ws_router
|
||||
|
||||
# Reset in-process TTL caches so stale data from a previous test can't bleed through.
|
||||
_state_router._SUMMARY_CACHE = None
|
||||
_state_router._SUMMARY_CACHE_AT = 0.0
|
||||
_ws_router._INDEX_CACHE = None
|
||||
_ws_router._INDEX_CACHE_AT = 0.0
|
||||
|
||||
yield
|
||||
engine = sqlalchemy.create_engine(_SYNC_URL)
|
||||
|
||||
@@ -3,7 +3,7 @@ id: CUST-WP-0041
|
||||
type: workplan
|
||||
title: "API Performance Optimization"
|
||||
domain: custodian
|
||||
status: todo
|
||||
status: done
|
||||
owner: custodian
|
||||
topic_slug: custodian
|
||||
created: "2026-05-15"
|
||||
|
||||
Reference in New Issue
Block a user