From 3c8cb513bb379e00ed4592cca1abf024c2491e57 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 23 Aug 2026 00:07:52 +0800 Subject: [PATCH] fix: await durable subscribe deletion reports --- app/adapters/external/server.py | 20 +++++++ app/api/dependencies/subscription.py | 2 +- app/application/subscription/delete.py | 5 +- app/startup/subscription.py | 2 +- tests/test_subscribe_delete_command.py | 80 ++++++++++++++++++++++++++ 5 files changed, 106 insertions(+), 3 deletions(-) diff --git a/app/adapters/external/server.py b/app/adapters/external/server.py index f94909151..9e2293192 100644 --- a/app/adapters/external/server.py +++ b/app/adapters/external/server.py @@ -662,6 +662,15 @@ class MoviePilotServerHelper: """ return cls._post_json(cls._server_url(cls._SUBSCRIBE_DONE_PATH), payload, timeout=5) + @classmethod + async def async_subscribe_done(cls, payload: Dict[str, Any]): + """异步完成订阅统计，并返回可检查 HTTP 状态的响应对象。""" + return await cls._async_post_json( + cls._server_url(cls._SUBSCRIBE_DONE_PATH), + payload, + timeout=5, + ) + @classmethod def subscribe_report(cls, subscribes: List[Dict[str, Any]]): """ @@ -874,6 +883,17 @@ class MoviePilotServerHelper: res = cls.subscribe_done(payload) return bool(res is not None and res.status_code == 200) + @classmethod + async def async_sub_done(cls, sub: dict) -> bool: + """异步完成订阅统计，并仅在服务端确认成功时返回 True。""" + if not settings.SUBSCRIBE_STATISTIC_SHARE: + return False + payload = cls._build_subscribe_statistic_payload(sub) + if not payload: + return False + res = await cls.async_subscribe_done(payload) + return bool(res is not None and res.status_code == 200) + @classmethod def sub_reg_async(cls, sub: dict) -> bool: """ diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index 2897e35c1..dc0357150 100644 --- a/app/api/dependencies/subscription.py +++ b/app/api/dependencies/subscription.py @@ -62,7 +62,7 @@ def get_delete_subscribe_command( repository=cast(SubscribeDeletionRepository, repository_port), unit_of_work=cast(DeleteUnitOfWork, unit_of_work), publish_deleted=_publish_subscribe_deleted, - report_deleted=MoviePilotServerHelper.sub_done_async, + report_deleted=MoviePilotServerHelper.async_sub_done, outbox=outbox, ) diff --git a/app/application/subscription/delete.py b/app/application/subscription/delete.py index 5a8c9334f..fe18c8e8e 100644 --- a/app/application/subscription/delete.py +++ b/app/application/subscription/delete.py @@ -3,6 +3,7 @@ from contextlib import AbstractAsyncContextManager from dataclasses import dataclass from datetime import datetime, timezone +import inspect from typing import Any, Awaitable, Callable, Mapping, Protocol, cast from uuid import uuid4 @@ -55,7 +56,7 @@ class AsyncUnitOfWork(Protocol): SubscribeDeletedPublisher = Callable[[dict[str, Any]], Awaitable[None]] -SubscribeDeletedReporter = Callable[[Mapping[str, object]], object] +SubscribeDeletedReporter = Callable[[Mapping[str, object]], object | Awaitable[object]] class DeleteSubscribeCommand: @@ -134,6 +135,8 @@ class DeleteSubscribeCommand: # 上报适配器会自行白名单过滤公开字段；传完整删除前快照可保留音乐实体维度， # 避免 Agent 与 API 入口收敛后丢失 music_type / total_tracks。 report_result = self._report_deleted(dict(candidate.event_payload)) + if inspect.isawaitable(report_result): + report_result = await report_result if report_result is False: raise RuntimeError("订阅删除统计上报未确认") if self._outbox: diff --git a/app/startup/subscription.py b/app/startup/subscription.py index 91611be7c..bf7ac718f 100644 --- a/app/startup/subscription.py +++ b/app/startup/subscription.py @@ -145,7 +145,7 @@ async def delete_subscribe_scope(): repository=SubscribeOper(session), unit_of_work=SqlAlchemyAsyncUnitOfWork(session), publish_deleted=_publish_deleted, - report_deleted=MoviePilotServerHelper.sub_done_async, + report_deleted=MoviePilotServerHelper.async_sub_done, outbox=SqlAlchemyAsyncOutboxStager(session), ) diff --git a/tests/test_subscribe_delete_command.py b/tests/test_subscribe_delete_command.py index 0c49827a8..1811d632d 100644 --- a/tests/test_subscribe_delete_command.py +++ b/tests/test_subscribe_delete_command.py @@ -116,6 +116,28 @@ def _command( ) +def _async_report_command(candidate, calls, result=True, error=None, outbox=None): + """构造异步统计 reporter，验证命令可等待真实远端确认。""" + async def publish(payload): + """记录删除事件。""" + calls.append(("event", payload["subscribe_id"], payload)) + + async def report(payload): + """记录异步统计并按需返回未确认或抛错。""" + calls.append(("report", payload)) + if error: + raise error + return result + + return DeleteSubscribeCommand( + repository=_Repository(candidate, calls), + unit_of_work=_UnitOfWork(calls), + publish_deleted=publish, + report_deleted=report, + outbox=outbox, + ) + + @pytest.mark.asyncio async def test_owner_delete_commits_before_event_and_report(): """owner 删除成功时必须先提交，再按原顺序发送事件和上报。""" @@ -265,6 +287,64 @@ async def test_delete_outbox_stage_failure_rolls_back_business_delete(): ] +@pytest.mark.asyncio +async def test_async_reporter_completes_report_intent_only_after_confirmation(): + """异步 reporter 确认成功后才允许收口统计 intent。""" + calls = [] + command = _async_report_command(_candidate(), calls, outbox=_Outbox(calls)) + + 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", + "event", "outbox_complete", "report", "outbox_complete", + ] + + +@pytest.mark.asyncio +async def test_async_reporter_false_keeps_report_intent_pending(): + """异步 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 [call[0] for call in calls] == [ + "get", "delete", "outbox_stage", "outbox_stage", "commit", + "event", "outbox_complete", "report", + ] + + +@pytest.mark.asyncio +async def test_async_reporter_error_keeps_report_intent_pending(): + """异步 reporter 异常时必须保留待重试统计 intent。""" + calls = [] + command = _async_report_command( + _candidate(), + calls, + error=RuntimeError("remote failed"), + outbox=_Outbox(calls), + ) + + with pytest.raises(RuntimeError, match="remote failed"): + await command.execute( + 7, + SubscribeDeletionActor(username="alice", is_superuser=False), + ) + + assert [call[0] for call in calls] == [ + "get", "delete", "outbox_stage", "outbox_stage", "commit", + "event", "outbox_complete", "report", + ] + + @pytest.mark.asyncio async def test_repository_candidate_uses_loaded_orm_snapshot(monkeypatch): """DB 适配器只向应用层暴露权限字段和完整列快照。"""