Merge pull request #1092 from ipfs-force-community/feat/nv23

Feat/nv23

damocles-manager/cmd/damocles-manager/internal/util_sealer_sectors.go

@@ -2121,20 +2121,20 @@ func sectorInfo2SectorState(
 				spiece.DealInfo.IsCompatible = true
 			}
 
-			pieceInfo, err := cli.PiecesGetPieceInfo(context.Background(), lotusPiece.Piece.PieceCID)
-			if err != nil {
-				_, _ = fmt.Fprintf(os.Stdout, "get piece info %s: %s\n", lotusPiece.Piece.PieceCID, err)
-			} else if pieceInfo == nil {
-				_, _ = fmt.Fprintf(os.Stdout, "piece info not found %s\n", lotusPiece.Piece.PieceCID)
-			} else {
-				for _, deal := range pieceInfo.Deals {
-					if deal.DealID == lotusPiece.DealInfo.DealID {
-						spiece.DealInfo.Offset = deal.Offset
-						spiece.DealInfo.Length = deal.Length
-						break
-					}
-				}
-			}
+			// pieceInfo, err := cli.PiecesGetPieceInfo(context.Background(), lotusPiece.Piece.PieceCID)
+			// if err != nil {
+			// 	_, _ = fmt.Fprintf(os.Stdout, "get piece info %s: %s\n", lotusPiece.Piece.PieceCID, err)
+			// } else if pieceInfo == nil {
+			// 	_, _ = fmt.Fprintf(os.Stdout, "piece info not found %s\n", lotusPiece.Piece.PieceCID)
+			// } else {
+			// 	for _, deal := range pieceInfo.Deals {
+			// 		if deal.DealID == lotusPiece.DealInfo.DealID {
+			// 			spiece.DealInfo.Offset = deal.Offset
+			// 			spiece.DealInfo.Length = deal.Length
+			// 			break
+			// 		}
+			// 	}
+			// }
 		}
 
 		pieces = append(pieces, spiece)

damocles-manager/core/api.go

@@ -51,7 +51,7 @@ type SealerAPI interface {
 
 	SubmitPersistedEx(ctx context.Context, sid abi.SectorID, instanceName string, isUpgrade bool) (bool, error)
 
-	WaitSeed(context.Context, abi.SectorID) (WaitSeedResp, error)
+	WaitSeed(context.Context, abi.SectorID, abi.RegisteredSealProof) (WaitSeedResp, error)
 
 	SubmitProof(context.Context, abi.SectorID, ProofOnChainInfo, bool) (SubmitProofResp, error)
 

damocles-manager/modules/config.go

@@ -591,6 +591,8 @@ type MinerSealingConfig struct {
 	RequireNotificationSuccessUpdate bool
 
 	UseSyntheticPoRep bool
+
+	SealingSectorDeadlineDelayNi uint64
 }
 
 func defaultMinerSealingConfig() MinerSealingConfig {
@@ -611,6 +613,8 @@ func defaultMinerSealingConfig() MinerSealingConfig {
 		RequireNotificationSuccessUpdate: false,
 
 		UseSyntheticPoRep: false,
+
+		SealingSectorDeadlineDelayNi: 6,
 	}
 }
 

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

@@ -13,6 +13,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/ipfs/go-cid"
 
 	"github.com/filecoin-project/venus/venus-shared/actors/builtin/miner"
@@ -58,15 +59,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
@@ -306,6 +313,142 @@ 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 := getSectorCollateralNiPoRep(ctx, c.api, mid, &p, tok, expire)
+			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
+	})
+	deadline, err := getProvingDeadline(ctx, c.api, mid, tok)
+	if err != nil {
+		return fmt.Errorf("get miner proving deadline for %d: %w", mid, err)
+	}
+
+	// avoid to use current or next deadline
+	deadline = (deadline + mcfg.Sealing.SealingSectorDeadlineDelayNi) % miner.WPoStPeriodDeadlines
+
+	params := &miner14.ProveCommitSectorsNIParams{
+		Sectors:                  actInfos,
+		SealProofType:            sectorsMap[infos[0].Number].SectorType,
+		AggregateProofType:       arp,
+		ProvingDeadline:          deadline,
+		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/params.go

@@ -117,3 +117,48 @@ func getSectorCollateral(
 
 	return collateral, nil
 }
+
+func getSectorCollateralNiPoRep(
+	ctx context.Context,
+	stateMgr SealingAPI,
+	mid abi.ActorID,
+	p *core.SectorState,
+	tok core.TipSetToken,
+	expire abi.ChainEpoch,
+) (abi.TokenAmount, error) {
+	maddr, err := address.NewIDAddress(uint64(mid))
+	if err != nil {
+		return big.Zero(), fmt.Errorf("invalid miner actor id: %w", err)
+	}
+
+	collateral, err := stateMgr.StateMinerInitialPledgeCollateral(ctx, maddr, miner.SectorPreCommitInfo{
+		Expiration:   expire,
+		SectorNumber: p.ID.Number,
+		SealProof:    p.SectorType,
+
+		SealedCID:     p.Pre.CommR,
+		SealRandEpoch: p.Ticket.Epoch,
+	}, tok)
+	if err != nil {
+		return big.Zero(), fmt.Errorf("getting initial pledge collateral: %w", err)
+	}
+	return collateral, nil
+}
+
+func getProvingDeadline(
+	ctx context.Context,
+	stateMgr SealingAPI,
+	mid abi.ActorID,
+	tok core.TipSetToken,
+) (uint64, error) {
+	maddr, err := address.NewIDAddress(uint64(mid))
+	if err != nil {
+		return 0, fmt.Errorf("invalid miner actor id: %w", err)
+	}
+
+	info, err := stateMgr.StateMinerProvingDeadline(ctx, maddr, tok)
+	if err != nil {
+		return 0, fmt.Errorf("getting miner proving deadline: %w", err)
+	}
+	return info.Index, nil
+}