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/export_test.go

@@ -302,7 +302,7 @@ func (sr *subroundEndRound) ReceivedInvalidSignersInfo(cnsDta *consensus.Message
 }
 
 // VerifyInvalidSigners calls the unexported verifyInvalidSigners function
-func (sr *subroundEndRound) VerifyInvalidSigners(invalidSigners []byte) error {
+func (sr *subroundEndRound) VerifyInvalidSigners(invalidSigners []byte) ([]string, error) {
 	return sr.verifyInvalidSigners(invalidSigners)
 }
 
@@ -313,7 +313,7 @@ func (sr *subroundEndRound) GetMinConsensusGroupIndexOfManagedKeys() int {
 
 // CreateAndBroadcastInvalidSigners calls the unexported createAndBroadcastInvalidSigners function
 func (sr *subroundEndRound) CreateAndBroadcastInvalidSigners(invalidSigners []byte) {
-	sr.createAndBroadcastInvalidSigners(invalidSigners)
+	sr.createAndBroadcastInvalidSigners(invalidSigners, nil)
 }
 
 // GetFullMessagesForInvalidSigners calls the unexported getFullMessagesForInvalidSigners function

consensus/spos/bls/v2/subroundEndRound.go

@@ -150,14 +150,21 @@ func (sr *subroundEndRound) receivedInvalidSignersInfo(_ context.Context, cnsDta
 		return false
 	}
 
-	err := sr.verifyInvalidSigners(cnsDta.InvalidSigners)
+	invalidSignersCache := sr.InvalidSignersCache()
+	if invalidSignersCache.HasInvalidSigners(cnsDta.BlockHeaderHash, cnsDta.InvalidSigners) {
+		return false
+	}
+
+	invalidSignersPubKeys, err := sr.verifyInvalidSigners(cnsDta.InvalidSigners)
 	if err != nil {
 		log.Trace("receivedInvalidSignersInfo.verifyInvalidSigners", "error", err.Error())
 		return false
 	}
 
 	log.Debug("step 3: invalid signers info has been evaluated")
 
+	invalidSignersCache.AddInvalidSigners(cnsDta.BlockHeaderHash, cnsDta.InvalidSigners, invalidSignersPubKeys)
+
 	sr.PeerHonestyHandler().ChangeScore(
 		messageSender,
 		spos.GetConsensusTopicID(sr.ShardCoordinator()),
@@ -167,32 +174,37 @@ func (sr *subroundEndRound) receivedInvalidSignersInfo(_ context.Context, cnsDta
 	return true
 }
 
-func (sr *subroundEndRound) verifyInvalidSigners(invalidSigners []byte) error {
+func (sr *subroundEndRound) verifyInvalidSigners(invalidSigners []byte) ([]string, error) {
 	messages, err := sr.MessageSigningHandler().Deserialize(invalidSigners)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
+	pubKeys := make([]string, 0, len(messages))
 	for _, msg := range messages {
-		err = sr.verifyInvalidSigner(msg)
-		if err != nil {
-			return err
+		pubKey, errVerify := sr.verifyInvalidSigner(msg)
+		if errVerify != nil {
+			return nil, errVerify
+		}
+
+		if len(pubKey) > 0 {
+			pubKeys = append(pubKeys, pubKey)
 		}
 	}
 
-	return nil
+	return pubKeys, nil
 }
 
-func (sr *subroundEndRound) verifyInvalidSigner(msg p2p.MessageP2P) error {
+func (sr *subroundEndRound) verifyInvalidSigner(msg p2p.MessageP2P) (string, error) {
 	err := sr.MessageSigningHandler().Verify(msg)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	cnsMsg := &consensus.Message{}
 	err = sr.Marshalizer().Unmarshal(cnsMsg, msg.Data())
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	err = sr.SigningHandler().VerifySingleSignature(cnsMsg.PubKey, cnsMsg.BlockHeaderHash, cnsMsg.SignatureShare)
@@ -203,9 +215,11 @@ func (sr *subroundEndRound) verifyInvalidSigner(msg p2p.MessageP2P) error {
 			"error", err.Error(),
 		)
 		sr.applyBlacklistOnNode(msg.Peer())
+
+		return string(cnsMsg.PubKey), nil
 	}
 
-	return nil
+	return "", nil
 }
 
 func (sr *subroundEndRound) applyBlacklistOnNode(peer core.PeerID) {
@@ -501,24 +515,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, invalidPubKeys)
+	}
 
 	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
 	}
 
@@ -618,7 +631,7 @@ func (sr *subroundEndRound) getRandomManagedKeyProofSender() string {
 	return randManagedKey
 }
 
-func (sr *subroundEndRound) createAndBroadcastInvalidSigners(invalidSigners []byte) {
+func (sr *subroundEndRound) createAndBroadcastInvalidSigners(invalidSigners []byte, invalidSignersPubKeys []string) {
 	if !sr.ShouldConsiderSelfKeyInConsensus() {
 		return
 	}
@@ -646,6 +659,8 @@ func (sr *subroundEndRound) createAndBroadcastInvalidSigners(invalidSigners []by
 		invalidSigners,
 	)
 
+	sr.InvalidSignersCache().AddInvalidSigners(sr.GetData(), invalidSigners, invalidSignersPubKeys)
+
 	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(headerHash []byte, invalidSigners []byte) 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(headerHash []byte, invalidSigners []byte, invalidPublicKeys []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)
 	})
 }
 
@@ -1552,7 +1581,6 @@ func TestVerifyInvalidSigners(t *testing.T) {
 
 		container := consensusMocks.InitConsensusCore()
 
-		expectedErr := errors.New("expected err")
 		messageSigningHandler := &mock.MessageSigningHandlerStub{
 			DeserializeCalled: func(messagesBytes []byte) ([]p2p.MessageP2P, error) {
 				return nil, expectedErr
@@ -1563,7 +1591,7 @@ func TestVerifyInvalidSigners(t *testing.T) {
 
 		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
 
-		err := sr.VerifyInvalidSigners([]byte{})
+		_, err := sr.VerifyInvalidSigners([]byte{})
 		require.Equal(t, expectedErr, err)
 	})
 
@@ -1577,7 +1605,6 @@ func TestVerifyInvalidSigners(t *testing.T) {
 		}}
 		invalidSignersBytes, _ := container.Marshalizer().Marshal(invalidSigners)
 
-		expectedErr := errors.New("expected err")
 		messageSigningHandler := &mock.MessageSigningHandlerStub{
 			DeserializeCalled: func(messagesBytes []byte) ([]p2p.MessageP2P, error) {
 				require.Equal(t, invalidSignersBytes, messagesBytes)
@@ -1592,7 +1619,7 @@ func TestVerifyInvalidSigners(t *testing.T) {
 
 		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
 
-		err := sr.VerifyInvalidSigners(invalidSignersBytes)
+		_, err := sr.VerifyInvalidSigners(invalidSignersBytes)
 		require.Equal(t, expectedErr, err)
 	})
 
@@ -1634,7 +1661,7 @@ func TestVerifyInvalidSigners(t *testing.T) {
 
 		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
 
-		err := sr.VerifyInvalidSigners(invalidSignersBytes)
+		_, err := sr.VerifyInvalidSigners(invalidSignersBytes)
 		require.Nil(t, err)
 		require.True(t, wasCalled)
 	})
@@ -1662,7 +1689,7 @@ func TestVerifyInvalidSigners(t *testing.T) {
 
 		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
 
-		err := sr.VerifyInvalidSigners(invalidSignersBytes)
+		_, err := sr.VerifyInvalidSigners(invalidSignersBytes)
 		require.Nil(t, err)
 	})
 }
@@ -1704,25 +1731,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(headerHash []byte, invalidSigners []byte, invalidPublicKeys []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()