feat(submissions): create durable queued evaluations idempotently

backend/src/kf_oj/api/app.py

@@ -10,6 +10,7 @@ from kf_oj.api.body_limits import BodyLimitMiddleware
 from kf_oj.api.v1.identity import router as identity_router
 from kf_oj.api.v1.problems import router as problems_router
 from kf_oj.api.v1.router import router
+from kf_oj.api.v1.submissions import router as submission_router
 from kf_oj.api.v1.worker_artifacts import router as worker_router
 from kf_oj.core.logging import configure_logging
 from kf_oj.core.settings import Settings
@@ -18,6 +19,7 @@ from kf_oj.infra.artifacts.store import ArtifactStore
 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.service import SubmissionService
 from kf_oj.modules.submissions.worker_access import WorkerAccess
 
 
@@ -50,6 +52,9 @@ def create_app(settings: Settings | None = None) -> FastAPI:
                     app.state.sessions, artifacts, settings.judge_admission
                 )
                 if artifacts is not None:
+                    app.state.submissions = SubmissionService(
+                        app.state.sessions, app.state.problems, artifacts, settings.judge_admission
+                    )
                     app.state.worker_access = WorkerAccess(
                         app.state.sessions, app.state.problems, artifacts
                     )
@@ -58,6 +63,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
             app.state.identity = None
             app.state.problems = None
             app.state.worker_access = None
+            app.state.submissions = None
             if engine is not None:
                 await engine.dispose()
             if store is not None:
@@ -84,4 +90,5 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     app.include_router(identity_router, prefix="/api/v1")
     app.include_router(problems_router, prefix="/api/v1")
     app.include_router(worker_router, prefix="/api/v1")
+    app.include_router(submission_router, prefix="/api/v1")
     return app

backend/src/kf_oj/api/body_limits.py

@@ -14,6 +14,9 @@ class BodyLimitMiddleware:
             await self.app(scope, receive, send)
             return
         limit = 8192 if scope["path"] == "/api/v1/auth/login" else 1024 * 1024
+        if scope["path"] == "/api/v1/submissions":
+            # Max256KiB UTF-8 source, worst-case JSON escaping plus bounded metadata.
+            limit = 6 * 256 * 1024 + 8192
         headers = dict(scope.get("headers", []))
         length = headers.get(b"content-length")
         if length is not None:

backend/src/kf_oj/api/v1/submissions.py

@@ -0,0 +1,43 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+
+from kf_oj.api.dependencies import unsafe_principal
+from kf_oj.contracts.submissions import CreateSubmission, SubmissionCreated
+from kf_oj.infra.artifacts.store import ArtifactError
+from kf_oj.modules.identity.public import Principal
+from kf_oj.modules.submissions.public import (
+    IdempotencyConflict,
+    SubmissionBusy,
+    SubmissionUnavailable,
+)
+
+router = APIRouter(prefix="/submissions", tags=["submissions"])
+
+
+@router.post("", status_code=201, response_model=SubmissionCreated)
+async def create(
+    body: CreateSubmission,
+    request: Request,
+    response: Response,
+    principal: Annotated[Principal, Depends(unsafe_principal)],
+):
+    if request.headers.get("content-type", "").split(";", 1)[0] != "application/json":
+        raise HTTPException(415, "JSON request required")
+    service = getattr(request.app.state, "submissions", None)
+    if service is None:
+        raise HTTPException(503, "submission service unavailable")
+    try:
+        result = await service.create(principal, body)
+    except IdempotencyConflict:
+        raise HTTPException(409, "idempotency request conflict") from None
+    except SubmissionBusy:
+        raise HTTPException(429, "pending submission capacity exhausted") from None
+    except SubmissionUnavailable:
+        raise HTTPException(404, "submission capability unavailable") from None
+    except ArtifactError:
+        raise HTTPException(503, "source storage unavailable") from None
+    except ValueError:
+        raise HTTPException(422, "submission outside admission policy") from None
+    response.headers["Cache-Control"] = "no-store"
+    return result

backend/src/kf_oj/contracts/submissions.py

@@ -0,0 +1,37 @@
+from datetime import datetime
+from typing import Annotated, Literal
+
+from pydantic import Field, model_validator
+
+from kf_oj.contracts.judging import MAX_SOURCE_BYTES, Identifier, Language, WireModel
+
+EvaluationState = Literal[
+    "queued", "claimed", "compiling", "running", "cancelling", "finished", "cancelled", "failed"
+]
+
+
+class CreateSubmission(WireModel):
+    problem_version_id: Identifier
+    language: Language
+    idempotency_key: Identifier
+    source: Annotated[
+        str, Field(strict=True, min_length=1, max_length=MAX_SOURCE_BYTES, repr=False)
+    ]
+
+    @model_validator(mode="after")
+    def bounded_source(self):
+        try:
+            if len(self.source.encode("utf-8")) > MAX_SOURCE_BYTES:
+                raise ValueError("source exceeds byte policy")
+        except UnicodeError:
+            raise ValueError("source must be UTF-8") from None
+        return self
+
+
+class SubmissionCreated(WireModel):
+    submission_id: Identifier
+    evaluation_id: Identifier
+    problem_version_id: Identifier
+    language: Language
+    state: EvaluationState
+    created_at: datetime

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

@@ -8,3 +8,11 @@ class WorkerAccessDenied(ValueError):
 
 class WorkerReadBusy(ValueError):
     pass
+
+
+class IdempotencyConflict(ValueError):
+    pass
+
+
+class SubmissionBusy(ValueError):
+    pass

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

@@ -0,0 +1,135 @@
+"""Only initial queued evaluation/outbox writes; execution belongs to Coordinator."""
+
+import hashlib
+import json
+from uuid import uuid4
+
+from sqlalchemy import func, select, text
+
+from kf_oj.contracts.judging import JobManifest
+from kf_oj.contracts.submissions import CreateSubmission, SubmissionCreated
+from kf_oj.infra.artifacts.repository import ArtifactRepository
+from kf_oj.infra.messaging.models import Outbox
+from kf_oj.modules.identity.public import Principal
+from kf_oj.modules.problems.public import ProblemPort, ProblemUnavailable
+from kf_oj.modules.submissions.models import Evaluation, Submission
+from kf_oj.modules.submissions.public import (
+    IdempotencyConflict,
+    SubmissionBusy,
+    SubmissionUnavailable,
+)
+
+
+class SubmissionService:
+    def __init__(self, sessions, problems: ProblemPort, artifacts: ArtifactRepository, admission):
+        self.sessions, self.problems = sessions, problems
+        self.artifacts, self.admission = artifacts, admission
+
+    @staticmethod
+    def _response(submission, evaluation):
+        return SubmissionCreated(
+            submission_id=submission.id,
+            evaluation_id=evaluation.id,
+            problem_version_id=submission.problem_version_id,
+            language=submission.language,
+            state=evaluation.state,
+            created_at=submission.created_at,
+        )
+
+    async def create(self, principal: Principal, request: CreateSubmission) -> SubmissionCreated:
+        # Validate even trusted non-HTTP callers. No path or argv is accepted.
+        request = CreateSubmission.model_validate(request.model_dump())
+        source = request.source.encode("utf-8")
+        request_digest = hashlib.sha256(
+            json.dumps(
+                [
+                    str(request.problem_version_id),
+                    request.language.value,
+                    hashlib.sha256(source).hexdigest(),
+                ],
+                separators=(",", ":"),
+            ).encode()
+        ).hexdigest()
+        if self.admission is None:
+            raise SubmissionUnavailable("judge admission unavailable")
+        async with self.sessions.begin() as session:
+            # Bounded pending capacity and per-account idempotency serialize across API processes.
+            # Two-key namespace is separate from artifact/slug advisory locks.
+            await session.execute(text("SELECT pg_advisory_xact_lock(1262907470, 0)"))
+            existing = (
+                await session.execute(
+                    select(Submission, Evaluation)
+                    .join(Evaluation, Evaluation.submission_id == Submission.id)
+                    .where(
+                        Submission.account_id == principal.account_id,
+                        Submission.idempotency_key == request.idempotency_key,
+                    )
+                )
+            ).one_or_none()
+            if existing is not None:
+                if existing[0].request_digest != request_digest:
+                    raise IdempotencyConflict("idempotency key belongs to another request")
+                return self._response(*existing)
+            pending = ("queued", "claimed", "compiling", "running", "cancelling")
+            total = await session.scalar(
+                select(func.count()).select_from(Evaluation).where(Evaluation.state.in_(pending))
+            )
+            owned = await session.scalar(
+                select(func.count())
+                .select_from(Submission)
+                .join(Evaluation, Evaluation.submission_id == Submission.id)
+                .where(Submission.account_id == principal.account_id, Evaluation.state.in_(pending))
+            )
+            if total >= 64 or owned >= 8:
+                raise SubmissionBusy("pending submission capacity exhausted")
+            try:
+                version = await self.problems.for_submission(session, request.problem_version_id)
+            except ProblemUnavailable:
+                raise SubmissionUnavailable("problem version unavailable") from None
+            if request.language not in version.languages:
+                raise SubmissionUnavailable("language unavailable")
+            # Bytes fsync before metadata/ref transaction. Rollback may leave collectable bytes,
+            # but cannot leave an evaluation that refers to non-durable bytes.
+            reference = await self.artifacts.put(session, source)
+            submission_id, evaluation_id = uuid4(), uuid4()
+            manifest = JobManifest(
+                schema_version=1,
+                kind="judge.manifest",
+                evaluation_id=evaluation_id,
+                submission_id=submission_id,
+                problem_version_id=version.version_id,
+                language=request.language,
+                toolchain=version.toolchain,
+                source=reference,
+                cases=version.cases,
+                comparison="exact_bytes",
+                compile_limits=version.compile_limits,
+                run_limits=version.run_limits,
+                job_wall_time_us=version.job_wall_time_us,
+            )
+            self.admission.admit(manifest, 1)
+            submission = Submission(
+                id=submission_id,
+                account_id=principal.account_id,
+                problem_id=version.problem_id,
+                problem_version_id=version.version_id,
+                language=request.language.value,
+                source_digest=reference.digest,
+                idempotency_key=request.idempotency_key,
+                request_digest=request_digest,
+            )
+            session.add(submission)
+            await session.flush()
+            await session.execute(
+                Evaluation.__table__.insert().values(
+                    id=evaluation_id,
+                    submission_id=submission_id,
+                    problem_version_id=version.version_id,
+                )
+            )
+            session.add(Outbox(id=uuid4(), evaluation_id=evaluation_id))
+            await session.flush()
+            evaluation = await session.get(Evaluation, evaluation_id)
+            response = self._response(submission, evaluation)
+        # The transaction commit must succeed before any success is returned.
+        return response