Heartbeats chain simulator

cmd/node/config/external.toml

@@ -12,8 +12,8 @@
     Username          = ""
     Password          = ""
     # EnabledIndexes represents a slice of indexes that will be enabled for indexing. Full list is:
-    # ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations", "esdts"]
-    EnabledIndexes    = ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations", "esdts"]
+    # ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations", "esdts", "events"]
+    EnabledIndexes    = ["rating", "transactions", "blocks", "validators", "miniblocks", "rounds", "accounts", "accountshistory", "receipts", "scresults", "accountsesdt", "accountsesdthistory", "epochinfo", "scdeploys", "tokens", "tags", "logs", "delegators", "operations", "esdts", "events"]
 
 # EventNotifierConnector defines settings needed to configure and launch the event notifier component
 # HTTP event notifier connector integration will be DEPRECATED in the following iterations

integrationTests/chainSimulator/vm/esdtImprovements_test.go

@@ -3868,7 +3868,9 @@ func TestChainSimulator_CheckRolesWhichHasToBeSingular(t *testing.T) {
 }
 
 func TestChainSimulator_metaESDT_mergeMetaDataFromMultipleUpdates(t *testing.T) {
-	t.Parallel()
+	if testing.Short() {
+		t.Skip("this is not a short test")
+	}
 
 	baseIssuingCost := "1000"
 	cs, _ := getTestChainSimulatorWithDynamicNFTEnabled(t, baseIssuingCost)
@@ -4237,7 +4239,9 @@ func transferSpecialRoleToAddr(
 }
 
 func TestChainSimulator_dynamicNFT_mergeMetaDataFromMultipleUpdates(t *testing.T) {
-	t.Parallel()
+	if testing.Short() {
+		t.Skip("this is not a short test")
+	}
 
 	baseIssuingCost := "1000"
 	cs, _ := getTestChainSimulatorWithDynamicNFTEnabled(t, baseIssuingCost)
@@ -4385,7 +4389,9 @@ func TestChainSimulator_dynamicNFT_mergeMetaDataFromMultipleUpdates(t *testing.T
 }
 
 func TestChainSimulator_dynamicNFT_changeMetaDataForOneNFTShouldNotChangeOtherNonces(t *testing.T) {
-	t.Parallel()
+	if testing.Short() {
+		t.Skip("this is not a short test")
+	}
 
 	baseIssuingCost := "1000"
 	cs, _ := getTestChainSimulatorWithDynamicNFTEnabled(t, baseIssuingCost)
@@ -4498,7 +4504,9 @@ func TestChainSimulator_dynamicNFT_changeMetaDataForOneNFTShouldNotChangeOtherNo
 }
 
 func TestChainSimulator_dynamicNFT_updateBeforeCreateOnSameAccountShouldOverwrite(t *testing.T) {
-	t.Parallel()
+	if testing.Short() {
+		t.Skip("this is not a short test")
+	}
 
 	baseIssuingCost := "1000"
 	cs, _ := getTestChainSimulatorWithDynamicNFTEnabled(t, baseIssuingCost)
@@ -4595,7 +4603,9 @@ func TestChainSimulator_dynamicNFT_updateBeforeCreateOnSameAccountShouldOverwrit
 }
 
 func TestChainSimulator_dynamicNFT_updateBeforeCreateOnDifferentAccountsShouldMergeMetaDataWhenTransferred(t *testing.T) {
-	t.Parallel()
+	if testing.Short() {
+		t.Skip("this is not a short test")
+	}
 
 	baseIssuingCost := "1000"
 	cs, _ := getTestChainSimulatorWithDynamicNFTEnabled(t, baseIssuingCost)

node/chainSimulator/chainSimulator.go

@@ -11,7 +11,9 @@ import (
 	"time"
 
 	"github.com/multiversx/mx-chain-go/config"
+	"github.com/multiversx/mx-chain-go/factory"
 	"github.com/multiversx/mx-chain-go/node/chainSimulator/components"
+	"github.com/multiversx/mx-chain-go/node/chainSimulator/components/heartbeat"
 	"github.com/multiversx/mx-chain-go/node/chainSimulator/configs"
 	"github.com/multiversx/mx-chain-go/node/chainSimulator/dtos"
 	chainSimulatorErrors "github.com/multiversx/mx-chain-go/node/chainSimulator/errors"
@@ -130,18 +132,20 @@ func (s *simulator) createChainHandlers(args ArgsBaseChainSimulator) error {
 		return err
 	}
 
+	monitor := heartbeat.NewHeartbeatMonitor()
+
 	for idx := -1; idx < int(args.NumOfShards); idx++ {
 		shardIDStr := fmt.Sprintf("%d", idx)
 		if idx == -1 {
 			shardIDStr = "metachain"
 		}
 
-		node, errCreate := s.createTestNode(*outputConfigs, args, shardIDStr)
+		node, errCreate := s.createTestNode(*outputConfigs, args, shardIDStr, monitor)
 		if errCreate != nil {
 			return errCreate
 		}
 
-		chainHandler, errCreate := process.NewBlocksCreator(node)
+		chainHandler, errCreate := process.NewBlocksCreator(node, monitor)
 		if errCreate != nil {
 			return errCreate
 		}
@@ -195,7 +199,7 @@ func computeStartTimeBaseOnInitialRound(args ArgsChainSimulator) int64 {
 }
 
 func (s *simulator) createTestNode(
-	outputConfigs configs.ArgsConfigsSimulator, args ArgsBaseChainSimulator, shardIDStr string,
+	outputConfigs configs.ArgsConfigsSimulator, args ArgsBaseChainSimulator, shardIDStr string, monitor factory.HeartbeatV2Monitor,
 ) (process.NodeHandler, error) {
 	argsTestOnlyProcessorNode := components.ArgsTestOnlyProcessingNode{
 		Configs:                     outputConfigs.Configs,
@@ -214,6 +218,7 @@ func (s *simulator) createTestNode(
 		MetaChainConsensusGroupSize: args.MetaChainConsensusGroupSize,
 		RoundDurationInMillis:       args.RoundDurationInMillis,
 		VmQueryDelayAfterStartInMs:  args.VmQueryDelayAfterStartInMs,
+		Monitor:                     monitor,
 	}
 
 	return components.NewTestOnlyProcessingNode(argsTestOnlyProcessorNode)

node/chainSimulator/chainSimulator_test.go

@@ -82,13 +82,17 @@ func TestChainSimulator_GenerateBlocksShouldWork(t *testing.T) {
 	})
 	require.Nil(t, err)
 	require.NotNil(t, chainSimulator)
-
 	defer chainSimulator.Close()
 
 	time.Sleep(time.Second)
 
 	err = chainSimulator.GenerateBlocks(50)
 	require.Nil(t, err)
+
+	heartBeats, err := chainSimulator.GetNodeHandler(0).GetFacadeHandler().GetHeartbeats()
+	require.Nil(t, err)
+	require.Equal(t, 4, len(heartBeats))
+
 }
 
 func TestChainSimulator_GenerateBlocksAndEpochChangeShouldWork(t *testing.T) {

node/chainSimulator/components/heartbeat/heartBeat.go

@@ -0,0 +1,52 @@
+package heartbeat
+
+import (
+	"github.com/multiversx/mx-chain-core-go/core/check"
+	"github.com/multiversx/mx-chain-go/factory"
+	"github.com/multiversx/mx-chain-go/heartbeat"
+)
+
+type heartBeatComponents struct {
+	monitor factory.HeartbeatV2Monitor
+}
+
+// NewSyncedHeartbeatComponents will create a new instance of heartbeat components
+func NewSyncedHeartbeatComponents(monitor factory.HeartbeatV2Monitor) (factory.HeartbeatV2ComponentsHandler, error) {
+	if check.IfNil(monitor) {
+		return nil, heartbeat.ErrNilHeartbeatMonitor
+	}
+
+	return &heartBeatComponents{
+		monitor: monitor,
+	}, nil
+}
+
+// Create will do nothing
+func (h *heartBeatComponents) Create() error {
+	return nil
+}
+
+// Close will do nothing
+func (h *heartBeatComponents) Close() error {
+	return nil
+}
+
+// CheckSubcomponents will do nothing
+func (h *heartBeatComponents) CheckSubcomponents() error {
+	return nil
+}
+
+// String will return a string
+func (h *heartBeatComponents) String() string {
+	return "heartBeat"
+}
+
+// Monitor will return the monitor
+func (h *heartBeatComponents) Monitor() factory.HeartbeatV2Monitor {
+	return h.monitor
+}
+
+// IsInterfaceNil returns true if there is no value under the interface
+func (h *heartBeatComponents) IsInterfaceNil() bool {
+	return h == nil
+}

node/chainSimulator/components/heartbeat/monitor.go

@@ -0,0 +1,39 @@
+package heartbeat
+
+import (
+	"sync"
+
+	"github.com/multiversx/mx-chain-go/heartbeat/data"
+)
+
+type heartbeatMonitor struct {
+	heartbeats []data.PubKeyHeartbeat
+	mutex      sync.RWMutex
+}
+
+// NewHeartbeatMonitor will create a new instance of heartbeat monitor
+func NewHeartbeatMonitor() *heartbeatMonitor {
+	return &heartbeatMonitor{
+		heartbeats: make([]data.PubKeyHeartbeat, 0),
+	}
+}
+
+// GetHeartbeats will return the heartbeats
+func (hm *heartbeatMonitor) GetHeartbeats() []data.PubKeyHeartbeat {
+	hm.mutex.RLock()
+	defer hm.mutex.RUnlock()
+
+	return hm.heartbeats
+}
+
+// SetHeartbeats will set the provided heartbeats
+func (hm *heartbeatMonitor) SetHeartbeats(heartbeats []data.PubKeyHeartbeat) {
+	hm.mutex.Lock()
+	hm.heartbeats = heartbeats
+	hm.mutex.Unlock()
+}
+
+// IsInterfaceNil returns true if there is no value under the interface
+func (hm *heartbeatMonitor) IsInterfaceNil() bool {
+	return nil == hm
+}

node/chainSimulator/components/nodeFacade.go

@@ -3,6 +3,8 @@ package components
 import (
 	"errors"
 	"fmt"
+	"github.com/multiversx/mx-chain-go/factory"
+	heartbeat2 "github.com/multiversx/mx-chain-go/node/chainSimulator/components/heartbeat"
 	"strconv"
 	"time"
 
@@ -18,7 +20,7 @@ import (
 	"github.com/multiversx/mx-chain-go/process/mock"
 )
 
-func (node *testOnlyProcessingNode) createFacade(configs config.Configs, apiInterface APIConfigurator, vmQueryDelayAfterStartInMs uint64) error {
+func (node *testOnlyProcessingNode) createFacade(configs config.Configs, apiInterface APIConfigurator, vmQueryDelayAfterStartInMs uint64, monitor factory.HeartbeatV2Monitor) error {
 	log.Debug("creating api resolver structure")
 
 	err := node.createMetrics(configs)
@@ -73,6 +75,13 @@ func (node *testOnlyProcessingNode) createFacade(configs config.Configs, apiInte
 
 	flagsConfig := configs.FlagsConfig
 
+	heartbeatComponents, err := heartbeat2.NewSyncedHeartbeatComponents(monitor)
+	if err != nil {
+		return err
+	}
+
+	node.closeHandler.AddComponent(heartbeatComponents)
+
 	nd, err := nodePack.NewNode(
 		nodePack.WithStatusCoreComponents(node.StatusCoreComponents),
 		nodePack.WithCoreComponents(node.CoreComponentsHolder),
@@ -95,6 +104,7 @@ func (node *testOnlyProcessingNode) createFacade(configs config.Configs, apiInte
 		nodePack.WithNodeStopChannel(node.CoreComponentsHolder.ChanStopNodeProcess()),
 		nodePack.WithImportMode(configs.ImportDbConfig.IsImportDBMode),
 		nodePack.WithESDTNFTStorageHandler(node.ProcessComponentsHolder.ESDTDataStorageHandlerForAPI()),
+		nodePack.WithHeartbeatV2Components(heartbeatComponents),
 	)
 	if err != nil {
 		return errors.New("error creating node: " + err.Error())

node/chainSimulator/components/statusComponents.go

@@ -9,16 +9,22 @@ import (
 	"github.com/multiversx/mx-chain-core-go/core"
 	"github.com/multiversx/mx-chain-core-go/core/appStatusPolling"
 	"github.com/multiversx/mx-chain-core-go/core/check"
+	nodeData "github.com/multiversx/mx-chain-core-go/data"
+	outportCore "github.com/multiversx/mx-chain-core-go/data/outport"
 	factoryMarshalizer "github.com/multiversx/mx-chain-core-go/marshal/factory"
 	indexerFactory "github.com/multiversx/mx-chain-es-indexer-go/process/factory"
 	"github.com/multiversx/mx-chain-go/common"
 	"github.com/multiversx/mx-chain-go/common/statistics"
 	"github.com/multiversx/mx-chain-go/config"
+	"github.com/multiversx/mx-chain-go/epochStart"
+	"github.com/multiversx/mx-chain-go/epochStart/notifier"
 	"github.com/multiversx/mx-chain-go/errors"
+	"github.com/multiversx/mx-chain-go/factory"
 	"github.com/multiversx/mx-chain-go/integrationTests/mock"
 	"github.com/multiversx/mx-chain-go/outport"
-	"github.com/multiversx/mx-chain-go/outport/factory"
+	outportFactory "github.com/multiversx/mx-chain-go/outport/factory"
 	"github.com/multiversx/mx-chain-go/process"
+	"github.com/multiversx/mx-chain-go/sharding/nodesCoordinator"
 	"github.com/multiversx/mx-chain-go/testscommon"
 )
 
@@ -32,10 +38,11 @@ type statusComponentsHolder struct {
 	statusPollingIntervalSec int
 	cancelFunc               func()
 	mutex                    sync.RWMutex
+	nodesCoordinator         nodesCoordinator.NodesCoordinator
 }
 
 // CreateStatusComponents will create a new instance of status components holder
-func CreateStatusComponents(shardID uint32, appStatusHandler core.AppStatusHandler, statusPollingIntervalSec int, external config.ExternalConfig, coreComponents process.CoreComponentsHolder) (*statusComponentsHolder, error) {
+func CreateStatusComponents(shardID uint32, appStatusHandler core.AppStatusHandler, statusPollingIntervalSec int, external config.ExternalConfig, coreComponents factory.CoreComponentsHandler) (*statusComponentsHolder, error) {
 	if check.IfNil(appStatusHandler) {
 		return nil, core.ErrNilAppStatusHandler
 	}
@@ -51,12 +58,12 @@ func CreateStatusComponents(shardID uint32, appStatusHandler core.AppStatusHandl
 	if err != nil {
 		return nil, err
 	}
-	instance.outportHandler, err = factory.CreateOutport(&factory.OutportFactoryArgs{
+	instance.outportHandler, err = outportFactory.CreateOutport(&outportFactory.OutportFactoryArgs{
 		IsImportDB:                false,
 		ShardID:                   shardID,
 		RetrialInterval:           time.Second,
 		HostDriversArgs:           hostDriverArgs,
-		EventNotifierFactoryArgs:  &factory.EventNotifierFactoryArgs{},
+		EventNotifierFactoryArgs:  &outportFactory.EventNotifierFactoryArgs{},
 		ElasticIndexerFactoryArgs: makeElasticIndexerArgs(external, coreComponents),
 	})
 	if err != nil {
@@ -65,13 +72,45 @@ func CreateStatusComponents(shardID uint32, appStatusHandler core.AppStatusHandl
 	instance.softwareVersionChecker = &mock.SoftwareVersionCheckerMock{}
 	instance.managedPeerMonitor = &testscommon.ManagedPeersMonitorStub{}
 
+	if shardID == core.MetachainShardId {
+		coreComponents.EpochStartNotifierWithConfirm().RegisterHandler(instance.epochStartEventHandler())
+	}
+
 	instance.collectClosableComponents()
 
 	return instance, nil
 }
 
-func makeHostDriversArgs(external config.ExternalConfig) ([]factory.ArgsHostDriverFactory, error) {
-	argsHostDriverFactorySlice := make([]factory.ArgsHostDriverFactory, 0, len(external.HostDriversConfig))
+// SetNodesCoordinator will set the nodes coordinator
+func (s *statusComponentsHolder) SetNodesCoordinator(nodesCoordinator nodesCoordinator.NodesCoordinator) {
+	s.mutex.Lock()
+	s.nodesCoordinator = nodesCoordinator
+	s.mutex.Unlock()
+}
+
+func (s *statusComponentsHolder) epochStartEventHandler() epochStart.ActionHandler {
+	subscribeHandler := notifier.NewHandlerForEpochStart(func(hdr nodeData.HeaderHandler) {
+		currentEpoch := hdr.GetEpoch()
+		validatorsPubKeys, err := s.nodesCoordinator.GetAllEligibleValidatorsPublicKeys(currentEpoch)
+		if err != nil {
+			log.Warn("s.nodesCoordinator.GetAllEligibleValidatorPublicKeys for current epoch failed",
+				"epoch", currentEpoch,
+				"error", err.Error())
+		}
+
+		s.outportHandler.SaveValidatorsPubKeys(&outportCore.ValidatorsPubKeys{
+			ShardID:                hdr.GetShardID(),
+			ShardValidatorsPubKeys: outportCore.ConvertPubKeys(validatorsPubKeys),
+			Epoch:                  currentEpoch,
+		})
+
+	}, func(_ nodeData.HeaderHandler) {}, common.IndexerOrder)
+
+	return subscribeHandler
+}
+
+func makeHostDriversArgs(external config.ExternalConfig) ([]outportFactory.ArgsHostDriverFactory, error) {
+	argsHostDriverFactorySlice := make([]outportFactory.ArgsHostDriverFactory, 0, len(external.HostDriversConfig))
 	for idx := 0; idx < len(external.HostDriversConfig); idx++ {
 		hostConfig := external.HostDriversConfig[idx]
 		if !hostConfig.Enabled {
@@ -83,7 +122,7 @@ func makeHostDriversArgs(external config.ExternalConfig) ([]factory.ArgsHostDriv
 			return argsHostDriverFactorySlice, err
 		}
 
-		argsHostDriverFactorySlice = append(argsHostDriverFactorySlice, factory.ArgsHostDriverFactory{
+		argsHostDriverFactorySlice = append(argsHostDriverFactorySlice, outportFactory.ArgsHostDriverFactory{
 			Marshaller: marshaller,
 			HostConfig: hostConfig,
 		})