diff --git a/app/application/outbox.py b/app/application/outbox.py index b953fcccd..ec156dbb7 100644 --- a/app/application/outbox.py +++ b/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, diff --git a/app/application/subscription/complete.py b/app/application/subscription/complete.py index b5f7f750e..f36edd3a5 100644 --- a/app/application/subscription/complete.py +++ b/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: diff --git a/app/db/adapters/outbox.py b/app/db/adapters/outbox.py index c05b4b4e0..0aa372915 100644 --- a/app/db/adapters/outbox.py +++ b/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( diff --git a/tests/test_outbox.py b/tests/test_outbox.py index a42e2e9a1..12f2770e7 100644 --- a/tests/test_outbox.py +++ b/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() diff --git a/tests/test_subscription_completion_command.py b/tests/test_subscription_completion_command.py index 3282b6c22..e437c22cb 100644 --- a/tests/test_subscription_completion_command.py +++ b/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", + ]