From e11bc6f3d6123f8eaa68787c6b3f95ac80bc6de9 Mon Sep 17 00:00:00 2001 From: mac Date: Thu, 23 Jul 2026 12:04:37 +0800 Subject: [PATCH] feat(delivery): implement delivery scheduler and state machine Core delivery service with: task creation (idempotent, validated), background scheduler (claimAndReserve with SELECT FOR UPDATE SKIP LOCKED), 4-level concurrency control, resource quota enforcement, host allocation from candidate pool, AWX job dispatch, polling, TCP health check, instance registration, CloudDM integration (optional), and cancel support. 11-state machine with optimistic locking and full event audit trail. Relates-to: #97 --- server/internal/service/delivery.go | 565 +++++++++++++++++++++++ server/internal/service/delivery_test.go | 27 ++ 2 files changed, 592 insertions(+) create mode 100644 server/internal/service/delivery.go create mode 100644 server/internal/service/delivery_test.go diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go new file mode 100644 index 0000000..5ecdb9f --- /dev/null +++ b/server/internal/service/delivery.go @@ -0,0 +1,565 @@ +package service + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "regexp" + "strings" + "sync" + "time" + + "github.com/1024XEngineer/xinfra/server/internal/config" + "github.com/1024XEngineer/xinfra/server/internal/model" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +var dnsLabelPattern = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$`) + +type MySQLDeliveryInput struct { + BusinessLineID uint64 `json:"business_line_id" binding:"required"` + TargetID uint64 `json:"target_id" binding:"required"` + Namespace string `json:"namespace" binding:"required"` + InstanceName string `json:"instance_name" binding:"required"` + MySQLVersion string `json:"mysql_version"` + CPUMilli int64 `json:"cpu_milli" binding:"required"` + MemoryMi int64 `json:"memory_mi" binding:"required"` + StorageGi int64 `json:"storage_gi" binding:"required"` +} + +type deliveryPayload struct { + MySQLDeliveryInput + TargetType string `json:"target_type"` +} + +// targetMetadata describes the native VM候选节点池以及部署形态，存储在 DeploymentTarget.Metadata (JSON)。 +type targetMetadata struct { + Topology string `json:"topology"` + MySQLPort int `json:"mysql_port"` + Hosts []targetHost `json:"hosts"` +} + +type targetHost struct { + Name string `json:"name"` + IP string `json:"ip"` +} + +func parseTargetMetadata(raw string) targetMetadata { + meta := targetMetadata{} + if strings.TrimSpace(raw) != "" { + _ = json.Unmarshal([]byte(raw), &meta) + } + if meta.Topology == "" { + meta.Topology = "standalone" + } + if meta.MySQLPort == 0 { + meta.MySQLPort = 3307 + } + return meta +} + +// firstFreeHost 返回候选池中第一个未被占用的节点。 +func firstFreeHost(hosts []targetHost, occupied []string) *targetHost { + taken := make(map[string]bool, len(occupied)) + for _, h := range occupied { + taken[h] = true + } + for i := range hosts { + if !taken[hosts[i].Name] { + return &hosts[i] + } + } + return nil +} + +type DeliveryService struct { + db *gorm.DB + cfg config.Config + awx *AWXClient + audit *AuditService + executionMu sync.Mutex +} + +func (s *DeliveryService) DB() *gorm.DB { return s.db } + +func NewDeliveryService(cfg config.Config, db *gorm.DB, audit *AuditService) *DeliveryService { + return &DeliveryService{db: db, cfg: cfg, awx: NewAWXClient(cfg.AWXBaseURL, cfg.AWXToken), audit: audit} +} + +func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin bool, idempotencyKey string, input MySQLDeliveryInput) (*model.DeliveryTask, bool, error) { + idempotencyKey = strings.TrimSpace(idempotencyKey) + if idempotencyKey == "" || len(idempotencyKey) > 128 { + return nil, false, fmt.Errorf("Idempotency-Key header is required and must not exceed 128 characters") + } + if err := validateDeliveryInput(input); err != nil { + return nil, false, err + } + + var existing model.DeliveryTask + if err := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; err == nil { + return &existing, true, nil + } else if !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, false, err + } + + var target model.DeploymentTarget + if err := s.db.WithContext(ctx).First(&target, "id = ? AND enabled = ?", input.TargetID, true).Error; err != nil { + return nil, false, fmt.Errorf("deployment target is unavailable: %w", err) + } + if target.TargetType != "k8s" { + return nil, false, fmt.Errorf("target type %q is not supported in the MVP", target.TargetType) + } + if !isAdmin { + var count int64 + if err := s.db.WithContext(ctx).Model(&model.BusinessLineUser{}). + Where("business_line_id = ? AND user_id = ?", input.BusinessLineID, userID).Count(&count).Error; err != nil { + return nil, false, err + } + if count == 0 { + return nil, false, fmt.Errorf("user is not authorized for this business line") + } + } + if input.MySQLVersion == "" { + input.MySQLVersion = "8.0" + } + payload := deliveryPayload{MySQLDeliveryInput: input, TargetType: target.TargetType} + raw, err := json.Marshal(payload) + if err != nil { + return nil, false, err + } + digest := sha256.Sum256(raw) + task := model.DeliveryTask{ + ID: randomUUID(), + BusinessLineID: input.BusinessLineID, + RequestedBy: userID, + TargetType: target.TargetType, + TargetID: target.ID, + Namespace: input.Namespace, + InstanceName: input.InstanceName, + Status: model.TaskPending, + ImmutablePayload: string(raw), + PayloadHash: hex.EncodeToString(digest[:]), + IdempotencyKey: idempotencyKey, + } + if err := s.db.WithContext(ctx).Create(&task).Error; err != nil { + if lookupErr := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; lookupErr == nil { + return &existing, true, nil + } + return nil, false, err + } + _ = s.db.WithContext(ctx).Create(&model.TaskEvent{TaskID: task.ID, ToState: model.TaskPending, Message: "delivery task created"}).Error + return &task, false, nil +} + +func validateDeliveryInput(input MySQLDeliveryInput) error { + if len(input.Namespace) > 63 || !dnsLabelPattern.MatchString(input.Namespace) { + return fmt.Errorf("namespace must be a valid Kubernetes DNS label") + } + if len(input.InstanceName) > 63 || !dnsLabelPattern.MatchString(input.InstanceName) { + return fmt.Errorf("instance_name must be a valid Kubernetes DNS label") + } + if input.CPUMilli < 100 || input.CPUMilli > 64000 || input.MemoryMi < 256 || input.MemoryMi > 262144 || input.StorageGi < 1 || input.StorageGi > 16384 { + return fmt.Errorf("requested resources are outside the supported range") + } + return nil +} + +func (s *DeliveryService) ListTasks(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]model.DeliveryTask, error) { + query := s.db.WithContext(ctx).Order("created_at DESC") + if businessLineID != 0 { + query = query.Where("business_line_id = ?", businessLineID) + } + if !isAdmin { + query = query.Where("business_line_id IN (?)", s.db.Model(&model.BusinessLineUser{}).Select("business_line_id").Where("user_id = ?", userID)) + } + var tasks []model.DeliveryTask + return tasks, query.Find(&tasks).Error +} + +func (s *DeliveryService) GetTask(ctx context.Context, taskID string, userID uint64, isAdmin bool) (*model.DeliveryTask, []model.TaskEvent, error) { + query := s.db.WithContext(ctx).Where("id = ?", taskID) + if !isAdmin { + query = query.Where("business_line_id IN (?)", s.db.Model(&model.BusinessLineUser{}).Select("business_line_id").Where("user_id = ?", userID)) + } + var task model.DeliveryTask + if err := query.First(&task).Error; err != nil { + return nil, nil, err + } + var events []model.TaskEvent + if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).Order("id ASC").Find(&events).Error; err != nil { + return nil, nil, err + } + return &task, events, nil +} + +func (s *DeliveryService) Cancel(ctx context.Context, taskID string, userID uint64, isAdmin bool) error { + task, _, err := s.GetTask(ctx, taskID, userID, isAdmin) + if err != nil { + return err + } + if task.Status == model.TaskPending { + return s.transition(ctx, task, model.TaskCanceled, "canceled before dispatch", "") + } + if task.Status != model.TaskRunning && task.Status != model.TaskDispatching { + return fmt.Errorf("task in state %q cannot be canceled", task.Status) + } + var job model.ExecutionJob + if err := s.db.WithContext(ctx).Where("task_id = ?", task.ID).First(&job).Error; err != nil { + return err + } + if err := s.awx.Cancel(ctx, job.ExecutorJobID); err != nil { + return err + } + return s.transition(ctx, task, model.TaskCanceling, "cancel requested in AWX", "") +} + +func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryTask, error) { + var task model.DeliveryTask + var payload deliveryPayload + var target model.DeploymentTarget + dispatchable := false + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).Where("status = ?", model.TaskPending).Order("created_at ASC").First(&task).Error; err != nil { + return err + } + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&target, "id = ? AND enabled = ?", task.TargetID, true).Error; err != nil { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, "deployment target is unavailable") + } + if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, "stored deployment payload is invalid") + } + activeStates := []string{model.TaskValidating, model.TaskDispatching, model.TaskRunning, model.TaskCanceling} + checks := []struct { + query string + args []any + limit int + message string + }{ + {"status IN ?", []any{activeStates}, s.cfg.DeliveryGlobalLimit, "global concurrency limit reached"}, + {"status IN ? AND target_id = ?", []any{activeStates, task.TargetID}, s.cfg.DeliveryTargetLimit, "target concurrency limit reached"}, + {"status IN ? AND business_line_id = ?", []any{activeStates, task.BusinessLineID}, s.cfg.DeliveryBusinessLimit, "business line concurrency limit reached"}, + {"status IN ? AND target_id = ? AND namespace = ?", []any{activeStates, task.TargetID, task.Namespace}, 1, "namespace already has an active MySQL delivery"}, + } + for _, check := range checks { + var count int64 + if check.limit > 0 { + if err := tx.Model(&model.DeliveryTask{}).Where(check.query, check.args...).Count(&count).Error; err != nil { + return err + } + if count >= int64(check.limit) { + return fmt.Errorf("defer: %s", check.message) + } + } + } + var quota model.ResourceQuota + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("business_line_id = ? AND target_id = ?", task.BusinessLineID, task.TargetID).First("a).Error; err != nil { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, "resource quota is not configured") + } + type totals struct{ CPU, Memory, Storage, Instances int64 } + var used, reserved totals + 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 = ?", task.BusinessLineID, task.TargetID, "active").Scan(&used).Error; err != nil { + return 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 = ?", task.BusinessLineID, task.TargetID, "reserved").Scan(&reserved).Error; err != nil { + return err + } + if used.CPU+reserved.CPU+payload.CPUMilli > quota.CPUMilli || used.Memory+reserved.Memory+payload.MemoryMi > quota.MemoryMi || used.Storage+reserved.Storage+payload.StorageGi > quota.StorageGi || used.Instances+reserved.Instances+1 > quota.InstanceLimit { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, "resource quota is insufficient") + } + meta := parseTargetMetadata(target.Metadata) + if len(meta.Hosts) == 0 { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, "deployment target has no candidate hosts") + } + var occupied []string + occupiedExclude := []string{model.TaskExecutionFailed, model.TaskValidationFailed, model.TaskCanceled} + if err := tx.Model(&model.DeliveryTask{}).Where("target_id = ? AND target_host <> ? AND status NOT IN ?", task.TargetID, "", occupiedExclude).Pluck("target_host", &occupied).Error; err != nil { + return err + } + host := firstFreeHost(meta.Hosts, occupied) + if host == nil { + return fmt.Errorf("defer: no free host available on target") + } + reservation := model.ResourceReservation{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, InstanceCount: 1, Status: "reserved", ExpiresAt: time.Now().Add(time.Duration(s.cfg.ReservationTTLMinutes) * time.Minute)} + if err := tx.Create(&reservation).Error; err != nil { + return err + } + if err := tx.Model(&model.DeliveryTask{}).Where("id = ?", task.ID).Updates(map[string]any{"target_host": host.Name, "target_host_ip": host.IP, "mysql_port": meta.MySQLPort}).Error; err != nil { + return err + } + task.TargetHost = host.Name + task.TargetHostIP = host.IP + task.MySQLPort = meta.MySQLPort + if err := s.transitionTx(tx, &task, model.TaskDispatching, "resources reserved", ""); err != nil { + return err + } + dispatchable = true + return nil + }) + if err == nil && !dispatchable { + err = gorm.ErrRecordNotFound + } + return &task, err +} + +func (s *DeliveryService) failInTransaction(tx *gorm.DB, task *model.DeliveryTask, status, message string) error { + if err := s.transitionTx(tx, task, status, message, message); err != nil { + return err + } + return nil +} + +func (s *DeliveryService) transition(ctx context.Context, task *model.DeliveryTask, status, message, errorMessage string) error { + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { return s.transitionTx(tx, task, status, message, errorMessage) }) +} + +func (s *DeliveryService) transitionTx(tx *gorm.DB, task *model.DeliveryTask, status, message, errorMessage string) error { + from := task.Status + updates := map[string]any{"status": status, "error_message": errorMessage} + now := time.Now() + if status == model.TaskRunning { + updates["started_at"] = now + } + if status == model.TaskFinished || status == model.TaskExecutionFailed || status == model.TaskValidationFailed || status == model.TaskCanceled || status == model.TaskRegisterFailed { + updates["finished_at"] = now + } + result := tx.Model(&model.DeliveryTask{}).Where("id = ? AND status = ?", task.ID, from).Updates(updates) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return fmt.Errorf("task %s changed concurrently", task.ID) + } + task.Status = status + task.ErrorMessage = errorMessage + return tx.Create(&model.TaskEvent{TaskID: task.ID, FromState: from, ToState: status, Message: message}).Error +} + +func (s *DeliveryService) releaseReservation(tx *gorm.DB, taskID string) error { + return tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", taskID, "reserved").Update("status", "released").Error +} + +func randomUUID() string { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + panic(err) + } + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]) +} + +func mysqlReady(ctx context.Context, address string) error { + dialer := net.Dialer{Timeout: 5 * time.Second} + conn, err := dialer.DialContext(ctx, "tcp", address) + if err != nil { + return err + } + return conn.Close() +} + +func (s *DeliveryService) DispatchOnce(ctx context.Context) error { + task, err := s.claimAndReserve(ctx) + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) || strings.HasPrefix(err.Error(), "defer:") { + return nil + } + return err + } + _, _, err = s.CreateExecution(ctx, task.ID, task.PayloadHash, task.IdempotencyKey) + if err != nil { + return s.failTask(ctx, task, model.TaskExecutionFailed, err.Error()) + } + return nil +} + +func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHash, idempotencyKey string) (*model.ExecutionJob, bool, error) { + s.executionMu.Lock() + defer s.executionMu.Unlock() + var existing model.ExecutionJob + if err := s.db.WithContext(ctx).Where("task_id = ? OR idempotency_key = ?", taskID, idempotencyKey).First(&existing).Error; err == nil { + if existing.TaskID != taskID || existing.IdempotencyKey != idempotencyKey { + return nil, false, fmt.Errorf("idempotency key is already bound to another task") + } + return &existing, true, nil + } else if !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, false, err + } + var task model.DeliveryTask + if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil { + return nil, false, err + } + if task.PayloadHash != payloadHash || task.IdempotencyKey != idempotencyKey { + return nil, false, fmt.Errorf("execution request does not match the immutable task payload") + } + if task.Status != model.TaskDispatching { + return nil, false, fmt.Errorf("task in state %q is not ready for execution", task.Status) + } + var target model.DeploymentTarget + if err := s.db.WithContext(ctx).First(&target, "id = ? AND enabled = ?", task.TargetID, true).Error; err != nil { + return nil, false, err + } + var payload deliveryPayload + if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil { + return nil, false, err + } + meta := parseTargetMetadata(target.Metadata) + job, err := s.awx.Launch(ctx, target.AWXTemplateID, AWXLaunchRequest{InventoryID: target.AWXInventoryID, Limit: task.TargetHost, ExtraVars: map[string]any{ + "task_id": task.ID, "payload_hash": task.PayloadHash, + "target_hosts": task.TargetHost, "topology": meta.Topology, + "instance_name": payload.InstanceName, "mysql_port": task.MySQLPort, + "memory_mb": payload.MemoryMi, "storage_gb": payload.StorageGi, + }}) + if err != nil { + return nil, false, err + } + now := time.Now() + execution := model.ExecutionJob{TaskID: task.ID, IdempotencyKey: task.IdempotencyKey, ExecutorJobID: fmt.Sprint(job.ID), Status: "running", StartedAt: &now} + if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Create(&execution).Error; err != nil { + return err + } + return s.transitionTx(tx, &task, model.TaskRunning, "AWX job started", "") + }); err != nil { + return nil, false, err + } + return &execution, false, nil +} + +func (s *DeliveryService) PollOnce(ctx context.Context) error { + var jobs []model.ExecutionJob + if err := s.db.WithContext(ctx).Where("status = ?", "running").Find(&jobs).Error; err != nil { + return err + } + for _, execution := range jobs { + job, err := s.awx.GetJob(ctx, execution.ExecutorJobID) + if err != nil { + continue + } + switch strings.ToLower(job.Status) { + case "pending", "waiting", "running", "new": + continue + case "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()) + } + default: + _ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, "AWX job finished with status "+job.Status) + } + } + return nil +} + +func (s *DeliveryService) completeTask(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 + } + var payload deliveryPayload + if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil { + return err + } + addr := fmt.Sprintf("%s:%d", task.TargetHostIP, task.MySQLPort) + if err := mysqlReady(ctx, addr); err != nil { + return fmt.Errorf("MySQL health check failed: %w", err) + } + now := time.Now() + if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := s.transitionTx(tx, &task, model.TaskRegistering, "AWX succeeded and MySQL health check passed", ""); err != nil { + return err + } + instance := model.MySQLInstance{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, Namespace: payload.Namespace, Name: payload.InstanceName, NodeName: task.TargetHost, Host: task.TargetHostIP, Port: task.MySQLPort, Version: payload.MySQLVersion, Status: "active"} + if err := tx.Create(&instance).Error; err != nil { + return err + } + if err := tx.Create(&model.ResourceUsage{TaskID: task.ID, InstanceID: instance.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, InstanceCount: 1, Status: "active"}).Error; err != nil { + return err + } + if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", task.ID, "reserved").Update("status", "consumed").Error; err != nil { + return err + } + if err := tx.Model(&model.ExecutionJob{}).Where("task_id = ?", task.ID).Updates(map[string]any{"status": "successful", "finished_at": now}).Error; err != nil { + return err + } + return nil + }); err != nil { + return err + } + if err := s.RegisterCloudDM(ctx, task.ID); err != nil { + return s.transition(ctx, &task, model.TaskRegisterFailed, "CloudDM registration failed", err.Error()) + } + return s.transition(ctx, &task, model.TaskFinished, "MySQL delivery completed", "") +} + +func (s *DeliveryService) RegisterCloudDM(ctx context.Context, taskID string) error { + if s.cfg.CloudDMRegisterURL == "" { + return nil + } + var instance model.MySQLInstance + if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).First(&instance).Error; err != nil { + return err + } + body := map[string]any{"name": instance.Name, "host": instance.Host, "port": instance.Port, "username": "root", "database_type": "mysql"} + raw, _ := json.Marshal(body) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.CloudDMRegisterURL, bytes.NewReader(raw)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if s.cfg.CloudDMAPIToken != "" { + req.Header.Set("Authorization", "Bearer "+s.cfg.CloudDMAPIToken) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("CloudDM returned %s", resp.Status) + } + return nil +} + +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 + if err := tx.First(¤t, "id = ?", task.ID).Error; err != nil { + return err + } + if current.Status != model.TaskPending && current.Status != model.TaskDispatching && current.Status != model.TaskRunning && current.Status != model.TaskRegistering { + return nil + } + if err := s.transitionTx(tx, ¤t, status, message, message); err != nil { + return err + } + if status == model.TaskExecutionFailed || status == model.TaskValidationFailed || status == model.TaskCanceled { + return s.releaseReservation(tx, current.ID) + } + return nil + }) +} + +func (s *DeliveryService) Run(ctx context.Context) { + interval := time.Duration(s.cfg.DeliveryPollSeconds) * time.Second + if interval < time.Second { + interval = time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + _ = s.DispatchOnce(ctx) + _ = s.PollOnce(ctx) + } + } +} diff --git a/server/internal/service/delivery_test.go b/server/internal/service/delivery_test.go new file mode 100644 index 0000000..6d2a88b --- /dev/null +++ b/server/internal/service/delivery_test.go @@ -0,0 +1,27 @@ +package service + +import "testing" + +func TestValidateDeliveryInput(t *testing.T) { + valid := MySQLDeliveryInput{BusinessLineID: 1, TargetID: 1, Namespace: "team-a", InstanceName: "mysql-01", CPUMilli: 500, MemoryMi: 1024, StorageGi: 10} + if err := validateDeliveryInput(valid); err != nil { + t.Fatalf("valid input rejected: %v", err) + } + for name, input := range map[string]MySQLDeliveryInput{ + "uppercase namespace": valid, + "bad instance": valid, + "too little memory": valid, + } { + switch name { + case "uppercase namespace": + input.Namespace = "Team-A" + case "bad instance": + input.InstanceName = "mysql_01" + case "too little memory": + input.MemoryMi = 128 + } + if err := validateDeliveryInput(input); err == nil { + t.Errorf("%s was accepted", name) + } + } +}