From 6ba9b3b775a8cd60cc2283290b1de736bcf63d8e Mon Sep 17 00:00:00 2001 From: Nick Johnson Date: Mon, 24 Sep 2018 11:40:41 +0100 Subject: [PATCH] Add support for fetching logs based on blockhash ranges --- eth/filters/api.go | 63 ++++++++++++++------- eth/filters/filter.go | 128 +++++++++++++++++++++++++++++++----------- interfaces.go | 10 ++-- 3 files changed, 144 insertions(+), 57 deletions(-) diff --git a/eth/filters/api.go b/eth/filters/api.go index 5ed80a8875..362fe73181 100644 --- a/eth/filters/api.go +++ b/eth/filters/api.go @@ -320,6 +320,24 @@ func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) { return logsSub.ID, nil } +func (api *PublicFilterAPI) getBlockHash(ctx context.Context, number *big.Int, hash *common.Hash) (common.Hash, error) { + if hash != nil { + return *hash, nil + } else if number != nil { + header, err := api.backend.HeaderByNumber(ctx, rpc.BlockNumber(number.Int64())) + if err != nil { + return common.Hash{}, err + } + return header.Hash(), nil + } else { + header, err := api.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber) + if err != nil { + return common.Hash{}, err + } + return header.Hash(), nil + } +} + // GetLogs returns logs matching the given argument that are stored within the state. // // https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs @@ -329,14 +347,13 @@ func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([ // Block filter requested, construct a single-shot filter filter = NewBlockFilter(api.backend, *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() + begin, err := api.getBlockHash(ctx, crit.FromBlock, crit.FromBlockHash) + if err != nil { + return nil, err } - end := rpc.LatestBlockNumber.Int64() - if crit.ToBlock != nil { - end = crit.ToBlock.Int64() + end, err := api.getBlockHash(ctx, crit.ToBlock, crit.ToBlockHash) + if err != nil { + return nil, err } // Construct the range filter filter = NewRangeFilter(api.backend, begin, end, crit.Addresses, crit.Topics) @@ -384,14 +401,13 @@ func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*ty // Block filter requested, construct a single-shot filter filter = NewBlockFilter(api.backend, *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() + begin, err := api.getBlockHash(ctx, f.crit.FromBlock, f.crit.FromBlockHash) + if err != nil { + return nil, err } - end := rpc.LatestBlockNumber.Int64() - if f.crit.ToBlock != nil { - end = f.crit.ToBlock.Int64() + end, err := api.getBlockHash(ctx, f.crit.ToBlock, f.crit.ToBlockHash) + if err != nil { + return nil, err } // Construct the range filter filter = NewRangeFilter(api.backend, begin, end, f.crit.Addresses, f.crit.Topics) @@ -459,11 +475,13 @@ func returnLogs(logs []*types.Log) []*types.Log { // 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"` + BlockHash *common.Hash `json:"blockHash"` + FromBlock *rpc.BlockNumber `json:"fromBlock"` + FromBlockHash *common.Hash `json:"fromBlockHash"` + ToBlock *rpc.BlockNumber `json:"toBlock"` + ToBlockHash *common.Hash `json:"toBlockHash"` + Addresses interface{} `json:"address"` + Topics []interface{} `json:"topics"` } var raw input @@ -481,10 +499,15 @@ func (args *FilterCriteria) UnmarshalJSON(data []byte) error { if raw.FromBlock != nil { args.FromBlock = big.NewInt(raw.FromBlock.Int64()) } - + if raw.FromBlockHash != nil { + args.FromBlockHash = raw.FromBlockHash + } if raw.ToBlock != nil { args.ToBlock = big.NewInt(raw.ToBlock.Int64()) } + if raw.ToBlockHash != nil { + args.ToBlockHash = raw.ToBlockHash + } } args.Addresses = []common.Address{} diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 071613ad7a..492e1c32d9 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -56,14 +56,14 @@ type Filter struct { topics [][]common.Hash block common.Hash // Block hash if filtering a single block - begin, end int64 // Range interval if filtering multiple blocks + begin, end common.Hash // 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 NewRangeFilter(backend Backend, begin, end int64, addresses []common.Address, topics [][]common.Hash) *Filter { +func NewRangeFilter(backend Backend, begin, end common.Hash, 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. @@ -114,9 +114,49 @@ func newFilter(backend Backend, addresses []common.Address, topics [][]common.Ha } } +func (f *Filter) findCommonAncestor(ctx context.Context, begin, end *types.Header) (*types.Header, bool, error) { + var err error + var mainChain bool + + // If end is on the canonical chain, we can rewind efficiently + if header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(end.Number.Int64())); err != nil && header.Hash() == end.Hash() { + mainChain = true + end, err = f.backend.HeaderByNumber(ctx, rpc.BlockNumber(begin.Number.Int64())) + if err != nil { + return nil, false, err + } + } else { + mainChain = false + // Rewind until begin and end are at the same height + for end.Number.Cmp(begin.Number) > 0 { + end, err = f.backend.HeaderByHash(ctx, end.ParentHash) + if err != nil { + return nil, false, err + } + } + } + + // Rewind both until they match + for begin.Hash() != end.Hash() { + begin, err = f.backend.HeaderByHash(ctx, begin.ParentHash) + if err != nil { + return nil, false, err + } + + end, err = f.backend.HeaderByHash(ctx, end.ParentHash) + if err != nil { + return nil, false, err + } + } + + return end, mainChain, nil +} + // 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) { + var err error + // If we're doing singleton block filtering, execute and return if f.block != (common.Hash{}) { header, err := f.backend.HeaderByHash(ctx, f.block) @@ -128,48 +168,72 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) { } return f.blockLogs(ctx, header) } + // Figure out the limits of the filter range header, _ := f.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber) if header == nil { return nil, nil } - head := header.Number.Uint64() - if f.begin == -1 { - f.begin = int64(head) - } - end := uint64(f.end) - if f.end == -1 { - end = head - } - // Gather all indexed logs, and finish with non indexed ones - var ( - logs []*types.Log - err error - ) - size, sections := f.backend.BloomStatus() - if indexed := sections * size; indexed > uint64(f.begin) { - if indexed > end { - logs, err = f.indexedLogs(ctx, end) - } else { - logs, err = f.indexedLogs(ctx, indexed-1) - } + begin := header + if f.begin != (common.Hash{}) { + begin, err = f.backend.HeaderByHash(ctx, f.begin) if err != nil { - return logs, err + return nil, err } } - rest, err := f.unindexedLogs(ctx, end) + + end := header + if f.end != (common.Hash{}) { + end, err = f.backend.HeaderByHash(ctx, f.end) + if err != nil { + return nil, err + } + } + + ancestor, mainChain, err := f.findCommonAncestor(ctx, begin, end) + if err != nil { + return nil, err + } + + // Insert deletions of any reorg-ed logs + var logs []*types.Log + removed, err := f.unindexedLogs(ctx, ancestor.Hash(), begin.Hash()) + if err != nil { + return nil, err + } + for _, log := range removed { + log.Removed = true + logs = append(logs, log) + } + + // Gather all indexed logs, and finish with non indexed ones + if mainChain { + size, sections := f.backend.BloomStatus() + if indexed := sections * size; indexed > begin.Number.Uint64() { + if indexed > end.Number.Uint64() { + logs, err = f.indexedLogs(ctx, begin.Number.Uint64(), end.Number.Uint64()) + } else { + logs, err = f.indexedLogs(ctx, begin.Number.Uint64(), indexed-1) + } + if err != nil { + return logs, err + } + } + } + rest, err := f.unindexedLogs(ctx, begin.Hash(), end.Hash()) logs = append(logs, rest...) + f.begin = end.Hash() return logs, err } // 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) ([]*types.Log, error) { +func (f *Filter) indexedLogs(ctx context.Context, begin, end uint64) ([]*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) + session, err := f.matcher.Start(ctx, begin, end, matches) if err != nil { return nil, err } @@ -186,18 +250,15 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err // Abort if all matches have been fulfilled if !ok { err := session.Error() - if err == nil { - f.begin = int64(end) + 1 - } return logs, err } - f.begin = int64(number) + 1 // Retrieve the suggested block and pull any truly matching logs header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(number)) if header == nil || err != nil { return logs, err } + f.begin = header.Hash() found, err := f.checkMatches(ctx, header) if err != nil { return logs, err @@ -212,11 +273,11 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err // indexedLogs returns the logs matching the filter criteria based on raw block // iteration and bloom matching. -func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, error) { +func (f *Filter) unindexedLogs(ctx context.Context, begin, end common.Hash) ([]*types.Log, error) { var logs []*types.Log - for ; f.begin <= int64(end); f.begin++ { - header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin)) + for begin != end { + header, err := f.backend.HeaderByHash(ctx, end) if header == nil || err != nil { return logs, err } @@ -225,6 +286,7 @@ func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, e return logs, err } logs = append(logs, found...) + end = header.ParentHash } return logs, nil } diff --git a/interfaces.go b/interfaces.go index 26b0fcbc17..73baba5e81 100644 --- a/interfaces.go +++ b/interfaces.go @@ -131,10 +131,12 @@ type ContractCaller interface { // FilterQuery contains options for contract log filtering. type FilterQuery struct { - BlockHash *common.Hash // used by eth_getLogs, return logs only from block with this hash - FromBlock *big.Int // beginning of the queried range, nil means genesis block - ToBlock *big.Int // end of the range, nil means latest block - Addresses []common.Address // restricts matches to events created by specific contracts + BlockHash *common.Hash // used by eth_getLogs, return logs only from block with this hash + FromBlock *big.Int // beginning of the queried range, nil means genesis block + FromBlockHash *common.Hash // beginning of the queried range, as a block hash + ToBlock *big.Int // end of the range, nil means latest block + ToBlockHash *common.Hash // end of the range, as a block hash + Addresses []common.Address // restricts matches to events created by specific contracts // The Topic list restricts matches to particular event topics. Each event has a list // of topics. Topics matches a prefix of that list. An empty element slice matches any