-
Notifications
You must be signed in to change notification settings - Fork 187
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.
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
Add EigenDA client using handshake based authentication #551
Merged
Merged
Changes from all commits
Commits
Show all changes
13 commits
Select commit
Hold shift + click to select a range
4dbfa2a
Add EigenDA client using handshake based authentication
teddyknox 994c5fd
Respond to PR feedback with updates
teddyknox 715eb7e
Update to depend on existing low-level EigenDA client
teddyknox 79176c0
Add test suite for high-level EigenDA client
teddyknox 958d6ac
Move clients/ to api/clients/
teddyknox 2d0f899
Modularize EigenDA client encoding
teddyknox ce81fa7
Add test for codecs
teddyknox 5e4374a
Remove cert concept in favor of BlobInfo
teddyknox 7530b14
Fix dockerfiles with clients moved
teddyknox b97ffbc
Fix lint error
teddyknox 244c88d
Remove default quorums validation logic
teddyknox 44bfbf9
Add more validation for EigenDAClientConfig checks
teddyknox 179fdca
Update name of Check function
teddyknox File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,51 @@ | ||
package clients | ||
|
||
import ( | ||
"fmt" | ||
"time" | ||
) | ||
|
||
type EigenDAClientConfig struct { | ||
// RPC is the HTTP provider URL for the Data Availability node. | ||
RPC string | ||
|
||
// The total amount of time that the client will spend waiting for EigenDA to confirm a blob | ||
StatusQueryTimeout time.Duration | ||
|
||
// The amount of time to wait between status queries of a newly dispersed blob | ||
StatusQueryRetryInterval time.Duration | ||
|
||
// The total amount of time that the client will waiting for a response from the EigenDA disperser | ||
ResponseTimeout time.Duration | ||
|
||
// The quorum IDs to write blobs to using this client. Should not include default quorums 0 or 1. | ||
CustomQuorumIDs []uint | ||
|
||
// Signer private key in hex encoded format. This key should not be associated with an Ethereum address holding any funds. | ||
SignerPrivateKeyHex string | ||
|
||
// Whether to disable TLS for an insecure connection when connecting to a local EigenDA disperser instance. | ||
DisableTLS bool | ||
|
||
// The blob encoding version to use when writing blobs from the high level interface. | ||
PutBlobEncodingVersion BlobEncodingVersion | ||
} | ||
|
||
func (c *EigenDAClientConfig) CheckAndSetDefaults() error { | ||
if c.StatusQueryRetryInterval == 0 { | ||
c.StatusQueryRetryInterval = 5 * time.Second | ||
} | ||
if c.StatusQueryTimeout == 0 { | ||
c.StatusQueryTimeout = 25 * time.Minute | ||
} | ||
if c.ResponseTimeout == 0 { | ||
c.ResponseTimeout = 30 * time.Second | ||
} | ||
if len(c.SignerPrivateKeyHex) != 64 { | ||
return fmt.Errorf("EigenDAClientConfig.SignerPrivateKeyHex should be 64 hex characters long, should not have 0x prefix") | ||
} | ||
if len(c.RPC) == 0 { | ||
return fmt.Errorf("EigenDAClientConfig.RPC not set") | ||
} | ||
return nil | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,175 @@ | ||
package clients | ||
|
||
import ( | ||
"context" | ||
"encoding/base64" | ||
"encoding/hex" | ||
"fmt" | ||
"net" | ||
"time" | ||
|
||
grpcdisperser "github.com/Layr-Labs/eigenda/api/grpc/disperser" | ||
"github.com/Layr-Labs/eigenda/core/auth" | ||
"github.com/Layr-Labs/eigenda/disperser" | ||
"github.com/ethereum/go-ethereum/log" | ||
) | ||
|
||
type IEigenDAClient interface { | ||
GetBlob(ctx context.Context, BatchHeaderHash []byte, BlobIndex uint32) ([]byte, error) | ||
PutBlob(ctx context.Context, txData []byte) (*grpcdisperser.BlobInfo, error) | ||
} | ||
|
||
type EigenDAClient struct { | ||
Config EigenDAClientConfig | ||
Log log.Logger | ||
Client DisperserClient | ||
PutCodec BlobCodec | ||
} | ||
|
||
var _ IEigenDAClient = EigenDAClient{} | ||
|
||
func NewEigenDAClient(log log.Logger, config EigenDAClientConfig) (*EigenDAClient, error) { | ||
err := config.CheckAndSetDefaults() | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
||
host, port, err := net.SplitHostPort(config.RPC) | ||
if err != nil { | ||
return nil, fmt.Errorf("failed to parse EigenDA RPC: %w", err) | ||
} | ||
|
||
signer := auth.NewLocalBlobRequestSigner(config.SignerPrivateKeyHex) | ||
llConfig := NewConfig(host, port, config.ResponseTimeout, !config.DisableTLS) | ||
llClient := NewDisperserClient(llConfig, signer) | ||
|
||
codec, err := BlobEncodingVersionToCodec(config.PutBlobEncodingVersion) | ||
if err != nil { | ||
return nil, fmt.Errorf("error initializing EigenDA client: %w", err) | ||
} | ||
|
||
return &EigenDAClient{ | ||
Log: log, | ||
Config: config, | ||
Client: llClient, | ||
PutCodec: codec, | ||
}, nil | ||
} | ||
|
||
func (m EigenDAClient) GetBlob(ctx context.Context, BatchHeaderHash []byte, BlobIndex uint32) ([]byte, error) { | ||
data, err := m.Client.RetrieveBlob(ctx, BatchHeaderHash, BlobIndex) | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
||
if len(data) == 0 { | ||
return nil, fmt.Errorf("blob has length zero") | ||
} | ||
|
||
version := BlobEncodingVersion(data[0]) | ||
codec, err := BlobEncodingVersionToCodec(version) | ||
if err != nil { | ||
return nil, fmt.Errorf("error getting blob: %w", err) | ||
} | ||
|
||
rawData, err := codec.DecodeBlob(data) | ||
if err != nil { | ||
return nil, fmt.Errorf("error getting blob: %w", err) | ||
} | ||
|
||
return rawData, nil | ||
} | ||
|
||
func (m EigenDAClient) PutBlob(ctx context.Context, data []byte) (*grpcdisperser.BlobInfo, error) { | ||
resultChan, errorChan := m.PutBlobAsync(ctx, data) | ||
select { // no timeout here because we depend on the configured timeout in PutBlobAsync | ||
case result := <-resultChan: | ||
return result, nil | ||
case err := <-errorChan: | ||
return nil, err | ||
} | ||
} | ||
|
||
func (m EigenDAClient) PutBlobAsync(ctx context.Context, data []byte) (resultChan chan *grpcdisperser.BlobInfo, errChan chan error) { | ||
resultChan = make(chan *grpcdisperser.BlobInfo, 1) | ||
errChan = make(chan error, 1) | ||
go m.putBlob(ctx, data, resultChan, errChan) | ||
return | ||
} | ||
|
||
func (m EigenDAClient) putBlob(ctx context.Context, rawData []byte, resultChan chan *grpcdisperser.BlobInfo, errChan chan error) { | ||
m.Log.Info("Attempting to disperse blob to EigenDA") | ||
|
||
// encode blob | ||
if m.PutCodec == nil { | ||
errChan <- fmt.Errorf("PutCodec cannot be nil") | ||
return | ||
} | ||
data := m.PutCodec.EncodeBlob(rawData) | ||
|
||
customQuorumNumbers := make([]uint8, len(m.Config.CustomQuorumIDs)) | ||
for i, e := range m.Config.CustomQuorumIDs { | ||
customQuorumNumbers[i] = uint8(e) | ||
} | ||
|
||
// disperse blob | ||
blobStatus, requestID, err := m.Client.DisperseBlobAuthenticated(ctx, data, customQuorumNumbers) | ||
if err != nil { | ||
errChan <- fmt.Errorf("error initializing DisperseBlobAuthenticated() client: %w", err) | ||
return | ||
} | ||
|
||
// process response | ||
if *blobStatus == disperser.Failed { | ||
m.Log.Error("Unable to disperse blob to EigenDA, aborting", "err", err) | ||
errChan <- fmt.Errorf("reply status is %d", blobStatus) | ||
return | ||
} | ||
|
||
base64RequestID := base64.StdEncoding.EncodeToString(requestID) | ||
m.Log.Info("Blob dispersed to EigenDA, now waiting for confirmation", "requestID", base64RequestID) | ||
|
||
ticker := time.NewTicker(m.Config.StatusQueryRetryInterval) | ||
defer ticker.Stop() | ||
|
||
var cancel context.CancelFunc | ||
ctx, cancel = context.WithTimeout(ctx, m.Config.StatusQueryTimeout) | ||
defer cancel() | ||
|
||
for { | ||
select { | ||
case <-ctx.Done(): | ||
errChan <- fmt.Errorf("timed out waiting for EigenDA blob to confirm blob with request id=%s: %w", base64RequestID, ctx.Err()) | ||
return | ||
case <-ticker.C: | ||
statusRes, err := m.Client.GetBlobStatus(ctx, requestID) | ||
if err != nil { | ||
m.Log.Error("Unable to retrieve blob dispersal status, will retry", "requestID", base64RequestID, "err", err) | ||
continue | ||
} | ||
|
||
switch statusRes.Status { | ||
case grpcdisperser.BlobStatus_PROCESSING, grpcdisperser.BlobStatus_DISPERSING: | ||
m.Log.Info("Blob submitted, waiting for dispersal from EigenDA", "requestID", base64RequestID) | ||
case grpcdisperser.BlobStatus_FAILED: | ||
m.Log.Error("EigenDA blob dispersal failed in processing", "requestID", base64RequestID, "err", err) | ||
errChan <- fmt.Errorf("EigenDA blob dispersal failed in processing, requestID=%s: %w", base64RequestID, err) | ||
return | ||
case grpcdisperser.BlobStatus_INSUFFICIENT_SIGNATURES: | ||
m.Log.Error("EigenDA blob dispersal failed in processing with insufficient signatures", "requestID", base64RequestID, "err", err) | ||
errChan <- fmt.Errorf("EigenDA blob dispersal failed in processing with insufficient signatures, requestID=%s: %w", base64RequestID, err) | ||
return | ||
case grpcdisperser.BlobStatus_CONFIRMED: | ||
teddyknox marked this conversation as resolved.
Show resolved
Hide resolved
|
||
m.Log.Info("EigenDA blob confirmed, waiting for finalization", "requestID", base64RequestID) | ||
case grpcdisperser.BlobStatus_FINALIZED: | ||
batchHeaderHashHex := fmt.Sprintf("0x%s", hex.EncodeToString(statusRes.Info.BlobVerificationProof.BatchMetadata.BatchHeaderHash)) | ||
m.Log.Info("Successfully dispersed blob to EigenDA", "requestID", base64RequestID, "batchHeaderHash", batchHeaderHashHex) | ||
resultChan <- statusRes.Info | ||
return | ||
default: | ||
errChan <- fmt.Errorf("EigenDA blob dispersal failed in processing with reply status %d", statusRes.Status) | ||
return | ||
} | ||
} | ||
} | ||
} |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It would be better to rename the method name to make it clear to the caller that this method modifies the content of the receiver (or return a new config instead of modifying the receiver in place).