diff --git a/universalClient/chains/evm/event_listener.go b/universalClient/chains/evm/event_listener.go index 294ddcc3..8bd5c045 100644 --- a/universalClient/chains/evm/event_listener.go +++ b/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 diff --git a/universalClient/chains/evm/event_listener_test.go b/universalClient/chains/evm/event_listener_test.go index 71d76420..d9e493d3 100644 --- a/universalClient/chains/evm/event_listener_test.go +++ b/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) +}