Clean up send error handling; fix stalled send queue.

mailer/engine.py

@@ -12,8 +12,7 @@
 from django.core.mail.message import make_msgid
 
 from mailer.models import (
-    Message, MessageLog, RESULT_SUCCESS, RESULT_FAILURE, get_message_id,
-)
+    Message, MessageLog, RESULT_SUCCESS, RESULT_FAILURE, RESULT_ERROR, get_message_id)
 
 
 # when queue is empty, how long to wait (in seconds) before checking again
@@ -103,7 +102,7 @@ def release_lock(lock):
     logging.debug("released.")
 
 
-def send_all():
+def send_all():  # noqa: C901  # TODO: refactor to reduce complexity
     """
     Send all eligible messages in the queue.
     """
@@ -122,6 +121,7 @@ def send_all():
     start_time = time.time()
 
     deferred = 0
+    errored = 0
     sent = 0
 
     try:
@@ -154,19 +154,35 @@ def send_all():
                     logging.warning("message discarded due to failure in converting from DB. Added on '%s' with priority '%s'" % (message.when_added, message.priority))  # noqa
                 message.delete()
 
-            except (socket_error, smtplib.SMTPSenderRefused,
-                    smtplib.SMTPRecipientsRefused,
-                    smtplib.SMTPDataError,
-                    smtplib.SMTPAuthenticationError) as err:
+            except (smtplib.SMTPConnectError,
+                    smtplib.SMTPServerDisconnected) as err:
+                # Connection-related problem -- try again soon:
+                logging.info("message left queued due to transient failure: %s" % err)
+                MessageLog.objects.log(message, RESULT_FAILURE, log_message=str(err))
+                break  # connectivity issue; resume batch next time
+
+            except (socket_error,  # includes smtplib.SMTPException in python3.4+
+                    # TODO: consider handling socket_error/OSError/IOError as "retry soon"???
+                    smtplib.SMTPException) as err:
+                # Destination/recipient-related problem -- try again later:
                 message.defer()
                 logging.info("message deferred due to failure: %s" % err)
                 MessageLog.objects.log(message, RESULT_FAILURE, log_message=str(err))
                 deferred += 1
                 # Get new connection, it case the connection itself has an error.
                 connection = None
 
+            except Exception as err:
+                # Content-specific (or unknown) problem -- don't retry this message:
+                logging.info("message discarded due to error: %s" % err)
+                MessageLog.objects.log(message, RESULT_ERROR, log_message=str(err))
+                message.delete()
+                errored += 1
+                # Get new connection, it case the connection itself has an error.
+                connection = None
+
             # Check if we reached the limits for the current run
-            if _limits_reached(sent, deferred):
+            if _limits_reached(sent, deferred):  # TODO: add limits for errored???
                 break
 
             _throttle_emails()
@@ -175,7 +191,7 @@ def send_all():
         release_lock(lock)
 
     logging.info("")
-    logging.info("%s sent; %s deferred;" % (sent, deferred))
+    logging.info("%s sent; %s deferred; %s errored" % (sent, deferred, errored))
     logging.info("done in %.2f seconds" % (time.time() - start_time))
 
 

mailer/models.py

@@ -236,11 +236,13 @@ class Meta:
 RESULT_SUCCESS = "1"
 RESULT_DONT_SEND = "2"
 RESULT_FAILURE = "3"
+RESULT_ERROR = "4"
 
 RESULT_CODES = (
     (RESULT_SUCCESS, "success"),
     (RESULT_DONT_SEND, "don't send"),
-    (RESULT_FAILURE, "failure"),
+    (RESULT_FAILURE, "failure (retryable)"),
+    (RESULT_ERROR, "error (non-retryable)"),
     # @@@ other types of failure?
 )
 

mailer/tests.py

@@ -18,8 +18,8 @@
 
 import mailer
 from mailer import engine
-from mailer.models import (PRIORITY_DEFERRED, PRIORITY_HIGH, PRIORITY_LOW,
-                           PRIORITY_MEDIUM, RESULT_FAILURE, RESULT_SUCCESS,
+from mailer.models import (PRIORITY_DEFERRED, PRIORITY_HIGH, PRIORITY_LOW, PRIORITY_MEDIUM,
+                           RESULT_FAILURE, RESULT_ERROR, RESULT_SUCCESS,
                            DontSendEntry, Message, MessageLog, db_to_email,
                            email_to_db, make_message)
 
@@ -105,6 +105,66 @@ def test_retry_deferred(self):
             self.assertEqual(len(mail.outbox), 1)
             self.assertEqual(Message.objects.count(), 0)
 
+    def test_deferrable_send_errors(self):
+        """
+        Test that specific send errors are treated as deferrals
+        """
+        for deferrable_error in [
+            smtplib.SMTPSenderRefused(530, "Unauthorized sender", "from@example.com"),
+            smtplib.SMTPRecipientsRefused({"to@example.com": (422, "Mailbox full")}),
+            smtplib.SMTPDataError(501, "Syntax error"),
+            smtplib.SMTPAuthenticationError(530, "Invalid username/password"),
+        ]:
+            with self.settings(MAILER_EMAIL_BACKEND="mailer.tests.FailingMailerEmailBackend"):
+                with patch("mailer.tests.FailingMailerEmailBackend", side_effect=deferrable_error):
+                    mailer.send_mail("Subject", "Body", "from@example.com", ["to@example.com"])
+                    engine.send_all()
+
+            # Should be deferred and logged
+            self.assertEqual(Message.objects.count(), 1)
+            self.assertEqual(Message.objects.deferred().count(), 1)
+            self.assertEqual(MessageLog.objects.count(), 1)
+
+            Message.objects.all().delete()
+            MessageLog.objects.all().delete()
+
+    def test_transient_send_errors(self):
+        """
+        Test that temporary errors cleanly exit send, leaving queue intact
+        """
+        with self.settings(MAILER_EMAIL_BACKEND="mailer.tests.FailingMailerEmailBackend"):
+            with patch("mailer.tests.FailingMailerEmailBackend",
+                       side_effect=smtplib.SMTPConnectError(421, "Down for maintenance")
+                       ) as send_patch:
+                mailer.send_mail("Subject", "Body", "from@example.com", ["to1@example.com"])
+                mailer.send_mail("Subject", "Body", "from@example.com", ["to2@example.com"])
+                self.assertEqual(Message.objects.count(), 2)
+                engine.send_all()
+                send_patch.assert_called_once()
+
+        # Should both still be pending
+        self.assertEqual(Message.objects.count(), 2)
+
+        # And should have logged the failure
+        self.assertEqual(MessageLog.objects.count(), 1)
+        self.assertEqual(MessageLog.objects.all()[0].result, RESULT_FAILURE)
+
+    def test_unexpected_send_errors(self):
+        """
+        Test that messages with permanent content errors don't stall send queue
+        """
+        with self.settings(MAILER_EMAIL_BACKEND="mailer.tests.FailingMailerEmailBackend"):
+            with patch("mailer.tests.FailingMailerEmailBackend", side_effect=UnicodeDecodeError):
+                mailer.send_mail("Subject", "Body", "from@example.com", ["to@example.com"])
+                engine.send_all()
+
+        # Should not be pending or deferred
+        self.assertEqual(Message.objects.count(), 0)
+
+        # But should have logged it as an error
+        self.assertEqual(MessageLog.objects.count(), 1)
+        self.assertEqual(MessageLog.objects.all()[0].result, RESULT_ERROR)
+
     def test_purge_old_entries(self):
         # Send one successfully
         with self.settings(MAILER_EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend"):