Merge pull request #5666 from multiversx/equivalent_messages_filter

Equivalent messages filter on consensus topic

Author: sstanculeanu

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

@@ -40,3 +40,9 @@ func NewConsensusMessage(
 		InvalidSigners:     invalidSigners,
 	}
 }
+
+// EquivalentMessageInfo holds information about an equivalent message
+type EquivalentMessageInfo struct {
+	NumMessages uint64
+	Validated   bool
+}

consensus/mock/equivalentMessagesDebuggerStub.go

@@ -0,0 +1,22 @@
+package mock
+
+import (
+	"github.com/multiversx/mx-chain-go/consensus"
+)
+
+// EquivalentMessagesDebuggerStub -
+type EquivalentMessagesDebuggerStub struct {
+	DisplayEquivalentMessagesStatisticsCalled func(getDataHandler func() map[string]*consensus.EquivalentMessageInfo)
+}
+
+// DisplayEquivalentMessagesStatistics -
+func (stub *EquivalentMessagesDebuggerStub) DisplayEquivalentMessagesStatistics(getDataHandler func() map[string]*consensus.EquivalentMessageInfo) {
+	if stub.DisplayEquivalentMessagesStatisticsCalled != nil {
+		stub.DisplayEquivalentMessagesStatisticsCalled(getDataHandler)
+	}
+}
+
+// IsInterfaceNil -
+func (stub *EquivalentMessagesDebuggerStub) IsInterfaceNil() bool {
+	return stub == nil
+}

consensus/spos/debug/equivalentMessagesDebugger.go

@@ -0,0 +1,72 @@
+package debug
+
+import (
+	"fmt"
+
+	"github.com/multiversx/mx-chain-core-go/display"
+	"github.com/multiversx/mx-chain-go/consensus"
+	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]*consensus.EquivalentMessageInfo) {
+	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]*consensus.EquivalentMessageInfo) string {
+	header := []string{
+		"Block header hash",
+		"Equivalent messages received",
+	}
+
+	lines := make([]*display.LineData, 0, len(data))
+	idx := 0
+	for hash, info := range data {
+		horizontalLineAfter := idx == len(data)
+		line := []string{
+			hash,
+			fmt.Sprintf("%d", info.NumMessages),
+		}
+		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
+}

consensus/spos/debug/equivalentMessagesDebugger_test.go

@@ -0,0 +1,80 @@
+package debug
+
+import (
+	"testing"
+
+	"github.com/multiversx/mx-chain-go/consensus"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewEquivalentMessagesDebugger_IsInterfaceNil(t *testing.T) {
+	t.Parallel()
+
+	var debugger *equivalentMessagesDebugger
+	require.True(t, debugger.IsInterfaceNil())
+
+	debugger = NewEquivalentMessagesDebugger()
+	require.False(t, debugger.IsInterfaceNil())
+}
+
+func TestEquivalentMessagesDebugger_DisplayEquivalentMessagesStatistics(t *testing.T) {
+	t.Parallel()
+
+	t.Run("log level not trace should early exit", func(t *testing.T) {
+		t.Parallel()
+
+		defer func() {
+			r := recover()
+			if r != nil {
+				require.Fail(t, "should have not panicked")
+			}
+		}()
+
+		debugger := NewEquivalentMessagesDebugger()
+		debugger.DisplayEquivalentMessagesStatistics(func() map[string]*consensus.EquivalentMessageInfo {
+			return make(map[string]*consensus.EquivalentMessageInfo)
+		})
+	})
+	t.Run("nil get data handler should early exit", func(t *testing.T) {
+		t.Parallel()
+
+		defer func() {
+			r := recover()
+			if r != nil {
+				require.Fail(t, "should have not panicked")
+			}
+		}()
+
+		debugger := NewEquivalentMessagesDebugger()
+		debugger.shouldProcessDataFunc = func() bool {
+			return true
+		}
+
+		debugger.DisplayEquivalentMessagesStatistics(nil)
+	})
+	t.Run("should work", func(t *testing.T) {
+		t.Parallel()
+
+		defer func() {
+			r := recover()
+			if r != nil {
+				require.Fail(t, "should have not panicked")
+			}
+		}()
+
+		debugger := NewEquivalentMessagesDebugger()
+		debugger.shouldProcessDataFunc = func() bool {
+			return true
+		}
+
+		debugger.DisplayEquivalentMessagesStatistics(func() map[string]*consensus.EquivalentMessageInfo {
+			return map[string]*consensus.EquivalentMessageInfo{
+				"hash1": {NumMessages: 1, Validated: true},
+				"hash2": {NumMessages: 2, Validated: true},
+				"hash3": {NumMessages: 3, Validated: true},
+				"hash4": {NumMessages: 4, Validated: true},
+			}
+		})
+
+	})
+}