Piece deletion

harmony/harmonydb/sql/20260818-pdpv0-deletion-drain.sql

@@ -0,0 +1,53 @@
+-- PDPv0 scheduled-removal draining (filecoin-project/curio#1422).
+--
+-- PDPVerifier no longer applies scheduled piece removals inside
+-- nextProvingPeriod; the storage provider drains the queue explicitly with
+-- processPieceDeletions, and nextProvingPeriod reverts while the queue is
+-- non-empty. See https://github.com/FilOzone/pdp/pull/297.
+--
+-- This table coordinates removal draining. Rows are candidates rather than
+-- confirmed work: the task's first action is an on-chain queue read, and a row
+-- whose data set has an empty queue is simply dropped. So the seed below can be
+-- indiscriminate and needs no chain access at migration time.
+--
+-- Two writers: this one-time seed, which picks up data sets already carrying a
+-- removal queue at upgrade time (including any stuck by FilOzone/pdp#283), and
+-- proving-period code, which inserts a row when it observes confirmed delete
+-- intent that still needs explicit draining.
+
+CREATE TABLE IF NOT EXISTS pdpv0_deletion_drain (
+    data_set BIGINT PRIMARY KEY REFERENCES pdp_data_sets(id) ON DELETE CASCADE,
+
+    -- ON DELETE SET NULL so an abandoned or exhausted harmony task releases its
+    -- claim automatically, the same way pdp_data_sets.challenge_request_task_id
+    -- works. Without it a row lost mid-task would never be re-claimed.
+    task_id BIGINT REFERENCES harmony_task(id) ON DELETE SET NULL,
+
+    -- In-flight processPieceDeletions transaction. At most one per data set:
+    -- drains must be sequential because each one re-reads the queue length.
+    msg_hash TEXT DEFAULT NULL,
+
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+COMMENT ON TABLE pdpv0_deletion_drain IS
+    'Data sets that may have a non-empty PDPVerifier scheduled-removal queue to drain via processPieceDeletions.';
+
+-- The watcher only ever looks for unclaimed rows with no drain in flight.
+CREATE INDEX IF NOT EXISTS idx_pdpv0_deletion_drain_pending
+    ON pdpv0_deletion_drain (data_set)
+    WHERE task_id IS NULL AND msg_hash IS NULL;
+
+-- Reorg rollback locates drain rows by in-flight message.
+CREATE INDEX IF NOT EXISTS idx_pdpv0_deletion_drain_msg_hash
+    ON pdpv0_deletion_drain (msg_hash)
+    WHERE msg_hash IS NOT NULL;
+
+-- Reclaiming abandoned rows scans by task_id.
+CREATE INDEX IF NOT EXISTS idx_pdpv0_deletion_drain_task_id
+    ON pdpv0_deletion_drain (task_id)
+    WHERE task_id IS NOT NULL;
+
+INSERT INTO pdpv0_deletion_drain (data_set)
+SELECT id FROM pdp_data_sets
+ON CONFLICT (data_set) DO NOTHING;

pdp/contract/utils.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/jellydator/ttlcache/v2"
+	"golang.org/x/mod/semver"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"
@@ -56,6 +57,23 @@ const (
 	CapIpniPeerIDDeprecated = "IPNIPeerID"
 )
 
+const pdpVerifierProcessPieceDeletionsAfterVersion = "v3.4.0"
+
+func SemverVersion(version string) string {
+	if strings.HasPrefix(version, "v") {
+		return version
+	}
+	return "v" + version
+}
+
+func SupportsPieceDeletionProcessing(ctx context.Context, verifier *PDPVerifier) (bool, error) {
+	version, err := verifier.VERSION(EthCallOpts(ctx))
+	if err != nil {
+		return false, xerrors.Errorf("failed to get PDPVerifier version: %w", err)
+	}
+	return semver.Compare(SemverVersion(version), pdpVerifierProcessPieceDeletionsAfterVersion) > 0, nil
+}
+
 // PDPOfferingData converts a PDPOffering-like struct to capability key-value pairs
 type PDPOfferingData struct {
 	ServiceURL               string

pdp/handlers.go

@@ -1127,7 +1127,7 @@ func (p *PDPService) handleDeleteDataSetPiece(w http.ResponseWriter, r *http.Req
 		return
 	}
 	if len(queued) >= contract.ConservativeEnqueuedRemovalsLimit {
-		http.Error(w, fmt.Sprintf("data set %d already has %d scheduled removals queued (limit %d); retry after the next proving period flushes the queue",
+		http.Error(w, fmt.Sprintf("data set %d already has %d scheduled removals queued (limit %d); retry once they have been processed",
 			dataSetId, len(queued), contract.ConservativeEnqueuedRemovalsLimit), http.StatusTooManyRequests)
 		return
 	}
@@ -1206,6 +1206,7 @@ func (p *PDPService) handleDeleteDataSetPiece(w http.ResponseWriter, r *http.Req
 			log.Errorw("Failed to update rm_message_hash in pdp_data_set_pieces", "dataSetId", dataSetId, "pieceIDs", pieceIDsI64, "error", err)
 			return false, err
 		}
+
 		log.Infow("scheduled user requested deletion", "dataSetId", dataSetId, "pieceIDs", pieceIDsI64, "txHash", txHashLower)
 
 		return true, nil

pdpnode/tasks.go

@@ -79,13 +79,15 @@ func buildPDPTasks(ctx context.Context, d *Deps, chainSched *chainsched.CurioCha
 	pay.NewSettleWatcher(w)
 	pdpv0.NewDataSetDeleteWatcher(w)
 	pdpv0.NewCleanupPiecesWatcher(w)
+	pdpv0.NewProcessDeletionsWatcher(w)
 	pdpv0.NewProvingPeriodWatcher(w)
 	pdpv0.NewTerminateServiceWatcher(w)
 
 	tasks = append(tasks,
 		pdpv0.NewProveTask(db, ethClient, d.Chain, w, senderEth, d.CachedPieceReader, d.IndexStore),
 		pdpv0.NewNextProvingPeriodTask(db, ethClient, d.Chain, w, senderEth),
 		pdpv0.NewInitProvingPeriodTask(db, ethClient, d.Chain, w, senderEth),
+		pdpv0.NewProcessDeletionsTask(db, ethClient, d.Chain, w, senderEth),
 		pdpv0.NewPDPNotifyTask(ctx, db),
 		pdpv0.NewPDPPullPieceTask(ctx, db, d.PieceIO, cfg.Subsystems.PDPPullPieceMaxTasks),
 		pdpv0.NewTerminateServiceTask(db, ethClient, senderEth),

tasks/pdpv0/error_detection.go

@@ -31,6 +31,9 @@ var (
 	ErrPDPVerifierDataSetNotLive             abi.Error
 	ErrPDPVerifierInsufficientChallengeDelay abi.Error
 
+	ErrPDPVerifierPendingPieceDeletions abi.Error
+	ErrPDPVerifierNoPiecesToProve       abi.Error
+
 	// Unexpected proving invariant errors. Curio should not produce these in
 	// normal PDPv0 initPP/nextPP/prove flow; classify them explicitly so they
 	// alert and require investigation instead of entering recovery/backoff paths.
@@ -77,6 +80,16 @@ func init() {
 		panic("PDPVerifier ABI missing ExcessiveChallengeDelay error")
 	}
 
+	ErrPDPVerifierPendingPieceDeletions, ok = parsedPDPVerifier.Errors["PendingPieceDeletions"]
+	if !ok {
+		panic("PDPVerifier removal ABI missing PendingPieceDeletions error")
+	}
+
+	ErrPDPVerifierNoPiecesToProve, ok = parsedPDPVerifier.Errors["NoPiecesToProve"]
+	if !ok {
+		panic("PDPVerifier removal ABI missing NoPiecesToProve error")
+	}
+
 	parsedFWSS, err := FWSS.FilecoinWarmStorageServiceMetaData.GetAbi()
 	if err != nil {
 		panic("failed to parse FWSS ABI: " + err.Error())
@@ -218,7 +231,27 @@ func IsNextProvingPeriodEmptyDatasetError(err error) bool {
 	if err == nil {
 		return false
 	}
-	return strings.Contains(strings.ToLower(err.Error()), strings.ToLower(provingRevertNoLeavesForProvingPeriod))
+	errStr := strings.ToLower(err.Error())
+	return strings.Contains(errStr, strings.ToLower(provingRevertNoLeavesForProvingPeriod)) ||
+		strings.Contains(errStr, contractErrorSelector(ErrPDPVerifierNoPiecesToProve))
+}
+
+// IsPendingPieceDeletionsError returns true when nextProvingPeriod (or initPP)
+// refuses to roll over because the data set still has scheduled removals
+// queued. This is recoverable: the process deletions task processes the queue and the
+// proving-period task retries.
+func IsPendingPieceDeletionsError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(strings.ToLower(err.Error()), contractErrorSelector(ErrPDPVerifierPendingPieceDeletions))
+}
+
+func IsPDPVerifierDataSetNotLive(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(strings.ToLower(err.Error()), contractErrorSelector(ErrPDPVerifierDataSetNotLive))
 }
 
 // IsRefreshProvingStateError returns true when initPP/nextPP selected a

tasks/pdpv0/task_init_pp.go

@@ -127,6 +127,17 @@ func (ipp *InitProvingPeriodTask) Do(ctx context.Context, taskID harmonytask.Tas
 		}
 	}()
 
+	// initPP sends nextProvingPeriod calldata, so it reverts on pending
+	// deletions exactly as nextPP does. Drain first.
+	draining, err := hasDrainInFlight(ctx, ipp.db, dataSetId)
+	if err != nil {
+		return false, err
+	}
+	if draining {
+		log.Debugw("deferring initProvingPeriod until scheduled removals are drained", "dataSetId", dataSetId)
+		return true, nil
+	}
+
 	// Get the listener address for this data set from the PDPVerifier contract
 	pdpVerifier, err := contract.NewPDPVerifier(contract.ContractAddresses().PDPVerifier, ipp.ethClient)
 	if err != nil {

tasks/pdpv0/task_next_pp.go

@@ -128,6 +128,20 @@ func (n *NextProvingPeriodTask) Do(ctx context.Context, taskID harmonytask.TaskI
 		}
 	}()
 
+	// PDPVerifier reverts nextProvingPeriod while scheduled removals remain, so
+	// there is no point sending one while a drain is in flight. The drain
+	// watcher runs in an earlier phase; complete here and let the next tipset
+	// re-schedule this task. A queue with no drain in flight is handled by the
+	// PendingPieceDeletions revert instead.
+	draining, err := hasDrainInFlight(ctx, n.db, dataSetId)
+	if err != nil {
+		return false, err
+	}
+	if draining {
+		log.Debugw("deferring nextProvingPeriod until scheduled removals are drained", "dataSetId", dataSetId)
+		return true, nil
+	}
+
 	// Get the listener address for this data set from the PDPVerifier contract
 	pdpVerifier, err := contract.NewPDPVerifier(contract.ContractAddresses().PDPVerifier, n.ethClient)
 	if err != nil {
@@ -440,6 +454,17 @@ func (n *NextProvingPeriodTask) handleNextProvingPeriodPreflightError(ctx contex
 
 func handleNextProvingPeriodSendError(ctx context.Context, tx *harmonydb.Tx, provingSchedule *contract.IPDPProvingSchedule, al curioalerting.AlertingInterface, alertSubsystem string, dataSetId int64, currentHeight int64, sendErr error) error {
 	switch {
+	case IsPendingPieceDeletionsError(sendErr):
+		// PDPVerifier will not roll over while scheduled removals remain. This
+		// is recoverable and expected whenever a drain has not finished: make
+		// sure the data set is queued for draining and retry the task. It must
+		// not reach disableProvingForEmptyDataset or any terminal path.
+		if err := enqueueDeletionDrain(tx, dataSetId); err != nil {
+			return err
+		}
+		log.Warnw("Proving period scheduling blocked by pending piece deletions; draining first",
+			"dataSetId", dataSetId, "subsystem", alertSubsystem, "height", currentHeight, "error", sendErr)
+		return nil
 	case IsInsufficientChallengeDelayError(sendErr):
 		// The challenge epoch was too close to the current block. Retry the
 		// task so it recomputes challenge state and calldata instead of