[Feature]: Submit all blocks possible (#1607)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.
-->

## Overview
closes #1590 
<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 
-->
- This pr updates the logic of `DAClient.SubmitBlocks` so now it can
submit all blocks possible
## Checklist

<!-- 
Please complete the checklist to ensure that the PR is ready to be
reviewed.

IMPORTANT:
PRs should be left in Draft until the below checklist is completed.
-->

- [ ] New and updated code has appropriate documentation
- [ ] New and updated code has new and/or updated testing
- [ ] Required CI checks are passing
- [ ] Visual proof for any user facing features like CLI or
documentation updates
- [ ] Linked issues closed with keywords


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit


- **Bug Fixes**
- Improved error handling and messaging for block submission failures
due to serialization issues or size limits in the `SubmitBlocks` method.
- **Tests**
- Updated test cases `Test_submitBlocksToDA_BlockMarshalErrorCase1` and
`Test_submitBlocksToDA_BlockMarshalErrorCase2` to reflect changes in
block submission and error messaging.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: ThanhNhann

block/manager_test.go

@@ -329,9 +329,9 @@ func Test_submitBlocksToDA_BlockMarshalErrorCase1(t *testing.T) {
 
 	m := getManager(t, goDATest.NewDummyDA())
 
-	block1 := types.GetRandomBlock(uint64(0), 5)
-	block2 := types.GetRandomBlock(uint64(1), 5)
-	block3 := types.GetRandomBlock(uint64(2), 5)
+	block1 := types.GetRandomBlock(uint64(1), 5)
+	block2 := types.GetRandomBlock(uint64(2), 5)
+	block3 := types.GetRandomBlock(uint64(3), 5)
 
 	store := mocks.NewStore(t)
 	invalidateBlockHeader(block1)
@@ -354,20 +354,22 @@ func Test_submitBlocksToDA_BlockMarshalErrorCase1(t *testing.T) {
 	assert.Equal(3, len(blocks))
 }
 
-// Test_submitBlocksToDA_BlockMarshalErrorCase2: A and B are fair blocks, but C has a marshalling error. None of the blocks get submitted to DA layer.
+// Test_submitBlocksToDA_BlockMarshalErrorCase2: A and B are fair blocks, but C has a marshalling error
+// - Block A and B get submitted to DA layer not block C
 func Test_submitBlocksToDA_BlockMarshalErrorCase2(t *testing.T) {
 	assert := assert.New(t)
 	require := require.New(t)
 	ctx := context.Background()
 
 	m := getManager(t, goDATest.NewDummyDA())
 
-	block1 := types.GetRandomBlock(uint64(0), 5)
-	block2 := types.GetRandomBlock(uint64(1), 5)
-	block3 := types.GetRandomBlock(uint64(2), 5)
+	block1 := types.GetRandomBlock(uint64(1), 5)
+	block2 := types.GetRandomBlock(uint64(2), 5)
+	block3 := types.GetRandomBlock(uint64(3), 5)
 
 	store := mocks.NewStore(t)
 	invalidateBlockHeader(block3)
+	store.On("SetMetadata", ctx, LastSubmittedHeightKey, []byte(strconv.FormatUint(2, 10))).Return(nil)
 	store.On("GetMetadata", ctx, LastSubmittedHeightKey).Return(nil, ds.ErrNotFound)
 	store.On("GetBlock", ctx, uint64(1)).Return(block1, nil)
 	store.On("GetBlock", ctx, uint64(2)).Return(block2, nil)
@@ -379,12 +381,11 @@ func Test_submitBlocksToDA_BlockMarshalErrorCase2(t *testing.T) {
 	var err error
 	m.pendingBlocks, err = NewPendingBlocks(store, m.logger)
 	require.NoError(err)
-
 	err = m.submitBlocksToDA(ctx)
 	assert.ErrorContains(err, "failed to submit all blocks to DA layer")
 	blocks, err := m.pendingBlocks.getPendingBlocks(ctx)
 	assert.NoError(err)
-	assert.Equal(3, len(blocks))
+	assert.Equal(1, len(blocks))
 }
 
 // invalidateBlockHeader results in a block header that produces a marshalling error

da/da.go

@@ -119,35 +119,35 @@ func NewDAClient(da goDA.DA, gasPrice, gasMultiplier float64, ns goDA.Namespace,
 
 // SubmitBlocks submits blocks to DA.
 func (dac *DAClient) SubmitBlocks(ctx context.Context, blocks []*types.Block, maxBlobSize uint64, gasPrice float64) ResultSubmitBlocks {
-	var blobs [][]byte
-	var blobSize uint64
-	var submitted uint64
+	var (
+		blobs    [][]byte
+		blobSize uint64
+		message  string
+	)
 	for i := range blocks {
 		blob, err := blocks[i].MarshalBinary()
 		if err != nil {
-			return ResultSubmitBlocks{
-				BaseResult: BaseResult{
-					Code:    StatusError,
-					Message: "failed to serialize block",
-				},
-			}
+			message = fmt.Sprint("failed to serialize block", err)
+			dac.Logger.Info(message)
+			break
 		}
 		if blobSize+uint64(len(blob)) > maxBlobSize {
-			dac.Logger.Info("blob size limit reached", "maxBlobSize", maxBlobSize, "index", i, "blobSize", blobSize, "len(blob)", len(blob))
+			message = fmt.Sprint(ErrBlobSizeOverLimit.Error(), "blob size limit reached", "maxBlobSize", maxBlobSize, "index", i, "blobSize", blobSize, "len(blob)", len(blob))
+			dac.Logger.Info(message)
 			break
 		}
 		blobSize += uint64(len(blob))
-		submitted += 1
 		blobs = append(blobs, blob)
 	}
-	if submitted == 0 {
+	if len(blobs) == 0 {
 		return ResultSubmitBlocks{
 			BaseResult: BaseResult{
 				Code:    StatusError,
-				Message: "failed to submit blocks: oversized block: " + ErrBlobSizeOverLimit.Error(),
+				Message: "failed to submit blocks: no blobs generated " + message,
 			},
 		}
 	}
+
 	ctx, cancel := context.WithTimeout(ctx, dac.SubmitTimeout)
 	defer cancel()
 	ids, err := dac.DA.Submit(ctx, blobs, gasPrice, dac.Namespace)
@@ -186,7 +186,7 @@ func (dac *DAClient) SubmitBlocks(ctx context.Context, blocks []*types.Block, ma
 		BaseResult: BaseResult{
 			Code:           StatusSuccess,
 			DAHeight:       binary.LittleEndian.Uint64(ids[0]),
-			SubmittedCount: submitted,
+			SubmittedCount: uint64(len(ids)),
 		},
 	}
 }

da/da_test.go

@@ -274,7 +274,7 @@ func doTestSubmitOversizedBlock(t *testing.T, dalc *DAClient) {
 	oversizedBlock := types.GetRandomBlock(1, int(limit))
 	resp := dalc.SubmitBlocks(ctx, []*types.Block{oversizedBlock}, limit, -1)
 	assert.Equal(StatusError, resp.Code, "oversized block should throw error")
-	assert.Contains(resp.Message, "failed to submit blocks: oversized block: blob: over size limit")
+	assert.Contains(resp.Message, "failed to submit blocks: no blobs generated blob: over size limit")
 }
 
 func doTestSubmitSmallBlocksBatch(t *testing.T, dalc *DAClient) {