cherry picked PR for #10274 (#10300)

* ignore topic messages (except block topic) during optimistic sync

* address review comments

* nit pick fix

Co-authored-by: Nishant Das <nishdas93@gmail.com>

beacon-chain/sync/validate_aggregate_proof.go

@@ -41,6 +41,16 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms
 		return pubsub.ValidationIgnore, nil
 	}
 
+	// We should not attempt to process this message if the node is running in optimistic mode.
+	// We just ignore in p2p so that the peer is not penalized.
+	optimistic, err := s.cfg.chain.IsOptimistic(ctx)
+	if err != nil {
+		return pubsub.ValidationReject, err
+	}
+	if optimistic {
+		return pubsub.ValidationIgnore, nil
+	}
+
 	raw, err := s.decodePubsubMessage(msg)
 	if err != nil {
 		tracing.AnnotateError(span, err)

beacon-chain/sync/validate_aggregate_proof_test.go

@@ -686,3 +686,35 @@ func TestValidateAggregateAndProof_RejectWhenAttEpochDoesntEqualTargetEpoch(t *t
 	assert.NotNil(t, err)
 	assert.Equal(t, pubsub.ValidationReject, res)
 }
+
+func TestValidateAggregateAndProof_Optimistic(t *testing.T) {
+	p := p2ptest.NewTestP2P(t)
+	ctx := context.Background()
+
+	exit, s := setupValidExit(t)
+
+	r := &Service{
+		cfg: &config{
+			p2p: p,
+			chain: &mock.ChainService{
+				State:      s,
+				Optimistic: true,
+			},
+			initialSync: &mockSync.Sync{IsSyncing: false},
+		},
+	}
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, exit)
+	require.NoError(t, err)
+	topic := p2p.GossipTypeMapping[reflect.TypeOf(exit)]
+	m := &pubsub.Message{
+		Message: &pubsubpb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		},
+	}
+	res, err := r.validateAggregateAndProof(ctx, "", m)
+	assert.NoError(t, err)
+	valid := res == pubsub.ValidationIgnore
+	assert.Equal(t, true, valid, "Validation should have ignored the message")
+}

beacon-chain/sync/validate_attester_slashing.go

@@ -26,6 +26,16 @@ func (s *Service) validateAttesterSlashing(ctx context.Context, pid peer.ID, msg
 		return pubsub.ValidationIgnore, nil
 	}
 
+	// We should not attempt to process this message if the node is running in optimistic mode.
+	// We just ignore in p2p so that the peer is not penalized.
+	optimistic, err := s.cfg.chain.IsOptimistic(ctx)
+	if err != nil {
+		return pubsub.ValidationReject, err
+	}
+	if optimistic {
+		return pubsub.ValidationIgnore, nil
+	}
+
 	ctx, span := trace.StartSpan(ctx, "sync.validateAttesterSlashing")
 	defer span.End()
 

beacon-chain/sync/validate_attester_slashing_test.go

@@ -293,3 +293,34 @@ func TestSeenAttesterSlashingIndices(t *testing.T) {
 		assert.Equal(t, tc.seen, r.hasSeenAttesterSlashingIndices(tc.checkIndices1, tc.checkIndices2))
 	}
 }
+
+func TestValidateAttesterSlashing_Optimistic(t *testing.T) {
+	p := p2ptest.NewTestP2P(t)
+	ctx := context.Background()
+
+	slashing, s := setupValidAttesterSlashing(t)
+
+	r := &Service{
+		cfg: &config{
+			p2p:         p,
+			chain:       &mock.ChainService{State: s, Optimistic: true},
+			initialSync: &mockSync.Sync{IsSyncing: false},
+		},
+	}
+
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, slashing)
+	require.NoError(t, err)
+
+	topic := p2p.GossipTypeMapping[reflect.TypeOf(slashing)]
+	msg := &pubsub.Message{
+		Message: &pubsubpb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		},
+	}
+	res, err := r.validateAttesterSlashing(ctx, "foobar", msg)
+	assert.NoError(t, err)
+	valid := res == pubsub.ValidationIgnore
+	assert.Equal(t, true, valid, "Should have ignore this message")
+}

beacon-chain/sync/validate_beacon_attestation.go

@@ -40,6 +40,17 @@ func (s *Service) validateCommitteeIndexBeaconAttestation(ctx context.Context, p
 	if s.cfg.initialSync.Syncing() {
 		return pubsub.ValidationIgnore, nil
 	}
+
+	// We should not attempt to process this message if the node is running in optimistic mode.
+	// We just ignore in p2p so that the peer is not penalized.
+	optimistic, err := s.cfg.chain.IsOptimistic(ctx)
+	if err != nil {
+		return pubsub.ValidationReject, err
+	}
+	if optimistic {
+		return pubsub.ValidationIgnore, nil
+	}
+
 	ctx, span := trace.StartSpan(ctx, "sync.validateCommitteeIndexBeaconAttestation")
 	defer span.End()
 

beacon-chain/sync/validate_beacon_attestation_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"reflect"
 	"testing"
 	"time"
 
@@ -14,6 +15,7 @@ import (
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/signing"
 	dbtest "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
+	"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
 	p2ptest "github.com/prysmaticlabs/prysm/beacon-chain/p2p/testing"
 	mockSync "github.com/prysmaticlabs/prysm/beacon-chain/sync/initial-sync/testing"
 	lruwrpr "github.com/prysmaticlabs/prysm/cache/lru"
@@ -22,6 +24,7 @@ import (
 	"github.com/prysmaticlabs/prysm/encoding/bytesutil"
 	ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1/wrapper"
+	"github.com/prysmaticlabs/prysm/testing/assert"
 	"github.com/prysmaticlabs/prysm/testing/require"
 	"github.com/prysmaticlabs/prysm/testing/util"
 )
@@ -298,3 +301,34 @@ func TestService_validateCommitteeIndexBeaconAttestation(t *testing.T) {
 		})
 	}
 }
+
+func TestServiceValidateCommitteeIndexBeaconAttestation_Optimistic(t *testing.T) {
+	p := p2ptest.NewTestP2P(t)
+	ctx := context.Background()
+
+	slashing, s := setupValidAttesterSlashing(t)
+
+	r := &Service{
+		cfg: &config{
+			p2p:         p,
+			chain:       &mockChain.ChainService{State: s, Optimistic: true},
+			initialSync: &mockSync.Sync{IsSyncing: false},
+		},
+	}
+
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, slashing)
+	require.NoError(t, err)
+
+	topic := p2p.GossipTypeMapping[reflect.TypeOf(slashing)]
+	msg := &pubsub.Message{
+		Message: &pubsubpb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		},
+	}
+	res, err := r.validateCommitteeIndexBeaconAttestation(ctx, "foobar", msg)
+	assert.NoError(t, err)
+	valid := res == pubsub.ValidationIgnore
+	assert.Equal(t, true, valid, "Should have ignore this message")
+}

beacon-chain/sync/validate_proposer_slashing.go

@@ -26,6 +26,16 @@ func (s *Service) validateProposerSlashing(ctx context.Context, pid peer.ID, msg
 		return pubsub.ValidationIgnore, nil
 	}
 
+	// We should not attempt to process this message if the node is running in optimistic mode.
+	// We just ignore in p2p so that the peer is not penalized.
+	optimistic, err := s.cfg.chain.IsOptimistic(ctx)
+	if err != nil {
+		return pubsub.ValidationReject, err
+	}
+	if optimistic {
+		return pubsub.ValidationIgnore, nil
+	}
+
 	ctx, span := trace.StartSpan(ctx, "sync.validateProposerSlashing")
 	defer span.End()
 

beacon-chain/sync/validate_proposer_slashing_test.go

@@ -209,3 +209,33 @@ func TestValidateProposerSlashing_Syncing(t *testing.T) {
 	valid := res == pubsub.ValidationAccept
 	assert.Equal(t, false, valid, "Did not fail validation")
 }
+
+func TestValidateProposerSlashing_Optimistic(t *testing.T) {
+	p := p2ptest.NewTestP2P(t)
+	ctx := context.Background()
+
+	slashing, s := setupValidProposerSlashing(t)
+
+	r := &Service{
+		cfg: &config{
+			p2p:         p,
+			chain:       &mock.ChainService{State: s, Optimistic: true},
+			initialSync: &mockSync.Sync{IsSyncing: false},
+		},
+	}
+
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, slashing)
+	require.NoError(t, err)
+	topic := p2p.GossipTypeMapping[reflect.TypeOf(slashing)]
+	m := &pubsub.Message{
+		Message: &pubsubpb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		},
+	}
+	res, err := r.validateProposerSlashing(ctx, "", m)
+	assert.NoError(t, err)
+	valid := res == pubsub.ValidationIgnore
+	assert.Equal(t, true, valid, "Did not ignore the message")
+}

beacon-chain/sync/validate_sync_committee_message.go

@@ -57,6 +57,17 @@ func (s *Service) validateSyncCommitteeMessage(
 	if s.cfg.initialSync.Syncing() {
 		return pubsub.ValidationIgnore, nil
 	}
+
+	// We should not attempt to process this message if the node is running in optimistic mode.
+	// We just ignore in p2p so that the peer is not penalized.
+	optimistic, err := s.cfg.chain.IsOptimistic(ctx)
+	if err != nil {
+		return pubsub.ValidationReject, err
+	}
+	if optimistic {
+		return pubsub.ValidationIgnore, nil
+	}
+
 	if msg.Topic == nil {
 		return pubsub.ValidationReject, errInvalidTopic
 	}

beacon-chain/sync/validate_sync_committee_message_test.go

@@ -1,6 +1,7 @@
 package sync
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"reflect"
@@ -561,3 +562,34 @@ func Test_ignoreEmptyCommittee(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateSyncCommitteeMessage_Optimistic(t *testing.T) {
+	p := mockp2p.NewTestP2P(t)
+	ctx := context.Background()
+
+	slashing, s := setupValidAttesterSlashing(t)
+
+	r := &Service{
+		cfg: &config{
+			p2p:         p,
+			chain:       &mockChain.ChainService{State: s, Optimistic: true},
+			initialSync: &mockSync.Sync{IsSyncing: false},
+		},
+	}
+
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, slashing)
+	require.NoError(t, err)
+
+	topic := p2p.GossipTypeMapping[reflect.TypeOf(slashing)]
+	msg := &pubsub.Message{
+		Message: &pubsub_pb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		},
+	}
+	res, err := r.validateCommitteeIndexBeaconAttestation(ctx, "foobar", msg)
+	assert.NoError(t, err)
+	valid := res == pubsub.ValidationIgnore
+	assert.Equal(t, true, valid, "Should have ignore this message")
+}

beacon-chain/sync/validate_sync_contribution_proof.go

@@ -52,6 +52,17 @@ func (s *Service) validateSyncContributionAndProof(ctx context.Context, pid peer
 	if s.cfg.initialSync.Syncing() {
 		return pubsub.ValidationIgnore, nil
 	}
+
+	// We should not attempt to process this message if the node is running in optimistic mode.
+	// We just ignore in p2p so that the peer is not penalized.
+	optimistic, err := s.cfg.chain.IsOptimistic(ctx)
+	if err != nil {
+		return pubsub.ValidationReject, err
+	}
+	if optimistic {
+		return pubsub.ValidationIgnore, nil
+	}
+
 	m, err := s.readSyncContributionMessage(msg)
 	if err != nil {
 		tracing.AnnotateError(span, err)

beacon-chain/sync/validate_sync_contribution_proof_test.go

@@ -1,8 +1,10 @@
 package sync
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"reflect"
 	"testing"
 	"time"
 
@@ -890,7 +892,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) {
 	}
 }
 
-func TestService_ValidateSyncContributionAndProof_Broadcast(t *testing.T) {
+func TestValidateSyncContributionAndProof(t *testing.T) {
 	ctx := context.Background()
 	db := testingDB.SetupDB(t)
 	headRoot, keys := fillUpBlocksAndState(ctx, t, db)
@@ -1026,6 +1028,37 @@ func TestService_ValidateSyncContributionAndProof_Broadcast(t *testing.T) {
 	}
 }
 
+func TestValidateSyncContributionAndProof_Optimistic(t *testing.T) {
+	p := mockp2p.NewTestP2P(t)
+	ctx := context.Background()
+
+	slashing, s := setupValidAttesterSlashing(t)
+
+	r := &Service{
+		cfg: &config{
+			p2p:         p,
+			chain:       &mockChain.ChainService{State: s, Optimistic: true},
+			initialSync: &mockSync.Sync{IsSyncing: false},
+		},
+	}
+
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, slashing)
+	require.NoError(t, err)
+
+	topic := p2p.GossipTypeMapping[reflect.TypeOf(slashing)]
+	msg := &pubsub.Message{
+		Message: &pubsub_pb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		},
+	}
+	res, err := r.validateCommitteeIndexBeaconAttestation(ctx, "foobar", msg)
+	assert.NoError(t, err)
+	valid := res == pubsub.ValidationIgnore
+	assert.Equal(t, true, valid, "Should have ignore this message")
+}
+
 func fillUpBlocksAndState(ctx context.Context, t *testing.T, beaconDB db.Database) ([32]byte, []bls.SecretKey) {
 	gs, keys := util.DeterministicGenesisStateAltair(t, 64)
 	sCom, err := altair.NextSyncCommittee(ctx, gs)

beacon-chain/sync/validate_voluntary_exit.go

@@ -29,6 +29,16 @@ func (s *Service) validateVoluntaryExit(ctx context.Context, pid peer.ID, msg *p
 		return pubsub.ValidationIgnore, nil
 	}
 
+	// We should not attempt to process this message if the node is running in optimistic mode.
+	// We just ignore in p2p so that the peer is not penalized.
+	optimistic, err := s.cfg.chain.IsOptimistic(ctx)
+	if err != nil {
+		return pubsub.ValidationReject, err
+	}
+	if optimistic {
+		return pubsub.ValidationIgnore, nil
+	}
+
 	ctx, span := trace.StartSpan(ctx, "sync.validateVoluntaryExit")
 	defer span.End()
 

beacon-chain/sync/validate_voluntary_exit_test.go

@@ -195,3 +195,35 @@ func TestValidateVoluntaryExit_ValidExit_Syncing(t *testing.T) {
 	valid := res == pubsub.ValidationAccept
 	assert.Equal(t, false, valid, "Validation should have failed")
 }
+
+func TestValidateVoluntaryExit_Optimistic(t *testing.T) {
+	p := p2ptest.NewTestP2P(t)
+	ctx := context.Background()
+
+	exit, s := setupValidExit(t)
+
+	r := &Service{
+		cfg: &config{
+			p2p: p,
+			chain: &mock.ChainService{
+				State:      s,
+				Optimistic: true,
+			},
+			initialSync: &mockSync.Sync{IsSyncing: false},
+		},
+	}
+	buf := new(bytes.Buffer)
+	_, err := p.Encoding().EncodeGossip(buf, exit)
+	require.NoError(t, err)
+	topic := p2p.GossipTypeMapping[reflect.TypeOf(exit)]
+	m := &pubsub.Message{
+		Message: &pubsubpb.Message{
+			Data:  buf.Bytes(),
+			Topic: &topic,
+		},
+	}
+	res, err := r.validateVoluntaryExit(ctx, "", m)
+	assert.NoError(t, err)
+	valid := res == pubsub.ValidationIgnore
+	assert.Equal(t, true, valid, "Validation should have ignored the message")
+}