Merge branch 'develop' into bb_auctions

cmd/data-node/commands/start/node_pre.go

@@ -181,6 +181,12 @@ func (l *NodeCommand) persistentPre([]string) (err error) {
 		return fmt.Errorf("failed to apply data retention policies:%w", err)
 	}
 
+	// check that the schema version matches the latest migration, because if it doesn't queries might fail if rows/tables
+	// it expects to exist don't
+	if err := sqlstore.CheckSchemaVersionsSynced(l.Log, conf.SQLStore.ConnectionConfig, sqlstore.EmbedMigrations); err != nil {
+		return err
+	}
+
 	preLog.Info("Enabling SQL stores")
 
 	l.transactionalConnectionSource, err = sqlstore.NewTransactionalConnectionSource(l.ctx, preLog, l.conf.SQLStore.ConnectionConfig)

datanode/sqlstore/sqlstore.go

@@ -205,6 +205,44 @@ func RevertToSchemaVersionZero(log *logging.Logger, config ConnectionConfig, fs
 	return nil
 }
 
+// CheckSchemaVersionsSynced checks if if the current migrated version of the DB matches the latest available version.
+// If they are not in sync then we are running code expecting a particular version, but the rows/columns we ask for
+// might not exist.
+func CheckSchemaVersionsSynced(log *logging.Logger, config ConnectionConfig, fs fs.FS) error {
+	goose.SetBaseFS(fs)
+	goose.SetLogger(log.GooseLogger())
+	goose.SetVerbose(true)
+
+	log.Info("Checking database version matches code version")
+	poolConfig, err := config.GetPoolConfig()
+	if err != nil {
+		return fmt.Errorf("failed to get pool config: %w", err)
+	}
+
+	db := stdlib.OpenDB(*poolConfig.ConnConfig)
+	defer db.Close()
+
+	current, err := goose.GetDBVersion(db)
+	if err != nil {
+		return err
+	}
+
+	migrations, err := goose.CollectMigrations(SQLMigrationsDir, current, current+1)
+	if err != nil {
+		return err
+	}
+
+	if migrations.Len() != 0 {
+		last, err := migrations.Last()
+		if err != nil {
+			return err
+		}
+		return fmt.Errorf("Schema is version %d, latest should be %d", current, last.Version)
+	}
+
+	return nil
+}
+
 func WipeDatabaseAndMigrateSchemaToVersion(log *logging.Logger, config ConnectionConfig, version int64, fs fs.FS, verbose bool) error {
 	log = log.Named("db-wipe-migrate")
 	goose.SetBaseFS(fs)