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(transfer): fence queued tasks from recovery

Browse Source
This commit is contained in:
jxxghp
2026-09-02 13:31:58 +08:00
parent d0eddb86eb
commit 0f1627876e
5 changed files with 120 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -971,6 +971,7 @@ class TransferQueueService:
*,
register_task: Callable[[TransferTask], bool],
admit_task: Callable[[TransferTask], TransferAdmission],
claim_task: Callable[[TransferTask, TransferAdmission], None],
enqueue: Callable[[TransferQueue], None],
before_enqueue: Callable[[TransferTask], None],
enqueue_failed: Callable[[TransferTask, Exception], None],
@@ -981,6 +982,7 @@ class TransferQueueService:
"""保存队列用例依赖，避免 Application 服务绑定具体线程队列实现。"""
self._register_task = register_task
self._admit_task = admit_task
self._claim_task = claim_task
self._enqueue = enqueue
self._before_enqueue = before_enqueue
self._enqueue_failed = enqueue_failed
@@ -989,12 +991,13 @@ class TransferQueueService:
self._expire_tasks = expire_tasks
def put(self, task: TransferTask, callback: TransferCallback) -> bool:
"""先持久化准入事实再入队；任何前置失败都撤销内存作业视图。"""
"""先持久化并 claim 准入事实再入队；前置失败撤销内存作业视图。"""
if not task or not self._register_task(task):
return False
try:
admission = self._admit_task(task)
task.bind_admission_task_id(admission.task_id)
self._claim_task(task, admission)
except Exception:
self._remove_task(task.fileitem)
raise

36
app/chain/transfer/queue.py
View File

@@ -373,6 +373,7 @@ class TransferQueueOwner(_TransferOwnerBase):
return TransferQueueService(
register_task=self._TransferChain__put_to_jobview,
admit_task=self._TransferChain__admit_transfer,
claim_task=self._TransferChain__claim_admitted_task,
enqueue=self._queue.put,
before_enqueue=self._register_scrape_batch_task,
enqueue_failed=self._TransferChain__record_enqueue_failure,
@@ -615,7 +616,7 @@ class TransferQueueOwner(_TransferOwnerBase):
raise TransferLeaseLostError(f"整理任务租约已经失效：{task_id}")
def _TransferChain__claim_task_for_execution(self, task: TransferTask) -> None:
"""让普通队列任务在业务执行前取得唯一租约，恢复任务复用既有 token。"""
"""校验队列任务既有租约，并为兼容旧队列项补取唯一租约。"""
if task.preview:
return
self._TransferChain__ensure_lease_runtime_state()
@@ -643,6 +644,27 @@ class TransferQueueOwner(_TransferOwnerBase):
self._TransferChain__release_admission_claim(claimed, error=str(err))
raise
def _TransferChain__claim_admitted_task(
self,
task: TransferTask,
admission: TransferAdmission,
) -> None:
"""普通任务进入内存队列前取得租约，避免恢复调度抢占等待项。"""
claimed = self._transfer_admissions.claim_task(
task_id=admission.task_id,
owner_id=self._worker_owner_id,
lease_seconds=self._WORKER_LEASE_SECONDS,
)
if claimed is None:
raise TransferLeaseLostError(
f"整理任务已由其他 worker claim：{admission.task_id}"
)
try:
self._TransferChain__bind_claimed_admission(task, claimed)
except Exception as err:
self._TransferChain__release_admission_claim(claimed, error=str(err))
raise
def _TransferChain__release_task_claim(
self,
task: TransferTask,
@@ -728,6 +750,12 @@ class TransferQueueOwner(_TransferOwnerBase):
"""把已 claim 的恢复任务送入普通队列，禁止再次准入或二次 claim。"""
self._TransferChain__assert_owned_lease(task)
if not self._TransferChain__put_to_jobview(task):
logger.warning(
"恢复任务被内存作业视图判定为重复，未进入队列：task_id=%s, source=%s:%s",
task.admission_task_id,
task.fileitem.storage,
task.fileitem.path,
)
return False
try:
self._register_scrape_batch_task(task)
@@ -917,6 +945,12 @@ class TransferQueueOwner(_TransferOwnerBase):
"待整理文件回放收到关闭请求，已送入 %s 个文件，其余登记保持待处理",
replayed,
)
elif replayed == 0:
logger.warning(
"待整理文件回放未产生可执行队列任务：本次 claim %s 个；"
"逐任务原因已写入前序日志和 transferpending.last_error",
len(pendings),
)
else:
logger.info(f"✓ 待整理文件回放完成，{replayed} 个文件已重新送入整理链")

7
tests/test_transfer_pending_replay.py
View File

@@ -495,6 +495,8 @@ def test_replay_releases_claim_when_jobview_rejects_recovered_task(
admissions.claim_recoverable.return_value = [admission]
admissions.release_claim.return_value = True
chain = _build_chain(admissions)
warning = MagicMock()
monkeypatch.setattr("app.chain.transfer.queue.logger.warning", warning)
monkeypatch.setattr(
chain,
"_TransferChain__queue_planned_replay",
@@ -508,6 +510,11 @@ def test_replay_releases_claim_when_jobview_rejects_recovered_task(
lease_token="lease-task-1",
error="恢复任务未进入内存队列",
)
warning.assert_called_once_with(
"待整理文件回放未产生可执行队列任务：本次 claim %s 个；"
"逐任务原因已写入前序日志和 transferpending.last_error",
1,
)
assert chain._owned_leases == {}

69
tests/test_transfer_queue_service.py
View File

@@ -1,3 +1,4 @@
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock, patch
@@ -9,6 +10,7 @@ from app.application.transfer.workflow import (
TransferAdmission,
TransferPlanningInput,
TransferQueueService,
TransferTask,
)
from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository
from app.db.models.transferhistory import TransferHistory
@@ -44,6 +46,7 @@ def _service(**overrides):
updated_at="2026-08-27 10:00:00",
planning_input=_planning_input(),
)),
"claim_task": Mock(),
"enqueue": Mock(),
"before_enqueue": Mock(),
"enqueue_failed": Mock(),
@@ -56,7 +59,7 @@ def _service(**overrides):
def test_transfer_queue_service_put_preserves_registration_order():
"""入队必须先登记视图和 durable admission，再登记批次并写队列。"""
"""入队必须先登记视图、准入并取得租约，再登记批次并写队列。"""
calls = []
service, _ = _service(
register_task=lambda _task: calls.append("register") or True,
@@ -69,13 +72,14 @@ def test_transfer_queue_service_put_preserves_registration_order():
updated_at="2026-08-27 10:00:00",
planning_input=_planning_input(),
),
claim_task=lambda _task, _admission: calls.append("claim"),
before_enqueue=lambda _task: calls.append("batch"),
enqueue=lambda _item: calls.append("queue"),
)
task = make_task(1)
assert service.put(task, Mock()) is True
assert calls == ["register", "admit", "batch", "queue"]
assert calls == ["register", "admit", "claim", "batch", "queue"]
assert task.admission_task_id == "task-1"
@@ -87,6 +91,7 @@ def test_transfer_queue_service_rejects_duplicate_without_side_effects():
dependencies["before_enqueue"].assert_not_called()
dependencies["enqueue"].assert_not_called()
dependencies["admit_task"].assert_not_called()
dependencies["claim_task"].assert_not_called()
def test_transfer_queue_service_blocks_enqueue_when_admission_fails():
@@ -100,10 +105,70 @@ def test_transfer_queue_service_blocks_enqueue_when_admission_fails():
service.put(task, Mock())
dependencies["remove_task"].assert_called_once_with(task.fileitem)
dependencies["claim_task"].assert_not_called()
dependencies["before_enqueue"].assert_not_called()
dependencies["enqueue"].assert_not_called()
def test_transfer_queue_service_blocks_enqueue_when_claim_fails() -> None:
"""准入后的租约竞争失败必须撤销作业视图，不能留下等待占位。"""
service, dependencies = _service(
claim_task=Mock(side_effect=RuntimeError("already claimed")),
)
task = make_task(1)
with pytest.raises(RuntimeError, match="already claimed"):
service.put(task, Mock())
dependencies["remove_task"].assert_called_once_with(task.fileitem)
dependencies["before_enqueue"].assert_not_called()
dependencies["enqueue"].assert_not_called()
def test_transfer_queue_service_claim_fences_recovery_owner(tmp_path: Path) -> None:
"""普通任务等待内存队列期间必须持有租约，恢复 owner 不得抢占。"""
engine = create_engine(f"sqlite:///{tmp_path / 'queue-claim.db'}")
TransferHistory.__table__.create(engine)
TransferPending.__table__.create(engine)
factory = sessionmaker(bind=engine)
repository = TransactionalTransferAdmissionRepository(factory)
task = make_task(1)
task.bind_planning_input(_planning_input(task.fileitem.path))
def claim_task(item: TransferTask, admission: TransferAdmission) -> None:
"""使用真实仓储 claim 并把租约绑定到内存任务。"""
claimed = repository.claim_task(
task_id=admission.task_id,
owner_id="queue-worker",
lease_seconds=120,
)
assert claimed is not None
assert claimed.lease_owner is not None
assert claimed.lease_token is not None
item.bind_execution_lease(
owner_id=claimed.lease_owner,
lease_token=claimed.lease_token,
)
service, _ = _service(
admit_task=lambda item: repository.admit(
storage=item.fileitem.storage,
src_path=item.fileitem.path,
planning_input=item.planning_input,
),
claim_task=claim_task,
)
assert service.put(task, Mock()) is True
assert task.lease_token is not None
assert repository.claim_recoverable(
owner_id="recovery-worker",
limit=100,
lease_seconds=120,
) == []
engine.dispose()
def test_transfer_queue_service_keeps_admission_when_enqueue_fails():
"""内存入队失败必须记录原因并清理视图，durable admission 由仓储保留。"""
error = RuntimeError("queue closed")

12
tests/test_transfer_worker_lifecycle.py
View File

@@ -490,6 +490,12 @@ def test_durable_task_identity_flows_to_unsettled_terminal_claim_release(monkeyp
monkeypatch.setattr(global_vars, "STOP_EVENT", threading.Event())
assert chain.put_to_queue(task) is True
assert task.lease_token == "lease-durable-task-id"
admissions.claim_task.assert_called_once_with(
task_id="durable-task-id",
owner_id="worker-owner",
lease_seconds=120,
)
stop_event = threading.Event()
worker = threading.Thread(
target=chain._TransferChain__start_transfer,
@@ -503,11 +509,7 @@ def test_durable_task_identity_flows_to_unsettled_terminal_claim_release(monkeyp
assert worker.is_alive() is False
assert task.admission_task_id == "durable-task-id"
admissions.claim_task.assert_called_once_with(
task_id="durable-task-id",
owner_id="worker-owner",
lease_seconds=120,
)
admissions.claim_task.assert_called_once()
admissions.release_claim.assert_called_once_with(
task_id="durable-task-id",
lease_token="lease-durable-task-id",