Merge pull request #136 from AustralianBioCommons/fix-robust-email-sending

feat: robust email sending updates

Author: amandazhuyilan

db/models.py

@@ -1,5 +1,5 @@
 import uuid
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timezone
 from logging import getLogger
 from typing import Optional, Self
 
@@ -910,6 +910,7 @@ class EmailNotification(BaseModel, table=True):
     last_error: str | None = Field(default=None, sa_column=Column(String(1024), nullable=True))
     send_after: AwareDatetime | None = Field(default=None, sa_type=DateTime(timezone=True))
     sent_at: AwareDatetime | None = Field(default=None, sa_type=DateTime(timezone=True))
+    last_attempt_at: AwareDatetime | None = Field(default=None, sa_type=DateTime(timezone=True))
     created_at: AwareDatetime = Field(
         default_factory=lambda: datetime.now(timezone.utc),
         sa_type=DateTime(timezone=True),
@@ -920,9 +921,12 @@ class EmailNotification(BaseModel, table=True):
     )
 
     def mark_sending(self) -> None:
+        now = datetime.now(timezone.utc)
+        if self.last_attempt_at is None:
+            self.last_attempt_at = now
         self.status = EmailStatusEnum.SENDING
         self.attempts += 1
-        self.updated_at = datetime.now(timezone.utc)
+        self.updated_at = now
 
     def mark_sent(self) -> None:
         now = datetime.now(timezone.utc)
@@ -931,13 +935,18 @@ def mark_sent(self) -> None:
         self.updated_at = now
         self.last_error = None
 
-    def mark_failed(self, error: str, retry_delay_seconds: int | None = None) -> None:
+    def mark_failed(self, error: str) -> None:
         now = datetime.now(timezone.utc)
         self.status = EmailStatusEnum.FAILED
         self.last_error = error[:1024]
         self.updated_at = now
-        if retry_delay_seconds:
-            self.send_after = now + timedelta(seconds=retry_delay_seconds)
+        self.send_after = None
+
+    def schedule_retry(self, error: str, retry_time: datetime) -> None:
+        self.status = EmailStatusEnum.FAILED
+        self.last_error = error[:1024]
+        self.send_after = retry_time
+        self.updated_at = datetime.now(timezone.utc)
 
 
 class EmailChangeOtp(BaseModel, table=True):

migrations/versions/d6a5578732bc_email_scheduler_robust.py

@@ -0,0 +1,31 @@
+"""email_scheduler_robust
+
+Revision ID: d6a5578732bc
+Revises: 9f2d8c1b5d4e
+Create Date: 2025-11-24 15:37:19.537530
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'd6a5578732bc'
+down_revision: Union[str, None] = '9f2d8c1b5d4e'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('emailnotification', sa.Column('last_attempt_at', sa.DateTime(timezone=True), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('emailnotification', 'last_attempt_at')
+    # ### end Alembic commands ###

scheduled_tasks/email_retry.py

@@ -0,0 +1,60 @@
+import random
+from datetime import datetime, timedelta, timezone
+
+from botocore.exceptions import BotoCoreError, ClientError
+
+from db.models import EmailNotification
+
+EMAIL_QUEUE_BATCH_SIZE = 25
+EMAIL_MAX_ATTEMPTS = 2
+EMAIL_RETRY_DELAY_MIN_SECONDS = 15 * 60
+EMAIL_RETRY_DELAY_MAX_SECONDS = 30 * 60
+EMAIL_RETRY_WINDOW_SECONDS = 60 * 60
+EMAIL_JOB_ID_PREFIX = "email_notification_"
+
+TRANSIENT_CLIENT_ERROR_CODES = {
+    "Throttling",
+    "ThrottlingException",
+    "TooManyRequestsException",
+    "ServiceUnavailable",
+    "InternalFailure",
+    "InternalError",
+    "RequestThrottled",
+}
+
+
+def _ensure_aware(dt: datetime | None) -> datetime | None:
+    if dt is None:
+        return None
+    if dt.tzinfo is None:
+        return dt.replace(tzinfo=timezone.utc)
+    return dt
+
+
+def retry_deadline(notification: EmailNotification) -> datetime | None:
+    first_attempt = _ensure_aware(notification.last_attempt_at)
+    if first_attempt is None:
+        return None
+    return first_attempt + timedelta(seconds=EMAIL_RETRY_WINDOW_SECONDS)
+
+
+def can_schedule_notification(notification: EmailNotification, now: datetime) -> bool:
+    if notification.attempts >= EMAIL_MAX_ATTEMPTS:
+        return False
+    deadline = retry_deadline(notification)
+    return deadline is None or now < deadline
+
+
+def is_retryable_email_error(exc: Exception) -> bool:
+    if isinstance(exc, ClientError):
+        code = exc.response.get("Error", {}).get("Code")
+        return code in TRANSIENT_CLIENT_ERROR_CODES
+    if isinstance(exc, BotoCoreError):
+        return True
+    if isinstance(exc, OSError):
+        return True
+    return False
+
+
+def next_retry_delay_seconds() -> int:
+    return random.randint(EMAIL_RETRY_DELAY_MIN_SECONDS, EMAIL_RETRY_DELAY_MAX_SECONDS)

scheduled_tasks/tasks.py

@@ -1,10 +1,10 @@
 import re
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from uuid import UUID
 
 from loguru import logger
 from pydantic import BaseModel
-from sqlalchemy import or_
+from sqlalchemy import and_, or_
 from sqlmodel import Session, select
 
 from auth.management import get_management_token
@@ -23,6 +23,16 @@
 )
 from db.setup import get_db_session
 from db.types import GROUP_NAMES, ApprovalStatusEnum, EmailStatusEnum, GroupEnum
+from scheduled_tasks.email_retry import (
+    EMAIL_JOB_ID_PREFIX,
+    EMAIL_MAX_ATTEMPTS,
+    EMAIL_QUEUE_BATCH_SIZE,
+    EMAIL_RETRY_WINDOW_SECONDS,
+    can_schedule_notification,
+    is_retryable_email_error,
+    next_retry_delay_seconds,
+    retry_deadline,
+)
 from scheduled_tasks.scheduler import SCHEDULER
 from schemas.auth0 import (
     GROUP_ROLE_PATTERN,
@@ -31,10 +41,6 @@
 )
 from schemas.biocommons import Auth0UserData
 
-EMAIL_QUEUE_BATCH_SIZE = 25
-EMAIL_RETRY_DELAY_SECONDS = 300
-EMAIL_JOB_ID_PREFIX = "email_notification_"
-
 
 def _ensure_user_from_auth0(session: Session, user_data: Auth0UserData) -> tuple[BiocommonsUser, bool, bool]:
     """
@@ -72,7 +78,6 @@ def _get_group_membership_including_deleted(session: Session, user_id: str, grou
 
 async def process_email_queue(
     batch_size: int = EMAIL_QUEUE_BATCH_SIZE,
-    retry_delay_seconds: int = EMAIL_RETRY_DELAY_SECONDS,
 ) -> int:
     """
     Schedule pending email notifications for delivery.
@@ -84,8 +89,12 @@ async def process_email_queue(
         stmt = (
             select(EmailNotification)
             .where(
-                EmailNotification.status.in_(
-                    [EmailStatusEnum.PENDING, EmailStatusEnum.FAILED]
+                or_(
+                    EmailNotification.status == EmailStatusEnum.PENDING,
+                    and_(
+                        EmailNotification.status == EmailStatusEnum.FAILED,
+                        EmailNotification.send_after.is_not(None),
+                    ),
                 ),
                 or_(
                     EmailNotification.send_after.is_(None),
@@ -101,6 +110,15 @@ async def process_email_queue(
             return 0
         scheduled = 0
         for notification in notifications:
+            if not can_schedule_notification(notification, now):
+                logger.info(
+                    "Skipping email %s: retry window exhausted or max attempts reached",
+                    notification.id,
+                )
+                notification.status = EmailStatusEnum.FAILED
+                notification.send_after = None
+                session.add(notification)
+                continue
             notification.mark_sending()
             session.add(notification)
             session.flush()
@@ -110,7 +128,6 @@ async def process_email_queue(
                 args=[notification.id],
                 id=job_id,
                 replace_existing=True,
-                kwargs={"retry_delay_seconds": retry_delay_seconds},
             )
             scheduled += 1
         session.commit()
@@ -122,7 +139,6 @@ async def process_email_queue(
 
 async def send_email_notification(
     notification_id: UUID,
-    retry_delay_seconds: int = EMAIL_RETRY_DELAY_SECONDS,
 ) -> bool:
     """
     Deliver a single queued email notification.
@@ -141,10 +157,26 @@ async def send_email_notification(
                 notification.body_html,
             )
         except Exception as exc:  # noqa: BLE001
-            logger.warning(
-                "Failed to send email %s: %s", notification.id, exc
-            )
-            notification.mark_failed(str(exc), retry_delay_seconds)
+            logger.warning("Failed to send email %s: %s", notification.id, exc)
+            now = datetime.now(timezone.utc)
+            should_retry = is_retryable_email_error(exc)
+            deadline = retry_deadline(notification)
+            if deadline is None:
+                deadline = now + timedelta(seconds=EMAIL_RETRY_WINDOW_SECONDS)
+            attempts_remaining = notification.attempts < EMAIL_MAX_ATTEMPTS
+            if (
+                should_retry
+                and attempts_remaining
+                and now < deadline
+            ):
+                delay_seconds = next_retry_delay_seconds()
+                retry_time = now + timedelta(seconds=delay_seconds)
+                if retry_time <= deadline:
+                    notification.schedule_retry(str(exc), retry_time)
+                    session.add(notification)
+                    session.commit()
+                    return False
+            notification.mark_failed(str(exc))
             session.add(notification)
             session.commit()
             return False