Merge pull request #2007 from ElrondNetwork/EN-6833/Antiflood-observers

En 6833/antiflood observers

cmd/node/factory/structs.go

@@ -37,6 +37,7 @@ import (
 	"github.com/ElrondNetwork/elrond-go/process/block/preprocess"
 	"github.com/ElrondNetwork/elrond-go/process/coordinator"
 	"github.com/ElrondNetwork/elrond-go/process/economics"
+	"github.com/ElrondNetwork/elrond-go/process/factory"
 	"github.com/ElrondNetwork/elrond-go/process/factory/interceptorscontainer"
 	"github.com/ElrondNetwork/elrond-go/process/factory/metachain"
 	"github.com/ElrondNetwork/elrond-go/process/factory/shard"
@@ -765,7 +766,6 @@ func newShardInterceptorContainerFactory(
 		WhiteListHandler:        whiteListHandler,
 		WhiteListerVerifiedTxs:  whiteListerVerifiedTxs,
 		AntifloodHandler:        network.InputAntifloodHandler,
-		NonceConverter:          dataCore.Uint64ByteSliceConverter,
 	}
 	interceptorContainerFactory, err := interceptorscontainer.NewShardInterceptorsContainerFactory(shardInterceptorsContainerFactoryArgs)
 	if err != nil {
@@ -820,7 +820,6 @@ func newMetaInterceptorContainerFactory(
 		WhiteListHandler:        whiteListHandler,
 		WhiteListerVerifiedTxs:  whiteListerVerifiedTxs,
 		AntifloodHandler:        network.InputAntifloodHandler,
-		NonceConverter:          dataCore.Uint64ByteSliceConverter,
 	}
 	interceptorContainerFactory, err := interceptorscontainer.NewMetaInterceptorsContainerFactory(metaInterceptorsContainerFactoryArgs)
 	if err != nil {
@@ -1725,17 +1724,34 @@ func newValidatorStatisticsProcessor(
 	return validatorStatisticsProcessor, nil
 }
 
-// PrepareNetworkShardingCollector will create the network sharding collector and apply it to the network messenger
+// PrepareOpenTopics will set to the anti flood handler the topics for which
+// the node can receive messages from others than validators
+func PrepareOpenTopics(
+	antiflood mainFactory.P2PAntifloodHandler,
+	shardCoordinator sharding.Coordinator,
+) {
+	selfID := shardCoordinator.SelfId()
+	if selfID == core.MetachainShardId {
+		antiflood.SetTopicsForAll(core.HeartbeatTopic)
+		return
+	}
+
+	selfShardTxTopic := factory.TransactionTopic + core.CommunicationIdentifierBetweenShards(selfID, selfID)
+	antiflood.SetTopicsForAll(core.HeartbeatTopic, selfShardTxTopic)
+}
+
+// PrepareNetworkShardingCollector will create the network sharding collector and apply it to
+// the network messenger and antiflood handler
 func PrepareNetworkShardingCollector(
 	network *mainFactory.NetworkComponents,
 	config *config.Config,
 	nodesCoordinator sharding.NodesCoordinator,
 	coordinator sharding.Coordinator,
 	epochStartRegistrationHandler epochStart.RegistrationHandler,
-	epochShard uint32,
+	epochStart uint32,
 ) (*networksharding.PeerShardMapper, error) {
 
-	networkShardingCollector, err := createNetworkShardingCollector(config, nodesCoordinator, epochStartRegistrationHandler, epochShard)
+	networkShardingCollector, err := createNetworkShardingCollector(config, nodesCoordinator, epochStartRegistrationHandler, epochStart)
 	if err != nil {
 		return nil, err
 	}
@@ -1748,6 +1764,11 @@ func PrepareNetworkShardingCollector(
 		return nil, err
 	}
 
+	err = network.InputAntifloodHandler.SetPeerValidatorMapper(networkShardingCollector)
+	if err != nil {
+		return nil, err
+	}
+
 	return networkShardingCollector, nil
 }
 

cmd/node/main.go

@@ -2035,12 +2035,14 @@ func createNode(
 		nodesCoordinator,
 		shardCoordinator,
 		epochStartRegistrationHandler,
-		process.EpochStartTrigger.Epoch(),
+		process.EpochStartTrigger.MetaEpoch(),
 	)
 	if err != nil {
 		return nil, err
 	}
 
+	factory.PrepareOpenTopics(network.InputAntifloodHandler, shardCoordinator)
+
 	apiTxsByHashThrottler, err := throttler.NewNumGoRoutinesThrottler(maxNumGoRoutinesTxsByHashApi)
 	if err != nil {
 		return nil, err

consensus/mock/nodesCoordinatorMock.go

@@ -140,7 +140,7 @@ func (ncm *NodesCoordinatorMock) ComputeAdditionalLeaving([]*state.ShardValidato
 }
 
 // GetValidatorWithPublicKey -
-func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(_ []byte, _ uint32) (sharding.Validator, uint32, error) {
+func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(_ []byte) (sharding.Validator, uint32, error) {
 	panic("implement me")
 }
 

core/mock/nodesCoordinatorMock.go

@@ -19,7 +19,7 @@ type NodesCoordinatorMock struct {
 	GetValidatorsRewardsAddressesCalled      func(randomness []byte, round uint64, shardId uint32, epoch uint32) ([]string, error)
 	SetNodesPerShardsCalled                  func(nodes map[uint32][]sharding.Validator, epoch uint32) error
 	ComputeValidatorsGroupCalled             func(randomness []byte, round uint64, shardId uint32, epoch uint32) (validatorsGroup []sharding.Validator, err error)
-	GetValidatorWithPublicKeyCalled          func(publicKey []byte, epoch uint32) (validator sharding.Validator, shardId uint32, err error)
+	GetValidatorWithPublicKeyCalled          func(publicKey []byte) (validator sharding.Validator, shardId uint32, err error)
 	GetAllEligibleValidatorsPublicKeysCalled func(epoch uint32) (map[uint32][][]byte, error)
 	GetAllWaitingValidatorsPublicKeysCalled  func() (map[uint32][][]byte, error)
 	ConsensusGroupSizeCalled                 func(uint32) int
@@ -170,9 +170,9 @@ func (ncm *NodesCoordinatorMock) ConsensusGroupSize(shardId uint32) int {
 }
 
 // GetValidatorWithPublicKey -
-func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(publicKey []byte, epoch uint32) (sharding.Validator, uint32, error) {
+func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(publicKey []byte) (sharding.Validator, uint32, error) {
 	if ncm.GetValidatorWithPublicKeyCalled != nil {
-		return ncm.GetValidatorWithPublicKeyCalled(publicKey, epoch)
+		return ncm.GetValidatorWithPublicKeyCalled(publicKey)
 	}
 
 	if publicKey == nil {

dataRetriever/requestHandlers/requestHandler.go

@@ -21,6 +21,8 @@ var log = logger.GetOrCreate("dataretriever/requesthandlers")
 const minHashesToRequest = 10
 const timeToAccumulateTrieHashes = 100 * time.Millisecond
 
+//TODO move the keys definitions that are whitelisted in core and use them in InterceptedData implementations, Identifiers() function
+
 type resolverRequestHandler struct {
 	epoch                 uint32
 	shardID               uint32

epochStart/bootstrap/disabled/disabledAntiFloodHandler.go

@@ -42,6 +42,11 @@ func (a *antiFloodHandler) SetDebugger(_ process.AntifloodDebugger) error {
 func (a *antiFloodHandler) BlacklistPeer(_ core.PeerID, _ string, _ time.Duration) {
 }
 
+// IsOriginatorEligibleForTopic returns nil
+func (a *antiFloodHandler) IsOriginatorEligibleForTopic(_ core.PeerID, _ string) error {
+	return nil
+}
+
 // IsInterfaceNil returns true if there is no value under the interface
 func (a *antiFloodHandler) IsInterfaceNil() bool {
 	return a == nil

epochStart/bootstrap/disabled/disabledNodesCoordinator.go

@@ -65,7 +65,7 @@ func (n *nodesCoordinator) ComputeConsensusGroup(_ []byte, _ uint64, _ uint32, _
 }
 
 // GetValidatorWithPublicKey -
-func (n *nodesCoordinator) GetValidatorWithPublicKey(_ []byte, _ uint32) (validator sharding.Validator, shardId uint32, err error) {
+func (n *nodesCoordinator) GetValidatorWithPublicKey(_ []byte) (validator sharding.Validator, shardId uint32, err error) {
 	return nil, 0, nil
 }
 

epochStart/bootstrap/factory/epochStartInterceptorsContainerFactory.go

@@ -93,7 +93,6 @@ func NewEpochStartInterceptorsContainer(args ArgsEpochStartInterceptorContainer)
 		WhiteListHandler:        args.WhiteListHandler,
 		WhiteListerVerifiedTxs:  args.WhiteListerVerifiedTxs,
 		AntifloodHandler:        antiFloodHandler,
-		NonceConverter:          args.NonceConverter,
 	}
 
 	interceptorsContainerFactory, err := interceptorscontainer.NewMetaInterceptorsContainerFactory(containerFactoryArgs)

epochStart/bootstrap/syncEpochStartMeta.go

@@ -106,7 +106,6 @@ func NewEpochStartMetaSyncer(args ArgsNewEpochStartMetaSyncer) (*epochStartMetaS
 		HeaderIntegrityVerifier: headerIntegrityVerifier,
 		ValidityAttester:        disabled.NewValidityAttester(),
 		EpochStartTrigger:       disabled.NewEpochStartTrigger(),
-		NonceConverter:          args.NonceConverter,
 	}
 
 	interceptedMetaHdrDataFactory, err := interceptorsFactory.NewInterceptedMetaHeaderDataFactory(&argsInterceptedDataFactory)

epochStart/mock/nodesCoordinatorStub.go

@@ -147,9 +147,9 @@ func (ncm *NodesCoordinatorStub) GetSelectedPublicKeys(_ []byte, _ uint32, _ uin
 }
 
 // GetValidatorWithPublicKey -
-func (ncm *NodesCoordinatorStub) GetValidatorWithPublicKey(address []byte, _ uint32) (sharding.Validator, uint32, error) {
+func (ncm *NodesCoordinatorStub) GetValidatorWithPublicKey(publicKey []byte) (sharding.Validator, uint32, error) {
 	if ncm.GetValidatorWithPublicKeyCalled != nil {
-		return ncm.GetValidatorWithPublicKeyCalled(address)
+		return ncm.GetValidatorWithPublicKeyCalled(publicKey)
 	}
 	return nil, 0, nil
 }

factory/interface.go

@@ -34,7 +34,10 @@ type P2PAntifloodHandler interface {
 	ResetForTopic(topic string)
 	SetMaxMessagesForTopic(topic string, maxNum uint32)
 	SetDebugger(debugger process.AntifloodDebugger) error
+	SetPeerValidatorMapper(validatorMapper process.PeerValidatorMapper) error
+	SetTopicsForAll(topics ...string)
 	ApplyConsensusSize(size int)
 	BlacklistPeer(peer core.PeerID, reason string, duration time.Duration)
+	IsOriginatorEligibleForTopic(pid core.PeerID, topic string) error
 	IsInterfaceNil() bool
 }

heartbeat/mock/nodesCoordinatorMock.go

@@ -151,7 +151,7 @@ func (ncm *NodesCoordinatorMock) GetSelectedPublicKeys(_ []byte, _ uint32, _ uin
 }
 
 // GetValidatorWithPublicKey -
-func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(_ []byte, _ uint32) (sharding.Validator, uint32, error) {
+func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(_ []byte) (sharding.Validator, uint32, error) {
 	panic("implement me")
 }
 

integrationTests/mock/nilAntifloodHandler.go

@@ -44,6 +44,11 @@ func (nah *NilAntifloodHandler) ApplyConsensusSize(_ int) {
 func (nah *NilAntifloodHandler) BlacklistPeer(_ core.PeerID, _ string, _ time.Duration) {
 }
 
+// IsOriginatorEligibleForTopic returns nil
+func (nah *NilAntifloodHandler) IsOriginatorEligibleForTopic(_ core.PeerID, _ string) error {
+	return nil
+}
+
 // IsInterfaceNil returns true if there is no value under the interface
 func (nah *NilAntifloodHandler) IsInterfaceNil() bool {
 	return nah == nil

integrationTests/mock/nodesCoordinatorMock.go

@@ -139,9 +139,9 @@ func (ncm *NodesCoordinatorMock) GetSelectedPublicKeys(_ []byte, _ uint32, _ uin
 }
 
 // GetValidatorWithPublicKey -
-func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(address []byte, _ uint32) (sharding.Validator, uint32, error) {
+func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(publicKey []byte) (sharding.Validator, uint32, error) {
 	if ncm.GetValidatorWithPublicKeyCalled != nil {
-		return ncm.GetValidatorWithPublicKeyCalled(address)
+		return ncm.GetValidatorWithPublicKeyCalled(publicKey)
 	}
 	return nil, 0, nil
 }

integrationTests/testProcessorNode.go

@@ -688,7 +688,6 @@ func (tpn *TestProcessorNode) initInterceptors() {
 			WhiteListHandler:        tpn.WhiteListHandler,
 			WhiteListerVerifiedTxs:  tpn.WhiteListerVerifiedTxs,
 			AntifloodHandler:        &mock.NilAntifloodHandler{},
-			NonceConverter:          TestUint64Converter,
 		}
 		interceptorContainerFactory, _ := interceptorscontainer.NewMetaInterceptorsContainerFactory(metaIntercContFactArgs)
 
@@ -747,7 +746,6 @@ func (tpn *TestProcessorNode) initInterceptors() {
 			WhiteListHandler:        tpn.WhiteListHandler,
 			WhiteListerVerifiedTxs:  tpn.WhiteListerVerifiedTxs,
 			AntifloodHandler:        &mock.NilAntifloodHandler{},
-			NonceConverter:          TestUint64Converter,
 		}
 		interceptorContainerFactory, _ := interceptorscontainer.NewShardInterceptorsContainerFactory(shardInterContFactArgs)
 

node/mock/nodesCoordinatorMock.go

@@ -146,7 +146,7 @@ func (ncm *NodesCoordinatorMock) GetSelectedPublicKeys(_ []byte, _ uint32, _ uin
 }
 
 // GetValidatorWithPublicKey -
-func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(_ []byte, _ uint32) (sharding.Validator, uint32, error) {
+func (ncm *NodesCoordinatorMock) GetValidatorWithPublicKey(_ []byte) (sharding.Validator, uint32, error) {
 	panic("implement me")
 }
 

process/block/interceptedBlocks/argInterceptedBlockHeader.go

@@ -1,7 +1,6 @@
 package interceptedBlocks
 
 import (
-	"github.com/ElrondNetwork/elrond-go/data/typeConverters"
 	"github.com/ElrondNetwork/elrond-go/hashing"
 	"github.com/ElrondNetwork/elrond-go/marshal"
 	"github.com/ElrondNetwork/elrond-go/process"
@@ -18,5 +17,4 @@ type ArgInterceptedBlockHeader struct {
 	HeaderIntegrityVerifier process.InterceptedHeaderIntegrityVerifier
 	ValidityAttester        process.ValidityAttester
 	EpochStartTrigger       process.EpochStartTriggerHandler
-	NonceConverter          typeConverters.Uint64ByteSliceConverter
 }

process/block/interceptedBlocks/common.go

@@ -37,9 +37,6 @@ func checkBlockHeaderArgument(arg *ArgInterceptedBlockHeader) error {
 	if check.IfNil(arg.ValidityAttester) {
 		return process.ErrNilValidityAttester
 	}
-	if check.IfNil(arg.NonceConverter) {
-		return process.ErrNilUint64Converter
-	}
 
 	return nil
 }

process/block/interceptedBlocks/common_test.go

@@ -19,7 +19,6 @@ func createDefaultBlockHeaderArgument() *ArgInterceptedBlockHeader {
 		HeaderIntegrityVerifier: &mock.HeaderIntegrityVerifierStub{},
 		ValidityAttester:        &mock.ValidityAttesterStub{},
 		EpochStartTrigger:       &mock.EpochStartTriggerStub{},
-		NonceConverter:          mock.NewNonceHashConverterMock(),
 	}
 
 	return arg
@@ -135,17 +134,6 @@ func TestCheckBlockHeaderArgument_NilValidityAttesterShouldErr(t *testing.T) {
 	assert.Equal(t, process.ErrNilValidityAttester, err)
 }
 
-func TestCheckBlockHeaderArgument_NilNonceConverterShouldErr(t *testing.T) {
-	t.Parallel()
-
-	arg := createDefaultBlockHeaderArgument()
-	arg.NonceConverter = nil
-
-	err := checkBlockHeaderArgument(arg)
-
-	assert.Equal(t, process.ErrNilUint64Converter, err)
-}
-
 func TestCheckBlockHeaderArgument_ShouldWork(t *testing.T) {
 	t.Parallel()
 

process/block/interceptedBlocks/interceptedBlockHeader.go

@@ -7,7 +7,6 @@ import (
 	"github.com/ElrondNetwork/elrond-go/core"
 	"github.com/ElrondNetwork/elrond-go/data"
 	"github.com/ElrondNetwork/elrond-go/data/block"
-	"github.com/ElrondNetwork/elrond-go/data/typeConverters"
 	"github.com/ElrondNetwork/elrond-go/hashing"
 	"github.com/ElrondNetwork/elrond-go/marshal"
 	"github.com/ElrondNetwork/elrond-go/process"
@@ -29,7 +28,6 @@ type InterceptedHeader struct {
 	isForCurrentShard bool
 	validityAttester  process.ValidityAttester
 	epochStartTrigger process.EpochStartTriggerHandler
-	nonceConverter    typeConverters.Uint64ByteSliceConverter
 }
 
 // NewInterceptedHeader creates a new instance of InterceptedHeader struct
@@ -52,7 +50,6 @@ func NewInterceptedHeader(arg *ArgInterceptedBlockHeader) (*InterceptedHeader, e
 		shardCoordinator:  arg.ShardCoordinator,
 		validityAttester:  arg.ValidityAttester,
 		epochStartTrigger: arg.EpochStartTrigger,
-		nonceConverter:    arg.NonceConverter,
 	}
 	inHdr.processFields(arg.HdrBuff)
 
@@ -192,9 +189,10 @@ func (inHdr *InterceptedHeader) String() string {
 
 // Identifiers returns the identifiers used in requests
 func (inHdr *InterceptedHeader) Identifiers() [][]byte {
-	nonceBytes := inHdr.nonceConverter.ToByteSlice(inHdr.hdr.Nonce)
+	keyNonce := []byte(fmt.Sprintf("%d-%d", inHdr.hdr.ShardID, inHdr.hdr.Nonce))
+	keyEpoch := []byte(core.EpochStartIdentifier(inHdr.hdr.Epoch))
 
-	return [][]byte{inHdr.hash, nonceBytes}
+	return [][]byte{inHdr.hash, keyNonce, keyEpoch}
 }
 
 // IsInterfaceNil returns true if there is no value under the interface

process/block/interceptedBlocks/interceptedBlockHeader_test.go

@@ -30,7 +30,6 @@ func createDefaultShardArgument() *interceptedBlocks.ArgInterceptedBlockHeader {
 		HeaderIntegrityVerifier: &mock.HeaderIntegrityVerifierStub{},
 		ValidityAttester:        &mock.ValidityAttesterStub{},
 		EpochStartTrigger:       &mock.EpochStartTriggerStub{},
-		NonceConverter:          mock.NewNonceHashConverterMock(),
 	}
 
 	hdr := createMockShardHeader()

process/block/interceptedBlocks/interceptedMetaBlockHeader.go

@@ -3,9 +3,9 @@ package interceptedBlocks
 import (
 	"fmt"
 
+	"github.com/ElrondNetwork/elrond-go/core"
 	"github.com/ElrondNetwork/elrond-go/data"
 	"github.com/ElrondNetwork/elrond-go/data/block"
-	"github.com/ElrondNetwork/elrond-go/data/typeConverters"
 	"github.com/ElrondNetwork/elrond-go/hashing"
 	"github.com/ElrondNetwork/elrond-go/marshal"
 	"github.com/ElrondNetwork/elrond-go/process"
@@ -25,7 +25,6 @@ type InterceptedMetaHeader struct {
 	hash              []byte
 	validityAttester  process.ValidityAttester
 	epochStartTrigger process.EpochStartTriggerHandler
-	nonceConverter    typeConverters.Uint64ByteSliceConverter
 }
 
 // NewInterceptedMetaHeader creates a new instance of InterceptedMetaHeader struct
@@ -48,7 +47,6 @@ func NewInterceptedMetaHeader(arg *ArgInterceptedBlockHeader) (*InterceptedMetaH
 		shardCoordinator:  arg.ShardCoordinator,
 		validityAttester:  arg.ValidityAttester,
 		epochStartTrigger: arg.EpochStartTrigger,
-		nonceConverter:    arg.NonceConverter,
 	}
 	inHdr.processFields(arg.HdrBuff)
 
@@ -147,9 +145,10 @@ func (imh *InterceptedMetaHeader) String() string {
 
 // Identifiers returns the identifiers used in requests
 func (imh *InterceptedMetaHeader) Identifiers() [][]byte {
-	nonceBytes := imh.nonceConverter.ToByteSlice(imh.hdr.Nonce)
+	keyNonce := []byte(fmt.Sprintf("%d-%d", core.MetachainShardId, imh.hdr.Nonce))
+	keyEpoch := []byte(core.EpochStartIdentifier(imh.hdr.Epoch))
 
-	return [][]byte{imh.hash, nonceBytes}
+	return [][]byte{imh.hash, keyNonce, keyEpoch}
 }
 
 // IsInterfaceNil returns true if there is no value under the interface

process/block/interceptedBlocks/interceptedMetaBlockHeader_test.go

@@ -22,7 +22,6 @@ func createDefaultMetaArgument() *interceptedBlocks.ArgInterceptedBlockHeader {
 		HeaderIntegrityVerifier: &mock.HeaderIntegrityVerifierStub{},
 		ValidityAttester:        &mock.ValidityAttesterStub{},
 		EpochStartTrigger:       &mock.EpochStartTriggerStub{},
-		NonceConverter:          mock.NewNonceHashConverterMock(),
 	}
 
 	hdr := createMockMetaHeader()

process/errors.go

@@ -829,3 +829,9 @@ var ErrInvalidUnitValue = errors.New("invalid unit value")
 
 // ErrInvalidBadPeerThreshold signals that an invalid bad peer threshold has been provided
 var ErrInvalidBadPeerThreshold = errors.New("invalid bad peer threshold")
+
+// ErrNilPeerValidatorMapper signals that nil peer validator mapper has been provided
+var ErrNilPeerValidatorMapper = errors.New("nil peer validator mapper")
+
+// ErrOnlyValidatorsCanUseThisTopic signals that topic can be used by validator only
+var ErrOnlyValidatorsCanUseThisTopic = errors.New("only validators can use this topic")