姚章浩
8d838d9cb4
build / build (push) Successful in 1m45s
执行时真实时钟与持久授权回读（撤销/过期即停并保留预备份）、写集合写前落盘、 租约与 attempt 身份、恢复队列与原发送身份分离、成功收尾必须释放维护后再登记终态。
519 lines
24 KiB
Go
519 lines
24 KiB
Go
package postgres
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
|
||
"knowledge-governance/internal/governance"
|
||
)
|
||
|
||
// 发布执行身份：operation / attempt（epoch）的持久化。
|
||
//
|
||
// 一条铁律写在这里，别处不再重复实现：**谁写业务状态，就必须证明自己是当前代次**。
|
||
// epoch 守卫读 library_jobs.attempt_epoch/lease_owner，与 BeginLibraryAttempt 的激活
|
||
// 是同一把锁；失租代次的所有治理写入都会被拒绝，而不是靠调用方自觉。
|
||
|
||
// ErrLibraryAttemptHeld：同一作业还有未过期的租约。第二个执行者必须放弃，
|
||
// 不能"再认领一次"——那正是 F13 的双 claim。
|
||
var ErrLibraryAttemptHeld = errors.New("library job lease is held by another attempt")
|
||
|
||
// EnsureLibraryOperation 幂等创建/读取一次意图。
|
||
// 同一个 (scope,kind,operation_key) + 同一个 request_digest ⇒ 同一次意图（重放）；
|
||
// 摘要不一致 ⇒ 冲突，绝不把新意图塞进旧 operation。
|
||
func (s *Store) EnsureLibraryOperation(ctx context.Context, op governance.LibraryOperation) (governance.LibraryOperation, bool, error) {
|
||
if err := governance.ValidateLibraryOperation(op); err != nil {
|
||
return governance.LibraryOperation{}, false, err
|
||
}
|
||
var id string
|
||
err := s.db.QueryRow(ctx, `
|
||
INSERT INTO library_operations (id, scope, kind, target_snapshot_id, request_digest, operation_key, actor, created_at, target_id)
|
||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
|
||
ON CONFLICT (scope, kind, operation_key) DO NOTHING
|
||
RETURNING id`, op.ID, op.Scope, string(op.Kind), op.TargetSnapshotID, op.RequestDigest,
|
||
op.OperationKey, op.Actor, op.CreatedAt.UTC(), s.libraryTargetID).Scan(&id)
|
||
if err == nil {
|
||
return op, true, nil
|
||
}
|
||
if !errors.Is(err, pgx.ErrNoRows) {
|
||
return governance.LibraryOperation{}, false, fmt.Errorf("create library operation: %w", err)
|
||
}
|
||
existing, loadErr := s.libraryOperationByKey(ctx, op.Scope, op.Kind, op.OperationKey)
|
||
if loadErr != nil {
|
||
return governance.LibraryOperation{}, false, loadErr
|
||
}
|
||
if existing.RequestDigest != op.RequestDigest {
|
||
return existing, false, governance.ErrLibraryOperationConflict
|
||
}
|
||
return existing, false, nil
|
||
}
|
||
|
||
func (s *Store) libraryOperationByKey(ctx context.Context, scope string, kind governance.LibraryJobKind, key string) (governance.LibraryOperation, error) {
|
||
var id string
|
||
if err := s.db.QueryRow(ctx, `
|
||
SELECT id FROM library_operations WHERE scope=$1 AND kind=$2 AND operation_key=$3`,
|
||
scope, string(kind), key).Scan(&id); err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return governance.LibraryOperation{}, errors.New("library operation not found")
|
||
}
|
||
return governance.LibraryOperation{}, fmt.Errorf("load library operation: %w", err)
|
||
}
|
||
return s.LibraryOperation(ctx, id)
|
||
}
|
||
|
||
func (s *Store) LibraryOperation(ctx context.Context, id string) (governance.LibraryOperation, error) {
|
||
var op governance.LibraryOperation
|
||
var kind string
|
||
err := s.db.QueryRow(ctx, `
|
||
SELECT id, scope, kind, target_snapshot_id, request_digest, operation_key, actor, created_at
|
||
FROM library_operations WHERE id=$1 AND (kind='seal' OR target_id=$2)`, id, s.libraryTargetID).Scan(&op.ID, &op.Scope, &kind, &op.TargetSnapshotID,
|
||
&op.RequestDigest, &op.OperationKey, &op.Actor, &op.CreatedAt)
|
||
if err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return governance.LibraryOperation{}, err
|
||
}
|
||
return governance.LibraryOperation{}, fmt.Errorf("load library operation: %w", err)
|
||
}
|
||
op.Kind = governance.LibraryJobKind(kind)
|
||
return op, nil
|
||
}
|
||
|
||
// OperationAnchors 读不可变首次锚点。
|
||
func (s *Store) OperationAnchors(ctx context.Context, operationID string) (governance.LibraryAnchors, bool, error) {
|
||
// 锚点列允许 NULL：必须用可空类型扫描，否则空锚点会让整条读取失败
|
||
// （真库验证发现：NULL 扫进 *string 直接报错，而内存 fake 永远不会有这个形状）。
|
||
var (
|
||
snapshot sql.NullString
|
||
backupAt sql.NullTime
|
||
digest sql.NullString
|
||
fingerprint sql.NullString
|
||
recordedAt sql.NullTime
|
||
)
|
||
var bytes, rows sql.NullInt64
|
||
err := s.db.QueryRow(ctx, `
|
||
SELECT first_backup_snapshot_id, first_backup_bytes, first_backup_at, writeset_digest, writeset_rows,
|
||
target_fingerprint, anchors_at
|
||
FROM library_operations WHERE id=$1`, operationID).Scan(&snapshot, &bytes, &backupAt, &digest, &rows,
|
||
&fingerprint, &recordedAt)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return governance.LibraryAnchors{}, false, nil
|
||
}
|
||
if err != nil {
|
||
return governance.LibraryAnchors{}, false, fmt.Errorf("load library operation anchors: %w", err)
|
||
}
|
||
a := scanAnchors(snapshot, bytes, backupAt, digest, rows, fingerprint, recordedAt)
|
||
return a, a.RecordedAt != nil, nil
|
||
}
|
||
|
||
// scanAnchors 是锚点唯一的解码实现：NULL 与零值必须分清（未知 ≠ 0）。
|
||
func scanAnchors(snapshot sql.NullString, bytes sql.NullInt64, backupAt sql.NullTime, digest sql.NullString,
|
||
rows sql.NullInt64, fingerprint sql.NullString, recordedAt sql.NullTime) governance.LibraryAnchors {
|
||
a := governance.LibraryAnchors{FirstBackupSnapshotID: snapshot.String, WriteSetDigest: digest.String,
|
||
TargetFingerprint: fingerprint.String}
|
||
if bytes.Valid {
|
||
value := bytes.Int64
|
||
a.FirstBackupBytes = &value
|
||
}
|
||
if rows.Valid {
|
||
value := rows.Int64
|
||
a.WriteSetRows = &value
|
||
}
|
||
if backupAt.Valid {
|
||
value := backupAt.Time
|
||
a.FirstBackupAt = &value
|
||
}
|
||
if recordedAt.Valid {
|
||
value := recordedAt.Time
|
||
a.RecordedAt = &value
|
||
}
|
||
return a
|
||
}
|
||
|
||
// RecordOperationAnchors 只**填空**，不改已存在的锚点。
|
||
//
|
||
// 读-改-写必须在同一个事务里并锁住 operation 行（R2 附带项）：两个执行者同时"首次"
|
||
// 登记时，先读后写的实现会各自读到 NULL、各自写入，最后一个静默覆盖前一个——
|
||
// 首次锚点就不再是首次。这里 FOR UPDATE 串行化，写入后**逐字段回读核对**：
|
||
// 任何一个不可变字段与本次声称的值不一致就报错（而不是让调用方以为登记成功）。
|
||
func (s *Store) RecordOperationAnchors(ctx context.Context, operationID string, anchors governance.LibraryAnchors, now time.Time) error {
|
||
if operationID == "" {
|
||
return errors.New("record operation anchors requires an operation")
|
||
}
|
||
tx, err := s.db.Begin(ctx)
|
||
if err != nil {
|
||
return fmt.Errorf("begin operation anchors: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
var (
|
||
exSnapshot, exDigest, exFingerprint sql.NullString
|
||
exBytes, exRows sql.NullInt64
|
||
exBackupAt, exRecordedAt sql.NullTime
|
||
)
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT first_backup_snapshot_id, first_backup_bytes, first_backup_at, writeset_digest, writeset_rows,
|
||
target_fingerprint, anchors_at
|
||
FROM library_operations WHERE id=$1 FOR UPDATE`, operationID).Scan(&exSnapshot, &exBytes, &exBackupAt,
|
||
&exDigest, &exRows, &exFingerprint, &exRecordedAt)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return fmt.Errorf("operation %s 不存在：拒绝登记首次锚点", operationID)
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("load library operation anchors: %w", err)
|
||
}
|
||
existing := scanAnchors(exSnapshot, exBytes, exBackupAt, exDigest, exRows, exFingerprint, exRecordedAt)
|
||
// 已存在且不同 ⇒ 有人想把"首次"换掉（把半写状态当新基线），必须报错。
|
||
if err := compareAnchors(operationID, existing, anchors); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE library_operations SET
|
||
first_backup_snapshot_id = COALESCE(first_backup_snapshot_id, $2),
|
||
first_backup_bytes = COALESCE(first_backup_bytes, $3),
|
||
first_backup_at = COALESCE(first_backup_at, $4),
|
||
writeset_digest = COALESCE(writeset_digest, $5),
|
||
writeset_rows = COALESCE(writeset_rows, $6),
|
||
target_fingerprint = COALESCE(target_fingerprint, $7),
|
||
anchors_at = COALESCE(anchors_at, $8)
|
||
WHERE id=$1`, operationID, nullIfEmpty(anchors.FirstBackupSnapshotID), anchors.FirstBackupBytes,
|
||
anchors.FirstBackupAt, nullIfEmpty(anchors.WriteSetDigest), anchors.WriteSetRows,
|
||
nullIfEmpty(anchors.TargetFingerprint), now.UTC()); err != nil {
|
||
return fmt.Errorf("record library operation anchors: %w", err)
|
||
}
|
||
// 回读核对：NULL 与空串、nil 与 0 必须分清（真实 PG 才有这个差别）。
|
||
var (
|
||
afSnapshot, afDigest, afFingerprint sql.NullString
|
||
afBytes, afRows sql.NullInt64
|
||
afBackupAt, afRecordedAt sql.NullTime
|
||
)
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT first_backup_snapshot_id, first_backup_bytes, first_backup_at, writeset_digest, writeset_rows,
|
||
target_fingerprint, anchors_at
|
||
FROM library_operations WHERE id=$1`, operationID).Scan(&afSnapshot, &afBytes, &afBackupAt, &afDigest,
|
||
&afRows, &afFingerprint, &afRecordedAt); err != nil {
|
||
return fmt.Errorf("read back library operation anchors: %w", err)
|
||
}
|
||
after := scanAnchors(afSnapshot, afBytes, afBackupAt, afDigest, afRows, afFingerprint, afRecordedAt)
|
||
if err := verifyAnchorsPersisted(operationID, anchors, after); err != nil {
|
||
return err
|
||
}
|
||
return tx.Commit(ctx)
|
||
}
|
||
|
||
// compareAnchors 拒绝"用新值覆盖已登记的首次锚点"。
|
||
func compareAnchors(operationID string, existing, incoming governance.LibraryAnchors) error {
|
||
if existing.FirstBackupSnapshotID != "" && incoming.FirstBackupSnapshotID != "" &&
|
||
existing.FirstBackupSnapshotID != incoming.FirstBackupSnapshotID {
|
||
return fmt.Errorf("operation %s 的首次预备份是 %s，不能用 %s 覆盖",
|
||
operationID, existing.FirstBackupSnapshotID, incoming.FirstBackupSnapshotID)
|
||
}
|
||
if existing.WriteSetDigest != "" && incoming.WriteSetDigest != "" &&
|
||
existing.WriteSetDigest != incoming.WriteSetDigest {
|
||
return fmt.Errorf("operation %s 的首次写集合是 %s，不能用 %s 覆盖",
|
||
operationID, existing.WriteSetDigest, incoming.WriteSetDigest)
|
||
}
|
||
if existing.FirstBackupBytes != nil && incoming.FirstBackupBytes != nil &&
|
||
*existing.FirstBackupBytes != *incoming.FirstBackupBytes {
|
||
return fmt.Errorf("operation %s 的首次预备份字节是 %d，不能用 %d 覆盖",
|
||
operationID, *existing.FirstBackupBytes, *incoming.FirstBackupBytes)
|
||
}
|
||
if existing.TargetFingerprint != "" && incoming.TargetFingerprint != "" &&
|
||
existing.TargetFingerprint != incoming.TargetFingerprint {
|
||
return fmt.Errorf("operation %s 的目标指纹是 %s，不能用 %s 覆盖",
|
||
operationID, existing.TargetFingerprint, incoming.TargetFingerprint)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// verifyAnchorsPersisted 核对"本次声称的值确实落库了"：空值必须仍是 NULL，
|
||
// 有值必须逐字段相同。避免"写成功但被 NULL/0 混淆"的静默丢锚点。
|
||
func verifyAnchorsPersisted(operationID string, claimed, persisted governance.LibraryAnchors) error {
|
||
check := func(name, want, got string) error {
|
||
if want != "" && want != got {
|
||
return fmt.Errorf("operation %s 的 %s 回读为 %q，与声称的 %q 不一致", operationID, name, got, want)
|
||
}
|
||
return nil
|
||
}
|
||
if err := check("first_backup_snapshot_id", claimed.FirstBackupSnapshotID, persisted.FirstBackupSnapshotID); err != nil {
|
||
return err
|
||
}
|
||
if err := check("writeset_digest", claimed.WriteSetDigest, persisted.WriteSetDigest); err != nil {
|
||
return err
|
||
}
|
||
if err := check("target_fingerprint", claimed.TargetFingerprint, persisted.TargetFingerprint); err != nil {
|
||
return err
|
||
}
|
||
if claimed.FirstBackupBytes != nil &&
|
||
(persisted.FirstBackupBytes == nil || *persisted.FirstBackupBytes != *claimed.FirstBackupBytes) {
|
||
return fmt.Errorf("operation %s 的 first_backup_bytes 没有按声称值落库", operationID)
|
||
}
|
||
if claimed.WriteSetRows != nil &&
|
||
(persisted.WriteSetRows == nil || *persisted.WriteSetRows != *claimed.WriteSetRows) {
|
||
return fmt.Errorf("operation %s 的 writeset_rows 没有按声称值落库", operationID)
|
||
}
|
||
if persisted.RecordedAt == nil {
|
||
return fmt.Errorf("operation %s 的 anchors_at 仍为空：锚点没被真正登记", operationID)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// BeginLibraryAttempt 取得这个作业的执行权并激活一个新代次。
|
||
//
|
||
// 顺序与理由（R2）：
|
||
// 1. 锁作业行、再锁 **operation 行**：epoch 是 operation 维度的单调量，两个作业同时
|
||
// 接管同一个 operation 时，只有锁 operation 才能保证"下一个 epoch"不重号。
|
||
// 2. 同一次投递重复调用 ⇒ **幂等**：返回同一个活跃代次并续租，不新增谱系。
|
||
// 3. 有别的活跃代次：
|
||
// - 它的租约仍未过期 ⇒ 拒绝（ErrLibraryAttemptHeld）—— 不能把还在跑的代次踢掉；
|
||
// - 租约已过期（或就是本作业的旧代次）⇒ 结束它（abandoned，保留行），再开新代次。
|
||
// 4. 新代次 = max(epoch)+1，**新增一行**（历史不可变）；同时把同 operation 其它作业行
|
||
// 的租约清空，让它们的守卫立刻失效（旧 job 行上的 epoch/owner 字段还会留着，
|
||
// 所以守卫必须同时看 attempts 的活跃状态，见 libraryFence）。
|
||
func (s *Store) BeginLibraryAttempt(ctx context.Context, jobID, owner string, lease time.Duration, now time.Time) (governance.LibraryAttempt, error) {
|
||
if jobID == "" || owner == "" {
|
||
return governance.LibraryAttempt{}, errors.New("begin library attempt requires job and owner")
|
||
}
|
||
if lease <= 0 {
|
||
lease = 2 * time.Minute
|
||
}
|
||
tx, err := s.db.Begin(ctx)
|
||
if err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("begin library attempt: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
|
||
var operationID, leaseOwner, queueJobID string
|
||
var jobEpoch int
|
||
var jobLeaseUntil *time.Time
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT operation_id, attempt_epoch, lease_owner, lease_until, queue_job_id
|
||
FROM library_jobs WHERE id=$1 FOR UPDATE`, jobID).Scan(&operationID, &jobEpoch, &leaseOwner, &jobLeaseUntil, &queueJobID)
|
||
if err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("library job %s not found", jobID)
|
||
}
|
||
return governance.LibraryAttempt{}, fmt.Errorf("load library job for attempt: %w", err)
|
||
}
|
||
until := now.Add(lease).UTC()
|
||
|
||
// 旧作业（迁移前没有 operation）：没有代次谱系可建，只做租约互斥。
|
||
if operationID == "" {
|
||
if leaseOwner != "" && leaseOwner != owner && jobLeaseUntil != nil && jobLeaseUntil.After(now.UTC()) {
|
||
return governance.LibraryAttempt{}, ErrLibraryAttemptHeld
|
||
}
|
||
next := jobEpoch + 1
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE library_jobs SET attempt_epoch=$2, lease_owner=$3, lease_until=$4, updated_at=$5 WHERE id=$1`,
|
||
jobID, next, owner, until, now.UTC()); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("activate legacy library attempt: %w", err)
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("commit legacy library attempt: %w", err)
|
||
}
|
||
return governance.LibraryAttempt{JobID: jobID, Epoch: next, QueueJobID: queueJobID, Owner: owner,
|
||
State: governance.LibraryAttemptRunning, StartedAt: now.UTC(), LeaseUntil: &until}, nil
|
||
}
|
||
|
||
// epoch 分配必须串行到 operation 粒度。
|
||
if _, err := tx.Exec(ctx, `SELECT id FROM library_operations WHERE id=$1 FOR UPDATE`, operationID); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("lock library operation for attempt: %w", err)
|
||
}
|
||
|
||
// 当前活跃代次（部分唯一索引保证至多一行）。
|
||
var activeEpoch int
|
||
var activeJob, activeOwner, activeState string
|
||
var activeUntil *time.Time
|
||
activeErr := tx.QueryRow(ctx, `
|
||
SELECT epoch, job_id, owner, state, lease_until FROM library_attempts
|
||
WHERE operation_id=$1 AND state IN ('claimed','running')`, operationID).Scan(
|
||
&activeEpoch, &activeJob, &activeOwner, &activeState, &activeUntil)
|
||
switch {
|
||
case activeErr != nil && !errors.Is(activeErr, pgx.ErrNoRows):
|
||
return governance.LibraryAttempt{}, fmt.Errorf("load active library attempt: %w", activeErr)
|
||
case activeErr == nil && activeOwner == owner && activeJob == jobID:
|
||
// 同一次投递重复调用：续租，返回同一代次（不新增谱系）。
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE library_attempts SET heartbeat_at=$4, lease_until=$5 WHERE operation_id=$1 AND epoch=$2 AND owner=$3`,
|
||
operationID, activeEpoch, owner, now.UTC(), until); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("renew library attempt: %w", err)
|
||
}
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE library_jobs SET attempt_epoch=$2, lease_owner=$3, lease_until=$4, updated_at=$5 WHERE id=$1`,
|
||
jobID, activeEpoch, owner, until, now.UTC()); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("renew library job lease: %w", err)
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("commit library attempt: %w", err)
|
||
}
|
||
return governance.LibraryAttempt{OperationID: operationID, Epoch: activeEpoch, JobID: jobID,
|
||
QueueJobID: queueJobID, Owner: owner, State: governance.LibraryAttemptRunning,
|
||
StartedAt: now.UTC(), LeaseUntil: &until}, nil
|
||
case activeErr == nil && activeUntil != nil && activeUntil.After(now.UTC()):
|
||
// 别人（另一个作业或另一个执行者）仍持有未过期租约：让路，不许抢占。
|
||
return governance.LibraryAttempt{}, fmt.Errorf("%w（operation %s 的 %d 代次由 %s/%s 持有）",
|
||
ErrLibraryAttemptHeld, operationID, activeEpoch, activeJob, activeOwner)
|
||
case activeErr == nil:
|
||
// 租约已过期：结束它（行保留），再开新代次。
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE library_attempts SET state='abandoned', finished_at=$3, outcome=$4, lease_until=NULL
|
||
WHERE operation_id=$1 AND epoch=$2`, operationID, activeEpoch, now.UTC(),
|
||
"lease expired: superseded by a new attempt"); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("abandon stale library attempt: %w", err)
|
||
}
|
||
}
|
||
|
||
var maxEpoch *int
|
||
if err := tx.QueryRow(ctx, `SELECT MAX(epoch) FROM library_attempts WHERE operation_id=$1`,
|
||
operationID).Scan(&maxEpoch); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("load library attempt epoch: %w", err)
|
||
}
|
||
next := 1
|
||
if maxEpoch != nil && *maxEpoch >= next {
|
||
next = *maxEpoch + 1
|
||
}
|
||
attempt := governance.LibraryAttempt{
|
||
OperationID: operationID, Epoch: next, JobID: jobID, QueueJobID: queueJobID,
|
||
Owner: owner, State: governance.LibraryAttemptRunning, StartedAt: now.UTC(), LeaseUntil: &until,
|
||
}
|
||
if _, err := tx.Exec(ctx, `
|
||
INSERT INTO library_attempts (operation_id, epoch, job_id, queue_job_id, owner, state, started_at, heartbeat_at, lease_until)
|
||
VALUES ($1,$2,$3,$4,$5,'running',$6,$6,$7)`,
|
||
operationID, next, jobID, queueJobID, owner, now.UTC(), until); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("record library attempt: %w", err)
|
||
}
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE library_jobs SET attempt_epoch=$2, lease_owner=$3, lease_until=$4, updated_at=$5 WHERE id=$1`,
|
||
jobID, next, owner, until, now.UTC()); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("activate library attempt: %w", err)
|
||
}
|
||
// 同一 operation 的其它作业行（续作留下/旧作业）立刻失去租约：
|
||
// 它们的守卫写从这一刻起失败，即使行上的 epoch/owner 字段还没被改写。
|
||
if _, err := tx.Exec(ctx, `
|
||
UPDATE library_jobs SET lease_owner='', lease_until=NULL, updated_at=$3
|
||
WHERE operation_id=$1 AND id<>$2`, operationID, jobID, now.UTC()); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("revoke sibling library job leases: %w", err)
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return governance.LibraryAttempt{}, fmt.Errorf("commit library attempt: %w", err)
|
||
}
|
||
return attempt, nil
|
||
}
|
||
|
||
// FinishLibraryAttempt 落本次代次的终态。旧代次不能结束别人的代次：
|
||
// 只有 (operation, epoch) 与 owner 都匹配才写。
|
||
func (s *Store) FinishLibraryAttempt(ctx context.Context, attempt governance.LibraryAttempt, state governance.LibraryAttemptState, outcome string, now time.Time) error {
|
||
if attempt.OperationID == "" {
|
||
return nil // 旧作业没有 operation：没有代次可结
|
||
}
|
||
// 只允许结束**还没结束**的那一代：已经 finished/abandoned 的代次是历史事实，
|
||
// 迟到的 finish 不能把它的 outcome 改写成"成功"（否则接管后旧执行者一收尾，
|
||
// 历史就变成了对自己有利的版本，排障时无从分辨谁真的跑完）。
|
||
// 终态集合与服务端的 state CHECK 一致；`completed` 只允许从活跃态进来一次。
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_attempts SET state=$4, finished_at=$5, outcome=$6, lease_until=NULL
|
||
WHERE operation_id=$1 AND epoch=$2 AND owner=$3 AND state IN ('claimed','running')`,
|
||
attempt.OperationID, attempt.Epoch, attempt.Owner, string(state), now.UTC(), outcome)
|
||
if err != nil {
|
||
return fmt.Errorf("finish library attempt: %w", err)
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return governance.ErrLibraryEpochStale
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// LibraryAttempts 返回一次 operation 的执行谱系（按代次升序）：接管/恢复排查读它。
|
||
func (s *Store) LibraryAttempts(ctx context.Context, operationID string) ([]governance.LibraryAttempt, error) {
|
||
rows, err := s.db.Query(ctx, `
|
||
SELECT operation_id, epoch, job_id, queue_job_id, owner, state, started_at, heartbeat_at, lease_until, finished_at, outcome
|
||
FROM library_attempts WHERE operation_id=$1 ORDER BY epoch`, operationID)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("list library attempts: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
items := make([]governance.LibraryAttempt, 0, 4)
|
||
for rows.Next() {
|
||
var item governance.LibraryAttempt
|
||
var state string
|
||
if err := rows.Scan(&item.OperationID, &item.Epoch, &item.JobID, &item.QueueJobID, &item.Owner, &state,
|
||
&item.StartedAt, &item.HeartbeatAt, &item.LeaseUntil, &item.FinishedAt, &item.Outcome); err != nil {
|
||
return nil, fmt.Errorf("scan library attempt: %w", err)
|
||
}
|
||
item.State = governance.LibraryAttemptState(state)
|
||
items = append(items, item)
|
||
}
|
||
return items, rows.Err()
|
||
}
|
||
|
||
// LatestLibraryAttempt 取当前代次（无则 false）。
|
||
func (s *Store) LatestLibraryAttempt(ctx context.Context, operationID string) (governance.LibraryAttempt, bool, error) {
|
||
items, err := s.LibraryAttempts(ctx, operationID)
|
||
if err != nil || len(items) == 0 {
|
||
return governance.LibraryAttempt{}, false, err
|
||
}
|
||
return items[len(items)-1], true, nil
|
||
}
|
||
|
||
// RenewLibraryAttempt 续租：失租（0 行）即返回 ErrLibraryEpochStale，调用方必须停手。
|
||
func (s *Store) RenewLibraryAttempt(ctx context.Context, attempt governance.LibraryAttempt, lease time.Duration, now time.Time) error {
|
||
if lease <= 0 {
|
||
return errors.New("positive lease required")
|
||
}
|
||
tx, err := s.db.Begin(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
||
// Lock job first, exactly like BeginLibraryAttempt. Both leases are one fact;
|
||
// never resurrect an expired owner or commit half of a renewal.
|
||
tag, err := tx.Exec(ctx, `UPDATE library_jobs SET lease_until=clock_timestamp()+$4::interval
|
||
WHERE id=$1 AND attempt_epoch=$2 AND lease_owner=$3 AND lease_until>clock_timestamp()`, attempt.JobID, attempt.Epoch, attempt.Owner, lease.String())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return governance.ErrLibraryEpochStale
|
||
}
|
||
if attempt.OperationID != "" {
|
||
tag, err = tx.Exec(ctx, `UPDATE library_attempts SET heartbeat_at=clock_timestamp(),lease_until=clock_timestamp()+$4::interval
|
||
WHERE operation_id=$1 AND epoch=$2 AND owner=$3 AND state='running' AND lease_until>clock_timestamp()`, attempt.OperationID, attempt.Epoch, attempt.Owner, lease.String())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return governance.ErrLibraryEpochStale
|
||
}
|
||
}
|
||
return tx.Commit(ctx)
|
||
}
|
||
|
||
// guardLibraryJobEpoch 在事务内核"这次写入来自当前代次"。守卫失败一律返回
|
||
// ErrLibraryEpochStale，调用方不许把它当成"重试一下就好"。
|
||
func guardLibraryJobEpoch(ctx context.Context, tx pgx.Tx, jobID string, epoch int, owner string) error {
|
||
var currentEpoch int
|
||
var currentOwner string
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT attempt_epoch, lease_owner FROM library_jobs WHERE id=$1 FOR UPDATE`, jobID).Scan(¤tEpoch, ¤tOwner)
|
||
if err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return fmt.Errorf("library job %s not found", jobID)
|
||
}
|
||
return fmt.Errorf("load library job for epoch guard: %w", err)
|
||
}
|
||
if currentEpoch != epoch || (owner != "" && currentOwner != owner) {
|
||
return fmt.Errorf("%w（作业 %s：本代次 %d/%s，当前 %d/%s）", governance.ErrLibraryEpochStale,
|
||
jobID, epoch, owner, currentEpoch, currentOwner)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func nullIfEmpty(value string) any {
|
||
if value == "" {
|
||
return nil
|
||
}
|
||
return value
|
||
}
|