feat(pdp): add SP-to-SP piece pull endpoint (#864)

Implement POST /pdp/piece/pull for fetching pieces from other SPs.
Background task handles download, CommP verification, and retry with
failure tracking. Consolidates PieceCID parsing into shared utilities.

Closes: #828

Author: rvagg

cmd/curio/tasks/tasks.go

@@ -305,12 +305,13 @@ 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(ctx, db)
+			pdpPullPieceTask := pdp.NewPDPPullPieceTask(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)
 			payTask := pay.NewSettleTask(db, must.One(dependencies.EthClient.Val()), senderEth)
-			activeTasks = append(activeTasks, pdpProveTask, pdpNotifTask, pdpNextProvingPeriodTask, pdpInitProvingPeriodTask, pdpIndexingTask, pdpIpniTask, pdpTerminate, pdpDelete, payTask)
+			activeTasks = append(activeTasks, pdpProveTask, pdpNotifTask, pdpPullPieceTask, pdpNextProvingPeriodTask, pdpInitProvingPeriodTask, pdpIndexingTask, pdpIpniTask, pdpTerminate, pdpDelete, payTask)
 		}
 
 		indexingTask := indexing.NewIndexingTask(db, sc, iStore, pp, cfg, idxMax)

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

@@ -0,0 +1,30 @@
+-- PDP piece pull tables for SP-to-SP transfer
+--
+-- Provides idempotency and piece tracking for pull requests.
+-- Status is derived dynamically from parked_pieces, not stored here.
+CREATE TABLE pdp_piece_pulls (
+    id BIGSERIAL PRIMARY KEY,
+    service TEXT NOT NULL REFERENCES pdp_services(service_label) ON DELETE CASCADE,
+    extra_data_hash BYTEA NOT NULL,  -- sha256(extraData) for idempotency
+    data_set_id BIGINT NOT NULL DEFAULT 0,  -- 0 = create new dataset
+    record_keeper TEXT NOT NULL DEFAULT '',  -- required when data_set_id is 0
+    created_at TIMESTAMPTZ DEFAULT NOW(),
+
+    UNIQUE(service, extra_data_hash, data_set_id, record_keeper)
+);
+
+-- Tracks individual pieces within a pull request
+CREATE TABLE pdp_piece_pull_items (
+    fetch_id BIGINT NOT NULL REFERENCES pdp_piece_pulls(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,  -- pull task
+    failed BOOLEAN NOT NULL DEFAULT FALSE,  -- true if piece permanently failed
+    fail_reason TEXT,               -- error message when failed
+
+    PRIMARY KEY (fetch_id, piece_cid)
+);
+
+-- Index for cleanup queries
+CREATE INDEX idx_pdp_piece_pulls_created_at ON pdp_piece_pulls(created_at);

pdp/handlers.go

@@ -18,12 +18,8 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/go-chi/chi/v5"
-	"github.com/ipfs/go-cid"
-	"github.com/multiformats/go-multicodec"
 	"github.com/yugabyte/pgx/v5"
 
-	commcid "github.com/filecoin-project/go-fil-commcid"
-
 	"github.com/filecoin-project/curio/harmony/harmonydb"
 	"github.com/filecoin-project/curio/lib/paths"
 	"github.com/filecoin-project/curio/pdp/contract"
@@ -56,6 +52,8 @@ type PDPService struct {
 	sender    *message.SenderETH
 	ethClient *ethclient.Client
 	filClient PDPServiceNodeApi
+
+	pullHandler *PullHandler
 }
 
 type PDPServiceNodeApi interface {
@@ -64,14 +62,20 @@ type PDPServiceNodeApi interface {
 
 // NewPDPService creates a new instance of PDPService with the provided stores
 func NewPDPService(ctx context.Context, db *harmonydb.DB, stor paths.StashStore, ec *ethclient.Client, fc PDPServiceNodeApi, sn *message.SenderETH) *PDPService {
+	auth := &NullAuth{}
+	pullStore := NewDBPullStore(db)
+	pullValidator := NewEthCallValidator(ec, db)
+
 	p := &PDPService{
-		Auth:    &NullAuth{},
+		Auth:    auth,
 		db:      db,
 		storage: stor,
 
 		sender:    sn,
 		ethClient: ec,
 		filClient: fc,
+
+		pullHandler: NewPullHandler(auth, pullStore, pullValidator),
 	}
 
 	go p.cleanup(ctx)
@@ -146,6 +150,9 @@ func Routes(r *chi.Mux, p *PDPService) {
 
 	// POST /pdp/piece/uploads/{uploadUUID}
 	r.Post(path.Join(PDPRoutePath, "/piece/uploads/{uploadUUID}"), p.handleFinalizeStreamingUpload)
+
+	// POST /pdp/piece/pull - Pull pieces from other SPs
+	r.Post(path.Join(PDPRoutePath, "/piece/pull"), p.pullHandler.HandlePull)
 }
 
 // Handler functions
@@ -181,12 +188,12 @@ func (p *PDPService) handleGetPieceStatus(w http.ResponseWriter, r *http.Request
 	}
 
 	// Convert to v1 format (database stores v1)
-	pieceCidV1, err := asPieceCIDv1(pieceCidStr)
+	info, err := ParsePieceCid(pieceCidStr)
 	if err != nil {
 		http.Error(w, "Invalid pieceCid format: "+err.Error(), http.StatusBadRequest)
 		return
 	}
-	pieceCidV1Str := pieceCidV1.String()
+	pieceCidV1Str := info.CidV1.String()
 
 	// Query status from database
 	var result struct {
@@ -256,7 +263,7 @@ func (p *PDPService) handleGetPieceStatus(w http.ResponseWriter, r *http.Request
 	}
 
 	// Convert authoritative PieceCID back from v1 to v2 for external API
-	pieceCidV2, _, err := asPieceCIDv2(result.PieceCID, result.PieceRawSize)
+	pieceInfo, err := PieceCidV2FromV1Str(result.PieceCID, result.PieceRawSize)
 	if err != nil {
 		http.Error(w, "Failed to convert PieceCID to v2: "+err.Error(), http.StatusInternalServerError)
 		return
@@ -271,7 +278,7 @@ func (p *PDPService) handleGetPieceStatus(w http.ResponseWriter, r *http.Request
 		Retrieved   bool       `json:"retrieved"`
 		RetrievedAt *time.Time `json:"retrievedAt,omitempty"`
 	}{
-		PieceCID:    pieceCidV2.String(),
+		PieceCID:    pieceInfo.CidV2.String(),
 		Status:      result.Status,
 		Indexed:     result.Indexed,
 		Advertised:  result.Advertised,
@@ -573,25 +580,25 @@ func (p *PDPService) handleGetDataSet(w http.ResponseWriter, r *http.Request) {
 		pcv2Str, exists := aggregatePieceCIDs[piece.PieceID]
 		if !exists {
 			aggregateRawSize := pieceRawSizes[piece.PieceID]
-			pcv2, _, err := asPieceCIDv2(piece.PieceCid, aggregateRawSize)
+			pcInfo, err := PieceCidV2FromV1Str(piece.PieceCid, aggregateRawSize)
 			if err != nil {
 				http.Error(w, "Invalid PieceCID: "+err.Error(), http.StatusBadRequest)
 				return
 			}
-			pcv2Str = pcv2.String()
+			pcv2Str = pcInfo.CidV2.String()
 			aggregatePieceCIDs[piece.PieceID] = pcv2Str
 		}
 
 		// Use the raw size for the sub piece
-		spcv2, _, err := asPieceCIDv2(piece.SubPieceCID, piece.SubPieceRawSize)
+		spcInfo, err := PieceCidV2FromV1Str(piece.SubPieceCID, piece.SubPieceRawSize)
 		if err != nil {
 			http.Error(w, "Invalid SubPieceCID: "+err.Error(), http.StatusBadRequest)
 			return
 		}
 		response.Pieces = append(response.Pieces, PieceEntry{
 			PieceID:        piece.PieceID,
 			PieceCID:       pcv2Str,
-			SubPieceCID:    spcv2.String(),
+			SubPieceCID:    spcInfo.CidV2.String(),
 			SubPieceOffset: piece.SubPieceOffset,
 		})
 	}
@@ -1095,48 +1102,6 @@ func (p *PDPService) handleGetDataSetPiece(w http.ResponseWriter, r *http.Reques
 	}
 }
 
-func asPieceCIDv1(cidStr string) (cid.Cid, error) {
-	pieceCid, err := cid.Decode(cidStr)
-	if err != nil {
-		return cid.Undef, fmt.Errorf("failed to decode PieceCID: %w", err)
-	}
-	if pieceCid.Prefix().MhType == uint64(multicodec.Fr32Sha256Trunc254Padbintree) {
-		c1, _, err := commcid.PieceCidV1FromV2(pieceCid)
-		return c1, err
-	}
-	return pieceCid, nil
-}
-
-// asPieceCIDv2 converts a string to a PieceCIDv2. Where the input is expected to be a PieceCIDv1,
-// a size argument is required. Where it's expected to be a v2, the size argument is ignored. The
-// size either derived from the v2 or from the size argument in the case of a v1 is returned.
-func asPieceCIDv2(cidStr string, size uint64) (cid.Cid, uint64, error) {
-	pieceCid, err := cid.Decode(cidStr)
-	if err != nil {
-		return cid.Undef, 0, fmt.Errorf("failed to decode PieceCid: %w", err)
-	}
-	switch pieceCid.Prefix().MhType {
-	case uint64(multicodec.Sha2_256Trunc254Padded):
-		if size == 0 {
-			return cid.Undef, 0, fmt.Errorf("size must be provided for PieceCIDv1")
-		}
-		c, err := commcid.PieceCidV2FromV1(pieceCid, size)
-		if err != nil {
-			return cid.Undef, 0, err
-		}
-		return c, size, nil
-	case uint64(multicodec.Fr32Sha256Trunc254Padbintree):
-		// get the size from the CID, not the argument
-		_, size, err := commcid.PieceCidV2ToDataCommitment(pieceCid)
-		if err != nil {
-			return cid.Undef, 0, fmt.Errorf("failed to get size from Sha2_256Trunc254Padded PieceCid: %w", err)
-		}
-		return pieceCid, size, nil
-	default:
-		return cid.Undef, 0, fmt.Errorf("unsupported piece CID type: %d", pieceCid.Prefix().MhType)
-	}
-}
-
 func (p *PDPService) cleanup(ctx context.Context) {
 	rm := func(ctx context.Context, db *harmonydb.DB) {
 		var RefIDs []int64
@@ -1155,6 +1120,13 @@ func (p *PDPService) cleanup(ctx context.Context) {
 				log.Errorw("failed to delete non-finalized uploads", "error", err)
 			}
 		}
+
+		// Clean up old piece fetch records (older than 5 days)
+		// CASCADE deletes pdp_piece_fetch_items automatically
+		_, err = db.Exec(ctx, `DELETE FROM pdp_piece_fetches WHERE created_at < NOW() - INTERVAL '5 days'`)
+		if err != nil {
+			log.Errorw("failed to delete old piece fetch records", "error", err)
+		}
 	}
 
 	ticker := time.NewTicker(time.Minute * 5)

pdp/handlers_add.go

@@ -69,11 +69,11 @@ func (p *PDPService) transformAddPiecesRequest(ctx context.Context, serviceLabel
 			if subPieceEntry.SubPieceCID == "" {
 				return nil, nil, nil, errors.New("subPieceCid is required for each subPiece")
 			}
-			pieceCid, err := asPieceCIDv1(subPieceEntry.SubPieceCID)
+			info, err := ParsePieceCid(subPieceEntry.SubPieceCID)
 			if err != nil {
 				return nil, nil, nil, fmt.Errorf("invalid SubPiece: %w", err)
 			}
-			pieceCidString := pieceCid.String()
+			pieceCidString := info.CidV1.String()
 
 			addPieceReq.SubPieces[i].subPieceCIDv1 = pieceCidString // save it for to query subPieceInfoMap later
 
@@ -177,14 +177,14 @@ func (p *PDPService) transformAddPiecesRequest(ctx context.Context, serviceLabel
 			}
 
 			// Compare generated PieceCid with provided PieceCid
-			providedPieceCidv1, err := asPieceCIDv1(addPieceReq.PieceCID)
+			providedInfo, err := ParsePieceCid(addPieceReq.PieceCID)
 			if err != nil {
 				return false, fmt.Errorf("invalid provided PieceCid: %v", err)
 			}
-			pieces[i].pieceCIDv1 = providedPieceCidv1.String()
+			pieces[i].pieceCIDv1 = providedInfo.CidV1.String()
 
-			if !providedPieceCidv1.Equals(generatedPieceCid) {
-				return false, fmt.Errorf("provided PieceCid does not match generated PieceCid: %s != %s", providedPieceCidv1, generatedPieceCid)
+			if !providedInfo.CidV1.Equals(generatedPieceCid) {
+				return false, fmt.Errorf("provided PieceCid does not match generated PieceCid: %s != %s", providedInfo.CidV1, generatedPieceCid)
 			}
 		}