diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d79e20..dbdc228 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,9 @@ - 传输阶段在真实 UI 上不可见:此前 running 阶段是无条件 `UpdateLibraryStage`(读 owner 再写,存在竞态), 成功阶段在交接后按旧 owner 写必然 0 行,且 `library_job_stages` 里根本没有 handoff 行。 现在阶段创建/推进/成功/失败都与 job 行的所有权判定在同一事务里完成,交接提交后阶段一定是 succeeded。 +- 删除 handoff 路径残留的无条件 `UpdateLibraryStage`:它与 `BeginLibraryHandoff` 事务里的写重复, + 并在 Begin 返回到这次写之间给交接留出竞态窗口。现在 handoff 的所有阶段写只发生在 + Begin/Complete/Fail 三个持有 job 行锁的事务方法里;新增源码契约测试防止回流。 - 身份判定漏掉"目标权威基线":`library_baseline` 是目标侧真实捕获,但只登记 `production_effective` (证据 action=target_baseline_init / origin=target_baseline)。现在原产地按捕获证据判定: 封存=源、预备份=目标、带基线证据的 effective=目标;普通 effective 与 recovery_verified 只是引用, diff --git a/cmd/worker/library_handoff.go b/cmd/worker/library_handoff.go index d0e9fbd..7f13b73 100644 --- a/cmd/worker/library_handoff.go +++ b/cmd/worker/library_handoff.go @@ -115,6 +115,9 @@ func runLibraryHandoffForQueue(ctx context.Context, store *postgresstore.Store, return nil } // 阶段与所有权在同一事务里建立:不存在"先读 owner 再写 stage"的竞态窗口。 + // 阶段写一律只发生在 Begin/Complete/Fail 三个持有 job 行锁的事务方法里: + // 这里若再补一次无条件 UpdateLibraryStage,既重复事务里的写,又会在 + // Begin 返回与这次写之间被交接抢走——那正是上一轮退回的原因。 started, beginErr := store.BeginLibraryHandoff(ctx, payload.LibraryJobID, queueJobID, "正在把快照送到发布执行器并校验", now) if beginErr != nil { @@ -124,8 +127,6 @@ func runLibraryHandoffForQueue(ctx context.Context, store *postgresstore.Store, return fmt.Errorf("业务作业 %s 当前不由本次 handoff 持有(%q),未做任何修改", payload.LibraryJobID, record.QueueJobID) } - _ = store.UpdateLibraryStage(ctx, payload.LibraryJobID, libraryHandoffStage, "running", nil, nil, nil, nil, - "正在把快照送到发布执行器并校验", now) evidence, err := handoff.Push(ctx, want) if err != nil { message := "快照未送到发布执行器:" + err.Error() diff --git a/cmd/worker/library_handoff_stage_contract_test.go b/cmd/worker/library_handoff_stage_contract_test.go new file mode 100644 index 0000000..a635af3 --- /dev/null +++ b/cmd/worker/library_handoff_stage_contract_test.go @@ -0,0 +1,81 @@ +package main + +import ( + "bytes" + "context" + "os" + "testing" + + "knowledge-governance/internal/governance" +) + +// handoff 路径的 stage 写只允许发生在 Begin/Complete/Fail 这三个持有 job 行锁的 +// 事务方法里。这条契约测试挡的是"又在事务外面补一次无条件 UpdateLibraryStage" +// —— 它重复事务里的写,并在 Begin 返回与这次写之间给交接留出竞态窗口(上一轮退回点)。 +// +// 说明:这是源码契约,不能替代已经跑过的事务行为测试;它只负责让回流立刻变红。 +func TestHandoffStagesOnlyWrittenThroughOwnershipTransactions(t *testing.T) { + source, err := os.ReadFile("library_handoff.go") + if err != nil { + t.Fatalf("读不到 library_handoff.go:%v", err) + } + if bytes.Contains(source, []byte("store.UpdateLibraryStage(")) { + t.Fatal("handoff 路径又出现了无条件的 store.UpdateLibraryStage:阶段写必须在 Begin/Complete/Fail 的事务里") + } + for _, method := range []string{ + "store.BeginLibraryHandoff(", + "store.CompleteLibraryHandoff(", + "store.FailLibraryHandoff(", + } { + if !bytes.Contains(source, []byte(method)) { + t.Fatalf("handoff 路径没有走事务方法 %s", method) + } + } +} + +// 行为侧:所有权不属于本次 handoff 时,连阶段行都不该被建出来、作业状态也不动。 +func TestUnownedHandoffCreatesNoStageRow(t *testing.T) { + pool, store := fixturePool(t) + ctx := context.Background() + resetFixtureJobs(t, pool) + fixture := newHandoffFixture(t) + config := appconfigFixture(fixture) + snapshot := fixture.unknownSnapshotID() + job := createFixtureJob(t, store, "internal", snapshot) + recordSealedVersion(t, store, snapshot, "internal") + releaseID := governance.QueueJobIDFor(job.ID) + if _, err := pool.Exec(ctx, `UPDATE library_jobs SET state='running', queue_job_id=$2 WHERE id=$1`, + job.ID, releaseID); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `DELETE FROM library_job_stages WHERE job_id=$1 AND stage='handoff'`, job.ID); err != nil { + t.Fatal(err) + } + incoming := libraryHandoffPayload{LibraryJobID: job.ID, Scope: "internal", + TargetSnapshotID: snapshot, NextType: librarySyncJobType} + // 拿一个既不是当前 owner、也不是这份作业 handoff ID 的队列号:必须拒绝。 + if err := runLibraryHandoffForQueue(ctx, store, config, fixture.handoff, incoming, + "not-the-owner-"+snapshot[:8]); err == nil { + t.Fatal("错误前驱被接受了") + } + // 用这份作业真正的 handoff ID 重放:已交接 ⇒ 幂等成功,但也必须零写入。 + if err := runLibraryHandoffForQueue(ctx, store, config, fixture.handoff, incoming, + governance.LibraryHandoffQueueJobID(job.ID)); err != nil { + t.Fatalf("交接后的重放应当幂等成功,得到 %v", err) + } + var rows int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM library_job_stages WHERE job_id=$1 AND stage='handoff'`, + job.ID).Scan(&rows); err != nil { + t.Fatal(err) + } + if rows != 0 { + t.Fatalf("未被持有的 handoff 建出了 %d 行阶段记录", rows) + } + after, err := store.LibraryJob(ctx, job.ID) + if err != nil { + t.Fatal(err) + } + if after.State != governance.LibraryJobRunning || after.Error != "" { + t.Fatalf("未持有的 handoff 改动了作业:%s / %q", after.State, after.Error) + } +} diff --git a/scripts/test-scope.json b/scripts/test-scope.json index 306d6a6..b0352ba 100644 --- a/scripts/test-scope.json +++ b/scripts/test-scope.json @@ -297,6 +297,11 @@ "package": "knowledge-governance/cmd/worker", "test": "TestSnapshotOriginOnlyFromCaptureEvents", "reason": "需要隔离 fixture PostgreSQL:验证 test_sealed 与 production_effective 共存(正常成功链)仍按源绑定识别,以及真冲突/无捕获事件被拒。本机已实跑通过" + }, + { + "package": "knowledge-governance/cmd/worker", + "test": "TestUnownedHandoffCreatesNoStageRow", + "reason": "需要隔离 fixture PostgreSQL:验证未被持有的 handoff 不建阶段行、不改作业,且用真实 handoff ID 重放时幂等且零写入。CI 无测试库;本机已实跑(go test ./cmd/worker -count=1 → ok)" } ] }