forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add support for fetching logs based on blockhash ranges

Browse source
This commit is contained in:
Nick Johnson 2018-09-24 11:40:41 +01:00
parent a95a601f35
commit 6ba9b3b775
3 changed files with 144 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
eth/filters/api.go
View file

@ -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{}

128
eth/filters/filter.go
View file

@ -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
}

10
interfaces.go
View file

@ -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