diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py index 97563fe0e..0c3774c1c 100644 --- a/app/application/subscription/execution.py +++ b/app/application/subscription/execution.py @@ -200,6 +200,16 @@ class SubscriptionSearchRepository(Protocol): """释放尚未完成的任务租约，供停止或取消后恢复。""" ... + def defer_task( + self, + *, + task_id: str, + lease_token: str, + available_at: str, + ) -> bool: + """把临时站点预算冲突任务退回队列，并设置下一次领取时间。""" + ... + def is_cancel_requested(self, task_id: str) -> bool: """判断任务或所属批次是否已请求取消。""" ... diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index a134c8b45..83c8e1f46 100644 --- a/app/application/subscription/sitebudget.py +++ b/app/application/subscription/sitebudget.py @@ -15,11 +15,29 @@ class SubscriptionSearchCancelled(RuntimeError): """表示订阅搜索在可取消预算等待点终止。""" +@dataclass(frozen=True, slots=True) +class SubscriptionSiteBudgetDeferral: + """记录一次站点预算冲突及该站点最早可再次尝试的时间。""" + + site_id: int + retry_at: str + + +class SubscriptionSearchDeferred(RuntimeError): + """表示订阅搜索未失败，而是应在站点预算可用后重新入队。""" + + def __init__(self, *, retry_at: str, site_ids: tuple[int, ...]) -> None: + """保存队列恢复所需的时间和冲突站点，避免把临时冲突写成错误。""" + super().__init__(f"订阅搜索已延后，站点预算最早可重试：{retry_at}") + self.retry_at = retry_at + self.site_ids = site_ids + + class SubscriptionSiteBudgetUnavailable(RuntimeError): - """表示站点仍处于错误冷却或已有未释放租约。""" + """表示站点预算暂时不可用，调用方应记录为延后而非失败。""" def __init__(self, *, site_id: int, retry_at: str) -> None: - """保存站点和下一次可尝试时间，供批次聚合失败展示。""" + """保存站点和下一次可尝试时间，供订阅队列恢复。""" super().__init__(f"站点 {site_id} 冷却或已有在途搜索，最早可重试：{retry_at}") self.site_id = site_id self.retry_at = retry_at diff --git a/app/application/subscription/status.py b/app/application/subscription/status.py index bcdb143ed..4d51ca58c 100644 --- a/app/application/subscription/status.py +++ b/app/application/subscription/status.py @@ -153,6 +153,8 @@ class SubscriptionExecutionStatusService: state = phase = "cancelling" elif task.state == "running": state = phase = task.phase or "running" + elif task.state == "queued" and task.phase == "waiting_site_budget": + state = phase = "waiting_site_budget" else: state = phase = task.state return SubscriptionExecutionStatus( diff --git a/app/chain/search/contract.py b/app/chain/search/contract.py index daaaa56de..ef3bd7600 100644 --- a/app/chain/search/contract.py +++ b/app/chain/search/contract.py @@ -99,6 +99,8 @@ if TYPE_CHECKING: process: Callable[..., Any] record_subscription_site_budget_failure: Callable[..., Any] consume_subscription_site_budget_failures: Callable[..., Any] + record_subscription_site_budget_deferred: Callable[..., Any] + consume_subscription_site_budget_deferrals: Callable[..., Any] save_last_search_params: Callable[..., Any] search_by_id: Callable[..., Any] search_by_title: Callable[..., Any] diff --git a/app/chain/search/facade.py b/app/chain/search/facade.py index 89992baed..4c718c364 100644 --- a/app/chain/search/facade.py +++ b/app/chain/search/facade.py @@ -4,7 +4,10 @@ import threading from collections.abc import AsyncIterator, Callable from typing import Any, Optional, TypeVar, cast -from app.application.subscription.sitebudget import SubscriptionSiteBudget +from app.application.subscription.sitebudget import ( + SubscriptionSiteBudget, + SubscriptionSiteBudgetDeferral, +) from app.chain.base import ChainBase from app.chain.search.cache import SearchCacheOwner from app.chain.search.media import SearchMediaOwner @@ -49,6 +52,7 @@ class SearchChain(ChainBase): """仅为订阅搜索启用或清除站点预算，不影响其它搜索入口。""" self._subscription_site_budget = budget self._subscription_site_budget_failures: list[str] = [] + self._subscription_site_budget_deferrals: list[SubscriptionSiteBudgetDeferral] = [] self._subscription_site_budget_failure_lock = threading.Lock() def record_subscription_site_budget_failure(self, error: str) -> None: @@ -69,6 +73,27 @@ class SearchChain(ChainBase): self._subscription_site_budget_failures.clear() return failures + def record_subscription_site_budget_deferred( + self, + deferral: SubscriptionSiteBudgetDeferral, + ) -> None: + """线程安全地记录临时站点冲突，供订阅任务重新入队。""" + lock = getattr(self, "_subscription_site_budget_failure_lock", None) + if lock is None: + return + with lock: + self._subscription_site_budget_deferrals.append(deferral) + + def consume_subscription_site_budget_deferrals(self) -> tuple[SubscriptionSiteBudgetDeferral, ...]: + """读取并清空当前订阅搜索积累的站点预算延后事实。""" + lock = getattr(self, "_subscription_site_budget_failure_lock", None) + if lock is None: + return () + with lock: + deferrals = tuple(self._subscription_site_budget_deferrals) + self._subscription_site_budget_deferrals.clear() + return deferrals + # owner descriptor 经类访问后被 mypy 视为普通 Callable；运行时仍需取回原始 # classmethod 函数，才能保持 SearchChain 的直接 MRO 与既有绑定语义。 music_site_keywords = classmethod( # type: ignore[var-annotated] diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index dd564f041..e03a0d623 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -17,6 +17,7 @@ from app.application.site.observation import ( from app.application.site.sites import SitesHelper # pylint: disable=import-error,no-name-in-module from app.application.subscription.sitebudget import ( SubscriptionSiteBudget, + SubscriptionSiteBudgetDeferral, SubscriptionSiteBudgetUnavailable, ) from app.chain.search.contract import _SearchOwnerBase @@ -215,7 +216,12 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): try: claim = budget.acquire(site_id) except SubscriptionSiteBudgetUnavailable as error: - self.record_subscription_site_budget_failure(str(error)) + self.record_subscription_site_budget_deferred( + SubscriptionSiteBudgetDeferral( + site_id=error.site_id, + retry_at=error.retry_at, + ) + ) logger.debug(str(error)) return [] with capture_site_search_observation() as observation: diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index 18ab1f546..ddf8115fe 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -31,7 +31,9 @@ from app.application.subscription.observability import ( from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.sitebudget import ( SubscriptionSearchCancelled, + SubscriptionSearchDeferred, SubscriptionSiteBudget, + SubscriptionSiteBudgetDeferral, ) from app.chain.media import MediaChain from app.chain.search.facade import SearchChain @@ -323,6 +325,12 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): reason = "ttl_timeout" if execution_context.is_expired() else "cancelled" outcome = "failed" if reason == "ttl_timeout" else "cancelled" logger.debug(f"订阅 {subscribe.name} 搜索已在安全边界取消") + except SubscriptionSearchDeferred as deferred: + outcome = "skipped" + reason = "site_budget_deferred" + logger.debug( + f"订阅 {subscribe.name} 站点预算冲突，兼容搜索将在 {deferred.retry_at} 后重试" + ) except Exception as err: outcome = "failed" reason = "error" @@ -610,6 +618,20 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): "requeued" if system_stopped else "cancelled", "system_stop" if system_stopped else "cancelled", ) + except SubscriptionSearchDeferred as deferred: + requeued = queue.defer_task( + task_id=task_id, + lease_token=lease_token, + available_at=deferred.retry_at, + ) + if requeued: + logger.debug( + f"订阅 {subscribe.name} 站点预算冲突，已排队至 {deferred.retry_at} 后重试，" + f"sites={','.join(str(site_id) for site_id in deferred.site_ids)}" + ) + summary.record("requeued", "site_budget_deferred") + else: + logger.debug(f"订阅搜索任务 {task_id} 租约已变化，跳过重复站点预算重排队") except Exception as err: logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True) queue.finish_task( @@ -821,9 +843,12 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): filter_params=self.get_params(subscribe), ) site_budget_failures = searchchain.consume_subscription_site_budget_failures() + site_budget_deferrals = searchchain.consume_subscription_site_budget_deferrals() _ensure_execution_active(execution_context) if not contexts: logger.debug(f"订阅 {subscribe.keyword or subscribe.name} 未搜索到资源") + if not site_budget_failures: + self._raise_site_budget_deferral(site_budget_deferrals, execution_context) self.finish_subscribe_or_not( subscribe=subscribe, meta=meta, @@ -835,6 +860,8 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): matched = self._filter_search_contexts(subscribe, contexts) if not matched: logger.debug(f"订阅 {subscribe.name} 没有符合过滤条件的资源") + if not site_budget_failures: + self._raise_site_budget_deferral(site_budget_deferrals, execution_context) self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo, lefts=no_exists) self._raise_site_budget_failures(site_budget_failures) return subscribe @@ -862,6 +889,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): lefts=lefts, ) self._raise_site_budget_failures(site_budget_failures) + self._raise_site_budget_deferral(site_budget_deferrals, execution_context) return cast(Optional[SubscriptionSnapshot], current) @staticmethod @@ -870,6 +898,18 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): if failures: raise RuntimeError("；".join(failures)) + @staticmethod + def _raise_site_budget_deferral( + deferrals: tuple[SubscriptionSiteBudgetDeferral, ...], + execution_context: Optional[SubscriptionExecutionContext], + ) -> None: + """在没有下载副作用时，将临时站点冲突转换为持久队列延后。""" + if not deferrals or (execution_context and execution_context.download_started): + return + retry_at = min(deferrals, key=lambda item: item.retry_at).retry_at + site_ids = tuple(dict.fromkeys(item.site_id for item in deferrals)) + raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=site_ids) + def _filter_search_contexts( self, subscribe: SubscriptionSnapshot, diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py index c18263dfd..0017ed492 100644 --- a/app/db/adapters/subscriptionsearch.py +++ b/app/db/adapters/subscriptionsearch.py @@ -177,6 +177,22 @@ class TransactionalSubscriptionSearchRepository: ) ) + def defer_task( + self, + *, + task_id: str, + lease_token: str, + available_at: str, + ) -> bool: + """以站点预算的下一次可用时间重新排队任务。""" + return self._write( + lambda repository: repository.defer_task( + task_id=task_id, + lease_token=lease_token, + available_at=available_at, + ) + ) + def is_cancel_requested(self, task_id: str) -> bool: """查询任务或批次的取消请求。""" return self._read(lambda repository: repository.is_cancel_requested(task_id)) @@ -208,11 +224,9 @@ class TransactionalSubscriptionSearchRepository: owner=owner, lease_seconds=lease_seconds, ) - retry_at = ( - record.lease_expires_at - if record.lease_token and not acquired - else record.next_allowed_at - ) or record.next_allowed_at + retry_at = record.next_allowed_at + if not acquired and record.lease_token and record.lease_expires_at: + retry_at = max(retry_at, record.lease_expires_at) return SiteBudgetClaim( site_id=record.site_id, acquired=acquired, diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py index 5ccc7b48f..d35045859 100644 --- a/app/db/oper/subscriptionsearch.py +++ b/app/db/oper/subscriptionsearch.py @@ -363,6 +363,60 @@ class SubscriptionSearchOper(DbOper): self._refresh_batch(task.batch_id, now=now, error=None) return True + def defer_task( + self, + *, + task_id: str, + lease_token: str, + available_at: str, + ) -> bool: + """释放当前租约并在站点预算时间到达后恢复同一任务。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅搜索延后需要调用方提供同步 Session") + task = self._db.execute( + select(SubscriptionSearchTask).where( + SubscriptionSearchTask.task_id == task_id, + SubscriptionSearchTask.state == "running", + SubscriptionSearchTask.lease_token == lease_token, + ) + ).scalars().first() + if task is None: + return False + if bool(task.cancel_requested) or self._batch_cancel_requested(task.batch_id): + return self.finish_task( + task_id=task_id, + lease_token=lease_token, + state="cancelled", + error=None, + ) + now = utc_now_text() + updated = execute_dml( + self._db, + update(SubscriptionSearchTask) + .where( + SubscriptionSearchTask.id == task.id, + SubscriptionSearchTask.state == "running", + SubscriptionSearchTask.lease_token == lease_token, + ) + .values( + state="queued", + phase="waiting_site_budget", + current_site_id=None, + lease_owner=None, + lease_token=None, + lease_expires_at=None, + available_at=available_at, + updated_at=now, + finished_at=None, + last_error=None, + ), + execution_options={"synchronize_session": False}, + ) + if not updated: + return False + self._refresh_batch(task.batch_id, now=now, error=None) + return True + def is_cancel_requested(self, task_id: str) -> bool: """读取任务和批次取消标记。""" if not isinstance(self._db, Session): diff --git a/scripts/validation/subscription_governance_scale.py b/scripts/validation/subscription_governance_scale.py index eaf843d28..99679d8f7 100644 --- a/scripts/validation/subscription_governance_scale.py +++ b/scripts/validation/subscription_governance_scale.py @@ -379,7 +379,7 @@ def _run_site_pressure_case( start_barrier = threading.Barrier(2) owners = ("scale-owner-a", "scale-owner-b") chains: dict[str, SearchChain] = {} - owner_failures: dict[str, list[str]] = {owner: [] for owner in owners} + owner_deferrals: dict[str, list[Any]] = {owner: [] for owner in owners} for owner in owners: controller = SubscriptionSiteBudgetController( @@ -391,22 +391,21 @@ def _run_site_pressure_case( ) chain = object.__new__(SearchChain) chain.configure_subscription_site_budget(controller) - original_record_failure = chain.record_subscription_site_budget_failure + original_record_deferred = chain.record_subscription_site_budget_deferred - def record_failure( - error: str, + def record_deferred( + deferral: Any, *, - original=original_record_failure, + original=original_record_deferred, owner_name=owner, ) -> None: - """保留 wrapper 聚合失败并同步压力轮次观测。""" - original(error) - owner_failures[owner_name].append(error) - if "冷却或已有在途搜索" in error: - boundary.record_budget_rejection() - pressure_event.set() + """保留 wrapper 延后事实并同步压力轮次观测。""" + original(deferral) + owner_deferrals[owner_name].append(deferral) + boundary.record_budget_rejection() + pressure_event.set() - chain.record_subscription_site_budget_failure = record_failure + chain.record_subscription_site_budget_deferred = record_deferred def search_site_torrents(*, _owner=owner, **_kwargs: Any) -> list[str]: """将 SearchChain 的真实站点请求委托给固定边界。""" @@ -453,11 +452,10 @@ def _run_site_pressure_case( thread.join(timeout=_SITE_PRESSURE_SYNC_TIMEOUT) owners_finished = all(not thread.is_alive() for thread in threads) - initial_failures = [ - error + initial_deferrals = [ + deferral for owner in owners - for error in owner_failures[owner] - if "冷却或已有在途搜索" in error + for deferral in owner_deferrals[owner] ] successful_owners = [ owner @@ -469,10 +467,7 @@ def _run_site_pressure_case( for owner in owners if owner in owner_results and not owner_results[owner] - and any( - "冷却或已有在途搜索" in error - for error in owner_failures[owner] - ) + and owner_deferrals[owner] ] successful_owner = successful_owners[0] if len(successful_owners) == 1 else None blocked_owner = blocked_owners[0] if len(blocked_owners) == 1 else None @@ -482,7 +477,7 @@ def _run_site_pressure_case( and boundary.active_at_rejection == 1 and boundary.peak == 1 and boundary.active == 0 - and len(initial_failures) == 1 + and len(initial_deferrals) == 1 and len(successful_owners) == 1 and len(blocked_owners) == 1 and all(count == 1 for count in owner_invocations.values()) @@ -501,8 +496,8 @@ def _run_site_pressure_case( if successful_owner and blocked_owner: winner = chains[successful_owner] loser = chains[blocked_owner] - winner.consume_subscription_site_budget_failures() - loser.consume_subscription_site_budget_failures() + winner.consume_subscription_site_budget_deferrals() + loser.consume_subscription_site_budget_deferrals() if site_id != case.site_count: boundary.set_outcome("success") @@ -552,11 +547,11 @@ def _run_site_pressure_case( page=0, ) cooldown_calls = boundary.calls - calls_before_cooldown - cooldown_failures = loser.consume_subscription_site_budget_failures() + cooldown_deferrals = loser.consume_subscription_site_budget_deferrals() error_cooldown_blocked = bool( not cooldown_result and cooldown_calls == 0 - and cooldown_failures + and cooldown_deferrals ) error_cooldown_persisted = bool( error_record @@ -576,7 +571,7 @@ def _run_site_pressure_case( and error_cooldown_persisted ) ) - duplicate_site_claims_blocked += len(initial_failures) + duplicate_site_claims_blocked += len(initial_deferrals) successful_site_claims_reused += int(success_reused) error_cooldown_claims_blocked += int(error_cooldown_blocked) site_observations.append( @@ -589,7 +584,7 @@ def _run_site_pressure_case( "request_peak": boundary.peak, "request_calls": boundary.calls, "request_active_at_rejection": boundary.active_at_rejection, - "budget_rejections": len(initial_failures), + "budget_rejections": len(initial_deferrals), "success_reused": success_reused, "error_observed": error_observed, "error_cooldown_blocked": error_cooldown_blocked, diff --git a/tests/test_subscription_execution_status.py b/tests/test_subscription_execution_status.py index 625933f35..54a0a474e 100644 --- a/tests/test_subscription_execution_status.py +++ b/tests/test_subscription_execution_status.py @@ -84,6 +84,19 @@ def test_execution_status_exposes_site_wait_and_cancel_capability(): assert statuses[1].can_cancel is True +def test_execution_status_exposes_queued_site_wait_without_error(): + """重新入队的站点预算冲突应显示等待状态而不是失败。""" + repository = _Repository() + repository.tasks[2] = _task(2, state="queued", phase="waiting_site_budget") + + statuses = asyncio.run(SubscriptionExecutionStatusService(repository).for_subscriptions((2,))) + + assert statuses[2].state == "waiting_site_budget" + assert statuses[2].phase == "waiting_site_budget" + assert statuses[2].error is None + assert statuses[2].can_cancel is True + + def test_failed_search_exposes_safe_error(): """搜索失败文本必须压平且不暴露内部错误细节。""" repository = _Repository() diff --git a/tests/test_subscription_governance_joint.py b/tests/test_subscription_governance_joint.py index 4fafc9efb..6e592370a 100644 --- a/tests/test_subscription_governance_joint.py +++ b/tests/test_subscription_governance_joint.py @@ -52,8 +52,8 @@ class _MixedBudgetRepository: return True -def test_cooled_site_does_not_block_independent_site_or_hide_batch_failure(): - """慢站点跳过后其它站点仍完成，调用方同时收到可聚合失败。""" +def test_cooled_site_does_not_block_independent_site_or_mark_batch_failed(): + """冷却站点延后后其它站点仍完成，调用方只收到可重新入队的事实。""" repository = _MixedBudgetRepository() budget = SubscriptionSiteBudget( repository=repository, @@ -91,7 +91,9 @@ def test_cooled_site_does_not_block_independent_site_or_hide_batch_failure(): assert results == [2] assert repository.finished_sites == [2] + deferrals = chain.consume_subscription_site_budget_deferrals() + assert len(deferrals) == 1 + assert deferrals[0].site_id == 1 failures = chain.consume_subscription_site_budget_failures() - assert len(failures) == 1 - assert "站点 1" in failures[0] + assert failures == () assert progress.values[-1] == 100 diff --git a/tests/test_subscription_governance_scale.py b/tests/test_subscription_governance_scale.py index 921781b0e..aec06b778 100644 --- a/tests/test_subscription_governance_scale.py +++ b/tests/test_subscription_governance_scale.py @@ -179,22 +179,21 @@ def test_scale_validator_rejects_fake_site_pressure(monkeypatch, tmp_path, mutat def test_scale_validator_rejects_unfinished_site_wrapper(monkeypatch, tmp_path): - """失败已登记但 wrapper 尚未返回时，压力门禁必须拒绝并暴露未收口 owner。""" + """延后已登记但 wrapper 尚未返回时，压力门禁必须拒绝并暴露未收口 owner。""" original_wrapper = SearchChain._search_site_torrents_with_budget release_stalled = threading.Event() threads_before = set(threading.enumerate()) def stall_after_rejection(self, *, site, keyword, mtype, page): """模拟预算拒绝已记录、调用方却未取得返回值的挂起路径。""" - original_record_failure = self.record_subscription_site_budget_failure + original_record_deferred = self.record_subscription_site_budget_deferred - def record_failure_and_stall(error: str) -> None: + def record_deferred_and_stall(deferral) -> None: """在拒绝已登记后阻塞原始 wrapper 的返回。""" - original_record_failure(error) - if "冷却或已有在途搜索" in error: - release_stalled.wait() + original_record_deferred(deferral) + release_stalled.wait() - self.record_subscription_site_budget_failure = record_failure_and_stall + self.record_subscription_site_budget_deferred = record_deferred_and_stall try: return original_wrapper( self, @@ -204,7 +203,7 @@ def test_scale_validator_rejects_unfinished_site_wrapper(monkeypatch, tmp_path): page=page, ) finally: - self.record_subscription_site_budget_failure = original_record_failure + self.record_subscription_site_budget_deferred = original_record_deferred monkeypatch.setattr( SearchChain, diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index ad7a42085..e96847556 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -13,13 +13,17 @@ from sqlalchemy.orm import sessionmaker from app.application.site.observation import report_site_search_outcome from app.application.subscription.contract import SubscriptionSnapshot from app.application.subscription.execution import SubscriptionExecutionAdmission -from app.application.subscription.sitebudget import SubscriptionSearchCancelled +from app.application.subscription.sitebudget import ( + SubscriptionSearchCancelled, + SubscriptionSearchDeferred, +) from app.chain.search.facade import SearchChain from app.chain.subscribe import search as subscribe_search from app.chain.subscribe.facade import SubscribeChain from app.chain.subscribe.search import _search_task_available_at from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository from app.db.base import Base +from app.db.models.subscriptionsearch import SubscriptionSearchTask from app.modules.indexer import IndexerModule from app.schemas.types import MediaType @@ -249,6 +253,37 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke assert chain._subscription_execution_admission.release(lease) is True +def test_site_budget_conflict_requeues_task_without_batch_failure(tmp_path, monkeypatch): + """站点预算冲突应自动排队重试，不能把订阅批次置为失败。""" + subscribe = _subscribe(5) + chain = _chain(tmp_path, [subscribe]) + _make_tasks_ready(monkeypatch) + retry_at = (datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat(timespec="seconds") + monkeypatch.setattr( + chain, + "_process_search_subscription", + Mock(side_effect=SubscriptionSearchDeferred(retry_at=retry_at, site_ids=(31,))), + ) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(state="R") + + batch = chain.get_search_batch(batch_id) + assert batch.state == "queued" + assert batch.finished_count == 0 + assert batch.failed_count == 0 + assert batch.last_error is None + with chain.subscription_search_repository._session_factory() as session: + task = session.query(SubscriptionSearchTask).filter_by( + subscription_id=subscribe.id, + ).one() + assert task.state == "queued" + assert task.phase == "waiting_site_budget" + assert task.available_at == retry_at + assert task.last_error is None + assert chain.subscription_search_repository.claim_next(owner="worker-after-retry") is None + + def test_search_logs_one_bounded_start_and_finish_summary(tmp_path, monkeypatch): """Search INFO 只保留轮次摘要，并携带任务终态与耗时字段。""" subscribes = [_subscribe(50), _subscribe(51)] diff --git a/tests/test_subscription_search_queue.py b/tests/test_subscription_search_queue.py index 1dcf2d06d..62fc76cb5 100644 --- a/tests/test_subscription_search_queue.py +++ b/tests/test_subscription_search_queue.py @@ -132,6 +132,47 @@ def test_search_queue_phase_update_requires_current_lease(tmp_path): ) is True +def test_search_queue_defers_site_budget_conflict_until_retry_time(tmp_path): + """站点预算冲突应释放任务租约并保留同一任务等待后续恢复。""" + repository, engine = _repository(tmp_path) + enqueued = repository.enqueue(subscription_ids=(31,), source="fallback", priority=10) + running = repository.claim_next(owner="worker-a") + retry_at = (datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat(timespec="seconds") + + assert repository.defer_task( + task_id=running.task_id, + lease_token=running.lease_token, + available_at=retry_at, + ) is True + + batch = repository.get_batch(enqueued.batch.batch_id) + assert batch.state == "queued" + assert batch.finished_count == 0 + assert batch.failed_count == 0 + assert repository.claim_next(owner="worker-b") is None + + with Session(engine) as session: + task = session.execute( + select(SubscriptionSearchTask).where( + SubscriptionSearchTask.task_id == running.task_id + ) + ).scalar_one() + assert task.state == "queued" + assert task.phase == "waiting_site_budget" + assert task.available_at == retry_at + assert task.last_error is None + session.execute( + update(SubscriptionSearchTask) + .where(SubscriptionSearchTask.task_id == running.task_id) + .values(available_at="1970-01-01T00:00:00+00:00") + ) + session.commit() + + recovered = repository.claim_next(owner="worker-c") + assert recovered.task_id == running.task_id + assert recovered.attempt_count == 2 + + def test_search_queue_cancel_finishes_queued_and_running_tasks(tmp_path): """取消立即终止未发请求任务，运行中任务在租约边界收口。""" repository, engine = _repository(tmp_path) diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index f67ae8191..c53d77392 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -243,7 +243,7 @@ def test_skipped_search_releases_budget_without_external_interval(): def test_search_provider_reports_cooled_site_without_blocking_other_results(): - """错误冷却中的站点返回空页并记录聚合失败，而非阻塞 provider。""" + """错误冷却中的站点返回空页并记录延后，而非阻塞 provider 或制造失败。""" repository = _WaitingRepository() metrics = SubscriptionSiteBudgetMetrics() budget = SubscriptionSiteBudget( @@ -265,9 +265,12 @@ def test_search_provider_reports_cooled_site_without_blocking_other_results(): ) assert result == [] + deferrals = chain.consume_subscription_site_budget_deferrals() + assert len(deferrals) == 1 + assert deferrals[0].site_id == 11 + assert deferrals[0].retry_at failures = chain.consume_subscription_site_budget_failures() - assert len(failures) == 1 - assert "站点 11" in failures[0] + assert failures == () snapshot = metrics.snapshot() assert snapshot.request_count == 0 assert snapshot.cooldown_skip_count == 1