Problem: decode tx is unnecessary in tx listener (#491)

* Problem: decode tx is unnecessary in tx listener

* cleanup

CHANGELOG.md

@@ -51,6 +51,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 ### Improvements
 
 * (app) [#483](https://github.com/crypto-org-chain/ethermint/pull/483) Make keyring-backend client config accessible in app.
+* (rpc) [#491](https://github.com/crypto-org-chain/ethermint/pull/491) Avoid unnecessary tx decode in tx listener.
 
 ## v0.21.x-cronos
 

app/ante/handler_options.go

@@ -46,6 +46,7 @@ type HandlerOptions struct {
 	TxFeeChecker           ante.TxFeeChecker
 	DisabledAuthzMsgs      []string
 	ExtraDecorators        []sdk.AnteDecorator
+	PendingTxListener      PendingTxListener
 }
 
 func (options HandlerOptions) validate() error {
@@ -88,6 +89,7 @@ func newEthAnteHandler(ctx sdk.Context, options HandlerOptions, extra ...sdk.Ant
 		NewEthEmitEventDecorator(options.EvmKeeper), // emit eth tx hash and index at the very last ante handler.
 	}
 	decorators = append(decorators, extra...)
+	decorators = append(decorators, newTxListenerDecorator(options.PendingTxListener))
 	return sdk.ChainAnteDecorators(decorators...)
 }
 

app/ante/tx_listener.go

@@ -0,0 +1,33 @@
+package ante
+
+import (
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/ethereum/go-ethereum/common"
+	evmtypes "github.com/evmos/ethermint/x/evm/types"
+)
+
+type PendingTxListener func(common.Hash)
+
+type TxListenerDecorator struct {
+	pendingTxListener PendingTxListener
+}
+
+// newTxListenerDecorator creates a new TxListenerDecorator with the provided PendingTxListener.
+// CONTRACT: must be put at the last of the chained decorators
+func newTxListenerDecorator(pendingTxListener PendingTxListener) TxListenerDecorator {
+	return TxListenerDecorator{pendingTxListener}
+}
+
+func (d TxListenerDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate bool, next sdk.AnteHandler) (sdk.Context, error) {
+	if ctx.IsReCheckTx() {
+		return next(ctx, tx, simulate)
+	}
+	if ctx.IsCheckTx() && !simulate && d.pendingTxListener != nil {
+		for _, msg := range tx.GetMsgs() {
+			if ethTx, ok := msg.(*evmtypes.MsgEthereumTx); ok {
+				d.pendingTxListener(common.HexToHash(ethTx.Hash))
+			}
+		}
+	}
+	return next(ctx, tx, simulate)
+}

app/app.go

@@ -141,6 +141,7 @@ import (
 
 	consensusparamkeeper "github.com/cosmos/cosmos-sdk/x/consensus/keeper"
 	consensusparamtypes "github.com/cosmos/cosmos-sdk/x/consensus/types"
+	"github.com/ethereum/go-ethereum/common"
 
 	// Force-load the tracer engines to trigger registration due to Go-Ethereum v1.10.15 changes
 	_ "github.com/ethereum/go-ethereum/eth/tracers/js"
@@ -215,8 +216,6 @@ var (
 	_ servertypes.Application = (*EthermintApp)(nil)
 )
 
-type PendingTxListener func([]byte)
-
 // var _ server.Application (*EthermintApp)(nil)
 
 // EthermintApp implements an extended ABCI application. It is an application
@@ -233,7 +232,7 @@ type EthermintApp struct {
 
 	invCheckPeriod uint
 
-	pendingTxListeners []PendingTxListener
+	pendingTxListeners []ante.PendingTxListener
 
 	// keys to access the substores
 	keys    map[string]*storetypes.KVStoreKey
@@ -771,6 +770,7 @@ func (app *EthermintApp) setAnteHandler(txConfig client.TxConfig, maxGasWanted u
 			sdk.MsgTypeURL(&evmtypes.MsgEthereumTx{}),
 			sdk.MsgTypeURL(&vestingtypes.MsgCreateVestingAccount{}),
 		},
+		PendingTxListener: app.onPendingTx,
 	})
 	if err != nil {
 		panic(err)
@@ -951,19 +951,15 @@ func (app *EthermintApp) RegisterNodeService(clientCtx client.Context) {
 	node.RegisterNodeService(clientCtx, app.GRPCQueryRouter())
 }
 
-// RegisterPendingTxListener is used by json-rpc server to listen to pending transactions in CheckTx.
-func (app *EthermintApp) RegisterPendingTxListener(listener PendingTxListener) {
+// RegisterPendingTxListener is used by json-rpc server to listen to pending transactions callback.
+func (app *EthermintApp) RegisterPendingTxListener(listener ante.PendingTxListener) {
 	app.pendingTxListeners = append(app.pendingTxListeners, listener)
 }
 
-func (app *EthermintApp) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
-	res := app.BaseApp.CheckTx(req)
-	if res.Code == 0 && req.Type == abci.CheckTxType_New {
-		for _, listener := range app.pendingTxListeners {
-			listener(req.Tx)
-		}
+func (app *EthermintApp) onPendingTx(hash common.Hash) {
+	for _, listener := range app.pendingTxListeners {
+		listener(hash)
 	}
-	return res
 }
 
 // RegisterSwaggerAPI registers swagger route with API Server

rpc/stream/rpc.go

@@ -115,20 +115,8 @@ func (s *RPCStream) LogStream() *Stream[*ethtypes.Log] {
 }
 
 // ListenPendingTx is a callback passed to application to listen for pending transactions in CheckTx.
-func (s *RPCStream) ListenPendingTx(bytes []byte) {
-	tx, err := s.txDecoder(bytes)
-	if err != nil {
-		s.logger.Error("fail to decode tx", "error", err.Error())
-		return
-	}
-
-	var hashes []common.Hash
-	for _, msg := range tx.GetMsgs() {
-		if ethTx, ok := msg.(*evmtypes.MsgEthereumTx); ok {
-			hashes = append(hashes, ethTx.AsTransaction().Hash())
-		}
-	}
-	s.pendingTxStream.Add(hashes...)
+func (s *RPCStream) ListenPendingTx(hash common.Hash) {
+	s.pendingTxStream.Add(hash)
 }
 
 func (s *RPCStream) start(

server/json_rpc.go

@@ -29,18 +29,17 @@ import (
 	"github.com/cosmos/cosmos-sdk/server/types"
 	ethlog "github.com/ethereum/go-ethereum/log"
 	ethrpc "github.com/ethereum/go-ethereum/rpc"
-	"github.com/evmos/ethermint/app"
+	"github.com/evmos/ethermint/app/ante"
 	"github.com/evmos/ethermint/rpc"
 	"github.com/evmos/ethermint/rpc/stream"
-
 	"github.com/evmos/ethermint/server/config"
 	ethermint "github.com/evmos/ethermint/types"
 )
 
 const MaxRetry = 6
 
 type AppWithPendingTxStream interface {
-	RegisterPendingTxListener(listener app.PendingTxListener)
+	RegisterPendingTxListener(listener ante.PendingTxListener)
 }
 
 // StartJSONRPC starts the JSON-RPC server