remove double marshalling

Author: julienrbrt

block/internal/submitting/da_submitter.go

@@ -160,7 +160,7 @@ func (s *DASubmitter) recordFailure(reason common.DASubmitterFailureReason) {
 }
 
 // SubmitHeaders submits pending headers to DA layer
-func (s *DASubmitter) SubmitHeaders(ctx context.Context, headers []*types.SignedHeader, cache cache.Manager, signer signer.Signer) error {
+func (s *DASubmitter) SubmitHeaders(ctx context.Context, headers []*types.SignedHeader, marshalledHeaders [][]byte, cache cache.Manager, signer signer.Signer) error {
 	if len(headers) == 0 {
 		return nil
 	}
@@ -169,26 +169,30 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, headers []*types.Signed
 		return fmt.Errorf("signer is nil")
 	}
 
+	if len(marshalledHeaders) != len(headers) {
+		return fmt.Errorf("marshalledHeaders length (%d) does not match headers length (%d)", len(marshalledHeaders), len(headers))
+	}
+
 	s.logger.Info().Int("count", len(headers)).Msg("submitting headers to DA")
 
-	return submitToDA(s, ctx, headers,
-		func(header *types.SignedHeader) ([]byte, error) {
-			// A. Marshal the inner SignedHeader content to bytes (canonical representation for signing)
-			// This effectively signs "Fields 1-3" of the intended DAHeaderEnvelope.
-			contentBytes, err := header.MarshalBinary()
-			if err != nil {
-				return nil, fmt.Errorf("failed to marshal signed header for envelope signing: %w", err)
-			}
+	// Create DA envelopes from pre-marshalled headers
+	envelopes := make([][]byte, len(headers))
+	for i, header := range headers {
+		// Sign the pre-marshalled header content
+		envelopeSignature, err := signer.Sign(marshalledHeaders[i])
+		if err != nil {
+			return fmt.Errorf("failed to sign envelope for header %d: %w", i, err)
+		}
 
-			// B. Sign the contentBytes with the envelope signer (aggregator)
-			envelopeSignature, err := signer.Sign(contentBytes)
-			if err != nil {
-				return nil, fmt.Errorf("failed to sign envelope: %w", err)
-			}
+		// Create the envelope and marshal it
+		envelope, err := header.MarshalDAEnvelope(envelopeSignature)
+		if err != nil {
+			return fmt.Errorf("failed to marshal DA envelope for header %d: %w", i, err)
+		}
+		envelopes[i] = envelope
+	}
 
-			// C. Create the envelope and marshal it
-			return header.MarshalDAEnvelope(envelopeSignature)
-		},
+	return submitToDA(s, ctx, headers, envelopes,
 		func(submitted []*types.SignedHeader, res *datypes.ResultSubmit) {
 			for _, header := range submitted {
 				cache.SetHeaderDAIncluded(header.Hash().String(), res.Height, header.Height())
@@ -206,11 +210,15 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, headers []*types.Signed
 }
 
 // SubmitData submits pending data to DA layer
-func (s *DASubmitter) SubmitData(ctx context.Context, unsignedDataList []*types.SignedData, cache cache.Manager, signer signer.Signer, genesis genesis.Genesis) error {
+func (s *DASubmitter) SubmitData(ctx context.Context, unsignedDataList []*types.SignedData, marshalledData [][]byte, cache cache.Manager, signer signer.Signer, genesis genesis.Genesis) error {
 	if len(unsignedDataList) == 0 {
 		return nil
 	}
 
+	if len(marshalledData) != len(unsignedDataList) {
+		return fmt.Errorf("marshalledData length (%d) does not match unsignedDataList length (%d)", len(marshalledData), len(unsignedDataList))
+	}
+
 	// Sign the data (cache returns unsigned SignedData structs)
 	signedDataList, err := s.signData(unsignedDataList, signer, genesis)
 	if err != nil {
@@ -223,10 +231,17 @@ func (s *DASubmitter) SubmitData(ctx context.Context, unsignedDataList []*types.
 
 	s.logger.Info().Int("count", len(signedDataList)).Msg("submitting data to DA")
 
-	return submitToDA(s, ctx, signedDataList,
-		func(signedData *types.SignedData) ([]byte, error) {
-			return signedData.MarshalBinary()
-		},
+	// Filter marshalledData to match signedDataList (removes empty data)
+	filteredMarshalledData := make([][]byte, 0, len(signedDataList))
+	signedIdx := 0
+	for i, unsigned := range unsignedDataList {
+		if signedIdx < len(signedDataList) && unsigned.Height() == signedDataList[signedIdx].Height() {
+			filteredMarshalledData = append(filteredMarshalledData, marshalledData[i])
+			signedIdx++
+		}
+	}
+
+	return submitToDA(s, ctx, signedDataList, filteredMarshalledData,
 		func(submitted []*types.SignedData, res *datypes.ResultSubmit) {
 			for _, sd := range submitted {
 				cache.SetDataDAIncluded(sd.Data.DACommitment().String(), res.Height, sd.Height())
@@ -342,16 +357,15 @@ func submitToDA[T any](
 	s *DASubmitter,
 	ctx context.Context,
 	items []T,
-	marshalFn func(T) ([]byte, error),
+	marshaled [][]byte,
 	postSubmit func([]T, *datypes.ResultSubmit),
 	itemType string,
 	namespace []byte,
 	options []byte,
 	getTotalPendingFn func() uint64,
 ) error {
-	marshaled, err := marshalItems(ctx, items, marshalFn, itemType)
-	if err != nil {
-		return err
+	if len(items) != len(marshaled) {
+		return fmt.Errorf("items length (%d) does not match marshaled length (%d)", len(items), len(marshaled))
 	}
 
 	pol := defaultRetryPolicy(s.config.DA.MaxSubmitAttempts, s.config.DA.BlockTime.Duration)
@@ -513,55 +527,6 @@ func limitBatchBySize[T any](items []T, marshaled [][]byte, maxBytes int) ([]T,
 	return items[:count], marshaled[:count], nil
 }
 
-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 results from goroutines
-	resultCh := make(chan error, len(items))
-
-	// Marshal items concurrently
-	for i, item := range items {
-		go func(idx int, itm T) {
-			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
-			}
-		}(i, item)
-	}
-
-	// Wait for all goroutines to complete and check for errors
-	for i := 0; i < len(items); i++ {
-		if err := <-resultCh; err != nil {
-			return nil, err
-		}
-	}
-	return marshaled, nil
-}
-
 func waitForBackoffOrContext(ctx context.Context, backoff time.Duration) error {
 	if backoff <= 0 {
 		select {

block/internal/submitting/da_submitter_integration_test.go

@@ -101,11 +101,27 @@ func TestDASubmitter_SubmitHeadersAndData_MarksInclusionAndUpdatesLastSubmitted(
 	// Submit headers and data
 	headers, err := cm.GetPendingHeaders(context.Background())
 	require.NoError(t, err)
-	require.NoError(t, daSubmitter.SubmitHeaders(context.Background(), headers, cm, n))
+
+	// Marshal headers
+	marshalledHeaders := make([][]byte, len(headers))
+	for i, h := range headers {
+		data, err := h.MarshalBinary()
+		require.NoError(t, err)
+		marshalledHeaders[i] = data
+	}
+	require.NoError(t, daSubmitter.SubmitHeaders(context.Background(), headers, marshalledHeaders, cm, n))
 
 	dataList, err := cm.GetPendingData(context.Background())
 	require.NoError(t, err)
-	require.NoError(t, daSubmitter.SubmitData(context.Background(), dataList, cm, n, gen))
+
+	// Marshal data
+	marshalledData := make([][]byte, len(dataList))
+	for i, d := range dataList {
+		data, err := d.MarshalBinary()
+		require.NoError(t, err)
+		marshalledData[i] = data
+	}
+	require.NoError(t, daSubmitter.SubmitData(context.Background(), dataList, marshalledData, cm, n, gen))
 
 	// After submission, inclusion markers should be set
 	_, ok := cm.GetHeaderDAIncluded(hdr1.Hash().String())

block/internal/submitting/da_submitter_mocks_test.go

@@ -38,9 +38,6 @@ func newTestSubmitter(t *testing.T, mockClient *mocks.MockClient, override func(
 	return NewDASubmitter(mockClient, cfg, genesis.Genesis{} /*options=*/, common.BlockOptions{}, common.NopMetrics(), zerolog.Nop())
 }
 
-// marshal helper for simple items
-func marshalString(s string) ([]byte, error) { return []byte(s), nil }
-
 func TestSubmitToDA_MempoolRetry_IncreasesGasAndSucceeds(t *testing.T) {
 	t.Parallel()
 
@@ -68,12 +65,17 @@ func TestSubmitToDA_MempoolRetry_IncreasesGasAndSucceeds(t *testing.T) {
 	s := newTestSubmitter(t, client, nil)
 
 	items := []string{"a", "b", "c"}
+	marshalledItems := make([][]byte, 0, len(items))
+	for idx, item := range items {
+		marshalledItems[idx] = []byte(item)
+	}
+
 	ctx := context.Background()
 	err := submitToDA[string](
 		s,
 		ctx,
 		items,
-		marshalString,
+		marshalledItems,
 		func(_ []string, _ *datypes.ResultSubmit) {},
 		"item",
 		nsBz,
@@ -112,12 +114,17 @@ func TestSubmitToDA_UnknownError_RetriesSameGasThenSucceeds(t *testing.T) {
 	s := newTestSubmitter(t, client, nil)
 
 	items := []string{"x"}
+	marshalledItems := make([][]byte, 0, len(items))
+	for idx, item := range items {
+		marshalledItems[idx] = []byte(item)
+	}
+
 	ctx := context.Background()
 	err := submitToDA[string](
 		s,
 		ctx,
 		items,
-		marshalString,
+		marshalledItems,
 		func(_ []string, _ *datypes.ResultSubmit) {},
 		"item",
 		nsBz,
@@ -158,12 +165,17 @@ func TestSubmitToDA_TooBig_HalvesBatch(t *testing.T) {
 	s := newTestSubmitter(t, client, nil)
 
 	items := []string{"a", "b", "c", "d"}
+	marshalledItems := make([][]byte, 0, len(items))
+	for idx, item := range items {
+		marshalledItems[idx] = []byte(item)
+	}
+
 	ctx := context.Background()
 	err := submitToDA[string](
 		s,
 		ctx,
 		items,
-		marshalString,
+		marshalledItems,
 		func(_ []string, _ *datypes.ResultSubmit) {},
 		"item",
 		nsBz,
@@ -198,12 +210,17 @@ func TestSubmitToDA_SentinelNoGas_PreservesGasAcrossRetries(t *testing.T) {
 	s := newTestSubmitter(t, client, nil)
 
 	items := []string{"only"}
+	marshalledItems := make([][]byte, 0, len(items))
+	for idx, item := range items {
+		marshalledItems[idx] = []byte(item)
+	}
+
 	ctx := context.Background()
 	err := submitToDA[string](
 		s,
 		ctx,
 		items,
-		marshalString,
+		marshalledItems,
 		func(_ []string, _ *datypes.ResultSubmit) {},
 		"item",
 		nsBz,
@@ -237,12 +254,17 @@ func TestSubmitToDA_PartialSuccess_AdvancesWindow(t *testing.T) {
 	s := newTestSubmitter(t, client, nil)
 
 	items := []string{"a", "b", "c"}
+	marshalledItems := make([][]byte, 0, len(items))
+	for idx, item := range items {
+		marshalledItems[idx] = []byte(item)
+	}
+
 	ctx := context.Background()
 	err := submitToDA[string](
 		s,
 		ctx,
 		items,
-		marshalString,
+		marshalledItems,
 		func(submitted []string, _ *datypes.ResultSubmit) { totalSubmitted += len(submitted) },
 		"item",
 		nsBz,

block/internal/submitting/da_submitter_test.go

@@ -215,7 +215,14 @@ func TestDASubmitter_SubmitHeaders_Success(t *testing.T) {
 	// Get headers from cache and submit
 	headers, err := cm.GetPendingHeaders(ctx)
 	require.NoError(t, err)
-	err = submitter.SubmitHeaders(ctx, headers, cm, signer)
+	// Marshal headers
+	marshalledHeaders := make([][]byte, len(headers))
+	for i, h := range headers {
+		data, err := h.MarshalBinary()
+		require.NoError(t, err)
+		marshalledHeaders[i] = data
+	}
+	err = submitter.SubmitHeaders(ctx, headers, marshalledHeaders, cm, signer)
 	require.NoError(t, err)
 
 	// Verify headers are marked as DA included
@@ -237,7 +244,8 @@ func TestDASubmitter_SubmitHeaders_NoPendingHeaders(t *testing.T) {
 	// Get headers from cache (should be empty) and submit
 	headers, err := cm.GetPendingHeaders(ctx)
 	require.NoError(t, err)
-	err = submitter.SubmitHeaders(ctx, headers, cm, signer)
+	var marshalledHeaders [][]byte
+	err = submitter.SubmitHeaders(ctx, headers, marshalledHeaders, cm, signer)
 	require.NoError(t, err) // Should succeed with no action
 	mockDA.AssertNotCalled(t, "Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything)
 }
@@ -330,9 +338,16 @@ func TestDASubmitter_SubmitData_Success(t *testing.T) {
 	require.NoError(t, batch2.Commit())
 
 	// Get data from cache and submit
-	dataList, err := cm.GetPendingData(ctx)
+	signedDataList, err := cm.GetPendingData(ctx)
 	require.NoError(t, err)
-	err = submitter.SubmitData(ctx, dataList, cm, signer, gen)
+	// Marshal data
+	marshalledData := make([][]byte, len(signedDataList))
+	for i, d := range signedDataList {
+		data, err := d.MarshalBinary()
+		require.NoError(t, err)
+		marshalledData[i] = data
+	}
+	err = submitter.SubmitData(ctx, signedDataList, marshalledData, cm, signer, gen)
 	require.NoError(t, err)
 
 	// Verify data is marked as DA included
@@ -383,9 +398,17 @@ func TestDASubmitter_SubmitData_SkipsEmptyData(t *testing.T) {
 	require.NoError(t, batch.Commit())
 
 	// Get data from cache and submit - should succeed but skip empty data
-	dataList, err := cm.GetPendingData(ctx)
+	// Get data from cache and submit
+	signedDataList, err := cm.GetPendingData(ctx)
 	require.NoError(t, err)
-	err = submitter.SubmitData(ctx, dataList, cm, signer, gen)
+	// Marshal data
+	marshalledData := make([][]byte, len(signedDataList))
+	for i, d := range signedDataList {
+		data, err := d.MarshalBinary()
+		require.NoError(t, err)
+		marshalledData[i] = data
+	}
+	err = submitter.SubmitData(ctx, signedDataList, marshalledData, cm, signer, gen)
 	require.NoError(t, err)
 	mockDA.AssertNotCalled(t, "Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything)
 
@@ -404,7 +427,8 @@ func TestDASubmitter_SubmitData_NoPendingData(t *testing.T) {
 	// Get data from cache (should be empty) and submit
 	dataList, err := cm.GetPendingData(ctx)
 	require.NoError(t, err)
-	err = submitter.SubmitData(ctx, dataList, cm, signer, gen)
+	var marshalledData [][]byte
+	err = submitter.SubmitData(ctx, dataList, marshalledData, cm, signer, gen)
 	require.NoError(t, err) // Should succeed with no action
 	mockDA.AssertNotCalled(t, "Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything)
 }
@@ -443,9 +467,16 @@ func TestDASubmitter_SubmitData_NilSigner(t *testing.T) {
 	require.NoError(t, batch.Commit())
 
 	// Get data from cache and submit with nil signer - should fail
-	dataList, err := cm.GetPendingData(ctx)
+	signedDataList, err := cm.GetPendingData(ctx)
 	require.NoError(t, err)
-	err = submitter.SubmitData(ctx, dataList, cm, nil, gen)
+	// Marshal data
+	marshalledData := make([][]byte, len(signedDataList))
+	for i, d := range signedDataList {
+		data, err := d.MarshalBinary()
+		require.NoError(t, err)
+		marshalledData[i] = data
+	}
+	err = submitter.SubmitData(ctx, signedDataList, marshalledData, cm, nil, gen)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "signer is nil")
 	mockDA.AssertNotCalled(t, "Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything)