Skip to content

Commit 21bcde6

Browse files
hundt-corbalthundt-corbalt
authored
[CMCSMACD-4253] Run each migration in its own transaction (#8)
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 21bcde6

3 files changed

Lines changed: 150 additions & 108 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: 82 additions & 67 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,40 @@ 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
154131
}
155-
156-
if rollBackThroughIndex < 0 {
157-
return fmt.Errorf("Invalid target index %d", rollBackThroughIndex)
158-
}
159-
if rollBackThroughIndex >= firstUnappliedIndex {
160-
return fmt.Errorf("Migration %d has not been applied yet", rollBackThroughIndex)
132+
if firstUnappliedIndex >= len(migrations) {
133+
return false, nil
161134
}
162135

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-
}
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)
141+
}
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)
177145
}
178146

147+
committed = true
179148
err = tx.Commit()
180149
if err != nil {
181-
return fmt.Errorf("Error committing migrations: %w", err)
150+
return false, fmt.Errorf("Error committing migrations: %w", err)
182151
}
183-
committed = true
184-
return nil
152+
return true, nil
185153
}
186154

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-
}
155+
func rollbackOne(db *sqlx.DB, migrations []NamedMigration, rollBackThroughIndex int) (rolledBackIndex int, err error) {
196156
tx, err := db.Beginx()
197157
if err != nil {
198-
return fmt.Errorf("Error starting migrations transaction: %w", err)
158+
return -1, fmt.Errorf("Error starting migrations transaction: %w", err)
199159
}
200160
committed := false
201161
defer func() {
@@ -209,23 +169,78 @@ func Migrate(db *sqlx.DB, migrations []NamedMigration) error {
209169

210170
_, err = tx.Exec("LOCK TABLE migration")
211171
if err != nil {
212-
return fmt.Errorf("Error locking migration table: %w", err)
172+
return -1, fmt.Errorf("Error locking migration table: %w", err)
213173
}
214174

215175
firstUnappliedIndex, err := verifyMigrations(tx, migrations)
216176
if err != nil {
217-
return err
177+
return -1, err
218178
}
219179

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

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

‎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)