When disabled in config, completely disable index provider (filecoin-…

…project#558)

* feat: completely disable index provider

* fix: dont announce if index provider disabled

cmd/boostd/main.go

@@ -61,6 +61,7 @@ func before(cctx *cli.Context) error {
 	_ = logging.SetLogLevel("boostd", "INFO")
 	_ = logging.SetLogLevel("db", "INFO")
 	_ = logging.SetLogLevel("boost-prop", "INFO")
+	_ = logging.SetLogLevel("modules", "INFO")
 
 	if cliutil.IsVeryVerbose {
 		_ = logging.SetLogLevel("boostd", "DEBUG")

go.mod

@@ -8,7 +8,7 @@ replace github.com/filecoin-project/filecoin-ffi => ./extern/filecoin-ffi
 
 require (
 	contrib.go.opencensus.io/exporter/prometheus v0.4.0
-	github.com/BurntSushi/toml v0.4.1
+	github.com/BurntSushi/toml v1.1.0
 	github.com/alecthomas/jsonschema v0.0.0-20200530073317-71f438968921
 	github.com/buger/goterm v1.0.3
 	github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e
@@ -38,6 +38,7 @@ require (
 	github.com/filecoin-project/specs-actors v0.9.14
 	github.com/filecoin-project/specs-actors/v2 v2.3.6
 	github.com/filecoin-project/specs-storage v0.2.2
+	github.com/filecoin-project/storetheindex v0.4.14
 	github.com/gbrlsnchs/jwt/v3 v3.0.1
 	github.com/golang/mock v1.6.0
 	github.com/google/uuid v1.3.0
@@ -90,8 +91,8 @@ require (
 	github.com/prometheus/client_golang v1.12.1
 	github.com/raulk/clock v1.1.0
 	github.com/raulk/go-watchdog v1.2.0
-	github.com/stretchr/testify v1.7.0
-	github.com/urfave/cli/v2 v2.3.0
+	github.com/stretchr/testify v1.7.1
+	github.com/urfave/cli/v2 v2.8.1
 	github.com/whyrusleeping/base32 v0.0.0-20170828182744-c30ac30633cc
 	github.com/whyrusleeping/cbor-gen v0.0.0-20220302191723-37c43cae8e14
 	go.opencensus.io v0.23.0

indexprovider/disabledprovider.go

@@ -0,0 +1,48 @@
+package indexprovider
+
+import (
+	"context"
+	"fmt"
+	"github.com/filecoin-project/index-provider"
+	"github.com/filecoin-project/index-provider/metadata"
+	"github.com/filecoin-project/storetheindex/api/v0/ingest/schema"
+	"github.com/ipfs/go-cid"
+)
+
+type DisabledIndexProvider struct {
+}
+
+func NewDisabledIndexProvider() *DisabledIndexProvider {
+	return &DisabledIndexProvider{}
+}
+
+func (d DisabledIndexProvider) PublishLocal(ctx context.Context, advertisement schema.Advertisement) (cid.Cid, error) {
+	return cid.Undef, fmt.Errorf("could not publish locally: index provider disabled")
+}
+
+func (d DisabledIndexProvider) Publish(ctx context.Context, advertisement schema.Advertisement) (cid.Cid, error) {
+	return cid.Undef, fmt.Errorf("could not publish: index provider disabled")
+}
+
+func (d DisabledIndexProvider) RegisterMultihashLister(lister provider.MultihashLister) {
+}
+
+func (d DisabledIndexProvider) NotifyPut(ctx context.Context, contextID []byte, md metadata.Metadata) (cid.Cid, error) {
+	return cid.Undef, fmt.Errorf("could not notify put: index provider disabled")
+}
+
+func (d DisabledIndexProvider) NotifyRemove(ctx context.Context, contextID []byte) (cid.Cid, error) {
+	return cid.Undef, fmt.Errorf("could not notify remove: index provider disabled")
+}
+
+func (d DisabledIndexProvider) GetAdv(ctx context.Context, cid cid.Cid) (*schema.Advertisement, error) {
+	return nil, fmt.Errorf("could not get advertisement: index provider disabled")
+}
+
+func (d DisabledIndexProvider) GetLatestAdv(ctx context.Context) (cid.Cid, *schema.Advertisement, error) {
+	return cid.Undef, nil, fmt.Errorf("could not get latest advertisement: index provider disabled")
+}
+
+func (d DisabledIndexProvider) Shutdown() error {
+	return nil
+}

indexprovider/wrapper.go

@@ -2,6 +2,7 @@ package indexprovider
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"os"
@@ -35,6 +36,7 @@ var defaultDagStoreDir = "dagstore"
 
 type Wrapper struct {
 	cfg         lotus_config.DAGStoreConfig
+	enabled     bool
 	dealsDB     *db.DealsDB
 	legacyProv  lotus_storagemarket.StorageProvider
 	prov        provider.Interface
@@ -53,18 +55,28 @@ func NewWrapper(cfg lotus_config.DAGStoreConfig) func(lc fx.Lifecycle, r repo.Lo
 			cfg.RootDir = filepath.Join(r.Path(), defaultDagStoreDir)
 		}
 
+		_, isDisabled := prov.(*DisabledIndexProvider)
 		return &Wrapper{
 			dealsDB:     dealsDB,
 			legacyProv:  legacyProv,
 			prov:        prov,
 			dagStore:    dagStore,
 			meshCreator: meshCreator,
 			cfg:         cfg,
+			enabled:     !isDisabled,
 		}
 	}
 }
 
+func (w *Wrapper) Enabled() bool {
+	return w.enabled
+}
+
 func (w *Wrapper) IndexerAnnounceAllDeals(ctx context.Context) error {
+	if !w.enabled {
+		return errors.New("cannot announce all deals: index provider is disabled")
+	}
+
 	log.Info("will announce all Markets deals to Indexer")
 	err := w.legacyProv.AnnounceAllDealsToIndexer(ctx)
 	if err != nil {
@@ -145,6 +157,10 @@ func (w *Wrapper) Start(ctx context.Context) {
 }
 
 func (w *Wrapper) AnnounceBoostDeal(ctx context.Context, pds *types.ProviderDealState) (cid.Cid, error) {
+	if !w.enabled {
+		return cid.Undef, errors.New("cannot announce deal: index provider is disabled")
+	}
+
 	// Announce deal to network Indexer
 	fm := metadata.New(&metadata.GraphsyncFilecoinV1{
 		PieceCID:      pds.ClientDealProposal.Proposal.PieceCID,

node/builder.go

@@ -499,7 +499,7 @@ func ConfigBoost(c interface{}) Option {
 		Override(new(retrievalmarket.RetrievalProvider), lotus_modules.RetrievalProvider),
 		Override(HandleRetrievalKey, lotus_modules.HandleRetrieval),
 		Override(new(idxprov.MeshCreator), idxprov.NewMeshCreator),
-		Override(new(provider.Interface), lotus_modules.IndexProvider(cfg.IndexProvider)),
+		Override(new(provider.Interface), modules.IndexProvider(cfg.IndexProvider)),
 
 		// Lotus Markets (storage)
 		Override(new(lotus_dtypes.ProviderTransferNetwork), lotus_modules.NewProviderTransferNetwork),

node/config/load.go

@@ -32,7 +32,7 @@ func FromFile(path string, def interface{}) (interface{}, error) {
 // FromReader loads config from a reader instance.
 func FromReader(reader io.Reader, def interface{}) (interface{}, error) {
 	cfg := def
-	_, err := toml.DecodeReader(reader, cfg)
+	_, err := toml.NewDecoder(reader).Decode(cfg)
 	if err != nil {
 		return nil, err
 	}

node/modules/storageminer.go

@@ -355,9 +355,9 @@ func HandleBoostDeals(lc fx.Lifecycle, h host.Host, prov *storagemarket.Provider
 			// Start the Boost Index Provider.
 			// It overrides the multihash lister registered by the legacy
 			// index provider so it must start after the legacy SP.
-			log.Info("starting boost index provider")
+			log.Info("starting boost index provider wrapper")
 			idxProv.Start(ctx)
-			log.Info("boost index provider started successfully")
+			log.Info("boost index provider wrapper started successfully")
 			return nil
 		},
 		OnStop: func(ctx context.Context) error {

node/modules/storageminer_idxprov.go

@@ -0,0 +1,21 @@
+package modules
+
+import (
+	"github.com/filecoin-project/boost/indexprovider"
+	provider "github.com/filecoin-project/index-provider"
+	"github.com/filecoin-project/lotus/node/config"
+	lotus_modules "github.com/filecoin-project/lotus/node/modules"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/libp2p/go-libp2p-core/host"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+)
+
+func IndexProvider(cfg config.IndexProviderConfig) func(params lotus_modules.IdxProv, marketHost host.Host, dt dtypes.ProviderDataTransfer, maddr dtypes.MinerAddress, ps *pubsub.PubSub, nn dtypes.NetworkName) (provider.Interface, error) {
+	if !cfg.Enable {
+		log.Warnf("Starting Boost with index provider disabled - no announcements will be made to the index provider")
+		return func(params lotus_modules.IdxProv, marketHost host.Host, dt dtypes.ProviderDataTransfer, maddr dtypes.MinerAddress, ps *pubsub.PubSub, nn dtypes.NetworkName) (provider.Interface, error) {
+			return indexprovider.NewDisabledIndexProvider(), nil
+		}
+	}
+	return lotus_modules.IndexProvider(cfg)
+}

storagemarket/deal_execution.go

@@ -654,15 +654,20 @@ func (p *Provider) indexAndAnnounce(ctx context.Context, pub event.Emitter, deal
 		p.dealLogger.Infow(deal.DealUuid, "deal has successfully been registered in the dagstore")
 	}
 
-	// announce to the network indexer but do not fail the deal if the announcement fails
-	annCid, err := p.ip.AnnounceBoostDeal(ctx, deal)
-	if err != nil {
-		return &dealMakingError{
-			retry: types.DealRetryAuto,
-			error: fmt.Errorf("failed to announce deal to network indexer: %w", err),
+	// if the index provider is enabled
+	if p.ip.Enabled() {
+		// announce to the network indexer but do not fail the deal if the announcement fails
+		annCid, err := p.ip.AnnounceBoostDeal(ctx, deal)
+		if err != nil {
+			return &dealMakingError{
+				retry: types.DealRetryAuto,
+				error: fmt.Errorf("failed to announce deal to network indexer: %w", err),
+			}
 		}
+		p.dealLogger.Infow(deal.DealUuid, "announced deal to network indexer", "announcement-cid", annCid)
+	} else {
+		p.dealLogger.Infow(deal.DealUuid, "didn't announce deal because network indexer is disabled")
 	}
-	p.dealLogger.Infow(deal.DealUuid, "announced deal to network indexer", "announcement-cid", annCid)
 
 	if derr := p.updateCheckpoint(pub, deal, dealcheckpoints.IndexedAndAnnounced); derr != nil {
 		return derr

storagemarket/provider_test.go

@@ -1944,6 +1944,10 @@ func (td *testDeal) assertDealFailedNonRecoverable(t *testing.T, ctx context.Con
 
 type NoOpIndexProvider struct{}
 
+func (n *NoOpIndexProvider) Enabled() bool {
+	return true
+}
+
 func (n *NoOpIndexProvider) AnnounceBoostDeal(ctx context.Context, pds *types.ProviderDealState) (cid.Cid, error) {
 	return testutil.GenerateCid(), nil
 }

storagemarket/types/types.go

@@ -115,6 +115,7 @@ type ChainDealManager interface {
 }
 
 type IndexProvider interface {
+	Enabled() bool
 	AnnounceBoostDeal(ctx context.Context, pds *ProviderDealState) (cid.Cid, error)
 	Start(ctx context.Context)
 }