package transaction import ( "context" "database/sql" "errors" "fmt" "strings" "time" ) // backendService backend 容器部署事务使用的服务名常量。 // // 该常量同时用于部署表相关事务的归属判定与历史查询的条件过滤。 const backendService = "backend" // BackendContainerDeployment is the last container slot committed for backend. // Runtime inspection remains authoritative for the current external state; this // record distinguishes a fresh installation from loss or drift after deployment. // // BackendContainerDeployment backend 最近一次提交使用的容器槽位持久化记录。 // 它只保留单行(singleton_id=1),用于区分全新安装与部署后丢失/漂移:运行时 inspect // 始终是当前外部状态的权威来源,而本记录提供“机器是否曾成功部署”的持久化证据。 type BackendContainerDeployment struct { ActivePort int // 当前对外服务的端口,只能是 8080 或 8081。 ContainerName string // 生效容器的名称。 ImageDigest string // 生效容器镜像的摘要。 ContainerID string // 生效容器的 ID。 TransactionID string // 提交本次部署的事务 ID。 UpdatedAt time.Time // 记录最近一次更新(提交)时间(UTC)。 } // BackendContainerDeployment returns the single committed backend container state. // // BackendContainerDeployment 返回当前唯一已提交的 backend 容器部署记录。参数 ctx // 用于取消查询。若尚无部署记录则返回 ErrNotFound。 func (s *Store) BackendContainerDeployment(ctx context.Context) (BackendContainerDeployment, error) { return scanBackendContainerDeployment(s.db.QueryRowContext(ctx, ` SELECT active_port, container_name, image_digest, container_id, transaction_id, updated_at FROM backend_container_deployment WHERE singleton_id = 1`)) } // HasCommittedBackendContainerTransactionHistory reports whether this store // contains a committed container backend transaction. This is the legacy // deployment evidence used when a database predates the deployment table. // Failed and rolled-back first-install attempts do not mark a machine deployed. // // HasCommittedBackendContainerTransactionHistory 报告当前存储中是否存在已提交的 // backend 容器事务历史。当数据库版本早于部署表(缺少 backend_container_deployment // 单例记录)时,它作为历史部署证据使用:只要存在 service 为 backend 且幂等键匹配 // backend:container:* 且状态为 COMMITTED 的事务,即视为该机器已部署。失败或回滚的 // 首次安装尝试不会把机器标记为已部署。参数 ctx 用于取消查询,返回是否已部署的布尔值 // 及可能的错误。 func (s *Store) HasCommittedBackendContainerTransactionHistory(ctx context.Context) (bool, error) { var found int if err := s.db.QueryRowContext(ctx, ` SELECT EXISTS ( SELECT 1 FROM transactions WHERE service = ? AND idempotency_key GLOB 'backend:container:*' AND state = ? )`, backendService, StateCommitted).Scan(&found); err != nil { return false, fmt.Errorf("query committed backend container transaction history: %w", err) } return found == 1, nil } // CommitBackendContainerDeployment atomically records the active container and // commits its transaction. A crash cannot leave COMMITTED without the matching // deployment row, or publish a deployment row for an unfinished transaction. // // CommitBackendContainerDeployment 原子地记录生效容器并把其事务提交为 COMMITTED。 // 参数 ctx 用于取消操作;transactionID 为待提交的事务 ID,必须与 backend 服务匹配 // 且当前状态为 DRAINING;deployment 为要写入的容器部署信息;message 为随状态事件 // 记录的信息。返回值为更新后的最新事务快照。由于部署行写入与事务状态提交在同一个 // Serializable 事务内完成,崩溃不可能留下“已 COMMITTED 却没有对应部署行”或“未完成 // 事务却已发布部署行”的中间状态。 func (s *Store) CommitBackendContainerDeployment( ctx context.Context, transactionID string, deployment BackendContainerDeployment, message string, ) (Transaction, error) { if transactionID == "" || strings.TrimSpace(transactionID) != transactionID { return Transaction{}, errors.New("exact transaction ID is required") } if err := validateBackendContainerDeployment(deployment); err != nil { return Transaction{}, err } tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable}) if err != nil { return Transaction{}, fmt.Errorf("begin backend container deployment commit: %w", err) } defer tx.Rollback() record, err := getTransactionByID(ctx, tx, transactionID) if err != nil { return Transaction{}, err } if record.Service != backendService { return Transaction{}, fmt.Errorf("transaction %s service must be %s", transactionID, backendService) } if record.State != StateDraining { return Transaction{}, &TransitionError{From: record.State, To: StateCommitted} } now := s.now().UTC() _, err = tx.ExecContext(ctx, ` INSERT INTO backend_container_deployment ( singleton_id, active_port, container_name, image_digest, container_id, transaction_id, updated_at ) VALUES (1, ?, ?, ?, ?, ?, ?) ON CONFLICT(singleton_id) DO UPDATE SET active_port = excluded.active_port, container_name = excluded.container_name, image_digest = excluded.image_digest, container_id = excluded.container_id, transaction_id = excluded.transaction_id, updated_at = excluded.updated_at`, deployment.ActivePort, deployment.ContainerName, deployment.ImageDigest, deployment.ContainerID, transactionID, formatTime(now), ) if err != nil { return Transaction{}, fmt.Errorf("write backend container deployment: %w", err) } result, err := tx.ExecContext(ctx, ` UPDATE transactions SET state = ?, version = version + 1, updated_at = ? WHERE id = ? AND version = ? AND state = ?`, StateCommitted, formatTime(now), transactionID, record.Version, StateDraining, ) if err != nil { return Transaction{}, fmt.Errorf("commit backend container transaction state: %w", err) } rows, err := result.RowsAffected() if err != nil { return Transaction{}, fmt.Errorf("read committed backend container transaction rows: %w", err) } if rows != 1 { return Transaction{}, errors.New("backend container transaction changed concurrently") } if err := insertEvent(ctx, tx, transactionID, "", "TRANSACTION_STATE_CHANGED", record.State, StateCommitted, message, now); err != nil { return Transaction{}, err } if err := tx.Commit(); err != nil { return Transaction{}, fmt.Errorf("commit backend container deployment: %w", err) } record.State = StateCommitted record.Version++ record.UpdatedAt = now return record, nil } // validateBackendContainerDeployment 校验 backend 容器部署信息的合法性。 // // ActivePort 必须是 8080 或 8081;ContainerName、ImageDigest、ContainerID 均不能 // 为空且不能含有首尾空白(即必须是精确值)。全部通过返回 nil,否则返回描述具体 // 问题的错误。 func validateBackendContainerDeployment(deployment BackendContainerDeployment) error { if deployment.ActivePort != 8080 && deployment.ActivePort != 8081 { return fmt.Errorf("backend container deployment port must be 8080 or 8081: %d", deployment.ActivePort) } for _, field := range []struct { name string value string }{ {"container name", deployment.ContainerName}, {"image digest", deployment.ImageDigest}, {"container ID", deployment.ContainerID}, } { if field.value == "" || strings.TrimSpace(field.value) != field.value { return fmt.Errorf("exact backend container %s is required", field.name) } } return nil } // scanBackendContainerDeployment 从单行结果反序列化一条 backend 容器部署记录。 // // 参数 row 提供扫描能力,返回反序列化后的 BackendContainerDeployment。将持久化的 // updated_at 文本解析为 time.Time。未命中返回 ErrNotFound,其他解析失败返回包装后 // 的错误。 func scanBackendContainerDeployment(row rowScanner) (BackendContainerDeployment, error) { var deployment BackendContainerDeployment var updatedAt string if err := row.Scan( &deployment.ActivePort, &deployment.ContainerName, &deployment.ImageDigest, &deployment.ContainerID, &deployment.TransactionID, &updatedAt, ); err != nil { if errors.Is(err, sql.ErrNoRows) { return BackendContainerDeployment{}, ErrNotFound } return BackendContainerDeployment{}, fmt.Errorf("scan backend container deployment: %w", err) } parsed, err := parseTime(updatedAt) if err != nil { return BackendContainerDeployment{}, fmt.Errorf("parse backend container deployment update time: %w", err) } deployment.UpdatedAt = parsed return deployment, nil }