fix: update DASubmitterPendingBlobs to track total backlog

Changes the DASubmitterPendingBlobs metric to track the total number
of blobs awaiting submission across the entire queue, rather than just
the current batch being submitted. This provides better visibility into
the submission backlog for monitoring and alerting.

- Added getTotalPendingFn parameter to submitToDA()
- Updated metric to call NumPendingHeaders() or NumPendingData()
- Updated metric documentation and help text
- Updated all test calls with new parameter

Co-authored-by: Marko <tac0turtle@users.noreply.github.com>

Author: github-actions[bot]

block/internal/common/metrics.go

@@ -67,7 +67,7 @@ type Metrics struct {
 	// DA Submitter metrics
 	DASubmitterFailures          map[string]metrics.Counter // Counter with reason label
 	DASubmitterLastFailure       map[string]metrics.Gauge   // Timestamp gauge with reason label
-	DASubmitterPendingBlobs      metrics.Gauge              // Number of pending blobs
+	DASubmitterPendingBlobs      metrics.Gauge              // Total number of blobs awaiting submission (backlog)
 	DASubmitterResends           metrics.Counter            // Number of resend attempts
 }
 
@@ -362,7 +362,7 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 		Namespace: namespace,
 		Subsystem: MetricsSubsystem,
 		Name:      "da_submitter_pending_blobs",
-		Help:      "Number of blobs pending DA submission",
+		Help:      "Total number of blobs awaiting DA submission (backlog)",
 	}, labels).With(labelsAndValues...)
 
 	m.DASubmitterResends = prometheus.NewCounterFrom(stdprometheus.CounterOpts{

block/internal/submitting/da_submitter.go

@@ -231,6 +231,7 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager) er
 		s.namespaceBz,
 		[]byte(s.config.DA.SubmitOptions),
 		cache,
+		func() uint64 { return cache.NumPendingHeaders() },
 	)
 }
 
@@ -274,6 +275,7 @@ func (s *DASubmitter) SubmitData(ctx context.Context, cache cache.Manager, signe
 		s.namespaceDataBz,
 		[]byte(s.config.DA.SubmitOptions),
 		cache,
+		func() uint64 { return cache.NumPendingData() },
 	)
 }
 
@@ -344,6 +346,7 @@ func submitToDA[T any](
 	namespace []byte,
 	options []byte,
 	cache cache.Manager,
+	getTotalPendingFn func() uint64,
 ) error {
 	marshaled, err := marshalItems(ctx, items, marshalFn, itemType)
 	if err != nil {
@@ -368,9 +371,9 @@ func submitToDA[T any](
 		marshaled = batchMarshaled
 	}
 
-	// Update pending blobs metric
-	if s.metrics != nil {
-		s.metrics.DASubmitterPendingBlobs.Set(float64(len(items)))
+	// Update pending blobs metric to track total backlog
+	if s.metrics != nil && getTotalPendingFn != nil {
+		s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
 	}
 
 	// Start the retry loop
@@ -401,19 +404,19 @@ func submitToDA[T any](
 			s.logger.Info().Str("itemType", itemType).Float64("gasPrice", rs.GasPrice).Uint64("count", res.SubmittedCount).Msg("successfully submitted items to DA layer")
 			if int(res.SubmittedCount) == len(items) {
 				rs.Next(reasonSuccess, pol, gm, sentinelNoGas)
-				// Clear pending blobs on success
-				if s.metrics != nil {
-					s.metrics.DASubmitterPendingBlobs.Set(0)
+				// Update pending blobs metric to reflect total backlog
+				if s.metrics != nil && 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, gm, sentinelNoGas)
-			// Update pending blobs count
-			if s.metrics != nil {
-				s.metrics.DASubmitterPendingBlobs.Set(float64(len(items)))
+			// Update pending blobs count to reflect total backlog
+			if s.metrics != nil && getTotalPendingFn != nil {
+				s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
 			}
 
 		case coreda.StatusTooBig:
@@ -433,9 +436,9 @@ func submitToDA[T any](
 			marshaled = marshaled[:half]
 			s.logger.Debug().Int("newBatchSize", half).Msg("batch too big; halving and retrying")
 			rs.Next(reasonTooBig, pol, gm, sentinelNoGas)
-			// Update pending blobs count
-			if s.metrics != nil {
-				s.metrics.DASubmitterPendingBlobs.Set(float64(len(items)))
+			// Update pending blobs count to reflect total backlog
+			if s.metrics != nil && getTotalPendingFn != nil {
+				s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
 			}
 
 		case coreda.StatusNotIncludedInBlock:

block/internal/submitting/da_submitter_mocks_test.go

@@ -86,6 +86,7 @@ func TestSubmitToDA_MempoolRetry_IncreasesGasAndSucceeds(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
+		nil,
 	)
 	assert.NoError(t, err)
 
@@ -137,6 +138,7 @@ func TestSubmitToDA_UnknownError_RetriesSameGasThenSucceeds(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
+		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []float64{5.5, 5.5}, usedGas)
@@ -193,6 +195,7 @@ func TestSubmitToDA_TooBig_HalvesBatch(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
+		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []int{4, 2}, batchSizes)
@@ -242,6 +245,7 @@ func TestSubmitToDA_SentinelNoGas_PreservesGasAcrossRetries(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
+		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []float64{-1, -1}, usedGas)
@@ -282,6 +286,7 @@ func TestSubmitToDA_PartialSuccess_AdvancesWindow(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
+		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, 3, totalSubmitted)