perf: gRPC caching

decentralized-api/apiconfig/config.go

@@ -74,14 +74,17 @@ type ApiConfig struct {
 }
 
 type ChainNodeConfig struct {
-	Url              string `koanf:"url" json:"url"`
-	IsGenesis        bool   `koanf:"is_genesis" json:"is_genesis"`
-	SeedApiUrl       string `koanf:"seed_api_url" json:"seed_api_url"`
-	AccountPublicKey string `koanf:"account_public_key" json:"account_public_key"`
-	SignerKeyName    string `koanf:"signer_key_name" json:"signer_key_name"`
-	KeyringBackend   string `koanf:"keyring_backend" json:"keyring_backend"`
-	KeyringDir       string `koanf:"keyring_dir" json:"keyring_dir"`
-	KeyringPassword  string `json:"-"`
+	Url                  string `koanf:"url" json:"url"`
+	IsGenesis            bool   `koanf:"is_genesis" json:"is_genesis"`
+	SeedApiUrl           string `koanf:"seed_api_url" json:"seed_api_url"`
+	AccountPublicKey     string `koanf:"account_public_key" json:"account_public_key"`
+	SignerKeyName        string `koanf:"signer_key_name" json:"signer_key_name"`
+	KeyringBackend       string `koanf:"keyring_backend" json:"keyring_backend"`
+	KeyringDir           string `koanf:"keyring_dir" json:"keyring_dir"`
+	KeyringPassword      string `json:"-"`
+	QueryCacheEnabled    bool   `koanf:"query_cache_enabled" json:"query_cache_enabled"`
+	QueryCacheMaxEntries int    `koanf:"query_cache_max_entries" json:"query_cache_max_entries"`
+	QueryCacheMaxBytes   int64  `koanf:"query_cache_max_bytes" json:"query_cache_max_bytes"`
 	// MinGasPriceNgonka is the gas price in ngonka used for transaction fee calculation.
 	//
 	// When set to a non-zero value, the DAPI uses it as-is (this is an override

decentralized-api/apiconfig/config_manager.go

@@ -520,6 +520,21 @@ func readConfig(provider koanf.Provider) (Config, error) {
 	if err != nil {
 		log.Fatalf("error loading env: %v", err)
 	}
+	if !k.Exists("chain_node.query_cache_enabled") {
+		if err := k.Set("chain_node.query_cache_enabled", true); err != nil {
+			log.Fatalf("error setting default chain_node.query_cache_enabled: %v", err)
+		}
+	}
+	if !k.Exists("chain_node.query_cache_max_entries") {
+		if err := k.Set("chain_node.query_cache_max_entries", 200000); err != nil {
+			log.Fatalf("error setting default chain_node.query_cache_max_entries: %v", err)
+		}
+	}
+	if !k.Exists("chain_node.query_cache_max_bytes") {
+		if err := k.Set("chain_node.query_cache_max_bytes", int64(1<<30)); err != nil {
+			log.Fatalf("error setting default chain_node.query_cache_max_bytes: %v", err)
+		}
+	}
 	var config Config
 	err = k.Unmarshal("", &config)
 	if err != nil {

decentralized-api/cosmosclient/cosmosclient.go

@@ -35,6 +35,7 @@ import (
 	blstypes "github.com/productscience/inference/x/bls/types"
 	inferencetypes "github.com/productscience/inference/x/inference/types"
 	restrictionstypes "github.com/productscience/inference/x/restrictions/types"
+	"google.golang.org/grpc"
 )
 
 type InferenceCosmosClient struct {
@@ -44,6 +45,7 @@ type InferenceCosmosClient struct {
 	manager         tx_manager.TxManager
 	batchConsumer   *tx_manager.BatchConsumer
 	batchingEnabled bool
+	queryCache      *QueryCache
 }
 
 func NewInferenceCosmosClientWithRetry(
@@ -219,11 +221,18 @@ func NewInferenceCosmosClient(ctx context.Context, addressPrefix string, config
 		return nil, err
 	}
 
+	var queryCache *QueryCache
+	if nodeConfig.QueryCacheEnabled {
+		queryCache = NewQueryCacheWithLimits(nodeConfig.QueryCacheMaxEntries, nodeConfig.QueryCacheMaxBytes)
+		log.Printf("Query cache enabled (max_entries=%d, max_bytes=%d)", nodeConfig.QueryCacheMaxEntries, nodeConfig.QueryCacheMaxBytes)
+	}
+
 	client := &InferenceCosmosClient{
 		ctx:        ctx,
 		Address:    accAddress,
 		apiAccount: apiAccount,
 		manager:    mn,
+		queryCache: queryCache,
 	}
 
 	batchingCfg := config.GetTxBatchingConfig()
@@ -294,12 +303,44 @@ type CosmosMessageClient interface {
 	NewRestrictionsQueryClient() restrictionstypes.QueryClient
 	GetAddress() string
 	GetApiAccount() apiconfig.ApiAccount
+	SetQueryCacheHeightHint(height int64)
+	GetQueryCacheStats() *QueryCacheStats
+	ResetQueryCacheStats() bool
 }
 
 func (icc *InferenceCosmosClient) GetApiAccount() apiconfig.ApiAccount {
 	return icc.manager.GetApiAccount()
 }
 
+func (icc *InferenceCosmosClient) SetQueryCacheHeightHint(height int64) {
+	if icc.queryCache != nil {
+		icc.queryCache.SetHeightHint(height)
+	}
+}
+
+func (icc *InferenceCosmosClient) GetQueryCacheStats() *QueryCacheStats {
+	if icc.queryCache == nil {
+		return nil
+	}
+	stats := icc.queryCache.SnapshotStats()
+	return &stats
+}
+
+func (icc *InferenceCosmosClient) ResetQueryCacheStats() bool {
+	if icc.queryCache == nil {
+		return false
+	}
+	icc.queryCache.ResetStats()
+	return true
+}
+
+func (icc *InferenceCosmosClient) cachedConn() grpc.ClientConnInterface {
+	if icc.queryCache != nil {
+		return &CachingConn{inner: icc.manager.GetClientContext(), cache: icc.queryCache}
+	}
+	return icc.manager.GetClientContext()
+}
+
 func (icc *InferenceCosmosClient) GetClientContext() sdkclient.Context {
 	return icc.manager.GetClientContext()
 }
@@ -518,15 +559,15 @@ func (icc *InferenceCosmosClient) GetPartialUpgrades() (*inferencetypes.QueryAll
 }
 
 func (icc *InferenceCosmosClient) NewUpgradeQueryClient() upgradetypes.QueryClient {
-	return upgradetypes.NewQueryClient(icc.manager.GetClientContext())
+	return upgradetypes.NewQueryClient(icc.cachedConn())
 }
 
 func (icc *InferenceCosmosClient) NewInferenceQueryClient() inferencetypes.QueryClient {
-	return inferencetypes.NewQueryClient(icc.manager.GetClientContext())
+	return inferencetypes.NewQueryClient(icc.cachedConn())
 }
 
 func (icc *InferenceCosmosClient) NewCometQueryClient() cmtservice.ServiceClient {
-	return cmtservice.NewServiceClient(icc.manager.GetClientContext())
+	return cmtservice.NewServiceClient(icc.cachedConn())
 }
 
 func (icc *InferenceCosmosClient) SendTransactionSyncNoRetry(transaction proto.Message, dstMsg proto.Message) error {
@@ -583,9 +624,9 @@ func (icc *InferenceCosmosClient) SubmitPartialSignature(requestId []byte, slotI
 }
 
 func (icc *InferenceCosmosClient) NewBLSQueryClient() blstypes.QueryClient {
-	return blstypes.NewQueryClient(icc.manager.GetClientContext())
+	return blstypes.NewQueryClient(icc.cachedConn())
 }
 
 func (icc *InferenceCosmosClient) NewRestrictionsQueryClient() restrictionstypes.QueryClient {
-	return restrictionstypes.NewQueryClient(icc.manager.GetClientContext())
+	return restrictionstypes.NewQueryClient(icc.cachedConn())
 }

decentralized-api/cosmosclient/mock_cosmos_message_client.go

@@ -232,3 +232,13 @@ func (m *MockCosmosMessageClient) NewRestrictionsQueryClient() restrictionstypes
 	args := m.Called()
 	return args.Get(0).(restrictionstypes.QueryClient)
 }
+
+func (m *MockCosmosMessageClient) SetQueryCacheHeightHint(_ int64) {}
+
+func (m *MockCosmosMessageClient) GetQueryCacheStats() *QueryCacheStats {
+	return nil
+}
+
+func (m *MockCosmosMessageClient) ResetQueryCacheStats() bool {
+	return false
+}