Skip to content

feat: distributed retrieval operator socket #1403

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Mar 21, 2025
Merged

feat: distributed retrieval operator socket #1403

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
992665c
chore: update eigenlayer-middleware commit
hopeyen Mar 19, 2025
74a2119
chore: contract binding gen
hopeyen Mar 19, 2025
ce9c8f7
feat: rm thegraph from retriever client, use OperatorStateWithSocket
hopeyen Mar 19, 2025
2e704b2
refactor: simplify retriever start fn
hopeyen Mar 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions api/clients/mock/node_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ func NewNodeClient() *MockNodeClient {
return &MockNodeClient{}
}

func (c *MockNodeClient) GetBlobHeader(ctx context.Context, socket string, batchHeaderHash [32]byte, blobIndex uint32) (*core.BlobHeader, *merkletree.Proof, error) {
func (c *MockNodeClient) GetBlobHeader(ctx context.Context, socket core.OperatorSocket, batchHeaderHash [32]byte, blobIndex uint32) (*core.BlobHeader, *merkletree.Proof, error) {
args := c.Called(socket, batchHeaderHash, blobIndex)
var hashes [][]byte
if args.Get(1) != nil {
Expand All @@ -46,7 +46,7 @@ func (c *MockNodeClient) GetBlobHeader(ctx context.Context, socket string, batch
func (c *MockNodeClient) GetChunks(
ctx context.Context,
opID core.OperatorID,
opInfo *core.IndexedOperatorInfo,
opInfo *core.OperatorInfo,
batchHeaderHash [32]byte,
blobIndex uint32,
quorumID core.QuorumID,
Expand Down
10 changes: 5 additions & 5 deletions api/clients/node_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@ type RetrievedChunks struct {
}

type NodeClient interface {
GetBlobHeader(ctx context.Context, socket string, batchHeaderHash [32]byte, blobIndex uint32) (*core.BlobHeader, *merkletree.Proof, error)
GetChunks(ctx context.Context, opID core.OperatorID, opInfo *core.IndexedOperatorInfo, batchHeaderHash [32]byte, blobIndex uint32, quorumID core.QuorumID, chunksChan chan RetrievedChunks)
GetBlobHeader(ctx context.Context, socket core.OperatorSocket, batchHeaderHash [32]byte, blobIndex uint32) (*core.BlobHeader, *merkletree.Proof, error)
GetChunks(ctx context.Context, opID core.OperatorID, opInfo *core.OperatorInfo, batchHeaderHash [32]byte, blobIndex uint32, quorumID core.QuorumID, chunksChan chan RetrievedChunks)
}

type client struct {
Expand All @@ -36,12 +36,12 @@ func NewNodeClient(timeout time.Duration) NodeClient {

func (c client) GetBlobHeader(
ctx context.Context,
socket string,
socket core.OperatorSocket,
batchHeaderHash [32]byte,
blobIndex uint32,
) (*core.BlobHeader, *merkletree.Proof, error) {
conn, err := grpc.NewClient(
core.OperatorSocket(socket).GetV1RetrievalSocket(),
socket.GetV1RetrievalSocket(),
grpc.WithTransportCredentials(insecure.NewCredentials()),
)
if err != nil {
Expand Down Expand Up @@ -79,7 +79,7 @@ func (c client) GetBlobHeader(
func (c client) GetChunks(
ctx context.Context,
opID core.OperatorID,
opInfo *core.IndexedOperatorInfo,
opInfo *core.OperatorInfo,
batchHeaderHash [32]byte,
blobIndex uint32,
quorumID core.QuorumID,
Expand Down
18 changes: 9 additions & 9 deletions api/clients/retrieval_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ type BlobChunks struct {

type retrievalClient struct {
logger logging.Logger
indexedChainState core.IndexedChainState
chainState core.ChainState
assignmentCoordinator core.AssignmentCoordinator
nodeClient NodeClient
verifier encoding.Verifier
Expand All @@ -63,15 +63,15 @@ type retrievalClient struct {
// NewRetrievalClient creates a new retrieval client.
func NewRetrievalClient(
logger logging.Logger,
chainState core.IndexedChainState,
chainState core.ChainState,
assignmentCoordinator core.AssignmentCoordinator,
nodeClient NodeClient,
verifier encoding.Verifier,
numConnections int) (RetrievalClient, error) {

return &retrievalClient{
logger: logger.With("component", "RetrievalClient"),
indexedChainState: chainState,
chainState: chainState,
assignmentCoordinator: assignmentCoordinator,
nodeClient: nodeClient,
verifier: verifier,
Expand Down Expand Up @@ -104,11 +104,12 @@ func (r *retrievalClient) RetrieveBlobChunks(ctx context.Context,
batchRoot [32]byte,
quorumID core.QuorumID) (*BlobChunks, error) {

indexedOperatorState, err := r.indexedChainState.GetIndexedOperatorState(ctx, referenceBlockNumber, []core.QuorumID{quorumID})
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
}
operators, ok := indexedOperatorState.Operators[quorumID]
operators, ok := operatorState.Operators[quorumID]
if !ok {
return nil, fmt.Errorf("no quorum with ID: %d", quorumID)
}
Expand All @@ -118,7 +119,7 @@ func (r *retrievalClient) RetrieveBlobChunks(ctx context.Context,
var proof *merkletree.Proof
var proofVerified bool
for opID := range operators {
opInfo := indexedOperatorState.IndexedOperators[opID]
opInfo := operators[opID]
blobHeader, proof, err = r.nodeClient.GetBlobHeader(ctx, opInfo.Socket, batchHeaderHash, blobIndex)
if err != nil {
// try another operator
Expand Down Expand Up @@ -172,7 +173,7 @@ func (r *retrievalClient) RetrieveBlobChunks(ctx context.Context,
return nil, err
}

assignments, info, err := r.assignmentCoordinator.GetAssignments(indexedOperatorState.OperatorState, blobHeader.Length, quorumHeader)
assignments, info, err := r.assignmentCoordinator.GetAssignments(operatorState, blobHeader.Length, quorumHeader)
if err != nil {
return nil, errors.New("failed to get assignments")
}
Expand All @@ -181,8 +182,7 @@ func (r *retrievalClient) RetrieveBlobChunks(ctx context.Context,
chunksChan := make(chan RetrievedChunks, len(operators))
pool := workerpool.New(r.numConnections)
for opID := range operators {
opID := opID
opInfo := indexedOperatorState.IndexedOperators[opID]
opInfo := operators[opID]
pool.Submit(func() {
r.nodeClient.GetChunks(ctx, opID, opInfo, batchHeaderHash, blobIndex, quorumID, chunksChan)
})
Expand Down
7 changes: 1 addition & 6 deletions api/clients/retrieval_client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,12 +100,7 @@ func setup(t *testing.T) {
indexer = &indexermock.MockIndexer{}
indexer.On("Index").Return(nil).Once()

ics, err := coreindexer.NewIndexedChainState(chainState, indexer)
if err != nil {
panic("failed to create a new indexed chain state")
}

retrievalClient, err = clients.NewRetrievalClient(logger, ics, coordinator, nodeClient, v, 2)
retrievalClient, err = clients.NewRetrievalClient(logger, chainState, coordinator, nodeClient, v, 2)
if err != nil {
panic("failed to create a new retrieval client")
}
Expand Down
33 changes: 16 additions & 17 deletions api/clients/v2/retrieval_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,11 +34,11 @@ type RetrievalClient interface {
}

type retrievalClient struct {
logger logging.Logger
ethClient core.Reader
indexedChainState core.IndexedChainState
verifier encoding.Verifier
numConnections int
logger logging.Logger
ethClient core.Reader
chainState core.ChainState
verifier encoding.Verifier
numConnections int
}

var _ RetrievalClient = &retrievalClient{}
Expand All @@ -47,16 +47,16 @@ var _ RetrievalClient = &retrievalClient{}
func NewRetrievalClient(
logger logging.Logger,
ethClient core.Reader,
chainState core.IndexedChainState,
chainState core.ChainState,
verifier encoding.Verifier,
numConnections int,
) RetrievalClient {
return &retrievalClient{
logger: logger.With("component", "RetrievalClient"),
ethClient: ethClient,
indexedChainState: chainState,
verifier: verifier,
numConnections: numConnections,
logger: logger.With("component", "RetrievalClient"),
ethClient: ethClient,
chainState: chainState,
verifier: verifier,
numConnections: numConnections,
}
}

Expand All @@ -75,11 +75,11 @@ func (r *retrievalClient) GetBlob(
return nil, err
}

indexedOperatorState, err := r.indexedChainState.GetIndexedOperatorState(ctx, uint(referenceBlockNumber), []core.QuorumID{quorumID})
operatorState, err := r.chainState.GetOperatorStateWithSocket(ctx, uint(referenceBlockNumber), []core.QuorumID{quorumID})
if err != nil {
return nil, err
}
operators, ok := indexedOperatorState.Operators[quorumID]
operators, ok := operatorState.Operators[quorumID]
if !ok {
return nil, fmt.Errorf("no quorum with ID: %d", quorumID)
}
Expand All @@ -99,7 +99,7 @@ func (r *retrievalClient) GetBlob(
return nil, err
}

assignments, err := corev2.GetAssignments(indexedOperatorState.OperatorState, blobParam, quorumID)
assignments, err := corev2.GetAssignments(operatorState, blobParam, quorumID)
if err != nil {
return nil, errors.New("failed to get assignments")
}
Expand All @@ -108,8 +108,7 @@ func (r *retrievalClient) GetBlob(
chunksChan := make(chan clients.RetrievedChunks, len(operators))
pool := workerpool.New(r.numConnections)
for opID := range operators {
opID := opID
opInfo := indexedOperatorState.IndexedOperators[opID]
opInfo := operatorState.Operators[quorumID][opID]
pool.Submit(func() {
r.getChunksFromOperator(ctx, opID, opInfo, blobKey, quorumID, chunksChan)
})
Expand Down Expand Up @@ -161,7 +160,7 @@ func (r *retrievalClient) GetBlob(
func (r *retrievalClient) getChunksFromOperator(
ctx context.Context,
opID core.OperatorID,
opInfo *core.IndexedOperatorInfo,
opInfo *core.OperatorInfo,
blobKey corev2.BlobKey,
quorumID core.QuorumID,
chunksChan chan clients.RetrievedChunks,
Expand Down
2 changes: 1 addition & 1 deletion contracts/bindings/BLSApkRegistry/binding.go
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion contracts/bindings/EigenDACertVerifier/binding.go
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion contracts/bindings/EigenDAServiceManager/binding.go
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion contracts/bindings/EjectionManager/binding.go
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion contracts/bindings/MockRollup/binding.go
View file
Open in desktop

Large diffs are not rendered by default.

99 changes: 97 additions & 2 deletions contracts/bindings/OperatorStateRetriever/binding.go
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion contracts/bindings/RegistryCoordinator/binding.go
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion contracts/bindings/SocketRegistry/binding.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading
Loading