Job cancellation

[colingavin]

I'm investigating using this package in a project that would require the ability to cancel queued jobs. In particular, I would like to be able to:

• Cancel a specific job by its ID whether it is queued or already picked
• Have a mechanism for the consumer of a job to detect its cancellation and short circuit ongoing work
• See that a job was cancelled in the statistics table

Is this a feature that would be in scope to support? If so, I'd be interested in getting your approval on a design and supplying a PR.

[janbjorge]

@colingavin Moved to the ideas - discussion section.

Its a feature that has been requested a few times, and im open to all the help i can get. But my vision for pgqueuer is to keep this library slim and easy to maintain, but yet useful. I think my main concern is that i think its a bad idea to just invoke the .cancel() method on running tasks to be canceled as it feels like the risk of getting into a bad state is high.

The simplest way i can think of is to introduce a 'checkpoint object (co)', the checkpoint object will be know if the a current job id should be canceled. Leaving up to the specific entrypoint function to know at what points its 'safe' to be canceled.

[colingavin]

Great - thanks for the quick reply. I agree on your point about the risk of directly cancelling tasks, especially since existing API users may not have written their entrypoints to expect cancellation.

I would propose the following as general guidelines regarding cancellation behavior:

• When a queued (but not picked) job is cancelled, it is immediately moved to the statistics table and cannot be picked
• Cancellation of a picked job is best-effort
  • Consumers will attempt to be notified of cancellation, but it is OK if they miss this notification and finish work on the job
  • Entrypoints will continue work unless they cooperate in cancellation (as you suggested)
  • Cancelled picked jobs will not be retried
• The introduction of cancellation should not cause a significant decrease in performance benchmarks

Does that sound reasonable to you?

Open questions:

• What happens to picked jobs that are cancelled if the consumer crashes or is partitioned? Normally these would be retried. Who is responsible for cleaning them up & moving to the statistics table.
  • Potentially, consumers could dequeue cancelled jobs as part of their usual work loop and clean them up
• How should cancellation be communicated to consumers. I don't think the current NOTIFY payload contains the content of updates to the queue table. It would be inefficient for the consumer to check for cancellation of all active jobs any time it receives a notification.
  • Could add a separate change notification function specifically for cancellations.
  • Or could add necessary information to the existing notification payload.

[janbjorge]

I've moved over to number references, hopefully making thins a bit easier to split up as there are few ongoing topics.

Guidelines cancellation behavior

1. ✅ Might need a new type: 'canceled' or something.
2. 1. ✅ I think we need to add a new class context argument to the entrypoint functions, that we can use as signaling for cancellation. Consider if contextvars can be used (Unsure how at this point)
3. 2. ✅
4. 3. ✅
5. ✅

To me only open question (for now) is how do we add an extra argument to the entrypoint functions without breaking them.

Open questions

1. If its cancelled, and the consumer crashes (the entrypoint funciton has not yet failed), it should be logged as an cancellation. If however it function crashes, and then the consumer i think state can be a bit tricky to recover 100% correct. (Need to think / analyze a bit more here).
2. How many cancellations do we think we would get? Would it be fair to assume that the vast majority of jobs wont bed canceled? If so, then i think we can add a new event.type (canceled) that we emit at the same time as we delete the job from the queue table. The QM will pick up these events and dispatch them accordingly (set a canceled flag or similar).

[colingavin]

To me only open question (for now) is how do we add an extra argument to the entrypoint functions without breaking them.

The entrypoints already get a Job argument which contains the JobId. That could just be used to look up the cancellation information / checkpoint object on the QM without having any other arguments or a contextvar.

How many cancellations do we think we would get? Would it be fair to assume that the vast majority of jobs wont bed canceled? If so, then i think we can add a new event.type that we emit at the same time as we delete the job from the queue table.

For my use case, yes, most jobs will not be cancelled. Cancellation will be the exceptional case which would be triggered by user actions. In my particular case, a user would be starting a higher level task which may queue many individual jobs with PgQueuer. The user could cancel that task, and thus all the jobs that were queued, but most of these would be waiting in the queue and only a small number would be picked / in-flight. I think adding a new event type that is emitted on cancellation only of picked jobs makes sense.

Given this, how about roughly this design for the new APIs:

1. In Queries

async def cancel_jobs(self, jobs: list[models.JobId]) -> None

Immediately moves jobs that are in the queued state to the statistics table with status cancelled. Jobs that are in the picked state are changed to the new cancelled state but left in the queue table. For those jobs, a new notification is sent with event type jobs_cancelled and the list of IDs.

2. In QueueManager

def cancel_scope_for_job(self, job_id: models.JobId) -> anyio.CancelScope

Allows an entrypoint to fetch a CancelScope for its job ID. It can use this in one of two ways:

• Enter the cancel scope as a context manager to allow automatic cancellation at await points. This would look like:

@qm.entrypoint("task")
async def task(job: Job):
  with qm.cancel_scope_for_job(job.id):
    try:
      await do_some_work() # may raise `asyncio.CancelledError`
    finally:
      cleanup() # can clean up from cancellation here

• Check the cancel scope's cancel_called property at safe checkpoints to handle cancellation explicitly. This would look like:

@qm.entrypoint("task")
def task(job: Job): # explicit checkpoints can also be used in non-async entrypoints
  scope = qm.cancel_scope_for_job(job.id)
  for x in subtasks:
    if scope.cancel_called:
      return
    process(x)

[janbjorge]

Excellent! This works for me, i need to read up on https://anyio.readthedocs.io/en/stable/api.html#anyio.CancelScope but from a quick review it seems fit for purpose.

[janbjorge]

First draft: #91

[janbjorge]

Resolved by: #91