From 53caafc157e8c68bb1987202fec4016ec7cfbd73 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Fri, 9 Oct 2026 15:18:08 +0800 Subject: [PATCH] feat(coordinator): dispatch versioned runtime leases and reports --- backend/src/kf_oj/contracts/judging_v2.py | 8 +- backend/tests/test_coordinator.py | 15 +- backend/tests/test_judging_v2_contracts.py | 15 + .../tests/test_multi_runtime_coordinator.py | 241 ++++++++++++++++ .../crates/contracts/src/judging_versions.rs | 267 ++++++++++++++++++ .../contracts/tests/judging_v2_examples.rs | 74 +++++ judge/crates/coordinator/src/api.rs | 7 +- judge/crates/coordinator/src/config.rs | 5 +- judge/crates/coordinator/src/repository.rs | 188 +++++++----- judge/crates/worker/src/consumer.rs | 9 +- 10 files changed, 744 insertions(+), 85 deletions(-) create mode 100644 backend/tests/test_multi_runtime_coordinator.py diff --git a/backend/src/kf_oj/contracts/judging_v2.py b/backend/src/kf_oj/contracts/judging_v2.py index 1077fff..3a84e6b 100644 --- a/backend/src/kf_oj/contracts/judging_v2.py +++ b/backend/src/kf_oj/contracts/judging_v2.py @@ -331,10 +331,16 @@ NodeAdmission = v1.AdmissionPolicy | AdmissionPolicy | AdmissionRegistry def admit_manifest( policy: NodeAdmission | None, manifest: v1.JobManifest | JobManifest, attempt_number: int ) -> None: + if type(manifest) not in (v1.JobManifest, JobManifest): + raise ValueError("unsupported manifest model") + # Copied/constructed values must pass the original strict protocol validators too. + manifest = type(manifest).model_validate(manifest.model_dump(mode="json")) if isinstance(policy, AdmissionRegistry): policy.admit(manifest, attempt_number) elif manifest.schema_version == 1 and type(policy) is v1.AdmissionPolicy: - policy.admit(manifest, attempt_number) + v1.AdmissionPolicy.model_validate(policy.model_dump(mode="json")).admit( + manifest, attempt_number + ) elif manifest.schema_version == 2 and type(policy) is AdmissionPolicy: policy.admit(manifest, attempt_number) else: diff --git a/backend/tests/test_coordinator.py b/backend/tests/test_coordinator.py index 16c7769..4250280 100644 --- a/backend/tests/test_coordinator.py +++ b/backend/tests/test_coordinator.py @@ -27,8 +27,10 @@ ROOT = Path(__file__).resolve().parents[2] @asynccontextmanager -async def fixture(postgres_url, tmp_path, monkeypatch): - _, policy = import_fixture(tmp_path / "import") +async def fixture( + postgres_url, tmp_path, monkeypatch, *, import_builder=import_fixture, node_policy=None +): + _, policy = import_builder(tmp_path / "import") engine = create_async_engine(postgres_url) platform_role(engine) monkeypatch.setattr("kf_oj.api.app.build_engine", lambda _: engine) @@ -66,7 +68,7 @@ async def fixture(postgres_url, tmp_path, monkeypatch): database_url=coord_url, workers=workers, platform_token=platform, - admission=policy.model_dump(mode="json"), + admission=(node_policy or policy).model_dump(mode="json"), lease_seconds=5, recovery_seconds=1, ) @@ -134,17 +136,20 @@ async def fixture(postgres_url, tmp_path, monkeypatch): "cpu.max": "50000 100000", } - async def create(): + async def create(language="c17"): return await app.state.submissions.create( owner, CreateSubmission( problem_version_id=version, - language="c17", + language=language, idempotency_key=uuid4(), source="synthetic report fixture; never compiled on host", ), ) + client.fixture_app = app + client.fixture_account = owner + client.fixture_version = version client.coordinator_unit = unit client.coordinator_role = role yield client, c, workers, platform, create diff --git a/backend/tests/test_judging_v2_contracts.py b/backend/tests/test_judging_v2_contracts.py index 8c63439..b51dbff 100644 --- a/backend/tests/test_judging_v2_contracts.py +++ b/backend/tests/test_judging_v2_contracts.py @@ -220,3 +220,18 @@ def test_registry_is_explicit_for_each_protocol_and_retains_legacy_configuration raw["multi_runtime"]["prefetch"] = 2 with pytest.raises(ValueError): Settings(_env_file=None, judge_admission=raw) + + +def test_generic_admission_revalidates_constructed_values_in_both_generations(): + registry = fixture("judge-registry.valid", v2.AdmissionRegistry) + old = ROOT / "tests/fixtures/contracts/judge-manifest.valid.json" + legacy = v1.decode_wire(v1.JobManifest, old.read_bytes()) + for manifest in (legacy, fixture("judge-manifest.valid-python313", v2.JobManifest)): + for wrong in ( + manifest.model_copy(update={"schema_version": True}), + manifest.model_copy(update={"job_wall_time_us": 1}), + manifest.model_copy(update={"cases": ()}), + ): + with pytest.raises(ValueError): + v2.admit_manifest(registry, wrong, 1) + v2.admit_manifest(registry, legacy, 1) diff --git a/backend/tests/test_multi_runtime_coordinator.py b/backend/tests/test_multi_runtime_coordinator.py new file mode 100644 index 0000000..1e411f3 --- /dev/null +++ b/backend/tests/test_multi_runtime_coordinator.py @@ -0,0 +1,241 @@ +"""Actual finite Rust Coordinator/PG control acceptance; reports are synthetic, no execution.""" + +import asyncio +import copy +import hashlib +import json +from uuid import UUID, uuid4 + +import pytest +from test_coordinator import claim, fixture, manifest, notice, synthetic_report, worker_headers +from test_multi_runtime_versions import multi_import + +from kf_oj.contracts import judging_v2 as v2 +from kf_oj.contracts import judging_versions as versions + + +def report_v2(grant, task): + result = synthetic_report(grant, task) + result["schema_version"] = 2 + language = v2.Language(task["language"]) + kind = v2.runtime_contract(language)[3] + if kind == v2.ProgramKind.PYTHON: + files = [dict(path="main.py", artifact=task["source"])] + elif kind == v2.ProgramKind.JAVA: + files = [ + dict(path=name, artifact=dict(digest="d" * 64, bytes=256)) + for name in ("Helper.class", "Main$Inner.class", "Main.class") + ] + else: + files = [dict(path="main", artifact=dict(digest="d" * 64, bytes=4096))] + result["compilation"]["program"] = dict( + kind=kind.value, + binding=grant["binding"], + source=task["source"], + toolchain=task["toolchain"], + files=files, + ) + return result + + +def stage_v2(grant, language, sequence, stage="compiling", index=None): + return dict( + schema_version=2, + kind="judge.stage", + binding=grant["binding"], + sequence=sequence, + stage=stage, + case_index=index, + preparation_kind=v2.runtime_contract(v2.Language(language))[4].value, + ) + + +def test_actual_versioned_claim_canonical_snapshot_stage_and_report( + postgres_url, tmp_path, monkeypatch +): + async def run(): + async with fixture(postgres_url, tmp_path, monkeypatch, import_builder=multi_import) as ( + client, + c, + workers, + _, + create, + ): + for language in v2.Language: + created = await create(language.value) + grant = await claim(client, created, workers[0]) + assert grant["schema_version"] == 2 + task = await manifest(c, grant) + assert task["schema_version"] == 2 and task["language"] == language.value + parsed = versions.decode_manifest(json.dumps(task).encode()) + canonical = versions.manifest_bytes(parsed) + assert hashlib.sha256(canonical).hexdigest() == grant["manifest_digest"] + async with client.fixture_app.state.sessions() as s: + version = await client.fixture_app.state.problems.for_submission( + s, created.problem_version_id + ) + expected = version.manifest( + created.evaluation_id, created.submission_id, language, parsed.source + ) + assert versions.manifest_bytes(expected) == canonical + headers = worker_headers(workers[0], grant) + bad = stage_v2(grant, language, 1) + bad["preparation_kind"] = ( + "compile" if language == v2.Language.PYTHON313 else "syntax_check" + ) + assert ( + await client.post("/internal/stage", json=bad, headers=headers) + ).status_code == 422 + old = stage_v2(grant, language, 1) + old.update(schema_version=1) + old.pop("preparation_kind") + assert ( + await client.post("/internal/stage", json=old, headers=headers) + ).status_code == 422 + for sequence, phase, index in ( + (1, "compiling", None), + (2, "running", 0), + (3, "running", 1), + ): + stage = stage_v2(grant, language, sequence, phase, index) + assert ( + await client.post("/internal/stage", json=stage, headers=headers) + ).status_code == 204 + assert ( + await client.post("/internal/stage", json=stage, headers=headers) + ).status_code == 204 + result = report_v2(grant, task) + for invalid in ( + result | {"manifest_digest": "f" * 64}, + result | {"binding": grant["binding"] | {"fencing_version": 2}}, + ): + assert ( + await client.post("/internal/report", json=invalid, headers=headers) + ).status_code in (409, 422) + changed = copy.deepcopy(result) + changed["compilation"]["program"]["toolchain"]["image_digest"] = "e" * 64 + assert ( + await client.post("/internal/report", json=changed, headers=headers) + ).status_code == 422 + assert ( + await client.post("/internal/report", json=result, headers=headers) + ).status_code == 204 + assert ( + await client.post("/internal/report", json=result, headers=headers) + ).status_code == 204 + final = await c.fetchval( + "SELECT final_report FROM evaluations WHERE id=$1", created.evaluation_id + ) + assert json.loads(final) == result + assert ( + await c.fetchval( + "SELECT state FROM evaluations WHERE id=$1", created.evaluation_id + ) + == "finished" + ) + assert ( + await c.fetchval( + "SELECT count(*) FROM case_results WHERE evaluation_id=$1", + created.evaluation_id, + ) + == 2 + ) + # No real program/Worker/Root executed; accepted is a synthetic protocol value only. + + asyncio.run(run()) + + +@pytest.mark.parametrize("kind", ["legacy", "identity", "budget"]) +def test_actual_claim_refuses_missing_protocol_or_changed_capability_without_attempt( + postgres_url, + tmp_path, + monkeypatch, + kind, +): + from pathlib import Path + + from kf_oj.contracts.judging import decode_wire + + fixtures = Path(__file__).resolve().parents[2] / "tests/fixtures/contracts-v2" + registry = decode_wire( + v2.AdmissionRegistry, (fixtures / "judge-registry.valid.json").read_bytes() + ) + raw = registry.model_dump(mode="json") + if kind == "identity": + raw["multi_runtime"]["capabilities"][2]["toolchain"]["image_digest"] = "e" * 64 + elif kind == "budget": + raw["multi_runtime"]["capabilities"][2]["run_ceiling"]["memory_bytes"] = 32 * 1024 * 1024 + policy = registry.legacy if kind == "legacy" else v2.AdmissionRegistry.model_validate(raw) + + async def run(): + async with fixture( + postgres_url, + tmp_path, + monkeypatch, + import_builder=multi_import, + node_policy=policy, + ) as (client, c, workers, _, create): + created = await create("python313") + response = await client.post( + "/internal/claim", + json=notice(created.evaluation_id), + headers=worker_headers(workers[0]), + ) + assert response.status_code == 422 + assert await c.fetchval("SELECT count(*) FROM judge_attempts") == 0 + assert await c.fetchval("SELECT state FROM evaluations") == "queued" + assert (await client.get("/health/ready")).status_code == 200 + + asyncio.run(run()) + + +def test_actual_v2_cancel_requires_matching_stop_protocol_and_fence( + postgres_url, tmp_path, monkeypatch +): + async def run(): + async with fixture(postgres_url, tmp_path, monkeypatch, import_builder=multi_import) as ( + client, + c, + workers, + platform, + create, + ): + created = await create("java21") + grant = await claim(client, created, workers[0]) + assert ( + await client.post( + "/internal/cancel", + json=notice(created.evaluation_id), + headers={"Authorization": "Bearer " + platform}, + ) + ).status_code == 204 + stopped = dict( + schema_version=2, + kind="judge.stopped", + binding=grant["binding"], + cleanup_complete=True, + ) + headers = worker_headers(workers[0]) + assert ( + await client.post( + "/internal/stopped", json=stopped | {"schema_version": 1}, headers=headers + ) + ).status_code == 422 + bad = stopped | {"binding": grant["binding"] | {"attempt_id": str(uuid4())}} + assert ( + await client.post("/internal/stopped", json=bad, headers=headers) + ).status_code == 409 + assert ( + await client.post("/internal/stopped", json=stopped, headers=headers) + ).status_code == 204 + assert await c.fetchval("SELECT state FROM evaluations") == "cancelled" + assert ( + await c.fetchval( + "SELECT cleanup_complete FROM judge_attempts WHERE id=$1", + UUID(grant["binding"]["attempt_id"]), + ) + is True + ) + # This proves stop-report authority only; real JVM cleanup remains VM acceptance. + + asyncio.run(run()) diff --git a/judge/crates/contracts/src/judging_versions.rs b/judge/crates/contracts/src/judging_versions.rs index a8e2dce..f151c84 100644 --- a/judge/crates/contracts/src/judging_versions.rs +++ b/judge/crates/contracts/src/judging_versions.rs @@ -39,3 +39,270 @@ impl JudgementReport { } } } + +/// Operator explicitly declares accepted generations; legacy policies are never promoted. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(untagged)] +pub enum NodeAdmission { + Legacy(v1::AdmissionPolicy), + Multi(v2::AdmissionPolicy), + Registry(v2::AdmissionRegistry), +} +impl NodeAdmission { + pub fn validate(&self) -> v1::Validation { + match self { + Self::Legacy(policy) => policy.validate(), + Self::Multi(policy) => policy.validate(), + Self::Registry(policy) => policy.validate(), + } + } + pub fn admit(&self, manifest: &JobManifest, attempt_number: u32) -> v1::Validation { + self.validate()?; + match (self, manifest) { + (Self::Legacy(policy), JobManifest::V1(manifest)) => { + policy.admit(manifest, attempt_number) + } + (Self::Multi(policy), JobManifest::V2(manifest)) => { + policy.admit(manifest, attempt_number) + } + (Self::Registry(policy), manifest) => policy.admit(manifest, attempt_number), + _ => Err("node does not advertise this manifest protocol"), + } + } + fn capacity(&self) -> (u32, u32, u32) { + match self { + Self::Legacy(policy) => ( + policy.max_concurrent_jobs, + policy.prefetch, + policy.max_retries, + ), + Self::Multi(policy) => ( + policy.max_concurrent_jobs, + policy.prefetch, + policy.max_retries, + ), + Self::Registry(registry) => { + // Validation requires a nonempty registry and identical common node capacities. + if let Some(policy) = ®istry.legacy { + ( + policy.max_concurrent_jobs, + policy.prefetch, + policy.max_retries, + ) + } else if let Some(policy) = ®istry.multi_runtime { + ( + policy.max_concurrent_jobs, + policy.prefetch, + policy.max_retries, + ) + } else { + (0, 0, 0) + } + } + } + } + pub fn max_concurrent_jobs(&self) -> u32 { + self.capacity().0 + } + pub fn prefetch(&self) -> u32 { + self.capacity().1 + } + pub fn max_retries(&self) -> u32 { + self.capacity().2 + } +} + +macro_rules! versions { + ($name:ident) => { + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] + #[serde(untagged)] + pub enum $name { + V1(v1::$name), + V2(v2::$name), + } + }; +} +versions!(LeaseGrant); +versions!(StageReport); +versions!(StoppedReport); + +impl JobManifest { + pub fn version(&self) -> u16 { + match self { + Self::V1(_) => 1, + Self::V2(_) => 2, + } + } + pub fn cases(&self) -> &[v1::TestCase] { + match self { + Self::V1(m) => &m.cases, + Self::V2(m) => &m.cases, + } + } + pub fn compile_limits(&self) -> &v1::RunLimits { + match self { + Self::V1(m) => &m.compile_limits, + Self::V2(m) => &m.compile_limits, + } + } + pub fn problem_version_id(&self) -> v1::Identifier { + match self { + Self::V1(m) => m.problem_version_id, + Self::V2(m) => m.problem_version_id, + } + } + pub fn evaluation_id(&self) -> v1::Identifier { + match self { + Self::V1(m) => m.evaluation_id, + Self::V2(m) => m.evaluation_id, + } + } +} +impl LeaseGrant { + pub fn for_manifest( + manifest: &JobManifest, + binding: v1::LeaseBinding, + lease_token: v1::LeaseToken, + attempt_number: u32, + expires_at_unix_ms: u64, + manifest_digest: v1::Digest, + ) -> Self { + match manifest { + JobManifest::V1(_) => Self::V1(v1::LeaseGrant { + schema_version: crate::SchemaVersion, + kind: v1::LeaseKind::Value, + binding, + lease_token, + attempt_number, + expires_at_unix_ms, + manifest_digest, + }), + JobManifest::V2(_) => Self::V2(v2::LeaseGrant { + schema_version: v2::SchemaVersion, + kind: v1::LeaseKind::Value, + binding, + lease_token, + attempt_number, + expires_at_unix_ms, + manifest_digest, + }), + } + } +} +/// Shared state-machine projection only; serialization retains the original wire value. +pub struct StageFields<'a> { + pub binding: &'a v1::LeaseBinding, + pub sequence: u64, + pub stage: v1::Stage, + pub case_index: Option, +} +impl StageReport { + pub fn fields(&self) -> StageFields<'_> { + match self { + Self::V1(r) => StageFields { + binding: &r.binding, + sequence: r.sequence, + stage: r.stage, + case_index: r.case_index, + }, + Self::V2(r) => StageFields { + binding: &r.binding, + sequence: r.sequence, + stage: r.stage, + case_index: r.case_index, + }, + } + } + pub fn validate_for_manifest(&self, manifest: &JobManifest) -> v1::Validation { + match (self, manifest) { + (Self::V1(r), JobManifest::V1(m)) => { + r.validate()?; + if r.binding.evaluation_id != m.evaluation_id { + return Err("stage belongs to another manifest"); + } + Ok(()) + } + (Self::V2(r), JobManifest::V2(m)) => r.validate_for_manifest(m), + _ => Err("stage protocol differs from authorized manifest"), + } + } +} +pub struct ReportFields<'a> { + pub binding: &'a v1::LeaseBinding, + pub verdict: v1::Verdict, + pub failure_code: Option, + pub cleanup_complete: bool, + pub compilation_status: v1::CompilationStatus, + pub cases: &'a [v1::CaseReport], +} +impl JudgementReport { + pub fn fields(&self) -> ReportFields<'_> { + match self { + Self::V1(r) => ReportFields { + binding: &r.binding, + verdict: r.verdict, + failure_code: r.failure_code, + cleanup_complete: r.cleanup_complete, + compilation_status: r.compilation.status, + cases: &r.cases, + }, + Self::V2(r) => ReportFields { + binding: &r.binding, + verdict: r.verdict, + failure_code: r.failure_code, + cleanup_complete: r.cleanup_complete, + compilation_status: r.compilation.status, + cases: &r.cases, + }, + } + } + pub fn validate_program_budget(&self, manifest: &JobManifest) -> v1::Validation { + let limits = manifest.compile_limits(); + let fits = match self { + Self::V1(r) => r + .compilation + .program + .as_ref() + .is_none_or(|p| p.bytes <= limits.file_bytes), + Self::V2(r) => r.compilation.program.as_ref().is_none_or(|p| { + // Python reuses read-only source input, not a compiler output file. + p.kind == v2::ProgramKind::PythonSource + || (p + .files + .iter() + .all(|f| f.artifact.bytes <= limits.file_bytes) + && p.bytes() <= limits.workdir_bytes) + }), + }; + if fits { + Ok(()) + } else { + Err("compiled program exceeds output budget") + } + } +} +impl StoppedReport { + pub fn binding(&self) -> &v1::LeaseBinding { + match self { + Self::V1(r) => &r.binding, + Self::V2(r) => &r.binding, + } + } + pub fn cleanup_complete(&self) -> bool { + match self { + Self::V1(r) => r.cleanup_complete, + Self::V2(r) => r.cleanup_complete, + } + } + pub fn validate_for_manifest(&self, manifest: &JobManifest) -> v1::Validation { + match (self, manifest) { + (Self::V1(r), JobManifest::V1(_)) => r.validate()?, + (Self::V2(r), JobManifest::V2(_)) => r.validate()?, + _ => return Err("stop protocol differs from authorized manifest"), + } + if self.binding().evaluation_id != manifest.evaluation_id() { + return Err("stop belongs to another manifest"); + } + Ok(()) + } +} diff --git a/judge/crates/contracts/tests/judging_v2_examples.rs b/judge/crates/contracts/tests/judging_v2_examples.rs index 8af8176..87732c9 100644 --- a/judge/crates/contracts/tests/judging_v2_examples.rs +++ b/judge/crates/contracts/tests/judging_v2_examples.rs @@ -187,3 +187,77 @@ fn registry_explicitly_admits_both_protocols_without_promoting_legacy_policy() { wrong.multi_runtime.as_mut().unwrap().max_retries += 1; assert!(wrong.validate().is_err()); } + +#[test] +fn operator_admission_and_transport_dispatch_keep_generations_explicit() { + let original = fixtures().parent().unwrap().join("contracts"); + let legacy_policy: versions::NodeAdmission = + v1::decode_wire(&fs::read(original.join("judge-admission.valid.json")).unwrap()).unwrap(); + let multi_policy: versions::NodeAdmission = fixture("judge-admission.valid"); + let registry: versions::NodeAdmission = fixture("judge-registry.valid"); + let old_manifest: versions::JobManifest = + v1::decode_wire(&fs::read(original.join("judge-manifest.valid.json")).unwrap()).unwrap(); + let new_manifest: versions::JobManifest = fixture("judge-manifest.valid-python313"); + for policy in [&legacy_policy, &multi_policy, ®istry] { + policy.validate().unwrap(); + assert_eq!(policy.max_concurrent_jobs(), 1); + assert_eq!(policy.prefetch(), 1); + assert_eq!(policy.max_retries(), 2); + } + registry.admit(&old_manifest, 1).unwrap(); + registry.admit(&new_manifest, 1).unwrap(); + assert!(legacy_policy.admit(&new_manifest, 1).is_err()); + assert!(multi_policy.admit(&old_manifest, 1).is_err()); + let old_stage: versions::StageReport = + v1::decode_wire(&fs::read(original.join("judge-stage.valid.json")).unwrap()).unwrap(); + let new_stage: versions::StageReport = fixture("judge-stage.valid-python313"); + old_stage.validate_for_manifest(&old_manifest).unwrap(); + new_stage.validate_for_manifest(&new_manifest).unwrap(); + assert!(old_stage.validate_for_manifest(&new_manifest).is_err()); + assert!(new_stage.validate_for_manifest(&old_manifest).is_err()); + let old_stop: versions::StoppedReport = + v1::decode_wire(&fs::read(original.join("judge-stopped.valid.json")).unwrap()).unwrap(); + let new_stop: versions::StoppedReport = fixture("judge-stopped.valid"); + assert!(old_stop.validate_for_manifest(&new_manifest).is_err()); + assert!(new_stop.validate_for_manifest(&old_manifest).is_err()); + for manifest in [&old_manifest, &new_manifest] { + let legacy_grant: v1::LeaseGrant = + v1::decode_wire(&fs::read(original.join("judge-lease.valid.json")).unwrap()).unwrap(); + let grant = versions::LeaseGrant::for_manifest( + manifest, + legacy_grant.binding, + legacy_grant.lease_token, + legacy_grant.attempt_number, + legacy_grant.expires_at_unix_ms, + legacy_grant.manifest_digest, + ); + let value = serde_json::to_value(&grant).unwrap(); + assert_eq!( + value["schema_version"].as_u64().unwrap(), + u64::from(manifest.version()) + ); + v1::decode_wire::(&serde_json::to_vec(&value).unwrap()).unwrap(); + } +} + +#[test] +fn typed_compiler_output_checks_individual_files_and_collection_storage() { + let report: versions::JudgementReport = fixture("judge-result.valid-java21"); + let mut manifest: versions::JobManifest = fixture("judge-manifest.valid-java21"); + report.validate_program_budget(&manifest).unwrap(); + let versions::JobManifest::V2(m) = &mut manifest else { + unreachable!() + }; + m.compile_limits.stdout_bytes = 32; + m.compile_limits.stderr_bytes = 32; + m.compile_limits.file_bytes = 63; + m.validate().unwrap(); + assert!(report.validate_program_budget(&manifest).is_err()); + let versions::JobManifest::V2(m) = &mut manifest else { + unreachable!() + }; + m.compile_limits.file_bytes = 64; + m.compile_limits.workdir_bytes = 128; + m.validate().unwrap(); + assert!(report.validate_program_budget(&manifest).is_err()); +} diff --git a/judge/crates/coordinator/src/api.rs b/judge/crates/coordinator/src/api.rs index 25b5012..758a7ff 100644 --- a/judge/crates/coordinator/src/api.rs +++ b/judge/crates/coordinator/src/api.rs @@ -8,6 +8,7 @@ use axum::{ response::{IntoResponse, Response}, routing::{get, post}, }; +use kf_contracts::judging_versions::{JudgementReport, StageReport, StoppedReport}; use kf_contracts::{JudgeTaskNotification, judging::*}; use serde_json::{Value, json}; use std::{sync::Arc, time::Duration}; @@ -158,19 +159,19 @@ async fn progress(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult, h: HeaderMap, b: Bytes) -> ApiResult { let r: StageReport = decode(&b)?; - bound_worker(&app, &h, &r.binding)?; + bound_worker(&app, &h, r.fields().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)?; + bound_worker(&app, &h, r.fields().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)?; + bound_worker(&app, &h, r.binding())?; app.repository.stopped(&r).await?; Ok(StatusCode::NO_CONTENT) } diff --git a/judge/crates/coordinator/src/config.rs b/judge/crates/coordinator/src/config.rs index e8c139e..af50105 100644 --- a/judge/crates/coordinator/src/config.rs +++ b/judge/crates/coordinator/src/config.rs @@ -1,5 +1,6 @@ //! Service-only secrets from a private local file; never format database/token values. -use kf_contracts::judging::{AdmissionPolicy, Identifier, LeaseToken}; +use kf_contracts::judging::{Identifier, LeaseToken}; +use kf_contracts::judging_versions::NodeAdmission; use serde::Deserialize; use std::{collections::HashSet, fs, io, net::SocketAddr, path::Path}; #[derive(Deserialize)] @@ -15,7 +16,7 @@ pub struct Config { pub database_url: String, pub workers: Vec, pub platform_token: LeaseToken, - pub admission: AdmissionPolicy, + pub admission: NodeAdmission, pub lease_seconds: u32, pub recovery_seconds: u32, } diff --git a/judge/crates/coordinator/src/repository.rs b/judge/crates/coordinator/src/repository.rs index 907d079..e04e40a 100644 --- a/judge/crates/coordinator/src/repository.rs +++ b/judge/crates/coordinator/src/repository.rs @@ -1,6 +1,9 @@ //! Execution authority is applied under row locks, using database clock and fencing. use crate::database::{Database, Error, Result}; -use kf_contracts::{SchemaVersion, judging::*}; +use kf_contracts::judging_versions::{ + JobManifest, JudgementReport, LeaseGrant, NodeAdmission, StageReport, StoppedReport, +}; +use kf_contracts::{SchemaVersion, judging::*, judging_v2 as v2}; use serde_json::{Value, json}; use sha2::{Digest as _, Sha256}; use tokio_postgres::{Row, Transaction}; @@ -8,7 +11,7 @@ use uuid::Uuid; #[derive(Clone)] pub struct Repository { pub database: Database, - pub admission: AdmissionPolicy, + pub admission: NodeAdmission, pub lease_seconds: u32, } pub fn identifier(v: Uuid) -> Identifier { @@ -34,7 +37,7 @@ 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 h=t.query_opt("SELECT s.id,s.problem_version_id,s.language,s.source_digest,a.bytes,v.judging_schema_version,v.languages,v.language_configurations,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(); @@ -60,28 +63,76 @@ impl Repository { }, }); } - 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, + let schema: i16 = h.get("judging_schema_version"); + let source = ArtifactReference { + digest: h + .get::<_, String>("source_digest") + .try_into() + .map_err(invalid)?, + bytes: h.get::<_, i64>("bytes") as u64, }; - m.validate().map_err(invalid)?; + let m = if schema == 1 { + JobManifest::V1(kf_contracts::judging::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, + 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, + }) + } else if schema == 2 { + let configs: Vec = as_wire( + h.get::<_, Option>("language_configurations") + .ok_or(Error::Invalid)?, + )?; + let languages: Vec = as_wire(h.get("languages"))?; + if configs.is_empty() + || configs.len() > 4 + || configs.iter().map(|c| c.language).collect::>() != languages + || languages + .iter() + .collect::>() + .len() + != languages.len() + || ["toolchain", "compile_limits", "run_limits"] + .iter() + .any(|key| h.get::<_, Option>(*key).is_some()) + || h.get::<_, Option>("job_wall_time_us").is_some() + { + return Err(Error::Invalid); + } + let language: v2::Language = as_wire(json!(h.get::<_, String>("language")))?; + let config = configs + .into_iter() + .find(|c| c.language == language) + .ok_or(Error::Invalid)?; + JobManifest::V2(v2::JobManifest { + schema_version: v2::SchemaVersion, + kind: ManifestKind::Value, + evaluation_id: identifier(eid), + submission_id: identifier(sid), + problem_version_id: identifier(vid), + language, + source, + cases, + comparison: Comparison::Value, + toolchain: config.toolchain, + compile_limits: config.compile_limits, + run_limits: config.run_limits, + job_wall_time_us: config.job_wall_time_us, + }) + } else { + return Err(Error::Invalid); + }; + // Canonicalization runs the selected generation's full semantic validators. + m.canonical_bytes().map_err(invalid)?; Ok(m) } pub async fn claim(&self, eid: Identifier, worker: Identifier) -> Result> { @@ -91,7 +142,7 @@ impl Repository { 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); + 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") @@ -101,7 +152,7 @@ impl Repository { 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) { + if active >= i64::from(self.admission.max_concurrent_jobs()) { return Err(Error::Busy); } let m = self.manifest(&t, &e).await?; @@ -113,23 +164,22 @@ impl Repository { .get::<_, i64>("fencing_version") .checked_add(1) .ok_or(Error::Invalid)?; - let digest = hash(&manifest_bytes(&m).map_err(invalid)?); + let digest = hash(&m.canonical_bytes().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 { + let grant = LeaseGrant::for_manifest( + &m, + 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)?, - }; + serde_json::from_value(json!(token)).map_err(invalid)?, + number as u32, + row.get::<_, i64>(0) as u64, + digest.try_into().map_err(invalid)?, + ); t.commit().await?; Ok(Some(grant)) } @@ -188,28 +238,30 @@ impl Repository { ) } pub async fn stage(&self, report: &StageReport, token: &str) -> Result<()> { - report.validate().map_err(invalid)?; - let b = &report.binding; + let fields = report.fields(); + let b = fields.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 { + let m: JobManifest = as_wire(e.get("manifest"))?; + report.validate_for_manifest(&m).map_err(invalid)?; + let stage = match fields.stage { Stage::Compiling => "compiling", Stage::Running => "running", }; - let index = report.case_index.map(i32::from); + let index = fields.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 + if fields.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 { + if fields.sequence as i64 != sequence.checked_add(1).ok_or(Error::Invalid)? + || match fields.stage { Stage::Compiling => e.get::<_, String>("state") != "claimed" || old_stage.is_some(), Stage::Running => { !matches!( @@ -221,15 +273,14 @@ impl Repository { { return Err(Error::Rejected); } - let m: JobManifest = as_wire(e.get("manifest"))?; - if index.is_some_and(|i| i as usize >= m.cases.len()) { + 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), + &(fields.sequence as i64), &stage, &index, ], @@ -255,7 +306,8 @@ impl Repository { 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 retries = + (e.get::<_, i32>("max_attempts") as u32).min(self.admission.max_retries() + 1); let retry = !cancelled && clean && verdict == "system_error" @@ -283,7 +335,8 @@ impl Repository { Ok(()) } pub async fn report(&self, r: &JudgementReport, token: &str) -> Result<()> { - let b = &r.binding; + let fields = r.fields(); + let b = fields.binding; let mut s = self.database.session().await?; let t = s.client.transaction().await?; let e = self.bound(&t, b, Some(token)).await?; @@ -303,33 +356,27 @@ impl Repository { .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) + r.validate_program_budget(&m).map_err(invalid)?; + if !matches!(fields.verdict, Verdict::SystemError | Verdict::Cancelled) && (e.get::<_, String>("state") == "claimed" - || (r.compilation.status == CompilationStatus::Succeeded + || (fields.compilation_status == CompilationStatus::Succeeded && e.get::<_, String>("state") != "running")) { return Err(Error::Rejected); } - if !r.cases.is_empty() { + if !fields.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) + if fields.cases.len() > index + 1 + || (!matches!(fields.verdict, Verdict::SystemError | Verdict::Cancelled) + && fields.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 vid = m.problem_version_id().as_uuid(); + for (position, c) in fields.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?; } @@ -343,15 +390,15 @@ impl Repository { &[&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)?; + let verdict = serde_json::to_value(fields.verdict).map_err(invalid)?; + let code = serde_json::to_value(fields.failure_code).map_err(invalid)?; self.finish( &t, &e, b, verdict.as_str().ok_or(Error::Invalid)?, code.as_str(), - r.cleanup_complete, + fields.cleanup_complete, ) .await?; t.commit().await?; @@ -381,13 +428,14 @@ impl Repository { Ok(()) } pub async fn stopped(&self, r: &StoppedReport) -> Result<()> { - r.validate().map_err(invalid)?; - let b = &r.binding; + 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?; + let m: JobManifest = as_wire(e.get("manifest"))?; + r.validate_for_manifest(&m).map_err(invalid)?; if e.get::<_, String>("state") != "cancelling" { - if r.cleanup_complete && e.get::<_, Option>("cleanup_complete") == Some(true) { + if r.cleanup_complete() && e.get::<_, Option>("cleanup_complete") == Some(true) { return Ok(()); } return Err(Error::Rejected); @@ -404,7 +452,7 @@ impl Repository { b, "system_error", Some(code.as_deref().unwrap_or("lease_lost")), - r.cleanup_complete, + r.cleanup_complete(), ) .await?; t.commit().await?; diff --git a/judge/crates/worker/src/consumer.rs b/judge/crates/worker/src/consumer.rs index c23c29e..e17f2fd 100644 --- a/judge/crates/worker/src/consumer.rs +++ b/judge/crates/worker/src/consumer.rs @@ -1,6 +1,7 @@ //! Ordinary transport adapter. AMQP is a bounded wake-up; claim is sole authority. use crate::pipeline::{ArtifactPort, Attempt}; use futures_lite::{StreamExt, future::block_on}; +use kf_contracts::judging_versions::{JobManifest as VersionedManifest, NodeAdmission}; use kf_contracts::{JudgeTaskNotification, SchemaVersion, judging::*}; use kf_engine::submission::{StageFailure, judge_submission}; use kf_sandbox::{artifacts::StagePublication, factory::FactoryClient}; @@ -49,7 +50,7 @@ pub struct Config { pub amqp_url: String, pub socket: PathBuf, pub staging: PathBuf, - pub admission: AdmissionPolicy, + pub admission: NodeAdmission, } impl Config { pub fn load(path: &Path) -> io::Result { @@ -85,8 +86,8 @@ impl Config { } if !c.socket.is_absolute() || !c.staging.is_absolute() - || c.admission.max_concurrent_jobs != 1 - || c.admission.prefetch != 1 + || c.admission.max_concurrent_jobs() != 1 + || c.admission.prefetch() != 1 { return Err(error()); } @@ -292,7 +293,7 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu }; let m = artifacts.manifest()?; c.admission - .admit(&m, g.attempt_number) + .admit(&VersionedManifest::V1(m.clone()), g.attempt_number) .map_err(|_| error())?; let cancel = AtomicBool::new(false); let done = AtomicBool::new(false);