姚章浩
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 带原因；重启后恢复。
202 lines
8.5 KiB
Go
202 lines
8.5 KiB
Go
package httpapi
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"net/http"
|
||
"regexp"
|
||
"strings"
|
||
"time"
|
||
|
||
"knowledge-governance/internal/governance"
|
||
"knowledge-governance/internal/jobs"
|
||
)
|
||
|
||
// 页面发起封存：只投递一个作业，真正的停机/导出/恢复发生在主机侧 agent。
|
||
// 这里没有任何 Docker 访问，也没有来源凭据 —— 调用方只能选择范围与作业号。
|
||
const libraryCaptureJobType = "library.capture"
|
||
|
||
var libraryCaptureJobPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`)
|
||
|
||
type libraryCaptureRequest struct {
|
||
Scope string `json:"scope"`
|
||
Job string `json:"job"`
|
||
}
|
||
|
||
func libraryCaptureEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
|
||
var request libraryCaptureRequest
|
||
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
|
||
}
|
||
// The job id is the idempotency key handed to the agent: repeating the same
|
||
// scope+job returns the already sealed version instead of sealing twice.
|
||
captureJob := strings.TrimSpace(request.Job)
|
||
if captureJob == "" {
|
||
captureJob = "lib-" + request.Scope + "-" + time.Now().UTC().Format("20060102T150405")
|
||
}
|
||
if !libraryCaptureJobPattern.MatchString(captureJob) {
|
||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid job id"})
|
||
return
|
||
}
|
||
// 请求形状先校验（400），再看执行器（503）：不只看"配了 socket"——执行器可能
|
||
// 不可达、可能指向别的实例、可能降级。页面与接口必须用同一份判定，否则又会出现
|
||
// "点了才失败"或"封错库"。
|
||
if available, reason, executor := libraryExecutorAvailability(r.Context(), config); !available {
|
||
writeJSON(w, http.StatusServiceUnavailable, map[string]any{
|
||
"code": "library_capture_unavailable",
|
||
"reason": reason,
|
||
"executor": executor,
|
||
})
|
||
return
|
||
}
|
||
requestedBy := ""
|
||
if identity, ok := sessionFrom(r.Context()); ok {
|
||
requestedBy = identity.Username
|
||
}
|
||
payload, err := json.Marshal(map[string]string{
|
||
"scope": request.Scope, "capture_job": captureJob, "requested_by": requestedBy,
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot build capture job"})
|
||
return
|
||
}
|
||
now := time.Now().UTC()
|
||
// 业务作业先建：互斥由 library_jobs 的唯一索引保证，这样不会出现"队列里排着两个封存"。
|
||
// 队列投递失败或请求冲突时把业务作业取消，不留下永远 pending 的占位。
|
||
libraryJobID := "libseal-" + captureJob
|
||
createdSealJob := false
|
||
if store, ok := libraryStore(config); ok {
|
||
digest, digestErr := governance.LibraryRequestDigest(request.Scope, governance.LibraryJobSeal,
|
||
map[string]any{"capture_job": captureJob})
|
||
if digestErr == nil {
|
||
sealJob, reusedSeal, sealErr := store.CreateLibraryJob(r.Context(), governance.LibraryJob{
|
||
ID: libraryJobID, Scope: request.Scope, Kind: governance.LibraryJobSeal,
|
||
IdempotencyKey: "library.seal:" + captureJob, RequestDigest: digest, Actor: requestedBy,
|
||
}, governance.SealStages(), now)
|
||
switch {
|
||
case errors.Is(sealErr, governance.ErrLibraryJobActive):
|
||
writeJSON(w, http.StatusConflict, map[string]any{
|
||
"code": "library_seal_active", "error": "该目标已有在飞的封存作业，等它结束再发起",
|
||
})
|
||
return
|
||
case errors.Is(sealErr, governance.ErrLibraryJobIdempotencyConflict):
|
||
writeJSON(w, http.StatusConflict, map[string]any{
|
||
"code": "library_seal_scope_conflict", "error": "同一作业号已按另一个范围封存",
|
||
})
|
||
return
|
||
case sealErr != nil:
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create seal job"})
|
||
return
|
||
case !reusedSeal:
|
||
createdSealJob = true
|
||
default:
|
||
libraryJobID = sealJob.ID
|
||
}
|
||
}
|
||
}
|
||
// 队列作业 id 落进 jobs.id（varchar(36)）：原来拼 "library-capture-<scope>-<nano>"
|
||
// 是 45 个字符，真实请求 100% 被 PG 拒（value too long）——只是页面一直停在
|
||
// "未接线"上，没人走到这里。这里按列宽拼短 id，并留出 scope 全称。
|
||
requestedID := fmt.Sprintf("libcap-%s-%d", request.Scope, now.UnixNano())
|
||
// The idempotency key is the capture job id: the same logical capture must map
|
||
// to one queue row. Binding the scope into the key would silently create a
|
||
// second job instead of reporting a conflicting request, so the conflict is
|
||
// detected by comparing the stored payload below.
|
||
job, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{
|
||
ID: requestedID,
|
||
Type: libraryCaptureJobType,
|
||
Payload: payload,
|
||
AvailableAt: now,
|
||
MaxAttempts: 2,
|
||
IdempotencyKey: "library.capture:" + captureJob,
|
||
}, now)
|
||
if err != nil || job == nil {
|
||
if store, ok := libraryStore(config); ok && createdSealJob {
|
||
_ = store.CancelLibraryJob(r.Context(), libraryJobID, "队列投递失败，未开始封存", now)
|
||
}
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue capture job"})
|
||
return
|
||
}
|
||
if store, ok := libraryStore(config); ok {
|
||
if attachErr := store.AttachLibraryQueueJob(r.Context(), libraryJobID, job.ID, now); attachErr != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot attach capture job"})
|
||
return
|
||
}
|
||
}
|
||
// Enqueue returns the existing row for a reused key. Answer from that row, not
|
||
// from the current request: a reused job id with another scope must be refused
|
||
// instead of being reported as a job for the new scope.
|
||
var stored struct {
|
||
Scope string `json:"scope"`
|
||
CaptureJob string `json:"capture_job"`
|
||
RequestedBy string `json:"requested_by"`
|
||
}
|
||
cancelSeal := func(reason string) {
|
||
if store, ok := libraryStore(config); ok && createdSealJob {
|
||
_ = store.CancelLibraryJob(r.Context(), libraryJobID, reason, now)
|
||
}
|
||
}
|
||
if err = json.Unmarshal(job.Payload, &stored); err != nil || stored.CaptureJob != captureJob {
|
||
cancelSeal("请求冲突，未开始封存")
|
||
writeJSON(w, http.StatusConflict, map[string]any{
|
||
"code": "library_capture_conflict", "error": "该作业号已被另一个封存请求占用",
|
||
"existing_scope": stored.Scope, "requested_scope": request.Scope,
|
||
})
|
||
return
|
||
}
|
||
if stored.Scope != request.Scope {
|
||
cancelSeal("同一作业号已按另一个范围投递")
|
||
writeJSON(w, http.StatusConflict, map[string]any{
|
||
"code": "library_capture_scope_conflict", "error": "同一作业号已按另一个范围投递",
|
||
"existing_scope": stored.Scope, "requested_scope": request.Scope,
|
||
})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusAccepted, map[string]any{
|
||
"job_id": job.ID, "capture_job": stored.CaptureJob, "scope": stored.Scope,
|
||
"state": job.State, "reused": job.ID != requestedID, "requested_by": stored.RequestedBy,
|
||
})
|
||
}
|
||
|
||
// libraryCaptureJobs lists recent capture jobs so the page can show real state
|
||
// instead of an optimistic spinner. It reads the queue, not the archive.
|
||
func libraryCaptureJobs(w http.ResponseWriter, r *http.Request, config Config) {
|
||
// Ask the store for this type only. Reading the global queue and filtering in
|
||
// the handler would hide capture jobs behind unrelated traffic and make the
|
||
// page claim "idle" while a capture is still queued.
|
||
byType, ok := config.Store.(interface {
|
||
ListByType(ctx context.Context, jobType string, limit int) ([]jobs.Job, error)
|
||
})
|
||
if !ok {
|
||
writeJSON(w, http.StatusOK, map[string]any{"available": false,
|
||
"reason": "任务存储不支持按类型查询，页面不会据此判断空闲", "jobs": []any{}})
|
||
return
|
||
}
|
||
queue, err := byType.ListByType(r.Context(), libraryCaptureJobType, 20)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read capture jobs"})
|
||
return
|
||
}
|
||
views := make([]map[string]any, 0, len(queue))
|
||
for _, job := range queue {
|
||
var payload struct {
|
||
Scope string `json:"scope"`
|
||
CaptureJob string `json:"capture_job"`
|
||
RequestedBy string `json:"requested_by"`
|
||
}
|
||
_ = json.Unmarshal(job.Payload, &payload)
|
||
views = append(views, map[string]any{"job_id": job.ID, "capture_job": payload.CaptureJob,
|
||
"scope": payload.Scope, "state": job.State, "attempt": job.Attempt,
|
||
"last_error": job.LastError, "requested_by": payload.RequestedBy,
|
||
"created_at": job.CreatedAt, "updated_at": job.UpdatedAt})
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"available": true, "jobs": views})
|
||
}
|