fix(email): enforce shared send throttling

backend/alembic/versions/0018_email_send_rate_buckets.py

@@ -0,0 +1,44 @@
+"""add shared email send rate-limit buckets
+
+Revision ID: 0018_email_send_rate_buckets
+Revises: 0017_merge_newsdom_carddav_heads
+Create Date: 2026-08-19 00:00:00.000000
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0018_email_send_rate_buckets"
+down_revision = "0017_merge_newsdom_carddav_heads"
+branch_labels = None
+depends_on = None
+
+_BUCKET_TABLE = "email_send_rate_buckets"
+_EXPIRY_INDEX = "ix_email_send_rate_buckets_expires_at"
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+    if not inspector.has_table(_BUCKET_TABLE):
+        op.create_table(
+            _BUCKET_TABLE,
+            sa.Column("bucket_scope_hash", sa.String(length=64), nullable=False),
+            sa.Column("window_started_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("attempt_count", sa.Integer(), nullable=False),
+            sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("bucket_scope_hash"),
+        )
+    op.create_index(
+        _EXPIRY_INDEX,
+        _BUCKET_TABLE,
+        ["expires_at"],
+        if_not_exists=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(_EXPIRY_INDEX, table_name=_BUCKET_TABLE, if_exists=True)
+    op.drop_table(_BUCKET_TABLE, if_exists=True)

backend/api/emails.py

@@ -1,13 +1,11 @@
 from collections import defaultdict
-from threading import Lock
 from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import func, or_, select
 from db.session import get_db
 from db.models import Email
 from pydantic import BaseModel, EmailStr, Field, field_validator
 import datetime
-import time
 from typing import Literal
 from services.email_client import (
     EmailMessageParams,
@@ -39,6 +37,10 @@
     canonical_email_import_upload_filename,
     import_email_uploads,
 )
+from services.email_send_rate_limiter import (
+    EmailSendRateLimitUnavailable,
+    enforce_send_email_rate_limit,
+)
 from services.llm_provider_selection import resolve_runtime_llm_provider
 from services.text_safety import strip_html_markup
 import logging
@@ -50,34 +52,6 @@
 
 router = APIRouter(prefix="/api/emails")
 
-_SEND_EMAIL_RATE_LIMIT_MAX_ATTEMPTS = 10
-_SEND_EMAIL_RATE_LIMIT_WINDOW_SECONDS = 60.0
-_email_send_attempts_by_scope: dict[tuple[str | None, str], list[float]] = {}
-_email_send_rate_limit_lock = Lock()
-
-
-def _enforce_send_email_rate_limit(auth_context: AuthContext) -> None:
-    now = time.monotonic()
-    cutoff = now - _SEND_EMAIL_RATE_LIMIT_WINDOW_SECONDS
-    key = (auth_context.organization_id, auth_context.user_id)
-
-    # ponytail: process-local throttle; move to Redis when multi-worker send volume matters.
-    with _email_send_rate_limit_lock:
-        attempts = [
-            attempt
-            for attempt in _email_send_attempts_by_scope.get(key, [])
-            if attempt > cutoff
-        ]
-        if len(attempts) >= _SEND_EMAIL_RATE_LIMIT_MAX_ATTEMPTS:
-            _email_send_attempts_by_scope[key] = attempts
-            raise HTTPException(
-                status_code=429,
-                detail="Email send rate limit exceeded",
-            )
-        attempts.append(now)
-        _email_send_attempts_by_scope[key] = attempts
-
-
 def canonical_thread_key(email: Email) -> str:
     return (
         normalize_message_id(email.thread_id)
@@ -756,7 +730,18 @@ async def send_email_endpoint(
             in_reply_to=request.in_reply_to,
             references=request.references,
         )
-        _enforce_send_email_rate_limit(auth_context)
+        try:
+            rate_limit_decision = await enforce_send_email_rate_limit(db, auth_context)
+        except EmailSendRateLimitUnavailable as exc:
+            raise HTTPException(
+                status_code=503,
+                detail="Email send rate limiter unavailable",
+            ) from exc
+        if not rate_limit_decision.allowed:
+            raise HTTPException(
+                status_code=429,
+                detail="Email send rate limit exceeded",
+            )
         smtp_config = SmtpConfig(
             smtp_server=smtp_server,
             smtp_port=smtp_port,

backend/db/models.py

@@ -823,6 +823,33 @@ def owner_filters(cls, user_id: str, organization_id: str | None):
     )
 
 
+class EmailSendRateBucket(Base):
+    __tablename__ = "email_send_rate_buckets"
+
+    bucket_scope_hash: Mapped[str] = mapped_column(String(64), primary_key=True)
+    window_started_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False
+    )
+    attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    expires_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False
+    )
+    created_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.datetime.now(datetime.timezone.utc),
+        nullable=False,
+    )
+    updated_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.datetime.now(datetime.timezone.utc),
+        nullable=False,
+    )
+
+    __table_args__ = (
+        Index("ix_email_send_rate_buckets_expires_at", "expires_at"),
+    )
+
+
 class TicketTask(Base):
     __tablename__ = "ticket_tasks"
 

backend/services/email_send_rate_limiter.py

@@ -0,0 +1,179 @@
+"""Shared, fail-closed email send throttling."""
+
+from __future__ import annotations
+
+import datetime
+import hashlib
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Literal
+
+from sqlalchemy import bindparam, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.models import EmailSendRateBucket, SecurityAuditEvent
+
+if TYPE_CHECKING:
+    from api.auth import AuthContext
+
+logger = logging.getLogger(__name__)
+
+SEND_RATE_LIMIT_MAX_ATTEMPTS = 10
+SEND_RATE_LIMIT_WINDOW_SECONDS = 60
+SEND_RATE_LIMIT_NAMESPACE = "naruon-email-send-rate-limit"
+
+
+class EmailSendRateLimitUnavailable(RuntimeError):
+    """The shared rate-limit state cannot provide a trustworthy decision."""
+
+
+@dataclass(frozen=True)
+class EmailSendRateLimitDecision:
+    """A non-sensitive rate-limit decision returned to the send endpoint."""
+
+    allowed: bool
+    reason: Literal["allowed", "quota_exhausted"]
+
+
+def rate_limit_scope_hash(user_id: str, organization_id: str | None) -> str:
+    """Return a non-reversible identifier for one authorized send scope."""
+    organization_scope = organization_id or "<personal>"
+    value = f"{SEND_RATE_LIMIT_NAMESPACE}\0{organization_scope}\0{user_id}"
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _lock_key(scope_hash: str) -> int:
+    return int.from_bytes(bytes.fromhex(scope_hash[:16]), byteorder="big", signed=True)
+
+
+def _utc_now() -> datetime.datetime:
+    return datetime.datetime.now(datetime.timezone.utc)
+
+
+def _next_bucket_state(
+    bucket: EmailSendRateBucket | None,
+    now: datetime.datetime,
+) -> tuple[bool, int, datetime.datetime, datetime.datetime]:
+    """Calculate the next state without storing individual attempts."""
+    expires_at = now + datetime.timedelta(seconds=SEND_RATE_LIMIT_WINDOW_SECONDS)
+    if bucket is None or now >= bucket.expires_at:
+        return True, 1, now, expires_at
+    if bucket.attempt_count >= SEND_RATE_LIMIT_MAX_ATTEMPTS:
+        return False, bucket.attempt_count, bucket.window_started_at, bucket.expires_at
+    return (
+        True,
+        bucket.attempt_count + 1,
+        bucket.window_started_at,
+        bucket.expires_at,
+    )
+
+
+def _session_uses_postgresql(session: AsyncSession) -> bool:
+    try:
+        bind = session.get_bind()
+    except Exception:
+        return False
+    return getattr(getattr(bind, "dialect", None), "name", None) == "postgresql"
+
+
+def _audit_event(
+    auth_context: AuthContext,
+    *,
+    scope_hash: str,
+    decision: EmailSendRateLimitDecision,
+) -> SecurityAuditEvent:
+    return SecurityAuditEvent(
+        actor_user_id=auth_context.user_id,
+        actor_role=auth_context.role,
+        organization_id=auth_context.organization_id,
+        workspace_id=auth_context.workspace_id,
+        event_action=f"email_send_rate_limit.{decision.reason}",
+        resource_type="email_send_rate_limit",
+        resource_uid=f"email_send_scope:{scope_hash}",
+        evidence_source="services.email_send_rate_limiter",
+        detail_text=(
+            f"decision={decision.reason};"
+            f"window_seconds={SEND_RATE_LIMIT_WINDOW_SECONDS};"
+            f"max_attempts={SEND_RATE_LIMIT_MAX_ATTEMPTS}"
+        ),
+    )
+
+
+async def enforce_send_email_rate_limit(
+    session: AsyncSession,
+    auth_context: AuthContext,
+    *,
+    now: datetime.datetime | None = None,
+) -> EmailSendRateLimitDecision:
+    """Atomically reserve one send attempt in the shared PostgreSQL bucket.
+
+    PostgreSQL transaction advisory locking serializes the read/update/insert
+    decision across workers. A single row per scope plus ``expires_at`` keeps
+    attempts bounded; unavailable shared state fails closed instead of using a
+    process-local fallback.
+    """
+    if not _session_uses_postgresql(session):
+        raise EmailSendRateLimitUnavailable
+
+    observed_at = now or _utc_now()
+    scope_hash = rate_limit_scope_hash(
+        auth_context.user_id, auth_context.organization_id
+    )
+    try:
+        await session.execute(
+            select(func.pg_advisory_xact_lock(bindparam("lock_key"))),
+            {"lock_key": _lock_key(scope_hash)},
+        )
+        result = await session.execute(
+            select(EmailSendRateBucket)
+            .where(EmailSendRateBucket.bucket_scope_hash == scope_hash)
+            .with_for_update()
+        )
+        bucket = result.scalar_one_or_none()
+        allowed, attempt_count, window_started_at, expires_at = _next_bucket_state(
+            bucket, observed_at
+        )
+        if bucket is None:
+            session.add(
+                EmailSendRateBucket(
+                    bucket_scope_hash=scope_hash,
+                    window_started_at=window_started_at,
+                    attempt_count=attempt_count,
+                    expires_at=expires_at,
+                    created_at=observed_at,
+                    updated_at=observed_at,
+                )
+            )
+        else:
+            bucket.window_started_at = window_started_at
+            bucket.attempt_count = attempt_count
+            bucket.expires_at = expires_at
+            bucket.updated_at = observed_at
+
+        decision = EmailSendRateLimitDecision(
+            allowed=allowed,
+            reason="allowed" if allowed else "quota_exhausted",
+        )
+        session.add(
+            _audit_event(
+                auth_context,
+                scope_hash=scope_hash,
+                decision=decision,
+            )
+        )
+        await session.commit()
+        return decision
+    except Exception as exc:
+        try:
+            await session.rollback()
+        except Exception as rollback_exc:
+            logger.warning(
+                "Email send rate limiter rollback failed; error_type=%s",
+                type(rollback_exc).__name__,
+            )
+        logger.warning(
+            "Email send rate limiter decision unavailable; "
+            "event_action=email_send_rate_limit.unavailable error_type=%s",
+            type(exc).__name__,
+        )
+        raise EmailSendRateLimitUnavailable from exc