From 9afd96afdad01012b260878ddfbcbfc9a49047ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isabel=20Sch=C3=B6ps=20Thiel?= <155141998+IST-Github@users.noreply.github.com> Date: Mon, 1 Jan 2024 00:50:23 +0100 Subject: [PATCH] Delete eth/filters directory --- eth/filters/api.go | 609 ------------------- eth/filters/api_test.go | 185 ------ eth/filters/bench_test.go | 189 ------ eth/filters/filter.go | 422 ------------- eth/filters/filter_system.go | 606 ------------------ eth/filters/filter_system_test.go | 977 ------------------------------ eth/filters/filter_test.go | 389 ------------ 7 files changed, 3377 deletions(-) delete mode 100644 eth/filters/api.go delete mode 100644 eth/filters/api_test.go delete mode 100644 eth/filters/bench_test.go delete mode 100644 eth/filters/filter.go delete mode 100644 eth/filters/filter_system.go delete mode 100644 eth/filters/filter_system_test.go delete mode 100644 eth/filters/filter_test.go diff --git a/eth/filters/api.go b/eth/filters/api.go deleted file mode 100644 index a4eaa9cec8..0000000000 --- a/eth/filters/api.go +++ /dev/null @@ -1,609 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package filters - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "math/big" - "sync" - "time" - - "github.com/ethereum/go-ethereum" - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/common/hexutil" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/internal/ethapi" - "github.com/ethereum/go-ethereum/rpc" -) - -var ( - errInvalidTopic = errors.New("invalid topic(s)") - errFilterNotFound = errors.New("filter not found") - errInvalidBlockRange = errors.New("invalid block range params") - errExceedMaxTopics = errors.New("exceed max topics") -) - -// The maximum number of topic criteria allowed, vm.LOG4 - vm.LOG0 -const maxTopics = 4 - -// filter is a helper struct that holds meta information over the filter type -// and associated subscription in the event system. -type filter struct { - typ Type - deadline *time.Timer // filter is inactive when deadline triggers - hashes []common.Hash - fullTx bool - txs []*types.Transaction - crit FilterCriteria - logs []*types.Log - s *Subscription // associated subscription in event system -} - -// FilterAPI offers support to create and manage filters. This will allow external clients to retrieve various -// information related to the Ethereum protocol such as blocks, transactions and logs. -type FilterAPI struct { - sys *FilterSystem - events *EventSystem - filtersMu sync.Mutex - filters map[rpc.ID]*filter - timeout time.Duration -} - -// NewFilterAPI returns a new FilterAPI instance. -func NewFilterAPI(system *FilterSystem, lightMode bool) *FilterAPI { - api := &FilterAPI{ - sys: system, - events: NewEventSystem(system, lightMode), - filters: make(map[rpc.ID]*filter), - timeout: system.cfg.Timeout, - } - go api.timeoutLoop(system.cfg.Timeout) - - return api -} - -// timeoutLoop runs at the interval set by 'timeout' and deletes filters -// that have not been recently used. It is started when the API is created. -func (api *FilterAPI) timeoutLoop(timeout time.Duration) { - var toUninstall []*Subscription - ticker := time.NewTicker(timeout) - defer ticker.Stop() - for { - <-ticker.C - api.filtersMu.Lock() - for id, f := range api.filters { - select { - case <-f.deadline.C: - toUninstall = append(toUninstall, f.s) - delete(api.filters, id) - default: - continue - } - } - api.filtersMu.Unlock() - - // Unsubscribes are processed outside the lock to avoid the following scenario: - // event loop attempts broadcasting events to still active filters while - // Unsubscribe is waiting for it to process the uninstall request. - for _, s := range toUninstall { - s.Unsubscribe() - } - toUninstall = nil - } -} - -// NewPendingTransactionFilter creates a filter that fetches pending transactions -// as transactions enter the pending state. -// -// It is part of the filter package because this filter can be used through the -// `eth_getFilterChanges` polling method that is also used for log filters. -func (api *FilterAPI) NewPendingTransactionFilter(fullTx *bool) rpc.ID { - var ( - pendingTxs = make(chan []*types.Transaction) - pendingTxSub = api.events.SubscribePendingTxs(pendingTxs) - ) - - api.filtersMu.Lock() - api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, fullTx: fullTx != nil && *fullTx, deadline: time.NewTimer(api.timeout), txs: make([]*types.Transaction, 0), s: pendingTxSub} - api.filtersMu.Unlock() - - go func() { - for { - select { - case pTx := <-pendingTxs: - api.filtersMu.Lock() - if f, found := api.filters[pendingTxSub.ID]; found { - f.txs = append(f.txs, pTx...) - } - api.filtersMu.Unlock() - case <-pendingTxSub.Err(): - api.filtersMu.Lock() - delete(api.filters, pendingTxSub.ID) - api.filtersMu.Unlock() - return - } - } - }() - - return pendingTxSub.ID -} - -// NewPendingTransactions creates a subscription that is triggered each time a -// transaction enters the transaction pool. If fullTx is true the full tx is -// sent to the client, otherwise the hash is sent. -func (api *FilterAPI) NewPendingTransactions(ctx context.Context, fullTx *bool) (*rpc.Subscription, error) { - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() - - go func() { - txs := make(chan []*types.Transaction, 128) - pendingTxSub := api.events.SubscribePendingTxs(txs) - chainConfig := api.sys.backend.ChainConfig() - - for { - select { - case txs := <-txs: - // To keep the original behaviour, send a single tx hash in one notification. - // TODO(rjl493456442) Send a batch of tx hashes in one notification - latest := api.sys.backend.CurrentHeader() - for _, tx := range txs { - if fullTx != nil && *fullTx { - rpcTx := ethapi.NewRPCPendingTransaction(tx, latest, chainConfig) - notifier.Notify(rpcSub.ID, rpcTx) - } else { - notifier.Notify(rpcSub.ID, tx.Hash()) - } - } - case <-rpcSub.Err(): - pendingTxSub.Unsubscribe() - return - case <-notifier.Closed(): - pendingTxSub.Unsubscribe() - return - } - } - }() - - return rpcSub, nil -} - -// NewBlockFilter creates a filter that fetches blocks that are imported into the chain. -// It is part of the filter package since polling goes with eth_getFilterChanges. -func (api *FilterAPI) NewBlockFilter() rpc.ID { - var ( - headers = make(chan *types.Header) - headerSub = api.events.SubscribeNewHeads(headers) - ) - - api.filtersMu.Lock() - api.filters[headerSub.ID] = &filter{typ: BlocksSubscription, deadline: time.NewTimer(api.timeout), hashes: make([]common.Hash, 0), s: headerSub} - api.filtersMu.Unlock() - - go func() { - for { - select { - case h := <-headers: - api.filtersMu.Lock() - if f, found := api.filters[headerSub.ID]; found { - f.hashes = append(f.hashes, h.Hash()) - } - api.filtersMu.Unlock() - case <-headerSub.Err(): - api.filtersMu.Lock() - delete(api.filters, headerSub.ID) - api.filtersMu.Unlock() - return - } - } - }() - - return headerSub.ID -} - -// NewHeads send a notification each time a new (header) block is appended to the chain. -func (api *FilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) { - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - rpcSub := notifier.CreateSubscription() - - go func() { - headers := make(chan *types.Header) - headersSub := api.events.SubscribeNewHeads(headers) - - for { - select { - case h := <-headers: - notifier.Notify(rpcSub.ID, h) - case <-rpcSub.Err(): - headersSub.Unsubscribe() - return - case <-notifier.Closed(): - headersSub.Unsubscribe() - return - } - } - }() - - return rpcSub, nil -} - -// Logs creates a subscription that fires for all new log that match the given filter criteria. -func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) { - notifier, supported := rpc.NotifierFromContext(ctx) - if !supported { - return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported - } - - var ( - rpcSub = notifier.CreateSubscription() - matchedLogs = make(chan []*types.Log) - ) - - logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), matchedLogs) - if err != nil { - return nil, err - } - - go func() { - for { - select { - case logs := <-matchedLogs: - for _, log := range logs { - log := log - notifier.Notify(rpcSub.ID, &log) - } - case <-rpcSub.Err(): // client send an unsubscribe request - logsSub.Unsubscribe() - return - case <-notifier.Closed(): // connection dropped - logsSub.Unsubscribe() - return - } - } - }() - - return rpcSub, nil -} - -// FilterCriteria represents a request to create a new filter. -// Same as ethereum.FilterQuery but with UnmarshalJSON() method. -type FilterCriteria ethereum.FilterQuery - -// NewFilter creates a new filter and returns the filter id. It can be -// used to retrieve logs when the state changes. This method cannot be -// used to fetch logs that are already stored in the state. -// -// Default criteria for the from and to block are "latest". -// Using "latest" as block number will return logs for mined blocks. -// Using "pending" as block number returns logs for not yet mined (pending) blocks. -// In case logs are removed (chain reorg) previously returned logs are returned -// again but with the removed property set to true. -// -// In case "fromBlock" > "toBlock" an error is returned. -func (api *FilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) { - logs := make(chan []*types.Log) - logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), logs) - if err != nil { - return "", err - } - - api.filtersMu.Lock() - api.filters[logsSub.ID] = &filter{typ: LogsSubscription, crit: crit, deadline: time.NewTimer(api.timeout), logs: make([]*types.Log, 0), s: logsSub} - api.filtersMu.Unlock() - - go func() { - for { - select { - case l := <-logs: - api.filtersMu.Lock() - if f, found := api.filters[logsSub.ID]; found { - f.logs = append(f.logs, l...) - } - api.filtersMu.Unlock() - case <-logsSub.Err(): - api.filtersMu.Lock() - delete(api.filters, logsSub.ID) - api.filtersMu.Unlock() - return - } - } - }() - - return logsSub.ID, nil -} - -// GetLogs returns logs matching the given argument that are stored within the state. -func (api *FilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*types.Log, error) { - if len(crit.Topics) > maxTopics { - return nil, errExceedMaxTopics - } - var filter *Filter - if crit.BlockHash != nil { - // Block filter requested, construct a single-shot filter - filter = api.sys.NewBlockFilter(*crit.BlockHash, crit.Addresses, crit.Topics) - } else { - // Convert the RPC block numbers into internal representations - begin := rpc.LatestBlockNumber.Int64() - if crit.FromBlock != nil { - begin = crit.FromBlock.Int64() - } - end := rpc.LatestBlockNumber.Int64() - if crit.ToBlock != nil { - end = crit.ToBlock.Int64() - } - if begin > 0 && end > 0 && begin > end { - return nil, errInvalidBlockRange - } - // Construct the range filter - filter = api.sys.NewRangeFilter(begin, end, crit.Addresses, crit.Topics) - } - // Run the filter and return all the logs - logs, err := filter.Logs(ctx) - if err != nil { - return nil, err - } - return returnLogs(logs), err -} - -// UninstallFilter removes the filter with the given filter id. -func (api *FilterAPI) UninstallFilter(id rpc.ID) bool { - api.filtersMu.Lock() - f, found := api.filters[id] - if found { - delete(api.filters, id) - } - api.filtersMu.Unlock() - if found { - f.s.Unsubscribe() - } - - return found -} - -// GetFilterLogs returns the logs for the filter with the given id. -// If the filter could not be found an empty array of logs is returned. -func (api *FilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*types.Log, error) { - api.filtersMu.Lock() - f, found := api.filters[id] - api.filtersMu.Unlock() - - if !found || f.typ != LogsSubscription { - return nil, errFilterNotFound - } - - var filter *Filter - if f.crit.BlockHash != nil { - // Block filter requested, construct a single-shot filter - filter = api.sys.NewBlockFilter(*f.crit.BlockHash, f.crit.Addresses, f.crit.Topics) - } else { - // Convert the RPC block numbers into internal representations - begin := rpc.LatestBlockNumber.Int64() - if f.crit.FromBlock != nil { - begin = f.crit.FromBlock.Int64() - } - end := rpc.LatestBlockNumber.Int64() - if f.crit.ToBlock != nil { - end = f.crit.ToBlock.Int64() - } - // Construct the range filter - 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) - if err != nil { - return nil, err - } - return returnLogs(logs), nil -} - -// GetFilterChanges returns the logs for the filter with the given id since -// last time it was called. This can be used for polling. -// -// For pending transaction and block filters the result is []common.Hash. -// (pending)Log filters return []Log. -func (api *FilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) { - api.filtersMu.Lock() - defer api.filtersMu.Unlock() - - chainConfig := api.sys.backend.ChainConfig() - latest := api.sys.backend.CurrentHeader() - - if f, found := api.filters[id]; found { - if !f.deadline.Stop() { - // timer expired but filter is not yet removed in timeout loop - // receive timer value and reset timer - <-f.deadline.C - } - f.deadline.Reset(api.timeout) - - switch f.typ { - case BlocksSubscription: - hashes := f.hashes - f.hashes = nil - return returnHashes(hashes), nil - case PendingTransactionsSubscription: - if f.fullTx { - txs := make([]*ethapi.RPCTransaction, 0, len(f.txs)) - for _, tx := range f.txs { - txs = append(txs, ethapi.NewRPCPendingTransaction(tx, latest, chainConfig)) - } - f.txs = nil - return txs, nil - } else { - hashes := make([]common.Hash, 0, len(f.txs)) - for _, tx := range f.txs { - hashes = append(hashes, tx.Hash()) - } - f.txs = nil - return hashes, nil - } - case LogsSubscription, MinedAndPendingLogsSubscription: - logs := f.logs - f.logs = nil - return returnLogs(logs), nil - } - } - - return []interface{}{}, errFilterNotFound -} - -// returnHashes is a helper that will return an empty hash array case the given hash array is nil, -// otherwise the given hashes array is returned. -func returnHashes(hashes []common.Hash) []common.Hash { - if hashes == nil { - return []common.Hash{} - } - return hashes -} - -// returnLogs is a helper that will return an empty log array in case the given logs array is nil, -// otherwise the given logs array is returned. -func returnLogs(logs []*types.Log) []*types.Log { - if logs == nil { - return []*types.Log{} - } - return logs -} - -// UnmarshalJSON sets *args fields with given data. -func (args *FilterCriteria) UnmarshalJSON(data []byte) error { - type input struct { - BlockHash *common.Hash `json:"blockHash"` - FromBlock *rpc.BlockNumber `json:"fromBlock"` - ToBlock *rpc.BlockNumber `json:"toBlock"` - Addresses interface{} `json:"address"` - Topics []interface{} `json:"topics"` - } - - var raw input - if err := json.Unmarshal(data, &raw); err != nil { - return err - } - - if raw.BlockHash != nil { - if raw.FromBlock != nil || raw.ToBlock != nil { - // BlockHash is mutually exclusive with FromBlock/ToBlock criteria - return errors.New("cannot specify both BlockHash and FromBlock/ToBlock, choose one or the other") - } - args.BlockHash = raw.BlockHash - } else { - if raw.FromBlock != nil { - args.FromBlock = big.NewInt(raw.FromBlock.Int64()) - } - - if raw.ToBlock != nil { - args.ToBlock = big.NewInt(raw.ToBlock.Int64()) - } - } - - args.Addresses = []common.Address{} - - if raw.Addresses != nil { - // raw.Address can contain a single address or an array of addresses - switch rawAddr := raw.Addresses.(type) { - case []interface{}: - for i, addr := range rawAddr { - if strAddr, ok := addr.(string); ok { - addr, err := decodeAddress(strAddr) - if err != nil { - return fmt.Errorf("invalid address at index %d: %v", i, err) - } - args.Addresses = append(args.Addresses, addr) - } else { - return fmt.Errorf("non-string address at index %d", i) - } - } - case string: - addr, err := decodeAddress(rawAddr) - if err != nil { - return fmt.Errorf("invalid address: %v", err) - } - args.Addresses = []common.Address{addr} - default: - return errors.New("invalid addresses in query") - } - } - - // topics is an array consisting of strings and/or arrays of strings. - // JSON null values are converted to common.Hash{} and ignored by the filter manager. - if len(raw.Topics) > 0 { - args.Topics = make([][]common.Hash, len(raw.Topics)) - for i, t := range raw.Topics { - switch topic := t.(type) { - case nil: - // ignore topic when matching logs - - case string: - // match specific topic - top, err := decodeTopic(topic) - if err != nil { - return err - } - args.Topics[i] = []common.Hash{top} - - case []interface{}: - // or case e.g. [null, "topic0", "topic1"] - for _, rawTopic := range topic { - if rawTopic == nil { - // null component, match all - args.Topics[i] = nil - break - } - if topic, ok := rawTopic.(string); ok { - parsed, err := decodeTopic(topic) - if err != nil { - return err - } - args.Topics[i] = append(args.Topics[i], parsed) - } else { - return errInvalidTopic - } - } - default: - return errInvalidTopic - } - } - } - - return nil -} - -func decodeAddress(s string) (common.Address, error) { - b, err := hexutil.Decode(s) - if err == nil && len(b) != common.AddressLength { - err = fmt.Errorf("hex has invalid length %d after decoding; expected %d for address", len(b), common.AddressLength) - } - return common.BytesToAddress(b), err -} - -func decodeTopic(s string) (common.Hash, error) { - b, err := hexutil.Decode(s) - if err == nil && len(b) != common.HashLength { - err = fmt.Errorf("hex has invalid length %d after decoding; expected %d for topic", len(b), common.HashLength) - } - return common.BytesToHash(b), err -} diff --git a/eth/filters/api_test.go b/eth/filters/api_test.go deleted file mode 100644 index 822bc826f6..0000000000 --- a/eth/filters/api_test.go +++ /dev/null @@ -1,185 +0,0 @@ -// Copyright 2016 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package filters - -import ( - "encoding/json" - "fmt" - "testing" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/rpc" -) - -func TestUnmarshalJSONNewFilterArgs(t *testing.T) { - var ( - fromBlock rpc.BlockNumber = 0x123435 - toBlock rpc.BlockNumber = 0xabcdef - address0 = common.HexToAddress("70c87d191324e6712a591f304b4eedef6ad9bb9d") - address1 = common.HexToAddress("9b2055d370f73ec7d8a03e965129118dc8f5bf83") - topic0 = common.HexToHash("3ac225168df54212a25c1c01fd35bebfea408fdac2e31ddd6f80a4bbf9a5f1ca") - topic1 = common.HexToHash("9084a792d2f8b16a62b882fd56f7860c07bf5fa91dd8a2ae7e809e5180fef0b3") - topic2 = common.HexToHash("6ccae1c4af4152f460ff510e573399795dfab5dcf1fa60d1f33ac8fdc1e480ce") - ) - - // default values - var test0 FilterCriteria - if err := json.Unmarshal([]byte("{}"), &test0); err != nil { - t.Fatal(err) - } - if test0.FromBlock != nil { - t.Fatalf("expected nil, got %d", test0.FromBlock) - } - if test0.ToBlock != nil { - t.Fatalf("expected nil, got %d", test0.ToBlock) - } - if len(test0.Addresses) != 0 { - t.Fatalf("expected 0 addresses, got %d", len(test0.Addresses)) - } - if len(test0.Topics) != 0 { - t.Fatalf("expected 0 topics, got %d topics", len(test0.Topics)) - } - - // from, to block number - var test1 FilterCriteria - vector := fmt.Sprintf(`{"fromBlock":"%v","toBlock":"%v"}`, fromBlock, toBlock) - if err := json.Unmarshal([]byte(vector), &test1); err != nil { - t.Fatal(err) - } - if test1.FromBlock.Int64() != fromBlock.Int64() { - t.Fatalf("expected FromBlock %d, got %d", fromBlock, test1.FromBlock) - } - if test1.ToBlock.Int64() != toBlock.Int64() { - t.Fatalf("expected ToBlock %d, got %d", toBlock, test1.ToBlock) - } - - // single address - var test2 FilterCriteria - vector = fmt.Sprintf(`{"address": "%s"}`, address0.Hex()) - if err := json.Unmarshal([]byte(vector), &test2); err != nil { - t.Fatal(err) - } - if len(test2.Addresses) != 1 { - t.Fatalf("expected 1 address, got %d address(es)", len(test2.Addresses)) - } - if test2.Addresses[0] != address0 { - t.Fatalf("expected address %x, got %x", address0, test2.Addresses[0]) - } - - // multiple address - var test3 FilterCriteria - vector = fmt.Sprintf(`{"address": ["%s", "%s"]}`, address0.Hex(), address1.Hex()) - if err := json.Unmarshal([]byte(vector), &test3); err != nil { - t.Fatal(err) - } - if len(test3.Addresses) != 2 { - t.Fatalf("expected 2 addresses, got %d address(es)", len(test3.Addresses)) - } - if test3.Addresses[0] != address0 { - t.Fatalf("expected address %x, got %x", address0, test3.Addresses[0]) - } - if test3.Addresses[1] != address1 { - t.Fatalf("expected address %x, got %x", address1, test3.Addresses[1]) - } - - // single topic - var test4 FilterCriteria - vector = fmt.Sprintf(`{"topics": ["%s"]}`, topic0.Hex()) - if err := json.Unmarshal([]byte(vector), &test4); err != nil { - t.Fatal(err) - } - if len(test4.Topics) != 1 { - t.Fatalf("expected 1 topic, got %d", len(test4.Topics)) - } - if len(test4.Topics[0]) != 1 { - t.Fatalf("expected len(topics[0]) to be 1, got %d", len(test4.Topics[0])) - } - if test4.Topics[0][0] != topic0 { - t.Fatalf("got %x, expected %x", test4.Topics[0][0], topic0) - } - - // test multiple "AND" topics - var test5 FilterCriteria - vector = fmt.Sprintf(`{"topics": ["%s", "%s"]}`, topic0.Hex(), topic1.Hex()) - if err := json.Unmarshal([]byte(vector), &test5); err != nil { - t.Fatal(err) - } - if len(test5.Topics) != 2 { - t.Fatalf("expected 2 topics, got %d", len(test5.Topics)) - } - if len(test5.Topics[0]) != 1 { - t.Fatalf("expected 1 topic, got %d", len(test5.Topics[0])) - } - if test5.Topics[0][0] != topic0 { - t.Fatalf("got %x, expected %x", test5.Topics[0][0], topic0) - } - if len(test5.Topics[1]) != 1 { - t.Fatalf("expected 1 topic, got %d", len(test5.Topics[1])) - } - if test5.Topics[1][0] != topic1 { - t.Fatalf("got %x, expected %x", test5.Topics[1][0], topic1) - } - - // test optional topic - var test6 FilterCriteria - vector = fmt.Sprintf(`{"topics": ["%s", null, "%s"]}`, topic0.Hex(), topic2.Hex()) - if err := json.Unmarshal([]byte(vector), &test6); err != nil { - t.Fatal(err) - } - if len(test6.Topics) != 3 { - t.Fatalf("expected 3 topics, got %d", len(test6.Topics)) - } - if len(test6.Topics[0]) != 1 { - t.Fatalf("expected 1 topic, got %d", len(test6.Topics[0])) - } - if test6.Topics[0][0] != topic0 { - t.Fatalf("got %x, expected %x", test6.Topics[0][0], topic0) - } - if len(test6.Topics[1]) != 0 { - t.Fatalf("expected 0 topic, got %d", len(test6.Topics[1])) - } - if len(test6.Topics[2]) != 1 { - t.Fatalf("expected 1 topic, got %d", len(test6.Topics[2])) - } - if test6.Topics[2][0] != topic2 { - t.Fatalf("got %x, expected %x", test6.Topics[2][0], topic2) - } - - // test OR topics - var test7 FilterCriteria - vector = fmt.Sprintf(`{"topics": [["%s", "%s"], null, ["%s", null]]}`, topic0.Hex(), topic1.Hex(), topic2.Hex()) - if err := json.Unmarshal([]byte(vector), &test7); err != nil { - t.Fatal(err) - } - if len(test7.Topics) != 3 { - t.Fatalf("expected 3 topics, got %d topics", len(test7.Topics)) - } - if len(test7.Topics[0]) != 2 { - t.Fatalf("expected 2 topics, got %d topics", len(test7.Topics[0])) - } - if test7.Topics[0][0] != topic0 || test7.Topics[0][1] != topic1 { - t.Fatalf("invalid topics expected [%x,%x], got [%x,%x]", - topic0, topic1, test7.Topics[0][0], test7.Topics[0][1], - ) - } - if len(test7.Topics[1]) != 0 { - t.Fatalf("expected 0 topic, got %d topics", len(test7.Topics[1])) - } - if len(test7.Topics[2]) != 0 { - t.Fatalf("expected 0 topics, got %d topics", len(test7.Topics[2])) - } -} diff --git a/eth/filters/bench_test.go b/eth/filters/bench_test.go deleted file mode 100644 index 73b96b77af..0000000000 --- a/eth/filters/bench_test.go +++ /dev/null @@ -1,189 +0,0 @@ -// Copyright 2017 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package filters - -import ( - "context" - "fmt" - "testing" - "time" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/common/bitutil" - "github.com/ethereum/go-ethereum/core/bloombits" - "github.com/ethereum/go-ethereum/core/rawdb" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/node" -) - -func BenchmarkBloomBits512(b *testing.B) { - benchmarkBloomBits(b, 512) -} - -func BenchmarkBloomBits1k(b *testing.B) { - benchmarkBloomBits(b, 1024) -} - -func BenchmarkBloomBits2k(b *testing.B) { - benchmarkBloomBits(b, 2048) -} - -func BenchmarkBloomBits4k(b *testing.B) { - benchmarkBloomBits(b, 4096) -} - -func BenchmarkBloomBits8k(b *testing.B) { - benchmarkBloomBits(b, 8192) -} - -func BenchmarkBloomBits16k(b *testing.B) { - benchmarkBloomBits(b, 16384) -} - -func BenchmarkBloomBits32k(b *testing.B) { - benchmarkBloomBits(b, 32768) -} - -const benchFilterCnt = 2000 - -func benchmarkBloomBits(b *testing.B, sectionSize uint64) { - b.Skip("test disabled: this tests presume (and modify) an existing datadir.") - benchDataDir := node.DefaultDataDir() + "/geth/chaindata" - b.Log("Running bloombits benchmark section size:", sectionSize) - - db, err := rawdb.NewLevelDBDatabase(benchDataDir, 128, 1024, "", false) - if err != nil { - b.Fatalf("error opening database at %v: %v", benchDataDir, err) - } - head := rawdb.ReadHeadBlockHash(db) - if head == (common.Hash{}) { - b.Fatalf("chain data not found at %v", benchDataDir) - } - - clearBloomBits(db) - b.Log("Generating bloombits data...") - headNum := rawdb.ReadHeaderNumber(db, head) - if headNum == nil || *headNum < sectionSize+512 { - b.Fatalf("not enough blocks for running a benchmark") - } - - start := time.Now() - cnt := (*headNum - 512) / sectionSize - var dataSize, compSize uint64 - for sectionIdx := uint64(0); sectionIdx < cnt; sectionIdx++ { - bc, err := bloombits.NewGenerator(uint(sectionSize)) - if err != nil { - b.Fatalf("failed to create generator: %v", err) - } - var header *types.Header - for i := sectionIdx * sectionSize; i < (sectionIdx+1)*sectionSize; i++ { - hash := rawdb.ReadCanonicalHash(db, i) - if header = rawdb.ReadHeader(db, hash, i); header == nil { - b.Fatalf("Error creating bloomBits data") - return - } - bc.AddBloom(uint(i-sectionIdx*sectionSize), header.Bloom) - } - sectionHead := rawdb.ReadCanonicalHash(db, (sectionIdx+1)*sectionSize-1) - for i := 0; i < types.BloomBitLength; i++ { - data, err := bc.Bitset(uint(i)) - if err != nil { - b.Fatalf("failed to retrieve bitset: %v", err) - } - comp := bitutil.CompressBytes(data) - dataSize += uint64(len(data)) - compSize += uint64(len(comp)) - rawdb.WriteBloomBits(db, uint(i), sectionIdx, sectionHead, comp) - } - //if sectionIdx%50 == 0 { - // b.Log(" section", sectionIdx, "/", cnt) - //} - } - - d := time.Since(start) - b.Log("Finished generating bloombits data") - b.Log(" ", d, "total ", d/time.Duration(cnt*sectionSize), "per block") - b.Log(" data size:", dataSize, " compressed size:", compSize, " compression ratio:", float64(compSize)/float64(dataSize)) - - b.Log("Running filter benchmarks...") - start = time.Now() - - var ( - backend *testBackend - sys *FilterSystem - ) - for i := 0; i < benchFilterCnt; i++ { - if i%20 == 0 { - db.Close() - db, _ = rawdb.NewLevelDBDatabase(benchDataDir, 128, 1024, "", false) - backend = &testBackend{db: db, sections: cnt} - sys = NewFilterSystem(backend, Config{}) - } - var addr common.Address - addr[0] = byte(i) - addr[1] = byte(i / 256) - 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) - } - } - - d = time.Since(start) - b.Log("Finished running filter benchmarks") - b.Log(" ", d, "total ", d/time.Duration(benchFilterCnt), "per address", d*time.Duration(1000000)/time.Duration(benchFilterCnt*cnt*sectionSize), "per million blocks") - db.Close() -} - -//nolint:unused -func clearBloomBits(db ethdb.Database) { - var bloomBitsPrefix = []byte("bloomBits-") - fmt.Println("Clearing bloombits data...") - it := db.NewIterator(bloomBitsPrefix, nil) - for it.Next() { - db.Delete(it.Key()) - } - it.Release() -} - -func BenchmarkNoBloomBits(b *testing.B) { - b.Skip("test disabled: this tests presume (and modify) an existing datadir.") - benchDataDir := node.DefaultDataDir() + "/geth/chaindata" - b.Log("Running benchmark without bloombits") - db, err := rawdb.NewLevelDBDatabase(benchDataDir, 128, 1024, "", false) - if err != nil { - b.Fatalf("error opening database at %v: %v", benchDataDir, err) - } - head := rawdb.ReadHeadBlockHash(db) - if head == (common.Hash{}) { - b.Fatalf("chain data not found at %v", benchDataDir) - } - headNum := rawdb.ReadHeaderNumber(db, head) - - clearBloomBits(db) - - _, sys := newTestFilterSystem(b, db, Config{}) - - b.Log("Running filter benchmarks...") - start := time.Now() - filter := sys.NewRangeFilter(0, int64(*headNum), []common.Address{{}}, nil) - filter.Logs(context.Background()) - d := time.Since(start) - b.Log("Finished running filter benchmarks") - b.Log(" ", d, "total ", d*time.Duration(1000000)/time.Duration(*headNum+1), "per million blocks") - db.Close() -} diff --git a/eth/filters/filter.go b/eth/filters/filter.go deleted file mode 100644 index 83e3284a2b..0000000000 --- a/eth/filters/filter.go +++ /dev/null @@ -1,422 +0,0 @@ -// Copyright 2014 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package filters - -import ( - "context" - "errors" - "math/big" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/core/bloombits" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/rpc" -) - -// Filter can be used to retrieve and filter logs. -type Filter struct { - sys *FilterSystem - - addresses []common.Address - topics [][]common.Hash - - block *common.Hash // Block hash if filtering a single block - begin, end int64 // Range interval if filtering multiple blocks - - matcher *bloombits.Matcher -} - -// 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 { - // 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. - var filters [][][]byte - if len(addresses) > 0 { - filter := make([][]byte, len(addresses)) - for i, address := range addresses { - filter[i] = address.Bytes() - } - filters = append(filters, filter) - } - for _, topicList := range topics { - filter := make([][]byte, len(topicList)) - for i, topic := range topicList { - filter[i] = topic.Bytes() - } - filters = append(filters, filter) - } - size, _ := sys.backend.BloomStatus() - - // 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 -} - -// NewBlockFilter creates a new filter which directly inspects the contents of -// a block to figure out whether it is interesting or not. -func (sys *FilterSystem) NewBlockFilter(block common.Hash, addresses []common.Address, topics [][]common.Hash) *Filter { - // Create a generic filter and convert it into a block filter - filter := newFilter(sys, addresses, topics) - filter.block = &block - return filter -} - -// newFilter creates a generic filter that can either filter based on a block hash, -// or based on range queries. The search criteria needs to be explicitly set. -func newFilter(sys *FilterSystem, addresses []common.Address, topics [][]common.Hash) *Filter { - return &Filter{ - sys: sys, - addresses: addresses, - topics: topics, - } -} - -// Logs searches the blockchain for matching log entries, returning all from the -// first block that contains matches, updating the start of the filter accordingly. -func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) { - // If we're doing singleton block filtering, execute and return - if f.block != nil { - header, err := f.sys.backend.HeaderByHash(ctx, *f.block) - if err != nil { - return nil, err - } - if header == nil { - return nil, errors.New("unknown block") - } - 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, errInvalidBlockRange - } - - // 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 - 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) - if hdr == nil { - return 0, errors.New("latest header not found") - } - 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 - } - - var err error - // range query need to resolve the special begin/end block number - if f.begin, err = resolveSpecial(f.begin); err != nil { - return nil, err - } - if f.end, err = resolveSpecial(f.end); err != nil { - return nil, err - } - - 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 - } - } -} - -// rangeLogsAsync retrieves block-range logs that match the filter criteria asynchronously, -// it creates and returns two channels: one for delivering log data, and one for reporting errors. -func (f *Filter) rangeLogsAsync(ctx context.Context) (chan *types.Log, chan error) { - var ( - logChan = make(chan *types.Log) - errChan = make(chan error) - ) - - go func() { - defer func() { - close(errChan) - close(logChan) - }() - - // Gather all indexed logs, and finish with non indexed ones - var ( - end = uint64(f.end) - size, sections = f.sys.backend.BloomStatus() - err error - ) - if indexed := sections * size; indexed > uint64(f.begin) { - if indexed > end { - indexed = end + 1 - } - if err = f.indexedLogs(ctx, indexed-1, logChan); err != nil { - errChan <- err - return - } - } - - if err := f.unindexedLogs(ctx, end, logChan); err != nil { - errChan <- err - return - } - - errChan <- nil - }() - - return logChan, errChan -} - -// indexedLogs returns the logs matching the filter criteria based on the bloom -// bits indexed available locally or via the network. -func (f *Filter) indexedLogs(ctx context.Context, end uint64, logChan chan *types.Log) error { - // Create a matcher session and request servicing from the backend - matches := make(chan uint64, 64) - - session, err := f.matcher.Start(ctx, uint64(f.begin), end, matches) - if err != nil { - return err - } - defer session.Close() - - f.sys.backend.ServiceFilter(ctx, session) - - for { - select { - case number, ok := <-matches: - // Abort if all matches have been fulfilled - if !ok { - err := session.Error() - if err == nil { - f.begin = int64(end) + 1 - } - return err - } - f.begin = int64(number) + 1 - - // Retrieve the suggested block and pull any truly matching logs - header, err := f.sys.backend.HeaderByNumber(ctx, rpc.BlockNumber(number)) - if header == nil || err != nil { - return err - } - found, err := f.checkMatches(ctx, header) - if err != nil { - return err - } - for _, log := range found { - logChan <- log - } - - case <-ctx.Done(): - return ctx.Err() - } - } -} - -// unindexedLogs returns the logs matching the filter criteria based on raw block -// iteration and bloom matching. -func (f *Filter) unindexedLogs(ctx context.Context, end uint64, logChan chan *types.Log) error { - for ; f.begin <= int64(end); f.begin++ { - header, err := f.sys.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin)) - if header == nil || err != nil { - return err - } - found, err := f.blockLogs(ctx, header) - if err != nil { - return err - } - for _, log := range found { - select { - case logChan <- log: - case <-ctx.Done(): - return ctx.Err() - } - } - } - return nil -} - -// blockLogs returns the logs matching the filter criteria within a single block. -func (f *Filter) blockLogs(ctx context.Context, header *types.Header) ([]*types.Log, error) { - if bloomFilter(header.Bloom, f.addresses, f.topics) { - return f.checkMatches(ctx, header) - } - return nil, nil -} - -// checkMatches checks if the receipts belonging to the given header contain any log events that -// match the filter criteria. This function is called when the bloom filter signals a potential match. -// skipFilter signals all logs of the given block are requested. -func (f *Filter) checkMatches(ctx context.Context, header *types.Header) ([]*types.Log, error) { - hash := header.Hash() - // Logs in cache are partially filled with context data - // such as tx index, block hash, etc. - // Notably tx hash is NOT filled in because it needs - // access to block body data. - cached, err := f.sys.cachedLogElem(ctx, hash, header.Number.Uint64()) - if err != nil { - return nil, err - } - logs := filterLogs(cached.logs, nil, nil, f.addresses, f.topics) - if len(logs) == 0 { - return nil, nil - } - // Most backends will deliver un-derived logs, but check nevertheless. - if len(logs) > 0 && logs[0].TxHash != (common.Hash{}) { - return logs, nil - } - - body, err := f.sys.cachedGetBody(ctx, cached, hash, header.Number.Uint64()) - if err != nil { - return nil, err - } - for i, log := range logs { - // Copy log not to modify cache elements - logcopy := *log - logcopy.TxHash = body.Transactions[logcopy.TxIndex].Hash() - logs[i] = &logcopy - } - 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 || receipts == 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 -} - -// includes returns true if the element is present in the list. -func includes[T comparable](things []T, element T) bool { - for _, thing := range things { - if thing == element { - return true - } - } - return false -} - -// filterLogs creates a slice of logs matching the given criteria. -func filterLogs(logs []*types.Log, fromBlock, toBlock *big.Int, addresses []common.Address, topics [][]common.Hash) []*types.Log { - var check = func(log *types.Log) bool { - if fromBlock != nil && fromBlock.Int64() >= 0 && fromBlock.Uint64() > log.BlockNumber { - return false - } - if toBlock != nil && toBlock.Int64() >= 0 && toBlock.Uint64() < log.BlockNumber { - return false - } - if len(addresses) > 0 && !includes(addresses, log.Address) { - return false - } - // If the to filtered topics is greater than the amount of topics in logs, skip. - if len(topics) > len(log.Topics) { - return false - } - for i, sub := range topics { - if len(sub) == 0 { - continue // empty rule set == wildcard - } - if !includes(sub, log.Topics[i]) { - return false - } - } - return true - } - var ret []*types.Log - for _, log := range logs { - if check(log) { - ret = append(ret, log) - } - } - return ret -} - -func bloomFilter(bloom types.Bloom, addresses []common.Address, topics [][]common.Hash) bool { - if len(addresses) > 0 { - var included bool - for _, addr := range addresses { - if types.BloomLookup(bloom, addr) { - included = true - break - } - } - if !included { - return false - } - } - - for _, sub := range topics { - included := len(sub) == 0 // empty rule set == wildcard - for _, topic := range sub { - if types.BloomLookup(bloom, topic) { - included = true - break - } - } - if !included { - return false - } - } - return true -} diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go deleted file mode 100644 index f98a1f84ce..0000000000 --- a/eth/filters/filter_system.go +++ /dev/null @@ -1,606 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -// Package filters implements an ethereum filtering system for block, -// transactions and log events. -package filters - -import ( - "context" - "fmt" - "sync" - "sync/atomic" - "time" - - "github.com/ethereum/go-ethereum" - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/common/lru" - "github.com/ethereum/go-ethereum/core" - "github.com/ethereum/go-ethereum/core/bloombits" - "github.com/ethereum/go-ethereum/core/rawdb" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/event" - "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/params" - "github.com/ethereum/go-ethereum/rpc" -) - -// Config represents the configuration of the filter system. -type Config struct { - LogCacheSize int // maximum number of cached blocks (default: 32) - Timeout time.Duration // how long filters stay active (default: 5min) -} - -func (cfg Config) withDefaults() Config { - if cfg.Timeout == 0 { - cfg.Timeout = 5 * time.Minute - } - if cfg.LogCacheSize == 0 { - cfg.LogCacheSize = 32 - } - return cfg -} - -type Backend interface { - ChainDb() ethdb.Database - HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) - HeaderByHash(ctx context.Context, blockHash common.Hash) (*types.Header, error) - GetBody(ctx context.Context, hash common.Hash, number rpc.BlockNumber) (*types.Body, error) - GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) - GetLogs(ctx context.Context, blockHash common.Hash, number uint64) ([][]*types.Log, error) - PendingBlockAndReceipts() (*types.Block, types.Receipts) - - CurrentHeader() *types.Header - ChainConfig() *params.ChainConfig - SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription - SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription - SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription - SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription - SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription - - BloomStatus() (uint64, uint64) - ServiceFilter(ctx context.Context, session *bloombits.MatcherSession) -} - -// FilterSystem holds resources shared by all filters. -type FilterSystem struct { - backend Backend - logsCache *lru.Cache[common.Hash, *logCacheElem] - cfg *Config -} - -// NewFilterSystem creates a filter system. -func NewFilterSystem(backend Backend, config Config) *FilterSystem { - config = config.withDefaults() - return &FilterSystem{ - backend: backend, - logsCache: lru.NewCache[common.Hash, *logCacheElem](config.LogCacheSize), - cfg: &config, - } -} - -type logCacheElem struct { - logs []*types.Log - body atomic.Value -} - -// cachedLogElem loads block logs from the backend and caches the result. -func (sys *FilterSystem) cachedLogElem(ctx context.Context, blockHash common.Hash, number uint64) (*logCacheElem, error) { - cached, ok := sys.logsCache.Get(blockHash) - if ok { - return cached, nil - } - - logs, err := sys.backend.GetLogs(ctx, blockHash, number) - if err != nil { - return nil, err - } - if logs == nil { - return nil, fmt.Errorf("failed to get logs for block #%d (0x%s)", number, blockHash.TerminalString()) - } - // Database logs are un-derived. - // Fill in whatever we can (txHash is inaccessible at this point). - flattened := make([]*types.Log, 0) - var logIdx uint - for i, txLogs := range logs { - for _, log := range txLogs { - log.BlockHash = blockHash - log.BlockNumber = number - log.TxIndex = uint(i) - log.Index = logIdx - logIdx++ - flattened = append(flattened, log) - } - } - elem := &logCacheElem{logs: flattened} - sys.logsCache.Add(blockHash, elem) - return elem, nil -} - -func (sys *FilterSystem) cachedGetBody(ctx context.Context, elem *logCacheElem, hash common.Hash, number uint64) (*types.Body, error) { - if body := elem.body.Load(); body != nil { - return body.(*types.Body), nil - } - body, err := sys.backend.GetBody(ctx, hash, rpc.BlockNumber(number)) - if err != nil { - return nil, err - } - elem.body.Store(body) - return body, nil -} - -// Type determines the kind of filter and is used to put the filter in to -// the correct bucket when added. -type Type byte - -const ( - // UnknownSubscription indicates an unknown subscription type - UnknownSubscription Type = iota - // LogsSubscription queries for new or removed (chain reorg) logs - LogsSubscription - // PendingLogsSubscription queries for logs in pending blocks - PendingLogsSubscription - // MinedAndPendingLogsSubscription queries for logs in mined and pending blocks. - MinedAndPendingLogsSubscription - // PendingTransactionsSubscription queries for pending transactions entering - // the pending state - PendingTransactionsSubscription - // BlocksSubscription queries hashes for blocks that are imported - BlocksSubscription - // LastIndexSubscription keeps track of the last index - LastIndexSubscription -) - -const ( - // txChanSize is the size of channel listening to NewTxsEvent. - // The number is referenced from the size of tx pool. - txChanSize = 4096 - // rmLogsChanSize is the size of channel listening to RemovedLogsEvent. - rmLogsChanSize = 10 - // logsChanSize is the size of channel listening to LogsEvent. - logsChanSize = 10 - // chainEvChanSize is the size of channel listening to ChainEvent. - chainEvChanSize = 10 -) - -type subscription struct { - id rpc.ID - typ Type - created time.Time - logsCrit ethereum.FilterQuery - logs chan []*types.Log - txs chan []*types.Transaction - headers chan *types.Header - installed chan struct{} // closed when the filter is installed - err chan error // closed when the filter is uninstalled -} - -// EventSystem creates subscriptions, processes events and broadcasts them to the -// subscription which match the subscription criteria. -type EventSystem struct { - backend Backend - sys *FilterSystem - lightMode bool - lastHead *types.Header - - // Subscriptions - txsSub event.Subscription // Subscription for new transaction event - logsSub event.Subscription // Subscription for new log event - rmLogsSub event.Subscription // Subscription for removed log event - pendingLogsSub event.Subscription // Subscription for pending log event - chainSub event.Subscription // Subscription for new chain event - - // Channels - install chan *subscription // install filter for event notification - uninstall chan *subscription // remove filter for event notification - txsCh chan core.NewTxsEvent // Channel to receive new transactions event - logsCh chan []*types.Log // Channel to receive new log event - pendingLogsCh chan []*types.Log // Channel to receive new log event - rmLogsCh chan core.RemovedLogsEvent // Channel to receive removed log event - chainCh chan core.ChainEvent // Channel to receive new chain event -} - -// NewEventSystem creates a new manager that listens for event on the given mux, -// parses and filters them. It uses the all map to retrieve filter changes. The -// work loop holds its own index that is used to forward events to filters. -// -// The returned manager has a loop that needs to be stopped with the Stop function -// or by stopping the given mux. -func NewEventSystem(sys *FilterSystem, lightMode bool) *EventSystem { - m := &EventSystem{ - sys: sys, - backend: sys.backend, - lightMode: lightMode, - install: make(chan *subscription), - uninstall: make(chan *subscription), - txsCh: make(chan core.NewTxsEvent, txChanSize), - logsCh: make(chan []*types.Log, logsChanSize), - rmLogsCh: make(chan core.RemovedLogsEvent, rmLogsChanSize), - pendingLogsCh: make(chan []*types.Log, logsChanSize), - chainCh: make(chan core.ChainEvent, chainEvChanSize), - } - - // Subscribe events - m.txsSub = m.backend.SubscribeNewTxsEvent(m.txsCh) - m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh) - m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh) - m.chainSub = m.backend.SubscribeChainEvent(m.chainCh) - m.pendingLogsSub = m.backend.SubscribePendingLogsEvent(m.pendingLogsCh) - - // Make sure none of the subscriptions are empty - if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil || m.pendingLogsSub == nil { - log.Crit("Subscribe for event system failed") - } - - go m.eventLoop() - return m -} - -// Subscription is created when the client registers itself for a particular event. -type Subscription struct { - ID rpc.ID - f *subscription - es *EventSystem - unsubOnce sync.Once -} - -// Err returns a channel that is closed when unsubscribed. -func (sub *Subscription) Err() <-chan error { - return sub.f.err -} - -// Unsubscribe uninstalls the subscription from the event broadcast loop. -func (sub *Subscription) Unsubscribe() { - sub.unsubOnce.Do(func() { - uninstallLoop: - for { - // write uninstall request and consume logs/hashes. This prevents - // the eventLoop broadcast method to deadlock when writing to the - // filter event channel while the subscription loop is waiting for - // this method to return (and thus not reading these events). - select { - case sub.es.uninstall <- sub.f: - break uninstallLoop - case <-sub.f.logs: - case <-sub.f.txs: - case <-sub.f.headers: - } - } - - // wait for filter to be uninstalled in work loop before returning - // this ensures that the manager won't use the event channel which - // will probably be closed by the client asap after this method returns. - <-sub.Err() - }) -} - -// subscribe installs the subscription in the event broadcast loop. -func (es *EventSystem) subscribe(sub *subscription) *Subscription { - es.install <- sub - <-sub.installed - return &Subscription{ID: sub.id, f: sub, es: es} -} - -// SubscribeLogs creates a subscription that will write all logs matching the -// given criteria to the given logs channel. Default value for the from and to -// block is "latest". If the fromBlock > toBlock an error is returned. -func (es *EventSystem) SubscribeLogs(crit ethereum.FilterQuery, logs chan []*types.Log) (*Subscription, error) { - if len(crit.Topics) > maxTopics { - return nil, errExceedMaxTopics - } - var from, to rpc.BlockNumber - if crit.FromBlock == nil { - from = rpc.LatestBlockNumber - } else { - from = rpc.BlockNumber(crit.FromBlock.Int64()) - } - if crit.ToBlock == nil { - to = rpc.LatestBlockNumber - } else { - to = rpc.BlockNumber(crit.ToBlock.Int64()) - } - - // only interested in pending logs - if from == rpc.PendingBlockNumber && to == rpc.PendingBlockNumber { - return es.subscribePendingLogs(crit, logs), nil - } - // only interested in new mined logs - if from == rpc.LatestBlockNumber && to == rpc.LatestBlockNumber { - return es.subscribeLogs(crit, logs), nil - } - // only interested in mined logs within a specific block range - if from >= 0 && to >= 0 && to >= from { - return es.subscribeLogs(crit, logs), nil - } - // interested in mined logs from a specific block number, new logs and pending logs - if from >= rpc.LatestBlockNumber && to == rpc.PendingBlockNumber { - return es.subscribeMinedPendingLogs(crit, logs), nil - } - // interested in logs from a specific block number to new mined blocks - if from >= 0 && to == rpc.LatestBlockNumber { - return es.subscribeLogs(crit, logs), nil - } - return nil, errInvalidBlockRange -} - -// subscribeMinedPendingLogs creates a subscription that returned mined and -// pending logs that match the given criteria. -func (es *EventSystem) subscribeMinedPendingLogs(crit ethereum.FilterQuery, logs chan []*types.Log) *Subscription { - sub := &subscription{ - id: rpc.NewID(), - typ: MinedAndPendingLogsSubscription, - logsCrit: crit, - created: time.Now(), - logs: logs, - txs: make(chan []*types.Transaction), - headers: make(chan *types.Header), - installed: make(chan struct{}), - err: make(chan error), - } - return es.subscribe(sub) -} - -// subscribeLogs creates a subscription that will write all logs matching the -// given criteria to the given logs channel. -func (es *EventSystem) subscribeLogs(crit ethereum.FilterQuery, logs chan []*types.Log) *Subscription { - sub := &subscription{ - id: rpc.NewID(), - typ: LogsSubscription, - logsCrit: crit, - created: time.Now(), - logs: logs, - txs: make(chan []*types.Transaction), - headers: make(chan *types.Header), - installed: make(chan struct{}), - err: make(chan error), - } - return es.subscribe(sub) -} - -// subscribePendingLogs creates a subscription that writes contract event logs for -// transactions that enter the transaction pool. -func (es *EventSystem) subscribePendingLogs(crit ethereum.FilterQuery, logs chan []*types.Log) *Subscription { - sub := &subscription{ - id: rpc.NewID(), - typ: PendingLogsSubscription, - logsCrit: crit, - created: time.Now(), - logs: logs, - txs: make(chan []*types.Transaction), - headers: make(chan *types.Header), - installed: make(chan struct{}), - err: make(chan error), - } - return es.subscribe(sub) -} - -// SubscribeNewHeads creates a subscription that writes the header of a block that is -// imported in the chain. -func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscription { - sub := &subscription{ - id: rpc.NewID(), - typ: BlocksSubscription, - created: time.Now(), - logs: make(chan []*types.Log), - txs: make(chan []*types.Transaction), - headers: headers, - installed: make(chan struct{}), - err: make(chan error), - } - return es.subscribe(sub) -} - -// SubscribePendingTxs creates a subscription that writes transactions for -// transactions that enter the transaction pool. -func (es *EventSystem) SubscribePendingTxs(txs chan []*types.Transaction) *Subscription { - sub := &subscription{ - id: rpc.NewID(), - typ: PendingTransactionsSubscription, - created: time.Now(), - logs: make(chan []*types.Log), - txs: txs, - headers: make(chan *types.Header), - installed: make(chan struct{}), - err: make(chan error), - } - return es.subscribe(sub) -} - -type filterIndex map[Type]map[rpc.ID]*subscription - -func (es *EventSystem) handleLogs(filters filterIndex, ev []*types.Log) { - if len(ev) == 0 { - return - } - for _, f := range filters[LogsSubscription] { - matchedLogs := filterLogs(ev, f.logsCrit.FromBlock, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics) - if len(matchedLogs) > 0 { - f.logs <- matchedLogs - } - } -} - -func (es *EventSystem) handlePendingLogs(filters filterIndex, ev []*types.Log) { - if len(ev) == 0 { - return - } - for _, f := range filters[PendingLogsSubscription] { - matchedLogs := filterLogs(ev, nil, f.logsCrit.ToBlock, f.logsCrit.Addresses, f.logsCrit.Topics) - if len(matchedLogs) > 0 { - f.logs <- matchedLogs - } - } -} - -func (es *EventSystem) handleTxsEvent(filters filterIndex, ev core.NewTxsEvent) { - for _, f := range filters[PendingTransactionsSubscription] { - f.txs <- ev.Txs - } -} - -func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent) { - for _, f := range filters[BlocksSubscription] { - f.headers <- ev.Block.Header() - } - if es.lightMode && len(filters[LogsSubscription]) > 0 { - es.lightFilterNewHead(ev.Block.Header(), func(header *types.Header, remove bool) { - for _, f := range filters[LogsSubscription] { - if f.logsCrit.FromBlock != nil && header.Number.Cmp(f.logsCrit.FromBlock) < 0 { - continue - } - if f.logsCrit.ToBlock != nil && header.Number.Cmp(f.logsCrit.ToBlock) > 0 { - continue - } - if matchedLogs := es.lightFilterLogs(header, f.logsCrit.Addresses, f.logsCrit.Topics, remove); len(matchedLogs) > 0 { - f.logs <- matchedLogs - } - } - }) - } -} - -func (es *EventSystem) lightFilterNewHead(newHeader *types.Header, callBack func(*types.Header, bool)) { - oldh := es.lastHead - es.lastHead = newHeader - if oldh == nil { - return - } - newh := newHeader - // find common ancestor, create list of rolled back and new block hashes - var oldHeaders, newHeaders []*types.Header - for oldh.Hash() != newh.Hash() { - if oldh.Number.Uint64() >= newh.Number.Uint64() { - oldHeaders = append(oldHeaders, oldh) - oldh = rawdb.ReadHeader(es.backend.ChainDb(), oldh.ParentHash, oldh.Number.Uint64()-1) - } - if oldh.Number.Uint64() < newh.Number.Uint64() { - newHeaders = append(newHeaders, newh) - newh = rawdb.ReadHeader(es.backend.ChainDb(), newh.ParentHash, newh.Number.Uint64()-1) - if newh == nil { - // happens when CHT syncing, nothing to do - newh = oldh - } - } - } - // roll back old blocks - for _, h := range oldHeaders { - callBack(h, true) - } - // check new blocks (array is in reverse order) - for i := len(newHeaders) - 1; i >= 0; i-- { - callBack(newHeaders[i], false) - } -} - -// filter logs of a single header in light client mode -func (es *EventSystem) lightFilterLogs(header *types.Header, addresses []common.Address, topics [][]common.Hash, remove bool) []*types.Log { - if !bloomFilter(header.Bloom, addresses, topics) { - return nil - } - // Get the logs of the block - ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) - defer cancel() - cached, err := es.sys.cachedLogElem(ctx, header.Hash(), header.Number.Uint64()) - if err != nil { - return nil - } - unfiltered := append([]*types.Log{}, cached.logs...) - for i, log := range unfiltered { - // Don't modify in-cache elements - logcopy := *log - logcopy.Removed = remove - // Swap copy in-place - unfiltered[i] = &logcopy - } - logs := filterLogs(unfiltered, nil, nil, addresses, topics) - // Txhash is already resolved - if len(logs) > 0 && logs[0].TxHash != (common.Hash{}) { - return logs - } - // Resolve txhash - body, err := es.sys.cachedGetBody(ctx, cached, header.Hash(), header.Number.Uint64()) - if err != nil { - return nil - } - for _, log := range logs { - // logs are already copied, safe to modify - log.TxHash = body.Transactions[log.TxIndex].Hash() - } - return logs -} - -// eventLoop (un)installs filters and processes mux events. -func (es *EventSystem) eventLoop() { - // Ensure all subscriptions get cleaned up - defer func() { - es.txsSub.Unsubscribe() - es.logsSub.Unsubscribe() - es.rmLogsSub.Unsubscribe() - es.pendingLogsSub.Unsubscribe() - es.chainSub.Unsubscribe() - }() - - index := make(filterIndex) - for i := UnknownSubscription; i < LastIndexSubscription; i++ { - index[i] = make(map[rpc.ID]*subscription) - } - - for { - select { - case ev := <-es.txsCh: - es.handleTxsEvent(index, ev) - case ev := <-es.logsCh: - es.handleLogs(index, ev) - case ev := <-es.rmLogsCh: - es.handleLogs(index, ev.Logs) - case ev := <-es.pendingLogsCh: - es.handlePendingLogs(index, ev) - case ev := <-es.chainCh: - es.handleChainEvent(index, ev) - - case f := <-es.install: - if f.typ == MinedAndPendingLogsSubscription { - // the type are logs and pending logs subscriptions - index[LogsSubscription][f.id] = f - index[PendingLogsSubscription][f.id] = f - } else { - index[f.typ][f.id] = f - } - close(f.installed) - - case f := <-es.uninstall: - if f.typ == MinedAndPendingLogsSubscription { - // the type are logs and pending logs subscriptions - delete(index[LogsSubscription], f.id) - delete(index[PendingLogsSubscription], f.id) - } else { - delete(index[f.typ], f.id) - } - close(f.err) - - // System stopped - case <-es.txsSub.Err(): - return - case <-es.logsSub.Err(): - return - case <-es.rmLogsSub.Err(): - return - case <-es.chainSub.Err(): - return - } - } -} diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go deleted file mode 100644 index 27cad8826a..0000000000 --- a/eth/filters/filter_system_test.go +++ /dev/null @@ -1,977 +0,0 @@ -// Copyright 2016 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package filters - -import ( - "context" - "errors" - "fmt" - "math/big" - "math/rand" - "reflect" - "runtime" - "testing" - "time" - - "github.com/ethereum/go-ethereum" - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/consensus/ethash" - "github.com/ethereum/go-ethereum/core" - "github.com/ethereum/go-ethereum/core/bloombits" - "github.com/ethereum/go-ethereum/core/rawdb" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/event" - "github.com/ethereum/go-ethereum/internal/ethapi" - "github.com/ethereum/go-ethereum/params" - "github.com/ethereum/go-ethereum/rpc" -) - -type testBackend struct { - db ethdb.Database - sections uint64 - txFeed event.Feed - logsFeed event.Feed - rmLogsFeed event.Feed - pendingLogsFeed event.Feed - chainFeed event.Feed - pendingBlock *types.Block - pendingReceipts types.Receipts -} - -func (b *testBackend) ChainConfig() *params.ChainConfig { - return params.TestChainConfig -} - -func (b *testBackend) CurrentHeader() *types.Header { - hdr, _ := b.HeaderByNumber(context.TODO(), rpc.LatestBlockNumber) - return hdr -} - -func (b *testBackend) ChainDb() ethdb.Database { - return b.db -} - -func (b *testBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) { - var ( - hash common.Hash - num uint64 - ) - switch blockNr { - case rpc.LatestBlockNumber: - hash = rawdb.ReadHeadBlockHash(b.db) - number := rawdb.ReadHeaderNumber(b.db, hash) - if number == nil { - return nil, nil - } - num = *number - case rpc.FinalizedBlockNumber: - hash = rawdb.ReadFinalizedBlockHash(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) - } - return rawdb.ReadHeader(b.db, hash, num), nil -} - -func (b *testBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) { - number := rawdb.ReadHeaderNumber(b.db, hash) - if number == nil { - return nil, nil - } - return rawdb.ReadHeader(b.db, hash, *number), nil -} - -func (b *testBackend) GetBody(ctx context.Context, hash common.Hash, number rpc.BlockNumber) (*types.Body, error) { - if body := rawdb.ReadBody(b.db, hash, uint64(number)); body != nil { - return body, nil - } - return nil, errors.New("block body not found") -} - -func (b *testBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) { - if number := rawdb.ReadHeaderNumber(b.db, hash); number != nil { - if header := rawdb.ReadHeader(b.db, hash, *number); header != nil { - return rawdb.ReadReceipts(b.db, hash, *number, header.Time, params.TestChainConfig), nil - } - } - return nil, nil -} - -func (b *testBackend) GetLogs(ctx context.Context, hash common.Hash, number uint64) ([][]*types.Log, error) { - logs := rawdb.ReadLogs(b.db, hash, number) - return logs, nil -} - -func (b *testBackend) PendingBlockAndReceipts() (*types.Block, types.Receipts) { - return b.pendingBlock, b.pendingReceipts -} - -func (b *testBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription { - return b.txFeed.Subscribe(ch) -} - -func (b *testBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription { - return b.rmLogsFeed.Subscribe(ch) -} - -func (b *testBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription { - return b.logsFeed.Subscribe(ch) -} - -func (b *testBackend) SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription { - return b.pendingLogsFeed.Subscribe(ch) -} - -func (b *testBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription { - return b.chainFeed.Subscribe(ch) -} - -func (b *testBackend) BloomStatus() (uint64, uint64) { - return params.BloomBitsBlocks, b.sections -} - -func (b *testBackend) ServiceFilter(ctx context.Context, session *bloombits.MatcherSession) { - requests := make(chan chan *bloombits.Retrieval) - - go session.Multiplex(16, 0, requests) - go func() { - for { - // Wait for a service request or a shutdown - select { - case <-ctx.Done(): - return - - case request := <-requests: - task := <-request - - task.Bitsets = make([][]byte, len(task.Sections)) - for i, section := range task.Sections { - if rand.Int()%4 != 0 { // Handle occasional missing deliveries - head := rawdb.ReadCanonicalHash(b.db, (section+1)*params.BloomBitsBlocks-1) - task.Bitsets[i], _ = rawdb.ReadBloomBits(b.db, task.Bit, section, head) - } - } - request <- task - } - } - }() -} - -func newTestFilterSystem(t testing.TB, db ethdb.Database, cfg Config) (*testBackend, *FilterSystem) { - backend := &testBackend{db: db} - sys := NewFilterSystem(backend, cfg) - return backend, sys -} - -// TestBlockSubscription tests if a block subscription returns block hashes for posted chain events. -// It creates multiple subscriptions: -// - one at the start and should receive all posted chain events and a second (blockHashes) -// - one that is created after a cutoff moment and uninstalled after a second cutoff moment (blockHashes[cutoff1:cutoff2]) -// - one that is created after the second cutoff moment (blockHashes[cutoff2:]) -func TestBlockSubscription(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - backend, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - genesis = &core.Genesis{ - Config: params.TestChainConfig, - BaseFee: big.NewInt(params.InitialBaseFee), - } - _, chain, _ = core.GenerateChainWithGenesis(genesis, ethash.NewFaker(), 10, func(i int, gen *core.BlockGen) {}) - chainEvents = []core.ChainEvent{} - ) - - for _, blk := range chain { - chainEvents = append(chainEvents, core.ChainEvent{Hash: blk.Hash(), Block: blk}) - } - - chan0 := make(chan *types.Header) - sub0 := api.events.SubscribeNewHeads(chan0) - chan1 := make(chan *types.Header) - sub1 := api.events.SubscribeNewHeads(chan1) - - go func() { // simulate client - i1, i2 := 0, 0 - for i1 != len(chainEvents) || i2 != len(chainEvents) { - select { - case header := <-chan0: - if chainEvents[i1].Hash != header.Hash() { - t.Errorf("sub0 received invalid hash on index %d, want %x, got %x", i1, chainEvents[i1].Hash, header.Hash()) - } - i1++ - case header := <-chan1: - if chainEvents[i2].Hash != header.Hash() { - t.Errorf("sub1 received invalid hash on index %d, want %x, got %x", i2, chainEvents[i2].Hash, header.Hash()) - } - i2++ - } - } - - sub0.Unsubscribe() - sub1.Unsubscribe() - }() - - time.Sleep(1 * time.Second) - for _, e := range chainEvents { - backend.chainFeed.Send(e) - } - - <-sub0.Err() - <-sub1.Err() -} - -// TestPendingTxFilter tests whether pending tx filters retrieve all pending transactions that are posted to the event mux. -func TestPendingTxFilter(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - backend, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - - transactions = []*types.Transaction{ - types.NewTransaction(0, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - types.NewTransaction(1, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - types.NewTransaction(2, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - types.NewTransaction(3, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - types.NewTransaction(4, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - } - - hashes []common.Hash - ) - - fid0 := api.NewPendingTransactionFilter(nil) - - time.Sleep(1 * time.Second) - backend.txFeed.Send(core.NewTxsEvent{Txs: transactions}) - - timeout := time.Now().Add(1 * time.Second) - for { - results, err := api.GetFilterChanges(fid0) - if err != nil { - t.Fatalf("Unable to retrieve logs: %v", err) - } - - h := results.([]common.Hash) - hashes = append(hashes, h...) - if len(hashes) >= len(transactions) { - break - } - // check timeout - if time.Now().After(timeout) { - break - } - - time.Sleep(100 * time.Millisecond) - } - - if len(hashes) != len(transactions) { - t.Errorf("invalid number of transactions, want %d transactions(s), got %d", len(transactions), len(hashes)) - return - } - for i := range hashes { - if hashes[i] != transactions[i].Hash() { - t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), hashes[i]) - } - } -} - -// TestPendingTxFilterFullTx tests whether pending tx filters retrieve all pending transactions that are posted to the event mux. -func TestPendingTxFilterFullTx(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - backend, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - - transactions = []*types.Transaction{ - types.NewTransaction(0, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - types.NewTransaction(1, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - types.NewTransaction(2, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - types.NewTransaction(3, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - types.NewTransaction(4, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), - } - - txs []*ethapi.RPCTransaction - ) - - fullTx := true - fid0 := api.NewPendingTransactionFilter(&fullTx) - - time.Sleep(1 * time.Second) - backend.txFeed.Send(core.NewTxsEvent{Txs: transactions}) - - timeout := time.Now().Add(1 * time.Second) - for { - results, err := api.GetFilterChanges(fid0) - if err != nil { - t.Fatalf("Unable to retrieve logs: %v", err) - } - - tx := results.([]*ethapi.RPCTransaction) - txs = append(txs, tx...) - if len(txs) >= len(transactions) { - break - } - // check timeout - if time.Now().After(timeout) { - break - } - - time.Sleep(100 * time.Millisecond) - } - - if len(txs) != len(transactions) { - t.Errorf("invalid number of transactions, want %d transactions(s), got %d", len(transactions), len(txs)) - return - } - for i := range txs { - if txs[i].Hash != transactions[i].Hash() { - t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), txs[i].Hash) - } - } -} - -// TestLogFilterCreation test whether a given filter criteria makes sense. -// If not it must return an error. -func TestLogFilterCreation(t *testing.T) { - var ( - db = rawdb.NewMemoryDatabase() - _, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - - testCases = []struct { - crit FilterCriteria - success bool - }{ - // defaults - {FilterCriteria{}, true}, - // valid block number range - {FilterCriteria{FromBlock: big.NewInt(1), ToBlock: big.NewInt(2)}, true}, - // "mined" block range to pending - {FilterCriteria{FromBlock: big.NewInt(1), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())}, true}, - // new mined and pending blocks - {FilterCriteria{FromBlock: big.NewInt(rpc.LatestBlockNumber.Int64()), ToBlock: big.NewInt(rpc.PendingBlockNumber.Int64())}, true}, - // from block "higher" than to block - {FilterCriteria{FromBlock: big.NewInt(2), ToBlock: big.NewInt(1)}, false}, - // from block "higher" than to block - {FilterCriteria{FromBlock: big.NewInt(rpc.LatestBlockNumber.Int64()), ToBlock: big.NewInt(100)}, false}, - // from block "higher" than to block - {FilterCriteria{FromBlock: big.NewInt(rpc.PendingBlockNumber.Int64()), ToBlock: big.NewInt(100)}, false}, - // from block "higher" than to block - {FilterCriteria{FromBlock: big.NewInt(rpc.PendingBlockNumber.Int64()), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())}, false}, - // topics more then 4 - {FilterCriteria{Topics: [][]common.Hash{{}, {}, {}, {}, {}}}, false}, - } - ) - - for i, test := range testCases { - id, err := api.NewFilter(test.crit) - if err != nil && test.success { - t.Errorf("expected filter creation for case %d to success, got %v", i, err) - } - if err == nil { - api.UninstallFilter(id) - if !test.success { - t.Errorf("expected testcase %d to fail with an error", i) - } - } - } -} - -// TestInvalidLogFilterCreation tests whether invalid filter log criteria results in an error -// when the filter is created. -func TestInvalidLogFilterCreation(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - _, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - ) - - // different situations where log filter creation should fail. - // Reason: fromBlock > toBlock - testCases := []FilterCriteria{ - 0: {FromBlock: big.NewInt(rpc.PendingBlockNumber.Int64()), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())}, - 1: {FromBlock: big.NewInt(rpc.PendingBlockNumber.Int64()), ToBlock: big.NewInt(100)}, - 2: {FromBlock: big.NewInt(rpc.LatestBlockNumber.Int64()), ToBlock: big.NewInt(100)}, - 3: {Topics: [][]common.Hash{{}, {}, {}, {}, {}}}, - } - - for i, test := range testCases { - if _, err := api.NewFilter(test); err == nil { - t.Errorf("Expected NewFilter for case #%d to fail", i) - } - } -} - -// TestLogFilterUninstall tests invalid getLogs requests -func TestInvalidGetLogsRequest(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - _, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - blockHash = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111") - ) - - // Reason: Cannot specify both BlockHash and FromBlock/ToBlock) - testCases := []FilterCriteria{ - 0: {BlockHash: &blockHash, FromBlock: big.NewInt(100)}, - 1: {BlockHash: &blockHash, ToBlock: big.NewInt(500)}, - 2: {BlockHash: &blockHash, FromBlock: big.NewInt(rpc.LatestBlockNumber.Int64())}, - 3: {BlockHash: &blockHash, Topics: [][]common.Hash{{}, {}, {}, {}, {}}}, - } - - for i, test := range testCases { - if _, err := api.GetLogs(context.Background(), test); err == nil { - t.Errorf("Expected Logs for case #%d to fail", i) - } - } -} - -// TestInvalidGetRangeLogsRequest tests getLogs with invalid block range -func TestInvalidGetRangeLogsRequest(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - _, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - ) - - if _, err := api.GetLogs(context.Background(), FilterCriteria{FromBlock: big.NewInt(2), ToBlock: big.NewInt(1)}); err != errInvalidBlockRange { - t.Errorf("Expected Logs for invalid range return error, but got: %v", err) - } -} - -// TestLogFilter tests whether log filters match the correct logs that are posted to the event feed. -func TestLogFilter(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - backend, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - - firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111") - secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222") - thirdAddress = common.HexToAddress("0x3333333333333333333333333333333333333333") - notUsedAddress = common.HexToAddress("0x9999999999999999999999999999999999999999") - firstTopic = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111") - secondTopic = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222") - notUsedTopic = common.HexToHash("0x9999999999999999999999999999999999999999999999999999999999999999") - - // posted twice, once as regular logs and once as pending logs. - allLogs = []*types.Log{ - {Address: firstAddr}, - {Address: firstAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 1}, - {Address: secondAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 1}, - {Address: thirdAddress, Topics: []common.Hash{secondTopic}, BlockNumber: 2}, - {Address: thirdAddress, Topics: []common.Hash{secondTopic}, BlockNumber: 3}, - } - - expectedCase7 = []*types.Log{allLogs[3], allLogs[4], allLogs[0], allLogs[1], allLogs[2], allLogs[3], allLogs[4]} - expectedCase11 = []*types.Log{allLogs[1], allLogs[2], allLogs[1], allLogs[2]} - - testCases = []struct { - crit FilterCriteria - expected []*types.Log - id rpc.ID - }{ - // match all - 0: {FilterCriteria{}, allLogs, ""}, - // match none due to no matching addresses - 1: {FilterCriteria{Addresses: []common.Address{{}, notUsedAddress}, Topics: [][]common.Hash{nil}}, []*types.Log{}, ""}, - // match logs based on addresses, ignore topics - 2: {FilterCriteria{Addresses: []common.Address{firstAddr}}, allLogs[:2], ""}, - // match none due to no matching topics (match with address) - 3: {FilterCriteria{Addresses: []common.Address{secondAddr}, Topics: [][]common.Hash{{notUsedTopic}}}, []*types.Log{}, ""}, - // match logs based on addresses and topics - 4: {FilterCriteria{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}}, allLogs[3:5], ""}, - // match logs based on multiple addresses and "or" topics - 5: {FilterCriteria{Addresses: []common.Address{secondAddr, thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}}, allLogs[2:5], ""}, - // logs in the pending block - 6: {FilterCriteria{Addresses: []common.Address{firstAddr}, FromBlock: big.NewInt(rpc.PendingBlockNumber.Int64()), ToBlock: big.NewInt(rpc.PendingBlockNumber.Int64())}, allLogs[:2], ""}, - // mined logs with block num >= 2 or pending logs - 7: {FilterCriteria{FromBlock: big.NewInt(2), ToBlock: big.NewInt(rpc.PendingBlockNumber.Int64())}, expectedCase7, ""}, - // all "mined" logs with block num >= 2 - 8: {FilterCriteria{FromBlock: big.NewInt(2), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())}, allLogs[3:], ""}, - // all "mined" logs - 9: {FilterCriteria{ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())}, allLogs, ""}, - // all "mined" logs with 1>= block num <=2 and topic secondTopic - 10: {FilterCriteria{FromBlock: big.NewInt(1), ToBlock: big.NewInt(2), Topics: [][]common.Hash{{secondTopic}}}, allLogs[3:4], ""}, - // all "mined" and pending logs with topic firstTopic - 11: {FilterCriteria{FromBlock: big.NewInt(rpc.LatestBlockNumber.Int64()), ToBlock: big.NewInt(rpc.PendingBlockNumber.Int64()), Topics: [][]common.Hash{{firstTopic}}}, expectedCase11, ""}, - // match all logs due to wildcard topic - 12: {FilterCriteria{Topics: [][]common.Hash{nil}}, allLogs[1:], ""}, - } - ) - - // create all filters - for i := range testCases { - testCases[i].id, _ = api.NewFilter(testCases[i].crit) - } - - // raise events - time.Sleep(1 * time.Second) - if nsend := backend.logsFeed.Send(allLogs); nsend == 0 { - t.Fatal("Logs event not delivered") - } - if nsend := backend.pendingLogsFeed.Send(allLogs); nsend == 0 { - t.Fatal("Pending logs event not delivered") - } - - for i, tt := range testCases { - var fetched []*types.Log - timeout := time.Now().Add(1 * time.Second) - for { // fetch all expected logs - results, err := api.GetFilterChanges(tt.id) - if err != nil { - t.Fatalf("Unable to fetch logs: %v", err) - } - - fetched = append(fetched, results.([]*types.Log)...) - if len(fetched) >= len(tt.expected) { - break - } - // check timeout - if time.Now().After(timeout) { - break - } - - time.Sleep(100 * time.Millisecond) - } - - if len(fetched) != len(tt.expected) { - t.Errorf("invalid number of logs for case %d, want %d log(s), got %d", i, len(tt.expected), len(fetched)) - return - } - - for l := range fetched { - if fetched[l].Removed { - t.Errorf("expected log not to be removed for log %d in case %d", l, i) - } - if !reflect.DeepEqual(fetched[l], tt.expected[l]) { - t.Errorf("invalid log on index %d for case %d", l, i) - } - } - } -} - -// TestPendingLogsSubscription tests if a subscription receives the correct pending logs that are posted to the event feed. -func TestPendingLogsSubscription(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - backend, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, false) - - firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111") - secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222") - thirdAddress = common.HexToAddress("0x3333333333333333333333333333333333333333") - notUsedAddress = common.HexToAddress("0x9999999999999999999999999999999999999999") - firstTopic = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111") - secondTopic = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222") - thirdTopic = common.HexToHash("0x3333333333333333333333333333333333333333333333333333333333333333") - fourthTopic = common.HexToHash("0x4444444444444444444444444444444444444444444444444444444444444444") - notUsedTopic = common.HexToHash("0x9999999999999999999999999999999999999999999999999999999999999999") - - allLogs = [][]*types.Log{ - {{Address: firstAddr, Topics: []common.Hash{}, BlockNumber: 0}}, - {{Address: firstAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 1}}, - {{Address: secondAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 2}}, - {{Address: thirdAddress, Topics: []common.Hash{secondTopic}, BlockNumber: 3}}, - {{Address: thirdAddress, Topics: []common.Hash{secondTopic}, BlockNumber: 4}}, - { - {Address: thirdAddress, Topics: []common.Hash{firstTopic}, BlockNumber: 5}, - {Address: thirdAddress, Topics: []common.Hash{thirdTopic}, BlockNumber: 5}, - {Address: thirdAddress, Topics: []common.Hash{fourthTopic}, BlockNumber: 5}, - {Address: firstAddr, Topics: []common.Hash{firstTopic}, BlockNumber: 5}, - }, - } - - pendingBlockNumber = big.NewInt(rpc.PendingBlockNumber.Int64()) - - testCases = []struct { - crit ethereum.FilterQuery - expected []*types.Log - c chan []*types.Log - sub *Subscription - err chan error - }{ - // match all - { - ethereum.FilterQuery{FromBlock: pendingBlockNumber, ToBlock: pendingBlockNumber}, - flattenLogs(allLogs), - nil, nil, nil, - }, - // match none due to no matching addresses - { - ethereum.FilterQuery{Addresses: []common.Address{{}, notUsedAddress}, Topics: [][]common.Hash{nil}, FromBlock: pendingBlockNumber, ToBlock: pendingBlockNumber}, - nil, - nil, nil, nil, - }, - // match logs based on addresses, ignore topics - { - ethereum.FilterQuery{Addresses: []common.Address{firstAddr}, FromBlock: pendingBlockNumber, ToBlock: pendingBlockNumber}, - append(flattenLogs(allLogs[:2]), allLogs[5][3]), - nil, nil, nil, - }, - // match none due to no matching topics (match with address) - { - ethereum.FilterQuery{Addresses: []common.Address{secondAddr}, Topics: [][]common.Hash{{notUsedTopic}}, FromBlock: pendingBlockNumber, ToBlock: pendingBlockNumber}, - nil, - nil, nil, nil, - }, - // match logs based on addresses and topics - { - ethereum.FilterQuery{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}, FromBlock: pendingBlockNumber, ToBlock: pendingBlockNumber}, - append(flattenLogs(allLogs[3:5]), allLogs[5][0]), - nil, nil, nil, - }, - // match logs based on multiple addresses and "or" topics - { - ethereum.FilterQuery{Addresses: []common.Address{secondAddr, thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}, FromBlock: pendingBlockNumber, ToBlock: pendingBlockNumber}, - append(flattenLogs(allLogs[2:5]), allLogs[5][0]), - nil, nil, nil, - }, - // multiple pending logs, should match only 2 topics from the logs in block 5 - { - ethereum.FilterQuery{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{{firstTopic, fourthTopic}}, FromBlock: pendingBlockNumber, ToBlock: pendingBlockNumber}, - []*types.Log{allLogs[5][0], allLogs[5][2]}, - nil, nil, nil, - }, - // match none due to only matching new mined logs - { - ethereum.FilterQuery{}, - nil, - nil, nil, nil, - }, - // match none due to only matching mined logs within a specific block range - { - ethereum.FilterQuery{FromBlock: big.NewInt(1), ToBlock: big.NewInt(2)}, - nil, - nil, nil, nil, - }, - // match all due to matching mined and pending logs - { - ethereum.FilterQuery{FromBlock: big.NewInt(rpc.LatestBlockNumber.Int64()), ToBlock: big.NewInt(rpc.PendingBlockNumber.Int64())}, - flattenLogs(allLogs), - nil, nil, nil, - }, - // match none due to matching logs from a specific block number to new mined blocks - { - ethereum.FilterQuery{FromBlock: big.NewInt(1), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())}, - nil, - nil, nil, nil, - }, - } - ) - - // create all subscriptions, this ensures all subscriptions are created before the events are posted. - // on slow machines this could otherwise lead to missing events when the subscription is created after - // (some) events are posted. - for i := range testCases { - testCases[i].c = make(chan []*types.Log) - testCases[i].err = make(chan error, 1) - - var err error - testCases[i].sub, err = api.events.SubscribeLogs(testCases[i].crit, testCases[i].c) - if err != nil { - t.Fatalf("SubscribeLogs %d failed: %v\n", i, err) - } - } - - for n, test := range testCases { - i := n - tt := test - go func() { - defer tt.sub.Unsubscribe() - - var fetched []*types.Log - - timeout := time.After(1 * time.Second) - fetchLoop: - for { - select { - case logs := <-tt.c: - // Do not break early if we've fetched greater, or equal, - // to the number of logs expected. This ensures we do not - // deadlock the filter system because it will do a blocking - // send on this channel if another log arrives. - fetched = append(fetched, logs...) - case <-timeout: - break fetchLoop - } - } - - if len(fetched) != len(tt.expected) { - tt.err <- fmt.Errorf("invalid number of logs for case %d, want %d log(s), got %d", i, len(tt.expected), len(fetched)) - return - } - - for l := range fetched { - if fetched[l].Removed { - tt.err <- fmt.Errorf("expected log not to be removed for log %d in case %d", l, i) - return - } - if !reflect.DeepEqual(fetched[l], tt.expected[l]) { - tt.err <- fmt.Errorf("invalid log on index %d for case %d\n", l, i) - return - } - } - tt.err <- nil - }() - } - - // raise events - for _, ev := range allLogs { - backend.pendingLogsFeed.Send(ev) - } - - for i := range testCases { - err := <-testCases[i].err - if err != nil { - t.Fatalf("test %d failed: %v", i, err) - } - <-testCases[i].sub.Err() - } -} - -func TestLightFilterLogs(t *testing.T) { - t.Parallel() - - var ( - db = rawdb.NewMemoryDatabase() - backend, sys = newTestFilterSystem(t, db, Config{}) - api = NewFilterAPI(sys, true) - signer = types.HomesteadSigner{} - - firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111") - secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222") - thirdAddress = common.HexToAddress("0x3333333333333333333333333333333333333333") - notUsedAddress = common.HexToAddress("0x9999999999999999999999999999999999999999") - firstTopic = common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111") - secondTopic = common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222") - - // posted twice, once as regular logs and once as pending logs. - allLogs = []*types.Log{ - // Block 1 - {Address: firstAddr, Topics: []common.Hash{}, Data: []byte{}, BlockNumber: 2, Index: 0}, - // Block 2 - {Address: firstAddr, Topics: []common.Hash{firstTopic}, Data: []byte{}, BlockNumber: 3, Index: 0}, - {Address: secondAddr, Topics: []common.Hash{firstTopic}, Data: []byte{}, BlockNumber: 3, Index: 1}, - {Address: thirdAddress, Topics: []common.Hash{secondTopic}, Data: []byte{}, BlockNumber: 3, Index: 2}, - // Block 3 - {Address: thirdAddress, Topics: []common.Hash{secondTopic}, Data: []byte{}, BlockNumber: 4, Index: 0}, - } - - testCases = []struct { - crit FilterCriteria - expected []*types.Log - id rpc.ID - }{ - // match all - 0: {FilterCriteria{}, allLogs, ""}, - // match none due to no matching addresses - 1: {FilterCriteria{Addresses: []common.Address{{}, notUsedAddress}, Topics: [][]common.Hash{nil}}, []*types.Log{}, ""}, - // match logs based on addresses, ignore topics - 2: {FilterCriteria{Addresses: []common.Address{firstAddr}}, allLogs[:2], ""}, - // match logs based on addresses and topics - 3: {FilterCriteria{Addresses: []common.Address{thirdAddress}, Topics: [][]common.Hash{{firstTopic, secondTopic}}}, allLogs[3:5], ""}, - // all logs with block num >= 3 - 4: {FilterCriteria{FromBlock: big.NewInt(3), ToBlock: big.NewInt(5)}, allLogs[1:], ""}, - // all logs - 5: {FilterCriteria{FromBlock: big.NewInt(0), ToBlock: big.NewInt(5)}, allLogs, ""}, - // all logs with 1>= block num <=2 and topic secondTopic - 6: {FilterCriteria{FromBlock: big.NewInt(2), ToBlock: big.NewInt(3), Topics: [][]common.Hash{{secondTopic}}}, allLogs[3:4], ""}, - } - - key, _ = crypto.GenerateKey() - addr = crypto.PubkeyToAddress(key.PublicKey) - genesis = &core.Genesis{Config: params.TestChainConfig, - Alloc: core.GenesisAlloc{ - addr: {Balance: big.NewInt(params.Ether)}, - }, - } - receipts = []*types.Receipt{{ - Logs: []*types.Log{allLogs[0]}, - }, { - Logs: []*types.Log{allLogs[1], allLogs[2], allLogs[3]}, - }, { - Logs: []*types.Log{allLogs[4]}, - }} - ) - - _, blocks, _ := core.GenerateChainWithGenesis(genesis, ethash.NewFaker(), 4, func(i int, b *core.BlockGen) { - if i == 0 { - return - } - receipts[i-1].Bloom = types.CreateBloom(types.Receipts{receipts[i-1]}) - b.AddUncheckedReceipt(receipts[i-1]) - tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{Nonce: uint64(i - 1), To: &common.Address{}, Value: big.NewInt(1000), Gas: params.TxGas, GasPrice: b.BaseFee(), Data: nil}), signer, key) - b.AddTx(tx) - }) - for i, block := range blocks { - rawdb.WriteBlock(db, block) - rawdb.WriteCanonicalHash(db, block.Hash(), block.NumberU64()) - rawdb.WriteHeadBlockHash(db, block.Hash()) - if i > 0 { - rawdb.WriteReceipts(db, block.Hash(), block.NumberU64(), []*types.Receipt{receipts[i-1]}) - } - } - // create all filters - for i := range testCases { - id, err := api.NewFilter(testCases[i].crit) - if err != nil { - t.Fatal(err) - } - testCases[i].id = id - } - - // raise events - time.Sleep(1 * time.Second) - for _, block := range blocks { - backend.chainFeed.Send(core.ChainEvent{Block: block, Hash: common.Hash{}, Logs: allLogs}) - } - - for i, tt := range testCases { - var fetched []*types.Log - timeout := time.Now().Add(1 * time.Second) - for { // fetch all expected logs - results, err := api.GetFilterChanges(tt.id) - if err != nil { - t.Fatalf("Unable to fetch logs: %v", err) - } - fetched = append(fetched, results.([]*types.Log)...) - if len(fetched) >= len(tt.expected) { - break - } - // check timeout - if time.Now().After(timeout) { - break - } - - time.Sleep(100 * time.Millisecond) - } - - if len(fetched) != len(tt.expected) { - t.Errorf("invalid number of logs for case %d, want %d log(s), got %d", i, len(tt.expected), len(fetched)) - return - } - - for l := range fetched { - if fetched[l].Removed { - t.Errorf("expected log not to be removed for log %d in case %d", l, i) - } - expected := *tt.expected[l] - blockNum := expected.BlockNumber - 1 - expected.BlockHash = blocks[blockNum].Hash() - expected.TxHash = blocks[blockNum].Transactions()[0].Hash() - if !reflect.DeepEqual(fetched[l], &expected) { - t.Errorf("invalid log on index %d for case %d", l, i) - } - } - } -} - -// TestPendingTxFilterDeadlock tests if the event loop hangs when pending -// txes arrive at the same time that one of multiple filters is timing out. -// Please refer to #22131 for more details. -func TestPendingTxFilterDeadlock(t *testing.T) { - t.Parallel() - timeout := 100 * time.Millisecond - - var ( - db = rawdb.NewMemoryDatabase() - backend, sys = newTestFilterSystem(t, db, Config{Timeout: timeout}) - api = NewFilterAPI(sys, false) - done = make(chan struct{}) - ) - - go func() { - // Bombard feed with txes until signal was received to stop - i := uint64(0) - for { - select { - case <-done: - return - default: - } - - tx := types.NewTransaction(i, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil) - backend.txFeed.Send(core.NewTxsEvent{Txs: []*types.Transaction{tx}}) - i++ - } - }() - - // Create a bunch of filters that will - // timeout either in 100ms or 200ms - subs := make([]*Subscription, 20) - for i := 0; i < len(subs); i++ { - fid := api.NewPendingTransactionFilter(nil) - f, ok := api.filters[fid] - if !ok { - t.Fatalf("Filter %s should exist", fid) - } - subs[i] = f.s - // Wait for at least one tx to arrive in filter - for { - hashes, err := api.GetFilterChanges(fid) - if err != nil { - t.Fatalf("Filter should exist: %v\n", err) - } - if len(hashes.([]common.Hash)) > 0 { - break - } - runtime.Gosched() - } - } - - // Wait until filters have timed out and have been uninstalled. - for _, sub := range subs { - select { - case <-sub.Err(): - case <-time.After(1 * time.Second): - t.Fatalf("Filter timeout is hanging") - } - } -} - -func flattenLogs(pl [][]*types.Log) []*types.Log { - var logs []*types.Log - for _, l := range pl { - logs = append(logs, l...) - } - return logs -} diff --git a/eth/filters/filter_test.go b/eth/filters/filter_test.go deleted file mode 100644 index 1db917c960..0000000000 --- a/eth/filters/filter_test.go +++ /dev/null @@ -1,389 +0,0 @@ -// Copyright 2015 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package filters - -import ( - "context" - "encoding/json" - "math/big" - "strings" - "testing" - "time" - - "github.com/ethereum/go-ethereum/accounts/abi" - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/consensus/ethash" - "github.com/ethereum/go-ethereum/core" - "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/params" - "github.com/ethereum/go-ethereum/rpc" - "github.com/ethereum/go-ethereum/trie" -) - -func makeReceipt(addr common.Address) *types.Receipt { - receipt := types.NewReceipt(nil, false, 0) - receipt.Logs = []*types.Log{ - {Address: addr}, - } - receipt.Bloom = types.CreateBloom(types.Receipts{receipt}) - return receipt -} - -func BenchmarkFilters(b *testing.B) { - var ( - db, _ = rawdb.NewLevelDBDatabase(b.TempDir(), 0, 0, "", false) - _, sys = newTestFilterSystem(b, db, Config{}) - key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") - addr1 = crypto.PubkeyToAddress(key1.PublicKey) - addr2 = common.BytesToAddress([]byte("jeff")) - addr3 = common.BytesToAddress([]byte("ethereum")) - addr4 = common.BytesToAddress([]byte("random addresses please")) - - gspec = &core.Genesis{ - Alloc: core.GenesisAlloc{addr1: {Balance: big.NewInt(1000000)}}, - BaseFee: big.NewInt(params.InitialBaseFee), - Config: params.TestChainConfig, - } - ) - defer db.Close() - _, chain, receipts := core.GenerateChainWithGenesis(gspec, ethash.NewFaker(), 100010, func(i int, gen *core.BlockGen) { - switch i { - case 2403: - receipt := makeReceipt(addr1) - gen.AddUncheckedReceipt(receipt) - gen.AddUncheckedTx(types.NewTransaction(999, common.HexToAddress("0x999"), big.NewInt(999), 999, gen.BaseFee(), nil)) - case 1034: - receipt := makeReceipt(addr2) - gen.AddUncheckedReceipt(receipt) - gen.AddUncheckedTx(types.NewTransaction(999, common.HexToAddress("0x999"), big.NewInt(999), 999, gen.BaseFee(), nil)) - case 34: - receipt := makeReceipt(addr3) - gen.AddUncheckedReceipt(receipt) - gen.AddUncheckedTx(types.NewTransaction(999, common.HexToAddress("0x999"), big.NewInt(999), 999, gen.BaseFee(), nil)) - case 99999: - receipt := makeReceipt(addr4) - gen.AddUncheckedReceipt(receipt) - gen.AddUncheckedTx(types.NewTransaction(999, common.HexToAddress("0x999"), big.NewInt(999), 999, gen.BaseFee(), nil)) - } - }) - // The test txs are not properly signed, can't simply create a chain - // and then import blocks. TODO(rjl493456442) try to get rid of the - // manual database writes. - gspec.MustCommit(db, trie.NewDatabase(db, trie.HashDefaults)) - - for i, block := range chain { - rawdb.WriteBlock(db, block) - rawdb.WriteCanonicalHash(db, block.Hash(), block.NumberU64()) - rawdb.WriteHeadBlockHash(db, block.Hash()) - rawdb.WriteReceipts(db, block.Hash(), block.NumberU64(), receipts[i]) - } - b.ResetTimer() - - filter := sys.NewRangeFilter(0, -1, []common.Address{addr1, addr2, addr3, addr4}, nil) - - for i := 0; i < b.N; i++ { - logs, _ := filter.Logs(context.Background()) - if len(logs) != 4 { - b.Fatal("expected 4 logs, got", len(logs)) - } - } -} - -func TestFilters(t *testing.T) { - var ( - db = rawdb.NewMemoryDatabase() - _, sys = newTestFilterSystem(t, db, Config{}) - // Sender account - key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") - addr = crypto.PubkeyToAddress(key1.PublicKey) - signer = types.NewLondonSigner(big.NewInt(1)) - // Logging contract - contract = common.Address{0xfe} - contract2 = common.Address{0xff} - abiStr = `[{"inputs":[],"name":"log0","outputs":[],"stateMutability":"nonpayable","type":"function"},{"inputs":[{"internalType":"uint256","name":"t1","type":"uint256"}],"name":"log1","outputs":[],"stateMutability":"nonpayable","type":"function"},{"inputs":[{"internalType":"uint256","name":"t1","type":"uint256"},{"internalType":"uint256","name":"t2","type":"uint256"}],"name":"log2","outputs":[],"stateMutability":"nonpayable","type":"function"},{"inputs":[{"internalType":"uint256","name":"t1","type":"uint256"},{"internalType":"uint256","name":"t2","type":"uint256"},{"internalType":"uint256","name":"t3","type":"uint256"}],"name":"log3","outputs":[],"stateMutability":"nonpayable","type":"function"},{"inputs":[{"internalType":"uint256","name":"t1","type":"uint256"},{"internalType":"uint256","name":"t2","type":"uint256"},{"internalType":"uint256","name":"t3","type":"uint256"},{"internalType":"uint256","name":"t4","type":"uint256"}],"name":"log4","outputs":[],"stateMutability":"nonpayable","type":"function"}]` - /* - // SPDX-License-Identifier: GPL-3.0 - pragma solidity >=0.7.0 <0.9.0; - - contract Logger { - function log0() external { - assembly { - log0(0, 0) - } - } - - function log1(uint t1) external { - assembly { - log1(0, 0, t1) - } - } - - function log2(uint t1, uint t2) external { - assembly { - log2(0, 0, t1, t2) - } - } - - function log3(uint t1, uint t2, uint t3) external { - assembly { - log3(0, 0, t1, t2, t3) - } - } - - function log4(uint t1, uint t2, uint t3, uint t4) external { - assembly { - log4(0, 0, t1, t2, t3, t4) - } - } - } - */ - bytecode = common.FromHex("608060405234801561001057600080fd5b50600436106100575760003560e01c80630aa731851461005c5780632a4c08961461006657806378b9a1f314610082578063c670f8641461009e578063c683d6a3146100ba575b600080fd5b6100646100d6565b005b610080600480360381019061007b9190610143565b6100dc565b005b61009c60048036038101906100979190610196565b6100e8565b005b6100b860048036038101906100b391906101d6565b6100f2565b005b6100d460048036038101906100cf9190610203565b6100fa565b005b600080a0565b808284600080a3505050565b8082600080a25050565b80600080a150565b80828486600080a450505050565b600080fd5b6000819050919050565b6101208161010d565b811461012b57600080fd5b50565b60008135905061013d81610117565b92915050565b60008060006060848603121561015c5761015b610108565b5b600061016a8682870161012e565b935050602061017b8682870161012e565b925050604061018c8682870161012e565b9150509250925092565b600080604083850312156101ad576101ac610108565b5b60006101bb8582860161012e565b92505060206101cc8582860161012e565b9150509250929050565b6000602082840312156101ec576101eb610108565b5b60006101fa8482850161012e565b91505092915050565b6000806000806080858703121561021d5761021c610108565b5b600061022b8782880161012e565b945050602061023c8782880161012e565b935050604061024d8782880161012e565b925050606061025e8782880161012e565b9150509295919450925056fea264697066735822122073a4b156f487e59970dc1ef449cc0d51467268f676033a17188edafcee861f9864736f6c63430008110033") - - hash1 = common.BytesToHash([]byte("topic1")) - hash2 = common.BytesToHash([]byte("topic2")) - hash3 = common.BytesToHash([]byte("topic3")) - hash4 = common.BytesToHash([]byte("topic4")) - hash5 = common.BytesToHash([]byte("topic5")) - - gspec = &core.Genesis{ - Config: params.TestChainConfig, - Alloc: core.GenesisAlloc{ - addr: {Balance: big.NewInt(0).Mul(big.NewInt(100), big.NewInt(params.Ether))}, - contract: {Balance: big.NewInt(0), Code: bytecode}, - contract2: {Balance: big.NewInt(0), Code: bytecode}, - }, - BaseFee: big.NewInt(params.InitialBaseFee), - } - ) - - contractABI, err := abi.JSON(strings.NewReader(abiStr)) - if err != nil { - t.Fatal(err) - } - - // Hack: GenerateChainWithGenesis creates a new db. - // Commit the genesis manually and use GenerateChain. - _, err = gspec.Commit(db, trie.NewDatabase(db, nil)) - if err != nil { - t.Fatal(err) - } - chain, _ := core.GenerateChain(gspec.Config, gspec.ToBlock(), ethash.NewFaker(), db, 1000, func(i int, gen *core.BlockGen) { - switch i { - case 1: - data, err := contractABI.Pack("log1", hash1.Big()) - if err != nil { - t.Fatal(err) - } - tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{ - Nonce: 0, - GasPrice: gen.BaseFee(), - Gas: 30000, - To: &contract, - Data: data, - }), signer, key1) - gen.AddTx(tx) - tx2, _ := types.SignTx(types.NewTx(&types.LegacyTx{ - Nonce: 1, - GasPrice: gen.BaseFee(), - Gas: 30000, - To: &contract2, - Data: data, - }), signer, key1) - gen.AddTx(tx2) - case 2: - data, err := contractABI.Pack("log2", hash2.Big(), hash1.Big()) - if err != nil { - t.Fatal(err) - } - tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{ - Nonce: 2, - GasPrice: gen.BaseFee(), - Gas: 30000, - To: &contract, - Data: data, - }), signer, key1) - gen.AddTx(tx) - case 998: - data, err := contractABI.Pack("log1", hash3.Big()) - if err != nil { - t.Fatal(err) - } - tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{ - Nonce: 3, - GasPrice: gen.BaseFee(), - Gas: 30000, - To: &contract2, - Data: data, - }), signer, key1) - gen.AddTx(tx) - case 999: - data, err := contractABI.Pack("log1", hash4.Big()) - if err != nil { - t.Fatal(err) - } - tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{ - Nonce: 4, - GasPrice: gen.BaseFee(), - Gas: 30000, - To: &contract, - Data: data, - }), signer, key1) - gen.AddTx(tx) - } - }) - var l uint64 - bc, err := core.NewBlockChain(db, nil, gspec, nil, ethash.NewFaker(), vm.Config{}, nil, &l) - if err != nil { - t.Fatal(err) - } - _, err = bc.InsertChain(chain) - if err != nil { - t.Fatal(err) - } - - // Set block 998 as Finalized (-3) - bc.SetFinalized(chain[998].Header()) - - // Generate pending block - pchain, preceipts := core.GenerateChain(gspec.Config, chain[len(chain)-1], ethash.NewFaker(), db, 1, func(i int, gen *core.BlockGen) { - data, err := contractABI.Pack("log1", hash5.Big()) - if err != nil { - t.Fatal(err) - } - tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{ - Nonce: 5, - GasPrice: gen.BaseFee(), - Gas: 30000, - To: &contract, - Data: data, - }), signer, key1) - gen.AddTx(tx) - }) - sys.backend.(*testBackend).pendingBlock = pchain[0] - sys.backend.(*testBackend).pendingReceipts = preceipts[0] - - for i, tc := range []struct { - f *Filter - want string - err string - }{ - { - f: sys.NewBlockFilter(chain[2].Hash(), []common.Address{contract}, nil), - want: `[{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696332","0x0000000000000000000000000000000000000000000000000000746f70696331"],"data":"0x","blockNumber":"0x3","transactionHash":"0xdefe471992a07a02acdfbe33edaae22fbb86d7d3cec3f1b8e4e77702fb3acc1d","transactionIndex":"0x0","blockHash":"0x7a7556792ca7d37882882e2b001fe14833eaf81c2c7f865c9c771ec37a024f6b","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), []common.Address{contract}, [][]common.Hash{{hash1, hash2, hash3, hash4}}), - want: `[{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696331"],"data":"0x","blockNumber":"0x2","transactionHash":"0xa8028c655b6423204c8edfbc339f57b042d6bec2b6a61145d76b7c08b4cccd42","transactionIndex":"0x0","blockHash":"0x24417bb49ce44cfad65da68f33b510bf2a129c0d89ccf06acb6958b8585ccf34","logIndex":"0x0","removed":false},{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696332","0x0000000000000000000000000000000000000000000000000000746f70696331"],"data":"0x","blockNumber":"0x3","transactionHash":"0xdefe471992a07a02acdfbe33edaae22fbb86d7d3cec3f1b8e4e77702fb3acc1d","transactionIndex":"0x0","blockHash":"0x7a7556792ca7d37882882e2b001fe14833eaf81c2c7f865c9c771ec37a024f6b","logIndex":"0x0","removed":false},{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696334"],"data":"0x","blockNumber":"0x3e8","transactionHash":"0x9a87842100a638dfa5da8842b4beda691d2fd77b0c84b57f24ecfa9fb208f747","transactionIndex":"0x0","blockHash":"0xb360bad5265261c075ece02d3bf0e39498a6a76310482cdfd90588748e6c5ee0","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(900, 999, []common.Address{contract}, [][]common.Hash{{hash3}}), - }, - { - f: sys.NewRangeFilter(990, int64(rpc.LatestBlockNumber), []common.Address{contract2}, [][]common.Hash{{hash3}}), - want: `[{"address":"0xff00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696333"],"data":"0x","blockNumber":"0x3e7","transactionHash":"0x53e3675800c6908424b61b35a44e51ca4c73ca603e58a65b32c67968b4f42200","transactionIndex":"0x0","blockHash":"0x2e4620a2b426b0612ec6cad9603f466723edaed87f98c9137405dd4f7a2409ff","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(1, 10, []common.Address{contract}, [][]common.Hash{{hash2}, {hash1}}), - want: `[{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696332","0x0000000000000000000000000000000000000000000000000000746f70696331"],"data":"0x","blockNumber":"0x3","transactionHash":"0xdefe471992a07a02acdfbe33edaae22fbb86d7d3cec3f1b8e4e77702fb3acc1d","transactionIndex":"0x0","blockHash":"0x7a7556792ca7d37882882e2b001fe14833eaf81c2c7f865c9c771ec37a024f6b","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(1, 10, nil, [][]common.Hash{{hash1, hash2}}), - want: `[{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696331"],"data":"0x","blockNumber":"0x2","transactionHash":"0xa8028c655b6423204c8edfbc339f57b042d6bec2b6a61145d76b7c08b4cccd42","transactionIndex":"0x0","blockHash":"0x24417bb49ce44cfad65da68f33b510bf2a129c0d89ccf06acb6958b8585ccf34","logIndex":"0x0","removed":false},{"address":"0xff00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696331"],"data":"0x","blockNumber":"0x2","transactionHash":"0xdba3e2ea9a7d690b722d70ee605fd67ba4c00d1d3aecd5cf187a7b92ad8eb3df","transactionIndex":"0x1","blockHash":"0x24417bb49ce44cfad65da68f33b510bf2a129c0d89ccf06acb6958b8585ccf34","logIndex":"0x1","removed":false},{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696332","0x0000000000000000000000000000000000000000000000000000746f70696331"],"data":"0x","blockNumber":"0x3","transactionHash":"0xdefe471992a07a02acdfbe33edaae22fbb86d7d3cec3f1b8e4e77702fb3acc1d","transactionIndex":"0x0","blockHash":"0x7a7556792ca7d37882882e2b001fe14833eaf81c2c7f865c9c771ec37a024f6b","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), nil, [][]common.Hash{{common.BytesToHash([]byte("fail"))}}), - }, - { - f: sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), []common.Address{common.BytesToAddress([]byte("failmenow"))}, nil), - }, - { - f: sys.NewRangeFilter(0, int64(rpc.LatestBlockNumber), nil, [][]common.Hash{{common.BytesToHash([]byte("fail"))}, {hash1}}), - }, - { - f: sys.NewRangeFilter(int64(rpc.LatestBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), - want: `[{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696334"],"data":"0x","blockNumber":"0x3e8","transactionHash":"0x9a87842100a638dfa5da8842b4beda691d2fd77b0c84b57f24ecfa9fb208f747","transactionIndex":"0x0","blockHash":"0xb360bad5265261c075ece02d3bf0e39498a6a76310482cdfd90588748e6c5ee0","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(int64(rpc.FinalizedBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), - want: `[{"address":"0xff00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696333"],"data":"0x","blockNumber":"0x3e7","transactionHash":"0x53e3675800c6908424b61b35a44e51ca4c73ca603e58a65b32c67968b4f42200","transactionIndex":"0x0","blockHash":"0x2e4620a2b426b0612ec6cad9603f466723edaed87f98c9137405dd4f7a2409ff","logIndex":"0x0","removed":false},{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696334"],"data":"0x","blockNumber":"0x3e8","transactionHash":"0x9a87842100a638dfa5da8842b4beda691d2fd77b0c84b57f24ecfa9fb208f747","transactionIndex":"0x0","blockHash":"0xb360bad5265261c075ece02d3bf0e39498a6a76310482cdfd90588748e6c5ee0","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(int64(rpc.FinalizedBlockNumber), int64(rpc.FinalizedBlockNumber), nil, nil), - want: `[{"address":"0xff00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696333"],"data":"0x","blockNumber":"0x3e7","transactionHash":"0x53e3675800c6908424b61b35a44e51ca4c73ca603e58a65b32c67968b4f42200","transactionIndex":"0x0","blockHash":"0x2e4620a2b426b0612ec6cad9603f466723edaed87f98c9137405dd4f7a2409ff","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(int64(rpc.LatestBlockNumber), int64(rpc.FinalizedBlockNumber), nil, nil), - }, - { - f: sys.NewRangeFilter(int64(rpc.SafeBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), - err: "safe header not found", - }, - { - f: sys.NewRangeFilter(int64(rpc.SafeBlockNumber), int64(rpc.SafeBlockNumber), nil, nil), - err: "safe header not found", - }, - { - f: sys.NewRangeFilter(int64(rpc.LatestBlockNumber), int64(rpc.SafeBlockNumber), nil, nil), - err: "safe header not found", - }, - { - f: sys.NewRangeFilter(int64(rpc.PendingBlockNumber), int64(rpc.PendingBlockNumber), nil, nil), - want: `[{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696335"],"data":"0x","blockNumber":"0x3e9","transactionHash":"0x4110587c1b8d86edc85dce929a34127f1cb8809515a9f177c91c866de3eb0638","transactionIndex":"0x0","blockHash":"0xd5e8d4e4eb51a2a2a6ec20ef68a4c2801240743c8deb77a6a1d118ac3eefb725","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(int64(rpc.LatestBlockNumber), int64(rpc.PendingBlockNumber), nil, nil), - want: `[{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696334"],"data":"0x","blockNumber":"0x3e8","transactionHash":"0x9a87842100a638dfa5da8842b4beda691d2fd77b0c84b57f24ecfa9fb208f747","transactionIndex":"0x0","blockHash":"0xb360bad5265261c075ece02d3bf0e39498a6a76310482cdfd90588748e6c5ee0","logIndex":"0x0","removed":false},{"address":"0xfe00000000000000000000000000000000000000","topics":["0x0000000000000000000000000000000000000000000000000000746f70696335"],"data":"0x","blockNumber":"0x3e9","transactionHash":"0x4110587c1b8d86edc85dce929a34127f1cb8809515a9f177c91c866de3eb0638","transactionIndex":"0x0","blockHash":"0xd5e8d4e4eb51a2a2a6ec20ef68a4c2801240743c8deb77a6a1d118ac3eefb725","logIndex":"0x0","removed":false}]`, - }, - { - f: sys.NewRangeFilter(int64(rpc.PendingBlockNumber), int64(rpc.LatestBlockNumber), nil, nil), - err: errInvalidBlockRange.Error(), - }, - } { - logs, err := tc.f.Logs(context.Background()) - if err == nil && tc.err != "" { - t.Fatalf("test %d, expected error %q, got nil", i, tc.err) - } else if err != nil && err.Error() != tc.err { - t.Fatalf("test %d, expected error %q, got %q", i, tc.err, err.Error()) - } - if tc.want == "" && len(logs) == 0 { - continue - } - have, err := json.Marshal(logs) - if err != nil { - t.Fatal(err) - } - if string(have) != tc.want { - t.Fatalf("test %d, have:\n%s\nwant:\n%s", i, have, tc.want) - } - } - - t.Run("timeout", func(t *testing.T) { - f := sys.NewRangeFilter(0, -1, nil, nil) - ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Hour)) - defer cancel() - _, err := f.Logs(ctx) - if err == nil { - t.Fatal("expected error") - } - if err != context.DeadlineExceeded { - t.Fatalf("expected context.DeadlineExceeded, got %v", err) - } - }) -}