Some review addressing

api/api_storage.go

@@ -85,7 +85,10 @@ type StorageMiner interface {
 	SectorPreCommitFlush(ctx context.Context) (*cid.Cid, error) //perm:admin
 	// SectorPreCommitPending returns a list of pending PreCommit sectors to be sent in the next batch message
 	SectorPreCommitPending(ctx context.Context) ([]abi.SectorID, error) //perm:admin
+	// SectorCommitFlush immediately sends a Commit message with sectors aggregated for Commit.
+	// Returns null if message wasn't sent
 	SectorCommitFlush(ctx context.Context) (*cid.Cid, error) //perm:admin
+	// SectorCommitPending returns a list of pending Commit sectors to be sent in the next aggregate message
 	SectorCommitPending(ctx context.Context) ([]abi.SectorID, error) //perm:admin
 
 	// WorkerConnect tells the node to connect to workers RPC

chain/vm/gas.go

@@ -75,7 +75,7 @@ type Pricelist interface {
 	OnHashing(dataSize int) GasCharge
 	OnComputeUnsealedSectorCid(proofType abi.RegisteredSealProof, pieces []abi.PieceInfo) GasCharge
 	OnVerifySeal(info proof5.SealVerifyInfo) GasCharge
-	OnVerifyAggregateSeals() GasCharge
+	OnVerifyAggregateSeals(aggregate proof5.AggregateSealVerifyProofAndInfos) GasCharge
 	OnVerifyPost(info proof5.WindowPoStVerifyInfo) GasCharge
 	OnVerifyConsensusFault() GasCharge
 }
@@ -282,7 +282,7 @@ func (ps pricedSyscalls) BatchVerifySeals(inp map[address.Address][]proof5.SealV
 }
 
 func (ps pricedSyscalls) VerifyAggregateSeals(aggregate proof5.AggregateSealVerifyProofAndInfos) error {
-	ps.chargeGas(ps.pl.OnVerifyAggregateSeals())
+	ps.chargeGas(ps.pl.OnVerifyAggregateSeals(aggregate))
 	defer ps.chargeGas(gasOnActorExec)
 
 	return ps.under.VerifyAggregateSeals(aggregate)

chain/vm/gas_v0.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	proof2 "github.com/filecoin-project/specs-actors/v2/actors/runtime/proof"
+	proof5 "github.com/filecoin-project/specs-actors/v5/actors/runtime/proof"
 
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/big"
@@ -187,7 +188,7 @@ func (pl *pricelistV0) OnVerifySeal(info proof2.SealVerifyInfo) GasCharge {
 }
 
 // OnVerifyAggregateSeals
-func (pl *pricelistV0) OnVerifyAggregateSeals() GasCharge {
+func (pl *pricelistV0) OnVerifyAggregateSeals(aggregate proof5.AggregateSealVerifyProofAndInfos) GasCharge {
 	// TODO: this needs more cost tunning
 	return newGasCharge("OnVerifyAggregateSeals", pl.verifyAggregateSealBase, 0)
 }

cmd/lotus-shed/cron-count.go

@@ -1,99 +0,0 @@
-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-shed/main.go

@@ -20,7 +20,6 @@ func main() {
 		base32Cmd,
 		base16Cmd,
 		bitFieldCmd,
-		cronWcCmd,
 		frozenMinersCmd,
 		keyinfoCmd,
 		jwtCmd,

documentation/en/api-methods-miner.md

@@ -1561,6 +1561,8 @@ Response: `{}`
 
 
 ### SectorCommitFlush
+SectorCommitFlush immediately sends a Commit message with sectors aggregated for Commit.
+Returns null if message wasn't sent
 
 
 Perms: admin
@@ -1570,6 +1572,7 @@ Inputs: `null`
 Response: `null`
 
 ### SectorCommitPending
+SectorCommitPending returns a list of pending Commit sectors to be sent in the next aggregate message
 
 
 Perms: admin

extern/sector-storage/mock/mock.go

@@ -526,7 +526,6 @@ func (m mockVerifProver) VerifyAggregateSeals(aggregate proof5.AggregateSealVeri
 	for _, info := range aggregate.Infos {
 		sis = append(sis, info.Number)
 	}
-	fmt.Printf("VERSIS %+v\n", sis)
 
 	return bytes.Equal(aggregate.Proof, out), nil
 }
@@ -543,7 +542,6 @@ func (m mockVerifProver) AggregateSealProofs(aggregateInfo proof5.AggregateSealV
 	for _, info := range aggregateInfo.Infos {
 		sis = append(sis, info.Number)
 	}
-	fmt.Printf("AGGSIS %+v\n", sis)
 
 	return out, nil
 }

markets/storageadapter/ondealsectorcommitted.go

@@ -146,39 +146,13 @@ func (mgr *SectorCommittedManager) OnDealSectorPreCommitted(ctx context.Context,
 		}
 
 		// Extract the message parameters
-		switch msg.Method {
+		sn, err := dealSectorInPreCommitMsg(msg, res)
-		case miner.Methods.PreCommitSector:
+		if err != nil {
-			var params miner.SectorPreCommitInfo
+			return false, err
-			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
-				return false, xerrors.Errorf("unmarshal pre commit: %w", err)
 		}
 
-			// Check through the deal IDs associated with this message
+		if sn != nil {
-			for _, did := range params.DealIDs {
+			cb(*sn, false, nil)
-				if did == res.DealID {
-					// Found the deal ID in this message. Callback with the sector ID.
-					cb(params.SectorNumber, false, nil)
-					return false, nil
-				}
-			}
-		case miner.Methods.PreCommitSectorBatch:
-			var params miner5.PreCommitSectorBatchParams
-			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
-				return false, xerrors.Errorf("unmarshal pre commit: %w", err)
-			}
-
-			for _, precommit := range params.Sectors {
-				// Check through the deal IDs associated with this message
-				for _, did := range precommit.DealIDs {
-					if did == res.DealID {
-						// Found the deal ID in this message. Callback with the sector ID.
-						cb(precommit.SectorNumber, false, nil)
-						return false, nil
-					}
-				}
-			}
-		default:
-			return false, xerrors.Errorf("unexpected method %d", msg.Method)
 		}
 
 		// Didn't find the deal ID in this message, so keep looking
@@ -233,31 +207,7 @@ func (mgr *SectorCommittedManager) OnDealSectorCommitted(ctx context.Context, pr
 			return false, nil
 		}
 
-		switch msg.Method {
+		return sectorInCommitMsg(msg, sectorNumber)
-		case miner.Methods.ProveCommitSector:
-			var params miner.ProveCommitSectorParams
-			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
-				return false, xerrors.Errorf("failed to unmarshal prove commit sector params: %w", err)
-			}
-
-			return params.SectorNumber == sectorNumber, nil
-
-		case miner.Methods.ProveCommitAggregate:
-			var params miner5.ProveCommitAggregateParams
-			if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
-				return false, xerrors.Errorf("failed to unmarshal prove commit sector params: %w", err)
-			}
-
-			set, err := params.SectorNumbers.IsSet(uint64(sectorNumber))
-			if err != nil {
-				return false, xerrors.Errorf("checking if sectorNumber is set in commit aggregate message: %w", err)
-			}
-
-			return set, nil
-
-		default:
-			return false, nil
-		}
 	}
 
 	// The deal must be accepted by the deal proposal start epoch, so timeout
@@ -314,6 +264,73 @@ func (mgr *SectorCommittedManager) OnDealSectorCommitted(ctx context.Context, pr
 	return nil
 }
 
+// dealSectorInPreCommitMsg tries to find a sector containing the specified deal
+func dealSectorInPreCommitMsg(msg *types.Message, res sealing.CurrentDealInfo) (*abi.SectorNumber, error) {
+	switch msg.Method {
+	case miner.Methods.PreCommitSector:
+		var params miner.SectorPreCommitInfo
+		if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
+			return nil, xerrors.Errorf("unmarshal pre commit: %w", err)
+		}
+
+		// Check through the deal IDs associated with this message
+		for _, did := range params.DealIDs {
+			if did == res.DealID {
+				// Found the deal ID in this message. Callback with the sector ID.
+				return &params.SectorNumber, nil
+			}
+		}
+	case miner.Methods.PreCommitSectorBatch:
+		var params miner5.PreCommitSectorBatchParams
+		if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
+			return nil, xerrors.Errorf("unmarshal pre commit: %w", err)
+		}
+
+		for _, precommit := range params.Sectors {
+			// Check through the deal IDs associated with this message
+			for _, did := range precommit.DealIDs {
+				if did == res.DealID {
+					// Found the deal ID in this message. Callback with the sector ID.
+					return &precommit.SectorNumber, nil
+				}
+			}
+		}
+	default:
+		return nil, xerrors.Errorf("unexpected method %d", msg.Method)
+	}
+
+	return nil, nil
+}
+
+// sectorInCommitMsg checks if the provided message commits specified sector
+func sectorInCommitMsg(msg *types.Message, sectorNumber abi.SectorNumber) (bool, error) {
+	switch msg.Method {
+	case miner.Methods.ProveCommitSector:
+		var params miner.ProveCommitSectorParams
+		if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
+			return false, xerrors.Errorf("failed to unmarshal prove commit sector params: %w", err)
+		}
+
+		return params.SectorNumber == sectorNumber, nil
+
+	case miner.Methods.ProveCommitAggregate:
+		var params miner5.ProveCommitAggregateParams
+		if err := params.UnmarshalCBOR(bytes.NewReader(msg.Params)); err != nil {
+			return false, xerrors.Errorf("failed to unmarshal prove commit sector params: %w", err)
+		}
+
+		set, err := params.SectorNumbers.IsSet(uint64(sectorNumber))
+		if err != nil {
+			return false, xerrors.Errorf("checking if sectorNumber is set in commit aggregate message: %w", err)
+		}
+
+		return set, nil
+
+	default:
+		return false, nil
+	}
+}
+
 func (mgr *SectorCommittedManager) checkIfDealAlreadyActive(ctx context.Context, ts *types.TipSet, proposal *market.DealProposal, publishCid cid.Cid) (sealing.CurrentDealInfo, bool, error) {
 	res, err := mgr.dealInfo.GetCurrentDealInfo(ctx, ts.Key().Bytes(), proposal, publishCid)
 	if err != nil {