Handle duplicate block submission

block/manager.go

@@ -827,35 +827,42 @@ func (m *Manager) recordMetrics(block *types.Block) {
 }
 
 func (m *Manager) submitBlocksToDA(ctx context.Context) error {
-	submitted := false
+	submittedAllBlocks := false
 	backoff := initialBackoff
-	blocks := m.pendingBlocks.getPendingBlocks()
-	for attempt := 1; ctx.Err() == nil && !submitted && attempt <= maxSubmitAttempts; attempt++ {
-		res := m.dalc.SubmitBlocks(ctx, blocks)
+	blocksToSubmit := m.pendingBlocks.getPendingBlocks()
+	numTotalBlocks := len(blocksToSubmit)
+	numSubmittedBlocks := 0
+	attempt := 0
+	for ctx.Err() == nil && !submittedAllBlocks && attempt < maxSubmitAttempts {
+		res := m.dalc.SubmitBlocks(ctx, blocksToSubmit)
 		switch res.Code {
 		case da.StatusSuccess:
-			m.logger.Info("successfully submitted Rollkit block to DA layer", "daHeight", res.DAHeight, "count", res.SubmittedCount)
-			if int(res.SubmittedCount) == len(blocks) {
-				submitted = true
+			m.logger.Info("successfully submitted Rollkit blocks to DA layer", "daHeight", res.DAHeight, "count", res.SubmittedCount)
+			if res.SubmittedCount == uint64(len(blocksToSubmit)) {
+				submittedAllBlocks = true
 			}
-			submittedBlocks := blocks[:res.SubmittedCount]
+			submittedBlocks, notSubmittedBlocks := blocksToSubmit[:res.SubmittedCount], blocksToSubmit[res.SubmittedCount:]
+			numSubmittedBlocks += len(submittedBlocks)
 			for _, block := range submittedBlocks {
 				m.blockCache.setDAIncluded(block.Hash().String())
 			}
 			m.pendingBlocks.removeSubmittedBlocks(submittedBlocks)
-		case da.StatusError, da.StatusNotFound:
-			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
-			time.Sleep(backoff)
-			backoff = m.exponentialBackoff(backoff)
+			blocksToSubmit = notSubmittedBlocks
 		default:
-			m.logger.Error("DA layer unknown status", "error", res.Message, "attempt", attempt)
+			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
 			time.Sleep(backoff)
 			backoff = m.exponentialBackoff(backoff)
 		}
+		attempt += 1
 	}
 
-	if !submitted {
-		return fmt.Errorf("failed to submit block to DA layer after %d attempts", maxSubmitAttempts)
+	if !submittedAllBlocks {
+		return fmt.Errorf(
+			"failed to submit all blocks to DA layer, submitted %d of %d blocks after %d attempts",
+			numSubmittedBlocks,
+			numTotalBlocks,
+			maxSubmitAttempts,
+		)
 	}
 	return nil
 }

block/manager_test.go

@@ -5,12 +5,42 @@ import (
 	"testing"
 
 	cmtypes "github.com/cometbft/cometbft/types"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	goDATest "github.com/rollkit/go-da/test"
+
+	"github.com/rollkit/rollkit/da"
 	"github.com/rollkit/rollkit/store"
+	test "github.com/rollkit/rollkit/test/log"
 	"github.com/rollkit/rollkit/types"
 )
 
+// Returns a minimalistic block manager
+func getManager(t *testing.T) *Manager {
+	logger := test.NewFileLoggerCustom(t, test.TempLogFileName(t, t.Name()))
+	return &Manager{
+		dalc:       &da.DAClient{DA: goDATest.NewDummyDA(), GasPrice: -1, Logger: logger},
+		blockCache: NewBlockCache(),
+		logger:     logger,
+	}
+}
+
+// getBlockBiggerThan generates a block with the given height bigger than the specified limit.
+func getBlockBiggerThan(blockHeight, limit uint64) (*types.Block, error) {
+	for numTxs := 0; ; numTxs += 100 {
+		block := types.GetRandomBlock(blockHeight, numTxs)
+		blob, err := block.MarshalBinary()
+		if err != nil {
+			return nil, err
+		}
+
+		if uint64(len(blob)) > limit {
+			return block, nil
+		}
+	}
+}
+
 func TestInitialStateClean(t *testing.T) {
 	require := require.New(t)
 	genesisDoc, _ := types.GetGenesisWithPrivkey()
@@ -94,3 +124,75 @@ func TestIsDAIncluded(t *testing.T) {
 	m.blockCache.setDAIncluded(hash.String())
 	require.True(m.IsDAIncluded(hash))
 }
+
+func TestSubmitBlocksToDA(t *testing.T) {
+	require := require.New(t)
+	ctx := context.Background()
+
+	m := getManager(t)
+
+	maxDABlobSizeLimit, err := m.dalc.DA.MaxBlobSize(ctx)
+	require.NoError(err)
+
+	testCases := []struct {
+		name                        string
+		blocks                      []*types.Block
+		isErrExpected               bool
+		expectedPendingBlocksLength int
+	}{
+		{
+			name:                        "happy path, all blocks A, B, C combine to less than maxDABlobSize",
+			blocks:                      []*types.Block{types.GetRandomBlock(1, 5), types.GetRandomBlock(2, 5), types.GetRandomBlock(3, 5)},
+			isErrExpected:               false,
+			expectedPendingBlocksLength: 0,
+		},
+		{
+			name: "blocks A and B are submitted together without C because including C triggers blob size limit. C is submitted in a separate round",
+			blocks: func() []*types.Block {
+				// Find three blocks where two of them are under blob size limit
+				// but adding the third one exceeds the blob size limit
+				block1 := types.GetRandomBlock(1, 100)
+				blob1, err := block1.MarshalBinary()
+				require.NoError(err)
+
+				block2 := types.GetRandomBlock(2, 100)
+				blob2, err := block2.MarshalBinary()
+				require.NoError(err)
+
+				block3, err := getBlockBiggerThan(3, maxDABlobSizeLimit-uint64(len(blob1)+len(blob2)))
+				require.NoError(err)
+
+				return []*types.Block{block1, block2, block3}
+			}(),
+			isErrExpected:               false,
+			expectedPendingBlocksLength: 0,
+		},
+		{
+			name: "A and B are submitted successfully but C is too big on its own, so C never gets submitted",
+			blocks: func() []*types.Block {
+				numBlocks, numTxs := 3, 5
+				blocks := make([]*types.Block, numBlocks)
+				for i := 0; i < numBlocks-1; i++ {
+					blocks[i] = types.GetRandomBlock(uint64(i+1), numTxs)
+				}
+				blocks[2], err = getBlockBiggerThan(3, maxDABlobSizeLimit)
+				require.NoError(err)
+				return blocks
+			}(),
+			isErrExpected:               true,
+			expectedPendingBlocksLength: 1,
+		},
+	}
+
+	for _, tc := range testCases {
+		m.pendingBlocks = NewPendingBlocks()
+		t.Run(tc.name, func(t *testing.T) {
+			for _, block := range tc.blocks {
+				m.pendingBlocks.addPendingBlock(block)
+			}
+			err := m.submitBlocksToDA(ctx)
+			assert.Equal(t, tc.isErrExpected, err != nil)
+			assert.Equal(t, tc.expectedPendingBlocksLength, len(m.pendingBlocks.getPendingBlocks()))
+		})
+	}
+}