feat(submissions): bind final performance to trusted measurement epochs

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")

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):

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)

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())

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；存在纪元/投影时降级拒绝
+丢弃数据，须另行进行显式数据保留迁移，不能用删除成绩绕过。

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<Digest>,
 }
 #[derive(Deserialize)]
 #[serde(deny_unknown_fields)]

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;

judge/crates/coordinator/src/main.rs

@@ -14,6 +14,9 @@ struct Cli {
     config: Option<PathBuf>,
     #[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<uuid::Uuid>,
 }
 async fn run() -> Result<(), Box<dyn std::error::Error>> {
     let cli = Cli::parse();
@@ -25,7 +28,18 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
             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 {

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<String> {
+    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<u64> {
+    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<String>>("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<Option<i64>> {
+        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());
+    }
+}

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<Identifier, Digest>,
 }
 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<u64> {
+        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<Value>>(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<String>>("final_report_digest").as_deref() != Some(&digest)
+            || e.get::<_, Option<String>>("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?;