Move batchSubmissionLogic out of sequencer

Author: Manav-Aggarwal

block/daIncluder.go

@@ -29,7 +29,7 @@ func (m *Manager) DAIncluderLoop(ctx context.Context) {
 			headerHash := header.Hash()
 			dataHash := data.DACommitment()
 
-			if m.headerCache.IsDAIncluded(headerHash.String()) && (bytes.Equal(dataHash, dataHashForEmptyTxs) || m.dataCache.IsDAIncluded(dataHash.String())) {
+			if m.headerCache.IsDAIncluded(headerHash.String()) && (bytes.Equal(dataHash, DataHashForEmptyTxs) || m.dataCache.IsDAIncluded(dataHash.String())) {
 				// Both header and data are DA-included, so we can advance the height
 				if err := m.incrementDAIncludedHeight(ctx); err != nil {
 					break

block/manager.go

@@ -63,7 +63,7 @@ const (
 
 var (
 	// dataHashForEmptyTxs to be used while only syncing headers from DA and no p2p to get the Data for no txs scenarios, the syncing can proceed without getting stuck forever.
-	dataHashForEmptyTxs = []byte{110, 52, 11, 156, 255, 179, 122, 152, 156, 165, 68, 230, 187, 120, 10, 44, 120, 144, 29, 63, 179, 55, 56, 118, 133, 17, 163, 6, 23, 175, 160, 29}
+	DataHashForEmptyTxs = []byte{110, 52, 11, 156, 255, 179, 122, 152, 156, 165, 68, 230, 187, 120, 10, 44, 120, 144, 29, 63, 179, 55, 56, 118, 133, 17, 163, 6, 23, 175, 160, 29}
 
 	// initialBackoff defines initial value for block submission backoff
 	initialBackoff = 100 * time.Millisecond
@@ -158,6 +158,9 @@ type Manager struct {
 
 	// txNotifyCh is used to signal when new transactions are available
 	txNotifyCh chan struct{}
+
+	// batchSubmissionChan is used to submit batches to the sequencer
+	batchSubmissionChan chan coresequencer.Batch
 }
 
 // getInitialState tries to load lastState from Store, and if it's not available it reads genesis.
@@ -339,37 +342,38 @@ func NewManager(
 		dalc:      dalc,
 		daHeight:  &daH,
 		// channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
-		HeaderCh:       make(chan *types.SignedHeader, channelLength),
-		DataCh:         make(chan *types.Data, channelLength),
-		headerInCh:     make(chan NewHeaderEvent, eventInChLength),
-		dataInCh:       make(chan NewDataEvent, eventInChLength),
-		headerStoreCh:  make(chan struct{}, 1),
-		dataStoreCh:    make(chan struct{}, 1),
-		headerStore:    headerStore,
-		dataStore:      dataStore,
-		lastStateMtx:   new(sync.RWMutex),
-		lastBatchData:  lastBatchData,
-		headerCache:    cache.NewCache[types.SignedHeader](),
-		dataCache:      cache.NewCache[types.Data](),
-		retrieveCh:     make(chan struct{}, 1),
-		daIncluderCh:   make(chan struct{}, 1),
-		logger:         logger,
-		txsAvailable:   false,
-		pendingHeaders: pendingHeaders,
-		metrics:        seqMetrics,
-		sequencer:      sequencer,
-		exec:           exec,
-		gasPrice:       gasPrice,
-		gasMultiplier:  gasMultiplier,
-		txNotifyCh:     make(chan struct{}, 1), // Non-blocking channel
+		HeaderCh:            make(chan *types.SignedHeader, channelLength),
+		DataCh:              make(chan *types.Data, channelLength),
+		headerInCh:          make(chan NewHeaderEvent, eventInChLength),
+		dataInCh:            make(chan NewDataEvent, eventInChLength),
+		headerStoreCh:       make(chan struct{}, 1),
+		dataStoreCh:         make(chan struct{}, 1),
+		headerStore:         headerStore,
+		dataStore:           dataStore,
+		lastStateMtx:        new(sync.RWMutex),
+		lastBatchData:       lastBatchData,
+		headerCache:         cache.NewCache[types.SignedHeader](),
+		dataCache:           cache.NewCache[types.Data](),
+		retrieveCh:          make(chan struct{}, 1),
+		daIncluderCh:        make(chan struct{}, 1),
+		logger:              logger,
+		txsAvailable:        false,
+		pendingHeaders:      pendingHeaders,
+		metrics:             seqMetrics,
+		sequencer:           sequencer,
+		exec:                exec,
+		gasPrice:            gasPrice,
+		gasMultiplier:       gasMultiplier,
+		txNotifyCh:          make(chan struct{}, 1), // Non-blocking channel
+		batchSubmissionChan: make(chan coresequencer.Batch, eventInChLength),
 	}
 	agg.init(ctx)
 	// Set the default publishBlock implementation
 	agg.publishBlock = agg.publishBlockInternal
-
-	// Set the manager pointer in the sequencer if it is a *single.Sequencer
-	if s, ok := sequencer.(interface{ SetManager(*Manager) }); ok {
-		s.SetManager(agg)
+	if s, ok := agg.sequencer.(interface {
+		SetBatchSubmissionChan(chan coresequencer.Batch)
+	}); ok {
+		s.SetBatchSubmissionChan(agg.batchSubmissionChan)
 	}
 
 	return agg, nil
@@ -478,7 +482,7 @@ func (m *Manager) retrieveBatch(ctx context.Context) (*BatchData, error) {
 		if len(res.Batch.Transactions) == 0 {
 			errRetrieveBatch = ErrNoBatch
 		}
-		// Even if there are no transactions, update lastBatchData so we don’t
+		// Even if there are no transactions, update lastBatchData so we don't
 		// repeatedly emit the same empty batch, and persist it to metadata.
 		if err := m.store.SetMetadata(ctx, LastBatchDataKey, convertBatchDataToBytes(res.BatchData)); err != nil {
 			m.logger.Error("error while setting last batch hash", "error", err)
@@ -780,7 +784,7 @@ func (m *Manager) execCreateBlock(_ context.Context, height uint64, lastSignatur
 				Time:    uint64(batchData.UnixNano()), //nolint:gosec // why is time unix? (tac0turtle)
 			},
 			LastHeaderHash:  lastHeaderHash,
-			DataHash:        dataHashForEmptyTxs, // Use batchDataIDs when available
+			DataHash:        DataHashForEmptyTxs, // Use batchDataIDs when available
 			ConsensusHash:   make(types.Hash, 32),
 			AppHash:         m.lastState.AppHash,
 			ProposerAddress: m.genesis.ProposerAddress,
@@ -923,6 +927,11 @@ func (m *Manager) NotifyNewTransactions() {
 	}
 }
 
+// HeaderCache returns the headerCache used by the manager.
+func (m *Manager) HeaderCache() *cache.Cache[types.SignedHeader] {
+	return m.headerCache
+}
+
 // DataCache returns the dataCache used by the manager.
 func (m *Manager) DataCache() *cache.Cache[types.Data] {
 	return m.dataCache

block/submitter.go

@@ -8,6 +8,9 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	coreda "github.com/rollkit/rollkit/core/da"
+	coresequencer "github.com/rollkit/rollkit/core/sequencer"
+	"github.com/rollkit/rollkit/types"
+	pb "github.com/rollkit/rollkit/types/pb/rollkit/v1"
 )
 
 // HeaderSubmissionLoop is responsible for submitting headers to the DA layer.
@@ -103,7 +106,7 @@ daSubmitRetryLoop:
 			}
 			m.pendingHeaders.setLastSubmittedHeight(ctx, lastSubmittedHeight)
 			headersToSubmit = notSubmittedHeaders
-			m.daIncluderCh <- struct{}{}
+			m.sendNonBlockingSignalToDAIncluderCh()
 			// reset submission options when successful
 			// scale back gasPrice gradually
 			backoff = 0
@@ -142,3 +145,153 @@ daSubmitRetryLoop:
 	}
 	return nil
 }
+
+// BatchSubmissionLoop is responsible for submitting batches to the DA layer.
+func (m *Manager) BatchSubmissionLoop(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			m.logger.Info("Batch submission loop stopped")
+			return
+		case batch := <-m.batchSubmissionChan:
+			err := m.submitBatchToDA(ctx, batch)
+			if err != nil {
+				m.logger.Error("failed to submit batch to DA", "error", err)
+			}
+		}
+	}
+}
+
+// submitBatchToDA submits a batch of transactions to the Data Availability (DA) layer.
+// It implements a retry mechanism with exponential backoff and gas price adjustments
+// to handle various failure scenarios.
+func (m *Manager) submitBatchToDA(ctx context.Context, batch coresequencer.Batch) error {
+	currentBatch := batch
+	submittedAllTxs := false
+	var backoff time.Duration
+	totalTxCount := len(batch.Transactions)
+	submittedTxCount := 0
+	attempt := 0
+
+	// Store initial values to be able to reset or compare later
+	initialGasPrice, err := m.dalc.GasPrice(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get initial gas price: %w", err)
+	}
+	initialMaxBlobSize, err := m.dalc.MaxBlobSize(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get initial max blob size: %w", err)
+	}
+	maxBlobSize := initialMaxBlobSize
+	gasPrice := initialGasPrice
+
+	for !submittedAllTxs && attempt < maxSubmitAttempts {
+		// Wait for backoff duration or exit if context is done
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(backoff):
+		}
+
+		// Convert batch to protobuf and marshal
+		batchPb := &pb.Batch{
+			Txs: currentBatch.Transactions,
+		}
+		batchBz, err := proto.Marshal(batchPb)
+		if err != nil {
+			return fmt.Errorf("failed to marshal batch: %w", err)
+		}
+
+		// Attempt to submit the batch to the DA layer
+		res := m.dalc.Submit(ctx, [][]byte{batchBz}, maxBlobSize, gasPrice)
+
+		gasMultiplier, err := m.dalc.GasMultiplier(ctx)
+		if err != nil {
+			return fmt.Errorf("failed to get gas multiplier: %w", err)
+		}
+
+		switch res.Code {
+		case coreda.StatusSuccess:
+			// Count submitted transactions for this attempt
+			submittedTxs := int(res.SubmittedCount)
+			m.logger.Info("successfully submitted transactions to DA layer",
+				"gasPrice", gasPrice,
+				"height", res.Height,
+				"submittedTxs", submittedTxs,
+				"remainingTxs", len(currentBatch.Transactions)-submittedTxs)
+
+			// Update overall progress
+			submittedTxCount += submittedTxs
+
+			// Check if all transactions in the current batch were submitted
+			if submittedTxs == len(currentBatch.Transactions) {
+				submittedAllTxs = true
+			} else {
+				// Update the current batch to contain only the remaining transactions
+				currentBatch.Transactions = currentBatch.Transactions[submittedTxs:]
+			}
+
+			// Reset submission parameters after success
+			backoff = 0
+			maxBlobSize = initialMaxBlobSize
+
+			// Gradually reduce gas price on success, but not below initial price
+			if gasMultiplier > 0 && gasPrice != 0 {
+				gasPrice = gasPrice / gasMultiplier
+				if gasPrice < initialGasPrice {
+					gasPrice = initialGasPrice
+				}
+			}
+			m.logger.Debug("resetting DA layer submission options", "backoff", backoff, "gasPrice", gasPrice)
+
+			// Set DA included in manager's dataCache if all txs submitted and manager is set
+			if submittedAllTxs {
+				data := &types.Data{
+					Txs: make(types.Txs, len(currentBatch.Transactions)),
+				}
+				for i, tx := range currentBatch.Transactions {
+					data.Txs[i] = types.Tx(tx)
+				}
+				hash := data.DACommitment().String()
+				if err == nil {
+					m.DataCache().SetDAIncluded(hash)
+				}
+				m.sendNonBlockingSignalToDAIncluderCh()
+			}
+
+		case coreda.StatusNotIncludedInBlock, coreda.StatusAlreadyInMempool:
+			// For mempool-related issues, use a longer backoff and increase gas price
+			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
+			backoff = m.config.DA.BlockTime.Duration * time.Duration(m.config.DA.MempoolTTL)
+
+			// Increase gas price to prioritize the transaction
+			if gasMultiplier > 0 && gasPrice != 0 {
+				gasPrice = gasPrice * gasMultiplier
+			}
+			m.logger.Info("retrying DA layer submission with", "backoff", backoff, "gasPrice", gasPrice)
+
+		case coreda.StatusTooBig:
+			// If the blob is too big, reduce the max blob size
+			maxBlobSize = maxBlobSize / 4
+			fallthrough
+
+		default:
+			// For other errors, use exponential backoff
+			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
+			backoff = m.exponentialBackoff(backoff)
+		}
+
+		attempt += 1
+	}
+
+	// Return error if not all transactions were submitted after all attempts
+	if !submittedAllTxs {
+		return fmt.Errorf(
+			"failed to submit all transactions to DA layer, submitted %d txs (%d left) after %d attempts",
+			submittedTxCount,
+			totalTxCount-submittedTxCount,
+			attempt,
+		)
+	}
+	return nil
+}

block/sync.go

@@ -163,7 +163,7 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
 
 func (m *Manager) handleEmptyDataHash(ctx context.Context, header *types.Header) {
 	headerHeight := header.Height()
-	if bytes.Equal(header.DataHash, dataHashForEmptyTxs) {
+	if bytes.Equal(header.DataHash, DataHashForEmptyTxs) {
 		var lastDataHash types.Hash
 		var err error
 		var lastData *types.Data

block/sync_test.go

@@ -767,7 +767,7 @@ func TestHandleEmptyDataHash(t *testing.T) {
 	// Define the test data
 	headerHeight := 2
 	header := &types.Header{
-		DataHash: dataHashForEmptyTxs,
+		DataHash: DataHashForEmptyTxs,
 		BaseHeader: types.BaseHeader{
 			Height: 2,
 			Time:   uint64(time.Now().UnixNano()),

core/sequencer/dummy.go

@@ -12,8 +12,9 @@ import (
 
 // dummySequencer is a dummy implementation of the Sequencer interface for testing
 type dummySequencer struct {
-	mu      sync.RWMutex
-	batches map[string]*Batch
+	mu                  sync.RWMutex
+	batches             map[string]*Batch
+	batchSubmissionChan chan Batch
 }
 
 // NewDummySequencer creates a new dummy Sequencer instance
@@ -29,6 +30,9 @@ func (s *dummySequencer) SubmitRollupBatchTxs(ctx context.Context, req SubmitRol
 	defer s.mu.Unlock()
 
 	s.batches[string(req.RollupId)] = req.Batch
+	if req.Batch != nil && len(req.Batch.Transactions) > 0 {
+		s.batchSubmissionChan <- *req.Batch
+	}
 	return &SubmitRollupBatchTxsResponse{}, nil
 }
 
@@ -54,3 +58,7 @@ func (s *dummySequencer) VerifyBatch(ctx context.Context, req VerifyBatchRequest
 		Status: true,
 	}, nil
 }
+
+func (s *dummySequencer) SetBatchSubmissionChan(batchSubmissionChan chan Batch) {
+	s.batchSubmissionChan = batchSubmissionChan
+}

node/full.go

@@ -400,6 +400,7 @@ func (n *FullNode) Run(ctx context.Context) error {
 		go n.blockManager.AggregationLoop(ctx)
 		go n.reaper.Start(ctx)
 		go n.blockManager.HeaderSubmissionLoop(ctx)
+		go n.blockManager.BatchSubmissionLoop(ctx)
 		go n.headerPublishLoop(ctx)
 		go n.dataPublishLoop(ctx)
 		go n.blockManager.DAIncluderLoop(ctx)