Merge pull request #102 from InjectiveLabs/cleanup

Cleanup

Makefile

@@ -19,7 +19,7 @@ push:
 install: export GOPROXY=direct
 install: export VERSION_FLAGS="-X $(VERSION_PKG).GitCommit=$(GIT_COMMIT) -X $(VERSION_PKG).BuildDate=$(BUILD_DATE)"
 install:
-	$(DOCKER) && go install -tags muslc -ldflags $(VERSION_FLAGS) ./cmd/... || go install -ldflags $(VERSION_FLAGS) ./cmd/...
+	docker=$(DOCKER) && go install -tags muslc -ldflags $(VERSION_FLAGS) ./cmd/... || go install -ldflags $(VERSION_FLAGS) ./cmd/...
 
 .PHONY: install image push test gen
 

cmd/peggo/options.go

@@ -61,14 +61,12 @@ func initCosmosOptions(
 		Name:   "cosmos-grpc",
 		Desc:   "Cosmos GRPC querying endpoint",
 		EnvVar: "PEGGO_COSMOS_GRPC",
-		Value:  "tcp://localhost:9900",
 	})
 
 	*tendermintRPC = cmd.String(cli.StringOpt{
 		Name:   "tendermint-rpc",
 		Desc:   "Tendermint RPC endpoint",
 		EnvVar: "PEGGO_TENDERMINT_RPC",
-		Value:  "http://localhost:26657",
 	})
 
 	*cosmosGasPrices = cmd.String(cli.StringOpt{
@@ -290,14 +288,12 @@ func initConfig(cmd *cli.Cmd) Config {
 		Name:   "cosmos-grpc",
 		Desc:   "Cosmos GRPC querying endpoint",
 		EnvVar: "PEGGO_COSMOS_GRPC",
-		Value:  "tcp://localhost:9900",
 	})
 
 	cfg.tendermintRPC = cmd.String(cli.StringOpt{
 		Name:   "tendermint-rpc",
 		Desc:   "Tendermint RPC endpoint",
 		EnvVar: "PEGGO_TENDERMINT_RPC",
-		Value:  "http://localhost:26657",
 	})
 
 	cfg.cosmosGasPrices = cmd.String(cli.StringOpt{

cmd/peggo/orchestrator.go

@@ -2,12 +2,9 @@ package main
 
 import (
 	"context"
-	"github.com/InjectiveLabs/peggo/orchestrator/version"
 	"os"
-	"time"
 
-	ctypes "github.com/InjectiveLabs/sdk-go/chain/types"
-	ethcmn "github.com/ethereum/go-ethereum/common"
+	gethcommon "github.com/ethereum/go-ethereum/common"
 	cli "github.com/jawher/mow.cli"
 	"github.com/xlab/closer"
 	log "github.com/xlab/suplog"
@@ -16,6 +13,8 @@ import (
 	"github.com/InjectiveLabs/peggo/orchestrator/coingecko"
 	"github.com/InjectiveLabs/peggo/orchestrator/cosmos"
 	"github.com/InjectiveLabs/peggo/orchestrator/ethereum"
+	"github.com/InjectiveLabs/peggo/orchestrator/version"
+	chaintypes "github.com/InjectiveLabs/sdk-go/chain/types"
 )
 
 // startOrchestrator action runs an infinite loop,
@@ -24,7 +23,6 @@ import (
 // $ peggo orchestrator
 func orchestratorCmd(cmd *cli.Cmd) {
 	// orchestrator-specific CLI options
-	cfg := initConfig(cmd)
 
 	cmd.Before = func() {
 		initMetrics(cmd)
@@ -34,6 +32,8 @@ func orchestratorCmd(cmd *cli.Cmd) {
 		// ensure a clean exit
 		defer closer.Close()
 
+		cfg := initConfig(cmd)
+
 		log.WithFields(log.Fields{
 			"version":    version.AppVersion,
 			"git":        version.GitCommit,
@@ -71,47 +71,53 @@ func orchestratorCmd(cmd *cli.Cmd) {
 			log.WithError(err).Fatalln("failed to initialize Ethereum account")
 		}
 
-		log.WithFields(log.Fields{
-			"inj_addr": valAddress.String(),
-			"eth_addr": ethKeyFromAddress.String(),
-		}).Infoln("starting peggo service")
-
-		// Connect to Injective network
-		injNetwork, err := cosmos.NewNetwork(
-			*cfg.cosmosChainID,
-			valAddress.String(),
-			*cfg.cosmosGRPC,
-			*cfg.cosmosGasPrices,
-			*cfg.tendermintRPC,
-			cosmosKeyring,
-			signerFn,
-			personalSignFn,
+		log.WithFields(log.Fields{"inj_addr": valAddress.String(), "eth_addr": ethKeyFromAddress.String()}).Infoln("starting peggo service")
+
+		var (
+			injectiveNet       orchestrator.InjectiveNetwork
+			customEndpointRPCs = *cfg.cosmosGRPC != "" && *cfg.tendermintRPC != ""
 		)
-		orShutdown(err)
 
-		// See if the provided ETH address belongs to a validator and determine in which mode peggo should run
-		isValidator, err := isValidatorAddress(injNetwork.PeggyQueryClient, ethKeyFromAddress)
-		if err != nil {
-			log.WithError(err).Fatalln("failed to query current validator set on Injective")
+		if customEndpointRPCs {
+			injectiveNet, err = cosmos.NewCustomRPCNetwork(
+				*cfg.cosmosChainID,
+				valAddress.String(),
+				*cfg.cosmosGRPC,
+				*cfg.cosmosGasPrices,
+				*cfg.tendermintRPC,
+				cosmosKeyring,
+				personalSignFn,
+			)
+		} else {
+			// load balanced connection
+			injectiveNet, err = cosmos.NewLoadBalancedNetwork(
+				*cfg.cosmosChainID,
+				valAddress.String(),
+				*cfg.cosmosGasPrices,
+				cosmosKeyring,
+				personalSignFn,
+			)
 		}
 
+		orShutdown(err)
+
 		ctx, cancelFn := context.WithCancel(context.Background())
 		closer.Bind(cancelFn)
 
 		// Construct erc20 token mapping
-		peggyParams, err := injNetwork.PeggyParams(ctx)
+		peggyParams, err := injectiveNet.PeggyParams(ctx)
 		if err != nil {
 			log.WithError(err).Fatalln("failed to query peggy params, is injectived running?")
 		}
 
-		peggyContractAddr := ethcmn.HexToAddress(peggyParams.BridgeEthereumAddress)
-		injTokenAddr := ethcmn.HexToAddress(peggyParams.CosmosCoinErc20Contract)
+		peggyContractAddr := gethcommon.HexToAddress(peggyParams.BridgeEthereumAddress)
+		injTokenAddr := gethcommon.HexToAddress(peggyParams.CosmosCoinErc20Contract)
 
-		erc20ContractMapping := make(map[ethcmn.Address]string)
-		erc20ContractMapping[injTokenAddr] = ctypes.InjectiveCoin
+		erc20ContractMapping := make(map[gethcommon.Address]string)
+		erc20ContractMapping[injTokenAddr] = chaintypes.InjectiveCoin
 
 		// Connect to ethereum network
-		ethNetwork, err := ethereum.NewNetwork(
+		ethereumNet, err := ethereum.NewNetwork(
 			*cfg.ethNodeRPC,
 			peggyContractAddr,
 			ethKeyFromAddress,
@@ -127,8 +133,8 @@ func orchestratorCmd(cmd *cli.Cmd) {
 
 		// Create peggo and run it
 		peggo, err := orchestrator.NewPeggyOrchestrator(
-			injNetwork,
-			ethNetwork,
+			injectiveNet,
+			ethereumNet,
 			coingeckoFeed,
 			erc20ContractMapping,
 			*cfg.minBatchFeeUSD,
@@ -140,7 +146,7 @@ func orchestratorCmd(cmd *cli.Cmd) {
 		orShutdown(err)
 
 		go func() {
-			if err := peggo.Run(ctx, isValidator); err != nil {
+			if err := peggo.Run(ctx); err != nil {
 				log.Errorln(err)
 				os.Exit(1)
 			}
@@ -149,22 +155,3 @@ func orchestratorCmd(cmd *cli.Cmd) {
 		closer.Hold()
 	}
 }
-
-func isValidatorAddress(peggyQuery cosmos.PeggyQueryClient, addr ethcmn.Address) (bool, error) {
-	ctx, cancelFn := context.WithTimeout(context.Background(), time.Second*30)
-	defer cancelFn()
-
-	currentValset, err := peggyQuery.CurrentValset(ctx)
-	if err != nil {
-		return false, err
-	}
-
-	var isValidator bool
-	for _, validator := range currentValset.Members {
-		if ethcmn.HexToAddress(validator.EthereumAddress) == addr {
-			isValidator = true
-		}
-	}
-
-	return isValidator, nil
-}

cmd/peggo/tx.go

@@ -4,19 +4,10 @@ import (
 	"context"
 	"time"
 
-	rpchttp "github.com/cometbft/cometbft/rpc/client/http"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/connectivity"
-
 	cli "github.com/jawher/mow.cli"
 	"github.com/xlab/closer"
 	log "github.com/xlab/suplog"
 
-	"github.com/InjectiveLabs/sdk-go/chain/peggy/types"
-	chainclient "github.com/InjectiveLabs/sdk-go/client/chain"
-	"github.com/InjectiveLabs/sdk-go/client/common"
-
-
 	"github.com/InjectiveLabs/peggo/orchestrator/cosmos"
 )
 
@@ -135,44 +126,47 @@ func registerEthKeyCmd(cmd *cli.Cmd) {
 			return
 		}
 
-		clientCtx, err := chainclient.NewClientContext(*cosmosChainID, valAddress.String(), cosmosKeyring)
-		if err != nil {
-			log.WithError(err).Fatalln("failed to initialize cosmos client context")
-		}
-		clientCtx = clientCtx.WithNodeURI(*tendermintRPC)
+		var (
+			peggyBroadcastClient cosmos.PeggyBroadcastClient
+			customCosmosRPC      = *cosmosGRPC != "" && *tendermintRPC != ""
+		)
 
-		tmRPC, err := rpchttp.New(*tendermintRPC, "/websocket")
-		if err != nil {
-			log.WithError(err)
-		}
+		if customCosmosRPC {
+			net, err := cosmos.NewCustomRPCNetwork(
+				*cosmosChainID,
+				valAddress.String(),
+				*cosmosGRPC,
+				*cosmosGasPrices,
+				*tendermintRPC,
+				cosmosKeyring,
+				personalSignFn,
+			)
+
+			if err != nil {
+				log.Fatalln("failed to connect to Injective network")
+			}
 
-		clientCtx = clientCtx.WithClient(tmRPC)
-		daemonClient, err := chainclient.NewChainClient(clientCtx, *cosmosGRPC, common.OptionGasPrices(*cosmosGasPrices))
-		if err != nil {
-			log.WithError(err).WithFields(log.Fields{
-				"endpoint": *cosmosGRPC,
-			}).Fatalln("failed to connect to Cosmos daemon")
-		}
+			peggyBroadcastClient = net.PeggyBroadcastClient
+		} else {
+			net, err := cosmos.NewLoadBalancedNetwork(
+				*cosmosChainID,
+				valAddress.String(),
+				*cosmosGasPrices,
+				cosmosKeyring,
+				personalSignFn,
+			)
+
+			if err != nil {
+				log.Fatalln("failed to connect to Injective network")
+			}
 
-		log.Infoln("Waiting for injectived GRPC")
-		time.Sleep(1 * time.Second)
-
-		daemonWaitCtx, cancelWait := context.WithTimeout(context.Background(), time.Minute)
-		grpcConn := daemonClient.QueryClient()
-		waitForService(daemonWaitCtx, grpcConn)
-		peggyQuerier := types.NewQueryClient(grpcConn)
-		peggyBroadcaster := cosmos.NewPeggyBroadcastClient(
-			peggyQuerier,
-			daemonClient,
-			nil,
-			personalSignFn,
-		)
-		cancelWait()
+			peggyBroadcastClient = net.PeggyBroadcastClient
+		}
 
 		broadcastCtx, cancelFn := context.WithTimeout(context.Background(), 15*time.Second)
 		defer cancelFn()
 
-		if err = peggyBroadcaster.UpdatePeggyOrchestratorAddresses(broadcastCtx, ethKeyFromAddress, valAddress); err != nil {
+		if err = peggyBroadcastClient.UpdatePeggyOrchestratorAddresses(broadcastCtx, ethKeyFromAddress, valAddress); err != nil {
 			log.WithError(err).Errorln("failed to broadcast Tx")
 			time.Sleep(time.Second)
 			return
@@ -182,23 +176,3 @@ func registerEthKeyCmd(cmd *cli.Cmd) {
 			ethKeyFromAddress, valAddress.String())
 	}
 }
-
-// waitForService awaits an active ClientConn to a GRPC service.
-func waitForService(ctx context.Context, clientconn *grpc.ClientConn) {
-	for {
-		select {
-		case <-ctx.Done():
-			log.Fatalln("GRPC service wait timed out")
-		default:
-			state := clientconn.GetState()
-
-			if state != connectivity.Ready {
-				log.WithField("state", state.String()).Warningln("state of GRPC connection not ready")
-				time.Sleep(5 * time.Second)
-				continue
-			}
-
-			return
-		}
-	}
-}