fix: sealing: More complete snapdeals abort cleanup

.circleci/config.yml

@@ -953,6 +953,11 @@ workflows:
           suite: itest-worker
           target: "./itests/worker_test.go"
 
+      - test:
+          name: test-itest-worker_upgrade
+          suite: itest-worker_upgrade
+          target: "./itests/worker_upgrade_test.go"
+
       - test:
           name: test-unit-cli
           suite: utest-unit-cli

api/api_storage.go

@@ -417,6 +417,10 @@ func (st *SealSeed) Equals(ost *SealSeed) bool {
 
 type SectorState string
 
+func (s *SectorState) String() string {
+	return string(*s)
+}
+
 type AddrUse int
 
 const (

api/api_worker.go

@@ -39,13 +39,13 @@ type Worker interface {
 	SealPreCommit2(ctx context.Context, sector storiface.SectorRef, pc1o storiface.PreCommit1Out) (storiface.CallID, error)                                                                                  //perm:admin
 	SealCommit1(ctx context.Context, sector storiface.SectorRef, ticket abi.SealRandomness, seed abi.InteractiveSealRandomness, pieces []abi.PieceInfo, cids storiface.SectorCids) (storiface.CallID, error) //perm:admin
 	SealCommit2(ctx context.Context, sector storiface.SectorRef, c1o storiface.Commit1Out) (storiface.CallID, error)                                                                                         //perm:admin
-	FinalizeSector(ctx context.Context, sector storiface.SectorRef, keepUnsealed []storiface.Range) (storiface.CallID, error)                                                                                //perm:admin
-	FinalizeReplicaUpdate(ctx context.Context, sector storiface.SectorRef, keepUnsealed []storiface.Range) (storiface.CallID, error)                                                                         //perm:admin
+	FinalizeSector(ctx context.Context, sector storiface.SectorRef) (storiface.CallID, error)                                                                                                                //perm:admin
+	FinalizeReplicaUpdate(ctx context.Context, sector storiface.SectorRef) (storiface.CallID, error)                                                                                                         //perm:admin
 	ReplicaUpdate(ctx context.Context, sector storiface.SectorRef, pieces []abi.PieceInfo) (storiface.CallID, error)                                                                                         //perm:admin
 	ProveReplicaUpdate1(ctx context.Context, sector storiface.SectorRef, sectorKey, newSealed, newUnsealed cid.Cid) (storiface.CallID, error)                                                                //perm:admin
 	ProveReplicaUpdate2(ctx context.Context, sector storiface.SectorRef, sectorKey, newSealed, newUnsealed cid.Cid, vanillaProofs storiface.ReplicaVanillaProofs) (storiface.CallID, error)                  //perm:admin
 	GenerateSectorKeyFromData(ctx context.Context, sector storiface.SectorRef, commD cid.Cid) (storiface.CallID, error)                                                                                      //perm:admin
-	ReleaseUnsealed(ctx context.Context, sector storiface.SectorRef, safeToFree []storiface.Range) (storiface.CallID, error)                                                                                 //perm:admin
+	ReleaseUnsealed(ctx context.Context, sector storiface.SectorRef, keepUnsealed []storiface.Range) (storiface.CallID, error)                                                                               //perm:admin
 	MoveStorage(ctx context.Context, sector storiface.SectorRef, types storiface.SectorFileType) (storiface.CallID, error)                                                                                   //perm:admin
 	UnsealPiece(context.Context, storiface.SectorRef, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) (storiface.CallID, error)                                             //perm:admin
 	Fetch(context.Context, storiface.SectorRef, storiface.SectorFileType, storiface.PathType, storiface.AcquireMode) (storiface.CallID, error)                                                               //perm:admin

api/version.go

@@ -58,7 +58,7 @@ var (
 	FullAPIVersion1 = newVer(2, 3, 0)
 
 	MinerAPIVersion0  = newVer(1, 5, 0)
-	WorkerAPIVersion0 = newVer(1, 6, 0)
+	WorkerAPIVersion0 = newVer(1, 7, 0)
 )
 
 //nolint:varcheck,deadcode

cmd/lotus-seed/seed/seed.go

@@ -164,7 +164,7 @@ func presealSector(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, sid storiface.
 		return nil, xerrors.Errorf("commit: %w", err)
 	}
 
-	if err := sb.FinalizeSector(context.TODO(), sid, nil); err != nil {
+	if err := sb.FinalizeSector(context.TODO(), sid); err != nil {
 		return nil, xerrors.Errorf("trim cache: %w", err)
 	}
 

cmd/lotus-worker/main.go

@@ -369,7 +369,7 @@ var runCmd = &cli.Command{
 		}
 
 		if workerType == "" {
-			taskTypes = append(taskTypes, sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTProveReplicaUpdate1, sealtasks.TTFinalize, sealtasks.TTFinalizeReplicaUpdate)
+			taskTypes = append(taskTypes, sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTProveReplicaUpdate1, sealtasks.TTFinalize, sealtasks.TTFinalizeUnsealed, sealtasks.TTFinalizeReplicaUpdate)
 
 			if !cctx.Bool("no-default") {
 				workerType = sealtasks.WorkerSealing

documentation/en/api-v0-methods-worker.md

@@ -1601,13 +1601,7 @@ Inputs:
       "Number": 9
     },
     "ProofType": 8
-  },
-  [
-    {
-      "Offset": 1024,
-      "Size": 1024
-    }
-  ]
+  }
 ]
 ```
 
@@ -1636,13 +1630,7 @@ Inputs:
       "Number": 9
     },
     "ProofType": 8
-  },
-  [
-    {
-      "Offset": 1024,
-      "Size": 1024
-    }
-  ]
+  }
 ]
 ```
 

itests/kit/node_miner.go

@@ -224,3 +224,36 @@ func (tm *TestMiner) SectorsListNonGenesis(ctx context.Context) ([]abi.SectorNum
 
 	return l[tm.PresealSectors:], nil
 }
+
+type SchedInfo struct {
+	CallToWork   struct{}
+	EarlyRet     interface{}
+	ReturnedWork interface{}
+	SchedInfo    struct {
+		OpenWindows []string
+		Requests    []struct {
+			Priority int
+			SchedId  uuid.UUID
+			Sector   struct {
+				Miner  int
+				Number int
+			}
+			TaskType string
+		}
+	}
+	Waiting interface{}
+}
+
+func (tm *TestMiner) SchedInfo(ctx context.Context) SchedInfo {
+	schedb, err := tm.SealingSchedDiag(ctx, false)
+	require.NoError(tm.t, err)
+
+	j, err := json.MarshalIndent(&schedb, "", "  ")
+	require.NoError(tm.t, err)
+
+	var b SchedInfo
+	err = json.Unmarshal(j, &b)
+	require.NoError(tm.t, err)
+
+	return b
+}

itests/kit/node_opts.go

@@ -58,7 +58,7 @@ var DefaultNodeOpts = nodeOpts{
 	sectors:    DefaultPresealsPerBootstrapMiner,
 	sectorSize: abi.SectorSize(2 << 10), // 2KiB.
 
-	workerTasks:      []sealtasks.TaskType{sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTFinalize},
+	workerTasks:      []sealtasks.TaskType{sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTFinalize, sealtasks.TTFinalizeUnsealed},
 	workerStorageOpt: func(store paths.Store) paths.Store { return store },
 }
 
@@ -229,7 +229,7 @@ func WithWorkerName(n string) NodeOpt {
 	}
 }
 
-var WithSealWorkerTasks = WithTaskTypes([]sealtasks.TaskType{sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTFinalize, sealtasks.TTAddPiece, sealtasks.TTPreCommit1, sealtasks.TTPreCommit2, sealtasks.TTCommit2, sealtasks.TTUnseal})
+var WithSealWorkerTasks = WithTaskTypes(append([]sealtasks.TaskType{sealtasks.TTAddPiece, sealtasks.TTDataCid, sealtasks.TTPreCommit1, sealtasks.TTPreCommit2, sealtasks.TTCommit2, sealtasks.TTUnseal}, DefaultNodeOpts.workerTasks...))
 
 func WithWorkerStorage(transform func(paths.Store) paths.Store) NodeOpt {
 	return func(opts *nodeOpts) error {

itests/worker_test.go

@@ -2,13 +2,11 @@ package itests
 
 import (
 	"context"
-	"encoding/json"
 	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 
-	"github.com/google/uuid"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/xerrors"
@@ -79,7 +77,7 @@ func TestWorkerPledgeLocalFin(t *testing.T) {
 func TestWorkerDataCid(t *testing.T) {
 	ctx := context.Background()
 	_, miner, worker, _ := kit.EnsembleWorker(t, kit.WithAllSubsystems(), kit.ThroughRPC(), kit.WithNoLocalSealing(true),
-		kit.WithTaskTypes([]sealtasks.TaskType{sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTFinalize, sealtasks.TTDataCid, sealtasks.TTAddPiece, sealtasks.TTPreCommit1, sealtasks.TTPreCommit2, sealtasks.TTCommit2, sealtasks.TTUnseal})) // no mock proofs
+		kit.WithSealWorkerTasks) // no mock proofs
 
 	e, err := worker.Enabled(ctx)
 	require.NoError(t, err)
@@ -409,34 +407,14 @@ func TestWindowPostWorkerManualPoSt(t *testing.T) {
 func TestSchedulerRemoveRequest(t *testing.T) {
 	ctx := context.Background()
 	_, miner, worker, _ := kit.EnsembleWorker(t, kit.WithAllSubsystems(), kit.ThroughRPC(), kit.WithNoLocalSealing(true),
-		kit.WithTaskTypes([]sealtasks.TaskType{sealtasks.TTFetch, sealtasks.TTCommit1, sealtasks.TTFinalize, sealtasks.TTDataCid, sealtasks.TTAddPiece, sealtasks.TTPreCommit1, sealtasks.TTCommit2, sealtasks.TTUnseal})) // no mock proofs
+		kit.WithTaskTypes([]sealtasks.TaskType{sealtasks.TTAddPiece, sealtasks.TTPreCommit1})) // no mock proofs
 
 	//ens.InterconnectAll().BeginMining(50 * time.Millisecond)
 
 	e, err := worker.Enabled(ctx)
 	require.NoError(t, err)
 	require.True(t, e)
 
-	type info struct {
-		CallToWork struct {
-		} `json:"CallToWork"`
-		EarlyRet     interface{} `json:"EarlyRet"`
-		ReturnedWork interface{} `json:"ReturnedWork"`
-		SchedInfo    struct {
-			OpenWindows []string `json:"OpenWindows"`
-			Requests    []struct {
-				Priority int    `json:"Priority"`
-				SchedID  string `json:"SchedId"`
-				Sector   struct {
-					Miner  int `json:"Miner"`
-					Number int `json:"Number"`
-				} `json:"Sector"`
-				TaskType string `json:"TaskType"`
-			} `json:"Requests"`
-		} `json:"SchedInfo"`
-		Waiting interface{} `json:"Waiting"`
-	}
-
 	tocheck := miner.StartPledge(ctx, 1, 0, nil)
 	var sn abi.SectorNumber
 	for n := range tocheck {
@@ -453,39 +431,18 @@ func TestSchedulerRemoveRequest(t *testing.T) {
 	}
 
 	// Dump current scheduler info
-	schedb, err := miner.SealingSchedDiag(ctx, false)
-	require.NoError(t, err)
-
-	j, err := json.MarshalIndent(&schedb, "", "  ")
-	require.NoError(t, err)
-
-	var b info
-	err = json.Unmarshal(j, &b)
-	require.NoError(t, err)
-
-	var schedidb uuid.UUID
+	b := miner.SchedInfo(ctx)
 
 	// cast scheduler info and get the request UUID. Call the SealingRemoveRequest()
 	require.Len(t, b.SchedInfo.Requests, 1)
 	require.Equal(t, "seal/v0/precommit/2", b.SchedInfo.Requests[0].TaskType)
 
-	schedidb, err = uuid.Parse(b.SchedInfo.Requests[0].SchedID)
-	require.NoError(t, err)
-
-	err = miner.SealingRemoveRequest(ctx, schedidb)
+	err = miner.SealingRemoveRequest(ctx, b.SchedInfo.Requests[0].SchedId)
 	require.NoError(t, err)
 
 	// Dump the schduler again and compare the UUID if a request is present
 	// If no request present then pass the test
-	scheda, err := miner.SealingSchedDiag(ctx, false)
-	require.NoError(t, err)
-
-	k, err := json.MarshalIndent(&scheda, "", "  ")
-	require.NoError(t, err)
-
-	var a info
-	err = json.Unmarshal(k, &a)
-	require.NoError(t, err)
+	a := miner.SchedInfo(ctx)
 
 	require.Len(t, a.SchedInfo.Requests, 0)
 }