Skip to content

[WIP] eth/filters: remove support for pending logs #28623 #29574 #2015

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gzliudan wants to merge 1 commit into
base: dev-upgrade
Choose a base branch
from
+103 −637
Open

[WIP] eth/filters: remove support for pending logs #28623 #29574 #2015

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions accounts/abi/bind/backends/simulated.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,7 @@ func NewXDCSimulatedBackend(alloc types.GenesisAlloc, gasLimit uint64, chainConf

filterBackend := &filterBackend{database, blockchain, backend}
backend.filterSystem = filters.NewFilterSystem(filterBackend, filters.Config{})
backend.events = filters.NewEventSystem(backend.filterSystem, false)
backend.events = filters.NewEventSystem(backend.filterSystem)

blockchain.Client = backend

Expand Down Expand Up @@ -181,7 +181,7 @@ func NewSimulatedBackend(alloc types.GenesisAlloc, gasLimit uint64) *SimulatedBa

filterBackend := &filterBackend{database, blockchain, backend}
backend.filterSystem = filters.NewFilterSystem(filterBackend, filters.Config{})
backend.events = filters.NewEventSystem(backend.filterSystem, false)
backend.events = filters.NewEventSystem(backend.filterSystem)

header := backend.blockchain.CurrentBlock()
block := backend.blockchain.GetBlock(header.Hash(), header.Number.Uint64())
Expand Down Expand Up @@ -834,7 +834,7 @@ func (b *SimulatedBackend) FilterLogs(ctx context.Context, query ethereum.Filter
if query.FromBlock != nil {
from = query.FromBlock.Int64()
}
to := int64(-1)
to := int64(rpc.LatestBlockNumber)
if query.ToBlock != nil {
to = query.ToBlock.Int64()
}
Expand Down
2 changes: 1 addition & 1 deletion cmd/utils/flags.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1862,7 +1862,7 @@ func RegisterFilterAPI(stack *node.Node, backend ethapi.Backend, ethcfg *ethconf
})
stack.RegisterAPIs([]rpc.API{{
Namespace: "eth",
Service: filters.NewFilterAPI(filterSystem, false),
Service: filters.NewFilterAPI(filterSystem),
}})
return filterSystem
}
2 changes: 1 addition & 1 deletion eth/backend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,7 +426,7 @@ func (e *Ethereum) APIs() []rpc.API {
Service: downloader.NewDownloaderAPI(e.protocolManager.downloader, e.eventMux),
}, {
Namespace: "eth",
Service: filters.NewFilterAPI(filters.NewFilterSystem(e.APIBackend, filters.Config{LogCacheSize: e.config.FilterLogCacheSize}), false),
Service: filters.NewFilterAPI(filters.NewFilterSystem(e.APIBackend, filters.Config{LogCacheSize: e.config.FilterLogCacheSize})),
}, {
Namespace: "admin",
Service: NewAdminAPI(e),
Expand Down
21 changes: 11 additions & 10 deletions eth/filters/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,14 @@ import (
)

var (
errInvalidTopic = invalidParamsErr("invalid topic(s)")
errInvalidBlockRange = invalidParamsErr("invalid block range params")
errBlockHashWithRange = invalidParamsErr("can't specify fromBlock/toBlock with blockHash")
errUnknownBlock = errors.New("unknown block")
errFilterNotFound = errors.New("filter not found")
errExceedMaxTopics = errors.New("exceed max topics")
errExceedLogQueryLimit = errors.New("exceed max addresses or topics per search position")
errInvalidTopic = invalidParamsErr("invalid topic(s)")
errInvalidBlockRange = invalidParamsErr("invalid block range params")
errBlockHashWithRange = invalidParamsErr("can't specify fromBlock/toBlock with blockHash")
errPendingLogsUnsupported = invalidParamsErr("pending logs are not supported")
errUnknownBlock = errors.New("unknown block")
errFilterNotFound = errors.New("filter not found")
errExceedMaxTopics = errors.New("exceed max topics")
errExceedLogQueryLimit = errors.New("exceed max addresses or topics per search position")
)

type invalidParamsError struct {
Expand Down Expand Up @@ -83,10 +84,10 @@ type FilterAPI struct {
}

// NewFilterAPI returns a new FilterAPI instance.
func NewFilterAPI(system *FilterSystem, lightMode bool) *FilterAPI {
func NewFilterAPI(system *FilterSystem) *FilterAPI {
api := &FilterAPI{
sys: system,
events: NewEventSystem(system, lightMode),
events: NewEventSystem(system),
filters: make(map[rpc.ID]*filter),
timeout: system.cfg.Timeout,
logQueryLimit: system.cfg.LogQueryLimit,
Expand Down Expand Up @@ -481,7 +482,7 @@ func (api *FilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
txs := f.txs
f.txs = nil
return txs, nil
case LogsSubscription, MinedAndPendingLogsSubscription:
case LogsSubscription:
logs := f.logs
f.logs = nil
return returnLogs(logs), nil
Expand Down
80 changes: 28 additions & 52 deletions eth/filters/filter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,71 +110,63 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
}
return f.blockLogs(ctx, header)
}
var (
beginPending = f.begin == rpc.PendingBlockNumber.Int64()
endPending = f.end == rpc.PendingBlockNumber.Int64()
)

// special case for pending logs
if beginPending && !endPending {
return nil, errors.New("invalid block range")
// Disallow pending logs.
if f.begin == rpc.PendingBlockNumber.Int64() || f.end == rpc.PendingBlockNumber.Int64() {
return nil, errPendingLogsUnsupported
}

// Short-cut if all we care about is pending logs
if beginPending && endPending {
return f.pendingLogs(), nil
}

resolveSpecial := func(number int64) (int64, error) {
var hdr *types.Header
resolveSpecial := func(number int64) (uint64, error) {
switch number {
case rpc.LatestBlockNumber.Int64(), rpc.PendingBlockNumber.Int64():
// we should return head here since we've already captured
// that we need to get the pending logs in the pending boolean above
hdr, _ = f.sys.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber)
case rpc.LatestBlockNumber.Int64():
hdr, _ := f.sys.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber)
if hdr == nil {
return 0, errors.New("latest header not found")
}
return hdr.Number.Uint64(), nil
case rpc.FinalizedBlockNumber.Int64():
hdr, _ = f.sys.backend.HeaderByNumber(ctx, rpc.FinalizedBlockNumber)
hdr, _ := f.sys.backend.HeaderByNumber(ctx, rpc.FinalizedBlockNumber)
if hdr == nil {
return 0, errors.New("committed header not found")
return 0, errors.New("finalized header not found")
}
return hdr.Number.Uint64(), nil
case rpc.EarliestBlockNumber.Int64():
hdr, _ := f.sys.backend.HeaderByNumber(ctx, rpc.EarliestBlockNumber)
if hdr == nil {
return 0, errors.New("earliest header not found")
}
return hdr.Number.Uint64(), nil
default:
return number, nil
if number < 0 {
return 0, errors.New("negative block number")
}
return uint64(number), nil
}
return hdr.Number.Int64(), nil
}

var err error
// range query need to resolve the special begin/end block number
if f.begin, err = resolveSpecial(f.begin); err != nil {
begin, err := resolveSpecial(f.begin)
if err != nil {
return nil, err
}
if f.end, err = resolveSpecial(f.end); err != nil {
end, err := resolveSpecial(f.end)
if err != nil {
return nil, err
}
if f.rangeLimit != 0 && (uint64(f.end)-uint64(f.begin)) > f.rangeLimit {
if f.rangeLimit != 0 && (end-begin) > f.rangeLimit {
return nil, fmt.Errorf("exceed maximum block range: %d", f.rangeLimit)
}

f.begin = int64(begin)
f.end = int64(end)
logChan, errChan := f.rangeLogsAsync(ctx)
var logs []*types.Log
for {
select {
case log := <-logChan:
logs = append(logs, log)
case err := <-errChan:
if err != nil {
// if an error occurs during extraction, we do return the extracted data
return logs, err
}
// Append the pending ones
if endPending {
pendingLogs := f.pendingLogs()
logs = append(logs, pendingLogs...)
}
return logs, nil
return logs, err
Comment on lines 147 to +169
Copy link

@coderabbitai coderabbitai bot Jan 29, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Add explicit end ≥ begin validation before range arithmetic.

Right now, if end < begin and rangeLimit == 0, the filter silently returns empty results; when rangeLimit != 0, (end-begin) underflows and yields a misleading “exceed maximum block range” error. Consider returning an invalid-range error up front (consistent with SubscribeLogs).

💡 Suggested fix 
  end, err := resolveSpecial(f.end)
  if err != nil {
      return nil, err
  }
+ if end < begin {
+     return nil, errInvalidBlockRange
+ }
  if f.rangeLimit != 0 && (end-begin) > f.rangeLimit {
      return nil, fmt.Errorf("exceed maximum block range: %d", f.rangeLimit)
  }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
// range query need to resolve the special begin/end block number
if f.begin, err = resolveSpecial(f.begin); err != nil {
begin, err := resolveSpecial(f.begin)
if err != nil {
return nil, err
}
if f.end, err = resolveSpecial(f.end); err != nil {
end, err := resolveSpecial(f.end)
if err != nil {
return nil, err
}
if f.rangeLimit != 0 && (uint64(f.end)-uint64(f.begin)) > f.rangeLimit {
if f.rangeLimit != 0 && (end-begin) > f.rangeLimit {
return nil, fmt.Errorf("exceed maximum block range: %d", f.rangeLimit)
}
f.begin = int64(begin)
f.end = int64(end)
logChan, errChan := f.rangeLogsAsync(ctx)
var logs []*types.Log
for {
select {
case log := <-logChan:
logs = append(logs, log)
case err := <-errChan:
if err != nil {
// if an error occurs during extraction, we do return the extracted data
return logs, err
}
// Append the pending ones
if endPending {
pendingLogs := f.pendingLogs()
logs = append(logs, pendingLogs...)
}
return logs, nil
return logs, err
// range query need to resolve the special begin/end block number
begin, err := resolveSpecial(f.begin)
if err != nil {
return nil, err
}
end, err := resolveSpecial(f.end)
if err != nil {
return nil, err
}
if end < begin {
return nil, errInvalidBlockRange
}
if f.rangeLimit != 0 && (end-begin) > f.rangeLimit {
return nil, fmt.Errorf("exceed maximum block range: %d", f.rangeLimit)
}
f.begin = int64(begin)
f.end = int64(end)
logChan, errChan := f.rangeLogsAsync(ctx)
var logs []*types.Log
for {
select {
case log := <-logChan:
logs = append(logs, log)
case err := <-errChan:
return logs, err
🤖 Prompt for AI Agents 
In `@eth/filters/filter.go` around lines 147 - 169, Add an explicit validation in
the filter range resolution: after calling resolveSpecial for f.begin and f.end
(before assigning f.begin/f.end and before any arithmetic or rangeLimit checks
in the function that calls rangeLogsAsync), check if end < begin and return a
clear invalid-range error (consistent with SubscribeLogs) instead of proceeding;
update the code paths around resolveSpecial, the rangeLimit check, and the call
sites (e.g., where f.begin/f.end are set and rangeLogsAsync is invoked) to
ensure the new validation prevents silent empty results and underflowed
subtraction when computing (end - begin).

}
}
}
Expand Down Expand Up @@ -332,22 +324,6 @@ func (f *Filter) checkMatches(ctx context.Context, header *types.Header) ([]*typ
return logs, nil
}

// pendingLogs returns the logs matching the filter criteria within the pending block.
func (f *Filter) pendingLogs() []*types.Log {
block, receipts := f.sys.backend.PendingBlockAndReceipts()
if block == nil {
return nil
}
if bloomFilter(block.Bloom(), f.addresses, f.topics) {
var unfiltered []*types.Log
for _, r := range receipts {
unfiltered = append(unfiltered, r.Logs...)
}
return filterLogs(unfiltered, nil, nil, f.addresses, f.topics)
}
return nil
}

func includes(addresses []common.Address, a common.Address) bool {
for _, addr := range addresses {
if addr == a {
Expand Down
Loading