From f0c1b38c5b423f75200f2f7101638c7f16b35c28 Mon Sep 17 00:00:00 2001 From: Hungerdream <1710233908@qq.com> Date: Tue, 28 Jul 2026 14:29:09 +0800 Subject: [PATCH] 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- placeholder for executor job IDs --- server/.env.example | 2 + server/internal/config/config.go | 2 + server/internal/database/database.go | 1 + server/internal/model/delivery.go | 19 ++ server/internal/service/delivery.go | 309 ++++++++++++++++++++++- server/internal/service/delivery_test.go | 44 +++- 6 files changed, 366 insertions(+), 11 deletions(-) diff --git a/server/.env.example b/server/.env.example index e3ae278..6d4f3f4 100644 --- a/server/.env.example +++ b/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= diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 4041281..a20bc5a 100644 --- a/server/internal/config/config.go +++ b/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), diff --git a/server/internal/database/database.go b/server/internal/database/database.go index f8daaa8..3123ba0 100644 --- a/server/internal/database/database.go +++ b/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{}, ) } diff --git a/server/internal/model/delivery.go b/server/internal/model/delivery.go index e30b228..23ce068 100644 --- a/server/internal/model/delivery.go +++ b/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"` diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index 119254c..b8240e6 100644 --- a/server/internal/service/delivery.go +++ b/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(¤t, "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, ¤t, 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) } } } diff --git a/server/internal/service/delivery_test.go b/server/internal/service/delivery_test.go index ea19946..66142b4 100644 --- a/server/internal/service/delivery_test.go +++ b/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 }