姚章浩
72ea9726e7
acquired_at 没进 ON CONFLICT DO UPDATE，于是第一次建立基线写下的时间再也不会变： 恢复作业 16:59 才开窗口，页面却显示 13:37。库级与实例级是两份 SQL，都要改成 "同一作业重入（崩溃接管）保留旧值，新作业接管用本次时间"。 真实 PostgreSQL 用例覆盖两者：首开、同作业重入、释放后新作业接管、持有期间他人被拒。
200 lines
8.3 KiB
Go
200 lines
8.3 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"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 说明这次维护到底动了什么：
|
|
// - "instance"：入口与写入者被真的关掉（实例级），所有库都受影响；
|
|
// - "library"：只锁了治理库的一个 scope，目标入口/后台写入没被挡，不能当成系统维护。
|
|
type LibraryMaintenance struct {
|
|
Scope string `json:"scope"`
|
|
OwnerJobID string `json:"owner_job_id"`
|
|
TargetIdentity string `json:"target_identity"`
|
|
Level string `json:"level"`
|
|
TargetFingerprint string `json:"target_fingerprint,omitempty"`
|
|
Services []string `json:"services,omitempty"`
|
|
IngressState string `json:"ingress_state"`
|
|
State string `json:"state"`
|
|
OriginalServiceState string `json:"original_service_state"`
|
|
Note string `json:"note"`
|
|
AcquiredAt time.Time `json:"acquired_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
ReleasedAt *time.Time `json:"released_at,omitempty"`
|
|
}
|
|
|
|
// ErrMaintenanceHeld 表示目标维护窗口被别的作业占着：任何新的写入都必须拒绝。
|
|
var ErrMaintenanceHeld = errors.New("目标正在维护中，被其他作业占用")
|
|
|
|
const maintenanceColumns = `scope, owner_job_id, target_identity, level, target_fingerprint, services,
|
|
ingress_state, state, original_service_state, note, acquired_at, updated_at, released_at`
|
|
|
|
// AcquireLibraryMaintenance 原子地拿到维护权：
|
|
// 同一作业可以重入（接管后继续），别的作业一律拒绝——这不是"尽力而为"的锁。
|
|
func (s *Store) AcquireLibraryMaintenance(ctx context.Context, item LibraryMaintenance, now time.Time) (LibraryMaintenance, error) {
|
|
level := item.Level
|
|
if level == "" {
|
|
level = "library"
|
|
}
|
|
ingress := item.IngressState
|
|
if ingress == "" {
|
|
ingress = "unknown"
|
|
}
|
|
services := item.Services
|
|
if services == nil {
|
|
services = []string{}
|
|
}
|
|
raw, err := json.Marshal(services)
|
|
if err != nil {
|
|
return LibraryMaintenance{}, fmt.Errorf("acquire library maintenance: %w", err)
|
|
}
|
|
row := s.db.QueryRow(ctx, `
|
|
INSERT INTO library_maintenance (scope, owner_job_id, target_identity, level, target_fingerprint, services,
|
|
ingress_state, state, original_service_state, note, acquired_at, updated_at)
|
|
VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8,$9,$10,$11,$11)
|
|
ON CONFLICT (scope) DO UPDATE SET
|
|
owner_job_id = EXCLUDED.owner_job_id,
|
|
target_identity = EXCLUDED.target_identity,
|
|
level = EXCLUDED.level,
|
|
target_fingerprint = EXCLUDED.target_fingerprint,
|
|
services = EXCLUDED.services,
|
|
ingress_state = EXCLUDED.ingress_state,
|
|
state = EXCLUDED.state,
|
|
-- 同上：原运行状态只由第一次进入窗口时决定，重入不能覆盖。
|
|
original_service_state = CASE
|
|
WHEN library_maintenance.owner_job_id = EXCLUDED.owner_job_id
|
|
AND library_maintenance.original_service_state <> ''
|
|
THEN library_maintenance.original_service_state
|
|
ELSE EXCLUDED.original_service_state
|
|
END,
|
|
note = EXCLUDED.note,
|
|
-- 窗口的"起始时间"属于**这次**占用：只有同一作业重入（崩溃接管）才保留第一次的
|
|
-- 时间。少了这一条，页面会一直显示第一次建立基线时的 acquired_at，让人以为
|
|
-- 当前这个窗口从那时起就没释放过。
|
|
acquired_at = CASE
|
|
WHEN library_maintenance.owner_job_id = EXCLUDED.owner_job_id
|
|
THEN library_maintenance.acquired_at
|
|
ELSE EXCLUDED.acquired_at
|
|
END,
|
|
updated_at = EXCLUDED.updated_at,
|
|
released_at = NULL
|
|
-- 已经释放的窗口允许被新作业接管（否则第一个同步之后谁也再进不去）；
|
|
-- 仍在持有的窗口只允许原作业重入。
|
|
WHERE library_maintenance.released_at IS NOT NULL
|
|
OR library_maintenance.owner_job_id = EXCLUDED.owner_job_id
|
|
RETURNING `+maintenanceColumns,
|
|
item.Scope, item.OwnerJobID, item.TargetIdentity, level, item.TargetFingerprint, string(raw),
|
|
ingress, item.State, item.OriginalServiceState, item.Note, now.UTC())
|
|
acquired, err := scanMaintenance(row)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return LibraryMaintenance{}, ErrMaintenanceHeld
|
|
}
|
|
if err != nil {
|
|
return LibraryMaintenance{}, fmt.Errorf("acquire library maintenance: %w", err)
|
|
}
|
|
return acquired, nil
|
|
}
|
|
|
|
// UpdateLibraryMaintenanceState 只允许持有者改自己的窗口状态。
|
|
func (s *Store) UpdateLibraryMaintenanceState(ctx context.Context, scope, ownerJobID, state, note string, 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`, scope, ownerJobID, state, note, now.UTC())
|
|
if err != nil {
|
|
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
|
|
}
|
|
|
|
// SetLibraryMaintenanceIngress 记录入口的真实状态（closed/open），
|
|
// 页面据此区分"入口关了"和"只是容器停了"。
|
|
func (s *Store) SetLibraryMaintenanceIngress(ctx context.Context, scope, ownerJobID, state, note string, 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`, scope, ownerJobID, state, note, now.UTC())
|
|
if err != nil {
|
|
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
|
|
}
|
|
|
|
// ReleaseLibraryMaintenance 只在完整验证通过后调用；失败路径必须保持窗口。
|
|
func (s *Store) ReleaseLibraryMaintenance(ctx context.Context, scope, ownerJobID string, 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`, scope, ownerJobID, now.UTC())
|
|
if err != nil {
|
|
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
|
|
}
|
|
|
|
// LibraryMaintenanceOf 读当前窗口状态（页面与诊断用）。
|
|
func (s *Store) LibraryMaintenanceOf(ctx context.Context, scope string) (*LibraryMaintenance, error) {
|
|
row := s.db.QueryRow(ctx, `SELECT `+maintenanceColumns+` FROM library_maintenance WHERE scope=$1`, scope)
|
|
item, err := scanMaintenance(row)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &item, nil
|
|
}
|
|
|
|
func scanMaintenance(row pgx.Row) (LibraryMaintenance, error) {
|
|
var item LibraryMaintenance
|
|
var raw []byte
|
|
err := row.Scan(&item.Scope, &item.OwnerJobID, &item.TargetIdentity, &item.Level, &item.TargetFingerprint,
|
|
&raw, &item.IngressState, &item.State, &item.OriginalServiceState, &item.Note,
|
|
&item.AcquiredAt, &item.UpdatedAt, &item.ReleasedAt)
|
|
if err != nil {
|
|
return LibraryMaintenance{}, err
|
|
}
|
|
if len(raw) > 0 {
|
|
if err := json.Unmarshal(raw, &item.Services); err != nil {
|
|
return LibraryMaintenance{}, fmt.Errorf("maintenance services unreadable: %w", err)
|
|
}
|
|
}
|
|
return item, nil
|
|
}
|