Merge pull request #3350 from filecoin-project/fix/some-fsm-issues

Fix some fsm issues

api/api_full.go

@@ -325,7 +325,7 @@ type FullNode interface {
 	StateMinerAvailableBalance(context.Context, address.Address, types.TipSetKey) (types.BigInt, error)
 	// StateSectorPreCommitInfo returns the PreCommit info for the specified miner's sector
 	StateSectorPreCommitInfo(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error)
-	// StateSectorGetInfo returns the on-chain info for the specified miner's sector
+	// StateSectorGetInfo returns the on-chain info for the specified miner's sector. Returns null in case the sector info isn't found
 	// NOTE: returned info.Expiration may not be accurate in some cases, use StateSectorExpiration to get accurate
 	// expiration epoch
 	StateSectorGetInfo(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (*miner.SectorOnChainInfo, error)

api/api_storage.go

@@ -120,16 +120,18 @@ type SectorLog struct {
 }
 
 type SectorInfo struct {
-	SectorID  abi.SectorNumber
-	State     SectorState
-	CommD     *cid.Cid
-	CommR     *cid.Cid
-	Proof     []byte
-	Deals     []abi.DealID
-	Ticket    SealTicket
-	Seed      SealSeed
-	Retries   uint64
-	ToUpgrade bool
+	SectorID     abi.SectorNumber
+	State        SectorState
+	CommD        *cid.Cid
+	CommR        *cid.Cid
+	Proof        []byte
+	Deals        []abi.DealID
+	Ticket       SealTicket
+	Seed         SealSeed
+	PreCommitMsg *cid.Cid
+	CommitMsg    *cid.Cid
+	Retries      uint64
+	ToUpgrade    bool
 
 	LastErr string
 

chain/beacon/beacon.go

@@ -37,6 +37,10 @@ func ValidateBlockValues(b RandomBeacon, h *types.BlockHeader, prevEntry types.B
 		return nil
 	}
 
+	if len(h.BeaconEntries) == 0 {
+		return xerrors.Errorf("expected to have beacon entries in this block, but didn't find any")
+	}
+
 	last := h.BeaconEntries[len(h.BeaconEntries)-1]
 	if last.Round != maxRound {
 		return xerrors.Errorf("expected final beacon entry in block to be at round %d, got %d", maxRound, last.Round)

chain/stmgr/utils.go

@@ -157,7 +157,7 @@ func MinerSectorInfo(ctx context.Context, sm *StateManager, maddr address.Addres
 		return nil, err
 	}
 	if !ok {
-		return nil, xerrors.New("sector not found")
+		return nil, nil
 	}
 
 	return sectorInfo, nil

cmd/lotus-storage-miner/info.go

@@ -236,18 +236,24 @@ var stateList = []stateMeta{
 	{col: 39, state: "Total"},
 	{col: color.FgGreen, state: sealing.Proving},
 
+	{col: color.FgBlue, state: sealing.Empty},
+	{col: color.FgBlue, state: sealing.WaitDeals},
+
 	{col: color.FgRed, state: sealing.UndefinedSectorState},
-	{col: color.FgYellow, state: sealing.Empty},
 	{col: color.FgYellow, state: sealing.Packing},
 	{col: color.FgYellow, state: sealing.PreCommit1},
 	{col: color.FgYellow, state: sealing.PreCommit2},
 	{col: color.FgYellow, state: sealing.PreCommitting},
 	{col: color.FgYellow, state: sealing.PreCommitWait},
 	{col: color.FgYellow, state: sealing.WaitSeed},
 	{col: color.FgYellow, state: sealing.Committing},
+	{col: color.FgYellow, state: sealing.SubmitCommit},
 	{col: color.FgYellow, state: sealing.CommitWait},
 	{col: color.FgYellow, state: sealing.FinalizeSector},
 
+	{col: color.FgCyan, state: sealing.Removing},
+	{col: color.FgCyan, state: sealing.Removed},
+
 	{col: color.FgRed, state: sealing.FailedUnrecoverable},
 	{col: color.FgRed, state: sealing.SealPreCommit1Failed},
 	{col: color.FgRed, state: sealing.SealPreCommit2Failed},
@@ -259,6 +265,9 @@ var stateList = []stateMeta{
 	{col: color.FgRed, state: sealing.Faulty},
 	{col: color.FgRed, state: sealing.FaultReported},
 	{col: color.FgRed, state: sealing.FaultedFinal},
+	{col: color.FgRed, state: sealing.RemoveFailed},
+	{col: color.FgRed, state: sealing.DealsExpired},
+	{col: color.FgRed, state: sealing.RecoverDealIDs},
 }
 
 func init() {

cmd/lotus-storage-miner/sealing.go

@@ -155,6 +155,9 @@ var sealingJobsCmd = &cli.Command{
 
 		// oldest first
 		sort.Slice(lines, func(i, j int) bool {
+			if lines[i].RunWait != lines[j].RunWait {
+				return lines[i].RunWait < lines[j].RunWait
+			}
 			return lines[i].Start.Before(lines[j].Start)
 		})
 
@@ -170,10 +173,14 @@ var sealingJobsCmd = &cli.Command{
 		}
 
 		tw := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
-		_, _ = fmt.Fprintf(tw, "ID\tSector\tWorker\tHostname\tTask\tTime\n")
+		_, _ = fmt.Fprintf(tw, "ID\tSector\tWorker\tHostname\tTask\tState\tTime\n")
 
 		for _, l := range lines {
-			_, _ = fmt.Fprintf(tw, "%d\t%d\t%d\t%s\t%s\t%s\n", l.ID, l.Sector.Number, l.wid, workerHostnames[l.wid], l.Task.Short(), time.Now().Sub(l.Start).Truncate(time.Millisecond*100))
+			state := "running"
+			if l.RunWait != 0 {
+				state = fmt.Sprintf("assigned(%d)", l.RunWait-1)
+			}
+			_, _ = fmt.Fprintf(tw, "%d\t%d\t%d\t%s\t%s\t%s\t%s\n", l.ID, l.Sector.Number, l.wid, workerHostnames[l.wid], l.Task.Short(), state, time.Now().Sub(l.Start).Truncate(time.Millisecond*100))
 		}
 
 		return tw.Flush()

cmd/lotus-storage-miner/sectors.go

@@ -97,6 +97,8 @@ var sectorsStatusCmd = &cli.Command{
 		fmt.Printf("TicketH:\t%d\n", status.Ticket.Epoch)
 		fmt.Printf("Seed:\t\t%x\n", status.Seed.Value)
 		fmt.Printf("SeedH:\t\t%d\n", status.Seed.Epoch)
+		fmt.Printf("Precommit:\t%s\n", status.PreCommitMsg)
+		fmt.Printf("Commit:\t\t%s\n", status.CommitMsg)
 		fmt.Printf("Proof:\t\t%x\n", status.Proof)
 		fmt.Printf("Deals:\t\t%v\n", status.Deals)
 		fmt.Printf("Retries:\t%d\n", status.Retries)

documentation/en/api-methods.md

@@ -3642,7 +3642,7 @@ Response:
 ```
 
 ### StateSectorGetInfo
-StateSectorGetInfo returns the on-chain info for the specified miner's sector
+StateSectorGetInfo returns the on-chain info for the specified miner's sector. Returns null in case the sector info isn't found
 NOTE: returned info.Expiration may not be accurate in some cases, use StateSectorExpiration to get accurate
 expiration epoch
 

extern/sector-storage/manager.go

@@ -6,8 +6,7 @@ import (
 	"io"
 	"net/http"
 
-	"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
-
+	"github.com/hashicorp/go-multierror"
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/mitchellh/go-homedir"
@@ -17,6 +16,7 @@ import (
 	"github.com/filecoin-project/specs-storage/storage"
 
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
+	"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
 	"github.com/filecoin-project/lotus/extern/sector-storage/sealtasks"
 	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 	"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
@@ -463,25 +463,19 @@ func (m *Manager) Remove(ctx context.Context, sector abi.SectorID) error {
 		return xerrors.Errorf("acquiring sector lock: %w", err)
 	}
 
-	unsealed := stores.FTUnsealed
-	{
-		unsealedStores, err := m.index.StorageFindSector(ctx, sector, stores.FTUnsealed, 0, false)
-		if err != nil {
-			return xerrors.Errorf("finding unsealed sector: %w", err)
-		}
+	var err error
 
-		if len(unsealedStores) == 0 { // can be already removed
-			unsealed = stores.FTNone
-		}
+	if rerr := m.storage.Remove(ctx, sector, stores.FTSealed, true); rerr != nil {
+		err = multierror.Append(err, xerrors.Errorf("removing sector (sealed): %w", rerr))
+	}
+	if rerr := m.storage.Remove(ctx, sector, stores.FTCache, true); rerr != nil {
+		err = multierror.Append(err, xerrors.Errorf("removing sector (cache): %w", rerr))
+	}
+	if rerr := m.storage.Remove(ctx, sector, stores.FTUnsealed, true); rerr != nil {
+		err = multierror.Append(err, xerrors.Errorf("removing sector (unsealed): %w", rerr))
 	}
 
-	selector := newExistingSelector(m.index, sector, stores.FTCache|stores.FTSealed, false)
-
-	return m.sched.Schedule(ctx, sector, sealtasks.TTFinalize, selector,
-		schedFetch(sector, stores.FTCache|stores.FTSealed|unsealed, stores.PathStorage, stores.AcquireMove),
-		func(ctx context.Context, w Worker) error {
-			return w.Remove(ctx, sector)
-		})
+	return err
 }
 
 func (m *Manager) StorageLocal(ctx context.Context) (map[stores.ID]string, error) {

extern/sector-storage/request_queue.go

@@ -7,6 +7,11 @@ type requestQueue []*workerRequest
 func (q requestQueue) Len() int { return len(q) }
 
 func (q requestQueue) Less(i, j int) bool {
+	oneMuchLess, muchLess := q[i].taskType.MuchLess(q[j].taskType)
+	if oneMuchLess {
+		return muchLess
+	}
+
 	if q[i].priority != q[j].priority {
 		return q[i].priority > q[j].priority
 	}

extern/sector-storage/resources.go

@@ -22,17 +22,17 @@ func (r Resources) MultiThread() bool {
 
 var ResourceTable = map[sealtasks.TaskType]map[abi.RegisteredSealProof]Resources{
 	sealtasks.TTAddPiece: {
-		abi.RegisteredSealProof_StackedDrg64GiBV1: Resources{ // This is probably a bit conservative
-			MaxMemory: 64 << 30,
-			MinMemory: 64 << 30,
+		abi.RegisteredSealProof_StackedDrg64GiBV1: Resources{
+			MaxMemory: 8 << 30,
+			MinMemory: 8 << 30,
 
 			Threads: 1,
 
 			BaseMinMemory: 1 << 30,
 		},
-		abi.RegisteredSealProof_StackedDrg32GiBV1: Resources{ // This is probably a bit conservative
-			MaxMemory: 32 << 30,
-			MinMemory: 32 << 30,
+		abi.RegisteredSealProof_StackedDrg32GiBV1: Resources{
+			MaxMemory: 4 << 30,
+			MinMemory: 4 << 30,
 
 			Threads: 1,