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

Author: rvagg

cmd/curio/tasks/tasks.go

@@ -305,13 +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)
-			pdpFetchPieceTask := pdp.NewPDPFetchPieceTask(ctx, db, lstor, 4)
+			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, pdpFetchPieceTask, 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)

…y/harmonydb/sql/20260109-pdpv0-fetch.sql …ny/harmonydb/sql/20260109-pdpv0-pull.sqlharmony/harmonydb/sql/20260109-pdpv0-fetch.sql renamed to harmony/harmonydb/sql/20260109-pdpv0-pull.sql harmony/harmonydb/sql/20260109-pdpv0-fetch.sql renamed to harmony/harmonydb/sql/20260109-pdpv0-pull.sql

@@ -1,8 +1,8 @@
--- PDP piece fetch tables for SP-to-SP transfer
+-- PDP piece pull tables for SP-to-SP transfer
 --
--- Provides idempotency and piece tracking for fetch requests.
+-- Provides idempotency and piece tracking for pull requests.
 -- Status is derived dynamically from parked_pieces, not stored here.
-CREATE TABLE pdp_piece_fetches (
+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
@@ -13,18 +13,18 @@ CREATE TABLE pdp_piece_fetches (
     UNIQUE(service, extra_data_hash, data_set_id, record_keeper)
 );
 
--- Tracks individual pieces within a fetch request
-CREATE TABLE pdp_piece_fetch_items (
-    fetch_id BIGINT NOT NULL REFERENCES pdp_piece_fetches(id) ON DELETE CASCADE,
+-- 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,  -- fetch task
+    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_fetches_created_at ON pdp_piece_fetches(created_at);
+CREATE INDEX idx_pdp_piece_pulls_created_at ON pdp_piece_pulls(created_at);

pdp/handlers.go

@@ -53,7 +53,7 @@ type PDPService struct {
 	ethClient *ethclient.Client
 	filClient PDPServiceNodeApi
 
-	fetchHandler *FetchHandler
+	pullHandler *PullHandler
 }
 
 type PDPServiceNodeApi interface {
@@ -63,8 +63,8 @@ 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{}
-	fetchStore := NewDBFetchStore(db)
-	fetchValidator := NewEthCallValidator(ec, db)
+	pullStore := NewDBPullStore(db)
+	pullValidator := NewEthCallValidator(ec, db)
 
 	p := &PDPService{
 		Auth:    auth,
@@ -75,7 +75,7 @@ func NewPDPService(ctx context.Context, db *harmonydb.DB, stor paths.StashStore,
 		ethClient: ec,
 		filClient: fc,
 
-		fetchHandler: NewFetchHandler(auth, fetchStore, fetchValidator),
+		pullHandler: NewPullHandler(auth, pullStore, pullValidator),
 	}
 
 	go p.cleanup(ctx)
@@ -151,8 +151,8 @@ 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/fetch - Fetch pieces from other SPs
-	r.Post(path.Join(PDPRoutePath, "/piece/fetch"), p.fetchHandler.HandleFetch)
+	// POST /pdp/piece/pull - Pull pieces from other SPs
+	r.Post(path.Join(PDPRoutePath, "/piece/pull"), p.pullHandler.HandlePull)
 }
 
 // Handler functions