From 01f3bffcfde328c0dc4caa91d67215e737afa84a Mon Sep 17 00:00:00 2001 From: Lucca Martins Date: Thu, 4 Jun 2026 12:15:50 -0300 Subject: [PATCH 1/2] fix(filters): honor context cancellation in bor block-log range scan BorBlockLogsFilter.unindexedLogs iterated the full block range without ever checking the request context, so a cancelled request (client disconnect or an expired deadline) kept scanning and held its RPC worker until the whole range was exhausted. Upstream go-ethereum eth/filters/filter.go performs this check at the top of its unindexedLogs loop; the bor copy omitted it. Add the same select{<-ctx.Done()} guard so a range scan stops promptly when the caller is gone, and add tests covering both cancellation and deadline. --- eth/filters/bor_filter.go | 10 ++++ eth/filters/bor_filter_test.go | 101 +++++++++++++++++++++++++++++++++ 2 files changed, 111 insertions(+) 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..05c7715876 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,102 @@ func TestBorFilters(t *testing.T) { } } +// 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() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + db := NewMockDatabase(ctrl) + backend := NewMockBackend(ctrl) + + testBorConfig := params.TestChainConfig.Bor + + // Count only the per-block reads inside the scan loop. The single pre-loop + // head lookup uses rpc.LatestBlockNumber (negative); loop reads use the + // concrete, non-negative block number. + var loopBlockReads 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() + + // Cancel up-front: a scan that honors cancellation must bail out before + // doing any per-block work. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + // Range large enough that an unguarded loop would iterate hundreds of times. + filter := NewBorBlockLogsRangeFilter(backend, testBorConfig, 0, 4000, []common.Address{addr}, nil) + + 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() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + db := NewMockDatabase(ctrl) + backend := NewMockBackend(ctrl) + + testBorConfig := params.TestChainConfig.Bor + + var loopBlockReads 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() + + // 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() + + filter := NewBorBlockLogsRangeFilter(backend, testBorConfig, 0, 4000, []common.Address{addr}, nil) + + _, 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() From e4027ce0d2795c103b51d25dbd16b650cf00f360 Mon Sep 17 00:00:00 2001 From: Lucca Martins Date: Mon, 13 Jul 2026 13:06:31 -0300 Subject: [PATCH 2/2] test(filters): dedup cancellation test setup into a helper Extract the shared mock-backend wiring used by the cancellation and deadline tests into newCancellationTestFilter, removing the ~20 duplicated setup lines that tripped SonarCloud's new-code duplication gate (38.7% > 3%). --- eth/filters/bor_filter_test.go | 74 +++++++++++++++------------------- 1 file changed, 32 insertions(+), 42 deletions(-) diff --git a/eth/filters/bor_filter_test.go b/eth/filters/bor_filter_test.go index 05c7715876..9a306d095c 100644 --- a/eth/filters/bor_filter_test.go +++ b/eth/filters/bor_filter_test.go @@ -159,51 +159,60 @@ func TestBorFilters(t *testing.T) { } } -// 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() +// 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) - defer ctrl.Finish() + t.Cleanup(ctrl.Finish) db := NewMockDatabase(ctrl) backend := NewMockBackend(ctrl) - testBorConfig := params.TestChainConfig.Bor - - // Count only the per-block reads inside the scan loop. The single pre-loop - // head lookup uses rpc.LatestBlockNumber (negative); loop reads use the - // concrete, non-negative block number. - var loopBlockReads int32 + 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) + 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() - // Range large enough that an unguarded loop would iterate hundreds of times. - filter := NewBorBlockLogsRangeFilter(backend, testBorConfig, 0, 4000, []common.Address{addr}, nil) - 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)) + err, atomic.LoadInt32(loopBlockReads)) } - if got := atomic.LoadInt32(&loopBlockReads); got != 0 { + 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 { @@ -219,38 +228,19 @@ func TestBorFilterHonorsContextCancellation(t *testing.T) { func TestBorFilterHonorsContextDeadline(t *testing.T) { t.Parallel() - ctrl := gomock.NewController(t) - defer ctrl.Finish() - - db := NewMockDatabase(ctrl) - backend := NewMockBackend(ctrl) - - testBorConfig := params.TestChainConfig.Bor - - var loopBlockReads 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() + 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() - filter := NewBorBlockLogsRangeFilter(backend, testBorConfig, 0, 4000, []common.Address{addr}, nil) - _, 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)) + err, atomic.LoadInt32(loopBlockReads)) } - if got := atomic.LoadInt32(&loopBlockReads); got != 0 { + if got := atomic.LoadInt32(loopBlockReads); got != 0 { t.Fatalf("expected 0 in-loop block reads after honoring the deadline, got %d", got) } }