fixes

Author: julienrbrt

block/internal/cache/manager.go

@@ -315,10 +315,6 @@ func (m *implementation) SetLastSubmittedHeaderHeight(ctx context.Context, heigh
 	m.pendingHeaders.SetLastSubmittedHeaderHeight(ctx, height)
 }
 
-func (m *implementation) ResetInFlightHeaderHeight() {
-	m.pendingHeaders.ResetInFlightHeaderRange(0, 0)
-}
-
 func (m *implementation) ResetInFlightHeaderRange(start, end uint64) {
 	m.pendingHeaders.ResetInFlightHeaderRange(start, end)
 }

block/internal/cache/pending_base.go

@@ -87,6 +87,7 @@ func (pb *pendingBase[T]) getPending(ctx context.Context) ([]T, error) {
 	}
 
 	pb.inFlightMu.Lock()
+	lastSubmitted = pb.lastHeight.Load()
 	rangeStart, rangeEnd := findAvailableRange(pb.gaps, pb.inFlightClaims, lastSubmitted, storeHeight)
 	pb.inFlightMu.Unlock()
 

block/internal/submitting/da_submitter_tracing.go

@@ -36,7 +36,6 @@ func (t *tracedDASubmitter) SubmitHeaders(ctx context.Context, headers []*types.
 			attribute.Int("header.count", len(headers)),
 		),
 	)
-	defer span.End()
 
 	var totalBytes int
 	for _, h := range marshalledHeaders {
@@ -51,13 +50,30 @@ func (t *tracedDASubmitter) SubmitHeaders(ctx context.Context, headers []*types.
 		)
 	}
 
-	err := t.inner.SubmitHeaders(ctx, headers, marshalledHeaders, cache, signer, onSubmitError)
+	var wrappedOnError func(error)
+	if onSubmitError != nil {
+		wrappedOnError = func(err error) {
+			if err != nil {
+				span.RecordError(err)
+				span.SetStatus(codes.Error, err.Error())
+			}
+			span.End()
+			onSubmitError(err)
+		}
+	}
+
+	err := t.inner.SubmitHeaders(ctx, headers, marshalledHeaders, cache, signer, wrappedOnError)
 	if err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
+		span.End()
 		return err
 	}
 
+	if onSubmitError == nil {
+		span.End()
+	}
+
 	return nil
 }
 
@@ -67,7 +83,6 @@ func (t *tracedDASubmitter) SubmitData(ctx context.Context, signedDataList []*ty
 			attribute.Int("data.count", len(signedDataList)),
 		),
 	)
-	defer span.End()
 
 	var totalBytes int
 	for _, d := range marshalledData {
@@ -82,13 +97,30 @@ func (t *tracedDASubmitter) SubmitData(ctx context.Context, signedDataList []*ty
 		)
 	}
 
-	err := t.inner.SubmitData(ctx, signedDataList, marshalledData, cache, signer, genesis, onSubmitError)
+	var wrappedOnError func(error)
+	if onSubmitError != nil {
+		wrappedOnError = func(err error) {
+			if err != nil {
+				span.RecordError(err)
+				span.SetStatus(codes.Error, err.Error())
+			}
+			span.End()
+			onSubmitError(err)
+		}
+	}
+
+	err := t.inner.SubmitData(ctx, signedDataList, marshalledData, cache, signer, genesis, wrappedOnError)
 	if err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
+		span.End()
 		return err
 	}
 
+	if onSubmitError == nil {
+		span.End()
+	}
+
 	return nil
 }
 

block/internal/submitting/submitter.go

@@ -154,18 +154,18 @@ func (s *Submitter) Stop() error {
 	if s.cancel != nil {
 		s.cancel()
 	}
-	// Wait for goroutines to finish with a timeout to prevent hanging
 	done := make(chan struct{})
 	go func() {
 		s.wg.Wait()
+		s.daSubmitter.Close()
 		close(done)
 	}()
 	select {
 	case <-done:
 	case <-time.After(5 * time.Second):
 		s.logger.Warn().Msg("submitter shutdown timed out waiting for goroutines, proceeding anyway")
+		s.daSubmitter.Close()
 	}
-	s.daSubmitter.Close()
 	s.logger.Info().Msg("submitter stopped")
 	return nil
 }