Integrate Process Piece Deletions

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

@@ -0,0 +1,63 @@
+-- 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 is the work queue the drain watcher selects from. 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
+-- the DeletePiece intake path, which inserts a row alongside every
+-- schedulePieceDeletions send from here on.
+
+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,
+
+    -- Bumped when a drain send fails, bounding retries so a permanently failing
+    -- data set cannot spin forever.
+    failures BIGINT NOT NULL DEFAULT 0,
+
+    -- Set when the task claimed the row but could not act on it yet (for
+    -- example the challenge window has not closed). Keeps the watcher from
+    -- re-claiming and re-reading chain state on every tipset.
+    blocked_at TIMESTAMPTZ DEFAULT NULL,
+
+    created_at TIMESTAMPTZ NOT NULL DEFAULT TIMEZONE('UTC', 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;
+
+-- The confirmation watcher scans 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/removals.go

@@ -0,0 +1,128 @@
+package contract
+
+import (
+	"context"
+	"math/big"
+	"strconv"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"golang.org/x/xerrors"
+)
+
+// This file is hand-maintained, unlike the generated PDPVerifier bindings. It
+// carries the removal-queue entrypoint and custom errors added by
+// https://github.com/FilOzone/pdp/pull/297, which are not yet in the checked-in
+// ABI. Delete it and use the generated bindings once PDPVerifier.abi is
+// regenerated against a release containing processPieceDeletions.
+const removalQueueABIJSON = `[
+	{"type":"function","name":"processPieceDeletions","inputs":[{"name":"setId","type":"uint256","internalType":"uint256"},{"name":"removalCount","type":"uint256","internalType":"uint256"}],"outputs":[],"stateMutability":"nonpayable"},
+	{"type":"error","name":"OnlyStorageProvider","inputs":[]},
+	{"type":"error","name":"InvalidPieceDeletionBatch","inputs":[]},
+	{"type":"error","name":"EmptyRemovalBatch","inputs":[]},
+	{"type":"error","name":"PendingPieceDeletions","inputs":[{"name":"count","type":"uint256","internalType":"uint256"}]},
+	{"type":"error","name":"NoPiecesToProve","inputs":[]}
+]`
+
+var removalQueueABI abi.ABI
+
+func init() {
+	parsed, err := abi.JSON(strings.NewReader(removalQueueABIJSON))
+	if err != nil {
+		panic("failed to parse PDPVerifier removal-queue ABI fragment: " + err.Error())
+	}
+	removalQueueABI = parsed
+}
+
+// RemovalQueueABI exposes the hand-maintained fragment so callers can resolve
+// the new custom errors for revert classification.
+func RemovalQueueABI() *abi.ABI {
+	return &removalQueueABI
+}
+
+// PackProcessPieceDeletions builds calldata draining removalCount entries from
+// the tail of a data set's scheduled-removal queue.
+func PackProcessPieceDeletions(setID, removalCount *big.Int) ([]byte, error) {
+	data, err := removalQueueABI.Pack("processPieceDeletions", setID, removalCount)
+	if err != nil {
+		return nil, xerrors.Errorf("packing processPieceDeletions: %w", err)
+	}
+	return data, nil
+}
+
+// PieceDeletionProcessingMinVersion is the lowest PDPVerifier VERSION that
+// exposes processPieceDeletions. Below it, nextProvingPeriod still drains the
+// removal queue itself and Curio must not attempt to drain it separately.
+//
+// TODO: confirm against the release that lands FilOzone/pdp#297. The currently
+// deployed contract reports "3.4.0"; this is a placeholder for the next bump.
+const PieceDeletionProcessingMinVersion = "3.5.0"
+
+// SupportsPieceDeletionProcessing reports whether the deployed PDPVerifier
+// exposes processPieceDeletions.
+//
+// The value is read rather than cached for the process lifetime: PDPVerifier is
+// a UUPS proxy and can be upgraded in place underneath a running Curio.
+func SupportsPieceDeletionProcessing(ctx context.Context, verifier *PDPVerifier) (bool, error) {
+	version, err := verifier.VERSION(EthCallOpts(ctx))
+	if err != nil {
+		return false, xerrors.Errorf("reading PDPVerifier VERSION: %w", err)
+	}
+
+	cmp, err := compareSemver(version, PieceDeletionProcessingMinVersion)
+	if err != nil {
+		return false, xerrors.Errorf("comparing PDPVerifier version %q: %w", version, err)
+	}
+	return cmp >= 0, nil
+}
+
+// compareSemver orders two dotted version strings, returning -1, 0 or 1. Any
+// pre-release or build suffix is ignored: "3.5.0-rc1" compares equal to "3.5.0",
+// which is the conservative choice for a feature gate on a release candidate.
+func compareSemver(a, b string) (int, error) {
+	aParts, err := parseSemver(a)
+	if err != nil {
+		return 0, err
+	}
+	bParts, err := parseSemver(b)
+	if err != nil {
+		return 0, err
+	}
+
+	for i := range aParts {
+		switch {
+		case aParts[i] < bParts[i]:
+			return -1, nil
+		case aParts[i] > bParts[i]:
+			return 1, nil
+		}
+	}
+	return 0, nil
+}
+
+func parseSemver(v string) ([3]int, error) {
+	var out [3]int
+
+	v = strings.TrimSpace(v)
+	v = strings.TrimPrefix(v, "v")
+	if idx := strings.IndexAny(v, "-+"); idx >= 0 {
+		v = v[:idx]
+	}
+
+	fields := strings.Split(v, ".")
+	if len(fields) == 0 || len(fields) > 3 {
+		return out, xerrors.Errorf("malformed version %q", v)
+	}
+
+	for i, field := range fields {
+		n, err := strconv.Atoi(field)
+		if err != nil {
+			return out, xerrors.Errorf("malformed version %q: component %q is not a number", v, field)
+		}
+		if n < 0 {
+			return out, xerrors.Errorf("malformed version %q: negative component", v)
+		}
+		out[i] = n
+	}
+	return out, nil
+}

pdp/handlers.go

@@ -1114,6 +1114,43 @@ func (p *PDPService) handleDeleteDataSetPiece(w http.ResponseWriter, r *http.Req
 		return
 	}
 
+	// Removals can only be drained once the data set has a proving schedule: the
+	// drain waits for the challenge window to close, and there is no window
+	// without one. Accepting a removal before then would build a queue that can
+	// never be drained, while blocking initProvingPeriod from ever succeeding.
+	var proveAtEpoch sql.NullInt64
+	err = p.db.QueryRow(ctx, `SELECT prove_at_epoch FROM pdp_data_sets WHERE id = $1`, dataSetId).Scan(&proveAtEpoch)
+	if err != nil {
+		httpServerError(w, http.StatusInternalServerError, "Failed to read data set proving schedule", err)
+		return
+	}
+	if !proveAtEpoch.Valid {
+		http.Error(w, fmt.Sprintf("data set %d has not started proving yet; retry once its first proving period is initialized", dataSetId),
+			http.StatusTooManyRequests)
+		return
+	}
+
+	// A drain transaction in flight means the queue is mid-processing. Adding to
+	// it now re-fills what is being drained and pushes the proving-period
+	// rollover further out, so refuse until the drain settles.
+	//
+	// Keyed on an in-flight message rather than the mere presence of a drain
+	// row: the migration seeds a row for every data set, so row existence would
+	// refuse every deletion until that sweep completed.
+	var draining bool
+	err = p.db.QueryRow(ctx, `
+		SELECT EXISTS (SELECT 1 FROM pdpv0_deletion_drain WHERE data_set = $1 AND msg_hash IS NOT NULL)
+	`, dataSetId).Scan(&draining)
+	if err != nil {
+		httpServerError(w, http.StatusInternalServerError, "Failed to read removal drain state", err)
+		return
+	}
+	if draining {
+		http.Error(w, fmt.Sprintf("data set %d is draining previously scheduled removals; retry once they are processed", dataSetId),
+			http.StatusTooManyRequests)
+		return
+	}
+
 	// Soft gate: refuse if the data set's on-chain removal queue is already at our
 	// conservative ceiling. This keeps us well clear of the on-chain MAX_ENQUEUED_REMOVALS.
 	pdpVerifier, err := contract.NewPDPVerifier(contract.ContractAddresses().PDPVerifier, p.ethClient)
@@ -1127,7 +1164,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 +1243,19 @@ 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
 		}
+
+		// PDPVerifier no longer applies scheduled removals inside
+		// nextProvingPeriod, so the data set needs an explicit drain before it
+		// can roll over. Enqueue it in the same transaction as the send.
+		_, err = tx.Exec(`
+			INSERT INTO pdpv0_deletion_drain (data_set)
+			VALUES ($1)
+			ON CONFLICT (data_set) DO UPDATE SET blocked_at = NULL`, dataSetId)
+		if err != nil {
+			log.Errorw("Failed to enqueue removal drain", "dataSetId", dataSetId, "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,15 @@ var (
 	ErrPDPVerifierDataSetNotLive             abi.Error
 	ErrPDPVerifierInsufficientChallengeDelay abi.Error
 
+	// Removal-queue errors from FilOzone/pdp#297. These resolve against the
+	// hand-maintained ABI fragment in pdp/contract/removals.go until the
+	// generated PDPVerifier bindings carry them.
+	ErrPDPVerifierPendingPieceDeletions     abi.Error
+	ErrPDPVerifierInvalidPieceDeletionBatch abi.Error
+	ErrPDPVerifierEmptyRemovalBatch         abi.Error
+	ErrPDPVerifierOnlyStorageProvider       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 +86,33 @@ func init() {
 		panic("PDPVerifier ABI missing ExcessiveChallengeDelay error")
 	}
 
+	removalQueue := contract.RemovalQueueABI()
+
+	ErrPDPVerifierPendingPieceDeletions, ok = removalQueue.Errors["PendingPieceDeletions"]
+	if !ok {
+		panic("PDPVerifier removal ABI missing PendingPieceDeletions error")
+	}
+
+	ErrPDPVerifierInvalidPieceDeletionBatch, ok = removalQueue.Errors["InvalidPieceDeletionBatch"]
+	if !ok {
+		panic("PDPVerifier removal ABI missing InvalidPieceDeletionBatch error")
+	}
+
+	ErrPDPVerifierEmptyRemovalBatch, ok = removalQueue.Errors["EmptyRemovalBatch"]
+	if !ok {
+		panic("PDPVerifier removal ABI missing EmptyRemovalBatch error")
+	}
+
+	ErrPDPVerifierOnlyStorageProvider, ok = removalQueue.Errors["OnlyStorageProvider"]
+	if !ok {
+		panic("PDPVerifier removal ABI missing OnlyStorageProvider error")
+	}
+
+	ErrPDPVerifierNoPiecesToProve, ok = removalQueue.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())
@@ -214,11 +250,63 @@ func IsProvingPeriodNotInitializedError(err error) bool {
 
 // IsNextProvingPeriodEmptyDatasetError returns true when PDPVerifier refuses to
 // start the next proving period because the current proving set has no leaves.
+//
+// Both encodings are matched because one Curio build spans two contract
+// versions: the condition is a string revert before FilOzone/pdp#297 and the
+// NoPiecesToProve custom error afterwards. The underlying requirement --
+// dataSetLeafCount > 0 -- is the same, so neither form can be dropped until no
+// deployment runs the older contract.
 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 drain 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))
+}
+
+// IsStaleRemovalQueueViewError returns true when processPieceDeletions rejects
+// the requested batch because Curio's view of the queue is out of date -- the
+// queue shrank, or emptied, between the read and the send. Re-reading the queue
+// and retrying is the correct response.
+func IsStaleRemovalQueueViewError(err error) bool {
+	if err == nil {
+		return false
+	}
+	errStr := strings.ToLower(err.Error())
+	return strings.Contains(errStr, contractErrorSelector(ErrPDPVerifierInvalidPieceDeletionBatch)) ||
+		strings.Contains(errStr, contractErrorSelector(ErrPDPVerifierEmptyRemovalBatch))
+}
+
+// IsOnlyStorageProviderError returns true when PDPVerifier rejects a removal
+// call because the sender is not the data set's storage provider. This needs
+// operator attention rather than a retry.
+func IsOnlyStorageProviderError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(strings.ToLower(err.Error()), contractErrorSelector(ErrPDPVerifierOnlyStorageProvider))
+}
+
+// IsPDPVerifierDataSetNotLive returns true when PDPVerifier reports that a data
+// set is no longer live. In the removal pipeline this means the data set is
+// being deleted or cleaned up, so its removal queue no longer matters.
+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