Repository: ariga/atlas
Pull request #880 (2022-06-20 08:02:15)

sql/migrate: check revision state when attempting to compute pending …
…migration files


Patch:
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index 1dd396b1..28807a18 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -373,7 +373,8 @@ func (e *Executor) Lock(ctx context.Context) (schema.UnlockFunc, error) {
 	return unlock, nil
 }
 
-// Pending returns all pending (not applied) migration files in the migration directory.
+// Pending returns all pending (not applied) migration files in the migration directory. It will return an error, if
+// there is at least one revision in a "not-ok" state (error or ongoing).
 func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 	// Don't operate with a broken migration directory.
 	if err := Validate(e.dir); err != nil {
@@ -384,6 +385,16 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 	if err != nil {
 		return nil, fmt.Errorf("sql/migrate: execute: read revisions: %w", err)
 	}
+	// Check for all revisions to be "okay".
+	for _, r := range revs {
+		if r.ExecutionState != stateOK {
+			return nil, fmt.Errorf(
+				"sql/migrate: execute: dirty migration state: version %q has state %q",
+				r.Version,
+				r.ExecutionState,
+			)
+		}
+	}
 	// Select the correct migration files.
 	sc := e.dir.(Scanner)
 	migrations, err := sc.Files()
@@ -424,7 +435,8 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 	return migrations[len(revs):], nil
 }
 
-// Execute executes the given migration file on the database.
+// Execute executes the given migration file on the database. It does not check for the database to be clean before
+// attempting to apply the changes. This behavior is required to enabled "fixing" a broken state.
 func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 	r := &Revision{ExecutedAt: time.Now(), ExecutionState: stateOngoing}
 	if err := e.rrw.WriteRevision(ctx, r); err != nil {
@@ -479,7 +491,8 @@ func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 	return
 }
 
-// ExecuteN executes n pending migration files. If n<=0 all pending migration files are executed.
+// ExecuteN executes n pending migration files. If n<=0 all pending migration files are executed. It will not attempt
+// an execution if the database is not "clean" (has only successfully applied migrations).
 func (e *Executor) ExecuteN(ctx context.Context, n int) error {
 	unlock, err := e.Lock(ctx)
 	if err != nil {


Test patch:
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index 5f9ca7f8..52c52be1 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -9,6 +9,7 @@ import (
 	"database/sql"
 	_ "embed"
 	"errors"
+	"fmt"
 	"io"
 	"io/fs"
 	"os"
@@ -423,7 +424,7 @@ func TestExecutor(t *testing.T) {
 	require.True(t, drv.released())
 
 	// Unknown revision present.
-	*rrw = mockRevisionReadWriter(migrate.Revisions{&migrate.Revision{Version: "unknown"}})
+	*rrw = mockRevisionReadWriter(migrate.Revisions{&migrate.Revision{Version: "unknown", ExecutionState: "ok"}})
 	*drv = lockMockDriver{&mockDriver{}}
 	require.EqualError(t, ex.ExecuteN(context.Background(), 0),
 		`sql/migrate: execute: revisions and migrations mismatch: rev "1.a" <> file "unknown"`,
@@ -434,7 +435,7 @@ func TestExecutor(t *testing.T) {
 	require.True(t, drv.released())
 
 	// More revisions than migration files.
-	*rrw = mockRevisionReadWriter(migrate.Revisions{&migrate.Revision{Version: "unknown"}, rev1, rev2})
+	*rrw = mockRevisionReadWriter(migrate.Revisions{{Version: "unknown", ExecutionState: "ok"}, rev1, rev2})
 	*drv = lockMockDriver{&mockDriver{}}
 	require.EqualError(t, ex.ExecuteN(context.Background(), 0),
 		`sql/migrate: execute: revisions and migrations mismatch: more revisions than migrations`,
@@ -443,6 +444,29 @@ func TestExecutor(t *testing.T) {
 	require.Equal(t, drv.lockCounter, 1)
 	require.Equal(t, drv.unlockCounter, 1)
 	require.True(t, drv.released())
+
+	// Incomplete revisions (error or ongoing) present.
+	rev1.ExecutionState = "error"
+	*rrw = mockRevisionReadWriter(migrate.Revisions{rev1})
+	*drv = lockMockDriver{&mockDriver{}}
+	require.EqualError(t, ex.ExecuteN(context.Background(), 0), fmt.Sprintf(
+		"sql/migrate: execute: dirty migration state: version %q has state %q",
+		rev1.Version, "error",
+	))
+
+	require.Equal(t, drv.lockCounter, 1)
+	require.Equal(t, drv.unlockCounter, 1)
+	require.True(t, drv.released())
+
+	rev1.ExecutionState = "ongoing"
+	require.EqualError(t, ex.ExecuteN(context.Background(), 0), fmt.Sprintf(
+		"sql/migrate: execute: dirty migration state: version %q has state %q",
+		rev1.Version, "ongoing",
+	))
+
+	require.Equal(t, drv.lockCounter, 2)
+	require.Equal(t, drv.unlockCounter, 2)
+	require.True(t, drv.released())
 }
 
 type (