add some fixes

Author: vgonkivs

blob/blob.go

@@ -34,25 +34,22 @@ func (p Proof) verify(blob *Blob, header *header.ExtendedHeader) error {
 	if err != nil {
 		return err
 	}
-	toCoords, err := shwap.SampleCoordsFrom1DIndex(blob.Index()+len(shrs)-1, len(header.DAH.RowRoots))
-	if err != nil {
-		return err
-	}
 
 	hasher := share.NewSHA256Hasher()
-	for from, i := fromCoords.Row, 0; from <= toCoords.Row; from++ {
+	shareOffset := 0
+	for proofIdx := 0; proofIdx < len(p); proofIdx++ {
 		hasher.Reset()
-		sharesPerRow := p[i].End() - p[i].Start()
-		valid := p[i].VerifyInclusion(
+		sharesInProof := p[proofIdx].End() - p[proofIdx].Start()
+		valid := p[proofIdx].VerifyInclusion(
 			hasher,
 			blob.Namespace().Bytes(),
-			libshare.ToBytes(shrs[i:sharesPerRow]),
-			header.DAH.RowRoots[from],
+			libshare.ToBytes(shrs[shareOffset:shareOffset+sharesInProof]),
+			header.DAH.RowRoots[fromCoords.Row+proofIdx],
 		)
 		if !valid {
-			return errors.New("invalid share proof")
+			return ErrInvalidProof
 		}
-		i += sharesPerRow
+		shareOffset += sharesInProof
 	}
 	return nil
 }

blob/helper.go

@@ -75,7 +75,7 @@ func fromShwapBlob(shBlob *shwap.Blob, odsSize int) (*Blob, error) {
 		return nil, err
 	}
 
-	edsIndex, err := shwap.SampleCoordsAs1DIndex(coords, odsIndex*2)
+	edsIndex, err := shwap.SampleCoordsAs1DIndex(coords, odsSize*2)
 	if err != nil {
 		return nil, err
 	}

blob/metrics.go

@@ -3,6 +3,7 @@ package blob
 import (
 	"context"
 	"errors"
+	"github.com/celestiaorg/celestia-node/share/shwap"
 	"time"
 
 	"go.opentelemetry.io/otel"
@@ -91,7 +92,7 @@ func (m *metrics) observeRetrieval(ctx context.Context, duration time.Duration,
 	if err != nil {
 		errorType := errorTypeUnknown
 		switch {
-		case errors.Is(err, ErrBlobNotFound):
+		case errors.Is(err, shwap.ErrBlobNotFound):
 			errorType = errorTypeNotFound
 		case errors.Is(err, context.DeadlineExceeded):
 			errorType = errorTypeTimeout

blob/service.go

@@ -30,7 +30,6 @@ import (
 )
 
 var (
-	ErrBlobNotFound = errors.New("blob: not found")
 	ErrInvalidProof = errors.New("blob: invalid proof")
 
 	log    = logging.Logger("blob")
@@ -296,7 +295,7 @@ func (s *Service) GetProof(
 // the user will receive all found blobs along with a combined error message.
 //
 // All blobs will preserve the order of the namespaces that were requested.
-func (s *Service) GetAll(ctx context.Context, height uint64, namespaces []libshare.Namespace) (_ []*Blob, err error) {
+func (s *Service) GetAll(ctx context.Context, height uint64, namespaces []libshare.Namespace) (blobs []*Blob, err error) {
 	ctx, span := tracer.Start(ctx, "blob/get-all")
 	defer func() {
 		utils.SetStatusAndEnd(span, err)
@@ -318,49 +317,62 @@ func (s *Service) GetAll(ctx context.Context, height uint64, namespaces []libsha
 
 	header, err := s.headerGetter(ctx, height)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("err getting header: %w", err)
+	}
+
+	blobs, err = s.getAll(ctx, header, namespaces)
+	if err != nil {
+		return blobs, fmt.Errorf("err getting blobs: %w", err)
 	}
 
-	return s.getAll(ctx, header, namespaces)
+	namespaceStrings := make([]string, len(namespaces))
+	for i := range namespaces {
+		namespaceStrings[i] = namespaces[i].String()
+	}
+	span.SetAttributes(attribute.StringSlice("namespaces", namespaceStrings))
+	span.SetAttributes(attribute.Int("total", len(blobs)))
+	return blobs, nil
 }
 
 func (s *Service) getAll(
 	ctx context.Context,
 	header *header.ExtendedHeader,
 	namespaces []libshare.Namespace,
-) ([]*Blob, error) {
+) (_ []*Blob, err error) {
 	var (
-		span             = trace.SpanFromContext(ctx)
-		namespaceStrings = make([]string, len(namespaces))
-		resultBlobs      = make([][]*shwap.Blob, len(namespaces))
-		resultErr        = make([]error, len(namespaces))
-		wg               = sync.WaitGroup{}
+		resultBlobs = make([][]*shwap.Blob, len(namespaces))
+		resultErr   = make([]error, len(namespaces))
+		wg          = sync.WaitGroup{}
 	)
 
 	for i, namespace := range namespaces {
 		wg.Add(1)
 		go func(i int, namespace libshare.Namespace) {
 			defer wg.Done()
+			log.Debugw("retrieving all blobs from", "namespace", namespace.String(), "height", header.Height())
 			resultBlobs[i], resultErr[i] = s.shareGetter.GetBlobs(ctx, header, namespace)
+			if errors.Is(resultErr[i], shwap.ErrNotFound) {
+				resultErr[i] = nil
+			}
 		}(i, namespace)
-
-		namespaceStrings[i] = namespace.String()
 	}
-	span.SetAttributes(attribute.StringSlice("namespaces", namespaceStrings))
 	wg.Wait()
 
 	blbs := slices.Concat(resultBlobs...)
-	errs := errors.Join(resultErr...)
+	err = errors.Join(resultErr...)
+	if len(blbs) == 0 {
+		return nil, err
+	}
 
 	blobs := make([]*Blob, len(blbs))
 	for i, blb := range blbs {
-		blob, err := fromShwapBlob(blb, len(header.DAH.RowRoots)/2)
+		blob, convertErr := fromShwapBlob(blb, len(header.DAH.RowRoots)/2)
 		if err != nil {
-			errs = errors.Join(errs, err)
+			err = errors.Join(err, convertErr)
 		}
 		blobs[i] = blob
 	}
-	return blobs, errs
+	return blobs, err
 }
 
 // Included verifies that the blob was included in a specific height.
@@ -406,6 +418,9 @@ func (s *Service) Included(
 
 	shwapBlob, err := s.shareGetter.GetBlob(ctx, header, namespace, commitment)
 	if err != nil {
+		if errors.Is(err, shwap.ErrBlobNotFound) {
+			err = nil
+		}
 		return false, err
 	}
 

blob/service_test.go

@@ -409,15 +409,15 @@ func TestBlobService_Get(t *testing.T) {
 				innerGetter := service.shareGetter
 				getterWrapper := mock.NewMockGetter(ctrl)
 				getterWrapper.EXPECT().
-					GetNamespaceData(gomock.Any(), gomock.Any(), gomock.Any()).
+					GetBlobs(gomock.Any(), gomock.Any(), gomock.Any()).
 					DoAndReturn(
 						func(
 							ctx context.Context, h *header.ExtendedHeader, ns libshare.Namespace,
-						) (shwap.NamespaceData, error) {
+						) ([]*shwap.Blob, error) {
 							if ns.Equals(blobsWithDiffNamespaces[0].Namespace()) {
 								return nil, errors.New("internal error")
 							}
-							return innerGetter.GetNamespaceData(ctx, h, ns)
+							return innerGetter.GetBlobs(ctx, h, ns)
 						}).AnyTimes()
 
 				service.shareGetter = getterWrapper
@@ -929,7 +929,6 @@ func createService(ctx context.Context, t testing.TB, shares []libshare.Share) *
 			func(ctx context.Context,
 				h *header.ExtendedHeader,
 				ns libshare.Namespace,
-				com []byte,
 			) ([]*shwap.Blob, error) {
 				return accessor.Blobs(ctx, ns)
 			})

share/shwap/blob.go

@@ -19,6 +19,8 @@ const blobName = "blob_v0"
 
 var subtreeRootThreshold = appconsts.SubtreeRootThreshold
 
+var ErrBlobNotFound = errors.New("blob: not found")
+
 // Blob represents a retrieved blob from the data square containing
 // the blob data along with proofs for the verification and blob's index(position inside the ODS).
 type Blob struct {
@@ -50,7 +52,7 @@ func BlobFromShares(
 	colStart := 0
 	for rowStart := 0; rowStart < len(libShares); {
 		shr := libShares[rowStart][colStart]
-		if !shr.Namespace().Equals(namespace) || !shr.IsSequenceStart() {
+		if !shr.Namespace().Equals(namespace) || !shr.IsSequenceStart() || shr.IsPadding() {
 			colStart++
 			if colStart >= odsSize {
 				colStart = 0
@@ -75,11 +77,15 @@ func BlobFromShares(
 		rngShares := libShares[rowStart : to.Row+1]
 		shrs := make([]libshare.Share, 0, sharesAmount)
 		for i := range rngShares {
-			col := odsSize
+			startCol := 0
+			endCol := odsSize
+			if i == 0 {
+				startCol = from.Col // account for starting column on first row
+			}
 			if i == len(rngShares)-1 {
-				col = to.Col + 1
+				endCol = to.Col + 1
 			}
-			shrs = append(shrs, rngShares[i][:col]...)
+			shrs = append(shrs, rngShares[i][startCol:endCol]...)
 		}
 
 		blobs, err := libshare.ParseBlobs(shrs)
@@ -113,7 +119,7 @@ func BlobFromShares(
 			StartIndex:         fromIndex,
 		}, err
 	}
-	return nil, errors.New("blob not found")
+	return nil, ErrBlobNotFound
 }
 
 func (b *Blob) VerifyInclusion(roots [][]byte) error {
@@ -228,7 +234,7 @@ func BlobsFromShares(
 	colStart := 0
 	for rowStart := 0; rowStart < len(libShares); {
 		shr := libShares[rowStart][colStart]
-		if !shr.Namespace().Equals(namespace) || !shr.IsSequenceStart() {
+		if !shr.Namespace().Equals(namespace) || !shr.IsSequenceStart() || shr.IsPadding() {
 			colStart++
 			if colStart >= odsSize {
 				colStart = 0
@@ -265,7 +271,7 @@ func BlobsFromShares(
 	}
 
 	if len(blobs) == 0 {
-		return nil, errors.New("blobs were empty")
+		return nil, ErrNotFound
 	}
 	return blobs, nil
 }