Change detection of already executed migrations (#2665)

This changes the detection of already executed migrations for the
roomserver state block and keychange refactor. It now uses schema tables
provided by the database engine to check if the column was already
removed. We now also store the migration in the migrations table.

This should stop e.g. Postgres from logging errors like `ERROR: column
"event_nid" does not exist at character 8`.

internal/sqlutil/migrate.go

@@ -21,8 +21,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/matrix-org/dendrite/internal"
 	"github.com/sirupsen/logrus"
+
+	"github.com/matrix-org/dendrite/internal"
 )
 
 const createDBMigrationsSQL = "" +
@@ -95,11 +96,11 @@ func (m *Migrator) Up(ctx context.Context) error {
 		for i := range m.migrations {
 			now := time.Now().UTC().Format(time.RFC3339)
 			migration := m.migrations[i]
-			logrus.Debugf("Executing database migration '%s'", migration.Version)
 			// Skip migration if it was already executed
 			if _, ok := executedMigrations[migration.Version]; ok {
 				continue
 			}
+			logrus.Debugf("Executing database migration '%s'", migration.Version)
 			err = migration.Up(ctx, txn)
 			if err != nil {
 				return fmt.Errorf("unable to execute migration '%s': %w", migration.Version, err)
@@ -140,3 +141,19 @@ func (m *Migrator) ExecutedMigrations(ctx context.Context) (map[string]struct{},
 
 	return result, rows.Err()
 }
+
+// InsertMigration creates the migrations table if it doesn't exist and
+// inserts a migration given their name to the database.
+// This should only be used when manually inserting migrations.
+func InsertMigration(ctx context.Context, db *sql.DB, migrationName string) error {
+	_, err := db.ExecContext(ctx, createDBMigrationsSQL)
+	if err != nil {
+		return fmt.Errorf("unable to create db_migrations: %w", err)
+	}
+	_, err = db.ExecContext(ctx, insertVersionSQL,
+		migrationName,
+		time.Now().Format(time.RFC3339),
+		internal.VersionString(),
+	)
+	return err
+}

keyserver/storage/postgres/key_changes_table.go

@@ -17,8 +17,9 @@ package postgres
 import (
 	"context"
 	"database/sql"
+	"errors"
 
-	"github.com/lib/pq"
+	"github.com/sirupsen/logrus"
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
@@ -63,30 +64,37 @@ func NewPostgresKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
 		return s, err
 	}
 
+	if err = executeMigration(context.Background(), db); err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func executeMigration(ctx context.Context, db *sql.DB) error {
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column partition was removed from the table
-	var count int
+	migrationName := "keyserver: refactor key changes"
-	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan(&count)
+
-	if err == nil {
+	var cName string
-		m := sqlutil.NewMigrator(db)
+	err := db.QueryRowContext(ctx, "select column_name from information_schema.columns where table_name = 'keyserver_key_changes' AND column_name = 'partition'").Scan(&cName)
-		m.AddMigrations(sqlutil.Migration{
+	if err != nil {
-			Version: "keyserver: refactor key changes",
+		if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
-			Up:      deltas.UpRefactorKeyChanges,
+			if err = sqlutil.InsertMigration(ctx, db, migrationName); err != nil {
-		})
+				// not a fatal error, log and continue
-		return s, m.Up(context.Background())
+				logrus.WithError(err).Warnf("unable to manually insert migration '%s'", migrationName)
-	} else {
-		switch e := err.(type) {
-		case *pq.Error:
-			// ignore undefined_column (42703) errors, as this is expected at this point
-			if e.Code != "42703" {
-				return nil, err
 			}
-		default:
+			return nil
-			return nil, err
 		}
+		return err
 	}
-	return s, nil
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: migrationName,
+		Up:      deltas.UpRefactorKeyChanges,
+	})
+
+	return m.Up(ctx)
 }
 
 func (s *keyChangesStatements) Prepare() (err error) {

keyserver/storage/sqlite3/key_changes_table.go

@@ -17,6 +17,9 @@ package sqlite3
 import (
 	"context"
 	"database/sql"
+	"errors"
+
+	"github.com/sirupsen/logrus"
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
@@ -58,23 +61,40 @@ func NewSqliteKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
 	if err != nil {
 		return s, err
 	}
-	// TODO: Remove when we are sure we are not having goose artefacts in the db
+
-	// This forces an error, which indicates the migration is already applied, since the
+	if err = executeMigration(context.Background(), db); err != nil {
-	// column partition was removed from the table
+		return nil, err
-	var count int
-	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan(&count)
-	if err == nil {
-		m := sqlutil.NewMigrator(db)
-		m.AddMigrations(sqlutil.Migration{
-			Version: "keyserver: refactor key changes",
-			Up:      deltas.UpRefactorKeyChanges,
-		})
-		return s, m.Up(context.Background())
 	}
 
 	return s, nil
 }
 
+func executeMigration(ctx context.Context, db *sql.DB) error {
+	// TODO: Remove when we are sure we are not having goose artefacts in the db
+	// This forces an error, which indicates the migration is already applied, since the
+	// column partition was removed from the table
+	migrationName := "keyserver: refactor key changes"
+
+	var cName string
+	err := db.QueryRowContext(ctx, `SELECT p.name FROM sqlite_master AS m JOIN pragma_table_info(m.name) AS p WHERE m.name = 'keyserver_key_changes' AND p.name = 'partition'`).Scan(&cName)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
+			if err = sqlutil.InsertMigration(ctx, db, migrationName); err != nil {
+				// not a fatal error, log and continue
+				logrus.WithError(err).Warnf("unable to manually insert migration '%s'", migrationName)
+			}
+			return nil
+		}
+		return err
+	}
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: migrationName,
+		Up:      deltas.UpRefactorKeyChanges,
+	})
+	return m.Up(ctx)
+}
+
 func (s *keyChangesStatements) Prepare() (err error) {
 	if s.upsertKeyChangeStmt, err = s.db.Prepare(upsertKeyChangeSQL); err != nil {
 		return err

roomserver/storage/postgres/storage.go

@@ -16,10 +16,13 @@
 package postgres
 
 import (
+	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 
-	"github.com/lib/pq"
+	"github.com/sirupsen/logrus"
+
 	// Import the postgres database driver.
 	_ "github.com/lib/pq"
 
@@ -52,30 +55,8 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 
 	// Special case, since this migration uses several tables, so it needs to
 	// be sure that all tables are created first.
-	// TODO: Remove when we are sure we are not having goose artefacts in the db
+	if err = executeMigration(base.Context(), db); err != nil {
-	// This forces an error, which indicates the migration is already applied, since the
+		return nil, err
-	// column event_nid was removed from the table
-	var eventNID int
-	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan(&eventNID)
-	if err == nil {
-		m := sqlutil.NewMigrator(db)
-		m.AddMigrations(sqlutil.Migration{
-			Version: "roomserver: state blocks refactor",
-			Up:      deltas.UpStateBlocksRefactor,
-		})
-		if err = m.Up(base.Context()); err != nil {
-			return nil, err
-		}
-	} else {
-		switch e := err.(type) {
-		case *pq.Error:
-			// ignore undefined_column (42703) errors, as this is expected at this point
-			if e.Code != "42703" {
-				return nil, err
-			}
-		default:
-			return nil, err
-		}
 	}
 
 	// Then prepare the statements. Now that the migrations have run, any columns referred
@@ -87,6 +68,33 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 	return &d, nil
 }
 
+func executeMigration(ctx context.Context, db *sql.DB) error {
+	// TODO: Remove when we are sure we are not having goose artefacts in the db
+	// This forces an error, which indicates the migration is already applied, since the
+	// column event_nid was removed from the table
+	migrationName := "roomserver: state blocks refactor"
+
+	var cName string
+	err := db.QueryRowContext(ctx, "select column_name from information_schema.columns where table_name = 'roomserver_state_block' AND column_name = 'event_nid'").Scan(&cName)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
+			if err = sqlutil.InsertMigration(ctx, db, migrationName); err != nil {
+				// not a fatal error, log and continue
+				logrus.WithError(err).Warnf("unable to manually insert migration '%s'", migrationName)
+			}
+			return nil
+		}
+		return err
+	}
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: migrationName,
+		Up:      deltas.UpStateBlocksRefactor,
+	})
+
+	return m.Up(ctx)
+}
+
 func (d *Database) create(db *sql.DB) error {
 	if err := CreateEventStateKeysTable(db); err != nil {
 		return err

roomserver/storage/sqlite3/storage.go

@@ -18,9 +18,11 @@ package sqlite3
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 
 	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
 
 	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
@@ -61,20 +63,8 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 
 	// Special case, since this migration uses several tables, so it needs to
 	// be sure that all tables are created first.
-	// TODO: Remove when we are sure we are not having goose artefacts in the db
+	if err = executeMigration(base.Context(), db); err != nil {
-	// This forces an error, which indicates the migration is already applied, since the
+		return nil, err
-	// column event_nid was removed from the table
-	var eventNID int
-	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan(&eventNID)
-	if err == nil {
-		m := sqlutil.NewMigrator(db)
-		m.AddMigrations(sqlutil.Migration{
-			Version: "roomserver: state blocks refactor",
-			Up:      deltas.UpStateBlocksRefactor,
-		})
-		if err = m.Up(base.Context()); err != nil {
-			return nil, err
-		}
 	}
 
 	// Then prepare the statements. Now that the migrations have run, any columns referred
@@ -86,6 +76,32 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 	return &d, nil
 }
 
+func executeMigration(ctx context.Context, db *sql.DB) error {
+	// TODO: Remove when we are sure we are not having goose artefacts in the db
+	// This forces an error, which indicates the migration is already applied, since the
+	// column event_nid was removed from the table
+	migrationName := "roomserver: state blocks refactor"
+
+	var cName string
+	err := db.QueryRowContext(ctx, `SELECT p.name FROM sqlite_master AS m JOIN pragma_table_info(m.name) AS p WHERE m.name = 'roomserver_state_block' AND p.name = 'event_nid'`).Scan(&cName)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
+			if err = sqlutil.InsertMigration(ctx, db, migrationName); err != nil {
+				// not a fatal error, log and continue
+				logrus.WithError(err).Warnf("unable to manually insert migration '%s'", migrationName)
+			}
+			return nil
+		}
+		return err
+	}
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: migrationName,
+		Up:      deltas.UpStateBlocksRefactor,
+	})
+	return m.Up(ctx)
+}
+
 func (d *Database) create(db *sql.DB) error {
 	if err := CreateEventStateKeysTable(db); err != nil {
 		return err