mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2024-12-26 05:17:22 +00:00
f89fce9bd7
* add fork choice helpers * LMD GHOST basic implementation * add todos * add in the initial test and finalize rest of functions * added tests * more simulation, adding trivial case with only one option for a head * panics * begin helper function tests * more helper tests * higher vote wins * LMD Ghost full test * comment * add types * optimize * prepared * deposits for chain * lint * use the hash beacon block util * rem unnecessary db methods * comments * setup works
379 lines
12 KiB
Go
379 lines
12 KiB
Go
// Package blockchain defines the life-cycle and status of the beacon chain
|
|
// as well as the Ethereum Serenity beacon chain fork-choice rule based on
|
|
// Casper Proof of Stake finality.
|
|
package blockchain
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
gethTypes "github.com/ethereum/go-ethereum/core/types"
|
|
b "github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/core/state"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/db"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/powchain"
|
|
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
|
|
"github.com/prysmaticlabs/prysm/shared/event"
|
|
"github.com/prysmaticlabs/prysm/shared/hashutil"
|
|
"github.com/prysmaticlabs/prysm/shared/params"
|
|
"github.com/sirupsen/logrus"
|
|
)
|
|
|
|
var log = logrus.WithField("prefix", "blockchain")
|
|
|
|
// ChainService represents a service that handles the internal
|
|
// logic of managing the full PoS beacon chain.
|
|
type ChainService struct {
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
beaconDB *db.BeaconDB
|
|
web3Service *powchain.Web3Service
|
|
incomingBlockFeed *event.Feed
|
|
incomingBlockChan chan *pb.BeaconBlock
|
|
processedBlockChan chan *pb.BeaconBlock
|
|
canonicalBlockFeed *event.Feed
|
|
canonicalStateFeed *event.Feed
|
|
genesisTime time.Time
|
|
unProcessedBlocks map[uint64]*pb.BeaconBlock
|
|
unfinalizedBlocks map[[32]byte]*pb.BeaconState
|
|
enablePOWChain bool
|
|
}
|
|
|
|
// Config options for the service.
|
|
type Config struct {
|
|
BeaconBlockBuf int
|
|
IncomingBlockBuf int
|
|
Web3Service *powchain.Web3Service
|
|
BeaconDB *db.BeaconDB
|
|
DevMode bool
|
|
EnablePOWChain bool
|
|
}
|
|
|
|
// NewChainService instantiates a new service instance that will
|
|
// be registered into a running beacon node.
|
|
func NewChainService(ctx context.Context, cfg *Config) (*ChainService, error) {
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
return &ChainService{
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
beaconDB: cfg.BeaconDB,
|
|
web3Service: cfg.Web3Service,
|
|
incomingBlockChan: make(chan *pb.BeaconBlock, cfg.IncomingBlockBuf),
|
|
processedBlockChan: make(chan *pb.BeaconBlock),
|
|
incomingBlockFeed: new(event.Feed),
|
|
canonicalBlockFeed: new(event.Feed),
|
|
canonicalStateFeed: new(event.Feed),
|
|
unProcessedBlocks: make(map[uint64]*pb.BeaconBlock),
|
|
unfinalizedBlocks: make(map[[32]byte]*pb.BeaconState),
|
|
enablePOWChain: cfg.EnablePOWChain,
|
|
}, nil
|
|
}
|
|
|
|
// Start a blockchain service's main event loop.
|
|
func (c *ChainService) Start() {
|
|
log.Info("Starting service")
|
|
var err error
|
|
c.genesisTime, err = c.beaconDB.GenesisTime()
|
|
if err != nil {
|
|
log.Fatalf("Unable to retrieve genesis time - blockchain service could not start: %v", err)
|
|
return
|
|
}
|
|
// TODO(#675): Initialize unfinalizedBlocks map from disk in case this
|
|
// is a beacon node restarting.
|
|
go c.updateHead(c.processedBlockChan)
|
|
go c.blockProcessing(c.processedBlockChan)
|
|
}
|
|
|
|
// Stop the blockchain service's main event loop and associated goroutines.
|
|
func (c *ChainService) Stop() error {
|
|
defer c.cancel()
|
|
|
|
log.Info("Stopping service")
|
|
return nil
|
|
}
|
|
|
|
// Status always returns nil.
|
|
// TODO(1202): Add service health checks.
|
|
func (c *ChainService) Status() error {
|
|
return nil
|
|
}
|
|
|
|
// IncomingBlockFeed returns a feed that any service can send incoming p2p blocks into.
|
|
// The chain service will subscribe to this feed in order to process incoming blocks.
|
|
func (c *ChainService) IncomingBlockFeed() *event.Feed {
|
|
return c.incomingBlockFeed
|
|
}
|
|
|
|
// CanonicalBlockFeed returns a channel that is written to
|
|
// whenever a new block is determined to be canonical in the chain.
|
|
func (c *ChainService) CanonicalBlockFeed() *event.Feed {
|
|
return c.canonicalBlockFeed
|
|
}
|
|
|
|
// CanonicalStateFeed returns a feed that is written to
|
|
// whenever a new state is determined to be canonical in the chain.
|
|
func (c *ChainService) CanonicalStateFeed() *event.Feed {
|
|
return c.canonicalStateFeed
|
|
}
|
|
|
|
// doesPoWBlockExist checks if the referenced PoW block exists.
|
|
func (c *ChainService) doesPoWBlockExist(hash [32]byte) bool {
|
|
powBlock, err := c.web3Service.Client().BlockByHash(c.ctx, hash)
|
|
if err != nil {
|
|
log.Debugf("fetching PoW block corresponding to mainchain reference failed: %v", err)
|
|
return false
|
|
}
|
|
|
|
return powBlock != nil
|
|
}
|
|
|
|
// updateHead applies the fork choice rule to the beacon chain
|
|
// at the start of each new slot interval. The function looks
|
|
// at an in-memory slice of block hashes pending processing and
|
|
// selects the best block according to the in-protocol fork choice
|
|
// rule as canonical. This block is then persisted to storage.
|
|
func (c *ChainService) updateHead(processedBlock <-chan *pb.BeaconBlock) {
|
|
for {
|
|
select {
|
|
case <-c.ctx.Done():
|
|
return
|
|
case block := <-processedBlock:
|
|
if block == nil {
|
|
continue
|
|
}
|
|
|
|
h, err := hashutil.HashBeaconBlock(block)
|
|
if err != nil {
|
|
log.Errorf("Could not hash incoming block: %v", err)
|
|
continue
|
|
}
|
|
|
|
log.Info("Updating chain head...")
|
|
currentHead, err := c.beaconDB.GetChainHead()
|
|
if err != nil {
|
|
log.Errorf("Could not get current chain head: %v", err)
|
|
continue
|
|
}
|
|
currentState, err := c.beaconDB.GetState()
|
|
if err != nil {
|
|
log.Errorf("Could not get current beacon state: %v", err)
|
|
continue
|
|
}
|
|
|
|
blockState := c.unfinalizedBlocks[h]
|
|
|
|
var headUpdated bool
|
|
newHead := currentHead
|
|
// If both blocks have the same crystallized state root, we favor one which has
|
|
// the higher slot.
|
|
if bytes.Equal(currentHead.StateRootHash32, block.StateRootHash32) {
|
|
if block.Slot > currentHead.Slot {
|
|
newHead = block
|
|
headUpdated = true
|
|
}
|
|
// 2a. Pick the block with the higher last_finalized_slot.
|
|
// 2b. If same, pick the block with the higher last_justified_slot.
|
|
} else if blockState.FinalizedSlot > currentState.FinalizedSlot {
|
|
newHead = block
|
|
headUpdated = true
|
|
} else if blockState.FinalizedSlot == currentState.FinalizedSlot {
|
|
if blockState.JustifiedSlot > currentState.JustifiedSlot {
|
|
newHead = block
|
|
headUpdated = true
|
|
} else if blockState.JustifiedSlot == currentState.JustifiedSlot {
|
|
if block.Slot > currentHead.Slot {
|
|
newHead = block
|
|
headUpdated = true
|
|
}
|
|
}
|
|
}
|
|
|
|
// If no new head was found, we do not update the chain.
|
|
if !headUpdated {
|
|
log.Info("Chain head not updated")
|
|
continue
|
|
}
|
|
|
|
// TODO(#674): Handle chain reorgs.
|
|
newState := blockState
|
|
if err := c.beaconDB.UpdateChainHead(newHead, newState); err != nil {
|
|
log.Errorf("Failed to update chain: %v", err)
|
|
continue
|
|
}
|
|
log.WithField("blockHash", fmt.Sprintf("0x%x", h)).Info("Chain head block and state updated")
|
|
// We fire events that notify listeners of a new block in
|
|
// the case of a state transition. This is useful for the beacon node's gRPC
|
|
// server to stream these events to beacon clients.
|
|
// When the transition is a cycle transition, we stream the state containing the new validator
|
|
// assignments to clients.
|
|
if block.Slot%params.BeaconConfig().EpochLength == 0 {
|
|
c.canonicalStateFeed.Send(newState)
|
|
}
|
|
c.canonicalBlockFeed.Send(newHead)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *ChainService) blockProcessing(processedBlock chan<- *pb.BeaconBlock) {
|
|
subBlock := c.incomingBlockFeed.Subscribe(c.incomingBlockChan)
|
|
defer subBlock.Unsubscribe()
|
|
for {
|
|
select {
|
|
case <-c.ctx.Done():
|
|
log.Debug("Chain service context closed, exiting goroutine")
|
|
return
|
|
|
|
// Listen for a newly received incoming block from the feed. Blocks
|
|
// can be received either from the sync service, the RPC service,
|
|
// or via p2p.
|
|
case block := <-c.incomingBlockChan:
|
|
// Before sending the blocks for processing we check to see if the blocks
|
|
// are valid to continue being processed. If the slot number in the block
|
|
// has already been processed by the beacon node, we throw it away. If the
|
|
// slot number is too high to be processed in the current slot, we store
|
|
// it in a cache.
|
|
beaconState, err := c.beaconDB.GetState()
|
|
if err != nil {
|
|
log.Errorf("Unable to retrieve beacon state %v", err)
|
|
continue
|
|
}
|
|
|
|
currentSlot := beaconState.Slot
|
|
if currentSlot+1 < block.Slot {
|
|
c.unProcessedBlocks[block.Slot] = block
|
|
continue
|
|
}
|
|
|
|
if currentSlot+1 == block.Slot {
|
|
if err := c.receiveBlock(block, beaconState); err != nil {
|
|
log.Error(err)
|
|
processedBlock <- nil
|
|
continue
|
|
}
|
|
// Push the block to trigger the fork choice rule.
|
|
processedBlock <- block
|
|
} else {
|
|
log.Debugf(
|
|
"Block slot number is lower than the current slot in the beacon state %d",
|
|
block.Slot)
|
|
c.sendAndDeleteCachedBlocks(currentSlot, beaconState)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// receiveBlock is a function that defines the operations that are preformed on
|
|
// any block that is received from p2p layer or rpc. It checks the block to see
|
|
// if it passes the pre-processing conditions, if it does then the per slot
|
|
// state transition function is carried out on the block.
|
|
// spec:
|
|
// def process_block(block):
|
|
// if not block_pre_processing_conditions(block):
|
|
// return False
|
|
//
|
|
// # process skipped slots
|
|
//
|
|
// while (state.slot < block.slot - 1):
|
|
// state = slot_state_transition(state, block=None)
|
|
//
|
|
// # process slot with block
|
|
// state = slot_state_transition(state, block)
|
|
//
|
|
// # check state root
|
|
// if block.state_root == hash(state):
|
|
// return state
|
|
// else:
|
|
// return False # or throw or whatever
|
|
//
|
|
func (c *ChainService) receiveBlock(block *pb.BeaconBlock, beaconState *pb.BeaconState) error {
|
|
|
|
blockhash, err := hashutil.HashBeaconBlock(block)
|
|
if err != nil {
|
|
return fmt.Errorf("could not hash incoming block: %v", err)
|
|
}
|
|
|
|
if block.Slot == 0 {
|
|
return errors.New("cannot process a genesis block: received block with slot 0")
|
|
}
|
|
|
|
// Save blocks with higher slot numbers in cache.
|
|
if err := c.isBlockReadyForProcessing(block, beaconState); err != nil {
|
|
log.Debugf("block with hash %#x is not ready for processing: %v", blockhash, err)
|
|
return nil
|
|
}
|
|
|
|
prevBlock, err := c.beaconDB.GetChainHead()
|
|
if err != nil {
|
|
return fmt.Errorf("could not retrieve chain head %v", err)
|
|
}
|
|
|
|
// TODO(#716):Replace with tree-hashing algorithm.
|
|
blockRoot, err := hashutil.HashBeaconBlock(prevBlock)
|
|
if err != nil {
|
|
return fmt.Errorf("could not hash block %v", err)
|
|
}
|
|
|
|
log.WithField("slotNumber", block.Slot).Info("Executing state transition")
|
|
|
|
// Check for skipped slots and update the corresponding proposers
|
|
// randao layer.
|
|
for beaconState.Slot < block.Slot-1 {
|
|
beaconState, err = state.ExecuteStateTransition(beaconState, nil, blockRoot)
|
|
if err != nil {
|
|
return fmt.Errorf("could not execute state transition %v", err)
|
|
}
|
|
}
|
|
|
|
beaconState, err = state.ExecuteStateTransition(beaconState, block, blockRoot)
|
|
if err != nil {
|
|
return fmt.Errorf("could not execute state transition %v", err)
|
|
}
|
|
|
|
// TODO(#1074): Verify block.state_root == hash_tree_root(state)
|
|
// if there exists a block for the slot being processed.
|
|
|
|
if err := c.beaconDB.SaveBlock(block); err != nil {
|
|
return fmt.Errorf("failed to save block: %v", err)
|
|
}
|
|
if err := c.beaconDB.SaveUnfinalizedBlockState(beaconState); err != nil {
|
|
return fmt.Errorf("error persisting unfinalized block's state: %v", err)
|
|
}
|
|
|
|
log.WithField("hash", fmt.Sprintf("%#x", blockhash)).Debug("Processed beacon block")
|
|
|
|
// We keep a map of unfinalized blocks in memory along with their state
|
|
// pair to apply the fork choice rule.
|
|
c.unfinalizedBlocks[blockhash] = beaconState
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *ChainService) isBlockReadyForProcessing(block *pb.BeaconBlock, beaconState *pb.BeaconState) error {
|
|
|
|
var powBlockFetcher func(ctx context.Context, hash common.Hash) (*gethTypes.Block, error)
|
|
if c.enablePOWChain {
|
|
powBlockFetcher = c.web3Service.Client().BlockByHash
|
|
}
|
|
if err := b.IsValidBlock(c.ctx, beaconState, block, c.enablePOWChain,
|
|
c.beaconDB.HasBlock, powBlockFetcher, c.genesisTime); err != nil {
|
|
return fmt.Errorf("block does not fulfill pre-processing conditions %v", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// sendAndDeleteCachedBlocks checks if there is any block saved in the cache with a
|
|
// slot number equivalent to the current slot. If there is then the block is
|
|
// sent to the incoming block channel and deleted from the cache.
|
|
func (c *ChainService) sendAndDeleteCachedBlocks(currentSlot uint64, beaconState *pb.BeaconState) {
|
|
if block, ok := c.unProcessedBlocks[currentSlot+1]; ok {
|
|
if err := c.isBlockReadyForProcessing(block, beaconState); err == nil {
|
|
c.incomingBlockChan <- block
|
|
delete(c.unProcessedBlocks, currentSlot)
|
|
}
|
|
}
|
|
}
|