Unstick analysis running when the worker is dead.
A container rebuild left Emmaforo status=running with no process, so PCB start kept 409. Heal on GET now matches placement/PCB: complete if report.json exists, otherwise error.
This commit is contained in:
@@ -308,46 +308,68 @@ def mark_stale_running(
|
||||
def heal_if_pipeline_finished(
|
||||
storage: StorageBackend, user_id: str, project_id: str,
|
||||
) -> ProjectMeta | None:
|
||||
"""If meta says queued/running but events already ended with
|
||||
``pipeline_complete``, flip status to ``complete``.
|
||||
"""Unstick analysis ``queued``/``running`` when the worker is gone.
|
||||
|
||||
Covers zombies where the worker wrote the terminal event (and often
|
||||
the report) then died before the meta transition — e.g. container
|
||||
rebuild mid-shutdown. Returns updated meta, or ``None`` if no heal.
|
||||
- Last event ``pipeline_complete`` → ``complete``
|
||||
- Dead worker + ``report.json`` → ``complete``
|
||||
- Dead worker otherwise → ``error`` (so PCB/placement can start)
|
||||
"""
|
||||
meta = get_project(storage, user_id, project_id)
|
||||
if meta is None or meta.status not in (STATUS_RUNNING, STATUS_QUEUED):
|
||||
return None
|
||||
|
||||
events_prefix = f"{_project_prefix(user_id, project_id)}/events/"
|
||||
prefix = _project_prefix(user_id, project_id)
|
||||
events_prefix = f"{prefix}/events/"
|
||||
last = None
|
||||
try:
|
||||
keys = storage.list_prefix(events_prefix)
|
||||
event_keys = sorted(
|
||||
k for k in storage.list_prefix(events_prefix)
|
||||
if k.endswith(".json") and "/events/" in k
|
||||
)
|
||||
if event_keys:
|
||||
last = storage.read_json(event_keys[-1])
|
||||
except Exception:
|
||||
return None
|
||||
event_keys = sorted(
|
||||
k for k in keys if k.endswith(".json") and "/events/" in k
|
||||
)
|
||||
if not event_keys:
|
||||
return None
|
||||
last = None
|
||||
|
||||
if (last or {}).get("event") == "pipeline_complete":
|
||||
summary = (last.get("data") or {}).get("summary")
|
||||
try:
|
||||
return transition_status(
|
||||
storage, user_id, project_id,
|
||||
from_status={STATUS_RUNNING, STATUS_QUEUED},
|
||||
to_status=STATUS_COMPLETE,
|
||||
summary=summary if isinstance(summary, dict) else meta.summary,
|
||||
cancel_requested=False,
|
||||
pipeline_state=None,
|
||||
)
|
||||
except StatusConflict:
|
||||
return None
|
||||
|
||||
from backend.services import job_runner
|
||||
|
||||
exec_name = meta.execution_name or f"local/projects/{project_id}"
|
||||
try:
|
||||
last = storage.read_json(event_keys[-1])
|
||||
state = job_runner.get_execution_state(exec_name)
|
||||
except Exception:
|
||||
return None
|
||||
if (last or {}).get("event") != "pipeline_complete":
|
||||
state = "unknown"
|
||||
if state in ("pending", "running"):
|
||||
return None
|
||||
|
||||
summary = (last.get("data") or {}).get("summary")
|
||||
try:
|
||||
return transition_status(
|
||||
storage, user_id, project_id,
|
||||
from_status={STATUS_RUNNING, STATUS_QUEUED},
|
||||
to_status=STATUS_COMPLETE,
|
||||
summary=summary if isinstance(summary, dict) else meta.summary,
|
||||
cancel_requested=False,
|
||||
pipeline_state=None,
|
||||
)
|
||||
except StatusConflict:
|
||||
return None
|
||||
if storage.exists(f"{prefix}/report.json"):
|
||||
try:
|
||||
return transition_status(
|
||||
storage, user_id, project_id,
|
||||
from_status={STATUS_RUNNING, STATUS_QUEUED},
|
||||
to_status=STATUS_COMPLETE,
|
||||
cancel_requested=False,
|
||||
pipeline_state=None,
|
||||
)
|
||||
except StatusConflict:
|
||||
return None
|
||||
return mark_stale_running(
|
||||
storage, user_id, project_id,
|
||||
f"Analysis worker terminated ({state})",
|
||||
)
|
||||
|
||||
|
||||
def heal_if_placement_stuck(
|
||||
|
||||
Reference in New Issue
Block a user