Fix DA submission tracking and shutdown ordering

Author: julienrbrt

block/internal/submitting/da_submitter.go

@@ -223,11 +223,14 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, headers []*types.Signed
 
 	postSubmit := s.makeHeaderPostSubmit(ctx, cache)
 	namespace := s.client.GetHeaderNamespace()
+	submittedOffset := 0
 
 	s.wg.Go(func() {
 		s.submitWithRetry(ctx, envelopes, namespace, func(submittedCount int, daHeight uint64) {
 			if submittedCount > 0 {
-				postSubmit(headers[:submittedCount], &datypes.ResultSubmit{BaseResult: datypes.BaseResult{Code: datypes.StatusSuccess, SubmittedCount: uint64(submittedCount), Height: daHeight}})
+				end := submittedOffset + submittedCount
+				postSubmit(headers[submittedOffset:end], &datypes.ResultSubmit{BaseResult: datypes.BaseResult{Code: datypes.StatusSuccess, SubmittedCount: uint64(submittedCount), Height: daHeight}})
+				submittedOffset = end
 			}
 			if onSubmitSuccess != nil {
 				onSubmitSuccess()
@@ -280,11 +283,14 @@ func (s *DASubmitter) SubmitData(ctx context.Context, unsignedDataList []*types.
 
 	postSubmit := s.makeDataPostSubmit(ctx, cache)
 	namespace := s.client.GetDataNamespace()
+	submittedOffset := 0
 
 	s.wg.Go(func() {
 		s.submitWithRetry(ctx, signedDataListBz, namespace, func(submittedCount int, daHeight uint64) {
 			if submittedCount > 0 {
-				postSubmit(signedDataList[:submittedCount], &datypes.ResultSubmit{BaseResult: datypes.BaseResult{Code: datypes.StatusSuccess, SubmittedCount: uint64(submittedCount), Height: daHeight}})
+				end := submittedOffset + submittedCount
+				postSubmit(signedDataList[submittedOffset:end], &datypes.ResultSubmit{BaseResult: datypes.BaseResult{Code: datypes.StatusSuccess, SubmittedCount: uint64(submittedCount), Height: daHeight}})
+				submittedOffset = end
 			}
 			if onSubmitSuccess != nil {
 				onSubmitSuccess()
@@ -383,6 +389,15 @@ func (s *DASubmitter) submitWithRetry(
 		switch res.Code {
 		case datypes.StatusSuccess:
 			submitted := int(res.SubmittedCount)
+			if submitted <= 0 || submitted > len(marshaled) {
+				err := fmt.Errorf("invalid submitted count %d for batch size %d", submitted, len(marshaled))
+				s.recordFailure(common.DASubmitterFailureReasonUnknown)
+				s.logger.Error().Err(err).Str("itemType", itemType).Msg("DA layer returned invalid submitted count")
+				if onError != nil {
+					onError(err)
+				}
+				return
+			}
 			if onSuccess != nil {
 				onSuccess(submitted, res.Height)
 			}

block/internal/submitting/submitter.go

@@ -154,7 +154,6 @@ func (s *Submitter) Stop() error {
 	if s.cancel != nil {
 		s.cancel()
 	}
-	s.daSubmitter.Close()
 	// Wait for goroutines to finish with a timeout to prevent hanging
 	done := make(chan struct{})
 	go func() {
@@ -163,10 +162,10 @@ func (s *Submitter) Stop() error {
 	}()
 	select {
 	case <-done:
-		// All goroutines finished cleanly
 	case <-time.After(5 * time.Second):
 		s.logger.Warn().Msg("submitter shutdown timed out waiting for goroutines, proceeding anyway")
 	}
+	s.daSubmitter.Close()
 	s.logger.Info().Msg("submitter stopped")
 	return nil
 }
@@ -256,7 +255,9 @@ func (s *Submitter) daSubmissionLoop() {
 								s.sendCriticalError(fmt.Errorf("unrecoverable DA submission error: %w", err))
 								return
 							}
-							s.logger.Error().Err(err).Msg("failed to submit headers")
+							if err != nil {
+								s.logger.Error().Err(err).Msg("failed to submit headers")
+							}
 						}
 						if err := s.daSubmitter.SubmitHeaders(s.ctx, headers, marshalledHeaders, s.cache, s.signer, onSuccess, onError); err != nil {
 							if len(headers) > 0 {
@@ -334,7 +335,9 @@ func (s *Submitter) daSubmissionLoop() {
 								s.sendCriticalError(fmt.Errorf("unrecoverable DA submission error: %w", err))
 								return
 							}
-							s.logger.Error().Err(err).Msg("failed to submit data")
+							if err != nil {
+								s.logger.Error().Err(err).Msg("failed to submit data")
+							}
 						}
 						if err := s.daSubmitter.SubmitData(s.ctx, signedDataList, marshalledData, s.cache, s.signer, s.genesis, onSuccess, onError); err != nil {
 							if len(signedDataList) > 0 {