fix(outbox): serialize subscription completion delivery

app/application/outbox.py

@@ -20,6 +20,7 @@ SUBSCRIBE_DELETED_TOPIC = "subscribe.deleted"
 DOWNLOAD_ADDED_TOPIC = "download.added"
 TRANSFER_COMPLETED_TOPIC = "transfer.completed"
 TRANSFER_FAILED_TOPIC = "transfer.failed"
+OUTBOX_LEASE_SECONDS = 60
 
 DURABLE_EVENT_TOPICS: Mapping[EventType, str] = MappingProxyType({
     EventType.SubscribeAdded: SUBSCRIBE_ADDED_TOPIC,
@@ -126,6 +127,14 @@ class SyncOutboxTransaction(Protocol):
     def stage(self, intent: OutboxIntent, now: datetime) -> None:
         """把 intent 加入调用方事务，但不自行提交。"""
 
+    def claim_by_event_key(
+        self,
+        event_key: str,
+        now: datetime,
+        lease_until: datetime,
+    ) -> bool:
+        """在同步副作用前原子认领 intent，已被其他投递者持有时返回 False。"""
+
     def complete_by_event_key(
         self,
         event_key: str,
@@ -183,7 +192,7 @@ class OutboxDispatcher:
         handlers: dict[str, Callable[[ClaimedOutboxMessage], None]],
         *,
         max_attempts: int = 5,
-        lease_seconds: int = 60,
+        lease_seconds: int = OUTBOX_LEASE_SECONDS,
         clock: Callable[[], datetime] | None = None,
         close: Callable[[], None] | None = None,
         failure_observer: Callable[[bool], None] | None = None,

app/application/subscription/complete.py

@@ -4,10 +4,15 @@ from __future__ import annotations
 
 from collections.abc import Callable, Mapping
 from contextlib import AbstractContextManager
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Any, Protocol
 
-from app.application.outbox import OutboxIntent, SyncOutboxTransaction, SyncUnitOfWork
+from app.application.outbox import (
+    OUTBOX_LEASE_SECONDS,
+    OutboxIntent,
+    SyncOutboxTransaction,
+    SyncUnitOfWork,
+)
 
 
 class SubscriptionCompletionRepository(Protocol):
@@ -101,19 +106,35 @@ class CompleteSubscriptionCommand:
             self._unit_of_work.rollback()
             raise
 
-        notify()
-        if self._outbox and notification:
-            self._outbox.complete_by_event_key(
-                notification_key,
-                datetime.now(timezone.utc),
-            )
-        self._publish(event_payload)
+        if notification:
+            if self._claim_sync_delivery(notification_key):
+                notify()
+                self._complete_sync_delivery(notification_key)
+        else:
+            notify()
+        if self._claim_sync_delivery(event_key):
+            self._publish(event_payload)
+            self._complete_sync_delivery(event_key)
+        if self._claim_sync_delivery(report_key):
+            if report(report_payload["subscribe_info"]) is False:
+                raise RuntimeError("订阅完成统计上报未确认")
+            self._complete_sync_delivery(report_key)
+
+    def _claim_sync_delivery(self, event_key: str) -> bool:
+        """在同步副作用前取得 lease，已由恢复投递接管时跳过直投。"""
+        if self._outbox is None:
+            return True
+        now = datetime.now(timezone.utc)
+        return self._outbox.claim_by_event_key(
+            event_key,
+            now,
+            now + timedelta(seconds=OUTBOX_LEASE_SECONDS),
+        )
+
+    def _complete_sync_delivery(self, event_key: str) -> None:
+        """收口当前同步投递持有的 durable intent。"""
         if self._outbox:
             self._outbox.complete_by_event_key(event_key, datetime.now(timezone.utc))
-        if report(report_payload["subscribe_info"]) is False:
-            raise RuntimeError("订阅完成统计上报未确认")
-        if self._outbox:
-            self._outbox.complete_by_event_key(report_key, datetime.now(timezone.utc))
 
 
 def completion_event_key(subscribe_id: int, subscribe_info: Mapping[str, Any]) -> str:

app/db/adapters/outbox.py

@@ -91,6 +91,52 @@ class SqlAlchemyOutboxRepository:
             attempt=next_attempt,
         )
 
+    def claim_by_event_key(
+        self,
+        event_key: str,
+        now: datetime,
+        lease_until: datetime,
+    ) -> bool:
+        """按事件键原子认领同步投递，避免与 dispatcher 并发重复发送。"""
+        now_text = _iso(now)
+        candidate = self._session.execute(
+            select(OutboxMessage)
+            .where(
+                OutboxMessage.event_key == event_key,
+                OutboxMessage.status.in_(("pending", "processing")),
+                OutboxMessage.next_retry_at <= now_text,
+                or_(
+                    OutboxMessage.lease_until.is_(None),
+                    OutboxMessage.lease_until <= now_text,
+                ),
+            )
+            .limit(1)
+        ).scalars().first()
+        if candidate is None:
+            return False
+        claimed = execute_dml(
+            self._session,
+            update(OutboxMessage)
+            .where(
+                OutboxMessage.id == candidate.id,
+                OutboxMessage.attempt == candidate.attempt,
+                OutboxMessage.event_key == event_key,
+                OutboxMessage.status.in_(("pending", "processing")),
+                OutboxMessage.next_retry_at <= now_text,
+                or_(
+                    OutboxMessage.lease_until.is_(None),
+                    OutboxMessage.lease_until <= now_text,
+                ),
+            )
+            .values(
+                status="processing",
+                attempt=OutboxMessage.attempt + 1,
+                lease_until=_iso(lease_until),
+            ),
+        )
+        self._session.commit()
+        return bool(claimed)
+
     def complete(self, message_id: int, completed_at: datetime) -> None:
         """持久化完成终态并释放 lease。"""
         self._session.execute(

tests/test_outbox.py

@@ -1,12 +1,17 @@
 """durable side-effect outbox 原子性、认领、重试与幂等测试。"""
 
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from unittest.mock import MagicMock
 
 import pytest
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import sessionmaker
 
-from app.application.outbox import ClaimedOutboxMessage, OutboxDispatcher
+from app.application.outbox import ClaimedOutboxMessage, OutboxDispatcher, OutboxIntent
 from app.application.subscription.write import CreateSubscriptionCommand
+from app.db.adapters.outbox import SqlAlchemyOutboxRepository
+from app.db.base import Base
+from app.db.models.outbox import OutboxMessage
 
 
 class _Staged:
@@ -134,3 +139,35 @@ def test_dispatcher_marks_success_and_closes_owned_resource() -> None:
     repository.complete.assert_called_once_with(7, now)
     dispatcher.close()
     close.assert_called_once_with()
+
+
+def test_sync_outbox_claim_is_exclusive_for_event_key() -> None:
+    """同步投递与恢复投递竞争同一 intent 时只允许一个取得 lease。"""
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    factory = sessionmaker(bind=engine)
+    now = datetime(2026, 8, 24, tzinfo=timezone.utc)
+    lease_until = now + timedelta(seconds=60)
+    event_key = "subscribe.complete:7:tmdb:123:v1"
+
+    with factory() as session:
+        repository = SqlAlchemyOutboxRepository(session)
+        repository.stage(
+            OutboxIntent(event_key=event_key, topic="subscribe.complete", payload={}),
+            now,
+        )
+        session.commit()
+
+    with factory() as owner, factory() as competitor:
+        assert SqlAlchemyOutboxRepository(owner).claim_by_event_key(
+            event_key, now, lease_until
+        ) is True
+        assert SqlAlchemyOutboxRepository(competitor).claim_by_event_key(
+            event_key, now, lease_until
+        ) is False
+
+    with factory() as session:
+        message = session.execute(select(OutboxMessage)).scalar_one()
+    assert message.status == "processing"
+    assert message.attempt == 1
+    assert message.lease_until == lease_until.isoformat()

tests/test_subscription_completion_command.py

@@ -45,20 +45,33 @@ class _UnitOfWork:
 class _Outbox:
     """记录 intent 暂存与即时收口。"""
 
-    def __init__(self, calls: list[tuple]) -> None:
+    def __init__(self, calls: list[tuple], claim_result: bool = True) -> None:
         """保存共享调用序列。"""
         self.calls = calls
+        self.claim_result = claim_result
 
     def stage(self, intent, _now: datetime) -> None:
         """记录 durable intent。"""
         self.calls.append(("stage", intent))
 
+    def claim_by_event_key(self, event_key: str, _now: datetime, _lease_until: datetime) -> bool:
+        """记录同步投递认领结果。"""
+        self.calls.append(("claim", event_key))
+        return self.claim_result
+
     def complete_by_event_key(self, event_key: str, _now: datetime) -> None:
         """记录成功副作用对应的 intent 收口。"""
         self.calls.append(("complete", event_key))
 
 
-def _command(calls: list[tuple], *, publish_error=None, report_result=True, notify_error=None):
+def _command(
+    calls: list[tuple],
+    *,
+    publish_error=None,
+    report_result=True,
+    notify_error=None,
+    claim_result=True,
+):
     """构造可注入失败的完成命令。"""
     def notify() -> None:
         """记录通知。"""
@@ -80,7 +93,7 @@ def _command(calls: list[tuple], *, publish_error=None, report_result=True, noti
     return CompleteSubscriptionCommand(
         repository=_Repository(calls),
         unit_of_work=_UnitOfWork(calls),
-        outbox=_Outbox(calls),
+        outbox=_Outbox(calls, claim_result),
         publish=publish,
     ), notify, report
 
@@ -113,10 +126,10 @@ def test_completion_stages_business_and_independent_intents_before_commit(failur
     if failure == "notify":
         assert [call[0] for call in calls[5:]] == ["notify"]
     elif failure == "event":
-        assert [call[0] for call in calls[5:]] == ["notify", "event"]
+        assert [call[0] for call in calls[5:]] == ["notify", "claim", "event"]
     else:
         assert [call[0] for call in calls[5:]] == [
-            "notify", "event", "complete", "report",
+            "notify", "claim", "event", "complete", "claim", "report",
         ]
 
 
@@ -135,10 +148,11 @@ def test_completion_success_closes_event_then_report_intent():
 
     assert [call[0] for call in calls] == [
         "history", "delete", "stage", "stage", "commit",
-        "notify", "event", "complete", "report", "complete",
+        "notify", "claim", "event", "complete",
+        "claim", "report", "complete",
     ]
-    assert calls[6][1]["idempotency_key"] == calls[2][1].event_key
-    assert calls[8][1]["idempotency_key"] == calls[3][1].event_key
+    assert calls[7][1]["idempotency_key"] == calls[2][1].event_key
+    assert calls[10][1]["idempotency_key"] == calls[3][1].event_key
 
 
 def test_completion_stages_and_closes_notification_snapshot() -> None:
@@ -164,3 +178,23 @@ def test_completion_stages_and_closes_notification_snapshot() -> None:
     assert staged[1].payload["message"]["title"] == "完成"
     completed = [call[1] for call in calls if call[0] == "complete"]
     assert completed[0].endswith(":notification")
+
+
+def test_completion_skips_sync_delivery_owned_by_outbox_dispatcher() -> None:
+    """后台已认领 intent 时同步路径不得再次发送相同副作用。"""
+    calls = []
+    command, notify, report = _command(calls, claim_result=False)
+
+    command.execute(
+        7,
+        {"id": 7, "media_source": "tmdb", "media_id": "123", "season": 2},
+        {"title": "Test"},
+        notify=notify,
+        report=report,
+        notification={"title": "完成", "text": "Test"},
+    )
+
+    assert [call[0] for call in calls] == [
+        "history", "delete", "stage", "stage", "stage", "commit",
+        "claim", "claim", "claim",
+    ]