Implement validator payload retrieval

api/clients/v2/config.go

@@ -1,7 +1,7 @@
 package clients
 
 import (
-	"fmt"
+	"errors"
 	"time"
 
 	"github.com/Layr-Labs/eigenda/api/clients/codecs"
@@ -54,6 +54,24 @@ type RelayPayloadRetrieverConfig struct {
 	RelayTimeout time.Duration
 }
 
+// DistributedPayloadRetrieverConfig contains an embedded PayloadClientConfig, plus all additional configuration values
+// needed by a DistributedPayloadRetriever
+type DistributedPayloadRetrieverConfig struct {
+	PayloadClientConfig
+
+	// The timeout duration for calls to retrieve blobs from a given quorum.
+	RetrievalTimeout time.Duration
+
+	// The address of the BlsOperatorStateRetriever contract
+	BlsOperatorStateRetrieverAddr string
+
+	// The address of the EigenDAServiceManager contract
+	EigenDAServiceManagerAddr string
+
+	// The number of simultaneous connections to use when fetching chunks during distributed retrieval
+	ConnectionCount uint
+}
+
 // PayloadDisperserConfig contains an embedded PayloadClientConfig, plus all additional configuration values needed
 // by a PayloadDisperser
 type PayloadDisperserConfig struct {
@@ -105,11 +123,11 @@ func (cc *PayloadClientConfig) checkAndSetDefaults() error {
 	// BlobEncodingVersion may be 0, so don't do anything
 
 	if cc.EthRpcUrl == "" {
-		return fmt.Errorf("EthRpcUrl is required")
+		return errors.New("EthRpcUrl is required")
 	}
 
 	if cc.EigenDACertVerifierAddr == "" {
-		return fmt.Errorf("EigenDACertVerifierAddr is required")
+		return errors.New("EigenDACertVerifierAddr is required")
 	}
 
 	// Nothing to do for PayloadPolynomialForm
@@ -154,6 +172,51 @@ func (rc *RelayPayloadRetrieverConfig) checkAndSetDefaults() error {
 	return nil
 }
 
+// GetDefaultDistributedPayloadRetrieverConfig creates a DistributedPayloadRetrieverConfig with default values
+//
+// NOTE: The following fields do not have defined defaults and must always be specifically configured:
+// - EthRpcUrl
+// - EigenDACertVerifierAddr
+// - BlsOperatorStateRetrieverAddr
+// - EigenDAServiceManagerAddr
+func GetDefaultDistributedPayloadRetrieverConfig() *DistributedPayloadRetrieverConfig {
+	return &DistributedPayloadRetrieverConfig{
+		PayloadClientConfig: *getDefaultPayloadClientConfig(),
+		RetrievalTimeout:    20 * time.Second,
+		ConnectionCount:     10,
+	}
+}
+
+// checkAndSetDefaults checks an existing config struct and performs the following actions:
+//
+// 1. If a config value is 0, and a 0 value makes sense, do nothing.
+// 2. If a config value is 0, but a 0 value doesn't make sense and a default value is defined, then set it to the default.
+// 3. If a config value is 0, but a 0 value doesn't make sense and a default value isn't defined, return an error.
+func (rc *DistributedPayloadRetrieverConfig) checkAndSetDefaults() error {
+	err := rc.PayloadClientConfig.checkAndSetDefaults()
+	if err != nil {
+		return err
+	}
+
+	if rc.BlsOperatorStateRetrieverAddr == "" {
+		return errors.New("BlsOperatorStateRetrieverAddr is required")
+	}
+
+	if rc.EigenDAServiceManagerAddr == "" {
+		return errors.New("EigenDAServiceManagerAddr is required")
+	}
+
+	defaultConfig := GetDefaultDistributedPayloadRetrieverConfig()
+	if rc.RetrievalTimeout == 0 {
+		rc.RetrievalTimeout = defaultConfig.RetrievalTimeout
+	}
+	if rc.ConnectionCount == 0 {
+		rc.ConnectionCount = defaultConfig.ConnectionCount
+	}
+
+	return nil
+}
+
 // GetDefaultPayloadDisperserConfig creates a PayloadDisperserConfig with default values
 //
 // NOTE: EthRpcUrl and EigenDACertVerifierAddr do not have defined defaults. These must always be specifically configured.

api/clients/v2/distributed_payload_retriever.go

@@ -0,0 +1,210 @@
+package clients
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/Layr-Labs/eigenda/api/clients/codecs"
+	"github.com/Layr-Labs/eigenda/api/clients/v2/verification"
+	"github.com/Layr-Labs/eigenda/common/geth"
+	"github.com/Layr-Labs/eigenda/core"
+	"github.com/Layr-Labs/eigenda/core/eth"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
+	corev2 "github.com/Layr-Labs/eigenda/core/v2"
+	"github.com/Layr-Labs/eigenda/encoding"
+	"github.com/Layr-Labs/eigenda/encoding/kzg"
+	"github.com/Layr-Labs/eigenda/encoding/kzg/verifier"
+	"github.com/Layr-Labs/eigensdk-go/logging"
+	"github.com/consensys/gnark-crypto/ecc/bn254"
+	gethcommon "github.com/ethereum/go-ethereum/common"
+)
+
+// DistributedPayloadRetriever provides the ability to get payloads from the EigenDA nodes directly
+//
+// This struct is goroutine safe.
+type DistributedPayloadRetriever struct {
+	logger          logging.Logger
+	config          DistributedPayloadRetrieverConfig
+	codec           codecs.BlobCodec
+	retrievalClient RetrievalClient
+	g1Srs           []bn254.G1Affine
+}
+
+var _ PayloadRetriever = &DistributedPayloadRetriever{}
+
+// BuildDistributedPayloadRetriever builds a DistributedPayloadRetriever from config structs.
+func BuildDistributedPayloadRetriever(
+	logger logging.Logger,
+	distributedPayloadRetrieverConfig DistributedPayloadRetrieverConfig,
+	ethConfig geth.EthClientConfig,
+	thegraphConfig thegraph.Config,
+	kzgConfig kzg.KzgConfig,
+) (*DistributedPayloadRetriever, error) {
+	err := distributedPayloadRetrieverConfig.checkAndSetDefaults()
+	if err != nil {
+		return nil, fmt.Errorf("check and set config defaults: %w", err)
+	}
+
+	ethClient, err := geth.NewClient(ethConfig, gethcommon.Address{}, 0, logger)
+	if err != nil {
+		return nil, fmt.Errorf("new eth client: %w", err)
+	}
+
+	reader, err := eth.NewReader(
+		logger,
+		ethClient,
+		distributedPayloadRetrieverConfig.BlsOperatorStateRetrieverAddr,
+		distributedPayloadRetrieverConfig.EigenDAServiceManagerAddr)
+	if err != nil {
+		return nil, fmt.Errorf("new reader: %w", err)
+	}
+
+	chainState := eth.NewChainState(reader, ethClient)
+	indexedChainState := thegraph.MakeIndexedChainState(thegraphConfig, chainState, logger)
+
+	kzgVerifier, err := verifier.NewVerifier(&kzgConfig, nil)
+	if err != nil {
+		return nil, fmt.Errorf("new verifier: %w", err)
+	}
+
+	retrievalClient := NewRetrievalClient(
+		logger,
+		reader,
+		indexedChainState,
+		kzgVerifier,
+		int(distributedPayloadRetrieverConfig.ConnectionCount))
+
+	codec, err := codecs.CreateCodec(
+		distributedPayloadRetrieverConfig.PayloadPolynomialForm,
+		distributedPayloadRetrieverConfig.BlobEncodingVersion)
+	if err != nil {
+		return nil, fmt.Errorf("create codec: %w", err)
+	}
+
+	return &DistributedPayloadRetriever{
+		logger:          logger,
+		config:          distributedPayloadRetrieverConfig,
+		codec:           codec,
+		retrievalClient: retrievalClient,
+		g1Srs:           kzgVerifier.Srs.G1,
+	}, nil
+}
+
+// NewDistributedPayloadRetriever creates a new DistributedPayloadRetriever from already constructed objects
+func NewDistributedPayloadRetriever(
+	logger logging.Logger,
+	config DistributedPayloadRetrieverConfig,
+	codec codecs.BlobCodec,
+	retrievalClient RetrievalClient,
+	g1Srs []bn254.G1Affine,
+) (*DistributedPayloadRetriever, error) {
+	err := config.checkAndSetDefaults()
+	if err != nil {
+		return nil, fmt.Errorf("check and set config defaults: %w", err)
+	}
+
+	return &DistributedPayloadRetriever{
+		logger:          logger,
+		config:          config,
+		codec:           codec,
+		retrievalClient: retrievalClient,
+		g1Srs:           g1Srs,
+	}, nil
+}
+
+// GetPayload iteratively attempts to retrieve a given blob from the quorums listed in the EigenDACert.
+//
+// If the blob is successfully retrieved, then the blob verified against the EigenDACert. If the verification succeeds,
+// the blob is decoded to yield the payload (the original user data, with no padding or any modification), and the
+// payload is returned.
+func (pr *DistributedPayloadRetriever) GetPayload(
+	ctx context.Context,
+	eigenDACert *verification.EigenDACert,
+) ([]byte, error) {
+
+	blobKey, err := eigenDACert.ComputeBlobKey()
+	if err != nil {
+		return nil, fmt.Errorf("compute blob key: %w", err)
+	}
+
+	blobHeader := eigenDACert.BlobInclusionInfo.BlobCertificate.BlobHeader
+	commitment, err := verification.BlobCommitmentsBindingToInternal(&blobHeader.Commitment)
+	if err != nil {
+		return nil, fmt.Errorf("convert commitments binding to internal: %w", err)
+	}
+
+	// TODO (litt3): Add a feature which keeps chunks from previous quorums, and just fills in gaps
+	for _, quorumID := range blobHeader.QuorumNumbers {
+		blobBytes, err := pr.getBlobWithTimeout(
+			ctx,
+			*blobKey,
+			blobHeader.Version,
+			*commitment,
+			eigenDACert.BatchHeader.ReferenceBlockNumber,
+			quorumID)
+
+		if err != nil {
+			pr.logger.Warn(
+				"blob couldn't be retrieved from quorum",
+				"blobKey", blobKey.Hex(),
+				"quorumId", quorumID,
+				"error", err)
+			continue
+		}
+
+		err = verification.CheckBlobLength(blobBytes, commitment.Length)
+		if err != nil {
+			pr.logger.Warn("check blob length", "blobKey", blobKey.Hex(), "quorumID", quorumID, "error", err)
+			continue
+		}
+
+		valid, err := verification.GenerateAndCompareBlobCommitment(pr.g1Srs, blobBytes, commitment.Commitment)
+		if err != nil {
+			pr.logger.Warn(
+				"generate and compare blob commitment",
+				"blobKey", blobKey.Hex(), "quorumID", quorumID, "error", err)
+			continue
+		}
+		if !valid {
+			pr.logger.Warn("cert is invalid", "blobKey", blobKey.Hex(), "quorumID", quorumID)
+			continue
+		}
+
+		payload, err := pr.codec.DecodeBlob(blobBytes)
+		if err != nil {
+			pr.logger.Error(
+				`Cert verification was successful, but decode blob failed!
+					This is likely a problem with the local blob codec configuration,
+					but could potentially indicate a maliciously generated blob certificate.
+					It should not be possible for an honestly generated certificate to verify
+					for an invalid blob!`,
+				"blobKey", blobKey.Hex(), "quorumID", quorumID, "eigenDACert", eigenDACert, "error", err)
+			return nil, fmt.Errorf("decode blob: %w", err)
+		}
+
+		return payload, nil
+	}
+
+	return nil, fmt.Errorf("unable to retrieve payload from quorums %v", blobHeader.QuorumNumbers)
+}
+
+// getBlobWithTimeout attempts to get a blob from a given quorum, and times out based on config.RetrievalTimeout
+func (pr *DistributedPayloadRetriever) getBlobWithTimeout(
+	ctx context.Context,
+	blobKey corev2.BlobKey,
+	blobVersion corev2.BlobVersion,
+	blobCommitments encoding.BlobCommitments,
+	referenceBlockNumber uint32,
+	quorumID core.QuorumID) ([]byte, error) {
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, pr.config.RetrievalTimeout)
+	defer cancel()
+
+	return pr.retrievalClient.GetBlob(
+		timeoutCtx,
+		blobKey,
+		blobVersion,
+		blobCommitments,
+		uint64(referenceBlockNumber),
+		quorumID)
+}