inabox test retrieval from network

api/clients/v2/retrieval_client.go

@@ -21,7 +21,7 @@ import (
 // To retrieve a blob from the relay, use RelayClient instead.
 type RetrievalClient interface {
 	// GetBlob downloads chunks of a blob from operator network and reconstructs the blob.
-	GetBlob(ctx context.Context, blobHeader corev2.BlobHeader, referenceBlockNumber uint64, quorumID core.QuorumID) ([]byte, error)
+	GetBlob(ctx context.Context, blobHeader *corev2.BlobHeader, referenceBlockNumber uint64, quorumID core.QuorumID) ([]byte, error)
 }
 
 type retrievalClient struct {
@@ -49,7 +49,11 @@ func NewRetrievalClient(
 	}
 }
 
-func (r *retrievalClient) GetBlob(ctx context.Context, blobHeader corev2.BlobHeader, referenceBlockNumber uint64, quorumID core.QuorumID) ([]byte, error) {
+func (r *retrievalClient) GetBlob(ctx context.Context, blobHeader *corev2.BlobHeader, referenceBlockNumber uint64, quorumID core.QuorumID) ([]byte, error) {
+	if blobHeader == nil {
+		return nil, errors.New("blob header is nil")
+	}
+
 	blobKey, err := blobHeader.BlobKey()
 	if err != nil {
 		return nil, err

inabox/deploy/config.go

@@ -455,6 +455,8 @@ func (env *Config) generateRetrieverVars(ind int, key string, graphUrl, logPath,
 		RETRIEVER_VERBOSE:             "true",
 		RETRIEVER_CACHE_ENCODED_BLOBS: "false",
 		RETRIEVER_GRAPH_URL:           graphUrl,
+		RETRIEVER_GRAPH_BACKOFF:       "1s",
+		RETRIEVER_GRAPH_MAX_RETRIES:   "3",
 
 		RETRIEVER_INDEXER_PULL_INTERVAL: "1s",
 	}

inabox/tests/integration_suite_test.go

@@ -11,17 +11,17 @@ import (
 	"time"
 
 	"github.com/Layr-Labs/eigenda/api/clients"
+	clientsv2 "github.com/Layr-Labs/eigenda/api/clients/v2"
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/geth"
 	rollupbindings "github.com/Layr-Labs/eigenda/contracts/bindings/MockRollup"
 	"github.com/Layr-Labs/eigenda/core"
 	"github.com/Layr-Labs/eigenda/core/eth"
-	coreindexer "github.com/Layr-Labs/eigenda/core/indexer"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
 	corev2 "github.com/Layr-Labs/eigenda/core/v2"
 	"github.com/Layr-Labs/eigenda/encoding/kzg"
 	"github.com/Layr-Labs/eigenda/encoding/kzg/verifier"
 	"github.com/Layr-Labs/eigenda/inabox/deploy"
-	"github.com/Layr-Labs/eigenda/indexer"
 	"github.com/Layr-Labs/eigensdk-go/logging"
 	gcommon "github.com/ethereum/go-ethereum/common"
 	ethrpc "github.com/ethereum/go-ethereum/rpc"
@@ -48,6 +48,7 @@ var (
 	rpcClient           common.RPCEthClient
 	mockRollup          *rollupbindings.ContractMockRollup
 	retrievalClient     clients.RetrievalClient
+	retrievalClientV2   clientsv2.RetrievalClient
 	numConfirmations    int = 3
 	numRetries              = 0
 	relays                  = map[corev2.RelayKey]string{}
@@ -190,33 +191,31 @@ func setupRetrievalClient(testConfig *deploy.Config) error {
 		return err
 	}
 
-	indexer, err := coreindexer.CreateNewIndexer(
-		&indexer.Config{
-			PullInterval: 100 * time.Millisecond,
-		},
-		ethClient,
-		rpcClient,
-		testConfig.Retriever.RETRIEVER_EIGENDA_SERVICE_MANAGER,
-		logger,
-	)
+	graphBackoff, err := time.ParseDuration(testConfig.Retriever.RETRIEVER_GRAPH_BACKOFF)
 	if err != nil {
 		return err
 	}
-
-	ics, err := coreindexer.NewIndexedChainState(cs, indexer)
+	maxRetries, err := strconv.Atoi(testConfig.Retriever.RETRIEVER_GRAPH_MAX_RETRIES)
 	if err != nil {
 		return err
 	}
+	ics := thegraph.MakeIndexedChainState(thegraph.Config{
+		Endpoint:     testConfig.Retriever.RETRIEVER_GRAPH_URL,
+		PullInterval: graphBackoff,
+		MaxRetries:   maxRetries,
+	}, cs, logger)
 
 	retrievalClient, err = clients.NewRetrievalClient(logger, ics, agn, nodeClient, v, 10)
 	if err != nil {
 		return err
 	}
+	chainReader, err := eth.NewReader(logger, ethClient, testConfig.Retriever.RETRIEVER_BLS_OPERATOR_STATE_RETRIVER, testConfig.Retriever.RETRIEVER_EIGENDA_SERVICE_MANAGER)
+	if err != nil {
+		return err
+	}
+	retrievalClientV2 = clientsv2.NewRetrievalClient(logger, chainReader, ics, v, 10)
 
-	var ctx context.Context
-	ctx, cancel = context.WithCancel(context.Background())
-
-	return indexer.Index(ctx)
+	return ics.Start(context.Background())
 }
 
 var _ = AfterSuite(func() {

inabox/tests/integration_v2_test.go

@@ -1,11 +1,13 @@
 package integration_test
 
 import (
+	"bytes"
 	"context"
 	"crypto/rand"
 	"time"
 
 	clients "github.com/Layr-Labs/eigenda/api/clients/v2"
+	commonpb "github.com/Layr-Labs/eigenda/api/grpc/common/v2"
 	disperserpb "github.com/Layr-Labs/eigenda/api/grpc/disperser/v2"
 	"github.com/Layr-Labs/eigenda/core"
 	auth "github.com/Layr-Labs/eigenda/core/auth/v2"
@@ -62,6 +64,8 @@ var _ = Describe("Inabox v2 Integration", func() {
 		var reply2 *disperserpb.BlobStatusReply
 		var blobCert1 *corev2.BlobCertificate
 		var blobCert2 *corev2.BlobCertificate
+		var batchHeader1 *commonpb.BatchHeader
+		var batchHeader2 *commonpb.BatchHeader
 		for loop := true; loop; {
 			select {
 			case <-ctx.Done():
@@ -83,10 +87,10 @@ var _ = Describe("Inabox v2 Integration", func() {
 					continue
 				}
 
-				batchHeader := reply1.GetSignedBatch().GetHeader()
-				Expect(batchHeader).To(Not(BeNil()))
-				Expect(batchHeader.GetBatchRoot()).To(Not(BeNil()))
-				Expect(batchHeader.GetReferenceBlockNumber()).To(BeNumerically(">", 0))
+				batchHeader1 = reply1.GetSignedBatch().GetHeader()
+				Expect(batchHeader1).To(Not(BeNil()))
+				Expect(batchHeader1.GetBatchRoot()).To(Not(BeNil()))
+				Expect(batchHeader1.GetReferenceBlockNumber()).To(BeNumerically(">", 0))
 				attestation := reply1.GetSignedBatch().GetAttestation()
 				Expect(attestation).To(Not(BeNil()))
 				blobVerification := reply1.GetBlobVerificationInfo()
@@ -102,14 +106,14 @@ var _ = Describe("Inabox v2 Integration", func() {
 				Expect(err).To(BeNil())
 				_, err = blobCert1.BlobHeader.BlobKey()
 				Expect(err).To(BeNil())
-				verified, err := merkletree.VerifyProofUsing(certHash[:], false, proof, [][]byte{batchHeader.BatchRoot}, keccak256.New())
+				verified, err := merkletree.VerifyProofUsing(certHash[:], false, proof, [][]byte{batchHeader1.BatchRoot}, keccak256.New())
 				Expect(err).To(BeNil())
 				Expect(verified).To(BeTrue())
 
-				batchHeader = reply2.GetSignedBatch().GetHeader()
-				Expect(batchHeader).To(Not(BeNil()))
-				Expect(batchHeader.GetBatchRoot()).To(Not(BeNil()))
-				Expect(batchHeader.GetReferenceBlockNumber()).To(BeNumerically(">", 0))
+				batchHeader2 = reply2.GetSignedBatch().GetHeader()
+				Expect(batchHeader2).To(Not(BeNil()))
+				Expect(batchHeader2.GetBatchRoot()).To(Not(BeNil()))
+				Expect(batchHeader2.GetReferenceBlockNumber()).To(BeNumerically(">", 0))
 				attestation = reply2.GetSignedBatch().GetAttestation()
 				Expect(attestation).To(Not(BeNil()))
 				blobVerification = reply2.GetBlobVerificationInfo()
@@ -123,7 +127,7 @@ var _ = Describe("Inabox v2 Integration", func() {
 				Expect(err).To(BeNil())
 				certHash, err = blobCert2.Hash()
 				Expect(err).To(BeNil())
-				verified, err = merkletree.VerifyProofUsing(certHash[:], false, proof, [][]byte{batchHeader.BatchRoot}, keccak256.New())
+				verified, err = merkletree.VerifyProofUsing(certHash[:], false, proof, [][]byte{batchHeader2.BatchRoot}, keccak256.New())
 				Expect(err).To(BeNil())
 				Expect(verified).To(BeTrue())
 				// TODO(ian-shim): verify the blob onchain using a mock rollup contract
@@ -136,8 +140,6 @@ var _ = Describe("Inabox v2 Integration", func() {
 			Sockets: relays,
 		}, logger)
 		Expect(err).To(BeNil())
-		ctx, cancel = context.WithTimeout(context.Background(), time.Second*5)
-		defer cancel()
 
 		blob1Relays := make(map[corev2.RelayKey]struct{}, 0)
 		blob2Relays := make(map[corev2.RelayKey]struct{}, 0)
@@ -165,6 +167,21 @@ var _ = Describe("Inabox v2 Integration", func() {
 			}
 		}
 
-		// TODO(ian-shim): test retrieval from DA nodes via retrieval client
+		b, err := retrievalClientV2.GetBlob(ctx, blobCert1.BlobHeader, batchHeader1.ReferenceBlockNumber, 0)
+		Expect(err).To(BeNil())
+		restored := bytes.TrimRight(b, "\x00")
+		Expect(restored).To(Equal(paddedData1))
+		b, err = retrievalClientV2.GetBlob(ctx, blobCert1.BlobHeader, batchHeader1.ReferenceBlockNumber, 1)
+		restored = bytes.TrimRight(b, "\x00")
+		Expect(err).To(BeNil())
+		Expect(restored).To(Equal(paddedData1))
+		b, err = retrievalClientV2.GetBlob(ctx, blobCert2.BlobHeader, batchHeader2.ReferenceBlockNumber, 0)
+		restored = bytes.TrimRight(b, "\x00")
+		Expect(err).To(BeNil())
+		Expect(restored).To(Equal(paddedData2))
+		b, err = retrievalClientV2.GetBlob(ctx, blobCert2.BlobHeader, batchHeader2.ReferenceBlockNumber, 1)
+		restored = bytes.TrimRight(b, "\x00")
+		Expect(err).NotTo(BeNil())
+		Expect(restored).To(BeNil())
 	})
 })