Feat/db migration

rolling-shutter/chainobserver/db/collator/definition.go

@@ -16,7 +16,7 @@ var files embed.FS
 var Definition db.Definition
 
 func init() {
-	def, err := db.NewSQLCDefinition(files, "sql/", "chainobscollator", 1)
+	def, err := db.NewSQLCDefinition(files, "sql/", "chainobscollator")
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to initialize DB metadata")
 	}

rolling-shutter/chainobserver/db/keyper/definition.go

@@ -16,7 +16,7 @@ var files embed.FS
 var Definition db.Definition
 
 func init() {
-	def, err := db.NewSQLCDefinition(files, "sql/", "chainobskeyper", 1)
+	def, err := db.NewSQLCDefinition(files, "sql/", "chainobskeyper")
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to initialize DB metadata")
 	}

rolling-shutter/chainobserver/db/sync/definition.go

@@ -16,7 +16,7 @@ var Definition db.Definition
 
 func init() {
 	var err error
-	Definition, err = db.NewSQLCDefinition(files, "sql/", "chainobssync", 1)
+	Definition, err = db.NewSQLCDefinition(files, "sql/", "chainobssync")
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to initialize DB metadata")
 	}

rolling-shutter/keyper/database/definition.go

@@ -21,7 +21,7 @@ var files embed.FS
 var Definition db.Definition
 
 func init() {
-	sqlcDB, err := db.NewSQLCDefinition(files, "sql/", "keyper", 1)
+	sqlcDB, err := db.NewSQLCDefinition(files, "sql/", "keyper")
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to initialize DB")
 	}

rolling-shutter/keyperimpl/gnosis/database/definition.go

@@ -17,7 +17,7 @@ var files embed.FS
 var Definition db.Definition
 
 func init() {
-	def, err := db.NewSQLCDefinition(files, "sql/", "gnosiskeyper", 1)
+	def, err := db.NewSQLCDefinition(files, "sql/", "gnosiskeyper")
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to initialize DB metadata")
 	}

rolling-shutter/keyperimpl/gnosis/database/sql/migrations/V2_validatorRegistrations.sql

@@ -0,0 +1,6 @@
+-- schema-version: gnosiskeyper-2 --
+-- migrations need to start from V2... as file name, as the V1 was initial schema
+
+ALTER TABLE validator_registrations
+ DROP CONSTRAINT validator_registrations_pkey,
+ ADD PRIMARY KEY (block_number, tx_index, log_index, validator_index);

rolling-shutter/keyperimpl/gnosis/database/sql/schemas/gnosiskeyper.sql

@@ -66,4 +66,5 @@ CREATE TABLE validator_registrations_synced_until(
     enforce_one_row bool PRIMARY KEY DEFAULT true,
     block_hash bytea NOT NULL,
     block_number bigint NOT NULL CHECK (block_number >= 0)
-);
+);
+

rolling-shutter/keyperimpl/shutterservice/database/definition.go

@@ -17,7 +17,7 @@ var files embed.FS
 var Definition db.Definition
 
 func init() {
-	def, err := db.NewSQLCDefinition(files, "sql/", "shutterservicekeyper", 1)
+	def, err := db.NewSQLCDefinition(files, "sql/", "shutterservicekeyper")
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to initialize DB metadata")
 	}

rolling-shutter/medley/db/definition.go

@@ -14,7 +14,7 @@ var metaDefinition Definition
 
 func init() {
 	var err error
-	metaDefinition, err = NewSQLCDefinition(files, "sql/", "meta", 1)
+	metaDefinition, err = NewSQLCDefinition(files, "sql/", "meta")
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to initialize DB metadata")
 	}

rolling-shutter/medley/db/definitions.go

@@ -17,14 +17,23 @@ import (
 // pins the database to a specific role, e.g. "keyper-test" or "snapshot-keyper-production"
 // in order to prevent later usage of the database with commands that fulfill a different role.
 func InitDB(ctx context.Context, dbpool *pgxpool.Pool, role string, definition Definition) error {
+	// First check if schema exists and is valid
 	err := dbpool.BeginFunc(WrapContext(ctx, definition.Validate))
 	if err == nil {
 		shdb.AddConnectionInfo(log.Info(), dbpool).Msg("database already exists")
 		return nil
+	} else if errors.Is(err, ErrNeedsMigration) {
+		// Schema exists, just run migrations
+		shdb.AddConnectionInfo(log.Info(), dbpool).Msg("database exists, checking for migrations")
+		err = dbpool.BeginFunc(WrapContext(ctx, definition.Migrate))
+		if err != nil {
+			return errors.Wrap(err, "failed to apply migrations")
+		}
+		return nil
 	} else if errors.Is(err, ErrValueMismatch) {
 		return err
 	}
-
+	// Schema doesn't exist or is invalid, create it
 	err = dbpool.BeginFunc(WrapContext(ctx, definition.Create))
 	if err != nil {
 		return err
@@ -35,19 +44,23 @@ func InitDB(ctx context.Context, dbpool *pgxpool.Pool, role string, definition D
 		return err
 	}
 
+	// Run any migrations after initial creation
+	err = dbpool.BeginFunc(WrapContext(ctx, definition.Migrate))
+	if err != nil {
+		return errors.Wrap(err, "failed to apply migrations")
+	}
+
 	// For the outer DB initialisation, also set the database version to
 	// the overall "role", so that e.g. a snapshot keyper database won't be
 	// used by another keyper implementation, no matter if the schemas
 	// are compatible
-	err = dbpool.BeginFunc(
-		ctx,
-		func(tx pgx.Tx) error {
-			return InsertDBVersion(ctx, tx, role)
-		},
-	)
+	err = dbpool.BeginFunc(ctx, func(tx pgx.Tx) error {
+		return InsertDBVersion(ctx, tx, role)
+	})
 	if err != nil {
 		return err
 	}
+
 	shdb.AddConnectionInfo(log.Info(), dbpool).Msg("database initialized")
 	return nil
 }
@@ -120,17 +133,27 @@ func (d AggregateDefinition) Validate(ctx context.Context, tx pgx.Tx) error {
 	return nil
 }
 
+func (d AggregateDefinition) Migrate(ctx context.Context, tx pgx.Tx) error {
+	for def := range d.defs {
+		err := def.Migrate(ctx, tx)
+		if err != nil {
+			return errors.Wrapf(err, "migration failed for definition '%s'", def.Name())
+		}
+	}
+	return nil
+}
+
 type Definition interface {
 	Name() string
 	Create(context.Context, pgx.Tx) error
 	Init(context.Context, pgx.Tx) error
 	Validate(context.Context, pgx.Tx) error
+	Migrate(context.Context, pgx.Tx) error
 }
 
 type Schema struct {
-	Version int
-	Name    string
-	Path    string
+	Name string
+	Path string
 }
 
 type Migration struct {

rolling-shutter/medley/db/metadb.go

@@ -3,15 +3,17 @@ package db
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	"github.com/jackc/pgx/v4"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
 )
 
 var (
-	ErrValueMismatch = errors.New("database has unexpected value")
-	ErrKeyNotFound   = errors.New("key does not exist")
+	ErrValueMismatch  = errors.New("database has unexpected value")
+	ErrKeyNotFound    = errors.New("key does not exist")
+	ErrNeedsMigration = errors.New("needs migration")
 )
 
 var DatabaseVersionKey string = "database-version"
@@ -29,8 +31,8 @@ func InsertDBVersion(ctx context.Context, tx pgx.Tx, version string) error {
 	return insertMetaInf(ctx, tx, DatabaseVersionKey, version)
 }
 
-func InsertSchemaVersion(ctx context.Context, tx pgx.Tx, definitionName string, schema Schema) error {
-	return insertMetaInf(ctx, tx, MakeSchemaVersionKey(definitionName, schema.Name), fmt.Sprint(schema.Version))
+func InsertSchemaVersion(ctx context.Context, tx pgx.Tx, definitionName string, schema Schema, version int) error {
+	return insertMetaInf(ctx, tx, MakeSchemaVersionKey(definitionName, schema.Name), fmt.Sprint(version))
 }
 
 func insertMetaInf(ctx context.Context, tx pgx.Tx, key, val string) error {
@@ -42,9 +44,26 @@ func insertMetaInf(ctx context.Context, tx pgx.Tx, key, val string) error {
 	})
 }
 
+func UpdateSchemaVersion(ctx context.Context, tx pgx.Tx, defName string, schema Schema, version int) error {
+	return New(tx).UpdateMeta(ctx, UpdateMetaParams{
+		Key:   MakeSchemaVersionKey(defName, schema.Name),
+		Value: fmt.Sprint(version),
+	})
+}
+
 // ValidateSchemaVersion checks that the database schema is compatible.
-func ValidateSchemaVersion(ctx context.Context, tx pgx.Tx, definitionName string, schema Schema) error {
-	return expectMetaKeyVal(ctx, tx, MakeSchemaVersionKey(definitionName, schema.Name), fmt.Sprint(schema.Version))
+func ValidateSchemaVersion(ctx context.Context, tx pgx.Tx, definitionName string, schema Schema, version int) error {
+	haveVersion, err := GetSchemaVersion(ctx, tx, definitionName, schema)
+	if err != nil {
+		return err
+	}
+	if haveVersion < version {
+		return errors.Wrapf(ErrNeedsMigration, "expected version %d, have %d", version, haveVersion)
+	}
+	if haveVersion != version {
+		return errors.Wrapf(ErrValueMismatch, "expected version %d, have %d", version, haveVersion)
+	}
+	return nil
 }
 
 func expectMetaKeyVal(ctx context.Context, tx pgx.Tx, key, val string) error {
@@ -67,3 +86,18 @@ func expectMetaKeyVal(ctx context.Context, tx pgx.Tx, key, val string) error {
 func ValidateDatabaseVersion(ctx context.Context, tx pgx.Tx, version string) error {
 	return expectMetaKeyVal(ctx, tx, DatabaseVersionKey, version)
 }
+
+func GetSchemaVersion(ctx context.Context, tx pgx.Tx, definitionName string, schema Schema) (int, error) {
+	key := MakeSchemaVersionKey(definitionName, schema.Name)
+	haveVal, err := New(tx).GetMeta(ctx, key)
+	if err == pgx.ErrNoRows {
+		return 0, errors.Wrapf(ErrKeyNotFound, "key: %s", key)
+	} else if err != nil {
+		return 0, errors.Wrapf(err, "failed to get key '%s' from meta_inf table", key)
+	}
+	version, err := strconv.ParseInt(haveVal, 10, 0)
+	if err != nil {
+		return 0, errors.Wrapf(err, "failed to convert version '%s' from meta_inf table", key)
+	}
+	return int(version), nil
+}

rolling-shutter/medley/db/sql/queries/meta.sql

@@ -3,3 +3,6 @@ INSERT INTO meta_inf (key, value) VALUES ($1, $2);
 
 -- name: GetMeta :one
 SELECT value FROM meta_inf WHERE key = $1;
+
+-- name: UpdateMeta :exec
+UPDATE meta_inf SET value = $1 WHERE key = $2;