From 86bc008a815838110f374b2ea6c95381f1762bfc Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 17 Sep 2026 03:53:29 +0200 Subject: [PATCH] fix: a crashed job's crash line carries its job id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- backend/app/worker.py | 16 +++++++----- backend/tests/test_worker_supervisor.py | 34 +++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 7 deletions(-) diff --git a/backend/app/worker.py b/backend/app/worker.py index cecc2ac6..60e2c2fc 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -493,14 +493,16 @@ def _worker_loop(idx: int) -> None: if job_id is None: _stop.wait(settings.download_poll_seconds) continue - try: - # Tag every line this job produces (including the crash below) with the job id: a worker - # runs N jobs at once on N threads, so without it the interleaved output is unreadable. - with logs.bind(job_id=job_id): + # Tag every line this job produces (including the crash below) with the job id: a worker runs + # N jobs at once on N threads, so without it the interleaved output is unreadable. The bind + # wraps the handler too — bound inside the `try`, it was already reset by the time the crash + # line was logged. + with logs.bind(job_id=job_id): + try: _process_job(job_id) - except Exception: # noqa: BLE001 - log.exception("processing job %s crashed", job_id) - _fail_running(job_id, error="Worker error", error_code=None) + except Exception: # noqa: BLE001 + log.exception("processing job %s crashed", job_id) + _fail_running(job_id, error="Worker error", error_code=None) def main() -> None: diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py index 1898bc6f..bc4adb38 100644 --- a/backend/tests/test_worker_supervisor.py +++ b/backend/tests/test_worker_supervisor.py @@ -236,3 +236,37 @@ def test_stop_child_noop_if_already_exited(killpg_rec): proc.returncode = 0 # already gone worker._stop_child(proc, grace_s=0.01) 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}