Merge branch 'master' into nonsense/refactor-nodetype

node/builder.go

@@ -105,6 +105,8 @@ const (
 	HandleMigrateClientFundsKey
 	HandlePaymentChannelManagerKey
 
+	RelayIndexerMessagesKey
+
 	// miner
 	GetParamsKey
 	HandleMigrateProviderFundsKey

node/builder_chain.go

@@ -134,6 +134,8 @@ var ChainNode = Options(
 
 	Override(new(*full.GasPriceCache), full.NewGasPriceCache),
 
+	Override(RelayIndexerMessagesKey, modules.RelayIndexerMessages),
+
 	// Lite node API
 	ApplyIf(isLiteNode,
 		Override(new(messagepool.Provider), messagepool.NewProviderLite),

node/builder_miner.go

@@ -12,6 +12,7 @@ import (
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-fil-markets/storagemarket/impl/storedask"
 	"github.com/filecoin-project/go-state-types/abi"
+	provider "github.com/filecoin-project/index-provider"
 	storage2 "github.com/filecoin-project/specs-storage/storage"
 
 	"github.com/filecoin-project/lotus/api"
@@ -25,6 +26,7 @@ import (
 	sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
 	"github.com/filecoin-project/lotus/markets/dagstore"
 	"github.com/filecoin-project/lotus/markets/dealfilter"
+	"github.com/filecoin-project/lotus/markets/idxprov"
 	"github.com/filecoin-project/lotus/markets/retrievaladapter"
 	"github.com/filecoin-project/lotus/markets/sectoraccessor"
 	"github.com/filecoin-project/lotus/markets/storageadapter"
@@ -167,6 +169,8 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(new(dtypes.ProviderTransferNetwork), modules.NewProviderTransferNetwork),
 			Override(new(dtypes.ProviderTransport), modules.NewProviderTransport),
 			Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDataTransfer),
+			Override(new(idxprov.MeshCreator), idxprov.NewMeshCreator),
+			Override(new(provider.Interface), modules.IndexProvider(cfg.IndexProvider)),
 			Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
 			Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(cfg.Dealmaking, nil)),
 			Override(new(storagemarket.StorageProvider), modules.StorageProvider),

node/config/def.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"encoding"
+
 	"os"
 	"strconv"
 	"time"
@@ -182,6 +183,14 @@ func DefaultStorageMiner() *StorageMiner {
 			},
 		},
 
+		IndexProvider: IndexProviderConfig{
+			Enable:               false,
+			EntriesCacheCapacity: 1024,
+			EntriesChunkSize:     16384,
+			TopicName:            "/indexer/ingest/mainnet",
+			PurgeCacheOnStart:    false,
+		},
+
 		Subsystems: MinerSubsystemConfig{
 			EnableMining:        true,
 			EnableSealing:       true,
@@ -222,6 +231,7 @@ func DefaultStorageMiner() *StorageMiner {
 			GCInterval:                 Duration(1 * time.Minute),
 		},
 	}
+
 	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/def_test.go

@@ -52,3 +52,9 @@ func TestDefaultMinerRoundtrip(t *testing.T) {
 
 	require.True(t, reflect.DeepEqual(c, c2))
 }
+
+func TestDefaultStorageMiner_SetsIndexIngestTopic(t *testing.T) {
+	subject := DefaultStorageMiner()
+	require.False(t, subject.IndexProvider.Enable)
+	require.Equal(t, "/indexer/ingest/mainnet", subject.IndexProvider.TopicName)
+}

node/config/doc_gen.go

@@ -374,6 +374,49 @@ see https://docs.filecoin.io/mine/lotus/miner-configuration/#using-filters-for-f
 			Comment: ``,
 		},
 	},
+	"IndexProviderConfig": []DocField{
+		{
+			Name: "Enable",
+			Type: "bool",
+
+			Comment: `Enable set whether to enable indexing announcement to the network and expose endpoints that
+allow indexer nodes to process announcements. Disabled by default.`,
+		},
+		{
+			Name: "EntriesCacheCapacity",
+			Type: "int",
+
+			Comment: `EntriesCacheCapacity sets the maximum capacity to use for caching the indexing advertisement
+entries. Defaults to 1024 if not specified. The cache is evicted using LRU policy. The
+maximum storage used by the cache is a factor of EntriesCacheCapacity, EntriesChunkSize and
+the length of multihashes being advertised. For example, advertising 128-bit long multihashes
+with the default EntriesCacheCapacity, and EntriesChunkSize means the cache size can grow to
+256MiB when full.`,
+		},
+		{
+			Name: "EntriesChunkSize",
+			Type: "int",
+
+			Comment: `EntriesChunkSize sets the maximum number of multihashes to include in a single entries chunk.
+Defaults to 16384 if not specified. Note that chunks are chained together for indexing
+advertisements that include more multihashes than the configured EntriesChunkSize.`,
+		},
+		{
+			Name: "TopicName",
+			Type: "string",
+
+			Comment: `TopicName sets the topic name on which the changes to the advertised content are announced.
+Defaults to '/indexer/ingest/mainnet' if not specified.`,
+		},
+		{
+			Name: "PurgeCacheOnStart",
+			Type: "bool",
+
+			Comment: `PurgeCacheOnStart sets whether to clear any cached entries chunks when the provider engine
+starts. By default, the cache is rehydrated from previously cached entries stored in
+datastore if any is present.`,
+		},
+	},
 	"Libp2p": []DocField{
 		{
 			Name: "ListenAddresses",
@@ -849,6 +892,12 @@ Default is 20 (about once a week).`,
 
 			Comment: ``,
 		},
+		{
+			Name: "IndexProvider",
+			Type: "IndexProviderConfig",
+
+			Comment: ``,
+		},
 		{
 			Name: "Sealing",
 			Type: "SealingConfig",

node/config/types.go

@@ -43,13 +43,14 @@ type Backup struct {
 type StorageMiner struct {
 	Common
 
-	Subsystems MinerSubsystemConfig
-	Dealmaking DealmakingConfig
-	Sealing    SealingConfig
-	Storage    sectorstorage.SealerConfig
-	Fees       MinerFeeConfig
-	Addresses  MinerAddressConfig
-	DAGStore   DAGStoreConfig
+	Subsystems    MinerSubsystemConfig
+	Dealmaking    DealmakingConfig
+	IndexProvider IndexProviderConfig
+	Sealing       SealingConfig
+	Storage       sectorstorage.SealerConfig
+	Fees          MinerFeeConfig
+	Addresses     MinerAddressConfig
+	DAGStore      DAGStoreConfig
 }
 
 type DAGStoreConfig struct {
@@ -162,6 +163,35 @@ type DealmakingConfig struct {
 	RetrievalPricing *RetrievalPricing
 }
 
+type IndexProviderConfig struct {
+
+	// Enable set whether to enable indexing announcement to the network and expose endpoints that
+	// allow indexer nodes to process announcements. Disabled by default.
+	Enable bool
+
+	// EntriesCacheCapacity sets the maximum capacity to use for caching the indexing advertisement
+	// entries. Defaults to 1024 if not specified. The cache is evicted using LRU policy. The
+	// maximum storage used by the cache is a factor of EntriesCacheCapacity, EntriesChunkSize and
+	// the length of multihashes being advertised. For example, advertising 128-bit long multihashes
+	// with the default EntriesCacheCapacity, and EntriesChunkSize means the cache size can grow to
+	// 256MiB when full.
+	EntriesCacheCapacity int
+
+	// EntriesChunkSize sets the maximum number of multihashes to include in a single entries chunk.
+	// Defaults to 16384 if not specified. Note that chunks are chained together for indexing
+	// advertisements that include more multihashes than the configured EntriesChunkSize.
+	EntriesChunkSize int
+
+	// TopicName sets the topic name on which the changes to the advertised content are announced.
+	// Defaults to '/indexer/ingest/mainnet' if not specified.
+	TopicName string
+
+	// PurgeCacheOnStart sets whether to clear any cached entries chunks when the provider engine
+	// starts. By default, the cache is rehydrated from previously cached entries stored in
+	// datastore if any is present.
+	PurgeCacheOnStart bool
+}
+
 type RetrievalPricing struct {
 	Strategy string // possible values: "default", "external"
 

node/impl/full/chain.go

@@ -22,8 +22,6 @@ import (
 	ipld "github.com/ipfs/go-ipld-format"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/ipfs/go-merkledag"
-	"github.com/ipfs/go-path"
-	"github.com/ipfs/go-path/resolver"
 	mh "github.com/multiformats/go-multihash"
 	cbg "github.com/whyrusleeping/cbor-gen"
 
@@ -36,6 +34,8 @@ import (
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
+	"github.com/filecoin-project/lotus/lib/oldpath"
+	"github.com/filecoin-project/lotus/lib/oldpath/oldresolver"
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 )
 
@@ -550,17 +550,16 @@ func resolveOnce(bs blockstore.Blockstore, tse stmgr.Executor) func(ctx context.
 }
 
 func (a *ChainAPI) ChainGetNode(ctx context.Context, p string) (*api.IpldObject, error) {
-	ip, err := path.ParsePath(p)
+	ip, err := oldpath.ParsePath(p)
 	if err != nil {
 		return nil, xerrors.Errorf("parsing path: %w", err)
 	}
 
 	bs := a.ExposedBlockstore
 	bsvc := blockservice.New(bs, offline.Exchange(bs))
-
 	dag := merkledag.NewDAGService(bsvc)
 
-	r := &resolver.Resolver{
+	r := &oldresolver.Resolver{
 		DAG:         dag,
 		ResolveOnce: resolveOnce(bs, a.TsExec),
 	}

node/impl/net/protect.go

@@ -0,0 +1,35 @@
+package net
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+)
+
+const apiProtectTag = "api"
+
+func (a *NetAPI) NetProtectAdd(ctx context.Context, peers []peer.ID) error {
+	for _, p := range peers {
+		a.Host.ConnManager().Protect(p, apiProtectTag)
+	}
+
+	return nil
+}
+
+func (a *NetAPI) NetProtectRemove(ctx context.Context, peers []peer.ID) error {
+	for _, p := range peers {
+		a.Host.ConnManager().Unprotect(p, apiProtectTag)
+	}
+
+	return nil
+}
+
+func (a *NetAPI) NetProtectList(ctx context.Context) (result []peer.ID, err error) {
+	for _, conn := range a.Host.Network().Conns() {
+		if a.Host.ConnManager().IsProtected(conn.RemotePeer(), apiProtectTag) {
+			result = append(result, conn.RemotePeer())
+		}
+	}
+
+	return
+}

node/impl/storminer.go

@@ -1008,6 +1008,52 @@ func (sm *StorageMinerAPI) DagstoreGC(ctx context.Context) ([]api.DagstoreShardR
 	return ret, nil
 }
 
+func (sm *StorageMinerAPI) IndexerAnnounceDeal(ctx context.Context, proposalCid cid.Cid) error {
+	return sm.StorageProvider.AnnounceDealToIndexer(ctx, proposalCid)
+}
+
+func (sm *StorageMinerAPI) IndexerAnnounceAllDeals(ctx context.Context) error {
+	return sm.StorageProvider.AnnounceAllDealsToIndexer(ctx)
+}
+
+func (sm *StorageMinerAPI) DagstoreLookupPieces(ctx context.Context, cid cid.Cid) ([]api.DagstoreShardInfo, error) {
+	if sm.DAGStore == nil {
+		return nil, fmt.Errorf("dagstore not available on this node")
+	}
+
+	keys, err := sm.DAGStore.TopLevelIndex.GetShardsForMultihash(ctx, cid.Hash())
+	if err != nil {
+		return nil, err
+	}
+
+	var ret []api.DagstoreShardInfo
+
+	for _, k := range keys {
+		shard, err := sm.DAGStore.GetShardInfo(k)
+		if err != nil {
+			return nil, err
+		}
+
+		ret = append(ret, api.DagstoreShardInfo{
+			Key:   k.String(),
+			State: shard.ShardState.String(),
+			Error: func() string {
+				if shard.Error == nil {
+					return ""
+				}
+				return shard.Error.Error()
+			}(),
+		})
+	}
+
+	// order by key.
+	sort.SliceStable(ret, func(i, j int) bool {
+		return ret[i].Key < ret[j].Key
+	})
+
+	return ret, nil
+}
+
 func (sm *StorageMinerAPI) DealsList(ctx context.Context) ([]api.MarketDeal, error) {
 	return sm.listDeals(ctx)
 }

node/modules/lp2p/addrs.go

@@ -11,7 +11,7 @@ import (
 	mamask "github.com/whyrusleeping/multiaddr-filter"
 )
 
-func makeAddrsFactory(announce []string, noAnnounce []string) (p2pbhost.AddrsFactory, error) {
+func MakeAddrsFactory(announce []string, noAnnounce []string) (p2pbhost.AddrsFactory, error) {
 	var annAddrs []ma.Multiaddr
 	for _, addr := range announce {
 		maddr, err := ma.NewMultiaddr(addr)
@@ -59,7 +59,7 @@ func makeAddrsFactory(announce []string, noAnnounce []string) (p2pbhost.AddrsFac
 
 func AddrsFactory(announce []string, noAnnounce []string) func() (opts Libp2pOpts, err error) {
 	return func() (opts Libp2pOpts, err error) {
-		addrsFactory, err := makeAddrsFactory(announce, noAnnounce)
+		addrsFactory, err := MakeAddrsFactory(announce, noAnnounce)
 		if err != nil {
 			return opts, err
 		}

node/modules/lp2p/pubsub.go

@@ -114,6 +114,22 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 		InvalidMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
 	}
 
+	ingestTopicParams := &pubsub.TopicScoreParams{
+		// expected ~0.5 confirmed deals / min. sampled
+		TopicWeight: 0.1,
+
+		TimeInMeshWeight:  0.00027, // ~1/3600
+		TimeInMeshQuantum: time.Second,
+		TimeInMeshCap:     1,
+
+		FirstMessageDeliveriesWeight: 0.5,
+		FirstMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
+		FirstMessageDeliveriesCap:    100, // allowing for burstiness
+
+		InvalidMessageDeliveriesWeight: -1000,
+		InvalidMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
+	}
+
 	topicParams := map[string]*pubsub.TopicScoreParams{
 		build.BlocksTopic(in.Nn): {
 			// expected 10 blocks/min
@@ -208,6 +224,9 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 		drandTopics = append(drandTopics, topic)
 	}
 
+	// Index ingestion whitelist
+	topicParams[build.IndexerIngestTopic(in.Nn)] = ingestTopicParams
+
 	// IP colocation whitelist
 	var ipcoloWhitelist []*net.IPNet
 	for _, cidr := range in.Cfg.IPColocationWhitelist {
@@ -333,6 +352,7 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 	allowTopics := []string{
 		build.BlocksTopic(in.Nn),
 		build.MessagesTopic(in.Nn),
+		build.IndexerIngestTopic(in.Nn),
 	}
 	allowTopics = append(allowTopics, drandTopics...)
 	options = append(options,

node/modules/services.go

@@ -35,6 +35,7 @@ import (
 	"github.com/filecoin-project/lotus/lib/peermgr"
 	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"
 	"github.com/filecoin-project/lotus/node/repo"
@@ -198,6 +199,36 @@ func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub
 	waitForSync(stmgr, pubsubMsgsSyncEpochs, subscribe)
 }
 
+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
+}
+
 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 {

node/modules/storageminer.go

@@ -35,6 +35,7 @@ import (
 	"github.com/filecoin-project/go-state-types/big"
 	"github.com/filecoin-project/go-statestore"
 	"github.com/filecoin-project/go-storedcounter"
+	provider "github.com/filecoin-project/index-provider"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
@@ -61,6 +62,7 @@ import (
 	"github.com/filecoin-project/lotus/journal"
 	"github.com/filecoin-project/lotus/markets"
 	"github.com/filecoin-project/lotus/markets/dagstore"
+	"github.com/filecoin-project/lotus/markets/idxprov"
 	marketevents "github.com/filecoin-project/lotus/markets/loggers"
 	"github.com/filecoin-project/lotus/markets/pricing"
 	lotusminer "github.com/filecoin-project/lotus/miner"
@@ -584,10 +586,12 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 	h host.Host, ds dtypes.MetadataDS,
 	r repo.LockedRepo,
 	pieceStore dtypes.ProviderPieceStore,
+	indexer provider.Interface,
 	dataTransfer dtypes.ProviderDataTransfer,
 	spn storagemarket.StorageProviderNode,
 	df dtypes.StorageDealFilter,
 	dsw *dagstore.Wrapper,
+	meshCreator idxprov.MeshCreator,
 ) (storagemarket.StorageProvider, error) {
 	net := smnet.NewFromLibp2pHost(h)
 
@@ -612,11 +616,13 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 		namespace.Wrap(ds, datastore.NewKey("/deals/provider")),
 		store,
 		dsw,
+		indexer,
 		pieceStore,
 		dataTransfer,
 		spn,
 		address.Address(minerAddress),
 		storedAsk,
+		meshCreator,
 		opt,
 	)
 }

node/modules/storageminer_dagstore.go

@@ -7,6 +7,8 @@ import (
 	"path/filepath"
 	"strconv"
 
+	"github.com/libp2p/go-libp2p-core/host"
+
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
@@ -58,8 +60,8 @@ func NewMinerAPI(cfg config.DAGStoreConfig) func(fx.Lifecycle, repo.LockedRepo,
 // DAGStore constructs a DAG store using the supplied minerAPI, and the
 // user configuration. It returns both the DAGStore and the Wrapper suitable for
 // passing to markets.
-func DAGStore(cfg config.DAGStoreConfig) func(fx.Lifecycle, repo.LockedRepo, mdagstore.MinerAPI) (*dagstore.DAGStore, *mdagstore.Wrapper, error) {
-	return func(lc fx.Lifecycle, r repo.LockedRepo, minerAPI mdagstore.MinerAPI) (*dagstore.DAGStore, *mdagstore.Wrapper, error) {
+func DAGStore(cfg config.DAGStoreConfig) func(lc fx.Lifecycle, r repo.LockedRepo, minerAPI mdagstore.MinerAPI, h host.Host) (*dagstore.DAGStore, *mdagstore.Wrapper, error) {
+	return func(lc fx.Lifecycle, r repo.LockedRepo, minerAPI mdagstore.MinerAPI, h host.Host) (*dagstore.DAGStore, *mdagstore.Wrapper, error) {
 		// fall back to default root directory if not explicitly set in the config.
 		if cfg.RootDir == "" {
 			cfg.RootDir = filepath.Join(r.Path(), DefaultDAGStoreDir)
@@ -73,7 +75,7 @@ func DAGStore(cfg config.DAGStoreConfig) func(fx.Lifecycle, repo.LockedRepo, mda
 			}
 		}
 
-		dagst, w, err := mdagstore.NewDAGStore(cfg, minerAPI)
+		dagst, w, err := mdagstore.NewDAGStore(cfg, minerAPI, h)
 		if err != nil {
 			return nil, nil, xerrors.Errorf("failed to create DAG store: %w", err)
 		}

node/modules/storageminer_idxprov.go

@@ -0,0 +1,77 @@
+package modules
+
+import (
+	"context"
+
+	"github.com/filecoin-project/go-address"
+	provider "github.com/filecoin-project/index-provider"
+	"github.com/filecoin-project/index-provider/engine"
+	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/namespace"
+	"github.com/libp2p/go-libp2p-core/host"
+	"go.uber.org/fx"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/node/config"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+)
+
+type IdxProv struct {
+	fx.In
+
+	fx.Lifecycle
+	Datastore dtypes.MetadataDS
+}
+
+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"))
+		var opts = []engine.Option{
+			engine.WithDatastore(ipds),
+			engine.WithHost(marketHost),
+			engine.WithRetrievalAddrs(marketHost.Addrs()...),
+		}
+
+		llog := log.With("idxProvEnabled", cfg.Enable, "pid", marketHost.ID(), "retAddrs", marketHost.Addrs())
+		// If announcements to the network are enabled, then set options for datatransfer publisher.
+		if cfg.Enable {
+			// 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)
+			opts = append(opts,
+				engine.WithPublisherKind(engine.DataTransferPublisher),
+				engine.WithDataTransfer(dt),
+				engine.WithExtraGossipData(ma.Bytes()))
+			llog = llog.With("extraGossipData", ma)
+		} else {
+			opts = append(opts, engine.WithPublisherKind(engine.NoPublisher))
+		}
+
+		// Instantiate the index provider engine.
+		e, err := engine.New(opts...)
+		if err != nil {
+			return nil, xerrors.Errorf("creating indexer provider engine: %w", err)
+		}
+		llog.Info("Instantiated index provider engine")
+
+		args.Lifecycle.Append(fx.Hook{
+			OnStart: func(ctx context.Context) error {
+				// Note that the OnStart context is cancelled after startup. Its use in e.Start is
+				// to start up gossipsub publishers and restore cache, all of  which are completed
+				// before e.Start returns. Therefore, it is fine to reuse the give context.
+				if err := e.Start(ctx); err != nil {
+					return xerrors.Errorf("starting indexer provider engine: %w", err)
+				}
+				log.Infof("Started index provider engine")
+				return nil
+			},
+			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
+	}
+}