diff --git a/harmony/harmonydb/downgrade/20260818-pdpv0-deletion-drain.sql b/harmony/harmonydb/downgrade/20260818-pdpv0-deletion-drain.sql new file mode 100644 index 000000000..af87837cb --- /dev/null +++ b/harmony/harmonydb/downgrade/20260818-pdpv0-deletion-drain.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS pdpv0_deletion_drain; diff --git a/harmony/harmonydb/sql/20260818-pdpv0-deletion-drain.sql b/harmony/harmonydb/sql/20260818-pdpv0-deletion-drain.sql new file mode 100644 index 000000000..c89678859 --- /dev/null +++ b/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; diff --git a/pdp/contract/utils.go b/pdp/contract/utils.go index 8c5175876..feb6d34df 100644 --- a/pdp/contract/utils.go +++ b/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 diff --git a/pdp/handlers.go b/pdp/handlers.go index ce451b356..8721249b4 100644 --- a/pdp/handlers.go +++ b/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 diff --git a/pdpnode/tasks.go b/pdpnode/tasks.go index ecf530e8e..c15675b01 100644 --- a/pdpnode/tasks.go +++ b/pdpnode/tasks.go @@ -79,6 +79,7 @@ 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) @@ -86,6 +87,7 @@ func buildPDPTasks(ctx context.Context, d *Deps, chainSched *chainsched.CurioCha 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), diff --git a/tasks/pdpv0/error_detection.go b/tasks/pdpv0/error_detection.go index f0e7502d7..8ff61ec09 100644 --- a/tasks/pdpv0/error_detection.go +++ b/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 diff --git a/tasks/pdpv0/task_next_pp.go b/tasks/pdpv0/task_next_pp.go index ea3383a09..c8d7319cd 100644 --- a/tasks/pdpv0/task_next_pp.go +++ b/tasks/pdpv0/task_next_pp.go @@ -51,6 +51,10 @@ func NewNextProvingPeriodTask(db *harmonydb.DB, ethClient ethchain.EthClient, fi fil: fil, al: w.al, } + verifier, err := contract.NewPDPVerifier(contract.ContractAddresses().PDPVerifier, ethClient) + if err != nil { + panic(err) + } _ = w.AddWatcher(func(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient, al curioalerting.AlertingInterface, revert, apply *chainTypes.TipSet) { if apply == nil { @@ -63,13 +67,37 @@ func NewNextProvingPeriodTask(db *harmonydb.DB, ethClient ethchain.EthClient, fi } currentHeight := apply.Height() - err := db.Select(ctx, &toCallNext, ` - SELECT id - FROM pdp_data_sets - WHERE challenge_request_task_id IS NULL - AND (prove_at_epoch + challenge_window) <= $1 - AND unrecoverable_proving_failure_epoch IS NULL - `, currentHeight) + supported, err := contract.SupportsPieceDeletionProcessing(ctx, verifier) + if err != nil { + return + } + + // NextProvingPeriod should only schedule based on deletion drains rows after + // the pdp verifier has been upgraded. Until then no rows will ever be processed + // or removed from pdpv0_deletion_drains. + // TODO: remove this logic after upgrade -- https://github.com/filecoin-project/curio/issues/1455 + if supported { + err = db.Select(ctx, &toCallNext, ` + SELECT id + FROM pdp_data_sets + WHERE challenge_request_task_id IS NULL + AND (prove_at_epoch + challenge_window) <= $1 + AND unrecoverable_proving_failure_epoch IS NULL + AND NOT EXISTS ( + SELECT 1 + FROM pdpv0_deletion_drain d + WHERE d.data_set = pdp_data_sets.id + ) + `, currentHeight) + } else { + err = db.Select(ctx, &toCallNext, ` + SELECT id + FROM pdp_data_sets + WHERE challenge_request_task_id IS NULL + AND (prove_at_epoch + challenge_window) <= $1 + AND unrecoverable_proving_failure_epoch IS NULL + `, currentHeight) + } if err != nil && !errors.Is(err, pgx.ErrNoRows) { _ = al.EmitEvent(ctx, curioalerting.AlertEvent{ System: alertType, @@ -440,6 +468,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 diff --git a/tasks/pdpv0/task_process_deletions.go b/tasks/pdpv0/task_process_deletions.go new file mode 100644 index 000000000..d22cf63a3 --- /dev/null +++ b/tasks/pdpv0/task_process_deletions.go @@ -0,0 +1,343 @@ +package pdpv0 + +import ( + "context" + "errors" + "fmt" + "math/big" + "time" + + "github.com/ethereum/go-ethereum/accounts/abi" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/yugabyte/pgx/v5" + "golang.org/x/xerrors" + + "github.com/filecoin-project/curio/alertmanager/curioalerting" + "github.com/filecoin-project/curio/harmony/harmonydb" + "github.com/filecoin-project/curio/harmony/harmonytask" + "github.com/filecoin-project/curio/harmony/resources" + "github.com/filecoin-project/curio/harmony/taskhelp" + "github.com/filecoin-project/curio/lib/ethchain" + "github.com/filecoin-project/curio/lib/promise" + "github.com/filecoin-project/curio/pdp/contract" + "github.com/filecoin-project/curio/tasks/message" + "github.com/filecoin-project/curio/tasks/tasknames" + + chainTypes "github.com/filecoin-project/lotus/chain/types" +) + +const alertNameProcessDeletions = "ProcessDeletions" + +// reasonPDPProcessDeletions is the SenderETH reason for processPieceDeletions +// sends. Also registered with the reorg checker. +const reasonPDPProcessDeletions = "pdp-process-deletions" + +// processDeletionsBatchSize is the starting number of queue entries drained per +// transaction. It matches PDPVerifier's PiecesRemoved event chunk size, and sits +// well under the block gas limit. +// +// With ConservativeEnqueuedRemovalsLimit at 35 this never binds in steady state +// -- one message drains a period's whole queue. It exists for migration-seeded +// backlogs, which predate that limit and can run to a few hundred pieces. The +// halving loop below, not this constant, is what guarantees progress: the +// listener's gas cost is invisible to PDPVerifier, which is the root cause of +// FilOzone/pdp#283. +const processDeletionsBatchSize = 100 + +// processDeletionsScheduleLimit bounds how many data sets are claimed per tipset. +const processDeletionsScheduleLimit = 16 + +type ProcessDeletionsTask struct { + db *harmonydb.DB + ethClient ethchain.EthClient + sender *message.SenderETH + + fil ProcessDeletionsChainApi + + al curioalerting.AlertingInterface + + addFunc promise.Promise[harmonytask.AddTaskFunc] +} + +type ProcessDeletionsChainApi interface { + ChainHead(context.Context) (*chainTypes.TipSet, error) +} + +// NewProcessDeletionsTask drains PDPVerifier scheduled-removal queues. +// +// Scheduling is driven by pdpv0_deletion_drain coordination rows, gated on the +// same prove_at_epoch + challenge_window deadline as nextProvingPeriod. A NULL +// task_id means no Harmony task currently owns the row; a NULL msg_hash means no +// processPieceDeletions transaction is waiting for confirmation. +func NewProcessDeletionsTask(db *harmonydb.DB, ethClient ethchain.EthClient, fil ProcessDeletionsChainApi, w *Watcher, sender *message.SenderETH) *ProcessDeletionsTask { + p := &ProcessDeletionsTask{ + db: db, + ethClient: ethClient, + sender: sender, + fil: fil, + al: w.al, + } + verifier, err := contract.NewPDPVerifier(contract.ContractAddresses().PDPVerifier, ethClient) + if err != nil { + panic(err) + } + + _ = w.AddWatcher(func(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient, al curioalerting.AlertingInterface, revert, apply *chainTypes.TipSet) { + if apply == nil { + return + } + + // Only schedule work for process deletions once the pdp verifier has been upgraded + supported, err := contract.SupportsPieceDeletionProcessing(ctx, verifier) + if err != nil { + return + } + if !supported { + return + } + + var candidates []struct { + DataSetID int64 `db:"data_set"` + } + + currentHeight := apply.Height() + err = db.Select(ctx, &candidates, ` + SELECT d.data_set + FROM pdpv0_deletion_drain d + JOIN pdp_data_sets ds ON ds.id = d.data_set + WHERE d.task_id IS NULL + AND d.msg_hash IS NULL + AND (ds.prove_at_epoch + ds.challenge_window) <= $1 + ORDER BY d.data_set + LIMIT $2 + `, currentHeight, processDeletionsScheduleLimit) + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + _ = al.EmitEvent(ctx, curioalerting.AlertEvent{ + System: alertType, + Subsystem: alertNameProcessDeletions, + Message: fmt.Sprintf("failed to select data sets needing removal draining: %s", err), + }) + return + } + + for _, candidate := range candidates { + dataSetID := candidate.DataSetID + p.addFunc.Val(ctx)(func(id harmonytask.TaskID, tx *harmonydb.Tx) (shouldCommit bool, seriousError error) { + affected, err := tx.Exec(` + UPDATE pdpv0_deletion_drain + SET task_id = $1 + WHERE data_set = $2 AND task_id IS NULL AND msg_hash IS NULL + `, id, dataSetID) + if err != nil { + return false, xerrors.Errorf("failed to claim deletion drain row: %w", err) + } + if affected == 0 { + // Claimed elsewhere. + return false, nil + } + return true, nil + }) + } + }, WatcherOrderProcessDeletions) + + return p +} + +func (p *ProcessDeletionsTask) Do(ctx context.Context, taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error) { + var dataSetID int64 + err = p.db.QueryRow(ctx, `SELECT data_set FROM pdpv0_deletion_drain WHERE task_id = $1`, taskID).Scan(&dataSetID) + if errors.Is(err, pgx.ErrNoRows) { + return true, nil + } + if err != nil { + return false, xerrors.Errorf("failed to query deletion drain row: %w", err) + } + + defer func() { + if err != nil { + log.Errorw("Removal queue draining failed", "dataSetId", dataSetID, "error", err) + } + }() + + verifier, err := contract.NewPDPVerifier(contract.ContractAddresses().PDPVerifier, p.ethClient) + if err != nil { + return false, xerrors.Errorf("failed to instantiate PDPVerifier contract: %w", err) + } + + queued, err := verifier.GetScheduledRemovals(contract.EthCallOpts(ctx), big.NewInt(dataSetID)) + if err != nil { + // A data set in deletion or cleanup is no longer live, so its queue is + // moot. Drop the row rather than burning retries against it. + if IsPDPVerifierDataSetNotFound(err) || IsPDPVerifierDataSetNotLive(err) { + if dropErr := p.dropDrainRow(ctx, dataSetID, taskID); dropErr != nil { + return false, dropErr + } + log.Infow("dropping removal drain for data set that is no longer live", "dataSetId", dataSetID) + return true, nil + } + return false, xerrors.Errorf("failed to read scheduled removals for data set %d: %w", dataSetID, err) + } + + if len(queued) == 0 { + if err := processPendingPieceDeletes(ctx, p.db, verifier, dataSetID, nil); err != nil { + return false, xerrors.Errorf("failed to reconcile drained piece deletes for data set %d: %w", dataSetID, err) + } + if dropErr := p.dropDrainRow(ctx, dataSetID, taskID); dropErr != nil { + return false, dropErr + } + return true, nil + } + + if !stillOwned() { + return false, nil + } + + fromAddress, _, err := verifier.GetDataSetStorageProvider(contract.EthCallOpts(ctx), big.NewInt(dataSetID)) + if err != nil { + return false, xerrors.Errorf("failed to get storage provider for data set %d: %w", dataSetID, err) + } + + pabi, err := contract.PDPVerifierMetaData.GetAbi() + if err != nil { + return false, xerrors.Errorf("failed to get PDPVerifier metadata: %w", err) + } + + txHash, batchSize, err := p.sendProcessPieceDeletions(ctx, pabi, fromAddress, dataSetID, len(queued)) + if err != nil { + return false, xerrors.Errorf("failed to send processPieceDeletions: %w", err) + } + + comm, err := p.db.BeginTransaction(ctx, func(tx *harmonydb.Tx) (commit bool, err error) { + n, err := tx.Exec(` + UPDATE pdpv0_deletion_drain + SET msg_hash = $1, task_id = NULL + WHERE task_id = $2 + `, txHash.Hex(), taskID) + if err != nil { + return false, xerrors.Errorf("failed to record drain message: %w", err) + } + if n != 1 { + return false, xerrors.Errorf("expected to update 1 deletion drain row, updated %d", n) + } + + _, err = tx.Exec(` + INSERT INTO message_waits_eth (signed_tx_hash, tx_status) + VALUES ($1, 'pending') ON CONFLICT DO NOTHING + `, txHash.Hex()) + if err != nil { + return false, xerrors.Errorf("failed to insert drain message wait: %w", err) + } + return true, nil + }, harmonydb.OptionRetry()) + if err != nil { + return false, xerrors.Errorf("failed to commit drain state: %w", err) + } + if !comm { + return false, xerrors.Errorf("failed to commit drain state") + } + + log.Infow("submitted PDP processPieceDeletions", + "dataSetId", dataSetID, + "txHash", txHash.Hex(), + "batchSize", batchSize, + "queueLength", len(queued)) + + return true, nil +} + +// sendProcessPieceDeletions submits one drain batch, halving on gas-estimate +// failure. SenderETH estimates gas before submission, so a failed estimate means +// nothing was sent and a smaller retry cannot double-drain. +func (p *ProcessDeletionsTask) sendProcessPieceDeletions(ctx context.Context, pabi *abi.ABI, from common.Address, dataSet int64, queueLength int) (common.Hash, int, error) { + batchSize := processDeletionsBatchSize + if queueLength < batchSize { + batchSize = queueLength + } + + dataSetID := big.NewInt(dataSet) + + for { + data, err := pabi.Pack("processPieceDeletions", dataSetID, big.NewInt(int64(batchSize))) + if err != nil { + return common.Hash{}, 0, err + } + + txEth := types.NewTransaction( + 0, + contract.ContractAddresses().PDPVerifier, + big.NewInt(0), + 0, + nil, + data, + ) + + txHash, err := p.sender.Send(ctx, from, txEth, reasonPDPProcessDeletions) + if err == nil { + return txHash, batchSize, nil + } + + if !isCleanupPiecesGasEstimateOutOfGas(err) { + return common.Hash{}, 0, err + } + if batchSize == 1 { + return common.Hash{}, 0, xerrors.Errorf("processPieceDeletions gas estimate failed at batch size 1: %w", err) + } + + next := batchSize / 2 + if next == 0 { + next = 1 + } + log.Warnw("processPieceDeletions gas estimate failed; retrying with smaller batch", + "dataSetId", dataSetID, "batchSize", batchSize, "nextBatchSize", next, "err", err) + batchSize = next + } +} + +func (p *ProcessDeletionsTask) dropDrainRow(ctx context.Context, dataSetID int64, taskID harmonytask.TaskID) error { + _, err := p.db.Exec(ctx, `DELETE FROM pdpv0_deletion_drain WHERE data_set = $1 AND task_id = $2`, dataSetID, taskID) + if err != nil { + return xerrors.Errorf("failed to drop deletion drain row for data set %d: %w", dataSetID, err) + } + return nil +} + +func (p *ProcessDeletionsTask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) ([]harmonytask.TaskID, error) { + return ids, nil +} + +func (p *ProcessDeletionsTask) TypeDetails() harmonytask.TaskTypeDetails { + return harmonytask.TaskTypeDetails{ + Max: taskhelp.Max(16), + Name: tasknames.PDPv0_ProcDel, + TimeSensitive: true, + Cost: resources.Resources{ + Cpu: 0, + Gpu: 0, + Ram: 1 << 20, + }, + MaxFailures: 3, + RetryWait: taskhelp.RetryWaitExp(5*time.Second, 2), + } +} + +func (p *ProcessDeletionsTask) Adder(taskFunc harmonytask.AddTaskFunc) { + p.addFunc.Set(taskFunc) +} + +// enqueueDeletionDrain records that a data set may have removals to drain. +// +// Called from the proving-period tasks when PDPVerifier reports pending +// deletions, so that a data set whose drain row was lost -- or which had +// removals scheduled out of band -- is picked up rather than retrying a +// rollover that cannot succeed. +func enqueueDeletionDrain(tx *harmonydb.Tx, dataSetID int64) error { + _, err := tx.Exec(`INSERT INTO pdpv0_deletion_drain (data_set) VALUES ($1) ON CONFLICT DO NOTHING`, dataSetID) + if err != nil { + return xerrors.Errorf("failed to enqueue removal drain for data set %d: %w", dataSetID, err) + } + return nil +} + +var _ harmonytask.TaskInterface = &ProcessDeletionsTask{} +var _ = harmonytask.Reg(&ProcessDeletionsTask{}) diff --git a/tasks/pdpv0/task_prove.go b/tasks/pdpv0/task_prove.go index 7284d7302..033d75240 100644 --- a/tasks/pdpv0/task_prove.go +++ b/tasks/pdpv0/task_prove.go @@ -249,7 +249,27 @@ func (p *ProveTask) Do(ctx context.Context, taskID harmonytask.TaskID, stillOwne return p.handleProvePreflightError(ctx, dataSetId, currentHeight, xerrors.Errorf("failed to get next challenge epoch: %w", err)) } - if challengeEpoch.Sign() == 0 { // if challengeEpoch is 0 (NO_CHALLENGE_SCHEDULED), we need to disable proving + if challengeEpoch.Sign() == 0 { // NO_CHALLENGE_SCHEDULED + // A zero challenge epoch has two causes since FilOzone/pdp#297, and they + // need opposite handling. Leaf count tells them apart. + leafCount, err := pdpVerifier.GetDataSetLeafCount(contract.EthCallOpts(ctx), big.NewInt(dataSetId)) + if err != nil { + return p.handleProvePreflightError(ctx, dataSetId, currentHeight, xerrors.Errorf("failed to get data set leaf count: %w", err)) + } + + if leafCount.Sign() > 0 { + // Removals were processed before this period's proof, invalidating + // the challenge. There is nothing to prove this period, but the data + // set is healthy and its proving schedule is still valid. Complete + // without a proof and leave the schedule intact: the prove watcher + // already cleared challenge_request_msg_hash when it claimed this + // task, so the nextProvingPeriod watcher picks the data set up at + // prove_at_epoch + challenge_window and samples a fresh challenge. + log.Warnw("skipping proof; challenge invalidated by processed piece deletions", + "dataSetId", dataSetId, "taskID", taskID, "leafCount", leafCount.String()) + return true, nil + } + log.Infow("disabling proving", "dataSetId", dataSetId, "taskID", taskID, "reason", "no challenge epoch") err = p.disableProving(ctx, dataSetId) if err != nil { diff --git a/tasks/pdpv0/task_reorg_check.go b/tasks/pdpv0/task_reorg_check.go index c066e61d8..7c97f8301 100644 --- a/tasks/pdpv0/task_reorg_check.go +++ b/tasks/pdpv0/task_reorg_check.go @@ -51,6 +51,9 @@ const ( reasonPDPTerminateDataSet = "pdp-terminate-data-set" ) +// reasonPDPProcessDeletions is declared in task_process_deletions.go alongside +// the task that sends it. + var pdpv0SendReasons = []string{ reasonPDPMkDataset, reasonPDPCreateAndAdd, @@ -61,6 +64,7 @@ var pdpv0SendReasons = []string{ reasonPDPProve, reasonPDPTerminateSvc, reasonPDPTerminateDataSet, + reasonPDPProcessDeletions, } // ReorgCheckFilAPI is the minimal Filecoin API needed for finality depth. @@ -446,6 +450,8 @@ func (t *ReorgCheckTask) rollbackByReasonTx(ctx context.Context, tx *harmonydb.T return t.rollbackAddPiecesTx(ctx, tx, txHash) case reasonPDPDeletePiece: return t.rollbackDeletePieceTx(ctx, tx, txHash) + case reasonPDPProcessDeletions: + return t.rollbackProcessDeletionsTx(ctx, tx, txHash) case reasonPDPProvingInit, reasonPDPProvingPeriod: return t.rollbackProvingPeriodTx(ctx, tx, txHash) case reasonPDPProve: @@ -551,6 +557,61 @@ func (t *ReorgCheckTask) rollbackDeletePieceTx(ctx context.Context, tx *harmonyd return fmt.Sprintf("deletePiece rollback log-only would_unmark_rows=%d piecerefs_already_missing=%d", n, lostRefs), nil } +// rollbackProcessDeletionsTx handles a reorged-out processPieceDeletions send. +// +// The removals did not apply, so the pieces are live and re-queued on-chain and +// any local removed=TRUE for them is wrong. Clearing the drain row's msg_hash is +// safe and sufficient to resume: the drain task re-reads the on-chain queue and +// will process whatever is actually there. +// +// Unmarking pieces is deliberately log-only, matching rollbackDeletePieceTx. +// That is also where the real hazard is: piece GC keys off removed=TRUE and +// deletes the underlying data, so if it has already run the local state cannot +// be repaired regardless of what this does. +func (t *ReorgCheckTask) rollbackProcessDeletionsTx(ctx context.Context, tx *harmonydb.Tx, txHash string) (string, error) { + var dataSets []struct { + DataSet int64 `db:"data_set"` + } + err := tx.Select(&dataSets, `SELECT data_set FROM pdpv0_deletion_drain WHERE msg_hash = $1`, txHash) + if err != nil { + return "", err + } + + n, err := tx.Exec(` + UPDATE pdpv0_deletion_drain + SET msg_hash = NULL + WHERE msg_hash = $1`, txHash) + if err != nil { + return "", err + } + + lostRefs := 0 + for _, ds := range dataSets { + var cnt int + err = tx.QueryRow(` + SELECT COUNT(*) FROM pdp_data_set_pieces p + WHERE p.data_set = $1 + AND p.removed = TRUE + AND NOT EXISTS (SELECT 1 FROM pdp_piecerefs r WHERE r.id = p.pdp_pieceref)`, ds.DataSet).Scan(&cnt) + if err != nil { + return "", err + } + lostRefs += cnt + } + + if err := t.markWaitReorged(ctx, tx, txHash); err != nil { + return "", err + } + + logReorgCheck.Warnw("reorg processPieceDeletions rollback cleared drain message", + "tx", txHash, "drain_rows_reset", n, "data_sets", len(dataSets)) + if lostRefs > 0 { + logReorgCheck.Errorw("reorg processPieceDeletions rollback found removed pieces whose pieceref data is already cleaned up — possible DATA LOSS", + "tx", txHash, "lost_piecerefs", lostRefs) + } + return fmt.Sprintf("processPieceDeletions rollback drain_rows_reset=%d piecerefs_already_missing=%d", n, lostRefs), nil +} + func (t *ReorgCheckTask) rollbackProvingPeriodTx(ctx context.Context, tx *harmonydb.Tx, txHash string) (string, error) { if err := t.markWaitReorged(ctx, tx, txHash); err != nil { return "", err diff --git a/tasks/pdpv0/watch_process_deletions.go b/tasks/pdpv0/watch_process_deletions.go new file mode 100644 index 000000000..6f9cd8e8d --- /dev/null +++ b/tasks/pdpv0/watch_process_deletions.go @@ -0,0 +1,335 @@ +package pdpv0 + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "math/big" + + "github.com/ethereum/go-ethereum/core/types" + "golang.org/x/xerrors" + + "github.com/filecoin-project/curio/alertmanager/curioalerting" + "github.com/filecoin-project/curio/harmony/harmonydb" + "github.com/filecoin-project/curio/lib/ethchain" + "github.com/filecoin-project/curio/pdp/contract" + + chainTypes "github.com/filecoin-project/lotus/chain/types" +) + +const deletionDrainReconcileBatchLimit = 128 + +// NewProcessDeletionsWatcher reconciles confirmed processPieceDeletions +// transactions. +// +// Before FilOzone/pdp#297, nextProvingPeriod applied scheduled removals, so +// local removal tracking was reconciled off a confirmed nextProvingPeriod +// message. Removals are now applied by their own transaction, so that is the +// trigger. The watcher uses the PiecesRemoved event as the candidate set, then +// still verifies chain state before marking local pieces removed. +func NewProcessDeletionsWatcher(w *Watcher) { + if err := w.AddWatcher(func(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient, al curioalerting.AlertingInterface, revert, apply *chainTypes.TipSet) { + if err := reconcileDrainMessages(ctx, db, ethClient); err != nil { + log.Warnf("Failed to reconcile PDP removal drain messages: %s", err) + _ = al.EmitEvent(ctx, curioalerting.AlertEvent{ + System: alertType, + Subsystem: alertNameProcessDeletions, + Message: fmt.Sprintf("failed to reconcile PDP removal drain messages: %s", err), + }) + return + } + }, WatcherOrderProcessDeletions); err != nil { + panic(err) + } +} + +type drainMessage struct { + DataSetID int64 `db:"data_set"` + TxHash string `db:"msg_hash"` + TxStatus sql.NullString `db:"tx_status"` + TxSuccess sql.NullBool `db:"tx_success"` + TxReceipt []byte `db:"tx_receipt"` +} + +// reconcileDrainMessages handles in-flight drain messages that have reached a +// final state. +// +// Clearing msg_hash is what lets the drain watcher pick the data set up again: +// one drain transaction handles at most one batch, so a deep queue needs several +// passes. The local piece reconciliation must happen before that clear, so a +// process restart can replay from the in-flight drain row. +func reconcileDrainMessages(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient) error { + var messages []drainMessage + err := db.Select(ctx, &messages, ` + SELECT d.data_set, + d.msg_hash, + mwe.tx_status, + mwe.tx_success, + mwe.tx_receipt + FROM pdpv0_deletion_drain d + LEFT JOIN message_waits_eth mwe ON mwe.signed_tx_hash = d.msg_hash + WHERE d.msg_hash IS NOT NULL + ORDER BY d.data_set + LIMIT $1 + `, deletionDrainReconcileBatchLimit) + if err != nil { + return xerrors.Errorf("failed to select in-flight removal drains: %w", err) + } + + if len(messages) == 0 { + return nil + } + + pdpVerifier, err := contract.NewPDPVerifier(contract.ContractAddresses().PDPVerifier, ethClient) + if err != nil { + return xerrors.Errorf("failed to instantiate PDPVerifier contract: %w", err) + } + + pdpABI, err := contract.PDPVerifierMetaData.GetAbi() + if err != nil { + return xerrors.Errorf("failed to get PDP ABI: %w", err) + } + + event, exists := pdpABI.Events["PiecesRemoved"] + if !exists { + return xerrors.Errorf("PiecesRemoved event not found in ABI") + } + + pdpVerifierAddress := contract.ContractAddresses().PDPVerifier + for _, msg := range messages { + if !msg.TxStatus.Valid || msg.TxStatus.String != "confirmed" { + continue + } + + success := msg.TxSuccess.Valid && msg.TxSuccess.Bool + if !success { + // The drain did not apply. Clear the message so the data set is + // retried rather than stalling behind a dead transaction. + log.Errorw("PDP removal drain transaction failed", "dataSetId", msg.DataSetID, "txHash", msg.TxHash) + if _, err := db.Exec(ctx, ` + UPDATE pdpv0_deletion_drain + SET msg_hash = NULL WHERE data_set = $1 AND msg_hash = $2 + `, msg.DataSetID, msg.TxHash); err != nil { + return xerrors.Errorf("failed to clear failed removal drain %s: %w", msg.TxHash, err) + } + continue + } + + var txReceipt types.Receipt + err = json.Unmarshal(msg.TxReceipt, &txReceipt) + if err != nil { + return xerrors.Errorf("failed to unmarshal tx_receipt for tx %s: %w", msg.TxHash, err) + } + + var removedPieceIDs []int64 + for _, vLog := range txReceipt.Logs { + if vLog.Address == pdpVerifierAddress && len(vLog.Topics) > 0 && vLog.Topics[0] == event.ID { + prv, err := pdpVerifier.ParsePiecesRemoved(*vLog) + if err != nil { + return xerrors.Errorf("failed to parse PDP removal event: %w", err) + } + if !prv.SetId.IsInt64() { + return xerrors.Errorf("PiecesRemoved set id %s does not fit in int64", prv.SetId.String()) + } + eventDataSetID := prv.SetId.Int64() + if eventDataSetID != msg.DataSetID { + return xerrors.Errorf("PiecesRemoved event data set %d does not match drain row data set %d for tx %s", eventDataSetID, msg.DataSetID, msg.TxHash) + } + + pids := make([]int64, len(prv.PieceIds)) + for i, pid := range prv.PieceIds { + if !pid.IsInt64() { + return xerrors.Errorf("PiecesRemoved piece id %s does not fit in int64 for tx %s", pid.String(), msg.TxHash) + } + pids[i] = pid.Int64() + } + + removedPieceIDs = append(removedPieceIDs, pids...) + } + } + + if len(removedPieceIDs) == 0 { + return xerrors.Errorf("confirmed removal drain %s for data set %d had no PiecesRemoved event", msg.TxHash, msg.DataSetID) + } + + if err := processPendingPieceDeletes(ctx, db, pdpVerifier, msg.DataSetID, removedPieceIDs); err != nil { + return xerrors.Errorf("failed to process pending piece deletes for drain %s: %w", msg.TxHash, err) + } + + if _, err := db.Exec(ctx, ` + UPDATE pdpv0_deletion_drain + SET msg_hash = NULL WHERE data_set = $1 AND msg_hash = $2 + `, msg.DataSetID, msg.TxHash); err != nil { + return xerrors.Errorf("failed to clear confirmed removal drain %s: %w", msg.TxHash, err) + } + + log.Infow("PDP removal drain confirmed", "dataSetId", msg.DataSetID, "txHash", msg.TxHash, "piecesRemoved", len(removedPieceIDs)) + } + + return nil +} + +type pendingPieceDelete struct { + DataSetID int64 `db:"data_set"` + PieceID int64 `db:"piece_id"` + TxHash string `db:"rm_message_hash"` + TxStatus sql.NullString `db:"tx_status"` + TxSuccess sql.NullBool `db:"tx_success"` +} + +// processPendingPieceDeletes reconciles local piece-removal rows against +// PDPVerifier. When candidatePieceIDs is non-nil, it is the PiecesRemoved event +// payload from a confirmed processPieceDeletions transaction and limits the rows +// considered. When candidatePieceIDs is nil, all pending rows for the data set +// are reconciled against chain state. In both modes a confirmed +// schedulePieceDeletions transaction only records delete intent, and the piece +// must not be marked removed locally while PDPVerifier still reports it as +// scheduled or live, because piece GC keys off that flag and deletes the +// underlying data. +func processPendingPieceDeletes(ctx context.Context, db *harmonydb.DB, verifier *contract.PDPVerifier, dataSetID int64, candidatePieceIDs []int64) error { + var pendingDeletes []pendingPieceDelete + err := db.Select(ctx, &pendingDeletes, ` + SELECT psp.data_set, + psp.piece_id, + psp.rm_message_hash, + mwe.tx_status, + mwe.tx_success + FROM pdp_data_set_pieces psp + LEFT JOIN message_waits_eth mwe ON mwe.signed_tx_hash = psp.rm_message_hash + WHERE psp.data_set = $1 + AND psp.rm_message_hash IS NOT NULL + AND psp.removed = FALSE + ORDER BY psp.data_set, psp.piece_id + `, dataSetID) + if err != nil { + return xerrors.Errorf("failed to select pending piece deletes: %w", err) + } + if len(pendingDeletes) == 0 { + return nil + } + + var candidateSet map[int64]struct{} + if candidatePieceIDs != nil { + candidateSet = make(map[int64]struct{}, len(candidatePieceIDs)) + for _, pieceID := range candidatePieceIDs { + candidateSet[pieceID] = struct{}{} + } + } + + var scheduled map[int64]struct{} + var loadedScheduled bool + for _, piece := range pendingDeletes { + // Wait until the schedulePieceDeletions send has a final watcher result. + if !piece.TxStatus.Valid || piece.TxStatus.String != "confirmed" { + continue + } + // A confirmed row without tx_success is malformed for our purposes; clear + // the local delete intent so operators can resubmit cleanly. + if !piece.TxSuccess.Valid { + log.Errorf("invalid message_waits_eth state for piece delete tx %s", piece.TxHash) + if err := clearPendingPieceDelete(ctx, db, piece); err != nil { + return err + } + continue + } + // The schedule transaction failed, so no on-chain removal is pending. + if !piece.TxSuccess.Bool { + log.Errorf("failed to process pending piece delete as transaction %s failed", piece.TxHash) + if err := clearPendingPieceDelete(ctx, db, piece); err != nil { + return err + } + continue + } + + if candidateSet != nil { + if _, ok := candidateSet[piece.PieceID]; !ok { + continue + } + } + + if !loadedScheduled { + scheduled, err = getScheduledRemovalSet(ctx, verifier, dataSetID) + if err != nil { + return err + } + loadedScheduled = true + } + // Still scheduled means the removal has not been processed yet. Keep + // rm_message_hash set and leave removed=false. + if _, ok := scheduled[piece.PieceID]; ok { + continue + } + + // Once it is no longer scheduled, PieceLive is the final authority for + // whether the removal actually applied. + pieceID := big.NewInt(piece.PieceID) + live, err := verifier.PieceLive(contract.EthCallOpts(ctx), big.NewInt(dataSetID), pieceID) + if err != nil { + return xerrors.Errorf("failed to check if piece is live: %w", err) + } + if !live { + if err := markPendingPieceRemoved(ctx, db, piece); err != nil { + return err + } + log.Infow("piece removed on-chain, marking as removed in DB", "dataSetId", piece.DataSetID, "pieceID", piece.PieceID, "txHash", piece.TxHash) + continue + } + + log.Warnw("piece is live and not scheduled despite successful delete tx; clearing stale delete tracking", + "dataSetId", piece.DataSetID, "pieceID", piece.PieceID, "txHash", piece.TxHash) + if err := clearPendingPieceDelete(ctx, db, piece); err != nil { + return err + } + } + + return nil +} + +func getScheduledRemovalSet(ctx context.Context, verifier *contract.PDPVerifier, dataSetID int64) (map[int64]struct{}, error) { + removals, err := verifier.GetScheduledRemovals(contract.EthCallOpts(ctx), big.NewInt(dataSetID)) + if err != nil { + return nil, xerrors.Errorf("failed to get scheduled removals: %w", err) + } + + out := make(map[int64]struct{}, len(removals)) + for _, removal := range removals { + if removal.IsInt64() { + out[removal.Int64()] = struct{}{} + } + } + return out, nil +} + +func clearPendingPieceDelete(ctx context.Context, db *harmonydb.DB, piece pendingPieceDelete) error { + _, err := db.Exec(ctx, ` + UPDATE pdp_data_set_pieces + SET rm_message_hash = NULL + WHERE data_set = $1 + AND piece_id = $2 + AND rm_message_hash = $3 + AND removed = FALSE + `, piece.DataSetID, piece.PieceID, piece.TxHash) + if err != nil { + return xerrors.Errorf("failed to clear pending piece delete %s: %w", piece.TxHash, err) + } + return nil +} + +func markPendingPieceRemoved(ctx context.Context, db *harmonydb.DB, piece pendingPieceDelete) error { + affected, err := db.Exec(ctx, ` + UPDATE pdp_data_set_pieces + SET removed = TRUE + WHERE data_set = $1 + AND piece_id = $2 + AND rm_message_hash = $3 + AND removed = FALSE + `, piece.DataSetID, piece.PieceID, piece.TxHash) + if err != nil { + return xerrors.Errorf("failed to mark piece removed: %w", err) + } + if affected > 1 { + return xerrors.Errorf("expected to update at most 1 piece delete row, updated %d", affected) + } + return nil +} diff --git a/tasks/pdpv0/watch_proving_period.go b/tasks/pdpv0/watch_proving_period.go index 1723f0176..566d29354 100644 --- a/tasks/pdpv0/watch_proving_period.go +++ b/tasks/pdpv0/watch_proving_period.go @@ -6,6 +6,7 @@ import ( "fmt" "math/big" + "golang.org/x/exp/maps" "golang.org/x/xerrors" "github.com/filecoin-project/curio/alertmanager/curioalerting" @@ -66,12 +67,22 @@ func NewProvingPeriodWatcher(w *Watcher) { return } - if err := processPendingPieceDeletes(ctx, db, ethClient, readyDataSets); err != nil { - log.Warnf("Failed to process pending PDP piece deletes: %s", err) + if err := reconcilePendingPieceDeletesFromChain(ctx, db, ethClient, readyDataSets); err != nil { + log.Warnf("Failed to reconcile pending PDP piece deletes: %s", err) _ = al.EmitEvent(ctx, curioalerting.AlertEvent{ System: alertType, Subsystem: alertNameProvingPeriod, - Message: fmt.Sprintf("failed to process pending PDP piece deletes: %s", err), + Message: fmt.Sprintf("failed to reconcile pending PDP piece deletes: %s", err), + }) + return + } + + if err := enqueueDeletionDrainsForPendingDeletes(ctx, db, ethClient, readyDataSets); err != nil { + log.Warnf("Failed to enqueue PDP deletion drains: %s", err) + _ = al.EmitEvent(ctx, curioalerting.AlertEvent{ + System: alertType, + Subsystem: alertNameProvingPeriod, + Message: fmt.Sprintf("failed to enqueue PDP deletion drains: %s", err), }) return } @@ -186,9 +197,14 @@ func clearFailedProvingPeriodReconciliations(ctx context.Context, db *harmonydb. } // processEmptyProvingPeriods reconciles datasets whose confirmed initPP/nextPP -// message left no next challenge on-chain. This happens when nextProvingPeriod -// removes the final piece: PDPVerifier clears the challenge, but Curio may have -// already stored the now-stale prove_at_epoch. +// message left no next challenge on-chain. This happens when the final piece is +// removed: PDPVerifier clears the challenge, but Curio may have already stored +// the now-stale prove_at_epoch. +// +// A zero next challenge epoch is ambiguous since FilOzone/pdp#297: +// processPieceDeletions also clears it, on a data set that may still have +// leaves and only needs a fresh proving-period transition. Leaf count is the +// discriminator -- only a zero challenge with zero leaves is genuinely empty. func processEmptyProvingPeriods(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient, periods []confirmedProvingPeriod) error { if len(periods) == 0 { return nil @@ -218,18 +234,22 @@ func processEmptyProvingPeriods(ctx context.Context, db *harmonydb.DB, ethClient if err != nil { return xerrors.Errorf("failed to get leaf count for data set %d: %w", period.DataSetID, err) } - initReady := leafCount.Sign() > 0 + if leafCount.Sign() > 0 { + log.Debugw("skipping empty-period reset; challenge cleared but data set still has leaves", + "dataSetId", period.DataSetID, "leafCount", leafCount.String()) + continue + } affected, err := db.Exec(ctx, ` UPDATE pdp_data_sets SET challenge_request_msg_hash = NULL, prove_at_epoch = NULL, prev_challenge_request_epoch = NULL, - init_ready = $3 + init_ready = FALSE WHERE id = $1 AND challenge_request_msg_hash = $2 AND unrecoverable_proving_failure_epoch IS NULL - `, period.DataSetID, period.TxHash.String, initReady) + `, period.DataSetID, period.TxHash.String) if err != nil { return xerrors.Errorf("failed to reset empty proving period for data set %d: %w", period.DataSetID, err) } @@ -239,9 +259,7 @@ func processEmptyProvingPeriods(ctx context.Context, db *harmonydb.DB, ethClient if affected == 1 { log.Infow("reset empty proving period", "dataSetId", period.DataSetID, - "txHash", period.TxHash.String, - "leafCount", leafCount.String(), - "initReady", initReady) + "txHash", period.TxHash.String) } } @@ -263,36 +281,48 @@ func clearProvingPeriodReconcileNeeded(ctx context.Context, db *harmonydb.DB, pe return nil } -type pendingPieceDelete struct { - DataSetID int64 `db:"data_set"` - PieceID int64 `db:"piece_id"` - TxHash string `db:"rm_message_hash"` - TxStatus sql.NullString `db:"tx_status"` - TxSuccess sql.NullBool `db:"tx_success"` +func reconcilePendingPieceDeletesFromChain(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient, dataSets []int64) error { + if len(dataSets) == 0 { + return nil + } + + verifier, err := contract.NewPDPVerifier(contract.ContractAddresses().PDPVerifier, ethClient) + if err != nil { + return xerrors.Errorf("failed to instantiate PDPVerifier contract: %w", err) + } + + for _, dataSetID := range dataSets { + if err := processPendingPieceDeletes(ctx, db, verifier, dataSetID, nil); err != nil { + return xerrors.Errorf("failed to reconcile pending piece deletes for data set %d: %w", dataSetID, err) + } + } + + return nil +} + +type pendingPieceDeleteDrain struct { + DataSetID int64 `db:"data_set"` + PieceID int64 `db:"piece_id"` } -// processPendingPieceDeletes reconciles local piece-removal rows after -// nextProvingPeriod has applied scheduled removals on-chain. A confirmed -// schedulePieceDeletions transaction only records delete intent; the piece -// should not be marked removed locally while PDPVerifier still reports it as -// scheduled or live. -func processPendingPieceDeletes(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient, dataSets []int64) error { - var pendingDeletes []pendingPieceDelete +// enqueueDeletionDrainsForPendingDeletes records datasets whose confirmed +// delete intent is still present in PDPVerifier's scheduled-removal queue. +func enqueueDeletionDrainsForPendingDeletes(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient, dataSets []int64) error { + var pendingDeletes []pendingPieceDeleteDrain err := db.Select(ctx, &pendingDeletes, ` SELECT psp.data_set, - psp.piece_id, - psp.rm_message_hash, - mwe.tx_status, - mwe.tx_success + psp.piece_id FROM pdp_data_set_pieces psp - LEFT JOIN message_waits_eth mwe ON mwe.signed_tx_hash = psp.rm_message_hash + INNER JOIN message_waits_eth mwe ON mwe.signed_tx_hash = psp.rm_message_hash WHERE psp.data_set = ANY($1::bigint[]) AND psp.rm_message_hash IS NOT NULL AND psp.removed = FALSE + AND mwe.tx_status = 'confirmed' + AND mwe.tx_success = TRUE ORDER BY psp.data_set, psp.piece_id `, dataSets) if err != nil { - return xerrors.Errorf("failed to select pending piece deletes: %w", err) + return xerrors.Errorf("failed to select pending deletion drains: %w", err) } if len(pendingDeletes) == 0 { return nil @@ -304,29 +334,8 @@ func processPendingPieceDeletes(ctx context.Context, db *harmonydb.DB, ethClient } scheduledByDataSet := map[int64]map[int64]struct{}{} + drainQueuedByDataSet := map[int64]struct{}{} for _, piece := range pendingDeletes { - // Wait until the schedulePieceDeletions send has a final watcher result. - if !piece.TxStatus.Valid || piece.TxStatus.String != "confirmed" { - continue - } - // A confirmed row without tx_success is malformed for our purposes; clear - // the local delete intent so operators can resubmit cleanly. - if !piece.TxSuccess.Valid { - log.Errorf("invalid message_waits_eth state for piece delete tx %s", piece.TxHash) - if err := clearPendingPieceDelete(ctx, db, piece); err != nil { - return err - } - continue - } - // The schedule transaction failed, so no on-chain removal is pending. - if !piece.TxSuccess.Bool { - log.Errorf("failed to process pending piece delete as transaction %s failed", piece.TxHash) - if err := clearPendingPieceDelete(ctx, db, piece); err != nil { - return err - } - continue - } - scheduled, ok := scheduledByDataSet[piece.DataSetID] if !ok { scheduled, err = getScheduledRemovalSet(ctx, verifier, piece.DataSetID) @@ -335,81 +344,38 @@ func processPendingPieceDeletes(ctx context.Context, db *harmonydb.DB, ethClient } scheduledByDataSet[piece.DataSetID] = scheduled } - // Still scheduled means nextProvingPeriod has not processed the removal - // yet. Keep rm_message_hash set and leave removed=false. if _, ok := scheduled[piece.PieceID]; ok { - continue - } - - // Once it is no longer scheduled, PieceLive is the final authority for - // whether nextProvingPeriod actually removed it. - pieceID := big.NewInt(piece.PieceID) - live, err := verifier.PieceLive(contract.EthCallOpts(ctx), big.NewInt(piece.DataSetID), pieceID) - if err != nil { - return xerrors.Errorf("failed to check if piece is live: %w", err) - } - if !live { - if err := markPendingPieceRemoved(ctx, db, piece); err != nil { - return err + if _, queued := drainQueuedByDataSet[piece.DataSetID]; !queued { + if err := enqueueDeletionDrainFromProvingWatcher(ctx, db, piece.DataSetID); err != nil { + return err + } + drainQueuedByDataSet[piece.DataSetID] = struct{}{} } - log.Infow("piece removed on-chain, marking as removed in DB", "dataSetId", piece.DataSetID, "pieceID", piece.PieceID, "txHash", piece.TxHash) continue } - - log.Warnw("piece is live and not scheduled despite successful delete tx; clearing stale delete tracking", - "dataSetId", piece.DataSetID, "pieceID", piece.PieceID, "txHash", piece.TxHash) - if err := clearPendingPieceDelete(ctx, db, piece); err != nil { - return err - } } - return nil -} + ds := maps.Keys(drainQueuedByDataSet) -func getScheduledRemovalSet(ctx context.Context, verifier *contract.PDPVerifier, dataSetID int64) (map[int64]struct{}, error) { - removals, err := verifier.GetScheduledRemovals(contract.EthCallOpts(ctx), big.NewInt(dataSetID)) - if err != nil { - return nil, xerrors.Errorf("failed to get scheduled removals: %w", err) + if len(ds) > 0 { + log.Infow("Scheduled Deletion Process", "dataset", maps.Keys(drainQueuedByDataSet)) } - out := make(map[int64]struct{}, len(removals)) - for _, removal := range removals { - if removal.IsInt64() { - out[removal.Int64()] = struct{}{} - } - } - return out, nil -} - -func clearPendingPieceDelete(ctx context.Context, db *harmonydb.DB, piece pendingPieceDelete) error { - _, err := db.Exec(ctx, ` - UPDATE pdp_data_set_pieces - SET rm_message_hash = NULL - WHERE data_set = $1 - AND piece_id = $2 - AND rm_message_hash = $3 - AND removed = FALSE - `, piece.DataSetID, piece.PieceID, piece.TxHash) - if err != nil { - return xerrors.Errorf("failed to clear pending piece delete %s: %w", piece.TxHash, err) - } return nil } -func markPendingPieceRemoved(ctx context.Context, db *harmonydb.DB, piece pendingPieceDelete) error { - affected, err := db.Exec(ctx, ` - UPDATE pdp_data_set_pieces - SET removed = TRUE - WHERE data_set = $1 - AND piece_id = $2 - AND rm_message_hash = $3 - AND removed = FALSE - `, piece.DataSetID, piece.PieceID, piece.TxHash) +func enqueueDeletionDrainFromProvingWatcher(ctx context.Context, db *harmonydb.DB, dataSetID int64) error { + committed, err := db.BeginTransaction(ctx, func(tx *harmonydb.Tx) (commit bool, err error) { + if err := enqueueDeletionDrain(tx, dataSetID); err != nil { + return false, err + } + return true, nil + }, harmonydb.OptionRetry()) if err != nil { - return xerrors.Errorf("failed to mark piece removed: %w", err) + return xerrors.Errorf("failed to enqueue removal drain from proving watcher: %w", err) } - if affected > 1 { - return xerrors.Errorf("expected to update at most 1 piece delete row, updated %d", affected) + if !committed { + return xerrors.Errorf("failed to commit removal drain enqueue for data set %d", dataSetID) } return nil } diff --git a/tasks/pdpv0/watcher.go b/tasks/pdpv0/watcher.go index d327352b7..f74bb2cc5 100644 --- a/tasks/pdpv0/watcher.go +++ b/tasks/pdpv0/watcher.go @@ -24,6 +24,7 @@ const ( WatcherOrderPaymentSettle WatcherOrderDelete WatcherOrderCleanupPieces + WatcherOrderProcessDeletions WatcherOrderProving ) @@ -34,6 +35,7 @@ var watcherOrders = []WatcherOrder{ WatcherOrderPaymentSettle, WatcherOrderDelete, WatcherOrderCleanupPieces, + WatcherOrderProcessDeletions, WatcherOrderProving, } diff --git a/tasks/tasknames/names.go b/tasks/tasknames/names.go index 5fb100622..5f607d67c 100644 --- a/tasks/tasknames/names.go +++ b/tasks/tasknames/names.go @@ -110,4 +110,5 @@ const ( PDPv0_ChainSync = "PDPv0_ChainSync" PDPv0_DatasetIdx = "PDPv0_DatasetIdx" PDPv0_FixPiece = "PDPv0_FixPiece" + PDPv0_ProcDel = "PDPv0_ProcDel" )