fix: decode filtered logs in ws (backport: evmos#1781) (#251)

* fix decode logs in ws

* add test

* add change doc

* decode in single loop

* add ci timeout

* reset max-tx-gas-wanted

* fix rest filter api

* test batch tx with exist sc

* more filter tests

Author: mmsqe

.github/workflows/test.yml

@@ -105,6 +105,7 @@ jobs:
 
   integration_tests:
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     steps:
       - uses: actions/checkout@v2
       - uses: cachix/install-nix-action@v19

CHANGELOG.md

@@ -44,6 +44,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 * (rpc) [#1722](https://github.com/evmos/ethermint/pull/1722) Align revert response for `eth_estimateGas` and `eth_call` as Ethereum.
 * (rpc) [#1685](https://github.com/evmos/ethermint/pull/1685) Fix parse for websocket connID.
 * (rpc) [#1773](https://github.com/evmos/ethermint/pull/1773) Avoid channel get changed when concurrent subscribe happens.
+- (rpc) [#1781](https://github.com/evmos/ethermint/pull/1781) Fix decode log for multi topics in websocket subscribe.
 
 ### Features
 

rpc/namespaces/ethereum/eth/filters/api.go

@@ -399,15 +399,12 @@ func (api *PublicFilterAPI) Logs(ctx context.Context, crit filters.FilterCriteri
 					api.logger.Debug("event data type mismatch", "type", fmt.Sprintf("%T", ev.Data))
 					continue
 				}
-
-				txResponse, err := evmtypes.DecodeTxResponse(dataTx.TxResult.Result.Data)
+				txLogs, err := evmtypes.DecodeTxLogsFromEvents(dataTx.TxResult.Result.Data)
 				if err != nil {
-					api.logger.Error("fail to decode tx response", "error", err)
+					api.logger.Error("fail to decode tx response", "error", err.Error())
 					return
 				}
-
-				logs := FilterLogs(evmtypes.LogsToEthereum(txResponse.Logs), crit.FromBlock, crit.ToBlock, crit.Addresses, crit.Topics)
-
+				logs := FilterLogs(txLogs, crit.FromBlock, crit.ToBlock, crit.Addresses, crit.Topics)
 				for _, log := range logs {
 					_ = notifier.Notify(rpcSub.ID, log)
 				}
@@ -482,14 +479,12 @@ func (api *PublicFilterAPI) NewFilter(criteria filters.FilterCriteria) (rpc.ID,
 					api.logger.Debug("event data type mismatch", "type", fmt.Sprintf("%T", ev.Data))
 					continue
 				}
-
-				txResponse, err := evmtypes.DecodeTxResponse(dataTx.TxResult.Result.Data)
+				txLogs, err := evmtypes.DecodeTxLogsFromEvents(dataTx.TxResult.Result.Data)
 				if err != nil {
-					api.logger.Error("fail to decode tx response", "error", err)
+					api.logger.Error("fail to decode tx response", "error", err.Error())
 					return
 				}
-
-				logs := FilterLogs(evmtypes.LogsToEthereum(txResponse.Logs), criteria.FromBlock, criteria.ToBlock, criteria.Addresses, criteria.Topics)
+				logs := FilterLogs(txLogs, criteria.FromBlock, criteria.ToBlock, criteria.Addresses, criteria.Topics)
 
 				api.filtersMu.Lock()
 				if f, found := api.filters[filterID]; found {

rpc/websockets.go

@@ -574,14 +574,12 @@ func (api *pubSubAPI) subscribeLogs(wsConn *wsConn, subID rpc.ID, extra interfac
 					api.logger.Debug("event data type mismatch", "type", fmt.Sprintf("%T", event.Data))
 					continue
 				}
-
-				txResponse, err := evmtypes.DecodeTxResponse(dataTx.TxResult.Result.Data)
+				txLogs, err := evmtypes.DecodeTxLogsFromEvents(dataTx.TxResult.Result.Data)
 				if err != nil {
 					api.logger.Error("failed to decode tx response", "error", err.Error())
 					return
 				}
-
-				logs := rpcfilters.FilterLogs(evmtypes.LogsToEthereum(txResponse.Logs), crit.FromBlock, crit.ToBlock, crit.Addresses, crit.Topics)
+				logs := rpcfilters.FilterLogs(txLogs, crit.FromBlock, crit.ToBlock, crit.Addresses, crit.Topics)
 				if len(logs) == 0 {
 					continue
 				}

tests/integration_tests/network.py

@@ -9,7 +9,7 @@
 from web3.middleware import geth_poa_middleware
 
 from .cosmoscli import CosmosCLI
-from .utils import wait_for_port
+from .utils import supervisorctl, wait_for_port
 
 DEFAULT_CHAIN_BINARY = "ethermintd"
 
@@ -62,6 +62,9 @@ def cosmos_cli(self, i=0):
             self.base_dir / f"node{i}", self.node_rpc(i), self.chain_binary
         )
 
+    def supervisorctl(self, *args):
+        return supervisorctl(self.base_dir / "../tasks.ini", *args)
+
 
 class Geth:
     def __init__(self, w3):