[DO NOT MERGE]: Apply consensus and equivocation check before broadcast

beacon-chain/node/node.go

@@ -753,6 +753,11 @@ func (b *BeaconNode) registerRPCService(router *mux.Router) error {
 		return err
 	}
 
+	var regSyncService *regularsync.Service
+	if err := b.services.FetchService(&regSyncService); err != nil {
+		return err
+	}
+
 	var slasherService *slasher.Service
 	if features.Get().EnableSlasher {
 		if err := b.services.FetchService(&slasherService); err != nil {
@@ -834,6 +839,7 @@ func (b *BeaconNode) registerRPCService(router *mux.Router) error {
 		ProposerIdsCache:              b.proposerIdsCache,
 		BlockBuilder:                  b.fetchBuilderService(),
 		Router:                        router,
+		RegSyncService:                regSyncService,
 	})
 
 	return b.services.RegisterService(rpcService)

beacon-chain/rpc/eth/beacon/blocks.go

@@ -10,6 +10,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/feed"
 	blockfeed "github.com/prysmaticlabs/prysm/v4/beacon-chain/core/feed/block"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/helpers"
+	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/transition"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/filters"
 	rpchelpers "github.com/prysmaticlabs/prysm/v4/beacon-chain/rpc/eth/helpers"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/rpc/lookup"
@@ -1019,6 +1020,24 @@ func (bs *Server) submitBlock(ctx context.Context, blockRoot [fieldparams.RootLe
 		})
 	}()
 
+	if block == nil || block.IsNil() {
+		return errors.New("nil block")
+	}
+
+	b := block.Block()
+	parentState, err := bs.StateGenService.StateByRoot(ctx, b.ParentRoot())
+	if err != nil {
+		return errors.Wrap(err, "could not get parent state")
+	}
+	_, err = transition.ExecuteStateTransition(ctx, parentState, block)
+	if err != nil {
+		return errors.Wrap(err, "could not execute state transition")
+	}
+
+	if bs.EquivocationChecker.SeenProposerIndex(b.Slot(), b.ProposerIndex()) {
+		return fmt.Errorf("block exists in sync service, slot: %d, proposer index: %d", b.Slot(), b.ProposerIndex())
+	}
+
 	// Broadcast the new block to the network.
 	blockPb, err := block.Proto()
 	if err != nil {
@@ -1028,9 +1047,5 @@ func (bs *Server) submitBlock(ctx context.Context, blockRoot [fieldparams.RootLe
 		return status.Errorf(codes.Internal, "Could not broadcast block: %v", err)
 	}
 
-	if err := bs.BlockReceiver.ReceiveBlock(ctx, block, blockRoot); err != nil {
-		return status.Errorf(codes.Internal, "Could not process beacon block: %v", err)
-	}
-
 	return nil
 }

beacon-chain/rpc/eth/beacon/server.go

@@ -40,6 +40,7 @@ type Server struct {
 	OptimisticModeFetcher         blockchain.OptimisticModeFetcher
 	V1Alpha1ValidatorServer       *v1alpha1validator.Server
 	SyncChecker                   sync.Checker
+	EquivocationChecker           sync.EquivocationChecker
 	CanonicalHistory              *stategen.CanonicalHistory
 	ExecutionPayloadReconstructor execution.ExecutionPayloadReconstructor
 	FinalizationFetcher           blockchain.FinalizationFetcher

beacon-chain/rpc/service.go

@@ -104,6 +104,7 @@ type Config struct {
 	SyncCommitteeObjectPool       synccommittee.Pool
 	BLSChangesPool                blstoexec.PoolManager
 	SyncService                   chainSync.Checker
+	RegSyncService                chainSync.EquivocationChecker
 	Broadcaster                   p2p.Broadcaster
 	PeersFetcher                  p2p.PeersProvider
 	PeerManager                   p2p.PeerManager
@@ -335,6 +336,7 @@ func (s *Service) Start() {
 		ExecutionPayloadReconstructor: s.cfg.ExecutionPayloadReconstructor,
 		BLSChangesPool:                s.cfg.BLSChangesPool,
 		FinalizationFetcher:           s.cfg.FinalizationFetcher,
+		EquivocationChecker:           s.cfg.RegSyncService,
 	}
 	ethpbv1alpha1.RegisterNodeServer(s.grpcServer, nodeServer)
 	ethpbservice.RegisterBeaconNodeServer(s.grpcServer, nodeServerV1)

beacon-chain/sync/service.go

@@ -34,6 +34,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/state/stategen"
 	lruwrpr "github.com/prysmaticlabs/prysm/v4/cache/lru"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
+	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/runtime"
 	prysmTime "github.com/prysmaticlabs/prysm/v4/time"
@@ -110,6 +111,7 @@ type Service struct {
 	ctx                              context.Context
 	cancel                           context.CancelFunc
 	slotToPendingBlocks              *gcache.Cache
+	seenProposerIndexCache           []primitives.ValidatorIndex
 	seenPendingBlocks                map[[32]byte]bool
 	blkRootToPendingAtts             map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof
 	subHandler                       *subTopicHandler
@@ -146,14 +148,15 @@ func NewService(ctx context.Context, opts ...Option) *Service {
 	c := gcache.New(pendingBlockExpTime /* exp time */, 2*pendingBlockExpTime /* prune time */)
 	ctx, cancel := context.WithCancel(ctx)
 	r := &Service{
-		ctx:                  ctx,
-		cancel:               cancel,
-		chainStarted:         abool.New(),
-		cfg:                  &config{},
-		slotToPendingBlocks:  c,
-		seenPendingBlocks:    make(map[[32]byte]bool),
-		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
-		signatureChan:        make(chan *signatureVerifier, verifierLimit),
+		ctx:                    ctx,
+		cancel:                 cancel,
+		chainStarted:           abool.New(),
+		cfg:                    &config{},
+		slotToPendingBlocks:    c,
+		seenPendingBlocks:      make(map[[32]byte]bool),
+		blkRootToPendingAtts:   make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
+		signatureChan:          make(chan *signatureVerifier, verifierLimit),
+		seenProposerIndexCache: []primitives.ValidatorIndex{0}, // Index 0 is reserved for slot
 	}
 	for _, opt := range opts {
 		if err := opt(r); err != nil {
@@ -301,3 +304,7 @@ type Checker interface {
 	Status() error
 	Resync() error
 }
+
+type EquivocationChecker interface {
+	SeenProposerIndex(slot primitives.Slot, proposerIdx primitives.ValidatorIndex) bool
+}

beacon-chain/sync/validate_beacon_blocks.go

@@ -69,6 +69,15 @@ func (s *Service) validateBeaconBlockPubSub(ctx context.Context, pid peer.ID, ms
 		return pubsub.ValidationReject, errors.New("block.Block is nil")
 	}
 
+	parentState, err := s.cfg.stateGen.StateByRoot(ctx, blk.Block().ParentRoot())
+	if err != nil {
+		return pubsub.ValidationIgnore, err
+	}
+	if err = blocks.VerifyBlockSignatureUsingCurrentFork(parentState, blk); err != nil {
+		return pubsub.ValidationReject, err
+	}
+	s.insertProposerIndexCache(blk.Block().Slot(), blk.Block().ProposerIndex())
+
 	// Broadcast the block on a feed to notify other services in the beacon node
 	// of a received block (even if it does not process correctly through a state transition).
 	s.cfg.blockNotifier.BlockFeed().Send(&feed.Event{
@@ -384,3 +393,32 @@ func getBlockFields(b interfaces.ReadOnlySignedBeaconBlock) logrus.Fields {
 		"version":       b.Block().Version(),
 	}
 }
+
+func (s *Service) insertProposerIndexCache(slot primitives.Slot, proposerIdx primitives.ValidatorIndex) {
+	if s.cfg.chain.CurrentSlot() != slot {
+		return
+	}
+
+	switch {
+	case uint64(s.seenProposerIndexCache[0]) > uint64(slot):
+		return
+	case uint64(s.seenProposerIndexCache[0]) == uint64(slot):
+		s.seenProposerIndexCache = append(s.seenProposerIndexCache, proposerIdx)
+	case uint64(slot) > uint64(s.seenProposerIndexCache[0]):
+		// Overwrite slot in proposer index cache if it's higher.
+		s.seenProposerIndexCache = []primitives.ValidatorIndex{primitives.ValidatorIndex(slot)}
+		s.seenProposerIndexCache = append(s.seenProposerIndexCache, proposerIdx)
+	}
+}
+
+func (s *Service) SeenProposerIndex(slot primitives.Slot, proposerIdx primitives.ValidatorIndex) bool {
+	if s.seenProposerIndexCache[0] != primitives.ValidatorIndex(slot) {
+		return false
+	}
+	for _, i := range s.seenProposerIndexCache[1:] {
+		if i == proposerIdx {
+			return true
+		}
+	}
+	return false
+}