Add dispersing transistion check (#741)

Co-authored-by: Bowen Xue <ubuntu@ip-172-31-91-211.ec2.internal>

disperser/common/blobstore/shared_storage.go

@@ -19,6 +19,8 @@ const (
 	maxS3BlobFetchWorkers = 64
 )
 
+var errProcessingToDispersing = errors.New("blob transit to dispersing from non processing")
+
 // The shared blob store that the disperser is operating on.
 // The metadata store is backed by DynamoDB and the blob store is backed by S3.
 //
@@ -158,6 +160,18 @@ func (s *SharedBlobStore) MarkBlobConfirmed(ctx context.Context, existingMetadat
 }
 
 func (s *SharedBlobStore) MarkBlobDispersing(ctx context.Context, metadataKey disperser.BlobKey) error {
+	refreshedMetadata, err := s.GetBlobMetadata(ctx, metadataKey)
+	if err != nil {
+		s.logger.Error("error getting blob metadata while marking blobDispersing", "err", err)
+		return err
+	}
+
+	status := refreshedMetadata.BlobStatus
+	if status != disperser.Processing {
+		s.logger.Error("error marking blob as dispersing from non processing state", "blobKey", metadataKey.String(), "status", status)
+		return errProcessingToDispersing
+	}
+
 	return s.blobMetadataStore.SetBlobStatus(ctx, metadataKey, disperser.Dispersing)
 }