Merge pull request #1 from mtrisic/feature/remote_pr2_job

Feature/remote pr2 job

cmd/lotus-worker/main.go

@@ -36,6 +36,7 @@ import (
 	"github.com/filecoin-project/lotus/metrics"
 	"github.com/filecoin-project/lotus/node/modules"
 	"github.com/filecoin-project/lotus/node/repo"
+	marketauth "github.com/filecoin-project/lotus/sealing_market/market_auth"
 	"github.com/filecoin-project/lotus/storage/paths"
 	"github.com/filecoin-project/lotus/storage/sealer"
 	"github.com/filecoin-project/lotus/storage/sealer/sealtasks"
@@ -63,6 +64,7 @@ func main() {
 		waitQuietCmd,
 		resourcesCmd,
 		tasksCmd,
+		sealingMarketCmd,
 	}
 
 	app := &cli.App{
@@ -146,6 +148,20 @@ var stopCmd = &cli.Command{
 	},
 }
 
+var marketUriOptionalFlag = &cli.StringFlag{
+	Name:     marketURIFlagName,
+	Usage:    "The URL to query the market on",
+	Value:    "http://localhost:3000",
+	Required: false,
+}
+
+var useRemotePr2Flag = &cli.BoolFlag{
+	Name:     remotePr2FlagName,
+	Usage:    "If set, worker will not compute PR2 locally, but instead create a task for rempte sealers to compute it",
+	Value:    false,
+	Required: false,
+}
+
 var runCmd = &cli.Command{
 	Name:  "run",
 	Usage: "Start lotus worker",
@@ -283,6 +299,8 @@ var runCmd = &cli.Command{
 			Value:       true,
 			DefaultText: "inherits --addpiece",
 		},
+		marketUriOptionalFlag,
+		useRemotePr2Flag,
 	},
 	Before: func(cctx *cli.Context) error {
 		if cctx.IsSet("address") {
@@ -485,6 +503,17 @@ var runCmd = &cli.Command{
 			return err
 		}
 
+		useRemotePr2 := cctx.Bool(remotePr2FlagName)
+		if useRemotePr2 {
+			log.Infof("Using remote PR2 is set - worker will use remote sealers to compute PR2 \n")
+			log.Infof("Using \"%v\" as auth repo; \n", repoPath)
+			marketAuth, err := marketauth.New(cctx.String(marketURIFlagName), repoPath)
+			if err != nil {
+				return err
+			}
+			go marketAuth.PollRefresh(cctx.Context)
+		}
+
 		ok, err := r.Exists()
 		if err != nil {
 			return err
@@ -622,6 +651,9 @@ var runCmd = &cli.Command{
 				MaxParallelChallengeReads: cctx.Int("post-parallel-reads"),
 				ChallengeReadTimeout:      cctx.Duration("post-read-timeout"),
 				Name:                      cctx.String("name"),
+				UseRemotePr2:              cctx.Bool(remotePr2FlagName),
+				MarketURI:                 cctx.String(marketURIFlagName),
+				MarketConfigRepoPath:      cctx.String(FlagWorkerRepo),
 			}, remote, localStore, nodeApi, nodeApi, wsts),
 			LocalStore: localStore,
 			Storage:    lr,

cmd/lotus-worker/sealing_market.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+
+	marketauth "github.com/filecoin-project/lotus/sealing_market/market_auth"
+	"github.com/google/uuid"
+	"github.com/urfave/cli/v2"
+)
+
+const marketURIFlagName = "market-uri"
+const remotePr2FlagName = "remotePr2"
+
+var sealingMarketCmd = &cli.Command{
+	Name:  "sealing-market",
+	Usage: "Commands related to the sealing market",
+	Subcommands: []*cli.Command{
+		loginToSealingMarketCmd,
+	},
+}
+
+var loginToSealingMarketCmd = &cli.Command{
+	Name:        "register",
+	Usage:       "Register sealing worker using Sealing Market's OTP",
+	Description: "Used to register lotus-worker with the Sealing Market. You will need OTP code from the Sealing Market to complete the registration.",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:    FlagWorkerRepo,
+			Aliases: []string{FlagWorkerRepoDeprecation},
+			EnvVars: []string{"LOTUS_WORKER_PATH", "WORKER_PATH"},
+			Value:   "~/.lotusworker", // TODO: Consider XDG_DATA_HOME
+			Usage:   fmt.Sprintf("Specify worker repo path. flag %s and env WORKER_PATH are DEPRECATION, will REMOVE SOON", FlagWorkerRepoDeprecation),
+		},
+		marketUriFlag,
+	},
+	Action: func(ctx *cli.Context) error {
+		workerRepoPath := ctx.String(FlagWorkerRepo)
+		fmt.Printf("Repo: %s\n", workerRepoPath)
+
+		otp, err := findUuid(ctx)
+		if err != nil {
+			return err
+		}
+
+		auth, err := marketauth.New(ctx.String(marketURIFlagName), workerRepoPath)
+		if err != nil {
+			return fmt.Errorf("market auth: %w", err)
+		}
+		_, err = auth.Register(otp.String())
+		if err != nil {
+			return fmt.Errorf("error registering appliance: %w", err)
+		}
+
+		fmt.Println("successfully registered as a market appliance. you can now restart in daemon mode.")
+
+		return nil
+	},
+}
+
+func findUuid(ctx *cli.Context) (uuid.UUID, error) {
+	for _, v := range ctx.Args().Slice() {
+		uuid, err := uuid.Parse(v)
+		if err == nil {
+			return uuid, nil
+		}
+	}
+
+	return uuid.UUID{}, errors.New("could not find OTP in command line arguments. Please try again with a valid OTP")
+}
+
+var marketUriFlag = &cli.StringFlag{
+	Name:     marketURIFlagName,
+	Usage:    "The URL to query the market on",
+	Value:    "http://localhost:3000",
+	Required: true,
+}

sealing_market/jobs.go

@@ -0,0 +1,28 @@
+package sealing_market
+
+type Jobs interface {
+	*Pr2Input // *Pr2Input | *C2JobRequest| *C2JobRequest
+	JobType() string
+}
+
+type AddJobRequest[T Jobs] struct {
+	JobType string `json:"job_type"`
+	Input   T      `json:"input"`
+}
+
+type Pr2Input struct {
+	VannilaProofs     [][]byte `json:"vanilla_proofs"`
+	CommRNew          string   `json:"comm_r_new"`
+	CommROld          string   `json:"comm_r_old"` // cid.Cid
+	CommDNew          string   `json:"comm_d_new"`
+	StorageProviderId uint64   `json:"storage_provider_id"`
+	SectorId          uint64   `json:"sector_id"`
+	RegisteredProof   string   `json:"registered_proof"`
+}
+
+func (j Pr2Input) JobType() string { return "PR2" }
+
+type Output[T any] struct {
+	JobType string `json:"job_type"`
+	Output  T      `json:"output"`
+}