lotus/node/builder.go

415 lines
13 KiB
Go
Raw Normal View History

package node
import (
"context"
2019-07-04 15:50:48 +00:00
"errors"
"time"
logging "github.com/ipfs/go-log/v2"
2022-06-14 15:00:51 +00:00
metricsi "github.com/ipfs/go-metrics-interface"
dht "github.com/libp2p/go-libp2p-kad-dht"
2019-07-08 14:07:09 +00:00
pubsub "github.com/libp2p/go-libp2p-pubsub"
2019-07-05 10:06:28 +00:00
record "github.com/libp2p/go-libp2p-record"
2022-08-25 18:20:41 +00:00
ci "github.com/libp2p/go-libp2p/core/crypto"
"github.com/libp2p/go-libp2p/core/host"
"github.com/libp2p/go-libp2p/core/network"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/libp2p/go-libp2p/core/peerstore"
"github.com/libp2p/go-libp2p/core/routing"
2020-11-13 09:44:29 +00:00
"github.com/libp2p/go-libp2p/p2p/net/conngater"
"github.com/multiformats/go-multiaddr"
"go.uber.org/fx"
"golang.org/x/xerrors"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/lotus/api"
2021-08-26 13:59:54 +00:00
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/beacon"
"github.com/filecoin-project/lotus/chain/types"
2020-07-17 13:22:37 +00:00
"github.com/filecoin-project/lotus/journal"
2021-08-17 12:39:36 +00:00
"github.com/filecoin-project/lotus/journal/alerting"
"github.com/filecoin-project/lotus/lib/lotuslog"
2020-02-22 11:36:22 +00:00
"github.com/filecoin-project/lotus/lib/peermgr"
_ "github.com/filecoin-project/lotus/lib/sigs/bls"
_ "github.com/filecoin-project/lotus/lib/sigs/delegated"
_ "github.com/filecoin-project/lotus/lib/sigs/secp"
"github.com/filecoin-project/lotus/markets/storageadapter"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/impl/common"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/lotus/node/impl/net"
"github.com/filecoin-project/lotus/node/modules"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/modules/lp2p"
"github.com/filecoin-project/lotus/node/modules/testing"
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/storage/paths"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/lotus/system"
)
//nolint:deadcode,varcheck
2020-01-29 18:10:41 +00:00
var log = logging.Logger("builder")
2019-07-04 15:50:48 +00:00
// special is a type used to give keys to modules which
2022-08-29 14:25:30 +00:00
//
// can't really be identified by the returned type
2019-07-04 15:50:48 +00:00
type special struct{ id int }
//nolint:golint
2019-07-04 15:50:48 +00:00
var (
2020-04-28 13:21:56 +00:00
DefaultTransportsKey = special{0} // Libp2p option
DiscoveryHandlerKey = special{2} // Private type
AddrsFactoryKey = special{3} // Libp2p option
SmuxTransportKey = special{4} // Libp2p option
RelayKey = special{5} // Libp2p option
SecurityKey = special{6} // Libp2p option
BaseRoutingKey = special{7} // fx groups + multiret
NatPortMapKey = special{8} // Libp2p option
ConnectionManagerKey = special{9} // Libp2p option
2020-04-27 17:20:39 +00:00
AutoNATSvcKey = special{10} // Libp2p option
BandwidthReporterKey = special{11} // Libp2p option
2020-11-13 09:44:29 +00:00
ConnGaterKey = special{12} // libp2p option
integrate DAG store and CARv2 in deal-making (#6671) This commit removes badger from the deal-making processes, and moves to a new architecture with the dagstore as the cental component on the miner-side, and CARv2s on the client-side. Every deal that has been handed off to the sealing subsystem becomes a shard in the dagstore. Shards are mounted via the LotusMount, which teaches the dagstore how to load the related piece when serving retrievals. When the miner starts the Lotus for the first time with this patch, we will perform a one-time migration of all active deals into the dagstore. This is a lightweight process, and it consists simply of registering the shards in the dagstore. Shards are backed by the unsealed copy of the piece. This is currently a CARv1. However, the dagstore keeps CARv2 indices for all pieces, so when it's time to acquire a shard to serve a retrieval, the unsealed CARv1 is joined with its index (safeguarded by the dagstore), to form a read-only blockstore, thus taking the place of the monolithic badger. Data transfers have been adjusted to interface directly with CARv2 files. On inbound transfers (client retrievals, miner storage deals), we stream the received data into a CARv2 ReadWrite blockstore. On outbound transfers (client storage deals, miner retrievals), we serve the data off a CARv2 ReadOnly blockstore. Client-side imports are managed by the refactored *imports.Manager component (when not using IPFS integration). Just like it before, we use the go-filestore library to avoid duplicating the data from the original file in the resulting UnixFS DAG (concretely the leaves). However, the target of those imports are what we call "ref-CARv2s": CARv2 files placed under the `$LOTUS_PATH/imports` directory, containing the intermediate nodes in full, and the leaves as positional references to the original file on disk. Client-side retrievals are placed into CARv2 files in the location: `$LOTUS_PATH/retrievals`. A new set of `Dagstore*` JSON-RPC operations and `lotus-miner dagstore` subcommands have been introduced on the miner-side to inspect and manage the dagstore. Despite moving to a CARv2-backed system, the IPFS integration has been respected, and it continues to be possible to make storage deals with data held in an IPFS node, and to perform retrievals directly into an IPFS node. NOTE: because the "staging" and "client" Badger blockstores are no longer used, existing imports on the client will be rendered useless. On startup, Lotus will enumerate all imports and print WARN statements on the log for each import that needs to be reimported. These log lines contain these messages: - import lacks carv2 path; import will not work; please reimport - import has missing/broken carv2; please reimport At the end, we will print a "sanity check completed" message indicating the count of imports found, and how many were deemed broken. Co-authored-by: Aarsh Shah <aarshkshah1992@gmail.com> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com> Co-authored-by: Raúl Kripalani <raul@protocol.ai> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com>
2021-08-16 22:34:32 +00:00
DAGStoreKey = special{13} // constructor returns multiple values
2022-01-18 12:56:24 +00:00
ResourceManagerKey = special{14} // Libp2p option
2019-07-04 15:50:48 +00:00
)
type invoke int
// Invokes are called in the order they are defined.
2022-08-29 14:25:30 +00:00
//
2019-07-08 13:36:43 +00:00
//nolint:golint
2019-07-04 15:50:48 +00:00
const (
// InitJournal at position 0 initializes the journal global var as soon as
// the system starts, so that it's available for all other components.
InitJournalKey = invoke(iota)
// System processes.
InitMemoryWatchdog
2019-07-04 20:06:02 +00:00
2021-08-17 12:39:36 +00:00
// health checks
CheckFDLimit
// libp2p
PstoreAddSelfKeysKey
2019-07-04 15:50:48 +00:00
StartListeningKey
2019-10-11 00:31:06 +00:00
BootstrapKey
2019-07-04 15:50:48 +00:00
2019-07-08 13:36:43 +00:00
// filecoin
2019-07-09 22:58:51 +00:00
SetGenesisKey
2019-07-08 14:07:09 +00:00
2019-07-08 13:36:43 +00:00
RunHelloKey
RunChainExchangeKey
RunChainGraphsync
2019-10-17 08:57:56 +00:00
RunPeerMgrKey
2019-07-08 14:07:09 +00:00
HandleIncomingBlocksKey
HandleIncomingMessagesKey
HandleMigrateClientFundsKey
2020-07-28 23:16:47 +00:00
HandlePaymentChannelManagerKey
2019-08-14 20:27:10 +00:00
RelayIndexerMessagesKey
// miner
PreflightChecksKey
2019-12-04 19:44:15 +00:00
GetParamsKey
HandleMigrateProviderFundsKey
2019-08-02 16:25:10 +00:00
HandleDealsKey
2019-08-26 13:45:36 +00:00
HandleRetrievalKey
2019-08-14 20:27:10 +00:00
RunSectorServiceKey
2019-08-01 17:12:41 +00:00
// daemon
2019-07-23 22:34:13 +00:00
ExtractApiKey
2019-10-10 11:07:00 +00:00
HeadMetricsKey
SettlePaymentChannelsKey
2019-12-17 16:28:02 +00:00
RunPeerTaggerKey
SetupFallbackBlockstoresKey
2022-09-22 20:27:15 +00:00
GoRPCServer
2019-07-23 22:34:13 +00:00
2019-07-10 17:28:49 +00:00
SetApiEndpointKey
2019-07-04 15:50:48 +00:00
_nInvokes // keep this last
)
2019-07-10 13:06:04 +00:00
type Settings struct {
2019-07-04 20:06:02 +00:00
// modules is a map of constructors for DI
//
// In most cases the index will be a reflect. Type of element returned by
// the constructor, but for some 'constructors' it's hard to specify what's
// the return type should be (or the constructor returns fx group)
2019-07-04 15:50:48 +00:00
modules map[interface{}]fx.Option
// invokes are separate from modules as they can't be referenced by return
// type, and must be applied in correct order
invokes []fx.Option
nodeType repo.RepoType
2019-07-19 09:23:24 +00:00
2021-06-24 14:02:51 +00:00
Base bool // Base option applied
2019-07-10 13:06:04 +00:00
Config bool // Config option applied
Lite bool // Start node in "lite" mode
enableLibp2pNode bool
2019-07-04 15:50:48 +00:00
}
// Basic lotus-app services
2019-07-09 17:03:36 +00:00
func defaults() []Option {
return []Option{
2020-08-26 15:09:37 +00:00
// global system journal.
Override(new(journal.DisabledEvents), journal.EnvDisabledEvents),
2020-08-26 15:09:37 +00:00
Override(new(journal.Journal), modules.OpenFilesystemJournal),
2021-08-17 12:39:36 +00:00
Override(new(*alerting.Alerting), alerting.NewAlertingSystem),
Override(new(dtypes.NodeStartTime), FromVal(dtypes.NodeStartTime(time.Now()))),
2021-08-17 12:39:36 +00:00
2021-08-26 13:59:54 +00:00
Override(CheckFDLimit, modules.CheckFdLimit(build.DefaultFDLimit)),
2020-08-26 15:09:37 +00:00
2020-12-02 22:26:30 +00:00
Override(new(system.MemoryConstraints), modules.MemoryConstraints),
Override(InitMemoryWatchdog, modules.MemoryWatchdog),
Override(new(helpers.MetricsCtx), func() context.Context {
return metricsi.CtxScope(context.Background(), "lotus")
}),
Override(new(dtypes.ShutdownChan), make(chan struct{})),
// the great context in the sky, otherwise we can't DI build genesis; there has to be a better
// solution than this hack.
Override(new(context.Context), func(lc fx.Lifecycle, mctx helpers.MetricsCtx) context.Context {
return helpers.LifecycleCtx(mctx, lc)
}),
2019-07-09 17:03:36 +00:00
}
2019-07-04 15:50:48 +00:00
}
var LibP2P = Options(
// Host config
Override(new(dtypes.Bootstrapper), dtypes.Bootstrapper(false)),
// Host dependencies
Override(new(peerstore.Peerstore), lp2p.Peerstore),
Override(PstoreAddSelfKeysKey, lp2p.PstoreAddSelfKeys),
Override(StartListeningKey, lp2p.StartListening(config.DefaultFullNode().Libp2p.ListenAddresses)),
// Host settings
Override(DefaultTransportsKey, lp2p.DefaultTransports),
Override(AddrsFactoryKey, lp2p.AddrsFactory(nil, nil)),
Override(SmuxTransportKey, lp2p.SmuxTransport()),
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(DiscoveryHandlerKey, lp2p.DiscoveryHandler),
// Routing
Override(new(record.Validator), modules.RecordValidator),
Override(BaseRoutingKey, lp2p.BaseRouting),
Override(new(routing.Routing), lp2p.Routing),
// Services
Override(BandwidthReporterKey, lp2p.BandwidthCounter),
Override(AutoNATSvcKey, lp2p.AutoNATService),
// 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),
}
}),
// Services (connection management)
Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
Override(new(*conngater.BasicConnectionGater), lp2p.ConnGater),
Override(ConnGaterKey, lp2p.ConnGaterOption),
2022-01-18 12:56:24 +00:00
// Services (resource management)
Override(new(network.ResourceManager), lp2p.ResourceManager(200)),
2022-01-18 12:56:24 +00:00
Override(ResourceManagerKey, lp2p.ResourceManagerOption),
)
2019-07-19 09:23:24 +00:00
2021-06-28 16:17:22 +00:00
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 }
2022-02-10 16:33:38 +00:00
func isFullNode(s *Settings) bool {
return s.nodeType == repo.FullNode && !s.Lite
2022-02-10 16:33:38 +00:00
}
func isLiteNode(s *Settings) bool {
return s.nodeType == repo.FullNode && s.Lite
2022-02-10 16:33:38 +00:00
}
func Base() Option {
2019-07-19 09:23:24 +00:00
return Options(
2021-06-24 14:02:51 +00:00
func(s *Settings) error { s.Base = true; return nil }, // mark Base as applied
2019-07-19 09:23:24 +00:00
ApplyIf(func(s *Settings) bool { return s.Config },
2021-06-24 14:02:51 +00:00
Error(errors.New("the Base() option must be set before Config option")),
2019-07-19 09:23:24 +00:00
),
ApplyIf(func(s *Settings) bool { return s.enableLibp2pNode },
2021-07-07 11:31:56 +00:00
LibP2P,
2019-07-19 09:23:24 +00:00
),
ApplyIf(isFullOrLiteNode, ChainNode),
ApplyIf(IsType(repo.StorageMiner), MinerNode),
2019-07-19 09:23:24 +00:00
)
}
2019-07-10 13:06:04 +00:00
// Config sets up constructors based on the provided Config
2021-06-24 14:02:51 +00:00
func ConfigCommon(cfg *config.Common, enableLibp2pNode bool) Option {
// setup logging early
lotuslog.SetLevelsFromConfig(cfg.Logging.SubsystemLevels)
2019-07-04 15:50:48 +00:00
return Options(
2019-07-10 13:06:04 +00:00
func(s *Settings) error { s.Config = true; return nil },
2020-03-16 17:50:07 +00:00
Override(new(dtypes.APIEndpoint), func() (dtypes.APIEndpoint, error) {
return multiaddr.NewMultiaddr(cfg.API.ListenAddress)
}),
Override(SetApiEndpointKey, func(lr repo.LockedRepo, e dtypes.APIEndpoint) error {
return lr.SetAPIEndpoint(e)
}),
Override(new(paths.URLs), func(e dtypes.APIEndpoint) (paths.URLs, error) {
ip := cfg.API.RemoteListenAddress
var urls paths.URLs
2020-04-03 05:01:39 +00:00
urls = append(urls, "http://"+ip+"/remote") // TODO: This makes no assumptions, and probably could...
2020-03-16 17:50:07 +00:00
return urls, nil
}),
2021-06-24 14:02:51 +00:00
ApplyIf(func(s *Settings) bool { return s.Base }), // apply only if Base has already been applied
If(!enableLibp2pNode,
Override(new(api.Net), new(api.NetStub)),
2021-06-24 14:02:51 +00:00
Override(new(api.Common), From(new(common.CommonAPI))),
),
If(enableLibp2pNode,
Override(new(api.Net), From(new(net.NetAPI))),
2021-06-24 14:02:51 +00:00
Override(new(api.Common), From(new(common.CommonAPI))),
2019-07-04 15:50:48 +00:00
Override(StartListeningKey, lp2p.StartListening(cfg.Libp2p.ListenAddresses)),
2019-12-17 16:09:43 +00:00
Override(ConnectionManagerKey, lp2p.ConnectionManager(
cfg.Libp2p.ConnMgrLow,
cfg.Libp2p.ConnMgrHigh,
time.Duration(cfg.Libp2p.ConnMgrGrace),
cfg.Libp2p.ProtectedPeers)),
Override(new(network.ResourceManager), lp2p.ResourceManager(cfg.Libp2p.ConnMgrHigh)),
2020-05-14 01:10:49 +00:00
Override(new(*pubsub.PubSub), lp2p.GossipSub),
Override(new(*config.Pubsub), &cfg.Pubsub),
2019-10-11 02:45:45 +00:00
ApplyIf(func(s *Settings) bool { return len(cfg.Libp2p.BootstrapPeers) > 0 },
Override(new(dtypes.BootstrapPeers), modules.ConfigBootstrap(cfg.Libp2p.BootstrapPeers)),
),
2021-06-24 14:02:51 +00:00
Override(AddrsFactoryKey, lp2p.AddrsFactory(
cfg.Libp2p.AnnounceAddresses,
cfg.Libp2p.NoAnnounceAddresses)),
2021-08-27 17:14:29 +00:00
If(!cfg.Libp2p.DisableNatPortMap, Override(NatPortMapKey, lp2p.NatPortMap)),
2019-12-11 15:08:50 +00:00
),
2021-03-09 21:33:01 +00:00
Override(new(dtypes.MetadataDS), modules.Datastore(cfg.Backup.DisableMetadataLog)),
)
}
2019-07-10 15:38:35 +00:00
func Repo(r repo.Repo) Option {
return func(settings *Settings) error {
lr, err := r.Lock(settings.nodeType)
if err != nil {
return err
}
c, err := lr.Config()
if err != nil {
return err
}
return Options(
Override(new(repo.LockedRepo), modules.LockedRepo(lr)), // module handles closing
2019-07-16 16:02:51 +00:00
Override(new(ci.PrivKey), lp2p.PrivKey),
Override(new(ci.PubKey), ci.PrivKey.GetPublic),
Override(new(peer.ID), peer.IDFromPublicKey),
2019-07-23 20:23:44 +00:00
Override(new(types.KeyStore), modules.KeyStore),
Override(new(*dtypes.APIAlg), modules.APISecret),
ApplyIf(IsType(repo.FullNode), ConfigFullNode(c)),
ApplyIf(IsType(repo.StorageMiner), ConfigStorageMiner(c)),
)(settings)
}
2019-07-10 15:38:35 +00:00
}
2019-09-17 14:23:08 +00:00
type StopFunc func(context.Context) error
2019-07-02 13:05:43 +00:00
// New builds and starts new Filecoin node
2019-09-17 14:23:08 +00:00
func New(ctx context.Context, opts ...Option) (StopFunc, error) {
2019-07-10 13:06:04 +00:00
settings := Settings{
2020-09-14 03:00:08 +00:00
modules: map[interface{}]fx.Option{},
invokes: make([]fx.Option, _nInvokes),
2019-07-04 15:50:48 +00:00
}
2019-07-04 20:06:02 +00:00
// apply module options in the right order
2019-07-09 17:03:36 +00:00
if err := Options(Options(defaults()...), Options(opts...))(&settings); err != nil {
2020-02-27 23:14:15 +00:00
return nil, xerrors.Errorf("applying node options failed: %w", err)
2019-07-04 15:50:48 +00:00
}
2019-07-04 20:06:02 +00:00
// gather constructors for fx.Options
2019-07-04 15:50:48 +00:00
ctors := make([]fx.Option, 0, len(settings.modules))
for _, opt := range settings.modules {
ctors = append(ctors, opt)
}
2019-07-04 20:06:02 +00:00
// fill holes in invokes for use in fx.Options
2019-07-04 15:50:48 +00:00
for i, opt := range settings.invokes {
if opt == nil {
settings.invokes[i] = fx.Options()
}
}
app := fx.New(
2019-07-04 15:50:48 +00:00
fx.Options(ctors...),
fx.Options(settings.invokes...),
2019-07-09 17:03:36 +00:00
fx.NopLogger,
)
2019-07-04 20:06:02 +00:00
// TODO: we probably should have a 'firewall' for Closing signal
// on this context, and implement closing logic through lifecycles
// correctly
if err := app.Start(ctx); err != nil {
// comment fx.NopLogger few lines above for easier debugging
2020-02-27 23:14:15 +00:00
return nil, xerrors.Errorf("starting node: %w", err)
}
2019-09-17 14:23:08 +00:00
return app.Stop, nil
}
// In-memory / testing
2019-10-23 11:02:00 +00:00
func Test() Option {
2019-07-04 15:50:48 +00:00
return Options(
2019-10-23 11:02:00 +00:00
Unset(RunPeerMgrKey),
Unset(new(*peermgr.PeerMgr)),
Override(new(beacon.Schedule), testing.RandomBeacon),
Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{})),
)
}
// For 3rd party dep injection.
func WithRepoType(repoType repo.RepoType) func(s *Settings) error {
return func(s *Settings) error {
s.nodeType = repoType
return nil
}
}
func WithEnableLibp2pNode(enable bool) func(s *Settings) error {
return func(s *Settings) error {
s.enableLibp2pNode = enable
return nil
}
}
func WithInvokesKey(i invoke, resApi interface{}) func(s *Settings) error {
return func(s *Settings) error {
s.invokes[i] = fx.Populate(resApi)
return nil
}
}