auto_balance_test and handle auto-unbond and auto-bridge concurrency

Author: 0xSpuddy

cmd/main.go

@@ -138,9 +138,9 @@ func init() {
 	rootCmd.Flags().Duration("refresh-gas-estimates-interval", 12*time.Hour, "Interval for resetting cached gas estimates and gas-adjustment levels (<=0 disables)")
 
 	// Auto-bridge: keep wallet at a fixed balance by bridging the excess to Ethereum
-	rootCmd.Flags().Uint64("auto-balance-to-keep", 0, "Keep this amount of loya in the wallet; bridge any excess to Ethereum at --auto-balance-execution-time (0 = disabled)")
-	rootCmd.Flags().String("auto-balance-execution-time", "00:00", "UTC time to execute the auto-balance bridge (HH:MM, e.g. '03:00')")
-	rootCmd.Flags().String("auto-balance-bridge-to-eth-addr", "", "Ethereum address to bridge excess tokens to (required when auto-balance-to-keep > 0)")
+	rootCmd.Flags().Uint64(daemonflags.FlagAutoBalanceToKeep, 0, "Keep this amount of loya in the wallet; bridge any excess to Ethereum at --auto-balance-execution-time (0 = disabled)")
+	rootCmd.Flags().String(daemonflags.FlagAutoBalanceExecutionTime, "00:00", "UTC time to execute the auto-balance bridge (HH:MM, e.g. '03:00')")
+	rootCmd.Flags().String(daemonflags.FlagAutoBalanceBridgeToEthAddr, "", "Ethereum address to bridge excess tokens to (required when auto-balance-to-keep > 0)")
 
 	// Marking required flags
 	if err := rootCmd.MarkFlagRequired(flags.FlagHome); err != nil {

flags/flags.go

@@ -17,6 +17,10 @@ const (
 	FlagPriceDaemonLoopDelayMs = "price-daemon-loop-delay-ms"
 
 	FlagKeyringBackend = "keyring-backend"
+
+	FlagAutoBalanceToKeep          = "auto-balance-to-keep"
+	FlagAutoBalanceExecutionTime   = "auto-balance-execution-time"
+	FlagAutoBalanceBridgeToEthAddr = "auto-balance-bridge-to-eth-addr"
 )
 
 // Shared flags contains configuration flags shared by all daemons.

reporter/client/auto_balance_test.go

@@ -0,0 +1,59 @@
+package client
+
+import (
+	"testing"
+
+	"cosmossdk.io/math"
+
+	"github.com/stretchr/testify/require"
+	bridgetypes "github.com/tellor-io/layer/x/bridge/types"
+	oracletypes "github.com/tellor-io/layer/x/oracle/types"
+)
+
+func TestNormalizeAutoBalanceEthAddr(t *testing.T) {
+	addr, err := normalizeAutoBalanceEthAddr("0x0000000000000000000000000000000000000001")
+	require.NoError(t, err)
+	require.Equal(t, "0000000000000000000000000000000000000001", addr)
+
+	addr, err = normalizeAutoBalanceEthAddr("0000000000000000000000000000000000000001")
+	require.NoError(t, err)
+	require.Equal(t, "0000000000000000000000000000000000000001", addr)
+
+	addr, err = normalizeAutoBalanceEthAddr("")
+	require.NoError(t, err)
+	require.Empty(t, addr)
+
+	_, err = normalizeAutoBalanceEthAddr("not-an-address")
+	require.Error(t, err)
+}
+
+func TestParseAutoBalanceExecutionTime(t *testing.T) {
+	hour, minute, err := parseAutoBalanceExecutionTime("03:05")
+	require.NoError(t, err)
+	require.Equal(t, 3, hour)
+	require.Equal(t, 5, minute)
+
+	_, _, err = parseAutoBalanceExecutionTime("24:00")
+	require.Error(t, err)
+
+	_, _, err = parseAutoBalanceExecutionTime("03")
+	require.Error(t, err)
+}
+
+func TestIsBridgeDepositReportMsg(t *testing.T) {
+	require.True(t, isBridgeDepositReportMsg(&oracletypes.MsgSubmitValue{}))
+	require.False(t, isBridgeDepositReportMsg(&bridgetypes.MsgWithdrawTokens{}))
+}
+
+func TestShouldSkipAutoUnbond(t *testing.T) {
+	reporterStake := math.LegacyNewDec(1_000)
+	unbondAmount := math.NewInt(100)
+	tenPercent, err := math.LegacyNewDecFromStr("0.10")
+	require.NoError(t, err)
+	fivePercent, err := math.LegacyNewDecFromStr("0.05")
+	require.NoError(t, err)
+
+	require.False(t, shouldSkipAutoUnbond(reporterStake, math.LegacyZeroDec(), unbondAmount), "zero max percentage disables the cap")
+	require.False(t, shouldSkipAutoUnbond(reporterStake, tenPercent, unbondAmount))
+	require.True(t, shouldSkipAutoUnbond(reporterStake, fivePercent, unbondAmount))
+}

reporter/client/broadcast_message.go

@@ -222,13 +222,13 @@ func (c *Client) BroadcastTxMsgToChain(ctx context.Context) {
 				txCtx, cancel := context.WithTimeout(ctx, txBroadcastTimeout)
 				defer cancel()
 
-				if !txInfo.isBridge {
-					_, err := c.sendTx(txCtx, txInfo.QueryMetaId, false, txInfo.Msg)
+				if txInfo.isBridge && isBridgeDepositReportMsg(txInfo.Msg) {
+					c.HandleBridgeDepositTxInChannel(txCtx, txInfo)
+				} else {
+					_, err := c.sendTx(txCtx, txInfo.QueryMetaId, txInfo.isBridge, txInfo.Msg)
 					if err != nil {
 						c.logger.Error(fmt.Sprintf("Error sending tx: %v", err))
 					}
-				} else {
-					c.HandleBridgeDepositTxInChannel(txCtx, txInfo)
 				}
 			}(obj)
 
@@ -237,3 +237,8 @@ func (c *Client) BroadcastTxMsgToChain(ctx context.Context) {
 		}
 	}
 }
+
+func isBridgeDepositReportMsg(msg interface{}) bool {
+	_, ok := msg.(*oracletypes.MsgSubmitValue)
+	return ok
+}

reporter/client/client.go

@@ -4,16 +4,18 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	rpchttp "github.com/cometbft/cometbft/rpc/client/http"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/spf13/viper"
 	globalfeetypes "github.com/strangelove-ventures/globalfee/x/globalfee/types"
 	customquery "github.com/tellor-io/layer-daemons/custom_query"
-	"github.com/tellor-io/layer-daemons/flags"
+	daemonflags "github.com/tellor-io/layer-daemons/flags"
 	pricefeedtypes "github.com/tellor-io/layer-daemons/pricefeed/client/types"
 	pricefeedservertypes "github.com/tellor-io/layer-daemons/server/types/pricefeed"
 	tokenbridgetypes "github.com/tellor-io/layer-daemons/server/types/token_bridge"
@@ -122,7 +124,7 @@ func NewClient(logger log.Logger, valGasMin string) *Client {
 
 func (c *Client) Start(
 	ctx context.Context,
-	flags flags.DaemonFlags,
+	flags daemonflags.DaemonFlags,
 	grpcAddress string,
 	grpcClient daemontypes.GrpcClient,
 	marketParams []pricefeedtypes.MarketParam,
@@ -211,6 +213,9 @@ func (c *Client) Start(
 	c.refreshGasEstimatesInterval = viper.GetDuration("refresh-gas-estimates-interval")
 
 	if autoUnbondingFrequency > 0 {
+		if autoUnbondingFrequency > 21 {
+			return fmt.Errorf("auto-unbonding-frequency must be between 1 and 21 days when set, got: %d", autoUnbondingFrequency)
+		}
 		if autoUnbondingAmount == 0 {
 			return fmt.Errorf("auto-unbonding-amount must be greater than 0 when auto-unbonding-frequency is set")
 		}
@@ -245,15 +250,21 @@ func (c *Client) Start(
 	}
 
 	// Read and validate auto-balance-to-keep configuration
-	autoBalanceToKeep := viper.GetUint64("auto-balance-to-keep")
-	autoBalanceEthAddr := strings.TrimPrefix(viper.GetString("auto-balance-eth-addr"), "0x")
+	autoBalanceToKeep := viper.GetUint64(daemonflags.FlagAutoBalanceToKeep)
 	if autoBalanceToKeep > 0 {
+		autoBalanceEthAddr, err := normalizeAutoBalanceEthAddr(viper.GetString(daemonflags.FlagAutoBalanceBridgeToEthAddr))
+		if err != nil {
+			return err
+		}
 		if autoBalanceEthAddr == "" {
-			return fmt.Errorf("auto-balance-eth-addr is required when auto-balance-to-keep > 0")
+			return fmt.Errorf("%s is required when %s > 0", daemonflags.FlagAutoBalanceBridgeToEthAddr, daemonflags.FlagAutoBalanceToKeep)
+		}
+		if _, _, err := parseAutoBalanceExecutionTime(viper.GetString(daemonflags.FlagAutoBalanceExecutionTime)); err != nil {
+			return err
 		}
 		c.logger.Info("Auto balance-to-keep enabled",
 			"balance_to_keep_loya", autoBalanceToKeep,
-			"execution_time", viper.GetString("auto-balance-execution-time"),
+			"execution_time", viper.GetString(daemonflags.FlagAutoBalanceExecutionTime),
 			"eth_addr", "0x"+autoBalanceEthAddr,
 		)
 	} else {
@@ -312,7 +323,7 @@ func (c *Client) Start(
 func StartReporterDaemonTaskLoop(
 	client *Client,
 	ctx context.Context,
-	flags flags.DaemonFlags,
+	flags daemonflags.DaemonFlags,
 	wg *sync.WaitGroup,
 ) {
 	reporterCreated := false
@@ -477,6 +488,30 @@ func (c *Client) trySend(ctx context.Context, info TxChannelInfo) bool {
 	}
 }
 
+func normalizeAutoBalanceEthAddr(addr string) (string, error) {
+	addr = strings.TrimSpace(addr)
+	if addr == "" {
+		return "", nil
+	}
+	if !common.IsHexAddress(addr) {
+		return "", fmt.Errorf("%s must be a valid Ethereum address, got: %s", daemonflags.FlagAutoBalanceBridgeToEthAddr, addr)
+	}
+	return strings.TrimPrefix(common.HexToAddress(addr).Hex(), "0x"), nil
+}
+
+func parseAutoBalanceExecutionTime(executionTime string) (int, int, error) {
+	parts := strings.SplitN(executionTime, ":", 2)
+	if len(parts) != 2 {
+		return 0, 0, fmt.Errorf("invalid %s, expected HH:MM, got: %s", daemonflags.FlagAutoBalanceExecutionTime, executionTime)
+	}
+	hour, errH := strconv.Atoi(parts[0])
+	minute, errM := strconv.Atoi(parts[1])
+	if errH != nil || errM != nil || hour < 0 || hour > 23 || minute < 0 || minute > 59 {
+		return 0, 0, fmt.Errorf("invalid %s value, expected HH:MM in UTC, got: %s", daemonflags.FlagAutoBalanceExecutionTime, executionTime)
+	}
+	return hour, minute, nil
+}
+
 // Stop stops the reporter client gracefully
 func (c *Client) Stop() {
 	c.stopOnce.Do(func() {

reporter/client/reporter_monitors.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/shirou/gopsutil/v3/process"
 	"github.com/spf13/viper"
+	"github.com/tellor-io/layer-daemons/flags"
 	tokenbridgetipstypes "github.com/tellor-io/layer-daemons/server/types/token_bridge_tips"
 	oracletypes "github.com/tellor-io/layer/x/oracle/types"
 	reportertypes "github.com/tellor-io/layer/x/reporter/types"
@@ -305,9 +306,8 @@ func (c *Client) AutoUnbondStakePeriodically(ctx context.Context, wg *sync.WaitG
 				}
 			}
 
-			maxStakeAbleToWithdraw := reporterStake.Mul(maxStakePercentage)
-
-			if maxStakeAbleToWithdraw.LT(math.LegacyNewDecFromInt(unbondAmount)) {
+			if shouldSkipAutoUnbond(reporterStake, maxStakePercentage, unbondAmount) {
+				maxStakeAbleToWithdraw := reporterStake.Mul(maxStakePercentage)
 				c.logger.Info("Not enough stake to withdraw", "reporterStake", reporterStake, "maxStakeAbleToWithdraw", maxStakeAbleToWithdraw)
 				continue
 			}
@@ -323,31 +323,33 @@ func (c *Client) AutoUnbondStakePeriodically(ctx context.Context, wg *sync.WaitG
 	}
 }
 
+func shouldSkipAutoUnbond(reporterStake math.LegacyDec, maxStakePercentage math.LegacyDec, unbondAmount math.Int) bool {
+	if !maxStakePercentage.GT(math.LegacyZeroDec()) {
+		return false
+	}
+	return reporterStake.Mul(maxStakePercentage).LT(math.LegacyNewDecFromInt(unbondAmount))
+}
+
 // AutoBridgeWalletExcessPeriodically watches the wallet balance once per day at the configured
 // UTC time. Whenever the balance exceeds --auto-balance-to-keep (loya), the excess (minus a
-// 1 TRB gas reserve) is bridged to the Ethereum address supplied by --auto-balance-eth-addr.
+// 1 TRB gas reserve) is bridged to the Ethereum address supplied by --auto-balance-bridge-to-eth-addr.
 func (c *Client) AutoBridgeWalletExcessPeriodically(ctx context.Context, wg *sync.WaitGroup) {
 	defer wg.Done()
 
-	balanceToKeep := viper.GetUint64("auto-balance-to-keep")
+	balanceToKeep := viper.GetUint64(flags.FlagAutoBalanceToKeep)
 	if balanceToKeep == 0 {
 		c.logger.Info("Auto balance-to-keep is disabled")
 		return
 	}
 
-	ethAddr := strings.TrimPrefix(viper.GetString("auto-balance-eth-addr"), "0x")
-	executionTime := viper.GetString("auto-balance-execution-time")
-
-	// Parse HH:MM
-	parts := strings.SplitN(executionTime, ":", 2)
-	if len(parts) != 2 {
-		c.logger.Error("invalid auto-balance-execution-time, expected HH:MM", "value", executionTime)
+	ethAddr, err := normalizeAutoBalanceEthAddr(viper.GetString(flags.FlagAutoBalanceBridgeToEthAddr))
+	if err != nil {
+		c.logger.Error("invalid auto-balance bridge address", "error", err)
 		return
 	}
-	hour, errH := strconv.Atoi(parts[0])
-	minute, errM := strconv.Atoi(parts[1])
-	if errH != nil || errM != nil || hour < 0 || hour > 23 || minute < 0 || minute > 59 {
-		c.logger.Error("invalid auto-balance-execution-time value", "value", executionTime)
+	hour, minute, err := parseAutoBalanceExecutionTime(viper.GetString(flags.FlagAutoBalanceExecutionTime))
+	if err != nil {
+		c.logger.Error("invalid auto-balance execution time", "error", err)
 		return
 	}
 
@@ -402,7 +404,7 @@ func (c *Client) AutoBridgeWalletExcessPeriodically(ctx context.Context, wg *syn
 			Recipient: ethAddr,
 			Amount:    sdk.NewCoin("loya", amountToBridge),
 		}
-		c.txChan <- TxChannelInfo{Msg: msg, isBridge: true, NumRetries: 0, QueryMetaId: 0}
+		c.trySend(ctx, TxChannelInfo{Msg: msg, isBridge: true, NumRetries: 0, QueryMetaId: 0})
 	}
 }