resolve conflicts

eth/filters/api.go

@@ -37,17 +37,30 @@ import (
 )
 
 var (
-	errInvalidTopic       = errors.New("invalid topic(s)")
+	errInvalidTopic           = errors.New("invalid topic(s)")
-	errFilterNotFound     = errors.New("filter not found")
+	errFilterNotFound         = errors.New("filter not found")
-	errConnectDropped     = errors.New("connection dropped")
+	errConnectDropped         = errors.New("connection dropped")
-	errInvalidToBlock     = errors.New("log subscription does not support history block range")
+	errInvalidToBlock         = errors.New("log subscription does not support history block range")
-	errInvalidFromBlock   = errors.New("from block can be only a number, or \"safe\", or \"finalized\"")
+	errInvalidFromBlock       = errors.New("from block can be only a number, or \"safe\", or \"finalized\"")
-	errClientUnsubscribed = errors.New("client unsubscribed")
+	errClientUnsubscribed     = errors.New("client unsubscribed")
+	errExceedMaxTopics        = errors.New("exceeds max topics")
+	errExceedMaxAddresses     = errors.New("exceeds max addresses")
+	errBlockHashWithRange     = errors.New("cannot specify block hash with block range")
+	errInvalidBlockRange      = errors.New("invalid block range")
+	errUnknownBlock           = errors.New("unknown block")
+	errPendingLogsUnsupported = errors.New("pending logs are not supported")
+	errInvalidFromTo          = errors.New("invalid from and to block combination: from > to")
 )
 
 const (
 	// maxTrackedBlocks is the number of block hashes that will be tracked by subscription.
 	maxTrackedBlocks = 32 * 1024
+	// maxTopics is the maximum number of topics that can be specified in a filter
+	maxTopics = 4
+	// maxSubTopics is the maximum number of sub-topics that can be specified in a single topic position
+	maxSubTopics = 1000
+	// maxAddresses is the maximum number of addresses that can be specified in a filter
+	maxAddresses = 1000
 )
 
 // filter is a helper struct that holds meta information over the filter type
@@ -261,7 +274,6 @@ func (api *FilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
 // used in unit testing.
 type notifier interface {
 	Notify(id rpc.ID, data interface{}) error
-	Closed() <-chan interface{}
 }
 
 // Logs creates a subscription that fires for all historical
@@ -322,9 +334,6 @@ func (api *FilterAPI) liveLogs(notifier notifier, rpcSub *rpc.Subscription, crit
 			case <-rpcSub.Err(): // client send an unsubscribe request
 				logsSub.Unsubscribe()
 				return
-			case <-notifier.Closed(): // connection dropped
-				logsSub.Unsubscribe()
-				return
 			}
 		}
 	}()
@@ -448,8 +457,6 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
 
 			case <-rpcSub.Err(): // client send an unsubscribe request
 				return
-			case <-notifier.Closed(): // connection dropped
-				return
 			}
 		}
 	}()
@@ -462,20 +469,36 @@ func (api *FilterAPI) doHistLogs(ctx context.Context, from int64, addrs []common
 	// Fetch logs from a range of blocks.
 	fetchRange := func(from, to int64) error {
 		f := api.sys.NewRangeFilter(from, to, addrs, topics)
-		logsCh, errChan := f.rangeLogsAsync(ctx)
+		logs, err := f.Logs(ctx)
-		for {
+		if err != nil {
-			select {
+			return err
-			case logs := <-logsCh:
+		}
+		// Group logs by block and send them in batches
+		var currentBlockLogs []*types.Log
+		var currentBlockNumber uint64
+
+		for _, log := range logs {
+			if currentBlockNumber != log.BlockNumber && len(currentBlockLogs) > 0 {
+				// Send the current block's logs
 				select {
-				case histLogs <- logs:
+				case histLogs <- currentBlockLogs:
 				case <-ctx.Done():
-					// Flush out all logs until the range filter voluntarily exits.
+					return ctx.Err()
-					continue
 				}
-			case err := <-errChan:
+				currentBlockLogs = nil
-				return err
+			}
+			currentBlockLogs = append(currentBlockLogs, log)
+			currentBlockNumber = log.BlockNumber
+		}
+		// Send remaining logs if any
+		if len(currentBlockLogs) > 0 {
+			select {
+			case histLogs <- currentBlockLogs:
+			case <-ctx.Done():
+				return ctx.Err()
 			}
 		}
+		return nil
 	}
 
 	for {

eth/filters/filter_system.go

@@ -39,10 +39,6 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
-var (
-	errInvalidFromTo = errors.New("invalid from and to block combination: from > to")
-)
-
 // Config represents the configuration of the filter system.
 type Config struct {
 	LogCacheSize int           // maximum number of cached blocks (default: 32)
@@ -156,6 +152,8 @@ const (
 	UnknownSubscription Type = iota
 	// LogsSubscription queries for new or removed (chain reorg) logs
 	LogsSubscription
+	// MinedAndPendingLogsSubscription queries for both mined and pending logs
+	MinedAndPendingLogsSubscription
 	// PendingTransactionsSubscription queries for pending transactions entering
 	// the pending state
 	PendingTransactionsSubscription

eth/filters/filter_system_test.go

@@ -32,7 +32,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/filtermaps"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
@@ -40,7 +39,7 @@ import (
 	logger "github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/ethereum/go-ethereum/trie"
+	"github.com/ethereum/go-ethereum/triedb"
 )
 
 type testBackend struct {
@@ -185,7 +184,34 @@ func (b *testBackend) forwardLogEvents(logCh chan []*types.Log, removedLogCh cha
 	}()
 }
 
-func newTestFilterSystem(t testing.TB, db ethdb.Database, cfg Config) (*testBackend, *FilterSystem) {
+func (b *testBackend) startFilterMaps(history uint64, disabled bool, params filtermaps.Params) {
+	head := b.CurrentBlock()
+	chainView := filtermaps.NewChainView(b, head.Number.Uint64(), head.Hash())
+	config := filtermaps.Config{
+		History:        history,
+		Disabled:       disabled,
+		ExportFileName: "",
+	}
+	b.fm, _ = filtermaps.NewFilterMaps(b.db, chainView, 0, 0, params, config)
+	b.fm.Start()
+	b.fm.WaitIdle()
+}
+
+func (b *testBackend) stopFilterMaps() {
+	b.fm.Stop()
+	b.fm = nil
+}
+
+func (b *testBackend) setPending(block *types.Block, receipts types.Receipts) {
+	b.pendingBlock = block
+	b.pendingReceipts = receipts
+}
+
+func (b *testBackend) HistoryPruningCutoff() uint64 {
+	return 0
+}
+
+func newTestFilterSystem(db ethdb.Database, cfg Config) (*testBackend, *FilterSystem) {
 	backend := &testBackend{db: db}
 	sys := NewFilterSystem(backend, cfg)
 	return backend, sys
@@ -717,6 +743,7 @@ func TestLogsSubscription(t *testing.T) {
 
 	var (
 		db       = rawdb.NewMemoryDatabase()
+		tdb      = triedb.NewDatabase(db, triedb.HashDefaults)
 		signer   = types.HomesteadSigner{}
 		key, _   = crypto.GenerateKey()
 		addr     = crypto.PubkeyToAddress(key.PublicKey)
@@ -742,7 +769,7 @@ func TestLogsSubscription(t *testing.T) {
 
 	// Hack: GenerateChainWithGenesis creates a new db.
 	// Commit the genesis manually and use GenerateChain.
-	_, err := genesis.Commit(db, trie.NewDatabase(db))
+	_, err := genesis.Commit(db, tdb)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -752,7 +779,7 @@ func TestLogsSubscription(t *testing.T) {
 		tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{Nonce: uint64(i), To: &contract, Value: big.NewInt(0), Gas: 46000, GasPrice: b.BaseFee(), Data: common.FromHex(data)}), signer, key)
 		b.AddTx(tx)
 	})
-	bc, err := core.NewBlockChain(db, nil, genesis, nil, ethash.NewFaker(), vm.Config{}, nil, new(uint64))
+	bc, err := core.NewBlockChain(db, genesis, ethash.NewFaker(), nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -771,12 +798,15 @@ func TestLogsSubscription(t *testing.T) {
 	})
 	liveLogs := preceipts[0][0].Logs
 	var (
-		backend, sys = newTestFilterSystem(t, db, Config{})
+		backend, sys = newTestFilterSystem(db, Config{})
-		api          = NewFilterAPI(sys, false)
+		api          = NewFilterAPI(sys)
 		// Transfer(address indexed from, address indexed to, uint256 value);
 		topic = common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")
 	)
 
+	backend.startFilterMaps(0, false, filtermaps.RangeTestParams)
+	defer backend.stopFilterMaps()
+
 	i2h := func(i int) common.Hash { return common.BigToHash(big.NewInt(int64(i))) }
 
 	allLogs := []*types.Log{
@@ -886,9 +916,8 @@ func TestLogsSubscription(t *testing.T) {
 	backend.logsFeed.Send(liveLogs)
 
 	for i := range testCases {
-		err := <-testCases[i].err
+		if err := <-testCases[i].err; err != nil {
-		if err != nil {
+			t.Errorf("test %d failed: %v", i, err)
-			t.Fatalf("test %d failed: %v", i, err)
 		}
 	}
 }
@@ -971,19 +1000,20 @@ func calculateBalance(logs []*types.Log) map[common.Address]uint64 {
 func testLogsSubscriptionReorg(t *testing.T, oldChainMaker, newChainMaker, pendingMaker func(i int, b *core.BlockGen), oldChainLen, newChainLen, forkAt, reorgAt int, expected []*types.Log) {
 	var (
 		db           = rawdb.NewMemoryDatabase()
-		backend, sys = newTestFilterSystem(t, db, Config{})
+		tdb          = triedb.NewDatabase(db, triedb.HashDefaults)
-		api          = NewFilterAPI(sys, false)
+		backend, sys = newTestFilterSystem(db, Config{})
+		api          = NewFilterAPI(sys)
 		genesis      = &core.Genesis{Config: params.TestChainConfig, Alloc: genesisAlloc, GasLimit: 10000000000000}
 	)
 
 	// Hack: GenerateChainWithGenesis creates a new db.
 	// Commit the genesis manually and use GenerateChain.
-	_, err := genesis.Commit(db, trie.NewDatabase(db))
+	_, err := genesis.Commit(db, tdb)
 	if err != nil {
 		t.Fatal(err)
 	}
 	oldChain, _ := core.GenerateChain(genesis.Config, genesis.ToBlock(), ethash.NewFaker(), db, oldChainLen, oldChainMaker)
-	bc, err := core.NewBlockChain(db, nil, genesis, nil, ethash.NewFaker(), vm.Config{}, nil, new(uint64))
+	bc, err := core.NewBlockChain(db, genesis, ethash.NewFaker(), nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -1001,6 +1031,10 @@ func testLogsSubscriptionReorg(t *testing.T, oldChainMaker, newChainMaker, pendi
 	if err != nil {
 		t.Fatal(err)
 	}
+
+	backend.startFilterMaps(0, false, filtermaps.RangeTestParams)
+	defer backend.stopFilterMaps()
+
 	newChain, _ := core.GenerateChain(genesis.Config, oldChain[forkAt], ethash.NewFaker(), db, newChainLen, newChainMaker)
 	logger.Info("oldChain/newChain", "oldChain", fmt.Sprintf("%d..%d", oldChain[0].Number(), oldChain[len(oldChain)-1].Number()), "newChain", fmt.Sprintf("%d..%d", newChain[0].Number(), newChain[len(newChain)-1].Number()))
 

eth/filters/filter_test.go

@@ -456,7 +456,7 @@ func TestRangeLogs(t *testing.T) {
 	)
 
 	expEvent := func(expEvent int, expFirst, expAfterLast uint64) {
-		exp := rangeLogsTestEvent{expEvent, common.NewRange[uint64](expFirst, expAfterLast-expFirst)}
+		exp := rangeLogsTestEvent{expEvent, common.NewRange(expFirst, expAfterLast-expFirst)}
 		event++
 		ev := <-filter.rangeLogsTestHook
 		if ev != exp {