merged master

node/builder.go

@@ -384,6 +384,7 @@ var MinerNode = Options(
 
 	// Sector storage: Proofs
 	Override(new(ffiwrapper.Verifier), ffiwrapper.ProofVerifier),
+	Override(new(ffiwrapper.Prover), ffiwrapper.ProofProver),
 	Override(new(storage2.Prover), From(new(sectorstorage.SectorManager))),
 
 	// Sealing
@@ -407,6 +408,7 @@ var MinerNode = Options(
 	Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
 
 	// Markets (retrieval)
+	Override(new(sectorstorage.PieceProvider), sectorstorage.NewPieceProvider),
 	Override(new(dtypes.RetrievalPricingFunc), modules.RetrievalPricingFunc(config.DealmakingConfig{
 		RetrievalPricing: &config.RetrievalPricing{
 			Strategy: config.DefaultRetrievalPricing,

node/config/def.go

@@ -6,6 +6,8 @@ import (
 
 	"github.com/ipfs/go-cid"
 
+	miner5 "github.com/filecoin-project/specs-actors/v5/actors/builtin/miner"
+
 	"github.com/filecoin-project/lotus/chain/types"
 	sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
 )
@@ -113,6 +115,30 @@ type SealingConfig struct {
 
 	AlwaysKeepUnsealedCopy bool
 
+	// enable / disable precommit batching (takes effect after nv13)
+	BatchPreCommits bool
+	// maximum precommit batch size - batches will be sent immediately above this size
+	MaxPreCommitBatch int
+	MinPreCommitBatch int
+	// how long to wait before submitting a batch after crossing the minimum batch size
+	PreCommitBatchWait Duration
+	// time buffer for forceful batch submission before sectors in batch would start expiring
+	PreCommitBatchSlack Duration
+
+	// enable / disable commit aggregation (takes effect after nv13)
+	AggregateCommits bool
+	// maximum batched commit size - batches will be sent immediately above this size
+	MinCommitBatch int
+	MaxCommitBatch int
+	// how long to wait before submitting a batch after crossing the minimum batch size
+	CommitBatchWait Duration
+	// time buffer for forceful batch submission before sectors in batch would start expiring
+	CommitBatchSlack Duration
+
+	TerminateBatchMax  uint64
+	TerminateBatchMin  uint64
+	TerminateBatchWait Duration
+
 	// Keep this many sectors in sealing pipeline, start CC if needed
 	// todo TargetSealingSectors uint64
 
@@ -268,6 +294,22 @@ func DefaultStorageMiner() *StorageMiner {
 			MaxSealingSectorsForDeals: 0,
 			WaitDealsDelay:            Duration(time.Hour * 6),
 			AlwaysKeepUnsealedCopy:    true,
+
+			BatchPreCommits:     true,
+			MinPreCommitBatch:   1,                                  // we must have at least one proof to aggregate
+			MaxPreCommitBatch:   miner5.PreCommitSectorBatchMaxSize, //
+			PreCommitBatchWait:  Duration(24 * time.Hour),           // this can be up to 6 days
+			PreCommitBatchSlack: Duration(3 * time.Hour),
+
+			AggregateCommits: true,
+			MinCommitBatch:   miner5.MinAggregatedSectors, // we must have at least four proofs to aggregate
+			MaxCommitBatch:   miner5.MaxAggregatedSectors, // this is the maximum aggregation per FIP13
+			CommitBatchWait:  Duration(24 * time.Hour),    // this can be up to 6 days
+			CommitBatchSlack: Duration(1 * time.Hour),
+
+			TerminateBatchMin:  1,
+			TerminateBatchMax:  100,
+			TerminateBatchWait: Duration(5 * time.Minute),
 		},
 
 		Storage: sectorstorage.SealerConfig{

node/hello/hello.go

@@ -77,7 +77,7 @@ func (hs *Service) HandleStream(s inet.Stream) {
 		"hash", hmsg.GenesisHash)
 
 	if hmsg.GenesisHash != hs.syncer.Genesis.Cids()[0] {
-		log.Warnf("other peer has different genesis! (%s)", hmsg.GenesisHash)
+		log.Debugf("other peer has different genesis! (%s)", hmsg.GenesisHash)
 		_ = s.Conn().Close()
 		return
 	}

node/impl/client/client.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"sort"
 	"time"
 
 	"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
@@ -67,7 +68,8 @@ import (
 
 var DefaultHashFunction = uint64(mh.BLAKE2B_MIN + 31)
 
-const dealStartBufferHours uint64 = 49
+// 8 days ~=  SealDuration + PreCommit + MaxProveCommitDuration + 8 hour buffer
+const dealStartBufferHours uint64 = 8 * 24
 
 type API struct {
 	fx.In
@@ -96,7 +98,13 @@ func calcDealExpiration(minDuration uint64, md *dline.Info, startEpoch abi.Chain
 	minExp := startEpoch + abi.ChainEpoch(minDuration)
 
 	// Align on miners ProvingPeriodBoundary
-	return minExp + md.WPoStProvingPeriod - (minExp % md.WPoStProvingPeriod) + (md.PeriodStart % md.WPoStProvingPeriod) - 1
+	exp := minExp + md.WPoStProvingPeriod - (minExp % md.WPoStProvingPeriod) + (md.PeriodStart % md.WPoStProvingPeriod) - 1
+	// Should only be possible for miners created around genesis
+	for exp < minExp {
+		exp += md.WPoStProvingPeriod
+	}
+
+	return exp
 }
 
 func (a *API) imgr() *importmgr.Mgr {
@@ -829,6 +837,83 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 	return
 }
 
+func (a *API) ClientListRetrievals(ctx context.Context) ([]api.RetrievalInfo, error) {
+	deals, err := a.Retrieval.ListDeals()
+	if err != nil {
+		return nil, err
+	}
+	dataTransfersByID, err := a.transfersByID(ctx)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]api.RetrievalInfo, 0, len(deals))
+	for _, v := range deals {
+		// Find the data transfer associated with this deal
+		var transferCh *api.DataTransferChannel
+		if v.ChannelID != nil {
+			if ch, ok := dataTransfersByID[*v.ChannelID]; ok {
+				transferCh = &ch
+			}
+		}
+		out = append(out, a.newRetrievalInfoWithTransfer(transferCh, v))
+	}
+	sort.Slice(out, func(a, b int) bool {
+		return out[a].ID < out[b].ID
+	})
+	return out, nil
+}
+
+func (a *API) ClientGetRetrievalUpdates(ctx context.Context) (<-chan api.RetrievalInfo, error) {
+	updates := make(chan api.RetrievalInfo)
+
+	unsub := a.Retrieval.SubscribeToEvents(func(_ rm.ClientEvent, deal rm.ClientDealState) {
+		updates <- a.newRetrievalInfo(ctx, deal)
+	})
+
+	go func() {
+		defer unsub()
+		<-ctx.Done()
+	}()
+
+	return updates, nil
+}
+
+func (a *API) newRetrievalInfoWithTransfer(ch *api.DataTransferChannel, deal rm.ClientDealState) api.RetrievalInfo {
+	return api.RetrievalInfo{
+		PayloadCID:        deal.PayloadCID,
+		ID:                deal.ID,
+		PieceCID:          deal.PieceCID,
+		PricePerByte:      deal.PricePerByte,
+		UnsealPrice:       deal.UnsealPrice,
+		Status:            deal.Status,
+		Message:           deal.Message,
+		Provider:          deal.Sender,
+		BytesReceived:     deal.TotalReceived,
+		BytesPaidFor:      deal.BytesPaidFor,
+		TotalPaid:         deal.FundsSpent,
+		TransferChannelID: deal.ChannelID,
+		DataTransfer:      ch,
+	}
+}
+
+func (a *API) newRetrievalInfo(ctx context.Context, v rm.ClientDealState) api.RetrievalInfo {
+	// Find the data transfer associated with this deal
+	var transferCh *api.DataTransferChannel
+	if v.ChannelID != nil {
+		state, err := a.DataTransfer.ChannelState(ctx, *v.ChannelID)
+
+		// Note: If there was an error just ignore it, as the data transfer may
+		// be not found if it's no longer active
+		if err == nil {
+			ch := api.NewDataTransferChannel(a.Host.ID(), state)
+			ch.Stages = state.Stages()
+			transferCh = &ch
+		}
+	}
+
+	return a.newRetrievalInfoWithTransfer(transferCh, v)
+}
+
 type multiStoreRetrievalStore struct {
 	storeID multistore.StoreID
 	store   *multistore.Store

node/impl/full/chain.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/filecoin-project/lotus/build"
+
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
@@ -97,7 +99,12 @@ func (a *ChainAPI) ChainGetRandomnessFromTickets(ctx context.Context, tsk types.
 		return nil, xerrors.Errorf("loading tipset key: %w", err)
 	}
 
-	return a.Chain.GetChainRandomness(ctx, pts.Cids(), personalization, randEpoch, entropy)
+	// Doing this here is slightly nicer than doing it in the chainstore directly, but it's still bad for ChainAPI to reason about network upgrades
+	if randEpoch > build.UpgradeHyperdriveHeight {
+		return a.Chain.GetChainRandomnessLookingForward(ctx, pts.Cids(), personalization, randEpoch, entropy)
+	}
+
+	return a.Chain.GetChainRandomnessLookingBack(ctx, pts.Cids(), personalization, randEpoch, entropy)
 }
 
 func (a *ChainAPI) ChainGetRandomnessFromBeacon(ctx context.Context, tsk types.TipSetKey, personalization crypto.DomainSeparationTag, randEpoch abi.ChainEpoch, entropy []byte) (abi.Randomness, error) {
@@ -106,7 +113,12 @@ func (a *ChainAPI) ChainGetRandomnessFromBeacon(ctx context.Context, tsk types.T
 		return nil, xerrors.Errorf("loading tipset key: %w", err)
 	}
 
-	return a.Chain.GetBeaconRandomness(ctx, pts.Cids(), personalization, randEpoch, entropy)
+	// Doing this here is slightly nicer than doing it in the chainstore directly, but it's still bad for ChainAPI to reason about network upgrades
+	if randEpoch > build.UpgradeHyperdriveHeight {
+		return a.Chain.GetBeaconRandomnessLookingForward(ctx, pts.Cids(), personalization, randEpoch, entropy)
+	}
+
+	return a.Chain.GetBeaconRandomnessLookingBack(ctx, pts.Cids(), personalization, randEpoch, entropy)
 }
 
 func (a *ChainAPI) ChainGetBlock(ctx context.Context, msg cid.Cid) (*types.BlockHeader, error) {

node/impl/full/gas.go

@@ -267,7 +267,7 @@ func gasEstimateGasLimit(
 		return -1, xerrors.Errorf("getting key address: %w", err)
 	}
 
-	pending, ts := mpool.PendingFor(fromA)
+	pending, ts := mpool.PendingFor(ctx, fromA)
 	priorMsgs := make([]types.ChainMsg, 0, len(pending))
 	for _, m := range pending {
 		if m.Message.Nonce == msg.Nonce {

node/impl/full/mpool.go

@@ -60,7 +60,7 @@ func (a *MpoolAPI) MpoolSelect(ctx context.Context, tsk types.TipSetKey, ticketQ
 		return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
 
-	return a.Mpool.SelectMessages(ts, ticketQuality)
+	return a.Mpool.SelectMessages(ctx, ts, ticketQuality)
 }
 
 func (a *MpoolAPI) MpoolPending(ctx context.Context, tsk types.TipSetKey) ([]*types.SignedMessage, error) {
@@ -68,7 +68,7 @@ func (a *MpoolAPI) MpoolPending(ctx context.Context, tsk types.TipSetKey) ([]*ty
 	if err != nil {
 		return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
 	}
-	pending, mpts := a.Mpool.Pending()
+	pending, mpts := a.Mpool.Pending(ctx)
 
 	haveCids := map[cid.Cid]struct{}{}
 	for _, m := range pending {
@@ -122,16 +122,16 @@ func (a *MpoolAPI) MpoolPending(ctx context.Context, tsk types.TipSetKey) ([]*ty
 }
 
 func (a *MpoolAPI) MpoolClear(ctx context.Context, local bool) error {
-	a.Mpool.Clear(local)
+	a.Mpool.Clear(ctx, local)
 	return nil
 }
 
 func (m *MpoolModule) MpoolPush(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error) {
-	return m.Mpool.Push(smsg)
+	return m.Mpool.Push(ctx, smsg)
 }
 
 func (a *MpoolAPI) MpoolPushUntrusted(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error) {
-	return a.Mpool.PushUntrusted(smsg)
+	return a.Mpool.PushUntrusted(ctx, smsg)
 }
 
 func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec) (*types.SignedMessage, error) {
@@ -192,7 +192,7 @@ func (a *MpoolAPI) MpoolPushMessage(ctx context.Context, msg *types.Message, spe
 func (a *MpoolAPI) MpoolBatchPush(ctx context.Context, smsgs []*types.SignedMessage) ([]cid.Cid, error) {
 	var messageCids []cid.Cid
 	for _, smsg := range smsgs {
-		smsgCid, err := a.Mpool.Push(smsg)
+		smsgCid, err := a.Mpool.Push(ctx, smsg)
 		if err != nil {
 			return messageCids, err
 		}
@@ -204,7 +204,7 @@ func (a *MpoolAPI) MpoolBatchPush(ctx context.Context, smsgs []*types.SignedMess
 func (a *MpoolAPI) MpoolBatchPushUntrusted(ctx context.Context, smsgs []*types.SignedMessage) ([]cid.Cid, error) {
 	var messageCids []cid.Cid
 	for _, smsg := range smsgs {
-		smsgCid, err := a.Mpool.PushUntrusted(smsg)
+		smsgCid, err := a.Mpool.PushUntrusted(ctx, smsg)
 		if err != nil {
 			return messageCids, err
 		}

node/impl/remoteworker.go

@@ -38,6 +38,16 @@ func connectRemoteWorker(ctx context.Context, fa api.Common, url string) (*remot
 		return nil, xerrors.Errorf("creating jsonrpc client: %w", err)
 	}
 
+	wver, err := wapi.Version(ctx)
+	if err != nil {
+		closer()
+		return nil, err
+	}
+
+	if !wver.EqMajorMinor(api.WorkerAPIVersion0) {
+		return nil, xerrors.Errorf("unsupported worker api version: %s (expected %s)", wver, api.WorkerAPIVersion0)
+	}
+
 	return &remoteWorker{wapi, closer}, nil
 }
 

node/impl/storminer.go

@@ -32,6 +32,7 @@ import (
 	"github.com/filecoin-project/lotus/extern/sector-storage/stores"
 	"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
 	sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
+	"github.com/filecoin-project/lotus/extern/storage-sealing/sealiface"
 
 	"github.com/filecoin-project/lotus/api"
 	apitypes "github.com/filecoin-project/lotus/api/types"
@@ -244,13 +245,13 @@ func (sm *StorageMinerAPI) SectorsList(context.Context) ([]abi.SectorNumber, err
 		return nil, err
 	}
 
-	out := make([]abi.SectorNumber, len(sectors))
-	for i, sector := range sectors {
+	out := make([]abi.SectorNumber, 0, len(sectors))
+	for _, sector := range sectors {
 		if sector.State == sealing.UndefinedSectorState {
 			continue // sector ID not set yet
 		}
 
-		out[i] = sector.SectorNumber
+		out = append(out, sector.SectorNumber)
 	}
 	return out, nil
 }
@@ -373,10 +374,26 @@ func (sm *StorageMinerAPI) SectorTerminatePending(ctx context.Context) ([]abi.Se
 	return sm.Miner.TerminatePending(ctx)
 }
 
+func (sm *StorageMinerAPI) SectorPreCommitFlush(ctx context.Context) ([]sealiface.PreCommitBatchRes, error) {
+	return sm.Miner.SectorPreCommitFlush(ctx)
+}
+
+func (sm *StorageMinerAPI) SectorPreCommitPending(ctx context.Context) ([]abi.SectorID, error) {
+	return sm.Miner.SectorPreCommitPending(ctx)
+}
+
 func (sm *StorageMinerAPI) SectorMarkForUpgrade(ctx context.Context, id abi.SectorNumber) error {
 	return sm.Miner.MarkForUpgrade(id)
 }
 
+func (sm *StorageMinerAPI) SectorCommitFlush(ctx context.Context) ([]sealiface.CommitBatchRes, error) {
+	return sm.Miner.CommitFlush(ctx)
+}
+
+func (sm *StorageMinerAPI) SectorCommitPending(ctx context.Context) ([]abi.SectorID, error) {
+	return sm.Miner.CommitPending(ctx)
+}
+
 func (sm *StorageMinerAPI) WorkerConnect(ctx context.Context, url string) error {
 	w, err := connectRemoteWorker(ctx, sm, url)
 	if err != nil {

node/modules/client.go

@@ -134,8 +134,14 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 
 	// data-transfer push / pull channel restart configuration:
 	dtRestartConfig := dtimpl.ChannelRestartConfig(channelmonitor.Config{
-		// Wait up to 2m for the other side to respond to an Open channel message
-		AcceptTimeout: 2 * time.Minute,
+		// Disable Accept and Complete timeouts until this issue is resolved:
+		// https://github.com/filecoin-project/lotus/issues/6343#
+		// Wait for the other side to respond to an Open channel message
+		AcceptTimeout: 0,
+		// Wait for the other side to send a Complete message once all
+		// data has been sent / received
+		CompleteTimeout: 0,
+
 		// When an error occurs, wait a little while until all related errors
 		// have fired before sending a restart message
 		RestartDebounce: 10 * time.Second,
@@ -143,12 +149,6 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
 		RestartBackoff: time.Minute,
 		// After trying to restart 3 times, give up and fail the transfer
 		MaxConsecutiveRestarts: 3,
-		// After sending a restart message, the time to wait for the peer to
-		// respond with an ack of the restart
-		RestartAckTimeout: 30 * time.Second,
-		// Wait up to 10m for the other side to send a Complete message once all
-		// data has been sent / received
-		CompleteTimeout: 10 * time.Minute,
 	})
 	dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, dtRestartConfig)
 	if err != nil {

node/modules/dtypes/miner.go

@@ -74,10 +74,12 @@ type ConsiderUnverifiedStorageDealsConfigFunc func() (bool, error)
 // disable or enable unverified storage deal acceptance.
 type SetConsiderUnverifiedStorageDealsConfigFunc func(bool) error
 
-// SetSealingDelay sets how long a sector waits for more deals before sealing begins.
+// SetSealingConfigFunc is a function which is used to
+// sets the sealing config.
 type SetSealingConfigFunc func(sealiface.Config) error
 
-// GetSealingDelay returns how long a sector waits for more deals before sealing begins.
+// GetSealingConfigFunc is a function which is used to
+// get the sealing config.
 type GetSealingConfigFunc func() (sealiface.Config, error)
 
 // SetExpectedSealDurationFunc is a function which is used to set how long sealing is expected to take.

node/modules/storageminer.go

@@ -101,7 +101,7 @@ func GetParams(spt abi.RegisteredSealProof) error {
 	}
 
 	// TODO: We should fetch the params for the actual proof type, not just based on the size.
-	if err := paramfetch.GetParams(context.TODO(), build.ParametersJSON(), uint64(ssize)); err != nil {
+	if err := paramfetch.GetParams(context.TODO(), build.ParametersJSON(), build.SrsJSON(), uint64(ssize)); err != nil {
 		return xerrors.Errorf("fetching proof parameters: %w", err)
 	}
 
@@ -204,6 +204,7 @@ type StorageMinerParams struct {
 	Sealer             sectorstorage.SectorManager
 	SectorIDCounter    sealing.SectorIDCounter
 	Verifier           ffiwrapper.Verifier
+	Prover             ffiwrapper.Prover
 	GetSealingConfigFn dtypes.GetSealingConfigFunc
 	Journal            journal.Journal
 	AddrSel            *storage.AddressSelector
@@ -220,6 +221,7 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
 			h      = params.Host
 			sc     = params.SectorIDCounter
 			verif  = params.Verifier
+			prover = params.Prover
 			gsd    = params.GetSealingConfigFn
 			j      = params.Journal
 			as     = params.AddrSel
@@ -237,7 +239,7 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
 			return nil, err
 		}
 
-		sm, err := storage.NewMiner(api, maddr, h, ds, sealer, sc, verif, gsd, fc, j, as)
+		sm, err := storage.NewMiner(api, maddr, h, ds, sealer, sc, verif, prover, gsd, fc, j, as)
 		if err != nil {
 			return nil, err
 		}
@@ -849,6 +851,22 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
 				MaxSealingSectorsForDeals: cfg.MaxSealingSectorsForDeals,
 				WaitDealsDelay:            config.Duration(cfg.WaitDealsDelay),
 				AlwaysKeepUnsealedCopy:    cfg.AlwaysKeepUnsealedCopy,
+
+				BatchPreCommits:     cfg.BatchPreCommits,
+				MinPreCommitBatch:   cfg.MinPreCommitBatch,
+				MaxPreCommitBatch:   cfg.MaxPreCommitBatch,
+				PreCommitBatchWait:  config.Duration(cfg.PreCommitBatchWait),
+				PreCommitBatchSlack: config.Duration(cfg.PreCommitBatchSlack),
+
+				AggregateCommits: cfg.AggregateCommits,
+				MinCommitBatch:   cfg.MinCommitBatch,
+				MaxCommitBatch:   cfg.MaxCommitBatch,
+				CommitBatchWait:  config.Duration(cfg.CommitBatchWait),
+				CommitBatchSlack: config.Duration(cfg.CommitBatchSlack),
+
+				TerminateBatchMax:  cfg.TerminateBatchMax,
+				TerminateBatchMin:  cfg.TerminateBatchMin,
+				TerminateBatchWait: config.Duration(cfg.TerminateBatchWait),
 			}
 		})
 		return
@@ -864,6 +882,22 @@ func NewGetSealConfigFunc(r repo.LockedRepo) (dtypes.GetSealingConfigFunc, error
 				MaxSealingSectorsForDeals: cfg.Sealing.MaxSealingSectorsForDeals,
 				WaitDealsDelay:            time.Duration(cfg.Sealing.WaitDealsDelay),
 				AlwaysKeepUnsealedCopy:    cfg.Sealing.AlwaysKeepUnsealedCopy,
+
+				BatchPreCommits:     cfg.Sealing.BatchPreCommits,
+				MinPreCommitBatch:   cfg.Sealing.MinPreCommitBatch,
+				MaxPreCommitBatch:   cfg.Sealing.MaxPreCommitBatch,
+				PreCommitBatchWait:  time.Duration(cfg.Sealing.PreCommitBatchWait),
+				PreCommitBatchSlack: time.Duration(cfg.Sealing.PreCommitBatchSlack),
+
+				AggregateCommits: cfg.Sealing.AggregateCommits,
+				MinCommitBatch:   cfg.Sealing.MinCommitBatch,
+				MaxCommitBatch:   cfg.Sealing.MaxCommitBatch,
+				CommitBatchWait:  time.Duration(cfg.Sealing.CommitBatchWait),
+				CommitBatchSlack: time.Duration(cfg.Sealing.CommitBatchSlack),
+
+				TerminateBatchMax:  cfg.Sealing.TerminateBatchMax,
+				TerminateBatchMin:  cfg.Sealing.TerminateBatchMin,
+				TerminateBatchWait: time.Duration(cfg.Sealing.TerminateBatchWait),
 			}
 		})
 		return

node/node_test.go

@@ -58,6 +58,12 @@ func TestAPIDealFlow(t *testing.T) {
 	t.Run("TestPublishDealsBatching", func(t *testing.T) {
 		test.TestPublishDealsBatching(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
 	})
+	t.Run("TestOfflineDealFlow", func(t *testing.T) {
+		test.TestOfflineDealFlow(t, builder.MockSbBuilder, blockTime, dealStartEpoch, false)
+	})
+	t.Run("TestOfflineDealFlowFastRetrieval", func(t *testing.T) {
+		test.TestOfflineDealFlow(t, builder.MockSbBuilder, blockTime, dealStartEpoch, true)
+	})
 }
 
 func TestBatchDealInput(t *testing.T) {
@@ -66,6 +72,7 @@ func TestBatchDealInput(t *testing.T) {
 	logging.SetLogLevel("chain", "ERROR")
 	logging.SetLogLevel("sub", "ERROR")
 	logging.SetLogLevel("storageminer", "ERROR")
+	logging.SetLogLevel("sectors", "DEBUG")
 
 	blockTime := 10 * time.Millisecond
 
@@ -170,6 +177,15 @@ func TestPledgeSectors(t *testing.T) {
 	})
 }
 
+func TestPledgeBatching(t *testing.T) {
+	t.Run("100", func(t *testing.T) {
+		test.TestPledgeBatching(t, builder.MockSbBuilder, 50*time.Millisecond, 100)
+	})
+	t.Run("100-before-nv13", func(t *testing.T) {
+		test.TestPledgeBeforeNv13(t, builder.MockSbBuilder, 50*time.Millisecond, 100)
+	})
+}
+
 func TestTapeFix(t *testing.T) {
 	logging.SetLogLevel("miner", "ERROR")
 	logging.SetLogLevel("chainstore", "ERROR")
@@ -186,6 +202,7 @@ func TestWindowedPost(t *testing.T) {
 	}
 
 	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("gen", "ERROR")
 	logging.SetLogLevel("chainstore", "ERROR")
 	logging.SetLogLevel("chain", "ERROR")
 	logging.SetLogLevel("sub", "ERROR")
@@ -234,6 +251,7 @@ func TestWindowPostDispute(t *testing.T) {
 		t.Skip("this takes a few minutes, set LOTUS_TEST_WINDOW_POST=1 to run")
 	}
 	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("gen", "ERROR")
 	logging.SetLogLevel("chainstore", "ERROR")
 	logging.SetLogLevel("chain", "ERROR")
 	logging.SetLogLevel("sub", "ERROR")
@@ -247,6 +265,7 @@ func TestWindowPostDisputeFails(t *testing.T) {
 		t.Skip("this takes a few minutes, set LOTUS_TEST_WINDOW_POST=1 to run")
 	}
 	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("gen", "ERROR")
 	logging.SetLogLevel("chainstore", "ERROR")
 	logging.SetLogLevel("chain", "ERROR")
 	logging.SetLogLevel("sub", "ERROR")
@@ -255,11 +274,40 @@ func TestWindowPostDisputeFails(t *testing.T) {
 	test.TestWindowPostDisputeFails(t, builder.MockSbBuilder, 2*time.Millisecond)
 }
 
+func TestWindowPostBaseFeeNoBurn(t *testing.T) {
+	if os.Getenv("LOTUS_TEST_WINDOW_POST") != "1" {
+		t.Skip("this takes a few minutes, set LOTUS_TEST_WINDOW_POST=1 to run")
+	}
+	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("gen", "ERROR")
+	logging.SetLogLevel("chainstore", "ERROR")
+	logging.SetLogLevel("chain", "ERROR")
+	logging.SetLogLevel("sub", "ERROR")
+	logging.SetLogLevel("storageminer", "ERROR")
+
+	test.TestWindowPostBaseFeeNoBurn(t, builder.MockSbBuilder, 2*time.Millisecond)
+}
+
+func TestWindowPostBaseFeeBurn(t *testing.T) {
+	if os.Getenv("LOTUS_TEST_WINDOW_POST") != "1" {
+		t.Skip("this takes a few minutes, set LOTUS_TEST_WINDOW_POST=1 to run")
+	}
+	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("gen", "ERROR")
+	logging.SetLogLevel("chainstore", "ERROR")
+	logging.SetLogLevel("chain", "ERROR")
+	logging.SetLogLevel("sub", "ERROR")
+	logging.SetLogLevel("storageminer", "ERROR")
+
+	test.TestWindowPostBaseFeeBurn(t, builder.MockSbBuilder, 2*time.Millisecond)
+}
+
 func TestDeadlineToggling(t *testing.T) {
 	if os.Getenv("LOTUS_TEST_DEADLINE_TOGGLING") != "1" {
 		t.Skip("this takes a few minutes, set LOTUS_TEST_DEADLINE_TOGGLING=1 to run")
 	}
 	logging.SetLogLevel("miner", "ERROR")
+	logging.SetLogLevel("gen", "ERROR")
 	logging.SetLogLevel("chainstore", "ERROR")
 	logging.SetLogLevel("chain", "ERROR")
 	logging.SetLogLevel("sub", "ERROR")
@@ -267,3 +315,9 @@ func TestDeadlineToggling(t *testing.T) {
 
 	test.TestDeadlineToggling(t, builder.MockSbBuilder, 2*time.Millisecond)
 }
+
+func TestVerifiedClientTopUp(t *testing.T) {
+	logging.SetLogLevel("storageminer", "FATAL")
+	logging.SetLogLevel("chain", "ERROR")
+	test.AddVerifiedClient(t, builder.MockSbBuilder)
+}

node/test/builder.go

@@ -12,6 +12,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/filecoin-project/go-state-types/network"
+
 	"github.com/gorilla/mux"
 	"golang.org/x/xerrors"
 
@@ -276,12 +278,26 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
 		maddrs = append(maddrs, maddr)
 		genms = append(genms, *genm)
 	}
+
+	rkhKey, err := wallet.GenerateKey(types.KTSecp256k1)
+	if err != nil {
+		return nil, nil
+	}
+
+	vrk := genesis.Actor{
+		Type:    genesis.TAccount,
+		Balance: big.Mul(big.Div(big.NewInt(int64(build.FilBase)), big.NewInt(100)), big.NewInt(int64(build.FilecoinPrecision))),
+		Meta:    (&genesis.AccountMeta{Owner: rkhKey.Address}).ActorMeta(),
+	}
+	keys = append(keys, rkhKey)
+
 	templ := &genesis.Template{
+		NetworkVersion:   network.Version0,
 		Accounts:         genaccs,
 		Miners:           genms,
 		NetworkName:      "test",
 		Timestamp:        uint64(time.Now().Unix() - 10000), // some time sufficiently far in the past
-		VerifregRootKey:  gen.DefaultVerifregRootkeyActor,
+		VerifregRootKey:  vrk,
 		RemainderAccount: gen.DefaultRemainderAccountActor,
 	}
 
@@ -306,6 +322,7 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
 
 			fullOpts[i].Opts(fulls),
 		)
+
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -319,6 +336,10 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
 		fulls[i].Stb = storageBuilder(fulls[i], mn, node.Options())
 	}
 
+	if _, err := fulls[0].FullNode.WalletImport(ctx, &rkhKey.KeyInfo); err != nil {
+		t.Fatal(err)
+	}
+
 	for i, def := range storage {
 		// TODO: support non-bootstrap miners
 		if i != 0 {
@@ -439,12 +460,26 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 		maddrs = append(maddrs, maddr)
 		genms = append(genms, *genm)
 	}
+
+	rkhKey, err := wallet.GenerateKey(types.KTSecp256k1)
+	if err != nil {
+		return nil, nil
+	}
+
+	vrk := genesis.Actor{
+		Type:    genesis.TAccount,
+		Balance: big.Mul(big.Div(big.NewInt(int64(build.FilBase)), big.NewInt(100)), big.NewInt(int64(build.FilecoinPrecision))),
+		Meta:    (&genesis.AccountMeta{Owner: rkhKey.Address}).ActorMeta(),
+	}
+	keys = append(keys, rkhKey)
+
 	templ := &genesis.Template{
+		NetworkVersion:   network.Version0,
 		Accounts:         genaccs,
 		Miners:           genms,
 		NetworkName:      "test",
 		Timestamp:        uint64(time.Now().Unix()) - (build.BlockDelaySecs * 20000),
-		VerifregRootKey:  gen.DefaultVerifregRootkeyActor,
+		VerifregRootKey:  vrk,
 		RemainderAccount: gen.DefaultRemainderAccountActor,
 	}
 
@@ -466,6 +501,7 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 			node.Test(),
 
 			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
+			node.Override(new(ffiwrapper.Prover), mock.MockProver),
 
 			// so that we subscribe to pubsub topics immediately
 			node.Override(new(dtypes.Bootstrapper), dtypes.Bootstrapper(true)),
@@ -493,11 +529,16 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 			node.Override(new(sectorstorage.Unsealer), node.From(new(*mock.SectorMgr))),
 			node.Override(new(sectorstorage.PieceProvider), node.From(new(*mock.SectorMgr))),
 
-			node.Unset(new(*sectorstorage.Manager)),
 			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
+			node.Override(new(ffiwrapper.Prover), mock.MockProver),
+			node.Unset(new(*sectorstorage.Manager)),
 		))
 	}
 
+	if _, err := fulls[0].FullNode.WalletImport(ctx, &rkhKey.KeyInfo); err != nil {
+		t.Fatal(err)
+	}
+
 	for i, def := range storage {
 		// TODO: support non-bootstrap miners
 
@@ -536,10 +577,9 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 			node.Override(new(sectorstorage.Unsealer), node.From(new(*mock.SectorMgr))),
 			node.Override(new(sectorstorage.PieceProvider), node.From(new(*mock.SectorMgr))),
 
-			node.Unset(new(*sectorstorage.Manager)),
-
 			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
-
+			node.Override(new(ffiwrapper.Prover), mock.MockProver),
+			node.Unset(new(*sectorstorage.Manager)),
 			opts,
 		))