use headblock for prunePostBlockOperationPools, remove duplicate mark…

…InclusionBLStoExecutionChange calls (#12085)

* removing duplicate function

* moved markInclusion for bls to use headblock instead of processed block

* updating based on internal feedback

* addressing some comments

* addressing feedback from slack

* fixing conflict

* making changes based on suggestions on slack

* reverting a change

* making chases based on potuz's comments

* removing one additional block copy

* clarifying comments

beacon-chain/blockchain/process_block.go

@@ -152,9 +152,6 @@ func (s *Service) onBlock(ctx context.Context, signed interfaces.ReadOnlySignedB
 	if err := s.handleBlockAttestations(ctx, signed.Block(), postState); err != nil {
 		return errors.Wrap(err, "could not handle block's attestations")
 	}
-	if err := s.handleBlockBLSToExecChanges(signed.Block()); err != nil {
-		return errors.Wrap(err, "could not handle block's BLSToExecutionChanges")
-	}
 
 	s.InsertSlashingsToForkChoiceStore(ctx, signed.Block().Body().AttesterSlashings())
 	if isValidPayload {
@@ -214,6 +211,8 @@ func (s *Service) onBlock(ctx context.Context, signed interfaces.ReadOnlySignedB
 	}
 	newBlockHeadElapsedTime.Observe(float64(time.Since(start).Milliseconds()))
 
+	// verify conditions for FCU, notifies FCU, and saves the new head.
+	// This function also prunes attestations, other similar operations happen in prunePostBlockOperationPools.
 	if err := s.forkchoiceUpdateWithExecution(ctx, headRoot, s.CurrentSlot()+1); err != nil {
 		return err
 	}
@@ -597,8 +596,8 @@ func (s *Service) savePostStateInfo(ctx context.Context, r [32]byte, b interface
 }
 
 // This removes the attestations in block `b` from the attestation mem pool.
-func (s *Service) pruneAttsFromPool(b interfaces.ReadOnlySignedBeaconBlock) error {
-	atts := b.Block().Body().Attestations()
+func (s *Service) pruneAttsFromPool(headBlock interfaces.ReadOnlySignedBeaconBlock) error {
+	atts := headBlock.Block().Body().Attestations()
 	for _, att := range atts {
 		if helpers.IsAggregated(att) {
 			if err := s.cfg.AttPool.DeleteAggregatedAttestation(att); err != nil {

beacon-chain/blockchain/receive_block.go

@@ -1,6 +1,7 @@
 package blockchain
 
 import (
+	"bytes"
 	"context"
 
 	"github.com/pkg/errors"
@@ -48,17 +49,16 @@ func (s *Service) ReceiveBlock(ctx context.Context, block interfaces.ReadOnlySig
 
 	s.cfg.ForkChoiceStore.Lock()
 	defer s.cfg.ForkChoiceStore.Unlock()
-
 	// Apply state transition on the new block.
 	if err := s.onBlock(ctx, blockCopy, blockRoot); err != nil {
 		err := errors.Wrap(err, "could not process block")
 		tracing.AnnotateError(span, err)
 		return err
 	}
 
-	// Handle post block operations such as attestations and exits.
-	if err := s.handlePostBlockOperations(blockCopy.Block()); err != nil {
-		return err
+	// Handle post block operations such as pruning exits and bls messages if incoming block is the head
+	if err := s.prunePostBlockOperationPools(ctx, blockCopy, blockRoot); err != nil {
+		log.WithError(err).Error("Could not prune canonical objects from pool ")
 	}
 
 	// Have we been finalizing? Should we start saving hot states to db?
@@ -157,29 +157,40 @@ func (s *Service) ReceiveAttesterSlashing(ctx context.Context, slashing *ethpb.A
 	s.InsertSlashingsToForkChoiceStore(ctx, []*ethpb.AttesterSlashing{slashing})
 }
 
-func (s *Service) handlePostBlockOperations(b interfaces.ReadOnlyBeaconBlock) error {
+// prunePostBlockOperationPools only runs on new head otherwise should return a nil.
+func (s *Service) prunePostBlockOperationPools(ctx context.Context, blk interfaces.ReadOnlySignedBeaconBlock, root [32]byte) error {
+	headRoot, err := s.HeadRoot(ctx)
+	if err != nil {
+		return err
+	}
+	// By comparing the current headroot, that has already gone through forkchoice,
+	// we can assume that if equal the current block root is canonical.
+	if !bytes.Equal(headRoot, root[:]) {
+		return nil
+	}
+
 	// Mark block exits as seen so we don't include same ones in future blocks.
-	for _, e := range b.Body().VoluntaryExits() {
+	for _, e := range blk.Block().Body().VoluntaryExits() {
 		s.cfg.ExitPool.MarkIncluded(e)
 	}
 
 	// Mark block BLS changes as seen so we don't include same ones in future blocks.
-	if err := s.handleBlockBLSToExecChanges(b); err != nil {
+	if err := s.markIncludedBlockBLSToExecChanges(blk.Block()); err != nil {
 		return errors.Wrap(err, "could not process BLSToExecutionChanges")
 	}
 
 	//  Mark attester slashings as seen so we don't include same ones in future blocks.
-	for _, as := range b.Body().AttesterSlashings() {
+	for _, as := range blk.Block().Body().AttesterSlashings() {
 		s.cfg.SlashingPool.MarkIncludedAttesterSlashing(as)
 	}
 	return nil
 }
 
-func (s *Service) handleBlockBLSToExecChanges(blk interfaces.ReadOnlyBeaconBlock) error {
-	if blk.Version() < version.Capella {
+func (s *Service) markIncludedBlockBLSToExecChanges(headBlock interfaces.ReadOnlyBeaconBlock) error {
+	if headBlock.Version() < version.Capella {
 		return nil
 	}
-	changes, err := blk.Body().BLSToExecutionChanges()
+	changes, err := headBlock.Body().BLSToExecutionChanges()
 	if err != nil {
 		return errors.Wrap(err, "could not get BLSToExecutionChanges")
 	}

beacon-chain/blockchain/receive_block_test.go

@@ -357,7 +357,7 @@ func TestHandleBlockBLSToExecutionChanges(t *testing.T) {
 		}
 		blk, err := blocks.NewBeaconBlock(pbb)
 		require.NoError(t, err)
-		require.NoError(t, service.handleBlockBLSToExecChanges(blk))
+		require.NoError(t, service.markIncludedBlockBLSToExecChanges(blk))
 	})
 
 	t.Run("Post Capella no changes", func(t *testing.T) {
@@ -367,7 +367,7 @@ func TestHandleBlockBLSToExecutionChanges(t *testing.T) {
 		}
 		blk, err := blocks.NewBeaconBlock(pbb)
 		require.NoError(t, err)
-		require.NoError(t, service.handleBlockBLSToExecChanges(blk))
+		require.NoError(t, service.markIncludedBlockBLSToExecChanges(blk))
 	})
 
 	t.Run("Post Capella some changes", func(t *testing.T) {
@@ -389,7 +389,7 @@ func TestHandleBlockBLSToExecutionChanges(t *testing.T) {
 
 		pool.InsertBLSToExecChange(signedChange)
 		require.Equal(t, true, pool.ValidatorExists(idx))
-		require.NoError(t, service.handleBlockBLSToExecChanges(blk))
+		require.NoError(t, service.markIncludedBlockBLSToExecChanges(blk))
 		require.Equal(t, false, pool.ValidatorExists(idx))
 	})
 }