#683 keychain sdk configure multiple nodes endpoints

cmd/wardenkms/health_check_response.go

@@ -0,0 +1,23 @@
+package main
+
+type HealthCheckResponse struct {
+	// The number of nodes that are online
+	Online uint `json:"online"`
+
+	// The total number of nodes
+	Total uint `json:"total"`
+
+	// The consensus threshold
+	Threshold uint8 `json:"threshold"`
+
+	// Node statuses
+	Nodes []NodeStatus `json:"nodes"`
+}
+
+type NodeStatus struct {
+	// The address of the node
+	Address string `json:"address"`
+
+	// The status of the node
+	Status string `json:"status"`
+}

cmd/wardenkms/wardenkms.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"encoding/binary"
+	"encoding/json"
 	"fmt"
 	"log"
 	"log/slog"
@@ -20,11 +21,10 @@ import (
 )
 
 type Config struct {
-	ChainID      string `env:"CHAIN_ID, default=warden_1337-1"`
-	GRPCURL      string `env:"GRPC_URL, default=localhost:9090"`
-	GRPCInsecure bool   `env:"GRPC_INSECURE, default=true"`
-	Mnemonic     string `env:"MNEMONIC, default=exclude try nephew main caught favorite tone degree lottery device tissue tent ugly mouse pelican gasp lava flush pen river noise remind balcony emerge"`
-	KeychainId   uint64 `env:"KEYCHAIN_ID, default=1"`
+	ChainID    string                `env:"CHAIN_ID, default=warden_1337-1"`
+	GRPCURLs   GrpcNodeConfigDecoder `env:"GRPC_URLS, default=[{\"GRPCUrl\":\"localhost:9090\",\"GRPCInsecure\":true}] "`
+	Mnemonic   string                `env:"MNEMONIC, default=exclude try nephew main caught favorite tone degree lottery device tissue tent ugly mouse pelican gasp lava flush pen river noise remind balcony emerge"`
+	KeychainId uint64                `env:"KEYCHAIN_ID, default=1"`
 
 	KeyringMnemonic string `env:"KEYRING_MNEMONIC, required"`
 	KeyringPassword string `env:"KEYRING_PASSWORD, required"`
@@ -38,6 +38,27 @@ type Config struct {
 	HttpAddr string `env:"HTTP_ADDR, default=:8080"`
 
 	LogLevel slog.Level `env:"LOG_LEVEL, default=debug"`
+
+	ConsensusNodeThreshold uint8 `env:"CONSENSUS_NODE_THRESHOLD, default=1"`
+}
+
+type GrpcNodeConfig struct {
+	GRPCUrl      string
+	GRPCInsecure bool
+}
+
+type GrpcNodeConfigDecoder []GrpcNodeConfig
+
+func (sd *GrpcNodeConfigDecoder) Decode(value string) error {
+	nodeConfigs := make([]GrpcNodeConfig, 0)
+
+	if err := json.Unmarshal([]byte(value), &nodeConfigs); err != nil {
+		return fmt.Errorf("invalid map json: %w", err)
+	}
+
+	*sd = nodeConfigs
+
+	return nil
 }
 
 func main() {
@@ -56,18 +77,26 @@ func main() {
 		return
 	}
 
+	grpcConfigs, err := mapGrpcConfig(cfg.GRPCURLs)
+	if err != nil {
+		logger.Error("failed to initialize grpc configs", "error", err)
+		return
+	}
+
 	app := keychain.NewApp(keychain.Config{
-		Logger:        logger,
-		ChainID:       cfg.ChainID,
-		GRPCURL:       cfg.GRPCURL,
-		GRPCInsecure:  cfg.GRPCInsecure,
-		Mnemonic:      cfg.Mnemonic,
-		KeychainID:    cfg.KeychainId,
-		GasLimit:      cfg.GasLimit,
-		BatchInterval: cfg.BatchInterval,
-		BatchSize:     cfg.BatchSize,
-		TxTimeout:     cfg.TxTimeout,
-		TxFees:        sdk.NewCoins(sdk.NewCoin("award", math.NewInt(cfg.TxFee))),
+		BasicConfig: keychain.BasicConfig{
+			Logger:        logger,
+			ChainID:       cfg.ChainID,
+			Mnemonic:      cfg.Mnemonic,
+			KeychainID:    cfg.KeychainId,
+			GasLimit:      cfg.GasLimit,
+			BatchInterval: cfg.BatchInterval,
+			BatchSize:     cfg.BatchSize,
+			TxTimeout:     cfg.TxTimeout,
+			TxFees:        sdk.NewCoins(sdk.NewCoin("award", math.NewInt(cfg.TxFee))),
+		},
+		GRPCConfigs:            grpcConfigs,
+		ConsensusNodeThreshold: cfg.ConsensusNodeThreshold,
 	})
 
 	app.SetKeyRequestHandler(func(w keychain.KeyResponseWriter, req *keychain.KeyRequest) {
@@ -120,11 +149,41 @@ func main() {
 	if cfg.HttpAddr != "" {
 		logger.Info("starting HTTP server", "addr", cfg.HttpAddr)
 		http.HandleFunc("/healthcheck", func(w http.ResponseWriter, r *http.Request) {
-			if app.ConnectionState() == connectivity.Ready {
+			connectionStates := app.ConnectionState()
+
+			readyConnectionsCount := uint(0)
+			nodes := make([]NodeStatus, 0, len(connectionStates))
+
+			for url, state := range connectionStates {
+				if state == connectivity.Ready {
+					readyConnectionsCount += 1
+				}
+
+				nodes = append(nodes, NodeStatus{
+					Address: url,
+					Status:  state.String(),
+				})
+			}
+
+			bytes, err := json.Marshal(HealthCheckResponse{
+				Online:    readyConnectionsCount,
+				Total:     uint(len(connectionStates)),
+				Nodes:     nodes,
+				Threshold: cfg.ConsensusNodeThreshold,
+			})
+
+			if err != nil {
+				w.WriteHeader(http.StatusInternalServerError)
+				return
+			}
+
+			if readyConnectionsCount >= uint(cfg.ConsensusNodeThreshold) {
 				w.WriteHeader(http.StatusOK)
 			} else {
 				w.WriteHeader(http.StatusServiceUnavailable)
 			}
+
+			_, _ = w.Write(bytes)
 		})
 		go func() { _ = http.ListenAndServe(cfg.HttpAddr, nil) }()
 	}
@@ -144,3 +203,24 @@ func bigEndianBytesFromUint32(n uint64) ([4]byte, error) {
 	binary.BigEndian.PutUint32(b, uint32(n))
 	return [4]byte(b), nil
 }
+
+func mapGrpcConfig(value GrpcNodeConfigDecoder) ([]keychain.GrpcNodeConfig, error) {
+	var nodesLength = len(value)
+	if nodesLength == 0 {
+		return nil, fmt.Errorf("GRPCUrls must be specified")
+	}
+
+	result := make([]keychain.GrpcNodeConfig, 0, nodesLength)
+	for _, item := range value {
+		if item.GRPCUrl == "" {
+			return nil, fmt.Errorf("GRPCUrl must be specified")
+		}
+
+		result = append(result, keychain.GrpcNodeConfig{
+			GRPCInsecure: item.GRPCInsecure,
+			GRPCURL:      item.GRPCUrl,
+		})
+	}
+
+	return result, nil
+}

docker-compose.yml

@@ -15,7 +15,7 @@ services:
       WARDEND_GRPC_ADDRESS: 0.0.0.0:9090
       WARDEND_RPC_LADDR: tcp://0.0.0.0:26657
       WARDEND_RPC_CORS_ALLOWED_ORIGINS: "*"
-      WARDEND_MINIMUM_GAS_PRICES: "0uward"
+      WARDEND_MINIMUM_GAS_PRICES: "0award"
 
   faucet:
     build:

keychain-sdk/config.go

@@ -7,23 +7,14 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
-// Config is the configuration for the Keychain.
-type Config struct {
+type BasicConfig struct {
 	// Logger is the logger to use for the Keychain.
 	// If nil, no logging will be done.
 	Logger *slog.Logger
 
 	// ChainID is the chain ID of the chain to connect to.
 	ChainID string
 
-	// GRPCURL is the URL of the gRPC server to connect to.
-	// e.g. "localhost:9090"
-	GRPCURL string
-
-	// GRPCInsecure determines whether to allow an insecure connection to the
-	// gRPC server.
-	GRPCInsecure bool
-
 	// KeychainID is the ID of the keychain this instance will fetch requests
 	// for.
 	KeychainID uint64
@@ -58,3 +49,25 @@ type Config struct {
 	// failed (but the blockchain might still include in a block later).
 	TxTimeout time.Duration
 }
+
+// Config is the configuration for the Keychain.
+type Config struct {
+	BasicConfig
+
+	// ConsensusNodeThreshold represents the number of nodes required to execute a pending key/sign request.
+	ConsensusNodeThreshold uint8
+
+	// GRPCURLs is the list of URLs of the gRPC server to connect to.
+	// e.g. "localhost:9090"
+	GRPCConfigs []GrpcNodeConfig
+}
+
+type GrpcNodeConfig struct {
+	// GRPCInsecure determines whether to allow an insecure connection to the
+	// gRPC server.
+	GRPCInsecure bool
+
+	// GRPCURL is the URL of the gRPC server to connect to.
+	// e.g. "localhost:9090"
+	GRPCURL string
+}

keychain-sdk/example_keychain_test.go

@@ -18,21 +18,24 @@ func Main() {
 	}))
 
 	app := keychain.NewApp(keychain.Config{
-		Logger: logger, // not required, but recommended
+		BasicConfig: keychain.BasicConfig{
+			Logger: logger, // not required, but recommended
 
-		// setup the connection to the Warden Protocol node
-		ChainID:      "warden",
-		GRPCURL:      "localhost:9090",
-		GRPCInsecure: true,
+			// setup the connection to the Warden Protocol node
+			ChainID: "warden",
 
-		// setup the account used to write txs
-		KeychainID: 1,
-		Mnemonic:   "virus boat radio apple pilot ask vault exhaust again state doll stereo slide exhibit scissors miss attack boat budget egg bird mask more trick",
+			// setup the account used to write txs
+			KeychainID: 1,
+			Mnemonic:   "virus boat radio apple pilot ask vault exhaust again state doll stereo slide exhibit scissors miss attack boat budget egg bird mask more trick",
 
-		// setup throughput for batching responses
-		GasLimit:      400000,
-		BatchInterval: 8 * time.Second,
-		BatchSize:     10,
+			// setup throughput for batching responses
+			GasLimit:      400000,
+			BatchInterval: 8 * time.Second,
+			BatchSize:     10,
+		},
+
+		GRPCConfigs:            []keychain.GrpcNodeConfig{{GRPCURL: "localhost:9090", GRPCInsecure: false}},
+		ConsensusNodeThreshold: 1,
 	})
 
 	app.SetKeyRequestHandler(func(w keychain.KeyResponseWriter, req *keychain.KeyRequest) {

keychain-sdk/internal/tracker/tracker.go

@@ -1,35 +1,72 @@
 package tracker
 
 import (
+	"fmt"
 	"sync"
 )
 
+type Action int
+
+const (
+	ActionSkip Action = iota
+	ActionProcess
+)
+
+type stringSet map[string]struct{}
+
+// add safely adds a string to the set
+func (s stringSet) add(str string) bool {
+	if _, exists := s[str]; exists {
+		return false
+	}
+	s[str] = struct{}{}
+	return true
+}
+
 type T struct {
-	rw       sync.RWMutex
-	ingested map[uint64]struct{}
+	threshold uint8
+	rw        sync.RWMutex
+	ingested  map[uint64]stringSet
 }
 
-func New() *T {
+func New(threshold uint8) *T {
 	return &T{
-		ingested: make(map[uint64]struct{}),
+		threshold: threshold,
+		ingested:  make(map[uint64]stringSet),
 	}
 }
 
-func (t *T) IsNew(id uint64) bool {
-	t.rw.RLock()
-	defer t.rw.RUnlock()
-	_, ok := t.ingested[id]
-	return !ok
+func (t *T) ingestTracker(id uint64) stringSet {
+	value, ok := t.ingested[id]
+	if !ok {
+		t.ingested[id] = make(stringSet)
+
+		return t.ingested[id]
+	}
+
+	return value
 }
 
-func (t *T) Ingested(id uint64) {
+func (t *T) Ingest(id uint64, ingesterId string) (Action, error) {
 	t.rw.Lock()
 	defer t.rw.Unlock()
-	t.ingested[id] = struct{}{}
+
+	value := t.ingestTracker(id)
+
+	if !value.add(ingesterId) {
+		return ActionSkip, fmt.Errorf("already ingested")
+	}
+
+	if uint64(len(value)) == uint64(t.threshold) {
+		return ActionProcess, nil
+	}
+
+	return ActionSkip, nil
 }
 
 func (t *T) Done(id uint64) {
 	t.rw.Lock()
 	defer t.rw.Unlock()
+
 	delete(t.ingested, id)
 }