sync with develop branch

changelog.md

@@ -21,6 +21,8 @@
 ### Fixes
 
 * [1554](https://github.com/zeta-chain/node/pull/1554) - Screen out unconfirmed UTXOs that are not created by TSS itself
+* [1560](https://github.com/zeta-chain/node/issues/1560) - Zetaclient post evm-chain outtx hashes only when receipt is available
+* [1516](https://github.com/zeta-chain/node/issues/1516) - Unprivileged outtx tracker removal
 * [1537](https://github.com/zeta-chain/node/issues/1537) - Sanity check events of ZetaSent/ZetaReceived/ZetaRevertedWithdrawn/Deposited
 * [1530](https://github.com/zeta-chain/node/pull/1530) - Outbound tx confirmation/inclusion enhancement
 * [1496](https://github.com/zeta-chain/node/issues/1496) - post block header for enabled EVM chains only
@@ -36,6 +38,7 @@
 * [1525](https://github.com/zeta-chain/node/pull/1525) - relax EVM chain block header length check 1024->4096
 * [1522](https://github.com/zeta-chain/node/pull/1522/files) - block `distribution` module account from receiving zeta
 * [1528](https://github.com/zeta-chain/node/pull/1528) - fix panic caused on decoding malformed BTC addresses
+* [1555](https://github.com/zeta-chain/node/pull/1555) - Reduce websocket message limit to 10MB
 
 ### Refactoring
 
@@ -52,6 +55,8 @@
 * Remove chain id from the index for observer mapper and rename it to observer set.
 * Add logger to smoke tests
 * [1521](https://github.com/zeta-chain/node/pull/1521) - replace go-tss lib version with one that reverts back to thorchain tss-lib
+* [1558](https://github.com/zeta-chain/node/pull/1558) - change log level for gas stability pool iteration error
+* Update --ledger flag hint
 
 ### Chores
 * [1446](https://github.com/zeta-chain/node/pull/1446) - renamed file `zetaclientd/aux.go` to `zetaclientd/utils.go` to avoid complaints from go package resolver. 

cmd/zetacored/main.go

@@ -34,7 +34,7 @@ func processError(err error) {
 	if strings.Contains(err.Error(), "cannot set custom bip32 path with ledger") {
 		printNotice([]string{
 			"note: --ledger flag can't be used with Ethereum HD path (used by default)",
-			"use --hd-path=\"\" in the command to use Cosmos HD path",
+			"Please set a blank path with --hd-path=\"\" to use Cosmos HD path instead.",
 		})
 		os.Exit(1)
 	}

rpc/websockets.go

@@ -47,7 +47,7 @@ import (
 )
 
 const (
-	messageSizeLimit = 32 * 1024 * 1024 // 32MB
+	messageSizeLimit = 10 * 1024 * 1024 // 10MB
 
 )
 

x/crosschain/keeper/msg_server_add_to_outtx_tracker.go

@@ -29,6 +29,24 @@ func (k msgServer) AddToOutTxTracker(goCtx context.Context, msg *types.MsgAddToO
 		return nil, observertypes.ErrSupportedChains
 	}
 
+	// the cctx must exist
+	cctx, err := k.CctxByNonce(ctx, &types.QueryGetCctxByNonceRequest{
+		ChainID: msg.ChainId,
+		Nonce:   msg.Nonce,
+	})
+	if err != nil {
+		return nil, cosmoserrors.Wrap(err, "CcxtByNonce failed")
+	}
+	if cctx == nil || cctx.CrossChainTx == nil {
+		return nil, cosmoserrors.Wrapf(types.ErrCannotFindCctx, "no corresponding cctx found for chain %d, nonce %d", msg.ChainId, msg.Nonce)
+	}
+	// tracker submission is only allowed when the cctx is pending
+	if !IsPending(*cctx.CrossChainTx) {
+		// garbage tracker (for any reason) is harmful to outTx observation and should be removed
+		k.RemoveOutTxTracker(ctx, msg.ChainId, msg.Nonce)
+		return &types.MsgAddToOutTxTrackerResponse{IsRemoved: true}, nil
+	}
+
 	if msg.Proof == nil { // without proof, only certain accounts can send this message
 		adminPolicyAccount := k.zetaObserverKeeper.GetParams(ctx).GetAdminPolicyAccount(observertypes.Policy_Type_group1)
 		isAdmin := msg.Creator == adminPolicyAccount
@@ -53,18 +71,6 @@ func (k msgServer) AddToOutTxTracker(goCtx context.Context, msg *types.MsgAddToO
 		isProven = true
 	}
 
-	cctx, err := k.CctxByNonce(ctx, &types.QueryGetCctxByNonceRequest{
-		ChainID: msg.ChainId,
-		Nonce:   msg.Nonce,
-	})
-	if err != nil || cctx == nil || cctx.CrossChainTx == nil {
-		return nil, cosmoserrors.Wrap(types.ErrCannotFindCctx, "cannot add out tx: no corresponding cctx found")
-	}
-	if !IsPending(*cctx.CrossChainTx) {
-		k.RemoveOutTxTracker(ctx, msg.ChainId, msg.Nonce)
-		return &types.MsgAddToOutTxTrackerResponse{IsRemoved: true}, nil
-	}
-
 	tracker, found := k.GetOutTxTracker(ctx, msg.ChainId, msg.Nonce)
 	hash := types.TxHashList{
 		TxHash:   msg.TxHash,

x/crosschain/module.go

@@ -190,7 +190,7 @@ func (AppModule) ConsensusVersion() uint64 { return 4 }
 func (am AppModule) BeginBlock(ctx sdk.Context, _ abci.RequestBeginBlock) {
 	err := am.keeper.IterateAndUpdateCctxGasPrice(ctx)
 	if err != nil {
-		ctx.Logger().Error("Error iterating and updating pending cctx gas price", "err", err.Error())
+		ctx.Logger().Info("Error iterating and updating pending cctx gas price", "err", err.Error())
 	}
 }
 

zetaclient/bitcoin_client.go

@@ -365,9 +365,11 @@ func (ob *BitcoinChainClient) observeInTx() error {
 		}
 
 		// add block header to zetacore
-		err = ob.postBlockHeader(bn)
-		if err != nil {
-			ob.logger.WatchInTx.Warn().Err(err).Msgf("observeInTxBTC: error posting block header %d", bn)
+		if flags.BlockHeaderVerificationFlags != nil && flags.BlockHeaderVerificationFlags.IsBtcTypeChainEnabled {
+			err = ob.postBlockHeader(bn)
+			if err != nil {
+				ob.logger.WatchInTx.Warn().Err(err).Msgf("observeInTxBTC: error posting block header %d", bn)
+			}
 		}
 
 		tssAddress := ob.Tss.BTCAddress()

zetaclient/evm_client.go

@@ -885,7 +885,7 @@ func (ob *EVMChainClient) observeInTX(sampledLogger zerolog.Logger) error {
 	lastScannedDeposited := ob.observeERC20Deposited(startBlock, toBlock)
 
 	// task 3: query the incoming tx to TSS address (read at most 100 blocks in one go)
-	lastScannedTssRecvd := ob.observeTssRecvd(startBlock, toBlock)
+	lastScannedTssRecvd := ob.observeTssRecvd(startBlock, toBlock, flags)
 
 	// note: using lowest height for all 3 events is not perfect, but it's simple and good enough
 	lastScannedLowest := lastScannedZetaSent
@@ -1068,7 +1068,7 @@ func (ob *EVMChainClient) observeERC20Deposited(startBlock, toBlock uint64) uint
 
 // observeTssRecvd queries the incoming gas asset to TSS address and posts to zetacore
 // returns the last block successfully scanned
-func (ob *EVMChainClient) observeTssRecvd(startBlock, toBlock uint64) uint64 {
+func (ob *EVMChainClient) observeTssRecvd(startBlock, toBlock uint64, flags observertypes.CrosschainFlags) uint64 {
 	// check TSS address (after keygen, ob.Tss.pubkey will be updated)
 	tssAddress := ob.Tss.EVMAddress()
 	if tssAddress == (ethcommon.Address{}) {
@@ -1080,7 +1080,9 @@ func (ob *EVMChainClient) observeTssRecvd(startBlock, toBlock uint64) uint64 {
 	for bn := startBlock; bn <= toBlock; bn++ {
 		// post new block header (if any) to zetacore and ignore error
 		// TODO: consider having a independent ticker(from TSS scaning) for posting block headers
-		if common.IsHeaderSupportedEvmChain(ob.chain.ChainId) { // post block header for supported chains
+		if flags.BlockHeaderVerificationFlags != nil &&
+			flags.BlockHeaderVerificationFlags.IsEthTypeChainEnabled &&
+			common.IsHeaderSupportedEvmChain(ob.chain.ChainId) { // post block header for supported chains
 			err := ob.postBlockHeader(toBlock)
 			if err != nil {
 				ob.logger.ExternalChainWatcher.Error().Err(err).Msg("error posting block header")

zetaclient/evm_signer.go

@@ -9,6 +9,7 @@ import (
 	"math/rand"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
@@ -24,6 +25,10 @@ import (
 	observertypes "github.com/zeta-chain/zetacore/x/observer/types"
 )
 
+const (
+	OutTxTrackerReportTimeout = 10 * time.Minute
+)
+
 type EVMSigner struct {
 	client                      EVMRPCClient
 	chain                       *common.Chain
@@ -36,6 +41,10 @@ type EVMSigner struct {
 	erc20CustodyContractAddress ethcommon.Address
 	logger                      zerolog.Logger
 	ts                          *TelemetryServer
+
+	// for outTx tracker report only
+	mu                     *sync.Mutex
+	outTxHashBeingReported map[string]bool
 }
 
 var _ ChainSigner = &EVMSigner{}
@@ -83,7 +92,9 @@ func NewEVMSigner(
 		logger: logger.With().
 			Str("chain", chain.ChainName.String()).
 			Str("module", "EVMSigner").Logger(),
-		ts: ts,
+		ts:                     ts,
+		mu:                     &sync.Mutex{},
+		outTxHashBeingReported: make(map[string]bool),
 	}, nil
 }
 
@@ -566,11 +577,7 @@ func (signer *EVMSigner) TryProcessOutTx(
 				log.Warn().Err(err).Msgf("OutTx Broadcast error")
 				retry, report := HandleBroadcastError(err, strconv.FormatUint(send.GetCurrentOutTxParam().OutboundTxTssNonce, 10), toChain.String(), outTxHash)
 				if report {
-					zetaHash, err := zetaBridge.AddTxHashToOutTxTracker(toChain.ChainId, tx.Nonce(), outTxHash, nil, "", -1)
-					if err != nil {
-						logger.Err(err).Msgf("Unable to add to tracker on ZetaCore: nonce %d chain %s outTxHash %s", send.GetCurrentOutTxParam().OutboundTxTssNonce, toChain, outTxHash)
-					}
-					logger.Info().Msgf("Broadcast to core successful %s", zetaHash)
+					signer.reportToOutTxTracker(zetaBridge, toChain.ChainId, tx.Nonce(), outTxHash, logger)
 				}
 				if !retry {
 					break
@@ -579,15 +586,62 @@ func (signer *EVMSigner) TryProcessOutTx(
 				continue
 			}
 			logger.Info().Msgf("Broadcast success: nonce %d to chain %s outTxHash %s", send.GetCurrentOutTxParam().OutboundTxTssNonce, toChain, outTxHash)
-			zetaHash, err := zetaBridge.AddTxHashToOutTxTracker(toChain.ChainId, tx.Nonce(), outTxHash, nil, "", -1)
-			if err != nil {
-				logger.Err(err).Msgf("Unable to add to tracker on ZetaCore: nonce %d chain %s outTxHash %s", send.GetCurrentOutTxParam().OutboundTxTssNonce, toChain, outTxHash)
-			}
-			logger.Info().Msgf("Broadcast to core successful %s", zetaHash)
+			signer.reportToOutTxTracker(zetaBridge, toChain.ChainId, tx.Nonce(), outTxHash, logger)
 			break // successful broadcast; no need to retry
 		}
+	}
+}
 
+// reportToOutTxTracker reports outTxHash to tracker only when tx receipt is available
+func (signer *EVMSigner) reportToOutTxTracker(zetaBridge ZetaCoreBridger, chainID int64, nonce uint64, outTxHash string, logger zerolog.Logger) {
+	// skip if already being reported
+	signer.mu.Lock()
+	defer signer.mu.Unlock()
+	if _, found := signer.outTxHashBeingReported[outTxHash]; found {
+		logger.Info().Msgf("reportToOutTxTracker: outTxHash %s for chain %d nonce %d is being reported", outTxHash, chainID, nonce)
+		return
 	}
+	signer.outTxHashBeingReported[outTxHash] = true // mark as being reported
+
+	// report to outTx tracker with goroutine
+	go func() {
+		defer func() {
+			signer.mu.Lock()
+			delete(signer.outTxHashBeingReported, outTxHash)
+			signer.mu.Unlock()
+		}()
+
+		// try fetching tx receipt for 10 minutes
+		tStart := time.Now()
+		for {
+			if time.Since(tStart) > OutTxTrackerReportTimeout { // give up after 10 minutes
+				logger.Info().Msgf("reportToOutTxTracker: outTxHash report timeout for chain %d nonce %d outTxHash %s", chainID, nonce, outTxHash)
+				return
+			}
+			receipt, err := signer.client.TransactionReceipt(context.TODO(), ethcommon.HexToHash(outTxHash))
+			if err != nil {
+				logger.Info().Err(err).Msgf("reportToOutTxTracker: receipt not available for chain %d nonce %d outTxHash %s", chainID, nonce, outTxHash)
+				time.Sleep(10 * time.Second)
+				continue
+			}
+			if receipt != nil {
+				_, isPending, err := signer.client.TransactionByHash(context.TODO(), ethcommon.HexToHash(outTxHash))
+				if err != nil || isPending {
+					logger.Info().Err(err).Msgf("reportToOutTxTracker: error getting tx or tx is pending for chain %d nonce %d outTxHash %s", chainID, nonce, outTxHash)
+					time.Sleep(10 * time.Second)
+					continue
+				}
+				break
+			}
+		}
+
+		// report to outTxTracker
+		zetaHash, err := zetaBridge.AddTxHashToOutTxTracker(chainID, nonce, outTxHash, nil, "", -1)
+		if err != nil {
+			logger.Err(err).Msgf("reportToOutTxTracker: unable to add to tracker on ZetaCore for chain %d nonce %d outTxHash %s", chainID, nonce, outTxHash)
+		}
+		logger.Info().Msgf("reportToOutTxTracker: reported outTxHash to core successful %s, chain %d nonce %d outTxHash %s", zetaHash, chainID, nonce, outTxHash)
+	}()
 }
 
 // SignERC20WithdrawTx