Unique recurring task

[Unkn0wnCat]

Hi everyone,

for my project I need to have a task run at regular intervals to go through the database and check which entries are due for refresh. I don't want every worker instance to run this, and would like the task to be assigned like any other "normal" task.

For this I need to create a task that may only exist once, but I am stuck on how to facilitate that, as there are N frontend servers and N workers and no single one is supposed to be an authority.

Is this possible? How can I create a task exactly once?

Thanks in advance,
Kevin

[hibiken]

@Unkn0wnCat Thank you for opening a discussion!

Have you taken a look at this wiki page on periodic-tasks (https://github.com/hibiken/asynq/wiki/Periodic-Tasks)?

Update:
TLDR; If you have multiple Schedulers running to create periodic tasks but want to avoid enqueueing duplicate tasks, try using TaskID and Retention options.

// This ensures there's only one task with ID "foo" in the default queue, and the task will be kept in the queue for x duration after the task has been successfully processed.
client.Enqueue(task, asynq.TaskID("foo"), asynq.Retention(x))

Also, check issue #369 to see if there's alternative solution available (e.g. We may decide to add another variant of Unique option to keep the uniqueness lock until TTL expires, regardless of the task being processed or not).

[Unkn0wnCat]

Ah that makes sense, but weirdly enough it still won't work...

I've changed the enqueue line to the following:

	taskInfo, err := job.client.Enqueue(task, asynq.Unique(1*time.Minute), asynq.Retention(1*time.Minute))

Both tasks show up in asynqmon:

Instances:

I've tried passing a dummy payload as well instead of nil but the result was the same:

[hibiken]

Sorry, I was wrong :( Current implementation releases the lock when task gets processed successfully (regardless of whether retention option is provided or not).

Give this confusion and the issue #369, maybe we should revisit the unique option semantics. We could let the lock expire only after the given TTL expires. Let me reopen the issue and think about this for a bit.

In the meantime, is there a way to handle duplicate tasks within the handler (i.e. detect duplicate task and ignore).

[hibiken]

Following up:
Is there a way to use TaskID option instead of Unique? Suggested workaround mentioned in #369 uses the TaskID option.

client.Enqueue(task, asynq.TaskID("foo"), asynq.Retention(1 * time.Minute))

[Unkn0wnCat]

When using TaskID and checking for the asynq.ErrTaskIDConflict error it works perfectly! Thank you so much for having taken the time to help me with this, and keep up the good work with this amazing package :)

Edit: Can you add this solution to your original response, then I can mark it as the answer and others can use it too?

[hibiken]

Updated the first comment. Thanks for providing all the infos to look into this issue together 🙏