diff --git a/consensus/polybft/eventtracker/event_tracker.go b/consensus/polybft/eventtracker/event_tracker.go index f4877b1d35..9b619eed89 100644 --- a/consensus/polybft/eventtracker/event_tracker.go +++ b/consensus/polybft/eventtracker/event_tracker.go @@ -29,12 +29,6 @@ type EventTrackerConfig struct { // RPCEndpoint is the full json rpc url on some node on a tracked chain RPCEndpoint string - // StartBlockFromConfig represents a starting block from which tracker starts to - // track events from a tracked chain. - // This is only relevant on the first start of the tracker. After it processes blocks, - // it will start from the last processed block, and not from the StartBlockFromConfig. - StartBlockFromConfig uint64 - // NumBlockConfirmations defines how many blocks must pass from a certain block, // to consider that block as final on the tracked chain. // This is very important for reorgs, and events from the given block will only be @@ -50,13 +44,13 @@ type EventTrackerConfig struct { // from memory, and continue to the next batch) SyncBatchSize uint64 - // MaxBacklogSize defines how many blocks we will sync up from the latest block on tracked chain. + // NumOfBlocksToReconcile 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 MaxBacklogSize, - // we tell the tracker to sync only latestBlock.Number - MaxBacklogSize number of blocks. - MaxBacklogSize uint64 + // 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. + NumOfBlocksToReconcile uint64 // PollInterval defines a time interval in which tracker polls json rpc node // for latest block on the tracked chain. @@ -122,13 +116,19 @@ func NewEventTracker(config *EventTrackerConfig) (*EventTracker, error) { return nil, err } - var definiteLastProcessedBlock uint64 - if config.StartBlockFromConfig > 0 { - definiteLastProcessedBlock = config.StartBlockFromConfig - 1 - } + definiteLastProcessedBlock := lastProcessedBlock + + if lastProcessedBlock == 0 && config.NumOfBlocksToReconcile > 0 { + latestBlock, err := config.BlockProvider.GetBlockByNumber(ethgo.Latest, false) + if err != nil { + return nil, err + } - if lastProcessedBlock > definiteLastProcessedBlock { - definiteLastProcessedBlock = lastProcessedBlock + 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 + } } return &EventTracker{ @@ -170,11 +170,10 @@ func (e *EventTracker) Close() { func (e *EventTracker) Start() error { e.config.Logger.Info("Starting event tracker", "jsonRpcEndpoint", e.config.RPCEndpoint, - "startBlockFromConfig", e.config.StartBlockFromConfig, "numBlockConfirmations", e.config.NumBlockConfirmations, "pollInterval", e.config.PollInterval, "syncBatchSize", e.config.SyncBatchSize, - "maxBacklogSize", e.config.MaxBacklogSize, + "numOfBlocksToReconcile", e.config.NumOfBlocksToReconcile, "logFilter", e.config.LogFilter, ) @@ -294,9 +293,10 @@ func (e *EventTracker) getNewState(latestBlock *ethgo.Block) error { startBlock := lastProcessedBlock + 1 // sanitize startBlock from which we will start polling for blocks - if latestBlock.Number > e.config.MaxBacklogSize && - latestBlock.Number-e.config.MaxBacklogSize > lastProcessedBlock { - startBlock = latestBlock.Number - e.config.MaxBacklogSize + if e.config.NumOfBlocksToReconcile > 0 && + latestBlock.Number > e.config.NumOfBlocksToReconcile && + latestBlock.Number-e.config.NumOfBlocksToReconcile > lastProcessedBlock { + startBlock = latestBlock.Number - e.config.NumOfBlocksToReconcile } // get blocks in batches diff --git a/consensus/polybft/eventtracker/event_tracker_test.go b/consensus/polybft/eventtracker/event_tracker_test.go index 987d33f567..be7b0d1dd8 100644 --- a/consensus/polybft/eventtracker/event_tracker_test.go +++ b/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, 1000)) + tracker, err := NewEventTracker(createTestTrackerConfig(t, 10, 10, 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, 1000)) + tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 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, 1000)) + tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 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, 1000)) + tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, 10, 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, 1000)) + tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 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, 1000)) + tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 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, 1000)) + tracker, err := NewEventTracker(createTestTrackerConfig(t, numBlockConfirmations, batchSize, 0)) require.NoError(t, err) tracker.config.BlockProvider = blockProviderMock @@ -582,19 +582,19 @@ func TestEventTracker_TrackBlock(t *testing.T) { }) } -func createTestTrackerConfig(t *testing.T, numBlockConfirmations, batchSize, maxBacklogSize uint64) *EventTrackerConfig { +func createTestTrackerConfig(t *testing.T, numBlockConfirmations, batchSize, + numOfBlocksToReconcile uint64) *EventTrackerConfig { t.Helper() var stateSyncEvent contractsapi.StateSyncedEvent return &EventTrackerConfig{ - RPCEndpoint: "http://some-rpc-url.com", - StartBlockFromConfig: 0, - NumBlockConfirmations: numBlockConfirmations, - SyncBatchSize: batchSize, - MaxBacklogSize: maxBacklogSize, - PollInterval: 2 * time.Second, - Logger: hclog.NewNullLogger(), + RPCEndpoint: "http://some-rpc-url.com", + NumBlockConfirmations: numBlockConfirmations, + SyncBatchSize: batchSize, + NumOfBlocksToReconcile: numOfBlocksToReconcile, + PollInterval: 2 * time.Second, + Logger: hclog.NewNullLogger(), LogFilter: map[ethgo.Address][]ethgo.Hash{ ethgo.ZeroAddress: {stateSyncEvent.Sig()}, }, diff --git a/consensus/polybft/state_sync_manager.go b/consensus/polybft/state_sync_manager.go index 1b6ed88ad5..1772da08a9 100644 --- a/consensus/polybft/state_sync_manager.go +++ b/consensus/polybft/state_sync_manager.go @@ -145,16 +145,15 @@ func (s *stateSyncManager) setupNewTracker() error { var stateSyncEvent contractsapi.StateSyncedEvent tracker, err := eventtracker.NewEventTracker(&eventtracker.EventTrackerConfig{ - RPCEndpoint: s.config.jsonrpcAddr, - StartBlockFromConfig: s.config.stateSenderStartBlock, - NumBlockConfirmations: s.config.numBlockConfirmations, - SyncBatchSize: 5, // this should be configurable - MaxBacklogSize: 10_000, // this should be configurable - PollInterval: s.config.blockTrackerPollInterval, - Logger: s.logger, - Store: store, - EventSubscriber: s, - BlockProvider: clt.Eth(), + RPCEndpoint: s.config.jsonrpcAddr, + NumBlockConfirmations: s.config.numBlockConfirmations, + SyncBatchSize: 5, // this should be configurable + NumOfBlocksToReconcile: 0, // this should be configurable + PollInterval: s.config.blockTrackerPollInterval, + Logger: s.logger, + Store: store, + EventSubscriber: s, + BlockProvider: clt.Eth(), LogFilter: map[ethgo.Address][]ethgo.Hash{ ethgo.Address(s.config.stateSenderAddr): {stateSyncEvent.Sig()}, },