优化加载后删除节点的逻辑；和数据库查询性能

2026-04-02 19:31:04 +08:00
parent 74df53889c
commit 543c3ff906
17 changed files with 571 additions and 32 deletions
@@ -146,3 +146,39 @@ async def test_api_datasource_failed_update_does_not_use_unrelated_op_trace() ->
    assert updated.sync_log is not None
    assert "API链路[submit-failed] PUT /material/plan" not in updated.sync_log
    assert "未找到匹配 biz_id 的请求记录" in updated.sync_log
+
+
+@pytest.mark.asyncio
+async def test_api_datasource_failed_create_uses_push_id_trace_when_data_id_empty() -> None:
+    collection = DataCollection("remote")
+    ds = ApiDataSource(base_url="http://example.com", uid="u", secret="s")
+    ds.set_collection(collection)
+
+    node = _build_node(node_id="n-create-fail", data_id="", name="X")
+    with node.allow_core_state_write("test"):
+        node.binding_status = BindingStatus.NORMAL
+        node.action = SyncAction.CREATE
+        node.status = SyncStatus.IN_PROGRESS
+    await collection.add(node)
+
+    ds.client._record_trace(
+        method="POST",
+        endpoint="/construction/create",
+        url="http://example.com/construction/create",
+        request_params=None,
+        request_payload={"push_id": "push-create-fail", "data": {"task_type": "jxfx"}},
+        response={"code": 400, "message": "施工类型错误"},
+        elapsed_ms=12,
+    )
+
+    await ds._handle_task_result(
+        cast(Any, _Handler()),
+        TaskResult.failed(node_id=node.node_id, error="Create failed: 施工类型错误", push_id="push-create-fail"),
+        {},
+    )
+
+    updated = collection.get(node.node_id)
+    assert updated is not None
+    assert updated.sync_log is not None
+    assert "API链路[submit-failed] POST /construction/create" in updated.sync_log
+    assert "response={\"code\":400,\"message\":\"施工类型错误\"}" in updated.sync_log
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import pytest
+
+from sync_state_machine.domain.construction.api_handler import ConstructionTaskApiHandler
+from sync_state_machine.domain.construction.sync_node import ConstructionTaskSyncNode
+
+
+@pytest.mark.asyncio
+async def test_construction_create_failure_preserves_push_id(monkeypatch: pytest.MonkeyPatch) -> None:
+    handler = ConstructionTaskApiHandler()
+    handler.api_client = object()
+
+    async def _fake_create(client, data, push_id: str, *, steps=None):
+        del client, data, steps
+        raise Exception(f"API error for {push_id}")
+
+    monkeypatch.setattr(
+        "sync_state_machine.domain.construction.api_handler.api_create_construction_task",
+        _fake_create,
+    )
+
+    node = ConstructionTaskSyncNode(
+        node_id="n1",
+        data_id="",
+        data={
+            "id": "",
+            "project_id": "p1",
+            "contract_id": "c1",
+            "show_name": "架线工程",
+            "construction_section": "SC",
+            "status": -1,
+            "task_type": "jxfx",
+            "plan_start_date": "2024-06-18",
+            "plan_complete_date": "2024-12-05",
+            "actual_complete_date": "2025-12-29",
+            "complete_quantity": 2.0,
+            "complete_rate": 100.0,
+            "contract_quantity": 2.0,
+            "plan_node": [],
+            "statistic_date": "2025-12-29",
+            "is_crucial": False,
+            "delay_days": None,
+        },
+    )
+    node.context["project_id"] = "p1"
+
+    results = await handler.create_all([node])
+
+    assert len(results) == 1
+    result = results[0]
+    assert result.status.value == "failed"
+    assert result.push_id is not None
+    assert result.error == f"API error for {result.push_id}"
@@ -52,7 +52,13 @@ def test_e01_bootstrap_accepts_all_combinations(action: SyncAction, status: Sync
    original_error = node.error

    is_zombie = _is_create_zombie(node)
-    decision = e01_bootstrap(runtime, node=node, is_create_zombie=is_zombie)
+    decision = e01_bootstrap(
+        runtime,
+        node=node,
+        is_create_zombie=is_zombie,
+        bootstrap_source_present=True,
+        has_binding_record=False,
+    )

    if is_zombie:
        assert decision.to_state == "S15"
@@ -66,3 +72,48 @@ def test_e01_bootstrap_accepts_all_combinations(action: SyncAction, status: Sync
        assert node.action == SyncAction.NONE
        assert node.status == SyncStatus.PENDING
        assert node.error is None
+
+
+def test_e01_bootstrap_cleans_unbound_persistence_only_node_on_source_miss() -> None:
+    runtime = StateMachineRuntime(StateMachineConfig.load(CFG_PATH))
+    node = _new_node(
+        node_id="e01-stale-unbound",
+        action=SyncAction.NONE,
+        status=SyncStatus.PENDING,
+        data_id="id-1",
+    )
+
+    decision = e01_bootstrap(
+        runtime,
+        node=node,
+        is_create_zombie=False,
+        bootstrap_source_present=False,
+        has_binding_record=False,
+    )
+
+    assert decision.to_state == "S15"
+
+
+def test_e01_bootstrap_isolates_bound_node_on_source_miss() -> None:
+    runtime = StateMachineRuntime(StateMachineConfig.load(CFG_PATH))
+    node = _new_node(
+        node_id="e01-stale-bound",
+        action=SyncAction.NONE,
+        status=SyncStatus.PENDING,
+        data_id="id-1",
+    )
+
+    decision = e01_bootstrap(
+        runtime,
+        node=node,
+        is_create_zombie=False,
+        bootstrap_source_present=False,
+        has_binding_record=True,
+    )
+
+    assert decision.to_state == "S17"
+    assert node.binding_status == BindingStatus.ABNORMAL
+    assert node.action == SyncAction.NONE
+    assert node.status == SyncStatus.FAILED
+    assert node.context["bootstrap_blocked"] is True
+    assert "BOOTSTRAP_SOURCE_MISSING" in (node.error or "")
@@ -7,7 +7,10 @@ import pytest

 from sync_state_machine.common.binding import BindingManager
 from sync_state_machine.common.collection import DataCollection
+from sync_state_machine.common.sqlite_backend import SQLitePersistenceBackend
+from sync_state_machine.common.types import BindingStatus, SyncAction, SyncStatus
 from sync_state_machine.datasource.jsonl import JsonlDataSource
+from sync_state_machine.domain.project.sync_node import ProjectSyncNode
 import sync_state_machine.pipeline.full_sync_pipeline as full_sync_pipeline_module
 from sync_state_machine.pipeline.full_sync_pipeline import FullSyncPipeline, NodeTypeLoadError, WriteOutcome

@@ -30,6 +33,37 @@ class _FakeDataSource:
        return None


+class _BootstrapRefreshDataSource(_FakeDataSource):
+    def __init__(self, *, node_data: dict[str, str]) -> None:
+        self.node_data = dict(node_data)
+
+    async def load_all(self, order=None, data_type=None, raise_on_error=False) -> None:
+        del data_type, raise_on_error
+        if self.collection is None:
+            raise RuntimeError("collection not set")
+        if not order or tuple(order) != ("project",):
+            return
+
+        existing = self.collection.get_by_data_id("project", self.node_data["id"])
+        if existing is None:
+            node = ProjectSyncNode(
+                node_id="bootstrap-refreshed-project",
+                data_id=self.node_data["id"],
+                data=self.node_data,
+            )
+            node.append_log(f"加载来源: {self.__class__.__name__}.load 新增节点")
+            await self.collection.add(node)
+            return
+
+        existing.set_data(self.node_data)
+        existing.set_origin_data(self.node_data)
+        if existing.context.pop("_loaded_from_persistence", False):
+            existing.append_log(f"加载来源: persistence -> {self.__class__.__name__}.load 覆盖刷新")
+        else:
+            existing.append_log(f"加载来源: {self.__class__.__name__}.load 覆盖刷新")
+        await self.collection.update(existing)
+
+
 class _StubStrategy:
    def __init__(self, node_type: str):
        self.node_type = node_type
@@ -348,4 +382,73 @@ async def test_run_persists_before_reraising_load_failure(monkeypatch: pytest.Mo
        await pipeline.run()

    assert calls == ["bootstrap", "sync", "persist"]
-    assert persistence.close_calls == 1
+    assert persistence.close_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_phase_bootstrap_refreshes_persisted_nodes_before_e01_cleanup(tmp_path: Path) -> None:
+    db_path = tmp_path / "bootstrap-refresh.db"
+    persistence = SQLitePersistenceBackend(str(db_path))
+    await persistence.initialize()
+
+    seed_remote = DataCollection("remote", persistence=persistence, auto_persist=False)
+    stale_node = ProjectSyncNode(
+        node_id="persisted-project-node",
+        data_id="proj-bootstrap-1",
+        data={
+            "id": "proj-bootstrap-1",
+            "type": "project",
+            "name": "Persisted Project",
+            "code": "P-BOOT",
+            "status": "active",
+            "progress_type": "C",
+            "project_type": "onshore_wind",
+        },
+        action=SyncAction.NONE,
+        status=SyncStatus.PENDING,
+        binding_status=BindingStatus.WARNING,
+    )
+    await seed_remote.add(stale_node)
+    await seed_remote.persist()
+
+    local_collection = DataCollection("local", persistence=persistence, auto_persist=False)
+    remote_collection = DataCollection("remote", persistence=persistence, auto_persist=False)
+    binding_manager = BindingManager(persistence, auto_persist=False)
+
+    pipeline = FullSyncPipeline(
+        local_datasource=_FakeDataSource(),
+        remote_datasource=_BootstrapRefreshDataSource(
+            node_data={
+                "id": "proj-bootstrap-1",
+                "type": "project",
+                "name": "Refreshed Project",
+                "code": "P-BOOT",
+                "status": "active",
+                "progress_type": "C",
+                "project_type": "onshore_wind",
+            }
+        ),
+        strategies=[_StubStrategy("project")],
+        persistence=persistence,
+        local_collection=local_collection,
+        remote_collection=remote_collection,
+        binding_manager=binding_manager,
+        node_types=["project"],
+        load_order=["project"],
+        sync_order=["project"],
+        enable_persistence=True,
+    )
+
+    await pipeline.phase_bootstrap()
+
+    refreshed = remote_collection.get("persisted-project-node")
+    assert refreshed is not None
+    assert refreshed.binding_status == BindingStatus.UNCHECKED
+    assert refreshed.action == SyncAction.NONE
+    assert refreshed.status == SyncStatus.PENDING
+    assert remote_collection.is_bootstrap_blocked(refreshed) is False
+    assert refreshed.sync_log is not None
+    assert "加载来源: persistence 恢复节点" in refreshed.sync_log
+    assert "加载来源: persistence -> _BootstrapRefreshDataSource.load 覆盖刷新" in refreshed.sync_log
+
+    await pipeline.close()
@@ -55,4 +55,22 @@ async def test_scope_partitioning_and_copy_between_scopes(tmp_path: Path) -> Non
    await reloaded_global.load_from_persistence()
    assert len(reloaded_global.filter(node_type="test_project")) == 1

+    await persistence.close()
+
+
+@pytest.mark.asyncio
+async def test_sqlite_backend_creates_restore_join_index(tmp_path: Path) -> None:
+    db_path = tmp_path / "scope_partitioning.db"
+    persistence = SQLitePersistenceBackend(str(db_path))
+    await persistence.initialize()
+
+    columns, rows = PersistenceBackend.query_records(
+        backend="sqlite",
+        db_path=str(db_path),
+        sql="PRAGMA index_list('nodes')",
+    )
+
+    assert columns == ["seq", "name", "unique", "origin", "partial"]
+    assert any(row["name"] == "idx_nodes_scope_type_node_id" for row in rows)
+
    await persistence.close()