Merge pull request #343 from spacemeshos/324-keep-challenges-db-in-se…

…parate-dir

Add option to store DBs in a different location

config/config.go

@@ -20,6 +20,7 @@ import (
 )
 
 const (
+	defaultDbDirName                = "db"
 	defaultDataDirname              = "data"
 	defaultLogDirname               = "logs"
 	defaultMaxLogFiles              = 3
@@ -45,7 +46,8 @@ var (
 type Config struct {
 	PoetDir         string  `long:"poetdir"        description:"The base directory that contains poet's data, logs, configuration file, etc."`
 	ConfigFile      string  `long:"configfile"     description:"Path to configuration file"                                                   short:"c"`
-	DataDir         string  `long:"datadir"        description:"The directory to store poet's data within"                                    short:"b"`
+	DataDir         string  `long:"datadir"        description:"The directory to store poet's data within."                                   short:"b"`
+	DbDir           string  `long:"dbdir"          description:"The directory to store DBs within"`
 	LogDir          string  `long:"logdir"         description:"Directory to log output."`
 	DebugLog        bool    `long:"debuglog"       description:"Enable debug logs"`
 	JSONLog         bool    `long:"jsonlog"        description:"Whether to log in JSON format"`
@@ -72,6 +74,7 @@ func DefaultConfig() *Config {
 	return &Config{
 		PoetDir:         poetDir,
 		DataDir:         filepath.Join(poetDir, defaultDataDirname),
+		DbDir:           filepath.Join(poetDir, defaultDbDirName),
 		LogDir:          filepath.Join(poetDir, defaultLogDirname),
 		MaxLogFiles:     defaultMaxLogFiles,
 		MaxLogFileSize:  defaultMaxLogFileSize,
@@ -125,6 +128,9 @@ func SetupConfig(cfg *Config) (*Config, error) {
 		if cfg.LogDir == defaultCfg.LogDir {
 			cfg.LogDir = filepath.Join(cfg.PoetDir, defaultLogDirname)
 		}
+		if cfg.DbDir == defaultCfg.DbDir {
+			cfg.DbDir = filepath.Join(cfg.PoetDir, defaultDbDirName)
+		}
 	}
 
 	// Create the poet directory if it doesn't already exist.

db/migrate.go

@@ -0,0 +1,67 @@
+package db
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/syndtr/goleveldb/leveldb"
+	"go.uber.org/zap"
+
+	"github.com/spacemeshos/poet/logging"
+)
+
+// Temporary code to migrate a database to a new location.
+// It opens both DBs and copies all the data from the old DB to the new one.
+func Migrate(ctx context.Context, targetDbDir, oldDbDir string) error {
+	log := logging.FromContext(ctx).With(zap.String("oldDbDir", oldDbDir), zap.String("targetDbDir", targetDbDir))
+	if oldDbDir == targetDbDir {
+		log.Debug("skipping in-place DB migration")
+		return nil
+	}
+	if _, err := os.Stat(oldDbDir); os.IsNotExist(err) {
+		log.Debug("skipping DB migration - old DB doesn't exist")
+		return nil
+	}
+	log.Info("migrating DB location")
+
+	oldDb, err := leveldb.OpenFile(oldDbDir, nil)
+	if err != nil {
+		return fmt.Errorf("opening old DB: %w", err)
+	}
+	defer oldDb.Close()
+
+	targetDb, err := leveldb.OpenFile(targetDbDir, nil)
+	if err != nil {
+		return fmt.Errorf("opening target DB: %w", err)
+	}
+	defer targetDb.Close()
+
+	trans, err := targetDb.OpenTransaction()
+	if err != nil {
+		return fmt.Errorf("opening new DB transaction: %w", err)
+	}
+	iter := oldDb.NewIterator(nil, nil)
+	defer iter.Release()
+	for iter.Next() {
+		if err := trans.Put(iter.Key(), iter.Value(), nil); err != nil {
+			trans.Discard()
+			return fmt.Errorf("migrating key %X: %w", iter.Key(), err)
+		}
+	}
+	iter.Release()
+	if err := trans.Commit(); err != nil {
+		return fmt.Errorf("committing DB transaction: %w", err)
+	}
+
+	// Remove old DB
+	log.Info("removing the old DB")
+	if err := oldDb.Close(); err != nil {
+		return fmt.Errorf("closing old DB: %w", err)
+	}
+	if err := os.RemoveAll(oldDbDir); err != nil {
+		return fmt.Errorf("removing old DB: %w", err)
+	}
+	log.Info("DB migrated to new location")
+	return nil
+}

db/migrate_test.go

@@ -0,0 +1,79 @@
+package db_test
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/syndtr/goleveldb/leveldb"
+
+	"github.com/spacemeshos/poet/db"
+)
+
+var kvs = map[string][]byte{
+	"key":  []byte("value"),
+	"key2": []byte("value2"),
+	"key3": []byte("value3"),
+}
+
+func TestMigrateDb(t *testing.T) {
+	// open a database and write some data
+	oldDbPath := t.TempDir()
+	oldDb, err := leveldb.OpenFile(oldDbPath, nil)
+	require.NoError(t, err)
+	defer oldDb.Close()
+	for k, v := range kvs {
+		require.NoError(t, oldDb.Put([]byte(k), v, nil))
+	}
+	oldDb.Close()
+
+	// migrate the database
+	newDbPath := t.TempDir()
+	require.NoError(t, db.Migrate(context.Background(), newDbPath, oldDbPath))
+
+	// open the new database and check that the data was copied
+	newDb, err := leveldb.OpenFile(newDbPath, nil)
+	require.NoError(t, err)
+	defer newDb.Close()
+
+	for k, v := range kvs {
+		value, err := newDb.Get([]byte(k), nil)
+		require.NoError(t, err)
+		require.Equal(t, v, value)
+	}
+
+	// old DB should be removed
+	_, err = os.Stat(oldDbPath)
+	require.ErrorIs(t, err, os.ErrNotExist)
+}
+
+func TestSkipMigrateInPlace(t *testing.T) {
+	// open a database and write some data
+	dbPath := t.TempDir()
+	database, err := leveldb.OpenFile(dbPath, nil)
+	require.NoError(t, err)
+	defer database.Close()
+	for k, v := range kvs {
+		require.NoError(t, database.Put([]byte(k), v, nil))
+	}
+	database.Close()
+
+	// migrate the database
+	require.NoError(t, db.Migrate(context.Background(), dbPath, dbPath))
+
+	// open the new database and check that the data was copied
+	database, err = leveldb.OpenFile(dbPath, nil)
+	require.NoError(t, err)
+	defer database.Close()
+
+	for k, v := range kvs {
+		value, err := database.Get([]byte(k), nil)
+		require.NoError(t, err)
+		require.Equal(t, v, value)
+	}
+}
+
+func TestSkipMigrateSrcDoesntExist(t *testing.T) {
+	require.NoError(t, db.Migrate(context.Background(), t.TempDir(), "i-dont-exist"))
+}

migrations/00_dbdir.go

@@ -0,0 +1,67 @@
+package migrations
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+
+	"go.uber.org/zap"
+
+	"github.com/spacemeshos/poet/config"
+	"github.com/spacemeshos/poet/db"
+	"github.com/spacemeshos/poet/logging"
+)
+
+func migrateDbDir(ctx context.Context, cfg *config.Config) error {
+	if err := migrateProofsDb(ctx, cfg); err != nil {
+		return fmt.Errorf("migrating proofs DB: %w", err)
+	}
+	if err := migrateRoundsDbs(ctx, cfg); err != nil {
+		return fmt.Errorf("migrating rounds DBs: %w", err)
+	}
+
+	return nil
+}
+
+func migrateRoundsDbs(ctx context.Context, cfg *config.Config) error {
+	roundsDataDir := filepath.Join(cfg.DataDir, "rounds")
+	// check if dir exists
+	if _, err := os.Stat(roundsDataDir); os.IsNotExist(err) {
+		return nil
+	}
+
+	logger := logging.FromContext(ctx)
+	logger.Info("migrating rounds DBs", zap.String("datadir", cfg.DataDir))
+	entries, err := os.ReadDir(roundsDataDir)
+	if err != nil {
+		return err
+	}
+
+	for _, entry := range entries {
+		if !entry.IsDir() {
+			continue
+		}
+
+		if _, err := strconv.ParseUint(entry.Name(), 10, 32); err != nil {
+			return fmt.Errorf("entry is not a uint32 %s", entry.Name())
+		}
+
+		dbdir := filepath.Join(cfg.DbDir, "rounds", entry.Name())
+		oldDbDir := filepath.Join(roundsDataDir, entry.Name(), "challengesDb")
+		if err := db.Migrate(ctx, dbdir, oldDbDir); err != nil {
+			return fmt.Errorf("migrating round DB from %s: %w", oldDbDir, err)
+		}
+	}
+	return nil
+}
+
+func migrateProofsDb(ctx context.Context, cfg *config.Config) error {
+	proofsDbPath := filepath.Join(cfg.DbDir, "proofs")
+	oldProofsDbPath := filepath.Join(cfg.DataDir, "proofs")
+	if err := db.Migrate(ctx, proofsDbPath, oldProofsDbPath); err != nil {
+		return fmt.Errorf("migrating proofs DB %s -> %s: %w", oldProofsDbPath, proofsDbPath, err)
+	}
+	return nil
+}

migrations/00_dbdir_test.go

@@ -0,0 +1,58 @@
+package migrations
+
+import (
+	"context"
+	"path/filepath"
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/syndtr/goleveldb/leveldb"
+
+	"github.com/spacemeshos/poet/config"
+)
+
+func TestMigrateRoundsDb(t *testing.T) {
+	// Prepare
+	cfg := config.Config{
+		DataDir: t.TempDir(),
+		DbDir:   t.TempDir(),
+	}
+	for i := 0; i < 5; i++ {
+		id := strconv.Itoa(i)
+		oldDb, err := leveldb.OpenFile(filepath.Join(cfg.DataDir, "rounds", id, "challengesDb"), nil)
+		require.NoError(t, err)
+		defer oldDb.Close()
+		require.NoError(t, oldDb.Put([]byte(id+"key"), []byte("value"), nil))
+		require.NoError(t, oldDb.Put([]byte(id+"key2"), []byte("value2"), nil))
+		oldDb.Close()
+	}
+	// Act
+	require.NoError(t, migrateRoundsDbs(context.Background(), &cfg))
+
+	// Verify
+	for i := 0; i < 5; i++ {
+		id := strconv.Itoa(i)
+		db, err := leveldb.OpenFile(filepath.Join(cfg.DbDir, "rounds", id), nil)
+		require.NoError(t, err)
+		defer db.Close()
+
+		v, err := db.Get([]byte(id+"key"), nil)
+		require.NoError(t, err)
+		require.Equal(t, []byte("value"), v)
+
+		v, err = db.Get([]byte(id+"key2"), nil)
+		require.NoError(t, err)
+		require.Equal(t, []byte("value2"), v)
+
+		db.Close()
+	}
+}
+
+func TestMigrateRoundsDb_NothingToMigrate(t *testing.T) {
+	cfg := config.Config{
+		DataDir: t.TempDir(),
+		DbDir:   t.TempDir(),
+	}
+	require.NoError(t, migrateRoundsDbs(context.Background(), &cfg))
+}

migrations/migrations.go

@@ -0,0 +1,16 @@
+package migrations
+
+import (
+	"context"
+
+	"github.com/spacemeshos/poet/config"
+	"github.com/spacemeshos/poet/logging"
+)
+
+func Migrate(ctx context.Context, cfg *config.Config) error {
+	ctx = logging.NewContext(ctx, logging.FromContext(ctx).Named("migrations"))
+	if err := migrateDbDir(ctx, cfg); err != nil {
+		return err
+	}
+	return nil
+}

migrations/migrations_test.go

@@ -0,0 +1,19 @@
+package migrations_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/spacemeshos/poet/config"
+	"github.com/spacemeshos/poet/migrations"
+)
+
+func TestMigrate(t *testing.T) {
+	cfg := config.DefaultConfig()
+	cfg.PoetDir = t.TempDir()
+	cfg.DataDir = t.TempDir()
+	cfg.DbDir = t.TempDir()
+	require.NoError(t, migrations.Migrate(context.Background(), cfg))
+}

poet.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/spacemeshos/poet/config"
 	"github.com/spacemeshos/poet/logging"
+	"github.com/spacemeshos/poet/migrations"
 	"github.com/spacemeshos/poet/server"
 )
 
@@ -72,6 +73,11 @@ func poetMain() (err error) {
 	logger.Sugar().
 		Infof("version: %s, dir: %v, datadir: %v, genesis: %v", version, cfg.PoetDir, cfg.DataDir, cfg.Service.Genesis)
 
+	// Migrate data if needed
+	if err := migrations.Migrate(ctx, cfg); err != nil {
+		return fmt.Errorf("migrations failed: %w", err)
+	}
+
 	if cfg.MetricsPort != nil {
 		// Start Prometheus
 		lis, err := net.Listen("tcp", fmt.Sprintf(":%v", *cfg.MetricsPort))

server/server.go

@@ -64,7 +64,7 @@ func New(ctx context.Context, cfg config.Config) (*Server, error) {
 		}
 	}
 
-	svc, err := service.NewService(ctx, cfg.Service, cfg.DataDir)
+	svc, err := service.NewService(ctx, cfg.Service, cfg.DbDir, cfg.DataDir)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create Service: %v", err)
 	}
@@ -120,7 +120,7 @@ func (s *Server) Start(ctx context.Context) error {
 		}),
 	}
 
-	proofsDbPath := filepath.Join(s.cfg.DataDir, "proofs")
+	proofsDbPath := filepath.Join(s.cfg.DbDir, "proofs")
 	proofsDb, err := service.NewProofsDatabase(proofsDbPath, s.svc.ProofsChan())
 	if err != nil {
 		return fmt.Errorf("failed to create proofs DB: %w", err)