prysm-pulse/beacon-chain/blockchain/block_processing.go

package blockchain

import (
	"context"
	"fmt"

	b "github.com/prysmaticlabs/prysm/beacon-chain/core/blocks"
	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
	"github.com/prysmaticlabs/prysm/beacon-chain/core/state"
	"github.com/prysmaticlabs/prysm/beacon-chain/core/validators"
	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"
	"go.opencensus.io/trace"
)

// BlockReceiver interface defines the methods in the blockchain service which
// directly receives a new block from other services and applies the full processing pipeline.
type BlockReceiver interface {
	CanonicalBlockFeed() *event.Feed
	ReceiveBlock(ctx context.Context, block *pb.BeaconBlock) (*pb.BeaconState, error)
}

// BlockProcessor defines a common interface for methods useful for directly applying state transitions
// to beacon blocks and generating a new beacon state from the Ethereum 2.0 core primitives.
type BlockProcessor interface {
	VerifyBlockValidity(block *pb.BeaconBlock, beaconState *pb.BeaconState) error
	ApplyBlockStateTransition(ctx context.Context, block *pb.BeaconBlock, beaconState *pb.BeaconState) (*pb.BeaconState, error)
}

// ReceiveBlock is a function that defines the operations that are preformed on
// any block that is received from p2p layer or rpc. It performs the following actions: It checks the block to see
// 1. Verify a block passes pre-processing conditions
// 2. Save and broadcast the block via p2p to other peers
// 3. Apply the block state transition function and account for skip slots.
// 4. Process and cleanup any block operations, such as attestations and deposits, which would need to be
//    either included or flushed from the beacon node's runtime.
func (c *ChainService) ReceiveBlock(ctx context.Context, block *pb.BeaconBlock) (*pb.BeaconState, error) {
	ctx, span := trace.StartSpan(ctx, "beacon-chain.blockchain.ReceiveBlock")
	defer span.End()
	beaconState, err := c.beaconDB.State(ctx)
	if err != nil {
		return nil, fmt.Errorf("could not retrieve beacon state: %v", err)
	}

	// We first verify the block's basic validity conditions.
	if err := c.VerifyBlockValidity(block, beaconState); err != nil {
		return nil, fmt.Errorf("block with slot %d is not ready for processing: %v", block.Slot, err)
	}

	// We save the block to the DB and broadcast it to our peers.
	if err := c.SaveAndBroadcastBlock(ctx, block); err != nil {
		return nil, fmt.Errorf(
			"could not save and broadcast beacon block with slot %d: %v",
			block.Slot-params.BeaconConfig().GenesisSlot, err,
		)
	}

	log.WithField("slotNumber", block.Slot-params.BeaconConfig().GenesisSlot).Info(
		"Executing state transition")

	// We then apply the block state transition accordingly to obtain the resulting beacon state.
	beaconState, err = c.ApplyBlockStateTransition(ctx, block, beaconState)
	if err != nil {
		return nil, fmt.Errorf("could not apply block state transition: %v", err)
	}

	log.WithFields(logrus.Fields{
		"slotNumber":     block.Slot - params.BeaconConfig().GenesisSlot,
		"justifiedEpoch": beaconState.JustifiedEpoch - params.BeaconConfig().GenesisEpoch,
		"finalizedEpoch": beaconState.FinalizedEpoch - params.BeaconConfig().GenesisEpoch,
	}).Info("State transition complete")

	// We process the block's contained deposits, attestations, and other operations
	// and that may need to be stored or deleted from the beacon node's persistent storage.
	if err := c.CleanupBlockOperations(ctx, block); err != nil {
		return nil, fmt.Errorf("could not process block deposits, attestations, and other operations: %v", err)
	}

	log.WithField("slot", block.Slot-params.BeaconConfig().GenesisSlot).Info("Processed beacon block")
	return beaconState, nil
}

// ApplyBlockStateTransition runs the Ethereum 2.0 state transition function
// to produce a new beacon state and also accounts for skip slots occurring.
//
//  def apply_block_state_transition(block):
//  	# 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, error
//		else:
//			return nil, error  # or throw or whatever
//
func (c *ChainService) ApplyBlockStateTransition(
	ctx context.Context, block *pb.BeaconBlock, beaconState *pb.BeaconState,
) (*pb.BeaconState, error) {
	// Retrieve the last processed beacon block's hash root.
	headRoot, err := c.ChainHeadRoot()
	if err != nil {
		return nil, fmt.Errorf("could not retrieve chain head root: %v", err)
	}

	// Check for skipped slots.
	numSkippedSlots := 0
	for beaconState.Slot < block.Slot-1 {
		beaconState, err = c.runStateTransition(headRoot, nil, beaconState)
		if err != nil {
			return nil, fmt.Errorf("could not execute state transition without block %v", err)
		}
		numSkippedSlots++
	}
	if numSkippedSlots > 0 {
		log.Warnf("Processed %d skipped slots", numSkippedSlots)
	}

	beaconState, err = c.runStateTransition(headRoot, block, beaconState)
	if err != nil {
		return nil, fmt.Errorf("could not execute state transition with block %v", err)
	}
	return beaconState, nil
}

// VerifyBlockValidity cross-checks the block against the pre-processing conditions from
// Ethereum 2.0, namely:
//   The parent block with root block.parent_root has been processed and accepted.
//   The node has processed its state up to slot, block.slot - 1.
//   The Ethereum 1.0 block pointed to by the state.processed_pow_receipt_root has been processed and accepted.
//   The node's local clock time is greater than or equal to state.genesis_time + block.slot * SECONDS_PER_SLOT.
func (c *ChainService) VerifyBlockValidity(block *pb.BeaconBlock, beaconState *pb.BeaconState) error {
	if block.Slot == params.BeaconConfig().GenesisSlot {
		return fmt.Errorf("cannot process a genesis block: received block with slot %d",
			block.Slot-params.BeaconConfig().GenesisSlot)
	}
	powBlockFetcher := c.web3Service.Client().BlockByHash
	if err := b.IsValidBlock(c.ctx, beaconState, block,
		c.beaconDB.HasBlock, powBlockFetcher, c.genesisTime); err != nil {
		return fmt.Errorf("block does not fulfill pre-processing conditions %v", err)
	}
	return nil
}

// SaveAndBroadcastBlock stores the block in persistent storage and then broadcasts it to
// peers via p2p. Blocks which have already been saved are not processed again via p2p, which is why
// the order of operations is important in this function to prevent infinite p2p loops.
func (c *ChainService) SaveAndBroadcastBlock(ctx context.Context, block *pb.BeaconBlock) error {
	blockRoot, err := hashutil.HashBeaconBlock(block)
	if err != nil {
		return fmt.Errorf("could not tree hash incoming block: %v", err)
	}
	if err := c.beaconDB.SaveBlock(block); err != nil {
		return fmt.Errorf("failed to save block: %v", err)
	}
	// Announce the new block to the network.
	c.p2p.Broadcast(ctx, &pb.BeaconBlockAnnounce{
		Hash:       blockRoot[:],
		SlotNumber: block.Slot,
	})
	return nil
}

// CleanupBlockOperations processes and cleans up any block operations relevant to the beacon node
// such as attestations, exits, and deposits. We update the latest seen attestation by validator
// in the local node's runtime, cleanup and remove pending deposits which have been included in the block
// from our node's local cache, and process validator exits and more.
func (c *ChainService) CleanupBlockOperations(ctx context.Context, block *pb.BeaconBlock) error {
	// Forward processed block to operation pool to remove individual operation from DB.
	if c.opsPoolService.IncomingProcessedBlockFeed().Send(block) == 0 {
		log.Error("Sent processed block to no subscribers")
	}

	// Update attestation store with latest attestation target.
	log.Info("Updating latest attestation target")
	for _, att := range block.Body.Attestations {
		if err := c.attsService.UpdateLatestAttestation(c.ctx, att); err != nil {
			return fmt.Errorf("failed to update latest attestation for store: %v", err)
		}
		log.WithFields(
			logrus.Fields{
				"attestationSlot": att.Data.Slot - params.BeaconConfig().GenesisSlot,
				"justifiedEpoch":  att.Data.JustifiedEpoch - params.BeaconConfig().GenesisEpoch,
			},
		).Info("Attestation store updated")
	}

	// Remove pending deposits from the deposit queue.
	for _, dep := range block.Body.Deposits {
		c.beaconDB.RemovePendingDeposit(ctx, dep)
	}
	return nil
}

// runStateTransition executes the Ethereum 2.0 core state transition for the beacon chain and
// updates important checkpoints and local persistent data during epoch transitions. It serves as a wrapper
// around the more low-level, core state transition function primitive.
func (c *ChainService) runStateTransition(
	headRoot [32]byte, block *pb.BeaconBlock, beaconState *pb.BeaconState,
) (*pb.BeaconState, error) {
	beaconState, err := state.ExecuteStateTransition(
		c.ctx,
		beaconState,
		block,
		headRoot,
		&state.TransitionConfig{
			VerifySignatures: false, // We disable signature verification for now.
			Logging:          true,  // We enable logging in this state transition call.
		},
	)
	if err != nil {
		return nil, fmt.Errorf("could not execute state transition %v", err)
	}
	log.WithField(
		"slotsSinceGenesis", beaconState.Slot-params.BeaconConfig().GenesisSlot,
	).Info("Slot transition successfully processed")

	if block != nil {
		log.WithField(
			"slotsSinceGenesis", beaconState.Slot-params.BeaconConfig().GenesisSlot,
		).Info("Block transition successfully processed")
	}

	if helpers.IsEpochEnd(beaconState.Slot) {
		// Save activated validators of this epoch to public key -> index DB.
		if err := c.saveValidatorIdx(beaconState); err != nil {
			return nil, fmt.Errorf("could not save validator index: %v", err)
		}
		// Delete exited validators of this epoch to public key -> index DB.
		if err := c.deleteValidatorIdx(beaconState); err != nil {
			return nil, fmt.Errorf("could not delete validator index: %v", err)
		}
		// Update FFG checkpoints in DB.
		if err := c.updateFFGCheckPts(beaconState); err != nil {
			return nil, fmt.Errorf("could not update FFG checkpts: %v", err)
		}
		// Save Historical States.
		if err := c.beaconDB.SaveHistoricalState(beaconState); err != nil {
			return nil, fmt.Errorf("could not save historical state: %v", err)
		}
		log.WithField(
			"SlotsSinceGenesis", beaconState.Slot-params.BeaconConfig().GenesisSlot,
		).Info("Epoch transition successfully processed")
	}
	return beaconState, nil
}

// saveValidatorIdx saves the validators public key to index mapping in DB, these
// validators were activated from current epoch. After it saves, current epoch key
// is deleted from ActivatedValidators mapping.
func (c *ChainService) saveValidatorIdx(state *pb.BeaconState) error {
	activatedValidators := validators.ActivatedValFromEpoch(helpers.CurrentEpoch(state))
	for _, idx := range activatedValidators {
		pubKey := state.ValidatorRegistry[idx].Pubkey
		if err := c.beaconDB.SaveValidatorIndex(pubKey, int(idx)); err != nil {
			return fmt.Errorf("could not save validator index: %v", err)
		}
	}
	validators.DeleteActivatedVal(helpers.CurrentEpoch(state))
	return nil
}

// deleteValidatorIdx deletes the validators public key to index mapping in DB, the
// validators were exited from current epoch. After it deletes, current epoch key
// is deleted from ExitedValidators mapping.
func (c *ChainService) deleteValidatorIdx(state *pb.BeaconState) error {
	exitedValidators := validators.ExitedValFromEpoch(helpers.CurrentEpoch(state))
	for _, idx := range exitedValidators {
		pubKey := state.ValidatorRegistry[idx].Pubkey
		if err := c.beaconDB.DeleteValidatorIndex(pubKey); err != nil {
			return fmt.Errorf("could not delete validator index: %v", err)
		}
	}
	validators.DeleteExitedVal(helpers.CurrentEpoch(state))
	return nil
}