Invalid signers cache

consensus/spos/bls/proxy/subroundsHandler_test.go

@@ -81,6 +81,7 @@ func getDefaultArgumentsSubroundHandler() (*SubroundsHandlerArgs, *spos.Consensu
 	consensusCore.SetEnableEpochsHandler(epochsEnable)
 	consensusCore.SetEquivalentProofsPool(&dataRetriever.ProofsPoolMock{})
 	consensusCore.SetEpochNotifier(epochNotifier)
+	consensusCore.SetInvalidSignersCache(&consensus.InvalidSignersCacheMock{})
 	handlerArgs.ConsensusCoreHandler = consensusCore
 
 	return handlerArgs, consensusCore

consensus/spos/bls/v2/subroundEndRound.go

@@ -150,6 +150,12 @@ func (sr *subroundEndRound) receivedInvalidSignersInfo(_ context.Context, cnsDta
 		return false
 	}
 
+	invalidSignersHash := sr.Hasher().Compute(string(cnsDta.InvalidSigners))
+	invalidSignersCache := sr.InvalidSignersCache()
+	if invalidSignersCache.HasInvalidSigners(string(invalidSignersHash)) {
+		return false
+	}
+
 	err := sr.verifyInvalidSigners(cnsDta.InvalidSigners)
 	if err != nil {
 		log.Trace("receivedInvalidSignersInfo.verifyInvalidSigners", "error", err.Error())
@@ -158,6 +164,8 @@ func (sr *subroundEndRound) receivedInvalidSignersInfo(_ context.Context, cnsDta
 
 	log.Debug("step 3: invalid signers info has been evaluated")
 
+	invalidSignersCache.AddInvalidSigners(string(invalidSignersHash))
+
 	sr.PeerHonestyHandler().ChangeScore(
 		messageSender,
 		spos.GetConsensusTopicID(sr.ShardCoordinator()),
@@ -501,24 +509,23 @@ func (sr *subroundEndRound) handleInvalidSignersOnAggSigFail() ([]byte, []byte,
 	invalidPubKeys, err := sr.verifyNodesOnAggSigFail(ctx)
 	cancel()
 	if err != nil {
-		log.Debug("doEndRoundJobByNode.verifyNodesOnAggSigFail", "error", err.Error())
+		log.Debug("handleInvalidSignersOnAggSigFail.verifyNodesOnAggSigFail", "error", err.Error())
 		return nil, nil, err
 	}
 
-	_, err = sr.getFullMessagesForInvalidSigners(invalidPubKeys)
+	invalidSigners, err := sr.getFullMessagesForInvalidSigners(invalidPubKeys)
 	if err != nil {
-		log.Debug("doEndRoundJobByNode.getFullMessagesForInvalidSigners", "error", err.Error())
+		log.Debug("handleInvalidSignersOnAggSigFail.getFullMessagesForInvalidSigners", "error", err.Error())
 		return nil, nil, err
 	}
 
-	// TODO: handle invalid signers broadcast without flooding the network
-	// if len(invalidSigners) > 0 {
-	// 	sr.createAndBroadcastInvalidSigners(invalidSigners)
-	// }
+	if len(invalidSigners) > 0 {
+		sr.createAndBroadcastInvalidSigners(invalidSigners)
+	}
 
 	bitmap, sig, err := sr.computeAggSigOnValidNodes()
 	if err != nil {
-		log.Debug("doEndRoundJobByNode.computeAggSigOnValidNodes", "error", err.Error())
+		log.Debug("handleInvalidSignersOnAggSigFail.computeAggSigOnValidNodes", "error", err.Error())
 		return nil, nil, err
 	}
 
@@ -646,6 +653,9 @@ func (sr *subroundEndRound) createAndBroadcastInvalidSigners(invalidSigners []by
 		invalidSigners,
 	)
 
+	invalidSignersHash := sr.Hasher().Compute(string(invalidSigners))
+	sr.InvalidSignersCache().AddInvalidSigners(string(invalidSignersHash))
+
 	err = sr.BroadcastMessenger().BroadcastConsensusMessage(cnsMsg)
 	if err != nil {
 		log.Debug("doEndRoundJob.BroadcastConsensusMessage", "error", err.Error())

consensus/spos/bls/v2/subroundEndRound_test.go

@@ -1502,6 +1502,27 @@ func TestSubroundEndRound_ReceivedInvalidSignersInfo(t *testing.T) {
 		res := sr.ReceivedInvalidSignersInfo(&cnsData)
 		assert.False(t, res)
 	})
+	t.Run("invalid signers cache already has this message", func(t *testing.T) {
+		t.Parallel()
+
+		container := consensusMocks.InitConsensusCore()
+		invalidSignersCache := &consensusMocks.InvalidSignersCacheMock{
+			HasInvalidSignersCalled: func(hash string) bool {
+				return true
+			},
+		}
+		container.SetInvalidSignersCache(invalidSignersCache)
+
+		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
+		cnsData := consensus.Message{
+			BlockHeaderHash: []byte("X"),
+			PubKey:          []byte("A"),
+			InvalidSigners:  []byte("invalidSignersData"),
+		}
+
+		res := sr.ReceivedInvalidSignersInfo(&cnsData)
+		assert.False(t, res)
+	})
 	t.Run("invalid signers data", func(t *testing.T) {
 		t.Parallel()
 
@@ -1528,6 +1549,13 @@ func TestSubroundEndRound_ReceivedInvalidSignersInfo(t *testing.T) {
 		t.Parallel()
 
 		container := consensusMocks.InitConsensusCore()
+		wasAddInvalidSignersCalled := false
+		invalidSignersCache := &consensusMocks.InvalidSignersCacheMock{
+			AddInvalidSignersCalled: func(hash string) {
+				wasAddInvalidSignersCalled = true
+			},
+		}
+		container.SetInvalidSignersCache(invalidSignersCache)
 
 		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
 		sr.SetHeader(&block.HeaderV2{
@@ -1541,6 +1569,7 @@ func TestSubroundEndRound_ReceivedInvalidSignersInfo(t *testing.T) {
 
 		res := sr.ReceivedInvalidSignersInfo(&cnsData)
 		assert.True(t, res)
+		require.True(t, wasAddInvalidSignersCalled)
 	})
 }
 
@@ -1704,25 +1733,34 @@ func TestSubroundEndRound_CreateAndBroadcastInvalidSigners(t *testing.T) {
 
 		expectedInvalidSigners := []byte("invalid signers")
 
-		wasCalled := false
+		wasBroadcastConsensusMessageCalled := false
 		container := consensusMocks.InitConsensusCore()
 		messenger := &consensusMocks.BroadcastMessengerMock{
 			BroadcastConsensusMessageCalled: func(message *consensus.Message) error {
 				assert.Equal(t, expectedInvalidSigners, message.InvalidSigners)
-				wasCalled = true
+				wasBroadcastConsensusMessageCalled = true
 				wg.Done()
 				return nil
 			},
 		}
 		container.SetBroadcastMessenger(messenger)
+
+		wasAddInvalidSignersCalled := false
+		invalidSignersCache := &consensusMocks.InvalidSignersCacheMock{
+			AddInvalidSignersCalled: func(hash string) {
+				wasAddInvalidSignersCalled = true
+			},
+		}
+		container.SetInvalidSignersCache(invalidSignersCache)
 		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
 		sr.SetSelfPubKey("A")
 
 		sr.CreateAndBroadcastInvalidSigners(expectedInvalidSigners)
 
 		wg.Wait()
 
-		require.True(t, wasCalled)
+		require.True(t, wasBroadcastConsensusMessageCalled)
+		require.True(t, wasAddInvalidSignersCalled)
 	})
 }
 

consensus/spos/bls/v2/subroundStartRound.go

@@ -101,6 +101,7 @@ func (sr *subroundStartRound) doStartRoundJob(_ context.Context) bool {
 	topic := spos.GetConsensusTopicID(sr.ShardCoordinator())
 	sr.GetAntiFloodHandler().ResetForTopic(topic)
 	sr.worker.ResetConsensusMessages()
+	sr.worker.ResetInvalidSignersCache()
 
 	return true
 }

consensus/spos/consensusCore.go

@@ -43,6 +43,7 @@ type ConsensusCore struct {
 	enableEpochsHandler           common.EnableEpochsHandler
 	equivalentProofsPool          consensus.EquivalentProofsPool
 	epochNotifier                 process.EpochNotifier
+	invalidSignersCache           InvalidSignersCache
 }
 
 // ConsensusCoreArgs store all arguments that are needed to create a ConsensusCore object
@@ -72,6 +73,7 @@ type ConsensusCoreArgs struct {
 	EnableEpochsHandler           common.EnableEpochsHandler
 	EquivalentProofsPool          consensus.EquivalentProofsPool
 	EpochNotifier                 process.EpochNotifier
+	InvalidSignersCache           InvalidSignersCache
 }
 
 // NewConsensusCore creates a new ConsensusCore instance
@@ -104,6 +106,7 @@ func NewConsensusCore(
 		enableEpochsHandler:           args.EnableEpochsHandler,
 		equivalentProofsPool:          args.EquivalentProofsPool,
 		epochNotifier:                 args.EpochNotifier,
+		invalidSignersCache:           args.InvalidSignersCache,
 	}
 
 	err := ValidateConsensusCore(consensusCore)
@@ -239,6 +242,11 @@ func (cc *ConsensusCore) EquivalentProofsPool() consensus.EquivalentProofsPool {
 	return cc.equivalentProofsPool
 }
 
+// InvalidSignersCache returns the invalid signers cache component
+func (cc *ConsensusCore) InvalidSignersCache() InvalidSignersCache {
+	return cc.invalidSignersCache
+}
+
 // SetBlockchain sets blockchain handler
 func (cc *ConsensusCore) SetBlockchain(blockChain data.ChainHandler) {
 	cc.blockChain = blockChain
@@ -364,6 +372,11 @@ func (cc *ConsensusCore) SetEpochNotifier(epochNotifier process.EpochNotifier) {
 	cc.epochNotifier = epochNotifier
 }
 
+// SetInvalidSignersCache sets the invalid signers cache
+func (cc *ConsensusCore) SetInvalidSignersCache(cache InvalidSignersCache) {
+	cc.invalidSignersCache = cache
+}
+
 // IsInterfaceNil returns true if there is no value under the interface
 func (cc *ConsensusCore) IsInterfaceNil() bool {
 	return cc == nil

consensus/spos/consensusCoreValidator.go

@@ -88,6 +88,9 @@ func ValidateConsensusCore(container ConsensusCoreHandler) error {
 	if check.IfNil(container.EpochStartRegistrationHandler()) {
 		return ErrNilEpochStartNotifier
 	}
+	if check.IfNil(container.InvalidSignersCache()) {
+		return ErrNilInvalidSignersCache
+	}
 
 	return nil
 }

consensus/spos/consensusCoreValidator_test.go

@@ -46,6 +46,7 @@ func initConsensusDataContainer() *spos.ConsensusCore {
 	enableEpochsHandler := &enableEpochsHandlerMock.EnableEpochsHandlerStub{}
 	proofsPool := &dataRetriever.ProofsPoolMock{}
 	epochNotifier := &epochNotifierMock.EpochNotifierStub{}
+	invalidSignersCache := &consensusMocks.InvalidSignersCacheMock{}
 
 	consensusCore, _ := spos.NewConsensusCore(&spos.ConsensusCoreArgs{
 		BlockChain:                    blockChain,
@@ -73,6 +74,7 @@ func initConsensusDataContainer() *spos.ConsensusCore {
 		EnableEpochsHandler:           enableEpochsHandler,
 		EquivalentProofsPool:          proofsPool,
 		EpochNotifier:                 epochNotifier,
+		InvalidSignersCache:           invalidSignersCache,
 	})
 
 	return consensusCore
@@ -372,6 +374,17 @@ func TestConsensusContainerValidator_ValidateNilEpochStartRegistrationHandlerSho
 	assert.Equal(t, spos.ErrNilEpochStartNotifier, err)
 }
 
+func TestConsensusContainerValidator_ValidateNilInvalidSignersCacheShouldFail(t *testing.T) {
+	t.Parallel()
+
+	container := initConsensusDataContainer()
+	container.SetInvalidSignersCache(nil)
+
+	err := spos.ValidateConsensusCore(container)
+
+	assert.Equal(t, spos.ErrNilInvalidSignersCache, err)
+}
+
 func TestConsensusContainerValidator_ShouldWork(t *testing.T) {
 	t.Parallel()
 

consensus/spos/consensusCore_test.go

@@ -41,6 +41,7 @@ func createDefaultConsensusCoreArgs() *spos.ConsensusCoreArgs {
 		EnableEpochsHandler:           consensusCoreMock.EnableEpochsHandler(),
 		EquivalentProofsPool:          consensusCoreMock.EquivalentProofsPool(),
 		EpochNotifier:                 consensusCoreMock.EpochNotifier(),
+		InvalidSignersCache:           &consensus.InvalidSignersCacheMock{},
 	}
 	return args
 }

consensus/spos/errors.go

@@ -276,3 +276,9 @@ var ErrNilEpochNotifier = errors.New("nil epoch notifier")
 
 // ErrNilEpochStartNotifier signals that nil epoch start notifier has been provided
 var ErrNilEpochStartNotifier = errors.New("nil epoch start notifier")
+
+// ErrInvalidSignersAlreadyReceived signals that an invalid signers message has been already received
+var ErrInvalidSignersAlreadyReceived = errors.New("invalid signers already received")
+
+// ErrNilInvalidSignersCache signals that nil invalid signers has been provided
+var ErrNilInvalidSignersCache = errors.New("nil invalid signers cache")

consensus/spos/interface.go

@@ -48,6 +48,7 @@ type ConsensusCoreHandler interface {
 	EnableEpochsHandler() common.EnableEpochsHandler
 	EquivalentProofsPool() consensus.EquivalentProofsPool
 	EpochNotifier() process.EpochNotifier
+	InvalidSignersCache() InvalidSignersCache
 	IsInterfaceNil() bool
 }
 
@@ -127,6 +128,8 @@ type WorkerHandler interface {
 	ResetConsensusMessages()
 	// ResetConsensusRoundState resets the consensus round state when transitioning to a different consensus version
 	ResetConsensusRoundState()
+	// ResetInvalidSignersCache resets the invalid signers cache
+	ResetInvalidSignersCache()
 	// IsInterfaceNil returns true if there is no value under the interface
 	IsInterfaceNil() bool
 }
@@ -265,3 +268,11 @@ type RoundThresholdHandler interface {
 	FallbackThreshold(subroundId int) int
 	SetFallbackThreshold(subroundId int, threshold int)
 }
+
+// InvalidSignersCache encapsulates the methods needed for a invalid signers cache
+type InvalidSignersCache interface {
+	AddInvalidSigners(hash string)
+	HasInvalidSigners(hash string) bool
+	Reset()
+	IsInterfaceNil() bool
+}

consensus/spos/invalidSignersCache.go

@@ -0,0 +1,49 @@
+package spos
+
+import "sync"
+
+type invalidSignersCache struct {
+	sync.RWMutex
+	invalidSignersMap map[string]struct{}
+}
+
+// NewInvalidSignersCache returns a new instance of invalidSignersCache
+func NewInvalidSignersCache() *invalidSignersCache {
+	return &invalidSignersCache{
+		invalidSignersMap: make(map[string]struct{}),
+	}
+}
+
+// AddInvalidSigners adds the provided hash into the internal map if it does not exist
+func (cache *invalidSignersCache) AddInvalidSigners(hash string) {
+	if len(hash) == 0 {
+		return
+	}
+
+	cache.Lock()
+	defer cache.Unlock()
+
+	cache.invalidSignersMap[hash] = struct{}{}
+}
+
+// HasInvalidSigners check whether the provided hash exists in int internal map or not
+func (cache *invalidSignersCache) HasInvalidSigners(hash string) bool {
+	cache.RLock()
+	defer cache.RUnlock()
+
+	_, has := cache.invalidSignersMap[hash]
+	return has
+}
+
+// Reset clears the internal map
+func (cache *invalidSignersCache) Reset() {
+	cache.Lock()
+	defer cache.Unlock()
+
+	cache.invalidSignersMap = make(map[string]struct{})
+}
+
+// IsInterfaceNil returns true if there is no value under the interface
+func (cache *invalidSignersCache) IsInterfaceNil() bool {
+	return cache == nil
+}