Merge remote-tracking branch 'origin/master' into feat/election-post

node/modules/chain.go

@@ -41,14 +41,17 @@ func ChainExchange(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt
 	return exch
 }
 
-func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub) *chain.MessagePool {
-	mp := chain.NewMessagePool(sm, ps)
+func MessagePool(lc fx.Lifecycle, sm *stmgr.StateManager, ps *pubsub.PubSub, ds dtypes.MetadataDS) (*chain.MessagePool, error) {
+	mp, err := chain.NewMessagePool(sm, ps, ds)
+	if err != nil {
+		return nil, xerrors.Errorf("constructing mpool: %w", err)
+	}
 	lc.Append(fx.Hook{
 		OnStop: func(_ context.Context) error {
 			return mp.Close()
 		},
 	})
-	return mp
+	return mp, nil
 }
 
 func ChainBlockstore(r repo.LockedRepo) (dtypes.ChainBlockstore, error) {

node/modules/core.go

@@ -3,6 +3,9 @@ package modules
 import (
 	"context"
 	"crypto/rand"
+	"io"
+	"io/ioutil"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -14,8 +17,6 @@ import (
 	"github.com/libp2p/go-libp2p-core/peerstore"
 	record "github.com/libp2p/go-libp2p-record"
 	"golang.org/x/xerrors"
-	"io"
-	"io/ioutil"
 )
 
 var log = logging.Logger("modules")

node/modules/lp2p/pubsub.go

@@ -1,21 +1,45 @@
 package lp2p
 
 import (
+	"context"
+
 	host "github.com/libp2p/go-libp2p-core/host"
+	peer "github.com/libp2p/go-libp2p-core/peer"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	ma "github.com/multiformats/go-multiaddr"
 	"go.uber.org/fx"
 
 	"github.com/filecoin-project/lotus/node/modules/helpers"
 )
 
-func FloodSub(pubsubOptions ...pubsub.Option) interface{} {
-	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host) (service *pubsub.PubSub, err error) {
-		return pubsub.NewFloodSub(helpers.LifecycleCtx(mctx, lc), host, pubsubOptions...)
+type PubsubOpt func(host.Host) pubsub.Option
+
+func PubsubTracer() PubsubOpt {
+	return func(host host.Host) pubsub.Option {
+		pi, err := peer.AddrInfoFromP2pAddr(ma.StringCast("/ip4/147.75.67.199/tcp/4001/p2p/QmTd6UvR47vUidRNZ1ZKXHrAFhqTJAD27rKL9XYghEKgKX"))
+		if err != nil {
+			panic(err)
+		}
+
+		tr, err := pubsub.NewRemoteTracer(context.TODO(), host, *pi)
+		if err != nil {
+			panic(err)
+		}
+
+		return pubsub.WithEventTracer(tr)
 	}
 }
 
-func GossipSub(pubsubOptions ...pubsub.Option) interface{} {
+func GossipSub(pubsubOptions ...PubsubOpt) interface{} {
 	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host) (service *pubsub.PubSub, err error) {
-		return pubsub.NewGossipSub(helpers.LifecycleCtx(mctx, lc), host, pubsubOptions...)
+		return pubsub.NewGossipSub(helpers.LifecycleCtx(mctx, lc), host, paresOpts(host, pubsubOptions)...)
 	}
 }
+
+func paresOpts(host host.Host, in []PubsubOpt) []pubsub.Option {
+	out := make([]pubsub.Option, len(in))
+	for k, v := range in {
+		out[k] = v(host)
+	}
+	return out
+}