feat(delivery): add automatic rollback state machine for failed deployments

Introduce a compensating workflow that launches the dedicated AWX
rollback template when a deployment cannot be started or fails midway.

- add task states: rollback_pending, rolling_back, rolled_back,
  rollback_failed, rollback_acknowledged
- add RollbackJob model to track the compensating AWX run separately
  from the deploy run, preserving both job IDs for audit
- hold resource reservations in 'rollback' status until cleanup
  succeeds so a failed cleanup cannot be masked by a later delivery
- poll rollback jobs with a 2-minute launch timeout; an unknown launch
  result surfaces as a recoverable failure instead of re-launching
- protect finished/register_failed/rolled-back tasks from rollback;
  register_failed keeps the healthy instance and its resource usage
- add DELIVERY_ROLLBACK_TEMPLATE_ID config; without it, failures are
  marked rollback_failed and require manual cleanup
- use unique pending-<task_id> placeholder for executor job IDs

server/.env.example

@@ -30,6 +30,8 @@ AWX_BASE_URL=
 AWX_TOKEN=
 AWX_USERNAME=
 AWX_PASSWORD=
+# AWX Job Template ID for ansible/mysql-rollback.yml; required for automatic cleanup
+DELIVERY_ROLLBACK_TEMPLATE_ID=0
 DELIVERY_SERVICE_TOKEN=
 CLOUDDM_REGISTER_URL=
 CLOUDDM_API_TOKEN=

server/internal/config/config.go

@@ -63,6 +63,7 @@ type Config struct {
 	AWXToken                  string
 	AWXUsername               string
 	AWXPassword               string
+	RollbackTemplateID        uint64
 	DeliveryServiceToken      string
 	DeliverySchedulerEnabled  bool
 	DeliveryPollSeconds       int
@@ -133,6 +134,7 @@ func Load() Config {
 		AWXToken:                  env("AWX_TOKEN", ""),
 		AWXUsername:               env("AWX_USERNAME", ""),
 		AWXPassword:               env("AWX_PASSWORD", ""),
+		RollbackTemplateID:        uint64(envInt("DELIVERY_ROLLBACK_TEMPLATE_ID", 0)),
 		DeliveryServiceToken:      env("DELIVERY_SERVICE_TOKEN", ""),
 		DeliverySchedulerEnabled:  envBool("DELIVERY_SCHEDULER_ENABLED", false),
 		DeliveryPollSeconds:       envInt("DELIVERY_POLL_SECONDS", 5),

server/internal/database/database.go

@@ -27,6 +27,7 @@ func AutoMigrate(db *gorm.DB) error {
 		&model.MySQLInstance{},
 		&model.ResourceUsage{},
 		&model.ExecutionJob{},
+		&model.RollbackJob{},
 		&model.TaskEvent{},
 	)
 }

server/internal/model/delivery.go

@@ -14,6 +14,11 @@ const (
 	TaskValidationFailed = "validation_failed"
 	TaskCanceling        = "canceling"
 	TaskCanceled         = "canceled"
+	TaskRollbackPending  = "rollback_pending"
+	TaskRollingBack      = "rolling_back"
+	TaskRolledBack       = "rolled_back"
+	TaskRollbackFailed   = "rollback_failed"
+	TaskRollbackAck      = "rollback_acknowledged"
 )
 
 type ResourceQuota struct {
@@ -110,6 +115,20 @@ type ExecutionJob struct {
 	UpdatedAt      time.Time  `json:"updated_at"`
 }
 
+// RollbackJob tracks the compensating AWX run independently from the deploy run.
+// Keeping a separate record preserves both job IDs for audit and retry tooling.
+type RollbackJob struct {
+	ID            uint64     `gorm:"primaryKey" json:"id"`
+	TaskID        string     `gorm:"size:36;not null;uniqueIndex" json:"task_id"`
+	ExecutorJobID string     `gorm:"size:128;not null" json:"executor_job_id"`
+	Status        string     `gorm:"size:32;not null;index" json:"status"`
+	Reason        string     `gorm:"type:text" json:"reason"`
+	StartedAt     *time.Time `json:"started_at,omitempty"`
+	FinishedAt    *time.Time `json:"finished_at,omitempty"`
+	CreatedAt     time.Time  `json:"created_at"`
+	UpdatedAt     time.Time  `json:"updated_at"`
+}
+
 type TaskEvent struct {
 	ID        uint64    `gorm:"primaryKey" json:"id"`
 	TaskID    string    `gorm:"size:36;not null;index" json:"task_id"`

server/internal/service/delivery.go

@@ -310,6 +310,8 @@ var supportedMySQLVersions = map[string]bool{"8.0": true, "8.4": true}
 // 但调度器仍是单主机模型且复制编排未自动化，本期仅放开 standalone。
 var supportedTopologies = map[string]bool{"standalone": true}
 
+const rollbackLaunchTimeout = 2 * time.Minute
+
 var supportedCharsets = map[string]bool{"utf8mb4": true, "utf8": true, "gbk": true, "latin1": true}
 
 // 高级参数档位白名单（与 docs/mysql-parameter-selection.md 保持一致）
@@ -587,7 +589,7 @@ func checkResourceQuota(tx *gorm.DB, businessLineID, targetID uint64, payload de
 	if err := tx.Model(&model.ResourceUsage{}).Select("COALESCE(SUM(cpu_milli),0) cpu, COALESCE(SUM(memory_mi),0) memory, COALESCE(SUM(storage_gi),0) storage, COALESCE(SUM(instance_count),0) instances").Where("business_line_id = ? AND target_id = ? AND status = ?", businessLineID, targetID, "active").Scan(&used).Error; err != nil {
 		return false, err
 	}
-	if err := tx.Model(&model.ResourceReservation{}).Select("COALESCE(SUM(cpu_milli),0) cpu, COALESCE(SUM(memory_mi),0) memory, COALESCE(SUM(storage_gi),0) storage, COALESCE(SUM(instance_count),0) instances").Where("business_line_id = ? AND target_id = ? AND status = ? AND expires_at > ?", businessLineID, targetID, "reserved", time.Now()).Scan(&reserved).Error; err != nil {
+	if err := tx.Model(&model.ResourceReservation{}).Select("COALESCE(SUM(cpu_milli),0) cpu, COALESCE(SUM(memory_mi),0) memory, COALESCE(SUM(storage_gi),0) storage, COALESCE(SUM(instance_count),0) instances").Where("business_line_id = ? AND target_id = ? AND (status = ? OR (status = ? AND expires_at > ?))", businessLineID, targetID, "rollback", "reserved", time.Now()).Scan(&reserved).Error; err != nil {
 		return false, err
 	}
 	return used.CPU+reserved.CPU+payload.CPUMilli <= quota.CPUMilli &&
@@ -614,7 +616,7 @@ func (s *DeliveryService) transitionTx(tx *gorm.DB, task *model.DeliveryTask, st
 	if status == model.TaskRunning {
 		updates["started_at"] = now
 	}
-	if status == model.TaskFinished || status == model.TaskExecutionFailed || status == model.TaskValidationFailed || status == model.TaskCanceled || status == model.TaskRegisterFailed {
+	if status == model.TaskFinished || status == model.TaskExecutionFailed || status == model.TaskValidationFailed || status == model.TaskCanceled || status == model.TaskRegisterFailed || status == model.TaskRolledBack || status == model.TaskRollbackFailed || status == model.TaskRollbackAck {
 		updates["finished_at"] = now
 	}
 	result := tx.Model(&model.DeliveryTask{}).Where("id = ? AND status = ?", task.ID, from).Updates(updates)
@@ -662,7 +664,7 @@ func (s *DeliveryService) DispatchOnce(ctx context.Context) error {
 	}
 	_, _, err = s.CreateExecution(ctx, task.ID, task.PayloadHash, task.IdempotencyKey)
 	if err != nil {
-		return s.failTask(ctx, task, model.TaskExecutionFailed, err.Error())
+		return s.beginRollback(ctx, task.ID, "deployment could not be started: "+err.Error())
 	}
 	return nil
 }
@@ -700,7 +702,7 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa
 	meta := parseTargetMetadata(target.Metadata)
 	// Persist execution record BEFORE launching AWX to ensure crash recovery.
 	now := time.Now()
-	execution := model.ExecutionJob{TaskID: task.ID, IdempotencyKey: task.IdempotencyKey, ExecutorJobID: "pending", Status: "launching", StartedAt: &now}
+	execution := model.ExecutionJob{TaskID: task.ID, IdempotencyKey: task.IdempotencyKey, ExecutorJobID: "pending-" + task.ID, Status: "launching", StartedAt: &now}
 	if err := s.db.WithContext(ctx).Create(&execution).Error; err != nil {
 		return nil, false, err
 	}
@@ -785,8 +787,8 @@ func (s *DeliveryService) PollOnce(ctx context.Context) error {
 	for _, execution := range jobs {
 		job, err := s.awx.GetJob(ctx, execution.ExecutorJobID)
 		if err != nil {
-			s.finishExecution(ctx, &execution, "failed")
-			_ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, "poll AWX job "+execution.ExecutorJobID+": "+err.Error())
+			// A transient AWX/API failure is not evidence that deployment failed.
+			// Keep the job running and retry on the next scheduler tick.
 			continue
 		}
 		switch strings.ToLower(job.Status) {
@@ -795,14 +797,14 @@ func (s *DeliveryService) PollOnce(ctx context.Context) error {
 		case "successful":
 			s.finishExecution(ctx, &execution, "successful")
 			if err := s.completeTask(ctx, execution.TaskID); err != nil {
-				_ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskValidationFailed, err.Error())
+				_ = s.beginRollback(ctx, execution.TaskID, err.Error())
 			}
 		case "canceled":
 			s.finishExecution(ctx, &execution, "canceled")
-			_ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskCanceled, "AWX job was canceled")
+			_ = s.beginRollback(ctx, execution.TaskID, "AWX deployment job was canceled")
 		default:
 			s.finishExecution(ctx, &execution, "failed")
-			_ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, "AWX job finished with status "+job.Status)
+			_ = s.beginRollback(ctx, execution.TaskID, "AWX job finished with status "+job.Status)
 		}
 	}
 	return nil
@@ -849,11 +851,41 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error
 		return err
 	}
 	if err := s.RegisterCloudDM(ctx, task.ID); err != nil {
-		return s.transition(ctx, &task, model.TaskRegisterFailed, "CloudDM registration failed", err.Error())
+		message := fmt.Sprintf("CloudDM registration failed: %v", err)
+		// The database is already healthy and accounted for at this point. A
+		// registration outage must not trigger destructive instance rollback.
+		if transitionErr := s.transition(ctx, &task, model.TaskRegisterFailed,
+			"MySQL delivered; CloudDM registration failed and can be retried", message); transitionErr != nil {
+			return fmt.Errorf("%s; cannot record register_failed: %w", message, transitionErr)
+		}
+		return nil
 	}
 	return s.transition(ctx, &task, model.TaskFinished, "MySQL delivery completed", "")
 }
 
+// RetryCloudDMRegistration retries only the external registration step. The
+// instance and its resource usage remain active throughout the retry.
+func (s *DeliveryService) RetryCloudDMRegistration(ctx context.Context, taskID string) error {
+	var task model.DeliveryTask
+	if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
+		return err
+	}
+	if task.Status != model.TaskRegisterFailed {
+		return fmt.Errorf("task %s is in state %q and cannot retry CloudDM registration", taskID, task.Status)
+	}
+	if err := s.transition(ctx, &task, model.TaskRegistering, "CloudDM registration retry started", ""); err != nil {
+		return err
+	}
+	if err := s.RegisterCloudDM(ctx, taskID); err != nil {
+		message := fmt.Sprintf("CloudDM registration failed: %v", err)
+		if transitionErr := s.transition(ctx, &task, model.TaskRegisterFailed, "CloudDM registration retry failed; instance remains active", message); transitionErr != nil {
+			return fmt.Errorf("%s; cannot restore register_failed: %w", message, transitionErr)
+		}
+		return err
+	}
+	return s.transition(ctx, &task, model.TaskFinished, "CloudDM registration completed", "")
+}
+
 func (s *DeliveryService) RegisterCloudDM(ctx context.Context, taskID string) error {
 	if s.cfg.CloudDMRegisterURL == "" {
 		return nil
@@ -883,6 +915,262 @@ func (s *DeliveryService) RegisterCloudDM(ctx context.Context, taskID string) er
 	return nil
 }
 
+// beginRollback changes the task into the compensating workflow and launches
+// the dedicated AWX rollback template. The deploy reservation is deliberately
+// kept until rollback succeeds so a failed cleanup cannot be silently
+// overwritten by a later delivery.
+func (s *DeliveryService) beginRollback(ctx context.Context, taskID, reason string) error {
+	var task model.DeliveryTask
+	if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
+		return err
+	}
+	if rollbackProtectedStatus(task.Status) {
+		return nil
+	}
+
+	if s.cfg.RollbackTemplateID == 0 {
+		if err := s.transition(ctx, &task, model.TaskRollbackFailed,
+			"automatic rollback is not configured; manual cleanup is required", reason); err != nil {
+			return err
+		}
+		return s.holdReservationForRollback(ctx, task.ID)
+	}
+	if err := s.transition(ctx, &task, model.TaskRollbackPending, "automatic rollback queued", reason); err != nil {
+		return err
+	}
+	if err := s.holdReservationForRollback(ctx, task.ID); err != nil {
+		return s.markRollbackFailed(ctx, task.ID, "cannot hold resources during rollback: "+err.Error())
+	}
+	return s.launchRollbackJob(ctx, &task, reason)
+}
+
+func rollbackProtectedStatus(status string) bool {
+	switch status {
+	case model.TaskRollbackPending, model.TaskRollingBack, model.TaskRolledBack, model.TaskRollbackFailed,
+		model.TaskRollbackAck, model.TaskRegisterFailed, model.TaskFinished, model.TaskCanceled:
+		return true
+	default:
+		return false
+	}
+}
+
+func (s *DeliveryService) launchRollbackJob(ctx context.Context, task *model.DeliveryTask, reason string) error {
+	target, err := s.getTarget(ctx, task.TargetID)
+	if err != nil {
+		return s.markRollbackFailed(ctx, task.ID, "rollback target is unavailable: "+err.Error())
+	}
+	var payload deliveryPayload
+	if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
+		return s.markRollbackFailed(ctx, task.ID, "rollback payload is invalid: "+err.Error())
+	}
+	now := time.Now()
+	var rollback model.RollbackJob
+	if err := s.db.WithContext(ctx).Where("task_id = ?", task.ID).First(&rollback).Error; errors.Is(err, gorm.ErrRecordNotFound) {
+		rollback = model.RollbackJob{TaskID: task.ID}
+	} else if err != nil {
+		return s.markRollbackFailed(ctx, task.ID, "cannot load rollback job: "+err.Error())
+	}
+	rollback.ExecutorJobID = "pending-rollback-" + task.ID
+	rollback.Status = "launching"
+	rollback.Reason = reason
+	rollback.StartedAt = &now
+	rollback.FinishedAt = nil
+	if err := s.db.WithContext(ctx).Save(&rollback).Error; err != nil {
+		return s.markRollbackFailed(ctx, task.ID, "cannot persist rollback job: "+err.Error())
+	}
+	job, err := s.awx.Launch(ctx, s.cfg.RollbackTemplateID, AWXLaunchRequest{
+		InventoryID: target.AWXInventoryID,
+		Limit:       task.TargetHost,
+		ExtraVars:   rollbackExtraVars(task, payload),
+	})
+	if err != nil {
+		_ = s.db.WithContext(ctx).Model(&rollback).Updates(map[string]any{"status": "failed", "finished_at": time.Now()}).Error
+		return s.markRollbackFailed(ctx, task.ID, "cannot launch rollback AWX job: "+err.Error())
+	}
+	if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var current model.DeliveryTask
+		if err := tx.First(&current, "id = ?", task.ID).Error; err != nil {
+			return err
+		}
+		if current.Status != model.TaskRollbackPending {
+			return fmt.Errorf("task %s is no longer pending rollback (state %q)", task.ID, current.Status)
+		}
+		if err := tx.Model(&rollback).Updates(map[string]any{"executor_job_id": fmt.Sprint(job.ID), "status": "running"}).Error; err != nil {
+			return err
+		}
+		return s.transitionTx(tx, &current, model.TaskRollingBack, "rollback AWX job started", reason)
+	}); err != nil {
+		return s.markRollbackFailed(ctx, task.ID, "cannot record rollback AWX job: "+err.Error())
+	}
+	return nil
+}
+
+func (s *DeliveryService) RetryRollback(ctx context.Context, taskID string) error {
+	var task model.DeliveryTask
+	if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
+		return err
+	}
+	if task.Status != model.TaskRollbackFailed {
+		return fmt.Errorf("task %s is in state %q and cannot retry rollback", taskID, task.Status)
+	}
+	if s.cfg.RollbackTemplateID == 0 {
+		return fmt.Errorf("automatic rollback is not configured")
+	}
+	reason := "manual rollback retry requested"
+	if err := s.transition(ctx, &task, model.TaskRollbackPending, "manual rollback retry queued", reason); err != nil {
+		return err
+	}
+	if err := s.holdReservationForRollback(ctx, task.ID); err != nil {
+		_ = s.markRollbackFailed(ctx, task.ID, "cannot hold resources during rollback retry: "+err.Error())
+		return err
+	}
+	return s.launchRollbackJob(ctx, &task, reason)
+}
+
+// AcknowledgeRollbackRelease is an explicit platform-admin escape hatch after
+// the operator has verified the target host. It releases bookkeeping only; it
+// does not claim that the automatic rollback succeeded.
+func (s *DeliveryService) AcknowledgeRollbackRelease(ctx context.Context, taskID string) error {
+	now := time.Now()
+	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var task model.DeliveryTask
+		if err := tx.First(&task, "id = ?", taskID).Error; err != nil {
+			return err
+		}
+		if task.Status != model.TaskRollbackFailed {
+			return fmt.Errorf("task %s is in state %q and cannot acknowledge rollback release", taskID, task.Status)
+		}
+		if err := tx.Model(&model.MySQLInstance{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "rollback_acknowledged", "updated_at": now}).Error; err != nil {
+			return err
+		}
+		if err := tx.Model(&model.ResourceUsage{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "released", "released_at": now, "updated_at": now}).Error; err != nil {
+			return err
+		}
+		if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", taskID, "rollback").Update("status", "released").Error; err != nil {
+			return err
+		}
+		return s.transitionTx(tx, &task, model.TaskRollbackAck, "rollback release acknowledged by platform admin", "")
+	})
+}
+
+func rollbackExtraVars(task *model.DeliveryTask, payload deliveryPayload) map[string]any {
+	return map[string]any{
+		"target_hosts":  task.TargetHost,
+		"instance_name": payload.InstanceName,
+		"data_disk":     payload.DataDisk,
+		"task_id":       task.ID,
+		"rollback":      true,
+	}
+}
+
+func (s *DeliveryService) holdReservationForRollback(ctx context.Context, taskID string) error {
+	return s.db.WithContext(ctx).Model(&model.ResourceReservation{}).
+		Where("task_id = ? AND status IN ?", taskID, []string{"reserved", "consumed"}).
+		Update("status", "rollback").Error
+}
+
+func (s *DeliveryService) markRollbackFailed(ctx context.Context, taskID, message string) error {
+	var task model.DeliveryTask
+	if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
+		return err
+	}
+	if task.Status == model.TaskRolledBack || task.Status == model.TaskFinished || task.Status == model.TaskCanceled ||
+		task.Status == model.TaskRollbackFailed || task.Status == model.TaskRollbackAck || task.Status == model.TaskRegisterFailed {
+		return nil
+	}
+	return s.transition(ctx, &task, model.TaskRollbackFailed, "automatic rollback failed; manual cleanup is required", message)
+}
+
+func (s *DeliveryService) PollRollbackOnce(ctx context.Context) error {
+	var jobs []model.RollbackJob
+	if err := s.db.WithContext(ctx).Where("status IN ?", []string{"launching", "running"}).Find(&jobs).Error; err != nil {
+		return err
+	}
+	for _, rollback := range jobs {
+		if rollback.Status == "launching" {
+			if !rollbackLaunchExpired(rollback, time.Now()) {
+				continue
+			}
+			// The AWX launch result is unknown. Do not blindly launch a second
+			// job; surface a recoverable failure for an explicit admin retry.
+			expired, err := s.expireRollbackLaunch(ctx, rollback.ID)
+			if err != nil {
+				return err
+			}
+			if !expired {
+				continue
+			}
+			_ = s.markRollbackFailed(ctx, rollback.TaskID, "rollback launch timed out before its AWX job ID was recorded; manual retry is required")
+			continue
+		}
+		job, err := s.awx.GetJob(ctx, rollback.ExecutorJobID)
+		if err != nil {
+			// Do not declare cleanup failed because AWX is temporarily unreachable.
+			continue
+		}
+		switch strings.ToLower(job.Status) {
+		case "pending", "waiting", "running", "new":
+			continue
+		case "successful":
+			_ = s.finishRollbackJob(ctx, &rollback, "successful")
+			if err := s.completeRollback(ctx, rollback.TaskID); err != nil {
+				_ = s.markRollbackFailed(ctx, rollback.TaskID, err.Error())
+			}
+		case "canceled":
+			_ = s.finishRollbackJob(ctx, &rollback, "canceled")
+			_ = s.markRollbackFailed(ctx, rollback.TaskID, "rollback AWX job was canceled")
+		default:
+			_ = s.finishRollbackJob(ctx, &rollback, "failed")
+			_ = s.markRollbackFailed(ctx, rollback.TaskID, "rollback AWX job finished with status "+job.Status)
+		}
+	}
+	return nil
+}
+
+func rollbackLaunchExpired(rollback model.RollbackJob, now time.Time) bool {
+	if rollback.Status != "launching" {
+		return false
+	}
+	if rollback.StartedAt == nil {
+		return true
+	}
+	return !now.Before(rollback.StartedAt.Add(rollbackLaunchTimeout))
+}
+
+func (s *DeliveryService) expireRollbackLaunch(ctx context.Context, rollbackID uint64) (bool, error) {
+	result := s.db.WithContext(ctx).Model(&model.RollbackJob{}).
+		Where("id = ? AND status = ?", rollbackID, "launching").
+		Updates(map[string]any{"status": "launch_timeout", "finished_at": time.Now()})
+	return result.RowsAffected == 1, result.Error
+}
+
+func (s *DeliveryService) finishRollbackJob(ctx context.Context, rollback *model.RollbackJob, status string) error {
+	return s.db.WithContext(ctx).Model(rollback).Updates(map[string]any{"status": status, "finished_at": time.Now()}).Error
+}
+
+func (s *DeliveryService) completeRollback(ctx context.Context, taskID string) error {
+	now := time.Now()
+	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var task model.DeliveryTask
+		if err := tx.First(&task, "id = ?", taskID).Error; err != nil {
+			return err
+		}
+		if task.Status != model.TaskRollingBack {
+			return fmt.Errorf("task %s is in state %q, cannot complete rollback", taskID, task.Status)
+		}
+		if err := tx.Model(&model.MySQLInstance{}).Where("task_id = ?", taskID).Updates(map[string]any{"status": "rolled_back", "updated_at": now}).Error; err != nil {
+			return err
+		}
+		if err := tx.Model(&model.ResourceUsage{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "released", "released_at": now, "updated_at": now}).Error; err != nil {
+			return err
+		}
+		if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status IN ?", taskID, []string{"reserved", "consumed", "rollback"}).Update("status", "released").Error; err != nil {
+			return err
+		}
+		return s.transitionTx(tx, &task, model.TaskRolledBack, "MySQL instance rollback completed", "")
+	})
+}
+
 func (s *DeliveryService) failTask(ctx context.Context, task *model.DeliveryTask, status, message string) error {
 	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		var current model.DeliveryTask
@@ -916,6 +1204,7 @@ func (s *DeliveryService) Run(ctx context.Context) {
 		case <-ticker.C:
 			_ = s.DispatchOnce(ctx)
 			_ = s.PollOnce(ctx)
+			_ = s.PollRollbackOnce(ctx)
 		}
 	}
 }

server/internal/service/delivery_test.go

@@ -1,6 +1,11 @@
 package service
 
-import "testing"
+import (
+	"testing"
+	"time"
+
+	"github.com/1024XEngineer/xinfra/server/internal/model"
+)
 
 func intPtr(v int) *int { return &v }
 
@@ -124,3 +129,40 @@ func TestAllocatePort(t *testing.T) {
 		t.Fatal("exhausted pool still allocated a port")
 	}
 }
+
+func TestRollbackExtraVarsTargetsOnlyTheAllocatedInstance(t *testing.T) {
+	task := &model.DeliveryTask{ID: "task-1", TargetHost: "db-01"}
+	payload := deliveryPayload{MySQLDeliveryInput: MySQLDeliveryInput{InstanceName: "mysql-a", DataDisk: "/disk1"}}
+	vars := rollbackExtraVars(task, payload)
+	if vars["target_hosts"] != "db-01" || vars["instance_name"] != "mysql-a" || vars["data_disk"] != "/disk1" {
+		t.Fatalf("rollback vars target the wrong instance: %#v", vars)
+	}
+	if vars["rollback"] != true {
+		t.Fatalf("rollback marker missing: %#v", vars)
+	}
+}
+
+func TestRegisterFailedIsProtectedFromRollback(t *testing.T) {
+	if !rollbackProtectedStatus(model.TaskRegisterFailed) {
+		t.Fatal("register_failed must preserve the healthy instance and resource usage")
+	}
+	if rollbackProtectedStatus(model.TaskValidationFailed) {
+		t.Fatal("validation_failed must still be eligible for cleanup rollback")
+	}
+}
+
+func TestRollbackLaunchExpired(t *testing.T) {
+	now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC)
+	started := now.Add(-rollbackLaunchTimeout - time.Second)
+	if !rollbackLaunchExpired(model.RollbackJob{Status: "launching", StartedAt: &started}, now) {
+		t.Fatal("stale launching rollback job must be recoverable")
+	}
+	if rollbackLaunchExpired(model.RollbackJob{Status: "launching", StartedAt: ptrTime(now.Add(-rollbackLaunchTimeout + time.Second))}, now) {
+		t.Fatal("recent launching rollback job must remain pending")
+	}
+	if rollbackLaunchExpired(model.RollbackJob{Status: "running", StartedAt: &started}, now) {
+		t.Fatal("running rollback job is not a launch timeout")
+	}
+}
+
+func ptrTime(v time.Time) *time.Time { return &v }