feat(submissions): query bounded anonymous performance snapshots

backend/migrations/versions/0004_performance_watermark.py

@@ -0,0 +1,21 @@
+"""Indexed group watermark for bounded aggregate cache invalidation."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0004_performance_watermark"
+down_revision = "0003_submission_performance"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "ix_performance_group_watermark",
+        "evaluation_performance",
+        ["comparison_group", sa.text("finished_at DESC"), sa.text("evaluation_id DESC")],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_performance_group_watermark", table_name="evaluation_performance")

backend/src/kf_oj/api/app.py

@@ -20,6 +20,7 @@ from kf_oj.infra.coordinator import CoordinatorControl
 from kf_oj.infra.db.session import build_engine, build_session_factory
 from kf_oj.modules.identity.service import IdentityService
 from kf_oj.modules.problems.service import ProblemService
+from kf_oj.modules.submissions.performance import SubmissionPerformanceQueries
 from kf_oj.modules.submissions.queries import SubmissionQueries
 from kf_oj.modules.submissions.service import SubmissionService
 from kf_oj.modules.submissions.worker_access import WorkerAccess
@@ -42,6 +43,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
                 app.state.identity = IdentityService(
                     app.state.sessions, lifetime_seconds=settings.session_lifetime_seconds
                 )
+                app.state.submission_performance = SubmissionPerformanceQueries(app.state.sessions)
                 artifacts = None
                 if settings.artifact_root is not None:
                     store = ArtifactStore(
@@ -77,6 +79,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
             app.state.worker_access = None
             app.state.submissions = None
             app.state.submission_queries = None
+            app.state.submission_performance = None
             if engine is not None:
                 await engine.dispose()
             if store is not None:

backend/src/kf_oj/modules/submissions/models.py

@@ -169,6 +169,12 @@ class EvaluationPerformance(Base):
         CheckConstraint("report_digest ~ '^[0-9a-f]{64}$'", name="report_digest"),
         CheckConstraint("cpu_time_us IS NULL OR cpu_time_us >= 0", name="cpu"),
         CheckConstraint("memory_peak_bytes IS NULL OR memory_peak_bytes >= 0", name="memory"),
+        Index(
+            "ix_performance_group_watermark",
+            "comparison_group",
+            text("finished_at DESC"),
+            text("evaluation_id DESC"),
+        ),
         Index(
             "ix_performance_cpu_time_us_cohort",
             "comparison_group",

backend/src/kf_oj/modules/submissions/performance.py

@@ -0,0 +1,217 @@
+"""Read-only, bounded anonymous cohorts from Coordinator's immutable projection."""
+
+import asyncio
+import json
+import time
+from collections import OrderedDict
+from uuid import UUID
+
+from sqlalchemy import select, text
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from kf_oj.contracts.judging import I64_MAX
+from kf_oj.contracts.judging_versions import decode_report
+from kf_oj.contracts.submission_performance import (
+    MAX_PERFORMANCE_SAMPLES,
+    MIN_PERFORMANCE_SAMPLES,
+    PerformanceBucket,
+    PerformanceMetric,
+    SubmissionPerformance,
+)
+from kf_oj.modules.identity.public import Principal
+from kf_oj.modules.submissions.models import Evaluation, EvaluationPerformance, Submission
+from kf_oj.modules.submissions.public import SubmissionUnavailable
+
+CACHE_TTL_SECONDS = 30
+CACHE_CAPACITY = 128
+HISTOGRAM_BUCKETS = 24
+
+
+def metric(unit, current, rows=(), *, reason=None) -> PerformanceMetric:
+    """Integer arithmetic throughout; conversion for display belongs to the client."""
+    truncated = len(rows) > MAX_PERFORMANCE_SAMPLES
+    rows = rows[:MAX_PERFORMANCE_SAMPLES]
+    count = len(rows)
+    base = dict(
+        unit=unit,
+        current=None if current is None else str(current),
+        sample_count=count,
+        window_truncated=truncated,
+        oldest_sample_at=rows[-1][1] if rows else None,
+    )
+    if reason is None:
+        reason = (
+            "measurement_missing"
+            if current is None
+            else ("insufficient_samples" if count < MIN_PERFORMANCE_SAMPLES else None)
+        )
+    if reason is not None:
+        return PerformanceMetric(
+            **base,
+            available=False,
+            reason=reason,
+            strictly_slower_count=None,
+            tied_count=None,
+            buckets=(),
+        )
+    values = [row[0] for row in rows]
+    end = max(1, current, max(values))
+    step = (end + HISTOGRAM_BUCKETS - 1) // HISTOGRAM_BUCKETS
+    # Last upper bound includes the maximum, including I64_MAX. No overflow/padding.
+    boundaries = list(range(0, end, step)) + [end]
+    counts = [0] * (len(boundaries) - 1)
+    for value in values:
+        counts[min(value // step, len(counts) - 1)] += 1
+    return PerformanceMetric(
+        **base,
+        available=True,
+        reason=None,
+        strictly_slower_count=sum(value > current for value in values),
+        tied_count=sum(value == current for value in values),
+        buckets=tuple(
+            PerformanceBucket(
+                start=str(start),
+                end=str(stop),
+                upper_inclusive=index == len(counts) - 1,
+                count=counts[index],
+            )
+            for index, (start, stop) in enumerate(zip(boundaries, boundaries[1:], strict=False))
+        ),
+    )
+
+
+class SubmissionPerformanceQueries:
+    def __init__(self, sessions: async_sessionmaker):
+        self.sessions = sessions
+        # No queued aggregate workload. At most one bounded in-memory row cohort per process.
+        self._slot = asyncio.Lock()
+        # Cache contains small immutable aggregate DTOs, never 10,000-row arrays.
+        self._cache: OrderedDict[tuple, tuple[float, SubmissionPerformance]] = OrderedDict()
+
+    async def detail(self, principal: Principal, submission_id: UUID) -> SubmissionPerformance:
+        if self._slot.locked():
+            raise ValueError("performance query capacity unavailable")
+        async with self._slot:
+            try:
+                async with asyncio.timeout(3), self.sessions() as session, session.begin():
+                    await session.execute(
+                        text("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY")
+                    )
+                    await session.execute(text("SET LOCAL statement_timeout='2s'"))
+                    await session.execute(text("SET LOCAL lock_timeout='500ms'"))
+                    # Ownership checked even on a cache hit. Session authentication is upstream.
+                    row = (
+                        await session.execute(
+                            select(Submission, Evaluation, EvaluationPerformance)
+                            .join(Evaluation, Evaluation.submission_id == Submission.id)
+                            .outerjoin(
+                                EvaluationPerformance,
+                                EvaluationPerformance.evaluation_id == Evaluation.id,
+                            )
+                            .where(
+                                Submission.id == submission_id,
+                                Submission.account_id == principal.account_id,
+                            )
+                        )
+                    ).one_or_none()
+                    if row is None:
+                        raise SubmissionUnavailable("submission unavailable")
+                    submission, evaluation, projection = row
+                    as_of = await session.scalar(text("SELECT clock_timestamp()"))
+                    accepted = (
+                        evaluation.state == "finished"
+                        and evaluation.verdict == "accepted"
+                        and evaluation.failure_code is None
+                    )
+                    if projection is not None and (
+                        not accepted
+                        or projection.attempt_id != evaluation.current_attempt_id
+                        or projection.report_digest != evaluation.final_report_digest
+                    ):
+                        raise ValueError("stored performance binding invalid")
+                    cpu = memory = None
+                    reason = None if accepted else "not_accepted"
+                    if projection is not None:
+                        cpu, memory = projection.cpu_time_us, projection.memory_peak_bytes
+                    elif accepted:
+                        # A single owned legacy report, never a full-table JSON scan or promotion.
+                        report = decode_report(
+                            json.dumps(evaluation.final_report, ensure_ascii=False).encode()
+                        )
+                        if (
+                            report.binding.evaluation_id != evaluation.id
+                            or report.binding.attempt_id != evaluation.current_attempt_id
+                            or report.verdict.value != "accepted"
+                            or not report.cleanup_complete
+                        ):
+                            raise ValueError("stored result binding invalid")
+                        cpu, memory = report.run_cpu_time_us, report.run_memory_peak_bytes
+                        reason = "environment_unverified"
+                    if any(v is not None and not 0 <= v <= I64_MAX for v in (cpu, memory)):
+                        raise ValueError("stored measurement out of range")
+                    base = dict(
+                        submission_id=submission.id,
+                        problem_version_id=submission.problem_version_id,
+                        language=submission.language,
+                        as_of=as_of,
+                        comparison_group=None
+                        if projection is None
+                        else projection.comparison_group,
+                    )
+                    if projection is None:
+                        return SubmissionPerformance(
+                            **base,
+                            cpu=metric("cpu_microseconds", cpu, reason=reason),
+                            memory=metric("bytes", memory, reason=reason),
+                        )
+                    group = projection.comparison_group
+                    watermark = (
+                        await session.execute(
+                            select(
+                                EvaluationPerformance.finished_at,
+                                EvaluationPerformance.evaluation_id,
+                            )
+                            .where(EvaluationPerformance.comparison_group == group)
+                            .order_by(
+                                EvaluationPerformance.finished_at.desc(),
+                                EvaluationPerformance.evaluation_id.desc(),
+                            )
+                            .limit(1)
+                        )
+                    ).one()
+                    key = (submission.id, projection.report_digest, group, tuple(watermark))
+                    cached = self._cache.get(key)
+                    now = time.monotonic()
+                    if cached is not None and now - cached[0] < CACHE_TTL_SECONDS:
+                        self._cache.move_to_end(key)
+                        return cached[1]
+                    metrics = []
+                    for column, unit, current in [
+                        (EvaluationPerformance.cpu_time_us, "cpu_microseconds", cpu),
+                        (EvaluationPerformance.memory_peak_bytes, "bytes", memory),
+                    ]:
+                        rows = (
+                            await session.execute(
+                                select(column, EvaluationPerformance.finished_at)
+                                .where(
+                                    EvaluationPerformance.comparison_group == group,
+                                    EvaluationPerformance.evaluation_id != evaluation.id,
+                                    column.is_not(None),
+                                )
+                                .order_by(
+                                    EvaluationPerformance.finished_at.desc(),
+                                    EvaluationPerformance.evaluation_id.desc(),
+                                )
+                                .limit(MAX_PERFORMANCE_SAMPLES + 1)
+                            )
+                        ).all()
+                        metrics.append(metric(unit, current, rows))
+                    result = SubmissionPerformance(**base, cpu=metrics[0], memory=metrics[1])
+                    self._cache[key] = (now, result)
+                    self._cache.move_to_end(key)
+                    while len(self._cache) > CACHE_CAPACITY:
+                        self._cache.popitem(last=False)
+                    return result
+            except (SQLAlchemyError, TimeoutError) as error:
+                raise ValueError("performance database unavailable") from error

backend/tests/test_performance_queries.py

@@ -0,0 +1,112 @@
+"""Integer-domain tests and finite PG read integration, not measured executions."""
+
+import asyncio
+from datetime import UTC, datetime
+
+import pytest
+from test_coordinator import claim, fixture, manifest, stages, synthetic_report, worker_headers
+
+from kf_oj.contracts.judging import I64_MAX
+from kf_oj.modules.submissions.performance import metric
+
+
+@pytest.mark.parametrize(
+    "current,values,greater,ties",
+    [
+        (0, [0, 0, 1, 2, 2], 3, 2),
+        (
+            9_007_199_254_740_993,
+            [9_007_199_254_740_992, 9_007_199_254_740_993, 9_007_199_254_740_994, I64_MAX, 0],
+            2,
+            1,
+        ),
+        (I64_MAX, [0, 1, 2, 3, 4], 0, 0),
+        (0, [0, 0, 0, 0, 0], 0, 5),
+    ],
+)
+def test_exact_metrics_keep_zero_ties_and_extreme_values(current, values, greater, ties):
+    now = datetime.now(UTC)
+    result = metric("cpu_microseconds", current, [(v, now) for v in values])
+    assert result.available and result.current == str(current)
+    assert result.strictly_slower_count == greater and result.tied_count == ties
+    assert sum(b.count for b in result.buckets) == len(values)
+    assert len(result.buckets) <= 24
+    for value in values:
+        assert (
+            sum(
+                int(b.start) <= value < int(b.end) or (b.upper_inclusive and value == int(b.end))
+                for b in result.buckets
+            )
+            == 1
+        )
+
+
+def test_recent_window_missing_and_insufficient_are_distinct():
+    now = datetime.now(UTC)
+    limited = metric("bytes", 0, [(1, now)] * 10_001)
+    assert limited.sample_count == 10_000 and limited.window_truncated
+    assert limited.strictly_slower_count == 10_000
+    assert metric("bytes", None).reason == "measurement_missing"
+    assert metric("bytes", 0).reason == "insufficient_samples"
+    assert metric("bytes", 0, [(0, now)] * 4).buckets == ()
+
+
+def test_actual_owned_read_snapshot_cache_invalidation_and_history(
+    postgres_url, tmp_path, monkeypatch
+):
+    async def run():
+        async with fixture(
+            postgres_url, tmp_path, monkeypatch, measurement_epochs=["a" * 64] * 2
+        ) as (client, connection, workers, _, create):
+            app = client.fixture_app
+            owner = client.fixture_account
+            created_items = []
+            for _ in range(6):
+                created = await create()
+                grant = await claim(client, created, workers[0])
+                await stages(client, workers[0], grant)
+                report = synthetic_report(grant, await manifest(connection, grant))
+                result = await client.post(
+                    "/internal/report", json=report, headers=worker_headers(workers[0], grant)
+                )
+                assert result.status_code == 204, result.text
+                created_items.append(created)
+            target = created_items[0].submission_id
+            first = await app.state.submission_performance.detail(owner, target)
+            assert first.cpu.sample_count == 5 and first.cpu.tied_count == 5
+            assert first.cpu.strictly_slower_count == 0 and first.memory.tied_count == 5
+            cached = await app.state.submission_performance.detail(owner, target)
+            assert cached == first
+            pending = await create()
+            assert (
+                await app.state.submission_performance.detail(owner, pending.submission_id)
+            ).cpu.reason == "not_accepted"
+            grant = await claim(client, pending, workers[0])
+            await stages(client, workers[0], grant)
+            report = synthetic_report(grant, await manifest(connection, grant))
+            assert (
+                await client.post(
+                    "/internal/report", json=report, headers=worker_headers(workers[0], grant)
+                )
+            ).status_code == 204
+            updated = await app.state.submission_performance.detail(owner, target)
+            assert updated.cpu.sample_count == 6 and updated.cpu.tied_count == 6
+            assert updated.as_of >= first.as_of
+            page = await app.state.submission_queries.page(
+                owner, None, 2, client.fixture_version, "c17"
+            )
+            assert len(page.items) == 2 and page.next_after
+            older = await app.state.submission_queries.page(
+                owner, page.next_after, 2, client.fixture_version, "c17"
+            )
+            assert len(older.items) == 2
+            assert not {i.submission_id for i in page.items} & {
+                i.submission_id for i in older.items
+            }
+            assert not (
+                await app.state.submission_queries.page(
+                    owner, None, 2, client.fixture_version, "java21"
+                )
+            ).items
+
+    asyncio.run(run())