Refactor out submitAllTxs

Author: Manav-Aggarwal

block/submitter.go

@@ -146,7 +146,7 @@ func (m *Manager) BatchSubmissionLoop(ctx context.Context) {
 				continue
 			}
 
-			err = m.submitBatchToDA(ctx, signedData)
+			err = m.submitDataToDA(ctx, signedData)
 			if err != nil {
 				m.logger.Error("failed to submit batch to DA", "error", err)
 			}
@@ -185,7 +185,7 @@ func (m *Manager) createSignedDataFromBatch(batch *coresequencer.Batch) (*types.
 	}, nil
 }
 
-// submitBatchToDA submits a batch of transactions to the Data Availability (DA) layer.
+// submitBatchToDA submits signed data to the Data Availability (DA) layer.
 // It implements a retry mechanism with exponential backoff and gas price adjustments
 // to handle various failure scenarios.
 //
@@ -198,28 +198,25 @@ func (m *Manager) createSignedDataFromBatch(batch *coresequencer.Batch) (*types.
 // - On other errors: Uses exponential backoff
 //
 // It returns an error if not all transactions could be submitted after all attempts.
-func (m *Manager) submitBatchToDA(ctx context.Context, signedData *types.SignedData) error {
-	currentSignedData := signedData
-	submittedAllTxs := false
+func (m *Manager) submitDataToDA(ctx context.Context, signedData *types.SignedData) error {
 	var backoff time.Duration
-	totalTxCount := len(currentSignedData.Txs)
-	submittedTxCount := 0
 	attempt := 0
+	submitted := false
 
 	// Store initial values to be able to reset or compare later
 	initialGasPrice := m.gasPrice
 	gasPrice := initialGasPrice
 
 daSubmitRetryLoop:
-	for !submittedAllTxs && attempt < maxSubmitAttempts {
+	for !submitted && attempt < maxSubmitAttempts {
 		// Wait for backoff duration or exit if context is done
 		select {
 		case <-ctx.Done():
 			break daSubmitRetryLoop
 		case <-time.After(backoff):
 		}
 
-		signedDataBz, err := currentSignedData.MarshalBinary()
+		signedDataBz, err := signedData.MarshalBinary()
 		if err != nil {
 			return fmt.Errorf("failed to marshal signed data: %w", err)
 		}
@@ -235,22 +232,11 @@ daSubmitRetryLoop:
 
 		switch res.Code {
 		case coreda.StatusSuccess:
-			submittedTxs := int(res.SubmittedCount)
-			m.logger.Info("successfully submitted transactions to DA layer",
+			m.logger.Info("successfully submitted data to DA layer",
 				"gasPrice", gasPrice,
-				"height", res.Height,
-				"submittedTxs", submittedTxs,
-				"remainingTxs", len(currentSignedData.Txs)-submittedTxs)
-
-			submittedTxCount += submittedTxs
-
-			// Check if all transactions in the current batch were submitted
-			if submittedTxs == len(currentSignedData.Txs) {
-				submittedAllTxs = true
-			} else {
-				// Update the current batch to contain only the remaining transactions
-				currentSignedData.Txs = currentSignedData.Txs[submittedTxs:]
-			}
+				"height", res.Height)
+
+			submitted = true
 
 			// Reset submission parameters after success
 			backoff = 0
@@ -263,11 +249,10 @@ daSubmitRetryLoop:
 				}
 			}
 			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 {
-				m.DataCache().SetDAIncluded(signedData.DACommitment().String())
-				m.sendNonBlockingSignalToDAIncluderCh()
-			}
+
+			m.DataCache().SetDAIncluded(signedData.DACommitment().String())
+			m.sendNonBlockingSignalToDAIncluderCh()
+			return nil
 
 		case coreda.StatusNotIncludedInBlock, coreda.StatusAlreadyInMempool:
 			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
@@ -291,11 +276,9 @@ daSubmitRetryLoop:
 	}
 
 	// Return error if not all transactions were submitted after all attempts
-	if !submittedAllTxs {
+	if !submitted {
 		return fmt.Errorf(
-			"failed to submit all transactions to DA layer, submitted %d txs (%d left) after %d attempts",
-			submittedTxCount,
-			totalTxCount-submittedTxCount,
+			"failed to submit data to DA layer after %d attempts",
 			attempt,
 		)
 	}

block/submitter_test.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/libp2p/go-libp2p/core/crypto"
 	coreda "github.com/rollkit/rollkit/core/da"
-	coresequencer "github.com/rollkit/rollkit/core/sequencer"
 	"github.com/rollkit/rollkit/pkg/cache"
 	"github.com/rollkit/rollkit/pkg/config"
 	"github.com/rollkit/rollkit/pkg/signer/noop"
@@ -41,13 +40,11 @@ func newTestManagerWithDA(t *testing.T, da *mocks.DA) (m *Manager) {
 	}
 }
 
-// TestSubmitBatchToDA_Success verifies that submitBatchToDA succeeds when the DA layer accepts the batch.
-func TestSubmitBatchToDA_Success(t *testing.T) {
+// TestSubmitDataToDA_Success verifies that submitDataToDA succeeds when the DA layer accepts the batch.
+func TestSubmitDataToDA_Success(t *testing.T) {
 	da := &mocks.DA{}
 	m := newTestManagerWithDA(t, da)
 
-	batch := coresequencer.Batch{Transactions: [][]byte{[]byte("tx1"), []byte("tx2")}}
-
 	// Simulate DA success
 	da.On("GasMultiplier", mock.Anything).Return(2.0, nil)
 	da.On("SubmitWithOptions", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
@@ -58,36 +55,36 @@ func TestSubmitBatchToDA_Success(t *testing.T) {
 	addr, err := m.signer.GetAddress()
 	require.NoError(t, err)
 
+	transactions := [][]byte{[]byte("tx1"), []byte("tx2")}
+
 	signedData := types.SignedData{
 		Data: types.Data{
-			Txs: make(types.Txs, len(batch.Transactions)),
+			Txs: make(types.Txs, len(transactions)),
 		},
 		Signer: types.Signer{
 			Address: addr,
 			PubKey:  pubKey,
 		},
 	}
 
-	signedData.Txs = make(types.Txs, len(batch.Transactions))
-	for i, tx := range batch.Transactions {
+	signedData.Txs = make(types.Txs, len(transactions))
+	for i, tx := range transactions {
 		signedData.Txs[i] = types.Tx(tx)
 	}
 
 	signature, err := m.getDataSignature(&signedData.Data)
 	require.NoError(t, err)
 	signedData.Signature = signature
 
-	err = m.submitBatchToDA(context.Background(), &signedData)
+	err = m.submitDataToDA(context.Background(), &signedData)
 	assert.NoError(t, err)
 }
 
-// TestSubmitBatchToDA_Failure verifies that submitBatchToDA returns an error for various DA failures.
-func TestSubmitBatchToDA_Failure(t *testing.T) {
+// TestSubmitDataToDA_Failure verifies that submitDataToDA returns an error for various DA failures.
+func TestSubmitDataToDA_Failure(t *testing.T) {
 	da := &mocks.DA{}
 	m := newTestManagerWithDA(t, da)
 
-	batch := coresequencer.Batch{Transactions: [][]byte{[]byte("tx1"), []byte("tx2")}}
-
 	// Table-driven test for different DA error scenarios
 	testCases := []struct {
 		name    string
@@ -110,18 +107,20 @@ func TestSubmitBatchToDA_Failure(t *testing.T) {
 			addr, err := m.signer.GetAddress()
 			require.NoError(t, err)
 
+			transactions := [][]byte{[]byte("tx1"), []byte("tx2")}
+
 			signedData := types.SignedData{
 				Data: types.Data{
-					Txs: make(types.Txs, len(batch.Transactions)),
+					Txs: make(types.Txs, len(transactions)),
 				},
 				Signer: types.Signer{
 					Address: addr,
 					PubKey:  pubKey,
 				},
 			}
 
-			signedData.Txs = make(types.Txs, len(batch.Transactions))
-			for i, tx := range batch.Transactions {
+			signedData.Txs = make(types.Txs, len(transactions))
+			for i, tx := range transactions {
 				signedData.Txs[i] = types.Tx(tx)
 			}
 
@@ -130,7 +129,7 @@ func TestSubmitBatchToDA_Failure(t *testing.T) {
 			signedData.Signature = signature
 
 			// Expect an error from submitBatchToDA
-			err = m.submitBatchToDA(context.Background(), &signedData)
+			err = m.submitDataToDA(context.Background(), &signedData)
 			assert.Error(t, err, "expected error")
 		})
 	}