feat: Add reservations expiration tracking to DataAPI (#1733)

* Initialize subgraph

* Initialize subgraph

* Initialize subgraph

* Save payment subgraph state

* fix

* Add reservation collector to dataapi

* Add query examples + more test coverage

* Remove docker compose

* Add scripts for updating abi and adds a CI check when subgraph tests run

* fix flags

* fix lint error

* Fix rebase issues

* Inject the registry

* wrap errors

* expose more info from reservation data

* fmt

* rename updateCounts to updateMetrics

* Document subgraph.NewApi in ejections tool

Author: dmanc

disperser/cmd/dataapi/config.go

@@ -28,6 +28,7 @@ type Config struct {
 	PrometheusApiAddr            string
 	SubgraphApiBatchMetadataAddr string
 	SubgraphApiOperatorStateAddr string
+	SubgraphApiPaymentsAddr      string
 	ServerMode                   string
 	AllowOrigins                 []string
 
@@ -63,6 +64,7 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 		SocketAddr:                    ctx.GlobalString(flags.SocketAddrFlag.Name),
 		SubgraphApiBatchMetadataAddr:  ctx.GlobalString(flags.SubgraphApiBatchMetadataAddrFlag.Name),
 		SubgraphApiOperatorStateAddr:  ctx.GlobalString(flags.SubgraphApiOperatorStateAddrFlag.Name),
+		SubgraphApiPaymentsAddr:       ctx.GlobalString(flags.SubgraphApiPaymentsAddrFlag.Name),
 		BLSOperatorStateRetrieverAddr: ctx.GlobalString(flags.BlsOperatorStateRetrieverFlag.Name),
 		EigenDAServiceManagerAddr:     ctx.GlobalString(flags.EigenDAServiceManagerFlag.Name),
 		EigenDADirectory:              ctx.GlobalString(flags.EigenDADirectoryFlag.Name),

disperser/cmd/dataapi/flags/flags.go

@@ -59,15 +59,15 @@ var (
 	}
 	SubgraphApiBatchMetadataAddrFlag = cli.StringFlag{
 		Name: common.PrefixFlag(FlagPrefix, "sub-batch-metadata-socket-addr"),
-		//We need the socket address of the subgraph batch metadata api to pull the subgraph data from.
-		Usage:    "the socket address of the subgraph batch metadata api",
+		//We need the URL of the subgraph batch metadata api to pull the subgraph data from.
+		Usage:    "the URL of the subgraph batch metadata api",
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "SUBGRAPH_BATCH_METADATA_API_SOCKET_ADDR"),
 		Required: true,
 	}
 	SubgraphApiOperatorStateAddrFlag = cli.StringFlag{
 		Name: common.PrefixFlag(FlagPrefix, "sub-op-state-socket-addr"),
-		//We need the socket address of the subgraph operator state api to pull the subgraph data from.
-		Usage:    "the socket address of the subgraph operator state api",
+		//We need the URL of the subgraph operator state api to pull the subgraph data from.
+		Usage:    "the URL of the subgraph operator state api",
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "SUBGRAPH_OPERATOR_STATE_API_SOCKET_ADDR"),
 		Required: true,
 	}
@@ -77,6 +77,13 @@ var (
 		Required: false,
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "EIGENDA_DIRECTORY"),
 	}
+	SubgraphApiPaymentsAddrFlag = cli.StringFlag{
+		Name: common.PrefixFlag(FlagPrefix, "sub-payments-socket-addr"),
+		//We need the URL of the subgraph payments api to pull the subgraph data from.
+		Usage:    "the URL of the subgraph payments api",
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "SUBGRAPH_PAYMENTS_API_SOCKET_ADDR"),
+		Required: true,
+	}
 	BlsOperatorStateRetrieverFlag = cli.StringFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "bls-operator-state-retriever"),
 		Usage:    "[Deprecated: use EigenDADirectory instead] Address of the BLS operator state Retriever",
@@ -153,6 +160,7 @@ var requiredFlags = []cli.Flag{
 	S3BucketNameFlag,
 	SubgraphApiBatchMetadataAddrFlag,
 	SubgraphApiOperatorStateAddrFlag,
+	SubgraphApiPaymentsAddrFlag,
 	PrometheusServerURLFlag,
 	PrometheusServerUsernameFlag,
 	PrometheusServerSecretFlag,

disperser/cmd/dataapi/main.go

@@ -95,7 +95,7 @@ func RunDataApi(ctx *cli.Context) error {
 	var (
 		reg               = prometheus.NewRegistry()
 		promClient        = dataapi.NewPrometheusClient(promApi, config.PrometheusConfig.Cluster)
-		subgraphApi       = subgraph.NewApi(config.SubgraphApiBatchMetadataAddr, config.SubgraphApiOperatorStateAddr)
+		subgraphApi       = subgraph.NewApi(config.SubgraphApiBatchMetadataAddr, config.SubgraphApiOperatorStateAddr, config.SubgraphApiPaymentsAddr)
 		subgraphClient    = dataapi.NewSubgraphClient(subgraphApi, logger)
 		chainState        = coreeth.NewChainState(tx, client)
 		indexedChainState = thegraph.MakeIndexedChainState(config.ChainStateConfig, chainState, logger)
@@ -108,6 +108,11 @@ func RunDataApi(ctx *cli.Context) error {
 			Registry:    reg,
 			Backend:     blobstorev2.BackendDynamoDB,
 		})
+
+		// Register reservation collector
+		reservationCollector := serverv2.NewReservationExpirationCollector(subgraphClient, logger)
+		reg.MustRegister(reservationCollector)
+
 		metrics := dataapi.NewMetrics(config.ServerVersion, reg, blobMetadataStorev2, config.MetricsConfig.HTTPPort, logger)
 		serverv2, err := serverv2.NewServerV2(
 			dataapi.Config{

disperser/dataapi/Makefile

@@ -15,3 +15,7 @@ generate-swagger-v2:
 	swag fmt --dir ./v2
 
 generate-swagger: generate-swagger-v1 generate-swagger-v2
+
+run: build
+	@echo "  >  Running dataapi..."
+	cd .. && ./bin/dataapi

disperser/dataapi/metrics.go

@@ -25,8 +25,6 @@ type MetricsConfig struct {
 }
 
 type Metrics struct {
-	registry *prometheus.Registry
-
 	NumRequests    *prometheus.CounterVec
 	CacheHitsTotal *prometheus.CounterVec
 	Latency        *prometheus.SummaryVec
@@ -40,14 +38,15 @@ type Metrics struct {
 	SemversStakePctQuorum1 *prometheus.GaugeVec
 	SemversStakePctQuorum2 *prometheus.GaugeVec
 
+	registry *prometheus.Registry
 	httpPort string
 	logger   logging.Logger
 }
 
 func NewMetrics(serverVersion uint, reg *prometheus.Registry, blobMetadataStore interface{}, httpPort string, logger logging.Logger) *Metrics {
 	namespace := "eigenda_dataapi"
 	if reg == nil {
-		reg = prometheus.NewRegistry()
+		panic("registry must not be nil")
 	}
 
 	reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))

disperser/dataapi/server_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/consensys/gnark-crypto/ecc/bn254/fp"
 	gethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/gin-gonic/gin"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -53,7 +54,7 @@ var (
 
 	serverVersion     = uint(1)
 	mockTx            = &coremock.MockWriter{}
-	metrics           = dataapi.NewMetrics(serverVersion, nil, nil, "9001", mockLogger)
+	metrics           = dataapi.NewMetrics(serverVersion, prometheus.NewRegistry(), nil, "9001", mockLogger)
 	opId0, _          = core.OperatorIDFromHex("e22dae12a0074f20b8fc96a0489376db34075e545ef60c4845d264a732568311")
 	opId1, _          = core.OperatorIDFromHex("e23cae12a0074f20b8fc96a0489376db34075e545ef60c4845d264b732568312")
 	mockChainState, _ = coremock.NewChainDataMock(map[uint8]map[core.OperatorID]int{
@@ -73,7 +74,7 @@ var (
 	})
 	_                               = mockTx.On("GetCurrentBlockNumber").Return(uint32(1), nil)
 	_                               = mockTx.On("GetQuorumCount").Return(uint8(2), nil)
-	testDataApiServer, _            = dataapi.NewServer(config, blobstore, prometheusClient, subgraphClient, mockTx, mockChainState, mockIndexedChainState, mockLogger, dataapi.NewMetrics(serverVersion, nil, nil, "9001", mockLogger), &MockGRPCConnection{}, nil, nil)
+	testDataApiServer, _            = dataapi.NewServer(config, blobstore, prometheusClient, subgraphClient, mockTx, mockChainState, mockIndexedChainState, mockLogger, dataapi.NewMetrics(serverVersion, prometheus.NewRegistry(), nil, "9001", mockLogger), &MockGRPCConnection{}, nil, nil)
 	expectedRequestedAt             = uint64(5567830000000000000)
 	expectedDataLength              = 32
 	expectedBatchId                 = uint32(99)

disperser/dataapi/subgraph/api.go

@@ -29,23 +29,27 @@ type (
 		QueryOperatorRemovedFromQuorum(ctx context.Context, startBlock, endBlock uint32) ([]*OperatorQuorum, error)
 		QueryOperatorEjectionsGteBlockTimestampByOperatorId(ctx context.Context, blockTimestamp uint64, operatorId string, first uint, skip uint) ([]*OperatorEjection, error)
 		QueryOperatorEjectionsGteBlockTimestamp(ctx context.Context, blockTimestamp uint64, first uint, skip uint) ([]*OperatorEjection, error)
+		QueryReservations(ctx context.Context, currentTimestamp uint64, first, skip int) ([]*Reservation, error)
 	}
 
 	api struct {
 		uiMonitoringGql  *graphql.Client
 		operatorStateGql *graphql.Client
+		paymentsGql      *graphql.Client
 	}
 )
 
 var _ Api = (*api)(nil)
 
-func NewApi(uiMonitoringSocketAddr string, operatorStateSocketAddr string) *api {
+func NewApi(uiMonitoringSocketAddr string, operatorStateSocketAddr string, paymentsSocketAddr string) *api {
 	once.Do(func() {
 		uiMonitoringGql := graphql.NewClient(uiMonitoringSocketAddr, nil)
 		operatorStateGql := graphql.NewClient(operatorStateSocketAddr, nil)
+		paymentsGql := graphql.NewClient(paymentsSocketAddr, nil)
 		instance = &api{
 			uiMonitoringGql:  uiMonitoringGql,
 			operatorStateGql: operatorStateGql,
+			paymentsGql:      paymentsGql,
 		}
 	})
 	return instance
@@ -301,3 +305,17 @@ func (a *api) QueryOperatorRemovedFromQuorum(ctx context.Context, startBlock, en
 	}
 	return removedFromQuorums, nil
 }
+
+func (a *api) QueryReservations(ctx context.Context, currentTimestamp uint64, first, skip int) ([]*Reservation, error) {
+	variables := map[string]any{
+		"currentTimestamp": graphql.Int(currentTimestamp),
+		"first":            graphql.Int(first),
+		"skip":             graphql.Int(skip),
+	}
+	result := new(queryReservations)
+	err := a.paymentsGql.Query(ctx, result, variables)
+	if err != nil {
+		return nil, err
+	}
+	return result.Reservations, nil
+}

disperser/dataapi/subgraph/mock/api.go

@@ -206,3 +206,14 @@ func (m *MockSubgraphApi) QueryOperatorEjectionsGteBlockTimestampByOperatorId(ct
 
 	return value, args.Error(1)
 }
+
+func (m *MockSubgraphApi) QueryReservations(ctx context.Context, currentTimestamp uint64, first, skip int) ([]*subgraph.Reservation, error) {
+	args := m.Called()
+
+	var value []*subgraph.Reservation
+	if args.Get(0) != nil {
+		value = args.Get(0).([]*subgraph.Reservation)
+	}
+
+	return value, args.Error(1)
+}

disperser/dataapi/subgraph/queries.go

@@ -81,7 +81,14 @@ type (
 		BlockNumber uint32
 		Metadata    *Operator
 	}
-
+	Reservation struct {
+		Account          graphql.String
+		SymbolsPerSecond graphql.String
+		QuorumNumbers    graphql.String
+		QuorumSplits     graphql.String
+		StartTimestamp   graphql.String
+		EndTimestamp     graphql.String
+	}
 	queryBatches struct {
 		Batches []*Batches `graphql:"batches(orderDirection: $orderDirection, orderBy: $orderBy, first: $first, skip: $skip)"`
 	}
@@ -118,4 +125,7 @@ type (
 	queryOperatorEjectedsByOperatorID struct {
 		OperatorEjections []*OperatorEjection `graphql:"operatorEjecteds(orderBy: blockTimestamp, where: {and: [{blockTimestamp_gte: $blockTimestamp_gte}, {operatorId: $operatorId}]}, first: $first, skip: $skip)"`
 	}
+	queryReservations struct {
+		Reservations []*Reservation `graphql:"reservations(where: {startTimestamp_lte: $currentTimestamp, endTimestamp_gte: $currentTimestamp}, orderBy: startTimestamp, orderDirection: asc, first: $first, skip: $skip)"`
+	}
 )