Feat/HeartbeatV2

cmd/node/config/config.toml

@@ -896,3 +896,29 @@
     NumCrossShardPeers  = 2
     NumIntraShardPeers  = 1
     NumFullHistoryPeers = 3
+
+[HeartbeatV2]
+    PeerAuthenticationTimeBetweenSendsInSec          = 7200 # 2h
+    PeerAuthenticationTimeBetweenSendsWhenErrorInSec = 60   # 1min
+    PeerAuthenticationThresholdBetweenSends          = 0.1  # 10%
+    HeartbeatTimeBetweenSendsInSec                   = 60   # 1min
+    HeartbeatTimeBetweenSendsWhenErrorInSec          = 60   # 1min
+    HeartbeatThresholdBetweenSends                   = 0.1  # 10%
+    MaxNumOfPeerAuthenticationInResponse             = 10
+    HeartbeatExpiryTimespanInSec                     = 3600 # 1h
+    MinPeersThreshold                                = 0.8  # 80%
+    DelayBetweenRequestsInSec                        = 10   # 10sec
+    MaxTimeoutInSec                                  = 7200 # 2h
+    DelayBetweenConnectionNotificationsInSec         = 60   # 1min
+    MaxMissingKeysInRequest                          = 1000
+    MaxDurationPeerUnresponsiveInSec                 = 900  # 15min
+    HideInactiveValidatorIntervalInSec               = 3600 # 1h
+    HardforkTimeBetweenSendsInSec                    = 10   # 10sec
+    [HeartbeatV2.PeerAuthenticationPool]
+        DefaultSpanInSec = 3600 # 1h
+        CacheExpiryInSec = 3600 # 1h
+    [HeartbeatV2.HeartbeatPool]
+        Name = "HeartbeatPool"
+        Capacity = 1000
+        Type = "SizeLRU"
+        SizeInBytes = 314572800 #300MB

cmd/node/config/enableEpochs.toml

@@ -199,6 +199,9 @@
         { EpochEnable = 1, MaxNumNodes = 56, NodesToShufflePerShard = 2 }
     ]
 
+    # HeartbeatDisableEpoch represents the epoch when heartbeat v1 messages stop being sent and processed
+    HeartbeatDisableEpoch = 2
+
 [GasSchedule]
     GasScheduleByEpochs = [
         { StartEpoch = 0, FileName = "gasScheduleV1.toml" },

cmd/node/config/prefs.toml

@@ -18,11 +18,11 @@
    # It is highly recommended to enable this flag on an observer (not on a validator node)
    FullArchive = false
 
-   # PreferredConnections holds an array containing the public keys of the nodes to connect with (in top of other connections)
+   # PreferredConnections holds an array containing valid ips or peer ids from nodes to connect with (in top of other connections)
    # Example:
    # PreferredConnections = [
-   #    "eb2a13ec773924df2c7d1e92ff1c08d1c3b14218dc6a780b269ef12b9c098971f71851c212103720d40f92380c306a0c1a5e606f043f034188c3fcb95170112158730e2c53cd6c79331ce73df921675d71488f6287aa1ddca297756a98239584",
-   #    "eb2a13ec773924df2c7d1e92ff1c08d1c3b14218dc6a780b269ef12b9c098971f71851c212103720d40f92380c306a0c1a5e606f043f034188c3fcb95170112158730e2c53cd6c79331ce73df921675d71488f6287aa1ddca297756a98239584"
+   #    "127.0.0.10",
+   #    "16Uiu2HAm6yvbp1oZ6zjnWsn9FdRqBSaQkbhELyaThuq48ybdorrr"
    # ]
 
    PreferredConnections = []

common/constants.go

@@ -63,6 +63,15 @@ const GenesisTxSignatureString = "GENESISGENESISGENESISGENESISGENESISGENESISGENE
 // HeartbeatTopic is the topic used for heartbeat signaling
 const HeartbeatTopic = "heartbeat"
 
+// HeartbeatV2Topic is the topic used for heartbeatV2 signaling
+const HeartbeatV2Topic = "heartbeatV2"
+
+// PeerAuthenticationTopic is the topic used for peer authentication signaling
+const PeerAuthenticationTopic = "peerAuthentication"
+
+// ConnectionTopic represents the topic used when sending the new connection message data
+const ConnectionTopic = "connection"
+
 // PathShardPlaceholder represents the placeholder for the shard ID in paths
 const PathShardPlaceholder = "[S]"
 
@@ -485,6 +494,9 @@ const (
 	// MetricWaitingListFixEnableEpoch represents the epoch when the waiting list fix is enabled
 	MetricWaitingListFixEnableEpoch = "erd_waiting_list_fix_enable_epoch"
 
+	// MetricHeartbeatDisableEpoch represents the epoch when heartbeat v1 messages stop being sent and processed
+	MetricHeartbeatDisableEpoch = "erd_heartbeat_disable_epoch"
+
 	// MetricMaxNodesChangeEnableEpoch holds configuration for changing the maximum number of nodes and the enabling epoch
 	MetricMaxNodesChangeEnableEpoch = "erd_max_nodes_change_enable_epoch"
 

config/config.go

@@ -103,6 +103,34 @@ type SoftwareVersionConfig struct {
 	PollingIntervalInMinutes int
 }
 
+// HeartbeatV2Config will hold the configuration for heartbeat v2
+type HeartbeatV2Config struct {
+	PeerAuthenticationTimeBetweenSendsInSec          int64
+	PeerAuthenticationTimeBetweenSendsWhenErrorInSec int64
+	PeerAuthenticationThresholdBetweenSends          float64
+	HeartbeatTimeBetweenSendsInSec                   int64
+	HeartbeatTimeBetweenSendsWhenErrorInSec          int64
+	HeartbeatThresholdBetweenSends                   float64
+	MaxNumOfPeerAuthenticationInResponse             int
+	HeartbeatExpiryTimespanInSec                     int64
+	MinPeersThreshold                                float32
+	DelayBetweenRequestsInSec                        int64
+	MaxTimeoutInSec                                  int64
+	DelayBetweenConnectionNotificationsInSec         int64
+	MaxMissingKeysInRequest                          uint32
+	MaxDurationPeerUnresponsiveInSec                 int64
+	HideInactiveValidatorIntervalInSec               int64
+	PeerAuthenticationPool                           PeerAuthenticationPoolConfig
+	HeartbeatPool                                    CacheConfig
+	HardforkTimeBetweenSendsInSec                    int64
+}
+
+// PeerAuthenticationPoolConfig will hold the configuration for peer authentication pool
+type PeerAuthenticationPoolConfig struct {
+	DefaultSpanInSec int
+	CacheExpiryInSec int
+}
+
 // Config will hold the entire application configuration parameters
 type Config struct {
 	MiniBlocksStorage               StorageConfig
@@ -162,6 +190,7 @@ type Config struct {
 	Antiflood           AntifloodConfig
 	ResourceStats       ResourceStatsConfig
 	Heartbeat           HeartbeatConfig
+	HeartbeatV2         HeartbeatV2Config
 	ValidatorStatistics ValidatorStatisticsConfig
 	GeneralSettings     GeneralSettingsConfig
 	Consensus           ConsensusConfig

config/epochConfig.go

@@ -77,6 +77,7 @@ type EnableEpochs struct {
 	SCRSizeInvariantOnBuiltInResultEnableEpoch        uint32
 	CheckCorrectTokenIDForTransferRoleEnableEpoch     uint32
 	FailExecutionOnEveryAPIErrorEnableEpoch           uint32
+	HeartbeatDisableEpoch                             uint32
 }
 
 // GasScheduleByEpochs represents a gas schedule toml entry that will be applied from the provided epoch

config/tomlConfig_test.go

@@ -654,6 +654,9 @@ func TestEnableEpochConfig(t *testing.T) {
 	# FailExecutionOnEveryAPIErrorEnableEpoch represent the epoch when new protection in VM is enabled to fail all wrong API calls
 	FailExecutionOnEveryAPIErrorEnableEpoch = 53
 
+    # HeartbeatDisableEpoch represents the epoch when heartbeat v1 messages stop being sent and processed
+    HeartbeatDisableEpoch = 54
+
     # MaxNodesChangeEnableEpoch holds configuration for changing the maximum number of nodes and the enabling epoch
     MaxNodesChangeEnableEpoch = [
         { EpochEnable = 44, MaxNumNodes = 2169, NodesToShufflePerShard = 80 },
@@ -734,6 +737,7 @@ func TestEnableEpochConfig(t *testing.T) {
 			TransformToMultiShardCreateEnableEpoch:      51,
 			ESDTRegisterAndSetAllRolesEnableEpoch:       52,
 			FailExecutionOnEveryAPIErrorEnableEpoch:     53,
+			HeartbeatDisableEpoch:                       54,
 		},
 		GasSchedule: GasScheduleConfig{
 			GasScheduleByEpochs: []GasScheduleByEpochs{

consensus/interface.go

@@ -80,7 +80,7 @@ type P2PMessenger interface {
 // The interface assures that the collected data will be used by the p2p network sharding components
 type NetworkShardingCollector interface {
 	UpdatePeerIDInfo(pid core.PeerID, pk []byte, shardID uint32)
-	UpdatePeerIdSubType(pid core.PeerID, peerSubType core.P2PPeerSubType)
+	PutPeerIdSubType(pid core.PeerID, peerSubType core.P2PPeerSubType)
 	GetPeerInfo(pid core.PeerID) core.P2PPeerInfo
 	IsInterfaceNil() bool
 }

consensus/spos/worker.go

@@ -51,8 +51,6 @@ type Worker struct {
 	headerIntegrityVerifier process.HeaderIntegrityVerifier
 	appStatusHandler        core.AppStatusHandler
 
-	networkShardingCollector consensus.NetworkShardingCollector
-
 	receivedMessages      map[consensus.MessageType][]*consensus.Message
 	receivedMessagesCalls map[consensus.MessageType]func(ctx context.Context, msg *consensus.Message) bool
 
@@ -79,30 +77,29 @@ type Worker struct {
 
 // WorkerArgs holds the consensus worker arguments
 type WorkerArgs struct {
-	ConsensusService         ConsensusService
-	BlockChain               data.ChainHandler
-	BlockProcessor           process.BlockProcessor
-	ScheduledProcessor       consensus.ScheduledProcessor
-	Bootstrapper             process.Bootstrapper
-	BroadcastMessenger       consensus.BroadcastMessenger
-	ConsensusState           *ConsensusState
-	ForkDetector             process.ForkDetector
-	Marshalizer              marshal.Marshalizer
-	Hasher                   hashing.Hasher
-	RoundHandler             consensus.RoundHandler
-	ShardCoordinator         sharding.Coordinator
-	PeerSignatureHandler     crypto.PeerSignatureHandler
-	SyncTimer                ntp.SyncTimer
-	HeaderSigVerifier        HeaderSigVerifier
-	HeaderIntegrityVerifier  process.HeaderIntegrityVerifier
-	ChainID                  []byte
-	NetworkShardingCollector consensus.NetworkShardingCollector
-	AntifloodHandler         consensus.P2PAntifloodHandler
-	PoolAdder                PoolAdder
-	SignatureSize            int
-	PublicKeySize            int
-	AppStatusHandler         core.AppStatusHandler
-	NodeRedundancyHandler    consensus.NodeRedundancyHandler
+	ConsensusService        ConsensusService
+	BlockChain              data.ChainHandler
+	BlockProcessor          process.BlockProcessor
+	ScheduledProcessor      consensus.ScheduledProcessor
+	Bootstrapper            process.Bootstrapper
+	BroadcastMessenger      consensus.BroadcastMessenger
+	ConsensusState          *ConsensusState
+	ForkDetector            process.ForkDetector
+	Marshalizer             marshal.Marshalizer
+	Hasher                  hashing.Hasher
+	RoundHandler            consensus.RoundHandler
+	ShardCoordinator        sharding.Coordinator
+	PeerSignatureHandler    crypto.PeerSignatureHandler
+	SyncTimer               ntp.SyncTimer
+	HeaderSigVerifier       HeaderSigVerifier
+	HeaderIntegrityVerifier process.HeaderIntegrityVerifier
+	ChainID                 []byte
+	AntifloodHandler        consensus.P2PAntifloodHandler
+	PoolAdder               PoolAdder
+	SignatureSize           int
+	PublicKeySize           int
+	AppStatusHandler        core.AppStatusHandler
+	NodeRedundancyHandler   consensus.NodeRedundancyHandler
 }
 
 // NewWorker creates a new Worker object
@@ -128,28 +125,27 @@ func NewWorker(args *WorkerArgs) (*Worker, error) {
 	}
 
 	wrk := Worker{
-		consensusService:         args.ConsensusService,
-		blockChain:               args.BlockChain,
-		blockProcessor:           args.BlockProcessor,
-		scheduledProcessor:       args.ScheduledProcessor,
-		bootstrapper:             args.Bootstrapper,
-		broadcastMessenger:       args.BroadcastMessenger,
-		consensusState:           args.ConsensusState,
-		forkDetector:             args.ForkDetector,
-		marshalizer:              args.Marshalizer,
-		hasher:                   args.Hasher,
-		roundHandler:             args.RoundHandler,
-		shardCoordinator:         args.ShardCoordinator,
-		peerSignatureHandler:     args.PeerSignatureHandler,
-		syncTimer:                args.SyncTimer,
-		headerSigVerifier:        args.HeaderSigVerifier,
-		headerIntegrityVerifier:  args.HeaderIntegrityVerifier,
-		appStatusHandler:         args.AppStatusHandler,
-		networkShardingCollector: args.NetworkShardingCollector,
-		antifloodHandler:         args.AntifloodHandler,
-		poolAdder:                args.PoolAdder,
-		nodeRedundancyHandler:    args.NodeRedundancyHandler,
-		closer:                   closing.NewSafeChanCloser(),
+		consensusService:        args.ConsensusService,
+		blockChain:              args.BlockChain,
+		blockProcessor:          args.BlockProcessor,
+		scheduledProcessor:      args.ScheduledProcessor,
+		bootstrapper:            args.Bootstrapper,
+		broadcastMessenger:      args.BroadcastMessenger,
+		consensusState:          args.ConsensusState,
+		forkDetector:            args.ForkDetector,
+		marshalizer:             args.Marshalizer,
+		hasher:                  args.Hasher,
+		roundHandler:            args.RoundHandler,
+		shardCoordinator:        args.ShardCoordinator,
+		peerSignatureHandler:    args.PeerSignatureHandler,
+		syncTimer:               args.SyncTimer,
+		headerSigVerifier:       args.HeaderSigVerifier,
+		headerIntegrityVerifier: args.HeaderIntegrityVerifier,
+		appStatusHandler:        args.AppStatusHandler,
+		antifloodHandler:        args.AntifloodHandler,
+		poolAdder:               args.PoolAdder,
+		nodeRedundancyHandler:   args.NodeRedundancyHandler,
+		closer:                  closing.NewSafeChanCloser(),
 	}
 
 	wrk.consensusMessageValidator = consensusMessageValidatorObj
@@ -232,9 +228,6 @@ func checkNewWorkerParams(args *WorkerArgs) error {
 	if len(args.ChainID) == 0 {
 		return ErrInvalidChainID
 	}
-	if check.IfNil(args.NetworkShardingCollector) {
-		return ErrNilNetworkShardingCollector
-	}
 	if check.IfNil(args.AntifloodHandler) {
 		return ErrNilAntifloodHandler
 	}
@@ -381,8 +374,6 @@ func (wrk *Worker) ProcessReceivedMessage(message p2p.MessageP2P, fromConnectedP
 		return err
 	}
 
-	wrk.networkShardingCollector.UpdatePeerIDInfo(message.Peer(), cnsMsg.PubKey, wrk.shardCoordinator.SelfId())
-
 	isMessageWithBlockBody := wrk.consensusService.IsMessageWithBlockBody(msgType)
 	isMessageWithBlockHeader := wrk.consensusService.IsMessageWithBlockHeader(msgType)
 	isMessageWithBlockBodyAndHeader := wrk.consensusService.IsMessageWithBlockBodyAndHeader(msgType)

consensus/spos/worker_test.go

@@ -22,7 +22,6 @@ import (
 	"github.com/ElrondNetwork/elrond-go/process"
 	"github.com/ElrondNetwork/elrond-go/testscommon"
 	"github.com/ElrondNetwork/elrond-go/testscommon/hashingMocks"
-	"github.com/ElrondNetwork/elrond-go/testscommon/p2pmocks"
 	statusHandlerMock "github.com/ElrondNetwork/elrond-go/testscommon/statusHandler"
 	"github.com/stretchr/testify/assert"
 )
@@ -86,30 +85,29 @@ func createDefaultWorkerArgs(appStatusHandler core.AppStatusHandler) *spos.Worke
 
 	peerSigHandler := &mock.PeerSignatureHandler{Signer: singleSignerMock, KeyGen: keyGeneratorMock}
 	workerArgs := &spos.WorkerArgs{
-		ConsensusService:         blsService,
-		BlockChain:               blockchainMock,
-		BlockProcessor:           blockProcessor,
-		ScheduledProcessor:       scheduledProcessor,
-		Bootstrapper:             bootstrapperMock,
-		BroadcastMessenger:       broadcastMessengerMock,
-		ConsensusState:           consensusState,
-		ForkDetector:             forkDetectorMock,
-		Marshalizer:              marshalizerMock,
-		Hasher:                   hasher,
-		RoundHandler:             roundHandlerMock,
-		ShardCoordinator:         shardCoordinatorMock,
-		PeerSignatureHandler:     peerSigHandler,
-		SyncTimer:                syncTimerMock,
-		HeaderSigVerifier:        &mock.HeaderSigVerifierStub{},
-		HeaderIntegrityVerifier:  &mock.HeaderIntegrityVerifierStub{},
-		ChainID:                  chainID,
-		NetworkShardingCollector: &p2pmocks.NetworkShardingCollectorStub{},
-		AntifloodHandler:         createMockP2PAntifloodHandler(),
-		PoolAdder:                poolAdder,
-		SignatureSize:            SignatureSize,
-		PublicKeySize:            PublicKeySize,
-		AppStatusHandler:         appStatusHandler,
-		NodeRedundancyHandler:    &mock.NodeRedundancyHandlerStub{},
+		ConsensusService:        blsService,
+		BlockChain:              blockchainMock,
+		BlockProcessor:          blockProcessor,
+		ScheduledProcessor:      scheduledProcessor,
+		Bootstrapper:            bootstrapperMock,
+		BroadcastMessenger:      broadcastMessengerMock,
+		ConsensusState:          consensusState,
+		ForkDetector:            forkDetectorMock,
+		Marshalizer:             marshalizerMock,
+		Hasher:                  hasher,
+		RoundHandler:            roundHandlerMock,
+		ShardCoordinator:        shardCoordinatorMock,
+		PeerSignatureHandler:    peerSigHandler,
+		SyncTimer:               syncTimerMock,
+		HeaderSigVerifier:       &mock.HeaderSigVerifierStub{},
+		HeaderIntegrityVerifier: &mock.HeaderIntegrityVerifierStub{},
+		ChainID:                 chainID,
+		AntifloodHandler:        createMockP2PAntifloodHandler(),
+		PoolAdder:               poolAdder,
+		SignatureSize:           SignatureSize,
+		PublicKeySize:           PublicKeySize,
+		AppStatusHandler:        appStatusHandler,
+		NodeRedundancyHandler:   &mock.NodeRedundancyHandlerStub{},
 	}
 
 	return workerArgs
@@ -320,17 +318,6 @@ func TestWorker_NewWorkerEmptyChainIDShouldFail(t *testing.T) {
 	assert.Equal(t, spos.ErrInvalidChainID, err)
 }
 
-func TestWorker_NewWorkerNilNetworkShardingCollectorShouldFail(t *testing.T) {
-	t.Parallel()
-
-	workerArgs := createDefaultWorkerArgs(&statusHandlerMock.AppStatusHandlerStub{})
-	workerArgs.NetworkShardingCollector = nil
-	wrk, err := spos.NewWorker(workerArgs)
-
-	assert.Nil(t, wrk)
-	assert.Equal(t, spos.ErrNilNetworkShardingCollector, err)
-}
-
 func TestWorker_NewWorkerNilAntifloodHandlerShouldFail(t *testing.T) {
 	t.Parallel()