Repository navigation
How to safely re-enqueue an aborted job? #485
Description
I need to perform something similar to:
await Job("id", pool).abort()
await pool.enqueue_job("func", _job_id="id")
However, sometimes this can lead to the abort of the freshly enqueued job. From my observations, this always happens if the aborted job is not found. Waiting until a key in the abort queue expires (1 minute, not configurable) seems to help. My questions are then:
- Is waiting for
arq.constants.abort_job_max_ageguaranteed to be enough for the freshly enqueued job to not be aborted? - Is it possible to not abort a job if it's not found? Simply checking the
job.status()is a race condition.
Activity
sashkent3 commentedon Nov 19, 2024
I believe I've found the culprit of my problem. Calling
await Job("id", pool).abort()isn't ever safe. If the job isn't found, the specifiedjob_idwill remain in theabort_jobs_ssuntil a job with the specified id is run. At that point, the job will be "aborted before start".The comment here is incorrect. Items in the
abort_jobs_ssolder thanabort_job_max_ageare not deleted. The line here seems rather confusing. The worker removes items from theabort_jobs_sswhich areabort_job_max_age(60 ms) in the future. Such items can only appear if theJob.abortwas called immediately after the above line but before the pipeline is executed.If the only intended purpose here was to deliver on the
abort_job_max_agecomment's promise, the line should probably look like this:I'm willing to submit a PR if the issue is confirmed and the proposed resolution is accepted by the maintainers.
Also, it seems like the value of the
abort_job_max_ageconstant was specified in milliseconds by mistake, and the 60-second max-age was intended.drizzt commentedon Nov 23, 2024
hi, did you see if your line fixes the problem?
sashkent3 commentedon Nov 23, 2024
I did very limited testing so take it for what it's worth. But yes, the fix seems to be working for me.
SteniMariyaThomas commentedon Dec 26, 2024
@sashkent3 The enqueued job always be in queue until it taken for execution. Then how will it be not found while aborting.
Can you explain the scenario in which the job is not found
sashkent3 commentedon Dec 26, 2024
@SteniMariyaThomas well, nothing prevents the user from running
await Job("id", pool).abort()while there's no job with the id"id". For example, this could be a first command on a fresh (empty) queue.