[v2] s3 Blob Store

common/aws/s3/client.go

@@ -72,6 +72,17 @@ func NewClient(ctx context.Context, cfg commonaws.ClientConfig, logger logging.L
 	return ref, err
 }
 
+func (s *client) CreateBucket(ctx context.Context, bucket string) error {
+	_, err := s.s3Client.CreateBucket(ctx, &s3.CreateBucketInput{
+		Bucket: aws.String(bucket),
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (s *client) DownloadObject(ctx context.Context, bucket string, key string) ([]byte, error) {
 	var partMiBs int64 = 10
 	downloader := manager.NewDownloader(s.s3Client, func(d *manager.Downloader) {

common/aws/s3/s3.go

@@ -3,6 +3,7 @@ package s3
 import "context"
 
 type Client interface {
+	CreateBucket(ctx context.Context, bucket string) error
 	DownloadObject(ctx context.Context, bucket string, key string) ([]byte, error)
 	UploadObject(ctx context.Context, bucket string, key string, data []byte) error
 	DeleteObject(ctx context.Context, bucket string, key string) error

common/mock/s3_client.go

@@ -17,6 +17,10 @@ func NewS3Client() *S3Client {
 	return &S3Client{bucket: make(map[string][]byte)}
 }
 
+func (s *S3Client) CreateBucket(ctx context.Context, bucket string) error {
+	return nil
+}
+
 func (s *S3Client) DownloadObject(ctx context.Context, bucket string, key string) ([]byte, error) {
 	data, ok := s.bucket[key]
 	if !ok {

disperser/common/v2/blobstore/blob_metadata_store_test.go → disperser/common/v2/blobstore/blobstore_test.go

@@ -6,11 +6,11 @@ import (
 	"math/big"
 	"os"
 	"testing"
-	"time"
 
 	"github.com/Layr-Labs/eigenda/common/aws"
 	"github.com/Layr-Labs/eigenda/common/aws/dynamodb"
 	test_utils "github.com/Layr-Labs/eigenda/common/aws/dynamodb/utils"
+	"github.com/Layr-Labs/eigenda/common/aws/s3"
 	"github.com/Layr-Labs/eigenda/encoding"
 	"github.com/Layr-Labs/eigenda/inabox/deploy"
 	"github.com/Layr-Labs/eigensdk-go/logging"
@@ -31,10 +31,13 @@ var (
 	deployLocalStack bool
 	localStackPort   = "4571"
 
+	s3Client          s3.Client
 	dynamoClient      *dynamodb.Client
+	blobStore         *blobstore.BlobStore
 	blobMetadataStore *blobstore.BlobMetadataStore
 
 	UUID              = uuid.New()
+	s3BucketName      = "test-eigenda-blobstore"
 	metadataTableName = fmt.Sprintf("test-BlobMetadata-%v", UUID)
 
 	mockCommitment = encoding.BlobCommitments{}
@@ -82,7 +85,19 @@ func setup(m *testing.M) {
 		panic("failed to create dynamodb client: " + err.Error())
 	}
 
-	blobMetadataStore = blobstore.NewBlobMetadataStore(dynamoClient, logger, metadataTableName, time.Hour)
+	blobMetadataStore = blobstore.NewBlobMetadataStore(dynamoClient, logger, metadataTableName)
+
+	s3Client, err = s3.NewClient(context.Background(), cfg, logger)
+	if err != nil {
+		teardown()
+		panic("failed to create s3 client: " + err.Error())
+	}
+	err = s3Client.CreateBucket(context.Background(), s3BucketName)
+	if err != nil {
+		teardown()
+		panic("failed to create s3 bucket: " + err.Error())
+	}
+	blobStore = blobstore.NewBlobStore(s3BucketName, s3Client, logger)
 
 	var X1, Y1 fp.Element
 	X1 = *X1.SetBigInt(big.NewInt(1))

disperser/common/v2/blobstore/dynamo_store.go → disperser/common/v2/blobstore/dynamo_metadata_store.go

@@ -32,16 +32,14 @@ type BlobMetadataStore struct {
 	dynamoDBClient *commondynamodb.Client
 	logger         logging.Logger
 	tableName      string
-	ttl            time.Duration
 }
 
-func NewBlobMetadataStore(dynamoDBClient *commondynamodb.Client, logger logging.Logger, tableName string, ttl time.Duration) *BlobMetadataStore {
-	logger.Debugf("creating blob metadata store v2 with table %s with TTL: %s", tableName, ttl)
+func NewBlobMetadataStore(dynamoDBClient *commondynamodb.Client, logger logging.Logger, tableName string) *BlobMetadataStore {
+	logger.Debugf("creating blob metadata store v2 with table %s", tableName)
 	return &BlobMetadataStore{
 		dynamoDBClient: dynamoDBClient,
 		logger:         logger.With("component", "blobMetadataStoreV2"),
 		tableName:      tableName,
-		ttl:            ttl,
 	}
 }
 

disperser/common/v2/blobstore/s3_blob_store.go

@@ -0,0 +1,42 @@
+package blobstore
+
+import (
+	"context"
+
+	"github.com/Layr-Labs/eigenda/common/aws/s3"
+	"github.com/Layr-Labs/eigensdk-go/logging"
+)
+
+type BlobStore struct {
+	bucketName string
+	s3Client   s3.Client
+	logger     logging.Logger
+}
+
+func NewBlobStore(s3BucketName string, s3Client s3.Client, logger logging.Logger) *BlobStore {
+	return &BlobStore{
+		bucketName: s3BucketName,
+		s3Client:   s3Client,
+		logger:     logger,
+	}
+}
+
+// StoreBlob adds a blob to the blob store
+func (b *BlobStore) StoreBlob(ctx context.Context, blobKey string, data []byte) error {
+	err := b.s3Client.UploadObject(ctx, b.bucketName, blobKey, data)
+	if err != nil {
+		b.logger.Errorf("failed to upload blob in bucket %s: %v", b.bucketName, err)
+		return err
+	}
+	return nil
+}
+
+// GetBlob retrieves a blob from the blob store
+func (b *BlobStore) GetBlob(ctx context.Context, blobKey string) ([]byte, error) {
+	data, err := b.s3Client.DownloadObject(ctx, b.bucketName, blobKey)
+	if err != nil {
+		b.logger.Errorf("failed to download blob from bucket %s: %v", b.bucketName, err)
+		return nil, err
+	}
+	return data, nil
+}

disperser/common/v2/blobstore/s3_blob_store_test.go

@@ -0,0 +1,22 @@
+package blobstore_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStoreGetBlob(t *testing.T) {
+	err := blobStore.StoreBlob(context.Background(), "testBlobKey", []byte("testBlobData"))
+	assert.NoError(t, err)
+	data, err := blobStore.GetBlob(context.Background(), "testBlobKey")
+	assert.NoError(t, err)
+	assert.Equal(t, []byte("testBlobData"), data)
+}
+
+func TestGetBlobNotFound(t *testing.T) {
+	data, err := blobStore.GetBlob(context.Background(), "nonExistentBlobKey")
+	assert.Error(t, err)
+	assert.Nil(t, data)
+}