Merge pull request #3438 from filecoin-project/asr/changelog

Lotus version 0.5.7

.github/labels.yml

@@ -26,11 +26,14 @@
   color: 00A4E2
   description: "Area: Chain/VM"
 - name: area/chain/sync
-  color: 00A4E2
+  color: 00A4E4
   description: "Area: Chain/Sync"
 - name: area/chain/misc
-  color: 00A4E2
+  color: 00A4E6
   description: "Area: Chain/Misc"
+- name: area/markets
+  color: 00A4E8
+  description: "Area: Markets"  
 - name: area/sealing/fsm
   color: 0bb1ed
   description: "Area: Sealing/FSM"

CHANGELOG.md

@@ -1,5 +1,20 @@
 # Lotus changelog
 
+# 0.5.7 / 2020-08-31
+
+This patch release includes some bugfixes and enhancements to the sector lifecycle and message pool logic. 
+
+## Changes
+
+- Rebuild unsealed infos on miner restart (https://github.com/filecoin-project/lotus/pull/3401)
+- CLI to attach storage paths to workers (https://github.com/filecoin-project/lotus/pull/3405)
+- Do not select negative performing message chains for inclusion (https://github.com/filecoin-project/lotus/pull/3392)
+- Remove a redundant error-check (https://github.com/filecoin-project/lotus/pull/3421)
+- Correctly move unsealed sectors in `FinalizeSectors` (https://github.com/filecoin-project/lotus/pull/3424)
+- Improve worker selection logic (https://github.com/filecoin-project/lotus/pull/3425)
+- Don't use context to close bitswap (https://github.com/filecoin-project/lotus/pull/3430)
+- Correctly estimate gas premium when there is only one message on chain (https://github.com/filecoin-project/lotus/pull/3428)
+
 # 0.5.6 / 2020-08-29
 
 Hotfix release that fixes a panic in the sealing scheduler (https://github.com/filecoin-project/lotus/pull/3389).

api/api_worker.go

@@ -27,11 +27,13 @@ type WorkerAPI interface {
 
 	storage.Sealer
 
-	MoveStorage(ctx context.Context, sector abi.SectorID) error
+	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)
 
+	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)

api/apistruct/struct.go

@@ -317,7 +317,8 @@ type WorkerStruct struct {
 		FinalizeSector  func(context.Context, abi.SectorID, []storage.Range) error                                                                                                                                 `perm:"admin"`
 		ReleaseUnsealed func(ctx context.Context, sector abi.SectorID, safeToFree []storage.Range) error                                                                                                           `perm:"admin"`
 		Remove          func(ctx context.Context, sector abi.SectorID) error                                                                                                                                       `perm:"admin"`
-		MoveStorage     func(ctx context.Context, sector abi.SectorID) error                                                                                                                                       `perm:"admin"`
+		MoveStorage     func(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error                                                                                                          `perm:"admin"`
+		StorageAddLocal func(ctx context.Context, path string) error                                                                                                                                               `perm:"admin"`
 
 		UnsealPiece func(context.Context, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) error `perm:"admin"`
 		ReadPiece   func(context.Context, io.Writer, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize) (bool, error)           `perm:"admin"`
@@ -1219,8 +1220,12 @@ func (w *WorkerStruct) Remove(ctx context.Context, sector abi.SectorID) error {
 	return w.Internal.Remove(ctx, sector)
 }
 
-func (w *WorkerStruct) MoveStorage(ctx context.Context, sector abi.SectorID) error {
-	return w.Internal.MoveStorage(ctx, sector)
+func (w *WorkerStruct) MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error {
+	return w.Internal.MoveStorage(ctx, sector, types)
+}
+
+func (w *WorkerStruct) StorageAddLocal(ctx context.Context, path string) error {
+	return w.Internal.StorageAddLocal(ctx, path)
 }
 
 func (w *WorkerStruct) UnsealPiece(ctx context.Context, id abi.SectorID, index storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize, randomness abi.SealRandomness, c cid.Cid) error {

build/version.go

@@ -25,7 +25,7 @@ func buildType() string {
 }
 
 // BuildVersion is the local build version, set by build system
-const BuildVersion = "0.5.6"
+const BuildVersion = "0.5.7"
 
 func UserVersion() string {
 	return BuildVersion + buildType() + CurrentCommit
@@ -53,7 +53,7 @@ func (ve Version) EqMajorMinor(v2 Version) bool {
 }
 
 // APIVersion is a semver version of the rpc api exposed
-var APIVersion Version = newVer(0, 13, 0)
+var APIVersion Version = newVer(0, 14, 0)
 
 //nolint:varcheck,deadcode
 const (

chain/messagepool/messagepool.go

@@ -204,9 +204,7 @@ func New(api Provider, ds dtypes.MetadataDS, netName dtypes.NetworkName) (*Messa
 
 	cfg, err := loadConfig(ds)
 	if err != nil {
-		if err != nil {
-			return nil, xerrors.Errorf("error loading mpool config: %w", err)
-		}
+		return nil, xerrors.Errorf("error loading mpool config: %w", err)
 	}
 
 	mp := &MessagePool{

chain/messagepool/repub.go

@@ -108,7 +108,7 @@ func (mp *MessagePool) republishPendingMessages() error {
 
 		// we can't fit the current chain but there is gas to spare
 		// trim it and push it down
-		chain.Trim(gasLimit, mp, baseFee, ts, false)
+		chain.Trim(gasLimit, mp, baseFee, ts)
 		for j := i; j < len(chains)-1; j++ {
 			if chains[j].Before(chains[j+1]) {
 				break

chain/messagepool/selection.go

@@ -217,7 +217,7 @@ tailLoop:
 	for gasLimit >= minGas && last < len(chains) {
 		// trim if necessary
 		if chains[last].gasLimit > gasLimit {
-			chains[last].Trim(gasLimit, mp, baseFee, ts, false)
+			chains[last].Trim(gasLimit, mp, baseFee, ts)
 		}
 
 		// push down if it hasn't been invalidated
@@ -284,7 +284,7 @@ tailLoop:
 			}
 
 			// dependencies fit, just trim it
-			chain.Trim(gasLimit-depGasLimit, mp, baseFee, ts, false)
+			chain.Trim(gasLimit-depGasLimit, mp, baseFee, ts)
 			last += i
 			continue tailLoop
 		}
@@ -389,7 +389,7 @@ func (mp *MessagePool) selectMessagesGreedy(curTs, ts *types.TipSet) ([]*types.S
 tailLoop:
 	for gasLimit >= minGas && last < len(chains) {
 		// trim
-		chains[last].Trim(gasLimit, mp, baseFee, ts, false)
+		chains[last].Trim(gasLimit, mp, baseFee, ts)
 
 		// push down if it hasn't been invalidated
 		if chains[last].valid {
@@ -462,15 +462,27 @@ func (mp *MessagePool) selectPriorityMessages(pending map[address.Address]map[ui
 		}
 	}
 
+	if len(chains) == 0 {
+		return nil, gasLimit
+	}
+
 	// 2. Sort the chains
 	sort.Slice(chains, func(i, j int) bool {
 		return chains[i].Before(chains[j])
 	})
 
-	// 3. Merge chains until the block limit; we are willing to include negative performing chains
-	//    as these are messages from our own miners
+	if len(chains) != 0 && chains[0].gasPerf < 0 {
+		log.Warnw("all priority messages in mpool have negative gas performance", "bestGasPerf", chains[0].gasPerf)
+		return nil, gasLimit
+	}
+
+	// 3. Merge chains until the block limit, as long as they have non-negative gas performance
 	last := len(chains)
 	for i, chain := range chains {
+		if chain.gasPerf < 0 {
+			break
+		}
+
 		if chain.gasLimit <= gasLimit {
 			gasLimit -= chain.gasLimit
 			result = append(result, chain.msgs...)
@@ -484,8 +496,8 @@ func (mp *MessagePool) selectPriorityMessages(pending map[address.Address]map[ui
 
 tailLoop:
 	for gasLimit >= minGas && last < len(chains) {
-		// trim, without discarding negative performing messages
-		chains[last].Trim(gasLimit, mp, baseFee, ts, true)
+		// trim, discarding negative performing messages
+		chains[last].Trim(gasLimit, mp, baseFee, ts)
 
 		// push down if it hasn't been invalidated
 		if chains[last].valid {
@@ -503,6 +515,12 @@ tailLoop:
 			if !chain.valid {
 				continue
 			}
+
+			// if gasPerf < 0 we have no more profitable chains
+			if chain.gasPerf < 0 {
+				break tailLoop
+			}
+
 			// does it fit in the bock?
 			if chain.gasLimit <= gasLimit {
 				gasLimit -= chain.gasLimit
@@ -515,9 +533,9 @@ tailLoop:
 			continue tailLoop
 		}
 
-		// the merge loop ended after processing all the chains and we probably still have gas to spare
-		// -- mark the end.
-		last = len(chains)
+		// the merge loop ended after processing all the chains and we probably still have gas to spare;
+		// end the loop
+		break
 	}
 
 	return result, gasLimit
@@ -755,9 +773,9 @@ func (mc *msgChain) Before(other *msgChain) bool {
 		(mc.gasPerf == other.gasPerf && mc.gasReward.Cmp(other.gasReward) > 0)
 }
 
-func (mc *msgChain) Trim(gasLimit int64, mp *MessagePool, baseFee types.BigInt, ts *types.TipSet, priority bool) {
+func (mc *msgChain) Trim(gasLimit int64, mp *MessagePool, baseFee types.BigInt, ts *types.TipSet) {
 	i := len(mc.msgs) - 1
-	for i >= 0 && (mc.gasLimit > gasLimit || (!priority && mc.gasPerf < 0)) {
+	for i >= 0 && (mc.gasLimit > gasLimit || mc.gasPerf < 0) {
 		gasReward := mp.getGasReward(mc.msgs[i], baseFee, ts)
 		mc.gasReward = new(big.Int).Sub(mc.gasReward, gasReward)
 		mc.gasLimit -= mc.msgs[i].Message.GasLimit

cli/cmd.go

@@ -75,6 +75,8 @@ func flagForAPI(t repo.RepoType) string {
 		return "api"
 	case repo.StorageMiner:
 		return "miner-api"
+	case repo.Worker:
+		return "worker-api"
 	default:
 		panic(fmt.Sprintf("Unknown repo type: %v", t))
 	}
@@ -86,6 +88,8 @@ func flagForRepo(t repo.RepoType) string {
 		return "repo"
 	case repo.StorageMiner:
 		return "miner-repo"
+	case repo.Worker:
+		return "worker-repo"
 	default:
 		panic(fmt.Sprintf("Unknown repo type: %v", t))
 	}
@@ -97,6 +101,8 @@ func envForRepo(t repo.RepoType) string {
 		return "FULLNODE_API_INFO"
 	case repo.StorageMiner:
 		return "MINER_API_INFO"
+	case repo.Worker:
+		return "WORKER_API_INFO"
 	default:
 		panic(fmt.Sprintf("Unknown repo type: %v", t))
 	}
@@ -109,6 +115,8 @@ func envForRepoDeprecation(t repo.RepoType) string {
 		return "FULLNODE_API_INFO"
 	case repo.StorageMiner:
 		return "STORAGE_API_INFO"
+	case repo.Worker:
+		return "WORKER_API_INFO"
 	default:
 		panic(fmt.Sprintf("Unknown repo type: %v", t))
 	}
@@ -234,6 +242,15 @@ func GetStorageMinerAPI(ctx *cli.Context, opts ...jsonrpc.Option) (api.StorageMi
 	return client.NewStorageMinerRPC(ctx.Context, addr, headers, opts...)
 }
 
+func GetWorkerAPI(ctx *cli.Context) (api.WorkerAPI, jsonrpc.ClientCloser, error) {
+	addr, headers, err := GetRawAPI(ctx, repo.Worker)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return client.NewWorkerRPC(ctx.Context, addr, headers)
+}
+
 func DaemonContext(cctx *cli.Context) context.Context {
 	if mtCtx, ok := cctx.App.Metadata[metadataTraceContext]; ok {
 		return mtCtx.(context.Context)

cmd/lotus-seal-worker/info.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/urfave/cli/v2"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/chain/types"
+	lcli "github.com/filecoin-project/lotus/cli"
+)
+
+var infoCmd = &cli.Command{
+	Name:  "info",
+	Usage: "Print worker info",
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := lcli.GetWorkerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := lcli.ReqContext(cctx)
+
+		ver, err := api.Version(ctx)
+		if err != nil {
+			return xerrors.Errorf("getting version: %w", err)
+		}
+
+		fmt.Println("Worker version: ", ver)
+		fmt.Print("CLI version: ")
+		cli.VersionPrinter(cctx)
+		fmt.Println()
+
+		info, err := api.Info(ctx)
+		if err != nil {
+			return xerrors.Errorf("getting info: %w", err)
+		}
+
+		fmt.Printf("Hostname: %s\n", info.Hostname)
+		fmt.Printf("CPUs: %d; GPUs: %v\n", info.Resources.CPUs, info.Resources.GPUs)
+		fmt.Printf("RAM: %s; Swap: %s\n", types.SizeStr(types.NewInt(info.Resources.MemPhysical)), types.SizeStr(types.NewInt(info.Resources.MemSwap)))
+		fmt.Printf("Reserved memory: %s\n", types.SizeStr(types.NewInt(info.Resources.MemReserved)))
+		fmt.Println()
+
+		paths, err := api.Paths(ctx)
+		if err != nil {
+			return xerrors.Errorf("getting path info: %w", err)
+		}
+
+		for _, path := range paths {
+			fmt.Printf("%s:\n", path.ID)
+			fmt.Printf("\tWeight: %d; Use: ", path.Weight)
+			if path.CanSeal || path.CanStore {
+				fmt.Printf("Weight: %d; Use: ", path.Weight)
+				if path.CanSeal {
+					fmt.Print("Seal ")
+				}
+				if path.CanStore {
+					fmt.Print("Store")
+				}
+				fmt.Println("")
+			} else {
+				fmt.Print("Use: ReadOnly")
+			}
+			fmt.Printf("\tLocal: %s\n", path.LocalPath)
+		}
+
+		return nil
+	},
+}

cmd/lotus-seal-worker/main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/gorilla/mux"
 	logging "github.com/ipfs/go-log/v2"
+	manet "github.com/multiformats/go-multiaddr/net"
 	"github.com/urfave/cli/v2"
 	"golang.org/x/xerrors"
 
@@ -46,10 +47,10 @@ const FlagWorkerRepoDeprecation = "workerrepo"
 func main() {
 	lotuslog.SetupLogLevels()
 
-	log.Info("Starting lotus worker")
-
 	local := []*cli.Command{
 		runCmd,
+		infoCmd,
+		storageCmd,
 	}
 
 	app := &cli.App{
@@ -153,6 +154,8 @@ var runCmd = &cli.Command{
 		return nil
 	},
 	Action: func(cctx *cli.Context) error {
+		log.Info("Starting lotus worker")
+
 		if !cctx.Bool("enable-gpu-proving") {
 			if err := os.Setenv("BELLMAN_NO_GPU", "true"); err != nil {
 				return xerrors.Errorf("could not set no-gpu env: %+v", err)
@@ -342,6 +345,8 @@ var runCmd = &cli.Command{
 				SealProof: spt,
 				TaskTypes: taskTypes,
 			}, remote, localStore, nodeApi),
+			localStore: localStore,
+			ls:         lr,
 		}
 
 		mux := mux.NewRouter()
@@ -383,6 +388,32 @@ var runCmd = &cli.Command{
 			return err
 		}
 
+		{
+			a, err := net.ResolveTCPAddr("tcp", address)
+			if err != nil {
+				return xerrors.Errorf("parsing address: %w", err)
+			}
+
+			ma, err := manet.FromNetAddr(a)
+			if err != nil {
+				return xerrors.Errorf("creating api multiaddress: %w", err)
+			}
+
+			if err := lr.SetAPIEndpoint(ma); err != nil {
+				return xerrors.Errorf("setting api endpoint: %w", err)
+			}
+
+			ainfo, err := lcli.GetAPIInfo(cctx, repo.StorageMiner)
+			if err != nil {
+				return xerrors.Errorf("could not get miner API info: %w", err)
+			}
+
+			// TODO: ideally this would be a token with some permissions dropped
+			if err := lr.SetAPIToken(ainfo.Token); err != nil {
+				return xerrors.Errorf("setting api token: %w", err)
+			}
+		}
+
 		log.Info("Waiting for tasks")
 
 		go func() {

cmd/lotus-seal-worker/rpc.go

@@ -3,19 +3,44 @@ package main
 import (
 	"context"
 
+	"github.com/mitchellh/go-homedir"
+	"golang.org/x/xerrors"
+
 	"github.com/filecoin-project/specs-storage/storage"
 
-	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
-
 	"github.com/filecoin-project/lotus/build"
+	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 )
 
 type worker struct {
 	*sectorstorage.LocalWorker
+
+	localStore *stores.Local
+	ls         stores.LocalStorage
 }
 
 func (w *worker) Version(context.Context) (build.Version, error) {
 	return build.APIVersion, nil
 }
 
+func (w *worker) StorageAddLocal(ctx context.Context, path string) error {
+	path, err := homedir.Expand(path)
+	if err != nil {
+		return xerrors.Errorf("expanding local path: %w", err)
+	}
+
+	if err := w.localStore.OpenPath(ctx, path); err != nil {
+		return xerrors.Errorf("opening local path: %w", err)
+	}
+
+	if err := w.ls.SetStorage(func(sc *stores.StorageConfig) {
+		sc.StoragePaths = append(sc.StoragePaths, stores.LocalPath{Path: path})
+	}); err != nil {
+		return xerrors.Errorf("get storage config: %w", err)
+	}
+
+	return nil
+}
+
 var _ storage.Sealer = &worker{}

cmd/lotus-seal-worker/storage.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
+	"github.com/google/uuid"
+	"github.com/mitchellh/go-homedir"
+	"github.com/urfave/cli/v2"
+	"golang.org/x/xerrors"
+
+	lcli "github.com/filecoin-project/lotus/cli"
+	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
+)
+
+const metaFile = "sectorstore.json"
+
+var storageCmd = &cli.Command{
+	Name:  "storage",
+	Usage: "manage sector storage",
+	Subcommands: []*cli.Command{
+		storageAttachCmd,
+	},
+}
+
+var storageAttachCmd = &cli.Command{
+	Name:  "attach",
+	Usage: "attach local storage path",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:  "init",
+			Usage: "initialize the path first",
+		},
+		&cli.Uint64Flag{
+			Name:  "weight",
+			Usage: "(for init) path weight",
+			Value: 10,
+		},
+		&cli.BoolFlag{
+			Name:  "seal",
+			Usage: "(for init) use path for sealing",
+		},
+		&cli.BoolFlag{
+			Name:  "store",
+			Usage: "(for init) use path for long-term storage",
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		nodeApi, closer, err := lcli.GetWorkerAPI(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+		ctx := lcli.ReqContext(cctx)
+
+		if !cctx.Args().Present() {
+			return xerrors.Errorf("must specify storage path to attach")
+		}
+
+		p, err := homedir.Expand(cctx.Args().First())
+		if err != nil {
+			return xerrors.Errorf("expanding path: %w", err)
+		}
+
+		if cctx.Bool("init") {
+			if err := os.MkdirAll(p, 0755); err != nil {
+				if !os.IsExist(err) {
+					return err
+				}
+			}
+
+			_, err := os.Stat(filepath.Join(p, metaFile))
+			if !os.IsNotExist(err) {
+				if err == nil {
+					return xerrors.Errorf("path is already initialized")
+				}
+				return err
+			}
+
+			cfg := &stores.LocalStorageMeta{
+				ID:       stores.ID(uuid.New().String()),
+				Weight:   cctx.Uint64("weight"),
+				CanSeal:  cctx.Bool("seal"),
+				CanStore: cctx.Bool("store"),
+			}
+
+			if !(cfg.CanStore || cfg.CanSeal) {
+				return xerrors.Errorf("must specify at least one of --store of --seal")
+			}
+
+			b, err := json.MarshalIndent(cfg, "", "  ")
+			if err != nil {
+				return xerrors.Errorf("marshaling storage config: %w", err)
+			}
+
+			if err := ioutil.WriteFile(filepath.Join(p, metaFile), b, 0644); err != nil {
+				return xerrors.Errorf("persisting storage metadata (%s): %w", filepath.Join(p, metaFile), err)
+			}
+		}
+
+		return nodeApi.StorageAddLocal(ctx, p)
+	},
+}

documentation/en/api-methods.md

@@ -198,7 +198,7 @@ Response:
 ```json
 {
   "Version": "string value",
-  "APIVersion": 3328,
+  "APIVersion": 3584,
   "BlockDelay": 42
 }
 ```

documentation/en/mpool.md

@@ -133,7 +133,7 @@ type MpoolConfig struct {
 
 The meaning of these fields is as follows:
 - `PriorityAddrs` -- these are the addresses of actors whose pending messages should always
-  be included in a block during message selection, regardless of profitability.
+  be included in a block during message selection, as long as they are profitable.
   Miners should configure their own worker addresses so that they include their own messages
   when they produce a new block.
   Default is empty.

extern/sector-storage/localworker.go

@@ -208,8 +208,8 @@ func (l *LocalWorker) Remove(ctx context.Context, sector abi.SectorID) error {
 	return err
 }
 
-func (l *LocalWorker) MoveStorage(ctx context.Context, sector abi.SectorID) error {
-	if err := l.storage.MoveStorage(ctx, sector, l.scfg.SealProofType, stores.FTSealed|stores.FTCache); err != nil {
+func (l *LocalWorker) MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error {
+	if err := l.storage.MoveStorage(ctx, sector, l.scfg.SealProofType, types); err != nil {
 		return xerrors.Errorf("moving sealed data to storage: %w", err)
 	}
 

extern/sector-storage/manager.go

@@ -31,7 +31,7 @@ type URLs []string
 type Worker interface {
 	ffiwrapper.StorageSealer
 
-	MoveStorage(ctx context.Context, sector abi.SectorID) error
+	MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error
 
 	Fetch(ctx context.Context, s abi.SectorID, ft stores.SectorFileType, ptype stores.PathType, am stores.AcquireMode) error
 	UnsealPiece(context.Context, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) error
@@ -441,7 +441,7 @@ func (m *Manager) FinalizeSector(ctx context.Context, sector abi.SectorID, keepU
 	err = m.sched.Schedule(ctx, sector, sealtasks.TTFetch, fetchSel,
 		schedFetch(sector, stores.FTCache|stores.FTSealed|moveUnsealed, stores.PathStorage, stores.AcquireMove),
 		func(ctx context.Context, w Worker) error {
-			return w.MoveStorage(ctx, sector)
+			return w.MoveStorage(ctx, sector, stores.FTCache|stores.FTSealed|moveUnsealed)
 		})
 	if err != nil {
 		return xerrors.Errorf("moving sector to storage: %w", err)

extern/sector-storage/sched.go

@@ -437,6 +437,10 @@ func (sh *scheduler) trySched() {
 			log.Debugf("SCHED ASSIGNED sqi:%d sector %d task %s to window %d", sqi, task.sector.Number, task.taskType, wnd)
 
 			windows[wnd].allocated.add(wr, needRes)
+			// TODO: We probably want to re-sort acceptableWindows here based on new
+			//  workerHandle.utilization + windows[wnd].allocated.utilization (workerHandle.utilization is used in all
+			//  task selectors, but not in the same way, so need to figure out how to do that in a non-O(n^2 way), and
+			//  without additional network roundtrips (O(n^2) could be avoided by turning acceptableWindows.[] into heaps))
 
 			selectedWindow = wnd
 			break

extern/sector-storage/sched_resources.go

@@ -108,3 +108,17 @@ func (a *activeResources) utilization(wr storiface.WorkerResources) float64 {
 
 	return max
 }
+
+func (wh *workerHandle) utilization() float64 {
+	wh.lk.Lock()
+	u := wh.active.utilization(wh.info.Resources)
+	u += wh.preparing.utilization(wh.info.Resources)
+	wh.lk.Unlock()
+	wh.wndLk.Lock()
+	for _, window := range wh.activeWindows {
+		u += window.allocated.utilization(wh.info.Resources)
+	}
+	wh.wndLk.Unlock()
+
+	return u
+}

extern/sector-storage/sched_test.go

@@ -82,7 +82,7 @@ func (s *schedTestWorker) AddPiece(ctx context.Context, sector abi.SectorID, pie
 	panic("implement me")
 }
 
-func (s *schedTestWorker) MoveStorage(ctx context.Context, sector abi.SectorID) error {
+func (s *schedTestWorker) MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error {
 	panic("implement me")
 }
 

extern/sector-storage/selector_alloc.go

@@ -59,7 +59,7 @@ func (s *allocSelector) Ok(ctx context.Context, task sealtasks.TaskType, spt abi
 }
 
 func (s *allocSelector) Cmp(ctx context.Context, task sealtasks.TaskType, a, b *workerHandle) (bool, error) {
-	return a.active.utilization(a.info.Resources) < b.active.utilization(b.info.Resources), nil
+	return a.utilization() < b.utilization(), nil
 }
 
 var _ WorkerSelector = &allocSelector{}

extern/sector-storage/selector_existing.go

@@ -61,7 +61,7 @@ func (s *existingSelector) Ok(ctx context.Context, task sealtasks.TaskType, spt
 }
 
 func (s *existingSelector) Cmp(ctx context.Context, task sealtasks.TaskType, a, b *workerHandle) (bool, error) {
-	return a.active.utilization(a.info.Resources) < b.active.utilization(b.info.Resources), nil
+	return a.utilization() < b.utilization(), nil
 }
 
 var _ WorkerSelector = &existingSelector{}

extern/sector-storage/selector_task.go

@@ -42,7 +42,7 @@ func (s *taskSelector) Cmp(ctx context.Context, _ sealtasks.TaskType, a, b *work
 		return len(atasks) < len(btasks), nil // prefer workers which can do less
 	}
 
-	return a.active.utilization(a.info.Resources) < b.active.utilization(b.info.Resources), nil
+	return a.utilization() < b.utilization(), nil
 }
 
 var _ WorkerSelector = &allocSelector{}

extern/sector-storage/testworker_test.go

@@ -85,7 +85,7 @@ func (t *testWorker) Remove(ctx context.Context, sector abi.SectorID) error {
 	panic("implement me")
 }
 
-func (t *testWorker) MoveStorage(ctx context.Context, sector abi.SectorID) error {
+func (t *testWorker) MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error {
 	panic("implement me")
 }
 

extern/storage-sealing/fsm.go

@@ -376,12 +376,33 @@ func (m *Sealing) restartSectors(ctx context.Context) error {
 		return xerrors.Errorf("getting the sealing delay: %w", err)
 	}
 
+	m.unsealedInfoMap.lk.Lock()
+	defer m.unsealedInfoMap.lk.Unlock()
 	for _, sector := range trackedSectors {
 		if err := m.sectors.Send(uint64(sector.SectorNumber), SectorRestart{}); err != nil {
 			log.Errorf("restarting sector %d: %+v", sector.SectorNumber, err)
 		}
 
 		if sector.State == WaitDeals {
+
+			// put the sector in the unsealedInfoMap
+			if _, ok := m.unsealedInfoMap.infos[sector.SectorNumber]; ok {
+				// something's funky here, but probably safe to move on
+				log.Warnf("sector %v was already in the unsealedInfoMap when restarting", sector.SectorNumber)
+			} else {
+				ui := UnsealedSectorInfo{}
+				for _, p := range sector.Pieces {
+					if p.DealInfo != nil {
+						ui.numDeals++
+					}
+					ui.stored += p.Piece.Size
+					ui.pieceSizes = append(ui.pieceSizes, p.Piece.Size.Unpadded())
+				}
+
+				m.unsealedInfoMap.infos[sector.SectorNumber] = ui
+			}
+
+			// start a fresh timer for the sector
 			if cfg.WaitDealsDelay > 0 {
 				timer := time.NewTimer(cfg.WaitDealsDelay)
 				go func() {

node/impl/full/gas.go

@@ -109,27 +109,22 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 		return prices[i].price.GreaterThan(prices[j].price)
 	})
 
-	// todo: account for how full blocks are
-
 	at := build.BlockGasTarget * int64(blocks) / 2
-	prev := big.Zero()
-
-	premium := big.Zero()
+	prev1, prev2 := big.Zero(), big.Zero()
 	for _, price := range prices {
+		prev1, prev2 = price.price, prev1
 		at -= price.limit
 		if at > 0 {
-			prev = price.price
 			continue
 		}
-
-		if prev.Equals(big.Zero()) {
-			return types.BigAdd(price.price, big.NewInt(1)), nil
-		}
-
-		premium = types.BigAdd(big.Div(types.BigAdd(price.price, prev), types.NewInt(2)), big.NewInt(1))
 	}
 
-	if types.BigCmp(premium, big.Zero()) == 0 {
+	premium := prev1
+	if prev2.Sign() != 0 {
+		premium = big.Div(types.BigAdd(prev1, prev2), types.NewInt(2))
+	}
+
+	if types.BigCmp(premium, types.NewInt(MinGasPremium)) < 0 {
 		switch nblocksincl {
 		case 1:
 			premium = types.NewInt(2 * MinGasPremium)
@@ -144,7 +139,7 @@ func (a *GasAPI) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64,
 	const precision = 32
 	// mean 1, stddev 0.005 => 95% within +-1%
 	noise := 1 + rand.NormFloat64()*0.005
-	premium = types.BigMul(premium, types.NewInt(uint64(noise*(1<<precision))))
+	premium = types.BigMul(premium, types.NewInt(uint64(noise*(1<<precision))+1))
 	premium = types.BigDiv(premium, types.NewInt(1<<precision))
 	return premium, nil
 }

node/modules/chain.go

@@ -38,7 +38,9 @@ func ChainExchange(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt
 	// (so bitswap uses /chain/ipfs/bitswap/1.0.0 internally for chain sync stuff)
 	bitswapNetwork := network.NewFromIpfsHost(host, rt, network.Prefix("/chain"))
 	bitswapOptions := []bitswap.Option{bitswap.ProvideEnabled(false)}
-	exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, bs, bitswapOptions...)
+
+	// Use just exch.Close(), closing the context is not needed
+	exch := bitswap.New(mctx, bitswapNetwork, bs, bitswapOptions...)
 	lc.Append(fx.Hook{
 		OnStop: func(ctx context.Context) error {
 			return exch.Close()

node/modules/storageminer.go

@@ -275,13 +275,14 @@ func StagingDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBloc
 
 	bitswapNetwork := network.NewFromIpfsHost(h, rt)
 	bitswapOptions := []bitswap.Option{bitswap.ProvideEnabled(false)}
-	exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, ibs, bitswapOptions...)
+	exch := bitswap.New(mctx, bitswapNetwork, ibs, bitswapOptions...)
 
 	bsvc := blockservice.New(ibs, exch)
 	dag := merkledag.NewDAGService(bsvc)
 
 	lc.Append(fx.Hook{
 		OnStop: func(_ context.Context) error {
+			// blockservice closes the exchange
 			return bsvc.Close()
 		},
 	})