🕰️ 向量时钟 & 因果一致性

❓问题：分布式系统中没有全局时钟

在单机程序中，判断"哪个操作先发生"很简单——看时间戳。但在分布式系统中，不存在可靠的全局时钟。

经典困境：Last Write Wins (LWW) 的失败

假设三个节点 A、B、C 管理同一个 key：

Happens-Before 关系 (→)

Lamport 定义了 happens-before 关系（记作 →），三条规则：

并发：如果既非 a → b 也非 b → a，则 a 和 b 是并发的，记作 a || b。

🔢Lamport 时钟：偏序的第一步

Lamport 时钟是最简单的逻辑时钟——一个单调递增的整数计数器。

算法

代码实现

class LamportClock:
    """Lamport 逻辑时钟"""
    def __init__(self):
        self.counter = 0

    def event(self) -> int:
        """进程内事件"""
        self.counter += 1
        return self.counter

    def send(self) -> tuple:
        """发送消息：先递增，附带时间戳"""
        self.counter += 1
        return (self.counter,)  # (timestamp,)

    def receive(self, msg_timestamp: int) -> int:
        """接收消息：取 max + 1"""
        self.counter = max(self.counter, msg_timestamp) + 1
        return self.counter

# 使用示例
clock_a = LamportClock()
clock_b = LamportClock()

# A 发生本地事件
t1 = clock_a.event()  # A: 1

# A → B 发送消息
ts = clock_a.send()   # A: 2, 消息附带 ts=2
t2 = clock_b.receive(ts[0])  # B: max(0, 2) + 1 = 3

print(f"A={clock_a.counter}, B={clock_b.counter}")  # A=2, B=3

性质

🕐向量时钟：捕获因果

向量时钟由 Colin Fidge (1988) 和 Friedemann Mattern (1989) 独立提出。核心思想：用 N 维向量代替单一整数，每个维度对应一个进程。

定义

N 个进程的系统中，向量时钟 VC 是长度为 N 的整数数组 [v₀, v₁, ..., v_{N-1}]。

进程 i 的向量时钟记为 VC_i，其中 VC_i[j] 表示进程 i 所知道的进程 j 的最新事件数。

算法

代码实现

from typing import List

class VectorClock:
    """向量时钟实现"""
    def __init__(self, node_id: int, total_nodes: int):
        self.id = node_id
        self.vc = [0] * total_nodes

    def event(self) -> List[int]:
        """进程内事件：递增自己的分量"""
        self.vc[self.id] += 1
        return self.vc.copy()

    def send(self) -> List[int]:
        """发送消息：先递增，附带整个 VC"""
        self.vc[self.id] += 1
        return self.vc.copy()

    def receive(self, msg_vc: List[int]) -> List[int]:
        """接收消息：逐分量取 max，再递增自己"""
        for k in range(len(self.vc)):
            self.vc[k] = max(self.vc[k], msg_vc[k])
        self.vc[self.id] += 1
        return self.vc.copy()

    def __repr__(self):
        return f"VC{self.id}={self.vc}"

# --- 使用示例 ---
# 3 个节点
vc_a = VectorClock(0, 3)  # A=节点0
vc_b = VectorClock(1, 3)  # B=节点1
vc_c = VectorClock(2, 3)  # C=节点2

# A 本地事件
vc_a.event()        # VC_A = [1, 0, 0]

# A 发消息给 B
msg = vc_a.send()   # VC_A = [2, 0, 0]
vc_b.receive(msg)   # VC_B = max([0,0,0],[2,0,0]) + [0,1,0] = [2,1,0]

# C 本地事件（与 A、B 并发）
vc_c.event()        # VC_C = [0, 0, 1]

print(vc_a)  # VC0=[2, 0, 0]
print(vc_b)  # VC1=[2, 1, 0]
print(vc_c)  # VC2=[0, 0, 1]

比较规则：因果关系的判断

交互式模拟器

完整示例：为什么向量时钟能区分因果和并发

⚖️Lamport 时钟 vs 向量时钟：关键区别

为什么 Lamport 时钟"丢了信息"？

考虑两个事件：

• 事件 a 在节点 A，LC(a) = 3
• 事件 b 在节点 B，LC(b) = 5

Lamport 时钟只能告诉你 LC(a) < LC(b)。但可能有三种情况：

1. a → b（A 的消息传到了 B）
2. b → a（B 的消息传到了 A——不太可能因为 LC(b) > LC(a)，但并非绝对）
3. a || b（并发）

向量时钟保留了每个维度的信息：VC(a)=[3,0] vs VC(b)=[0,5]，一看就知道是并发。Lamport 时钟把这些维度压成了一个数——信息不可逆地丢失了。

🔗因果一致性：比线性一致弱，比最终一致强

向量时钟是工具，因果一致性是目标。根据 Jepsen 的分类，因果一致性（Causal Consistency）是在始终可用的系统中能保证的最强一致性模型。

因果一致性的定义

根据 Ahamad, Neiger, Burns 等人 1993 年的论文以及 Mahajan, Alvisi, Dahlin 的形式化：

1. 因果可见性（CausalVisibility）：如果 a → b，则所有看到 b 的进程也必须看到 a
2. 因果仲裁（CausalArbitration）：因果关系的顺序在仲裁（冲突解决）中必须被保留
3. 返回值一致性（RVal）：读操作返回的值必须反映最近的所有写入

因果一致性的可用性

根据 Mahajan 等人的证明（"Consistency, Availability, and Convergence"），以及 Viotti 和 Vukolić 的综述（arXiv:1512.00168）：

向量时钟如何实现因果一致性

1. 每个写操作附带向量时钟
2. 读操作返回值时，同时返回对应的向量时钟
3. 如果 VC(a) < VC(b)，系统保证所有节点先看到 a 再看到 b
4. 如果 VC(a) || VC(b)，两个操作可以以任意顺序出现（但最终收敛）

🛒实战：Amazon Dynamo 的向量时钟

Amazon Dynamo（DeCandia et al., 2007, "Dynamo: Amazon's Highly Available Key-value Store"）是最著名的向量时钟工业应用。

Dynamo 的设计选择

Dynamo 向量时钟的实际格式

Dynamo 的 VC 不是简单的节点 ID 计数器。由于节点可能动态加入/离开，Dynamo 使用 (node_id, counter) 对的列表：

# Dynamo 风格的向量时钟
# 格式: [(node_id, counter), ...]
#
# 例如：
dynamo_vc_v1 = [("node-A", 1)]                    # A 写入 1 次
dynamo_vc_v2 = [("node-A", 2)]                    # A 又写入 1 次（v2 > v1）
dynamo_vc_v3 = [("node-A", 1), ("node-B", 1)]     # A 写入后 B 写入（v3 > v1）

# 并发情况：A 和 B 各自独立更新
dynamo_vc_v4 = [("node-A", 2)]                    # A 独立更新 2 次
dynamo_vc_v5 = [("node-B", 1)]                    # B 独立更新 1 次
# v4 和 v5 并发！都是 v1 的后续，但彼此不可比较

Sibling（兄弟版本）问题

Dynamo 在 Amazon 的实际部署数据

根据 Dynamo 论文：

受 Dynamo 启发的系统

🔬进阶：Dotted Version Vectors (DVV)

标准向量时钟有一个微妙的 bug：在多副本场景下可能丢失因果关系。这个 bug 被 Paulo Sérgio Almeida 等人在 2014 年的论文中揭示，并提出了 DVV 作为修复方案。

标准向量时钟的 Bug

DVV 的修复

DVV (Dotted Version Vectors) 将这两个角色分离：

• Dot（点）：精确标识一次写入事件 (server_id, counter)
• Version Vector（版本向量）：记录所有已知的 dot

# DVV 数据结构
class DottedVersionVector:
    def __init__(self):
        self.dot = None     # (server_id, counter) — 标识当前版本
        self.vv = {}        # {server_id: max_counter} — 已知的所有 dot

    def create(self, server_id: str):
        """Server 处理写入时创建新 dot"""
        counter = self.vv.get(server_id, 0) + 1
        self.dot = (server_id, counter)
        self.vv[server_id] = counter
        return self

    def merge(self, other):
        """合并两个 DVV"""
        # 合并 version vectors
        for sid, cnt in other.vv.items():
            self.vv[sid] = max(self.vv.get(sid, 0), cnt)
        # 保留 dot（如果 other.dot 不在合并后的 vv 中，它就是并发 sibling）
        return self

    def dominates(self, other) -> bool:
        """判断是否因果支配 other"""
        if other.dot:
            sid, cnt = other.dot
            if self.vv.get(sid, 0) >= cnt:
                return True  # other 的 dot 已被我们知道
        return False

Riak 的迁移

Riak 从标准向量时钟迁移到 DVV（Riak 2.0+），解决了 sibling 爆增问题。根据 Riak 文档：

🔀Git：最成功的向量时钟实例

你可能每天都在用向量时钟——只是没意识到。Git 的提交历史就是一个向量时钟实例。

Git 如何体现向量时钟

Git 提交作为"事件"

# Git 的 commit hash 就是"事件 ID"
# 每个 commit 记录了 parent(s)，这构成了 happens-before 关系

# 线性历史（因果有序）
A -- B -- C -- D
# A → B → C → D

# 分叉（并发）
A -- B -- C (main)
      \
       D -- E (feature)
# C 和 D 是并发的！C || D

# Merge（合并向量时钟）
A -- B -- C ------ M (main)
      \           /
       D -- E ---   (feature)
# M 的 parent 是 C 和 E，等价于 VC merge

Git 的 "Last Write Wins" 问题

从 Git 理解向量时钟的本质

🤖Agent 系统中的因果顺序

多 Agent 系统天然是分布式系统——多个 Agent 实例并行工作，需要协调"谁做了什么"。

场景 1：多 Agent 协作的因果依赖

场景 2：记忆同步的并发冲突

{name: "Alice", email: "bob@x.com"}  # CRDT merge

场景 3：工具调用链的部分失败

Agent 记忆系统中的向量时钟设计

class AgentMemoryWithVC:
    """带向量时钟的 Agent 记忆系统"""
    def __init__(self, agent_id: str, cluster_size: int):
        self.agent_id = agent_id
        self.agent_idx = self._resolve_index(agent_id)
        self.vc = [0] * cluster_size
        self.memories = {}  # key -> (value, vc)

    def write(self, key: str, value: str):
        """写入记忆，附带向量时钟"""
        self.vc[self.agent_idx] += 1
        self.memories[key] = (value, self.vc.copy())

    def read(self, key: str):
        """读取记忆，返回 (value, vc)"""
        return self.memories.get(key, (None, None))

    def sync(self, other_memories: dict):
        """从其他 Agent 同步记忆"""
        for key, (value, other_vc) in other_memories.items():
            if key not in self.memories:
                # 新 key，直接采纳
                self.memories[key] = (value, other_vc)
                self._merge_vc(other_vc)
            else:
                local_value, local_vc = self.memories[key]
                cmp = self._compare_vc(local_vc, other_vc)
                if cmp == 'less':
                    # 远端更新，采纳
                    self.memories[key] = (value, other_vc)
                    self._merge_vc(other_vc)
                elif cmp == 'concurrent':
                    # 并发冲突！保留两个版本
                    self.memories[key] = self._merge_values(
                        local_value, local_vc, value, other_vc
                    )
                    self._merge_vc(other_vc)
                # else: local 更新，忽略

    def _compare_vc(self, a, b):
        a_less = any(a[i] < b[i] for i in range(len(a)))
        b_less = any(a[i] > b[i] for i in range(len(a)))
        if a_less and b_less: return 'concurrent'
        if a_less: return 'less'
        if b_less: return 'greater'
        return 'equal'

    def _merge_vc(self, other_vc):
        for i in range(len(self.vc)):
            self.vc[i] = max(self.vc[i], other_vc[i])

    def _merge_values(self, val1, vc1, val2, vc2):
        """应用层合并策略——默认保留两个版本"""
        # 实际中可以用 CRDT 或让 Agent 自己决定
        return (f"<> v1={val1} v2={val2}", vc1)

💻代码实战

完整向量时钟 + 冲突检测实现

"""
向量时钟完整实现：包括冲突检测、版本合并、垃圾回收
"""
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import time

class CausalRelation(Enum):
    BEFORE = "before"       # a → b
    AFTER = "after"         # b → a
    CONCURRENT = "concurrent"  # a || b
    EQUAL = "equal"         # a 和 b 相同

@dataclass
class VersionedValue:
    """带版本号的值"""
    value: any
    vc: List[int]
    timestamp: float = field(default_factory=time.time)

class VectorClockStore:
    """带向量时钟的 KV 存储"""
    def __init__(self, node_id: int, cluster_size: int):
        self.node_id = node_id
        self.cluster_size = cluster_size
        self.vc = [0] * cluster_size        # 当前节点的向量时钟
        self.data: Dict[str, List[VersionedValue]] = {}  # key -> [siblings]

    def _increment(self):
        """递增自己的分量"""
        self.vc[self.node_id] += 1

    def _merge_vc(self, other_vc: List[int]):
        """合并另一个 VC 到自己的 VC"""
        for i in range(self.cluster_size):
            self.vc[i] = max(self.vc[i], other_vc[i])

    @staticmethod
    def compare_vc(a: List[int], b: List[int]) -> CausalRelation:
        """比较两个向量时钟的因果关系"""
        a_less = any(a[i] < b[i] for i in range(len(a)))
        b_less = any(a[i] > b[i] for i in range(len(a)))

        if not a_less and not b_less:
            return CausalRelation.EQUAL
        if a_less and not b_less:
            return CausalRelation.BEFORE
        if b_less and not a_less:
            return CausalRelation.AFTER
        return CausalRelation.CONCURRENT

    def put(self, key: str, value: any):
        """写入值"""
        self._increment()
        new_vc = self.vc.copy()
        new_val = VersionedValue(value=value, vc=new_vc)

        if key not in self.data:
            self.data[key] = [new_val]
            return

        # 检查与现有 siblings 的关系
        new_siblings = []
        dominated = False
        for existing in self.data[key]:
            rel = self.compare_vc(new_vc, existing.vc)
            if rel == CausalRelation.BEFORE:
                # 新值因果先于已有值 → 新值被支配，丢弃
                dominated = True
                new_siblings.append(existing)
            elif rel == CausalRelation.AFTER:
                # 新值因果后于已有值 → 已有值被支配，丢弃
                pass  # 不保留 existing
            else:
                # 并发或相等 → 保留为 sibling
                new_siblings.append(existing)

        if not dominated:
            new_siblings.append(new_val)

        self.data[key] = new_siblings

    def get(self, key: str) -> List[VersionedValue]:
        """读取值（可能返回多个 siblings）"""
        return self.data.get(key, [])

    def merge(self, key: str, remote_values: List[VersionedValue],
              remote_vc: List[int]):
        """从远程节点合并数据"""
        self._merge_vc(remote_vc)

        if key not in self.data:
            self.data[key] = remote_values.copy()
            return

        # 合并 siblings
        all_values = self.data[key] + remote_values
        merged = self._deduplicate_siblings(all_values)
        self.data[key] = merged

    def _deduplicate(self, values: List[VersionedValue]) -> List[VersionedValue]:
        """去除被支配的 siblings"""
        result = []
        for v in values:
            dominated = False
            for other in result:
                rel = self.compare_vc(v.vc, other.vc)
                if rel == CausalRelation.BEFORE:
                    dominated = True
                    break
                elif rel == CausalRelation.AFTER:
                    result.remove(other)
                    break
            if not dominated:
                result.append(v)
        return result

    def _deduplicate_siblings(self, values):
        return self._deduplicate(values)

    @property
    def clock(self) -> List[int]:
        return self.vc.copy()

    def siblings_count(self, key: str) -> int:
        """返回某个 key 的 sibling 数量"""
        return len(self.data.get(key, []))

# === 使用示例 ===
if __name__ == "__main__":
    # 3 节点集群
    store_a = VectorClockStore(node_id=0, cluster_size=3)
    store_b = VectorClockStore(node_id=1, cluster_size=3)

    # A 写入 "color" = "red"
    store_a.put("color", "red")
    print(f"A after put 'red': VC={store_a.clock}")

    # B 并发写入 "color" = "blue"（不知道 A 的写入）
    store_b.put("color", "blue")
    print(f"B after put 'blue': VC={store_b.clock}")

    # A 再写入一次（基于 "red"）
    store_a.put("color", "green")
    print(f"A after put 'green': VC={store_a.clock}")

    # 模拟 A 收到 B 的数据并合并
    b_values = store_b.get("color")
    store_a.merge("color", b_values, store_b.clock)
    print(f"A after merge: VC={store_a.clock}")
    print(f"Siblings: {[(v.value, v.vc) for v in store_a.get('color')]}")
    # 输出：两个 sibling ("green", [2,0,0]) 和 ("blue", [0,1,0])
    # 因为它们是并发的！

向量时钟垃圾回收

向量时钟不会自动缩小。长时间运行的系统中，需要清理不再需要的旧版本。

def gc_vector_clocks(store: VectorClockStore, max_siblings: int = 10):
    """向量时钟垃圾回收"""
    for key in list(store.data.keys()):
        siblings = store.data[key]
        if len(siblings) <= 1:
            continue

        # 策略 1：去除被支配的旧版本
        store.data[key] = store._deduplicate_siblings(siblings)

        # 策略 2：如果 sibling 仍然太多，按时间戳排序保留最近的
        if len(store.data[key]) > max_siblings:
            store.data[key].sort(key=lambda v: v.timestamp, reverse=True)
            # 保留最新的 max_siblings 个
            store.data[key] = store.data[key][:max_siblings]

💣踩坑指南

Cassandra 的教训

🌳选型决策树

只需要给事件排序？
├─ YES → Lamport 时钟（O(1) 空间，实现简单）
└─ NO → 需要检测并发冲突？
    ├─ YES → 并发频率低（<1%）？
    │   ├─ YES → 标准向量时钟（如 Dynamo/Riak）
    │   └─ NO → Dotted Version Vectors (DVV)
    └─ NO → 需要保证因果一致性？
        ├─ YES → 向量时钟 + 因果广播协议
        └─ NO → 最终一致性 + LWW 足矣
            （如 Cassandra 场景）

更具体地：

你的数据能接受偶尔丢数据吗？
├─ YES → LWW（Cassandra 模式）
└─ NO → 你能接受多个 sibling 需要合并吗？
    ├─ YES → 向量时钟 / DVV（Riak 模式）
    └─ NO → CRDT（自动合并，参见 crdt.html）
        或 线性一致性（强一致，牺牲可用性）

系统对比速查

关键参考文献

🔗相关阅读