用户 A (北京)          用户 B (纽约)
    │                       │
    ▼                       ▼
 写入 X=1                写入 X=2
    │                       │
    ▼                       ▼
  节点 1                 节点 2
    │                       │
    └──────────┬────────────┘
               ▼
          数据同步
               ▼
        X 的最终值是？

节点 A 时间: 12:00:00.000
节点 B 时间: 12:00:00.500  (快 500ms)

用户操作:
  T+0ms   (A时间)  用户写入数据
  T+100ms (B时间)  用户读取数据

问题：读取操作"看起来"发生在写入之前！

漂移率：~10^-6 (1 微秒/秒)

一天内的累积误差：
  1 μs/s × 86400 s = 86.4 ms

一个月内的累积误差：
  86.4 ms × 30 ≈ 2.6 秒

NTP 同步精度：
- 局域网：~1 ms
- 广域网：~10-100 ms

问题场景：
  节点 A 时间：12:00:00.000
  节点 B 时间：12:00:00.050 (通过NTP同步)
  
  事件 1 (A): 12:00:00.010
  事件 2 (B): 12:00:00.040
  
  实际：事件 2 比事件 1 晚 30ms
  但时钟显示：事件 2 比事件 1 早 20ms！

→ 是分布式事件集合上的最小关系，满足：

1. 本地顺序：
   ∀e, f ∈ Events(process_i): index(e) < index(f) ⟹ e → f

2. 消息顺序：
   ∀send(m), receive(m): send(m) → receive(m)

3. 传递性：
   ∀a, b, c: a → b ∧ b → c ⟹ a → c

进程 P1          进程 P2
   │                │
   ▼                ▼
  A: x=1          C: y=2
   │                │
   ▼                ▼
  B: send(m)      D: receive(m)
   │                │
   ▼                ▼
  E: x=2          F: y=3

Happens-Before 关系：
  A → B → D → F
  C → D (通过消息)
  A → E
  C → F

并发关系：
  A || C (A不影响C，C不影响A)
  E || F (E不影响F，F不影响E)

偏序关系 ≤ 满足：
1. 自反性：∀a: a ≤ a
2. 反对称性：a ≤ b ∧ b ≤ a ⟹ a = b
3. 传递性：a ≤ b ∧ b ≤ c ⟹ a ≤ c

注意：不要求完备性（不是所有元素都可比较）

全序关系 ≤ 满足偏序的所有性质，加上：
4. 完备性：∀a, b: a ≤ b ∨ b ≤ a

任意两个元素都可比较

class LamportClock:
    """
    Lamport 逻辑时钟实现
    """
    
    def __init__(self, process_id: str):
        self.process_id = process_id
        self.time = 0
    
    def local_event(self) -> int:
        """本地事件发生"""
        self.time += 1
        return self.time
    
    def send_message(self) -> tuple:
        """发送消息"""
        self.time += 1
        return (self.time, self.process_id)
    
    def receive_message(self, msg_time: int) -> int:
        """接收消息"""
        self.time = max(self.time, msg_time) + 1
        return self.time

进程 P1 (Lamport Clock)    进程 P2 (Lamport Clock)
   │                              │
   ▼ T=1                          ▼ T=1
  A: x=1                        C: y=2
   │                              │
   ▼ T=2                          ▼ T=2
  B: send(m, T=2) ─────────────► D: receive(m)
                                        T = max(2,2)+1 = 3
   │                              │
   ▼ T=3                          ▼ T=4
  E: x=2                        F: y=3

如果 A → B，则 L(A) < L(B)

但逆命题不成立！
  L(A) < L(B) ↛ A → B

反例：
  A (P1, T=1): x=1
  C (P2, T=1): y=2
  
  L(A) = 1 < L(C) = 1? 不成立，需要打破平局
  
  实际上：L(A) = (1, P1), L(C) = (1, P2)
  通过进程 ID 打破平局：(1, P1) < (1, P2)
  
  但 A 和 C 是并发的！

from typing import Dict, List

class VectorClock:
    """
    向量时钟实现
    每个进程维护一个向量，记录所有进程的时钟
    """
    
    def __init__(self, process_id: str, num_processes: int):
        self.process_id = process_id
        self.num_processes = num_processes
        # 向量: [时钟_0, 时钟_1, ..., 时钟_n-1]
        self.vector = [0] * num_processes
        self.process_index = int(process_id[-1])  # 简化的进程索引
    
    def local_event(self) -> List[int]:
        """本地事件发生"""
        self.vector[self.process_index] += 1
        return self.vector.copy()
    
    def send_message(self) -> tuple:
        """发送消息"""
        self.vector[self.process_index] += 1
        return (self.vector.copy(), self.process_id)
    
    def receive_message(self, msg_vector: List[int]) -> List[int]:
        """接收消息"""
        # 逐元素取最大值
        for i in range(self.num_processes):
            self.vector[i] = max(self.vector[i], msg_vector[i])
        self.vector[self.process_index] += 1
        return self.vector.copy()
    
    def compare(self, other: List[int]) -> str:
        """
        比较两个向量时钟
        返回: '<', '>', '=', '||' (并发)
        """
        less = False
        greater = False
        
        for i in range(self.num_processes):
            if self.vector[i] < other[i]:
                less = True
            elif self.vector[i] > other[i]:
                greater = True
        
        if less and not greater:
            return '<'
        elif greater and not less:
            return '>'
        elif not less and not greater:
            return '='
        else:
            return '||'  # 并发

进程 P0 (VC)           进程 P1 (VC)
   │                       │
   ▼ [1,0]                 ▼ [0,1]
  A: x=1                 C: y=2
   │                       │
   ▼ [2,0]                 ▼ [0,2]
  B: send(m) ───────────► D: receive(m)
   │                      [2,2] = max([2,0], [0,2]) + [0,1]
   ▼ [3,0]                 ▼ [2,3]
  E: x=2                 F: y=3

向量时钟比较：
  VC(A) = [1,0], VC(C) = [0,1]
  A[0]=1 > C[0]=0, A[1]=0 < C[1]=1
  结果：A || C (并发)
  
  VC(B) = [2,0], VC(D) = [2,2]
  B[0]=2 = D[0]=2, B[1]=0 < D[1]=2
  结果：B < D (B happens-before D)

定理：对于任意两个事件 e1, e2

1. e1 → e2  ⟺  VC(e1) < VC(e2) (逐元素小于)
2. e1 || e2 ⟺  VC(e1) || VC(e2) (不可比较)

证明：
(=>) 如果 e1 → e2，则存在因果链
     每步因果链都会增加至少一个进程的时钟
     因此 VC(e2) 在每个分量上都 ≥ VC(e1)
     且至少有一个分量严格大于

(<=) 如果 VC(e1) < VC(e2)，则 e2 的向量在每个分量上
     都至少和 e1 一样大。这意味着 e2 知道 e1 发生的
     所有信息，因此 e1 必然 happens-before e2

class VersionVector:
    """
    版本向量用于检测多主复制中的冲突
    """
    
    def __init__(self, replica_id: str):
        self.replica_id = replica_id
        self.versions: Dict[str, int] = {replica_id: 0}
    
    def increment(self):
        """本地修改数据时调用"""
        self.versions[self.replica_id] += 1
    
    def merge(self, other: Dict[str, int]):
        """合并来自其他副本的版本向量"""
        for replica, version in other.items():
            self.versions[replica] = max(
                self.versions.get(replica, 0),
                version
            )
    
    def compare(self, other: Dict[str, int]) -> str:
        """
        比较两个版本向量，判断数据关系
        返回: 'ancestor', 'descendant', 'conflict', 'same'
        """
        dominates = False
        dominated = False
        
        all_replicas = set(self.versions.keys()) | set(other.keys())
        
        for replica in all_replicas:
            v1 = self.versions.get(replica, 0)
            v2 = other.get(replica, 0)
            
            if v1 > v2:
                dominates = True
            elif v2 > v1:
                dominated = True
        
        if dominates and dominated:
            return 'conflict'  # 需要冲突解决
        elif dominates:
            return 'descendant'  # self 是 other 的后代
        elif dominated:
            return 'ancestor'  # self 是 other 的祖先
        else:
            return 'same'

# 使用示例
replica_a = VersionVector('A')
replica_b = VersionVector('B')

# A 修改数据
replica_a.increment()  # versions: {A: 1}

# B 修改数据（并发）
replica_b.increment()  # versions: {B: 1}

# 同步时检测冲突
result = replica_a.compare(replica_b.versions)
# result = 'conflict' - 需要手动合并或自动解决

class IntervalTreeClock:
    """
    区间树时钟（简化概念演示）
    实际实现更复杂，涉及树的合并与拆分
    """
    
    def __init__(self, id: str = None):
        self.id = id
        self.tree = {'start': 0, 'end': 1, 'value': 0}
    
    def fork(self) -> 'IntervalTreeClock':
        """创建新进程时拆分时钟区间"""
        new_clock = IntervalTreeClock()
        # 将当前区间拆分为两半
        mid = (self.tree['start'] + self.tree['end']) / 2
        new_clock.tree = {
            'start': mid,
            'end': self.tree['end'],
            'value': 0
        }
        self.tree['end'] = mid
        return new_clock
    
    def event(self):
        """记录事件"""
        self.tree['value'] += 1
    
    def join(self, other: 'IntervalTreeClock'):
        """进程退出时合并时钟区间"""
        # 合并两棵区间树
        pass

NTP 同步误差来源：
1. 网络不对称延迟
   - 请求路径延迟 ≠ 响应路径延迟
   - 误差 = (delay_out - delay_in) / 2

2. 时钟漂移
   - 石英晶体频率不稳定
   - 温度、电压影响振荡频率

3. 网络抖动
   - 排队延迟变化
   - 路由变化

典型精度：
- 局域网：1-10 ms
- 广域网：10-100 ms
- 互联网：100+ ms

场景：基于时间戳的乐观锁

T+0ms    客户端 A 读取数据，timestamp=100
T+50ms   客户端 B 读取数据，timestamp=150 (时钟快50ms)
T+100ms  客户端 A 写入数据，timestamp=200
T+120ms  客户端 B 写入数据，timestamp=170

结果：B 的写入"看起来"比 A 早，被错误地接受！

// TrueTime 返回一个时间区间 [earliest, latest]
// 保证真实时间一定落在这个区间内

tt := spanner.Now()  // 返回 TTinterval

// TTinterval 结构
type TTinterval struct {
    Earliest time.Time  // 最早可能时间
    Latest   time.Time  // 最晚可能时间
}

// 不确定性边界（epsilon）
epsilon := tt.Latest.Sub(tt.Earliest)
// 典型值：1-7 ms

TrueTime 实现：

1. 参考时钟源
   ├── GPS 接收器（原子钟级别精度）
   └── 原子钟（本地备份）

2. 时间同步协议
   ├── Marzullo 算法（容错时钟同步）
   └── 定期与参考源同步

3. 不确定性估计
   └── ε = 同步间隔 × 时钟漂移率 + 网络延迟

Spanner 保证：
如果事务 T1 在 T2 开始之前提交，
则 T1 的提交时间戳 < T2 的提交时间戳

实现方式：
- 提交时等待：sleep(ε)
- 确保时间戳差距大于不确定性边界

事务延迟增加：
- 写事务：+ ε (通常 2-7 ms)
- 读事务：无额外延迟（使用最新时间戳）

权衡：
- 获得：外部一致性保证
- 付出：小幅延迟增加

时钟模型演进：

1. 理想时钟（单机）
   时间 = 单调递增的标量
   
2. 物理时钟（分布式，传统）
   时间 = 有偏差的标量
   问题：隐藏了不确定性
   
3. 区间时钟（TrueTime）
   时间 = [earliest, latest]
   优势：显式建模不确定性
   
4. 向量时钟（逻辑）
   时间 = 因果关系的偏序
   优势：完全正确，无不确定性

class HybridClock:
    """
    混合逻辑时钟（结合物理时间与逻辑时间）
    用于 CockroachDB 等系统
    """
    
    def __init__(self):
        self.wall_time = 0  # 物理时间戳
        self.logical = 0    # 逻辑计数器
    
    def now(self) -> tuple:
        """获取当前时间"""
        current_wall = get_physical_time()
        
        if current_wall > self.wall_time:
            # 物理时间前进，重置逻辑计数
            self.wall_time = current_wall
            self.logical = 0
        else:
            # 物理时间停滞或回退，增加逻辑计数
            self.logical += 1
        
        return (self.wall_time, self.logical)
    
    def update(self, other: tuple):
        """接收其他节点的时间"""
        other_wall, other_logical = other
        
        if other_wall > self.wall_time:
            self.wall_time = other_wall
            self.logical = other_logical + 1
        elif other_wall == self.wall_time:
            self.logical = max(self.logical, other_logical + 1)
        # else: other_wall < self.wall_time，忽略

class DynamoNode:
    def put(self, key: str, value: any, context: dict = None):
        """
        写入数据
        context: 来自之前读取的向量时钟
        """
        # 1. 生成新的向量时钟
        new_clock = self.increment_vector_clock(context)
        
        # 2. 存储数据版本
        version = DataVersion(
            value=value,
            vector_clock=new_clock,
            timestamp=time.now()
        )
        
        # 3. 异步复制到 N 个节点
        self.replicate(key, version)
    
    def get(self, key: str) -> list:
        """
        读取数据
        可能返回多个版本（冲突时）
        """
        versions = self.read_from_replicas(key)
        
        # 检查版本关系
        result = []
        for v in versions:
            # 如果 v 不是任何其他版本的祖先，保留
            if not self.is_ancestor(v, versions):
                result.append(v)
        
        return result
    
    def is_ancestor(self, version, all_versions) -> bool:
        """检查 version 是否是其他版本的祖先"""
        for other in all_versions:
            if other != version:
                if version.vector_clock < other.vector_clock:
                    return True
        return False

当 get() 返回多个版本时，应用层需要合并：

1. 购物车示例（自然合并）
   版本 A: {apple: 1, banana: 2}
   版本 B: {apple: 2, orange: 1}
   合并: {apple: 2, banana: 2, orange: 1}  # 取最大值

2. 计数器示例（CRDT）
   使用 G-Counter 或 PN-Counter

3. 通用场景（Last-Write-Wins）
   选择时间戳最新的版本
   （可能丢失更新）

本地层（Local Cluster）：
- 保证线性一致性
- 低延迟读写
- 不跨数据中心

全球层（Global Cluster）：
- 异步复制
- 保证因果一致性
- 跨数据中心

class COPSClient:
    def __init__(self):
        self.dependencies = set()  # 当前事务依赖的版本
    
    def get(self, key: str) -> Value:
        value = self.local_cluster.read(key)
        # 记录依赖
        self.dependencies.add((key, value.version))
        return value
    
    def put(self, key: str, value: any):
        # 写入时携带所有依赖
        self.local_cluster.write(
            key=key,
            value=value,
            dependencies=self.dependencies
        )
        # 清空依赖，开始新事务
        self.dependencies = set()

如果操作 A 在操作 B 之前（A → B），
则所有节点看到 A 的效果后，才能看到 B 的效果。

实现方式：
- 写入时携带前置依赖
- 远程节点必须满足所有依赖后才能应用写入

1. 事务支持
   - 快照读（Snapshot Read）
   - 原子写（Atomic Write）

2. 因果一致性
   - 显式依赖追踪
   - 异步复制

3. CRDT 数据类型
   - G-Counter（增长计数器）
   - PN-Counter（正负计数器）
   - G-Set（增长集合）
   - OR-Set（观察移除集合）
   - LWW-Register（最后写入获胜寄存器）

class GCounter:
    """
    增长计数器 CRDT
    只支持增加操作，天然无冲突
    """
    
    def __init__(self, replica_id: str):
        self.replica_id = replica_id
        self.counters: Dict[str, int] = {}
    
    def increment(self):
        """本地增加"""
        self.counters[self.replica_id] = \
            self.counters.get(self.replica_id, 0) + 1
    
    def value(self) -> int:
        """查询当前值"""
        return sum(self.counters.values())
    
    def merge(self, other: 'GCounter'):
        """合并其他副本的计数器"""
        for replica, count in other.counters.items():
            self.counters[replica] = max(
                self.counters.get(replica, 0),
                count
            )

传统架构：
  状态 = 当前数据库中的值
  
事件溯源：
  状态 = fold(初始状态, 事件序列)
  
示例：银行账户
  初始余额: 0
  事件 1: Deposit(100)  → 余额: 100
  事件 2: Withdraw(30)   → 余额: 70
  事件 3: Deposit(50)    → 余额: 120
  
  当前状态 = 0 + 100 - 30 + 50 = 120

1. 天然因果一致性
   - 事件顺序 = 因果关系
   - 并发事件可以并行处理

2. 完美审计
   - 完整历史记录
   - 任意时刻状态可重建

3. 灵活查询
   - 投影（Projection）到不同视图
   - CQRS 模式

CRDT 必须满足：
1. 交换律：a ⊔ b = b ⊔ a
2. 结合律：(a ⊔ b) ⊔ c = a ⊔ (b ⊔ c)
3. 幂等律：a ⊔ a = a

其中 ⊔ 是合并操作（join）

状态型 CRDT（State-based）：
- 传输完整状态
- 合并操作：⊔
- 示例：G-Set, LWW-Register

操作型 CRDT（Operation-based）：
- 传输操作
- 需要可靠广播
- 示例：Woot（协同编辑）

比特币：
- 区块时间戳由矿工设置
- 不要求精确，只需大致有序
- 难度调整维持平均 10 分钟出块

以太坊：
- 时间戳必须大于父区块
- 允许 ±15 秒误差
- 用于合约中的 now 变量

区块链通过哈希链保证因果序：

Block N:   [..., Timestamp, Hash_N-1] → Hash_N
Block N+1: [..., Timestamp, Hash_N]   → Hash_N+1

每个区块引用前一个区块的哈希，
形成不可篡改的因果链。

分布式系统时间观的演进：

1978 ─ Lamport: Time, Clocks, and Ordering
       └── 提出 Happens-Before 关系
       └── 逻辑时钟的概念

1988 ─ 向量时钟（Fidge, Mattern）
       └── 可判断并发关系
       └── 但空间开销大

2000s ─ 版本向量（多主复制）
        └── 冲突检测
        └── Dynamo, Riak

2012 ─ Spanner TrueTime
       └── 显式建模时钟不确定性
       └── 外部一致性保证

2010s ─ CRDT 普及
        └── 无冲突复制
        └── 事件溯源架构

2020s ─ 混合策略
        └── 逻辑时钟 + 物理时钟
        └── 因果一致性 + 最终一致性

在分布式系统中：
- 物理时间是测量的便利工具
- 因果关系是系统的本质属性
- 并发是常态，顺序是特例

Spanner 的启示：
- 不要隐藏时钟误差
- 量化不确定性边界
- 在系统设计层面处理

Dynamo 的启示：
- 接受短期不一致
- 应用层解决冲突
- 可用性优先于强一致性

一致性需求分析：

需要线性一致性？
├── 是 → 使用 Paxos/Raft
│   └── 代价：性能、可用性
│
└── 否 → 需要因果一致性？
    ├── 是 → 使用向量时钟 + 应用层合并
    │   └── 适合：社交、协作
    │
    └── 否 → 使用最终一致性
        ├── 有冲突可能？→ CRDT
        └── 无冲突？→ 简单异步复制