fix(metrics): proper count of pending blobs metrics

Author: julienrbrt

block/internal/submitting/da_submitter.go

@@ -235,7 +235,6 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, headers []*types.Signed
 		"header",
 		s.client.GetHeaderNamespace(),
 		[]byte(s.config.DA.SubmitOptions),
-		func() uint64 { return cache.NumPendingHeaders() },
 	)
 }
 
@@ -435,7 +434,6 @@ func (s *DASubmitter) SubmitData(ctx context.Context, unsignedDataList []*types.
 		"data",
 		s.client.GetDataNamespace(),
 		[]byte(s.config.DA.SubmitOptions),
-		func() uint64 { return cache.NumPendingData() },
 	)
 }
 
@@ -545,7 +543,6 @@ func submitToDA[T any](
 	itemType string,
 	namespace []byte,
 	options []byte,
-	getTotalPendingFn func() uint64,
 ) error {
 	if len(items) != len(marshaled) {
 		return fmt.Errorf("items length (%d) does not match marshaled length (%d)", len(items), len(marshaled))
@@ -570,11 +567,6 @@ func submitToDA[T any](
 		marshaled = batchMarshaled
 	}
 
-	// Update pending blobs metric to track total backlog
-	if getTotalPendingFn != nil {
-		s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
-	}
-
 	// Start the retry loop
 	for rs.Attempt < pol.MaxAttempts {
 		// Record resend metric for retry attempts (not the first attempt)
@@ -615,20 +607,12 @@ func submitToDA[T any](
 			s.logger.Info().Str("itemType", itemType).Uint64("count", res.SubmittedCount).Msg("successfully submitted items to DA layer")
 			if int(res.SubmittedCount) == len(items) {
 				rs.Next(reasonSuccess, pol)
-				// Update pending blobs metric to reflect total backlog
-				if getTotalPendingFn != nil {
-					s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
-				}
 				return nil
 			}
 			// partial success: advance window
 			items = items[res.SubmittedCount:]
 			marshaled = marshaled[res.SubmittedCount:]
 			rs.Next(reasonSuccess, pol)
-			// Update pending blobs count to reflect total backlog
-			if getTotalPendingFn != nil {
-				s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
-			}
 
 		case datypes.StatusTooBig:
 			// Record failure metric
@@ -649,10 +633,6 @@ func submitToDA[T any](
 			marshaled = marshaled[:half]
 			s.logger.Debug().Int("newBatchSize", half).Msg("batch too big; halving and retrying")
 			rs.Next(reasonTooBig, pol)
-			// Update pending blobs count to reflect total backlog
-			if getTotalPendingFn != nil {
-				s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
-			}
 
 		case datypes.StatusNotIncludedInBlock:
 			// Record failure metric

block/internal/submitting/da_submitter_mocks_test.go

@@ -80,7 +80,6 @@ func TestSubmitToDA_MempoolRetry_IncreasesGasAndSucceeds(t *testing.T) {
 		"item",
 		nsBz,
 		opts,
-		nil,
 	)
 	assert.NoError(t, err)
 
@@ -129,7 +128,6 @@ func TestSubmitToDA_UnknownError_RetriesSameGasThenSucceeds(t *testing.T) {
 		"item",
 		nsBz,
 		opts,
-		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []float64{-1, -1}, usedGas)
@@ -180,7 +178,6 @@ func TestSubmitToDA_TooBig_HalvesBatch(t *testing.T) {
 		"item",
 		nsBz,
 		opts,
-		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []int{4, 2}, batchSizes)
@@ -225,7 +222,6 @@ func TestSubmitToDA_SentinelNoGas_PreservesGasAcrossRetries(t *testing.T) {
 		"item",
 		nsBz,
 		opts,
-		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []float64{-1, -1}, usedGas)
@@ -269,7 +265,6 @@ func TestSubmitToDA_PartialSuccess_AdvancesWindow(t *testing.T) {
 		"item",
 		nsBz,
 		opts,
-		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, 3, totalSubmitted)

block/internal/submitting/submitter.go

@@ -185,6 +185,7 @@ func (s *Submitter) daSubmissionLoop() {
 		case <-ticker.C:
 			// Check if we should submit headers based on batching strategy
 			headersNb := s.cache.NumPendingHeaders()
+
 			if headersNb > 0 {
 				lastSubmitNanos := s.lastHeaderSubmit.Load()
 				timeSinceLastSubmit := time.Since(time.Unix(0, lastSubmitNanos))
@@ -305,6 +306,9 @@ func (s *Submitter) daSubmissionLoop() {
 					}()
 				}
 			}
+
+			// Update metrics with current pending counts
+			s.metrics.DASubmitterPendingBlobs.Set(float64(headersNb + dataNb))
 		}
 	}
 }