move migration table updating SQL into a migration transaction

atc/db/migration/migration.go

@@ -379,41 +379,46 @@ func (m *migrator) recordMigrationFailure(migration migration, migrationErr erro
 	return migrationErr
 }
 
-func (m *migrator) runMigration(migration migration, strategy encryption.Strategy) error {
-	var err error
+func (m *migrator) runMigration(migration migration, strategy encryption.Strategy) (err error) {
+	tx, err := m.db.Begin()
+	if err != nil {
+		return err
+	}
 
-	switch migration.Strategy {
-	case GoMigration:
-		err = migrations.NewMigrations(m.db, strategy).Run(migration.Name)
+	defer func() {
 		if err != nil {
-			return m.recordMigrationFailure(
+			err = m.recordMigrationFailure(
 				migration,
-				fmt.Errorf("migration '%s' failed: %w", migration.Name, err),
+				fmt.Errorf("migration '%s' failed and was rolled back: %w", migration.Name, err),
 				false,
 			)
-		}
-	case SQLMigration:
-		_, err = m.db.Exec(migration.Statements)
-		if err != nil {
-			// rollback in case the migration was BEGIN ... COMMIT and failed
-			//
-			// note that this succeeds and does a no-op (with a warning) if no
-			// transaction was opened; we're just OK with that
-			_, rbErr := m.db.Exec(`ROLLBACK`)
+
+			rbErr := tx.Rollback()
 			if rbErr != nil {
-				return multierror.Append(err, fmt.Errorf("rollback failed: %w", rbErr))
+				err = multierror.Append(err, fmt.Errorf("rollback failed: %w", rbErr))
 			}
+		}
+	}()
 
-			return m.recordMigrationFailure(
-				migration,
-				fmt.Errorf("migration '%s' failed and was rolled back: %w", migration.Name, err),
-				false,
-			)
+	switch migration.Strategy {
+	case GoMigration:
+		err = migrations.NewMigrations(tx, strategy).Run(migration.Name)
+		if err != nil {
+			return err
+		}
+	case SQLMigration:
+		_, err = tx.Exec(migration.Statements)
+		if err != nil {
+			return err
 		}
 	}
 
-	_, err = m.db.Exec("INSERT INTO migrations_history (version, tstamp, direction, status, dirty) VALUES ($1, current_timestamp, $2, 'passed', false)", migration.Version, migration.Direction)
-	return err
+	_, err = tx.Exec("INSERT INTO migrations_history (version, tstamp, direction, status, dirty) VALUES ($1, current_timestamp, $2, 'passed', false)", migration.Version, migration.Direction)
+	if err != nil {
+		return err
+	}
+
+	return tx.Commit()
 }
 
 func (helper *migrator) Up(newKey, oldKey *encryption.Key) error {

atc/db/migration/migration_test.go

@@ -197,9 +197,7 @@ var _ = Describe("Migration", func() {
 				migrator := migration.NewMigratorForMigrations(db, lockFactory, fstest.MapFS{
 					"1000_test_table_created.up.sql": &fstest.MapFile{
 						Data: []byte(`
-							BEGIN;
 							CREATE TABLE some_table (id integer);
-							COMMIT;
 						`),
 					},
 				})
@@ -227,9 +225,7 @@ var _ = Describe("Migration", func() {
 				migrator := migration.NewMigratorForMigrations(db, lockFactory, fstest.MapFS{
 					"1000_test_table_created.up.sql": &fstest.MapFile{
 						Data: []byte(`
-							BEGIN;
 							CREATE TABLE some_table (id integer);
-							COMMIT;
 						`),
 					},
 				})
@@ -248,16 +244,12 @@ var _ = Describe("Migration", func() {
 				migrator := migration.NewMigratorForMigrations(db, lockFactory, fstest.MapFS{
 					"1000_test_table_created.up.sql": &fstest.MapFile{
 						Data: []byte(`
-							BEGIN;
 							CREATE TABLE some_table (id integer);
-							COMMIT;
 						`),
 					},
 					"1001_test_table_created.up.sql": &fstest.MapFile{
 						Data: []byte(`
-							BEGIN;
 							DROP TABLE some_table;
-							COMMIT;
 						`),
 					},
 				})
@@ -271,23 +263,17 @@ var _ = Describe("Migration", func() {
 					migrator := migration.NewMigratorForMigrations(db, lockFactory, fstest.MapFS{
 						"1000_test_table_created.up.sql": &fstest.MapFile{
 							Data: []byte(`
-								BEGIN;
 								CREATE TABLE some_table (id integer);
-								COMMIT;
 							`),
 						},
 						"1000_test_table_created.down.sql": &fstest.MapFile{
 							Data: []byte(`
-								BEGIN;
 								DROP TABLE some_table;
-								COMMIT;
 							`),
 						},
 						"1001_drop_bogus_table.up.sql": &fstest.MapFile{
 							Data: []byte(`
-								BEGIN;
 								DROP TABLE some_bogus_table;
-								COMMIT;
 							`),
 						},
 					})
@@ -304,9 +290,7 @@ var _ = Describe("Migration", func() {
 				migrator := migration.NewMigratorForMigrations(db, lockFactory, fstest.MapFS{
 					"1000_test_table_created.up.sql": &fstest.MapFile{
 						Data: []byte(`
-							BEGIN;
 							CREATE TABLE some_table (id integer);
-							COMMIT;
 						`),
 					},
 				})
@@ -325,9 +309,7 @@ var _ = Describe("Migration", func() {
 				migrator := migration.NewMigratorForMigrations(db, lockFactory, fstest.MapFS{
 					"1000_test_table_created.up.sql": &fstest.MapFile{
 						Data: []byte(`
-							BEGIN;
 							CREATE TABLE some_table (id integer);
-							COMMIT;
 						`),
 					},
 				})

atc/db/migration/migrations/1510262030_initial_schema.down.sql

@@ -1,53 +1,53 @@
-BEGIN;
 
-  DROP MATERIALIZED VIEW
-    latest_completed_builds_per_job,
-    next_builds_per_job,
-    transition_builds_per_job
-    CASCADE;
 
-  DROP SEQUENCE
-    config_version_seq,
-    one_off_name;
+DROP MATERIALIZED VIEW
+  latest_completed_builds_per_job,
+  next_builds_per_job,
+  transition_builds_per_job
+  CASCADE;
 
-  DROP TABLE
-    base_resource_types,
-    build_events,
-    build_image_resource_caches,
-    build_inputs,
-    build_outputs,
-    builds,
-    cache_invalidator,
-    containers,
-    independent_build_inputs,
-    jobs,
-    jobs_serial_groups,
-    next_build_inputs,
-    pipelines,
-    pipes,
-    resource_cache_uses,
-    resource_caches,
-    resource_config_check_sessions,
-    resource_configs,
-    resource_types,
-    resources,
-    teams,
-    versioned_resources,
-    volumes,
-    worker_base_resource_types,
-    worker_resource_caches,
-    worker_resource_config_check_sessions,
-    worker_task_caches,
-    workers
-    CASCADE;
+DROP SEQUENCE
+  config_version_seq,
+  one_off_name;
+
+DROP TABLE
+  base_resource_types,
+  build_events,
+  build_image_resource_caches,
+  build_inputs,
+  build_outputs,
+  builds,
+  cache_invalidator,
+  containers,
+  independent_build_inputs,
+  jobs,
+  jobs_serial_groups,
+  next_build_inputs,
+  pipelines,
+  pipes,
+  resource_cache_uses,
+  resource_caches,
+  resource_config_check_sessions,
+  resource_configs,
+  resource_types,
+  resources,
+  teams,
+  versioned_resources,
+  volumes,
+  worker_base_resource_types,
+  worker_resource_caches,
+  worker_resource_config_check_sessions,
+  worker_task_caches,
+  workers
+  CASCADE;
+
+DROP TYPE
+  build_status,
+  container_state,
+  container_stage,
+  container_state_old,
+  volume_state,
+  volume_state_old,
+  worker_state;
 
-  DROP TYPE
-    build_status,
-    container_state,
-    container_stage,
-    container_state_old,
-    volume_state,
-    volume_state_old,
-    worker_state;
 
-COMMIT;