fix(library): 维护释放与服务恢复完成之后才暴露作业成功
build / build (push) Successful in 1m32s

原本 execute 写完生效版本就写 succeeded,而服务恢复、入口开放与维护锁释放发生在
窗口返回之后。任一步失败时外部已看到成功,重启接管又只认非终态,于是目标永远
留在维护里(本轮真实现象:同步 succeeded 后立即恢复被 409)。

现在内容校验与版本登记完成后只记"待收尾":finishLibrarySuccess 先确认目标写入者
与入口健康,再按 scope → instance 顺序释放两把锁(每一步错误都上抛,不再 `_ =` 丢弃),
最后才写 succeeded;任一步失败即落失败并保持 needs_restore。释放与终态写入用
context.WithoutCancel 有界包住,停机不会留下半成品。接管的 finalize-only 判定改为
按证据(finalize 阶段成功或未顶掉的生效版本),只复核与登记,不重写生产。

测试覆盖:调用顺序(成功后两把锁必已释放)、健康失败不释放不报成功、三步释放失败
各自不报成功、终态写入失败提示等待接管、接管证据判定正负例。
This commit is contained in:
姚章浩
2026-10-29 14:02:09 +08:00
parent c14dd01f08
commit 8d2ed8ebe6
4 changed files with 308 additions and 25 deletions
+8
View File
@@ -38,6 +38,14 @@
### Fixed
- **作业成功暴露早于维护结束**:原来 `execute` 里写完生效版本就 `AdvanceLibraryJob(succeeded)`,
而服务恢复、入口开放与维护锁释放要在窗口返回后才做。任何一步失败时外部已经看到
`succeeded`,重启接管又只认非终态,锁就永远挂着。现在内容校验/版本登记之后只记
"待收尾",`finishLibrarySuccess` 依次确认目标写入者与入口健康、按 **scope → instance**
顺序释放两把锁(每一步错误都上抛,不再丢弃),最后才写 `succeeded`;任一步失败都把
作业落成失败、维护保持 `needs_restore`。释放与终态写入用 `context.WithoutCancel`
包住,停机不会留下半成品。接管判定改为按证据(finalize 阶段成功或存在未顶掉的生效
版本)转 finalize-only,只复核与登记,不重写生产。
- **封存读资源失败只剩一句 `read library resource`**:对象缺失、桶错、端点解析不到
都会走到同一个字符串,真实封存直接卡在这里无法判断方向。现在带上资源行 id 与底层
原因(`read library resource <id>: <cause>`),不含凭据。
+124
View File
@@ -0,0 +1,124 @@
package main
import (
"context"
"errors"
"fmt"
"time"
"knowledge-governance/internal/governance"
postgresstore "knowledge-governance/internal/platform/postgres"
)
// libraryMaintenanceStore 是"成功收尾"用到的存储能力。抽成接口是为了能在测试里
// 让某一步失败,验证"没全部完成就不许暴露成功"。
type libraryMaintenanceStore interface {
SetInstanceMaintenanceIngress(ctx context.Context, instance, ownerJobID, state, note string, now time.Time) error
SetLibraryMaintenanceIngress(ctx context.Context, scope, ownerJobID, state, note string, now time.Time) error
UpdateInstanceMaintenanceState(ctx context.Context, instance, ownerJobID, state, note string, now time.Time) error
UpdateLibraryMaintenanceState(ctx context.Context, scope, ownerJobID, state, note string, now time.Time) error
ReleaseInstanceMaintenance(ctx context.Context, instance, ownerJobID string, now time.Time) error
ReleaseLibraryMaintenance(ctx context.Context, scope, ownerJobID string, now time.Time) error
AdvanceLibraryJob(ctx context.Context, jobID string, state governance.LibraryJobState, stage, message, recovery, snapshotID, targetSnapshotID string, now time.Time) error
}
// releaseWindowLocks 是唯一释放路径。
//
// 顺序刻意是"scope 先、instance 后":instance 行是跨 scope 的互斥来源,最后放它
// 才能保证任何新作业在全部记录落定之前都进不来。每一步的错误都必须向上暴露——
// 只丢日志会让"锁其实没放"这件事变成页面上的成功。
func releaseWindowLocks(ctx context.Context, store libraryMaintenanceStore, instance, scope, jobID, note string, now time.Time) error {
if err := store.SetLibraryMaintenanceIngress(ctx, scope, jobID, "open", note, now); err != nil {
return fmt.Errorf("记录入口已开放失败(scope):%w", err)
}
if err := store.SetInstanceMaintenanceIngress(ctx, instance, jobID, "open", note, now); err != nil {
return fmt.Errorf("记录入口已开放失败(instance):%w", err)
}
if err := store.UpdateLibraryMaintenanceState(ctx, scope, jobID, "verified", note, now); err != nil {
return fmt.Errorf("维护状态写入失败(scope):%w", err)
}
if err := store.ReleaseLibraryMaintenance(ctx, scope, jobID, now); err != nil {
return fmt.Errorf("维护占用释放失败(scope):%w", err)
}
if err := store.UpdateInstanceMaintenanceState(ctx, instance, jobID, "verified", note, now); err != nil {
return fmt.Errorf("维护状态写入失败(instance):%w", err)
}
if err := store.ReleaseInstanceMaintenance(ctx, instance, jobID, now); err != nil {
return fmt.Errorf("实例占用释放失败(instance):%w", err)
}
return nil
}
// healthChecker 是"目标写入者与入口是否真的恢复健康"的判定口(测试可注入失败)。
type healthChecker interface {
VerifyRunning(ctx context.Context) error
}
// successBoundary 描述一次"内容已通过验证与版本登记,但还没到可以对外报成功"的作业。
type successBoundary struct {
Ready bool
Backup string
Target string
FinishedAt time.Time
}
// finishLibrarySuccess 是唯一的成功收尾入口:只有以下全部成立才暴露 succeeded——
// 目标服务健康、入口开放、scope 与 instance 两把维护锁都释放完成、终态写入成功。
//
// 为什么不能先标成功再释放:恢复/开入口/放锁都可能失败,先写 succeeded 会让外部
// (页面、API、重启后的接管逻辑)看到一条"成功了但目标仍在维护"的作业;而
// InterruptedLibraryJobs 只接管非终态,终态作业不会再被接管,锁就永远挂着。
func finishLibrarySuccess(ctx context.Context, store libraryMaintenanceStore, health healthChecker,
instance, scope, jobID string, boundary successBoundary) error {
// 释放与终态写入必须能在 ctx 取消后继续:停机不该留下"锁放了但作业还在跑"或
// 反过来"作业成功了锁还挂着"的任一半成品。
bounded, cancel := context.WithTimeout(context.WithoutCancel(ctx), 60*time.Second)
defer cancel()
holdForRepair := func(reason string) error {
note := "内容已验证,但收尾未完成,目标保持维护:" + reason
_ = store.UpdateLibraryMaintenanceState(bounded, scope, jobID, "needs_restore", note, time.Now().UTC())
_ = store.UpdateInstanceMaintenanceState(bounded, instance, jobID, "needs_restore", note, time.Now().UTC())
_ = store.AdvanceLibraryJob(bounded, jobID, governance.LibraryJobFailed, "finalize", note,
"生产内容已写入并校验;维护保持,按预备份或人工核验后再开放", boundary.Backup, boundary.Target, time.Now().UTC())
return errors.New(note)
}
if err := health.VerifyRunning(bounded); err != nil {
return holdForRepair("目标服务或入口未恢复健康:" + err.Error())
}
if err := releaseWindowLocks(bounded, store, instance, scope, jobID,
"写入与校验完成,目标已按原运行状态恢复", time.Now().UTC()); err != nil {
return holdForRepair(err.Error())
}
finishedAt := time.Now().UTC()
if err := store.AdvanceLibraryJob(bounded, jobID, governance.LibraryJobSucceeded, "finalize", "", "",
boundary.Backup, boundary.Target, finishedAt); err != nil {
// 锁已释放但终态没写下去:作业保持非终态,重启后会被接管(接管走
// finalize-only,只复核与登记,不重写生产)。
return fmt.Errorf("维护已释放但作业终态写入失败(等待接管):%w", err)
}
return nil
}
// needsFinalizeOnly 判断一个被接管的作业是否"内容与版本都已落定、只差收尾"。
// 判据只认真实证据:终态版本记录存在且未被顶掉,或 finalize 阶段已成功。
func needsFinalizeOnly(job governance.LibraryJob, versions []governance.LibraryVersion) bool {
if job.TargetSnapshotID == "" {
return false
}
for _, stage := range job.Stages {
if stage.Stage == "finalize" && stage.State == "succeeded" {
return true
}
}
for _, version := range versions {
if version.Kind == governance.LibraryVersionProductionEffective && version.SupersededAt == nil &&
version.SnapshotID == job.TargetSnapshotID {
return true
}
}
return false
}
var _ libraryMaintenanceStore = (*postgresstore.Store)(nil)
+149
View File
@@ -0,0 +1,149 @@
package main
import (
"context"
"errors"
"strings"
"testing"
"time"
"knowledge-governance/internal/governance"
)
type fakeFinishStore struct {
calls []string
failOn string
advanced []governance.LibraryJobState
}
func (f *fakeFinishStore) record(name string) error {
f.calls = append(f.calls, name)
if f.failOn == name {
return errors.New("injected " + name + " failure")
}
return nil
}
func (f *fakeFinishStore) SetInstanceMaintenanceIngress(context.Context, string, string, string, string, time.Time) error {
return f.record("instance_ingress")
}
func (f *fakeFinishStore) SetLibraryMaintenanceIngress(context.Context, string, string, string, string, time.Time) error {
return f.record("scope_ingress")
}
func (f *fakeFinishStore) UpdateInstanceMaintenanceState(context.Context, string, string, string, string, time.Time) error {
return f.record("instance_state")
}
func (f *fakeFinishStore) UpdateLibraryMaintenanceState(context.Context, string, string, string, string, time.Time) error {
return f.record("scope_state")
}
func (f *fakeFinishStore) ReleaseInstanceMaintenance(context.Context, string, string, time.Time) error {
return f.record("instance_release")
}
func (f *fakeFinishStore) ReleaseLibraryMaintenance(context.Context, string, string, time.Time) error {
return f.record("scope_release")
}
func (f *fakeFinishStore) AdvanceLibraryJob(_ context.Context, _ string, state governance.LibraryJobState,
_, _, _, _, _ string, _ time.Time) error {
f.advanced = append(f.advanced, state)
return f.record("advance_" + string(state))
}
type fakeHealth struct{ err error }
func (h fakeHealth) VerifyRunning(context.Context) error { return h.err }
func boundary() successBoundary {
return successBoundary{Ready: true, Backup: "backup-1", Target: "snap-1", FinishedAt: time.Now().UTC()}
}
// 首次观察到成功时,两把维护锁必须已经释放,且终态写入排在最后。
func TestFinishLibrarySuccessReleasesEveryLockBeforeSucceeded(t *testing.T) {
store := &fakeFinishStore{}
if err := finishLibrarySuccess(context.Background(), store, fakeHealth{}, "kgseal-weknora", "internal", "job-1", boundary()); err != nil {
t.Fatal(err)
}
want := []string{"scope_ingress", "instance_ingress", "scope_state", "scope_release",
"instance_state", "instance_release", "advance_succeeded"}
if strings.Join(store.calls, ",") != strings.Join(want, ",") {
t.Fatalf("call order = %v", store.calls)
}
if len(store.advanced) != 1 || store.advanced[0] != governance.LibraryJobSucceeded {
t.Fatalf("advanced = %v", store.advanced)
}
}
// 目标服务/入口没恢复健康:不许放锁,不许报成功,作业落成失败并保持维护。
func TestFinishLibrarySuccessKeepsMaintenanceWhenTargetIsNotHealthy(t *testing.T) {
store := &fakeFinishStore{}
err := finishLibrarySuccess(context.Background(), store, fakeHealth{err: errors.New("app not healthy")},
"kgseal-weknora", "internal", "job-2", boundary())
if err == nil || !strings.Contains(err.Error(), "未恢复健康") {
t.Fatalf("err = %v", err)
}
for _, call := range store.calls {
if strings.HasSuffix(call, "_release") || call == "advance_succeeded" {
t.Fatalf("released or reported success despite unhealthy target: %v", store.calls)
}
}
if len(store.advanced) != 1 || store.advanced[0] != governance.LibraryJobFailed {
t.Fatalf("advanced = %v", store.advanced)
}
}
// 任一步释放失败:不报成功,作业失败,维护状态写 needs_restore(等人工/接管)。
func TestFinishLibrarySuccessStopsOnReleaseFailure(t *testing.T) {
for _, step := range []string{"scope_release", "instance_state", "instance_release"} {
t.Run(step, func(t *testing.T) {
store := &fakeFinishStore{failOn: step}
err := finishLibrarySuccess(context.Background(), store, fakeHealth{}, "kgseal-weknora", "internal", "job-3", boundary())
if err == nil {
t.Fatalf("%s failure was swallowed", step)
}
for _, call := range store.calls {
if call == "advance_succeeded" {
t.Fatalf("reported success after %s failed: %v", step, store.calls)
}
}
if len(store.advanced) != 1 || store.advanced[0] != governance.LibraryJobFailed {
t.Fatalf("advanced = %v", store.advanced)
}
})
}
}
// 终态写入失败:锁已释放,但不能声称成功;作业保持非终态等接管。
func TestFinishLibrarySuccessReportsTerminalWriteFailure(t *testing.T) {
store := &fakeFinishStore{failOn: "advance_succeeded"}
err := finishLibrarySuccess(context.Background(), store, fakeHealth{}, "kgseal-weknora", "internal", "job-4", boundary())
if err == nil || !strings.Contains(err.Error(), "等待接管") {
t.Fatalf("err = %v", err)
}
}
// 接管判定只认证据:finalize 阶段成功或存在未被顶掉的生效版本。
func TestNeedsFinalizeOnlyUsesRealEvidence(t *testing.T) {
job := governance.LibraryJob{Scope: "internal", Kind: governance.LibraryJobSync, TargetSnapshotID: "snap-1"}
if needsFinalizeOnly(job, nil) {
t.Fatal("job without evidence treated as finalized")
}
withStage := job
withStage.Stages = []governance.LibraryStage{{Stage: "finalize", State: "succeeded"}}
if !needsFinalizeOnly(withStage, nil) {
t.Fatal("finalize stage ignored")
}
versions := []governance.LibraryVersion{{Kind: governance.LibraryVersionProductionEffective, SnapshotID: "snap-1"}}
if !needsFinalizeOnly(job, versions) {
t.Fatal("effective version ignored")
}
superseded := []governance.LibraryVersion{{Kind: governance.LibraryVersionProductionEffective,
SnapshotID: "snap-1", SupersededAt: ptrTime(time.Now())}}
if needsFinalizeOnly(job, superseded) {
t.Fatal("superseded version treated as evidence")
}
other := []governance.LibraryVersion{{Kind: governance.LibraryVersionProductionEffective, SnapshotID: "snap-2"}}
if needsFinalizeOnly(job, other) {
t.Fatal("version of another snapshot treated as evidence")
}
}
func ptrTime(value time.Time) *time.Time { return &value }
+27 -25
View File
@@ -117,6 +117,13 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
// 绝不重新备份——重新备份会把"半写状态"当成新基线,覆盖真正可回滚的那份。
previous, err := store.LibraryJob(ctx, payload.LibraryJobID)
resumed := err == nil && previous.SnapshotID != ""
if resumed {
// 上次可能死在"锁已放、终态没写"之间:这种情况下只复核与登记,不重写生产。
if versions, versionErr := store.LibraryVersions(ctx, payload.Scope, 50); versionErr == nil &&
needsFinalizeOnly(previous, versions) {
payload.FinalizeOnly = true
}
}
if resumed {
recorder.Stage(ctx, "prebackup", "skipped", "接管失败前已创建的预备份 "+previous.SnapshotID)
}
@@ -257,6 +264,7 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
if resumed {
contentState = contentHalfWritten
}
pending := successBoundary{}
tracked := &contentTrackedTarget{inner: target, onApply: func() { contentState = contentHalfWritten }}
allowRestore := func(error) bool {
return contentState == contentUntouched || contentState == contentSucceeded || contentState == contentRolledBack
@@ -291,8 +299,9 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
return errors.New(message)
}
recorder.Stage(ctx, "finalize", "succeeded", "已登记生效版本(仅收尾,未重写生产)")
_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobSucceeded, "finalize", "", "",
previous.SnapshotID, payload.TargetSnapshotID, time.Now().UTC())
// 成功要等维护锁释放完才暴露:这里只记"待收尾"。
pending = successBoundary{Ready: true, Backup: previous.SnapshotID,
Target: payload.TargetSnapshotID, FinishedAt: time.Now().UTC()}
return nil
}
@@ -341,13 +350,10 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
return errors.New(message)
}
recorder.Stage(ctx, "finalize", "succeeded", "已登记生效版本")
if advanceErr := store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobSucceeded, "finalize", "", "",
outcome.Backup.SnapshotID, payload.TargetSnapshotID, finishedAt); advanceErr != nil {
// 状态落不下去时不要把队列作业报成功:让上层可见并等待接管。
return fmt.Errorf("作业状态写入失败: %w", advanceErr)
}
log.Printf("library %s succeeded job=%s scope=%s snapshot=%s backup=%s", kind,
payload.LibraryJobID, payload.Scope, payload.TargetSnapshotID, outcome.Backup.SnapshotID)
// 内容与版本都落定了,但还不能报成功:服务健康、入口开放、两把维护锁
// 释放完成才允许暴露 succeeded(见 finishLibrarySuccess)。
pending = successBoundary{Ready: true, Backup: outcome.Backup.SnapshotID,
Target: payload.TargetSnapshotID, FinishedAt: finishedAt}
return nil
case "rolled_back":
contentState = contentRolledBack
@@ -408,10 +414,19 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
}
return windowErr
case contentState == contentSucceeded:
if err := releaseWindowLocks(ctx, store, identity.Instance, payload.Scope, payload.LibraryJobID,
"写入与校验完成,目标已按原运行状态恢复", now2); err != nil {
return fmt.Errorf("目标已恢复但维护占用释放失败: %w", err)
if !pending.Ready {
// 内容状态说成功但没有收尾凭据(不可能两者都缺):按未收尾处理,保持维护。
note := "内容状态为已验证,但缺少收尾凭据,目标保持维护"
_ = store.UpdateLibraryMaintenanceState(ctx, payload.Scope, payload.LibraryJobID, "needs_restore", note, now2)
_ = store.UpdateInstanceMaintenanceState(ctx, identity.Instance, payload.LibraryJobID, "needs_restore", note, now2)
return errors.New(note)
}
if finishErr := finishLibrarySuccess(ctx, store, manager, identity.Instance, payload.Scope,
payload.LibraryJobID, pending); finishErr != nil {
return finishErr
}
log.Printf("library %s succeeded job=%s scope=%s snapshot=%s backup=%s", kind,
payload.LibraryJobID, payload.Scope, payload.TargetSnapshotID, pending.Backup)
return windowErr
case contentState == contentRolledBack:
if err := releaseWindowLocks(ctx, store, identity.Instance, payload.Scope, payload.LibraryJobID,
@@ -429,19 +444,6 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
}
}
// releaseWindowLocks 是唯一释放路径:先记录入口已恢复,再放 scope 行与实例锁。
// 顺序反过来会出现"实例锁放了、scope 还挂着"的中间态,页面就会说正在维护。
func releaseWindowLocks(ctx context.Context, store *postgresstore.Store, instance, scope, jobID, note string, now time.Time) error {
_ = store.SetInstanceMaintenanceIngress(ctx, instance, jobID, "open", note, now)
_ = store.SetLibraryMaintenanceIngress(ctx, scope, jobID, "open", note, now)
_ = store.UpdateInstanceMaintenanceState(ctx, instance, jobID, "verified", note, now)
_ = store.UpdateLibraryMaintenanceState(ctx, scope, jobID, "verified", note, now)
if err := store.ReleaseInstanceMaintenance(ctx, instance, jobID, now); err != nil {
return err
}
return store.ReleaseLibraryMaintenance(ctx, scope, jobID, now)
}
// resumeInterruptedLibraryJobs 在 worker 启动时接管中断的写作业:
// 已经持久化预备份的重新排队(同一个业务作业 id,阶段与备份都接着用),
// 连预备份都没完成的才判失败——那种情况没有可回滚的基线,不能盲写。