From fce752226750a8d1b39e611837c80af8144f909a Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Sun, 11 Oct 2026 03:46:16 +0800 Subject: [PATCH] feat(submissions): bind final performance to trusted measurement epochs --- .../versions/0003_submission_performance.py | 119 ++++++++++++++++++ .../src/kf_oj/modules/submissions/models.py | 64 ++++++++++ backend/tests/test_coordinator.py | 12 +- backend/tests/test_performance_projection.py | 115 +++++++++++++++++ judge/crates/coordinator/README.md | 12 ++ judge/crates/coordinator/src/config.rs | 6 +- judge/crates/coordinator/src/lib.rs | 1 + judge/crates/coordinator/src/main.rs | 14 +++ judge/crates/coordinator/src/performance.rs | 113 +++++++++++++++++ judge/crates/coordinator/src/repository.rs | 42 ++++++- 10 files changed, 495 insertions(+), 3 deletions(-) create mode 100644 backend/migrations/versions/0003_submission_performance.py create mode 100644 backend/tests/test_performance_projection.py create mode 100644 judge/crates/coordinator/src/performance.rs diff --git a/backend/migrations/versions/0003_submission_performance.py b/backend/migrations/versions/0003_submission_performance.py new file mode 100644 index 0000000..345a9ab --- /dev/null +++ b/backend/migrations/versions/0003_submission_performance.py @@ -0,0 +1,119 @@ +"""Trusted final performance projection; no re-encoding or promotion of old reports.""" + +import sqlalchemy as sa +from alembic import op + +revision = "0003_submission_performance" +down_revision = "0002_multi_runtime" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("judge_attempts", sa.Column("measurement_epoch", sa.String(64), nullable=True)) + op.create_check_constraint( + op.f("ck_judge_attempts_measurement_epoch"), + "judge_attempts", + "measurement_epoch IS NULL OR measurement_epoch ~ '^[0-9a-f]{64}$'", + ) + op.create_table( + "evaluation_performance", + sa.Column("evaluation_id", sa.Uuid(), primary_key=True), + sa.Column("submission_id", sa.Uuid(), nullable=False, unique=True), + sa.Column("attempt_id", sa.Uuid(), nullable=False, unique=True), + sa.Column("problem_version_id", sa.Uuid(), nullable=False), + sa.Column("language", sa.String(16), nullable=False), + sa.Column("comparison_group", sa.String(64), nullable=False), + sa.Column("measurement_epoch", sa.String(64), nullable=False), + sa.Column("report_digest", sa.String(64), nullable=False), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("cpu_time_us", sa.BigInteger(), nullable=True), + sa.Column("memory_peak_bytes", sa.BigInteger(), nullable=True), + sa.ForeignKeyConstraint( + ["evaluation_id", "problem_version_id"], + ["evaluations.id", "evaluations.problem_version_id"], + ), + sa.ForeignKeyConstraint( + ["submission_id", "problem_version_id"], + ["submissions.id", "submissions.problem_version_id"], + ), + sa.ForeignKeyConstraint( + ["attempt_id", "evaluation_id"], ["judge_attempts.id", "judge_attempts.evaluation_id"] + ), + sa.CheckConstraint("language IN ('c17','cpp20','python313','java21')", name="language"), + sa.CheckConstraint("comparison_group ~ '^[0-9a-f]{64}$'", name="comparison_group"), + sa.CheckConstraint("measurement_epoch ~ '^[0-9a-f]{64}$'", name="measurement_epoch"), + sa.CheckConstraint("report_digest ~ '^[0-9a-f]{64}$'", name="report_digest"), + sa.CheckConstraint("cpu_time_us IS NULL OR cpu_time_us >= 0", name="cpu"), + sa.CheckConstraint("memory_peak_bytes IS NULL OR memory_peak_bytes >= 0", name="memory"), + ) + for metric in ("cpu_time_us", "memory_peak_bytes"): + op.create_index( + f"ix_performance_{metric}_cohort", + "evaluation_performance", + ["comparison_group", sa.text("finished_at DESC"), sa.text("evaluation_id DESC")], + postgresql_where=sa.text(f"{metric} IS NOT NULL"), + ) + op.create_index( + "ix_submissions_owned_problem_history", + "submissions", + ["account_id", "problem_version_id", sa.text("created_at DESC"), sa.text("id DESC")], + ) + op.execute(""" + CREATE FUNCTION public.kf_trusted_performance() RETURNS trigger LANGUAGE plpgsql + SET search_path = pg_catalog, public AS $$ + BEGIN + IF TG_OP != 'INSERT' THEN + RAISE EXCEPTION 'final performance is immutable' USING ERRCODE = '23514'; + END IF; + IF NOT EXISTS ( + SELECT 1 FROM public.evaluations e + JOIN public.judge_attempts a ON a.id=e.current_attempt_id AND a.evaluation_id=e.id + JOIN public.submissions s ON s.id=e.submission_id + WHERE e.id=NEW.evaluation_id AND e.state='finished' AND e.verdict='accepted' + AND e.failure_code IS NULL AND a.cleanup_complete IS TRUE + AND e.current_attempt_id=NEW.attempt_id AND s.id=NEW.submission_id + AND s.problem_version_id=NEW.problem_version_id AND s.language=NEW.language + AND a.measurement_epoch=NEW.measurement_epoch + AND e.final_report_digest=NEW.report_digest AND a.report_digest=NEW.report_digest + AND e.updated_at=NEW.finished_at + AND e.final_report->>'cleanup_complete'='true' + AND (e.final_report->>'run_cpu_time_us')::bigint + IS NOT DISTINCT FROM NEW.cpu_time_us + AND (e.final_report->>'run_memory_peak_bytes')::bigint + IS NOT DISTINCT FROM NEW.memory_peak_bytes + ) THEN + RAISE EXCEPTION 'performance requires a bound trusted final AC' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; + END $$ + """) + op.execute("REVOKE ALL ON FUNCTION public.kf_trusted_performance() FROM PUBLIC") + op.execute(""" + CREATE TRIGGER kf_trusted_performance BEFORE INSERT OR UPDATE OR DELETE + ON public.evaluation_performance FOR EACH ROW + EXECUTE FUNCTION public.kf_trusted_performance() + """) + op.execute("REVOKE ALL ON evaluation_performance FROM PUBLIC,kf_oj_platform,kf_oj_coordinator") + op.execute("GRANT SELECT ON evaluation_performance TO kf_oj_platform,kf_oj_coordinator") + op.execute("GRANT INSERT ON evaluation_performance TO kf_oj_coordinator") + # Existing attempt INSERT permission covers the new column. No UPDATE epoch permission: + # an unbound historical attempt must never be promoted by changing current configuration. + + +def downgrade() -> None: + op.execute(""" + DO $$ BEGIN + IF EXISTS(SELECT 1 FROM evaluation_performance) OR + EXISTS(SELECT 1 FROM judge_attempts WHERE measurement_epoch IS NOT NULL) THEN + RAISE EXCEPTION 'cannot discard trusted measurement bindings' + USING ERRCODE = '23514'; + END IF; + END $$ + """) + op.drop_table("evaluation_performance") + op.execute("DROP FUNCTION public.kf_trusted_performance()") + op.drop_index("ix_submissions_owned_problem_history", table_name="submissions") + op.drop_constraint(op.f("ck_judge_attempts_measurement_epoch"), "judge_attempts", type_="check") + op.drop_column("judge_attempts", "measurement_epoch") diff --git a/backend/src/kf_oj/modules/submissions/models.py b/backend/src/kf_oj/modules/submissions/models.py index d0ed4f5..4898cf4 100644 --- a/backend/src/kf_oj/modules/submissions/models.py +++ b/backend/src/kf_oj/modules/submissions/models.py @@ -15,6 +15,7 @@ from sqlalchemy import ( String, UniqueConstraint, func, + text, ) from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column @@ -38,6 +39,13 @@ class Submission(CreatedAt, Base): ["source_digest", "artifact_state"], ["artifacts.digest", "artifacts.state"] ), Index("ix_submissions_account_history", "account_id", "created_at", "id"), + Index( + "ix_submissions_owned_problem_history", + "account_id", + "problem_version_id", + text("created_at DESC"), + text("id DESC"), + ), ) id: Mapped[UUID] = mapped_column(primary_key=True) @@ -112,6 +120,10 @@ class JudgeAttempt(CreatedAt, Base): CheckConstraint("fencing_version > 0", name="fence"), CheckConstraint("lease_token_hash ~ '^[0-9a-f]{64}$'", name="token_digest"), CheckConstraint("manifest_digest ~ '^[0-9a-f]{64}$'", name="manifest_digest"), + CheckConstraint( + "measurement_epoch IS NULL OR measurement_epoch ~ '^[0-9a-f]{64}$'", + name="measurement_epoch", + ), CheckConstraint("sequence >= 0", name="sequence"), Index("ix_judge_attempts_expiry", "expires_at", "evaluation_id"), ) @@ -132,6 +144,58 @@ class JudgeAttempt(CreatedAt, Base): report: Mapped[dict | None] = mapped_column(JSONB) report_digest: Mapped[str | None] = mapped_column(String(64)) cleanup_complete: Mapped[bool | None] = mapped_column(Boolean) + measurement_epoch: Mapped[str | None] = mapped_column(String(64)) + + +class EvaluationPerformance(Base): + """Coordinator-only immutable final projection; the platform only reads it.""" + + __tablename__ = "evaluation_performance" + __table_args__ = ( + ForeignKeyConstraint( + ["evaluation_id", "problem_version_id"], + ["evaluations.id", "evaluations.problem_version_id"], + ), + ForeignKeyConstraint( + ["submission_id", "problem_version_id"], + ["submissions.id", "submissions.problem_version_id"], + ), + ForeignKeyConstraint( + ["attempt_id", "evaluation_id"], ["judge_attempts.id", "judge_attempts.evaluation_id"] + ), + CheckConstraint("language IN ('c17','cpp20','python313','java21')", name="language"), + CheckConstraint("comparison_group ~ '^[0-9a-f]{64}$'", name="comparison_group"), + CheckConstraint("measurement_epoch ~ '^[0-9a-f]{64}$'", name="measurement_epoch"), + CheckConstraint("report_digest ~ '^[0-9a-f]{64}$'", name="report_digest"), + CheckConstraint("cpu_time_us IS NULL OR cpu_time_us >= 0", name="cpu"), + CheckConstraint("memory_peak_bytes IS NULL OR memory_peak_bytes >= 0", name="memory"), + Index( + "ix_performance_cpu_time_us_cohort", + "comparison_group", + text("finished_at DESC"), + text("evaluation_id DESC"), + postgresql_where=text("cpu_time_us IS NOT NULL"), + ), + Index( + "ix_performance_memory_peak_bytes_cohort", + "comparison_group", + text("finished_at DESC"), + text("evaluation_id DESC"), + postgresql_where=text("memory_peak_bytes IS NOT NULL"), + ), + ) + + evaluation_id: Mapped[UUID] = mapped_column(primary_key=True) + submission_id: Mapped[UUID] = mapped_column(unique=True) + attempt_id: Mapped[UUID] = mapped_column(unique=True) + problem_version_id: Mapped[UUID] + language: Mapped[str] = mapped_column(String(16)) + comparison_group: Mapped[str] = mapped_column(String(64)) + measurement_epoch: Mapped[str] = mapped_column(String(64)) + report_digest: Mapped[str] = mapped_column(String(64)) + finished_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + cpu_time_us: Mapped[int | None] = mapped_column(BigInteger) + memory_peak_bytes: Mapped[int | None] = mapped_column(BigInteger) class CaseResult(Base): diff --git a/backend/tests/test_coordinator.py b/backend/tests/test_coordinator.py index 9ac63e7..3eef0dc 100644 --- a/backend/tests/test_coordinator.py +++ b/backend/tests/test_coordinator.py @@ -28,7 +28,13 @@ ROOT = Path(__file__).resolve().parents[2] @asynccontextmanager async def fixture( - postgres_url, tmp_path, monkeypatch, *, import_builder=import_fixture, node_policy=None + postgres_url, + tmp_path, + monkeypatch, + *, + import_builder=import_fixture, + node_policy=None, + measurement_epochs=None, ): _, policy = import_builder(tmp_path / "import") engine = create_async_engine(postgres_url) @@ -59,6 +65,9 @@ async def fixture( 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)] + if measurement_epochs is not None: + for worker, epoch in zip(workers, measurement_epochs, strict=True): + worker["measurement_epoch"] = epoch platform = secrets.token_hex(32) config = tmp_path / "coordinator.json" config.write_text( @@ -152,6 +161,7 @@ async def fixture( client.fixture_version = version client.coordinator_unit = unit client.coordinator_role = role + client.coordinator_config = config yield client, c, workers, platform, create finally: subprocess.run(["systemctl", "--user", "stop", unit], check=True, capture_output=True) diff --git a/backend/tests/test_performance_projection.py b/backend/tests/test_performance_projection.py new file mode 100644 index 0000000..9a0e5e8 --- /dev/null +++ b/backend/tests/test_performance_projection.py @@ -0,0 +1,115 @@ +"""Finite actual control-plane PG/HTTP tests; all reports explicitly synthetic. + +No user program runs here. Privilege/tamper/fault acceptance runs in the VM. +""" + +import asyncio +import subprocess + +from test_coordinator import ( + ROOT, + claim, + fixture, + manifest, + stages, + synthetic_report, + worker_headers, +) + + +def test_final_projection_atomic_idempotent_and_bound(postgres_url, tmp_path, monkeypatch): + async def run(): + async with fixture( + postgres_url, tmp_path, monkeypatch, measurement_epochs=["a" * 64, "b" * 64] + ) as (client, c, workers, _, create): + groups = [] + for index in [0, 0, 1]: + created = await create() + worker = workers[index] + grant = await claim(client, created, worker) + await stages(client, worker, grant) + report = synthetic_report(grant, await manifest(c, grant)) + result = await client.post( + "/internal/report", json=report, headers=worker_headers(worker, grant) + ) + assert result.status_code == 204, result.text + row = await c.fetchrow( + "SELECT p.*,e.final_report_digest,a.report_digest AS attempt_digest " + "FROM evaluation_performance p JOIN evaluations e ON e.id=p.evaluation_id " + "JOIN judge_attempts a ON a.id=p.attempt_id WHERE p.evaluation_id=$1", + created.evaluation_id, + ) + assert row is not None + assert row["cpu_time_us"] == report["run_cpu_time_us"] + assert row["memory_peak_bytes"] == report["run_memory_peak_bytes"] + assert row["measurement_epoch"] == worker["measurement_epoch"] + assert row["report_digest"] == row["final_report_digest"] == row["attempt_digest"] + groups.append(row["comparison_group"]) + repeated = await client.post( + "/internal/report", json=report, headers=worker_headers(worker, grant) + ) + assert repeated.status_code == 204 + assert ( + await c.fetchval( + "SELECT count(*) FROM evaluation_performance WHERE evaluation_id=$1", + created.evaluation_id, + ) + == 1 + ) + backfill = await asyncio.to_thread( + subprocess.run, + [ + str(ROOT / "judge/target/debug/kf-coordinator"), + "--config", + str(client.coordinator_config), + "--backfill-performance", + str(created.evaluation_id), + ], + capture_output=True, + text=True, + timeout=15, + ) + assert backfill.returncode == 0, backfill.stderr + assert backfill.stdout.strip() == "trusted performance rows created: 0" + assert groups[0] == groups[1] and groups[2] != groups[0] + assert await c.fetchval("SELECT count(*) FROM evaluation_performance") == 3 + # Read grants rather than attempting denied writes on this host. + grants = await c.fetchrow( + "SELECT has_table_privilege('kf_oj_platform','evaluation_performance','SELECT') " + "AS read,has_table_privilege('kf_oj_platform','evaluation_performance','INSERT') " + "AS platform_insert,has_table_privilege('kf_oj_coordinator'," + "'evaluation_performance','INSERT') AS coord_insert,has_table_privilege(" + "'kf_oj_coordinator','evaluation_performance','UPDATE') AS coord_update," + "has_column_privilege('kf_oj_coordinator','judge_attempts'," + "'measurement_epoch','UPDATE') AS epoch_update" + ) + assert dict(grants) == dict( + read=True, + platform_insert=False, + coord_insert=True, + coord_update=False, + epoch_update=False, + ) + + asyncio.run(run()) + + +def test_unbound_legacy_and_nonaccepted_stay_out(postgres_url, tmp_path, monkeypatch): + async def run(): + async with fixture(postgres_url, tmp_path, monkeypatch) as (client, c, workers, _, create): + for verdict in ["accepted", "wrong_answer"]: + created = await create() + grant = await claim(client, created, workers[0]) + await stages(client, workers[0], grant) + report = synthetic_report(grant, await manifest(c, grant), verdict) + result = await client.post( + "/internal/report", json=report, headers=worker_headers(workers[0], grant) + ) + assert result.status_code == 204, result.text + assert await c.fetchval( + "SELECT final_report_digest IS NOT NULL FROM evaluations WHERE id=$1", + created.evaluation_id, + ) + assert await c.fetchval("SELECT count(*) FROM evaluation_performance") == 0 + + asyncio.run(run()) diff --git a/judge/crates/coordinator/README.md b/judge/crates/coordinator/README.md index d58adbb..70d92b4 100644 --- a/judge/crates/coordinator/README.md +++ b/judge/crates/coordinator/README.md @@ -19,3 +19,15 @@ 清理失败保留 `cleanup_failed`，只有已确认停止的基础设施失败可按有限预算重试。 恢复扫描每批最多16项，待领取事件投递30秒后仍无执行权时补偿重新通知。 Worker 的实际停止证明与执行隔离需独立 VM 验收；普通数据库测试不代表 Linux 安全验收。 + +性能投影由 Coordinator 与验证后的最终 AC 同事务写入，平台仅有 SELECT 权限。 +每个 Worker 可配置 `measurement_epoch`（64 位小写十六进制摘要）。运维须先核对实际 +硬件、内核、CPU 调度/资源上限及测量配置，再固定该纪元；上述条件变化时更换纪元。 +未配置纪元时正常判题与历史查询保留，统计不宣称可比较。组键还绑定不可变题目版本、 +语言、精确工具链/镜像及编译/运行预算；它不含学生源码或提交身份。 + +`kf-coordinator --config /etc/kf-judge/coordinator.json --backfill-performance UUID` +是单条、有数据库期限的受控修复入口。它重新验证已保存报告的绑定、清单与摘要， +仅从原尝试已绑定的纪元生成缺失投影；重复运行不增加样本。新配置不能给无纪元的旧尝试 +补造可信环境。原判题报告不改写。升级只有 Backend Alembic；存在纪元/投影时降级拒绝 +丢弃数据，须另行进行显式数据保留迁移，不能用删除成绩绕过。 diff --git a/judge/crates/coordinator/src/config.rs b/judge/crates/coordinator/src/config.rs index af50105..a4a9af7 100644 --- a/judge/crates/coordinator/src/config.rs +++ b/judge/crates/coordinator/src/config.rs @@ -1,5 +1,5 @@ //! Service-only secrets from a private local file; never format database/token values. -use kf_contracts::judging::{Identifier, LeaseToken}; +use kf_contracts::judging::{Digest, Identifier, LeaseToken}; use kf_contracts::judging_versions::NodeAdmission; use serde::Deserialize; use std::{collections::HashSet, fs, io, net::SocketAddr, path::Path}; @@ -8,6 +8,10 @@ use std::{collections::HashSet, fs, io, net::SocketAddr, path::Path}; pub struct WorkerIdentity { pub id: Identifier, pub token: LeaseToken, + /// Operator-attested measurement setup; absent means no comparable statistics. + /// Rotate whenever hardware/kernel/CPU scheduling or measurement setup changes. + #[serde(default)] + pub measurement_epoch: Option, } #[derive(Deserialize)] #[serde(deny_unknown_fields)] diff --git a/judge/crates/coordinator/src/lib.rs b/judge/crates/coordinator/src/lib.rs index c7f5e34..4c706be 100644 --- a/judge/crates/coordinator/src/lib.rs +++ b/judge/crates/coordinator/src/lib.rs @@ -2,4 +2,5 @@ pub mod api; pub mod config; pub mod database; +pub mod performance; pub mod repository; diff --git a/judge/crates/coordinator/src/main.rs b/judge/crates/coordinator/src/main.rs index 614d637..ddf34da 100644 --- a/judge/crates/coordinator/src/main.rs +++ b/judge/crates/coordinator/src/main.rs @@ -14,6 +14,9 @@ struct Cli { config: Option, #[arg(long, env = "KF_COORDINATOR_LISTEN", default_value = "127.0.0.1:8100")] listen: SocketAddr, + /// One bounded, operator-only repair from an already epoch-bound final report. + #[arg(long, requires = "config")] + backfill_performance: Option, } async fn run() -> Result<(), Box> { let cli = Cli::parse(); @@ -25,7 +28,18 @@ async fn run() -> Result<(), Box> { database: Database::new(&c.database_url)?, admission: c.admission.clone(), lease_seconds: c.lease_seconds, + measurement_epochs: c + .workers + .iter() + .filter_map(|w| w.measurement_epoch.clone().map(|epoch| (w.id, epoch))) + .collect(), }; + if let Some(evaluation) = cli.backfill_performance { + let evaluation = serde_json::from_value(serde_json::json!(evaluation))?; + let created = repository.backfill_performance(evaluation).await?; + println!("trusted performance rows created: {created}"); + return Ok(()); + } let r = repository.clone(); let period = c.recovery_seconds; task = Some(tokio::spawn(async move { diff --git a/judge/crates/coordinator/src/performance.rs b/judge/crates/coordinator/src/performance.rs new file mode 100644 index 0000000..b15671b --- /dev/null +++ b/judge/crates/coordinator/src/performance.rs @@ -0,0 +1,113 @@ +//! Same-transaction final projections. An old unbound attempt stays unbound. +use crate::{ + database::{Error, Result}, + repository::hash, +}; +use kf_contracts::{ + judging::Digest, + judging_versions::{JobManifest, JudgementReport}, +}; +use serde_json::{Value, json}; +use tokio_postgres::Transaction; + +pub fn comparison_group(manifest: &JobManifest, epoch: &str) -> Result { + let _: Digest = epoch.to_owned().try_into().map_err(|_| Error::Invalid)?; + let value = serde_json::to_value(manifest).map_err(|_| Error::Invalid)?; + let group = json!({ + "version": 1, "measurement_epoch": epoch, + "schema_version": value["schema_version"], + "problem_version_id": value["problem_version_id"], + "language": value["language"], "toolchain": value["toolchain"], + "compile_limits": value["compile_limits"], "run_limits": value["run_limits"], + "job_wall_time_us": value["job_wall_time_us"], + }); + Ok(hash( + &serde_json::to_vec(&group).map_err(|_| Error::Invalid)?, + )) +} + +pub async fn record( + transaction: &Transaction<'_>, + manifest: &JobManifest, + report: &JudgementReport, + digest: &str, +) -> Result { + let fields = report.fields(); + let b = fields.binding; + let Some(row) = transaction + .query_opt( + "SELECT a.measurement_epoch,s.id,s.language,e.updated_at FROM evaluations e \ + JOIN judge_attempts a ON a.id=e.current_attempt_id AND a.evaluation_id=e.id \ + JOIN submissions s ON s.id=e.submission_id WHERE e.id=$1 AND a.id=$2 \ + AND e.state='finished' AND e.verdict='accepted' AND e.failure_code IS NULL \ + AND a.cleanup_complete IS TRUE AND e.final_report_digest=$3 AND a.report_digest=$3", + &[&b.evaluation_id.as_uuid(), &b.attempt_id.as_uuid(), &digest], + ) + .await? + else { + return Ok(0); + }; + let Some(epoch) = row.get::<_, Option>("measurement_epoch") else { + return Ok(0); + }; + let group = comparison_group(manifest, &epoch)?; + let value: Value = serde_json::to_value(report).map_err(|_| Error::Invalid)?; + let measurement = |name: &str| -> Result> { + if value[name].is_null() { + return Ok(None); + } + let value = value[name].as_u64().ok_or(Error::Invalid)?; + Ok(Some(i64::try_from(value).map_err(|_| Error::Invalid)?)) + }; + let cpu = measurement("run_cpu_time_us")?; + let memory = measurement("run_memory_peak_bytes")?; + let submission: uuid::Uuid = row.get("id"); + let language: String = row.get("language"); + let finished: std::time::SystemTime = row.get("updated_at"); + Ok(transaction + .execute( + "INSERT INTO evaluation_performance(evaluation_id,submission_id,attempt_id,\ + problem_version_id,language,comparison_group,measurement_epoch,report_digest,\ + finished_at,cpu_time_us,memory_peak_bytes) \ + VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT(evaluation_id) DO NOTHING", + &[ + &b.evaluation_id.as_uuid(), + &submission, + &b.attempt_id.as_uuid(), + &manifest.problem_version_id().as_uuid(), + &language, + &group, + &epoch, + &digest, + &finished, + &cpu, + &memory, + ], + ) + .await?) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn grouping_ignores_submission_identity_and_separates_epoch_and_language() { + let bytes = + include_bytes!("../../../../tests/fixtures/contracts/judge-manifest.valid.json"); + let mut value: Value = serde_json::from_slice(bytes).unwrap(); + let manifest: JobManifest = serde_json::from_value(value.clone()).unwrap(); + let epoch = "a".repeat(64); + let group = comparison_group(&manifest, &epoch).unwrap(); + value["evaluation_id"] = json!("11111111-1111-4111-8111-111111111111"); + value["submission_id"] = json!("22222222-2222-4222-8222-222222222222"); + value["source"]["digest"] = json!("c".repeat(64)); + let other: JobManifest = serde_json::from_value(value.clone()).unwrap(); + assert_eq!(group, comparison_group(&other, &epoch).unwrap()); + assert_ne!(group, comparison_group(&other, &"b".repeat(64)).unwrap()); + value["language"] = json!("c17"); + let other: JobManifest = serde_json::from_value(value).unwrap(); + assert_ne!(group, comparison_group(&other, &epoch).unwrap()); + assert!(comparison_group(&manifest, "unverified").is_err()); + } +} diff --git a/judge/crates/coordinator/src/repository.rs b/judge/crates/coordinator/src/repository.rs index a30bc6e..09a45ac 100644 --- a/judge/crates/coordinator/src/repository.rs +++ b/judge/crates/coordinator/src/repository.rs @@ -6,6 +6,7 @@ use kf_contracts::judging_versions::{ use kf_contracts::{SchemaVersion, judging::*, judging_v2 as v2}; use serde_json::{Value, json}; use sha2::{Digest as _, Sha256}; +use std::collections::HashMap; use tokio_postgres::{Row, Transaction}; use uuid::Uuid; #[derive(Clone)] @@ -13,6 +14,7 @@ pub struct Repository { pub database: Database, pub admission: NodeAdmission, pub lease_seconds: u32, + pub measurement_epochs: HashMap, } pub fn identifier(v: Uuid) -> Identifier { v.to_string().try_into().expect("non-nil generated UUID") @@ -165,7 +167,8 @@ impl Repository { .checked_add(1) .ok_or(Error::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?; + let epoch = self.measurement_epochs.get(&worker).map(|e| e.as_str()); + 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,measurement_epoch) VALUES($1,$2,$3,$4,$5,$6,clock_timestamp()+$7::bigint*interval '1 second',$8,$9,$10) 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,&epoch]).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::for_manifest( &m, @@ -401,9 +404,46 @@ impl Repository { fields.cleanup_complete, ) .await?; + crate::performance::record(&t, &m, r, &digest).await?; t.commit().await?; Ok(()) } + pub async fn backfill_performance(&self, evaluation: Identifier) -> Result { + let mut s = self.database.session().await?; + let t = s.client.transaction().await?; + let row = t + .query_opt( + "SELECT final_report FROM evaluations WHERE id=$1 FOR UPDATE", + &[&evaluation.as_uuid()], + ) + .await? + .ok_or(Error::Rejected)?; + let value: Value = row.get::<_, Option>(0).ok_or(Error::Rejected)?; + let report: JudgementReport = as_wire(value.clone())?; + let b = report.fields().binding; + if b.evaluation_id != evaluation { + return Err(Error::Rejected); + } + let e = self.bound(&t, b, None).await?; + let manifest: JobManifest = as_wire(e.get("manifest"))?; + let manifest_digest: Digest = e + .get::<_, String>("manifest_digest") + .try_into() + .map_err(invalid)?; + report + .validate_for_manifest(&manifest, &manifest_digest) + .map_err(invalid)?; + report.validate_program_budget(&manifest).map_err(invalid)?; + let digest = hash(&serde_json::to_vec(&as_value(&report)?).map_err(invalid)?); + if e.get::<_, Option>("final_report_digest").as_deref() != Some(&digest) + || e.get::<_, Option>("report_digest").as_deref() != Some(&digest) + { + return Err(Error::Rejected); + } + let count = crate::performance::record(&t, &manifest, &report, &digest).await?; + t.commit().await?; + Ok(count) + } pub async fn cancel(&self, eid: Identifier) -> Result<()> { let mut s = self.database.session().await?; let t = s.client.transaction().await?;