Make pgroll pull and pgroll migrate command able to handle incompatible migration file formats (#812)

Ensure that the `pgroll pull` and `pgroll migrate` commands are able to
operate safely when either:

* The local migration directory contains migrations that can not be
deserialized by the current version of `pgroll`.
* The migration history in the target database contains migrations that
can not be deserialized by the current version of `pgroll`.

This ensures that these commands work in the presence of older migration
formats, either locally or in the remote schema history.

This is done by adding a new `RawMigration` type to the `migrations`
package, which does not deserialize the `Operations` in the migration
but leaves them as a `json.RawMessage`.

The following methods in the `roll` and `state` packages are then
updated to work with and return `RawMigrations`:

* `SchemaHistory` - used to return the schema history from the target
database.
* `MissingMigrations` - used to find remote migrations that don't exist
in the local history
* `UnappliedMigrations` - used to find local migrations that don't exist
in the remote.

By updating these methods and the `pull` and `migrate` commands that use
them, these commands can work safely even with local or remote
migrations that can't be deserialized by the current version of
`pgroll`.

Tests are added for both the `MissingMigrations` and
`UnappliedMigrations` methods to ensure that they work with
un-deserializable migrations.

Part of #770

Author: andrew-farries

cmd/migrate.go

@@ -3,13 +3,16 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"time"
 
 	"github.com/spf13/cobra"
 
 	"github.com/xataio/pgroll/pkg/backfill"
+	"github.com/xataio/pgroll/pkg/migrations"
+	"github.com/xataio/pgroll/pkg/roll"
 )
 
 func migrateCmd() *cobra.Command {
@@ -82,13 +85,13 @@ func migrateCmd() *cobra.Command {
 			// Run all migrations after the latest version up to the final migration,
 			// completing each one.
 			for _, mig := range migs[:len(migs)-1] {
-				if err := runMigration(ctx, m, mig, true, backfillConfig); err != nil {
+				if err := runRawMigration(ctx, m, mig, true, backfillConfig); err != nil {
 					return fmt.Errorf("failed to run migration file %q: %w", mig.Name, err)
 				}
 			}
 
 			// Run the final migration, completing it only if requested.
-			return runMigration(ctx, m, migs[len(migs)-1], complete, backfillConfig)
+			return runRawMigration(ctx, m, migs[len(migs)-1], complete, backfillConfig)
 		},
 	}
 
@@ -98,3 +101,12 @@ func migrateCmd() *cobra.Command {
 
 	return migrateCmd
 }
+
+func runRawMigration(ctx context.Context, m *roll.Roll, migration *migrations.RawMigration, complete bool, c *backfill.Config) error {
+	parsedMigration, err := migrations.ParseMigration(migration)
+	if err != nil {
+		return fmt.Errorf("failed to parse migration %q: %w", migration.Name, err)
+	}
+
+	return runMigration(ctx, m, parsedMigration, complete, c)
+}

cmd/pull.go

@@ -87,7 +87,7 @@ func pullCmd() *cobra.Command {
 // WriteToFile writes the migration to a file in `targetDir`, prefixing the
 // filename with `prefix`. The output format defaults to YAML, but can
 // be changed to JSON by setting `useJSON` to true.
-func writeMigrationToFile(m *migrations.Migration, targetDir, prefix string, useJSON bool) error {
+func writeMigrationToFile(m *migrations.RawMigration, targetDir, prefix string, useJSON bool) error {
 	err := os.MkdirAll(targetDir, 0o755)
 	if err != nil {
 		return err

pkg/migrations/migrations.go

@@ -6,10 +6,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 
 	_ "github.com/lib/pq"
-	"sigs.k8s.io/yaml"
 
 	"github.com/xataio/pgroll/pkg/db"
 	"github.com/xataio/pgroll/pkg/schema"
@@ -54,10 +52,13 @@ type RequiresSchemaRefreshOperation interface {
 type (
 	Operations []Operation
 	Migration  struct {
-		Name string `json:"name,omitempty"`
-
+		Name       string     `json:"name,omitempty"`
 		Operations Operations `json:"operations"`
 	}
+	RawMigration struct {
+		Name       string          `json:"name"`
+		Operations json.RawMessage `json:"operations"`
+	}
 )
 
 // Validate will check that the migration can be applied to the given schema
@@ -105,22 +106,3 @@ func (m *Migration) ContainsRawSQLOperation() bool {
 	}
 	return false
 }
-
-// WriteAsJSON writes the migration to the given writer in JSON format
-func (m *Migration) WriteAsJSON(w io.Writer) error {
-	encoder := json.NewEncoder(w)
-	encoder.SetIndent("", "  ")
-
-	return encoder.Encode(m)
-}
-
-// WriteAsYAML writes the migration to the given writer in YAML format
-func (m *Migration) WriteAsYAML(w io.Writer) error {
-	yml, err := yaml.Marshal(m)
-	if err != nil {
-		return err
-	}
-
-	_, err = w.Write(yml)
-	return err
-}

pkg/migrations/op_common.go

@@ -72,8 +72,9 @@ func CollectFilesFromDir(dir fs.FS) ([]string, error) {
 	return migrationFiles, nil
 }
 
-// ReadMigration opens the migration file and reads the migration.
-func ReadMigration(dir fs.FS, filename string) (*Migration, error) {
+// ReadMigration opens the migration file and reads the migration as a
+// RawMigration.
+func ReadRawMigration(dir fs.FS, filename string) (*RawMigration, error) {
 	file, err := dir.Open(filename)
 	if err != nil {
 		return nil, fmt.Errorf("opening migration file: %w", err)
@@ -85,7 +86,7 @@ func ReadMigration(dir fs.FS, filename string) (*Migration, error) {
 		return nil, err
 	}
 
-	mig := Migration{}
+	mig := RawMigration{}
 	switch filepath.Ext(filename) {
 	case ".json":
 		dec := json.NewDecoder(bytes.NewReader(byteValue))
@@ -106,6 +107,29 @@ func ReadMigration(dir fs.FS, filename string) (*Migration, error) {
 	return &mig, nil
 }
 
+// ParseMigration converts a RawMigration to a fully parsed Migration
+func ParseMigration(raw *RawMigration) (*Migration, error) {
+	var ops Operations
+	if err := json.Unmarshal(raw.Operations, &ops); err != nil {
+		return nil, fmt.Errorf("parsing operations: %w", err)
+	}
+
+	return &Migration{
+		Name:       raw.Name,
+		Operations: ops,
+	}, nil
+}
+
+// ReadMigration reads and parses a migration file
+func ReadMigration(dir fs.FS, filename string) (*Migration, error) {
+	raw, err := ReadRawMigration(dir, filename)
+	if err != nil {
+		return nil, err
+	}
+
+	return ParseMigration(raw)
+}
+
 // UnmarshalJSON deserializes the list of operations from a JSON array.
 func (v *Operations) UnmarshalJSON(data []byte) error {
 	var tmp []map[string]json.RawMessage
@@ -279,3 +303,22 @@ func operationFromName(name OpName) (Operation, error) {
 	}
 	return nil, fmt.Errorf("unknown migration type: %v", name)
 }
+
+// WriteAsJSON writes the migration to the given writer in JSON format
+func (m *RawMigration) WriteAsJSON(w io.Writer) error {
+	encoder := json.NewEncoder(w)
+	encoder.SetIndent("", "  ")
+
+	return encoder.Encode(m)
+}
+
+// WriteAsYAML writes the migration to the given writer in YAML format
+func (m *RawMigration) WriteAsYAML(w io.Writer) error {
+	yml, err := yaml.Marshal(m)
+	if err != nil {
+		return err
+	}
+
+	_, err = w.Write(yml)
+	return err
+}

pkg/roll/missing.go

@@ -13,7 +13,7 @@ import (
 // MissingMigrations returns the slice of migrations that have been applied to
 // the target database but are missing from the local migrations directory
 // `dir`.
-func (m *Roll) MissingMigrations(ctx context.Context, dir fs.FS) ([]*migrations.Migration, error) {
+func (m *Roll) MissingMigrations(ctx context.Context, dir fs.FS) ([]*migrations.RawMigration, error) {
 	// Determine the latest version of the database
 	latestVersion, err := m.State().LatestVersion(ctx, m.Schema())
 	if err != nil {
@@ -34,7 +34,7 @@ func (m *Roll) MissingMigrations(ctx context.Context, dir fs.FS) ([]*migrations.
 	// Create a set of local migration names for fast lookup
 	localMigNames := make(map[string]struct{}, len(files))
 	for _, file := range files {
-		mig, err := migrations.ReadMigration(dir, file)
+		mig, err := migrations.ReadRawMigration(dir, file)
 		if err != nil {
 			return nil, fmt.Errorf("reading migration file %s: %w", file, err)
 		}
@@ -49,7 +49,7 @@ func (m *Roll) MissingMigrations(ctx context.Context, dir fs.FS) ([]*migrations.
 
 	// Find all migrations that have been applied to the database but are missing
 	// from the local directory
-	migs := make([]*migrations.Migration, 0, len(history))
+	migs := make([]*migrations.RawMigration, 0, len(history))
 	for _, h := range history {
 		if _, ok := localMigNames[h.Migration.Name]; ok {
 			continue

pkg/roll/missing_test.go

@@ -195,6 +195,74 @@ func TestMissingMigrations(t *testing.T) {
 	})
 }
 
+func TestMissingMigrationsWithOldMigrationFormats(t *testing.T) {
+	t.Parallel()
+
+	t.Run("local directory contains an un-deserializable migration", func(t *testing.T) {
+		fs := fstest.MapFS{
+			"01_migration_1.json": &fstest.MapFile{Data: unDeserializableMigration(t, "01_migration_1")},
+		}
+
+		testutils.WithMigratorAndConnectionToContainer(t, func(roll *roll.Roll, _ *sql.DB) {
+			ctx := context.Background()
+
+			// Apply migrations to the target database
+			for _, migration := range []*migrations.Migration{
+				exampleMig(t, "02_migration_2"),
+				exampleMig(t, "03_migration_3"),
+			} {
+				err := roll.Start(ctx, migration, backfill.NewConfig())
+				require.NoError(t, err)
+				err = roll.Complete(ctx)
+				require.NoError(t, err)
+			}
+
+			// Get missing migrations
+			migs, err := roll.MissingMigrations(ctx, fs)
+			require.NoError(t, err)
+
+			// Assert that migrations 2 and 3 are missing in the local directory
+			require.Len(t, migs, 2)
+			require.Equal(t, "02_migration_2", migs[0].Name)
+			require.Equal(t, "03_migration_3", migs[1].Name)
+		})
+	})
+
+	t.Run("remote migration history contains an un-deserializable migration", func(t *testing.T) {
+		fs := fstest.MapFS{}
+
+		testutils.WithMigratorAndConnectionToContainer(t, func(roll *roll.Roll, db *sql.DB) {
+			ctx := context.Background()
+
+			// Apply migrations to the target database
+			for _, migration := range []*migrations.Migration{
+				exampleMig(t, "01_migration_1"),
+			} {
+				err := roll.Start(ctx, migration, backfill.NewConfig())
+				require.NoError(t, err)
+				err = roll.Complete(ctx)
+				require.NoError(t, err)
+			}
+
+			// Modify the first migration in the schema history to be un-deserializable; in
+			// practice this could happen if the migration was applied with an older
+			// version of pgroll that had a different migration format
+			_, err := db.ExecContext(ctx, `UPDATE pgroll.migrations
+				SET migration = REPLACE(migration::text, '"up"', '"upxxx"')::jsonb
+				WHERE name = '01_migration_1'`)
+			require.NoError(t, err)
+
+			// Get missing migrations
+			migs, err := roll.MissingMigrations(ctx, fs)
+			require.NoError(t, err)
+
+			// Assert that the un-deserializable migration is missing in the local directory
+			require.Len(t, migs, 1)
+			require.Equal(t, "01_migration_1", migs[0].Name)
+		})
+	})
+}
+
 func exampleMig(t *testing.T, name string) *migrations.Migration {
 	t.Helper()
 

pkg/roll/unapplied.go

@@ -16,7 +16,7 @@ import (
 //
 // If the local order of migrations does not match the order of migrations in
 // the schema history, an `ErrMismatchedMigration` error is returned.
-func (m *Roll) UnappliedMigrations(ctx context.Context, dir fs.FS) ([]*migrations.Migration, error) {
+func (m *Roll) UnappliedMigrations(ctx context.Context, dir fs.FS) ([]*migrations.RawMigration, error) {
 	latestVersion, err := m.State().LatestVersion(ctx, m.Schema())
 	if err != nil {
 		return nil, fmt.Errorf("determining latest version: %w", err)
@@ -36,7 +36,7 @@ func (m *Roll) UnappliedMigrations(ctx context.Context, dir fs.FS) ([]*migration
 	var idx int
 	if latestVersion != nil {
 		for _, file := range files {
-			migration, err := migrations.ReadMigration(dir, file)
+			migration, err := migrations.ReadRawMigration(dir, file)
 			if err != nil {
 				return nil, fmt.Errorf("reading migration file %q: %w", file, err)
 			}
@@ -54,9 +54,9 @@ func (m *Roll) UnappliedMigrations(ctx context.Context, dir fs.FS) ([]*migration
 	}
 
 	// Return all unapplied migrations
-	migs := make([]*migrations.Migration, 0, len(files))
+	migs := make([]*migrations.RawMigration, 0, len(files))
 	for _, file := range files[idx:] {
-		migration, err := migrations.ReadMigration(dir, file)
+		migration, err := migrations.ReadRawMigration(dir, file)
 		if err != nil {
 			return nil, fmt.Errorf("reading migration file %q: %w", file, err)
 		}