re-add comments

Author: julienrbrt

block/internal/submitting/da_submitter.go

@@ -199,6 +199,7 @@ func (s *DASubmitter) Close() {
 	s.wg.Wait()
 }
 
+// SubmitHeaders submits pending headers to DA layer
 func (s *DASubmitter) SubmitHeaders(ctx context.Context, headers []*types.SignedHeader, marshalledHeaders [][]byte, cache cache.Manager, signer signer.Signer, onSubmitSuccess func(), onSubmitError func(error)) error {
 	if len(headers) == 0 {
 		return nil
@@ -257,6 +258,7 @@ func (s *DASubmitter) makeHeaderPostSubmit(ctx context.Context, cache cache.Mana
 	}
 }
 
+// SubmitData submits pending data to DA layer
 func (s *DASubmitter) SubmitData(ctx context.Context, unsignedDataList []*types.SignedData, marshalledData [][]byte, cache cache.Manager, signer signer.Signer, genesis genesis.Genesis, onSubmitSuccess func(), onSubmitError func(error)) error {
 	if len(unsignedDataList) == 0 {
 		return nil
@@ -459,6 +461,7 @@ func limitBatchBySizeBytes(marshaled [][]byte, maxBytes uint64) ([][]byte, bool)
 	return marshaled[:count], false
 }
 
+// recordFailure records a DA submission failure in metrics
 func (s *DASubmitter) recordFailure(reason common.DASubmitterFailureReason) {
 	counter, ok := s.metrics.DASubmitterFailures[reason]
 	if !ok {
@@ -472,10 +475,13 @@ func (s *DASubmitter) recordFailure(reason common.DASubmitterFailureReason) {
 	}
 }
 
+// createDAEnvelopes creates signed DA envelopes for the given headers.
+// It uses caching to avoid re-signing on retries and parallel signing for new envelopes.
 func (s *DASubmitter) createDAEnvelopes(ctx context.Context, headers []*types.SignedHeader, marshalledHeaders [][]byte, signer signer.Signer) ([][]byte, error) {
 	envelopes := make([][]byte, len(headers))
 
-	var needSigning []int
+	// First pass: check cache for already-signed envelopes
+	var needSigning []int // indices that need signing
 	for i, header := range headers {
 		height := header.Height()
 		if cached := s.getCachedEnvelope(height); cached != nil {
@@ -485,6 +491,7 @@ func (s *DASubmitter) createDAEnvelopes(ctx context.Context, headers []*types.Si
 		}
 	}
 
+	// If all envelopes were cached, we're done
 	if len(needSigning) == 0 {
 		s.logger.Debug().Int("cached", len(headers)).Msg("all envelopes retrieved from cache")
 		return envelopes, nil
@@ -495,7 +502,9 @@ func (s *DASubmitter) createDAEnvelopes(ctx context.Context, headers []*types.Si
 		Int("to_sign", len(needSigning)).
 		Msg("signing DA envelopes")
 
+	// For small batches, sign sequentially to avoid goroutine overhead
 	if len(needSigning) <= 2 || s.signingWorkers <= 1 {
+		// Send jobs
 		for _, i := range needSigning {
 			envelope, err := s.signAndCacheEnvelope(ctx, headers[i], marshalledHeaders[i], signer)
 			if err != nil {
@@ -506,9 +515,11 @@ func (s *DASubmitter) createDAEnvelopes(ctx context.Context, headers []*types.Si
 		return envelopes, nil
 	}
 
+	// Parallel signing for larger batches
 	return s.signEnvelopesParallel(ctx, headers, marshalledHeaders, envelopes, needSigning, signer)
 }
 
+// signEnvelopesParallel signs envelopes in parallel using a worker pool.
 func (s *DASubmitter) signEnvelopesParallel(
 	ctx context.Context,
 	headers []*types.SignedHeader,
@@ -529,6 +540,7 @@ func (s *DASubmitter) signEnvelopesParallel(
 	jobs := make(chan signJob, len(needSigning))
 	results := make(chan signResult, len(needSigning))
 
+	// Start workers
 	numWorkers := min(s.signingWorkers, len(needSigning))
 	var wg sync.WaitGroup
 	for range numWorkers {
@@ -545,11 +557,13 @@ func (s *DASubmitter) signEnvelopesParallel(
 	}
 	close(jobs)
 
+	// Wait for workers to finish and close results
 	go func() {
 		wg.Wait()
 		close(results)
 	}()
 
+	// Collect results
 	var firstErr error
 	for result := range results {
 		if result.err != nil && firstErr == nil {
@@ -568,26 +582,33 @@ func (s *DASubmitter) signEnvelopesParallel(
 	return envelopes, nil
 }
 
+// signAndCacheEnvelope signs a single header and caches the result.
 func (s *DASubmitter) signAndCacheEnvelope(ctx context.Context, header *types.SignedHeader, marshalledHeader []byte, signer signer.Signer) ([]byte, error) {
+	// Sign the pre-marshalled header content
 	envelopeSignature, err := signer.Sign(ctx, marshalledHeader)
 	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 nil, fmt.Errorf("failed to marshal DA envelope: %w", err)
 	}
 
+	// Cache for potential retries
 	s.setCachedEnvelope(header.Height(), envelope)
 
 	return envelope, nil
 }
 
+// getCachedEnvelope retrieves a cached envelope for the given height.
+// Uses lazy invalidation: entries at or below lastSubmittedHeight are considered invalid.
 func (s *DASubmitter) getCachedEnvelope(height uint64) []byte {
 	if s.envelopeCache == nil {
 		return nil
 	}
+	// Lazy invalidation: don't return cached data for already-submitted heights
 	if height <= s.lastSubmittedHeight.Load() {
 		return nil
 	}
@@ -597,16 +618,20 @@ func (s *DASubmitter) getCachedEnvelope(height uint64) []byte {
 	return nil
 }
 
+// setCachedEnvelope stores an envelope in the cache.
+// Does not cache heights that have already been submitted.
 func (s *DASubmitter) setCachedEnvelope(height uint64, envelope []byte) {
 	if s.envelopeCache == nil {
 		return
 	}
+	// Don't cache already-submitted heights
 	if height <= s.lastSubmittedHeight.Load() {
 		return
 	}
 	s.envelopeCache.Add(height, envelope)
 }
 
+// signData signs unsigned SignedData structs returned from cache
 func (s *DASubmitter) signData(ctx context.Context, unsignedDataList []*types.SignedData, unsignedDataListBz [][]byte, signer signer.Signer, genesis genesis.Genesis) ([]*types.SignedData, [][]byte, error) {
 	if signer == nil {
 		return nil, nil, fmt.Errorf("signer is nil")
@@ -635,6 +660,7 @@ func (s *DASubmitter) signData(ctx context.Context, unsignedDataList []*types.Si
 	signedDataListBz := make([][]byte, 0, len(unsignedDataListBz))
 
 	for i, unsignedData := range unsignedDataList {
+		// Skip empty data
 		if len(unsignedData.Txs) == 0 {
 			continue
 		}
@@ -663,6 +689,10 @@ func (s *DASubmitter) signData(ctx context.Context, unsignedDataList []*types.Si
 	return signedDataList, signedDataListBz, nil
 }
 
+// mergeSubmitOptions merges the base submit options with a signing address.
+// If the base options are valid JSON, the signing address is added to the JSON object.
+// Otherwise, a new JSON object is created with just the signing address.
+// Returns the base options unchanged if no signing address is provided.
 func mergeSubmitOptions(baseOptions []byte, signingAddress string) ([]byte, error) {
 	if signingAddress == "" {
 		return baseOptions, nil