diff --git a/AGENTS_CHANGELOG.md b/AGENTS_CHANGELOG.md new file mode 100644 index 0000000..4336acb --- /dev/null +++ b/AGENTS_CHANGELOG.md @@ -0,0 +1,129 @@ +# AGENTS Changelog + +多 agent 并行时的共享变更记录。开工前先读本文，避免重复实现或与已变更的接口冲突。 +条目按时间倒序排列；破坏性变更（接口签名、协议字段等）会明确标注。 + +--- + +## 2026-08-22 对账结果可见性 + 容器镜像 tag 展示 + +### 诊断与对账输出 + +- `ymsctl status/doctor/reconcile` 的状态展示区分 `HEALTHY`、`REPAIRABLE`、`DRIFT`。 +- `ymsctl reconcile --service backend` 明确标记为只读修复计划，并提示追加 `--apply`。 +- `ymsctl reconcile --service backend --apply` 成功后输出 `APPLIED`、对账事务 ID 和重新核对结果。 +- `daemonapi.Diagnosis` 新增 `RepairApplied`、`RepairTransactionID` 字段；自动修复完成后由服务端填充。 +- 对账修复仍通过 SQLite 事务与 coordinator 执行，未改变只读命令的行为。 + +### 容器镜像显示与校验 + +- 容器更新创建容器时使用用户提供的 tag 引用，使 `docker ps` / `docker inspect .Config.Image` 显示可读 tag。 +- `backendexecutor.Request` 新增 `DisplayImageReference`，仅用于容器显示引用。 +- 实际镜像仍使用不可变 digest 引用拉取、校验和事务记录；容器创建阶段继续校验镜像 ID，提交阶段同时接受 digest 引用或对应 tag 引用。 +- 现有已创建容器不会被改写；下一次更新对应槽位后显示 tag。 + +### 验证与产物 + +- `go test ./...`、`go vet ./...`、`git diff --check` 通过。 +- 生成 x86_64 RPM：`dist/yms-daemon-0.1.0-20260822072334.f536987a.x86_64.rpm`。 +- RPM SHA256：`a3df708e54a95dadbf67487b4cd25dfa7599557237241638d00f38d390cf581e`。 + +--- + +## 2026-08-17 容器停止超时配置化 + 示例配置支持 native/container 快速切换 + +### 新增配置项 + +- `deploymentconfig.Backend` 新增 `stop_grace_seconds *int`（`[backend]` 下，可省略）。 + - 容器后端停止旧容器时，SIGTERM 后到强制终止前的等待秒数；省略时默认 30。 + - 配置值覆盖容器创建时的 9000 秒 StopTimeout；原生后端忽略该字段。 + - `validateExactDocumentKeys` 的 backend 允许键新增 `stop_grace_seconds`；`Validate` 要求其为正整数。 +- `backendupdate.Updater` 新增 `containerStopGraceSeconds`，由 `NewContainer` 从配置读取（缺省 30）。 +- `containerStopOperation` 新增 `stopGraceSeconds` 字段，`switchAndCommitContainer` 与 `rollbackContainer` 传 `u.containerStopGraceSeconds`。 + +### 示例配置 + +- 重写 `packaging/etc/yms-daemon/yms-daemon.toml`：同时展示 native（默认启用）与 container（整体注释）两套 `[backend]` 块，顶部注释说明快速切换步骤与约束（type 必须与现场一致、native↔container 需走迁移事务）。 + +### 测试 + +`deploymentconfig/config_test.go` 新增：`stop_grace_seconds` 合法值接受、非正整数拒绝。全量 build/vet/test 通过。 + +--- + +## 2026-08-17 修复旧容器停不掉与启动日志跟随 + +### 破坏性变更 + +`containerengine.Engine.StopContainer` 签名由 `StopContainer(context.Context, string) error` 改为 `StopContainer(context.Context, string, int) error`（新增超时秒数参数）。 + +### 修复旧容器「kill 不掉」 + +- 根因：`MobyEngine.StopContainer` 用 `ContainerStopOptions{}`（不传超时），Docker 沿用容器创建时设置的 `StopTimeout=9000` 秒，Spring Boot 优雅停机长期不退时旧容器无法停止。 +- `internal/containerengine/engine.go` / `moby.go`：`StopContainer` 增加 `timeoutSeconds`，`ContainerStop` 显式传 `Timeout`。 +- `internal/backendupdate/container.go`：新增常量 `containerStopGraceSeconds = 30`，`containerStopOperation.Apply` 传入 30 秒，超时后 Docker 强制 SIGKILL。 + +### 修复启动日志跟随 + +- 根因：`ContainerLogs` 用 `Follow:false` 一次性读快照（读于容器刚启动时，Spring Boot 尚未输出），且 `RestartContainer` 的 `StartLog` 为 false。 +- `internal/containerengine/moby.go`：`ContainerLogs` 改为 `Follow:true` + `io.Pipe` 持续流式返回。 +- `internal/backendexecutor/executor.go`：`startAndCheck` 改为后台 goroutine 跟随读日志（`streamContainerLogs`），健康检查结束后取消跟随。 +- `internal/backendupdate/container.go`：`RestartContainer` 的 `StartLog` 改为 true。 + +### 测试 + +三个测试 fake 的 `StopContainer` 签名同步更新。全量 build/vet/test 通过。 + +--- + +## 2026-08-17 实现 container 后端 restart + +`ymsctl restart --service backend` 的 container 后端此前直接返回「restart is not implemented for container backend」，本次补齐。 + +### 变更文件 + +- `internal/backendexecutor/executor.go`：新增 `ImageAcquisitionPresent = "present"` 模式——镜像已在本地，跳过 pull/load，仅在 `prepare` 核对本地镜像 digest + platform。 +- `internal/backendupdate/container.go`：新增 `RestartContainer`；新增 `inputTypeContainerRestart` 常量；`runContainerUpdate` 按 `InputType` 区分 `present`/`pull`；提取 `validateContainerInputs` 复用前置校验。 +- `internal/backendupdate/restart.go`：`Restart` 按 `backend.type` 分派，container → `RestartContainer`，native 走原逻辑。 + +### 语义（对齐 native restart 决策 50） + +- 复用本地镜像，不重新 pull，不依赖 registry。 +- 每次调用执行一次真实槽位轮转，幂等键用随机 `restart:` 前缀，不复用历史 update 的内容幂等键。 +- 无部署记录、gateway 活动端口漂移、活动容器身份不一致时拒绝。 +- 未完成 restart 事务由同一命令恢复；复用 `runContainerUpdate` + `switchAndCommitContainer`。 + +### 测试 + +`container_test.go` 新增：正常 restart（轮转 + 不 pull + 提交部署记录）、无部署记录拒绝、gateway 漂移拒绝。全量 build/vet/test 通过。 + +--- + +## 2026-08-17 实现 status / doctor / reconcile + +依据 `CLI_NAMING_AND_OPERATIONS_PLAN.md`（已并入 `PLAN.md` 附录）第 3.3 / 4 / 5 节，实现三个诊断与对账子命令。 + +### 破坏性变更 + +1. `daemonserver.New` 签名：由 `New(socketPath, updater, logger)` 改为 `New(socketPath, updater, diagnoser, logger)`。 +2. `daemonapi.Request` 新增 `Apply bool`（`json:"apply"`），仅 `OperationReconcile` 使用。 +3. `daemonapi.Response` 新增 `Diagnosis *Diagnosis`（`json:"diagnosis,omitempty"`）。 +4. `daemonapi` 新增操作常量 `OperationStatus/OperationDoctor/OperationReconcile`，及诊断级别常量 `DiagnosisLevelOK/Fixable/Drift`。 +5. `hostnginx.NewController` 已是 2 参数（`configPath, nginxExecutable`），重载走 `nginx -s reload`（并行期间他人改动，本轮按现状接入）。 + +### 新增 / 修改文件 + +- 新增 `internal/backendstatus/`：`diagnoser.go`、`container.go`、`native.go`、`reconcile.go`、`container_test.go`。 +- 修改 `daemonapi/protocol.go`、`daemonserver/server.go`、`daemonclient/client.go`、`main.go`、`main_test.go`、`daemonserver/server_test.go`。 + +### 诊断模型 + +- 级别：`ok`（一致）/ `fixable`（可自动修复）/ `drift`（需人工）。 +- `Healthy = true` 当且仅当无 `drift` 项。 +- 展示过滤：`status` 只展示 drift/fixable，`doctor` 展示全部，`reconcile` 只展示 fixable。 +- container 诊断以 `backend_container_deployment` 记录为事实来源；native 只读诊断（无自动修复）。 + +### 待办 + +- native 自动修复未实现（JAR 链接漂移修复需 filestore identity + release 校验）。 +- `coverage.out` 未重新生成（不含新包）。 diff --git a/internal/backendexecutor/executor.go b/internal/backendexecutor/executor.go index ec6b8a6..fae3459 100644 --- a/internal/backendexecutor/executor.go +++ b/internal/backendexecutor/executor.go @@ -56,6 +56,8 @@ const ( ImageAcquisitionLoad = "load" // ImageAcquisitionPull 表示通过远程仓库拉取方式获取后端镜像。 ImageAcquisitionPull = "pull" + // ImageAcquisitionPresent 表示镜像已在本地，不执行拉取或加载，只核对镜像身份。 + ImageAcquisitionPresent = "present" ) // Request 携带 update 包与本地部署配置提供的精确取值。 @@ -67,6 +69,9 @@ type Request struct { ArchivePath string // ImageReference 后端镜像的精确引用，执行器不解析其语义。 ImageReference string + // DisplayImageReference 容器配置中用于可读展示的镜像引用。非空时仅影响 docker ps 的显示， + // 实际镜像身份仍必须由 ExpectedImageDigest 校验通过。 + DisplayImageReference string // ExpectedImageDigest 后端镜像期望的清单摘要。 ExpectedImageDigest string // Platform 后端镜像显式指定的操作系统与架构。 @@ -239,12 +244,23 @@ func (e *Executor) prepare(ctx context.Context, transactionID string, request Re if _, err := e.coordinator.ExecuteStep(ctx, transactionID, pullIntent(request), operation); err != nil { return err } + case ImageAcquisitionPresent: + // 镜像已在本地，无外部副作用，跳过拉取，后续 InspectImage 时核对摘要。 } image, err := e.engine.InspectImage(ctx, request.ImageReference) if err != nil { return fmt.Errorf("inspect prepared backend image: %w", err) } + if request.ImageAcquisition == ImageAcquisitionPresent { + matches, err := imageMatches(image, request.ExpectedImageDigest, request.Platform) + if err != nil { + return err + } + if !matches { + return fmt.Errorf("present backend image %s does not match expected digest %s", request.ImageReference, request.ExpectedImageDigest) + } + } removeOperation := &removeContainerOperation{engine: e.engine, name: request.ContainerName} if _, err := e.coordinator.ExecuteStep(ctx, transactionID, removeIntent(request), removeOperation); err != nil { return err @@ -258,27 +274,18 @@ func (e *Executor) prepare(ctx context.Context, transactionID string, request Re return err } -// startAndCheck 先启动非活动后端容器，若请求要求读取启动日志则逐行回传， -// 最后执行健康检查步骤等待后端 Actuator 健康。 +// startAndCheck 先启动非活动后端容器，若请求要求读取启动日志则在后台跟随回传， +// 随后执行健康检查步骤等待后端 Actuator 健康，健康检查结束后停止日志跟随。 func (e *Executor) startAndCheck(ctx context.Context, transactionID string, request Request) error { startOperation := &startContainerOperation{engine: e.engine, name: request.ContainerName} if _, err := e.coordinator.ExecuteStep(ctx, transactionID, startIntent(request), startOperation); err != nil { return err } + var cancelLog context.CancelFunc if request.StartLog && request.LogReporter != nil { - logs, err := e.engine.ContainerLogs(ctx, request.ContainerName) - if err == nil { - scanner := bufio.NewScanner(logs) - for scanner.Scan() { - request.LogReporter(scanner.Text()) - } - _ = logs.Close() - if err := scanner.Err(); err != nil { - return fmt.Errorf("read container startup logs: %w", err) - } - } else { - request.LogReporter("unable to read container startup logs: " + err.Error()) - } + var logCtx context.Context + logCtx, cancelLog = context.WithCancel(ctx) + go e.streamContainerLogs(logCtx, request.ContainerName, request.LogReporter) } healthOperation := &healthOperation{ engine: e.engine, @@ -288,9 +295,27 @@ func (e *Executor) startAndCheck(ctx context.Context, transactionID string, requ timeout: healthTimeout, } _, err := e.coordinator.ExecuteStep(ctx, transactionID, healthIntent(request), healthOperation) + if cancelLog != nil { + cancelLog() + } return err } +// streamContainerLogs 跟随读取指定容器的日志并逐行上报，直到 ctx 被取消。 +// 它由 startAndCheck 在后台 goroutine 中调用，健康检查期间持续吐出容器启动日志。 +func (e *Executor) streamContainerLogs(ctx context.Context, name string, report func(string)) { + logs, err := e.engine.ContainerLogs(ctx, name) + if err != nil { + report("unable to read container startup logs: " + err.Error()) + return + } + defer logs.Close() + scanner := bufio.NewScanner(logs) + for scanner.Scan() { + report(scanner.Text()) + } +} + // validateRequest 对请求字段做静态校验，确保所有取值精确且自洽。 // 任一字段不符合要求时返回描述性错误。 func validateRequest(request Request) error { @@ -303,6 +328,10 @@ func validateRequest(request Request) error { if request.ArchivePath != "" { return errors.New("pull image acquisition does not accept an archive path") } + case ImageAcquisitionPresent: + if request.ArchivePath != "" { + return errors.New("present image acquisition does not accept an archive path") + } default: return fmt.Errorf("unsupported image acquisition: %q", request.ImageAcquisition) } @@ -394,9 +423,13 @@ func directDirectory(path, description string) error { // containerSpec 根据请求构造后端容器的完整规格，包括名称、镜像引用、平台、环境变量、 // 宿主机网络模式、重启策略、绑定挂载、用户与停止超时。 func containerSpec(request Request) containerengine.ContainerSpec { + imageReference := request.ImageReference + if request.DisplayImageReference != "" { + imageReference = request.DisplayImageReference + } return containerengine.ContainerSpec{ Name: request.ContainerName, - ImageReference: request.ImageReference, + ImageReference: imageReference, Platform: request.Platform, Environment: []string{ request.PortEnvironmentKey + "=" + strconv.Itoa(request.Port), diff --git a/internal/backendupdate/container.go b/internal/backendupdate/container.go index 1a3cf4f..45a9fe1 100644 --- a/internal/backendupdate/container.go +++ b/internal/backendupdate/container.go @@ -23,6 +23,15 @@ import ( // inputTypeContainerImage 标识以容器镜像方式执行后端更新的输入类型。 const inputTypeContainerImage = "container-image" +// inputTypeContainerRestart 标识容器后端同版本重启的输入类型，用于事务恢复时区分 +// 普通更新与重启，从而在重启时复用本地镜像而不重新拉取。 +const inputTypeContainerRestart = "container-restart" + +// defaultContainerStopGraceSeconds 停止旧容器时，发送停止信号后到强制终止前的默认等待秒数， +// 当 backend.stop_grace_seconds 未配置时使用。该值覆盖容器创建时设置的 9000 秒 StopTimeout， +// 避免旧容器因优雅停机超时长期无法退出。 +const defaultContainerStopGraceSeconds = 30 + // persistedContainerRequest 容器后端更新请求的持久化形态，以 JSON 存入事务 // 记录。事务在提交、重试或补偿时依赖这些字段重建执行上下文。 type persistedContainerRequest struct { @@ -69,19 +78,8 @@ func (u *Updater) UpdateContainerImage(ctx context.Context, imageReference strin if strings.TrimSpace(imageReference) != imageReference || imageReference == "" { return transaction.Transaction{}, errors.New("exact container image reference is required") } - configInfo, err := os.Lstat(u.containerConfigSource) - if err != nil { - return transaction.Transaction{}, fmt.Errorf("inspect backend configuration %s: %w", u.containerConfigSource, err) - } - if !configInfo.Mode().IsRegular() || configInfo.Mode()&os.ModeSymlink != 0 { - return transaction.Transaction{}, fmt.Errorf("backend configuration is not a direct regular file: %s", u.containerConfigSource) - } - tmpInfo, err := os.Lstat(u.containerTmpSource) - if err != nil { - return transaction.Transaction{}, fmt.Errorf("inspect backend temporary directory %s: %w", u.containerTmpSource, err) - } - if !tmpInfo.IsDir() || tmpInfo.Mode()&os.ModeSymlink != 0 { - return transaction.Transaction{}, fmt.Errorf("backend temporary path is not a direct directory: %s", u.containerTmpSource) + if err := u.validateContainerInputs(); err != nil { + return transaction.Transaction{}, err } active, err := u.store.ActiveTransaction(ctx) @@ -183,6 +181,142 @@ func (u *Updater) UpdateContainerImage(ctx context.Context, imageReference strin return u.runContainerUpdate(ctx, record, request, created, report) } +// RestartContainer 使用当前已提交部署记录的镜像执行一次同版本零停机轮转：它复用本地 +// 镜像（不重新拉取），把同一镜像部署到非活动槽位，健康检查通过后切换流量并停止旧容器。 +// +// 参数 ctx 用于控制整个重启过程的取消；report 用于回传实时进度，可为 nil。返回值为本次 +// 重启对应的事务记录以及错误。若不存在已提交部署记录、现场与部署记录不一致，或存在其他 +// 未完成事务，则拒绝新建重启；存在未完成的重启事务时复用续跑。 +func (u *Updater) RestartContainer(ctx context.Context, report ProgressReporter) (transaction.Transaction, error) { + if u.containerExecutor == nil || u.engine == nil { + return transaction.Transaction{}, errors.New("container backend updater is not configured") + } + if err := u.validateContainerInputs(); err != nil { + return transaction.Transaction{}, err + } + + deployment, err := u.store.BackendContainerDeployment(ctx) + if err != nil { + return transaction.Transaction{}, fmt.Errorf("container backend restart requires a committed deployment: %w", err) + } + + active, err := u.store.ActiveTransaction(ctx) + if err == nil { + var request persistedContainerRequest + if decodeErr := decodeContainerRequest(active.Request, &request); decodeErr != nil { + return active, decodeErr + } + if active.Service != serviceBackend || request.InputType != inputTypeContainerRestart { + return active, &transaction.ActiveTransactionError{TransactionID: active.ID} + } + return u.runContainerUpdate(ctx, active, request, false, report) + } + if !errors.Is(err, transaction.ErrNotFound) { + return transaction.Transaction{}, err + } + + before, err := u.gateway.Read() + if err != nil { + return transaction.Transaction{}, err + } + if before.ActivePort != deployment.ActivePort { + return transaction.Transaction{}, fmt.Errorf("container backend restart requires gateway active port %d, got %d", deployment.ActivePort, before.ActivePort) + } + + activeContainer, err := u.engine.InspectContainer(ctx, deployment.ContainerName) + if err != nil { + return transaction.Transaction{}, fmt.Errorf("inspect active backend container %s: %w", deployment.ContainerName, err) + } + if activeContainer.ID != deployment.ContainerID { + return transaction.Transaction{}, fmt.Errorf("active backend container %s identity does not match committed deployment", deployment.ContainerName) + } + if !activeContainer.Running || activeContainer.Dead { + return transaction.Transaction{}, fmt.Errorf("active backend container %s is not running", deployment.ContainerName) + } + + image, err := u.engine.InspectImage(ctx, activeContainer.ImageReference) + if err != nil { + return transaction.Transaction{}, fmt.Errorf("inspect active backend image %s: %w", activeContainer.ImageReference, err) + } + if image.Platform.OS == "" || image.Platform.Architecture == "" { + return transaction.Transaction{}, errors.New("active backend image does not report an exact platform") + } + + targetPort := otherPort(deployment.ActivePort) + targetSlot, err := u.config.Backend.SlotForPort(targetPort) + if err != nil { + return transaction.Transaction{}, err + } + previousSlot, err := u.config.Backend.SlotForPort(deployment.ActivePort) + if err != nil { + return transaction.Transaction{}, err + } + afterContent, err := hostnginx.RenderBackendPort(before.Content, targetPort) + if err != nil { + return transaction.Transaction{}, err + } + transactionID := rand.Text() + transactionRoot := filepath.Join(u.workRoot, transactionID) + if err := os.MkdirAll(transactionRoot, 0o750); err != nil { + return transaction.Transaction{}, fmt.Errorf("create container backend transaction directory: %w", err) + } + beforePath := filepath.Join(transactionRoot, "gateway.before.conf") + afterPath := filepath.Join(transactionRoot, "gateway.after.conf") + if err := writeImmutableFile(beforePath, before.Content, 0o640); err != nil { + return transaction.Transaction{}, err + } + if err := writeImmutableFile(afterPath, afterContent, 0o640); err != nil { + return transaction.Transaction{}, err + } + request := persistedContainerRequest{ + InputType: inputTypeContainerRestart, ImageReference: activeContainer.ImageReference, + ImmutableReference: activeContainer.ImageReference, ImageDigest: deployment.ImageDigest, Platform: image.Platform, + TargetPort: targetPort, TargetContainer: targetSlot.ContainerName, + PreviousPort: deployment.ActivePort, PreviousContainer: previousSlot.ContainerName, + TargetHealthEndpoint: targetSlot.HealthEndpoint, + StartLog: true, + GatewayBeforePath: beforePath, GatewayAfterPath: afterPath, + GatewayReceiptPath: filepath.Join(transactionRoot, "gateway.applied"), + } + content, err := json.Marshal(request) + if err != nil { + return transaction.Transaction{}, fmt.Errorf("encode container backend restart request: %w", err) + } + record, created, err := u.store.CreateTransaction(ctx, transaction.CreateRequest{ + ID: transactionID, IdempotencyKey: serviceBackend + ":restart:" + rand.Text(), + Source: sourceLocalCLI, Service: serviceBackend, Request: content, + }) + if err != nil { + return transaction.Transaction{}, err + } + if !created { + if err := decodeContainerRequest(record.Request, &request); err != nil { + return record, err + } + } + return u.runContainerUpdate(ctx, record, request, created, report) +} + +// validateContainerInputs 校验容器后端的配置源文件与临时目录满足部署前置条件： +// 配置源必须是直接普通文件，临时目录必须是直接目录。 +func (u *Updater) validateContainerInputs() error { + configInfo, err := os.Lstat(u.containerConfigSource) + if err != nil { + return fmt.Errorf("inspect backend configuration %s: %w", u.containerConfigSource, err) + } + if !configInfo.Mode().IsRegular() || configInfo.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("backend configuration is not a direct regular file: %s", u.containerConfigSource) + } + tmpInfo, err := os.Lstat(u.containerTmpSource) + if err != nil { + return fmt.Errorf("inspect backend temporary directory %s: %w", u.containerTmpSource, err) + } + if !tmpInfo.IsDir() || tmpInfo.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("backend temporary path is not a direct directory: %s", u.containerTmpSource) + } + return nil +} + // resolveContainerSlots 在选择目标槽位前，核对已提交的部署、网关以及两个精确 // 容器名是否一致。全新安装没有部署记录、没有已提交的容器事务历史，也没有槽位 // 容器，会从非路由槽位启动，这样只有健康检查通过后才会暴露流量。 @@ -318,9 +452,14 @@ func (u *Updater) runContainerUpdate(ctx context.Context, record transaction.Tra if record.State.Terminal() { return terminalResult(record) } + imageAcquisition := backendexecutor.ImageAcquisitionPull + if request.InputType == inputTypeContainerRestart { + imageAcquisition = backendexecutor.ImageAcquisitionPresent + } executorRequest := backendexecutor.Request{ - ImageAcquisition: backendexecutor.ImageAcquisitionPull, - ImageReference: request.ImmutableReference, ExpectedImageDigest: request.ImageDigest, Platform: request.Platform, + ImageAcquisition: imageAcquisition, + ImageReference: request.ImmutableReference, DisplayImageReference: request.ImageReference, + ExpectedImageDigest: request.ImageDigest, Platform: request.Platform, ContainerName: request.TargetContainer, Port: request.TargetPort, PortEnvironmentKey: deploymentconfig.ContainerPortEnvironment, ConfigSource: u.containerConfigSource, ConfigTarget: u.containerConfigTarget, @@ -400,7 +539,7 @@ func (u *Updater) switchAndCommitContainer(ctx context.Context, transactionID st if err := waitContext(ctx, u.drain); err != nil { return err } - operation := &containerStopOperation{engine: u.engine, name: request.PreviousContainer} + operation := &containerStopOperation{engine: u.engine, name: request.PreviousContainer, stopGraceSeconds: u.containerStopGraceSeconds} if _, err := u.coordinator.ExecuteStep(ctx, transactionID, stopPreviousContainerIntent(request), operation); err != nil { return err } @@ -412,7 +551,7 @@ func (u *Updater) switchAndCommitContainer(ctx context.Context, transactionID st if !target.Running || target.Dead || target.ID == "" { return fmt.Errorf("target backend container %s is not running with an exact identity", request.TargetContainer) } - if target.ImageReference != request.ImmutableReference { + if target.ImageReference != request.ImmutableReference && target.ImageReference != request.ImageReference { return fmt.Errorf("target backend container %s image reference does not match the transaction", request.TargetContainer) } _, err = u.store.CommitBackendContainerDeployment(ctx, transactionID, transaction.BackendContainerDeployment{ @@ -450,7 +589,7 @@ func (u *Updater) rollbackContainer(ctx context.Context, transactionID string, r } gateway := &gatewayOperation{controller: u.gateway, before: after, after: before, receiptPath: request.GatewayReceiptPath + ".restore"} _, gatewayErr := u.coordinator.ExecuteStep(ctx, transactionID, containerGatewayRestoreIntent(request), gateway) - stop := &containerStopOperation{engine: u.engine, name: request.TargetContainer} + stop := &containerStopOperation{engine: u.engine, name: request.TargetContainer, stopGraceSeconds: u.containerStopGraceSeconds} _, stopErr := u.coordinator.ExecuteStep(ctx, transactionID, stopTargetContainerIntent(request), stop) if err := errors.Join(gatewayErr, stopErr); err != nil { return errors.Join(cause, err) @@ -465,11 +604,13 @@ type containerStopOperation struct { engine containerengine.Engine // name 待停止的容器名。 name string + // stopGraceSeconds 发送停止信号后到强制终止前的等待秒数。 + stopGraceSeconds int } // Apply 停止指定容器，若容器不存在则视为已满足（幂等成功）。 func (o *containerStopOperation) Apply(ctx context.Context) error { - err := o.engine.StopContainer(ctx, o.name) + err := o.engine.StopContainer(ctx, o.name, o.stopGraceSeconds) if errors.Is(err, containerengine.ErrNotFound) { return nil } diff --git a/internal/backendupdate/container_test.go b/internal/backendupdate/container_test.go index ab026dc..24d398b 100644 --- a/internal/backendupdate/container_test.go +++ b/internal/backendupdate/container_test.go @@ -190,6 +190,94 @@ func TestContainerUpdaterRejectsMissingCommittedContainer(t *testing.T) { } } +// TestContainerUpdaterRestartsContainerWithLocalImage 验证 container 后端重启会复用当前已提交部署 +// 记录的镜像：轮转到非活跃槽位、健康检查通过后切流并停止旧容器，且整个过程不重新拉取镜像。 +func TestContainerUpdaterRestartsContainerWithLocalImage(t *testing.T) { + ctx := context.Background() + updater, store, engine, gateway := newContainerUpdaterFixture(t, map[string]containerengine.Container{ + "backend-8080": { + ID: "committed-backend-8080", Name: "backend-8080", Running: true, + ImageReference: "harbor.ymswell.asia/ymswell/glory-ymswell@" + containerTestDigest, + }, + }, 0) + commitContainerDeployment(t, store, transaction.BackendContainerDeployment{ + ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: containerTestDigest, ContainerID: "committed-backend-8080", + }) + + record, err := updater.RestartContainer(ctx, nil) + if err != nil { + t.Fatalf("restart container backend: %v", err) + } + if record.State != transaction.StateCommitted || gateway.snapshot.ActivePort != 8081 { + t.Fatalf("unexpected committed container restart: record=%+v gateway=%+v", record, gateway.snapshot) + } + if engine.containers["backend-8080"].Running { + t.Fatal("previous backend container is still running after restart") + } + if engine.pullCalls != 0 { + t.Fatalf("restart must reuse the local image without pulling: pulls=%d", engine.pullCalls) + } + assertCommittedContainerDeployment(t, store, record.ID, 8081, "backend-8081", "container-id-backend-8081") +} + +// TestContainerUpdaterRestartRejectsWithoutDeployment 验证不存在已提交部署记录时，container 后端重启被拒绝。 +func TestContainerUpdaterRestartRejectsWithoutDeployment(t *testing.T) { + updater, _, _, _ := newContainerUpdaterFixture(t, map[string]containerengine.Container{}, 0) + + _, err := updater.RestartContainer(context.Background(), nil) + if err == nil || !strings.Contains(err.Error(), "requires a committed deployment") { + t.Fatalf("unexpected restart without deployment: %v", err) + } +} + +// TestContainerUpdaterRestartRejectsGatewayDrift 验证宿主 Nginx 活动端口与部署记录不一致时，container 后端重启被拒绝。 +func TestContainerUpdaterRestartRejectsGatewayDrift(t *testing.T) { + updater, store, _, gateway := newContainerUpdaterFixture(t, map[string]containerengine.Container{ + "backend-8080": { + ID: "committed-backend-8080", Name: "backend-8080", Running: true, + ImageReference: "harbor.ymswell.asia/ymswell/glory-ymswell@" + containerTestDigest, + }, + }, 0) + commitContainerDeployment(t, store, transaction.BackendContainerDeployment{ + ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: containerTestDigest, ContainerID: "committed-backend-8080", + }) + gateway.snapshot.ActivePort = 8081 + + _, err := updater.RestartContainer(context.Background(), nil) + if err == nil || !strings.Contains(err.Error(), "requires gateway active port 8080, got 8081") { + t.Fatalf("unexpected restart with gateway drift: %v", err) + } +} + +// commitContainerDeployment 通过合法的状态机推进写入一条 container 部署记录，供重启测试预置现场。 +func commitContainerDeployment(t *testing.T, store *transaction.Store, deployment transaction.BackendContainerDeployment) { + t.Helper() + ctx := context.Background() + transactionID := "seed-" + deployment.ContainerName + _, _, err := store.CreateTransaction(ctx, transaction.CreateRequest{ + ID: transactionID, IdempotencyKey: "backend:container:seed:" + deployment.ContainerName, Source: sourceLocalCLI, + Service: serviceBackend, Request: []byte(`{"inputType":"container-image"}`), + }) + if err != nil { + t.Fatalf("create seed backend transaction: %v", err) + } + for _, state := range []transaction.State{ + transaction.StateValidating, + transaction.StatePrepared, + transaction.StateStarting, + transaction.StateSwitching, + transaction.StateVerifying, + transaction.StateDraining, + } { + if _, err := store.Transition(ctx, transactionID, state, "seed committed deployment"); err != nil { + t.Fatalf("transition seed backend transaction to %s: %v", state, err) + } + } + if _, err := store.CommitBackendContainerDeployment(ctx, transactionID, deployment, "seed committed backend container deployment"); err != nil { + t.Fatalf("commit seed backend deployment: %v", err) + } +} + // newContainerUpdaterFixture 构造容器更新器的测试夹具，返回更新器、事务存储、假容器引擎和内存网关。 // containers 指定引擎初始存在的容器；drain 指定旧容器停止前的排水等待时长。 func newContainerUpdaterFixture( @@ -263,7 +351,7 @@ func newContainerUpdaterFixture( // 环境变量和挂载点均符合容器更新契约。 func assertCreatedContainerSpec(t *testing.T, request containerengine.ContainerSpec) { t.Helper() - if request.Name != "backend-8081" || request.ImageReference != "harbor.ymswell.asia/ymswell/glory-ymswell@"+containerTestDigest { + if request.Name != "backend-8081" || request.ImageReference != containerTestImage { t.Fatalf("unexpected target container identity: %+v", request) } if request.NetworkMode != "host" || request.RestartPolicy.Name != "no" || request.User != "0:0" { @@ -325,13 +413,17 @@ type containerUpdateEngine struct { stopped []string lastCreateSpec containerengine.ContainerSpec healthChecks int + pullCalls int } // Ping 返回 nil，模拟引擎连通性检查始终成功。 func (e *containerUpdateEngine) Ping(context.Context) error { return nil } -// PullImage 返回 nil，模拟镜像拉取始终成功。 -func (e *containerUpdateEngine) PullImage(context.Context, string) error { return nil } +// PullImage 记录一次拉取调用并返回 nil，模拟镜像拉取始终成功。 +func (e *containerUpdateEngine) PullImage(context.Context, string) error { + e.pullCalls++ + return nil +} // LoadImage 返回 nil，模拟从流加载镜像始终成功。 func (e *containerUpdateEngine) LoadImage(context.Context, io.Reader) error { return nil } @@ -381,7 +473,7 @@ func (e *containerUpdateEngine) ContainerLogs(context.Context, string) (io.ReadC } // StopContainer 记录被停止的容器名称并将该容器标记为已退出；若容器不存在则返回 containerengine.ErrNotFound。 -func (e *containerUpdateEngine) StopContainer(_ context.Context, name string) error { +func (e *containerUpdateEngine) StopContainer(_ context.Context, name string, _ int) error { e.stopped = append(e.stopped, name) record, found := e.containers[name] if !found { diff --git a/internal/daemonapi/protocol.go b/internal/daemonapi/protocol.go index 8c057cd..105f7bc 100644 --- a/internal/daemonapi/protocol.go +++ b/internal/daemonapi/protocol.go @@ -8,6 +8,12 @@ const ( OperationUpdate = "update" // OperationRestart 表示重启操作，即不更换制品、仅重启当前后端服务。 OperationRestart = "restart" + // OperationStatus 表示状态查询操作，读取 SQLite 并校验外部系统与活动槽位是否一致。 + OperationStatus = "status" + // OperationDoctor 表示只读诊断操作，输出全部漂移项与建议动作。 + OperationDoctor = "doctor" + // OperationReconcile 表示对账操作，只读生成修复计划；携带 Apply 时执行自动修复。 + OperationReconcile = "reconcile" // InputTypeRepackZIP 表示输入类型为重新打包后的 ZIP 包。 InputTypeRepackZIP = "repack-zip" // InputTypeNativeJAR 表示输入类型为原生后端 JAR 文件。 @@ -20,6 +26,16 @@ const ( ResponseResult = "result" ) +// 诊断项级别的稳定取值，用于 status/doctor/reconcile 的结果展示。 +const ( + // DiagnosisLevelOK 表示该项与 SQLite 事实来源一致，无需处理。 + DiagnosisLevelOK = "ok" + // DiagnosisLevelFixable 表示该项不一致，但可由 reconcile --apply 自动修复。 + DiagnosisLevelFixable = "fixable" + // DiagnosisLevelDrift 表示该项不一致且无法自动修复，需要人工确认。 + DiagnosisLevelDrift = "drift" +) + // Request 表示客户端通过 Unix Socket 提交给守护进程的一次请求。 // 各字段按操作类型选择性填充，未使用的字段保持空值。 type Request struct { @@ -35,6 +51,8 @@ type Request struct { ImageReference string `json:"imageReference"` // StartLog 表示容器更新完成后是否输出容器启动日志，仅对容器镜像输入生效。 StartLog bool `json:"startLog"` + // Apply 仅用于 OperationReconcile：为 true 时执行自动修复动作，否则只生成修复计划。 + Apply bool `json:"apply"` } // Response 表示守护进程返回给客户端的一次响应。 @@ -50,4 +68,34 @@ type Response struct { Message string `json:"message,omitempty"` // Error 表示操作失败时的错误信息，为空说明操作执行成功。 Error string `json:"error,omitempty"` + // Diagnosis 表示 status/doctor/reconcile 的结构化诊断结果，仅在对应操作成功时返回。 + Diagnosis *Diagnosis `json:"diagnosis,omitempty"` +} + +// Diagnosis status/doctor/reconcile 的结构化诊断结果，由守护进程生成并通过结果响应返回。 +type Diagnosis struct { + // Service 诊断的目标服务名。 + Service string `json:"service"` + // Type 组件运行类型，取值 native 或 container。 + Type string `json:"type"` + // Healthy 表示是否存在需要人工处理的漂移项（drift）；只有 ok 与 fixable 项时为 true。 + Healthy bool `json:"healthy"` + // Items 按诊断顺序排列的诊断项列表。 + Items []DiagnosisItem `json:"items"` + // RepairApplied 表示本次请求是否实际执行了 reconcile --apply 修复动作。 + RepairApplied bool `json:"repairApplied,omitempty"` + // RepairTransactionID 表示本次对账修复事务标识，仅在 RepairApplied 为 true 时返回。 + RepairTransactionID string `json:"repairTransactionId,omitempty"` +} + +// DiagnosisItem 单条诊断结论，说明某项现场状态与 SQLite 事实来源是否一致。 +type DiagnosisItem struct { + // Level 诊断级别，取值 DiagnosisLevelOK / DiagnosisLevelFixable / DiagnosisLevelDrift。 + Level string `json:"level"` + // Code 稳定标识，供程序与测试识别，不参与展示。 + Code string `json:"code"` + // Message 对该诊断项的人类可读描述。 + Message string `json:"message"` + // Action 仅在 Level 为 fixable 时给出建议的修复动作描述。 + Action string `json:"action,omitempty"` } diff --git a/main.go b/main.go index b4f41b2..b29da28 100644 --- a/main.go +++ b/main.go @@ -13,9 +13,11 @@ import ( "os" "os/signal" "path/filepath" + "strings" "syscall" "time" + "yms-daemon/internal/backendstatus" "yms-daemon/internal/backendupdate" "yms-daemon/internal/containerengine" "yms-daemon/internal/daemonapi" @@ -125,6 +127,27 @@ func run(ctx context.Context, arguments []string, stdout io.Writer, stderr io.Wr fmt.Fprintf(stdout, "transaction=%s state=%s\n", response.TransactionID, response.State) } return 0 + case "status", "doctor", "reconcile": + request, err := parseDiagnosisArgs(arguments[0], arguments[1:], stderr) + if err != nil { + fmt.Fprintln(stderr, err) + return 2 + } + var diagnosis daemonapi.Diagnosis + switch arguments[0] { + case "status": + diagnosis, err = daemonclient.Status(ctx, runtimepaths.Socket, request.service) + case "doctor": + diagnosis, err = daemonclient.Doctor(ctx, runtimepaths.Socket, request.service) + case "reconcile": + diagnosis, err = daemonclient.Reconcile(ctx, runtimepaths.Socket, request.service, request.apply) + } + if err != nil { + fmt.Fprintf(stderr, "ymsctl %s failed: %v\n", arguments[0], err) + return 1 + } + writeDiagnosis(stdout, arguments[0], diagnosis) + return 0 case "help", "-h", "--help": writeUsage(stdout) return 0 @@ -235,6 +258,37 @@ func parseRestartArgs(arguments []string, output io.Writer) (restartArguments, e return restartArguments{service: *service, quite: *quite}, nil } +// diagnosisArguments 保存 status/doctor/reconcile 子命令解析后的参数。 +type diagnosisArguments struct { + // service 目标服务名，当前仅接受 backend。 + service string + // apply 为真时表示 reconcile 执行自动修复，其余命令不接受该选项。 + apply bool +} + +// parseDiagnosisArgs 解析 status/doctor/reconcile 子命令的 flag 参数并做业务校验。 +// 它要求 --service 必须为 backend，且不接受任何位置参数；--apply 仅允许 reconcile 使用。 +// 校验失败时向 output 写入原因并返回 error。 +func parseDiagnosisArgs(command string, arguments []string, output io.Writer) (diagnosisArguments, error) { + flags := flag.NewFlagSet(command, flag.ContinueOnError) + flags.SetOutput(output) + service := flags.String("service", "", "service to diagnose") + apply := flags.Bool("apply", false, "apply automatic fixes (reconcile only)") + if err := flags.Parse(arguments); err != nil { + return diagnosisArguments{}, err + } + if flags.NArg() != 0 { + return diagnosisArguments{}, errors.New(command + " does not accept positional arguments") + } + if *service != serviceBackend { + return diagnosisArguments{}, errors.New("--service currently accepts only backend") + } + if *apply && command != "reconcile" { + return diagnosisArguments{}, errors.New("--apply requires reconcile") + } + return diagnosisArguments{service: *service, apply: *apply}, nil +} + // runServe 以守护进程服务端角色运行：初始化日志、获取进程锁、加载部署配置、 // 打开事务存储与协调器、构建 Nginx 控制器与后端更新器，最后通过 daemonserver 开始服务。 // 它使用具名返回 result 以便在各资源清理阶段合并所有 Close 错误，最终返回服务端的运行错误。 @@ -273,6 +327,7 @@ func runServe(ctx context.Context) (result error) { } httpClient := &http.Client{Timeout: 5 * time.Second} var updater *backendupdate.Updater + var diagnoser *backendstatus.Diagnoser switch config.Backend.Type { case deploymentconfig.BackendTypeNative: releaseStore, err := filestore.New(config.Backend.ReleaseDir) @@ -284,12 +339,18 @@ func runServe(ctx context.Context) (result error) { return err } updater, err = backendupdate.New(config, runtimepaths.WorkRoot, store, coordinator, releaseStore, units, gateway, httpClient, logger) + if err == nil { + diagnoser, err = backendstatus.New(config, store, coordinator, nil, units, gateway) + } case deploymentconfig.BackendTypeContainer: var engine *containerengine.MobyEngine engine, err = containerengine.NewMobyEngine() if err == nil { defer func() { result = errors.Join(result, engine.Close()) }() updater, err = backendupdate.NewContainer(config, runtimepaths.WorkRoot, store, coordinator, engine, gateway, httpClient, logger) + if err == nil { + diagnoser, err = backendstatus.New(config, store, coordinator, engine, nil, gateway) + } } default: err = fmt.Errorf("unsupported backend.type %q", config.Backend.Type) @@ -297,7 +358,7 @@ func runServe(ctx context.Context) (result error) { if err != nil { return err } - server, err := daemonserver.New(runtimepaths.Socket, updater, logger) + server, err := daemonserver.New(runtimepaths.Socket, updater, diagnoser, logger) if err != nil { return err } @@ -312,6 +373,9 @@ func writeUsage(output io.Writer) { fmt.Fprintln(output, " ymsctl update --service backend --container-image [--no-start-log] [--quite]") fmt.Fprintln(output, " ymsctl restart --service backend [--quite]") fmt.Fprintln(output, " ymsctl list [--limit ] [--service backend] [--state ] [--json]") + fmt.Fprintln(output, " ymsctl status --service backend") + fmt.Fprintln(output, " ymsctl doctor --service backend") + fmt.Fprintln(output, " ymsctl reconcile --service backend [--apply]") } // writeUpdateProgress 把一条进度事件格式化写入 output：状态字段左对齐占 14 列， @@ -324,6 +388,51 @@ func writeUpdateProgress(output io.Writer, event daemonapi.Response) { fmt.Fprintf(output, "%-14s %s\n", state, event.Message) } +// writeDiagnosis 把诊断结果格式化写入 output。 +// status 只展示需要关注的漂移与可修复项，doctor 展示全部诊断项， +// reconcile 只展示可自动修复项作为修复计划。 +func writeDiagnosis(output io.Writer, command string, diagnosis daemonapi.Diagnosis) { + state := "HEALTHY" + if !diagnosis.Healthy { + state = "DRIFT" + } else { + for _, item := range diagnosis.Items { + if item.Level == daemonapi.DiagnosisLevelFixable { + state = "REPAIRABLE" + break + } + } + } + fmt.Fprintf(output, "service=%s type=%s status=%s\n", diagnosis.Service, diagnosis.Type, state) + if diagnosis.RepairApplied { + fmt.Fprintf(output, "APPLIED 对账修复已执行 transaction=%s\n", diagnosis.RepairTransactionID) + fmt.Fprintln(output, "RESULT 现场已重新核对") + } + shown := 0 + for _, item := range diagnosis.Items { + switch command { + case "status": + if item.Level == daemonapi.DiagnosisLevelOK { + continue + } + case "reconcile": + if item.Level != daemonapi.DiagnosisLevelFixable { + continue + } + } + shown++ + fmt.Fprintf(output, "%-8s %s\n", strings.ToUpper(item.Level), item.Message) + if item.Action != "" { + fmt.Fprintf(output, "%-8s %s\n", "action", item.Action) + } + } + if shown == 0 { + fmt.Fprintln(output, "(no items)") + } else if command == "reconcile" && !diagnosis.RepairApplied { + fmt.Fprintln(output, "PLAN 仅生成修复计划，未执行；如需执行请追加 --apply") + } +} + // runList 执行 list 子命令：解析过滤参数，打开事务存储读取最近事务， // 按是否指定 --json 决定以稳定 JSON 或制表符分隔的表格形式输出到 stdout。 // 它支持 --limit、--service、--state、--json 四个选项，返回查询或输出阶段的错误。 diff --git a/main_test.go b/main_test.go index d36cb4b..cf2c2f9 100644 --- a/main_test.go +++ b/main_test.go @@ -171,3 +171,88 @@ func TestRunPrintsUsageWithoutCommand(t *testing.T) { t.Fatal("missing usage output") } } + +// TestParseDiagnosisArgsAcceptsService 验证 status/doctor/reconcile 子命令能接受 backend 服务， +// 且 reconcile 能接受 --apply。 +func TestParseDiagnosisArgsAcceptsService(t *testing.T) { + for _, command := range []string{"status", "doctor", "reconcile"} { + request, err := parseDiagnosisArgs(command, []string{"--service", "backend"}, &bytes.Buffer{}) + if err != nil { + t.Fatalf("parse %s arguments: %v", command, err) + } + if request.service != serviceBackend { + t.Fatalf("unexpected %s service: %+v", command, request) + } + } + request, err := parseDiagnosisArgs("reconcile", []string{"--service", "backend", "--apply"}, &bytes.Buffer{}) + if err != nil { + t.Fatalf("parse reconcile apply arguments: %v", err) + } + if !request.apply { + t.Fatal("reconcile --apply was not retained") + } +} + +// TestParseDiagnosisArgsRejectsUnsupportedInput 验证诊断子命令拒绝缺失服务、 +// 非 backend 服务、位置参数以及非 reconcile 命令使用 --apply。 +func TestParseDiagnosisArgsRejectsUnsupportedInput(t *testing.T) { + for name, test := range map[string]struct { + command string + arguments []string + }{ + "missing service": {"status", nil}, + "other service": {"doctor", []string{"--service", "frontend"}}, + "positional": {"status", []string{"--service", "backend", "extra"}}, + "apply on status": {"status", []string{"--service", "backend", "--apply"}}, + "apply on doctor": {"doctor", []string{"--service", "backend", "--apply"}}, + } { + t.Run(name, func(t *testing.T) { + if _, err := parseDiagnosisArgs(test.command, test.arguments, &bytes.Buffer{}); err == nil { + t.Fatal("expected diagnosis argument rejection") + } + }) + } +} + +// TestWriteDiagnosisFiltersByCommand 验证 writeDiagnosis 按子命令过滤诊断项： +// status 只展示漂移与可修复项，doctor 展示全部，reconcile 只展示可修复项。 +func TestWriteDiagnosisFiltersByCommand(t *testing.T) { + diagnosis := daemonapi.Diagnosis{ + Service: "backend", + Type: "container", + Healthy: true, + Items: []daemonapi.DiagnosisItem{ + {Level: daemonapi.DiagnosisLevelOK, Code: "active", Message: "活动容器健康"}, + {Level: daemonapi.DiagnosisLevelFixable, Code: "nginx", Message: "Nginx 指向错误", Action: "切流到 8080"}, + }, + } + + var statusOut, doctorOut, reconcileOut bytes.Buffer + writeDiagnosis(&statusOut, "status", diagnosis) + writeDiagnosis(&doctorOut, "doctor", diagnosis) + writeDiagnosis(&reconcileOut, "reconcile", diagnosis) + + if !strings.Contains(statusOut.String(), "Nginx 指向错误") || strings.Contains(statusOut.String(), "活动容器健康") { + t.Fatalf("status should hide ok items: %q", statusOut.String()) + } + if !strings.Contains(doctorOut.String(), "活动容器健康") || !strings.Contains(doctorOut.String(), "Nginx 指向错误") { + t.Fatalf("doctor should show all items: %q", doctorOut.String()) + } + if !strings.Contains(reconcileOut.String(), "Nginx 指向错误") || strings.Contains(reconcileOut.String(), "活动容器健康") { + t.Fatalf("reconcile should only show fixable items: %q", reconcileOut.String()) + } + if !strings.Contains(statusOut.String(), "status=REPAIRABLE") { + t.Fatalf("status should distinguish fixable drift: %q", statusOut.String()) + } + if !strings.Contains(reconcileOut.String(), "未执行") || !strings.Contains(reconcileOut.String(), "--apply") { + t.Fatalf("reconcile plan should explain that no fix was applied: %q", reconcileOut.String()) + } + + var appliedOut bytes.Buffer + diagnosis.RepairApplied = true + diagnosis.RepairTransactionID = "TX-1" + writeDiagnosis(&appliedOut, "reconcile", diagnosis) + if !strings.Contains(appliedOut.String(), "APPLIED") || !strings.Contains(appliedOut.String(), "TX-1") { + t.Fatalf("reconcile apply should show transaction: %q", appliedOut.String()) + } +}