feat(blob): extend blob struct with index field (#3165)

What was done:
* added index to the blob struct;
* introduced a new `parser` entity that collects shares and transforms
them to blob;
* added tests to compare shares received from share service by
coordinates that have gotten from blob index;

blob/blob.go

@@ -9,13 +9,14 @@ import (
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 
 	"github.com/celestiaorg/celestia-app/pkg/appconsts"
-	"github.com/celestiaorg/celestia-app/pkg/shares"
 	"github.com/celestiaorg/celestia-app/x/blob/types"
 	"github.com/celestiaorg/nmt"
 
 	"github.com/celestiaorg/celestia-node/share"
 )
 
+var errEmptyShares = errors.New("empty shares")
+
 // Commitment is a Merkle Root of the subtree built from shares of the Blob.
 // It is computed by splitting the blob into shares and building the Merkle subtree to be included
 // after Submit.
@@ -31,6 +32,9 @@ func (com Commitment) Equal(c Commitment) bool {
 }
 
 // Proof is a collection of nmt.Proofs that verifies the inclusion of the data.
+// Proof proves the WHOLE namespaced data for the particular row.
+// TODO (@vgonkivs): rework `Proof` in order to prove a particular blob.
+// https://github.com/celestiaorg/celestia-node/issues/2303
 type Proof []*nmt.Proof
 
 func (p Proof) Len() int { return len(p) }
@@ -99,6 +103,10 @@ type Blob struct {
 	// the celestia-node's namespace type
 	// this is to avoid converting to and from app's type
 	namespace share.Namespace
+
+	// index represents the index of the blob's first share in the EDS.
+	// Only retrieved, on-chain blobs will have the index set. Default is -1.
+	index int
 }
 
 // NewBlobV0 constructs a new blob from the provided Namespace and data.
@@ -127,19 +135,30 @@ func NewBlob(shareVersion uint8, namespace share.Namespace, data []byte) (*Blob,
 	if err != nil {
 		return nil, err
 	}
-	return &Blob{Blob: blob, Commitment: com, namespace: namespace}, nil
+	return &Blob{Blob: blob, Commitment: com, namespace: namespace, index: -1}, nil
 }
 
 // Namespace returns blob's namespace.
 func (b *Blob) Namespace() share.Namespace {
 	return b.namespace
 }
 
+// Index returns the blob's first share index in the EDS.
+// Only retrieved, on-chain blobs will have the index set. Default is -1.
+func (b *Blob) Index() int {
+	return b.index
+}
+
+func (b *Blob) compareCommitments(com Commitment) bool {
+	return bytes.Equal(b.Commitment, com)
+}
+
 type jsonBlob struct {
 	Namespace    share.Namespace `json:"namespace"`
 	Data         []byte          `json:"data"`
 	ShareVersion uint32          `json:"share_version"`
 	Commitment   Commitment      `json:"commitment"`
+	Index        int             `json:"index"`
 }
 
 func (b *Blob) MarshalJSON() ([]byte, error) {
@@ -148,6 +167,7 @@ func (b *Blob) MarshalJSON() ([]byte, error) {
 		Data:         b.Data,
 		ShareVersion: b.ShareVersion,
 		Commitment:   b.Commitment,
+		Index:        b.index,
 	}
 	return json.Marshal(blob)
 }
@@ -165,39 +185,6 @@ func (b *Blob) UnmarshalJSON(data []byte) error {
 	b.Blob.ShareVersion = blob.ShareVersion
 	b.Commitment = blob.Commitment
 	b.namespace = blob.Namespace
+	b.index = blob.Index
 	return nil
 }
-
-// buildBlobsIfExist takes shares and tries building the Blobs from them.
-// It will build blobs either until appShares will be empty or the first incomplete blob will
-// appear, so in this specific case it will return all built blobs + remaining shares.
-func buildBlobsIfExist(appShares []shares.Share) ([]*Blob, []shares.Share, error) {
-	if len(appShares) == 0 {
-		return nil, nil, errors.New("empty shares received")
-	}
-	blobs := make([]*Blob, 0, len(appShares))
-	for {
-		length, err := appShares[0].SequenceLen()
-		if err != nil {
-			return nil, nil, err
-		}
-
-		amount := shares.SparseSharesNeeded(length)
-		if amount > len(appShares) {
-			return blobs, appShares, nil
-		}
-
-		b, err := parseShares(appShares[:amount])
-		if err != nil {
-			return nil, nil, err
-		}
-
-		// only 1 blob will be created bc we passed the exact amount of shares
-		blobs = append(blobs, b[0])
-
-		if amount == len(appShares) {
-			return blobs, nil, nil
-		}
-		appShares = appShares[amount:]
-	}
-}

blob/blob_test.go

@@ -1,7 +1,7 @@
 package blob
 
 import (
-	"reflect"
+	"bytes"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -14,7 +14,7 @@ import (
 )
 
 func TestBlob(t *testing.T) {
-	appBlobs, err := blobtest.GenerateV0Blobs([]int{1}, false)
+	appBlobs, err := blobtest.GenerateV0Blobs([]int{16}, false)
 	require.NoError(t, err)
 	blob, err := convertBlobs(appBlobs...)
 	require.NoError(t, err)
@@ -53,10 +53,12 @@ func TestBlob(t *testing.T) {
 			expectedRes: func(t *testing.T) {
 				sh, err := BlobsToShares(blob...)
 				require.NoError(t, err)
-				b, err := SharesToBlobs(sh)
+				shares, err := toAppShares(sh...)
 				require.NoError(t, err)
-				assert.Equal(t, len(b), 1)
-				assert.Equal(t, blob[0].Commitment, b[0].Commitment)
+				p := &parser{length: len(shares), shares: shares}
+				b, err := p.parse()
+				require.NoError(t, err)
+				assert.Equal(t, blob[0].Commitment, b.Commitment)
 			},
 		},
 		{
@@ -67,7 +69,8 @@ func TestBlob(t *testing.T) {
 
 				newBlob := &Blob{}
 				require.NoError(t, newBlob.UnmarshalJSON(data))
-				require.True(t, reflect.DeepEqual(blob[0], newBlob))
+				require.True(t, bytes.Equal(blob[0].Blob.Data, newBlob.Data))
+				require.True(t, bytes.Equal(blob[0].Commitment, newBlob.Commitment))
 			},
 		},
 	}

blob/helper.go

@@ -11,55 +11,6 @@ import (
 	"github.com/celestiaorg/celestia-node/share"
 )
 
-// SharesToBlobs takes raw shares and converts them to the blobs.
-func SharesToBlobs(rawShares []share.Share) ([]*Blob, error) {
-	if len(rawShares) == 0 {
-		return nil, ErrBlobNotFound
-	}
-
-	appShares, err := toAppShares(rawShares...)
-	if err != nil {
-		return nil, err
-	}
-	return parseShares(appShares)
-}
-
-// parseShares takes shares and converts them to the []*Blob.
-func parseShares(appShrs []shares.Share) ([]*Blob, error) {
-	shareSequences, err := shares.ParseShares(appShrs, true)
-	if err != nil {
-		return nil, err
-	}
-
-	// ensure that sequence length is not 0
-	if len(shareSequences) == 0 {
-		return nil, ErrBlobNotFound
-	}
-
-	blobs := make([]*Blob, len(shareSequences))
-	for i, sequence := range shareSequences {
-		data, err := sequence.RawData()
-		if err != nil {
-			return nil, err
-		}
-		if len(data) == 0 {
-			continue
-		}
-
-		shareVersion, err := sequence.Shares[0].Version()
-		if err != nil {
-			return nil, err
-		}
-
-		blob, err := NewBlob(shareVersion, sequence.Namespace.Bytes(), data)
-		if err != nil {
-			return nil, err
-		}
-		blobs[i] = blob
-	}
-	return blobs, nil
-}
-
 // BlobsToShares accepts blobs and convert them to the Shares.
 func BlobsToShares(blobs ...*Blob) ([]share.Share, error) {
 	b := make([]types.Blob, len(blobs))
@@ -75,7 +26,7 @@ func BlobsToShares(blobs ...*Blob) ([]share.Share, error) {
 
 	sort.Slice(b, func(i, j int) bool {
 		val := bytes.Compare(b[i].NamespaceID, b[j].NamespaceID)
-		return val <= 0
+		return val < 0
 	})
 
 	rawShares, err := shares.SplitBlobs(b...)
@@ -84,3 +35,22 @@ func BlobsToShares(blobs ...*Blob) ([]share.Share, error) {
 	}
 	return shares.ToBytes(rawShares), nil
 }
+
+// toAppShares converts node's raw shares to the app shares, skipping padding
+func toAppShares(shrs ...share.Share) ([]shares.Share, error) {
+	appShrs := make([]shares.Share, 0, len(shrs))
+	for _, shr := range shrs {
+		bShare, err := shares.NewShare(shr)
+		if err != nil {
+			return nil, err
+		}
+		appShrs = append(appShrs, *bShare)
+	}
+	return appShrs, nil
+}
+
+func calculateIndex(rowLength, blobIndex int) (row, col int) {
+	row = blobIndex / rowLength
+	col = blobIndex - (row * rowLength)
+	return
+}