fix streaming upload

Author: LexLuthr

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

@@ -98,13 +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,
-			PieceIO:    piecestore.New(d.Stor, d.LocalStore, d.Si),
-			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
 
@@ -179,13 +178,18 @@ 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:
 
 **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`.
 2. **PDPv0_PullPiece** downloads the piece from the external URL, computes and verifies CommP, and stores the result in `parked_pieces` + `pdp_piecerefs`.

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());

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

@@ -0,0 +1,40 @@
+/*
+  Streaming-upload timestamps represent absolute instants. Convert legacy
+  naive columns, if present, by interpreting their stored values as UTC.
+  Existing TIMESTAMPTZ columns already store absolute instants and are left
+  unchanged.
+*/
+
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1
+        FROM information_schema.columns
+        WHERE table_schema = 'public'
+          AND table_name = 'pdp_piece_streaming_uploads'
+          AND column_name = 'created_at'
+          AND data_type = 'timestamp without time zone'
+    ) THEN
+        ALTER TABLE pdp_piece_streaming_uploads
+            ALTER COLUMN created_at DROP DEFAULT;
+        ALTER TABLE pdp_piece_streaming_uploads
+            ALTER COLUMN created_at TYPE TIMESTAMPTZ
+            USING created_at AT TIME ZONE 'UTC';
+    END IF;
+
+    IF EXISTS (
+        SELECT 1
+        FROM information_schema.columns
+        WHERE table_schema = 'public'
+          AND table_name = 'pdp_piece_streaming_uploads'
+          AND column_name = 'completed_at'
+          AND data_type = 'timestamp without time zone'
+    ) THEN
+        ALTER TABLE pdp_piece_streaming_uploads
+            ALTER COLUMN completed_at TYPE TIMESTAMPTZ
+            USING completed_at AT TIME ZONE 'UTC';
+    END IF;
+END $$;
+
+ALTER TABLE pdp_piece_streaming_uploads
+    ALTER COLUMN created_at SET DEFAULT NOW();