From 65e89a08cae102a5422e62a1d78b92afe396a563 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=C5=81ukasz=20Magiera?= <magik6k@users.noreply.github.com>
Date: Fri, 5 Apr 2024 20:58:00 +0200
Subject: [PATCH] fix: sealing: Use PRU3 After Nv22 (#11831)

---
 storage/pipeline/states_replica_update.go |  8 +++-
 storage/pipeline/states_sealing.go        | 45 ++++++++++++-----------
 2 files changed, 31 insertions(+), 22 deletions(-)

diff --git a/storage/pipeline/states_replica_update.go b/storage/pipeline/states_replica_update.go
index 380078e75..a0d92891c 100644
--- a/storage/pipeline/states_replica_update.go
+++ b/storage/pipeline/states_replica_update.go
@@ -11,6 +11,7 @@ import (
 	"github.com/filecoin-project/go-state-types/big"
 	"github.com/filecoin-project/go-state-types/builtin"
 	"github.com/filecoin-project/go-state-types/exitcode"
+	"github.com/filecoin-project/go-state-types/network"
 	"github.com/filecoin-project/go-statemachine"
 
 	"github.com/filecoin-project/lotus/api"
@@ -170,7 +171,12 @@ func (m *Sealing) handleSubmitReplicaUpdate(ctx statemachine.Context, sector Sec
 
 	// figure out message type
 
-	pams, deals, err := m.processPieces(ctx.Context(), sector)
+	nv, err := m.Api.StateNetworkVersion(ctx.Context(), ts.Key())
+	if err != nil {
+		log.Errorf("failed to get network version: %+v", err)
+	}
+
+	pams, deals, err := m.processPieces(ctx.Context(), sector, nv >= network.Version22)
 	if err != nil {
 		log.Errorf("failed to process pieces: %+v", err)
 		return ctx.Send(SectorSubmitReplicaUpdateFailed{})
diff --git a/storage/pipeline/states_sealing.go b/storage/pipeline/states_sealing.go
index 4f40ac7c7..81ee85853 100644
--- a/storage/pipeline/states_sealing.go
+++ b/storage/pipeline/states_sealing.go
@@ -748,30 +748,33 @@ func (m *Sealing) handleSubmitCommit(ctx statemachine.Context, sector SectorInfo
 // processPieces returns either:
 // - a list of piece activation manifests
 // - a list of deal IDs, if all non-filler pieces are deal-id pieces
-func (m *Sealing) processPieces(ctx context.Context, sector SectorInfo) ([]miner.PieceActivationManifest, []abi.DealID, error) {
+func (m *Sealing) processPieces(ctx context.Context, sector SectorInfo, forceDDO bool) ([]miner.PieceActivationManifest, []abi.DealID, error) {
 	pams := make([]miner.PieceActivationManifest, 0, len(sector.Pieces))
 	dealIDs := make([]abi.DealID, 0, len(sector.Pieces))
-	var hasDDO bool
+	hasDDO := forceDDO
 
-	for _, piece := range sector.Pieces {
-		piece := piece
+	if !forceDDO {
+		// if not forcing DDO, check if we have any DDO pieces
+		for _, piece := range sector.Pieces {
+			piece := piece
 
-		// first figure out if this is a ddo sector
-		err := piece.handleDealInfo(handleDealInfoParams{
-			FillerHandler: func(info UniversalPieceInfo) error {
-				// Fillers are implicit (todo review: Are they??)
-				return nil
-			},
-			BuiltinMarketHandler: func(info UniversalPieceInfo) error {
-				return nil
-			},
-			DDOHandler: func(info UniversalPieceInfo) error {
-				hasDDO = true
-				return nil
-			},
-		})
-		if err != nil {
-			return nil, nil, xerrors.Errorf("handleDealInfo: %w", err)
+			// first figure out if this is a ddo sector
+			err := piece.handleDealInfo(handleDealInfoParams{
+				FillerHandler: func(info UniversalPieceInfo) error {
+					// Fillers are implicit (todo review: Are they??)
+					return nil
+				},
+				BuiltinMarketHandler: func(info UniversalPieceInfo) error {
+					return nil
+				},
+				DDOHandler: func(info UniversalPieceInfo) error {
+					hasDDO = true
+					return nil
+				},
+			})
+			if err != nil {
+				return nil, nil, xerrors.Errorf("handleDealInfo: %w", err)
+			}
 		}
 	}
 	for _, piece := range sector.Pieces {
@@ -847,7 +850,7 @@ func (m *Sealing) handleSubmitCommitAggregate(ctx statemachine.Context, sector S
 		return ctx.Send(SectorCommitFailed{xerrors.Errorf("sector had nil commR or commD")})
 	}
 
-	pams, dealIDs, err := m.processPieces(ctx.Context(), sector)
+	pams, dealIDs, err := m.processPieces(ctx.Context(), sector, false)
 	if err != nil {
 		return err
 	}