Minibatch store interface (#635)

disperser/batcher/inmem/minibatch_store.go

@@ -0,0 +1,103 @@
+package inmem
+
+import (
+	"fmt"
+
+	"github.com/Layr-Labs/eigenda/disperser/batcher"
+	"github.com/Layr-Labs/eigensdk-go/logging"
+	"github.com/google/uuid"
+)
+
+type minibatchStore struct {
+	// BatchRecords maps batch IDs to batch records
+	BatchRecords map[uuid.UUID]*batcher.BatchRecord
+	// MinibatchRecords maps batch IDs to a map from minibatch indices to minibatch records
+	MinibatchRecords map[uuid.UUID]map[uint]*batcher.MinibatchRecord
+	// DispersalRequests maps batch IDs to a map from minibatch indices to dispersal requests
+	DispersalRequests map[uuid.UUID]map[uint]*batcher.DispersalRequest
+	// DispersalResponses maps batch IDs to a map from minibatch indices to dispersal responses
+	DispersalResponses map[uuid.UUID]map[uint]*batcher.DispersalResponse
+
+	logger logging.Logger
+}
+
+var _ batcher.MinibatchStore = (*minibatchStore)(nil)
+
+func NewMinibatchStore(logger logging.Logger) batcher.MinibatchStore {
+	return &minibatchStore{
+		BatchRecords:       make(map[uuid.UUID]*batcher.BatchRecord),
+		MinibatchRecords:   make(map[uuid.UUID]map[uint]*batcher.MinibatchRecord),
+		DispersalRequests:  make(map[uuid.UUID]map[uint]*batcher.DispersalRequest),
+		DispersalResponses: make(map[uuid.UUID]map[uint]*batcher.DispersalResponse),
+
+		logger: logger,
+	}
+}
+
+func (m *minibatchStore) PutBatch(batch *batcher.BatchRecord) error {
+	m.BatchRecords[batch.ID] = batch
+
+	return nil
+}
+
+func (m *minibatchStore) GetBatch(batchID uuid.UUID) (*batcher.BatchRecord, error) {
+	b, ok := m.BatchRecords[batchID]
+	if !ok {
+		return nil, fmt.Errorf("batch not found")
+	}
+	return b, nil
+}
+
+func (m *minibatchStore) PutMiniBatch(minibatch *batcher.MinibatchRecord) error {
+	if _, ok := m.MinibatchRecords[minibatch.BatchID]; !ok {
+		m.MinibatchRecords[minibatch.BatchID] = make(map[uint]*batcher.MinibatchRecord)
+	}
+	m.MinibatchRecords[minibatch.BatchID][minibatch.MinibatchIndex] = minibatch
+
+	return nil
+}
+
+func (m *minibatchStore) GetMiniBatch(batchID uuid.UUID, minibatchIndex uint) (*batcher.MinibatchRecord, error) {
+	if _, ok := m.MinibatchRecords[batchID]; !ok {
+		return nil, nil
+	}
+	return m.MinibatchRecords[batchID][minibatchIndex], nil
+}
+
+func (m *minibatchStore) PutDispersalRequest(request *batcher.DispersalRequest) error {
+	if _, ok := m.DispersalRequests[request.BatchID]; !ok {
+		m.DispersalRequests[request.BatchID] = make(map[uint]*batcher.DispersalRequest)
+	}
+	m.DispersalRequests[request.BatchID][request.MinibatchIndex] = request
+
+	return nil
+}
+
+func (m *minibatchStore) GetDispersalRequest(batchID uuid.UUID, minibatchIndex uint) (*batcher.DispersalRequest, error) {
+	if _, ok := m.DispersalRequests[batchID]; !ok {
+		return nil, nil
+	}
+
+	return m.DispersalRequests[batchID][minibatchIndex], nil
+}
+
+func (m *minibatchStore) PutDispersalResponse(response *batcher.DispersalResponse) error {
+	if _, ok := m.DispersalResponses[response.BatchID]; !ok {
+		m.DispersalResponses[response.BatchID] = make(map[uint]*batcher.DispersalResponse)
+	}
+	m.DispersalResponses[response.BatchID][response.MinibatchIndex] = response
+
+	return nil
+}
+
+func (m *minibatchStore) GetDispersalResponse(batchID uuid.UUID, minibatchIndex uint) (*batcher.DispersalResponse, error) {
+	if _, ok := m.DispersalResponses[batchID]; !ok {
+		return nil, nil
+	}
+
+	return m.DispersalResponses[batchID][minibatchIndex], nil
+}
+
+func (m *minibatchStore) GetPendingBatch() (*batcher.BatchRecord, error) {
+	return nil, nil
+}

disperser/batcher/inmem/minibatch_store_test.go

@@ -0,0 +1,98 @@
+package inmem_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/Layr-Labs/eigenda/core"
+	"github.com/Layr-Labs/eigenda/disperser/batcher"
+	"github.com/Layr-Labs/eigenda/disperser/batcher/inmem"
+	gcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+)
+
+func newMinibatchStore() batcher.MinibatchStore {
+	return inmem.NewMinibatchStore(nil)
+}
+
+func TestPutBatch(t *testing.T) {
+	s := newMinibatchStore()
+	id, err := uuid.NewV7()
+	assert.NoError(t, err)
+
+	batch := &batcher.BatchRecord{
+		ID:                   id,
+		CreatedAt:            time.Now().UTC(),
+		ReferenceBlockNumber: 1,
+		HeaderHash:           [32]byte{1},
+		AggregatePubKey:      nil,
+		AggregateSignature:   nil,
+	}
+	err = s.PutBatch(batch)
+	assert.NoError(t, err)
+	b, err := s.GetBatch(batch.ID)
+	assert.NoError(t, err)
+	assert.Equal(t, batch, b)
+}
+
+func TestPutMiniBatch(t *testing.T) {
+	s := newMinibatchStore()
+	id, err := uuid.NewV7()
+	assert.NoError(t, err)
+	minibatch := &batcher.MinibatchRecord{
+		BatchID:              id,
+		MinibatchIndex:       12,
+		BlobHeaderHashes:     [][32]byte{{1}},
+		BatchSize:            1,
+		ReferenceBlockNumber: 1,
+	}
+	err = s.PutMiniBatch(minibatch)
+	assert.NoError(t, err)
+	m, err := s.GetMiniBatch(minibatch.BatchID, minibatch.MinibatchIndex)
+	assert.NoError(t, err)
+	assert.Equal(t, minibatch, m)
+}
+
+func TestPutDispersalRequest(t *testing.T) {
+	s := newMinibatchStore()
+	id, err := uuid.NewV7()
+	assert.NoError(t, err)
+	request := &batcher.DispersalRequest{
+		BatchID:         id,
+		MinibatchIndex:  0,
+		OperatorID:      core.OperatorID([32]byte{1}),
+		OperatorAddress: gcommon.HexToAddress("0x0"),
+		NumBlobs:        1,
+		RequestedAt:     time.Now().UTC(),
+	}
+	err = s.PutDispersalRequest(request)
+	assert.NoError(t, err)
+	r, err := s.GetDispersalRequest(request.BatchID, request.MinibatchIndex)
+	assert.NoError(t, err)
+	assert.Equal(t, request, r)
+}
+
+func TestPutDispersalResponse(t *testing.T) {
+	s := newMinibatchStore()
+	id, err := uuid.NewV7()
+	assert.NoError(t, err)
+	response := &batcher.DispersalResponse{
+		DispersalRequest: batcher.DispersalRequest{
+			BatchID:         id,
+			MinibatchIndex:  0,
+			OperatorID:      core.OperatorID([32]byte{1}),
+			OperatorAddress: gcommon.HexToAddress("0x0"),
+			NumBlobs:        1,
+			RequestedAt:     time.Now().UTC(),
+		},
+		Signatures:  nil,
+		RespondedAt: time.Now().UTC(),
+		Error:       nil,
+	}
+	err = s.PutDispersalResponse(response)
+	assert.NoError(t, err)
+	r, err := s.GetDispersalResponse(response.BatchID, response.MinibatchIndex)
+	assert.NoError(t, err)
+	assert.Equal(t, response, r)
+}

disperser/batcher/minibatch_store.go

@@ -0,0 +1,54 @@
+package batcher
+
+import (
+	"time"
+
+	"github.com/Layr-Labs/eigenda/core"
+	gcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/google/uuid"
+)
+
+type BatchRecord struct {
+	ID                   uuid.UUID
+	CreatedAt            time.Time
+	ReferenceBlockNumber uint
+	HeaderHash           [32]byte
+	AggregatePubKey      *core.G2Point
+	AggregateSignature   *core.Signature
+}
+
+type MinibatchRecord struct {
+	BatchID              uuid.UUID
+	MinibatchIndex       uint
+	BlobHeaderHashes     [][32]byte
+	BatchSize            uint64 // in bytes
+	ReferenceBlockNumber uint
+}
+
+type DispersalRequest struct {
+	BatchID        uuid.UUID
+	MinibatchIndex uint
+	core.OperatorID
+	OperatorAddress gcommon.Address
+	NumBlobs        uint
+	RequestedAt     time.Time
+}
+
+type DispersalResponse struct {
+	DispersalRequest
+	Signatures  []*core.Signature
+	RespondedAt time.Time
+	Error       error
+}
+
+type MinibatchStore interface {
+	PutBatch(batch *BatchRecord) error
+	GetBatch(batchID uuid.UUID) (*BatchRecord, error)
+	PutMiniBatch(minibatch *MinibatchRecord) error
+	GetMiniBatch(batchID uuid.UUID, minibatchIndex uint) (*MinibatchRecord, error)
+	PutDispersalRequest(request *DispersalRequest) error
+	GetDispersalRequest(batchID uuid.UUID, minibatchIndex uint) (*DispersalRequest, error)
+	PutDispersalResponse(response *DispersalResponse) error
+	GetDispersalResponse(batchID uuid.UUID, minibatchIndex uint) (*DispersalResponse, error)
+	GetPendingBatch() (*BatchRecord, error)
+}