Skip to content

Commit caa7c99

Browse files
author
Chris Hundt
committed
[CMCSMACD-4253] Run each migration in its own transaction
Some types of changes (like adding new enum values) cannot be used until after the transaction including them is committed.[0] As part of this change, the verification program is updated to: - include a migration that fails if it is included in the same transaction as the previous migration - call Migrate() on all migrations once before calling it on each one individually which reproduces the error that this change fixes. [0] https://www.postgresql.org/docs/16/sql-altertype.html
1 parent ce1fb12 commit caa7c99

3 files changed

Lines changed: 152 additions & 107 deletions

File tree

‎cmd/schema-test/migrations.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,4 +68,13 @@ var allMigrations = []migrations.NamedMigration{
6868
`DROP TYPE type1_old`,
6969
}),
7070
},
71+
{
72+
Name: "Create a view referencing a new enum value",
73+
Migration: migrations.StaticMigration([]string{
74+
`CREATE VIEW v AS SELECT * FROM table3 WHERE v = 'type1val3'`,
75+
}),
76+
Reverse: migrations.StaticMigration([]string{
77+
`DROP VIEW v`,
78+
}),
79+
},
7180
}

‎migrations/migration.go‎

Lines changed: 84 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -105,33 +105,10 @@ func verifyMigrations(tx *sqlx.Tx, migrations []NamedMigration) (firstUnappliedM
105105
return firstUnappliedMigrationIndex, nil
106106
}
107107

108-
func doMigrations(tx *sqlx.Tx, migrations []NamedMigration, startIndex int) error {
109-
for index := startIndex; index < len(migrations); index++ {
110-
migration := migrations[index]
111-
log.Printf("Performing migration %d (%q)", index, migration.Name)
112-
err := migrations[index].Migration.DoMigration(tx)
113-
if err != nil {
114-
return fmt.Errorf("Error performing migration %d (%q): %w", index, migration.Name, err)
115-
}
116-
_, err = tx.Exec(`INSERT INTO migration ("index", name) VALUES ($1, $2)`, index, migration.Name)
117-
if err != nil {
118-
return fmt.Errorf("Error recording migration %d (%q): %w", index, migration.Name, err)
119-
}
120-
}
121-
122-
return nil
123-
}
124-
125-
// Rollback runs the Reverse migrations for all the input migrations with index >= rollBackThroughIndex.
126-
// The input migrations must include all migrations, not just the ones to roll back.
127-
func Rollback(db *sqlx.DB, migrations []NamedMigration, rollBackThroughIndex int) error {
128-
err := ensureMigrationsTableExists(db)
129-
if err != nil {
130-
return err
131-
}
108+
func migrateOne(db *sqlx.DB, migrations []NamedMigration) (bool, error) {
132109
tx, err := db.Beginx()
133110
if err != nil {
134-
return fmt.Errorf("Error starting migrations transaction: %w", err)
111+
return false, fmt.Errorf("Error starting migrations transaction: %w", err)
135112
}
136113
committed := false
137114
defer func() {
@@ -145,57 +122,43 @@ func Rollback(db *sqlx.DB, migrations []NamedMigration, rollBackThroughIndex int
145122

146123
_, err = tx.Exec("LOCK TABLE migration")
147124
if err != nil {
148-
return fmt.Errorf("Error locking migration table: %w", err)
125+
return false, fmt.Errorf("Error locking migration table: %w", err)
149126
}
150127

151128
firstUnappliedIndex, err := verifyMigrations(tx, migrations)
152129
if err != nil {
153-
return err
130+
return false, err
131+
}
132+
if firstUnappliedIndex >= len(migrations) {
133+
return false, nil
154134
}
155135

156-
if rollBackThroughIndex < 0 {
157-
return fmt.Errorf("Invalid target index %d", rollBackThroughIndex)
136+
migration := migrations[firstUnappliedIndex]
137+
log.Printf("Performing migration %d (%q)", firstUnappliedIndex, migration.Name)
138+
err = migrations[firstUnappliedIndex].Migration.DoMigration(tx)
139+
if err != nil {
140+
return false, fmt.Errorf("Error performing migration %d (%q): %w", firstUnappliedIndex, migration.Name, err)
158141
}
159-
if rollBackThroughIndex >= firstUnappliedIndex {
160-
return fmt.Errorf("Migration %d has not been applied yet", rollBackThroughIndex)
142+
_, err = tx.Exec(`INSERT INTO migration ("index", name) VALUES ($1, $2)`, firstUnappliedIndex, migration.Name)
143+
if err != nil {
144+
return false, fmt.Errorf("Error recording migration %d (%q): %w", firstUnappliedIndex, migration.Name, err)
161145
}
162-
163-
for index := firstUnappliedIndex - 1; index >= rollBackThroughIndex; index-- {
164-
migration := migrations[index]
165-
if migration.Reverse == nil {
166-
return fmt.Errorf("No Reverse for migration %d (%q)", index, migration.Name)
167-
}
168-
log.Printf("Reversing migration %d (%q)", index, migration.Name)
169-
err := migrations[index].Reverse.DoMigration(tx)
170-
if err != nil {
171-
return fmt.Errorf("Error reversing migration %d (%q): %w", index, migration.Name, err)
172-
}
173-
_, err = tx.Exec(`DELETE FROM migration WHERE "index"=$1`, index)
174-
if err != nil {
175-
return fmt.Errorf("Error deleting migration row %d (%q): %w", index, migration.Name, err)
176-
}
146+
if err != nil {
147+
return false, err
177148
}
178149

150+
committed = true
179151
err = tx.Commit()
180152
if err != nil {
181-
return fmt.Errorf("Error committing migrations: %w", err)
153+
return false, fmt.Errorf("Error committing migrations: %w", err)
182154
}
183-
committed = true
184-
return nil
155+
return true, nil
185156
}
186157

187-
// Migrate does the following:
188-
// 1. Verifies that the `migration` table exists, and creates it if it does not.
189-
// 2. Verifies that the existing migrations recorded in the database match (by name and order) the migrations given as the argument.
190-
// 3. Performs any migrations that are not yet recorded in the database.
191-
func Migrate(db *sqlx.DB, migrations []NamedMigration) error {
192-
err := ensureMigrationsTableExists(db)
193-
if err != nil {
194-
return err
195-
}
158+
func rollbackOne(db *sqlx.DB, migrations []NamedMigration, rollBackThroughIndex int) (rolledBackIndex int, err error) {
196159
tx, err := db.Beginx()
197160
if err != nil {
198-
return fmt.Errorf("Error starting migrations transaction: %w", err)
161+
return -1, fmt.Errorf("Error starting migrations transaction: %w", err)
199162
}
200163
committed := false
201164
defer func() {
@@ -209,23 +172,78 @@ func Migrate(db *sqlx.DB, migrations []NamedMigration) error {
209172

210173
_, err = tx.Exec("LOCK TABLE migration")
211174
if err != nil {
212-
return fmt.Errorf("Error locking migration table: %w", err)
175+
return -1, fmt.Errorf("Error locking migration table: %w", err)
213176
}
214177

215178
firstUnappliedIndex, err := verifyMigrations(tx, migrations)
216179
if err != nil {
217-
return err
180+
return -1, err
218181
}
219182

220-
err = doMigrations(tx, migrations, firstUnappliedIndex)
183+
if rollBackThroughIndex < 0 {
184+
return -1, fmt.Errorf("Invalid target index %d", rollBackThroughIndex)
185+
}
186+
if rollBackThroughIndex >= firstUnappliedIndex {
187+
return -1, fmt.Errorf("Migration %d has not been applied yet", rollBackThroughIndex)
188+
}
189+
190+
index := firstUnappliedIndex - 1
191+
migration := migrations[index]
192+
if migration.Reverse == nil {
193+
return -1, fmt.Errorf("No Reverse for migration %d (%q)", index, migration.Name)
194+
}
195+
log.Printf("Reversing migration %d (%q)", index, migration.Name)
196+
err = migrations[index].Reverse.DoMigration(tx)
221197
if err != nil {
222-
return err
198+
return -1, fmt.Errorf("Error reversing migration %d (%q): %w", index, migration.Name, err)
199+
}
200+
_, err = tx.Exec(`DELETE FROM migration WHERE "index"=$1`, index)
201+
if err != nil {
202+
return -1, fmt.Errorf("Error deleting migration row %d (%q): %w", index, migration.Name, err)
223203
}
224204

205+
committed = true
225206
err = tx.Commit()
226207
if err != nil {
227-
return fmt.Errorf("Error committing migrations: %w", err)
208+
return -1, fmt.Errorf("Error committing migrations: %w", err)
209+
}
210+
return index, nil
211+
}
212+
213+
// Rollback runs the Reverse migrations for all the input migrations with index >= rollBackThroughIndex.
214+
// The input migrations must include all migrations, not just the ones to roll back.
215+
func Rollback(db *sqlx.DB, migrations []NamedMigration, rollBackThroughIndex int) error {
216+
err := ensureMigrationsTableExists(db)
217+
if err != nil {
218+
return err
219+
}
220+
for {
221+
rolledBackIndex, err := rollbackOne(db, migrations, rollBackThroughIndex)
222+
if err != nil {
223+
return err
224+
}
225+
if rolledBackIndex == rollBackThroughIndex {
226+
return nil
227+
}
228+
}
229+
}
230+
231+
// Migrate does the following:
232+
// 1. Verifies that the `migration` table exists, and creates it if it does not.
233+
// 2. Verifies that the existing migrations recorded in the database match (by name and order) the migrations given as the argument.
234+
// 3. Performs any migrations that are not yet recorded in the database.
235+
func Migrate(db *sqlx.DB, migrations []NamedMigration) error {
236+
err := ensureMigrationsTableExists(db)
237+
if err != nil {
238+
return err
239+
}
240+
for {
241+
migrated, err := migrateOne(db, migrations)
242+
if err != nil {
243+
return err
244+
}
245+
if !migrated {
246+
return nil
247+
}
228248
}
229-
committed = true
230-
return nil
231249
}

‎migrations/verify.go‎

Lines changed: 59 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -65,15 +65,61 @@ func verifyNoTables(db *sqlx.DB) error {
6565
return errors.New("Existing tables found. You must run SchemaTest on an empty database.")
6666
}
6767

68+
func migrateAndRollback(emptyDBConfig *PostgresConfig, db *sqlx.DB, allMigrations []NamedMigration, migrateToIndex, rollbackThroughIndex int, repeatForward bool) error {
69+
beforeMigrate, err := dump(emptyDBConfig)
70+
if err != nil {
71+
return fmt.Errorf("Error calling pg_dump: %s", err)
72+
}
73+
err = Migrate(db, allMigrations[:migrateToIndex+1])
74+
if err != nil {
75+
return fmt.Errorf("Migrate to %q failed: %s", allMigrations[migrateToIndex].Name, err)
76+
}
77+
afterMigrate, err := dump(emptyDBConfig)
78+
if err != nil {
79+
return fmt.Errorf("Error calling pg_dump: %s", err)
80+
}
81+
err = Rollback(db, allMigrations, rollbackThroughIndex)
82+
if err != nil {
83+
return fmt.Errorf("Rollback through %q failed: %s", allMigrations[rollbackThroughIndex].Name, err)
84+
}
85+
afterRollback, err := dump(emptyDBConfig)
86+
if err != nil {
87+
return fmt.Errorf("Error calling pg_dump: %s", err)
88+
}
89+
if string(beforeMigrate) != string(afterRollback) {
90+
fmt.Printf("%s\n", cmp.Diff(string(beforeMigrate), string(afterRollback)))
91+
return fmt.Errorf("Dump after rollback through %q did not match the dump before the migration", allMigrations[rollbackThroughIndex].Name)
92+
}
93+
if repeatForward {
94+
err = Migrate(db, allMigrations[:migrateToIndex+1])
95+
if err != nil {
96+
return fmt.Errorf("Migration to %q failed: %s", allMigrations[migrateToIndex].Name, err)
97+
}
98+
afterMigrateAgain, err := dump(emptyDBConfig)
99+
if err != nil {
100+
return fmt.Errorf("Error calling pg_dump: %s", err)
101+
}
102+
if string(afterMigrate) != string(afterMigrateAgain) {
103+
fmt.Printf("%s\n", cmp.Diff(string(afterMigrate), string(afterMigrateAgain)))
104+
return fmt.Errorf("Dump after re-migration of %q did not match dump after first migration", allMigrations[migrateToIndex].Name)
105+
}
106+
}
107+
return err
108+
}
109+
68110
// Schema test expects a new *empty* postgres database.
69-
// It will, for each migration:
70-
// 1. Apply the migration
71-
// 2. Reverse the migration
72-
// 3. Apply the migration again
111+
// It will:
112+
// 1. Apply all migrations
113+
// 2. Reverse all migrations
114+
// 3. For each migration:
115+
// a. Apply the migration
116+
// b. Reverse the migration
117+
// c. Apply the migration again
118+
//
73119
// Before and after each step it will use pg_dump to dump the database schema.
74120
// It will verify that:
75-
// A. The schema is the same after step 2 as before step 1.
76-
// B. The schema is the same after step 3 as after step 1.
121+
// A. The schema is the same after reversing as before applying.
122+
// B. (If re-applying) The schema is the same after applying as after re-applying.
77123
//
78124
// You must have `pg_dump` in your `PATH` to run this.
79125
func SchemaTest(emptyDBConfig *PostgresConfig, allMigrations []NamedMigration) error {
@@ -103,42 +149,14 @@ func SchemaTest(emptyDBConfig *PostgresConfig, allMigrations []NamedMigration) e
103149
if err != nil {
104150
return fmt.Errorf("Setting up migrations table failed: %s", err)
105151
}
106-
for idx, migration := range allMigrations {
107-
beforeMigrate, err := dump(emptyDBConfig)
108-
if err != nil {
109-
return fmt.Errorf("Error calling pg_dump: %s", err)
110-
}
111-
err = Migrate(db, allMigrations[:idx+1])
112-
if err != nil {
113-
return fmt.Errorf("Migration %q failed: %s", migration.Name, err)
114-
}
115-
afterMigrate, err := dump(emptyDBConfig)
116-
if err != nil {
117-
return fmt.Errorf("Error calling pg_dump: %s", err)
118-
}
119-
err = Rollback(db, allMigrations, idx)
120-
if err != nil {
121-
return fmt.Errorf("Rollback to %q failed: %s", migration.Name, err)
122-
}
123-
afterRollback, err := dump(emptyDBConfig)
124-
if err != nil {
125-
return fmt.Errorf("Error calling pg_dump: %s", err)
126-
}
127-
if string(beforeMigrate) != string(afterRollback) {
128-
fmt.Printf("%s\n", cmp.Diff(string(beforeMigrate), string(afterRollback)))
129-
return fmt.Errorf("Dump after rollback of %q did not match the dump before the migration", migration.Name)
130-
}
131-
err = Migrate(db, allMigrations[:idx+1])
132-
if err != nil {
133-
return fmt.Errorf("Migration %q failed: %s", migration.Name, err)
134-
}
135-
afterMigrateAgain, err := dump(emptyDBConfig)
152+
err = migrateAndRollback(emptyDBConfig, db, allMigrations, len(allMigrations)-1, 0, false)
153+
if err != nil {
154+
return err
155+
}
156+
for idx := range allMigrations {
157+
err := migrateAndRollback(emptyDBConfig, db, allMigrations, idx, idx, true)
136158
if err != nil {
137-
return fmt.Errorf("Error calling pg_dump: %s", err)
138-
}
139-
if string(afterMigrate) != string(afterMigrateAgain) {
140-
fmt.Printf("%s\n", cmp.Diff(string(afterMigrate), string(afterMigrateAgain)))
141-
return fmt.Errorf("Dump after re-migration of %q did not match dump after first migration", migration.Name)
159+
return err
142160
}
143161
}
144162
return nil

0 commit comments

Comments
 (0)