diff --git a/CHANGELOG.md b/CHANGELOG.md index a976945..49dd5a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,14 @@ ### Fixed +- **接管在版本查询失败时 fail-open**:`resumed` 后版本查询出错被忽略、`FinalizeOnly` 仍为 + false,于是"已成功登记并释放窗口、只差终态"的作业会被**重新写一遍目标**。现在接管先按 + 证据分类:`finalize` 阶段成功是独立证据(不受版本查询影响);查询失败且无阶段证据 ⇒ + **未知**,零写入、作业失败可重试;本快照的生效版本已被别的快照顶替 ⇒ **拒绝覆盖**; + 只有真正没有收尾证据时才按正常流程。 +- **维护释放部分成功后的重试不再永久卡死**:`Update*/Release*` 对"已由本作业释放"的行 + 返回可判别的 `ErrMaintenanceReleased`,收尾重试视其为已完成;别的作业来释放仍是 + "未持有"错误。真 PG 负例:同作业重复 Update/Release 得到已释放,跨作业得到未持有。 - **作业成功暴露早于维护结束**:原来 `execute` 里写完生效版本就 `AdvanceLibraryJob(succeeded)`, 而服务恢复、入口开放与维护锁释放要在窗口返回后才做。任何一步失败时外部已经看到 `succeeded`,重启接管又只认非终态,锁就永远挂着。现在内容校验/版本登记之后只记 diff --git a/cmd/worker/library_finish.go b/cmd/worker/library_finish.go index 0b1628c..e8ba1eb 100644 --- a/cmd/worker/library_finish.go +++ b/cmd/worker/library_finish.go @@ -28,22 +28,22 @@ type libraryMaintenanceStore interface { // 才能保证任何新作业在全部记录落定之前都进不来。每一步的错误都必须向上暴露—— // 只丢日志会让"锁其实没放"这件事变成页面上的成功。 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 { + if err := store.SetLibraryMaintenanceIngress(ctx, scope, jobID, "open", note, now); !releasedOrNil(err) { return fmt.Errorf("记录入口已开放失败(scope):%w", err) } - if err := store.SetInstanceMaintenanceIngress(ctx, instance, jobID, "open", note, now); err != nil { + if err := store.SetInstanceMaintenanceIngress(ctx, instance, jobID, "open", note, now); !releasedOrNil(err) { return fmt.Errorf("记录入口已开放失败(instance):%w", err) } - if err := store.UpdateLibraryMaintenanceState(ctx, scope, jobID, "verified", note, now); err != nil { + if err := store.UpdateLibraryMaintenanceState(ctx, scope, jobID, "verified", note, now); !releasedOrNil(err) { return fmt.Errorf("维护状态写入失败(scope):%w", err) } - if err := store.ReleaseLibraryMaintenance(ctx, scope, jobID, now); err != nil { + if err := store.ReleaseLibraryMaintenance(ctx, scope, jobID, now); !releasedOrNil(err) { return fmt.Errorf("维护占用释放失败(scope):%w", err) } - if err := store.UpdateInstanceMaintenanceState(ctx, instance, jobID, "verified", note, now); err != nil { + if err := store.UpdateInstanceMaintenanceState(ctx, instance, jobID, "verified", note, now); !releasedOrNil(err) { return fmt.Errorf("维护状态写入失败(instance):%w", err) } - if err := store.ReleaseInstanceMaintenance(ctx, instance, jobID, now); err != nil { + if err := store.ReleaseInstanceMaintenance(ctx, instance, jobID, now); !releasedOrNil(err) { return fmt.Errorf("实例占用释放失败(instance):%w", err) } return nil @@ -101,24 +101,62 @@ func finishLibrarySuccess(ctx context.Context, store libraryMaintenanceStore, he return nil } -// needsFinalizeOnly 判断一个被接管的作业是否"内容与版本都已落定、只差收尾"。 -// 判据只认真实证据:终态版本记录存在且未被顶掉,或 finalize 阶段已成功。 -func needsFinalizeOnly(job governance.LibraryJob, versions []governance.LibraryVersion) bool { - if job.TargetSnapshotID == "" { - return false - } +// takeoverKind 是接管时"这个作业走到哪一步"的分类。 +// +// 四类必须分开:把"版本查不到"当成"没有收尾证据",就会把只差收尾的作业重新写一遍 +// 目标;把"曾被顶替"当成正常重放,就会用旧快照覆盖更新的生产版本。 +type takeoverKind string + +const ( + // takeoverFresh:没有任何收尾证据,按正常流程(有预备份则复用,写前仍备份)。 + takeoverFresh takeoverKind = "fresh" + // takeoverPendingFinal:内容与版本已落定,只差收尾(只复核与登记,不重写)。 + takeoverPendingFinal takeoverKind = "pending_final" + // takeoverSuperseded:本快照曾经生效、后来被别的快照顶替,不得自动覆盖。 + takeoverSuperseded takeoverKind = "superseded" + // takeoverUnknown:版本状态读不到且没有阶段证据,属"未知",不得写入。 + takeoverUnknown takeoverKind = "unknown" +) + +// classifyTakeover 只按证据分类: +// - finalize 阶段成功是独立证据(来自作业自己的阶段行),不因版本查询失败而丢弃; +// - 本快照存在 released/被顶替的版本 ⇒ superseded; +// - 本快照存在未被顶掉的生效版本 ⇒ pending_final; +// - 查询失败且无阶段证据 ⇒ unknown(可重试,不当成没有证据)。 +func classifyTakeover(job governance.LibraryJob, versions []governance.LibraryVersion, versionsErr error) takeoverKind { + finalized := false for _, stage := range job.Stages { if stage.Stage == "finalize" && stage.State == "succeeded" { - return true + finalized = true } } + if versionsErr != nil && !finalized { + return takeoverUnknown + } + superseded, effective := false, false for _, version := range versions { - if version.Kind == governance.LibraryVersionProductionEffective && version.SupersededAt == nil && - version.SnapshotID == job.TargetSnapshotID { - return true + if version.Kind != governance.LibraryVersionProductionEffective || version.SnapshotID != job.TargetSnapshotID { + continue + } + if version.SupersededAt != nil { + superseded = true + } else { + effective = true } } - return false + if superseded && !effective { + return takeoverSuperseded + } + if effective || finalized { + return takeoverPendingFinal + } + return takeoverFresh } var _ libraryMaintenanceStore = (*postgresstore.Store)(nil) + +// releasedOrNil:这一行已经由本作业释放过,说明该步在上一次尝试里已经完成。 +// 部分成功后重试幂等靠它——否则"scope 放了、instance 没放"的作业永远重试不下去。 +func releasedOrNil(err error) bool { + return err == nil || errors.Is(err, postgresstore.ErrMaintenanceReleased) +} diff --git a/cmd/worker/library_finish_test.go b/cmd/worker/library_finish_test.go index 29d797c..08eca3c 100644 --- a/cmd/worker/library_finish_test.go +++ b/cmd/worker/library_finish_test.go @@ -8,12 +8,22 @@ import ( "time" "knowledge-governance/internal/governance" + postgresstore "knowledge-governance/internal/platform/postgres" ) type fakeFinishStore struct { - calls []string - failOn string - advanced []governance.LibraryJobState + calls []string + failOn string + alreadyReleased bool + advanced []governance.LibraryJobState +} + +// 模拟"这一行已由同一个作业释放过"(真实 Store 返回 ErrMaintenanceReleased)。 +func (f *fakeFinishStore) releasedSentinel() error { + if f.alreadyReleased { + return postgresstore.ErrMaintenanceReleased + } + return nil } func (f *fakeFinishStore) record(name string) error { @@ -25,21 +35,39 @@ func (f *fakeFinishStore) record(name string) error { } func (f *fakeFinishStore) SetInstanceMaintenanceIngress(context.Context, string, string, string, string, time.Time) error { + if err := f.releasedSentinel(); err != nil { + return err + } return f.record("instance_ingress") } func (f *fakeFinishStore) SetLibraryMaintenanceIngress(context.Context, string, string, string, string, time.Time) error { + if err := f.releasedSentinel(); err != nil { + return err + } return f.record("scope_ingress") } func (f *fakeFinishStore) UpdateInstanceMaintenanceState(context.Context, string, string, string, string, time.Time) error { + if err := f.releasedSentinel(); err != nil { + return err + } return f.record("instance_state") } func (f *fakeFinishStore) UpdateLibraryMaintenanceState(context.Context, string, string, string, string, time.Time) error { + if err := f.releasedSentinel(); err != nil { + return err + } return f.record("scope_state") } func (f *fakeFinishStore) ReleaseInstanceMaintenance(context.Context, string, string, time.Time) error { + if err := f.releasedSentinel(); err != nil { + return err + } return f.record("instance_release") } func (f *fakeFinishStore) ReleaseLibraryMaintenance(context.Context, string, string, time.Time) error { + if err := f.releasedSentinel(); err != nil { + return err + } return f.record("scope_release") } func (f *fakeFinishStore) AdvanceLibraryJob(_ context.Context, _ string, state governance.LibraryJobState, @@ -120,29 +148,61 @@ func TestFinishLibrarySuccessReportsTerminalWriteFailure(t *testing.T) { } } -// 接管判定只认证据:finalize 阶段成功或存在未被顶掉的生效版本。 -func TestNeedsFinalizeOnlyUsesRealEvidence(t *testing.T) { +// 接管分类:阶段证据独立成立;查询失败=未知(不得写入);被顶替=不得覆盖。 +func TestClassifyTakeoverSeparatesUnknownPendingSupersededAndFresh(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, + finalized := job + finalized.Stages = []governance.LibraryStage{{Stage: "finalize", State: "succeeded"}} + effective := []governance.LibraryVersion{{Kind: governance.LibraryVersionProductionEffective, SnapshotID: "snap-1"}} + supersededVersion := []governance.LibraryVersion{{Kind: governance.LibraryVersionProductionEffective, SnapshotID: "snap-1", SupersededAt: ptrTime(time.Now())}} - if needsFinalizeOnly(job, superseded) { - t.Fatal("superseded version treated as evidence") + unrelated := []governance.LibraryVersion{{Kind: governance.LibraryVersionProductionEffective, SnapshotID: "snap-9"}} + queryErr := errors.New("version storage temporarily unavailable") + + cases := []struct { + name string + job governance.LibraryJob + versions []governance.LibraryVersion + err error + want takeoverKind + }{ + // 关键回归:查询失败 + 没有阶段证据 ⇒ 未知,不能当成"没有收尾证据"继续写。 + {"query fails without stage evidence", job, nil, queryErr, takeoverUnknown}, + // 阶段证据先独立成立:查询失败不影响"只差收尾"的判定。 + {"query fails but finalize stage succeeded", finalized, nil, queryErr, takeoverPendingFinal}, + {"effective version for this snapshot", job, effective, nil, takeoverPendingFinal}, + {"finalize stage succeeded", finalized, nil, nil, takeoverPendingFinal}, + {"this snapshot was superseded", job, supersededVersion, nil, takeoverSuperseded}, + {"no evidence at all", job, nil, nil, takeoverFresh}, + {"another snapshot effective but this one has no row", job, unrelated, nil, takeoverFresh}, + {"empty target snapshot", governance.LibraryJob{Scope: "internal"}, nil, nil, takeoverFresh}, } - other := []governance.LibraryVersion{{Kind: governance.LibraryVersionProductionEffective, SnapshotID: "snap-2"}} - if needsFinalizeOnly(job, other) { - t.Fatal("version of another snapshot treated as evidence") + for _, test := range cases { + t.Run(test.name, func(t *testing.T) { + if got := classifyTakeover(test.job, test.versions, test.err); got != test.want { + t.Fatalf("classify = %q, want %q", got, test.want) + } + }) + } +} + +// 部分释放后重试必须幂等:已由本作业释放的行不能再算失败。 +func TestReleaseWindowLocksToleratesAlreadyReleasedRows(t *testing.T) { + store := &fakeFinishStore{} + if err := releaseWindowLocks(context.Background(), store, "kgseal-weknora", "internal", "job-1", + "verified", time.Now().UTC()); err != nil { + t.Fatal(err) + } + store = &fakeFinishStore{alreadyReleased: true} + if err := releaseWindowLocks(context.Background(), store, "kgseal-weknora", "internal", "job-1", + "verified", time.Now().UTC()); err != nil { + t.Fatalf("retry after a partial release must be idempotent: %v", err) + } + // 真正"不是本作业持有"的错误仍然要暴露。 + store = &fakeFinishStore{failOn: "instance_release"} + if err := releaseWindowLocks(context.Background(), store, "kgseal-weknora", "internal", "job-1", + "verified", time.Now().UTC()); err == nil { + t.Fatal("a foreign owner error was swallowed") } } diff --git a/cmd/worker/library_handlers.go b/cmd/worker/library_handlers.go index e256eae..331c6eb 100644 --- a/cmd/worker/library_handlers.go +++ b/cmd/worker/library_handlers.go @@ -118,9 +118,17 @@ 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) { + versions, versionErr := store.LibraryVersions(ctx, payload.Scope, 50) + switch classifyTakeover(previous, versions, versionErr) { + case takeoverUnknown: + // 版本状态读不到、又没有阶段证据:不能当成"没有收尾证据"继续写目标。 + return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, + "接管时读不到该目标的版本状态,无法判断是只差收尾还是要重做;未对目标执行任何写入,稍后重试", "prebackup") + case takeoverSuperseded: + return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, + "本快照的生效版本已被别的快照顶替,拒绝用旧快照覆盖当前生产版本", "prebackup") + case takeoverPendingFinal: + // 上次可能死在"锁已放、终态没写"之间:只复核与登记,不重写生产。 payload.FinalizeOnly = true } } diff --git a/internal/platform/postgres/library_instance_maintenance.go b/internal/platform/postgres/library_instance_maintenance.go index 6a6b468..6223b44 100644 --- a/internal/platform/postgres/library_instance_maintenance.go +++ b/internal/platform/postgres/library_instance_maintenance.go @@ -92,6 +92,9 @@ WHERE instance=$1 AND owner_job_id=$2 AND released_at IS NULL`, instance, ownerJ return fmt.Errorf("update instance maintenance: %w", err) } if tag.RowsAffected() == 0 { + if s.releasedByThisJob(ctx, "library_instance_maintenance", "instance", instance, ownerJobID) { + return ErrMaintenanceReleased + } return errors.New("instance maintenance not held by this job") } return nil @@ -108,6 +111,9 @@ WHERE instance=$1 AND owner_job_id=$2 AND released_at IS NULL`, instance, ownerJ return fmt.Errorf("update instance maintenance ingress: %w", err) } if tag.RowsAffected() == 0 { + if s.releasedByThisJob(ctx, "library_instance_maintenance", "instance", instance, ownerJobID) { + return ErrMaintenanceReleased + } return errors.New("instance maintenance not held by this job") } return nil @@ -121,6 +127,9 @@ WHERE instance=$1 AND owner_job_id=$2 AND released_at IS NULL`, instance, ownerJ return fmt.Errorf("release instance maintenance: %w", err) } if tag.RowsAffected() == 0 { + if s.releasedByThisJob(ctx, "library_instance_maintenance", "instance", instance, ownerJobID) { + return ErrMaintenanceReleased + } return errors.New("instance maintenance not held by this job") } return nil diff --git a/internal/platform/postgres/library_maintenance.go b/internal/platform/postgres/library_maintenance.go index 27d6a34..62eca4a 100644 --- a/internal/platform/postgres/library_maintenance.go +++ b/internal/platform/postgres/library_maintenance.go @@ -10,6 +10,19 @@ import ( "github.com/jackc/pgx/v5" ) +var ErrMaintenanceReleased = errors.New("maintenance already released by this job") + +// releasedByThisJob 判断"零行更新"是因为本作业已经释放过,还是根本没持有。 +func (s *Store) releasedByThisJob(ctx context.Context, table, keyColumn, key, ownerJobID string) bool { + var owner string + var released *time.Time + query := "SELECT owner_job_id, released_at FROM " + table + " WHERE " + keyColumn + "=$1" + if err := s.db.QueryRow(ctx, query, key).Scan(&owner, &released); err != nil { + return false + } + return owner == ownerJobID && released != nil +} + // LibraryMaintenance 是目标维护窗口的持有记录:一个 scope 只能有一个持有者。 // // Level 说明这次维护到底动了什么: @@ -97,6 +110,9 @@ WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL`, scope, ownerJobID, return fmt.Errorf("update library maintenance: %w", err) } if tag.RowsAffected() == 0 { + if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) { + return ErrMaintenanceReleased + } return errors.New("library maintenance not held by this job") } return nil @@ -114,6 +130,9 @@ WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL`, scope, ownerJobID, return fmt.Errorf("update library maintenance ingress: %w", err) } if tag.RowsAffected() == 0 { + if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) { + return ErrMaintenanceReleased + } return errors.New("library maintenance not held by this job") } return nil @@ -127,6 +146,9 @@ WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL`, scope, ownerJobID, return fmt.Errorf("release library maintenance: %w", err) } if tag.RowsAffected() == 0 { + if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) { + return ErrMaintenanceReleased + } return errors.New("library maintenance not held by this job") } return nil