From 74753ff6b8e28ad840dfebd12c52db3c6a30f5d9 Mon Sep 17 00:00:00 2001 From: Jules GRIVOT PELISSON Date: Thu, 29 Oct 2026 16:04:11 -0400 Subject: [PATCH] FIX: keep a resumed run's tool name when the server restarted while it was paused --- server/main.py | 7 ++++++ server/telemetry.py | 32 +++++++++++++++++++++++++ server/tests/test_quality_control.py | 35 ++++++++++++++++++++++++++++ server/tests/test_telemetry.py | 26 +++++++++++++++++++++ 4 files changed, 100 insertions(+) diff --git a/server/main.py b/server/main.py index d2fdfd3..62ae705 100644 --- a/server/main.py +++ b/server/main.py @@ -3265,6 +3265,13 @@ async def resume_run(run_id: str, request: Request, tool = get_tool(tool_name) except KeyError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) + # A restart while the run waited at its checkpoint dropped its ledger + # record; the name it was registered under is still beside its events. + registered = await anyio.to_thread.run_sync(runs.meta, run_id) + telemetry.record_run_resumed( + run_id, registered.get("tool") or tool_name, + registered.get("client"), batch=registered.get("batch"), + started_at=registered.get("at")) if not already_staged: # A rewind stages before it narrows, and the upload's handles are # read once: staging the same form twice would find every file empty diff --git a/server/telemetry.py b/server/telemetry.py index ef39fef..2ccbbc6 100644 --- a/server/telemetry.py +++ b/server/telemetry.py @@ -372,6 +372,38 @@ def record_run_start(run_id: str, tool: str, client: str = None, pass +def record_run_resumed(run_id: str, tool: str, client: str = None, + batch: Optional[dict] = None, + started_at: Optional[float] = None) -> None: + """A paused run is carried on. Fills in only what the ledger lacks. + + A run that paused before this process started is not in the ledger: the + history keeps finished runs only, so a restart while a run waits at its + checkpoint loses the record `record_run_start` made. Without this, the + first grant after the resume creates a blank one and the run is listed + under no tool at all. `started_at` is when the run was first registered, + so it reads the same as a run that paused and resumed without a restart. + """ + try: + with _ledger_lock: + record = _ledger_record(run_id) + if record is None: + return + if not record["tool"] and tool: + record["tool"] = str(tool)[:100] + if not record["client"] and client: + record["client"] = str(client)[:64] + if not record["batch"] and batch: + record["batch"] = dict(batch) + if started_at and record["ended_at"] is None: + try: + record["started_at"] = min(record["started_at"], float(started_at)) + except (TypeError, ValueError): + pass + except Exception: # noqa: BLE001 - telemetry must never fail a run + pass + + # How many files a single input is walked over before its size is reported as # partial. Deliberately smaller than the disk sweep's: this runs on the request # path, microseconds before the tool is invoked, and an unpacked cohort can be diff --git a/server/tests/test_quality_control.py b/server/tests/test_quality_control.py index 9831f57..70d2807 100644 --- a/server/tests/test_quality_control.py +++ b/server/tests/test_quality_control.py @@ -755,6 +755,41 @@ async def carried_on(tool_name, request, background_tasks, resume_from=None): real_finish(run_id, runs.PHASE_DONE) +def test_a_run_resumed_after_a_restart_is_listed_under_its_tool(tmp_path, monkeypatch): + """The ledger lives in memory and keeps finished runs only, so a server + restarted while a run waited at its checkpoint resumed it with no record: + the admin panel then listed the run as "(unnamed)".""" + from fastapi.testclient import TestClient + import main + import telemetry + from wire import runs + + job = _ran(tmp_path, "01_ALI_CBCT") + with open(os.path.join(job, "job.json"), "w", encoding="utf-8") as handle: + json.dump({"tool": "Test_Tool"}, handle) + run_id = runs.register(uuid.uuid4().hex, tool="Test_Tool", client="a-workstation") + runs.pause(run_id, job, "ALI_CBCT") + telemetry.reset() # the restart + + async def carried_on(tool_name, request, background_tasks, resume_from=None): + return {"carried": "on"} + + monkeypatch.setattr(main, "_run_tool", carried_on) + real_finish = runs.finish + monkeypatch.setattr(runs, "finish", lambda rid, phase, **kw: None) + + client = TestClient(main.app) + response = client.post( + f"/runs/{run_id}/resume", + headers={"Authorization": "Bearer " + main.settings.API_TOKEN}, + ) + assert response.status_code == 200 + record = telemetry.ledger_record(run_id) + assert record is not None + assert (record["tool"], record["client"]) == ("Test_Tool", "a-workstation") + real_finish(run_id, runs.PHASE_DONE) + + class _Background: """Enough of BackgroundTasks to record what was queued.""" diff --git a/server/tests/test_telemetry.py b/server/tests/test_telemetry.py index f3b5dc0..c74852c 100644 --- a/server/tests/test_telemetry.py +++ b/server/tests/test_telemetry.py @@ -186,6 +186,32 @@ def test_the_ledger_is_bounded_and_keeps_the_newest(): assert ledger[0]["run_id"] == f"run-{telemetry.LEDGER_SIZE + 19}" +def test_a_run_resumed_after_a_restart_keeps_its_name(): + """A restart while a run waits at its checkpoint drops its record; the + grant after the resume used to make a blank one, listed under no tool.""" + telemetry.record_run_resumed("run-r", "AREG", "a-workstation", + started_at=1000.0) + telemetry.record_run_grant("run-r", channels=1) + telemetry.record_run_end("run-r", "done", "done") + + record = telemetry.run_ledger()[0] + assert record["tool"] == "AREG" + assert record["client"] == "a-workstation" + assert record["started_at"] == 1000.0 + assert [row["tool"] for row in telemetry.tool_uptime()] == ["AREG"] + + +def test_a_resume_does_not_rename_a_run_the_ledger_already_knows(): + telemetry.record_run_start("run-k", "AREG", "first") + started = telemetry.run_ledger()[0]["started_at"] + telemetry.record_run_resumed("run-k", "ASO", "second", + started_at=started + 60) + + record = telemetry.run_ledger()[0] + assert (record["tool"], record["client"]) == ("AREG", "first") + assert record["started_at"] == started + + def test_ending_a_run_nobody_started_is_ignored(): telemetry.record_run_end("never-seen", "done") assert telemetry.run_ledger() == []