resolve merge conflicts

node/builder.go

@@ -207,6 +207,203 @@ func isFullOrLiteNode(s *Settings) bool { return s.nodeType == repo.FullNode }
 func isFullNode(s *Settings) bool       { return s.nodeType == repo.FullNode && !s.Lite }
 func isLiteNode(s *Settings) bool       { return s.nodeType == repo.FullNode && s.Lite }
 
+<<<<<<< HEAD
+=======
+// Chain node provides access to the Filecoin blockchain, by setting up a full
+// validator node, or by delegating some actions to other nodes (lite mode)
+var ChainNode = Options(
+	// Full node or lite node
+	// TODO: Fix offline mode
+
+	// Consensus settings
+	Override(new(dtypes.DrandSchedule), modules.BuiltinDrandConfig),
+	Override(new(stmgr.UpgradeSchedule), stmgr.DefaultUpgradeSchedule()),
+	Override(new(dtypes.NetworkName), modules.NetworkName),
+	Override(new(modules.Genesis), modules.ErrorGenesis),
+	Override(new(dtypes.AfterGenesisSet), modules.SetGenesis),
+	Override(SetGenesisKey, modules.DoSetGenesis),
+	Override(new(beacon.Schedule), modules.RandomSchedule),
+
+	// Network bootstrap
+	Override(new(dtypes.BootstrapPeers), modules.BuiltinBootstrap),
+	Override(new(dtypes.DrandBootstrap), modules.DrandBootstrap),
+
+	// Consensus: crypto dependencies
+	Override(new(ffiwrapper.Verifier), ffiwrapper.ProofVerifier),
+
+	// Consensus: VM
+	Override(new(vm.SyscallBuilder), vm.Syscalls),
+
+	// Consensus: Chain storage/access
+	Override(new(*store.ChainStore), modules.ChainStore),
+	Override(new(*stmgr.StateManager), modules.StateManager),
+	Override(new(dtypes.ChainBitswap), modules.ChainBitswap),
+	Override(new(dtypes.ChainBlockService), modules.ChainBlockService), // todo: unused
+
+	// Consensus: Chain sync
+
+	// We don't want the SyncManagerCtor to be used as an fx constructor, but rather as a value.
+	// It will be called implicitly by the Syncer constructor.
+	Override(new(chain.SyncManagerCtor), func() chain.SyncManagerCtor { return chain.NewSyncManager }),
+	Override(new(*chain.Syncer), modules.NewSyncer),
+	Override(new(exchange.Client), exchange.NewClient),
+
+	// Chain networking
+	Override(new(*hello.Service), hello.NewHelloService),
+	Override(new(exchange.Server), exchange.NewServer),
+	Override(new(*peermgr.PeerMgr), peermgr.NewPeerMgr),
+
+	// Chain mining API dependencies
+	Override(new(*slashfilter.SlashFilter), modules.NewSlashFilter),
+
+	// Service: Message Pool
+	Override(new(dtypes.DefaultMaxFeeFunc), modules.NewDefaultMaxFeeFunc),
+	Override(new(*messagepool.MessagePool), modules.MessagePool),
+	Override(new(*dtypes.MpoolLocker), new(dtypes.MpoolLocker)),
+
+	// Shared graphsync (markets, serving chain)
+	Override(new(dtypes.Graphsync), modules.Graphsync(config.DefaultFullNode().Client.SimultaneousTransfers)),
+
+	// Service: Wallet
+	Override(new(*messagesigner.MessageSigner), messagesigner.NewMessageSigner),
+	Override(new(*wallet.LocalWallet), wallet.NewWallet),
+	Override(new(wallet.Default), From(new(*wallet.LocalWallet))),
+	Override(new(api.Wallet), From(new(wallet.MultiWallet))),
+
+	// Service: Payment channels
+	Override(new(paychmgr.PaychAPI), From(new(modules.PaychAPI))),
+	Override(new(*paychmgr.Store), modules.NewPaychStore),
+	Override(new(*paychmgr.Manager), modules.NewManager),
+	Override(HandlePaymentChannelManagerKey, modules.HandlePaychManager),
+	Override(SettlePaymentChannelsKey, settler.SettlePaymentChannels),
+
+	// Markets (common)
+	Override(new(*discoveryimpl.Local), modules.NewLocalDiscovery),
+
+	// Markets (retrieval)
+	Override(new(discovery.PeerResolver), modules.RetrievalResolver),
+	Override(new(retrievalmarket.RetrievalClient), modules.RetrievalClient),
+	Override(new(dtypes.ClientDataTransfer), modules.NewClientGraphsyncDataTransfer),
+
+	// Markets (storage)
+	Override(new(*market.FundManager), market.NewFundManager),
+	Override(new(dtypes.ClientDatastore), modules.NewClientDatastore),
+	Override(new(storagemarket.StorageClient), modules.StorageClient),
+	Override(new(storagemarket.StorageClientNode), storageadapter.NewClientNodeAdapter),
+	Override(HandleMigrateClientFundsKey, modules.HandleMigrateClientFunds),
+
+	Override(new(*full.GasPriceCache), full.NewGasPriceCache),
+
+	// Lite node API
+	ApplyIf(isLiteNode,
+		Override(new(messagepool.Provider), messagepool.NewProviderLite),
+		Override(new(messagesigner.MpoolNonceAPI), From(new(modules.MpoolNonceAPI))),
+		Override(new(full.ChainModuleAPI), From(new(api.Gateway))),
+		Override(new(full.GasModuleAPI), From(new(api.Gateway))),
+		Override(new(full.MpoolModuleAPI), From(new(api.Gateway))),
+		Override(new(full.StateModuleAPI), From(new(api.Gateway))),
+		Override(new(stmgr.StateManagerAPI), rpcstmgr.NewRPCStateManager),
+	),
+
+	// Full node API / service startup
+	ApplyIf(isFullNode,
+		Override(new(messagepool.Provider), messagepool.NewProvider),
+		Override(new(messagesigner.MpoolNonceAPI), From(new(*messagepool.MessagePool))),
+		Override(new(full.ChainModuleAPI), From(new(full.ChainModule))),
+		Override(new(full.GasModuleAPI), From(new(full.GasModule))),
+		Override(new(full.MpoolModuleAPI), From(new(full.MpoolModule))),
+		Override(new(full.StateModuleAPI), From(new(full.StateModule))),
+		Override(new(stmgr.StateManagerAPI), From(new(*stmgr.StateManager))),
+
+		Override(RunHelloKey, modules.RunHello),
+		Override(RunChainExchangeKey, modules.RunChainExchange),
+		Override(RunPeerMgrKey, modules.RunPeerMgr),
+		Override(HandleIncomingMessagesKey, modules.HandleIncomingMessages),
+		Override(HandleIncomingBlocksKey, modules.HandleIncomingBlocks),
+	),
+)
+
+var MinerNode = Options(
+	// API dependencies
+	Override(new(api.Common), From(new(common.CommonAPI))),
+	Override(new(sectorstorage.StorageAuth), modules.StorageAuth),
+
+	// Actor config
+	Override(new(dtypes.MinerAddress), modules.MinerAddress),
+	Override(new(dtypes.MinerID), modules.MinerID),
+	Override(new(abi.RegisteredSealProof), modules.SealProofType),
+	Override(new(dtypes.NetworkName), modules.StorageNetworkName),
+
+	// Sector storage
+	Override(new(*stores.Index), stores.NewIndex),
+	Override(new(stores.SectorIndex), From(new(*stores.Index))),
+	Override(new(stores.LocalStorage), From(new(repo.LockedRepo))),
+	Override(new(*sectorstorage.Manager), modules.SectorStorage),
+	Override(new(sectorstorage.SectorManager), From(new(*sectorstorage.Manager))),
+	Override(new(storiface.WorkerReturn), From(new(sectorstorage.SectorManager))),
+
+	// Sector storage: Proofs
+	Override(new(ffiwrapper.Verifier), ffiwrapper.ProofVerifier),
+	Override(new(ffiwrapper.Prover), ffiwrapper.ProofProver),
+	Override(new(storage2.Prover), From(new(sectorstorage.SectorManager))),
+
+	// Sealing
+	Override(new(sealing.SectorIDCounter), modules.SectorIDCounter),
+	Override(GetParamsKey, modules.GetParams),
+
+	// Mining / proving
+	Override(new(*slashfilter.SlashFilter), modules.NewSlashFilter),
+	Override(new(*storage.Miner), modules.StorageMiner(config.DefaultStorageMiner().Fees)),
+	Override(new(*miner.Miner), modules.SetupBlockProducer),
+	Override(new(gen.WinningPoStProver), storage.NewWinningPoStProver),
+
+	Override(new(*storage.AddressSelector), modules.AddressSelector(nil)),
+
+	// Markets
+	Override(new(dtypes.StagingMultiDstore), modules.StagingMultiDatastore),
+	Override(new(dtypes.StagingBlockstore), modules.StagingBlockstore),
+	Override(new(dtypes.StagingDAG), modules.StagingDAG),
+	Override(new(dtypes.StagingGraphsync), modules.StagingGraphsync),
+	Override(new(dtypes.ProviderPieceStore), modules.NewProviderPieceStore),
+	Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
+
+	// Markets (retrieval)
+	Override(new(retrievalmarket.RetrievalProvider), modules.RetrievalProvider),
+	Override(new(dtypes.RetrievalDealFilter), modules.RetrievalDealFilter(nil)),
+	Override(HandleRetrievalKey, modules.HandleRetrieval),
+
+	// Markets (storage)
+	Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
+	Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
+	Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(nil)),
+	Override(new(storagemarket.StorageProvider), modules.StorageProvider),
+	Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{})),
+	Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(nil, nil)),
+	Override(HandleMigrateProviderFundsKey, modules.HandleMigrateProviderFunds),
+	Override(HandleDealsKey, modules.HandleDeals),
+
+	// Config (todo: get a real property system)
+	Override(new(dtypes.ConsiderOnlineStorageDealsConfigFunc), modules.NewConsiderOnlineStorageDealsConfigFunc),
+	Override(new(dtypes.SetConsiderOnlineStorageDealsConfigFunc), modules.NewSetConsideringOnlineStorageDealsFunc),
+	Override(new(dtypes.ConsiderOnlineRetrievalDealsConfigFunc), modules.NewConsiderOnlineRetrievalDealsConfigFunc),
+	Override(new(dtypes.SetConsiderOnlineRetrievalDealsConfigFunc), modules.NewSetConsiderOnlineRetrievalDealsConfigFunc),
+	Override(new(dtypes.StorageDealPieceCidBlocklistConfigFunc), modules.NewStorageDealPieceCidBlocklistConfigFunc),
+	Override(new(dtypes.SetStorageDealPieceCidBlocklistConfigFunc), modules.NewSetStorageDealPieceCidBlocklistConfigFunc),
+	Override(new(dtypes.ConsiderOfflineStorageDealsConfigFunc), modules.NewConsiderOfflineStorageDealsConfigFunc),
+	Override(new(dtypes.SetConsiderOfflineStorageDealsConfigFunc), modules.NewSetConsideringOfflineStorageDealsFunc),
+	Override(new(dtypes.ConsiderOfflineRetrievalDealsConfigFunc), modules.NewConsiderOfflineRetrievalDealsConfigFunc),
+	Override(new(dtypes.SetConsiderOfflineRetrievalDealsConfigFunc), modules.NewSetConsiderOfflineRetrievalDealsConfigFunc),
+	Override(new(dtypes.ConsiderVerifiedStorageDealsConfigFunc), modules.NewConsiderVerifiedStorageDealsConfigFunc),
+	Override(new(dtypes.SetConsiderVerifiedStorageDealsConfigFunc), modules.NewSetConsideringVerifiedStorageDealsFunc),
+	Override(new(dtypes.ConsiderUnverifiedStorageDealsConfigFunc), modules.NewConsiderUnverifiedStorageDealsConfigFunc),
+	Override(new(dtypes.SetConsiderUnverifiedStorageDealsConfigFunc), modules.NewSetConsideringUnverifiedStorageDealsFunc),
+	Override(new(dtypes.SetSealingConfigFunc), modules.NewSetSealConfigFunc),
+	Override(new(dtypes.GetSealingConfigFunc), modules.NewGetSealConfigFunc),
+	Override(new(dtypes.SetExpectedSealDurationFunc), modules.NewSetExpectedSealDurationFunc),
+	Override(new(dtypes.GetExpectedSealDurationFunc), modules.NewGetExpectedSealDurationFunc),
+)
+
+>>>>>>> master
 // Online sets up basic libp2p node
 func Online() Option {
 

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"
 )
@@ -93,6 +95,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
 
@@ -248,6 +274,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

@@ -67,7 +67,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

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/storminer.go

@@ -31,6 +31,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"
@@ -205,13 +206,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
 }
@@ -344,10 +345,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/storageminer.go

@@ -100,7 +100,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)
 	}
 
@@ -203,6 +203,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
@@ -219,6 +220,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
@@ -236,7 +238,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
 		}
@@ -842,6 +844,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
@@ -857,6 +875,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

@@ -66,6 +66,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
 
@@ -162,6 +163,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")
@@ -178,6 +188,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")
@@ -226,6 +237,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")
@@ -239,6 +251,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")
@@ -247,11 +260,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")
@@ -259,3 +301,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

@@ -278,13 +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,
 	}
 
@@ -309,6 +322,7 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
 
 			fullOpts[i].Opts(fulls),
 		)
+
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -322,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 {
@@ -442,13 +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,
 	}
 
@@ -473,6 +504,7 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 
 			node.Override(new(messagepool.Provider), messagepool.NewProvider),
 			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)),
@@ -496,10 +528,15 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 				return mock.NewMockSectorMgr(nil), nil
 			}),
 			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
 
@@ -541,7 +578,8 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
 				return mgr, nil
 			}),
 			node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
-			//node.Unset(new(*sectorstorage.Manager)),
+			node.Override(new(ffiwrapper.Prover), mock.MockProver),
+			// node.Unset(new(*sectorstorage.Manager)),
 			opts,
 		))