diff --git a/internal/transaction/node_ssr_container_deployment.go b/internal/transaction/node_ssr_container_deployment.go new file mode 100644 index 0000000..eac1a0b --- /dev/null +++ b/internal/transaction/node_ssr_container_deployment.go @@ -0,0 +1,114 @@ +package transaction + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" + "time" +) + +const nodeSsrService = "nodeSsr" + +// NodeSsrContainerDeployment 保存 Node SSR 最近一次提交的活动槽位。 +type NodeSsrContainerDeployment struct { + ActivePort int + ContainerName string + ImageDigest string + ContainerID string + TransactionID string + UpdatedAt time.Time +} + +// NodeSsrContainerDeployment 返回当前 Node SSR 部署记录。 +func (s *Store) NodeSsrContainerDeployment(ctx context.Context) (NodeSsrContainerDeployment, error) { + return scanNodeSsrContainerDeployment(s.db.QueryRowContext(ctx, ` + SELECT active_port, container_name, image_digest, container_id, transaction_id, updated_at + FROM node_ssr_container_deployment WHERE singleton_id = 1`)) +} + +// CommitNodeSsrContainerDeployment 原子写入 Node SSR 部署记录并提交事务。 +func (s *Store) CommitNodeSsrContainerDeployment(ctx context.Context, transactionID string, deployment NodeSsrContainerDeployment, message string) (Transaction, error) { + if transactionID == "" || strings.TrimSpace(transactionID) != transactionID { + return Transaction{}, errors.New("exact transaction ID is required") + } + if err := validateNodeSsrContainerDeployment(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 Node SSR container deployment commit: %w", err) + } + defer tx.Rollback() + record, err := getTransactionByID(ctx, tx, transactionID) + if err != nil { + return Transaction{}, err + } + if record.Service != nodeSsrService { + return Transaction{}, fmt.Errorf("transaction %s service must be %s", transactionID, nodeSsrService) + } + if record.State != StateDraining { + return Transaction{}, &TransitionError{From: record.State, To: StateCommitted} + } + now := s.now().UTC() + _, err = tx.ExecContext(ctx, ` + INSERT INTO node_ssr_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 Node SSR 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 Node SSR transaction state: %w", err) + } + rows, err := result.RowsAffected() + if err != nil || rows != 1 { + return Transaction{}, errors.New("Node SSR 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 Node SSR container deployment: %w", err) + } + record.State = StateCommitted + record.Version++ + record.UpdatedAt = now + return record, nil +} + +func validateNodeSsrContainerDeployment(deployment NodeSsrContainerDeployment) error { + if deployment.ActivePort != 18910 && deployment.ActivePort != 28910 { + return fmt.Errorf("Node SSR container deployment port must be 18910 or 28910: %d", deployment.ActivePort) + } + for name, value := range map[string]string{"container name": deployment.ContainerName, "image digest": deployment.ImageDigest, "container ID": deployment.ContainerID} { + if value == "" || strings.TrimSpace(value) != value { + return fmt.Errorf("exact Node SSR container %s is required", name) + } + } + return nil +} + +func scanNodeSsrContainerDeployment(row rowScanner) (NodeSsrContainerDeployment, error) { + var deployment NodeSsrContainerDeployment + 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 NodeSsrContainerDeployment{}, ErrNotFound + } + return NodeSsrContainerDeployment{}, fmt.Errorf("scan Node SSR container deployment: %w", err) + } + parsed, err := parseTime(updatedAt) + if err != nil { + return NodeSsrContainerDeployment{}, fmt.Errorf("parse Node SSR deployment update time: %w", err) + } + deployment.UpdatedAt = parsed + return deployment, nil +} diff --git a/internal/transaction/store.go b/internal/transaction/store.go index a8edf25..c7d195a 100644 --- a/internal/transaction/store.go +++ b/internal/transaction/store.go @@ -17,7 +17,7 @@ import ( "github.com/ncruces/go-sqlite3/driver" ) -const schemaVersion = 2 +const schemaVersion = 3 // schemaV1 SQLite 数据库的首版 schema，定义事务、步骤与事件三张核心表。 // @@ -95,6 +95,21 @@ CREATE TABLE backend_container_deployment ( PRAGMA user_version = 2; ` +// schemaV3 新增 Node SSR 容器部署单例表。 +const schemaV3 = ` +CREATE TABLE node_ssr_container_deployment ( + singleton_id INTEGER PRIMARY KEY CHECK (singleton_id = 1), + active_port INTEGER NOT NULL CHECK (active_port IN (18910, 28910)), + container_name TEXT NOT NULL, + image_digest TEXT NOT NULL, + container_id TEXT NOT NULL, + transaction_id TEXT NOT NULL REFERENCES transactions(id), + updated_at TEXT NOT NULL +) STRICT; + +PRAGMA user_version = 3; +` + // Store 服务端 SQLite 事务记录。一个进程只应创建一个 Store。 // // 它封装了底层 *sql.DB，并把单连接访问（MaxOpenConns/MaxIdleConns 均为 1）作为 @@ -222,6 +237,8 @@ func migrate(ctx context.Context, db *sql.DB) error { script = schemaV1 case 2: script = schemaV2 + case 3: + script = schemaV3 default: return fmt.Errorf("sqlite migration script is missing for version %d", nextVersion) }