CLI for precommit batching

cmd/lotus-shed/cron-count.go

@@ -0,0 +1,99 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/lotus/build"
+	lcli "github.com/filecoin-project/lotus/cli"
+	"github.com/urfave/cli/v2"
+	"golang.org/x/xerrors"
+)
+
+var cronWcCmd = &cli.Command{
+	Name:        "cron-wc",
+	Description: "cron stats",
+	Subcommands: []*cli.Command{
+		minerDeadlineCronCountCmd,
+	},
+}
+
+var minerDeadlineCronCountCmd = &cli.Command{
+	Name:        "deadline",
+	Description: "list all addresses of miners with active deadline crons",
+	Action: func(c *cli.Context) error {
+		return countDeadlineCrons(c)
+	},
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:  "tipset",
+			Usage: "specify tipset state to search on (pass comma separated array of cids)",
+		},
+	},
+}
+
+func findDeadlineCrons(c *cli.Context) (map[address.Address]struct{}, error) {
+	api, acloser, err := lcli.GetFullNodeAPI(c)
+	if err != nil {
+		return nil, err
+	}
+	defer acloser()
+	ctx := lcli.ReqContext(c)
+
+	ts, err := lcli.LoadTipSet(ctx, c, api)
+	if err != nil {
+		return nil, err
+	}
+	if ts == nil {
+		ts, err = api.ChainHead(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	mAddrs, err := api.StateListMiners(ctx, ts.Key())
+	if err != nil {
+		return nil, err
+	}
+	activeMiners := make(map[address.Address]struct{})
+	for _, mAddr := range mAddrs {
+		// All miners have active cron before v4.
+		// v4 upgrade epoch is last epoch running v3 epoch and api.StateReadState reads
+		// parent state, so v4 state isn't read until upgrade epoch + 2
+		if ts.Height() <= build.UpgradeTurboHeight+1 {
+			activeMiners[mAddr] = struct{}{}
+			continue
+		}
+		st, err := api.StateReadState(ctx, mAddr, ts.Key())
+		if err != nil {
+			return nil, err
+		}
+		minerState, ok := st.State.(map[string]interface{})
+		if !ok {
+			return nil, xerrors.Errorf("internal error: failed to cast miner state to expected map type")
+		}
+
+		activeDlineIface, ok := minerState["DeadlineCronActive"]
+		if !ok {
+			return nil, xerrors.Errorf("miner %s had no deadline state, is this a v3 state root?", mAddr)
+		}
+		active := activeDlineIface.(bool)
+		if active {
+			activeMiners[mAddr] = struct{}{}
+		}
+	}
+
+	return activeMiners, nil
+}
+
+func countDeadlineCrons(c *cli.Context) error {
+	activeMiners, err := findDeadlineCrons(c)
+	if err != nil {
+		return err
+	}
+	for addr := range activeMiners {
+		fmt.Printf("%s\n", addr)
+	}
+
+	return nil
+}

cmd/lotus-storage-miner/sectors.go

@@ -45,7 +45,7 @@ var sectorsCmd = &cli.Command{
 		sectorsStartSealCmd,
 		sectorsSealDelayCmd,
 		sectorsCapacityCollateralCmd,
-		sectorsPendingCommit,
+		sectorsBatching,
 	},
 }
 
@@ -970,9 +970,18 @@ var sectorsUpdateCmd = &cli.Command{
 	},
 }
 
-var sectorsPendingCommit = &cli.Command{
+var sectorsBatching = &cli.Command{
+	Name:  "batching",
+	Usage: "manage batch sector operations",
+	Subcommands: []*cli.Command{
+		sectorsBatchingPendingCommit,
+		sectorsBatchingPendingPreCommit,
+	},
+}
+
+var sectorsBatchingPendingCommit = &cli.Command{
 	Name:  "pending-commit",
-	Usage: "list sectors waiting in batch queue",
+	Usage: "list sectors waiting in commit batch queue",
 	Flags: []cli.Flag{
 		&cli.BoolFlag{
 			Name:  "publish-now",
@@ -1017,6 +1026,53 @@ var sectorsPendingCommit = &cli.Command{
 	},
 }
 
+var sectorsBatchingPendingPreCommit = &cli.Command{
+	Name:  "pending-precommit",
+	Usage: "list sectors waiting in precommit batch queue",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:  "publish-now",
+			Usage: "send a batch now",
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := lcli.GetStorageMinerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := lcli.ReqContext(cctx)
+
+		if cctx.Bool("publish-now") {
+			cid, err := api.SectorPreCommitFlush(ctx)
+			if err != nil {
+				return xerrors.Errorf("flush: %w", err)
+			}
+			if cid == nil {
+				return xerrors.Errorf("no sectors to publish")
+			}
+
+			fmt.Println("sector batch published: ", cid)
+			return nil
+		}
+
+		pending, err := api.SectorPreCommitPending(ctx)
+		if err != nil {
+			return xerrors.Errorf("getting pending deals: %w", err)
+		}
+
+		if len(pending) > 0 {
+			for _, sector := range pending {
+				fmt.Println(sector.Number)
+			}
+			return nil
+		}
+
+		fmt.Println("No sectors queued to be committed")
+		return nil
+	},
+}
+
 func yesno(b bool) string {
 	if b {
 		return color.GreenString("YES")

extern/storage-sealing/commit_batch.go

@@ -20,6 +20,7 @@ import (
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
+	"github.com/filecoin-project/lotus/chain/actors/policy"
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 )
 
@@ -333,7 +334,7 @@ func (b *CommitBatcher) Stop(ctx context.Context) error {
 }
 
 func getSectorDeadline(curEpoch abi.ChainEpoch, si SectorInfo) time.Time {
-	deadlineEpoch := si.TicketEpoch
+	deadlineEpoch := si.TicketEpoch + policy.MaxPreCommitRandomnessLookback
 	for _, p := range si.Pieces {
 		if p.DealInfo == nil {
 			continue