feat: add local index directory

Internally this is still refered to as the piece directory

Co-authored-by: dirkmc <dirkmdev@gmail.com>
Co-authored-by: Anton Evangelatov <anton.evangelatov@gmail.com>

.circleci/config.yml

@@ -341,9 +341,6 @@ workflows:
           target: "`go list ./... | grep -v boost/itests`"
 
       - test:
-          name: test-itest-ipni
-          suite: itest-ipni
-          target: "./itests/ipni_publish_test.go"
-
-      - lid-docker-compose
-
+          name: local index directory
+          suite: all
+          cwd: "./extern/boostd-data"

api/api.go

@@ -8,6 +8,8 @@ import (
 	smtypes "github.com/filecoin-project/boost/storagemarket/types"
 	"github.com/filecoin-project/go-address"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
+	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
+	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-state-types/abi"
 	lapi "github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -33,31 +35,30 @@ type Boost interface {
 	Net
 
 	// MethodGroup: Boost
-	BoostIndexerAnnounceAllDeals(ctx context.Context) error                                                                                     //perm:admin
-	BoostIndexerListMultihashes(ctx context.Context, proposalCid cid.Cid) ([]multihash.Multihash, error)                                        //perm:admin
-	BoostIndexerAnnounceLatest(ctx context.Context) (cid.Cid, error)                                                                            //perm:admin
-	BoostIndexerAnnounceLatestHttp(ctx context.Context, urls []string) (cid.Cid, error)                                                         //perm:admin
-	BoostOfflineDealWithData(ctx context.Context, dealUuid uuid.UUID, filePath string, delAfterImport bool) (*ProviderDealRejectionInfo, error) //perm:admin
-	BoostDeal(ctx context.Context, dealUuid uuid.UUID) (*smtypes.ProviderDealState, error)                                                      //perm:admin
-	BoostDealBySignedProposalCid(ctx context.Context, proposalCid cid.Cid) (*smtypes.ProviderDealState, error)                                  //perm:admin
-	BoostDummyDeal(context.Context, smtypes.DealParams) (*ProviderDealRejectionInfo, error)                                                     //perm:admin
-	BoostDagstoreRegisterShard(ctx context.Context, key string) error                                                                           //perm:admin
-	BoostDagstoreDestroyShard(ctx context.Context, key string) error                                                                            //perm:admin
-	BoostDagstoreInitializeShard(ctx context.Context, key string) error                                                                         //perm:admin
-	BoostDagstoreInitializeAll(ctx context.Context, params DagstoreInitializeAllParams) (<-chan DagstoreInitializeAllEvent, error)              //perm:admin
-	BoostDagstoreRecoverShard(ctx context.Context, key string) error                                                                            //perm:admin
-	BoostDagstoreGC(ctx context.Context) ([]DagstoreShardResult, error)                                                                         //perm:admin
-	BoostDagstorePiecesContainingMultihash(ctx context.Context, mh multihash.Multihash) ([]cid.Cid, error)                                      //perm:read
-	BoostDagstoreListShards(ctx context.Context) ([]DagstoreShardInfo, error)                                                                   //perm:admin
-	BoostMakeDeal(context.Context, smtypes.DealParams) (*ProviderDealRejectionInfo, error)                                                      //perm:write
+	BoostIndexerAnnounceAllDeals(ctx context.Context) error                                                                        //perm:admin
+	BoostIndexerListMultihashes(ctx context.Context, proposalCid cid.Cid) ([]multihash.Multihash, error)                           //perm:admin
+	BoostOfflineDealWithData(ctx context.Context, dealUuid uuid.UUID, filePath string) (*ProviderDealRejectionInfo, error)         //perm:admin
+	BoostDeal(ctx context.Context, dealUuid uuid.UUID) (*smtypes.ProviderDealState, error)                                         //perm:admin
+	BoostDealBySignedProposalCid(ctx context.Context, proposalCid cid.Cid) (*smtypes.ProviderDealState, error)                     //perm:admin
+	BoostDummyDeal(context.Context, smtypes.DealParams) (*ProviderDealRejectionInfo, error)                                        //perm:admin
+	BoostDagstoreRegisterShard(ctx context.Context, key string) error                                                              //perm:admin
+	BoostDagstoreDestroyShard(ctx context.Context, key string) error                                                               //perm:admin
+	BoostDagstoreInitializeShard(ctx context.Context, key string) error                                                            //perm:admin
+	BoostDagstoreInitializeAll(ctx context.Context, params DagstoreInitializeAllParams) (<-chan DagstoreInitializeAllEvent, error) //perm:admin
+	BoostDagstoreRecoverShard(ctx context.Context, key string) error                                                               //perm:admin
+	BoostDagstoreGC(ctx context.Context) ([]DagstoreShardResult, error)                                                            //perm:admin
+	BoostDagstorePiecesContainingMultihash(ctx context.Context, mh multihash.Multihash) ([]cid.Cid, error)                         //perm:read
+	BoostDagstoreListShards(ctx context.Context) ([]DagstoreShardInfo, error)                                                      //perm:admin
+	BoostMakeDeal(context.Context, smtypes.DealParams) (*ProviderDealRejectionInfo, error)                                         //perm:write
 
 	// MethodGroup: Blockstore
 	BlockstoreGet(ctx context.Context, c cid.Cid) ([]byte, error)  //perm:read
 	BlockstoreHas(ctx context.Context, c cid.Cid) (bool, error)    //perm:read
 	BlockstoreGetSize(ctx context.Context, c cid.Cid) (int, error) //perm:read
 
 	// MethodGroup: PieceDirectory
-	PdBuildIndexForPieceCid(ctx context.Context, piececid cid.Cid) error //perm:admin
+	PdBuildIndexForPieceCid(ctx context.Context, piececid cid.Cid) error        //perm:admin
+	PdMarkIndexErrored(ctx context.Context, piececid cid.Cid, err string) error //perm:admin
 
 	// RuntimeSubsystems returns the subsystems that are enabled
 	// in this instance.

cmd/boost/retrieve_cmd.go

@@ -25,6 +25,11 @@ import (
 	"github.com/ipfs/go-cid"
 	flatfs "github.com/ipfs/go-ds-flatfs"
 	levelds "github.com/ipfs/go-ds-leveldb"
+	blockstore "github.com/ipfs/go-ipfs-blockstore"
+	offline "github.com/ipfs/go-ipfs-exchange-offline"
+	files "github.com/ipfs/go-ipfs-files"
+	"github.com/ipfs/go-merkledag"
+	unixfile "github.com/ipfs/go-unixfs/file"
 	"github.com/ipld/go-car"
 	"github.com/ipld/go-ipld-prime"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"

cmd/boostd/index.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"fmt"
-
 	bcli "github.com/filecoin-project/boost/cli"
 	lcli "github.com/filecoin-project/lotus/cli"
 	"github.com/ipfs/go-cid"
@@ -15,8 +14,6 @@ var indexProvCmd = &cli.Command{
 	Subcommands: []*cli.Command{
 		indexProvAnnounceAllCmd,
 		indexProvListMultihashesCmd,
-		indexProvAnnounceLatest,
-		indexProvAnnounceLatestHttp,
 	},
 }
 
@@ -72,58 +69,6 @@ var indexProvListMultihashesCmd = &cli.Command{
 		for _, mh := range mhs {
 			fmt.Println("  " + mh.String())
 		}
-
-		return nil
-	},
-}
-
-var indexProvAnnounceLatest = &cli.Command{
-	Name:  "announce-latest",
-	Usage: "Re-publish the latest existing advertisement to pubsub",
-	Action: func(cctx *cli.Context) error {
-		ctx := lcli.ReqContext(cctx)
-
-		napi, closer, err := bcli.GetBoostAPI(cctx)
-		if err != nil {
-			return err
-		}
-		defer closer()
-
-		c, err := napi.BoostIndexerAnnounceLatest(ctx)
-		if err != nil {
-			return err
-		}
-
-		fmt.Printf("Announced advertisement with cid %s\n", c)
-		return nil
-	},
-}
-
-var indexProvAnnounceLatestHttp = &cli.Command{
-	Name:  "announce-latest-http",
-	Usage: "Re-publish the latest existing advertisement to specific indexers over http",
-	Flags: []cli.Flag{
-		&cli.StringSliceFlag{
-			Name:     "announce-url",
-			Usage:    "The url(s) to announce to. If not specified, announces to the http urls in config",
-			Required: false,
-		},
-	},
-	Action: func(cctx *cli.Context) error {
-		ctx := lcli.ReqContext(cctx)
-
-		napi, closer, err := bcli.GetBoostAPI(cctx)
-		if err != nil {
-			return err
-		}
-		defer closer()
-
-		c, err := napi.BoostIndexerAnnounceLatestHttp(ctx, cctx.StringSlice("announce-url"))
-		if err != nil {
-			return err
-		}
-
-		fmt.Printf("Announced advertisement to indexers over http with cid %s\n", c)
 		return nil
 	},
 }

cmd/boostd/main.go

@@ -50,7 +50,6 @@ func main() {
 			dagstoreCmd,
 			netCmd,
 			pieceDirCmd,
-			recoverCmd,
 		},
 	}
 	app.Setup()
@@ -86,10 +85,6 @@ func before(cctx *cli.Context) error {
 		_ = logging.SetLogLevel("migrator", "DEBUG")
 		_ = logging.SetLogLevel("piecedir", "DEBUG")
 		_ = logging.SetLogLevel("fxlog", "DEBUG")
-		_ = logging.SetLogLevel("unsmgr", "DEBUG")
-		_ = logging.SetLogLevel("piecedoc", "DEBUG")
-		_ = logging.SetLogLevel("piecedirectory", "DEBUG")
-		_ = logging.SetLogLevel("sectorstatemgr", "DEBUG")
 	}
 
 	return nil

cmd/boostd/piecedir.go

@@ -16,6 +16,7 @@ var pieceDirCmd = &cli.Command{
 	Usage: "Manage Local Index Directory",
 	Subcommands: []*cli.Command{
 		pdIndexGenerate,
+		pdIndexMarkErroredCmd,
 	},
 }
 
@@ -55,3 +56,45 @@ var pdIndexGenerate = &cli.Command{
 		return nil
 	},
 }
+
+var pdIndexMarkErroredCmd = &cli.Command{
+	Name:  "mark-index",
+	Usage: "Mark an index errored for a given piece in the local index directory",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:     "piece-cid",
+			Usage:    "piece-cid of the index that will be marked as errored",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "error",
+			Usage:    "error message",
+			Required: true,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		ctx := lcli.ReqContext(cctx)
+
+		// parse piececid
+		piececid, err := cid.Decode(cctx.String("piece-cid"))
+		if err != nil {
+			return err
+		}
+
+		boostApi, ncloser, err := bcli.GetBoostAPI(cctx)
+		if err != nil {
+			return fmt.Errorf("getting boost api: %w", err)
+		}
+		defer ncloser()
+
+		errMsg := cctx.String("error")
+		err = boostApi.PdMarkIndexErrored(ctx, piececid, errMsg)
+		if err != nil {
+			return err
+		}
+
+		fmt.Printf("Marked %s as errored with \"%s\"\n", piececid, errMsg)
+
+		return nil
+	},
+}

cmd/booster-bitswap/run.go

@@ -5,12 +5,11 @@ import (
 	"net/http"
 	_ "net/http/pprof"
 
+	"github.com/filecoin-project/boost/cmd/booster-bitswap/filters"
+	"github.com/filecoin-project/boost/cmd/booster-bitswap/remoteblockstore"
 	"github.com/filecoin-project/boost/cmd/lib"
-	"github.com/filecoin-project/boost/cmd/lib/filters"
-	"github.com/filecoin-project/boost/cmd/lib/remoteblockstore"
 	"github.com/filecoin-project/boost/metrics"
 	"github.com/filecoin-project/boost/piecedirectory"
-	bdclient "github.com/filecoin-project/boostd-data/client"
 	"github.com/filecoin-project/boostd-data/shared/tracing"
 	lcli "github.com/filecoin-project/lotus/cli"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -158,28 +157,14 @@ var runCmd = &cli.Command{
 		defer storageCloser()
 
 		// Connect to the local index directory service
-		cl := bdclient.NewStore()
-		defer cl.Close(ctx)
-		err = cl.Dial(ctx, cctx.String("api-lid"))
+		pdClient := piecedirectory.NewStore()
+		defer pdClient.Close(ctx)
+		err = pdClient.Dial(ctx, cctx.String("api-lid"))
 		if err != nil {
 			return fmt.Errorf("connecting to local index directory service: %w", err)
 		}
 
 		// Create the bitswap host
-		bitswapBlockMetrics := remoteblockstore.BlockMetrics{
-			GetRequestCount:             metrics.BitswapRblsGetRequestCount,
-			GetFailResponseCount:        metrics.BitswapRblsGetFailResponseCount,
-			GetSuccessResponseCount:     metrics.BitswapRblsGetSuccessResponseCount,
-			BytesSentCount:              metrics.BitswapRblsBytesSentCount,
-			HasRequestCount:             metrics.BitswapRblsHasRequestCount,
-			HasFailResponseCount:        metrics.BitswapRblsHasFailResponseCount,
-			HasSuccessResponseCount:     metrics.BitswapRblsHasSuccessResponseCount,
-			GetSizeRequestCount:         metrics.BitswapRblsGetSizeRequestCount,
-			GetSizeFailResponseCount:    metrics.BitswapRblsGetSizeFailResponseCount,
-			GetSizeSuccessResponseCount: metrics.BitswapRblsGetSizeSuccessResponseCount,
-		}
-
-		// Create the server API
 		port := cctx.Int("port")
 		repoDir, err := homedir.Expand(cctx.String(FlagRepo.Name))
 		if err != nil {
@@ -197,8 +182,8 @@ var runCmd = &cli.Command{
 			return fmt.Errorf("starting block filter: %w", err)
 		}
 		pr := &piecedirectory.SectorAccessorAsPieceReader{SectorAccessor: sa}
-		pd := piecedirectory.NewPieceDirectory(cl, pr, cctx.Int("add-index-throttle"))
-		remoteStore := remoteblockstore.NewRemoteBlockstore(pd, &bitswapBlockMetrics)
+		piecedirectory := piecedirectory.NewPieceDirectory(pdClient, pr, cctx.Int("add-index-throttle"))
+		remoteStore := remoteblockstore.NewRemoteBlockstore(piecedirectory)
 		server := NewBitswapServer(remoteStore, host, multiFilter)
 
 		var proxyAddrInfo *peer.AddrInfo
@@ -211,7 +196,7 @@ var runCmd = &cli.Command{
 		}
 
 		// Start the local index directory
-		pd.Start(ctx)
+		piecedirectory.Start(ctx)
 
 		// Start the bitswap server
 		log.Infof("Starting booster-bitswap node on port %d", port)