From 54cc3d368d3ca43d3f4eb25f28dadb46f1c8be98 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 5 Sep 2026 18:09:25 +0800 Subject: [PATCH] refactor(subscribe): isolate queued search execution --- app/application/subscription/execution.py | 11 + app/chain/subscribe/contract.py | 1 + app/chain/subscribe/notify.py | 32 +- app/chain/subscribe/search.py | 317 +++-------------- app/chain/subscribe/searchtask.py | 330 ++++++++++++++++++ app/db/adapters/subscriptionsearch.py | 90 ++++- app/startup/composition/runtime.py | 5 +- docs/architecture-overview.md | 4 +- docs/architecture/optimization-checklist.md | 2 +- .../architecture/dependency-baseline.json | 21 +- tests/test_subscription_search_queue.py | 40 +++ 11 files changed, 548 insertions(+), 305 deletions(-) create mode 100644 app/chain/subscribe/searchtask.py diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py index 72a815b15..e35d43284 100644 --- a/app/application/subscription/execution.py +++ b/app/application/subscription/execution.py @@ -206,6 +206,17 @@ class SubscriptionSearchRepository(Protocol): """按订阅 ID 和各自到期时间建立或合并活动任务。""" ... + async def async_enqueue( + self, + *, + subscription_ids: tuple[int, ...], + source: str, + priority: int, + available_at_by_subscription: Optional[Mapping[int, str]] = None, + ) -> SearchEnqueueResult: + """在异步会话中建立或合并活动任务。""" + ... + def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]: """按优先级和稳定游标认领下一条可执行任务。""" ... diff --git a/app/chain/subscribe/contract.py b/app/chain/subscribe/contract.py index 301f4203e..5c52772fe 100644 --- a/app/chain/subscribe/contract.py +++ b/app/chain/subscribe/contract.py @@ -36,6 +36,7 @@ if TYPE_CHECKING: _SubscribeChain__async_apply_episodes_refresh: Callable[..., Awaitable[Any]] _SubscribeChain__async_notify_subscribe_create_failure: Callable[..., Awaitable[Any]] _SubscribeChain__async_post_subscribe_added: Callable[..., Awaitable[Any]] + _SubscribeChain__async_queue_new_subscription_search: Callable[..., Awaitable[Any]] _SubscribeChain__build_completion_notification: Callable[..., Any] _SubscribeChain__build_subscribe_notification: Callable[..., Any] _SubscribeChain__download_best_version_with_full_pack_first: Callable[..., Any] diff --git a/app/chain/subscribe/notify.py b/app/chain/subscribe/notify.py index d971e2b45..7c2c3b12a 100644 --- a/app/chain/subscribe/notify.py +++ b/app/chain/subscribe/notify.py @@ -18,7 +18,6 @@ from app.domain.context import ( MediaInfo, ) from app.domain.meta.metabase import MetaBase -from app.runtime.execution import run_in_threadpool from app.runtime.log import logger from app.schemas.common import JsonData from app.schemas.message import Message as _SchemaMessage @@ -145,17 +144,16 @@ class SubscribeNotificationOwner(_SubscribeOwnerBase): ) try: self._SubscribeChain__queue_new_subscription_search(subscribe_id) - except Exception as error: - logger.warning( - "订阅已保存，但自动搜索暂时没有安排成功，" - f"系统会在下一次检查时重试：{error}" - ) + except Exception: + logger.warning("订阅已保存，但自动搜索暂时没有安排成功，系统会在下一次检查时重试") + logger.debug("安排订阅自动搜索失败", exc_info=True) try: report_delivered = _subscription_share_snapshot().report_added( self._SubscribeChain__subscribe_report_payload(context) ) - except Exception as error: - logger.warning(f"订阅新增统计上报失败，将由后台重试：{error}") + except Exception: + logger.warning("订阅新增统计暂时没有上报成功，系统会在后台重试") + logger.debug("订阅新增统计上报失败", exc_info=True) return False if not report_delivered: logger.warning("订阅新增统计上报未确认，将由后台重试") @@ -186,21 +184,17 @@ class SubscribeNotificationOwner(_SubscribeOwnerBase): }, ) try: - await run_in_threadpool( - self._SubscribeChain__queue_new_subscription_search, - subscribe_id, - ) - except Exception as error: - logger.warning( - "订阅已保存，但自动搜索暂时没有安排成功，" - f"系统会在下一次检查时重试：{error}" - ) + await self._SubscribeChain__async_queue_new_subscription_search(subscribe_id) + except Exception: + logger.warning("订阅已保存，但自动搜索暂时没有安排成功，系统会在下一次检查时重试") + logger.debug("安排订阅自动搜索失败", exc_info=True) try: report_delivered = await _subscription_share_snapshot().async_report_added( self._SubscribeChain__subscribe_report_payload(context) ) - except Exception as error: - logger.warning(f"订阅新增统计上报失败，将由后台重试：{error}") + except Exception: + logger.warning("订阅新增统计暂时没有上报成功，系统会在后台重试") + logger.debug("订阅新增统计上报失败", exc_info=True) return False if not report_delivered: logger.warning("订阅新增统计上报未确认，将由后台重试") diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index 6b5cb4563..0946c1e5c 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -19,7 +19,6 @@ from app.application.subscription.execution import ( SearchTaskSnapshot, SubscriptionExecutionContext, SubscriptionSearchRepository, - handle_subscription_search_deferred, raise_subscription_site_budget_deferral, raise_subscription_site_budget_failures, ) @@ -28,20 +27,22 @@ from app.application.subscription.observability import ( SearchTaskOutcome, batch_finished_count, batch_progress_text, - finish_returned_search_task, inline_search_result, ) from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.sitebudget import ( SubscriptionSearchCancelled, SubscriptionSearchDeferred, - SubscriptionSiteBudget, ) from app.chain.media import MediaChain from app.chain.search.facade import SearchChain from app.chain.subscribe.contract import _SubscribeOwnerBase from app.chain.subscribe.identity import subscribe_recognize_kwargs from app.chain.subscribe.metadata import apply_subscription_classification +from app.chain.subscribe.searchtask import ( + SubscriptionSearchTaskRunner, + retry_at_after, +) from app.domain.context import ( MediaInfo, ) @@ -54,8 +55,6 @@ from app.schemas.types import ( ) _NEW_SUBSCRIPTION_EDIT_SECONDS = 60 -_FOREGROUND_RETRY_SECONDS = 5 -_BACKGROUND_RETRY_SECONDS = 10 def _ensure_execution_active( @@ -70,22 +69,6 @@ def _ensure_execution_active( raise TimeoutError("这次搜索用时过长，已停止") -def _update_search_task_phase( - queue: SubscriptionSearchRepository, - task_id: str, - lease_token: str, - phase: str, - current_site_id: Optional[int] = None, -) -> None: - """以当前任务租约持久化业务阶段，过期执行者不得覆盖新状态。""" - queue.update_task_phase( - task_id=task_id, - lease_token=lease_token, - phase=phase, - current_site_id=current_site_id, - ) - - def _search_source_and_priority( *, sid: Optional[int], @@ -104,13 +87,6 @@ def _search_source_and_priority( return "new", 50 -def _retry_at_after(seconds: int) -> str: - """返回指定秒数后的 UTC 时间，供短暂等待任务重新入队。""" - return (datetime.now(timezone.utc) + timedelta(seconds=max(1, seconds))).isoformat( - timespec="seconds" - ) - - def _search_task_available_at( source: str, subscription_ids: tuple[int, ...], @@ -132,36 +108,6 @@ def _search_task_available_at( return available_at -def _skip_search_task( - queue: SubscriptionSearchRepository, - task: SearchTaskSnapshot, - reason: str, -) -> bool: - """以 skipped 终态收口未执行任务，并保留可见原因。""" - if task.lease_token is None: - return False - return queue.finish_task( - task_id=task.task_id, - lease_token=task.lease_token, - state="skipped", - error=reason, - ) - - -def _release_cancelled_or_stopped_search_task( - queue: SubscriptionSearchRepository, - task_id: str, - lease_token: str, - system_stopped: bool, -) -> bool: - """用户取消落终态，系统停机仅释放任务以供重启恢复。""" - return queue.release_task( - task_id=task_id, - lease_token=lease_token, - cancelled=not system_stopped, - ) - - class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): """订阅主动搜索入口、队列提交与消费协调 owner。""" @@ -528,212 +474,30 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): summary: SearchExecutionSummary, ) -> Optional[int]: """执行一条已认领任务，返回实际进入搜索处理的订阅 ID。""" - if task.lease_token is None: - logger.error("这次订阅搜索暂时无法开始，系统稍后会重新处理") - summary.record("failed", "missing_lease") - return None - task_id = str(task.task_id) - lease_token = task.lease_token - cancelled = partial(queue.is_cancel_requested, task_id) stop_state = getattr(self, "stop_state", runtime_stop_state) - if cancelled(): - queue.release_task(task_id=task_id, lease_token=lease_token, cancelled=True) - summary.record("cancelled", "cancelled") - return None - subscribe = self.subscription_repository.get(task.subscription_id) - if subscribe is None: - queue.finish_task( - task_id=task_id, - lease_token=lease_token, - state="cancelled", - error="订阅已删除", - ) - summary.record("cancelled", "missing_subscription") - return None - self._report_search_progress(progress_callback, subscribe, index, limit) - recent_retry_at = self._recent_subscription_retry_at(subscribe, task.source) - if recent_retry_at: - queue.defer_task( - task_id=task_id, - lease_token=lease_token, - available_at=recent_retry_at, - phase="scheduled", - message="订阅刚刚创建，保存好设置后会自动开始搜索", - ) - summary.record("requeued", "recent_subscription") - return None - lease = self._subscription_execution_admission.try_acquire( - subscription_id=subscribe.id, - operation="search", - ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL, + runner = SubscriptionSearchTaskRunner( + queue=queue, + task=task, + owner=owner, + searchchain=searchchain, + index=index, + limit=limit, + progress_callback=progress_callback, + summary=summary, + subscription_repository=self.subscription_repository, + execution_admission=self._subscription_execution_admission, + execution_ttl=self._SUBSCRIPTION_EXECUTION_TTL, + recent_retry_at=self._recent_subscription_retry_at, + process_subscription=self._process_search_subscription, + reset_subscription=partial( + self._SubscribeChain__apply_subscribe_update, + update_data={"state": "R"}, + scene="search_reset", + ), + report_progress=self._report_search_progress, + stop_state=stop_state, ) - if lease is None: - if task.source in {"manual", "targeted", "new"}: - retry_seconds = ( - _FOREGROUND_RETRY_SECONDS - if task.source in {"manual", "targeted"} - else _BACKGROUND_RETRY_SECONDS - ) - queue.defer_task( - task_id=task_id, - lease_token=lease_token, - available_at=_retry_at_after(retry_seconds), - phase="waiting_subscription", - message="这个订阅正在处理，结束后会自动继续搜索", - ) - summary.record("requeued", "admission_conflict") - logger.debug(f"订阅《{subscribe.name}》正在处理，搜索会在结束后自动继续") - else: - _skip_search_task( - queue, - task, - "这个订阅正在处理，本次自动检查无需重复执行", - ) - summary.record("skipped", "admission_conflict") - return None - phase_changed = partial(_update_search_task_phase, queue, task_id, lease_token) - execution_context = SubscriptionExecutionContext( - lease=lease, - admission=self._subscription_execution_admission, - task_id=task_id, - cancel_requested=lambda: cancelled() or stop_state.is_system_stopped, - phase_changed=phase_changed, - ) - current = subscribe - try: - current = self.subscription_repository.get(task.subscription_id) - if current is None: - queue.finish_task( - task_id=task_id, - lease_token=lease_token, - state="cancelled", - error="订阅已删除", - ) - summary.record("cancelled", "missing_subscription") - return None - if current.state == "S": - _skip_search_task(queue, task, "订阅已暂停，这次没有搜索") - summary.record("skipped", "paused") - return None - searchchain.configure_subscription_site_budget( - SubscriptionSiteBudget( - repository=queue, - owner=f"{owner}:{task_id}", - cancelled=execution_context.should_stop, - stop_state=stop_state, - phase_changed=phase_changed, - metrics=summary.site_metrics, - ) - ) - current = self._process_search_subscription( - current, - searchchain, - execution_context=execution_context, - ) - system_stopped = stop_state.is_system_stopped - cancel_requested = False if system_stopped else cancelled() - subscription_id, outcome, reason = finish_returned_search_task( - queue=queue, - task_id=task_id, - lease_token=lease_token, - subscription_id=task.subscription_id, - execution_context=execution_context, - system_stopped=system_stopped, - cancel_requested=cancel_requested, - ) - summary.record(outcome, reason) - return subscription_id - except SubscriptionSearchCancelled: - if execution_context.is_expired() and not execution_context.is_cancel_requested(): - queue.finish_task( - task_id=task_id, - lease_token=lease_token, - state="failed", - error="这次搜索用时过长，已停止，可稍后重试", - ) - summary.record("failed", "ttl_timeout") - else: - system_stopped = stop_state.is_system_stopped - _release_cancelled_or_stopped_search_task( - queue, - task_id, - lease_token, - system_stopped, - ) - summary.record( - "requeued" if system_stopped else "cancelled", - "system_stop" if system_stopped else "cancelled", - ) - except SubscriptionSearchDeferred as deferred: - handle_subscription_search_deferred( - queue, task_id, lease_token, deferred, summary.record - ) - except Exception as err: - logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True) - queue.finish_task( - task_id=task_id, - lease_token=lease_token, - state="failed", - error=str(err), - ) - summary.record("failed", "error") - finally: - self._cleanup_search_task( - queue=queue, - searchchain=searchchain, - subscribe=subscribe, - current=current, - lease=lease, - progress_callback=progress_callback, - index=index, - limit=limit, - summary=summary, - ) - return None - - def _cleanup_search_task( - self, - *, - queue: SubscriptionSearchRepository, - searchchain: SearchChain, - subscribe: SubscriptionSnapshot, - current: Optional[SubscriptionSnapshot], - lease: Any, - progress_callback: Optional[Callable[..., None]], - index: int, - limit: int, - summary: SearchExecutionSummary, - ) -> None: - """清理站点预算和订阅状态，并在所有异常路径释放 owner。""" - try: - searchchain.configure_subscription_site_budget(None) - except Exception as err: - logger.error( - f"订阅《{subscribe.name}》结束站点访问时遇到问题，" - f"系统稍后会继续处理：{str(err)}", - exc_info=True, - ) - try: - if current and current.state == "N": - self._SubscribeChain__apply_subscribe_update( - current, - {"state": "R"}, - scene="search_reset", - ) - except Exception as err: - logger.error(f"订阅《{subscribe.name}》搜索结束后未能恢复正常状态：{str(err)}", exc_info=True) - finally: - released = self._subscription_execution_admission.release(lease) - if not released: - summary.release_failures += 1 - logger.error(f"订阅《{subscribe.name}》的搜索状态没有正常恢复，系统稍后会继续处理") - self._report_search_progress( - progress_callback, - subscribe, - index, - limit, - finished=True, - ) + return runner.execute() def resume_search_queue( self, @@ -837,7 +601,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): return None retry_seconds = max(1, int(remaining) + 1) logger.debug(f"新订阅《{subscribe.name}》将在约 {retry_seconds} 秒后自动开始搜索") - return _retry_at_after(retry_seconds) + return retry_at_after(retry_seconds) def _SubscribeChain__queue_new_subscription_search( self, @@ -865,6 +629,33 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): logger.info(f"已安排新订阅《{subscribe.name}》自动搜索，保存好设置后会自动开始") return enqueued.active_batch_ids[0] if enqueued.active_batch_ids else None + async def _SubscribeChain__async_queue_new_subscription_search( + self, + subscribe_id: int, + ) -> Optional[str]: + """在异步订阅提交后安排自动搜索。""" + queue: Optional[SubscriptionSearchRepository] = getattr( + self, "subscription_search_repository", None + ) + if queue is None: + return None + subscribe = await self.subscription_repository.async_get(subscribe_id) + if subscribe is None or subscribe.state != "N": + return None + available_at = self._recent_subscription_retry_at(subscribe, "new") + enqueued = await queue.async_enqueue( + subscription_ids=(subscribe_id,), + source="new", + priority=50, + available_at_by_subscription={ + subscribe_id: available_at + or datetime.now(timezone.utc).isoformat(timespec="seconds") + }, + ) + if enqueued.created_count: + logger.info(f"已安排新订阅《{subscribe.name}》自动搜索，保存好设置后会自动开始") + return enqueued.active_batch_ids[0] if enqueued.active_batch_ids else None + @staticmethod def _wait_before_scheduled_search( sid: Optional[int], diff --git a/app/chain/subscribe/searchtask.py b/app/chain/subscribe/searchtask.py new file mode 100644 index 000000000..ca783efee --- /dev/null +++ b/app/chain/subscribe/searchtask.py @@ -0,0 +1,330 @@ +"""订阅搜索队列单任务执行。""" + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from functools import partial +from typing import Callable, Optional + +from app.application.subscription.contract import ( + SubscriptionRepository, + SubscriptionSnapshot, +) +from app.application.subscription.execution import ( + SearchTaskSnapshot, + SubscriptionExecutionAdmission, + SubscriptionExecutionContext, + SubscriptionExecutionLease, + SubscriptionSearchRepository, + handle_subscription_search_deferred, +) +from app.application.subscription.observability import ( + SearchExecutionSummary, + finish_returned_search_task, +) +from app.application.subscription.sitebudget import ( + SubscriptionSearchCancelled, + SubscriptionSearchDeferred, + SubscriptionSiteBudget, +) +from app.chain.search.facade import SearchChain +from app.runtime.log import logger +from app.runtime.stop import StopState + +_FOREGROUND_RETRY_SECONDS = 5 +_BACKGROUND_RETRY_SECONDS = 10 + + +def retry_at_after(seconds: int) -> str: + """返回指定秒数后的 UTC 时间，供短暂等待任务重新入队。""" + return (datetime.now(timezone.utc) + timedelta(seconds=max(1, seconds))).isoformat( + timespec="seconds" + ) + + +def skip_search_task( + queue: SubscriptionSearchRepository, + task: SearchTaskSnapshot, + reason: str, +) -> bool: + """以 skipped 终态收口未执行任务，并保留可见原因。""" + if task.lease_token is None: + return False + return queue.finish_task( + task_id=task.task_id, + lease_token=task.lease_token, + state="skipped", + error=reason, + ) + + +@dataclass(slots=True) +class SubscriptionSearchTaskRunner: + """执行一条已认领的订阅搜索任务，并负责完整收尾。""" + + queue: SubscriptionSearchRepository + task: SearchTaskSnapshot + owner: str + searchchain: SearchChain + index: int + limit: int + progress_callback: Optional[Callable[..., None]] + summary: SearchExecutionSummary + subscription_repository: SubscriptionRepository + execution_admission: SubscriptionExecutionAdmission + execution_ttl: int + recent_retry_at: Callable[[SubscriptionSnapshot, str], Optional[str]] + process_subscription: Callable[..., Optional[SubscriptionSnapshot]] + reset_subscription: Callable[[SubscriptionSnapshot], SubscriptionSnapshot] + report_progress: Callable[..., None] + stop_state: StopState + + def execute(self) -> Optional[int]: + """执行当前任务，返回实际进入搜索处理的订阅 ID。""" + lease_token = self.task.lease_token + if lease_token is None: + logger.error("这次订阅搜索暂时无法开始，系统稍后会重新处理") + self.summary.record("failed", "missing_lease") + return None + + task_id = str(self.task.task_id) + cancelled = partial(self.queue.is_cancel_requested, task_id) + if cancelled(): + self.queue.release_task(task_id=task_id, lease_token=lease_token, cancelled=True) + self.summary.record("cancelled", "cancelled") + return None + + subscribe = self.subscription_repository.get(self.task.subscription_id) + if subscribe is None: + self._finish_missing_subscription(task_id, lease_token) + return None + + self.report_progress( + self.progress_callback, + subscribe, + self.index, + self.limit, + ) + recent_retry_at = self.recent_retry_at(subscribe, self.task.source) + if recent_retry_at: + self.queue.defer_task( + task_id=task_id, + lease_token=lease_token, + available_at=recent_retry_at, + phase="scheduled", + message="订阅刚刚创建，保存好设置后会自动开始搜索", + ) + self.summary.record("requeued", "recent_subscription") + return None + + execution_lease = self.execution_admission.try_acquire( + subscription_id=subscribe.id, + operation="search", + ttl_seconds=self.execution_ttl, + ) + if execution_lease is None: + self._handle_active_subscription(subscribe, task_id, lease_token) + return None + return self._execute_owned_task( + subscribe=subscribe, + task_id=task_id, + lease_token=lease_token, + execution_lease=execution_lease, + cancelled=cancelled, + ) + + def _finish_missing_subscription(self, task_id: str, lease_token: str) -> None: + """把已经删除的订阅任务标记为取消。""" + self.queue.finish_task( + task_id=task_id, + lease_token=lease_token, + state="cancelled", + error="订阅已删除", + ) + self.summary.record("cancelled", "missing_subscription") + + def _handle_active_subscription( + self, + subscribe: SubscriptionSnapshot, + task_id: str, + lease_token: str, + ) -> None: + """根据搜索来源延后或跳过正在处理的订阅。""" + if self.task.source in {"manual", "targeted", "new"}: + retry_seconds = ( + _FOREGROUND_RETRY_SECONDS + if self.task.source in {"manual", "targeted"} + else _BACKGROUND_RETRY_SECONDS + ) + self.queue.defer_task( + task_id=task_id, + lease_token=lease_token, + available_at=retry_at_after(retry_seconds), + phase="waiting_subscription", + message="这个订阅正在处理，结束后会自动继续搜索", + ) + self.summary.record("requeued", "admission_conflict") + logger.debug(f"订阅《{subscribe.name}》正在处理，搜索会在结束后自动继续") + return + skip_search_task( + self.queue, + self.task, + "这个订阅正在处理，本次自动检查无需重复执行", + ) + self.summary.record("skipped", "admission_conflict") + + def _execute_owned_task( + self, + *, + subscribe: SubscriptionSnapshot, + task_id: str, + lease_token: str, + execution_lease: SubscriptionExecutionLease, + cancelled: Callable[[], bool], + ) -> Optional[int]: + """在持有订阅执行权时完成搜索、异常处理和资源释放。""" + phase_changed = partial(self._update_phase, task_id, lease_token) + execution_context = SubscriptionExecutionContext( + lease=execution_lease, + admission=self.execution_admission, + task_id=task_id, + cancel_requested=lambda: cancelled() or self.stop_state.is_system_stopped, + phase_changed=phase_changed, + ) + current: Optional[SubscriptionSnapshot] = subscribe + try: + current = self.subscription_repository.get(self.task.subscription_id) + 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, + owner=f"{self.owner}:{task_id}", + cancelled=execution_context.should_stop, + stop_state=self.stop_state, + phase_changed=phase_changed, + metrics=self.summary.site_metrics, + ) + ) + current = self.process_subscription( + current, + self.searchchain, + execution_context=execution_context, + ) + system_stopped = self.stop_state.is_system_stopped + cancel_requested = False if system_stopped else cancelled() + subscription_id, outcome, reason = finish_returned_search_task( + queue=self.queue, + task_id=task_id, + lease_token=lease_token, + subscription_id=self.task.subscription_id, + execution_context=execution_context, + system_stopped=system_stopped, + cancel_requested=cancel_requested, + ) + self.summary.record(outcome, reason) + return subscription_id + except SubscriptionSearchCancelled: + self._handle_cancelled_search(task_id, lease_token, execution_context) + except SubscriptionSearchDeferred as deferred: + handle_subscription_search_deferred( + self.queue, + task_id, + lease_token, + deferred, + self.summary.record, + ) + except Exception as err: + logger.error(f"订阅《{subscribe.name}》搜索失败：{str(err)}", exc_info=True) + self.queue.finish_task( + task_id=task_id, + lease_token=lease_token, + state="failed", + error=str(err), + ) + self.summary.record("failed", "error") + finally: + self._cleanup(subscribe, current, execution_lease) + return None + + def _update_phase( + self, + task_id: str, + lease_token: str, + phase: str, + current_site_id: Optional[int] = None, + ) -> None: + """保存当前任务的用户可见阶段和正在访问的站点。""" + self.queue.update_task_phase( + task_id=task_id, + lease_token=lease_token, + phase=phase, + current_site_id=current_site_id, + ) + + def _handle_cancelled_search( + self, + task_id: str, + lease_token: str, + execution_context: SubscriptionExecutionContext, + ) -> None: + """区分搜索超时、用户停止和系统关闭。""" + if execution_context.is_expired() and not execution_context.is_cancel_requested(): + self.queue.finish_task( + task_id=task_id, + lease_token=lease_token, + state="failed", + error="这次搜索用时过长，已停止，可稍后重试", + ) + self.summary.record("failed", "ttl_timeout") + return + system_stopped = self.stop_state.is_system_stopped + self.queue.release_task( + task_id=task_id, + lease_token=lease_token, + cancelled=not system_stopped, + ) + self.summary.record( + "requeued" if system_stopped else "cancelled", + "system_stop" if system_stopped else "cancelled", + ) + + def _cleanup( + self, + subscribe: SubscriptionSnapshot, + current: Optional[SubscriptionSnapshot], + execution_lease: SubscriptionExecutionLease, + ) -> None: + """清理站点访问、订阅状态和本轮执行权。""" + try: + self.searchchain.configure_subscription_site_budget(None) + except Exception as err: + logger.error( + f"订阅《{subscribe.name}》结束站点访问时遇到问题，" + f"系统稍后会继续处理：{str(err)}", + exc_info=True, + ) + try: + if current and current.state == "N": + self.reset_subscription(current) + except Exception as err: + logger.error( + f"订阅《{subscribe.name}》搜索结束后未能恢复正常状态：{str(err)}", + exc_info=True, + ) + finally: + released = self.execution_admission.release(execution_lease) + if not released: + self.summary.release_failures += 1 + logger.error(f"订阅《{subscribe.name}》的搜索状态没有正常恢复，系统稍后会继续处理") + self.report_progress( + self.progress_callback, + subscribe, + self.index, + self.limit, + finished=True, + ) diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py index d15cbcaaf..087313d63 100644 --- a/app/db/adapters/subscriptionsearch.py +++ b/app/db/adapters/subscriptionsearch.py @@ -1,9 +1,11 @@ """订阅搜索持久队列的 SQLAlchemy 适配器。""" from collections.abc import Callable, Mapping +from contextlib import AbstractAsyncContextManager from datetime import datetime, timedelta, timezone from typing import Optional, TypeVar +from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session from app.application.subscription.execution import ( @@ -17,7 +19,7 @@ from app.db.models.subscriptionsearch import ( SubscriptionSearchTask, ) from app.db.oper.subscriptionsearch import SubscriptionSearchOper -from app.db.uow import SqlAlchemyUnitOfWork +from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork T = TypeVar("T") _BUSY_SITE_RETRY_SECONDS = 10 @@ -68,12 +70,42 @@ def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot: ) -class TransactionalSubscriptionSearchRepository: - """使用短事务实现订阅搜索队列端口。""" +def _enqueue_result( + repository: SubscriptionSearchOper, + *, + subscription_ids: tuple[int, ...], + source: str, + priority: int, + available_at_by_subscription: Optional[Mapping[int, str]], +) -> SearchEnqueueResult: + """在当前事务中创建搜索批次并投影返回结果。""" + record, created, coalesced, active_batch_ids = repository.enqueue( + subscription_ids=subscription_ids, + source=source, + priority=priority, + available_at_by_subscription=available_at_by_subscription, + ) + return SearchEnqueueResult( + batch=_batch(record), + created_count=created, + coalesced_count=coalesced, + active_batch_ids=active_batch_ids, + ) - def __init__(self, session_factory: Callable[[], Session]) -> None: - """保存由组合根注入的同步 Session 工厂。""" + +class TransactionalSubscriptionSearchRepository: + """使用同步或异步短事务实现订阅搜索队列端口。""" + + def __init__( + self, + session_factory: Callable[[], Session], + async_session_factory: Optional[ + Callable[[], AbstractAsyncContextManager[AsyncSession]] + ] = None, + ) -> None: + """保存由组合根注入的同步和异步 Session 工厂。""" self._session_factory = session_factory + self._async_session_factory = async_session_factory def _read(self, operation: Callable[[SubscriptionSearchOper], T]) -> T: """在短 Session 中执行一次只读查询。""" @@ -92,6 +124,22 @@ class TransactionalSubscriptionSearchRepository: unit_of_work.rollback() raise + async def _async_write(self, operation: Callable[[SubscriptionSearchOper], T]) -> T: + """在短异步事务中执行一次队列状态变更。""" + if self._async_session_factory is None: + raise RuntimeError("订阅搜索异步写入尚未配置") + async with self._async_session_factory() as session: + unit_of_work = SqlAlchemyAsyncUnitOfWork(session) + try: + result = await session.run_sync( + lambda sync_session: operation(SubscriptionSearchOper(sync_session)) + ) + await unit_of_work.commit() + return result + except Exception: + await unit_of_work.rollback() + raise + def enqueue( self, *, @@ -101,22 +149,34 @@ class TransactionalSubscriptionSearchRepository: available_at_by_subscription: Optional[Mapping[int, str]] = None, ) -> SearchEnqueueResult: """创建批次并返回 single-flight 合并计数。""" - def operation(repository: SubscriptionSearchOper) -> SearchEnqueueResult: - """在同一事务内创建批次和任务。""" - record, created, coalesced, active_batch_ids = repository.enqueue( + return self._write( + lambda repository: _enqueue_result( + repository, subscription_ids=subscription_ids, source=source, priority=priority, available_at_by_subscription=available_at_by_subscription, ) - return SearchEnqueueResult( - batch=_batch(record), - created_count=created, - coalesced_count=coalesced, - active_batch_ids=active_batch_ids, - ) + ) - return self._write(operation) + async def async_enqueue( + self, + *, + subscription_ids: tuple[int, ...], + source: str, + priority: int, + available_at_by_subscription: Optional[Mapping[int, str]] = None, + ) -> SearchEnqueueResult: + """在短异步事务中创建批次并返回合并计数。""" + return await self._async_write( + lambda repository: _enqueue_result( + repository, + subscription_ids=subscription_ids, + source=source, + priority=priority, + available_at_by_subscription=available_at_by_subscription, + ) + ) def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]: """认领下一任务并返回脱离 Session 的快照。""" diff --git a/app/startup/composition/runtime.py b/app/startup/composition/runtime.py index 2ec5d2502..0cd2186d4 100644 --- a/app/startup/composition/runtime.py +++ b/app/startup/composition/runtime.py @@ -139,7 +139,10 @@ def compose_runtime_dependencies() -> RuntimeDependencies: transfer_execution=TransactionalTransferExecutionRepository(SessionFactory), message_helper=message_helper_factory(), message_queue=MessageQueueManager(auto_start=False), - subscription_search=TransactionalSubscriptionSearchRepository(SessionFactory), + subscription_search=TransactionalSubscriptionSearchRepository( + SessionFactory, + async_session_scope, + ), ) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 089815c92..72c017d6c 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -754,8 +754,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 972 | -| 内部导入边 | 8,208 | +| Python 模块 | 973 | +| 内部导入边 | 8,220 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 0d8ff2887..9f6278f5d 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 972 / 8,208 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 973 / 8,220 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 0d978aeee..fa510b925 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 8208, - "edge_sha256": "1a7eee327467c99451ce338d7d2670b74c2e71aeefcf36ac27b6668cb30a71aa", + "edge_count": 8220, + "edge_sha256": "4ef4fb9843d0e3a1d356f297ff2dafca35794b7811b7574d64bda71e114f2418", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -4779,7 +4779,6 @@ "app.chain.subscribe.notify -> app.domain.meta", "app.chain.subscribe.notify -> app.domain.meta.metabase", "app.chain.subscribe.notify -> app.runtime", - "app.chain.subscribe.notify -> app.runtime.execution", "app.chain.subscribe.notify -> app.runtime.log", "app.chain.subscribe.notify -> app.schemas", "app.chain.subscribe.notify -> app.schemas.common", @@ -4879,6 +4878,7 @@ "app.chain.subscribe.search -> app.chain.subscribe.contract", "app.chain.subscribe.search -> app.chain.subscribe.identity", "app.chain.subscribe.search -> app.chain.subscribe.metadata", + "app.chain.subscribe.search -> app.chain.subscribe.searchtask", "app.chain.subscribe.search -> app.domain", "app.chain.subscribe.search -> app.domain.context", "app.chain.subscribe.search -> app.domain.meta", @@ -4888,6 +4888,18 @@ "app.chain.subscribe.search -> app.runtime.stop", "app.chain.subscribe.search -> app.schemas", "app.chain.subscribe.search -> app.schemas.types", + "app.chain.subscribe.searchtask -> app.application", + "app.chain.subscribe.searchtask -> app.application.subscription", + "app.chain.subscribe.searchtask -> app.application.subscription.contract", + "app.chain.subscribe.searchtask -> app.application.subscription.execution", + "app.chain.subscribe.searchtask -> app.application.subscription.observability", + "app.chain.subscribe.searchtask -> app.application.subscription.sitebudget", + "app.chain.subscribe.searchtask -> app.chain", + "app.chain.subscribe.searchtask -> app.chain.search", + "app.chain.subscribe.searchtask -> app.chain.search.facade", + "app.chain.subscribe.searchtask -> app.runtime", + "app.chain.subscribe.searchtask -> app.runtime.log", + "app.chain.subscribe.searchtask -> app.runtime.stop", "app.chain.system -> app.application", "app.chain.system -> app.application.configuration", "app.chain.system -> app.chain", @@ -9301,7 +9313,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 972, + "module_count": 973, "modules": [ "app", "app.adapters", @@ -9712,6 +9724,7 @@ "app.chain.subscribe.reconcile", "app.chain.subscribe.refresh", "app.chain.subscribe.search", + "app.chain.subscribe.searchtask", "app.chain.system", "app.chain.theaudiodb", "app.chain.tmdb", diff --git a/tests/test_subscription_search_queue.py b/tests/test_subscription_search_queue.py index 778c5166b..f2f29f74e 100644 --- a/tests/test_subscription_search_queue.py +++ b/tests/test_subscription_search_queue.py @@ -1,8 +1,11 @@ """订阅搜索持久队列、single-flight、租约和取消测试。""" +from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone +import pytest from sqlalchemy import create_engine, select, update +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.orm import Session, sessionmaker from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository @@ -17,6 +20,43 @@ def _repository(tmp_path): return TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)), engine +@pytest.mark.asyncio +async def test_search_queue_async_enqueue_uses_async_session(tmp_path): + """异步新增订阅应通过 AsyncSession 入队，并可由同步消费者继续认领。""" + database_path = tmp_path / "async-search-queue.db" + async_engine = create_async_engine(f"sqlite+aiosqlite:///{database_path}") + async_factory = async_sessionmaker(bind=async_engine, expire_on_commit=False) + + @asynccontextmanager + async def async_session_scope(): + """为测试队列提供独立异步会话。""" + async with async_factory() as session: + yield session + + async with async_engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + sync_engine = create_engine(f"sqlite:///{database_path}") + repository = TransactionalSubscriptionSearchRepository( + sessionmaker(bind=sync_engine), + async_session_scope, + ) + + try: + enqueued = await repository.async_enqueue( + subscription_ids=(101,), + source="new", + priority=50, + ) + claimed = repository.claim_next(owner="worker-async") + + assert enqueued.created_count == 1 + assert claimed is not None + assert claimed.subscription_id == 101 + finally: + sync_engine.dispose() + await async_engine.dispose() + + def test_search_queue_coalesces_active_subscription_and_raises_priority(tmp_path): """重叠入口只保留一个活动任务，手工请求可提高优先级。""" repository, _engine = _repository(tmp_path)