feat: support multiple ethereum grpc endpoints

cmd/peggo/bridge.go

@@ -119,13 +119,13 @@ func deployGravityCmd() *cobra.Command {
 			}
 
 			// ETH RPC
-			ethRPCEndpoint := konfig.String(flagEthRPC)
-			ethRPC, err := ethclient.Dial(ethRPCEndpoint)
+			InitEthRPCManager(konfig)
+			ethRPC, err := ethManager.GetEthClient()
 			if err != nil {
-				return fmt.Errorf("failed to dial Ethereum RPC node: %w", err)
+				return err
 			}
 
-			auth, err := buildTransactOpts(konfig, ethRPC)
+			auth, err := buildTransactOpts(konfig)
 			if err != nil {
 				return err
 			}
@@ -200,13 +200,13 @@ func deployERC20Cmd() *cobra.Command {
 				return err
 			}
 
-			ethRPCEndpoint := konfig.String(flagEthRPC)
-			ethRPC, err := ethclient.Dial(ethRPCEndpoint)
+			InitEthRPCManager(konfig)
+			ethRPC, err := ethManager.GetEthClient()
 			if err != nil {
-				return fmt.Errorf("failed to dial Ethereum RPC node: %w", err)
+				return err
 			}
 
-			auth, err := buildTransactOpts(konfig, ethRPC)
+			auth, err := buildTransactOpts(konfig)
 			if err != nil {
 				return err
 			}
@@ -341,13 +341,13 @@ network starting.`,
 				return err
 			}
 
-			ethRPCEndpoint := konfig.String(flagEthRPC)
-			ethRPC, err := ethclient.Dial(ethRPCEndpoint)
+			InitEthRPCManager(konfig)
+			ethRPC, err := ethManager.GetEthClient()
 			if err != nil {
-				return fmt.Errorf("failed to dial Ethereum RPC node: %w", err)
+				return err
 			}
 
-			auth, err := buildTransactOpts(konfig, ethRPC)
+			auth, err := buildTransactOpts(konfig)
 			if err != nil {
 				return err
 			}
@@ -403,10 +403,10 @@ func sendToCosmosCmd() *cobra.Command {
 				return err
 			}
 
-			ethRPCEndpoint := konfig.String(flagEthRPC)
-			ethRPC, err := ethclient.Dial(ethRPCEndpoint)
+			InitEthRPCManager(konfig)
+			ethRPC, err := ethManager.GetEthClient()
 			if err != nil {
-				return fmt.Errorf("failed to dial Ethereum RPC node: %w", err)
+				return err
 			}
 
 			gravityAddr := args[0]
@@ -425,7 +425,7 @@ func sendToCosmosCmd() *cobra.Command {
 				}
 			}
 
-			auth, err := buildTransactOpts(konfig, ethRPC)
+			auth, err := buildTransactOpts(konfig)
 			if err != nil {
 				return err
 			}
@@ -468,7 +468,9 @@ Transaction: %s
 	return cmd
 }
 
-func buildTransactOpts(konfig *koanf.Koanf, ethClient *ethclient.Client) (*bind.TransactOpts, error) {
+func buildTransactOpts(konfig *koanf.Koanf) (*bind.TransactOpts, error) {
+	InitEthRPCManager(konfig)
+
 	ethPrivKeyHexStr := konfig.String(flagEthPK)
 
 	privKey, err := ethcrypto.ToECDSA(ethcmn.FromHex(ethPrivKeyHexStr))
@@ -487,15 +489,15 @@ func buildTransactOpts(konfig *koanf.Koanf, ethClient *ethclient.Client) (*bind.
 
 	fromAddress := ethcrypto.PubkeyToAddress(*publicKeyECDSA)
 
-	nonce, err := ethClient.PendingNonceAt(goCtx, fromAddress)
+	nonce, err := ethManager.PendingNonceAt(goCtx, fromAddress)
 	if err != nil {
 		return nil, err
 	}
 
 	goCtx, cancel = context.WithTimeout(context.Background(), 15*time.Second)
 	defer cancel()
 
-	ethChainID, err := ethClient.ChainID(goCtx)
+	ethChainID, err := ethManager.ChainID(goCtx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get Ethereum chain ID: %w", err)
 	}
@@ -516,7 +518,7 @@ func buildTransactOpts(konfig *koanf.Koanf, ethClient *ethclient.Client) (*bind.
 		gasPrice = big.NewInt(gasPriceInt)
 
 	default:
-		gasPrice, err = ethClient.SuggestGasPrice(context.Background())
+		gasPrice, err = ethManager.SuggestGasPrice(context.Background())
 		if err != nil {
 			return nil, fmt.Errorf("failed to get Ethereum gas estimate: %w", err)
 		}
@@ -564,7 +566,7 @@ func approveERC20(konfig *koanf.Koanf, ethRPC *ethclient.Client, erc20AddrStr, g
 		return fmt.Errorf("failed to create ERC20 contract instance: %w", err)
 	}
 
-	auth, err := buildTransactOpts(konfig, ethRPC)
+	auth, err := buildTransactOpts(konfig)
 	if err != nil {
 		return err
 	}

cmd/peggo/flags.go

@@ -2,6 +2,7 @@
 package peggo
 
 import (
+	"fmt"
 	"net/url"
 	"strings"
 
@@ -37,6 +38,7 @@ const (
 	flagEthPK                   = "eth-pk"
 	flagEthUseLedger            = "eth-use-ledger"
 	flagEthRPC                  = "eth-rpc"
+	flagEthRPCs                 = "eth-rpcs"
 	flagEthGasAdjustment        = "eth-gas-price-adjustment"
 	flagEthGasLimitAdjustment   = "eth-gas-limit-adjustment"
 	flagEthAlchemyWS            = "eth-alchemy-ws"
@@ -93,8 +95,10 @@ func ethereumOptsFlagSet() *pflag.FlagSet {
 	fs := pflag.NewFlagSet("", pflag.ContinueOnError)
 
 	fs.String(flagEthRPC, "http://localhost:8545", "Specify the RPC address of an Ethereum node")
+	fs.StringSlice(flagEthRPCs, []string{"http://localhost:8545"}, "Specify RPC addresses of one or more Ethereum nodes")
 	fs.Float64(flagEthGasAdjustment, float64(1.3), "Specify a gas price adjustment for Ethereum transactions")
 	fs.Float64(flagEthGasLimitAdjustment, float64(1.2), "Specify a gas limit adjustment for Ethereum transactions")
+	_ = fs.MarkDeprecated(flagEthRPC, fmt.Sprintf("Use the '%s' flag instead to provide one or more Ethereum RPC instances", flagEthRPCs))
 
 	return fs
 }
@@ -103,8 +107,10 @@ func bridgeFlagSet() *pflag.FlagSet {
 	fs := pflag.NewFlagSet("", pflag.ContinueOnError)
 
 	fs.String(flagEthRPC, "http://localhost:8545", "Specify the RPC address of an Ethereum node")
+	fs.String(flagEthRPCs, "http://localhost:8545", "Specify comma-separated RPC addresses of one or more Ethereum nodes")
 	fs.Int64(flagEthGasPrice, 0, "The Ethereum gas price to include in the transaction; If zero, gas price will be estimated")
 	fs.Int64(flagEthGasLimit, 6000000, "The Ethereum gas limit to include in the transaction")
+	_ = fs.MarkDeprecated(flagEthRPC, fmt.Sprintf("Use the '%s' flag instead to provide one or more Ethereum RPC instances", flagEthRPCs))
 
 	return fs
 }

cmd/peggo/grpcclient.go

@@ -0,0 +1,139 @@
+package peggo
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/knadh/koanf"
+	"github.com/pkg/errors"
+)
+
+var ethManager *EthRPCManager
+
+type EthRPCManager struct {
+	currentEndpoint int // the slice index of the endpoint currently used
+	client          *rpc.Client
+	konfig          *koanf.Koanf
+}
+
+// initializes the single instance of EthRPCManager with a given config (uses flagEthRPCs).
+// no-op if already initialized, even if konfig would be different.
+func InitEthRPCManager(konfig *koanf.Koanf) {
+	if ethManager == nil {
+		ethManager = &EthRPCManager{
+			konfig: konfig,
+		}
+	}
+}
+
+// closes and sets to nil the stored eth RPC client
+func (em *EthRPCManager) CloseClient() {
+	if em.client != nil {
+		em.client.Close()
+		em.client = nil
+	}
+}
+
+// closes the current client and dials configured ethereum rpc endpoints in a roundrobin fashion until one
+// is connected. returns an error if no endpoints ar configured or all dials failed
+func (em *EthRPCManager) DialNext() error {
+	if em.konfig == nil {
+		return errors.New("ethRPCManager konfig is nil")
+	}
+
+	rpcs := strings.Split(strings.ReplaceAll(em.konfig.String(flagEthRPCs), " ", ""), ",")
+
+	em.CloseClient()
+
+	dialIndex := func(i int) bool {
+		if cli, err := rpc.Dial(rpcs[i]); err == nil {
+			em.currentEndpoint = i
+			em.client = cli
+			return true
+		}
+		// todo: should likely log the error
+		return false
+	}
+
+	// first tries all endpoints in the slice after the current index
+	for i := range rpcs {
+		if i > em.currentEndpoint && dialIndex(i) {
+			return nil
+		}
+	}
+
+	// then tries remaining endpoints from the beginning of the slice
+	for i := range rpcs {
+		if i <= em.currentEndpoint && dialIndex(i) {
+			return nil
+		}
+	}
+
+	return errors.New(fmt.Sprintf("could not dial any of the %d Ethereum RPC endpoints configured", len(rpcs)))
+}
+
+// returns the current eth RPC client, dialing one first if nonexistent
+func (em *EthRPCManager) GetClient() (*rpc.Client, error) {
+	if em.client == nil {
+		if err := em.DialNext(); err != nil {
+			return nil, err
+		}
+	}
+	return em.client, nil
+}
+
+// returns the current eth RPC client, dialing one first if nonexistent
+func (em *EthRPCManager) GetEthClient() (*ethclient.Client, error) {
+	cli, err := em.GetClient()
+	if err != nil {
+		return nil, err
+	}
+	return ethclient.NewClient(cli), nil
+}
+
+// wraps ethclient.PendingNonceAt, also closing client if PendingNonceAt returns an error
+func (em *EthRPCManager) PendingNonceAt(ctx context.Context, addr common.Address) (uint64, error) {
+	cli, err := em.GetEthClient()
+	if err != nil {
+		return 0, err
+	}
+	nonce, err := cli.PendingNonceAt(ctx, addr)
+	if err != nil {
+		em.CloseClient()
+		return 0, err
+	}
+	return nonce, nil
+}
+
+// wraps ethclient.ChainID, also closing client if ChainID returns an error
+func (em *EthRPCManager) ChainID(ctx context.Context) (*big.Int, error) {
+	cli, err := em.GetEthClient()
+	if err != nil {
+		return nil, err
+	}
+	id, err := cli.ChainID(ctx)
+	if err != nil {
+		em.CloseClient()
+		return nil, err
+	}
+	return id, nil
+}
+
+// wraps ethclient.SuggestGasPrice, also closing client if SuggestGasPrice returns an error
+func (em *EthRPCManager) SuggestGasPrice(ctx context.Context) (*big.Int, error) {
+	cli, err := em.GetEthClient()
+	if err != nil {
+		return nil, err
+	}
+	price, err := cli.SuggestGasPrice(ctx)
+	if err != nil {
+		em.CloseClient()
+		return nil, err
+	}
+	return price, nil
+}

cmd/peggo/orchestrator.go

@@ -11,7 +11,6 @@ import (
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	ethcmn "github.com/ethereum/go-ethereum/common"
-	ethrpc "github.com/ethereum/go-ethereum/rpc"
 	"github.com/rs/zerolog"
 	"github.com/spf13/cobra"
 	rpchttp "github.com/tendermint/tendermint/rpc/client/http"
@@ -123,13 +122,11 @@ func getOrchestratorCmd() *cobra.Command {
 				return fmt.Errorf("failed to initialize Ethereum account: %w", err)
 			}
 
-			ethRPCEndpoint := konfig.String(flagEthRPC)
-			ethRPC, err := ethrpc.Dial(ethRPCEndpoint)
+			ethRPC, err := ethManager.GetClient()
 			if err != nil {
 				return fmt.Errorf("failed to dial Ethereum RPC node: %w", err)
 			}
 
-			fmt.Fprintf(os.Stderr, "Connected to Ethereum RPC: %s\n", ethRPCEndpoint)
 			ethProvider := provider.NewEVMProvider(ethRPC)
 
 			ethGasPriceAdjustment := konfig.Float64(flagEthGasAdjustment)

cmd/peggo/peggo.go

@@ -21,7 +21,7 @@ func NewRootCmd() *cobra.Command {
 		Long: `Peggo is a companion executable for orchestrating a Gravity validator.
 
 Inputs in the CLI commands can be provided via flags or environment variables. If
-using the later, prefix the environment variable with PEGGO_ and the named of the
+using the latter, prefix the environment variable with PEGGO_ and the named of the
 flag (e.g. PEGGO_COSMOS_PK).`,
 	}
 
@@ -81,7 +81,7 @@ func parseServerConfig(cmd *cobra.Command) (*koanf.Koanf, error) {
 	konfig := koanf.New(".")
 
 	// load from file first (if provided)
-	// TODO: Support config files if/when needed.
+	// TODO: Support config files if/when needed. Koanf also supports watch the config file for changes
 	// if configPath := ctx.String(config.ConfigPath); len(configPath) != 0 {
 	// 	if err := konfig.Load(file.Provider(configPath), toml.Parser()); err != nil {
 	// 		return nil, err

test/e2e/e2e_setup_test.go

@@ -545,7 +545,7 @@ func (s *IntegrationTestSuite) runContractDeployment() {
 				"peggo",
 				"bridge",
 				"deploy-gravity",
-				"--eth-rpc",
+				"--eth-rpcs",
 				fmt.Sprintf("http://%s:8545", s.ethResource.Container.Name[1:]),
 				"--cosmos-grpc",
 				fmt.Sprintf("tcp://%s:9090", s.valResources[0].Container.Name[1:]),