feat(processor): add gap detection for block processing

Add ClickHouse gap detection to find missing blocks between the oldest
processed block and the current block. This catches gaps that fall
outside the maxPendingBlockRange window.

Changes:
- Add GetIncompleteBlocksInRange() to state manager for range queries
- Add GapStateProvider interface extending StateProvider
- Add GetGaps() method to limiter for full-range gap detection
- Add GapDetectionConfig with configurable scan interval, batch size,
  and lookback range
- Add periodic gap scanner goroutine to processor
- Exclude maxPendingBlockRange window from gap scan to avoid double work

The gap scanner runs periodically (default: 5m) and reprocesses any
incomplete blocks found outside the limiter's normal detection window.

Author: Savid

pkg/processor/tracker/limiter.go

@@ -3,6 +3,7 @@ package tracker
 import (
 	"context"
 	"fmt"
+	"math/big"
 
 	"github.com/ethpandaops/execution-processor/pkg/common"
 	"github.com/sirupsen/logrus"
@@ -15,6 +16,13 @@ type StateProvider interface {
 	MarkBlockComplete(ctx context.Context, blockNumber uint64, network, processor string) error
 }
 
+// GapStateProvider extends StateProvider with gap detection capabilities.
+type GapStateProvider interface {
+	StateProvider
+	GetIncompleteBlocksInRange(ctx context.Context, network, processor string, minBlock, maxBlock uint64, limit int) ([]uint64, error)
+	GetMinMaxStoredBlocks(ctx context.Context, network, processor string) (*big.Int, *big.Int, error)
+}
+
 // LimiterConfig holds configuration for the Limiter.
 type LimiterConfig struct {
 	MaxPendingBlockRange int
@@ -207,3 +215,76 @@ func (l *Limiter) ValidateBatchWithinLeash(ctx context.Context, startBlock uint6
 
 	return nil
 }
+
+// GetGaps returns incomplete blocks outside the maxPendingBlockRange window.
+// If lookbackRange is 0, scans from the oldest stored block.
+// This performs a full-range scan for gap detection, excluding the recent window
+// that is already handled by IsBlockedByIncompleteBlocks.
+func (l *Limiter) GetGaps(ctx context.Context, currentBlock uint64, lookbackRange uint64, limit int) ([]uint64, error) {
+	gapProvider, ok := l.stateProvider.(GapStateProvider)
+	if !ok {
+		return nil, fmt.Errorf("state provider does not support gap detection")
+	}
+
+	var minBlock uint64
+
+	if lookbackRange == 0 {
+		// Unlimited: scan from oldest stored block
+		minStored, _, err := gapProvider.GetMinMaxStoredBlocks(ctx, l.network, l.processor)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get min stored block: %w", err)
+		}
+
+		if minStored == nil {
+			// No blocks stored yet
+			return nil, nil
+		}
+
+		minBlock = minStored.Uint64()
+	} else {
+		// Limited: scan from currentBlock - lookbackRange
+		if currentBlock > lookbackRange {
+			minBlock = currentBlock - lookbackRange
+		}
+	}
+
+	// Calculate maxBlock to exclude the window handled by IsBlockedByIncompleteBlocks.
+	// The limiter already handles blocks within [currentBlock - maxPendingBlockRange, currentBlock],
+	// so we only scan up to (currentBlock - maxPendingBlockRange - 1) to avoid double work.
+	maxBlock := currentBlock
+
+	if l.config.MaxPendingBlockRange > 0 {
+		exclusionWindow := uint64(l.config.MaxPendingBlockRange) //nolint:gosec // validated in config
+
+		if currentBlock > exclusionWindow {
+			maxBlock = currentBlock - exclusionWindow - 1
+		} else {
+			// Current block is within the exclusion window, nothing to scan
+			return nil, nil
+		}
+	}
+
+	// Ensure minBlock doesn't exceed maxBlock
+	if minBlock > maxBlock {
+		return nil, nil
+	}
+
+	gaps, err := gapProvider.GetIncompleteBlocksInRange(
+		ctx, l.network, l.processor,
+		minBlock, maxBlock, limit,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get incomplete blocks in range: %w", err)
+	}
+
+	if len(gaps) > 0 {
+		l.log.WithFields(logrus.Fields{
+			"min_block": minBlock,
+			"max_block": maxBlock,
+			"gap_count": len(gaps),
+			"first_gap": gaps[0],
+		}).Debug("Found gaps in block range")
+	}
+
+	return gaps, nil
+}

pkg/processor/tracker/limiter_gap_test.go

@@ -0,0 +1,292 @@
+package tracker
+
+import (
+	"context"
+	"math/big"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// mockGapStateProvider implements GapStateProvider for testing.
+type mockGapStateProvider struct {
+	mockStateProvider
+	incompleteBlocksInRange []uint64
+	minStoredBlock          *big.Int
+	maxStoredBlock          *big.Int
+	getIncompleteErr        error
+	getMinMaxErr            error
+}
+
+func (m *mockGapStateProvider) GetIncompleteBlocksInRange(
+	_ context.Context, _, _ string, _, _ uint64, _ int,
+) ([]uint64, error) {
+	if m.getIncompleteErr != nil {
+		return nil, m.getIncompleteErr
+	}
+
+	return m.incompleteBlocksInRange, nil
+}
+
+func (m *mockGapStateProvider) GetMinMaxStoredBlocks(
+	_ context.Context, _, _ string,
+) (*big.Int, *big.Int, error) {
+	if m.getMinMaxErr != nil {
+		return nil, nil, m.getMinMaxErr
+	}
+
+	return m.minStoredBlock, m.maxStoredBlock, nil
+}
+
+func TestGetGaps_FindsMissingBlocks(t *testing.T) {
+	// Setup: blocks 5,6,7,9,10...100 are complete, block 8 is incomplete
+	// With maxPendingBlockRange=2 and currentBlock=100, gap scanner looks at blocks up to 97
+	// (excluding the window 98-100 that the limiter handles)
+	mockProvider := &mockGapStateProvider{
+		minStoredBlock:          big.NewInt(5),
+		maxStoredBlock:          big.NewInt(100),
+		incompleteBlocksInRange: []uint64{8},
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 2})
+
+	ctx := context.Background()
+	currentBlock := uint64(100)
+	lookbackRange := uint64(0) // Unlimited
+
+	// Gap scanner searches [5, 97] (excludes 98-100 handled by limiter)
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Equal(t, []uint64{8}, gaps)
+}
+
+func TestGetGaps_RespectsLookbackRange(t *testing.T) {
+	mockProvider := &mockGapStateProvider{
+		// GetMinMaxStoredBlocks should NOT be called when lookbackRange is set
+		incompleteBlocksInRange: []uint64{75, 80},
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 2})
+
+	ctx := context.Background()
+	currentBlock := uint64(100)
+	lookbackRange := uint64(50) // Only look back 50 blocks
+
+	// Should query from block 50 (100 - 50) to 100
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Equal(t, []uint64{75, 80}, gaps)
+}
+
+func TestGetGaps_NoGaps(t *testing.T) {
+	mockProvider := &mockGapStateProvider{
+		minStoredBlock:          big.NewInt(1),
+		maxStoredBlock:          big.NewInt(100),
+		incompleteBlocksInRange: []uint64{},
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 2})
+
+	ctx := context.Background()
+	currentBlock := uint64(100)
+	lookbackRange := uint64(0)
+
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Empty(t, gaps)
+}
+
+func TestGetGaps_DoesNotLookBeforeOldestStoredBlock(t *testing.T) {
+	// Ensure we don't query for blocks before the oldest stored block
+	mockProvider := &mockGapStateProvider{
+		// Oldest stored is 50, so we should only query from 50 onwards
+		minStoredBlock:          big.NewInt(50),
+		maxStoredBlock:          big.NewInt(100),
+		incompleteBlocksInRange: []uint64{},
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 2})
+
+	ctx := context.Background()
+	currentBlock := uint64(100)
+	lookbackRange := uint64(0) // Unlimited
+
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Empty(t, gaps)
+}
+
+func TestGetGaps_NoBlocksStored(t *testing.T) {
+	mockProvider := &mockGapStateProvider{
+		minStoredBlock: nil, // No blocks stored
+		maxStoredBlock: nil,
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 2})
+
+	ctx := context.Background()
+	currentBlock := uint64(100)
+	lookbackRange := uint64(0)
+
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Nil(t, gaps)
+}
+
+func TestGetGaps_StateProviderDoesNotSupportGapDetection(t *testing.T) {
+	// Use the basic mockStateProvider which doesn't implement GapStateProvider
+	mockProvider := &mockStateProvider{}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 2})
+
+	ctx := context.Background()
+	currentBlock := uint64(100)
+	lookbackRange := uint64(0)
+
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "state provider does not support gap detection")
+	assert.Nil(t, gaps)
+}
+
+func TestGetGaps_MultipleGaps(t *testing.T) {
+	mockProvider := &mockGapStateProvider{
+		minStoredBlock:          big.NewInt(1),
+		maxStoredBlock:          big.NewInt(100),
+		incompleteBlocksInRange: []uint64{5, 10, 15, 20, 25},
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 2})
+
+	ctx := context.Background()
+	currentBlock := uint64(100)
+	lookbackRange := uint64(0)
+
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Equal(t, []uint64{5, 10, 15, 20, 25}, gaps)
+}
+
+func TestGetGaps_LookbackRangeGreaterThanCurrentBlock(t *testing.T) {
+	// When lookbackRange is greater than currentBlock, minBlock should be 0
+	// With maxPendingBlockRange=2 and currentBlock=10, gap scanner looks at blocks up to 7
+	mockProvider := &mockGapStateProvider{
+		incompleteBlocksInRange: []uint64{3},
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 2})
+
+	ctx := context.Background()
+	currentBlock := uint64(10)
+	lookbackRange := uint64(100) // Greater than currentBlock
+
+	// Gap scanner searches [0, 7] (excludes 8-10 handled by limiter)
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Equal(t, []uint64{3}, gaps)
+}
+
+func TestGetGaps_ExcludesMaxPendingBlockRangeWindow(t *testing.T) {
+	// Verify that gaps within maxPendingBlockRange are NOT returned
+	// because they're already handled by IsBlockedByIncompleteBlocks
+	mockProvider := &mockGapStateProvider{
+		minStoredBlock:          big.NewInt(1),
+		maxStoredBlock:          big.NewInt(100),
+		incompleteBlocksInRange: []uint64{50}, // Gap at block 50, outside the exclusion window
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 5})
+
+	ctx := context.Background()
+	currentBlock := uint64(100)
+	lookbackRange := uint64(0)
+
+	// With maxPendingBlockRange=5, gap scanner searches [1, 94] (excludes 95-100)
+	// Block 50 should be found since it's outside the exclusion window
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Equal(t, []uint64{50}, gaps)
+}
+
+func TestGetGaps_CurrentBlockWithinExclusionWindow(t *testing.T) {
+	// When currentBlock is smaller than or equal to maxPendingBlockRange,
+	// there's nothing to scan outside the exclusion window
+	mockProvider := &mockGapStateProvider{
+		minStoredBlock:          big.NewInt(1),
+		maxStoredBlock:          big.NewInt(5),
+		incompleteBlocksInRange: []uint64{2},
+	}
+
+	limiter := NewLimiter(&LimiterDeps{
+		Log:           logrus.NewEntry(logrus.New()),
+		StateProvider: mockProvider,
+		Network:       "mainnet",
+		Processor:     "simple",
+	}, LimiterConfig{MaxPendingBlockRange: 10}) // Larger than currentBlock
+
+	ctx := context.Background()
+	currentBlock := uint64(5)
+	lookbackRange := uint64(0)
+
+	// currentBlock (5) <= maxPendingBlockRange (10), so nothing to scan
+	gaps, err := limiter.GetGaps(ctx, currentBlock, lookbackRange, 100)
+
+	require.NoError(t, err)
+	assert.Nil(t, gaps)
+}