add possibility to poll without saving (#225)

### TL;DR

Refactored the Poller to separate block polling from data storage operations.

### What changed?

- Split the `Poll` method into two separate methods:
  - `PollWithoutSaving`: Handles only the polling of blocks without storing data
  - `StageResults`: Handles the storage of polled block data
- Moved the worker results handling logic into these new methods
- Added functionality to track and return the highest block number polled
- Created a helper method `convertPollResultsToBlockData` to transform poll results into block data

### How to test?

1. Run the existing test suite to ensure all functionality works as expected
2. Test the new `PollWithoutSaving` method by calling it directly and verifying it returns the correct block data without saving
3. Test the `StageResults` method by passing block data and verifying it's properly stored
4. Verify that the `Poll` method still returns the highest block number correctly

### Why make this change?

This refactoring improves the separation of concerns in the Poller, making it more flexible by allowing clients to poll blocks without immediately saving the data. This enables more advanced use cases where data processing might be needed before storage, and makes the code more maintainable by breaking down the monolithic polling function into smaller, more focused components.

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

- **Refactor**
  - Enhanced the polling process to improve data handling and error reporting, resulting in more reliable background operations without affecting user-facing features.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: iuwqyir

internal/orchestrator/poller.go

@@ -165,9 +165,27 @@ func (p *Poller) Start(ctx context.Context) {
 }
 
 func (p *Poller) Poll(ctx context.Context, blockNumbers []*big.Int) (lastPolledBlock *big.Int) {
+	blockData, failedResults := p.PollWithoutSaving(ctx, blockNumbers)
+	if len(blockData) > 0 || len(failedResults) > 0 {
+		p.StageResults(blockData, failedResults)
+	}
+
+	var highestBlockNumber *big.Int
+	if len(blockData) > 0 {
+		highestBlockNumber = blockData[0].Block.Number
+		for _, block := range blockData {
+			if block.Block.Number.Cmp(highestBlockNumber) > 0 {
+				highestBlockNumber = new(big.Int).Set(block.Block.Number)
+			}
+		}
+	}
+	return highestBlockNumber
+}
+
+func (p *Poller) PollWithoutSaving(ctx context.Context, blockNumbers []*big.Int) ([]common.BlockData, []rpc.GetFullBlockResult) {
 	if len(blockNumbers) < 1 {
 		log.Debug().Msg("No blocks to poll, skipping")
-		return
+		return nil, nil
 	}
 	endBlock := blockNumbers[len(blockNumbers)-1]
 	if endBlock != nil {
@@ -180,8 +198,60 @@ func (p *Poller) Poll(ctx context.Context, blockNumbers []*big.Int) (lastPolledB
 
 	worker := worker.NewWorker(p.rpc)
 	results := worker.Run(ctx, blockNumbers)
-	p.handleWorkerResults(results)
-	return endBlock
+	blockData, failedResults := p.convertPollResultsToBlockData(results)
+	return blockData, failedResults
+}
+
+func (p *Poller) convertPollResultsToBlockData(results []rpc.GetFullBlockResult) ([]common.BlockData, []rpc.GetFullBlockResult) {
+	var successfulResults []rpc.GetFullBlockResult
+	var failedResults []rpc.GetFullBlockResult
+
+	for _, result := range results {
+		if result.Error != nil {
+			bn := "<unknown>"
+			if result.BlockNumber != nil {
+				bn = result.BlockNumber.String()
+			}
+			log.Warn().Err(result.Error).Msgf("Error fetching block data for block %s", bn)
+			failedResults = append(failedResults, result)
+		} else {
+			successfulResults = append(successfulResults, result)
+		}
+	}
+
+	blockData := make([]common.BlockData, 0, len(successfulResults))
+	for _, result := range successfulResults {
+		blockData = append(blockData, common.BlockData{
+			Block:        result.Data.Block,
+			Logs:         result.Data.Logs,
+			Transactions: result.Data.Transactions,
+			Traces:       result.Data.Traces,
+		})
+	}
+	return blockData, failedResults
+}
+
+func (p *Poller) StageResults(blockData []common.BlockData, failedResults []rpc.GetFullBlockResult) {
+	startTime := time.Now()
+	metrics.PolledBatchSize.Set(float64(len(blockData)))
+	if len(blockData) > 0 {
+		if err := p.storage.StagingStorage.InsertStagingData(blockData); err != nil {
+			e := fmt.Errorf("error inserting block data: %v", err)
+			log.Error().Err(e)
+			for _, result := range blockData {
+				failedResults = append(failedResults, rpc.GetFullBlockResult{
+					BlockNumber: result.Block.Number,
+					Error:       e,
+				})
+			}
+		}
+	}
+	log.Debug().Str("metric", "staging_insert_duration").Msgf("StagingStorage.InsertStagingData duration: %f", time.Since(startTime).Seconds())
+	metrics.StagingInsertDuration.Observe(time.Since(startTime).Seconds())
+
+	if len(failedResults) > 0 {
+		p.handleBlockFailures(failedResults)
+	}
 }
 
 func (p *Poller) reachedPollLimit(blockNumber *big.Int) bool {
@@ -230,49 +300,6 @@ func (p *Poller) createBlockNumbersForRange(startBlock *big.Int, endBlock *big.I
 	return blockNumbers
 }
 
-func (p *Poller) handleWorkerResults(results []rpc.GetFullBlockResult) {
-	var successfulResults []rpc.GetFullBlockResult
-	var failedResults []rpc.GetFullBlockResult
-
-	for _, result := range results {
-		if result.Error != nil {
-			log.Warn().Err(result.Error).Msgf("Error fetching block data for block %s", result.BlockNumber.String())
-			failedResults = append(failedResults, result)
-		} else {
-			successfulResults = append(successfulResults, result)
-		}
-	}
-
-	blockData := make([]common.BlockData, 0, len(successfulResults))
-	for _, result := range successfulResults {
-		blockData = append(blockData, common.BlockData{
-			Block:        result.Data.Block,
-			Logs:         result.Data.Logs,
-			Transactions: result.Data.Transactions,
-			Traces:       result.Data.Traces,
-		})
-	}
-
-	startTime := time.Now()
-	if err := p.storage.StagingStorage.InsertStagingData(blockData); err != nil {
-		e := fmt.Errorf("error inserting block data: %v", err)
-		log.Error().Err(e)
-		for _, result := range successfulResults {
-			failedResults = append(failedResults, rpc.GetFullBlockResult{
-				BlockNumber: result.BlockNumber,
-				Error:       e,
-			})
-		}
-		metrics.PolledBatchSize.Set(float64(len(blockData)))
-	}
-	log.Debug().Str("metric", "staging_insert_duration").Msgf("StagingStorage.InsertStagingData duration: %f", time.Since(startTime).Seconds())
-	metrics.StagingInsertDuration.Observe(time.Since(startTime).Seconds())
-
-	if len(failedResults) > 0 {
-		p.handleBlockFailures(failedResults)
-	}
-}
-
 func (p *Poller) handleBlockFailures(results []rpc.GetFullBlockResult) {
 	var blockFailures []common.BlockFailure
 	for _, result := range results {