Improves/request retry

apps/bitcoin/rpc.go

@@ -9,10 +9,10 @@ import (
 	"math/big"
 	"net/http"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/MixinNetwork/mixin/logger"
+	"github.com/MixinNetwork/safe/util"
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/shopspring/decimal"
@@ -367,15 +367,11 @@ func callBitcoinRPCUntilSufficient(rpc, method string, params []any) ([]byte, er
 			return res, nil
 		}
 		logger.Printf("callBitcoinRPC(%s, %s, %v) => %v", rpc, method, params, err)
-		reason := strings.ToLower(err.Error())
-		switch {
-		case strings.Contains(reason, "timeout"):
-		case strings.Contains(reason, "eof"):
-		case strings.Contains(reason, "handshake"):
-		default:
-			return res, err
+		if util.CheckRetryableError(err) {
+			time.Sleep(7 * time.Second)
+			continue
 		}
-		time.Sleep(7 * time.Second)
+		return res, err
 	}
 }
 

apps/ethereum/account.go

@@ -319,6 +319,7 @@ func FetchBalanceFromKey(ctx context.Context, rpc, key string) (*big.Int, error)
 	if err != nil {
 		return nil, err
 	}
+	defer client.Close()
 
 	balance, err := client.BalanceAt(ctx, *addr, nil)
 	if err != nil {

apps/ethereum/rpc.go

@@ -16,9 +16,11 @@
 
 	"github.com/MixinNetwork/mixin/logger"
 	"github.com/MixinNetwork/safe/apps/ethereum/abi"
+	"github.com/MixinNetwork/safe/util"
 	"github.com/ethereum/go-ethereum"
 	ga "github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethclient"
 )
@@ -141,10 +143,10 @@
 	if err != nil {
 		return nil, err
 	}
-	var b RPCBlockWithTransactions
+	var b *RPCBlockWithTransactions
 	err = json.Unmarshal(res, &b)
-	if err != nil {
-		return nil, err
+	if err != nil || b == nil {
+		return nil, fmt.Errorf("RPCGetBlockWithTransactions(%d) => Unmarshal() => %v, %v", height, err, b)
 	}
 	blockHeight, err := ethereumNumberToUint64(b.Number)
 	if err != nil {
@@ -154,7 +156,7 @@
 	for _, tx := range b.Tx {
 		tx.BlockHash = b.Hash
 	}
-	return &b, err
+	return b, err
 }
 
 func RPCGetGasPrice(rpc string) (*big.Int, error) {
@@ -284,12 +286,6 @@
 }
 
 func GetERC20TransferLogFromBlock(ctx context.Context, rpc string, chain, height int64) ([]*Transfer, error) {
-	client, err := ethclient.Dial(rpc)
-	if err != nil {
-		return nil, err
-	}
-	defer client.Close()
-
 	logTransferSig := []byte("Transfer(address,address,uint256)")
 	logTransferSigHash := crypto.Keccak256Hash(logTransferSig)
 	query := ethereum.FilterQuery{
@@ -301,7 +297,7 @@
 	if err != nil {
 		log.Fatal(err)
 	}
-	logs, err := client.FilterLogs(ctx, query)
+	logs, err := filterLogsUntilSufficient(ctx, rpc, query)
 	if err != nil {
 		return nil, err
 	}
@@ -340,15 +336,31 @@
 			return res, nil
 		}
 		logger.Printf("callEthereumRPC(%s, %s, %v) => %v", rpc, method, params, err)
-		reason := strings.ToLower(err.Error())
-		switch {
-		case strings.Contains(reason, "timeout"):
-		case strings.Contains(reason, "eof"):
-		case strings.Contains(reason, "handshake"):
-		default:
-			return res, err
+		if util.CheckRetryableError(err) {
+			time.Sleep(7 * time.Second)
+			continue
+		}
+		return res, err
+	}
+}
+
+func filterLogsUntilSufficient(ctx context.Context, rpc string, query ethereum.FilterQuery) ([]types.Log, error) {
+	client, err := ethclient.Dial(rpc)
+	if err != nil {
+		return nil, err
+	}
+	defer client.Close()
+
+	for {
+		logs, err := client.FilterLogs(ctx, query)
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
+		if util.CheckRetryableError(err) {
+			time.Sleep(7 * time.Second)
+			continue
 		}
-		time.Sleep(7 * time.Second)
+		return logs, err
 	}
 }
 

apps/mixin/rpc.go

@@ -7,10 +7,10 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"strings"
 	"time"
 
 	"github.com/MixinNetwork/mixin/logger"
+	"github.com/MixinNetwork/safe/util"
 )
 
 type WithdrawalData struct {
@@ -69,16 +69,11 @@ func callMixinRPCUntilSufficient(rpc, method string, params []any) ([]byte, erro
 			return res, nil
 		}
 		logger.Printf("callMixinRPC(%s, %s, %v) => %v", rpc, method, params, err)
-		reason := strings.ToLower(err.Error())
-		switch {
-		case strings.Contains(reason, "timeout"):
-		case strings.Contains(reason, "eof"):
-		case strings.Contains(reason, "handshake"):
-		case strings.Contains(reason, "invalid character"):
-		default:
-			return res, err
+		if util.CheckRetryableError(err) {
+			time.Sleep(7 * time.Second)
+			continue
 		}
-		time.Sleep(7 * time.Second)
+		return res, err
 	}
 }
 

cmd/transaction.go

@@ -1,13 +1,11 @@
 package cmd
 
 import (
+	"context"
 	"encoding/base64"
 	"encoding/hex"
-	"encoding/json"
 	"fmt"
-	"net/http"
 	"strings"
-	"time"
 
 	"github.com/MixinNetwork/safe/apps/bitcoin"
 	"github.com/MixinNetwork/safe/apps/ethereum"
@@ -22,6 +20,7 @@ import (
 )
 
 func GenerateTestTransactionProposal(c *cli.Context) error {
+	ctx := context.Background()
 	chain := c.Int("chain")
 	switch chain {
 	case common.SafeChainBitcoin:
@@ -43,7 +42,7 @@ func GenerateTestTransactionProposal(c *cli.Context) error {
 
 	addr := abi.GetFactoryAssetAddress("0x11EC02748116A983deeD59235302C3139D6e8cdD", common.SafeBitcoinChainId, "BTC", "Bitcoin", holder)
 	assetKey := strings.ToLower(addr.String())
-	bondId := fetchAssetId(ethereum.GenerateAssetId(common.SafeChainPolygon, assetKey))
+	bondId := fetchAssetId(ctx, ethereum.GenerateAssetId(common.SafeChainPolygon, assetKey))
 
 	extra := []byte(receiver)
 	sid := uuid.Must(uuid.NewV4()).String()
@@ -94,24 +93,13 @@ func GenerateTestTransactionApproval(c *cli.Context) error {
 	return nil
 }
 
-func fetchAssetId(mixinId string) string {
-	client := &http.Client{Timeout: 10 * time.Second}
-	path := "https://api.mixin.one/network/assets/" + mixinId
-	resp, err := client.Get(path)
-	if err != nil {
+func fetchAssetId(ctx context.Context, mixinId string) string {
+	asset, err := common.SafeReadAssetUntilSufficient(ctx, mixinId)
+	if err != nil || asset == nil {
 		panic(mixinId)
 	}
-	defer resp.Body.Close()
-
-	var body struct {
-		Data struct {
-			AssetId string `json:"asset_id"`
-			MixinId string `json:"mixin_id"`
-		} `json:"data"`
-	}
-	json.NewDecoder(resp.Body).Decode(&body)
-	if body.Data.MixinId != mixinId {
+	if asset.KernelAssetID != mixinId {
 		panic(mixinId)
 	}
-	return body.Data.AssetId
+	return asset.AssetID
 }

common/util.go

@@ -12,7 +12,6 @@ import (
 	"strings"
 
 	"github.com/MixinNetwork/mixin/crypto"
-	"github.com/MixinNetwork/safe/mtg"
 	"github.com/MixinNetwork/safe/util"
 	"github.com/fox-one/mixin-sdk-go/v2/mixinnet"
 )
@@ -79,7 +78,7 @@ func ExpandTilde(path string) string {
 }
 
 func CheckRetryableError(err error) bool {
-	return mtg.CheckRetryableError(err) || CheckTransactionLockedError(err)
+	return util.CheckRetryableError(err) || CheckTransactionLockedError(err)
 }
 
 func CheckTransactionLockedError(err error) bool {

common/wallet.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/MixinNetwork/bot-api-go-client/v3"
+	"github.com/MixinNetwork/mixin/logger"
 	"github.com/fox-one/mixin-sdk-go/v2"
 	"github.com/fox-one/mixin-sdk-go/v2/mixinnet"
 	"github.com/shopspring/decimal"
@@ -36,6 +37,10 @@ func (mw *MixinWallet) drainOutputsFromNetwork(ctx context.Context) {
 		}
 		members := []string{mw.client.ClientID}
 		utxos, err := listUnspentUTXOsUntilSufficient(ctx, mw.client, members, 1, "", checkpoint)
+		if CheckRetryableError(err) {
+			logger.Verbosef("listUnspentUTXOsUntilSufficient() => %v", err)
+			continue
+		}
 		if err != nil {
 			panic(err)
 		}

keeper/network.go

@@ -4,12 +4,9 @@ import (
 	"context"
 	"encoding/binary"
 	"encoding/hex"
-	"encoding/json"
 	"fmt"
 	"math/big"
-	"net/http"
 	"slices"
-	"strings"
 	"time"
 
 	"github.com/MixinNetwork/mixin/crypto"
@@ -19,6 +16,7 @@ import (
 	"github.com/MixinNetwork/safe/common"
 	"github.com/MixinNetwork/safe/keeper/store"
 	"github.com/MixinNetwork/safe/mtg"
+	"github.com/MixinNetwork/safe/util"
 	"github.com/gofrs/uuid/v5"
 	"github.com/shopspring/decimal"
 )
@@ -240,40 +238,19 @@ func (node *Node) fetchAssetMetaFromMessengerOrEthereum(ctx context.Context, id,
 	return asset, node.store.WriteAssetMeta(ctx, asset)
 }
 
-func (node *Node) fetchMixinAsset(_ context.Context, id string) (*store.Asset, error) {
-	client := &http.Client{Timeout: 10 * time.Second}
-	path := node.conf.MixinMessengerAPI + "/network/assets/" + id
-	resp, err := client.Get(path)
-	if err != nil {
-		return nil, err
-	}
-	defer common.CloseOrPanic(resp.Body)
-
-	var body struct {
-		Data *struct {
-			AssetId   string      `json:"asset_id"`
-			MixinId   crypto.Hash `json:"mixin_id"`
-			AssetKey  string      `json:"asset_key"`
-			Symbol    string      `json:"symbol"`
-			Name      string      `json:"name"`
-			Precision uint32      `json:"precision"`
-			ChainId   string      `json:"chain_id"`
-		} `json:"data"`
-	}
-	err = json.NewDecoder(resp.Body).Decode(&body)
-	if err != nil {
+func (node *Node) fetchMixinAsset(ctx context.Context, id string) (*store.Asset, error) {
+	asset, err := common.SafeReadAssetUntilSufficient(ctx, id)
+	if err != nil || asset == nil {
 		return nil, err
 	}
-	asset := body.Data
-
 	return &store.Asset{
-		AssetId:   asset.AssetId,
-		MixinId:   asset.MixinId.String(),
+		AssetId:   asset.AssetID,
+		MixinId:   asset.KernelAssetID,
 		AssetKey:  asset.AssetKey,
 		Symbol:    asset.Symbol,
 		Name:      asset.Name,
-		Decimals:  asset.Precision,
-		Chain:     common.SafeAssetIdChain(asset.ChainId),
+		Decimals:  uint32(asset.Precision),
+		Chain:     common.SafeAssetIdChainNoPanic(asset.ChainID),
 		CreatedAt: time.Now().UTC(),
 	}, nil
 }
@@ -287,16 +264,15 @@ func (node *Node) fetchAssetMeta(ctx context.Context, id string) (*store.Asset,
 	for {
 		meta, err = node.fetchMixinAsset(ctx, id)
 		if err == nil {
+			if meta == nil {
+				return nil, fmt.Errorf("fetchAssetMeta(%s) => nil", id)
+			}
 			return meta, node.store.WriteAssetMeta(ctx, meta)
 		}
-		reason := strings.ToLower(err.Error())
-		switch {
-		case strings.Contains(reason, "timeout"):
-		case strings.Contains(reason, "eof"):
-		case strings.Contains(reason, "handshake"):
-		default:
-			return nil, err
+		if util.CheckRetryableError(err) {
+			time.Sleep(2 * time.Second)
+			continue
 		}
-		time.Sleep(2 * time.Second)
+		return nil, err
 	}
 }

messenger/mixin.go

@@ -13,7 +13,7 @@ import (
 	"github.com/MixinNetwork/bot-api-go-client/v3"
 	"github.com/MixinNetwork/mixin/logger"
 	"github.com/MixinNetwork/safe/common"
-	"github.com/MixinNetwork/safe/mtg"
+	"github.com/MixinNetwork/safe/util"
 	"github.com/fox-one/mixin-sdk-go/v2"
 	"github.com/gofrs/uuid/v5"
 )
@@ -216,7 +216,7 @@ func (mm *MixinMessenger) SyncAck() bool {
 func (mm *MixinMessenger) sendMessagesWithoutTimeout(ctx context.Context, batch []*mixin.MessageRequest) error {
 	for {
 		err := mm.client.SendMessages(ctx, batch)
-		if err != nil && mtg.CheckRetryableError(err) {
+		if err != nil && util.CheckRetryableError(err) {
 			logger.Printf("messenger.sendMessagesWithoutTimeout(retry, %d) => %v", len(batch), err)
 			time.Sleep(3 * time.Second)
 			continue

mtg/deposit.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/MixinNetwork/mixin/logger"
+	"github.com/MixinNetwork/safe/util"
 )
 
 type DepositEntry struct {
@@ -61,7 +62,7 @@ func (grp *Group) readOutputDepositUntilSufficientImpl(ctx context.Context, id s
 		var deposit *SafeDepositView
 		err := grp.mixin.Get(ctx, fmt.Sprintf("/safe/outputs/%s/deposit", id), nil, &deposit)
 		logger.Verbosef("Group.readOutputDeposit(%s) => %v %v\n", id, deposit, err)
-		if CheckRetryableError(err) {
+		if util.CheckRetryableError(err) {
 			time.Sleep(3 * time.Second)
 			continue
 		}