Delete pdp roots from proofset (#384)

* Core api and prove task logic changes

* Update pdptool to process deletes

* SQL ordering fix

* NoSQL just track removals on chain

* Log removals

* correct pdptool output

* correct delete api interface

* remove

* remove

* Update tasks/pdp/task_prove.go

Co-authored-by: Łukasz Magiera <magik6k@users.noreply.github.com>

* review response

* Fix up

---------

Co-authored-by: zenground0 <ZenGround0@users.noreply.github.com>
Co-authored-by: Łukasz Magiera <magik6k@users.noreply.github.com>

cmd/pdptool/main.go

@@ -48,6 +48,7 @@ func main() {
 			getProofSetCmd,       // retrieve the details of a proof set from the PDP service
 
 			addRootsCmd,
+			removeRootsCmd, // schedule roots for removal after next proof submission
 		},
 	}
 	app.Setup()
@@ -992,3 +993,76 @@ var addRootsCmd = &cli.Command{
 		return nil
 	},
 }
+
+var removeRootsCmd = &cli.Command{
+	Name:  "remove-roots",
+	Usage: "Schedule roots for removal after next proof submission",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:     "service-url",
+			Usage:    "URL of the PDP service",
+			Required: true,
+		},
+		&cli.Uint64Flag{
+			Name:     "proof-set-id",
+			Usage:    "ID of the proof set to which roots will be added",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "service-name",
+			Usage:    "Service Name to include in the JWT token",
+			Required: true,
+		},
+		&cli.Uint64Flag{
+			Name:     "root-id",
+			Usage:    "Root ID for removal",
+			Required: true,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		serviceURL := cctx.String("service-url")
+		serviceName := cctx.String("service-name")
+		proofSetID := cctx.Uint64("proof-set-id")
+		rootID := cctx.Uint64("root-id")
+
+		// Load the private key (implement `loadPrivateKey` according to your setup)
+		privKey, err := loadPrivateKey()
+		if err != nil {
+			return fmt.Errorf("failed to load private key: %v", err)
+		}
+
+		// Create the JWT token (implement `createJWTToken` according to your setup)
+		jwtToken, err := createJWTToken(serviceName, privKey)
+		if err != nil {
+			return fmt.Errorf("failed to create JWT token: %v", err)
+		}
+
+		// Construct the POST URL
+		deleteURL := fmt.Sprintf("%s/pdp/proof-sets/%d/roots/%d", serviceURL, proofSetID, rootID)
+		fmt.Printf("Delete URL: %s\n", deleteURL)
+
+		// Create the POST request
+		req, err := http.NewRequest("DELETE", deleteURL, nil)
+		if err != nil {
+			return fmt.Errorf("failed to create request: %v", err)
+		}
+		req.Header.Set("Authorization", "Bearer "+jwtToken)
+		req.Header.Set("Content-Type", "application/json")
+
+		// Send the request
+		client := &http.Client{}
+		resp, err := client.Do(req)
+		if err != nil {
+			return fmt.Errorf("failed to send request: %v", err)
+		}
+
+		// Read and display the response
+		if resp.StatusCode == http.StatusNoContent {
+			fmt.Printf("Root %d scheduled for removal from proof set ID %d.\n", rootID, proofSetID)
+		} else {
+			return fmt.Errorf("failed to add roots, status code %d", resp.StatusCode)
+		}
+
+		return nil
+	},
+}

cmd/pdptool/pdp_cleanup.sql

@@ -0,0 +1,9 @@
+-- Clear tables in correct order (respecting foreign key constraints)
+TRUNCATE TABLE curio.pdp_proof_sets CASCADE;
+TRUNCATE TABLE curio.pdp_piecerefs CASCADE;
+TRUNCATE TABLE curio.pdp_proofset_root_adds CASCADE;
+TRUNCATE TABLE curio.pdp_proofset_roots CASCADE;
+TRUNCATE TABLE curio.pdp_prove_tasks CASCADE;
+TRUNCATE TABLE curio.pdp_proofset_creates CASCADE;
+TRUNCATE TABLE curio.pdp_piece_uploads CASCADE;
+

pdp/handlers.go

@@ -883,6 +883,129 @@ func (p *PDPService) handleAddRootToProofSet(w http.ResponseWriter, r *http.Requ
 	w.WriteHeader(http.StatusCreated)
 }
 
+func (p *PDPService) handleDeleteProofSetRoot(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	// Step 1: Verify that the request is authorized using ECDSA JWT
+	serviceLabel, err := p.verifyJWTToken(r)
+	if err != nil {
+		http.Error(w, "Unauthorized: "+err.Error(), http.StatusUnauthorized)
+		return
+	}
+
+	// Step 2: Extract parameters from the URL
+	proofSetIdStr := chi.URLParam(r, "proofSetID")
+	if proofSetIdStr == "" {
+		http.Error(w, "Missing proof set ID in URL", http.StatusBadRequest)
+		return
+	}
+	rootIdStr := chi.URLParam(r, "rootID")
+	if rootIdStr == "" {
+		http.Error(w, "Missing root ID in URL", http.StatusBadRequest)
+		return
+	}
+
+	// Convert proofSetId to uint64
+	proofSetID, err := strconv.ParseUint(proofSetIdStr, 10, 64)
+	if err != nil {
+		http.Error(w, "Invalid proof set ID format", http.StatusBadRequest)
+		return
+	}
+	rootID, err := strconv.ParseUint(rootIdStr, 10, 64)
+	if err != nil {
+		http.Error(w, "Invalid root ID format", http.StatusBadRequest)
+		return
+	}
+
+	// check if the proofset belongs to the service in pdp_proof_sets
+	var proofSetService string
+	err = p.db.QueryRow(ctx, `
+			SELECT service 
+			FROM pdp_proof_sets
+			WHERE id = $1
+		`, proofSetID).Scan(&proofSetService)
+	if err != nil {
+		if err == pgx.ErrNoRows {
+			http.Error(w, "Proof set not found", http.StatusNotFound)
+			return
+		}
+		http.Error(w, "Failed to retrieve proof set: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	if proofSetService != serviceLabel {
+		// same as when actually not found to avoid leaking information in obvious ways
+		http.Error(w, "Proof set not found", http.StatusNotFound)
+		return
+	}
+
+	// Get the ABI and pack the transaction data
+	abiData, err := contract.PDPVerifierMetaData.GetAbi()
+	if err != nil {
+		http.Error(w, "Failed to get contract ABI: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Pack the method call data
+	data, err := abiData.Pack("scheduleRemovals",
+		big.NewInt(int64(proofSetID)),
+		[]*big.Int{big.NewInt(int64(rootID))},
+		[]byte{},
+	)
+	if err != nil {
+		http.Error(w, "Failed to pack method call: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Get the sender address
+	fromAddress, err := p.getSenderAddress(ctx)
+	if err != nil {
+		http.Error(w, "Failed to get sender address: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Prepare the transaction
+	ethTx := types.NewTransaction(
+		0, // nonce will be set by SenderETH
+		contract.ContractAddresses().PDPVerifier,
+		big.NewInt(0), // value
+		0,             // gas limit (will be estimated)
+		nil,           // gas price (will be set by SenderETH)
+		data,
+	)
+
+	// Send the transaction
+	reason := "pdp-delete-root"
+	txHash, err := p.sender.Send(ctx, fromAddress, ethTx, reason)
+	if err != nil {
+		http.Error(w, "Failed to send transaction: "+err.Error(), http.StatusInternalServerError)
+		log.Errorf("Failed to send transaction: %+v", err)
+		return
+	}
+
+	// Schedule deletion of the root from the proof set using a transaction
+	_, err = p.db.BeginTransaction(ctx, func(tx *harmonydb.Tx) (bool, error) {
+		// Insert into message_waits_eth
+		_, err := tx.Exec(`
+			INSERT INTO message_waits_eth (signed_tx_hash, tx_status)
+			VALUES ($1, $2)
+		`, txHash.Hex(), "pending")
+		if err != nil {
+			return false, err
+		}
+
+		return true, nil
+	}, harmonydb.OptionRetry())
+
+	if err != nil {
+		http.Error(w, "Failed to schedule delete root: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Return 204 No Content on successful deletion
+	w.WriteHeader(http.StatusNoContent)
+
+}
+
 func (p *PDPService) handleGetProofSetRoot(w http.ResponseWriter, r *http.Request) {
 	// Spec snippet:
 	// ### GET /proof-sets/{set id}/roots/{root id}
@@ -930,31 +1053,6 @@ func (p *PDPService) handleGetProofSetRoot(w http.ResponseWriter, r *http.Reques
 	}*/
 }
 
-func (p *PDPService) handleDeleteProofSetRoot(w http.ResponseWriter, r *http.Request) {
-	// Spec snippet:
-	// ### DEL /proof-sets/{set id}/roots/{root id}
-
-	proofSetIDStr := chi.URLParam(r, "proofSetID")
-	proofSetID, err := strconv.ParseInt(proofSetIDStr, 10, 64)
-	if err != nil {
-		http.Error(w, "Invalid proof set ID", http.StatusBadRequest)
-		return
-	}
-
-	rootIDStr := chi.URLParam(r, "rootID")
-	rootID, err := strconv.ParseInt(rootIDStr, 10, 64)
-	if err != nil {
-		http.Error(w, "Invalid root ID", http.StatusBadRequest)
-		return
-	}
-
-	_ = proofSetID
-	_ = rootID
-
-	// Respond with 204 No Content
-	w.WriteHeader(http.StatusNoContent)
-}
-
 // Data models corresponding to the updated schema
 
 // PDPOwnerAddress represents the owner address with its private key

tasks/pdp/task_prove.go

@@ -170,19 +170,19 @@ func (p *ProveTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
 	}
 
 	pdpContracts := contract.ContractAddresses()
-	pdpServiceAddress := pdpContracts.PDPVerifier
+	pdpVerifierAddress := pdpContracts.PDPVerifier
 
-	pdpService, err := contract.NewPDPVerifier(pdpServiceAddress, p.ethClient)
+	pdpVerifier, err := contract.NewPDPVerifier(pdpVerifierAddress, p.ethClient)
 	if err != nil {
-		return false, xerrors.Errorf("failed to instantiate PDPVerifier contract at %s: %w", pdpServiceAddress.Hex(), err)
+		return false, xerrors.Errorf("failed to instantiate PDPVerifier contract at %s: %w", pdpVerifierAddress.Hex(), err)
 	}
 
 	callOpts := &bind.CallOpts{
 		Context: ctx,
 	}
 
 	// Proof parameters
-	challengeEpoch, err := pdpService.GetNextChallengeEpoch(callOpts, big.NewInt(proofSetID))
+	challengeEpoch, err := pdpVerifier.GetNextChallengeEpoch(callOpts, big.NewInt(proofSetID))
 	if err != nil {
 		return false, xerrors.Errorf("failed to get next challenge epoch: %w", err)
 	}
@@ -192,7 +192,7 @@ func (p *ProveTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
 		return false, xerrors.Errorf("failed to get chain randomness from beacon for pdp prove: %w", err)
 	}
 
-	proofs, err := p.GenerateProofs(ctx, pdpService, proofSetID, seed, contract.NumChallenges)
+	proofs, err := p.GenerateProofs(ctx, pdpVerifier, proofSetID, seed, contract.NumChallenges)
 	if err != nil {
 		return false, xerrors.Errorf("failed to generate proofs: %w", err)
 	}
@@ -233,7 +233,7 @@ func (p *ProveTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
 	// If gas used is 0 fee is maximized
 	gasFee := big.NewInt(0)
 	log.Infow("PDP Prove Task", "gasFeeEstimate", gasFee)
-	proofFee, err := pdpService.CalculateProofFee(callOpts, big.NewInt(proofSetID), gasFee)
+	proofFee, err := pdpVerifier.CalculateProofFee(callOpts, big.NewInt(proofSetID), gasFee)
 	if err != nil {
 		return false, xerrors.Errorf("failed to calculate proof fee: %w", err)
 	}
@@ -247,7 +247,6 @@ func (p *ProveTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
 	if err != nil {
 		return false, xerrors.Errorf("failed to get sender address: %w", err)
 	}
-	pdpVerifierAddress := contract.ContractAddresses().PDPVerifier
 
 	// Prepare the transaction (nonce will be set to 0, SenderETH will assign it)
 	txEth := types.NewTransaction(
@@ -271,6 +270,12 @@ func (p *ProveTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
 		return false, xerrors.Errorf("failed to send transaction: %w", err)
 	}
 
+	// Remove the roots previously scheduled for deletion
+	err = p.cleanupDeletedRoots(ctx, proofSetID, pdpVerifier)
+	if err != nil {
+		return false, xerrors.Errorf("failed to cleanup deleted roots: %w", err)
+	}
+
 	log.Infow("PDP Prove Task: transaction sent", "txHash", txHash, "proofSetID", proofSetID, "taskID", taskID)
 
 	// Task completed successfully
@@ -631,6 +636,65 @@ func (p *ProveTask) getSenderAddress(ctx context.Context) (common.Address, error
 	return address, nil
 }
 
+func (p *ProveTask) cleanupDeletedRoots(ctx context.Context, proofSetID int64, pdpVerifier *contract.PDPVerifier) error {
+
+	removals, err := pdpVerifier.GetScheduledRemovals(nil, big.NewInt(proofSetID))
+	if err != nil {
+		return xerrors.Errorf("failed to get scheduled removals: %w", err)
+	}
+
+	// Execute cleanup in a transaction
+	ok, err := p.db.BeginTransaction(ctx, func(tx *harmonydb.Tx) (bool, error) {
+
+		for _, removeID := range removals {
+			log.Infow("cleanupDeletedRoots", "removeID", removeID)
+			// Get the pdp_pieceref ID for the root before deleting
+			var pdpPieceRefID int64
+			err := tx.QueryRow(`
+                SELECT pdp_pieceref 
+                FROM pdp_proofset_roots 
+                WHERE proofset = $1 AND root_id = $2
+            `, proofSetID, removeID.Int64()).Scan(&pdpPieceRefID)
+			if err != nil {
+				if errors.Is(err, sql.ErrNoRows) {
+					// Root already deleted, skip
+					continue
+				}
+				return false, xerrors.Errorf("failed to get piece ref for root %d: %w", removeID, err)
+			}
+
+			// Delete the parked piece ref, this will cascade to the pdp piece ref too
+			_, err = tx.Exec(`
+				DELETE FROM parked_piece_refs
+				WHERE ref_id = $1
+			`, pdpPieceRefID)
+			if err != nil {
+				return false, xerrors.Errorf("failed to delete parked piece ref %d: %w", pdpPieceRefID, err)
+			}
+
+			// Delete the root entry
+			_, err = tx.Exec(`
+                DELETE FROM pdp_proofset_roots 
+                WHERE proofset = $1 AND root_id = $2
+            `, proofSetID, removeID)
+			if err != nil {
+				return false, xerrors.Errorf("failed to delete root %d: %w", removeID, err)
+			}
+		}
+
+		return true, nil
+	}, harmonydb.OptionRetry())
+
+	if err != nil {
+		return xerrors.Errorf("failed to cleanup deleted roots: %w", err)
+	}
+	if !ok {
+		return xerrors.Errorf("database delete not committed")
+	}
+
+	return nil
+}
+
 func (p *ProveTask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) (*harmonytask.TaskID, error) {
 	if len(ids) == 0 {
 		return nil, nil