Fixes

Author: jakeloo

internal/orchestrator/poller.go

@@ -230,9 +230,9 @@ func (p *Poller) Request(ctx context.Context, blockNumbers []*big.Int) []common.
 		return nil
 	}
 
-	p.lastPolledBlockMutex.Lock()
+	p.lastRequestedBlockMutex.Lock()
 	p.lastRequestedBlock = new(big.Int).Set(highestBlockNumber)
-	p.lastPolledBlockMutex.Unlock()
+	p.lastRequestedBlockMutex.Unlock()
 	return blockData
 }
 
@@ -292,14 +292,14 @@ func (p *Poller) reachedPollLimit(blockNumber *big.Int) bool {
 }
 
 func (p *Poller) getNextBlockRange(ctx context.Context) ([]*big.Int, error) {
-	p.blockRangeMutex.Lock()
-	defer p.blockRangeMutex.Unlock()
-
 	latestBlock, err := p.rpc.GetLatestBlockNumber(ctx)
 	if err != nil {
 		return nil, err
 	}
 
+	p.blockRangeMutex.Lock()
+	defer p.blockRangeMutex.Unlock()
+
 	p.lastPendingFetchBlockMutex.Lock()
 	lastPendingFetchBlock := new(big.Int).Set(p.lastPendingFetchBlock)
 	p.lastPendingFetchBlockMutex.Unlock()
@@ -323,16 +323,16 @@ func (p *Poller) getNextBlockRange(ctx context.Context) ([]*big.Int, error) {
 		return nil, nil
 	}
 
+	p.lastPendingFetchBlockMutex.Lock()
+	p.lastPendingFetchBlock = new(big.Int).Set(endBlock)
+	p.lastPendingFetchBlockMutex.Unlock()
+
 	log.Debug().
 		Str("last_pending_block", lastPendingFetchBlock.String()).
 		Str("last_polled_block", lastPolledBlock.String()).
 		Str("last_requested_block", lastRequestedBlock.String()).
 		Msgf("GetNextBlockRange for poller workers")
 
-	p.lastPendingFetchBlockMutex.Lock()
-	p.lastPendingFetchBlock = new(big.Int).Set(endBlock)
-	p.lastPendingFetchBlockMutex.Unlock()
-
 	return p.createBlockNumbersForRange(startBlock, endBlock), nil
 }
 

internal/orchestrator/poller_test.go

@@ -10,4 +10,4 @@ import (
 func TestPollerPlaceholder(t *testing.T) {
 	// Placeholder test to keep the test file valid
 	t.Skip("Poller tests need to be rewritten for new implementation")
-}
+}

internal/orchestrator/reorg_handler_test.go

@@ -10,4 +10,4 @@ import (
 func TestReorgHandlerPlaceholder(t *testing.T) {
 	// Placeholder test to keep the test file valid
 	t.Skip("Reorg handler tests need to be rewritten for new implementation")
-}
+}

internal/storage/badger.go

@@ -206,7 +206,11 @@ func (bc *BadgerConnector) Close() error {
 		bc.gcTicker.Stop()
 		close(bc.stopGC)
 	}
-	close(bc.stopRangeUpdate)
+	select {
+	case <-bc.stopRangeUpdate:
+	default:
+		close(bc.stopRangeUpdate)
+	}
 	return bc.db.Close()
 }
 

internal/worker/worker.go

@@ -26,6 +26,10 @@ const (
 	SourceTypeStaging SourceType = "staging"
 )
 
+const (
+	DEFAULT_RPC_CHUNK_SIZE = 25
+)
+
 // String returns the string representation of the source type
 func (s SourceType) String() string {
 	return string(s)
@@ -41,9 +45,13 @@ type Worker struct {
 }
 
 func NewWorker(rpc rpc.IRPCClient) *Worker {
+	chunk := rpc.GetBlocksPerRequest().Blocks
+	if chunk <= 0 {
+		chunk = DEFAULT_RPC_CHUNK_SIZE
+	}
 	return &Worker{
 		rpc:          rpc,
-		rpcChunkSize: rpc.GetBlocksPerRequest().Blocks,
+		rpcChunkSize: chunk,
 		rpcSemaphore: make(chan struct{}, 20),
 	}
 }
@@ -395,7 +403,7 @@ func (w *Worker) Run(ctx context.Context, blockNumbers []*big.Int) []rpc.GetFull
 	}
 
 	if !success {
-		sourceType := SourceTypeRPC
+		sourceType = SourceTypeRPC
 		results, errors = w.processBatchWithRetry(ctx, blockNumbers, sourceType, w.fetchFromRPC)
 		success = len(results) > 0 && len(errors) == 0
 	}