Reclaim the jobs of a worker that died without unwinding (#150) - #172
Merged
danielplohmann merged 2 commits intoSep 8, 2026
Merged
Conversation
A worker killed with SIGKILL (the OOM killer's way) never runs its context manager's exit, so its registration and the lock on the job it was running stay behind. Nothing cleared such a lock: next() only hands out unlocked jobs, and release_orphaned_jobs() only released jobs of unregistered workers, which a killed worker is not. The job was stranded for good, and worse, get_cached_job_id() handed its id to every identical resubmission, which then waited on work that could never run (danielplohmann#150, eight such jobs measured on a live instance). Registration alone is not liveness. Workers now write a heartbeat when they register and refresh it off their poll loop (throttled to a third of the queue timeout), and a registered worker whose heartbeat is older than the timeout is dead: release_orphaned_jobs() releases its unfinished jobs with one attempt fewer, failing a job out of attempts and notifying its dependents like the error path does, and drops the dead registration. It runs when a worker starts, as before, and from the poll loop once per timeout. get_cached_job_id() only serves a job that is finished, waiting, or in flight on a live worker. A registration without any heartbeat is a process from before heartbeats existed that was not shut down cleanly, and is treated as dead. The lock's own locked_at is still not used for this: it is only refreshed by progress reporting, so a healthy long job would look stale (see repair()).
…worker one timeout of grace Jobs run synchronously in the poll loop, so a job longer than the queue timeout left its worker without a heartbeat and another poller reclaimed the job while it was still being processed. A daemon thread now refreshes the heartbeat from registration to unregistration, and a heartbeat also re-adds the registration, so a worker judged dead by mistake serves the next jobs again. A registered worker without any heartbeat (a rolling upgrade: an older worker busy with a job) is stamped once and judged after one timeout, instead of being unregistered on sight.
This was referenced Sep 8, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #150 (a hard-killed worker strands its job forever, and the stranded job is then served to identical resubmissions).
Cause
A worker killed with SIGKILL (the OOM killer's way) never runs its context manager's exit, so its registration and the lock on the job it was running stay behind.
next()only hands out unlocked jobs, andrelease_orphaned_jobs()only released jobs of unregistered workers, which a killed worker is not. So the job stayed locked for good, andget_cached_job_id()handed its id to every identical resubmission, which then waited on work that could never run.Fix
Registration alone is not liveness, so workers now carry a heartbeat:
registerWorker()records a heartbeat next to the registration;next()refreshes it off the poll loop, throttled to a third of the queue timeout (one write per 100 s at the default 300 s).release_orphaned_jobs()releases the unfinished jobs of every worker that is not live, with one attempt fewer; a job out of attempts fails and notifies its dependents, likeJob.error()does; the queue counters follow. Dead registrations are dropped, so the worker list stops accumulating ids of long-gone processes. It runs when a worker starts, as before, and from the poll loop once per timeout.get_cached_job_id()only serves a job that is finished, waiting, or in flight on a live worker.locked_atis still not used for liveness, for the reasonrepair()documents (a healthy long job would look stale).Known limit, unchanged from before: if the parent of a spawning worker is killed while its child still runs the job, the child may still complete it after the reclaim, so the job can run twice. The timeout bounds when that window opens.
Verification
tests/testMongoQueue.pyWorkerLivenessTest(7 tests): registration writes and removal of the heartbeat; liveness by heartbeat age, including the no-heartbeat case; a dead worker's job is reclaimed with one attempt fewer while a live worker's is kept and the dead registration pruned; a job never registered for; reclaiming the last attempt fails the job and frees its dependent, with the failed counter incremented; the cached-job lookup skips the stranded job, serves it again once reclaimed, and serves a finished one regardless; polling heartbeats only past the interval and reclaims past the timeout.ruff check,ruff format --check,ty checkclean.kill -9'd while running one. The job was left withlocked_byset,finished_at: None,attempts_left: 3and the registration still present, exactly as the issue describes. A replacement worker started, processed the other queued jobs, and at 295 s (the dead heartbeat's age passing the 300 s timeout) released the stranded job, ran it and finished it with a result; the dead registration was gone from the worker list.Note: this touches
get_cached_job_idnext to #160; the two merge cleanly (kept both conditions) and that combination is what the live instance ran for this verification.