diff --git a/backend/src/kf_oj/api/app.py b/backend/src/kf_oj/api/app.py index 180a52a..4889ef3 100644 --- a/backend/src/kf_oj/api/app.py +++ b/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 diff --git a/backend/src/kf_oj/api/body_limits.py b/backend/src/kf_oj/api/body_limits.py index ea65bc1..5405096 100644 --- a/backend/src/kf_oj/api/body_limits.py +++ b/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: diff --git a/backend/src/kf_oj/api/v1/submissions.py b/backend/src/kf_oj/api/v1/submissions.py new file mode 100644 index 0000000..316c2ac --- /dev/null +++ b/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 diff --git a/backend/src/kf_oj/contracts/submissions.py b/backend/src/kf_oj/contracts/submissions.py new file mode 100644 index 0000000..a92ecd0 --- /dev/null +++ b/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 diff --git a/backend/src/kf_oj/modules/submissions/public.py b/backend/src/kf_oj/modules/submissions/public.py index 67a2d83..612abaa 100644 --- a/backend/src/kf_oj/modules/submissions/public.py +++ b/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 diff --git a/backend/src/kf_oj/modules/submissions/service.py b/backend/src/kf_oj/modules/submissions/service.py new file mode 100644 index 0000000..0d8158d --- /dev/null +++ b/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 diff --git a/backend/tests/test_submission_create.py b/backend/tests/test_submission_create.py new file mode 100644 index 0000000..e8411f3 --- /dev/null +++ b/backend/tests/test_submission_create.py @@ -0,0 +1,222 @@ +import asyncio +import json +import secrets +from uuid import uuid4 + +import httpx +import pytest +from sqlalchemy import func, select, text +from sqlalchemy.exc import DBAPIError +from sqlalchemy.ext.asyncio import create_async_engine +from test_problems import import_fixture, platform_role + +from kf_oj.api.app import create_app +from kf_oj.contracts.judging import MAX_SOURCE_BYTES, Language +from kf_oj.contracts.submissions import CreateSubmission +from kf_oj.core.settings import Settings +from kf_oj.infra.artifacts.models import Artifact +from kf_oj.infra.messaging.models import Outbox +from kf_oj.modules.submissions.models import Evaluation, Submission +from kf_oj.modules.submissions.public import IdempotencyConflict, SubmissionUnavailable + + +def test_submission_input_utf8_and_unknown_fields(): + base = dict(problem_version_id=str(uuid4()), language="c17", idempotency_key=str(uuid4())) + CreateSubmission(**base, source="é" * (MAX_SOURCE_BYTES // 2)) + for extra in [ + dict(source="é" * (MAX_SOURCE_BYTES // 2 + 1)), + dict(source="x", argv=[]), + dict(source="x", user_id=str(uuid4())), + dict(source=""), + dict(source="\ud800"), + ]: + with pytest.raises(ValueError): + CreateSubmission(**base, **extra) + + +async def setup_app(postgres_url, tmp_path): + _, policy = import_fixture(tmp_path / "import") + app = create_app( + Settings( + _env_file=None, + environment="test", + database_url=postgres_url, + artifact_root=tmp_path / "artifacts", + judge_admission=policy, + ) + ) + engine = create_async_engine(postgres_url) + platform_role(engine) + # Test creates separate permission-constrained platform connections used by application. + app_engine_build = __import__("kf_oj.api.app", fromlist=["build_engine"]) + return app, engine, app_engine_build + + +def test_actual_concurrent_idempotency_version_and_outbox(postgres_url, tmp_path, monkeypatch): + async def run(): + app, engine, mod = await setup_app(postgres_url, tmp_path) + monkeypatch.setattr(mod, "build_engine", lambda _: engine) + async with app.router.lifespan_context(app): + account = await app.state.identity.create_account( + "sub_fixture", secrets.token_urlsafe(24) + ) + version = await app.state.problems.import_directory(tmp_path / "import") + request = CreateSubmission( + problem_version_id=version, + language="c17", + idempotency_key=uuid4(), + source="synthetic source, never executed", + ) + results = await asyncio.gather( + *(app.state.submissions.create(account, request) for _ in range(8)) + ) + assert len({r.submission_id for r in results}) == 1 + async with app.state.sessions() as s: + for model in [Submission, Evaluation, Outbox]: + assert await s.scalar(select(func.count()).select_from(model)) == 1 + stored = await s.get(Submission, results[0].submission_id) + data = await app.state.problems.artifacts.read( + s, app.state.problems.artifacts.store.reference(request.source.encode()) + ) + assert data == request.source.encode() and stored.language == "c17" + with pytest.raises(IdempotencyConflict): + await app.state.submissions.create( + account, request.model_copy(update={"source": "different source"}) + ) + admin = create_async_engine(postgres_url) + async with admin.begin() as c: + await c.execute(text("SET LOCAL ROLE kf_oj_coordinator")) + await c.execute( + text("UPDATE evaluations SET state='finished',verdict='accepted' WHERE id=:id"), + {"id": results[0].evaluation_id}, + ) + await admin.dispose() + assert (await app.state.submissions.create(account, request)).state == "finished" + cpp = await app.state.submissions.create( + account, + request.model_copy(update={"language": Language.CPP20, "idempotency_key": uuid4()}), + ) + assert cpp.submission_id != results[0].submission_id + async with app.state.sessions.begin() as s: + await s.execute(text("UPDATE problems SET enabled=false")) + assert (await app.state.submissions.create(account, request)).submission_id == results[ + 0 + ].submission_id + with pytest.raises(SubmissionUnavailable): + await app.state.submissions.create( + account, request.model_copy(update={"idempotency_key": uuid4()}) + ) + + asyncio.run(run()) + + +def test_actual_database_failure_rolls_back_entire_creation(postgres_url, tmp_path, monkeypatch): + async def run(): + app, engine, mod = await setup_app(postgres_url, tmp_path) + monkeypatch.setattr(mod, "build_engine", lambda _: engine) + async with app.router.lifespan_context(app): + account = await app.state.identity.create_account( + "rollback_fixture", secrets.token_urlsafe(24) + ) + version = await app.state.problems.import_directory(tmp_path / "import") + admin = create_async_engine(postgres_url) + async with admin.begin() as c: + await c.execute( + text( + """CREATE FUNCTION test_reject_outbox() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'synthetic rollback'; END $$""" + ) + ) + await c.execute( + text( + "CREATE TRIGGER test_reject_outbox BEFORE INSERT ON outbox " + "FOR EACH ROW EXECUTE FUNCTION test_reject_outbox()" + ) + ) + request = CreateSubmission( + problem_version_id=version, + language="c17", + idempotency_key=uuid4(), + source="rollback synthetic bytes", + ) + with pytest.raises(DBAPIError): + await app.state.submissions.create(account, request) + async with app.state.sessions() as s: + for model in [Submission, Evaluation, Outbox]: + assert await s.scalar(select(func.count()).select_from(model)) == 0 + ref = app.state.problems.artifacts.store.reference(request.source.encode()) + assert await s.get(Artifact, ref.digest) is None + # Bytes can survive rollback as a collectable orphan; no durable reference exists. + assert app.state.problems.artifacts.store.read(ref) == request.source.encode() + async with admin.begin() as c: + await c.execute(text("DROP TRIGGER test_reject_outbox ON outbox")) + await c.execute(text("DROP FUNCTION test_reject_outbox()")) + assert (await app.state.submissions.create(account, request)).state == "queued" + await admin.dispose() + + asyncio.run(run()) + + +def test_actual_authenticated_creation_request_boundaries(postgres_url, tmp_path, monkeypatch): + async def run(): + app, engine, mod = await setup_app(postgres_url, tmp_path) + monkeypatch.setattr(mod, "build_engine", lambda _: engine) + async with app.router.lifespan_context(app): + password = secrets.token_urlsafe(24) + await app.state.identity.create_account("api_submission", password) + version = await app.state.problems.import_directory(tmp_path / "import") + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://localhost:8000" + ) as client: + req = dict( + problem_version_id=str(version), + language="c17", + idempotency_key=str(uuid4()), + source="synthetic ordinary data", + ) + assert (await client.post("/api/v1/submissions", json=req)).status_code == 403 + login = await client.post( + "/api/v1/auth/login", + json=dict(username="api_submission", password=password), + headers={"Origin": "http://localhost:8000"}, + ) + assert login.status_code == 200 + headers = { + "Origin": "http://localhost:8000", + "X-CSRF-Token": login.json()["csrf_token"], + } + assert ( + await client.post("/api/v1/submissions", json=req, headers=headers) + ).status_code == 201 + assert ( + await client.post( + "/api/v1/submissions", json=req | {"user_id": str(uuid4())}, headers=headers + ) + ).status_code == 422 + assert ( + await client.post( + "/api/v1/submissions", json=req | {"language": "python"}, headers=headers + ) + ).status_code == 422 + assert ( + await client.post( + "/api/v1/submissions", json=req | {"source": "changed"}, headers=headers + ) + ).status_code == 409 + assert ( + await client.post( + "/api/v1/submissions", + content=b"x" * (6 * MAX_SOURCE_BYTES + 8193), + headers=headers | {"Content-Type": "application/json"}, + ) + ).status_code == 413 + # JSON escaping can exceed old1MiB body ceiling while valid source remains256KiB. + req["idempotency_key"] = str(uuid4()) + req["source"] = "\x01" * MAX_SOURCE_BYTES + assert len(json.dumps(req)) > 1024 * 1024 + assert ( + await client.post("/api/v1/submissions", json=req, headers=headers) + ).status_code == 201 + + asyncio.run(run()) diff --git a/contracts/openapi/platform.v1.json b/contracts/openapi/platform.v1.json index b5c6b13..6c68835 100644 --- a/contracts/openapi/platform.v1.json +++ b/contracts/openapi/platform.v1.json @@ -25,6 +25,38 @@ "title": "ArtifactReference", "type": "object" }, + "CreateSubmission": { + "additionalProperties": false, + "properties": { + "idempotency_key": { + "format": "uuid", + "title": "Idempotency Key", + "type": "string" + }, + "language": { + "$ref": "#/components/schemas/Language" + }, + "problem_version_id": { + "format": "uuid", + "title": "Problem Version Id", + "type": "string" + }, + "source": { + "maxLength": 262144, + "minLength": 1, + "title": "Source", + "type": "string" + } + }, + "required": [ + "problem_version_id", + "language", + "idempotency_key", + "source" + ], + "title": "CreateSubmission", + "type": "object" + }, "CurrentUser": { "additionalProperties": false, "properties": { @@ -468,6 +500,58 @@ "title": "RunLimits", "type": "object" }, + "SubmissionCreated": { + "additionalProperties": false, + "properties": { + "created_at": { + "format": "date-time", + "title": "Created At", + "type": "string" + }, + "evaluation_id": { + "format": "uuid", + "title": "Evaluation Id", + "type": "string" + }, + "language": { + "$ref": "#/components/schemas/Language" + }, + "problem_version_id": { + "format": "uuid", + "title": "Problem Version Id", + "type": "string" + }, + "state": { + "enum": [ + "queued", + "claimed", + "compiling", + "running", + "cancelling", + "finished", + "cancelled", + "failed" + ], + "title": "State", + "type": "string" + }, + "submission_id": { + "format": "uuid", + "title": "Submission Id", + "type": "string" + } + }, + "required": [ + "submission_id", + "evaluation_id", + "problem_version_id", + "language", + "state", + "created_at" + ], + "title": "SubmissionCreated", + "type": "object" + }, "TestCase": { "additionalProperties": false, "properties": { @@ -942,6 +1026,47 @@ "problems" ] } + }, + "/api/v1/submissions": { + "post": { + "operationId": "create_api_v1_submissions_post", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CreateSubmission" + } + } + }, + "required": true + }, + "responses": { + "201": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubmissionCreated" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Create", + "tags": [ + "submissions" + ] + } } } } diff --git a/frontend/src/api/generated/schema.d.ts b/frontend/src/api/generated/schema.d.ts index b9cf25f..46a1fe1 100644 --- a/frontend/src/api/generated/schema.d.ts +++ b/frontend/src/api/generated/schema.d.ts @@ -143,6 +143,23 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/submissions": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Create */ + post: operations["create_api_v1_submissions_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; } export type webhooks = Record; export interface components { @@ -154,6 +171,22 @@ export interface components { /** Digest */ digest: string; }; + /** CreateSubmission */ + CreateSubmission: { + /** + * Idempotency Key + * Format: uuid + */ + idempotency_key: string; + language: components["schemas"]["Language"]; + /** + * Problem Version Id + * Format: uuid + */ + problem_version_id: string; + /** Source */ + source: string; + }; /** CurrentUser */ CurrentUser: { /** @@ -342,6 +375,35 @@ export interface components { /** Workdir Bytes */ workdir_bytes: number; }; + /** SubmissionCreated */ + SubmissionCreated: { + /** + * Created At + * Format: date-time + */ + created_at: string; + /** + * Evaluation Id + * Format: uuid + */ + evaluation_id: string; + language: components["schemas"]["Language"]; + /** + * Problem Version Id + * Format: uuid + */ + problem_version_id: string; + /** + * State + * @enum {string} + */ + state: "queued" | "claimed" | "compiling" | "running" | "cancelling" | "finished" | "cancelled" | "failed"; + /** + * Submission Id + * Format: uuid + */ + submission_id: string; + }; /** TestCase */ TestCase: { /** @@ -611,4 +673,37 @@ export interface operations { }; }; }; + create_api_v1_submissions_post: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["CreateSubmission"]; + }; + }; + responses: { + /** @description Successful Response */ + 201: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["SubmissionCreated"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; }