[Grogu] fix by spec (#178)

* signaller can only add pending signal id

* separate build tx from broadcast tx

* add sync max block height

* only parallel on calculate gas

* add debug log in pending signal id set and remove

* fix cmd grogu

* fix atomic bug, calculate gas bug

* update buildSignedTx to log more error

* fix bug

* fix again

* fix format

* implement updater with chain subscribe

* signaller use the same now time

* fix lint, reduce updater log

* fix

* add start bothan script for local testing

* separate clients from context

* delete unnecessary new line

* cleaning for filtering and prepare submit price process

* cleanup

* fix format

* remove dup keyring from the submitter

* fix merge

* clean up broadcast function

* continue on some clients create error, extract as a function

* updater implement multi-clients

* fix from comments

* move function buildSignedTx to submitter

* update reference source query to support non-tx situation

* fix error message

* fix from comments

---------

Co-authored-by: Kitipong Sirirueangsakul <[email protected]>

Author: colmazia

cmd/grogu/cmd/keys.go

@@ -4,8 +4,8 @@ import (
 	"bufio"
 	"fmt"
 	"strings"
+	"sync/atomic"
 
-	"github.com/cometbft/cometbft/rpc/client/http"
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/client/flags"
 	"github.com/cosmos/cosmos-sdk/client/input"
@@ -168,7 +168,7 @@ func keysListCmd(ctx *grogu.Context) *cobra.Command {
 		Use:     "list",
 		Aliases: []string{"l"},
 		Short:   "List all the keys in the keychain",
-		Args:    cobra.ExactArgs(0),
+		Args:    cobra.NoArgs,
 		RunE:    createKeysListRunE(ctx),
 	}
 
@@ -184,26 +184,26 @@ func createKeysListRunE(ctx *grogu.Context) func(cmd *cobra.Command, args []stri
 	return func(cmd *cobra.Command, args []string) error {
 		bandConfig := band.MakeEncodingConfig()
 
-		nodesURIs := strings.Split(viper.GetString(flagNodes), ",")
-		clients := make([]client.Context, 0, len(nodesURIs))
-		for _, URI := range nodesURIs {
-			httpClient, err := http.New(URI, "/websocket")
-			if err != nil {
-				continue
-			}
-			cl := client.Context{
-				Client:            httpClient,
-				ChainID:           ctx.Config.ChainID,
-				Codec:             bandConfig.Marshaler,
-				InterfaceRegistry: bandConfig.InterfaceRegistry,
-				Keyring:           ctx.Keyring,
-				TxConfig:          bandConfig.TxConfig,
-				BroadcastMode:     flags.BroadcastSync,
-			}
-			clients = append(clients, cl)
+		clientCtx := client.Context{
+			ChainID:           ctx.Config.ChainID,
+			Codec:             bandConfig.Marshaler,
+			InterfaceRegistry: bandConfig.InterfaceRegistry,
+			Keyring:           ctx.Keyring,
+			TxConfig:          bandConfig.TxConfig,
+			BroadcastMode:     flags.BroadcastSync,
+		}
+
+		nodeURIs := strings.Split(viper.GetString(flagNodes), ",")
+		clients, stopClients, err := createClients(nodeURIs)
+		if err != nil {
+			return err
 		}
+		defer stopClients()
 
-		feedQuerier := querier.NewFeedQuerier(clients)
+		maxBlockHeight := new(atomic.Int64)
+		maxBlockHeight.Store(0)
+
+		feedQuerier := querier.NewFeedQuerier(clientCtx, clients, maxBlockHeight)
 
 		keys, err := ctx.Keyring.List()
 		if err != nil {
@@ -237,6 +237,7 @@ func createKeysListRunE(ctx *grogu.Context) func(cmd *cobra.Command, args []stri
 					s = ":x:"
 				}
 			}
+
 			_, err = emoji.Printf("%s%s => %s\n", s, key.Name, address.String())
 			if err != nil {
 				return err

cmd/grogu/cmd/run.go

@@ -5,12 +5,12 @@ import (
 	"os/signal"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 
 	bothan "github.com/bandprotocol/bothan/bothan-api/client/go-client"
 	"github.com/cometbft/cometbft/libs/log"
-	"github.com/cometbft/cometbft/rpc/client/http"
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/client/flags"
 	sdk "github.com/cosmos/cosmos-sdk/types"
@@ -22,6 +22,7 @@ import (
 	"github.com/bandprotocol/chain/v2/grogu/querier"
 	"github.com/bandprotocol/chain/v2/grogu/signaller"
 	"github.com/bandprotocol/chain/v2/grogu/submitter"
+	"github.com/bandprotocol/chain/v2/grogu/updater"
 	"github.com/bandprotocol/chain/v2/pkg/logger"
 	"github.com/bandprotocol/chain/v2/x/feeds/types"
 )
@@ -83,39 +84,38 @@ func createRunE(ctx *context.Context) func(cmd *cobra.Command, args []string) er
 		bandConfig := band.MakeEncodingConfig()
 		chainID := viper.GetString(flags.FlagChainID)
 
-		// Split Node URIs and create RPC clients
-		nodesURIs := strings.Split(viper.GetString(flagNodes), ",")
-		clients := make([]client.Context, 0, len(nodesURIs))
-
-		for _, URI := range nodesURIs {
-			httpClient, err := http.New(URI, "/websocket")
-			if err != nil {
-				return err
-			}
-			cl := client.Context{
-				Client:            httpClient,
-				ChainID:           chainID,
-				Codec:             bandConfig.Marshaler,
-				InterfaceRegistry: bandConfig.InterfaceRegistry,
-				Keyring:           ctx.Keyring,
-				TxConfig:          bandConfig.TxConfig,
-				BroadcastMode:     flags.BroadcastSync,
-			}
-			clients = append(clients, cl)
-		}
-
-		// set up Queriers
-		authQuerier := querier.NewAuthQuerier(clients)
-		feedQuerier := querier.NewFeedQuerier(clients)
-		txQuerier := querier.NewTxQuerier(clients)
-
 		// Initialize logger
 		allowLevel, err := log.AllowLevel(ctx.Config.LogLevel)
 		if err != nil {
 			return err
 		}
 		l := logger.New(allowLevel)
 
+		// Split Node URIs and create RPC clients
+		clientCtx := client.Context{
+			ChainID:           chainID,
+			Codec:             bandConfig.Marshaler,
+			InterfaceRegistry: bandConfig.InterfaceRegistry,
+			Keyring:           ctx.Keyring,
+			TxConfig:          bandConfig.TxConfig,
+			BroadcastMode:     flags.BroadcastSync,
+		}
+
+		nodeURIs := strings.Split(viper.GetString(flagNodes), ",")
+		clients, stopClients, err := createClients(nodeURIs)
+		if err != nil {
+			return err
+		}
+		defer stopClients()
+
+		// Set up Queriers
+		maxBlockHeight := new(atomic.Int64)
+		maxBlockHeight.Store(0)
+
+		authQuerier := querier.NewAuthQuerier(clientCtx, clients, maxBlockHeight)
+		feedQuerier := querier.NewFeedQuerier(clientCtx, clients, maxBlockHeight)
+		txQuerier := querier.NewTxQuerier(clientCtx, clients)
+
 		// Setup Bothan service
 		timeout, err := time.ParseDuration(ctx.Config.BothanTimeout)
 		if err != nil {
@@ -165,9 +165,9 @@ func createRunE(ctx *context.Context) func(cmd *cobra.Command, args []string) er
 
 		// Setup Submitter
 		submitterService, err := submitter.New(
+			clientCtx,
 			clients,
 			l,
-			ctx.Keyring,
 			submitSignalPriceCh,
 			authQuerier,
 			txQuerier,
@@ -182,19 +182,36 @@ func createRunE(ctx *context.Context) func(cmd *cobra.Command, args []string) er
 			return err
 		}
 
-		// Start all
-		go signallerService.Start()
-		go submitterService.Start()
+		// Setup Updater
+		maxCurrentFeedEventHeight := new(atomic.Int64)
+		maxCurrentFeedEventHeight.Store(0)
 
-		l.Info("Grogu has started")
+		maxUpdateRefSourceEventHeight := new(atomic.Int64)
+		maxUpdateRefSourceEventHeight.Store(0)
+
+		updaterService := updater.New(
+			feedQuerier,
+			clients,
+			l,
+			maxCurrentFeedEventHeight,
+			maxUpdateRefSourceEventHeight,
+		)
 
 		// Listen for termination signals for graceful shutdown
 		sigChan := make(chan os.Signal, 1)
 		signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
 
+		// Start all services
+		go updaterService.Start(sigChan)
+		go signallerService.Start()
+		go submitterService.Start()
+
+		l.Info("Grogu has started")
+
 		<-sigChan
 		l.Info("Received stop signal, shutting down")
 		l.Info("Grogu has stopped")
+
 		return nil
 	}
 }

cmd/grogu/cmd/utils.go

@@ -0,0 +1,38 @@
+package cmd
+
+import (
+	"fmt"
+
+	"github.com/cometbft/cometbft/rpc/client/http"
+)
+
+func createClients(nodeURIs []string) ([]*http.HTTP, func(), error) {
+	clients := make([]*http.HTTP, 0, len(nodeURIs))
+	for _, uri := range nodeURIs {
+		httpClient, err := http.New(uri, "/websocket")
+		if err != nil {
+			fmt.Printf("[Grogu] failed to create HTTP client with error: %v\n", err)
+			continue
+		}
+
+		if err = httpClient.Start(); err != nil {
+			fmt.Printf("[Grogu] failed to start HTTP client with error: %v\n", err)
+			continue
+		}
+
+		clients = append(clients, httpClient)
+	}
+
+	if len(clients) == 0 {
+		return nil, nil, fmt.Errorf("no clients are available")
+	}
+
+	// Function to stop all clients created so far
+	stopClients := func() {
+		for _, client := range clients {
+			_ = client.Stop()
+		}
+	}
+
+	return clients, stopClients, nil
+}

cmd/grogu/main.go

@@ -1,4 +1,4 @@
-package grogu
+package main
 
 import (
 	"fmt"

grogu/querier/auth.go

@@ -1,23 +1,32 @@
 package querier
 
 import (
+	"sync/atomic"
+
+	"github.com/cometbft/cometbft/rpc/client/http"
 	"github.com/cosmos/cosmos-sdk/client"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	auth "github.com/cosmos/cosmos-sdk/x/auth/types"
 )
 
 type AuthQuerier struct {
-	queryClients []auth.QueryClient
+	queryClients   []auth.QueryClient
+	maxBlockHeight *atomic.Int64
 }
 
-func NewAuthQuerier(clients []client.Context) *AuthQuerier {
+func NewAuthQuerier(
+	clientContext client.Context,
+	clients []*http.HTTP,
+	maxBlockHeight *atomic.Int64,
+) *AuthQuerier {
 	queryClients := make([]auth.QueryClient, 0, len(clients))
 	for _, cl := range clients {
-		queryClients = append(queryClients, auth.NewQueryClient(cl))
+		queryClients = append(queryClients, auth.NewQueryClient(clientContext.WithClient(cl)))
 	}
 
 	return &AuthQuerier{
 		queryClients,
+		maxBlockHeight,
 	}
 }
 
@@ -28,5 +37,5 @@ func (q *AuthQuerier) QueryAccount(address sdk.Address) (*auth.QueryAccountRespo
 	}
 
 	in := auth.QueryAccountRequest{Address: address.String()}
-	return getMaxBlockHeightResponse(fs, &in)
+	return getMaxBlockHeightResponse(fs, &in, q.maxBlockHeight)
 }

grogu/querier/feed.go

@@ -1,23 +1,27 @@
 package querier
 
 import (
+	"sync/atomic"
+
+	"github.com/cometbft/cometbft/rpc/client/http"
 	"github.com/cosmos/cosmos-sdk/client"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 
 	feeds "github.com/bandprotocol/chain/v2/x/feeds/types"
 )
 
 type FeedQuerier struct {
-	queryClients []feeds.QueryClient
+	queryClients   []feeds.QueryClient
+	maxBlockHeight *atomic.Int64
 }
 
-func NewFeedQuerier(clients []client.Context) *FeedQuerier {
+func NewFeedQuerier(clientContext client.Context, clients []*http.HTTP, maxBlockHeight *atomic.Int64) *FeedQuerier {
 	queryClients := make([]feeds.QueryClient, 0, len(clients))
 	for _, cl := range clients {
-		queryClients = append(queryClients, feeds.NewQueryClient(cl))
+		queryClients = append(queryClients, feeds.NewQueryClient(clientContext.WithClient(cl)))
 	}
 
-	return &FeedQuerier{queryClients}
+	return &FeedQuerier{queryClients, maxBlockHeight}
 }
 
 func (q *FeedQuerier) QueryValidValidator(valAddress sdk.ValAddress) (*feeds.QueryValidValidatorResponse, error) {
@@ -34,7 +38,7 @@ func (q *FeedQuerier) QueryValidValidator(valAddress sdk.ValAddress) (*feeds.Que
 		Validator: valAddress.String(),
 	}
 
-	return getMaxBlockHeightResponse(fs, &in)
+	return getMaxBlockHeightResponse(fs, &in, q.maxBlockHeight)
 }
 
 func (q *FeedQuerier) QueryIsFeeder(
@@ -50,7 +54,7 @@ func (q *FeedQuerier) QueryIsFeeder(
 		FeederAddress:    feeder.String(),
 		ValidatorAddress: validator.String(),
 	}
-	return getMaxBlockHeightResponse(fs, &in)
+	return getMaxBlockHeightResponse(fs, &in, q.maxBlockHeight)
 }
 
 func (q *FeedQuerier) QueryValidatorPrices(valAddress sdk.ValAddress) (*feeds.QueryValidatorPricesResponse, error) {
@@ -66,7 +70,7 @@ func (q *FeedQuerier) QueryValidatorPrices(valAddress sdk.ValAddress) (*feeds.Qu
 	in := feeds.QueryValidatorPricesRequest{
 		Validator: valAddress.String(),
 	}
-	return getMaxBlockHeightResponse(fs, &in)
+	return getMaxBlockHeightResponse(fs, &in, q.maxBlockHeight)
 }
 
 func (q *FeedQuerier) QueryParams() (*feeds.QueryParamsResponse, error) {
@@ -76,7 +80,7 @@ func (q *FeedQuerier) QueryParams() (*feeds.QueryParamsResponse, error) {
 	}
 
 	in := feeds.QueryParamsRequest{}
-	return getMaxBlockHeightResponse(fs, &in)
+	return getMaxBlockHeightResponse(fs, &in, q.maxBlockHeight)
 }
 
 func (q *FeedQuerier) QueryCurrentFeeds() (*feeds.QueryCurrentFeedsResponse, error) {
@@ -90,5 +94,19 @@ func (q *FeedQuerier) QueryCurrentFeeds() (*feeds.QueryCurrentFeedsResponse, err
 	}
 
 	in := feeds.QueryCurrentFeedsRequest{}
-	return getMaxBlockHeightResponse(fs, &in)
+	return getMaxBlockHeightResponse(fs, &in, q.maxBlockHeight)
+}
+
+func (q *FeedQuerier) QueryReferenceSourceConfig() (*feeds.QueryReferenceSourceConfigResponse, error) {
+	fs := make(
+		[]QueryFunction[feeds.QueryReferenceSourceConfigRequest, feeds.QueryReferenceSourceConfigResponse],
+		0,
+		len(q.queryClients),
+	)
+	for _, queryClient := range q.queryClients {
+		fs = append(fs, queryClient.ReferenceSourceConfig)
+	}
+
+	in := feeds.QueryReferenceSourceConfigRequest{}
+	return getMaxBlockHeightResponse(fs, &in, q.maxBlockHeight)
 }