fix(email): enforce shared send throttling

CHANGELOG.md

@@ -1,4 +1,7 @@
 ## [Unreleased]
+- 이메일 발송 연결 중 요청을 취소하면 남은 연결을 즉시 정리합니다. 직접 발송과 등록된 Connector에 같은 수정이 적용되며, 취소를 발송 성공으로 처리하지 않습니다.
+- 이메일 발송 제한을 모든 worker가 공유하는 PostgreSQL rolling window로
+  적용하고, 제한 상태를 확인할 수 없으면 발송 전에 안전하게 중단합니다.
 - 데이터 저장·검색 검사가 실제 DB 연결 없이 건너뛰어져도 성공으로 보이던 검증 공백을 보완했습니다. 새 DB 설치와 반복 업그레이드 후 전체 백엔드 검사를 실행하며, 아직 실제 배포 환경 검증을 뜻하지는 않습니다.
 - Starlette `TestClient`의 기존 `httpx2==2.5.0` pin을 core 개발·테스트 의존성으로 승격하고, deprecated `httpx` fallback 경고 억제를 제거했습니다.
 - 긴 이메일·첨부 본문을 의미 단위 청크로 임베딩한 뒤 기존 email/attachment 벡터 계약으로 평균화하고, 청크 요청·벡터 누적을 제한된 창으로 처리합니다. OpenAI `text-embedding-3-*`에는 저장 차원(`1536`)을 직접 요청하도록 보강했습니다. 합성 메일 fixture 5건(70청크)과 provider 요청 계약으로 1,536차원 벡터 경로를 검증했으며, 실행 시 선택한 임베딩 제공자에 본문·파싱된 첨부 텍스트를 전송할 수 있습니다. 회사 기밀 데이터는 fixture·commit·PR·log에 포함하지 않습니다.

backend/alembic/versions/0018_security_audit_events.py

@@ -0,0 +1,84 @@
+"""Formalize durable security audit events in the Alembic upgrade path.
+
+Revision ID: 0018_security_audit_events
+Revises: 0017_merge_newsdom_carddav_heads
+Create Date: 2026-09-05 00:00:00.000000
+
+Older installations can already contain ``security_audit_events`` because the
+legacy bootstrap path created it outside Alembic. This revision is therefore
+idempotent: it creates the table when absent and reconciles the model-owned
+indexes when the table already exists. Downgrade intentionally preserves the
+durable audit table and its evidence rather than deleting security history that
+may predate this revision.
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0018_security_audit_events"
+down_revision = "0017_merge_newsdom_carddav_heads"
+branch_labels = None
+depends_on = None
+
+_TABLE = "security_audit_events"
+_INDEXES: tuple[tuple[str, list[str]], ...] = (
+    ("ix_security_audit_events_actor_user_id", ["actor_user_id"]),
+    ("ix_security_audit_events_actor_role", ["actor_role"]),
+    ("ix_security_audit_events_organization_id", ["organization_id"]),
+    ("ix_security_audit_events_workspace_id", ["workspace_id"]),
+    ("ix_security_audit_events_event_action", ["event_action"]),
+    ("ix_security_audit_events_resource_type", ["resource_type"]),
+    ("ix_security_audit_events_resource_uid", ["resource_uid"]),
+    ("ix_security_audit_events_observed_at", ["observed_at"]),
+    (
+        "ix_security_audit_events_scope_time",
+        ["organization_id", "workspace_id", "observed_at"],
+    ),
+    (
+        "ix_security_audit_events_actor_scope",
+        ["actor_user_id", "organization_id", "workspace_id"],
+    ),
+)
+
+
+def upgrade() -> None:
+    """Create the audit schema missing from Alembic-managed upgrades."""
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+
+    if not inspector.has_table(_TABLE):
+        op.create_table(
+            _TABLE,
+            sa.Column("event_uid", sa.String(), nullable=False),
+            sa.Column("actor_user_id", sa.String(), nullable=False),
+            sa.Column("actor_role", sa.String(), nullable=False),
+            sa.Column("organization_id", sa.String(), nullable=True),
+            sa.Column("workspace_id", sa.String(), nullable=False),
+            sa.Column("event_action", sa.String(), nullable=False),
+            sa.Column("resource_type", sa.String(), nullable=False),
+            sa.Column("resource_uid", sa.String(), nullable=True),
+            sa.Column("evidence_source", sa.String(), nullable=False),
+            sa.Column("detail_text", sa.Text(), nullable=True),
+            sa.Column(
+                "observed_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.text("CURRENT_TIMESTAMP"),
+            ),
+            sa.PrimaryKeyConstraint("event_uid"),
+        )
+
+    for index_name, column_names in _INDEXES:
+        op.create_index(
+            index_name,
+            _TABLE,
+            column_names,
+            if_not_exists=True,
+        )
+
+
+def downgrade() -> None:
+    """Preserve durable security evidence created before or after this revision."""
+    # This revision reconciles a table that may predate Alembic ownership.
+    # Dropping it on downgrade could destroy security evidence owned by the
+    # earlier bootstrap path, so schema rollback deliberately leaves it intact.

backend/alembic/versions/0020_merge_send_registry.py

@@ -0,0 +1,18 @@
+"""Join shared-send audit and workspace registry migration histories.
+
+Both parents own durable data. This revision only reconciles the graph so the
+normal ``upgrade head`` path applies both prerequisites without deleting either.
+"""
+
+revision = "0020_merge_send_registry"
+down_revision = ("0018_security_audit_events", "0019_email_read_state_repair")
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Join already-applied parent revisions without changing their data."""
+
+
+def downgrade() -> None:
+    """Split revision bookkeeping without deleting parent-owned data."""

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)
@@ -729,6 +703,11 @@ async def send_email_endpoint(
             smtp_port = tenant_config.smtp_port
             smtp_username = tenant_config.smtp_username
             smtp_password = tenant_config.smtp_password
+            # The production dependency is an AsyncSession. Lightweight unit-test
+            # doubles do not own a pooled connection and may omit rollback().
+            rollback = getattr(db, "rollback", None)
+            if rollback is not None:
+                await rollback()
             validate_smtp_destination(smtp_server, smtp_port)
         except Exception as exc:
             if "ENCRYPTION_KEY is required" in str(exc):
@@ -756,7 +735,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(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/services/email_client.py

@@ -400,7 +400,7 @@ async def _connect_validated_smtp_socket(
             ),
             timeout=SMTP_TIMEOUT_SECONDS,
         )
-    except Exception:
+    except (Exception, asyncio.CancelledError):
         smtp_socket.close()
         raise
     return smtp_socket

backend/services/email_send_rate_limiter.py

@@ -0,0 +1,186 @@
+"""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, delete, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.models import SecurityAuditEvent
+from db.session import AsyncSessionLocal
+
+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, workspace_id: str
+) -> 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{workspace_id}\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 _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,
+    observed_at: datetime.datetime,
+) -> 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",
+        observed_at=observed_at,
+        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(
+    auth_context: AuthContext,
+    *,
+    now: datetime.datetime | None = None,
+) -> EmailSendRateLimitDecision:
+    """Atomically reserve one send attempt in a rolling PostgreSQL window.
+
+    A limiter-owned transaction prevents committing unrelated request work.
+    PostgreSQL advisory locking serializes the count-and-record decision across
+    workers. Production timestamps come from PostgreSQL after the scope lock so
+    worker clock skew and lock wait do not weaken the real-time quota. Expired
+    allowed rows are transient reservation state and are pruned under that same
+    lock; durable quota-denial audit evidence is retained. ``now`` remains an
+    explicit deterministic test seam.
+    """
+    observed_at = now
+    scope_hash = rate_limit_scope_hash(
+        auth_context.user_id,
+        auth_context.organization_id,
+        auth_context.workspace_id,
+    )
+    scope_uid = f"email_send_scope:{scope_hash}"
+    async with AsyncSessionLocal() as session:
+        if not _session_uses_postgresql(session):
+            raise EmailSendRateLimitUnavailable
+        try:
+            await session.connection(
+                execution_options={"isolation_level": "READ COMMITTED"}
+            )
+            await session.execute(
+                select(func.pg_advisory_xact_lock(bindparam("lock_key"))),
+                {"lock_key": _lock_key(scope_hash)},
+            )
+            if observed_at is None:
+                database_clock = await session.execute(select(func.clock_timestamp()))
+                observed_at = database_clock.scalar_one()
+            window_started_at = observed_at - datetime.timedelta(
+                seconds=SEND_RATE_LIMIT_WINDOW_SECONDS
+            )
+            await session.execute(
+                delete(SecurityAuditEvent).where(
+                    SecurityAuditEvent.resource_uid == scope_uid,
+                    SecurityAuditEvent.event_action
+                    == "email_send_rate_limit.allowed",
+                    SecurityAuditEvent.observed_at <= window_started_at,
+                )
+            )
+            result = await session.execute(
+                select(func.count())
+                .select_from(SecurityAuditEvent)
+                .where(
+                    SecurityAuditEvent.resource_uid == scope_uid,
+                    SecurityAuditEvent.event_action
+                    == "email_send_rate_limit.allowed",
+                    SecurityAuditEvent.observed_at > window_started_at,
+                )
+            )
+            allowed = result.scalar_one() < SEND_RATE_LIMIT_MAX_ATTEMPTS
+            decision = EmailSendRateLimitDecision(
+                allowed=allowed,
+                reason="allowed" if allowed else "quota_exhausted",
+            )
+            record_decision = allowed
+            if not allowed:
+                denied_result = await session.execute(
+                    select(func.count())
+                    .select_from(SecurityAuditEvent)
+                    .where(
+                        SecurityAuditEvent.resource_uid == scope_uid,
+                        SecurityAuditEvent.event_action
+                        == "email_send_rate_limit.quota_exhausted",
+                        SecurityAuditEvent.observed_at > window_started_at,
+                    )
+                )
+                record_decision = denied_result.scalar_one() == 0
+            if record_decision:
+                session.add(
+                    _audit_event(
+                        auth_context,
+                        scope_hash=scope_hash,
+                        decision=decision,
+                        observed_at=observed_at,
+                    )
+                )
+            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