fix: await durable subscribe deletion reports

This commit is contained in:
jxxghp
2026-08-23 00:07:52 +08:00
parent 8171e16f8d
commit 3c8cb513bb
5 changed files with 106 additions and 3 deletions
+20
View File
@@ -662,6 +662,15 @@ class MoviePilotServerHelper:
""" """
return cls._post_json(cls._server_url(cls._SUBSCRIBE_DONE_PATH), payload, timeout=5) 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 @classmethod
def subscribe_report(cls, subscribes: List[Dict[str, Any]]): def subscribe_report(cls, subscribes: List[Dict[str, Any]]):
""" """
@@ -874,6 +883,17 @@ class MoviePilotServerHelper:
res = cls.subscribe_done(payload) res = cls.subscribe_done(payload)
return bool(res is not None and res.status_code == 200) 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 @classmethod
def sub_reg_async(cls, sub: dict) -> bool: def sub_reg_async(cls, sub: dict) -> bool:
""" """
+1 -1
View File
@@ -62,7 +62,7 @@ def get_delete_subscribe_command(
repository=cast(SubscribeDeletionRepository, repository_port), repository=cast(SubscribeDeletionRepository, repository_port),
unit_of_work=cast(DeleteUnitOfWork, unit_of_work), unit_of_work=cast(DeleteUnitOfWork, unit_of_work),
publish_deleted=_publish_subscribe_deleted, publish_deleted=_publish_subscribe_deleted,
report_deleted=MoviePilotServerHelper.sub_done_async, report_deleted=MoviePilotServerHelper.async_sub_done,
outbox=outbox, outbox=outbox,
) )
+4 -1
View File
@@ -3,6 +3,7 @@
from contextlib import AbstractAsyncContextManager from contextlib import AbstractAsyncContextManager
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timezone from datetime import datetime, timezone
import inspect
from typing import Any, Awaitable, Callable, Mapping, Protocol, cast from typing import Any, Awaitable, Callable, Mapping, Protocol, cast
from uuid import uuid4 from uuid import uuid4
@@ -55,7 +56,7 @@ class AsyncUnitOfWork(Protocol):
SubscribeDeletedPublisher = Callable[[dict[str, Any]], Awaitable[None]] SubscribeDeletedPublisher = Callable[[dict[str, Any]], Awaitable[None]]
SubscribeDeletedReporter = Callable[[Mapping[str, object]], object] SubscribeDeletedReporter = Callable[[Mapping[str, object]], object | Awaitable[object]]
class DeleteSubscribeCommand: class DeleteSubscribeCommand:
@@ -134,6 +135,8 @@ class DeleteSubscribeCommand:
# 上报适配器会自行白名单过滤公开字段;传完整删除前快照可保留音乐实体维度, # 上报适配器会自行白名单过滤公开字段;传完整删除前快照可保留音乐实体维度,
# 避免 Agent 与 API 入口收敛后丢失 music_type / total_tracks。 # 避免 Agent 与 API 入口收敛后丢失 music_type / total_tracks。
report_result = self._report_deleted(dict(candidate.event_payload)) report_result = self._report_deleted(dict(candidate.event_payload))
if inspect.isawaitable(report_result):
report_result = await report_result
if report_result is False: if report_result is False:
raise RuntimeError("订阅删除统计上报未确认") raise RuntimeError("订阅删除统计上报未确认")
if self._outbox: if self._outbox:
+1 -1
View File
@@ -145,7 +145,7 @@ async def delete_subscribe_scope():
repository=SubscribeOper(session), repository=SubscribeOper(session),
unit_of_work=SqlAlchemyAsyncUnitOfWork(session), unit_of_work=SqlAlchemyAsyncUnitOfWork(session),
publish_deleted=_publish_deleted, publish_deleted=_publish_deleted,
report_deleted=MoviePilotServerHelper.sub_done_async, report_deleted=MoviePilotServerHelper.async_sub_done,
outbox=SqlAlchemyAsyncOutboxStager(session), outbox=SqlAlchemyAsyncOutboxStager(session),
) )
+80
View File
@@ -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 @pytest.mark.asyncio
async def test_owner_delete_commits_before_event_and_report(): async def test_owner_delete_commits_before_event_and_report():
"""owner 删除成功时必须先提交,再按原顺序发送事件和上报。""" """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 @pytest.mark.asyncio
async def test_repository_candidate_uses_loaded_orm_snapshot(monkeypatch): async def test_repository_candidate_uses_loaded_orm_snapshot(monkeypatch):
"""DB 适配器只向应用层暴露权限字段和完整列快照。""" """DB 适配器只向应用层暴露权限字段和完整列快照。"""