fix: upload-to-first-download latency

alertmanager/alerts.go

@@ -173,7 +173,6 @@ var pdpTasks = []string{
 	tasknames.PDPDelDataSet,
 	tasknames.PDPInitPP,
 	tasknames.PDPProvingPeriod,
-	tasknames.PDPNotify,
 	tasknames.PDPCommP,
 	tasknames.PDPSaveCache,
 	tasknames.AggregatePDPDeal,
@@ -183,7 +182,6 @@ var pdpTasks = []string{
 	tasknames.PDPv0_SaveCache,
 	tasknames.PDPv0_InitPP,
 	tasknames.PDPv0_ProvPeriod,
-	tasknames.PDPv0_Notify,
 }
 
 // taskFailureCheckWith is the parameterized core shared by taskFailureCheck

cmd/pdptool/main.go

@@ -12,7 +12,6 @@ import (
 	"encoding/pem"
 	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"os"
 	"strconv"
@@ -375,50 +374,7 @@ var piecePrepareCmd = &cli.Command{
 	},
 }
 
-func startLocalNotifyServer() (string, chan struct{}, error) {
-	var notifyReceived chan struct{}
-	var server *http.Server
-	var ln net.Listener
-
-	notifyReceived = make(chan struct{})
-	var err error
-	ln, err = net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		return "", nil, fmt.Errorf("failed to start local HTTP server: %v", err)
-	}
-	serverAddr := fmt.Sprintf("http://%s/notify", ln.Addr().String())
-
-	mux := http.NewServeMux()
-	mux.HandleFunc("/notify", func(w http.ResponseWriter, r *http.Request) {
-		fmt.Println("Received notification from server.")
-		b, err := io.ReadAll(r.Body)
-		if err != nil {
-			fmt.Printf("Failed to read notification body: %v\n", err)
-			w.WriteHeader(http.StatusInternalServerError)
-			return
-		}
-		fmt.Printf("Notification body: %s\n", string(b))
-		w.WriteHeader(http.StatusOK)
-		// Signal that notification was received
-		close(notifyReceived)
-	})
-
-	server = &http.Server{Handler: mux}
-
-	go func() {
-		if err := server.Serve(ln); err != nil && err != http.ErrServerClosed {
-			fmt.Printf("HTTP server error: %v\n", err)
-		}
-	}()
-
-	defer func() {
-		_ = server.Close()
-		_ = ln.Close()
-	}()
-	return serverAddr, notifyReceived, nil
-}
-
-func uploadOnePiece(client *http.Client, serviceURL string, reqBody []byte, jwtToken string, r io.ReadSeeker, pieceSize int64, localNotifWait bool, notifyReceived chan struct{}, verbose bool) error {
+func uploadOnePiece(client *http.Client, serviceURL string, reqBody []byte, jwtToken string, r io.ReadSeeker, pieceSize int64, verbose bool) error {
 	req, err := http.NewRequest("POST", serviceURL+"/pdp/piece", bytes.NewReader(reqBody))
 	if err != nil {
 		return fmt.Errorf("failed to create request: %v", err)
@@ -491,13 +447,6 @@ func uploadOnePiece(client *http.Client, serviceURL string, reqBody []byte, jwtT
 			body, _ := io.ReadAll(uploadResp.Body)
 			return fmt.Errorf("upload failed with status code %d: %s", uploadResp.StatusCode, string(body))
 		}
-		if localNotifWait {
-			if verbose {
-				fmt.Println("Waiting for server notification...")
-			}
-			<-notifyReceived
-		}
-
 		return nil
 	default:
 		body, _ := io.ReadAll(resp.Body)
@@ -523,20 +472,11 @@ var pieceUploadCmd = &cli.Command{
 			Name:  "service-name",
 			Usage: "Service Name to include in the JWT token (used if --jwt-token is not provided)",
 		},
-		&cli.StringFlag{
-			Name:     "notify-url",
-			Usage:    "Notification URL",
-			Required: false,
-		},
 		&cli.StringFlag{
 			Name:  "hash-type",
 			Usage: "Hash type to use for verification (sha256 or commp)",
 			Value: "sha256",
 		},
-		&cli.BoolFlag{
-			Name:  "local-notif-wait",
-			Usage: "Wait for server notification by spawning a temporary local HTTP server",
-		},
 	},
 	Action: func(cctx *cli.Context) error {
 		inputFile := cctx.Args().Get(0)
@@ -546,10 +486,8 @@ var pieceUploadCmd = &cli.Command{
 
 		serviceURL := cctx.String("service-url")
 		jwtToken := cctx.String("jwt-token")
-		notifyURL := cctx.String("notify-url")
 		serviceName := cctx.String("service-name")
 		hashType := cctx.String("hash-type")
-		localNotifWait := cctx.Bool("local-notif-wait")
 
 		if jwtToken == "" {
 			if serviceName == "" {
@@ -566,20 +504,8 @@ var pieceUploadCmd = &cli.Command{
 			return fmt.Errorf("invalid hash type: %s", hashType)
 		}
 
-		if localNotifWait && notifyURL != "" {
-			return fmt.Errorf("cannot specify both --notify-url and --local-notif-wait")
-		}
-
-		var notifyReceived chan struct{}
 		var err error
 
-		if localNotifWait {
-			notifyURL, notifyReceived, err = startLocalNotifyServer()
-			if err != nil {
-				return fmt.Errorf("failed to start local HTTP server: %v", err)
-			}
-		}
-
 		// Open input file
 		file, err := os.Open(inputFile)
 		if err != nil {
@@ -626,15 +552,12 @@ var pieceUploadCmd = &cli.Command{
 			return fmt.Errorf("unsupported hash type: %s", hashType)
 		}
 
-		if notifyURL != "" {
-			reqData["notify"] = notifyURL
-		}
 		reqBody, err = json.Marshal(reqData)
 		if err != nil {
 			return fmt.Errorf("failed to marshal request data: %v", err)
 		}
 		client := &http.Client{}
-		if err := uploadOnePiece(client, serviceURL, reqBody, jwtToken, file, pieceSize, localNotifWait, notifyReceived, true); err != nil {
+		if err := uploadOnePiece(client, serviceURL, reqBody, jwtToken, file, pieceSize, true); err != nil {
 			return fmt.Errorf("failed to upload piece: %v", err)
 		}
 
@@ -662,20 +585,11 @@ var uploadFileCmd = &cli.Command{
 			Name:  "service-name",
 			Usage: "Service Name to include in the JWT token (used if --jwt-token is not provided)",
 		},
-		&cli.StringFlag{
-			Name:     "notify-url",
-			Usage:    "Notification URL",
-			Required: false,
-		},
 		&cli.StringFlag{
 			Name:  "hash-type",
 			Usage: "Hash type to use for verification (sha256 or commp)",
 			Value: "sha256",
 		},
-		&cli.BoolFlag{
-			Name:  "local-notif-wait",
-			Usage: "Wait for server notification by spawning a temporary local HTTP server",
-		},
 		&cli.BoolFlag{
 			Name:  "verbose",
 			Usage: "Verbose output",
@@ -702,8 +616,6 @@ var uploadFileCmd = &cli.Command{
 		jwtToken := cctx.String("jwt-token")
 		serviceName := cctx.String("service-name")
 		hashType := cctx.String("hash-type")
-		localNotifWait := cctx.Bool("local-notif-wait")
-		notifyURL := cctx.String("notify-url")
 		verbose := cctx.Bool("verbose")
 		dryRun := cctx.Bool("dry-run")
 		chunkFileName := cctx.String("chunk-file")
@@ -754,15 +666,6 @@ var uploadFileCmd = &cli.Command{
 			bar = progressbar.NewOptions(int(fileSize/chunkSize), progressbar.OptionSetDescription("Uploading..."))
 		}
 
-		// Setup local server if needed
-		var notifyReceived chan struct{}
-		if localNotifWait {
-			notifyURL, notifyReceived, err = startLocalNotifyServer()
-			if err != nil {
-				return fmt.Errorf("failed to start local HTTP server: %v", err)
-			}
-		}
-
 		// group piece aggregations for tracking as onchain pieces into sector size chunks
 		type pieceSetInfo struct {
 			pieces      []abi.PieceInfo
@@ -820,16 +723,13 @@ var uploadFileCmd = &cli.Command{
 					return fmt.Errorf("unsupported hash type: %s", hashType)
 				}
 
-				if notifyURL != "" {
-					reqData["notify"] = notifyURL
-				}
 				reqBody, err = json.Marshal(reqData)
 				if err != nil {
 					return fmt.Errorf("failed to marshal request data: %v", err)
 				}
 
 				// Upload the piece
-				err = uploadOnePiece(client, serviceURL, reqBody, jwtToken, chunkReader, int64(n), localNotifWait, notifyReceived, verbose)
+				err = uploadOnePiece(client, serviceURL, reqBody, jwtToken, chunkReader, int64(n), verbose)
 				if err != nil {
 					return fmt.Errorf("failed to upload piece: %v", err)
 				}
@@ -1530,20 +1430,11 @@ var streamingPieceUploadCmd = &cli.Command{
 			Name:  "service-name",
 			Usage: "Service Name to include in the JWT token (used if --jwt-token is not provided)",
 		},
-		&cli.StringFlag{
-			Name:     "notify-url",
-			Usage:    "Notification URL",
-			Required: false,
-		},
 		&cli.StringFlag{
 			Name:  "hash-type",
 			Usage: "Hash type to use for verification (sha256 or commp)",
 			Value: "commp",
 		},
-		&cli.BoolFlag{
-			Name:  "local-notif-wait",
-			Usage: "Wait for server notification by spawning a temporary local HTTP server",
-		},
 	},
 	Action: func(cctx *cli.Context) error {
 		inputFile := cctx.Args().Get(0)
@@ -1553,10 +1444,8 @@ var streamingPieceUploadCmd = &cli.Command{
 
 		serviceURL := cctx.String("service-url")
 		jwtToken := cctx.String("jwt-token")
-		notifyURL := cctx.String("notify-url")
 		serviceName := cctx.String("service-name")
 		hashType := cctx.String("hash-type")
-		localNotifWait := cctx.Bool("local-notif-wait")
 
 		if jwtToken == "" {
 			if serviceName == "" {
@@ -1573,20 +1462,8 @@ var streamingPieceUploadCmd = &cli.Command{
 			return fmt.Errorf("invalid hash type: %s", hashType)
 		}
 
-		if localNotifWait && notifyURL != "" {
-			return fmt.Errorf("cannot specify both --notify-url and --local-notif-wait")
-		}
-
-		var notifyReceived chan struct{}
 		var err error
 
-		if localNotifWait {
-			notifyURL, notifyReceived, err = startLocalNotifyServer()
-			if err != nil {
-				return fmt.Errorf("failed to start local HTTP server: %v", err)
-			}
-		}
-
 		// Open the input file
 		file, err := os.Open(inputFile)
 		if err != nil {
@@ -1715,17 +1592,12 @@ var streamingPieceUploadCmd = &cli.Command{
 
 		type finalize struct {
 			PieceCID string `json:"pieceCid"`
-			Notify   string `json:"notify,omitempty"`
 		}
 
 		bd := finalize{
 			PieceCID: pcid2.String(),
 		}
 
-		if notifyURL != "" {
-			bd.Notify = notifyURL
-		}
-
 		bodyBytes, err := json.Marshal(bd)
 		if err != nil {
 			return fmt.Errorf("failed to marshal finalize request body: %v", err)
@@ -1758,10 +1630,6 @@ var streamingPieceUploadCmd = &cli.Command{
 
 		fmt.Printf("Piece CID: %s\n", pcid2.String())
 		fmt.Println("Piece uploaded successfully.")
-		if localNotifWait {
-			fmt.Println("Waiting for server notification...")
-			<-notifyReceived
-		}
 		return nil
 	},
 }

cuhttp/server.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/filecoin-project/curio/cuhttp/servicedeps"
 	"github.com/filecoin-project/curio/deps"
+	"github.com/filecoin-project/curio/lib/piecestore"
 	mhttp "github.com/filecoin-project/curio/market/http"
 	"github.com/filecoin-project/curio/market/libp2p"
 	"github.com/filecoin-project/curio/pdp"
@@ -97,12 +98,12 @@ func attachRouters(ctx context.Context, r *chi.Mux, d *deps.Deps, sd *ServiceDep
 
 	if sd.EthSender != nil {
 		if err := pdp.MountRoutes(ctx, r, pdp.MountDeps{
-			DB:         d.DB,
-			LocalStore: d.LocalStore,
-			EthClient:  must.One(d.EthClient.Get()),
-			Chain:      d.Chain,
-			EthSender:  sd.EthSender,
-			AlertTask:  sd.AlertTask,
+			DB:        d.DB,
+			PieceIO:   piecestore.New(d.Stor, d.LocalStore, d.Si),
+			EthClient: must.One(d.EthClient.Get()),
+			Chain:     d.Chain,
+			EthSender: sd.EthSender,
+			AlertTask: sd.AlertTask,
 		}, ipp); err != nil {
 			return nil, err
 		}

documentation/en/experimental-features/PDPCURIOSPEC.md

@@ -20,7 +20,6 @@ TODO: the pdp datasets table etc
 ## Managing Pieces
 TODO: add, upload, delete, pull, info
 TODO: the pdp pieceref table etc
-TODO: "finalization" notify task
 
 # Storage
 
@@ -143,7 +142,7 @@ Harmony tasks are created through three trigger mechanisms:
 
 - **Chain handlers** — Registered via `chainsched.AddHandler`, these callbacks fire on every chain head change. They inspect on-chain state (e.g. transaction receipts, epoch thresholds) and call `AddTask` to insert work into the harmony_task queue when conditions are met. The proving-cycle tasks (InitPP, ProvPeriod, Prove) use this mechanism so they respond immediately to new tipsets.
 - **IAmBored** — An optional callback in `TaskTypeDetails` that the task engine invokes when a machine has spare capacity and no queued work exists for that task type. A `passcall.Every(duration, ...)` wrapper rate-limits invocations. Tasks like TerminateFWSS (1 min), DeleteDataSet (1 hour), and Settle (12 hours) use IAmBored because they generate work opportunistically rather than in response to chain events.
-- **Polling** — Some tasks use a dedicated poller goroutine that periodically queries the database for pending work and calls `AddTask`. The Notify (2s) and PullPiece (10s) tasks use this pattern because their triggers are purely database-driven (new uploads or pull requests) with no chain dependency.
+- **Polling** — Some tasks use a dedicated poller goroutine that periodically queries the database for pending work and calls `AddTask`. PullPiece uses this pattern because its trigger is a database-backed pull request rather than a chain event.
 
 All three mechanisms funnel through `harmonytask.AddTask()`, which atomically inserts a task record. The main poller loop (every 3s) then discovers unowned tasks and assigns them to machines with available resources.
 
@@ -154,7 +153,6 @@ All three mechanisms funnel through `harmonytask.AddTask()`, which atomically in
 | `PDPv0_InitPP` | `InitProvingPeriodTask` | `tasks/pdpv0/task_init_pp.go` | Chain handler |
 | `PDPv0_ProvPeriod` | `NextProvingPeriodTask` | `tasks/pdpv0/task_next_pp.go` | Chain handler |
 | `PDPv0_Prove` | `ProveTask` | `tasks/pdpv0/task_prove.go` | Chain handler |
-| `PDPv0_Notify` | `PDPNotifyTask` | `tasks/pdpv0/notify_task.go` | Polling (2s) |
 | `PDPv0_PullPiece` | `PDPPullPieceTask` | `tasks/pdpv0/task_pull_piece.go` | Polling (10s) |
 | `PDPv0_Indexing` | `PDPIndexingTask` | `tasks/indexing/task_pdp_v0_indexing.go` | IAmBored (3s) |
 | `PDPv0_IPNI` | `PDPIPNITask` | `tasks/indexing/task_pdp_v0_ipni.go` | IAmBored (30s) |
@@ -180,11 +178,17 @@ The sections below describe how tasks and watchers connect to form the PDP lifec
 
 ## Piece Ingestion
 
-There are two paths for getting pieces into the system:
+There are three paths for getting pieces into the system:
 
-**Direct upload path:**
-1. Client uploads piece data via HTTP. The data is written to `parked_pieces`.
-2. When the upload completes (`parked_pieces.complete = TRUE`), **PDPv0_Notify** picks it up, sends an HTTP callback to `notify_url` if configured, and moves the reference from `pdp_piece_uploads` to `pdp_piecerefs`.
+**Known-CID direct upload path:**
+1. The client posts the PieceCID. If a complete long-term copy already exists, the handler creates its `parked_piece_refs` and `pdp_piecerefs` rows immediately.
+2. Otherwise, the client PUTs the bytes to the returned upload URL. The handler claims a `parked_pieces` row, writes the request body once directly to final piece storage while computing CommP, and validates the declared size and PieceCID.
+3. After the write succeeds, one database transaction marks the parked piece complete, creates `pdp_piecerefs`, and deletes the upload row. The legacy `notify` request field is accepted for compatibility but this path does not call the URL or schedule a notify task.
+
+**Streaming upload path:**
+1. The client creates an upload session and PUTs bytes without declaring a PieceCID. The handler claims a provisional `parked_pieces` identity for that session, then streams the request once directly into final piece storage while calculating CommP and the raw size.
+2. After the PieceCID is known, one transaction promotes the provisional row to the calculated identity. If a matching complete piece already exists, the session is pointed at that piece and the unreferenced provisional copy is left for normal parked-piece cleanup.
+3. The client finalizes with the calculated PieceCID. The handler validates it, creates `pdp_piecerefs`, and deletes the streaming session in one transaction. It does not use scratch space, create an intermediate `pdp_piece_uploads` row, or schedule a notify task.
 
 **Pull path:**
 1. Client submits a pull request via HTTP, creating a row in `pdp_piece_pull_items`.

harmony/harmonydb/downgrade/20260820-pdp-streaming-timestamptz.sql

@@ -0,0 +1,4 @@
+-- The canonical pre-migration schema already used TIMESTAMPTZ for both
+-- columns, so retain their types and restore only the previous default.
+ALTER TABLE pdp_piece_streaming_uploads
+    ALTER COLUMN created_at SET DEFAULT TIMEZONE('UTC', NOW());