Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions server/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
32 changes: 32 additions & 0 deletions server/telemetry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
35 changes: 35 additions & 0 deletions server/tests/test_quality_control.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down
26 changes: 26 additions & 0 deletions server/tests/test_telemetry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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() == []
Expand Down
Loading