yms-daemon/internal/backendupdate/updater.go

// Package backendupdate 负责在单台服务器上编排一次原生后端（或容器后端）更新，
// 覆盖从构件准备、切换流量到提交或补偿的完整事务流程。它通过 transaction 包维护
// 可恢复的更新事务，并在失败时执行补偿以回滚到更新前的状态。
package backendupdate

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"path/filepath"
	"time"

	"yms-daemon/internal/backendexecutor"
	"yms-daemon/internal/containerengine"
	"yms-daemon/internal/daemonapi"
	"yms-daemon/internal/deploymentconfig"
	"yms-daemon/internal/filestore"
	"yms-daemon/internal/hostnginx"
	"yms-daemon/internal/nativebackendexecutor"
	"yms-daemon/internal/systemd"
	"yms-daemon/internal/transaction"
	"yms-daemon/internal/updatepackage"
)

const (
	// serviceBackend 本更新服务在事务中的服务标识名。
	serviceBackend = "backend"
	// sourceLocalCLI 事务来源标识，表示更新由本地命令行发起。
	sourceLocalCLI = "local-cli"
	// drainDuration 切换后旧单元/旧容器的默认排空时长。
	drainDuration = 5 * time.Second
	// directReleaseDigestLength 直接部署构件在 release 目录下的摘要路径截取长度。
	directReleaseDigestLength = 12
	// inputTypeCurrentRelease 标识以当前运行发行版执行重启的输入类型。
	inputTypeCurrentRelease = "current-native-release"
	// legacyUnit8080 端口 8080 对应的遗留 systemd 单元名。
	legacyUnit8080 = "yms.service"
	// legacyUnit8081 端口 8081 对应的遗留 systemd 单元名。
	legacyUnit8081 = "ymsback.service"
)

// Updater 在单台服务器上执行当前原生（或容器）后端契约。它持有部署配置、事务
// 存储、协调器以及各类执行器，是后端更新流程的核心入口与状态载体。
type Updater struct {
	// config 部署配置，描述后端类型、槽位与路径等信息。
	config deploymentconfig.Config
	// workRoot 事务工作目录的根路径。
	workRoot string
	// store 事务存储，用于创建、读取与推进更新事务。
	store *transaction.Store
	// coordinator 事务协调器，用于以可恢复方式执行单个步骤。
	coordinator *transaction.Coordinator
	// releaseStore 发行版文件存储，供执行器使用。
	releaseStore *filestore.Store
	// units systemd 管理器，用于检查与停止后端单元。
	units systemd.Manager
	// gateway 宿主 Nginx 配置控制器，用于读取与应用上游配置。
	gateway gatewayController
	// executor 原生后端执行器，负责运行与健康检查原生后端。
	executor nativeExecutor
	// containerExecutor 容器后端执行器，负责运行与健康检查容器后端。
	containerExecutor containerExecutor
	// engine 容器引擎，仅在容器后端更新时使用。
	engine containerengine.Engine
	// containerConfigSource 容器后端配置文件在宿主机上的源路径。
	containerConfigSource string
	// containerConfigTarget 容器后端配置文件在容器内的目标路径。
	containerConfigTarget string
	// containerTmpSource 容器后端临时目录在宿主机上的源路径。
	containerTmpSource string
	// containerTmpTarget 容器后端临时目录在容器内的目标路径。
	containerTmpTarget string
	// containerStopGraceSeconds 停止容器后端旧容器时，发送停止信号后到强制终止前的等待秒数。
	containerStopGraceSeconds int
	// logger 结构化日志记录器。
	logger *slog.Logger
	// drain 切换后旧单元/旧容器的排空时长。
	drain time.Duration
}

// gatewayController 抽象宿主 Nginx 配置的读取与应用，便于测试与替换实现。
type gatewayController interface {
	// Read 返回当前宿主 Nginx 配置快照。
	Read() (hostnginx.Snapshot, error)
	// Apply 将给定快照应用到宿主 Nginx。
	Apply(context.Context, hostnginx.Snapshot) error
}

// nativeExecutor 抽象原生后端执行器的运行能力。
type nativeExecutor interface {
	// Run 执行原生后端运行流程，request 携带构件与目标槽位等信息。
	Run(context.Context, string, nativebackendexecutor.Request) error
}

// containerExecutor 抽象容器后端执行器的运行能力。
type containerExecutor interface {
	// Run 执行容器后端运行流程，request 携带镜像与容器参数等信息。
	Run(context.Context, string, backendexecutor.Request) error
}

// New 创建完整的原生后端更新编排器。
//
// 参数 config 为部署配置且必须校验通过且后端类型为 native；workRoot 必须是干净
// 的绝对路径；store 与 coordinator 提供事务能力；releaseStore 提供发行版存储；
// units 提供 systemd 管理；gateway 提供 Nginx 控制；httpClient 供执行器进行健康
// 检查；logger 可为 nil，缺省使用默认日志器。返回构造完成的 Updater，若参数非法
// 或执行器创建失败则返回错误。
func New(
	config deploymentconfig.Config,
	workRoot string,
	store *transaction.Store,
	coordinator *transaction.Coordinator,
	releaseStore *filestore.Store,
	units systemd.Manager,
	gateway gatewayController,
	httpClient *http.Client,
	logger *slog.Logger,
) (*Updater, error) {
	if err := config.Validate(); err != nil {
		return nil, err
	}
	if config.Backend.Type != deploymentconfig.BackendTypeNative {
		return nil, errors.New("backend.type must be native")
	}
	if !filepath.IsAbs(workRoot) || filepath.Clean(workRoot) != workRoot {
		return nil, errors.New("backend update work root must be a clean absolute path")
	}
	if store == nil || coordinator == nil || releaseStore == nil || units == nil || gateway == nil {
		return nil, errors.New("backend update dependencies are required")
	}
	if logger == nil {
		logger = slog.Default()
	}
	executor, err := nativebackendexecutor.New(store, coordinator, releaseStore, units, httpClient)
	if err != nil {
		return nil, err
	}
	return &Updater{
		config:       config,
		workRoot:     workRoot,
		store:        store,
		coordinator:  coordinator,
		releaseStore: releaseStore,
		units:        units,
		gateway:      gateway,
		executor:     executor,
		logger:       logger,
		drain:        drainDuration,
	}, nil
}

// NewContainer 创建 Docker 独立容器后端更新编排器。
//
// 参数与 New 类似，但要求后端类型为 container 且 daemon 环境为 dev，engine 提供
// 容器引擎能力。返回构造完成的 Updater，若参数非法或执行器创建失败则返回错误。
func NewContainer(
	config deploymentconfig.Config,
	workRoot string,
	store *transaction.Store,
	coordinator *transaction.Coordinator,
	engine containerengine.Engine,
	gateway gatewayController,
	httpClient *http.Client,
	logger *slog.Logger,
) (*Updater, error) {
	if err := config.Validate(); err != nil {
		return nil, err
	}
	if config.Backend.Type != deploymentconfig.BackendTypeContainer {
		return nil, errors.New("backend.type must be container")
	}
	if config.Daemon.Environment != deploymentconfig.EnvironmentDev {
		return nil, errors.New("--container-image requires daemon.environment = dev")
	}
	if !filepath.IsAbs(workRoot) || filepath.Clean(workRoot) != workRoot {
		return nil, errors.New("container backend update work root must be a clean absolute path")
	}
	if store == nil || coordinator == nil || engine == nil || gateway == nil {
		return nil, errors.New("container backend update dependencies are required")
	}
	if logger == nil {
		logger = slog.Default()
	}
	executor, err := backendexecutor.New(store, coordinator, engine, httpClient)
	if err != nil {
		return nil, err
	}
	stopGraceSeconds := defaultContainerStopGraceSeconds
	if config.Backend.StopGraceSeconds != nil {
		stopGraceSeconds = *config.Backend.StopGraceSeconds
	}
	return &Updater{
		config: config, workRoot: workRoot, store: store, coordinator: coordinator, gateway: gateway,
		containerExecutor: executor, engine: engine, logger: logger, drain: drainDuration,
		containerConfigSource:     deploymentconfig.ContainerConfigSource,
		containerConfigTarget:     deploymentconfig.ContainerConfigTarget,
		containerTmpSource:        deploymentconfig.ContainerTmpSource,
		containerTmpTarget:        deploymentconfig.ContainerTmpTarget,
		containerStopGraceSeconds: stopGraceSeconds,
	}, nil
}

// UpdateRepack 应用由绝对本地路径指定的 repack ZIP 更新。
//
// 参数 packagePath 是 repack ZIP 的绝对路径；report 用于回传进度，可为 nil。
// 返回本次更新对应的事务记录以及错误。容器后端不支持该更新方式。
func (u *Updater) UpdateRepack(ctx context.Context, packagePath string, report ProgressReporter) (transaction.Transaction, error) {
	if u.config.Backend.Type == deploymentconfig.BackendTypeContainer {
		return transaction.Transaction{}, errors.New("repack ZIP update is not implemented for container backend")
	}
	reportProgress(report, Progress{Message: "Validating repack ZIP"})
	updatePackage, err := updatepackage.OpenBackendNative(packagePath)
	if err != nil {
		return transaction.Transaction{}, err
	}
	defer updatePackage.Close()
	reportProgress(report, Progress{Message: "Repack ZIP validated"})
	return u.update(ctx, updateInput{
		IdempotencyKey:   serviceBackend + ":" + updatePackage.PackageSHA256,
		InputType:        daemonapi.InputTypeRepackZIP,
		SourcePath:       updatePackage.PackagePath,
		SourceSHA256:     updatePackage.PackageSHA256,
		CustomerCode:     updatePackage.CustomerCode,
		VersionID:        updatePackage.VersionID,
		ArtifactID:       updatePackage.ArtifactID,
		ArtifactFileName: updatePackage.FileName,
		ArtifactIdentity: updatePackage.Identity,
		ReleasePath:      updatePackage.FileName,
		Materialize:      updatePackage.ExtractArtifact,
	}, report)
}

// UpdateNativeJAR 应用一个直接复制到服务器的 JAR 更新。
//
// 参数 jarPath 是 JAR 的本地路径；report 用于回传进度，可为 nil。返回本次更新
// 对应的事务记录以及错误。仅原生后端支持该更新方式。
func (u *Updater) UpdateNativeJAR(ctx context.Context, jarPath string, report ProgressReporter) (transaction.Transaction, error) {
	if u.config.Backend.Type == deploymentconfig.BackendTypeContainer {
		return transaction.Transaction{}, errors.New("--native-jar requires backend.type = native")
	}
	reportProgress(report, Progress{Message: "Validating direct native backend JAR and computing SHA-256"})
	jar, err := updatepackage.OpenDirectNativeJAR(jarPath)
	if err != nil {
		return transaction.Transaction{}, err
	}
	reportProgress(report, Progress{Message: "Direct native backend JAR validated: sha256=" + jar.SHA256})
	return u.update(ctx, updateInput{
		IdempotencyKey:   serviceBackend + ":" + jar.SHA256,
		InputType:        daemonapi.InputTypeNativeJAR,
		SourcePath:       jar.Path,
		SourceSHA256:     jar.SHA256,
		ArtifactFileName: jar.FileName,
		ArtifactIdentity: jar.Identity,
		ReleasePath:      filepath.Join("direct", jar.SHA256[:directReleaseDigestLength], jar.FileName),
		Materialize:      jar.CopyArtifact,
	}, report)
}

// update 原生后端更新（含重启）的统一执行入口：创建或恢复事务、物化构件、
// 运行执行器、再切换与提交。input 描述本次更新输入，report 用于回传进度。
func (u *Updater) update(ctx context.Context, input updateInput, report ProgressReporter) (transaction.Transaction, error) {
	existing, request, created, err := u.createOrResume(ctx, input)
	if err != nil {
		return transaction.Transaction{}, err
	}
	operation := operationLabel(input.InputType)
	transactionMessage := "Resuming backend " + operation + " transaction"
	if created {
		transactionMessage = "Created backend " + operation + " transaction"
	}
	transactionMessage += " " + existing.ID
	reportProgress(report, Progress{TransactionID: existing.ID, State: existing.State, Message: transactionMessage})
	if !created && existing.State.Terminal() {
		return terminalResult(existing)
	}
	if err := os.MkdirAll(filepath.Dir(request.ArtifactPath), 0o750); err != nil {
		return u.fail(ctx, existing.ID, fmt.Errorf("create backend transaction work directory: %w", err))
	}
	reportProgress(report, Progress{TransactionID: existing.ID, State: existing.State, Message: "Staging backend artifact in transaction workspace"})
	if err := ensureTransactionArtifact(input, request); err != nil {
		return u.fail(ctx, existing.ID, err)
	}

	executorRequest := nativebackendexecutor.Request{
		ArtifactPath:       request.ArtifactPath,
		ArtifactIdentity:   request.ArtifactIdentity,
		ReleasePath:        request.ReleasePath,
		SlotJarPath:        request.TargetSlotJAR,
		PreviousSlotTarget: request.PreviousSlotTarget,
		UnitName:           request.TargetUnit,
		Port:               request.TargetPort,
		HealthEndpoint:     request.TargetHealthEndpoint,
		Progress: func(state transaction.State, message string) {
			reportProgress(report, Progress{TransactionID: existing.ID, State: state, Message: message})
		},
	}
	switch existing.State {
	case transaction.StateCreated, transaction.StateValidating, transaction.StatePrepared, transaction.StateStarting:
		if err := u.executor.Run(ctx, existing.ID, executorRequest); err != nil {
			return u.currentWithError(ctx, existing.ID, err)
		}
	case transaction.StateSwitching, transaction.StateVerifying, transaction.StateDraining:
	case transaction.StateRollingBack:
		if err := u.executor.Run(ctx, existing.ID, executorRequest); err != nil {
			return u.currentWithError(ctx, existing.ID, err)
		}
	default:
		return u.currentWithError(ctx, existing.ID, fmt.Errorf("backend update cannot resume transaction %s in state %s", existing.ID, existing.State))
	}
	current, err := u.store.Transaction(ctx, existing.ID)
	if err != nil {
		return transaction.Transaction{}, err
	}
	if current.State.Terminal() {
		return terminalResult(current)
	}
	if err := u.switchAndCommit(ctx, existing.ID, request, report); err != nil {
		return u.currentWithError(ctx, existing.ID, err)
	}
	return u.store.Transaction(ctx, existing.ID)
}

// terminalResult 根据终态事务记录返回结果：已提交则正常返回，否则返回带有状态
// 信息的错误。
func terminalResult(record transaction.Transaction) (transaction.Transaction, error) {
	if record.State == transaction.StateCommitted {
		return record, nil
	}
	return record, fmt.Errorf("backend update transaction %s is terminal in state %s", record.ID, record.State)
}

// createOrResume 为给定更新输入创建新事务，或在幂等键命中时恢复已存在事务。
//
// 该函数先读取当前 Nginx 快照以确定目标端口与槽位，再在事务工作目录下固化网关
// 快照与兼容性链接备份，最后创建持久化请求并写入事务存储。返回值为事务记录、
// 持久化请求、是否新建以及错误。
func (u *Updater) createOrResume(ctx context.Context, input updateInput) (transaction.Transaction, persistedRequest, bool, error) {
	gatewayBefore, err := u.gateway.Read()
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	targetPort := otherPort(gatewayBefore.ActivePort)
	targetSlot, err := u.config.Backend.SlotForPort(targetPort)
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	currentSlot, err := u.config.Backend.SlotForPort(gatewayBefore.ActivePort)
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	previousUnit, err := u.resolveCurrentUnit(ctx, gatewayBefore.ActivePort, currentSlot.Unit)
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	previousSlotTarget, err := readOptionalSymlink(targetSlot.JAR)
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}

	transactionID := rand.Text()
	transactionRoot := filepath.Join(u.workRoot, transactionID)
	if err := os.MkdirAll(transactionRoot, 0o750); err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, fmt.Errorf("create backend transaction directory: %w", err)
	}
	gatewayAfterContent, err := hostnginx.RenderBackendPort(gatewayBefore.Content, targetPort)
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	gatewayBeforePath := filepath.Join(transactionRoot, "gateway.before.conf")
	gatewayAfterPath := filepath.Join(transactionRoot, "gateway.after.conf")
	if err := writeImmutableFile(gatewayBeforePath, gatewayBefore.Content, 0o640); err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	if err := writeImmutableFile(gatewayAfterPath, gatewayAfterContent, 0o640); err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}

	activeBefore, err := snapshotPath(u.config.Backend.ActiveJAR, filepath.Join(transactionRoot, "active-jar.before"))
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	request := persistedRequest{
		InputType:            input.InputType,
		SourcePath:           input.SourcePath,
		SourceSHA256:         input.SourceSHA256,
		CustomerCode:         input.CustomerCode,
		VersionID:            input.VersionID,
		ArtifactID:           input.ArtifactID,
		ArtifactFileName:     input.ArtifactFileName,
		ArtifactPath:         filepath.Join(transactionRoot, "backend.jar"),
		ArtifactIdentity:     input.ArtifactIdentity,
		ReleasePath:          input.ReleasePath,
		TargetPort:           targetPort,
		TargetUnit:           targetSlot.Unit,
		TargetSlotJAR:        targetSlot.JAR,
		TargetHealthEndpoint: targetSlot.HealthEndpoint,
		PreviousSlotTarget:   previousSlotTarget,
		PreviousGatewayPort:  gatewayBefore.ActivePort,
		PreviousUnit:         previousUnit,
		GatewayBeforePath:    gatewayBeforePath,
		GatewayAfterPath:     gatewayAfterPath,
		GatewayReceiptPath:   filepath.Join(transactionRoot, "gateway.applied"),
		ActiveJARPath:        u.config.Backend.ActiveJAR,
		ActiveJARBefore:      activeBefore,
	}
	requestJSON, err := json.Marshal(request)
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, fmt.Errorf("encode backend update request: %w", err)
	}
	record, created, err := u.store.CreateTransaction(ctx, transaction.CreateRequest{
		ID:             transactionID,
		IdempotencyKey: input.IdempotencyKey,
		Source:         sourceLocalCLI,
		Service:        serviceBackend,
		Request:        requestJSON,
	})
	if err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	if created {
		return record, request, true, nil
	}
	var persisted persistedRequest
	if err := decodePersistedRequest(record.Request, &persisted); err != nil {
		return transaction.Transaction{}, persistedRequest{}, false, err
	}
	if persisted.InputType != input.InputType || persisted.SourceSHA256 != input.SourceSHA256 {
		return transaction.Transaction{}, persistedRequest{}, false, errors.New("persisted backend transaction input identity mismatch")
	}
	return record, persisted, false, nil
}

// switchAndCommit 执行原生后端更新的切换与提交：切换 Nginx 上游到目标端口、更新
// 兼容性 JAR 链接、排空并停止前一单元，最终把事务迁入 Committed 状态。任何切换
// 阶段失败都会触发补偿。
func (u *Updater) switchAndCommit(ctx context.Context, transactionID string, request persistedRequest, report ProgressReporter) error {
	operation := operationLabel(request.InputType)
	before, err := readGatewaySnapshot(request.GatewayBeforePath, request.PreviousGatewayPort)
	if err != nil {
		return err
	}
	after, err := readGatewaySnapshot(request.GatewayAfterPath, request.TargetPort)
	if err != nil {
		return err
	}
	for {
		record, err := u.store.Transaction(ctx, transactionID)
		if err != nil {
			return err
		}
		switch record.State {
		case transaction.StateSwitching:
			reportProgress(report, Progress{TransactionID: transactionID, State: record.State, Message: fmt.Sprintf("Switching host Nginx backend traffic to port %d", request.TargetPort)})
			gatewayOperation := &gatewayOperation{
				controller:  u.gateway,
				before:      before,
				after:       after,
				receiptPath: request.GatewayReceiptPath,
			}
			if _, err := u.coordinator.ExecuteStep(ctx, transactionID, gatewaySwitchIntent(request, before, after), gatewayOperation); err != nil {
				return u.rollbackAfterPreparation(ctx, transactionID, request, before, after, err)
			}
			if _, err := u.store.Transition(ctx, transactionID, transaction.StateVerifying, "host Nginx now routes backend traffic to the healthy native slot"); err != nil {
				return err
			}
		case transaction.StateVerifying:
			reportProgress(report, Progress{TransactionID: transactionID, State: record.State, Message: "Updating the active compatibility JAR link"})
			installedPath := filepath.Join(u.config.Backend.ReleaseDir, request.ReleasePath)
			activeOperation := &pathOperation{
				path:    request.ActiveJARPath,
				before:  request.ActiveJARBefore,
				desired: pathState{Kind: pathKindSymlink, Target: installedPath},
			}
			if _, err := u.coordinator.ExecuteStep(ctx, transactionID, activeLinkIntent(request, installedPath), activeOperation); err != nil {
				return u.rollbackAfterPreparation(ctx, transactionID, request, before, after, err)
			}
			if _, err := u.store.Transition(ctx, transactionID, transaction.StateDraining, "backend compatibility link committed; previous unit draining"); err != nil {
				return err
			}
		case transaction.StateDraining:
			reportProgress(report, Progress{TransactionID: transactionID, State: record.State, Message: fmt.Sprintf("Draining previous backend unit for %s", u.drain)})
			if err := waitContext(ctx, u.drain); err != nil {
				return err
			}
			reportProgress(report, Progress{TransactionID: transactionID, State: record.State, Message: "Stopping previous backend unit " + request.PreviousUnit})
			stopOperation := &unitStopOperation{units: u.units, name: request.PreviousUnit}
			if _, err := u.coordinator.ExecuteStep(ctx, transactionID, stopPreviousUnitIntent(request), stopOperation); err != nil {
				return err
			}
			_, err = u.store.Transition(ctx, transactionID, transaction.StateCommitted, "native backend "+operation+" committed")
			if err == nil {
				reportProgress(report, Progress{TransactionID: transactionID, State: transaction.StateCommitted, Message: "Native backend " + operation + " committed"})
			}
			return err
		case transaction.StateCommitted:
			reportProgress(report, Progress{TransactionID: transactionID, State: record.State, Message: "Native backend " + operation + " already committed"})
			return nil
		default:
			return fmt.Errorf("backend commit cannot continue transaction %s in state %s", transactionID, record.State)
		}
	}
}

// reportProgress 在 report 非空时向其投递一条进度事件，report 为 nil 时静默忽略。
func reportProgress(report ProgressReporter, progress Progress) {
	if report != nil {
		report(progress)
	}
}

// rollbackAfterPreparation 在切换阶段失败后执行原生后端补偿：恢复兼容性链接、
// 恢复 Nginx 上游、停止目标单元并恢复目标槽位，最后迁入 RolledBack 状态。cause
// 为触发补偿的原始错误，会与补偿过程中的错误合并返回。
func (u *Updater) rollbackAfterPreparation(ctx context.Context, transactionID string, request persistedRequest, before hostnginx.Snapshot, after hostnginx.Snapshot, cause error) error {
	record, readErr := u.store.Transaction(ctx, transactionID)
	if readErr != nil {
		return errors.Join(cause, readErr)
	}
	if record.State != transaction.StateRollingBack {
		if _, err := u.store.Transition(ctx, transactionID, transaction.StateRollingBack, "native backend post-start compensation started"); err != nil {
			return errors.Join(cause, err)
		}
	}

	activeRestore := &pathOperation{
		path:    request.ActiveJARPath,
		before:  pathState{Kind: pathKindSymlink, Target: filepath.Join(u.config.Backend.ReleaseDir, request.ReleasePath)},
		desired: request.ActiveJARBefore,
	}
	_, activeErr := u.coordinator.ExecuteStep(ctx, transactionID, activeLinkRestoreIntent(request), activeRestore)
	gatewayRestore := &gatewayOperation{
		controller:  u.gateway,
		before:      after,
		after:       before,
		receiptPath: request.GatewayReceiptPath + ".restore",
	}
	_, gatewayErr := u.coordinator.ExecuteStep(ctx, transactionID, gatewayRestoreIntent(request), gatewayRestore)
	stopTarget := &unitStopOperation{units: u.units, name: request.TargetUnit}
	_, stopErr := u.coordinator.ExecuteStep(ctx, transactionID, stopTargetUnitIntent(request), stopTarget)
	installedPath := filepath.Join(u.config.Backend.ReleaseDir, request.ReleasePath)
	previousSlotState := pathState{Kind: pathKindAbsent}
	if request.PreviousSlotTarget != "" {
		previousSlotState = pathState{Kind: pathKindSymlink, Target: request.PreviousSlotTarget}
	}
	slotRestore := &pathOperation{
		path:    request.TargetSlotJAR,
		before:  pathState{Kind: pathKindSymlink, Target: installedPath},
		desired: previousSlotState,
	}
	_, slotErr := u.coordinator.ExecuteStep(ctx, transactionID, restoreTargetSlotIntent(request, installedPath), slotRestore)
	if err := errors.Join(activeErr, gatewayErr, stopErr, slotErr); err != nil {
		return errors.Join(cause, err)
	}
	_, transitionErr := u.store.Transition(ctx, transactionID, transaction.StateRolledBack, "native backend post-start compensation completed")
	return errors.Join(cause, transitionErr)
}

// resolveCurrentUnit 确定当前活动端口实际运行的后端单元名。它同时检查配置单元与
// 遗留单元，要求二者恰好一个在运行，并返回运行中的那个。
func (u *Updater) resolveCurrentUnit(ctx context.Context, port int, configuredUnit string) (string, error) {
	legacyUnit, err := legacyUnitForPort(port)
	if err != nil {
		return "", err
	}
	configured, err := u.units.Inspect(ctx, configuredUnit)
	if err != nil {
		return "", fmt.Errorf("inspect configured active-port unit %s: %w", configuredUnit, err)
	}
	legacy, err := u.units.Inspect(ctx, legacyUnit)
	if err != nil {
		return "", fmt.Errorf("inspect legacy active-port unit %s: %w", legacyUnit, err)
	}
	configuredRunning := unitRunning(configured)
	legacyRunning := unitRunning(legacy)
	if configuredRunning == legacyRunning {
		return "", fmt.Errorf("backend port %d requires exactly one running unit, configured=%s(%s), legacy=%s(%s)", port, configuredUnit, configured.ActiveState, legacyUnit, legacy.ActiveState)
	}
	if configuredRunning {
		return configuredUnit, nil
	}
	return legacyUnit, nil
}

// fail 把事务迁入 Failed 状态并返回带错误的事务记录。
func (u *Updater) fail(ctx context.Context, transactionID string, cause error) (transaction.Transaction, error) {
	_, transitionErr := u.store.Transition(ctx, transactionID, transaction.StateFailed, cause.Error())
	return u.currentWithError(ctx, transactionID, errors.Join(cause, transitionErr))
}

// currentWithError 读取事务当前记录并把给定错误与读取错误合并返回，便于调用方在
// 出错时仍拿到最新事务状态。
func (u *Updater) currentWithError(ctx context.Context, transactionID string, cause error) (transaction.Transaction, error) {
	record, err := u.store.Transaction(ctx, transactionID)
	return record, errors.Join(cause, err)
}

// otherPort 返回给定后端端口的对侧端口：8080 与 8081 互换。
func otherPort(port int) int {
	if port == deploymentconfig.BackendPort8080 {
		return deploymentconfig.BackendPort8081
	}
	return deploymentconfig.BackendPort8080
}

// legacyUnitForPort 返回给定后端端口对应的遗留 systemd 单元名，仅支持 8080 与
// 8081 两个端口。
func legacyUnitForPort(port int) (string, error) {
	switch port {
	case deploymentconfig.BackendPort8080:
		return legacyUnit8080, nil
	case deploymentconfig.BackendPort8081:
		return legacyUnit8081, nil
	default:
		return "", fmt.Errorf("unsupported legacy backend port: %d", port)
	}
}

// unitRunning 判断单元是否处于运行状态：既非 inactive 也非 failed 即视为运行。
func unitRunning(unit systemd.Unit) bool {
	return unit.ActiveState != "inactive" && unit.ActiveState != "failed"
}

// readOptionalSymlink 读取指定路径的符号链接目标；若路径不存在则返回空字符串，
// 若路径存在但不是符号链接则报错。
func readOptionalSymlink(path string) (string, error) {
	info, err := os.Lstat(path)
	if errors.Is(err, os.ErrNotExist) {
		return "", nil
	}
	if err != nil {
		return "", fmt.Errorf("inspect native backend target slot %s: %w", path, err)
	}
	if info.Mode()&os.ModeSymlink == 0 {
		return "", fmt.Errorf("native backend target slot is not a symbolic link: %s", path)
	}
	target, err := os.Readlink(path)
	if err != nil {
		return "", fmt.Errorf("read native backend target slot %s: %w", path, err)
	}
	return target, nil
}

// ensureTransactionArtifact 确保构件已物化到事务工作路径：若已存在则校验其为直接
// 普通文件且身份匹配，否则调用 input.Materialize 进行物化。
func ensureTransactionArtifact(input updateInput, request persistedRequest) error {
	info, err := os.Lstat(request.ArtifactPath)
	if errors.Is(err, os.ErrNotExist) {
		return input.Materialize(request.ArtifactPath)
	}
	if err != nil {
		return fmt.Errorf("inspect extracted native backend artifact: %w", err)
	}
	if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 {
		return errors.New("extracted native backend artifact is not a direct regular file")
	}
	return verifyFileIdentity(request.ArtifactPath, request.ArtifactIdentity)
}

// decodePersistedRequest 将持久化的原生后端更新请求 JSON 反序列化到目标结构体，
// 并禁止出现未知字段。
func decodePersistedRequest(content json.RawMessage, request *persistedRequest) error {
	decoder := json.NewDecoder(bytes.NewReader(content))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(request); err != nil {
		return fmt.Errorf("decode persisted backend update request: %w", err)
	}
	return nil
}

// waitContext 等待指定时长，或直到 ctx 被取消。返回 ctx 取消错误或 nil。
func waitContext(ctx context.Context, duration time.Duration) error {
	timer := time.NewTimer(duration)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}