fix: F-2026-18802 | [Dual Defense] EVM FilterLogs Fixed 9000-Block Chunks Stall Cursor With No Adaptive Shrink

universalClient/chains/evm/event_listener.go

@@ -208,58 +208,94 @@ func (el *EventListener) processNewBlocks(
 	}
 
 	// Process blocks in range
-	if err := el.processBlockRange(ctx, *currentBlock, latestBlock, topics); err != nil {
-		return fmt.Errorf("failed to process block range: %w", err)
+	nextBlock, rangeErr := el.processBlockRange(ctx, *currentBlock, latestBlock, topics)
+
+	// Commit whatever was covered even when a later chunk failed. Holding the
+	// cursor back would re-read the blocks already handled on every tick, so one
+	// unreadable window would sit in front of everything behind it indefinitely.
+	if nextBlock > *currentBlock {
+		if err := el.updateLastProcessedBlock(nextBlock - 1); err != nil {
+			el.logger.Error().Err(err).Msg("failed to update last processed block")
+			// Don't return error - continue processing
+		}
+		*currentBlock = nextBlock
 	}
 
-	// Update last processed block in database
-	if err := el.updateLastProcessedBlock(latestBlock); err != nil {
-		el.logger.Error().Err(err).Msg("failed to update last processed block")
-		// Don't return error - continue processing
+	if rangeErr != nil {
+		return fmt.Errorf("failed to process block range: %w", rangeErr)
 	}
-
-	// Move to next block
-	*currentBlock = latestBlock + 1
 	return nil
 }
 
-// processBlockRange processes events in a range of blocks
+// Block span for a single eth_getLogs call. Providers cap the result set rather
+// than the block count, so a dense window can be rejected at a span that is
+// normally fine. maxBlockRange is the optimistic starting point and minBlockRange
+// the floor we stop shrinking at.
+const (
+	maxBlockRange uint64 = 9000 // Safe under the 10000 RPC limit
+	minBlockRange uint64 = 100
+)
+
+// processBlockRange processes events in a range of blocks, returning the first
+// block it did not cover. That is fromBlock when nothing was processed and
+// toBlock+1 when everything was, so the caller can commit partial progress
+// whether or not an error is also returned.
+//
+// A rejected query is retried over a smaller span rather than abandoned: the
+// limit is on results, so halving until the window fits gets past a dense range
+// that a fixed span cannot. Shrinking is linear rather than a recursive split,
+// which would issue exponentially many calls against a range that keeps failing.
 func (el *EventListener) processBlockRange(
 	ctx context.Context,
 	fromBlock, toBlock uint64,
 	topics []ethcommon.Hash,
-) error {
-	const maxBlockRange uint64 = 9000 // Safe under the 10000 RPC limit
+) (uint64, error) {
+	span := maxBlockRange
+	nextFrom := fromBlock
 
-	currentFrom := fromBlock
-
-	// Process in chunks if the range is too large
-	for currentFrom <= toBlock {
-		currentTo := currentFrom + maxBlockRange - 1
-		if currentTo > toBlock {
+	for nextFrom <= toBlock {
+		currentTo := nextFrom + span - 1
+		if currentTo > toBlock || currentTo < nextFrom { // second test catches overflow
 			currentTo = toBlock
 		}
 
-		// Log chunk processing for large ranges
-		blockRange := currentTo - currentFrom + 1
+		blockRange := currentTo - nextFrom + 1
 		if blockRange > 1000 {
 			el.logger.Debug().
-				Uint64("from_block", currentFrom).
+				Uint64("from_block", nextFrom).
 				Uint64("to_block", currentTo).
 				Uint64("range_size", blockRange).
 				Msg("processing block chunk")
 		}
 
-		// Process chunk
-		if err := el.processBlockChunk(ctx, currentFrom, currentTo, topics); err != nil {
-			return fmt.Errorf("failed to process chunk %d-%d: %w", currentFrom, currentTo, err)
+		if err := el.processBlockChunk(ctx, nextFrom, currentTo, topics); err != nil {
+			// Halve what was actually attempted, not the nominal span: near the end
+			// of a range the span is clamped to toBlock, so shrinking the span alone
+			// would resend the identical query until it dropped below the remainder.
+			if blockRange > minBlockRange {
+				span = blockRange / 2
+				if span < minBlockRange {
+					span = minBlockRange
+				}
+				el.logger.Warn().
+					Err(err).
+					Uint64("from_block", nextFrom).
+					Uint64("to_block", currentTo).
+					Uint64("retry_span", span).
+					Msg("log query failed, retrying the same start over a smaller span")
+				continue
+			}
+
+			// At the floor the span is no longer the problem. Report the failure
+			// and leave the cursor here: skipping ahead would drop any deposits in
+			// these blocks permanently, which is worse than waiting for the RPC.
+			return nextFrom, fmt.Errorf("failed to process chunk %d-%d at minimum span: %w", nextFrom, currentTo, err)
 		}
 
-		// Move to next chunk
-		currentFrom = currentTo + 1
+		nextFrom = currentTo + 1
 	}
 
-	return nil
+	return nextFrom, nil
 }
 
 // processBlockChunk processes a single chunk of blocks

universalClient/chains/evm/event_listener_test.go

@@ -2,6 +2,14 @@ package evm
 
 import (
 	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -416,3 +424,248 @@ func TestEventListener_ContextCancellationStopsGoroutine(t *testing.T) {
 	el.Stop()
 	assert.False(t, el.IsRunning())
 }
+
+// logQueryServer serves eth_getLogs, rejecting any query whose block span exceeds
+// maxSpan the way a provider rejects an over-large result set, and recording the
+// spans it was asked for so tests can assert how the client adapted.
+type logQueryServer struct {
+	maxSpan  uint64
+	failFrom uint64 // when non-zero, reject any query overlapping this block onwards
+	mu       sync.Mutex
+	asked    [][2]uint64
+	served   [][2]uint64 // only the queries that actually returned logs
+}
+
+func (s *logQueryServer) record(from, to uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.asked = append(s.asked, [2]uint64{from, to})
+}
+
+func (s *logQueryServer) spans() [][2]uint64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([][2]uint64(nil), s.asked...)
+}
+
+func (s *logQueryServer) servedSpans() [][2]uint64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([][2]uint64(nil), s.served...)
+}
+
+func (s *logQueryServer) recordServed(from, to uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.served = append(s.served, [2]uint64{from, to})
+}
+
+func (s *logQueryServer) start(t *testing.T) *RPCClient {
+	t.Helper()
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		w.Header().Set("Content-Type", "application/json")
+
+		if !strings.Contains(string(body), "eth_getLogs") {
+			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":"0x1"}`))
+			return
+		}
+
+		var req struct {
+			Params []struct {
+				FromBlock string `json:"fromBlock"`
+				ToBlock   string `json:"toBlock"`
+			} `json:"params"`
+		}
+		_ = json.Unmarshal(body, &req)
+		from, _ := strconv.ParseUint(strings.TrimPrefix(req.Params[0].FromBlock, "0x"), 16, 64)
+		to, _ := strconv.ParseUint(strings.TrimPrefix(req.Params[0].ToBlock, "0x"), 16, 64)
+		s.record(from, to)
+
+		overSpan := to-from+1 > s.maxSpan
+		stuck := s.failFrom != 0 && to >= s.failFrom
+		if overSpan || stuck {
+			_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"error":{"code":-32005,"message":"query returned more than 10000 results"}}`))
+			return
+		}
+		s.recordServed(from, to)
+		_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":[]}`))
+	}))
+	t.Cleanup(srv.Close)
+
+	rpcClient, err := NewRPCClient([]string{srv.URL}, 1, zerolog.Nop())
+	require.NoError(t, err)
+	t.Cleanup(rpcClient.Close)
+	return rpcClient
+}
+
+func newRangeListener(t *testing.T, rpcClient *RPCClient) *EventListener {
+	t.Helper()
+	el, err := NewEventListener(rpcClient, "0x1111111111111111111111111111111111111111",
+		"0x2222222222222222222222222222222222222222", "eip155:1", nil, nil, testDB(t), 10, nil, zerolog.Nop())
+	require.NoError(t, err)
+	return el
+}
+
+// Providers cap the result set, not the block count, so a dense window is
+// rejected at a span that is normally fine. A fixed span retries the same
+// rejected query forever and the cursor never moves past it.
+func TestProcessBlockRange_ShrinksSpanUntilTheQueryFits(t *testing.T) {
+	srv := &logQueryServer{maxSpan: 1000} // anything wider than 1000 blocks is rejected
+	el := newRangeListener(t, srv.start(t))
+
+	next, err := el.processBlockRange(context.Background(), 1, 2000, nil)
+	require.NoError(t, err)
+	assert.Equal(t, uint64(2001), next, "the whole range must end up covered")
+
+	spans := srv.spans()
+	require.NotEmpty(t, spans)
+
+	// The first attempt is optimistic, and every retry restarts at the same block
+	// rather than skipping the blocks that were rejected.
+	assert.Equal(t, uint64(1), spans[0][0])
+	assert.Equal(t, uint64(2000), spans[0][1], "first attempt spans the whole range")
+
+	var widths []uint64
+	for _, s := range spans {
+		if s[0] == 1 {
+			widths = append(widths, s[1]-s[0]+1)
+		}
+	}
+	require.Greater(t, len(widths), 1, "must retry the same start over a smaller span")
+	for i := 1; i < len(widths); i++ {
+		assert.Less(t, widths[i], widths[i-1], "each retry must be narrower")
+	}
+}
+
+// A window that cannot be read even at the floor must not be stepped over:
+// the blocks may contain deposits, and skipping them loses those permanently.
+func TestProcessBlockRange_DoesNotSkipAnUnreadableWindow(t *testing.T) {
+	srv := &logQueryServer{maxSpan: maxBlockRange, failFrom: 1} // every query fails
+	el := newRangeListener(t, srv.start(t))
+
+	next, err := el.processBlockRange(context.Background(), 1, 500, nil)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "minimum span")
+	assert.Equal(t, uint64(1), next, "cursor must stay put, not advance past unread blocks")
+}
+
+// Work already done must be committed. Holding the cursor at the start would
+// re-read the earlier chunks on every tick, so one bad window would sit in front
+// of everything behind it.
+func TestProcessBlockRange_ReportsPartialProgressOnFailure(t *testing.T) {
+	// First 9000 blocks are readable; anything from 9001 always fails.
+	srv := &logQueryServer{maxSpan: maxBlockRange, failFrom: 9001}
+	el := newRangeListener(t, srv.start(t))
+
+	next, err := el.processBlockRange(context.Background(), 1, 20000, nil)
+	require.Error(t, err)
+	assert.Equal(t, uint64(9001), next, "must report the first block it could not cover")
+}
+
+func TestProcessBlockRange_SinglePassWhenNothingIsRejected(t *testing.T) {
+	srv := &logQueryServer{maxSpan: maxBlockRange}
+	el := newRangeListener(t, srv.start(t))
+
+	next, err := el.processBlockRange(context.Background(), 1, 500, nil)
+	require.NoError(t, err)
+	assert.Equal(t, uint64(501), next)
+	assert.Len(t, srv.spans(), 1, "a range that fits must not be split")
+}
+
+// assertExactCoverage checks that the served queries tile [from,to] with no gap
+// and no block fetched twice. A gap is a block whose logs are never read, which
+// for an inbound is a deposit nobody observes.
+func assertExactCoverage(t *testing.T, served [][2]uint64, from, to uint64) {
+	t.Helper()
+
+	sort.Slice(served, func(i, j int) bool { return served[i][0] < served[j][0] })
+
+	require.NotEmpty(t, served, "nothing was fetched for %d-%d", from, to)
+	assert.Equal(t, from, served[0][0], "coverage must start at the first block")
+	assert.Equal(t, to, served[len(served)-1][1], "coverage must end at the last block")
+
+	for i := 1; i < len(served); i++ {
+		prevEnd, thisStart := served[i-1][1], served[i][0]
+		assert.Equal(t, prevEnd+1, thisStart,
+			"chunk %d starts at %d but the previous ended at %d", i, thisStart, prevEnd)
+	}
+
+	var covered uint64
+	for _, c := range served {
+		require.LessOrEqual(t, c[0], c[1], "chunk %d-%d is inverted", c[0], c[1])
+		covered += c[1] - c[0] + 1
+	}
+	assert.Equal(t, to-from+1, covered, "total blocks covered must equal the range size")
+}
+
+// Every block in the range must be fetched exactly once, whatever the span ends
+// up being. Off-by-one at a chunk boundary would silently skip a block.
+func TestProcessBlockRange_CoversEveryBlockExactlyOnce(t *testing.T) {
+	cases := []struct {
+		name       string
+		from, to   uint64
+		serverSpan uint64 // widest query the server will accept
+	}{
+		{"single block", 1, 1, maxBlockRange},
+		{"single block at zero", 0, 0, maxBlockRange},
+		{"range starting at zero", 0, 500, maxBlockRange},
+		{"exactly one full span", 1, maxBlockRange, maxBlockRange},
+		{"one block past a full span", 1, maxBlockRange + 1, maxBlockRange},
+		{"one block short of a full span", 1, maxBlockRange - 1, maxBlockRange},
+		{"several full spans", 1, maxBlockRange * 3, maxBlockRange},
+		{"several spans plus a remainder", 1, maxBlockRange*2 + 137, maxBlockRange},
+		{"forced shrink, divisible", 1, 2000, 1000},
+		{"forced shrink, not divisible", 1, 2500, 333},
+		{"forced shrink to the floor", 1, 1000, minBlockRange},
+		{"shrink with an odd start", 4097, 9999, 700},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := &logQueryServer{maxSpan: tc.serverSpan}
+			el := newRangeListener(t, srv.start(t))
+
+			next, err := el.processBlockRange(context.Background(), tc.from, tc.to, nil)
+			require.NoError(t, err)
+			assert.Equal(t, tc.to+1, next, "must report the range as fully covered")
+
+			assertExactCoverage(t, srv.servedSpans(), tc.from, tc.to)
+		})
+	}
+}
+
+// After a shrink the walk continues at the smaller span. The blocks either side
+// of the failure boundary must still be covered exactly once.
+func TestProcessBlockRange_NoGapAroundAShrink(t *testing.T) {
+	srv := &logQueryServer{maxSpan: 750}
+	el := newRangeListener(t, srv.start(t))
+
+	next, err := el.processBlockRange(context.Background(), 100, 3100, nil)
+	require.NoError(t, err)
+	assert.Equal(t, uint64(3101), next)
+
+	assertExactCoverage(t, srv.servedSpans(), 100, 3100)
+}
+
+// Across successive polls the caller resumes from the block the previous call
+// reported, so a partial range must hand back a boundary that leaves no hole.
+func TestProcessBlockRange_ResumeAfterPartialLeavesNoGap(t *testing.T) {
+	// Blocks from 5001 are unreadable, so the first call stops there.
+	srv := &logQueryServer{maxSpan: maxBlockRange, failFrom: 5001}
+	el := newRangeListener(t, srv.start(t))
+
+	next, err := el.processBlockRange(context.Background(), 1, 8000, nil)
+	require.Error(t, err)
+	assertExactCoverage(t, srv.servedSpans(), 1, next-1)
+
+	// The obstruction clears and the caller resumes from where it stopped.
+	srv2 := &logQueryServer{maxSpan: maxBlockRange}
+	el2 := newRangeListener(t, srv2.start(t))
+
+	final, err := el2.processBlockRange(context.Background(), next, 8000, nil)
+	require.NoError(t, err)
+	assert.Equal(t, uint64(8001), final)
+	assertExactCoverage(t, srv2.servedSpans(), next, 8000)
+}