From 7cd29962808d691d466fafcacbaf57ce9d9a62c7 Mon Sep 17 00:00:00 2001 From: Harsh Vardhan Date: Thu, 13 Aug 2026 19:02:38 +0000 Subject: [PATCH] fix(email): use SELECT FOR UPDATE SKIP LOCKED to prevent duplicate email dispatch stack_email_notification reads unprocessed EmailNotificationLog records, dispatches sub-tasks, then marks records as processed. In multi-pod deployments or when execution time exceeds the 5-minute Celery Beat schedule, two workers simultaneously read the same unprocessed rows, both dispatch sub-tasks, and both mark records processed. Every subscriber then receives duplicate emails. Fix: wrap the SELECT and the processed_at UPDATE inside a single atomic transaction and use select_for_update(skip_locked=True). A concurrent worker skips already-locked rows. Rows are marked processed inside the transaction before dispatch so no row is dispatched twice. Fixes makeplane/plane#9602 Signed-off-by: harsh4vardhan --- .../plane/bgtasks/email_notification_task.py | 26 +++++++++++++------ 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/apps/api/plane/bgtasks/email_notification_task.py b/apps/api/plane/bgtasks/email_notification_task.py index 5cf1d52af91..536d36edcf2 100644 --- a/apps/api/plane/bgtasks/email_notification_task.py +++ b/apps/api/plane/bgtasks/email_notification_task.py @@ -45,15 +45,30 @@ def release_lock(lock_id): @shared_task def stack_email_notification(): - # get all email notifications - email_notifications = EmailNotificationLog.objects.filter(processed_at__isnull=True).order_by("receiver").values() + # Claim unprocessed records inside an atomic transaction using SELECT FOR UPDATE SKIP LOCKED. + # This prevents concurrent Celery workers (e.g., overlapping Beat runs on multi-pod deployments) + # from reading the same rows and dispatching duplicate emails. + from django.db import transaction as db_transaction + with db_transaction.atomic(): + email_notifications = list( + EmailNotificationLog.objects.filter(processed_at__isnull=True) + .select_for_update(skip_locked=True) + .order_by("receiver") + .values() + ) + + if not email_notifications: + return + + # Mark the claimed rows as processed immediately so other workers skip them. + claimed_ids = [n.get("id") for n in email_notifications] + EmailNotificationLog.objects.filter(pk__in=claimed_ids).update(processed_at=timezone.now()) # Create the below format for each of the issues # {"issue_id" : { "actor_id1": [ { data }, { data } ], "actor_id2": [ { data }, { data } ] }} # Convert to unique receivers list receivers = list(set([str(notification.get("receiver_id")) for notification in email_notifications])) - processed_notifications = [] # Loop through all the issues to create the emails for receiver_id in receivers: # Notification triggered for the receiver @@ -67,8 +82,6 @@ def stack_email_notification(): payload.setdefault(receiver_notification.get("entity_identifier"), {}).setdefault( str(receiver_notification.get("triggered_by_id")), [] ).append(receiver_notification.get("data")) - # append processed notifications - processed_notifications.append(receiver_notification.get("id")) email_notification_ids.append(receiver_notification.get("id")) # Create emails for all the issues @@ -80,9 +93,6 @@ def stack_email_notification(): email_notification_ids=email_notification_ids, ) - # Update the email notification log - EmailNotificationLog.objects.filter(pk__in=processed_notifications).update(processed_at=timezone.now()) - def create_payload(notification_data): # return format {"actor_id": { "key": { "old_value": [], "new_value": [] } }}