fix: a crashed job's crash line carries its job id
The bind sat inside the try, so it was already reset when the crash line and the cleanup were logged — the one line most worth finding by job id had none.
This commit is contained in:
@@ -493,14 +493,16 @@ def _worker_loop(idx: int) -> None:
|
|||||||
if job_id is None:
|
if job_id is None:
|
||||||
_stop.wait(settings.download_poll_seconds)
|
_stop.wait(settings.download_poll_seconds)
|
||||||
continue
|
continue
|
||||||
try:
|
# Tag every line this job produces (including the crash below) with the job id: a worker runs
|
||||||
# Tag every line this job produces (including the crash below) with the job id: a worker
|
# N jobs at once on N threads, so without it the interleaved output is unreadable. The bind
|
||||||
# runs N jobs at once on N threads, so without it the interleaved output is unreadable.
|
# wraps the handler too — bound inside the `try`, it was already reset by the time the crash
|
||||||
with logs.bind(job_id=job_id):
|
# line was logged.
|
||||||
|
with logs.bind(job_id=job_id):
|
||||||
|
try:
|
||||||
_process_job(job_id)
|
_process_job(job_id)
|
||||||
except Exception: # noqa: BLE001
|
except Exception: # noqa: BLE001
|
||||||
log.exception("processing job %s crashed", job_id)
|
log.exception("processing job %s crashed", job_id)
|
||||||
_fail_running(job_id, error="Worker error", error_code=None)
|
_fail_running(job_id, error="Worker error", error_code=None)
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
def main() -> None:
|
||||||
|
|||||||
@@ -236,3 +236,37 @@ def test_stop_child_noop_if_already_exited(killpg_rec):
|
|||||||
proc.returncode = 0 # already gone
|
proc.returncode = 0 # already gone
|
||||||
worker._stop_child(proc, grace_s=0.01)
|
worker._stop_child(proc, grace_s=0.01)
|
||||||
assert killpg_rec == []
|
assert killpg_rec == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_crashed_jobs_crash_line_and_cleanup_carry_the_job_id(monkeypatch):
|
||||||
|
"""The job id is bound for the WHOLE job, the crash handler included: bound inside the `try`, it
|
||||||
|
was already reset when the one line that says the job crashed was logged."""
|
||||||
|
from app import logs
|
||||||
|
|
||||||
|
seen: dict[str, object] = {}
|
||||||
|
claims = iter([42])
|
||||||
|
|
||||||
|
def claim():
|
||||||
|
job = next(claims, None)
|
||||||
|
if job is None:
|
||||||
|
worker._stop.set()
|
||||||
|
return job
|
||||||
|
|
||||||
|
def crash(_job_id):
|
||||||
|
raise RuntimeError("boom")
|
||||||
|
|
||||||
|
monkeypatch.setattr(worker, "_claim_job", claim)
|
||||||
|
monkeypatch.setattr(worker, "_process_job", crash)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
worker.log, "exception", lambda *a, **k: seen.setdefault("crash", logs._job_id.get())
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
worker, "_fail_running", lambda *a, **k: seen.setdefault("cleanup", logs._job_id.get())
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(worker.settings, "download_poll_seconds", 0)
|
||||||
|
worker._stop.clear()
|
||||||
|
try:
|
||||||
|
worker._worker_loop(0)
|
||||||
|
finally:
|
||||||
|
worker._stop.clear()
|
||||||
|
assert seen == {"crash": 42, "cleanup": 42}
|
||||||
|
|||||||
Reference in New Issue
Block a user