Question/Feature request: Is there a way to ensure that a task is only enqueued once?

[philippfreyer]

I am using steady-queue for background task management for update signals. These signals can be paused, if a remote system is currently not reachable. If it becomes reachable again, the update signal may be rescheduled. This usually results in a lot of signals being rescheduled, though (I need to refactor that still).
However, this also means that not all tasks are processed until my next retry check happens, resulting in my queue growing out of proportion, since tasks are then enqueued again.

Is there a way to prevent that with steady-queue?
In other words: Since concurrency is already relying on a key to ensure that a version of a task is only executed once in parallel, would it be possible to use the same key to prevent enqueueing a task twice into the same queue, if it is already in the queue?

[knifecake]

Hello! Hmmm that's an interesting use case!

For now, I want to maintain feature-parity with what Solid Queue provides so I don't think this will get added to the package. Also, I suspect in most cases, making the job idempotent and exiting early could do the trick. Nevertheless, I'm open to creating a contrib subpackage that could include this and an Email sending backend (I was going to wait for Django but it's gonna be at least a year before they include it).

For your particular use case, I understand that even that approach is too much overhead, that's why you want to prevent enqueueing altogether. If you're already using concurrency limited tasks, then you can query the Job table before enqueueing and skip enqueueing if a task with your concurrency key already exists.

I think something like this should work:

from steady_queue.models import Job

def enqueue_unique(task, *args, **kwargs):
    key = task.concurrency_key
    if callable(key):
        key = key(*args, **kwargs)

    if Job.objects.filter(concurrency_key=key, finished_at__isnull=True).exists():
        return None

    return task.enqueue(*args, **kwargs)

enqueue_unique(your_task, *args, **kwargs)

☝️ this solution is prone to race conditions but if you combine it with idempotent jobs, you could still use it and save the load of enqueueing most of them at all.

Does this help at all?

[philippfreyer]

Thank you for the snippet, that helps a lot, actually! Race conditions are less of an issue for my specific implementation, since my issue is based on a periodically scheduled job.

I fully understand the wish for feature-parity. Until you may include this functionality in a contrib subpackage (no pressure, though), I will add your snippet into my local utility collection, alongside a comment to change it out with your implementation, in case you decide to include one.