feat: implement batch size validation and automatic splitting for DA

.gitignore

@@ -14,4 +14,5 @@ build
 .DS_Store
 coverage.out
 execution/evm/jwttoken
-target
+target
+/.claude/settings.local.json

block/submitter.go

@@ -155,7 +155,28 @@ func submitToDA[T any](
 			m.logger.Info("DA layer submission canceled due to context cancellation", "attempt", attempt)
 			return nil
 		case coreda.StatusTooBig:
-			fallthrough
+			m.logger.Warn("DA layer submission failed due to blob size limit", "error", res.Message, "attempt", attempt, "batchSize", len(remaining))
+			// Record failed DA submission (will retry)
+			m.recordDAMetrics("submission", DAModeFail)
+
+			// Implement batch splitting when blob is too big
+			if len(remaining) > 1 {
+				// Split the batch in half to reduce size
+				splitPoint := len(remaining) / 2
+				m.logger.Info("splitting batch due to size limit", "originalSize", len(remaining), "newSize", splitPoint)
+
+				// Keep only the first half for this attempt
+				remaining = remaining[:splitPoint]
+				marshaled = marshaled[:splitPoint]
+				remLen = len(remaining)
+
+				// Reset backoff since we're trying with a smaller batch
+				backoff = 0
+			} else {
+				// If we have only 1 item and it's still too big, we can't split further
+				m.logger.Error("single item exceeds DA blob size limit", "itemType", itemType, "attempt", attempt)
+				backoff = m.exponentialBackoff(backoff)
+			}
 		default:
 			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
 			// Record failed DA submission (will retry)

block/submitter_test.go

@@ -432,21 +432,21 @@ func TestCreateSignedDataToSubmit(t *testing.T) {
 func fillPendingHeaders(ctx context.Context, t *testing.T, pendingHeaders *PendingHeaders, chainID string, numBlocks uint64) {
 	t.Helper()
 
-	store := pendingHeaders.base.store
+	s := pendingHeaders.base.store
 	for i := uint64(0); i < numBlocks; i++ {
 		height := i + 1
 		header, data := types.GetRandomBlock(height, 0, chainID)
 		sig := &header.Signature
-		err := store.SaveBlockData(ctx, header, data, sig)
+		err := s.SaveBlockData(ctx, header, data, sig)
 		require.NoError(t, err, "failed to save block data for header at height %d", height)
-		err = store.SetHeight(ctx, height)
+		err = s.SetHeight(ctx, height)
 		require.NoError(t, err, "failed to set store height for header at height %d", height)
 	}
 }
 
 func fillPendingData(ctx context.Context, t *testing.T, pendingData *PendingData, chainID string, numBlocks uint64) {
 	t.Helper()
-	store := pendingData.base.store
+	s := pendingData.base.store
 	txNum := 1
 	for i := uint64(0); i < numBlocks; i++ {
 		height := i + 1
@@ -457,9 +457,9 @@ func fillPendingData(ctx context.Context, t *testing.T, pendingData *PendingData
 			txNum++
 		}
 		sig := &header.Signature
-		err := store.SaveBlockData(ctx, header, data, sig)
+		err := s.SaveBlockData(ctx, header, data, sig)
 		require.NoError(t, err, "failed to save block data for data at height %d", height)
-		err = store.SetHeight(ctx, height)
+		err = s.SetHeight(ctx, height)
 		require.NoError(t, err, "failed to set store height for data at height %d", height)
 	}
 }
@@ -565,6 +565,102 @@ func TestSubmitDataToDA_WithMetricsRecorder(t *testing.T) {
 	mockSequencer.AssertExpectations(t)
 }
 
+// TestSubmitToDA_ItChunksBatchWhenSizeExceedsLimit verifies that when DA submission
+// fails with StatusTooBig, the submitter automatically splits the batch in half and
+// retries until successful. This prevents infinite retry loops when batches exceed
+// DA layer size limits.
+func TestSubmitToDA_ItChunksBatchWhenSizeExceedsLimit(t *testing.T) {
+
+	da := &mocks.MockDA{}
+	m := newTestManagerWithDA(t, da)
+	ctx := context.Background()
+
+	// Fill with items that would be too big as a single batch
+	largeItemCount := uint64(10)
+	fillPendingHeaders(ctx, t, m.pendingHeaders, "TestFix", largeItemCount)
+
+	headers, err := m.pendingHeaders.getPendingHeaders(ctx)
+	require.NoError(t, err)
+	require.Len(t, headers, int(largeItemCount))
+
+	var submitAttempts int
+	var batchSizes []int
+
+	// Mock DA behavior:
+	// - First call: full batch (10 items) -> StatusTooBig
+	// - Second call: split batch (5 items) -> Success
+	da.On("SubmitWithOptions", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			submitAttempts++
+			blobs := args.Get(1).([]coreda.Blob)
+			batchSizes = append(batchSizes, len(blobs))
+			t.Logf("DA Submit attempt %d: batch size %d", submitAttempts, len(blobs))
+		}).
+		Return(nil, coreda.ErrBlobSizeOverLimit).Once() // First attempt fails
+
+	da.On("SubmitWithOptions", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			submitAttempts++
+			blobs := args.Get(1).([]coreda.Blob)
+			batchSizes = append(batchSizes, len(blobs))
+			t.Logf("DA Submit attempt %d: batch size %d", submitAttempts, len(blobs))
+		}).
+		Return([]coreda.ID{getDummyID(1, []byte("id1")), getDummyID(1, []byte("id2")), getDummyID(1, []byte("id3")), getDummyID(1, []byte("id4")), getDummyID(1, []byte("id5"))}, nil).Once() // Second attempt succeeds
+
+	err = m.submitHeadersToDA(ctx, headers)
+
+	assert.NoError(t, err, "Should succeed by splitting large batch")
+	assert.Equal(t, 2, submitAttempts, "Should make 2 attempts: 1 large batch + 1 split batch")
+	assert.Equal(t, []int{10, 5}, batchSizes, "Should try full batch, then split in half")
+
+	// The first 5 items should be successfully submitted
+	// Note: Our current implementation only handles one split cycle per call
+	// The remaining 5 items would be handled in the next submission loop iteration
+}
+
+// TestSubmitToDA_SingleItemTooLarge verifies behavior when even a single item
+// exceeds DA size limits and cannot be split further. This should result in
+// exponential backoff and eventual failure after maxSubmitAttempts.
+func TestSubmitToDA_SingleItemTooLarge(t *testing.T) {
+	// Skip this test normally to avoid long wait times, but keep it for manual testing
+	t.Skip("Skipping long-running test. Uncomment to test single item too large scenario manually.")
+
+	da := &mocks.MockDA{}
+	m := newTestManagerWithDA(t, da)
+	ctx := context.Background()
+
+	// Create a single header that will always be "too big"
+	fillPendingHeaders(ctx, t, m.pendingHeaders, "TestSingleLarge", 1)
+
+	headers, err := m.pendingHeaders.getPendingHeaders(ctx)
+	require.NoError(t, err)
+	require.Len(t, headers, 1)
+
+	var submitAttempts int
+
+	// Mock DA to always return "too big" for this single item
+	da.On("SubmitWithOptions", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			submitAttempts++
+			blobs := args.Get(1).([]coreda.Blob)
+			t.Logf("DA Submit attempt %d: batch size %d (single item too large)", submitAttempts, len(blobs))
+		}).
+		Return(nil, coreda.ErrBlobSizeOverLimit) // Always fails
+
+	// This should fail after maxSubmitAttempts (30) attempts
+	err = m.submitHeadersToDA(ctx, headers)
+
+	// Expected behavior: Should fail after exhausting all attempts
+	assert.Error(t, err, "Should fail when single item is too large")
+	assert.Contains(t, err.Error(), "failed to submit all header(s) to DA layer")
+	assert.Contains(t, err.Error(), "after 30 attempts") // maxSubmitAttempts
+
+	// Should have made exactly maxSubmitAttempts (30) attempts
+	assert.Equal(t, 30, submitAttempts, "Should make exactly maxSubmitAttempts before giving up")
+
+	da.AssertExpectations(t)
+}
+
 func getDummyID(height uint64, commitment []byte) coreda.ID {
 	id := make([]byte, len(commitment)+8)
 	binary.LittleEndian.PutUint64(id, height)

da/jsonrpc/client.go

@@ -140,46 +140,34 @@ func (api *API) Submit(ctx context.Context, blobs []da.Blob, gasPrice float64, _
 }
 
 // SubmitWithOptions submits the Blobs to Data Availability layer with additional options.
-// It checks blobs against MaxBlobSize and submits only those that fit.
+// It validates the entire batch against MaxBlobSize before submission.
+// If any blob or the total batch size exceeds limits, it returns ErrBlobSizeOverLimit.
 func (api *API) SubmitWithOptions(ctx context.Context, inputBlobs []da.Blob, gasPrice float64, _ []byte, options []byte) ([]da.ID, error) {
 	maxBlobSize := api.MaxBlobSize
 
-	var (
-		blobsToSubmit [][]byte = make([][]byte, 0, len(inputBlobs))
-		currentSize   uint64
-		oversizeBlobs int
-	)
+	if len(inputBlobs) == 0 {
+		return []da.ID{}, nil
+	}
 
+	// Validate each blob individually and calculate total size
+	var totalSize uint64
 	for i, blob := range inputBlobs {
 		blobLen := uint64(len(blob))
 		if blobLen > maxBlobSize {
-			api.Logger.Warn("Individual blob exceeds MaxBlobSize, cannot submit", "index", i, "blobSize", blobLen, "maxBlobSize", maxBlobSize)
-			oversizeBlobs++
-			continue
-		}
-		if currentSize+blobLen > maxBlobSize {
-			api.Logger.Info("Blob size limit reached for batch", "maxBlobSize", maxBlobSize, "index", i, "currentSize", currentSize, "nextBlobSize", blobLen)
-			break
+			api.Logger.Warn("Individual blob exceeds MaxBlobSize", "index", i, "blobSize", blobLen, "maxBlobSize", maxBlobSize)
+			return nil, da.ErrBlobSizeOverLimit
 		}
-		currentSize += blobLen
-		blobsToSubmit = append(blobsToSubmit, blob)
+		totalSize += blobLen
 	}
 
-	if oversizeBlobs > 0 {
-		api.Logger.Error("Blobs exceeded size limit", "oversize_count", oversizeBlobs, "total_blobs", len(inputBlobs))
+	// Validate total batch size
+	if totalSize > maxBlobSize {
+		api.Logger.Warn("Total batch size exceeds MaxBlobSize", "totalSize", totalSize, "maxBlobSize", maxBlobSize, "numBlobs", len(inputBlobs))
 		return nil, da.ErrBlobSizeOverLimit
 	}
 
-	if len(blobsToSubmit) == 0 {
-		api.Logger.Info("No blobs to submit after filtering by size")
-		if len(inputBlobs) > 0 {
-			return nil, da.ErrBlobSizeOverLimit
-		}
-		return []da.ID{}, nil
-	}
-
-	api.Logger.Debug("Making RPC call", "method", "SubmitWithOptions", "num_blobs_original", len(inputBlobs), "num_blobs_to_submit", len(blobsToSubmit), "gas_price", gasPrice, "namespace", string(api.Namespace))
-	res, err := api.Internal.SubmitWithOptions(ctx, blobsToSubmit, gasPrice, api.Namespace, options)
+	api.Logger.Debug("Making RPC call", "method", "SubmitWithOptions", "num_blobs", len(inputBlobs), "total_size", totalSize, "gas_price", gasPrice, "namespace", string(api.Namespace))
+	res, err := api.Internal.SubmitWithOptions(ctx, inputBlobs, gasPrice, api.Namespace, options)
 	if err != nil {
 		if strings.Contains(err.Error(), context.Canceled.Error()) {
 			api.Logger.Debug("RPC call canceled due to context cancellation", "method", "SubmitWithOptions")

da/jsonrpc/client_test.go

@@ -0,0 +1,126 @@
+package jsonrpc
+
+import (
+	"context"
+	"testing"
+
+	logging "github.com/ipfs/go-log/v2"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/evstack/ev-node/core/da"
+)
+
+// TestSubmitWithOptions_SizeValidation tests the corrected behavior of SubmitWithOptions
+// where it validates the entire batch before submission and returns ErrBlobSizeOverLimit
+// if the batch is too large, instead of silently dropping blobs.
+func TestSubmitWithOptions_SizeValidation(t *testing.T) {
+	logger := logging.Logger("test")
+
+	testCases := []struct {
+		name          string
+		maxBlobSize   uint64
+		inputBlobs    []da.Blob
+		expectError   bool
+		expectedError error
+		description   string
+	}{
+		{
+			name:        "Empty input",
+			maxBlobSize: 1000,
+			inputBlobs:  []da.Blob{},
+			expectError: false,
+			description: "Empty input should return empty result without error",
+		},
+		{
+			name:        "Single blob within limit",
+			maxBlobSize: 1000,
+			inputBlobs:  []da.Blob{make([]byte, 500)},
+			expectError: false,
+			description: "Single blob smaller than limit should succeed",
+		},
+		{
+			name:          "Single blob exceeds limit",
+			maxBlobSize:   1000,
+			inputBlobs:    []da.Blob{make([]byte, 1500)},
+			expectError:   true,
+			expectedError: da.ErrBlobSizeOverLimit,
+			description:   "Single blob larger than limit should fail",
+		},
+		{
+			name:        "Multiple blobs within limit",
+			maxBlobSize: 1000,
+			inputBlobs:  []da.Blob{make([]byte, 300), make([]byte, 400), make([]byte, 200)},
+			expectError: false,
+			description: "Multiple blobs totaling less than limit should succeed",
+		},
+		{
+			name:          "Multiple blobs exceed total limit",
+			maxBlobSize:   1000,
+			inputBlobs:    []da.Blob{make([]byte, 400), make([]byte, 400), make([]byte, 400)},
+			expectError:   true,
+			expectedError: da.ErrBlobSizeOverLimit,
+			description:   "Multiple blobs totaling more than limit should fail completely",
+		},
+		{
+			name:          "Mixed: some blobs fit, total exceeds limit",
+			maxBlobSize:   1000,
+			inputBlobs:    []da.Blob{make([]byte, 100), make([]byte, 200), make([]byte, 800)},
+			expectError:   true,
+			expectedError: da.ErrBlobSizeOverLimit,
+			description:   "Should fail completely, not partially submit blobs that fit",
+		},
+		{
+			name:          "One blob exceeds limit individually",
+			maxBlobSize:   1000,
+			inputBlobs:    []da.Blob{make([]byte, 300), make([]byte, 1500), make([]byte, 200)},
+			expectError:   true,
+			expectedError: da.ErrBlobSizeOverLimit,
+			description:   "Should fail if any individual blob exceeds limit",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Create API with test configuration
+			api := &API{
+				Logger:      logger,
+				MaxBlobSize: tc.maxBlobSize,
+				Namespace:   []byte("test"),
+			}
+
+			// Mock the Internal.SubmitWithOptions to always succeed if called
+			// This tests that our validation logic works before reaching the actual RPC call
+			mockCalled := false
+			api.Internal.SubmitWithOptions = func(ctx context.Context, blobs []da.Blob, gasPrice float64, namespace []byte, options []byte) ([]da.ID, error) {
+				mockCalled = true
+				// Return mock IDs for successful submissions
+				ids := make([]da.ID, len(blobs))
+				for i := range blobs {
+					ids[i] = []byte{byte(i)}
+				}
+				return ids, nil
+			}
+
+			// Call SubmitWithOptions
+			ctx := context.Background()
+			result, err := api.SubmitWithOptions(ctx, tc.inputBlobs, 1.0, nil, nil)
+
+			// Verify expectations
+			if tc.expectError {
+				assert.Error(t, err, tc.description)
+				if tc.expectedError != nil {
+					assert.ErrorIs(t, err, tc.expectedError, tc.description)
+				}
+				assert.Nil(t, result, "Result should be nil on error")
+				assert.False(t, mockCalled, "Internal RPC should not be called when validation fails")
+			} else {
+				assert.NoError(t, err, tc.description)
+				assert.NotNil(t, result, "Result should not be nil on success")
+				if len(tc.inputBlobs) > 0 {
+					assert.True(t, mockCalled, "Internal RPC should be called for valid submissions")
+					assert.Len(t, result, len(tc.inputBlobs), "Should return IDs for all submitted blobs")
+				}
+			}
+		})
+	}
+}