eth/filters: move check for unfinalized queries (#1085)

* eth/filters: move check for unfinalized queries

* eth/filters, eth/api: update rpc numbers in unfinalized query mode (#1086)

* Align semantics of negative RPC numbers with upstream

* nits (#1089)

* nits

* sort conditions

* revert resolveSpecial changes

---------

Co-authored-by: Ceyhun Onur <[email protected]>

---------

Co-authored-by: Ceyhun Onur <[email protected]>

Author: Darioush Jalali

accounts/abi/bind/backends/simulated.go

@@ -768,7 +768,7 @@ func (b *SimulatedBackend) FilterLogs(ctx context.Context, query interfaces.Filt
 			to = query.ToBlock.Int64()
 		}
 		// Construct the range filter
-		filter, _ = b.filterSystem.NewRangeFilter(from, to, query.Addresses, query.Topics)
+		filter = b.filterSystem.NewRangeFilter(from, to, query.Addresses, query.Topics)
 	}
 	// Run the filter and return all the logs
 	logs, err := filter.Logs(ctx)
@@ -911,7 +911,7 @@ func (fb *filterBackend) EventMux() *event.TypeMux { panic("not supported") }
 
 func (fb *filterBackend) HeaderByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Header, error) {
 	switch number {
-	case rpc.PendingBlockNumber, rpc.AcceptedBlockNumber:
+	case rpc.PendingBlockNumber, rpc.FinalizedBlockNumber:
 		if block := fb.backend.acceptedBlock; block != nil {
 			return block.Header(), nil
 		}

eth/api.go

@@ -197,7 +197,11 @@ func (api *DebugAPI) DumpBlock(blockNr rpc.BlockNumber) (state.Dump, error) {
 	}
 	var header *types.Header
 	if blockNr.IsAccepted() {
-		header = api.eth.LastAcceptedBlock().Header()
+		if api.eth.APIBackend.isLatestAndAllowed(blockNr) {
+			header = api.eth.blockchain.CurrentHeader()
+		} else {
+			header = api.eth.LastAcceptedBlock().Header()
+		}
 	} else {
 		block := api.eth.blockchain.GetBlockByNumber(uint64(blockNr))
 		if block == nil {
@@ -241,7 +245,11 @@ func (api *DebugAPI) AccountRange(blockNrOrHash rpc.BlockNumberOrHash, start hex
 	if number, ok := blockNrOrHash.Number(); ok {
 		var header *types.Header
 		if number.IsAccepted() {
-			header = api.eth.LastAcceptedBlock().Header()
+			if api.eth.APIBackend.isLatestAndAllowed(number) {
+				header = api.eth.blockchain.CurrentHeader()
+			} else {
+				header = api.eth.LastAcceptedBlock().Header()
+			}
 		} else {
 			block := api.eth.blockchain.GetBlockByNumber(uint64(number))
 			if block == nil {

eth/api_backend.go

@@ -92,6 +92,9 @@ func (b *EthAPIBackend) HeaderByNumber(ctx context.Context, number rpc.BlockNumb
 	// identically.
 	acceptedBlock := b.eth.LastAcceptedBlock()
 	if number.IsAccepted() {
+		if b.isLatestAndAllowed(number) {
+			return b.eth.blockchain.CurrentHeader(), nil
+		}
 		return acceptedBlock.Header(), nil
 	}
 
@@ -156,6 +159,10 @@ func (b *EthAPIBackend) BlockByNumber(ctx context.Context, number rpc.BlockNumbe
 	// identically.
 	acceptedBlock := b.eth.LastAcceptedBlock()
 	if number.IsAccepted() {
+		if b.isLatestAndAllowed(number) {
+			header := b.eth.blockchain.CurrentBlock()
+			return b.eth.blockchain.GetBlock(header.Hash(), header.Number.Uint64()), nil
+		}
 		return acceptedBlock, nil
 	}
 
@@ -486,3 +493,7 @@ func (b *EthAPIBackend) StateAtTransaction(ctx context.Context, block *types.Blo
 func (b *EthAPIBackend) MinRequiredTip(ctx context.Context, header *types.Header) (*big.Int, error) {
 	return dummy.MinRequiredTip(b.ChainConfig(), header)
 }
+
+func (b *EthAPIBackend) isLatestAndAllowed(number rpc.BlockNumber) bool {
+	return number.IsLatest() && b.IsAllowUnfinalizedQueries()
+}

eth/filters/api.go

@@ -439,11 +439,7 @@ func (api *FilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*type
 			end = crit.ToBlock.Int64()
 		}
 		// Construct the range filter
-		var err error
-		filter, err = api.sys.NewRangeFilter(begin, end, crit.Addresses, crit.Topics)
-		if err != nil {
-			return nil, err
-		}
+		filter = api.sys.NewRangeFilter(begin, end, crit.Addresses, crit.Topics)
 	}
 	// Run the filter and return all the logs
 	logs, err := filter.Logs(ctx)
@@ -497,11 +493,7 @@ func (api *FilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*types.Lo
 			end = f.crit.ToBlock.Int64()
 		}
 		// Construct the range filter
-		var err error
-		filter, err = api.sys.NewRangeFilter(begin, end, f.crit.Addresses, f.crit.Topics)
-		if err != nil {
-			return nil, err
-		}
+		filter = api.sys.NewRangeFilter(begin, end, f.crit.Addresses, f.crit.Topics)
 	}
 	// Run the filter and return all the logs
 	logs, err := filter.Logs(ctx)

eth/filters/bench_test.go

@@ -38,7 +38,6 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/bitutil"
 	"github.com/ethereum/go-ethereum/ethdb"
-	"github.com/stretchr/testify/require"
 )
 
 func BenchmarkBloomBits512(b *testing.B) {
@@ -147,8 +146,7 @@ func benchmarkBloomBits(b *testing.B, sectionSize uint64) {
 		var addr common.Address
 		addr[0] = byte(i)
 		addr[1] = byte(i / 256)
-		filter, err := sys.NewRangeFilter(0, int64(cnt*sectionSize-1), []common.Address{addr}, nil)
-		require.NoError(b, err)
+		filter := sys.NewRangeFilter(0, int64(cnt*sectionSize-1), []common.Address{addr}, nil)
 		if _, err := filter.Logs(context.Background()); err != nil {
 			b.Error("filter.Logs error:", err)
 		}
@@ -191,8 +189,7 @@ func BenchmarkNoBloomBits(b *testing.B) {
 
 	b.Log("Running filter benchmarks...")
 	start := time.Now()
-	filter, err := sys.NewRangeFilter(0, int64(*headNum), []common.Address{{}}, nil)
-	require.NoError(b, err)
+	filter := sys.NewRangeFilter(0, int64(*headNum), []common.Address{{}}, nil)
 	filter.Logs(context.Background())
 	d := time.Since(start)
 	b.Log("Finished running filter benchmarks")

eth/filters/filter.go

@@ -53,10 +53,7 @@ type Filter struct {
 
 // NewRangeFilter creates a new filter which uses a bloom filter on blocks to
 // figure out whether a particular block is interesting or not.
-func (sys *FilterSystem) NewRangeFilter(begin, end int64, addresses []common.Address, topics [][]common.Hash) (*Filter, error) {
-	allowUnfinalizedQueries := sys.backend.IsAllowUnfinalizedQueries()
-	acceptedBlock := sys.backend.LastAcceptedBlock()
-
+func (sys *FilterSystem) NewRangeFilter(begin, end int64, addresses []common.Address, topics [][]common.Hash) *Filter {
 	// Flatten the address and topic filter clauses into a single bloombits filter
 	// system. Since the bloombits are not positional, nil topics are permitted,
 	// which get flattened into a nil byte slice.
@@ -77,24 +74,14 @@ func (sys *FilterSystem) NewRangeFilter(begin, end int64, addresses []common.Add
 	}
 	size, _ := sys.backend.BloomStatus()
 
-	if !allowUnfinalizedQueries && acceptedBlock != nil {
-		lastAccepted := acceptedBlock.Number().Int64()
-		if begin >= 0 && begin > lastAccepted {
-			return nil, fmt.Errorf("requested from block %d after last accepted block %d", begin, lastAccepted)
-		}
-		if end >= 0 && end > lastAccepted {
-			return nil, fmt.Errorf("requested to block %d after last accepted block %d", end, lastAccepted)
-		}
-	}
-
 	// Create a generic filter and convert it into a range filter
 	filter := newFilter(sys, addresses, topics)
 
 	filter.matcher = bloombits.NewMatcher(size, filters)
 	filter.begin = begin
 	filter.end = end
 
-	return filter, nil
+	return filter
 }
 
 // NewBlockFilter creates a new filter which directly inspects the contents of
@@ -130,6 +117,21 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
 		}
 		return f.blockLogs(ctx, header)
 	}
+
+	// Disallow blocks past the last accepted block if the backend does not
+	// allow unfinalized queries.
+	allowUnfinalizedQueries := f.sys.backend.IsAllowUnfinalizedQueries()
+	acceptedBlock := f.sys.backend.LastAcceptedBlock()
+	if !allowUnfinalizedQueries && acceptedBlock != nil {
+		lastAccepted := acceptedBlock.Number().Int64()
+		if f.begin >= 0 && f.begin > lastAccepted {
+			return nil, fmt.Errorf("requested from block %d after last accepted block %d", f.begin, lastAccepted)
+		}
+		if f.end >= 0 && f.end > lastAccepted {
+			return nil, fmt.Errorf("requested to block %d after last accepted block %d", f.end, lastAccepted)
+		}
+	}
+
 	// Short-cut if all we care about is pending logs
 	if f.begin == rpc.PendingBlockNumber.Int64() {
 		if f.end != rpc.PendingBlockNumber.Int64() {
@@ -149,14 +151,38 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
 		return nil, nil
 	}
 	var (
-		head = header.Number.Uint64()
-		end  = uint64(f.end)
+		head = header.Number.Int64()
 	)
-	if f.begin < 0 {
-		f.begin = int64(head)
+
+	resolveSpecial := func(number int64) (int64, error) {
+		var hdr *types.Header
+		switch number {
+		case rpc.LatestBlockNumber.Int64():
+			return head, nil
+		case 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
+			return head, nil
+		case rpc.FinalizedBlockNumber.Int64():
+			hdr, _ = f.sys.backend.HeaderByNumber(ctx, rpc.FinalizedBlockNumber)
+			if hdr == nil {
+				return 0, errors.New("finalized header not found")
+			}
+		case rpc.SafeBlockNumber.Int64():
+			hdr, _ = f.sys.backend.HeaderByNumber(ctx, rpc.SafeBlockNumber)
+			if hdr == nil {
+				return 0, errors.New("safe header not found")
+			}
+		default:
+			return number, nil
+		}
+		return hdr.Number.Int64(), nil
 	}
-	if f.end < 0 {
-		end = head
+	if f.begin, err = resolveSpecial(f.begin); err != nil {
+		return nil, err
+	}
+	if f.end, err = resolveSpecial(f.end); err != nil {
+		return nil, err
 	}
 
 	// When querying unfinalized data without a populated end block, it is
@@ -165,18 +191,19 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
 	// We error in this case to prevent a bad UX where the caller thinks there
 	// are no logs from the specified beginning to end (when in reality there may
 	// be some).
-	if end < uint64(f.begin) {
-		return nil, fmt.Errorf("begin block %d is greater than end block %d", f.begin, end)
+	if f.end < f.begin {
+		return nil, fmt.Errorf("begin block %d is greater than end block %d", f.begin, f.end)
 	}
 
 	// If the requested range of blocks exceeds the maximum number of blocks allowed by the backend
 	// return an error instead of searching for the logs.
-	if maxBlocks := f.sys.backend.GetMaxBlocksPerRequest(); int64(end)-f.begin >= maxBlocks && maxBlocks > 0 {
-		return nil, fmt.Errorf("requested too many blocks from %d to %d, maximum is set to %d", f.begin, int64(end), maxBlocks)
+	if maxBlocks := f.sys.backend.GetMaxBlocksPerRequest(); f.end-f.begin >= maxBlocks && maxBlocks > 0 {
+		return nil, fmt.Errorf("requested too many blocks from %d to %d, maximum is set to %d", f.begin, f.end, maxBlocks)
 	}
 	// Gather all indexed logs, and finish with non indexed ones
 	var (
 		logs           []*types.Log
+		end            = uint64(f.end)
 		size, sections = f.sys.backend.BloomStatus()
 	)
 	if indexed := sections * size; indexed > uint64(f.begin) {

eth/filters/filter_system_test.go

@@ -95,13 +95,26 @@ func (b *testBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumbe
 		num  uint64
 	)
 	switch blockNr {
-	case rpc.LatestBlockNumber, rpc.AcceptedBlockNumber:
+	case rpc.FinalizedBlockNumber:
+		var err error
+		hash, err = rawdb.ReadAcceptorTip(b.db)
+		if err != nil {
+			return nil, err
+		}
+		number := rawdb.ReadHeaderNumber(b.db, hash)
+		if number == nil {
+			return nil, nil
+		}
+		num = *number
+	case rpc.LatestBlockNumber, rpc.PendingBlockNumber:
 		hash = rawdb.ReadHeadBlockHash(b.db)
 		number := rawdb.ReadHeaderNumber(b.db, hash)
 		if number == nil {
 			return nil, nil
 		}
 		num = *number
+	case rpc.SafeBlockNumber:
+		return nil, errors.New("safe block not found")
 	default:
 		num = uint64(blockNr)
 		hash = rawdb.ReadCanonicalHash(b.db, num)

eth/filters/filter_test.go

@@ -63,9 +63,9 @@ func BenchmarkFilters(b *testing.B) {
 		addr4   = common.BytesToAddress([]byte("random addresses please"))
 
 		gspec = &core.Genesis{
-			Config:  params.TestChainConfig,
 			Alloc:   core.GenesisAlloc{addr1: {Balance: big.NewInt(1000000)}},
 			BaseFee: big.NewInt(1),
+			Config:  params.TestChainConfig,
 		}
 	)
 	defer db.Close()
@@ -94,6 +94,7 @@ func BenchmarkFilters(b *testing.B) {
 	// and then import blocks. TODO(rjl493456442) try to get rid of the
 	// manual database writes.
 	gspec.MustCommit(db)
+
 	for i, block := range chain {
 		rawdb.WriteBlock(db, block)
 		rawdb.WriteCanonicalHash(db, block.Hash(), block.NumberU64())
@@ -102,8 +103,7 @@ func BenchmarkFilters(b *testing.B) {
 	}
 	b.ResetTimer()
 
-	filter, err := sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), []common.Address{addr1, addr2, addr3, addr4}, nil)
-	require.NoError(b, err)
+	filter := sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), []common.Address{addr1, addr2, addr3, addr4}, nil)
 
 	for i := 0; i < b.N; i++ {
 		filter.begin = 0
@@ -191,8 +191,12 @@ func TestFilters(t *testing.T) {
 		rawdb.WriteReceipts(db, block.Hash(), block.NumberU64(), receipts[i])
 	}
 
-	filter, err := sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), []common.Address{addr}, [][]common.Hash{{hash1, hash2, hash3, hash4}})
+	// Set block 998 as Finalized (-3)
+	// rawdb.WriteFinalizedBlockHash(db, chain[998].Hash())
+	err = rawdb.WriteAcceptorTip(db, chain[998].Hash())
 	require.NoError(t, err)
+
+	filter := sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), []common.Address{addr}, [][]common.Hash{{hash1, hash2, hash3, hash4}})
 	logs, _ := filter.Logs(context.Background())
 	if len(logs) != 4 {
 		t.Error("expected 4 log, got", len(logs))
@@ -203,46 +207,39 @@ func TestFilters(t *testing.T) {
 		wantHashes []common.Hash
 	}{
 		{
-			mustNewRangeFilter(t, sys, 900, 999, []common.Address{addr}, [][]common.Hash{{hash3}}),
+			sys.NewRangeFilter(900, 999, []common.Address{addr}, [][]common.Hash{{hash3}}),
 			[]common.Hash{hash3},
 		}, {
-			mustNewRangeFilter(t, sys, 990, int64(rpc.LatestBlockNumber), []common.Address{addr}, [][]common.Hash{{hash3}}),
+			sys.NewRangeFilter(990, int64(rpc.LatestBlockNumber), []common.Address{addr}, [][]common.Hash{{hash3}}),
 			[]common.Hash{hash3},
 		}, {
-			mustNewRangeFilter(t, sys, 1, 10, nil, [][]common.Hash{{hash1, hash2}}),
+			sys.NewRangeFilter(1, 10, nil, [][]common.Hash{{hash1, hash2}}),
 			[]common.Hash{hash1, hash2},
 		}, {
-			mustNewRangeFilter(t, sys, 0, int64(rpc.LatestBlockNumber), nil, [][]common.Hash{{common.BytesToHash([]byte("fail"))}}),
+			sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), nil, [][]common.Hash{{common.BytesToHash([]byte("fail"))}}),
 			nil,
 		}, {
-			mustNewRangeFilter(t, sys, 0, int64(rpc.LatestBlockNumber), []common.Address{common.BytesToAddress([]byte("failmenow"))}, nil),
+			sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), []common.Address{common.BytesToAddress([]byte("failmenow"))}, nil),
 			nil,
 		}, {
-			mustNewRangeFilter(t, sys, 0, int64(rpc.LatestBlockNumber), nil, [][]common.Hash{{common.BytesToHash([]byte("fail"))}, {hash1}}),
+			sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), nil, [][]common.Hash{{common.BytesToHash([]byte("fail"))}, {hash1}}),
 			nil,
 		}, {
-			mustNewRangeFilter(t, sys, int64(rpc.LatestBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), []common.Hash{hash4},
+			sys.NewRangeFilter(int64(rpc.LatestBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), []common.Hash{hash4},
 		}, {
-			// Note: modified from go-ethereum since we don't have FinalizedBlock
-			mustNewRangeFilter(t, sys, int64(rpc.AcceptedBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), []common.Hash{hash4},
+			sys.NewRangeFilter(int64(rpc.FinalizedBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), []common.Hash{hash3, hash4},
 		}, {
-			// Note: modified from go-ethereum since we don't have FinalizedBlock
-			mustNewRangeFilter(t, sys, int64(rpc.AcceptedBlockNumber), int64(rpc.AcceptedBlockNumber), nil, nil), []common.Hash{hash4},
+			sys.NewRangeFilter(int64(rpc.FinalizedBlockNumber), int64(rpc.FinalizedBlockNumber), nil, nil), []common.Hash{hash3},
 		}, {
-			// Note: modified from go-ethereum since we don't have FinalizedBlock
-			mustNewRangeFilter(t, sys, int64(rpc.LatestBlockNumber), -3, nil, nil), []common.Hash{hash4},
+			sys.NewRangeFilter(int64(rpc.LatestBlockNumber), int64(rpc.FinalizedBlockNumber), nil, nil), nil,
 		}, {
-			// Note: modified from go-ethereum since we don't have SafeBlock
-			mustNewRangeFilter(t, sys, int64(rpc.AcceptedBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), []common.Hash{hash4},
+			sys.NewRangeFilter(int64(rpc.SafeBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), nil,
 		}, {
-			// Note: modified from go-ethereum since we don't have SafeBlock
-			mustNewRangeFilter(t, sys, int64(rpc.AcceptedBlockNumber), int64(rpc.AcceptedBlockNumber), nil, nil), []common.Hash{hash4},
+			sys.NewRangeFilter(int64(rpc.SafeBlockNumber), int64(rpc.SafeBlockNumber), nil, nil), nil,
 		}, {
-			// Note: modified from go-ethereum since we don't have SafeBlock
-			mustNewRangeFilter(t, sys, int64(rpc.LatestBlockNumber), int64(rpc.AcceptedBlockNumber), nil, nil), []common.Hash{hash4},
-		},
-		{
-			mustNewRangeFilter(t, sys, int64(rpc.PendingBlockNumber), int64(rpc.PendingBlockNumber), nil, nil), nil,
+			sys.NewRangeFilter(int64(rpc.LatestBlockNumber), int64(rpc.SafeBlockNumber), nil, nil), nil,
+		}, {
+			sys.NewRangeFilter(int64(rpc.PendingBlockNumber), int64(rpc.PendingBlockNumber), nil, nil), nil,
 		},
 	} {
 		logs, _ := tc.f.Logs(context.Background())
@@ -261,10 +258,3 @@ func TestFilters(t *testing.T) {
 		}
 	}
 }
-
-func mustNewRangeFilter(t *testing.T, sys *FilterSystem, begin, end int64, addresses []common.Address, topics [][]common.Hash) *Filter {
-	t.Helper()
-	f, err := sys.NewRangeFilter(begin, end, addresses, topics)
-	require.NoError(t, err)
-	return f
-}