姚章浩
a7d643635a
build / build (push) Successful in 1m27s
R1 维护/发布守卫硬编码 j.id=$1：维护类入口的 $1 是 scope/instance，作业号在 $2， 导致 update/release/ingress 六个入口正常路径全部匹配不到作业。改为按被更新行自己的 owner_job_id 生成条件，并同时要求该代次在 library_attempts 里仍活跃。 R2 认领只改 jobs.attempt_epoch 不改 attempts.epoch，第二次接管即错位；唯一索引 (job_id) 与"每次接管新 attempt 且历史不可变"矛盾。改为同作业多代次：锁作业行与 operation 行、同 owner 幂等续租、活跃未过期者让路、过期/本作业旧代次标 abandoned、 新代次 epoch=max+1，并清同 operation 其它作业行租约。迟到 write/renew/finish 一律 ErrLibraryEpochStale（finish 补了"只能在活跃态进来一次"）。 R3 续作不再新建 -resume 作业：PrepareLibraryJobResume 让原业务作业回 pending 并显式 结束上一代，维护窗口/journal/预备份/写集合仍按原作业 id 归属，代次由 worker 起手递增。 R4 执行域准入改为注册 handler 前读事实：目标实况（KB/租户/schema/索引）、旧 release 路由绑定的 KB、未释放维护窗口持有者；未知或重叠即拒绝执行（不做维护动作、不停既有服务）。 同时自纠 5 项设计偏离：current 次序键取意图创建时间（旧意图迟到事实被拒）、首次锚点 改事务 FOR UPDATE + 回读且 NULL≠0、版本事件冲突只做加法且 xmax=0 才算新事实、 恢复作为新 kind=recover 意图推进 current、隔离测试加测试库护栏并清理残留 schema。 测试：真实 PG（kg-pgtest-writeset，一用例一 schema）新增 8 条集成用例覆盖上述正负例； go test ./... 全绿（50 包）、go vet/build rc=0、web tsc + vitest 485 通过、make verify 16 项通过。
248 lines
11 KiB
Go
248 lines
11 KiB
Go
package postgres
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
|
||
"knowledge-governance/internal/governance"
|
||
)
|
||
|
||
// 写入栅栏：所有"业务状态 / 维护锁"的写都必须证明调用方仍是**当前活跃代次**。
|
||
//
|
||
// 两条条件缺一不可：
|
||
// 1. 作业行上的 (attempt_epoch, lease_owner) 就是这次的 (epoch, owner)；
|
||
// 2. library_attempts 里那一代次仍是 active（claimed/running）。
|
||
//
|
||
// 为什么第 2 条不能省（R2）：接管/续跑会激活新一代次，而**旧作业行上的 epoch/owner
|
||
// 字段仍原样保留**。只比较旧作业行，旧执行器的守卫会继续通过——它就能继续写业务状态、
|
||
// 释放别人刚拿到的锁。把"该代次仍活跃"作为条件后，旧代次被结束的那一刻起，
|
||
// 它的所有守卫写都会失败。
|
||
//
|
||
// 参数契约（R1）：作业 id 取自**被更新行自己的列**（`library_maintenance.owner_job_id`
|
||
// 等），不写死 `$1`——维护类更新的 `$1` 是 scope/instance，写死 `$1` 会让守卫永远匹配
|
||
// 不到作业，正常写与释放全部失败。列表达式由调用方传入，序号漂移不可能再发生。
|
||
|
||
// libraryFence 生成"调用方仍是当前活跃代次"的谓词。jobIDExpr 是**列表达式**。
|
||
func libraryFence(jobIDExpr string, epochParam, ownerParam int) string {
|
||
return fmt.Sprintf(`EXISTS (
|
||
SELECT 1 FROM library_jobs j
|
||
WHERE j.id = %s
|
||
AND j.attempt_epoch = $%d
|
||
AND j.lease_owner = $%d
|
||
AND (
|
||
j.operation_id = ''
|
||
OR EXISTS (
|
||
SELECT 1 FROM library_attempts a
|
||
WHERE a.operation_id = j.operation_id
|
||
AND a.epoch = j.attempt_epoch
|
||
AND a.owner = j.lease_owner
|
||
AND a.state IN ('claimed','running')
|
||
)
|
||
)
|
||
)`, jobIDExpr, epochParam, ownerParam)
|
||
}
|
||
|
||
// AdvanceLibraryJobGuarded 只在调用方仍是该作业当前代次时写业务状态。
|
||
func (s *Store) AdvanceLibraryJobGuarded(ctx context.Context, attempt governance.LibraryAttempt,
|
||
state governance.LibraryJobState, stage, message, recovery, snapshotID, targetSnapshotID string, now time.Time) error {
|
||
finished := any(nil)
|
||
if state == governance.LibraryJobSucceeded || state == governance.LibraryJobFailed || state == governance.LibraryJobCancelled {
|
||
finished = now.UTC()
|
||
}
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_jobs SET
|
||
state=$2,
|
||
stage = CASE WHEN $3='' THEN stage ELSE $3 END,
|
||
error = CASE WHEN $4='' THEN error ELSE $4 END,
|
||
recovery = CASE WHEN $5='' THEN recovery ELSE $5 END,
|
||
snapshot_id = CASE WHEN $6='' THEN snapshot_id ELSE $6 END,
|
||
target_snapshot_id = CASE WHEN $7='' THEN target_snapshot_id ELSE $7 END,
|
||
updated_at=$8,
|
||
finished_at=COALESCE($9, finished_at)
|
||
WHERE id=$1 AND `+libraryFence("$1", 10, 11),
|
||
attempt.JobID, string(state), stage, message, recovery, snapshotID, targetSnapshotID, now.UTC(), finished,
|
||
attempt.Epoch, attempt.Owner)
|
||
if err != nil {
|
||
return fmt.Errorf("advance library job (guarded): %w", err)
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return s.fencedWriteError(ctx, attempt, "advance library job")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// UpdateLibraryStageGuarded 是阶段进度的守卫版：失租代次不能再写"我这一阶段成功了"。
|
||
func (s *Store) UpdateLibraryStageGuarded(ctx context.Context, attempt governance.LibraryAttempt,
|
||
stage, state string, total, completed, bytesTotal, bytesDone *int64, detail string, now time.Time) error {
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_job_stages SET
|
||
state = CASE WHEN $3='' THEN state ELSE $3 END,
|
||
total = COALESCE($4, total),
|
||
completed = COALESCE($5, completed),
|
||
bytes_total = COALESCE($6, bytes_total),
|
||
bytes_done = COALESCE($7, bytes_done),
|
||
detail = CASE WHEN $8='' THEN detail ELSE $8 END,
|
||
updated_at = $9
|
||
WHERE job_id=$1 AND stage=$2 AND `+libraryFence("$1", 10, 11),
|
||
attempt.JobID, stage, state, total, completed, bytesTotal, bytesDone, detail, now.UTC(),
|
||
attempt.Epoch, attempt.Owner)
|
||
if err != nil {
|
||
return fmt.Errorf("update library stage (guarded): %w", err)
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return s.fencedWriteError(ctx, attempt, "update library stage")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ReleaseLibraryMaintenanceGuarded 释放 scope 维护锁，但只认当前代次。
|
||
func (s *Store) ReleaseLibraryMaintenanceGuarded(ctx context.Context, scope, ownerJobID string,
|
||
attempt governance.LibraryAttempt, now time.Time) error {
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_maintenance SET state='released', released_at=$3, updated_at=$3
|
||
WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL
|
||
AND `+libraryFence("library_maintenance.owner_job_id", 4, 5), scope, ownerJobID, now.UTC(),
|
||
attempt.Epoch, attempt.Owner)
|
||
if err != nil {
|
||
return fmt.Errorf("release library maintenance (guarded): %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) {
|
||
return ErrMaintenanceReleased
|
||
}
|
||
return s.fencedWriteError(ctx, attempt, "release library maintenance")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ReleaseInstanceMaintenanceGuarded 释放实例维护锁，但只认当前代次。
|
||
func (s *Store) ReleaseInstanceMaintenanceGuarded(ctx context.Context, instance, ownerJobID string,
|
||
attempt governance.LibraryAttempt, now time.Time) error {
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_instance_maintenance SET state='released', released_at=$3, updated_at=$3
|
||
WHERE instance=$1 AND owner_job_id=$2 AND released_at IS NULL
|
||
AND `+libraryFence("library_instance_maintenance.owner_job_id", 4, 5), instance, ownerJobID, now.UTC(),
|
||
attempt.Epoch, attempt.Owner)
|
||
if err != nil {
|
||
return fmt.Errorf("release instance maintenance (guarded): %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
if s.releasedByThisJob(ctx, "library_instance_maintenance", "instance", instance, ownerJobID) {
|
||
return ErrMaintenanceReleased
|
||
}
|
||
return s.fencedWriteError(ctx, attempt, "release instance maintenance")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// UpdateLibraryMaintenanceStateGuarded / UpdateInstanceMaintenanceStateGuarded 同理：
|
||
// 失租代次不能把窗口改成 needs_restore / verified 去影响当前代次的判断。
|
||
func (s *Store) UpdateLibraryMaintenanceStateGuarded(ctx context.Context, scope, ownerJobID, state, note string,
|
||
attempt governance.LibraryAttempt, now time.Time) error {
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_maintenance SET state=$3, note=$4, updated_at=$5
|
||
WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL
|
||
AND `+libraryFence("library_maintenance.owner_job_id", 6, 7), scope, ownerJobID, state, note, now.UTC(),
|
||
attempt.Epoch, attempt.Owner)
|
||
if err != nil {
|
||
return fmt.Errorf("update library maintenance (guarded): %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) {
|
||
return ErrMaintenanceReleased
|
||
}
|
||
return s.fencedWriteError(ctx, attempt, "update library maintenance")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Store) UpdateInstanceMaintenanceStateGuarded(ctx context.Context, instance, ownerJobID, state, note string,
|
||
attempt governance.LibraryAttempt, now time.Time) error {
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_instance_maintenance SET state=$3, note=$4, updated_at=$5
|
||
WHERE instance=$1 AND owner_job_id=$2 AND released_at IS NULL
|
||
AND `+libraryFence("library_instance_maintenance.owner_job_id", 6, 7), instance, ownerJobID, state, note,
|
||
now.UTC(), attempt.Epoch, attempt.Owner)
|
||
if err != nil {
|
||
return fmt.Errorf("update instance maintenance (guarded): %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
if s.releasedByThisJob(ctx, "library_instance_maintenance", "instance", instance, ownerJobID) {
|
||
return ErrMaintenanceReleased
|
||
}
|
||
return s.fencedWriteError(ctx, attempt, "update instance maintenance")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// fencedWriteError 区分"作业不存在"与"代次已过期"：两者都要停下，但文案不能误导运维。
|
||
func (s *Store) fencedWriteError(ctx context.Context, attempt governance.LibraryAttempt, action string) error {
|
||
var epoch int
|
||
var owner, operation string
|
||
err := s.db.QueryRow(ctx, `SELECT attempt_epoch, lease_owner, operation_id FROM library_jobs WHERE id=$1`,
|
||
attempt.JobID).Scan(&epoch, &owner, &operation)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return fmt.Errorf("%s：作业 %s 不存在", action, attempt.JobID)
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("%s：读不到作业代次：%w", action, err)
|
||
}
|
||
var state string
|
||
if operation != "" {
|
||
_ = s.db.QueryRow(ctx, `SELECT state FROM library_attempts WHERE operation_id=$1 AND epoch=$2`,
|
||
operation, attempt.Epoch).Scan(&state)
|
||
}
|
||
return fmt.Errorf("%w（%s：本代次 %d/%s，作业行 %d/%s，代次状态 %q）", governance.ErrLibraryEpochStale,
|
||
action, attempt.Epoch, attempt.Owner, epoch, owner, state)
|
||
}
|
||
|
||
func (s *Store) SetInstanceMaintenanceIngressGuarded(ctx context.Context, instance, ownerJobID, state, note string,
|
||
attempt governance.LibraryAttempt, now time.Time) error {
|
||
if state != "closed" && state != "open" {
|
||
return errors.New("invalid ingress state")
|
||
}
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_instance_maintenance SET ingress_state=$3, note=$4, updated_at=$5
|
||
WHERE instance=$1 AND owner_job_id=$2 AND released_at IS NULL
|
||
AND `+libraryFence("library_instance_maintenance.owner_job_id", 6, 7), instance, ownerJobID, state, note,
|
||
now.UTC(), attempt.Epoch, attempt.Owner)
|
||
if err != nil {
|
||
return fmt.Errorf("update instance maintenance ingress (guarded): %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
if s.releasedByThisJob(ctx, "library_instance_maintenance", "instance", instance, ownerJobID) {
|
||
return ErrMaintenanceReleased
|
||
}
|
||
return s.fencedWriteError(ctx, attempt, "update instance maintenance ingress")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// SetLibraryMaintenanceIngressGuarded 记录入口状态也需要代次守卫：
|
||
// "入口已关"是隔离证据，失租代次写它等于伪造证据。
|
||
func (s *Store) SetLibraryMaintenanceIngressGuarded(ctx context.Context, scope, ownerJobID, state, note string,
|
||
attempt governance.LibraryAttempt, now time.Time) error {
|
||
if state != "closed" && state != "open" {
|
||
return errors.New("invalid ingress state")
|
||
}
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_maintenance SET ingress_state=$3, note=$4, updated_at=$5
|
||
WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL
|
||
AND `+libraryFence("library_maintenance.owner_job_id", 6, 7), scope, ownerJobID, state, note, now.UTC(),
|
||
attempt.Epoch, attempt.Owner)
|
||
if err != nil {
|
||
return fmt.Errorf("update library maintenance ingress (guarded): %w", err)
|
||
}
|
||
if tag.RowsAffected() == 0 {
|
||
if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) {
|
||
return ErrMaintenanceReleased
|
||
}
|
||
return s.fencedWriteError(ctx, attempt, "update library maintenance ingress")
|
||
}
|
||
return nil
|
||
}
|