Enable minibatcher

core/data.go

@@ -275,6 +275,9 @@ type BlobHeader struct {
 
 	// AccountID is the account that is paying for the blob to be stored
 	AccountID AccountID
+
+	// Reference block number
+	ReferenceBlockNumber uint
 }
 
 func (b *BlobHeader) GetQuorumInfo(quorum QuorumID) *BlobQuorumInfo {

disperser/batcher/batch_confirmer.go

@@ -25,7 +25,7 @@ type BatchConfirmerConfig struct {
 	DispersalStatusCheckInterval time.Duration
 	AttestationTimeout           time.Duration
 	SRSOrder                     int
-	NumConnections               int
+	NumConnections               uint
 	MaxNumRetriesPerBlob         uint
 }
 
@@ -46,6 +46,7 @@ type BatchConfirmer struct {
 
 	ethClient common.EthClient
 	logger    logging.Logger
+	Metrics   *Metrics
 }
 
 func NewBatchConfirmer(
@@ -62,6 +63,7 @@ func NewBatchConfirmer(
 	txnManager TxnManager,
 	minibatcher *Minibatcher,
 	logger logging.Logger,
+	metrics *Metrics,
 ) (*BatchConfirmer, error) {
 	return &BatchConfirmer{
 		BatchConfirmerConfig: config,
@@ -76,6 +78,7 @@ func NewBatchConfirmer(
 		Transactor:         transactor,
 		TransactionManager: txnManager,
 		Minibatcher:        minibatcher,
+		Metrics:            metrics,
 
 		ethClient: ethClient,
 		logger:    logger.With("component", "BatchConfirmer"),
@@ -285,6 +288,11 @@ func (b *BatchConfirmer) handleFailure(ctx context.Context, batchID uuid.UUID, b
 			continue
 		}
 
+		if reason == FailNoSignatures {
+			b.Metrics.UpdateCompletedBlob(int(metadata.RequestMetadata.BlobSize), disperser.InsufficientSignatures)
+		} else {
+			b.Metrics.UpdateCompletedBlob(int(metadata.RequestMetadata.BlobSize), disperser.Failed)
+		}
 		numPermanentFailures++
 	}
 
@@ -324,6 +332,7 @@ func (b *BatchConfirmer) HandleSingleBatch(ctx context.Context) error {
 		}
 	}
 
+	b.logger.Info("Processing batch", "batchID", batch.ID)
 	// Make sure all minibatches in the batch have been dispersed
 	batchDispersed := false
 	stateUpdateTicker.Reset(b.DispersalStatusCheckInterval)
@@ -332,6 +341,7 @@ func (b *BatchConfirmer) HandleSingleBatch(ctx context.Context) error {
 	for !batchDispersed {
 		select {
 		case <-ctxWithTimeout.Done():
+			b.logger.Error("timed out waiting for batch to disperse")
 			return ctxWithTimeout.Err()
 		case <-stateUpdateTicker.C:
 			batchDispersed, err = b.MinibatchStore.BatchDispersed(ctx, batch.ID, batch.NumMinibatches)

disperser/batcher/batch_confirmer_test.go

@@ -15,7 +15,6 @@ import (
 	bat "github.com/Layr-Labs/eigenda/disperser/batcher"
 	batcherinmem "github.com/Layr-Labs/eigenda/disperser/batcher/inmem"
 	batchermock "github.com/Layr-Labs/eigenda/disperser/batcher/mock"
-	batmock "github.com/Layr-Labs/eigenda/disperser/batcher/mock"
 	"github.com/Layr-Labs/eigenda/disperser/common/inmem"
 	dmock "github.com/Layr-Labs/eigenda/disperser/mock"
 	"github.com/Layr-Labs/eigenda/encoding"
@@ -57,7 +56,7 @@ func makeBatchConfirmer(t *testing.T) *batchConfirmerComponents {
 	dispatcher := dmock.NewDispatcher(state)
 	blobStore := inmem.NewBlobStore()
 	ethClient := &cmock.MockEthClient{}
-	txnManager := batmock.NewTxnManager()
+	txnManager := batchermock.NewTxnManager()
 	minibatchStore := batcherinmem.NewMinibatchStore(logger)
 	encodingWorkerPool := workerpool.New(10)
 	encoderProver, err = makeTestProver()
@@ -76,7 +75,7 @@ func makeBatchConfirmer(t *testing.T) *batchConfirmerComponents {
 		MaxNumConnections:         10,
 		MaxNumRetriesPerBlob:      2,
 		MaxNumRetriesPerDispersal: 1,
-	}, blobStore, minibatchStore, dispatcher, mockChainState, assignmentCoordinator, encodingStreamer, ethClient, pool, logger)
+	}, blobStore, minibatchStore, dispatcher, mockChainState, encodingStreamer, pool, logger, metrics)
 	assert.NoError(t, err)
 
 	config := bat.BatchConfirmerConfig{
@@ -88,7 +87,7 @@ func makeBatchConfirmer(t *testing.T) *batchConfirmerComponents {
 		SRSOrder:                     3000,
 		MaxNumRetriesPerBlob:         2,
 	}
-	b, err := bat.NewBatchConfirmer(config, blobStore, minibatchStore, dispatcher, mockChainState, assignmentCoordinator, encodingStreamer, agg, ethClient, transactor, txnManager, minibatcher, logger)
+	b, err := bat.NewBatchConfirmer(config, blobStore, minibatchStore, dispatcher, mockChainState, assignmentCoordinator, encodingStreamer, agg, ethClient, transactor, txnManager, minibatcher, logger, metrics)
 	assert.NoError(t, err)
 
 	ethClient.On("BlockNumber").Return(uint64(initialBlock), nil)

disperser/batcher/batcher.go

@@ -41,6 +41,7 @@ type QuorumInfo struct {
 
 type TimeoutConfig struct {
 	EncodingTimeout     time.Duration
+	DispersalTimeout    time.Duration
 	AttestationTimeout  time.Duration
 	ChainReadTimeout    time.Duration
 	ChainWriteTimeout   time.Duration
@@ -64,6 +65,11 @@ type Config struct {
 
 	TargetNumChunks          uint
 	MaxBlobsToFetchFromStore int
+
+	EnableMinibatch              bool
+	BatchstoreTableName          string
+	MinibatcherConfig            MinibatcherConfig
+	DispersalStatusCheckInterval time.Duration
 }
 
 type Batcher struct {
@@ -105,6 +111,10 @@ func NewBatcher(
 	metrics *Metrics,
 	heartbeatChan chan time.Time,
 ) (*Batcher, error) {
+	if config.EnableMinibatch {
+		return nil, errors.New("minibatch is not supported")
+	}
+
 	batchTrigger := NewEncodedSizeNotifier(
 		make(chan struct{}, 1),
 		uint64(config.BatchSizeMBLimit)*1024*1024, // convert to bytes

disperser/batcher/batcher_test.go

@@ -41,7 +41,6 @@ type batcherComponents struct {
 	transactor       *coremock.MockTransactor
 	txnManager       *batchermock.MockTxnManager
 	blobStore        disperser.BlobStore
-	encoderClient    *disperser.LocalEncoderClient
 	encodingStreamer *bat.EncodingStreamer
 	ethClient        *cmock.MockEthClient
 	dispatcher       *dmock.Dispatcher
@@ -153,7 +152,6 @@ func makeBatcher(t *testing.T) (*batcherComponents, *bat.Batcher, func() []time.
 			transactor:       transactor,
 			txnManager:       txnManager,
 			blobStore:        blobStore,
-			encoderClient:    encoderClient,
 			encodingStreamer: b.EncodingStreamer,
 			ethClient:        ethClient,
 			dispatcher:       dispatcher,

disperser/batcher/batchstore/minibatch_store.go

@@ -2,6 +2,7 @@ package batchstore
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"time"
@@ -35,7 +36,7 @@ type MinibatchStore struct {
 
 var _ batcher.MinibatchStore = (*MinibatchStore)(nil)
 
-func NewMinibatchStore(dynamoDBClient *commondynamodb.Client, logger logging.Logger, tableName string, ttl time.Duration) *MinibatchStore {
+func NewMinibatchStore(dynamoDBClient *commondynamodb.Client, logger logging.Logger, tableName string, ttl time.Duration) batcher.MinibatchStore {
 	logger.Debugf("creating minibatch store with table %s with TTL: %s", tableName, ttl)
 	return &MinibatchStore{
 		dynamoDBClient: dynamoDBClient,
@@ -145,6 +146,9 @@ func MarshalDispersal(response *batcher.MinibatchDispersal) (map[string]types.At
 	if err != nil {
 		return nil, err
 	}
+	if response.Error != nil {
+		fields["Error"] = &types.AttributeValueMemberS{Value: response.DispersalResponse.Error.Error()}
+	}
 	fields["BatchID"] = &types.AttributeValueMemberS{Value: response.BatchID.String()}
 	fields["SK"] = &types.AttributeValueMemberS{Value: dispersalSKPrefix + fmt.Sprintf("%d#%s", response.MinibatchIndex, response.OperatorID.Hex())}
 	fields["OperatorID"] = &types.AttributeValueMemberS{Value: response.OperatorID.Hex()}
@@ -158,6 +162,10 @@ func MarshalDispersalResponse(response *batcher.DispersalResponse) (map[string]t
 	if err != nil {
 		return nil, err
 	}
+	if response.Error != nil {
+		fields["Error"] = &types.AttributeValueMemberS{Value: response.Error.Error()}
+	}
+	fields["RespondedAt"] = &types.AttributeValueMemberN{Value: fmt.Sprintf("%d", response.RespondedAt.UTC().Unix())}
 	return fields, nil
 }
 
@@ -192,6 +200,22 @@ func UnmarshalBatchID(item commondynamodb.Item) (*uuid.UUID, error) {
 	return &batchID, nil
 }
 
+func UnmarshalError(item commondynamodb.Item) (error, error) {
+	type Error struct {
+		Error string
+	}
+	var e Error
+	err := attributevalue.UnmarshalMap(item, &e)
+	if err != nil {
+		return nil, err
+	}
+
+	if e.Error == "" {
+		return nil, nil
+	}
+	return errors.New(e.Error), nil
+}
+
 func UnmarshalBlobKey(item commondynamodb.Item) (*disperser.BlobKey, error) {
 	blobKey := disperser.BlobKey{}
 	err := attributevalue.UnmarshalMap(item, &blobKey)
@@ -257,6 +281,12 @@ func UnmarshalDispersal(item commondynamodb.Item) (*batcher.MinibatchDispersal,
 	}
 	response.OperatorID = *operatorID
 
+	dispersalError, err := UnmarshalError(item)
+	if err != nil {
+		return nil, err
+	}
+	response.OperatorID = *operatorID
+	response.Error = dispersalError
 	response.RespondedAt = response.RespondedAt.UTC()
 	response.RequestedAt = response.RequestedAt.UTC()
 	return &response, nil

disperser/batcher/batchstore/minibatch_store_test.go

@@ -2,6 +2,7 @@ package batchstore_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"testing"
@@ -35,7 +36,7 @@ var (
 	localStackPort   = "4566"
 
 	dynamoClient   *dynamodb.Client
-	minibatchStore *batchstore.MinibatchStore
+	minibatchStore batcher.MinibatchStore
 
 	UUID               = uuid.New()
 	minibatchTableName = fmt.Sprintf("test-MinibatchStore-%v", UUID)
@@ -237,6 +238,34 @@ func TestPutDispersal(t *testing.T) {
 	assert.Equal(t, response, r)
 }
 
+func TestDispersalError(t *testing.T) {
+	ctx := context.Background()
+	id, err := uuid.NewV7()
+	assert.NoError(t, err)
+	ts := time.Now().Truncate(time.Second).UTC()
+	opID := core.OperatorID([32]byte{123})
+	response := &batcher.MinibatchDispersal{
+		BatchID:         id,
+		MinibatchIndex:  0,
+		OperatorID:      opID,
+		OperatorAddress: gcommon.HexToAddress("0x0"),
+		Socket:          "socket",
+		NumBlobs:        1,
+		RequestedAt:     ts,
+		DispersalResponse: batcher.DispersalResponse{
+			Signatures:  nil,
+			RespondedAt: ts,
+			Error:       errors.New("error"),
+		},
+	}
+	err = minibatchStore.PutDispersal(ctx, response)
+	assert.NoError(t, err)
+	r, err := minibatchStore.GetDispersal(ctx, response.BatchID, response.MinibatchIndex, opID)
+	assert.NoError(t, err)
+	assert.Equal(t, response, r)
+	assert.Equal(t, response.DispersalResponse.Error.Error(), r.DispersalResponse.Error.Error())
+}
+
 func TestDispersalStatus(t *testing.T) {
 	ctx := context.Background()
 	id, err := uuid.NewV7()

disperser/batcher/encoding_streamer.go

@@ -492,7 +492,8 @@ func (e *EncodingStreamer) CreateMinibatch(ctx context.Context) (*batch, error)
 			metadataByKey[blobKey] = result.BlobMetadata
 			blobQuorums[blobKey] = make([]*core.BlobQuorumInfo, 0)
 			blobHeader := &core.BlobHeader{
-				BlobCommitments: *result.Commitment,
+				BlobCommitments:      *result.Commitment,
+				ReferenceBlockNumber: e.ReferenceBlockNumber,
 			}
 			blobHeaderByKey[blobKey] = blobHeader
 			encodedBlobByKey[blobKey] = core.EncodedBlob{

disperser/batcher/grpc/dispatcher.go

@@ -409,11 +409,12 @@ func getBlobMessage(blob *core.EncodedBlobMessage, useGnarkBundleEncoding bool)
 
 	return &node.Blob{
 		Header: &node.BlobHeader{
-			Commitment:       commitData,
-			LengthCommitment: &lengthCommitData,
-			LengthProof:      &lengthProofData,
-			Length:           uint32(blob.BlobHeader.Length),
-			QuorumHeaders:    quorumHeaders,
+			Commitment:           commitData,
+			LengthCommitment:     &lengthCommitData,
+			LengthProof:          &lengthProofData,
+			Length:               uint32(blob.BlobHeader.Length),
+			QuorumHeaders:        quorumHeaders,
+			ReferenceBlockNumber: uint32(blob.BlobHeader.ReferenceBlockNumber),
 		},
 		Bundles: bundles,
 	}, nil

disperser/batcher/minibatch_store.go

@@ -45,7 +45,7 @@ type DispersalResponse struct {
 	// The signatures are compressed using G1Affine.Bytes(), to be decompressed using G1Affine.SetBytes()
 	Signatures  [][32]byte
 	RespondedAt time.Time
-	Error       error
+	Error       error `dynamodbav:"-"`
 }
 
 type MinibatchDispersal struct {