diff --git a/src/backend/core/api/viewsets/draft.py b/src/backend/core/api/viewsets/draft.py
index a2bf2c826..ffb7d2bf1 100644
--- a/src/backend/core/api/viewsets/draft.py
+++ b/src/backend/core/api/viewsets/draft.py
@@ -278,21 +278,31 @@ def put(self, request, message_id: str):
# mailbox can edit (404 otherwise).
sender_mailbox = self._resolve_editable_sender_mailbox(request.user, sender_id)
- # Get the draft message
+ # Lock the message row and re-check ``is_draft`` under it: a
+ # concurrent send finalizes the message (``is_draft=False``) and
+ # hands it to the outbound worker, so an autosave racing it must
+ # either fully commit before the send proceeds, or observe the
+ # finalized state and 404 — never rewrite the recipients of a
+ # message already being delivered. The view runs in a single
+ # transaction (``@transaction.atomic`` above), so the lock is held
+ # until the request commits.
try:
- message = models.Message.objects.select_related("thread", "draft_blob").get(
- id=message_id,
- is_draft=True,
- # Ensure the user has access to this thread
- thread__accesses__mailbox=sender_mailbox,
- thread__accesses__role=enums.ThreadAccessRoleChoices.EDITOR,
+ message = (
+ models.Message.objects.select_for_update(of=("self",))
+ .select_related("thread", "draft_blob")
+ .get(
+ id=message_id,
+ is_draft=True,
+ # Ensure the user has access to this thread
+ thread__accesses__mailbox=sender_mailbox,
+ thread__accesses__role=enums.ThreadAccessRoleChoices.EDITOR,
+ )
)
except models.Message.DoesNotExist as exc:
raise drf.exceptions.NotFound(
"Draft message not found, is not a draft, or access denied."
) from exc
- # Update draft using the new function
updated_message = update_draft(
sender_mailbox, message, request.data, user=request.user
)
diff --git a/src/backend/core/api/viewsets/send.py b/src/backend/core/api/viewsets/send.py
index d8eb1b50d..e0cb9ee8a 100644
--- a/src/backend/core/api/viewsets/send.py
+++ b/src/backend/core/api/viewsets/send.py
@@ -96,43 +96,6 @@ def post(self, request):
except models.Mailbox.DoesNotExist as e:
raise drf_exceptions.NotFound("Sender mailbox not found.") from e
- try:
- message = (
- models.Message.objects.select_related("sender")
- .prefetch_related(
- "thread__accesses", "recipients__contact", "attachments__blob"
- )
- .get(
- id=message_id,
- is_draft=True,
- thread__accesses__mailbox=mailbox_sender,
- )
- )
- except models.Message.DoesNotExist as e:
- raise drf_exceptions.NotFound(
- "Draft message not found or does not belong to the specified sender mailbox."
- ) from e
-
- self.check_object_permissions(request, message)
-
- # The sender mailbox itself must be authorised to send on this thread.
- # ``IsAllowedToAccess`` only proves the user can SEND through *some*
- # mailbox holding EDITOR access to the thread — not necessarily
- # ``mailbox_sender``. Re-check against the specific ``senderId`` so a
- # VIEWER on the sender mailbox cannot send as it by piggy-backing on a
- # SENDER role they hold on a different mailbox sharing the thread.
- can_send_as_sender = models.ThreadAccess.objects.filter(
- thread=message.thread,
- mailbox=mailbox_sender,
- role=enums.ThreadAccessRoleChoices.EDITOR,
- mailbox__accesses__user=request.user,
- mailbox__accesses__role__in=enums.MAILBOX_ROLES_CAN_SEND,
- ).exists()
- if not can_send_as_sender:
- raise drf_exceptions.PermissionDenied(
- "You do not have permission to send as this mailbox."
- )
-
# Pre-generate the Celery task id so we can return it to the caller
# while still deferring the actual dispatch to ``transaction.on_commit``
# below — the broker must never receive a delivery task for a message
@@ -140,6 +103,50 @@ def post(self, request):
task_id = str(uuid.uuid4())
with transaction.atomic():
+ # Fetch under a row lock: the draft PUT takes the same lock
+ # before rewriting the recipients, so the recipient set and
+ # draft state read here (MIME headers, recipient cap) can no
+ # longer change between this read and the commit that hands
+ # the message to the worker. The prefetches run under the
+ # lock too, so they observe the same snapshot.
+ try:
+ message = (
+ models.Message.objects.select_for_update(of=("self",))
+ .select_related("sender")
+ .prefetch_related(
+ "thread__accesses", "recipients__contact", "attachments__blob"
+ )
+ .get(
+ id=message_id,
+ is_draft=True,
+ thread__accesses__mailbox=mailbox_sender,
+ )
+ )
+ except models.Message.DoesNotExist as e:
+ raise drf_exceptions.NotFound(
+ "Draft message not found or does not belong to the specified sender mailbox."
+ ) from e
+
+ self.check_object_permissions(request, message)
+
+ # The sender mailbox itself must be authorised to send on this thread.
+ # ``IsAllowedToAccess`` only proves the user can SEND through *some*
+ # mailbox holding EDITOR access to the thread — not necessarily
+ # ``mailbox_sender``. Re-check against the specific ``senderId`` so a
+ # VIEWER on the sender mailbox cannot send as it by piggy-backing on a
+ # SENDER role they hold on a different mailbox sharing the thread.
+ can_send_as_sender = models.ThreadAccess.objects.filter(
+ thread=message.thread,
+ mailbox=mailbox_sender,
+ role=enums.ThreadAccessRoleChoices.EDITOR,
+ mailbox__accesses__user=request.user,
+ mailbox__accesses__role__in=enums.MAILBOX_ROLES_CAN_SEND,
+ ).exists()
+ if not can_send_as_sender:
+ raise drf_exceptions.PermissionDenied(
+ "You do not have permission to send as this mailbox."
+ )
+
prepared = prepare_outbound_message(
mailbox_sender,
message,
diff --git a/src/backend/core/mda/draft.py b/src/backend/core/mda/draft.py
index a07c7abc8..5fad83cc1 100644
--- a/src/backend/core/mda/draft.py
+++ b/src/backend/core/mda/draft.py
@@ -463,6 +463,16 @@ def update_draft(
drf.exceptions.PermissionDenied: If access denied to thread
"""
+ # Recipients become the outbound envelope the moment the message is
+ # finalized: rewriting them (delete + recreate) on a sent message
+ # resets delivery statuses and makes the retry task deliver the same
+ # email again. Keep the invariant structural, not just in callers'
+ # WHERE clauses.
+ if message.pk and not message.is_draft:
+ raise drf.exceptions.ValidationError(
+ "Cannot update a message that is no longer a draft."
+ )
+
updated_fields = []
thread_updated_fields = []
diff --git a/src/backend/core/mda/outbound.py b/src/backend/core/mda/outbound.py
index 68209fb5a..47eac9bea 100644
--- a/src/backend/core/mda/outbound.py
+++ b/src/backend/core/mda/outbound.py
@@ -7,7 +7,8 @@
from django.conf import settings
from django.core.cache import cache
-from django.db import transaction
+from django.core.exceptions import ValidationError
+from django.db import DatabaseError, transaction
from django.utils import timezone
import rest_framework as drf
@@ -503,6 +504,48 @@ def _finalize_sent_message(
message.thread.update_stats()
+def _save_recipient_status(
+ recipient: models.MessageRecipient, update_fields: list
+) -> None:
+ """Persist a recipient's delivery outcome, tolerating concurrent deletion.
+
+ The row can vanish while the worker is on the wire (a draft rewrite
+ racing the send used to do this); losing one status write must not
+ crash the whole delivery. A plain ``save`` is kept — not a queryset
+ UPDATE — so the post_save signals (thread stats, search reindex)
+ still fire on success. The savepoint keeps the surrounding
+ transaction usable when the row is gone (a 0-row UPDATE marks the
+ atomic block for rollback), and ``ValidationError`` covers the
+ variant where a concurrent rewrite already recreated an identical
+ row (``validate_unique`` collision on the stale instance).
+
+ Only the vanished-row case is absorbed. If the row still exists, the
+ write failed for another reason (deadlock, statement timeout, dropped
+ connection) and is re-raised: silently leaving a delivered recipient
+ at a NULL status would hand it to ``retry_messages_task``, which
+ re-enters ``send_message`` and delivers the same email a second time.
+ """
+ try:
+ with transaction.atomic():
+ recipient.save(update_fields=update_fields)
+ except (DatabaseError, ValidationError):
+ if models.MessageRecipient.objects.filter(pk=recipient.pk).exists():
+ logger.error(
+ "Failed to persist status %s of recipient row %s of message %s",
+ recipient.delivery_status,
+ recipient.pk,
+ recipient.message_id,
+ )
+ raise
+ logger.warning(
+ "Recipient row %s of message %s vanished during delivery; "
+ "status %s not recorded",
+ recipient.pk,
+ recipient.message_id,
+ recipient.delivery_status,
+ )
+
+
def send_message(message: models.Message, force_mta_out: bool = False):
"""Send an existing Message, internally or externally.
@@ -544,8 +587,8 @@ def send_message(message: models.Message, force_mta_out: bool = False):
for recipient in message.recipients.all():
recipient.delivery_status = MessageDeliveryStatusChoices.FAILED
recipient.delivery_message = "Internal error: failed to parse email"
- recipient.save(
- update_fields=["delivery_status", "delivery_message"]
+ _save_recipient_status(
+ recipient, ["delivery_status", "delivery_message"]
)
return
@@ -593,10 +636,11 @@ def _mark_delivered(
status,
json.dumps(error or "nil"),
)
+ recipient = envelope_to[recipient_email]
if delivered:
# TODO also update message.updated_at?
- envelope_to[recipient_email].delivered_at = timezone.now()
- envelope_to[recipient_email].delivery_message = None
+ recipient.delivered_at = timezone.now()
+ recipient.delivery_message = None
# Same-instance delivery gets its own SENT_INTERNAL
# status, distinct from external SENT_EXTERNAL, so the
# internal/external split stays visible in the data and
@@ -609,46 +653,35 @@ def _mark_delivered(
# internal mail (see the note on
# ``MessageDeliveryStatusChoices`` and the
# ``force_mta_out`` guard in ``mda/selfcheck.py``).
- envelope_to[recipient_email].delivery_status = (
+ recipient.delivery_status = (
MessageDeliveryStatusChoices.SENT_INTERNAL
if internal
else MessageDeliveryStatusChoices.SENT_EXTERNAL
)
- envelope_to[recipient_email].save(
- update_fields=[
- "delivered_at",
- "delivery_message",
- "delivery_status",
- ]
- )
- elif retry and envelope_to[recipient_email].retry_count < len(
- RETRY_INTERVALS
- ):
- envelope_to[recipient_email].retry_at = (
- timezone.now()
- + RETRY_INTERVALS[envelope_to[recipient_email].retry_count]
- )
- envelope_to[recipient_email].retry_count += 1
- envelope_to[
- recipient_email
- ].delivery_status = MessageDeliveryStatusChoices.RETRY
- envelope_to[recipient_email].delivery_message = error
- envelope_to[recipient_email].save(
- update_fields=[
- "retry_at",
- "retry_count",
- "delivery_status",
- "delivery_message",
- ]
+ status_fields = [
+ "delivered_at",
+ "delivery_message",
+ "delivery_status",
+ ]
+ elif retry and recipient.retry_count < len(RETRY_INTERVALS):
+ recipient.retry_at = (
+ timezone.now() + RETRY_INTERVALS[recipient.retry_count]
)
+ recipient.retry_count += 1
+ recipient.delivery_status = MessageDeliveryStatusChoices.RETRY
+ recipient.delivery_message = error
+ status_fields = [
+ "retry_at",
+ "retry_count",
+ "delivery_status",
+ "delivery_message",
+ ]
else:
- envelope_to[
- recipient_email
- ].delivery_status = MessageDeliveryStatusChoices.FAILED
- envelope_to[recipient_email].delivery_message = error
- envelope_to[recipient_email].save(
- update_fields=["delivery_status", "delivery_message"]
- )
+ recipient.delivery_status = MessageDeliveryStatusChoices.FAILED
+ recipient.delivery_message = error
+ status_fields = ["delivery_status", "delivery_message"]
+
+ _save_recipient_status(recipient, status_fields)
external_recipients = set()
for recipient_email in envelope_to:
@@ -755,24 +788,19 @@ def _mark_delivered(
sender_domain.name,
)
+ statuses = None
try:
statuses = send_outbound_message(
external_recipients, message, blob_content
)
- for recipient_email, status in statuses.items():
- _mark_delivered(
- recipient_email,
- status["delivered"],
- False,
- status.get("error"),
- status.get("retry", False),
- status.get("smtp_host"),
- status.get("proxy_host"),
- )
except Exception as e: # pylint: disable=broad-exception-caught
logger.error(
"Failed to send outbound message: %s", e, exc_info=True
)
+
+ if statuses is None:
+ # The send itself failed before the MTA reported anything:
+ # nothing was accepted, retry everyone.
for recipient_email in external_recipients:
_mark_delivered(
recipient_email,
@@ -781,6 +809,39 @@ def _mark_delivered(
"Internal error while delivering",
True,
)
+ else:
+ # ``statuses`` is the authoritative MTA outcome. Record
+ # each entry independently: one malformed entry must not
+ # abort the others, and above all must not flip a
+ # recipient the MTA already accepted back to RETRY — the
+ # retry task would send the same email a second time.
+ for recipient_email, status in statuses.items():
+ try:
+ _mark_delivered(
+ recipient_email,
+ status["delivered"],
+ False,
+ status.get("error"),
+ status.get("retry", False),
+ status.get("smtp_host"),
+ status.get("proxy_host"),
+ )
+ except Exception: # pylint: disable=broad-exception-caught
+ logger.exception(
+ "Failed to record delivery status of a "
+ "recipient of message %s",
+ message.id,
+ )
+ # Recipients the MTA never reported on: outcome unknown,
+ # retry them.
+ for recipient_email in external_recipients - set(statuses):
+ _mark_delivered(
+ recipient_email,
+ False,
+ False,
+ "Internal error while delivering",
+ True,
+ )
finally:
# Always release the lock when done
cache.delete(lock_key)
diff --git a/src/backend/core/tests/api/test_draft_send_race_guards.py b/src/backend/core/tests/api/test_draft_send_race_guards.py
new file mode 100644
index 000000000..eba7bd6fb
--- /dev/null
+++ b/src/backend/core/tests/api/test_draft_send_race_guards.py
@@ -0,0 +1,208 @@
+"""Endpoint-level guards of the draft-update/send race.
+
+The MDA-layer invariants are covered in
+``core/tests/mda/test_outbound_recipient_race.py``; these tests pin the
+HTTP contract on top of them:
+
+* the draft PUT scopes its (locked) fetch to ``is_draft=True``, so an
+ autosave landing after the send finalized the message gets a 404
+ instead of rewriting the recipients of a message being delivered;
+* the send endpoint scopes its (locked) fetch the same way, so a message
+ already handed to the outbound worker cannot be sent twice;
+* both fetches actually take the row lock (``FOR UPDATE``) that
+ serializes them against each other.
+"""
+
+# pylint: disable=unused-argument
+
+from unittest.mock import MagicMock, patch
+
+from django.db import connection
+from django.test.utils import CaptureQueriesContext
+from django.urls import reverse
+
+import pytest
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from core import enums, factories
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture(name="user")
+def fixture_user():
+ """A user holding the sender role on the mailbox."""
+ return factories.UserFactory()
+
+
+@pytest.fixture(name="mailbox")
+def fixture_mailbox(user):
+ """A mailbox the user can draft and send from."""
+ mailbox = factories.MailboxFactory()
+ factories.MailboxAccessFactory(
+ mailbox=mailbox, user=user, role=enums.MailboxRoleChoices.SENDER
+ )
+ return mailbox
+
+
+@pytest.fixture(name="thread")
+def fixture_thread(mailbox):
+ """A thread the mailbox can edit."""
+ thread = factories.ThreadFactory()
+ factories.ThreadAccessFactory(
+ mailbox=mailbox,
+ thread=thread,
+ role=enums.ThreadAccessRoleChoices.EDITOR,
+ )
+ return thread
+
+
+@pytest.fixture(name="draft_message")
+def fixture_draft_message(thread, mailbox):
+ """A draft message in the editable thread."""
+ return factories.MessageFactory(
+ thread=thread,
+ sender=factories.ContactFactory(mailbox=mailbox),
+ is_draft=True,
+ subject="Draft under race",
+ )
+
+
+@pytest.fixture(name="finalized_message")
+def fixture_finalized_message(draft_message):
+ """The same message once a send has finalized it."""
+ draft_message.is_draft = False
+ draft_message.is_sender = True
+ draft_message.save()
+ return draft_message
+
+
+@pytest.fixture(name="client")
+def fixture_client(user):
+ """An authenticated API client."""
+ client = APIClient()
+ client.force_authenticate(user=user)
+ return client
+
+
+def locked_message_fetches(queries):
+ """SQL of the captured queries that lock a ``messages_message`` row."""
+ return [
+ q["sql"]
+ for q in queries.captured_queries
+ if "FOR UPDATE" in q["sql"] and '"messages_message"' in q["sql"]
+ ]
+
+
+def test_put_draft_on_finalized_message_returns_404(client, mailbox, finalized_message):
+ """An autosave landing after the send finalized the message gets a 404."""
+ to_contact = factories.ContactFactory(mailbox=mailbox, email="to@example.com")
+ recipient = factories.MessageRecipientFactory(
+ message=finalized_message,
+ contact=to_contact,
+ type=enums.MessageRecipientTypeChoices.TO,
+ )
+
+ url = reverse("draft-message-detail", kwargs={"message_id": finalized_message.id})
+ response = client.put(
+ url,
+ {
+ "senderId": str(mailbox.id),
+ "subject": "Rewritten during delivery",
+ "to": ["other@example.com"],
+ },
+ format="json",
+ )
+
+ assert response.status_code == status.HTTP_404_NOT_FOUND
+ # The envelope of the message being delivered was left untouched.
+ assert list(finalized_message.recipients.all()) == [recipient]
+ finalized_message.refresh_from_db()
+ assert finalized_message.subject == "Draft under race"
+
+
+def test_put_draft_takes_row_lock(client, mailbox, draft_message):
+ """The draft PUT must fetch the message FOR UPDATE.
+
+ The lock is what serializes an autosave against a concurrent send;
+ removing ``select_for_update`` from the view would reopen the race
+ without failing any 404 assertion, hence this SQL-level check.
+ """
+ url = reverse("draft-message-detail", kwargs={"message_id": draft_message.id})
+ with CaptureQueriesContext(connection) as queries:
+ response = client.put(
+ url,
+ {"senderId": str(mailbox.id), "subject": "Still a draft"},
+ format="json",
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ locking_fetches = locked_message_fetches(queries)
+ assert locking_fetches, "draft PUT no longer locks the message row"
+ # The draft-state predicate must live in the locked query itself: an
+ # unlocked ``is_draft`` read followed by a separate locked fetch would
+ # reopen the race between the state check and the lock.
+ assert any('"is_draft"' in sql for sql in locking_fetches), (
+ "the locked draft PUT fetch no longer checks the draft state"
+ )
+
+
+def test_send_finalized_message_returns_404(client, mailbox, finalized_message):
+ """A message already handed to the worker cannot be sent again.
+
+ The delivery mocks pin the actual invariant: a 404 returned *after*
+ having enqueued the outbound task would still double-send.
+ """
+ with (
+ patch("core.api.viewsets.send.prepare_outbound_message") as mock_prepare,
+ patch("core.api.viewsets.send.send_message_task") as mock_task,
+ ):
+ response = client.post(
+ reverse("send-message"),
+ {
+ "messageId": str(finalized_message.id),
+ "senderId": str(mailbox.id),
+ "textBody": "Hello",
+ "htmlBody": "Hello
",
+ },
+ format="json",
+ )
+
+ assert response.status_code == status.HTTP_404_NOT_FOUND
+ mock_prepare.assert_not_called()
+ mock_task.apply_async.assert_not_called()
+
+
+def test_send_takes_row_lock(client, mailbox, draft_message):
+ """The send endpoint must fetch the message FOR UPDATE.
+
+ Same rationale as the draft PUT: the lock pair is the fix under
+ test, and only the captured SQL proves it is still taken.
+ """
+ with (
+ patch("core.api.viewsets.send.prepare_outbound_message") as mock_prepare,
+ patch("core.api.viewsets.send.send_message_task") as mock_task,
+ ):
+ mock_prepare.return_value = True
+ mock_task.apply_async.return_value = MagicMock(id="task-123")
+ with CaptureQueriesContext(connection) as queries:
+ response = client.post(
+ reverse("send-message"),
+ {
+ "messageId": str(draft_message.id),
+ "senderId": str(mailbox.id),
+ "textBody": "Hello",
+ "htmlBody": "Hello
",
+ },
+ format="json",
+ )
+
+ assert response.status_code == status.HTTP_200_OK
+ locking_fetches = locked_message_fetches(queries)
+ assert locking_fetches, "send endpoint no longer locks the message row"
+ # Same rationale as the draft PUT: the state check must be part of
+ # the locked fetch, not a separate unlocked read.
+ assert any('"is_draft"' in sql for sql in locking_fetches), (
+ "the locked send fetch no longer checks the draft state"
+ )
diff --git a/src/backend/core/tests/mda/conftest.py b/src/backend/core/tests/mda/conftest.py
new file mode 100644
index 000000000..4e61a7b7f
--- /dev/null
+++ b/src/backend/core/tests/mda/conftest.py
@@ -0,0 +1,52 @@
+"""Shared fixtures for MDA tests."""
+
+import pytest
+
+from core import enums, factories, models
+
+
+@pytest.fixture(name="relay_settings")
+def fixture_relay_settings(settings):
+ """SMTP relay configuration used by outbound send tests."""
+ settings.MTA_OUT_MODE = "relay"
+ settings.MTA_OUT_RELAY_HOST = "smtp.test:1025"
+ settings.MTA_OUT_RELAY_USERNAME = "smtp_user"
+ settings.MTA_OUT_RELAY_PASSWORD = "smtp_pass"
+ settings.OPENSEARCH_INDEX_THREADS = False
+
+
+@pytest.fixture(name="sendable_message")
+def fixture_sendable_message():
+ """A finalized outbound message with one external TO recipient."""
+ sender_contact = factories.ContactFactory(email="sender@sendtest.com")
+ mailbox = sender_contact.mailbox
+ thread = factories.ThreadFactory()
+ factories.ThreadAccessFactory(
+ mailbox=mailbox,
+ thread=thread,
+ role=enums.ThreadAccessRoleChoices.EDITOR,
+ )
+ message = factories.MessageFactory(
+ thread=thread,
+ sender=sender_contact,
+ is_draft=False,
+ is_sender=True,
+ subject="Race repro",
+ )
+ message.blob = factories.BlobFactory(
+ mailbox=mailbox,
+ content=(
+ b"From: sender@sendtest.com\n"
+ b"To: to@example.com\n"
+ b"Subject: Race repro\n\nBody"
+ ),
+ content_type="message/rfc822",
+ )
+ message.save()
+ to_contact = factories.ContactFactory(mailbox=mailbox, email="to@example.com")
+ factories.MessageRecipientFactory(
+ message=message,
+ contact=to_contact,
+ type=models.MessageRecipientTypeChoices.TO,
+ )
+ return message
diff --git a/src/backend/core/tests/mda/test_outbound_recipient_race.py b/src/backend/core/tests/mda/test_outbound_recipient_race.py
new file mode 100644
index 000000000..edae98724
--- /dev/null
+++ b/src/backend/core/tests/mda/test_outbound_recipient_race.py
@@ -0,0 +1,225 @@
+"""Regression tests: recipient rows vanishing while the worker is on the wire.
+
+A draft-update racing the send used to rewrite the recipients
+(delete + get_or_create → new UUIDs) while the outbound worker held the old
+rows, making the post-SMTP status save crash the whole delivery with
+``DatabaseError: Save with update_fields did not affect any rows.`` (or a
+``validate_unique`` collision once the rewrite had fully committed).
+
+Fixes under test:
+* ``update_draft`` refuses messages that are no longer drafts;
+* the worker tolerates a vanished recipient row (warning, not crash) while
+ still firing the post_save signals — thread stats, search reindex — on
+ the nominal path;
+* the SMTP-failure fallback never flips an already-recorded recipient back
+ to RETRY (which would double-send);
+* only the vanished-row case is absorbed: a status save failing while the
+ row still exists (deadlock, timeout) surfaces as an error instead of
+ being mislabeled "vanished" — the misleading label hid writes whose loss
+ makes the retry task deliver the same email twice.
+"""
+
+from unittest.mock import patch
+
+from django.db import DatabaseError
+
+import pytest
+import rest_framework as drf
+
+from core import enums, factories, models
+from core.mda import outbound
+from core.mda.draft import update_draft
+
+pytestmark = pytest.mark.django_db
+
+
+@patch("core.mda.outbound.send_smtp_mail")
+def test_recipient_deleted_during_smtp_does_not_crash(
+ mock_smtp_send, sendable_message, relay_settings
+):
+ """A recipient deleted mid-SMTP is logged, not fatal."""
+ message = sendable_message
+ original_recipient = message.recipients.get()
+
+ def delete_lands_mid_smtp(*args, **kwargs):
+ models.MessageRecipient.objects.filter(pk=original_recipient.pk).delete()
+ return {"to@example.com": {"delivered": True, "error": None}}
+
+ mock_smtp_send.side_effect = delete_lands_mid_smtp
+
+ with patch.object(outbound.logger, "warning") as mock_warning:
+ outbound.send_message(message)
+
+ assert any(
+ "vanished during delivery" in str(call.args[0])
+ for call in mock_warning.call_args_list
+ )
+
+
+@patch("core.mda.outbound.send_smtp_mail")
+def test_recipient_recreated_during_smtp_does_not_crash(
+ mock_smtp_send, sendable_message, relay_settings
+):
+ """A recipient rewrite (delete + recreate) mid-SMTP is logged, not fatal.
+
+ Simulated with raw ORM calls: the API path can no longer do this
+ (row lock in the draft PUT + ``is_draft`` guard in ``update_draft``).
+ """
+ message = sendable_message
+ original_recipient = message.recipients.get()
+
+ def rewrite_lands_mid_smtp(*args, **kwargs):
+ contact = original_recipient.contact
+ models.MessageRecipient.objects.filter(pk=original_recipient.pk).delete()
+ models.MessageRecipient.objects.create(
+ message=message,
+ contact=contact,
+ type=models.MessageRecipientTypeChoices.TO,
+ )
+ return {"to@example.com": {"delivered": True, "error": None}}
+
+ mock_smtp_send.side_effect = rewrite_lands_mid_smtp
+
+ with patch.object(outbound.logger, "warning") as mock_warning:
+ outbound.send_message(message)
+
+ assert any(
+ "vanished during delivery" in str(call.args[0])
+ for call in mock_warning.call_args_list
+ )
+ recreated = models.MessageRecipient.objects.get(message=message)
+ assert recreated.id != original_recipient.id
+
+
+def test_update_draft_refuses_finalized_message(sendable_message):
+ """The recipient-rewrite invariant is structural, not just in views."""
+ message = sendable_message
+
+ with pytest.raises(drf.exceptions.ValidationError, match="no longer a draft"):
+ update_draft(message.sender.mailbox, message, {"to": ["to@example.com"]})
+
+
+@patch("core.mda.outbound.send_smtp_mail")
+def test_delivery_status_save_still_updates_thread_stats(
+ mock_smtp_send, sendable_message, relay_settings
+):
+ """The nominal status save must keep firing post_save side effects.
+
+ ``has_delivery_pending`` is recomputed from a MessageRecipient
+ post_save signal (batched by ThreadStatsUpdateDeferrer); replacing the
+ save with a queryset UPDATE would leave every sent thread stuck in
+ "pending" state.
+ """
+ message = sendable_message
+ thread = message.thread
+ thread.update_stats()
+ assert thread.has_delivery_pending is True
+
+ mock_smtp_send.return_value = {"to@example.com": {"delivered": True, "error": None}}
+
+ outbound.send_message(message)
+
+ thread.refresh_from_db()
+ assert thread.has_delivery_pending is False
+
+
+@patch("core.mda.outbound.send_smtp_mail")
+def test_malformed_status_entry_does_not_flip_delivered_recipients(
+ mock_smtp_send, sendable_message, relay_settings
+):
+ """One bad MTA status entry must not corrupt the other recipients.
+
+ The malformed entry comes first so a naive sequential loop would abort
+ before recording the delivered one, then re-mark it RETRY — and the
+ retry task would send the same email a second time.
+ """
+ message = sendable_message
+ mailbox = message.sender.mailbox
+ cc_contact = factories.ContactFactory(mailbox=mailbox, email="cc@example.com")
+ factories.MessageRecipientFactory(
+ message=message,
+ contact=cc_contact,
+ type=models.MessageRecipientTypeChoices.CC,
+ )
+
+ mock_smtp_send.return_value = {
+ "cc@example.com": None, # malformed: crashes on subscript
+ "to@example.com": {"delivered": True, "error": None},
+ }
+
+ outbound.send_message(message)
+
+ delivered = message.recipients.get(contact__email="to@example.com")
+ assert delivered.delivery_status == enums.MessageDeliveryStatusChoices.SENT_EXTERNAL
+ # The malformed entry's recipient keeps its unset status: the retry
+ # task will pick it up, which is the correct outcome for an unknown
+ # delivery result.
+ pending = message.recipients.get(contact__email="cc@example.com")
+ assert pending.delivery_status is None
+
+
+@patch("core.mda.outbound.send_smtp_mail")
+def test_smtp_send_failure_marks_all_recipients_for_retry(
+ mock_smtp_send, sendable_message, relay_settings
+):
+ """When the MTA reported nothing, every recipient goes to RETRY."""
+ message = sendable_message
+ mock_smtp_send.side_effect = ConnectionError("relay unreachable")
+
+ outbound.send_message(message)
+
+ recipient = message.recipients.get()
+ assert recipient.delivery_status == enums.MessageDeliveryStatusChoices.RETRY
+
+
+def test_save_recipient_status_reraises_when_row_still_exists(sendable_message):
+ """A DB failure with the row still present must propagate, not be absorbed.
+
+ Absorbing it as "vanished row" leaves ``delivery_status`` NULL: the
+ retry task selects NULL statuses and re-enters ``send_message``, so a
+ recipient the MTA already accepted would receive the email twice —
+ with only a benign-looking warning in the logs.
+ """
+ recipient = sendable_message.recipients.get()
+ recipient.delivery_status = enums.MessageDeliveryStatusChoices.SENT_EXTERNAL
+
+ with (
+ patch.object(
+ models.MessageRecipient, "save", side_effect=DatabaseError("deadlock")
+ ),
+ pytest.raises(DatabaseError),
+ ):
+ outbound._save_recipient_status(recipient, ["delivery_status"])
+
+
+@patch("core.mda.outbound.send_smtp_mail")
+def test_transient_db_error_is_not_mistaken_for_vanished_row(
+ mock_smtp_send, sendable_message, relay_settings
+):
+ """End to end, a deadlock on the status save surfaces as an error.
+
+ The per-recipient handler in ``send_message`` keeps the delivery loop
+ alive, but the failure must be logged as an error ("Failed to record
+ delivery status"), never as the benign "vanished during delivery"
+ warning.
+ """
+ message = sendable_message
+ mock_smtp_send.return_value = {"to@example.com": {"delivered": True, "error": None}}
+
+ with (
+ patch.object(
+ models.MessageRecipient, "save", side_effect=DatabaseError("deadlock")
+ ),
+ patch.object(outbound.logger, "exception") as mock_exception,
+ patch.object(outbound.logger, "warning") as mock_warning,
+ ):
+ outbound.send_message(message)
+
+ assert any(
+ "Failed to record delivery status" in str(call.args[0])
+ for call in mock_exception.call_args_list
+ )
+ assert not any(
+ "vanished during delivery" in str(call.args[0])
+ for call in mock_warning.call_args_list
+ )
diff --git a/src/frontend/src/features/forms/components/message-form/index.tsx b/src/frontend/src/features/forms/components/message-form/index.tsx
index 72738e7ee..c6588029d 100644
--- a/src/frontend/src/features/forms/components/message-form/index.tsx
+++ b/src/frontend/src/features/forms/components/message-form/index.tsx
@@ -128,6 +128,12 @@ export const MessageForm = forwardRef(({
return localStorage.getItem(PREFER_SEND_MODE_KEY) as PreferSendMode ?? PreferSendMode.SEND;
});
const saveDraftPromiseRef = useRef | null>(null);
+ // Blocks any non-forced draft save while a send is in flight: a draft PUT
+ // racing the send rewrites the recipients of a message already being
+ // delivered server-side. A ref (not state) so the guard is visible
+ // synchronously from the autosave timer and saveDraftInner, and survives
+ // the re-arms done by saveDraftInner's finally and the [draft] effect.
+ const isSendingRef = useRef(false);
const [isSubmitting, setIsSubmitting] = useState(false);
const [currentTime, setCurrentTime] = useState(new Date());
const autoSaveTimerRef = useRef(null);
@@ -470,6 +476,8 @@ export const MessageForm = forwardRef(({
* Auto-save draft every 30 seconds
*/
const startAutoSave = () => {
+ // No autosave while a send is in flight (see isSendingRef).
+ if (isSendingRef.current) return;
// Clear existing timer
if (autoSaveTimerRef.current) {
clearInterval(autoSaveTimerRef.current);
@@ -522,6 +530,7 @@ export const MessageForm = forwardRef(({
* Returns the draft id on success.
*/
const saveDraftInner = async (force = false): Promise => {
+ if (isSendingRef.current && !force) return draft?.id;
if (saveDraftPromiseRef.current) return saveDraftPromiseRef.current;
const data = form.getValues();
@@ -626,6 +635,17 @@ export const MessageForm = forwardRef(({
if (!canSendMessages || !composerRef.current) return;
setIsSubmitting(true);
+ // Block autosave for the whole send: the ref also guards
+ // startAutoSave against the re-arms performed by saveDraftInner's
+ // finally and the [draft] effect during the awaits below.
+ isSendingRef.current = true;
+ stopAutoSave();
+
+ const abortSend = () => {
+ isSendingRef.current = false;
+ setIsSubmitting(false);
+ if (draftRef.current) startAutoSave();
+ };
try {
// Wait for any in-progress draft save to complete
@@ -634,7 +654,7 @@ export const MessageForm = forwardRef(({
// Ensure a draft exists before sending (creates one on-the-fly if needed)
const messageId = draft?.id ?? await ensureDraft();
if (!messageId) {
- setIsSubmitting(false);
+ abortSend();
return;
}
@@ -643,7 +663,9 @@ export const MessageForm = forwardRef(({
// ![]()
DOM elements and triggers unwanted blob download requests).
const { htmlBody, textBody } = await composerRef.current.exportContent();
- stopAutoSave();
+ // A blur-triggered save may have started while we awaited above;
+ // let it settle so no draft PUT is in flight when the send fires.
+ await saveDraftPromiseRef.current;
// Send (and "send and archive") moves the thread out of the drafts
// filter — and possibly out of the inbox when archived. Drop the
// pin upfront so the eventual refetch is authoritative.
@@ -657,10 +679,17 @@ export const MessageForm = forwardRef(({
textBody: MailHelper.attachDriveAttachmentsToTextBody(textBody, data.driveAttachments),
archive,
}
+ }, {
+ onSuccess: () => {
+ isSendingRef.current = false;
+ },
+ // The draft stays open when the send fails server-side;
+ // resume the autosave loop blocked for the send.
+ onError: abortSend,
});
} catch (error) {
console.warn("Error in handleSubmit:", error);
- setIsSubmitting(false);
+ abortSend();
}
};