From 3de4f84501c001583820482b9735ec5eff5d1efc Mon Sep 17 00:00:00 2001 From: Anton Krytskyi Date: Thu, 6 Aug 2026 20:06:19 +0300 Subject: [PATCH 01/16] developer warning only for high-activity phase --- .../notification_campaigns_detail.html | 2 +- osf/email/notification_campaign.py | 34 +++++++--- osf_tests/test_notification_campaign.py | 64 +++++++++++++++++++ 3 files changed, 89 insertions(+), 11 deletions(-) diff --git a/admin/templates/notifications/notification_campaigns_detail.html b/admin/templates/notifications/notification_campaigns_detail.html index 1c75d1f382a..6e2cec6babc 100644 --- a/admin/templates/notifications/notification_campaigns_detail.html +++ b/admin/templates/notifications/notification_campaigns_detail.html @@ -158,7 +158,7 @@

Progress

{% if notification_campaign.developer_reminder_sent %}
Warning! - The campaign exceeded the expected timeframe ({{ metadata.execution.time_window }}s). A reminder was sent. + The high-activity phase exceeded the expected timeframe ({{ metadata.execution.time_window }}s). A reminder was sent.
{% endif %} {% endif %} diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index a6db5fc9ca0..d9363594691 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -295,6 +295,7 @@ def start_notification_campaign(campaign_id, restart_failed=False, restart_stuck high_activity_tasks = build_campaign_group( min_activity=activity_threshold, spam=False, + developer_reminder=True, **batch_task_kwargs ) if high_activity_tasks: @@ -319,7 +320,14 @@ def start_notification_campaign(campaign_id, restart_failed=False, restart_stuck @celery_app.task(name='email.send_campaign_batch', ignore_result=False) -def send_campaign_batch(context, recipients_ids, notification_type_name='blank', campaign_id=None, run_id=None): +def send_campaign_batch( + context, + recipients_ids, + notification_type_name='blank', + campaign_id=None, + run_id=None, + developer_reminder=False, +): campaign = NotificationCampaign.objects.get(id=campaign_id) if campaign.run_id != run_id: return @@ -342,15 +350,21 @@ def send_campaign_batch(context, recipients_ids, notification_type_name='blank', sentry.log_message(message) return - execution_time_window = campaign.metadata.get('execution', {}).get('time_window', settings.DEFAULT_CAMPAIGN_WINDOW_TIME) - if campaign.started_at < timezone.now() - timedelta(seconds=execution_time_window): - if not campaign.developer_reminder_sent: - message = (f'[Notification Campaign #{campaign_id}] WARNING: ' - f'Exceeded execution time window ({execution_time_window}s): name={campaign.name}.') - logger.warning(message) - sentry.log_message(message) - campaign.developer_reminder_sent = True - campaign.save() + if developer_reminder: + execution_time_window = campaign.metadata.get('execution', {}).get('time_window', settings.DEFAULT_CAMPAIGN_WINDOW_TIME) + if campaign.started_at < timezone.now() - timedelta(seconds=execution_time_window): + # Atomic claim so concurrent high-activity batches only alert once + updated = NotificationCampaign.objects.filter( + pk=campaign_id, + developer_reminder_sent=False, + ).update(developer_reminder_sent=True) + if updated: + message = ( + f'[Notification Campaign #{campaign_id}] WARNING: Campaign {campaign.name} exceeded ' + f'its high-activity execution time window ({execution_time_window} seconds).' + ) + logger.warning(message) + sentry.log_message(message) recipients_qs = NotificationCampaignRecipient.objects.filter(id__in=recipients_ids).select_related('user') recipient_records = [] diff --git a/osf_tests/test_notification_campaign.py b/osf_tests/test_notification_campaign.py index a54e1f16fed..a26604afe43 100644 --- a/osf_tests/test_notification_campaign.py +++ b/osf_tests/test_notification_campaign.py @@ -619,6 +619,70 @@ def test_send_campaign_batch_logs_when_time_window_exceeded(self, mock_sentry, r notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, + developer_reminder=True, + ) + + running_campaign.refresh_from_db() + assert running_campaign.developer_reminder_sent is True + mock_sentry.assert_called_once() + + @mock.patch('osf.email.notification_campaign.sentry.log_message') + def test_send_campaign_batch_skips_reminder_without_developer_reminder_flag( + self, mock_sentry, running_campaign + ): + user = UserFactory() + running_campaign.started_at = timezone.now() - timedelta(seconds=9) + running_campaign.metadata['execution']['time_window'] = 8 + running_campaign.save() + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) + + with mock.patch.object(NotificationType, 'emit'): + send_campaign_batch( + context={}, + recipients_ids=[recipient.id], + notification_type_name='blank', + campaign_id=running_campaign.id, + run_id=running_campaign.run_id, + ) + + running_campaign.refresh_from_db() + assert running_campaign.developer_reminder_sent is False + mock_sentry.assert_not_called() + + @mock.patch('osf.email.notification_campaign.sentry.log_message') + def test_send_campaign_batch_reminder_claimed_only_once(self, mock_sentry, running_campaign): + user_a = UserFactory() + user_b = UserFactory() + running_campaign.started_at = timezone.now() - timedelta(seconds=9) + running_campaign.metadata['execution']['time_window'] = 8 + running_campaign.save() + create_campaign_recipients( + Q(**{'id__in': [user_a.id, user_b.id]}), + campaign_id=running_campaign.id, + ) + recipients = list( + NotificationCampaignRecipient.objects.filter(campaign=running_campaign).values_list( + 'id', flat=True + ) + ) + + with mock.patch.object(NotificationType, 'emit'): + send_campaign_batch( + context={}, + recipients_ids=[recipients[0]], + notification_type_name='blank', + campaign_id=running_campaign.id, + run_id=running_campaign.run_id, + developer_reminder=True, + ) + send_campaign_batch( + context={}, + recipients_ids=[recipients[1]], + notification_type_name='blank', + campaign_id=running_campaign.id, + run_id=running_campaign.run_id, + developer_reminder=True, ) running_campaign.refresh_from_db() From f2cafd10df062d9ee27f2e87fcb2ab73b6515bd4 Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Tue, 25 Aug 2026 23:38:04 +0300 Subject: [PATCH 02/16] [ENG-11968] Fix 'updated_at' field not being updated (#11867) * Add updated_at field to campaign save operations in process_campaign_retry and send_campaign_batch --- osf/email/notification_campaign.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index d9363594691..fc62cc85810 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -210,7 +210,7 @@ def process_campaign_retry(*args, **kwargs): logger.info(message) sentry.log_message(message) campaign.retries += 1 - campaign.save(update_fields=['retries']) + campaign.save(update_fields=['retries', 'updated_at']) retry_group = build_campaign_group( batch_size=batch_size, campaign_id=campaign_id, @@ -429,7 +429,7 @@ def send_campaign_batch( stats = get_campaign_recipient_stats(campaign_id) notification_campaign.sent_count = stats['sent_count'] notification_campaign.failed_count = stats['failed_count'] - notification_campaign.save(update_fields=['sent_count', 'failed_count']) + notification_campaign.save(update_fields=['sent_count', 'failed_count', 'updated_at']) batch_finished_at = timezone.now() batch_run_time = (batch_finished_at - batch_started_at).total_seconds() From ec29e9fb5c855bca383f22050532e84a0897b8d4 Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Wed, 26 Aug 2026 16:35:31 +0300 Subject: [PATCH 03/16] [ENG-11874] Create a periodic task to clean up recipients (#11869) * Add task to delete old notification campaign recipients * Implement notification campaign recipients cleanup with configurable settings and add tests --- .../test_campaign_recipient_cleanup.py | 115 ++++++++++++++++++ notifications/tasks.py | 36 +++++- website/settings/defaults.py | 6 + website/settings/local-ci.py | 2 + 4 files changed, 158 insertions(+), 1 deletion(-) create mode 100644 api_tests/notifications/test_campaign_recipient_cleanup.py diff --git a/api_tests/notifications/test_campaign_recipient_cleanup.py b/api_tests/notifications/test_campaign_recipient_cleanup.py new file mode 100644 index 00000000000..963540d0ae4 --- /dev/null +++ b/api_tests/notifications/test_campaign_recipient_cleanup.py @@ -0,0 +1,115 @@ +from datetime import timedelta + +import pytest +from django.utils import timezone + +from website import settings +from osf.models import ( + NotificationCampaign, + NotificationCampaignRecipient, + NotificationTypeEnum, +) +from osf_tests.factories import AuthUserFactory +from notifications.tasks import delete_notification_campaign_recipients + + +@pytest.mark.django_db +class TestDeleteNotificationCampaignRecipients: + + @pytest.fixture(autouse=True) + def setup(self): + self.notification_type = NotificationTypeEnum.BLANK.instance + + def test_deletes_recipients_for_old_campaigns(self): + now = timezone.now() + cutoff = now - settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE + + campaign = NotificationCampaign.objects.create( + name='Old campaign', + notification_type=self.notification_type, + completed_at=cutoff - timedelta(seconds=1), + ) + + recipients = NotificationCampaignRecipient.objects.bulk_create([ + NotificationCampaignRecipient( + campaign=campaign, + user_id=AuthUserFactory().id, + ), + NotificationCampaignRecipient( + campaign=campaign, + user_id=AuthUserFactory().id, + ), + ]) + + delete_notification_campaign_recipients() + + assert not NotificationCampaignRecipient.objects.filter( + id__in=[recipient.id for recipient in recipients], + ).exists() + + def test_does_not_delete_recipients_for_recent_campaigns(self): + now = timezone.now() + cutoff = now - settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE + + campaign = NotificationCampaign.objects.create( + name='Recent campaign', + notification_type=self.notification_type, + completed_at=cutoff + timedelta(seconds=1), + ) + + recipient = NotificationCampaignRecipient.objects.create( + campaign=campaign, + user_id=AuthUserFactory().id, + ) + + delete_notification_campaign_recipients() + + assert NotificationCampaignRecipient.objects.filter( + id=recipient.id, + ).exists() + + def test_does_not_delete_recipients_for_incomplete_campaigns(self): + campaign = NotificationCampaign.objects.create( + name='Incomplete campaign', + notification_type=self.notification_type, + completed_at=None, + ) + + recipient = NotificationCampaignRecipient.objects.create( + campaign=campaign, + user_id=AuthUserFactory().id, + ) + + delete_notification_campaign_recipients() + + assert NotificationCampaignRecipient.objects.filter( + id=recipient.id, + ).exists() + + def test_deletes_recipients_in_batches(self): + now = timezone.now() + cutoff = now - settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE + + campaign = NotificationCampaign.objects.create( + name='Large campaign', + notification_type=self.notification_type, + completed_at=cutoff - timedelta(seconds=1), + ) + + NotificationCampaignRecipient.objects.bulk_create([ + NotificationCampaignRecipient( + campaign=campaign, + user_id=AuthUserFactory().id, + ) + for _ in range(10) + ]) + + assert NotificationCampaignRecipient.objects.filter( + campaign=campaign, + ).count() == 10 + + delete_notification_campaign_recipients() + + assert not NotificationCampaignRecipient.objects.filter( + campaign=campaign, + ).exists() diff --git a/notifications/tasks.py b/notifications/tasks.py index 9775a44fc7a..3f58770ef79 100644 --- a/notifications/tasks.py +++ b/notifications/tasks.py @@ -11,7 +11,7 @@ from framework.postcommit_tasks.handlers import run_postcommit from osf.models import OSFUser, Notification, NotificationTypeEnum, EmailTask, RegistrationProvider, \ - CollectionProvider, AbstractProvider + CollectionProvider, AbstractProvider, NotificationCampaign, NotificationCampaignRecipient from framework.sentry import log_message from osf.registrations.utils import get_registration_provider_submissions_url from osf.utils.permissions import ADMIN @@ -544,3 +544,37 @@ def delete_batch( logger.info(f'Deleted {deleted} rows from {model_name}') delete_batch.delay(app_label, model_name, filters, order_field, batch_size) + + +@celery_app.task( + bind=True, + name='notifications.tasks.delete_notification_campaign_recipients', +) +def delete_notification_campaign_recipients(self): + """Delete recipients for old notification campaigns.""" + + cutoff = timezone.now() - settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE + + campaigns = NotificationCampaign.objects.filter( + completed_at__lt=cutoff, + ).values_list('id', flat=True) + + for campaign_id in campaigns.iterator(): + total_deleted = 0 + while True: + recipient_ids = list( + NotificationCampaignRecipient.objects + .filter(campaign_id=campaign_id) + .values_list('id', flat=True)[:settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_BATCH_SIZE] + ) + + if not recipient_ids: + break + + deleted, _ = NotificationCampaignRecipient.objects.filter( + id__in=recipient_ids, + ).delete() + + total_deleted += deleted + + logger.info(f'Deleted {total_deleted} recipients for campaign {campaign_id}') diff --git a/website/settings/defaults.py b/website/settings/defaults.py index 7305c08b1bd..4030fe5940f 100644 --- a/website/settings/defaults.py +++ b/website/settings/defaults.py @@ -190,6 +190,8 @@ def parent_dir(path): NO_LOGIN_OSF4M_WAIT_TIME = timedelta(weeks=52) # 1 year for "We miss you at OSF" email to users created from OSF4M NOTIFICATIONS_CLEANUP_AGE = timedelta(weeks=12) # 3 months to clean up old notifications and email tasks NOTIFICATIONS_CLEANUP_BATCH_SIZE = 10000 # Batch size for notifications and email tasks cleanup +NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE = timedelta(weeks=12) # 3 months to clean up old notification campaign recipients +NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_BATCH_SIZE = 5000 # Batch size for notification campaign recipients cleanup # Notification campaign execution defaults (overridable per campaign in admin metadata) DEFAULT_CAMPAIGN_ACTIVITY_THRESHOLD = 3 # Users at/above this activity total are scheduled in the high-activity phase @@ -712,6 +714,10 @@ class CeleryConfig: 'schedule': crontab(minute=0, hour=7), # Daily 2 a.m 'kwargs': {'dry_run': False}, }, + 'delete_notification_campaign_recipients': { + 'task': 'notifications.tasks.delete_notification_campaign_recipients', + 'schedule': crontab(minute=0, hour=3, day_of_month=1), + }, 'clear_expired_sessions': { 'task': 'osf.management.commands.clear_expired_sessions', 'schedule': crontab(minute=0, hour=5), # Daily 12 a.m diff --git a/website/settings/local-ci.py b/website/settings/local-ci.py index eec0e7070e2..1c3636405fb 100644 --- a/website/settings/local-ci.py +++ b/website/settings/local-ci.py @@ -67,6 +67,8 @@ class CeleryConfig(defaults.CeleryConfig): NO_ADDON_WAIT_TIME = timedelta(weeks=8) NO_LOGIN_WAIT_TIME = timedelta(weeks=4) NO_LOGIN_OSF4M_WAIT_TIME = timedelta(weeks=6) +NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE = timedelta(weeks=12) # 3 months to clean up old notification campaign recipients +NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_BATCH_SIZE = 10 # Batch size for notification campaign recipients cleanup # Configuration for "We miss you at OSF" email (`NotificationTypeEnum.USER_NO_LOGIN`) MAX_DAILY_NO_LOGIN_EMAILS = None From e0fe6c1e853e7ddf1fd01a7d98a38d3d20bc44f4 Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Wed, 26 Aug 2026 16:40:04 +0300 Subject: [PATCH 04/16] [ENG-11873] Delete test campaigns (#11877) * Add delete functionality for notification campaigns with permissions check --- admin/notifications/urls.py | 1 + admin/notifications/views.py | 16 ++++++++++ .../notification_campaigns_detail.html | 20 ++++++++++++ .../notification_campaigns_list.html | 10 ++++++ admin_tests/notifications/test_campaigns.py | 31 +++++++++++++++++++ osf/migrations/__init__.py | 1 + 6 files changed, 79 insertions(+) diff --git a/admin/notifications/urls.py b/admin/notifications/urls.py index c9c03343fd8..213bd369256 100644 --- a/admin/notifications/urls.py +++ b/admin/notifications/urls.py @@ -17,4 +17,5 @@ re_path(r'notification_campaigns_recipients_preview/$', views.NotificationCampaignsRecipientsPreview.as_view(), name='notification_campaigns_recipients_preview'), re_path(r'notification_campaigns_recipients_list/$', views.NotificationCampaignsRecipientsView.as_view(), name='notification_campaigns_recipients_list'), re_path(r'notification_campaigns_start/(?P\d+)/$', views.StartNotificationCampaign.as_view(), name='notification_campaigns_start'), + re_path(r'notification_campaigns_delete/(?P\d+)/$', views.DeleteNotificationCampaign.as_view(), name='notification_campaigns_delete'), ] diff --git a/admin/notifications/views.py b/admin/notifications/views.py index fc27b02417b..b47896d61eb 100644 --- a/admin/notifications/views.py +++ b/admin/notifications/views.py @@ -431,6 +431,7 @@ def get_context_data(self, *args, **kwargs): if k not in {'filters', 'context', 'execution', 'template'} }, 'allow_restart_stuck': True if timezone.now() - notification_campaign.updated_at > timedelta(minutes=15) else False, + 'delete_allowed': notification_campaign.status != NotificationCampaignStatus.RUNNING } if notification_campaign.status != NotificationCampaignStatus.CREATED: @@ -748,3 +749,18 @@ def post(self, request, *args, **kwargs): 'notifications:notification_campaigns_detail', pk=notification_campaign.pk, ) + +class DeleteNotificationCampaign(PermissionRequiredMixin, View): + permission_required = 'osf.delete_notificationcampaign' + + def post(self, request, *args, **kwargs): + notification_campaign = get_object_or_404( + NotificationCampaign, + pk=kwargs['pk'], + ) + if notification_campaign.status == NotificationCampaignStatus.RUNNING: + messages.error(request, 'Cannot delete a running campaign.') + return redirect('notifications:notification_campaigns_detail', pk=notification_campaign.pk) + notification_campaign.delete() + messages.success(request, f'Notification campaign {notification_campaign.name} deleted successfully.') + return redirect('notifications:notification_campaigns_list') diff --git a/admin/templates/notifications/notification_campaigns_detail.html b/admin/templates/notifications/notification_campaigns_detail.html index 6e2cec6babc..b2f661d69da 100644 --- a/admin/templates/notifications/notification_campaigns_detail.html +++ b/admin/templates/notifications/notification_campaigns_detail.html @@ -197,6 +197,18 @@

Progress

Cancel Campaign +
+ {% csrf_token %} + +
@@ -451,5 +463,13 @@

Additional Metadata

}); } + const deleteForm = document.getElementById("delete-campaign-form"); + + if (deleteForm) { + deleteForm.addEventListener("submit", function (e) { + confirmCampaign(this, e); + }); + } + {% endblock %} diff --git a/admin/templates/notifications/notification_campaigns_list.html b/admin/templates/notifications/notification_campaigns_list.html index 16b609cefac..486fa2d5814 100644 --- a/admin/templates/notifications/notification_campaigns_list.html +++ b/admin/templates/notifications/notification_campaigns_list.html @@ -5,6 +5,16 @@ List of Notification Campaigns {% endblock title %} {% block content %} +
+ {% if messages %} +
    + {% for message in messages %} + {{ message }} + + {% endfor %} +
+ {% endif %} +

List of Notification Campaigns

{% if active_campaign %}

Active Campaign

diff --git a/admin_tests/notifications/test_campaigns.py b/admin_tests/notifications/test_campaigns.py index a213975163c..218dcad95ec 100644 --- a/admin_tests/notifications/test_campaigns.py +++ b/admin_tests/notifications/test_campaigns.py @@ -16,6 +16,7 @@ NotificationCampaignDetail, NotificationCampaignsList, StartNotificationCampaign, + DeleteNotificationCampaign, ) from admin_tests.utilities import setup_form_view from osf.models import NotificationType @@ -419,3 +420,33 @@ def test_start_rejects_when_another_campaign_is_running(self): mock_start.assert_not_called() self.campaign.refresh_from_db() assert self.campaign.status == NotificationCampaignStatus.CREATED + +class TestNotificationCampaignDeleteView(AdminTestCase): + + def setUp(self): + super().setUp() + self.user = AuthUserFactory() + self.notification_type, _ = NotificationType.objects.get_or_create( + name='blank', + defaults={'subject': 'Test', 'template': 'Hello'}, + ) + self.campaign = NotificationCampaign.objects.create( + name='Campaign', + notification_type=self.notification_type, + metadata={'filters': {}, 'context': {}, 'execution': {}}, + ) + + def test_delete_requires_change_notificationcampaign_permission(self): + request = RequestFactory().post( + reverse('notifications:notification_campaigns_delete', kwargs={'pk': self.campaign.pk}) + ) + request.user = self.user + patch_messages(request) + + with self.assertRaises(PermissionDenied): + DeleteNotificationCampaign.as_view()(request, pk=self.campaign.pk) + + grant_permission(self.user, 'delete_notificationcampaign') + response = DeleteNotificationCampaign.as_view()(request, pk=self.campaign.pk) + assert response.status_code == 302 + assert not NotificationCampaign.objects.filter(pk=self.campaign.pk).exists() diff --git a/osf/migrations/__init__.py b/osf/migrations/__init__.py index 344cf381f5a..487082d1169 100644 --- a/osf/migrations/__init__.py +++ b/osf/migrations/__init__.py @@ -119,6 +119,7 @@ def get_admin_write_permissions(): 'change_emailtask', 'delete_emailtask', 'change_notificationcampaign', + 'delete_notificationcampaign', ]) From 97ce6dbdd8eb2ce29ecc3505a116602af8e1a0a6 Mon Sep 17 00:00:00 2001 From: antkryt Date: Mon, 31 Aug 2026 16:52:29 +0300 Subject: [PATCH 05/16] [ENG-11997] Exclude Unconfirmed Accounts from Email Campaigns (#11878) * Add exclude unconfirmed users option to notification campaign --- admin/notifications/views.py | 7 +- .../notification_campaigns_detail.html | 18 ++-- .../notification_campaing_create.html | 42 ++++++++- osf/email/notification_campaign.py | 24 +++-- osf_tests/test_notification_campaign.py | 93 +++++++++++++++++++ 5 files changed, 158 insertions(+), 26 deletions(-) diff --git a/admin/notifications/views.py b/admin/notifications/views.py index b47896d61eb..d7b208afd02 100644 --- a/admin/notifications/views.py +++ b/admin/notifications/views.py @@ -19,7 +19,7 @@ from mako.parsetree import ControlLine from string import Formatter from osf.email import _render_email_html -from osf.email.notification_campaign import FILTER_PRESETS, counter_subquery, build_query +from osf.email.notification_campaign import FILTER_PRESETS, counter_subquery, build_campaign_filter_query from website import settings from urllib.parse import urlencode @@ -642,10 +642,7 @@ def get_queryset(self): raw_filters = self.request.GET.get('filters', None) if raw_filters: json_filters = json.loads(raw_filters) - if predefined := json_filters.get('predefined'): - query = Q(**FILTER_PRESETS.get(predefined, {})) - else: - query = build_query(json_filters.get('manual')) + query = build_campaign_filter_query(json_filters) qs = OSFUser.objects.filter(query) qs = qs.annotate( diff --git a/admin/templates/notifications/notification_campaigns_detail.html b/admin/templates/notifications/notification_campaigns_detail.html index b2f661d69da..abec27feb24 100644 --- a/admin/templates/notifications/notification_campaigns_detail.html +++ b/admin/templates/notifications/notification_campaigns_detail.html @@ -279,25 +279,19 @@

General

Recipient Filters

- {% if not "predefined" in metadata.filters %} - - {% if not "predefined" in metadata.filters %} - {% if metadata.filters.manual %} - {% include "notifications/campaign_filter_group.html" with group=metadata.filters.manual is_root=True %} - {% else %} -

No filters configured.

- {% endif %} - {% endif %} - - {% elif "predefined" in metadata.filters %} - + {% if "predefined" in metadata.filters %}
Predefined Filter {{ metadata.filters.predefined }}
+ {% endif %} + {% if metadata.filters.manual %} + {% include "notifications/campaign_filter_group.html" with group=metadata.filters.manual is_root=True %} + {% elif not "predefined" in metadata.filters %} +

No filters configured.

{% endif %} Recipient Filters
+
+ +

+ When enabled, users who have not confirmed their accounts + are excluded from the recipient list. +

+
+ Execution } function buildFilters() { + let filters; if (filterMode.value === "predefined") { - return { + filters = { "predefined": document.getElementById("filter-row").value }; + } else { + const root = builder.querySelector(":scope > .group"); + filters = {"manual": serializeGroup(root)}; + } + + if (document.getElementById("exclude-unconfirmed").checked) { + const confirmed = { + field: "date_confirmed", + lookup: "isnull", + value: false, + }; + if (filters.manual) { + filters.manual = { + operator: "AND", + children: [filters.manual, confirmed], + }; + } else { + filters.manual = { + operator: "AND", + children: [confirmed], + }; + } } - const root = builder.querySelector(":scope > .group"); - return {"manual": serializeGroup(root)} + return filters; } function updateFiltersInput() { diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index fc62cc85810..c96b33558da 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -3,7 +3,7 @@ from osf.models import NotificationType, NotificationTypeEnum, OSFUser, UserActivityCounter, Email from osf.models.spam import SpamStatus from django.db import transaction -from django.db.models import OuterRef, Subquery, Case, When, CharField, Count, Q +from django.db.models import OuterRef, Subquery, Case, When, CharField, Count, Q, BooleanField from django.db.models.functions import Coalesce from framework.celery_tasks import app as celery_app from celery import group, chain @@ -54,6 +54,9 @@ def build_query(node): if lookup == 'in': value = [v.strip() for v in value.split(',')] + if lookup == 'isnull': + value = BooleanField().to_python(value) + if lookup in negated_lookups: return ~Q(**{ f'{node["field"]}__{negated_lookups[lookup]}': value @@ -79,6 +82,18 @@ def build_query(node): return query + +def build_campaign_filter_query(filters): + """AND together optional predefined and manual filter clauses.""" + filters = filters or {} + query = Q() + if predefined := filters.get('predefined'): + query &= Q(**FILTER_PRESETS.get(predefined, {})) + if manual := filters.get('manual'): + query &= build_query(manual) + return query + + def create_campaign_recipients(filters, campaign_id): qs = ( OSFUser.objects @@ -260,14 +275,11 @@ def start_notification_campaign(campaign_id, restart_failed=False, restart_stuck if hasattr(NotificationTypeEnum, notification_type_name): del getattr(NotificationTypeEnum, notification_type_name).instance - if predefined_filter_name := filters.get('predefined'): - filters = Q(**FILTER_PRESETS.get(predefined_filter_name, {})) - else: - filters = build_query(filters.get('manual', [])) + recipient_filters = build_campaign_filter_query(filters) if not restart_failed and not restart_stuck: recipients_creation_started_at = timezone.now() - create_campaign_recipients(filters=filters, campaign_id=campaign_id) + create_campaign_recipients(filters=recipient_filters, campaign_id=campaign_id) campaign.recipient_count = NotificationCampaignRecipient.objects.filter(campaign_id=campaign_id).count() campaign.save() recipients_creation_finished_at = timezone.now() diff --git a/osf_tests/test_notification_campaign.py b/osf_tests/test_notification_campaign.py index a26604afe43..890d9522a3a 100644 --- a/osf_tests/test_notification_campaign.py +++ b/osf_tests/test_notification_campaign.py @@ -135,6 +135,61 @@ def test_or_combines_usernames(self): assert plain_user.id in user_ids assert other_email.id not in user_ids + def test_isnull_false_excludes_unconfirmed_accounts(self): + confirmed = UserFactory() + unconfirmed = UserFactory(date_confirmed=None, is_registered=False) + + query = build_query({ + 'operator': 'AND', + 'children': [ + { + 'field': 'id', + 'lookup': 'in', + 'value': f'{confirmed.id},{unconfirmed.id}', + }, + { + 'field': 'date_confirmed', + 'lookup': 'isnull', + 'value': False, + }, + ], + }) + user_ids = set(OSFUser.objects.filter(query).values_list('id', flat=True)) + + assert user_ids == {confirmed.id} + + def test_build_campaign_filter_query_ands_predefined_and_manual(self): + from osf.email.notification_campaign import build_campaign_filter_query + + confirmed = UserFactory() + unconfirmed = UserFactory(date_confirmed=None, is_registered=False) + inactive = UserFactory() + inactive.is_disabled = True + inactive.save() + assert inactive.is_active is False + + query = build_campaign_filter_query({ + 'predefined': 'active', + 'manual': { + 'operator': 'AND', + 'children': [ + { + 'field': 'id', + 'lookup': 'in', + 'value': f'{confirmed.id},{unconfirmed.id},{inactive.id}', + }, + { + 'field': 'date_confirmed', + 'lookup': 'isnull', + 'value': False, + }, + ], + }, + }) + user_ids = set(OSFUser.objects.filter(query).values_list('id', flat=True)) + + assert user_ids == {confirmed.id} + class TestCreateCampaignRecipients: @@ -442,6 +497,44 @@ def test_start_creates_recipients_and_schedules_workflow(self, mock_chain, campa mock_chain.assert_called_once() mock_chain.return_value.apply_async.assert_called_once() + @mock.patch('osf.email.notification_campaign.chain') + def test_start_excludes_unconfirmed_accounts_when_enabled(self, mock_chain, campaign): + confirmed = UserFactory() + unconfirmed = UserFactory(date_confirmed=None, is_registered=False) + _set_activity(confirmed, 100) + _set_activity(unconfirmed, 100) + + campaign.metadata['filters'] = { + 'predefined': 'active', + 'manual': { + 'operator': 'AND', + 'children': [ + { + 'field': 'id', + 'lookup': 'in', + 'value': f'{confirmed.id},{unconfirmed.id}', + }, + { + 'field': 'date_confirmed', + 'lookup': 'isnull', + 'value': False, + }, + ], + }, + } + campaign.run_id = uuid.uuid4() + campaign.save() + mock_chain.return_value.apply_async = mock.Mock() + + start_notification_campaign(campaign.id) + + recipient_ids = set( + NotificationCampaignRecipient.objects.filter(campaign=campaign).values_list( + 'user_id', flat=True + ) + ) + assert recipient_ids == {confirmed.id} + @mock.patch('osf.email.notification_campaign.chain') def test_start_restart_failed_does_not_recreate_recipients(self, mock_chain, campaign): user = UserFactory() From 00395170fc7fb99fdb26b44a44c180fefe8024f3 Mon Sep 17 00:00:00 2001 From: antkryt Date: Mon, 31 Aug 2026 17:06:56 +0300 Subject: [PATCH 06/16] [ENG-12046] Fix sendgrid bulk personalization (#11882) * Fix batch personalizations --- osf/email/__init__.py | 40 ++++++++++++++++++------- osf/email/notification_campaign.py | 7 ++++- osf_tests/test_notification_campaign.py | 24 +++++++++++++++ 3 files changed, 60 insertions(+), 11 deletions(-) diff --git a/osf/email/__init__.py b/osf/email/__init__.py index 1cf39af809b..4193d9e5212 100644 --- a/osf/email/__init__.py +++ b/osf/email/__init__.py @@ -238,7 +238,33 @@ def send_email_over_smtp(to_email, notification_type, context, email_context): email.attach(attachment_name, attachment_content) email.send() -def send_email_with_send_grid(to_addr, notification_type, context, email_context=None): +def _build_sendgrid_personalizations(to_list, email_context=None, is_multiple=False): + """Build SendGrid personalizations for one shared message or one message per recipient. + + When ``is_multiple`` is True, each address gets its own personalization (separate + delivery; recipients do not see each other). CC/BCC are omitted in that mode to + avoid duplicating copies per recipient. + """ + email_context = email_context or {} + + if is_multiple: + # If we attached the same CC/BCC to every personalization in is_multiple mode, + # a batch of N recipients would produce N separate emails each including that CC + # so the CC address would get N copies of the same message + # TODO: decide if it's safe to add the CC/BCC to each personalization + return [{'to': [{'email': addr}]} for addr in to_list] + + personalization = {'to': [{'email': addr} for addr in to_list]} + cc_addr = email_context.get('cc_addr') + if cc_addr: + personalization['cc'] = [{'email': a} for a in ([cc_addr] if isinstance(cc_addr, str) else cc_addr)] + bcc_addr = email_context.get('bcc_addr') + if bcc_addr: + personalization['bcc'] = [{'email': a} for a in ([bcc_addr] if isinstance(bcc_addr, str) else bcc_addr)] + return [personalization] + + +def send_email_with_send_grid(to_addr, notification_type, context, email_context=None, *, is_multiple=False): email_context = email_context or {} to_list = [to_addr] if isinstance(to_addr, str) else [a for a in (to_addr or []) if a] @@ -256,18 +282,12 @@ def send_email_with_send_grid(to_addr, notification_type, context, email_context subject_tpl = getattr(notification_type, 'subject', None) subject = subject_tpl.format(**context) if subject_tpl else f'Notification: {getattr(notification_type, "name", "OSF")}' - personalization = {'to': [{'email': addr} for addr in to_list]} - cc_addr = email_context.get('cc_addr') - if cc_addr: - personalization['cc'] = [{'email': a} for a in ([cc_addr] if isinstance(cc_addr, str) else cc_addr)] - bcc_addr = email_context.get('bcc_addr') - if bcc_addr: - personalization['bcc'] = [{'email': a} for a in ([bcc_addr] if isinstance(bcc_addr, str) else bcc_addr)] - payload = { 'from': {'email': from_email}, 'subject': subject, - 'personalizations': [personalization], + 'personalizations': _build_sendgrid_personalizations( + to_list, email_context=email_context, is_multiple=is_multiple + ), 'content': [ {'type': 'text/html', 'value': html}, ], diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index c96b33558da..fa662f21a7d 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -395,7 +395,12 @@ def send_campaign_batch( # NOTE: sendgrid bulk send feature has not been fully implemented and tested recipient_emails = list(valid_emails_qs.values_list('recipient_address', flat=True)) try: - send_email_with_send_grid(to_addr=recipient_emails, notification_type=notification_type, context=context) + send_email_with_send_grid( + to_addr=recipient_emails, + notification_type=notification_type, + context=context, + is_multiple=True, + ) valid_emails_qs.update(status=NotificationCampaignRecipientStatus.SENT, error_message=None) except Exception as exc: message = (f'[Notification Campaign #{campaign_id}] ERROR: ' diff --git a/osf_tests/test_notification_campaign.py b/osf_tests/test_notification_campaign.py index 890d9522a3a..822e01f14d6 100644 --- a/osf_tests/test_notification_campaign.py +++ b/osf_tests/test_notification_campaign.py @@ -6,6 +6,7 @@ from django.utils import timezone from django.db.models import Q +from osf.email import _build_sendgrid_personalizations from osf.email.notification_campaign import ( create_campaign_recipients, get_campaign_recipient_batches, @@ -666,10 +667,33 @@ def test_send_campaign_batch_sendgrid_bulk_success(self, mock_sendgrid, running_ recipient.refresh_from_db() running_campaign.refresh_from_db() mock_sendgrid.assert_called_once() + assert mock_sendgrid.call_args.kwargs.get('is_multiple') is True assert recipient.status == NotificationCampaignRecipientStatus.SENT assert running_campaign.sent_count == 1 assert running_campaign.failed_count == 0 + def test_build_sendgrid_personalizations_shared_to_list(self): + personalizations = _build_sendgrid_personalizations( + ['a@example.com', 'b@example.com'], + is_multiple=False, + ) + assert personalizations == [{ + 'to': [ + {'email': 'a@example.com'}, + {'email': 'b@example.com'}, + ], + }] + + def test_build_sendgrid_personalizations_one_per_recipient(self): + personalizations = _build_sendgrid_personalizations( + ['a@example.com', 'b@example.com'], + is_multiple=True, + ) + assert personalizations == [ + {'to': [{'email': 'a@example.com'}]}, + {'to': [{'email': 'b@example.com'}]}, + ] + @mock.patch( 'osf.email.notification_campaign.send_email_with_send_grid', side_effect=Exception('bulk failed'), From 03f47c983adfcf7da30685e43f51ca7bed91212e Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Tue, 1 Sep 2026 17:01:47 +0300 Subject: [PATCH 07/16] [ENG-11995] Add a bounded dispatcher for Celery tasks (#11886) * Add batch_id field to NotificationCampaignRecipient and update status choices; add campaign dispatcher * Reorder status choices in NotificationCampaignRecipient model * Update recipient status to QUEUED in batch processing and adjust tests accordingly * Refactor assign_batch_id_to_recipients to use min_activity, max_activity, and spam filters; update dispatch_campaign to handle new priority group structure * Fix unit tests --- osf/email/notification_campaign.py | 253 ++++++++++-------- ...tioncampaignrecipient_batch_id_and_more.py | 23 ++ osf/models/notification_campaign.py | 2 + osf_tests/test_notification_campaign.py | 124 +++++---- website/settings/defaults.py | 3 + 5 files changed, 236 insertions(+), 169 deletions(-) create mode 100644 osf/migrations/0050_notificationcampaignrecipient_batch_id_and_more.py diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index fa662f21a7d..137206a785b 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -1,12 +1,11 @@ import logging - +import uuid from osf.models import NotificationType, NotificationTypeEnum, OSFUser, UserActivityCounter, Email from osf.models.spam import SpamStatus from django.db import transaction from django.db.models import OuterRef, Subquery, Case, When, CharField, Count, Q, BooleanField from django.db.models.functions import Coalesce from framework.celery_tasks import app as celery_app -from celery import group, chain from django.utils import timezone from datetime import timedelta from osf.models.notification_campaign import NotificationCampaign, NotificationCampaignRecipient, NotificationCampaignStatus, NotificationCampaignRecipientStatus @@ -119,70 +118,6 @@ def create_campaign_recipients(filters, campaign_id): ) -def get_campaign_recipient_batches( - campaign_id, - batch_size, - restart_failed=False, - min_activity=None, - max_activity=None, - spam=None, -): - qs = NotificationCampaignRecipient.objects.filter( - campaign_id=campaign_id, - ) - - if restart_failed: - qs = qs.filter(status=NotificationCampaignRecipientStatus.FAILED) - else: - qs = qs.filter(status=NotificationCampaignRecipientStatus.PENDING) - - # Minimum and maximum activity are mutually exclusive and use the same threshold. - if min_activity is not None: - qs = qs.filter(activity_score__gte=min_activity) - - if max_activity is not None: - qs = qs.filter(activity_score__lt=max_activity) - - if spam is True: - qs = qs.filter(user__spam_status=SpamStatus.SPAM) - elif spam is False: - qs = qs.exclude(user__spam_status=SpamStatus.SPAM) - - yield from batched( - qs.values_list('id', flat=True).iterator(chunk_size=batch_size), - batch_size, - ) - -def build_campaign_group( - campaign_id, - batch_size, - restart_failed=False, - min_activity=None, - max_activity=None, - spam=None, - **send_kwargs, -): - tasks = [] - - for batch in get_campaign_recipient_batches( - campaign_id=campaign_id, - batch_size=batch_size, - restart_failed=restart_failed, - min_activity=min_activity, - max_activity=max_activity, - spam=spam, - ): - tasks.append( - send_campaign_batch.si( - recipients_ids=batch, - campaign_id=campaign_id, - **send_kwargs, - ) - ) - - return group(tasks) - - def get_campaign_recipient_stats(campaign_id): return NotificationCampaignRecipient.objects.filter( campaign_id=campaign_id @@ -215,7 +150,6 @@ def process_campaign_retry(*args, **kwargs): if campaign.status != NotificationCampaignStatus.CANCELLED: failed_recipients = NotificationCampaignRecipient.objects.filter(campaign=campaign, status=NotificationCampaignRecipientStatus.FAILED) max_retries = campaign.metadata.get('execution', {}).get('max_retries', settings.DEFAULT_CAMPAIGN_MAX_RETRIES) - batch_size = campaign.metadata.get('execution', {}).get('batch_size', settings.DEFAULT_CAMPAIGN_BATCH_SIZE) failed_recipients_count = failed_recipients.count() if failed_recipients_count: if campaign.retries < max_retries: @@ -226,18 +160,13 @@ def process_campaign_retry(*args, **kwargs): sentry.log_message(message) campaign.retries += 1 campaign.save(update_fields=['retries', 'updated_at']) - retry_group = build_campaign_group( - batch_size=batch_size, - campaign_id=campaign_id, - restart_failed=True, - notification_type_name=campaign.notification_type.name, - context=campaign.metadata.get('context', {}), - run_id=campaign.run_id, + + dispatch_campaign.apply_async( + args=[campaign_id, campaign.run_id], + kwargs={ + 'restart_failed': True, + }, ) - chain( - retry_group, - process_campaign_retry.si(campaign_id=campaign_id, run_id=campaign.run_id), - ).apply_async() return final_status = NotificationCampaignStatus.PARTIALLY_COMPLETED @@ -269,7 +198,6 @@ def process_campaign_retry(*args, **kwargs): def start_notification_campaign(campaign_id, restart_failed=False, restart_stuck=False): campaign = NotificationCampaign.objects.get(id=campaign_id) filters = campaign.metadata.get('filters', {}) - context = campaign.metadata.get('context', {}) notification_type_name = campaign.notification_type.name if hasattr(NotificationTypeEnum, notification_type_name): @@ -291,56 +219,156 @@ def start_notification_campaign(campaign_id, restart_failed=False, restart_stuck logger.info(message) sentry.log_message(message) + dispatch_campaign.apply_async( + args=[campaign_id, campaign.run_id], + kwargs={ + 'restart_failed': restart_failed, + 'restart_stuck': restart_stuck, + }, + ) + + +def assign_batch_id_to_recipients( + campaign_id, + batch_size, + restart_failed=False, + min_activity=None, + max_activity=None, + spam=None, +): + batch_id = uuid.uuid4() + filters = Q(campaign_id=campaign_id) + + if restart_failed: + filters &= Q( + status=NotificationCampaignRecipientStatus.FAILED, + ) + else: + filters &= Q( + status=NotificationCampaignRecipientStatus.PENDING, + batch_id__isnull=True, + ) + + if min_activity is not None: + filters &= Q(activity_score__gte=min_activity) + if max_activity is not None: + filters &= Q(activity_score__lt=max_activity) + + if spam is not None: + filters &= Q(user__spam_status=SpamStatus.SPAM) if spam else ~Q(user__spam_status=SpamStatus.SPAM) + + recipient_ids = NotificationCampaignRecipient.objects.filter( + filters + ).values_list('id', flat=True)[:batch_size] + + updated = NotificationCampaignRecipient.objects.filter(id__in=recipient_ids).update(batch_id=batch_id, status=NotificationCampaignRecipientStatus.QUEUED) + + if updated == 0: + return None + + return batch_id + +@celery_app.task(bind=True, name='email.dispatch_campaign') +def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_stuck=False): + campaign = NotificationCampaign.objects.get(id=campaign_id) + if campaign.run_id != run_id: + return + + if campaign.status == NotificationCampaignStatus.CANCELLED: + logger.warning(f"Campaign {campaign_id} was cancelled") + return + if campaign.status != NotificationCampaignStatus.RUNNING: + message = f'[Notification Campaign #{campaign_id}] ERROR: Campaign {campaign.name} is not in RUNNING status.' + logger.error(message) + sentry.log_message(message) + return + + queued_batches_count = NotificationCampaignRecipient.objects.filter( + campaign=campaign, + batch_id__isnull=False, + status=NotificationCampaignRecipientStatus.QUEUED + ).order_by().values('batch_id').distinct().count() + execution = campaign.metadata.get('execution', {}) batch_size = execution.get('batch_size', settings.DEFAULT_CAMPAIGN_BATCH_SIZE) activity_threshold = execution.get('activity_threshold', settings.DEFAULT_CAMPAIGN_ACTIVITY_THRESHOLD) - batch_task_kwargs = dict( - batch_size=batch_size, - campaign_id=campaign_id, - restart_failed=restart_failed, - notification_type_name=notification_type_name, - context=context, - run_id=campaign.run_id + notification_type_name = campaign.notification_type.name + to_queue = settings.MAX_QUEUED_CAMPAIGN_BATCHES - queued_batches_count + + priority_groups = ( + { + 'min_activity': activity_threshold, + 'spam': False, + }, + { + 'max_activity': activity_threshold, + 'spam': False, + }, + { + 'spam': True, + }, ) + total_new_queued_batches = 0 + new_queued_batches = 0 + for priority_group in priority_groups: + no_more_recipients = False + for _ in range(to_queue): + batch_id = assign_batch_id_to_recipients( + campaign_id, + batch_size=batch_size, + restart_failed=restart_failed, + **priority_group, + ) - workflow = [] - high_activity_tasks = build_campaign_group( - min_activity=activity_threshold, - spam=False, - developer_reminder=True, - **batch_task_kwargs - ) - if high_activity_tasks: - workflow.append(high_activity_tasks) + if batch_id is None: + no_more_recipients = True + break - low_activity_tasks = build_campaign_group( - max_activity=activity_threshold, - spam=False, - **batch_task_kwargs - ) - if low_activity_tasks: - workflow.append(low_activity_tasks) + send_campaign_batch.delay( + context=campaign.metadata.get('context', {}), + batch_id=batch_id, + campaign_id=campaign_id, + notification_type_name=notification_type_name, + run_id=run_id, + developer_reminder=True if priority_group == 'high_activity' else False, + ) + new_queued_batches += 1 - spam_users_tasks = build_campaign_group( - spam=True, - **batch_task_kwargs - ) - if spam_users_tasks: - workflow.append(spam_users_tasks) + to_queue -= new_queued_batches + total_new_queued_batches += new_queued_batches + new_queued_batches = 0 + if to_queue <= 0: + break + + logger.info(f'[Notification Campaign #{campaign_id}] INFO: Dispatched {total_new_queued_batches} new batches for campaign {campaign.name}.') - chain(*workflow, process_campaign_retry.si(campaign_id=campaign_id, run_id=campaign.run_id)).apply_async() + if no_more_recipients: + process_campaign_retry.delay( + campaign_id=campaign_id, run_id=campaign.run_id + ) + else: + self.apply_async( + args=[campaign_id, run_id], + kwargs={ + 'restart_failed': restart_failed, + 'restart_stuck': restart_stuck, + }, + countdown=settings.CAMPAIGN_DISPATCH_INTERVAL.total_seconds(), + ) @celery_app.task(name='email.send_campaign_batch', ignore_result=False) def send_campaign_batch( context, - recipients_ids, + batch_id=None, notification_type_name='blank', campaign_id=None, run_id=None, developer_reminder=False, ): campaign = NotificationCampaign.objects.get(id=campaign_id) + recipients_qs = NotificationCampaignRecipient.objects.filter(batch_id=batch_id).select_related('user') + if campaign.run_id != run_id: return if campaign.status == NotificationCampaignStatus.CANCELLED: @@ -357,6 +385,10 @@ def send_campaign_batch( if campaign.status != NotificationCampaignStatus.FAILED: campaign.status = NotificationCampaignStatus.FAILED campaign.save() + recipients_qs.update( + status=NotificationCampaignRecipientStatus.FAILED, + error_message='Notification type not found', + ) message = f'[Notification Campaign #{campaign_id}] ERROR: Batch failed due to none notification_type (template)' logger.error(message) sentry.log_message(message) @@ -378,7 +410,6 @@ def send_campaign_batch( logger.warning(message) sentry.log_message(message) - recipients_qs = NotificationCampaignRecipient.objects.filter(id__in=recipients_ids).select_related('user') recipient_records = [] recipients_qs_annotated = recipients_qs.annotate( recipient_address=Case( diff --git a/osf/migrations/0050_notificationcampaignrecipient_batch_id_and_more.py b/osf/migrations/0050_notificationcampaignrecipient_batch_id_and_more.py new file mode 100644 index 00000000000..8081dedf15f --- /dev/null +++ b/osf/migrations/0050_notificationcampaignrecipient_batch_id_and_more.py @@ -0,0 +1,23 @@ +# Generated by Django 4.2.26 on 2026-08-28 10:01 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('osf', '0049_project_enter'), + ] + + operations = [ + migrations.AddField( + model_name='notificationcampaignrecipient', + name='batch_id', + field=models.UUIDField(blank=True, db_index=True, null=True), + ), + migrations.AlterField( + model_name='notificationcampaignrecipient', + name='status', + field=models.CharField(choices=[('queued', 'Queued'), ('pending', 'Pending'), ('sent', 'Sent'), ('failed', 'Failed'), ('skipped', 'Skipped'), ('postponed', 'Postponed')], db_index=True, default='pending', max_length=20), + ), + ] diff --git a/osf/models/notification_campaign.py b/osf/models/notification_campaign.py index 4b03337f25e..261e445da51 100644 --- a/osf/models/notification_campaign.py +++ b/osf/models/notification_campaign.py @@ -13,6 +13,7 @@ class NotificationCampaignStatus(models.TextChoices): ENDED = 'ended', 'Ended' class NotificationCampaignRecipientStatus(models.TextChoices): + QUEUED = 'queued', 'Queued' PENDING = 'pending', 'Pending' SENT = 'sent', 'Sent' FAILED = 'failed', 'Failed' @@ -120,6 +121,7 @@ class NotificationCampaignRecipient(models.Model): error_message = models.TextField(null=True, blank=True) activity_score = models.IntegerField(default=0) + batch_id = models.UUIDField(null=True, blank=True, db_index=True) class Meta: unique_together = ('campaign', 'user') diff --git a/osf_tests/test_notification_campaign.py b/osf_tests/test_notification_campaign.py index 822e01f14d6..1ec91118c17 100644 --- a/osf_tests/test_notification_campaign.py +++ b/osf_tests/test_notification_campaign.py @@ -8,8 +8,8 @@ from osf.email import _build_sendgrid_personalizations from osf.email.notification_campaign import ( + assign_batch_id_to_recipients, create_campaign_recipients, - get_campaign_recipient_batches, get_campaign_recipient_stats, process_campaign_retry, send_campaign_batch, @@ -59,26 +59,23 @@ def _set_activity(user, total): defaults={'total': total, 'action': {}, 'date': {}}, ) +def reset_batch_ids(campaign_id): + NotificationCampaignRecipient.objects.filter(campaign_id=campaign_id).update(batch_id=None) def _recipient_user_ids(campaign_id, **batch_kwargs): """Flatten all batches into an list of user ids and preserve order""" - user_ids = [] - for batch in get_campaign_recipient_batches(campaign_id=campaign_id, batch_size=1000, **batch_kwargs): - recipients = NotificationCampaignRecipient.objects.filter(id__in=batch) - by_id = {r.id: r.user_id for r in recipients} - user_ids.extend(by_id[recipient_id] for recipient_id in batch) + + batch_id = assign_batch_id_to_recipients(campaign_id=campaign_id, batch_size=1000, **batch_kwargs) + recipients = NotificationCampaignRecipient.objects.filter(batch_id=batch_id) + user_ids = [r.user_id for r in recipients] return user_ids def _recipient_scores(campaign_id, **batch_kwargs): """Flatten all batches into an list of activity scores and preserve order""" - scores = [] - for batch in get_campaign_recipient_batches(campaign_id=campaign_id, batch_size=1000, **batch_kwargs): - by_id = { - r.id: r.activity_score - for r in NotificationCampaignRecipient.objects.filter(id__in=batch) - } - scores.extend(by_id[recipient_id] for recipient_id in batch) + batch_id = assign_batch_id_to_recipients(campaign_id=campaign_id, batch_size=1000, **batch_kwargs) + recipients = NotificationCampaignRecipient.objects.filter(batch_id=batch_id) + scores = [r.activity_score for r in recipients] return scores @@ -253,6 +250,7 @@ def test_ordered_by_activity_score_descending(self, campaign): scores = _recipient_scores(campaign.id) assert scores == sorted(scores, reverse=True) + reset_batch_ids(campaign.id) user_ids = _recipient_user_ids(campaign.id) assert user_ids == [newer.id, mid.id, older.id] @@ -360,6 +358,7 @@ def test_high_activity_non_spam(self, campaign, users_and_recipients): min_activity=data['threshold'], spam=False, ) + assert set(user_ids) == {data['high'].id, data['flagged'].id} def test_low_activity_non_spam_includes_zero(self, campaign, users_and_recipients): @@ -396,8 +395,14 @@ def test_batches_respect_batch_size(self, campaign): Q(**{'id__in': [u.id for u in users]}), campaign_id=campaign.id, ) + batches = [] + while True: + batch_id = assign_batch_id_to_recipients(campaign_id=campaign.id, batch_size=2) + if not batch_id: + break + + batches.append(list(NotificationCampaignRecipient.objects.filter(batch_id=batch_id).values_list('user_id', flat=True))) - batches = list(get_campaign_recipient_batches(campaign_id=campaign.id, batch_size=2)) assert [len(batch) for batch in batches] == [2, 2, 1] flat = {recipient_id for batch in batches for recipient_id in batch} assert len(flat) == 5 @@ -408,10 +413,10 @@ def test_restart_failed_only_returns_failed(self, campaign, users_and_recipients failed.status = NotificationCampaignRecipientStatus.FAILED failed.save(update_fields=['status']) - pending_ids = _recipient_user_ids(campaign.id, spam=False, min_activity=data['threshold']) + pending_ids = _recipient_user_ids(campaign.id, min_activity=data['threshold'], spam=False) assert data['high'].id not in pending_ids - failed_ids = _recipient_user_ids(campaign.id, restart_failed=True) + failed_ids = _recipient_user_ids(campaign.id, restart_failed=True, min_activity=data['threshold'], spam=False) assert failed_ids == [data['high'].id] def test_ignore_conflicts_on_duplicate_create(self, campaign): @@ -468,8 +473,8 @@ def test_campaign_start_restart_stuck_counts(self, mock_delay, campaign): restart_stuck=True, ) - @mock.patch('osf.email.notification_campaign.chain') - def test_start_creates_recipients_and_schedules_workflow(self, mock_chain, campaign): + @mock.patch('osf.email.notification_campaign.dispatch_campaign.apply_async') + def test_start_creates_recipients_and_schedules_workflow(self, mock_dispatch_campaign, campaign): high = UserFactory() low = UserFactory() spam = UserFactory() @@ -484,8 +489,6 @@ def test_start_creates_recipients_and_schedules_workflow(self, mock_chain, campa campaign.run_id = uuid.uuid4() campaign.save() - mock_chain.return_value.apply_async = mock.Mock() - start_notification_campaign(campaign.id) campaign.refresh_from_db() @@ -495,11 +498,10 @@ def test_start_creates_recipients_and_schedules_workflow(self, mock_chain, campa } assert recipients == {high.id: 250, low.id: 10, spam.id: 0} assert campaign.recipient_count == 3 - mock_chain.assert_called_once() - mock_chain.return_value.apply_async.assert_called_once() + mock_dispatch_campaign.assert_called_once() - @mock.patch('osf.email.notification_campaign.chain') - def test_start_excludes_unconfirmed_accounts_when_enabled(self, mock_chain, campaign): + @mock.patch('osf.email.notification_campaign.dispatch_campaign') + def test_start_excludes_unconfirmed_accounts_when_enabled(self, mock_dispatch_campaign, campaign): confirmed = UserFactory() unconfirmed = UserFactory(date_confirmed=None, is_registered=False) _set_activity(confirmed, 100) @@ -525,7 +527,7 @@ def test_start_excludes_unconfirmed_accounts_when_enabled(self, mock_chain, camp } campaign.run_id = uuid.uuid4() campaign.save() - mock_chain.return_value.apply_async = mock.Mock() + mock_dispatch_campaign.return_value.apply_async = mock.Mock() start_notification_campaign(campaign.id) @@ -536,8 +538,8 @@ def test_start_excludes_unconfirmed_accounts_when_enabled(self, mock_chain, camp ) assert recipient_ids == {confirmed.id} - @mock.patch('osf.email.notification_campaign.chain') - def test_start_restart_failed_does_not_recreate_recipients(self, mock_chain, campaign): + @mock.patch('osf.email.notification_campaign.dispatch_campaign') + def test_start_restart_failed_does_not_recreate_recipients(self, mock_dispatch_campaign, campaign): user = UserFactory() _set_activity(user, 50) create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) @@ -550,21 +552,21 @@ def test_start_restart_failed_does_not_recreate_recipients(self, mock_chain, cam 'manual': {'operator': 'AND', 'children': [{'field': 'id', 'lookup': 'in', 'value': str(user.id)}]}, } campaign.save() - mock_chain.return_value.apply_async = mock.Mock() + mock_dispatch_campaign.return_value.apply_async = mock.Mock() start_notification_campaign(campaign.id, restart_failed=True) assert NotificationCampaignRecipient.objects.filter(campaign=campaign).count() == 1 assert NotificationCampaignRecipient.objects.get(pk=recipient.pk).status == NotificationCampaignRecipientStatus.FAILED - @mock.patch('osf.email.notification_campaign.chain') - def test_start_restart_stuck_does_not_recreate_recipients(self, mock_chain, campaign): + @mock.patch('osf.email.notification_campaign.dispatch_campaign') + def test_start_restart_stuck_does_not_recreate_recipients(self, mock_dispatch_campaign, campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) campaign.run_id = uuid.uuid4() campaign.recipient_count = 1 campaign.save() - mock_chain.return_value.apply_async = mock.Mock() + mock_dispatch_campaign.return_value.apply_async = mock.Mock() start_notification_campaign(campaign.id, restart_stuck=True) @@ -587,11 +589,12 @@ def running_campaign(self, campaign): def test_send_campaign_batch_marks_recipients_sent(self, mock_emit, running_campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -608,11 +611,12 @@ def test_send_campaign_batch_marks_recipients_sent(self, mock_emit, running_camp def test_send_campaign_batch_marks_recipients_failed(self, mock_sentry, mock_emit, running_campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -631,11 +635,12 @@ def test_send_campaign_batch_skips_invalid_email_addresses(self, running_campaig user.emails.all().delete() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -654,11 +659,12 @@ def test_send_campaign_batch_sendgrid_bulk_success(self, mock_sendgrid, running_ running_campaign.metadata['sendgrid_bulk'] = True running_campaign.save(update_fields=['metadata']) create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -704,11 +710,12 @@ def test_send_campaign_batch_sendgrid_bulk_failure(self, mock_sentry, mock_sendg running_campaign.metadata['sendgrid_bulk'] = True running_campaign.save(update_fields=['metadata']) create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -727,12 +734,12 @@ def test_send_campaign_batch_logs_when_time_window_exceeded(self, mock_sentry, r running_campaign.metadata['execution']['time_window'] = 8 running_campaign.save() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) - recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) with mock.patch.object(NotificationType, 'emit'): send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -752,12 +759,12 @@ def test_send_campaign_batch_skips_reminder_without_developer_reminder_flag( running_campaign.metadata['execution']['time_window'] = 8 running_campaign.save() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) - recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) with mock.patch.object(NotificationType, 'emit'): send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -778,16 +785,13 @@ def test_send_campaign_batch_reminder_claimed_only_once(self, mock_sentry, runni Q(**{'id__in': [user_a.id, user_b.id]}), campaign_id=running_campaign.id, ) - recipients = list( - NotificationCampaignRecipient.objects.filter(campaign=running_campaign).values_list( - 'id', flat=True - ) - ) + batch_id_1 = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) + batch_id_2 = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) with mock.patch.object(NotificationType, 'emit'): send_campaign_batch( context={}, - recipients_ids=[recipients[0]], + batch_id=batch_id_1, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -795,7 +799,7 @@ def test_send_campaign_batch_reminder_claimed_only_once(self, mock_sentry, runni ) send_campaign_batch( context={}, - recipients_ids=[recipients[1]], + batch_id=batch_id_2, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -809,11 +813,12 @@ def test_send_campaign_batch_reminder_claimed_only_once(self, mock_sentry, runni def test_send_campaign_batch_marks_failed_when_notification_type_missing(self, running_campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='does-not-exist', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -822,41 +827,43 @@ def test_send_campaign_batch_marks_failed_when_notification_type_missing(self, r running_campaign.refresh_from_db() recipient.refresh_from_db() assert running_campaign.status == NotificationCampaignStatus.FAILED - assert recipient.status == NotificationCampaignRecipientStatus.PENDING + assert recipient.status == NotificationCampaignRecipientStatus.FAILED def test_send_campaign_batch_skips_stale_run_id(self, running_campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=uuid.uuid4(), ) recipient.refresh_from_db() - assert recipient.status == NotificationCampaignRecipientStatus.PENDING + assert recipient.status == NotificationCampaignRecipientStatus.QUEUED def test_send_campaign_batch_skips_cancelled_campaign(self, running_campaign): user = UserFactory() running_campaign.status = NotificationCampaignStatus.CANCELLED running_campaign.save(update_fields=['status']) create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, ) recipient.refresh_from_db() - assert recipient.status == NotificationCampaignRecipientStatus.PENDING + assert recipient.status == NotificationCampaignRecipientStatus.QUEUED @mock.patch.object(NotificationType, 'emit') def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, mock_emit, running_campaign): @@ -867,11 +874,12 @@ def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, m user.emails.create(address='fallback@example.com') create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -980,8 +988,8 @@ def test_process_campaign_retry_keeps_cancelled_status_and_syncs_stats(self, moc assert campaign.completed_at is not None mock_sentry.assert_called_once() - @mock.patch('osf.email.notification_campaign.chain') - def test_process_campaign_retry_retries_failed_recipients(self, mock_chain, campaign): + @mock.patch('osf.email.notification_campaign.dispatch_campaign.apply_async') + def test_process_campaign_retry_retries_failed_recipients(self, mock_dispatch_campaign, campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) recipient = NotificationCampaignRecipient.objects.get(campaign=campaign, user=user) @@ -991,14 +999,14 @@ def test_process_campaign_retry_retries_failed_recipients(self, mock_chain, camp campaign.retries = 0 campaign.status = NotificationCampaignStatus.RUNNING campaign.save() - mock_chain.return_value.apply_async = mock.Mock() + mock_dispatch_campaign.return_value.apply_async = mock.Mock() process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) campaign.refresh_from_db() assert campaign.retries == 1 assert campaign.status == NotificationCampaignStatus.RUNNING - mock_chain.assert_called_once() + mock_dispatch_campaign.assert_called_once() def test_process_campaign_retry_marks_partially_completed_after_max_retries(self, campaign): user = UserFactory() diff --git a/website/settings/defaults.py b/website/settings/defaults.py index 4030fe5940f..bd03f08cf4d 100644 --- a/website/settings/defaults.py +++ b/website/settings/defaults.py @@ -198,6 +198,8 @@ def parent_dir(path): DEFAULT_CAMPAIGN_BATCH_SIZE = 1000 DEFAULT_CAMPAIGN_WINDOW_TIME = 28800 # 8 hours DEFAULT_CAMPAIGN_MAX_RETRIES = 3 +MAX_QUEUED_CAMPAIGN_BATCHES = 100 # Maximum number of queued campaign batches allowed before new batches are rejected. This is to prevent runaway campaigns from overwhelming the system. +CAMPAIGN_DISPATCH_INTERVAL = timedelta(minutes=5) # Minimum time between dispatching campaign batches. # The following are rough estimates so we can log to sentry those batches and sendgrid quests which run longer than normal ESTIMATED_PER_REQUEST_THRESHOLD = 0.3 # On production, sending one email via SendGrid takes 0.20 ~ 0.50 seconds, set default alert threshold at 0.30s ESTIMATED_BATCH_RUN_TIME_THRESHOLD = 300 # On production, with batch size 1000, we expect each batch to finish within 300s (5m) @@ -508,6 +510,7 @@ class CeleryConfig: 'scripts.check_manual_restart_approval', 'scripts.enhanced_stuck_registration_audit', 'email.start_notification_campaign', + 'email.dispatch_campaign', } background_migration_modules = { From b6083170cf37a29282bfb086167951cd9a0ca587 Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Wed, 2 Sep 2026 16:57:55 +0300 Subject: [PATCH 08/16] [ENG-12079] Make dispatcher settings configurable through admin app for DevTesting purpose (#11895) * Add max queued batches and dispatch interval fields to NotificationCampaign forms * redo migrations * unit tests * Update campaign dispatch interval to use integer minutes instead of timedelta * Fix CAMPAIGN_DISPATCH_INTERVAL which uses sec instead of min --------- Co-authored-by: Longze Chen --- admin/notifications/forms.py | 12 ++++++++++ admin/notifications/views.py | 2 ++ .../notification_campaing_create.html | 22 +++++++++++++++++++ admin_tests/notifications/test_campaigns.py | 10 +++++++-- osf/email/notification_campaign.py | 5 +++-- ...and_more.py => 0053_post_enter_project.py} | 4 ++-- website/settings/defaults.py | 2 +- 7 files changed, 50 insertions(+), 7 deletions(-) rename osf/migrations/{0050_notificationcampaignrecipient_batch_id_and_more.py => 0053_post_enter_project.py} (86%) diff --git a/admin/notifications/forms.py b/admin/notifications/forms.py index 94b571dbb65..6d4050c0c3f 100644 --- a/admin/notifications/forms.py +++ b/admin/notifications/forms.py @@ -50,6 +50,18 @@ class NotificationCampaignCreateForm(forms.ModelForm): initial=False, ) + max_queued_batches = forms.IntegerField( + min_value=1, + initial=settings.MAX_QUEUED_CAMPAIGN_BATCHES, + help_text='Maximum number of queued campaign batches allowed before new batches are rejected.', + ) + + dispatch_interval = forms.IntegerField( + min_value=1, + initial=settings.CAMPAIGN_DISPATCH_INTERVAL, + help_text='The time in seconds between dispatches of campaign batches.', + ) + class Meta: model = NotificationCampaign fields = ( diff --git a/admin/notifications/views.py b/admin/notifications/views.py index d7b208afd02..d9ac34aafa0 100644 --- a/admin/notifications/views.py +++ b/admin/notifications/views.py @@ -582,6 +582,8 @@ def form_valid(self, form): 'max_retries': form.cleaned_data['max_retries'], 'activity_threshold': form.cleaned_data['activity_threshold'], 'time_window': form.cleaned_data['time_window'], + 'max_queued_batches': form.cleaned_data['max_queued_batches'], + 'dispatch_interval': form.cleaned_data['dispatch_interval'], }, 'sendgrid_bulk': form.cleaned_data.get('sendgrid_bulk', False), } diff --git a/admin/templates/notifications/notification_campaing_create.html b/admin/templates/notifications/notification_campaing_create.html index ce1b832c10d..af66df3786f 100644 --- a/admin/templates/notifications/notification_campaing_create.html +++ b/admin/templates/notifications/notification_campaing_create.html @@ -280,6 +280,28 @@

Execution

> + + Max Queued Batches + + + + + + Dispatch Interval + + + +
diff --git a/admin_tests/notifications/test_campaigns.py b/admin_tests/notifications/test_campaigns.py index 218dcad95ec..771081e5b24 100644 --- a/admin_tests/notifications/test_campaigns.py +++ b/admin_tests/notifications/test_campaigns.py @@ -61,6 +61,8 @@ def _valid_form_data(notification_type, **overrides): 'activity_threshold': settings.DEFAULT_CAMPAIGN_ACTIVITY_THRESHOLD, 'sendgrid_bulk': False, 'time_window': 8 * 60 * 60, + 'max_queued_batches': settings.MAX_QUEUED_CAMPAIGN_BATCHES, + 'dispatch_interval': settings.CAMPAIGN_DISPATCH_INTERVAL, } data.update(overrides) return data @@ -261,7 +263,9 @@ def test_form_valid_persists_execution_metadata(self): max_retries=4, activity_threshold=77, sendgrid_bulk=True, - time_window=28800 + time_window=28800, + max_queued_batches=10, + dispatch_interval=60, ), ) request.user = self.user @@ -283,7 +287,9 @@ def test_form_valid_persists_execution_metadata(self): 'batch_size': 25, 'max_retries': 4, 'activity_threshold': 77, - 'time_window': 28800 + 'time_window': 28800, + 'max_queued_batches': 10, + 'dispatch_interval': 60, } assert campaign.metadata['sendgrid_bulk'] is True assert campaign.metadata['filters'] == {'predefined': 'active'} diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index 137206a785b..58c9cff28ba 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -292,8 +292,9 @@ def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_s execution = campaign.metadata.get('execution', {}) batch_size = execution.get('batch_size', settings.DEFAULT_CAMPAIGN_BATCH_SIZE) activity_threshold = execution.get('activity_threshold', settings.DEFAULT_CAMPAIGN_ACTIVITY_THRESHOLD) + max_queued_batches = execution.get('max_queued_batches', settings.MAX_QUEUED_CAMPAIGN_BATCHES) notification_type_name = campaign.notification_type.name - to_queue = settings.MAX_QUEUED_CAMPAIGN_BATCHES - queued_batches_count + to_queue = max_queued_batches - queued_batches_count priority_groups = ( { @@ -353,7 +354,7 @@ def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_s 'restart_failed': restart_failed, 'restart_stuck': restart_stuck, }, - countdown=settings.CAMPAIGN_DISPATCH_INTERVAL.total_seconds(), + countdown=execution.get('dispatch_interval', settings.CAMPAIGN_DISPATCH_INTERVAL), ) diff --git a/osf/migrations/0050_notificationcampaignrecipient_batch_id_and_more.py b/osf/migrations/0053_post_enter_project.py similarity index 86% rename from osf/migrations/0050_notificationcampaignrecipient_batch_id_and_more.py rename to osf/migrations/0053_post_enter_project.py index 8081dedf15f..521ddc81ee7 100644 --- a/osf/migrations/0050_notificationcampaignrecipient_batch_id_and_more.py +++ b/osf/migrations/0053_post_enter_project.py @@ -1,4 +1,4 @@ -# Generated by Django 4.2.26 on 2026-08-28 10:01 +# Generated by Django 4.2.26 on 2026-09-02 09:16 from django.db import migrations, models @@ -6,7 +6,7 @@ class Migration(migrations.Migration): dependencies = [ - ('osf', '0049_project_enter'), + ('osf', '0052_downloadevent_download_channel'), ] operations = [ diff --git a/website/settings/defaults.py b/website/settings/defaults.py index 3114cacd595..a5a52176a24 100644 --- a/website/settings/defaults.py +++ b/website/settings/defaults.py @@ -198,7 +198,7 @@ def parent_dir(path): DEFAULT_CAMPAIGN_WINDOW_TIME = 28800 # 8 hours DEFAULT_CAMPAIGN_MAX_RETRIES = 3 MAX_QUEUED_CAMPAIGN_BATCHES = 100 # Maximum number of queued campaign batches allowed before new batches are rejected. This is to prevent runaway campaigns from overwhelming the system. -CAMPAIGN_DISPATCH_INTERVAL = timedelta(minutes=5) # Minimum time between dispatching campaign batches. +CAMPAIGN_DISPATCH_INTERVAL = 300 # 5 min (300 sec), minimum time before checking and dispatching new campaign batches. # The following are rough estimates so we can log to sentry those batches and sendgrid quests which run longer than normal ESTIMATED_PER_REQUEST_THRESHOLD = 0.3 # On production, sending one email via SendGrid takes 0.20 ~ 0.50 seconds, set default alert threshold at 0.30s ESTIMATED_BATCH_RUN_TIME_THRESHOLD = 300 # On production, with batch size 1000, we expect each batch to finish within 300s (5m) From a9a6d36d4b7d26278e81f0f7d687f408a8123a24 Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Fri, 4 Sep 2026 18:44:48 +0300 Subject: [PATCH 09/16] [ENG-12090] Make recipients creation a separate step/task (#11898) * Add create recipients functionality for notification campaigns * Fix developer reminder bug * Adjust unit tests --- admin/notifications/urls.py | 1 + admin/notifications/views.py | 27 ++++++++- .../notification_campaigns_detail.html | 13 ++++- osf/email/notification_campaign.py | 57 ++++++++++++------- osf/models/notification_campaign.py | 10 ++++ osf_tests/test_notification_campaign.py | 8 ++- 6 files changed, 91 insertions(+), 25 deletions(-) diff --git a/admin/notifications/urls.py b/admin/notifications/urls.py index 213bd369256..de7139bb491 100644 --- a/admin/notifications/urls.py +++ b/admin/notifications/urls.py @@ -17,5 +17,6 @@ re_path(r'notification_campaigns_recipients_preview/$', views.NotificationCampaignsRecipientsPreview.as_view(), name='notification_campaigns_recipients_preview'), re_path(r'notification_campaigns_recipients_list/$', views.NotificationCampaignsRecipientsView.as_view(), name='notification_campaigns_recipients_list'), re_path(r'notification_campaigns_start/(?P\d+)/$', views.StartNotificationCampaign.as_view(), name='notification_campaigns_start'), + re_path(r'notification_campaigns_create_recipients/(?P\d+)/$', views.CreateNotificationCampaignRecipients.as_view(), name='notification_campaigns_create_recipients'), re_path(r'notification_campaigns_delete/(?P\d+)/$', views.DeleteNotificationCampaign.as_view(), name='notification_campaigns_delete'), ] diff --git a/admin/notifications/views.py b/admin/notifications/views.py index d9ac34aafa0..bef2ceed94d 100644 --- a/admin/notifications/views.py +++ b/admin/notifications/views.py @@ -431,7 +431,8 @@ def get_context_data(self, *args, **kwargs): if k not in {'filters', 'context', 'execution', 'template'} }, 'allow_restart_stuck': True if timezone.now() - notification_campaign.updated_at > timedelta(minutes=15) else False, - 'delete_allowed': notification_campaign.status != NotificationCampaignStatus.RUNNING + 'delete_allowed': notification_campaign.status != NotificationCampaignStatus.RUNNING, + 'start_allowed': notification_campaign.status == NotificationCampaignStatus.CREATED and metadata.get('recipients_creation_finished', False), } if notification_campaign.status != NotificationCampaignStatus.CREATED: @@ -586,6 +587,7 @@ def form_valid(self, form): 'dispatch_interval': form.cleaned_data['dispatch_interval'], }, 'sendgrid_bulk': form.cleaned_data.get('sendgrid_bulk', False), + 'recipients_creation_finished': False, } try: _render_email_html(form.instance.notification_type, form.cleaned_data['context']) @@ -749,6 +751,29 @@ def post(self, request, *args, **kwargs): pk=notification_campaign.pk, ) +class CreateNotificationCampaignRecipients(PermissionRequiredMixin, View): + permission_required = 'osf.change_notificationcampaign' + + def post(self, request, *args, **kwargs): + notification_campaign = get_object_or_404( + NotificationCampaign, + pk=kwargs['pk'], + ) + + if notification_campaign.status != NotificationCampaignStatus.CREATED: + messages.error(request, 'Recipients can only be created for campaigns in CREATED status.') + return redirect( + 'notifications:notification_campaigns_detail', + pk=notification_campaign.pk, + ) + + notification_campaign.create_recipients() + + return redirect( + 'notifications:notification_campaigns_detail', + pk=notification_campaign.pk, + ) + class DeleteNotificationCampaign(PermissionRequiredMixin, View): permission_required = 'osf.delete_notificationcampaign' diff --git a/admin/templates/notifications/notification_campaigns_detail.html b/admin/templates/notifications/notification_campaigns_detail.html index abec27feb24..883b6e3e482 100644 --- a/admin/templates/notifications/notification_campaigns_detail.html +++ b/admin/templates/notifications/notification_campaigns_detail.html @@ -169,10 +169,21 @@

Progress

style="display:inline;" > {% csrf_token %} - +
+ {% csrf_token %} + +
Date: Tue, 8 Sep 2026 16:13:45 -0400 Subject: [PATCH 10/16] Re-do migration after merge release 26.20 --- ...{0053_post_enter_project.py => 0054_post_enter_project.py} | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) rename osf/migrations/{0053_post_enter_project.py => 0054_post_enter_project.py} (86%) diff --git a/osf/migrations/0053_post_enter_project.py b/osf/migrations/0054_post_enter_project.py similarity index 86% rename from osf/migrations/0053_post_enter_project.py rename to osf/migrations/0054_post_enter_project.py index 521ddc81ee7..3c867947385 100644 --- a/osf/migrations/0053_post_enter_project.py +++ b/osf/migrations/0054_post_enter_project.py @@ -1,4 +1,4 @@ -# Generated by Django 4.2.26 on 2026-09-02 09:16 +# Generated by Django 4.2.26 on 2026-09-08 20:13 from django.db import migrations, models @@ -6,7 +6,7 @@ class Migration(migrations.Migration): dependencies = [ - ('osf', '0052_downloadevent_download_channel'), + ('osf', '0053_gdpr_delete_and_orcid_revoke'), ] operations = [ From ffda8a88d54e9dbfdd781050bb8c98acbb6cd718 Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Wed, 9 Sep 2026 17:13:16 +0300 Subject: [PATCH 11/16] [ENG-12138] Bypass notification.emit() and use sendgrid directly to reduce overhead / resource (#11906) * Refactor email sending logic to consolidate SMTP and SendGrid functionality into a single send_email function, improving maintainability and readability. * Fix capture_notifications * Update emit mocks --- osf/email/__init__.py | 36 ++++++++++++++++++++++--- osf/email/notification_campaign.py | 11 +++++--- osf/models/notification.py | 31 +++++---------------- osf_tests/test_notification_campaign.py | 22 +++++++-------- tests/utils.py | 8 +++--- 5 files changed, 61 insertions(+), 47 deletions(-) diff --git a/osf/email/__init__.py b/osf/email/__init__.py index 4193d9e5212..6f399dca55f 100644 --- a/osf/email/__init__.py +++ b/osf/email/__init__.py @@ -199,7 +199,7 @@ def _safe_categories(cats): out.append(c) return out[:10] -def send_email_over_smtp(to_email, notification_type, context, email_context): +def send_email_over_smtp(to_email, notification_type, context, email_context, rendered_html=None): if waffle.switch_is_active(features.ENABLE_MAILHOG): host = settings.MAILHOG_HOST port = settings.MAILHOG_PORT @@ -212,7 +212,7 @@ def send_email_over_smtp(to_email, notification_type, context, email_context): raise NotImplementedError('MAIL_SERVER or MAIL_PORT is not set') subject = None if not notification_type.subject else notification_type.subject.format(**context) - body_html = _render_email_html(notification_type, context) + body_html = rendered_html or _render_email_html(notification_type, context) or '

(no content)

' email = EmailMessage( subject=subject, @@ -264,7 +264,7 @@ def _build_sendgrid_personalizations(to_list, email_context=None, is_multiple=Fa return [personalization] -def send_email_with_send_grid(to_addr, notification_type, context, email_context=None, *, is_multiple=False): +def send_email_with_send_grid(to_addr, notification_type, context, email_context=None, *, is_multiple=False, rendered_html=None): email_context = email_context or {} to_list = [to_addr] if isinstance(to_addr, str) else [a for a in (to_addr or []) if a] @@ -277,7 +277,7 @@ def send_email_with_send_grid(to_addr, notification_type, context, email_context logging.error('SendGrid: missing SENDGRID_FROM_EMAIL/FROM_EMAIL') return False - html = _render_email_html(notification_type, context) or '

(no content)

' + html = rendered_html or _render_email_html(notification_type, context) or '

(no content)

' subject_tpl = getattr(notification_type, 'subject', None) subject = subject_tpl.format(**context) if subject_tpl else f'Notification: {getattr(notification_type, "name", "OSF")}' @@ -358,3 +358,31 @@ def send_email_with_send_grid(to_addr, notification_type, context, email_context else: logging.error('SendGrid hit a blocked socket error: %r | payload=%s', exc, payload) raise + +def send_email(recipient_address, notification_type, event_context=None, email_context=None, rendered_html=None): + """ + Send an email using either SMTP or SendGrid based on settings and feature flags. + """ + if waffle.switch_is_active(features.ENABLE_MAILHOG): + send_email_over_smtp( + recipient_address, + notification_type, + event_context, + email_context, + rendered_html=rendered_html, + ) + + if not settings.LOCAL_MODE: + send_email_with_send_grid( + recipient_address, + notification_type, + event_context, + email_context, + rendered_html=rendered_html, + ) + + if settings.LOCAL_MODE and not waffle.switch_is_active(features.ENABLE_MAILHOG): + logging.warning( + 'Both ENABLE_MAILHOG and LOCAL_MODE are disabled. Emails will not be sent to MailHog or real email addresses. ' + 'Turn on ENABLE_MAILHOG to send emails to MailHog for testing, or turn on LOCAL_MODE to send emails with SendGrid.' + ) diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index 7c4bfba6e9b..70826926435 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -9,7 +9,7 @@ from django.utils import timezone from datetime import timedelta from osf.models.notification_campaign import NotificationCampaign, NotificationCampaignRecipient, NotificationCampaignStatus, NotificationCampaignRecipientStatus -from osf.email import send_email_with_send_grid +from osf.email import send_email_with_send_grid, _render_email_html, send_email from framework import sentry from website import settings from itertools import batched @@ -454,13 +454,16 @@ def send_campaign_batch( sentry.log_message(message) valid_emails_qs.update(status=NotificationCampaignRecipientStatus.FAILED, error_message=str(exc)) else: + rendered_html = _render_email_html(notification_type, context) for recipient in valid_emails_qs: notification_started_at = timezone.now() try: - notification_type.emit( - user=recipient.user, + send_email( + recipient_address=recipient.recipient_address, + notification_type=notification_type, event_context=context, - save=False, # Too many write operations + email_context=context, + rendered_html=rendered_html ) recipient.status = NotificationCampaignRecipientStatus.SENT recipient.error_message = None diff --git a/osf/models/notification.py b/osf/models/notification.py index d3071442af6..3f9a66b7daa 100644 --- a/osf/models/notification.py +++ b/osf/models/notification.py @@ -1,12 +1,10 @@ import logging -import waffle from django.db import models from django.utils import timezone from api.base import settings as api_settings -from website import settings as osf_settings -from osf import email, features +from osf import email class Notification(models.Model): @@ -38,27 +36,12 @@ def send( f"\nemail_context={email_context}" ) - if waffle.switch_is_active(features.ENABLE_MAILHOG): - email.send_email_over_smtp( - recipient_address, - self.subscription.notification_type, - self.event_context, - email_context - ) - - if not osf_settings.LOCAL_MODE: - email.send_email_with_send_grid( - recipient_address, - self.subscription.notification_type, - self.event_context, - email_context - ) - - if osf_settings.LOCAL_MODE and not waffle.switch_is_active(features.ENABLE_MAILHOG): - logging.warning( - 'Both ENABLE_MAILHOG and LOCAL_MODE are disabled. Emails will not be sent to MailHog or real email addresses. ' - 'Turn on ENABLE_MAILHOG to send emails to MailHog for testing, or turn on LOCAL_MODE to send emails with SendGrid.' - ) + email.send_email( + recipient_address=recipient_address, + notification_type=self.subscription.notification_type, + event_context=self.event_context, + email_context=email_context + ) if save: self.mark_sent() diff --git a/osf_tests/test_notification_campaign.py b/osf_tests/test_notification_campaign.py index dd936403193..8f6f4a4cc52 100644 --- a/osf_tests/test_notification_campaign.py +++ b/osf_tests/test_notification_campaign.py @@ -587,8 +587,8 @@ def running_campaign(self, campaign): campaign.save() return campaign - @mock.patch.object(NotificationType, 'emit') - def test_send_campaign_batch_marks_recipients_sent(self, mock_emit, running_campaign): + @mock.patch('osf.email.notification_campaign.send_email') + def test_send_campaign_batch_marks_recipients_sent(self, mock_send_email, running_campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) @@ -606,11 +606,11 @@ def test_send_campaign_batch_marks_recipients_sent(self, mock_emit, running_camp running_campaign.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.SENT assert running_campaign.sent_count == 1 - mock_emit.assert_called_once() + mock_send_email.assert_called_once() - @mock.patch.object(NotificationType, 'emit', side_effect=Exception('send failed')) + @mock.patch('osf.email.notification_campaign.send_email', side_effect=Exception('send failed')) @mock.patch('osf.email.notification_campaign.sentry.log_exception') - def test_send_campaign_batch_marks_recipients_failed(self, mock_sentry, mock_emit, running_campaign): + def test_send_campaign_batch_marks_recipients_failed(self, mock_sentry, mock_send_email, running_campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) @@ -739,7 +739,7 @@ def test_send_campaign_batch_logs_when_time_window_exceeded(self, mock_sentry, r batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) mock_sentry.reset_mock() - with mock.patch.object(NotificationType, 'emit'): + with mock.patch('osf.email.notification_campaign.send_email'): send_campaign_batch( context={}, batch_id=batch_id, @@ -765,7 +765,7 @@ def test_send_campaign_batch_skips_reminder_without_developer_reminder_flag( batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) mock_sentry.reset_mock() - with mock.patch.object(NotificationType, 'emit'): + with mock.patch('osf.email.notification_campaign.send_email'): send_campaign_batch( context={}, batch_id=batch_id, @@ -793,7 +793,7 @@ def test_send_campaign_batch_reminder_claimed_only_once(self, mock_sentry, runni batch_id_2 = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) mock_sentry.reset_mock() - with mock.patch.object(NotificationType, 'emit'): + with mock.patch('osf.email.notification_campaign.send_email'): send_campaign_batch( context={}, batch_id=batch_id_1, @@ -870,8 +870,8 @@ def test_send_campaign_batch_skips_cancelled_campaign(self, running_campaign): recipient.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.QUEUED - @mock.patch.object(NotificationType, 'emit') - def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, mock_emit, running_campaign): + @mock.patch('osf.email.notification_campaign.send_email') + def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, mock_send_email, running_campaign): user = UserFactory() user.username = 'invalid' user.save(update_fields=['username']) @@ -895,7 +895,7 @@ def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, m assert recipient.status == NotificationCampaignRecipientStatus.SENT assert running_campaign.sent_count == 1 assert running_campaign.failed_count == 0 - mock_emit.assert_called_once() + mock_send_email.assert_called_once() class TestNotificationCampaignCancel: diff --git a/tests/utils.py b/tests/utils.py index ba208573f37..54616ad86d8 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -320,7 +320,7 @@ def _wrapped_emit(self, *emit_args, **emit_kwargs): _real_send_over_smtp = _osf_email.send_email_over_smtp _real_send_with_sendgrid = _osf_email.send_email_with_send_grid - def _fake_send_over_smtp(to_email, notification_type, context=None, email_context=None): + def _fake_send_over_smtp(to_email, notification_type, context=None, email_context=None, rendered_html=None): captured['emails'].append({ 'protocol': 'smtp', 'to': to_email, @@ -329,9 +329,9 @@ def _fake_send_over_smtp(to_email, notification_type, context=None, email_contex 'email_context': email_context.copy() if isinstance(email_context, dict) else email_context, }) if passthrough: - return _real_send_over_smtp(to_email, notification_type, context, email_context) + return _real_send_over_smtp(to_email, notification_type, context, email_context, rendered_html) - def _fake_send_with_sendgrid(user, notification_type, context=None, email_context=None): + def _fake_send_with_sendgrid(user, notification_type, context=None, email_context=None, rendered_html=None): captured['emails'].append({ 'protocol': 'sendgrid', 'to': user, @@ -340,7 +340,7 @@ def _fake_send_with_sendgrid(user, notification_type, context=None, email_contex 'email_context': email_context.copy() if isinstance(email_context, dict) else email_context, }) if passthrough: - return _real_send_with_sendgrid(user, notification_type, context, email_context) + return _real_send_with_sendgrid(user, notification_type, context, email_context, rendered_html) patches.extend([ mock.patch('osf.email.send_email_over_smtp', new=_fake_send_over_smtp), From 59afb3ee0721f0dae3723f0a070fa45ee23fa974 Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Thu, 10 Sep 2026 17:40:13 +0300 Subject: [PATCH 12/16] ENG-12152] Fix admin progress data (#11909) * Remove recipient_count reset on campaign start to preserve state --- osf/models/notification_campaign.py | 1 - 1 file changed, 1 deletion(-) diff --git a/osf/models/notification_campaign.py b/osf/models/notification_campaign.py index 5429f82d6a3..ef3441233fb 100644 --- a/osf/models/notification_campaign.py +++ b/osf/models/notification_campaign.py @@ -84,7 +84,6 @@ def start(self, restart_failed=False, restart_stuck=False): self.started_at = timezone.now() self.run_id = uuid.uuid4() if not restart_failed and not restart_stuck: - self.recipient_count = 0 self.sent_count = 0 self.failed_count = 0 self.retries = 0 From 9b7fc50ea2c74daaf14708cc5dfadc779bb401df Mon Sep 17 00:00:00 2001 From: antkryt Date: Fri, 11 Sep 2026 18:29:14 +0300 Subject: [PATCH 13/16] [ENG-12091] Implement POC error handling/monitoring for SendGrid bulk send (#11905) * reshape process_campaign_retry task to handle bulk send * add sendgrid webhook * minor fixes; add webhook local test docs * fix tests * remove unnecessary comment * minor fixes; add unit tests for event processing * move process sendgrid events to medium queue * move process sendgrid events to external_low queue --- admin/notifications/forms.py | 3 + .../notification_campaing_create.html | 6 + api/base/urls.py | 1 + api/sendgrid/__init__.py | 0 api/sendgrid/permissions.py | 38 +++ api/sendgrid/urls.py | 9 + api/sendgrid/views.py | 87 +++++ osf/email/__init__.py | 59 ++-- osf/email/notification_campaign.py | 308 +++++++++++++----- osf_tests/test_notification_campaign.py | 235 ++++++++++++- website/settings/defaults.py | 7 +- 11 files changed, 658 insertions(+), 95 deletions(-) create mode 100644 api/sendgrid/__init__.py create mode 100644 api/sendgrid/permissions.py create mode 100644 api/sendgrid/urls.py create mode 100644 api/sendgrid/views.py diff --git a/admin/notifications/forms.py b/admin/notifications/forms.py index 6d4050c0c3f..ea9417a89c2 100644 --- a/admin/notifications/forms.py +++ b/admin/notifications/forms.py @@ -48,6 +48,9 @@ class NotificationCampaignCreateForm(forms.ModelForm): sendgrid_bulk = forms.BooleanField( required=False, initial=False, + help_text=( + 'SendGrid caps personalizations at about 1000 per request - keep Batch Size ≤ 1000 for bulk requests' + ), ) max_queued_batches = forms.IntegerField( diff --git a/admin/templates/notifications/notification_campaing_create.html b/admin/templates/notifications/notification_campaing_create.html index af66df3786f..37b8fe01e3b 100644 --- a/admin/templates/notifications/notification_campaing_create.html +++ b/admin/templates/notifications/notification_campaing_create.html @@ -278,6 +278,9 @@

Execution

name="sendgrid_bulk" value="{{ form.sendgrid_bulk.initial }}" > +

+ {{ form.sendgrid_bulk.help_text }} +

@@ -300,6 +303,9 @@

Execution

name="dispatch_interval" value="{{ form.dispatch_interval.initial }}" > +

+ {{ form.dispatch_interval.help_text }} +

diff --git a/api/base/urls.py b/api/base/urls.py index 63984efd287..8572f9ae6b0 100644 --- a/api/base/urls.py +++ b/api/base/urls.py @@ -21,6 +21,7 @@ re_path(r'^ia/', include('api.ia.urls', namespace='ia')), re_path(r'^banners/', include('api.banners.urls', namespace='banners')), re_path(r'^crossref/', include('api.crossref.urls', namespace='crossref')), + re_path(r'^sendgrid/', include('api.sendgrid.urls', namespace='sendgrid')), re_path(r'^chronos/', include('api.chronos.urls', namespace='chronos')), re_path(r'^cedar_metadata_templates/', include('api.cedar_metadata_templates.urls', namespace='cedar-metadata-templates')), re_path(r'^cedar_metadata_records/', include('api.cedar_metadata_records.urls', namespace='cedar-metadata-records')), diff --git a/api/sendgrid/__init__.py b/api/sendgrid/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/api/sendgrid/permissions.py b/api/sendgrid/permissions.py new file mode 100644 index 00000000000..067eaa3168a --- /dev/null +++ b/api/sendgrid/permissions.py @@ -0,0 +1,38 @@ +from rest_framework import permissions +from rest_framework import exceptions +from sendgrid.helpers.eventwebhook import EventWebhook, EventWebhookHeader + +from framework import sentry +from website import settings + + +class RequestComesFromSendGrid(permissions.BasePermission): + """Verify that the request comes from SendGrid via signed Event Webhook. + + Uses ECDSA signature verification against the raw request body as documented at: + https://www.twilio.com/docs/sendgrid/for-developers/tracking-events/getting-started-event-webhook-security-features + """ + + def has_permission(self, request, view): + if request.method != 'POST': + raise exceptions.MethodNotAllowed(method=request.method) + + public_key = settings.SENDGRID_EVENT_WEBHOOK_PUBLIC_KEY + if not public_key: + sentry.log_message('SendGrid Event Webhook public key is not configured') + return False + + signature = request.headers.get(EventWebhookHeader.SIGNATURE) + timestamp = request.headers.get(EventWebhookHeader.TIMESTAMP) + if not signature or not timestamp: + error_message = 'SendGrid Event Webhook signature headers required' + sentry.log_message(error_message) + raise exceptions.ParseError(error_message) + + # Must verify the raw body; re-serializing parsed JSON breaks the signature. + payload = request.body.decode('utf-8') + event_webhook = EventWebhook(public_key) + if not event_webhook.verify_signature(payload, signature, timestamp): + raise exceptions.ParseError('Invalid SendGrid Event Webhook signature') + + return True diff --git a/api/sendgrid/urls.py b/api/sendgrid/urls.py new file mode 100644 index 00000000000..64ce9714bf3 --- /dev/null +++ b/api/sendgrid/urls.py @@ -0,0 +1,9 @@ +from django.urls import re_path + +from api.sendgrid import views + +app_name = 'osf' + +urlpatterns = [ + re_path(r'^events/$', views.SendGridEventWebhook.as_view(), name=views.SendGridEventWebhook.view_name), +] diff --git a/api/sendgrid/views.py b/api/sendgrid/views.py new file mode 100644 index 00000000000..45f14c235f1 --- /dev/null +++ b/api/sendgrid/views.py @@ -0,0 +1,87 @@ +import json +import logging + +from django.http import HttpResponse +from django.views.decorators.csrf import csrf_exempt +from rest_framework.views import APIView + +from api.sendgrid.permissions import RequestComesFromSendGrid +from osf.email.notification_campaign import ( + CAMPAIGN_CUSTOM_ARG_KEYS, + process_sendgrid_campaign_events, +) + +logger = logging.getLogger(__name__) + + +def _is_campaign_related_event(event): + """SendGrid flattens personalization ``custom_args`` onto each event object.""" + if not isinstance(event, dict): + return False + return all(event.get(key) for key in CAMPAIGN_CUSTOM_ARG_KEYS) + + +class SendGridEventWebhook(APIView): + """Receive SendGrid Event Webhook POSTs and enqueue campaign event handling. + + Mounted under the ``_/`` namespace (no user auth). Non-campaign events are + ignored; campaign-tagged events are processed asynchronously. + + Local testing + ------------- + Requests must be ECDSA-signed by SendGrid (see ``RequestComesFromSendGrid``), + so plain ``curl`` posts are rejected unless you forge a valid signature or + disable the signature check. Use a public tunnel and a real SendGrid Event Webhook + instead: + + 1. Run the API (``localhost:8000``) and a Celery worker (events are enqueued + to ``email.process_sendgrid_campaign_events``). + 2. Expose the API, e.g. ``ngrok http 8000``. Add ``*.ngrok-free.dev`` in + ``ALLOWED_HOSTS`` in the settings. See https://ngrok.com/docs/share-localhost/overview + for more details. + 3. In SendGrid → Mail Settings → Event Webhook: + - POST URL: ``https:///_/sendgrid/events/`` + - Enable Signed Event Webhook; copy the verification key into + ``SENDGRID_EVENT_WEBHOOK_PUBLIC_KEY`` (``website/settings/local.py``). + - Subscribe at least to ``delivered``, ``bounce``, and ``dropped``. + 4. Send a notification campaign email (personalization ``custom_args`` must + include ``campaign_id``, ``campaign_recipient_id``, and ``run_id``). + SendGrid's "Test Your Integration" payload lacks those keys, so this view + accepts it with 200 but does not update campaign recipients. + """ + + view_name = 'sendgrid_event_webhook' + view_category = 'sendgrid' + + authentication_classes = () + permission_classes = ( + RequestComesFromSendGrid, + ) + + @csrf_exempt + def dispatch(self, request, *args, **kwargs): + return super().dispatch(request, *args, **kwargs) + + def get_serializer_class(self): + return None + + def post(self, request): + try: + events = json.loads(request.body) + except (TypeError, ValueError, json.JSONDecodeError): + return HttpResponse('Invalid JSON', status=400) + + if isinstance(events, dict): + events = [events] + if not isinstance(events, list): + return HttpResponse('Expected a JSON array of events', status=400) + + # Process only campaign-related events + # TODO: add handling for non-campaign events here if required + campaign_events = [event for event in events if _is_campaign_related_event(event)] + + if campaign_events: + process_sendgrid_campaign_events.delay(campaign_events) + logger.info(f'Enqueued {len(campaign_events)} campaign-related SendGrid event(s) of {len(events)} received') + + return HttpResponse('Events received', status=200) diff --git a/osf/email/__init__.py b/osf/email/__init__.py index 6f399dca55f..54be37c898d 100644 --- a/osf/email/__init__.py +++ b/osf/email/__init__.py @@ -238,30 +238,51 @@ def send_email_over_smtp(to_email, notification_type, context, email_context, re email.attach(attachment_name, attachment_content) email.send() +def _email_objects(addrs): + if not addrs: + return None + if isinstance(addrs, str): + addrs = [addrs] + return [{'email': a} for a in addrs] + + def _build_sendgrid_personalizations(to_list, email_context=None, is_multiple=False): - """Build SendGrid personalizations for one shared message or one message per recipient. + """Build SendGrid personalizations. When ``is_multiple`` is True, each address gets its own personalization (separate - delivery; recipients do not see each other). CC/BCC are omitted in that mode to - avoid duplicating copies per recipient. + delivery; recipients do not see each other). + + Optional ``email_context`` keys: + - ``custom_args``: dict applied to every personalization + - ``custom_args_list``: list of dicts parallel to ``to_list`` (used when + ``is_multiple`` is True; each entry is attached to that recipient only) """ email_context = email_context or {} - - if is_multiple: - # If we attached the same CC/BCC to every personalization in is_multiple mode, - # a batch of N recipients would produce N separate emails each including that CC - # so the CC address would get N copies of the same message - # TODO: decide if it's safe to add the CC/BCC to each personalization - return [{'to': [{'email': addr}]} for addr in to_list] - - personalization = {'to': [{'email': addr} for addr in to_list]} - cc_addr = email_context.get('cc_addr') - if cc_addr: - personalization['cc'] = [{'email': a} for a in ([cc_addr] if isinstance(cc_addr, str) else cc_addr)] - bcc_addr = email_context.get('bcc_addr') - if bcc_addr: - personalization['bcc'] = [{'email': a} for a in ([bcc_addr] if isinstance(bcc_addr, str) else bcc_addr)] - return [personalization] + cc = _email_objects(email_context.get('cc_addr')) + bcc = _email_objects(email_context.get('bcc_addr')) + shared_custom_args = email_context.get('custom_args') + custom_args_list = email_context.get('custom_args_list') or [] + + def personalization(recipients, custom_args=None): + item = {'to': [{'email': a} for a in recipients]} + if cc: + item['cc'] = cc + if bcc: + item['bcc'] = bcc + if custom_args: + item['custom_args'] = {str(k): str(v) for k, v in custom_args.items()} + return item + + if not is_multiple: + return [personalization(to_list, shared_custom_args)] + + return [ + personalization( + [addr], + custom_args_list[i] if i < len(custom_args_list) else shared_custom_args, + ) + for i, addr in enumerate(to_list) + ] def send_email_with_send_grid(to_addr, notification_type, context, email_context=None, *, is_multiple=False, rendered_html=None): diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index 70826926435..c36b6c25451 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -3,7 +3,7 @@ from osf.models import NotificationType, NotificationTypeEnum, OSFUser, UserActivityCounter, Email from osf.models.spam import SpamStatus from django.db import transaction -from django.db.models import OuterRef, Subquery, Case, When, CharField, Count, Q, BooleanField +from django.db.models import OuterRef, Subquery, Case, When, Value, CharField, Count, Q, BooleanField, TextField from django.db.models.functions import Coalesce from framework.celery_tasks import app as celery_app from django.utils import timezone @@ -23,6 +23,11 @@ 'internal': {'is_active': True, 'is_staff': True, 'username__endswith': '@cos.io'}, } +# Flattened onto SendGrid Event Webhook payloads via personalization custom_args. +CAMPAIGN_CUSTOM_ARG_KEYS = ('campaign_id', 'campaign_recipient_id', 'run_id') +SENDGRID_SUCCESS_EVENTS = frozenset({'delivered'}) +SENDGRID_FAILURE_EVENTS = frozenset({'bounce', 'dropped'}) + first_email_subquery = ( Email.objects .filter(user=OuterRef('user_id')) @@ -37,6 +42,39 @@ ) +class NotificationCampaignTask(celery_app.Task): + """Shared guards for notification campaign Celery tasks.""" + + abstract = True + + def get_campaign(self, campaign_id, run_id=None, *, abort_if_cancelled=True): + """Load a campaign, or return None if the task should no-op. + """ + campaign = NotificationCampaign.objects.get(id=campaign_id) + if run_id is not None and campaign.run_id != run_id: + return None + if abort_if_cancelled and campaign.status == NotificationCampaignStatus.CANCELLED: + logger.warning(f"Campaign {campaign_id} was cancelled") + return None + return campaign + + def sync_campaign_stats(self, campaign): + stats = get_campaign_recipient_stats(campaign.id) + campaign.recipient_count = stats['recipient_count'] + campaign.sent_count = stats['sent_count'] + campaign.failed_count = stats['failed_count'] + return stats + + def finish_campaign(self, campaign, status=None): + """Sync recipient counters, set completed_at once, optionally update status, and save.""" + self.sync_campaign_stats(campaign) + if campaign.completed_at is None: + campaign.completed_at = timezone.now() + if status is not None: + campaign.status = status + campaign.save() + + def build_query(node): """ Convert a filter tree into a Django Q object. @@ -160,65 +198,176 @@ def get_campaign_recipient_stats(campaign_id): ), ) -@celery_app.task(name='email.process_campaign_retry') -def process_campaign_retry(*args, **kwargs): - campaign_id = kwargs.get('campaign_id') - campaign = NotificationCampaign.objects.get(id=campaign_id) - if kwargs.get('run_id') != campaign.run_id: + +@celery_app.task(name='email.process_sendgrid_campaign_events') +def process_sendgrid_campaign_events(events): + """Update campaign recipients from filtered SendGrid Event Webhook events. + + Expects events that already include campaign ``custom_args`` + (``campaign_id``, ``campaign_recipient_id``, ``run_id``). Only ``QUEUED`` + or ``FAILED`` recipients whose event ``run_id`` matches the campaign's + current run are updated; delayed webhooks from a prior run are ignored. + + A ``delivered`` event wins over failure events for the same recipient + (including a prior ``FAILED`` from an earlier webhook) so a confirmed + delivery is never left failed (and retried). + """ + campaign_ids = { + event.get('campaign_id') + for event in events + if event.get('campaign_id', False) + } + if not campaign_ids: return - final_status = NotificationCampaignStatus.COMPLETED - - if campaign.status != NotificationCampaignStatus.CANCELLED: - failed_recipients = NotificationCampaignRecipient.objects.filter(campaign=campaign, status=NotificationCampaignRecipientStatus.FAILED) - max_retries = campaign.metadata.get('execution', {}).get('max_retries', settings.DEFAULT_CAMPAIGN_MAX_RETRIES) - failed_recipients_count = failed_recipients.count() - if failed_recipients_count: - if campaign.retries < max_retries: - message = (f'[Notification Campaign #{campaign_id}] WARNING: ' - f'Retrying {failed_recipients_count} failed recipients, ' - f'previous retry attempts: {campaign.retries}/{max_retries}') - logger.info(message) - sentry.log_message(message) - campaign.retries += 1 - campaign.save(update_fields=['retries', 'updated_at']) + current_run_ids = { + str(campaign.id): str(campaign.run_id) + for campaign in NotificationCampaign.objects.filter( + id__in=campaign_ids, + run_id__isnull=False, + ) + } + if not current_run_ids: + return - dispatch_campaign.apply_async( - args=[campaign_id, campaign.run_id], - kwargs={ - 'restart_failed': True, - }, - ) - return + success_ids = set() + failed = dict() - final_status = NotificationCampaignStatus.PARTIALLY_COMPLETED - else: - message = f'[Notification Campaign #{campaign_id}] WARNING: Campaign {campaign.name} was cancelled.' - logger.info(message) - sentry.log_message(message) + for event in events: + campaign_id = event.get('campaign_id', '') + current_run_id = current_run_ids.get(str(campaign_id), None) + if current_run_id is None or event.get('run_id') != current_run_id: + continue - # Refresh in case the campaign was cancelled while we were running. - campaign.refresh_from_db(fields=['status', 'completed_at']) + event_type = event.get('event') + recipient_id = event.get('campaign_recipient_id') + if not recipient_id: + continue + try: + recipient_pk = int(recipient_id) + except (TypeError, ValueError): + continue + + if event_type in SENDGRID_SUCCESS_EVENTS: + success_ids.add(recipient_pk) + failed.pop(recipient_pk, None) + elif event_type in SENDGRID_FAILURE_EVENTS: + if recipient_pk in success_ids: + continue + error_message = event.get('reason') or event.get('type') or event_type + failed[recipient_pk] = error_message + + if success_ids: + NotificationCampaignRecipient.objects.filter( + id__in=success_ids, + status__in=[ + NotificationCampaignRecipientStatus.QUEUED, + NotificationCampaignRecipientStatus.FAILED, + ], + ).update(status=NotificationCampaignRecipientStatus.SENT, error_message=None) - # Sync statistics regardless of status. - stats = get_campaign_recipient_stats(campaign_id) - campaign.recipient_count = stats['recipient_count'] - campaign.sent_count = stats['sent_count'] - campaign.failed_count = stats['failed_count'] + if failed: + failed_errors = [ + When(id=recipient_pk, then=Value(error_message)) + for recipient_pk, error_message in failed.items() + ] + NotificationCampaignRecipient.objects.filter( + id__in=failed.keys(), + status=NotificationCampaignRecipientStatus.QUEUED, + ).update( + status=NotificationCampaignRecipientStatus.FAILED, + error_message=Case( + *failed_errors, + default=Value('SendGrid delivery failed'), + output_field=TextField(), + ), + ) - if campaign.completed_at is None: - campaign.completed_at = timezone.now() - # Don't overwrite CANCELLED. - if campaign.status != NotificationCampaignStatus.CANCELLED: - campaign.status = final_status +@celery_app.task(bind=True, base=NotificationCampaignTask, name='email.process_campaign_retry') +def process_campaign_retry(self, campaign_id, run_id): + campaign = self.get_campaign(campaign_id, run_id, abort_if_cancelled=False) + if campaign is None: + return - campaign.save() + campaign.refresh_from_db() + execution = campaign.metadata.get('execution', {}) + queued_qs = NotificationCampaignRecipient.objects.filter( + campaign=campaign, + status=NotificationCampaignRecipientStatus.QUEUED, + ) + if queued_qs.exists(): + delivery_timeout = execution.get('delivery_timeout', settings.DEFAULT_CAMPAIGN_DELIVERY_TIMEOUT) + reference_time = campaign.started_at or campaign.created_at + if timezone.now() - reference_time < timedelta(seconds=delivery_timeout): + # Still waiting for in-flight sends / SendGrid delivery webhooks. + self.sync_campaign_stats(campaign) + campaign.save() + process_campaign_retry.apply_async( + kwargs={'campaign_id': campaign_id, 'run_id': campaign.run_id}, + countdown=execution.get('dispatch_interval', settings.CAMPAIGN_DISPATCH_INTERVAL), + ) + return -@celery_app.task(name='email.start_notification_campaign') -def start_notification_campaign(campaign_id, restart_failed=False, restart_stuck=False): - campaign = NotificationCampaign.objects.get(id=campaign_id) + timed_out = queued_qs.update( + status=NotificationCampaignRecipientStatus.FAILED, + error_message='SendGrid delivery timeout', + ) + message = ( + f'[Notification Campaign #{campaign_id}] WARNING: ' + f'Marked {timed_out} queued recipients as FAILED after delivery timeout ' + f'({delivery_timeout}s) for campaign {campaign.name}.' + ) + logger.warning(message) + sentry.log_message(message) + + # Do not retry timed-out deliveries; close the run as partially completed. + self.finish_campaign(campaign, NotificationCampaignStatus.PARTIALLY_COMPLETED) + return + + if campaign.status == NotificationCampaignStatus.CANCELLED: + message = f'[Notification Campaign #{campaign_id}] WARNING: Campaign {campaign.name} was cancelled.' + logger.info(message) + sentry.log_message(message) + self.finish_campaign(campaign) + return + + failed_recipients_count = NotificationCampaignRecipient.objects.filter( + campaign=campaign, + status=NotificationCampaignRecipientStatus.FAILED, + ).count() + max_retries = execution.get('max_retries', settings.DEFAULT_CAMPAIGN_MAX_RETRIES) + + if failed_recipients_count: + if campaign.retries < max_retries: + message = (f'[Notification Campaign #{campaign_id}] WARNING: ' + f'Retrying {failed_recipients_count} failed recipients, ' + f'previous retry attempts: {campaign.retries}/{max_retries}') + logger.info(message) + sentry.log_message(message) + campaign.retries += 1 + campaign.save() + + dispatch_campaign.apply_async( + args=[campaign_id, campaign.run_id], + kwargs={ + 'restart_failed': True, + }, + ) + return + + final_status = NotificationCampaignStatus.PARTIALLY_COMPLETED + else: + final_status = NotificationCampaignStatus.COMPLETED + + self.finish_campaign(campaign, final_status) + +@celery_app.task(bind=True, base=NotificationCampaignTask, name='email.start_notification_campaign') +def start_notification_campaign(self, campaign_id, restart_failed=False, restart_stuck=False): + campaign = self.get_campaign(campaign_id) + if campaign is None: + return notification_type_name = campaign.notification_type.name if hasattr(NotificationTypeEnum, notification_type_name): @@ -278,15 +427,12 @@ def assign_batch_id_to_recipients( return batch_id -@celery_app.task(bind=True, name='email.dispatch_campaign') -def dispatch_campaign(self, campaign_id, run_id, restart_failed=False): - campaign = NotificationCampaign.objects.get(id=campaign_id) - if campaign.run_id != run_id: +@celery_app.task(bind=True, base=NotificationCampaignTask, name='email.dispatch_campaign') +def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_stuck=False): + campaign = self.get_campaign(campaign_id, run_id) + if campaign is None: return - if campaign.status == NotificationCampaignStatus.CANCELLED: - logger.warning(f"Campaign {campaign_id} was cancelled") - return if campaign.status != NotificationCampaignStatus.RUNNING: message = f'[Notification Campaign #{campaign_id}] ERROR: Campaign {campaign.name} is not in RUNNING status.' logger.error(message) @@ -324,6 +470,7 @@ def dispatch_campaign(self, campaign_id, run_id, restart_failed=False): ) total_new_queued_batches = 0 new_queued_batches = 0 + no_more_recipients = False for priority_group in priority_groups: no_more_recipients = False developer_reminder = priority_group.pop('developer_reminder', False) @@ -371,8 +518,9 @@ def dispatch_campaign(self, campaign_id, run_id, restart_failed=False): ) -@celery_app.task(name='email.send_campaign_batch', ignore_result=False) +@celery_app.task(bind=True, base=NotificationCampaignTask, name='email.send_campaign_batch', ignore_result=False) def send_campaign_batch( + self, context, batch_id=None, notification_type_name='blank', @@ -380,14 +528,12 @@ def send_campaign_batch( run_id=None, developer_reminder=False, ): - campaign = NotificationCampaign.objects.get(id=campaign_id) + campaign = self.get_campaign(campaign_id, run_id) + if campaign is None: + return + recipients_qs = NotificationCampaignRecipient.objects.filter(batch_id=batch_id).select_related('user') - if campaign.run_id != run_id: - return - if campaign.status == NotificationCampaignStatus.CANCELLED: - logger.warning(f"Campaign {campaign_id} was cancelled") - return batch_started_at = timezone.now() if hasattr(NotificationTypeEnum, notification_type_name): notification_type = getattr(NotificationTypeEnum, notification_type_name).instance @@ -437,16 +583,27 @@ def send_campaign_batch( invalid_emails_qs.update(status=NotificationCampaignRecipientStatus.SKIPPED, error_message='Invalid email address') if campaign.metadata.get('sendgrid_bulk', False): - # NOTE: sendgrid bulk send feature has not been fully implemented and tested - recipient_emails = list(valid_emails_qs.values_list('recipient_address', flat=True)) + recipients = list(valid_emails_qs) + recipient_emails = [] + custom_args_list = [] + for recipient in recipients: + recipient_emails.append(recipient.recipient_address) + custom_args_list.append( + { + 'campaign_recipient_id': str(recipient.id), + 'campaign_id': str(campaign_id), + 'run_id': str(run_id), + } + ) try: send_email_with_send_grid( to_addr=recipient_emails, notification_type=notification_type, context=context, + email_context={'custom_args_list': custom_args_list}, is_multiple=True, ) - valid_emails_qs.update(status=NotificationCampaignRecipientStatus.SENT, error_message=None) + # Leave QUEUED until SendGrid Event Webhook confirms delivery. except Exception as exc: message = (f'[Notification Campaign #{campaign_id}] ERROR: ' f'Campaign {campaign.name} sendgrid bulk request failed, error={str(exc)}') @@ -462,8 +619,14 @@ def send_campaign_batch( recipient_address=recipient.recipient_address, notification_type=notification_type, event_context=context, - email_context=context, - rendered_html=rendered_html + email_context={ + 'custom_args': { + 'campaign_recipient_id': str(recipient.id), + 'campaign_id': str(campaign_id), + 'run_id': str(run_id), + }, + }, + rendered_html=rendered_html, ) recipient.status = NotificationCampaignRecipientStatus.SENT recipient.error_message = None @@ -486,15 +649,14 @@ def send_campaign_batch( f'campaign_name={campaign.name}') logger.warning(message) sentry.log_message(message) - NotificationCampaignRecipient.objects.bulk_update(recipient_records, ['status', 'error_message']) + if recipient_records: + NotificationCampaignRecipient.objects.bulk_update(recipient_records, ['status', 'error_message']) # Lock the campaign row so concurrent batches cannot overwrite counters with a stale aggregate snapshot with transaction.atomic(): notification_campaign = NotificationCampaign.objects.select_for_update().get(pk=campaign_id) - stats = get_campaign_recipient_stats(campaign_id) - notification_campaign.sent_count = stats['sent_count'] - notification_campaign.failed_count = stats['failed_count'] - notification_campaign.save(update_fields=['sent_count', 'failed_count', 'updated_at']) + self.sync_campaign_stats(notification_campaign) + notification_campaign.save(update_fields=['sent_count', 'failed_count', 'recipient_count', 'updated_at']) batch_finished_at = timezone.now() batch_run_time = (batch_finished_at - batch_started_at).total_seconds() diff --git a/osf_tests/test_notification_campaign.py b/osf_tests/test_notification_campaign.py index 8f6f4a4cc52..04504c16c75 100644 --- a/osf_tests/test_notification_campaign.py +++ b/osf_tests/test_notification_campaign.py @@ -8,10 +8,12 @@ from osf.email import _build_sendgrid_personalizations from osf.email.notification_campaign import ( + NotificationCampaignTask, assign_batch_id_to_recipients, create_campaign_recipients, get_campaign_recipient_stats, process_campaign_retry, + process_sendgrid_campaign_events, send_campaign_batch, start_notification_campaign, build_query, @@ -30,6 +32,47 @@ pytestmark = pytest.mark.django_db +class TestNotificationCampaignTask: + + @pytest.fixture + def task(self): + return NotificationCampaignTask() + + def test_get_campaign_returns_campaign(self, task, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + loaded = task.get_campaign(campaign.id, campaign.run_id) + assert loaded.pk == campaign.pk + + def test_get_campaign_skips_stale_run_id(self, task, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + assert task.get_campaign(campaign.id, uuid.uuid4()) is None + + def test_get_campaign_skips_cancelled_by_default(self, task, campaign): + campaign.status = NotificationCampaignStatus.CANCELLED + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['status', 'run_id']) + assert task.get_campaign(campaign.id, campaign.run_id) is None + + def test_get_campaign_can_include_cancelled(self, task, campaign): + campaign.status = NotificationCampaignStatus.CANCELLED + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['status', 'run_id']) + loaded = task.get_campaign(campaign.id, campaign.run_id, abort_if_cancelled=False) + assert loaded.pk == campaign.pk + + def test_sync_campaign_stats(self, task, campaign): + user = UserFactory() + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) + NotificationCampaignRecipient.objects.filter(campaign=campaign).update( + status=NotificationCampaignRecipientStatus.SENT + ) + stats = task.sync_campaign_stats(campaign) + assert stats['sent_count'] == 1 + assert campaign.sent_count == 1 + + @pytest.fixture def notification_type(): notification_type, _ = NotificationType.objects.get_or_create(name='blank') @@ -676,8 +719,14 @@ def test_send_campaign_batch_sendgrid_bulk_success(self, mock_sendgrid, running_ running_campaign.refresh_from_db() mock_sendgrid.assert_called_once() assert mock_sendgrid.call_args.kwargs.get('is_multiple') is True - assert recipient.status == NotificationCampaignRecipientStatus.SENT - assert running_campaign.sent_count == 1 + email_context = mock_sendgrid.call_args.kwargs.get('email_context') or {} + assert email_context['custom_args_list'] == [{ + 'campaign_recipient_id': str(recipient.id), + 'campaign_id': str(running_campaign.id), + 'run_id': str(running_campaign.run_id), + }] + assert recipient.status == NotificationCampaignRecipientStatus.QUEUED + assert running_campaign.sent_count == 0 assert running_campaign.failed_count == 0 def test_build_sendgrid_personalizations_shared_to_list(self): @@ -702,6 +751,28 @@ def test_build_sendgrid_personalizations_one_per_recipient(self): {'to': [{'email': 'b@example.com'}]}, ] + def test_build_sendgrid_personalizations_attaches_custom_args(self): + personalizations = _build_sendgrid_personalizations( + ['a@example.com', 'b@example.com'], + email_context={ + 'custom_args_list': [ + {'campaign_recipient_id': '1', 'campaign_id': '9'}, + {'campaign_recipient_id': '2', 'campaign_id': '9'}, + ], + }, + is_multiple=True, + ) + assert personalizations == [ + { + 'to': [{'email': 'a@example.com'}], + 'custom_args': {'campaign_recipient_id': '1', 'campaign_id': '9'}, + }, + { + 'to': [{'email': 'b@example.com'}], + 'custom_args': {'campaign_recipient_id': '2', 'campaign_id': '9'}, + }, + ] + @mock.patch( 'osf.email.notification_campaign.send_email_with_send_grid', side_effect=Exception('bulk failed'), @@ -1031,3 +1102,163 @@ def test_process_campaign_retry_marks_partially_completed_after_max_retries(self assert campaign.failed_count == 1 assert campaign.recipient_count == 1 assert campaign.completed_at is not None + + @mock.patch('osf.email.notification_campaign.process_campaign_retry.apply_async') + def test_process_campaign_retry_waits_for_queued_recipients(self, mock_apply_async, campaign): + user = UserFactory() + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) + NotificationCampaignRecipient.objects.filter(campaign=campaign).update( + status=NotificationCampaignRecipientStatus.QUEUED + ) + campaign.run_id = uuid.uuid4() + campaign.status = NotificationCampaignStatus.RUNNING + campaign.started_at = timezone.now() + campaign.metadata['execution']['dispatch_interval'] = 60 + campaign.save() + + process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) + + campaign.refresh_from_db() + assert campaign.status == NotificationCampaignStatus.RUNNING + assert campaign.completed_at is None + assert campaign.recipient_count == 1 + assert campaign.sent_count == 0 + assert campaign.failed_count == 0 + mock_apply_async.assert_called_once_with( + kwargs={'campaign_id': campaign.id, 'run_id': campaign.run_id}, + countdown=60, + ) + + def test_process_campaign_retry_times_out_queued_marks_partial_without_retry(self, campaign): + user = UserFactory() + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) + NotificationCampaignRecipient.objects.filter(campaign=campaign).update( + status=NotificationCampaignRecipientStatus.QUEUED + ) + campaign.run_id = uuid.uuid4() + campaign.retries = 0 + campaign.status = NotificationCampaignStatus.RUNNING + campaign.started_at = timezone.now() - timedelta(seconds=10) + campaign.metadata['execution']['delivery_timeout'] = 1 + campaign.save() + + process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) + + recipient = NotificationCampaignRecipient.objects.get(campaign=campaign, user=user) + assert recipient.status == NotificationCampaignRecipientStatus.FAILED + assert recipient.error_message == 'SendGrid delivery timeout' + campaign.refresh_from_db() + assert campaign.status == NotificationCampaignStatus.PARTIALLY_COMPLETED + assert campaign.retries == 0 + assert campaign.failed_count == 1 + assert campaign.completed_at is not None + + @mock.patch('osf.email.notification_campaign.process_campaign_retry.apply_async') + @mock.patch('osf.email.notification_campaign.dispatch_campaign.apply_async') + def test_process_campaign_retry_waits_for_queued_before_retrying_failed( + self, mock_dispatch_campaign, mock_apply_async, campaign + ): + queued_user = UserFactory() + failed_user = UserFactory() + create_campaign_recipients( + Q(**{'id__in': [queued_user.id, failed_user.id]}), + campaign_id=campaign.id, + ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=queued_user).update( + status=NotificationCampaignRecipientStatus.QUEUED + ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=failed_user).update( + status=NotificationCampaignRecipientStatus.FAILED + ) + campaign.run_id = uuid.uuid4() + campaign.retries = 0 + campaign.status = NotificationCampaignStatus.RUNNING + campaign.started_at = timezone.now() + campaign.save() + + process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) + + campaign.refresh_from_db() + assert campaign.status == NotificationCampaignStatus.RUNNING + assert campaign.retries == 0 + assert campaign.completed_at is None + mock_apply_async.assert_called_once() + mock_dispatch_campaign.assert_not_called() + + +class TestProcessSendgridCampaignEvents: + + def _event(self, campaign, recipient, event, **extra): + payload = { + 'event': event, + 'campaign_id': str(campaign.id), + 'campaign_recipient_id': str(recipient.id), + 'run_id': str(campaign.run_id), + } + payload.update(extra) + return payload + + def _queued_recipient(self, campaign, user=None): + user = user or UserFactory() + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) + recipient = NotificationCampaignRecipient.objects.get(campaign=campaign, user=user) + recipient.status = NotificationCampaignRecipientStatus.QUEUED + recipient.save(update_fields=['status']) + return recipient + + def test_delivered_marks_queued_sent(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._queued_recipient(campaign) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'delivered'), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.SENT + assert recipient.error_message is None + + def test_failure_marks_queued_failed(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._queued_recipient(campaign) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'bounce', reason='mailbox full'), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.FAILED + assert recipient.error_message == 'mailbox full' + + def test_delivered_overrides_failed(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._queued_recipient(campaign) + recipient.status = NotificationCampaignRecipientStatus.FAILED + recipient.error_message = 'bounce' + recipient.save(update_fields=['status', 'error_message']) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'delivered'), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.SENT + assert recipient.error_message is None + + def test_ignores_stale_run_id(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._queued_recipient(campaign) + + process_sendgrid_campaign_events([{ + 'event': 'delivered', + 'campaign_id': str(campaign.id), + 'campaign_recipient_id': str(recipient.id), + 'run_id': str(uuid.uuid4()), + }]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.QUEUED diff --git a/website/settings/defaults.py b/website/settings/defaults.py index 9ff403b24b8..16ecf034ed2 100644 --- a/website/settings/defaults.py +++ b/website/settings/defaults.py @@ -170,6 +170,8 @@ def parent_dir(path): # OR, if using Sendgrid's API # WARNING: If `SENDGRID_WHITELIST_MODE` is True, SENDGRID_API_KEY = None +# Public verification key from SendGrid Event Webhook (Mail Settings -> Event Webhook -> Signed Event Webhook) +SENDGRID_EVENT_WEBHOOK_PUBLIC_KEY = None # Mailchimp MAILCHIMP_API_KEY = None @@ -197,6 +199,7 @@ def parent_dir(path): DEFAULT_CAMPAIGN_BATCH_SIZE = 1000 DEFAULT_CAMPAIGN_WINDOW_TIME = 28800 # 8 hours DEFAULT_CAMPAIGN_MAX_RETRIES = 3 +DEFAULT_CAMPAIGN_DELIVERY_TIMEOUT = 86400 # 24 hours; mark remaining QUEUED as FAILED after this from started_at MAX_QUEUED_CAMPAIGN_BATCHES = 100 # Maximum number of queued campaign batches allowed before new batches are rejected. This is to prevent runaway campaigns from overwhelming the system. CAMPAIGN_DISPATCH_INTERVAL = 300 # 5 min (300 sec), minimum time before checking and dispatching new campaign batches. # The following are rough estimates so we can log to sentry those batches and sendgrid quests which run longer than normal @@ -451,7 +454,9 @@ class CeleryConfig: 'website.identifiers.tasks.task__update_verified_links' } - external_low_modules = {} + external_low_modules = { + 'email.process_sendgrid_campaign_events', + } account_status_changes_modules = {} From ddc3d497751bfd429e6b5dd99fff0a91d3442fef Mon Sep 17 00:00:00 2001 From: Longze Chen Date: Fri, 11 Sep 2026 13:03:55 -0400 Subject: [PATCH 14/16] Redo post-ENTER migration after 26.21 merge --- ...> 0055_notificationcampaignrecipient_batch_id_and_more.py} | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) rename osf/migrations/{0054_post_enter_project.py => 0055_notificationcampaignrecipient_batch_id_and_more.py} (87%) diff --git a/osf/migrations/0054_post_enter_project.py b/osf/migrations/0055_notificationcampaignrecipient_batch_id_and_more.py similarity index 87% rename from osf/migrations/0054_post_enter_project.py rename to osf/migrations/0055_notificationcampaignrecipient_batch_id_and_more.py index 3c867947385..9cc303d5fcb 100644 --- a/osf/migrations/0054_post_enter_project.py +++ b/osf/migrations/0055_notificationcampaignrecipient_batch_id_and_more.py @@ -1,4 +1,4 @@ -# Generated by Django 4.2.26 on 2026-09-08 20:13 +# Generated by Django 4.2.26 on 2026-09-11 17:03 from django.db import migrations, models @@ -6,7 +6,7 @@ class Migration(migrations.Migration): dependencies = [ - ('osf', '0053_gdpr_delete_and_orcid_revoke'), + ('osf', '0054_merge_20260909_reconcile'), ] operations = [ From cee567207b674a74c0a844cddf0ac7e85818984a Mon Sep 17 00:00:00 2001 From: Ostap Zherebetskyi Date: Tue, 15 Sep 2026 16:06:55 +0300 Subject: [PATCH 15/16] Update recipient count logic in create_campaign_recipients to reflect processed records --- osf/email/notification_campaign.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index c36b6c25451..3ce4b2bd1d0 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -148,6 +148,7 @@ def create_campaign_recipients(filters=None, campaign_id=None): 'activity_score', ) ) + processed_records = 0 for rows in batched(qs.iterator(chunk_size=BULK_CREATE_SIZE), BULK_CREATE_SIZE): NotificationCampaignRecipient.objects.bulk_create( @@ -163,10 +164,11 @@ def create_campaign_recipients(filters=None, campaign_id=None): update_fields=['activity_score'], unique_fields=['campaign', 'user'], ) + processed_records += len(rows) - campaign.recipient_count = NotificationCampaignRecipient.objects.filter(campaign_id=campaign_id).count() + campaign.recipient_count = processed_records campaign.metadata['recipients_creation_finished'] = True - campaign.save(update_fields=['recipient_count', 'metadata']) + campaign.save() recipients_creation_finished_at = timezone.now() recipients_creation_run_time = (recipients_creation_finished_at - recipients_creation_started_at) From bcc09d0d1d2e709dacfb30582340f9b56bb5e338 Mon Sep 17 00:00:00 2001 From: Ostap-Zherebetskyi Date: Thu, 17 Sep 2026 16:36:47 +0300 Subject: [PATCH 16/16] [ENG-12196] Make "excluding unconfirmed" checked by default (#11926) * Add default checked state for exclude unconfirmed accounts checkbox --- admin/templates/notifications/notification_campaing_create.html | 1 + 1 file changed, 1 insertion(+) diff --git a/admin/templates/notifications/notification_campaing_create.html b/admin/templates/notifications/notification_campaing_create.html index 37b8fe01e3b..5ca5e46f03b 100644 --- a/admin/templates/notifications/notification_campaing_create.html +++ b/admin/templates/notifications/notification_campaing_create.html @@ -168,6 +168,7 @@

Recipient Filters

Exclude unconfirmed accounts