From bb2945122280701f81d4a4252ce4b2e87d556c45 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 10 Oct 2026 12:17:07 +0800 Subject: [PATCH] fix: isolate unhandled durable transfer failures --- app/application/transfer/execution.py | 5 + app/chain/transfer/execution.py | 17 ++- app/db/adapters/transfer/execution.py | 63 ++++++++- app/db/models/transferpending.py | 31 +++++ app/db/oper/transferpending.py | 12 ++ docs/mcp-api.md | 5 + skills/moviepilot-api/SKILL.md | 8 +- tests/test_builtin_skill_boundaries.py | 2 +- tests/test_transfer_execution_persistence.py | 134 +++++++++++++++++++ tests/test_transfer_worker_lifecycle.py | 39 ++++++ 10 files changed, 310 insertions(+), 6 deletions(-) diff --git a/app/application/transfer/execution.py b/app/application/transfer/execution.py index 729c9c2f3..e97effedd 100644 --- a/app/application/transfer/execution.py +++ b/app/application/transfer/execution.py @@ -508,6 +508,11 @@ class TransferExecutionRepository(Protocol): def get_snapshot(self, *, task_id: str) -> Optional[TransferExecutionSnapshot]: """读取任务执行快照。""" + def isolate_unhandled_failure( + self, *, task_id: str, lease_token: str, error: str, + ) -> bool: + """将步骤账本之外的异常计入失败次数并隔离复核，不推测外部操作结果。""" + async def async_get_snapshot( self, *, diff --git a/app/chain/transfer/execution.py b/app/chain/transfer/execution.py index bcfd7553c..9d84e8ce3 100644 --- a/app/chain/transfer/execution.py +++ b/app/chain/transfer/execution.py @@ -14,6 +14,8 @@ from app.application.history.retry import ( from app.application.transfer.execution import ( TransferExecutionCheckpoint, TransferExecutionCommand, + TransferExecutionConflictError, + TransferExecutionLeaseLostError, TransferExecutionRepository, TransferExecutionSnapshot, TransferOperationObservation, @@ -313,7 +315,20 @@ class TransferExecutionOwner(_TransferOwnerBase): *, preview: bool, ) -> Tuple[bool, str]: - """收口整理执行异常；人工复核只告警并保留 durable 状态。""" + """在插件可替换执行入口之外收口异常，防止未记账失败无限回放。""" + if (not preview and not task.terminal_settled and task.admission_task_id and task.lease_token + and not isinstance(error, (TransferAdmissionConflictError, TransferLeaseLostError, + TransferExecutionConflictError, TransferExecutionLeaseLostError, + _TransferRetryDeferred, _TransferRetryExhausted, + _TransferManualReviewRequired))): + try: + isolated = self.transfer_execution_repository.isolate_unhandled_failure( + task_id=task.admission_task_id, lease_token=task.lease_token, error=str(error), + ) + if isolated: + error = _TransferManualReviewRequired(str(error)) + except Exception as record_error: + logger.error(f"记录整理执行异常失败：{record_error}") if isinstance(error, _TransferManualReviewRequired): logger.warning( f"{task.fileitem.name} 已转入人工复核，未自动重放；" diff --git a/app/db/adapters/transfer/execution.py b/app/db/adapters/transfer/execution.py index 546c5e61e..1d608ea50 100644 --- a/app/db/adapters/transfer/execution.py +++ b/app/db/adapters/transfer/execution.py @@ -55,6 +55,10 @@ from app.db.oper.transferpending import TransferPendingOper from app.db.uow import SqlAlchemyUnitOfWork _LEGACY_REVIEW_STEP_KIND = "legacy_execution_review" +_UNHANDLED_REVIEW_STEP_KIND = "unhandled_execution_review" +_REVIEW_STEP_KINDS = {_LEGACY_REVIEW_STEP_KIND, _UNHANDLED_REVIEW_STEP_KIND} +# 与迁移保留的最大序号相邻，复核凭据不占用从零连续递增的真实执行序号。 +_UNHANDLED_REVIEW_STEP_ORDINAL = 2_147_483_646 def _local_now() -> datetime: @@ -100,8 +104,61 @@ class TransactionalTransferExecutionRepository: def _execution_steps( steps: list[TransferExecutionStepModel], ) -> list[TransferExecutionStepModel]: - """排除只用于迁移人工审计、不得参与真实执行的 synthetic 步骤。""" - return [step for step in steps if step.kind != _LEGACY_REVIEW_STEP_KIND] + """排除只用于人工审计、不得参与真实执行的 synthetic 步骤。""" + return [step for step in steps if step.kind not in _REVIEW_STEP_KINDS] + + def isolate_unhandled_failure( + self, *, task_id: str, lease_token: str, error: str, + ) -> bool: + """原子隔离未记账异常并保存可发现的复核凭据，保留所有真实步骤证据。""" + now_utc, updated_at = self._times() + with self._session_factory() as session: + transaction = SqlAlchemyUnitOfWork(session) + try: + pending_oper = TransferPendingOper(session) + updated = pending_oper.stage_isolate_unhandled_failure( + task_id=task_id, lease_token=lease_token, error=error, + now_utc=now_utc, updated_at=updated_at, + ) + if updated != 1: + transaction.commit() + return False + session.expire_all() + pending = pending_oper.get_by_task_id(task_id=task_id) + if pending is None: + raise TransferExecutionConflictError("异常隔离后未找到整理任务") + self._stage_unhandled_review(session, pending=pending, error=error, updated_at=updated_at) + transaction.commit() + return True + except Exception: + self._rollback(transaction) + raise + + @staticmethod + def _stage_unhandled_review( + session: Session, *, pending: TransferPending, error: str, updated_at: str, + ) -> None: + """创建或更新独立复核凭据；同一任务再次异常时复用凭据并保留复核审计。""" + oper = TransferExecutionStepOper(session) + step = next((item for item in oper.list_by_task_id(task_id=pending.task_id) + if item.kind == _UNHANDLED_REVIEW_STEP_KIND), None) + if step is None: + intent = TransferStepIntent.create( + task_id=pending.task_id, + checkpoint_fingerprint=pending.input_fingerprint, + ordinal=_UNHANDLED_REVIEW_STEP_ORDINAL, + phase="review", kind=_UNHANDLED_REVIEW_STEP_KIND, + payload={"source": {"storage": pending.storage, "path": pending.src_path}}, + ) + step = oper.stage_prepare( + task_id=pending.task_id, operation_id=intent.operation_id, + checkpoint_fingerprint=intent.checkpoint_fingerprint, + ordinal=intent.ordinal, phase=intent.phase, kind=intent.kind, + intent_version=intent.version, intent_payload=dict(intent.payload), now_time=updated_at, + ) + step.state = TransferStepState.MANUAL_REVIEW.value + step.last_error = error + step.updated_at = updated_at @staticmethod def _project_step(step: TransferExecutionStepModel) -> TransferExecutionStep: @@ -1533,7 +1590,7 @@ class TransactionalTransferExecutionRepository: decision is TransferManualReviewDecision.APPLIED and step is not None and step.task_id == task_id - and step.kind == _LEGACY_REVIEW_STEP_KIND + and step.kind in _REVIEW_STEP_KINDS ): raise TransferExecutionConflictError( "这条整理步骤无法确认是否已经执行，请人工确认文件状态后再继续；" diff --git a/app/db/models/transferpending.py b/app/db/models/transferpending.py index 56993634d..e1458df92 100644 --- a/app/db/models/transferpending.py +++ b/app/db/models/transferpending.py @@ -1127,6 +1127,37 @@ class TransferPending(Base): execution_options={"synchronize_session": False}, ) + @classmethod + def isolate_unhandled_failure( + cls, db: Session, *, task_id: str, lease_token: str, + error: str, now_utc: str, updated_at: str, + ) -> int: + """以当前租约收口未记账异常；已释放租约的重试和复核不得重复计数。""" + return execute_dml( + db, + update(cls).where( + cls.task_id == task_id, + cls.state.in_(("accepted", "planned", "provider_pending")), + cls.execution_state.in_(("not_started", "running", "retry_wait")), + cls.execution_payload.is_(None), + cls.terminal_history_id.is_(None), + cls.lease_token == lease_token, + cls.lease_expires_at.is_not(None), + cls.lease_expires_at > now_utc, + ).values( + execution_state="manual_review", + retry_count=cls.retry_count + 1, + retry_due_at=None, + lease_owner=None, + lease_token=None, + lease_expires_at=None, + heartbeat_at=None, + last_error=error, + updated_at=updated_at, + ), + execution_options={"synchronize_session": False}, + ) + @classmethod def record_planning_failure(cls, db: Session, *, task_id: str, lease_token: str, error: str, diff --git a/app/db/oper/transferpending.py b/app/db/oper/transferpending.py index b03e17c97..7802c97be 100644 --- a/app/db/oper/transferpending.py +++ b/app/db/oper/transferpending.py @@ -12,6 +12,18 @@ class TransferPendingOper(DbOper): 整理链的文件重新送回去，避免挂载故障重启后永久漏件。 """ + def stage_isolate_unhandled_failure( + self, *, task_id: str, lease_token: str, error: str, + now_utc: str, updated_at: str, + ) -> int: + """在调用方事务中暂存异常隔离，失败计数与租约释放使用同一 CAS。""" + return self._execute_sync_write( + lambda session: TransferPending.isolate_unhandled_failure( + session, task_id=task_id, lease_token=lease_token, + error=error, now_utc=now_utc, updated_at=updated_at, + ) + ) + def stage_admit(self, *, task_id: str, storage: str, src_path: str, state: str, now_time: str, input_version: int, planning_input: dict[str, Any], diff --git a/docs/mcp-api.md b/docs/mcp-api.md index c56d9e2a7..fef758174 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -435,6 +435,11 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`，不再返 | GET | `/api/v1/transfer/tasks/{task_id}/manual-review` | 管理员查询单个 durable 人工复核任务详情；仅可读取 `manual_review` 或已经人工判定的 `retry_wait` 任务，其余状态按不存在处理 | | POST | `/api/v1/transfer/tasks/{task_id}/manual-review` | 管理员判定处于 `manual_review` 的 durable 整理步骤；请求包含 `operation_id`、`decision=not_applied|applied`，`reason` 选填，省略或空白时记为空字符串，最多 2000 字符；`applied` 还必须提供 `result_payload`。`failed` 不属于公开决策，失败终态只能由持租约的 durable 结算写入；响应仅返回任务、操作、决策、后续状态和复核修订号 | +步骤账本之外的未处理异常（例如插件替换整理入口后在 `finally` 抛错）会记一次失败， +并立即进入 `manual_review` 停止自动回放。详情中的 `unhandled_execution_review` 是诊断凭据， +不代表某个文件操作已执行；应先修复或关闭异常插件并确认文件状态，必要时人工回滚， +再以 `not_applied` 交回调度器。该凭据不接受 `applied`，既有真实步骤仍按原证据恢复。 + #### SMB 下载器监控与服务端整理 下载器原生保存路径与 MoviePilot 的存储访问路径是两个地址空间。下载器配置的 diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index 1f2b0bff3..24624e656 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -1,6 +1,6 @@ --- name: moviepilot-api -version: 47 +version: 48 description: >- Use this skill for MoviePilot product operations such as media search, torrent search, downloads, subscriptions, library checks, sites, storage, workflows, @@ -67,6 +67,12 @@ allowed-api-operations: >- # MoviePilot API +Unhandled exceptions outside the durable transfer step ledger stop automatic +replay and create an `unhandled_execution_review` diagnostic. Repair or disable +the failing plugin and inspect the files before resolving it as `not_applied`; +roll back untracked effects first when needed. This diagnostic rejects `applied` +and does not replace the evidence for any recorded file operations. + For relevant prior conversations or previous tool evidence, use the host-provided `session_search` tool when available. Discover a session, then read its anchored messages; past results are context, not proof of current state or permission to diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index 27ebaaecd..3449e0569 100644 --- a/tests/test_builtin_skill_boundaries.py +++ b/tests/test_builtin_skill_boundaries.py @@ -117,7 +117,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None: "command-dispatch": "2", "database-operation": "9", "feedback-issue": "11", - "moviepilot-api": "47", + "moviepilot-api": "48", "moviepilot-update": "5", "organize-files": "5", "transfer-failed-retry": "5", diff --git a/tests/test_transfer_execution_persistence.py b/tests/test_transfer_execution_persistence.py index 07046614d..aae6a0c13 100644 --- a/tests/test_transfer_execution_persistence.py +++ b/tests/test_transfer_execution_persistence.py @@ -14,6 +14,7 @@ from app.application.transfer.execution import ( TransferExecutionConflictError, TransferExecutionState, TransferManualReviewDecision, + TransferManualReviewQuery, TransferOperationObservation, TransferOperationObservationState, TransferStepIntent, @@ -220,6 +221,139 @@ def _intent(*, task_id: str = "task-1", ordinal: int = 0) -> TransferStepIntent: ) +@pytest.mark.parametrize("state", ["accepted", "planned"]) +@pytest.mark.parametrize("retry_count", [0, 318]) +@pytest.mark.parametrize("execution_state", ["not_started", "retry_wait"]) +def test_unhandled_failure_is_counted_once_and_stops_recovery( + execution_store, state, retry_count, execution_state, +): + """补丁在步骤外抛异常时立即隔离，即使旧任务已反复领取也不再自动回放。""" + _seed_pending(execution_store, state=state, with_checkpoint=state == "planned") + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + pending.retry_count = retry_count + pending.attempt_count = 319 + pending.execution_state = execution_state + session.commit() + repository, _ = _repository(execution_store) + error = "TransferChain has no attribute _TransferChain__finish_scrape_batch_task" + assert repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error=error) + assert not repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error=error) + snapshot = repository.get_snapshot(task_id="task-1") + assert snapshot.state is TransferExecutionState.MANUAL_REVIEW + assert snapshot.retry_count == retry_count + 1 + assert snapshot.last_error == error + assert snapshot.steps == () + view = TransferManualReviewQuery(repository).get(task_id="task-1") + assert view.step.kind == "unhandled_execution_review" + assert view.step.error == error + assert TransferManualReviewQuery(repository).list().total == 1 + admissions = TransactionalTransferAdmissionRepository(execution_store) + assert admissions.claim_recoverable(owner_id="recovery", limit=10, lease_seconds=120) == [] + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + assert pending.attempt_count == 319 + assert pending.lease_owner is None + assert pending.lease_token is None + assert pending.lease_expires_at is None + assert pending.heartbeat_at is None + assert pending.retry_due_at is None + + +@pytest.mark.parametrize("state,token,expires", [ + ("retry_wait", None, "2099-01-01 00:00:00.000000"), + ("manual_review", "lease-1", "2099-01-01 00:00:00.000000"), + ("failed", "lease-1", "2099-01-01 00:00:00.000000"), + ("settling", "lease-1", "2099-01-01 00:00:00.000000"), + ("running", "new-owner-token", "2099-01-01 00:00:00.000000"), + ("running", "lease-1", "2000-01-01 00:00:00.000000"), +]) +def test_unhandled_failure_cannot_overwrite_existing_disposition_or_lease( + execution_store, state, token, expires, +): + """步骤已决定重试或终态、结算中及租约失效时，不重复计数或覆盖新 owner。""" + _seed_pending(execution_store) + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + pending.execution_state = state + pending.lease_token = token + pending.lease_expires_at = expires + pending.last_error = "previous evidence" + session.commit() + repository, _ = _repository(execution_store) + assert not repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer") + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + assert pending.execution_state == state + assert pending.retry_count == 0 + assert pending.last_error == "previous evidence" + assert pending.lease_token == token + assert session.scalar(select(TransferExecutionStep)) is None + + +def test_unhandled_failure_isolation_rolls_back_with_review_evidence(execution_store, monkeypatch): + """复核凭据写入失败时必须同时回滚状态、失败次数和租约释放。""" + _seed_pending(execution_store) + repository, _ = _repository(execution_store) + + def fail_review(*_args, **_kwargs): + """模拟复核凭据无法写入。""" + raise RuntimeError("review write failed") + + monkeypatch.setattr(repository, "_stage_unhandled_review", fail_review) + with pytest.raises(RuntimeError, match="review write failed"): + repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer") + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + assert pending.execution_state == "not_started" + assert pending.retry_count == 0 + assert pending.lease_token == "lease-1" + + +def test_unhandled_review_preserves_started_step_and_resumes_after_manual_confirmation(execution_store): + """复核未记账异常后，真实 STARTED 仍需严格探测，诊断步骤不参与执行或结算。""" + _seed_pending(execution_store) + repository, command = _repository(execution_store) + prepared = command.prepare(task_id="task-1", lease_token="lease-1", intent=_intent()) + started = command.begin(task_id="task-1", lease_token="lease-1", operation_id=prepared.operation_id) + assert repository.isolate_unhandled_failure(task_id="task-1", lease_token="lease-1", error="outer") + snapshot = repository.get_snapshot(task_id="task-1") + assert snapshot.steps == (started,) + review = repository.get_manual_review(task_id="task-1") + with pytest.raises(TransferExecutionConflictError, match="人工回滚"): + command.resolve_manual_review( + task_id="task-1", operation_id=review.step.operation_id, + decision=TransferManualReviewDecision.APPLIED, actor="admin", reason="checked", + result=TransferStepResult(payload={"success": True}), + ) + resolved = command.resolve_manual_review( + task_id="task-1", operation_id=review.step.operation_id, + decision=TransferManualReviewDecision.NOT_APPLIED, actor="admin", reason="plugin rolled back", + ) + assert resolved.state is TransferExecutionState.RETRY_WAIT + assert repository.get_manual_review(task_id="task-1").review_revision == 1 + admission = TransactionalTransferAdmissionRepository(execution_store).claim_task( + task_id="task-1", owner_id="recovery", lease_seconds=120, + ) + command.prepare(task_id="task-1", lease_token=admission.lease_token, intent=_intent()) + resumed = command.restart_after_not_applied( + task_id="task-1", lease_token=admission.lease_token, step=started, + evidence=TransferStepResult(payload={"observation": "not_applied"}), + ) + completed = command.complete( + task_id="task-1", lease_token=admission.lease_token, step=resumed, + result=TransferStepResult(payload={"success": True}), + ) + assert completed.state is TransferStepState.SUCCEEDED + assert repository.isolate_unhandled_failure(task_id="task-1", lease_token=admission.lease_token, error="again") + with execution_store() as session: + assert len(session.scalars(select(TransferExecutionStep)).all()) == 2 + snapshot = repository.get_snapshot(task_id="task-1") + assert len(snapshot.steps) == 1 + assert snapshot.steps[0].state is TransferStepState.SUCCEEDED + assert snapshot.retry_count == 2 + + def test_stable_operation_and_checkpoint_identities_are_canonical(): """字段顺序不能改变 operation ID 或执行 checkpoint 指纹。""" first = build_transfer_operation_id( diff --git a/tests/test_transfer_worker_lifecycle.py b/tests/test_transfer_worker_lifecycle.py index 0e7edd365..9d4f921f6 100644 --- a/tests/test_transfer_worker_lifecycle.py +++ b/tests/test_transfer_worker_lifecycle.py @@ -925,6 +925,45 @@ def test_worker_manual_review_logs_queue_guidance_without_traceback(monkeypatch) assert chain._queue.unfinished_tasks == 0 +def test_worker_isolates_exception_from_plugin_replacement_finally(monkeypatch) -> None: + """插件替换整个执行方法并在 finally 抛异常时，worker 外层仍必须隔离持久任务。""" + chain = _build_chain() + task = make_task(1) + task.bind_admission_task_id("plugin-task") + task.bind_execution_lease(owner_id="worker-owner", lease_token="plugin-lease") + chain._owned_leases = {"plugin-task": ("plugin-lease", time.monotonic() + 120)} + chain.jobview = MagicMock() + chain.jobview.pending_total.return_value = 1 + chain._progress = MagicMock() + chain._active_tasks = chain._processed_num = chain._fail_num = chain._total_num = 0 + chain._transfer_executions = MagicMock() + chain._transfer_executions.isolate_unhandled_failure.return_value = True + chain.transfer_execution_repository = chain._transfer_executions + chain._TransferChain__fail_transfer_task = MagicMock() + stop_event = threading.Event() + + def plugin_replacement(*, task, callback): + """重现第三方补丁在 finally 调用已不存在的私有方法。""" + del callback + try: + return True, "" + finally: + stop_event.set() + chain._TransferChain__finish_scrape_batch_task(task) + + chain._TransferChain__handle_transfer = plugin_replacement + chain._queue.put(TransferQueue(task=task)) + monkeypatch.setattr(global_vars, "STOP_EVENT", threading.Event()) + chain._TransferChain__start_transfer(stop_event) + + isolation = chain._transfer_executions.isolate_unhandled_failure + isolation.assert_called_once() + assert isolation.call_args.kwargs["task_id"] == "plugin-task" + assert "_TransferChain__finish_scrape_batch_task" in isolation.call_args.kwargs["error"] + assert chain._queue.unfinished_tasks == 0 + assert chain._owned_leases == {} + + def test_heartbeat_refreshes_current_token_and_forgets_lost_lease() -> None: """heartbeat 成功应刷新本地期限，CAS 拒绝后必须立即停止本地推进资格。""" chain = _build_chain()