fix(library): 接管不再把版本查询失败当成没有收尾证据

resumed 分支里 LibraryVersions 的错误被忽略，FinalizeOnly 保持 false 后继续完整
apply：对"已登记版本并释放窗口、只差终态"的作业，这等于把同一份内容重写一遍目标，
与"最终只复核"的承诺相反。

现在接管先分类：finalize 阶段成功作为独立证据（不受版本查询影响）→ 只收尾；查询
失败且无阶段证据 → 未知，零写入、可重试；本快照被别的快照顶替 → 拒绝覆盖；其余才
按正常流程。

同时修释放幂等：Update*/Release* 对"已由本作业释放"的行返回 ErrMaintenanceReleased，
收尾重试视为已完成（部分释放后不会永久卡死），别的作业仍报未持有。真 PG 负例验证
两种情形。

CHANGELOG.md

@@ -38,6 +38,14 @@
 
 ### Fixed
 
+- **接管在版本查询失败时 fail-open**：`resumed` 后版本查询出错被忽略、`FinalizeOnly` 仍为
+  false，于是"已成功登记并释放窗口、只差终态"的作业会被**重新写一遍目标**。现在接管先按
+  证据分类：`finalize` 阶段成功是独立证据（不受版本查询影响）；查询失败且无阶段证据 ⇒
+  **未知**，零写入、作业失败可重试；本快照的生效版本已被别的快照顶替 ⇒ **拒绝覆盖**；
+  只有真正没有收尾证据时才按正常流程。
+- **维护释放部分成功后的重试不再永久卡死**：`Update*/Release*` 对"已由本作业释放"的行
+  返回可判别的 `ErrMaintenanceReleased`，收尾重试视其为已完成；别的作业来释放仍是
+  "未持有"错误。真 PG 负例：同作业重复 Update/Release 得到已释放，跨作业得到未持有。
 - **作业成功暴露早于维护结束**：原来 `execute` 里写完生效版本就 `AdvanceLibraryJob(succeeded)`，
   而服务恢复、入口开放与维护锁释放要在窗口返回后才做。任何一步失败时外部已经看到
   `succeeded`，重启接管又只认非终态，锁就永远挂着。现在内容校验/版本登记之后只记

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)
+}

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")
 	}
 }
 

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
 		}
 	}

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

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