fix: check crosschain flags to stop inbound/outbound; get rid of outtx tracker iteration timeout

changelog.md

@@ -63,6 +63,8 @@
 * [1883](https://github.com/zeta-chain/node/issues/1883) - zetaclient should check 'IsSupported' flag to pause/unpause a specific chain
 * [1633](https://github.com/zeta-chain/node/issues/1633) - zetaclient should be able to pick up new connector and erc20Custody addresses
 * [1944](https://github.com/zeta-chain/node/pull/1944) - fix evm signer unit tests
+* [1888](https://github.com/zeta-chain/node/issues/1888) - zetaclient should stop inbound/outbound txs according to cross-chain flags
+* [1970](https://github.com/zeta-chain/node/issues/1970) - remove the timeout in the evm outtx tracker processing thread
 
 ### Chores
 

zetaclient/bitcoin/bitcoin_client.go

@@ -333,6 +333,9 @@
 	for {
 		select {
 		case <-ticker.C():
+			if flags := ob.coreContext.GetCrossChainFlags(); !flags.IsInboundEnabled {
+				continue
+			}
 			if !ob.GetChainParams().IsSupported {
 				continue
 			}
@@ -384,12 +387,6 @@
 }
 
 func (ob *BTCChainClient) ObserveInTx() error {
-	// make sure inbound TXS / Send is enabled by the protocol
-	flags := ob.coreContext.GetCrossChainFlags()
-	if !flags.IsInboundEnabled {
-		return errors.New("inbound TXS / Send has been disabled by the protocol")
-	}
-
 	// get and update latest block height
 	cnt, err := ob.rpcClient.GetBlockCount()
 	if err != nil {
@@ -438,6 +435,7 @@
 		}
 
 		// add block header to zetabridge
+		flags := ob.coreContext.GetCrossChainFlags()
 		if flags.BlockHeaderVerificationFlags != nil && flags.BlockHeaderVerificationFlags.IsBtcTypeChainEnabled {
 			err = ob.postBlockHeader(bn)
 			if err != nil {
@@ -583,11 +581,20 @@
 
 // WatchGasPrice watches Bitcoin chain for gas rate and post to zetacore
 func (ob *BTCChainClient) WatchGasPrice() {
+	// report gas price right away as the ticker takes time to kick in
+	err := ob.PostGasPrice()
+	if err != nil {
+		ob.logger.GasPrice.Error().Err(err).Msgf("PostGasPrice error for chain %d", ob.chain.ChainId)
+	}
+
+	// start gas price ticker
 	ticker, err := clienttypes.NewDynamicTicker("Bitcoin_WatchGasPrice", ob.GetChainParams().GasPriceTicker)
 	if err != nil {
 		ob.logger.GasPrice.Error().Err(err).Msg("error creating ticker")
 		return
 	}
+	ob.logger.GasPrice.Info().Msgf("WatchGasPrice started for chain %d with interval %d",
+		ob.chain.ChainId, ob.GetChainParams().GasPriceTicker)
 
 	defer ticker.Stop()
 	for {
@@ -1125,6 +1132,9 @@
 	for {
 		select {
 		case <-ticker.C():
+			if flags := ob.coreContext.GetCrossChainFlags(); !flags.IsOutboundEnabled {
+				continue
+			}
 			if !ob.GetChainParams().IsSupported {
 				continue
 			}

zetaclient/bitcoin/bitcoin_client_test.go

@@ -712,12 +712,3 @@ func TestGetBtcEventErrors(t *testing.T) {
 		require.Nil(t, event)
 	})
 }
-
-func TestBTCChainClient_ObserveInTx(t *testing.T) {
-	t.Run("should return error", func(t *testing.T) {
-		// create mainnet mock client
-		btcClient := MockBTCClientMainnet()
-		err := btcClient.ObserveInTx()
-		require.ErrorContains(t, err, "inbound TXS / Send has been disabled by the protocol")
-	})
-}

zetaclient/bitcoin/inbound_tracker.go

@@ -22,6 +22,9 @@
 	for {
 		select {
 		case <-ticker.C():
+			if flags := ob.coreContext.GetCrossChainFlags(); !flags.IsInboundEnabled {
+				continue
+			}
 			if !ob.GetChainParams().IsSupported {
 				continue
 			}

zetaclient/evm/evm_client.go

@@ -613,13 +613,6 @@
 
 // WatchOutTx watches evm chain for outgoing txs status
 func (ob *ChainClient) WatchOutTx() {
-	// read env variables if set
-	timeoutNonce, err := strconv.Atoi(os.Getenv("OS_TIMEOUT_NONCE"))
-	if err != nil || timeoutNonce <= 0 {
-		timeoutNonce = 100 * 3 // process up to 100 hashes
-	}
-	ob.logger.OutTx.Info().Msgf("WatchOutTx: using timeoutNonce %d seconds", timeoutNonce)
-
 	ticker, err := clienttypes.NewDynamicTicker(fmt.Sprintf("EVM_WatchOutTx_%d", ob.chain.ChainId), ob.GetChainParams().OutTxTicker)
 	if err != nil {
 		ob.logger.OutTx.Error().Err(err).Msg("error creating ticker")
@@ -630,16 +623,16 @@
 	for {
 		select {
 		case <-ticker.C():
+			if flags := ob.coreContext.GetCrossChainFlags(); !flags.IsOutboundEnabled {
+				continue
+			}
 			if !ob.GetChainParams().IsSupported {
 				continue
 			}
 			trackers, err := ob.zetaClient.GetAllOutTxTrackerByChain(ob.chain.ChainId, interfaces.Ascending)
 			if err != nil {
 				continue
 			}
-			//FIXME: remove this timeout here to ensure that all trackers are queried
-			outTimeout := time.After(time.Duration(timeoutNonce) * time.Second)
-		TRACKERLOOP:
 			for _, tracker := range trackers {
 				nonceInt := tracker.Nonce
 				if ob.isTxConfirmed(nonceInt) { // Go to next tracker if this one already has a confirmed tx
@@ -649,20 +642,14 @@
 				var receipt *ethtypes.Receipt
 				var transaction *ethtypes.Transaction
 				for _, txHash := range tracker.HashList {
-					select {
-					case <-outTimeout:
-						ob.logger.OutTx.Warn().Msgf("WatchOutTx: timeout on chain %d nonce %d", ob.chain.ChainId, nonceInt)
-						break TRACKERLOOP
-					default:
-						if recpt, tx, ok := ob.checkConfirmedTx(txHash.TxHash, nonceInt); ok {
-							txCount++
-							receipt = recpt
-							transaction = tx
-							ob.logger.OutTx.Info().Msgf("WatchOutTx: confirmed outTx %s for chain %d nonce %d", txHash.TxHash, ob.chain.ChainId, nonceInt)
-							if txCount > 1 {
-								ob.logger.OutTx.Error().Msgf(
-									"WatchOutTx: checkConfirmedTx passed, txCount %d chain %d nonce %d receipt %v transaction %v", txCount, ob.chain.ChainId, nonceInt, receipt, transaction)
-							}
+					if recpt, tx, ok := ob.checkConfirmedTx(txHash.TxHash, nonceInt); ok {
+						txCount++
+						receipt = recpt
+						transaction = tx
+						ob.logger.OutTx.Info().Msgf("WatchOutTx: confirmed outTx %s for chain %d nonce %d", txHash.TxHash, ob.chain.ChainId, nonceInt)
+						if txCount > 1 {
+							ob.logger.OutTx.Error().Msgf(
+								"WatchOutTx: checkConfirmedTx passed, txCount %d chain %d nonce %d receipt %v transaction %v", txCount, ob.chain.ChainId, nonceInt, receipt, transaction)
 						}
 					}
 				}
@@ -854,6 +841,9 @@
 	for {
 		select {
 		case <-ticker.C():
+			if flags := ob.coreContext.GetCrossChainFlags(); !flags.IsInboundEnabled {
+				continue
+			}
 			if !ob.GetChainParams().IsSupported {
 				sampledLogger.Info().Msgf("WatchInTx: chain %d is not supported", ob.chain.ChainId)
 				continue
@@ -918,12 +908,6 @@
 }
 
 func (ob *ChainClient) observeInTX(sampledLogger zerolog.Logger) error {
-	// make sure inbound TXS / Send is enabled by the protocol
-	flags := ob.coreContext.GetCrossChainFlags()
-	if !flags.IsInboundEnabled {
-		return errors.New("inbound TXS / Send has been disabled by the protocol")
-	}
-
 	// get and update latest block height
 	blockNumber, err := ob.evmClient.BlockNumber(context.Background())
 	if err != nil {
@@ -961,7 +945,7 @@
 	lastScannedDeposited := ob.ObserveERC20Deposited(startBlock, toBlock)
 
 	// task 3: query the incoming tx to TSS address (read at most 100 blocks in one go)
-	lastScannedTssRecvd := ob.ObserverTSSReceive(startBlock, toBlock, flags)
+	lastScannedTssRecvd := ob.ObserverTSSReceive(startBlock, toBlock)
 
 	// note: using lowest height for all 3 events is not perfect, but it's simple and good enough
 	lastScannedLowest := lastScannedZetaSent
@@ -1138,11 +1122,12 @@
 
 // ObserverTSSReceive queries the incoming gas asset to TSS address and posts to zetabridge
 // returns the last block successfully scanned
-func (ob *ChainClient) ObserverTSSReceive(startBlock, toBlock uint64, flags observertypes.CrosschainFlags) uint64 {
+func (ob *ChainClient) ObserverTSSReceive(startBlock, toBlock uint64) uint64 {
 	// query incoming gas asset
 	for bn := startBlock; bn <= toBlock; bn++ {
 		// post new block header (if any) to zetabridge and ignore error
 		// TODO: consider having a independent ticker(from TSS scaning) for posting block headers
+		flags := ob.coreContext.GetCrossChainFlags()
 		if flags.BlockHeaderVerificationFlags != nil &&
 			flags.BlockHeaderVerificationFlags.IsEthTypeChainEnabled &&
 			chains.IsHeaderSupportedEvmChain(ob.chain.ChainId) { // post block header for supported chains
@@ -1165,23 +1150,20 @@
 
 // WatchGasPrice watches evm chain for gas prices and post to zetacore
 func (ob *ChainClient) WatchGasPrice() {
-	ob.logger.GasPrice.Info().Msg("WatchGasPrice starting...")
+	// report gas price right away as the ticker takes time to kick in
 	err := ob.PostGasPrice()
 	if err != nil {
-		height, err := ob.zetaClient.GetBlockHeight()
-		if err != nil {
-			ob.logger.GasPrice.Error().Err(err).Msg("GetBlockHeight error")
-		} else {
-			ob.logger.GasPrice.Error().Err(err).Msgf("PostGasPrice error at zeta block : %d  ", height)
-		}
+		ob.logger.GasPrice.Error().Err(err).Msgf("PostGasPrice error for chain %d", ob.chain.ChainId)
 	}
 
+	// start gas price ticker
 	ticker, err := clienttypes.NewDynamicTicker(fmt.Sprintf("EVM_WatchGasPrice_%d", ob.chain.ChainId), ob.GetChainParams().GasPriceTicker)
 	if err != nil {
 		ob.logger.GasPrice.Error().Err(err).Msg("NewDynamicTicker error")
 		return
 	}
-	ob.logger.GasPrice.Info().Msgf("WatchGasPrice started with interval %d", ob.GetChainParams().GasPriceTicker)
+	ob.logger.GasPrice.Info().Msgf("WatchGasPrice started for chain %d with interval %d",
+		ob.chain.ChainId, ob.GetChainParams().GasPriceTicker)
 
 	defer ticker.Stop()
 	for {
@@ -1192,12 +1174,7 @@
 			}
 			err = ob.PostGasPrice()
 			if err != nil {
-				height, err := ob.zetaClient.GetBlockHeight()
-				if err != nil {
-					ob.logger.GasPrice.Error().Err(err).Msg("GetBlockHeight error")
-				} else {
-					ob.logger.GasPrice.Error().Err(err).Msgf("PostGasPrice error at zeta block : %d  ", height)
-				}
+				ob.logger.GasPrice.Error().Err(err).Msgf("PostGasPrice error for chain %d", ob.chain.ChainId)
 			}
 			ticker.UpdateInterval(ob.GetChainParams().GasPriceTicker, ob.logger.GasPrice)
 		case <-ob.stop:

zetaclient/evm/inbounds.go

@@ -8,20 +8,19 @@
 	"strings"
 
 	sdkmath "cosmossdk.io/math"
+	ethcommon "github.com/ethereum/go-ethereum/common"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/onrik/ethrpc"
 	"github.com/pkg/errors"
 	"github.com/zeta-chain/protocol-contracts/pkg/contracts/evm/erc20custody.sol"
 	"github.com/zeta-chain/protocol-contracts/pkg/contracts/evm/zetaconnector.non-eth.sol"
 	"github.com/zeta-chain/zetacore/pkg/chains"
 	"github.com/zeta-chain/zetacore/pkg/coin"
+	"github.com/zeta-chain/zetacore/pkg/constant"
+	"github.com/zeta-chain/zetacore/x/crosschain/types"
 	"github.com/zeta-chain/zetacore/zetaclient/compliance"
 	"github.com/zeta-chain/zetacore/zetaclient/config"
 	clienttypes "github.com/zeta-chain/zetacore/zetaclient/types"
-
-	ethcommon "github.com/ethereum/go-ethereum/common"
-	"github.com/zeta-chain/zetacore/pkg/constant"
-	"github.com/zeta-chain/zetacore/x/crosschain/types"
 	"github.com/zeta-chain/zetacore/zetaclient/zetabridge"
 	"golang.org/x/net/context"
 )
@@ -43,6 +42,9 @@
 	for {
 		select {
 		case <-ticker.C():
+			if flags := ob.coreContext.GetCrossChainFlags(); !flags.IsInboundEnabled {
+				continue
+			}
 			if !ob.GetChainParams().IsSupported {
 				continue
 			}

zetaclient/zetacore_observer.go

@@ -5,21 +5,18 @@
 	"math"
 	"time"
 
+	sdkmath "cosmossdk.io/math"
 	ethcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/rs/zerolog"
 	"github.com/zeta-chain/zetacore/pkg/chains"
+	"github.com/zeta-chain/zetacore/x/crosschain/types"
+	observertypes "github.com/zeta-chain/zetacore/x/observer/types"
 	appcontext "github.com/zeta-chain/zetacore/zetaclient/app_context"
 	"github.com/zeta-chain/zetacore/zetaclient/bitcoin"
 	corecontext "github.com/zeta-chain/zetacore/zetaclient/core_context"
 	"github.com/zeta-chain/zetacore/zetaclient/interfaces"
-	"github.com/zeta-chain/zetacore/zetaclient/outtxprocessor"
-
-	observertypes "github.com/zeta-chain/zetacore/x/observer/types"
-
-	sdkmath "cosmossdk.io/math"
-
-	"github.com/rs/zerolog"
-	"github.com/zeta-chain/zetacore/x/crosschain/types"
 	"github.com/zeta-chain/zetacore/zetaclient/metrics"
+	"github.com/zeta-chain/zetacore/zetaclient/outtxprocessor"
 )
 
 const (
@@ -138,46 +135,49 @@
 
 					// schedule keysign for pending cctxs on each chain
 					coreContext := appContext.ZetaCoreContext()
-					supportedChains := coreContext.GetEnabledChains()
-					for _, c := range supportedChains {
-						if c.ChainId == co.bridge.ZetaChain().ChainId {
-							continue
-						}
-						// update chain parameters for signer and chain client
-						signer, err := co.GetUpdatedSigner(coreContext, c.ChainId)
-						if err != nil {
-							co.logger.ZetaChainWatcher.Error().Err(err).Msgf("startCctxScheduler: getUpdatedSigner failed for chain %d", c.ChainId)
-							continue
-						}
-						ob, err := co.GetUpdatedChainClient(coreContext, c.ChainId)
-						if err != nil {
-							co.logger.ZetaChainWatcher.Error().Err(err).Msgf("startCctxScheduler: getTargetChainOb failed for chain %d", c.ChainId)
-							continue
-						}
-						if !ob.GetChainParams().IsSupported {
-							co.logger.ZetaChainWatcher.Info().Msgf("startCctxScheduler: chain %d is not supported", c.ChainId)
-							continue
-						}
-
-						cctxList, totalPending, err := co.bridge.ListPendingCctx(c.ChainId)
-						if err != nil {
-							co.logger.ZetaChainWatcher.Error().Err(err).Msgf("startCctxScheduler: ListPendingCctx failed for chain %d", c.ChainId)
-							continue
-						}
-						// Set Pending transactions prometheus gauge
-						metrics.PendingTxsPerChain.WithLabelValues(c.ChainName.String()).Set(float64(totalPending))
-
-						// #nosec G701 range is verified
-						zetaHeight := uint64(bn)
-						if chains.IsEVMChain(c.ChainId) {
-							co.scheduleCctxEVM(outTxMan, zetaHeight, c.ChainId, cctxList, ob, signer)
-						} else if chains.IsBitcoinChain(c.ChainId) {
-							co.scheduleCctxBTC(outTxMan, zetaHeight, c.ChainId, cctxList, ob, signer)
-						} else {
-							co.logger.ZetaChainWatcher.Error().Msgf("startCctxScheduler: unsupported chain %d", c.ChainId)
-							continue
+					if flags := coreContext.GetCrossChainFlags(); flags.IsOutboundEnabled {
+						supportedChains := coreContext.GetEnabledChains()
+						for _, c := range supportedChains {
+							if c.ChainId == co.bridge.ZetaChain().ChainId {
+								continue
+							}
+							// update chain parameters for signer and chain client
+							signer, err := co.GetUpdatedSigner(coreContext, c.ChainId)
+							if err != nil {
+								co.logger.ZetaChainWatcher.Error().Err(err).Msgf("startCctxScheduler: getUpdatedSigner failed for chain %d", c.ChainId)
+								continue
+							}
+							ob, err := co.GetUpdatedChainClient(coreContext, c.ChainId)
+							if err != nil {
+								co.logger.ZetaChainWatcher.Error().Err(err).Msgf("startCctxScheduler: getTargetChainOb failed for chain %d", c.ChainId)
+								continue
+							}
+							if !ob.GetChainParams().IsSupported {
+								co.logger.ZetaChainWatcher.Info().Msgf("startCctxScheduler: chain %d is not supported", c.ChainId)
+								continue
+							}
+
+							cctxList, totalPending, err := co.bridge.ListPendingCctx(c.ChainId)
+							if err != nil {
+								co.logger.ZetaChainWatcher.Error().Err(err).Msgf("startCctxScheduler: ListPendingCctx failed for chain %d", c.ChainId)
+								continue
+							}
+							// Set Pending transactions prometheus gauge
+							metrics.PendingTxsPerChain.WithLabelValues(c.ChainName.String()).Set(float64(totalPending))
+
+							// #nosec G701 range is verified
+							zetaHeight := uint64(bn)
+							if chains.IsEVMChain(c.ChainId) {
+								co.scheduleCctxEVM(outTxMan, zetaHeight, c.ChainId, cctxList, ob, signer)
+							} else if chains.IsBitcoinChain(c.ChainId) {
+								co.scheduleCctxBTC(outTxMan, zetaHeight, c.ChainId, cctxList, ob, signer)
+							} else {
+								co.logger.ZetaChainWatcher.Error().Msgf("startCctxScheduler: unsupported chain %d", c.ChainId)
+								continue
+							}
 						}
 					}
+
 					// update last processed block number
 					lastBlockNum = bn
 					metrics.LastCoreBlockNumber.Set(float64(lastBlockNum))