From 94716484b759c0a77d59851b04acf0bb2a532fea Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 1 Sep 2026 07:11:06 +0800 Subject: [PATCH] fix(transfer): resolve durable failure recovery issues --- app/api/dependencies/history.py | 1 + app/application/historymutation.py | 37 +++++- app/application/transfer/execution.py | 35 +++++ app/application/transfer/workflow.py | 4 + app/chain/base.py | 2 +- app/chain/transfer/plan.py | 74 +++++++---- app/chain/transfer/records.py | 33 +++-- app/chain/transfer/workflow.py | 29 ++-- app/db/adapters/transfer/execution.py | 106 +++++++++++++++ app/db/models/transferhistory.py | 35 +++++ app/db/models/transferpending.py | 50 ++++++- app/db/oper/transferhistory.py | 17 +++ app/db/oper/transferpending.py | 17 +++ app/modules/filemanager/transhandler.py | 3 +- tests/test_chain_durable_events.py | 1 + tests/test_history_mutation_command.py | 52 +++++++- tests/test_manual_transfer_history.py | 55 ++++++++ tests/test_transfer_durable_retry_owner.py | 132 ++++++++++++++++--- tests/test_transfer_execution_persistence.py | 125 ++++++++++++++++++ tests/test_transfer_failed_retry_budget.py | 3 +- tests/test_transfer_planning_checkpoint.py | 27 ++++ tests/test_transfer_planning_compat.py | 4 +- 22 files changed, 762 insertions(+), 80 deletions(-) diff --git a/app/api/dependencies/history.py b/app/api/dependencies/history.py index d6941fe72..bafa16fcd 100644 --- a/app/api/dependencies/history.py +++ b/app/api/dependencies/history.py @@ -91,6 +91,7 @@ def get_transfer_history_mutation_command( return TransferHistoryMutationCommand( repository=runtime.history.transfer_mutation_repository(db), download_repository=runtime.history.download_repository(db), + transfer_execution_repository=runtime.history.transfer_execution_repository, unit_of_work=runtime.persistence.sync_transaction(db), file_item_factory=lambda payload: _SchemaFileItem(**payload), file_exists=storage_chain.exists, diff --git a/app/application/historymutation.py b/app/application/historymutation.py index 7c80adecf..41ee9b0df 100644 --- a/app/application/historymutation.py +++ b/app/application/historymutation.py @@ -6,6 +6,10 @@ from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional, Protocol +from app.application.transfer.execution import ( + TransferExecutionCommand, + TransferExecutionRepository, +) from app.schemas.common import JsonData from app.schemas.history import TransferHistoryDeleteResult, TransferHistoryDeleteStep @@ -34,6 +38,11 @@ class TransferHistoryMutationRecord(Protocol): """返回持久整理任务标识。""" ... + @property + def transfer_settlement_revision(self) -> Optional[int]: + """返回持久整理失败回执的结算版本。""" + ... + @property def src(self) -> Optional[str]: """返回源路径。""" @@ -132,6 +141,7 @@ class TransferHistoryMutationCommand: *, repository: TransferHistoryMutationRepository, download_repository: DownloadFileMutationRepository, + transfer_execution_repository: TransferExecutionRepository, unit_of_work: HistoryUnitOfWork, file_item_factory: Callable[[dict[str, JsonData]], Any], delete_media_file: Callable[[Any], bool], @@ -142,6 +152,7 @@ class TransferHistoryMutationCommand: """保存历史事务、存储删除、事件和失败状态清理端口。""" self._repository = repository self._download_repository = download_repository + self._transfer_execution_repository = transfer_execution_repository self._unit_of_work = unit_of_work self._file_item_factory = file_item_factory self._file_exists = file_exists or (lambda _fileitem: True) @@ -166,12 +177,28 @@ class TransferHistoryMutationCommand: message="记录不存在", ) if history.transfer_task_id: - return TransferHistoryDeleteResult( - source=TransferHistoryDeleteStep(status="not_requested"), - destination=TransferHistoryDeleteStep(status="not_requested"), - history="retained", - message="持久整理失败记录不可删除，请使用重试或人工复核入口", + settlement_revision = history.transfer_settlement_revision + if not settlement_revision: + return TransferHistoryDeleteResult( + source=TransferHistoryDeleteStep(status="not_requested"), + destination=TransferHistoryDeleteStep(status="not_requested"), + history="retained", + message="持久整理失败记录缺少结算版本，请刷新后重试", + ) + discard = TransferExecutionCommand( + self._transfer_execution_repository + ).discard_failed( + task_id=history.transfer_task_id, + history_id=history_id, + settlement_revision=settlement_revision, ) + if not discard.discarded: + return TransferHistoryDeleteResult( + source=TransferHistoryDeleteStep(status="not_requested"), + destination=TransferHistoryDeleteStep(status="not_requested"), + history="retained", + message=discard.message, + ) destination_result = self._delete_file( history.dest_fileitem if delete_destination else None, diff --git a/app/application/transfer/execution.py b/app/application/transfer/execution.py index 7e145cbe4..2b55c8d6f 100644 --- a/app/application/transfer/execution.py +++ b/app/application/transfer/execution.py @@ -435,6 +435,15 @@ class TransferRetryRequestResult: message: str +@dataclass(frozen=True, slots=True) +class TransferFailureDiscardResult: + """描述终态失败任务是否已解除历史绑定并删除执行证据。""" + + discarded: bool + state: Optional[TransferExecutionState] + message: str + + @dataclass(frozen=True, slots=True) class TransferManualReviewResult: """描述一次已持久审计的人工判定及后续调度状态。""" @@ -595,6 +604,15 @@ class TransferExecutionRepository(Protocol): ) -> TransferRetryRequestResult: """仅把 FAILED 终态转入到期可 claim 的 retry_wait。""" + def discard_failed( + self, + *, + task_id: str, + history_id: int, + settlement_revision: int, + ) -> TransferFailureDiscardResult: + """仅放弃与指定失败回执完全匹配的无租约 FAILED 任务。""" + def resolve_manual_review( self, *, @@ -810,6 +828,22 @@ class TransferExecutionCommand: requested_by=requested_by, ) + def discard_failed( + self, + *, + task_id: str, + history_id: int, + settlement_revision: int, + ) -> TransferFailureDiscardResult: + """放弃确定失败任务，使对应历史恢复为普通可维护记录。""" + if not task_id or history_id <= 0 or settlement_revision <= 0: + raise ValueError("放弃失败整理任务缺少任务、历史或结算版本") + return self._repository.discard_failed( + task_id=task_id, + history_id=history_id, + settlement_revision=settlement_revision, + ) + def resolve_manual_review( self, *, @@ -924,6 +958,7 @@ __all__ = [ "TransferExecutionCommand", "TransferExecutionConflictError", "TransferExecutionError", + "TransferFailureDiscardResult", "TransferExecutionLeaseLostError", "TransferOperationObservation", "TransferOperationObservationState", diff --git a/app/application/transfer/workflow.py b/app/application/transfer/workflow.py index 9f0c14010..193ddf13c 100644 --- a/app/application/transfer/workflow.py +++ b/app/application/transfer/workflow.py @@ -705,6 +705,10 @@ class TransferAdmissionConflictError(ValueError): """同一源文件以不同规划输入重复准入时抛出的冲突错误。""" +class TransferPlanningRejectedError(ValueError): + """规划输入缺少业务必需信息且继续自动重试不会改变结果。""" + + class TransferPlanningStateError(RuntimeError): """计划检查点无法从当前持久状态推进时抛出的状态错误。""" diff --git a/app/chain/base.py b/app/chain/base.py index 83da87c2a..50571ea97 100644 --- a/app/chain/base.py +++ b/app/chain/base.py @@ -951,7 +951,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, """调用文件管理模块生成无文件写入的冻结整理计划。""" return cast( "Optional[TransferPlanCheckpoint]", - self.run_module( + self.run_module_strict( "plan_transfer", fileitem=fileitem, meta=meta, diff --git a/app/chain/transfer/plan.py b/app/chain/transfer/plan.py index 5059c8e5d..46380477c 100644 --- a/app/chain/transfer/plan.py +++ b/app/chain/transfer/plan.py @@ -17,6 +17,7 @@ from app.application.transfer.workflow import ( TransferLeaseLostError, TransferPlanCheckpoint, TransferPlanningInput, + TransferPlanningRejectedError, TransferPlanningStateError, TransferProviderInvocationSnapshot, TransferProviderReference, @@ -274,13 +275,34 @@ class TransferPlanningOwner(_TransferOwnerBase): ) planning_input = task.planning_input or self._TransferChain__build_planning_input(task) task.bind_planning_input(planning_input) + checkpoint = self._TransferChain__build_planning_rejection_checkpoint( + task, + error=error, + planning_input=planning_input, + ) + persisted = self._TransferChain__persist_transfer_checkpoint( + task, + planning_input=planning_input, + checkpoint=checkpoint, + ) + task.bind_plan_checkpoint(persisted) + return self._plan_checkpoint_and_execute(task) + + def _TransferChain__build_planning_rejection_checkpoint( + self, + task: TransferTask, + *, + error: str, + planning_input: TransferPlanningInput, + ) -> TransferPlanCheckpoint: + """把确定性的宿主规划错误冻结为可重放的零步骤失败计划。""" source_path = task.fileitem.path resolved_transfer_type = ( task.transfer_type or planning_input.requested_transfer_type or "copy" ) - checkpoint = TransferPlanCheckpoint( + return TransferPlanCheckpoint( planning_input=planning_input, target_storage=task.fileitem.storage, root_target_path=source_path, @@ -300,13 +322,6 @@ class TransferPlanningOwner(_TransferOwnerBase): overwrite_mode=planning_input.overwrite_mode, rejection_error=error, ) - persisted = self._TransferChain__persist_transfer_checkpoint( - task, - planning_input=planning_input, - checkpoint=checkpoint, - ) - task.bind_plan_checkpoint(persisted) - return self._plan_checkpoint_and_execute(task) def _TransferChain__execute_planning_rejection( self, @@ -675,25 +690,32 @@ class TransferPlanningOwner(_TransferOwnerBase): source_oper: Any, ) -> TransferPlanCheckpoint: """在 provider 未接管时生成纯宿主计划，不执行任何文件写入。""" - checkpoint = cast( - Optional[TransferPlanCheckpoint], - self.plan_transfer( - fileitem=task.fileitem, - meta=cast(MetaBase, task.meta), - mediainfo=cast(Union[MediaInfo, MusicInfo], task.mediainfo), - target_directory=task.target_directory, - target_storage=task.target_storage, - target_path=task.target_path, - transfer_type=task.transfer_type, - episodes_info=task.episodes_info, - scrape=task.scrape, - library_type_folder=task.library_type_folder, - library_category_folder=task.library_category_folder, - source_oper=source_oper, - preview=bool(task.preview), + try: + checkpoint = cast( + Optional[TransferPlanCheckpoint], + self.plan_transfer( + fileitem=task.fileitem, + meta=cast(MetaBase, task.meta), + mediainfo=cast(Union[MediaInfo, MusicInfo], task.mediainfo), + target_directory=task.target_directory, + target_storage=task.target_storage, + target_path=task.target_path, + transfer_type=task.transfer_type, + episodes_info=task.episodes_info, + scrape=task.scrape, + library_type_folder=task.library_type_folder, + library_category_folder=task.library_category_folder, + source_oper=source_oper, + preview=bool(task.preview), + planning_input=planning_input, + ), + ) + except TransferPlanningRejectedError as error: + return self._TransferChain__build_planning_rejection_checkpoint( + task, + error=str(error), planning_input=planning_input, - ), - ) + ) if checkpoint is None: raise RuntimeError("文件整理模块未返回规划检查点") return checkpoint diff --git a/app/chain/transfer/records.py b/app/chain/transfer/records.py index 3fb38f443..4f9935a63 100644 --- a/app/chain/transfer/records.py +++ b/app/chain/transfer/records.py @@ -13,6 +13,7 @@ from app.application.history import ( resolve_history, ) from app.application.transfer.execution import ( + TransferExecutionCommand, TransferExecutionRepository, ) from app.chain._contracts import TransferMixinHost @@ -477,8 +478,8 @@ class ManualHistoryMixin(_TransferOwnerBase): ) -> Optional[Tuple[bool, str]]: """将 durable 历史重试交还持久调度器，旧历史返回 ``None``。 - durable 任务的历史、目标文件和步骤证据共同描述一次可恢复执行。任何历史 - 入口都只能登记重试意图，不能删除这些证据后重新准入一条并行任务。 + 普通重试和 AI 接管只登记重试意图；显式重新整理由 + ``_delete_manual_transfer_history`` 先放弃确定失败任务，再重新准入。 :param history: 整理历史 :param requested_by: 发起重试的稳定入口身份 @@ -495,15 +496,25 @@ class ManualHistoryMixin(_TransferOwnerBase): history: TransferHistorySnapshot, transfer_history_oper: TransferHistoryRepository, ) -> Tuple[bool, str]: - """删除手动重整历史；非成功移动记录同时清理可能存在的旧目标。""" - durable_retry = self._request_durable_transfer_retry( - history, - requested_by="manual_reorganize", - ) - if durable_retry is not None: - _, message = durable_retry - # 返回 False 阻止旧 caller 把“已登记”误当成“历史已删除”并继续执行。 - return False, message + """删除手动重整历史；失败 durable 回执先原子放弃再清理旧目标。""" + task_id = getattr(history, "transfer_task_id", None) + if task_id: + settlement_revision = getattr( + history, + "transfer_settlement_revision", + None, + ) + if not settlement_revision: + return False, "持久整理失败记录缺少结算版本，请刷新后重试" + discard = TransferExecutionCommand( + self.transfer_execution_repository + ).discard_failed( + task_id=task_id, + history_id=history.id, + settlement_revision=settlement_revision, + ) + if not discard.discarded: + return False, discard.message if ( history.dest_fileitem and not ManualHistoryMixin._is_successful_move_history(history) diff --git a/app/chain/transfer/workflow.py b/app/chain/transfer/workflow.py index 7da42772d..e4d1296a8 100644 --- a/app/chain/transfer/workflow.py +++ b/app/chain/transfer/workflow.py @@ -710,20 +710,21 @@ class TransferWorkflowOwner(_TransferOwnerBase): reorganize or not transferd.status ) if should_reorganize: - durable_retry = self._request_durable_transfer_retry( - transferd, - requested_by="manual_reorganize", - ) - if durable_retry is not None: - accepted, message = durable_retry - if accepted: - logger.info(message) - else: - all_success = False - logger.error(message) - err_msgs.append(message) - # durable 历史只登记唯一调度重试，不在当前调用重新准入。 - continue + if not reorganize: + durable_retry = self._request_durable_transfer_retry( + transferd, + requested_by="manual_reorganize", + ) + if durable_retry is not None: + accepted, message = durable_retry + if accepted: + logger.info(message) + else: + all_success = False + logger.error(message) + err_msgs.append(message) + # 普通失败重试仍由唯一 durable 调度器继续原计划。 + continue state, message = self._delete_manual_transfer_history( history=transferd, transfer_history_oper=transfer_history_oper, diff --git a/app/db/adapters/transfer/execution.py b/app/db/adapters/transfer/execution.py index c2558fb75..75a06c257 100644 --- a/app/db/adapters/transfer/execution.py +++ b/app/db/adapters/transfer/execution.py @@ -18,6 +18,7 @@ from app.application.transfer.execution import ( TransferExecutionSnapshot, TransferExecutionState, TransferExecutionStep, + TransferFailureDiscardResult, TransferManualReviewDecision, TransferManualReviewPage, TransferManualReviewResult, @@ -42,6 +43,7 @@ from app.db.models.transferexecutionstep import ( ) from app.db.models.transferpending import TransferPending from app.db.oper.transferexecutionstep import TransferExecutionStepOper +from app.db.oper.transferhistory import TransferHistoryOper from app.db.oper.transferpending import TransferPendingOper from app.db.uow import SqlAlchemyUnitOfWork @@ -1197,6 +1199,110 @@ class TransactionalTransferExecutionRepository: self._rollback(transaction) raise + def discard_failed( + self, + *, + task_id: str, + history_id: int, + settlement_revision: int, + ) -> TransferFailureDiscardResult: + """原子删除指定 FAILED pending、步骤证据并解除历史回执映射。""" + with self._session_factory() as session: + transaction = SqlAlchemyUnitOfWork(session) + try: + pending_oper = TransferPendingOper(session) + history_oper = TransferHistoryOper(session) + pending = pending_oper.get_by_task_id(task_id=task_id) + if pending is None: + history = history_oper.get(history_id) + if history is None or history.transfer_task_id is None: + return TransferFailureDiscardResult( + discarded=True, + state=None, + message="失败整理任务已由并发请求放弃", + ) + return TransferFailureDiscardResult( + discarded=False, + state=None, + message="未找到与失败历史匹配的整理任务，请刷新后重试", + ) + + state = TransferExecutionState(pending.execution_state) + if state is not TransferExecutionState.FAILED: + message = ( + "人工复核任务必须先完成专门判定" + if state is TransferExecutionState.MANUAL_REVIEW + else "整理任务当前状态不能放弃" + ) + return TransferFailureDiscardResult( + discarded=False, + state=state, + message=message, + ) + if pending.lease_owner is not None or pending.lease_token is not None: + return TransferFailureDiscardResult( + discarded=False, + state=state, + message="整理任务仍被执行器占用，不能放弃", + ) + if ( + pending.terminal_history_id != history_id + or pending.settlement_revision != settlement_revision + ): + return TransferFailureDiscardResult( + discarded=False, + state=state, + message="整理任务失败回执已变化，请刷新后重试", + ) + + deleted = pending_oper.stage_delete_terminal_failure( + task_id=task_id, + history_id=history_id, + settlement_revision=settlement_revision, + ) + if deleted != 1: + session.expire_all() + pending = pending_oper.get_by_task_id(task_id=task_id) + history = history_oper.get(history_id) + if pending is None and ( + history is None or history.transfer_task_id is None + ): + return TransferFailureDiscardResult( + discarded=True, + state=None, + message="失败整理任务已由并发请求放弃", + ) + return TransferFailureDiscardResult( + discarded=False, + state=( + TransferExecutionState(pending.execution_state) + if pending is not None + else None + ), + message="整理任务状态已变化，未放弃失败任务", + ) + + # PostgreSQL/启用外键的 SQLite 会级联删除；显式清理兼容独立测试库。 + TransferExecutionStepOper(session).stage_delete_task(task_id=task_id) + detached = history_oper.stage_detach_failed_transfer_task( + history_id=history_id, + task_id=task_id, + settlement_revision=settlement_revision, + ) + if detached != 1: + raise TransferExecutionConflictError( + "失败整理任务删除后无法解除历史回执映射" + ) + transaction.commit() + return TransferFailureDiscardResult( + discarded=True, + state=TransferExecutionState.FAILED, + message="已放弃失败整理任务", + ) + except Exception: + self._rollback(transaction) + raise + def resolve_manual_review( self, *, diff --git a/app/db/models/transferhistory.py b/app/db/models/transferhistory.py index dc40003a6..05566baf2 100644 --- a/app/db/models/transferhistory.py +++ b/app/db/models/transferhistory.py @@ -627,6 +627,41 @@ class TransferHistory(Base): update(cls).where(cls.id == historyid).values(download_hash=download_hash) ) + @classmethod + def detach_failed_transfer_task( + cls, + db: Session, + *, + history_id: int, + task_id: str, + settlement_revision: int, + ) -> int: + """ + 解除指定失败回执与 durable 任务的精确映射。 + + :param db: 数据库会话 + :param history_id: 失败终态历史标识 + :param task_id: 稳定任务标识 + :param settlement_revision: 失败回执结算版本 + :return: 更新的历史数 + """ + if history_id <= 0 or not task_id or settlement_revision <= 0: + return 0 + return execute_dml( + db, + update(cls) + .where( + cls.id == history_id, + cls.transfer_task_id == task_id, + cls.transfer_settlement_revision == settlement_revision, + ) + .values( + transfer_task_id=None, + transfer_settlement_revision=None, + ), + execution_options={"synchronize_session": False}, + ) + @classmethod def replace_by_src(cls, db: Session, **kwargs) -> "TransferHistory": """ diff --git a/app/db/models/transferpending.py b/app/db/models/transferpending.py index 8e3430b9a..8c4df6d93 100644 --- a/app/db/models/transferpending.py +++ b/app/db/models/transferpending.py @@ -27,7 +27,12 @@ from sqlalchemy.sql.elements import ColumnElement from app.db.base import Base, execute_dml, get_id_column _TRANSFER_EXECUTION_STEP = table("transferexecutionstep", column("task_id")) -_TRANSFER_HISTORY = table("transferhistory", column("transfer_task_id")) +_TRANSFER_HISTORY = table( + "transferhistory", + column("id"), + column("transfer_task_id"), + column("transfer_settlement_revision"), +) class TransferPending(Base): @@ -737,6 +742,49 @@ class TransferPending(Base): execution_options={"synchronize_session": False}, ) + @classmethod + def delete_terminal_failure( + cls, + db: Session, + *, + task_id: str, + history_id: int, + settlement_revision: int, + ) -> int: + """ + 仅删除仍与指定失败历史精确绑定的无租约终态任务。 + + 历史存在性子查询把用户看到的失败回执纳入同一个 CAS，避免陈旧页面误删 + 已重试、已人工判定或已产生新结算版本的任务。 + :param db: 数据库会话 + :param task_id: 稳定任务标识 + :param history_id: 失败终态历史标识 + :param settlement_revision: 失败回执结算版本 + :return: 删除的任务数，1 表示成功取得放弃权 + """ + if not task_id or history_id <= 0 or settlement_revision <= 0: + return 0 + return execute_dml( + db, + delete(cls).where( + cls.task_id == task_id, + cls.execution_state == "failed", + cls.terminal_history_id == history_id, + cls.settlement_revision == settlement_revision, + cls.lease_owner.is_(None), + cls.lease_token.is_(None), + exists( + select(_TRANSFER_HISTORY.c.id).where( + _TRANSFER_HISTORY.c.id == history_id, + _TRANSFER_HISTORY.c.transfer_task_id == cls.task_id, + _TRANSFER_HISTORY.c.transfer_settlement_revision + == cls.settlement_revision, + ) + ), + ), + execution_options={"synchronize_session": False}, + ) + @classmethod def delete_terminal_success( cls, diff --git a/app/db/oper/transferhistory.py b/app/db/oper/transferhistory.py index 0530d7fc2..698fde61b 100644 --- a/app/db/oper/transferhistory.py +++ b/app/db/oper/transferhistory.py @@ -509,6 +509,23 @@ class TransferHistoryOper(DbOper): self._db.flush() return history + def stage_detach_failed_transfer_task( + self, + *, + history_id: int, + task_id: str, + settlement_revision: int, + ) -> int: + """在调用方事务内解除指定失败历史与 durable 任务的精确映射。""" + if not isinstance(self._db, Session): + raise RuntimeError("整理失败回执解绑需要调用方提供同步 Session") + return TransferHistory.detach_failed_transfer_task( + self._db, + history_id=history_id, + task_id=task_id, + settlement_revision=settlement_revision, + ) + def update_download_hash(self, historyid, download_hash): """ 补充转移记录download_hash diff --git a/app/db/oper/transferpending.py b/app/db/oper/transferpending.py index e594d8c17..d7e291447 100644 --- a/app/db/oper/transferpending.py +++ b/app/db/oper/transferpending.py @@ -521,6 +521,23 @@ class TransferPendingOper(DbOper): ) ) + def stage_delete_terminal_failure( + self, + *, + task_id: str, + history_id: int, + settlement_revision: int, + ) -> int: + """暂存与指定失败回执精确匹配的无租约 pending 删除。""" + return self._execute_sync_write( + lambda session: TransferPending.delete_terminal_failure( + session, + task_id=task_id, + history_id=history_id, + settlement_revision=settlement_revision, + ) + ) + def stage_delete_terminal_success( self, *, diff --git a/app/modules/filemanager/transhandler.py b/app/modules/filemanager/transhandler.py index 762c46503..478271b80 100644 --- a/app/modules/filemanager/transhandler.py +++ b/app/modules/filemanager/transhandler.py @@ -20,6 +20,7 @@ from app.application.transfer.workflow import ( TransferPlanCheckpoint, TransferPlanItem, TransferPlanningInput, + TransferPlanningRejectedError, ) from app.domain.context import MediaInfo, MusicInfo from app.domain.meta.metabase import MetaBase @@ -380,7 +381,7 @@ class TransHandler: preview=preview, skip_reason="未识别到文件集数，识别为特典/附加视频文件", ) - raise ValueError("未识别到文件集数") + raise TransferPlanningRejectedError("未识别到文件集数") planning_meta.end_season = None if planning_meta.total_season: planning_meta.total_season = 1 diff --git a/tests/test_chain_durable_events.py b/tests/test_chain_durable_events.py index f61aef49d..0ec0e6aec 100644 --- a/tests/test_chain_durable_events.py +++ b/tests/test_chain_durable_events.py @@ -1059,6 +1059,7 @@ def test_receipt_replay_survives_real_history_command_cleanup(cleanup): command = TransferHistoryMutationCommand( repository=TransferHistoryOper(session), download_repository=Mock(), + transfer_execution_repository=Mock(), unit_of_work=SqlAlchemyUnitOfWork(session), file_item_factory=Mock(), delete_media_file=Mock(return_value=True), diff --git a/tests/test_history_mutation_command.py b/tests/test_history_mutation_command.py index 3e78385a3..09921ed2e 100644 --- a/tests/test_history_mutation_command.py +++ b/tests/test_history_mutation_command.py @@ -7,6 +7,10 @@ from app.application.history import ( DownloadHistoryMutationCommand, TransferHistoryMutationCommand, ) +from app.application.transfer.execution import ( + TransferExecutionState, + TransferFailureDiscardResult, +) def _transfer_command(*, history=None, delete_result=True, commit_error=None): @@ -14,11 +18,13 @@ def _transfer_command(*, history=None, delete_result=True, commit_error=None): repository = Mock() repository.get.return_value = history download_repository = Mock() + transfer_execution_repository = Mock() unit_of_work = Mock() unit_of_work.commit.side_effect = commit_error dependencies = { "repository": repository, "download_repository": download_repository, + "transfer_execution_repository": transfer_execution_repository, "unit_of_work": unit_of_work, "file_item_factory": lambda payload: SimpleNamespace(**payload), "delete_media_file": Mock(return_value=delete_result), @@ -38,6 +44,7 @@ def _history(): src_fileitem={"path": "/downloads/demo.mkv"}, dest_fileitem={"path": "/media/demo.mkv"}, transfer_task_id=None, + transfer_settlement_revision=None, ) @@ -129,22 +136,57 @@ def test_transfer_truncate_uses_single_transaction(): dependencies["unit_of_work"].commit.assert_called_once_with() -def test_transfer_delete_rejects_durable_receipt_before_file_side_effects(): - """durable 回执不能被历史 API 连同源或目标文件一起删除。""" +def test_transfer_delete_rejects_nonfailed_durable_receipt_before_file_side_effects(): + """非 FAILED durable 回执不能被历史 API 连同文件一起删除。""" history = _history() history.transfer_task_id = "task-durable" + history.transfer_settlement_revision = 2 command, dependencies = _transfer_command(history=history) + dependencies["transfer_execution_repository"].discard_failed.return_value = ( + TransferFailureDiscardResult( + discarded=False, + state=TransferExecutionState.MANUAL_REVIEW, + message="人工复核任务必须先完成专门判定", + ) + ) result = command.delete(7, delete_source=True, delete_destination=True) assert result.success is False - assert result.message == "持久整理失败记录不可删除，请使用重试或人工复核入口" + assert result.message == "人工复核任务必须先完成专门判定" assert result.history == "retained" dependencies["delete_media_file"].assert_not_called() dependencies["repository"].stage_delete.assert_not_called() dependencies["unit_of_work"].commit.assert_not_called() +def test_transfer_delete_discards_failed_durable_receipt_before_cleanup(): + """FAILED durable 回执应先放弃任务，再按普通历史执行文件和记录删除。""" + history = _history() + history.transfer_task_id = "task-durable" + history.transfer_settlement_revision = 3 + command, dependencies = _transfer_command(history=history) + dependencies["transfer_execution_repository"].discard_failed.return_value = ( + TransferFailureDiscardResult( + discarded=True, + state=TransferExecutionState.FAILED, + message="已放弃失败整理任务", + ) + ) + + result = command.delete(7, delete_destination=True) + + assert result.success is True + dependencies["transfer_execution_repository"].discard_failed.assert_called_once_with( + task_id="task-durable", + history_id=7, + settlement_revision=3, + ) + dependencies["delete_media_file"].assert_called_once() + dependencies["repository"].stage_delete.assert_called_once_with(7) + dependencies["unit_of_work"].commit.assert_called_once_with() + + def test_transfer_destination_delete_failure_keeps_history_and_reports_step(): """目标文件删除失败时必须保留历史，并向调用方报告失败步骤。""" command, dependencies = _transfer_command( @@ -169,6 +211,7 @@ def test_transfer_delete_commits_completed_source_when_destination_fails(): dependencies = { "repository": repository, "download_repository": Mock(), + "transfer_execution_repository": Mock(), "unit_of_work": Mock(), "file_item_factory": lambda payload: SimpleNamespace(**payload), "file_exists": Mock(return_value=True), @@ -201,6 +244,9 @@ def test_transfer_delete_treats_missing_requested_file_as_completed(): command = TransferHistoryMutationCommand( repository=dependencies["repository"], download_repository=dependencies["download_repository"], + transfer_execution_repository=dependencies[ + "transfer_execution_repository" + ], unit_of_work=dependencies["unit_of_work"], file_item_factory=dependencies["file_item_factory"], file_exists=dependencies["file_exists"], diff --git a/tests/test_manual_transfer_history.py b/tests/test_manual_transfer_history.py index 7a86bca20..913ecad60 100644 --- a/tests/test_manual_transfer_history.py +++ b/tests/test_manual_transfer_history.py @@ -622,6 +622,61 @@ def test_manual_reorganize_removes_success_history_and_old_target(monkeypatch): assert planned == [fileitem.path] +def test_explicit_durable_reorganize_discards_old_task_and_replans(monkeypatch): + """显式重新整理 durable 失败历史时不得登记原计划重试，应清理后重新规划。""" + chain = make_transfer_chain() + fileitem = make_fileitem("/downloads/Test.Show.S01E01.mkv") + history = SimpleNamespace( + id=12, + transfer_task_id="transfer-task-12", + transfer_settlement_revision=1, + status=False, + mode="copy", + dest_fileitem=None, + download_hash=None, + downloader=None, + src=fileitem.path, + src_storage=fileitem.storage, + ) + planned = [] + deleted = [] + _patch_transfer_planning( + monkeypatch, + chain, + fileitem, + history, + planned, + deleted, + ) + monkeypatch.setattr( + chain, + "_request_durable_transfer_retry", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("显式重新整理不得登记旧 durable 计划重试") + ), + ) + monkeypatch.setattr( + chain, + "_delete_manual_transfer_history", + lambda history, transfer_history_oper: deleted.append( + ("history", history.id) + ) or (True, ""), + ) + + state, message = TransferChain.do_transfer( + chain, + fileitem=fileitem, + background=False, + manual=True, + reorganize=True, + ) + + assert state is True + assert message == "" + assert deleted == [("history", 12)] + assert planned == [fileitem.path] + + def test_manual_reorganize_keeps_successful_move_target_as_source(monkeypatch): """成功移动后的目标是当前重整源，只能删历史记录，不能先删除文件。""" chain = make_transfer_chain() diff --git a/tests/test_transfer_durable_retry_owner.py b/tests/test_transfer_durable_retry_owner.py index b6c7e7489..1ec8119ae 100644 --- a/tests/test_transfer_durable_retry_owner.py +++ b/tests/test_transfer_durable_retry_owner.py @@ -4,6 +4,7 @@ from types import SimpleNamespace from app.application.transfer.execution import ( TransferExecutionState, + TransferFailureDiscardResult, TransferRetryRequestResult, ) from app.chain.transfer import TransferChain @@ -31,6 +32,26 @@ class _RetryCommand: return self.result +class _DiscardCommand: + """记录显式重新整理提交的失败任务放弃请求。""" + + calls: list[tuple[object, dict]] = [] + result = TransferFailureDiscardResult( + discarded=True, + state=TransferExecutionState.FAILED, + message="已放弃失败整理任务", + ) + + def __init__(self, repository: object) -> None: + """保存测试仓储实例。""" + self._repository = repository + + def discard_failed(self, **kwargs) -> TransferFailureDiscardResult: + """记录放弃请求并返回用例指定结果。""" + self.calls.append((self._repository, kwargs)) + return self.result + + def _install_retry_port(monkeypatch) -> object: """构造不会接触数据库的 execution 端口并安装命令替身。""" repository = object() @@ -48,6 +69,22 @@ def _install_retry_port(monkeypatch) -> object: return repository +def _install_discard_port(monkeypatch) -> object: + """构造不会接触数据库的 execution 端口并安装放弃命令替身。""" + repository = object() + _DiscardCommand.calls = [] + _DiscardCommand.result = TransferFailureDiscardResult( + discarded=True, + state=TransferExecutionState.FAILED, + message="已放弃失败整理任务", + ) + monkeypatch.setattr( + "app.chain.transfer.records.TransferExecutionCommand", + _DiscardCommand, + ) + return repository + + def test_durable_history_redo_only_requests_persistent_retry(monkeypatch): """durable 重做不得检查源文件、重新识别或重新准入执行。""" repository = _install_retry_port(monkeypatch) @@ -91,12 +128,13 @@ def test_durable_history_redo_only_requests_persistent_retry(monkeypatch): ] -def test_durable_manual_cleanup_keeps_target_history_and_failure_budget(monkeypatch): - """手动重整命中 durable 历史时不得先删目标、历史或失败计数。""" - repository = _install_retry_port(monkeypatch) +def test_durable_manual_cleanup_discards_task_and_removes_old_state(monkeypatch): + """显式重整命中 FAILED durable 历史时应放弃任务并清理旧状态。""" + repository = _install_discard_port(monkeypatch) history = SimpleNamespace( id=82, transfer_task_id="transfer-task-82", + transfer_settlement_revision=4, status=False, mode="copy", src="/downloads/source.mkv", @@ -107,22 +145,23 @@ def test_durable_manual_cleanup_keeps_target_history_and_failure_budget(monkeypa "type": "file", }, ) - history_port = SimpleNamespace( - delete=lambda _history_id: (_ for _ in ()).throw( - AssertionError("durable 历史不得删除") - ) - ) + deleted = [] + cleared = [] + history_port = SimpleNamespace(delete=lambda history_id: deleted.append( + ("history", history_id) + )) monkeypatch.setattr( "app.chain.transfer.records.StorageChain", - lambda: (_ for _ in ()).throw( - AssertionError("durable 目标不得删除") + lambda: SimpleNamespace( + exists=lambda _fileitem: True, + delete_media_file=lambda fileitem: deleted.append( + ("target", fileitem.path) + ) or True, ), ) monkeypatch.setattr( "app.chain.transfer.records.clear_transfer_failures", - lambda *_args: (_ for _ in ()).throw( - AssertionError("durable 失败计数不得清零") - ), + lambda *args: cleared.append(args), ) chain = object.__new__(TransferChain) chain.transfer_execution_repository = repository @@ -132,9 +171,72 @@ def test_durable_manual_cleanup_keeps_target_history_and_failure_budget(monkeypa transfer_history_oper=history_port, ) + assert state is True + assert message == "" + assert _DiscardCommand.calls == [ + ( + repository, + { + "task_id": "transfer-task-82", + "history_id": 82, + "settlement_revision": 4, + }, + ) + ] + assert deleted == [ + ("target", "/library/source.mkv"), + ("history", 82), + ] + assert cleared == [("/downloads/source.mkv", "local")] + + +def test_durable_manual_cleanup_rejects_nonfailed_state(monkeypatch): + """非 FAILED durable 任务被拒绝后不得删除目标、历史或失败计数。""" + repository = _install_discard_port(monkeypatch) + _DiscardCommand.result = TransferFailureDiscardResult( + discarded=False, + state=TransferExecutionState.MANUAL_REVIEW, + message="人工复核任务必须先完成专门判定", + ) + history = SimpleNamespace( + id=85, + transfer_task_id="transfer-task-85", + transfer_settlement_revision=2, + status=False, + mode="copy", + src="/downloads/source.mkv", + src_storage="local", + dest_fileitem={ + "storage": "local", + "path": "/library/source.mkv", + "type": "file", + }, + ) + monkeypatch.setattr( + "app.chain.transfer.records.StorageChain", + lambda: (_ for _ in ()).throw(AssertionError("拒绝后不得删除目标")), + ) + monkeypatch.setattr( + "app.chain.transfer.records.clear_transfer_failures", + lambda *_args: (_ for _ in ()).throw( + AssertionError("拒绝后不得清理失败计数") + ), + ) + chain = object.__new__(TransferChain) + chain.transfer_execution_repository = repository + history_port = SimpleNamespace( + delete=lambda _history_id: (_ for _ in ()).throw( + AssertionError("拒绝后不得删除历史") + ) + ) + + state, message = chain._delete_manual_transfer_history( + history=history, + transfer_history_oper=history_port, + ) + assert state is False - assert message == "整理任务已登记重试" - assert _RetryCommand.calls[0][1]["requested_by"] == "manual_reorganize" + assert message == "人工复核任务必须先完成专门判定" def test_durable_ai_button_bypasses_agent_and_requests_scheduler(monkeypatch): diff --git a/tests/test_transfer_execution_persistence.py b/tests/test_transfer_execution_persistence.py index d3e3a0dca..4d7589269 100644 --- a/tests/test_transfer_execution_persistence.py +++ b/tests/test_transfer_execution_persistence.py @@ -134,6 +134,56 @@ def _seed_pending( session.commit() +def _seed_failed_receipt( + factory, + *, + task_id: str = "task-1", + execution_state: str = "failed", + settlement_revision: int = 1, +) -> int: + """写入带失败历史和步骤证据的无租约 durable 终态任务。""" + _seed_pending(factory, task_id=task_id) + with factory() as session: + pending = session.scalar( + select(TransferPending).where(TransferPending.task_id == task_id) + ) + assert pending is not None + history = TransferHistory( + transfer_task_id=task_id, + transfer_settlement_revision=settlement_revision, + src=f"/{task_id}.mkv", + src_storage="local", + status=False, + errmsg="terminal failure", + ) + session.add(history) + session.flush() + pending.execution_state = execution_state + pending.settlement_revision = settlement_revision + pending.terminal_history_id = history.id + pending.lease_owner = None + pending.lease_token = None + pending.lease_expires_at = None + session.add(TransferExecutionStep( + task_id=task_id, + operation_id=f"operation-{task_id}", + checkpoint_fingerprint=_plan_fingerprint(task_id), + ordinal=0, + phase="transfer", + kind="materialize_target", + state="failed", + attempt_token=None, + attempt_count=1, + intent_version=1, + intent_payload={"path": f"/{task_id}.mkv"}, + last_error="terminal failure", + prepared_at="2026-08-27 09:00:00", + updated_at="2026-08-27 09:30:00", + )) + session.commit() + return history.id + + def _repository(factory, token_values: list[str] | None = None): """构造固定时钟与可预测 attempt token 的执行命令。""" repository = TransactionalTransferExecutionRepository( @@ -833,6 +883,81 @@ def test_user_retry_is_single_generation_and_rejects_manual_review(execution_sto assert "人工" in rejected.message +def test_discard_failed_removes_execution_evidence_and_detaches_history( + execution_store, +) -> None: + """放弃匹配的 FAILED 任务应删除执行证据，并把历史恢复为普通记录。""" + history_id = _seed_failed_receipt(execution_store) + _, command = _repository(execution_store) + + result = command.discard_failed( + task_id="task-1", + history_id=history_id, + settlement_revision=1, + ) + + assert result.discarded is True + assert result.state is TransferExecutionState.FAILED + with execution_store() as session: + assert session.scalar(select(TransferPending)) is None + assert session.scalar(select(TransferExecutionStep)) is None + history = session.scalar(select(TransferHistory)) + assert history is not None + assert history.id == history_id + assert history.transfer_task_id is None + assert history.transfer_settlement_revision is None + + +@pytest.mark.parametrize("execution_state", ("retry_wait", "manual_review")) +def test_discard_failed_rejects_nonfailed_execution_state( + execution_store, + execution_state, +) -> None: + """等待重试和人工复核任务不得通过历史维护入口删除执行证据。""" + history_id = _seed_failed_receipt( + execution_store, + execution_state=execution_state, + ) + _, command = _repository(execution_store) + + result = command.discard_failed( + task_id="task-1", + history_id=history_id, + settlement_revision=1, + ) + + assert result.discarded is False + assert result.state is TransferExecutionState(execution_state) + with execution_store() as session: + assert session.scalar(select(TransferPending)) is not None + assert session.scalar(select(TransferExecutionStep)) is not None + history = session.scalar(select(TransferHistory)) + assert history is not None + assert history.transfer_task_id == "task-1" + + +def test_discard_failed_rejects_stale_settlement_revision(execution_store) -> None: + """陈旧页面携带的结算版本不能放弃已经变化的失败任务。""" + history_id = _seed_failed_receipt(execution_store, settlement_revision=3) + _, command = _repository(execution_store) + + result = command.discard_failed( + task_id="task-1", + history_id=history_id, + settlement_revision=2, + ) + + assert result.discarded is False + assert "已变化" in result.message + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + history = session.scalar(select(TransferHistory)) + assert pending is not None + assert pending.settlement_revision == 3 + assert history is not None + assert history.transfer_settlement_revision == 3 + + def test_manual_not_applied_decision_is_audited_and_schedules_same_step( execution_store, ) -> None: diff --git a/tests/test_transfer_failed_retry_budget.py b/tests/test_transfer_failed_retry_budget.py index da80ebaf6..1e0e4058b 100644 --- a/tests/test_transfer_failed_retry_budget.py +++ b/tests/test_transfer_failed_retry_budget.py @@ -10,7 +10,6 @@ from types import SimpleNamespace from unittest.mock import Mock from app import schemas -from app.runtime.config import settings from app.application.history import ( HistoryGateAction, TransferHistoryMutationCommand, @@ -19,6 +18,7 @@ from app.application.history import ( failed_retry_count, record_transfer_failure, ) +from app.runtime.config import settings def _reset_failed_retries(src_path, storage=None): @@ -115,6 +115,7 @@ def test_delete_transfer_history_endpoint_clears_retry_count(monkeypatch): command = TransferHistoryMutationCommand( repository=repository, download_repository=Mock(), + transfer_execution_repository=Mock(), unit_of_work=Mock(), file_item_factory=lambda payload: SimpleNamespace(**payload), delete_media_file=Mock(return_value=True), diff --git a/tests/test_transfer_planning_checkpoint.py b/tests/test_transfer_planning_checkpoint.py index 8f73259b8..bbba18f92 100644 --- a/tests/test_transfer_planning_checkpoint.py +++ b/tests/test_transfer_planning_checkpoint.py @@ -1344,6 +1344,33 @@ def test_checkpoint_commit_failure_blocks_executor_and_keeps_accepted(): assert "commit failed" in repository.record_planning_failure.call_args.kwargs["error"] +def test_host_planning_value_error_commits_rejection_instead_of_retrying(): + """确定性的宿主规划错误必须形成失败终态，不能留在 accepted 周期回放。""" + repository = Mock() + task = _task() + planning_input = _planning_input() + task.bind_admission_task_id("task-planning-rejection") + _bind_planning_input(task, planning_input) + repository.checkpoint_plan.side_effect = lambda **kwargs: _planned_admission( + task, + kwargs["checkpoint"], + ) + chain = _chain(repository=repository) + chain.plan_transfer.side_effect = transfer_application.TransferPlanningRejectedError( + "未识别到文件集数" + ) + + result = chain._plan_checkpoint_and_execute(task) + + assert result.success is False + assert result.message == "未识别到文件集数" + assert task.plan_checkpoint is not None + assert task.plan_checkpoint.rejection_error == "未识别到文件集数" + assert task.plan_checkpoint.items == () + assert task.execution_checkpoint is not None + repository.record_planning_failure.assert_not_called() + + def test_post_commit_crash_replays_frozen_plan_without_replanning(): """commit 后执行崩溃仍保留 planned checkpoint，重启后跳过 rename 规划。""" repository = Mock() diff --git a/tests/test_transfer_planning_compat.py b/tests/test_transfer_planning_compat.py index 50736fdb5..d37550fda 100644 --- a/tests/test_transfer_planning_compat.py +++ b/tests/test_transfer_planning_compat.py @@ -205,7 +205,7 @@ def test_chain_base_plan_transfer_keeps_internal_dto_type_only(): """规划入口运行时只做内部调度，不要求导入或重复导出 DTO。""" chain = object.__new__(ChainBase) checkpoint = Mock(name="checkpoint") - chain.run_module = Mock(return_value=checkpoint) + chain.run_module_strict = Mock(return_value=checkpoint) meta = Mock(name="meta") mediainfo = Mock(name="mediainfo") @@ -216,7 +216,7 @@ def test_chain_base_plan_transfer_keeps_internal_dto_type_only(): ) assert returned is checkpoint - chain.run_module.assert_called_once_with( + chain.run_module_strict.assert_called_once_with( "plan_transfer", fileitem=_fileitem(), meta=meta,