poll new blocks in the committer when in live mode (#226)

### TL;DR

Added support for real-time block committing in addition to backfill mode.

### What changed?

- Added a new `getBlockToCommitUntil` function that determines the end block for committing based on work mode
- In real-time mode, the committer now checks the latest block from RPC and uses that as the upper bound if it's less than the calculated end block
- Created a `fetchBlockDataToCommit` function that handles different data retrieval strategies based on work mode:
  - In backfill mode: retrieves data from staging storage
  - In real-time mode: uses a `BoundlessPoller` to fetch data directly without saving to staging first
- Updated the `getBlockNumbersToCommit` function to accept a context parameter
- Refactored `getSequentialBlockDataToCommit` to use the new functions

### How to test?

1. Run the application in real-time mode and verify it correctly commits blocks up to the latest block from RPC
2. Run the application in backfill mode and verify it commits blocks in batches of `blocksPerCommit`
3. Verify that when switching between modes, the committer correctly adjusts its behavior

### Why make this change?

This change enables the committer to operate in two distinct modes:
1. Backfill mode: for historical data processing where blocks are first staged then committed
2. Real-time mode: for processing current blocks directly from the RPC without staging

This dual-mode approach improves efficiency by allowing real-time processing to skip the staging step when immediate data is needed, while maintaining the ability to process historical data in batches.

<!-- This is an auto-generated comment: release notes by coderabbit.ai -->

## Summary by CodeRabbit

- **New Features**
	- Improved block data fetching logic to dynamically adjust commit ranges based on current mode and blockchain state.
- **Bug Fixes**
	- Updated gap handling to skip unnecessary operations in live mode, reducing redundant processing.
- **Tests**
	- Enhanced test coverage to reflect new context-based method signatures and work mode handling.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: iuwqyir

internal/orchestrator/committer.go

@@ -105,7 +105,7 @@ func (c *Committer) Start(ctx context.Context) {
 	}
 }
 
-func (c *Committer) getBlockNumbersToCommit() ([]*big.Int, error) {
+func (c *Committer) getBlockNumbersToCommit(ctx context.Context) ([]*big.Int, error) {
 	startTime := time.Now()
 	defer func() {
 		log.Debug().Str("metric", "get_block_numbers_to_commit_duration").Msgf("getBlockNumbersToCommit duration: %f", time.Since(startTime).Seconds())
@@ -129,7 +129,10 @@ func (c *Committer) getBlockNumbersToCommit() ([]*big.Int, error) {
 	}
 
 	startBlock := new(big.Int).Add(latestCommittedBlockNumber, big.NewInt(1))
-	endBlock := new(big.Int).Add(latestCommittedBlockNumber, big.NewInt(int64(c.blocksPerCommit)))
+	endBlock, err := c.getBlockToCommitUntil(ctx, latestCommittedBlockNumber)
+	if err != nil {
+		return nil, fmt.Errorf("error getting block to commit until: %v", err)
+	}
 
 	blockCount := new(big.Int).Sub(endBlock, startBlock).Int64() + 1
 	blockNumbers := make([]*big.Int, blockCount)
@@ -140,26 +143,64 @@ func (c *Committer) getBlockNumbersToCommit() ([]*big.Int, error) {
 	return blockNumbers, nil
 }
 
+func (c *Committer) getBlockToCommitUntil(ctx context.Context, latestCommittedBlockNumber *big.Int) (*big.Int, error) {
+	untilBlock := new(big.Int).Add(latestCommittedBlockNumber, big.NewInt(int64(c.blocksPerCommit)))
+	if c.workMode == WorkModeBackfill {
+		return untilBlock, nil
+	} else {
+		// get latest block from RPC and if that's less than until block, return that
+		latestBlock, err := c.rpc.GetLatestBlockNumber(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("error getting latest block from RPC: %v", err)
+		}
+		if latestBlock.Cmp(untilBlock) < 0 {
+			log.Debug().Msgf("Committing until latest block: %s", latestBlock.String())
+			return latestBlock, nil
+		}
+		return untilBlock, nil
+	}
+}
+
+func (c *Committer) fetchBlockDataToCommit(ctx context.Context, blockNumbers []*big.Int) ([]common.BlockData, error) {
+	if c.workMode == WorkModeBackfill {
+		startTime := time.Now()
+		blocksData, err := c.storage.StagingStorage.GetStagingData(storage.QueryFilter{BlockNumbers: blockNumbers, ChainId: c.rpc.GetChainID()})
+		log.Debug().Str("metric", "get_staging_data_duration").Msgf("StagingStorage.GetStagingData duration: %f", time.Since(startTime).Seconds())
+		metrics.GetStagingDataDuration.Observe(time.Since(startTime).Seconds())
+
+		if err != nil {
+			return nil, fmt.Errorf("error fetching blocks to commit: %v", err)
+		}
+		if len(blocksData) == 0 {
+			log.Warn().Msgf("Committer didn't find the following range in staging: %v - %v", blockNumbers[0].Int64(), blockNumbers[len(blockNumbers)-1].Int64())
+			c.handleMissingStagingData(ctx, blockNumbers)
+			return nil, nil
+		}
+		return blocksData, nil
+	} else {
+		poller := NewBoundlessPoller(c.rpc, c.storage)
+		blocksData, err := poller.PollWithoutSaving(ctx, blockNumbers)
+		if err != nil {
+			return nil, fmt.Errorf("poller error: %v", err)
+		}
+		return blocksData, nil
+	}
+}
+
 func (c *Committer) getSequentialBlockDataToCommit(ctx context.Context) ([]common.BlockData, error) {
-	blocksToCommit, err := c.getBlockNumbersToCommit()
+	blocksToCommit, err := c.getBlockNumbersToCommit(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("error determining blocks to commit: %v", err)
 	}
 	if len(blocksToCommit) == 0 {
 		return nil, nil
 	}
 
-	startTime := time.Now()
-	blocksData, err := c.storage.StagingStorage.GetStagingData(storage.QueryFilter{BlockNumbers: blocksToCommit, ChainId: c.rpc.GetChainID()})
-	log.Debug().Str("metric", "get_staging_data_duration").Msgf("StagingStorage.GetStagingData duration: %f", time.Since(startTime).Seconds())
-	metrics.GetStagingDataDuration.Observe(time.Since(startTime).Seconds())
-
+	blocksData, err := c.fetchBlockDataToCommit(ctx, blocksToCommit)
 	if err != nil {
-		return nil, fmt.Errorf("error fetching blocks to commit: %v", err)
+		return nil, err
 	}
 	if len(blocksData) == 0 {
-		log.Warn().Msgf("Committer didn't find the following range in staging: %v - %v", blocksToCommit[0].Int64(), blocksToCommit[len(blocksToCommit)-1].Int64())
-		c.handleMissingStagingData(ctx, blocksToCommit)
 		return nil, nil
 	}
 
@@ -250,6 +291,11 @@ func (c *Committer) handleGap(ctx context.Context, expectedStartBlockNumber *big
 	// record the first missed block number in prometheus
 	metrics.MissedBlockNumbers.Set(float64(expectedStartBlockNumber.Int64()))
 
+	if c.workMode == WorkModeLive {
+		log.Debug().Msgf("Skipping gap handling in live mode. Expected block %s, actual first block %s", expectedStartBlockNumber.String(), actualFirstBlock.Number.String())
+		return nil
+	}
+
 	poller := NewBoundlessPoller(c.rpc, c.storage)
 
 	missingBlockCount := new(big.Int).Sub(actualFirstBlock.Number, expectedStartBlockNumber).Int64()

internal/orchestrator/committer_test.go

@@ -25,6 +25,7 @@ func TestNewCommitter(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 
 	assert.NotNil(t, committer)
 	assert.Equal(t, DEFAULT_COMMITTER_TRIGGER_INTERVAL, committer.triggerIntervalMs)
@@ -40,12 +41,13 @@ func TestGetBlockNumbersToCommit(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(100), nil)
 
-	blockNumbers, err := committer.getBlockNumbersToCommit()
+	blockNumbers, err := committer.getBlockNumbersToCommit(context.Background())
 
 	assert.NoError(t, err)
 	assert.Equal(t, committer.blocksPerCommit, len(blockNumbers))
@@ -63,12 +65,13 @@ func TestGetBlockNumbersToCommitWithoutConfiguredAndNotStored(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(0), nil)
 
-	blockNumbers, err := committer.getBlockNumbersToCommit()
+	blockNumbers, err := committer.getBlockNumbersToCommit(context.Background())
 
 	assert.NoError(t, err)
 	assert.Equal(t, committer.blocksPerCommit, len(blockNumbers))
@@ -89,12 +92,13 @@ func TestGetBlockNumbersToCommitWithConfiguredAndNotStored(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(0), nil)
 
-	blockNumbers, err := committer.getBlockNumbersToCommit()
+	blockNumbers, err := committer.getBlockNumbersToCommit(context.Background())
 
 	assert.NoError(t, err)
 	assert.Equal(t, committer.blocksPerCommit, len(blockNumbers))
@@ -115,12 +119,13 @@ func TestGetBlockNumbersToCommitWithConfiguredAndStored(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(2000), nil)
 
-	blockNumbers, err := committer.getBlockNumbersToCommit()
+	blockNumbers, err := committer.getBlockNumbersToCommit(context.Background())
 
 	assert.NoError(t, err)
 	assert.Equal(t, committer.blocksPerCommit, len(blockNumbers))
@@ -138,12 +143,13 @@ func TestGetBlockNumbersToCommitWithoutConfiguredAndStored(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(2000), nil)
 
-	blockNumbers, err := committer.getBlockNumbersToCommit()
+	blockNumbers, err := committer.getBlockNumbersToCommit(context.Background())
 
 	assert.NoError(t, err)
 	assert.Equal(t, committer.blocksPerCommit, len(blockNumbers))
@@ -164,12 +170,13 @@ func TestGetBlockNumbersToCommitWithStoredHigherThanInMemory(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(2000), nil)
 
-	blockNumbers, err := committer.getBlockNumbersToCommit()
+	blockNumbers, err := committer.getBlockNumbersToCommit(context.Background())
 
 	assert.NoError(t, err)
 	assert.Equal(t, committer.blocksPerCommit, len(blockNumbers))
@@ -190,12 +197,13 @@ func TestGetBlockNumbersToCommitWithStoredLowerThanInMemory(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(99), nil)
 
-	blockNumbers, err := committer.getBlockNumbersToCommit()
+	blockNumbers, err := committer.getBlockNumbersToCommit(context.Background())
 
 	assert.NoError(t, err)
 	assert.Equal(t, 0, len(blockNumbers))
@@ -214,12 +222,13 @@ func TestGetBlockNumbersToCommitWithStoredEqualThanInMemory(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(2000), nil)
 
-	blockNumbers, err := committer.getBlockNumbersToCommit()
+	blockNumbers, err := committer.getBlockNumbersToCommit(context.Background())
 
 	assert.NoError(t, err)
 	assert.Equal(t, committer.blocksPerCommit, len(blockNumbers))
@@ -239,6 +248,7 @@ func TestGetSequentialBlockDataToCommit(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
@@ -273,6 +283,7 @@ func TestGetSequentialBlockDataToCommitWithDuplicateBlocks(t *testing.T) {
 		StagingStorage: mockStagingStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 	chainID := big.NewInt(1)
 
 	mockRPC.EXPECT().GetChainID().Return(chainID)
@@ -311,6 +322,7 @@ func TestCommit(t *testing.T) {
 		OrchestratorStorage: mockOrchestratorStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 
 	blockData := []common.BlockData{
 		{Block: common.Block{Number: big.NewInt(101)}},
@@ -336,6 +348,7 @@ func TestHandleGap(t *testing.T) {
 		OrchestratorStorage: mockOrchestratorStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 
 	expectedStartBlockNumber := big.NewInt(100)
 	actualFirstBlock := common.Block{Number: big.NewInt(105)}
@@ -459,6 +472,7 @@ func TestHandleMissingStagingData(t *testing.T) {
 	}
 
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 
 	chainID := big.NewInt(1)
 	mockRPC.EXPECT().GetChainID().Return(chainID)
@@ -505,6 +519,7 @@ func TestHandleMissingStagingDataIsPolledWithCorrectBatchSize(t *testing.T) {
 	}
 
 	committer := NewCommitter(mockRPC, mockStorage)
+	committer.workMode = WorkModeBackfill
 
 	chainID := big.NewInt(1)
 	mockRPC.EXPECT().GetChainID().Return(chainID)