yms-daemon/internal/nativebackendexecutor/executor.go

// Package nativebackendexecutor 负责准备并启动一个由外部显式配置的原生后端槽位。
// 它把后端 JAR 安装到发布目录、把槽位软链接指向新版本、启动对应的 systemd 单元并等待健康检查通过，
// 从而把持久化的事务推进到 switching 状态。网关流量切换被刻意排除在本包职责之外。
package nativebackendexecutor

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"yms-daemon/internal/filestore"
	"yms-daemon/internal/healthcheck"
	"yms-daemon/internal/systemd"
	"yms-daemon/internal/transaction"
)

const (
	// healthPath 健康检查端点必须精确匹配的 URL 路径。
	healthPath = "/yms/actuator/health"
	// healthTimeout 等待原生后端 Actuator 健康检查就绪的最长时限。
	healthTimeout = 120 * time.Second
	// healthInterval 健康检查器轮询端点时的间隔。
	healthInterval = time.Second
	// stepInstallJar “安装不可变原生后端 JAR”事务步骤的持久化键。
	stepInstallJar = "backend.native.jar.install"
	// stepBindSlot “绑定非活跃槽位软链接”事务步骤的持久化键。
	stepBindSlot = "backend.native.slot.bind"
	// stepStartUnit “启动非活跃原生后端 systemd 单元”事务步骤的持久化键。
	stepStartUnit = "backend.native.service.start"
	// stepCheckHealth “等待原生后端 Actuator 健康检查”事务步骤的持久化键。
	stepCheckHealth = "backend.native.health"
	// stepStopUnit “停止失败的非活跃原生后端 systemd 单元”事务步骤的持久化键。
	stepStopUnit = "backend.native.service.stop"
	// stepRestoreSlot “恢复非活跃槽位软链接”事务步骤的持久化键。
	stepRestoreSlot = "backend.native.slot.restore"
	// activeState systemd 单元的“活跃”运行状态标识。
	activeState = "active"
	// inactiveState systemd 单元的“非活跃”运行状态标识。
	inactiveState = "inactive"
	// failedState systemd 单元的“失败”运行状态标识。
	failedState = "failed"
)

// Request 汇集了不可变更新请求中的精确值与本地部署配置。
// 其中 UnitName、SlotJarPath 与 PreviousSlotTarget 都是不透明值，绝不从文件名或端口号推导得出。
type Request struct {
	// ArtifactPath 待安装后端 JAR 的绝对源路径。
	ArtifactPath string
	// ArtifactIdentity 后端 JAR 的期望身份（大小与 SHA-256），用于校验与发布存储。
	ArtifactIdentity filestore.Identity
	// ReleasePath 发布存储内的本地相对路径。
	ReleasePath string
	// SlotJarPath 槽位软链接的绝对路径。
	SlotJarPath string
	// PreviousSlotTarget 槽位软链接此前指向的绝对路径，首次部署时可为空。
	PreviousSlotTarget string
	// UnitName 非活跃后端对应的精确 systemd 单元名。
	UnitName string
	// Port 后端监听的端口，只允许 8080 或 8081。
	Port int
	// HealthEndpoint 健康检查的 HTTP URL，路径必须精确等于 healthPath。
	HealthEndpoint string
	// Progress 可选的回调，用于把事务状态与进度消息上报给上层；为空时不回调。
	Progress func(transaction.State, string)
}

// actuatorChecker 抽象了 Actuator 健康检查器，供 Executor 依赖注入使用。
// 它把真实的 healthcheck.ActuatorChecker 与测试替身统一起来。
type actuatorChecker interface {
	// Check 对给定端点执行一次健康检查并立即返回报告、就绪标志与错误。
	Check(context.Context, string, healthcheck.RunningProbe) (healthcheck.ActuatorReport, bool, error)
	// Wait 反复检查端点直到就绪或超过给定时限，并返回最终报告与错误。
	Wait(context.Context, string, time.Duration, healthcheck.RunningProbe) (healthcheck.ActuatorReport, error)
}

// Executor 在非活跃原生后端健康之后，把持久化事务推进到 switching 状态。
// 它负责校验、安装、绑定槽位、启动单元与健康检查，但不负责切换网关流量。
type Executor struct {
	store        *transaction.Store
	coordinator  *transaction.Coordinator
	releaseStore *filestore.Store
	units        systemd.Manager
	checker      actuatorChecker
}

// New 构造一个原生后端执行器。
// store、coordinator、releaseStore、units 为 nil 时返回错误；httpClient 用于构造 Actuator 健康检查器。
// 返回值是就绪可用的 *Executor；错误只在缺少必要依赖或健康检查器构造失败时非空。
func New(store *transaction.Store, coordinator *transaction.Coordinator, releaseStore *filestore.Store, units systemd.Manager, httpClient *http.Client) (*Executor, error) {
	if store == nil {
		return nil, errors.New("transaction store is required")
	}
	if coordinator == nil {
		return nil, errors.New("transaction coordinator is required")
	}
	if releaseStore == nil {
		return nil, errors.New("native backend release store is required")
	}
	if units == nil {
		return nil, errors.New("systemd manager is required")
	}
	checker, err := healthcheck.NewActuatorChecker(httpClient, healthInterval)
	if err != nil {
		return nil, err
	}
	return &Executor{store: store, coordinator: coordinator, releaseStore: releaseStore, units: units, checker: checker}, nil
}

// Run 从事务持久化的状态恢复并继续推进，直到 switching 状态后返回。
// ctx 用于取消与超时；transactionID 是待推进的事务 ID，空白时返回错误；request 携带不可变请求与部署配置。
// 它通过协调器的排他执行保证同一事务不会并发运行，且不切换网关流量。
// 返回值是执行过程中的错误；若事务已达到 switching 状态则返回 nil。
func (e *Executor) Run(ctx context.Context, transactionID string, request Request) error {
	if strings.TrimSpace(transactionID) == "" {
		return errors.New("transaction ID is required")
	}
	return e.coordinator.RunExclusive(ctx, func(ctx context.Context) error {
		return e.run(ctx, transactionID, request)
	})
}

// run 在排他锁内的实际状态机，根据持久化事务状态执行对应步骤。
// ctx 用于取消与超时；transactionID 定位事务；request 携带不可变请求与部署配置。
// 它循环读取事务状态并推进：校验、安装与绑定、启动与健康检查、补偿恢复，直至 switching 或终止状态。
// 返回值是推进过程中的错误；可恢复错误会被原样返回以便重试，不可恢复错误会写入 failed 状态。
func (e *Executor) run(ctx context.Context, transactionID string, request Request) error {
	for {
		record, err := e.store.Transaction(ctx, transactionID)
		if err != nil {
			return err
		}
		switch record.State {
		case transaction.StateCreated:
			if _, err := e.store.Transition(ctx, transactionID, transaction.StateValidating, "native backend validation started"); err != nil {
				return err
			}
		case transaction.StateValidating:
			reportProgress(request, record.State, "Validating inactive native backend slot")
			if err := e.validate(ctx, request); err != nil {
				_, transitionErr := e.store.Transition(ctx, transactionID, transaction.StateFailed, err.Error())
				return errors.Join(err, transitionErr)
			}
			if _, err := e.store.Transition(ctx, transactionID, transaction.StatePrepared, "native backend inputs validated"); err != nil {
				return err
			}
		case transaction.StatePrepared:
			reportProgress(request, record.State, "Installing backend JAR and binding the inactive slot")
			if _, err := e.prepare(ctx, transactionID, request); err != nil {
				return e.failUnlessRecoverable(ctx, transactionID, err)
			}
			if _, err := e.store.Transition(ctx, transactionID, transaction.StateStarting, "native backend slot prepared"); err != nil {
				return err
			}
		case transaction.StateStarting:
			reportProgress(request, record.State, fmt.Sprintf("Preparing to start %s on port %d", request.UnitName, request.Port))
			installedPath, err := e.prepare(ctx, transactionID, request)
			if err != nil {
				return e.failUnlessRecoverable(ctx, transactionID, err)
			}
			if err := e.startAndCheck(ctx, transactionID, request); err != nil {
				if recoverable(err) {
					return err
				}
				return e.rollbackBeforeSwitch(ctx, transactionID, request, installedPath, err)
			}
			if _, err := e.store.Transition(ctx, transactionID, transaction.StateSwitching, "native backend is healthy"); err != nil {
				return err
			}
		case transaction.StateSwitching:
			return nil
		case transaction.StateRollingBack:
			reportProgress(request, record.State, "Resuming native backend preparation compensation")
			installed, found, err := e.releaseStore.Inspect(request.ReleasePath, request.ArtifactIdentity)
			if err != nil {
				return fmt.Errorf("inspect native backend JAR while resuming compensation: %w", err)
			}
			if !found {
				return errors.New("installed native backend JAR is missing while compensation is pending")
			}
			if err := e.compensateBeforeSwitch(ctx, transactionID, request, installed.Path); err != nil {
				return err
			}
			if _, err := e.store.Transition(ctx, transactionID, transaction.StateRolledBack, "native backend preparation rollback resumed and completed"); err != nil {
				return err
			}
			return nil
		default:
			return fmt.Errorf("native backend executor cannot run transaction %s in state %s", transactionID, record.State)
		}
	}
}

// validate 校验请求与部署前置条件。
// ctx 用于取消；request 携带待校验的请求与配置。
// 校验内容包括：请求字段合法性、JAR 源文件为普通文件且大小匹配、槽位软链接初始状态正确、
// 以及目标 systemd 单元必须处于非活跃或失败状态。返回错误时不会改动任何文件或事务状态。
func (e *Executor) validate(ctx context.Context, request Request) error {
	if err := validateRequest(request); err != nil {
		return err
	}
	info, err := os.Lstat(request.ArtifactPath)
	if err != nil {
		return fmt.Errorf("inspect native backend JAR %s: %w", request.ArtifactPath, err)
	}
	if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 {
		return fmt.Errorf("native backend JAR is not a regular file: %s", request.ArtifactPath)
	}
	if info.Size() != request.ArtifactIdentity.Size {
		return fmt.Errorf("native backend JAR size mismatch: got %d, want %d", info.Size(), request.ArtifactIdentity.Size)
	}
	if err := inspectInitialSlot(request.SlotJarPath, request.PreviousSlotTarget); err != nil {
		return err
	}
	unit, err := e.units.Inspect(ctx, request.UnitName)
	if err != nil {
		return fmt.Errorf("inspect inactive native backend unit: %w", err)
	}
	if unit.ActiveState != inactiveState && unit.ActiveState != failedState {
		return fmt.Errorf("native backend unit %s must be inactive or failed before preparation, active state is %q", request.UnitName, unit.ActiveState)
	}
	return nil
}

// prepare 通过协调器执行安装 JAR 与绑定槽位两个步骤，具备幂等性。
// ctx 用于取消；transactionID 定位事务；request 携带源路径、发布路径、身份与槽位配置。
// 返回已安装 JAR 的绝对路径；错误来自任一事务步骤或发布存储检查失败。
func (e *Executor) prepare(ctx context.Context, transactionID string, request Request) (string, error) {
	installOperation := &installJarOperation{
		store:       e.releaseStore,
		sourcePath:  request.ArtifactPath,
		releasePath: request.ReleasePath,
		identity:    request.ArtifactIdentity,
	}
	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, installIntent(request), installOperation); err != nil {
		return "", err
	}
	installed, found, err := e.releaseStore.Inspect(request.ReleasePath, request.ArtifactIdentity)
	if err != nil {
		return "", fmt.Errorf("inspect installed native backend JAR: %w", err)
	}
	if !found {
		return "", errors.New("installed native backend JAR is missing after completed install step")
	}

	bindOperation := &slotLinkOperation{
		path:           request.SlotJarPath,
		desiredTarget:  installed.Path,
		previousTarget: request.PreviousSlotTarget,
	}
	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, bindIntent(request, installed.Path), bindOperation); err != nil {
		return "", err
	}
	return installed.Path, nil
}

// startAndCheck 通过协调器执行启动 systemd 单元与等待健康检查两个步骤。
// ctx 用于取消；transactionID 定位事务；request 提供单元名、健康端点与进度回调。
// 返回值是任一事务步骤的错误；健康检查成功后还会上报“状态为 UP”的进度。
func (e *Executor) startAndCheck(ctx context.Context, transactionID string, request Request) error {
	reportProgress(request, transaction.StateStarting, "Starting native backend unit "+request.UnitName)
	startOperation := &unitStartOperation{units: e.units, name: request.UnitName}
	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, startIntent(request), startOperation); err != nil {
		return err
	}
	reportProgress(request, transaction.StateStarting, fmt.Sprintf("Waiting up to %s for Actuator health: %s", healthTimeout, request.HealthEndpoint))
	healthOperation := &healthOperation{
		units:    e.units,
		checker:  e.checker,
		unitName: request.UnitName,
		endpoint: request.HealthEndpoint,
		timeout:  healthTimeout,
	}
	_, err := e.coordinator.ExecuteStep(ctx, transactionID, healthIntent(request), healthOperation)
	if err == nil {
		reportProgress(request, transaction.StateStarting, "Actuator health status is UP")
	}
	return err
}

// reportProgress 在 request.Progress 非空时向调用方上报进度。
// request 提供 Progress 回调；state 是当前事务状态；message 是进度描述。
// 该函数无返回值且无副作用（除了可选回调），Progress 为空时直接跳过。
func reportProgress(request Request, state transaction.State, message string) {
	if request.Progress != nil {
		request.Progress(state, message)
	}
}

// rollbackBeforeSwitch 在切换网关前因启动或健康检查失败而启动补偿流程。
// ctx 用于取消；transactionID 定位事务；request 提供单元名与槽位配置；
// installedPath 本次已安装的 JAR 路径；cause 是触发补偿的原始错误。
// 它先把事务置为 rolling back，再补偿并置为 rolled back，最终总是返回 cause 以保留原始错误上下文。
func (e *Executor) rollbackBeforeSwitch(ctx context.Context, transactionID string, request Request, installedPath string, cause error) error {
	if _, err := e.store.Transition(ctx, transactionID, transaction.StateRollingBack, "native backend preparation failed; compensation started"); err != nil {
		return errors.Join(cause, err)
	}
	if err := e.compensateBeforeSwitch(ctx, transactionID, request, installedPath); err != nil {
		return errors.Join(cause, err)
	}
	if _, err := e.store.Transition(ctx, transactionID, transaction.StateRolledBack, "native backend preparation rolled back"); err != nil {
		return errors.Join(cause, err)
	}
	return cause
}

// compensateBeforeSwitch 执行切换前的补偿：停止后端单元并把槽位软链接恢复到先前的目标。
// ctx 用于取消；transactionID 定位事务；request 提供单元名与槽位配置；installedPath 是本次安装的 JAR 路径。
// 返回值是停止或恢复步骤的错误；两个步骤都通过协调器执行以保持幂等。
func (e *Executor) compensateBeforeSwitch(ctx context.Context, transactionID string, request Request, installedPath string) error {
	stopOperation := &unitStopOperation{units: e.units, name: request.UnitName}
	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, stopIntent(request), stopOperation); err != nil {
		return err
	}
	restoreOperation := &slotLinkOperation{
		path:           request.SlotJarPath,
		desiredTarget:  request.PreviousSlotTarget,
		previousTarget: installedPath,
	}
	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, restoreIntent(request, installedPath), restoreOperation); err != nil {
		return err
	}
	return nil
}

// failUnlessRecoverable 判断错误是否可恢复：可恢复时原样返回，否则把事务置为 failed 状态。
// ctx 用于取消；transactionID 定位事务；cause 是待判定的错误。
// 返回值为 cause 与（可选的）状态转换错误的合并结果。
func (e *Executor) failUnlessRecoverable(ctx context.Context, transactionID string, cause error) error {
	if recoverable(cause) {
		return cause
	}
	_, transitionErr := e.store.Transition(ctx, transactionID, transaction.StateFailed, cause.Error())
	return errors.Join(cause, transitionErr)
}

// recoverable 判断错误是否允许重试，即是否为不确定步骤错误或步骤冲突错误。
// cause 待判定的错误。若可通过重试恢复则返回 true，否则返回 false。
func recoverable(cause error) bool {
	var uncertain *transaction.UncertainStepError
	return errors.As(cause, &uncertain) || errors.Is(cause, transaction.ErrStepConflict)
}

// validateRequest 校验 Request 中所有字段的合法性，不产生任何副作用。
// request 待校验的请求。任一字段不满足约束即返回错误；全部通过则返回 nil。
func validateRequest(request Request) error {
	if !filepath.IsAbs(request.ArtifactPath) {
		return errors.New("native backend JAR path must be absolute")
	}
	if err := request.ArtifactIdentity.Validate(); err != nil {
		return fmt.Errorf("invalid native backend JAR identity: %w", err)
	}
	if !filepath.IsLocal(request.ReleasePath) || request.ReleasePath == "." {
		return fmt.Errorf("native backend release path must be a local relative path: %q", request.ReleasePath)
	}
	if !filepath.IsAbs(request.SlotJarPath) {
		return errors.New("native backend slot JAR path must be absolute")
	}
	if request.PreviousSlotTarget != "" && !filepath.IsAbs(request.PreviousSlotTarget) {
		return errors.New("previous native backend slot target must be empty or absolute")
	}
	if request.UnitName == "" || strings.TrimSpace(request.UnitName) != request.UnitName {
		return errors.New("exact native backend systemd unit name is required")
	}
	if request.Port != 8080 && request.Port != 8081 {
		return fmt.Errorf("native backend port must be 8080 or 8081: %d", request.Port)
	}
	parsed, err := url.ParseRequestURI(request.HealthEndpoint)
	if err != nil || parsed.Scheme != "http" || parsed.Host == "" || parsed.Path != healthPath {
		return fmt.Errorf("health endpoint must be an HTTP URL with exact path %s", healthPath)
	}
	if parsed.Port() != strconv.Itoa(request.Port) {
		return fmt.Errorf("health endpoint port must equal native backend port %d", request.Port)
	}
	return nil
}

// inspectInitialSlot 校验槽位软链接的初始状态与期望的先前目标一致。
// path 槽位软链接路径；previousTarget 是期望指向的先前目标，首次部署时可为空。
// 返回错误的条件包括：软链接缺失但与期望不符、路径不是软链接、目标不符，或先前目标不是普通文件。
func inspectInitialSlot(path, previousTarget string) error {
	info, err := os.Lstat(path)
	if errors.Is(err, os.ErrNotExist) {
		if previousTarget == "" {
			return nil
		}
		return fmt.Errorf("native backend slot link %s is missing; expected target %s", path, previousTarget)
	}
	if err != nil {
		return fmt.Errorf("inspect native backend slot link %s: %w", path, err)
	}
	if info.Mode()&os.ModeSymlink == 0 {
		return fmt.Errorf("native backend slot path is not a symbolic link: %s", path)
	}
	target, err := os.Readlink(path)
	if err != nil {
		return fmt.Errorf("read native backend slot link %s: %w", path, err)
	}
	if target != previousTarget {
		return fmt.Errorf("native backend slot link %s targets %q, expected %q", path, target, previousTarget)
	}
	if previousTarget != "" {
		targetInfo, err := os.Lstat(previousTarget)
		if err != nil {
			return fmt.Errorf("inspect previous native backend slot target %s: %w", previousTarget, err)
		}
		if !targetInfo.Mode().IsRegular() || targetInfo.Mode()&os.ModeSymlink != 0 {
			return fmt.Errorf("previous native backend slot target is not a regular file: %s", previousTarget)
		}
	}
	return nil
}

// installIntent 构造“安装不可变原生后端 JAR”的事务步骤意图。
// request 提供源路径、发布路径与身份。返回值是序列化后的事务步骤意图。
func installIntent(request Request) transaction.StepIntent {
	return intent(stepInstallJar, "install immutable native backend JAR", struct {
		SourcePath  string             `json:"sourcePath"`
		ReleasePath string             `json:"releasePath"`
		Identity    filestore.Identity `json:"identity"`
	}{request.ArtifactPath, request.ReleasePath, request.ArtifactIdentity})
}

// bindIntent 构造“绑定非活跃槽位软链接”的事务步骤意图。
// request 提供槽位路径与先前目标；installedPath 是本次安装后的 JAR 路径。返回值是事务步骤意图。
func bindIntent(request Request, installedPath string) transaction.StepIntent {
	return intent(stepBindSlot, "bind inactive native backend slot", struct {
		SlotJarPath    string `json:"slotJarPath"`
		PreviousTarget string `json:"previousTarget"`
		InstalledPath  string `json:"installedPath"`
	}{request.SlotJarPath, request.PreviousSlotTarget, installedPath})
}

// startIntent 构造“启动非活跃原生后端 systemd 单元”的事务步骤意图。
// request 提供单元名。返回值是事务步骤意图。
func startIntent(request Request) transaction.StepIntent {
	return intent(stepStartUnit, "start inactive native backend systemd unit", struct {
		UnitName string `json:"unitName"`
	}{request.UnitName})
}

// healthIntent 构造“等待原生后端 Actuator 健康检查”的事务步骤意图。
// request 提供单元名与端点，超时使用包级常量 healthTimeout。返回值是事务步骤意图。
func healthIntent(request Request) transaction.StepIntent {
	return intent(stepCheckHealth, "wait for native backend Actuator health", struct {
		UnitName string        `json:"unitName"`
		Endpoint string        `json:"endpoint"`
		Timeout  time.Duration `json:"timeout"`
	}{request.UnitName, request.HealthEndpoint, healthTimeout})
}

// stopIntent 构造“停止失败的非活跃原生后端 systemd 单元”的事务步骤意图。
// request 提供单元名。返回值是事务步骤意图。
func stopIntent(request Request) transaction.StepIntent {
	return intent(stepStopUnit, "stop failed inactive native backend systemd unit", struct {
		UnitName string `json:"unitName"`
	}{request.UnitName})
}

// restoreIntent 构造“恢复非活跃槽位软链接”的事务步骤意图。
// request 提供槽位路径与先前目标；installedPath 是本次安装后需被替换的 JAR 路径。返回值是事务步骤意图。
func restoreIntent(request Request, installedPath string) transaction.StepIntent {
	return intent(stepRestoreSlot, "restore inactive native backend slot", struct {
		SlotJarPath    string `json:"slotJarPath"`
		InstalledPath  string `json:"installedPath"`
		PreviousTarget string `json:"previousTarget"`
	}{request.SlotJarPath, installedPath, request.PreviousSlotTarget})
}

// intent 用键、名称与值构造事务步骤意图，其中值会被序列化为 JSON。
// key 步骤的持久化键；name 是人类可读的步骤名；value 是待序列化的载荷。
// 序列化失败时直接 panic（内部载荷应始终可序列化）。返回值是事务步骤意图。
func intent(key, name string, value any) transaction.StepIntent {
	payload, err := json.Marshal(value)
	if err != nil {
		panic(fmt.Sprintf("marshal internal step intent: %v", err))
	}
	return transaction.StepIntent{Key: key, Name: name, Intent: payload}
}