Support specifying PieceCID in retrievals

retrievalmarket/impl/blockunsealing/blockunsealing.go

@@ -26,15 +26,16 @@ type loaderWithUnsealing struct {
 	pieceStore piecestore.PieceStore
 	carIO      pieceio.CarIO
 	unsealer   UnsealingFunc
+	pieceCid   *cid.Cid
 }
 
 // UnsealingFunc is a function that unseals sectors at a given offset and length
 type UnsealingFunc func(ctx context.Context, sectorId uint64, offset uint64, length uint64) (io.ReadCloser, error)
 
 // NewLoaderWithUnsealing creates a loader that will attempt to read blocks from the blockstore but unseal the piece
 // as needed using the passed unsealing function
-func NewLoaderWithUnsealing(ctx context.Context, bs blockstore.Blockstore, pieceStore piecestore.PieceStore, carIO pieceio.CarIO, unsealer UnsealingFunc) LoaderWithUnsealing {
-	return &loaderWithUnsealing{ctx, bs, pieceStore, carIO, unsealer}
+func NewLoaderWithUnsealing(ctx context.Context, bs blockstore.Blockstore, pieceStore piecestore.PieceStore, carIO pieceio.CarIO, unsealer UnsealingFunc, pieceCid *cid.Cid) LoaderWithUnsealing {
+	return &loaderWithUnsealing{ctx, bs, pieceStore, carIO, unsealer, pieceCid}
 }
 
 func (lu *loaderWithUnsealing) Load(lnk ipld.Link, lnkCtx ipld.LinkContext) (io.Reader, error) {
@@ -66,13 +67,21 @@ func (lu *loaderWithUnsealing) Load(lnk ipld.Link, lnkCtx ipld.LinkContext) (io.
 }
 
 func (lu *loaderWithUnsealing) attemptUnseal(c cid.Cid) error {
+	var err error
+	var reader io.Reader
+	var cidInfo piecestore.CIDInfo
+
+	// if the deal proposal specified a Piece CID, only check that piece
+	if lu.pieceCid != nil {
+		reader, err = lu.firstSuccessfulUnsealByPieceCID(*lu.pieceCid)
+	} else {
+		cidInfo, err = lu.pieceStore.GetCIDInfo(c)
+		if err != nil {
+			return xerrors.Errorf("error looking up information on CID: %w", err)
+		}
 
-	cidInfo, err := lu.pieceStore.GetCIDInfo(c)
-	if err != nil {
-		return xerrors.Errorf("error looking up information on CID: %w", err)
+		reader, err = lu.firstSuccessfulUnseal(cidInfo)
 	}
-
-	reader, err := lu.firstSuccessfulUnseal(cidInfo)
 	// no successful unseal
 	if err != nil {
 		return xerrors.Errorf("Unable to unseal piece: %w", err)
@@ -87,21 +96,32 @@ func (lu *loaderWithUnsealing) attemptUnseal(c cid.Cid) error {
 	return nil
 }
 
-func (lu *loaderWithUnsealing) firstSuccessfulUnseal(cidInfo piecestore.CIDInfo) (io.ReadCloser, error) {
+func (lu *loaderWithUnsealing) firstSuccessfulUnseal(payloadCidInfo piecestore.CIDInfo) (io.ReadCloser, error) {
+	var lastErr error
+	for _, pieceBlockLocation := range payloadCidInfo.PieceBlockLocations {
+		reader, err := lu.firstSuccessfulUnsealByPieceCID(pieceBlockLocation.PieceCID)
+		if err == nil {
+			return reader, nil
+		}
+		lastErr = err
+	}
+	return nil, lastErr
+}
+
+func (lu *loaderWithUnsealing) firstSuccessfulUnsealByPieceCID(pieceCID cid.Cid) (io.ReadCloser, error) {
+	pieceInfo, err := lu.pieceStore.GetPieceInfo(pieceCID)
+	if err != nil {
+		return nil, err
+	}
+
 	// try to unseal data from all pieces
 	lastErr := xerrors.New("no sectors found to unseal from")
-	for _, pieceBlockLocation := range cidInfo.PieceBlockLocations {
-		pieceInfo, err := lu.pieceStore.GetPieceInfo(pieceBlockLocation.PieceCID)
-		if err != nil {
-			continue
-		}
-		for _, deal := range pieceInfo.Deals {
-			reader, err := lu.unsealer(lu.ctx, deal.SectorID, deal.Offset, deal.Length)
-			if err == nil {
-				return reader, nil
-			}
-			lastErr = err
+	for _, deal := range pieceInfo.Deals {
+		reader, err := lu.unsealer(lu.ctx, deal.SectorID, deal.Offset, deal.Length)
+		if err == nil {
+			return reader, nil
 		}
+		lastErr = err
 	}
 	return nil, lastErr
 }