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
2026-11-02 21:34:32 +08:00 · 2026-10-13 07:13:42 +08:00
parent 0c640be189
commit 435703573e
5 changed files with 104 additions and 41 deletions
--- 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
--- 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,
--- 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,
--- 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
--- 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"