perf(fiber-da): per-item concurrent Uploads on Submit

Fan out one goroutine per item in fiber DA Submit, calling fiber.Upload
concurrently with the caller's ctx. Settlement throughput now scales
linearly with the batch size: previously ev-node's submitter could
only have one Upload in flight per stream (header + data, mutex-locked
in submitter.go), and each Submit further serialized the batch into
one big flatten-encoded blob. With fan-out, a Submit of N items
becomes N concurrent Uploads, and Fibre's ~1.5 s per-Upload latency
amortizes across N.

The result-aggregation honors submitToDA's "prefix of successes"
contract: SubmittedCount = N means items [0..N) succeeded and the
caller will retry [N..end). Reporting interleaved successes would
double-submit blobs and waste escrow; matching prefix semantics keeps
the retry contract intact even when individual Uploads fail
out-of-order.

Pair changes in submitting/da_submitter.go:
- limitBatchBySize gains a maxItems cap (was total-bytes-only). Each
  item is still bounded by maxItemBytes (chain ceiling), but the
  total batch is now bounded by item count, letting multiple
  full-size items flow through one Submit.
- retryPolicy adds MaxItems with a sensible non-fiber default of 1
  (preserves legacy single-item-per-Submit semantics for backends
  that flatten a batch into one blob).
- For the fiber backend, MaxItems is bumped to 16 — covers a 5 min
  run at 1 b/s production with 4–8 pending blocks while leaving
  headroom for memory pressure under MaxBlobSize-sized items.

Wire-format follow-up (see TODO in fiber_client.go::Submit): the
retrieve path in this file still uses splitBlobs which assumes the
old single-prefixed-blob format. Per-item Uploads now produce raw
blobs with their own BlobIDs; retrieve needs an update to read each
BlobID separately. The bench's aggregator-only setup never invokes
retrieve so this is unblocked for measurement but blocks merging to
production until addressed.

Author: Wondertan

block/internal/da/fiber_client.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog"
@@ -87,62 +88,96 @@ func (c *fiberDAClient) Submit(ctx context.Context, data [][]byte, _ float64, na
 		}
 	}
 
-	// Single-item fast path: avoid the MaxBlobSize-sized allocation +
-	// memcpy that flattenBlobs would do just to wrap one item in the
-	// 8-byte count/length prefix. With per-item caps already saturating
-	// MaxBlobBytes for data blobs, this is the steady-state path.
+	// Per-item concurrent Upload. Fibre's per-Upload latency is
+	// dominated by validator signature aggregation (~1.5 s on a
+	// healthy network) and does not scale up linearly under multiple
+	// in-flight Uploads, so settlement throughput scales with the
+	// number of concurrent items submitted in a single batch. Each
+	// item gets its own goroutine, its own Upload call, and its own
+	// BlobID in the result; the previous flatten step was both
+	// memory-wasteful (a MaxBlobSize-sized memcpy on every Submit)
+	// and inherently serial (one Upload per Submit).
 	//
-	// TODO: wire-format compat — splitBlobs always expects the prefix,
-	// so any retriever (full node syncer, light client) downloading a
-	// blob written via this fast path will fail to decode. Address
-	// alongside the concurrent-uploads change by switching to a
-	// per-item Upload model where flatten is no longer needed.
-	var blob []byte
-	if len(data) == 1 {
-		blob = data[0]
-	} else {
-		blob = flattenBlobs(data)
+	// TODO: wire-format compat — old splitBlobs assumed all items in
+	// a Submit were written as a single prefixed blob. With per-item
+	// Uploads, retrievers must treat each BlobID separately. The
+	// retrieve path in this file still uses splitBlobs and will need
+	// a follow-up to read the new per-item blobs as raw payloads.
+	nsID := namespace[len(namespace)-10:]
+	type uploadResult struct {
+		idx int
+		id  []byte
+		err error
+	}
+	results := make([]uploadResult, len(data))
+	var wg sync.WaitGroup
+	for i := range data {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			res, err := c.fiber.Upload(ctx, nsID, data[i])
+			if err != nil {
+				results[i] = uploadResult{idx: i, err: err}
+				return
+			}
+			id := make([]byte, len(res.BlobID))
+			copy(id, res.BlobID)
+			results[i] = uploadResult{idx: i, id: id}
+		}(i)
+	}
+	wg.Wait()
+
+	// Walk results in submission order. submitToDA's retry logic
+	// expects "prefix of successes": SubmittedCount=N means items
+	// [0..N) succeeded and the caller will re-submit items [N..end)
+	// on the next attempt. Reporting interleaved successes would
+	// double-submit blobs and waste escrow; matching prefix
+	// semantics keeps the contract intact even when individual
+	// Uploads fail out-of-order.
+	ids := make([][]byte, 0, len(data))
+	var firstErr error
+	for _, r := range results {
+		if r.err != nil {
+			firstErr = r.err
+			break
+		}
+		ids = append(ids, r.id)
 	}
 
-	// Honor the caller's context so Upload returns promptly on
-	// shutdown / parent cancellation. The previous context.Background()
-	// kept Uploads alive past node shutdown and contributed to the
-	// "payment promise already processed" warnings we saw in early
-	// runs (a stale Upload would settle after the node had stopped
-	// tracking it).
-	result, err := c.fiber.Upload(ctx, namespace[len(namespace)-10:], blob)
-	if err != nil {
+	if len(ids) == 0 && firstErr != nil {
 		code := datypes.StatusError
 		switch {
-		case errors.Is(err, context.Canceled):
+		case errors.Is(firstErr, context.Canceled):
 			code = datypes.StatusContextCanceled
-		case errors.Is(err, context.DeadlineExceeded):
+		case errors.Is(firstErr, context.DeadlineExceeded):
 			code = datypes.StatusContextDeadline
 		}
-
-		c.logger.Error().Err(err).Msg("fiber upload failed")
-
+		c.logger.Error().Err(firstErr).Msg("fiber upload failed")
 		return datypes.ResultSubmit{
 			BaseResult: datypes.BaseResult{
-				Code: code,
-				Message: fmt.Sprintf("fiber upload failed for blob: %v", err),
-				// On error nothing settled — the previous len(data)-1
-				// reported all-but-one as submitted on full failure,
-				// which lied to the caller's retry/postSubmit logic.
+				Code:           code,
+				Message:        fmt.Sprintf("fiber upload failed for blob: %v", firstErr),
 				SubmittedCount: 0,
 				BlobSize:       blobSize,
 				Timestamp:      time.Now(),
 			},
 		}
 	}
 
-	c.logger.Debug().Int("num_ids", len(data)).Uint64("height", 0 /* TODO */).Msg("fiber DA submission successful")
+	if firstErr != nil {
+		c.logger.Warn().Err(firstErr).
+			Int("submitted", len(ids)).
+			Int("total", len(data)).
+			Msg("fiber upload partial success — caller will retry the remainder")
+	}
+
+	c.logger.Debug().Int("num_ids", len(ids)).Uint64("height", 0 /* TODO */).Msg("fiber DA submission successful")
 
 	return datypes.ResultSubmit{
 		BaseResult: datypes.BaseResult{
 			Code:           datypes.StatusSuccess,
-			IDs:            [][]byte{result.BlobID},
-			SubmittedCount: uint64(len(data)),
+			IDs:            ids,
+			SubmittedCount: uint64(len(ids)),
 			Height:         0, /* TODO */
 			BlobSize:       blobSize,
 			Timestamp:      time.Now(),

block/internal/submitting/da_submitter.go

@@ -43,8 +43,28 @@ type retryPolicy struct {
 	MinBackoff   time.Duration
 	MaxBackoff   time.Duration
 	MaxBlobBytes uint64
+	// MaxItems caps the number of items packed into a single Submit
+	// call. DA clients that fan out per-item Uploads (fiber) benefit
+	// linearly from larger batches — settlement throughput scales
+	// with concurrency until per-Upload latency dominates. Default
+	// 1 preserves legacy single-item-per-Submit semantics for
+	// backends that flatten a batch into one blob (JSON-RPC blob
+	// client). The fiber path overrides this from config.
+	MaxItems int
 }
 
+// defaultBatchItems is the conservative default for non-fiber backends
+// that historically expected one item per Submit call. The fiber path
+// raises this via config because it can fan out per-item Uploads.
+const defaultBatchItems = 1
+
+// fiberDefaultBatchItems is the upper bound on items packed into a
+// single fiber Submit. Each item gets its own concurrent Upload, so
+// this caps the per-batch goroutine fan-out. 16 covers a 5 min run at
+// 1 b/s production with 4–8 pending blocks while leaving headroom for
+// memory pressure; tunable via config when the cleanup TODO lands.
+const fiberDefaultBatchItems = 16
+
 func defaultRetryPolicy(maxAttempts int, maxDuration time.Duration) retryPolicy {
 	return retryPolicy{
 		MaxAttempts: maxAttempts,
@@ -57,6 +77,7 @@ func defaultRetryPolicy(maxAttempts int, maxDuration time.Duration) retryPolicy
 		// the duplication is what made packed-block-larger-than-cap
 		// failures non-obvious. See common/consts.go.
 		MaxBlobBytes: common.DefaultMaxBlobSize,
+		MaxItems:     defaultBatchItems,
 	}
 }
 
@@ -578,12 +599,19 @@ func submitToDA[T any](
 	}
 
 	pol := defaultRetryPolicy(s.config.DA.MaxSubmitAttempts, s.config.DA.BlockTime.Duration)
+	// Fiber's DA client fans out per-item Uploads concurrently, so
+	// packing more items per Submit lifts settlement throughput. For
+	// non-fiber backends the default of 1 preserves the legacy
+	// flatten-one-blob behavior.
+	if s.config.DA.IsFiberEnabled() {
+		pol.MaxItems = fiberDefaultBatchItems
+	}
 
 	rs := retryState{Attempt: 0, Backoff: 0}
 
 	// Limit this submission to a single size-capped batch
 	if len(marshaled) > 0 {
-		batchItems, batchMarshaled, err := limitBatchBySize(items, marshaled, pol.MaxBlobBytes)
+		batchItems, batchMarshaled, err := limitBatchBySize(items, marshaled, pol.MaxBlobBytes, pol.MaxItems)
 		if err != nil {
 			s.logger.Error().
 				Str("itemType", itemType).
@@ -694,27 +722,42 @@ func submitToDA[T any](
 	return fmt.Errorf("failed to submit all %s(s) to DA layer after %d attempts", itemType, rs.Attempt)
 }
 
-// limitBatchBySize returns a prefix of items whose total marshaled size does not exceed maxBytes.
-// If the first item exceeds maxBytes, it returns ErrOversizedItem which is unrecoverable.
-func limitBatchBySize[T any](items []T, marshaled [][]byte, maxBytes uint64) ([]T, [][]byte, error) {
-	total := uint64(0)
+// limitBatchBySize returns a prefix of items whose per-item marshaled size
+// fits within maxItemBytes. The total batch size is bounded by item count
+// (maxItems), not by total bytes — DA clients that can fan out per-item
+// Uploads (e.g. the fiber DA client) settle each item in its own
+// concurrent Upload call, so packing more items per batch lifts the
+// effective settlement throughput. DA clients that flatten a batch into
+// a single blob still get one item per call when maxItems == 1.
+//
+// If the first item exceeds maxItemBytes, returns ErrOversizedItem
+// (unrecoverable). If no items fit at all (empty inputs), returns a
+// distinct error so the caller can distinguish "nothing to send".
+//
+// TODO(throughput-cleanup): see common/consts.go — maxItemBytes is the
+// per-item chain ceiling, separate from the raw-tx budget driving
+// FilterTxs. Once that split lands, the duplicate-cap-everywhere
+// problem these fixes work around goes away.
+func limitBatchBySize[T any](items []T, marshaled [][]byte, maxItemBytes uint64, maxItems int) ([]T, [][]byte, error) {
+	if maxItems <= 0 {
+		maxItems = 1
+	}
 	count := 0
 	for i := range items {
+		if count >= maxItems {
+			break
+		}
 		sz := uint64(len(marshaled[i]))
-		if sz > maxBytes {
+		if sz > maxItemBytes {
 			if i == 0 {
-				return nil, nil, fmt.Errorf("%w: item size %d exceeds max %d", common.ErrOversizedItem, sz, maxBytes)
+				return nil, nil, fmt.Errorf("%w: item size %d exceeds max %d", common.ErrOversizedItem, sz, maxItemBytes)
 			}
 			break
 		}
-		if total+sz > maxBytes {
-			break
-		}
-		total += sz
 		count++
 	}
 	if count == 0 {
-		return nil, nil, fmt.Errorf("no items fit within %d bytes", maxBytes)
+		return nil, nil, fmt.Errorf("no items fit within %d bytes", maxItemBytes)
 	}
 	return items[:count], marshaled[:count], nil
 }