process niporep sector commit info and push message on chain #1086

Author: beeleelee

damocles-manager/modules/impl/commitmgr/commit_processor.go

@@ -14,6 +14,7 @@ import (
 	"github.com/filecoin-project/go-state-types/big"
 	stbuiltin "github.com/filecoin-project/go-state-types/builtin"
 	miner13 "github.com/filecoin-project/go-state-types/builtin/v13/miner"
+	miner14 "github.com/filecoin-project/go-state-types/builtin/v14/miner"
 
 	"github.com/filecoin-project/venus/venus-shared/actors/builtin/miner"
 
@@ -129,15 +130,21 @@ func (c CommitProcessor) Process(
 
 	ddoSectors := make([]core.SectorState, 0)
 	builtinMarketSectors := make([]core.SectorState, 0)
+	niporepSectors := make([]core.SectorState, 0)
 	for i := range sectors {
-		if sectors[i].HasBuiltinMarketDeal() {
+		if sectors[i].SectorType.IsNonInteractive() {
+			niporepSectors = append(niporepSectors, sectors[i])
+		} else if sectors[i].HasBuiltinMarketDeal() {
 			builtinMarketSectors = append(builtinMarketSectors, sectors[i])
 		} else {
 			ddoSectors = append(ddoSectors, sectors[i])
 		}
 	}
 
 	aggregate := c.ShouldBatch(mid) && len(sectors) >= core.MinAggregatedSectors
+	if len(niporepSectors) > 0 {
+		return c.ProcessNiPoRep(ctx, niporepSectors, mid, ctrlAddr, tok, nv, aggregate)
+	}
 	if nv >= MinDDONetworkVersion {
 		if err := c.ProcessV2(ctx, ddoSectors, mid, ctrlAddr, tok, nv, aggregate); err != nil {
 			return err
@@ -376,6 +383,135 @@ func (c CommitProcessor) ProcessV2(
 	return nil
 }
 
+func (c CommitProcessor) ProcessNiPoRep(
+	ctx context.Context,
+	sectors []core.SectorState,
+	mid abi.ActorID,
+	ctrlAddr address.Address,
+	tok core.TipSetToken,
+	nv network.Version,
+	batch bool,
+) error {
+	// Notice: If a sector in sectors has been sent, it's cid failed should be changed already.
+	plog := log.With("proc", "prove", "miner", mid, "ctrl", ctrlAddr.String(), "len", len(sectors))
+
+	start := time.Now()
+	defer func() {
+		plog.Infof("finished process, elapsed %s", time.Since(start))
+	}()
+
+	defer updateSector(ctx, c.smgr, sectors, plog)
+
+	mcfg, err := c.config.MinerConfig(mid)
+	if err != nil {
+		return fmt.Errorf("get miner config for %d: %w", mid, err)
+	}
+
+	arp, err := c.aggregateProofType(nv)
+	if err != nil {
+		return fmt.Errorf("get aggregate proof type: %w", err)
+	}
+
+	infos := []core.AggregateSealVerifyInfo{}
+	sectorsMap := map[abi.SectorNumber]core.SectorState{}
+	failed := map[abi.SectorID]struct{}{}
+	actInfos := []miner14.SectorNIActivationInfo{}
+
+	collateral := big.Zero()
+	for i, p := range sectors {
+		sectorsMap[p.ID.Number] = sectors[i]
+		expire, err := c.sectorExpiration(ctx, &p)
+		if err != nil {
+			plog.Errorf("get sector expiration for %d failed: %s\n", p.ID.Number, err)
+			failed[sectors[i].ID] = struct{}{}
+			continue
+		}
+
+		if mcfg.Commitment.Prove.SendFund {
+			sc, err := getSectorCollateral(ctx, c.api, mid, p.ID.Number, tok)
+			if err != nil {
+				plog.Errorf("get sector collateral for %d failed: %s\n", p.ID.Number, err)
+				failed[sectors[i].ID] = struct{}{}
+				continue
+			}
+
+			collateral = big.Add(collateral, sc)
+		}
+
+		infos = append(infos, core.AggregateSealVerifyInfo{
+			Number:                p.ID.Number,
+			Randomness:            abi.SealRandomness(p.Ticket.Ticket),
+			InteractiveRandomness: abi.InteractiveSealRandomness(p.Seed.Seed),
+			SealedCID:             p.Pre.CommR,
+			UnsealedCID:           p.Pre.CommD,
+		})
+
+		actInfos[i] = miner14.SectorNIActivationInfo{
+			SealingNumber: p.ID.Number,
+			SealerID:      mid,
+			SealedCID:     p.Pre.CommR,
+			SectorNumber:  p.ID.Number,
+			SealRandEpoch: p.Seed.Epoch,
+			Expiration:    expire,
+		}
+	}
+
+	if len(infos) == 0 {
+		return nil
+	}
+
+	sort.Slice(infos, func(i, j int) bool {
+		return infos[i].Number < infos[j].Number
+	})
+
+	sort.Slice(actInfos, func(i, j int) bool {
+		return actInfos[i].SealingNumber < actInfos[j].SealingNumber
+	})
+
+	params := &miner14.ProveCommitSectorsNIParams{
+		Sectors:                  actInfos,
+		SealProofType:            sectorsMap[infos[0].Number].SectorType,
+		AggregateProofType:       arp,
+		ProvingDeadline:          7,
+		RequireActivationSuccess: true,
+	}
+
+	proofs := make([][]byte, 0)
+	for i := range infos {
+		proofs = append(proofs, sectorsMap[infos[i].Number].Proof.Proof)
+	}
+
+	params.AggregateProof, err = c.prover.AggregateSealProofs(ctx, core.AggregateSealVerifyProofAndInfos{
+		Miner:          mid,
+		SealProof:      sectorsMap[infos[0].Number].SectorType,
+		AggregateProof: arp,
+		Infos:          infos,
+	}, proofs)
+
+	if err != nil {
+		return fmt.Errorf("aggregate sector failed: %w", err)
+	}
+
+	enc := new(bytes.Buffer)
+	if err := params.MarshalCBOR(enc); err != nil {
+		return fmt.Errorf("couldn't serialize ProveCommitAggregateParams: %w", err)
+	}
+
+	ccid, err := pushMessage(ctx, ctrlAddr, mid, collateral, stbuiltin.MethodsMiner.ProveCommitSectorsNI,
+		c.msgClient, &mcfg.Commitment.Prove.Batch.FeeConfig, enc.Bytes(), plog)
+	if err != nil {
+		return fmt.Errorf("push aggregate prove message failed: %w", err)
+	}
+
+	for i := range sectors {
+		if _, ok := failed[sectors[i].ID]; !ok {
+			sectors[i].MessageInfo.CommitCid = &ccid
+		}
+	}
+
+	return nil
+}
+
 func (CommitProcessor) aggregateProofType(nv network.Version) (abi.RegisteredAggregationProof, error) {
 	if nv < network.Version16 {
 		return abi.RegisteredAggregationProof_SnarkPackV1, nil

damocles-manager/modules/impl/commitmgr/policy.go

@@ -144,3 +144,135 @@ func (PreCommitProcessor) ccSectorLifetime(
 
 	return ccLifetimeEpochs - provingPeriod*2, nil
 }
+
+func (pp CommitProcessor) sectorExpiration(ctx context.Context, state *core.SectorState) (abi.ChainEpoch, error) {
+	mcfg, err := pp.config.MinerConfig(state.ID.Miner)
+	if err != nil {
+		return 0, fmt.Errorf("get miner config: %w", err)
+	}
+
+	tok, height, err := pp.api.ChainHead(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("get chain head: %w", err)
+	}
+
+	nv, err := pp.api.StateNetworkVersion(ctx, tok)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get network version: %w", err)
+	}
+
+	av, err := actors.VersionForNetwork(nv)
+	if err != nil {
+		return 0, fmt.Errorf("unsupported network version: %w", err)
+	}
+
+	mpcd, err := policy.GetMaxProveCommitDuration(av, state.SectorType)
+	if err != nil {
+		return 0, fmt.Errorf("getting max prove commit duration: %w", err)
+	}
+
+	expiration, err := pp.sectorEnd(height, nv, state, mcfg.Sector.LifetimeDays, miner.WPoStProvingPeriod)
+	if err != nil {
+		return 0, fmt.Errorf("calculate sector end: %w", err)
+	}
+
+	//revive:disable-next-line:line-length-limit
+	if minExpiration := state.Ticket.Epoch + policy.MaxPreCommitRandomnessLookback + mpcd + policy.MinSectorExpiration; expiration < minExpiration {
+		expiration = minExpiration
+	}
+
+	// Assume: both precommit msg & commit msg land on chain as early as possible
+	maxLifetime, err := policy.GetMaxSectorExpirationExtension(nv)
+	if err != nil {
+		return 0, fmt.Errorf("get max sector expiration extension: %w", err)
+	}
+	maxExpiration := height + policy.GetPreCommitChallengeDelay() + maxLifetime
+	if expiration > maxExpiration {
+		expiration = maxExpiration
+	}
+
+	return expiration, nil
+}
+
+func (pp CommitProcessor) sectorEnd(
+	height abi.ChainEpoch,
+	nv network.Version,
+	state *core.SectorState,
+	lifetimeDays uint64,
+	provingPeriod abi.ChainEpoch,
+) (abi.ChainEpoch, error) {
+	var end *abi.ChainEpoch
+
+	for _, p := range state.SectorPiece() {
+		if !p.HasDealInfo() {
+			continue
+		}
+
+		endEpoch := p.EndEpoch()
+		if endEpoch < height {
+			log.Warnf("piece schedule %+v ended before current epoch %d", p, height)
+			continue
+		}
+
+		if end == nil || *end < endEpoch {
+			tmp := endEpoch
+			end = &tmp
+		}
+	}
+
+	if end == nil {
+		// no deal pieces, get expiration for committed capacity sector
+		expirationDuration, err := pp.ccSectorLifetime(
+			nv,
+			abi.ChainEpoch(lifetimeDays*policy.EpochsInDay),
+			provingPeriod,
+		)
+		if err != nil {
+			return 0, fmt.Errorf("failed to get cc sector lifetime: %w", err)
+		}
+
+		tmp := height + expirationDuration
+		end = &tmp
+	}
+
+	// Ensure there is at least one day for the PC message to land without falling below min sector lifetime
+	// TODO: The "one day" should probably be a config, though it doesn't matter too much
+	minExp := height + policy.GetMinSectorExpiration() + miner.WPoStProvingPeriod
+	if *end < minExp {
+		end = &minExp
+	}
+
+	return *end, nil
+}
+
+func (CommitProcessor) ccSectorLifetime(
+	nv network.Version,
+	ccLifetimeEpochs abi.ChainEpoch,
+	provingPeriod abi.ChainEpoch,
+) (abi.ChainEpoch, error) {
+	maxExpiration, err := policy.GetMaxSectorExpirationExtension(nv)
+	if err != nil {
+		return 0, fmt.Errorf("get max sector expiration extension: %w", err)
+	}
+
+	if ccLifetimeEpochs == 0 {
+		ccLifetimeEpochs = maxExpiration
+	}
+
+	if minExpiration := abi.ChainEpoch(policy.MinSectorExpiration); ccLifetimeEpochs < minExpiration {
+		log.Warnf(
+			"value for CommittedCapacitySectorLifetime is too short, using default minimum (%d epochs)",
+			minExpiration,
+		)
+		return minExpiration, nil
+	}
+	if ccLifetimeEpochs > maxExpiration {
+		log.Warnf(
+			"value for CommittedCapacitySectorLifetime is too long, using default maximum (%d epochs)",
+			maxExpiration,
+		)
+		return maxExpiration, nil
+	}
+
+	return ccLifetimeEpochs - provingPeriod*2, nil
+}