dac: max blob size (#1385)

<!--
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

Fixes #1233

<!--
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.
-->

## 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.
-->

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

Depends on evstack/go-da#23

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

- **New Features**
- Improved block submission process with retries and exponential backoff
for increased reliability.
- Added error handling for block size limitations and existence issues
during submission.

- **Enhancements**
- Block submission now includes a count of successfully submitted blocks
for better tracking.
- Partial reset functionality added to the management of pending blocks
based on submission status.

- **Bug Fixes**
- Fixed an issue where block submission could fail silently by ensuring
all pending blocks are confirmed as submitted before completion.

- **Documentation**
- Updated function documentation to reflect new error handling and
submission count features.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: MSevey

block/manager.go

@@ -738,21 +738,26 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error {
 	submitted := false
 	backoff := initialBackoff
 	for attempt := 1; ctx.Err() == nil && !submitted && attempt <= maxSubmitAttempts; attempt++ {
-		res := m.dalc.SubmitBlocks(ctx, m.pendingBlocks.getPendingBlocks())
-		if res.Code == da.StatusSuccess {
-			m.logger.Info("successfully submitted Rollkit block to DA layer", "daHeight", res.DAHeight)
-			submitted = true
-		} else {
+		blocks := m.pendingBlocks.getPendingBlocks()
+		res := m.dalc.SubmitBlocks(ctx, blocks)
+		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.pendingBlocks.resetPendingBlocks(res.SubmittedCount)
+		case da.StatusError, da.StatusNotFound, da.StatusUnknown:
 			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
 			time.Sleep(backoff)
 			backoff = m.exponentialBackoff(backoff)
+		default:
 		}
 	}
 
 	if !submitted {
 		return fmt.Errorf("failed to submit block to DA layer after %d attempts", maxSubmitAttempts)
 	}
-	m.pendingBlocks.resetPendingBlocks()
 	return nil
 }
 

block/pending_blocks.go

@@ -37,8 +37,11 @@ func (pb *PendingBlocks) addPendingBlock(block *types.Block) {
 	pb.pendingBlocks = append(pb.pendingBlocks, block)
 }
 
-func (pb *PendingBlocks) resetPendingBlocks() {
+func (pb *PendingBlocks) resetPendingBlocks(submitted uint64) {
 	pb.mtx.Lock()
 	defer pb.mtx.Unlock()
-	pb.pendingBlocks = make([]*types.Block, 0)
+	if submitted > uint64(len(pb.pendingBlocks)) {
+		submitted = uint64(len(pb.pendingBlocks))
+	}
+	pb.pendingBlocks = pb.pendingBlocks[submitted:]
 }

da/da.go

@@ -14,8 +14,13 @@ import (
 	pb "github.com/rollkit/rollkit/types/pb/rollkit"
 )
 
-// ErrBlobNotFound is used to indicate that the blob was not found.
-var ErrBlobNotFound = errors.New("blob: not found")
+var (
+	// ErrBlobNotFound is used to indicate that the blob was not found.
+	ErrBlobNotFound = errors.New("blob: not found")
+
+	// ErrBlobSizeOverLimit is used to indicate that the blob size is over limit
+	ErrBlobSizeOverLimit = errors.New("blob: over size limit")
+)
 
 // StatusCode is a type for DA layer return status.
 // TODO: define an enum of different non-happy-path cases
@@ -39,6 +44,8 @@ type BaseResult struct {
 	Message string
 	// DAHeight informs about a height on Data Availability Layer for given result.
 	DAHeight uint64
+	// SubmittedCount is the number of successfully submitted blocks.
+	SubmittedCount uint64
 }
 
 // ResultSubmitBlocks contains information returned from DA layer after blocks submission.
@@ -65,7 +72,18 @@ type DAClient struct {
 
 // SubmitBlocks submits blocks to DA.
 func (dac *DAClient) SubmitBlocks(ctx context.Context, blocks []*types.Block) ResultSubmitBlocks {
-	blobs := make([][]byte, len(blocks))
+	var blobs [][]byte
+	var blobSize uint64
+	maxBlobSize, err := dac.DA.MaxBlobSize()
+	if err != nil {
+		return ResultSubmitBlocks{
+			BaseResult: BaseResult{
+				Code:    StatusError,
+				Message: "unable to get DA max blob size",
+			},
+		}
+	}
+	var submitted uint64
 	for i := range blocks {
 		blob, err := blocks[i].MarshalBinary()
 		if err != nil {
@@ -76,7 +94,21 @@ func (dac *DAClient) SubmitBlocks(ctx context.Context, blocks []*types.Block) Re
 				},
 			}
 		}
-		blobs[i] = blob
+		if blobSize+uint64(len(blob)) > maxBlobSize {
+			dac.Logger.Info("blob size limit reached", "maxBlobSize", maxBlobSize, "index", i, "blobSize", blobSize, "len(blob)", len(blob))
+			break
+		}
+		blobSize += uint64(len(blob))
+		submitted += 1
+		blobs = append(blobs, blob)
+	}
+	if submitted == 0 {
+		return ResultSubmitBlocks{
+			BaseResult: BaseResult{
+				Code:    StatusError,
+				Message: "failed to submit blocks: oversized block: " + ErrBlobSizeOverLimit.Error(),
+			},
+		}
 	}
 	ids, _, err := dac.DA.Submit(blobs)
 	if err != nil {
@@ -88,10 +120,20 @@ func (dac *DAClient) SubmitBlocks(ctx context.Context, blocks []*types.Block) Re
 		}
 	}
 
+	if len(ids) == 0 {
+		return ResultSubmitBlocks{
+			BaseResult: BaseResult{
+				Code:    StatusError,
+				Message: "failed to submit blocks: unexpected len(ids): 0",
+			},
+		}
+	}
+
 	return ResultSubmitBlocks{
 		BaseResult: BaseResult{
-			Code:     StatusSuccess,
-			DAHeight: binary.LittleEndian.Uint64(ids[0]),
+			Code:           StatusSuccess,
+			DAHeight:       binary.LittleEndian.Uint64(ids[0]),
+			SubmittedCount: submitted,
 		},
 	}
 }

da/da.md

@@ -8,6 +8,8 @@ Rollkit provides a wrapper for [go-da][go-da], a generic data availability inter
 
 Given a set of blocks to be submitted to DA by the block manager, the `SubmitBlocks` first encodes the blocks using protobuf (the encoded data are called blobs) and invokes the `Submit` method on the underlying DA implementation. On successful submission (`StatusSuccess`), the DA block height which included in the rollup blocks is returned.
 
+To make sure that the serialised blocks don't exceed the underlying DA's blob limits, it fetches the the blob size limit by calling `Config` which returns the limit as `uint64` bytes, then includes serialised blocks until the limit is reached. If the limit is reached, it submits the partial set and returns the count of successfully submitted blocks as `SubmittedCount`. The caller should retry with the remaining blocks until all the blocks are submitted. If the first block itself is over the limit, it throws an error.
+
 The `Submit` call may result in an error (`StatusError`) based on the underlying DA implementations on following scenarios:
 
 * the total blobs size exceeds the underlying DA's limits (includes empty blobs)