fixup! feat(pdp): add SP-to-SP piece fetch endpoint

Author: rvagg

cmd/curio/tasks/tasks.go

@@ -305,11 +305,12 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 			pdpNextProvingPeriodTask := pdp.NewNextProvingPeriodTask(db, must.One(dependencies.EthClient.Val()), dependencies.Chain, chainSched, es)
 			pdpInitProvingPeriodTask := pdp.NewInitProvingPeriodTask(db, must.One(dependencies.EthClient.Val()), dependencies.Chain, chainSched, es)
 			pdpNotifTask := pdp.NewPDPNotifyTask(db)
+			pdpFetchPieceTask := pdp.NewPDPFetchPieceTask(ctx, db, lstor, 4)
 			pdpIndexingTask := indexing.NewPDPIndexingTask(db, iStore, dependencies.CachedPieceReader, cfg, idxMax)
 			pdpIpniTask := indexing.NewPDPIPNITask(db, sc, dependencies.CachedPieceReader, cfg, idxMax)
 			pdpTerminate := pdp.NewTerminateServiceTask(db, must.One(dependencies.EthClient.Val()), senderEth)
 			pdpDelete := pdp.NewDeleteDataSetTask(db, must.One(dependencies.EthClient.Val()), senderEth)
-			activeTasks = append(activeTasks, pdpNotifTask, pdpProveTask, pdpNextProvingPeriodTask, pdpInitProvingPeriodTask, pdpIndexingTask, pdpIpniTask, pdpTerminate, pdpDelete)
+			activeTasks = append(activeTasks, pdpNotifTask, pdpFetchPieceTask, pdpProveTask, pdpNextProvingPeriodTask, pdpInitProvingPeriodTask, pdpIndexingTask, pdpIpniTask, pdpTerminate, pdpDelete)
 		}
 
 		indexingTask := indexing.NewIndexingTask(db, sc, iStore, pp, cfg, idxMax)

harmony/harmonydb/sql/20260109-pdpv0-fetch.sql

@@ -18,6 +18,8 @@ CREATE TABLE pdp_piece_fetch_items (
     fetch_id BIGINT NOT NULL REFERENCES pdp_piece_fetches(id) ON DELETE CASCADE,
     piece_cid TEXT NOT NULL,        -- PieceCIDv1 (for joins with parked_pieces)
     piece_raw_size BIGINT NOT NULL, -- raw size to reconstruct PieceCIDv2 for API
+    source_url TEXT NOT NULL,       -- external SP URL to fetch from
+    task_id BIGINT REFERENCES harmony_task(id) ON DELETE SET NULL,  -- fetch task
     failed BOOLEAN NOT NULL DEFAULT FALSE,  -- true if piece permanently failed
     fail_reason TEXT,               -- error message when failed
 

lib/ffi/piece_funcs.go

@@ -8,10 +8,6 @@ import (
 
 	"golang.org/x/xerrors"
 
-	commcid "github.com/filecoin-project/go-fil-commcid"
-	commp "github.com/filecoin-project/go-fil-commp-hashhash"
-	"github.com/filecoin-project/go-state-types/abi"
-
 	"github.com/filecoin-project/curio/harmony/harmonytask"
 	storiface "github.com/filecoin-project/curio/lib/storiface"
 )
@@ -87,82 +83,3 @@ func (sb *SealCalls) PieceReader(ctx context.Context, id storiface.PieceNumber)
 func (sb *SealCalls) RemovePiece(ctx context.Context, id storiface.PieceNumber) error {
 	return sb.Sectors.storage.Remove(ctx, id.Ref().ID, storiface.FTPiece, true, nil)
 }
-
-func (sb *SealCalls) WriteUploadPiece(ctx context.Context, pieceID storiface.PieceNumber, size int64, data io.Reader, storageType storiface.PathType, verifySize bool) (abi.PieceInfo, uint64, error) {
-	// Use storageType in AcquireSector
-	paths, pathIDs, done, err := sb.Sectors.AcquireSector(ctx, nil, pieceID.Ref(), storiface.FTNone, storiface.FTPiece, storageType)
-	if err != nil {
-		return abi.PieceInfo{}, 0, err
-	}
-	skipDeclare := storiface.FTPiece
-
-	defer func() {
-		done(skipDeclare)
-	}()
-
-	dest := paths.Piece
-	tempDest := dest + storiface.TempSuffix
-
-	destFile, err := os.OpenFile(tempDest, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
-	if err != nil {
-		return abi.PieceInfo{}, 0, xerrors.Errorf("creating temp piece file '%s': %w", tempDest, err)
-	}
-
-	removeTemp := true
-	defer func() {
-		if removeTemp {
-			rerr := os.Remove(tempDest)
-			if rerr != nil {
-				log.Errorf("removing temp file: %+v", rerr)
-			}
-		}
-	}()
-
-	copyStart := time.Now()
-
-	wr := new(commp.Calc)
-	writers := io.MultiWriter(wr, destFile)
-
-	n, err := io.CopyBuffer(writers, io.LimitReader(data, size), make([]byte, 8<<20))
-	if err != nil {
-		_ = destFile.Close()
-		return abi.PieceInfo{}, 0, xerrors.Errorf("copying piece data: %w", err)
-	}
-
-	if err := destFile.Close(); err != nil {
-		return abi.PieceInfo{}, 0, xerrors.Errorf("closing temp piece file: %w", err)
-	}
-
-	if verifySize && n != size {
-		return abi.PieceInfo{}, 0, xerrors.Errorf("short write: %d", n)
-	}
-
-	digest, pieceSize, err := wr.Digest()
-	if err != nil {
-		return abi.PieceInfo{}, 0, xerrors.Errorf("computing piece digest: %w", err)
-	}
-
-	pcid, err := commcid.DataCommitmentV1ToCID(digest)
-	if err != nil {
-		return abi.PieceInfo{}, 0, xerrors.Errorf("computing piece CID: %w", err)
-	}
-	psize := abi.PaddedPieceSize(pieceSize)
-
-	copyEnd := time.Now()
-
-	log.Infow("wrote piece", "piece", pieceID, "size", n, "duration", copyEnd.Sub(copyStart), "dest", dest, "MiB/s", float64(size)/(1<<20)/copyEnd.Sub(copyStart).Seconds())
-
-	if err := os.Rename(tempDest, dest); err != nil {
-		return abi.PieceInfo{}, 0, xerrors.Errorf("rename temp piece to dest %s -> %s: %w", tempDest, dest, err)
-	}
-
-	skipDeclare = storiface.FTNone
-
-	removeTemp = false
-
-	if err := sb.ensureOneCopy(ctx, pieceID.Ref().ID, pathIDs, storiface.FTPiece); err != nil {
-		return abi.PieceInfo{}, 0, xerrors.Errorf("ensure one copy: %w", err)
-	}
-
-	return abi.PieceInfo{PieceCID: pcid, Size: psize}, uint64(n), nil
-}