email: stack_email_notification has no distributed lock - duplicate emails sent on concurrent Celery Beat runs

[harsh4vardhan]

Bug Description

stack_email_notification reads unprocessed EmailNotificationLog records, dispatches email sub-tasks, then marks them as processed. There is no distributed lock around this task. On multi-pod deployments where two Celery Beat instances run, or when the task takes longer than its 5-minute schedule, two workers process the same unprocessed records and each dispatches a full set of email sub-tasks - resulting in duplicate emails to all subscribers.

Affected file

�pps/api/plane/bgtasks/email_notification_task.py:

`python
email_notifications = EmailNotificationLog.objects.filter(processed_at__isnull=True)... # line 49 - read

...

send_email_notification.delay(...) # lines 76-83 - dispatch sub-tasks (before marking processed)

...

EmailNotificationLog.objects.filter(pk__in=processed_notifications).update(
processed_at=timezone.now()
) # line 84 - marked processed AFTER dispatch
`

The send_email_notification sub-task uses a Redis lock (lines 34-43), but releases it after sending. If the second batch of sub-tasks runs after the 300-second lock TTL expires, they acquire the lock and send again.

Failure scenario

1. Celery Beat schedules stack_email_notification at T=0 and T=5min.
2. At T=0, worker 1 reads 50 unprocessed records and starts dispatching sub-tasks.
3. At T=5min, worker 2 reads the same 50 records (still processed_at IS NULL) and dispatches another 50 sub-tasks.
4. Both batches complete. Every subscriber receives two copies of each email notification.

Also reproducible with a single Beat instance if the task takes longer than 5 minutes (large notification backlog).

Fix

Mark records as processed (within a SELECT FOR UPDATE lock) before dispatching sub-tasks:

`python
with transaction.atomic():
notifications = EmailNotificationLog.objects.select_for_update(skip_locked=True).filter(
processed_at__isnull=True, ...
)
ids = list(notifications.values_list("pk", flat=True))
EmailNotificationLog.objects.filter(pk__in=ids).update(processed_at=timezone.now())

Now dispatch sub-tasks - records are already marked so a second worker skips them

for notification in EmailNotificationLog.objects.filter(pk__in=ids):
send_email_notification.delay(...)
`

skip_locked=True ensures a second concurrent worker skips already-locked rows rather than waiting.

Environment

Plane develop branch (2026-08-13).