From 218fa38f8e97d0ab33ec25bf8e0e84dec97ad779 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Sun, 11 Oct 2026 03:51:33 +0800 Subject: [PATCH] feat(submissions): query bounded anonymous performance snapshots --- .../versions/0004_performance_watermark.py | 21 ++ backend/src/kf_oj/api/app.py | 3 + .../src/kf_oj/modules/submissions/models.py | 6 + .../kf_oj/modules/submissions/performance.py | 217 ++++++++++++++++++ backend/tests/test_performance_queries.py | 112 +++++++++ 5 files changed, 359 insertions(+) create mode 100644 backend/migrations/versions/0004_performance_watermark.py create mode 100644 backend/src/kf_oj/modules/submissions/performance.py create mode 100644 backend/tests/test_performance_queries.py diff --git a/backend/migrations/versions/0004_performance_watermark.py b/backend/migrations/versions/0004_performance_watermark.py new file mode 100644 index 0000000..9fade76 --- /dev/null +++ b/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") diff --git a/backend/src/kf_oj/api/app.py b/backend/src/kf_oj/api/app.py index 1fe70f7..7f85704 100644 --- a/backend/src/kf_oj/api/app.py +++ b/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: diff --git a/backend/src/kf_oj/modules/submissions/models.py b/backend/src/kf_oj/modules/submissions/models.py index 4898cf4..d106d06 100644 --- a/backend/src/kf_oj/modules/submissions/models.py +++ b/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", diff --git a/backend/src/kf_oj/modules/submissions/performance.py b/backend/src/kf_oj/modules/submissions/performance.py new file mode 100644 index 0000000..727a577 --- /dev/null +++ b/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 diff --git a/backend/tests/test_performance_queries.py b/backend/tests/test_performance_queries.py new file mode 100644 index 0000000..3e06816 --- /dev/null +++ b/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())