From dfd2668c11ee21fcd1fa2ca6b1e3db705a97abed Mon Sep 17 00:00:00 2001 From: Sylvain Zimmer Date: Fri, 17 Oct 2025 19:19:11 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B(retries)=20make=20message=20send?= =?UTF-8?q?=20retries=20more=20reliable=20(#386)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a comprehensive test and a fix for failing "send" tasks --- .../core/management/commands/retry_send.py | 159 ------ src/backend/core/mda/outbound.py | 6 + src/backend/core/mda/tasks.py | 26 +- src/backend/core/tests/mda/test_outbound.py | 2 + src/backend/core/tests/mda/test_retry.py | 479 ++++++++++++++++++ 5 files changed, 503 insertions(+), 169 deletions(-) delete mode 100644 src/backend/core/management/commands/retry_send.py create mode 100644 src/backend/core/tests/mda/test_retry.py diff --git a/src/backend/core/management/commands/retry_send.py b/src/backend/core/management/commands/retry_send.py deleted file mode 100644 index 1c523c9c..00000000 --- a/src/backend/core/management/commands/retry_send.py +++ /dev/null @@ -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) diff --git a/src/backend/core/mda/outbound.py b/src/backend/core/mda/outbound.py index e10ff338..55d1a63d 100644 --- a/src/backend/core/mda/outbound.py +++ b/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 diff --git a/src/backend/core/mda/tasks.py b/src/backend/core/mda/tasks.py index 6e12f607..4e948aec 100644 --- a/src/backend/core/mda/tasks.py +++ b/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(): diff --git a/src/backend/core/tests/mda/test_outbound.py b/src/backend/core/tests/mda/test_outbound.py index cfb43c12..06caaba6 100644 --- a/src/backend/core/tests/mda/test_outbound.py +++ b/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 diff --git a/src/backend/core/tests/mda/test_retry.py b/src/backend/core/tests/mda/test_retry.py new file mode 100644 index 00000000..d0e3f3de --- /dev/null +++ b/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()