Remove redundant IndexProvider config now that host is shared

Now that host is shared having separate config for listen and announce
addresses along with graphsync limit makes no sense since all of that is
shared with the markets host and datatransfer manager.

documentation/en/default-lotus-miner-config.toml

@@ -273,30 +273,6 @@
   # env var: LOTUS_INDEXPROVIDER_PUBLISHERKIND
   #PublisherKind = "dtsync"
 
-  # Binding address for the libp2p host contacted by indexer nodes to sync the list of advertised
-  # multihashes. Note that when port is set to 0 a random port is generated at runtime and may be
-  # different on every restart. The format of the strings specified must conform to multiaddress;
-  # see https://multiformats.io/multiaddr/
-  #
-  # type: []string
-  # env var: LOTUS_INDEXPROVIDER_LISTENADDRESSES
-  #ListenAddresses = ["/ip4/0.0.0.0/tcp/0", "/ip6/::/tcp/0"]
-
-  # The address the endpoints at which the data associated to the advertised
-  # multihashes can be retrieved. If not specified, the ListenAddresses are used instead. The format
-  # of the strings specified must conform to multiaddress; see https://multiformats.io/multiaddr/
-  #
-  # type: []string
-  # env var: LOTUS_INDEXPROVIDER_ANNOUNCEADDRESSES
-  #AnnounceAddresses = []
-
-  # The maximum number of simultaneous requests syncing the list of advertised multihashes between
-  # the indexers and the index provider.
-  #
-  # type: uint64
-  # env var: LOTUS_INDEXPROVIDER_MAXSIMULTANEOUSTRANSFERS
-  #MaxSimultaneousTransfers = 20
-
   [IndexProvider.HttpPublisher]
     # env var: LOTUS_INDEXPROVIDER_HTTPPUBLISHER_LISTENMULTIADDR
     #ListenMultiaddr = "/ip4/0.0.0.0/tcp/3104/http"

markets/idxprov/host.go

@@ -1,5 +0,0 @@
-package idxprov
-
-import "github.com/libp2p/go-libp2p-core/host"
-
-type Host host.Host

markets/idxprov/mesh.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
 
 	"github.com/filecoin-project/lotus/api/v1api"
@@ -13,21 +14,23 @@ import (
 
 var log = logging.Logger("idxprov")
 
+const protectTag = "index-provider-gossipsub"
+
 type MeshCreator interface {
 	Connect(ctx context.Context) error
 }
 
 type Libp2pMeshCreator struct {
 	fullnodeApi v1api.FullNode
-	idxProvHost Host
+	marketsHost host.Host
 }
 
 func (mc Libp2pMeshCreator) Connect(ctx context.Context) error {
 
-	// Add the index provider's host ID to list of protected peers first, before any attempt to
-	// connect to full node.
-	idxProvID := mc.idxProvHost.ID()
-	if err := mc.fullnodeApi.NetProtectAdd(ctx, []peer.ID{idxProvID}); err != nil {
+	// Add the markets host ID to list of daemon's protected peers first, before any attempt to
+	// connect to full node over libp2p.
+	marketsPeerID := mc.marketsHost.ID()
+	if err := mc.fullnodeApi.NetProtectAdd(ctx, []peer.ID{marketsPeerID}); err != nil {
 		return fmt.Errorf("failed to call NetProtectAdd on the full node, err: %w", err)
 	}
 
@@ -38,17 +41,17 @@ func (mc Libp2pMeshCreator) Connect(ctx context.Context) error {
 
 	// Connect to the full node, ask it to protect the connection and protect the connection on
 	// markets end too.
-	if err := mc.idxProvHost.Connect(ctx, faddrs); err != nil {
+	if err := mc.marketsHost.Connect(ctx, faddrs); err != nil {
 		return fmt.Errorf("failed to connect index provider host with the full node: %w", err)
 	}
-	mc.idxProvHost.ConnManager().Protect(faddrs.ID, "index-provider-gossipsub")
+	mc.marketsHost.ConnManager().Protect(faddrs.ID, protectTag)
 
 	log.Debugw("successfully connected to full node and asked it protect indexer provider peer conn", "fullNodeInfo", faddrs.String(),
-		"idxProviderPeerId", idxProvID)
+		"peerId", marketsPeerID)
 
 	return nil
 }
 
-func NewMeshCreator(fullnodeApi v1api.FullNode, idxProvHost Host) MeshCreator {
-	return Libp2pMeshCreator{fullnodeApi, idxProvHost}
+func NewMeshCreator(fullnodeApi v1api.FullNode, marketsHost host.Host) MeshCreator {
+	return Libp2pMeshCreator{fullnodeApi, marketsHost}
 }

node/builder_miner.go

@@ -170,7 +170,6 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(new(dtypes.ProviderTransport), modules.NewProviderTransport),
 			Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDataTransfer),
 			Override(new(idxprov.MeshCreator), idxprov.NewMeshCreator),
-			Override(new(idxprov.Host), modules.IndexProviderHost(cfg.IndexProvider)),
 			Override(new(provider.Interface), modules.IndexProvider(cfg.IndexProvider)),
 			Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
 			Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(cfg.Dealmaking, nil)),

node/config/def.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"encoding"
+
 	"os"
 	"strconv"
 	"time"
@@ -182,13 +183,7 @@ func DefaultStorageMiner() *StorageMiner {
 		},
 
 		IndexProvider: IndexProviderConfig{
-			ListenAddresses: []string{
-				"/ip4/0.0.0.0/tcp/0",
-				"/ip6/::/tcp/0",
-			},
-			AnnounceAddresses: []string{},
-
-			MaxSimultaneousTransfers: DefaultSimultaneousTransfers,
+			Ingest: ipconfig.NewIngest(),
 		},
 
 		Subsystems: MinerSubsystemConfig{
@@ -232,10 +227,6 @@ func DefaultStorageMiner() *StorageMiner {
 		},
 	}
 
-	// TODO: Remove hardcoded defaults once provider library exposes them.
-	// See: https://github.com/filecoin-project/index-provider/issues/108
-	cfg.IndexProvider.Ingest = ipconfig.NewIngest()
-
 	cfg.Common.API.ListenAddress = "/ip4/127.0.0.1/tcp/2345/http"
 	cfg.Common.API.RemoteListenAddress = "127.0.0.1:2345"
 	return cfg

node/config/doc_gen.go

@@ -358,32 +358,6 @@ see https://docs.filecoin.io/mine/lotus/miner-configuration/#using-filters-for-f
 			Comment: ``,
 		},
 	},
-	"IndexProviderConfig": []DocField{
-		{
-			Name: "ListenAddresses",
-			Type: "[]string",
-
-			Comment: `Binding address for the libp2p host contacted by indexer nodes to sync the list of advertised
-multihashes. Note that when port is set to 0 a random port is generated at runtime and may be
-different on every restart. The format of the strings specified must conform to multiaddress;
-see https://multiformats.io/multiaddr/`,
-		},
-		{
-			Name: "AnnounceAddresses",
-			Type: "[]string",
-
-			Comment: `The address the endpoints at which the data associated to the advertised
-multihashes can be retrieved. If not specified, the ListenAddresses are used instead. The format
-of the strings specified must conform to multiaddress; see https://multiformats.io/multiaddr/`,
-		},
-		{
-			Name: "MaxSimultaneousTransfers",
-			Type: "uint64",
-
-			Comment: `The maximum number of simultaneous requests syncing the list of advertised multihashes between
-the indexers and the index provider.`,
-		},
-	},
 	"Libp2p": []DocField{
 		{
 			Name: "ListenAddresses",

node/config/types.go

@@ -163,21 +163,6 @@ type DealmakingConfig struct {
 
 type IndexProviderConfig struct {
 	config.Ingest
-
-	// Binding address for the libp2p host contacted by indexer nodes to sync the list of advertised
-	// multihashes. Note that when port is set to 0 a random port is generated at runtime and may be
-	// different on every restart. The format of the strings specified must conform to multiaddress;
-	// see https://multiformats.io/multiaddr/
-	ListenAddresses []string
-
-	// The address the endpoints at which the data associated to the advertised
-	// multihashes can be retrieved. If not specified, the ListenAddresses are used instead. The format
-	// of the strings specified must conform to multiaddress; see https://multiformats.io/multiaddr/
-	AnnounceAddresses []string
-
-	// The maximum number of simultaneous requests syncing the list of advertised multihashes between
-	// the indexers and the index provider.
-	MaxSimultaneousTransfers uint64
 }
 
 type RetrievalPricing struct {

node/modules/storageminer_idxprov.go

@@ -2,181 +2,70 @@ package modules
 
 import (
 	"context"
-	"fmt"
-	"os"
-	"path/filepath"
-
-	"golang.org/x/xerrors"
 
 	"github.com/ipfs/go-datastore"
-	"github.com/ipfs/go-graphsync"
-	graphsyncimpl "github.com/ipfs/go-graphsync/impl"
-	gsnet "github.com/ipfs/go-graphsync/network"
-	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
-	"github.com/libp2p/go-libp2p"
-	ci "github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/peerstore"
 	"go.uber.org/fx"
 
 	"github.com/filecoin-project/go-address"
-	datatransfer "github.com/filecoin-project/go-data-transfer"
-	dtimpl "github.com/filecoin-project/go-data-transfer/impl"
-	dtnet "github.com/filecoin-project/go-data-transfer/network"
-	dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
 	provider "github.com/filecoin-project/index-provider"
 	"github.com/filecoin-project/index-provider/engine"
 	"github.com/ipfs/go-datastore/namespace"
 	"github.com/libp2p/go-libp2p-core/host"
+	"golang.org/x/xerrors"
 
-	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/markets/idxprov"
-	marketevents "github.com/filecoin-project/lotus/markets/loggers"
 	"github.com/filecoin-project/lotus/node/config"
 	"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/repo"
 )
 
 type IdxProv struct {
 	fx.In
 
-	helpers.MetricsCtx
 	fx.Lifecycle
-	Repo      repo.LockedRepo
 	Datastore dtypes.MetadataDS
 	PeerID    peer.ID
 	peerstore.Peerstore
 }
 
-func IndexProviderHost(cfg config.IndexProviderConfig) func(IdxProv) (idxprov.Host, error) {
-	return func(args IdxProv) (idxprov.Host, error) {
-		pkey := args.Peerstore.PrivKey(args.PeerID)
-		if pkey == nil {
-			return nil, fmt.Errorf("missing private key for node ID: %s", args.PeerID.Pretty())
-		}
-
-		h, err := createIndexProviderHost(args.MetricsCtx, args.Lifecycle, pkey, args.Peerstore, cfg.ListenAddresses, cfg.AnnounceAddresses)
-		if err != nil {
-			return nil, xerrors.Errorf("creating indexer provider host: %w", err)
-		}
-
-		return h, nil
-	}
-}
-
-func IndexProvider(cfg config.IndexProviderConfig) func(params IdxProv, marketHost host.Host, h idxprov.Host, maddr dtypes.MinerAddress) (provider.Interface, error) {
-	return func(args IdxProv, marketHost host.Host, h idxprov.Host, maddr dtypes.MinerAddress) (provider.Interface, error) {
+func IndexProvider(cfg config.IndexProviderConfig) func(params IdxProv, marketHost host.Host, dt dtypes.ProviderDataTransfer, maddr dtypes.MinerAddress) (provider.Interface, error) {
+	return func(args IdxProv, marketHost host.Host, dt dtypes.ProviderDataTransfer, maddr dtypes.MinerAddress) (provider.Interface, error) {
 		ipds := namespace.Wrap(args.Datastore, datastore.NewKey("/index-provider"))
 
 		pkey := args.Peerstore.PrivKey(args.PeerID)
 		if pkey == nil {
-			return nil, fmt.Errorf("missing private key for node ID: %s", args.PeerID.Pretty())
+			return nil, xerrors.Errorf("missing private key for node ID: %s", args.PeerID)
 		}
 
-		dt, err := newIndexProviderDataTransfer(cfg, args.MetricsCtx, args.Lifecycle, args.Repo, h, ipds)
-		if err != nil {
-			return nil, err
-		}
-
-		var maddrs []string
+		var retAdds []string
 		for _, a := range marketHost.Addrs() {
-			maddrs = append(maddrs, a.String())
+			retAdds = append(retAdds, a.String())
 		}
 
 		// Get the miner ID and set as extra gossip data.
 		// The extra data is required by the lotus-specific index-provider gossip message validators.
 		ma := address.Address(maddr)
-		log.Infof("Using extra gossip data in index provider engine: %s", ma.String())
+		log.Infof("Using extra gossip data in index provider engine: %s", ma)
 
-		e, err := engine.New(cfg.Ingest, pkey, dt, h, ipds, maddrs, engine.WithExtraGossipData(ma.Bytes()))
+		e, err := engine.New(cfg.Ingest, pkey, dt, marketHost, ipds, retAdds, engine.WithExtraGossipData(ma.Bytes()))
 		if err != nil {
 			return nil, xerrors.Errorf("creating indexer provider engine: %w", err)
 		}
 
 		args.Lifecycle.Append(fx.Hook{
 			OnStart: func(ctx context.Context) error {
-				// Start the engine
-				err := e.Start(ctx)
-				if err != nil {
-					return xerrors.Errorf("starting indexer provider engine: %s", err)
+				if err := e.Start(ctx); err != nil {
+					return xerrors.Errorf("starting indexer provider engine: %w", err)
 				}
-
 				return nil
 			},
-			OnStop: func(ctx context.Context) error {
-				return e.Shutdown()
+			OnStop: func(_ context.Context) error {
+				if err := e.Shutdown(); err != nil {
+					return xerrors.Errorf("shutting down indexer provider engine: %w", err)
+				}
+				return nil
 			},
 		})
-
 		return e, nil
 	}
 }
-
-func createIndexProviderHost(mctx helpers.MetricsCtx, lc fx.Lifecycle, pkey ci.PrivKey, pstore peerstore.Peerstore, listenAddrs []string, announceAddrs []string) (host.Host, error) {
-	addrsFactory, err := lp2p.MakeAddrsFactory(announceAddrs, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	opts := []libp2p.Option{
-		libp2p.Identity(pkey),
-		libp2p.Peerstore(pstore),
-		libp2p.ListenAddrStrings(listenAddrs...),
-		libp2p.AddrsFactory(addrsFactory),
-		libp2p.Ping(true),
-		libp2p.UserAgent("lotus-indexer-provider-" + build.UserVersion()),
-	}
-
-	h, err := libp2p.New(opts...)
-	if err != nil {
-		return nil, err
-	}
-
-	lc.Append(fx.Hook{
-		OnStop: func(ctx context.Context) error {
-			return h.Close()
-		},
-	})
-
-	return h, nil
-}
-
-func newIndexProviderDataTransfer(cfg config.IndexProviderConfig, mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, h host.Host, ds datastore.Batching) (datatransfer.Manager, error) {
-	net := dtnet.NewFromLibp2pHost(h)
-
-	// Set up graphsync
-	gs := newIndexProviderGraphsync(cfg, mctx, lc, h)
-
-	// Set up data transfer
-	dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/transfers"))
-	transport := dtgstransport.NewTransport(h.ID(), gs)
-	dtPath := filepath.Join(r.Path(), "indexer-provider", "data-transfer")
-	err := os.MkdirAll(dtPath, 0755) //nolint: gosec
-	if err != nil && !os.IsExist(err) {
-		return nil, xerrors.Errorf("creating indexer provider data transfer dir %s: %w", dtPath, err)
-	}
-
-	dt, err := dtimpl.NewDataTransfer(dtDs, net, transport)
-	if err != nil {
-		return nil, xerrors.Errorf("creating indexer provider data transfer module: %w", err)
-	}
-
-	dt.OnReady(marketevents.ReadyLogger("indexer-provider data transfer"))
-	lc.Append(fx.Hook{
-		OnStart: dt.Start,
-		OnStop:  dt.Stop,
-	})
-	return dt, nil
-}
-
-func newIndexProviderGraphsync(cfg config.IndexProviderConfig, mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host) graphsync.GraphExchange {
-	graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
-	return graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc),
-		graphsyncNetwork,
-		cidlink.DefaultLinkSystem(),
-		graphsyncimpl.MaxInProgressIncomingRequests(cfg.MaxSimultaneousTransfers),
-		graphsyncimpl.MaxLinksPerIncomingRequests(config.MaxTraversalLinks),
-		graphsyncimpl.MaxLinksPerOutgoingRequests(config.MaxTraversalLinks))
-}