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/PDPVerifier.abi

@@ -172,6 +172,24 @@
     "outputs": [],
     "stateMutability": "nonpayable"
   },
+  {
+    "type": "function",
+    "name": "announceUpgradePlan",
+    "inputs": [
+      {
+        "name": "nextImplementation",
+        "type": "address",
+        "internalType": "address"
+      },
+      {
+        "name": "delayEpochs",
+        "type": "uint96",
+        "internalType": "uint96"
+      }
+    ],
+    "outputs": [],
+    "stateMutability": "nonpayable"
+  },
   {
     "type": "function",
     "name": "calculateProofFee",
@@ -791,6 +809,19 @@
     "outputs": [],
     "stateMutability": "nonpayable"
   },
+  {
+    "type": "function",
+    "name": "legacyPieceStorageIdLimit",
+    "inputs": [],
+    "outputs": [
+      {
+        "name": "",
+        "type": "uint64",
+        "internalType": "uint64"
+      }
+    ],
+    "stateMutability": "view"
+  },
   {
     "type": "function",
     "name": "migrate",
@@ -900,6 +931,24 @@
     ],
     "stateMutability": "view"
   },
+  {
+    "type": "function",
+    "name": "processPieceDeletions",
+    "inputs": [
+      {
+        "name": "setId",
+        "type": "uint256",
+        "internalType": "uint256"
+      },
+      {
+        "name": "removalCount",
+        "type": "uint256",
+        "internalType": "uint256"
+      }
+    ],
+    "outputs": [],
+    "stateMutability": "nonpayable"
+  },
   {
     "type": "function",
     "name": "proposeDataSetStorageProvider",
@@ -1232,6 +1281,43 @@
     ],
     "anonymous": false
   },
+  {
+    "type": "event",
+    "name": "PiecesAddedV2",
+    "inputs": [
+      {
+        "name": "setId",
+        "type": "uint256",
+        "indexed": true,
+        "internalType": "uint256"
+      },
+      {
+        "name": "firstPieceId",
+        "type": "uint256",
+        "indexed": false,
+        "internalType": "uint256"
+      },
+      {
+        "name": "pieceCids",
+        "type": "tuple[]",
+        "indexed": false,
+        "internalType": "struct Cids.PackedCid[]",
+        "components": [
+          {
+            "name": "header",
+            "type": "bytes32",
+            "internalType": "bytes32"
+          },
+          {
+            "name": "root",
+            "type": "bytes32",
+            "internalType": "bytes32"
+          }
+        ]
+      }
+    ],
+    "anonymous": false
+  },
   {
     "type": "event",
     "name": "PiecesRemoved",
@@ -1251,6 +1337,25 @@
     ],
     "anonymous": false
   },
+  {
+    "type": "event",
+    "name": "PiecesScheduledForRemoval",
+    "inputs": [
+      {
+        "name": "setId",
+        "type": "uint256",
+        "indexed": true,
+        "internalType": "uint256"
+      },
+      {
+        "name": "pieceIds",
+        "type": "uint256[]",
+        "indexed": false,
+        "internalType": "uint256[]"
+      }
+    ],
+    "anonymous": false
+  },
   {
     "type": "event",
     "name": "PossessionProven",
@@ -1375,6 +1480,11 @@
       }
     ]
   },
+  {
+    "type": "error",
+    "name": "CidTooShort",
+    "inputs": []
+  },
   {
     "type": "error",
     "name": "CleanupDepositRequired",
@@ -1421,6 +1531,11 @@
     "name": "ERC1967NonPayable",
     "inputs": []
   },
+  {
+    "type": "error",
+    "name": "EmptyRemovalBatch",
+    "inputs": []
+  },
   {
     "type": "error",
     "name": "ExcessiveChallengeDelay",
@@ -1474,29 +1589,65 @@
       }
     ]
   },
+  {
+    "type": "error",
+    "name": "InvalidCommPv2DigestLength",
+    "inputs": []
+  },
+  {
+    "type": "error",
+    "name": "InvalidCommPv2MultihashLength",
+    "inputs": []
+  },
+  {
+    "type": "error",
+    "name": "InvalidCommPv2Prefix",
+    "inputs": []
+  },
+  {
+    "type": "error",
+    "name": "InvalidImplementation",
+    "inputs": [
+      {
+        "name": "implementation",
+        "type": "address",
+        "internalType": "address"
+      }
+    ]
+  },
   {
     "type": "error",
     "name": "InvalidInitialization",
     "inputs": []
   },
+  {
+    "type": "error",
+    "name": "InvalidPieceDeletionBatch",
+    "inputs": []
+  },
   {
     "type": "error",
     "name": "MaxPiecesMustBePositive",
     "inputs": []
   },
   {
     "type": "error",
-    "name": "NotInitializing",
+    "name": "NoPiecesToProve",
     "inputs": []
   },
   {
     "type": "error",
-    "name": "OnlyStorageProviderCanCleanupPieces",
+    "name": "NonMinimalUvarint",
     "inputs": []
   },
   {
     "type": "error",
-    "name": "OnlyStorageProviderCanDelete",
+    "name": "NotInitializing",
+    "inputs": []
+  },
+  {
+    "type": "error",
+    "name": "OnlyStorageProvider",
     "inputs": []
   },
   {
@@ -1521,6 +1672,22 @@
       }
     ]
   },
+  {
+    "type": "error",
+    "name": "PendingPieceDeletions",
+    "inputs": [
+      {
+        "name": "count",
+        "type": "uint256",
+        "internalType": "uint256"
+      }
+    ]
+  },
+  {
+    "type": "error",
+    "name": "PieceMetadataOverflow",
+    "inputs": []
+  },
   {
     "type": "error",
     "name": "TransferFailed",
@@ -1541,5 +1708,15 @@
         "internalType": "bytes32"
       }
     ]
+  },
+  {
+    "type": "error",
+    "name": "UnterminatedUvarint",
+    "inputs": []
+  },
+  {
+    "type": "error",
+    "name": "UvarintOverflow",
+    "inputs": []
   }
 ]

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),