diff --git a/api/openapi.yaml b/api/openapi.yaml index a07a8a7..aecfad0 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -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": [ diff --git a/cmd/api/main.go b/cmd/api/main.go index 23dad2d..0a652f3 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -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, diff --git a/cmd/worker/library_capture_identity.go b/cmd/worker/library_capture_identity.go new file mode 100644 index 0000000..7110ecc --- /dev/null +++ b/cmd/worker/library_capture_identity.go @@ -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 +} diff --git a/cmd/worker/library_capture_identity_test.go b/cmd/worker/library_capture_identity_test.go new file mode 100644 index 0000000..fcf21bd --- /dev/null +++ b/cmd/worker/library_capture_identity_test.go @@ -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) + } +} diff --git a/cmd/worker/library_executor_status.go b/cmd/worker/library_executor_status.go new file mode 100644 index 0000000..f28c5b2 --- /dev/null +++ b/cmd/worker/library_executor_status.go @@ -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) + } + } +} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 6e5b1eb..d0d6f2c 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -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), diff --git a/internal/appconfig/config.go b/internal/appconfig/config.go index 1bd8e59..19ef36d 100644 --- a/internal/appconfig/config.go +++ b/internal/appconfig/config.go @@ -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() +} diff --git a/internal/httpapi/library_capture.go b/internal/httpapi/library_capture.go index 76fe671..4aefe34 100644 --- a/internal/httpapi/library_capture.go +++ b/internal/httpapi/library_capture.go @@ -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 diff --git a/internal/httpapi/library_capture_test.go b/internal/httpapi/library_capture_test.go index 9b7a41d..2cd6263 100644 --- a/internal/httpapi/library_capture_test.go +++ b/internal/httpapi/library_capture_test.go @@ -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()) } diff --git a/internal/httpapi/library_executor.go b/internal/httpapi/library_executor.go new file mode 100644 index 0000000..3265a9c --- /dev/null +++ b/internal/httpapi/library_executor.go @@ -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 +} diff --git a/internal/httpapi/library_executor_test.go b/internal/httpapi/library_executor_test.go new file mode 100644 index 0000000..7c5da52 --- /dev/null +++ b/internal/httpapi/library_executor_test.go @@ -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) + } +} diff --git a/internal/httpapi/library_flow.go b/internal/httpapi/library_flow.go index b35d328..19f355f 100644 --- a/internal/httpapi/library_flow.go +++ b/internal/httpapi/library_flow.go @@ -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) diff --git a/internal/httpapi/library_restore.go b/internal/httpapi/library_restore.go index 14edde6..3e1239f 100644 --- a/internal/httpapi/library_restore.go +++ b/internal/httpapi/library_restore.go @@ -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()) diff --git a/internal/httpapi/library_snapshots.go b/internal/httpapi/library_snapshots.go index ded1b11..bbdab31 100644 --- a/internal/httpapi/library_snapshots.go +++ b/internal/httpapi/library_snapshots.go @@ -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) diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index caf8b0d..1b1c4cf 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -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) diff --git a/internal/httpapi/server_test.go b/internal/httpapi/server_test.go index b4bfef6..9861428 100644 --- a/internal/httpapi/server_test.go +++ b/internal/httpapi/server_test.go @@ -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 diff --git a/internal/libraryagent/agent.go b/internal/libraryagent/agent.go index 8f9b6a6..65fd3fc 100644 --- a/internal/libraryagent/agent.go +++ b/internal/libraryagent/agent.go @@ -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 } diff --git a/internal/libraryagent/client.go b/internal/libraryagent/client.go index 78dba81..8822718 100644 --- a/internal/libraryagent/client.go +++ b/internal/libraryagent/client.go @@ -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"}) diff --git a/internal/libraryagent/health_test.go b/internal/libraryagent/health_test.go new file mode 100644 index 0000000..d682105 --- /dev/null +++ b/internal/libraryagent/health_test.go @@ -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) + } +} diff --git a/internal/libraryexecutor/executor.go b/internal/libraryexecutor/executor.go new file mode 100644 index 0000000..1bf8fe7 --- /dev/null +++ b/internal/libraryexecutor/executor.go @@ -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, "" +} diff --git a/internal/libraryexecutor/executor_test.go b/internal/libraryexecutor/executor_test.go new file mode 100644 index 0000000..dbfcef6 --- /dev/null +++ b/internal/libraryexecutor/executor_test.go @@ -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"} diff --git a/internal/platform/postgres/library_executor_status.go b/internal/platform/postgres/library_executor_status.go new file mode 100644 index 0000000..a77f98c --- /dev/null +++ b/internal/platform/postgres/library_executor_status.go @@ -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 +} diff --git a/internal/platform/postgres/library_executor_status_integration_test.go b/internal/platform/postgres/library_executor_status_integration_test.go new file mode 100644 index 0000000..8f81388 --- /dev/null +++ b/internal/platform/postgres/library_executor_status_integration_test.go @@ -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) + } +} diff --git a/internal/platform/postgres/migrations/000074_library_executor_status.down.sql b/internal/platform/postgres/migrations/000074_library_executor_status.down.sql new file mode 100644 index 0000000..5441531 --- /dev/null +++ b/internal/platform/postgres/migrations/000074_library_executor_status.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS library_executor_status; \ No newline at end of file diff --git a/internal/platform/postgres/migrations/000074_library_executor_status.up.sql b/internal/platform/postgres/migrations/000074_library_executor_status.up.sql new file mode 100644 index 0000000..cf92a02 --- /dev/null +++ b/internal/platform/postgres/migrations/000074_library_executor_status.up.sql @@ -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 +); \ No newline at end of file diff --git a/internal/platform/postgres/migrations_test.go b/internal/platform/postgres/migrations_test.go index f0c76f1..e15f24e 100644 --- a/internal/platform/postgres/migrations_test.go +++ b/internal/platform/postgres/migrations_test.go @@ -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)