""" Full Sync Pipeline - Implements document-specified end-to-end workflow. """ from __future__ import annotations import asyncio import logging from typing import Any, Dict, List, Optional, TYPE_CHECKING from pathlib import Path if TYPE_CHECKING: from ..datasource.datasource import BaseDataSource from ..common.collection import DataCollection from ..common.binding import BindingManager from ..common.persistence import PersistenceBackend from ..sync_system.strategy import BaseSyncStrategy from ..sync_system.config import OrphanAction, UpdateDirection from ..engine import StateMachineConfig, StateMachineRuntime from ..engine import e41_post_create_ready from ..sync_system.strategy_ops import finalize_peer_creating_sources from ..sync_system.strategy_ops.post_check_ops import evaluate_consistency_for_node_type from .summary_report import print_pipeline_summary logger = logging.getLogger(__name__) 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, node_types: Optional[List[str]] = None, load_order: Optional[List[str]] = None, sync_order: Optional[List[str]] = None, enable_persistence: bool = True, # 项目过滤 target_project_ids: Optional[List[str]] = None, local_target_project_ids: Optional[List[str]] = None, remote_target_project_ids: Optional[List[str]] = None, ): # 验证必需参数（由工厂函数创建） 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.node_types = node_types or [s.node_type for s in self.strategies] 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.target_project_ids = target_project_ids or [] self.local_target_project_ids = ( list(local_target_project_ids) if local_target_project_ids is not None else list(self.target_project_ids) ) self.remote_target_project_ids = ( list(remote_target_project_ids) if remote_target_project_ids is not None else list(self.target_project_ids) ) 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 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)) async def run(self) -> Dict[str, Any]: try: await self.phase_bootstrap() self._logger.info("✅ Phase bootstrap completed") await self.phase_sync_by_node_type() self._logger.info("✅ Phase sync-by-node-type completed") post_ok = await self.phase_post_check() self.stats["post_check_passed"] = post_ok self._logger.info("✅ Phase post-check completed") await self.phase_persistence() self._logger.info("✅ Phase persistence completed") return self.stats finally: await self.close() async def phase_sync_by_node_type(self) -> None: strategy_map = {s.node_type: s for s in self.strategies} for node_type in self.sync_order: strategy = strategy_map.get(node_type) if strategy is None: continue try: section_width = 96 self._logger.info("\n" + "-" * section_width) self._logger.info(f"🧩 Strategy start: {node_type}") self._logger.info("-" * section_width) await strategy.bind() self._logger.info(f"✅ Strategy bind: {node_type}") if strategy.skip_sync: self._logger.info(f"⏭️ Skipping create/update for {node_type} (skip_sync=True)") continue created = await strategy.create() self._logger.info(f"✅ Strategy create: {node_type} (created={len(created)})") create_written = await self.commit_creates(node_type) await self.normalize_create_success_to_update_entry(node_type) if create_written: await self._reload_node_type(node_type, reason="create wrote data") updated = await strategy.update() self._logger.info(f"✅ Strategy update: {node_type} (updated={len(updated)})") update_written = await self.commit_updates(node_type) if update_written: await self._reload_node_type(node_type, reason="update wrote data") self._logger.info(f"✅ Strategy done: {node_type}") except Exception as exc: self.stats["failed"] += 1 self._logger.error(f"❌ Strategy failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") continue @staticmethod def _is_noop_strategy(strategy: BaseSyncStrategy[Any]) -> bool: cfg = strategy.config return ( cfg.local_orphan_action == OrphanAction.NONE and cfg.remote_orphan_action == OrphanAction.NONE and cfg.update_direction == UpdateDirection.NONE ) async def phase_bootstrap(self) -> None: await self._load_persistence_state() self._logger.info("✅ Bootstrap: persistence loaded") await self._load_from_datasource() self._logger.info("✅ Bootstrap: datasource fetched") if self.local_target_project_ids or self.remote_target_project_ids: local_deleted = await self.local_collection.filter_by_project_ids(self.local_target_project_ids) remote_deleted = await self.remote_collection.filter_by_project_ids(self.remote_target_project_ids) if local_deleted > 0 or remote_deleted > 0: self._logger.info(f"🔍 Project filtering: deleted {local_deleted} local, {remote_deleted} remote nodes") async def phase_bind(self) -> None: strategy_map = {s.node_type: s for s in self.strategies} for node_type in self.sync_order: strategy = strategy_map.get(node_type) if strategy is None: continue try: await strategy.bind() except Exception as exc: self.stats["failed"] += 1 self._logger.error(f"❌ Strategy bind failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") continue async def phase_create(self) -> None: strategy_map = {s.node_type: s for s in self.strategies} for node_type in self.sync_order: strategy = strategy_map.get(node_type) if strategy is None: continue if strategy.skip_sync: self._logger.info(f"⏭️ Skipping create for {node_type} (skip_sync=True)") continue try: created = await strategy.create() self._logger.info(f"✅ Strategy create: {node_type} (created={len(created)})") create_written = await self.commit_creates(node_type) await self.normalize_create_success_to_update_entry(node_type) if create_written: await self._reload_node_type(node_type, reason="create wrote data") except Exception as exc: self.stats["failed"] += 1 self._logger.error(f"❌ Strategy create failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") continue async def phase_update(self) -> None: strategy_map = {s.node_type: s for s in self.strategies} for node_type in self.sync_order: strategy = strategy_map.get(node_type) if strategy is None: continue if strategy.skip_sync: self._logger.info(f"⏭️ Skipping update for {node_type} (skip_sync=True)") continue try: updated = await strategy.update() self._logger.info(f"✅ Strategy update: {node_type} (updated={len(updated)})") update_written = await self.commit_updates(node_type) if update_written: await self._reload_node_type(node_type, reason="update wrote data") await self._evaluate_consistency_for_node_type(node_type) except Exception as exc: self.stats["failed"] += 1 self._logger.error(f"❌ Strategy update failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") continue async def close(self) -> None: """关闭所有资源（DataSource、Persistence）""" if self._closed: return # 已经关闭过，避免重复关闭 self._closed = True async def _safe_close(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) await _safe_close("remote_datasource", self.remote_datasource.close()) await _safe_close("local_datasource", self.local_datasource.close()) await _safe_close("persistence", self.persistence.close()) async def _load_persistence_state(self) -> None: """加载持久化数据并清理僵尸节点""" if not self.enable_persistence: return # 加载持久化数据 await self.local_collection.load_from_persistence() await self.remote_collection.load_from_persistence() for node_type in self.node_types: await self.binding_manager.load_from_persistence(node_type) self._logger.info("🔄 Persistence loaded (state preserved; pending E01 normalization)") # 加载后 reset 清理：清理 CREATE 失败的僵尸绑定 if not self.strategies: raise RuntimeError("no strategies configured: cannot run bootstrap event E01") total_cleaned = await BaseSyncStrategy.run_reset( 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} CREATE-failed zombies cleaned") self._logger.info("🔄 Post-load reset cleanup completed") async def _load_from_datasource(self) -> None: """从数据源加载数据""" self.local_datasource.set_collection(self.local_collection) await self.local_datasource.load_all(order=self.load_order) self.remote_datasource.set_collection(self.remote_collection) await self.remote_datasource.load_all(order=self.load_order) for node_type in self.node_types: self.stats["loaded"] += len(self.local_collection.filter(node_type=node_type)) self.stats["loaded"] += len(self.remote_collection.filter(node_type=node_type)) 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 async def _reload_node_type(self, node_type: str, *, reason: str) -> None: self._logger.info(f"🔄 Reload after write: node_type={node_type}, reason={reason}") await self.local_datasource.load_all(order=[node_type]) await self.remote_datasource.load_all(order=[node_type]) 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) 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_list_item_fields = dict(strategy.config.post_check_ignore_list_item_fields) if strategy else {} # 从远端 handler 的 update_schemas 提取只读字段过滤白名单 post_check_fields: List[str] | None = None try: remote_handler = self.remote_datasource.get_handler(node_type) fields = remote_handler.get_update_fields() if fields: post_check_fields = fields except Exception: pass 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, depend_fields=depend_fields, compare_to_remote=compare_to_remote, ignore_list_item_fields=ignore_list_item_fields, post_check_fields=post_check_fields, ) self._consistency_by_type[node_type] = result return result async def commit_creates(self, node_type: str) -> bool: """提交 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") wrote = (local_success + remote_success) > 0 return wrote 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) -> bool: """提交 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") wrote = (local_success + remote_success) > 0 return wrote async def phase_post_check(self) -> bool: """同步结果报告。 所有类型均已同步完毕后，统一做一次全量 reload（触发后端派生字段计算）， 再对所有类型做一致性评估，确保 post-check 结果反映最终状态。 """ self._logger.info("🔄 Post-check: reloading all node types for final consistency evaluation...") for node_type in self.node_types: await self._reload_node_type(node_type, reason="post-check final reload") for node_type in self.node_types: 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() self._logger.info("🔍 Persisting remote collection...") await self.remote_collection.persist() self._logger.info("🔍 Persisting binding manager...") await self.binding_manager.persist() 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, )