task_init_pp.go

package pdpv0

import (
	"context"
	"errors"
	"fmt"
	"math/big"
	"strings"
	"time"

	"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 alertNameInitPP = "InitProvingPeriod"

type InitProvingPeriodTask struct {
	db        *harmonydb.DB
	ethClient ethchain.EthClient
	sender    *message.SenderETH

	fil NextProvingPeriodTaskChainApi

	al curioalerting.AlertingInterface

	addFunc promise.Promise[harmonytask.AddTaskFunc]
}

type InitProvingPeriodTaskChainApi interface {
	ChainHead(context.Context) (*chainTypes.TipSet, error)
}

func NewInitProvingPeriodTask(db *harmonydb.DB, ethClient ethchain.EthClient, fil NextProvingPeriodTaskChainApi, w *Watcher, sender *message.SenderETH) *InitProvingPeriodTask {
	ipp := &InitProvingPeriodTask{
		db:        db,
		ethClient: ethClient,
		sender:    sender,
		fil:       fil,
		al:        w.al,
	}

	_ = w.AddWatcher(func(ctx context.Context, db *harmonydb.DB, ethClient ethchain.EthClient, al curioalerting.AlertingInterface, revert, apply *chainTypes.TipSet) {
		if apply == nil {
			return
		}

		// Now query the db for data sets needing nextProvingPeriod inital call
		var toCallInit []struct {
			DataSetId int64 `db:"id"`
		}

		err := db.Select(ctx, &toCallInit, `
                SELECT id
                FROM pdp_data_sets
                WHERE challenge_request_task_id IS NULL
                  AND init_ready AND prove_at_epoch IS NULL
                  AND unrecoverable_proving_failure_epoch IS NULL
	            `)
		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
			_ = al.EmitEvent(ctx, curioalerting.AlertEvent{
				System:    alertType,
				Subsystem: alertNameInitPP,
				Message:   fmt.Sprintf("failed to select data sets needing initProvingPeriod: %s", err),
			})
			return
		}

		for _, ps := range toCallInit {
			ipp.addFunc.Val(ctx)(func(id harmonytask.TaskID, tx *harmonydb.Tx) (shouldCommit bool, seriousError error) {
				// Update pdp_data_sets to set challenge_request_task_id = id
				affected, err := tx.Exec(`
                        UPDATE pdp_data_sets
                        SET challenge_request_task_id = $1
                        WHERE id = $2 AND challenge_request_task_id IS NULL
                    `, id, ps.DataSetId)
				if err != nil {
					return false, xerrors.Errorf("failed to update pdp_data_sets: %w", err)
				}
				if affected == 0 {
					// Someone else might have already scheduled the task
					return false, nil
				}

				return true, nil
			})
		}
	}, WatcherOrderProving)

	return ipp
}

func (ipp *InitProvingPeriodTask) Do(ctx context.Context, taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error) {
	// Select the data set where challenge_request_task_id = taskID
	var dataSetId int64

	err = ipp.db.QueryRow(ctx, `
        SELECT id
        FROM pdp_data_sets
        WHERE challenge_request_task_id = $1
    `, taskID).Scan(&dataSetId)
	if errors.Is(err, pgx.ErrNoRows) {
		// No matching data set, task is done (something weird happened, and e.g another task was spawned in place of this one)
		return true, nil
	}
	if err != nil {
		return false, xerrors.Errorf("failed to query pdp_data_sets: %w", err)
	}

	defer func() {
		if err != nil {
			log.Errorw("Initial challenge window scheduling failed", "dataSetId", dataSetId, "error", err)
			err = fmt.Errorf("failed to set up initial proving period for dataset %d: %w", dataSetId, err)
		}
	}()

	// 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 {
		return false, xerrors.Errorf("failed to instantiate PDPVerifier contract: %w", err)
	}

	// Check if the data set has any leaves (pieces) before attempting to initialize proving period
	leafCount, err := pdpVerifier.GetDataSetLeafCount(contract.EthCallOpts(ctx), big.NewInt(dataSetId))
	if err != nil {
		return false, xerrors.Errorf("failed to get leaf count for data set %d: %w", dataSetId, err)
	}
	if leafCount.Cmp(big.NewInt(0)) == 0 {
		// No leaves in the data set, we cannot prove anything
		// Initialization is only triggered when thre are leaves (after add piece lands), or we strongly suspect that there are
		// So we disable proving for this dataset if we end up having no leaves
		log.Warnw("Initial challange window scheduling skipped", "dataSetId", dataSetId, "reason", "no leaves")
		_, err = ipp.db.Exec(ctx, `
			UPDATE pdp_data_sets
			SET init_ready = FALSE
			WHERE id = $1
			`, dataSetId)
		if err != nil {
			return false, xerrors.Errorf("failed to disable proving (caused by no leaves): %w", err)
		}

		return true, nil
	}

	listenerAddr, err := pdpVerifier.GetDataSetListener(contract.EthCallOpts(ctx), big.NewInt(dataSetId))
	if err != nil {
		return false, xerrors.Errorf("failed to get listener address for data set %d: %w", dataSetId, err)
	}

	// Get the proving schedule from the listener (handles view contract indirection)
	provingSchedule, err := contract.GetProvingScheduleFromListener(ctx, listenerAddr, ipp.ethClient)
	if err != nil {
		return false, xerrors.Errorf("failed to get proving schedule from listener: %w", err)
	}

	config, err := provingSchedule.GetPDPConfig(contract.EthCallOpts(ctx))
	if err != nil {
		return false, xerrors.Errorf("failed to GetPDPConfig: %w", err)
	}

	init_prove_at := config.InitChallengeWindowStart.Add(config.InitChallengeWindowStart, config.ChallengeWindow.Div(config.ChallengeWindow, big.NewInt(2))) // Give a buffer of 1/2 challenge window epochs so that we are still within challenge window
	// Instantiate the PDPVerifier contract
	pdpContracts := contract.ContractAddresses()
	pdpVeriferAddress := pdpContracts.PDPVerifier

	// Prepare the transaction data
	abiData, err := contract.PDPVerifierMetaData.GetAbi()
	if err != nil {
		return false, xerrors.Errorf("failed to get PDPVerifier ABI: %w", err)
	}

	data, err := abiData.Pack("nextProvingPeriod", big.NewInt(dataSetId), init_prove_at, []byte{})
	if err != nil {
		return false, xerrors.Errorf("failed to pack data: %w", err)
	}

	// Prepare the transaction
	txEth := types.NewTransaction(
		0,                 // nonce (will be set by sender)
		pdpVeriferAddress, // to
		big.NewInt(0),     // value
		0,                 // gasLimit (to be estimated)
		nil,               // gasPrice (to be set by sender)
		data,              // data
	)

	if !stillOwned() {
		// Task was abandoned, don't send the transaction
		return false, nil
	}

	fromAddress, _, err := pdpVerifier.GetDataSetStorageProvider(contract.EthCallOpts(ctx), big.NewInt(dataSetId))
	if err != nil {
		return false, xerrors.Errorf("failed to get default sender address: %w", err)
	}

	// Get the current tipset
	ts, err := ipp.fil.ChainHead(ctx)
	if err != nil {
		return false, xerrors.Errorf("failed to get chain head: %w", err)
	}

	// Send the transaction
	reason := "pdp-proving-init"
	txHash, sendErr := ipp.sender.Send(ctx, fromAddress, txEth, reason)
	if sendErr != nil {
		currentHeight := int64(ts.Height())
		comm, err := ipp.db.BeginTransaction(ctx, func(tx *harmonydb.Tx) (commit bool, err error) {
			handleErr := handleNextProvingPeriodSendError(ctx, tx, provingSchedule, ipp.al, alertNameInitPP, dataSetId, currentHeight, sendErr)
			if handleErr != nil {
				return false, xerrors.Errorf("failed to handle proving send error: %w", handleErr)
			}
			return true, nil
		}, harmonydb.OptionRetry())
		if err != nil {
			return false, xerrors.Errorf("failed to send transaction: %w", err)
		}
		if !comm {
			return false, xerrors.Errorf("failed to commit transaction")
		}
		return true, nil
	}

	txHashLower := strings.ToLower(txHash.Hex())

	// Update the database in a transaction
	_, err = ipp.db.BeginTransaction(ctx, func(tx *harmonydb.Tx) (bool, error) {
		// Update pdp_data_sets
		affected, err := tx.Exec(`
            UPDATE pdp_data_sets
            SET challenge_request_msg_hash = $1,
                prev_challenge_request_epoch = $2,
				prove_at_epoch = $3
            WHERE id = $4
        `, txHashLower, ts.Height(), init_prove_at.Uint64(), dataSetId)
		if err != nil {
			return false, xerrors.Errorf("failed to update pdp_data_sets: %w", err)
		}
		if affected == 0 {
			return false, xerrors.Errorf("pdp_data_sets update affected 0 rows")
		}

		// Insert into message_waits_eth
		_, err = tx.Exec(`
            INSERT INTO message_waits_eth (signed_tx_hash, tx_status)
            VALUES ($1, 'pending') ON CONFLICT DO NOTHING
        `, txHashLower)
		if err != nil {
			return false, xerrors.Errorf("failed to insert into message_waits_eth: %w", err)
		}

		return true, nil
	})
	if err != nil {
		return false, xerrors.Errorf("failed to perform database transaction: %w", err)
	}
	log.Infow("Initial challenge window scheduled", "dataSetId", dataSetId, "epoch", init_prove_at)

	// Task completed successfully
	return true, nil
}

func (ipp *InitProvingPeriodTask) CanAccept(ids []harmonytask.TaskID, engine *harmonytask.TaskEngine) ([]harmonytask.TaskID, error) {
	return ids, nil
}

func (ipp *InitProvingPeriodTask) TypeDetails() harmonytask.TaskTypeDetails {
	return harmonytask.TaskTypeDetails{
		Name: tasknames.PDPv0_InitPP,
		// Handoff from data onboarding (PDPv0_Notify → PDPv0_PullPiece → PDPv0_SaveCache).
		// InitPP checks on-chain leaf count before the first challenge request; proving
		// continues PDPv0_InitPP → PDPv0_Prove.
		MayFollow: []string{tasknames.PDPv0_SaveCache},
		Cost: resources.Resources{
			Cpu: 0,
			Gpu: 0,
			Ram: 1 << 20,
		},
		MaxFailures: 3, // Set retry limit to 3 attempts
		RetryWait:   taskhelp.RetryWaitExp(5*time.Second, 2),
	}
}

func (ipp *InitProvingPeriodTask) Adder(taskFunc harmonytask.AddTaskFunc) {
	ipp.addFunc.Set(taskFunc)
}

var _ = harmonytask.Reg(&InitProvingPeriodTask{})