diff --git a/app/application/subscription/complete.py b/app/application/subscription/complete.py index f36edd3a5..a3b3bfdbc 100644 --- a/app/application/subscription/complete.py +++ b/app/application/subscription/complete.py @@ -13,6 +13,7 @@ from app.application.outbox import ( SyncOutboxTransaction, SyncUnitOfWork, ) +from app.runtime.log import logger class SubscriptionCompletionRepository(Protocol): @@ -116,9 +117,15 @@ class CompleteSubscriptionCommand: 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) + try: + report_delivered = report(report_payload["subscribe_info"]) + except Exception as error: + logger.warning(f"订阅完成统计上报失败,将由后台重试:{error}") + else: + if report_delivered is False: + logger.warning("订阅完成统计上报未确认,将由后台重试") + else: + self._complete_sync_delivery(report_key) def _claim_sync_delivery(self, event_key: str) -> bool: """在同步副作用前取得 lease,已由恢复投递接管时跳过直投。""" diff --git a/app/application/subscription/delete.py b/app/application/subscription/delete.py index 1149aaefd..2bd2fd4e7 100644 --- a/app/application/subscription/delete.py +++ b/app/application/subscription/delete.py @@ -14,6 +14,7 @@ from app.application.outbox import ( SyncUnitOfWork, SUBSCRIBE_DELETED_TOPIC, ) +from app.runtime.log import logger from app.schemas.event import SubscribeDeletedEventData @@ -153,16 +154,20 @@ class DeleteSubscribeCommand: ) # 上报适配器会自行白名单过滤公开字段;传完整删除前快照可保留音乐实体维度, # 避免 Agent 与 API 入口收敛后丢失 music_type / total_tracks。 - report_result = self._report_deleted(effects.report_payload) - if inspect.isawaitable(report_result): - report_result = await report_result - if report_result is False: - raise RuntimeError("订阅删除统计上报未确认") - if self._outbox: - await self._outbox.complete_by_event_key( - effects.report_intent.event_key, - datetime.now(timezone.utc), - ) + try: + report_result = self._report_deleted(effects.report_payload) + if inspect.isawaitable(report_result): + report_result = await report_result + except Exception as error: + logger.warning(f"订阅删除统计上报失败,将由后台重试:{error}") + else: + if report_result is False: + logger.warning("订阅删除统计上报未确认,将由后台重试") + elif self._outbox: + await self._outbox.complete_by_event_key( + effects.report_intent.event_key, + datetime.now(timezone.utc), + ) return True @@ -216,13 +221,18 @@ class SyncDeleteSubscribeCommand: effects.event_intent.event_key, datetime.now(timezone.utc), ) - if self._report_deleted(effects.report_payload) is False: - raise RuntimeError("订阅删除统计上报未确认") - if self._outbox: - self._outbox.complete_by_event_key( - effects.report_intent.event_key, - datetime.now(timezone.utc), - ) + try: + report_result = self._report_deleted(effects.report_payload) + except Exception as error: + logger.warning(f"订阅删除统计上报失败,将由后台重试:{error}") + else: + if report_result is False: + logger.warning("订阅删除统计上报未确认,将由后台重试") + elif self._outbox: + self._outbox.complete_by_event_key( + effects.report_intent.event_key, + datetime.now(timezone.utc), + ) return True diff --git a/app/application/subscription/write.py b/app/application/subscription/write.py index 1c9a86ae7..17216cadb 100644 --- a/app/application/subscription/write.py +++ b/app/application/subscription/write.py @@ -28,8 +28,8 @@ from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType # 而后续按身份去重也会失效,所以必须在查询与建模之前短路 INCOMPLETE_IDENTITY = (0, "媒体身份不完整") -AfterCommitEffect = Callable[[int], None] -AsyncAfterCommitEffect = Callable[[int], Awaitable[None]] +AfterCommitEffect = Callable[[int], bool | None] +AsyncAfterCommitEffect = Callable[[int], Awaitable[bool | None]] class SubscriptionOutboxStager(Protocol): @@ -57,7 +57,7 @@ class SubscribeWriter(Protocol): after_commit: Optional[AfterCommitEffect] = None, notification: Mapping[str, object] | None = None, ) -> Tuple[int, str]: - """同步新增订阅,并在事务成功后执行外部副作用。""" + """同步新增订阅;提交后回调返回 False 时保留统计 intent 待重试。""" async def async_add( self, @@ -67,7 +67,7 @@ class SubscribeWriter(Protocol): after_commit: Optional[AsyncAfterCommitEffect] = None, notification: Mapping[str, object] | None = None, ) -> Tuple[int, str]: - """异步新增订阅,并在事务成功后执行外部副作用。""" + """异步新增订阅;提交后回调返回 False 时保留统计 intent 待重试。""" class StagedSubscription(Protocol): @@ -382,7 +382,7 @@ def add_subscribe( :param mediainfo: 识别结果 :param subscribe_oper: 复用的订阅操作对象,未传时由启动组合根提供 - :param after_commit: 数据提交后执行的消息、事件或上报编排 + :param after_commit: 提交后副作用编排;返回 False 表示统计 intent 等待重试 :param kwargs: 订阅设置;owner_scope 为真时按用户名限定查重范围 :return: (订阅 ID, 结果说明);ID 为 0 表示未新增 """ @@ -420,7 +420,7 @@ async def async_add_subscribe( :param mediainfo: 识别结果 :param subscribe_oper: 复用的订阅操作对象,未传时由启动组合根提供 - :param after_commit: 数据提交后执行的异步消息、事件或上报编排 + :param after_commit: 异步提交后副作用编排;返回 False 表示统计 intent 等待重试 :param kwargs: 订阅设置;owner_scope 为真时按用户名限定查重范围 :return: (订阅 ID, 结果说明);ID 为 0 表示未新增 """ diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index 8239039fc..a0a39401d 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -569,8 +569,8 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): self, subscribe_id: int, context: _SubscribePostCommitContext, - ) -> None: - """同步执行提交后消息、事件和统计,异常不再触碰数据库事务。""" + ) -> bool: + """同步执行提交后消息和事件;统计失败留给 outbox 重试。""" if context.notification: self.post_message(_SchemaMessage.model_validate(context.notification)) self.eventmanager.send_event(EventType.SubscribeAdded, { @@ -582,17 +582,23 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): "username": context.username, "mediainfo": context.mediainfo.to_dict(), }) - if not MoviePilotServerHelper.sub_reg_durable( - self.__subscribe_report_payload(context) - ): - raise RuntimeError("订阅新增统计上报未确认") + try: + report_delivered = MoviePilotServerHelper.sub_reg_durable( + self.__subscribe_report_payload(context) + ) + except Exception as error: + logger.warning(f"订阅新增统计上报失败,将由后台重试:{error}") + return False + if not report_delivered: + logger.warning("订阅新增统计上报未确认,将由后台重试") + return report_delivered async def __async_post_subscribe_added( self, subscribe_id: int, context: _SubscribePostCommitContext, - ) -> None: - """异步执行提交后消息、事件和统计,保持与同步入口相同顺序。""" + ) -> bool: + """异步执行提交后消息和事件;统计失败留给 outbox 重试。""" if context.notification: await self.async_post_message( _SchemaMessage.model_validate(context.notification) @@ -606,10 +612,16 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): "username": context.username, "mediainfo": context.mediainfo.to_dict(), }) - if not await MoviePilotServerHelper.async_sub_reg_durable( - self.__subscribe_report_payload(context) - ): - raise RuntimeError("订阅新增统计上报未确认") + try: + report_delivered = await MoviePilotServerHelper.async_sub_reg_durable( + self.__subscribe_report_payload(context) + ) + except Exception as error: + logger.warning(f"订阅新增统计上报失败,将由后台重试:{error}") + return False + if not report_delivered: + logger.warning("订阅新增统计上报未确认,将由后台重试") + return report_delivered @staticmethod def __build_subscribe_create_context( @@ -958,9 +970,9 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): self.__build_subscribe_notification(context), ) - def _after_commit(subscribe_id: int) -> None: + def _after_commit(subscribe_id: int) -> bool: """把同步提交后的副作用委托给单一顺序实现。""" - self.__post_subscribe_added(subscribe_id, post_commit_context) + return self.__post_subscribe_added(subscribe_id, post_commit_context) sid, err_msg = add_subscribe( mediainfo=context.mediainfo, @@ -985,9 +997,12 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): self.__build_subscribe_notification(context), ) - async def _after_commit(subscribe_id: int) -> None: + async def _after_commit(subscribe_id: int) -> bool: """把异步提交后的副作用委托给单一顺序实现。""" - await self.__async_post_subscribe_added(subscribe_id, post_commit_context) + return await self.__async_post_subscribe_added( + subscribe_id, + post_commit_context, + ) sid, err_msg = await async_add_subscribe( mediainfo=context.mediainfo, diff --git a/app/db/adapters/subscription.py b/app/db/adapters/subscription.py index c8d34b4b6..062f3bbda 100644 --- a/app/db/adapters/subscription.py +++ b/app/db/adapters/subscription.py @@ -58,9 +58,9 @@ class TransactionalSubscribeWriter: ) def delivered(subscribe_id: int) -> None: - """执行旧 post-commit 编排,全部成功后收口 durable intent。""" + """执行提交后编排,分别收口已确认的 durable intent。""" if after_commit: - after_commit(subscribe_id) + report_delivered = after_commit(subscribe_id) outbox.complete_by_event_key( subscription_added_event_key(subscribe_id, payload), datetime.now(timezone.utc), @@ -70,10 +70,11 @@ class TransactionalSubscribeWriter: subscription_added_notification_key(subscribe_id, payload), datetime.now(timezone.utc), ) - outbox.complete_by_event_key( - subscription_added_report_key(subscribe_id, payload), - datetime.now(timezone.utc), - ) + if report_delivered is not False: + outbox.complete_by_event_key( + subscription_added_report_key(subscribe_id, payload), + datetime.now(timezone.utc), + ) return command.execute( identity, @@ -103,9 +104,9 @@ class TransactionalSubscribeWriter: ) async def delivered(subscribe_id: int) -> None: - """异步执行旧编排,全部成功后收口 durable intent。""" + """异步执行提交后编排,分别收口已确认的 durable intent。""" if after_commit: - await after_commit(subscribe_id) + report_delivered = await after_commit(subscribe_id) await outbox.complete_by_event_key( subscription_added_event_key(subscribe_id, payload), datetime.now(timezone.utc), @@ -115,10 +116,11 @@ class TransactionalSubscribeWriter: subscription_added_notification_key(subscribe_id, payload), datetime.now(timezone.utc), ) - await outbox.complete_by_event_key( - subscription_added_report_key(subscribe_id, payload), - datetime.now(timezone.utc), - ) + if report_delivered is not False: + await outbox.complete_by_event_key( + subscription_added_report_key(subscribe_id, payload), + datetime.now(timezone.utc), + ) return await command.execute( identity, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 755f8d506..285a8c61f 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6702, - "edge_sha256": "a6d5e5b0ccca6f84fe5d8aa5acb0402e7253958913da8d9624a23bd80ceab748", + "edge_count": 6706, + "edge_sha256": "dc1249bc5f0ae05ec11c680236cf4dd389258cf883fbed66ed57b53875f9ef0e", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -2863,6 +2863,8 @@ "app.application.storage -> app.schemas.types", "app.application.subscription.complete -> app.application", "app.application.subscription.complete -> app.application.outbox", + "app.application.subscription.complete -> app.runtime", + "app.application.subscription.complete -> app.runtime.log", "app.application.subscription.contract -> app.domain", "app.application.subscription.contract -> app.domain.meta", "app.application.subscription.contract -> app.domain.meta.metabase", @@ -2873,6 +2875,8 @@ "app.application.subscription.contract -> app.schemas.types", "app.application.subscription.delete -> app.application", "app.application.subscription.delete -> app.application.outbox", + "app.application.subscription.delete -> app.runtime", + "app.application.subscription.delete -> app.runtime.log", "app.application.subscription.delete -> app.schemas", "app.application.subscription.delete -> app.schemas.event", "app.application.subscription.identity -> app.application", diff --git a/tests/test_subscribe_create_command.py b/tests/test_subscribe_create_command.py index 3b970875f..21575f031 100644 --- a/tests/test_subscribe_create_command.py +++ b/tests/test_subscribe_create_command.py @@ -4,6 +4,7 @@ import asyncio from unittest.mock import AsyncMock, Mock import pytest +from sqlalchemy import select from app.application.subscription.write import ( AsyncCreateSubscriptionCommand, @@ -11,6 +12,7 @@ from app.application.subscription.write import ( add_subscribe, async_add_subscribe, ) +from app.db.models.outbox import OutboxMessage from app.db.models.subscribe import Subscribe from app.db.oper.subscribe import SubscribeOper, SubscribeStageResult from app.domain.context import MediaInfo @@ -190,6 +192,55 @@ def test_default_sync_writer_persists_once_and_reuses_duplicate(db) -> None: ] +def test_default_sync_writer_keeps_failed_report_pending_without_raising(db) -> None: + """新增统计未确认时接口仍成功,事件 intent 收口而统计 intent 等待重试。""" + db.watermark(Subscribe, OutboxMessage) + media = _media("arch-221-report-pending") + + subscribe_id, message = add_subscribe( + mediainfo=media, + after_commit=lambda _subscribe_id: False, + ) + + assert subscribe_id > 0 + assert message == "新增订阅成功" + intents = db.session.execute( + select(OutboxMessage) + .where(OutboxMessage.event_key.contains(media.media_id)) + .order_by(OutboxMessage.id) + ).scalars().all() + assert [(intent.topic, intent.status) for intent in intents] == [ + ("subscribe.added", "completed"), + ("subscribe.added.report", "pending"), + ] + + +def test_default_async_writer_keeps_failed_report_pending_without_raising(db) -> None: + """异步新增入口同样返回成功,并只留下统计 intent 等待重试。""" + db.watermark(Subscribe, OutboxMessage) + media = _media("arch-221-async-report-pending") + + async def report_failed(_subscribe_id: int) -> bool: + """模拟异步统计接口未确认。""" + return False + + subscribe_id, message = asyncio.run( + async_add_subscribe(mediainfo=media, after_commit=report_failed) + ) + + assert subscribe_id > 0 + assert message == "新增订阅成功" + intents = db.session.execute( + select(OutboxMessage) + .where(OutboxMessage.event_key.contains(media.media_id)) + .order_by(OutboxMessage.id) + ).scalars().all() + assert [(intent.topic, intent.status) for intent in intents] == [ + ("subscribe.added", "completed"), + ("subscribe.added.report", "pending"), + ] + + def test_stage_add_reuses_explicit_session_without_commit(db, monkeypatch) -> None: """Oper 将调用方 Session 传给 Model 查询原语,暂存期间不自行提交。""" db.watermark(Subscribe) diff --git a/tests/test_subscribe_delete_command.py b/tests/test_subscribe_delete_command.py index 6b08eec97..ffc832e2f 100644 --- a/tests/test_subscribe_delete_command.py +++ b/tests/test_subscribe_delete_command.py @@ -200,7 +200,14 @@ def _async_report_command(candidate, calls, result=True, error=None, outbox=None ) -def _sync_command(candidate, calls, commit_error=None, delete_error=None, outbox=None): +def _sync_command( + candidate, + calls, + commit_error=None, + delete_error=None, + outbox=None, + report_result=True, +): """构造可观察事务和副作用顺序的同步订阅删除命令。""" def publish(payload): """记录同步删除事件。""" @@ -209,7 +216,7 @@ def _sync_command(candidate, calls, commit_error=None, delete_error=None, outbox def report(payload): """记录同步删除统计。""" calls.append(("report", payload)) - return True + return report_result return SyncDeleteSubscribeCommand( repository=_SyncRepository(candidate, calls, delete_error), @@ -321,15 +328,14 @@ async def test_event_failure_happens_after_commit_and_stops_report(): @pytest.mark.asyncio async def test_report_failure_happens_after_commit_and_event(): - """上报失败保持原有传播语义,且不得改变已经提交和发出的事件。""" + """上报异常不得把已经提交的删除误报为失败。""" calls = [] command = _command(_candidate(), calls, report_error=RuntimeError("report failed")) - with pytest.raises(RuntimeError, match="report failed"): - await command.execute( - 7, - SubscribeDeletionActor(username="alice", is_superuser=False), - ) + assert await command.execute( + 7, + SubscribeDeletionActor(username="alice", is_superuser=False), + ) is True assert [call[0] for call in calls] == ["get", "delete", "commit", "event", "report"] @@ -408,15 +414,14 @@ async def test_async_reporter_completes_report_intent_only_after_confirmation(): @pytest.mark.asyncio async def test_async_reporter_false_keeps_report_intent_pending(): - """异步 reporter 未确认时必须保留待重试统计 intent。""" + """异步 reporter 未确认时返回成功并保留待重试统计 intent。""" calls = [] command = _async_report_command(_candidate(), calls, result=False, outbox=_Outbox(calls)) - with pytest.raises(RuntimeError, match="未确认"): - await command.execute( - 7, - SubscribeDeletionActor(username="alice", is_superuser=False), - ) + assert await command.execute( + 7, + SubscribeDeletionActor(username="alice", is_superuser=False), + ) is True assert [call[0] for call in calls] == [ "get", "delete", "outbox_stage", "outbox_stage", "commit", @@ -426,7 +431,7 @@ async def test_async_reporter_false_keeps_report_intent_pending(): @pytest.mark.asyncio async def test_async_reporter_error_keeps_report_intent_pending(): - """异步 reporter 异常时必须保留待重试统计 intent。""" + """异步 reporter 异常时返回成功并保留待重试统计 intent。""" calls = [] command = _async_report_command( _candidate(), @@ -435,11 +440,10 @@ async def test_async_reporter_error_keeps_report_intent_pending(): outbox=_Outbox(calls), ) - with pytest.raises(RuntimeError, match="remote failed"): - await command.execute( - 7, - SubscribeDeletionActor(username="alice", is_superuser=False), - ) + assert await command.execute( + 7, + SubscribeDeletionActor(username="alice", is_superuser=False), + ) is True assert [call[0] for call in calls] == [ "get", "delete", "outbox_stage", "outbox_stage", "commit", @@ -505,6 +509,26 @@ def test_sync_delete_uses_same_durable_effect_order(): ] assert calls[2][1].topic == "subscribe.deleted" assert calls[3][1].topic == "subscribe.deleted.report" + + +def test_sync_delete_report_failure_returns_success_and_keeps_intent_pending(): + """同步删除统计未确认时仍返回成功,且不收口 report intent。""" + calls = [] + command = _sync_command( + _candidate(), + calls, + outbox=_SyncOutbox(calls), + report_result=False, + ) + + assert command.execute( + 7, + SubscribeDeletionActor(username="alice", is_superuser=False), + ) is True + assert [call[0] for call in calls] == [ + "get", "delete", "outbox_stage", "outbox_stage", "commit", + "event", "outbox_complete", "report", + ] assert calls[7][1] == _candidate().event_payload diff --git a/tests/test_subscription_completion_command.py b/tests/test_subscription_completion_command.py index e437c22cb..dfc63a447 100644 --- a/tests/test_subscription_completion_command.py +++ b/tests/test_subscription_completion_command.py @@ -71,6 +71,7 @@ def _command( report_result=True, notify_error=None, claim_result=True, + report_error=None, ): """构造可注入失败的完成命令。""" def notify() -> None: @@ -88,6 +89,8 @@ def _command( def report(payload) -> bool: """记录完成统计。""" calls.append(("report", payload)) + if report_error: + raise report_error return report_result return CompleteSubscriptionCommand( @@ -98,14 +101,13 @@ def _command( ), notify, report -@pytest.mark.parametrize("failure", ["event", "report", "notify"]) +@pytest.mark.parametrize("failure", ["event", "notify"]) def test_completion_stages_business_and_independent_intents_before_commit(failure): """完成事务先提交业务和两个 intent,提交后按通知、事件、统计顺序执行。""" calls = [] command, notify, report = _command( calls, publish_error=RuntimeError("event failed") if failure == "event" else None, - report_result=False if failure == "report" else True, notify_error=RuntimeError("notify failed") if failure == "notify" else None, ) @@ -127,10 +129,47 @@ def test_completion_stages_business_and_independent_intents_before_commit(failur assert [call[0] for call in calls[5:]] == ["notify"] elif failure == "event": assert [call[0] for call in calls[5:]] == ["notify", "claim", "event"] - else: - assert [call[0] for call in calls[5:]] == [ - "notify", "claim", "event", "complete", "claim", "report", - ] + + +def test_completion_report_failure_returns_success_and_keeps_intent_pending(): + """统计未确认不得误报完成失败,且 report intent 必须留待重试。""" + calls = [] + command, notify, report = _command(calls, report_result=False) + + command.execute( + 7, + {"id": 7, "media_source": "tmdb", "media_id": "123", "season": 2}, + {"title": "Test"}, + notify=notify, + report=report, + ) + + assert [call[0] for call in calls] == [ + "history", "delete", "stage", "stage", "commit", + "notify", "claim", "event", "complete", "claim", "report", + ] + + +def test_completion_report_error_returns_success_and_keeps_intent_pending(): + """统计上报抛出异常也不得覆盖已经成功提交的完成结果。""" + calls = [] + command, notify, report = _command( + calls, + report_error=RuntimeError("remote failed"), + ) + + command.execute( + 7, + {"id": 7, "media_source": "tmdb", "media_id": "123", "season": 2}, + {"title": "Test"}, + notify=notify, + report=report, + ) + + assert [call[0] for call in calls] == [ + "history", "delete", "stage", "stage", "commit", + "notify", "claim", "event", "complete", "claim", "report", + ] def test_completion_success_closes_event_then_report_intent():