Update flags

core/thegraph/config.go

@@ -0,0 +1,53 @@
+package thegraph
+
+import (
+	"time"
+
+	"github.com/Layr-Labs/eigenda/common"
+	"github.com/urfave/cli"
+)
+
+const (
+	EndpointFlagName     = "thegraph.endpoint"
+	PullIntervalFlagName = "thegraph.pull_interval"
+	MaxRetriesFlagName   = "thegraph.max_retries"
+)
+
+type Config struct {
+	Endpoint     string        // The Graph endpoint
+	PullInterval time.Duration // The interval to pull data from The Graph
+	MaxRetries   int           // The maximum number of retries to pull data from The Graph
+}
+
+func CLIFlags(envPrefix string) []cli.Flag {
+	return []cli.Flag{
+		cli.StringFlag{
+			Name:     EndpointFlagName,
+			Usage:    "The Graph endpoint",
+			Required: true,
+			EnvVar:   common.PrefixEnvVar(envPrefix, "GRAPH_URL"),
+		},
+		cli.DurationFlag{
+			Name:   PullIntervalFlagName,
+			Usage:  "Pull interval for retries",
+			Value:  100 * time.Millisecond,
+			EnvVar: common.PrefixEnvVar(envPrefix, "GRAPH_PULL_INTERVAL"),
+		},
+		cli.UintFlag{
+			Name:   MaxRetriesFlagName,
+			Usage:  "The maximum number of retries",
+			Value:  5,
+			EnvVar: common.PrefixEnvVar(envPrefix, "GRAPH_MAX_RETRIES"),
+		},
+	}
+}
+
+func ReadCLIConfig(ctx *cli.Context) Config {
+
+	return Config{
+		Endpoint:     ctx.String(EndpointFlagName),
+		PullInterval: ctx.Duration(PullIntervalFlagName),
+		MaxRetries:   ctx.Int(MaxRetriesFlagName),
+	}
+
+}

core/thegraph/state.go

@@ -87,6 +87,17 @@ type (
 
 var _ IndexedChainState = (*indexedChainState)(nil)
 
+func MakeIndexedChainState(config Config, cs core.ChainState, logger logging.Logger) *indexedChainState {
+
+	logger.Info("Using graph node")
+	querier := graphql.NewClient(config.Endpoint, nil)
+
+	// RetryQuerier is a wrapper around the GraphQLQuerier that retries queries on failure
+	retryQuerier := NewRetryQuerier(querier, config.PullInterval, config.MaxRetries)
+
+	return NewIndexedChainState(cs, retryQuerier, logger)
+}
+
 func NewIndexedChainState(cs core.ChainState, querier GraphQLQuerier, logger logging.Logger) *indexedChainState {
 	return &indexedChainState{
 		ChainState: cs,

disperser/cmd/batcher/config.go

@@ -4,6 +4,7 @@ import (
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/aws"
 	"github.com/Layr-Labs/eigenda/common/geth"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
 	"github.com/Layr-Labs/eigenda/disperser/batcher"
 	"github.com/Layr-Labs/eigenda/disperser/cmd/batcher/flags"
 	"github.com/Layr-Labs/eigenda/disperser/common/blobstore"
@@ -23,7 +24,7 @@ type Config struct {
 	MetricsConfig    batcher.MetricsConfig
 	IndexerConfig    indexer.Config
 	FireblocksConfig common.FireblocksConfig
-	GraphUrl         string
+	ChainStateConfig thegraph.Config
 	UseGraph         bool
 
 	IndexerDataDir string
@@ -75,8 +76,8 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 			HTTPPort:      ctx.GlobalString(flags.MetricsHTTPPort.Name),
 			EnableMetrics: ctx.GlobalBool(flags.EnableMetrics.Name),
 		},
+		ChainStateConfig:              thegraph.ReadCLIConfig(ctx),
 		UseGraph:                      ctx.Bool(flags.UseGraphFlag.Name),
-		GraphUrl:                      ctx.GlobalString(flags.GraphUrlFlag.Name),
 		BLSOperatorStateRetrieverAddr: ctx.GlobalString(flags.BlsOperatorStateRetrieverFlag.Name),
 		EigenDAServiceManagerAddr:     ctx.GlobalString(flags.EigenDAServiceManagerFlag.Name),
 		IndexerDataDir:                ctx.GlobalString(flags.IndexerDataDirFlag.Name),

disperser/cmd/batcher/flags/flags.go

@@ -6,6 +6,7 @@ import (
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/aws"
 	"github.com/Layr-Labs/eigenda/common/geth"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
 	"github.com/Layr-Labs/eigenda/indexer"
 	"github.com/urfave/cli"
 )
@@ -59,12 +60,6 @@ var (
 		Required: true,
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "ENABLE_METRICS"),
 	}
-	GraphUrlFlag = cli.StringFlag{
-		Name:     common.PrefixFlag(FlagPrefix, "graph-url"),
-		Usage:    "The url of the graph node",
-		Required: true,
-		EnvVar:   common.PrefixEnvVar(envVarPrefix, "GRAPH_URL"),
-	}
 	UseGraphFlag = cli.BoolFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "use-graph"),
 		Usage:    "Whether to use the graph node",
@@ -194,7 +189,6 @@ var requiredFlags = []cli.Flag{
 	EigenDAServiceManagerFlag,
 	EncoderSocket,
 	EnableMetrics,
-	GraphUrlFlag,
 	BatchSizeLimitFlag,
 	UseGraphFlag,
 	SRSOrderFlag,
@@ -227,4 +221,5 @@ func init() {
 	Flags = append(Flags, indexer.CLIFlags(envVarPrefix)...)
 	Flags = append(Flags, aws.ClientFlags(envVarPrefix, FlagPrefix)...)
 	Flags = append(Flags, common.FireblocksCLIFlags(envVarPrefix, FlagPrefix)...)
+	Flags = append(Flags, thegraph.CLIFlags(envVarPrefix)...)
 }

disperser/cmd/batcher/main.go

@@ -8,8 +8,6 @@ import (
 	"os"
 	"time"
 
-	"github.com/shurcooL/graphql"
-
 	"github.com/Layr-Labs/eigenda/common"
 	coreindexer "github.com/Layr-Labs/eigenda/core/indexer"
 	"github.com/Layr-Labs/eigenda/core/thegraph"
@@ -140,13 +138,9 @@ func RunBatcher(ctx *cli.Context) error {
 	var ics core.IndexedChainState
 	if config.UseGraph {
 		logger.Info("Using graph node")
-		querier := graphql.NewClient(config.GraphUrl, nil)
-
-		// RetryQuerier is a wrapper around the GraphQLQuerier that retries queries on failure
-		retryQuerier := thegraph.NewRetryQuerier(querier, 100*time.Millisecond, config.EthClientConfig.NumRetries)
 
-		logger.Info("Connecting to subgraph", "url", config.GraphUrl)
-		ics = thegraph.NewIndexedChainState(cs, retryQuerier, logger)
+		logger.Info("Connecting to subgraph", "url", config.ChainStateConfig.Endpoint)
+		ics = thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
 	} else {
 		logger.Info("Using built-in indexer")
 

operators/churner/cmd/main.go

@@ -5,7 +5,6 @@ import (
 	"log"
 	"net"
 	"os"
-	"time"
 
 	pb "github.com/Layr-Labs/eigenda/api/grpc/churner"
 	"github.com/Layr-Labs/eigenda/common"
@@ -17,7 +16,6 @@ import (
 	"github.com/Layr-Labs/eigenda/operators/churner"
 	"github.com/Layr-Labs/eigenda/operators/churner/flags"
 	gethcommon "github.com/ethereum/go-ethereum/common"
-	"github.com/shurcooL/graphql"
 	"github.com/urfave/cli"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
@@ -83,12 +81,11 @@ func run(ctx *cli.Context) error {
 
 	cs := coreeth.NewChainState(tx, gethClient)
 
-	querier := graphql.NewClient(config.GraphUrl, nil)
+	logger.Info("Using graph node")
 
-	// RetryQuerier is a wrapper around the GraphQLQuerier that retries queries on failure
-	retryQuerier := thegraph.NewRetryQuerier(querier, 100*time.Millisecond, config.EthClientConfig.NumRetries)
+	logger.Info("Connecting to subgraph", "url", config.ChainStateConfig.Endpoint)
+	indexer := thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
 
-	indexer := thegraph.NewIndexedChainState(cs, retryQuerier, logger)
 	metrics := churner.NewMetrics(config.MetricsConfig.HTTPPort, logger)
 
 	cn, err := churner.NewChurner(config, indexer, tx, logger, metrics)

operators/churner/config.go

@@ -5,15 +5,16 @@ import (
 
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/geth"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
 	"github.com/Layr-Labs/eigenda/operators/churner/flags"
 	"github.com/urfave/cli"
 )
 
 type Config struct {
-	EthClientConfig geth.EthClientConfig
-	LoggerConfig    common.LoggerConfig
-	GraphUrl        string
-	MetricsConfig   MetricsConfig
+	EthClientConfig  geth.EthClientConfig
+	LoggerConfig     common.LoggerConfig
+	MetricsConfig    MetricsConfig
+	ChainStateConfig thegraph.Config
 
 	BLSOperatorStateRetrieverAddr string
 	EigenDAServiceManagerAddr     string
@@ -29,7 +30,7 @@ func NewConfig(ctx *cli.Context) (*Config, error) {
 	return &Config{
 		EthClientConfig:               geth.ReadEthClientConfig(ctx),
 		LoggerConfig:                  *loggerConfig,
-		GraphUrl:                      ctx.GlobalString(flags.GraphUrlFlag.Name),
+		ChainStateConfig:              thegraph.ReadCLIConfig(ctx),
 		BLSOperatorStateRetrieverAddr: ctx.GlobalString(flags.BlsOperatorStateRetrieverFlag.Name),
 		EigenDAServiceManagerAddr:     ctx.GlobalString(flags.EigenDAServiceManagerFlag.Name),
 		PerPublicKeyRateLimit:         ctx.GlobalDuration(flags.PerPublicKeyRateLimit.Name),

operators/churner/flags/flags.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/geth"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
 	"github.com/Layr-Labs/eigenda/indexer"
 	"github.com/urfave/cli"
 )
@@ -31,12 +32,6 @@ var (
 		Required: true,
 		EnvVar:   common.PrefixEnvVar(envPrefix, "GRPC_PORT"),
 	}
-	GraphUrlFlag = cli.StringFlag{
-		Name:     common.PrefixFlag(FlagPrefix, "indexer-graph-url"),
-		Usage:    "The url of the subgraph to query",
-		Required: true,
-		EnvVar:   common.PrefixEnvVar(envPrefix, "GRAPH_URL"),
-	}
 	BlsOperatorStateRetrieverFlag = cli.StringFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "bls-operator-state-retriever"),
 		Usage:    "Address of the BLS Operator State Retriever",
@@ -75,7 +70,6 @@ var (
 var requiredFlags = []cli.Flag{
 	HostnameFlag,
 	GrpcPortFlag,
-	GraphUrlFlag,
 	BlsOperatorStateRetrieverFlag,
 	EigenDAServiceManagerFlag,
 	EnableMetrics,
@@ -94,4 +88,5 @@ func init() {
 	Flags = append(Flags, geth.EthClientFlags(envPrefix)...)
 	Flags = append(Flags, common.LoggerCLIFlags(envPrefix, FlagPrefix)...)
 	Flags = append(Flags, indexer.CLIFlags(envPrefix)...)
+	Flags = append(Flags, thegraph.CLIFlags(envPrefix)...)
 }

retriever/cmd/main.go

@@ -6,7 +6,6 @@ import (
 	"log"
 	"net"
 	"os"
-	"time"
 
 	pb "github.com/Layr-Labs/eigenda/api/grpc/retriever"
 	"github.com/Layr-Labs/eigenda/clients"
@@ -23,7 +22,6 @@ import (
 	"github.com/Layr-Labs/eigenda/retriever/flags"
 	gethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/shurcooL/graphql"
 	"github.com/urfave/cli"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
@@ -108,13 +106,9 @@ func RetrieverMain(ctx *cli.Context) error {
 	var ics core.IndexedChainState
 	if config.UseGraph {
 		logger.Info("Using graph node")
-		querier := graphql.NewClient(config.GraphUrl, nil)
 
-		// RetryQuerier is a wrapper around the GraphQLQuerier that retries queries on failure
-		retryQuerier := thegraph.NewRetryQuerier(querier, 100*time.Millisecond, config.EthClientConfig.NumRetries)
-
-		logger.Info("Connecting to subgraph", "url", config.GraphUrl)
-		ics = thegraph.NewIndexedChainState(cs, retryQuerier, logger)
+		logger.Info("Connecting to subgraph", "url", config.ChainStateConfig.Endpoint)
+		ics = thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
 	} else {
 		logger.Info("Using built-in indexer")
 

retriever/config.go

@@ -5,25 +5,26 @@ import (
 
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/geth"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
 	"github.com/Layr-Labs/eigenda/encoding/kzg"
 	"github.com/Layr-Labs/eigenda/indexer"
 	"github.com/Layr-Labs/eigenda/retriever/flags"
 	"github.com/urfave/cli"
 )
 
 type Config struct {
-	EncoderConfig   kzg.KzgConfig
-	EthClientConfig geth.EthClientConfig
-	LoggerConfig    common.LoggerConfig
-	IndexerConfig   indexer.Config
-	MetricsConfig   MetricsConfig
+	EncoderConfig    kzg.KzgConfig
+	EthClientConfig  geth.EthClientConfig
+	LoggerConfig     common.LoggerConfig
+	IndexerConfig    indexer.Config
+	MetricsConfig    MetricsConfig
+	ChainStateConfig thegraph.Config
 
 	IndexerDataDir                string
 	Timeout                       time.Duration
 	NumConnections                int
 	BLSOperatorStateRetrieverAddr string
 	EigenDAServiceManagerAddr     string
-	GraphUrl                      string
 	UseGraph                      bool
 }
 
@@ -40,12 +41,12 @@ func NewConfig(ctx *cli.Context) (*Config, error) {
 		MetricsConfig: MetricsConfig{
 			HTTPPort: ctx.GlobalString(flags.MetricsHTTPPortFlag.Name),
 		},
+		ChainStateConfig:              thegraph.ReadCLIConfig(ctx),
 		IndexerDataDir:                ctx.GlobalString(flags.IndexerDataDirFlag.Name),
 		Timeout:                       ctx.Duration(flags.TimeoutFlag.Name),
 		NumConnections:                ctx.Int(flags.NumConnectionsFlag.Name),
 		BLSOperatorStateRetrieverAddr: ctx.GlobalString(flags.BlsOperatorStateRetrieverFlag.Name),
 		EigenDAServiceManagerAddr:     ctx.GlobalString(flags.EigenDAServiceManagerFlag.Name),
-		GraphUrl:                      ctx.GlobalString(flags.GraphUrlFlag.Name),
 		UseGraph:                      ctx.GlobalBool(flags.UseGraphFlag.Name),
 	}, nil
 }

retriever/flags/flags.go

@@ -3,6 +3,7 @@ package flags
 import (
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/geth"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
 	"github.com/Layr-Labs/eigenda/encoding/kzg"
 	"github.com/Layr-Labs/eigenda/indexer"
 	"github.com/urfave/cli"
@@ -67,12 +68,6 @@ var (
 		Value:    "9100",
 		EnvVar:   common.PrefixEnvVar(envPrefix, "METRICS_HTTP_PORT"),
 	}
-	GraphUrlFlag = cli.StringFlag{
-		Name:     common.PrefixFlag(FlagPrefix, "graph-url"),
-		Usage:    "The url of the graph node",
-		Required: false,
-		EnvVar:   common.PrefixEnvVar(envPrefix, "GRAPH_URL"),
-	}
 	UseGraphFlag = cli.BoolFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "use-graph"),
 		Usage:    "Whether to use the graph node",
@@ -93,7 +88,6 @@ var optionalFlags = []cli.Flag{
 	NumConnectionsFlag,
 	IndexerDataDirFlag,
 	MetricsHTTPPortFlag,
-	GraphUrlFlag,
 	UseGraphFlag,
 }
 
@@ -106,4 +100,5 @@ func init() {
 	Flags = append(Flags, geth.EthClientFlags(envPrefix)...)
 	Flags = append(Flags, common.LoggerCLIFlags(envPrefix, FlagPrefix)...)
 	Flags = append(Flags, indexer.CLIFlags(envPrefix)...)
+	Flags = append(Flags, thegraph.CLIFlags(envPrefix)...)
 }