chore: replace filecoin-project/dagstore with ipfs-force-community/da…

…gstore
ipfs-force-community · Jun 28, 2023 · e204651 · e204651
1 parent 4cd6b86
commit e204651
Show file tree

Hide file tree

Showing 16 changed files with 219 additions and 303 deletions.
diff --git a/config/config.go b/config/config.go
@@ -88,7 +88,8 @@ type DAGStoreConfig struct {
 	// ReadDiretly enable to read piece storage directly skip transient file
 	UseTransient bool
 
-	MysqlShard *Mysql
+	// EnableMysqlShard used to config whether to save shard state to mysql
+	EnableMysqlShard bool
 }
 
 type MongoTopIndex struct {

diff --git a/config/def_config.go b/config/def_config.go
@@ -62,13 +62,6 @@ var DefaultMarketConfig = &MarketConfig{
 		MaxConcurrentIndex:         5,
 		MaxConcurrencyStorageCalls: 100,
 		GCInterval:                 Duration(1 * time.Minute),
-		MysqlShard: &Mysql{
-			ConnectionString: "",
-			MaxOpenConn:      100,
-			MaxIdleConn:      100,
-			ConnMaxLifeTime:  "1m",
-			Debug:            false,
-		},
 	},
 
 	SimultaneousTransfersForRetrieval:        DefaultSimultaneousTransfers,

diff --git a/dagstore/modules.go b/dagstore/modules.go
@@ -53,7 +53,13 @@ func CreateAndStartMarketAPI(ctx metrics.MetricsCtx, lc fx.Lifecycle, r *config.
 // 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 NewWrapperDAGStore(ctx metrics.MetricsCtx, lc fx.Lifecycle, homeDir *config.HomeDir, cfg *config.DAGStoreConfig, minerAPI MarketAPI) (*dagstore.DAGStore, stores.DAGStoreWrapper, error) {
+func NewWrapperDAGStore(ctx metrics.MetricsCtx,
+	lc fx.Lifecycle,
+	homeDir *config.HomeDir,
+	cfg *config.DAGStoreConfig,
+	minerAPI MarketAPI,
+	repo repo.Repo,
+) (*dagstore.DAGStore, stores.DAGStoreWrapper, error) {
 	// fall back to default root directory if not explicitly set in the config.
 	if cfg.RootDir == "" {
 		cfg.RootDir = filepath.Join(string(*homeDir), DefaultDAGStoreDir)
@@ -67,7 +73,7 @@ func NewWrapperDAGStore(ctx metrics.MetricsCtx, lc fx.Lifecycle, homeDir *config
 		}
 	}
 
-	dagst, w, err := NewDAGStore(ctx, cfg, minerAPI)
+	dagst, w, err := NewDAGStore(ctx, cfg, minerAPI, repo)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to create DAG store: %w", err)
 	}

diff --git a/dagstore/shard.go b/dagstore/shard.go
diff --git a/dagstore/wrapper.go b/dagstore/wrapper.go
@@ -19,6 +19,8 @@ import (
 
 	"github.com/filecoin-project/go-statemachine/fsm"
 	"github.com/ipfs-force-community/droplet/v2/config"
+	"github.com/ipfs-force-community/droplet/v2/models/badger"
+	"github.com/ipfs-force-community/droplet/v2/models/repo"
 	carindex "github.com/ipld/go-car/v2/index"
 
 	"github.com/filecoin-project/dagstore"
@@ -53,7 +55,11 @@ type Wrapper struct {
 
 var _ stores.DAGStoreWrapper = (*Wrapper)(nil)
 
-func NewDAGStore(ctx context.Context, cfg *config.DAGStoreConfig, marketApi MarketAPI) (*dagstore.DAGStore, *Wrapper, error) {
+func NewDAGStore(ctx context.Context,
+	cfg *config.DAGStoreConfig,
+	marketApi MarketAPI,
+	repo repo.Repo,
+) (*dagstore.DAGStore, *Wrapper, error) {
 	// construct the DAG Store.
 	registry := mount.NewRegistry()
 	if err := registry.Register(marketScheme, mountTemplate(marketApi, cfg.UseTransient)); err != nil {
@@ -85,12 +91,14 @@ func NewDAGStore(ctx context.Context, cfg *config.DAGStoreConfig, marketApi Mark
 		return nil, nil, fmt.Errorf("failed to create dagstore datastore in %s: %w", datastoreDir, err)
 	}
 
-	if cfg.MysqlShard != nil && len(cfg.MysqlShard.ConnectionString) != 0 {
-		shardRepo, err := newShardRepo(cfg.MysqlShard)
-		if err != nil {
-			return nil, nil, fmt.Errorf("failed to shard repo: %v", err)
+	var shardRepo dagstore.ShardRepo
+	if cfg.EnableMysqlShard {
+		shardRepo = repo.ShardRepo()
+		if _, ok := shardRepo.(*badger.Shard); ok {
+			return nil, nil, fmt.Errorf("EnableMysqlShard is true, must use mysql to store shard state")
 		}
-		dstore = &shardWrapper{ds: dstore, shardRepo: shardRepo}
+	} else {
+		shardRepo = dagstore.NewBadgerShardRepo(dstore)
 	}
 
 	irepo, err := index.NewFSRepo(indexDir)
@@ -101,7 +109,7 @@ func NewDAGStore(ctx context.Context, cfg *config.DAGStoreConfig, marketApi Mark
 	dCfg := dagstore.Config{
 		TransientsDir: transientsDir,
 		IndexRepo:     irepo,
-		Datastore:     dstore,
+		ShardRepo:     shardRepo,
 		MountRegistry: registry,
 		FailureCh:     failureCh,
 		TraceCh:       traceCh,
@@ -509,14 +517,3 @@ func (w *Wrapper) Close() error {
 
 	return nil
 }
-
-type shardWrapper struct {
-	ds ds.Batching
-	*shardRepo
-}
-
-func (s *shardWrapper) Batch(ctx context.Context) (ds.Batch, error) {
-	return s.ds.Batch(ctx)
-}
-
-var _ ds.Batching = &shardWrapper{}
diff --git a/dagstore/wrapper_test.go b/dagstore/wrapper_test.go
@@ -36,7 +36,7 @@ func TestWrapperLoadShard(t *testing.T) {
 	dagst, w, err := NewDAGStore(ctx, &config.DAGStoreConfig{
 		RootDir:    t.TempDir(),
 		GCInterval: config.Duration(1 * time.Millisecond),
-	}, mockLotusMount{})
+	}, mockLotusMount{}, nil)
 	require.NoError(t, err)
 
 	defer dagst.Close() //nolint:errcheck
@@ -101,7 +101,7 @@ func TestWrapperBackground(t *testing.T) {
 	dagst, w, err := NewDAGStore(ctx, &config.DAGStoreConfig{
 		RootDir:    t.TempDir(),
 		GCInterval: config.Duration(1 * time.Millisecond),
-	}, mockLotusMount{})
+	}, mockLotusMount{}, nil)
 	require.NoError(t, err)
 
 	defer dagst.Close() //nolint:errcheck

diff --git a/go.mod b/go.mod
@@ -198,7 +198,7 @@ require (
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-random v0.0.0-20190219211222-123a90aedc0c // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
-	github.com/jbenet/goprocess v0.1.4
+	github.com/jbenet/goprocess v0.1.4 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.2 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
@@ -314,6 +314,7 @@ require (
 )
 
 replace (
+	github.com/filecoin-project/dagstore => github.com/ipfs-force-community/dagstore v0.4.4-0.20230628060530-4b25fff4d833
 	github.com/filecoin-project/filecoin-ffi => ./extern/filecoin-ffi
 	github.com/filecoin-project/go-jsonrpc => github.com/ipfs-force-community/go-jsonrpc v0.1.7
 )
diff --git a/go.sum b/go.sum
@@ -351,9 +351,6 @@ github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL
 github.com/fatih/color v1.12.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
 github.com/fatih/color v1.13.0 h1:8LOYc1KYPPmyKMuN8QV2DNRWNbLo6LZ0iLs8+mlH53w=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
-github.com/filecoin-project/dagstore v0.5.2/go.mod h1:mdqKzYrRBHf1pRMthYfMv3n37oOw0Tkx7+TxPt240M0=
-github.com/filecoin-project/dagstore v0.6.0 h1:/ntQJEgCb8QfXqTVRFOCapUYmAvtoaNOtZRxzpJhbgU=
-github.com/filecoin-project/dagstore v0.6.0/go.mod h1:YKn4qXih+/2xQWpfJsaKGOi4POw5vH5grDmfPCCnx8g=
 github.com/filecoin-project/go-address v0.0.3/go.mod h1:jr8JxKsYx+lQlQZmF5i2U0Z+cGQ59wMIps/8YW/lDj8=
 github.com/filecoin-project/go-address v0.0.5/go.mod h1:jr8JxKsYx+lQlQZmF5i2U0Z+cGQ59wMIps/8YW/lDj8=
 github.com/filecoin-project/go-address v0.0.6/go.mod h1:7B0/5DA13n6nHkB8bbGx1gWzG/dbTsZ0fgOJVGsM3TE=
@@ -826,6 +823,8 @@ github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod
 github.com/influxdata/influxdb1-client v0.0.0-20200827194710-b269163b24ab/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
 github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 h1:W9WBk7wlPfJLvMCdtV4zPulc4uCPrlywQOmbFOhgQNU=
 github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
+github.com/ipfs-force-community/dagstore v0.4.4-0.20230628060530-4b25fff4d833 h1:2twcWpKmKa7iGXjw+wuN0suXavrJS5U7RYjCJml8qSI=
+github.com/ipfs-force-community/dagstore v0.4.4-0.20230628060530-4b25fff4d833/go.mod h1:YKn4qXih+/2xQWpfJsaKGOi4POw5vH5grDmfPCCnx8g=
 github.com/ipfs-force-community/go-jsonrpc v0.1.7 h1:e0ZTapGFhDY54j0QpRYN54Q3FHawUBQAM1KvXOzZtYY=
 github.com/ipfs-force-community/go-jsonrpc v0.1.7/go.mod h1:jBSvPTl8V1N7gSTuCR4bis8wnQnIjHbRPpROol6iQKM=
 github.com/ipfs-force-community/metrics v1.0.0/go.mod h1:mn40SioMuKtjmRumHFy/fJ26Pn028XuDjUJE9dorjyw=

diff --git a/models/badger/db.go b/models/badger/db.go
@@ -248,6 +248,10 @@ func (r *BadgerRepo) RetrievalDealRepo() repo.IRetrievalDealRepo {
 	return NewRetrievalDealRepo(r.dsParams.RetrievalDealsDs)
 }
 
+func (r *BadgerRepo) ShardRepo() repo.IShardRepo {
+	return NewShardRepo()
+}
+
 func (r *BadgerRepo) Close() error {
 	// todo: to implement
 	return nil

diff --git a/models/badger/shard.go b/models/badger/shard.go
@@ -0,0 +1,31 @@
+package badger
+
+import (
+	"context"
+
+	"github.com/filecoin-project/dagstore"
+)
+
+// dagstore already implements ShardRepo, so we don't need to it again.
+// https://github.com/ipfs-force-community/dagstore/blob/master/shard_repo.go#L27
+type Shard struct{}
+
+func NewShardRepo() *Shard {
+	return &Shard{}
+}
+
+func (s *Shard) SaveShard(ctx context.Context, shard *dagstore.PersistedShard) error {
+	panic("implement me")
+}
+func (s *Shard) GetShard(ctx context.Context, key string) (*dagstore.PersistedShard, error) {
+	panic("implement me")
+}
+func (s *Shard) ListShards(ctx context.Context) ([]*dagstore.PersistedShard, error) {
+	panic("implement me")
+}
+func (s *Shard) HasShard(ctx context.Context, key string) (bool, error) {
+	panic("implement me")
+}
+func (s *Shard) DeleteShard(ctx context.Context, key string) error {
+	panic("implement me")
+}