```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) 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, best_justified_checkpoint=justified_checkpoint, blocks={anchor_root: copy(anchor_block)}, block_states={anchor_root: copy(anchor_state)}, checkpoint_states={justified_checkpoint: copy(anchor_state)}, ) ``` ```python def get_slots_since_genesis(store: Store) -> int: return (store.time - store.genesis_time) // SECONDS_PER_SLOT ``` ```python def get_current_slot(store: Store) -> Slot: return Slot(GENESIS_SLOT + get_slots_since_genesis(store)) ``` ```python def compute_slots_since_epoch_start(slot: Slot) -> int: return slot - compute_start_slot_at_epoch(compute_epoch_at_slot(slot)) ``` ```python def get_ancestor(store: Store, root: Root, slot: Slot) -> Root: block = store.blocks[root] if block.slot > slot: return get_ancestor(store, block.parent_root, slot) elif block.slot == slot: return root else: # root is older than queried slot, thus a skip slot. Return most recent root prior to slot return root ``` ```python def get_latest_attesting_balance(store: Store, root: Root) -> Gwei: state = store.checkpoint_states[store.justified_checkpoint] active_indices = get_active_validator_indices(state, get_current_epoch(state)) return Gwei(sum( state.validators[i].effective_balance for i in active_indices if (i in store.latest_messages and get_ancestor(store, store.latest_messages[i].root, store.blocks[root].slot) == root) )) ``` ```python def filter_block_tree(store: Store, block_root: Root, blocks: Dict[Root, BeaconBlock]) -> bool: block = store.blocks[block_root] children = [ root for root in store.blocks.keys() if store.blocks[root].parent_root == block_root ] # If any children branches contain expected finalized/justified checkpoints, # add to filtered block-tree and signal viability to parent. if any(children): filter_block_tree_result = [filter_block_tree(store, child, blocks) for child in children] if any(filter_block_tree_result): blocks[block_root] = block return True return False # If leaf block, check finalized/justified checkpoints as matching latest. head_state = store.block_states[block_root] correct_justified = ( store.justified_checkpoint.epoch == GENESIS_EPOCH or head_state.current_justified_checkpoint == store.justified_checkpoint ) correct_finalized = ( store.finalized_checkpoint.epoch == GENESIS_EPOCH or head_state.finalized_checkpoint == store.finalized_checkpoint ) # If expected finalized/justified, add to viable block-tree and signal viability to parent. if correct_justified and correct_finalized: blocks[block_root] = block return True # Otherwise, branch not viable return False ``` ```python def get_filtered_block_tree(store: Store) -> Dict[Root, BeaconBlock]: """ Retrieve a filtered block tree from ``store``, only returning branches whose leaf state's justified/finalized info agrees with that in ``store``. """ base = store.justified_checkpoint.root blocks: Dict[Root, BeaconBlock] = {} filter_block_tree(store, base, blocks) return blocks ``` ```python def get_head(store: Store) -> Root: # Get filtered block tree that only includes viable branches blocks = get_filtered_block_tree(store) # Execute the LMD-GHOST fork choice head = store.justified_checkpoint.root while True: children = [ root for root in blocks.keys() if blocks[root].parent_root == head ] if len(children) == 0: return head # Sort by latest attesting balance with ties broken lexicographically head = max(children, key=lambda root: (get_latest_attesting_balance(store, root), root)) ``` ```python def should_update_justified_checkpoint(store: Store, new_justified_checkpoint: Checkpoint) -> bool: """ To address the bouncing attack, only update conflicting justified checkpoints in the fork choice if in the early slots of the epoch. Otherwise, delay incorporation of new justified checkpoint until next epoch boundary. See https://ethresear.ch/t/prevention-of-bouncing-attack-on-ffg/6114 for more detailed analysis and discussion. """ if compute_slots_since_epoch_start(get_current_slot(store)) < SAFE_SLOTS_TO_UPDATE_JUSTIFIED: return True justified_slot = compute_start_slot_at_epoch(store.justified_checkpoint.epoch) if not get_ancestor(store, new_justified_checkpoint.root, justified_slot) == store.justified_checkpoint.root: return False return True ``` ```python def validate_on_attestation(store: Store, attestation: Attestation) -> None: target = attestation.data.target # Attestations must be from the current or previous epoch current_epoch = compute_epoch_at_slot(get_current_slot(store)) # Use GENESIS_EPOCH for previous when genesis to avoid underflow previous_epoch = current_epoch - 1 if current_epoch > GENESIS_EPOCH else GENESIS_EPOCH # If attestation target is from a future epoch, delay consideration until the epoch arrives assert target.epoch in [current_epoch, previous_epoch] assert target.epoch == compute_epoch_at_slot(attestation.data.slot) # Attestations target be for a known block. If target block is unknown, delay consideration until the block is found assert target.root in store.blocks # Attestations must be for a known block. If block is unknown, delay consideration until the block is found assert attestation.data.beacon_block_root in store.blocks # Attestations must not be for blocks in the future. If not, the attestation should not be considered assert store.blocks[attestation.data.beacon_block_root].slot <= attestation.data.slot # LMD vote must be consistent with FFG vote target target_slot = compute_start_slot_at_epoch(target.epoch) assert target.root == get_ancestor(store, attestation.data.beacon_block_root, target_slot) # Attestations can only affect the fork choice of subsequent slots. # Delay consideration in the fork choice until their slot is in the past. assert get_current_slot(store) >= attestation.data.slot + 1 ``` ```python def store_target_checkpoint_state(store: Store, target: Checkpoint) -> None: # Store target checkpoint state if not yet seen if target not in store.checkpoint_states: base_state = copy(store.block_states[target.root]) if base_state.slot < compute_start_slot_at_epoch(target.epoch): process_slots(base_state, compute_start_slot_at_epoch(target.epoch)) store.checkpoint_states[target] = base_state ``` ```python def update_latest_messages(store: Store, attesting_indices: Sequence[ValidatorIndex], attestation: Attestation) -> None: target = attestation.data.target beacon_block_root = attestation.data.beacon_block_root for i in attesting_indices: if i not in store.latest_messages or target.epoch > store.latest_messages[i].epoch: store.latest_messages[i] = LatestMessage(epoch=target.epoch, root=beacon_block_root) ``` ```python def on_tick(store: Store, time: uint64) -> None: previous_slot = get_current_slot(store) # update store time store.time = time current_slot = get_current_slot(store) # Not a new epoch, return if not (current_slot > previous_slot and compute_slots_since_epoch_start(current_slot) == 0): return # Update store.justified_checkpoint if a better checkpoint is known if store.best_justified_checkpoint.epoch > store.justified_checkpoint.epoch: store.justified_checkpoint = store.best_justified_checkpoint ``` ```python def on_block(store: Store, signed_block: SignedBeaconBlock) -> None: block = signed_block.message # Parent block must be known assert block.parent_root in store.block_states # Make a copy of the state to avoid mutability issues pre_state = copy(store.block_states[block.parent_root]) # Blocks cannot be in the future. If they are, their consideration must be delayed until the are in the past. assert get_current_slot(store) >= block.slot # Check that block is later than the finalized epoch slot (optimization to reduce calls to get_ancestor) finalized_slot = compute_start_slot_at_epoch(store.finalized_checkpoint.epoch) assert block.slot > finalized_slot # Check block is a descendant of the finalized block at the checkpoint finalized slot assert get_ancestor(store, block.parent_root, finalized_slot) == store.finalized_checkpoint.root # Check the block is valid and compute the post-state state = pre_state.copy() state_transition(state, signed_block, True) # Add new block to the store store.blocks[hash_tree_root(block)] = block # Add new state for this block to the store store.block_states[hash_tree_root(block)] = state # Update justified checkpoint if state.current_justified_checkpoint.epoch > store.justified_checkpoint.epoch: if state.current_justified_checkpoint.epoch > store.best_justified_checkpoint.epoch: store.best_justified_checkpoint = state.current_justified_checkpoint if should_update_justified_checkpoint(store, state.current_justified_checkpoint): store.justified_checkpoint = state.current_justified_checkpoint # Update finalized checkpoint if state.finalized_checkpoint.epoch > store.finalized_checkpoint.epoch: store.finalized_checkpoint = state.finalized_checkpoint # Potentially update justified if different from store if store.justified_checkpoint != state.current_justified_checkpoint: # Update justified if new justified is later than store justified if state.current_justified_checkpoint.epoch > store.justified_checkpoint.epoch: store.justified_checkpoint = state.current_justified_checkpoint return # Update justified if store justified is not in chain with finalized checkpoint finalized_slot = compute_start_slot_at_epoch(store.finalized_checkpoint.epoch) ancestor_at_finalized_slot = get_ancestor(store, store.justified_checkpoint.root, finalized_slot) if ancestor_at_finalized_slot != store.finalized_checkpoint.root: store.justified_checkpoint = state.current_justified_checkpoint ``` ```python def on_attestation(store: Store, attestation: Attestation) -> None: """ Run ``on_attestation`` upon receiving a new ``attestation`` from either within a block or directly on the wire. An ``attestation`` that is asserted as invalid may be valid at a later time, consider scheduling it for later processing in such case. """ validate_on_attestation(store, attestation) store_target_checkpoint_state(store, attestation.data.target) # Get state at the `target` to fully validate attestation target_state = store.checkpoint_states[attestation.data.target] indexed_attestation = get_indexed_attestation(target_state, attestation) assert is_valid_indexed_attestation(target_state, indexed_attestation) # Update latest messages for attesting indices update_latest_messages(store, indexed_attestation.attesting_indices, attestation) ```