allow carstore to use multiple directories, round robin style (#812)

bluesky-social · Nov 16, 2024 · 05b8751 · 05b8751
2 parents 0fc4c7e + ffe7fb6
commit 05b8751
Show file tree

Hide file tree

Showing 11 changed files with 55 additions and 28 deletions.
diff --git a/carstore/bs.go b/carstore/bs.go
@@ -62,21 +62,23 @@ type CarStore interface {
 }
 
 type FileCarStore struct {
-	meta    *CarStoreGormMeta
-	rootDir string
+	meta     *CarStoreGormMeta
+	rootDirs []string
 
 	lscLk          sync.Mutex
 	lastShardCache map[models.Uid]*CarShard
 }
 
-func NewCarStore(meta *gorm.DB, root string) (CarStore, error) {
-	if _, err := os.Stat(root); err != nil {
-		if !os.IsNotExist(err) {
-			return nil, err
-		}
+func NewCarStore(meta *gorm.DB, roots []string) (CarStore, error) {
+	for _, root := range roots {
+		if _, err := os.Stat(root); err != nil {
+			if !os.IsNotExist(err) {
+				return nil, err
+			}
 
-		if err := os.Mkdir(root, 0775); err != nil {
-			return nil, err
+			if err := os.Mkdir(root, 0775); err != nil {
+				return nil, err
+			}
 		}
 	}
 	if err := meta.AutoMigrate(&CarShard{}, &blockRef{}); err != nil {
@@ -88,7 +90,7 @@ func NewCarStore(meta *gorm.DB, root string) (CarStore, error) {
 
 	return &FileCarStore{
 		meta:           &CarStoreGormMeta{meta: meta},
-		rootDir:        root,
+		rootDirs:       roots,
 		lastShardCache: make(map[models.Uid]*CarShard),
 	}, nil
 }
@@ -541,9 +543,14 @@ func (ds *DeltaSession) GetSize(ctx context.Context, c cid.Cid) (int, error) {
 func fnameForShard(user models.Uid, seq int) string {
 	return fmt.Sprintf("sh-%d-%d", user, seq)
 }
+
+func (cs *FileCarStore) dirForUser(user models.Uid) string {
+	return cs.rootDirs[int(user)%len(cs.rootDirs)]
+}
+
 func (cs *FileCarStore) openNewShardFile(ctx context.Context, user models.Uid, seq int) (*os.File, string, error) {
 	// TODO: some overwrite protections
-	fname := filepath.Join(cs.rootDir, fnameForShard(user, seq))
+	fname := filepath.Join(cs.dirForUser(user), fnameForShard(user, seq))
 	fi, err := os.Create(fname)
 	if err != nil {
 		return nil, "", err
@@ -557,7 +564,7 @@ func (cs *FileCarStore) writeNewShardFile(ctx context.Context, user models.Uid,
 	defer span.End()
 
 	// TODO: some overwrite protections
-	fname := filepath.Join(cs.rootDir, fnameForShard(user, seq))
+	fname := filepath.Join(cs.dirForUser(user), fnameForShard(user, seq))
 	if err := os.WriteFile(fname, data, 0664); err != nil {
 		return "", err
 	}
@@ -982,7 +989,7 @@ func (cs *FileCarStore) openNewCompactedShardFile(ctx context.Context, user mode
 	// TODO: some overwrite protections
 	// NOTE CreateTemp is used for creating a non-colliding file, but we keep it and don't delete it so don't think of it as "temporary".
 	// This creates "sh-%d-%d%s" with some random stuff in the last position
-	fi, err := os.CreateTemp(cs.rootDir, fnameForShard(user, seq))
+	fi, err := os.CreateTemp(cs.dirForUser(user), fnameForShard(user, seq))
 	if err != nil {
 		return nil, "", err
 	}

diff --git a/carstore/repo_test.go b/carstore/repo_test.go
@@ -30,8 +30,13 @@ func testCarStore() (CarStore, func(), error) {
 		return nil, nil, err
 	}
 
-	sharddir := filepath.Join(tempdir, "shards")
-	if err := os.MkdirAll(sharddir, 0775); err != nil {
+	sharddir1 := filepath.Join(tempdir, "shards1")
+	if err := os.MkdirAll(sharddir1, 0775); err != nil {
+		return nil, nil, err
+	}
+
+	sharddir2 := filepath.Join(tempdir, "shards2")
+	if err := os.MkdirAll(sharddir2, 0775); err != nil {
 		return nil, nil, err
 	}
 
@@ -45,7 +50,7 @@ func testCarStore() (CarStore, func(), error) {
 		return nil, nil, err
 	}
 
-	cs, err := NewCarStore(db, sharddir)
+	cs, err := NewCarStore(db, []string{sharddir1, sharddir2})
 	if err != nil {
 		return nil, nil, err
 	}

diff --git a/cmd/bigsky/main.go b/cmd/bigsky/main.go
@@ -200,6 +200,11 @@ func run(args []string) error {
 			EnvVars: []string{"RELAY_NUM_COMPACTION_WORKERS"},
 			Value:   2,
 		},
+		&cli.StringSliceFlag{
+			Name:    "carstore-shard-dirs",
+			Usage:   "specify list of shard directories for carstore storage, overrides default storage within datadir",
+			EnvVars: []string{"RELAY_CARSTORE_SHARD_DIRS"},
+		},
 	}
 
 	app.Action = runBigsky
@@ -312,8 +317,18 @@ func runBigsky(cctx *cli.Context) error {
 		}
 	}
 
-	os.MkdirAll(filepath.Dir(csdir), os.ModePerm)
-	cstore, err := carstore.NewCarStore(csdb, csdir)
+	csdirs := []string{csdir}
+	if paramDirs := cctx.StringSlice("carstore-shard-dirs"); len(paramDirs) > 0 {
+		csdirs = paramDirs
+	}
+
+	for _, csd := range csdirs {
+		if err := os.MkdirAll(filepath.Dir(csd), os.ModePerm); err != nil {
+			return err
+		}
+	}
+
+	cstore, err := carstore.NewCarStore(csdb, csdirs)
 	if err != nil {
 		return err
 	}

diff --git a/cmd/laputa/main.go b/cmd/laputa/main.go
@@ -158,7 +158,7 @@ func run(args []string) {
 			}
 		}
 
-		cstore, err := carstore.NewCarStore(csdb, csdir)
+		cstore, err := carstore.NewCarStore(csdb, []string{csdir})
 		if err != nil {
 			return err
 		}

diff --git a/cmd/supercollider/main.go b/cmd/supercollider/main.go
@@ -565,7 +565,7 @@ func initSpeedyRepoMan(key *godid.PrivKey) (*repomgr.RepoManager, *godid.PrivKey
 		return nil, nil, err
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		return nil, nil, err
 	}

diff --git a/events/dbpersist_test.go b/events/dbpersist_test.go
@@ -301,7 +301,7 @@ func setupDBs(t testing.TB) (*gorm.DB, *gorm.DB, carstore.CarStore, string, erro
 		return nil, nil, nil, "", err
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		return nil, nil, nil, "", err
 	}

diff --git a/indexer/posts_test.go b/indexer/posts_test.go
@@ -50,7 +50,7 @@ func testIndexer(t *testing.T) *testIx {
 		t.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		t.Fatal(err)
 	}

diff --git a/pds/handlers_test.go b/pds/handlers_test.go
@@ -29,7 +29,7 @@ func testCarStore(t *testing.T, db *gorm.DB) (carstore.CarStore, func()) {
 		t.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(db, sharddir)
+	cs, err := carstore.NewCarStore(db, []string{sharddir})
 	if err != nil {
 		t.Fatal(err)
 	}

diff --git a/repomgr/bench_test.go b/repomgr/bench_test.go
@@ -54,7 +54,7 @@ func BenchmarkRepoMgrCreates(b *testing.B) {
 		b.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		b.Fatal(err)
 	}

diff --git a/repomgr/ingest_test.go b/repomgr/ingest_test.go
@@ -50,7 +50,7 @@ func TestLoadNewRepo(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -80,7 +80,7 @@ func testCarstore(t *testing.T, dir string) carstore.CarStore {
 		t.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		t.Fatal(err)
 	}

diff --git a/testing/utils.go b/testing/utils.go
@@ -117,7 +117,7 @@ func SetupPDS(ctx context.Context, suffix string, plc plc.PLCClient) (*TestPDS,
 		return nil, err
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		return nil, err
 	}
@@ -550,7 +550,7 @@ func SetupRelay(ctx context.Context, didr plc.PLCClient) (*TestRelay, error) {
 		return nil, err
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		return nil, err
 	}