diff --git a/eth/filters/api.go b/eth/filters/api.go index cfd48bc7cb..445fa54650 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -37,17 +37,30 @@ import ( ) var ( - errInvalidTopic = errors.New("invalid topic(s)") - errFilterNotFound = errors.New("filter not found") - errConnectDropped = errors.New("connection dropped") - 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\"") - errClientUnsubscribed = errors.New("client unsubscribed") + errInvalidTopic = errors.New("invalid topic(s)") + errFilterNotFound = errors.New("filter not found") + errConnectDropped = errors.New("connection dropped") + 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\"") + 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) - for { - select { - case logs := <-logsCh: + logs, err := f.Logs(ctx) + if err != nil { + return err + } + // 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. - continue + return ctx.Err() } - case err := <-errChan: - return err + currentBlockLogs = nil + } + 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 { diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index c5f8318826..f635217402 100644 --- a/eth/filters/filter_system.go +++ b/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 diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index a361487aa7..2dab89e02a 100644 --- a/eth/filters/filter_system_test.go +++ b/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{}) - api = NewFilterAPI(sys, false) + backend, sys = newTestFilterSystem(db, Config{}) + 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 != nil { - t.Fatalf("test %d failed: %v", i, err) + if err := <-testCases[i].err; err != nil { + t.Errorf("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{}) - api = NewFilterAPI(sys, false) + tdb = triedb.NewDatabase(db, triedb.HashDefaults) + 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())) diff --git a/eth/filters/filter_test.go b/eth/filters/filter_test.go index e99b2f6caa..b05b0d888c 100644 --- a/eth/filters/filter_test.go +++ b/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 {