nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-02 13:28:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: isolate unhandled durable transfer failures

Browse Source
This commit is contained in:
jxxghp
2026-10-10 12:17:07 +08:00
parent 736b6686c3
commit bb29451222
10 changed files with 310 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
app/application/transfer/execution.py
View File

@@ -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,
*,

17
app/chain/transfer/execution.py
View File

@@ -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} 已转入人工复核，未自动重放；"

63
app/db/adapters/transfer/execution.py
View File

@@ -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(
"这条整理步骤无法确认是否已经执行，请人工确认文件状态后再继续；"

31
app/db/models/transferpending.py
View File

@@ -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,

12
app/db/oper/transferpending.py
View File

@@ -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],

5
docs/mcp-api.md
View File

@@ -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 的存储访问路径是两个地址空间。下载器配置的

8
skills/moviepilot-api/SKILL.md
View File

@@ -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

2
tests/test_builtin_skill_boundaries.py
View File

@@ -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",

134
tests/test_transfer_execution_persistence.py
View File

@@ -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(

39
tests/test_transfer_worker_lifecycle.py
View File

@@ -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()