temp: add socket read fn

Author: hopeyen

api/clients/retrieval_client.go

@@ -106,8 +106,25 @@ func (r *retrievalClient) RetrieveBlobChunks(ctx context.Context,
 
 	operatorState, err := r.chainState.GetOperatorStateWithSocket(ctx, referenceBlockNumber, []core.QuorumID{quorumID})
 	if err != nil {
+		r.logger.Error("failed to get operator state", "err", err)
 		return nil, err
 	}
+
+	// Log operator state in a structured way
+	if operatorState != nil && operatorState.Operators != nil {
+		if operators, ok := operatorState.Operators[quorumID]; ok {
+			for opID, opInfo := range operators {
+				r.logger.Info("operator info",
+					"quorumID", quorumID,
+					"operatorID", opID.Hex(),
+					"stake", opInfo.Stake.String(),
+					"index", opInfo.Index,
+					"socket", opInfo.Socket,
+				)
+			}
+		}
+	}
+
 	operators, ok := operatorState.Operators[quorumID]
 	if !ok {
 		return nil, fmt.Errorf("no quorum with ID: %d", quorumID)

core/chainio.go

@@ -19,7 +19,7 @@ type OperatorStake struct {
 type OperatorStakeWithSocket struct {
 	OperatorID OperatorID
 	Stake      *big.Int
-	Socket     string
+	Socket     OperatorSocket
 }
 
 type OperatorToChurn struct {
@@ -51,6 +51,10 @@ type Reader interface {
 	// The indices of the operators within each quorum are also returned.
 	GetOperatorStakesForQuorums(ctx context.Context, quorums []QuorumID, blockNumber uint32) (OperatorStakes, error)
 
+	// GetOperatorStakesWithSocketForQuorums returns the stakes of all operators within the supplied quorums. The returned stakes are for the block number supplied.
+	// The indices of the operators within each quorum are also returned.
+	GetOperatorStakesWithSocketForQuorums(ctx context.Context, quorums []QuorumID, blockNumber uint32) (OperatorStakesWithSocket, error)
+
 	// GetOperatorStakesWithSocket returns the operator state with socket for the given block number and quorums.
 	GetOperatorStakesWithSocket(ctx context.Context, operatorID OperatorID, blockNumber uint32) (OperatorStakesWithSocket, []QuorumID, error)
 

core/eth/reader.go

@@ -461,7 +461,7 @@ func (t *Reader) GetOperatorStakesWithSocket(ctx context.Context, operator core.
 			state[quorumID][operatorIndex] = core.OperatorStakeWithSocket{
 				Stake:      op.Stake,
 				OperatorID: op.OperatorId,
-				Socket:     result.Sockets[i][j],
+				Socket:     core.OperatorSocket(result.Sockets[i][j]),
 			}
 		}
 	}
@@ -552,7 +552,7 @@ func (t *Reader) GetOperatorStakesWithSocketForQuorums(ctx context.Context, quor
 			state[quorumID][operatorIndex] = core.OperatorStakeWithSocket{
 				Stake:      op.Stake,
 				OperatorID: op.OperatorId,
-				Socket:     result.Sockets[i][j],
+				Socket:     core.OperatorSocket(result.Sockets[i][j]),
 			}
 		}
 	}

core/eth/state.go

@@ -42,12 +42,12 @@ func (cs *ChainState) GetOperatorState(ctx context.Context, blockNumber uint, qu
 }
 
 func (cs *ChainState) GetOperatorStateWithSocket(ctx context.Context, blockNumber uint, quorums []core.QuorumID) (*core.OperatorState, error) {
-	operatorsByQuorum, err := cs.Tx.GetOperatorStakesForQuorums(ctx, quorums, uint32(blockNumber))
+	operatorsByQuorum, err := cs.Tx.GetOperatorStakesWithSocketForQuorums(ctx, quorums, uint32(blockNumber))
 	if err != nil {
 		return nil, err
 	}
 
-	return getOperatorState(operatorsByQuorum, uint32(blockNumber))
+	return getOperatorStateWithSocket(operatorsByQuorum, uint32(blockNumber))
 }
 
 func (cs *ChainState) GetCurrentBlockNumber(ctx context.Context) (uint, error) {
@@ -97,3 +97,36 @@ func getOperatorState(operatorsByQuorum core.OperatorStakes, blockNumber uint32)
 
 	return state, nil
 }
+
+func getOperatorStateWithSocket(operatorsByQuorum core.OperatorStakesWithSocket, blockNumber uint32) (*core.OperatorState, error) {
+	operators := make(map[core.QuorumID]map[core.OperatorID]*core.OperatorInfo)
+	totals := make(map[core.QuorumID]*core.OperatorInfo)
+
+	for quorumID, quorum := range operatorsByQuorum {
+		totalStake := big.NewInt(0)
+		operators[quorumID] = make(map[core.OperatorID]*core.OperatorInfo)
+
+		for ind, op := range quorum {
+			operators[quorumID][op.OperatorID] = &core.OperatorInfo{
+				Stake:  op.Stake,
+				Index:  core.OperatorIndex(ind),
+				Socket: core.OperatorSocket(op.Socket),
+			}
+			totalStake.Add(totalStake, op.Stake)
+		}
+
+		totals[quorumID] = &core.OperatorInfo{
+			Stake:  totalStake,
+			Index:  core.OperatorIndex(len(quorum)),
+			Socket: core.OperatorSocket(""),
+		}
+	}
+
+	state := &core.OperatorState{
+		Operators:   operators,
+		Totals:      totals,
+		BlockNumber: uint(blockNumber),
+	}
+
+	return state, nil
+}

retriever/cmd/main.go

@@ -17,7 +17,6 @@ import (
 	"github.com/Layr-Labs/eigenda/common/healthcheck"
 	"github.com/Layr-Labs/eigenda/core"
 	"github.com/Layr-Labs/eigenda/core/eth"
-	"github.com/Layr-Labs/eigenda/core/thegraph"
 	"github.com/Layr-Labs/eigenda/encoding/kzg/verifier"
 	"github.com/Layr-Labs/eigenda/retriever"
 	retrivereth "github.com/Layr-Labs/eigenda/retriever/eth"
@@ -100,12 +99,9 @@ func RetrieverMain(ctx *cli.Context) error {
 		log.Fatalln("could not start tcp listener", err)
 	}
 
-	logger.Info("Connecting to subgraph", "url", config.ChainStateConfig.Endpoint)
-	ics := thegraph.MakeIndexedChainState(config.ChainStateConfig, cs, logger)
-
 	if config.EigenDAVersion == 1 {
 		agn := &core.StdAssignmentCoordinator{}
-		retrievalClient, err := clients.NewRetrievalClient(logger, ics, agn, nodeClient, v, config.NumConnections)
+		retrievalClient, err := clients.NewRetrievalClient(logger, cs, agn, nodeClient, v, config.NumConnections)
 		if err != nil {
 			log.Fatalln("could not start tcp listener", err)
 		}
@@ -131,8 +127,8 @@ func RetrieverMain(ctx *cli.Context) error {
 	}
 
 	if config.EigenDAVersion == 2 {
-		retrievalClient := clientsv2.NewRetrievalClient(logger, tx, ics, v, config.NumConnections)
-		retrieverServiceServer := retrieverv2.NewServer(config, logger, retrievalClient, ics)
+		retrievalClient := clientsv2.NewRetrievalClient(logger, tx, cs, v, config.NumConnections)
+		retrieverServiceServer := retrieverv2.NewServer(config, logger, retrievalClient, cs)
 		if err = retrieverServiceServer.Start(context.Background()); err != nil {
 			log.Fatalln("failed to start retriever service server", err)
 		}

retriever/config.go

@@ -6,18 +6,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/encoding/kzg"
 	"github.com/Layr-Labs/eigenda/retriever/flags"
 	"github.com/urfave/cli"
 )
 
 type Config struct {
-	EncoderConfig    kzg.KzgConfig
-	EthClientConfig  geth.EthClientConfig
-	LoggerConfig     common.LoggerConfig
-	MetricsConfig    MetricsConfig
-	ChainStateConfig thegraph.Config
+	EncoderConfig   kzg.KzgConfig
+	EthClientConfig geth.EthClientConfig
+	LoggerConfig    common.LoggerConfig
+	MetricsConfig   MetricsConfig
 
 	Timeout                       time.Duration
 	NumConnections                int
@@ -44,7 +42,6 @@ func NewConfig(ctx *cli.Context) (*Config, error) {
 		MetricsConfig: MetricsConfig{
 			HTTPPort: ctx.GlobalString(flags.MetricsHTTPPortFlag.Name),
 		},
-		ChainStateConfig:              thegraph.ReadCLIConfig(ctx),
 		Timeout:                       ctx.Duration(flags.TimeoutFlag.Name),
 		NumConnections:                ctx.Int(flags.NumConnectionsFlag.Name),
 		BLSOperatorStateRetrieverAddr: ctx.GlobalString(flags.BlsOperatorStateRetrieverFlag.Name),

retriever/flags/flags.go

@@ -3,7 +3,6 @@ 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/urfave/cli"
 )
@@ -91,5 +90,4 @@ func init() {
 	Flags = append(Flags, kzg.CLIFlags(envPrefix)...)
 	Flags = append(Flags, geth.EthClientFlags(envPrefix)...)
 	Flags = append(Flags, common.LoggerCLIFlags(envPrefix, FlagPrefix)...)
-	Flags = append(Flags, thegraph.CLIFlags(envPrefix)...)
 }