How to prevent Payload Queue jobs from running multiple times?

[MurzNN]

I have jobs that send notifications on doc update to multiple users (actually, PUSH, but let's be email, as in the example) on the afterChange hook, here is the code:

payload.jobs.queue({ task: 'sendEmail', input: { to: 'user1@example.com' } }) 
payload.jobs.queue({ task: 'sendEmail', input: { to: 'user2@example.com' } }) 
...
payload.jobs.queue({ task: 'sendEmail', input: { to: 'userN@example.com' } }) 
payload.jobs.run()

It works great, but sometimes users receive the same notifications multiple times! This seem happens when several documents updated simultaneously, and the run() from the second afterChange hook also takes into work tasks from the first afterChange hook, as they are not finished yet.

So, seems Payload jobs runner doesn't lock the jobs, taken from the queue, right?

And the question is how to prevent this issue and enable locking of jobs, to prevent multiple execution of them?

[MurzNN]

There is a enableConcurrencyControl flag that looks suitable to resolve this issue, but actually - not. This flag is intended to not process two jobs with the same key simultaneously, but my issue is that the SAME job is executed twice (and more), not different jobs to the same email (that is okay, actually).

[MurzNN]

A working workaround using this option could be returning the job ID as the key, but it isn't available in the concurency function arguments:

    concurrency?: ConcurrencyConfig<TTaskSlugOrInputOutput extends keyof TypedJobs['tasks'] ? TypedJobs['tasks'][TTaskSlugOrInputOutput]['input'] : TTaskSlugOrInputOutput extends TaskInputOutput ? TTaskSlugOrInputOutput['input'] : object>;

only input values available there.

[MurzNN]

Tried to resolve this issue with the concurrency function, but the issue is still here :(

[MurzNN]

Seems found a relevant bug report, related to this issue: #16043

[barry166]

I would not call payload.jobs.run() from every afterChange hook. Let the hook only queue the notification jobs, then have one runner drain that queue: the payload jobs:run --cron ... bin process, autoRun, or the /payload-jobs/run endpoint from a cron.

The linked #16043 is the important bit: current job claiming is not a safe multi-runner claim for the same queue yet, and the source comment says it sets processing: true early only to "reduce the chance" of another worker picking the same job. Until that is fixed, run one worker per queue or shard work into separate queues, and make the notification side effect idempotent with a key like docId:userId:event so a duplicate execution cannot send twice.

enableConcurrencyControl is still useful for "do not run two jobs with the same concurrency key at once", but it is not the same as an atomic dequeue/claim fix for the exact same job.

[MurzNN]

Thanks for the detailed explanation! So, the suggestion is still to trigger them by cron, not instant, right? But this will lead to significant delays. For example, a chat with 100+ users, new message posted and I need to send 100+ PUSH notifications. Sending the new message notifications with around 30 seconds delay is not a good user experience, right? 😉

And this will not save us from parallel runs, if we gather a lot of notifications to send during a single cron run (10_000+) - it can take more than 1 minute to drain. And when the time for the next cron come, we face the same issue again.

For now, I workarounded the issue by just creating a new unique name of the queue for each job - this is not very convenient, but, at least, works ;)

[MurzNN]

I see a PR #17441 that is trying to fix this issue - awesome!

[smoothdvd]

Add key and supersedes: true in concurrency. This will reduce the duplicated chance.

[jayblast-spec]

The real fix here doesn't depend on Payload's internal job claiming at all: treat the notification send itself as the idempotency boundary, not the job dequeue. Add a dedupe table (or a Redis key if you already have Redis) keyed on whatever uniquely identifies "this notification, once" — something like (taskSlug, dedupeKey) where dedupeKey is docId:userId:event.

Before sending, try to insert that key with a unique constraint (Postgres/Mongo unique index) or SETNX in Redis. If the insert/SETNX fails because the key already exists, skip the send — someone already handled it. If it succeeds, you own this notification and proceed.

This sidesteps the queue's internal claiming race entirely, because it doesn't matter how many times the job runs or how many workers pick it up concurrently — the side effect itself can only fire once per key. It's also worth doing even after #17441 lands, since that PR hardens job claiming at the queue level, but any at-least-once queue can still redeliver a job after a worker crashes mid-execution but after the side effect already ran. Idempotency at the point of the actual external effect is the only thing that's airtight against that class of failure — the queue-level fix reduces how often you'll need the fallback, but doesn't remove the need for it.

Concretely for a notification send:

const dedupeKey = `${docId}:${userId}:${event}`
const claimed = await tryClaim(taskSlug, dedupeKey) // unique insert or Redis SETNX with TTL
if (!claimed) return // already sent, no-op
await sendNotification(...)

Set a TTL/expiry on the dedupe record that's comfortably longer than your job's max retry window, so it doesn't grow unbounded but still covers realistic redelivery windows.