no v1 to v2 index migration

Author: LexLuthr

extern/filecoin-ffi

@@ -19,6 +19,8 @@ import (
 	"golang.org/x/sync/errgroup"
 	"golang.org/x/xerrors"
 
+	commcid "github.com/filecoin-project/go-fil-commcid"
+
 	"github.com/filecoin-project/curio/deps/config"
 )
 
@@ -349,10 +351,10 @@ func (i *IndexStore) RemoveIndexes(ctx context.Context, pieceCidv2 cid.Cid) erro
 	return nil
 }
 
-// PieceInfo contains PieceCidV2 and BlockSize
+// PieceInfo contains PieceCid and BlockSize. PieceCid can be either v1 or v2.
 type PieceInfo struct {
-	PieceCidV2 cid.Cid
-	BlockSize  uint64
+	PieceCid  cid.Cid
+	BlockSize uint64
 }
 
 // PiecesContainingMultihash gets all pieces that contain a multihash along with their BlockSize
@@ -369,8 +371,8 @@ func (i *IndexStore) PiecesContainingMultihash(ctx context.Context, m multihash.
 			return nil, fmt.Errorf("parsing piece cid: %w", err)
 		}
 		pieces = append(pieces, PieceInfo{
-			PieceCidV2: pcid,
-			BlockSize:  blockSize,
+			PieceCid:  pcid,
+			BlockSize: blockSize,
 		})
 	}
 	if err := iter.Close(); err != nil {
@@ -397,21 +399,41 @@ func (i *IndexStore) GetOffset(ctx context.Context, pieceCidv2 cid.Cid, hash mul
 }
 
 func (i *IndexStore) GetPieceHashRange(ctx context.Context, piecev2 cid.Cid, start multihash.Multihash, num int64) ([]multihash.Multihash, error) {
-	qry := "SELECT PayloadMultihash FROM PieceBlockOffsetSize WHERE PieceCid = ? AND PayloadMultihash >= ? ORDER BY PayloadMultihash ASC LIMIT ?"
-	iter := i.session.Query(qry, piecev2.Bytes(), []byte(start), num).WithContext(ctx).Iter()
+	getHashes := func(pieceCid cid.Cid, start multihash.Multihash, num int64) ([]multihash.Multihash, error) {
+		qry := "SELECT PayloadMultihash FROM PieceBlockOffsetSize WHERE PieceCid = ? AND PayloadMultihash >= ? ORDER BY PayloadMultihash ASC LIMIT ?"
+		iter := i.session.Query(qry, pieceCid.Bytes(), []byte(start), num).WithContext(ctx).Iter()
 
-	var hashes []multihash.Multihash
-	var r []byte
-	for iter.Scan(&r) {
-		m := multihash.Multihash(r)
-		hashes = append(hashes, m)
+		var hashes []multihash.Multihash
+		var r []byte
+		for iter.Scan(&r) {
+			m := multihash.Multihash(r)
+			hashes = append(hashes, m)
 
-		// Allocate new r, preallocating the typical size of a multihash (36 bytes)
-		r = make([]byte, 0, 36)
+			// Allocate new r, preallocating the typical size of a multihash (36 bytes)
+			r = make([]byte, 0, 36)
+		}
+		if err := iter.Close(); err != nil {
+			return nil, xerrors.Errorf("iterating piece hash range (P:0x%02x, H:0x%02x, n:%d): %w", pieceCid.Bytes(), []byte(start), num, err)
+		}
+		return hashes, nil
 	}
-	if err := iter.Close(); err != nil {
-		return nil, xerrors.Errorf("iterating piece hash range (P:0x%02x, H:0x%02x, n:%d): %w", piecev2.Bytes(), []byte(start), num, err)
+
+	hashes, err := getHashes(piecev2, start, num)
+	if err != nil {
+		return nil, err
 	}
+
+	if len(hashes) == 0 {
+		pcid1, _, err := commcid.PieceCidV1FromV2(piecev2)
+		if err != nil {
+			return nil, xerrors.Errorf("getting piece cid v1 from v2: %w", err)
+		}
+		hashes, err = getHashes(pcid1, start, num)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	if len(hashes) != int(num) {
 		return nil, xerrors.Errorf("expected %d hashes, got %d (possibly missing indexes)", num, len(hashes))
 	}

market/indexstore/indexstore.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"encoding/hex"
 	"errors"
+	"fmt"
 	"io"
 	"time"
 
@@ -177,6 +178,43 @@ func (p *ServeChunker) getEntry(rctx context.Context, block cid.Cid, speculated
 		return nil, xerrors.Errorf("parsing piece CID: %w", err)
 	}
 
+	// Convert to pcid2 if needed
+	yes := commcidv2.IsPieceCidV2(pieceCidv2)
+	if !yes {
+		var rawSize int64
+		var singlePiece bool
+		err := p.db.QueryRow(ctx, `WITH meta AS (
+											  SELECT piece_size
+											  FROM market_piece_metadata
+											  WHERE piece_cid = $1
+											),
+											exact AS (
+											  SELECT COUNT(*) AS n, MIN(piece_size) AS piece_size
+											  FROM meta
+											),
+											raw AS (
+											  SELECT MAX(mpd.raw_size) AS raw_size
+											  FROM market_piece_deal mpd
+											  WHERE mpd.piece_cid   = $1
+												AND mpd.piece_length = (SELECT piece_size FROM exact)
+												AND (SELECT n FROM exact) = 1
+											)
+											SELECT
+											  COALESCE((SELECT raw_size FROM raw), 0)        AS raw_size,
+											  ((SELECT n FROM exact) = 1)                    AS has_single_metadata;`, pieceCidv2.String()).Scan(&rawSize, &singlePiece)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get piece metadata: %w", err)
+		}
+		if !singlePiece {
+			return nil, fmt.Errorf("more than 1 piece metadata found for piece cid %s, please use piece cid v2", pieceCidv2.String())
+		}
+		pcid2, err := commcidv2.PieceCidV2FromV1(pieceCidv2, uint64(rawSize))
+		if err != nil {
+			return nil, fmt.Errorf("failed to convert piece cid v1 to v2: %w", err)
+		}
+		pieceCidv2 = pcid2
+	}
+
 	if leave, ok := p.noSkipCache.Get(pieceCidv2); !ok || time.Now().After(leave) {
 		skip, err := p.checkIsEntrySkip(ctx, block)
 		if err != nil {

market/ipni/chunker/serve-chunker.go

@@ -115,7 +115,7 @@ func (ro *RemoteBlockstore) Get(ctx context.Context, c cid.Cid) (b blocks.Block,
 	var merr error
 	for _, piece := range pieces {
 		data, err := func() ([]byte, error) {
-			reader, _, err := ro.cpr.GetSharedPieceReader(ctx, piece.PieceCidV2)
+			reader, _, err := ro.cpr.GetSharedPieceReader(ctx, piece.PieceCid)
 			if err != nil {
 				return nil, fmt.Errorf("getting piece reader: %w", err)
 			}
@@ -124,19 +124,19 @@ func (ro *RemoteBlockstore) Get(ctx context.Context, c cid.Cid) (b blocks.Block,
 			}(reader)
 
 			// Get the offset of the block within the piece (CAR file)
-			offset, err := ro.idxApi.GetOffset(ctx, piece.PieceCidV2, c.Hash()) // This can be pieceCidV2 or pieceCidV1, but we don't care because we are feeding back the db output
+			offset, err := ro.idxApi.GetOffset(ctx, piece.PieceCid, c.Hash()) // This can be pieceCidV2 or pieceCidV1, but we don't care because we are feeding back the db output
 			if err != nil {
-				return nil, fmt.Errorf("getting offset/size for cid %s in piece %s: %w", c, piece.PieceCidV2, err)
+				return nil, fmt.Errorf("getting offset/size for cid %s in piece %s: %w", c, piece.PieceCid, err)
 			}
 
 			// Seek to the section offset
 			readerAt := io.NewSectionReader(reader, int64(offset), int64(piece.BlockSize+MaxCarBlockPrefixSize))
 			readCid, data, err := util.ReadNode(bufio.NewReader(readerAt))
 			if err != nil {
-				return nil, fmt.Errorf("reading data for block %s from reader for piece %s: %w", c, piece.PieceCidV2, err)
+				return nil, fmt.Errorf("reading data for block %s from reader for piece %s: %w", c, piece.PieceCid, err)
 			}
 			if !bytes.Equal(readCid.Hash(), c.Hash()) {
-				return nil, fmt.Errorf("read block %s from reader for piece %s, but expected block %s", readCid, piece.PieceCidV2, c)
+				return nil, fmt.Errorf("read block %s from reader for piece %s, but expected block %s", readCid, piece.PieceCid, c)
 			}
 			return data, nil
 		}()

market/retrieval/remoteblockstore/remoteblockstore.go

@@ -101,6 +101,10 @@ func (I *IPNITask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done b
 		return false, xerrors.Errorf("getting ipni task params: %w", err)
 	}
 
+	if len(tasks) == 0 {
+		return true, nil
+	}
+
 	if len(tasks) != 1 {
 		return false, xerrors.Errorf("expected 1 ipni task params, got %d", len(tasks))
 	}
@@ -511,7 +515,7 @@ func (I *IPNITask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskE
 	type task struct {
 		TaskID    harmonytask.TaskID `db:"task_id"`
 		ID        string             `db:"id"`
-		StorageID string             `db:"storage_id"`
+		StorageID sql.NullString     `db:"storage_id"`
 		IsRm      bool               `db:"is_rm"`
 	}
 
@@ -563,8 +567,8 @@ func (I *IPNITask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskE
 		var mk12Tasks []task
 		err := I.db.Select(ctx, &mk12Tasks, `
 			SELECT dp.task_id, dp.id, l.storage_id FROM ipni_task dp
-				INNER JOIN sector_location l ON dp.sp_id = l.miner_id AND dp.sector = l.sector_num
-				WHERE dp.task_id = ANY ($1) AND l.sector_filetype = 1`, indIDs)
+				LEFT JOIN sector_location l ON dp.sp_id = l.miner_id AND dp.sector = l.sector_num
+				WHERE dp.task_id = ANY ($1) AND (l.sector_filetype IS NULL OR l.sector_filetype = 1)`, indIDs)
 		if err != nil {
 			return nil, xerrors.Errorf("getting storage details: %w", err)
 		}
@@ -608,13 +612,29 @@ func (I *IPNITask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskE
 			continue
 		}
 
+		acceptables[t.TaskID] = false // note the task was found
+
+		if !t.StorageID.Valid {
+			// no unsealed copy
+			return &t.TaskID, nil
+		}
+
 		for _, l := range ls {
-			if string(l.ID) == t.StorageID {
+			if string(l.ID) == t.StorageID.String {
 				return &t.TaskID, nil
 			}
 		}
 	}
 
+	// special case for orphan tasks which are created for non-announced pieces
+	for taskID, notAccepted := range acceptables {
+		if !notAccepted {
+			continue
+		}
+
+		return &taskID, nil
+	}
+
 	return nil, nil
 }
 

tasks/indexing/task_ipni.go

@@ -83,6 +83,10 @@ func (P *PDPIPNITask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (don
 		return false, xerrors.Errorf("getting ipni task params: %w", err)
 	}
 
+	if len(tasks) == 0 {
+		return true, nil
+	}
+
 	if len(tasks) != 1 {
 		return false, xerrors.Errorf("expected 1 ipni task params, got %d", len(tasks))
 	}