fix(migration): stop a half-applied startup migration from committing silently (#6182)

* fix(traffic): check maintenance commits and IP-limit errors

* fix(migrations): propagate transactional failures

---------

Co-authored-by: n0ctal <293235942+n0ctal@users.noreply.github.com>

internal/web/service/inbound_migration.go

@@ -33,13 +33,15 @@ func (s *InboundService) MigrationRemoveOrphanedTraffics() {
 	}
 }
 
-func (s *InboundService) MigrationRequirements() {
+func (s *InboundService) MigrationRequirements() (err error) {
 	db := database.GetDB()
 	tx := db.Begin()
-	var err error
 	defer func() {
 		if err == nil {
-			tx.Commit()
+			if commitErr := tx.Commit().Error; commitErr != nil {
+				err = commitErr
+				return
+			}
 			if !database.IsPostgres() {
 				if dbErr := db.Exec(`VACUUM "main"`).Error; dbErr != nil {
 					logger.Warningf("VACUUM failed: %v", dbErr)
@@ -76,8 +78,8 @@ func (s *InboundService) MigrationRequirements() {
 	// SQLite (no PG :: casts).
 	if database.IsPostgres() {
 		// Use DO block so it is idempotent and doesn't fail if already boolean.
-		normalizeBool := func(table, col string) {
-			tx.Exec(fmt.Sprintf(`
+		normalizeBool := func(table, col string) error {
+			return tx.Exec(fmt.Sprintf(`
 				DO $$
 				BEGIN
 					IF EXISTS (
@@ -88,14 +90,13 @@ func (s *InboundService) MigrationRequirements() {
 						ALTER TABLE %s ALTER COLUMN %s
 							TYPE boolean USING (CASE WHEN %s::text IN ('1','true','t','yes') THEN true ELSE false END);
 					END IF;
-				END $$;`, table, col, table, col, col))
+				END $$;`, table, col, table, col, col)).Error
+		}
+		for _, column := range [][2]string{{"inbounds", "enable"}, {"client_traffics", "enable"}, {"nodes", "enable"}, {"clients", "enable"}, {"api_tokens", "enabled"}, {"outbound_subscriptions", "enabled"}} {
+			if err = normalizeBool(column[0], column[1]); err != nil {
+				return
+			}
 		}
-		normalizeBool("inbounds", "enable")
-		normalizeBool("client_traffics", "enable")
-		normalizeBool("nodes", "enable")
-		normalizeBool("clients", "enable")
-		normalizeBool("api_tokens", "enabled")
-		normalizeBool("outbound_subscriptions", "enabled")
 	}
 
 	// Fix inbounds based problems
@@ -160,7 +161,8 @@ func (s *InboundService) MigrationRequirements() {
 				delete(settings, "testseed")
 			}
 
-			modifiedSettings, err := json.MarshalIndent(settings, "", "  ")
+			var modifiedSettings []byte
+			modifiedSettings, err = json.MarshalIndent(settings, "", "  ")
 			if err != nil {
 				return
 			}
@@ -169,30 +171,39 @@ func (s *InboundService) MigrationRequirements() {
 		}
 
 		// Add client traffic row for all clients which has email
-		modelClients, err := s.GetClients(inbounds[inbound_index])
+		var modelClients []model.Client
+		modelClients, err = s.GetClients(inbounds[inbound_index])
 		if err != nil {
 			return
 		}
 		for _, modelClient := range modelClients {
 			if len(modelClient.Email) > 0 {
 				var count int64
-				tx.Model(xray.ClientTraffic{}).Where("email = ?", modelClient.Email).Count(&count)
+				if err = tx.Model(xray.ClientTraffic{}).Where("email = ?", modelClient.Email).Count(&count).Error; err != nil {
+					return
+				}
 				if count == 0 {
-					_ = s.AddClientStat(tx, inbounds[inbound_index].Id, &modelClient)
+					if err = s.AddClientStat(tx, inbounds[inbound_index].Id, &modelClient); err != nil {
+						return
+					}
 				}
 			}
 		}
 
 		// Heal clients table for installs where the one-shot seeder
 		// skipped clients due to a tgId-string unmarshal error.
-		if syncErr := s.clientService.SyncInbound(tx, inbounds[inbound_index].Id, modelClients); syncErr != nil {
-			logger.Warning("MigrationRequirements sync clients failed:", syncErr)
+		if err = s.clientService.SyncInbound(tx, inbounds[inbound_index].Id, modelClients); err != nil {
+			return
 		}
 	}
-	tx.Save(inbounds)
+	if err = tx.Save(inbounds).Error; err != nil {
+		return
+	}
 
 	// Remove orphaned traffics
-	tx.Where("inbound_id = 0").Delete(xray.ClientTraffic{})
+	if err = tx.Where("inbound_id = 0").Delete(xray.ClientTraffic{}).Error; err != nil {
+		return
+	}
 
 	// Migrate old MultiDomain to External Proxy
 	var externalProxy []struct {
@@ -238,8 +249,14 @@ func (s *InboundService) MigrationRequirements() {
 			}
 		}
 		stream["externalProxy"] = reverses
-		newStream, _ := json.MarshalIndent(stream, " ", "  ")
-		tx.Model(model.Inbound{}).Where("id = ?", ep.Id).Update("stream_settings", newStream)
+		newStream, marshalErr := json.MarshalIndent(stream, " ", "  ")
+		if marshalErr != nil {
+			err = marshalErr
+			return
+		}
+		if err = tx.Model(model.Inbound{}).Where("id = ?", ep.Id).Update("stream_settings", newStream).Error; err != nil {
+			return
+		}
 	}
 
 	// Legacy tag cleanup for old auto-generated tags (e.g. "0.0.0.0:443-...").
@@ -256,10 +273,13 @@ func (s *InboundService) MigrationRequirements() {
 	if err != nil {
 		return
 	}
+	return err
 }
 
 func (s *InboundService) MigrateDB() {
-	s.MigrationRequirements()
+	if err := s.MigrationRequirements(); err != nil {
+		logger.Errorf("MigrationRequirements failed: %v", err)
+	}
 	s.MigrationRemoveOrphanedTraffics()
 	s.MigrationRestoreVisionFlow()
 }