From 435703573e8811d8a7e43a7356cce69f3e09a5ff Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 13 Oct 2026 07:13:42 +0800 Subject: [PATCH] fix(transfer): persist provider checkpoint planning time Write planned_at atomically for provider_pending checkpoints and refresh it when host planning takes over. Cover real admission-to-execution persistence and distinguish incomplete plan fields from inconsistent execution checkpoint fields. Refs #6973 --- app/db/adapters/transfer/execution.py | 4 +- app/db/models/transferpending.py | 12 +- tests/test_transfer_execution_migration.py | 2 +- tests/test_transfer_execution_persistence.py | 121 ++++++++++++++----- tests/test_transfer_lease_persistence.py | 6 +- 5 files changed, 104 insertions(+), 41 deletions(-) diff --git a/app/db/adapters/transfer/execution.py b/app/db/adapters/transfer/execution.py index 1d608ea50..d658eda54 100644 --- a/app/db/adapters/transfer/execution.py +++ b/app/db/adapters/transfer/execution.py @@ -214,7 +214,7 @@ class TransactionalTransferExecutionRepository: and pending.execution_fingerprint is not None ) if has_checkpoint_column != has_checkpoint_identity: - raise TransferExecutionConflictError("整理任务记录不完整，请重新识别文件后再整理") + raise TransferExecutionConflictError("整理任务执行检查点字段不一致，请重新识别文件后再整理") checkpoint = None if pending.execution_payload is not None: checkpoint = TransferExecutionCheckpoint.from_payload( @@ -322,7 +322,7 @@ class TransactionalTransferExecutionRepository: or pending.checkpoint_payload is None or pending.planned_at is None ): - raise TransferExecutionConflictError("整理任务记录不完整，请重新识别文件后再整理") + raise TransferExecutionConflictError("整理任务计划检查点字段不完整，请重新识别文件后再整理") try: checkpoint = TransferPlanCheckpoint.from_payload( pending.checkpoint_payload diff --git a/app/db/models/transferpending.py b/app/db/models/transferpending.py index e1458df92..677faede5 100644 --- a/app/db/models/transferpending.py +++ b/app/db/models/transferpending.py @@ -44,8 +44,9 @@ class TransferPending(Base): 蒸发。而已经稳定落地的文件不会再产生任何监控事件，也不会有新的补偿扫描起点 ——结果就是永久漏件，只能靠人工比对补整理。 - 准入时保存版本化规划输入和指纹；纯规划完成后以同一行原子保存完整有序计划并 - 推进到 planned。重启恢复可直接消费已规划路径，避免再次触发 rename 等插件事件。 + 准入时保存版本化规划输入和指纹；provider 调用快照与宿主文件计划分别原子推进到 + provider_pending、planned，并记录本阶段的计划冻结时间。重启恢复直接消费检查点， + 避免再次触发 rename 等插件事件。 所有执行期 mutation 都以稳定任务身份和租约 token 进行 CAS fencing。 """ @@ -1080,7 +1081,10 @@ class TransferPending(Base): lease_token: str, now_time: str, updated_at: str) -> int: """ - 以输入指纹为 CAS 条件原子保存计划并推进到已规划。 + 以输入指纹为 CAS 条件原子保存检查点并推进到本阶段可执行状态。 + + provider 调用快照也属于冻结计划，必须同事务写入 planned_at 才能准备执行； + 后续回退到宿主计划时，以新的检查点提交时间刷新该字段。 :param db: 数据库会话 :param task_id: 稳定任务标识 :param input_fingerprint: 规划输入规范 JSON 指纹 @@ -1110,7 +1114,7 @@ class TransferPending(Base): "last_error": None, "updated_at": updated_at, } - if target_state == "planned": + if target_state in ("planned", "provider_pending"): values["planned_at"] = updated_at return execute_dml( db, diff --git a/tests/test_transfer_execution_migration.py b/tests/test_transfer_execution_migration.py index 73b8b437b..bada0afdf 100644 --- a/tests/test_transfer_execution_migration.py +++ b/tests/test_transfer_execution_migration.py @@ -840,7 +840,7 @@ def test_migrated_legacy_reviews_are_discoverable_resolvable_and_retryable( assert pending.state in {"accepted", "planned"} with pytest.raises( TransferExecutionConflictError, - match="整理任务记录不完整，请重新识别文件后再整理|整理任务尚未准备完成，请稍后再试", + match="整理任务计划检查点字段不完整，请重新识别文件后再整理|整理任务尚未准备完成，请稍后再试", ): command.prepare( task_id=task_id, diff --git a/tests/test_transfer_execution_persistence.py b/tests/test_transfer_execution_persistence.py index aae6a0c13..a50d25159 100644 --- a/tests/test_transfer_execution_persistence.py +++ b/tests/test_transfer_execution_persistence.py @@ -418,6 +418,35 @@ def test_prepare_rejects_task_without_executable_plan( assert session.scalar(select(TransferExecutionStep)) is None +@pytest.mark.parametrize("field", ["checkpoint_version", "checkpoint_payload", "planned_at"]) +def test_prepare_rejects_incomplete_plan_fields(execution_store, field): + """缺少计划冻结字段仍须拒绝执行，并准确指出计划检查点缺失。""" + _seed_pending(execution_store) + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + setattr(pending, field, None) + session.commit() + _, command = _repository(execution_store) + with pytest.raises(TransferExecutionConflictError, match="计划检查点字段不完整"): + command.prepare(task_id="task-1", lease_token="lease-1", intent=_intent()) + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + assert pending.execution_state == "not_started" + assert session.scalar(select(TransferExecutionStep)) is None + + +def test_snapshot_reports_inconsistent_execution_checkpoint_fields(execution_store): + """执行检查点列不一致的提示应与计划冻结字段缺失区分。""" + _seed_pending(execution_store) + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + pending.execution_payload = {"outcome": "succeeded"} + session.commit() + repository, _ = _repository(execution_store) + with pytest.raises(TransferExecutionConflictError, match="执行检查点字段不一致"): + repository.get_snapshot(task_id="task-1") + + def test_prepare_rejects_noncanonical_plan_fingerprint(execution_store): """步骤 intent 必须精确绑定当前持久计划的 canonical 指纹。""" _seed_pending(execution_store) @@ -607,11 +636,24 @@ def test_success_path_persists_steps_and_execution_checkpoint(execution_store): def test_provider_predecessor_remains_owned_after_host_plan_promotion( - execution_store, + execution_store, monkeypatch, ): - """provider 回退升级宿主计划后，序号零证据仍应被严格重建并纳入 checkpoint。""" - task_id = "task-1" - planning_input = _planning_input(task_id) + """真实准入冻结 provider 后可执行，回退升级宿主计划仍保留序号零证据。""" + admission_repository = TransactionalTransferAdmissionRepository(execution_store) + planning_input = _planning_input("task-1") + admitted = admission_repository.admit( + storage="local", + src_path="/task-1.mkv", + planning_input=planning_input, + ) + task_id = admitted.task_id + claimed = admission_repository.claim_task( + task_id=task_id, + owner_id="provider-worker", + lease_seconds=3600, + ) + assert claimed is not None + lease_token = claimed.lease_token provider = TransferProviderReference( plugin_id="provider-a", plugin_name="Provider A", @@ -637,13 +679,18 @@ def test_provider_predecessor_remains_owned_after_host_plan_promotion( legacy_transfer_providers=(provider,), provider_invocation=invocation, ) - _seed_pending(execution_store) + monkeypatch.setattr(admission_repository, "_now", lambda: "2026-08-27 09:00:00") + admission_repository.checkpoint_plan( + task_id=task_id, + lease_token=lease_token, + input_fingerprint=planning_input.fingerprint, + checkpoint=provider_checkpoint, + ) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None - pending.state = "provider_pending" - pending.checkpoint_payload = provider_checkpoint.to_payload() - session.commit() + assert pending.state == "provider_pending" + assert pending.planned_at == "2026-08-27 09:00:00" _, command = _repository(execution_store) provider_intent = TransferStepIntent.create( task_id=task_id, @@ -660,30 +707,42 @@ def test_provider_predecessor_remains_owned_after_host_plan_promotion( ) prepared = command.prepare( task_id=task_id, - lease_token="lease-1", + lease_token=lease_token, intent=provider_intent, ) - started = command.begin( - task_id=task_id, - lease_token="lease-1", - operation_id=prepared.operation_id, - ) - provider_succeeded = command.complete( - task_id=task_id, - lease_token="lease-1", - step=started, - result=TransferStepResult(payload={"handled": False}), - ) - promoted_checkpoint = replace( - _plan_checkpoint(task_id), - pre_execution_cleanup_completed=True, - ) with execution_store() as session: pending = session.scalar(select(TransferPending)) assert pending is not None - pending.state = "planned" - pending.checkpoint_payload = promoted_checkpoint.to_payload() - session.commit() + assert pending.execution_state == "running" + assert pending.state == "provider_pending" + assert session.scalar(select(TransferExecutionStep)).state == "prepared" + started = command.begin( + task_id=task_id, + lease_token=lease_token, + operation_id=prepared.operation_id, + ) + provider_succeeded = command.complete( + task_id=task_id, + lease_token=lease_token, + step=started, + result=TransferStepResult(payload={"handled": False}), + ) + promoted_checkpoint = replace( + _plan_checkpoint("task-1"), + pre_execution_cleanup_completed=True, + ) + monkeypatch.setattr(admission_repository, "_now", lambda: "2026-08-27 09:01:00") + admission_repository.checkpoint_plan( + task_id=task_id, + lease_token=lease_token, + input_fingerprint=planning_input.fingerprint, + checkpoint=promoted_checkpoint, + ) + with execution_store() as session: + pending = session.scalar(select(TransferPending)) + assert pending is not None + assert pending.state == "planned" + assert pending.planned_at == "2026-08-27 09:01:00" promoted_fingerprint = build_transfer_checkpoint_fingerprint( promoted_checkpoint.to_payload() ) @@ -697,17 +756,17 @@ def test_provider_predecessor_remains_owned_after_host_plan_promotion( ) host_prepared = command.prepare( task_id=task_id, - lease_token="lease-1", + lease_token=lease_token, intent=host_intent, ) host_started = command.begin( task_id=task_id, - lease_token="lease-1", + lease_token=lease_token, operation_id=host_prepared.operation_id, ) host_succeeded = command.complete( task_id=task_id, - lease_token="lease-1", + lease_token=lease_token, step=host_started, result=TransferStepResult(payload={"dest_exists": True}), ) @@ -720,7 +779,7 @@ def test_provider_predecessor_remains_owned_after_host_plan_promotion( ) snapshot = command.checkpoint( task_id=task_id, - lease_token="lease-1", + lease_token=lease_token, checkpoint=execution_checkpoint, ) assert snapshot.state is TransferExecutionState.SETTLING diff --git a/tests/test_transfer_lease_persistence.py b/tests/test_transfer_lease_persistence.py index 5547dc76b..b2069a042 100644 --- a/tests/test_transfer_lease_persistence.py +++ b/tests/test_transfer_lease_persistence.py @@ -726,12 +726,12 @@ def test_planning_writes_require_current_unexpired_lease( assert planned.attempt_count == 2 -def test_provider_checkpoint_sets_planned_time_only_after_host_plan( +def test_provider_checkpoint_sets_planned_time_and_host_plan_refreshes_it( repository_factory, lease_clock, business_clock, ) -> None: - """provider 快照不是规划完成，planned_at 只记录首次宿主完整计划。""" + """provider 阶段提交冻结时间，宿主计划刷新该时间，幂等重试不改动。""" repository = repository_factory() path = "/downloads/provider-plan.mkv" planning_input = _planning_input(path) @@ -756,7 +756,7 @@ def test_provider_checkpoint_sets_planned_time_only_after_host_plan( checkpoint=_provider_checkpoint(planning_input), ) provider_snapshot = _pending_snapshot(repository, admitted.task_id) - assert provider_snapshot["planned_at"] is None + assert provider_snapshot["planned_at"] == "2026-08-27 18:02:00" assert provider_snapshot["updated_at"] == "2026-08-27 18:02:00" business_clock["now"] = "2026-08-27 18:03:00"