From 0f1627876e1ac22b522cb44e88cadcb7dc3535bb Mon Sep 17 00:00:00 2001 From: jxxghp Date: Wed, 2 Sep 2026 13:31:58 +0800 Subject: [PATCH] fix(transfer): fence queued tasks from recovery --- app/application/transfer/workflow.py | 5 +- app/chain/transfer/queue.py | 36 ++++++++++++- tests/test_transfer_pending_replay.py | 7 +++ tests/test_transfer_queue_service.py | 69 ++++++++++++++++++++++++- tests/test_transfer_worker_lifecycle.py | 12 +++-- 5 files changed, 120 insertions(+), 9 deletions(-) diff --git a/app/application/transfer/workflow.py b/app/application/transfer/workflow.py index 9f0c14010..eede0013e 100644 --- a/app/application/transfer/workflow.py +++ b/app/application/transfer/workflow.py @@ -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 diff --git a/app/chain/transfer/queue.py b/app/chain/transfer/queue.py index 2c3facc9b..d65334c65 100644 --- a/app/chain/transfer/queue.py +++ b/app/chain/transfer/queue.py @@ -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} 个文件已重新送入整理链") diff --git a/tests/test_transfer_pending_replay.py b/tests/test_transfer_pending_replay.py index 1341d04b7..46b2662cb 100644 --- a/tests/test_transfer_pending_replay.py +++ b/tests/test_transfer_pending_replay.py @@ -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 == {} diff --git a/tests/test_transfer_queue_service.py b/tests/test_transfer_queue_service.py index 247d746ee..662959493 100644 --- a/tests/test_transfer_queue_service.py +++ b/tests/test_transfer_queue_service.py @@ -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") diff --git a/tests/test_transfer_worker_lifecycle.py b/tests/test_transfer_worker_lifecycle.py index 4df3169b7..fc903fd8c 100644 --- a/tests/test_transfer_worker_lifecycle.py +++ b/tests/test_transfer_worker_lifecycle.py @@ -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",