Equivalent messages filter on consensus topic

cmd/node/config/enableEpochs.toml

@@ -287,6 +287,9 @@
     # FixGasRemainingForSaveKeyValueBuiltinFunctionEnableEpoch represents the epoch when the fix for the remaining gas in the SaveKeyValue builtin function is enabled
     FixGasRemainingForSaveKeyValueBuiltinFunctionEnableEpoch = 3
 
+    # EquivalentMessagesEnableEpoch represents the epoch when the equivalent messages are enabled
+    EquivalentMessagesEnableEpoch = 3
+
     # BLSMultiSignerEnableEpoch represents the activation epoch for different types of BLS multi-signers
     BLSMultiSignerEnableEpoch = [
         { EnableEpoch = 0, Type = "no-KOSK" },

common/constants.go

@@ -1001,5 +1001,6 @@ const (
 	NFTStopCreateFlag                                  core.EnableEpochFlag = "NFTStopCreateFlag"
 	FixGasRemainingForSaveKeyValueFlag                 core.EnableEpochFlag = "FixGasRemainingForSaveKeyValueFlag"
 	IsChangeOwnerAddressCrossShardThroughSCFlag        core.EnableEpochFlag = "IsChangeOwnerAddressCrossShardThroughSCFlag"
+	EquivalentMessagesFlag                             core.EnableEpochFlag = "EquivalentMessagesFlag"
 	// all new flags must be added to createAllFlagsMap method, as part of enableEpochsHandler allFlagsDefined
 )

common/enablers/enableEpochsHandler.go

@@ -695,6 +695,12 @@ func (handler *enableEpochsHandler) createAllFlagsMap() {
 			},
 			activationEpoch: handler.enableEpochsConfig.ChangeOwnerAddressCrossShardThroughSCEnableEpoch,
 		},
+		common.EquivalentMessagesFlag: {
+			isActiveInEpoch: func(epoch uint32) bool {
+				return epoch >= handler.enableEpochsConfig.EquivalentMessagesEnableEpoch
+			},
+			activationEpoch: handler.enableEpochsConfig.EquivalentMessagesEnableEpoch,
+		},
 	}
 }
 

common/enablers/enableEpochsHandler_test.go

@@ -110,6 +110,7 @@ func createEnableEpochsConfig() config.EnableEpochs {
 		NFTStopCreateEnableEpoch:                                 92,
 		FixGasRemainingForSaveKeyValueBuiltinFunctionEnableEpoch: 93,
 		ChangeOwnerAddressCrossShardThroughSCEnableEpoch:         94,
+		EquivalentMessagesEnableEpoch:                    		  95,
 	}
 }
 
@@ -297,6 +298,7 @@ func TestEnableEpochsHandler_IsFlagEnabled(t *testing.T) {
 	require.True(t, handler.IsFlagEnabled(common.NFTStopCreateFlag))
 	require.True(t, handler.IsFlagEnabled(common.FixGasRemainingForSaveKeyValueFlag))
 	require.True(t, handler.IsFlagEnabled(common.IsChangeOwnerAddressCrossShardThroughSCFlag))
+	require.True(t, handler.IsFlagEnabled(common.EquivalentMessagesFlag))
 }
 
 func TestEnableEpochsHandler_GetActivationEpoch(t *testing.T) {
@@ -407,6 +409,7 @@ func TestEnableEpochsHandler_GetActivationEpoch(t *testing.T) {
 	require.Equal(t, cfg.NFTStopCreateEnableEpoch, handler.GetActivationEpoch(common.NFTStopCreateFlag))
 	require.Equal(t, cfg.ChangeOwnerAddressCrossShardThroughSCEnableEpoch, handler.GetActivationEpoch(common.IsChangeOwnerAddressCrossShardThroughSCFlag))
 	require.Equal(t, cfg.FixGasRemainingForSaveKeyValueBuiltinFunctionEnableEpoch, handler.GetActivationEpoch(common.FixGasRemainingForSaveKeyValueFlag))
+	require.Equal(t, cfg.EquivalentMessagesEnableEpoch, handler.GetActivationEpoch(common.EquivalentMessagesFlag))
 }
 
 func TestEnableEpochsHandler_IsInterfaceNil(t *testing.T) {

config/epochConfig.go

@@ -108,6 +108,7 @@ type EnableEpochs struct {
 	NFTStopCreateEnableEpoch                                 uint32
 	ChangeOwnerAddressCrossShardThroughSCEnableEpoch         uint32
 	FixGasRemainingForSaveKeyValueBuiltinFunctionEnableEpoch uint32
+	EquivalentMessagesEnableEpoch                    		 uint32
 	BLSMultiSignerEnableEpoch                                []MultiSignerConfig
 }
 

config/tomlConfig_test.go

@@ -857,6 +857,9 @@ func TestEnableEpochConfig(t *testing.T) {
     # FixGasRemainingForSaveKeyValueBuiltinFunctionEnableEpoch represents the epoch when the fix for the remaining gas in the SaveKeyValue builtin function is enabled
     FixGasRemainingForSaveKeyValueBuiltinFunctionEnableEpoch = 91
 
+	# EquivalentMessagesEnableEpoch represents the epoch when the equivalent messages are enabled
+	EquivalentMessagesEnableEpoch = 92
+
     # MaxNodesChangeEnableEpoch holds configuration for changing the maximum number of nodes and the enabling epoch
     MaxNodesChangeEnableEpoch = [
         { EpochEnable = 44, MaxNumNodes = 2169, NodesToShufflePerShard = 80 },
@@ -968,6 +971,7 @@ func TestEnableEpochConfig(t *testing.T) {
 			NFTStopCreateEnableEpoch:                                 89,
 			ChangeOwnerAddressCrossShardThroughSCEnableEpoch:         90,
 			FixGasRemainingForSaveKeyValueBuiltinFunctionEnableEpoch: 91,
+			EquivalentMessagesEnableEpoch:                            92,
 			MaxNodesChangeEnableEpoch: []MaxNodesChangeConfig{
 				{
 					EpochEnable:            44,

consensus/mock/equivalentMessagesDebuggerStub.go

@@ -0,0 +1,18 @@
+package mock
+
+// EquivalentMessagesDebuggerStub -
+type EquivalentMessagesDebuggerStub struct {
+	DisplayEquivalentMessagesStatisticsCalled func(getDataHandler func() map[string]uint64)
+}
+
+// DisplayEquivalentMessagesStatistics -
+func (stub *EquivalentMessagesDebuggerStub) DisplayEquivalentMessagesStatistics(getDataHandler func() map[string]uint64) {
+	if stub.DisplayEquivalentMessagesStatisticsCalled != nil {
+		stub.DisplayEquivalentMessagesStatisticsCalled(getDataHandler)
+	}
+}
+
+// IsInterfaceNil -
+func (stub *EquivalentMessagesDebuggerStub) IsInterfaceNil() bool {
+	return stub == nil
+}

consensus/mock/sposWorkerMock.go

@@ -28,6 +28,7 @@ type SposWorkerMock struct {
 	ReceivedHeaderCalled                   func(headerHandler data.HeaderHandler, headerHash []byte)
 	SetAppStatusHandlerCalled              func(ash core.AppStatusHandler) error
 	ResetConsensusMessagesCalled           func()
+	RemoveAllEquivalentMessagesCalled      func()
 }
 
 // AddReceivedMessageCall -
@@ -108,6 +109,13 @@ func (sposWorkerMock *SposWorkerMock) ResetConsensusMessages() {
 	}
 }
 
+// RemoveAllEquivalentMessages -
+func (sposWorkerMock *SposWorkerMock) RemoveAllEquivalentMessages() {
+	if sposWorkerMock.RemoveAllEquivalentMessagesCalled != nil {
+		sposWorkerMock.RemoveAllEquivalentMessagesCalled()
+	}
+}
+
 // IsInterfaceNil returns true if there is no value under the interface
 func (sposWorkerMock *SposWorkerMock) IsInterfaceNil() bool {
 	return sposWorkerMock == nil

consensus/spos/bls/blsSubroundsFactory.go

@@ -154,6 +154,7 @@ func (fct *factory) generateStartRoundSubround() error {
 		fct.worker.ExecuteStoredMessages,
 		fct.worker.ResetConsensusMessages,
 		fct.sentSignaturesTracker,
+		fct.worker.RemoveAllEquivalentMessages,
 	)
 	if err != nil {
 		return err

consensus/spos/bls/blsSubroundsFactory_test.go

@@ -39,6 +39,10 @@ func executeStoredMessages() {
 func resetConsensusMessages() {
 }
 
+// removeAllEquivalentMessages removes all equivalent messages
+func removeAllEquivalentMessages() {
+}
+
 func initRoundHandlerMock() *mock.RoundHandlerMock {
 	return &mock.RoundHandlerMock{
 		RoundIndex: 0,

consensus/spos/bls/subroundStartRound.go

@@ -24,6 +24,7 @@ type subroundStartRound struct {
 	processingThresholdPercentage int
 	executeStoredMessages         func()
 	resetConsensusMessages        func()
+	removeAllEquivalentMessages   func()
 
 	outportHandler       outport.OutportHandler
 	sentSignatureTracker spos.SentSignaturesTracker
@@ -37,6 +38,7 @@ func NewSubroundStartRound(
 	executeStoredMessages func(),
 	resetConsensusMessages func(),
 	sentSignatureTracker spos.SentSignaturesTracker,
+	removeAllEquivalentMessages func(),
 ) (*subroundStartRound, error) {
 	err := checkNewSubroundStartRoundParams(
 		baseSubround,
@@ -56,12 +58,16 @@ func NewSubroundStartRound(
 	if check.IfNil(sentSignatureTracker) {
 		return nil, spos.ErrNilSentSignatureTracker
 	}
+	if removeAllEquivalentMessages == nil {
+		return nil, fmt.Errorf("%w for removeAllEquivalentMessages function", spos.ErrNilFunctionHandler)
+	}
 
 	srStartRound := subroundStartRound{
 		Subround:                      baseSubround,
 		processingThresholdPercentage: processingThresholdPercentage,
 		executeStoredMessages:         executeStoredMessages,
 		resetConsensusMessages:        resetConsensusMessages,
+		removeAllEquivalentMessages:   removeAllEquivalentMessages,
 		outportHandler:                disabled.NewDisabledOutport(),
 		sentSignatureTracker:          sentSignatureTracker,
 		outportMutex:                  sync.RWMutex{},
@@ -110,6 +116,7 @@ func (sr *subroundStartRound) doStartRoundJob(_ context.Context) bool {
 	topic := spos.GetConsensusTopicID(sr.ShardCoordinator())
 	sr.GetAntiFloodHandler().ResetForTopic(topic)
 	sr.resetConsensusMessages()
+	sr.removeAllEquivalentMessages()
 	return true
 }
 

consensus/spos/bls/subroundStartRound_test.go

@@ -24,6 +24,7 @@ func defaultSubroundStartRoundFromSubround(sr *spos.Subround) (bls.SubroundStart
 		executeStoredMessages,
 		resetConsensusMessages,
 		&mock.SentSignatureTrackerStub{},
+		removeAllEquivalentMessages,
 	)
 
 	return startRound, err
@@ -37,6 +38,7 @@ func defaultWithoutErrorSubroundStartRoundFromSubround(sr *spos.Subround) bls.Su
 		executeStoredMessages,
 		resetConsensusMessages,
 		&mock.SentSignatureTrackerStub{},
+		removeAllEquivalentMessages,
 	)
 
 	return startRound
@@ -76,6 +78,7 @@ func initSubroundStartRoundWithContainer(container spos.ConsensusCoreHandler) bl
 		executeStoredMessages,
 		resetConsensusMessages,
 		&mock.SentSignatureTrackerStub{},
+		removeAllEquivalentMessages,
 	)
 
 	return srStartRound
@@ -118,6 +121,7 @@ func TestNewSubroundStartRound(t *testing.T) {
 			executeStoredMessages,
 			resetConsensusMessages,
 			&mock.SentSignatureTrackerStub{},
+			removeAllEquivalentMessages,
 		)
 
 		assert.Nil(t, srStartRound)
@@ -133,6 +137,7 @@ func TestNewSubroundStartRound(t *testing.T) {
 			executeStoredMessages,
 			resetConsensusMessages,
 			&mock.SentSignatureTrackerStub{},
+			removeAllEquivalentMessages,
 		)
 
 		assert.Nil(t, srStartRound)
@@ -149,6 +154,7 @@ func TestNewSubroundStartRound(t *testing.T) {
 			nil,
 			resetConsensusMessages,
 			&mock.SentSignatureTrackerStub{},
+			removeAllEquivalentMessages,
 		)
 
 		assert.Nil(t, srStartRound)
@@ -165,12 +171,30 @@ func TestNewSubroundStartRound(t *testing.T) {
 			executeStoredMessages,
 			nil,
 			&mock.SentSignatureTrackerStub{},
+			removeAllEquivalentMessages,
 		)
 
 		assert.Nil(t, srStartRound)
 		assert.ErrorIs(t, err, spos.ErrNilFunctionHandler)
 		assert.Contains(t, err.Error(), "resetConsensusMessages")
 	})
+	t.Run("nil removeAllEquivalentMessages function handler should error", func(t *testing.T) {
+		t.Parallel()
+
+		srStartRound, err := bls.NewSubroundStartRound(
+			sr,
+			extend,
+			bls.ProcessingThresholdPercent,
+			executeStoredMessages,
+			resetConsensusMessages,
+			&mock.SentSignatureTrackerStub{},
+			nil,
+		)
+
+		assert.Nil(t, srStartRound)
+		assert.ErrorIs(t, err, spos.ErrNilFunctionHandler)
+		assert.Contains(t, err.Error(), "removeAllEquivalentMessages")
+	})
 	t.Run("nil sent signatures tracker should error", func(t *testing.T) {
 		t.Parallel()
 
@@ -181,6 +205,7 @@ func TestNewSubroundStartRound(t *testing.T) {
 			executeStoredMessages,
 			resetConsensusMessages,
 			nil,
+			removeAllEquivalentMessages,
 		)
 
 		assert.Nil(t, srStartRound)
@@ -562,6 +587,7 @@ func TestSubroundStartRound_InitCurrentRoundShouldMetrics(t *testing.T) {
 			displayStatistics,
 			executeStoredMessages,
 			&mock.SentSignatureTrackerStub{},
+			removeAllEquivalentMessages,
 		)
 		srStartRound.Check()
 		assert.True(t, wasCalled)
@@ -605,6 +631,7 @@ func TestSubroundStartRound_InitCurrentRoundShouldMetrics(t *testing.T) {
 			displayStatistics,
 			executeStoredMessages,
 			&mock.SentSignatureTrackerStub{},
+			removeAllEquivalentMessages,
 		)
 		srStartRound.Check()
 		assert.True(t, wasCalled)
@@ -668,6 +695,7 @@ func TestSubroundStartRound_InitCurrentRoundShouldMetrics(t *testing.T) {
 			displayStatistics,
 			executeStoredMessages,
 			&mock.SentSignatureTrackerStub{},
+			removeAllEquivalentMessages,
 		)
 		srStartRound.Check()
 		assert.True(t, wasMetricConsensusStateCalled)
@@ -735,6 +763,7 @@ func TestSubroundStartRound_InitCurrentRoundShouldMetrics(t *testing.T) {
 			displayStatistics,
 			executeStoredMessages,
 			&mock.SentSignatureTrackerStub{},
+			removeAllEquivalentMessages,
 		)
 		srStartRound.Check()
 		assert.True(t, wasMetricConsensusStateCalled)

consensus/spos/debug/equivalentMessagesDebugger.go

@@ -0,0 +1,71 @@
+package debug
+
+import (
+	"fmt"
+
+	"github.com/multiversx/mx-chain-core-go/display"
+	logger "github.com/multiversx/mx-chain-logger-go"
+)
+
+var log = logger.GetOrCreate("debug/equivalentmessages")
+
+type equivalentMessagesDebugger struct {
+	shouldProcessDataFunc func() bool
+}
+
+// NewEquivalentMessagesDebugger returns a new instance of equivalentMessagesDebugger
+func NewEquivalentMessagesDebugger() *equivalentMessagesDebugger {
+	debugger := &equivalentMessagesDebugger{
+		shouldProcessDataFunc: isLogTrace,
+	}
+
+	return debugger
+}
+
+// DisplayEquivalentMessagesStatistics prints all the equivalent messages
+func (debugger *equivalentMessagesDebugger) DisplayEquivalentMessagesStatistics(getDataHandler func() map[string]uint64) {
+	if !debugger.shouldProcessDataFunc() {
+		return
+	}
+	if getDataHandler == nil {
+		return
+	}
+
+	dataMap := getDataHandler()
+	log.Trace(fmt.Sprintf("Equivalent messages statistics for current round\n%s", dataToString(dataMap)))
+}
+
+func dataToString(data map[string]uint64) string {
+	header := []string{
+		"Block header hash",
+		"Equivalent messages received",
+	}
+
+	lines := make([]*display.LineData, 0, len(data))
+	idx := 0
+	for hash, cnt := range data {
+		horizontalLineAfter := idx == len(data)
+		line := []string{
+			hash,
+			fmt.Sprintf("%d", cnt),
+		}
+		lines = append(lines, display.NewLineData(horizontalLineAfter, line))
+		idx++
+	}
+
+	table, err := display.CreateTableString(header, lines)
+	if err != nil {
+		return "error creating p2p stats table: " + err.Error()
+	}
+
+	return table
+}
+
+func isLogTrace() bool {
+	return log.GetLevel() == logger.LogTrace
+}
+
+// IsInterfaceNil returns true if there is no value under the interface
+func (debugger *equivalentMessagesDebugger) IsInterfaceNil() bool {
+	return debugger == nil
+}