姚章浩
47eb23cf68
页面上的"可发起封存"原来只判 LIBRARY_AGENT_SOCKET 非空：正式栈的 API 根本没挂
那个 socket，而 agent 曾顶着正式单元名冻结候选实例——配置非空、页面可点、真发起
却封错库。
现在：agent 新增 health 动作（不占封存操作锁、不跑子进程，只回答身份/降级/忙），
worker 每 30 秒探测并写入迁移 074 的 library_executor_status；overview、快照列表与
POST /library/captures 共用同一份判定（过期、不可达、实例不符、降级、正忙都拒绝并
给原因）。封存结果由 agent 附上自报身份，worker 在停写入者之前再探测一次、执行后
用结果复核，不一致就不登记测试封存——心跳只能证明"刚才"。
期望身份用 LIBRARY_SOURCE_*（封存源），不是 LIBRARY_TARGET_*（同步目标），否则
正确的执行器会被判成不匹配。另加 GET /library/restore-availability：恢复能否执行由
双闸、该快照的恢复授权与实例维护状态决定，页面不再写死 disabled。
真跑（候选隔离栈，两处 socket 隔离不变）：overview 返回 executor
{kgmaint-weknora/kgmaint-app}；停 agent 后转不可达并 503 带原因；重启后恢复。
321 lines
14 KiB
Go
321 lines
14 KiB
Go
package httpapi
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"knowledge-governance/internal/governance"
|
||
"knowledge-governance/internal/jobs"
|
||
)
|
||
|
||
// 按指定整库版本恢复：与同步分开的作业类型，需要**单独的管理员授权**，
|
||
// 授权摘要绑定"恢复到这个快照"这一件事，不能拿同步授权顶替。
|
||
const libraryRestoreJobType = "library.restore"
|
||
|
||
func libraryRestoreParamsDigest(scope, snapshotID string) (string, error) {
|
||
return governance.LibraryRequestDigest(scope, governance.LibraryJobRestore,
|
||
map[string]any{"restore_snapshot_id": snapshotID})
|
||
}
|
||
|
||
type libraryRestoreRequest struct {
|
||
Scope string `json:"scope"`
|
||
SnapshotID string `json:"snapshot_id"`
|
||
}
|
||
|
||
func libraryRestoreEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
|
||
if r.Method != http.MethodPost {
|
||
writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"})
|
||
return
|
||
}
|
||
identity, ok := sessionFrom(r.Context())
|
||
if !ok || roleRank(identity.Role) < roleRank(roleAdmin) {
|
||
writeJSON(w, http.StatusForbidden, map[string]any{"error": "按版本恢复仅管理员可发起"})
|
||
return
|
||
}
|
||
store, ok := libraryStore(config)
|
||
if !ok {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_restore_unavailable"})
|
||
return
|
||
}
|
||
var request libraryRestoreRequest
|
||
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&request); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"})
|
||
return
|
||
}
|
||
if request.Scope != "internal" && request.Scope != "external" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "scope must be internal or external"})
|
||
return
|
||
}
|
||
if strings.TrimSpace(request.SnapshotID) == "" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "snapshot_id required"})
|
||
return
|
||
}
|
||
if body, held := instanceMaintenanceGuard(r.Context(), config, request.Scope); held {
|
||
writeJSON(w, http.StatusConflict, body)
|
||
return
|
||
}
|
||
digest, err := libraryRestoreParamsDigest(request.Scope, request.SnapshotID)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot digest params"})
|
||
return
|
||
}
|
||
// 双闸与恢复演练证据同样适用于恢复：恢复也是写生产。
|
||
reasons := make([]string, 0, 4)
|
||
if !config.ProductionReleaseRequestsEnabled {
|
||
reasons = append(reasons, "生产写入闸口关闭")
|
||
}
|
||
if !config.ProductionBackupRestoreVerified {
|
||
reasons = append(reasons, "没有已核验的生产恢复演练证据")
|
||
}
|
||
authorizationOK := false
|
||
if storeWithAuth, ok := config.Store.(interface {
|
||
ActiveLibraryAuthorization(ctx context.Context, scope, snapshotID, digest string, now time.Time) (governance.LibraryReleaseAuthorization, bool, error)
|
||
}); ok {
|
||
_, found, authErr := storeWithAuth.ActiveLibraryAuthorization(r.Context(), request.Scope, request.SnapshotID, digest, time.Now().UTC())
|
||
authorizationOK = authErr == nil && found
|
||
}
|
||
if !authorizationOK {
|
||
reasons = append(reasons, "没有针对该快照的恢复授权")
|
||
}
|
||
if len(reasons) > 0 {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_restore_blocked", "reasons": reasons})
|
||
return
|
||
}
|
||
now := time.Now().UTC()
|
||
// 作业 id 会作为维护窗口的 job 标识落进 freeze journal，而冻结协调器只接受
|
||
// [A-Za-z0-9_-]：带 "." 的时间戳会让恢复在开窗口时被拒（真跑报 invalid freeze request），
|
||
// 所以用纳秒整数，和同步作业保持一致。
|
||
jobID := "librestore-" + request.Scope + "-" + strconv.FormatInt(now.UnixNano(), 10)
|
||
job, reused, err := store.CreateLibraryJob(r.Context(), governance.LibraryJob{
|
||
ID: jobID, Scope: request.Scope, Kind: governance.LibraryJobRestore,
|
||
IdempotencyKey: "library.restore:" + request.Scope + ":" + request.SnapshotID,
|
||
RequestDigest: digest, TargetSnapshotID: request.SnapshotID, Actor: identity.Username,
|
||
}, governance.SyncStages(), now)
|
||
if errors.Is(err, governance.ErrLibraryJobActive) {
|
||
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_restore_active", "error": "该目标已有在飞作业"})
|
||
return
|
||
}
|
||
if errors.Is(err, governance.ErrLibraryJobIdempotencyConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_restore_conflict", "error": "同一恢复请求参数不一致"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create restore job"})
|
||
return
|
||
}
|
||
if reused && (job.State == governance.LibraryJobFailed || job.State == governance.LibraryJobCancelled) {
|
||
job, reused, err = store.CreateLibraryJob(r.Context(), governance.LibraryJob{
|
||
ID: jobID + "-retry", Scope: request.Scope, Kind: governance.LibraryJobRestore,
|
||
IdempotencyKey: "library.restore:" + request.Scope + ":" + request.SnapshotID + ":retry:" + strconv.FormatInt(now.UnixNano(), 10),
|
||
RequestDigest: digest, TargetSnapshotID: request.SnapshotID, Actor: identity.Username,
|
||
}, governance.SyncStages(), now)
|
||
if err != nil || reused {
|
||
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_restore_retry_conflict", "error": "无法为重试创建新作业"})
|
||
return
|
||
}
|
||
}
|
||
if reused {
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": true})
|
||
return
|
||
}
|
||
payload, err := json.Marshal(map[string]string{
|
||
"library_job_id": job.ID, "scope": request.Scope, "target_snapshot_id": request.SnapshotID,
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot build restore payload"})
|
||
return
|
||
}
|
||
queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
|
||
ID: governance.QueueJobIDFor(job.ID), Type: libraryRestoreJobType, Payload: payload, AvailableAt: now,
|
||
MaxAttempts: 1, IdempotencyKey: "library.restore.queue:" + job.ID,
|
||
}, now)
|
||
if err != nil || queueJob == nil {
|
||
_ = store.CancelLibraryJob(r.Context(), job.ID, "队列投递失败，未开始恢复", now)
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue restore job"})
|
||
return
|
||
}
|
||
if attachErr := store.AttachLibraryQueueJob(r.Context(), job.ID, queueJob.ID, now); attachErr != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot attach restore job"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": false})
|
||
}
|
||
|
||
// libraryRestoreAvailability 回答"现在能不能把生产恢复到这份快照"。
|
||
//
|
||
// 为什么需要它：页面原来把"恢复至该版本"按钮写死为 disabled，理由是"未接线"——
|
||
// 但接口、授权、恢复作业其实都已经在跑。按钮该由真实条件决定：双闸打开、
|
||
// 有这份快照的恢复授权、目标实例没有在维护。
|
||
type libraryRestoreAvailabilityView struct {
|
||
Allowed bool `json:"allowed"`
|
||
Reasons []string `json:"reasons"`
|
||
Scope string `json:"scope"`
|
||
SnapshotID string `json:"snapshot_id"`
|
||
Authorization libraryAuthorizationView `json:"authorization"`
|
||
}
|
||
|
||
func libraryRestoreAvailability(w http.ResponseWriter, r *http.Request, config Config) {
|
||
scope, err := scopeFromQuery(r)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
|
||
return
|
||
}
|
||
snapshotID := strings.TrimSpace(r.URL.Query().Get("snapshot_id"))
|
||
if snapshotID == "" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "snapshot_id required"})
|
||
return
|
||
}
|
||
view := libraryRestoreAvailabilityView{Scope: scope, SnapshotID: snapshotID, Reasons: []string{}}
|
||
if !config.ProductionReleaseRequestsEnabled {
|
||
view.Reasons = append(view.Reasons, "生产写入闸口关闭（PRODUCTION_RELEASE_REQUESTS_ENABLED=false）")
|
||
}
|
||
if !config.ProductionBackupRestoreVerified {
|
||
view.Reasons = append(view.Reasons, "没有已核验的生产恢复演练证据（PRODUCTION_BACKUP_RESTORE_VERIFIED=false）")
|
||
}
|
||
digest, err := libraryRestoreParamsDigest(scope, snapshotID)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot digest params"})
|
||
return
|
||
}
|
||
grant, found := governance.LibraryReleaseAuthorization{}, false
|
||
if storeWithAuth, ok := config.Store.(interface {
|
||
ActiveLibraryAuthorization(ctx context.Context, scope, snapshotID, digest string, now time.Time) (governance.LibraryReleaseAuthorization, bool, error)
|
||
}); ok {
|
||
grant, found, err = storeWithAuth.ActiveLibraryAuthorization(r.Context(), scope, snapshotID, digest, time.Now().UTC())
|
||
if err != nil {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"allowed": false,
|
||
"reasons": []string{"读取恢复授权失败：" + err.Error()}})
|
||
return
|
||
}
|
||
}
|
||
view.Authorization = authorizationViewOf(grant, found)
|
||
if !found {
|
||
view.Reasons = append(view.Reasons, "没有针对该快照的恢复授权：先签发 restore 授权（绑定同一快照）")
|
||
}
|
||
if _, held := instanceMaintenanceGuard(r.Context(), config, scope); held {
|
||
view.Reasons = append(view.Reasons, "目标实例正在维护中，等当前作业释放后再恢复")
|
||
}
|
||
view.Allowed = len(view.Reasons) == 0
|
||
writeJSON(w, http.StatusOK, view)
|
||
}
|
||
|
||
// 崩溃续作：失败作业可以按原参数重新排队，生成新作业（旧作业保留为审计记录）。
|
||
func libraryJobResume(w http.ResponseWriter, r *http.Request, config Config) {
|
||
identity, ok := sessionFrom(r.Context())
|
||
// 续作会重新写生产，权限与恢复一致：只有管理员能触发。
|
||
if !ok || roleRank(identity.Role) < roleRank(roleAdmin) {
|
||
writeJSON(w, http.StatusForbidden, map[string]any{"error": "续作需要管理员权限"})
|
||
return
|
||
}
|
||
store, ok := libraryStore(config)
|
||
if !ok {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"error": "作业存储不可用"})
|
||
return
|
||
}
|
||
sourceID := strings.TrimSpace(r.PathValue("id"))
|
||
source, err := store.LibraryJob(r.Context(), sourceID)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusNotFound, map[string]any{"error": "作业不存在"})
|
||
return
|
||
}
|
||
if source.State != governance.LibraryJobFailed && source.State != governance.LibraryJobCancelled {
|
||
writeJSON(w, http.StatusConflict, map[string]any{"error": "只有失败或已取消的作业可以续作"})
|
||
return
|
||
}
|
||
// 与同步/恢复一致：闸口、恢复演练证据与授权都要在 API 侧 fail-closed，
|
||
// 不能让前台接受一个 worker 必然拒绝的请求。
|
||
reasons := make([]string, 0, 3)
|
||
if !config.ProductionReleaseRequestsEnabled {
|
||
reasons = append(reasons, "生产写入闸口关闭")
|
||
}
|
||
if !config.ProductionBackupRestoreVerified {
|
||
reasons = append(reasons, "没有已核验的生产恢复演练证据")
|
||
}
|
||
var digest string
|
||
switch source.Kind {
|
||
case governance.LibraryJobSync:
|
||
digest, _ = librarySyncParamsDigest(source.Scope, source.TargetSnapshotID)
|
||
case governance.LibraryJobRestore:
|
||
digest, _ = libraryRestoreParamsDigest(source.Scope, source.TargetSnapshotID)
|
||
}
|
||
authorizationOK := false
|
||
if storeWithAuth, ok := config.Store.(interface {
|
||
ActiveLibraryAuthorization(ctx context.Context, scope, snapshotID, digest string, now time.Time) (governance.LibraryReleaseAuthorization, bool, error)
|
||
}); ok && digest != "" {
|
||
_, found, authErr := storeWithAuth.ActiveLibraryAuthorization(r.Context(), source.Scope,
|
||
source.TargetSnapshotID, digest, time.Now().UTC())
|
||
authorizationOK = authErr == nil && found
|
||
}
|
||
if !authorizationOK {
|
||
reasons = append(reasons, "没有针对该快照的有效授权")
|
||
}
|
||
if len(reasons) > 0 {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_resume_blocked", "reasons": reasons})
|
||
return
|
||
}
|
||
|
||
now := time.Now().UTC()
|
||
// 幂等键按源作业固定：重复点击返回同一个续作作业，而不是每次新建。
|
||
jobID := sourceID + "-resume"
|
||
var kind governance.LibraryJobKind
|
||
queueType := ""
|
||
switch source.Kind {
|
||
case governance.LibraryJobSync:
|
||
kind, queueType = governance.LibraryJobSync, librarySyncJobType
|
||
case governance.LibraryJobRestore:
|
||
kind, queueType = governance.LibraryJobRestore, libraryRestoreJobType
|
||
default:
|
||
writeJSON(w, http.StatusConflict, map[string]any{"error": "封存作业由封存入口重发，不走续作"})
|
||
return
|
||
}
|
||
job, reused, err := store.CreateLibraryJob(r.Context(), governance.LibraryJob{
|
||
ID: jobID, Scope: source.Scope, Kind: kind,
|
||
IdempotencyKey: "library.resume:" + sourceID,
|
||
RequestDigest: digest, TargetSnapshotID: source.TargetSnapshotID, Actor: identity.Username,
|
||
}, governance.SyncStages(), now)
|
||
if errors.Is(err, governance.ErrLibraryJobActive) {
|
||
writeJSON(w, http.StatusConflict, map[string]any{"error": "该目标已有在飞作业"})
|
||
return
|
||
}
|
||
if errors.Is(err, governance.ErrLibraryJobIdempotencyConflict) {
|
||
writeJSON(w, http.StatusConflict, map[string]any{"error": "同一续作请求参数不一致"})
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create resumed job"})
|
||
return
|
||
}
|
||
if reused {
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "resumed_from": sourceID, "reused": true})
|
||
return
|
||
}
|
||
// 写入已完成、只差登记版本的情况：续作只做幂等收尾，不重写生产。
|
||
body := map[string]any{"library_job_id": job.ID, "scope": source.Scope,
|
||
"target_snapshot_id": source.TargetSnapshotID}
|
||
if source.Stage == "finalize" && source.SnapshotID != "" {
|
||
body["finalize_only"] = true
|
||
}
|
||
payload, _ := json.Marshal(body)
|
||
queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
|
||
ID: governance.QueueJobIDFor(job.ID), Type: queueType, Payload: payload, AvailableAt: now,
|
||
MaxAttempts: 1, IdempotencyKey: "library.resume.queue:" + job.ID,
|
||
}, now)
|
||
if err != nil || queueJob == nil {
|
||
_ = store.CancelLibraryJob(r.Context(), job.ID, "队列投递失败", now)
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue resumed job"})
|
||
return
|
||
}
|
||
// 附不上队列号就等于作业没人跑：必须显式失败并取消，不能静默返回 202。
|
||
if attachErr := store.AttachLibraryQueueJob(r.Context(), job.ID, queueJob.ID, now); attachErr != nil {
|
||
_ = store.CancelLibraryJob(r.Context(), job.ID, "队列归属写入失败", now)
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot attach resumed job"})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "resumed_from": sourceID, "reused": false})
|
||
}
|