eth_getFilterLogs, eth_getLogs implementation (#248)

CHANGELOG.md

@@ -58,3 +58,9 @@ Ref: https://keepachangelog.com/en/1.0.0/
 	* Update uninstallFilter and getFilterChanges accordingly
 	* uninstallFilter stops the polling goroutine
 	* getFilterChanges returns the filter's internal list of block hashes and resets it
+
+* (rpc) [\#54](https://github.com/ChainSafe/ethermint/issues/54) [\#55](https://github.com/ChainSafe/ethermint/issues/55) 
+  Implement eth_getFilterLogs and eth_getLogs
+  * for a given filter, look through each block for transactions. If there are transactions in the block, get the logs from it, and filter using the filterLogs method
+  * eth_getLogs and eth_getFilterChanges for log filters use the same underlying method as eth_getFilterLogs
+  * update HandleMsgEthereumTx to store logs using the ethereum hash

rpc/backend.go

@@ -20,6 +20,7 @@ type Backend interface {
 	// Used by block filter; also used for polling
 	BlockNumber() (hexutil.Uint64, error)
 	GetBlockByNumber(blockNum BlockNumber, fullTx bool) (map[string]interface{}, error)
+	GetBlockByHash(hash common.Hash, fullTx bool) (map[string]interface{}, error)
 	getEthBlockByNumber(height int64, fullTx bool) (map[string]interface{}, error)
 	getGasLimit() (int64, error)
 
@@ -62,6 +63,21 @@ func (e *EthermintBackend) GetBlockByNumber(blockNum BlockNumber, fullTx bool) (
 	return e.getEthBlockByNumber(value, fullTx)
 }
 
+// GetBlockByHash returns the block identified by hash.
+func (e *EthermintBackend) GetBlockByHash(hash common.Hash, fullTx bool) (map[string]interface{}, error) {
+	res, _, err := e.cliCtx.Query(fmt.Sprintf("custom/%s/%s/%s", types.ModuleName, evm.QueryHashToHeight, hash.Hex()))
+	if err != nil {
+		return nil, err
+	}
+
+	var out types.QueryResBlockNumber
+	if err := e.cliCtx.Codec.UnmarshalJSON(res, &out); err != nil {
+		return nil, err
+	}
+
+	return e.getEthBlockByNumber(out.Number, fullTx)
+}
+
 func (e *EthermintBackend) getEthBlockByNumber(height int64, fullTx bool) (map[string]interface{}, error) {
 	// Remove this check when 0 query is fixed ref: (https://github.com/tendermint/tendermint/issues/4014)
 	var blkNumPtr *int64
@@ -82,7 +98,7 @@ func (e *EthermintBackend) getEthBlockByNumber(height int64, fullTx bool) (map[s
 
 	var (
 		gasUsed      *big.Int
-		transactions []interface{}
+		transactions []common.Hash
 	)
 
 	if fullTx {
@@ -96,7 +112,7 @@ func (e *EthermintBackend) getEthBlockByNumber(height int64, fullTx bool) (map[s
 	} else {
 		// TODO: Gas used not saved and cannot be calculated by hashes
 		// Return slice of transaction hashes
-		transactions = make([]interface{}, len(block.Block.Txs))
+		transactions = make([]common.Hash, len(block.Block.Txs))
 		for i, tx := range block.Block.Txs {
 			transactions[i] = common.BytesToHash(tx.Hash())
 		}
@@ -142,13 +158,17 @@ func (e *EthermintBackend) getGasLimit() (int64, error) {
 func (e *EthermintBackend) GetTxLogs(txHash common.Hash) ([]*ethtypes.Log, error) {
 	// do we need to use the block height somewhere?
 	ctx := e.cliCtx
+
 	res, _, err := ctx.QueryWithData(fmt.Sprintf("custom/%s/%s/%s", types.ModuleName, evm.QueryTxLogs, txHash.Hex()), nil)
 	if err != nil {
 		return nil, err
 	}
 
-	var out types.QueryETHLogs
-	e.cliCtx.Codec.MustUnmarshalJSON(res, &out)
+	out := new(types.QueryETHLogs)
+	if err := e.cliCtx.Codec.UnmarshalJSON(res, &out); err != nil {
+		return nil, err
+	}
+
 	return out.Logs, nil
 }
 

rpc/eth_api.go

@@ -293,8 +293,6 @@ func (e *PublicEthAPI) SendTransaction(args params.SendTxArgs) (common.Hash, err
 	// Broadcast transaction
 	res, err := e.cliCtx.BroadcastTx(txBytes)
 	// If error is encountered on the node, the broadcast will not return an error
-	// TODO: Remove res log
-	fmt.Println(res.RawLog)
 	if err != nil {
 		return common.Hash{}, err
 	}
@@ -471,14 +469,7 @@ func (e *PublicEthAPI) EstimateGas(args CallArgs) (hexutil.Uint64, error) {
 
 // GetBlockByHash returns the block identified by hash.
 func (e *PublicEthAPI) GetBlockByHash(hash common.Hash, fullTx bool) (map[string]interface{}, error) {
-	res, _, err := e.cliCtx.Query(fmt.Sprintf("custom/%s/%s/%s", types.ModuleName, evm.QueryHashToHeight, hash.Hex()))
-	if err != nil {
-		return nil, err
-	}
-
-	var out types.QueryResBlockNumber
-	e.cliCtx.Codec.MustUnmarshalJSON(res, &out)
-	return e.backend.getEthBlockByNumber(out.Number, fullTx)
+	return e.backend.GetBlockByHash(hash, fullTx)
 }
 
 // GetBlockByNumber returns the block identified by number.
@@ -488,7 +479,7 @@ func (e *PublicEthAPI) GetBlockByNumber(blockNum BlockNumber, fullTx bool) (map[
 
 func formatBlock(
 	header tmtypes.Header, size int, gasLimit int64,
-	gasUsed *big.Int, transactions []interface{}, bloom ethtypes.Bloom,
+	gasUsed *big.Int, transactions interface{}, bloom ethtypes.Bloom,
 ) map[string]interface{} {
 	return map[string]interface{}{
 		"number":           hexutil.Uint64(header.Height),
@@ -507,13 +498,13 @@ func formatBlock(
 		"gasLimit":         hexutil.Uint64(gasLimit), // Static gas limit
 		"gasUsed":          (*hexutil.Big)(gasUsed),
 		"timestamp":        hexutil.Uint64(header.Time.Unix()),
-		"transactions":     transactions,
+		"transactions":     transactions.([]common.Hash),
 		"uncles":           nil,
 	}
 }
 
-func convertTransactionsToRPC(cliCtx context.CLIContext, txs []tmtypes.Tx, blockHash common.Hash, height uint64) ([]interface{}, *big.Int, error) {
-	transactions := make([]interface{}, len(txs))
+func convertTransactionsToRPC(cliCtx context.CLIContext, txs []tmtypes.Tx, blockHash common.Hash, height uint64) ([]common.Hash, *big.Int, error) {
+	transactions := make([]common.Hash, len(txs))
 	gasUsed := big.NewInt(0)
 
 	for i, tx := range txs {
@@ -523,10 +514,11 @@ func convertTransactionsToRPC(cliCtx context.CLIContext, txs []tmtypes.Tx, block
 		}
 		// TODO: Remove gas usage calculation if saving gasUsed per block
 		gasUsed.Add(gasUsed, ethTx.Fee())
-		transactions[i], err = newRPCTransaction(*ethTx, blockHash, &height, uint64(i))
+		tx, err := newRPCTransaction(*ethTx, blockHash, &height, uint64(i))
 		if err != nil {
 			return nil, nil, err
 		}
+		transactions[i] = tx.Hash
 	}
 
 	return transactions, gasUsed, nil

rpc/filter_api.go

@@ -1,12 +1,9 @@
 package rpc
 
 import (
-	"encoding/json"
-	"fmt"
-	"math/big"
+	"errors"
 
 	"github.com/cosmos/cosmos-sdk/client/context"
-	"github.com/cosmos/ethermint/x/evm/types"
 
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/filters"
@@ -62,62 +59,21 @@ func (e *PublicFilterAPI) UninstallFilter(id rpc.ID) bool {
 // If the filter is a block filter, it returns an array of block hashes.
 // If the filter is a pending transaction filter, it returns an array of transaction hashes.
 func (e *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
+	if e.filters[id] == nil {
+		return nil, errors.New("invalid filter ID")
+	}
 	return e.filters[id].getFilterChanges()
 }
 
 // GetFilterLogs returns an array of all logs matching filter with given id.
-func (e *PublicFilterAPI) GetFilterLogs(id rpc.ID) []*ethtypes.Log {
+func (e *PublicFilterAPI) GetFilterLogs(id rpc.ID) ([]*ethtypes.Log, error) {
 	return e.filters[id].getFilterLogs()
 }
 
 // GetLogs returns logs matching the given argument that are stored within the state.
 //
 // https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs
 func (e *PublicFilterAPI) GetLogs(criteria filters.FilterCriteria) ([]*ethtypes.Log, error) {
-	var filter *Filter
-	if criteria.BlockHash != nil {
-		/*
-			Still need to add blockhash in prepare function for log entry
-		*/
-		filter = NewFilterWithBlockHash(e.backend, &criteria)
-		results := e.getLogs()
-		logs := filterLogs(results, nil, nil, filter.addresses, filter.topics)
-		return logs, nil
-	}
-	// Convert the RPC block numbers into internal representations
-	begin := rpc.LatestBlockNumber.Int64()
-	if criteria.FromBlock != nil {
-		begin = criteria.FromBlock.Int64()
-	}
-	from := big.NewInt(begin)
-	end := rpc.LatestBlockNumber.Int64()
-	if criteria.ToBlock != nil {
-		end = criteria.ToBlock.Int64()
-	}
-	to := big.NewInt(end)
-	results := e.getLogs()
-	logs := filterLogs(results, from, to, criteria.Addresses, criteria.Topics)
-
-	return returnLogs(logs), nil
-}
-
-func (e *PublicFilterAPI) getLogs() (results []*ethtypes.Log) {
-	l, _, err := e.cliCtx.QueryWithData(fmt.Sprintf("custom/%s/logs", types.ModuleName), nil)
-	if err != nil {
-		fmt.Printf("error from querier %e ", err)
-	}
-
-	if err := json.Unmarshal(l, &results); err != nil {
-		panic(err)
-	}
-	return results
-}
-
-// 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 []*ethtypes.Log) []*ethtypes.Log {
-	if logs == nil {
-		return []*ethtypes.Log{}
-	}
-	return logs
+	filter := NewFilter(e.backend, &criteria)
+	return filter.getFilterLogs()
 }

rpc/filters.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/filters"
+	"github.com/ethereum/go-ethereum/log"
 )
 
 /*
@@ -37,7 +38,7 @@ type Filter struct {
 
 // NewFilter returns a new Filter
 func NewFilter(backend Backend, criteria *filters.FilterCriteria) *Filter {
-	return &Filter{
+	filter := &Filter{
 		backend:   backend,
 		fromBlock: criteria.FromBlock,
 		toBlock:   criteria.ToBlock,
@@ -46,6 +47,8 @@ func NewFilter(backend Backend, criteria *filters.FilterCriteria) *Filter {
 		typ:       logFilter,
 		stopped:   false,
 	}
+
+	return filter
 }
 
 // NewFilterWithBlockHash returns a new Filter with a blockHash.
@@ -139,28 +142,107 @@ func (f *Filter) getFilterChanges() (interface{}, error) {
 	case pendingTxFilter:
 		// TODO
 	case logFilter:
-		// TODO
+		return f.getFilterLogs()
 	}
 
-	return nil, nil
+	return nil, errors.New("unsupported filter")
 }
 
-func (f *Filter) getFilterLogs() []*ethtypes.Log {
-	// TODO
-	return nil
+func (f *Filter) getFilterLogs() ([]*ethtypes.Log, error) {
+	ret := []*ethtypes.Log{}
+
+	// filter specific block only
+	if f.blockHash != nil {
+		block, err := f.backend.GetBlockByHash(*f.blockHash, true)
+		if err != nil {
+			return nil, err
+		}
+
+		// if the logsBloom == 0, there are no logs in that block
+		if txs, ok := block["transactions"].([]common.Hash); !ok {
+			return ret, nil
+		} else if len(txs) != 0 {
+			return f.checkMatches(block)
+		}
+	}
+
+	// filter range of blocks
+	num, err := f.backend.BlockNumber()
+	if err != nil {
+		return nil, err
+	}
+
+	// if f.fromBlock is set to 0, set it to the latest block number
+	if f.fromBlock == nil || f.fromBlock.Cmp(big.NewInt(0)) == 0 {
+		f.fromBlock = big.NewInt(int64(num))
+	}
+
+	// if f.toBlock is set to 0, set it to the latest block number
+	if f.toBlock == nil || f.toBlock.Cmp(big.NewInt(0)) == 0 {
+		f.toBlock = big.NewInt(int64(num))
+	}
+
+	log.Debug("[ethAPI] Retrieving filter logs", "fromBlock", f.fromBlock, "toBlock", f.toBlock,
+		"topics", f.topics, "addresses", f.addresses)
+
+	from := f.fromBlock.Int64()
+	to := f.toBlock.Int64()
+
+	for i := from; i <= to; i++ {
+		block, err := f.backend.GetBlockByNumber(NewBlockNumber(big.NewInt(i)), true)
+		if err != nil {
+			f.err = err
+			log.Debug("[ethAPI] Cannot get block", "block", block["number"], "error", err)
+			break
+		}
+
+		log.Debug("[ethAPI] filtering", "block", block)
+
+		// TODO: block logsBloom is often set in the wrong block
+		// if the logsBloom == 0, there are no logs in that block
+
+		if txs, ok := block["transactions"].([]common.Hash); !ok {
+			continue
+		} else if len(txs) != 0 {
+			logs, err := f.checkMatches(block)
+			if err != nil {
+				f.err = err
+				break
+			}
+
+			ret = append(ret, logs...)
+		}
+	}
+
+	return ret, nil
 }
 
-func includes(addresses []common.Address, a common.Address) bool {
-	for _, addr := range addresses {
-		if addr == a {
-			return true
+func (f *Filter) checkMatches(block map[string]interface{}) ([]*ethtypes.Log, error) {
+	transactions, ok := block["transactions"].([]common.Hash)
+	if !ok {
+		return nil, errors.New("invalid block transactions")
+	}
+
+	unfiltered := []*ethtypes.Log{}
+
+	for _, tx := range transactions {
+		logs, err := f.backend.GetTxLogs(common.BytesToHash(tx[:]))
+		if err != nil {
+			return nil, err
 		}
+
+		unfiltered = append(unfiltered, logs...)
 	}
 
-	return false
+	return filterLogs(unfiltered, f.fromBlock, f.toBlock, f.addresses, f.topics), nil
 }
 
 // filterLogs creates a slice of logs matching the given criteria.
+// [] -> anything
+// [A] -> A in first position of log topics, anything after
+// [null, B] -> anything in first position, B in second position
+// [A, B] -> A in first position and B in second position
+// [[A, B], [A, B]] -> A or B in first position, A or B in second position
 func filterLogs(logs []*ethtypes.Log, fromBlock, toBlock *big.Int, addresses []common.Address, topics [][]common.Hash) []*ethtypes.Log {
 	var ret []*ethtypes.Log
 Logs:
@@ -194,3 +276,13 @@ Logs:
 	}
 	return ret
 }
+
+func includes(addresses []common.Address, a common.Address) bool {
+	for _, addr := range addresses {
+		if addr == a {
+			return true
+		}
+	}
+
+	return false
+}