Repository navigation
When RqueueMessageEnqueuer::enqueueUnique() is going to be properly implemented? #259
Description
Activity
@Tsar thanks for flagging this! We can now implement enqueue deduplication since we have a message metadata store in place. The issue only occurs during the enqueue flow, and we can add a check before sending the message to the queue. It should be a small change, I’ll take a look soon. If you’re up for it, feel free to send a PR. The update would primarily involve adding a duplicate check in RqueueMessageEnqueuerImpl.
Reacted by Ioann VolkovCan you test this with
3.4.0Hi @sonus21,
Thanks!
I tried the new version. Now it looks like there's another problem:enqueueUnique()returnsfalsewhen I try to enqueue the task with the same id as completed task.Here's how my test looks like:
- I have a task which sleeps for 100ms.
- I call
enqueueUnique()for it twice withmessageId = "unique-task". - First attempt should return
true, second should returnfalse. This works now. - Then the test waits until the task is completed and tries to call
enqueueUnique()again. It returnsfalse. I would expecttrue.
That’s the expected behavior,
enqueueUniqueis designed to prevent enqueuing tasks that have already been completed. Your expectation doesn’t align with how uniqueness is enforced.If you need different behavior, you’ll have to implement a custom workaround. One known approach is to perform your own uniqueness check using
rqueueMessageMetadataService.getByMessageId("queue", "messageId")and inspect the message status accordingly. You can also use Redis based locking to prevent parallel enqueues.In such scenarios, if you intend to enqueue the same message ID again, you should ensure that the corresponding MessageMetadata and any related state (such as job state) are properly deleted beforehand.
So do you actually store unique completed tasks forever?
All tasks are stored in Redis, technically it can be stored forever but it has default ttl of 30 minutes (post execution).
Is this a configurable setting?
It looks like if I set 0 (meaning completed tasks are immediately removed from Redis), thenenqueueUnique()would work like I need it to.yes, you can set that to zero or -1 it will delete immediately.
rqueue.message.durability.in-terminal-state=0 rqueue.job.durability.in-terminal-state=0Reacted by Ioann Volkov
Is your feature request related to a problem? Please describe.
I tried replacing db-scheduler with Rqueue in my project, because it fills the queue with a lot of tasks (a few million), db-scheduler has bottleneck on DB table single-row operations performance.
It looked really promising, I practically finished the implementation. But as soon as I started testing, I realized, that
RqueueMessageEnqueuer::enqueueUnique()is not properly implemented. There are some TODOs in the code of the library.I can't have same item processed in parallel, that's why I need to guarantee uniqueness of the item in the queue.
Describe the solution you'd like
I'd like
RqueueMessageEnqueuer::enqueueUnique()to returnfalseand not append anything to the queue in case item with same id is currently processed or pending or scheduled.Describe alternatives you've considered
I've seen message deduplication guide.
But it looks like it's for solving another problem: