resolve conflicts

node/builder.go

@@ -134,6 +134,8 @@ type Settings struct {
 	Base   bool // Base option applied
 	Config bool // Config option applied
 	Lite   bool // Start node in "lite" mode
+
+	enableLibp2pNode bool
 }
 
 // Basic lotus-app services
@@ -210,15 +212,15 @@ 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 }
 
-func Base(r repo.Repo) Option {
+func Base() Option {
 	return Options(
 		func(s *Settings) error { s.Base = true; return nil }, // mark Base as applied
 		ApplyIf(func(s *Settings) bool { return s.Config },
 			Error(errors.New("the Base() option must be set before Config option")),
 		),
-
-		ApplyIfEnableLibP2P(r, LibP2P),
-
+		ApplyIf(func(s *Settings) bool { return s.enableLibp2pNode },
+			LibP2P,
+		),
 		ApplyIf(isFullOrLiteNode, ChainNode),
 		ApplyIf(IsType(repo.StorageMiner), MinerNode),
 	)
@@ -299,6 +301,10 @@ func Repo(r repo.Repo) Option {
 			Override(new(dtypes.UniversalBlockstore), modules.UniversalBlockstore),
 
 			If(cfg.EnableSplitstore,
+				If(cfg.Splitstore.ColdStoreType == "universal",
+					Override(new(dtypes.ColdBlockstore), From(new(dtypes.UniversalBlockstore)))),
+				If(cfg.Splitstore.ColdStoreType == "discard",
+					Override(new(dtypes.ColdBlockstore), modules.DiscardColdBlockstore)),
 				If(cfg.Splitstore.HotStoreType == "badger",
 					Override(new(dtypes.HotBlockstore), modules.BadgerHotBlockstore)),
 				Override(new(dtypes.SplitBlockstore), modules.SplitBlockstore(cfg)),
@@ -373,7 +379,7 @@ func New(ctx context.Context, opts ...Option) (StopFunc, error) {
 		fx.Options(ctors...),
 		fx.Options(settings.invokes...),
 
-		//fx.NopLogger,
+		fx.NopLogger,
 	)
 
 	// TODO: we probably should have a 'firewall' for Closing signal

node/builder_chain.go

@@ -204,6 +204,7 @@ func FullAPI(out *api.FullNode, fopts ...FullOption) Option {
 	return Options(
 		func(s *Settings) error {
 			s.nodeType = repo.FullNode
+			s.enableLibp2pNode = true
 			return nil
 		},
 		Options(fopts...),

node/builder_miner.go

@@ -68,7 +68,7 @@ func ConfigStorageMiner(c interface{}) Option {
 		return Error(xerrors.New("retrieval pricing policy must be either default or external"))
 	}
 
-	enableLibp2pNode := cfg.Subsystems.EnableStorageMarket // we enable libp2p nodes if the storage market subsystem is enabled, otherwise we don't
+	enableLibp2pNode := cfg.Subsystems.EnableMarkets // we enable libp2p nodes if the storage market subsystem is enabled, otherwise we don't
 
 	return Options(
 		ConfigCommon(&cfg.Common, enableLibp2pNode),
@@ -128,7 +128,7 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(new(stores.SectorIndex), From(new(modules.MinerSealingService))),
 		),
 
-		If(cfg.Subsystems.EnableStorageMarket,
+		If(cfg.Subsystems.EnableMarkets,
 			// Markets
 			Override(new(dtypes.StagingBlockstore), modules.StagingBlockstore),
 			Override(new(dtypes.StagingDAG), modules.StagingDAG),
@@ -204,7 +204,7 @@ func ConfigStorageMiner(c interface{}) Option {
 	)
 }
 
-func StorageMiner(out *api.StorageMiner) Option {
+func StorageMiner(out *api.StorageMiner, subsystemsCfg config.MinerSubsystemConfig) Option {
 	return Options(
 		ApplyIf(func(s *Settings) bool { return s.Config },
 			Error(errors.New("the StorageMiner option must be set before Config option")),
@@ -212,6 +212,7 @@ func StorageMiner(out *api.StorageMiner) Option {
 
 		func(s *Settings) error {
 			s.nodeType = repo.StorageMiner
+			s.enableLibp2pNode = subsystemsCfg.EnableMarkets
 			return nil
 		},
 

node/config/def.go

@@ -62,7 +62,7 @@ type MinerSubsystemConfig struct {
 	EnableMining        bool
 	EnableSealing       bool
 	EnableSectorStorage bool
-	EnableStorageMarket bool
+	EnableMarkets       bool
 
 	SealerApiInfo      string // if EnableSealing == false
 	SectorIndexApiInfo string // if EnableSectorStorage == false
@@ -136,6 +136,13 @@ type SealingConfig struct {
 	// Run sector finalization before submitting sector proof to the chain
 	FinalizeEarly bool
 
+	// Whether to use available miner balance for sector collateral instead of sending it with each message
+	CollateralFromMinerBalance bool
+	// Minimum available balance to keep in the miner actor before sending it with messages
+	AvailableBalanceBuffer types.FIL
+	// Don't send collateral with messages even if there is no available balance in the miner actor
+	DisableCollateralFallback bool
+
 	// enable / disable precommit batching (takes effect after nv13)
 	BatchPreCommits bool
 	// maximum precommit batch size - batches will be sent immediately above this size
@@ -193,9 +200,10 @@ type MinerFeeConfig struct {
 }
 
 type MinerAddressConfig struct {
-	PreCommitControl []string
-	CommitControl    []string
-	TerminateControl []string
+	PreCommitControl   []string
+	CommitControl      []string
+	TerminateControl   []string
+	DealPublishControl []string
 
 	// DisableOwnerFallback disables usage of the owner address for messages
 	// sent automatically
@@ -240,12 +248,9 @@ type Chainstore struct {
 }
 
 type Splitstore struct {
-	HotStoreType         string
-	TrackingStoreType    string
-	MarkSetType          string
-	EnableFullCompaction bool
-	EnableGC             bool // EXPERIMENTAL
-	Archival             bool
+	ColdStoreType string
+	HotStoreType  string
+	MarkSetType   string
 }
 
 // // Full Node
@@ -316,7 +321,9 @@ func DefaultFullNode() *FullNode {
 		Chainstore: Chainstore{
 			EnableSplitstore: false,
 			Splitstore: Splitstore{
-				HotStoreType: "badger",
+				ColdStoreType: "universal",
+				HotStoreType:  "badger",
+				MarkSetType:   "map",
 			},
 		},
 	}
@@ -334,6 +341,10 @@ func DefaultStorageMiner() *StorageMiner {
 			AlwaysKeepUnsealedCopy:    true,
 			FinalizeEarly:             false,
 
+			CollateralFromMinerBalance: false,
+			AvailableBalanceBuffer:     types.FIL(big.Zero()),
+			DisableCollateralFallback:  false,
+
 			BatchPreCommits:     true,
 			MaxPreCommitBatch:   miner5.PreCommitSectorBatchMaxSize, // up to 256 sectors
 			PreCommitBatchWait:  Duration(24 * time.Hour),           // this should be less than 31.5 hours, which is the expiration of a precommit ticket
@@ -399,7 +410,7 @@ func DefaultStorageMiner() *StorageMiner {
 			EnableMining:        true,
 			EnableSealing:       true,
 			EnableSectorStorage: true,
-			EnableStorageMarket: true,
+			EnableMarkets:       true,
 		},
 
 		Fees: MinerFeeConfig{
@@ -422,8 +433,10 @@ func DefaultStorageMiner() *StorageMiner {
 		},
 
 		Addresses: MinerAddressConfig{
-			PreCommitControl: []string{},
-			CommitControl:    []string{},
+			PreCommitControl:   []string{},
+			CommitControl:      []string{},
+			TerminateControl:   []string{},
+			DealPublishControl: []string{},
 		},
 	}
 	cfg.Common.API.ListenAddress = "/ip4/127.0.0.1/tcp/2345/http"

node/enable_libp2p.go

@@ -1,40 +0,0 @@
-package node
-
-import (
-	"github.com/filecoin-project/lotus/node/config"
-	"github.com/filecoin-project/lotus/node/repo"
-)
-
-func ApplyIfEnableLibP2P(r repo.Repo, opts ...Option) Option {
-	return ApplyIf(func(settings *Settings) bool {
-		lr, err := r.Lock(settings.nodeType)
-		if err != nil {
-			// TODO: log error
-			return false
-		}
-		c, err := lr.Config()
-		if err != nil {
-			// TODO: log error
-			return false
-		}
-
-		defer lr.Close() //nolint:errcheck
-
-		switch settings.nodeType {
-		case repo.FullNode:
-			return true
-		case repo.StorageMiner:
-			cfg, ok := c.(*config.StorageMiner)
-			if !ok {
-				// TODO: log error
-				return false
-			}
-
-			enableLibP2P := cfg.Subsystems.EnableStorageMarket
-			return enableLibP2P
-		default:
-			// TODO: log error
-			return false
-		}
-	}, opts...)
-}

node/impl/client/client.go

@@ -684,6 +684,8 @@ func readSubscribeEvents(ctx context.Context, dealID retrievalmarket.DealID, sub
 			return nil
 		case rm.DealStatusRejected:
 			return xerrors.Errorf("Retrieval Proposal Rejected: %s", state.Message)
+		case rm.DealStatusCancelled:
+			return xerrors.Errorf("Retrieval was cancelled externally: %s", state.Message)
 		case
 			rm.DealStatusDealNotFound,
 			rm.DealStatusErrored:

node/impl/storminer.go

@@ -429,6 +429,11 @@ func (sm *StorageMinerAPI) MarketListRetrievalDeals(ctx context.Context) ([]retr
 	deals := sm.RetrievalProvider.ListDeals()
 
 	for _, deal := range deals {
+		if deal.ChannelID != nil {
+			if deal.ChannelID.Initiator == "" || deal.ChannelID.Responder == "" {
+				deal.ChannelID = nil // don't try to push unparsable peer IDs over jsonrpc
+			}
+		}
 		out = append(out, deal)
 	}
 

node/modules/blockstore.go

@@ -37,6 +37,10 @@ func UniversalBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.Locked
 	return bs, err
 }
 
+func DiscardColdBlockstore(lc fx.Lifecycle, bs dtypes.UniversalBlockstore) (dtypes.ColdBlockstore, error) {
+	return blockstore.NewDiscardStore(bs), nil
+}
+
 func BadgerHotBlockstore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.HotBlockstore, error) {
 	path, err := r.SplitstorePath()
 	if err != nil {
@@ -66,19 +70,16 @@ func BadgerHotBlockstore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.HotBlocksto
 	return bs, nil
 }
 
-func SplitBlockstore(cfg *config.Chainstore) func(lc fx.Lifecycle, r repo.LockedRepo, ds dtypes.MetadataDS, cold dtypes.UniversalBlockstore, hot dtypes.HotBlockstore) (dtypes.SplitBlockstore, error) {
-	return func(lc fx.Lifecycle, r repo.LockedRepo, ds dtypes.MetadataDS, cold dtypes.UniversalBlockstore, hot dtypes.HotBlockstore) (dtypes.SplitBlockstore, error) {
+func SplitBlockstore(cfg *config.Chainstore) func(lc fx.Lifecycle, r repo.LockedRepo, ds dtypes.MetadataDS, cold dtypes.ColdBlockstore, hot dtypes.HotBlockstore) (dtypes.SplitBlockstore, error) {
+	return func(lc fx.Lifecycle, r repo.LockedRepo, ds dtypes.MetadataDS, cold dtypes.ColdBlockstore, hot dtypes.HotBlockstore) (dtypes.SplitBlockstore, error) {
 		path, err := r.SplitstorePath()
 		if err != nil {
 			return nil, err
 		}
 
 		cfg := &splitstore.Config{
-			TrackingStoreType:    cfg.Splitstore.TrackingStoreType,
-			MarkSetType:          cfg.Splitstore.MarkSetType,
-			EnableFullCompaction: cfg.Splitstore.EnableFullCompaction,
-			EnableGC:             cfg.Splitstore.EnableGC,
-			Archival:             cfg.Splitstore.Archival,
+			MarkSetType:       cfg.Splitstore.MarkSetType,
+			DiscardColdBlocks: cfg.Splitstore.ColdStoreType == "discard",
 		}
 		ss, err := splitstore.Open(path, ds, hot, cold, cfg)
 		if err != nil {

node/modules/dtypes/storage.go

@@ -22,9 +22,12 @@ import (
 type MetadataDS datastore.Batching
 
 type (
-	// UniversalBlockstore is the cold blockstore.
+	// UniversalBlockstore is the universal blockstore backend.
 	UniversalBlockstore blockstore.Blockstore
 
+	// ColdBlockstore is the Cold blockstore abstraction for the splitstore
+	ColdBlockstore blockstore.Blockstore
+
 	// HotBlockstore is the Hot blockstore abstraction for the splitstore
 	HotBlockstore blockstore.Blockstore
 

node/modules/storageminer.go

@@ -188,6 +188,15 @@ func AddressSelector(addrConf *config.MinerAddressConfig) func() (*storage.Addre
 			as.TerminateControl = append(as.TerminateControl, addr)
 		}
 
+		for _, s := range addrConf.DealPublishControl {
+			addr, err := address.NewFromString(s)
+			if err != nil {
+				return nil, xerrors.Errorf("parsing deal publishing control address: %w", err)
+			}
+
+			as.DealPublishControl = append(as.DealPublishControl, addr)
+		}
+
 		return as, nil
 	}
 }
@@ -691,7 +700,6 @@ func LocalStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStora
 }
 
 func RemoteStorage(lstor *stores.Local, si stores.SectorIndex, sa sectorstorage.StorageAuth, sc sectorstorage.SealerConfig) *stores.Remote {
-	fmt.Printf("setting RemoteStorage: %#v \n", sa) // TODO(anteva): remove me prior to merge to master
 	return stores.NewRemote(lstor, si, http.Header(sa), sc.ParallelFetchLimit, &stores.DefaultPartialFileHandler{})
 }
 
@@ -724,11 +732,11 @@ func StorageAuth(ctx helpers.MetricsCtx, ca v0api.Common) (sectorstorage.Storage
 	return sectorstorage.StorageAuth(headers), nil
 }
 
-func StorageAuthWithURL(url string) func(ctx helpers.MetricsCtx, ca v0api.Common) (sectorstorage.StorageAuth, error) {
+func StorageAuthWithURL(apiInfo string) func(ctx helpers.MetricsCtx, ca v0api.Common) (sectorstorage.StorageAuth, error) {
 	return func(ctx helpers.MetricsCtx, ca v0api.Common) (sectorstorage.StorageAuth, error) {
-		s := strings.Split(url, ":")
+		s := strings.Split(apiInfo, ":")
 		if len(s) != 2 {
-			return nil, errors.New("unexpected format of URL")
+			return nil, errors.New("unexpected format of `apiInfo`")
 		}
 		headers := http.Header{}
 		headers.Add("Authorization", "Bearer "+s[0])
@@ -873,6 +881,10 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
 				AlwaysKeepUnsealedCopy:    cfg.AlwaysKeepUnsealedCopy,
 				FinalizeEarly:             cfg.FinalizeEarly,
 
+				CollateralFromMinerBalance: cfg.CollateralFromMinerBalance,
+				AvailableBalanceBuffer:     types.FIL(cfg.AvailableBalanceBuffer),
+				DisableCollateralFallback:  cfg.DisableCollateralFallback,
+
 				BatchPreCommits:     cfg.BatchPreCommits,
 				MaxPreCommitBatch:   cfg.MaxPreCommitBatch,
 				PreCommitBatchWait:  config.Duration(cfg.PreCommitBatchWait),
@@ -903,6 +915,10 @@ func ToSealingConfig(cfg *config.StorageMiner) sealiface.Config {
 		AlwaysKeepUnsealedCopy:    cfg.Sealing.AlwaysKeepUnsealedCopy,
 		FinalizeEarly:             cfg.Sealing.FinalizeEarly,
 
+		CollateralFromMinerBalance: cfg.Sealing.CollateralFromMinerBalance,
+		AvailableBalanceBuffer:     types.BigInt(cfg.Sealing.AvailableBalanceBuffer),
+		DisableCollateralFallback:  cfg.Sealing.DisableCollateralFallback,
+
 		BatchPreCommits:     cfg.Sealing.BatchPreCommits,
 		MaxPreCommitBatch:   cfg.Sealing.MaxPreCommitBatch,
 		PreCommitBatchWait:  time.Duration(cfg.Sealing.PreCommitBatchWait),