Equivalent messages filter on consensus topic

cmd/node/config/enableEpochs.toml

@@ -278,6 +278,9 @@
     # ScToScLogEventEnableEpoch represents the epoch when the sc to sc log event feature is enabled
     ScToScLogEventEnableEpoch = 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/enablers/enableEpochsHandler.go

@@ -130,6 +130,7 @@ func (handler *enableEpochsHandler) EpochConfirmed(epoch uint32, _ uint64) {
 	handler.setFlagValue(epoch >= handler.enableEpochsConfig.FixDelegationChangeOwnerOnAccountEnableEpoch, handler.fixDelegationChangeOwnerOnAccountFlag, "fixDelegationChangeOwnerOnAccountFlag", epoch, handler.enableEpochsConfig.FixDelegationChangeOwnerOnAccountEnableEpoch)
 	handler.setFlagValue(epoch >= handler.enableEpochsConfig.SCProcessorV2EnableEpoch, handler.scProcessorV2Flag, "scProcessorV2Flag", epoch, handler.enableEpochsConfig.SCProcessorV2EnableEpoch)
 	handler.setFlagValue(epoch >= handler.enableEpochsConfig.DynamicGasCostForDataTrieStorageLoadEnableEpoch, handler.dynamicGasCostForDataTrieStorageLoadFlag, "dynamicGasCostForDataTrieStorageLoadFlag", epoch, handler.enableEpochsConfig.DynamicGasCostForDataTrieStorageLoadEnableEpoch)
+	handler.setFlagValue(epoch >= handler.enableEpochsConfig.EquivalentMessagesEnableEpoch, handler.equivalentMessagesFlag, "equivalentMessagesFlag", epoch, handler.enableEpochsConfig.EquivalentMessagesEnableEpoch)
 }
 
 func (handler *enableEpochsHandler) setFlagValue(value bool, flag *atomic.Flag, flagName string, epoch uint32, flagEpoch uint32) {

common/enablers/enableEpochsHandler_test.go

@@ -104,6 +104,7 @@ func createEnableEpochsConfig() config.EnableEpochs {
 		FixDelegationChangeOwnerOnAccountEnableEpoch:      87,
 		DeterministicSortOnValidatorsInfoEnableEpoch:      79,
 		ScToScLogEventEnableEpoch:                         88,
+		EquivalentMessagesEnableEpoch:                     89,
 	}
 }
 
@@ -247,6 +248,7 @@ func TestNewEnableEpochsHandler_EpochConfirmed(t *testing.T) {
 		assert.True(t, handler.IsTransferToMetaFlagEnabled())
 		assert.True(t, handler.IsESDTNFTImprovementV1FlagEnabled())
 		assert.True(t, handler.FixDelegationChangeOwnerOnAccountEnabled())
+		assert.True(t, handler.IsEquivalentMessagesFlagEnabled())
 	})
 	t.Run("flags with == condition should not be set, the ones with >= should be set", func(t *testing.T) {
 		t.Parallel()
@@ -366,6 +368,7 @@ func TestNewEnableEpochsHandler_EpochConfirmed(t *testing.T) {
 		assert.True(t, handler.IsTransferToMetaFlagEnabled())
 		assert.True(t, handler.IsESDTNFTImprovementV1FlagEnabled())
 		assert.True(t, handler.FixDelegationChangeOwnerOnAccountEnabled())
+		assert.True(t, handler.IsEquivalentMessagesFlagEnabled())
 	})
 	t.Run("flags with < should be set", func(t *testing.T) {
 		t.Parallel()
@@ -480,6 +483,7 @@ func TestNewEnableEpochsHandler_EpochConfirmed(t *testing.T) {
 		assert.False(t, handler.IsTransferToMetaFlagEnabled())
 		assert.False(t, handler.IsESDTNFTImprovementV1FlagEnabled())
 		assert.False(t, handler.FixDelegationChangeOwnerOnAccountEnabled())
+		assert.False(t, handler.IsEquivalentMessagesFlagEnabled())
 	})
 }
 

common/enablers/epochFlags.go

@@ -102,6 +102,7 @@ type epochFlagsHolder struct {
 	autoBalanceDataTriesFlag                    *atomic.Flag
 	fixDelegationChangeOwnerOnAccountFlag       *atomic.Flag
 	dynamicGasCostForDataTrieStorageLoadFlag    *atomic.Flag
+	equivalentMessagesFlag                      *atomic.Flag
 }
 
 func newEpochFlagsHolder() *epochFlagsHolder {
@@ -203,6 +204,7 @@ func newEpochFlagsHolder() *epochFlagsHolder {
 		autoBalanceDataTriesFlag:                    &atomic.Flag{},
 		fixDelegationChangeOwnerOnAccountFlag:       &atomic.Flag{},
 		dynamicGasCostForDataTrieStorageLoadFlag:    &atomic.Flag{},
+		equivalentMessagesFlag:                      &atomic.Flag{},
 	}
 }
 
@@ -694,7 +696,7 @@ func (holder *epochFlagsHolder) IsSetGuardianEnabled() bool {
 	return holder.setGuardianFlag.IsSet()
 }
 
-// IsScToScLogEventFlagEnabled returns true if scToScLogEventFlag is enabled
+// IsScToScEventLogEnabled returns true if scToScLogEventFlag is enabled
 func (holder *epochFlagsHolder) IsScToScEventLogEnabled() bool {
 	return holder.scToScLogEventFlag.IsSet()
 }
@@ -743,3 +745,8 @@ func (holder *epochFlagsHolder) FixDelegationChangeOwnerOnAccountEnabled() bool
 func (holder *epochFlagsHolder) IsDynamicGasCostForDataTrieStorageLoadEnabled() bool {
 	return holder.dynamicGasCostForDataTrieStorageLoadFlag.IsSet()
 }
+
+// IsEquivalentMessagesFlagEnabled returns true if equivalentMessagesFlag is enabled
+func (holder *epochFlagsHolder) IsEquivalentMessagesFlagEnabled() bool {
+	return holder.equivalentMessagesFlag.IsSet()
+}

common/interface.go

@@ -396,6 +396,7 @@ type EnableEpochsHandler interface {
 	IsAutoBalanceDataTriesEnabled() bool
 	IsDynamicGasCostForDataTrieStorageLoadEnabled() bool
 	FixDelegationChangeOwnerOnAccountEnabled() bool
+	IsEquivalentMessagesFlagEnabled() bool
 
 	IsInterfaceNil() bool
 }

config/epochConfig.go

@@ -105,6 +105,7 @@ type EnableEpochs struct {
 	ConsistentTokensValuesLengthCheckEnableEpoch      uint32
 	FixDelegationChangeOwnerOnAccountEnableEpoch      uint32
 	DynamicGasCostForDataTrieStorageLoadEnableEpoch   uint32
+	EquivalentMessagesEnableEpoch                     uint32
 	BLSMultiSignerEnableEpoch                         []MultiSignerConfig
 }
 

config/tomlConfig_test.go

@@ -835,6 +835,9 @@ func TestEnableEpochConfig(t *testing.T) {
 	# ScToScLogEventEnableEpoch represents the epoch when the sc to sc log event feature is enabled
 	ScToScLogEventEnableEpoch = 88
 
+	# EquivalentMessagesEnableEpoch represents the epoch when the equivalent messages are enabled
+	EquivalentMessagesEnableEpoch = 89
+
     # MaxNodesChangeEnableEpoch holds configuration for changing the maximum number of nodes and the enabling epoch
     MaxNodesChangeEnableEpoch = [
         { EpochEnable = 44, MaxNumNodes = 2169, NodesToShufflePerShard = 80 },
@@ -911,12 +914,12 @@ func TestEnableEpochConfig(t *testing.T) {
 			StorageAPICostOptimizationEnableEpoch:             54,
 			TransformToMultiShardCreateEnableEpoch:            55,
 			ESDTRegisterAndSetAllRolesEnableEpoch:             56,
-			ScheduledMiniBlocksEnableEpoch:                 57,
-			CorrectJailedNotUnstakedEmptyQueueEpoch:         58,
-			DoNotReturnOldBlockInBlockchainHookEnableEpoch:            59,
-			AddFailedRelayedTxToInvalidMBsDisableEpoch:       60,
-			SCRSizeInvariantOnBuiltInResultEnableEpoch:              61,
-			CheckCorrectTokenIDForTransferRoleEnableEpoch:                   62,
+			ScheduledMiniBlocksEnableEpoch:                    57,
+			CorrectJailedNotUnstakedEmptyQueueEpoch:           58,
+			DoNotReturnOldBlockInBlockchainHookEnableEpoch:    59,
+			AddFailedRelayedTxToInvalidMBsDisableEpoch:        60,
+			SCRSizeInvariantOnBuiltInResultEnableEpoch:        61,
+			CheckCorrectTokenIDForTransferRoleEnableEpoch:     62,
 			DisableExecByCallerEnableEpoch:                    63,
 			RefactorContextEnableEpoch:                        64,
 			FailExecutionOnEveryAPIErrorEnableEpoch:           65,
@@ -926,7 +929,8 @@ func TestEnableEpochConfig(t *testing.T) {
 			ESDTMetadataContinuousCleanupEnableEpoch:          69,
 			MiniBlockPartialExecutionEnableEpoch:              70,
 			FixAsyncCallBackArgsListEnableEpoch:               71,
-			FixOldTokenLiquidityEnableEpoch:                   72,RuntimeMemStoreLimitEnableEpoch:                 73,
+			FixOldTokenLiquidityEnableEpoch:                   72,
+			RuntimeMemStoreLimitEnableEpoch:                   73,
 			SetSenderInEeiOutputTransferEnableEpoch:           74,
 			RefactorPeersMiniBlocksEnableEpoch:                75,
 			MaxBlockchainHookCountersEnableEpoch:              76,
@@ -942,6 +946,7 @@ func TestEnableEpochConfig(t *testing.T) {
 			ConsistentTokensValuesLengthCheckEnableEpoch:      86,
 			FixDelegationChangeOwnerOnAccountEnableEpoch:      87,
 			ScToScLogEventEnableEpoch:                         88,
+			EquivalentMessagesEnableEpoch:                     89,
 			MaxNodesChangeEnableEpoch: []MaxNodesChangeConfig{
 				{
 					EpochEnable:            44,
@@ -954,7 +959,7 @@ func TestEnableEpochConfig(t *testing.T) {
 					NodesToShufflePerShard: 80,
 				},
 			},
-			DeterministicSortOnValidatorsInfoEnableEpoch: 66,
+			DeterministicSortOnValidatorsInfoEnableEpoch:    66,
 			DynamicGasCostForDataTrieStorageLoadEnableEpoch: 64,
 			BLSMultiSignerEnableEpoch: []MultiSignerConfig{
 				{

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

@@ -145,6 +145,7 @@ func (fct *factory) generateStartRoundSubround() error {
 		processingThresholdPercent,
 		fct.worker.ExecuteStoredMessages,
 		fct.worker.ResetConsensusMessages,
+		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
 }
@@ -35,6 +36,7 @@ func NewSubroundStartRound(
 	processingThresholdPercentage int,
 	executeStoredMessages func(),
 	resetConsensusMessages func(),
+	removeAllEquivalentMessages func(),
 ) (*subroundStartRound, error) {
 	err := checkNewSubroundStartRoundParams(
 		baseSubround,
@@ -48,6 +50,7 @@ func NewSubroundStartRound(
 		processingThresholdPercentage: processingThresholdPercentage,
 		executeStoredMessages:         executeStoredMessages,
 		resetConsensusMessages:        resetConsensusMessages,
+		removeAllEquivalentMessages:   removeAllEquivalentMessages,
 		outportHandler:                disabled.NewDisabledOutport(),
 		outportMutex:                  sync.RWMutex{},
 	}
@@ -95,6 +98,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

@@ -23,6 +23,7 @@ func defaultSubroundStartRoundFromSubround(sr *spos.Subround) (bls.SubroundStart
 		bls.ProcessingThresholdPercent,
 		executeStoredMessages,
 		resetConsensusMessages,
+		removeAllEquivalentMessages,
 	)
 
 	return startRound, err
@@ -35,6 +36,7 @@ func defaultWithoutErrorSubroundStartRoundFromSubround(sr *spos.Subround) bls.Su
 		bls.ProcessingThresholdPercent,
 		executeStoredMessages,
 		resetConsensusMessages,
+		removeAllEquivalentMessages,
 	)
 
 	return startRound
@@ -73,6 +75,7 @@ func initSubroundStartRoundWithContainer(container spos.ConsensusCoreHandler) bl
 		bls.ProcessingThresholdPercent,
 		executeStoredMessages,
 		resetConsensusMessages,
+		removeAllEquivalentMessages,
 	)
 
 	return srStartRound
@@ -92,6 +95,7 @@ func TestSubroundStartRound_NewSubroundStartRoundNilSubroundShouldFail(t *testin
 		bls.ProcessingThresholdPercent,
 		executeStoredMessages,
 		resetConsensusMessages,
+		removeAllEquivalentMessages,
 	)
 
 	assert.Nil(t, srStartRound)
@@ -464,6 +468,7 @@ func TestSubroundStartRound_InitCurrentRoundShouldMetrics(t *testing.T) {
 			bls.ProcessingThresholdPercent,
 			displayStatistics,
 			executeStoredMessages,
+			removeAllEquivalentMessages,
 		)
 		srStartRound.Check()
 		assert.True(t, wasCalled)
@@ -506,6 +511,7 @@ func TestSubroundStartRound_InitCurrentRoundShouldMetrics(t *testing.T) {
 			bls.ProcessingThresholdPercent,
 			displayStatistics,
 			executeStoredMessages,
+			removeAllEquivalentMessages,
 		)
 		srStartRound.Check()
 		assert.True(t, wasCalled)
@@ -568,6 +574,7 @@ func TestSubroundStartRound_InitCurrentRoundShouldMetrics(t *testing.T) {
 			bls.ProcessingThresholdPercent,
 			displayStatistics,
 			executeStoredMessages,
+			removeAllEquivalentMessages,
 		)
 		srStartRound.Check()
 		assert.True(t, wasMetricConsensusStateCalled)
@@ -634,6 +641,7 @@ func TestSubroundStartRound_InitCurrentRoundShouldMetrics(t *testing.T) {
 			bls.ProcessingThresholdPercent,
 			displayStatistics,
 			executeStoredMessages,
+			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
+}