impr: support batch eth txs (#901)

* support batch eth tx

Closes: 896

Allow multiple MsgEthereumTx in single tx

* fix transaction receipt api

* fix tx receipt api and accumulate tx gas used

* fix lint

* fix test

* fix rpc test

* cleanup

* fix cumulativeGasUsed and gasUsed

* fix lint

* Update app/ante/eth.go

Co-authored-by: Federico Kunze Küllmer <31522760+fedekunze@users.noreply.github.com>

* Update app/ante/eth.go

Co-authored-by: Federico Kunze Küllmer <31522760+fedekunze@users.noreply.github.com>

* Update rpc/ethereum/backend/utils.go

Co-authored-by: Federico Kunze Küllmer <31522760+fedekunze@users.noreply.github.com>

* pr suggestions

* typo

* fix lint

Co-authored-by: Federico Kunze Küllmer <31522760+fedekunze@users.noreply.github.com>

rpc/ethereum/backend/backend.go

@@ -66,7 +66,6 @@ type Backend interface {
 	HeaderByNumber(blockNum types.BlockNumber) (*ethtypes.Header, error)
 	HeaderByHash(blockHash common.Hash) (*ethtypes.Header, error)
 	PendingTransactions() ([]*sdk.Tx, error)
-	GetTransactionLogs(txHash common.Hash) ([]*ethtypes.Log, error)
 	GetTransactionCount(address common.Address, blockNum types.BlockNumber) (*hexutil.Uint64, error)
 	SendTransaction(args evmtypes.TransactionArgs) (common.Hash, error)
 	GetCoinbase() (sdk.AccAddress, error)
@@ -536,18 +535,6 @@ func (e *EVMBackend) HeaderByHash(blockHash common.Hash) (*ethtypes.Header, erro
 	return ethHeader, nil
 }
 
-// GetTransactionLogs returns the logs given a transaction hash.
-// It returns an error if there's an encoding error.
-// If no logs are found for the tx hash, the error is nil.
-func (e *EVMBackend) GetTransactionLogs(txHash common.Hash) ([]*ethtypes.Log, error) {
-	tx, err := e.GetTxByEthHash(txHash)
-	if err != nil {
-		return nil, err
-	}
-
-	return TxLogsFromEvents(tx.TxResult.Events)
-}
-
 // PendingTransactions returns the transactions that are in the transaction pool
 // and have a from address that is one of the accounts this node manages.
 func (e *EVMBackend) PendingTransactions() ([]*sdk.Tx, error) {
@@ -578,12 +565,12 @@ func (e *EVMBackend) GetLogsByHeight(height *int64) ([][]*ethtypes.Log, error) {
 
 	blockLogs := [][]*ethtypes.Log{}
 	for _, txResult := range blockRes.TxsResults {
-		logs, err := TxLogsFromEvents(txResult.Events)
+		logs, err := AllTxLogsFromEvents(txResult.Events)
 		if err != nil {
 			return nil, err
 		}
 
-		blockLogs = append(blockLogs, logs)
+		blockLogs = append(blockLogs, logs...)
 	}
 
 	return blockLogs, nil
@@ -667,7 +654,7 @@ func (e *EVMBackend) GetTransactionByHash(txHash common.Hash) (*types.RPCTransac
 		}
 
 		for _, tx := range txs {
-			msg, err := evmtypes.UnwrapEthereumMsg(tx)
+			msg, err := evmtypes.UnwrapEthereumMsg(tx, txHash)
 			if err != nil {
 				// not ethereum tx
 				continue
@@ -696,16 +683,18 @@ func (e *EVMBackend) GetTransactionByHash(txHash common.Hash) (*types.RPCTransac
 		return nil, errors.New("invalid ethereum tx")
 	}
 
+	msgIndex, attrs := types.FindTxAttributes(res.TxResult.Events, hexTx)
+	if msgIndex < 0 {
+		return nil, fmt.Errorf("ethereum tx not found in msgs: %s", hexTx)
+	}
+
 	tx, err := e.clientCtx.TxConfig.TxDecoder()(res.Tx)
 	if err != nil {
 		return nil, err
 	}
 
-	if len(tx.GetMsgs()) != 1 {
-		return nil, errors.New("invalid ethereum tx")
-	}
-
-	msg, ok := tx.GetMsgs()[0].(*evmtypes.MsgEthereumTx)
+	// the `msgIndex` is inferred from tx events, should be within the bound.
+	msg, ok := tx.GetMsgs()[msgIndex].(*evmtypes.MsgEthereumTx)
 	if !ok {
 		return nil, errors.New("invalid ethereum tx")
 	}
@@ -718,7 +707,7 @@ func (e *EVMBackend) GetTransactionByHash(txHash common.Hash) (*types.RPCTransac
 
 	// Try to find txIndex from events
 	found := false
-	txIndex, err := types.TxIndexFromEvents(res.TxResult.Events)
+	txIndex, err := types.GetUint64Attribute(attrs, evmtypes.AttributeKeyTxIndex)
 	if err == nil {
 		found = true
 	} else {
@@ -1033,7 +1022,6 @@ func (e *EVMBackend) BaseFee(height int64) (*big.Int, error) {
 // GetEthereumMsgsFromTendermintBlock returns all real MsgEthereumTxs from a Tendermint block.
 // It also ensures consistency over the correct txs indexes across RPC endpoints
 func (e *EVMBackend) GetEthereumMsgsFromTendermintBlock(block *tmrpctypes.ResultBlock, blockRes *tmrpctypes.ResultBlockResults) []*evmtypes.MsgEthereumTx {
-	// nolint: prealloc
 	var result []*evmtypes.MsgEthereumTx
 
 	txResults := blockRes.TxsResults
@@ -1050,16 +1038,15 @@ func (e *EVMBackend) GetEthereumMsgsFromTendermintBlock(block *tmrpctypes.Result
 			e.logger.Debug("failed to decode transaction in block", "height", block.Block.Height, "error", err.Error())
 			continue
 		}
-		if len(tx.GetMsgs()) != 1 {
-			continue
-		}
 
-		ethMsg, ok := tx.GetMsgs()[0].(*evmtypes.MsgEthereumTx)
-		if !ok {
-			continue
-		}
+		for _, msg := range tx.GetMsgs() {
+			ethMsg, ok := msg.(*evmtypes.MsgEthereumTx)
+			if !ok {
+				continue
+			}
 
-		result = append(result, ethMsg)
+			result = append(result, ethMsg)
+		}
 	}
 
 	return result

rpc/ethereum/backend/utils.go

@@ -182,44 +182,76 @@ func (e *EVMBackend) getAccountNonce(accAddr common.Address, pending bool, heigh
 	// add the uncommitted txs to the nonce counter
 	// only supports `MsgEthereumTx` style tx
 	for _, tx := range pendingTxs {
-		msg, err := evmtypes.UnwrapEthereumMsg(tx)
-		if err != nil {
-			// not ethereum tx
-			continue
-		}
+		for _, msg := range (*tx).GetMsgs() {
+			ethMsg, ok := msg.(*evmtypes.MsgEthereumTx)
+			if !ok {
+				// not ethereum tx
+				break
+			}
 
-		sender, err := msg.GetSender(e.chainID)
-		if err != nil {
-			continue
-		}
-		if sender == accAddr {
-			nonce++
+			sender, err := ethMsg.GetSender(e.chainID)
+			if err != nil {
+				continue
+			}
+			if sender == accAddr {
+				nonce++
+			}
 		}
 	}
 
 	return nonce, nil
 }
 
-// TxLogsFromEvents parses ethereum logs from cosmos events
-func TxLogsFromEvents(events []abci.Event) ([]*ethtypes.Log, error) {
-	logs := make([]*evmtypes.Log, 0)
+// AllTxLogsFromEvents parses all ethereum logs from cosmos events
+func AllTxLogsFromEvents(events []abci.Event) ([][]*ethtypes.Log, error) {
+	allLogs := make([][]*ethtypes.Log, 0, 4)
 	for _, event := range events {
 		if event.Type != evmtypes.EventTypeTxLog {
 			continue
 		}
 
-		for _, attr := range event.Attributes {
-			if !bytes.Equal(attr.Key, []byte(evmtypes.AttributeKeyTxLog)) {
-				continue
-			}
-
-			var log evmtypes.Log
-			if err := json.Unmarshal(attr.Value, &log); err != nil {
-				return nil, err
-			}
-
-			logs = append(logs, &log)
+		logs, err := ParseTxLogsFromEvent(event)
+		if err != nil {
+			return nil, err
 		}
+
+		allLogs = append(allLogs, logs)
+	}
+	return allLogs, nil
+}
+
+// TxLogsFromEvents parses ethereum logs from cosmos events for specific msg index
+func TxLogsFromEvents(events []abci.Event, msgIndex int) ([]*ethtypes.Log, error) {
+	for _, event := range events {
+		if event.Type != evmtypes.EventTypeTxLog {
+			continue
+		}
+
+		if msgIndex > 0 {
+			// not the eth tx we want
+			msgIndex--
+			continue
+		}
+
+		return ParseTxLogsFromEvent(event)
+	}
+	return nil, fmt.Errorf("eth tx logs not found for message index %d", msgIndex)
+}
+
+// ParseTxLogsFromEvent parse tx logs from one event
+func ParseTxLogsFromEvent(event abci.Event) ([]*ethtypes.Log, error) {
+	logs := make([]*evmtypes.Log, 0, len(event.Attributes))
+	for _, attr := range event.Attributes {
+		if !bytes.Equal(attr.Key, []byte(evmtypes.AttributeKeyTxLog)) {
+			continue
+		}
+
+		var log evmtypes.Log
+		if err := json.Unmarshal(attr.Value, &log); err != nil {
+			return nil, err
+		}
+
+		logs = append(logs, &log)
 	}
 	return evmtypes.LogsToEthereum(logs), nil
 }

rpc/ethereum/namespaces/eth/api.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"math"
 	"math/big"
-	"strings"
 
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/params"
@@ -389,7 +388,20 @@ func (e *PublicAPI) GetCode(address common.Address, blockNrOrHash rpctypes.Block
 // GetTransactionLogs returns the logs given a transaction hash.
 func (e *PublicAPI) GetTransactionLogs(txHash common.Hash) ([]*ethtypes.Log, error) {
 	e.logger.Debug("eth_getTransactionLogs", "hash", txHash)
-	return e.backend.GetTransactionLogs(txHash)
+
+	hexTx := txHash.Hex()
+	res, err := e.backend.GetTxByEthHash(txHash)
+	if err != nil {
+		e.logger.Debug("tx not found", "hash", hexTx, "error", err.Error())
+		return nil, nil
+	}
+
+	msgIndex, _ := rpctypes.FindTxAttributes(res.TxResult.Events, hexTx)
+	if msgIndex < 0 {
+		return nil, fmt.Errorf("ethereum tx not found in msgs: %s", hexTx)
+	}
+	// parse tx logs from events
+	return backend.TxLogsFromEvents(res.TxResult.Events, msgIndex)
 }
 
 // Sign signs the provided data using the private key of address via Geth's signature standard.
@@ -556,7 +568,8 @@ func (e *PublicAPI) Resend(ctx context.Context, args evmtypes.TransactionArgs, g
 	}
 
 	for _, tx := range pending {
-		p, err := evmtypes.UnwrapEthereumMsg(tx)
+		// FIXME does Resend api possible at all?  https://github.com/tharsis/ethermint/issues/905
+		p, err := evmtypes.UnwrapEthereumMsg(tx, common.Hash{})
 		if err != nil {
 			// not valid ethereum tx
 			continue
@@ -768,6 +781,11 @@ func (e *PublicAPI) GetTransactionReceipt(hash common.Hash) (map[string]interfac
 		return nil, nil
 	}
 
+	msgIndex, attrs := rpctypes.FindTxAttributes(res.TxResult.Events, hexTx)
+	if msgIndex < 0 {
+		return nil, fmt.Errorf("ethereum tx not found in msgs: %s", hexTx)
+	}
+
 	resBlock, err := e.clientCtx.Client.Block(e.ctx, &res.Height)
 	if err != nil {
 		e.logger.Debug("block not found", "height", res.Height, "error", err.Error())
@@ -780,13 +798,15 @@ func (e *PublicAPI) GetTransactionReceipt(hash common.Hash) (map[string]interfac
 		return nil, fmt.Errorf("failed to decode tx: %w", err)
 	}
 
-	msg, err := evmtypes.UnwrapEthereumMsg(&tx)
-	if err != nil {
-		e.logger.Debug("invalid tx", "error", err.Error())
-		return nil, err
+	// the `msgIndex` is inferred from tx events, should be within the bound.
+	msg := tx.GetMsgs()[msgIndex]
+	ethMsg, ok := msg.(*evmtypes.MsgEthereumTx)
+	if !ok {
+		e.logger.Debug(fmt.Sprintf("invalid tx type: %T", msg))
+		return nil, fmt.Errorf("invalid tx type: %T", msg)
 	}
 
-	txData, err := evmtypes.UnpackTxData(msg.Data)
+	txData, err := evmtypes.UnpackTxData(ethMsg.Data)
 	if err != nil {
 		e.logger.Error("failed to unpack tx data", "error", err.Error())
 		return nil, err
@@ -799,37 +819,61 @@ func (e *PublicAPI) GetTransactionReceipt(hash common.Hash) (map[string]interfac
 		return nil, nil
 	}
 
-	for i := 0; i <= int(res.Index) && i < len(blockRes.TxsResults); i++ {
+	for i := 0; i < int(res.Index) && i < len(blockRes.TxsResults); i++ {
 		cumulativeGasUsed += uint64(blockRes.TxsResults[i].GasUsed)
 	}
+	cumulativeGasUsed += rpctypes.AccumulativeGasUsedOfMsg(res.TxResult.Events, msgIndex)
+
+	var gasUsed uint64
+	if len(tx.GetMsgs()) == 1 {
+		// backward compatibility
+		gasUsed = uint64(res.TxResult.GasUsed)
+	} else {
+		gasUsed, err = rpctypes.GetUint64Attribute(attrs, evmtypes.AttributeKeyTxGasUsed)
+		if err != nil {
+			return nil, err
+		}
+	}
 
 	// Get the transaction result from the log
+	_, found := attrs[evmtypes.AttributeKeyEthereumTxFailed]
 	var status hexutil.Uint
-	if strings.Contains(res.TxResult.GetLog(), evmtypes.AttributeKeyEthereumTxFailed) {
+	if found {
 		status = hexutil.Uint(ethtypes.ReceiptStatusFailed)
 	} else {
 		status = hexutil.Uint(ethtypes.ReceiptStatusSuccessful)
 	}
 
-	from, err := msg.GetSender(e.chainIDEpoch)
+	from, err := ethMsg.GetSender(e.chainIDEpoch)
 	if err != nil {
 		return nil, err
 	}
 
-	logs, err := e.backend.GetTransactionLogs(hash)
+	// parse tx logs from events
+	logs, err := backend.TxLogsFromEvents(res.TxResult.Events, msgIndex)
 	if err != nil {
 		e.logger.Debug("logs not found", "hash", hexTx, "error", err.Error())
 	}
 
-	// get eth index based on block's txs
-	var txIndex uint64
-	msgs := e.backend.GetEthereumMsgsFromTendermintBlock(resBlock, blockRes)
-	for i := range msgs {
-		if msgs[i].Hash == hexTx {
-			txIndex = uint64(i)
-			break
+	// Try to find txIndex from events
+	found = false
+	txIndex, err := rpctypes.GetUint64Attribute(attrs, evmtypes.AttributeKeyTxIndex)
+	if err == nil {
+		found = true
+	} else {
+		// Fallback to find tx index by iterating all valid eth transactions
+		msgs := e.backend.GetEthereumMsgsFromTendermintBlock(resBlock, blockRes)
+		for i := range msgs {
+			if msgs[i].Hash == hexTx {
+				txIndex = uint64(i)
+				found = true
+				break
+			}
 		}
 	}
+	if !found {
+		return nil, errors.New("can't find index of ethereum tx")
+	}
 
 	receipt := map[string]interface{}{
 		// Consensus fields: These fields are defined by the Yellow Paper
@@ -842,7 +886,7 @@ func (e *PublicAPI) GetTransactionReceipt(hash common.Hash) (map[string]interfac
 		// They are stored in the chain database.
 		"transactionHash": hash,
 		"contractAddress": nil,
-		"gasUsed":         hexutil.Uint64(res.TxResult.GasUsed),
+		"gasUsed":         hexutil.Uint64(gasUsed),
 		"type":            hexutil.Uint(txData.TxType()),
 
 		// Inclusion information: These fields provide information about the inclusion of the
@@ -888,24 +932,26 @@ func (e *PublicAPI) GetPendingTransactions() ([]*rpctypes.RPCTransaction, error)
 
 	result := make([]*rpctypes.RPCTransaction, 0, len(txs))
 	for _, tx := range txs {
-		msg, err := evmtypes.UnwrapEthereumMsg(tx)
-		if err != nil {
-			// not valid ethereum tx
-			continue
-		}
+		for _, msg := range (*tx).GetMsgs() {
+			ethMsg, ok := msg.(*evmtypes.MsgEthereumTx)
+			if !ok {
+				// not valid ethereum tx
+				break
+			}
 
-		rpctx, err := rpctypes.NewTransactionFromMsg(
-			msg,
-			common.Hash{},
-			uint64(0),
-			uint64(0),
-			e.chainIDEpoch,
-		)
-		if err != nil {
-			return nil, err
-		}
+			rpctx, err := rpctypes.NewTransactionFromMsg(
+				ethMsg,
+				common.Hash{},
+				uint64(0),
+				uint64(0),
+				e.chainIDEpoch,
+			)
+			if err != nil {
+				return nil, err
+			}
 
-		result = append(result, rpctx)
+			result = append(result, rpctx)
+		}
 	}
 
 	return result, nil

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

@@ -31,7 +31,6 @@ type Backend interface {
 	GetLogsByNumber(blockNum types.BlockNumber) ([][]*ethtypes.Log, error)
 	BlockBloom(height *int64) (ethtypes.Bloom, error)
 
-	GetTransactionLogs(txHash common.Hash) ([]*ethtypes.Log, error)
 	BloomStatus() (uint64, uint64)
 
 	RPCFilterCap() int32

rpc/ethereum/types/utils.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/hex"
-	"errors"
 	"fmt"
 	"math/big"
 	"strconv"
@@ -25,21 +24,21 @@ import (
 )
 
 // RawTxToEthTx returns a evm MsgEthereum transaction from raw tx bytes.
-func RawTxToEthTx(clientCtx client.Context, txBz tmtypes.Tx) (*evmtypes.MsgEthereumTx, error) {
+func RawTxToEthTx(clientCtx client.Context, txBz tmtypes.Tx) ([]*evmtypes.MsgEthereumTx, error) {
 	tx, err := clientCtx.TxConfig.TxDecoder()(txBz)
 	if err != nil {
 		return nil, sdkerrors.Wrap(sdkerrors.ErrJSONUnmarshal, err.Error())
 	}
 
-	if len(tx.GetMsgs()) != 1 {
-		return nil, errors.New("not ethereum tx")
+	ethTxs := make([]*evmtypes.MsgEthereumTx, len(tx.GetMsgs()))
+	for i, msg := range tx.GetMsgs() {
+		ethTx, ok := msg.(*evmtypes.MsgEthereumTx)
+		if !ok {
+			return nil, fmt.Errorf("invalid message type %T, expected %T", msg, &evmtypes.MsgEthereumTx{})
+		}
+		ethTxs[i] = ethTx
 	}
-
-	ethTx, ok := tx.GetMsgs()[0].(*evmtypes.MsgEthereumTx)
-	if !ok {
-		return nil, fmt.Errorf("invalid msg type %T, expected %T", tx, evmtypes.MsgEthereumTx{})
-	}
-	return ethTx, nil
+	return ethTxs, nil
 }
 
 // EthHeaderFromTendermint is an util function that returns an Ethereum Header
@@ -256,25 +255,80 @@ func BaseFeeFromEvents(events []abci.Event) *big.Int {
 	return nil
 }
 
-// TxIndexFromEvents parses the tx index from cosmos events
-func TxIndexFromEvents(events []abci.Event) (uint64, error) {
+// FindTxAttributes returns the msg index of the eth tx in cosmos tx, and the attributes,
+// returns -1 and nil if not found.
+func FindTxAttributes(events []abci.Event, txHash string) (int, map[string]string) {
+	msgIndex := -1
 	for _, event := range events {
 		if event.Type != evmtypes.EventTypeEthereumTx {
 			continue
 		}
 
-		for _, attr := range event.Attributes {
-			if bytes.Equal(attr.Key, []byte(evmtypes.AttributeKeyTxIndex)) {
-				result, err := strconv.ParseInt(string(attr.Value), 10, 64)
-				if err != nil {
-					return 0, err
-				}
-				if result < 0 {
-					return 0, errors.New("negative tx index")
-				}
-				return uint64(result), nil
-			}
+		msgIndex++
+
+		value := FindAttribute(event.Attributes, []byte(evmtypes.AttributeKeyEthereumTxHash))
+		if !bytes.Equal(value, []byte(txHash)) {
+			continue
 		}
+
+		// found, convert attributes to map for later lookup
+		attrs := make(map[string]string, len(event.Attributes))
+		for _, attr := range event.Attributes {
+			attrs[string(attr.Key)] = string(attr.Value)
+		}
+		return msgIndex, attrs
 	}
-	return 0, errors.New("not found")
+	// not found
+	return -1, nil
+}
+
+// FindAttribute find event attribute with specified key, if not found returns nil.
+func FindAttribute(attrs []abci.EventAttribute, key []byte) []byte {
+	for _, attr := range attrs {
+		if !bytes.Equal(attr.Key, key) {
+			continue
+		}
+		return attr.Value
+	}
+	return nil
+}
+
+// GetUint64Attribute parses the uint64 value from event attributes
+func GetUint64Attribute(attrs map[string]string, key string) (uint64, error) {
+	value, found := attrs[key]
+	if !found {
+		return 0, fmt.Errorf("tx index attribute not found: %s", key)
+	}
+	var result int64
+	result, err := strconv.ParseInt(value, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	if result < 0 {
+		return 0, fmt.Errorf("negative tx index: %d", result)
+	}
+	return uint64(result), nil
+}
+
+// AccumulativeGasUsedOfMsg accumulate the gas used by msgs before `msgIndex`.
+func AccumulativeGasUsedOfMsg(events []abci.Event, msgIndex int) (gasUsed uint64) {
+	for _, event := range events {
+		if event.Type != evmtypes.EventTypeEthereumTx {
+			continue
+		}
+
+		if msgIndex < 0 {
+			break
+		}
+		msgIndex--
+
+		value := FindAttribute(event.Attributes, []byte(evmtypes.AttributeKeyTxGasUsed))
+		var result int64
+		result, err := strconv.ParseInt(string(value), 10, 64)
+		if err != nil {
+			continue
+		}
+		gasUsed += uint64(result)
+	}
+	return
 }

rpc/websockets.go

@@ -683,44 +683,46 @@ func (api *pubSubAPI) subscribePendingTransactions(wsConn *wsConn) (rpc.ID, erro
 			select {
 			case ev := <-txsCh:
 				data, _ := ev.Data.(tmtypes.EventDataTx)
-				ethTx, err := types.RawTxToEthTx(api.clientCtx, data.Tx)
+				ethTxs, err := types.RawTxToEthTx(api.clientCtx, data.Tx)
 				if err != nil {
 					// not ethereum tx
-					panic("debug")
+					continue
 				}
 
 				api.filtersMu.RLock()
-				for subID, wsSub := range api.filters {
-					subID := subID
-					wsSub := wsSub
-					if wsSub.query != query {
-						continue
-					}
-					// write to ws conn
-					res := &SubscriptionNotification{
-						Jsonrpc: "2.0",
-						Method:  "eth_subscription",
-						Params: &SubscriptionResult{
-							Subscription: subID,
-							Result:       ethTx.Hash,
-						},
-					}
+				for _, ethTx := range ethTxs {
+					for subID, wsSub := range api.filters {
+						subID := subID
+						wsSub := wsSub
+						if wsSub.query != query {
+							continue
+						}
+						// write to ws conn
+						res := &SubscriptionNotification{
+							Jsonrpc: "2.0",
+							Method:  "eth_subscription",
+							Params: &SubscriptionResult{
+								Subscription: subID,
+								Result:       ethTx.Hash,
+							},
+						}
 
-					err = wsSub.wsConn.WriteJSON(res)
-					if err != nil {
-						api.logger.Debug("error writing header, will drop peer", "error", err.Error())
+						err = wsSub.wsConn.WriteJSON(res)
+						if err != nil {
+							api.logger.Debug("error writing header, will drop peer", "error", err.Error())
 
-						try(func() {
-							api.filtersMu.Lock()
-							defer api.filtersMu.Unlock()
+							try(func() {
+								api.filtersMu.Lock()
+								defer api.filtersMu.Unlock()
 
-							if err != websocket.ErrCloseSent {
-								_ = wsSub.wsConn.Close()
-							}
+								if err != websocket.ErrCloseSent {
+									_ = wsSub.wsConn.Close()
+								}
 
-							delete(api.filters, subID)
-							close(wsSub.unsubscribed)
-						}, api.logger, "closing websocket peer sub")
+								delete(api.filters, subID)
+								close(wsSub.unsubscribed)
+							}, api.logger, "closing websocket peer sub")
+						}
 					}
 				}
 				api.filtersMu.RUnlock()