feat(processor): add batch block fetching for improved throughput (#69)

Implement batch block fetching to request multiple blocks at once using
BatchCallContext RPC, with proper backpressure handling and contiguous-only
batching.

Changes:
- Add BlocksByNumbers to Node/DataSource interfaces for batch RPC
- Implement batch RPC using go-ethereum's BatchCallContext
- Add GetAvailableCapacity and ValidateBatchWithinLeash to Limiter
- Add NextBlocks method to State Manager for batch number generation
- Add InitBlocks batch method to PendingTracker using Redis pipeline
- Refactor ProcessNextBlock in all processors to support batch fetching
- Add exponential backoff with jitter for backpressure handling

The batch size is limited by min(maxPendingBlockRange, availableCapacity)
and stops at the first missing/not-found block to maintain contiguity.

Author: Savid

pkg/ethereum/execution/embedded_node.go

@@ -36,6 +36,10 @@ type DataSource interface {
 	// BlockByNumber returns the block at the given number.
 	BlockByNumber(ctx context.Context, number *big.Int) (Block, error)
 
+	// BlocksByNumbers returns blocks at the given numbers.
+	// Returns blocks up to the first not-found (contiguous only).
+	BlocksByNumbers(ctx context.Context, numbers []*big.Int) ([]Block, error)
+
 	// BlockReceipts returns all receipts for the block at the given number.
 	BlockReceipts(ctx context.Context, number *big.Int) ([]Receipt, error)
 
@@ -161,6 +165,11 @@ func (n *EmbeddedNode) BlockByNumber(ctx context.Context, number *big.Int) (Bloc
 	return n.source.BlockByNumber(ctx, number)
 }
 
+// BlocksByNumbers delegates to the DataSource.
+func (n *EmbeddedNode) BlocksByNumbers(ctx context.Context, numbers []*big.Int) ([]Block, error) {
+	return n.source.BlocksByNumbers(ctx, numbers)
+}
+
 // BlockReceipts delegates to the DataSource.
 func (n *EmbeddedNode) BlockReceipts(ctx context.Context, number *big.Int) ([]Receipt, error) {
 	return n.source.BlockReceipts(ctx, number)

pkg/ethereum/execution/embedded_node_test.go

@@ -170,6 +170,21 @@ func (m *MockDataSource) IsSynced() bool {
 	return args.Bool(0)
 }
 
+func (m *MockDataSource) BlocksByNumbers(ctx context.Context, numbers []*big.Int) ([]execution.Block, error) {
+	args := m.Called(ctx, numbers)
+
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	val, ok := args.Get(0).([]execution.Block)
+	if !ok {
+		return nil, args.Error(1)
+	}
+
+	return val, args.Error(1)
+}
+
 func TestEmbeddedNode_Creation(t *testing.T) {
 	log := logrus.New()
 	log.SetLevel(logrus.ErrorLevel)

pkg/ethereum/execution/geth/rpc.go

@@ -5,11 +5,13 @@ package geth
 import (
 	"context"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"math/big"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/rpc"
 
 	pcommon "github.com/ethpandaops/execution-processor/pkg/common"
@@ -71,6 +73,116 @@ func (n *RPCNode) blockByNumber(ctx context.Context, blockNumber *big.Int) (exec
 	return NewBlockAdapter(block), nil
 }
 
+// blocksByNumbers fetches multiple blocks using batch RPC calls.
+// Returns blocks up to the first not-found (contiguous only).
+func (n *RPCNode) blocksByNumbers(ctx context.Context, numbers []*big.Int) ([]execution.Block, error) {
+	if len(numbers) == 0 {
+		return []execution.Block{}, nil
+	}
+
+	start := time.Now()
+	network := n.Metadata().ChainID()
+
+	// Prepare batch calls using json.RawMessage to handle null responses
+	batch := make([]rpc.BatchElem, len(numbers))
+	results := make([]*json.RawMessage, len(numbers))
+
+	for i, num := range numbers {
+		results[i] = new(json.RawMessage)
+		batch[i] = rpc.BatchElem{
+			Method: "eth_getBlockByNumber",
+			Args:   []interface{}{toBlockNumArg(num), true}, // true = include transactions
+			Result: results[i],
+		}
+	}
+
+	// Execute batch call
+	err := n.rpcClient.BatchCallContext(ctx, batch)
+
+	duration := time.Since(start)
+
+	// Record batch RPC metrics
+	status := statusSuccess
+	if err != nil {
+		status = statusError
+	}
+
+	pcommon.RPCCallDuration.WithLabelValues(
+		fmt.Sprintf("%d", network),
+		n.config.Name,
+		"eth_getBlockByNumber_batch",
+		status,
+	).Observe(duration.Seconds())
+
+	pcommon.RPCCallsTotal.WithLabelValues(
+		fmt.Sprintf("%d", network),
+		n.config.Name,
+		"eth_getBlockByNumber_batch",
+		status,
+	).Inc()
+
+	if err != nil {
+		return nil, fmt.Errorf("batch call failed: %w", err)
+	}
+
+	// Process results, stopping at first not-found (contiguity requirement)
+	blocks := make([]execution.Block, 0, len(numbers))
+
+	for i, elem := range batch {
+		// Check for individual call error - stop at first error for contiguity
+		// We intentionally don't return this error as we want partial results
+		if elem.Error != nil {
+			break
+		}
+
+		// Check for nil/not-found block (null JSON response)
+		if results[i] == nil || len(*results[i]) == 0 || string(*results[i]) == "null" {
+			// Block not found - stop here for contiguity
+			break
+		}
+
+		// Parse the block from JSON
+		block, parseErr := parseBlockFromJSON(*results[i])
+		if parseErr != nil {
+			// Parse error - stop here for contiguity
+			break
+		}
+
+		blocks = append(blocks, NewBlockAdapter(block))
+	}
+
+	return blocks, nil //nolint:nilerr // Intentionally returning partial results for contiguity
+}
+
+// toBlockNumArg converts a block number to the RPC argument format.
+func toBlockNumArg(number *big.Int) string {
+	if number == nil {
+		return "latest"
+	}
+
+	return fmt.Sprintf("0x%x", number)
+}
+
+// parseBlockFromJSON parses a types.Block from JSON-RPC response.
+func parseBlockFromJSON(raw json.RawMessage) (*types.Block, error) {
+	// Use go-ethereum's internal header structure for unmarshaling
+	var head *types.Header
+	if err := json.Unmarshal(raw, &head); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal block header: %w", err)
+	}
+
+	// Parse transactions separately
+	var body struct {
+		Transactions []*types.Transaction `json:"transactions"`
+	}
+
+	if err := json.Unmarshal(raw, &body); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal block body: %w", err)
+	}
+
+	return types.NewBlockWithHeader(head).WithBody(types.Body{Transactions: body.Transactions}), nil
+}
+
 // getTraceParams returns VM trace parameters with configurable options.
 func getTraceParams(hash string, options execution.TraceOptions) []any {
 	return []any{

pkg/ethereum/execution/geth/rpc_node.go

@@ -308,6 +308,12 @@ func (n *RPCNode) BlockByNumber(ctx context.Context, number *big.Int) (execution
 	return n.blockByNumber(ctx, number)
 }
 
+// BlocksByNumbers returns blocks at the given numbers using batch RPC.
+// Returns blocks up to the first not-found (contiguous only).
+func (n *RPCNode) BlocksByNumbers(ctx context.Context, numbers []*big.Int) ([]execution.Block, error) {
+	return n.blocksByNumbers(ctx, numbers)
+}
+
 // BlockReceipts returns all receipts for the block at the given number.
 func (n *RPCNode) BlockReceipts(ctx context.Context, number *big.Int) ([]execution.Receipt, error) {
 	return n.blockReceipts(ctx, number)

pkg/ethereum/execution/interface.go

@@ -41,6 +41,11 @@ type Node interface {
 	// BlockByNumber returns the block at the given number.
 	BlockByNumber(ctx context.Context, number *big.Int) (Block, error)
 
+	// BlocksByNumbers returns blocks at the given numbers using batch RPC.
+	// Returns blocks up to the first not-found (contiguous only).
+	// If a block is not found, the returned slice contains all blocks before that point.
+	BlocksByNumbers(ctx context.Context, numbers []*big.Int) ([]Block, error)
+
 	// BlockReceipts returns all receipts for the block at the given number.
 	BlockReceipts(ctx context.Context, number *big.Int) ([]Receipt, error)
 

pkg/ethereum/pool_test.go

@@ -98,6 +98,15 @@ func (m *MockNode) BlockByNumber(_ context.Context, number *big.Int) (execution.
 	return &MockBlock{number: number}, nil
 }
 
+func (m *MockNode) BlocksByNumbers(_ context.Context, numbers []*big.Int) ([]execution.Block, error) {
+	blocks := make([]execution.Block, len(numbers))
+	for i, num := range numbers {
+		blocks[i] = &MockBlock{number: num}
+	}
+
+	return blocks, nil
+}
+
 func (m *MockNode) BlockReceipts(_ context.Context, _ *big.Int) ([]execution.Receipt, error) {
 	return []execution.Receipt{}, nil
 }
@@ -806,3 +815,12 @@ func (ds *testDataSource) ClientType() string {
 func (ds *testDataSource) IsSynced() bool {
 	return true
 }
+
+func (ds *testDataSource) BlocksByNumbers(_ context.Context, numbers []*big.Int) ([]execution.Block, error) {
+	blocks := make([]execution.Block, len(numbers))
+	for i, num := range numbers {
+		blocks[i] = &MockBlock{number: num}
+	}
+
+	return blocks, nil
+}

pkg/processor/defaults.go

@@ -25,4 +25,13 @@ const (
 
 	// DefaultLeaderRenewalInterval is the default renewal interval for leader election.
 	DefaultLeaderRenewalInterval = 3 * time.Second
+
+	// DefaultBackpressureBackoffMin is the minimum backoff duration when backpressure is detected.
+	DefaultBackpressureBackoffMin = 10 * time.Millisecond
+
+	// DefaultBackpressureBackoffMax is the maximum backoff duration when backpressure persists.
+	DefaultBackpressureBackoffMax = 1 * time.Second
+
+	// DefaultBackpressureJitterFraction is the fraction of backoff to add as random jitter (0.25 = 25%).
+	DefaultBackpressureJitterFraction = 0.25
 )

pkg/processor/manager.go

@@ -27,6 +27,7 @@ import (
 	"context"
 	"fmt"
 	"math/big"
+	"math/rand/v2"
 	"strings"
 	"sync"
 	"time"
@@ -101,6 +102,9 @@ type Manager struct {
 	// Track queue high water marks
 	queueHighWaterMarks map[string]int
 	queueMetricsMutex   sync.RWMutex
+
+	// Backpressure backoff tracking
+	backpressureBackoff time.Duration
 }
 
 func NewManager(log logrus.FieldLogger, config *Config, pool *ethereum.Pool, state *s.Manager, redis *r.Client, redisPrefix string) (*Manager, error) {
@@ -749,15 +753,50 @@ func (m *Manager) runBlockProcessing(ctx context.Context) {
 			if m.config.Interval > 0 {
 				// Fixed interval mode - always sleep the configured duration
 				time.Sleep(m.config.Interval)
+				m.backpressureBackoff = 0 // Reset backoff in fixed interval mode
 			} else if !workDone {
-				// Zero interval mode with no work - small backoff to prevent CPU spin
-				time.Sleep(DefaultNoWorkBackoff)
+				// Zero interval mode with no work - exponential backoff with jitter
+				m.backpressureBackoff = m.calculateBackpressureBackoff(m.backpressureBackoff)
+				time.Sleep(m.backpressureBackoff)
+			} else {
+				// Zero interval with work done - reset backoff and loop immediately
+				m.backpressureBackoff = 0
 			}
-			// Zero interval with work done - loop immediately (no sleep)
 		}
 	}
 }
 
+// calculateBackpressureBackoff calculates the next backoff duration using exponential backoff with jitter.
+func (m *Manager) calculateBackpressureBackoff(current time.Duration) time.Duration {
+	var next time.Duration
+
+	if current == 0 {
+		// Start with minimum backoff
+		next = DefaultBackpressureBackoffMin
+	} else {
+		// Double the current backoff (exponential)
+		next = current * 2
+	}
+
+	// Cap at maximum backoff
+	if next > DefaultBackpressureBackoffMax {
+		next = DefaultBackpressureBackoffMax
+	}
+
+	// Add jitter (up to 25% of the backoff duration)
+	//nolint:gosec // G404: Weak RNG is fine for backoff jitter - no security requirement
+	jitter := time.Duration(rand.Float64() * DefaultBackpressureJitterFraction * float64(next))
+	next += jitter
+
+	m.log.WithFields(logrus.Fields{
+		"previous_backoff": current,
+		"next_backoff":     next,
+		"jitter":           jitter,
+	}).Debug("Calculated backpressure backoff")
+
+	return next
+}
+
 // monitorQueues monitors queue health and archived items.
 func (m *Manager) monitorQueues(ctx context.Context) {
 	// Get Redis options for Asynq Inspector