Merge pull request #104 from Hungerdream/fix/delivery-review-feedback

fix(delivery): address review feedback on idempotency, crash recovery, and security (#97)

ansible/mysql-deploy.yml

@@ -6,6 +6,10 @@
   any_errors_fatal: true
   vars:
     mysql_instance: "{{ instance_name }}"
+    mysql_version_value: "{{ mysql_version | default('8.0') }}"
+    mysql_package_map:
+      "8.0": "mysql-server=8.0.46-0ubuntu0.24.04.3"
+    mysql_package_name: "{{ mysql_package_map[mysql_version_value] }}"
     mysql_port_value: "{{ mysql_port | default(3307) | int }}"
     mysql_memory_mb_value: "{{ memory_mb | default(2048) | int }}"
     mysql_storage_gb_value: "{{ storage_gb | default(20) | int }}"
@@ -22,6 +26,7 @@
         that:
           - topology == 'standalone'
           - mysql_instance is match('^[a-z0-9][a-z0-9-]{0,62}$')
+          - mysql_version_value in mysql_package_map
           - (mysql_port_value | int) >= 1024
           - (mysql_port_value | int) <= 65535
           - (mysql_memory_mb_value | int) >= 1024
@@ -73,7 +78,7 @@
 
     - name: Install Ubuntu MySQL package
       ansible.builtin.apt:
-        name: mysql-server=8.0.46-0ubuntu0.24.04.3
+        name: "{{ mysql_package_name }}"
         state: present
         update_cache: true
         cache_valid_time: 3600
@@ -191,25 +196,28 @@
       ansible.builtin.shell: |
         set -euo pipefail
         client_file="$(mktemp)"
-        trap 'rm -f "$client_file"' EXIT
+        sql_file="$(mktemp)"
-        chmod 600 "$client_file"
+        trap 'rm -f "$client_file" "$sql_file"' EXIT
-        cat >"$client_file" <<EOF
+        chmod 600 "$client_file" "$sql_file"
+        cat >"$client_file" <<'EOF'
         [client]
         user=root
         password={{ mysql_root_password_value }}
         socket={{ mysql_run_dir }}/mysql.sock
         EOF
         if ! /usr/bin/mysql --defaults-extra-file="$client_file" -e 'SELECT 1' >/dev/null 2>&1; then
-          /usr/bin/mysql --protocol=socket --socket={{ mysql_run_dir }}/mysql.sock -uroot <<'SQL'
+          cat >"$sql_file" <<'EOF'
-        ALTER USER 'root'@'localhost' IDENTIFIED BY '{{ mysql_root_password_value }}';
+        ALTER USER 'root'@'localhost' IDENTIFIED BY '{{ mysql_root_password_value | replace("'", "''") }}';
-        SQL
+        EOF
+          /usr/bin/mysql --protocol=socket --socket={{ mysql_run_dir }}/mysql.sock -uroot <"$sql_file"
         fi
-        /usr/bin/mysql --defaults-extra-file="$client_file" <<'SQL'
+        cat >"$sql_file" <<'EOF'
-        CREATE USER IF NOT EXISTS 'xinfra_admin'@'%' IDENTIFIED BY '{{ mysql_admin_password_value }}';
+        CREATE USER IF NOT EXISTS 'xinfra_admin'@'%' IDENTIFIED BY '{{ mysql_admin_password_value | replace("'", "''") }}';
-        ALTER USER 'xinfra_admin'@'%' IDENTIFIED BY '{{ mysql_admin_password_value }}';
+        ALTER USER 'xinfra_admin'@'%' IDENTIFIED BY '{{ mysql_admin_password_value | replace("'", "''") }}';
         GRANT ALL PRIVILEGES ON *.* TO 'xinfra_admin'@'%' WITH GRANT OPTION;
         FLUSH PRIVILEGES;
-        SQL
+        EOF
+        /usr/bin/mysql --defaults-extra-file="$client_file" <"$sql_file"
       args:
         executable: /bin/bash
       changed_when: false

server/internal/service/delivery.go

@@ -105,6 +105,9 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin
 
 	var existing model.DeliveryTask
 	if err := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; err == nil {
+		if existing.RequestedBy != userID {
+			return nil, false, fmt.Errorf("idempotency key is already in use by another user")
+		}
 		return &existing, true, nil
 	} else if !errors.Is(err, gorm.ErrRecordNotFound) {
 		return nil, false, err
@@ -159,6 +162,8 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin
 	return &task, false, nil
 }
 
+var supportedMySQLVersions = map[string]bool{"8.0": true}
+
 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")
@@ -166,8 +171,11 @@ func validateDeliveryInput(input MySQLDeliveryInput) error {
 	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 {
+	if input.CPUMilli < 100 || input.CPUMilli > 64000 || input.MemoryMi < 1024 || input.MemoryMi > 4096 || input.StorageGi < 10 || input.StorageGi > 100 {
-		return fmt.Errorf("requested resources are outside the supported range")
+		return fmt.Errorf("requested resources are outside the supported range (memory: 1024-4096 MiB, storage: 10-100 GiB)")
+	}
+	if input.MySQLVersion != "" && !supportedMySQLVersions[input.MySQLVersion] {
+		return fmt.Errorf("unsupported mysql_version %q, supported: 8.0", input.MySQLVersion)
 	}
 	return nil
 }
@@ -268,7 +276,7 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT
 		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 {
+		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 > ?", task.BusinessLineID, task.TargetID, "reserved", time.Now()).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 {
@@ -411,19 +419,25 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa
 		return nil, false, err
 	}
 	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}
+	if err := s.db.WithContext(ctx).Create(&execution).Error; err != nil {
+		return nil, false, err
+	}
 	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,
+		"mysql_version": payload.MySQLVersion,
 	}})
 	if err != nil {
+		_ = s.db.WithContext(ctx).Model(&execution).Updates(map[string]any{"status": "failed", "finished_at": time.Now()})
 		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 {
+		if err := tx.Model(&execution).Updates(map[string]any{"executor_job_id": fmt.Sprint(job.ID), "status": "running"}).Error; err != nil {
 			return err
 		}
 		return s.transitionTx(tx, &task, model.TaskRunning, "AWX job started", "")
@@ -447,16 +461,26 @@ func (s *DeliveryService) PollOnce(ctx context.Context) error {
 		case "pending", "waiting", "running", "new":
 			continue
 		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())
 			}
+		case "canceled":
+			s.finishExecution(ctx, &execution, "canceled")
+			_ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskCanceled, "AWX 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)
 		}
 	}
 	return nil
 }
 
+func (s *DeliveryService) finishExecution(ctx context.Context, execution *model.ExecutionJob, status string) {
+	now := time.Now()
+	_ = s.db.WithContext(ctx).Model(execution).Updates(map[string]any{"status": status, "finished_at": now})
+}
+
 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 {
@@ -533,7 +557,7 @@ func (s *DeliveryService) failTask(ctx context.Context, task *model.DeliveryTask
 		if err := tx.First(&current, "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 {
+		if current.Status != model.TaskPending && current.Status != model.TaskDispatching && current.Status != model.TaskRunning && current.Status != model.TaskRegistering && current.Status != model.TaskCanceling {
 			return nil
 		}
 		if err := s.transitionTx(tx, &current, status, message, message); err != nil {

server/internal/service/delivery_test.go

@@ -7,10 +7,19 @@ func TestValidateDeliveryInput(t *testing.T) {
 	if err := validateDeliveryInput(valid); err != nil {
 		t.Fatalf("valid input rejected: %v", err)
 	}
+	withVersion := valid
+	withVersion.MySQLVersion = "8.0"
+	if err := validateDeliveryInput(withVersion); err != nil {
+		t.Fatalf("valid input with version 8.0 rejected: %v", err)
+	}
 	for name, input := range map[string]MySQLDeliveryInput{
-		"uppercase namespace": valid,
+		"uppercase namespace":  valid,
-		"bad instance":        valid,
+		"bad instance":         valid,
-		"too little memory":   valid,
+		"too little memory":    valid,
+		"too much memory":      valid,
+		"too little storage":   valid,
+		"too much storage":     valid,
+		"unsupported version":  valid,
 	} {
 		switch name {
 		case "uppercase namespace":
@@ -18,7 +27,15 @@ func TestValidateDeliveryInput(t *testing.T) {
 		case "bad instance":
 			input.InstanceName = "mysql_01"
 		case "too little memory":
-			input.MemoryMi = 128
+			input.MemoryMi = 512
+		case "too much memory":
+			input.MemoryMi = 8192
+		case "too little storage":
+			input.StorageGi = 5
+		case "too much storage":
+			input.StorageGi = 200
+		case "unsupported version":
+			input.MySQLVersion = "5.7"
 		}
 		if err := validateDeliveryInput(input); err == nil {
 			t.Errorf("%s was accepted", name)