feat(delivery): sync mysql service ledger and credentials

server/internal/service/delivery.go

@@ -12,6 +12,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"regexp"
@@ -128,6 +129,24 @@ type MySQLServiceLedgerItem struct {
 	Namespace   string `json:"namespace"`
 }
 
+type MySQLInspectResult struct {
+	ID              uint64         `json:"id"`
+	TaskID          string         `json:"task_id"`
+	InstanceName    string         `json:"instance_name"`
+	Host            string         `json:"host"`
+	Status          string         `json:"status"`
+	ServiceState    string         `json:"service_state"`
+	PortListening   bool           `json:"port_listening"`
+	ConfigExists    bool           `json:"config_exists"`
+	InstallExists   bool           `json:"install_exists"`
+	BaseExists      bool           `json:"base_exists"`
+	DataExists      bool           `json:"data_exists"`
+	RunExists       bool           `json:"run_exists"`
+	ExpectedDataDir string         `json:"expected_data_dir"`
+	ActualDataDir   string         `json:"actual_data_dir"`
+	Metadata        map[string]any `json:"metadata,omitempty"`
+}
+
 type DeploymentCredentialView struct {
 	Service      string `json:"service"`
 	InstanceName string `json:"instance_name"`
@@ -373,6 +392,9 @@ func (s *DeliveryService) ListTargets(ctx context.Context, component string) ([]
 		return nil, err
 	}
 	component = strings.ToLower(strings.TrimSpace(component))
+	if component == "mysql" {
+		return s.listNamedDeliveryTargets(ctx, templates, s.cfg.MySQLDeliveryTemplateName)
+	}
 	var targets []DeliveryTarget
 	for _, template := range templates {
 		if template.Inventory == 0 {
@@ -393,11 +415,42 @@ func (s *DeliveryService) ListTargets(ctx context.Context, component string) ([]
 	return targets, nil
 }
 
+func (s *DeliveryService) listNamedDeliveryTargets(ctx context.Context, templates []AWXJobTemplate, name string) ([]DeliveryTarget, error) {
+	name = strings.TrimSpace(name)
+	if name == "" {
+		return nil, fmt.Errorf("DELIVERY_MYSQL_TEMPLATE_NAME must be configured")
+	}
+	var matched []AWXJobTemplate
+	for _, template := range templates {
+		if template.Name == name {
+			matched = append(matched, template)
+		}
+	}
+	if len(matched) == 0 {
+		return nil, fmt.Errorf("AWX job template named %q was not found", name)
+	}
+	if len(matched) > 1 {
+		return nil, fmt.Errorf("multiple AWX job templates named %q found", name)
+	}
+	template := matched[0]
+	if template.Inventory == 0 {
+		return nil, fmt.Errorf("AWX job template %q does not bind an inventory", name)
+	}
+	target, err := s.awxDeliveryTarget(ctx, template)
+	if err != nil {
+		return nil, err
+	}
+	return []DeliveryTarget{target}, nil
+}
+
 func (s *DeliveryService) getTarget(ctx context.Context, templateID uint64) (DeliveryTarget, error) {
 	template, err := s.awx.GetJobTemplate(ctx, templateID)
 	if err != nil {
 		return DeliveryTarget{}, fmt.Errorf("deployment target is unavailable: %w", err)
 	}
+	if expected := strings.TrimSpace(s.cfg.MySQLDeliveryTemplateName); expected != "" && template.Name != expected {
+		return DeliveryTarget{}, fmt.Errorf("deployment target must be AWX job template %q", expected)
+	}
 	return s.awxDeliveryTarget(ctx, *template)
 }
 
@@ -702,7 +755,7 @@ func ensureInstanceNameAvailable(tx *gorm.DB, businessLineID uint64, component,
 
 	var resultCount int64
 	if err := tx.Model(&model.DeploymentResult{}).
-		Where("business_line_id = ? AND component = ? AND instance_name = ? AND status = ?", businessLineID, component, instanceName, "active").
+		Where("business_line_id = ? AND component = ? AND instance_name = ? AND status IN ?", businessLineID, component, instanceName, occupiedDeploymentStatuses()).
 		Count(&resultCount).Error; err != nil {
 		return err
 	}
@@ -722,6 +775,10 @@ var supportedTopologies = map[string]bool{"standalone": true}
 
 const rollbackLaunchTimeout = 2 * time.Minute
 
+func occupiedDeploymentStatuses() []string {
+	return []string{"active", "running", "stopped", "moved", "unknown"}
+}
+
 var supportedCharsets = map[string]bool{"utf8mb4": true, "utf8": true, "gbk": true, "latin1": true}
 
 // 高级参数档位白名单（与 docs/mysql-parameter-selection.md 保持一致）
@@ -869,16 +926,8 @@ func (s *DeliveryService) ListMySQLServiceLedger(ctx context.Context, userID uin
 	if businessLineID == 0 {
 		return nil, fmt.Errorf("business_line_id is required")
 	}
-	if !isAdmin {
-		var count int64
-		if err := s.db.WithContext(ctx).Model(&model.BusinessLineUser{}).
-			Where("business_line_id = ? AND user_id = ?", businessLineID, userID).
-			Count(&count).Error; err != nil {
-			return nil, err
-		}
-		if count == 0 {
-			return nil, fmt.Errorf("user is not authorized for this business line")
-		}
+	if err := s.ensureBusinessLineAccess(ctx, userID, isAdmin, businessLineID); err != nil {
+		return nil, err
 	}
 	var businessLine model.BusinessLine
 	if err := s.db.WithContext(ctx).First(&businessLine, "id = ?", businessLineID).Error; err != nil {
@@ -886,7 +935,7 @@ func (s *DeliveryService) ListMySQLServiceLedger(ctx context.Context, userID uin
 	}
 	var instances []model.DeploymentResult
 	if err := s.db.WithContext(ctx).
-		Where("business_line_id = ? AND component = ? AND service_type = ? AND status = ?", businessLineID, "mysql", "database", "active").
+		Where("business_line_id = ? AND component = ? AND service_type = ? AND status IN ?", businessLineID, "mysql", "database", []string{"active", "running", "stopped", "moved", "unknown"}).
 		Order("created_at DESC").
 		Find(&instances).Error; err != nil {
 		return nil, err
@@ -907,10 +956,25 @@ func (s *DeliveryService) ListMySQLServiceLedger(ctx context.Context, userID uin
 	}
 	items := make([]MySQLServiceLedgerItem, 0, len(instances))
 	for _, instance := range instances {
+		s.ensureMySQLDeploymentMetadata(ctx, &instance)
 		status := "健康"
 		statusClass := "ok"
 		healthy := 1
-		if taskStatuses[instance.TaskID] == model.TaskRegisterFailed {
+		switch instance.Status {
+		case "stopped":
+			status = "已关闭"
+			statusClass = "warn"
+			healthy = 0
+		case "moved":
+			status = "路径变更"
+			statusClass = "warn"
+			healthy = 0
+		case "unknown":
+			status = "未知"
+			statusClass = "warn"
+			healthy = 0
+		}
+		if taskStatuses[instance.TaskID] == model.TaskRegisterFailed && healthy == 1 {
 			status = "注册异常"
 			statusClass = "warn"
 		}
@@ -935,6 +999,244 @@ func (s *DeliveryService) ListMySQLServiceLedger(ctx context.Context, userID uin
 	return items, nil
 }
 
+func (s *DeliveryService) SyncMySQLInstanceStatuses(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]MySQLInspectResult, error) {
+	if businessLineID == 0 {
+		return nil, fmt.Errorf("business_line_id is required")
+	}
+	if err := s.ensureBusinessLineAccess(ctx, userID, isAdmin, businessLineID); err != nil {
+		return nil, err
+	}
+	inspectTemplateName := strings.TrimSpace(s.cfg.MySQLInspectTemplateName)
+	if inspectTemplateName == "" {
+		return nil, fmt.Errorf("DELIVERY_MYSQL_INSPECT_TEMPLATE_NAME must be configured")
+	}
+	inspectTemplate, err := s.awxJobTemplateByName(ctx, inspectTemplateName)
+	if err != nil {
+		return nil, err
+	}
+	var instances []model.DeploymentResult
+	if err := s.db.WithContext(ctx).
+		Where("business_line_id = ? AND component = ? AND service_type = ? AND status IN ?", businessLineID, "mysql", "database", occupiedDeploymentStatuses()).
+		Order("target_id ASC, created_at DESC").
+		Find(&instances).Error; err != nil {
+		return nil, err
+	}
+	if len(instances) == 0 {
+		return []MySQLInspectResult{}, nil
+	}
+	byTarget := map[uint64][]model.DeploymentResult{}
+	for i := range instances {
+		s.ensureMySQLDeploymentMetadata(ctx, &instances[i])
+		byTarget[instances[i].TargetID] = append(byTarget[instances[i].TargetID], instances[i])
+	}
+	out := make([]MySQLInspectResult, 0, len(instances))
+	for targetID, group := range byTarget {
+		targetTemplate, err := s.awx.GetJobTemplate(ctx, targetID)
+		if err != nil {
+			return nil, fmt.Errorf("load AWX target template %d: %w", targetID, err)
+		}
+		extraInstances := make([]map[string]any, 0, len(group))
+		hosts := make([]string, 0, len(group))
+		seenHosts := map[string]struct{}{}
+		for _, instance := range group {
+			meta := metadataMap(instance.Metadata)
+			host := strings.TrimSpace(instance.NodeName)
+			if host == "" {
+				host = strings.TrimSpace(instance.Host)
+			}
+			if host != "" {
+				if _, ok := seenHosts[host]; !ok {
+					hosts = append(hosts, host)
+					seenHosts[host] = struct{}{}
+				}
+			}
+			extraInstances = append(extraInstances, map[string]any{
+				"id":            instance.ID,
+				"task_id":       instance.TaskID,
+				"instance_name": instance.InstanceName,
+				"target_host":   host,
+				"host":          instance.Host,
+				"port":          instance.Port,
+				"data_disk":     stringValue(meta["data_disk"]),
+				"base_dir":      stringValue(meta["base_dir"]),
+				"install_dir":   stringValue(meta["install_dir"]),
+				"data_dir":      stringValue(meta["data_dir"]),
+				"run_dir":       stringValue(meta["run_dir"]),
+				"config_file":   stringValue(meta["config_file"]),
+				"service_name":  stringValue(meta["service_name"]),
+			})
+		}
+		sort.Strings(hosts)
+		job, err := s.awx.Launch(ctx, inspectTemplate.ID, AWXLaunchRequest{
+			InventoryID: targetTemplate.Inventory,
+			Limit:       strings.Join(hosts, ","),
+			ExtraVars: map[string]any{
+				"target_hosts":    strings.Join(hosts, ","),
+				"mysql_instances": extraInstances,
+			},
+		})
+		if err != nil {
+			return nil, fmt.Errorf("launch AWX MySQL inspect job: %w", err)
+		}
+		done, err := s.awx.WaitJob(ctx, strconv.FormatUint(job.ID, 10), time.Duration(s.cfg.MySQLInspectTimeoutSeconds)*time.Second)
+		if err != nil {
+			return nil, err
+		}
+		stdout, err := s.awx.JobStdout(ctx, strconv.FormatUint(job.ID, 10))
+		if err != nil {
+			return nil, err
+		}
+		if done.Status != "successful" {
+			return nil, fmt.Errorf("AWX MySQL inspect job %d finished with status %s: %s", done.ID, done.Status, truncateForEvent(stdout, 1000))
+		}
+		results := parseMySQLInspectResults(stdout)
+		if len(results) == 0 {
+			return nil, fmt.Errorf("AWX MySQL inspect job %d returned no machine-readable results", job.ID)
+		}
+		if err := s.applyMySQLInspectResults(ctx, results); err != nil {
+			return nil, err
+		}
+		out = append(out, results...)
+	}
+	return out, nil
+}
+
+func (s *DeliveryService) awxJobTemplateByName(ctx context.Context, name string) (*AWXJobTemplate, error) {
+	name = strings.TrimSpace(name)
+	if name == "" {
+		return nil, fmt.Errorf("AWX job template name is required")
+	}
+	templates, err := s.awx.ListJobTemplates(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var matched *AWXJobTemplate
+	for i := range templates {
+		if templates[i].Name != name {
+			continue
+		}
+		if matched != nil {
+			return nil, fmt.Errorf("multiple AWX job templates named %q found", name)
+		}
+		item := templates[i]
+		matched = &item
+	}
+	if matched == nil {
+		return nil, fmt.Errorf("AWX job template named %q was not found", name)
+	}
+	return matched, nil
+}
+
+func (s *DeliveryService) ensureBusinessLineAccess(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) error {
+	if isAdmin {
+		return nil
+	}
+	var count int64
+	if err := s.db.WithContext(ctx).Model(&model.BusinessLineUser{}).
+		Where("business_line_id = ? AND user_id = ?", businessLineID, userID).
+		Count(&count).Error; err != nil {
+		return err
+	}
+	if count == 0 {
+		return fmt.Errorf("user is not authorized for this business line")
+	}
+	return nil
+}
+
+func parseMySQLInspectResults(stdout string) []MySQLInspectResult {
+	const marker = "XINFRA_MYSQL_INSPECT_RESULT_B64="
+	var out []MySQLInspectResult
+	for _, line := range strings.Split(stdout, "\n") {
+		idx := strings.Index(line, marker)
+		if idx < 0 {
+			continue
+		}
+		token := strings.Trim(strings.TrimSpace(line[idx+len(marker):]), "\\\",")
+		end := 0
+		for end < len(token) {
+			ch := token[end]
+			if (ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z') || (ch >= '0' && ch <= '9') || ch == '+' || ch == '/' || ch == '=' {
+				end++
+				continue
+			}
+			break
+		}
+		token = token[:end]
+		rawBytes, err := base64.StdEncoding.DecodeString(token)
+		if err != nil {
+			continue
+		}
+		var items []MySQLInspectResult
+		if err := json.Unmarshal(rawBytes, &items); err == nil {
+			out = append(out, items...)
+		}
+	}
+	return out
+}
+
+func (s *DeliveryService) applyMySQLInspectResults(ctx context.Context, results []MySQLInspectResult) error {
+	now := time.Now()
+	for _, result := range results {
+		status := normalizeMySQLInspectStatus(result.Status)
+		if result.ID == 0 || status == "" {
+			continue
+		}
+		var current model.DeploymentResult
+		if err := s.db.WithContext(ctx).First(&current, "id = ?", result.ID).Error; err != nil {
+			return err
+		}
+		meta := metadataMap(current.Metadata)
+		meta["last_probe"] = map[string]any{
+			"checked_at":        now.Format(time.RFC3339),
+			"status":            status,
+			"host":              result.Host,
+			"service_state":     result.ServiceState,
+			"port_listening":    result.PortListening,
+			"config_exists":     result.ConfigExists,
+			"install_exists":    result.InstallExists,
+			"base_exists":       result.BaseExists,
+			"data_exists":       result.DataExists,
+			"run_exists":        result.RunExists,
+			"expected_data_dir": result.ExpectedDataDir,
+			"actual_data_dir":   result.ActualDataDir,
+		}
+		if status == "moved" && result.ActualDataDir != "" {
+			meta["actual_data_dir"] = result.ActualDataDir
+		}
+		result.Metadata = meta
+		if err := s.db.WithContext(ctx).Model(&model.DeploymentResult{}).
+			Where("id = ?", result.ID).
+			Updates(map[string]any{"status": status, "metadata": string(mustJSON(meta)), "updated_at": now}).Error; err != nil {
+			return err
+		}
+		if status == "deleted" {
+			if err := s.db.WithContext(ctx).Model(&model.ResourceUsage{}).
+				Where("instance_id = ? AND status = ?", result.ID, "active").
+				Updates(map[string]any{"status": "released", "released_at": now, "updated_at": now}).Error; err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func normalizeMySQLInspectStatus(status string) string {
+	switch strings.ToLower(strings.TrimSpace(status)) {
+	case "running", "stopped", "deleted", "moved", "unknown":
+		return strings.ToLower(strings.TrimSpace(status))
+	default:
+		return "unknown"
+	}
+}
+
+func truncateForEvent(value string, limit int) string {
+	value = strings.TrimSpace(value)
+	if limit <= 0 || len(value) <= limit {
+		return value
+	}
+	return value[:limit] + "..."
+}
+
 func (s *DeliveryService) storeDeploymentCredentials(ctx context.Context, task *model.DeliveryTask, credentials map[string]string, status string) error {
 	for key, password := range credentials {
 		username, host, ok := strings.Cut(key, "@")
@@ -1350,7 +1652,7 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT
 		}
 		var instancePorts []int
 		if err := tx.Model(&model.DeploymentResult{}).
-			Where("component = ? AND service_type = ? AND node_name = ? AND status = ?", "mysql", "database", host.Name, "active").
+			Where("component = ? AND service_type = ? AND node_name = ? AND status IN ?", "mysql", "database", host.Name, occupiedDeploymentStatuses()).
 			Pluck("port", &instancePorts).Error; err != nil {
 			return err
 		}
@@ -1457,6 +1759,115 @@ func mustJSON(value any) []byte {
 	return raw
 }
 
+func metadataMap(raw string) map[string]any {
+	out := map[string]any{}
+	if strings.TrimSpace(raw) != "" {
+		_ = json.Unmarshal([]byte(raw), &out)
+	}
+	return out
+}
+
+func mysqlDeploymentMetadata(task model.DeliveryTask, payload deliveryPayload) map[string]any {
+	dataDisk := strings.TrimRight(strings.TrimSpace(payload.DataDisk), "/")
+	if dataDisk == "" {
+		dataDisk = "/data"
+	}
+	instance := payload.InstanceName
+	if instance == "" {
+		instance = task.InstanceName
+	}
+	mysqlVersion := payload.MySQLVersion
+	if mysqlVersion == "" {
+		mysqlVersion = "8.0"
+	}
+	topology := payload.Topology
+	if topology == "" {
+		topology = "standalone"
+	}
+	timezone := payload.Timezone
+	if timezone == "" {
+		timezone = "+08:00"
+	}
+	lowerCaseTableNames := 1
+	if payload.LowerCaseTableNames != nil {
+		lowerCaseTableNames = *payload.LowerCaseTableNames
+	}
+	characterSet := payload.CharacterSet
+	if characterSet == "" {
+		characterSet = "utf8mb4"
+	}
+	collation := payload.Collation
+	if collation == "" {
+		collation = "utf8mb4_general_ci"
+	}
+	baseDir := dataDisk + "/mysql-delivery/" + instance
+	return map[string]any{
+		"metadata_version":               2,
+		"component":                      "mysql",
+		"service_type":                   "database",
+		"business_line_id":               task.BusinessLineID,
+		"target_id":                      task.TargetID,
+		"target_type":                    task.TargetType,
+		"namespace":                      payload.Namespace,
+		"instance_name":                  instance,
+		"target_host":                    task.TargetHost,
+		"target_host_ip":                 task.TargetHostIP,
+		"mysql_port":                     task.MySQLPort,
+		"mysql_version":                  mysqlVersion,
+		"topology":                       topology,
+		"cpu_milli":                      payload.CPUMilli,
+		"memory_mi":                      payload.MemoryMi,
+		"storage_gi":                     payload.StorageGi,
+		"data_disk":                      dataDisk,
+		"base_dir":                       baseDir,
+		"install_dir":                    "/opt/mysql-delivery/" + instance,
+		"data_dir":                       baseDir + "/data",
+		"log_dir":                        baseDir + "/logs",
+		"binlog_dir":                     baseDir + "/logs/binlog",
+		"redo_dir":                       baseDir + "/logs/redo",
+		"tmp_dir":                        baseDir + "/tmp",
+		"run_dir":                        "/run/mysql-delivery-" + instance,
+		"config_file":                    "/etc/mysql/mysql-delivery/" + instance + ".cnf",
+		"service_name":                   "mysql-delivery@" + instance + ".service",
+		"timezone":                       timezone,
+		"lower_case_table_names":         lowerCaseTableNames,
+		"character_set":                  characterSet,
+		"collation":                      collation,
+		"max_connections":                payload.MaxConnections,
+		"innodb_redo_log_capacity":       payload.InnodbRedoLogCapacity,
+		"innodb_flush_log_at_trx_commit": payload.InnodbFlushLogAtTrxCommit,
+		"sync_binlog":                    payload.SyncBinlog,
+		"innodb_io_capacity":             payload.InnodbIOCapacity,
+		"long_query_time":                payload.LongQueryTime,
+		"binlog_expire_logs_seconds":     payload.BinlogExpireLogsSeconds,
+		"max_binlog_size":                payload.MaxBinlogSize,
+	}
+}
+
+func (s *DeliveryService) ensureMySQLDeploymentMetadata(ctx context.Context, result *model.DeploymentResult) map[string]any {
+	meta := metadataMap(result.Metadata)
+	if fmt.Sprint(meta["metadata_version"]) == "2" && strings.TrimSpace(stringValue(meta["data_dir"])) != "" && strings.TrimSpace(stringValue(meta["config_file"])) != "" {
+		return meta
+	}
+	var task model.DeliveryTask
+	if err := s.db.WithContext(ctx).First(&task, "id = ?", result.TaskID).Error; err != nil {
+		return meta
+	}
+	var payload deliveryPayload
+	if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
+		return meta
+	}
+	enriched := mysqlDeploymentMetadata(task, payload)
+	for key, value := range meta {
+		enriched[key] = value
+	}
+	if raw := string(mustJSON(enriched)); raw != result.Metadata {
+		_ = s.db.WithContext(ctx).Model(result).Update("metadata", raw).Error
+		result.Metadata = raw
+	}
+	return enriched
+}
+
 func mysqlReady(ctx context.Context, address string) error {
 	dialer := net.Dialer{Timeout: 5 * time.Second}
 	conn, err := dialer.DialContext(ctx, "tcp", address)
@@ -1892,6 +2303,7 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error
 		if err := s.transitionTx(tx, &task, model.TaskRegistering, "AWX succeeded and MySQL health check passed", ""); err != nil {
 			return err
 		}
+		metadata := mysqlDeploymentMetadata(task, payload)
 		result := model.DeploymentResult{
 			TaskID:         task.ID,
 			BusinessLineID: task.BusinessLineID,
@@ -1905,7 +2317,7 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error
 			Port:           task.MySQLPort,
 			Version:        payload.MySQLVersion,
 			Status:         "active",
-			Metadata:       string(mustJSON(map[string]any{"data_disk": payload.DataDisk})),
+			Metadata:       string(mustJSON(metadata)),
 		}
 		if result.Component == "" {
 			result.Component = "mysql"
@@ -1999,6 +2411,56 @@ func cloudDMClientTimeZone(value string) string {
 	return value
 }
 
+func cloudDMDataSourceIDFromResponse(raw []byte) (uint64, bool) {
+	var value any
+	if len(bytes.TrimSpace(raw)) == 0 {
+		return 0, false
+	}
+	if err := json.Unmarshal(raw, &value); err != nil {
+		return 0, false
+	}
+	return cloudDMDataSourceIDFromValue(value)
+}
+
+func cloudDMDataSourceIDFromValue(value any) (uint64, bool) {
+	switch typed := value.(type) {
+	case float64:
+		if typed <= 0 || typed != float64(uint64(typed)) {
+			return 0, false
+		}
+		return uint64(typed), true
+	case string:
+		id, err := strconv.ParseUint(strings.TrimSpace(typed), 10, 64)
+		if err != nil || id == 0 {
+			return 0, false
+		}
+		return id, true
+	case map[string]any:
+		for _, key := range []string{"dataSourceId", "datasourceId", "dsId", "id", "data"} {
+			if id, ok := cloudDMDataSourceIDFromValue(typed[key]); ok {
+				return id, true
+			}
+		}
+	}
+	return 0, false
+}
+
+func cloudDMMetadata(existing map[string]any, instance model.DeploymentResult, dataSourceID uint64) map[string]any {
+	meta := map[string]any{}
+	for key, value := range existing {
+		meta[key] = value
+	}
+	meta["clouddm"] = map[string]any{
+		"data_source_id":       dataSourceID,
+		"external_resource_id": cloudDMExternalResourceID(instance.ID),
+		"registered_at":        time.Now().Format(time.RFC3339),
+		"delete_status":        "",
+		"delete_error":         "",
+		"deleted_at":           "",
+	}
+	return meta
+}
+
 func buildCloudDMRegisterRequest(instance model.DeploymentResult, payload deliveryPayload, password string) cloudDMRegisterRequest {
 	description := strings.TrimSpace(payload.InstanceDesc)
 	if description == "" {
@@ -2073,8 +2535,18 @@ func (s *DeliveryService) RegisterCloudDM(ctx context.Context, taskID string) er
 		return err
 	}
 	defer resp.Body.Close()
+	respBody, _ := io.ReadAll(resp.Body)
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return fmt.Errorf("CloudDM returned %s", resp.Status)
+		return fmt.Errorf("CloudDM returned %s: %s", resp.Status, strings.TrimSpace(string(respBody)))
+	}
+	dataSourceID, ok := cloudDMDataSourceIDFromResponse(respBody)
+	if !ok {
+		return fmt.Errorf("CloudDM registration response did not include dataSourceId: %s", strings.TrimSpace(string(respBody)))
+	}
+	meta := metadataMap(instance.Metadata)
+	meta = cloudDMMetadata(meta, instance, dataSourceID)
+	if err := s.db.WithContext(ctx).Model(&instance).Update("metadata", string(mustJSON(meta))).Error; err != nil {
+		return fmt.Errorf("save CloudDM dataSourceId: %w", err)
 	}
 	return nil
 }
@@ -2204,7 +2676,7 @@ func (s *DeliveryService) AcknowledgeRollbackRelease(ctx context.Context, taskID
 		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.DeploymentResult{}).Where("task_id = ? AND component = ? AND service_type = ? AND status = ?", taskID, "mysql", "database", "active").Updates(map[string]any{"status": "rollback_acknowledged", "updated_at": now}).Error; err != nil {
+		if err := tx.Model(&model.DeploymentResult{}).Where("task_id = ? AND component = ? AND service_type = ? AND status IN ?", taskID, "mysql", "database", occupiedDeploymentStatuses()).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 {

server/internal/service/delivery_test.go

@@ -1,6 +1,7 @@
 package service
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"testing"
 	"time"
@@ -88,6 +89,18 @@ func TestValidateDeliveryInput(t *testing.T) {
 	}
 }
 
+func TestParseMySQLInspectResults(t *testing.T) {
+	payload := `[{"id":7,"task_id":"task-1","instance_name":"mysql-01","host":"k8s-01","status":"running","port_listening":true}]`
+	stdout := `ok: [k8s-01] => {"msg": "XINFRA_MYSQL_INSPECT_RESULT_B64=` + base64.StdEncoding.EncodeToString([]byte(payload)) + `"}`
+	items := parseMySQLInspectResults(stdout)
+	if len(items) != 1 {
+		t.Fatalf("len(items) = %d, want 1", len(items))
+	}
+	if items[0].ID != 7 || items[0].Status != "running" || !items[0].PortListening {
+		t.Fatalf("unexpected inspect result: %+v", items[0])
+	}
+}
+
 func TestFirstFreeHost(t *testing.T) {
 	hosts := []targetHost{{Name: "node-a"}, {Name: "node-b"}}
 	if h := firstFreeHost(hosts, nil, 1); h == nil || h.Name != "node-a" {
@@ -164,6 +177,29 @@ func TestBuildCloudDMRegisterRequest(t *testing.T) {
 	}
 }
 
+func TestCloudDMDataSourceIDFromResponse(t *testing.T) {
+	tests := []struct {
+		name string
+		body string
+		want uint64
+		ok   bool
+	}{
+		{name: "direct data", body: `{"data":123}`, want: 123, ok: true},
+		{name: "nested data source id", body: `{"code":0,"data":{"dataSourceId":456}}`, want: 456, ok: true},
+		{name: "nested id", body: `{"success":true,"data":{"id":789}}`, want: 789, ok: true},
+		{name: "string ds id", body: `{"dsId":"321"}`, want: 321, ok: true},
+		{name: "missing", body: `{"code":0,"message":"ok"}`, want: 0, ok: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, ok := cloudDMDataSourceIDFromResponse([]byte(tt.body))
+			if got != tt.want || ok != tt.ok {
+				t.Fatalf("cloudDMDataSourceIDFromResponse() = %d, %v; want %d, %v", got, ok, tt.want, tt.ok)
+			}
+		})
+	}
+}
+
 func TestRollbackExtraVarsTargetsOnlyTheAllocatedInstance(t *testing.T) {
 	task := &model.DeliveryTask{ID: "task-1", TargetHost: "db-01"}
 	payload := deliveryPayload{MySQLDeliveryInput: MySQLDeliveryInput{InstanceName: "mysql-a", DataDisk: "/disk1"}}