feat(library): 封存执行器可用性与还原条件来自真实观测

页面上的"可发起封存"原来只判 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 带原因;重启后恢复。
This commit is contained in:
姚章浩
2026-10-29 13:01:37 +08:00
parent a06945516f
commit 47eb23cf68
26 changed files with 1351 additions and 113 deletions
+45
View File
@@ -1312,6 +1312,51 @@
}
}
},
"/api/v1/library/restore-availability": {
"get": {
"tags": [
"Runtime"
],
"operationId": "getLibraryRestoreAvailability",
"summary": "这份快照现在能不能恢复到生产:双闸、恢复授权与目标实例维护状态逐条说明",
"parameters": [
{
"name": "scope",
"in": "query",
"required": false,
"schema": {
"type": "string",
"enum": [
"internal",
"external"
]
}
},
{
"name": "snapshot_id",
"in": "query",
"required": true,
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"500": {
"$ref": "#/components/responses/InternalError"
},
"503": {
"description": "读取恢复授权失败(不是\"可恢复\",是\"无法判定\")"
}
}
}
},
"/api/v1/library/authorizations": {
"get": {
"tags": [
+2
View File
@@ -204,7 +204,9 @@ func run(ctx context.Context, getenv func(string) string) error {
LibraryArchiveRoot: config.LibraryArchiveRoot,
LibraryTargetInstance: config.LibraryTargetInstance,
LibrarySourceInstance: config.SourceInstance(),
LibrarySourceProject: config.SourceProject(),
LibraryAgentSocket: config.LibraryAgentSocket,
LibraryExecutorStatusMaxAge: config.LibraryExecutorStatusMaxAge,
LibraryCapacityBudgetBytes: config.LibraryCapacityBudgetBytes,
LibraryProductionDatabaseURL: config.LibraryProductionDatabaseURL,
LibraryProductionInternalKB: config.LibraryProductionInternalKB,
+89
View File
@@ -0,0 +1,89 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"knowledge-governance/internal/libraryagent"
)
// 封存执行的身份核对分两次:执行前一次、执行后一次。
//
// 为什么心跳不够:心跳说明"上一次探测时"那个 socket 后面是哪个实例,而从心跳到
// 真正停写入者之间,socket 可能被换成另一个 agent(换挂载、换单元、误启正式单元)。
// 执行前探测把窗口缩到最小;执行后复核 agent 自报的身份,则把"换过 socket"这件事
// 变成一次显式的作业失败,而不是悄悄封错了库还登记成测试封存。
// captureExecutorPrecheck 在动源写入者之前确认执行器就是期望的那个实例。
func captureExecutorPrecheck(ctx context.Context, client libraryagent.ClientConfig, expectedProject string, probeTimeout time.Duration) error {
if strings.TrimSpace(expectedProject) == "" {
return errors.New("未配置封存来源实例身份(LIBRARY_SOURCE_PROJECT),拒绝在无法核对执行器的情况下停写入者")
}
if probeTimeout <= 0 {
probeTimeout = 3 * time.Second
}
bounded, cancel := context.WithTimeout(ctx, probeTimeout)
defer cancel()
result, err := libraryagent.Health(bounded, client)
if err != nil {
return fmt.Errorf("封存执行器不可达,拒绝停写入者:%w", err)
}
identity, err := agentIdentityOf(result.Payload)
if err != nil {
return err
}
if identity.Project != expectedProject {
return fmt.Errorf("封存执行器指向的实例不匹配(agent=%q,期望=%q),拒绝停写入者",
identity.Project, expectedProject)
}
if identity.Degraded != "" {
return fmt.Errorf("封存执行器处于降级状态(%s),拒绝停写入者", identity.Degraded)
}
return nil
}
// captureExecutorPostcheck 用封存结果里 agent 自报的身份复核:不一致就当作
// "封存可能来自别的实例",调用方不得把它登记成测试封存。
func captureExecutorPostcheck(payload json.RawMessage, expectedProject string) error {
identity, err := agentIdentityOf(payload)
if err != nil {
return err
}
if identity.Project == "" {
return errors.New("封存结果里没有执行器身份,无法证明这次封存来自期望实例")
}
if identity.Project != expectedProject {
return fmt.Errorf("封存结果来自另一个实例(agent=%q,期望=%q),该快照不能当作本实例的测试封存",
identity.Project, expectedProject)
}
return nil
}
// agentIdentity 是 agent 在 health/封存结果里自报的身份(无凭据)。
type agentIdentity struct {
Project string `json:"project"`
SourceApp string `json:"source_app"`
Degraded string `json:"degraded"`
}
func agentIdentityOf(payload json.RawMessage) (agentIdentity, error) {
if len(payload) == 0 {
return agentIdentity{}, errors.New("执行器应答没有身份信息")
}
var document struct {
Identity agentIdentity `json:"agent_identity"`
agentIdentity
}
if err := json.Unmarshal(payload, &document); err != nil {
return agentIdentity{}, errors.New("执行器身份无法解析")
}
// health 把身份摊平在顶层;封存结果把它放在 agent_identity 下。
if document.Identity.Project != "" {
return document.Identity, nil
}
return document.agentIdentity, nil
}
@@ -0,0 +1,98 @@
package main
import (
"context"
"encoding/json"
"net"
"os"
"path/filepath"
"strings"
"testing"
"time"
"knowledge-governance/internal/libraryagent"
)
func shortDir(t *testing.T) string {
t.Helper()
// unix socket 路径有长度上限,测试目录要短。
dir, err := os.MkdirTemp("/tmp", "kgcap-")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
return dir
}
func fakeAgent(t *testing.T, payload map[string]any) string {
t.Helper()
socket := filepath.Join(shortDir(t), "agent.sock")
listener, err := net.Listen("unix", socket)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { listener.Close() })
go func() {
for {
conn, acceptErr := listener.Accept()
if acceptErr != nil {
return
}
var request map[string]any
_ = json.NewDecoder(conn).Decode(&request)
body, _ := json.Marshal(payload)
_ = json.NewEncoder(conn).Encode(map[string]any{"ok": true, "action": "health", "payload": json.RawMessage(body)})
conn.Close()
}
}()
return socket
}
// 心跳说"可用"不等于执行时还是那个实例:执行前的探测必须能挡住换过的 socket。
func TestCaptureExecutorPrecheckRejectsAWrongInstance(t *testing.T) {
socket := fakeAgent(t, map[string]any{"project": "weknora071full", "source_app": "WKNFull-app"})
client := libraryagent.ClientConfig{Socket: socket, Timeout: 2 * time.Second}
err := captureExecutorPrecheck(context.Background(), client, "kgmaint-weknora", 2*time.Second)
if err == nil || !strings.Contains(err.Error(), "指向的实例不匹配") {
t.Fatalf("err = %v", err)
}
if !strings.Contains(err.Error(), "weknora071full") {
t.Fatalf("err should name the actual agent project: %v", err)
}
}
func TestCaptureExecutorPrecheckAcceptsTheExpectedInstance(t *testing.T) {
socket := fakeAgent(t, map[string]any{"project": "kgmaint-weknora", "source_app": "kgmaint-app"})
client := libraryagent.ClientConfig{Socket: socket, Timeout: 2 * time.Second}
if err := captureExecutorPrecheck(context.Background(), client, "kgmaint-weknora", 2*time.Second); err != nil {
t.Fatal(err)
}
// 没有期望身份就不许停写入者:那不是"跳过检查",是不能动生产。
if err := captureExecutorPrecheck(context.Background(), client, "", 2*time.Second); err == nil ||
!strings.Contains(err.Error(), "LIBRARY_SOURCE_PROJECT") {
t.Fatalf("empty expectation accepted: %v", err)
}
// 降级同样挡住。
degraded := fakeAgent(t, map[string]any{"project": "kgmaint-weknora", "degraded": "写入者恢复失败"})
if err := captureExecutorPrecheck(context.Background(), libraryagent.ClientConfig{Socket: degraded, Timeout: 2 * time.Second},
"kgmaint-weknora", 2*time.Second); err == nil || !strings.Contains(err.Error(), "降级") {
t.Fatalf("degraded executor accepted: %v", err)
}
}
// 执行后复核:结果里自报的身份不符,这次封存就不能被登记成测试封存。
func TestCaptureExecutorPostcheckUsesTheIdentityInTheResult(t *testing.T) {
good := json.RawMessage(`{"snapshot_id":"abc","agent_identity":{"project":"kgmaint-weknora"}}`)
if err := captureExecutorPostcheck(good, "kgmaint-weknora"); err != nil {
t.Fatalf("matching identity rejected: %v", err)
}
swapped := json.RawMessage(`{"snapshot_id":"abc","agent_identity":{"project":"weknora071full"}}`)
err := captureExecutorPostcheck(swapped, "kgmaint-weknora")
if err == nil || !strings.Contains(err.Error(), "另一个实例") {
t.Fatalf("err = %v", err)
}
if err := captureExecutorPostcheck(json.RawMessage(`{"snapshot_id":"abc"}`), "kgmaint-weknora"); err == nil ||
!strings.Contains(err.Error(), "没有执行器身份") {
t.Fatalf("missing identity accepted: %v", err)
}
}
+52
View File
@@ -0,0 +1,52 @@
package main
import (
"context"
"log"
"time"
"knowledge-governance/internal/appconfig"
"knowledge-governance/internal/libraryexecutor"
postgresstore "knowledge-governance/internal/platform/postgres"
)
// 为什么由 worker 观测、页面读库:agent socket 只挂载在 worker 容器里,API 那个
// LIBRARY_AGENT_SOCKET 即使配了也 dial 不到。谁真能执行就由谁写下观测,
// 否则页面上的"可发起"只是配置字符串的回声。
const libraryExecutorProbeInterval = 30 * time.Second
// libraryExecutorProbeTimeout 是一次 health 往返的上限:它只是同 uid unix socket
// 上的一次 JSON 往返,慢到秒级就说明执行器有问题,不该拖住心跳。
const libraryExecutorProbeTimeout = 3 * time.Second
func recordLibraryExecutorStatus(ctx context.Context, store *postgresstore.Store, config appconfig.Config) {
// 归属用**封存源**实例:agent 冻结的是源库写入者,和同步目标(kgseal)不是一套。
instance := config.SourceInstance()
if instance == "" {
// 没有实例身份,观测就无处归属:写进去只会让页面拿别的实例的状态。
log.Print("library executor probe skipped: LIBRARY_SOURCE_INSTANCE is not configured")
return
}
observation := libraryexecutor.Probe(ctx, instance, config.LibraryAgentSocket, libraryExecutorProbeTimeout, nil)
if err := store.UpsertLibraryExecutorStatus(ctx, observation, config.WorkerID, time.Now().UTC()); err != nil {
log.Printf("library executor status not recorded: %v", err)
}
}
// runLibraryExecutorProbe 先观测一次再进循环:页面在 worker 刚起来时就该有真实状态。
func runLibraryExecutorProbe(ctx context.Context, store *postgresstore.Store, config appconfig.Config, interval time.Duration) {
if interval <= 0 {
interval = libraryExecutorProbeInterval
}
recordLibraryExecutorStatus(ctx, store, config)
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
recordLibraryExecutorStatus(ctx, store, config)
}
}
}
+17 -16
View File
@@ -8,7 +8,6 @@ import (
"os"
"os/signal"
"path/filepath"
"strings"
"syscall"
"time"
@@ -58,21 +57,6 @@ func (r libraryStageRecorder) Stage(ctx context.Context, stage, state, detail st
func int64Ptr(value int64) *int64 { return &value }
// redactDSN 只保留库名,避免把连接串写进维护记录。
func redactDSN(dsn string) string {
if dsn == "" {
return ""
}
if index := strings.LastIndex(dsn, "/"); index >= 0 && index+1 < len(dsn) {
name := dsn[index+1:]
if cut := strings.Index(name, "?"); cut >= 0 {
name = name[:cut]
}
return name
}
return "configured"
}
// stageSize 是某一类数据在快照里的真实总量(行数/字节)。
type stageSize struct {
count int64
@@ -205,6 +189,8 @@ func run(ctx context.Context, getenv func(string) string) error {
go runRecordsSyncScheduler(ctx, mirrorStore, func(error) { log.Print("QA local mirror scheduler unavailable; last committed data retained") })
}
go weeklyMailLoop(ctx, pool, getenv)
// 封存执行器状态:worker 是唯一能连 agent socket 的一方,它写观测、页面读观测。
go runLibraryExecutorProbe(ctx, store, config, libraryExecutorProbeInterval)
go runSourceRefreshScheduler(ctx, store, config.SourceRefreshSchedulerInterval, func(err error) {
log.Printf("worker %s source refresh scheduler error: %v", config.WorkerID, err)
})
@@ -284,6 +270,13 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", "dead letter release storage unavailable", "", "", "", time.Now().UTC())
return fmt.Errorf("load dead letter releases: %w", releaseErr)
}
// 心跳只能证明"刚才"是哪个实例:动写入者之前再核一次身份。
if identityErr := captureExecutorPrecheck(ctx, agentConfig, config.SourceProject(), 5*time.Second); identityErr != nil {
_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, identityErr.Error(), time.Now().UTC())
_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", identityErr.Error(),
"未停任何写入者;修正执行器身份后重试", "", "", time.Now().UTC())
return identityErr
}
result, err := libraryagent.CaptureReleased(ctx, agentConfig, payload.Scope, payload.CaptureJob, released)
finishedAt := time.Now().UTC()
if err != nil {
@@ -294,6 +287,14 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
log.Printf("library capture failed scope=%s capture_job=%s: %v", payload.Scope, payload.CaptureJob, err)
return err
}
// 执行后复核:结果里 agent 自报的身份必须还是期望实例,否则这次封存
// 可能来自别的栈,绝不能登记成"本实例的测试封存"。
if identityErr := captureExecutorPostcheck(result.Payload, config.SourceProject()); identityErr != nil {
_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, identityErr.Error(), finishedAt)
_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", identityErr.Error(),
"快照已产生但来源身份不符,未登记为测试封存;核对执行器后重新封存", "", "", finishedAt)
return identityErr
}
// 封存产物成版本记录:只登记"测试封存",绝不写成生产生效版本。
versionErr := store.RecordLibraryVersion(ctx, governance.LibraryVersion{
ID: governance.LibraryVersionID(governance.LibraryVersionTestSealed, payload.Scope, result.SnapshotID),
+26 -5
View File
@@ -99,7 +99,10 @@ type Config struct {
LibraryTargetInstance string
// LibrarySourceInstance 是死信解除记录绑定的"来源实例"名字(封存源,不是同步目标)。
// 留空时回落到 LibraryTargetInstance,保持单实例部署的旧行为。
LibrarySourceInstance string
LibrarySourceInstance string
// LibrarySourceProject 是封存执行器应该指向的实例名(agent -project)。
// 留空时等于来源实例名:环境里两者本来就同名(kgmaint-weknora)。
LibrarySourceProject string
LibraryTargetProject string
LibraryTargetWriters []string
LibraryTargetNonWriters map[string]string
@@ -115,10 +118,13 @@ type Config struct {
LibraryTargetJournalRoot string
// LibraryAgentSocket 是主机侧封存 agent 的 unix socket(只由 worker 连接)。
// 留空即"未接线":页面不显示可用的封存入口,接口返回原因而不是假装已排队。
LibraryAgentSocket string
VersionHistoryLinkOrigin string
VersionHistoryHTTPTimeout time.Duration
VersionHistoryMaxPages int
LibraryAgentSocket string
// LibraryExecutorStatusMaxAge 是执行器状态的有效期:超过它就是"无法确认",
// 不是"仍然可用"。worker 每 30 秒观测一次,默认 120 秒足够容错。
LibraryExecutorStatusMaxAge time.Duration
VersionHistoryLinkOrigin string
VersionHistoryHTTPTimeout time.Duration
VersionHistoryMaxPages int
}
func Load(getenv func(string) string) (Config, error) {
@@ -203,6 +209,7 @@ func Load(getenv func(string) string) (Config, error) {
LibraryProductionS3SecretKey: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_SECRET_KEY")),
LibraryTargetInstance: strings.TrimSpace(getenv("LIBRARY_TARGET_INSTANCE")),
LibrarySourceInstance: strings.TrimSpace(getenv("LIBRARY_SOURCE_INSTANCE")),
LibrarySourceProject: strings.TrimSpace(getenv("LIBRARY_SOURCE_PROJECT")),
LibraryTargetProject: strings.TrimSpace(getenv("LIBRARY_TARGET_PROJECT")),
LibraryTargetWriters: splitList(getenv("LIBRARY_TARGET_WRITERS")),
LibraryTargetNonWriters: splitDeclarations(getenv("LIBRARY_TARGET_NON_WRITERS")),
@@ -219,6 +226,7 @@ func Load(getenv func(string) string) (Config, error) {
LibraryCapacityBudgetBytes: capacityBudget,
LibraryRetentionVersions: retentionVersions,
LibraryRetentionDays: retentionDays,
LibraryExecutorStatusMaxAge: 2 * time.Minute,
VersionHistoryLinkOrigin: strings.TrimSpace(getenv("VERSION_HISTORY_LINK_ORIGIN")),
VersionHistoryHTTPTimeout: 15 * time.Second,
VersionHistoryMaxPages: valueOrDefaultInt(getenv("VERSION_HISTORY_MAX_PAGES"), 200),
@@ -241,6 +249,9 @@ func Load(getenv func(string) string) (Config, error) {
if config.SourceRefreshSchedulerInterval < 5*time.Second {
return Config{}, errors.New("SOURCE_REFRESH_SCHEDULER_INTERVAL must be at least 5s")
}
if config.LibraryExecutorStatusMaxAge, err = durationOrDefault(getenv("LIBRARY_EXECUTOR_STATUS_MAX_AGE"), config.LibraryExecutorStatusMaxAge); err != nil {
return Config{}, fmt.Errorf("LIBRARY_EXECUTOR_STATUS_MAX_AGE: %w", err)
}
if raw := getenv("RELEASE_WORKER_ENABLED"); raw != "" {
config.ReleaseWorkerEnabled, err = strconv.ParseBool(raw)
if err != nil {
@@ -575,3 +586,13 @@ func (config Config) SourceInstance() string {
}
return config.LibraryTargetInstance
}
// SourceProject 是封存执行器自报 project 的期望值。它描述的是**封存源**
// (agent 冻结哪一套写入者),不是同步目标:拿目标实例去比会把正确的执行器判成
// 不匹配(本机候选栈就是这种形状:源 kgmaint、目标 kgseal)。
func (config Config) SourceProject() string {
if config.LibrarySourceProject != "" {
return config.LibrarySourceProject
}
return config.SourceInstance()
}
+11 -7
View File
@@ -26,13 +26,6 @@ type libraryCaptureRequest struct {
}
func libraryCaptureEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
if config.LibraryAgentSocket == "" {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{
"code": "library_capture_unwired",
"reason": "未接线:主机侧封存 agent 没有挂载,页面无法发起封存",
})
return
}
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"})
@@ -52,6 +45,17 @@ func libraryCaptureEnqueue(w http.ResponseWriter, r *http.Request, config Config
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
+5 -26
View File
@@ -17,31 +17,11 @@ func captureRequest(body string) *http.Request {
return request
}
// An unwired agent must refuse clearly instead of queueing a job nobody can run.
func TestLibraryCaptureRefusedWhenAgentUnwired(t *testing.T) {
store := &fakeJobStore{}
recorder := httptest.NewRecorder()
libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"internal"}`), Config{Store: store})
if recorder.Code != http.StatusServiceUnavailable {
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
}
var body map[string]any
if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
t.Fatal(err)
}
if body["code"] != "library_capture_unwired" || body["reason"] == "" {
t.Fatalf("unexpected body %v", body)
}
if store.enqueued.Type != "" {
t.Fatal("unwired agent still enqueued a job")
}
}
func TestLibraryCaptureEnqueuesFixedJobShape(t *testing.T) {
store := &fakeJobStore{enqueueEchoRequestedID: true}
recorder := httptest.NewRecorder()
libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"external","job":"lib-external-review"}`),
Config{Store: store, LibraryAgentSocket: "/run/library-agent/agent.sock"})
captureReadyConfig(store))
if recorder.Code != http.StatusAccepted {
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
}
@@ -76,7 +56,7 @@ func TestLibraryCaptureRejectsInvalidRequest(t *testing.T) {
} {
store := &fakeJobStore{}
recorder := httptest.NewRecorder()
libraryCaptureEnqueue(recorder, captureRequest(body), Config{Store: store, LibraryAgentSocket: "/tmp/agent.sock"})
libraryCaptureEnqueue(recorder, captureRequest(body), captureReadyConfig(store))
if recorder.Code != http.StatusBadRequest {
t.Fatalf("body %q status=%d", body, recorder.Code)
}
@@ -92,8 +72,7 @@ func TestLibraryCaptureQueueIDFitsTheColumn(t *testing.T) {
for _, scope := range []string{"internal", "external"} {
store := &fakeJobStore{enqueueEchoRequestedID: true}
recorder := httptest.NewRecorder()
libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"`+scope+`"}`),
Config{Store: store, LibraryAgentSocket: "/tmp/agent.sock"})
libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"`+scope+`"}`), captureReadyConfig(store))
if recorder.Code != http.StatusAccepted {
t.Fatalf("%s status=%d body=%s", scope, recorder.Code, recorder.Body.String())
}
@@ -146,7 +125,7 @@ func TestLibraryCaptureRejectsScopeConflictForReusedJob(t *testing.T) {
Payload: json.RawMessage(`{"scope":"internal","capture_job":"lib-shared","requested_by":"yaozhanghao"}`)}}
recorder := httptest.NewRecorder()
libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"external","job":"lib-shared"}`),
Config{Store: store, LibraryAgentSocket: "/run/library-agent/agent.sock"})
captureReadyConfig(store))
if recorder.Code != http.StatusConflict {
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
}
@@ -166,7 +145,7 @@ func TestLibraryCaptureReplayReportsStoredJob(t *testing.T) {
Payload: json.RawMessage(`{"scope":"internal","capture_job":"lib-shared","requested_by":"yaozhanghao"}`)}}
recorder := httptest.NewRecorder()
libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"internal","job":"lib-shared"}`),
Config{Store: store, LibraryAgentSocket: "/run/library-agent/agent.sock"})
captureReadyConfig(store))
if recorder.Code != http.StatusAccepted {
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
}
+77
View File
@@ -0,0 +1,77 @@
package httpapi
import (
"context"
"time"
"knowledge-governance/internal/libraryexecutor"
)
// libraryExecutorView 是"封存执行器到底能不能用"的页面口径。
//
// 为什么不是 `LIBRARY_AGENT_SOCKET != ""`:正式栈里 API 根本没有挂载那个 socket,
// 配置非空并不代表链路通;更危险的是 agent 可能顶着正式单元名却在冻结另一个实例。
// 这里只认 worker 写入的真实观测,并把观测时间与执行器自报的身份一起给出来。
type libraryExecutorView struct {
Available bool `json:"available"`
Reason string `json:"reason,omitempty"`
Instance string `json:"instance,omitempty"`
Project string `json:"project,omitempty"`
SourceApp string `json:"source_app,omitempty"`
ArchiveRoot string `json:"archive_root,omitempty"`
Busy bool `json:"busy"`
Degraded string `json:"degraded,omitempty"`
ObservedAt *time.Time `json:"observed_at,omitempty"`
}
type libraryExecutorStore interface {
LibraryExecutorStatus(ctx context.Context, instance string) (*libraryexecutor.Observation, error)
}
// libraryExecutorAvailability 把"配置里的期望"与"worker 观测到的真实身份"合起来判定:
// 观测不存在、过期、不可达、实例不符、降级、正忙,任何一条都不可用,并给出原因。
func libraryExecutorAvailability(ctx context.Context, config Config) (bool, string, libraryExecutorView) {
// 期望身份是**封存源**:agent 冻结的是源库写入者,与同步目标(LIBRARY_TARGET_*)
// 本来就是两套名字(本机候选栈:源 kgmaint、目标 kgseal)。
instance := config.LibrarySourceInstance
view := libraryExecutorView{Instance: instance}
if instance == "" {
return false, "未配置封存来源实例身份(LIBRARY_SOURCE_INSTANCE),无法确认执行器归属", view
}
store, ok := config.Store.(libraryExecutorStore)
if !ok {
return false, "执行器状态存储不可用", view
}
observation, err := store.LibraryExecutorStatus(ctx, instance)
if err != nil {
return false, "读取执行器状态失败:" + err.Error(), view
}
if observation == nil {
return false, "封存执行器还没有上报状态(worker 未运行,或实例身份与 worker 不一致)", view
}
available, reason := libraryexecutor.Evaluate(*observation,
libraryexecutor.Expectation{Instance: instance, Project: config.LibrarySourceProject,
ArchiveRoot: config.LibraryArchiveRoot},
executorStatusMaxAge(config), time.Now().UTC())
observedAt := observation.ObservedAt
view.Available = available
view.Reason = reason
view.Project = observation.Identity.Project
view.SourceApp = observation.Identity.SourceApp
view.ArchiveRoot = observation.Identity.ArchiveRoot
view.Busy = observation.Busy
view.Degraded = observation.Degraded
view.ObservedAt = &observedAt
return available, reason, view
}
// defaultLibraryExecutorStatusMaxAge 与 worker 的探测间隔(30s)配套:
// 没配上限绝不等于"永不过期"——那会把一次陈旧的"可用"永久留在页面上。
const defaultLibraryExecutorStatusMaxAge = 2 * time.Minute
func executorStatusMaxAge(config Config) time.Duration {
if config.LibraryExecutorStatusMaxAge > 0 {
return config.LibraryExecutorStatusMaxAge
}
return defaultLibraryExecutorStatusMaxAge
}
+184
View File
@@ -0,0 +1,184 @@
package httpapi
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"knowledge-governance/internal/libraryexecutor"
)
// fakeJobStore 只是"存储"替身;执行器状态同样是它的一条记录。
func (s *fakeJobStore) LibraryExecutorStatus(_ context.Context, instance string) (*libraryexecutor.Observation, error) {
if s.executorStatus == nil || s.executorStatus.Instance != instance {
return nil, nil
}
return s.executorStatus, nil
}
// healthyExecutor 返回一份"worker 刚观测到、指向本实例"的记录。
func healthyExecutor(now time.Time) *libraryexecutor.Observation {
return &libraryexecutor.Observation{
Instance: "kgmaint-weknora", Reachable: true, ObservedAt: now,
Identity: libraryexecutor.Identity{Project: "kgmaint-weknora", SourceApp: "kgmaint-app",
ArchiveRoot: "/tmp/kg-e2e/archive", Writers: "app"},
}
}
// captureReadyConfig 是"执行器真的可用"时的配置:实例、期望项目与观测三者一致。
func captureReadyConfig(store *fakeJobStore) Config {
store.executorStatus = healthyExecutor(time.Now().UTC())
return Config{
Store: store,
LibraryAgentSocket: "/run/library-agent/agent.sock",
LibrarySourceInstance: "kgmaint-weknora",
LibrarySourceProject: "kgmaint-weknora",
LibraryArchiveRoot: "/var/lib/knowledge-governance/library",
}
}
func decodeCaptureRefusal(t *testing.T, recorder *httptest.ResponseRecorder) (string, string) {
t.Helper()
var body struct {
Code string `json:"code"`
Reason string `json:"reason"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
t.Fatalf("body %s: %v", recorder.Body.String(), err)
}
return body.Code, body.Reason
}
// 页面上的"可发起"必须等于接口真的会接受:配置了 socket 但执行器不可达、
// 指向别的实例、降级或过期时,都必须带着原因拒绝,而不是先排队再失败。
func TestLibraryCaptureRefusesWhenExecutorIsNotVerified(t *testing.T) {
now := time.Now().UTC()
tests := []struct {
name string
config func(*Config)
text string
}{
{name: "no instance identity", config: func(c *Config) {
c.LibrarySourceInstance = ""
}, text: "LIBRARY_SOURCE_INSTANCE"},
{name: "nothing reported yet", config: func(c *Config) {
c.Store.(*fakeJobStore).executorStatus = nil
}, text: "还没有上报状态"},
{name: "unreachable", config: func(c *Config) {
c.Store.(*fakeJobStore).executorStatus.Reachable = false
c.Store.(*fakeJobStore).executorStatus.Error = "library agent is unreachable"
}, text: "不可达"},
{name: "stale observation", config: func(c *Config) {
c.Store.(*fakeJobStore).executorStatus.ObservedAt = now.Add(-10 * time.Minute)
}, text: "已过期"},
{name: "points at another instance", config: func(c *Config) {
c.Store.(*fakeJobStore).executorStatus.Identity.Project = "weknora071full"
}, text: "指向的实例不匹配"},
{name: "degraded", config: func(c *Config) {
c.Store.(*fakeJobStore).executorStatus.Degraded = "写入者恢复失败"
}, text: "降级"},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
store := &fakeJobStore{}
config := captureReadyConfig(store)
test.config(&config)
recorder := httptest.NewRecorder()
libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"internal"}`), config)
if recorder.Code != http.StatusServiceUnavailable {
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
}
code, reason := decodeCaptureRefusal(t, recorder)
if code != "library_capture_unavailable" {
t.Fatalf("code=%q body=%s", code, recorder.Body.String())
}
if reason == "" || !strings.Contains(reason, test.text) {
t.Fatalf("reason=%q want it to mention %q", reason, test.text)
}
if store.enqueued.Type != "" {
t.Fatal("refused capture still enqueued a job")
}
})
}
}
// The overview must report the same reason the capture endpoint would use.
func TestLibraryOverviewReportsExecutorIdentityAndReason(t *testing.T) {
store := &fakeLibraryStore{}
config := captureReadyConfig(&store.fakeJobStore)
// libraryJobStore 的方法在 fakeLibraryStore 上,执行器状态在嵌入的 fakeJobStore 上。
config.Store = store
recorder := httptest.NewRecorder()
libraryOverviewHandler(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/library/overview?scope=internal", nil), config)
if recorder.Code != http.StatusOK {
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
}
var body struct {
CaptureAvailable bool `json:"capture_available"`
Executor struct {
Available bool `json:"available"`
Instance string `json:"instance"`
Project string `json:"project"`
SourceApp string `json:"source_app"`
ArchiveRoot string `json:"archive_root"`
} `json:"executor"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
t.Fatal(err)
}
if !body.CaptureAvailable || !body.Executor.Available {
t.Fatalf("overview did not report an available executor: %s", recorder.Body.String())
}
if body.Executor.Project != "kgmaint-weknora" || body.Executor.SourceApp != "kgmaint-app" ||
body.Executor.ArchiveRoot != "/tmp/kg-e2e/archive" || body.Executor.Instance != "kgmaint-weknora" {
t.Fatalf("executor view = %+v", body.Executor)
}
store.fakeJobStore.executorStatus.Identity.Project = "weknora071full"
recorder = httptest.NewRecorder()
libraryOverviewHandler(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/library/overview?scope=internal", nil), config)
if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
t.Fatal(err)
}
if body.CaptureAvailable {
t.Fatalf("overview still says available after the executor moved to another instance: %s", recorder.Body.String())
}
}
// "恢复至该版本"必须由真实条件决定:没有 restore 授权就明确说出来。
func TestLibraryRestoreAvailabilityExplainsMissingAuthorization(t *testing.T) {
store := &fakeJobStore{}
config := Config{Store: store, ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true}
recorder := httptest.NewRecorder()
libraryRestoreAvailability(recorder, httptest.NewRequest(http.MethodGet,
"/api/v1/library/restore-availability?scope=internal&snapshot_id=snap-1", nil), config)
if recorder.Code != http.StatusOK {
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
}
var body struct {
Allowed bool `json:"allowed"`
Reasons []string `json:"reasons"`
Authorization struct {
State string `json:"state"`
} `json:"authorization"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
t.Fatal(err)
}
if body.Allowed || body.Authorization.State != "none" || len(body.Reasons) == 0 {
t.Fatalf("body=%s", recorder.Body.String())
}
found := false
for _, reason := range body.Reasons {
if strings.Contains(reason, "恢复授权") {
found = true
}
}
if !found {
t.Fatalf("reasons = %v", body.Reasons)
}
}
+5 -6
View File
@@ -307,6 +307,7 @@ type libraryOverview struct {
Authorization libraryAuthorizationView `json:"authorization"`
CaptureAvailable bool `json:"capture_available"`
CaptureReason string `json:"capture_reason,omitempty"`
Executor libraryExecutorView `json:"executor"`
Now time.Time `json:"now"`
}
@@ -563,12 +564,10 @@ func libraryOverviewHandler(w http.ResponseWriter, r *http.Request, config Confi
}
now := time.Now().UTC()
overview := libraryOverview{Scope: scope, Jobs: []governance.LibraryJob{}, Now: now,
CaptureAvailable: strings.TrimSpace(config.LibraryAgentSocket) != "",
Gate: libraryGateView{Reasons: []string{}},
Diff: libraryDiffView{State: "unknown", Reason: "生产与测试的文档级差异读取未接线", Added: []any{}, Updated: []any{}, Removed: []any{}}}
if !overview.CaptureAvailable {
overview.CaptureReason = "主机侧封存 agent 未接线"
}
Gate: libraryGateView{Reasons: []string{}},
Diff: libraryDiffView{State: "unknown", Reason: "生产与测试的文档级差异读取未接线", Added: []any{}, Updated: []any{}, Removed: []any{}}}
// 可发起封存 = worker 真实探测到执行器可用且指向本实例;原因逐条给全。
overview.CaptureAvailable, overview.CaptureReason, overview.Executor = libraryExecutorAvailability(r.Context(), config)
// 测试封存来自归档:它只是测试侧事实。
snapshots, snapErr := readLibrarySnapshots(r.Context(), config.LibraryArchiveRoot)
+58
View File
@@ -146,6 +146,64 @@ func libraryRestoreEnqueue(w http.ResponseWriter, r *http.Request, config Config
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())
+4 -5
View File
@@ -85,15 +85,14 @@ const libraryCaptureNote = "封存由运维命令 scripts/library-capture.sh cap
func librarySnapshotsList(w http.ResponseWriter, r *http.Request, config Config) {
view, err := readLibrarySnapshots(r.Context(), config.LibraryArchiveRoot)
view.CaptureAvailable = config.LibraryAgentSocket != ""
if !view.CaptureAvailable {
view.CaptureReason = "未接线:主机侧封存 agent 没有挂载,页面只能查看已封存版本"
}
// 能不能发起封存只认 worker 的真实观测,不认"配置里写了 socket 路径"。
view.CaptureAvailable, view.CaptureReason, _ = libraryExecutorAvailability(r.Context(), config)
if err != nil {
status, reason := libraryArchiveErrorStatus(err, "归档不可读")
writeJSON(w, status, librarySnapshotList{Available: false, Reason: reason,
Snapshots: []librarySnapshotView{}, Verification: libraryVerificationManifest,
VerificationNote: libraryVerificationNote, CaptureNote: libraryCaptureNote})
VerificationNote: libraryVerificationNote, CaptureNote: libraryCaptureNote,
CaptureAvailable: view.CaptureAvailable, CaptureReason: view.CaptureReason})
return
}
writeJSON(w, http.StatusOK, view)
+9
View File
@@ -152,6 +152,11 @@ type Config struct {
LibrarySourceInstance string
// LibraryAgentSocket 是主机侧封存 agent 的 unix socket;留空即页面不能发起封存。
LibraryAgentSocket string
// LibrarySourceProject 是执行器应指向的实例名(agent -project,封存源)。
// 与它不一致时按"封错库"处理:不可发起,并说明差在哪。
LibrarySourceProject string
// LibraryExecutorStatusMaxAge 是执行器观测的有效期;超过即"无法确认"。
LibraryExecutorStatusMaxAge time.Duration
// LibraryCapacityBudgetBytes 是整库归档容量预算;0 = 未设置(容量门禁不放行)。
LibraryCapacityBudgetBytes int64
// 整库生产目标的只读探测配置:首次发布要能证明"生产可读且为空"。
@@ -510,6 +515,10 @@ func NewHandler(config Config) (http.Handler, error) {
mux.HandleFunc("POST /api/v1/library/restores", func(w http.ResponseWriter, r *http.Request) {
libraryRestoreEnqueue(w, r, config)
})
// 页面问"这份快照现在能不能恢复":条件来自双闸、授权与实例维护,而不是写死的按钮。
mux.HandleFunc("GET /api/v1/library/restore-availability", func(w http.ResponseWriter, r *http.Request) {
libraryRestoreAvailability(w, r, config)
})
// 失败/取消作业续作:按原参数重新排队,旧作业保留审计。
mux.HandleFunc("POST /api/v1/library/jobs/{id}/resume", func(w http.ResponseWriter, r *http.Request) {
libraryJobResume(w, r, config)
+27 -24
View File
@@ -15,6 +15,7 @@ import (
"knowledge-governance/internal/governance"
"knowledge-governance/internal/jobs"
"knowledge-governance/internal/libraryexecutor"
postgresstore "knowledge-governance/internal/platform/postgres"
)
@@ -1501,30 +1502,32 @@ type fakeJobStore struct {
activateSettingsErr error
// enqueueEchoRequestedID:幂等"首次入队"语义 —— 返回调用方请求的那个 id
// (默认 false 时返回 enqueueResult,即"复用已有任务"的路径)。
enqueueEchoRequestedID bool
listSettingsConnectionID string
listSettingsErr error
settingsRevisions []governance.SourceConnectionRevision
jobFailureFilter governance.JobFailureFilter
jobFailurePage governance.JobFailurePage
jobFailureErr error
retryBatchRequest *governance.RetryPublicationsBatchRequest
retryBatchOperation string
retryBatchResult governance.RetryPublicationsBatchResult
cancelBatchRequest *governance.CancelPublicationsBatchRequest
cancelBatchOperation string
cancelBatchResult governance.RetryPublicationsBatchResult
cancelBatchReplayed bool
cancelBatchErr error
retryBatchReplayed bool
retryBatchErr error
requeueFailureRequest *governance.RequeueArtifactFetchRequest
requeueFailureOperation string
requeueFailureResult governance.RequeueArtifactFetchResult
requeueFailureReplayed bool
requeueFailureErr error
enqueued jobs.EnqueueRequest
enqueueResult *jobs.Job
enqueueEchoRequestedID bool
listSettingsConnectionID string
listSettingsErr error
settingsRevisions []governance.SourceConnectionRevision
jobFailureFilter governance.JobFailureFilter
jobFailurePage governance.JobFailurePage
jobFailureErr error
retryBatchRequest *governance.RetryPublicationsBatchRequest
retryBatchOperation string
retryBatchResult governance.RetryPublicationsBatchResult
cancelBatchRequest *governance.CancelPublicationsBatchRequest
cancelBatchOperation string
cancelBatchResult governance.RetryPublicationsBatchResult
cancelBatchReplayed bool
cancelBatchErr error
retryBatchReplayed bool
retryBatchErr error
requeueFailureRequest *governance.RequeueArtifactFetchRequest
requeueFailureOperation string
requeueFailureResult governance.RequeueArtifactFetchResult
requeueFailureReplayed bool
requeueFailureErr error
enqueued jobs.EnqueueRequest
enqueueResult *jobs.Job
// executorStatus 是封存执行器的"最近一次观测";nil 表示还没有任何上报。
executorStatus *libraryexecutor.Observation
listResult []jobs.Job
listType string
listByTypeResult []jobs.Job
+100 -14
View File
@@ -89,13 +89,45 @@ const maxReleasedDeadLetters = 64
type Agent struct {
config Config
mu sync.Mutex
// stateMu 只保护下面两个"给人看"的字段:健康应答不能被一次 15 分钟的封存
// 卡住(否则页面会把"正在封存"读成"执行器不可达")。
stateMu sync.RWMutex
// degraded is non-empty when a failed operation could not be followed by a
// verified recovery. While set, capture requests are refused: leaving the
// source writer stopped is worse than refusing work.
degraded string
busy bool
wg sync.WaitGroup
}
// captureState 是健康应答要暴露的执行器状态:是否在跑、是否降级。
type captureState struct {
Busy bool `json:"busy"`
Degraded string `json:"degraded"`
}
func (a *Agent) snapshotState() captureState {
a.stateMu.RLock()
defer a.stateMu.RUnlock()
return captureState{Busy: a.busy, Degraded: a.degraded}
}
func (a *Agent) setBusy(busy bool) {
a.stateMu.Lock()
defer a.stateMu.Unlock()
a.busy = busy
}
func (a *Agent) setDegraded(reason string) {
a.stateMu.Lock()
defer a.stateMu.Unlock()
a.degraded = reason
}
func (a *Agent) clearDegraded() {
a.setDegraded("")
}
func New(config Config) (*Agent, error) {
if !filepath.IsAbs(config.Socket) || !filepath.IsAbs(config.ArchiveRoot) || !filepath.IsAbs(config.JournalRoot) {
return nil, errors.New("library agent paths must be absolute")
@@ -143,7 +175,7 @@ func (a *Agent) Serve(ctx context.Context) error {
// writer from a previous killed run. Recover before accepting work; if that
// fails, stay degraded and refuse captures instead of pretending to be ready.
if err := a.recoverWriters(ctx); err != nil {
a.degraded = err.Error()
a.setDegraded(err.Error())
log.Printf("library agent: pending freeze recovery failed: %v", err)
}
go func() {
@@ -194,13 +226,18 @@ func (a *Agent) handle(ctx context.Context, conn net.Conn) {
// execute serialises operations and maps a request onto the one fixed command.
func (a *Agent) execute(ctx context.Context, req request) response {
// health 不进操作锁、不跑子进程:调用方要的是"这个执行器指向谁、能不能干活",
// 而一次正在进行的封存不该让这个回答变成"不可达"。
if req.Action == "health" {
return a.health()
}
a.mu.Lock()
defer a.mu.Unlock()
out := response{Action: req.Action, Scope: req.Scope, Job: req.Job}
// Degraded blocks new captures, but recover and status must stay reachable:
// otherwise there is no way back to a verified state.
if a.degraded != "" && req.Action == "capture" {
out.Error = "封存执行边界处于降级状态(上次恢复失败:" + a.degraded + "),先执行 recover"
if state := a.snapshotState(); state.Degraded != "" && req.Action == "capture" {
out.Error = "封存执行边界处于降级状态(上次恢复失败:" + state.Degraded + "),先执行 recover"
return out
}
var args []string
@@ -242,6 +279,10 @@ func (a *Agent) execute(ctx context.Context, req request) response {
args = append(args, a.config.declaration()...)
bounded, cancel := context.WithTimeout(ctx, a.config.MaxDuration)
defer cancel()
if req.Action == "capture" || req.Action == "recover" {
a.setBusy(true)
defer a.setBusy(false)
}
payload, err := a.config.run(bounded, a.config.Binary, args...)
if err != nil {
detail := "operation failed"
@@ -254,7 +295,7 @@ func (a *Agent) execute(ctx context.Context, req request) response {
// verify.
if req.Action == "capture" || req.Action == "recover" {
if recoveryErr := a.recoverWriters(ctx); recoveryErr != nil {
a.degraded = recoveryErr.Error()
a.setDegraded(recoveryErr.Error())
out.Error = detail + ";写入者恢复失败:" + recoveryErr.Error() + "(已阻断后续封存,需人工处理)"
return out
}
@@ -266,7 +307,7 @@ func (a *Agent) execute(ctx context.Context, req request) response {
return out
}
if req.Action == "recover" {
a.degraded = ""
a.clearDegraded()
}
out.OK = true
if json.Valid(payload) {
@@ -274,9 +315,28 @@ func (a *Agent) execute(ctx context.Context, req request) response {
} else {
out.Payload = json.RawMessage(`{}`)
}
// 封存结果必须带上"这次是谁执行的":调用方据此复核,防止心跳与执行之间
// socket 被换成另一个 agent(真正的封错库事故就是这样发生的)。
if req.Action == "capture" {
out.Payload = a.attachIdentity(out.Payload)
}
return out
}
// attachIdentity 把执行器身份并进封存结果(保留 CLI 原有的字段,如 snapshot_id)。
func (a *Agent) attachIdentity(payload json.RawMessage) json.RawMessage {
var document map[string]any
if err := json.Unmarshal(payload, &document); err != nil {
return payload
}
document["agent_identity"] = a.config.identity()
encoded, err := json.Marshal(document)
if err != nil {
return payload
}
return json.RawMessage(encoded)
}
// recoverWriters runs the idempotent recovery command with its own budget, even
// when the triggering context is already cancelled (timeout / shutdown).
func (a *Agent) recoverWriters(ctx context.Context) error {
@@ -285,7 +345,7 @@ func (a *Agent) recoverWriters(ctx context.Context) error {
if _, err := a.config.run(bounded, a.config.Binary, append([]string{"recover"}, a.config.declaration()...)...); err != nil {
return err
}
a.degraded = ""
a.clearDegraded()
return nil
}
@@ -332,12 +392,38 @@ func (c Config) run(ctx context.Context, name string, args ...string) ([]byte, e
// Describe returns a redacted view for logs and health checks.
func (a *Agent) Describe() map[string]any {
return map[string]any{
"socket": a.config.Socket,
"archive_root": a.config.ArchiveRoot,
"journal_root": a.config.JournalRoot,
"project": a.config.Project,
"writers": a.config.Writers,
"scope_choice": "request may select internal or external only",
}
view := a.config.identity()
view["scope_choice"] = "request may select internal or external only"
return view
}
// health 是"这个执行器指向谁、现在能不能干活"的唯一回答。它只暴露容器名、
// 路径与声明,不含任何凭据;调用方据此核对源身份,而不是靠配置猜。
func (a *Agent) health() response {
state := a.snapshotState()
payload := a.config.identity()
payload["busy"] = state.Busy
payload["degraded"] = state.Degraded
encoded, err := json.Marshal(payload)
if err != nil {
return response{Action: "health", Error: "cannot encode health"}
}
return response{OK: true, Action: "health", Payload: json.RawMessage(encoded)}
}
// identity 被日志与健康应答共用一份,避免两处描述漂移。
func (c Config) identity() map[string]any {
identity := map[string]any{
"project": c.Project,
"archive_root": c.ArchiveRoot,
"journal_root": c.JournalRoot,
"source_app": c.SourceApp,
"source_pg": c.SourcePG,
"source_es": c.SourceES,
"writers": c.Writers,
}
if c.SourceS3Endpoint != "" {
identity["source_s3_endpoint"] = c.SourceS3Endpoint
}
return identity
}
+6
View File
@@ -46,6 +46,12 @@ func Status(ctx context.Context, config ClientConfig) (Result, error) {
return call(ctx, config, map[string]any{"action": "status"})
}
// Health 只问三件事:这个执行器指向谁(project/容器/归档)、是否在跑、是否降级。
// 它不跑子进程、不占操作锁,所以可以当心跳用;Status 会去查源库,代价高得多。
func Health(ctx context.Context, config ClientConfig) (Result, error) {
return call(ctx, config, map[string]any{"action": "health"})
}
// Recover restores writers from an interrupted freeze journal.
func Recover(ctx context.Context, config ClientConfig) (Result, error) {
return call(ctx, config, map[string]any{"action": "recover"})
+108
View File
@@ -0,0 +1,108 @@
package libraryagent
import (
"context"
"encoding/json"
"os"
"testing"
"time"
)
func serveAgent(t *testing.T, agent *Agent, socket string) {
t.Helper()
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
go func() { _ = agent.Serve(ctx) }()
deadline := time.Now().Add(2 * time.Second)
for {
if _, err := os.Stat(socket); err == nil {
return
}
if time.Now().After(deadline) {
t.Fatal("agent socket never appeared")
}
time.Sleep(10 * time.Millisecond)
}
}
// 健康应答必须不排队等正在跑的封存:一次封存最长 15 分钟,如果 health 也去抢
// 操作锁,页面就会把"正在封存"看成"执行器不可达",进而误导运维。
func TestHealthAnswersWhileACaptureHoldsTheOperationLock(t *testing.T) {
agent, socket, calls := testAgent(t)
serveAgent(t, agent, socket)
// 占住操作锁并置 busy,模拟一次正在进行的封存。
agent.mu.Lock()
defer agent.mu.Unlock()
agent.setBusy(true)
defer agent.setBusy(false)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
result, err := Health(ctx, ClientConfig{Socket: socket, Timeout: 2 * time.Second})
if err != nil {
t.Fatalf("health while busy: %v", err)
}
var payload struct {
Project string `json:"project"`
Busy bool `json:"busy"`
}
if err = json.Unmarshal(result.Payload, &payload); err != nil {
t.Fatalf("health payload: %v", err)
}
if payload.Project != "weknora071full" || !payload.Busy {
t.Fatalf("health payload = %+v", payload)
}
if len(captureCalls(calls())) != 0 {
t.Fatalf("health must not run the capture binary, calls = %v", calls())
}
}
// 降级状态必须能被调用方看见:否则页面只能看到"可发起",而真发起会被拒。
func TestHealthReportsDegradedState(t *testing.T) {
agent, socket, _ := testAgent(t)
serveAgent(t, agent, socket)
// 启动恢复成功会清空降级;这里在它之后模拟"恢复失败"留下的降级状态。
agent.setDegraded("写入者恢复失败")
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
result, err := Health(ctx, ClientConfig{Socket: socket, Timeout: 2 * time.Second})
if err != nil {
t.Fatal(err)
}
var payload struct {
Degraded string `json:"degraded"`
SourceApp string `json:"source_app"`
Writers string `json:"writers"`
}
if err = json.Unmarshal(result.Payload, &payload); err != nil {
t.Fatal(err)
}
if payload.Degraded != "写入者恢复失败" || payload.SourceApp != "WKNFull-app" || payload.Writers != "app" {
t.Fatalf("health payload = %+v", payload)
}
}
// 封存结果必须自带"这次是谁执行的":调用方据此在事后复核,防止心跳与执行之间
// socket 被换成另一个实例。
func TestCaptureResultCarriesTheAgentIdentity(t *testing.T) {
agent, socket, _ := testAgent(t)
serveAgent(t, agent, socket)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
result, err := Capture(ctx, ClientConfig{Socket: socket, Timeout: 5 * time.Second}, "internal", "lib-internal-identity")
if err != nil {
t.Fatal(err)
}
var payload struct {
SnapshotID string `json:"snapshot_id"`
Identity struct {
Project string `json:"project"`
} `json:"agent_identity"`
}
if err = json.Unmarshal(result.Payload, &payload); err != nil {
t.Fatal(err)
}
if payload.SnapshotID != "abc" || payload.Identity.Project != "weknora071full" {
t.Fatalf("payload = %s", result.Payload)
}
}
+122
View File
@@ -0,0 +1,122 @@
// Package libraryexecutor 回答一个问题:主机侧封存执行器现在是不是真的可用,
// 而且是不是指向我们这个实例。
//
// 为什么不能在页面里读一个配置字符串了事:正式栈曾经出现过"agent 顶着正式单元名
// 却在冻结候选实例"的事故——配置非空、页面显示可发起、真发起却封错了库。可用的
// 定义必须是**运行时观测**:socket 通、执行器自报的实例与期望一致、没有降级。
package libraryexecutor
import (
"context"
"encoding/json"
"fmt"
"time"
"knowledge-governance/internal/libraryagent"
)
// Identity 是执行器自报的身份:它要冻结哪个容器的写入者、归档写到哪。
// 只有容器名与路径,没有任何凭据。
type Identity struct {
Project string `json:"project"`
ArchiveRoot string `json:"archive_root"`
JournalRoot string `json:"journal_root"`
SourceApp string `json:"source_app"`
SourcePG string `json:"source_pg"`
SourceES string `json:"source_es"`
Writers string `json:"writers"`
SourceS3Endpoint string `json:"source_s3_endpoint,omitempty"`
}
// Observation 是一次真实探测的结果,会被落库供页面读取。
type Observation struct {
Instance string `json:"instance"`
Reachable bool `json:"reachable"`
Busy bool `json:"busy"`
Degraded string `json:"degraded,omitempty"`
Identity Identity `json:"identity"`
Error string `json:"error,omitempty"`
ObservedAt time.Time `json:"observed_at"`
}
// Expectation 是治理侧对执行器的要求。ArchiveRoot 不参与判定:容器里的路径是
// 挂载后的路径,和宿主机上的真实路径本来就不是一个字符串;用它当门禁只会造成
// 永久误报,所以只作为观测值展示给运维比对。
type Expectation struct {
Instance string
Project string
ArchiveRoot string
}
// Probe 用一次短超时的 health 往返取回执行器身份。它不跑子进程、不碰源库,
// 所以可以当心跳反复调用。instance 是"这次观测属于哪个实例",由调用方的
// LIBRARY_TARGET_INSTANCE 决定。
func Probe(ctx context.Context, instance, socket string, timeout time.Duration, now func() time.Time) Observation {
if now == nil {
now = time.Now
}
observation := Observation{Instance: instance, ObservedAt: now().UTC()}
if socket == "" {
observation.Error = "未配置主机侧封存 agent socket(LIBRARY_AGENT_SOCKET)"
return observation
}
if timeout <= 0 {
timeout = 3 * time.Second
}
result, err := libraryagent.Health(ctx, libraryagent.ClientConfig{Socket: socket, Timeout: timeout})
if err != nil {
observation.Error = err.Error()
return observation
}
var payload struct {
Identity
Busy bool `json:"busy"`
Degraded string `json:"degraded"`
}
if err = json.Unmarshal(result.Payload, &payload); err != nil {
observation.Error = "执行器健康应答无法解析"
return observation
}
observation.Reachable = true
observation.Busy = payload.Busy
observation.Degraded = payload.Degraded
observation.Identity = payload.Identity
return observation
}
// Evaluate 把"观测 + 期望 + 新鲜度"换算成页面上的可用性与原因。
// 返回的原因必须能直接告诉运维该去改什么,而不是"不可用"三个字。
func Evaluate(observation Observation, want Expectation, maxAge time.Duration, now time.Time) (bool, string) {
if want.Instance == "" {
return false, "未配置封存来源实例身份(LIBRARY_SOURCE_INSTANCE),无法确认执行器归属"
}
if observation.Instance != want.Instance {
return false, fmt.Sprintf("执行器状态来自另一个实例(%q),不是 %q", observation.Instance, want.Instance)
}
age := now.Sub(observation.ObservedAt)
if maxAge > 0 && age > maxAge {
return false, fmt.Sprintf("执行器状态已过期:%d 秒前的观测,上限 %d 秒(worker 未上报?)",
int(age.Seconds()), int(maxAge.Seconds()))
}
if !observation.Reachable {
reason := "主机侧封存执行器不可达"
if observation.Error != "" {
reason += ":" + observation.Error
}
return false, reason
}
if want.Project == "" {
return false, "未配置执行器源身份(LIBRARY_SOURCE_PROJECT),无法核对 agent 指向哪个实例"
}
if observation.Identity.Project != want.Project {
return false, fmt.Sprintf("执行器指向的实例不匹配:agent=%q,期望=%q",
observation.Identity.Project, want.Project)
}
if observation.Degraded != "" {
return false, "执行器处于降级状态(上次恢复失败):" + observation.Degraded
}
if observation.Busy {
return false, "执行器正在执行另一个封存/恢复,等它结束再发起"
}
return true, ""
}
+133
View File
@@ -0,0 +1,133 @@
package libraryexecutor
import (
"context"
"encoding/json"
"net"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
func shortSocketDir(t *testing.T) string {
t.Helper()
// unix socket 路径上限 ~104 字节,TMPDIR(本机是 .hermes/cache/scratch)太长,
// 测试里用一个短目录并立刻清理。
dir, err := os.MkdirTemp("/tmp", "kgex-")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
return dir
}
// 可用性必须来自运行时观测,而不是"配置里写了 socket 路径"。
func TestProbeReadsIdentityFromTheAgentOverTheSocket(t *testing.T) {
root := shortSocketDir(t)
socket := filepath.Join(root, "agent.sock")
listener, err := net.Listen("unix", socket)
if err != nil {
t.Fatal(err)
}
defer listener.Close()
go func() {
for {
conn, acceptErr := listener.Accept()
if acceptErr != nil {
return
}
var request map[string]any
_ = json.NewDecoder(conn).Decode(&request)
payload, _ := json.Marshal(map[string]any{
"project": "kgmaint-weknora", "archive_root": "/tmp/kg-e2e/archive",
"journal_root": "/tmp/kg-e2e/journal", "source_app": "kgmaint-app",
"source_pg": "kgmaint-postgres", "source_es": "WKNFull-elasticsearch",
"writers": "app", "busy": true, "degraded": "",
})
_ = json.NewEncoder(conn).Encode(map[string]any{"ok": true, "action": "health", "payload": json.RawMessage(payload)})
conn.Close()
}
}()
observation := Probe(context.Background(), "kgmaint-weknora", socket, 2*time.Second, nil)
if !observation.Reachable {
t.Fatalf("observation = %+v", observation)
}
if observation.Identity.Project != "kgmaint-weknora" || observation.Identity.SourceApp != "kgmaint-app" || !observation.Busy {
t.Fatalf("identity = %+v", observation.Identity)
}
if observation.Instance != "kgmaint-weknora" || observation.ObservedAt.IsZero() {
t.Fatalf("observation = %+v", observation)
}
}
func TestProbeReportsUnreachableAgentWithoutFakingSuccess(t *testing.T) {
observation := Probe(context.Background(), "kgmaint-weknora", filepath.Join(t.TempDir(), "absent.sock"), time.Second, nil)
if observation.Reachable || observation.Error == "" {
t.Fatalf("observation = %+v", observation)
}
available, reason := Evaluate(observation, Expectation{Instance: "kgmaint-weknora", Project: "kgmaint-weknora"},
time.Minute, time.Now().UTC())
if available || !strings.Contains(reason, "不可达") {
t.Fatalf("available=%v reason=%q", available, reason)
}
}
func TestEvaluateRejectsEveryIdentityAndFreshnessProblem(t *testing.T) {
now := time.Date(2026, 10, 29, 8, 0, 0, 0, time.UTC)
healthy := Observation{
Instance: "kgmaint-weknora", Reachable: true, ObservedAt: now,
Identity: Identity{Project: "kgmaint-weknora", SourceApp: "kgmaint-app"},
}
want := Expectation{Instance: "kgmaint-weknora", Project: "kgmaint-weknora"}
if available, reason := Evaluate(healthy, want, 2*time.Minute, now); !available || reason != "" {
t.Fatalf("healthy executor rejected: %v %q", available, reason)
}
tests := []struct {
name string
mut func(*Observation)
want func(Expectation) Expectation
text string
}{
{name: "missing instance config", want: func(Expectation) Expectation { return expectEmptyInstance },
text: "LIBRARY_SOURCE_INSTANCE"},
{name: "row from another instance", mut: func(o *Observation) { o.Instance = "someone-else" },
text: "另一个实例"},
{name: "stale", mut: func(o *Observation) { o.ObservedAt = now.Add(-10 * time.Minute) },
text: "已过期"},
{name: "unreachable", mut: func(o *Observation) { o.Reachable = false; o.Error = "socket missing" },
text: "不可达:socket missing"},
{name: "project not configured", want: func(Expectation) Expectation { return Expectation{Instance: "kgmaint-weknora"} },
text: "LIBRARY_SOURCE_PROJECT"},
{name: "wrong source project", mut: func(o *Observation) { o.Identity.Project = "weknora071full" },
text: "指向的实例不匹配"},
{name: "degraded", mut: func(o *Observation) { o.Degraded = "写入者恢复失败" },
text: "降级状态"},
{name: "busy", mut: func(o *Observation) { o.Busy = true },
text: "正在执行另一个封存"},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
observation := healthy
if test.mut != nil {
test.mut(&observation)
}
expectation := want
if test.want != nil {
expectation = test.want(want)
}
available, reason := Evaluate(observation, expectation, 2*time.Minute, now)
if available {
t.Fatalf("accepted: %+v", observation)
}
if !strings.Contains(reason, test.text) {
t.Fatalf("reason = %q, want it to mention %q", reason, test.text)
}
})
}
}
var expectEmptyInstance = Expectation{Project: "kgmaint-weknora"}
@@ -0,0 +1,71 @@
package postgres
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"knowledge-governance/internal/libraryexecutor"
)
// 为什么落库而不是 API 直接探测:agent socket 只挂在 worker 容器里,API 那个
// LIBRARY_AGENT_SOCKET 即使配了也 dial 不到。谁真能执行就由谁观测,页面读结果。
const executorStatusColumns = `instance, reachable, busy, degraded, identity, status_error, observed_at`
// UpsertLibraryExecutorStatus 记录一次真实探测。降级/不可达同样要写:
// "最新一次观测是什么"比"最后一条好消息"更有用。
func (s *Store) UpsertLibraryExecutorStatus(ctx context.Context, observation libraryexecutor.Observation, observer string, now time.Time) error {
identity, err := json.Marshal(observation.Identity)
if err != nil {
return fmt.Errorf("encode library executor identity: %w", err)
}
if observation.Instance == "" {
return errors.New("library executor observation needs an instance")
}
observedAt := observation.ObservedAt
if observedAt.IsZero() {
observedAt = now
}
_, err = s.db.Exec(ctx, `
INSERT INTO library_executor_status (instance, reachable, busy, degraded, identity, status_error, observer, observed_at)
VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7,$8)
ON CONFLICT (instance) DO UPDATE SET
reachable = EXCLUDED.reachable,
busy = EXCLUDED.busy,
degraded = EXCLUDED.degraded,
identity = EXCLUDED.identity,
status_error = EXCLUDED.status_error,
observer = EXCLUDED.observer,
observed_at = EXCLUDED.observed_at`,
observation.Instance, observation.Reachable, observation.Busy, observation.Degraded,
string(identity), observation.Error, observer, observedAt.UTC())
if err != nil {
return fmt.Errorf("record library executor status: %w", err)
}
return nil
}
// LibraryExecutorStatus 读某个实例最近一次观测;没有记录返回 nil(不是"可用")。
func (s *Store) LibraryExecutorStatus(ctx context.Context, instance string) (*libraryexecutor.Observation, error) {
row := s.db.QueryRow(ctx, `SELECT `+executorStatusColumns+` FROM library_executor_status WHERE instance=$1`, instance)
var observation libraryexecutor.Observation
var raw []byte
err := row.Scan(&observation.Instance, &observation.Reachable, &observation.Busy, &observation.Degraded,
&raw, &observation.Error, &observation.ObservedAt)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("read library executor status: %w", err)
}
if len(raw) > 0 {
if err := json.Unmarshal(raw, &observation.Identity); err != nil {
return nil, fmt.Errorf("library executor identity unreadable: %w", err)
}
}
return &observation, nil
}
@@ -0,0 +1,70 @@
package postgres
import (
"context"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"knowledge-governance/internal/libraryexecutor"
)
// 执行器状态是"最新一次观测",不是"最后一条好消息":
// 降级与不可达必须覆盖上一次的健康记录,否则页面会一直显示可发起。
func TestLibraryExecutorStatusKeepsTheLatestObservation(t *testing.T) {
databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
if databaseURL == "" {
t.Skip("TEST_DATABASE_URL is not set")
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, databaseURL)
if err != nil {
t.Fatalf("connect: %v", err)
}
defer pool.Close()
if err := ApplyMigrations(ctx, pool); err != nil {
t.Fatalf("migrate: %v", err)
}
store := NewStore(pool)
now := time.Date(2026, 10, 29, 8, 0, 0, 0, time.UTC)
if missing, err := store.LibraryExecutorStatus(ctx, "kgmaint-weknora"); err != nil || missing != nil {
t.Fatalf("missing row = %+v, %v", missing, err)
}
healthy := libraryexecutor.Observation{
Instance: "kgmaint-weknora", Reachable: true, ObservedAt: now,
Identity: libraryexecutor.Identity{Project: "kgmaint-weknora", SourceApp: "kgmaint-app", ArchiveRoot: "/tmp/kg-e2e/archive"},
}
if err := store.UpsertLibraryExecutorStatus(ctx, healthy, "worker-1", now); err != nil {
t.Fatalf("upsert healthy: %v", err)
}
read, err := store.LibraryExecutorStatus(ctx, "kgmaint-weknora")
if err != nil {
t.Fatalf("read healthy: %v", err)
}
if read == nil || !read.Reachable || read.Identity.Project != "kgmaint-weknora" || read.Identity.SourceApp != "kgmaint-app" {
t.Fatalf("healthy row = %+v", read)
}
if !read.ObservedAt.Equal(now) {
t.Fatalf("observed_at = %v, want %v", read.ObservedAt, now)
}
degraded := healthy
degraded.Reachable = false
degraded.ObservedAt = now.Add(30 * time.Second)
degraded.Degraded = ""
degraded.Error = "library agent is unreachable"
if err := store.UpsertLibraryExecutorStatus(ctx, degraded, "worker-1", degraded.ObservedAt); err != nil {
t.Fatalf("upsert unreachable: %v", err)
}
read, err = store.LibraryExecutorStatus(ctx, "kgmaint-weknora")
if err != nil || read == nil {
t.Fatalf("read after unreachable: %+v %v", read, err)
}
if read.Reachable || read.Error != "library agent is unreachable" || !read.ObservedAt.Equal(degraded.ObservedAt) {
t.Fatalf("row kept a stale good state: %+v", read)
}
}
@@ -0,0 +1 @@
DROP TABLE IF EXISTS library_executor_status;
@@ -0,0 +1,16 @@
-- 封存执行器的运行时状态:页面上的"可以发起封存"必须来自真实观测,而不是
-- "配置里写了 socket 路径"。正式栈曾出现过 agent 顶着正式单元名却在冻结候选实例:
-- 配置非空、页面显示可发起、真发起却封错了库。
--
-- worker 探测 agent 的 health(不跑子进程、不碰源库),把"它指向哪个实例、
-- 是否降级/正在跑、什么时候观测的"落库;API 只读这张表并核对身份,不再猜。
CREATE TABLE IF NOT EXISTS library_executor_status (
instance VARCHAR(128) PRIMARY KEY,
reachable BOOLEAN NOT NULL DEFAULT FALSE,
busy BOOLEAN NOT NULL DEFAULT FALSE,
degraded TEXT NOT NULL DEFAULT '',
identity JSONB NOT NULL DEFAULT '{}'::jsonb,
status_error TEXT NOT NULL DEFAULT '',
observer VARCHAR(64) NOT NULL DEFAULT '',
observed_at TIMESTAMPTZ NOT NULL
);
+15 -10
View File
@@ -11,40 +11,45 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) {
if err != nil {
t.Fatalf("loadMigrations() error = %v", err)
}
if len(migrations) != 73 {
t.Fatalf("migration count = %d, want 73", len(migrations))
if len(migrations) != 74 {
t.Fatalf("migration count = %d, want 74", len(migrations))
}
// 74 记录封存执行器的运行时状态:页面可用性来自真实观测,不来自配置字符串。
if executor := migrations[len(migrations)-1]; executor.Version != 74 ||
!strings.Contains(executor.UpSQL, "library_executor_status") {
t.Fatalf("migration 74 = %#v", executor)
}
// 73 记录死信精确解除:保留上游行,只绑定指纹、重放引用与观测摘要。
if release := migrations[len(migrations)-1]; release.Version != 73 ||
if release := migrations[len(migrations)-2]; release.Version != 73 ||
!strings.Contains(release.UpSQL, "library_dead_letter_releases") {
t.Fatalf("migration 73 = %#v", release)
}
// 72 让维护占用按实例互斥(internal/external 不再各有各的锁)。
if instance := migrations[len(migrations)-2]; instance.Version != 72 ||
if instance := migrations[len(migrations)-3]; instance.Version != 72 ||
!strings.Contains(instance.UpSQL, "library_instance_maintenance") {
t.Fatalf("migration 72 = %#v", instance)
}
// 71 让维护记录能区分实例级维护与仅治理库互斥。
if level := migrations[len(migrations)-3]; level.Version != 71 ||
if level := migrations[len(migrations)-4]; level.Version != 71 ||
!strings.Contains(level.UpSQL, "level") || !strings.Contains(level.UpSQL, "ingress_state") {
t.Fatalf("migration 71 = %#v", level)
}
if maint := migrations[len(migrations)-4]; maint.Version != 70 || !strings.Contains(maint.UpSQL, "library_maintenance") {
if maint := migrations[len(migrations)-5]; maint.Version != 70 || !strings.Contains(maint.UpSQL, "library_maintenance") {
t.Fatalf("migration 70 = %#v", maint)
}
if tail := migrations[len(migrations)-5]; tail.Version != 69 ||
if tail := migrations[len(migrations)-6]; tail.Version != 69 ||
!strings.Contains(tail.UpSQL, "library_jobs_write_target") {
t.Fatalf("migration 69 = %#v", tail)
}
if restore := migrations[len(migrations)-6]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") {
if restore := migrations[len(migrations)-7]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") {
t.Fatalf("migration 68 = %#v", restore)
}
if last := migrations[len(migrations)-7]; last.Version != 67 ||
if last := migrations[len(migrations)-8]; last.Version != 67 ||
!strings.Contains(last.UpSQL, "library_release_authorizations") ||
!strings.Contains(last.UpSQL, "params_digest") {
t.Fatalf("migration 67 = %#v", last)
}
if flow := migrations[len(migrations)-8]; flow.Version != 66 ||
if flow := migrations[len(migrations)-9]; flow.Version != 66 ||
!strings.Contains(flow.UpSQL, "CREATE TABLE library_versions") ||
!strings.Contains(flow.UpSQL, "library_jobs_active") {
t.Fatalf("migration 66 = %#v", flow)