Merge pull request #621 from data-preservation-programs/feat/pdp-scheduling-runner

Implement PDP schedule execution path and config wiring

Author: anjor

cmd/run/dealpusher.go

@@ -1,6 +1,8 @@
 package run
 
 import (
+	"time"
+
 	"github.com/cockroachdb/errors"
 	"github.com/data-preservation-programs/singularity/database"
 	"github.com/data-preservation-programs/singularity/service"
@@ -25,6 +27,26 @@ var DealPusherCmd = &cli.Command{
 			Aliases:     []string{"M"},
 			DefaultText: "Unlimited",
 		},
+		&cli.IntFlag{
+			Name:  "pdp-batch-size",
+			Usage: "Number of roots to include in each PDP add-roots transaction",
+			Value: 128,
+		},
+		&cli.Uint64Flag{
+			Name:  "pdp-gas-limit",
+			Usage: "Gas limit for PDP on-chain transactions",
+			Value: 5000000,
+		},
+		&cli.Uint64Flag{
+			Name:  "pdp-confirmation-depth",
+			Usage: "Number of block confirmations required for PDP transactions",
+			Value: 5,
+		},
+		&cli.DurationFlag{
+			Name:  "pdp-poll-interval",
+			Usage: "Polling interval for PDP transaction confirmation checks",
+			Value: 30 * time.Second,
+		},
 	},
 	Action: func(c *cli.Context) error {
 		db, closer, err := database.OpenFromCLI(c)
@@ -39,7 +61,24 @@ var DealPusherCmd = &cli.Command{
 			return errors.WithStack(err)
 		}
 
-		dm, err := dealpusher.NewDealPusher(db, c.String("lotus-api"), c.String("lotus-token"), c.Uint("deal-attempts"), c.Uint("max-replication-factor"))
+		pdpCfg := dealpusher.PDPSchedulingConfig{
+			BatchSize:         c.Int("pdp-batch-size"),
+			GasLimit:          c.Uint64("pdp-gas-limit"),
+			ConfirmationDepth: c.Uint64("pdp-confirmation-depth"),
+			PollingInterval:   c.Duration("pdp-poll-interval"),
+		}
+		if err := pdpCfg.Validate(); err != nil {
+			return errors.WithStack(err)
+		}
+
+		dm, err := dealpusher.NewDealPusher(
+			db,
+			c.String("lotus-api"),
+			c.String("lotus-token"),
+			c.Uint("deal-attempts"),
+			c.Uint("max-replication-factor"),
+			dealpusher.WithPDPSchedulingConfig(pdpCfg),
+		)
 		if err != nil {
 			return errors.WithStack(err)
 		}

docs/en/cli-reference/run/deal-pusher.md

@@ -16,7 +16,6 @@ import (
 	"github.com/data-preservation-programs/singularity/service/healthcheck"
 	"github.com/data-preservation-programs/singularity/util"
 	"github.com/data-preservation-programs/singularity/util/keystore"
-	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/crypto"
 	"github.com/google/uuid"
 	"github.com/ipfs/go-cid"
@@ -458,39 +457,6 @@ func (d *DealPusher) resolveScheduleDealType(schedule *model.Schedule) model.Dea
 	return d.scheduleDealTypeResolver(schedule)
 }
 
-func defaultPDPSchedulingConfig() PDPSchedulingConfig {
-	return PDPSchedulingConfig{
-		BatchSize:         128,
-		GasLimit:          5_000_000,
-		ConfirmationDepth: 5,
-		PollingInterval:   30 * time.Second,
-	}
-}
-
-// inferScheduleDealType uses the provider address protocol as the discriminator:
-// delegated (f4) addresses are FEVM contracts that speak PDP, everything else
-// is a traditional miner actor that speaks market deals.
-func inferScheduleDealType(schedule *model.Schedule) model.DealType {
-	if schedule == nil {
-		return model.DealTypeMarket
-	}
-	providerAddr, err := address.NewFromString(schedule.Provider)
-	if err != nil {
-		return model.DealTypeMarket
-	}
-	if providerAddr.Protocol() == address.Delegated {
-		return model.DealTypePDP
-	}
-	return model.DealTypeMarket
-}
-
-func (d *DealPusher) runPDPSchedule(_ context.Context, _ *model.Schedule) (model.ScheduleState, error) {
-	if d.pdpProofSetManager == nil || d.pdpTxConfirmer == nil {
-		return model.ScheduleError, errors.New("pdp scheduling dependencies are not configured")
-	}
-	return model.ScheduleError, errors.New("pdp scheduling path is not implemented")
-}
-
 func NewDealPusher(db *gorm.DB, lotusURL string,
 	lotusToken string, numAttempts uint, maxReplicas uint, opts ...Option,
 ) (*DealPusher, error) {