Merge pull request #586 from shutter-network/feat/db-migration

blockchainluffy · web-flow · commit 9967cb719cc6 · 2025-08-08T16:44:38.000+05:30
Feat/db migration
diff --git a/rolling-shutter/chainobserver/db/collator/definition.go b/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")
 	}
diff --git a/rolling-shutter/chainobserver/db/keyper/definition.go b/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")
 	}
diff --git a/rolling-shutter/chainobserver/db/sync/definition.go b/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")
 	}
diff --git a/rolling-shutter/keyper/database/definition.go b/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")
 	}
diff --git a/rolling-shutter/keyperimpl/gnosis/config.go b/rolling-shutter/keyperimpl/gnosis/config.go
@@ -100,30 +100,32 @@ func (c *Config) GetAddress() common.Address {
 }
 
 type GnosisConfig struct {
-	Node                     *configuration.EthnodeConfig `shconfig:",required"`
-	Contracts                *GnosisContractsConfig       `shconfig:",required"`
-	EncryptedGasLimit        uint64                       `shconfig:",required"`
-	MinGasPerTransaction     uint64                       `shconfig:",required"`
-	MaxTxPointerAge          uint64                       `shconfig:",required"`
-	SecondsPerSlot           uint64                       `shconfig:",required"`
-	SlotsPerEpoch            uint64                       `shconfig:",required"`
-	GenesisSlotTimestamp     uint64                       `shconfig:",required"`
-	SyncStartBlockNumber     uint64                       `shconfig:",required"`
-	SyncMonitorCheckInterval uint64                       `shconfig:",required"`
+	Node                                   *configuration.EthnodeConfig `shconfig:",required"`
+	Contracts                              *GnosisContractsConfig       `shconfig:",required"`
+	EncryptedGasLimit                      uint64                       `shconfig:",required"`
+	MinGasPerTransaction                   uint64                       `shconfig:",required"`
+	MaxTxPointerAge                        uint64                       `shconfig:",required"`
+	SecondsPerSlot                         uint64                       `shconfig:",required"`
+	SlotsPerEpoch                          uint64                       `shconfig:",required"`
+	GenesisSlotTimestamp                   uint64                       `shconfig:",required"`
+	SyncStartBlockNumber                   uint64                       `shconfig:",required"`
+	SyncMonitorCheckInterval               uint64                       `shconfig:",required"`
+	EnableAggregateValidatorRegistrationV1 bool                         `shconfig:",required"`
 }
 
 func NewGnosisConfig() *GnosisConfig {
 	c := &GnosisConfig{
-		Node:                     configuration.NewEthnodeConfig(),
-		Contracts:                NewGnosisContractsConfig(),
-		EncryptedGasLimit:        0,
-		MinGasPerTransaction:     0,
-		MaxTxPointerAge:          0,
-		SecondsPerSlot:           0,
-		SlotsPerEpoch:            0,
-		GenesisSlotTimestamp:     0,
-		SyncStartBlockNumber:     0,
-		SyncMonitorCheckInterval: 0,
+		Node:                                   configuration.NewEthnodeConfig(),
+		Contracts:                              NewGnosisContractsConfig(),
+		EncryptedGasLimit:                      0,
+		MinGasPerTransaction:                   0,
+		MaxTxPointerAge:                        0,
+		SecondsPerSlot:                         0,
+		SlotsPerEpoch:                          0,
+		GenesisSlotTimestamp:                   0,
+		SyncStartBlockNumber:                   0,
+		SyncMonitorCheckInterval:               0,
+		EnableAggregateValidatorRegistrationV1: false,
 	}
 	c.Init()
 	return c
@@ -161,6 +163,7 @@ func (c *GnosisConfig) SetExampleValues() error {
 	c.GenesisSlotTimestamp = 1665410700
 	c.SyncStartBlockNumber = 0
 	c.SyncMonitorCheckInterval = 30
+	c.EnableAggregateValidatorRegistrationV1 = false
 	return nil
 }
 
diff --git a/rolling-shutter/keyperimpl/gnosis/database/definition.go b/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")
 	}
diff --git a/rolling-shutter/keyperimpl/gnosis/database/sql/migrations/V2_validatorRegistrations.sql b/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);
diff --git a/rolling-shutter/keyperimpl/gnosis/database/sql/schemas/gnosiskeyper.sql b/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)
-);
+);
+
diff --git a/rolling-shutter/keyperimpl/gnosis/keyper.go b/rolling-shutter/keyperimpl/gnosis/keyper.go
@@ -248,12 +248,13 @@ func (kpr *Keyper) initValidatorSyncer(ctx context.Context) error {
 		return errors.Wrap(err, "failed to instantiate validator registry contract")
 	}
 	kpr.validatorSyncer = &ValidatorSyncer{
-		Contract:             validatorRegistryContract,
-		DBPool:               kpr.dbpool,
-		BeaconAPIClient:      kpr.beaconAPIClient,
-		ExecutionClient:      validatorSyncerClient,
-		ChainID:              chainID.Uint64(),
-		SyncStartBlockNumber: kpr.config.Gnosis.SyncStartBlockNumber,
+		Contract:                               validatorRegistryContract,
+		DBPool:                                 kpr.dbpool,
+		BeaconAPIClient:                        kpr.beaconAPIClient,
+		ExecutionClient:                        validatorSyncerClient,
+		ChainID:                                chainID.Uint64(),
+		SyncStartBlockNumber:                   kpr.config.Gnosis.SyncStartBlockNumber,
+		EnableAggregateValidatorRegistrationV1: kpr.config.Gnosis.EnableAggregateValidatorRegistrationV1,
 	}
 
 	// Perform an initial sync now because it might take some time and doing so during regular
diff --git a/rolling-shutter/keyperimpl/gnosis/validatorsyncer.go b/rolling-shutter/keyperimpl/gnosis/validatorsyncer.go
@@ -29,12 +29,13 @@ const (
 )
 
 type ValidatorSyncer struct {
-	Contract             *validatorRegistryBindings.Validatorregistry
-	DBPool               *pgxpool.Pool
-	BeaconAPIClient      *beaconapiclient.Client
-	ExecutionClient      *ethclient.Client
-	ChainID              uint64
-	SyncStartBlockNumber uint64
+	Contract                               *validatorRegistryBindings.Validatorregistry
+	DBPool                                 *pgxpool.Pool
+	BeaconAPIClient                        *beaconapiclient.Client
+	ExecutionClient                        *ethclient.Client
+	ChainID                                uint64
+	SyncStartBlockNumber                   uint64
+	EnableAggregateValidatorRegistrationV1 bool
 }
 
 func (v *ValidatorSyncer) Sync(ctx context.Context, header *types.Header) error {
@@ -226,8 +227,12 @@ func (v *ValidatorSyncer) filterEvents(
 				evLog.Warn().Msg("ignoring registration message with invalid signature")
 				continue
 			}
+		} else if v.EnableAggregateValidatorRegistrationV1 && msg.Version == validatorregistry.AggregateValidatorRegistrationMessageVersion {
+			if !validatorregistry.VerifyAggregateSignature(sig, pubKeys, msg) {
+				evLog.Warn().Msg("ignoring validator registration message with invalid signature")
+				continue
+			}
 		} else {
-			// TODO: this disables aggregate message
 			evLog.Warn().Msg("ignoring validator registration message as the version is not compatible")
 			continue
 		}
diff --git a/rolling-shutter/keyperimpl/shutterservice/database/definition.go b/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")
 	}
diff --git a/rolling-shutter/medley/db/definition.go b/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")
 	}
diff --git a/rolling-shutter/medley/db/definitions.go b/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 {
diff --git a/rolling-shutter/medley/db/meta.sqlc.gen.go b/rolling-shutter/medley/db/meta.sqlc.gen.go
diff --git a/rolling-shutter/medley/db/metadb.go b/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
+}
diff --git a/rolling-shutter/medley/db/sql/queries/meta.sql b/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;
diff --git a/rolling-shutter/medley/db/sqlc.go b/rolling-shutter/medley/db/sqlc.go
diff --git a/rolling-shutter/migration-guide-dev.md b/rolling-shutter/migration-guide-dev.md
diff --git a/rolling-shutter/snapshot/database/definition.go b/rolling-shutter/snapshot/database/definition.go

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,7 @@ var files embed.FS`
`16`	`16`	`var Definition db.Definition`
`17`	`17`
`18`	`18`	`func init() {`
`19`		`- def, err := db.NewSQLCDefinition(files, "sql/", "chainobscollator", 1)`
	`19`	`+ def, err := db.NewSQLCDefinition(files, "sql/", "chainobscollator")`
`20`	`20`	`if err != nil {`
`21`	`21`	`log.Fatal().Err(err).Msg("failed to initialize DB metadata")`
`22`	`22`	`}`
Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,7 @@ var Definition db.Definition`
`16`	`16`
`17`	`17`	`func init() {`
`18`	`18`	`var err error`
`19`		`- Definition, err = db.NewSQLCDefinition(files, "sql/", "chainobssync", 1)`
	`19`	`+ Definition, err = db.NewSQLCDefinition(files, "sql/", "chainobssync")`
`20`	`20`	`if err != nil {`
`21`	`21`	`log.Fatal().Err(err).Msg("failed to initialize DB metadata")`
`22`	`22`	`}`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ var files embed.FS`
`21`	`21`	`var Definition db.Definition`
`22`	`22`
`23`	`23`	`func init() {`
`24`		`- sqlcDB, err := db.NewSQLCDefinition(files, "sql/", "keyper", 1)`
	`24`	`+ sqlcDB, err := db.NewSQLCDefinition(files, "sql/", "keyper")`
`25`	`25`	`if err != nil {`
`26`	`26`	`log.Fatal().Err(err).Msg("failed to initialize DB")`
`27`	`27`	`}`
Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@ var files embed.FS`
`17`	`17`	`var Definition db.Definition`
`18`	`18`
`19`	`19`	`func init() {`
`20`		`- def, err := db.NewSQLCDefinition(files, "sql/", "gnosiskeyper", 1)`
	`20`	`+ def, err := db.NewSQLCDefinition(files, "sql/", "gnosiskeyper")`
`21`	`21`	`if err != nil {`
`22`	`22`	`log.Fatal().Err(err).Msg("failed to initialize DB metadata")`
`23`	`23`	`}`