feat!: differentiate namespaced padded shares (#1341)

Implements the proposal in
#783 (comment).

Closes #1136
Closes #783
Opens #1344

---------

Co-authored-by: Callum Waters <[email protected]>

pkg/shares/namespaced_padding.go

@@ -2,6 +2,7 @@ package shares
 
 import (
 	"bytes"
+	"encoding/binary"
 
 	"github.com/celestiaorg/celestia-app/pkg/appconsts"
 	"github.com/celestiaorg/nmt/namespace"
@@ -12,15 +13,20 @@ import (
 // conforms to non-interactive default rules. The ns parameter provided should
 // be the namespace of the blob that precedes this padding in the data square.
 func NamespacedPaddedShare(ns namespace.ID) Share {
-	infoByte, err := NewInfoByte(appconsts.ShareVersionZero, false)
+	infoByte, err := NewInfoByte(appconsts.ShareVersionZero, true)
 	if err != nil {
 		panic(err)
 	}
-	padding := bytes.Repeat([]byte{0}, appconsts.ShareSize-appconsts.NamespaceSize-appconsts.ShareInfoBytes)
+
+	sequenceLen := make([]byte, appconsts.SequenceLenBytes)
+	binary.BigEndian.PutUint32(sequenceLen, uint32(0))
+
+	padding := bytes.Repeat([]byte{0}, appconsts.FirstSparseShareContentSize)
 
 	share := make([]byte, 0, appconsts.ShareSize)
 	share = append(share, ns...)
 	share = append(share, byte(infoByte))
+	share = append(share, sequenceLen...)
 	share = append(share, padding...)
 	return share
 }
@@ -33,3 +39,16 @@ func NamespacedPaddedShares(ns namespace.ID, n int) []Share {
 	}
 	return shares
 }
+
+func IsNamespacedPadded(s Share) (bool, error) {
+	isSequenceStart, err := s.IsSequenceStart()
+	if err != nil {
+		return false, err
+	}
+	sequenceLen, err := s.SequenceLen()
+	if err != nil {
+		return false, err
+	}
+
+	return isSequenceStart && sequenceLen == 0, nil
+}

pkg/shares/namespaced_padding_test.go

@@ -13,7 +13,8 @@ func TestNamespacedPaddedShare(t *testing.T) {
 
 	want, _ := zeroPadIfNecessary([]byte{
 		1, 1, 1, 1, 1, 1, 1, 1, // namespace ID
-		0x00, // info byte
+		1,          // info byte
+		0, 0, 0, 0, // sequence len
 	}, appconsts.ShareSize)
 
 	got := NamespacedPaddedShare(namespaceOne).ToBytes()
@@ -25,11 +26,60 @@ func TestNamespacedPaddedShares(t *testing.T) {
 
 	want, _ := zeroPadIfNecessary([]byte{
 		1, 1, 1, 1, 1, 1, 1, 1, // namespace ID
-		0x00, // info byte
+		1,          // info byte
+		0, 0, 0, 0, // sequence len
 	}, appconsts.ShareSize)
 
 	shares := NamespacedPaddedShares(namespaceOne, 2)
 	for _, share := range shares {
 		assert.Equal(t, want, share.ToBytes())
 	}
 }
+
+func TestIsNamespacedPadded(t *testing.T) {
+	type testCase struct {
+		name    string
+		share   Share
+		want    bool
+		wantErr bool
+	}
+	emptyShare := Share{}
+	blobShare, _ := zeroPadIfNecessary([]byte{
+		1, 1, 1, 1, 1, 1, 1, 1, // namespace ID
+		1,          // info byte
+		0, 0, 0, 1, // sequence len
+		0xff, // data
+	}, appconsts.ShareSize)
+
+	testCases := []testCase{
+		{
+			name:  "namespaced padded share",
+			share: NamespacedPaddedShare(namespace.ID{1, 1, 1, 1, 1, 1, 1, 1}),
+			want:  true,
+		},
+		{
+			name:    "empty share",
+			share:   emptyShare,
+			want:    false,
+			wantErr: true,
+		},
+		{
+			name:    "blob share",
+			share:   blobShare,
+			want:    false,
+			wantErr: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := IsNamespacedPadded(tc.share)
+			if tc.wantErr {
+				assert.Error(t, err)
+				return
+			}
+			assert.NoError(t, err)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}

pkg/shares/parse_sparse_shares.go

@@ -31,6 +31,14 @@ func parseSparseShares(rawShares [][]byte, supportedShareVersions []uint8) (blob
 			return nil, fmt.Errorf("unsupported share version %v is not present in supported share versions %v", version, supportedShareVersions)
 		}
 
+		isNamespacedPadded, err := IsNamespacedPadded(share)
+		if err != nil {
+			return nil, err
+		}
+		if isNamespacedPadded {
+			continue
+		}
+
 		isStart, err := share.IsSequenceStart()
 		if err != nil {
 			return nil, err