EIP-7732 -- 分叉选择

fradamt 发布于 2025-08-19 阅读 21

EIP-7732 对以太坊分叉选择算法进行了重大修改，引入了 PayloadStatus（COMMITTED、EMPTY、FULL）区分共识块和执行负载状态，新增 PTC（Payload Timeliness Committee）投票机制，并通过 ForkChoiceNode 结构体在 LMD-GHOST 中同时考虑区块根和负载状态。

**注意**：本文档仍在开发中，面向研究人员和实现者。

- [介绍](#introduction)
- [常量](#constants)
- [容器](#containers)
  - [新的 `ForkChoiceNode`](#new-ForkChoiceNode)
- [辅助函数](#helpers)
  - [修改后的 `LatestMessage`](#modified-latestmessage)
  - [修改后的 `update_latest_messages`](#modified-update_latest_messages)
  - [修改后的 `Store`](#modified-store)
  - [修改后的 `get_forkchoice_store`](#modified-get_forkchoice_store)
  - [`notify_ptc_messages`](#notify_ptc_messages)
  - [`is_payload_present`](#is_payload_present)
  - [`is_parent_node_full`](#is_parent_node_full)
  - [修改后的 `get_ancestor`](#modified-get_ancestor)
  - [修改后的 `get_checkpoint_block`](#modified-get_checkpoint_block)
  - [`is_supporting_vote`](#is_supporting_vote)
  - [修改后的 `get_weight`](#modified-get_weight)
  - [修改后的 `get_head`](#modified-get_head)
- [更新后的分叉选择处理器](#updated-fork-choice-handlers)
  - [修改后的 `on_block`](#modified-on_block)
- [新的分叉选择处理器](#new-fork-choice-handlers)
  - [新的 `on_execution_payload`](#new-on_execution_payload)
  - [`seconds_into_slot`](#seconds_into_slot)
  - [修改后的 `on_tick_per_slot`](#modified-on_tick_per_slot)
  - [`on_payload_attestation_message`](#on_payload_attestation_message)
  - [修改后的 `validate_merge_block`](#modified-validate_merge_block)

## 介绍

这是伴随 EIP-7732 升级的分叉选择修改。

## 常量

| 名称                           | 值                            |
| ------------------------------ | ----------------------------- |
| `PAYLOAD_TIMELY_THRESHOLD`     | `PTC_SIZE // 2` (= 256)       |
| `INTERVALS_PER_SLOT`           | `4` # [在 EIP-7732 中修改]    |

## 容器

### 新的 `PayloadStatus`

```python
class PayloadStatus(IntEnum):
    COMMITTED = 0
    EMPTY = 1
    FULL = 2

```

### 新的 `ForkChoiceNode`

```python
class ForkChoiceNode(Container):
    root: Root
    payload_status: PayloadStatus
```

## 辅助函数

### 修改后的 `LatestMessage`

**注意**：该类已修改为跟踪 Slot 而不是 Epoch。

```python
@dataclass(eq=True, frozen=True)
class LatestMessage(object):
    slot: Slot
    root: Root
    payload_present: boolean
```

### 修改后的 `update_latest_messages`

**注意**：函数 `update_latest_messages` 已更新为使用 Attestation Slot 而不是 Target。请注意，此函数仅在已验证的 Attestation 上调用，且验证者不能在同一个 Epoch 内进行两次 Attestation（否则构成双重投票，Equivocating）。另请注意，Target Epoch 编号和 Slot 编号在 `validate_on_attestation` 中已通过验证。

```python
def update_latest_messages(
    store: Store, attesting_indices: Sequence[ValidatorIndex], attestation: Attestation
) -> None:
    slot = attestation.data.slot
    beacon_block_root = attestation.data.beacon_block_root
    payload_present = (attestation.data.index == 1)
    non_equivocating_attesting_indices = [
        i for i in attesting_indices if i not in store.equivocating_indices
    ]
    for i in non_equivocating_attesting_indices:
        if i not in store.latest_messages or slot > store.latest_messages[i].slot:
            store.latest_messages[i] = LatestMessage(
                slot=slot,
                root=beacon_block_root,
                payload_present=payload_present
            )
```

### 修改后的 `Store`

**注意**：`Store` 已修改为跟踪“空”共识区块的中间状态，即那些对应的 Execution Payload 尚未揭示或尚未上链的共识区块。

```python
@dataclass
class Store(object):
    time: uint64
    genesis_time: uint64
    justified_checkpoint: Checkpoint
    finalized_checkpoint: Checkpoint
    unrealized_justified_checkpoint: Checkpoint
    unrealized_finalized_checkpoint: Checkpoint
    proposer_boost_root: Root
    equivocating_indices: Set[ValidatorIndex]
    blocks: Dict[Root, BeaconBlock] = field(default_factory=dict)
    block_states: Dict[Root, BeaconState] = field(default_factory=dict)
    block_timeliness: Dict[Root, boolean] = field(default_factory=dict)
    checkpoint_states: Dict[Checkpoint, BeaconState] = field(default_factory=dict)
    latest_messages: Dict[ValidatorIndex, LatestMessage] = field(default_factory=dict)
    unrealized_justifications: Dict[Root, Checkpoint] = field(default_factory=dict)
    payload_states: Dict[Root, BeaconState] = field(default_factory=dict)  # [EIP-7732 中新增]
    ptc_vote: Dict[Root, Vector[boolean, PTC_SIZE]] = field(default_factory=dict)  # [EIP-7732 中新增]
```

### 修改后的 `get_forkchoice_store`

```python
def get_forkchoice_store(anchor_state: BeaconState, anchor_block: BeaconBlock) -> Store:
    assert anchor_block.state_root == hash_tree_root(anchor_state)
    anchor_root = hash_tree_root(anchor_block)
    anchor_epoch = get_current_epoch(anchor_state)
    justified_checkpoint = Checkpoint(epoch=anchor_epoch, root=anchor_root)
    finalized_checkpoint = Checkpoint(epoch=anchor_epoch, root=anchor_root)
    proposer_boost_root = Root()
    return Store(
        time=uint64(anchor_state.genesis_time + SECONDS_PER_SLOT * anchor_state.slot),
        genesis_time=anchor_state.genesis_time,
        justified_checkpoint=justified_checkpoint,
        finalized_checkpoint=finalized_checkpoint,
        unrealized_justified_checkpoint=justified_checkpoint,
        unrealized_finalized_checkpoint=finalized_checkpoint,
        proposer_boost_root=proposer_boost_root,
        equivocating_indices=set(),
        blocks={anchor_root: copy(anchor_block)},
        block_states={anchor_root: copy(anchor_state)},
        checkpoint_states={justified_checkpoint: copy(anchor_state)},
        unrealized_justifications={anchor_root: justified_checkpoint},
        payload_states={anchor_root: copy(anchor_state)},  # [EIP-7732 中新增]
        ptc_vote={anchor_root: Vector[boolean, PTC_SIZE]()},
    )
```

### `notify_ptc_messages`

```python
def notify_ptc_messages(
    store: Store, state: BeaconState, payload_attestations: Sequence[PayloadAttestation]
) -> None:
    """
    从 ``payload_attestations`` 中提取 ``PayloadAttestationMessage`` 列表并用它们更新 Store。
    这些 Payload Attestation 被假定已包含在信标区块中，因此无需签名验证。
    """
    if state.slot == 0:
        return
    for payload_attestation in payload_attestations:
        indexed_payload_attestation = get_indexed_payload_attestation(
            state, Slot(state.slot - 1), payload_attestation
        )
        for idx in indexed_payload_attestation.attesting_indices:
            on_payload_attestation_message(
                store,
                PayloadAttestationMessage(
                    validator_index=idx,
                    data=payload_attestation.data,
                    signature=BLSSignature(),
                    is_from_block=True,
                ),
            )
```

### `is_payload_present`

```python
def is_payload_present(store: Store, beacon_block_root: Root) -> bool:
    """
    返回具有 Root ``beacon_block_root`` 的信标区块的 Execution Payload 是否被 PTC 投票为存在，并且在本地被判定为可用。
    """
    # 信标区块 Root 必须是已知的
    assert beacon_block_root in store.ptc_vote
    return (
        sum(store.ptc_vote[beacon_block_root]) > PAYLOAD_TIMELY_THRESHOLD
        and beacon_block_root in store.payload_states
    )
```

### `get_parent_payload_status`

```python
def get_parent_payload_status(store: Store, block: BeaconBlock) -> PayloadStatus:
    parent = store.blocks[block.parent_root]
    parent_block_hash = block.body.signed_execution_payload_header.message.parent_block_hash
    message_block_hash = parent.body.signed_execution_payload_header.message.block_hash
    return PayloadStatus.FULL if parent_block_hash == message_block_hash else PayloadStatus.EMPTY
```

### `is_parent_node_full`

```python
def is_parent_node_full(store: Store, block: BeaconBlock) -> bool:
    return get_parent_payload_status(store, block) == PayloadStatus.FULL
```

### 修改后的 `get_ancestor`

**注意**：`get_ancestor` 已修改为返回链是基于 **空** 区块还是 **满** 区块。

```python
def get_ancestor(store: Store, root: Root, slot: Slot) -> ForkChoiceNode:
    """
    返回具有 ``root`` 的信标区块在 ``slot`` 处的祖先信标区块 Root 和 Payload 状态。
    如果具有 ``root`` 的信标区块已经处于 ``slot`` 或我们正在请求“未来”的祖先，则返回 ``PayloadStatus.COMMITTED``。
    """
    block = store.blocks[root]
    if block.slot <= slot:
        return ForkChoiceNode(root=root, payload_status=PayloadStatus.COMMITTED)

parent = store.blocks[block.parent_root]
    if parent.slot > slot:
        return get_ancestor(store, block.parent_root, slot)
    else: 
        return ForkChoiceNode(
        root=block.parent_root,
        payload_status=get_parent_payload_status(store, block),
    )
```

### 修改后的 `get_checkpoint_block`

**注意**：`get_checkpoint_block` 已修改为使用新的 `get_ancestor`。

```python
def get_checkpoint_block(store: Store, root: Root, epoch: Epoch) -> Root:
    """
    计算区块 ``root`` 链中 epoch ``epoch`` 的 Checkpoint 区块。
    """
    epoch_first_slot = compute_start_slot_at_epoch(epoch)
    return get_ancestor(store, root, epoch_first_slot).root
```

### `is_supporting_vote`

```python
def is_supporting_vote(store: Store, node: ForkChoiceNode, message: LatestMessage) -> bool:
    """
    返回对 ``message.root`` 的投票是否支持包含信标区块 ``node.root`` 的链，
    该链在 ``node`` 的 Slot 期间以 ``node.payload_status`` 指示的 Payload 内容作为 Head。
    """
    block = store.blocks[node.root]
    if node.root == message.root:
        if node.payload_status == PayloadStatus.COMMITTED: 
            return True
        if message.slot <= block.slot:
            return False
        if message.payload_present:
            return node.payload_status == PayloadStatus.FULL
        else:
            return node.payload_status == PayloadStatus.EMPTY

else:
        ancestor = get_ancestor(store, message.root, block.slot)
        return node.root == ancestor.root and (
            node.payload_status == PayloadStatus.COMMITTED  
            or node.payload_status == ancestor.payload_status
        )
```

### 修改后的 `get_weight`

```python
def get_weight(store: Store, node: ForkChoiceNode) -> Gwei:
    # 在决定前一个 Slot 的区块的 Payload 状态时，
    # 如果 PTC 的大多数成员为其投票，或者当前的 Proposer 基于该区块构建了区块，
    # 我们就认为 Payload 存在
    if (
        store.blocks[node.root].slot + 1 == get_current_slot(store)
        and node.payload_status != PayloadStatus.COMMITTED
    ):
        proposer_root = store.proposer_boost_root
        if is_payload_present(store, node.root) or (
            proposer_root != Root()
            and store.blocks[proposer_root].parent_root == node.root
            and is_parent_node_full(store, store.blocks[proposer_root])
        ):
            # 如果我们认为 Payload 存在，FULL 状态的权重
            # 将高于 EMPTY，反之亦然
            return Gwei(1) if node.payload_status == PayloadStatus.FULL else Gwei(0)
        else:
            return Gwei(1) if node.payload_status == PayloadStatus.EMPTY else Gwei(0)
    else:
        state = store.checkpoint_states[store.justified_checkpoint]
        unslashed_and_active_indices = [
            i
            for i in get_active_validator_indices(state, get_current_epoch(state))
            if not state.validators[i].slashed
        ]
        attestation_score = Gwei(
            sum(
                state.validators[i].effective_balance
                for i in unslashed_and_active_indices
                if (
                    i in store.latest_messages
                    and i not in store.equivocating_indices
                    and is_supporting_vote(store, node, store.latest_messages[i])
                )
            )
        )

if store.proposer_boost_root == Root():
            # 如果未设置 ``proposer_boost_root``，则仅返回 Attestation 分数
            return attestation_score

# 如果设置了 ``proposer_boost_root``，则计算 Proposer 分数
        proposer_score = Gwei(0)

# ``proposer_boost_root`` 被视为对当前 Slot 中
        # Proposer 区块的投票。Proposer Boost
        # 会相应地应用于所有祖先
        message = LatestMessage(
            slot=get_current_slot(store),
            root=store.proposer_boost_root,
            payload_present=False,
        )
        if is_supporting_vote(store, node, message):
            proposer_score = get_proposer_score(store)

return attestation_score + proposer_score
```

### 新的 `get_fork_choice_children`

```python
def get_fork_choice_children(store: Store, blocks: Dict[Root, BeaconBlock], node: ForkChoiceNode) -> Sequence[ForkChoiceNode]:
    if node.payload_status == PayloadStatus.COMMITTED:
        return [
        ForkChoiceNode(root=node.root, payload_status=payload_status) 
        for payload_status in (PayloadStatus.FULL, PayloadStatus.EMPTY)
        ]
    else:
        return [
            ForkChoiceNode(root=root, payload_status=PayloadStatus.COMMITTED) 
            for root in blocks.keys()
            if (
                blocks[root].parent_root == node.root
                and node.payload_status == get_parent_payload_status(store, blocks[root])
            )
        ]
```

### 修改后的 `get_head`

**注意**：`get_head` 已修改为使用新的 `get_weight` 函数。它返回对应于 Head 区块的 `ForkChoiceNode` 对象。

```python
def get_head(store: Store) -> ForkChoiceNode:
    # 获取仅包含可行分支的过滤后的区块树
    blocks = get_filtered_block_tree(store)
    # 执行 LMD-GHOST 分叉选择
    head = ForkChoiceNode(
        root=store.justified_checkpoint.root,
        payload_status=PayloadStatus.COMMITTED,
    )
    
    while True:
        children = get_fork_choice_children(store, blocks, head)
        if len(children) == 0:
            return head
        # 按最新的投票余额排序，平局时按字典顺序打破
        head = max(children, key=lambda child: (get_weight(store, child), child.root, child.payload_status))
```

## 更新后的分叉选择处理器

### 修改后的 `on_block`

**注意**：处理器 `on_block` 已修改为不仅根据父区块 Root，还根据父区块 Hash 来考虑给定共识信标区块的 Pre state。此外，我们将 Blob 数据可用性的检查延迟到 Execution Payload 的处理过程中。

```python
def on_block(store: Store, signed_block: SignedBeaconBlock) -> None:
    """
    收到新区块时运行 ``on_block``。
    """
    block = signed_block.message
    # 父区块必须是已知的
    assert block.parent_root in store.block_states

# 检查此区块是构建在空的还是满的父区块上
    parent_block = store.blocks[block.parent_root]
    header = block.body.signed_execution_payload_header.message
    parent_header = parent_block.body.signed_execution_payload_header.message
    # 制作状态副本以避免可变性问题
    if is_parent_node_full(store, block):
        assert block.parent_root in store.payload_states
        state = copy(store.payload_states[block.parent_root])
    else:
        assert header.parent_block_hash == parent_header.parent_block_hash
        state = copy(store.block_states[block.parent_root])

# 区块不能处于未来。如果是，必须延迟考虑它们，直到它们处于过去。
    current_slot = get_current_slot(store)
    assert current_slot >= block.slot

# 检查区块是否晚于已 Finalized 的 Epoch Slot（减少调用 get_ancestor 的优化）
    finalized_slot = compute_start_slot_at_epoch(store.finalized_checkpoint.epoch)
    assert block.slot > finalized_slot
    # 检查区块是否是 Checkpoint Finalized Slot 处已 Finalized 区块的后代
    finalized_checkpoint_block = get_checkpoint_block(
        store,
        block.parent_root,
        store.finalized_checkpoint.epoch,
    )
    assert store.finalized_checkpoint.root == finalized_checkpoint_block

# 检查区块是否有效并计算 Post-state
    block_root = hash_tree_root(block)
    state_transition(state, signed_block, True)

# 将新区块添加到 Store
    store.blocks[block_root] = block
    # 为此区块向 Store 添加新状态
    store.block_states[block_root] = state
    # 为此区块向 Store 添加新的 PTC 投票
    store.ptc_vote[block_root] = [False] * PTC_SIZE

# 通知 Store 关于区块中的 payload_attestations
    notify_ptc_messages(store, state, block.body.payload_attestations)
    # 如果区块及时，则添加 Proposer 分数 Boost
    time_into_slot = (store.time - store.genesis_time) % SECONDS_PER_SLOT
    is_before_attesting_interval = time_into_slot < SECONDS_PER_SLOT // INTERVALS_PER_SLOT
    is_timely = get_current_slot(store) == block.slot and is_before_attesting_interval
    store.block_timeliness[hash_tree_root(block)] = is_timely

# 如果区块及时且不与现有区块冲突，则添加 Proposer 分数 Boost
    is_first_block = store.proposer_boost_root == Root()
    if is_timely and is_first_block:
        store.proposer_boost_root = hash_tree_root(block)

# 必要时更新 Store 中的 Checkpoint
    update_checkpoints(store, state.current_justified_checkpoint, state.finalized_checkpoint)

# 提前计算未实现的 Justification 和 Finality。
    compute_pulled_up_tip(store, block_root)
```

## 新的分叉选择处理器

### 新的 `on_execution_payload`

处理器 `on_execution_payload` 在节点收到用于同步的 `SignedExecutionPayloadEnvelope` 时被调用。

```python
def on_execution_payload(store: Store, signed_envelope: SignedExecutionPayloadEnvelope) -> None:
    """
    收到新 Execution Payload 时运行 ``on_execution_payload``。
    """
    envelope = signed_envelope.message
    # 相应的信标区块 Root 需要是已知的
    assert envelope.beacon_block_root in store.block_states

# 检查 Blob 数据是否可用
    # 如果不可用，此 Payload 可能会被排队，并在 Blob 数据可用时随后被考虑
    assert is_data_available(envelope.beacon_block_root, envelope.blob_kzg_commitments)

# 制作状态副本以避免可变性问题
    state = copy(store.block_states[envelope.beacon_block_root])

# 处理 Execution Payload
    process_execution_payload(state, signed_envelope, EXECUTION_ENGINE)

# 为此 Payload 向 Store 添加新状态
    store.payload_states[envelope.beacon_block_root] = state
```

### `on_payload_attestation_message`

```python
def on_payload_attestation_message(
    store: Store, ptc_message: PayloadAttestationMessage, is_from_block: bool = False
) -> None:
    """
    直接从网络收到新的 ``ptc_message`` 时运行 ``on_payload_attestation_message``。
    """
    # 信标区块 Root 必须是已知的
    data = ptc_message.data
    # PTC Attestation 必须针对已知区块。如果区块未知，延迟考虑直到找到该区块
    state = store.block_states[data.beacon_block_root]
    ptc = get_ptc(state, data.slot)
    # PTC 投票只能更改其分配的信标区块的投票，否则提前返回
    if data.slot != state.slot:
        return
    # 检查投票者是否来自 PTC
    assert ptc_message.validator_index in ptc

# 验证签名，并检查如果它来自网络，它是否针对当前 Slot
    if not is_from_block:
        # 检查 Attestation 是否针对当前 Slot
        assert data.slot == get_current_slot(store)
        # 验证签名
        assert is_valid_indexed_payload_attestation(
            state,
            IndexedPayloadAttestation(
                attesting_indices=[ptc_message.validator_index],
                data=data,
                signature=ptc_message.signature,
            ),
        )
    # 更新区块的 PTC 投票
    ptc_index = ptc.index(ptc_message.validator_index)
    ptc_vote = store.ptc_vote[data.beacon_block_root]
    ptc_vote[ptc_index] = data.payload_present

# 仅当区块针对当前 Slot 且处于早期时，才使用来自区块的 Attestation 更新 Payload Boost
    if is_from_block and data.slot + 1 != get_current_slot(store):
        return
    time_into_slot = (store.time - store.genesis_time) % SECONDS_PER_SLOT
    if is_from_block and time_into_slot >= SECONDS_PER_SLOT // INTERVALS_PER_SLOT:
        return
```

### `validate_on_attestation`

```python
def validate_on_attestation(store: Store, attestation: Attestation, is_from_block: bool) -> None:
    target = attestation.data.target

# 如果给定的 Attestation 不是来自信标区块消息，我们必须检查 Target Epoch 范围。
    if not is_from_block:
        validate_target_epoch_against_current_time(store, attestation)

# 检查 Epoch 编号和 Slot 编号是否匹配
    assert target.epoch == compute_epoch_at_slot(attestation.data.slot)

# Attestation Target 必须针对已知区块。如果 Target 区块未知，延迟考虑直到找到该区块
    assert target.root in store.blocks

# Attestation 必须针对已知区块。如果区块未知，延迟考虑直到找到该区块
    assert attestation.data.beacon_block_root in store.blocks
    # Attestation 不得针对未来的区块。如果不是，则不应考虑该 Attestation
    slot = store.blocks[attestation.data.beacon_block_root].slot
    assert slot <= attestation.data.slot

# [EIP-7732 中新增] 
    assert attestation.data.index in [0,1]
    if slot == attestation.data.slot:
        assert attestation.data.index == 0

# LMD 投票必须与 FFG 投票 Target 一致
    assert target.root == get_checkpoint_block(
        store, attestation.data.beacon_block_root, target.epoch
    )

# Attestation 只能影响后续 Slot 的分叉选择。
    # 延迟在分叉选择中的考虑，直到它们的 Slot 处于过去。
    assert get_current_slot(store) >= attestation.data.slot + 1
```

### 修改后的 `validate_merge_block`

函数 `validate_merge_block` 已针对测试目的进行了修改

```python
def validate_merge_block(block: BeaconBlock) -> None:
    """
    检查 Execution Payload 的父 PoW 区块是否是有效的终端 PoW 区块。

注意：不可用的 PoW 区块稍后可能会变得可用，
    客户端软件可以延迟调用 ``validate_merge_block``，
    直到 PoW 区块变得可用。
    """
    if TERMINAL_BLOCK_HASH != Hash32():
        # 如果使用 `TERMINAL_BLOCK_HASH` 作为覆盖，则必须达到激活 Epoch。
        assert compute_epoch_at_slot(block.slot) >= TERMINAL_BLOCK_HASH_ACTIVATION_EPOCH
        assert (
            block.body.signed_execution_payload_header.message.parent_block_hash
            == TERMINAL_BLOCK_HASH
        )
        return

pow_block = get_pow_block(block.body.signed_execution_payload_header.message.parent_block_hash)
    # 检查 `pow_block` 是否可用
    assert pow_block is not None
    pow_parent = get_pow_block(pow_block.parent_hash)
    # 检查 `pow_parent` 是否可用
    assert pow_parent is not None
    # 检查 `pow_block` 是否是有效的终端 PoW 区块
    assert is_valid_terminal_pow_block(pow_block, pow_parent)
```

>- 原文链接： [github.com/fradamt/conse...](https://github.com/fradamt/consensus-specs/blob/epbs-payload-bit/specs/_features/eip7732/fork-choice.md)
>- 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～

注意：本文档仍在开发中，面向研究人员和实现者。

介绍
常量
容器
- 新的 ForkChoiceNode
辅助函数
更新后的分叉选择处理器
- 修改后的 on_block
新的分叉选择处理器

介绍

这是伴随 EIP-7732 升级的分叉选择修改。

常量

名称	值
`PAYLOAD_TIMELY_THRESHOLD`	`PTC_SIZE // 2` (= 256)
`INTERVALS_PER_SLOT`	`4` # [在 EIP-7732 中修改]

容器

新的 `PayloadStatus`

class PayloadStatus(IntEnum):
    COMMITTED = 0
    EMPTY = 1
    FULL = 2

新的 `ForkChoiceNode`

class ForkChoiceNode(Container):
    root: Root
    payload_status: PayloadStatus

辅助函数

修改后的 `LatestMessage`

注意：该类已修改为跟踪 Slot 而不是 Epoch。

@dataclass(eq=True, frozen=True)
class LatestMessage(object):
    slot: Slot
    root: Root
    payload_present: boolean

修改后的 `update_latest_messages`

注意：函数 update_latest_messages 已更新为使用 Attestation Slot 而不是 Target。请注意，此函数仅在已验证的 Attestation 上调用，且验证者不能在同一个 Epoch 内进行两次 Attestation（否则构成双重投票，Equivocating）。另请注意，Target Epoch 编号和 Slot 编号在 validate_on_attestation 中已通过验证。

def update_latest_messages(
    store: Store, attesting_indices: Sequence[ValidatorIndex], attestation: Attestation
) -> None:
    slot = attestation.data.slot
    beacon_block_root = attestation.data.beacon_block_root
    payload_present = (attestation.data.index == 1)
    non_equivocating_attesting_indices = [
        i for i in attesting_indices if i not in store.equivocating_indices
    ]
    for i in non_equivocating_attesting_indices:
        if i not in store.latest_messages or slot > store.latest_messages[i].slot:
            store.latest_messages[i] = LatestMessage(
                slot=slot,
                root=beacon_block_root,
                payload_present=payload_present
            )

修改后的 `Store`

注意：Store 已修改为跟踪“空”共识区块的中间状态，即那些对应的 Execution Payload 尚未揭示或尚未上链的共识区块。

@dataclass
class Store(object):
    time: uint64
    genesis_time: uint64
    justified_checkpoint: Checkpoint
    finalized_checkpoint: Checkpoint
    unrealized_justified_checkpoint: Checkpoint
    unrealized_finalized_checkpoint: Checkpoint
    proposer_boost_root: Root
    equivocating_indices: Set[ValidatorIndex]
    blocks: Dict[Root, BeaconBlock] = field(default_factory=dict)
    block_states: Dict[Root, BeaconState] = field(default_factory=dict)
    block_timeliness: Dict[Root, boolean] = field(default_factory=dict)
    checkpoint_states: Dict[Checkpoint, BeaconState] = field(default_factory=dict)
    latest_messages: Dict[ValidatorIndex, LatestMessage] = field(default_factory=dict)
    unrealized_justifications: Dict[Root, Checkpoint] = field(default_factory=dict)
    payload_states: Dict[Root, BeaconState] = field(default_factory=dict)  # [EIP-7732 中新增]
    ptc_vote: Dict[Root, Vector[boolean, PTC_SIZE]] = field(default_factory=dict)  # [EIP-7732 中新增]

修改后的 `get_forkchoice_store`

def get_forkchoice_store(anchor_state: BeaconState, anchor_block: BeaconBlock) -> Store:
    assert anchor_block.state_root == hash_tree_root(anchor_state)
    anchor_root = hash_tree_root(anchor_block)
    anchor_epoch = get_current_epoch(anchor_state)
    justified_checkpoint = Checkpoint(epoch=anchor_epoch, root=anchor_root)
    finalized_checkpoint = Checkpoint(epoch=anchor_epoch, root=anchor_root)
    proposer_boost_root = Root()
    return Store(
        time=uint64(anchor_state.genesis_time + SECONDS_PER_SLOT * anchor_state.slot),
        genesis_time=anchor_state.genesis_time,
        justified_checkpoint=justified_checkpoint,
        finalized_checkpoint=finalized_checkpoint,
        unrealized_justified_checkpoint=justified_checkpoint,
        unrealized_finalized_checkpoint=finalized_checkpoint,
        proposer_boost_root=proposer_boost_root,
        equivocating_indices=set(),
        blocks={anchor_root: copy(anchor_block)},
        block_states={anchor_root: copy(anchor_state)},
        checkpoint_states={justified_checkpoint: copy(anchor_state)},
        unrealized_justifications={anchor_root: justified_checkpoint},
        payload_states={anchor_root: copy(anchor_state)},  # [EIP-7732 中新增]
        ptc_vote={anchor_root: Vector[boolean, PTC_SIZE]()},
    )

`notify_ptc_messages`

def notify_ptc_messages(
    store: Store, state: BeaconState, payload_attestations: Sequence[PayloadAttestation]
) -> None:
    """
    从 ``payload_attestations`` 中提取 ``PayloadAttestationMessage`` 列表并用它们更新 Store。
    这些 Payload Attestation 被假定已包含在信标区块中，因此无需签名验证。
    """
    if state.slot == 0:
        return
    for payload_attestation in payload_attestations:
        indexed_payload_attestation = get_indexed_payload_attestation(
            state, Slot(state.slot - 1), payload_attestation
        )
        for idx in indexed_payload_attestation.attesting_indices:
            on_payload_attestation_message(
                store,
                PayloadAttestationMessage(
                    validator_index=idx,
                    data=payload_attestation.data,
                    signature=BLSSignature(),
                    is_from_block=True,
                ),
            )

`is_payload_present`

def is_payload_present(store: Store, beacon_block_root: Root) -> bool:
    """
    返回具有 Root ``beacon_block_root`` 的信标区块的 Execution Payload 是否被 PTC 投票为存在，并且在本地被判定为可用。
    """
    # 信标区块 Root 必须是已知的
    assert beacon_block_root in store.ptc_vote
    return (
        sum(store.ptc_vote[beacon_block_root]) > PAYLOAD_TIMELY_THRESHOLD
        and beacon_block_root in store.payload_states
    )

`get_parent_payload_status`

def get_parent_payload_status(store: Store, block: BeaconBlock) -> PayloadStatus:
    parent = store.blocks[block.parent_root]
    parent_block_hash = block.body.signed_execution_payload_header.message.parent_block_hash
    message_block_hash = parent.body.signed_execution_payload_header.message.block_hash
    return PayloadStatus.FULL if parent_block_hash == message_block_hash else PayloadStatus.EMPTY

`is_parent_node_full`

def is_parent_node_full(store: Store, block: BeaconBlock) -> bool:
    return get_parent_payload_status(store, block) == PayloadStatus.FULL

修改后的 `get_ancestor`

注意：get_ancestor 已修改为返回链是基于空区块还是满区块。

def get_ancestor(store: Store, root: Root, slot: Slot) -> ForkChoiceNode:
    """
    返回具有 ``root`` 的信标区块在 ``slot`` 处的祖先信标区块 Root 和 Payload 状态。
    如果具有 ``root`` 的信标区块已经处于 ``slot`` 或我们正在请求“未来”的祖先，则返回 ``PayloadStatus.COMMITTED``。
    """
    block = store.blocks[root]
    if block.slot &lt;= slot:
        return ForkChoiceNode(root=root, payload_status=PayloadStatus.COMMITTED)

    parent = store.blocks[block.parent_root]
    if parent.slot > slot:
        return get_ancestor(store, block.parent_root, slot)
    else: 
        return ForkChoiceNode(
        root=block.parent_root,
        payload_status=get_parent_payload_status(store, block),
    )

修改后的 `get_checkpoint_block`

注意：get_checkpoint_block 已修改为使用新的 get_ancestor。

def get_checkpoint_block(store: Store, root: Root, epoch: Epoch) -> Root:
    """
    计算区块 ``root`` 链中 epoch ``epoch`` 的 Checkpoint 区块。
    """
    epoch_first_slot = compute_start_slot_at_epoch(epoch)
    return get_ancestor(store, root, epoch_first_slot).root

`is_supporting_vote`

def is_supporting_vote(store: Store, node: ForkChoiceNode, message: LatestMessage) -> bool:
    """
    返回对 ``message.root`` 的投票是否支持包含信标区块 ``node.root`` 的链，
    该链在 ``node`` 的 Slot 期间以 ``node.payload_status`` 指示的 Payload 内容作为 Head。
    """
    block = store.blocks[node.root]
    if node.root == message.root:
        if node.payload_status == PayloadStatus.COMMITTED: 
            return True
        if message.slot &lt;= block.slot:
            return False
        if message.payload_present:
            return node.payload_status == PayloadStatus.FULL
        else:
            return node.payload_status == PayloadStatus.EMPTY

    else:
        ancestor = get_ancestor(store, message.root, block.slot)
        return node.root == ancestor.root and (
            node.payload_status == PayloadStatus.COMMITTED  
            or node.payload_status == ancestor.payload_status
        )

修改后的 `get_weight`

def get_weight(store: Store, node: ForkChoiceNode) -> Gwei:
    # 在决定前一个 Slot 的区块的 Payload 状态时，
    # 如果 PTC 的大多数成员为其投票，或者当前的 Proposer 基于该区块构建了区块，
    # 我们就认为 Payload 存在
    if (
        store.blocks[node.root].slot + 1 == get_current_slot(store)
        and node.payload_status != PayloadStatus.COMMITTED
    ):
        proposer_root = store.proposer_boost_root
        if is_payload_present(store, node.root) or (
            proposer_root != Root()
            and store.blocks[proposer_root].parent_root == node.root
            and is_parent_node_full(store, store.blocks[proposer_root])
        ):
            # 如果我们认为 Payload 存在，FULL 状态的权重
            # 将高于 EMPTY，反之亦然
            return Gwei(1) if node.payload_status == PayloadStatus.FULL else Gwei(0)
        else:
            return Gwei(1) if node.payload_status == PayloadStatus.EMPTY else Gwei(0)
    else:
        state = store.checkpoint_states[store.justified_checkpoint]
        unslashed_and_active_indices = [
            i
            for i in get_active_validator_indices(state, get_current_epoch(state))
            if not state.validators[i].slashed
        ]
        attestation_score = Gwei(
            sum(
                state.validators[i].effective_balance
                for i in unslashed_and_active_indices
                if (
                    i in store.latest_messages
                    and i not in store.equivocating_indices
                    and is_supporting_vote(store, node, store.latest_messages[i])
                )
            )
        )

        if store.proposer_boost_root == Root():
            # 如果未设置 ``proposer_boost_root``，则仅返回 Attestation 分数
            return attestation_score

        # 如果设置了 ``proposer_boost_root``，则计算 Proposer 分数
        proposer_score = Gwei(0)

        # ``proposer_boost_root`` 被视为对当前 Slot 中
        # Proposer 区块的投票。Proposer Boost
        # 会相应地应用于所有祖先
        message = LatestMessage(
            slot=get_current_slot(store),
            root=store.proposer_boost_root,
            payload_present=False,
        )
        if is_supporting_vote(store, node, message):
            proposer_score = get_proposer_score(store)

        return attestation_score + proposer_score

新的 `get_fork_choice_children`

def get_fork_choice_children(store: Store, blocks: Dict[Root, BeaconBlock], node: ForkChoiceNode) -> Sequence[ForkChoiceNode]:
    if node.payload_status == PayloadStatus.COMMITTED:
        return [
        ForkChoiceNode(root=node.root, payload_status=payload_status) 
        for payload_status in (PayloadStatus.FULL, PayloadStatus.EMPTY)
        ]
    else:
        return [
            ForkChoiceNode(root=root, payload_status=PayloadStatus.COMMITTED) 
            for root in blocks.keys()
            if (
                blocks[root].parent_root == node.root
                and node.payload_status == get_parent_payload_status(store, blocks[root])
            )
        ]

修改后的 `get_head`

注意：get_head 已修改为使用新的 get_weight 函数。它返回对应于 Head 区块的 ForkChoiceNode 对象。

def get_head(store: Store) -> ForkChoiceNode:
    # 获取仅包含可行分支的过滤后的区块树
    blocks = get_filtered_block_tree(store)
    # 执行 LMD-GHOST 分叉选择
    head = ForkChoiceNode(
        root=store.justified_checkpoint.root,
        payload_status=PayloadStatus.COMMITTED,
    )
    
    while True:
        children = get_fork_choice_children(store, blocks, head)
        if len(children) == 0:
            return head
        # 按最新的投票余额排序，平局时按字典顺序打破
        head = max(children, key=lambda child: (get_weight(store, child), child.root, child.payload_status))

更新后的分叉选择处理器

修改后的 `on_block`

注意：处理器 on_block 已修改为不仅根据父区块 Root，还根据父区块 Hash 来考虑给定共识信标区块的 Pre state。此外，我们将 Blob 数据可用性的检查延迟到 Execution Payload 的处理过程中。

def on_block(store: Store, signed_block: SignedBeaconBlock) -> None:
    """
    收到新区块时运行 ``on_block``。
    """
    block = signed_block.message
    # 父区块必须是已知的
    assert block.parent_root in store.block_states

    # 检查此区块是构建在空的还是满的父区块上
    parent_block = store.blocks[block.parent_root]
    header = block.body.signed_execution_payload_header.message
    parent_header = parent_block.body.signed_execution_payload_header.message
    # 制作状态副本以避免可变性问题
    if is_parent_node_full(store, block):
        assert block.parent_root in store.payload_states
        state = copy(store.payload_states[block.parent_root])
    else:
        assert header.parent_block_hash == parent_header.parent_block_hash
        state = copy(store.block_states[block.parent_root])

    # 区块不能处于未来。如果是，必须延迟考虑它们，直到它们处于过去。
    current_slot = get_current_slot(store)
    assert current_slot >= block.slot

    # 检查区块是否晚于已 Finalized 的 Epoch Slot（减少调用 get_ancestor 的优化）
    finalized_slot = compute_start_slot_at_epoch(store.finalized_checkpoint.epoch)
    assert block.slot > finalized_slot
    # 检查区块是否是 Checkpoint Finalized Slot 处已 Finalized 区块的后代
    finalized_checkpoint_block = get_checkpoint_block(
        store,
        block.parent_root,
        store.finalized_checkpoint.epoch,
    )
    assert store.finalized_checkpoint.root == finalized_checkpoint_block

    # 检查区块是否有效并计算 Post-state
    block_root = hash_tree_root(block)
    state_transition(state, signed_block, True)

    # 将新区块添加到 Store
    store.blocks[block_root] = block
    # 为此区块向 Store 添加新状态
    store.block_states[block_root] = state
    # 为此区块向 Store 添加新的 PTC 投票
    store.ptc_vote[block_root] = [False] * PTC_SIZE

    # 通知 Store 关于区块中的 payload_attestations
    notify_ptc_messages(store, state, block.body.payload_attestations)
    # 如果区块及时，则添加 Proposer 分数 Boost
    time_into_slot = (store.time - store.genesis_time) % SECONDS_PER_SLOT
    is_before_attesting_interval = time_into_slot &lt; SECONDS_PER_SLOT // INTERVALS_PER_SLOT
    is_timely = get_current_slot(store) == block.slot and is_before_attesting_interval
    store.block_timeliness[hash_tree_root(block)] = is_timely

    # 如果区块及时且不与现有区块冲突，则添加 Proposer 分数 Boost
    is_first_block = store.proposer_boost_root == Root()
    if is_timely and is_first_block:
        store.proposer_boost_root = hash_tree_root(block)

    # 必要时更新 Store 中的 Checkpoint
    update_checkpoints(store, state.current_justified_checkpoint, state.finalized_checkpoint)

    # 提前计算未实现的 Justification 和 Finality。
    compute_pulled_up_tip(store, block_root)

新的分叉选择处理器

新的 `on_execution_payload`

处理器 on_execution_payload 在节点收到用于同步的 SignedExecutionPayloadEnvelope 时被调用。

def on_execution_payload(store: Store, signed_envelope: SignedExecutionPayloadEnvelope) -> None:
    """
    收到新 Execution Payload 时运行 ``on_execution_payload``。
    """
    envelope = signed_envelope.message
    # 相应的信标区块 Root 需要是已知的
    assert envelope.beacon_block_root in store.block_states

    # 检查 Blob 数据是否可用
    # 如果不可用，此 Payload 可能会被排队，并在 Blob 数据可用时随后被考虑
    assert is_data_available(envelope.beacon_block_root, envelope.blob_kzg_commitments)

    # 制作状态副本以避免可变性问题
    state = copy(store.block_states[envelope.beacon_block_root])

    # 处理 Execution Payload
    process_execution_payload(state, signed_envelope, EXECUTION_ENGINE)

    # 为此 Payload 向 Store 添加新状态
    store.payload_states[envelope.beacon_block_root] = state

`on_payload_attestation_message`

def on_payload_attestation_message(
    store: Store, ptc_message: PayloadAttestationMessage, is_from_block: bool = False
) -> None:
    """
    直接从网络收到新的 ``ptc_message`` 时运行 ``on_payload_attestation_message``。
    """
    # 信标区块 Root 必须是已知的
    data = ptc_message.data
    # PTC Attestation 必须针对已知区块。如果区块未知，延迟考虑直到找到该区块
    state = store.block_states[data.beacon_block_root]
    ptc = get_ptc(state, data.slot)
    # PTC 投票只能更改其分配的信标区块的投票，否则提前返回
    if data.slot != state.slot:
        return
    # 检查投票者是否来自 PTC
    assert ptc_message.validator_index in ptc

    # 验证签名，并检查如果它来自网络，它是否针对当前 Slot
    if not is_from_block:
        # 检查 Attestation 是否针对当前 Slot
        assert data.slot == get_current_slot(store)
        # 验证签名
        assert is_valid_indexed_payload_attestation(
            state,
            IndexedPayloadAttestation(
                attesting_indices=[ptc_message.validator_index],
                data=data,
                signature=ptc_message.signature,
            ),
        )
    # 更新区块的 PTC 投票
    ptc_index = ptc.index(ptc_message.validator_index)
    ptc_vote = store.ptc_vote[data.beacon_block_root]
    ptc_vote[ptc_index] = data.payload_present

    # 仅当区块针对当前 Slot 且处于早期时，才使用来自区块的 Attestation 更新 Payload Boost
    if is_from_block and data.slot + 1 != get_current_slot(store):
        return
    time_into_slot = (store.time - store.genesis_time) % SECONDS_PER_SLOT
    if is_from_block and time_into_slot >= SECONDS_PER_SLOT // INTERVALS_PER_SLOT:
        return

`validate_on_attestation`

def validate_on_attestation(store: Store, attestation: Attestation, is_from_block: bool) -> None:
    target = attestation.data.target

    # 如果给定的 Attestation 不是来自信标区块消息，我们必须检查 Target Epoch 范围。
    if not is_from_block:
        validate_target_epoch_against_current_time(store, attestation)

    # 检查 Epoch 编号和 Slot 编号是否匹配
    assert target.epoch == compute_epoch_at_slot(attestation.data.slot)

    # Attestation Target 必须针对已知区块。如果 Target 区块未知，延迟考虑直到找到该区块
    assert target.root in store.blocks

    # Attestation 必须针对已知区块。如果区块未知，延迟考虑直到找到该区块
    assert attestation.data.beacon_block_root in store.blocks
    # Attestation 不得针对未来的区块。如果不是，则不应考虑该 Attestation
    slot = store.blocks[attestation.data.beacon_block_root].slot
    assert slot &lt;= attestation.data.slot

    # [EIP-7732 中新增] 
    assert attestation.data.index in [0,1]
    if slot == attestation.data.slot:
        assert attestation.data.index == 0

    # LMD 投票必须与 FFG 投票 Target 一致
    assert target.root == get_checkpoint_block(
        store, attestation.data.beacon_block_root, target.epoch
    )

    # Attestation 只能影响后续 Slot 的分叉选择。
    # 延迟在分叉选择中的考虑，直到它们的 Slot 处于过去。
    assert get_current_slot(store) >= attestation.data.slot + 1

修改后的 `validate_merge_block`

函数 validate_merge_block 已针对测试目的进行了修改

def validate_merge_block(block: BeaconBlock) -> None:
    """
    检查 Execution Payload 的父 PoW 区块是否是有效的终端 PoW 区块。

    注意：不可用的 PoW 区块稍后可能会变得可用，
    客户端软件可以延迟调用 ``validate_merge_block``，
    直到 PoW 区块变得可用。
    """
    if TERMINAL_BLOCK_HASH != Hash32():
        # 如果使用 `TERMINAL_BLOCK_HASH` 作为覆盖，则必须达到激活 Epoch。
        assert compute_epoch_at_slot(block.slot) >= TERMINAL_BLOCK_HASH_ACTIVATION_EPOCH
        assert (
            block.body.signed_execution_payload_header.message.parent_block_hash
            == TERMINAL_BLOCK_HASH
        )
        return

    pow_block = get_pow_block(block.body.signed_execution_payload_header.message.parent_block_hash)
    # 检查 `pow_block` 是否可用
    assert pow_block is not None
    pow_parent = get_pow_block(pow_block.parent_hash)
    # 检查 `pow_parent` 是否可用
    assert pow_parent is not None
    # 检查 `pow_block` 是否是有效的终端 PoW 区块
    assert is_valid_terminal_pow_block(pow_block, pow_parent)

原文链接： github.com/fradamt/conse...

登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～

0 条评论

介绍

常量

容器

新的 PayloadStatus

新的 ForkChoiceNode

辅助函数

修改后的 LatestMessage

修改后的 update_latest_messages

修改后的 Store

修改后的 get_forkchoice_store

notify_ptc_messages

is_payload_present

get_parent_payload_status

is_parent_node_full

修改后的 get_ancestor

修改后的 get_checkpoint_block

is_supporting_vote

修改后的 get_weight

新的 get_fork_choice_children

修改后的 get_head