En 6833/antiflood observers

cmd/node/factory/structs.go

@@ -1750,10 +1750,10 @@ func PrepareNetworkShardingCollector(
 	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
 	}

cmd/node/main.go

@@ -2038,7 +2038,7 @@ func createNode(
 		nodesCoordinator,
 		shardCoordinator,
 		epochStartRegistrationHandler,
-		process.EpochStartTrigger.Epoch(),
+		process.EpochStartTrigger.MetaEpoch(),
 	)
 	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 {

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/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
 }

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/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
 }

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/interceptors/multiDataInterceptor.go

@@ -170,7 +170,7 @@ func (mdi *MultiDataInterceptor) ProcessReceivedMessage(message p2p.MessageP2P,
 		log.Debug("got message from peer on topic only for validators", "originator",
 			p2p.PeerIdToShortString(message.Peer()),
 			"topic", mdi.topic,
-			"err", err)
+			"err", errOriginator)
 	}
 
 	return lastErrEncountered

process/interceptors/singleDataInterceptor.go

@@ -101,7 +101,7 @@ func (sdi *SingleDataInterceptor) ProcessReceivedMessage(message p2p.MessageP2P,
 	if !isWhiteListed && errOriginator != nil {
 		log.Debug("got message from peer on topic only for validators",
 			"originator", p2p.PeerIdToShortString(message.Peer()), "topic",
-			sdi.topic, "err", err)
+			sdi.topic, "err", errOriginator)
 		sdi.throttler.EndProcessing()
 		return errOriginator
 	}

process/mock/nodesCoordinatorMock.go

@@ -21,7 +21,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() (map[uint32][][]byte, error)
 	GetAllWaitingValidatorsPublicKeysCalled  func() (map[uint32][][]byte, error)
 	GetAllLeavingValidatorsPublicKeysCalled  func() (map[uint32][][]byte, error)
@@ -214,9 +214,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 {

process/peer/process_test.go

@@ -1400,7 +1400,7 @@ func TestValidatorStatisticsProcessor_CheckForMissedBlocksWithRoundDifferenceGre
 		GetAllEligibleValidatorsPublicKeysCalled: func() (map[uint32][][]byte, error) {
 			return validatorPublicKeys, nil
 		},
-		GetValidatorWithPublicKeyCalled: func(publicKey []byte, _ uint32) (sharding.Validator, uint32, error) {
+		GetValidatorWithPublicKeyCalled: func(publicKey []byte) (sharding.Validator, uint32, error) {
 			validator, _ := sharding.NewValidator(publicKey, defaultChancesSelection, 1)
 			return validator, 0, nil
 		},
@@ -1459,7 +1459,7 @@ func TestValidatorStatisticsProcessor_CheckForMissedBlocksWithRoundDifferenceGre
 		GetAllEligibleValidatorsPublicKeysCalled: func() (map[uint32][][]byte, error) {
 			return validatorPublicKeys, nil
 		},
-		GetValidatorWithPublicKeyCalled: func(publicKey []byte, _ uint32) (sharding.Validator, uint32, error) {
+		GetValidatorWithPublicKeyCalled: func(publicKey []byte) (sharding.Validator, uint32, error) {
 			validator, _ := sharding.NewValidator(publicKey, defaultChancesSelection, 1)
 			return validator, 0, nil
 		},
@@ -1661,7 +1661,7 @@ func DoComputeMissingBlocks(
 		ConsensusGroupSizeCalled: func(uint32) int {
 			return consensusGroupSize
 		},
-		GetValidatorWithPublicKeyCalled: func(publicKey []byte, _ uint32) (sharding.Validator, uint32, error) {
+		GetValidatorWithPublicKeyCalled: func(publicKey []byte) (sharding.Validator, uint32, error) {
 			validator, _ := sharding.NewValidator(publicKey, defaultChancesSelection, 1)
 			return validator, 0, nil
 		},

sharding/indexHashedNodesCoordinator.go

@@ -53,15 +53,14 @@ func (v validatorList) Less(i, j int) bool {
 
 // TODO: add a parameter for shardID  when acting as observer
 type epochNodesConfig struct {
-	nbShards                uint32
-	shardID                 uint32
-	eligibleMap             map[uint32][]Validator
-	waitingMap              map[uint32][]Validator
-	selectors               map[uint32]RandomSelector
-	publicKeyToValidatorMap map[string]*validatorWithShardID
-	leavingMap              map[uint32][]Validator
-	newList                 []Validator
-	mutNodesMaps            sync.RWMutex
+	nbShards     uint32
+	shardID      uint32
+	eligibleMap  map[uint32][]Validator
+	waitingMap   map[uint32][]Validator
+	selectors    map[uint32]RandomSelector
+	leavingMap   map[uint32][]Validator
+	newList      []Validator
+	mutNodesMaps sync.RWMutex
 }
 
 type indexHashedNodesCoordinator struct {
@@ -85,6 +84,7 @@ type indexHashedNodesCoordinator struct {
 	loadingFromDisk               atomic.Value
 	shuffledOutHandler            ShuffledOutHandler
 	startEpoch                    uint32
+	publicKeyToValidatorMap       map[string]*validatorWithShardID
 }
 
 // NewIndexHashedNodesCoordinator creates a new index hashed group selector
@@ -124,6 +124,7 @@ func NewIndexHashedNodesCoordinator(arguments ArgNodesCoordinator) (*indexHashed
 		shardIDAsObserver:             arguments.ShardIDAsObserver,
 		shuffledOutHandler:            arguments.ShuffledOutHandler,
 		startEpoch:                    arguments.StartEpoch,
+		publicKeyToValidatorMap:       make(map[string]*validatorWithShardID),
 	}
 
 	ihgs.loadingFromDisk.Store(false)
@@ -134,6 +135,7 @@ func NewIndexHashedNodesCoordinator(arguments ArgNodesCoordinator) (*indexHashed
 		return nil, err
 	}
 
+	ihgs.fillPublicKeyToValidatorMap()
 	err = ihgs.saveState(ihgs.savedStateKey)
 	if err != nil {
 		log.Error("saving initial nodes coordinator config failed",
@@ -240,7 +242,6 @@ func (ihgs *indexHashedNodesCoordinator) setNodesPerShards(
 	nodesConfig.eligibleMap = eligible
 	nodesConfig.waitingMap = waiting
 	nodesConfig.leavingMap = leaving
-	nodesConfig.publicKeyToValidatorMap = ihgs.createPublicKeyToValidatorMap(eligible, waiting)
 	nodesConfig.shardID = ihgs.computeShardForSelfPublicKey(nodesConfig)
 	nodesConfig.selectors, err = ihgs.createSelectors(nodesConfig)
 	if err != nil {
@@ -338,25 +339,13 @@ func (ihgs *indexHashedNodesCoordinator) searchConsensusForKey(key []byte) []Val
 }
 
 // GetValidatorWithPublicKey gets the validator with the given public key
-func (ihgs *indexHashedNodesCoordinator) GetValidatorWithPublicKey(
-	publicKey []byte,
-	epoch uint32,
-) (Validator, uint32, error) {
+func (ihgs *indexHashedNodesCoordinator) GetValidatorWithPublicKey(publicKey []byte) (Validator, uint32, error) {
 	if len(publicKey) == 0 {
 		return nil, 0, ErrNilPubKey
 	}
 	ihgs.mutNodesConfig.RLock()
-	nodesConfig, ok := ihgs.nodesConfig[epoch]
+	v, ok := ihgs.publicKeyToValidatorMap[string(publicKey)]
 	ihgs.mutNodesConfig.RUnlock()
-
-	if !ok {
-		return nil, 0, fmt.Errorf("%w epoch=%v", ErrEpochNodesConfigDoesNotExist, epoch)
-	}
-
-	nodesConfig.mutNodesMaps.RLock()
-	defer nodesConfig.mutNodesMaps.RUnlock()
-
-	v, ok := nodesConfig.publicKeyToValidatorMap[string(publicKey)]
 	if ok {
 		return v.validator, v.shardID, nil
 	}
@@ -585,6 +574,7 @@ func (ihgs *indexHashedNodesCoordinator) EpochStartPrepare(metaHdr data.HeaderHa
 		log.Error("set nodes per shard failed", "error", err.Error())
 	}
 
+	ihgs.fillPublicKeyToValidatorMap()
 	err = ihgs.saveState(randomness)
 	if err != nil {
 		log.Error("saving nodes coordinator config failed", "error", err.Error())
@@ -604,6 +594,35 @@ func (ihgs *indexHashedNodesCoordinator) EpochStartPrepare(metaHdr data.HeaderHa
 	ihgs.consensusGroupCacher.Clear()
 }
 
+func (ihgs *indexHashedNodesCoordinator) fillPublicKeyToValidatorMap() {
+	ihgs.mutNodesConfig.Lock()
+	defer ihgs.mutNodesConfig.Unlock()
+
+	index := 0
+	epochList := make([]uint32, len(ihgs.nodesConfig))
+	mapAllValidators := make(map[uint32]map[string]*validatorWithShardID)
+	for epoch, epochConfig := range ihgs.nodesConfig {
+		epochConfig.mutNodesMaps.RLock()
+		mapAllValidators[epoch] = ihgs.createPublicKeyToValidatorMap(epochConfig.eligibleMap, epochConfig.waitingMap)
+		epochConfig.mutNodesMaps.RUnlock()
+
+		epochList[index] = epoch
+		index++
+	}
+
+	sort.Slice(epochList, func(i, j int) bool {
+		return epochList[i] < epochList[j]
+	})
+
+	ihgs.publicKeyToValidatorMap = make(map[string]*validatorWithShardID)
+	for _, epoch := range epochList {
+		validatorsForEpoch := mapAllValidators[epoch]
+		for pubKey, vInfo := range validatorsForEpoch {
+			ihgs.publicKeyToValidatorMap[pubKey] = vInfo
+		}
+	}
+}
+
 func (ihgs *indexHashedNodesCoordinator) createSortedListFromMap(validatorsMap map[uint32][]Validator) []Validator {
 	sortedList := make([]Validator, 0)
 	for _, validators := range validatorsMap {

sharding/indexHashedNodesCoordinatorWithRater_test.go

@@ -277,7 +277,7 @@ func TestIndexHashedGroupSelectorWithRater_GetValidatorWithPublicKeyShouldReturn
 	nc, _ := NewIndexHashedNodesCoordinator(arguments)
 	ihgs, _ := NewIndexHashedNodesCoordinatorWithRater(nc, &mock.RaterMock{})
 
-	_, _, err := ihgs.GetValidatorWithPublicKey(nil, 0)
+	_, _, err := ihgs.GetValidatorWithPublicKey(nil)
 	assert.Equal(t, ErrNilPubKey, err)
 }
 
@@ -314,7 +314,7 @@ func TestIndexHashedGroupSelectorWithRater_GetValidatorWithPublicKeyShouldReturn
 	nc, _ := NewIndexHashedNodesCoordinator(arguments)
 	ihgs, _ := NewIndexHashedNodesCoordinatorWithRater(nc, &mock.RaterMock{})
 
-	_, _, err := ihgs.GetValidatorWithPublicKey([]byte("pk1"), 0)
+	_, _, err := ihgs.GetValidatorWithPublicKey([]byte("pk1"))
 	assert.Equal(t, ErrValidatorNotFound, err)
 }
 
@@ -365,15 +365,15 @@ func TestIndexHashedGroupSelectorWithRater_GetValidatorWithPublicKeyShouldWork(t
 	nc, _ := NewIndexHashedNodesCoordinator(arguments)
 	ihgs, _ := NewIndexHashedNodesCoordinatorWithRater(nc, &mock.RaterMock{})
 
-	_, shardId, err := ihgs.GetValidatorWithPublicKey([]byte("pk0_meta"), 0)
+	_, shardId, err := ihgs.GetValidatorWithPublicKey([]byte("pk0_meta"))
 	assert.Nil(t, err)
 	assert.Equal(t, core.MetachainShardId, shardId)
 
-	_, shardId, err = ihgs.GetValidatorWithPublicKey([]byte("pk1_shard0"), 0)
+	_, shardId, err = ihgs.GetValidatorWithPublicKey([]byte("pk1_shard0"))
 	assert.Nil(t, err)
 	assert.Equal(t, uint32(0), shardId)
 
-	_, shardId, err = ihgs.GetValidatorWithPublicKey([]byte("pk2_shard1"), 0)
+	_, shardId, err = ihgs.GetValidatorWithPublicKey([]byte("pk2_shard1"))
 	assert.Nil(t, err)
 	assert.Equal(t, uint32(1), shardId)
 }

sharding/indexHashedNodesCoordinator_test.go

@@ -691,7 +691,7 @@ func TestIndexHashedNodesCoordinator_GetValidatorWithPublicKeyShouldReturnErrNil
 	arguments := createArguments()
 	ihgs, _ := NewIndexHashedNodesCoordinator(arguments)
 
-	_, _, err := ihgs.GetValidatorWithPublicKey(nil, 0)
+	_, _, err := ihgs.GetValidatorWithPublicKey(nil)
 	require.Equal(t, ErrNilPubKey, err)
 }
 
@@ -701,7 +701,7 @@ func TestIndexHashedNodesCoordinator_GetValidatorWithPublicKeyShouldReturnErrVal
 	arguments := createArguments()
 	ihgs, _ := NewIndexHashedNodesCoordinator(arguments)
 
-	_, _, err := ihgs.GetValidatorWithPublicKey([]byte("pk1"), 0)
+	_, _, err := ihgs.GetValidatorWithPublicKey([]byte("pk1"))
 	require.Equal(t, ErrValidatorNotFound, err)
 }
 
@@ -749,32 +749,22 @@ func TestIndexHashedNodesCoordinator_GetValidatorWithPublicKeyShouldWork(t *test
 	}
 	ihgs, _ := NewIndexHashedNodesCoordinator(arguments)
 
-	v, shardId, err := ihgs.GetValidatorWithPublicKey([]byte("pk0_meta"), 0)
+	v, shardId, err := ihgs.GetValidatorWithPublicKey([]byte("pk0_meta"))
 	require.Nil(t, err)
 	require.Equal(t, core.MetachainShardId, shardId)
 	require.Equal(t, []byte("pk0_meta"), v.PubKey())
 
-	v, shardId, err = ihgs.GetValidatorWithPublicKey([]byte("pk1_shard0"), 0)
+	v, shardId, err = ihgs.GetValidatorWithPublicKey([]byte("pk1_shard0"))
 	require.Nil(t, err)
 	require.Equal(t, uint32(0), shardId)
 	require.Equal(t, []byte("pk1_shard0"), v.PubKey())
 
-	v, shardId, err = ihgs.GetValidatorWithPublicKey([]byte("pk2_shard1"), 0)
+	v, shardId, err = ihgs.GetValidatorWithPublicKey([]byte("pk2_shard1"))
 	require.Nil(t, err)
 	require.Equal(t, uint32(1), shardId)
 	require.Equal(t, []byte("pk2_shard1"), v.PubKey())
 }
 
-func TestNewIndexHashedNodesCoordinator_GetValidatorWithPublicKeyNotExistingEpoch(t *testing.T) {
-	t.Parallel()
-
-	arguments := createArguments()
-	ihgs, _ := NewIndexHashedNodesCoordinator(arguments)
-
-	_, _, err := ihgs.GetValidatorWithPublicKey(arguments.EligibleNodes[0][0].PubKey(), 1)
-	require.True(t, errors.Is(err, ErrEpochNodesConfigDoesNotExist))
-}
-
 func TestIndexHashedGroupSelector_GetAllEligibleValidatorsPublicKeys(t *testing.T) {
 	t.Parallel()
 

sharding/interface.go

@@ -30,7 +30,7 @@ type NodesCoordinator interface {
 	NodesCoordinatorHelper
 	PublicKeysSelector
 	ComputeConsensusGroup(randomness []byte, round uint64, shardId uint32, epoch uint32) (validatorsGroup []Validator, err error)
-	GetValidatorWithPublicKey(publicKey []byte, epoch uint32) (validator Validator, shardId uint32, err error)
+	GetValidatorWithPublicKey(publicKey []byte) (validator Validator, shardId uint32, err error)
 	LoadState(key []byte) error
 	GetSavedStateKey() []byte
 	ShardIdForEpoch(epoch uint32) (uint32, error)