Mirko Holler
30b5119a2f
Type-tags queue jobs with a "kind" field (back-compat: missing kind defaults to "tomo"), and adds "command" jobs that reconfigure the beamline (currently: absolute moves on an allow-listed device, plus an optimize_idgap stub) between tomograms instead of running a scan. Command jobs are added via tomo_queue_add_command() against a curated named-action registry (_TOMO_QUEUE_ACTIONS) - not arbitrary code - so every queued action is auditable and safe to persist across a kernel restart. Validation is two-layer: schema checks at add time, and each action re-checks its own arguments against live hardware at execution time, which is the actual safety boundary. Crash-resume for a command job re-runs the whole step sequence from the top if every step is idempotent, or prompts the operator otherwise. flomni_webpage_generator.py is guarded against command jobs, which have no "params" key. See AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md for the full design.
90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
"""Checklist section B5 -- real crash-resume via SIGKILL. See TOMO_QUEUE_TESTING.md.
|
|
|
|
Needs a real OS process to kill: a SIGKILLed process never reaches
|
|
tomo_queue_execute()'s except-block, so the running job's status is left
|
|
stuck at "running" -- exactly the "kernel crashed mid-scan" scenario the doc
|
|
calls for, and not reproducible by raising an exception in-process (that's
|
|
B6, in test_tomo_queue_crash.py).
|
|
"""
|
|
|
|
import signal
|
|
|
|
from _queue_helpers import add_short_job, spawn_queue_subprocess, wait_until
|
|
|
|
|
|
def test_real_crash_resume(flomni_sim, bec_services_config_file_path, tmp_path):
|
|
flomni = flomni_sim
|
|
|
|
add_short_job(flomni, "job1")
|
|
add_short_job(flomni, "job2")
|
|
jobs = flomni._tomo_queue_proxy.as_list()
|
|
job1_id, job2_id = jobs[0]["id"], jobs[1]["id"]
|
|
|
|
flomni.progress.reset()
|
|
|
|
proc = spawn_queue_subprocess(bec_services_config_file_path, tmp_path / "run1.log")
|
|
try:
|
|
# Wait until job1 is genuinely mid-scan (subtomo 3 of 8, at
|
|
# FAST_STEPSIZE's N=1-projection-per-subtomo) before killing -- not
|
|
# the very first projection, so this is unambiguously a *resume*,
|
|
# not indistinguishable from a fresh start.
|
|
wait_until(lambda: (flomni.progress.get("subtomo") or 0) >= 3, timeout=90)
|
|
|
|
# One atomic snapshot -- reading subtomo/angle/heartbeat as three
|
|
# separate round-trips could catch values from different instants.
|
|
before = flomni.progress.as_dict()
|
|
|
|
proc.send_signal(signal.SIGKILL)
|
|
proc.wait(timeout=10)
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.wait(timeout=10)
|
|
|
|
jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
|
|
assert jobs[job1_id]["status"] == "running", (
|
|
"a SIGKILLed job never reaches tomo_queue_execute()'s except-block, so "
|
|
f"it should be stuck at 'running', not {jobs[job1_id]['status']!r}"
|
|
)
|
|
assert jobs[job2_id]["status"] == "pending", "job2 must be untouched by job1's crash"
|
|
|
|
# Resume: a second, fresh subprocess (a stand-in for "operator restarts
|
|
# their kernel") should pick job1 up mid-scan, not restart it.
|
|
proc2 = spawn_queue_subprocess(bec_services_config_file_path, tmp_path / "run2.log")
|
|
try:
|
|
after = wait_until(
|
|
lambda: _snapshot_if_new_heartbeat(flomni, before["heartbeat"]), timeout=60
|
|
)
|
|
returncode = proc2.wait(timeout=180)
|
|
finally:
|
|
if proc2.poll() is None:
|
|
proc2.kill()
|
|
proc2.wait(timeout=10)
|
|
|
|
log_text = (tmp_path / "run2.log").read_text()
|
|
assert returncode == 0, f"resume subprocess failed (see log):\n{log_text[-4000:]}"
|
|
|
|
assert after["subtomo"] == before["subtomo"], (
|
|
f"resume should re-acquire the exact in-flight subtomo {before['subtomo']}, "
|
|
f"got {after['subtomo']}"
|
|
)
|
|
assert after["angle"] == before["angle"], (
|
|
f"resume should re-acquire the exact in-flight angle {before['angle']}, "
|
|
f"got {after['angle']}"
|
|
)
|
|
|
|
jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
|
|
assert jobs[job1_id]["status"] == "done", "job1 should have completed after resuming"
|
|
assert jobs[job2_id]["status"] == "done", "job2 should have run after job1 finished"
|
|
|
|
|
|
def _snapshot_if_new_heartbeat(flomni, heartbeat_before):
|
|
"""Return a full progress snapshot the first time ``heartbeat`` differs
|
|
from its pre-kill value (i.e. a fresh `_tomo_scan_at_angle` attempt has
|
|
started), else None so wait_until() keeps polling.
|
|
"""
|
|
snapshot = flomni.progress.as_dict()
|
|
if snapshot.get("heartbeat") != heartbeat_before:
|
|
return snapshot
|
|
return None
|