🐛(retries) make message send retries more reliable (#386)

Adds a comprehensive test and a fix for failing "send" tasks

src/backend/core/management/commands/retry_send.py

@@ -1,159 +0,0 @@
-"""Management command to retry sending a message to failed/retry recipients."""
-
-import logging
-
-from django.core.management.base import BaseCommand, CommandError
-from django.db import transaction
-
-from core import models
-from core.enums import MessageDeliveryStatusChoices
-from core.mda.tasks import retry_messages_task
-
-logger = logging.getLogger(__name__)
-
-
-class Command(BaseCommand):
-    """Management command to retry sending message(s) to recipients with retry status."""
-
-    help = "Retry sending message(s) to recipients with retry status. Without --force, delegates to celery task (respects retry timing). With --force, processes immediately (ignores retry delays). Specify message_id for single message, or omit for bulk processing."
-
-    def add_arguments(self, parser):
-        """Define optional argument for message ID."""
-        parser.add_argument(
-            "message_id",
-            nargs="?",
-            help="ID of the message to retry sending (if not provided, retry all retryable messages)",
-        )
-        parser.add_argument(
-            "--force-mta-out",
-            action="store_true",
-            help="Force sending through external MTA even for local recipients",
-        )
-        parser.add_argument(
-            "--batch-size",
-            type=int,
-            default=100,
-            help="Number of messages to process in each batch (default: 100)",
-        )
-        parser.add_argument(
-            "--force",
-            action="store_true",
-            help="Force immediate retry by resetting retry_at timestamps (ignores retry delays)",
-        )
-
-    def handle(self, *args, **options):
-        """
-        Retry sending messages to recipients with retry status.
-
-        Without --force: Delegates to celery task (respects retry timing).
-        With --force: Processes immediately (resets retry_at timestamps).
-        """
-        message_id = options.get("message_id")
-        force_mta_out = options.get("force_mta_out", False)
-        batch_size = options.get("batch_size", 100)
-        force = options.get("force", False)
-
-        if force:
-            # Handle force operations: reset timestamps and delegate to celery task
-            if message_id:
-                # Reset timestamps for single message and delegate
-                self._reset_and_delegate_single(message_id, force_mta_out)
-            else:
-                # Reset timestamps for all messages and delegate
-                self._reset_and_delegate_all(force_mta_out, batch_size)
-        else:
-            # Delegate to celery task for non-force operations
-            self._delegate_to_celery_task(message_id, force_mta_out, batch_size)
-
-    def _delegate_to_celery_task(self, message_id, force_mta_out, batch_size):
-        """Delegate retry operations to celery task synchronously and print result."""
-        self.stdout.write("Running retry operations via celery task (synchronously)...")
-
-        result = retry_messages_task.apply(
-            args=(),
-            kwargs={
-                "message_id": message_id,
-                "force_mta_out": force_mta_out,
-                "batch_size": batch_size,
-            },
-        )
-        if result.successful():
-            self.stdout.write(
-                self.style.SUCCESS(f"Task completed successfully: {result.get()}")
-            )
-        else:
-            self.stdout.write(self.style.ERROR(f"Task failed: {result.result}"))
-
-    def _reset_and_delegate_single(self, message_id, force_mta_out):
-        """Reset retry_at timestamps for single message and delegate to celery task."""
-        try:
-            message = models.Message.objects.get(id=message_id)
-        except models.Message.DoesNotExist:
-            raise CommandError(
-                f"Message with ID '{message_id}' does not exist."
-            ) from None
-
-        # Check if message is a draft
-        if message.is_draft:
-            raise CommandError(
-                f"Message '{message_id}' is still a draft and cannot be sent."
-            )
-
-        # Get recipients with retry status
-        retry_recipients = message.recipients.filter(
-            delivery_status__in=[
-                MessageDeliveryStatusChoices.RETRY,
-                # MessageDeliveryStatusChoices.FAILED,
-            ]
-        )
-
-        if not retry_recipients.exists():
-            self.stdout.write(
-                self.style.WARNING(
-                    f"No recipients with retry status found for message '{message_id}'"
-                )
-            )
-            return
-
-        # Reset retry_at timestamps
-        with transaction.atomic():
-            updated_count = retry_recipients.update(retry_at=None)
-            self.stdout.write(
-                f"Reset retry_at timestamp for {updated_count} recipient(s) of message '{message_id}'"
-            )
-
-        # Delegate to celery task
-        self._delegate_to_celery_task(message_id, force_mta_out, 100)
-
-    def _reset_and_delegate_all(self, force_mta_out, batch_size):
-        """Reset retry_at timestamps for all messages and delegate to celery task."""
-        # Find all messages with retryable recipients
-        messages_with_retries = models.Message.objects.filter(
-            is_draft=False,
-            recipients__delivery_status=MessageDeliveryStatusChoices.RETRY,
-        ).distinct()
-
-        total_messages = messages_with_retries.count()
-
-        if total_messages == 0:
-            self.stdout.write(
-                self.style.WARNING("No messages with retryable recipients found")
-            )
-            return
-
-        self.stdout.write(
-            f"Found {total_messages} message(s) with retryable recipients"
-        )
-
-        # Reset retry_at timestamps for all recipients
-        with transaction.atomic():
-            updated_count = models.MessageRecipient.objects.filter(
-                delivery_status=MessageDeliveryStatusChoices.RETRY
-            ).update(retry_at=None)
-
-            self.stdout.write(
-                f"Reset retry_at timestamp for {updated_count} recipient(s) across all messages"
-            )
-
-        # Delegate to celery task
-        self._delegate_to_celery_task(None, force_mta_out, batch_size)

src/backend/core/mda/outbound.py

@@ -232,6 +232,12 @@ def send_message(message: models.Message, force_mta_out: bool = False):
     This part is called asynchronously from the celery worker.
     """
 
+    # Refuse to send messages that are draft or not senders
+    if message.is_draft:
+        raise ValueError("Cannot send a draft message")
+    if not message.is_sender:
+        raise ValueError("Cannot send a message we are not sender of")
+
     # Create a unique lock key for this message to prevent double sends
     lock_key = f"send_message_lock:{message.id}"
     lock_timeout = 1800  # 30 minutes timeout for the lock

src/backend/core/mda/tasks.py

@@ -23,7 +23,6 @@ def send_message_task(self, message_id, force_mta_out=False, must_archive=False)
 
     Args:
         message_id: The ID of the message to send
-        mime_data: The MIME data dictionary
         force_mta_out: Whether to force sending via MTA
 
     Returns:
@@ -149,12 +148,19 @@ def retry_messages_task(self, message_id=None, force_mta_out=False, batch_size=1
         total_messages = 1
     else:
         # Bulk mode - find all messages with retryable recipients that are ready for retry
-        message_filter_q = Q(
-            is_draft=False,
-            recipients__delivery_status=MessageDeliveryStatusChoices.RETRY,
-        ) & (
-            Q(recipients__retry_at__isnull=True)
-            | Q(recipients__retry_at__lte=timezone.now())
+        message_filter_q = (
+            Q(
+                is_draft=False,
+                is_sender=True,
+            )
+            & (
+                Q(recipients__delivery_status=MessageDeliveryStatusChoices.RETRY)
+                | Q(recipients__delivery_status__isnull=True)
+            )
+            & (
+                Q(recipients__retry_at__isnull=True)
+                | Q(recipients__retry_at__lte=timezone.now())
+            )
         )
 
         messages_to_process = list(
@@ -197,10 +203,10 @@ def retry_messages_task(self, message_id=None, force_mta_out=False, batch_size=1
         for message in batch_messages:
             try:
                 # Get recipients with retry status that are ready for retry
-                retry_filter_q = Q(
-                    delivery_status=MessageDeliveryStatusChoices.RETRY
+                retry_filter_q = (
+                    Q(delivery_status=MessageDeliveryStatusChoices.RETRY)
+                    | Q(delivery_status__isnull=True)
                 ) & (Q(retry_at__isnull=True) | Q(retry_at__lte=timezone.now()))
-
                 retry_recipients = message.recipients.filter(retry_filter_q)
 
                 if retry_recipients.exists():

src/backend/core/tests/mda/test_outbound.py

@@ -114,6 +114,7 @@ class TestSendOutboundMessage:
             thread=thread,
             sender=sender_contact,
             is_draft=False,
+            is_sender=True,
             subject="Test Outbound",
         )
         # Create a blob with the raw MIME content
@@ -466,6 +467,7 @@ class TestSendMessageRedisLock(TransactionTestCase):
             thread=self.thread,
             sender=self.sender_contact,
             is_draft=False,
+            is_sender=True,
             subject="Test Lock",
         )
         # Create a blob with the raw MIME content

src/backend/core/tests/mda/test_retry.py

@@ -0,0 +1,479 @@
+"""Tests for the core.mda.tasks retry functionality."""
+# pylint: disable=unused-argument
+
+from unittest.mock import patch
+
+from django.utils import timezone
+
+import pytest
+
+from core import enums, factories, models
+from core.mda import tasks
+
+
+@pytest.mark.django_db
+class TestRetryMessagesTask:
+    """Unit tests for the retry_messages_task function."""
+
+    @pytest.fixture
+    def mailbox_sender(self):
+        """Create a test mailbox sender."""
+        return factories.MailboxFactory()
+
+    @pytest.fixture
+    def thread(self, mailbox_sender):
+        """Create a test thread."""
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox_sender,
+            thread=thread,
+            role=enums.ThreadAccessRoleChoices.EDITOR,
+        )
+        return thread
+
+    @pytest.fixture
+    def message_with_recipients(self, mailbox_sender, thread):
+        """Create a message with recipients in various delivery states."""
+        sender_contact = factories.ContactFactory(mailbox=mailbox_sender)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender_contact,
+            is_draft=False,
+            is_sender=True,
+            subject="Test Retry Message",
+        )
+
+        # Create recipients with different delivery statuses
+        to_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="to@example.com"
+        )
+        cc_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="cc@example.com"
+        )
+        bcc_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="bcc@example.com"
+        )
+
+        # Recipient with RETRY status
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=to_contact,
+            type=models.MessageRecipientTypeChoices.TO,
+            delivery_status=enums.MessageDeliveryStatusChoices.RETRY,
+            retry_at=timezone.now() - timezone.timedelta(minutes=1),  # Ready for retry
+            retry_count=1,
+        )
+
+        # Recipient with null delivery_status (failed mid-route)
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=cc_contact,
+            type=models.MessageRecipientTypeChoices.CC,
+            delivery_status=None,  # This simulates prepare_message() done but no send
+            retry_at=None,
+            retry_count=0,
+        )
+
+        # Recipient with SENT status (should not be retried)
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=bcc_contact,
+            type=models.MessageRecipientTypeChoices.BCC,
+            delivery_status=enums.MessageDeliveryStatusChoices.SENT,
+            delivered_at=timezone.now(),
+        )
+
+        return message
+
+    @pytest.fixture
+    def draft_message(self, mailbox_sender, thread):
+        """Create a draft message (should not be retryable)."""
+        sender_contact = factories.ContactFactory(mailbox=mailbox_sender)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender_contact,
+            is_draft=True,  # Still a draft
+            is_sender=True,
+            subject="Draft Message",
+        )
+        return message
+
+    @patch("core.mda.tasks.send_message")
+    def test_retry_single_message_success(
+        self, mock_send_message, message_with_recipients
+    ):
+        """Test retrying a single message by ID."""
+        message = message_with_recipients
+
+        # Mock successful send
+        mock_send_message.return_value = None
+
+        result = tasks.retry_messages_task.apply(args=[str(message.id)]).get()
+
+        # Verify the result
+        assert result["success"] is True
+        assert result["message_id"] == str(message.id)
+        assert result["success_count"] == 1
+        assert result["error_count"] == 0
+        assert result["processed_messages"] == 1
+
+        # Verify send_message was called
+        mock_send_message.assert_called_once_with(message, force_mta_out=False)
+
+    def test_retry_nonexistent_message(self):
+        """Test retrying a non-existent message."""
+        fake_message_id = "00000000-0000-0000-0000-000000000000"
+
+        result = tasks.retry_messages_task.apply(args=[fake_message_id]).get()
+
+        # Verify the result
+        assert result["success"] is False
+        assert "does not exist" in result["error"]
+
+    def test_retry_draft_message(self, draft_message):
+        """Test retrying a draft message (should fail)."""
+        result = tasks.retry_messages_task.apply(args=[str(draft_message.id)]).get()
+
+        # Verify the result
+        assert result["success"] is False
+        assert "is still a draft" in result["error"]
+
+    @patch("core.mda.tasks.send_message")
+    def test_retry_bulk_mode(self, mock_send_message, message_with_recipients):
+        """Test retrying messages in bulk mode (no message_id specified)."""
+        message = message_with_recipients
+
+        # Mock successful send
+        mock_send_message.return_value = None
+
+        result = tasks.retry_messages_task.apply().get()
+
+        # Verify the result
+        assert result["success"] is True
+        assert result["total_messages"] == 1
+        assert result["success_count"] == 1
+        assert result["error_count"] == 0
+        assert result["processed_messages"] == 1
+
+        # Verify send_message was called
+        mock_send_message.assert_called_once_with(message, force_mta_out=False)
+
+    def test_retry_no_messages_ready(self, mailbox_sender, thread):
+        """Test retry when no messages are ready for retry."""
+        sender_contact = factories.ContactFactory(mailbox=mailbox_sender)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender_contact,
+            is_draft=False,
+            is_sender=True,
+            subject="No Retry Message",
+        )
+
+        # Create recipients that are not ready for retry
+        to_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="to@example.com"
+        )
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=to_contact,
+            type=models.MessageRecipientTypeChoices.TO,
+            delivery_status=enums.MessageDeliveryStatusChoices.RETRY,
+            retry_at=timezone.now() + timezone.timedelta(hours=1),  # Not ready yet
+            retry_count=1,
+        )
+
+        result = tasks.retry_messages_task.apply().get()
+
+        # Verify the result
+        assert result["success"] is True
+        assert result["total_messages"] == 0
+        assert result["processed_messages"] == 0
+        assert result["success_count"] == 0
+        assert result["error_count"] == 0
+        assert "No messages ready for retry" in result["message"]
+
+    @patch("core.mda.tasks.send_message")
+    def test_retry_failed_send_task_mid_route(
+        self, mock_send_message, mailbox_sender, thread
+    ):
+        """Test retry when send_message_task() failed mid-route (null delivery_status)."""
+        sender_contact = factories.ContactFactory(mailbox=mailbox_sender)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender_contact,
+            is_draft=False,
+            is_sender=True,
+            sent_at=timezone.now() - timezone.timedelta(minutes=1),
+            subject="Failed Mid-Route Message",
+        )
+
+        # Create recipients with null delivery_status (simulating prepare_message() done but no send)
+        to_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="to@example.com"
+        )
+        cc_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="cc@example.com"
+        )
+
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=to_contact,
+            type=models.MessageRecipientTypeChoices.TO,
+            delivery_status=None,  # Null status - prepare_message() done but no send
+            retry_at=None,
+            retry_count=0,
+        )
+
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=cc_contact,
+            type=models.MessageRecipientTypeChoices.CC,
+            delivery_status=None,  # Null status - prepare_message() done but no send
+            retry_at=None,
+            retry_count=0,
+        )
+
+        # Mock successful send
+        mock_send_message.return_value = None
+
+        result = tasks.retry_messages_task.apply(args=[str(message.id)]).get()
+
+        # Verify the result
+        assert result["success"] is True
+        assert result["message_id"] == str(message.id)
+        assert result["success_count"] == 1
+        assert result["error_count"] == 0
+
+        # Verify send_message was called
+        mock_send_message.assert_called_once_with(message, force_mta_out=False)
+
+    @patch("core.mda.tasks.send_message")
+    def test_retry_timing_respect(self, mock_send_message, mailbox_sender, thread):
+        """Test that retry respects retry timing (retry_at field)."""
+        sender_contact = factories.ContactFactory(mailbox=mailbox_sender)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender_contact,
+            is_draft=False,
+            is_sender=True,
+            subject="Timing Test Message",
+        )
+
+        # Create recipients with different retry timing
+        ready_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="ready@example.com"
+        )
+        not_ready_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="notready@example.com"
+        )
+
+        # Recipient ready for retry
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=ready_contact,
+            type=models.MessageRecipientTypeChoices.TO,
+            delivery_status=enums.MessageDeliveryStatusChoices.RETRY,
+            retry_at=timezone.now() - timezone.timedelta(minutes=1),  # Ready
+            retry_count=1,
+        )
+
+        # Recipient not ready for retry yet
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=not_ready_contact,
+            type=models.MessageRecipientTypeChoices.CC,
+            delivery_status=enums.MessageDeliveryStatusChoices.RETRY,
+            retry_at=timezone.now() + timezone.timedelta(hours=1),  # Not ready yet
+            retry_count=1,
+        )
+
+        # Mock successful send
+        mock_send_message.return_value = None
+
+        result = tasks.retry_messages_task.apply(args=[str(message.id)]).get()
+
+        # Verify the result - should only process the ready recipient
+        assert result["success"] is True
+        assert result["success_count"] == 1
+
+        # Verify send_message was called
+        mock_send_message.assert_called_once_with(message, force_mta_out=False)
+
+    @patch("core.mda.tasks.send_message")
+    def test_retry_batch_processing(self, mock_send_message, mailbox_sender, thread):
+        """Test retry batch processing functionality."""
+        # Create multiple messages
+        messages = []
+        for i in range(5):
+            sender_contact = factories.ContactFactory(mailbox=mailbox_sender)
+            message = factories.MessageFactory(
+                thread=thread,
+                sender=sender_contact,
+                is_draft=False,
+                is_sender=True,
+                subject=f"Batch Test Message {i}",
+            )
+
+            # Add recipients ready for retry
+            to_contact = factories.ContactFactory(
+                mailbox=mailbox_sender, email=f"to{i}@example.com"
+            )
+            factories.MessageRecipientFactory(
+                message=message,
+                contact=to_contact,
+                type=models.MessageRecipientTypeChoices.TO,
+                delivery_status=enums.MessageDeliveryStatusChoices.RETRY,
+                retry_at=timezone.now() - timezone.timedelta(minutes=1),
+                retry_count=1,
+            )
+            messages.append(message)
+
+        # Mock successful send
+        mock_send_message.return_value = None
+
+        result = tasks.retry_messages_task.apply(
+            kwargs={"batch_size": 2}
+        ).get()  # Process in batches of 2
+
+        # Verify the result
+        assert result["success"] is True
+        assert result["total_messages"] == 5
+        assert result["success_count"] == 5
+        assert result["error_count"] == 0
+        assert result["processed_messages"] == 5
+
+        # Verify send_message was called for each message
+        assert mock_send_message.call_count == 5
+
+    @patch("core.mda.tasks.send_message")
+    def test_retry_mixed_recipient_statuses(
+        self, mock_send_message, mailbox_sender, thread
+    ):
+        """Test retry with recipients in various delivery states."""
+        sender_contact = factories.ContactFactory(mailbox=mailbox_sender)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender_contact,
+            is_draft=False,
+            is_sender=True,
+            subject="Mixed Status Message",
+        )
+
+        # Create recipients with different statuses
+        retry_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="retry@example.com"
+        )
+        null_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="null@example.com"
+        )
+        sent_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="sent@example.com"
+        )
+        failed_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="failed@example.com"
+        )
+
+        # RETRY status - ready for retry
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=retry_contact,
+            type=models.MessageRecipientTypeChoices.TO,
+            delivery_status=enums.MessageDeliveryStatusChoices.RETRY,
+            retry_at=timezone.now() - timezone.timedelta(minutes=1),
+            retry_count=1,
+        )
+
+        # NULL status - failed mid-route
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=null_contact,
+            type=models.MessageRecipientTypeChoices.CC,
+            delivery_status=None,
+            retry_at=None,
+            retry_count=0,
+        )
+
+        # SENT status - should not be retried
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=sent_contact,
+            type=models.MessageRecipientTypeChoices.CC,
+            delivery_status=enums.MessageDeliveryStatusChoices.SENT,
+            delivered_at=timezone.now(),
+        )
+
+        # FAILED status - should not be retried
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=failed_contact,
+            type=models.MessageRecipientTypeChoices.BCC,
+            delivery_status=enums.MessageDeliveryStatusChoices.FAILED,
+            delivery_message="Permanent failure",
+        )
+
+        # Mock successful send
+        mock_send_message.return_value = None
+
+        result = tasks.retry_messages_task.apply(args=[str(message.id)]).get()
+
+        # Verify the result - should process 2 recipients (RETRY and NULL)
+        assert result["success"] is True
+        assert result["message_id"] == str(message.id)
+        assert result["success_count"] == 1  # One message processed successfully
+        assert result["error_count"] == 0
+        assert result["processed_messages"] == 1
+
+        # Verify send_message was called
+        mock_send_message.assert_called_once_with(message, force_mta_out=False)
+
+    @patch("core.mda.tasks.send_message")
+    def test_retry_message_with_no_retryable_recipients(
+        self, mock_send_message, mailbox_sender, thread
+    ):
+        """Test retry when message has no recipients ready for retry."""
+        sender_contact = factories.ContactFactory(mailbox=mailbox_sender)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender_contact,
+            is_draft=False,
+            is_sender=True,
+            subject="No Retryable Recipients Message",
+        )
+
+        # Create recipients that are not retryable
+        sent_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="sent@example.com"
+        )
+        failed_contact = factories.ContactFactory(
+            mailbox=mailbox_sender, email="failed@example.com"
+        )
+
+        # SENT status - should not be retried
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=sent_contact,
+            type=models.MessageRecipientTypeChoices.TO,
+            delivery_status=enums.MessageDeliveryStatusChoices.SENT,
+            delivered_at=timezone.now(),
+        )
+
+        # FAILED status - should not be retried
+        factories.MessageRecipientFactory(
+            message=message,
+            contact=failed_contact,
+            type=models.MessageRecipientTypeChoices.CC,
+            delivery_status=enums.MessageDeliveryStatusChoices.FAILED,
+            delivery_message="Permanent failure",
+        )
+
+        result = tasks.retry_messages_task.apply(args=[str(message.id)]).get()
+
+        # Verify the result - should process the message but not call send_message
+        assert result["success"] is True
+        assert result["message_id"] == str(message.id)
+        assert result["success_count"] == 0  # No recipients to retry
+        assert result["error_count"] == 0
+        assert result["processed_messages"] == 1
+
+        # Verify send_message was NOT called because no recipients were retryable
+        mock_send_message.assert_not_called()