Add operator state cache to IndexedChainState

core/indexer/state.go

@@ -2,16 +2,21 @@ package indexer
 
 import (
 	"context"
+	"encoding/binary"
 	"errors"
 
 	"github.com/Layr-Labs/eigenda/core"
 	"github.com/Layr-Labs/eigenda/indexer"
+	lru "github.com/hashicorp/golang-lru/v2"
 )
 
+const operatorStateCacheSize = 32
+
 type IndexedChainState struct {
 	core.ChainState
 
-	Indexer indexer.Indexer
+	Indexer            indexer.Indexer
+	operatorStateCache *lru.Cache[string, *core.IndexedOperatorState]
 }
 
 var _ core.IndexedChainState = (*IndexedChainState)(nil)
@@ -20,10 +25,15 @@ func NewIndexedChainState(
 	chainState core.ChainState,
 	indexer indexer.Indexer,
 ) (*IndexedChainState, error) {
+	operatorStateCache, err := lru.New[string, *core.IndexedOperatorState](operatorStateCacheSize)
+	if err != nil {
+		return nil, err
+	}
 
 	return &IndexedChainState{
-		ChainState: chainState,
-		Indexer:    indexer,
+		ChainState:         chainState,
+		Indexer:            indexer,
+		operatorStateCache: operatorStateCache,
 	}, nil
 }
 
@@ -32,6 +42,11 @@ func (ics *IndexedChainState) Start(ctx context.Context) error {
 }
 
 func (ics *IndexedChainState) GetIndexedOperatorState(ctx context.Context, blockNumber uint, quorums []core.QuorumID) (*core.IndexedOperatorState, error) {
+	// Check if the indexed operator state has been cached
+	cacheKey := computeCacheKey(blockNumber, quorums)
+	if val, ok := ics.operatorStateCache.Get(cacheKey); ok {
+		return val, nil
+	}
 
 	pubkeys, sockets, err := ics.getObjects(blockNumber)
 	if err != nil {
@@ -73,11 +88,11 @@ func (ics *IndexedChainState) GetIndexedOperatorState(ctx context.Context, block
 		AggKeys:          aggKeys,
 	}
 
+	ics.operatorStateCache.Add(cacheKey, state)
 	return state, nil
 }
 
 func (ics *IndexedChainState) GetIndexedOperators(ctx context.Context, blockNumber uint) (map[core.OperatorID]*core.IndexedOperatorInfo, error) {
-
 	pubkeys, sockets, err := ics.getObjects(blockNumber)
 	if err != nil {
 		return nil, err
@@ -138,3 +153,10 @@ func (ics *IndexedChainState) getObjects(blockNumber uint) (*OperatorPubKeys, Op
 	return pubkeys, sockets, nil
 
 }
+
+func computeCacheKey(blockNumber uint, quorumIDs []uint8) string {
+	bytes := make([]byte, 8+len(quorumIDs))
+	binary.LittleEndian.PutUint64(bytes, uint64(blockNumber))
+	copy(bytes[8:], quorumIDs)
+	return string(bytes)
+}

core/thegraph/state.go

@@ -2,6 +2,7 @@ package thegraph
 
 import (
 	"context"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"math"
@@ -10,15 +11,17 @@ import (
 	"github.com/Layr-Labs/eigenda/core"
 	"github.com/Layr-Labs/eigensdk-go/logging"
 	"github.com/consensys/gnark-crypto/ecc/bn254"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/shurcooL/graphql"
 )
 
 const (
-	defaultInterval      = time.Second
-	maxInterval          = 5 * time.Minute
-	maxEntriesPerQuery   = 1000
-	startRetriesInterval = time.Second * 5
-	startMaxRetries      = 6
+	defaultInterval        = time.Second
+	maxInterval            = 5 * time.Minute
+	maxEntriesPerQuery     = 1000
+	startRetriesInterval   = time.Second * 5
+	startMaxRetries        = 6
+	operatorStateCacheSize = 32
 )
 
 type (
@@ -79,14 +82,14 @@ type (
 		core.ChainState
 		querier GraphQLQuerier
 
-		logger logging.Logger
+		logger             logging.Logger
+		operatorStateCache *lru.Cache[string, *core.IndexedOperatorState]
 	}
 )
 
 var _ IndexedChainState = (*indexedChainState)(nil)
 
-func MakeIndexedChainState(config Config, cs core.ChainState, logger logging.Logger) *indexedChainState {
-
+func MakeIndexedChainState(config Config, cs core.ChainState, logger logging.Logger) (*indexedChainState, error) {
 	logger.Info("Using graph node")
 	querier := graphql.NewClient(config.Endpoint, nil)
 
@@ -96,12 +99,18 @@ func MakeIndexedChainState(config Config, cs core.ChainState, logger logging.Log
 	return NewIndexedChainState(cs, retryQuerier, logger)
 }
 
-func NewIndexedChainState(cs core.ChainState, querier GraphQLQuerier, logger logging.Logger) *indexedChainState {
-	return &indexedChainState{
-		ChainState: cs,
-		querier:    querier,
-		logger:     logger.With("component", "IndexedChainState"),
+func NewIndexedChainState(cs core.ChainState, querier GraphQLQuerier, logger logging.Logger) (*indexedChainState, error) {
+	operatorStateCache, err := lru.New[string, *core.IndexedOperatorState](operatorStateCacheSize)
+	if err != nil {
+		return nil, err
 	}
+
+	return &indexedChainState{
+		ChainState:         cs,
+		querier:            querier,
+		logger:             logger.With("component", "IndexedChainState"),
+		operatorStateCache: operatorStateCache,
+	}, nil
 }
 
 func (ics *indexedChainState) Start(ctx context.Context) error {
@@ -124,6 +133,12 @@ func (ics *indexedChainState) Start(ctx context.Context) error {
 }
 
 func (ics *indexedChainState) GetIndexedOperatorState(ctx context.Context, blockNumber uint, quorums []core.QuorumID) (*core.IndexedOperatorState, error) {
+	// Check if the indexed operator state has been cached
+	cacheKey := computeCacheKey(blockNumber, quorums)
+	if val, ok := ics.operatorStateCache.Get(cacheKey); ok {
+		return val, nil
+	}
+
 	operatorState, err := ics.ChainState.GetOperatorState(ctx, blockNumber, quorums)
 	if err != nil {
 		return nil, err
@@ -173,6 +188,8 @@ func (ics *indexedChainState) GetIndexedOperatorState(ctx context.Context, block
 		IndexedOperators: indexedOperators,
 		AggKeys:          aggKeys,
 	}
+
+	ics.operatorStateCache.Add(cacheKey, state)
 	return state, nil
 }
 
@@ -365,3 +382,10 @@ func convertIndexedOperatorInfoGqlToIndexedOperatorInfo(operator *IndexedOperato
 		Socket:   string(operator.SocketUpdates[0].Socket),
 	}, nil
 }
+
+func computeCacheKey(blockNumber uint, quorumIDs []uint8) string {
+	bytes := make([]byte, 8+len(quorumIDs))
+	binary.LittleEndian.PutUint64(bytes, uint64(blockNumber))
+	copy(bytes[8:], quorumIDs)
+	return string(bytes)
+}

core/thegraph/state_integration_test.go

@@ -86,7 +86,8 @@ func TestIndexerIntegration(t *testing.T) {
 	tx, err := eth.NewWriter(logger, client, testConfig.EigenDA.OperatorStateRetreiver, testConfig.EigenDA.ServiceManager)
 	assert.NoError(t, err)
 
-	cs := thegraph.NewIndexedChainState(eth.NewChainState(tx, client), graphql.NewClient(graphUrl, nil), logger)
+	cs, err := thegraph.NewIndexedChainState(eth.NewChainState(tx, client), graphql.NewClient(graphUrl, nil), logger)
+	assert.NoError(t, err)
 	time.Sleep(5 * time.Second)
 
 	err = cs.Start(context.Background())

core/thegraph/state_test.go

@@ -80,7 +80,9 @@ func TestIndexedChainState_GetIndexedOperatorState(t *testing.T) {
 		}
 	}
 
-	cs := thegraph.NewIndexedChainState(chainState, querier, logger)
+	cs, err := thegraph.NewIndexedChainState(chainState, querier, logger)
+	assert.NoError(t, err)
+
 	err = cs.Start(context.Background())
 	assert.NoError(t, err)
 
@@ -148,7 +150,9 @@ func TestIndexedChainState_GetIndexedOperatorStateMissingOperator(t *testing.T)
 		}
 	}
 
-	cs := thegraph.NewIndexedChainState(chainState, querier, logger)
+	cs, err := thegraph.NewIndexedChainState(chainState, querier, logger)
+	assert.NoError(t, err)
+
 	err = cs.Start(context.Background())
 	assert.NoError(t, err)
 
@@ -229,7 +233,9 @@ func TestIndexedChainState_GetIndexedOperatorStateExtraOperator(t *testing.T) {
 		}
 	}
 
-	cs := thegraph.NewIndexedChainState(chainState, querier, logger)
+	cs, err := thegraph.NewIndexedChainState(chainState, querier, logger)
+	assert.NoError(t, err)
+
 	err = cs.Start(context.Background())
 	assert.NoError(t, err)
 
@@ -283,7 +289,9 @@ func TestIndexedChainState_GetIndexedOperatorInfoByOperatorId(t *testing.T) {
 		}
 	}
 
-	cs := thegraph.NewIndexedChainState(chainState, querier, logger)
+	cs, err := thegraph.NewIndexedChainState(chainState, querier, logger)
+	assert.NoError(t, err)
+
 	err = cs.Start(context.Background())
 	assert.NoError(t, err)
 

disperser/batcher/encoding_streamer.go

@@ -2,7 +2,6 @@
 
 import (
 	"context"
-	"encoding/binary"
 	"errors"
 	"fmt"
 	"strings"
@@ -14,14 +13,13 @@
 	"github.com/Layr-Labs/eigenda/disperser"
 	"github.com/Layr-Labs/eigenda/encoding"
 	"github.com/Layr-Labs/eigensdk-go/logging"
-	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/wealdtech/go-merkletree/v2"
 	grpc_metadata "google.golang.org/grpc/metadata"
 )
 
 const encodingInterval = 2 * time.Second
 
 const operatorStateCacheSize = 32
 
 var errNoEncodedResults = errors.New("no encoded results")
 
@@ -81,8 +79,6 @@
 
 	// Used to keep track of the last evaluated key for fetching metadatas
 	exclusiveStartKey *disperser.BlobStoreExclusiveStartKey
-
-	operatorStateCache *lru.Cache[string, *core.IndexedOperatorState]
 }
 
 type batch struct {
@@ -116,10 +112,7 @@
 	if config.EncodingQueueLimit <= 0 {
 		return nil, errors.New("EncodingQueueLimit should be greater than 0")
 	}
-	operatorStateCache, err := lru.New[string, *core.IndexedOperatorState](operatorStateCacheSize)
-	if err != nil {
-		return nil, err
-	}
+
 	return &EncodingStreamer{
 		StreamerConfig:         config,
 		EncodedBlobstore:       newEncodedBlobStore(logger),
@@ -135,7 +128,6 @@
 		batcherMetrics:         batcherMetrics,
 		logger:                 logger.With("component", "EncodingStreamer"),
 		exclusiveStartKey:      nil,
-		operatorStateCache:     operatorStateCache,
 	}, nil
 }
 
@@ -680,16 +672,12 @@
 		i++
 	}
 
-	cacheKey := computeCacheKey(blockNumber, quorumIds)
-	if val, ok := e.operatorStateCache.Get(cacheKey); ok {
-		return val, nil
-	}
 	// GetIndexedOperatorState should return state for valid quorums only
 	state, err := e.chainState.GetIndexedOperatorState(ctx, blockNumber, quorumIds)
 	if err != nil {
 		return nil, fmt.Errorf("error getting operator state at block number %d: %w", blockNumber, err)
 	}
-	e.operatorStateCache.Add(cacheKey, state)
+
 	return state, nil
 }
 
@@ -715,10 +703,3 @@
 	}
 	return validMetadata
 }
-
-func computeCacheKey(blockNumber uint, quorumIDs []uint8) string {
-	bytes := make([]byte, 8+len(quorumIDs))
-	binary.LittleEndian.PutUint64(bytes, uint64(blockNumber))
-	copy(bytes[8:], quorumIDs)
-	return string(bytes)
-}

disperser/cmd/batcher/main.go

@@ -202,7 +202,10 @@ func RunBatcher(ctx *cli.Context) error {
 		logger.Info("Using graph node")
 
 		logger.Info("Connecting to subgraph", "url", config.ChainStateConfig.Endpoint)
-		ics = thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
+		ics, err = thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
+		if err != nil {
+			return err
+		}
 	} else {
 		logger.Info("Using built-in indexer")
 

disperser/cmd/controller/main.go

@@ -3,9 +3,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/collectors"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"log"
 	"net/http"
 	"os"
@@ -25,6 +22,9 @@ import (
 	gethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/gammazero/workerpool"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/urfave/cli"
 )
 
@@ -126,7 +126,10 @@ func RunController(ctx *cli.Context) error {
 		logger.Info("Using graph node")
 
 		logger.Info("Connecting to subgraph", "url", config.ChainStateConfig.Endpoint)
-		ics = thegraph.MakeIndexedChainState(config.ChainStateConfig, chainState, logger)
+		ics, err = thegraph.MakeIndexedChainState(config.ChainStateConfig, chainState, logger)
+		if err != nil {
+			return err
+		}
 	} else {
 		logger.Info("Using built-in indexer")
 		rpcClient, err := rpc.Dial(config.EthClientConfig.RPCURLs[0])

disperser/cmd/dataapi/main.go

@@ -90,14 +90,18 @@ func RunDataApi(ctx *cli.Context) error {
 		return err
 	}
 
+	chainState := coreeth.NewChainState(tx, client)
+	indexedChainState, err := thegraph.MakeIndexedChainState(config.ChainStateConfig, chainState, logger)
+	if err != nil {
+		return err
+	}
+
 	var (
 		promClient        = dataapi.NewPrometheusClient(promApi, config.PrometheusConfig.Cluster)
 		blobMetadataStore = blobstore.NewBlobMetadataStore(dynamoClient, logger, config.BlobstoreConfig.TableName, 0)
 		sharedStorage     = blobstore.NewSharedStorage(config.BlobstoreConfig.BucketName, s3Client, blobMetadataStore, logger)
 		subgraphApi       = subgraph.NewApi(config.SubgraphApiBatchMetadataAddr, config.SubgraphApiOperatorStateAddr)
 		subgraphClient    = dataapi.NewSubgraphClient(subgraphApi, logger)
-		chainState        = coreeth.NewChainState(tx, client)
-		indexedChainState = thegraph.MakeIndexedChainState(config.ChainStateConfig, chainState, logger)
 		metrics           = dataapi.NewMetrics(blobMetadataStore, config.MetricsConfig.HTTPPort, logger)
 		server            = dataapi.NewServer(
 			dataapi.Config{

operators/churner/cmd/main.go

@@ -84,7 +84,10 @@ func run(ctx *cli.Context) error {
 	logger.Info("Using graph node")
 
 	logger.Info("Connecting to subgraph", "url", config.ChainStateConfig.Endpoint)
-	indexer := thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
+	indexer, err := thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
+	if err != nil {
+		log.Fatalln("could not create indexer", err)
+	}
 
 	metrics := churner.NewMetrics(config.MetricsConfig.HTTPPort, logger)
 

relay/cmd/main.go

@@ -80,7 +80,10 @@ func RunRelay(ctx *cli.Context) error {
 	}
 
 	cs := coreeth.NewChainState(tx, client)
-	ics := thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
+	ics, err := thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
+	if err != nil {
+		return fmt.Errorf("failed to create indexed chain state: %w", err)
+	}
 
 	server, err := relay.NewServer(
 		context.Background(),