diff --git a/eth/filters/bor_filter.go b/eth/filters/bor_filter.go index 27a6564976..7e7065326d 100644 --- a/eth/filters/bor_filter.go +++ b/eth/filters/bor_filter.go @@ -128,6 +128,16 @@ func (f *BorBlockLogsFilter) unindexedLogs(ctx context.Context, end uint64) ([]* sprintLength := f.borConfig.CalculateSprint(uint64(f.begin)) for ; f.begin <= int64(end); f.begin = f.begin + int64(sprintLength) { + // Honor context cancellation (client disconnect or deadline) so a large + // range scan stops promptly instead of holding an RPC worker busy with + // work whose result the caller will never read. Mirrors the check in + // upstream go-ethereum eth/filters/filter.go's unindexedLogs. + select { + case <-ctx.Done(): + return logs, ctx.Err() + default: + } + header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin)) if header == nil || err != nil { return logs, err diff --git a/eth/filters/bor_filter_test.go b/eth/filters/bor_filter_test.go index b787cca138..9a306d095c 100644 --- a/eth/filters/bor_filter_test.go +++ b/eth/filters/bor_filter_test.go @@ -1,13 +1,18 @@ package filters import ( + "context" + "errors" "math/big" + "sync/atomic" "testing" + "time" "github.com/ethereum/go-ethereum/common" types "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/params" + "github.com/ethereum/go-ethereum/rpc" "go.uber.org/mock/gomock" ) @@ -154,6 +159,92 @@ func TestBorFilters(t *testing.T) { } } +// newCancellationTestFilter builds a bor range filter wired to a mock backend +// that counts per-block header reads inside the scan loop, returning the filter +// and a pointer to that counter. A test can then assert the loop bailed out +// early (counter == 0) instead of scanning the whole range. Only the per-block +// reads are counted: the single pre-loop head lookup uses rpc.LatestBlockNumber +// (negative), while loop reads use the concrete, non-negative block number. +func newCancellationTestFilter(t *testing.T) (*BorBlockLogsFilter, *int32) { + t.Helper() + + ctrl := gomock.NewController(t) + t.Cleanup(ctrl.Finish) + + db := NewMockDatabase(ctrl) + backend := NewMockBackend(ctrl) + + loopBlockReads := new(int32) + backend.EXPECT().ChainDb().Return(db).AnyTimes() + backend.EXPECT().HeaderByNumber(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, number rpc.BlockNumber) (*types.Header, error) { + if number >= 0 { + atomic.AddInt32(loopBlockReads, 1) + } + return newTestHeader(1), nil + }).AnyTimes() + backend.EXPECT().GetBorBlockReceipt(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() + + // Range large enough that an unguarded loop would iterate hundreds of times. + filter := NewBorBlockLogsRangeFilter(backend, params.TestChainConfig.Bor, 0, 4000, []common.Address{addr}, nil) + + return filter, loopBlockReads +} + +// TestBorFilterHonorsContextCancellation asserts that a range scan stops as soon +// as the request context is cancelled (e.g. the RPC client disconnected, or a +// deadline fired) instead of scanning the whole range and keeping an RPC worker +// busy with work whose result nobody will read. This mirrors the cancellation +// check upstream go-ethereum performs in eth/filters/filter.go's unindexedLogs. +func TestBorFilterHonorsContextCancellation(t *testing.T) { + t.Parallel() + + filter, loopBlockReads := newCancellationTestFilter(t) + + // Cancel up-front: a scan that honors cancellation must bail out before + // doing any per-block work. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + logs, err := filter.Logs(ctx) + + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got err=%v (loop ran %d block reads, ignoring cancellation)", + err, atomic.LoadInt32(loopBlockReads)) + } + if got := atomic.LoadInt32(loopBlockReads); got != 0 { + t.Fatalf("expected 0 in-loop block reads after honoring cancellation, got %d", got) + } + if len(logs) != 0 { + t.Fatalf("expected no logs on a cancelled scan, got %d", len(logs)) + } +} + +// TestBorFilterHonorsContextDeadline asserts the scan also stops once the +// context deadline has passed. Because the loop now observes the context, any +// deadline attached upstream (an operator-configured RPC timeout, or a deadline +// the caller sets) bounds the scan -- previously the loop ran to completion +// regardless of the deadline. +func TestBorFilterHonorsContextDeadline(t *testing.T) { + t.Parallel() + + filter, loopBlockReads := newCancellationTestFilter(t) + + // Deadline already in the past: the scan must stop before any per-block work. + ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Hour)) + defer cancel() + + _, err := filter.Logs(ctx) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected context.DeadlineExceeded, got err=%v (loop ran %d block reads, ignoring the deadline)", + err, atomic.LoadInt32(loopBlockReads)) + } + if got := atomic.LoadInt32(loopBlockReads); got != 0 { + t.Fatalf("expected 0 in-loop block reads after honoring the deadline, got %d", got) + } +} + func TestBorFilters_SkipOnBeginAtStateSync(t *testing.T) { t.Parallel()