update metrics not to double count blob failures

disperser/batcher/batcher.go

@@ -362,21 +362,28 @@ func (b *Batcher) ProcessConfirmedBatch(ctx context.Context, receiptOrErr *Recei
 
 func (b *Batcher) handleFailure(ctx context.Context, blobMetadatas []*disperser.BlobMetadata, reason FailReason) error {
 	var result *multierror.Error
+	numPermanentFailures := 0
 	for _, metadata := range blobMetadatas {
 		b.EncodingStreamer.RemoveEncodedBlob(metadata)
-		err := b.Queue.HandleBlobFailure(ctx, metadata, b.MaxNumRetriesPerBlob)
+		retry, err := b.Queue.HandleBlobFailure(ctx, metadata, b.MaxNumRetriesPerBlob)
 		if err != nil {
 			b.logger.Error("HandleSingleBatch: error handling blob failure", "err", err)
 			// Append the error
 			result = multierror.Append(result, err)
 		}
+
+		if retry {
+			continue
+		}
+
 		if reason == FailNoSignatures {
 			b.Metrics.UpdateCompletedBlob(int(metadata.RequestMetadata.BlobSize), disperser.InsufficientSignatures)
 		} else {
 			b.Metrics.UpdateCompletedBlob(int(metadata.RequestMetadata.BlobSize), disperser.Failed)
 		}
+		numPermanentFailures++
 	}
-	b.Metrics.UpdateBatchError(reason, len(blobMetadatas))
+	b.Metrics.UpdateBatchError(reason, numPermanentFailures)
 
 	// Return the error(s)
 	return result.ErrorOrNil()

disperser/batcher/encoding_streamer.go

@@ -611,7 +611,7 @@ func (e *EncodingStreamer) validateMetadataQuorums(metadatas []*disperser.BlobMe
 		if valid {
 			validMetadata = append(validMetadata, metadata)
 		} else {
-			err := e.blobStore.HandleBlobFailure(context.Background(), metadata, 0)
+			_, err := e.blobStore.HandleBlobFailure(context.Background(), metadata, 0)
 			if err != nil {
 				e.logger.Error("error handling blob failure", "err", err)
 			}

disperser/batcher/finalizer.go

@@ -179,7 +179,7 @@ func (f *finalizer) updateBlobs(ctx context.Context, metadatas []*disperser.Blob
 		confirmationBlockNumber, err := f.getTransactionBlockNumber(ctx, confirmationMetadata.ConfirmationInfo.ConfirmationTxnHash)
 		if errors.Is(err, ethereum.NotFound) {
 			// The confirmed block is finalized, but the transaction is not found. It means the transaction should be considered forked/invalid and the blob should be considered as failed.
-			err := f.blobStore.HandleBlobFailure(ctx, m, f.maxNumRetriesPerBlob)
+			_, err := f.blobStore.HandleBlobFailure(ctx, m, f.maxNumRetriesPerBlob)
 			if err != nil {
 				f.logger.Error("error marking blob as failed", "blobKey", blobKey.String(), "err", err)
 			}

disperser/common/blobstore/shared_storage.go

@@ -233,11 +233,11 @@ func (s *SharedBlobStore) GetBlobMetadata(ctx context.Context, metadataKey dispe
 	return s.blobMetadataStore.GetBlobMetadata(ctx, metadataKey)
 }
 
-func (s *SharedBlobStore) HandleBlobFailure(ctx context.Context, metadata *disperser.BlobMetadata, maxRetry uint) error {
+func (s *SharedBlobStore) HandleBlobFailure(ctx context.Context, metadata *disperser.BlobMetadata, maxRetry uint) (bool, error) {
 	if metadata.NumRetries < maxRetry {
-		return s.IncrementBlobRetryCount(ctx, metadata)
+		return true, s.IncrementBlobRetryCount(ctx, metadata)
 	} else {
-		return s.MarkBlobFailed(ctx, metadata.GetBlobKey())
+		return false, s.MarkBlobFailed(ctx, metadata.GetBlobKey())
 	}
 }
 

disperser/disperser.go

@@ -165,7 +165,8 @@ type BlobStore interface {
 	// GetBlobMetadata returns a blob metadata given a metadata key
 	GetBlobMetadata(ctx context.Context, blobKey BlobKey) (*BlobMetadata, error)
 	// HandleBlobFailure handles a blob failure by either incrementing the retry count or marking the blob as failed
-	HandleBlobFailure(ctx context.Context, metadata *BlobMetadata, maxRetry uint) error
+	// Returns a boolean indicating whether the blob should be retried and an error
+	HandleBlobFailure(ctx context.Context, metadata *BlobMetadata, maxRetry uint) (bool, error)
 }
 
 type Dispatcher interface {