cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge branch 'master' into refactor/lib/blockstore

Browse Source
This commit is contained in:
Raúl Kripalani
2021-02-28 19:55:23 +00:00
parent 8601e5da3a 191a05da48
commit 7f0f7d0b36
149 changed files with 4352 additions and 1640 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+240 -191
View File
@@ -171,6 +171,7 @@ type Settings struct {
Lite bool // Start node in "lite" mode
}
// Basic lotus-app services
func defaults() []Option {
return []Option{
// global system journal.
@@ -184,65 +185,256 @@ func defaults() []Option {
return metricsi.CtxScope(context.Background(), "lotus")
}),
Override(new(record.Validator), modules.RecordValidator),
Override(new(dtypes.Bootstrapper), dtypes.Bootstrapper(false)),
Override(new(dtypes.ShutdownChan), make(chan struct{})),
// Filecoin modules
}
}
func libp2p() Option {
return Options(
Override(new(peerstore.Peerstore), pstoremem.NewPeerstore),
var LibP2P = Options(
// Host config
Override(new(dtypes.Bootstrapper), dtypes.Bootstrapper(false)),
Override(DefaultTransportsKey, lp2p.DefaultTransports),
// Host dependencies
Override(new(peerstore.Peerstore), pstoremem.NewPeerstore),
Override(PstoreAddSelfKeysKey, lp2p.PstoreAddSelfKeys),
Override(StartListeningKey, lp2p.StartListening(config.DefaultFullNode().Libp2p.ListenAddresses)),
Override(new(lp2p.RawHost), lp2p.Host),
Override(new(host.Host), lp2p.RoutedHost),
Override(new(lp2p.BaseIpfsRouting), lp2p.DHTRouting(dht.ModeAuto)),
// Host settings
Override(DefaultTransportsKey, lp2p.DefaultTransports),
Override(AddrsFactoryKey, lp2p.AddrsFactory(nil, nil)),
Override(SmuxTransportKey, lp2p.SmuxTransport(true)),
Override(RelayKey, lp2p.NoRelay()),
Override(SecurityKey, lp2p.Security(true, false)),
Override(DiscoveryHandlerKey, lp2p.DiscoveryHandler),
Override(AddrsFactoryKey, lp2p.AddrsFactory(nil, nil)),
Override(SmuxTransportKey, lp2p.SmuxTransport(true)),
Override(RelayKey, lp2p.NoRelay()),
Override(SecurityKey, lp2p.Security(true, false)),
// Host
Override(new(lp2p.RawHost), lp2p.Host),
Override(new(host.Host), lp2p.RoutedHost),
Override(new(lp2p.BaseIpfsRouting), lp2p.DHTRouting(dht.ModeAuto)),
Override(BaseRoutingKey, lp2p.BaseRouting),
Override(new(routing.Routing), lp2p.Routing),
Override(DiscoveryHandlerKey, lp2p.DiscoveryHandler),
Override(NatPortMapKey, lp2p.NatPortMap),
Override(BandwidthReporterKey, lp2p.BandwidthCounter),
// Routing
Override(new(record.Validator), modules.RecordValidator),
Override(BaseRoutingKey, lp2p.BaseRouting),
Override(new(routing.Routing), lp2p.Routing),
Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
Override(AutoNATSvcKey, lp2p.AutoNATService),
// Services
Override(NatPortMapKey, lp2p.NatPortMap),
Override(BandwidthReporterKey, lp2p.BandwidthCounter),
Override(AutoNATSvcKey, lp2p.AutoNATService),
Override(new(*dtypes.ScoreKeeper), lp2p.ScoreKeeper),
Override(new(*pubsub.PubSub), lp2p.GossipSub),
Override(new(*config.Pubsub), func(bs dtypes.Bootstrapper) *config.Pubsub {
return &config.Pubsub{
Bootstrapper: bool(bs),
}
}),
// Services (pubsub)
Override(new(*dtypes.ScoreKeeper), lp2p.ScoreKeeper),
Override(new(*pubsub.PubSub), lp2p.GossipSub),
Override(new(*config.Pubsub), func(bs dtypes.Bootstrapper) *config.Pubsub {
return &config.Pubsub{
Bootstrapper: bool(bs),
}
}),
Override(PstoreAddSelfKeysKey, lp2p.PstoreAddSelfKeys),
Override(StartListeningKey, lp2p.StartListening(config.DefaultFullNode().Libp2p.ListenAddresses)),
Override(new(*conngater.BasicConnectionGater), lp2p.ConnGater),
Override(ConnGaterKey, lp2p.ConnGaterOption),
)
}
// Services (connection management)
Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
Override(new(*conngater.BasicConnectionGater), lp2p.ConnGater),
Override(ConnGaterKey, lp2p.ConnGaterOption),
)
func isType(t repo.RepoType) func(s *Settings) bool {
return func(s *Settings) bool { return s.nodeType == t }
}
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 }
// 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.WalletAPI), From(new(wallet.MultiWallet))),
// Service: Payment channels
Override(new(*paychmgr.Store), paychmgr.NewStore),
Override(new(*paychmgr.Manager), paychmgr.NewManager),
Override(HandlePaymentChannelManagerKey, paychmgr.HandleManager),
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),
// Lite node API
ApplyIf(isLiteNode,
Override(new(messagesigner.MpoolNonceAPI), From(new(modules.MpoolNonceAPI))),
Override(new(full.ChainModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.GasModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.MpoolModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.StateModuleAPI), From(new(api.GatewayAPI))),
Override(new(stmgr.StateManagerAPI), modules.NewRPCStateManager),
),
// Full node API / service startup
ApplyIf(isFullNode,
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(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)),
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),
)
// Online sets up basic libp2p node
func Online() Option {
isFullOrLiteNode := func(s *Settings) bool { return s.nodeType == repo.FullNode }
isFullNode := func(s *Settings) bool { return s.nodeType == repo.FullNode && !s.Lite }
isLiteNode := func(s *Settings) bool { return s.nodeType == repo.FullNode && s.Lite }
return Options(
// make sure that online is applied before Config.
@@ -252,158 +444,10 @@ func Online() Option {
Error(errors.New("the Online option must be set before Config option")),
),
libp2p(),
LibP2P,
// common
Override(new(*slashfilter.SlashFilter), modules.NewSlashFilter),
// Full node or lite node
ApplyIf(isFullOrLiteNode,
// TODO: Fix offline mode
Override(new(dtypes.BootstrapPeers), modules.BuiltinBootstrap),
Override(new(dtypes.DrandBootstrap), modules.DrandBootstrap),
Override(new(dtypes.DrandSchedule), modules.BuiltinDrandConfig),
Override(new(ffiwrapper.Verifier), ffiwrapper.ProofVerifier),
Override(new(vm.SyscallBuilder), vm.Syscalls),
Override(new(*store.ChainStore), modules.ChainStore),
Override(new(stmgr.UpgradeSchedule), stmgr.DefaultUpgradeSchedule()),
Override(new(*stmgr.StateManager), modules.StateManager),
Override(new(*wallet.LocalWallet), wallet.NewWallet),
Override(new(wallet.Default), From(new(*wallet.LocalWallet))),
Override(new(api.WalletAPI), From(new(wallet.MultiWallet))),
Override(new(*messagesigner.MessageSigner), messagesigner.NewMessageSigner),
Override(new(dtypes.ChainBitswap), modules.ChainBitswap),
Override(new(dtypes.ChainBlockService), modules.ChainBlockService),
// Filecoin services
// 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),
Override(new(*messagepool.MessagePool), modules.MessagePool),
Override(new(dtypes.DefaultMaxFeeFunc), modules.NewDefaultMaxFeeFunc),
Override(new(modules.Genesis), modules.ErrorGenesis),
Override(new(dtypes.AfterGenesisSet), modules.SetGenesis),
Override(SetGenesisKey, modules.DoSetGenesis),
Override(new(dtypes.NetworkName), modules.NetworkName),
Override(new(*hello.Service), hello.NewHelloService),
Override(new(exchange.Server), exchange.NewServer),
Override(new(*peermgr.PeerMgr), peermgr.NewPeerMgr),
Override(new(dtypes.Graphsync), modules.Graphsync(config.DefaultFullNode().Client.SimultaneousTransfers)),
Override(new(*dtypes.MpoolLocker), new(dtypes.MpoolLocker)),
Override(new(*discoveryimpl.Local), modules.NewLocalDiscovery),
Override(new(discovery.PeerResolver), modules.RetrievalResolver),
Override(new(retrievalmarket.RetrievalClient), modules.RetrievalClient),
Override(new(dtypes.ClientDatastore), modules.NewClientDatastore),
Override(new(dtypes.ClientDataTransfer), modules.NewClientGraphsyncDataTransfer),
Override(new(storagemarket.StorageClient), modules.StorageClient),
Override(new(storagemarket.StorageClientNode), storageadapter.NewClientNodeAdapter),
Override(new(beacon.Schedule), modules.RandomSchedule),
Override(new(*paychmgr.Store), paychmgr.NewStore),
Override(new(*paychmgr.Manager), paychmgr.NewManager),
Override(new(*market.FundManager), market.NewFundManager),
Override(HandleMigrateClientFundsKey, modules.HandleMigrateClientFunds),
Override(HandlePaymentChannelManagerKey, paychmgr.HandleManager),
Override(SettlePaymentChannelsKey, settler.SettlePaymentChannels),
),
// Lite node
ApplyIf(isLiteNode,
Override(new(messagesigner.MpoolNonceAPI), From(new(modules.MpoolNonceAPI))),
Override(new(full.ChainModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.GasModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.MpoolModuleAPI), From(new(api.GatewayAPI))),
Override(new(full.StateModuleAPI), From(new(api.GatewayAPI))),
Override(new(stmgr.StateManagerAPI), modules.NewRPCStateManager),
),
// Full node
ApplyIf(isFullNode,
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),
),
// miner
ApplyIf(isType(repo.StorageMiner),
Override(new(api.Common), From(new(common.CommonAPI))),
Override(new(sectorstorage.StorageAuth), modules.StorageAuth),
Override(new(*stores.Index), stores.NewIndex),
Override(new(stores.SectorIndex), From(new(*stores.Index))),
Override(new(dtypes.MinerID), modules.MinerID),
Override(new(dtypes.MinerAddress), modules.MinerAddress),
Override(new(abi.RegisteredSealProof), modules.SealProofType),
Override(new(stores.LocalStorage), From(new(repo.LockedRepo))),
Override(new(sealing.SectorIDCounter), modules.SectorIDCounter),
Override(new(*sectorstorage.Manager), modules.SectorStorage),
Override(new(ffiwrapper.Verifier), ffiwrapper.ProofVerifier),
Override(new(sectorstorage.SectorManager), From(new(*sectorstorage.Manager))),
Override(new(storage2.Prover), From(new(sectorstorage.SectorManager))),
Override(new(storiface.WorkerReturn), From(new(sectorstorage.SectorManager))),
Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
Override(new(*storage.Miner), modules.StorageMiner(config.DefaultStorageMiner().Fees)),
Override(new(*storage.AddressSelector), modules.AddressSelector(nil)),
Override(new(dtypes.NetworkName), modules.StorageNetworkName),
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(retrievalmarket.RetrievalProvider), modules.RetrievalProvider),
Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
Override(new(dtypes.ProviderPieceStore), modules.NewProviderPieceStore),
Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(nil)),
Override(new(dtypes.RetrievalDealFilter), modules.RetrievalDealFilter(nil)),
Override(new(storagemarket.StorageProvider), modules.StorageProvider),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(nil)),
Override(HandleMigrateProviderFundsKey, modules.HandleMigrateProviderFunds),
Override(HandleRetrievalKey, modules.HandleRetrieval),
Override(GetParamsKey, modules.GetParams),
Override(HandleDealsKey, modules.HandleDeals),
Override(new(gen.WinningPoStProver), storage.NewWinningPoStProver),
Override(new(*miner.Miner), modules.SetupBlockProducer),
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),
),
ApplyIf(isFullOrLiteNode, ChainNode),
ApplyIf(isType(repo.StorageMiner), MinerNode),
)
}
@@ -519,6 +563,10 @@ func ConfigStorageMiner(c interface{}) Option {
Override(new(dtypes.RetrievalDealFilter), modules.RetrievalDealFilter(dealfilter.CliRetrievalDealFilter(cfg.Dealmaking.RetrievalFilter))),
),
Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(&cfg.Fees, storageadapter.PublishMsgConfig{
Period: time.Duration(cfg.Dealmaking.PublishMsgPeriod),
MaxDealsPerMsg: cfg.Dealmaking.MaxDealsPerPublishMsg,
})),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(&cfg.Fees)),
Override(new(sectorstorage.SealerConfig), cfg.Storage),
@@ -646,5 +694,6 @@ func Test() Option {
Unset(RunPeerMgrKey),
Unset(new(*peermgr.PeerMgr)),
Override(new(beacon.Schedule), testing.RandomBeacon),
Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{})),
)
}
node/config/def.go
+11 -1
View File
@@ -48,6 +48,12 @@ type DealmakingConfig struct {
ConsiderUnverifiedStorageDeals bool
PieceCidBlocklist []cid.Cid
ExpectedSealDuration Duration
// The amount of time to wait for more deals to arrive before
// publishing
PublishMsgPeriod Duration
// The maximum number of deals to include in a single PublishStorageDeals
// message
MaxDealsPerPublishMsg uint64
Filter string
RetrievalFilter string
@@ -64,6 +70,8 @@ type SealingConfig struct {
MaxSealingSectorsForDeals uint64
WaitDealsDelay Duration
AlwaysKeepUnsealedCopy bool
}
type MinerFeeConfig struct {
@@ -206,7 +214,9 @@ func DefaultStorageMiner() *StorageMiner {
ConsiderUnverifiedStorageDeals: true,
PieceCidBlocklist: []cid.Cid{},
// TODO: It'd be nice to set this based on sector size
ExpectedSealDuration: Duration(time.Hour * 24),
ExpectedSealDuration: Duration(time.Hour * 24),
PublishMsgPeriod: Duration(time.Hour),
MaxDealsPerPublishMsg: 8,
},
Fees: MinerFeeConfig{
node/hello/cbor_gen.go
+3
View File
@@ -5,6 +5,7 @@ package hello
import (
"fmt"
"io"
"sort"
abi "github.com/filecoin-project/go-state-types/abi"
cid "github.com/ipfs/go-cid"
@@ -13,6 +14,8 @@ import (
)
var _ = xerrors.Errorf
var _ = cid.Undef
var _ = sort.Sort
var lengthBufHelloMessage = []byte{132}
node/impl/client/client.go
+10
View File
@@ -711,6 +711,16 @@ func (a *API) ClientCalcCommP(ctx context.Context, inpath string) (*api.CommPRet
return nil, err
}
// check that the data is a car file; if it's not, retrieval won't work
_, _, err = car.ReadHeader(bufio.NewReader(rdr))
if err != nil {
return nil, xerrors.Errorf("not a car file: %w", err)
}
if _, err := rdr.Seek(0, io.SeekStart); err != nil {
return nil, xerrors.Errorf("seek to start: %w", err)
}
pieceReader, pieceSize := padreader.New(rdr, uint64(stat.Size()))
commP, err := ffiwrapper.GeneratePieceCIDFromFile(arbitraryProofType, pieceReader, pieceSize)
node/impl/full/state.go
+56 -48
View File
@@ -6,7 +6,6 @@ import (
"strconv"
cid "github.com/ipfs/go-cid"
cbor "github.com/ipfs/go-ipld-cbor"
"go.uber.org/fx"
"golang.org/x/xerrors"
@@ -42,6 +41,7 @@ import (
type StateModuleAPI interface {
MsigGetAvailableBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (types.BigInt, error)
MsigGetVested(ctx context.Context, addr address.Address, start types.TipSetKey, end types.TipSetKey) (types.BigInt, error)
MsigGetPending(ctx context.Context, addr address.Address, tsk types.TipSetKey) ([]*api.MsigTransaction, error)
StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
StateDealProviderCollateralBounds(ctx context.Context, size abi.PaddedPieceSize, verified bool, tsk types.TipSetKey) (api.DealCollateralBounds, error)
StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
@@ -54,8 +54,8 @@ type StateModuleAPI interface {
StateMinerProvingDeadline(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*dline.Info, error)
StateMinerPower(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error)
StateNetworkVersion(ctx context.Context, key types.TipSetKey) (network.Version, error)
StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error)
StateSearchMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error)
StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error)
StateVerifiedClientStatus(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.StoragePower, error)
StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error)
}
@@ -93,19 +93,20 @@ func (a *StateAPI) StateNetworkName(ctx context.Context) (dtypes.NetworkName, er
}
func (a *StateAPI) StateMinerSectors(ctx context.Context, addr address.Address, sectorNos *bitfield.BitField, tsk types.TipSetKey) ([]*miner.SectorOnChainInfo, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
act, err := a.StateManager.LoadActorTsk(ctx, addr, tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
return nil, xerrors.Errorf("failed to load miner actor: %w", err)
}
return stmgr.GetMinerSectorSet(ctx, a.StateManager, ts, addr, sectorNos)
mas, err := miner.Load(a.StateManager.ChainStore().Store(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load miner actor state: %w", err)
}
return mas.LoadSectors(sectorNos)
}
func (a *StateAPI) StateMinerActiveSectors(ctx context.Context, maddr address.Address, tsk types.TipSetKey) ([]*miner.SectorOnChainInfo, error) { // TODO: only used in cli
ts, err := a.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
act, err := a.StateManager.LoadActorTsk(ctx, maddr, tsk)
if err != nil {
return nil, xerrors.Errorf("failed to load miner actor: %w", err)
@@ -121,7 +122,7 @@ func (a *StateAPI) StateMinerActiveSectors(ctx context.Context, maddr address.Ad
return nil, xerrors.Errorf("merge partition active sets: %w", err)
}
return stmgr.GetMinerSectorSet(ctx, a.StateManager, ts, maddr, &activeSectors)
return mas.LoadSectors(&activeSectors)
}
func (m *StateModule) StateMinerInfo(ctx context.Context, actor address.Address, tsk types.TipSetKey) (miner.MinerInfo, error) {
@@ -425,38 +426,12 @@ func (a *StateAPI) StateReplay(ctx context.Context, tsk types.TipSetKey, mc cid.
}, nil
}
func stateForTs(ctx context.Context, ts *types.TipSet, cstore *store.ChainStore, smgr *stmgr.StateManager) (*state.StateTree, error) {
if ts == nil {
ts = cstore.GetHeaviestTipSet()
}
st, _, err := smgr.TipSetState(ctx, ts)
if err != nil {
return nil, err
}
buf := blockstore.NewBuffered(cstore.Blockstore())
cst := cbor.NewCborStore(buf)
return state.LoadStateTree(cst, st)
}
func (a *StateAPI) stateForTs(ctx context.Context, ts *types.TipSet) (*state.StateTree, error) {
return stateForTs(ctx, ts, a.Chain, a.StateManager)
}
func (m *StateModule) stateForTs(ctx context.Context, ts *types.TipSet) (*state.StateTree, error) {
return stateForTs(ctx, ts, m.Chain, m.StateManager)
}
func (m *StateModule) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) {
ts, err := m.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
state, err := m.stateForTs(ctx, ts)
if err != nil {
return nil, xerrors.Errorf("computing tipset state failed: %w", err)
}
return state.GetActor(actor)
return m.StateManager.LoadActor(ctx, actor, ts)
}
func (m *StateModule) StateLookupID(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) {
@@ -482,17 +457,12 @@ func (a *StateAPI) StateReadState(ctx context.Context, actor address.Address, ts
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
state, err := a.stateForTs(ctx, ts)
if err != nil {
return nil, xerrors.Errorf("getting state for tipset: %w", err)
}
act, err := state.GetActor(actor)
act, err := a.StateManager.LoadActor(ctx, actor, ts)
if err != nil {
return nil, xerrors.Errorf("getting actor: %w", err)
}
blk, err := state.Store.(*cbor.BasicIpldStore).Blocks.Get(act.Head)
blk, err := a.Chain.Blockstore().Get(act.Head)
if err != nil {
return nil, xerrors.Errorf("getting actor head: %w", err)
}
@@ -504,6 +474,7 @@ func (a *StateAPI) StateReadState(ctx context.Context, actor address.Address, ts
return &api.ActorState{
Balance: act.Balance,
Code: act.Code,
State: oif,
}, nil
}
@@ -528,6 +499,7 @@ func (a *StateAPI) StateDecodeParams(ctx context.Context, toAddr address.Address
// This is on StateAPI because miner.Miner requires this, and MinerAPI requires miner.Miner
func (a *StateAPI) MinerGetBaseInfo(ctx context.Context, maddr address.Address, epoch abi.ChainEpoch, tsk types.TipSetKey) (*api.MiningBaseInfo, error) {
// XXX: Gets the state by computing the tipset state, instead of looking at the parent.
return stmgr.MinerGetBaseInfo(ctx, a.StateManager, a.Beacon, tsk, epoch, maddr, a.ProofVerifier)
}
@@ -1008,6 +980,40 @@ func (m *StateModule) MsigGetVested(ctx context.Context, addr address.Address, s
return types.BigSub(startLk, endLk), nil
}
func (m *StateModule) MsigGetPending(ctx context.Context, addr address.Address, tsk types.TipSetKey) ([]*api.MsigTransaction, error) {
ts, err := m.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
act, err := m.StateManager.LoadActor(ctx, addr, ts)
if err != nil {
return nil, xerrors.Errorf("failed to load multisig actor: %w", err)
}
msas, err := multisig.Load(m.Chain.Store(ctx), act)
if err != nil {
return nil, xerrors.Errorf("failed to load multisig actor state: %w", err)
}
var out = []*api.MsigTransaction{}
if err := msas.ForEachPendingTxn(func(id int64, txn multisig.Transaction) error {
out = append(out, &api.MsigTransaction{
ID: id,
To: txn.To,
Value: txn.Value,
Method: txn.Method,
Params: txn.Params,
Approved: txn.Approved,
})
return nil
}); err != nil {
return nil, err
}
return out, nil
}
var initialPledgeNum = types.NewInt(110)
var initialPledgeDen = types.NewInt(100)
@@ -1337,11 +1343,11 @@ func (a *StateAPI) StateCirculatingSupply(ctx context.Context, tsk types.TipSetK
return types.EmptyInt, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
sTree, err := a.stateForTs(ctx, ts)
sTree, err := a.StateManager.ParentState(ts)
if err != nil {
return types.EmptyInt, err
}
return a.StateManager.GetCirculatingSupply(ctx, ts.Height(), sTree)
return a.StateManager.GetCirculatingSupply(ctx, ts.Height()-1, sTree)
}
func (a *StateAPI) StateVMCirculatingSupplyInternal(ctx context.Context, tsk types.TipSetKey) (api.CirculatingSupply, error) {
@@ -1358,7 +1364,7 @@ func stateVMCirculatingSupplyInternal(
return api.CirculatingSupply{}, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
sTree, err := stateForTs(ctx, ts, cstore, smgr)
sTree, err := smgr.ParentState(ts)
if err != nil {
return api.CirculatingSupply{}, err
}
@@ -1372,5 +1378,7 @@ func (m *StateModule) StateNetworkVersion(ctx context.Context, tsk types.TipSetK
return network.VersionMax, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
// TODO: Height-1 to be consistent with the rest of the APIs?
// But that's likely going to break a bunch of stuff.
return m.StateManager.GetNtwkVersion(ctx, ts.Height()), nil
}
node/impl/storminer.go
+14 -3
View File
@@ -32,6 +32,7 @@ import (
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/apistruct"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/markets/storageadapter"
"github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/node/impl/common"
"github.com/filecoin-project/lotus/node/modules/dtypes"
@@ -55,9 +56,10 @@ type StorageMinerAPI struct {
IStorageMgr sectorstorage.SectorManager
*stores.Index
storiface.WorkerReturn
DataTransfer dtypes.ProviderDataTransfer
Host host.Host
AddrSel *storage.AddressSelector
DataTransfer dtypes.ProviderDataTransfer
Host host.Host
AddrSel *storage.AddressSelector
DealPublisher *storageadapter.DealPublisher
DS dtypes.MetadataDS
@@ -501,6 +503,15 @@ func (sm *StorageMinerAPI) MarketDataTransferUpdates(ctx context.Context) (<-cha
return channels, nil
}
func (sm *StorageMinerAPI) MarketPendingDeals(ctx context.Context) (api.PendingDealInfo, error) {
return sm.DealPublisher.PendingDeals(), nil
}
func (sm *StorageMinerAPI) MarketPublishPendingDeals(ctx context.Context) error {
sm.DealPublisher.ForcePublishPendingDeals()
return nil
}
func (sm *StorageMinerAPI) DealsList(ctx context.Context) ([]api.MarketDeal, error) {
return sm.listDeals(ctx)
}
node/modules/chain.go
+1 -1
View File
@@ -75,7 +75,7 @@ func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds
}
func ChainRawBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.ChainRawBlockstore, error) {
bs, err := r.Blockstore(repo.BlockstoreChain)
bs, err := r.Blockstore(helpers.LifecycleCtx(mctx, lc), repo.BlockstoreChain)
if err != nil {
return nil, err
}
node/modules/client.go
+7 -5
View File
@@ -39,6 +39,7 @@ import (
"github.com/filecoin-project/lotus/node/impl/full"
payapi "github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/node/repo/importmgr"
"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
@@ -57,18 +58,18 @@ func HandleMigrateClientFunds(lc fx.Lifecycle, ds dtypes.MetadataDS, wallet full
if xerrors.Is(err, datastore.ErrNotFound) {
return nil
}
log.Errorf("client funds migration - getting datastore value: %w", err)
log.Errorf("client funds migration - getting datastore value: %v", err)
return nil
}
var value abi.TokenAmount
if err = value.UnmarshalCBOR(bytes.NewReader(b)); err != nil {
log.Errorf("client funds migration - unmarshalling datastore value: %w", err)
log.Errorf("client funds migration - unmarshalling datastore value: %v", err)
return nil
}
_, err = fundMgr.Reserve(ctx, addr, addr, value)
if err != nil {
log.Errorf("client funds migration - reserving funds (wallet %s, addr %s, funds %d): %w",
log.Errorf("client funds migration - reserving funds (wallet %s, addr %s, funds %d): %v",
addr, addr, value, err)
return nil
}
@@ -78,8 +79,9 @@ func HandleMigrateClientFunds(lc fx.Lifecycle, ds dtypes.MetadataDS, wallet full
})
}
func ClientMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.ClientMultiDstore, error) {
ds, err := r.Datastore("/client")
func ClientMultiDatastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.ClientMultiDstore, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
ds, err := r.Datastore(ctx, "/client")
if err != nil {
return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
}
node/modules/mpoolnonceapi.go
+32 -3
View File
@@ -4,6 +4,7 @@ import (
"context"
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/node/impl/full"
@@ -21,13 +22,41 @@ type MpoolNonceAPI struct {
StateAPI full.StateAPI
}
// GetNonce gets the nonce from actor state
// GetNonce gets the nonce from current chain head.
func (a *MpoolNonceAPI) GetNonce(addr address.Address) (uint64, error) {
act, err := a.StateAPI.StateGetActor(context.Background(), addr, types.EmptyTSK)
ts := a.StateAPI.Chain.GetHeaviestTipSet()
// make sure we have a key address so we can compare with messages
keyAddr, err := a.StateAPI.StateManager.ResolveToKeyAddress(context.TODO(), addr, ts)
if err != nil {
return 0, err
}
return act.Nonce, nil
// Load the last nonce from the state, if it exists.
highestNonce := uint64(0)
if baseActor, err := a.StateAPI.StateManager.LoadActorRaw(context.TODO(), addr, ts.ParentState()); err != nil {
if !xerrors.Is(err, types.ErrActorNotFound) {
return 0, err
}
} else {
highestNonce = baseActor.Nonce
}
// Otherwise, find the highest nonce in the tipset.
msgs, err := a.StateAPI.Chain.MessagesForTipset(ts)
if err != nil {
return 0, err
}
for _, msg := range msgs {
vmmsg := msg.VMMessage()
if vmmsg.From != keyAddr {
continue
}
if vmmsg.Nonce >= highestNonce {
highestNonce = vmmsg.Nonce + 1
}
}
return highestNonce, nil
}
var _ messagesigner.MpoolNonceAPI = (*MpoolNonceAPI)(nil)
node/modules/storage.go
+4 -2
View File
@@ -8,6 +8,7 @@ import (
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/backupds"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
)
@@ -27,8 +28,9 @@ func KeyStore(lr repo.LockedRepo) (types.KeyStore, error) {
return lr.KeyStore()
}
func Datastore(r repo.LockedRepo) (dtypes.MetadataDS, error) {
mds, err := r.Datastore("/metadata")
func Datastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.MetadataDS, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
mds, err := r.Datastore(ctx, "/metadata")
if err != nil {
return nil, err
}
node/modules/storageminer.go
+12 -8
View File
@@ -218,7 +218,7 @@ func StorageMiner(fc config.MinerFeeConfig) func(params StorageMinerParams) (*st
ctx := helpers.LifecycleCtx(mctx, lc)
fps, err := storage.NewWindowedPoStScheduler(api, fc, as, sealer, sealer, j, maddr)
fps, err := storage.NewWindowedPoStScheduler(api, fc, as, sealer, verif, sealer, j, maddr)
if err != nil {
return nil, err
}
@@ -293,19 +293,19 @@ func HandleMigrateProviderFunds(lc fx.Lifecycle, ds dtypes.MetadataDS, node api.
}
ts, err := node.ChainHead(ctx)
if err != nil {
log.Errorf("provider funds migration - getting chain head: %w", err)
log.Errorf("provider funds migration - getting chain head: %v", err)
return nil
}
mi, err := node.StateMinerInfo(ctx, address.Address(minerAddress), ts.Key())
if err != nil {
log.Errorf("provider funds migration - getting miner info %s: %w", minerAddress, err)
log.Errorf("provider funds migration - getting miner info %s: %v", minerAddress, err)
return nil
}
_, err = node.MarketReserveFunds(ctx, mi.Worker, address.Address(minerAddress), value)
if err != nil {
log.Errorf("provider funds migration - reserving funds (wallet %s, addr %s, funds %d): %w",
log.Errorf("provider funds migration - reserving funds (wallet %s, addr %s, funds %d): %v",
mi.Worker, minerAddress, value, err)
return nil
}
@@ -362,8 +362,9 @@ func NewProviderPieceStore(lc fx.Lifecycle, ds dtypes.MetadataDS) (dtypes.Provid
return ps, nil
}
func StagingMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.StagingMultiDstore, error) {
ds, err := r.Datastore("/staging")
func StagingMultiDatastore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.StagingMultiDstore, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
ds, err := r.Datastore(ctx, "/staging")
if err != nil {
return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
}
@@ -384,8 +385,9 @@ func StagingMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.StagingMu
// StagingBlockstore creates a blockstore for staging blocks for a miner
// in a storage deal, prior to sealing
func StagingBlockstore(r repo.LockedRepo) (dtypes.StagingBlockstore, error) {
stagingds, err := r.Datastore("/staging")
func StagingBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo) (dtypes.StagingBlockstore, error) {
ctx := helpers.LifecycleCtx(mctx, lc)
stagingds, err := r.Datastore(ctx, "/staging")
if err != nil {
return nil, err
}
@@ -810,6 +812,7 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
MaxSealingSectors: cfg.MaxSealingSectors,
MaxSealingSectorsForDeals: cfg.MaxSealingSectorsForDeals,
WaitDealsDelay: config.Duration(cfg.WaitDealsDelay),
AlwaysKeepUnsealedCopy: cfg.AlwaysKeepUnsealedCopy,
}
})
return
@@ -824,6 +827,7 @@ func NewGetSealConfigFunc(r repo.LockedRepo) (dtypes.GetSealingConfigFunc, error
MaxSealingSectors: cfg.Sealing.MaxSealingSectors,
MaxSealingSectorsForDeals: cfg.Sealing.MaxSealingSectorsForDeals,
WaitDealsDelay: time.Duration(cfg.Sealing.WaitDealsDelay),
AlwaysKeepUnsealedCopy: cfg.Sealing.AlwaysKeepUnsealedCopy,
}
})
return
node/node_test.go
+10 -5
View File
@@ -5,14 +5,12 @@ import (
"testing"
"time"
builder "github.com/filecoin-project/lotus/node/test"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/lib/lotuslog"
logging "github.com/ipfs/go-log/v2"
"github.com/filecoin-project/lotus/api/test"
"github.com/filecoin-project/lotus/chain/actors/policy"
"github.com/filecoin-project/lotus/lib/lotuslog"
builder "github.com/filecoin-project/lotus/node/test"
logging "github.com/ipfs/go-log/v2"
)
func init() {
@@ -57,6 +55,9 @@ func TestAPIDealFlow(t *testing.T) {
t.Run("TestFastRetrievalDealFlow", func(t *testing.T) {
test.TestFastRetrievalDealFlow(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
})
t.Run("TestPublishDealsBatching", func(t *testing.T) {
test.TestPublishDealsBatching(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
})
}
func TestAPIDealFlowReal(t *testing.T) {
@@ -91,6 +92,10 @@ func TestAPIDealFlowReal(t *testing.T) {
}
func TestDealMining(t *testing.T) {
if testing.Short() {
t.Skip("skipping test in short mode")
}
logging.SetLogLevel("miner", "ERROR")
logging.SetLogLevel("chainstore", "ERROR")
logging.SetLogLevel("chain", "ERROR")
node/repo/fsrepo.go
+2 -1
View File
@@ -2,6 +2,7 @@ package repo
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
@@ -299,7 +300,7 @@ func (fsr *fsLockedRepo) Close() error {
}
// Blockstore returns a blockstore for the provided data domain.
func (fsr *fsLockedRepo) Blockstore(domain BlockstoreDomain) (blockstore.Blockstore, error) {
func (fsr *fsLockedRepo) Blockstore(ctx context.Context, domain BlockstoreDomain) (blockstore.Blockstore, error) {
if domain != BlockstoreChain {
return nil, ErrInvalidBlockstoreDomain
}
node/repo/fsrepo_ds.go
+2 -1
View File
@@ -1,6 +1,7 @@
package repo
import (
"context"
"os"
"path/filepath"
@@ -67,7 +68,7 @@ func (fsr *fsLockedRepo) openDatastores(readonly bool) (map[string]datastore.Bat
return out, nil
}
func (fsr *fsLockedRepo) Datastore(ns string) (datastore.Batching, error) {
func (fsr *fsLockedRepo) Datastore(_ context.Context, ns string) (datastore.Batching, error) {
fsr.dsOnce.Do(func() {
fsr.ds, fsr.dsErr = fsr.openDatastores(fsr.readonly)
})
node/repo/interface.go
+9 -2
View File
@@ -1,6 +1,7 @@
package repo
import (
"context"
"errors"
"github.com/filecoin-project/lotus/blockstore"
@@ -51,10 +52,16 @@ type LockedRepo interface {
Close() error
// Returns datastore defined in this repo.
Datastore(namespace string) (datastore.Batching, error)
// The supplied context must only be used to initialize the datastore.
// The implementation should not retain the context for usage throughout
// the lifecycle.
Datastore(ctx context.Context, namespace string) (datastore.Batching, error)
// Blockstore returns an IPLD blockstore for the requested domain.
Blockstore(domain BlockstoreDomain) (blockstore.Blockstore, error)
// The supplied context must only be used to initialize the blockstore.
// The implementation should not retain the context for usage throughout
// the lifecycle.
Blockstore(ctx context.Context, domain BlockstoreDomain) (blockstore.Blockstore, error)
// Returns config in this repo
Config() (interface{}, error)
node/repo/memrepo.go
+3 -2
View File
@@ -1,6 +1,7 @@
package repo
import (
"context"
"encoding/json"
"io/ioutil"
"os"
@@ -236,7 +237,7 @@ func (lmem *lockedMemRepo) Close() error {
}
func (lmem *lockedMemRepo) Datastore(ns string) (datastore.Batching, error) {
func (lmem *lockedMemRepo) Datastore(_ context.Context, ns string) (datastore.Batching, error) {
if err := lmem.checkToken(); err != nil {
return nil, err
}
@@ -244,7 +245,7 @@ func (lmem *lockedMemRepo) Datastore(ns string) (datastore.Batching, error) {
return namespace.Wrap(lmem.mem.datastore, datastore.NewKey(ns)), nil
}
func (lmem *lockedMemRepo) Blockstore(domain BlockstoreDomain) (blockstore.Blockstore, error) {
func (lmem *lockedMemRepo) Blockstore(ctx context.Context, domain BlockstoreDomain) (blockstore.Blockstore, error) {
if domain != BlockstoreChain {
return nil, ErrInvalidBlockstoreDomain
}
node/test/builder.go
+11 -2
View File
@@ -77,7 +77,7 @@ func CreateTestStorageNode(ctx context.Context, t *testing.T, waddr address.Addr
})
require.NoError(t, err)
ds, err := lr.Datastore("/metadata")
ds, err := lr.Datastore(context.TODO(), "/metadata")
require.NoError(t, err)
err = ds.Put(datastore.NewKey("miner-address"), act.Bytes())
require.NoError(t, err)
@@ -288,7 +288,11 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
genMiner := maddrs[i]
wa := genms[i].Worker
storers[i] = CreateTestStorageNode(ctx, t, wa, genMiner, pk, f, mn, node.Options())
opts := def.Opts
if opts == nil {
opts = node.Options()
}
storers[i] = CreateTestStorageNode(ctx, t, wa, genMiner, pk, f, mn, opts)
if err := storers[i].StorageAddLocal(ctx, presealDirs[i]); err != nil {
t.Fatalf("%+v", err)
}
@@ -455,12 +459,17 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
}
}
opts := def.Opts
if opts == nil {
opts = node.Options()
}
storers[i] = CreateTestStorageNode(ctx, t, genms[i].Worker, maddrs[i], pidKeys[i], f, mn, node.Options(
node.Override(new(sectorstorage.SectorManager), func() (sectorstorage.SectorManager, error) {
return mock.NewMockSectorMgr(sectors), nil
}),
node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
node.Unset(new(*sectorstorage.Manager)),
opts,
))
if rpc {