[SocketRegistry] get operator socket (#863)

core/chainio.go

@@ -72,6 +72,9 @@ type Reader interface {
 	// GetOperatorSetParams returns operator set params for the quorum.
 	GetOperatorSetParams(ctx context.Context, quorumID QuorumID) (*OperatorSetParam, error)
 
+	// GetOperatorSocket returns a operator's socket.
+	GetOperatorSocket(ctx context.Context, operatorID OperatorID) (string, error)
+
 	// GetNumberOfRegisteredOperatorForQuorum returns the number of registered operators for the quorum.
 	GetNumberOfRegisteredOperatorForQuorum(ctx context.Context, quorumID QuorumID) (uint32, error)
 

core/eth/reader.go

@@ -15,6 +15,7 @@ import (
 	indexreg "github.com/Layr-Labs/eigenda/contracts/bindings/IIndexRegistry"
 	opstateretriever "github.com/Layr-Labs/eigenda/contracts/bindings/OperatorStateRetriever"
 	regcoordinator "github.com/Layr-Labs/eigenda/contracts/bindings/RegistryCoordinator"
+	socketreg "github.com/Layr-Labs/eigenda/contracts/bindings/SocketRegistry"
 	stakereg "github.com/Layr-Labs/eigenda/contracts/bindings/StakeRegistry"
 	"github.com/Layr-Labs/eigenda/core"
 	"github.com/Layr-Labs/eigensdk-go/logging"
@@ -37,6 +38,7 @@ type ContractBindings struct {
 	EigenDAServiceManager *eigendasrvmg.ContractEigenDAServiceManager
 	EjectionManager       *ejectionmg.ContractEjectionManager
 	AVSDirectory          *avsdir.ContractAVSDirectory
+	SocketRegistry        *socketreg.ContractSocketRegistry
 }
 
 type Reader struct {
@@ -164,10 +166,23 @@ func (t *Reader) updateContractBindings(blsOperatorStateRetrieverAddr, eigenDASe
 		return err
 	}
 
+	socketRegistryAddr, err := contractIRegistryCoordinator.SocketRegistry(&bind.CallOpts{})
+	if err != nil {
+		t.logger.Error("Failed to fetch SocketRegistry address", "err", err)
+		return err
+	}
+
+	contractSocketRegistry, err := socketreg.NewContractSocketRegistry(socketRegistryAddr, t.ethClient)
+	if err != nil {
+		t.logger.Error("Failed to fetch SocketRegistry contract", "err", err)
+		return err
+	}
+
 	t.bindings = &ContractBindings{
 		ServiceManagerAddr:    eigenDAServiceManagerAddr,
 		RegCoordinatorAddr:    registryCoordinatorAddr,
 		AVSDirectory:          contractAVSDirectory,
+		SocketRegistry:        contractSocketRegistry,
 		OpStateRetriever:      contractBLSOpStateRetr,
 		BLSApkRegistry:        contractBLSPubkeyReg,
 		IndexRegistry:         contractIIndexReg,
@@ -377,6 +392,12 @@ func (t *Reader) StakeRegistry(ctx context.Context) (gethcommon.Address, error)
 	})
 }
 
+func (t *Reader) SocketRegistry(ctx context.Context) (gethcommon.Address, error) {
+	return t.bindings.RegistryCoordinator.SocketRegistry(&bind.CallOpts{
+		Context: ctx,
+	})
+}
+
 func (t *Reader) OperatorIDToAddress(ctx context.Context, operatorId core.OperatorID) (gethcommon.Address, error) {
 	return t.bindings.BLSApkRegistry.PubkeyHashToOperator(&bind.CallOpts{
 		Context: ctx,
@@ -655,3 +676,16 @@ func (t *Reader) GetReservationWindow(ctx context.Context) (uint32, error) {
 	// contract is not implemented yet
 	return 0, nil
 }
+
+func (t *Reader) GetOperatorSocket(ctx context.Context, operatorId core.OperatorID) (string, error) {
+	socket, err := t.bindings.SocketRegistry.GetOperatorSocket(&bind.CallOpts{
+		Context: ctx,
+	}, [32]byte(operatorId))
+	if err != nil {
+		return "", err
+	}
+	if socket == "" {
+		return "", errors.New("operator socket string is empty, check operator with id: " + operatorId.Hex())
+	}
+	return socket, nil
+}

core/eth/state.go

@@ -51,6 +51,14 @@ func (cs *ChainState) GetCurrentBlockNumber() (uint, error) {
 	return uint(header.Number.Uint64()), nil
 }
 
+func (cs *ChainState) GetOperatorSocket(ctx context.Context, blockNumber uint, operator core.OperatorID) (string, error) {
+	socket, err := cs.Tx.GetOperatorSocket(ctx, operator)
+	if err != nil {
+		return "", err
+	}
+	return socket, nil
+}
+
 func getOperatorState(operatorsByQuorum core.OperatorStakes, blockNumber uint32) (*core.OperatorState, error) {
 	operators := make(map[core.QuorumID]map[core.OperatorID]*core.OperatorInfo)
 	totals := make(map[core.QuorumID]*core.OperatorInfo)

core/mock/state.go

@@ -238,6 +238,13 @@ func (d *ChainDataMock) GetOperatorStateByOperator(ctx context.Context, blockNum
 
 }
 
+func (d *ChainDataMock) GetOperatorSocket(ctx context.Context, blockNumber uint, operator core.OperatorID) (string, error) {
+
+	state := d.GetTotalOperatorState(ctx, blockNumber)
+
+	return state.IndexedOperatorState.IndexedOperators[operator].Socket, nil
+}
+
 func (d *ChainDataMock) GetIndexedOperatorState(ctx context.Context, blockNumber uint, quorums []core.QuorumID) (*core.IndexedOperatorState, error) {
 
 	state := d.GetTotalOperatorStateWithQuorums(ctx, blockNumber, quorums)

core/mock/writer.go

@@ -241,3 +241,9 @@ func (t *MockWriter) GetOnDemandPaymentByAccount(ctx context.Context, blockNumbe
 	result := args.Get(0)
 	return result.(core.OnDemandPayment), args.Error(1)
 }
+
+func (t *MockWriter) GetOperatorSocket(ctx context.Context, operatorID core.OperatorID) (string, error) {
+	args := t.Called()
+	result := args.Get(0)
+	return result.(string), args.Error(1)
+}

core/state.go

@@ -128,6 +128,7 @@ type ChainState interface {
 	GetCurrentBlockNumber() (uint, error)
 	GetOperatorState(ctx context.Context, blockNumber uint, quorums []QuorumID) (*OperatorState, error)
 	GetOperatorStateByOperator(ctx context.Context, blockNumber uint, operator OperatorID) (*OperatorState, error)
+	GetOperatorSocket(ctx context.Context, blockNumber uint, operator OperatorID) (string, error)
 }
 
 // ChainState is an interface for getting information about the current chain state.

disperser/dataapi/queried_operators_handlers.go

@@ -295,6 +295,19 @@ func (s *server) scanOperatorsHostInfo(ctx context.Context) (*SemverReportRespon
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch indexed operator info - %s", err)
 	}
+
+	// check operator socket registration against the indexed state
+	for operatorID, operatorInfo := range operators {
+		socket, err := s.chainState.GetOperatorSocket(context.Background(), currentBlock, operatorID)
+		if err != nil {
+			s.logger.Warn("failed to get operator socket", "operatorId", operatorID.Hex(), "error", err)
+			continue
+		}
+		if socket != operatorInfo.Socket {
+			s.logger.Warn("operator socket mismatch", "operatorId", operatorID.Hex(), "socket", socket, "operatorInfo", operatorInfo.Socket)
+		}
+	}
+
 	s.logger.Info("Queried indexed operators", "operators", len(operators), "block", currentBlock)
 	operatorState, err := s.chainState.GetOperatorState(context.Background(), currentBlock, []core.QuorumID{0, 1, 2})
 	if err != nil {

node/node.go

@@ -305,7 +305,6 @@ func (n *Node) Start(ctx context.Context) error {
 		n.Logger.Info("Registering node on chain with the following parameters:", "operatorId",
 			n.Config.ID.Hex(), "hostname", n.Config.Hostname, "dispersalPort", n.Config.DispersalPort,
 			"retrievalPort", n.Config.RetrievalPort, "churnerUrl", n.Config.ChurnerUrl, "quorumIds", fmt.Sprint(n.Config.QuorumIDList))
-		socket := string(core.MakeOperatorSocket(n.Config.Hostname, n.Config.DispersalPort, n.Config.RetrievalPort))
 		privateKey, err := crypto.HexToECDSA(n.Config.EthClientConfig.PrivateKeyString)
 		if err != nil {
 			return fmt.Errorf("NewClient: cannot parse private key: %w", err)
@@ -326,6 +325,15 @@ func (n *Node) Start(ctx context.Context) error {
 			return fmt.Errorf("failed to register the operator: %w", err)
 		}
 	} else {
+		registeredSocket, err := n.Transactor.GetOperatorSocket(ctx, n.Config.ID)
+		// Error out if registration on-chain is a requirement
+		if err != nil {
+			n.Logger.Warnf("failed to get operator socket: %w", err)
+		}
+		if registeredSocket != socket {
+			n.Logger.Warnf("registered socket %s does not match expected socket %s", registeredSocket, socket)
+		}
+
 		eigenDAUrl, ok := eigenDAUIMap[n.ChainID.String()]
 		if ok {
 			n.Logger.Infof("The node has successfully started. Note: if it's not opted in on %s, then please follow the EigenDA operator guide section in docs.eigenlayer.xyz to register", eigenDAUrl)

node/node_test.go

@@ -2,6 +2,7 @@ package node_test
 
 import (
 	"context"
+	"errors"
 	"os"
 	"runtime"
 	"testing"
@@ -105,6 +106,8 @@ func TestNodeStartNoAddress(t *testing.T) {
 	c := newComponents(t)
 	c.node.Config.RegisterNodeAtStart = false
 
+	c.tx.On("GetOperatorSocket", mock.Anything).Return("", errors.New("failed to get operator socket"))
+
 	err := c.node.Start(context.Background())
 	assert.NoError(t, err)
 }

test/integration_test.go

@@ -376,6 +376,8 @@ func mustMakeOperators(t *testing.T, cst *coremock.ChainDataMock, logger logging
 		tx.On("GetBlockStaleMeasure").Return(nil)
 		tx.On("GetStoreDurationBlocks").Return(nil)
 		tx.On("OperatorIDToAddress").Return(gethcommon.Address{1}, nil)
+		socket := core.MakeOperatorSocket(config.Hostname, config.DispersalPort, config.RetrievalPort)
+		tx.On("GetOperatorSocket", mock.Anything, mock.Anything).Return(socket.String(), nil)
 
 		noopMetrics := metrics.NewNoopMetrics()
 		reg := prometheus.NewRegistry()

tools/semverscan/cmd/main.go

@@ -87,6 +87,20 @@ func RunScan(ctx *cli.Context) error {
 			}
 		}
 	}
+
+	// check operator socket registration against the indexed state
+	for operatorID, operatorInfo := range operators {
+		socket, err := chainState.GetOperatorSocket(context.Background(), currentBlock, operatorID)
+		if err != nil {
+			logger.Warn("failed to get operator socket", "operatorId", operatorID.Hex(), "error", err)
+			continue
+		}
+		if socket != operatorInfo.Socket {
+			// delete operator from operators if there's a mistmatch?
+			logger.Warn("operator socket mismatch", "operatorId", operatorID.Hex(), "socket", socket, "operatorInfo", operatorInfo.Socket)
+		}
+	}
+
 	logger.Info("Queried operator state", "count", len(operators))
 
 	semvers := semver.ScanOperators(operators, operatorState, config.UseRetrievalClient, config.Workers, config.Timeout, logger)