From 776b300ebc7fdd8dce2c1069f0278454512c914e Mon Sep 17 00:00:00 2001 From: kai392 Date: Fri, 24 Jul 2026 20:54:59 +0800 Subject: [PATCH] fix: commit evaluation progress so it is visible while the run is in flight MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_touch_progress()` ended in `session.flush()` and the worker claimed its job with `session.flush()` too. A flush only writes into the worker's own transaction, and `process_once()` holds that single transaction open across the whole job body — up to `EVAL_TIMEOUT_SECONDS`, 2h by default. Under Postgres READ COMMITTED the API process cannot read uncommitted rows, so: * every parsed `[submission] item i/N` progress line was invisible until the run had already finished, leaving `current_phase` / `current_message` / `current_progress_*` blank on `/api/submissions/{id}` and on the public submission page for the entire evaluation; * `/api/jobs` reported a claimed job as `queued`, and the submission as un-started, for hours. Commit in both places instead. Committing the claim also makes it durable and releases the `SELECT ... FOR UPDATE` row lock early; the committed `queued` -> `running` transition is what keeps other workers off the row, so `skip_locked` claiming is unaffected. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/JOURNAL.md | 9 ++ .../src/eval_backend/services/eval_runner.py | 8 +- validator/src/eval_backend/worker.py | 8 +- validator/tests/test_progress_visibility.py | 146 ++++++++++++++++++ 4 files changed, 169 insertions(+), 2 deletions(-) create mode 100644 validator/tests/test_progress_visibility.py diff --git a/docs/JOURNAL.md b/docs/JOURNAL.md index 13a8715..d84b169 100644 --- a/docs/JOURNAL.md +++ b/docs/JOURNAL.md @@ -18,6 +18,15 @@ protocol. **Newest entries at the top.** Tag each entry with one or more of: --- +## 2026-07-24 — live evaluation progress was computed and then thrown away #mistake #gotcha +**Context:** `eval_runner` parses the `[submission] item i/N ...` lines the harness prints and stores them on the run (`phase`, `message`, `progress_current/total`); `/api/submissions/{id}` exposes them as `current_phase`/`current_message`/`current_progress_*` and the public submission page renders them. +**Expected:** the progress fields advance while an evaluation is running. +**Actual:** they stayed empty for the entire run and only appeared once it had already finished — the progress bar was always either blank or complete, never moving. `/api/jobs` likewise reported a job as `queued` for hours after a worker had claimed it. +**Root cause:** `_touch_progress()` ended in `session.flush()`, and the worker claimed its job with `session.flush()` too. A flush writes into the *worker's own* transaction — and `process_once()` holds that single transaction open across the whole job body, up to `EVAL_TIMEOUT_SECONDS` (2h by default). Under Postgres READ COMMITTED the API process cannot see uncommitted rows, so every progress update was invisible until the final `session.commit()`, by which point it was worthless. +**Fix / decision:** commit instead of flush in both places. Committing the claim also makes it durable and releases the `SELECT ... FOR UPDATE` row lock early; the `status = 'queued' -> 'running'` transition is what actually keeps other workers off the row, so `skip_locked` claiming is unaffected. Regression tests in `validator/tests/test_progress_visibility.py` assert that `_touch_progress` commits and that the worker has committed at least once before the job body starts. +**Gotcha for the test suite:** the `validator_session` fixture binds the Session to a Connection that already has an open transaction, so SQLAlchemy 2.0 joins it via SAVEPOINT — an inner `session.commit()` is still undone by the fixture's outer `transaction.rollback()`. Verified before relying on it; tests stay isolated. +**Follow-up:** `_local_attempt()` still does not forward an `on_line` callback to `_run_bash_stream()`, so local / local-fallback runs parse no progress lines at all even now that the updates are durable. Left out of this change to keep it reviewable. + ## 2026-07-12 — code grader: add resource limits on top of the HOME/secrets fix #security #decision **Context:** issue #71 — the code grader (`run_pass_at_1`) runs untrusted miner/LLM candidate code. The core secret-leak fix (isolated throwaway HOME/cwd, scrubbed env, `python -I`) already landed on main. **Finding:** main's sandbox closes the HOME/secrets exfiltration vector but has **no resource limits** — an untrusted candidate can still exhaust host memory or fork-bomb the eval box within its wall-clock timeout (verified: a 4 GiB `bytearray` allocation runs to completion and "passes" on main). diff --git a/validator/src/eval_backend/services/eval_runner.py b/validator/src/eval_backend/services/eval_runner.py index f06cb0b..fc7e13b 100644 --- a/validator/src/eval_backend/services/eval_runner.py +++ b/validator/src/eval_backend/services/eval_runner.py @@ -240,7 +240,13 @@ def _touch_progress( run.status = status submission.status = status submission.updated_at = _utcnow() - session.flush() + # Commit, not flush: the whole point of these updates is that another process + # (the API serving /api/submissions/{id}) can observe them *while* the + # evaluation is still running. A flush only makes them visible inside this + # transaction, and the worker holds that transaction open for the entire run + # (up to EVAL_TIMEOUT_SECONDS, 2h by default), so every progress line was + # discarded until the run was already over. + session.commit() def _consume_progress_line( diff --git a/validator/src/eval_backend/worker.py b/validator/src/eval_backend/worker.py index 3497ed3..a712c1b 100644 --- a/validator/src/eval_backend/worker.py +++ b/validator/src/eval_backend/worker.py @@ -84,7 +84,13 @@ def process_once(session_factory, settings: Settings) -> int: now = datetime.now(timezone.utc) job.claimed_at = now job.heartbeat_at = now - session.flush() + # Commit the claim before the (long) job body runs. A flush kept the claim + # inside this transaction, which stays open for the whole evaluation, so + # /api/jobs reported the job as "queued" and the submission as un-started + # for hours. Committing also makes the claim durable and releases the + # SELECT ... FOR UPDATE row lock; the status is now "queued" -> "running" + # in the database, which is what keeps other workers off this row. + session.commit() payload = job.payload_json or {} job_id = job.id submission_id = job.submission_id diff --git a/validator/tests/test_progress_visibility.py b/validator/tests/test_progress_visibility.py new file mode 100644 index 0000000..7996e0f --- /dev/null +++ b/validator/tests/test_progress_visibility.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +from eval_backend import worker +from eval_backend.core.config import Settings +from eval_backend.models import Artifact, EvaluationRun, JobQueue, Submission +from eval_backend.services import eval_runner +from eval_backend.services.eval_runner import EvaluationResult + + +def _build_settings(tmp_path: Path) -> Settings: + return Settings( + workspace_root=tmp_path / "workspaces", + artifact_root=tmp_path / "artifacts", + local_repo_dir=tmp_path, + eval_execution_mode="local_cpu", + eval_max_items=2, + ) + + +def _add_submission(session, checkpoint_path: Path, submission_id: str) -> Submission: + artifact = Artifact( + id=f"artifact-{submission_id}", + storage_backend="local", + storage_uri=str(checkpoint_path), + file_names_json=[checkpoint_path.name], + sha256="abc123", + size_bytes=checkpoint_path.stat().st_size, + mime_type="application/octet-stream", + meta_json={"checkpoint_path": str(checkpoint_path)}, + ) + session.add(artifact) + session.flush() + submission = Submission( + id=submission_id, + source="upload", + miner_id="miner-a", + benchmark_names_json=["math500"], + status="queued", + submission_artifact_id=artifact.id, + ) + session.add(submission) + session.flush() + artifact.submission_id = submission.id + session.flush() + return submission + + +def test_touch_progress_is_committed_not_just_flushed(validator_session, tmp_path, monkeypatch): + """Progress only helps if another process can read it mid-run.""" + session = validator_session + checkpoint_path = tmp_path / "theta.npy" + checkpoint_path.write_bytes(b"theta") + submission = _add_submission(session, checkpoint_path, "sub-progress") + run = EvaluationRun( + submission_id=submission.id, + benchmark_names_json=["math500"], + status="running", + ) + session.add(run) + session.flush() + + committed: list[bool] = [] + monkeypatch.setattr(session, "commit", lambda: committed.append(True)) + + eval_runner._touch_progress( + session, + run, + submission, + phase="evaluation_running", + message="item 3/20 running", + current=3, + total=20, + ) + + assert committed, "progress update was never committed, so the API cannot observe it" + assert run.phase == "evaluation_running" + assert run.progress_current == 3 + assert run.progress_total == 20 + + +def test_worker_commits_the_claim_before_the_job_body_runs(validator_session, tmp_path, monkeypatch): + """The claim must be durable before the (up to 2h) evaluation starts.""" + session = validator_session + settings = _build_settings(tmp_path) + checkpoint_path = tmp_path / "theta.npy" + checkpoint_path.write_bytes(b"theta") + submission = _add_submission(session, checkpoint_path, "sub-claim") + session.add( + JobQueue( + id="job-claim", + job_type="evaluation", + job_id=submission.id, + submission_id=submission.id, + queue_name="default", + status="queued", + priority=0, + dedupe_key="evaluation:sub-claim", + attempts=0, + max_attempts=3, + payload_json={"submission_id": submission.id}, + ) + ) + session.flush() + + commits: list[bool] = [] + real_commit = session.commit + + def _tracking_commit() -> None: + commits.append(True) + real_commit() + + monkeypatch.setattr(session, "commit", _tracking_commit) + # process_once() closes the session it is handed; the fixture owns this one. + monkeypatch.setattr(session, "close", lambda: None) + + observed: dict[str, object] = {} + + def _fake_evaluate_submission(session_, submission_, settings_, **kwargs): + observed["commits_before_eval"] = len(commits) + observed["job_status"] = session_.get(JobQueue, "job-claim").status + now = datetime.now(timezone.utc) + run = EvaluationRun( + submission_id=submission_.id, + benchmark_names_json=["math500"], + status="completed", + score=0.5, + started_at=now, + finished_at=now, + ) + session_.add(run) + session_.flush() + return EvaluationResult(run=run, score=0.5, metrics={}, stdout="", stderr="") + + monkeypatch.setattr(worker, "evaluate_submission", _fake_evaluate_submission) + + processed = worker.process_once(lambda: session, settings) + + assert processed == 1 + assert observed["job_status"] == "running" + assert observed["commits_before_eval"] >= 1, ( + "the worker held the claim in an uncommitted transaction for the whole run, " + "so /api/jobs still reported the job as queued" + )