refactor: polish submitter package (#2661)

## Overview

Better test coverage and minor refactoring.


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

Ex: Closes #<issue number>
-->

---------

Co-authored-by: julienrbrt <julien@rbrt.fr>

Author: alpe

block/internal/submitting/da_retry_test.go

@@ -3,10 +3,12 @@ package submitting
 import (
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestRetryStateNext_Table(t *testing.T) {
-	pol := RetryPolicy{
+	pol := retryPolicy{
 		MaxAttempts:      10,
 		MinBackoff:       100 * time.Millisecond,
 		MaxBackoff:       1 * time.Second,
@@ -16,105 +18,97 @@ func TestRetryStateNext_Table(t *testing.T) {
 		MaxGasMultiplier: 3.0,
 	}
 
-	tests := []struct {
-		name          string
+	tests := map[string]struct {
 		startGas      float64
 		startBackoff  time.Duration
 		reason        retryReason
-		gm            float64
+		gasMult       float64
 		sentinelNoGas bool
 		wantGas       float64
 		wantBackoff   time.Duration
 	}{
-		{
-			name:         "success reduces gas and resets backoff",
+		"success reduces gas and resets backoff": {
 			startGas:     9.0,
 			startBackoff: 500 * time.Millisecond,
 			reason:       reasonSuccess,
-			gm:           3.0,
+			gasMult:      3.0,
 			wantGas:      3.0, // 9 / 3
 			wantBackoff:  pol.MinBackoff,
 		},
-		{
-			name:         "success clamps very small gm to 1/Max, possibly increasing gas",
+		"success clamps very small gasMult to 1/Max, possibly increasing gas": {
 			startGas:     3.0,
 			startBackoff: 250 * time.Millisecond,
 			reason:       reasonSuccess,
-			gm:           0.01, // clamped to 1/MaxGasMultiplier = 1/3
+			gasMult:      0.01, // clamped to 1/MaxGasMultiplier = 1/3
 			wantGas:      9.0,  // 3 / (1/3)
 			wantBackoff:  pol.MinBackoff,
 		},
-		{
-			name:         "mempool increases gas and sets max backoff",
+		"mempool increases gas and sets max backoff": {
 			startGas:     2.0,
 			startBackoff: 0,
 			reason:       reasonMempool,
-			gm:           2.0,
+			gasMult:      2.0,
 			wantGas:      4.0, // 2 * 2
 			wantBackoff:  pol.MaxBackoff,
 		},
-		{
-			name:         "mempool clamps gas to max",
+		"mempool clamps gas to max": {
 			startGas:     9.5,
 			startBackoff: 0,
 			reason:       reasonMempool,
-			gm:           3.0,
+			gasMult:      3.0,
 			wantGas:      10.0, // 9.5 * 3 = 28.5 -> clamp 10
 			wantBackoff:  pol.MaxBackoff,
 		},
-		{
-			name:         "failure sets initial backoff",
+		"failure sets initial backoff": {
 			startGas:     1.0,
 			startBackoff: 0,
 			reason:       reasonFailure,
-			gm:           2.0,
+			gasMult:      2.0,
 			wantGas:      1.0, // unchanged
 			wantBackoff:  pol.MinBackoff,
 		},
-		{
-			name:         "failure doubles backoff capped at max",
+		"failure doubles backoff capped at max": {
 			startGas:     1.0,
 			startBackoff: 700 * time.Millisecond,
 			reason:       reasonFailure,
-			gm:           2.0,
+			gasMult:      2.0,
 			wantGas:      1.0,             // unchanged
 			wantBackoff:  1 * time.Second, // 700ms*2=1400ms -> clamp 1s
 		},
-		{
-			name:         "tooBig doubles backoff like failure",
+		"tooBig doubles backoff like failure": {
 			startGas:     1.0,
 			startBackoff: 100 * time.Millisecond,
 			reason:       reasonTooBig,
-			gm:           2.0,
+			gasMult:      2.0,
 			wantGas:      1.0,
 			wantBackoff:  200 * time.Millisecond,
 		},
-		{
-			name:          "sentinel no gas keeps gas unchanged on success",
+		"sentinel no gas keeps gas unchanged on success": {
 			startGas:      5.0,
 			startBackoff:  0,
 			reason:        reasonSuccess,
-			gm:            2.0,
+			gasMult:       2.0,
 			sentinelNoGas: true,
 			wantGas:       5.0,
 			wantBackoff:   pol.MinBackoff,
 		},
+		"undefined reason keeps gas unchanged and uses min backoff": {
+			startGas:     3.0,
+			startBackoff: 500 * time.Millisecond,
+			reason:       reasonUndefined,
+			gasMult:      2.0,
+			wantGas:      3.0,
+			wantBackoff:  0,
+		},
 	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			rs := retryState{Attempt: 0, Backoff: tc.startBackoff, GasPrice: tc.startGas}
+			rs.Next(tc.reason, pol, tc.gasMult, tc.sentinelNoGas)
 
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			rs := RetryState{Attempt: 0, Backoff: tc.startBackoff, GasPrice: tc.startGas}
-			rs.Next(tc.reason, pol, tc.gm, tc.sentinelNoGas)
-
-			if rs.GasPrice != tc.wantGas {
-				t.Fatalf("gas price: got %v, want %v", rs.GasPrice, tc.wantGas)
-			}
-			if rs.Backoff != tc.wantBackoff {
-				t.Fatalf("backoff: got %v, want %v", rs.Backoff, tc.wantBackoff)
-			}
-			if rs.Attempt != 1 {
-				t.Fatalf("attempt: got %d, want %d", rs.Attempt, 1)
-			}
+			assert.Equal(t, tc.wantGas, rs.GasPrice, "gas price")
+			assert.Equal(t, tc.wantBackoff, rs.Backoff, "backoff")
+			assert.Equal(t, 1, rs.Attempt, "attempt")
 		})
 	}
 }

block/internal/submitting/da_submitter.go

@@ -27,11 +27,11 @@ const (
 	defaultGasMultiplier         = 1.0
 	defaultMaxBlobSize           = 2 * 1024 * 1024 // 2MB fallback blob size limit
 	defaultMaxGasPriceClamp      = 1000.0
-	defaultMaxGasMultiplierClamp = 3.0
+	defaultMaxGasMultiplierClamp = 3.0 // must always > 0 to avoid division by zero
 )
 
-// RetryPolicy defines clamped bounds for retries, backoff, and gas pricing.
-type RetryPolicy struct {
+// retryPolicy defines clamped bounds for retries, backoff, and gas pricing.
+type retryPolicy struct {
 	MaxAttempts      int
 	MinBackoff       time.Duration
 	MaxBackoff       time.Duration
@@ -41,8 +41,20 @@ type RetryPolicy struct {
 	MaxGasMultiplier float64
 }
 
-// RetryState holds the current retry attempt, backoff, and gas price.
-type RetryState struct {
+func defaultRetryPolicy(maxAttempts int, maxDuration time.Duration) retryPolicy {
+	return retryPolicy{
+		MaxAttempts:      maxAttempts,
+		MinBackoff:       initialBackoff,
+		MaxBackoff:       maxDuration,
+		MinGasPrice:      defaultGasPrice,
+		MaxGasPrice:      defaultMaxGasPriceClamp,
+		MaxBlobBytes:     defaultMaxBlobSize,
+		MaxGasMultiplier: defaultMaxGasMultiplierClamp,
+	}
+}
+
+// retryState holds the current retry attempt, backoff, and gas price.
+type retryState struct {
 	Attempt  int
 	Backoff  time.Duration
 	GasPrice float64
@@ -51,14 +63,14 @@ type RetryState struct {
 type retryReason int
 
 const (
-	reasonInitial retryReason = iota
+	reasonUndefined retryReason = iota
 	reasonFailure
 	reasonMempool
 	reasonSuccess
 	reasonTooBig
 )
 
-func (rs *RetryState) Next(reason retryReason, pol RetryPolicy, gasMultiplier float64, sentinelNoGas bool) {
+func (rs *retryState) Next(reason retryReason, pol retryPolicy, gasMultiplier float64, sentinelNoGas bool) {
 	switch reason {
 	case reasonSuccess:
 		// Reduce gas price towards initial on success, if multiplier is available
@@ -73,22 +85,25 @@ func (rs *RetryState) Next(reason retryReason, pol RetryPolicy, gasMultiplier fl
 			rs.GasPrice = clamp(rs.GasPrice*m, pol.MinGasPrice, pol.MaxGasPrice)
 		}
 		// Honor mempool stalling by using max backoff window
-		rs.Backoff = clamp(pol.MaxBackoff, pol.MinBackoff, pol.MaxBackoff)
+		rs.Backoff = pol.MaxBackoff
 	case reasonFailure, reasonTooBig:
 		if rs.Backoff == 0 {
 			rs.Backoff = pol.MinBackoff
 		} else {
 			rs.Backoff *= 2
 		}
 		rs.Backoff = clamp(rs.Backoff, pol.MinBackoff, pol.MaxBackoff)
-	case reasonInitial:
+	default:
 		rs.Backoff = 0
 	}
 	rs.Attempt++
 }
 
 // clamp constrains a value between min and max bounds for any comparable type
 func clamp[T ~float64 | time.Duration](v, min, max T) T {
+	if min > max {
+		min, max = max, min
+	}
 	if v < min {
 		return min
 	}
@@ -98,34 +113,6 @@ func clamp[T ~float64 | time.Duration](v, min, max T) T {
 	return v
 }
 
-// getGasMultiplier fetches the gas multiplier from DA layer with fallback and clamping
-func (s *DASubmitter) getGasMultiplier(ctx context.Context, pol RetryPolicy) float64 {
-	gasMultiplier, err := s.da.GasMultiplier(ctx)
-	if err != nil || gasMultiplier <= 0 {
-		if s.config.DA.GasMultiplier > 0 {
-			return clamp(s.config.DA.GasMultiplier, 0.1, pol.MaxGasMultiplier)
-		}
-		s.logger.Warn().Err(err).Msg("failed to get gas multiplier from DA layer, using default 1.0")
-		return defaultGasMultiplier
-	}
-	return clamp(gasMultiplier, 0.1, pol.MaxGasMultiplier)
-}
-
-// initialGasPrice determines the starting gas price with clamping and sentinel handling
-func (s *DASubmitter) initialGasPrice(ctx context.Context, pol RetryPolicy) (price float64, sentinelNoGas bool) {
-	if s.config.DA.GasPrice == noGasPrice {
-		return noGasPrice, true
-	}
-	if s.config.DA.GasPrice > 0 {
-		return clamp(s.config.DA.GasPrice, pol.MinGasPrice, pol.MaxGasPrice), false
-	}
-	if gp, err := s.da.GasPrice(ctx); err == nil {
-		return clamp(gp, pol.MinGasPrice, pol.MaxGasPrice), false
-	}
-	s.logger.Warn().Msg("DA gas price unavailable; using default 0.0")
-	return pol.MinGasPrice, false
-}
-
 // DASubmitter handles DA submission operations
 type DASubmitter struct {
 	da      coreda.DA
@@ -152,6 +139,34 @@ func NewDASubmitter(
 	}
 }
 
+// getGasMultiplier fetches the gas multiplier from DA layer with fallback and clamping
+func (s *DASubmitter) getGasMultiplier(ctx context.Context, pol retryPolicy) float64 {
+	gasMultiplier, err := s.da.GasMultiplier(ctx)
+	if err != nil || gasMultiplier <= 0 {
+		if s.config.DA.GasMultiplier > 0 {
+			return clamp(s.config.DA.GasMultiplier, 0.1, pol.MaxGasMultiplier)
+		}
+		s.logger.Warn().Err(err).Msg("failed to get gas multiplier from DA layer, using default 1.0")
+		return defaultGasMultiplier
+	}
+	return clamp(gasMultiplier, 0.1, pol.MaxGasMultiplier)
+}
+
+// initialGasPrice determines the starting gas price with clamping and sentinel handling
+func (s *DASubmitter) initialGasPrice(ctx context.Context, pol retryPolicy) (price float64, sentinelNoGas bool) {
+	if s.config.DA.GasPrice == noGasPrice {
+		return noGasPrice, true
+	}
+	if s.config.DA.GasPrice > 0 {
+		return clamp(s.config.DA.GasPrice, pol.MinGasPrice, pol.MaxGasPrice), false
+	}
+	if gp, err := s.da.GasPrice(ctx); err == nil {
+		return clamp(gp, pol.MinGasPrice, pol.MaxGasPrice), false
+	}
+	s.logger.Warn().Msg("DA gas price unavailable; using default 0.0")
+	return pol.MinGasPrice, false
+}
+
 // SubmitHeaders submits pending headers to DA layer
 func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager) error {
 	headers, err := cache.GetPendingHeaders(ctx)
@@ -301,25 +316,17 @@ func submitToDA[T any](
 	options []byte,
 	cache cache.Manager,
 ) error {
-	marshaled, err := marshalItems(items, marshalFn, itemType)
+	marshaled, err := marshalItems(ctx, items, marshalFn, itemType)
 	if err != nil {
 		return err
 	}
 
 	// Build retry policy from config with sane defaults
-	pol := RetryPolicy{
-		MaxAttempts:      s.config.DA.MaxSubmitAttempts,
-		MinBackoff:       initialBackoff,
-		MaxBackoff:       s.config.DA.BlockTime.Duration,
-		MinGasPrice:      defaultGasPrice,
-		MaxGasPrice:      defaultMaxGasPriceClamp,
-		MaxBlobBytes:     defaultMaxBlobSize,
-		MaxGasMultiplier: defaultMaxGasMultiplierClamp,
-	}
+	pol := defaultRetryPolicy(s.config.DA.MaxSubmitAttempts, s.config.DA.BlockTime.Duration)
 
 	// Choose initial gas price with clamp
 	gasPrice, sentinelNoGas := s.initialGasPrice(ctx, pol)
-	rs := RetryState{Attempt: 0, Backoff: 0, GasPrice: gasPrice}
+	rs := retryState{Attempt: 0, Backoff: 0, GasPrice: gasPrice}
 	gm := s.getGasMultiplier(ctx, pol)
 
 	// Limit this submission to a single size-capped batch
@@ -421,35 +428,51 @@ func limitBatchBySize[T any](items []T, marshaled [][]byte, maxBytes int) ([]T,
 }
 
 func marshalItems[T any](
+	parentCtx context.Context,
 	items []T,
 	marshalFn func(T) ([]byte, error),
 	itemType string,
 ) ([][]byte, error) {
+	if len(items) == 0 {
+		return nil, nil
+	}
 	marshaled := make([][]byte, len(items))
+	ctx, cancel := context.WithCancel(parentCtx)
+	defer cancel()
+
+	// Semaphore to limit concurrency to 32 workers
+	sem := make(chan struct{}, 32)
 
-	// Use a channel to collect errors from goroutines
-	errCh := make(chan error, len(items))
+	// Use a channel to collect results from goroutines
+	resultCh := make(chan error, len(items))
 
 	// Marshal items concurrently
 	for i, item := range items {
 		go func(idx int, itm T) {
-			bz, err := marshalFn(itm)
-			if err != nil {
-				errCh <- fmt.Errorf("failed to marshal %s item at index %d: %w", itemType, idx, err)
-				return
+			sem <- struct{}{}
+			defer func() { <-sem }()
+
+			select {
+			case <-ctx.Done():
+				resultCh <- ctx.Err()
+			default:
+				bz, err := marshalFn(itm)
+				if err != nil {
+					resultCh <- fmt.Errorf("failed to marshal %s item at index %d: %w", itemType, idx, err)
+					return
+				}
+				marshaled[idx] = bz
+				resultCh <- nil
 			}
-			marshaled[idx] = bz
-			errCh <- nil
 		}(i, item)
 	}
 
 	// Wait for all goroutines to complete and check for errors
 	for i := 0; i < len(items); i++ {
-		if err := <-errCh; err != nil {
+		if err := <-resultCh; err != nil {
 			return nil, err
 		}
 	}
-
 	return marshaled, nil
 }