From 6e295d5d2fe343374691511e2ba86a0fdb158996 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 4 Oct 2026 00:15:28 +0800 Subject: [PATCH] fix(subscribe): preserve accepted searches for paused subscriptions --- app/chain/_music.py | 5 +- app/chain/subscribe/policy.py | 7 +- app/chain/subscribe/search.py | 8 +- app/chain/subscribe/searchtask.py | 5 +- docs/subscribe-lifecycle.md | 5 + .../architecture/complexity-baseline.json | 2 +- tests/test_music_subscribe.py | 97 +++++++++- tests/test_subscription_paused_download.py | 174 ++++++++++++++++++ tests/test_subscription_search_governance.py | 81 +++++++- 9 files changed, 361 insertions(+), 23 deletions(-) create mode 100644 tests/test_subscription_paused_download.py diff --git a/app/chain/_music.py b/app/chain/_music.py index e3ac64f00..c7e285b8e 100644 --- a/app/chain/_music.py +++ b/app/chain/_music.py @@ -491,7 +491,10 @@ class MusicSubscribeMixin: if current_subscribe is None: logger.info(f"音乐订阅 {subscribe.id} 已删除，放弃本轮下载提交") return - if current_subscribe.state == "S": + # 与影视订阅一致，暂停不撤销已接纳的搜索，自动匹配仍不能为暂停订阅提交下载。 + if current_subscribe.state == "S" and not ( + execution_context and execution_context.lease.operation == "search" + ): logger.info(f"音乐订阅 {current_subscribe.name} 已暂停，放弃本轮下载提交") return if self._SubscribeChain__candidate_contract_changed(subscribe, current_subscribe): diff --git a/app/chain/subscribe/policy.py b/app/chain/subscribe/policy.py index d56679cf5..fd082796e 100644 --- a/app/chain/subscribe/policy.py +++ b/app/chain/subscribe/policy.py @@ -302,7 +302,10 @@ class SubscribePolicyOwner(_SubscribePriorityPolicyOwner): if current is None: logger.info(f"订阅 {subscribe.id} 已删除，放弃本轮下载提交") return [], no_exists - if current.state == "S": + # 已接纳的 Search 可完成本次下载；无搜索上下文的 Match 等路径仍遵守暂停状态。 + if current.state == "S" and not ( + execution_context and execution_context.lease.operation == "search" + ): logger.info(f"订阅 {current.name} 已暂停，放弃本轮下载提交") return [], no_exists if self._SubscribeChain__candidate_contract_changed(subscribe, current): @@ -482,7 +485,7 @@ class SubscribePolicyOwner(_SubscribePriorityPolicyOwner): N: New（新建，未处理） R: Resolved（订阅中） P: Pending（待定，信息待进一步更新，允许搜索，不允许完成） - S: Suspended（暂停，订阅不参与任何动作，暂时停止处理） + S: Suspended（暂停自动调度，已接纳搜索及指定订阅补搜可继续执行） :return: 需要查询的状态列表（多个状态用逗号分隔） """ # 如果状态是 R 或 P，则视为一起搜索，返回 R,P 作为查询条件 diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index ad9ab2803..5ee189a18 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -165,6 +165,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): 执行订阅搜索。 scheduled_interval 仅供定时调度传入系统间隔；手动和指定目标搜索不受周期限制。 + 指定目标可对暂停订阅搜索一次，已接纳的搜索不因暂停中断，也不会仅因搜索而恢复订阅。 """ return self._execute_search( sid=sid, @@ -281,9 +282,8 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): reason: Optional[str] = "not_eligible" try: current = self.subscription_repository.get(subscribe.id) - if current is None or current.state == "S": - if current and current.state == "S": - logger.debug(f"订阅《{current.name}》已暂停，本次没有搜索") + # 与持久队列保持一致，暂停不取消本轮已选取的搜索。 + if current is None: continue processed_result = self._process_search_subscription( current, @@ -613,7 +613,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): state: Optional[str], scheduled_interval: Optional[int] = None, ) -> list[SubscriptionSnapshot]: - """按单条、指定批次或状态读取本轮搜索订阅。""" + """指定目标允许暂停订阅补搜；常规调度只从请求的活动状态选取新任务。""" repository = self.subscription_repository if sid: subscribe = repository.get(sid) diff --git a/app/chain/subscribe/searchtask.py b/app/chain/subscribe/searchtask.py index 2768062fb..dc80d36f7 100644 --- a/app/chain/subscribe/searchtask.py +++ b/app/chain/subscribe/searchtask.py @@ -198,10 +198,7 @@ class SubscriptionSearchTaskRunner: if current is None: self._finish_missing_subscription(task_id, lease_token) return None - if current.state == "S": - skip_search_task(self.queue, self.task, "订阅已暂停，这次没有搜索") - self.summary.record("skipped", "paused") - return None + # 暂停只影响后续自动调度；持久任务已接纳的一次搜索继续执行，停止须显式取消任务。 self.searchchain.configure_subscription_site_budget( SubscriptionSiteBudget( repository=self.queue, diff --git a/docs/subscribe-lifecycle.md b/docs/subscribe-lifecycle.md index b0e007f7f..fcb09b1c3 100644 --- a/docs/subscribe-lifecycle.md +++ b/docs/subscribe-lifecycle.md @@ -87,6 +87,11 @@ MoviePilot 当前以媒体类型和洗版方式组合出常用订阅模式。普 ### 自动搜索与站点恢复 +暂停订阅不会被常规自动调度选取，但可以通过指定订阅搜索执行一次补搜。暂停不取消已经入队或 +正在执行的搜索，包括等待站点恢复的任务；需要停止本次执行时，应取消对应搜索批次。 +搜索及下载提交沿用同一执行上下文，不会为了补搜临时启用订阅，也不会在搜索无结果时恢复订阅。 +下载成功后插件可以根据关联订阅的下载事件恢复订阅，已经完成的订阅仍按正常完成规则收口。 + `last_search` 记录完整搜索轮次的开始时间，供订阅独立周期或系统周期判断下一次搜索是否到期。 因站点暂时不可用而仅补查剩余站点时，不刷新此时间，避免坏站点持续推迟正常站点的新一轮搜索。 同一轮恢复保留待搜索站点；经过到期筛选的新周期会把尚未运行的恢复任务重新覆盖当前启用站点。 diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index da1f3eb84..59461c41c 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -59,7 +59,7 @@ "app/chain/subscribe/query.py:SubscribeQueryOwner.follow": 19, "app/chain/subscribe/query.py:SubscribeQueryOwner.subscribe_files_info": 24, "app/chain/subscribe/refresh.py:SubscribeRefreshOwner.backfill_existing_episodes": 16, - "app/chain/subscribe/search.py:_SubscribeSearchQueueCoordinator._execute_inline_search": 17, + "app/chain/subscribe/search.py:_SubscribeSearchQueueCoordinator._execute_inline_search": 16, "app/chain/transfer/execution.py:TransferExecutionOwner._TransferChain__perform_transfer": 31, "app/chain/transfer/filter.py:FileFilterMixin._restore_music_download_context": 24, "app/chain/transfer/music.py:_release_groups": 16, diff --git a/tests/test_music_subscribe.py b/tests/test_music_subscribe.py index 85f019f19..57767e7be 100644 --- a/tests/test_music_subscribe.py +++ b/tests/test_music_subscribe.py @@ -151,12 +151,12 @@ def _configure_subscription_write(chain, repository) -> None: chain.sync_subscription_mutation_scope = mutation_scope -def _execution_context(*, cancelled=None) -> SubscriptionExecutionContext: +def _execution_context(*, cancelled=None, operation="search") -> SubscriptionExecutionContext: """构造音乐订阅使用的独立执行上下文。""" admission = SubscriptionExecutionAdmission() lease = admission.try_acquire( subscription_id=7, - operation="search", + operation=operation, ttl_seconds=60, ) assert lease is not None @@ -330,7 +330,7 @@ def test_music_download_marks_shared_execution_context_before_side_effect(): def test_music_download_rechecks_paused_state_before_submission(): - """候选准备后暂停的音乐订阅不得进入下载器。""" + """没有搜索执行上下文时，暂停的音乐订阅不得提交下载。""" subscribe = _subscribe() paused = _subscribe(state="S") repository = Mock() @@ -343,12 +343,101 @@ def test_music_download_rechecks_paused_state_before_submission(): subscribe, _music_info(), [Context()], - execution_context=_execution_context(), ) download_chain.assert_not_called() +@pytest.mark.parametrize("prepared_state", ["R", "S"]) +def test_music_accepted_search_downloads_paused_subscription(prepared_state): + """已接纳搜索和指定暂停订阅补搜能提交候选，不主动恢复订阅状态。""" + prepared = _subscribe(state=prepared_state) + current = _subscribe(state="S") + target = _music_info() + candidate = Context( + torrent_info=TorrentInfo(title="周杰伦 - 晴天 FLAC", category=MediaType.MUSIC.value), + meta_info=MetaMusic.from_music_info(target), + media_info=target, + ) + execution = _execution_context() + repository = Mock() + repository.get.return_value = current + chain = SubscribeChain() + chain.subscription_repository = repository + chain.finish_subscribe_or_not = Mock() + download_chain = Mock() + + def submit(**kwargs): + """模拟外部提交，并验证策略传入真实搜索取消和副作用信号。""" + assert kwargs["contexts"] == [candidate] + assert kwargs["governance"].cancelled() is False + kwargs["governance"].mark_started() + return [candidate], None + + download_chain.batch_download.side_effect = submit + + with patch("app.chain._music.DownloadChain", return_value=download_chain): + chain._download_music_subscribe(prepared, target, [candidate], execution_context=execution) + + download_chain.batch_download.assert_called_once() + assert execution.download_started is True + chain.finish_subscribe_or_not.assert_called_once() + assert chain.finish_subscribe_or_not.call_args.kwargs["subscribe"] is current + assert chain.finish_subscribe_or_not.call_args.kwargs["downloads"] == [candidate] + repository.update.assert_not_called() + assert current.state == "S" + assert execution.admission.release(execution.lease) is True + + +def test_music_paused_match_cannot_submit_download(): + """匹配执行上下文不能使用搜索任务的暂停放行规则。""" + repository = Mock() + repository.get.return_value = _subscribe(state="S") + chain = SubscribeChain() + chain.subscription_repository = repository + chain.finish_subscribe_or_not = Mock() + execution = _execution_context(operation="match") + + with patch("app.chain._music.DownloadChain") as download_chain: + chain._download_music_subscribe( + _subscribe(), _music_info(), [Context()], execution_context=execution, + ) + + download_chain.assert_not_called() + chain.finish_subscribe_or_not.assert_not_called() + assert execution.download_started is False + assert execution.admission.release(execution.lease) is True + + +@pytest.mark.parametrize("cancel_at", ["entry", "preparing"]) +def test_music_paused_search_still_cancels_before_submission(cancel_at): + """暂停搜索的放行不能绕过入口或资源准备阶段的取消。""" + cancelled = [cancel_at == "entry"] + execution = _execution_context(cancelled=lambda: cancelled[0]) + + def on_phase(phase, _site_id): + if phase == cancel_at: + cancelled[0] = True + + execution.phase_changed = on_phase + repository = Mock() + repository.get.return_value = _subscribe(state="S") + chain = SubscribeChain() + chain.subscription_repository = repository + chain.finish_subscribe_or_not = Mock() + + with patch("app.chain._music.DownloadChain") as download_chain: + with pytest.raises(SubscriptionSearchCancelled): + chain._download_music_subscribe( + _subscribe(), _music_info(), [Context()], execution_context=execution, + ) + + download_chain.assert_not_called() + chain.finish_subscribe_or_not.assert_not_called() + assert execution.download_started is False + assert execution.admission.release(execution.lease) is True + + def test_music_subscribe_filters_declared_bitrate_and_format(): """音乐订阅应按规范化格式和最低码率过滤站点资源。""" subscribe = _subscribe(audio_format="MP3", min_bitrate=320000) diff --git a/tests/test_subscription_paused_download.py b/tests/test_subscription_paused_download.py new file mode 100644 index 000000000..c3d41283a --- /dev/null +++ b/tests/test_subscription_paused_download.py @@ -0,0 +1,174 @@ +"""暂停订阅中已接纳搜索的真实下载边界回归。""" + +from dataclasses import replace +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +import app.chain.download.batch as download_batch +import app.chain.download.selection as download_selection +from app.application.subscription.contract import SubscriptionSnapshot +from app.application.subscription.execution import ( + SubscriptionExecutionAdmission, + SubscriptionExecutionContext, +) +from app.chain.subscribe import policy as subscribe_policy +from app.chain.subscribe.facade import SubscribeChain +from app.schemas.mediaserver import NotExistMediaInfo +from app.schemas.types import MediaSource, MediaType +from tests.test_subscription_download_governance import ( + _context, + _download_chain, +) +from tests.test_subscription_download_governance import ( + _submission_dependencies as _submission_dependencies, +) + +pytestmark = pytest.mark.usefixtures("_submission_dependencies") + + +@pytest.fixture +def download_scope(monkeypatch): + """保留订阅策略、候选选择与单资源提交，只隔离存储和外部下载器。""" + current = SubscriptionSnapshot( + id=7, + name="Demo Show", + type=MediaType.TV.value, + media_source=MediaSource.TMDB, + media_id="77", + season=1, + total_episode=1, + state="S", + save_path="/downloads", + downloader="qb", + ) + context = _context() + missing = {"tmdb:77": {1: NotExistMediaInfo(season=1, episodes=[1], total_episode=1)}} + download = _download_chain() + download.download_torrent = MagicMock(return_value=(b"torrent", None, [1])) + download._active_download_failure_fingerprints = MagicMock(return_value={}) + download.download = MagicMock(return_value=("qb", "hash-paused-search", "Original", "accepted")) + chain = SubscribeChain.__new__(SubscribeChain) + chain.subscription_repository = SimpleNamespace(get=MagicMock(return_value=current)) + chain.check_and_handle_existing_media = MagicMock(return_value=(False, missing)) + helper = SimpleNamespace(sort_torrents=lambda contexts: contexts) + monkeypatch.setattr(download_selection, "_new_torrent_helper", lambda: helper) + monkeypatch.setattr(download_batch, "runtime_stop_state", SimpleNamespace(is_system_stopped=False)) + monkeypatch.setattr(download_selection, "runtime_stop_state", SimpleNamespace(is_system_stopped=False)) + monkeypatch.setattr(subscribe_policy, "DownloadChain", lambda: download) + return SimpleNamespace( + current=current, + context=context, + missing=missing, + download=download, + chain=chain, + ) + + +def _execution(operation="search", cancelled=lambda: False): + """固定执行时钟，保留真实租约、取消与副作用阶段信号。""" + admission = SubscriptionExecutionAdmission(clock=lambda: 0) + lease = admission.try_acquire(subscription_id=7, operation=operation, ttl_seconds=60) + assert lease is not None + return SubscriptionExecutionContext( + lease=lease, + admission=admission, + task_id="search-task-7", + cancel_requested=cancelled, + ) + + +def _submit(scope, prepared, execution_context): + return scope.chain._SubscribeChain__download_best_version_with_full_pack_first( + contexts=[scope.context], + no_exists={"tmdb:77": {1: NotExistMediaInfo(season=1, episodes=[1], total_episode=1)}}, + subscribe=prepared, + mediakey="tmdb:77", + execution_context=execution_context, + ) + + +@pytest.mark.parametrize("prepared_state", ["R", "S"]) +def test_accepted_search_downloads_while_subscription_is_paused(download_scope, prepared_state): + """搜索中途暂停和指定暂停订阅补搜均能提交，不主动恢复订阅状态。""" + scope = download_scope + execution = _execution() + + downloads, remaining = _submit(scope, replace(scope.current, state=prepared_state), execution) + + assert downloads == [scope.context] + assert remaining == {} + scope.download.download.assert_called_once() + assert scope.download.download.call_args.kwargs["downloader"] == "qb" + assert scope.download.download.call_args.kwargs["content"] == b"torrent" + scope.download._settle_download_success.assert_called_once() + assert execution.download_started is True + assert scope.current.state == "S" + assert execution.admission.release(execution.lease) is True + + +@pytest.mark.parametrize("operation", [None, "match"]) +def test_paused_subscription_blocks_download_without_search_execution(download_scope, operation): + """无搜索上下文和自动匹配不能借暂停订阅创建下载任务。""" + scope = download_scope + execution = _execution(operation) if operation else None + + downloads, remaining = _submit(scope, scope.current, execution) + + assert downloads == [] + assert remaining["tmdb:77"][1].episodes == [1] + scope.chain.check_and_handle_existing_media.assert_not_called() + scope.download.download.assert_not_called() + scope.download._settle_download_success.assert_not_called() + if execution: + assert execution.download_started is False + assert execution.admission.release(execution.lease) is True + + +def test_paused_search_still_honors_cancellation_at_downloader_boundary(download_scope): + """资源准备期间的取消仍在真实下载器调用前生效。""" + scope = download_scope + cancelled = [False] + execution = _execution(cancelled=lambda: cancelled[0]) + + def prepare_torrent(_torrent, **_kwargs): + cancelled[0] = True + return b"torrent", None, [1] + + scope.download.download_torrent.side_effect = prepare_torrent + + downloads, remaining = _submit(scope, scope.current, execution) + + scope.chain.check_and_handle_existing_media.assert_called_once() + scope.download.download_torrent.assert_called_once() + assert downloads == [] + assert remaining["tmdb:77"][1].episodes == [1] + scope.download.download.assert_not_called() + scope.download._settle_download_success.assert_not_called() + assert execution.download_started is False + assert execution.admission.release(execution.lease) is True + + +@pytest.mark.parametrize("change", ["deleted", "filter", "identity"]) +def test_paused_search_discards_invalidated_candidates(download_scope, change): + """暂停搜索的放行不能绕过删除、筛选或媒体身份变化。""" + scope = download_scope + prepared = replace(scope.current, state="R") + current = { + "deleted": None, + "filter": replace(scope.current, quality="2160p"), + "identity": replace(scope.current, media_id="88"), + }[change] + scope.chain.subscription_repository.get.return_value = current + execution = _execution() + + downloads, remaining = _submit(scope, prepared, execution) + + assert downloads == [] + assert remaining["tmdb:77"][1].episodes == [1] + scope.chain.check_and_handle_existing_media.assert_not_called() + scope.download.download.assert_not_called() + scope.download._settle_download_success.assert_not_called() + assert execution.download_started is False + assert execution.admission.release(execution.lease) is True diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index e8b60c6f9..ebd1b1866 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -703,25 +703,92 @@ def test_manual_search_bypasses_new_subscription_edit_wait(tmp_path, monkeypatch assert chain.get_search_batch(batch_id).state == "completed" -def test_paused_subscription_is_skipped_after_admission_refresh(tmp_path, monkeypatch): - """准入后重新读取到暂停订阅时不得开始搜索或伪造完成。""" +def test_accepted_search_continues_after_subscription_is_paused(tmp_path, monkeypatch): + """暂停不等于取消，已接纳的定时搜索继续使用最新订阅快照。""" subscribe = _subscribe(9) paused = replace(subscribe, state="S") chain = _chain(tmp_path, [subscribe]) _make_tasks_ready(monkeypatch) chain.subscription_repository.get = Mock(side_effect=(subscribe, paused)) - process = Mock(side_effect=AssertionError("暂停订阅不应进入 Search")) + process = Mock(return_value=paused) monkeypatch.setattr(chain, "_process_search_subscription", process) with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): batch_id = chain.search(state="R") batch = chain.get_search_batch(batch_id) + process.assert_called_once() + assert process.call_args.args[0] == paused + assert batch.state == "completed" + assert batch.finished_count == 1 + assert batch.skipped_count == 0 + + +@pytest.mark.parametrize("queued", [False, True]) +@pytest.mark.parametrize("manual", [False, True]) +def test_targeted_paused_search_preserves_subscription_state(tmp_path, monkeypatch, queued, manual): + """插件及手动指定补搜均能执行一次，未产生下载时不能擅自恢复订阅。""" + subscribe = replace(_subscribe(9), state="S") + chain = _chain(tmp_path, [subscribe]) + if not queued: + del chain.subscription_search_repository + _make_tasks_ready(monkeypatch) + process = Mock(return_value=subscribe) + reset = Mock(side_effect=AssertionError("补搜不能自行恢复暂停订阅")) + monkeypatch.setattr(chain, "_process_search_subscription", process) + monkeypatch.setattr(chain, "_SubscribeChain__apply_subscribe_update", reset) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(sid=subscribe.id, manual=manual) + + process.assert_called_once() + assert process.call_args.kwargs["execution_context"].lease.operation == "search" + assert chain.subscription_repository.get(subscribe.id).state == "S" + reset.assert_not_called() + if queued: + assert chain.get_search_batch(batch_id).state == "completed" + + +def test_scheduled_search_does_not_enqueue_paused_subscription(tmp_path, monkeypatch): + """暂停订阅不进入常规状态选取，放行已有任务不能生成新的自动搜索。""" + subscribe = replace(_subscribe(9), state="S") + chain = _chain(tmp_path, [subscribe]) + chain.subscription_repository.list = Mock( + side_effect=lambda states: [subscribe] if subscribe.state in states.split(",") else [], + ) + process = Mock(side_effect=AssertionError("暂停订阅不应产生新自动任务")) + monkeypatch.setattr(chain, "_process_search_subscription", process) + + batch_id = chain.search(state="R", scheduled_interval=24) + + assert batch_id is None + chain.subscription_repository.list.assert_called_once_with("R") + assert chain.subscription_search_repository.claim_next(owner="test") is None process.assert_not_called() - assert batch.state == "skipped" - assert batch.finished_count == 0 - assert batch.skipped_count == 1 - assert batch.last_error == "订阅已暂停，这次没有搜索" + + +@pytest.mark.parametrize("cancelled", [False, True]) +def test_paused_search_survives_queue_reopen_unless_cancelled(tmp_path, monkeypatch, cancelled): + """暂停后的持久任务跨消费者恢复继续执行，显式取消仍阻止搜索。""" + subscribe = _subscribe(9) + chain = _chain(tmp_path, [subscribe]) + _make_tasks_ready(monkeypatch) + with chain._search_queue_lock: + batch_id = chain.search(sid=subscribe.id) + assert chain.get_search_batch(batch_id).state == "queued" + if cancelled: + assert chain.cancel_search_batch(batch_id) + + paused = replace(subscribe, state="S") + resumed = _chain(tmp_path, [paused]) + process = Mock(return_value=paused) + monkeypatch.setattr(resumed, "_process_search_subscription", process) + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + resumed.resume_search_queue() + + assert process.call_count == (0 if cancelled else 1) + assert resumed.get_search_batch(batch_id).state == ("cancelled" if cancelled else "completed") + assert resumed.subscription_repository.get(subscribe.id).state == "S" def test_cleanup_failures_cannot_leak_subscription_admission(tmp_path, monkeypatch):