diff --git a/backend/src/kf_oj/contracts/judge_state.py b/backend/src/kf_oj/contracts/judge_state.py index ae04200..0fa4b10 100644 --- a/backend/src/kf_oj/contracts/judge_state.py +++ b/backend/src/kf_oj/contracts/judge_state.py @@ -27,7 +27,7 @@ class EvaluationState(StrEnum): self.QUEUED, }, self.RUNNING: {self.FINISHED, self.CANCELLING, self.FAILED, self.QUEUED}, - self.CANCELLING: {self.CANCELLED, self.FAILED}, + self.CANCELLING: {self.CANCELLED, self.FAILED, self.QUEUED}, } return target in transitions.get(self, set()) diff --git a/backend/src/kf_oj/contracts/judging.py b/backend/src/kf_oj/contracts/judging.py index 3851354..d5adcb1 100644 --- a/backend/src/kf_oj/contracts/judging.py +++ b/backend/src/kf_oj/contracts/judging.py @@ -352,12 +352,20 @@ class AdmissionPolicy(WireModel): raise ValueError("job exceeds trusted node budget") +class StoppedReport(WireModel): + schema_version: Version + kind: Literal["judge.stopped"] + binding: LeaseBinding + cleanup_complete: StrictBool + + WIRE_MODELS = { "judge-manifest": JobManifest, "judge-lease": LeaseGrant, "judge-stage": StageReport, "judge-result": JudgementReport, "judge-admission": AdmissionPolicy, + "judge-stopped": StoppedReport, } diff --git a/backend/tests/test_coordinator.py b/backend/tests/test_coordinator.py new file mode 100644 index 0000000..16c7769 --- /dev/null +++ b/backend/tests/test_coordinator.py @@ -0,0 +1,606 @@ +"""Actual Rust HTTP + PostgreSQL state tests, using explicitly synthetic reports. + +No program is compiled/run by this host suite. Root stop proofs and lease-loss +execution are separately required in the independent VM (F05/F06). +""" + +import asyncio +import json +import secrets +import socket +import subprocess +from contextlib import asynccontextmanager +from pathlib import Path +from uuid import UUID, uuid4 + +import asyncpg +import httpx +from sqlalchemy.engine import make_url +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.submissions import CreateSubmission +from kf_oj.core.settings import Settings + +ROOT = Path(__file__).resolve().parents[2] + + +@asynccontextmanager +async def fixture(postgres_url, tmp_path, monkeypatch): + _, policy = import_fixture(tmp_path / "import") + engine = create_async_engine(postgres_url) + platform_role(engine) + monkeypatch.setattr("kf_oj.api.app.build_engine", lambda _: engine) + app = create_app( + Settings( + _env_file=None, + environment="test", + database_url=postgres_url, + artifact_root=tmp_path / "artifacts", + judge_admission=policy, + ) + ) + url = make_url(postgres_url) + c = await asyncpg.connect( + url.set(drivername="postgresql").render_as_string(hide_password=False) + ) + role = "jp_coord_" + uuid4().hex + password = secrets.token_hex(32) + await c.execute(f"CREATE ROLE {role} LOGIN PASSWORD '{password}'") + await c.execute(f"GRANT kf_oj_coordinator TO {role}") + coord_url = url.set(drivername="postgresql", username=role, password=password).render_as_string( + hide_password=False + ) + # Only loopback control-plane service is started; finite actual systemd scope. + with socket.socket() as reserved: + reserved.bind(("127.0.0.1", 0)) + port = reserved.getsockname()[1] + workers = [dict(id=str(uuid4()), token=secrets.token_hex(32)) for _ in range(2)] + platform = secrets.token_hex(32) + config = tmp_path / "coordinator.json" + config.write_text( + json.dumps( + dict( + listen=f"127.0.0.1:{port}", + database_url=coord_url, + workers=workers, + platform_token=platform, + admission=policy.model_dump(mode="json"), + lease_seconds=5, + recovery_seconds=1, + ) + ) + ) + config.chmod(0o600) + unit = "kf-jp-coord-" + uuid4().hex + binary = ROOT / "judge/target/debug/kf-coordinator" + assert binary.is_file(), "build actual Coordinator before this acceptance" + subprocess.run( + [ + "systemd-run", + "--user", + f"--unit={unit}", + "--collect", + "--service-type=exec", + "--property=MemoryMax=128M", + "--property=MemorySwapMax=0", + "--property=CPUQuota=50%", + "--property=TasksMax=32", + "--property=RuntimeMaxSec=180", + "--property=NoNewPrivileges=yes", + str(binary), + "--config", + str(config), + ], + check=True, + capture_output=True, + ) + try: + async with app.router.lifespan_context(app): + owner = await app.state.identity.create_account( + "coordinator_fixture", secrets.token_urlsafe(24) + ) + version = await app.state.problems.import_directory(tmp_path / "import") + async with httpx.AsyncClient(base_url=f"http://127.0.0.1:{port}", timeout=10) as client: + for _ in range(100): + try: + if (await client.get("/health/ready")).status_code == 200: + break + except httpx.TransportError: + pass + await asyncio.sleep(0.05) + else: + raise AssertionError("actual Coordinator not ready") + props = subprocess.check_output( + [ + "systemctl", + "--user", + "show", + unit, + "--property=ControlGroup,MemoryMax,MemorySwapMax,TasksMax,CPUQuotaPerSecUSec,RuntimeMaxUSec,MainPID", + ], + text=True, + ) + values = dict(line.split("=", 1) for line in props.splitlines()) + group = Path("/sys/fs/cgroup") / values["ControlGroup"].lstrip("/") + assert { + n: (group / n).read_text().strip() + for n in ["memory.max", "memory.swap.max", "pids.max", "cpu.max"] + } == { + "memory.max": "134217728", + "memory.swap.max": "0", + "pids.max": "32", + "cpu.max": "50000 100000", + } + + async def create(): + return await app.state.submissions.create( + owner, + CreateSubmission( + problem_version_id=version, + language="c17", + idempotency_key=uuid4(), + source="synthetic report fixture; never compiled on host", + ), + ) + + client.coordinator_unit = unit + client.coordinator_role = role + yield client, c, workers, platform, create + finally: + subprocess.run(["systemctl", "--user", "stop", unit], check=True, capture_output=True) + assert ( + subprocess.run( + ["systemctl", "--user", "is-active", unit], capture_output=True + ).returncode + != 0 + ) + await c.execute(f"DROP ROLE {role}") + await c.close() + + +def notice(eid): + return dict(schema_version=1, kind="judge.evaluation", evaluation_id=str(eid)) + + +def worker_headers(worker, grant=None): + headers = {"Authorization": "Bearer " + worker["token"]} + if grant: + headers["X-Lease-Token"] = grant["lease_token"] + return headers + + +async def claim(client, created, worker): + r = await client.post( + "/internal/claim", json=notice(created.evaluation_id), headers=worker_headers(worker) + ) + assert r.status_code == 200, r.text + return r.json() + + +async def manifest(c, g): + v = await c.fetchval( + "SELECT manifest FROM judge_attempts WHERE id=$1", UUID(g["binding"]["attempt_id"]) + ) + return json.loads(v) + + +async def stages(client, worker, g): + h = worker_headers(worker, g) + for seq, stage, index in [(1, "compiling", None), (2, "running", 0), (3, "running", 1)]: + r = await client.post( + "/internal/stage", + headers=h, + json=dict( + schema_version=1, + kind="judge.stage", + binding=g["binding"], + sequence=seq, + stage=stage, + case_index=index, + ), + ) + assert r.status_code == 204, r.text + + +def synthetic_report(g, m, verdict="accepted"): + cases = [ + dict(case_id=c["case_id"], verdict=verdict, cpu_time_us=100, memory_peak_bytes=4096) + for c in m["cases"] + ] + return dict( + schema_version=1, + kind="judge.result", + binding=g["binding"], + manifest_digest=g["manifest_digest"], + compilation=dict( + status="succeeded", + program=dict(digest="d" * 64, bytes=4096), + diagnostics="", + diagnostics_truncated=False, + cpu_time_us=1000, + memory_peak_bytes=16384, + ), + cases=cases, + verdict=verdict, + cleanup_complete=True, + failure_code=None, + run_cpu_time_us=len(cases) * 100, + run_memory_peak_bytes=4096, + ) + + +def test_actual_auth_claim_competition_stage_fence_and_report(postgres_url, tmp_path, monkeypatch): + async def run(): + async with fixture(postgres_url, tmp_path, monkeypatch) as ( + client, + c, + workers, + platform, + create, + ): + created = await create() + n = notice(created.evaluation_id) + assert (await client.post("/internal/claim", json=n)).status_code == 401 + assert ( + await client.post( + "/internal/claim", json=n, headers={"Authorization": "Bearer " + platform} + ) + ).status_code == 409 + r = await client.post( + "/internal/claim", + content=b"x" * 262145, + headers=worker_headers(workers[0]) | {"Content-Type": "application/json"}, + ) + assert r.status_code == 413 + outcomes = await asyncio.gather( + *( + client.post("/internal/claim", json=n, headers=worker_headers(w)) + for w in workers + ) + ) + assert sorted(r.status_code for r in outcomes) == [200, 204] + g = next(r.json() for r in outcomes if r.status_code == 200) + w = next(w for w in workers if w["id"] == g["binding"]["worker_id"]) + other = next(w2 for w2 in workers if w2 != w) + assert await c.fetchval("SELECT count(*) FROM judge_attempts") == 1 + assert ( + await client.post( + "/internal/renew", json=g["binding"], headers=worker_headers(other, g) + ) + ).status_code == 409 + assert ( + await client.post( + "/internal/renew", + json=g["binding"] | {"fencing_version": 2}, + headers=worker_headers(w, g), + ) + ).status_code == 409 + bad = dict( + schema_version=1, + kind="judge.stage", + binding=g["binding"], + sequence=1, + stage="running", + case_index=0, + ) + assert ( + await client.post("/internal/stage", json=bad, headers=worker_headers(w, g)) + ).status_code == 409 + await stages(client, w, g) + progress = await client.post( + "/internal/progress", json=g["binding"], headers=worker_headers(w, g) + ) + assert progress.status_code == 200 + assert progress.json() == { + "state": "running", + "sequence": 3, + "stage": "running", + "case_index": 1, + } + subprocess.run( + ["systemctl", "--user", "restart", client.coordinator_unit], + check=True, + capture_output=True, + ) + progress = await client.post( + "/internal/progress", json=g["binding"], headers=worker_headers(w, g) + ) + assert progress.status_code == 200 and progress.json()["sequence"] == 3 + + assert ( + await client.post("/internal/stage", json=bad, headers=worker_headers(w, g)) + ).status_code == 409 + assert ( + await client.post( + "/internal/renew", json=g["binding"], headers=worker_headers(w, g) + ) + ).status_code == 200 + m = await manifest(c, g) + report = synthetic_report(g, m) + wrong = report | {"cases": list(reversed(report["cases"]))} + assert ( + await client.post("/internal/report", json=wrong, headers=worker_headers(w, g)) + ).status_code == 422 + assert await c.fetchval("SELECT count(*) FROM case_results") == 0 + good = await client.post("/internal/report", json=report, headers=worker_headers(w, g)) + assert good.status_code == 204, good.text + assert ( + await client.post("/internal/report", json=report, headers=worker_headers(w, g)) + ).status_code == 204 + assert ( + await client.post( + "/internal/report", + json=synthetic_report(g, m, "wrong_answer"), + headers=worker_headers(w, g), + ) + ).status_code == 409 + assert ( + await c.fetchval("SELECT state FROM evaluations WHERE id=$1", created.evaluation_id) + == "finished" + ) + assert await c.fetchval("SELECT count(*) FROM case_results") == 2 + assert ( + await client.post("/internal/claim", json=n, headers=worker_headers(w)) + ).status_code == 204 + + queued = await create() + await c.execute( + "UPDATE outbox SET delivered_at=clock_timestamp()-interval '31 seconds' " + "WHERE evaluation_id=$1", + queued.evaluation_id, + ) + for _ in range(60): + if await c.fetchval( + "SELECT delivered_at IS NULL FROM outbox WHERE evaluation_id=$1", + queued.evaluation_id, + ): + break + await asyncio.sleep(0.05) + else: + raise AssertionError("missing notification not compensated") + assert ( + await c.fetchval("SELECT state FROM evaluations WHERE id=$1", queued.evaluation_id) + == "queued" + ) + await claim(client, queued, w) + await c.execute(f"ALTER ROLE {client.coordinator_role} NOLOGIN") + assert (await client.get("/health/ready")).status_code == 503 + assert (await client.get("/health/live")).status_code == 200 + await c.execute(f"ALTER ROLE {client.coordinator_role} LOGIN") + + asyncio.run(run()) + + +def test_actual_report_transaction_rollback(postgres_url, tmp_path, monkeypatch): + async def run(): + async with fixture(postgres_url, tmp_path, monkeypatch) as (client, c, workers, _, create): + created = await create() + w = workers[0] + g = await claim(client, created, w) + await stages(client, w, g) + m = await manifest(c, g) + await c.execute("""CREATE FUNCTION test_case_reject() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN IF NEW.position=1 THEN + RAISE EXCEPTION 'synthetic rollback'; + END IF; RETURN NEW; END $$""") + await c.execute( + "CREATE TRIGGER test_case_reject BEFORE INSERT ON case_results " + "FOR EACH ROW EXECUTE FUNCTION test_case_reject()" + ) + r = await client.post( + "/internal/report", json=synthetic_report(g, m), headers=worker_headers(w, g) + ) + assert r.status_code == 503, r.text + assert await c.fetchval("SELECT count(*) FROM case_results") == 0 + assert await c.fetchval("SELECT report FROM judge_attempts") is None + assert await c.fetchval("SELECT state FROM evaluations") == "running" + await c.execute("DROP TRIGGER test_case_reject ON case_results") + await c.execute("DROP FUNCTION test_case_reject()") + assert ( + await client.post( + "/internal/report", json=synthetic_report(g, m), headers=worker_headers(w, g) + ) + ).status_code == 204 + + asyncio.run(run()) + + +def test_actual_expiry_stopping_quarantine_retry_and_old_attempt( + postgres_url, tmp_path, monkeypatch +): + async def run(): + async with fixture(postgres_url, tmp_path, monkeypatch) as (client, c, workers, _, create): + created = await create() + other = await create() + w = workers[0] + g = await claim(client, created, w) + # Expired server-side lease (not a client clock claim) causes actual recovery scan. + await c.execute( + "UPDATE judge_attempts SET expires_at=clock_timestamp()-interval '1 second'" + ) + assert ( + await client.post( + "/internal/renew", json=g["binding"], headers=worker_headers(w, g) + ) + ).status_code == 409 + for _ in range(40): + pending = ( + await client.get("/internal/pending-stops", headers=worker_headers(w)) + ).json() + if pending: + break + await asyncio.sleep(0.05) + assert pending[0]["binding"] == g["binding"] + stopped = dict( + schema_version=1, kind="judge.stopped", binding=g["binding"], cleanup_complete=False + ) + assert ( + await client.post("/internal/stopped", json=stopped, headers=worker_headers(w)) + ).status_code == 204 + assert ( + await c.fetchval( + "SELECT failure_code FROM evaluations WHERE id=$1", created.evaluation_id + ) + == "cleanup_failed" + ) + assert ( + await client.post( + "/internal/claim", json=notice(other.evaluation_id), headers=worker_headers(w) + ) + ).status_code == 429 + stopped["cleanup_complete"] = True + # Synthetic stop assertion only tests DB state. Actual Root proof is required in VM. + assert ( + await client.post("/internal/stopped", json=stopped, headers=worker_headers(w)) + ).status_code == 204 + assert ( + await client.post("/internal/stopped", json=stopped, headers=worker_headers(w)) + ).status_code == 204 + assert ( + await c.fetchval("SELECT state FROM evaluations WHERE id=$1", created.evaluation_id) + == "queued" + ) + await c.execute( + "UPDATE evaluations SET not_before=clock_timestamp() WHERE id=$1", + created.evaluation_id, + ) + fresh = await claim(client, created, w) + assert fresh["attempt_number"] == 2 and fresh["binding"]["fencing_version"] == 2 + assert ( + await client.post( + "/internal/stage", + json=dict( + schema_version=1, + kind="judge.stage", + binding=g["binding"], + sequence=1, + stage="compiling", + case_index=None, + ), + headers=worker_headers(w, g), + ) + ).status_code == 409 + assert ( + await client.post("/internal/stopped", json=stopped, headers=worker_headers(w)) + ).status_code == 409 + + asyncio.run(run()) + + +def test_actual_user_cancel_all_stages_and_bounded_infrastructure_retry( + postgres_url, tmp_path, monkeypatch +): + async def run(): + async with fixture(postgres_url, tmp_path, monkeypatch) as ( + client, + c, + workers, + platform, + create, + ): + w = workers[0] + for stage in ["queued", "compiling", "running"]: + created = await create() + g = None + if stage != "queued": + g = await claim(client, created, w) + r = dict( + schema_version=1, + kind="judge.stage", + binding=g["binding"], + sequence=1, + stage="compiling", + case_index=None, + ) + assert ( + await client.post("/internal/stage", json=r, headers=worker_headers(w, g)) + ).status_code == 204 + if stage == "running": + r |= dict(sequence=2, stage="running", case_index=0) + assert ( + await client.post( + "/internal/stage", json=r, headers=worker_headers(w, g) + ) + ).status_code == 204 + assert ( + await client.post( + "/internal/cancel", + json=notice(created.evaluation_id), + headers={"Authorization": "Bearer " + platform}, + ) + ).status_code == 204 + if g: + assert ( + await c.fetchval( + "SELECT state FROM evaluations WHERE id=$1", created.evaluation_id + ) + == "cancelling" + ) + assert ( + await client.post( + "/internal/renew", json=g["binding"], headers=worker_headers(w, g) + ) + ).status_code == 409 + r = dict( + schema_version=1, + kind="judge.stopped", + binding=g["binding"], + cleanup_complete=True, + ) + assert ( + await client.post("/internal/stopped", json=r, headers=worker_headers(w)) + ).status_code == 204 + assert ( + await c.fetchval( + "SELECT state FROM evaluations WHERE id=$1", created.evaluation_id + ) + == "cancelled" + ) + created = await create() + for number in range(1, 4): + g = await claim(client, created, w) + m = await manifest(c, g) + report = synthetic_report(g, m) | dict( + compilation=dict( + status="system_error", + program=None, + diagnostics="", + diagnostics_truncated=False, + cpu_time_us=None, + memory_peak_bytes=None, + ), + cases=[], + verdict="system_error", + failure_code="sandbox_unavailable", + run_cpu_time_us=None, + run_memory_peak_bytes=None, + ) + assert ( + await client.post("/internal/report", json=report, headers=worker_headers(w, g)) + ).status_code == 204 + state = await c.fetchval( + "SELECT state FROM evaluations WHERE id=$1", created.evaluation_id + ) + assert state == ("failed" if number == 3 else "queued") + if state == "queued": + await c.execute( + "UPDATE evaluations SET not_before=clock_timestamp() WHERE id=$1", + created.evaluation_id, + ) + assert ( + await client.post( + "/internal/claim", json=notice(created.evaluation_id), headers=worker_headers(w) + ) + ).status_code == 204 + assert ( + await c.fetchval( + "SELECT count(*) FROM judge_attempts WHERE evaluation_id=$1", + created.evaluation_id, + ) + == 3 + ) + + asyncio.run(run()) diff --git a/contracts/schemas/judge-stopped.v1.schema.json b/contracts/schemas/judge-stopped.v1.schema.json new file mode 100644 index 0000000..9e56902 --- /dev/null +++ b/contracts/schemas/judge-stopped.v1.schema.json @@ -0,0 +1,69 @@ +{ + "$comment": "Structural schema; also run versioned model validation for cross-field budgets, result consistency, UUID canonical form and UTF-8 byte limits.", + "$defs": { + "LeaseBinding": { + "additionalProperties": false, + "properties": { + "attempt_id": { + "format": "uuid", + "title": "Attempt Id", + "type": "string" + }, + "evaluation_id": { + "format": "uuid", + "title": "Evaluation Id", + "type": "string" + }, + "fencing_version": { + "maximum": 9223372036854775807, + "minimum": 1, + "title": "Fencing Version", + "type": "integer" + }, + "worker_id": { + "format": "uuid", + "title": "Worker Id", + "type": "string" + } + }, + "required": [ + "evaluation_id", + "attempt_id", + "worker_id", + "fencing_version" + ], + "title": "LeaseBinding", + "type": "object" + } + }, + "$id": "urn:kf-oj:judge-stopped:v1", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": false, + "properties": { + "binding": { + "$ref": "#/$defs/LeaseBinding" + }, + "cleanup_complete": { + "title": "Cleanup Complete", + "type": "boolean" + }, + "kind": { + "const": "judge.stopped", + "title": "Kind", + "type": "string" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + } + }, + "required": [ + "schema_version", + "kind", + "binding", + "cleanup_complete" + ], + "title": "StoppedReport", + "type": "object" +} diff --git a/judge/Cargo.lock b/judge/Cargo.lock index 59f5513..45f72a9 100644 --- a/judge/Cargo.lock +++ b/judge/Cargo.lock @@ -61,6 +61,17 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -119,6 +130,12 @@ dependencies = [ "tracing", ] +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + [[package]] name = "bitflags" version = "2.13.2" @@ -134,12 +151,27 @@ dependencies = [ "generic-array", ] +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + [[package]] name = "bumpalo" version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.12.1" @@ -152,6 +184,17 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.1", + "rand_core 0.10.1", +] + [[package]] name = "clap" version = "4.6.7" @@ -192,12 +235,24 @@ version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c133bc6a41be0d194c306b5506d15e6feeea7b1d6604bd3f8310dfb2ca96486" +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + [[package]] name = "colorchoice" version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -207,6 +262,15 @@ dependencies = [ "libc", ] +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + [[package]] name = "crypto-common" version = "0.1.7" @@ -217,14 +281,44 @@ dependencies = [ "typenum", ] +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "ctutils" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03bb0e1cc970d482d121d9a1744999169b69a07470b3d644a7894e53fcaf4574" +dependencies = [ + "cmov", +] + [[package]] name = "digest" version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ - "block-buffer", - "crypto-common", + "block-buffer 0.10.4", + "crypto-common 0.1.7", +] + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer 0.12.1", + "const-oid", + "crypto-common 0.2.2", + "ctutils", ] [[package]] @@ -237,6 +331,12 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "fallible-iterator" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -253,6 +353,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -261,6 +362,12 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -274,6 +381,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ "futures-core", + "futures-sink", "futures-task", "pin-project-lite", "slab", @@ -289,12 +397,45 @@ dependencies = [ "version_check", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi 5.3.0", + "wasip2", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi 6.0.0", + "rand_core 0.10.1", +] + [[package]] name = "heck" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest 0.11.3", +] + [[package]] name = "http" version = "1.5.0" @@ -340,6 +481,15 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "hybrid-array" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" +dependencies = [ + "typenum", +] + [[package]] name = "hyper" version = "1.11.1" @@ -415,9 +565,13 @@ dependencies = [ "clap", "kf-contracts", "serde", + "serde_json", + "sha2 0.10.9", "tokio", + "tokio-postgres", "tracing", "tracing-subscriber", + "uuid", ] [[package]] @@ -440,7 +594,7 @@ dependencies = [ "seccompiler", "serde", "serde_json", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -453,7 +607,7 @@ dependencies = [ "kf-sandbox", "serde", "serde_json", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -468,12 +622,33 @@ version = "0.2.190" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" +[[package]] +name = "libredox" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61ff90caf6077a803a240f62fdbe88645a890bbca49ef8174c3cb0404362171d" +dependencies = [ + "bitflags", + "libc", + "plain", + "redox_syscall 0.9.4", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.34" @@ -495,6 +670,16 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.3", +] + [[package]] name = "memchr" version = "2.8.3" @@ -539,18 +724,107 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall 0.5.18", + "smallvec", + "windows-link", +] + [[package]] name = "percent-encoding" version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "phf" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1562dc717473dbaa4c1f85a36410e03c047b2e7df7f45ee938fbef64ae7fadf" +dependencies = [ + "phf_shared", + "serde", +] + +[[package]] +name = "phf_shared" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e57fef6bc5981e38c2ce2d63bfa546861309f875b8a75f092d1d54ae2d64f266" +dependencies = [ + "siphasher", +] + [[package]] name = "pin-project-lite" version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "plain" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" + +[[package]] +name = "postgres-protocol" +version = "0.6.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514" +dependencies = [ + "base64", + "byteorder", + "bytes", + "fallible-iterator", + "hmac", + "md-5", + "memchr", + "rand 0.10.3", + "sha2 0.11.0", + "stringprep", +] + +[[package]] +name = "postgres-types" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "851ca9db4932932d69f3ea811b1abe63087a0f740a47692619dd40d4899b68be" +dependencies = [ + "bytes", + "fallible-iterator", + "postgres-protocol", + "serde_core", + "serde_json", + "uuid", +] + +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "proc-macro2" version = "1.0.107" @@ -569,6 +843,82 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + +[[package]] +name = "rand" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "redox_syscall" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "737970939a87c6fa31e7acad13307bccbb017a073b695b6089a2c484f929e20e" +dependencies = [ + "bitflags", +] + [[package]] name = "regex-automata" version = "0.4.18" @@ -611,6 +961,12 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "seccompiler" version = "0.5.0" @@ -693,8 +1049,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", - "digest", + "cpufeatures 0.2.17", + "digest 0.10.7", +] + +[[package]] +name = "sha2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.1", + "digest 0.11.3", ] [[package]] @@ -716,6 +1083,12 @@ dependencies = [ "libc", ] +[[package]] +name = "siphasher" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33f4fe9184a62d842c9ef383018f3306d8ba224fd9d836f56d7288308847c256" + [[package]] name = "slab" version = "0.4.12" @@ -738,6 +1111,17 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "stringprep" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1" +dependencies = [ + "unicode-bidi", + "unicode-normalization", + "unicode-properties", +] + [[package]] name = "strsim" version = "0.11.1" @@ -781,12 +1165,19 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "tinyvec" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee" + [[package]] name = "tokio" version = "1.53.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044" dependencies = [ + "bytes", "libc", "mio", "pin-project-lite", @@ -807,6 +1198,46 @@ dependencies = [ "syn 3.0.6", ] +[[package]] +name = "tokio-postgres" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b40d66d9b2cfe04b628173409368e58247e8eddbbd3b0e6c6ba1d09f20f6c9e" +dependencies = [ + "async-trait", + "byteorder", + "bytes", + "fallible-iterator", + "futures-channel", + "futures-util", + "log", + "parking_lot", + "percent-encoding", + "phf", + "pin-project-lite", + "postgres-protocol", + "postgres-types", + "rand 0.9.5", + "socket2", + "tokio", + "tokio-util", + "whoami", +] + +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + [[package]] name = "tower" version = "0.5.3" @@ -916,12 +1347,33 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "unicode-bidi" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" + [[package]] name = "unicode-ident" version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" +[[package]] +name = "unicode-normalization" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fd4f6878c9cb28d874b009da9e8d183b5abc80117c40bbd187a1fde336be6e8" +dependencies = [ + "tinyvec", +] + +[[package]] +name = "unicode-properties" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" + [[package]] name = "utf8parse" version = "0.2.2" @@ -934,6 +1386,7 @@ version = "1.27.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97277d36b9c3ace13e58fa6e753f8b0bbbf302a18dd193240d70f9e29681059a" dependencies = [ + "getrandom 0.4.3", "js-sys", "serde_core", "wasm-bindgen", @@ -957,6 +1410,21 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasip2" +version = "1.0.4+wasi-0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" +dependencies = [ + "wit-bindgen", +] + +[[package]] +name = "wasite" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b" + [[package]] name = "wasm-bindgen" version = "0.2.129" @@ -1002,6 +1470,27 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "web-sys" +version = "0.3.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88261b9deccee56594c11a3460c462c41f58d148598fe70ad77070126a68aba4" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "whoami" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d4a4db5077702ca3015d3d02d74974948aba2ad9e12ab7df718ee64ccd7e97d" +dependencies = [ + "libredox", + "wasite", + "web-sys", +] + [[package]] name = "windows-link" version = "0.2.1" @@ -1017,6 +1506,32 @@ dependencies = [ "windows-link", ] +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + +[[package]] +name = "zerocopy" +version = "0.8.62" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "86502bf56ac7c77571a32e2647bb2a15894565e981fb2a48d7bde2d91c965a9d" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.62" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5457206954b06561e2608c7e19cf58b1926586d999c246eebe4502f7e2039d1a" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/judge/crates/contracts/src/judging.rs b/judge/crates/contracts/src/judging.rs index 386671d..c28450f 100644 --- a/judge/crates/contracts/src/judging.rs +++ b/judge/crates/contracts/src/judging.rs @@ -91,6 +91,7 @@ kind!(ManifestKind, "judge.manifest"); kind!(LeaseKind, "judge.lease"); kind!(StageKind, "judge.stage"); kind!(ResultKind, "judge.result"); +kind!(StoppedKind, "judge.stopped"); kind!(ImageKind, "linux_gnu_cxx_x86_64_v1"); kind!(Comparison, "exact_bytes"); @@ -409,6 +410,19 @@ impl StageReport { Ok(()) } } +// A trusted ordinary Worker attests actual Root stop/recovery proof. No execution +// result is accepted through this path; current attempt/fence still bind it. +validated_struct!(StoppedReport { + schema_version: SchemaVersion, + kind: StoppedKind, + binding: LeaseBinding, + cleanup_complete: bool, +}); +impl StoppedReport { + pub fn validate(&self) -> Validation { + self.binding.validate() + } +} validated_struct!(CompilationReport { status: CompilationStatus, #[serde(deserialize_with = "required_option")] program: Option, diff --git a/judge/crates/contracts/src/state.rs b/judge/crates/contracts/src/state.rs index 776d27f..3ce9226 100644 --- a/judge/crates/contracts/src/state.rs +++ b/judge/crates/contracts/src/state.rs @@ -23,7 +23,7 @@ impl EvaluationState { | (Claimed, Compiling | Cancelling | Failed | Queued) | (Compiling, Running | Finished | Cancelling | Failed | Queued) | (Running, Finished | Cancelling | Failed | Queued) - | (Cancelling, Cancelled | Failed) + | (Cancelling, Cancelled | Failed | Queued) ) } } diff --git a/judge/crates/contracts/tests/judging_examples.rs b/judge/crates/contracts/tests/judging_examples.rs index ea1fc8d..9550a46 100644 --- a/judge/crates/contracts/tests/judging_examples.rs +++ b/judge/crates/contracts/tests/judging_examples.rs @@ -51,6 +51,8 @@ fn shared_judging_examples_match_python() { roundtrip::(&bytes, invalid, &name); } else if name.starts_with("judge-result.") { roundtrip::(&bytes, invalid, &name); + } else if name.starts_with("judge-stopped.") { + roundtrip::(&bytes, invalid, &name); } else if name.starts_with("judge-admission.") { roundtrip::(&bytes, invalid, &name); } else { diff --git a/judge/crates/coordinator/Cargo.toml b/judge/crates/coordinator/Cargo.toml index 66853bc..ae868d7 100644 --- a/judge/crates/coordinator/Cargo.toml +++ b/judge/crates/coordinator/Cargo.toml @@ -11,8 +11,12 @@ axum.workspace = true tokio.workspace = true clap.workspace = true serde.workspace = true +serde_json.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +tokio-postgres = { version = "=0.7.15", features = ["with-serde_json-1", "with-uuid-1"] } +sha2 = "=0.10.9" +uuid = { version = "1", features = ["v4", "serde"] } [lints] workspace = true diff --git a/judge/crates/coordinator/README.md b/judge/crates/coordinator/README.md new file mode 100644 index 0000000..d58adbb --- /dev/null +++ b/judge/crates/coordinator/README.md @@ -0,0 +1,21 @@ +# 持久化协调器 + +`kf-coordinator --config /etc/kf-judge/coordinator.json` 读取 0600、有界、单链接 +私有配置。`listen` 为 loopback 地址，`database_url` 仅数值 loopback 或绝对 Unix socket， +使用继承 `kf_oj_coordinator` 的普通数据库角色；Backend Alembic 是唯一迁移入口。 + +`workers` 配置各 Worker 的 UUID 与独立64位十六进制 `token`，`platform_token` +与它们不同；同时配置可信 `admission`、`lease_seconds`（5–300）、 +`recovery_seconds`（1–60）。内部身份与浏览器会话、短期执行租约分离。 +新增 `tokio-postgres 0.7.15` 负责有限数据库连接和事务，`sha2 0.10.9` 绑定摘要。 +依赖方向仍为 `coordinator → contracts`，不调用沙箱或维护另一套 migration。 + +`/health/live` 表示存活，`/health/ready` 核对数据库和角色权限。内部接口先认证， +请求有体积/期限/并发上限；claim 原子生成 attempt、可信数据库时间租约与递增 fence， +提交后才返回领取权。阶段更新、续租和报告都核对当前绑定与令牌；报告的用例、attempt +和终态同一事务提交，原样重复可确认，修改/旧 attempt 拒绝覆盖。 + +取消与过期先撤销执行权并进入 `cancelling`；实际清理证明到达前不释放容量。 +清理失败保留 `cleanup_failed`，只有已确认停止的基础设施失败可按有限预算重试。 +恢复扫描每批最多16项，待领取事件投递30秒后仍无执行权时补偿重新通知。 +Worker 的实际停止证明与执行隔离需独立 VM 验收；普通数据库测试不代表 Linux 安全验收。 diff --git a/judge/crates/coordinator/src/api.rs b/judge/crates/coordinator/src/api.rs index 4c0d7e7..25b5012 100644 --- a/judge/crates/coordinator/src/api.rs +++ b/judge/crates/coordinator/src/api.rs @@ -1,21 +1,211 @@ -use axum::{Json, Router, routing::get}; -use serde::Serialize; - -#[derive(Serialize)] -struct HealthResponse { - status: &'static str, - service: &'static str, - version: &'static str, +use crate::{config::Config, database::Error, repository::Repository}; +use axum::{ + Json, Router, + body::Bytes, + extract::{DefaultBodyLimit, Request, State}, + http::{HeaderMap, StatusCode}, + middleware::{self, Next}, + response::{IntoResponse, Response}, + routing::{get, post}, +}; +use kf_contracts::{JudgeTaskNotification, judging::*}; +use serde_json::{Value, json}; +use std::{sync::Arc, time::Duration}; +use tokio::sync::Semaphore; +#[derive(Clone)] +pub struct App { + pub repository: Repository, + pub config: Arc, + slots: Arc, } - -async fn liveness() -> Json { - Json(HealthResponse { - status: "ok", - service: "coordinator", - version: env!("CARGO_PKG_VERSION"), +impl App { + pub fn new(repository: Repository, config: Config) -> Self { + Self { + repository, + config: Arc::new(config), + slots: Arc::new(Semaphore::new(16)), + } + } +} +type ApiResult = Result; +pub struct ApiError(Error); +impl From for ApiError { + fn from(e: Error) -> Self { + Self(e) + } +} +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + let (status, code) = match self.0 { + Error::Busy => (StatusCode::TOO_MANY_REQUESTS, "capacity"), + Error::Unavailable => (StatusCode::SERVICE_UNAVAILABLE, "unavailable"), + Error::Rejected => (StatusCode::CONFLICT, "stale_or_foreign_authority"), + Error::Invalid => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_request"), + }; + (status, Json(json!({"error":code}))).into_response() + } +} +fn token(headers: &HeaderMap) -> Option<&str> { + headers + .get("authorization")? + .to_str() + .ok()? + .strip_prefix("Bearer ") +} +fn equal(a: &str, b: &str) -> bool { + a.len() == 64 + && b.len() == 64 + && a.as_bytes() + .iter() + .zip(b.as_bytes()) + .fold(0u8, |v, (a, b)| v | (a ^ b)) + == 0 +} +fn worker(app: &App, h: &HeaderMap) -> ApiResult { + let token = token(h).ok_or(ApiError(Error::Rejected))?; + app.config + .workers + .iter() + .find(|w| equal(token, w.token.expose_secret())) + .map(|w| w.id) + .ok_or(ApiError(Error::Rejected)) +} +fn platform(app: &App, h: &HeaderMap) -> ApiResult<()> { + if token(h).is_some_and(|v| equal(v, app.config.platform_token.expose_secret())) { + Ok(()) + } else { + Err(ApiError(Error::Rejected)) + } +} +fn lease(h: &HeaderMap) -> ApiResult<&str> { + let v = h + .get("x-lease-token") + .and_then(|v| v.to_str().ok()) + .ok_or(ApiError(Error::Rejected))?; + if v.len() != 64 { + return Err(ApiError(Error::Rejected)); + } + Ok(v) +} +fn decode(body: &Bytes) -> ApiResult { + decode_wire(body).map_err(|_| ApiError(Error::Invalid)) +} +fn bound_worker(app: &App, h: &HeaderMap, b: &LeaseBinding) -> ApiResult<()> { + if worker(app, h)? != b.worker_id { + return Err(ApiError(Error::Rejected)); + } + Ok(()) +} +async fn guard(State(app): State, req: Request, next: Next) -> Response { + if req.uri().path().starts_with("/internal/") + && !(worker(&app, req.headers()).is_ok() || platform(&app, req.headers()).is_ok()) + { + return ( + StatusCode::UNAUTHORIZED, + Json(json!({"error":"service_identity_required"})), + ) + .into_response(); + } + let Ok(_slot) = app.slots.clone().try_acquire_owned() else { + return ApiError(Error::Busy).into_response(); + }; + match tokio::time::timeout(Duration::from_secs(10), next.run(req)).await { + Ok(mut response) => { + response.headers_mut().insert( + "cache-control", + "no-store".parse().expect("constant header"), + ); + response + } + Err(_) => ( + StatusCode::REQUEST_TIMEOUT, + Json(json!({"error":"request_deadline"})), + ) + .into_response(), + } +} +async fn live() -> Json { + Json(json!({"status":"ok","service":"coordinator","version":env!("CARGO_PKG_VERSION")})) +} +async fn ready(State(app): State) -> ApiResult> { + app.repository.database.ready().await?; + Ok(Json(json!({"status":"ready"}))) +} +async fn claim(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult { + let w = worker(&app, &h)?; + let n: JudgeTaskNotification = decode(&b)?; + let eid = n + .evaluation_id + .to_string() + .try_into() + .map_err(|_| ApiError(Error::Invalid))?; + Ok(match app.repository.claim(eid, w).await? { + Some(g) => Json(g).into_response(), + None => StatusCode::NO_CONTENT.into_response(), }) } - -pub fn router() -> Router { - Router::new().route("/health/live", get(liveness)) +async fn renew(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult> { + let binding: LeaseBinding = decode(&b)?; + bound_worker(&app, &h, &binding)?; + Ok(Json( + json!({"expires_at_unix_ms":app.repository.renew(&binding,lease(&h)?).await?}), + )) +} +async fn progress(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult> { + let binding: LeaseBinding = decode(&b)?; + bound_worker(&app, &h, &binding)?; + Ok(Json(app.repository.progress(&binding, lease(&h)?).await?)) +} +async fn stage(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult { + let r: StageReport = decode(&b)?; + bound_worker(&app, &h, &r.binding)?; + app.repository.stage(&r, lease(&h)?).await?; + Ok(StatusCode::NO_CONTENT) +} +async fn report(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult { + let r: JudgementReport = decode(&b)?; + bound_worker(&app, &h, &r.binding)?; + app.repository.report(&r, lease(&h)?).await?; + Ok(StatusCode::NO_CONTENT) +} +async fn stopped(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult { + let r: StoppedReport = decode(&b)?; + bound_worker(&app, &h, &r.binding)?; + app.repository.stopped(&r).await?; + Ok(StatusCode::NO_CONTENT) +} +async fn pending(State(app): State, h: HeaderMap) -> ApiResult> { + Ok(Json(app.repository.pending_stops(worker(&app, &h)?).await?)) +} +async fn cancel(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult { + platform(&app, &h)?; + let n: JudgeTaskNotification = decode(&b)?; + app.repository + .cancel( + n.evaluation_id + .to_string() + .try_into() + .map_err(|_| ApiError(Error::Invalid))?, + ) + .await?; + Ok(StatusCode::NO_CONTENT) +} +pub fn router() -> Router { + Router::new().route("/health/live", get(live)) +} +pub fn application(app: App) -> Router { + Router::new() + .route("/health/live", get(live)) + .route("/health/ready", get(ready)) + .route("/internal/claim", post(claim)) + .route("/internal/renew", post(renew)) + .route("/internal/stage", post(stage)) + .route("/internal/progress", post(progress)) + .route("/internal/report", post(report)) + .route("/internal/stopped", post(stopped)) + .route("/internal/pending-stops", get(pending)) + .route("/internal/cancel", post(cancel)) + .layer(DefaultBodyLimit::max(MAX_WIRE_BYTES)) + .layer(middleware::from_fn_with_state(app.clone(), guard)) + .with_state(app) } diff --git a/judge/crates/coordinator/src/config.rs b/judge/crates/coordinator/src/config.rs new file mode 100644 index 0000000..e8c139e --- /dev/null +++ b/judge/crates/coordinator/src/config.rs @@ -0,0 +1,56 @@ +//! Service-only secrets from a private local file; never format database/token values. +use kf_contracts::judging::{AdmissionPolicy, Identifier, LeaseToken}; +use serde::Deserialize; +use std::{collections::HashSet, fs, io, net::SocketAddr, path::Path}; +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub struct WorkerIdentity { + pub id: Identifier, + pub token: LeaseToken, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Config { + pub listen: SocketAddr, + pub database_url: String, + pub workers: Vec, + pub platform_token: LeaseToken, + pub admission: AdmissionPolicy, + pub lease_seconds: u32, + pub recovery_seconds: u32, +} +impl Config { + pub fn load(path: &Path) -> io::Result { + let meta = fs::symlink_metadata(path)?; + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + if !meta.is_file() + || meta.nlink() != 1 + || meta.permissions().mode() & 0o777 != 0o600 + || meta.len() > 32768 + { + return Err(io::Error::other( + "private coordinator configuration required", + )); + } + let v: Self = serde_json::from_slice(&fs::read(path)?) + .map_err(|_| io::Error::other("invalid coordinator configuration"))?; + if !v.listen.ip().is_loopback() + || !(5..=300).contains(&v.lease_seconds) + || !(1..=60).contains(&v.recovery_seconds) + || v.workers.is_empty() + || v.workers.len() > 64 + || v.workers.iter().map(|w| w.id).collect::>().len() != v.workers.len() + || v.workers + .iter() + .any(|w| w.token.expose_secret() == v.platform_token.expose_secret()) + { + return Err(io::Error::other("invalid coordinator policy")); + } + let tokens: HashSet<_> = v.workers.iter().map(|w| w.token.expose_secret()).collect(); + if tokens.len() != v.workers.len() { + return Err(io::Error::other("ambiguous worker identity")); + } + v.admission.validate().map_err(io::Error::other)?; + Ok(v) + } +} diff --git a/judge/crates/coordinator/src/database.rs b/judge/crates/coordinator/src/database.rs new file mode 100644 index 0000000..a1a4264 --- /dev/null +++ b/judge/crates/coordinator/src/database.rs @@ -0,0 +1,90 @@ +//! Finite connection slots and fixed SQL timeouts. Backend Alembic owns schema. +use std::{net::IpAddr, sync::Arc, time::Duration}; +use tokio::{ + sync::{OwnedSemaphorePermit, Semaphore}, + task::JoinHandle, +}; +use tokio_postgres::{Client, Config, NoTls, config::Host}; +#[derive(Debug, Clone, Copy)] +pub enum Error { + Unavailable, + Busy, + Rejected, + Invalid, +} +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("coordinator operation refused") + } +} +impl std::error::Error for Error {} +impl From for Error { + fn from(_: tokio_postgres::Error) -> Self { + Self::Unavailable + } +} +pub type Result = std::result::Result; +#[derive(Clone)] +pub struct Database { + config: Config, + slots: Arc, +} +pub struct Session { + pub client: Client, + task: JoinHandle<()>, + _slot: OwnedSemaphorePermit, +} +impl Drop for Session { + fn drop(&mut self) { + self.task.abort(); + } +} +impl Database { + pub fn new(url: &str) -> Result { + let mut config: Config = url.parse().map_err(|_| Error::Invalid)?; + if config.get_hosts().is_empty() + || config.get_hosts().iter().any(|h| match h { + Host::Tcp(n) => !n.parse::().is_ok_and(|ip| ip.is_loopback()), + Host::Unix(p) => !p.is_absolute(), + }) + { + return Err(Error::Invalid); + } + config.connect_timeout(Duration::from_secs(3)); + config.application_name("kf-coordinator"); + config.options("-c statement_timeout=5000 -c lock_timeout=2000 -c idle_in_transaction_session_timeout=10000"); + Ok(Self { + config, + slots: Arc::new(Semaphore::new(4)), + }) + } + pub async fn session(&self) -> Result { + let slot = self + .slots + .clone() + .try_acquire_owned() + .map_err(|_| Error::Busy)?; + let (client, connection) = self.config.connect(NoTls).await?; + let task = tokio::spawn(async move { + let _ = connection.await; + }); + let session = Session { + client, + task, + _slot: slot, + }; + let row=session.client.query_one("SELECT NOT r.rolsuper AND NOT r.rolbypassrls AND pg_has_role(current_user,'kf_oj_coordinator','MEMBER') AND NOT has_table_privilege(current_user,'accounts','SELECT') AND NOT has_table_privilege(current_user,'problems','UPDATE') FROM pg_roles r WHERE r.rolname=session_user",&[]).await?; + if !row.get::<_, bool>(0) { + return Err(Error::Rejected); + } + Ok(session) + } + pub async fn ready(&self) -> Result<()> { + self.session() + .await? + .client + .query("SELECT 1 FROM evaluations LIMIT 1", &[]) + .await?; + Ok(()) + } +} diff --git a/judge/crates/coordinator/src/lib.rs b/judge/crates/coordinator/src/lib.rs index 9f05be0..c7f5e34 100644 --- a/judge/crates/coordinator/src/lib.rs +++ b/judge/crates/coordinator/src/lib.rs @@ -1,6 +1,5 @@ -//! Internal HTTP process and future lease/state authority. -//! -//! Only process liveness is exposed now. SQLx repositories and task transitions -//! are added after DR-01/PF-01 define their shared schema and concurrency rules. - +//! Sole execution-state writer. The Backend owns migrations and initial rows. pub mod api; +pub mod config; +pub mod database; +pub mod repository; diff --git a/judge/crates/coordinator/src/main.rs b/judge/crates/coordinator/src/main.rs index 95d91ba..614d637 100644 --- a/judge/crates/coordinator/src/main.rs +++ b/judge/crates/coordinator/src/main.rs @@ -1,31 +1,66 @@ use clap::Parser; -use std::net::SocketAddr; +use kf_coordinator::{ + api::{App, application, router}, + config::Config, + database::Database, + repository::Repository, +}; +use std::{net::SocketAddr, path::PathBuf}; use tokio::net::TcpListener; -use tracing_subscriber::EnvFilter; - #[derive(Parser)] -#[command(version, about = "KF judge coordinator HTTP shell")] +#[command(version, about = "KF persistent judge coordinator")] struct Cli { + #[arg(long, env = "KF_COORDINATOR_CONFIG")] + config: Option, #[arg(long, env = "KF_COORDINATOR_LISTEN", default_value = "127.0.0.1:8100")] listen: SocketAddr, } - -#[tokio::main] -async fn main() -> Result<(), Box> { - tracing_subscriber::fmt() - .json() - .with_env_filter( - EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")), - ) - .init(); - let listener = TcpListener::bind(Cli::parse().listen).await?; - tracing::info!(address = %listener.local_addr()?, "coordinator scaffold listening"); - axum::serve(listener, kf_coordinator::api::router()) - .with_graceful_shutdown(async { - if let Err(error) = tokio::signal::ctrl_c().await { - tracing::error!(%error, "failed to listen for shutdown signal"); +async fn run() -> Result<(), Box> { + let cli = Cli::parse(); + let mut task = None; + let (listen, app) = if let Some(path) = cli.config { + let c = Config::load(&path)?; + let listen = c.listen; + let repository = Repository { + database: Database::new(&c.database_url)?, + admission: c.admission.clone(), + lease_seconds: c.lease_seconds, + }; + let r = repository.clone(); + let period = c.recovery_seconds; + task = Some(tokio::spawn(async move { + let mut tick = tokio::time::interval(std::time::Duration::from_secs(u64::from(period))); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tick.tick().await; + let _ = r.recover().await; } + })); + (listen, application(App::new(repository, c))) + } else { + (cli.listen, router()) + }; + let listener = TcpListener::bind(listen).await?; + axum::serve(listener, app) + .with_graceful_shutdown(async { + let mut term = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("signal registration"); + tokio::select! {_=tokio::signal::ctrl_c()=>{},_=term.recv()=>{}} }) .await?; + if let Some(t) = task { + t.abort(); + } Ok(()) } +#[tokio::main] +async fn main() -> std::process::ExitCode { + match run().await { + Ok(()) => std::process::ExitCode::SUCCESS, + Err(_) => { + eprintln!("coordinator unavailable"); + std::process::ExitCode::from(2) + } + } +} diff --git a/judge/crates/coordinator/src/repository.rs b/judge/crates/coordinator/src/repository.rs new file mode 100644 index 0000000..907d079 --- /dev/null +++ b/judge/crates/coordinator/src/repository.rs @@ -0,0 +1,436 @@ +//! Execution authority is applied under row locks, using database clock and fencing. +use crate::database::{Database, Error, Result}; +use kf_contracts::{SchemaVersion, judging::*}; +use serde_json::{Value, json}; +use sha2::{Digest as _, Sha256}; +use tokio_postgres::{Row, Transaction}; +use uuid::Uuid; +#[derive(Clone)] +pub struct Repository { + pub database: Database, + pub admission: AdmissionPolicy, + pub lease_seconds: u32, +} +pub fn identifier(v: Uuid) -> Identifier { + v.to_string().try_into().expect("non-nil generated UUID") +} +pub fn hash(data: &[u8]) -> String { + format!("{:x}", Sha256::digest(data)) +} +fn secret() -> String { + format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()) +} +fn invalid(_: impl std::fmt::Debug) -> Error { + Error::Invalid +} +fn as_value(v: &T) -> Result { + serde_json::to_value(v).map_err(invalid) +} +fn as_wire(v: Value) -> Result { + let b = serde_json::to_vec(&v).map_err(invalid)?; + decode_wire(&b).map_err(invalid) +} +impl Repository { + async fn manifest(&self, t: &Transaction<'_>, e: &Row) -> Result { + let sid: Uuid = e.get("submission_id"); + let eid: Uuid = e.get("id"); + let h=t.query_opt("SELECT s.id,s.problem_version_id,s.language,s.source_digest,a.bytes,v.toolchain,v.compile_limits,v.run_limits,v.job_wall_time_us FROM submissions s JOIN problem_versions v ON v.id=s.problem_version_id JOIN artifacts a ON a.digest=s.source_digest AND a.state='ready' WHERE s.id=$1 AND v.published_at IS NOT NULL",&[&sid]).await?.ok_or(Error::Invalid)?; + let vid: Uuid = h.get("problem_version_id"); + let rows=t.query("SELECT c.id,c.position,c.input_digest,c.expected_digest,i.bytes input_bytes,o.bytes expected_bytes FROM problem_cases c JOIN artifacts i ON i.digest=c.input_digest AND i.state='ready' JOIN artifacts o ON o.digest=c.expected_digest AND o.state='ready' WHERE c.version_id=$1 ORDER BY c.position LIMIT 257",&[&vid]).await?; + let mut cases = Vec::new(); + for (n, r) in rows.iter().enumerate() { + if r.get::<_, i32>("position") != n as i32 { + return Err(Error::Invalid); + } + cases.push(TestCase { + case_id: identifier(r.get("id")), + input: ArtifactReference { + digest: r + .get::<_, String>("input_digest") + .try_into() + .map_err(invalid)?, + bytes: r.get::<_, i64>("input_bytes") as u64, + }, + expected: ArtifactReference { + digest: r + .get::<_, String>("expected_digest") + .try_into() + .map_err(invalid)?, + bytes: r.get::<_, i64>("expected_bytes") as u64, + }, + }); + } + let m = JobManifest { + schema_version: SchemaVersion, + kind: ManifestKind::Value, + evaluation_id: identifier(eid), + submission_id: identifier(sid), + problem_version_id: identifier(vid), + language: as_wire(json!(h.get::<_, String>("language")))?, + toolchain: as_wire(h.get("toolchain"))?, + source: ArtifactReference { + digest: h + .get::<_, String>("source_digest") + .try_into() + .map_err(invalid)?, + bytes: h.get::<_, i64>("bytes") as u64, + }, + cases, + comparison: Comparison::Value, + compile_limits: as_wire(h.get("compile_limits"))?, + run_limits: as_wire(h.get("run_limits"))?, + job_wall_time_us: h.get::<_, i64>("job_wall_time_us") as u64, + }; + m.validate().map_err(invalid)?; + Ok(m) + } + pub async fn claim(&self, eid: Identifier, worker: Identifier) -> Result> { + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + // A fixed node-capacity lock is separate from platform publication/digest locks. + t.execute("SELECT pg_advisory_xact_lock(1262907471,0)", &[]) + .await?; + let e=match t.query_opt("SELECT *,not_before<=clock_timestamp() due FROM evaluations WHERE id=$1 FOR UPDATE",&[&eid.as_uuid()]).await?{Some(e)=>e,None=>return Ok(None)}; + let max = (e.get::<_, i32>("max_attempts") as u32).min(self.admission.max_retries + 1); + if e.get::<_, String>("state") != "queued" + || !e.get::<_, bool>("due") + || e.get::<_, Option>("cancel_requested_at") + .is_some() + || e.get::<_, i32>("attempts_count") as u32 >= max + { + return Ok(None); + } + let active:i64=t.query_one("SELECT count(*) FROM evaluations WHERE state IN ('claimed','compiling','running','cancelling')",&[]).await?.get(0); + if active >= i64::from(self.admission.max_concurrent_jobs) { + return Err(Error::Busy); + } + let m = self.manifest(&t, &e).await?; + let number = e.get::<_, i32>("attempts_count") + 1; + self.admission.admit(&m, number as u32).map_err(invalid)?; + let attempt = Uuid::new_v4(); + let token = secret(); + let fence = e + .get::<_, i64>("fencing_version") + .checked_add(1) + .ok_or(Error::Invalid)?; + let digest = hash(&manifest_bytes(&m).map_err(invalid)?); + let row=t.query_one("INSERT INTO judge_attempts(id,evaluation_id,worker_id,number,fencing_version,lease_token_hash,expires_at,manifest,manifest_digest) VALUES($1,$2,$3,$4,$5,$6,clock_timestamp()+$7::bigint*interval '1 second',$8,$9) RETURNING floor(extract(epoch from expires_at)*1000)::bigint",&[&attempt,&eid.as_uuid(),&worker.as_uuid(),&number,&fence,&hash(token.as_bytes()),&i64::from(self.lease_seconds),&as_value(&m)?,&digest]).await?; + t.execute("UPDATE evaluations SET state='claimed',current_attempt_id=$2,attempts_count=$3,fencing_version=$4,updated_at=clock_timestamp() WHERE id=$1",&[&eid.as_uuid(),&attempt,&number,&fence]).await?; + let grant = LeaseGrant { + schema_version: SchemaVersion, + kind: LeaseKind::Value, + binding: LeaseBinding { + evaluation_id: eid, + attempt_id: identifier(attempt), + worker_id: worker, + fencing_version: fence as u64, + }, + lease_token: serde_json::from_value(json!(token)).map_err(invalid)?, + attempt_number: number as u32, + expires_at_unix_ms: row.get::<_, i64>(0) as u64, + manifest_digest: digest.try_into().map_err(invalid)?, + }; + t.commit().await?; + Ok(Some(grant)) + } + async fn bound( + &self, + t: &Transaction<'_>, + b: &LeaseBinding, + token: Option<&str>, + ) -> Result { + b.validate().map_err(invalid)?; + let row=t.query_opt("SELECT e.*,a.worker_id,a.fencing_version attempt_fence,a.expires_at>clock_timestamp() live,a.revoked_at,a.lease_token_hash,a.manifest,a.manifest_digest,a.sequence,a.stage,a.case_index,a.report_digest,a.cleanup_complete FROM evaluations e JOIN judge_attempts a ON a.id=e.current_attempt_id AND a.evaluation_id=e.id WHERE e.id=$1 AND a.id=$2 FOR UPDATE OF e,a",&[&b.evaluation_id.as_uuid(),&b.attempt_id.as_uuid()]).await?.ok_or(Error::Rejected)?; + if row.get::<_, Uuid>("worker_id") != b.worker_id.as_uuid() + || row.get::<_, i64>("fencing_version") as u64 != b.fencing_version + || row.get::<_, i64>("attempt_fence") as u64 != b.fencing_version + || token.is_some_and(|s| hash(s.as_bytes()) != row.get::<_, String>("lease_token_hash")) + { + return Err(Error::Rejected); + } + Ok(row) + } + fn live(row: &Row) -> Result<()> { + if !row.get::<_, bool>("live") + || row + .get::<_, Option>("revoked_at") + .is_some() + || row + .get::<_, Option>("cancel_requested_at") + .is_some() + || !matches!( + row.get::<_, String>("state").as_str(), + "claimed" | "compiling" | "running" + ) + { + Err(Error::Rejected) + } else { + Ok(()) + } + } + pub async fn renew(&self, b: &LeaseBinding, token: &str) -> Result { + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + let e = self.bound(&t, b, Some(token)).await?; + Self::live(&e)?; + let row=t.query_one("UPDATE judge_attempts SET expires_at=clock_timestamp()+$2::bigint*interval '1 second' WHERE id=$1 RETURNING floor(extract(epoch from expires_at)*1000)::bigint",&[&b.attempt_id.as_uuid(),&i64::from(self.lease_seconds)]).await?; + let time = row.get::<_, i64>(0) as u64; + t.commit().await?; + Ok(time) + } + pub async fn progress(&self, b: &LeaseBinding, token: &str) -> Result { + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + let e = self.bound(&t, b, Some(token)).await?; + Self::live(&e)?; + Ok( + json!({"state":e.get::<_,String>("state"),"sequence":e.get::<_,i64>("sequence"),"stage":e.get::<_,Option>("stage"),"case_index":e.get::<_,Option>("case_index")}), + ) + } + pub async fn stage(&self, report: &StageReport, token: &str) -> Result<()> { + report.validate().map_err(invalid)?; + let b = &report.binding; + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + let e = self.bound(&t, b, Some(token)).await?; + Self::live(&e)?; + let stage = match report.stage { + Stage::Compiling => "compiling", + Stage::Running => "running", + }; + let index = report.case_index.map(i32::from); + let sequence = e.get::<_, i64>("sequence"); + let old_stage = e.get::<_, Option>("stage"); + let old_index = e.get::<_, Option>("case_index"); + if report.sequence as i64 == sequence + && old_stage.as_deref() == Some(stage) + && old_index == index + { + return Ok(()); + } + if report.sequence as i64 != sequence.checked_add(1).ok_or(Error::Invalid)? + || match report.stage { + Stage::Compiling => e.get::<_, String>("state") != "claimed" || old_stage.is_some(), + Stage::Running => { + !matches!( + e.get::<_, String>("state").as_str(), + "compiling" | "running" + ) || index != Some(old_index.map_or(0, |v| v + 1)) + } + } + { + return Err(Error::Rejected); + } + let m: JobManifest = as_wire(e.get("manifest"))?; + if index.is_some_and(|i| i as usize >= m.cases.len()) { + return Err(Error::Rejected); + } + t.execute( + "UPDATE judge_attempts SET sequence=$2,stage=$3,case_index=$4 WHERE id=$1", + &[ + &b.attempt_id.as_uuid(), + &(report.sequence as i64), + &stage, + &index, + ], + ) + .await?; + t.execute( + "UPDATE evaluations SET state=$2,updated_at=clock_timestamp() WHERE id=$1", + &[&b.evaluation_id.as_uuid(), &stage], + ) + .await?; + t.commit().await?; + Ok(()) + } + async fn finish( + &self, + t: &Transaction<'_>, + e: &Row, + b: &LeaseBinding, + verdict: &str, + failure: Option<&str>, + clean: bool, + ) -> Result<()> { + let cancelled = e + .get::<_, Option>("cancel_requested_at") + .is_some(); + let retries = (e.get::<_, i32>("max_attempts") as u32).min(self.admission.max_retries + 1); + let retry = !cancelled + && clean + && verdict == "system_error" + && e.get::<_, i32>("attempts_count") < retries as i32; + let (state, v, code) = if !clean { + ("cancelling", "system_error", Some("cleanup_failed")) + } else if cancelled { + ("cancelled", "cancelled", None) + } else if retry { + ("queued", verdict, failure) + } else if verdict == "cancelled" { + ("cancelled", verdict, None) + } else if verdict == "system_error" { + ("failed", verdict, failure) + } else { + ("finished", verdict, None) + }; + let delay = 2_i64.pow((e.get::<_, i32>("attempts_count") as u32).min(5)); + t.execute("UPDATE judge_attempts SET cleanup_complete=$2,revoked_at=coalesce(revoked_at,clock_timestamp()) WHERE id=$1",&[&b.attempt_id.as_uuid(),&clean]).await?; + t.execute("UPDATE evaluations SET state=$2,verdict=$3,failure_code=$4,not_before=clock_timestamp()+$5::bigint*interval '1 second',updated_at=clock_timestamp() WHERE id=$1",&[&b.evaluation_id.as_uuid(),&state,&v,&code,&delay]).await?; + if retry { + t.execute("UPDATE evaluations SET verdict=NULL,failure_code=NULL,final_report=NULL,final_report_digest=NULL WHERE id=$1",&[&b.evaluation_id.as_uuid()]).await?; + t.execute("UPDATE outbox SET delivered_at=NULL,next_attempt_at=clock_timestamp()+$2::bigint*interval '1 second' WHERE evaluation_id=$1",&[&b.evaluation_id.as_uuid(),&delay]).await?; + } + Ok(()) + } + pub async fn report(&self, r: &JudgementReport, token: &str) -> Result<()> { + let b = &r.binding; + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + let e = self.bound(&t, b, Some(token)).await?; + let value = as_value(r)?; + let digest = hash(&serde_json::to_vec(&value).map_err(invalid)?); + if let Some(old) = e.get::<_, Option>("report_digest") { + if old == digest { + return Ok(()); + } else { + return Err(Error::Rejected); + } + } + Self::live(&e)?; + let m: JobManifest = as_wire(e.get("manifest"))?; + let md: Digest = e + .get::<_, String>("manifest_digest") + .try_into() + .map_err(invalid)?; + r.validate_for_manifest(&m, &md).map_err(invalid)?; + if r.compilation + .program + .as_ref() + .is_some_and(|p| p.bytes > m.compile_limits.file_bytes) + { + return Err(Error::Invalid); + } + if !matches!(r.verdict, Verdict::SystemError | Verdict::Cancelled) + && (e.get::<_, String>("state") == "claimed" + || (r.compilation.status == CompilationStatus::Succeeded + && e.get::<_, String>("state") != "running")) + { + return Err(Error::Rejected); + } + if !r.cases.is_empty() { + let index = e + .get::<_, Option>("case_index") + .ok_or(Error::Rejected)? as usize; + if r.cases.len() > index + 1 + || (!matches!(r.verdict, Verdict::SystemError | Verdict::Cancelled) + && r.cases.len() != index + 1) + { + return Err(Error::Rejected); + } + } + let vid = m.problem_version_id.as_uuid(); + for (position, c) in r.cases.iter().enumerate() { + let verdict = serde_json::to_value(c.verdict).map_err(invalid)?; + t.execute("INSERT INTO case_results(attempt_id,case_id,evaluation_id,problem_version_id,position,verdict,cpu_time_us,memory_peak_bytes) VALUES($1,$2,$3,$4,$5,$6,$7,$8)",&[&b.attempt_id.as_uuid(),&c.case_id.as_uuid(),&b.evaluation_id.as_uuid(),&vid,&(position as i32),&verdict.as_str().ok_or(Error::Invalid)?,&c.cpu_time_us.map(|v|v as i64),&c.memory_peak_bytes.map(|v|v as i64)]).await?; + } + t.execute( + "UPDATE judge_attempts SET report=$2,report_digest=$3 WHERE id=$1", + &[&b.attempt_id.as_uuid(), &value, &digest], + ) + .await?; + t.execute( + "UPDATE evaluations SET final_report=$2,final_report_digest=$3 WHERE id=$1", + &[&b.evaluation_id.as_uuid(), &value, &digest], + ) + .await?; + let verdict = serde_json::to_value(r.verdict).map_err(invalid)?; + let code = serde_json::to_value(r.failure_code).map_err(invalid)?; + self.finish( + &t, + &e, + b, + verdict.as_str().ok_or(Error::Invalid)?, + code.as_str(), + r.cleanup_complete, + ) + .await?; + t.commit().await?; + Ok(()) + } + pub async fn cancel(&self, eid: Identifier) -> Result<()> { + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + let e = t + .query_opt( + "SELECT state,current_attempt_id FROM evaluations WHERE id=$1 FOR UPDATE", + &[&eid.as_uuid()], + ) + .await? + .ok_or(Error::Rejected)?; + let state = e.get::<_, String>("state"); + if state == "queued" { + t.execute("UPDATE evaluations SET state='cancelled',verdict='cancelled',failure_code=NULL,final_report=NULL,final_report_digest=NULL,cancel_requested_at=clock_timestamp(),updated_at=clock_timestamp() WHERE id=$1",&[&eid.as_uuid()]).await?; + } else if matches!( + state.as_str(), + "claimed" | "compiling" | "running" | "cancelling" + ) { + t.execute("UPDATE evaluations SET state='cancelling',cancel_requested_at=coalesce(cancel_requested_at,clock_timestamp()),updated_at=clock_timestamp() WHERE id=$1",&[&eid.as_uuid()]).await?; + t.execute("UPDATE judge_attempts SET revoked_at=coalesce(revoked_at,clock_timestamp()) WHERE id=$1",&[&e.get::<_,Option>("current_attempt_id")]).await?; + } + t.commit().await?; + Ok(()) + } + pub async fn stopped(&self, r: &StoppedReport) -> Result<()> { + r.validate().map_err(invalid)?; + let b = &r.binding; + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + let e = self.bound(&t, b, None).await?; + if e.get::<_, String>("state") != "cancelling" { + if r.cleanup_complete && e.get::<_, Option>("cleanup_complete") == Some(true) { + return Ok(()); + } + return Err(Error::Rejected); + } + if e.get::<_, Option>("revoked_at") + .is_none() + { + return Err(Error::Rejected); + } + let code = e.get::<_, Option>("failure_code"); + self.finish( + &t, + &e, + b, + "system_error", + Some(code.as_deref().unwrap_or("lease_lost")), + r.cleanup_complete, + ) + .await?; + t.commit().await?; + Ok(()) + } + pub async fn pending_stops(&self, worker: Identifier) -> Result { + let s = self.database.session().await?; + let rows=s.client.query("SELECT e.id,a.id attempt_id,a.fencing_version,jsonb_array_length(a.manifest->'cases') case_count FROM evaluations e JOIN judge_attempts a ON a.id=e.current_attempt_id WHERE e.state='cancelling' AND a.worker_id=$1 ORDER BY e.updated_at,e.id LIMIT 16",&[&worker.as_uuid()]).await?; + Ok(Value::Array(rows.iter().map(|r|json!({"binding":LeaseBinding{evaluation_id:identifier(r.get("id")),attempt_id:identifier(r.get("attempt_id")),worker_id:worker,fencing_version:r.get::<_,i64>("fencing_version")as u64},"case_count":r.get::<_,i32>("case_count")})).collect())) + } + pub async fn recover(&self) -> Result { + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + let rows=t.query("SELECT e.id,e.current_attempt_id FROM evaluations e JOIN judge_attempts a ON a.id=e.current_attempt_id WHERE e.state IN ('claimed','compiling','running') AND (a.expires_at<=clock_timestamp() OR a.revoked_at IS NOT NULL) ORDER BY a.expires_at,e.id FOR UPDATE OF e SKIP LOCKED LIMIT 16",&[]).await?; + for r in &rows { + t.execute("UPDATE evaluations SET state='cancelling',failure_code='lease_lost',updated_at=clock_timestamp() WHERE id=$1",&[&r.get::<_,Uuid>("id")]).await?; + t.execute("UPDATE judge_attempts SET revoked_at=coalesce(revoked_at,clock_timestamp()) WHERE id=$1",&[&r.get::<_,Uuid>("current_attempt_id")]).await?; + } + // Re-notify queued rows if an earlier delivery was acknowledged without a claim. + // This is periodic bounded compensation, never execution authority or a busy loop. + let queued=t.query("SELECT e.id FROM evaluations e JOIN outbox o ON o.evaluation_id=e.id WHERE e.state='queued' AND e.not_before<=clock_timestamp() AND o.delivered_at("id")]).await?; + } + let count = (rows.len() + queued.len()) as u64; + t.commit().await?; + Ok(count) + } +} diff --git a/tests/fixtures/contracts/judge-state.transitions.json b/tests/fixtures/contracts/judge-state.transitions.json index 043353f..b51832d 100644 --- a/tests/fixtures/contracts/judge-state.transitions.json +++ b/tests/fixtures/contracts/judge-state.transitions.json @@ -24,7 +24,8 @@ ], "cancelling": [ "cancelled", - "failed" + "failed", + "queued" ], "finished": [], "cancelled": [], diff --git a/tests/fixtures/contracts/judge-stopped.invalid-cleanup-string.json b/tests/fixtures/contracts/judge-stopped.invalid-cleanup-string.json new file mode 100644 index 0000000..8716c81 --- /dev/null +++ b/tests/fixtures/contracts/judge-stopped.invalid-cleanup-string.json @@ -0,0 +1,11 @@ +{ + "schema_version": 1, + "kind": "judge.stopped", + "binding": { + "evaluation_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000001", + "attempt_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000007", + "worker_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000008", + "fencing_version": 1 + }, + "cleanup_complete": "true" +} diff --git a/tests/fixtures/contracts/judge-stopped.invalid-unknown.json b/tests/fixtures/contracts/judge-stopped.invalid-unknown.json new file mode 100644 index 0000000..6505498 --- /dev/null +++ b/tests/fixtures/contracts/judge-stopped.invalid-unknown.json @@ -0,0 +1,12 @@ +{ + "schema_version": 1, + "kind": "judge.stopped", + "binding": { + "evaluation_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000001", + "attempt_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000007", + "worker_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000008", + "fencing_version": 1 + }, + "cleanup_complete": true, + "command": "/bin/sh" +} diff --git a/tests/fixtures/contracts/judge-stopped.invalid-version.json b/tests/fixtures/contracts/judge-stopped.invalid-version.json new file mode 100644 index 0000000..a17c2b2 --- /dev/null +++ b/tests/fixtures/contracts/judge-stopped.invalid-version.json @@ -0,0 +1,11 @@ +{ + "schema_version": 2, + "kind": "judge.stopped", + "binding": { + "evaluation_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000001", + "attempt_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000007", + "worker_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000008", + "fencing_version": 1 + }, + "cleanup_complete": true +} diff --git a/tests/fixtures/contracts/judge-stopped.valid-pending.json b/tests/fixtures/contracts/judge-stopped.valid-pending.json new file mode 100644 index 0000000..1b832f8 --- /dev/null +++ b/tests/fixtures/contracts/judge-stopped.valid-pending.json @@ -0,0 +1,11 @@ +{ + "schema_version": 1, + "kind": "judge.stopped", + "binding": { + "evaluation_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000001", + "attempt_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000007", + "worker_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000008", + "fencing_version": 1 + }, + "cleanup_complete": false +} diff --git a/tests/fixtures/contracts/judge-stopped.valid.json b/tests/fixtures/contracts/judge-stopped.valid.json new file mode 100644 index 0000000..9917893 --- /dev/null +++ b/tests/fixtures/contracts/judge-stopped.valid.json @@ -0,0 +1,11 @@ +{ + "schema_version": 1, + "kind": "judge.stopped", + "binding": { + "evaluation_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000001", + "attempt_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000007", + "worker_id": "aaaaaaaa-aaaa-4aaa-8aaa-000000000008", + "fencing_version": 1 + }, + "cleanup_complete": true +}