姚章浩
6209072e43
原实现把 ProtocolEpoch 相等当作构建兼容证明：任何 40 位小写 hex SHA 加一个非空 display 都会被记成 compatible_upgrade 并允许激活，换一个 SHA 即可绕开同 SHA 的 display 绑定，构建身份仍是声明式。 收口为精确绑定运行完整 SHA：同 SHA 必须 display 相符，不同 SHA 一律失败关闭， 错误明确写"cross-sha compatible activation is disabled without a verifiable protocol build membership proof"。不新建签名基础设施，也不引入第二份 allowlist； 使用点与激活点调用同一函数，行为一致。台账不重写、旧激活行保留（不匹配即 held）。
622 lines
30 KiB
Go
622 lines
30 KiB
Go
package postgres
|
||
|
||
import (
|
||
"context"
|
||
"crypto/rand"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
|
||
"knowledge-governance/internal/governance"
|
||
"knowledge-governance/internal/libraryrelease"
|
||
)
|
||
|
||
// DS01：恢复标记派发的持久准入。
|
||
//
|
||
// 关键点不是"多了一张表"，而是发送许可的产生方式：
|
||
// - Prepare 只建立 prepared（不返回许可）；
|
||
// - Acquire 在**独立短事务**里按 job→operation→activation→active attempt→effect
|
||
// 的顺序取锁，校验当前 epoch/owner/lease（数据库时钟）、原作业/operation 锚点、
|
||
// 维护窗口、恢复投递身份与不可变绑定后做 prepared→dispatching 的 CAS，只有
|
||
// COMMIT 明确成功才返回一次性许可；
|
||
// - Inspect 纯读，永不返回许可；
|
||
// - 表本身拒绝状态回退、绑定变化、删行与激活撤销回退（迁移里的触发器）。
|
||
//
|
||
// 外层 WithLibraryExecution 的事务只 rollback、从不 commit，所以这里绝不能用它：
|
||
// 在那种事务里写 dispatching，发出请求后账会被回滚成"未派发"。
|
||
|
||
var ErrRestoreDispatchNotActivated = libraryrelease.ErrRestoreDispatchUnavailable
|
||
|
||
// 台账行的 protocol_version 仍是 1（派发协议身份，迁移 78 的 CHECK 约束）；证据文档
|
||
// 升到 v2 只改变"激活输入要证明什么"，不改表结构。两者不是同一个数字。
|
||
const restoreDispatchActivationLedgerVersion = 1
|
||
|
||
// ActivateRestoreDispatch 消费一份**可核验的**激活证据文件（协议 v2）。
|
||
//
|
||
// 证据只声明目标身份、维护作业、journal 文件名/摘要与 writer 集合；实际事实由
|
||
// verifyRestoreDispatchEvidence 直接读维护 journal、治理库维护行与现场构建/指纹
|
||
// 核出（见 library_restore_dispatch_activation.go）。没有配置受信 journal 根时
|
||
// 返回 ErrRestoreDispatchActivationDisabled，绝不退回"只验字符串就开门"。
|
||
func (s *Store) ActivateRestoreDispatch(ctx context.Context, evidencePath, evidenceDigest,
|
||
targetIdentity, targetFingerprint string, now time.Time) error {
|
||
evidence, err := libraryrelease.LoadRestoreDispatchActivationEvidence(evidencePath, evidenceDigest,
|
||
now.UTC(), libraryrelease.RestoreDispatchActivationMaxAge)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if evidence.TargetIdentity != targetIdentity {
|
||
return fmt.Errorf("restore dispatch activation evidence is for target %q, not %q", evidence.TargetIdentity, targetIdentity)
|
||
}
|
||
verified, err := s.verifyRestoreDispatchEvidence(ctx, evidence, targetFingerprint, now.UTC())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
record, err := json.Marshal(restoreDispatchActivationRecord{Request: evidence, Verified: verified})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if _, err := s.db.Exec(ctx, `
|
||
INSERT INTO library_index_restore_dispatch_activation
|
||
(target_identity, protocol_version, target_fingerprint, target_instance, governance_database,
|
||
protocol_build, activated_by, drain_evidence_digest, evidence, attested_at, activated_at)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10, $11)
|
||
ON CONFLICT (target_identity) DO NOTHING`,
|
||
evidence.TargetIdentity, restoreDispatchActivationLedgerVersion, evidence.TargetFingerprint, evidence.TargetInstance,
|
||
evidence.GovernanceDatabase, evidence.ProtocolBuild, evidence.AttestedBy,
|
||
verified.MaintenanceJournalSHA256, string(record), evidence.AttestedAt.UTC(), now.UTC()); err != nil {
|
||
return fmt.Errorf("activate restore dispatch: %w", err)
|
||
}
|
||
return s.requireRestoreDispatchActivation(ctx, targetIdentity)
|
||
}
|
||
|
||
func (s *Store) RevokeRestoreDispatch(ctx context.Context, targetIdentity string, now time.Time) error {
|
||
tag, err := s.db.Exec(ctx, `
|
||
UPDATE library_index_restore_dispatch_activation SET revoked_at=$2
|
||
WHERE target_identity=$1 AND revoked_at IS NULL`, targetIdentity, now.UTC())
|
||
if err != nil {
|
||
return fmt.Errorf("revoke restore dispatch: %w", err)
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return errors.New("restore dispatch activation is not active for this target")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Store) requireRestoreDispatchActivation(ctx context.Context, targetIdentity string) error {
|
||
var one int
|
||
err := s.db.QueryRow(ctx, `
|
||
SELECT 1 FROM library_index_restore_dispatch_activation
|
||
WHERE target_identity=$1 AND protocol_version=1 AND revoked_at IS NULL`, targetIdentity).Scan(&one)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return fmt.Errorf("%w（目标 %s 没有激活登记）", ErrRestoreDispatchNotActivated, targetIdentity)
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("read restore dispatch activation: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// lockRestoreDispatchActivation 在**同一个 CAS 事务**里锁激活行。撤销是同一行的
|
||
// UPDATE：撤销先完成 ⇒ 这里看到 revoked_at；CAS 先取锁 ⇒ 撤销必须等 CAS 提交，
|
||
// 之后按"已派发"的窄边界处理。两者只能有一个线性化顺序。
|
||
//
|
||
// 这里同时是**使用点**的权威域/构建身份重核：激活行里的 verified 事实必须与当前
|
||
// 连接的 PG 权威域（集群 system_identifier + 库 OID + primary）一致；运行构建必须
|
||
// 有完整 SHA，且与激活登记的 SHA 精确相同（跨 SHA 兼容默认关闭，epoch 相等不算
|
||
// 证明）；恢复目标指纹必须等于本次观察值。同名库、副本、
|
||
// unknown 构建或换目标都在这里失败关闭，而不是只在激活那一刻检查。
|
||
func lockRestoreDispatchActivation(ctx context.Context, tx pgx.Tx, targetIdentity, targetFingerprint string) error {
|
||
var revoked *time.Time
|
||
var evidenceRaw, rowFingerprint string
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT revoked_at, evidence::text, target_fingerprint FROM library_index_restore_dispatch_activation
|
||
WHERE target_identity=$1 AND protocol_version=1 FOR UPDATE`, targetIdentity).Scan(&revoked, &evidenceRaw, &rowFingerprint)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return fmt.Errorf("%w（目标 %s 没有激活登记）", ErrRestoreDispatchNotActivated, targetIdentity)
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("lock restore dispatch activation: %w", err)
|
||
}
|
||
if revoked != nil {
|
||
return fmt.Errorf("%w（目标 %s 的激活已于 %s 撤销）", ErrRestoreDispatchNotActivated, targetIdentity, revoked.UTC().Format(time.RFC3339))
|
||
}
|
||
var record restoreDispatchActivationRecord
|
||
if err := json.Unmarshal([]byte(evidenceRaw), &record); err != nil {
|
||
return fmt.Errorf("parse restore dispatch activation record: %w", err)
|
||
}
|
||
verified := record.Verified
|
||
// 激活行的 target_fingerprint 列是**维护窗口**指纹；binding 带的是本次恢复对
|
||
// 目标做的只读观察值。两层不能相互顶替，否则正常恢复会被窗口指纹误拒。
|
||
if rowFingerprint != verified.TargetFingerprint {
|
||
return fmt.Errorf("%w（目标 %s 的激活行指纹 %s 与登记证据 %s 不一致）",
|
||
ErrRestoreDispatchNotActivated, targetIdentity, rowFingerprint, verified.TargetFingerprint)
|
||
}
|
||
if verified.RestoreTargetFingerprint != targetFingerprint {
|
||
return fmt.Errorf("%w（目标 %s 的激活登记恢复目标指纹 %s，本次恢复目标指纹 %s）",
|
||
ErrRestoreDispatchNotActivated, targetIdentity, verified.RestoreTargetFingerprint, targetFingerprint)
|
||
}
|
||
authority, primary, _, err := observedGovernanceAuthority(ctx, tx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if verified.GovernanceAuthority.SystemIdentifier == "" || verified.GovernanceAuthority.DatabaseOID == "" {
|
||
return fmt.Errorf("%w（目标 %s 的激活登记没有治理权威域身份：拒绝仅凭库名启用）",
|
||
ErrRestoreDispatchNotActivated, targetIdentity)
|
||
}
|
||
if authority != verified.GovernanceAuthority {
|
||
return fmt.Errorf("%w（目标 %s 的激活登记属于治理域 %s/%s，当前是 %s/%s）",
|
||
ErrRestoreDispatchNotActivated, targetIdentity, verified.GovernanceAuthority.SystemIdentifier,
|
||
verified.GovernanceAuthority.DatabaseOID, authority.SystemIdentifier, authority.DatabaseOID)
|
||
}
|
||
if !primary {
|
||
return fmt.Errorf("%w（治理库当前处于恢复/副本角色，拒绝授予派发许可）", ErrRestoreDispatchNotActivated)
|
||
}
|
||
buildEvidence := libraryrelease.RestoreDispatchActivationEvidence{
|
||
ProtocolEpoch: verified.ProtocolEpoch,
|
||
ProtocolBuild: verified.ProtocolBuild,
|
||
ProtocolBuildSHA: verified.ProtocolBuildSHA,
|
||
}
|
||
if _, _, _, err := verifyRestoreDispatchBuild(buildEvidence); err != nil {
|
||
return fmt.Errorf("%w（目标 %s：%v）", ErrRestoreDispatchNotActivated, targetIdentity, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func restoreDispatchBindingDigest(b libraryrelease.RestoreMarkerBinding) (string, error) {
|
||
return b.Digest()
|
||
}
|
||
|
||
func (s *Store) InspectRestoreMarker(ctx context.Context, b libraryrelease.RestoreMarkerBinding) (libraryrelease.RestoreDispatchStatus, bool, error) {
|
||
digest, err := restoreDispatchBindingDigest(b)
|
||
if err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, false, err
|
||
}
|
||
var state, stored string
|
||
err = s.db.QueryRow(ctx, `
|
||
SELECT state, binding_digest FROM library_index_restore_dispatch
|
||
WHERE target_identity=$1 AND operation_id=$2 AND window_job_id=$3 AND plan_digest=$4
|
||
AND index_name=$5 AND document_id=$6 AND phase=$7`,
|
||
b.Key.TargetIdentity, b.Key.OperationID, b.Key.WindowJobID, b.Key.PlanDigest,
|
||
b.Key.Index, b.Key.DocumentID, b.Key.Phase).Scan(&state, &stored)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return libraryrelease.RestoreDispatchStatus{}, false, nil
|
||
}
|
||
if err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, false, fmt.Errorf("inspect restore dispatch: %w", err)
|
||
}
|
||
if stored != digest {
|
||
return libraryrelease.RestoreDispatchStatus{State: state, Held: true, BindingConflict: true,
|
||
Reason: "restore dispatch binding differs from the immutable plan"}, true, nil
|
||
}
|
||
switch state {
|
||
case libraryrelease.RestoreDispatchPrepared:
|
||
return libraryrelease.RestoreDispatchStatus{State: state}, true, nil
|
||
case libraryrelease.RestoreDispatchDispatching:
|
||
return libraryrelease.RestoreDispatchStatus{State: state, Held: true,
|
||
Reason: "dispatch outcome unknown: no further create is permitted"}, true, nil
|
||
case libraryrelease.RestoreDispatchMarkerConfirmed:
|
||
return libraryrelease.RestoreDispatchStatus{State: state, Held: true,
|
||
Reason: "marker already dispatched; only the original confirmation path may continue"}, true, nil
|
||
default:
|
||
return libraryrelease.RestoreDispatchStatus{State: state, Held: true,
|
||
Reason: "unrecognised dispatch state"}, true, nil
|
||
}
|
||
}
|
||
|
||
// UnresolvedRestoreDispatches 返回原 window 的权威台账全集（跨 plan 摘要）。
|
||
// 前置与收尾都查同一份记录：只遍历当前 plan 的文档列表会漏掉计划外/别名 effect。
|
||
func (s *Store) UnresolvedRestoreDispatches(ctx context.Context, targetIdentity, operationID, windowJobID string) ([]libraryrelease.RestoreDispatchRecord, error) {
|
||
rows, err := s.db.Query(ctx, `
|
||
SELECT plan_digest, index_name, document_id, state FROM library_index_restore_dispatch
|
||
WHERE target_identity=$1 AND operation_id=$2 AND window_job_id=$3
|
||
ORDER BY plan_digest, index_name, document_id`, targetIdentity, operationID, windowJobID)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("read restore dispatch window: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
var records []libraryrelease.RestoreDispatchRecord
|
||
for rows.Next() {
|
||
var record libraryrelease.RestoreDispatchRecord
|
||
if err := rows.Scan(&record.PlanDigest, &record.Index, &record.DocumentID, &record.State); err != nil {
|
||
return nil, err
|
||
}
|
||
records = append(records, record)
|
||
}
|
||
return records, rows.Err()
|
||
}
|
||
|
||
// rejectRestoreDispatchPlanAlias 拒绝同一逻辑 effect 用另一个 plan 摘要另起一行。
|
||
// key 里虽然有 plan_digest，但唯一约束允许"同 target/operation/window/index/doc/phase、
|
||
// 不同 plan"共存；那正是"换计划别名重开证明"的旁路。
|
||
func rejectRestoreDispatchPlanAlias(ctx context.Context, tx pgx.Tx, b libraryrelease.RestoreMarkerBinding) error {
|
||
var other string
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT plan_digest FROM library_index_restore_dispatch
|
||
WHERE target_identity=$1 AND operation_id=$2 AND window_job_id=$3
|
||
AND index_name=$4 AND document_id=$5 AND phase=$6 AND plan_digest<>$7
|
||
LIMIT 1 FOR UPDATE`,
|
||
b.Key.TargetIdentity, b.Key.OperationID, b.Key.WindowJobID,
|
||
b.Key.Index, b.Key.DocumentID, b.Key.Phase, b.Key.PlanDigest).Scan(&other)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("read restore dispatch plan alias: %w", err)
|
||
}
|
||
return fmt.Errorf("%w: effect already belongs to another plan digest", libraryrelease.ErrRestoreDispatchHeld)
|
||
}
|
||
|
||
// PrepareRestoreMarker 只建立 prepared：没有发送权，也不把已有 dispatching 抬回 prepared。
|
||
func (s *Store) PrepareRestoreMarker(ctx context.Context, b libraryrelease.RestoreMarkerBinding) (libraryrelease.RestoreDispatchStatus, error) {
|
||
digest, err := restoreDispatchBindingDigest(b)
|
||
if err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
if err := s.requireRestoreDispatchActivation(ctx, b.Key.TargetIdentity); err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
tx, err := s.db.Begin(ctx)
|
||
if err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("prepare restore dispatch: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
// 与 BeginLibraryAttempt 同序：先作业行、再 operation 行。这样"准备"不会与代次切换
|
||
// 交错成一份属于旧 identity 的台账。
|
||
if err := lockRestoreDispatchIdentity(ctx, tx, b); err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
if err := lockRestoreDispatchActivation(ctx, tx, b.Key.TargetIdentity, b.TargetFingerprint); err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
if err := rejectRestoreDispatchPlanAlias(ctx, tx, b); err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
if _, err := tx.Exec(ctx, `
|
||
INSERT INTO library_index_restore_dispatch
|
||
(target_identity, operation_id, window_job_id, plan_digest, index_name, document_id, phase, binding_digest, state)
|
||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'prepared')
|
||
ON CONFLICT ON CONSTRAINT library_index_restore_dispatch_identity DO NOTHING`,
|
||
b.Key.TargetIdentity, b.Key.OperationID, b.Key.WindowJobID, b.Key.PlanDigest,
|
||
b.Key.Index, b.Key.DocumentID, b.Key.Phase, digest); err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("record prepared restore dispatch: %w", err)
|
||
}
|
||
status, found, err := inspectRestoreDispatchTx(ctx, tx, b)
|
||
if err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
if !found {
|
||
return libraryrelease.RestoreDispatchStatus{}, errors.New("prepared restore dispatch was not durable")
|
||
}
|
||
if status.Held {
|
||
return status, fmt.Errorf("%w: %s", libraryrelease.ErrRestoreDispatchHeld, status.Reason)
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("commit prepared restore dispatch: %w", err)
|
||
}
|
||
return status, nil
|
||
}
|
||
|
||
// acquireRestoreMarker 的 beforeCommit 是提交结果不明这一档的测试接缝：钩子返回错误时
|
||
// 事务回滚，调用方不得发送。生产路径传 nil。
|
||
func (s *Store) acquireRestoreMarker(ctx context.Context, b libraryrelease.RestoreMarkerBinding, beforeCommit func() error) (*libraryrelease.RestoreSendPermit, libraryrelease.RestoreDispatchStatus, error) {
|
||
digest, err := restoreDispatchBindingDigest(b)
|
||
if err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
tx, err := s.db.Begin(ctx)
|
||
if err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("dispatch restore marker: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
if err := lockRestoreDispatchIdentity(ctx, tx, b); err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
// 撤销与 CAS 在同一事务线性化：这里拿激活行锁，撤销的 UPDATE 必须排队。
|
||
if err := lockRestoreDispatchActivation(ctx, tx, b.Key.TargetIdentity, b.TargetFingerprint); err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
// 当前代次必须属于这个原 window 作业，且租户/epoch/租约都仍有效。lease 用
|
||
// clock_timestamp()：等锁可能跨越租约到期，事务起始的 now() 不是判据。
|
||
var epoch int64
|
||
var owner, attemptState, attemptJob string
|
||
var leaseUntil *time.Time
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT epoch, owner, state, job_id, lease_until FROM library_attempts
|
||
WHERE operation_id=$1 AND state IN ('claimed','running') FOR UPDATE`, b.Key.OperationID).Scan(
|
||
&epoch, &owner, &attemptState, &attemptJob, &leaseUntil)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("%w: no active attempt for operation %s",
|
||
governance.ErrLibraryEpochStale, b.Key.OperationID)
|
||
}
|
||
if err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("read active attempt: %w", err)
|
||
}
|
||
if attemptJob != b.Key.WindowJobID {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("%w: active attempt belongs to job %s, not the original window %s",
|
||
governance.ErrLibraryEpochStale, attemptJob, b.Key.WindowJobID)
|
||
}
|
||
var leaseLive bool
|
||
if err := tx.QueryRow(ctx,
|
||
`SELECT COALESCE($1::timestamptz > clock_timestamp(), false)`, leaseUntil).Scan(&leaseLive); err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("read attempt lease: %w", err)
|
||
}
|
||
if owner != b.Owner || epoch != b.Epoch || !leaseLive {
|
||
return nil, libraryrelease.RestoreDispatchStatus{State: libraryrelease.RestoreDispatchPrepared, Held: true,
|
||
Reason: fmt.Sprintf("attempt %s#%d is not the live authorized owner", b.Owner, b.Epoch)},
|
||
fmt.Errorf("%w: attempt %s#%d is not current", governance.ErrLibraryEpochStale, b.Owner, b.Epoch)
|
||
}
|
||
if err := rejectRestoreDispatchPlanAlias(ctx, tx, b); err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
|
||
var id int64
|
||
var state, stored string
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT id, state, binding_digest FROM library_index_restore_dispatch
|
||
WHERE target_identity=$1 AND operation_id=$2 AND window_job_id=$3 AND plan_digest=$4
|
||
AND index_name=$5 AND document_id=$6 AND phase=$7 FOR UPDATE`,
|
||
b.Key.TargetIdentity, b.Key.OperationID, b.Key.WindowJobID, b.Key.PlanDigest,
|
||
b.Key.Index, b.Key.DocumentID, b.Key.Phase).Scan(&id, &state, &stored)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, errors.New("no prepared restore dispatch record: refusing to send")
|
||
}
|
||
if err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("lock restore dispatch: %w", err)
|
||
}
|
||
if stored != digest {
|
||
return nil, libraryrelease.RestoreDispatchStatus{State: state, Held: true, BindingConflict: true,
|
||
Reason: "restore dispatch binding differs from the immutable plan"}, libraryrelease.ErrRestoreDispatchHeld
|
||
}
|
||
if state != libraryrelease.RestoreDispatchPrepared {
|
||
return nil, libraryrelease.RestoreDispatchStatus{State: state, Held: true,
|
||
Reason: "dispatch already claimed; no second send is permitted"}, libraryrelease.ErrRestoreDispatchHeld
|
||
}
|
||
// effect 行锁是本次 CAS 前最后一个可能阻塞的锁。job/attempt/恢复投递的租约可能
|
||
// 正好在等它的时候过期；前面那几次检查发生在拿锁之前，不能充当 CAS 点的判据。
|
||
// 这些行锁都已被本事务持有，这里只重读、不改变锁序。
|
||
if err := recheckRestoreDispatchLeases(ctx, tx, b); err != nil {
|
||
var held *restoreDispatchLeaseExpired
|
||
if errors.As(err, &held) {
|
||
return nil, libraryrelease.RestoreDispatchStatus{State: libraryrelease.RestoreDispatchPrepared, Held: true,
|
||
Reason: held.Error()}, fmt.Errorf("%w: %s", governance.ErrLibraryEpochStale, held.Error())
|
||
}
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
|
||
var raw [16]byte
|
||
if _, err := rand.Read(raw[:]); err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, err
|
||
}
|
||
nonce := hex.EncodeToString(raw[:])
|
||
tag, err := tx.Exec(ctx, `
|
||
UPDATE library_index_restore_dispatch
|
||
SET state='dispatching', dispatch_owner=$2, dispatch_epoch=$3, dispatch_nonce=$4,
|
||
request_digest=$5, dispatch_committed_at=clock_timestamp(), updated_at=clock_timestamp()
|
||
WHERE id=$1 AND state='prepared'`, id, b.Owner, b.Epoch, nonce, b.RequestDigest)
|
||
if err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{}, fmt.Errorf("claim restore dispatch: %w", err)
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return nil, libraryrelease.RestoreDispatchStatus{State: state, Held: true,
|
||
Reason: "dispatch CAS lost"}, libraryrelease.ErrRestoreDispatchHeld
|
||
}
|
||
if beforeCommit != nil {
|
||
if err := beforeCommit(); err != nil {
|
||
// 提交结果不明：本调用不发送，也不把状态当成 prepared 重试。
|
||
return nil, libraryrelease.RestoreDispatchStatus{State: libraryrelease.RestoreDispatchDispatching, Held: true,
|
||
Reason: "commit outcome unknown"}, fmt.Errorf("commit restore dispatch: %w", err)
|
||
}
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return nil, libraryrelease.RestoreDispatchStatus{State: libraryrelease.RestoreDispatchDispatching, Held: true,
|
||
Reason: "commit outcome unknown"}, fmt.Errorf("commit restore dispatch: %w", err)
|
||
}
|
||
return libraryrelease.IssueRestoreSendPermit(b.Key, digest, nonce),
|
||
libraryrelease.RestoreDispatchStatus{State: libraryrelease.RestoreDispatchDispatching}, nil
|
||
}
|
||
|
||
func (s *Store) AcquireRestoreMarker(ctx context.Context, b libraryrelease.RestoreMarkerBinding) (*libraryrelease.RestoreSendPermit, libraryrelease.RestoreDispatchStatus, error) {
|
||
return s.acquireRestoreMarker(ctx, b, nil)
|
||
}
|
||
|
||
// ConfirmRestoreMarker 只记录真实回执（不再发送）。状态仍是 dispatching 时才允许；
|
||
// 回执本身来自本次受控发送或同计划标记的实时核实，不由客户端布尔值决定。
|
||
func (s *Store) ConfirmRestoreMarker(ctx context.Context, b libraryrelease.RestoreMarkerBinding, receipt []byte) error {
|
||
if len(receipt) == 0 || !json.Valid(receipt) {
|
||
return errors.New("restore marker confirmation requires a real receipt")
|
||
}
|
||
digest, err := restoreDispatchBindingDigest(b)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
tx, err := s.db.Begin(ctx)
|
||
if err != nil {
|
||
return fmt.Errorf("confirm restore marker: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback(ctx) }()
|
||
if err := lockRestoreDispatchIdentity(ctx, tx, b); err != nil {
|
||
return err
|
||
}
|
||
if err := lockRestoreDispatchActivation(ctx, tx, b.Key.TargetIdentity, b.TargetFingerprint); err != nil {
|
||
return err
|
||
}
|
||
var epoch int64
|
||
var owner, attemptJob string
|
||
var leaseUntil *time.Time
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT epoch, owner, job_id, lease_until FROM library_attempts
|
||
WHERE operation_id=$1 AND state IN ('claimed','running') FOR UPDATE`, b.Key.OperationID).Scan(&epoch, &owner, &attemptJob, &leaseUntil)
|
||
if err != nil {
|
||
return fmt.Errorf("confirm restore marker attempt: %w", err)
|
||
}
|
||
var leaseLive bool
|
||
if err := tx.QueryRow(ctx,
|
||
`SELECT COALESCE($1::timestamptz > clock_timestamp(), false)`, leaseUntil).Scan(&leaseLive); err != nil {
|
||
return fmt.Errorf("confirm restore marker lease: %w", err)
|
||
}
|
||
if attemptJob != b.Key.WindowJobID || owner != b.Owner || epoch != b.Epoch || !leaseLive {
|
||
return fmt.Errorf("%w: confirmation by a stale owner", governance.ErrLibraryEpochStale)
|
||
}
|
||
// 与 Acquire 同一个顺序：确认前的最后一把锁也要先拿到，再按数据库实时时钟重核
|
||
// 租约。否则"确认"可以在等锁跨过到期之后仍落库，把一次过期 owner 的回执写成事实。
|
||
var effectID int64
|
||
if err := tx.QueryRow(ctx, `
|
||
SELECT id FROM library_index_restore_dispatch
|
||
WHERE target_identity=$1 AND operation_id=$2 AND window_job_id=$3 AND plan_digest=$4
|
||
AND index_name=$5 AND document_id=$6 AND phase=$7 FOR UPDATE`,
|
||
b.Key.TargetIdentity, b.Key.OperationID, b.Key.WindowJobID, b.Key.PlanDigest,
|
||
b.Key.Index, b.Key.DocumentID, b.Key.Phase).Scan(&effectID); err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return errors.New("restore marker confirmation did not match a dispatch record")
|
||
}
|
||
return fmt.Errorf("lock restore marker for confirmation: %w", err)
|
||
}
|
||
if err := recheckRestoreDispatchLeases(ctx, tx, b); err != nil {
|
||
return fmt.Errorf("%w: %s", governance.ErrLibraryEpochStale, err)
|
||
}
|
||
tag, err := tx.Exec(ctx, `
|
||
UPDATE library_index_restore_dispatch
|
||
SET state='marker_confirmed', marker_receipt=$1::jsonb, confirmed_at=clock_timestamp(), updated_at=clock_timestamp()
|
||
WHERE target_identity=$2 AND operation_id=$3 AND window_job_id=$4 AND plan_digest=$5
|
||
AND index_name=$6 AND document_id=$7 AND phase=$8
|
||
AND state='dispatching' AND binding_digest=$9`,
|
||
string(receipt), b.Key.TargetIdentity, b.Key.OperationID, b.Key.WindowJobID, b.Key.PlanDigest,
|
||
b.Key.Index, b.Key.DocumentID, b.Key.Phase, digest)
|
||
if err != nil {
|
||
return fmt.Errorf("confirm restore marker: %w", err)
|
||
}
|
||
if tag.RowsAffected() != 1 {
|
||
return errors.New("restore marker confirmation did not match a dispatching record")
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return fmt.Errorf("commit restore marker confirmation: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// lockRestoreDispatchIdentity 以 job→operation 锁序把"这份许可是不是仍属于原作业"
|
||
// 钉死在原持久锚点上：目标/scope/来源快照、写集摘要（若已登记）、维护窗口与实例
|
||
// 指纹、恢复投递的 queue/operation 绑定都必须与不可变绑定一致。
|
||
func lockRestoreDispatchIdentity(ctx context.Context, tx pgx.Tx, b libraryrelease.RestoreMarkerBinding) error {
|
||
var operationID, scope, jobSource string
|
||
var leaseOwner string
|
||
var jobEpoch int64
|
||
var jobLease *time.Time
|
||
err := tx.QueryRow(ctx, `
|
||
SELECT operation_id, scope, target_snapshot_id, lease_owner, attempt_epoch, lease_until
|
||
FROM library_jobs WHERE id=$1 FOR UPDATE`, b.Key.WindowJobID).Scan(
|
||
&operationID, &scope, &jobSource, &leaseOwner, &jobEpoch, &jobLease)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return fmt.Errorf("window job %s not found", b.Key.WindowJobID)
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("lock window job: %w", err)
|
||
}
|
||
if operationID != b.Key.OperationID {
|
||
return errors.New("window job belongs to another operation")
|
||
}
|
||
if scope != b.Scope {
|
||
return fmt.Errorf("window job scope %s does not match binding %s", scope, b.Scope)
|
||
}
|
||
if jobSource == "" || jobSource != b.SourceSnapshot {
|
||
return fmt.Errorf("window job source snapshot %q does not match binding %q", jobSource, b.SourceSnapshot)
|
||
}
|
||
if leaseOwner != b.Owner || jobEpoch != b.Epoch || jobLease == nil {
|
||
return fmt.Errorf("%w: window job lease is not held by attempt %s#%d", governance.ErrLibraryEpochStale, b.Owner, b.Epoch)
|
||
}
|
||
var jobLeaseLive bool
|
||
if err := tx.QueryRow(ctx, `SELECT COALESCE($1::timestamptz > clock_timestamp(), false)`, jobLease).Scan(&jobLeaseLive); err != nil {
|
||
return fmt.Errorf("read window job lease: %w", err)
|
||
}
|
||
if !jobLeaseLive {
|
||
return fmt.Errorf("%w: window job lease expired while waiting for the dispatch CAS", governance.ErrLibraryEpochStale)
|
||
}
|
||
|
||
var opScope, opKind, opSource, requestDigest, writesetDigest, firstBackup string
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT scope, kind, target_snapshot_id, request_digest, COALESCE(writeset_digest,''), COALESCE(first_backup_snapshot_id,'')
|
||
FROM library_operations WHERE id=$1 FOR UPDATE`, b.Key.OperationID).Scan(
|
||
&opScope, &opKind, &opSource, &requestDigest, &writesetDigest, &firstBackup)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return fmt.Errorf("operation %s not found", b.Key.OperationID)
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("lock operation: %w", err)
|
||
}
|
||
if opScope != b.Scope {
|
||
return fmt.Errorf("operation scope %s does not match binding %s", opScope, b.Scope)
|
||
}
|
||
if opSource == "" || opSource != b.SourceSnapshot {
|
||
return fmt.Errorf("operation source snapshot %q does not match binding %q", opSource, b.SourceSnapshot)
|
||
}
|
||
if requestDigest == "" {
|
||
return errors.New("operation has no request digest anchor")
|
||
}
|
||
if writesetDigest != "" && writesetDigest != b.Key.PlanDigest {
|
||
return fmt.Errorf("operation plan digest %s does not match binding %s", writesetDigest, b.Key.PlanDigest)
|
||
}
|
||
if firstBackup != "" && firstBackup != b.BackupSnapshot {
|
||
return fmt.Errorf("operation backup snapshot %s does not match binding %s", firstBackup, b.BackupSnapshot)
|
||
}
|
||
_ = opKind
|
||
|
||
if b.TargetInstance != "" {
|
||
var ownerJob, ownerScope string
|
||
var released *time.Time
|
||
var state string
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT owner_job_id, owner_scope, state, released_at
|
||
FROM library_instance_maintenance WHERE instance=$1 FOR UPDATE`, b.TargetInstance).Scan(
|
||
&ownerJob, &ownerScope, &state, &released)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return fmt.Errorf("target instance %s has no maintenance window", b.TargetInstance)
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("lock instance maintenance: %w", err)
|
||
}
|
||
if released != nil || state == "released" {
|
||
return errors.New("target instance maintenance window is already released")
|
||
}
|
||
if ownerJob != b.Key.WindowJobID || ownerScope != b.Scope {
|
||
return fmt.Errorf("maintenance window belongs to %s (scope=%s), not window %s", ownerJob, ownerScope, b.Key.WindowJobID)
|
||
}
|
||
}
|
||
|
||
if b.RecoveryJobID != "" {
|
||
if err := lockAndVerifyRecoveryDelivery(ctx, tx, b); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func inspectRestoreDispatchTx(ctx context.Context, tx pgx.Tx, b libraryrelease.RestoreMarkerBinding) (libraryrelease.RestoreDispatchStatus, bool, error) {
|
||
digest, err := restoreDispatchBindingDigest(b)
|
||
if err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, false, err
|
||
}
|
||
var state, stored string
|
||
err = tx.QueryRow(ctx, `
|
||
SELECT state, binding_digest FROM library_index_restore_dispatch
|
||
WHERE target_identity=$1 AND operation_id=$2 AND window_job_id=$3 AND plan_digest=$4
|
||
AND index_name=$5 AND document_id=$6 AND phase=$7`,
|
||
b.Key.TargetIdentity, b.Key.OperationID, b.Key.WindowJobID, b.Key.PlanDigest,
|
||
b.Key.Index, b.Key.DocumentID, b.Key.Phase).Scan(&state, &stored)
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
return libraryrelease.RestoreDispatchStatus{}, false, nil
|
||
}
|
||
if err != nil {
|
||
return libraryrelease.RestoreDispatchStatus{}, false, fmt.Errorf("inspect restore dispatch: %w", err)
|
||
}
|
||
if stored != digest {
|
||
return libraryrelease.RestoreDispatchStatus{State: state, Held: true,
|
||
Reason: "restore dispatch binding differs from the immutable plan"}, true, nil
|
||
}
|
||
return libraryrelease.RestoreDispatchStatus{State: state,
|
||
Held: state != libraryrelease.RestoreDispatchPrepared}, true, nil
|
||
}
|