Merge pull request #3618 from filecoin-project/feat/async-restartable…

…-workers

Track sealing processes across lotus-miner restarts

api/api_common.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/google/uuid"
+
 	"github.com/filecoin-project/go-jsonrpc/auth"
 	metrics "github.com/libp2p/go-libp2p-core/metrics"
 	"github.com/libp2p/go-libp2p-core/network"
@@ -58,6 +60,9 @@ type Common interface {
 	// trigger graceful shutdown
 	Shutdown(context.Context) error
 
+	// Session returns a random UUID of api provider session
+	Session(context.Context) (uuid.UUID, error)
+
 	Closing(context.Context) (<-chan struct{}, error)
 }
 

api/api_full.go

@@ -374,6 +374,8 @@ type FullNode interface {
 	StateMinerInitialPledgeCollateral(context.Context, address.Address, miner.SectorPreCommitInfo, types.TipSetKey) (types.BigInt, error)
 	// StateMinerAvailableBalance returns the portion of a miner's balance that can be withdrawn or spent
 	StateMinerAvailableBalance(context.Context, address.Address, types.TipSetKey) (types.BigInt, error)
+	// StateMinerSectorAllocated checks if a sector is allocated
+	StateMinerSectorAllocated(context.Context, address.Address, abi.SectorNumber, types.TipSetKey) (bool, 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. Returns null in case the sector info isn't found

api/api_storage.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	datatransfer "github.com/filecoin-project/go-data-transfer"
+	"github.com/google/uuid"
 	"github.com/ipfs/go-cid"
 	"github.com/libp2p/go-libp2p-core/peer"
 
@@ -64,8 +65,9 @@ type StorageMiner interface {
 
 	// WorkerConnect tells the node to connect to workers RPC
 	WorkerConnect(context.Context, string) error
-	WorkerStats(context.Context) (map[uint64]storiface.WorkerStats, error)
-	WorkerJobs(context.Context) (map[uint64][]storiface.WorkerJob, error)
+	WorkerStats(context.Context) (map[uuid.UUID]storiface.WorkerStats, error)
+	WorkerJobs(context.Context) (map[uuid.UUID][]storiface.WorkerJob, error)
+	storiface.WorkerReturn
 
 	// SealingSchedDiag dumps internal sealing scheduler state
 	SealingSchedDiag(context.Context) (interface{}, error)

api/api_worker.go

@@ -2,15 +2,13 @@ package api
 
 import (
 	"context"
-	"io"
 
-	"github.com/ipfs/go-cid"
+	"github.com/google/uuid"
 
 	"github.com/filecoin-project/go-state-types/abi"
 	"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"
-	"github.com/filecoin-project/specs-storage/storage"
 
 	"github.com/filecoin-project/lotus/build"
 )
@@ -23,18 +21,12 @@ type WorkerAPI interface {
 	Paths(context.Context) ([]stores.StoragePath, error)
 	Info(context.Context) (storiface.WorkerInfo, error)
 
-	AddPiece(ctx context.Context, sector abi.SectorID, pieceSizes []abi.UnpaddedPieceSize, newPieceSize abi.UnpaddedPieceSize, pieceData storage.Data) (abi.PieceInfo, error)
+	storiface.WorkerCalls
 
-	storage.Sealer
-
-	MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error
-
-	UnsealPiece(context.Context, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) error
-	ReadPiece(context.Context, io.Writer, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize) (bool, error)
+	// Storage / Other
+	Remove(ctx context.Context, sector abi.SectorID) error
 
 	StorageAddLocal(ctx context.Context, path string) error
 
-	Fetch(context.Context, abi.SectorID, stores.SectorFileType, stores.PathType, stores.AcquireMode) error
-
-	Closing(context.Context) (<-chan struct{}, error)
+	Session(context.Context) (uuid.UUID, error)
 }

build/version.go

@@ -84,8 +84,8 @@ func VersionForType(nodeType NodeType) (Version, error) {
 // semver versions of the rpc api exposed
 var (
 	FullAPIVersion   = newVer(0, 17, 0)
-	MinerAPIVersion  = newVer(0, 16, 0)
-	WorkerAPIVersion = newVer(0, 15, 0)
+	MinerAPIVersion  = newVer(0, 17, 0)
+	WorkerAPIVersion = newVer(0, 16, 0)
 )
 
 //nolint:varcheck,deadcode

chain/messagepool/selection.go

@@ -12,6 +12,7 @@ import (
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
 	tbig "github.com/filecoin-project/go-state-types/big"
+
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/messagepool/gasguess"
 	"github.com/filecoin-project/lotus/chain/types"

cli/cmd.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"net/url"
 	"os"
 	"os/signal"
 	"strings"
@@ -206,7 +207,22 @@ func GetFullNodeAPI(ctx *cli.Context) (api.FullNode, jsonrpc.ClientCloser, error
 	return client.NewFullNodeRPC(ctx.Context, addr, headers)
 }
 
-func GetStorageMinerAPI(ctx *cli.Context, opts ...jsonrpc.Option) (api.StorageMiner, jsonrpc.ClientCloser, error) {
+type GetStorageMinerOptions struct {
+	PreferHttp bool
+}
+
+type GetStorageMinerOption func(*GetStorageMinerOptions)
+
+func StorageMinerUseHttp(opts *GetStorageMinerOptions) {
+	opts.PreferHttp = true
+}
+
+func GetStorageMinerAPI(ctx *cli.Context, opts ...GetStorageMinerOption) (api.StorageMiner, jsonrpc.ClientCloser, error) {
+	var options GetStorageMinerOptions
+	for _, opt := range opts {
+		opt(&options)
+	}
+
 	if tn, ok := ctx.App.Metadata["testnode-storage"]; ok {
 		return tn.(api.StorageMiner), func() {}, nil
 	}
@@ -216,7 +232,23 @@ func GetStorageMinerAPI(ctx *cli.Context, opts ...jsonrpc.Option) (api.StorageMi
 		return nil, nil, err
 	}
 
-	return client.NewStorageMinerRPC(ctx.Context, addr, headers, opts...)
+	if options.PreferHttp {
+		u, err := url.Parse(addr)
+		if err != nil {
+			return nil, nil, xerrors.Errorf("parsing miner api URL: %w", err)
+		}
+
+		switch u.Scheme {
+		case "ws":
+			u.Scheme = "http"
+		case "wss":
+			u.Scheme = "https"
+		}
+
+		addr = u.String()
+	}
+
+	return client.NewStorageMinerRPC(ctx.Context, addr, headers)
 }
 
 func GetWorkerAPI(ctx *cli.Context) (api.WorkerAPI, jsonrpc.ClientCloser, error) {

cmd/lotus-bench/main.go

@@ -26,7 +26,7 @@ import (
 	lcli "github.com/filecoin-project/lotus/cli"
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
 	"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper/basicfs"
-	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
+	"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
 	"github.com/filecoin-project/specs-storage/storage"
 
 	lapi "github.com/filecoin-project/lotus/api"
@@ -614,7 +614,7 @@ func runSeals(sb *ffiwrapper.Sealer, sbfs *basicfs.Provider, numSectors int, par
 					if !skipunseal {
 						log.Infof("[%d] Unsealing sector", i)
 						{
-							p, done, err := sbfs.AcquireSector(context.TODO(), abi.SectorID{Miner: mid, Number: 1}, stores.FTUnsealed, stores.FTNone, stores.PathSealing)
+							p, done, err := sbfs.AcquireSector(context.TODO(), abi.SectorID{Miner: mid, Number: 1}, storiface.FTUnsealed, storiface.FTNone, storiface.PathSealing)
 							if err != nil {
 								return xerrors.Errorf("acquire unsealed sector for removing: %w", err)
 							}