chain piece deletion post proving

Author: LexLuthr

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

@@ -5,16 +5,15 @@
 -- 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.
+-- 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
--- the DeletePiece intake path, which inserts a row alongside every
--- schedulePieceDeletions send from here on.
+-- 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,
@@ -28,16 +27,7 @@ CREATE TABLE IF NOT EXISTS pdpv0_deletion_drain (
     -- 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())
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
 );
 
 COMMENT ON TABLE pdpv0_deletion_drain IS
@@ -48,7 +38,7 @@ 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.
+-- 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;

pdp-process-piece-deletions-findings.md

@@ -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/contract/utils.go

@@ -31,11 +31,8 @@ var (
 	ErrPDPVerifierDataSetNotLive             abi.Error
 	ErrPDPVerifierInsufficientChallengeDelay abi.Error
 
-	ErrPDPVerifierPendingPieceDeletions     abi.Error
-	ErrPDPVerifierInvalidPieceDeletionBatch abi.Error
-	ErrPDPVerifierEmptyRemovalBatch         abi.Error
-	ErrPDPVerifierOnlyStorageProvider       abi.Error
-	ErrPDPVerifierNoPiecesToProve           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
@@ -83,29 +80,12 @@ func init() {
 		panic("PDPVerifier ABI missing ExcessiveChallengeDelay error")
 	}
 
-	removalQueue := contract.RemovalQueueABI()
-
-	ErrPDPVerifierPendingPieceDeletions, ok = removalQueue.Errors["PendingPieceDeletions"]
+	ErrPDPVerifierPendingPieceDeletions, ok = parsedPDPVerifier.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"]
+	ErrPDPVerifierNoPiecesToProve, ok = parsedPDPVerifier.Errors["NoPiecesToProve"]
 	if !ok {
 		panic("PDPVerifier removal ABI missing NoPiecesToProve error")
 	}
@@ -267,22 +247,6 @@ func IsPendingPieceDeletionsError(err error) bool {
 	return strings.Contains(strings.ToLower(err.Error()), contractErrorSelector(ErrPDPVerifierPendingPieceDeletions))
 }
 
-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))
-}
-
-func IsOnlyStorageProviderError(err error) bool {
-	if err == nil {
-		return false
-	}
-	return strings.Contains(strings.ToLower(err.Error()), contractErrorSelector(ErrPDPVerifierOnlyStorageProvider))
-}
-
 func IsPDPVerifierDataSetNotLive(err error) bool {
 	if err == nil {
 		return false

tasks/pdpv0/error_detection.go

@@ -464,7 +464,7 @@ func handleNextProvingPeriodSendError(ctx context.Context, tx *harmonydb.Tx, pro
 		}
 		log.Warnw("Proving period scheduling blocked by pending piece deletions; draining first",
 			"dataSetId", dataSetId, "subsystem", alertSubsystem, "height", currentHeight, "error", sendErr)
-		return 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