diff --git a/app/application/chain/context.py b/app/application/chain/context.py index 064e9e184..ac68d0110 100644 --- a/app/application/chain/context.py +++ b/app/application/chain/context.py @@ -33,6 +33,7 @@ if TYPE_CHECKING: SubscriptionMutationScope, SyncSubscriptionMutationScope, ) + from app.application.subscription.execution import SubscriptionSearchRepository from app.application.transfer.execution import TransferExecutionRepository from app.application.transfer.workflow import TransferAdmissionRepository @@ -74,6 +75,7 @@ class ChainRuntimeContext: media_server_repository: MediaServerRepository download_failure_repository: DownloadFailureRepository user_repository: ChainUserRepository + subscription_search_repository: Optional[SubscriptionSearchRepository] = None legacy_transfer_command: Optional[LegacyTransferCommand] = None durable_event_writer: Optional[ChainDurableEventWriter] = None configuration: ChainRuntimeConfig = field( diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py new file mode 100644 index 000000000..2cff3c333 --- /dev/null +++ b/app/application/subscription/execution.py @@ -0,0 +1,105 @@ +"""订阅搜索执行批次、任务与持久队列端口。""" + +from dataclasses import dataclass +from typing import Optional, Protocol + + +@dataclass(frozen=True, slots=True) +class SearchBatchSnapshot: + """订阅搜索批次的持久业务状态快照。""" + + batch_id: str + source: str + state: str + priority: int + total_count: int + finished_count: int + failed_count: int + cancelled_count: int + cancel_requested: bool + created_at: str + updated_at: str + started_at: Optional[str] = None + finished_at: Optional[str] = None + last_error: Optional[str] = None + + +@dataclass(frozen=True, slots=True) +class SearchTaskSnapshot: + """一个可认领、恢复和取消的订阅搜索任务快照。""" + + task_id: str + batch_id: str + subscription_id: int + source: str + priority: int + position: int + state: str + attempt_count: int + cancel_requested: bool + lease_token: Optional[str] + created_at: str + updated_at: str + started_at: Optional[str] = None + finished_at: Optional[str] = None + last_error: Optional[str] = None + + +@dataclass(frozen=True, slots=True) +class SearchEnqueueResult: + """一次批次入队结果，区分新任务与 single-flight 合并。""" + + batch: SearchBatchSnapshot + created_count: int + coalesced_count: int + + +class SubscriptionSearchRepository(Protocol): + """订阅搜索批次与任务的持久队列端口。""" + + def enqueue( + self, + *, + subscription_ids: tuple[int, ...], + source: str, + priority: int, + ) -> SearchEnqueueResult: + """按订阅 ID 建立批次，并合并已存在的活动任务。""" + ... + + def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]: + """按优先级和稳定游标认领下一条可执行任务。""" + ... + + def finish_task( + self, + *, + task_id: str, + lease_token: str, + state: str, + error: Optional[str] = None, + ) -> bool: + """以租约令牌收口任务，并推进所属批次聚合状态。""" + ... + + def release_task( + self, + *, + task_id: str, + lease_token: str, + cancelled: bool = False, + ) -> bool: + """释放尚未完成的任务租约，供停止或取消后恢复。""" + ... + + def is_cancel_requested(self, task_id: str) -> bool: + """判断任务或所属批次是否已请求取消。""" + ... + + def request_cancel(self, batch_id: str) -> bool: + """请求取消批次，并立即终止尚未发出的排队任务。""" + ... + + def get_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]: + """按稳定批次 ID 返回当前聚合状态。""" + ... diff --git a/app/chain/base.py b/app/chain/base.py index 50571ea97..23992b914 100644 --- a/app/chain/base.py +++ b/app/chain/base.py @@ -58,6 +58,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, self.async_filecache = context.async_file_cache self.site_repository = context.site_repository self.subscription_repository = context.subscription_repository + self.subscription_search_repository = context.subscription_search_repository self.subscription_mutation_scope = context.subscription_mutation_scope self.sync_subscription_mutation_scope = context.sync_subscription_mutation_scope self.subscription_delete_scope = context.subscription_delete_scope diff --git a/app/chain/subscribe/contract.py b/app/chain/subscribe/contract.py index cc2d1497a..6df68a62e 100644 --- a/app/chain/subscribe/contract.py +++ b/app/chain/subscribe/contract.py @@ -25,6 +25,7 @@ if TYPE_CHECKING: site_repository: Any subscription_completion_scope: Callable[..., Any] subscription_repository: Any + subscription_search_repository: Any sync_subscription_delete_scope: Callable[..., Any] sync_subscription_mutation_scope: Callable[..., Any] @@ -98,6 +99,7 @@ if TYPE_CHECKING: remote_list: Callable[..., Any] resolve_subscribe_missing: Callable[..., Any] reconcile_subscription_completion: Callable[..., Any] + resume_search_queue: Callable[..., Any] _SubscribeOwnerBase = _SubscribeOwnerHost else: diff --git a/app/chain/subscribe/facade.py b/app/chain/subscribe/facade.py index f74fab3e6..cfc161be4 100644 --- a/app/chain/subscribe/facade.py +++ b/app/chain/subscribe/facade.py @@ -49,6 +49,7 @@ class SubscribeChain( _interaction_handler_type = SubscribeInteractionHandler _rlock = threading.RLock() + _search_queue_lock = threading.Lock() _LOCK_TIMOUT = 3600 * 2 @classmethod diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index d77385914..77927769b 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -4,6 +4,7 @@ import random import time from datetime import datetime from typing import Any, Callable, Optional, cast +from uuid import uuid4 from app.application.configuration import get_configured_system_config from app.application.subscription.contract import ( @@ -13,6 +14,7 @@ from app.application.subscription.contract import ( subscribe_media_key, ) from app.application.subscription.query import SubscriptionQueryService +from app.application.subscription.execution import SearchBatchSnapshot, SubscriptionSearchRepository from app.chain.media import MediaChain from app.chain.search.facade import SearchChain from app.chain.subscribe.contract import _SubscribeOwnerBase @@ -81,7 +83,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): manual: Optional[bool] = False, progress_callback: Optional[Callable[..., None]] = None, sids: Optional[tuple[int, ...]] = None, - ) -> None: + ) -> Optional[str]: """ 执行订阅搜索。 @@ -102,7 +104,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): manual: Optional[bool] = False, progress_callback: Optional[Callable[..., None]] = None, sids: Optional[tuple[int, ...]] = None, - ) -> None: + ) -> Optional[str]: """ 订阅搜索 :param sid: 订阅ID，有值时只处理该订阅 @@ -112,6 +114,37 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): :param sids: 订阅ID集合，有值时按给定顺序处理 :return: 更新订阅状态为R或删除订阅 """ + queue = cast( + Optional[SubscriptionSearchRepository], + getattr(self, "subscription_search_repository", None), + ) + if queue is not None: + return self._execute_queued_search( + queue=queue, + sid=sid, + sids=sids, + state=state, + manual=manual, + progress_callback=progress_callback, + ) + self._execute_inline_search( + sid=sid, + sids=sids, + state=state, + manual=manual, + progress_callback=progress_callback, + ) + return None + + def _execute_inline_search( + self, + sid: Optional[int], + sids: Optional[tuple[int, ...]], + state: Optional[str], + manual: Optional[bool], + progress_callback: Optional[Callable[..., None]], + ) -> None: + """保留未注入持久队列宿主的旧串行锁语义。""" lock_acquired = self._acquire_run_lock("search", progress_callback) if not lock_acquired: return @@ -154,6 +187,216 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): self._rlock.release() logger.debug(f"search Lock released at {datetime.now()}") + def _execute_queued_search( + self, + *, + queue: SubscriptionSearchRepository, + sid: Optional[int], + sids: Optional[tuple[int, ...]], + state: Optional[str], + manual: Optional[bool], + progress_callback: Optional[Callable[..., None]], + ) -> str: + """将搜索转为持久任务并在无 Match 长锁的短租约中串行消费。""" + subscribes = self._load_search_subscriptions(sid=sid, sids=sids, state=state) + source, priority = self._search_source_and_priority( + sid=sid, + sids=sids, + state=state, + manual=manual, + ) + enqueued = queue.enqueue( + subscription_ids=tuple(subscribe.id for subscribe in subscribes), + source=source, + priority=priority, + ) + total = len(subscribes) + if progress_callback: + progress_callback( + value=0, + text=f"开始订阅搜索，共 {total} 个订阅 ...", + data={ + "batch_id": enqueued.batch.batch_id, + "total": total, + "finished": 0, + "coalesced": enqueued.coalesced_count, + }, + ) + processed = self._drain_search_queue( + queue=queue, + limit=max(1, enqueued.created_count + enqueued.coalesced_count), + progress_callback=progress_callback, + ) + processed_subscribes = [item for item in subscribes if item.id in processed] + self._notify_manual_search(manual, sid, sids, subscribes, processed_subscribes) + if progress_callback: + batch = queue.get_batch(enqueued.batch.batch_id) + progress_callback( + value=100, + text=self._batch_progress_text(batch), + data={ + "batch_id": enqueued.batch.batch_id, + "total": total, + "finished": len(processed_subscribes), + "coalesced": enqueued.coalesced_count, + }, + ) + return enqueued.batch.batch_id + + def _drain_search_queue( + self, + *, + queue: SubscriptionSearchRepository, + limit: int, + progress_callback: Optional[Callable[..., None]], + ) -> set[int]: + """有界消费可恢复任务；单任务失败不得阻止后续订阅。""" + queue_lock = getattr(self, "_search_queue_lock", None) + if queue_lock is not None and not queue_lock.acquire(blocking=False): + logger.debug("订阅搜索队列已有消费者，本轮仅保留持久任务") + return set() + owner = f"subscribe-search:{uuid4().hex}" + processed: set[int] = set() + try: + searchchain = SearchChain() + for index in range(1, limit + 1): + if runtime_stop_state.is_system_stopped: + break + task = queue.claim_next(owner=owner) + if task is None: + break + if not task.lease_token: + logger.error(f"订阅搜索任务 {task.task_id} 缺少租约令牌，跳过执行") + continue + if queue.is_cancel_requested(task.task_id): + queue.release_task( + task_id=task.task_id, + lease_token=task.lease_token, + cancelled=True, + ) + continue + subscribe = self.subscription_repository.get(task.subscription_id) + if subscribe is None: + queue.finish_task( + task_id=task.task_id, + lease_token=task.lease_token, + state="cancelled", + error="订阅已不存在", + ) + continue + self._report_search_progress(progress_callback, subscribe, index, limit) + if self._defer_recent_subscription(subscribe): + queue.finish_task( + task_id=task.task_id, + lease_token=task.lease_token, + state="completed", + ) + continue + current = subscribe + try: + current = self._process_search_subscription(subscribe, searchchain) + if queue.is_cancel_requested(task.task_id): + queue.release_task( + task_id=task.task_id, + lease_token=task.lease_token, + cancelled=True, + ) + else: + queue.finish_task( + task_id=task.task_id, + lease_token=task.lease_token, + state="completed", + ) + processed.add(subscribe.id) + except Exception as err: + logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True) + queue.finish_task( + task_id=task.task_id, + lease_token=task.lease_token, + state="failed", + error=str(err), + ) + finally: + if current and current.state == "N": + try: + self._SubscribeChain__apply_subscribe_update( + current, + {"state": "R"}, + scene="search_reset", + ) + except Exception as err: + logger.error( + f"订阅 {current.name} 搜索后状态重置失败：{str(err)}", + exc_info=True, + ) + self._report_search_progress(progress_callback, subscribe, index, limit, finished=True) + finally: + if queue_lock is not None: + queue_lock.release() + return processed + + def resume_search_queue( + self, + progress_callback: Optional[Callable[..., None]] = None, + limit: int = 50, + ) -> None: + """短周期恢复排队或租约过期任务，不创建新的 24 小时兜底批次。""" + queue = cast( + Optional[SubscriptionSearchRepository], + getattr(self, "subscription_search_repository", None), + ) + if queue is None: + return + self._drain_search_queue( + queue=queue, + limit=max(1, limit), + progress_callback=progress_callback, + ) + + @staticmethod + def _search_source_and_priority( + *, + sid: Optional[int], + sids: Optional[tuple[int, ...]], + state: Optional[str], + manual: Optional[bool], + ) -> tuple[str, int]: + """把兼容入口归一为持久来源和公平队列优先级。""" + if manual: + return "manual", 100 + if sid or sids is not None: + return "targeted", 80 + if state in {"R", "P"}: + return "fallback", 10 + return "new", 50 + + @staticmethod + def _batch_progress_text(batch: Optional[SearchBatchSnapshot]) -> str: + """把批次聚合终态转为兼容进度文案。""" + if batch is None: + return "订阅搜索任务已提交" + if batch.state == "failed": + return "订阅搜索完成，部分任务失败" + if batch.state == "cancelled": + return "订阅搜索已取消" + return "订阅搜索完成" + + def cancel_search_batch(self, batch_id: str) -> bool: + """请求取消持久搜索批次；未注入队列时返回失败。""" + queue = cast( + Optional[SubscriptionSearchRepository], + getattr(self, "subscription_search_repository", None), + ) + return bool(queue and queue.request_cancel(batch_id)) + + def get_search_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]: + """返回持久搜索批次状态；未注入队列时返回空。""" + queue = cast( + Optional[SubscriptionSearchRepository], + getattr(self, "subscription_search_repository", None), + ) + return queue.get_batch(batch_id) if queue else None + def _load_search_subscriptions( self, sid: Optional[int], diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py new file mode 100644 index 000000000..0a9daf9f9 --- /dev/null +++ b/app/db/adapters/subscriptionsearch.py @@ -0,0 +1,169 @@ +"""订阅搜索持久队列的 SQLAlchemy 适配器。""" + +from collections.abc import Callable +from typing import Optional, TypeVar + +from sqlalchemy.orm import Session + +from app.application.subscription.execution import ( + SearchBatchSnapshot, + SearchEnqueueResult, + SearchTaskSnapshot, +) +from app.db.models.subscriptionsearch import ( + SubscriptionSearchBatch, + SubscriptionSearchTask, +) +from app.db.oper.subscriptionsearch import SubscriptionSearchOper +from app.db.uow import SqlAlchemyUnitOfWork + +T = TypeVar("T") + + +def _batch(record: SubscriptionSearchBatch) -> SearchBatchSnapshot: + """在 Session 内投影不可变搜索批次快照。""" + return SearchBatchSnapshot( + batch_id=record.batch_id, + source=record.source, + state=record.state, + priority=record.priority, + total_count=record.total_count, + finished_count=record.finished_count, + failed_count=record.failed_count, + cancelled_count=record.cancelled_count, + cancel_requested=bool(record.cancel_requested), + created_at=record.created_at, + updated_at=record.updated_at, + started_at=record.started_at, + finished_at=record.finished_at, + last_error=record.last_error, + ) + + +def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot: + """在 Session 内投影不可变搜索任务快照。""" + return SearchTaskSnapshot( + task_id=record.task_id, + batch_id=record.batch_id, + subscription_id=record.subscription_id, + source=record.source, + priority=record.priority, + position=record.position, + state=record.state, + attempt_count=record.attempt_count, + cancel_requested=bool(record.cancel_requested), + lease_token=record.lease_token, + created_at=record.created_at, + updated_at=record.updated_at, + started_at=record.started_at, + finished_at=record.finished_at, + last_error=record.last_error, + ) + + +class TransactionalSubscriptionSearchRepository: + """使用短事务实现订阅搜索队列端口。""" + + def __init__(self, session_factory: Callable[[], Session]) -> None: + """保存由组合根注入的同步 Session 工厂。""" + self._session_factory = session_factory + + def _read(self, operation: Callable[[SubscriptionSearchOper], T]) -> T: + """在短 Session 中执行一次只读查询。""" + with self._session_factory() as session: + return operation(SubscriptionSearchOper(session)) + + def _write(self, operation: Callable[[SubscriptionSearchOper], T]) -> T: + """在短事务中执行一次队列状态变更。""" + with self._session_factory() as session: + unit_of_work = SqlAlchemyUnitOfWork(session) + try: + result = operation(SubscriptionSearchOper(session)) + unit_of_work.commit() + return result + except Exception: + unit_of_work.rollback() + raise + + def enqueue( + self, + *, + subscription_ids: tuple[int, ...], + source: str, + priority: int, + ) -> SearchEnqueueResult: + """创建批次并返回 single-flight 合并计数。""" + def operation(repository: SubscriptionSearchOper) -> SearchEnqueueResult: + """在同一事务内创建批次和任务。""" + record, created, coalesced = repository.enqueue( + subscription_ids=subscription_ids, + source=source, + priority=priority, + ) + return SearchEnqueueResult( + batch=_batch(record), + created_count=created, + coalesced_count=coalesced, + ) + + return self._write(operation) + + def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]: + """认领下一任务并返回脱离 Session 的快照。""" + return self._write( + lambda repository: ( + _task(record) + if (record := repository.claim_next(owner=owner, lease_seconds=lease_seconds)) is not None + else None + ) + ) + + def finish_task( + self, + *, + task_id: str, + lease_token: str, + state: str, + error: Optional[str] = None, + ) -> bool: + """以租约令牌收口任务。""" + return self._write( + lambda repository: repository.finish_task( + task_id=task_id, + lease_token=lease_token, + state=state, + error=error, + ) + ) + + def release_task( + self, + *, + task_id: str, + lease_token: str, + cancelled: bool = False, + ) -> bool: + """释放执行租约或收口取消。""" + return self._write( + lambda repository: repository.release_task( + task_id=task_id, + lease_token=lease_token, + cancelled=cancelled, + ) + ) + + def is_cancel_requested(self, task_id: str) -> bool: + """查询任务或批次的取消请求。""" + return self._read(lambda repository: repository.is_cancel_requested(task_id)) + + def request_cancel(self, batch_id: str) -> bool: + """请求取消批次。""" + return self._write(lambda repository: repository.request_cancel(batch_id)) + + def get_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]: + """查询批次聚合快照。""" + return self._read( + lambda repository: ( + _batch(record) if (record := repository.get_batch(batch_id)) is not None else None + ) + ) diff --git a/app/db/models/__init__.py b/app/db/models/__init__.py index b44e8d134..0d19a2c2f 100644 --- a/app/db/models/__init__.py +++ b/app/db/models/__init__.py @@ -35,6 +35,14 @@ _MODEL_EXPORTS = { "app.db.models.subscribehistory", "SubscribeHistory", ), + "SubscriptionSearchBatch": ( + "app.db.models.subscriptionsearch", + "SubscriptionSearchBatch", + ), + "SubscriptionSearchTask": ( + "app.db.models.subscriptionsearch", + "SubscriptionSearchTask", + ), "SystemConfig": ("app.db.models.systemconfig", "SystemConfig"), "TransferHistory": ("app.db.models.transferhistory", "TransferHistory"), "TransferExecutionStep": ( diff --git a/app/db/models/subscriptionsearch.py b/app/db/models/subscriptionsearch.py new file mode 100644 index 000000000..f51dc00bc --- /dev/null +++ b/app/db/models/subscriptionsearch.py @@ -0,0 +1,72 @@ +"""订阅搜索持久批次与任务模型。""" + +from typing import Optional + +from sqlalchemy import Index, Integer, String, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base, get_id_column + + +class SubscriptionSearchBatch(Base): + """记录一次订阅搜索请求及其可观察聚合终态。""" + + id = get_id_column() + batch_id: Mapped[str] = mapped_column(String(64), nullable=False) + source: Mapped[str] = mapped_column(String(32), nullable=False) + state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued") + priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + total_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + finished_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + failed_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + cancelled_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + cancel_requested: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + created_at: Mapped[str] = mapped_column(String(40), nullable=False) + updated_at: Mapped[str] = mapped_column(String(40), nullable=False) + started_at: Mapped[Optional[str]] = mapped_column(String(40)) + finished_at: Mapped[Optional[str]] = mapped_column(String(40)) + last_error: Mapped[Optional[str]] = mapped_column(Text) + + __table_args__ = ( + UniqueConstraint("batch_id", name="uq_subscriptionsearchbatch_batch_id"), + Index("ix_subscriptionsearchbatch_state_created", "state", "created_at", "id"), + ) + + +class SubscriptionSearchTask(Base): + """记录一个具有 single-flight、租约和恢复游标的订阅搜索任务。""" + + id = get_id_column() + task_id: Mapped[str] = mapped_column(String(64), nullable=False) + batch_id: Mapped[str] = mapped_column(String(64), nullable=False) + subscription_id: Mapped[int] = mapped_column(Integer, nullable=False) + active_key: Mapped[Optional[str]] = mapped_column(String(128)) + source: Mapped[str] = mapped_column(String(32), nullable=False) + priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + position: Mapped[int] = mapped_column(Integer, nullable=False) + state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued") + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + cancel_requested: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + lease_owner: Mapped[Optional[str]] = mapped_column(String(128)) + lease_token: Mapped[Optional[str]] = mapped_column(String(64)) + lease_expires_at: Mapped[Optional[str]] = mapped_column(String(40)) + created_at: Mapped[str] = mapped_column(String(40), nullable=False) + updated_at: Mapped[str] = mapped_column(String(40), nullable=False) + started_at: Mapped[Optional[str]] = mapped_column(String(40)) + finished_at: Mapped[Optional[str]] = mapped_column(String(40)) + last_error: Mapped[Optional[str]] = mapped_column(Text) + + __table_args__ = ( + UniqueConstraint("task_id", name="uq_subscriptionsearchtask_task_id"), + UniqueConstraint("active_key", name="uq_subscriptionsearchtask_active_key"), + Index( + "ix_subscriptionsearchtask_claim", + "state", + "priority", + "lease_expires_at", + "created_at", + "id", + ), + Index("ix_subscriptionsearchtask_batch_position", "batch_id", "position", "id"), + Index("ix_subscriptionsearchtask_subscription", "subscription_id", "created_at", "id"), + ) diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py new file mode 100644 index 000000000..d1a660516 --- /dev/null +++ b/app/db/oper/subscriptionsearch.py @@ -0,0 +1,385 @@ +"""订阅搜索批次与任务的持久队列读写。""" + +from datetime import datetime, timedelta, timezone +from typing import Optional +from uuid import uuid4 + +from sqlalchemy import and_, case, func, or_, select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.db.base import DbOper, execute_dml +from app.db.models.subscriptionsearch import ( + SubscriptionSearchBatch, + SubscriptionSearchTask, +) + + +def utc_now_text() -> str: + """返回可按字符串稳定排序的 UTC ISO 时间。""" + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +class SubscriptionSearchOper(DbOper): + """在调用方事务中维护搜索队列、租约和批次聚合。""" + + def enqueue( + self, + *, + subscription_ids: tuple[int, ...], + source: str, + priority: int, + ) -> tuple[SubscriptionSearchBatch, int, int]: + """创建批次，并以活动键合并同一订阅的重叠搜索入口。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅搜索入队需要调用方提供同步 Session") + now = utc_now_text() + batch = SubscriptionSearchBatch( + batch_id=uuid4().hex, + source=source, + state="queued", + priority=priority, + total_count=0, + created_at=now, + updated_at=now, + ) + self._db.add(batch) + self._db.flush() + created = 0 + coalesced = 0 + for position, subscription_id in enumerate(dict.fromkeys(subscription_ids)): + active_key = f"subscription:{subscription_id}" + task = SubscriptionSearchTask( + task_id=uuid4().hex, + batch_id=batch.batch_id, + subscription_id=subscription_id, + active_key=active_key, + source=source, + priority=priority, + position=position, + state="queued", + created_at=now, + updated_at=now, + ) + try: + with self._db.begin_nested(): + self._db.add(task) + self._db.flush() + created += 1 + except IntegrityError: + coalesced += 1 + execute_dml( + self._db, + update(SubscriptionSearchTask) + .where(SubscriptionSearchTask.active_key == active_key) + .values( + priority=case( + (SubscriptionSearchTask.priority < priority, priority), + else_=SubscriptionSearchTask.priority, + ), + updated_at=now, + ), + execution_options={"synchronize_session": False}, + ) + batch.total_count = created + if created == 0: + batch.state = "completed" + batch.finished_at = now + return batch, created, coalesced + + def claim_next(self, *, owner: str, lease_seconds: int) -> Optional[SubscriptionSearchTask]: + """使用 CAS 认领最高优先级任务，过期 running 任务可被恢复。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅搜索认领需要调用方提供同步 Session") + now = utc_now_text() + lease_expires_at = ( + datetime.now(timezone.utc) + timedelta(seconds=max(1, lease_seconds)) + ).isoformat(timespec="seconds") + for _attempt in range(5): + candidate = self._db.execute( + select(SubscriptionSearchTask) + .where( + SubscriptionSearchTask.cancel_requested == 0, + or_( + SubscriptionSearchTask.state == "queued", + and_( + SubscriptionSearchTask.state == "running", + or_( + SubscriptionSearchTask.lease_expires_at.is_(None), + SubscriptionSearchTask.lease_expires_at <= now, + ), + ), + ), + ) + .order_by( + SubscriptionSearchTask.priority.desc(), + SubscriptionSearchTask.created_at.asc(), + SubscriptionSearchTask.position.asc(), + SubscriptionSearchTask.id.asc(), + ) + .limit(1) + ).scalars().first() + if candidate is None: + return None + lease_token = uuid4().hex + claimed = execute_dml( + self._db, + update(SubscriptionSearchTask) + .where( + SubscriptionSearchTask.id == candidate.id, + SubscriptionSearchTask.cancel_requested == 0, + or_( + SubscriptionSearchTask.state == "queued", + and_( + SubscriptionSearchTask.state == "running", + or_( + SubscriptionSearchTask.lease_expires_at.is_(None), + SubscriptionSearchTask.lease_expires_at <= now, + ), + ), + ), + ) + .values( + state="running", + lease_owner=owner, + lease_token=lease_token, + lease_expires_at=lease_expires_at, + attempt_count=SubscriptionSearchTask.attempt_count + 1, + started_at=func.coalesce(SubscriptionSearchTask.started_at, now), + updated_at=now, + ), + execution_options={"synchronize_session": False}, + ) + if not claimed: + self._db.expire_all() + continue + execute_dml( + self._db, + update(SubscriptionSearchBatch) + .where( + SubscriptionSearchBatch.batch_id == candidate.batch_id, + SubscriptionSearchBatch.state.in_(("queued", "running")), + ) + .values( + state="running", + started_at=func.coalesce(SubscriptionSearchBatch.started_at, now), + updated_at=now, + ), + execution_options={"synchronize_session": False}, + ) + self._db.flush() + self._db.expire_all() + return self._db.execute( + select(SubscriptionSearchTask).where( + SubscriptionSearchTask.id == candidate.id, + SubscriptionSearchTask.lease_token == lease_token, + ) + ).scalars().first() + return None + + def finish_task( + self, + *, + task_id: str, + lease_token: str, + state: str, + error: Optional[str], + ) -> bool: + """以当前租约令牌收口任务，并重新计算批次终态。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅搜索收口需要调用方提供同步 Session") + if state not in {"completed", "failed", "cancelled"}: + raise ValueError(f"不支持的订阅搜索终态：{state}") + 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 + 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=state, + active_key=None, + lease_owner=None, + lease_token=None, + lease_expires_at=None, + updated_at=now, + finished_at=now, + last_error=error, + ), + execution_options={"synchronize_session": False}, + ) + if not updated: + return False + self._refresh_batch(task.batch_id, now=now, error=error) + return True + + def release_task( + self, + *, + task_id: str, + lease_token: str, + cancelled: bool, + ) -> 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 + should_cancel = cancelled or bool(task.cancel_requested) or self._batch_cancel_requested(task.batch_id) + if should_cancel: + return self.finish_task( + task_id=task_id, + lease_token=lease_token, + state="cancelled", + error=None, + ) + now = utc_now_text() + return bool(execute_dml( + self._db, + update(SubscriptionSearchTask) + .where( + SubscriptionSearchTask.id == task.id, + SubscriptionSearchTask.state == "running", + SubscriptionSearchTask.lease_token == lease_token, + ) + .values( + state="queued", + lease_owner=None, + lease_token=None, + lease_expires_at=None, + updated_at=now, + ), + execution_options={"synchronize_session": False}, + )) + + def is_cancel_requested(self, task_id: str) -> bool: + """读取任务和批次取消标记。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅搜索取消查询需要调用方提供同步 Session") + task = self._db.execute( + select(SubscriptionSearchTask).where(SubscriptionSearchTask.task_id == task_id) + ).scalars().first() + return bool(task and (task.cancel_requested or self._batch_cancel_requested(task.batch_id))) + + def request_cancel(self, batch_id: str) -> bool: + """标记批次取消，并立即收口尚未发出的排队任务。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅搜索取消需要调用方提供同步 Session") + batch = self._db.execute( + select(SubscriptionSearchBatch).where(SubscriptionSearchBatch.batch_id == batch_id) + ).scalars().first() + if batch is None or batch.state in {"completed", "failed", "cancelled"}: + return False + now = utc_now_text() + execute_dml( + self._db, + update(SubscriptionSearchBatch) + .where(SubscriptionSearchBatch.id == batch.id) + .values(cancel_requested=1, state="cancelling", updated_at=now), + execution_options={"synchronize_session": False}, + ) + execute_dml( + self._db, + update(SubscriptionSearchTask) + .where( + SubscriptionSearchTask.batch_id == batch_id, + SubscriptionSearchTask.state == "queued", + ) + .values( + state="cancelled", + active_key=None, + cancel_requested=1, + finished_at=now, + updated_at=now, + ), + execution_options={"synchronize_session": False}, + ) + execute_dml( + self._db, + update(SubscriptionSearchTask) + .where( + SubscriptionSearchTask.batch_id == batch_id, + SubscriptionSearchTask.state == "running", + ) + .values(cancel_requested=1, updated_at=now), + execution_options={"synchronize_session": False}, + ) + self._refresh_batch(batch_id, now=now, error=None) + return True + + def get_batch(self, batch_id: str) -> Optional[SubscriptionSearchBatch]: + """按稳定批次 ID 读取聚合记录。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅搜索批次查询需要调用方提供同步 Session") + return self._db.execute( + select(SubscriptionSearchBatch).where(SubscriptionSearchBatch.batch_id == batch_id) + ).scalars().first() + + def _batch_cancel_requested(self, batch_id: str) -> bool: + """在当前事务中读取批次取消标记。""" + return bool(self._db.execute( + select(SubscriptionSearchBatch.cancel_requested).where( + SubscriptionSearchBatch.batch_id == batch_id + ) + ).scalar()) + + def _refresh_batch(self, batch_id: str, *, now: str, error: Optional[str]) -> None: + """依据所属任务终态重新计算批次计数和聚合状态。""" + rows = self._db.execute( + select(SubscriptionSearchTask.state, func.count()) # pylint: disable=not-callable + .where(SubscriptionSearchTask.batch_id == batch_id) + .group_by(SubscriptionSearchTask.state) + ).all() + counts = {state: int(count) for state, count in rows} + completed = counts.get("completed", 0) + failed = counts.get("failed", 0) + cancelled = counts.get("cancelled", 0) + terminal = completed + failed + cancelled + batch = self.get_batch(batch_id) + if batch is None: + return + if terminal >= batch.total_count: + if failed: + state = "failed" + elif cancelled or batch.cancel_requested: + state = "cancelled" + else: + state = "completed" + finished_at = now + else: + state = "cancelling" if batch.cancel_requested else "running" + finished_at = None + execute_dml( + self._db, + update(SubscriptionSearchBatch) + .where(SubscriptionSearchBatch.id == batch.id) + .values( + state=state, + finished_count=completed, + failed_count=failed, + cancelled_count=cancelled, + updated_at=now, + finished_at=finished_at, + last_error=error or batch.last_error, + ), + execution_options={"synchronize_session": False}, + ) diff --git a/app/scheduler/catalog.py b/app/scheduler/catalog.py index aff10b69d..c3d4b0887 100644 --- a/app/scheduler/catalog.py +++ b/app/scheduler/catalog.py @@ -142,6 +142,13 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): "subscription", kwargs={"state": "N"}, ), + JobSpec( + "subscribe_search_queue", + "恢复订阅搜索队列", + services.resume_subscribe_search, + "subscription", + recovery=JobRecoveryPolicy.DURABLE_QUEUE, + ), JobSpec("subscribe_refresh", "订阅刷新", services.refresh_subscribe, "subscription"), JobSpec("subscribe_follow", "关注的订阅分享", services.follow_subscribe, "subscription"), JobSpec( @@ -246,6 +253,16 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): ) # 新增订阅时搜索（5分钟检查一次） + self._scheduler.add_job( + self.start, + "interval", + id="subscribe_search_queue", + name="恢复订阅搜索队列", + minutes=1, + next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=10), + kwargs={"job_id": "subscribe_search_queue"}, + ) + self._scheduler.add_job( self.start, "interval", diff --git a/app/scheduler/services.py b/app/scheduler/services.py index 652c027d5..c230edd97 100644 --- a/app/scheduler/services.py +++ b/app/scheduler/services.py @@ -17,6 +17,7 @@ class SchedulerServices: sync_mediaserver: JobCallable check_subscribe: JobCallable search_subscribe: JobCallable + resume_subscribe_search: JobCallable refresh_subscribe: JobCallable follow_subscribe: JobCallable process_transfer: JobCallable diff --git a/app/startup/composition/chain.py b/app/startup/composition/chain.py index 8895232d4..3be18bd01 100644 --- a/app/startup/composition/chain.py +++ b/app/startup/composition/chain.py @@ -91,6 +91,7 @@ def build_chain_runtime_context( module_dispatcher_factory=ModuleInvocationDispatcher, site_repository=dependencies.site, subscription_repository=dependencies.subscription, + subscription_search_repository=dependencies.subscription_search, subscription_mutation_scope=subscription_mutation_scope, sync_subscription_mutation_scope=sync_subscription_mutation_scope, subscription_delete_scope=delete_subscribe_scope, diff --git a/app/startup/composition/runtime.py b/app/startup/composition/runtime.py index ebc5eff64..c71ff5771 100644 --- a/app/startup/composition/runtime.py +++ b/app/startup/composition/runtime.py @@ -48,6 +48,7 @@ from app.startup.composition.context import ( ) if TYPE_CHECKING: + from app.application.subscription.execution import SubscriptionSearchRepository from app.application.messaging.message import MessageHelper, MessageQueueManager from app.startup.composition.agent import AgentComposition from app.startup.composition.configuration import ConfigurationComposition @@ -67,6 +68,7 @@ class RuntimeDependencies: transfer_execution: TransferExecutionRepository message_helper: MessageHelper message_queue: MessageQueueManager + subscription_search: SubscriptionSearchRepository | None = None @dataclass(frozen=True, slots=True) @@ -103,6 +105,7 @@ def compose_runtime_dependencies() -> RuntimeDependencies: TransactionalSubscriptionHistoryRepository, TransactionalSubscriptionRepository, ) + from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository from app.db.adapters.transfer.execution import ( TransactionalTransferExecutionRepository, ) @@ -132,6 +135,7 @@ def compose_runtime_dependencies() -> RuntimeDependencies: transfer_execution=TransactionalTransferExecutionRepository(SessionFactory), message_helper=message_helper_factory(), message_queue=MessageQueueManager(auto_start=False), + subscription_search=TransactionalSubscriptionSearchRepository(SessionFactory), ) diff --git a/app/startup/initializers/scheduler.py b/app/startup/initializers/scheduler.py index 00dd7cbd1..4e2e952c8 100644 --- a/app/startup/initializers/scheduler.py +++ b/app/startup/initializers/scheduler.py @@ -61,6 +61,7 @@ def configure_scheduler_services() -> None: sync_mediaserver=mediaserver_chain.sync, check_subscribe=subscribe_chain.check_and_reconcile, search_subscribe=subscribe_chain.search, + resume_subscribe_search=subscribe_chain.resume_search_queue, refresh_subscribe=subscribe_chain.refresh, follow_subscribe=subscribe_chain.follow, process_transfer=transfer_chain.process, diff --git a/database/versions/c1f4a8d2e6b9_3_0_19.py b/database/versions/c1f4a8d2e6b9_3_0_19.py new file mode 100644 index 000000000..c2d31f00f --- /dev/null +++ b/database/versions/c1f4a8d2e6b9_3_0_19.py @@ -0,0 +1,104 @@ +"""3.0.19 增加订阅搜索持久批次与任务队列。 + +Revision ID: c1f4a8d2e6b9 +Revises: a9d4f2c7e6b1 +Create Date: 2026-09-01 +""" + +# Alembic 的 op 是运行期代理，静态分析无法看到实际操作方法。 +# pylint: disable=no-member + +import sqlalchemy as sa +from alembic import op + +revision = "c1f4a8d2e6b9" +down_revision = "a9d4f2c7e6b1" +branch_labels = None +depends_on = None + +_BATCH_TABLE = "subscriptionsearchbatch" +_TASK_TABLE = "subscriptionsearchtask" + + +def _table_names() -> set[str]: + """返回当前数据库表名集合。""" + return set(sa.inspect(op.get_bind()).get_table_names()) + + +def upgrade() -> None: + """创建跨 SQLite/PostgreSQL 一致的订阅搜索治理表。""" + tables = _table_names() + if _BATCH_TABLE not in tables: + op.create_table( + _BATCH_TABLE, + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("batch_id", sa.String(length=64), nullable=False), + sa.Column("source", sa.String(length=32), nullable=False), + sa.Column("state", sa.String(length=32), nullable=False), + sa.Column("priority", sa.Integer(), nullable=False), + sa.Column("total_count", sa.Integer(), nullable=False), + sa.Column("finished_count", sa.Integer(), nullable=False), + sa.Column("failed_count", sa.Integer(), nullable=False), + sa.Column("cancelled_count", sa.Integer(), nullable=False), + sa.Column("cancel_requested", sa.Integer(), nullable=False), + sa.Column("created_at", sa.String(length=40), nullable=False), + sa.Column("updated_at", sa.String(length=40), nullable=False), + sa.Column("started_at", sa.String(length=40), nullable=True), + sa.Column("finished_at", sa.String(length=40), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.UniqueConstraint("batch_id", name="uq_subscriptionsearchbatch_batch_id"), + ) + op.create_index( + "ix_subscriptionsearchbatch_state_created", + _BATCH_TABLE, + ["state", "created_at", "id"], + ) + if _TASK_TABLE not in tables: + op.create_table( + _TASK_TABLE, + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("task_id", sa.String(length=64), nullable=False), + sa.Column("batch_id", sa.String(length=64), nullable=False), + sa.Column("subscription_id", sa.Integer(), nullable=False), + sa.Column("active_key", sa.String(length=128), nullable=True), + sa.Column("source", sa.String(length=32), nullable=False), + sa.Column("priority", sa.Integer(), nullable=False), + sa.Column("position", sa.Integer(), nullable=False), + sa.Column("state", sa.String(length=32), nullable=False), + sa.Column("attempt_count", sa.Integer(), nullable=False), + sa.Column("cancel_requested", sa.Integer(), nullable=False), + sa.Column("lease_owner", sa.String(length=128), nullable=True), + sa.Column("lease_token", sa.String(length=64), nullable=True), + sa.Column("lease_expires_at", sa.String(length=40), nullable=True), + sa.Column("created_at", sa.String(length=40), nullable=False), + sa.Column("updated_at", sa.String(length=40), nullable=False), + sa.Column("started_at", sa.String(length=40), nullable=True), + sa.Column("finished_at", sa.String(length=40), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.UniqueConstraint("task_id", name="uq_subscriptionsearchtask_task_id"), + sa.UniqueConstraint("active_key", name="uq_subscriptionsearchtask_active_key"), + ) + op.create_index( + "ix_subscriptionsearchtask_claim", + _TASK_TABLE, + ["state", "priority", "lease_expires_at", "created_at", "id"], + ) + op.create_index( + "ix_subscriptionsearchtask_batch_position", + _TASK_TABLE, + ["batch_id", "position", "id"], + ) + op.create_index( + "ix_subscriptionsearchtask_subscription", + _TASK_TABLE, + ["subscription_id", "created_at", "id"], + ) + + +def downgrade() -> None: + """移除订阅搜索治理表。""" + tables = _table_names() + if _TASK_TABLE in tables: + op.drop_table(_TASK_TABLE) + if _BATCH_TABLE in tables: + op.drop_table(_BATCH_TABLE) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index c4237083b..f350f3a73 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -754,8 +754,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 919 | -| 内部导入边 | 7,690 | +| Python 模块 | 925 | +| 内部导入边 | 7,728 | | 非平凡 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 a128011f6..2f4470d99 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 919 / 7,690 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 925 / 7,728 | `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/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index 9a0c26095..a28f3fd92 100644 --- a/docs/refactor/moviepilot-subscription-governance-roadmap.md +++ b/docs/refactor/moviepilot-subscription-governance-roadmap.md @@ -1,7 +1,7 @@ # MoviePilot 订阅执行治理 > 状态：`active（2026-09-01 已由 MoviePilot v3 接管）` -> 当前叶：`SUB-GOV-002A` +> 当前叶：`SUB-GOV-002B` > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、 > `V3-RDY-009A1C`、`V3-RDY-009A1D` > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环 @@ -232,14 +232,14 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 | `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `completed（2026-09-01）` | | `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `completed（2026-09-01）` | | `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `not_activated（2026-09-01）` | -| `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `in_progress` | -| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `pending` | +| `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `completed（2026-09-01）` | +| `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `in_progress` | | `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `pending` | | `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `pending` | | `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` | | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` | -当前只激活 `SUB-GOV-002A`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B +当前只激活 `SUB-GOV-002B`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。 ### 6.1 SUB-GOV-001A 验收证据 @@ -299,6 +299,21 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 等待仍主导批次的同 revision 证据；因此不激活 2/4 worker，不引入共享客户端线程安全与提交排序风险； - 后续若受控生产形态证据再次显示准备等待占主导，可从本条件叶重新开启，但不得绕过 SiteBudget 和串行提交。 +### 6.6 SUB-GOV-002A 验收证据 + +- 新增持久 `SubscriptionSearchBatch` 与 `SubscriptionSearchTask`，通过 Alembic `c1f4a8d2e6b9` 同时支持 + SQLite/PostgreSQL；批次记录聚合终态，任务记录稳定位置、优先级、attempt、租约和取消标记； +- 可空唯一 `active_key=subscription:` 是订阅级 single-flight owner：手工/定向任务只提高已排队任务优先级， + 不为同一订阅创建第二个活动任务；终态清空 active key，允许后续正常周期重新搜索； +- 配置了正式队列的 Search 路径不再取得 Match 类级锁，也不再执行逐订阅 `60–300s` 随机休眠；单任务异常 + 独立标记 failed 并继续后续订阅，批次最终暴露聚合失败； +- running 任务使用 token fencing 与 UTC 到期租约，进程重启后以相同 task identity、递增 attempt 恢复；停止时 + 可退回 queued，取消会立即终止未发请求任务并在运行任务返回租约边界时收口； +- 新增 `subscribe_search_queue` durable 调度，启动后 10 秒首次恢复、之后每分钟消费，且使用独立消费者锁保持 + 单进程串行；它不创建 24 小时兜底批次，也不阻塞日常 Match； +- 验证：队列/编排/调度专项 `43 passed`，迁移与声明式模型 `35 passed`，复杂度/并发/架构/迁移组合 + `119 passed`，错误级 Pylint 为 0，Alembic 唯一 head 为 `c1f4a8d2e6b9`。 + ## 7. 上线前验证与验收 ### 7.1 场景 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index c52557d1a..b1b952d46 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 7711, - "edge_sha256": "f1ff788de28fa0c486e529afa4dc7af5038587f997bf155c5fc434754ec2a4a3", + "edge_count": 7728, + "edge_sha256": "33cdb555e19d7a5d15dbed065c98908269c3528ec4d61c2ac5515420c0b4cfa4", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -4586,6 +4586,7 @@ "app.chain.subscribe.search -> app.application.configuration", "app.chain.subscribe.search -> app.application.subscription", "app.chain.subscribe.search -> app.application.subscription.contract", + "app.chain.subscribe.search -> app.application.subscription.execution", "app.chain.subscribe.search -> app.application.subscription.query", "app.chain.subscribe.search -> app.chain", "app.chain.subscribe.search -> app.chain.media", @@ -5130,6 +5131,15 @@ "app.db.adapters.subscription -> app.schemas", "app.db.adapters.subscription -> app.schemas.common", "app.db.adapters.subscription -> app.schemas.types", + "app.db.adapters.subscriptionsearch -> app.application", + "app.db.adapters.subscriptionsearch -> app.application.subscription", + "app.db.adapters.subscriptionsearch -> app.application.subscription.execution", + "app.db.adapters.subscriptionsearch -> app.db", + "app.db.adapters.subscriptionsearch -> app.db.models", + "app.db.adapters.subscriptionsearch -> app.db.models.subscriptionsearch", + "app.db.adapters.subscriptionsearch -> app.db.oper", + "app.db.adapters.subscriptionsearch -> app.db.oper.subscriptionsearch", + "app.db.adapters.subscriptionsearch -> app.db.uow", "app.db.adapters.transaction -> app.db", "app.db.adapters.transaction -> app.db.uow", "app.db.adapters.transfer.admission -> app.application", @@ -5268,6 +5278,8 @@ "app.db.models.subscribehistory -> app.db.models._constraints", "app.db.models.subscribehistory -> app.schemas", "app.db.models.subscribehistory -> app.schemas.types", + "app.db.models.subscriptionsearch -> app.db", + "app.db.models.subscriptionsearch -> app.db.base", "app.db.models.systemconfig -> app.db", "app.db.models.systemconfig -> app.db.base", "app.db.models.transferexecutionstep -> app.db", @@ -5374,6 +5386,10 @@ "app.db.oper.subscribehistory -> app.schemas.common", "app.db.oper.subscribehistory -> app.schemas.query", "app.db.oper.subscribehistory -> app.schemas.types", + "app.db.oper.subscriptionsearch -> app.db", + "app.db.oper.subscriptionsearch -> app.db.base", + "app.db.oper.subscriptionsearch -> app.db.models", + "app.db.oper.subscriptionsearch -> app.db.models.subscriptionsearch", "app.db.oper.systemconfig -> app.db", "app.db.oper.systemconfig -> app.db.base", "app.db.oper.systemconfig -> app.db.models", @@ -8216,6 +8232,7 @@ "app.startup.composition.runtime -> app.db.adapters.outbox", "app.startup.composition.runtime -> app.db.adapters.site", "app.startup.composition.runtime -> app.db.adapters.subscription", + "app.startup.composition.runtime -> app.db.adapters.subscriptionsearch", "app.startup.composition.runtime -> app.db.adapters.transfer", "app.startup.composition.runtime -> app.db.adapters.transfer.execution", "app.startup.composition.runtime -> app.db.oper", @@ -8804,7 +8821,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 921, + "module_count": 925, "modules": [ "app", "app.adapters", @@ -9095,6 +9112,7 @@ "app.application.subscription.complete", "app.application.subscription.contract", "app.application.subscription.delete", + "app.application.subscription.execution", "app.application.subscription.facts", "app.application.subscription.identity", "app.application.subscription.mutation", @@ -9229,6 +9247,7 @@ "app.db.adapters.query", "app.db.adapters.site", "app.db.adapters.subscription", + "app.db.adapters.subscriptionsearch", "app.db.adapters.transaction", "app.db.adapters.transfer", "app.db.adapters.transfer.admission", @@ -9262,6 +9281,7 @@ "app.db.models.siteuserdata", "app.db.models.subscribe", "app.db.models.subscribehistory", + "app.db.models.subscriptionsearch", "app.db.models.systemconfig", "app.db.models.transferexecutionstep", "app.db.models.transferhistory", @@ -9284,6 +9304,7 @@ "app.db.oper.site", "app.db.oper.subscribe", "app.db.oper.subscribehistory", + "app.db.oper.subscriptionsearch", "app.db.oper.systemconfig", "app.db.oper.transferexecutionstep", "app.db.oper.transferhistory", diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py new file mode 100644 index 000000000..6b052e7aa --- /dev/null +++ b/tests/test_subscription_search_governance.py @@ -0,0 +1,116 @@ +"""订阅搜索队列接入、锁隔离和批次失败治理测试。""" + +from dataclasses import replace +from datetime import datetime, timedelta +from unittest.mock import Mock, patch + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from app.application.subscription.contract import SubscriptionSnapshot +from app.chain.subscribe.facade import SubscribeChain +from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository +from app.db.base import Base +from app.schemas.types import MediaType + + +class _ForbiddenLock: + """正式队列路径若仍访问 Match 全局锁则立即失败。""" + + def acquire(self, **_kwargs): + """禁止搜索队列取得历史长锁。""" + raise AssertionError("持久搜索队列不应取得 Match 全局锁") + + def release(self): + """禁止搜索队列释放从未取得的历史长锁。""" + raise AssertionError("持久搜索队列不应释放 Match 全局锁") + + +class _SubscriptionRepository: + """为搜索队列返回稳定的多订阅快照。""" + + def __init__(self, subscribes: list[SubscriptionSnapshot]) -> None: + self._subscribes = {subscribe.id: subscribe for subscribe in subscribes} + + def list(self, _state: str = None) -> list[SubscriptionSnapshot]: + """返回全部测试订阅。""" + return list(self._subscribes.values()) + + def get(self, subscribe_id: int) -> SubscriptionSnapshot | None: + """按 ID 返回任务执行时重新读取的订阅。""" + return self._subscribes.get(subscribe_id) + + +def _subscribe(subscribe_id: int) -> SubscriptionSnapshot: + """构造越过新增保护期的活动电影订阅。""" + return SubscriptionSnapshot( + id=subscribe_id, + name=f"治理电影 {subscribe_id}", + year="2026", + type=MediaType.MOVIE.value, + media_source="themoviedb", + media_id=str(1000 + subscribe_id), + state="R", + date=(datetime.now() - timedelta(minutes=2)).strftime("%Y-%m-%d %H:%M:%S"), + ) + + +def _chain(tmp_path, subscribes: list[SubscriptionSnapshot]): + """构造注入持久队列且不初始化其他 Chain 依赖的搜索实例。""" + engine = create_engine(f"sqlite:///{tmp_path / 'search-governance.db'}") + Base.metadata.create_all(engine) + chain = object.__new__(SubscribeChain) + chain.subscription_repository = _SubscriptionRepository(subscribes) + chain.subscription_search_repository = TransactionalSubscriptionSearchRepository( + sessionmaker(bind=engine) + ) + chain.get_states_for_search = lambda state: state + chain._rlock = _ForbiddenLock() + return chain + + +def test_fallback_queue_executes_without_match_global_lock(tmp_path, monkeypatch): + """R/P 兜底搜索在持久队列中执行，不受日常 Match 长锁阻塞。""" + subscribes = [_subscribe(1), _subscribe(2)] + chain = _chain(tmp_path, subscribes) + processed = [] + monkeypatch.setattr( + chain, + "_process_search_subscription", + lambda subscribe, _searchchain: processed.append(subscribe.id) or subscribe, + ) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(state="R") + + batch = chain.get_search_batch(batch_id) + assert processed == [1, 2] + assert batch.state == "completed" + assert batch.finished_count == 2 + assert batch.failed_count == 0 + + +def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monkeypatch): + """单订阅异常不得中止批次后续任务，聚合终态必须暴露失败。""" + subscribes = [_subscribe(3), _subscribe(4)] + chain = _chain(tmp_path, subscribes) + processed = [] + + def process(subscribe, _searchchain): + """让首条失败并保持第二条正常完成。""" + processed.append(subscribe.id) + if subscribe.id == 3: + raise RuntimeError("provider timeout") + return replace(subscribe) + + 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) + assert processed == [3, 4] + assert batch.state == "failed" + assert batch.finished_count == 1 + assert batch.failed_count == 1 + assert batch.last_error == "provider timeout" diff --git a/tests/test_subscription_search_queue.py b/tests/test_subscription_search_queue.py new file mode 100644 index 000000000..25757da61 --- /dev/null +++ b/tests/test_subscription_search_queue.py @@ -0,0 +1,132 @@ +"""订阅搜索持久队列、single-flight、租约和取消测试。""" + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import create_engine, select, update +from sqlalchemy.orm import Session, sessionmaker + +from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository +from app.db.base import Base +from app.db.models.subscriptionsearch import SubscriptionSearchTask + + +def _repository(tmp_path): + """构造使用独立 SQLite 文件的事务型搜索队列。""" + engine = create_engine(f"sqlite:///{tmp_path / 'search-queue.db'}") + Base.metadata.create_all(engine) + return TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)), engine + + +def test_search_queue_coalesces_active_subscription_and_raises_priority(tmp_path): + """重叠入口只保留一个活动任务，手工请求可提高优先级。""" + repository, _engine = _repository(tmp_path) + + scheduled = repository.enqueue( + subscription_ids=(1, 2), + source="fallback", + priority=10, + ) + manual = repository.enqueue( + subscription_ids=(1,), + source="manual", + priority=100, + ) + + first = repository.claim_next(owner="worker-a") + second = repository.claim_next(owner="worker-b") + + assert scheduled.created_count == 2 + assert scheduled.coalesced_count == 0 + assert manual.created_count == 0 + assert manual.coalesced_count == 1 + assert manual.batch.state == "completed" + assert first.subscription_id == 1 + assert first.priority == 100 + assert second.subscription_id == 2 + assert first.task_id != second.task_id + + +def test_search_queue_recovers_expired_lease_with_same_task_identity(tmp_path): + """进程遗留的过期 running 任务应以新 token 恢复且 attempt 单调递增。""" + repository, engine = _repository(tmp_path) + repository.enqueue(subscription_ids=(3,), source="fallback", priority=10) + first = repository.claim_next(owner="worker-a", lease_seconds=900) + expired_at = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat(timespec="seconds") + with Session(engine) as session: + session.execute( + update(SubscriptionSearchTask) + .where(SubscriptionSearchTask.task_id == first.task_id) + .values(lease_expires_at=expired_at) + ) + session.commit() + + recovered = repository.claim_next(owner="worker-b", lease_seconds=900) + + assert recovered.task_id == first.task_id + assert recovered.subscription_id == first.subscription_id + assert recovered.lease_token != first.lease_token + assert recovered.attempt_count == 2 + + +def test_search_queue_cancel_finishes_queued_and_running_tasks(tmp_path): + """取消立即终止未发请求任务，运行中任务在租约边界收口。""" + repository, engine = _repository(tmp_path) + enqueued = repository.enqueue( + subscription_ids=(4, 5), + source="fallback", + priority=10, + ) + running = repository.claim_next(owner="worker-a") + + assert repository.request_cancel(enqueued.batch.batch_id) is True + assert repository.is_cancel_requested(running.task_id) is True + assert repository.release_task( + task_id=running.task_id, + lease_token=running.lease_token, + cancelled=True, + ) is True + + batch = repository.get_batch(enqueued.batch.batch_id) + with Session(engine) as session: + states = list( + session.execute( + select(SubscriptionSearchTask.state) + .where(SubscriptionSearchTask.batch_id == enqueued.batch.batch_id) + .order_by(SubscriptionSearchTask.position) + ).scalars() + ) + + assert states == ["cancelled", "cancelled"] + assert batch.state == "cancelled" + assert batch.cancelled_count == 2 + assert repository.claim_next(owner="worker-b") is None + + +def test_search_queue_finishes_batch_with_aggregated_failure(tmp_path): + """单任务失败不阻止后续任务，但批次最终暴露聚合失败。""" + repository, _engine = _repository(tmp_path) + enqueued = repository.enqueue( + subscription_ids=(6, 7), + source="fallback", + priority=10, + ) + first = repository.claim_next(owner="worker-a") + assert repository.finish_task( + task_id=first.task_id, + lease_token=first.lease_token, + state="failed", + error="site timeout", + ) is True + second = repository.claim_next(owner="worker-a") + assert repository.finish_task( + task_id=second.task_id, + lease_token=second.lease_token, + state="completed", + ) is True + + batch = repository.get_batch(enqueued.batch.batch_id) + + assert batch.state == "failed" + assert batch.finished_count == 1 + assert batch.failed_count == 1 + assert batch.last_error == "site timeout"