Suggested `retry_stalled_jobs` implementation fails on jobs with queueing lock

[jakajancar]

If another job with same queuing_lock is already queued, the suggested retry_stalled_jobs will fail:

[Log] Traceback (most recent call last):
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/sync_psycopg_connector.py", line 29, in wrap_exceptions
    yield
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/psycopg_connector.py", line 40, in wrap_exceptions
    yield
  File "/root/.local/share/uv/python/cpython-3.11.13-linux-aarch64-gnu/lib/python3.11/contextlib.py", line 97, in inner
    return await func(*args, **kwds)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/psycopg_connector.py", line 207, in execute_query_async
    await cursor.execute(query, self._wrap_json(arguments))
  File "/code/.venv/lib/python3.11/site-packages/psycopg/cursor_async.py", line 97, in execute
    raise ex.with_traceback(None)
psycopg.errors.UniqueViolation: duplicate key value violates unique constraint "procrastinate_jobs_queueing_lock_idx_v1"
DETAIL:  Key (queueing_lock)=(shop/rfq/27782868-bc7d-48c9-8dfa-40cbd59d894d) already exists.
CONTEXT:  SQL statement "UPDATE procrastinate_jobs
        SET status = 'todo'::procrastinate_job_status,
            attempts = attempts + 1,
            scheduled_at = retry_at,
            priority = COALESCE(new_priority, priority),
            queue_name = COALESCE(new_queue_name, queue_name),
            lock = COALESCE(new_lock, lock)
        WHERE id = job_id AND status = 'doing'
        RETURNING id"
PL/pgSQL function procrastinate_retry_job_v1(bigint,timestamp with time zone,integer,character varying,character varying) line 16 at SQL statement

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/worker.py", line 266, in _process_job
    job_result.result = await ensure_async()
                        ^^^^^^^^^^^^^^^^^^^^
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/worker.py", line 252, in ensure_async
    task_result = await await_func(*job_args, **job.task_kwargs)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/code/uptool/tasks.py", line 13, in retry_stalled_jobs
    await app.job_manager.retry_job(job)
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/manager.py", line 470, in retry_job
    await self.retry_job_by_id_async(
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/manager.py", line 506, in retry_job_by_id_async
    await self.connector.execute_query_async(
  File "/root/.local/share/uv/python/cpython-3.11.13-linux-aarch64-gnu/lib/python3.11/contextlib.py", line 96, in inner
    async with self._recreate_cm():
  File "/root/.local/share/uv/python/cpython-3.11.13-linux-aarch64-gnu/lib/python3.11/contextlib.py", line 231, in __aexit__
    await self.gen.athrow(typ, value, traceback)
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/psycopg_connector.py", line 39, in wrap_exceptions
    with sync_psycopg_connector.wrap_exceptions():
  File "/root/.local/share/uv/python/cpython-3.11.13-linux-aarch64-gnu/lib/python3.11/contextlib.py", line 158, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/code/.venv/lib/python3.11/site-packages/procrastinate/sync_psycopg_connector.py", line 40, in wrap_exceptions
    raise exceptions.UniqueViolation(
procrastinate.exceptions.UniqueViolation: 
    A unique constraint is violated. The constraint name is available in
    ``exception.constraint_name``.

No problem to catch UniqueViolation, but a user should really not have to worry about this stuff, the whole retrying stalled jobs should just be done automatically :(

[jakajancar]

This is our current workaround:

# This job should not set `lock`. If it does, a stalled `retry_stalled_jobs` will prevent all un-stalling.
# `retry_job` seems to be atomic, so it should be OK if two `retry_stalled_jobs` jobs run at the same time.
@app.periodic(cron="* * * * *", queueing_lock="retry_stalled_jobs")
@app.task()
async def retry_stalled_jobs(timestamp: int):
    stalled_jobs = await app.job_manager.get_stalled_jobs()
    for job in stalled_jobs:
        try:
            await app.job_manager.retry_job(job)
        except UniqueViolation as e:
            if e.queueing_lock:
                # Another job for same queueing_lock already enqueued, ignore error and fail the job.
                # Currently deletes it, assuming worker has delete_jobs="always".
                await app.job_manager.finish_job(job, Status.FAILED, True)
            else:
                raise

But I believe it's actually problematic, because if we have:

• Job 1 - todo - 0 attempts
• Job 2 - doing (stalled) - 3 attempts

On execution this will delete the stalled job, meaning the next one will be attempted immediately.

• Job 1 - doing - 0 attempts

And if a new one gets enqueued in between, the pattern will continue forever. Meaning a job that is crashing a worker will completely hog a worker instead of being throttled by the retry policy.

Not sure what the right solution here is.

[ewjoachim]

I think that tasks that have a queuing lock generally don't need retrying. My suggestion would be that retry_stalled_jobs only retries job that don't have a queuing lock? Do you see a use case where it's relevant that a task with a queuing lock is retried upon stalling?

[jakajancar]

Yeah, we have a use-case:

Some update happens (e.g. via an API) that requires a slow async update (e.g. image generation or push to 3rd party system), so it schedules a refresh task. If said updates happen multiple times, there is no point in enqueueing multiple refreshes, they should be deduplicated, but it must happen at least once/be retried.

(in this case the fact that in Procrastinate there can be 1 doing (lock) and 1 queued task (queueing lock) concurrently works perfectly: the refresh already in progress would likely not pick up the latest data... this setup ensures the entire refresh "happens after" the update that triggered it)

[ewjoachim]

Make sense.

I don't have a clear idea what the right path should be but if anyone wants to weigh in, feel free :)