yms-daemon/internal/backendupdate/restart.go

package backendupdate

import (
	"context"
	"crypto/rand"
	"errors"
	"fmt"
	"os"
	"path/filepath"

	"yms-daemon/internal/deploymentconfig"
	"yms-daemon/internal/transaction"
	"yms-daemon/internal/updatepackage"
)

// Restart 使用兼容性 JAR 链接当前指向的精确发行版执行一次零停机轮换：它把
// 当前运行中的发行版部署到非活动槽位，然后切换流量并停止前一单元，从而在
// 不更换版本的前提下完成一次重启。
//
// 参数 ctx 用于控制整个更新过程的取消；report 用于回传实时进度，可为 nil。
// 返回值为本次重启对应的事务记录以及错误。若后端类型为容器，则直接返回错误；
// 若已存在相关活动事务，则尝试复用并续跑。
func (u *Updater) Restart(ctx context.Context, report ProgressReporter) (transaction.Transaction, error) {
	if u.config.Backend.Type == deploymentconfig.BackendTypeContainer {
		return u.RestartContainer(ctx, report)
	}
	reportProgress(report, Progress{Message: "Resolving the current native backend release"})
	active, err := u.store.ActiveTransaction(ctx)
	if err == nil {
		if active.Service != serviceBackend {
			return active, &transaction.ActiveTransactionError{TransactionID: active.ID}
		}
		var request persistedRequest
		if decodeErr := decodePersistedRequest(active.Request, &request); decodeErr != nil {
			return active, decodeErr
		}
		if request.InputType != inputTypeCurrentRelease {
			return active, &transaction.ActiveTransactionError{TransactionID: active.ID}
		}
		input, inputErr := u.persistedRestartInput(request, active.IdempotencyKey)
		if inputErr != nil {
			return active, inputErr
		}
		return u.update(ctx, input, report)
	}
	if !errors.Is(err, transaction.ErrNotFound) {
		return transaction.Transaction{}, err
	}

	sourcePath, err := currentReleaseSource(u.config.Backend.ActiveJAR)
	if err != nil {
		return transaction.Transaction{}, err
	}
	jar, err := updatepackage.OpenDirectNativeJAR(sourcePath)
	if err != nil {
		return transaction.Transaction{}, err
	}
	releasePath, err := u.restartReleasePath(jar)
	if err != nil {
		return transaction.Transaction{}, err
	}
	restartID := rand.Text()
	return u.update(ctx, updateInput{
		IdempotencyKey:   serviceBackend + ":restart:" + restartID,
		InputType:        inputTypeCurrentRelease,
		SourcePath:       jar.Path,
		SourceSHA256:     jar.SHA256,
		ArtifactFileName: jar.FileName,
		ArtifactIdentity: jar.Identity,
		ReleasePath:      releasePath,
		Materialize:      jar.CopyArtifact,
	}, report)
}

// currentReleaseSource 解析兼容性 JAR 链接实际指向的源文件路径。若链接为
// 直接普通文件则原样返回；若为符号链接则要求其为绝对目标并解析到最终文件。
func currentReleaseSource(activeJAR string) (string, error) {
	info, err := os.Lstat(activeJAR)
	if err != nil {
		return "", fmt.Errorf("inspect active compatibility JAR %s: %w", activeJAR, err)
	}
	if info.Mode().IsRegular() && info.Mode()&os.ModeSymlink == 0 {
		return activeJAR, nil
	}
	if info.Mode()&os.ModeSymlink == 0 {
		return "", fmt.Errorf("active compatibility JAR is neither a direct regular file nor symbolic link: %s", activeJAR)
	}
	target, err := os.Readlink(activeJAR)
	if err != nil {
		return "", fmt.Errorf("read active compatibility JAR link %s: %w", activeJAR, err)
	}
	if !filepath.IsAbs(target) {
		return "", fmt.Errorf("active compatibility JAR link target must be absolute: %s", target)
	}
	resolved, err := filepath.EvalSymlinks(target)
	if err != nil {
		return "", fmt.Errorf("resolve active compatibility JAR target %s: %w", target, err)
	}
	return resolved, nil
}

// restartReleasePath 计算当前 JAR 在 release 目录下的相对安装路径。若当前 JAR
// 位于 release 目录内，则返回其相对路径；否则落到 direct 目录下的规范路径。
func (u *Updater) restartReleasePath(jar updatepackage.DirectNativeJAR) (string, error) {
	resolvedReleaseDir, err := filepath.EvalSymlinks(u.config.Backend.ReleaseDir)
	if err != nil {
		return "", fmt.Errorf("resolve native backend release directory: %w", err)
	}
	relative, err := filepath.Rel(resolvedReleaseDir, jar.Path)
	if err != nil {
		return "", fmt.Errorf("compare current JAR with native backend release directory: %w", err)
	}
	if relative != "." && filepath.IsLocal(relative) {
		return relative, nil
	}
	return filepath.Join("direct", jar.SHA256[:directReleaseDigestLength], jar.FileName), nil
}

// persistedRestartInput 由已持久化的重启请求重建 updateInput，供续跑已存在的
// 重启事务使用。其中物化函数从 restartMaterializer 得到。
func (u *Updater) persistedRestartInput(request persistedRequest, idempotencyKey string) (updateInput, error) {
	materialize, err := u.restartMaterializer(request)
	if err != nil {
		return updateInput{}, err
	}
	return updateInput{
		IdempotencyKey:   idempotencyKey,
		InputType:        request.InputType,
		SourcePath:       request.SourcePath,
		SourceSHA256:     request.SourceSHA256,
		ArtifactFileName: request.ArtifactFileName,
		ArtifactIdentity: request.ArtifactIdentity,
		ReleasePath:      request.ReleasePath,
		Materialize:      materialize,
	}, nil
}

// restartMaterializer 为续跑的重启事务重建构件物化函数。若已物化的构件仍存在
// 且为直接普通文件，则返回一个表示“恢复期间构件消失”的失败函数；否则尝试从
// 源路径或 release 目录重新定位与身份匹配的 JAR 并提供其复制函数。
func (u *Updater) restartMaterializer(request persistedRequest) (func(string) error, error) {
	if info, err := os.Lstat(request.ArtifactPath); err == nil {
		if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 {
			return nil, fmt.Errorf("persisted backend restart artifact is not a direct regular file: %s", request.ArtifactPath)
		}
		return func(string) error {
			return errors.New("persisted backend restart artifact disappeared during resume")
		}, nil
	} else if !errors.Is(err, os.ErrNotExist) {
		return nil, fmt.Errorf("inspect persisted backend restart artifact: %w", err)
	}

	paths := []string{request.SourcePath, filepath.Join(u.config.Backend.ReleaseDir, request.ReleasePath)}
	for _, sourcePath := range paths {
		jar, err := updatepackage.OpenDirectNativeJAR(sourcePath)
		if err != nil {
			continue
		}
		if jar.SHA256 == request.SourceSHA256 && jar.Identity == request.ArtifactIdentity {
			return jar.CopyArtifact, nil
		}
	}
	return nil, fmt.Errorf("backend restart cannot recover artifact with identity %+v", request.ArtifactIdentity)
}