p2p: Return error if no ChainState found for a peer (#9502)

* p2p: Return error if no ChainState found for a peer

* add commentary

(cherry picked from commit dad03ade77)

beacon-chain/blockchain/head.go

@@ -15,6 +15,7 @@ import (
 	ethpbv1 "github.com/prysmaticlabs/prysm/proto/eth/v1"
 	"github.com/prysmaticlabs/prysm/proto/interfaces"
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
+	"github.com/prysmaticlabs/prysm/shared/featureconfig"
 	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/slotutil"
 	"github.com/sirupsen/logrus"
@@ -143,6 +144,10 @@ func (s *Service) saveHead(ctx context.Context, headRoot [32]byte) error {
 			},
 		})
 
+		if err := s.saveOrphanedAtts(ctx, bytesutil.ToBytes32(r)); err != nil {
+			return err
+		}
+
 		reorgCount.Inc()
 	}
 
@@ -360,3 +365,40 @@ func (s *Service) notifyNewHeadEvent(
 	})
 	return nil
 }
+
+// This saves the attestations inside the beacon block with respect to root `orphanedRoot` back into the
+// attestation pool. It also filters out the attestations that is one epoch older as a
+// defense so invalid attestations don't flow into the attestation pool.
+func (s *Service) saveOrphanedAtts(ctx context.Context, orphanedRoot [32]byte) error {
+	if !featureconfig.Get().CorrectlyInsertOrphanedAtts {
+		return nil
+	}
+
+	orphanedBlk, err := s.cfg.BeaconDB.Block(ctx, orphanedRoot)
+	if err != nil {
+		return err
+	}
+
+	if orphanedBlk == nil || orphanedBlk.IsNil() {
+		return errors.New("orphaned block can't be nil")
+	}
+
+	for _, a := range orphanedBlk.Block().Body().Attestations() {
+		// Is the attestation one epoch older.
+		if a.Data.Slot+params.BeaconConfig().SlotsPerEpoch < s.CurrentSlot() {
+			continue
+		}
+		if helpers.IsAggregated(a) {
+			if err := s.cfg.AttPool.SaveAggregatedAttestation(a); err != nil {
+				return err
+			}
+		} else {
+			if err := s.cfg.AttPool.SaveUnaggregatedAttestation(a); err != nil {
+				return err
+			}
+		}
+		saveOrphanedAttCount.Inc()
+	}
+
+	return nil
+}

beacon-chain/blockchain/head_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"testing"
+	"time"
 
 	types "github.com/prysmaticlabs/eth2-types"
 	mock "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing"
@@ -13,6 +14,8 @@ import (
 	ethpbv1 "github.com/prysmaticlabs/prysm/proto/eth/v1"
 	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/proto/eth/v1alpha1/wrapper"
+	"github.com/prysmaticlabs/prysm/shared/featureconfig"
+	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/testutil"
 	"github.com/prysmaticlabs/prysm/shared/testutil/assert"
 	"github.com/prysmaticlabs/prysm/shared/testutil/require"
@@ -36,18 +39,17 @@ func TestSaveHead_Different(t *testing.T) {
 	beaconDB := testDB.SetupDB(t)
 	service := setupBeaconChain(t, beaconDB)
 
-	oldRoot := [32]byte{'A'}
+	testutil.NewBeaconBlock()
+	oldBlock := wrapper.WrappedPhase0SignedBeaconBlock(
+		testutil.NewBeaconBlock(),
+	)
+	require.NoError(t, service.cfg.BeaconDB.SaveBlock(context.Background(), oldBlock))
+	oldRoot, err := oldBlock.Block().HashTreeRoot()
+	require.NoError(t, err)
 	service.head = &head{
-		slot: 0,
-		root: oldRoot,
-		block: wrapper.WrappedPhase0SignedBeaconBlock(
-			&ethpb.SignedBeaconBlock{
-				Block: &ethpb.BeaconBlock{
-					Slot:      0,
-					StateRoot: make([]byte, 32),
-				},
-			},
-		),
+		slot:  0,
+		root:  oldRoot,
+		block: oldBlock,
 	}
 
 	newHeadSignedBlock := testutil.NewBeaconBlock()
@@ -79,18 +81,16 @@ func TestSaveHead_Different_Reorg(t *testing.T) {
 	beaconDB := testDB.SetupDB(t)
 	service := setupBeaconChain(t, beaconDB)
 
-	oldRoot := [32]byte{'A'}
+	oldBlock := wrapper.WrappedPhase0SignedBeaconBlock(
+		testutil.NewBeaconBlock(),
+	)
+	require.NoError(t, service.cfg.BeaconDB.SaveBlock(context.Background(), oldBlock))
+	oldRoot, err := oldBlock.Block().HashTreeRoot()
+	require.NoError(t, err)
 	service.head = &head{
-		slot: 0,
-		root: oldRoot,
-		block: wrapper.WrappedPhase0SignedBeaconBlock(
-			&ethpb.SignedBeaconBlock{
-				Block: &ethpb.BeaconBlock{
-					Slot:      0,
-					StateRoot: make([]byte, 32),
-				},
-			},
-		),
+		slot:  0,
+		root:  oldRoot,
+		block: oldBlock,
 	}
 
 	reorgChainParent := [32]byte{'B'}
@@ -214,3 +214,55 @@ func Test_notifyNewHeadEvent(t *testing.T) {
 		require.DeepSSZEqual(t, wanted, eventHead)
 	})
 }
+
+func TestSaveOrphanedAtts(t *testing.T) {
+	resetCfg := featureconfig.InitWithReset(&featureconfig.Flags{
+		CorrectlyInsertOrphanedAtts: true,
+	})
+	defer resetCfg()
+
+	genesis, keys := testutil.DeterministicGenesisState(t, 64)
+	b, err := testutil.GenerateFullBlock(genesis, keys, testutil.DefaultBlockGenConfig(), 1)
+	assert.NoError(t, err)
+	r, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	beaconDB := testDB.SetupDB(t)
+	service := setupBeaconChain(t, beaconDB)
+	service.genesisTime = time.Now()
+
+	require.NoError(t, service.cfg.BeaconDB.SaveBlock(ctx, wrapper.WrappedPhase0SignedBeaconBlock(b)))
+	require.NoError(t, service.saveOrphanedAtts(ctx, r))
+
+	require.Equal(t, len(b.Block.Body.Attestations), service.cfg.AttPool.AggregatedAttestationCount())
+	savedAtts := service.cfg.AttPool.AggregatedAttestations()
+	atts := b.Block.Body.Attestations
+	require.DeepSSZEqual(t, atts, savedAtts)
+}
+
+func TestSaveOrphanedAtts_CanFilter(t *testing.T) {
+	resetCfg := featureconfig.InitWithReset(&featureconfig.Flags{
+		CorrectlyInsertOrphanedAtts: true,
+	})
+	defer resetCfg()
+
+	genesis, keys := testutil.DeterministicGenesisState(t, 64)
+	b, err := testutil.GenerateFullBlock(genesis, keys, testutil.DefaultBlockGenConfig(), 1)
+	assert.NoError(t, err)
+	r, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	beaconDB := testDB.SetupDB(t)
+	service := setupBeaconChain(t, beaconDB)
+	service.genesisTime = time.Now().Add(time.Duration(-1*int64(params.BeaconConfig().SlotsPerEpoch+1)*int64(params.BeaconConfig().SecondsPerSlot)) * time.Second)
+
+	require.NoError(t, service.cfg.BeaconDB.SaveBlock(ctx, wrapper.WrappedPhase0SignedBeaconBlock(b)))
+	require.NoError(t, service.saveOrphanedAtts(ctx, r))
+
+	require.Equal(t, 0, service.cfg.AttPool.AggregatedAttestationCount())
+	savedAtts := service.cfg.AttPool.AggregatedAttestations()
+	atts := b.Block.Body.Attestations
+	require.DeepNotSSZEqual(t, atts, savedAtts)
+}

beacon-chain/blockchain/metrics.go

@@ -100,6 +100,10 @@ var (
 		Name: "beacon_reorg_total",
 		Help: "Count the number of times beacon chain has a reorg",
 	})
+	saveOrphanedAttCount = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "saved_orphaned_att_total",
+		Help: "Count the number of times an orphaned attestation is saved",
+	})
 	attestationInclusionDelay = promauto.NewHistogram(
 		prometheus.HistogramOpts{
 			Name:    "attestation_inclusion_delay_slots",

beacon-chain/blockchain/process_attestation_helpers.go

@@ -54,22 +54,13 @@ func (s *Service) getAttPreState(ctx context.Context, c *ethpb.Checkpoint) (ifac
 				return nil, errors.Wrapf(err, "could not process slots up to epoch %d", c.Epoch)
 			}
 		}
-		if err := s.checkpointStateCache.AddCheckpointState(c, baseState); err != nil {
-			return nil, errors.Wrap(err, "could not saved checkpoint state to cache")
-		}
-		return baseState, nil
 	}
-
-	// To avoid sharing the same state across checkpoint state cache and hot state cache,
-	// we don't add the state to check point cache.
-	has, err := s.cfg.StateGen.HasStateInCache(ctx, bytesutil.ToBytes32(c.Root))
-	if err != nil {
-		return nil, err
-	}
-	if !has {
-		if err := s.checkpointStateCache.AddCheckpointState(c, baseState); err != nil {
-			return nil, errors.Wrap(err, "could not saved checkpoint state to cache")
-		}
+	// Sharing the same state across caches is perfectly fine here, the fetching
+	// of attestation prestate is by far the most accessed state fetching pattern in
+	// the beacon node. An extra state instance cached isn't an issue in the bigger
+	// picture.
+	if err := s.checkpointStateCache.AddCheckpointState(c, baseState); err != nil {
+		return nil, errors.Wrap(err, "could not save checkpoint state to cache")
 	}
 	return baseState, nil
 

beacon-chain/blockchain/process_block.go

@@ -141,6 +141,10 @@ func (s *Service) onBlock(ctx context.Context, signed interfaces.SignedBeaconBlo
 			log.WithError(err).Warn("Could not update head")
 		}
 
+		if err := s.pruneCanonicalAttsFromPool(ctx, blockRoot, signed); err != nil {
+			return err
+		}
+
 		// Send notification of the processed block to the state feed.
 		s.cfg.StateNotifier.StateFeed().Send(&feed.Event{
 			Type: statefeed.BlockProcessed,
@@ -305,6 +309,10 @@ func (s *Service) handleBlockAfterBatchVerify(ctx context.Context, signed interf
 		if err := s.updateFinalized(ctx, fCheckpoint); err != nil {
 			return err
 		}
+		if featureconfig.Get().UpdateHeadTimely {
+			s.prevFinalizedCheckpt = s.finalizedCheckpt
+			s.finalizedCheckpt = fCheckpoint
+		}
 	}
 	return nil
 }
@@ -404,3 +412,33 @@ func (s *Service) savePostStateInfo(ctx context.Context, r [32]byte, b interface
 	}
 	return nil
 }
+
+// This removes the attestations from the mem pool. It will only remove the attestations if input root `r` is canonical,
+// meaning the block `b` is part of the canonical chain.
+func (s *Service) pruneCanonicalAttsFromPool(ctx context.Context, r [32]byte, b interfaces.SignedBeaconBlock) error {
+	if !featureconfig.Get().CorrectlyPruneCanonicalAtts {
+		return nil
+	}
+
+	canonical, err := s.IsCanonical(ctx, r)
+	if err != nil {
+		return err
+	}
+	if !canonical {
+		return nil
+	}
+
+	atts := b.Block().Body().Attestations()
+	for _, att := range atts {
+		if helpers.IsAggregated(att) {
+			if err := s.cfg.AttPool.DeleteAggregatedAttestation(att); err != nil {
+				return err
+			}
+		} else {
+			if err := s.cfg.AttPool.DeleteUnaggregatedAttestation(att); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}

beacon-chain/blockchain/process_block_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/prysmaticlabs/prysm/proto/interfaces"
 	"github.com/prysmaticlabs/prysm/shared/attestationutil"
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
+	"github.com/prysmaticlabs/prysm/shared/featureconfig"
 	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/testutil"
 	"github.com/prysmaticlabs/prysm/shared/testutil/assert"
@@ -1001,3 +1002,49 @@ func TestInsertFinalizedDeposits(t *testing.T) {
 		assert.DeepEqual(t, [][]byte(nil), d.Proof, "Proofs are not empty")
 	}
 }
+
+func TestRemoveBlockAttestationsInPool_Canonical(t *testing.T) {
+	resetCfg := featureconfig.InitWithReset(&featureconfig.Flags{
+		CorrectlyPruneCanonicalAtts: true,
+	})
+	defer resetCfg()
+
+	genesis, keys := testutil.DeterministicGenesisState(t, 64)
+	b, err := testutil.GenerateFullBlock(genesis, keys, testutil.DefaultBlockGenConfig(), 1)
+	assert.NoError(t, err)
+	r, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	beaconDB := testDB.SetupDB(t)
+	service := setupBeaconChain(t, beaconDB)
+	require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(ctx, &pb.StateSummary{Root: r[:]}))
+	require.NoError(t, service.cfg.BeaconDB.SaveGenesisBlockRoot(ctx, r))
+
+	atts := b.Block.Body.Attestations
+	require.NoError(t, service.cfg.AttPool.SaveAggregatedAttestations(atts))
+	require.NoError(t, service.pruneCanonicalAttsFromPool(ctx, r, wrapper.WrappedPhase0SignedBeaconBlock(b)))
+	require.Equal(t, 0, service.cfg.AttPool.AggregatedAttestationCount())
+}
+
+func TestRemoveBlockAttestationsInPool_NonCanonical(t *testing.T) {
+	resetCfg := featureconfig.InitWithReset(&featureconfig.Flags{
+		CorrectlyPruneCanonicalAtts: true,
+	})
+	defer resetCfg()
+
+	genesis, keys := testutil.DeterministicGenesisState(t, 64)
+	b, err := testutil.GenerateFullBlock(genesis, keys, testutil.DefaultBlockGenConfig(), 1)
+	assert.NoError(t, err)
+	r, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	beaconDB := testDB.SetupDB(t)
+	service := setupBeaconChain(t, beaconDB)
+
+	atts := b.Block.Body.Attestations
+	require.NoError(t, service.cfg.AttPool.SaveAggregatedAttestations(atts))
+	require.NoError(t, service.pruneCanonicalAttsFromPool(ctx, r, wrapper.WrappedPhase0SignedBeaconBlock(b)))
+	require.Equal(t, 1, service.cfg.AttPool.AggregatedAttestationCount())
+}

beacon-chain/blockchain/receive_block.go

@@ -48,6 +48,11 @@ func (s *Service) ReceiveBlock(ctx context.Context, block interfaces.SignedBeaco
 		if err := s.updateHead(ctx, s.getJustifiedBalances()); err != nil {
 			log.WithError(err).Warn("Could not update head")
 		}
+
+		if err := s.pruneCanonicalAttsFromPool(ctx, blockRoot, block); err != nil {
+			return err
+		}
+
 		// Send notification of the processed block to the state feed.
 		s.cfg.StateNotifier.StateFeed().Send(&feed.Event{
 			Type: statefeed.BlockProcessed,

beacon-chain/blockchain/receive_block_test.go

@@ -8,6 +8,7 @@ import (
 
 	types "github.com/prysmaticlabs/eth2-types"
 	blockchainTesting "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/state"
 	testDB "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
 	"github.com/prysmaticlabs/prysm/beacon-chain/forkchoice/protoarray"
 	"github.com/prysmaticlabs/prysm/beacon-chain/operations/attestations"
@@ -17,6 +18,7 @@ import (
 	"github.com/prysmaticlabs/prysm/proto/eth/v1alpha1/wrapper"
 	"github.com/prysmaticlabs/prysm/proto/interfaces"
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
+	"github.com/prysmaticlabs/prysm/shared/featureconfig"
 	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/testutil"
 	"github.com/prysmaticlabs/prysm/shared/testutil/assert"
@@ -283,6 +285,62 @@ func TestService_ReceiveBlockBatch(t *testing.T) {
 	}
 }
 
+func TestService_ReceiveBlockBatch_UpdateFinalizedCheckpoint(t *testing.T) {
+	// Must enable head timely feature flag to test this.
+	resetCfg := featureconfig.InitWithReset(&featureconfig.Flags{
+		UpdateHeadTimely: true,
+	})
+	defer resetCfg()
+
+	ctx := context.Background()
+	genesis, keys := testutil.DeterministicGenesisState(t, 64)
+
+	// Generate 5 epochs worth of blocks.
+	var blks []interfaces.SignedBeaconBlock
+	var roots [][32]byte
+	copied := genesis.Copy()
+	for i := types.Slot(1); i < params.BeaconConfig().SlotsPerEpoch*5; i++ {
+		b, err := testutil.GenerateFullBlock(copied, keys, testutil.DefaultBlockGenConfig(), i)
+		assert.NoError(t, err)
+		copied, err = state.ExecuteStateTransition(context.Background(), copied, wrapper.WrappedPhase0SignedBeaconBlock(b))
+		assert.NoError(t, err)
+		r, err := b.Block.HashTreeRoot()
+		require.NoError(t, err)
+		blks = append(blks, wrapper.WrappedPhase0SignedBeaconBlock(b))
+		roots = append(roots, r)
+	}
+
+	beaconDB := testDB.SetupDB(t)
+	genesisBlockRoot, err := genesis.HashTreeRoot(ctx)
+	require.NoError(t, err)
+	cfg := &Config{
+		BeaconDB: beaconDB,
+		ForkChoiceStore: protoarray.New(
+			0, // justifiedEpoch
+			0, // finalizedEpoch
+			genesisBlockRoot,
+		),
+		StateNotifier: &blockchainTesting.MockStateNotifier{RecordEvents: false},
+		StateGen:      stategen.New(beaconDB),
+	}
+	s, err := NewService(ctx, cfg)
+	require.NoError(t, err)
+	err = s.saveGenesisData(ctx, genesis)
+	require.NoError(t, err)
+	gBlk, err := s.cfg.BeaconDB.GenesisBlock(ctx)
+	require.NoError(t, err)
+
+	gRoot, err := gBlk.Block().HashTreeRoot()
+	require.NoError(t, err)
+	s.finalizedCheckpt = &ethpb.Checkpoint{Root: gRoot[:]}
+
+	// Process 5 epochs worth of blocks.
+	require.NoError(t, s.ReceiveBlockBatch(ctx, blks, roots))
+
+	// Finalized epoch must be updated.
+	require.Equal(t, types.Epoch(2), s.finalizedCheckpt.Epoch)
+}
+
 func TestService_HasInitSyncBlock(t *testing.T) {
 	s, err := NewService(context.Background(), &Config{StateNotifier: &blockchainTesting.MockStateNotifier{}})
 	require.NoError(t, err)

beacon-chain/forkchoice/protoarray/store.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"math"
 
 	"github.com/pkg/errors"
 	types "github.com/prysmaticlabs/eth2-types"
@@ -411,8 +410,13 @@ func (s *Store) applyWeightChanges(ctx context.Context, justifiedEpoch, finalize
 			if int(n.weight)+nodeDelta < 0 {
 				n.weight = 0
 			} else {
+				// Absolute value of node delta.
+				d := nodeDelta
+				if nodeDelta < 0 {
+					d *= -1
+				}
 				// Subtract node's weight.
-				n.weight -= uint64(math.Abs(float64(nodeDelta)))
+				n.weight -= uint64(d)
 			}
 		} else {
 			// Add node's weight.

beacon-chain/gateway/helpers.go

@@ -1,8 +1,6 @@
 package gateway
 
 import (
-	"net/http"
-
 	gwruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	pbrpc "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1"
 	ethpbv1 "github.com/prysmaticlabs/prysm/proto/eth/v1"
@@ -65,9 +63,6 @@ func DefaultConfig(enableDebugRPCEndpoints bool) MuxConfig {
 			},
 		}),
 	)
-	muxHandler := func(h http.Handler, w http.ResponseWriter, req *http.Request) {
-		h.ServeHTTP(w, req)
-	}
 	v1Alpha1PbHandler := gateway.PbMux{
 		Registrations: v1Alpha1Registrations,
 		Patterns:      []string{"/eth/v1alpha1/"},
@@ -80,7 +75,6 @@ func DefaultConfig(enableDebugRPCEndpoints bool) MuxConfig {
 	}
 
 	return MuxConfig{
-		Handler:       muxHandler,
 		V1PbMux:       v1PbHandler,
 		V1Alpha1PbMux: v1Alpha1PbHandler,
 	}

beacon-chain/gateway/helpers_test.go

@@ -10,7 +10,6 @@ import (
 func TestDefaultConfig(t *testing.T) {
 	t.Run("Without debug endpoints", func(t *testing.T) {
 		cfg := DefaultConfig(false)
-		assert.NotNil(t, cfg.Handler)
 		assert.NotNil(t, cfg.V1PbMux.Mux)
 		require.Equal(t, 1, len(cfg.V1PbMux.Patterns))
 		assert.Equal(t, "/eth/v1/", cfg.V1PbMux.Patterns[0])
@@ -23,7 +22,6 @@ func TestDefaultConfig(t *testing.T) {
 
 	t.Run("With debug endpoints", func(t *testing.T) {
 		cfg := DefaultConfig(true)
-		assert.NotNil(t, cfg.Handler)
 		assert.NotNil(t, cfg.V1PbMux.Mux)
 		require.Equal(t, 1, len(cfg.V1PbMux.Patterns))
 		assert.Equal(t, "/eth/v1/", cfg.V1PbMux.Patterns[0])

beacon-chain/p2p/peers/BUILD.bazel

@@ -19,6 +19,7 @@ go_library(
         "@com_github_libp2p_go_libp2p_core//peer:go_default_library",
         "@com_github_multiformats_go_multiaddr//:go_default_library",
         "@com_github_multiformats_go_multiaddr//net:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
         "@com_github_prysmaticlabs_eth2_types//:go_default_library",
         "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
     ],

beacon-chain/p2p/peers/status.go

@@ -33,6 +33,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/peer"
 	ma "github.com/multiformats/go-multiaddr"
 	manet "github.com/multiformats/go-multiaddr/net"
+	"github.com/pkg/errors"
 	types "github.com/prysmaticlabs/eth2-types"
 	"github.com/prysmaticlabs/go-bitfield"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
@@ -74,6 +75,11 @@ const (
 	MaxBackOffDuration = 5000
 )
 
+// ErrNoPeerStatus is returned when there is a map entry for a given peer but there is no chain
+// status for that peer. This should happen in rare circumstances only, but is a very possible
+// scenario in a chaotic and adversarial network.
+var ErrNoPeerStatus = errors.New("no chain status for peer")
+
 // Status is the structure holding the peer status information.
 type Status struct {
 	ctx       context.Context
@@ -190,10 +196,17 @@ func (p *Status) SetChainState(pid peer.ID, chainState *pb.Status) {
 }
 
 // ChainState gets the chain state of the given remote peer.
-// This can return nil if there is no known chain state for the peer.
 // This will error if the peer does not exist.
+// This will error if there is no known chain state for the peer.
 func (p *Status) ChainState(pid peer.ID) (*pb.Status, error) {
-	return p.scorers.PeerStatusScorer().PeerStatus(pid)
+	s, err := p.scorers.PeerStatusScorer().PeerStatus(pid)
+	if err != nil {
+		return nil, err
+	}
+	if s == nil {
+		return nil, ErrNoPeerStatus
+	}
+	return s, nil
 }
 
 // IsActive checks if a peers is active and returns the result appropriately.

beacon-chain/p2p/peers/status_test.go

@@ -302,6 +302,32 @@ func TestPeerChainState(t *testing.T) {
 	}
 }
 
+func TestPeerWithNilChainState(t *testing.T) {
+	maxBadResponses := 2
+	p := peers.NewStatus(context.Background(), &peers.StatusConfig{
+		PeerLimit: 30,
+		ScorerParams: &scorers.Config{
+			BadResponsesScorerConfig: &scorers.BadResponsesScorerConfig{
+				Threshold: maxBadResponses,
+			},
+		},
+	})
+
+	id, err := peer.Decode("16Uiu2HAkyWZ4Ni1TpvDS8dPxsozmHY85KaiFjodQuV6Tz5tkHVeR")
+	require.NoError(t, err)
+	address, err := ma.NewMultiaddr("/ip4/213.202.254.180/tcp/13000")
+	require.NoError(t, err, "Failed to create address")
+	direction := network.DirInbound
+	p.Add(new(enr.Record), id, address, direction)
+
+	p.SetChainState(id, nil)
+
+	resChainState, err := p.ChainState(id)
+	require.Equal(t, peers.ErrNoPeerStatus, err)
+	var nothing *pb.Status
+	require.Equal(t, resChainState, nothing)
+}
+
 func TestPeerBadResponses(t *testing.T) {
 	maxBadResponses := 2
 	p := peers.NewStatus(context.Background(), &peers.StatusConfig{

beacon-chain/powchain/log_processing.go

@@ -140,14 +140,16 @@ func (s *Service) ProcessDepositLog(ctx context.Context, depositLog gethTypes.Lo
 	}
 	s.depositTrie.Insert(depositHash[:], int(index))
 
-	proof, err := s.depositTrie.MerkleProof(int(index))
-	if err != nil {
-		return errors.Wrap(err, "Unable to generate merkle proof for deposit")
-	}
-
 	deposit := &ethpb.Deposit{
-		Data:  depositData,
-		Proof: proof,
+		Data: depositData,
+	}
+	// Only generate the proofs during pre-genesis.
+	if !s.chainStartData.Chainstarted {
+		proof, err := s.depositTrie.MerkleProof(int(index))
+		if err != nil {
+			return errors.Wrap(err, "Unable to generate merkle proof for deposit")
+		}
+		deposit.Proof = proof
 	}
 
 	// We always store all historical deposits in the DB.

beacon-chain/state/stategen/replay.go

@@ -25,6 +25,7 @@ func (s *State) ReplayBlocks(
 	defer span.End()
 
 	var err error
+	log.Debugf("Replaying state from slot %d till slot %d", state.Slot(), targetSlot)
 	// The input block list is sorted in decreasing slots order.
 	if len(signed) > 0 {
 		for i := len(signed) - 1; i >= 0; i-- {

beacon-chain/sync/BUILD.bazel

@@ -73,6 +73,7 @@ go_library(
         "//shared/abool:go_default_library",
         "//shared/bls:go_default_library",
         "//shared/bytesutil:go_default_library",
+        "//shared/featureconfig:go_default_library",
         "//shared/interfaces:go_default_library",
         "//shared/messagehandler:go_default_library",
         "//shared/mputil:go_default_library",

beacon-chain/sync/initial-sync/round_robin.go

@@ -304,7 +304,15 @@ func (s *Service) isProcessedBlock(ctx context.Context, blk interfaces.SignedBea
 	if err != nil {
 		return false
 	}
-	if blk.Block().Slot() <= finalizedSlot || (s.cfg.DB.HasBlock(ctx, blkRoot) || s.cfg.Chain.HasInitSyncBlock(blkRoot)) {
+	// If block is before our finalized checkpoint
+	// we do not process it.
+	if blk.Block().Slot() <= finalizedSlot {
+		return true
+	}
+	blockExistsInDB := s.cfg.DB.HasBlock(ctx, blkRoot) || s.cfg.Chain.HasInitSyncBlock(blkRoot)
+	// If block exists in our db and is before or equal to our current head
+	// we ignore it.
+	if blockExistsInDB && s.cfg.Chain.HeadSlot() >= blk.Block().Slot() {
 		return true
 	}
 	return false

beacon-chain/sync/pending_attestations_queue_test.go

@@ -120,8 +120,8 @@ func TestProcessPendingAtts_HasBlockSaveUnAggregatedAtt(t *testing.T) {
 				}},
 			AttPool: attestations.NewPool(),
 		},
-		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
-		seenAttestationCache: c,
+		blkRootToPendingAtts:             make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
+		seenUnAggregatedAttestationCache: c,
 	}
 
 	sb = testutil.NewBeaconBlock()
@@ -237,8 +237,8 @@ func TestProcessPendingAtts_NoBroadcastWithBadSignature(t *testing.T) {
 				}},
 			AttPool: attestations.NewPool(),
 		},
-		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
-		seenAttestationCache: c,
+		blkRootToPendingAtts:             make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
+		seenUnAggregatedAttestationCache: c,
 	}
 
 	r.blkRootToPendingAtts[r32] = []*ethpb.SignedAggregateAttestationAndProof{{Message: aggregateAndProof, Signature: aggreSig}}
@@ -317,8 +317,8 @@ func TestProcessPendingAtts_HasBlockSaveAggregatedAtt(t *testing.T) {
 				}},
 			AttPool: attestations.NewPool(),
 		},
-		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
-		seenAttestationCache: c,
+		blkRootToPendingAtts:           make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
+		seenAggregatedAttestationCache: c,
 	}
 
 	sb = testutil.NewBeaconBlock()

beacon-chain/sync/pending_blocks_queue.go

@@ -307,9 +307,14 @@ func (s *Service) deleteBlockFromPendingQueue(slot types.Slot, b interfaces.Sign
 		return nil
 	}
 
+	// Defensive check to ignore nil blocks
+	if err := helpers.VerifyNilBeaconBlock(b); err != nil {
+		return err
+	}
+
 	newBlks := make([]interfaces.SignedBeaconBlock, 0, len(blks))
 	for _, blk := range blks {
-		if sszutil.DeepEqual(blk, b) {
+		if sszutil.DeepEqual(blk.Proto(), b.Proto()) {
 			continue
 		}
 		newBlks = append(newBlks, blk)

beacon-chain/sync/pending_blocks_queue_test.go

@@ -524,3 +524,65 @@ func TestService_ProcessPendingBlockOnCorrectSlot(t *testing.T) {
 	require.NoError(t, r.processPendingBlocks(context.Background()))
 	assert.Equal(t, 2, len(r.slotToPendingBlocks.Items()), "Incorrect size for slot to pending blocks cache")
 }
+
+func TestService_ProcessBadPendingBlocks(t *testing.T) {
+	ctx := context.Background()
+	db := dbtest.SetupDB(t)
+
+	p1 := p2ptest.NewTestP2P(t)
+	mockChain := mock.ChainService{Genesis: time.Unix(time.Now().Unix()-int64(params.BeaconConfig().SecondsPerSlot), 0),
+		FinalizedCheckPoint: &ethpb.Checkpoint{
+			Epoch: 0,
+		}}
+	r := &Service{
+		cfg: &Config{
+			P2P:      p1,
+			DB:       db,
+			Chain:    &mockChain,
+			StateGen: stategen.New(db),
+		},
+		slotToPendingBlocks: gcache.New(time.Second, 2*time.Second),
+		seenPendingBlocks:   make(map[[32]byte]bool),
+	}
+	err := r.initCaches()
+	require.NoError(t, err)
+
+	beaconState, privKeys := testutil.DeterministicGenesisState(t, 100)
+	parentBlock := testutil.NewBeaconBlock()
+	require.NoError(t, db.SaveBlock(ctx, wrapper.WrappedPhase0SignedBeaconBlock(parentBlock)))
+	bRoot, err := parentBlock.Block.HashTreeRoot()
+	require.NoError(t, err)
+	require.NoError(t, db.SaveState(ctx, beaconState, bRoot))
+	require.NoError(t, db.SaveStateSummary(ctx, &pb.StateSummary{Root: bRoot[:]}))
+	copied := beaconState.Copy()
+	require.NoError(t, copied.SetSlot(1))
+	proposerIdx, err := helpers.BeaconProposerIndex(copied)
+	require.NoError(t, err)
+
+	st, err := testutil.NewBeaconState()
+	require.NoError(t, err)
+	mockChain.Root = bRoot[:]
+	mockChain.State = st
+
+	b1 := testutil.NewBeaconBlock()
+	b1.Block.ParentRoot = bRoot[:]
+	b1.Block.Slot = 1
+	b1Root, err := b1.Block.HashTreeRoot()
+	require.NoError(t, err)
+	b1.Block.ProposerIndex = proposerIdx
+	b1.Signature, err = helpers.ComputeDomainAndSign(beaconState, 0, b1.Block, params.BeaconConfig().DomainBeaconProposer, privKeys[proposerIdx])
+	require.NoError(t, err)
+
+	b := testutil.NewBeaconBlock()
+	b.Block.Slot = 55
+	b.Block.ParentRoot = []byte{'A', 'B', 'C'}
+	bA := wrapper.WrappedPhase0SignedBeaconBlock(b)
+	assert.NoError(t, err)
+
+	// Add block1 for slot 55
+	require.NoError(t, r.insertBlockToPendingQueue(b.Block.Slot, bA, b1Root))
+	bB := wrapper.WrappedPhase0SignedBeaconBlock(testutil.NewBeaconBlock())
+	assert.NoError(t, err)
+	// remove with a different block from the same slot.
+	require.NoError(t, r.deleteBlockFromPendingQueue(b.Block.Slot, bB, b1Root))
+}

beacon-chain/sync/service.go

@@ -40,7 +40,8 @@ var _ shared.Service = (*Service)(nil)
 
 const rangeLimit = 1024
 const seenBlockSize = 1000
-const seenAttSize = 10000
+const seenUnaggregatedAttSize = 20000
+const seenAggregatedAttSize = 1024
 const seenExitSize = 100
 const seenProposerSlashingSize = 100
 const badBlockSize = 1000
@@ -85,29 +86,31 @@ type blockchainService interface {
 // Service is responsible for handling all run time p2p related operations as the
 // main entry point for network messages.
 type Service struct {
-	cfg                       *Config
-	ctx                       context.Context
-	cancel                    context.CancelFunc
-	slotToPendingBlocks       *gcache.Cache
-	seenPendingBlocks         map[[32]byte]bool
-	blkRootToPendingAtts      map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof
-	pendingAttsLock           sync.RWMutex
-	pendingQueueLock          sync.RWMutex
-	chainStarted              *abool.AtomicBool
-	validateBlockLock         sync.RWMutex
-	rateLimiter               *limiter
-	seenBlockLock             sync.RWMutex
-	seenBlockCache            *lru.Cache
-	seenAttestationLock       sync.RWMutex
-	seenAttestationCache      *lru.Cache
-	seenExitLock              sync.RWMutex
-	seenExitCache             *lru.Cache
-	seenProposerSlashingLock  sync.RWMutex
-	seenProposerSlashingCache *lru.Cache
-	seenAttesterSlashingLock  sync.RWMutex
-	seenAttesterSlashingCache map[uint64]bool
-	badBlockCache             *lru.Cache
-	badBlockLock              sync.RWMutex
+	cfg                              *Config
+	ctx                              context.Context
+	cancel                           context.CancelFunc
+	slotToPendingBlocks              *gcache.Cache
+	seenPendingBlocks                map[[32]byte]bool
+	blkRootToPendingAtts             map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof
+	pendingAttsLock                  sync.RWMutex
+	pendingQueueLock                 sync.RWMutex
+	chainStarted                     *abool.AtomicBool
+	validateBlockLock                sync.RWMutex
+	rateLimiter                      *limiter
+	seenBlockLock                    sync.RWMutex
+	seenBlockCache                   *lru.Cache
+	seenAggregatedAttestationLock    sync.RWMutex
+	seenAggregatedAttestationCache   *lru.Cache
+	seenUnAggregatedAttestationLock  sync.RWMutex
+	seenUnAggregatedAttestationCache *lru.Cache
+	seenExitLock                     sync.RWMutex
+	seenExitCache                    *lru.Cache
+	seenProposerSlashingLock         sync.RWMutex
+	seenProposerSlashingCache        *lru.Cache
+	seenAttesterSlashingLock         sync.RWMutex
+	seenAttesterSlashingCache        map[uint64]bool
+	badBlockCache                    *lru.Cache
+	badBlockLock                     sync.RWMutex
 }
 
 // NewService initializes new regular sync service.
@@ -194,7 +197,11 @@ func (s *Service) initCaches() error {
 	if err != nil {
 		return err
 	}
-	attCache, err := lru.New(seenAttSize)
+	aggregatedAttCache, err := lru.New(seenAggregatedAttSize)
+	if err != nil {
+		return err
+	}
+	unAggregatedAttCache, err := lru.New(seenUnaggregatedAttSize)
 	if err != nil {
 		return err
 	}
@@ -211,7 +218,8 @@ func (s *Service) initCaches() error {
 		return err
 	}
 	s.seenBlockCache = blkCache
-	s.seenAttestationCache = attCache
+	s.seenAggregatedAttestationCache = aggregatedAttCache
+	s.seenUnAggregatedAttestationCache = unAggregatedAttCache
 	s.seenExitCache = exitCache
 	s.seenAttesterSlashingCache = make(map[uint64]bool)
 	s.seenProposerSlashingCache = proposerSlashingCache

beacon-chain/sync/subscriber_beacon_aggregate_proof_test.go

@@ -22,7 +22,7 @@ func TestBeaconAggregateProofSubscriber_CanSaveAggregatedAttestation(t *testing.
 			AttPool:             attestations.NewPool(),
 			AttestationNotifier: (&mock.ChainService{}).OperationNotifier(),
 		},
-		seenAttestationCache: c,
+		seenUnAggregatedAttestationCache: c,
 	}
 
 	a := &ethpb.SignedAggregateAttestationAndProof{
@@ -46,7 +46,7 @@ func TestBeaconAggregateProofSubscriber_CanSaveUnaggregatedAttestation(t *testin
 			AttPool:             attestations.NewPool(),
 			AttestationNotifier: (&mock.ChainService{}).OperationNotifier(),
 		},
-		seenAttestationCache: c,
+		seenUnAggregatedAttestationCache: c,
 	}
 
 	a := &ethpb.SignedAggregateAttestationAndProof{

beacon-chain/sync/subscriber_beacon_blocks.go

@@ -8,6 +8,7 @@ import (
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/state/interop"
 	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/proto/eth/v1alpha1/wrapper"
+	"github.com/prysmaticlabs/prysm/shared/featureconfig"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -37,12 +38,13 @@ func (s *Service) beaconBlockSubscriber(ctx context.Context, msg proto.Message)
 		return err
 	}
 
-	// Delete attestations from the block in the pool to avoid inclusion in future block.
-	if err := s.deleteAttsInPool(block.Body().Attestations()); err != nil {
-		log.Debugf("Could not delete attestations in pool: %v", err)
-		return nil
+	if !featureconfig.Get().CorrectlyPruneCanonicalAtts {
+		// Delete attestations from the block in the pool to avoid inclusion in future block.
+		if err := s.deleteAttsInPool(block.Body().Attestations()); err != nil {
+			log.Debugf("Could not delete attestations in pool: %v", err)
+			return nil
+		}
 	}
-
 	return err
 }
 

beacon-chain/sync/validate_aggregate_proof.go

@@ -205,19 +205,19 @@ func (s *Service) validateBlockInAttestation(ctx context.Context, satt *ethpb.Si
 
 // Returns true if the node has received aggregate for the aggregator with index and target epoch.
 func (s *Service) hasSeenAggregatorIndexEpoch(epoch types.Epoch, aggregatorIndex types.ValidatorIndex) bool {
-	s.seenAttestationLock.RLock()
-	defer s.seenAttestationLock.RUnlock()
+	s.seenAggregatedAttestationLock.RLock()
+	defer s.seenAggregatedAttestationLock.RUnlock()
 	b := append(bytesutil.Bytes32(uint64(epoch)), bytesutil.Bytes32(uint64(aggregatorIndex))...)
-	_, seen := s.seenAttestationCache.Get(string(b))
+	_, seen := s.seenAggregatedAttestationCache.Get(string(b))
 	return seen
 }
 
 // Set aggregate's aggregator index target epoch as seen.
 func (s *Service) setAggregatorIndexEpochSeen(epoch types.Epoch, aggregatorIndex types.ValidatorIndex) {
-	s.seenAttestationLock.Lock()
-	defer s.seenAttestationLock.Unlock()
+	s.seenAggregatedAttestationLock.Lock()
+	defer s.seenAggregatedAttestationLock.Unlock()
 	b := append(bytesutil.Bytes32(uint64(epoch)), bytesutil.Bytes32(uint64(aggregatorIndex))...)
-	s.seenAttestationCache.Add(string(b), true)
+	s.seenAggregatedAttestationCache.Add(string(b), true)
 }
 
 // This validates the aggregator's index in state is within the beacon committee.

beacon-chain/sync/validate_aggregate_proof_test.go

@@ -121,8 +121,8 @@ func TestValidateAggregateAndProof_NoBlock(t *testing.T) {
 			AttPool:     attestations.NewPool(),
 			Chain:       &mock.ChainService{},
 		},
-		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
-		seenAttestationCache: c,
+		blkRootToPendingAtts:           make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
+		seenAggregatedAttestationCache: c,
 	}
 	err = r.initCaches()
 	require.NoError(t, err)
@@ -194,7 +194,7 @@ func TestValidateAggregateAndProof_NotWithinSlotRange(t *testing.T) {
 			AttPool:             attestations.NewPool(),
 			AttestationNotifier: (&mock.ChainService{}).OperationNotifier(),
 		},
-		seenAttestationCache: c,
+		seenAggregatedAttestationCache: c,
 	}
 	err = r.initCaches()
 	require.NoError(t, err)
@@ -276,8 +276,8 @@ func TestValidateAggregateAndProof_ExistedInPool(t *testing.T) {
 				State: beaconState},
 			AttestationNotifier: (&mock.ChainService{}).OperationNotifier(),
 		},
-		seenAttestationCache: c,
-		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
+		seenAggregatedAttestationCache: c,
+		blkRootToPendingAtts:           make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
 	}
 	err = r.initCaches()
 	require.NoError(t, err)
@@ -372,7 +372,7 @@ func TestValidateAggregateAndProof_CanValidate(t *testing.T) {
 			AttPool:             attestations.NewPool(),
 			AttestationNotifier: (&mock.ChainService{}).OperationNotifier(),
 		},
-		seenAttestationCache: c,
+		seenAggregatedAttestationCache: c,
 	}
 	err = r.initCaches()
 	require.NoError(t, err)
@@ -466,7 +466,7 @@ func TestVerifyIndexInCommittee_SeenAggregatorEpoch(t *testing.T) {
 			AttPool:             attestations.NewPool(),
 			AttestationNotifier: (&mock.ChainService{}).OperationNotifier(),
 		},
-		seenAttestationCache: c,
+		seenAggregatedAttestationCache: c,
 	}
 	err = r.initCaches()
 	require.NoError(t, err)
@@ -575,7 +575,7 @@ func TestValidateAggregateAndProof_BadBlock(t *testing.T) {
 			AttPool:             attestations.NewPool(),
 			AttestationNotifier: (&mock.ChainService{}).OperationNotifier(),
 		},
-		seenAttestationCache: c,
+		seenAggregatedAttestationCache: c,
 	}
 	err = r.initCaches()
 	require.NoError(t, err)
@@ -668,7 +668,7 @@ func TestValidateAggregateAndProof_RejectWhenAttEpochDoesntEqualTargetEpoch(t *t
 			AttPool:             attestations.NewPool(),
 			AttestationNotifier: (&mock.ChainService{}).OperationNotifier(),
 		},
-		seenAttestationCache: c,
+		seenAggregatedAttestationCache: c,
 	}
 	err = r.initCaches()
 	require.NoError(t, err)

beacon-chain/sync/validate_beacon_attestation.go

@@ -204,21 +204,21 @@ func (s *Service) validateUnaggregatedAttWithState(ctx context.Context, a *eth.A
 
 // Returns true if the attestation was already seen for the participating validator for the slot.
 func (s *Service) hasSeenCommitteeIndicesSlot(slot types.Slot, committeeID types.CommitteeIndex, aggregateBits []byte) bool {
-	s.seenAttestationLock.RLock()
-	defer s.seenAttestationLock.RUnlock()
+	s.seenUnAggregatedAttestationLock.RLock()
+	defer s.seenUnAggregatedAttestationLock.RUnlock()
 	b := append(bytesutil.Bytes32(uint64(slot)), bytesutil.Bytes32(uint64(committeeID))...)
 	b = append(b, aggregateBits...)
-	_, seen := s.seenAttestationCache.Get(string(b))
+	_, seen := s.seenUnAggregatedAttestationCache.Get(string(b))
 	return seen
 }
 
 // Set committee's indices and slot as seen for incoming attestations.
 func (s *Service) setSeenCommitteeIndicesSlot(slot types.Slot, committeeID types.CommitteeIndex, aggregateBits []byte) {
-	s.seenAttestationLock.Lock()
-	defer s.seenAttestationLock.Unlock()
+	s.seenUnAggregatedAttestationLock.Lock()
+	defer s.seenUnAggregatedAttestationLock.Unlock()
 	b := append(bytesutil.Bytes32(uint64(slot)), bytesutil.Bytes32(uint64(committeeID))...)
 	b = append(b, aggregateBits...)
-	s.seenAttestationCache.Add(string(b), true)
+	s.seenUnAggregatedAttestationCache.Add(string(b), true)
 }
 
 // hasBlockAndState returns true if the beacon node knows about a block and associated state in the

beacon-chain/sync/validate_beacon_attestation_test.go

@@ -45,8 +45,8 @@ func TestService_validateCommitteeIndexBeaconAttestation(t *testing.T) {
 			Chain:               chain,
 			AttestationNotifier: (&mockChain.ChainService{}).OperationNotifier(),
 		},
-		blkRootToPendingAtts: make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
-		seenAttestationCache: c,
+		blkRootToPendingAtts:             make(map[[32]byte][]*ethpb.SignedAggregateAttestationAndProof),
+		seenUnAggregatedAttestationCache: c,
 	}
 	err = s.initCaches()
 	require.NoError(t, err)

deps.bzl

@@ -1063,6 +1063,13 @@ def prysm_deps():
         sum = "h1:1r7pUrabqp18hOBcwBwiTsbnFeTZHV9eER/QT5JVZxY=",
         version = "v0.0.0-20200121045136-8c9f03a8e57e",
     )
+
+    go_repository(
+        name = "com_github_golang_jwt_jwt",
+        importpath = "github.com/golang-jwt/jwt",
+        sum = "h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keLg81eXfW3O+oY=",
+        version = "v3.2.2+incompatible",
+    )
     go_repository(
         name = "com_github_golang_lint",
         importpath = "github.com/golang/lint",
@@ -2774,8 +2781,8 @@ def prysm_deps():
     go_repository(
         name = "com_github_prysmaticlabs_go_bitfield",
         importpath = "github.com/prysmaticlabs/go-bitfield",
-        sum = "h1:9rrmgQval2GOmtMAgGLdqcCEZLraNaN3k2mY+07cx64=",
-        version = "v0.0.0-20210706153858-5cb5ce8bdbfe",
+        sum = "h1:0tVE4tdWQK9ZpYygoV7+vS6QkDvQVySboMVEIxBJmXw=",
+        version = "v0.0.0-20210809151128-385d8c5e3fb7",
     )
     go_repository(
         name = "com_github_prysmaticlabs_prombbolt",

go.mod

@@ -87,7 +87,7 @@ require (
 	github.com/prometheus/prom2json v1.3.0
 	github.com/prometheus/tsdb v0.10.0 // indirect
 	github.com/prysmaticlabs/eth2-types v0.0.0-20210303084904-c9735a06829d
-	github.com/prysmaticlabs/go-bitfield v0.0.0-20210706153858-5cb5ce8bdbfe
+	github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7
 	github.com/prysmaticlabs/prombbolt v0.0.0-20210126082820-9b7adba6db7c
 	github.com/prysmaticlabs/protoc-gen-go-cast v0.0.0-20210504233148-1e141af6a0a1
 	github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc

go.sum

@@ -1074,8 +1074,8 @@ github.com/prysmaticlabs/bazel-go-ethereum v0.0.0-20210707101027-e8523651bf6f/go
 github.com/prysmaticlabs/eth2-types v0.0.0-20210303084904-c9735a06829d h1:1dN7YAqMN3oAJ0LceWcyv/U4jHLh+5urnSnr4br6zg4=
 github.com/prysmaticlabs/eth2-types v0.0.0-20210303084904-c9735a06829d/go.mod h1:kOmQ/zdobQf7HUohDTifDNFEZfNaSCIY5fkONPL+dWU=
 github.com/prysmaticlabs/go-bitfield v0.0.0-20210108222456-8e92c3709aa0/go.mod h1:hCwmef+4qXWjv0jLDbQdWnL0Ol7cS7/lCSS26WR+u6s=
-github.com/prysmaticlabs/go-bitfield v0.0.0-20210706153858-5cb5ce8bdbfe h1:9rrmgQval2GOmtMAgGLdqcCEZLraNaN3k2mY+07cx64=
-github.com/prysmaticlabs/go-bitfield v0.0.0-20210706153858-5cb5ce8bdbfe/go.mod h1:wmuf/mdK4VMD+jA9ThwcUKjg3a2XWM9cVfFYjDyY4j4=
+github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 h1:0tVE4tdWQK9ZpYygoV7+vS6QkDvQVySboMVEIxBJmXw=
+github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7/go.mod h1:wmuf/mdK4VMD+jA9ThwcUKjg3a2XWM9cVfFYjDyY4j4=
 github.com/prysmaticlabs/grpc-gateway/v2 v2.3.1-0.20210622145107-ca3041e1b380 h1:KzQOksIZB8poBiMk8h5Txzbp/OoBLFhS3H20ZN06hWg=
 github.com/prysmaticlabs/grpc-gateway/v2 v2.3.1-0.20210622145107-ca3041e1b380/go.mod h1:IOyTYjcIO0rkmnGBfJTL0NJ11exy/Tc2QEuv7hCXp24=
 github.com/prysmaticlabs/prombbolt v0.0.0-20210126082820-9b7adba6db7c h1:9PHRCuO/VN0s9k+RmLykho7AjDxblNYI5bYKed16NPU=

shared/cmd/flags.go

@@ -257,13 +257,63 @@ func LoadFlagsFromConfig(cliCtx *cli.Context, flags []cli.Flag) error {
 // ValidateNoArgs insures that the application is not run with erroneous arguments or flags.
 // This function should be used in the app.Before, whenever the application supports a default command.
 func ValidateNoArgs(ctx *cli.Context) error {
+	commandList := ctx.App.Commands
+	parentCommand := ctx.Command
+	isParamForFlag := false
 	for _, a := range ctx.Args().Slice() {
-		if strings.HasPrefix(a, "-") {
+		// We don't validate further if
+		// the following value is actually
+		// a parameter for a flag.
+		if isParamForFlag {
+			isParamForFlag = false
 			continue
 		}
-		if c := ctx.App.Command(a); c == nil {
+		if strings.HasPrefix(a, "-") || strings.HasPrefix(a, "--") {
+			// In the event our flag doesn't specify
+			// the relevant argument with an equal
+			// sign, we can assume the next argument
+			// is the relevant value for the flag.
+			flagName := strings.TrimPrefix(a, "--")
+			flagName = strings.TrimPrefix(flagName, "-")
+			if !strings.Contains(a, "=") && !isBoolFlag(parentCommand, flagName) {
+				isParamForFlag = true
+			}
+			continue
+		}
+		c := checkCommandList(commandList, a)
+		if c == nil {
 			return fmt.Errorf("unrecognized argument: %s", a)
 		}
+		// Set the command list as the subcommand's
+		// from the current selected parent command.
+		commandList = c.Subcommands
+		parentCommand = c
+	}
+	return nil
+}
+
+// verifies that the provided command is in the command list.
+func checkCommandList(commands []*cli.Command, name string) *cli.Command {
+	for _, c := range commands {
+		if c.Name == name {
+			return c
+		}
 	}
 	return nil
 }
+
+func isBoolFlag(com *cli.Command, name string) bool {
+	for _, f := range com.Flags {
+		switch bFlag := f.(type) {
+		case *cli.BoolFlag:
+			if bFlag.Name == name {
+				return true
+			}
+		case *altsrc.BoolFlag:
+			if bFlag.Name == name {
+				return true
+			}
+		}
+	}
+	return false
+}

shared/cmd/flags_test.go

@@ -55,6 +55,24 @@ func TestValidateNoArgs(t *testing.T) {
 		Commands: []*cli.Command{
 			{
 				Name: "bar",
+				Subcommands: []*cli.Command{
+					{
+						Name: "subComm1",
+						Subcommands: []*cli.Command{
+							{
+								Name: "subComm3",
+							},
+						},
+					},
+					{
+						Name: "subComm2",
+						Subcommands: []*cli.Command{
+							{
+								Name: "subComm4",
+							},
+						},
+					},
+				},
 			},
 		},
 	}
@@ -71,4 +89,151 @@ func TestValidateNoArgs(t *testing.T) {
 	// It should fail on unregistered flag (default logic in urfave/cli).
 	err = app.Run([]string{"command", "bar", "--baz"})
 	require.ErrorContains(t, "flag provided but not defined", err)
+
+	// Handle Nested Subcommands
+
+	err = app.Run([]string{"command", "bar", "subComm1"})
+	require.NoError(t, err)
+
+	err = app.Run([]string{"command", "bar", "subComm2"})
+	require.NoError(t, err)
+
+	// Should fail from unknown subcommands.
+	err = app.Run([]string{"command", "bar", "subComm3"})
+	require.ErrorContains(t, "unrecognized argument: subComm3", err)
+
+	err = app.Run([]string{"command", "bar", "subComm4"})
+	require.ErrorContains(t, "unrecognized argument: subComm4", err)
+
+	// Should fail with invalid double nested subcommands.
+	err = app.Run([]string{"command", "bar", "subComm1", "subComm2"})
+	require.ErrorContains(t, "unrecognized argument: subComm2", err)
+
+	err = app.Run([]string{"command", "bar", "subComm1", "subComm4"})
+	require.ErrorContains(t, "unrecognized argument: subComm4", err)
+
+	err = app.Run([]string{"command", "bar", "subComm2", "subComm1"})
+	require.ErrorContains(t, "unrecognized argument: subComm1", err)
+
+	err = app.Run([]string{"command", "bar", "subComm2", "subComm3"})
+	require.ErrorContains(t, "unrecognized argument: subComm3", err)
+
+	// Should pass with correct nested double subcommands.
+	err = app.Run([]string{"command", "bar", "subComm1", "subComm3"})
+	require.NoError(t, err)
+
+	err = app.Run([]string{"command", "bar", "subComm2", "subComm4"})
+	require.NoError(t, err)
+}
+
+func TestValidateNoArgs_SubcommandFlags(t *testing.T) {
+	app := &cli.App{
+		Before: ValidateNoArgs,
+		Action: func(c *cli.Context) error {
+			return nil
+		},
+		Flags: []cli.Flag{
+			&cli.StringFlag{
+				Name: "foo",
+			},
+		},
+		Commands: []*cli.Command{
+			{
+				Name: "bar",
+				Subcommands: []*cli.Command{
+					{
+						Name: "subComm1",
+						Subcommands: []*cli.Command{
+							{
+								Name: "subComm3",
+							},
+						},
+						Flags: []cli.Flag{
+							&cli.StringFlag{
+								Name: "barfoo2",
+							},
+							&cli.BoolFlag{
+								Name: "barfoo99",
+							},
+						},
+					},
+					{
+						Name: "subComm2",
+						Subcommands: []*cli.Command{
+							{
+								Name: "subComm4",
+							},
+						},
+						Flags: []cli.Flag{
+							&cli.StringFlag{
+								Name: "barfoo3",
+							},
+							&cli.BoolFlag{
+								Name: "barfoo100",
+							},
+						},
+					},
+				},
+				Flags: []cli.Flag{
+					&cli.StringFlag{
+						Name: "barfoo1",
+					},
+				},
+			},
+		},
+	}
+
+	// It should not work with a bogus argument
+	err := app.Run([]string{"command", "foo"})
+	require.ErrorContains(t, "unrecognized argument: foo", err)
+	// It should work with registered flags
+	err = app.Run([]string{"command", "--foo=bar"})
+	require.NoError(t, err)
+
+	// It should work with registered flags with spaces.
+	err = app.Run([]string{"command", "--foo", "bar"})
+	require.NoError(t, err)
+
+	// Handle Nested Subcommands and its flags
+
+	err = app.Run([]string{"command", "bar", "--barfoo1=xyz"})
+	require.NoError(t, err)
+
+	err = app.Run([]string{"command", "bar", "--barfoo1", "xyz"})
+	require.NoError(t, err)
+
+	// Should pass with correct nested double subcommands.
+	err = app.Run([]string{"command", "bar", "subComm1", "--barfoo2=xyz"})
+	require.NoError(t, err)
+
+	err = app.Run([]string{"command", "bar", "subComm1", "--barfoo2", "xyz"})
+	require.NoError(t, err)
+
+	err = app.Run([]string{"command", "bar", "subComm2", "--barfoo3=xyz"})
+	require.NoError(t, err)
+
+	err = app.Run([]string{"command", "bar", "subComm2", "--barfoo3", "xyz"})
+	require.NoError(t, err)
+
+	err = app.Run([]string{"command", "bar", "subComm2", "--barfoo3"})
+	require.ErrorContains(t, "flag needs an argument", err)
+
+	err = app.Run([]string{"command", "bar", "subComm1", "--barfoo99"})
+	require.NoError(t, err)
+
+	// Test edge case with boolean flags, as they do not require spaced arguments.
+	app.CommandNotFound = func(context *cli.Context, s string) {
+		require.Equal(t, "garbage", s)
+	}
+	err = app.Run([]string{"command", "bar", "subComm1", "--barfoo99", "garbage"})
+	require.ErrorContains(t, "unrecognized argument: garbage", err)
+
+	err = app.Run([]string{"command", "bar", "subComm1", "--barfoo99", "garbage", "subComm3"})
+	require.ErrorContains(t, "unrecognized argument: garbage", err)
+
+	err = app.Run([]string{"command", "bar", "subComm2", "--barfoo100", "garbage"})
+	require.ErrorContains(t, "unrecognized argument: garbage", err)
+
+	err = app.Run([]string{"command", "bar", "subComm2", "--barfoo100", "garbage", "subComm4"})
+	require.ErrorContains(t, "unrecognized argument: garbage", err)
 }

shared/featureconfig/config.go

@@ -73,6 +73,10 @@ type Flags struct {
 
 	// EnableSlashingProtectionPruning for the validator client.
 	EnableSlashingProtectionPruning bool
+
+	// Bug fixes related flags.
+	CorrectlyInsertOrphanedAtts bool
+	CorrectlyPruneCanonicalAtts bool
 }
 
 var featureConfig *Flags
@@ -183,18 +187,28 @@ func ConfigureBeaconChain(ctx *cli.Context) {
 		log.WithField(enableNextSlotStateCache.Name, enableNextSlotStateCache.Usage).Warn(enabledFeatureFlag)
 		cfg.EnableNextSlotStateCache = true
 	}
-	if ctx.Bool(updateHeadTimely.Name) {
-		log.WithField(updateHeadTimely.Name, updateHeadTimely.Usage).Warn(enabledFeatureFlag)
-		cfg.UpdateHeadTimely = true
+	cfg.UpdateHeadTimely = true
+	if ctx.Bool(disableUpdateHeadTimely.Name) {
+		log.WithField(disableUpdateHeadTimely.Name, disableUpdateHeadTimely.Usage).Warn(enabledFeatureFlag)
+		cfg.UpdateHeadTimely = false
 	}
 	cfg.ProposerAttsSelectionUsingMaxCover = true
 	if ctx.Bool(disableProposerAttsSelectionUsingMaxCover.Name) {
 		log.WithField(disableProposerAttsSelectionUsingMaxCover.Name, disableProposerAttsSelectionUsingMaxCover.Usage).Warn(enabledFeatureFlag)
 		cfg.ProposerAttsSelectionUsingMaxCover = false
 	}
-	if ctx.Bool(enableOptimizedBalanceUpdate.Name) {
-		log.WithField(enableOptimizedBalanceUpdate.Name, enableOptimizedBalanceUpdate.Usage).Warn(enabledFeatureFlag)
-		cfg.EnableOptimizedBalanceUpdate = true
+	cfg.EnableOptimizedBalanceUpdate = true
+	if ctx.Bool(disableOptimizedBalanceUpdate.Name) {
+		log.WithField(disableOptimizedBalanceUpdate.Name, disableOptimizedBalanceUpdate.Usage).Warn(enabledFeatureFlag)
+		cfg.EnableOptimizedBalanceUpdate = false
+	}
+	if ctx.Bool(correctlyInsertOrphanedAtts.Name) {
+		log.WithField(correctlyInsertOrphanedAtts.Name, correctlyInsertOrphanedAtts.Usage).Warn(enabledFeatureFlag)
+		cfg.CorrectlyInsertOrphanedAtts = true
+	}
+	if ctx.Bool(correctlyPruneCanonicalAtts.Name) {
+		log.WithField(correctlyPruneCanonicalAtts.Name, correctlyPruneCanonicalAtts.Usage).Warn(enabledFeatureFlag)
+		cfg.CorrectlyPruneCanonicalAtts = true
 	}
 	Init(cfg)
 }

shared/featureconfig/deprecated_flags.go

@@ -42,6 +42,16 @@ var (
 		Usage:  deprecatedUsage,
 		Hidden: true,
 	}
+	deprecatedUpdateHeadTimely = &cli.BoolFlag{
+		Name:   "update-head-timely",
+		Usage:  deprecatedUsage,
+		Hidden: true,
+	}
+	deprecatedEnableOptimizedBalanceUpdate = &cli.BoolFlag{
+		Name:   "enable-optimized-balance-update",
+		Usage:  deprecatedUsage,
+		Hidden: true,
+	}
 )
 
 var deprecatedFlags = []cli.Flag{
@@ -52,4 +62,6 @@ var deprecatedFlags = []cli.Flag{
 	deprecatedDisableEth1DataMajorityVote,
 	deprecatedDisableBlst,
 	deprecatedProposerAttsSelectionUsingMaxCover,
+	deprecatedUpdateHeadTimely,
+	deprecatedEnableOptimizedBalanceUpdate,
 }

shared/featureconfig/flags.go

@@ -106,9 +106,9 @@ var (
 		Name:  "enable-next-slot-state-cache",
 		Usage: "Improves attesting and proposing efficiency by caching the next slot state at the end of the current slot",
 	}
-	updateHeadTimely = &cli.BoolFlag{
-		Name:  "update-head-timely",
-		Usage: "Improves update head time by updating head right after state transition",
+	disableUpdateHeadTimely = &cli.BoolFlag{
+		Name:  "disable-update-head-timely",
+		Usage: "Disables updating head right after state transition",
 	}
 	disableProposerAttsSelectionUsingMaxCover = &cli.BoolFlag{
 		Name:  "disable-proposer-atts-selection-using-max-cover",
@@ -118,9 +118,9 @@ var (
 		Name:  "enable-slashing-protection-pruning",
 		Usage: "Enables the pruning of the validator client's slashing protectin database",
 	}
-	enableOptimizedBalanceUpdate = &cli.BoolFlag{
-		Name:  "enable-optimized-balance-update",
-		Usage: "Enables the optimized method of updating validator balances.",
+	disableOptimizedBalanceUpdate = &cli.BoolFlag{
+		Name:  "disable-optimized-balance-update",
+		Usage: "Disable the optimized method of updating validator balances.",
 	}
 	enableDoppelGangerProtection = &cli.BoolFlag{
 		Name: "enable-doppelganger",
@@ -128,6 +128,16 @@ var (
 			"a foolproof method to find duplicate instances in the network. Your validator will still be" +
 			" vulnerable if it is being run in unsafe configurations.",
 	}
+	correctlyInsertOrphanedAtts = &cli.BoolFlag{
+		Name: "correctly-insert-orphaned-atts",
+		Usage: "This fixes a bug where orphaned attestations don't get reinserted back to mem pool. This improves validator profitability and overall network health," +
+			"see issue #9441 for further detail",
+	}
+	correctlyPruneCanonicalAtts = &cli.BoolFlag{
+		Name: "correctly-prune-canonical-atts",
+		Usage: "This fixes a bug where any block attestations can get incorrectly pruned. This improves validator profitability and overall network health," +
+			"see issue #9443 for further detail",
+	}
 )
 
 // devModeFlags holds list of flags that are set when development mode is on.
@@ -135,8 +145,8 @@ var devModeFlags = []cli.Flag{
 	enableLargerGossipHistory,
 	enableNextSlotStateCache,
 	forceOptMaxCoverAggregationStategy,
-	updateHeadTimely,
-	enableOptimizedBalanceUpdate,
+	correctlyInsertOrphanedAtts,
+	correctlyPruneCanonicalAtts,
 }
 
 // ValidatorFlags contains a list of all the feature flags that apply to the validator client.
@@ -186,9 +196,11 @@ var BeaconChainFlags = append(deprecatedFlags, []cli.Flag{
 	disableBroadcastSlashingFlag,
 	enableNextSlotStateCache,
 	forceOptMaxCoverAggregationStategy,
-	updateHeadTimely,
+	disableUpdateHeadTimely,
 	disableProposerAttsSelectionUsingMaxCover,
-	enableOptimizedBalanceUpdate,
+	disableOptimizedBalanceUpdate,
+	correctlyInsertOrphanedAtts,
+	correctlyPruneCanonicalAtts,
 }...)
 
 // E2EBeaconChainFlags contains a list of the beacon chain feature flags to be tested in E2E.
@@ -196,4 +208,6 @@ var E2EBeaconChainFlags = []string{
 	"--attestation-aggregation-strategy=opt_max_cover",
 	"--dev",
 	"--use-check-point-cache",
+	"--correctly-insert-orphaned-atts",
+	"--correctly-prune-canonical-atts",
 }

shared/gateway/gateway.go

@@ -131,9 +131,12 @@ func (g *Gateway) Start() {
 	}
 
 	corsMux := g.corsMiddleware(g.mux)
-	g.mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
-		g.muxHandler(corsMux, w, r)
-	})
+
+	if g.muxHandler != nil {
+		g.mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+			g.muxHandler(corsMux, w, r)
+		})
+	}
 
 	g.server = &http.Server{
 		Addr:    g.gatewayAddr,