理解共识算法----Hotstuff

HotStuff也是一种基于拜占庭容错（BFT）的区块链共识算法，由VMware Research团队于2019年提出，后成为Libra 区块链的核心共识协议。其核心创新在于通过线性视图切换（Linear View Change）和门限签名（Threshold Signatures）实现高效流水线化处理，大幅降低传统BFT算法的通信复杂度。

这里主要有两个关键的优化点：

引入了门限签名减少通信复杂度。门限签名：允许一组参与者合作生成一个有效的数字签名，但只要达到预定的门限数量（阈值）即可完成签名，而无需所有参与者都参与。
相对于PBFT在消息通信方面需要的O(n^2)、O(n^3)的复杂度，HotStuff只需要O(n)的复杂度。
流程更加简单，使其共识流程能够流水线化。

HotStuff 原论文：https://arxiv.org/pdf/1803.05069

HotStuff 论文中译版：https://josephucas.github.io/2022/09/26/HotStuff%EF%BC%9ABFT-Consensus-in-the-Lens-of-Blockchain/

Hotstuff共识算法流程

QC(Quorum Certificate)

理解Hotstuff共识算法，理解QC至关重要：

QC(Quorum Certificate 法定人数证明）是一组签名的集合，证明某个提案（Proposal）或区块已经获得了足够数量的验证者（节点）投票认可，达到了法定多数（通常是超过2/3的验证者）同意。

HotStuff 中的共识流程由多个阶段组成（Prepare、Pre-commit、Commit、Decide），每个阶段节点都会对某个区块进行投票。当收到超过 2/3 验证者的投票后，节点会将这些投票签名聚合成一个QC。这个 QC 绑定了被投票区块的哈希，且经过聚合签名（例如 BLS 聚合签名）或多签，任何节点都可以验证这个 QC 的有效性。QC的基础就是门限签名。

共识流程

如上图所示，Hotstuff 是一个三阶段投票共识算法，主要的共识步骤如下：

Proposal

领导者根据当前共识状态，构造一个新区块（Proposal），这个区块会引用一个之前阶段的 QC，确保链的连续性和安全性，并将该提案广播给所有验证者（节点）。在开始新一轮的共识时，新的leader会收到所有验证者发来的自己的Highest QC（这里可以先理解为自己看到的最长链）。新的领导者会根据这个Highest QC选择在哪条最长链上继续提出新的区块。

Prepare 阶段

各个节点收到提案后，验证区块合法性和前置条件（如引用的 QC 是否有效）。验证通过后，节点发出 Prepare 投票（一个对该区块的签名），并发送给领导者。领导者收集到超过 2f+1 的 Prepare 投票后，聚合成一个 Prepare QC。

Pre-commit 阶段

领导者将带有 Prepare QC 的区块提案再次广播。节点收到后，验证 Prepare QC 并确认上阶段完成。之后，节点发出 Pre-commit 投票给领导者。领导者收集足够的 Pre-commit 投票后，生成 Pre-commit QC。

Commit 阶段

领导者将带有 Pre-commit QC 的区块广播。节点验证 Pre-commit QC，确认区块进入 Commit 阶段。节点发出 Commit 投票给领导者。领导者收集超过 2f+1 Commit 投票后，生成 Commit QC。

Decide 阶段

当节点收到带有 Commit QC 的区块后，区块被最终确定。节点将区块提交执行（应用状态机），并更新本地链状态。该区块成为共识链上不可回退的区块。如上图该阶段也是开始新一轮共识的一环。

总结图示（来源见水印）

超时流程处理

HotStuff 将共识过程划分为多个视图（View），每个视图对应一个领导者。每个节点维护一个超时计时器，针对当前视图。计时器从进入新视图开始计时。如果在预定时间内未收到领导者的有效提案或未完成共识阶段，超时触发。超时触发后，节点会向其他节点广播一个Timeout 消息，表明当前视图的领导者未能及时推进共识。

Timeout 消息中通常包含：当前视图号、最新的 QC、节点签名等信息。当某个节点收到超过 2f+1 个来自不同节点的 Timeout 消息时，便可将这些超时签名聚合成一个 TC（Timeout Certificate）。

TC 是一种“法定人数超时证明”，表明大量节点认为当前视图失效。收到 TC 后，新领导者据此决定切换视图。新领导者基于收集到的最高 QC（或 TC 中包含的状态）构造新的提案，确保安全性不被破坏。新视图启动后，所有节点重置超时计时器，开始对新领导者的提案进行投票。

Hotstuff共识算法关键点

三阶段投票避免Lock以及线性视图切换

看过上一篇Tendermint共识算法的了解到其为了避免view-change可能会出现的区块“分叉”、二义性问题，会使用Lock的方式来保证view-change过程的安全性。而Hotstuff使用QC来避免使用Lock，同时因为QC已经表明某一阶段“共识”在里面，所以避免了所有验证者节点都进行广播的场景，而只是与leader进行消息的沟通，这样使消息的传输复杂度从O(n^2)变成了O(n)，也就是这里说的线性试图切换。

两阶段投票+QC为什么不行

看过Tendermint共识算法可能第一反应是，为什么加了QC之后，Hotstuff共识算法不能是两阶段投票+QC，而是变成三阶段投票呢？

如果是两阶段投票，那么Precommit阶段之后区块就已经被锁定而可以被提交了，那么如果在Precommit阶段有一部分节点收到了主节点发来的 prepare QC，另一些节点没有收到，那么view-change之后，如果新的leader在新view上发送一个新的区块，而忽略上一个区块，那么那部分已经收到了主节点发来的 prepare QC，就会出现了二义性，不知道该认同哪一个了。

如果你仔细思考的话，其实三阶段投票也会有这个问题？这里Hotstuff使用Highest QC的方式来解决，见下文。

什么是Highest QC

其实就是该节点能够“看”到的最长链，如果在Commit阶段，已经收到了Precommit QC，那么该节点的Highest QC就会更新到该区块。新的leader在开始新的共识的时候，会收集验证者节点提供的Highest QC，而新的leader在收集超过2f+1个验证者报告的 Highest QC，从收集的 QC 中选出“最高”的 QC（最大视图号或高度），并以选中的 QC 对应区块为父区块，构建新提案。

这样的做法类似于一种Lock，但与Tendermint不同的是这里不用传递复杂的Lock消息，验证者节点只需要传递自己的Highest QC即可。而新的leader任务也变得简单，只需要在这些收集到的Highest QC，选择一个 “最高”的就可以。

Pipline 模式

因为这里各个阶段投票都是类似，那么在进行每个阶段的时候，就可以“捎带”下一个view的投票信息，因为有三个投票阶段，那么就可以有三个view共识在“流水”进行。

图片来源：https://blog.csdn.net/ganzr/article/details/110879463

简单代码实现

来自AI，供学习使用

import time
import threading
from collections import defaultdict

# 参数设置
N = 4  # 节点数 (N=3f+1)
F = 1  # 最大容忍拜占庭节点数
QUORUM = 2 * F + 1  # 法定人数

# 共识阶段
PHASES = ['Prepare', 'PreCommit', 'Commit']

class Block:
    def __init__(self, parent_qc, height, proposer, data):
        self.parent_qc = parent_qc  # 父QC，None表示创世块
        self.height = height
        self.proposer = proposer
        self.data = data
        self.qc = None  # 该区块对应的QC

    def __repr__(self):
        return f"Block(h={self.height}, proposer={self.proposer})"

class QC:
    def __init__(self, block, phase, signatures):
        self.block = block
        self.phase = phase
        self.signatures = signatures  # 节点ID列表表示签名

    def __repr__(self):
        return f"QC(phase={self.phase}, block_height={self.block.height}, sigs={len(self.signatures)})"

class Message:
    def __init__(self, sender, block, phase, qc):
        self.sender = sender
        self.block = block
        self.phase = phase
        self.qc = qc  # 之前阶段的QC作为安全引用

class Node:
    def __init__(self, node_id, network):
        self.node_id = node_id
        self.network = network
        self.locked_qc = None  # 锁定的QC，保证安全
        self.highest_qc = None  # 见过的最高QC，用于视图切换
        self.current_view = 0
        self.timeout = 5  # 超时时间秒
        self.timer = None
        self.leader_id = None
        self.phase_votes = defaultdict(lambda: defaultdict(set))  # block => phase => set(node_id)
        self.committed_blocks = []
        self.running = True

    def start(self):
        threading.Thread(target=self.run, daemon=True).start()

    def run(self):
        while self.running:
            if self.is_leader():
                self.propose_block()
            self.reset_timer()
            time.sleep(1)

    def reset_timer(self):
        if self.timer:
            self.timer.cancel()
        self.timer = threading.Timer(self.timeout, self.on_timeout)
        self.timer.start()

    def on_timeout(self):
        print(f"[Node {self.node_id}] Timeout in view {self.current_view}, trigger view change")
        self.current_view += 1
        self.leader_id = self.network.get_leader(self.current_view)
        # 视图切换时上报最高QC
        self.network.broadcast_view_change(self.node_id, self.highest_qc, self.current_view)

    def is_leader(self):
        return self.node_id == self.network.get_leader(self.current_view)

    def propose_block(self):
        # 领导者构造新区块
        parent_qc = self.highest_qc or self.network.genesis_qc
        height = parent_qc.block.height + 1
        block = Block(parent_qc, height, self.node_id, f"txs at height {height}")
        print(f"[Leader {self.node_id}] Proposing {block} in view {self.current_view}")

        # 领导者广播 Prepare 消息
        msg = Message(self.node_id, block, 'Prepare', parent_qc)
        self.network.broadcast_message(msg)

        # 等待投票形成 Prepare QC（简化：直接同步收集）
        prepare_qc = self.collect_votes(block, 'Prepare')
        if not prepare_qc:
            print(f"[Leader {self.node_id}] Failed to form Prepare QC")
            return

        # PreCommit 阶段
        msg = Message(self.node_id, block, 'PreCommit', prepare_qc)
        self.network.broadcast_message(msg)
        precommit_qc = self.collect_votes(block, 'PreCommit')
        if not precommit_qc:
            print(f"[Leader {self.node_id}] Failed to form PreCommit QC")
            return

        # Commit 阶段
        msg = Message(self.node_id, block, 'Commit', precommit_qc)
        self.network.broadcast_message(msg)
        commit_qc = self.collect_votes(block, 'Commit')
        if not commit_qc:
            print(f"[Leader {self.node_id}] Failed to form Commit QC")
            return

        # 决定阶段
        self.commit_block(block, commit_qc)

    def collect_votes(self, block, phase):
        # 简化：等待节点回复投票消息并收集
        timeout = 3
        votes = set()
        start = time.time()
        while time.time() - start < timeout:
            votes = self.phase_votes[block][phase]
            if len(votes) >= QUORUM:
                qc = QC(block, phase, list(votes))
                # 更新最高QC
                if (self.highest_qc is None) or (qc.block.height > self.highest_qc.block.height):
                    self.highest_qc = qc
                return qc
            time.sleep(0.1)
        return None

    def receive_message(self, msg):
        # 验证消息合法性（简化，不做密码学验证）
        # 检查是否延续最高QC安全链
        parent_height = msg.qc.block.height if msg.qc else 0
        if msg.block.height != parent_height + 1:
            print(f"[Node {self.node_id}] Reject {msg.phase} msg with bad height")
            return

        # 模拟投票
        self.vote(msg.block, msg.phase)

    def vote(self, block, phase):
        # 模拟投票延迟
        time.sleep(0.1)

        # 投票只能对比锁定QC高度，不能回退
        if self.locked_qc and block.parent_qc.block.height < self.locked_qc.block.height:
            print(f"[Node {self.node_id}] Vote rejected due to locked QC")
            return

        # 记录投票
        self.phase_votes[block][phase].add(self.node_id)

        # 锁定 QC，确保安全
        if phase == 'PreCommit':
            self.locked_qc = QC(block, phase, [self.node_id])

    def commit_block(self, block, qc):
        print(f"[Node {self.node_id}] Commit {block} with {qc}")
        self.committed_blocks.append(block)
        self.locked_qc = qc
        self.highest_qc = qc

class Network:
    def __init__(self):
        self.nodes = {}
        self.genesis_block = Block(None, 0, -1, "Genesis")
        self.genesis_qc = QC(self.genesis_block, 'Commit', [])
        self.lock = threading.Lock()
        self.view_change_votes = defaultdict(list)  # view => list of (node_id, highest_qc)

    def add_node(self, node):
        self.nodes[node.node_id] = node

    def get_leader(self, view):
        # 简单轮换leader
        return view % N

    def broadcast_message(self, msg):
        # 模拟网络广播
        for node in self.nodes.values():
            if node.node_id != msg.sender:
                threading.Thread(target=node.receive_message, args=(msg,), daemon=True).start()

    def broadcast_view_change(self, node_id, highest_qc, view):
        with self.lock:
            self.view_change_votes[view].append((node_id, highest_qc))
        print(f"[Network] Node {node_id} reports view change to view {view} with QC height "
              f"{highest_qc.block.height if highest_qc else 'None'}")

        # 简单实现：当收到超过2f+1节点视图切换报告，切换新leader
        with self.lock:
            votes = self.view_change_votes[view]
            if len(votes) >= QUORUM:
                leader = self.get_leader(view)
                print(f"[Network] View {view} activated. New leader is Node {leader}")
                # 广播新视图信息(可以扩展)

def main():
    network = Network()
    nodes = [Node(i, network) for i in range(N)]
    for n in nodes:
        network.add_node(n)

    # 设置初始leader
    for n in nodes:
        n.leader_id = network.get_leader(n.current_view)

    # 启动所有节点
    for n in nodes:
        n.start()

    # 运行一段时间演示
    try:
        time.sleep(30)
    except KeyboardInterrupt:
        pass

    # 停止节点
    for n in nodes:
        n.running = False

    print("Committed blocks at nodes:")
    for n in nodes:
        print(f"Node {n.node_id}: {[b.height for b in n.committed_blocks]}")

if __name__ == "__main__":
    main()

本作品采用知识共享署名 4.0 国际许可协议进行许可