Merge pull request #84 from thinkt4nk/chore/send-messages-failure-retry

send messages retry failure handling

Author: thinkt4nk

entity_emailer/interface.py

@@ -1,15 +1,16 @@
+from datetime import datetime
 import json
 import sys
 import traceback
 
-from datetime import datetime
-from django.db import transaction
+from ambition_utils.transaction import durable
+from django.conf import settings
 from django.core import mail
+from django.db import transaction
 from entity_event import context_loader
 
 from entity_emailer.models import Email
 from entity_emailer.signals import pre_send, email_exception
-
 from entity_emailer.utils import get_medium, get_from_email_address, get_subscribed_email_addresses, \
     create_email_message, extract_email_subject_from_html_content
 
@@ -20,6 +21,7 @@ class EntityEmailerInterface(object):
     """
 
     @classmethod
+    @durable
     def send_unsent_scheduled_emails(cls):
         """
         Send out any scheduled emails that are unsent
@@ -30,7 +32,8 @@ def send_unsent_scheduled_emails(cls):
         email_medium = get_medium()
         to_send = Email.objects.filter(
             scheduled__lte=current_time,
-            sent__isnull=True
+            sent__isnull=True,
+            num_tries__lt=settings.ENTITY_EMAILER_MAX_SEND_MESSAGE_TRIES
         ).select_related(
             'event__source'
         ).prefetch_related(
@@ -95,12 +98,13 @@ def send_unsent_scheduled_emails(cls):
                 try:
                     # Send mail
                     connection.send_messages([email.get('message')])
+                    # Update the email model sent value
+                    email_model = email.get('model')
+                    email_model.sent = current_time
+                    email_model.save(update_fields=['sent'])
                 except Exception as e:
                     cls.save_email_exception(email.get('model'), e)
 
-        # Update the emails as sent
-        to_send.update(sent=current_time)
-
     @staticmethod
     def convert_events_to_emails():
         """
@@ -163,7 +167,8 @@ def save_email_exception(cls, email, e):
             exception_message += ': {}'.format(json.dumps(e.to_dict()))
 
         email.exception = exception_message
-        email.save(update_fields=['exception'])
+        email.num_tries += 1
+        email.save(update_fields=['exception', 'num_tries'])
 
         # Fire the email exception event
         email_exception.send(

entity_emailer/migrations/0004_email_num_tries.py

@@ -0,0 +1,18 @@
+# Generated by Django 2.2.13 on 2021-03-02 21:05
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('entity_emailer', '0003_email_exception'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='email',
+            name='num_tries',
+            field=models.IntegerField(default=0),
+        ),
+    ]

entity_emailer/models.py

@@ -105,6 +105,9 @@ class Email(models.Model):
     # The time that the email was actually sent, or None if the email is still waiting to be sent
     sent = models.DateTimeField(null=True, default=None)
 
+    # Number of attempts to send this email message, used only in retry logic
+    num_tries = models.IntegerField(default=0)
+
     # Any exception that occurred when attempting to send the email last
     exception = models.TextField(default=None, null=True)
 

entity_emailer/tests/interface_tests.py

@@ -5,6 +5,7 @@
 from django.core import mail
 from django.core.mail import EmailMultiAlternatives
 from django.core.management import call_command
+from django.db import utils
 from django.test import TestCase, SimpleTestCase
 from django.test.utils import override_settings
 from django_dynamic_fixture import G
@@ -378,41 +379,13 @@ def test_multiple_events_only_following_true(self):
         self.assertEquals(email.subject, '')
         self.assertEquals(email.scheduled, datetime(2013, 1, 2))
 
-    @freeze_time('2013-1-2')
-    def test_bulk_multiple_events_only_following_true(self):
-        """
-        Handles bulk creating events and tests the unique constraint of the duplicated subscription which would cause
-        a bulk create error if it wasn't handled
-        """
-        source = G(Source)
-        e = G(Entity)
-        other_e = G(Entity)
-
-        G(Subscription, entity=e, source=source, medium=self.email_medium, only_following=True)
-        G(Subscription, entity=e, source=source, medium=self.email_medium, only_following=True)
-        G(Subscription, entity=other_e, source=source, medium=self.email_medium, only_following=True)
-        email_context = {
-            'entity_emailer_template': 'template',
-            'entity_emailer_subject': 'hi',
-        }
-        G(Event, source=source, context=email_context)
-        event = G(Event, source=source, context=email_context)
-        G(EventActor, event=event, entity=e)
-
-        EntityEmailerInterface.bulk_convert_events_to_emails()
-
-        email = Email.objects.get()
-        self.assertEquals(set(email.recipients.all()), set([e]))
-        self.assertEquals(email.event.context, email_context)
-        self.assertEquals(email.subject, '')
-        self.assertEquals(email.scheduled, datetime(2013, 1, 2))
-
 
 @freeze_time('2014-01-05')
 class SendUnsentScheduledEmailsTest(TestCase):
     def setUp(self):
         G(Medium, name='email')
 
+    @override_settings(DISABLE_DURABILITY_CHECKING=True)
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
     @patch.object(Event, 'render', spec_set=True)
     def test_sends_all_scheduled_emails_no_email_addresses(self, render_mock, address_mock):
@@ -423,6 +396,7 @@ def test_sends_all_scheduled_emails_no_email_addresses(self, render_mock, addres
         EntityEmailerInterface.send_unsent_scheduled_emails()
         self.assertEqual(len(mail.outbox), 0)
 
+    @override_settings(DISABLE_DURABILITY_CHECKING=True)
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
     @patch.object(Event, 'render', spec_set=True)
     def test_sends_all_scheduled_emails(self, render_mock, address_mock):
@@ -436,6 +410,7 @@ def test_sends_all_scheduled_emails(self, render_mock, address_mock):
 
             self.assertEqual(2, mock_connection.return_value.__enter__.return_value.send_messages.call_count)
 
+    @override_settings(DISABLE_DURABILITY_CHECKING=True)
     @patch('entity_emailer.interface.pre_send')
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
     @patch.object(Event, 'render', spec_set=True)
@@ -468,6 +443,7 @@ def test_send_signals(self, render_mock, address_mock, mock_pre_send):
             })
             self.assertIsInstance(kwargs['message'], EmailMultiAlternatives)
 
+    @override_settings(DISABLE_DURABILITY_CHECKING=True)
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
     @patch.object(Event, 'render', spec_set=True)
     def test_sends_email_with_specified_from_address(self, render_mock, address_mock):
@@ -482,6 +458,7 @@ def test_sends_email_with_specified_from_address(self, render_mock, address_mock
             args = mock_connection.return_value.__enter__.return_value.send_messages.call_args
             self.assertEqual(args[0][0][0].from_email, from_address)
 
+    @override_settings(DISABLE_DURABILITY_CHECKING=True)
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
     @patch.object(Event, 'render', spec_set=True)
     def test_sends_no_future_emails(self, render_mock, address_mock):
@@ -491,6 +468,7 @@ def test_sends_no_future_emails(self, render_mock, address_mock):
         EntityEmailerInterface.send_unsent_scheduled_emails()
         self.assertEqual(len(mail.outbox), 0)
 
+    @override_settings(DISABLE_DURABILITY_CHECKING=True)
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
     @patch.object(Event, 'render', spec_set=True)
     def test_sends_no_sent_emails(self, render_mock, address_mock):
@@ -500,6 +478,7 @@ def test_sends_no_sent_emails(self, render_mock, address_mock):
         EntityEmailerInterface.send_unsent_scheduled_emails()
         self.assertEqual(len(mail.outbox), 0)
 
+    @override_settings(DISABLE_DURABILITY_CHECKING=True)
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
     @patch.object(Event, 'render', spec_set=True)
     def test_updates_times(self, render_mock, address_mock):
@@ -510,6 +489,7 @@ def test_updates_times(self, render_mock, address_mock):
         sent_email = Email.objects.filter(sent__isnull=False)
         self.assertEqual(sent_email.count(), 1)
 
+    @override_settings(DISABLE_DURABILITY_CHECKING=True)
     @patch('entity_emailer.interface.email_exception')
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
     @patch.object(Event, 'render', spec_set=True)
@@ -535,19 +515,26 @@ def test_exceptions(self, render_mock, address_mock, mock_email_exception):
         with patch(settings.EMAIL_BACKEND) as mock_connection:
             EntityEmailerInterface.send_unsent_scheduled_emails()
 
-            # Assert that both emails were marked as sent
-            self.assertEqual(Email.objects.filter(sent__isnull=False).count(), 2)
+            # Assert that only one email was marked as sent
+            self.assertEqual(Email.objects.filter(sent__isnull=False).count(), 1)
 
             # Assert that only one email is actually sent through backend
             self.assertEquals(1, mock_connection.call_count)
-            # Assert that one email raised an exception
-            exception_email = Email.objects.get(sent__isnull=False, exception__isnull=False)
+            # Assert that one email raised an exception and that the tries count was updated
+            exception_email = Email.objects.get(exception__isnull=False, num_tries=1)
             self.assertIsNotNone(exception_email)
             self.assertTrue('Exception: test' in exception_email.exception)
 
+    def test_send_unsent_emails_is_durable(self):
+        """
+        Verify that the process to send unsent emails is durable and will not be rolled-back on exit from any caller
+        """
+        self.assertRaises(utils.ProgrammingError, EntityEmailerInterface.send_unsent_scheduled_emails)
+
+    @override_settings(DISABLE_DURABILITY_CHECKING=True, ENTITY_EMAILER_MAX_SEND_MESSAGE_TRIES=2)
     @patch.object(Event, 'render', spec_set=True)
     @patch('entity_emailer.interface.get_subscribed_email_addresses')
-    def test_send_exceptions(self, mock_get_subscribed_addresses, mock_render):
+    def test_send_exceptions_and_retry(self, mock_get_subscribed_addresses, mock_render):
         """
         Verifies that when a single email raises an exception from within the backend, the batch is still
         updated as sent and the failed email is saved with the exception
@@ -574,16 +561,44 @@ def to_dict(self):
 
             EntityEmailerInterface.send_unsent_scheduled_emails()
 
-        # Verify that both emails are marked as sent
-        self.assertEquals(2, Email.objects.filter(sent__isnull=False).count())
+        # Verify that only one email is marked as sent
+        self.assertEquals(1, Email.objects.filter(sent__isnull=False).count())
         # Verify that the failed email was saved with the exception
-        actual_failed_email = Email.objects.get(sent__isnull=False, exception__isnull=False)
+        actual_failed_email = Email.objects.get(exception__isnull=False, num_tries=1)
         self.assertEquals(failed_email.id, actual_failed_email.id)
         self.assertEquals(
             'test: {}'.format(json.dumps({'message': 'test'})),
             actual_failed_email.exception
         )
 
+        # Verify that a subsequent attempt to send unscheduled emails will retry the failed email
+        with patch(settings.EMAIL_BACKEND) as mock_connection:
+            # Mock side effect for sending email
+            mock_connection.return_value.__enter__.return_value.send_messages.side_effect = (
+                TestEmailSendMessageException('test')
+            )
+
+            EntityEmailerInterface.send_unsent_scheduled_emails()
+
+            # Verify only called once, with the failed email
+            self.assertEquals(1, mock_connection.return_value.__enter__.return_value.send_messages.call_count)
+
+        # Verify num tries updated
+        actual_failed_email = Email.objects.get(exception__isnull=False, num_tries=2)
+        self.assertEquals(failed_email.id, actual_failed_email.id)
+
+        # Verify that a subsequent attempt to send unscheduled emails will find no emails to send
+        with patch(settings.EMAIL_BACKEND) as mock_connection:
+
+            EntityEmailerInterface.send_unsent_scheduled_emails()
+
+            # Verify only called once, with the failed email
+            self.assertEquals(0, mock_connection.return_value.__enter__.return_value.send_messages.call_count)
+
+        # Verify num tries not updated
+        actual_failed_email = Email.objects.get(exception__isnull=False, num_tries=2)
+        self.assertEquals(failed_email.id, actual_failed_email.id)
+
 
 class CreateEmailObjectTest(TestCase):
     def test_no_html(self):

requirements/requirements.txt

@@ -4,5 +4,6 @@ django-db-mutex>=2.0.0
 django-entity>=5.0.0
 django-entity-event>=2.0.0
 ambition-django-uuidfield>=0.5.0
+ambition-utils>=2.2.0
 
 beautifulsoup4>=4.3.2

settings.py

@@ -35,6 +35,7 @@ def configure_settings():
             raise RuntimeError('Unsupported test DB {0}'.format(test_db))
 
         settings.configure(
+            ENTITY_EMAILER_MAX_SEND_MESSAGE_TRIES=3,
             DATABASES={
                 'default': db_config,
             },