refactor piece doctor

cmd/boostd/recover.go

@@ -17,6 +17,7 @@ import (
 	"github.com/filecoin-project/boost/cmd/lib"
 	"github.com/filecoin-project/boost/db"
 	"github.com/filecoin-project/boost/piecedirectory"
+	bdclient "github.com/filecoin-project/boostd-data/client"
 	"github.com/filecoin-project/boostd-data/model"
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-commp-utils/writer"
@@ -190,14 +191,14 @@ func action(cctx *cli.Context) error {
 	if ignoreLID {
 		pd = nil
 	} else {
-		pdClient := piecedirectory.NewStore()
-		defer pdClient.Close(ctx)
-		err = pdClient.Dial(ctx, cctx.String("api-lid"))
+		cl := bdclient.NewStore()
+		defer cl.Close(ctx)
+		err = cl.Dial(ctx, cctx.String("api-lid"))
 		if err != nil {
 			return fmt.Errorf("connecting to local index directory service: %w", err)
 		}
 		pr := &piecedirectory.SectorAccessorAsPieceReader{SectorAccessor: sa}
-		pd = piecedirectory.NewPieceDirectory(pdClient, pr, cctx.Int("add-index-throttle"))
+		pd = piecedirectory.NewPieceDirectory(cl, pr, cctx.Int("add-index-throttle"))
 		pd.Start(ctx)
 	}
 

cmd/booster-bitswap/run.go

@@ -10,6 +10,7 @@ import (
 	"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"
@@ -157,9 +158,9 @@ var runCmd = &cli.Command{
 		defer storageCloser()
 
 		// Connect to the local index directory service
-		pdClient := piecedirectory.NewStore()
-		defer pdClient.Close(ctx)
-		err = pdClient.Dial(ctx, cctx.String("api-lid"))
+		cl := bdclient.NewStore()
+		defer cl.Close(ctx)
+		err = cl.Dial(ctx, cctx.String("api-lid"))
 		if err != nil {
 			return fmt.Errorf("connecting to local index directory service: %w", err)
 		}
@@ -196,8 +197,8 @@ var runCmd = &cli.Command{
 			return fmt.Errorf("starting block filter: %w", err)
 		}
 		pr := &piecedirectory.SectorAccessorAsPieceReader{SectorAccessor: sa}
-		piecedirectory := piecedirectory.NewPieceDirectory(pdClient, pr, cctx.Int("add-index-throttle"))
-		remoteStore := remoteblockstore.NewRemoteBlockstore(piecedirectory, &bitswapBlockMetrics)
+		pd := piecedirectory.NewPieceDirectory(cl, pr, cctx.Int("add-index-throttle"))
+		remoteStore := remoteblockstore.NewRemoteBlockstore(pd, &bitswapBlockMetrics)
 		server := NewBitswapServer(remoteStore, host, multiFilter)
 
 		var proxyAddrInfo *peer.AddrInfo
@@ -210,7 +211,7 @@ var runCmd = &cli.Command{
 		}
 
 		// Start the local index directory
-		piecedirectory.Start(ctx)
+		pd.Start(ctx)
 
 		// Start the bitswap server
 		log.Infof("Starting booster-bitswap node on port %d", port)

cmd/booster-http/run.go

@@ -14,6 +14,7 @@ import (
 	"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/model"
 	"github.com/filecoin-project/boostd-data/shared/tracing"
 	"github.com/filecoin-project/dagstore/mount"
@@ -127,9 +128,9 @@ var runCmd = &cli.Command{
 
 		// Connect to the local index directory service
 		ctx := lcli.ReqContext(cctx)
-		pdClient := piecedirectory.NewStore()
-		defer pdClient.Close(ctx)
-		err := pdClient.Dial(ctx, cctx.String("api-lid"))
+		cl := bdclient.NewStore()
+		defer cl.Close(ctx)
+		err := cl.Dial(ctx, cctx.String("api-lid"))
 		if err != nil {
 			return fmt.Errorf("connecting to local index directory service: %w", err)
 		}
@@ -168,7 +169,7 @@ var runCmd = &cli.Command{
 
 		// Create the server API
 		pr := &piecedirectory.SectorAccessorAsPieceReader{SectorAccessor: sa}
-		piecedirectory := piecedirectory.NewPieceDirectory(pdClient, pr, cctx.Int("add-index-throttle"))
+		pd := piecedirectory.NewPieceDirectory(cl, pr, cctx.Int("add-index-throttle"))
 
 		opts := &HttpServerOptions{
 			ServePieces:              servePieces,
@@ -199,11 +200,11 @@ var runCmd = &cli.Command{
 				GetSizeFailResponseCount:    metrics.HttpRblsGetSizeFailResponseCount,
 				GetSizeSuccessResponseCount: metrics.HttpRblsGetSizeSuccessResponseCount,
 			}
-			rbs := remoteblockstore.NewRemoteBlockstore(piecedirectory, &httpBlockMetrics)
+			rbs := remoteblockstore.NewRemoteBlockstore(pd, &httpBlockMetrics)
 			filtered := filters.NewFilteredBlockstore(rbs, multiFilter)
 			opts.Blockstore = filtered
 		}
-		sapi := serverApi{ctx: ctx, piecedirectory: piecedirectory, sa: sa}
+		sapi := serverApi{ctx: ctx, piecedirectory: pd, sa: sa}
 		server := NewHttpServer(
 			cctx.String("base-path"),
 			cctx.Int("port"),
@@ -212,7 +213,7 @@ var runCmd = &cli.Command{
 		)
 
 		// Start the local index directory
-		piecedirectory.Start(ctx)
+		pd.Start(ctx)
 
 		// Start the server
 		log.Infof("Starting booster-http node on port %d with base path '%s'",

extern/boostd-data/clientutil/clientutil.go

@@ -0,0 +1,28 @@
+package clientutil
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/filecoin-project/boostd-data/client"
+	"github.com/filecoin-project/boostd-data/svc"
+)
+
+func NewTestStore(ctx context.Context) *client.Store {
+	bdsvc, err := svc.NewLevelDB("")
+	if err != nil {
+		panic(err)
+	}
+	ln, err := bdsvc.Start(ctx, "localhost:0")
+	if err != nil {
+		panic(err)
+	}
+
+	cl := client.NewStore()
+	err = cl.Dial(ctx, fmt.Sprintf("ws://%s", ln.String()))
+	if err != nil {
+		panic(err)
+	}
+
+	return cl
+}

extern/boostd-data/cmd/run.go

@@ -163,7 +163,7 @@ func runAction(cctx *cli.Context, dbType string, store *svc.Service) error {
 
 	// Start the server
 	addr := cctx.String("addr")
-	err = store.Start(ctx, addr)
+	_, err = store.Start(ctx, addr)
 	if err != nil {
 		return fmt.Errorf("starting %s store: %w", dbType, err)
 	}

extern/boostd-data/svc/svc.go

@@ -54,15 +54,15 @@ func MakeLevelDBDir(repoPath string) (string, error) {
 	return repoPath, nil
 }
 
-func (s *Service) Start(ctx context.Context, addr string) error {
+func (s *Service) Start(ctx context.Context, addr string) (net.Addr, error) {
 	ln, err := net.Listen("tcp", addr)
 	if err != nil {
-		return fmt.Errorf("setting up listener for local index directory service: %w", err)
+		return nil, fmt.Errorf("setting up listener for local index directory service: %w", err)
 	}
 
 	err = s.Impl.Start(ctx)
 	if err != nil {
-		return fmt.Errorf("starting local index directory service: %w", err)
+		return nil, fmt.Errorf("starting local index directory service: %w", err)
 	}
 
 	server := jsonrpc.NewServer()
@@ -97,5 +97,5 @@ func (s *Service) Start(ctx context.Context, addr string) error {
 		<-done
 	}()
 
-	return nil
+	return ln.Addr(), nil
 }

extern/boostd-data/svc/svc_size_test.go

@@ -4,16 +4,19 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math"
+	"math/rand"
+	"testing"
+	"time"
+
 	"github.com/filecoin-project/boost/testutil"
 	"github.com/filecoin-project/boostd-data/client"
 	"github.com/filecoin-project/boostd-data/model"
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
+	"github.com/ipld/go-car/v2/index"
+	"github.com/multiformats/go-multihash"
 	"github.com/stretchr/testify/require"
-	"math"
-	"math/rand"
-	"testing"
-	"time"
 )
 
 var tlg = logging.Logger("tlg")
@@ -33,7 +36,7 @@ func TestSizeLimit(t *testing.T) {
 		bdsvc, err := NewLevelDB("")
 		require.NoError(t, err)
 
-		testSizeLimit(ctx, t, bdsvc, "localhost:8042")
+		testSizeLimit(ctx, t, bdsvc, "localhost:0")
 	})
 
 	t.Run("yugabyte", func(t *testing.T) {
@@ -43,17 +46,17 @@ func TestSizeLimit(t *testing.T) {
 
 		bdsvc := NewYugabyte(TestYugabyteSettings)
 
-		addr := "localhost:8044"
+		addr := "localhost:0"
 		testSizeLimit(ctx, t, bdsvc, addr)
 	})
 }
 
 func testSizeLimit(ctx context.Context, t *testing.T, bdsvc *Service, addr string) {
-	err := bdsvc.Start(ctx, addr)
+	ln, err := bdsvc.Start(ctx, addr)
 	require.NoError(t, err)
 
 	cl := client.NewStore()
-	err = cl.Dial(context.Background(), fmt.Sprintf("ws://%s", addr))
+	err = cl.Dial(context.Background(), fmt.Sprintf("ws://%s", ln.String()))
 	require.NoError(t, err)
 	defer cl.Close(ctx)
 
@@ -119,3 +122,20 @@ func generateRandomCid(baseCid []byte) (cid.Cid, error) {
 
 	return c, nil
 }
+
+func toEntries(idx index.Index) (map[string]uint64, bool) {
+	it, ok := idx.(index.IterableIndex)
+	if !ok {
+		return nil, false
+	}
+
+	entries := make(map[string]uint64)
+	err := it.ForEach(func(mh multihash.Multihash, o uint64) error {
+		entries[mh.String()] = o
+		return nil
+	})
+	if err != nil {
+		return nil, false
+	}
+	return entries, true
+}

extern/boostd-data/svc/svc_test.go

@@ -449,23 +449,6 @@ func compareIndices(subject, subjectDb index.Index) (bool, error) {
 	return equal, nil
 }
 
-func toEntries(idx index.Index) (map[string]uint64, bool) {
-	it, ok := idx.(index.IterableIndex)
-	if !ok {
-		return nil, false
-	}
-
-	entries := make(map[string]uint64)
-	err := it.ForEach(func(mh multihash.Multihash, o uint64) error {
-		entries[mh.String()] = o
-		return nil
-	})
-	if err != nil {
-		return nil, false
-	}
-	return entries, true
-}
-
 func TestCleanup(t *testing.T) {
 	_ = logging.SetLogLevel("*", "debug")