diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e99a7b..a976945 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 : `），不含凭据。 diff --git a/cmd/worker/library_finish.go b/cmd/worker/library_finish.go new file mode 100644 index 0000000..0b1628c --- /dev/null +++ b/cmd/worker/library_finish.go @@ -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) diff --git a/cmd/worker/library_finish_test.go b/cmd/worker/library_finish_test.go new file mode 100644 index 0000000..29d797c --- /dev/null +++ b/cmd/worker/library_finish_test.go @@ -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 } diff --git a/cmd/worker/library_handlers.go b/cmd/worker/library_handlers.go index bdb3973..e256eae 100644 --- a/cmd/worker/library_handlers.go +++ b/cmd/worker/library_handlers.go @@ -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，阶段与备份都接着用）， // 连预备份都没完成的才判失败——那种情况没有可回滚的基线，不能盲写。