strepsiades
802 lines
34 KiB
Python
802 lines
34 KiB
Python
"""
|
||
Full Sync Pipeline - Implements document-specified end-to-end workflow.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING, Any, Dict, List, Optional
|
||
|
||
if TYPE_CHECKING:
|
||
from ..datasource.datasource import BaseDataSource
|
||
|
||
from ..datasource.jsonl import JsonlDataSource
|
||
from ..common.collection import DataCollection
|
||
from ..common.binding import BindingManager
|
||
from ..logging import get_logger
|
||
from ..common.persistence import PersistenceBackend
|
||
from ..sync_system.config import UpdateDirection
|
||
from ..sync_system.strategy import BaseSyncStrategy
|
||
from ..engine import StateMachineConfig, StateMachineRuntime
|
||
from ..engine import e41_post_create_ready
|
||
from ..sync_system.strategy_ops import finalize_peer_creating_sources, run_phase2_cleanup
|
||
from ..sync_system.strategy_ops.bind_ops import refresh_bind_data_id
|
||
from ..sync_system.strategy_ops.post_check_ops import evaluate_consistency_for_node_type
|
||
from .summary_report import print_pipeline_summary
|
||
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class NodeTypeLoadError(RuntimeError):
|
||
"""Raised when a datasource load for a node type fails and the pipeline must stop."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class WriteOutcome:
|
||
local_written: bool = False
|
||
remote_written: bool = False
|
||
|
||
@property
|
||
def any_written(self) -> bool:
|
||
return self.local_written or self.remote_written
|
||
|
||
|
||
class FullSyncPipeline:
|
||
"""
|
||
Full pipeline phases:
|
||
1) bootstrap
|
||
2) bind
|
||
3) create
|
||
4) update
|
||
5) post-check
|
||
6) persistence
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
local_datasource: "BaseDataSource",
|
||
remote_datasource: "BaseDataSource",
|
||
strategies: List[BaseSyncStrategy],
|
||
persistence: PersistenceBackend,
|
||
local_collection: DataCollection,
|
||
remote_collection: DataCollection,
|
||
binding_manager: BindingManager,
|
||
scope: str = "global",
|
||
pipeline_name: str | None = None,
|
||
bootstrap_binding_node_types: Optional[List[str]] = None,
|
||
ephemeral_node_types: Optional[List[str]] = None,
|
||
node_types: Optional[List[str]] = None,
|
||
load_order: Optional[List[str]] = None,
|
||
sync_order: Optional[List[str]] = None,
|
||
enable_persistence: bool = True,
|
||
):
|
||
# 验证必需参数（由工厂函数创建）
|
||
if strategies is None:
|
||
raise ValueError("strategies is required (use factory function to create pipeline)")
|
||
if local_collection is None or remote_collection is None or binding_manager is None:
|
||
raise ValueError("collections and binding_manager are required (use factory function to create pipeline)")
|
||
|
||
self.local_datasource = local_datasource
|
||
self.remote_datasource = remote_datasource
|
||
self.strategies = strategies
|
||
self.scope = scope
|
||
self.pipeline_name = pipeline_name or f"pipeline[{scope}]"
|
||
self.node_types = node_types or [s.node_type for s in self.strategies]
|
||
self.bootstrap_binding_node_types = list(dict.fromkeys(bootstrap_binding_node_types or []))
|
||
self.ephemeral_node_types = list(dict.fromkeys(ephemeral_node_types or []))
|
||
self.load_order = load_order or self.node_types
|
||
self.sync_order = sync_order or self.node_types
|
||
self.persistence = persistence
|
||
self.enable_persistence = enable_persistence
|
||
|
||
self.local_collection = local_collection
|
||
self.remote_collection = remote_collection
|
||
self.binding_manager = binding_manager
|
||
|
||
# 统一状态机 runtime 注入（唯一入口）
|
||
runtime = self._resolve_pipeline_runtime()
|
||
self.sm_runtime = runtime
|
||
self.local_collection.set_state_machine_runtime(runtime)
|
||
self.remote_collection.set_state_machine_runtime(runtime)
|
||
self.local_datasource.set_state_machine_runtime(runtime)
|
||
self.remote_datasource.set_state_machine_runtime(runtime)
|
||
for strategy in self.strategies:
|
||
strategy.sm_runtime = runtime
|
||
|
||
self.stats: Dict[str, Any] = {
|
||
"loaded": 0,
|
||
"synced": 0,
|
||
"failed": 0,
|
||
"post_check_passed": True,
|
||
}
|
||
self._consistency_by_type: Dict[str, Dict[str, Any]] = {}
|
||
self._logger = logger
|
||
self._closed = False
|
||
self._loaded_node_types: set[str] = set()
|
||
self._preloaded_local_nodes: List[Any] = []
|
||
self._preloaded_remote_nodes: List[Any] = []
|
||
self._preloaded_bindings: Dict[str, Dict[str, Optional[str]]] = {}
|
||
|
||
def _resolve_pipeline_runtime(self) -> StateMachineRuntime:
|
||
if self.strategies:
|
||
runtime = self.strategies[0].sm_runtime
|
||
if runtime is not None:
|
||
return runtime
|
||
|
||
cfg_path = Path(__file__).resolve().parents[1] / "config" / "node_state_machine.yaml"
|
||
return StateMachineRuntime(StateMachineConfig.load(cfg_path))
|
||
|
||
def _strategy_map(self) -> Dict[str, BaseSyncStrategy]:
|
||
return {strategy.node_type: strategy for strategy in self.strategies}
|
||
|
||
def _iter_sync_strategies(self):
|
||
strategy_map = self._strategy_map()
|
||
for node_type in self.sync_order:
|
||
strategy = strategy_map.get(node_type)
|
||
if strategy is not None:
|
||
yield node_type, strategy
|
||
|
||
def _get_strategy(self, node_type: str) -> BaseSyncStrategy | None:
|
||
return next((item for item in self.strategies if item.node_type == node_type), None)
|
||
|
||
def _log_phase_completed(self, phase_name: str) -> None:
|
||
self._logger.info("✅ Phase %s completed (%s)", phase_name, self.pipeline_name)
|
||
|
||
def _log_strategy_start(self, node_type: str) -> None:
|
||
section_width = 96
|
||
self._logger.info("\n" + "-" * section_width)
|
||
self._logger.info("🧩 Strategy start: %s", node_type)
|
||
self._logger.info("-" * section_width)
|
||
|
||
def _log_strategy_result(self, action: str, node_type: str, count: int | None = None) -> None:
|
||
if count is None:
|
||
self._logger.info("✅ Strategy %s: %s", action, node_type)
|
||
return
|
||
self._logger.info("✅ Strategy %s: %s (%s=%d)", action, node_type, action, count)
|
||
|
||
def _log_strategy_done(self, node_type: str) -> None:
|
||
self._logger.info("✅ Strategy done: %s", node_type)
|
||
|
||
def _log_strategy_failure(self, stage: str, node_type: str, exc: Exception) -> None:
|
||
self._logger.error(
|
||
"❌ Strategy %s failed but pipeline continues: %s | %s: %s",
|
||
stage,
|
||
node_type,
|
||
type(exc).__name__,
|
||
exc,
|
||
)
|
||
|
||
def _log_reload_skipped(self, node_type: str, reason: str) -> None:
|
||
self._logger.info("⏭️ Reload skipped for JSONL pipeline: node_type=%s reason=%s", node_type, reason)
|
||
|
||
async def _safe_close(self, name: str, close_coro, timeout: float = 3.0) -> None:
|
||
try:
|
||
await asyncio.wait_for(close_coro, timeout=timeout)
|
||
except asyncio.TimeoutError:
|
||
self._logger.warning("⚠️ close timeout: %s (>%ss)", name, timeout)
|
||
except Exception as exc:
|
||
self._logger.warning("⚠️ close failed: %s (%s)", name, exc)
|
||
|
||
async def run(self) -> Dict[str, Any]:
|
||
persistence_started = False
|
||
try:
|
||
await self.phase_bootstrap()
|
||
self._log_phase_completed("bootstrap")
|
||
|
||
await self.phase_sync_by_node_type()
|
||
self._log_phase_completed("sync-by-node-type")
|
||
|
||
post_ok = await self.phase_post_check()
|
||
self.stats["post_check_passed"] = post_ok
|
||
self._log_phase_completed("post-check")
|
||
|
||
persistence_started = True
|
||
await self.phase_persistence()
|
||
self._log_phase_completed("persistence")
|
||
return self.stats
|
||
except Exception as exc:
|
||
if not persistence_started:
|
||
await self._persist_after_failure(exc)
|
||
raise
|
||
finally:
|
||
await self.close()
|
||
|
||
async def _persist_after_failure(self, cause: Exception) -> None:
|
||
if not self.enable_persistence:
|
||
self._logger.info(
|
||
"⏭️ Persistence disabled after pipeline failure: %s (%s)",
|
||
type(cause).__name__,
|
||
self.pipeline_name,
|
||
)
|
||
return
|
||
|
||
self._logger.info(
|
||
"🔄 Pipeline failed before persistence, persisting current state: %s | %s",
|
||
type(cause).__name__,
|
||
cause,
|
||
)
|
||
try:
|
||
await self.phase_persistence()
|
||
self._logger.info("✅ Phase persistence completed after failure (%s)", self.pipeline_name)
|
||
except Exception as persist_exc:
|
||
self._logger.error(
|
||
"❌ Phase persistence failed after pipeline failure: %s | %s",
|
||
type(persist_exc).__name__,
|
||
persist_exc,
|
||
)
|
||
|
||
async def phase_sync_by_node_type(self) -> None:
|
||
self._prepare_datasources()
|
||
for node_type, strategy in self._iter_sync_strategies():
|
||
try:
|
||
self._log_strategy_start(node_type)
|
||
|
||
await self._load_node_type(node_type, reason="pre-strategy refresh")
|
||
|
||
# Load failure is a hard stop: create/update cannot safely treat missing data as empty data.
|
||
await strategy.bind()
|
||
self._log_strategy_result("bind", node_type)
|
||
|
||
if strategy.skip_sync:
|
||
self._logger.info("⏭️ Skipping create/update for %s (skip_sync=True)", node_type)
|
||
continue
|
||
|
||
created = await strategy.create()
|
||
self._log_strategy_result("create", node_type, len(created))
|
||
created_local_node_ids = {
|
||
node.node_id for node in created if self.local_collection.get_by_node_id(node.node_id) is not None
|
||
}
|
||
created_remote_node_ids = {
|
||
node.node_id for node in created if self.remote_collection.get_by_node_id(node.node_id) is not None
|
||
}
|
||
create_outcome = await self.commit_creates(node_type)
|
||
await self.normalize_create_success_to_update_entry(node_type)
|
||
if create_outcome.any_written:
|
||
await self._reload_node_type_after_write(
|
||
node_type,
|
||
reason="create wrote data",
|
||
reload_local=create_outcome.local_written,
|
||
reload_remote=create_outcome.remote_written,
|
||
)
|
||
await refresh_bind_data_id(
|
||
node_type=node_type,
|
||
local_collection=self.local_collection,
|
||
remote_collection=self.remote_collection,
|
||
binding_manager=self.binding_manager,
|
||
local_node_ids=created_local_node_ids,
|
||
remote_node_ids=created_remote_node_ids,
|
||
)
|
||
|
||
updated = await strategy.update()
|
||
self._log_strategy_result("update", node_type, len(updated))
|
||
update_outcome = await self.commit_updates(node_type)
|
||
if update_outcome.any_written:
|
||
await self._reload_node_type_after_write(
|
||
node_type,
|
||
reason="update wrote data",
|
||
reload_local=update_outcome.local_written,
|
||
reload_remote=update_outcome.remote_written,
|
||
)
|
||
|
||
self._log_strategy_done(node_type)
|
||
except NodeTypeLoadError:
|
||
raise
|
||
except Exception as exc:
|
||
self.stats["failed"] += 1
|
||
self._log_strategy_failure("sync", node_type, exc)
|
||
continue
|
||
|
||
async def phase_bootstrap(self) -> None:
|
||
await self._load_persistence_state()
|
||
self._logger.info("✅ Bootstrap: persistence loaded")
|
||
|
||
await self._apply_preloaded_state()
|
||
self._logger.info("✅ Bootstrap: shared preload applied")
|
||
|
||
self._prepare_datasources()
|
||
self._logger.info("✅ Bootstrap: datasource prepared")
|
||
|
||
await self._bootstrap_refresh_latest_data()
|
||
self._logger.info("✅ Bootstrap: datasource refresh completed")
|
||
|
||
await self._run_bootstrap_reset_cleanup()
|
||
self._logger.info("✅ Bootstrap: reset cleanup completed")
|
||
|
||
await self._apply_project_scope_cleanup()
|
||
|
||
async def phase_bind(self) -> None:
|
||
self._prepare_datasources()
|
||
for node_type, strategy in self._iter_sync_strategies():
|
||
try:
|
||
await strategy.bind()
|
||
except Exception as exc:
|
||
self.stats["failed"] += 1
|
||
self._log_strategy_failure("bind", node_type, exc)
|
||
continue
|
||
|
||
async def phase_create(self) -> None:
|
||
self._prepare_datasources()
|
||
for node_type, strategy in self._iter_sync_strategies():
|
||
if strategy.skip_sync:
|
||
self._logger.info("⏭️ Skipping create for %s (skip_sync=True)", node_type)
|
||
continue
|
||
try:
|
||
created = await strategy.create()
|
||
self._log_strategy_result("create", node_type, len(created))
|
||
created_local_node_ids = {
|
||
node.node_id for node in created if self.local_collection.get_by_node_id(node.node_id) is not None
|
||
}
|
||
created_remote_node_ids = {
|
||
node.node_id for node in created if self.remote_collection.get_by_node_id(node.node_id) is not None
|
||
}
|
||
create_outcome = await self.commit_creates(node_type)
|
||
await self.normalize_create_success_to_update_entry(node_type)
|
||
if create_outcome.any_written:
|
||
await self._reload_node_type_after_write(
|
||
node_type,
|
||
reason="create wrote data",
|
||
reload_local=create_outcome.local_written,
|
||
reload_remote=create_outcome.remote_written,
|
||
)
|
||
await refresh_bind_data_id(
|
||
node_type=node_type,
|
||
local_collection=self.local_collection,
|
||
remote_collection=self.remote_collection,
|
||
binding_manager=self.binding_manager,
|
||
local_node_ids=created_local_node_ids,
|
||
remote_node_ids=created_remote_node_ids,
|
||
)
|
||
except Exception as exc:
|
||
self.stats["failed"] += 1
|
||
self._log_strategy_failure("create", node_type, exc)
|
||
continue
|
||
|
||
async def phase_update(self) -> None:
|
||
self._prepare_datasources()
|
||
for node_type, strategy in self._iter_sync_strategies():
|
||
if strategy.skip_sync:
|
||
self._logger.info("⏭️ Skipping update for %s (skip_sync=True)", node_type)
|
||
continue
|
||
try:
|
||
updated = await strategy.update()
|
||
self._log_strategy_result("update", node_type, len(updated))
|
||
update_outcome = await self.commit_updates(node_type)
|
||
if update_outcome.any_written:
|
||
await self._reload_node_type_after_write(
|
||
node_type,
|
||
reason="update wrote data",
|
||
reload_local=update_outcome.local_written,
|
||
reload_remote=update_outcome.remote_written,
|
||
)
|
||
await self._evaluate_consistency_for_node_type(node_type)
|
||
except Exception as exc:
|
||
self.stats["failed"] += 1
|
||
self._log_strategy_failure("update", node_type, exc)
|
||
continue
|
||
|
||
async def close(self) -> None:
|
||
"""关闭所有资源（DataSource、Persistence）"""
|
||
if self._closed:
|
||
return # 已经关闭过，避免重复关闭
|
||
self._closed = True
|
||
|
||
await self._safe_close("remote_datasource", self.remote_datasource.close())
|
||
await self._safe_close("local_datasource", self.local_datasource.close())
|
||
await self._safe_close("persistence", self.persistence.close())
|
||
|
||
async def _load_persistence_state(self) -> None:
|
||
"""加载持久化数据到内存，等待 bootstrap datasource refresh 与 E01 cleanup。"""
|
||
if not self.enable_persistence:
|
||
return
|
||
|
||
excluded_node_types = self.ephemeral_node_types or None
|
||
|
||
# 加载持久化数据
|
||
await self.local_collection.load_from_persistence_excluding(excluded_node_types)
|
||
await self.remote_collection.load_from_persistence_excluding(excluded_node_types)
|
||
binding_node_types = [
|
||
node_type
|
||
for node_type in dict.fromkeys([*self.node_types, *self.bootstrap_binding_node_types])
|
||
if node_type not in set(self.ephemeral_node_types)
|
||
]
|
||
for node_type in binding_node_types:
|
||
await self.binding_manager.load_from_persistence(node_type)
|
||
self._logger.info("🔄 Persistence loaded (state preserved; pending E01 normalization)")
|
||
|
||
def _resolve_bootstrap_refresh_node_types(self) -> List[str]:
|
||
local_types = {node.node_type for node in self.local_collection.filter()}
|
||
remote_types = {node.node_type for node in self.remote_collection.filter()}
|
||
candidate_types = local_types | remote_types
|
||
if not candidate_types:
|
||
return []
|
||
|
||
ordered: List[str] = []
|
||
for node_type in [*self.load_order, *self.bootstrap_binding_node_types, *sorted(candidate_types)]:
|
||
if node_type in candidate_types and node_type not in ordered:
|
||
ordered.append(node_type)
|
||
return ordered
|
||
|
||
async def _bootstrap_refresh_latest_data(self) -> None:
|
||
if not self.enable_persistence:
|
||
return
|
||
|
||
node_types = self._resolve_bootstrap_refresh_node_types()
|
||
if not node_types:
|
||
self._logger.info("⏭️ Bootstrap datasource refresh skipped: no persisted node types")
|
||
return
|
||
|
||
for node_type in node_types:
|
||
self._logger.info("🔄 Bootstrap refresh node_type=%s from datasource", node_type)
|
||
await self._load_node_type_from_datasource(
|
||
self.local_datasource,
|
||
datasource_role="local",
|
||
node_type=node_type,
|
||
reason="bootstrap persistence refresh",
|
||
)
|
||
await self._load_node_type_from_datasource(
|
||
self.remote_datasource,
|
||
datasource_role="remote",
|
||
node_type=node_type,
|
||
reason="bootstrap persistence refresh",
|
||
)
|
||
|
||
async def _run_bootstrap_reset_cleanup(self) -> None:
|
||
if not self.enable_persistence:
|
||
return
|
||
if not self.strategies:
|
||
raise RuntimeError("no strategies configured: cannot run bootstrap event E01")
|
||
|
||
total_cleaned = await run_phase2_cleanup(
|
||
node_types=self.node_types,
|
||
local_collection=self.local_collection,
|
||
remote_collection=self.remote_collection,
|
||
binding_manager=self.binding_manager,
|
||
runtime=self.sm_runtime,
|
||
)
|
||
if total_cleaned > 0:
|
||
self._logger.info(f"🧹 Reset cleanup: {total_cleaned} bootstrap zombies cleaned")
|
||
|
||
async def _apply_preloaded_state(self) -> None:
|
||
if self._preloaded_local_nodes:
|
||
await self.local_collection.import_nodes(self._preloaded_local_nodes)
|
||
if self._preloaded_remote_nodes:
|
||
await self.remote_collection.import_nodes(self._preloaded_remote_nodes)
|
||
for node_type, bindings in self._preloaded_bindings.items():
|
||
self.binding_manager.import_bindings(node_type, bindings)
|
||
|
||
def preload_shared_state(
|
||
self,
|
||
*,
|
||
local_nodes: List[Any],
|
||
remote_nodes: List[Any],
|
||
bindings_by_type: Dict[str, Dict[str, Optional[str]]],
|
||
) -> None:
|
||
self._preloaded_local_nodes = list(local_nodes)
|
||
self._preloaded_remote_nodes = list(remote_nodes)
|
||
self._preloaded_bindings = {
|
||
node_type: dict(bindings)
|
||
for node_type, bindings in bindings_by_type.items()
|
||
}
|
||
|
||
def _prepare_datasources(self) -> None:
|
||
"""为后续按 node_type 加载准备 collection 绑定。"""
|
||
self.local_datasource.set_collection(self.local_collection)
|
||
self.remote_datasource.set_collection(self.remote_collection)
|
||
|
||
def _get_project_scope_filter_ids(self, datasource: "BaseDataSource") -> List[str]:
|
||
try:
|
||
handler = datasource.get_handler("project")
|
||
get_data_id_filter = getattr(handler, "get_data_id_filter", None)
|
||
if callable(get_data_id_filter):
|
||
raw_filter = get_data_id_filter()
|
||
if isinstance(raw_filter, list):
|
||
return list(dict.fromkeys(raw_filter))
|
||
except Exception:
|
||
pass
|
||
return []
|
||
|
||
async def _apply_project_scope_cleanup(self) -> None:
|
||
local_project_ids = self._get_project_scope_filter_ids(self.local_datasource)
|
||
remote_project_ids = self._get_project_scope_filter_ids(self.remote_datasource)
|
||
if not local_project_ids and not remote_project_ids:
|
||
return
|
||
|
||
deleted_local = await self.local_collection.filter_by_project_ids(local_project_ids)
|
||
deleted_remote = await self.remote_collection.filter_by_project_ids(remote_project_ids)
|
||
deleted_bindings = await self.binding_manager.prune_missing_nodes(
|
||
local_collection=self.local_collection,
|
||
remote_collection=self.remote_collection,
|
||
node_types=list(dict.fromkeys([*self.node_types, *self.bootstrap_binding_node_types])),
|
||
)
|
||
|
||
if deleted_local or deleted_remote or deleted_bindings:
|
||
self._logger.info(
|
||
"🧹 Project-scope cleanup: local_deleted=%d remote_deleted=%d bindings_deleted=%d",
|
||
deleted_local,
|
||
deleted_remote,
|
||
deleted_bindings,
|
||
)
|
||
|
||
async def _load_node_type(self, node_type: str, *, reason: str) -> None:
|
||
self._logger.info(f"🔄 Load node_type={node_type}, reason={reason}")
|
||
await self._load_node_type_from_datasource(
|
||
self.local_datasource,
|
||
datasource_role="local",
|
||
node_type=node_type,
|
||
reason=reason,
|
||
)
|
||
await self._load_node_type_from_datasource(
|
||
self.remote_datasource,
|
||
datasource_role="remote",
|
||
node_type=node_type,
|
||
reason=reason,
|
||
)
|
||
|
||
if node_type in self._loaded_node_types:
|
||
return
|
||
|
||
self._loaded_node_types.add(node_type)
|
||
self.stats["loaded"] += len(self.local_collection.filter(node_type=node_type))
|
||
self.stats["loaded"] += len(self.remote_collection.filter(node_type=node_type))
|
||
|
||
async def _load_node_type_from_datasource(
|
||
self,
|
||
datasource: "BaseDataSource",
|
||
*,
|
||
datasource_role: str,
|
||
node_type: str,
|
||
reason: str,
|
||
) -> None:
|
||
try:
|
||
await datasource.load_all(order=[node_type], raise_on_error=True)
|
||
except Exception as exc:
|
||
raise NodeTypeLoadError(
|
||
f"Load failed for node_type={node_type}, datasource={datasource_role}, reason={reason}: {exc}"
|
||
) from exc
|
||
|
||
def _get_action_success_count(self, datasource: "BaseDataSource", node_type: str, action_key: str) -> int:
|
||
summary = datasource.get_action_summary().get(node_type, {})
|
||
action_summary = summary.get(action_key, {}) if isinstance(summary, dict) else {}
|
||
return int(action_summary.get("success", 0)) if isinstance(action_summary, dict) else 0
|
||
|
||
def _should_skip_reload(self) -> bool:
|
||
return isinstance(self.local_datasource, JsonlDataSource) and isinstance(self.remote_datasource, JsonlDataSource)
|
||
|
||
async def _reload_node_type(self, node_type: str, *, reason: str) -> None:
|
||
if self._should_skip_reload():
|
||
self._log_reload_skipped(node_type, reason)
|
||
return
|
||
await self._load_node_type(node_type, reason=reason)
|
||
|
||
async def _reload_node_type_after_write(
|
||
self,
|
||
node_type: str,
|
||
*,
|
||
reason: str,
|
||
reload_local: bool,
|
||
reload_remote: bool,
|
||
) -> None:
|
||
if not reload_local and not reload_remote:
|
||
return
|
||
|
||
if self._should_skip_reload():
|
||
self._log_reload_skipped(node_type, reason)
|
||
return
|
||
|
||
self._logger.info(
|
||
"🔄 Reload node_type=%s after write, reason=%s, local=%s, remote=%s",
|
||
node_type,
|
||
reason,
|
||
reload_local,
|
||
reload_remote,
|
||
)
|
||
if reload_local:
|
||
await self._load_node_type_from_datasource(
|
||
self.local_datasource,
|
||
datasource_role="local",
|
||
node_type=node_type,
|
||
reason=reason,
|
||
)
|
||
if reload_remote:
|
||
await self._load_node_type_from_datasource(
|
||
self.remote_datasource,
|
||
datasource_role="remote",
|
||
node_type=node_type,
|
||
reason=reason,
|
||
)
|
||
|
||
async def _evaluate_consistency_for_node_type(self, node_type: str) -> Dict[str, Any]:
|
||
strategy = next((item for item in self.strategies if item.node_type == node_type), None)
|
||
if strategy is None:
|
||
raise RuntimeError(f"Missing strategy for node_type={node_type}")
|
||
depend_fields = dict(strategy.config.depend_fields) if strategy and strategy.config.depend_fields else {}
|
||
compare_to_remote = True
|
||
if strategy and strategy.config.update_direction == UpdateDirection.PULL:
|
||
compare_to_remote = False
|
||
ignore_fields = list(
|
||
dict.fromkeys([
|
||
*strategy.config.compare_ignore_fields,
|
||
*strategy.config.post_check_ignore_fields,
|
||
])
|
||
)
|
||
|
||
result = await evaluate_consistency_for_node_type(
|
||
node_type=node_type,
|
||
local_collection=self.local_collection,
|
||
remote_collection=self.remote_collection,
|
||
binding_manager=self.binding_manager,
|
||
logger=self._logger,
|
||
normalize_compare_payload=strategy.normalize_compare_payload,
|
||
depend_fields=depend_fields,
|
||
compare_to_remote=compare_to_remote,
|
||
ignore_fields=ignore_fields,
|
||
)
|
||
self._consistency_by_type[node_type] = result
|
||
return result
|
||
|
||
async def commit_creates(self, node_type: str) -> WriteOutcome:
|
||
"""提交 CREATE 操作（包含异步轮询）"""
|
||
# 执行同步（sync_all 内部会根据 action 筛选节点）
|
||
await self.local_datasource.sync_all(
|
||
order=[node_type],
|
||
poll_async_tasks=True
|
||
)
|
||
|
||
await self.remote_datasource.sync_all(
|
||
order=[node_type],
|
||
poll_async_tasks=True
|
||
)
|
||
|
||
await self._finalize_peer_creating_sources(node_type)
|
||
|
||
# 更新统计
|
||
local_nodes = self.local_collection.filter(node_type=node_type)
|
||
remote_nodes = self.remote_collection.filter(node_type=node_type)
|
||
self.stats["synced"] += len([n for n in local_nodes if n.action.value == "CREATE"])
|
||
self.stats["synced"] += len([n for n in remote_nodes if n.action.value == "CREATE"])
|
||
self.stats["failed"] += len([n for n in local_nodes if n.status.value == "FAILED"])
|
||
self.stats["failed"] += len([n for n in remote_nodes if n.status.value == "FAILED"])
|
||
|
||
local_success = self._get_action_success_count(self.local_datasource, node_type, "create")
|
||
remote_success = self._get_action_success_count(self.remote_datasource, node_type, "create")
|
||
return WriteOutcome(
|
||
local_written=local_success > 0,
|
||
remote_written=remote_success > 0,
|
||
)
|
||
|
||
async def _finalize_peer_creating_sources(self, node_type: str) -> None:
|
||
finalized_success, finalized_failed = await finalize_peer_creating_sources(
|
||
node_type=node_type,
|
||
local_collection=self.local_collection,
|
||
remote_collection=self.remote_collection,
|
||
binding_manager=self.binding_manager,
|
||
runtime=self.sm_runtime,
|
||
)
|
||
|
||
if finalized_success or finalized_failed:
|
||
self._logger.info(
|
||
f"🔁 Peer-creating finalize: node_type={node_type}, success={finalized_success}, failed={finalized_failed}"
|
||
)
|
||
|
||
async def normalize_create_success_to_update_entry(self, node_type: str) -> None:
|
||
"""将 CREATE 成功节点从 S11C 归并到 S01，作为 update 起点。"""
|
||
normalized = 0
|
||
for collection in (self.local_collection, self.remote_collection):
|
||
candidates = collection.filter_by_state_ids(
|
||
node_type=node_type,
|
||
state_ids=["S11C"],
|
||
)
|
||
for node in candidates:
|
||
if node.action.value != "CREATE":
|
||
continue
|
||
decision = e41_post_create_ready(self.sm_runtime, node=node, execute_action="CREATE")
|
||
if decision.to_state != "S01":
|
||
raise RuntimeError(
|
||
f"[{node_type}] E41 expected S01, got {decision.to_state} for node={node.node_id}"
|
||
)
|
||
normalized += 1
|
||
|
||
if normalized > 0:
|
||
self._logger.info(f"🔁 Post-create normalization: {normalized} nodes moved S11C->S01 for {node_type}")
|
||
|
||
async def commit_updates(self, node_type: str) -> WriteOutcome:
|
||
"""提交 UPDATE 操作（包含异步轮询）"""
|
||
# 执行同步（sync_all 内部会根据 action 筛选节点）
|
||
await self.local_datasource.sync_all(
|
||
order=[node_type],
|
||
poll_async_tasks=True
|
||
)
|
||
|
||
await self.remote_datasource.sync_all(
|
||
order=[node_type],
|
||
poll_async_tasks=True
|
||
)
|
||
|
||
# 更新统计
|
||
local_nodes = self.local_collection.filter(node_type=node_type)
|
||
remote_nodes = self.remote_collection.filter(node_type=node_type)
|
||
self.stats["synced"] += len([n for n in local_nodes if n.action.value == "UPDATE"])
|
||
self.stats["synced"] += len([n for n in remote_nodes if n.action.value == "UPDATE"])
|
||
self.stats["failed"] += len([n for n in local_nodes if n.status.value == "FAILED"])
|
||
self.stats["failed"] += len([n for n in remote_nodes if n.status.value == "FAILED"])
|
||
|
||
local_success = self._get_action_success_count(self.local_datasource, node_type, "update")
|
||
remote_success = self._get_action_success_count(self.remote_datasource, node_type, "update")
|
||
return WriteOutcome(
|
||
local_written=local_success > 0,
|
||
remote_written=remote_success > 0,
|
||
)
|
||
|
||
async def phase_post_check(self) -> bool:
|
||
"""同步结果报告。
|
||
|
||
所有类型均已同步完毕后，统一做一次全量 reload（触发后端派生字段计算），
|
||
再对所有类型做一致性评估，确保 post-check 结果反映最终状态。
|
||
"""
|
||
if self._should_skip_reload():
|
||
self._logger.info("⏭️ Post-check reload skipped for JSONL pipeline")
|
||
else:
|
||
self._logger.info("🔄 Post-check: reloading all node types for final consistency evaluation...")
|
||
for node_type in self.node_types:
|
||
strategy = self._get_strategy(node_type)
|
||
if strategy is not None and strategy.skip_post_check:
|
||
self._logger.info("⏭️ Post-check reload skipped: node_type=%s", node_type)
|
||
continue
|
||
await self._reload_node_type(node_type, reason="post-check final reload")
|
||
for node_type in self.node_types:
|
||
strategy = self._get_strategy(node_type)
|
||
if strategy is not None and strategy.skip_post_check:
|
||
self._logger.info("⏭️ Post-check skipped: node_type=%s", node_type)
|
||
self._consistency_by_type[node_type] = {
|
||
"ok": True,
|
||
"checked_pairs": 0,
|
||
"mismatch_count": 0,
|
||
"skipped": True,
|
||
}
|
||
continue
|
||
await self._evaluate_consistency_for_node_type(node_type)
|
||
|
||
self.stats["consistency"] = {
|
||
node_type: {
|
||
"ok": item.get("ok", True),
|
||
"checked_pairs": item.get("checked_pairs", 0),
|
||
"mismatch_count": item.get("mismatch_count", 0),
|
||
}
|
||
for node_type, item in self._consistency_by_type.items()
|
||
}
|
||
await self.print_summary()
|
||
return True # 总是返回 True，不阻断流程
|
||
|
||
async def phase_persistence(self) -> None:
|
||
"""持久化所有数据。"""
|
||
if not self.enable_persistence:
|
||
self._logger.info("⏭️ Persistence disabled, skipping...")
|
||
return
|
||
|
||
self._logger.info("🔍 Persisting local collection...")
|
||
await self.local_collection.persist(exclude_node_types=self.ephemeral_node_types)
|
||
self._logger.info("🔍 Persisting remote collection...")
|
||
await self.remote_collection.persist(exclude_node_types=self.ephemeral_node_types)
|
||
self._logger.info("🔍 Persisting binding manager...")
|
||
await self.binding_manager.persist(exclude_node_types=self.ephemeral_node_types)
|
||
self._logger.info("✅ All data persisted successfully")
|
||
|
||
# ========== Summary & Reporting ==========
|
||
|
||
async def print_summary(self) -> None:
|
||
"""打印同步结果摘要"""
|
||
await print_pipeline_summary(
|
||
logger=self._logger,
|
||
node_types=self.node_types,
|
||
binding_manager=self.binding_manager,
|
||
local_datasource=self.local_datasource,
|
||
remote_datasource=self.remote_datasource,
|
||
local_collection=self.local_collection,
|
||
remote_collection=self.remote_collection,
|
||
consistency_by_type=self._consistency_by_type,
|
||
stats=self.stats,
|
||
)
|