Update database migrations, remove goose (#2264)

* Add new db migration

* Update migrations
Remove goose

* Add possibility to test direct upgrades

* Try to fix WASM test

* Add checks for specific migrations

* Remove AddMigration
Use WithTransaction
Add Dendrite version to table

* Fix linter issues

* Update tests

* Update comments, outdent if

* Namespace migrations

* Add direct upgrade tests, skipping over one version

* Split migrations

* Update go version in CI

* Fix copy&paste mistake

* Use contexts in migrations

Co-authored-by: kegsay <kegan@matrix.org>
Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>

4 files changed, 63 insertions, 62 deletions

diff --git a/roomserver/storage/sqlite3/deltas/20201028212440_add_forgotten_column.go b/roomserver/storage/sqlite3/deltas/20201028212440_add_forgotten_column.go
index d08ab02d..4c002e33 100644
--- a/roomserver/storage/sqlite3/deltas/20201028212440_add_forgotten_column.go
+++ b/roomserver/storage/sqlite3/deltas/20201028212440_add_forgotten_column.go
@@ -15,24 +15,13 @@
 package deltas
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
-
-	"github.com/matrix-org/dendrite/internal/sqlutil"
-	"github.com/pressly/goose"
 )
 
-func LoadFromGoose() {
-	goose.AddMigration(UpAddForgottenColumn, DownAddForgottenColumn)
-	goose.AddMigration(UpStateBlocksRefactor, DownStateBlocksRefactor)
-}
-
-func LoadAddForgottenColumn(m *sqlutil.Migrations) {
-	m.AddMigration(UpAddForgottenColumn, DownAddForgottenColumn)
-}
-
-func UpAddForgottenColumn(tx *sql.Tx) error {
-	_, err := tx.Exec(`	ALTER TABLE roomserver_membership RENAME TO roomserver_membership_tmp;
+func UpAddForgottenColumn(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `	ALTER TABLE roomserver_membership RENAME TO roomserver_membership_tmp;
 CREATE TABLE IF NOT EXISTS roomserver_membership (
 		room_nid INTEGER NOT NULL,
 		target_nid INTEGER NOT NULL,
@@ -57,8 +46,8 @@ DROP TABLE roomserver_membership_tmp;`)
 	return nil
 }
 
-func DownAddForgottenColumn(tx *sql.Tx) error {
-	_, err := tx.Exec(`	ALTER TABLE roomserver_membership RENAME TO roomserver_membership_tmp;
+func DownAddForgottenColumn(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `	ALTER TABLE roomserver_membership RENAME TO roomserver_membership_tmp;
 CREATE TABLE IF NOT EXISTS roomserver_membership (
 		room_nid INTEGER NOT NULL,
 		target_nid INTEGER NOT NULL,
diff --git a/roomserver/storage/sqlite3/deltas/2021041615092700_state_blocks_refactor.go b/roomserver/storage/sqlite3/deltas/2021041615092700_state_blocks_refactor.go
index 8f5ab8fc..00978121 100644
--- a/roomserver/storage/sqlite3/deltas/2021041615092700_state_blocks_refactor.go
+++ b/roomserver/storage/sqlite3/deltas/2021041615092700_state_blocks_refactor.go
@@ -21,40 +21,35 @@ import (
 	"fmt"
 
 	"github.com/matrix-org/dendrite/internal"
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/util"
 	"github.com/sirupsen/logrus"
 )
 
-func LoadStateBlocksRefactor(m *sqlutil.Migrations) {
-	m.AddMigration(UpStateBlocksRefactor, DownStateBlocksRefactor)
-}
-
 // nolint:gocyclo
-func UpStateBlocksRefactor(tx *sql.Tx) error {
+func UpStateBlocksRefactor(ctx context.Context, tx *sql.Tx) error {
 	logrus.Warn("Performing state storage upgrade. Please wait, this may take some time!")
 	defer logrus.Warn("State storage upgrade complete")
 
 	var maxsnapshotid int
 	var maxblockid int
-	if err := tx.QueryRow(`SELECT IFNULL(MAX(state_snapshot_nid),0) FROM roomserver_state_snapshots;`).Scan(&maxsnapshotid); err != nil {
-		return fmt.Errorf("tx.QueryRow.Scan (count snapshots): %w", err)
+	if err := tx.QueryRowContext(ctx, `SELECT IFNULL(MAX(state_snapshot_nid),0) FROM roomserver_state_snapshots;`).Scan(&maxsnapshotid); err != nil {
+		return fmt.Errorf("tx.QueryRowContext.Scan (count snapshots): %w", err)
 	}
-	if err := tx.QueryRow(`SELECT IFNULL(MAX(state_block_nid),0) FROM roomserver_state_block;`).Scan(&maxblockid); err != nil {
-		return fmt.Errorf("tx.QueryRow.Scan (count snapshots): %w", err)
+	if err := tx.QueryRowContext(ctx, `SELECT IFNULL(MAX(state_block_nid),0) FROM roomserver_state_block;`).Scan(&maxblockid); err != nil {
+		return fmt.Errorf("tx.QueryRowContext.Scan (count snapshots): %w", err)
 	}
 	maxsnapshotid++
 	maxblockid++
 	oldMaxSnapshotID := maxsnapshotid
 
-	if _, err := tx.Exec(`ALTER TABLE roomserver_state_block RENAME TO _roomserver_state_block;`); err != nil {
-		return fmt.Errorf("tx.Exec: %w", err)
+	if _, err := tx.ExecContext(ctx, `ALTER TABLE roomserver_state_block RENAME TO _roomserver_state_block;`); err != nil {
+		return fmt.Errorf("tx.ExecContext: %w", err)
 	}
-	if _, err := tx.Exec(`ALTER TABLE roomserver_state_snapshots RENAME TO _roomserver_state_snapshots;`); err != nil {
-		return fmt.Errorf("tx.Exec: %w", err)
+	if _, err := tx.ExecContext(ctx, `ALTER TABLE roomserver_state_snapshots RENAME TO _roomserver_state_snapshots;`); err != nil {
+		return fmt.Errorf("tx.ExecContext: %w", err)
 	}
-	_, err := tx.Exec(`
+	_, err := tx.ExecContext(ctx, `
 		CREATE TABLE IF NOT EXISTS roomserver_state_block (
 			state_block_nid INTEGER PRIMARY KEY AUTOINCREMENT,
 			state_block_hash BLOB UNIQUE,
@@ -62,9 +57,9 @@ func UpStateBlocksRefactor(tx *sql.Tx) error {
 		);
 	`)
 	if err != nil {
-		return fmt.Errorf("tx.Exec: %w", err)
+		return fmt.Errorf("tx.ExecContext: %w", err)
 	}
-	_, err = tx.Exec(`
+	_, err = tx.ExecContext(ctx, `
 		CREATE TABLE IF NOT EXISTS roomserver_state_snapshots (
 			state_snapshot_nid INTEGER PRIMARY KEY AUTOINCREMENT,
 			state_snapshot_hash BLOB UNIQUE,
@@ -73,11 +68,11 @@ func UpStateBlocksRefactor(tx *sql.Tx) error {
 	  	);
 	`)
 	if err != nil {
-		return fmt.Errorf("tx.Exec: %w", err)
+		return fmt.Errorf("tx.ExecContext: %w", err)
 	}
-	snapshotrows, err := tx.Query(`SELECT state_snapshot_nid, room_nid, state_block_nids FROM _roomserver_state_snapshots;`)
+	snapshotrows, err := tx.QueryContext(ctx, `SELECT state_snapshot_nid, room_nid, state_block_nids FROM _roomserver_state_snapshots;`)
 	if err != nil {
-		return fmt.Errorf("tx.Query: %w", err)
+		return fmt.Errorf("tx.QueryContext: %w", err)
 	}
 	defer internal.CloseAndLogIfError(context.TODO(), snapshotrows, "rows.close() failed")
 	for snapshotrows.Next() {
@@ -99,7 +94,7 @@ func UpStateBlocksRefactor(tx *sql.Tx) error {
 			// in question a state snapshot NID of 0 to indicate 'no snapshot'.
 			// If we don't do this, we'll fail the assertions later on which try to ensure we didn't forget
 			// any snapshots.
-			_, err = tx.Exec(
+			_, err = tx.ExecContext(ctx,
 				`UPDATE roomserver_events SET state_snapshot_nid = 0 WHERE event_type_nid = $1 AND event_state_key_nid = $2 AND state_snapshot_nid = $3`,
 				types.MRoomCreateNID, types.EmptyStateKeyNID, snapshot,
 			)
@@ -109,9 +104,9 @@ func UpStateBlocksRefactor(tx *sql.Tx) error {
 		}
 		for _, block := range blocks {
 			if err = func() error {
-				blockrows, berr := tx.Query(`SELECT event_nid FROM _roomserver_state_block WHERE state_block_nid = $1`, block)
+				blockrows, berr := tx.QueryContext(ctx, `SELECT event_nid FROM _roomserver_state_block WHERE state_block_nid = $1`, block)
 				if berr != nil {
-					return fmt.Errorf("tx.Query (event nids from old block): %w", berr)
+					return fmt.Errorf("tx.QueryContext (event nids from old block): %w", berr)
 				}
 				defer internal.CloseAndLogIfError(context.TODO(), blockrows, "rows.close() failed")
 				events := types.EventNIDs{}
@@ -129,14 +124,14 @@ func UpStateBlocksRefactor(tx *sql.Tx) error {
 				}
 
 				var blocknid types.StateBlockNID
-				err = tx.QueryRow(`
+				err = tx.QueryRowContext(ctx, `
 					INSERT INTO roomserver_state_block (state_block_nid, state_block_hash, event_nids)
 						VALUES ($1, $2, $3)
 						ON CONFLICT (state_block_hash) DO UPDATE SET event_nids=$3
 						RETURNING state_block_nid
 				`, maxblockid, events.Hash(), eventjson).Scan(&blocknid)
 				if err != nil {
-					return fmt.Errorf("tx.QueryRow.Scan (insert new block): %w", err)
+					return fmt.Errorf("tx.QueryRowContext.Scan (insert new block): %w", err)
 				}
 				maxblockid++
 				newblocks = append(newblocks, blocknid)
@@ -151,22 +146,22 @@ func UpStateBlocksRefactor(tx *sql.Tx) error {
 			}
 
 			var newsnapshot types.StateSnapshotNID
-			err = tx.QueryRow(`
+			err = tx.QueryRowContext(ctx, `
 				INSERT INTO roomserver_state_snapshots (state_snapshot_nid, state_snapshot_hash, room_nid, state_block_nids)
 					VALUES ($1, $2, $3, $4)
 					ON CONFLICT (state_snapshot_hash) DO UPDATE SET room_nid=$3
 					RETURNING state_snapshot_nid
 			`, maxsnapshotid, newblocks.Hash(), room, newblocksjson).Scan(&newsnapshot)
 			if err != nil {
-				return fmt.Errorf("tx.QueryRow.Scan (insert new snapshot): %w", err)
+				return fmt.Errorf("tx.QueryRowContext.Scan (insert new snapshot): %w", err)
 			}
 			maxsnapshotid++
-			_, err = tx.Exec(`UPDATE roomserver_events SET state_snapshot_nid=$1 WHERE state_snapshot_nid=$2 AND state_snapshot_nid<$3`, newsnapshot, snapshot, maxsnapshotid)
+			_, err = tx.ExecContext(ctx, `UPDATE roomserver_events SET state_snapshot_nid=$1 WHERE state_snapshot_nid=$2 AND state_snapshot_nid<$3`, newsnapshot, snapshot, maxsnapshotid)
 			if err != nil {
-				return fmt.Errorf("tx.Exec (update events): %w", err)
+				return fmt.Errorf("tx.ExecContext (update events): %w", err)
 			}
-			if _, err = tx.Exec(`UPDATE roomserver_rooms SET state_snapshot_nid=$1 WHERE state_snapshot_nid=$2 AND state_snapshot_nid<$3`, newsnapshot, snapshot, maxsnapshotid); err != nil {
-				return fmt.Errorf("tx.Exec (update rooms): %w", err)
+			if _, err = tx.ExecContext(ctx, `UPDATE roomserver_rooms SET state_snapshot_nid=$1 WHERE state_snapshot_nid=$2 AND state_snapshot_nid<$3`, newsnapshot, snapshot, maxsnapshotid); err != nil {
+				return fmt.Errorf("tx.ExecContext (update rooms): %w", err)
 			}
 		}
 	}
@@ -175,13 +170,13 @@ func UpStateBlocksRefactor(tx *sql.Tx) error {
 	// If we do, this is a problem if Dendrite tries to load the snapshot as it will not exist
 	// in roomserver_state_snapshots
 	var count int64
-	if err = tx.QueryRow(`SELECT COUNT(*) FROM roomserver_events WHERE state_snapshot_nid < $1 AND state_snapshot_nid != 0`, oldMaxSnapshotID).Scan(&count); err != nil {
+	if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM roomserver_events WHERE state_snapshot_nid < $1 AND state_snapshot_nid != 0`, oldMaxSnapshotID).Scan(&count); err != nil {
 		return fmt.Errorf("assertion query failed: %s", err)
 	}
 	if count > 0 {
 		var res sql.Result
 		var c int64
-		res, err = tx.Exec(`UPDATE roomserver_events SET state_snapshot_nid = 0 WHERE state_snapshot_nid < $1 AND state_snapshot_nid != 0`, oldMaxSnapshotID)
+		res, err = tx.ExecContext(ctx, `UPDATE roomserver_events SET state_snapshot_nid = 0 WHERE state_snapshot_nid < $1 AND state_snapshot_nid != 0`, oldMaxSnapshotID)
 		if err != nil && err != sql.ErrNoRows {
 			return fmt.Errorf("failed to reset invalid state snapshots: %w", err)
 		}
@@ -191,23 +186,23 @@ func UpStateBlocksRefactor(tx *sql.Tx) error {
 			return fmt.Errorf("expected to reset %d event(s) but only updated %d event(s)", count, c)
 		}
 	}
-	if err = tx.QueryRow(`SELECT COUNT(*) FROM roomserver_rooms WHERE state_snapshot_nid < $1 AND state_snapshot_nid != 0`, oldMaxSnapshotID).Scan(&count); err != nil {
+	if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM roomserver_rooms WHERE state_snapshot_nid < $1 AND state_snapshot_nid != 0`, oldMaxSnapshotID).Scan(&count); err != nil {
 		return fmt.Errorf("assertion query failed: %s", err)
 	}
 	if count > 0 {
 		return fmt.Errorf("%d rooms exist in roomserver_rooms which have not been converted to a new state_snapshot_nid; this is a bug, please report", count)
 	}
 
-	if _, err = tx.Exec(`DROP TABLE _roomserver_state_snapshots;`); err != nil {
+	if _, err = tx.ExecContext(ctx, `DROP TABLE _roomserver_state_snapshots;`); err != nil {
 		return fmt.Errorf("tx.Exec (delete old snapshot table): %w", err)
 	}
-	if _, err = tx.Exec(`DROP TABLE _roomserver_state_block;`); err != nil {
+	if _, err = tx.ExecContext(ctx, `DROP TABLE _roomserver_state_block;`); err != nil {
 		return fmt.Errorf("tx.Exec (delete old block table): %w", err)
 	}
 
 	return nil
 }
 
-func DownStateBlocksRefactor(tx *sql.Tx) error {
+func DownStateBlocksRefactor(ctx context.Context, tx *sql.Tx) error {
 	panic("Downgrading state storage is not supported")
 }
diff --git a/roomserver/storage/sqlite3/membership_table.go b/roomserver/storage/sqlite3/membership_table.go
index 9dda82fb..f3303eb0 100644
--- a/roomserver/storage/sqlite3/membership_table.go
+++ b/roomserver/storage/sqlite3/membership_table.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/roomserver/storage/sqlite3/deltas"
 	"github.com/matrix-org/dendrite/roomserver/storage/tables"
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -148,7 +149,15 @@ type membershipStatements struct {
 
 func CreateMembershipTable(db *sql.DB) error {
 	_, err := db.Exec(membershipSchema)
-	return err
+	if err != nil {
+		return err
+	}
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: "roomserver: add forgotten column",
+		Up:      deltas.UpAddForgottenColumn,
+	})
+	return m.Up(context.Background())
 }
 
 func PrepareMembershipTable(db *sql.DB) (tables.Membership, error) {
diff --git a/roomserver/storage/sqlite3/storage.go b/roomserver/storage/sqlite3/storage.go
index 8325fdad..bb9c15b5 100644
--- a/roomserver/storage/sqlite3/storage.go
+++ b/roomserver/storage/sqlite3/storage.go
@@ -54,17 +54,25 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 	// db.SetMaxOpenConns(20)
 
 	// Create the tables.
-	if err := d.create(db); err != nil {
+	if err = d.create(db); err != nil {
 		return nil, err
 	}
 
-	// Then execute the migrations. By this point the tables are created with the latest
-	// schemas.
-	m := sqlutil.NewMigrations()
-	deltas.LoadAddForgottenColumn(m)
-	deltas.LoadStateBlocksRefactor(m)
-	if err := m.RunDeltas(db, dbProperties); err != nil {
-		return nil, err
+	// 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
+	// This forces an error, which indicates the migration is already applied, since the
+	// column event_nid was removed from the table
+	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan()
+	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