""" Full Sync Pipeline - Implements document-specified end-to-end workflow. """ from __future__ import annotations import asyncio import logging from dataclasses import dataclass from typing import Any, Dict, List, Optional, TYPE_CHECKING from pathlib import Path 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.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, 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__) @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)) async def run(self) -> Dict[str, Any]: try: await self.phase_bootstrap() self._logger.info("✅ Phase bootstrap completed (%s)", self.pipeline_name) await self.phase_sync_by_node_type() self._logger.info("✅ Phase sync-by-node-type completed (%s)", self.pipeline_name) post_ok = await self.phase_post_check() self.stats["post_check_passed"] = post_ok self._logger.info("✅ Phase post-check completed (%s)", self.pipeline_name) await self.phase_persistence() self._logger.info("✅ Phase persistence completed (%s)", self.pipeline_name) return self.stats finally: await self.close() async def phase_sync_by_node_type(self) -> None: self._prepare_datasources() 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 self._load_node_type(node_type, reason="pre-strategy refresh") 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)})") 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._logger.info(f"✅ Strategy update: {node_type} (updated={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._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 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._apply_project_scope_cleanup() async def phase_bind(self) -> None: self._prepare_datasources() 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: self._prepare_datasources() 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)})") 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._logger.error(f"❌ Strategy create failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") continue async def phase_update(self) -> None: self._prepare_datasources() 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_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._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 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)") # 加载后 reset 清理：清理 CREATE 失败的僵尸绑定 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} CREATE-failed zombies cleaned") self._logger.info("🔄 Post-load reset cleanup completed") 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.local_datasource.load_all(order=[node_type]) await self.remote_datasource.load_all(order=[node_type]) 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)) 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._logger.info("⏭️ Reload skipped for JSONL pipeline: node_type=%s reason=%s", 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._logger.info("⏭️ Reload skipped for JSONL pipeline: node_type=%s reason=%s", 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.local_datasource.load_all(order=[node_type]) if reload_remote: 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) 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 = next((item for item in self.strategies if item.node_type == node_type), None) 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 = next((item for item in self.strategies if item.node_type == node_type), None) 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, )