don't count metadata not found as internal error

disperser/apiserver/server.go

@@ -510,7 +510,9 @@ func (s *DispersalServer) GetBlobStatus(ctx context.Context, req *pb.BlobStatusR
 	s.logger.Debug("metadataKey", "metadataKey", metadataKey.String())
 	metadata, err := s.blobStore.GetBlobMetadata(ctx, metadataKey)
 	if err != nil {
-		// TODO: we need to distinguish NOT_FOUND from actual internal error.
+		if errors.Is(err, disperser.ErrMetadataNotFound) {
+			return nil, api.NewInvalidArgError("no metadata found for the requestID")
+		}
 		return nil, api.NewInternalError(fmt.Sprintf("failed to get blob metadata, blobkey: %s", metadataKey.String()))
 	}
 
@@ -639,7 +641,10 @@ func (s *DispersalServer) RetrieveBlob(ctx context.Context, req *pb.RetrieveBlob
 	blobMetadata, err := s.blobStore.GetMetadataInBatch(ctx, batchHeaderHash32, blobIndex)
 	if err != nil {
 		s.logger.Error("Failed to retrieve blob metadata", "err", err)
-		// TODO: we need to distinguish NOT_FOUND from actual internal error.
+		if errors.Is(err, disperser.ErrMetadataNotFound) {
+			s.metrics.IncrementFailedBlobRequestNum(codes.NotFound.String(), "", "RetrieveBlob")
+			return nil, api.NewInvalidArgError("no metadata found for the given batch header hash and blob index")
+		}
 		s.metrics.IncrementFailedBlobRequestNum(codes.Internal.String(), "", "RetrieveBlob")
 		return nil, api.NewInternalError("failed to get blob metadata, please retry")
 	}

disperser/common/blobstore/blob_metadata_store.go

@@ -61,6 +61,11 @@ func (s *BlobMetadataStore) GetBlobMetadata(ctx context.Context, metadataKey dis
 			Value: metadataKey.MetadataHash,
 		},
 	})
+
+	if item == nil {
+		return nil, fmt.Errorf("%w: metadata not found for key %s", disperser.ErrMetadataNotFound, metadataKey)
+	}
+
 	if err != nil {
 		return nil, err
 	}
@@ -199,7 +204,7 @@ func (s *BlobMetadataStore) GetBlobMetadataInBatch(ctx context.Context, batchHea
 	}
 
 	if len(items) == 0 {
-		return nil, fmt.Errorf("there is no metadata for batch %s and blob index %d", batchHeaderHash, blobIndex)
+		return nil, fmt.Errorf("%w: there is no metadata for batch %s and blob index %d", disperser.ErrMetadataNotFound, batchHeaderHash, blobIndex)
 	}
 
 	if len(items) > 1 {

disperser/errors.go

@@ -3,5 +3,6 @@ package disperser
 import "errors"
 
 var (
-	ErrBlobNotFound = errors.New("blob not found")
+	ErrBlobNotFound     = errors.New("blob not found")
+	ErrMetadataNotFound = errors.New("metadata not found")
 )