refactor: prevent concurrent migration (#7311)

* refactor: prevent concurrent migration

* fix: python test

Author: klesh

backend/server/api/api.go

@@ -49,6 +49,7 @@ WARNING: Performing migration may wipe collected data for consistency and re-col
 To proceed, please send a request to <config-ui-endpoint>/api/proceed-db-migration (or <devlake-endpoint>/proceed-db-migration).
 Alternatively, you may downgrade back to the previous DevLake version.
 `
+const DB_MIGRATING = `Database migration is in progress. Please wait until it is completed.`
 
 var basicRes context.BasicRes
 
@@ -95,12 +96,6 @@ func SetupApiServer(router *gin.Engine) {
 
 	// Endpoint to proceed database migration
 	router.GET("/proceed-db-migration", func(ctx *gin.Context) {
-		// Check if migration requires confirmation
-		if !services.MigrationRequireConfirmation() {
-			// Return success response
-			shared.ApiOutputSuccess(ctx, nil, http.StatusOK)
-			return
-		}
 		// Execute database migration
 		err := services.ExecuteMigration()
 		if err != nil {
@@ -114,15 +109,22 @@ func SetupApiServer(router *gin.Engine) {
 
 	// Restrict access if database migration is required
 	router.Use(func(ctx *gin.Context) {
-		if !services.MigrationRequireConfirmation() {
-			return
+		serviceStatus := services.CurrentStatus()
+		if serviceStatus == services.SERVICE_STATUS_WAIT_CONFIRM {
+			// Return error response
+			shared.ApiOutputError(
+				ctx,
+				errors.HttpStatus(http.StatusPreconditionRequired).New(DB_MIGRATION_REQUIRED),
+			)
+			ctx.Abort()
+		} else if serviceStatus == services.SERVICE_STATUS_MIGRATING {
+			// Return error response
+			shared.ApiOutputError(
+				ctx,
+				errors.HttpStatus(http.StatusPreconditionRequired).New(DB_MIGRATING),
+			)
+			ctx.Abort()
 		}
-		// Return error response
-		shared.ApiOutputError(
-			ctx,
-			errors.HttpStatus(http.StatusPreconditionRequired).New(DB_MIGRATION_REQUIRED),
-		)
-		ctx.Abort()
 	})
 
 	// Add swagger handlers

backend/server/services/init.go

@@ -18,6 +18,7 @@ limitations under the License.
 package services
 
 import (
+	"sync"
 	"time"
 
 	"github.com/apache/incubator-devlake/core/config"
@@ -84,6 +85,7 @@ func GetMigrator() plugin.Migrator {
 }
 
 // Init the services module
+// Should not be called concurrently
 func Init() {
 	InitResources()
 
@@ -115,9 +117,21 @@ func Init() {
 	}
 }
 
+var statusLock sync.Mutex
+
 // ExecuteMigration executes all pending migration scripts and initialize services module
+// This might be called concurrently across multiple API requests
 func ExecuteMigration() errors.Error {
+	statusLock.Lock()
+	defer statusLock.Unlock()
+	if serviceStatus == SERVICE_STATUS_MIGRATING {
+		return errors.BadInput.New("already migrating")
+	}
+	if serviceStatus == SERVICE_STATUS_READY {
+		return nil
+	}
 	serviceStatus = SERVICE_STATUS_MIGRATING
+	statusLock.Unlock() // unlock to allow other API requests to check the status
 	// apply all pending migration scripts
 	err := migrator.Execute()
 	if err != nil {
@@ -130,11 +144,11 @@ func ExecuteMigration() errors.Error {
 
 	// initialize pipeline server, mainly to start the pipeline consuming process
 	pipelineServiceInit()
+	statusLock.Lock()
 	serviceStatus = SERVICE_STATUS_READY
 	return nil
 }
 
-// MigrationRequireConfirmation returns if there were migration scripts waiting to be executed
-func MigrationRequireConfirmation() bool {
-	return migrator.HasPendingScripts()
+func CurrentStatus() string {
+	return serviceStatus
 }