Add new event tracker flags to server command

Author: goran-ethernal

command/server/config/config.go

@@ -32,13 +32,17 @@ type Config struct {
 	JSONLogFormat            bool       `json:"json_log_format" yaml:"json_log_format"`
 	CorsAllowedOrigins       []string   `json:"cors_allowed_origins" yaml:"cors_allowed_origins"`
 
-	Relayer               bool   `json:"relayer" yaml:"relayer"`
-	NumBlockConfirmations uint64 `json:"num_block_confirmations" yaml:"num_block_confirmations"`
+	Relayer bool `json:"relayer" yaml:"relayer"`
 
 	ConcurrentRequestsDebug uint64 `json:"concurrent_requests_debug" yaml:"concurrent_requests_debug"`
 	WebSocketReadLimit      uint64 `json:"web_socket_read_limit" yaml:"web_socket_read_limit"`
 
 	MetricsInterval time.Duration `json:"metrics_interval" yaml:"metrics_interval"`
+
+	// event tracker
+	NumBlockConfirmations    uint64 `json:"num_block_confirmations" yaml:"num_block_confirmations"`
+	TrackerSyncBatchSize     uint64 `json:"tracker_sync_batch_size" yaml:"tracker_sync_batch_size"`
+	TrackerBlocksToReconcile uint64 `json:"tracker_blocks_to_reconcile" yaml:"tracker_blocks_to_reconcile"`
 }
 
 // Telemetry holds the config details for metric services.
@@ -81,10 +85,6 @@ const (
 	// requests with fromBlock/toBlock values (e.g. eth_getLogs)
 	DefaultJSONRPCBlockRangeLimit uint64 = 1000
 
-	// DefaultNumBlockConfirmations minimal number of child blocks required for the parent block to be considered final
-	// on ethereum epoch lasts for 32 blocks. more details: https://www.alchemy.com/overviews/ethereum-commitment-levels
-	DefaultNumBlockConfirmations uint64 = 64
-
 	// DefaultConcurrentRequestsDebug specifies max number of allowed concurrent requests for debug endpoints
 	DefaultConcurrentRequestsDebug uint64 = 32
 
@@ -96,6 +96,25 @@ const (
 	// DefaultMetricsInterval specifies the time interval after which Prometheus metrics will be generated.
 	// A value of 0 means the metrics are disabled.
 	DefaultMetricsInterval time.Duration = time.Second * 8
+
+	// event tracker
+
+	// DefaultNumBlockConfirmations minimal number of child blocks required for the parent block to be considered final
+	// on ethereum epoch lasts for 32 blocks. more details: https://www.alchemy.com/overviews/ethereum-commitment-levels
+	DefaultNumBlockConfirmations uint64 = 64
+
+	// DefaultTrackerSyncBatchSize defines a default batch size of blocks that will be gotten from tracked chain,
+	// when tracker is out of sync and needs to sync a number of blocks.
+	DefaultTrackerSyncBatchSize uint64 = 10
+
+	// DefaultTrackerBlocksToReconcile defines how default number blocks that tracker
+	// will sync up from the latest block on tracked chain.
+	// If a node that has tracker, was offline for days, months, a year, it will miss a lot of blocks.
+	// In the meantime, we expect the rest of nodes to have collected the desired events and did their
+	// logic with them, continuing consensus and relayer stuff.
+	// In order to not waste too much unnecessary time in syncing all those blocks, with NumOfBlocksToReconcile,
+	// we tell the tracker to sync only latestBlock.Number - NumOfBlocksToReconcile number of blocks.
+	DefaultTrackerBlocksToReconcile uint64 = 10000
 )
 
 // DefaultConfig returns the default server configuration
@@ -131,10 +150,13 @@ func DefaultConfig() *Config {
 		JSONRPCBatchRequestLimit: DefaultJSONRPCBatchRequestLimit,
 		JSONRPCBlockRangeLimit:   DefaultJSONRPCBlockRangeLimit,
 		Relayer:                  false,
-		NumBlockConfirmations:    DefaultNumBlockConfirmations,
 		ConcurrentRequestsDebug:  DefaultConcurrentRequestsDebug,
 		WebSocketReadLimit:       DefaultWebSocketReadLimit,
 		MetricsInterval:          DefaultMetricsInterval,
+		// event tracker
+		NumBlockConfirmations:    DefaultNumBlockConfirmations,
+		TrackerSyncBatchSize:     DefaultTrackerSyncBatchSize,
+		TrackerBlocksToReconcile: DefaultTrackerBlocksToReconcile,
 	}
 }
 

command/server/params.go

@@ -37,13 +37,17 @@ const (
 	corsOriginFlag               = "access-control-allow-origins"
 	logFileLocationFlag          = "log-to"
 
-	relayerFlag               = "relayer"
-	numBlockConfirmationsFlag = "num-block-confirmations"
+	relayerFlag = "relayer"
 
 	concurrentRequestsDebugFlag = "concurrent-requests-debug"
 	webSocketReadLimitFlag      = "websocket-read-limit"
 
 	metricsIntervalFlag = "metrics-interval"
+
+	// event tracker
+	numBlockConfirmationsFlag    = "num-block-confirmations"
+	trackerSyncBatchSizeFlag     = "tracker-sync-batch-size"
+	trackerBlocksToReconcileFlag = "tracker-blocks-to-reconcile"
 )
 
 // Flags that are deprecated, but need to be preserved for
@@ -185,8 +189,11 @@ func (p *serverParams) generateConfig() *server.Config {
 		JSONLogFormat:      p.rawConfig.JSONLogFormat,
 		LogFilePath:        p.logFileLocation,
 
-		Relayer:               p.relayer,
-		NumBlockConfirmations: p.rawConfig.NumBlockConfirmations,
-		MetricsInterval:       p.rawConfig.MetricsInterval,
+		Relayer:         p.relayer,
+		MetricsInterval: p.rawConfig.MetricsInterval,
+		// event tracker
+		NumBlockConfirmations:    p.rawConfig.NumBlockConfirmations,
+		TrackerSyncBatchSize:     p.rawConfig.TrackerSyncBatchSize,
+		TrackerBlocksToReconcile: p.rawConfig.TrackerBlocksToReconcile,
 	}
 }

command/server/server.go

@@ -214,13 +214,6 @@ func setFlags(cmd *cobra.Command) {
 		"start the state sync relayer service (PolyBFT only)",
 	)
 
-	cmd.Flags().Uint64Var(
-		&params.rawConfig.NumBlockConfirmations,
-		numBlockConfirmationsFlag,
-		defaultConfig.NumBlockConfirmations,
-		"minimal number of child blocks required for the parent block to be considered final",
-	)
-
 	cmd.Flags().Uint64Var(
 		&params.rawConfig.ConcurrentRequestsDebug,
 		concurrentRequestsDebugFlag,
@@ -242,6 +235,38 @@ func setFlags(cmd *cobra.Command) {
 		"the interval (in seconds) at which special metrics are generated. a value of zero means the metrics are disabled",
 	)
 
+	// event tracker config
+	cmd.Flags().Uint64Var(
+		&params.rawConfig.NumBlockConfirmations,
+		numBlockConfirmationsFlag,
+		defaultConfig.NumBlockConfirmations,
+		"minimal number of child blocks required for the parent block to be considered final",
+	)
+
+	cmd.Flags().Uint64Var(
+		&params.rawConfig.TrackerSyncBatchSize,
+		trackerSyncBatchSizeFlag,
+		defaultConfig.TrackerSyncBatchSize,
+		`defines a batch size of blocks that will be gotten from tracked chain,
+		when tracker is out of sync and needs to sync a number of blocks.
+		(e.g., SyncBatchSize = 10, trackers last processed block is 10, latest block on tracked chain is 100,
+		it will get blocks 11-20, get logs from confirmed blocks of given batch, remove processed confirm logs
+		from memory, and continue to the next batch)`,
+	)
+
+	cmd.Flags().Uint64Var(
+		&params.rawConfig.TrackerBlocksToReconcile,
+		trackerBlocksToReconcileFlag,
+		defaultConfig.TrackerBlocksToReconcile,
+		`defines how many blocks we will sync up from the latest block on tracked chain.
+		If a node that has tracker, was offline for days, months, a year, it will miss a lot of blocks.
+		In the meantime, we expect the rest of nodes to have collected the desired events and did their
+		logic with them, continuing consensus and relayer stuff.
+		In order to not waste too much unnecessary time in syncing all those blocks, with NumOfBlocksToReconcile,
+		we tell the tracker to sync only latestBlock.Number - NumOfBlocksToReconcile number of blocks.
+		If 0 is set to this flag, event tracker will sync all the blocks from tracked chain`,
+	)
+
 	setLegacyFlags(cmd)
 
 	setDevFlags(cmd)

consensus/consensus.go

@@ -84,8 +84,11 @@ type Params struct {
 	SecretsManager secrets.SecretsManager
 	BlockTime      uint64
 
-	NumBlockConfirmations uint64
-	MetricsInterval       time.Duration
+	MetricsInterval time.Duration
+	// event tracker
+	NumBlockConfirmations    uint64
+	TrackerSyncBatchSize     uint64
+	TrackerBlocksToReconcile uint64
 }
 
 // Factory is the factory function to create a discovery consensus

consensus/polybft/consensus_runtime.go

@@ -74,16 +74,19 @@ type guardedDataDTO struct {
 
 // runtimeConfig is a struct that holds configuration data for given consensus runtime
 type runtimeConfig struct {
-	PolyBFTConfig         *PolyBFTConfig
-	DataDir               string
-	Key                   *wallet.Key
-	State                 *State
-	blockchain            blockchainBackend
-	polybftBackend        polybftBackend
-	txPool                txPoolInterface
-	bridgeTopic           topic
-	numBlockConfirmations uint64
-	consensusConfig       *consensus.Config
+	PolyBFTConfig   *PolyBFTConfig
+	DataDir         string
+	Key             *wallet.Key
+	State           *State
+	blockchain      blockchainBackend
+	polybftBackend  polybftBackend
+	txPool          txPoolInterface
+	bridgeTopic     topic
+	consensusConfig *consensus.Config
+	// event tracker
+	numBlockConfirmations    uint64
+	trackerSyncBatchSize     uint64
+	trackerBlocksToReconcile uint64
 }
 
 // consensusRuntime is a struct that provides consensus runtime features like epoch, state and event management
@@ -197,15 +200,18 @@ func (c *consensusRuntime) initStateSyncManager(logger hcf.Logger) error {
 			logger.Named("state-sync-manager"),
 			c.config.State,
 			&stateSyncConfig{
-				key:                      c.config.Key,
-				stateSenderAddr:          stateSenderAddr,
-				stateSenderStartBlock:    c.config.PolyBFTConfig.Bridge.EventTrackerStartBlocks[stateSenderAddr],
-				jsonrpcAddr:              c.config.PolyBFTConfig.Bridge.JSONRPCEndpoint,
-				dataDir:                  c.config.DataDir,
-				topic:                    c.config.bridgeTopic,
-				maxCommitmentSize:        maxCommitmentSize,
+				key:                   c.config.Key,
+				stateSenderAddr:       stateSenderAddr,
+				stateSenderStartBlock: c.config.PolyBFTConfig.Bridge.EventTrackerStartBlocks[stateSenderAddr],
+				jsonrpcAddr:           c.config.PolyBFTConfig.Bridge.JSONRPCEndpoint,
+				dataDir:               c.config.DataDir,
+				topic:                 c.config.bridgeTopic,
+				maxCommitmentSize:     maxCommitmentSize,
+				// event tracker
 				numBlockConfirmations:    c.config.numBlockConfirmations,
 				blockTrackerPollInterval: c.config.PolyBFTConfig.BlockTrackerPollInterval.Duration,
+				trackerSyncBatchSize:     c.config.trackerSyncBatchSize,
+				trackerBlocksToReconcile: c.config.trackerBlocksToReconcile,
 			},
 			c,
 		)

consensus/polybft/eventtracker/event_tracker.go

@@ -87,7 +87,7 @@ type EventTracker struct {
 //
 // Example Usage:
 //
-//	config := &EventTracker{
+//	config := &EventTrackerConfig{
 //		RpcEndpoint:           "http://some-json-rpc-url.com",
 //		StartBlockFromConfig:  100_000,
 //		NumBlockConfirmations: 10,
@@ -110,32 +110,36 @@ type EventTracker struct {
 //
 // Outputs:
 //   - A new instance of the EventTracker struct.
-func NewEventTracker(config *EventTrackerConfig) (*EventTracker, error) {
+func NewEventTracker(config *EventTrackerConfig, startBlockFromGenesis uint64) (*EventTracker, error) {
 	lastProcessedBlock, err := config.Store.GetLastProcessedBlock()
 	if err != nil {
 		return nil, err
 	}
 
-	definiteLastProcessedBlock := lastProcessedBlock
+	if lastProcessedBlock == 0 {
+		lastProcessedBlock = startBlockFromGenesis
 
-	if lastProcessedBlock == 0 && config.NumOfBlocksToReconcile > 0 {
-		latestBlock, err := config.BlockProvider.GetBlockByNumber(ethgo.Latest, false)
-		if err != nil {
-			return nil, err
-		}
+		if config.NumOfBlocksToReconcile > 0 {
+			latestBlock, err := config.BlockProvider.GetBlockByNumber(ethgo.Latest, false)
+			if err != nil {
+				return nil, err
+			}
 
-		if latestBlock.Number > config.NumOfBlocksToReconcile {
-			// if this is a fresh start, then we should start syncing from
-			// latestBlock.Number - NumOfBlocksToReconcile
-			definiteLastProcessedBlock = latestBlock.Number - config.NumOfBlocksToReconcile
+			if latestBlock.Number > config.NumOfBlocksToReconcile &&
+				startBlockFromGenesis < latestBlock.Number-config.NumOfBlocksToReconcile {
+				// if this is a fresh start, and we missed too much blocks,
+				// then we should start syncing from
+				// latestBlock.Number - NumOfBlocksToReconcile
+				lastProcessedBlock = latestBlock.Number - config.NumOfBlocksToReconcile
+			}
 		}
 	}
 
 	return &EventTracker{
 		config:         config,
 		closeCh:        make(chan struct{}),
 		blockTracker:   blocktracker.NewJSONBlockTracker(config.BlockProvider),
-		blockContainer: NewTrackerBlockContainer(definiteLastProcessedBlock),
+		blockContainer: NewTrackerBlockContainer(lastProcessedBlock),
 	}, nil
 }
 
@@ -175,6 +179,7 @@ func (e *EventTracker) Start() error {
 		"syncBatchSize", e.config.SyncBatchSize,
 		"numOfBlocksToReconcile", e.config.NumOfBlocksToReconcile,
 		"logFilter", e.config.LogFilter,
+		"lastBlockProcessed", e.blockContainer.LastProcessedBlock(),
 	)
 
 	ctx, cancelFn := context.WithCancel(context.Background())

consensus/polybft/eventtracker/event_tracker_test.go

@@ -89,7 +89,7 @@ func TestEventTracker_TrackBlock(t *testing.T) {
 	t.Run("Add block by block - no confirmed blocks", func(t *testing.T) {
 		t.Parallel()
 
-		tracker, err := NewEventTracker(createTestTrackerConfig(t, 10, 10, 0))
+		tracker, err := NewEventTracker(createTestTrackerConfig(t, 10, 10, 0), 0)
 
 		require.NoError(t, err)
 
@@ -131,7 +131,7 @@ func TestEventTracker_TrackBlock(t *testing.T) {
 		blockProviderMock := new(mockProvider)
 		blockProviderMock.On("GetLogs", mock.Anything).Return([]*ethgo.Log{}, nil).Once()
 
-		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 0))
+		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 0), 0)
 		require.NoError(t, err)
 
 		tracker.config.BlockProvider = blockProviderMock
@@ -197,7 +197,7 @@ func TestEventTracker_TrackBlock(t *testing.T) {
 		blockProviderMock := new(mockProvider)
 		blockProviderMock.On("GetLogs", mock.Anything).Return(logs, nil).Once()
 
-		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 0))
+		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 0), 0)
 		require.NoError(t, err)
 
 		tracker.config.BlockProvider = blockProviderMock
@@ -265,7 +265,7 @@ func TestEventTracker_TrackBlock(t *testing.T) {
 		blockProviderMock := new(mockProvider)
 		blockProviderMock.On("GetLogs", mock.Anything).Return(nil, errors.New("some error occurred")).Once()
 
-		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 0))
+		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 0), 0)
 		require.NoError(t, err)
 
 		tracker.config.BlockProvider = blockProviderMock
@@ -335,7 +335,7 @@ func TestEventTracker_TrackBlock(t *testing.T) {
 		// just mock the call, it will use the provider.blocks map to handle proper returns
 		blockProviderMock.On("GetBlockByNumber", mock.Anything, mock.Anything).Return(nil, nil).Times(int(numOfMissedBlocks))
 
-		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 0))
+		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 0), 0)
 		require.NoError(t, err)
 
 		tracker.config.BlockProvider = blockProviderMock
@@ -420,7 +420,7 @@ func TestEventTracker_TrackBlock(t *testing.T) {
 		// just mock the call, it will use the provider.blocks map to handle proper returns
 		blockProviderMock.On("GetBlockByNumber", mock.Anything, mock.Anything).Return(nil, nil).Times(int(numOfMissedBlocks + numOfCachedBlocks))
 
-		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 0))
+		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 0), 0)
 		require.NoError(t, err)
 
 		tracker.config.BlockProvider = blockProviderMock
@@ -515,7 +515,7 @@ func TestEventTracker_TrackBlock(t *testing.T) {
 		// just mock the call, it will use the provider.blocks map to handle proper returns
 		blockProviderMock.On("GetBlockByNumber", mock.Anything, mock.Anything).Return(nil, nil).Times(int(numOfCachedBlocks))
 
-		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 0))
+		tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 0), 0)
 		require.NoError(t, err)
 
 		tracker.config.BlockProvider = blockProviderMock

consensus/polybft/polybft.go

@@ -543,16 +543,19 @@ func (p *Polybft) Start() error {
 // initRuntime creates consensus runtime
 func (p *Polybft) initRuntime() error {
 	runtimeConfig := &runtimeConfig{
-		PolyBFTConfig:         p.consensusConfig,
-		Key:                   p.key,
-		DataDir:               p.dataDir,
-		State:                 p.state,
-		blockchain:            p.blockchain,
-		polybftBackend:        p,
-		txPool:                p.txPool,
-		bridgeTopic:           p.bridgeTopic,
-		numBlockConfirmations: p.config.NumBlockConfirmations,
-		consensusConfig:       p.config.Config,
+		PolyBFTConfig:   p.consensusConfig,
+		Key:             p.key,
+		DataDir:         p.dataDir,
+		State:           p.state,
+		blockchain:      p.blockchain,
+		polybftBackend:  p,
+		txPool:          p.txPool,
+		bridgeTopic:     p.bridgeTopic,
+		consensusConfig: p.config.Config,
+		// event tracker
+		numBlockConfirmations:    p.config.NumBlockConfirmations,
+		trackerSyncBatchSize:     p.config.TrackerSyncBatchSize,
+		trackerBlocksToReconcile: p.config.TrackerBlocksToReconcile,
 	}
 
 	runtime, err := newConsensusRuntime(p.logger, runtimeConfig)