lotus/node/modules/services.go

291 lines
8.2 KiB
Go
Raw Normal View History

2019-07-08 14:07:09 +00:00
package modules
import (
2020-08-11 12:48:32 +00:00
"context"
"os"
"strconv"
"time"
2020-08-11 12:48:32 +00:00
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
eventbus "github.com/libp2p/go-eventbus"
event "github.com/libp2p/go-libp2p-core/event"
2019-07-08 14:07:09 +00:00
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/peer"
2019-07-08 14:07:09 +00:00
pubsub "github.com/libp2p/go-libp2p-pubsub"
"go.uber.org/fx"
"golang.org/x/xerrors"
2020-09-29 11:53:30 +00:00
"github.com/filecoin-project/go-fil-markets/discovery"
discoveryimpl "github.com/filecoin-project/go-fil-markets/discovery/impl"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/beacon"
2020-04-14 03:05:19 +00:00
"github.com/filecoin-project/lotus/chain/beacon/drand"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/exchange"
2019-12-01 23:11:43 +00:00
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/stmgr"
2020-04-14 03:05:19 +00:00
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/sub"
"github.com/filecoin-project/lotus/chain/types"
2020-08-11 12:48:32 +00:00
"github.com/filecoin-project/lotus/journal"
2021-08-17 12:55:35 +00:00
"github.com/filecoin-project/lotus/journal/fsjournal"
2020-02-22 11:36:22 +00:00
"github.com/filecoin-project/lotus/lib/peermgr"
2020-09-29 11:53:30 +00:00
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"github.com/filecoin-project/lotus/node/hello"
"github.com/filecoin-project/lotus/node/impl/full"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
2020-08-11 12:48:32 +00:00
"github.com/filecoin-project/lotus/node/repo"
2019-07-08 14:07:09 +00:00
)
var pubsubMsgsSyncEpochs = 10
func init() {
if s := os.Getenv("LOTUS_MSGS_SYNC_EPOCHS"); s != "" {
val, err := strconv.Atoi(s)
if err != nil {
log.Errorf("failed to parse LOTUS_MSGS_SYNC_EPOCHS: %s", err)
return
}
pubsubMsgsSyncEpochs = val
}
}
func RunHello(mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host, svc *hello.Service) error {
2019-07-08 14:07:09 +00:00
h.SetStreamHandler(hello.ProtocolID, svc.HandleStream)
sub, err := h.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted), eventbus.BufSize(1024))
if err != nil {
return xerrors.Errorf("failed to subscribe to event bus: %w", err)
}
ctx := helpers.LifecycleCtx(mctx, lc)
go func() {
for evt := range sub.Out() {
pic := evt.(event.EvtPeerIdentificationCompleted)
2019-07-08 14:07:09 +00:00
go func() {
if err := svc.SayHello(ctx, pic.Peer); err != nil {
protos, _ := h.Peerstore().GetProtocols(pic.Peer)
agent, _ := h.Peerstore().Get(pic.Peer, "AgentVersion")
if protosContains(protos, hello.ProtocolID) {
log.Warnw("failed to say hello", "error", err, "peer", pic.Peer, "supported", protos, "agent", agent)
} else {
log.Debugw("failed to say hello", "error", err, "peer", pic.Peer, "supported", protos, "agent", agent)
}
2019-07-08 14:07:09 +00:00
return
}
}()
}
}()
return nil
2019-07-08 14:07:09 +00:00
}
func protosContains(protos []string, search string) bool {
for _, p := range protos {
if p == search {
return true
}
}
return false
}
2019-10-17 08:57:56 +00:00
func RunPeerMgr(mctx helpers.MetricsCtx, lc fx.Lifecycle, pmgr *peermgr.PeerMgr) {
go pmgr.Run(helpers.LifecycleCtx(mctx, lc))
}
func RunChainExchange(h host.Host, svc exchange.Server) {
h.SetStreamHandler(exchange.ChainExchangeProtocolID, svc.HandleStream) // new
2019-07-08 14:07:09 +00:00
}
func waitForSync(stmgr *stmgr.StateManager, epochs int, subscribe func()) {
2020-09-08 07:55:14 +00:00
nearsync := time.Duration(epochs*int(build.BlockDelaySecs)) * time.Second
2019-07-08 14:07:09 +00:00
// early check, are we synced at start up?
ts := stmgr.ChainStore().GetHeaviestTipSet()
timestamp := ts.MinTimestamp()
2020-09-08 07:55:14 +00:00
timestampTime := time.Unix(int64(timestamp), 0)
if build.Clock.Since(timestampTime) < nearsync {
subscribe()
return
2019-07-08 14:07:09 +00:00
}
// we are not synced, subscribe to head changes and wait for sync
stmgr.ChainStore().SubscribeHeadChanges(func(rev, app []*types.TipSet) error {
if len(app) == 0 {
return nil
}
latest := app[0].MinTimestamp()
for _, ts := range app[1:] {
timestamp := ts.MinTimestamp()
if timestamp > latest {
latest = timestamp
}
}
2020-09-08 07:55:14 +00:00
latestTime := time.Unix(int64(latest), 0)
if build.Clock.Since(latestTime) < nearsync {
subscribe()
return store.ErrNotifeeDone
}
return nil
})
}
2021-09-02 16:07:23 +00:00
func HandleIncomingBlocks(mctx helpers.MetricsCtx,
lc fx.Lifecycle,
ps *pubsub.PubSub,
s *chain.Syncer,
bserv dtypes.ChainBlockService,
chain *store.ChainStore,
cns consensus.Consensus,
h host.Host,
nn dtypes.NetworkName) {
ctx := helpers.LifecycleCtx(mctx, lc)
v := sub.NewBlockValidator(
2021-09-02 16:07:23 +00:00
h.ID(), chain, cns,
func(p peer.ID) {
ps.BlacklistPeer(p)
h.ConnManager().TagPeer(p, "badblock", -1000)
})
2020-02-17 05:51:18 +00:00
2020-03-31 23:13:37 +00:00
if err := ps.RegisterTopicValidator(build.BlocksTopic(nn), v.Validate); err != nil {
2020-02-17 05:51:18 +00:00
panic(err)
}
log.Infof("subscribing to pubsub topic %s", build.BlocksTopic(nn))
2019-07-08 14:07:09 +00:00
blocksub, err := ps.Subscribe(build.BlocksTopic(nn)) //nolint
2019-07-08 14:07:09 +00:00
if err != nil {
panic(err)
}
go sub.HandleIncomingBlocks(ctx, blocksub, s, bserv, h.ConnManager())
}
func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, stmgr *stmgr.StateManager, mpool *messagepool.MessagePool, h host.Host, nn dtypes.NetworkName, bootstrapper dtypes.Bootstrapper) {
ctx := helpers.LifecycleCtx(mctx, lc)
v := sub.NewMessageValidator(h.ID(), mpool)
2020-02-17 05:51:18 +00:00
2020-03-31 23:13:37 +00:00
if err := ps.RegisterTopicValidator(build.MessagesTopic(nn), v.Validate); err != nil {
2020-02-17 05:51:18 +00:00
panic(err)
}
subscribe := func() {
log.Infof("subscribing to pubsub topic %s", build.MessagesTopic(nn))
msgsub, err := ps.Subscribe(build.MessagesTopic(nn)) //nolint
if err != nil {
panic(err)
}
go sub.HandleIncomingMessages(ctx, mpool, msgsub)
}
if bootstrapper {
subscribe()
return
}
// wait until we are synced within 10 epochs -- env var can override
waitForSync(stmgr, pubsubMsgsSyncEpochs, subscribe)
2019-07-08 14:07:09 +00:00
}
2019-08-01 17:12:41 +00:00
func RelayIndexerMessages(lc fx.Lifecycle, ps *pubsub.PubSub, nn dtypes.NetworkName, h host.Host, chainModule full.ChainModuleAPI, stateModule full.StateModuleAPI) error {
topicName := build.IndexerIngestTopic(nn)
v := sub.NewIndexerMessageValidator(h.ID(), chainModule, stateModule)
if err := ps.RegisterTopicValidator(topicName, v.Validate); err != nil {
return xerrors.Errorf("failed to register validator for topic %s, err: %w", topicName, err)
}
topicHandle, err := ps.Join(topicName)
if err != nil {
return xerrors.Errorf("failed to join pubsub topic %s: %w", topicName, err)
}
cancelFunc, err := topicHandle.Relay()
if err != nil {
return xerrors.Errorf("failed to relay to pubsub messages for topic %s: %w", topicName, err)
}
// Cancel message relay on shutdown.
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
cancelFunc()
return nil
},
})
log.Infof("relaying messages for pubsub topic %s", topicName)
return nil
}
2020-09-29 11:53:30 +00:00
func NewLocalDiscovery(lc fx.Lifecycle, ds dtypes.MetadataDS) (*discoveryimpl.Local, error) {
local, err := discoveryimpl.NewLocal(namespace.Wrap(ds, datastore.NewKey("/deals/local")))
if err != nil {
return nil, err
}
local.OnReady(marketevents.ReadyLogger("discovery"))
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
return local.Start(ctx)
},
})
return local, nil
}
2020-09-29 11:53:30 +00:00
func RetrievalResolver(l *discoveryimpl.Local) discovery.PeerResolver {
return discoveryimpl.Multi(l)
2019-08-26 13:45:36 +00:00
}
type RandomBeaconParams struct {
fx.In
PubSub *pubsub.PubSub `optional:"true"`
2020-06-23 19:10:27 +00:00
Cs *store.ChainStore
DrandConfig dtypes.DrandSchedule
}
func BuiltinDrandConfig() dtypes.DrandSchedule {
return build.DrandConfigSchedule()
}
2021-12-17 09:42:09 +00:00
func RandomSchedule(lc fx.Lifecycle, mctx helpers.MetricsCtx, p RandomBeaconParams, _ dtypes.AfterGenesisSet) (beacon.Schedule, error) {
gen, err := p.Cs.GetGenesis(helpers.LifecycleCtx(mctx, lc))
2020-04-14 03:05:19 +00:00
if err != nil {
return nil, err
}
shd := beacon.Schedule{}
for _, dc := range p.DrandConfig {
bc, err := drand.NewDrandBeacon(gen.Timestamp, build.BlockDelaySecs, p.PubSub, dc.Config)
if err != nil {
return nil, xerrors.Errorf("creating drand beacon: %w", err)
}
shd = append(shd, beacon.BeaconPoint{Start: dc.Start, Beacon: bc})
}
return shd, nil
}
2020-08-11 12:48:32 +00:00
func OpenFilesystemJournal(lr repo.LockedRepo, lc fx.Lifecycle, disabled journal.DisabledEvents) (journal.Journal, error) {
2021-08-17 12:55:35 +00:00
jrnl, err := fsjournal.OpenFSJournal(lr, disabled)
2020-08-11 12:48:32 +00:00
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error { return jrnl.Close() },
})
return jrnl, err
}