CAP 三角：

        C (Consistency)
       / \
      /   \
     /     \
    /  选择  \
   /   区域   \
  A --------- P
(Availability) (Partition Tolerance)

智策平台的选择：AP + 最终一致性
  - 保证可用性和分区容错性
  - 通过事件驱动实现最终一致性
  - 在关键路径上提供强一致性选项

一致性需求矩阵：

场景                        一致性要求    原因
──────────────────────────────────────────────────────
Ontology Schema 变更        强一致性      Schema 变更影响全局
World 创建/删除             强一致性      World 是隔离边界
用户认证/授权               强一致性      安全相关不能有窗口期
对象实例 CRUD               最终一致性    可接受短暂不一致
派生属性重算                最终一致性    可接受短暂过时值
规则评估触发                最终一致性    可接受毫秒级延迟
World 分支间数据            快照隔离      分支间完全隔离
审计日志                    最终一致性    只要不丢就行

数据所有权：

onto-control (Control Layer)：
  ├── Ontology Schema（ObjectType, LinkType, ActionType）
  ├── World 元数据（World, Branch, Tag）
  ├── 用户和权限（User, Role, Permission）
  └── 数据源：PostgreSQL（强一致性）

onto-data (Data Layer)：
  ├── 对象实例（Object instances）
  ├── 关系实例（Link instances）
  ├── 时序数据（Time series）
  └── 数据源：Doris + Iceberg/Nessie（最终一致性）

onto-intelligence (Reasoning & Decision Layer + Agent Runtime Layer)：
  ├── 规则定义和评估结果
  ├── 决策定义和执行结果
  ├── 派生属性定义和计算值
  └── 数据源：PostgreSQL + Redis（混合）

跨进程数据一致性 = 事件驱动 + 补偿机制

Nessie 版本树：

main (World: prod)
  │
  ├── commit-001: 初始化 ObjectTypes
  ├── commit-002: 添加 InventoryRecord
  ├── commit-003: 库存数据同步
  │
  ├──── branch: staging (World: staging)
  │     ├── commit-004: 新增 SupplierScore 属性
  │     └── commit-005: 修改评分计算逻辑
  │
  └──── branch: dev-feature-x (World: dev)
        ├── commit-006: 实验性 ObjectType
        └── commit-007: 测试数据

# Nessie 乐观并发控制示例
class NessieOCCWriter:
    """使用 Nessie OCC 进行并发安全写入"""

    def __init__(self, nessie_client):
        self.nessie = nessie_client

    async def write_with_occ(
        self,
        branch: str,
        table_id: str,
        data: Any,
        max_retries: int = 3,
    ):
        """乐观并发写入"""
        for attempt in range(max_retries):
            try:
                # 1. 读取当前分支的 HEAD hash
                branch_ref = await self.nessie.get_reference(branch)
                current_hash = branch_ref.hash

                # 2. 执行写入操作
                # (此时其他 writer 可能也在写)
                operations = self._prepare_operations(
                    table_id, data
                )

                # 3. 提交时带上 expected_hash
                # 如果 hash 已变（其他人先提交了），提交失败
                await self.nessie.commit(
                    branch=branch,
                    operations=operations,
                    expected_hash=current_hash,  # OCC 关键
                    message=f"Update {table_id}",
                )

                return  # 成功

            except NessieConflictException:
                # 4. 冲突：其他 writer 在我们之前提交了
                if attempt < max_retries - 1:
                    # 重试：重新读取最新 hash，重新准备操作
                    await asyncio.sleep(0.1 * (2 ** attempt))
                    continue
                else:
                    raise ConcurrencyConflictError(
                        f"Failed after {max_retries} retries "
                        f"on branch {branch}"
                    )

为什么选择 OCC 而不是悲观锁：

场景                    OCC              悲观锁
──────────────────────────────────────────────────
低冲突（正常情况）      极快，无锁等待     需要获取/释放锁
高冲突（罕见）          重试开销           阻塞等待
死锁风险                无                 有
分布式复杂度            低                 高（分布式锁）
吞吐量                  高                 受限于锁粒度

智策平台场景分析：
  - 不同 World 的写入完全隔离 → 无冲突
  - 同一 World 内并发写入 → 低冲突（不同对象）
  - 同一对象并发写入 → 极低概率

结论：OCC 是最佳选择

Kafka 顺序性保证：

保证 1：同一 Partition 内消息严格有序
  Partition 0: [msg-1] → [msg-2] → [msg-3]  ← 严格有序

保证 2：同一 Key 的消息路由到同一 Partition
  Key = "WH-001::PROD-ABC"
  → hash("WH-001::PROD-ABC") % num_partitions = Partition 3
  → 该实体的所有变更事件都在 Partition 3

保证 3：Consumer 按 Partition 顺序消费
  Consumer 读 Partition 3: [evt-1] → [evt-2] → [evt-3]

分区键设计策略（每个 Topic 不同）：

Topic                   分区键                    原因
──────────────────────────────────────────────────────────
cdc.raw.*               source_pk                 同一实体变更有序
onto.changes.objects    world_id + object_pk      同一世界同一对象有序
onto.changes.links      world_id + link_pk        同一世界同一关系有序
domain.alerts           world_id + alert_type     同类告警有序
domain.decisions        world_id + decision_id    同一决策有序
domain.actions          world_id + action_id      同一 Action 有序

跨实体因果关系问题：

场景：Object A 的变更触发 Object B 的变更
  - A 的变更在 Partition 1
  - B 的变更在 Partition 5
  - 消费者 X 先处理 Partition 5（B 的变更）
  - 消费者 Y 后处理 Partition 1（A 的变更）
  - 结果：B 在 A 之前被处理！

解决方案：事件中携带因果链信息

{
  "event_id": "evt-002",
  "caused_by": "evt-001",     // 因果关系标记
  "causal_order": 2,          // 因果序号
  "world_id": "world-prod"
}

处理逻辑：
  if event.caused_by is not None:
    # 检查前因事件是否已处理
    if not is_processed(event.caused_by):
      # 推迟处理，等待前因事件
      defer(event, retry_after=100ms)
    else:
      process(event)

class IdempotentConsumer:
    """幂等消费者"""

    def __init__(self, processed_store):
        self.processed = processed_store  # Redis Set

    async def consume(self, event: Event):
        # 1. 检查是否已处理过
        event_key = f"processed:{event.event_id}"
        if await self.processed.exists(event_key):
            logger.info(f"Skipping duplicate: {event.event_id}")
            return

        # 2. 处理事件
        await self._process(event)

        # 3. 标记为已处理（TTL = 7天）
        await self.processed.set(
            event_key, "1", ex=7 * 24 * 3600
        )

    async def _process(self, event: Event):
        """实际处理逻辑（必须是幂等的）"""
        # 使用 upsert 而不是 insert
        # 使用 conditional update 而不是 blind update
        await self.store.upsert(
            key=event.object_pk,
            value=event.after_properties,
            version=event.version,  # 版本号检查
        )

World 隔离架构：

World: prod (main branch)
  ┌────────────────────────────────────────┐
  │ Nessie Branch: main                    │
  │ Doris Database: world_prod             │
  │ Kafka Consumer Group: cg-world-prod    │
  │                                        │
  │ ObjectTypes: [完整的生产 Schema]        │
  │ Objects: [生产数据]                     │
  │ Rules: [生产规则]                       │
  │ Decisions: [生产决策]                   │
  └────────────────────────────────────────┘

World: staging (staging branch)
  ┌────────────────────────────────────────┐
  │ Nessie Branch: staging                 │
  │ Doris Database: world_staging          │
  │ Kafka Consumer Group: cg-world-staging │
  │                                        │
  │ ObjectTypes: [staging Schema]           │
  │ Objects: [staging 数据]                 │
  │ Rules: [测试中的新规则]                 │
  │ Decisions: [测试中的新决策]             │
  └────────────────────────────────────────┘

隔离保证：
  - 不同 World 的数据完全隔离
  - 修改 staging 不影响 prod
  - 只有通过 Merge 操作才能将变更从一个 World 传播到另一个

Saga 模式（以创建对象实例为例）：

Step 1: onto-control — 验证 ObjectType 存在
  成功 → Step 2
  失败 → 返回错误

Step 2: onto-data — 写入 Doris
  成功 → Step 3
  失败 → 无需补偿（Step 1 无副作用）

Step 3: onto-data — 写入 Iceberg
  成功 → Step 4
  失败 → 补偿 Step 2（从 Doris 删除）

Step 4: onto-data — 发布变更事件到 Kafka
  成功 → 完成
  失败 → 补偿 Step 3（从 Iceberg 删除）
         补偿 Step 2（从 Doris 删除）

class ObjectCreateSaga:
    """对象创建 Saga"""

    async def execute(self, request: CreateObjectRequest):
        compensations = []

        try:
            # Step 1: 验证 Schema
            await self.control_client.validate_object_type(
                request.world_id, request.object_type
            )

            # Step 2: 写入 Doris
            doris_result = await self.doris_writer.insert(
                request.world_id, request.object_type,
                request.properties
            )
            compensations.append(
                lambda: self.doris_writer.delete(
                    request.world_id, doris_result.pk
                )
            )

            # Step 3: 写入 Iceberg
            iceberg_result = await self.iceberg_writer.append(
                request.world_id, request.object_type,
                request.properties
            )
            compensations.append(
                lambda: self.iceberg_writer.delete(
                    request.world_id, iceberg_result.file_id
                )
            )

            # Step 4: 发布事件
            await self.event_publisher.publish(
                "onto.changes.objects",
                ObjectChangeEvent(
                    world_id=request.world_id,
                    object_type=request.object_type,
                    change_type="CREATE",
                    after_properties=request.properties,
                )
            )

            return doris_result

        except Exception as e:
            # 逆序执行补偿
            for compensation in reversed(compensations):
                try:
                    await compensation()
                except Exception as comp_error:
                    logger.error(
                        f"Compensation failed: {comp_error}"
                    )
            raise

三路合并（Three-Way Merge）：

找到共同祖先（Merge Base），比较三个版本：

  Merge Base (分叉点)
       │
  ┌────┴────┐
  │         │
  ▼         ▼
Source    Target
(staging) (prod)

对于每个对象/属性：
  Base    Source  Target  → 操作
  ─────────────────────────────────
  A=1     A=1    A=1     → 无变更（三方相同）
  A=1     A=2    A=1     → 取 Source（Source 修改了）
  A=1     A=1    A=3     → 取 Target（Target 修改了）
  A=1     A=2    A=3     → 冲突！需要解决
  -       A=2    -       → 取 Source（Source 新增）
  A=1     -      A=1     → 删除（Source 删除了）
  A=1     A=2    -       → 冲突！一边修改一边删除

class WorldMergeService:
    """World 三路合并服务"""

    async def merge(
        self,
        source_world: str,
        target_world: str,
        merge_strategy: str = "AUTO",
    ) -> MergeResult:

        # 1. 找到 Merge Base
        base_ref = await self.nessie.find_merge_base(
            source_world, target_world
        )

        # 2. 获取三个版本的数据快照
        base_snapshot = await self._get_snapshot(base_ref)
        source_snapshot = await self._get_snapshot(source_world)
        target_snapshot = await self._get_snapshot(target_world)

        # 3. 逐对象比较
        conflicts = []
        operations = []

        all_keys = set(
            list(base_snapshot.keys()) +
            list(source_snapshot.keys()) +
            list(target_snapshot.keys())
        )

        for key in all_keys:
            base_val = base_snapshot.get(key)
            source_val = source_snapshot.get(key)
            target_val = target_snapshot.get(key)

            result = self._three_way_compare(
                base_val, source_val, target_val
            )

            if result.is_conflict:
                conflicts.append(Conflict(
                    key=key,
                    base=base_val,
                    source=source_val,
                    target=target_val,
                    conflict_type=result.conflict_type,
                ))
            elif result.has_change:
                operations.append(result.operation)

        # 4. 处理冲突
        if conflicts:
            if merge_strategy == "AUTO":
                # 自动解决：Source 优先
                for conflict in conflicts:
                    operations.append(
                        MergeOperation(
                            key=conflict.key,
                            value=conflict.source,
                            resolution="SOURCE_WINS",
                        )
                    )
            elif merge_strategy == "MANUAL":
                # 返回冲突列表，等待用户解决
                return MergeResult(
                    status="CONFLICTS",
                    conflicts=conflicts,
                    merge_id=generate_merge_id(),
                )

        # 5. 应用合并操作到 Target
        await self._apply_operations(
            target_world, operations
        )

        return MergeResult(
            status="MERGED",
            operations_applied=len(operations),
            conflicts_resolved=len(conflicts),
        )

冲突解决策略：

策略 1: SOURCE_WINS（源优先）
  - Source 的值覆盖 Target
  - 适用：staging → prod 的部署

策略 2: TARGET_WINS（目标优先）
  - Target 的值保留
  - 适用：prod 数据不能被覆盖的场景

策略 3: MANUAL（手动解决）
  - 返回冲突列表
  - 用户在 UI 上逐个解决
  - 适用：重要的合并操作

策略 4: TIMESTAMP_WINS（最新优先）
  - 取 updated_at 更新的值
  - 适用：数据同步场景

冲突类型：
  ┌─────────────────────────────────────────┐
  │ MODIFY_MODIFY: 双方都修改了同一属性      │
  │ MODIFY_DELETE: 一方修改，另一方删除       │
  │ SCHEMA_CONFLICT: ObjectType 定义冲突     │
  │ RULE_CONFLICT: 规则定义冲突              │
  └─────────────────────────────────────────┘

三个进程间的一致性保证（无分布式事务）：

场景：修改 ObjectType 属性

1. onto-control 更新 Schema（PostgreSQL 事务）
   └── 发布事件: schema.updated

2. onto-data 收到事件，更新存储层映射
   └── Doris 表结构变更
   └── Iceberg Schema Evolution
   └── 发布事件: storage.schema.updated

3. onto-intelligence 收到事件，更新派生属性依赖
   └── 更新 DAG 图
   └── 触发受影响的派生属性重算

一致性窗口：
  Step 1 完成 → Step 2 完成: ~200ms
  Step 2 完成 → Step 3 完成: ~500ms

在这个窗口内：
  - 新 Schema 已定义，但存储层还在更新
  - 查询可能返回旧 Schema 的数据
  - 这是可接受的最终一致性

class VersionVector:
    """版本向量 - 检测跨进程一致性"""

    def __init__(self):
        self.versions = {
            "control": 0,   # onto-control 的版本
            "data": 0,      # onto-data 的版本
            "intelligence": 0,  # onto-intelligence 的版本
        }

    def increment(self, process: str):
        self.versions[process] += 1

    def is_consistent(self) -> bool:
        """检查三个进程是否一致"""
        # 所有进程的版本应该相同或差距 <= 1
        values = list(self.versions.values())
        return max(values) - min(values) <= 1

    def get_lagging_process(self) -> str | None:
        """找出落后的进程"""
        max_ver = max(self.versions.values())
        for process, ver in self.versions.items():
            if ver < max_ver:
                return process
        return None

定期一致性检查（每 5 分钟）：

1. 检查 Schema 一致性：
   control_schema = onto-control.get_schema(world_id)
   data_schema = onto-data.get_schema(world_id)
   if control_schema.version != data_schema.version:
     trigger_schema_sync(world_id)

2. 检查对象计数一致性：
   control_count = onto-control.get_object_count(world_id, type)
   data_count = onto-data.get_object_count(world_id, type)
   if abs(control_count - data_count) > threshold:
     trigger_reconciliation(world_id, type)

3. 检查派生属性新鲜度：
   for derived_prop in all_derived_properties:
     last_calc_time = get_last_calculation_time(derived_prop)
     if now() - last_calc_time > max_staleness:
       trigger_recalculation(derived_prop)

一致性选择矩阵：

                    读延迟要求
                    低(< 10ms)    中(< 100ms)   高(< 1s)
写入频率 ──────────────────────────────────────────────
  低     │ 强一致性       强一致性        最终一致性
(< 10/s) │ (直接查主库)   (直接查主库)    (事件驱动)
         │
  中     │ 最终一致性     最终一致性       最终一致性
(< 100/s)│ (缓存+失效)   (事件驱动)      (事件驱动)
         │
  高     │ 最终一致性     最终一致性       最终一致性
(> 100/s)│ (CQRS)       (CQRS)          (批量处理)

智策平台各场景的位置：
  Ontology Schema:   低写入 × 低延迟 → 强一致性
  对象实例查询:       中写入 × 低延迟 → 最终一致性(缓存)
  派生属性:          中写入 × 中延迟 → 最终一致性(事件)
  审计日志:          高写入 × 高延迟 → 最终一致性(批量)

CQRS（Command Query Responsibility Segregation）：

写入路径（Command）：
  Client → Gateway → onto-data → Doris (写入)
                                   ↓
                              Kafka Event
                                   ↓
                              onto-data → 更新读模型

读取路径（Query）：
  Client → Gateway → onto-data → Doris (读取)
                                   ↑
                              读模型（已物化的视图）

好处：
  - 写入和读取可以独立优化
  - 读取路径可以使用物化视图
  - 写入不会阻塞读取

Action 可能被重复执行的场景：

1. Temporal Worker 崩溃重启 → Activity 重试
2. 网络超时 → 客户端不知道是否成功 → 重试
3. Kafka 消息重复消费 → 触发重复 Action
4. 手动重放 → 运维人员重新触发

如果 Action 不幂等：
  创建采购订单 × 3次 = 3个重复订单 = 财务灾难

class IdempotentActionExecutor:
    """幂等 Action 执行器"""

    def __init__(self, idempotency_store):
        self.store = idempotency_store  # Redis

    async def execute(
        self,
        action_id: str,
        idempotency_key: str,
        action_fn,
        params: dict,
    ):
        # 1. 检查是否已有执行结果
        existing = await self.store.get(idempotency_key)
        if existing:
            logger.info(
                f"Action {action_id} already executed, "
                f"returning cached result"
            )
            return existing

        # 2. 获取执行锁（防止并发重复执行）
        lock = await self.store.acquire_lock(
            f"lock:{idempotency_key}",
            timeout=60,
        )

        if not lock:
            # 其他实例正在执行，等待结果
            return await self._wait_for_result(
                idempotency_key
            )

        try:
            # 3. 执行 Action
            result = await action_fn(**params)

            # 4. 存储结果（TTL = 7天）
            await self.store.set(
                idempotency_key,
                result,
                ex=7 * 24 * 3600,
            )

            return result

        finally:
            await self.store.release_lock(lock)

    def generate_idempotency_key(
        self, action_type: str, params: dict
    ) -> str:
        """生成幂等键"""
        # 基于 Action 类型和核心参数生成确定性 key
        key_parts = [
            action_type,
            params.get("world_id", ""),
            params.get("object_type", ""),
            params.get("object_pk", ""),
        ]
        content = "|".join(key_parts)
        return hashlib.sha256(content.encode()).hexdigest()

一致性监控仪表盘：

指标 1: 事件处理延迟（Event Processing Lag）
  kafka_consumer_lag{topic="onto.changes.*"}
  告警：lag > 1000 条 → WARNING
  告警：lag > 10000 条 → CRITICAL

指标 2: 跨进程版本差异
  version_vector_diff{process_pair="control-data"}
  告警：diff > 5 → WARNING

指标 3: 派生属性新鲜度
  derived_property_staleness_seconds
  告警：staleness > 60s → WARNING

指标 4: 合并冲突率
  world_merge_conflict_rate
  告警：rate > 10% → 需要检查工作流

指标 5: 幂等键命中率
  idempotency_cache_hit_rate
  正常：< 1%（偶尔重试）
  告警：> 10%（系统可能有问题）

一致性修复工具集：

1. schema-reconcile: 强制同步 Schema
   $ onto-admin schema-reconcile --world prod

2. object-recount: 重新计算对象计数
   $ onto-admin object-recount --world prod --type InventoryRecord

3. derived-recalc: 强制重算派生属性
   $ onto-admin derived-recalc --world prod --property stockLevel

4. event-replay: 重放 Kafka 事件
   $ onto-admin event-replay --topic onto.changes.objects \
       --from-offset 12345 --to-offset 12400

5. merge-verify: 验证合并结果
   $ onto-admin merge-verify --merge-id merge-001