Feat/equivalent messages

cmd/node/config/config.toml

@@ -951,3 +951,14 @@
     # MaxRoundsOfInactivityAccepted defines the number of rounds missed by a main or higher level backup machine before
     # the current machine will take over and propose/sign blocks. Used in both single-key and multi-key modes.
     MaxRoundsOfInactivityAccepted = 3
+
+# ConsensusGradualBroadcast defines how validators will broadcast the aggregated final info, based on their consensus index
+[ConsensusGradualBroadcast]
+    GradualIndexBroadcastDelay = [
+        # All validators will broadcast the message right away
+        { EndIndex = 0, DelayInMilliseconds = 0 },
+    ]
+
+[InterceptedDataVerifier]
+    CacheSpanInSec = 30
+    CacheExpiryInSec = 30

cmd/node/config/enableEpochs.toml

@@ -318,6 +318,12 @@
     # CryptoOpcodesV2EnableEpoch represents the epoch when BLSMultiSig, Secp256r1 and other opcodes are enabled
     CryptoOpcodesV2EnableEpoch = 4
 
+    # EquivalentMessagesEnableEpoch represents the epoch when the equivalent messages are enabled
+    EquivalentMessagesEnableEpoch = 4
+
+    # FixedOrderInConsensusEnableEpoch represents the epoch when the fixed order in consensus is enabled
+    FixedOrderInConsensusEnableEpoch = 4
+
     # BLSMultiSignerEnableEpoch represents the activation epoch for different types of BLS multi-signers
     BLSMultiSignerEnableEpoch = [
         { EnableEpoch = 0, Type = "no-KOSK" },

cmd/node/factory/interface.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/multiversx/mx-chain-core-go/core"
 	"github.com/multiversx/mx-chain-core-go/data"
+
 	"github.com/multiversx/mx-chain-go/p2p"
 )
 
@@ -14,6 +15,8 @@ type HeaderSigVerifierHandler interface {
 	VerifyLeaderSignature(header data.HeaderHandler) error
 	VerifyRandSeedAndLeaderSignature(header data.HeaderHandler) error
 	VerifySignature(header data.HeaderHandler) error
+	VerifySignatureForHash(header data.HeaderHandler, hash []byte, pubkeysBitmap []byte, signature []byte) error
+	VerifyHeaderProof(headerProof data.HeaderProofHandler) error
 	IsInterfaceNil() bool
 }
 

common/common.go

@@ -0,0 +1,21 @@
+package common
+
+import (
+	"github.com/multiversx/mx-chain-core-go/core"
+	"github.com/multiversx/mx-chain-core-go/data"
+)
+
+// IsEpochChangeBlockForFlagActivation returns true if the provided header is the first one after the specified flag's activation
+func IsEpochChangeBlockForFlagActivation(header data.HeaderHandler, enableEpochsHandler EnableEpochsHandler, flag core.EnableEpochFlag) bool {
+	isStartOfEpochBlock := header.IsStartOfEpochBlock()
+	isBlockInActivationEpoch := header.GetEpoch() == enableEpochsHandler.GetActivationEpoch(flag)
+
+	return isStartOfEpochBlock && isBlockInActivationEpoch
+}
+
+// IsFlagEnabledAfterEpochsStartBlock returns true if the flag is enabled for the header, but it is not the epoch start block
+func IsFlagEnabledAfterEpochsStartBlock(header data.HeaderHandler, enableEpochsHandler EnableEpochsHandler, flag core.EnableEpochFlag) bool {
+	isFlagEnabled := enableEpochsHandler.IsFlagEnabledInEpoch(flag, header.GetEpoch())
+	isEpochStartBlock := IsEpochChangeBlockForFlagActivation(header, enableEpochsHandler, flag)
+	return isFlagEnabled && !isEpochStartBlock
+}

common/constants.go

@@ -94,6 +94,9 @@ const ConnectionTopic = "connection"
 // ValidatorInfoTopic is the topic used for validatorInfo signaling
 const ValidatorInfoTopic = "validatorInfo"
 
+// EquivalentProofsTopic is the topic used for equivalent proofs
+const EquivalentProofsTopic = "equivalentProofs"
+
 // MetricCurrentRound is the metric for monitoring the current round of a node
 const MetricCurrentRound = "erd_current_round"
 
@@ -840,8 +843,10 @@ const (
 	ChainParametersOrder
 	// NodesCoordinatorOrder defines the order in which NodesCoordinator is notified of a start of epoch event
 	NodesCoordinatorOrder
-	// ConsensusOrder defines the order in which Consensus is notified of a start of epoch event
-	ConsensusOrder
+	// ConsensusHandlerOrder defines the order in which ConsensusHandler is notified of a start of epoch event
+	ConsensusHandlerOrder
+	// ConsensusStartRoundOrder defines the order in which Consensus StartRound subround is notified of a start of epoch event
+	ConsensusStartRoundOrder
 	// NetworkShardingOrder defines the order in which the network sharding subsystem is notified of a start of epoch event
 	NetworkShardingOrder
 	// IndexerOrder defines the order in which indexer is notified of a start of epoch event
@@ -1222,5 +1227,7 @@ const (
 	DynamicESDTFlag                                    core.EnableEpochFlag = "DynamicEsdtFlag"
 	EGLDInESDTMultiTransferFlag                        core.EnableEpochFlag = "EGLDInESDTMultiTransferFlag"
 	CryptoOpcodesV2Flag                                core.EnableEpochFlag = "CryptoOpcodesV2Flag"
+	EquivalentMessagesFlag                             core.EnableEpochFlag = "EquivalentMessagesFlag"
+	FixedOrderInConsensusFlag                          core.EnableEpochFlag = "FixedOrderInConsensusFlag"
 	// all new flags must be added to createAllFlagsMap method, as part of enableEpochsHandler allFlagsDefined
 )

common/enablers/enableEpochsHandler.go

@@ -750,6 +750,18 @@ func (handler *enableEpochsHandler) createAllFlagsMap() {
 			},
 			activationEpoch: handler.enableEpochsConfig.CryptoOpcodesV2EnableEpoch,
 		},
+		common.EquivalentMessagesFlag: {
+			isActiveInEpoch: func(epoch uint32) bool {
+				return epoch >= handler.enableEpochsConfig.EquivalentMessagesEnableEpoch
+			},
+			activationEpoch: handler.enableEpochsConfig.EquivalentMessagesEnableEpoch,
+		},
+		common.FixedOrderInConsensusFlag: {
+			isActiveInEpoch: func(epoch uint32) bool {
+				return epoch >= handler.enableEpochsConfig.FixedOrderInConsensusEnableEpoch
+			},
+			activationEpoch: handler.enableEpochsConfig.FixedOrderInConsensusEnableEpoch,
+		},
 	}
 }
 

common/enablers/enableEpochsHandler_test.go

@@ -115,10 +115,12 @@ func createEnableEpochsConfig() config.EnableEpochs {
 		StakingV4Step2EnableEpoch:                                98,
 		StakingV4Step3EnableEpoch:                                99,
 		AlwaysMergeContextsInEEIEnableEpoch:                      100,
-		CleanupAuctionOnLowWaitingListEnableEpoch: 101,
+		CleanupAuctionOnLowWaitingListEnableEpoch:                101,
 		DynamicESDTEnableEpoch:                                   102,
 		EGLDInMultiTransferEnableEpoch:                           103,
 		CryptoOpcodesV2EnableEpoch:                               104,
+		EquivalentMessagesEnableEpoch:                            105,
+		FixedOrderInConsensusEnableEpoch:                         106,
 	}
 }
 
@@ -319,6 +321,8 @@ func TestEnableEpochsHandler_IsFlagEnabled(t *testing.T) {
 	require.True(t, handler.IsFlagEnabled(common.StakingV4StartedFlag))
 	require.True(t, handler.IsFlagEnabled(common.AlwaysMergeContextsInEEIFlag))
 	require.True(t, handler.IsFlagEnabled(common.DynamicESDTFlag))
+	require.True(t, handler.IsFlagEnabled(common.EquivalentMessagesFlag))
+	require.True(t, handler.IsFlagEnabled(common.FixedOrderInConsensusFlag))
 }
 
 func TestEnableEpochsHandler_GetActivationEpoch(t *testing.T) {
@@ -438,6 +442,8 @@ func TestEnableEpochsHandler_GetActivationEpoch(t *testing.T) {
 	require.Equal(t, cfg.DynamicESDTEnableEpoch, handler.GetActivationEpoch(common.DynamicESDTFlag))
 	require.Equal(t, cfg.EGLDInMultiTransferEnableEpoch, handler.GetActivationEpoch(common.EGLDInESDTMultiTransferFlag))
 	require.Equal(t, cfg.CryptoOpcodesV2EnableEpoch, handler.GetActivationEpoch(common.CryptoOpcodesV2Flag))
+	require.Equal(t, cfg.EquivalentMessagesEnableEpoch, handler.GetActivationEpoch(common.EquivalentMessagesFlag))
+	require.Equal(t, cfg.FixedOrderInConsensusEnableEpoch, handler.GetActivationEpoch(common.FixedOrderInConsensusFlag))
 }
 
 func TestEnableEpochsHandler_IsInterfaceNil(t *testing.T) {

config/config.go

@@ -225,9 +225,12 @@ type Config struct {
 	Requesters            RequesterConfig
 	VMOutputCacher        CacheConfig
 
-	PeersRatingConfig   PeersRatingConfig
-	PoolsCleanersConfig PoolsCleanersConfig
-	Redundancy          RedundancyConfig
+	PeersRatingConfig         PeersRatingConfig
+	PoolsCleanersConfig       PoolsCleanersConfig
+	Redundancy                RedundancyConfig
+	ConsensusGradualBroadcast ConsensusGradualBroadcastConfig
+
+	InterceptedDataVerifier InterceptedDataVerifierConfig
 }
 
 // PeersRatingConfig will hold settings related to peers rating
@@ -667,3 +670,20 @@ type ChainParametersByEpochConfig struct {
 	MetachainMinNumNodes        uint32
 	Adaptivity                  bool
 }
+
+// IndexBroadcastDelay holds a pair of starting consensus index and the delay the nodes should wait before broadcasting final info
+type IndexBroadcastDelay struct {
+	EndIndex            int
+	DelayInMilliseconds uint64
+}
+
+// ConsensusGradualBroadcastConfig holds the configuration for the consensus final info gradual broadcast
+type ConsensusGradualBroadcastConfig struct {
+	GradualIndexBroadcastDelay []IndexBroadcastDelay
+}
+
+// InterceptedDataVerifierConfig holds the configuration for the intercepted data verifier
+type InterceptedDataVerifierConfig struct {
+	CacheSpanInSec   uint64
+	CacheExpiryInSec uint64
+}

config/epochConfig.go

@@ -117,6 +117,8 @@ type EnableEpochs struct {
 	DynamicESDTEnableEpoch                                   uint32
 	EGLDInMultiTransferEnableEpoch                           uint32
 	CryptoOpcodesV2EnableEpoch                               uint32
+	EquivalentMessagesEnableEpoch                            uint32
+	FixedOrderInConsensusEnableEpoch                         uint32
 	BLSMultiSignerEnableEpoch                                []MultiSignerConfig
 }
 

config/tomlConfig_test.go

@@ -162,6 +162,14 @@ func TestTomlParser(t *testing.T) {
 		Redundancy: RedundancyConfig{
 			MaxRoundsOfInactivityAccepted: 3,
 		},
+		ConsensusGradualBroadcast: ConsensusGradualBroadcastConfig{
+			GradualIndexBroadcastDelay: []IndexBroadcastDelay{
+				{
+					EndIndex:            0,
+					DelayInMilliseconds: 0,
+				},
+			},
+		},
 	}
 	testString := `
 [GeneralSettings]
@@ -268,6 +276,13 @@ func TestTomlParser(t *testing.T) {
     # MaxRoundsOfInactivityAccepted defines the number of rounds missed by a main or higher level backup machine before
     # the current machine will take over and propose/sign blocks. Used in both single-key and multi-key modes.
     MaxRoundsOfInactivityAccepted = 3
+
+# ConsensusGradualBroadcast defines how validators will broadcast the aggregated final info, based on their consensus index
+[ConsensusGradualBroadcast]
+    GradualIndexBroadcastDelay = [
+        # All validators will broadcast the message right away
+        { EndIndex = 0, DelayInMilliseconds = 0 },
+    ]
 `
 	cfg := Config{}
 
@@ -890,6 +905,12 @@ func TestEnableEpochConfig(t *testing.T) {
     # CryptoOpcodesV2EnableEpoch represents the epoch when BLSMultiSig, Secp256r1 and other opcodes are enabled
     CryptoOpcodesV2EnableEpoch = 98
 
+	# EquivalentMessagesEnableEpoch represents the epoch when the equivalent messages are enabled
+	EquivalentMessagesEnableEpoch = 99
+
+	# FixedOrderInConsensusEnableEpoch represents the epoch when the fixed order in consensus is enabled
+	FixedOrderInConsensusEnableEpoch = 100
+
     # MaxNodesChangeEnableEpoch holds configuration for changing the maximum number of nodes and the enabling epoch
     MaxNodesChangeEnableEpoch = [
         { EpochEnable = 44, MaxNumNodes = 2169, NodesToShufflePerShard = 80 },
@@ -1006,6 +1027,8 @@ func TestEnableEpochConfig(t *testing.T) {
 			DynamicESDTEnableEpoch:                                   96,
 			EGLDInMultiTransferEnableEpoch:                           97,
 			CryptoOpcodesV2EnableEpoch:                               98,
+			EquivalentMessagesEnableEpoch:                            99,
+			FixedOrderInConsensusEnableEpoch:                         100,
 			MaxNodesChangeEnableEpoch: []MaxNodesChangeConfig{
 				{
 					EpochEnable:            44,

consensus/broadcast/commonMessenger.go

@@ -7,41 +7,29 @@ import (
 	"github.com/multiversx/mx-chain-core-go/core"
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/core/partitioning"
-	"github.com/multiversx/mx-chain-core-go/data"
+	"github.com/multiversx/mx-chain-core-go/data/block"
 	"github.com/multiversx/mx-chain-core-go/hashing"
 	"github.com/multiversx/mx-chain-core-go/marshal"
 	crypto "github.com/multiversx/mx-chain-crypto-go"
+	logger "github.com/multiversx/mx-chain-logger-go"
+
 	"github.com/multiversx/mx-chain-go/common"
 	"github.com/multiversx/mx-chain-go/consensus"
 	"github.com/multiversx/mx-chain-go/consensus/spos"
 	"github.com/multiversx/mx-chain-go/process"
 	"github.com/multiversx/mx-chain-go/process/factory"
 	"github.com/multiversx/mx-chain-go/sharding"
-	logger "github.com/multiversx/mx-chain-logger-go"
 )
 
 var log = logger.GetOrCreate("consensus/broadcast")
 
-// delayedBroadcaster exposes functionality for handling the consensus members broadcasting of delay data
-type delayedBroadcaster interface {
-	SetLeaderData(data *delayedBroadcastData) error
-	SetValidatorData(data *delayedBroadcastData) error
-	SetHeaderForValidator(vData *validatorHeaderBroadcastData) error
-	SetBroadcastHandlers(
-		mbBroadcast func(mbData map[uint32][]byte, pkBytes []byte) error,
-		txBroadcast func(txData map[string][][]byte, pkBytes []byte) error,
-		headerBroadcast func(header data.HeaderHandler, pkBytes []byte) error,
-	) error
-	Close()
-}
-
 type commonMessenger struct {
 	marshalizer             marshal.Marshalizer
 	hasher                  hashing.Hasher
 	messenger               consensus.P2PMessenger
 	shardCoordinator        sharding.Coordinator
 	peerSignatureHandler    crypto.PeerSignatureHandler
-	delayedBlockBroadcaster delayedBroadcaster
+	delayedBlockBroadcaster DelayedBroadcaster
 	keysHandler             consensus.KeysHandler
 }
 
@@ -58,6 +46,7 @@ type CommonMessengerArgs struct {
 	MaxValidatorDelayCacheSize uint32
 	AlarmScheduler             core.TimersScheduler
 	KeysHandler                consensus.KeysHandler
+	DelayedBroadcaster         DelayedBroadcaster
 }
 
 func checkCommonMessengerNilParameters(
@@ -93,6 +82,9 @@ func checkCommonMessengerNilParameters(
 	if check.IfNil(args.KeysHandler) {
 		return ErrNilKeysHandler
 	}
+	if check.IfNil(args.DelayedBroadcaster) {
+		return ErrNilDelayedBroadcaster
+	}
 
 	return nil
 }
@@ -195,6 +187,18 @@ func (cm *commonMessenger) BroadcastBlockData(
 	}
 }
 
+// PrepareBroadcastEquivalentProof sets the proof into the delayed block broadcaster
+func (cm *commonMessenger) PrepareBroadcastEquivalentProof(
+	proof *block.HeaderProof,
+	consensusIndex int,
+	pkBytes []byte,
+) {
+	err := cm.delayedBlockBroadcaster.SetFinalProofForValidator(proof, consensusIndex, pkBytes)
+	if err != nil {
+		log.Error("commonMessenger.PrepareBroadcastEquivalentProof", "error", err)
+	}
+}
+
 func (cm *commonMessenger) extractMetaMiniBlocksAndTransactions(
 	miniBlocks map[uint32][]byte,
 	transactions map[string][][]byte,
@@ -241,3 +245,18 @@ func (cm *commonMessenger) broadcast(topic string, data []byte, pkBytes []byte)
 
 	cm.messenger.BroadcastUsingPrivateKey(topic, data, pid, skBytes)
 }
+
+func (cm *commonMessenger) broadcastEquivalentProof(proof *block.HeaderProof, pkBytes []byte, topic string) error {
+	if check.IfNilReflect(proof) {
+		return spos.ErrNilHeaderProof
+	}
+
+	msgProof, err := cm.marshalizer.Marshal(proof)
+	if err != nil {
+		return err
+	}
+
+	cm.broadcast(topic, msgProof, pkBytes)
+
+	return nil
+}