diff --git a/CHANGELOG.md b/CHANGELOG.md index 4bd8c31..fa62e17 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,34 @@ # Changelog +## 开发快照（死信精确解除：保留上游行，只按指纹放行） + +### Added + +- **死信解除契约（治理侧）**：上游 `task_dead_letters` 没有"已解决"标记， + purge/DELETE 都会销毁"这条任务失败过"的唯一证据。新增 + `library_dead_letter_releases`（迁移 073）记录**一条精确死信**的解除：绑定实时行 + 逐字节指纹（`sha256(to_jsonb(row))`，先压紧空白）、重放引用（队列/任务号/时间/ + 证据引用）、源侧观测摘要、操作人与时间。上游行一个字节都不动。 +- `GET/POST /api/v1/library/dead-letter-releases`：POST 需工程师及以上 + + `X-Governance-Action: dead-letter-release`，拒绝条件全部显式——观测摘要不符（被 + 改写/格式化）、源侧未结清、观测里没有这条死信（已被删除）、重放早于失败时间、 + 重放之后同一任务键又出现新死信（说明重放没成功）、实例不一致。 +- `libraryctl dead-letters`（只读）：打印整库死信原始行 + 结算计数 + 观测摘要， + 供治理平台绑定；平台自己不连源库、不复制"自报成功"。 + +### Changed + +- 结算/封存不再"死信存在即阻断"，而是"每条实时死信都要被精确解除命中"： + `checkLibrarySourceSettlementReleased` / `ExportLibraryWithReleases` 对每条行 + 重算指纹，未命中、行被改过、出现新行都继续阻断并点名。 +- 解除指纹经 worker → agent → `libraryctl -released-dead-letters` 逐段校验 + （`sha256:<64 hex>`，最多 64 条，不接受通配），agent 的固定命令形态不变。 + +### Fixed + +- 指纹对空白敏感会让"同一份内容"两边算不出同一个值（PostgreSQL `to_jsonb::text` + 带空格、Go `json.Marshal` 不带）；指纹改为先 `json.Compact` 再摘要。 + ## 开发快照（封存入口接线：库级窗口接上实例级闸门） ### Fixed diff --git a/api/openapi.yaml b/api/openapi.yaml index 50f3501..a07a8a7 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -898,6 +898,109 @@ } } }, + "/api/v1/library/dead-letter-releases": { + "get": { + "tags": [ + "Runtime" + ], + "operationId": "listLibraryDeadLetterReleases", + "summary": "列出本实例的死信解除记录（只读审计视图；上游死信行保持原样）", + "responses": { + "200": { + "$ref": "#/components/responses/Envelope" + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "500": { + "$ref": "#/components/responses/InternalError" + }, + "503": { + "description": "实例身份未配置或解除存储不可用" + } + } + }, + "post": { + "tags": [ + "Runtime" + ], + "operationId": "createLibraryDeadLetterRelease", + "summary": "按精确指纹记录一条死信解除（需要工程师及以上与 X-Governance-Action；不删除上游行，行变化即失效）", + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "dead_letter_id": { + "type": "integer" + }, + "observation": { + "type": "object" + }, + "observation_digest": { + "type": "string" + }, + "replay": { + "type": "object", + "properties": { + "queue": { + "type": "string" + }, + "task_id": { + "type": "string" + }, + "at": { + "type": "string", + "format": "date-time" + }, + "evidence": { + "type": "string" + } + }, + "required": [ + "queue", + "task_id", + "at" + ] + }, + "note": { + "type": "string" + } + }, + "required": [ + "dead_letter_id", + "observation", + "observation_digest", + "replay" + ] + } + } + } + }, + "responses": { + "200": { + "description": "解除记录已落库（或返回同一指纹的已存记录）" + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "403": { + "description": "权限不足或缺少写意图头" + }, + "409": { + "description": "解除被拒绝：指纹/结清/重放证据不满足" + }, + "500": { + "$ref": "#/components/responses/InternalError" + }, + "503": { + "description": "实例身份未配置或解除存储不可用" + } + } + } + }, "/api/v1/library/storage": { "get": { "tags": [ diff --git a/cmd/libraryctl/main.go b/cmd/libraryctl/main.go index c04b8b0..f5c653a 100644 --- a/cmd/libraryctl/main.go +++ b/cmd/libraryctl/main.go @@ -32,6 +32,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" + "knowledge-governance/internal/librarydeadletter" "knowledge-governance/internal/libraryfreeze" "knowledge-governance/internal/librarymaintenance" "knowledge-governance/internal/librarysnapshot" @@ -59,6 +60,9 @@ type options struct { job string dryRun bool snapshot string + // releasedDeadLetters 是治理平台审计过的精确解除指纹（逗号分隔）。它只让这些 + // 确切的行不再阻断；实时行必须逐字节命中，否则照旧拒绝。 + releasedDeadLetters string } func main() { @@ -94,6 +98,8 @@ func main() { runErr = runVerify(opts) case "recover": runErr = runRecover(ctx, opts) + case "dead-letters": + runErr = runDeadLetters(ctx, opts) default: usage() os.Exit(2) @@ -129,6 +135,8 @@ Flags (defaults target the local WKNFull test stack): -job (capture; default: lib--) -dry-run (capture: export + validate only, no archive, no stop) -snapshot (verify) + -released-dead-letters a,b (capture/preflight; 治理平台精确解除的死信指纹) + dead-letters (read-only) 打印整库死信原始行与观测摘要，供治理平台审计 `) } @@ -166,6 +174,7 @@ func parseFlags(command string, args []string) (options, []string, error) { set.StringVar(&opts.job, "job", "", "capture job id") set.BoolVar(&opts.dryRun, "dry-run", false, "export and validate without archiving or stopping") set.StringVar(&opts.snapshot, "snapshot", "", "snapshot id for verify") + set.StringVar(&opts.releasedDeadLetters, "released-dead-letters", "", "audited dead-letter fingerprints (comma separated)") if err := set.Parse(args); err != nil { return opts, nil, err } @@ -662,7 +671,11 @@ func runPreflight(ctx context.Context, opts options) error { if err != nil { return err } - if err = v071.LibrarySourceSettled(ctx, pool); err != nil { + released, err := parseReleasedDeadLetters(opts.releasedDeadLetters) + if err != nil { + return err + } + if err = v071.LibrarySourceSettledWithReleases(ctx, pool, released); err != nil { return err } search, err := v071.NewSearchExporter(config.Search) @@ -717,6 +730,10 @@ func runCapture(ctx context.Context, opts options) error { if opts.scope != "internal" && opts.scope != "external" { return errors.New("-scope must be internal or external") } + released, err := parseReleasedDeadLetters(opts.releasedDeadLetters) + if err != nil { + return err + } if opts.dryRun { return runPreflight(ctx, opts) } @@ -831,7 +848,7 @@ func runCapture(ctx context.Context, opts options) error { v071.LibraryCaptureRequest{JobID: job, Scope: opts.scope, Identity: v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb}, Compatibility: contract, CreatedAt: time.Now().UTC(), MaxDocuments: opts.maxDocuments, - VerifyContract: verifyContract}) + VerifyContract: verifyContract, ReleasedDeadLetters: released}) if err != nil { printJSON(map[string]any{"job": job, "state": result.State, "snapshot_id": result.SnapshotID, "error": err.Error()}) return err @@ -1079,3 +1096,56 @@ func printJSON(value any) { encoder.SetIndent("", " ") _ = encoder.Encode(value) } + +// parseReleasedDeadLetters 把逗号分隔的指纹转成列表，并在本地先校验形状：一个 +// 打错的指纹不应该走到源库才失败，更不应该被当成通配符。 +func parseReleasedDeadLetters(raw string) ([]string, error) { + out := []string{} + for _, item := range strings.Split(raw, ",") { + item = strings.TrimSpace(item) + if item == "" { + continue + } + if !librarydeadletter.ValidFingerprint(item) { + return nil, errors.New("released dead letter entries must be sha256:<64 hex> fingerprints") + } + out = append(out, item) + } + if len(out) > 64 { + return nil, errors.New("at most 64 dead letter fingerprints per capture") + } + return out, nil +} + +// runDeadLetters 只读：打印整库死信原始行与观测摘要。治理平台的解除记录绑定这份 +// 摘要；平台自己不连源库，因此观测必须由能读源库的一方生成并原样提交。 +func runDeadLetters(ctx context.Context, opts options) error { + if opts.scope != "internal" && opts.scope != "external" { + return errors.New("-scope must be internal or external") + } + binding, err := scopeBindings(ctx, opts.governancePG) + if err != nil { + return err + } + config, err := discoverSource(ctx, opts, binding) + if err != nil { + return err + } + pool, err := openSourceDatabase(ctx, config) + if err != nil { + return err + } + defer pool.Close() + kb := config.Scopes[opts.scope] + tenant, err := tenantAndKnowledgeBase(ctx, pool, kb) + if err != nil { + return err + } + payload, err := v071.ObserveDeadLetters(ctx, pool, opts.project, tenant, kb) + if err != nil { + return err + } + digest := librarydeadletter.Fingerprint(payload) + printJSON(map[string]any{"observation": json.RawMessage(payload), "observation_digest": digest}) + return nil +} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 814e7a8..a8a52df 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -276,7 +276,15 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st "主机侧正在暂停写入者、导出并恢复", startedAt); stageErr != nil { log.Printf("library seal job %s stage update failed: %v", libraryJobID, stageErr) } - result, err := libraryagent.Capture(ctx, agentConfig, payload.Scope, payload.CaptureJob) + // 平台审计过的精确死信解除指纹：读取失败必须阻断（fail closed）， + // 不能因为存储不可用就静默按"没有解除"或"全部解除"处理。 + released, releaseErr := store.DeadLetterReleaseFingerprints(ctx, config.LibraryTargetInstance) + if releaseErr != nil { + _ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, releaseErr.Error(), time.Now().UTC()) + _ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", "dead letter release storage unavailable", "", "", "", time.Now().UTC()) + return fmt.Errorf("load dead letter releases: %w", releaseErr) + } + result, err := libraryagent.CaptureReleased(ctx, agentConfig, payload.Scope, payload.CaptureJob, released) finishedAt := time.Now().UTC() if err != nil { // 导出失败时阶段与作业都要留痕；恢复责任在 agent，不会因为作业失败而假装来源已恢复。 diff --git a/internal/governance/library_flow.go b/internal/governance/library_flow.go index 34f10bd..ed8bb7b 100644 --- a/internal/governance/library_flow.go +++ b/internal/governance/library_flow.go @@ -150,3 +150,34 @@ type LibraryReleaseAuthorization struct { func (a LibraryReleaseAuthorization) ValidAt(now time.Time) bool { return a.RevokedAt == nil && now.Before(a.ExpiresAt) } + +// LibraryDeadLetterRelease 是治理平台对一条上游死信的精确解除记录。上游行保持原样， +// 这条记录只说明"这一行已被审计、重放并结清"；指纹一变就不再命中。 +type LibraryDeadLetterRelease struct { + ID string `json:"id"` + Instance string `json:"instance"` + TenantID uint64 `json:"tenant_id"` + KnowledgeBaseID string `json:"kb_id"` + DeadLetterID int64 `json:"dead_letter_id"` + Fingerprint string `json:"fingerprint"` + TaskType string `json:"task_type"` + Scope string `json:"scope"` + ScopeID string `json:"scope_id"` + RelatedID string `json:"related_id"` + ReplayQueue string `json:"replay_queue"` + ReplayTaskID string `json:"replay_task_id"` + ReplayedAt time.Time `json:"replayed_at"` + ReplayEvidence string `json:"replay_evidence"` + ObservationDigest string `json:"observation_digest"` + Observation json.RawMessage `json:"observation"` + Note string `json:"note"` + Actor string `json:"actor"` + CreatedAt time.Time `json:"created_at"` +} + +// LibraryDeadLetterReleaseID 生成解除记录 ID。同一实例同一指纹永远得到同一个 ID， +// 重复提交返回已存记录而不是造第二条审计。 +func LibraryDeadLetterReleaseID(instance, fingerprint string) string { + sum := sha256.Sum256([]byte(instance + "|" + fingerprint)) + return "dldr-" + hex.EncodeToString(sum[:])[:24] +} diff --git a/internal/httpapi/library_deadletters.go b/internal/httpapi/library_deadletters.go new file mode 100644 index 0000000..d879fcd --- /dev/null +++ b/internal/httpapi/library_deadletters.go @@ -0,0 +1,104 @@ +package httpapi + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "time" + + "knowledge-governance/internal/governance" + "knowledge-governance/internal/librarydeadletter" +) + +// 死信解除：上游 task_dead_letters 没有"已解决"标记。这里只写治理库的审计记录， +// 永不删除上游行；解绑的是精确指纹，行一变就不再命中，封存照旧阻断。 +type deadLetterReleaseStore interface { + RecordDeadLetterRelease(context.Context, governance.LibraryDeadLetterRelease) (governance.LibraryDeadLetterRelease, bool, error) + DeadLetterReleases(context.Context, string, int) ([]governance.LibraryDeadLetterRelease, error) +} + +type deadLetterReleaseRequest struct { + DeadLetterID int64 `json:"dead_letter_id"` + Replay librarydeadletter.Replay `json:"replay"` + Note string `json:"note"` + Observation json.RawMessage `json:"observation"` + ObservationDigest string `json:"observation_digest"` +} + +func deadLetterReleaseHandlers(w http.ResponseWriter, r *http.Request, config Config) { + identity, ok := requireSessionIdentity(w, r) + if !ok { + return + } + store, ok := config.Store.(deadLetterReleaseStore) + if !ok { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "死信解除存储不可用"}) + return + } + instance := strings.TrimSpace(config.LibraryTargetInstance) + if instance == "" { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{ + "code": "dead_letter_release_unwired", "reason": "未接线：没有配置来源实例身份，无法绑定解除记录", + }) + return + } + if r.Method == http.MethodGet { + releases, err := store.DeadLetterReleases(r.Context(), instance, 100) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "读取死信解除记录失败"}) + return + } + writeJSON(w, http.StatusOK, map[string]any{"instance": instance, "releases": releases}) + return + } + // 写操作需要工程师及以上，并要求显式意图头：避免普通点击意外解除。 + if roleRank(identity.Role) < roleRank(roleEngineer) { + writeJSON(w, http.StatusForbidden, map[string]string{"error": "forbidden"}) + return + } + if r.Header.Get("X-Governance-Action") != "dead-letter-release" { + writeJSON(w, http.StatusForbidden, map[string]string{"error": "write intent required"}) + return + } + var request deadLetterReleaseRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&request); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"}) + return + } + now := time.Now().UTC() + release, err := librarydeadletter.Verify(librarydeadletter.Request{ + Project: instance, DeadLetterID: request.DeadLetterID, Replay: request.Replay, + Note: request.Note, Observation: request.Observation, Digest: request.ObservationDigest, + }, instance, now) + if err != nil { + writeJSON(w, http.StatusConflict, map[string]any{"code": "dead_letter_release_rejected", "error": err.Error()}) + return + } + stored, reused, err := store.RecordDeadLetterRelease(r.Context(), governance.LibraryDeadLetterRelease{ + ID: governance.LibraryDeadLetterReleaseID(instance, release.Fingerprint), + Instance: instance, + TenantID: release.TenantID, + KnowledgeBaseID: release.KnowledgeBaseID, + DeadLetterID: release.DeadLetterID, + Fingerprint: release.Fingerprint, + TaskType: release.TaskType, + Scope: release.Scope, + ScopeID: release.ScopeID, + RelatedID: release.RelatedID, + ReplayQueue: release.Replay.Queue, + ReplayTaskID: release.Replay.TaskID, + ReplayedAt: release.Replay.At, + ReplayEvidence: release.Replay.Evidence, + ObservationDigest: release.Digest, + Observation: release.Observation, + Note: release.Note, + Actor: identity.Username, + CreatedAt: now, + }) + if err != nil { + writeJSON(w, http.StatusConflict, map[string]any{"code": "dead_letter_release_conflict", "error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, map[string]any{"release": stored, "reused": reused}) +} diff --git a/internal/httpapi/library_deadletters_test.go b/internal/httpapi/library_deadletters_test.go new file mode 100644 index 0000000..75c0769 --- /dev/null +++ b/internal/httpapi/library_deadletters_test.go @@ -0,0 +1,114 @@ +package httpapi + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "knowledge-governance/internal/governance" + "knowledge-governance/internal/librarydeadletter" +) + +type deadLetterStoreFake struct { + fakeJobStore + releases []governance.LibraryDeadLetterRelease + recorded int +} + +func (f *deadLetterStoreFake) RecordDeadLetterRelease(_ context.Context, release governance.LibraryDeadLetterRelease) (governance.LibraryDeadLetterRelease, bool, error) { + f.recorded++ + f.releases = append(f.releases, release) + return release, false, nil +} + +func (f *deadLetterStoreFake) DeadLetterReleases(context.Context, string, int) ([]governance.LibraryDeadLetterRelease, error) { + return f.releases, nil +} + +func deadLetterRequest(t *testing.T, role, action string, body string) *http.Request { + t.Helper() + request := httptest.NewRequest(http.MethodPost, "/api/v1/library/dead-letter-releases", strings.NewReader(body)) + request.Header.Set("Content-Type", "application/json") + if action != "" { + request.Header.Set("X-Governance-Action", action) + } + return request.WithContext(context.WithValue(request.Context(), sessionKey{}, sessionIdentity{Username: "tester", Role: role})) +} + +const deadLetterRow = `{"id": 7, "payload": {}, "scope": "knowledge", "scope_id": "k1", "failed_at": "2026-10-28T10:00:00+00:00", "last_error": "boom", "task_type": "wiki:ingest", "tenant_id": 10001, "fail_count": 3, "related_id": "k1"}` + +func deadLetterBody(t *testing.T, replayAt time.Time) string { + t.Helper() + observation := librarydeadletter.Observation{ + Project: "kgmaint-weknora", TenantID: 10001, KnowledgeBaseID: "kb-1", + CapturedAt: time.Now().UTC(), Rows: []json.RawMessage{json.RawMessage(deadLetterRow)}, + } + raw, err := json.Marshal(observation) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(map[string]any{ + "dead_letter_id": 7, "observation": json.RawMessage(raw), + "observation_digest": librarydeadletter.Fingerprint(raw), + "replay": map[string]any{"queue": "default", "task_id": "task-1", + "at": replayAt.Format(time.RFC3339), "evidence": "run_now 200"}, + }) + if err != nil { + t.Fatal(err) + } + return string(body) +} + +func TestDeadLetterReleaseRequiresIntentAndRole(t *testing.T) { + store := &deadLetterStoreFake{} + config := Config{Store: store, LibraryTargetInstance: "kgmaint-weknora"} + recorder := httptest.NewRecorder() + deadLetterReleaseHandlers(recorder, deadLetterRequest(t, roleViewer, "dead-letter-release", deadLetterBody(t, time.Now().Add(-time.Hour))), config) + if recorder.Code != http.StatusForbidden || store.recorded != 0 { + t.Fatalf("viewer recorded a release: %d %s", recorder.Code, recorder.Body.String()) + } + recorder = httptest.NewRecorder() + deadLetterReleaseHandlers(recorder, deadLetterRequest(t, roleEngineer, "", deadLetterBody(t, time.Now().Add(-time.Hour))), config) + if recorder.Code != http.StatusForbidden || store.recorded != 0 { + t.Fatalf("missing write intent recorded a release: %d", recorder.Code) + } +} + +func TestDeadLetterReleaseRefusesUnwiredInstanceAndBadEvidence(t *testing.T) { + store := &deadLetterStoreFake{} + recorder := httptest.NewRecorder() + deadLetterReleaseHandlers(recorder, deadLetterRequest(t, roleEngineer, "dead-letter-release", + deadLetterBody(t, time.Now().Add(-time.Hour))), Config{Store: store}) + if recorder.Code != http.StatusServiceUnavailable { + t.Fatalf("unwired instance accepted: %d", recorder.Code) + } + // 摘要被改写：必须拒绝，而不是"看着像就放过"。 + config := Config{Store: store, LibraryTargetInstance: "kgmaint-weknora"} + broken := strings.Replace(deadLetterBody(t, time.Now().Add(-time.Hour)), "sha256:", "sha256:0", 1) + recorder = httptest.NewRecorder() + deadLetterReleaseHandlers(recorder, deadLetterRequest(t, roleEngineer, "dead-letter-release", broken), config) + if recorder.Code != http.StatusConflict || store.recorded != 0 { + t.Fatalf("tampered observation accepted: %d", recorder.Code) + } +} + +func TestDeadLetterReleaseRecordsExactFingerprint(t *testing.T) { + store := &deadLetterStoreFake{} + config := Config{Store: store, LibraryTargetInstance: "kgmaint-weknora"} + recorder := httptest.NewRecorder() + deadLetterReleaseHandlers(recorder, deadLetterRequest(t, roleEngineer, "dead-letter-release", + deadLetterBody(t, time.Now().Add(-30*time.Minute))), config) + if recorder.Code != http.StatusOK || store.recorded != 1 { + t.Fatalf("release not recorded: %d %s", recorder.Code, recorder.Body.String()) + } + recorded := store.releases[0] + if recorded.Actor != "tester" || recorded.Instance != "kgmaint-weknora" || + recorded.Fingerprint != librarydeadletter.Fingerprint([]byte(deadLetterRow)) || + recorded.DeadLetterID != 7 || recorded.ReplayTaskID != "task-1" { + t.Fatalf("release lost its audit binding: %+v", recorded) + } +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 5f7fe6c..7b70c85 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -432,6 +432,13 @@ func NewHandler(config Config) (http.Handler, error) { mux.HandleFunc("GET /api/v1/library/captures", func(w http.ResponseWriter, r *http.Request) { libraryCaptureJobs(w, r, config) }) + // 死信解除：只写治理库审计，不删上游行；封存时按精确指纹放行。 + mux.HandleFunc("GET /api/v1/library/dead-letter-releases", func(w http.ResponseWriter, r *http.Request) { + deadLetterReleaseHandlers(w, r, config) + }) + mux.HandleFunc("POST /api/v1/library/dead-letter-releases", func(w http.ResponseWriter, r *http.Request) { + deadLetterReleaseHandlers(w, r, config) + }) // 整库状态：测试封存 / 生产生效 / 生产预备份 三类版本 + 差异 + 持久作业。 // 归档可能很大：加请求级超时，读不动就明确降级，别让页面请求挂死。 mux.HandleFunc("GET /api/v1/library/overview", func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/libraryagent/agent.go b/internal/libraryagent/agent.go index 315b60e..8f9b6a6 100644 --- a/internal/libraryagent/agent.go +++ b/internal/libraryagent/agent.go @@ -63,6 +63,9 @@ type request struct { Action string `json:"action"` Scope string `json:"scope,omitempty"` Job string `json:"job,omitempty"` + // Released 只允许精确死信指纹（sha256:<64 hex>）。它不改变固定命令形态： + // 还能传的只有"这些确切的行已由治理平台审计解除"，CLI 会逐字节复核。 + Released []string `json:"released,omitempty"` } type response struct { @@ -76,6 +79,11 @@ type response struct { var jobPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) +// 死信解除只能按精确指纹传入，不接受通配、实例或任务类型级解除。 +var fingerprintPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) + +const maxReleasedDeadLetters = 64 + // Agent serves capture requests. Only one operation runs at a time: a library // capture owns the writer group, so overlapping requests would race the window. type Agent struct { @@ -214,6 +222,19 @@ func (a *Agent) execute(ctx context.Context, req request) response { if req.Job != "" { args = append(args, "-job", req.Job) } + if len(req.Released) > 0 { + if len(req.Released) > maxReleasedDeadLetters { + out.Error = "too many released dead-letter fingerprints" + return out + } + for _, fingerprint := range req.Released { + if !fingerprintPattern.MatchString(fingerprint) { + out.Error = "released dead letters must be sha256:<64 hex> fingerprints" + return out + } + } + args = append(args, "-released-dead-letters", strings.Join(req.Released, ",")) + } default: out.Error = "unsupported action" return out diff --git a/internal/libraryagent/agent_test.go b/internal/libraryagent/agent_test.go index 11774d1..15e62d5 100644 --- a/internal/libraryagent/agent_test.go +++ b/internal/libraryagent/agent_test.go @@ -380,3 +380,32 @@ func TestAgentPinsTheHostSideObjectEndpoint(t *testing.T) { } } } + +// 死信解除只能按精确指纹透传：形状不对就拒绝，且固定命令其余部分不受影响。 +func TestAgentForwardsOnlyExactDeadLetterFingerprints(t *testing.T) { + agent, socket, calls := testAgent(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = agent.Serve(ctx) }() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if conn, err := net.Dial("unix", socket); err == nil { + conn.Close() + break + } + time.Sleep(10 * time.Millisecond) + } + fingerprint := "sha256:" + strings.Repeat("a", 64) + ok := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"j1","released":["`+fingerprint+`"]}`) + if !ok.OK { + t.Fatalf("exact fingerprint refused: %s", ok.Error) + } + last := calls()[len(calls())-1] + if !strings.Contains(last, "-released-dead-letters "+fingerprint) { + t.Fatalf("fingerprint not forwarded: %s", last) + } + bad := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"j2","released":["*"]}`) + if bad.OK || !strings.Contains(bad.Error, "fingerprints") { + t.Fatalf("wildcard release accepted: %+v", bad) + } +} diff --git a/internal/libraryagent/client.go b/internal/libraryagent/client.go index a1d2fca..78dba81 100644 --- a/internal/libraryagent/client.go +++ b/internal/libraryagent/client.go @@ -29,20 +29,29 @@ type Result struct { // Capture asks the agent to seal one library scope. The job id makes the call // idempotent: repeating it returns the already sealed candidate. func Capture(ctx context.Context, config ClientConfig, scope, job string) (Result, error) { - return call(ctx, config, map[string]string{"action": "capture", "scope": scope, "job": job}) + return CaptureReleased(ctx, config, scope, job, nil) +} + +// CaptureReleased 在 Capture 之上带上治理平台已审计的精确死信解除指纹。 +func CaptureReleased(ctx context.Context, config ClientConfig, scope, job string, released []string) (Result, error) { + request := map[string]any{"action": "capture", "scope": scope, "job": job} + if len(released) > 0 { + request["released"] = released + } + return call(ctx, config, request) } // Status probes the agent (source wiring, writer group, archive usage). func Status(ctx context.Context, config ClientConfig) (Result, error) { - return call(ctx, config, map[string]string{"action": "status"}) + return call(ctx, config, map[string]any{"action": "status"}) } // Recover restores writers from an interrupted freeze journal. func Recover(ctx context.Context, config ClientConfig) (Result, error) { - return call(ctx, config, map[string]string{"action": "recover"}) + return call(ctx, config, map[string]any{"action": "recover"}) } -func call(ctx context.Context, config ClientConfig, request map[string]string) (Result, error) { +func call(ctx context.Context, config ClientConfig, request map[string]any) (Result, error) { if config.Socket == "" { return Result{}, errors.New("library agent socket is not configured") } diff --git a/internal/librarydeadletter/deadletter.go b/internal/librarydeadletter/deadletter.go new file mode 100644 index 0000000..bf52046 --- /dev/null +++ b/internal/librarydeadletter/deadletter.go @@ -0,0 +1,253 @@ +// Package librarydeadletter 定义上游死信（task_dead_letters）的精确解除契约。 +// +// 为什么需要它：WeKnora v0.7.1 的 task_dead_letters 没有"已解决"标记，导出/封存 +// 按"存在即阻断"处理。上游只提供两种运维动作——整队列 purge 与直接 DELETE， +// 两者都会销毁"这条任务失败过"的唯一证据，不能当作产品能力。治理平台因此保留 +// 上游原始行，另外记录一条绑定到精确指纹的解除记录：指纹必须与实时行逐字节一致， +// 行变化或新死信出现都不再命中，继续阻断。 +package librarydeadletter + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strings" + "time" +) + +// FingerprintPrefix 让指纹在日志/审计里自解释：算法一旦更换无需猜。 +const FingerprintPrefix = "sha256:" + +// Fingerprint 对上游用 to_jsonb(row)::text 取回的行文本求摘要。 +// +// 为什么按原始行文本而不是逐字段重组：PostgreSQL 的 jsonb 文本已规范化（键有序）， +// 任何字段变化都会改变字节；逐字段重组会漏掉上游新增列，等于给"行被改过"开后门。 +func Fingerprint(raw []byte) string { + // 先压紧空白：PostgreSQL 的 to_jsonb::text 带空格，Go 的 json.Marshal 不带， + // 两边必须对同一份内容得到同一个指纹，否则解除永远命中不上。 + compacted := raw + if len(raw) > 0 { + var buffer bytes.Buffer + if err := json.Compact(&buffer, raw); err == nil { + compacted = buffer.Bytes() + } + } + sum := sha256.Sum256(compacted) + return FingerprintPrefix + hex.EncodeToString(sum[:]) +} + +// ValidFingerprint 校验调用方传入的指纹形状，避免把通配符或截断值当解除依据。 +func ValidFingerprint(value string) bool { + if !strings.HasPrefix(value, FingerprintPrefix) { + return false + } + body := strings.TrimPrefix(value, FingerprintPrefix) + if len(body) != 64 { + return false + } + _, err := hex.DecodeString(body) + return err == nil +} + +// Row 是 task_dead_letters 的一行。字段与上游 000041 迁移一致；Payload 保留原始 +// JSON 文本，因为任务参数本身就是证据。 +type Row struct { + ID int64 `json:"id"` + TenantID uint64 `json:"tenant_id"` + TaskType string `json:"task_type"` + Scope string `json:"scope"` + ScopeID string `json:"scope_id"` + RelatedID string `json:"related_id"` + Payload json.RawMessage `json:"payload"` + LastError string `json:"last_error"` + FailCount int `json:"fail_count"` + FailedAt time.Time `json:"failed_at"` +} + +// Settlement 只记录与死信无关的两个计数：未完成文档与待办操作。死信单独成表， +// 由指纹逐条解释，不能用"总数"含糊过去。 +type Settlement struct { + Unsettled int64 `json:"unsettled"` + PendingOps int64 `json:"pending_ops"` +} + +// Observation 是主机侧（能读源库的一方）生成的只读观测。治理平台不直接连源库， +// 因此解除记录绑定的是这份观测的摘要：平台重算摘要，换一个字节就不再接受。 +type Observation struct { + Project string `json:"project"` + TenantID uint64 `json:"tenant_id"` + KnowledgeBaseID string `json:"kb_id"` + CapturedAt time.Time `json:"captured_at"` + Settlement Settlement `json:"settlement"` + Rows []json.RawMessage `json:"dead_letters"` +} + +// Replay 记录"真实重放"的引用。平台无法自己向上游发起重放（那需要源侧管理员权限， +// 属于放大授权），所以只接受可核对的引用，并把"重放之后没有再次失败"作为机器检查项： +// 同一任务键在重放时间之后若出现新的死信行，解除无效。 +type Replay struct { + Queue string `json:"queue"` + TaskID string `json:"task_id"` + At time.Time `json:"at"` + Evidence string `json:"evidence"` +} + +// Request 是创建解除记录的输入。Observation 保持原始字节，摘要按这串字节计算。 +type Request struct { + Project string `json:"project"` + TenantID uint64 `json:"tenant_id"` + KnowledgeBaseID string `json:"kb_id"` + DeadLetterID int64 `json:"dead_letter_id"` + Replay Replay `json:"replay"` + Note string `json:"note"` + Observation json.RawMessage `json:"observation"` + Digest string `json:"observation_digest"` +} + +// Release 是验证通过后落库的解除记录。 +type Release struct { + Project string + TenantID uint64 + KnowledgeBaseID string + DeadLetterID int64 + Fingerprint string + TaskType string + Scope string + ScopeID string + RelatedID string + Replay Replay + Observation json.RawMessage + Digest string + Note string +} + +// Verify 检查解除请求是否可被接受。它不信任调用方自报的结论： +// 指纹由观测里的原始行重算，结清计数必须为 0，重放时间必须晚于失败时间， +// 且同一任务键在重放之后不得再有新的失败行。 +func Verify(req Request, expectedProject string, now time.Time) (Release, error) { + if expectedProject == "" { + return Release{}, errors.New("死信解除需要平台侧已配置的实例身份") + } + if req.Project != "" && req.Project != expectedProject { + return Release{}, fmt.Errorf("解除记录实例 %q 与当前实例 %q 不一致", req.Project, expectedProject) + } + if req.DeadLetterID <= 0 { + return Release{}, errors.New("dead_letter_id 必须指定一条精确死信") + } + if len(req.Observation) == 0 || req.Digest == "" { + return Release{}, errors.New("缺少源侧观测或摘要") + } + if got := Fingerprint(req.Observation); got != req.Digest { + return Release{}, errors.New("观测摘要与提交的观测不一致（可能被改写或格式化）") + } + var observation Observation + if err := json.Unmarshal(req.Observation, &observation); err != nil { + return Release{}, errors.New("源侧观测不是合法 JSON") + } + if observation.Project != expectedProject { + return Release{}, fmt.Errorf("观测实例 %q 与当前实例 %q 不一致", observation.Project, expectedProject) + } + if observation.Settlement.Unsettled != 0 || observation.Settlement.PendingOps != 0 { + return Release{}, fmt.Errorf("源侧尚未结清：未完成文档 %d、待办操作 %d", + observation.Settlement.Unsettled, observation.Settlement.PendingOps) + } + if observation.CapturedAt.IsZero() || observation.CapturedAt.After(now.Add(time.Minute)) { + return Release{}, errors.New("源侧观测时间缺失或晚于当前时间") + } + if req.Replay.Queue == "" || req.Replay.TaskID == "" { + return Release{}, errors.New("重放必须给出队列与任务号") + } + if req.Replay.At.IsZero() { + return Release{}, errors.New("重放必须给出实际发生时间") + } + if req.Replay.At.After(now.Add(time.Minute)) { + return Release{}, errors.New("重放时间晚于当前时间") + } + if req.TenantID != 0 && observation.TenantID != 0 && req.TenantID != observation.TenantID { + return Release{}, errors.New("请求租户与观测租户不一致") + } + if req.KnowledgeBaseID != "" && observation.KnowledgeBaseID != "" && req.KnowledgeBaseID != observation.KnowledgeBaseID { + return Release{}, errors.New("请求知识库与观测知识库不一致") + } + + var target *Row + var targetFingerprint string + for _, raw := range observation.Rows { + var row Row + if err := json.Unmarshal(raw, &row); err != nil { + return Release{}, errors.New("观测里的死信行不是合法 JSON") + } + if row.ID != req.DeadLetterID { + continue + } + if target != nil { + return Release{}, errors.New("观测里出现重复的死信 id") + } + copied := row + target = &copied + targetFingerprint = Fingerprint(raw) + } + if target == nil { + return Release{}, errors.New("观测里没有这条死信：上游记录可能已被删除，不能用解除掩盖删除") + } + if target.FailedAt.After(req.Replay.At) { + return Release{}, errors.New("重放时间早于这条死信的失败时间") + } + // 重放失败会再写一条死信行（asynq 中间件对每次耗尽都落一行）。同一任务键在 + // 重放之后仍有新行，说明重放没有成功，或者又失败了一次。 + for _, raw := range observation.Rows { + var row Row + if err := json.Unmarshal(raw, &row); err != nil { + return Release{}, errors.New("观测里的死信行不是合法 JSON") + } + if row.ID == target.ID { + continue + } + if sameKey(row, *target) && row.FailedAt.After(req.Replay.At) { + return Release{}, fmt.Errorf("重放之后同一任务仍产生新死信（id=%d），不能解除", row.ID) + } + } + if target.TenantID != observation.TenantID { + return Release{}, errors.New("死信租户与观测租户不一致") + } + return Release{ + Project: observation.Project, + TenantID: target.TenantID, + KnowledgeBaseID: observation.KnowledgeBaseID, + DeadLetterID: target.ID, + Fingerprint: targetFingerprint, + TaskType: target.TaskType, + Scope: target.Scope, + ScopeID: target.ScopeID, + RelatedID: target.RelatedID, + Replay: req.Replay, + Observation: req.Observation, + Digest: req.Digest, + Note: req.Note, + }, nil +} + +func sameKey(a, b Row) bool { + return a.TenantID == b.TenantID && a.TaskType == b.TaskType && a.Scope == b.Scope && + a.ScopeID == b.ScopeID && a.RelatedID == b.RelatedID +} + +// Unreleased 返回没有被精确解除记录覆盖的死信指纹。任何一行不匹配都阻断： +// 行被改过、出现新行、或解除记录缺失，三种情况都必须看得见。 +func Unreleased(rows [][]byte, released map[string]bool) ([]string, error) { + missing := make([]string, 0, len(rows)) + for _, raw := range rows { + var row Row + if err := json.Unmarshal(raw, &row); err != nil { + return nil, errors.New("源库死信行无法解析，按阻断处理") + } + digest := Fingerprint(raw) + if !released[digest] { + missing = append(missing, fmt.Sprintf("%d:%s", row.ID, digest)) + } + } + return missing, nil +} diff --git a/internal/librarydeadletter/deadletter_test.go b/internal/librarydeadletter/deadletter_test.go new file mode 100644 index 0000000..398bbb4 --- /dev/null +++ b/internal/librarydeadletter/deadletter_test.go @@ -0,0 +1,113 @@ +package librarydeadletter + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +const sampleRow = `{"id": 7, "payload": {"id": "k1"}, "scope": "knowledge", "scope_id": "k1", "failed_at": "2026-10-28T10:00:00+00:00", "last_error": "boom", "task_type": "wiki:ingest", "tenant_id": 10001, "fail_count": 3, "related_id": "k1"}` + +func observation(t *testing.T, rows ...string) (json.RawMessage, string) { + t.Helper() + value := Observation{ + Project: "kgmaint-weknora", TenantID: 10001, KnowledgeBaseID: "kb-1", + CapturedAt: time.Date(2026, 10, 29, 7, 0, 0, 0, time.UTC), + Settlement: Settlement{Unsettled: 0, PendingOps: 0}, + } + for _, row := range rows { + value.Rows = append(value.Rows, json.RawMessage(row)) + } + raw, err := json.Marshal(value) + if err != nil { + t.Fatalf("marshal observation: %v", err) + } + return raw, Fingerprint(raw) +} + +func request(t *testing.T, raw json.RawMessage, digest string) Request { + t.Helper() + return Request{ + Project: "kgmaint-weknora", DeadLetterID: 7, + Replay: Replay{Queue: "default", TaskID: "task-1", At: time.Date(2026, 10, 29, 6, 30, 0, 0, time.UTC), + Evidence: "run_now 200"}, + Observation: raw, Digest: digest, + } +} + +func TestVerifyAcceptsExactReleasedDeadLetter(t *testing.T) { + raw, digest := observation(t, sampleRow) + release, err := Verify(request(t, raw, digest), "kgmaint-weknora", time.Date(2026, 10, 29, 7, 5, 0, 0, time.UTC)) + if err != nil { + t.Fatalf("exact release rejected: %v", err) + } + if release.DeadLetterID != 7 || release.Fingerprint != Fingerprint([]byte(sampleRow)) || + release.TaskType != "wiki:ingest" || release.TenantID != 10001 { + t.Fatalf("release lost its binding: %+v", release) + } +} + +func TestVerifyRejectsUnsafeReleases(t *testing.T) { + cases := []struct { + name string + mutate func(*Request, *Observation) + want string + }{ + {"观察摘要被改写", func(r *Request, _ *Observation) { r.Digest = Fingerprint([]byte("other")) }, "观测摘要"}, + {"未结清", func(_ *Request, o *Observation) { o.Settlement.PendingOps = 1 }, "尚未结清"}, + {"断言的死信不存在", func(r *Request, _ *Observation) { r.DeadLetterID = 8 }, "没有这条死信"}, + {"重放早于失败", func(r *Request, _ *Observation) { + r.Replay.At = time.Date(2026, 10, 28, 9, 0, 0, 0, time.UTC) + }, "早于"}, + {"重放没有队列", func(r *Request, _ *Observation) { r.Replay.Queue = "" }, "队列"}, + {"重放后同一任务又失败", func(_ *Request, o *Observation) { + o.Rows = append(o.Rows, json.RawMessage( + `{"id": 9, "payload": {"id": "k1"}, "scope": "knowledge", "scope_id": "k1", "failed_at": "2026-10-29T06:45:00+00:00", "last_error": "boom", "task_type": "wiki:ingest", "tenant_id": 10001, "fail_count": 3, "related_id": "k1"}`)) + }, "新死信"}, + {"实例不一致", func(r *Request, _ *Observation) { r.Project = "other" }, "不一致"}, + } + for _, test := range cases { + t.Run(test.name, func(t *testing.T) { + raw, digest := observation(t, sampleRow) + var body Observation + if err := json.Unmarshal(raw, &body); err != nil { + t.Fatal(err) + } + req := request(t, raw, digest) + test.mutate(&req, &body) + if test.name == "重放后同一任务又失败" || test.name == "未结清" { + reserialized, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + req.Observation, req.Digest = reserialized, Fingerprint(reserialized) + } + if _, err := Verify(req, "kgmaint-weknora", time.Date(2026, 10, 29, 7, 5, 0, 0, time.UTC)); err == nil || + !strings.Contains(err.Error(), test.want) { + t.Fatalf("unsafe release accepted: %v", err) + } + }) + } +} + +func TestVerifyRejectsUnconfiguredInstance(t *testing.T) { + raw, digest := observation(t, sampleRow) + if _, err := Verify(request(t, raw, digest), "", time.Now()); err == nil { + t.Fatal("release accepted without a configured instance") + } +} + +func TestUnreleasedNamesEveryRow(t *testing.T) { + rows := [][]byte{[]byte(sampleRow), []byte(strings.Replace(sampleRow, `"id": 7`, `"id": 8`, 1))} + missing, err := Unreleased(rows, map[string]bool{Fingerprint(rows[0]): true}) + if err != nil { + t.Fatal(err) + } + if len(missing) != 1 || !strings.HasPrefix(missing[0], "8:sha256:") { + t.Fatalf("unexpected coverage: %v", missing) + } + if _, err := Unreleased([][]byte{[]byte("not json")}, nil); err == nil { + t.Fatal("unparseable row was treated as released") + } +} diff --git a/internal/platform/postgres/library_dead_letters.go b/internal/platform/postgres/library_dead_letters.go new file mode 100644 index 0000000..bd7960d --- /dev/null +++ b/internal/platform/postgres/library_dead_letters.go @@ -0,0 +1,149 @@ +package postgres + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + + "knowledge-governance/internal/governance" +) + +// 死信解除只写治理库：上游 task_dead_letters 一个字节都不动。为什么这样： +// 上游没有"已解决"标记，purge/DELETE 会销毁失败证据；治理侧保留审计并绑定 +// 实时行指纹，行变了就不会再命中，阻断自动恢复。 + +// RecordDeadLetterRelease 幂等写入一条解除记录。同实例同指纹重复提交返回已存记录， +// 但内容不一致（重放引用/观测摘要不同）必须报冲突，不能悄悄覆盖审计。 +func (s *Store) RecordDeadLetterRelease(ctx context.Context, release governance.LibraryDeadLetterRelease) (governance.LibraryDeadLetterRelease, bool, error) { + if release.ID == "" || release.Instance == "" || release.Fingerprint == "" || release.Actor == "" { + return governance.LibraryDeadLetterRelease{}, false, errors.New("dead letter release requires identity, fingerprint and actor") + } + observation := release.Observation + if len(observation) == 0 { + observation = json.RawMessage(`{}`) + } + var id string + err := s.db.QueryRow(ctx, ` +INSERT INTO library_dead_letter_releases (id, instance, tenant_id, kb_id, dead_letter_id, fingerprint, + task_type, scope, scope_id, related_id, replay_queue, replay_task_id, replayed_at, replay_evidence, + observation_digest, observation, note, actor, created_at) +VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19) +ON CONFLICT (instance, fingerprint) DO NOTHING +RETURNING id`, + release.ID, release.Instance, release.TenantID, release.KnowledgeBaseID, release.DeadLetterID, + release.Fingerprint, release.TaskType, release.Scope, release.ScopeID, release.RelatedID, + release.ReplayQueue, release.ReplayTaskID, release.ReplayedAt.UTC(), release.ReplayEvidence, + release.ObservationDigest, observation, release.Note, release.Actor, release.CreatedAt.UTC()).Scan(&id) + if err == nil { + stored, loadErr := s.DeadLetterRelease(ctx, id) + return stored, false, loadErr + } + if !errors.Is(err, pgx.ErrNoRows) { + return governance.LibraryDeadLetterRelease{}, false, fmt.Errorf("record dead letter release: %w", err) + } + existing, loadErr := s.deadLetterReleaseByFingerprint(ctx, release.Instance, release.Fingerprint) + if loadErr != nil { + return governance.LibraryDeadLetterRelease{}, false, loadErr + } + if existing.ObservationDigest != release.ObservationDigest || existing.ReplayTaskID != release.ReplayTaskID || + !existing.ReplayedAt.Equal(release.ReplayedAt.UTC()) { + return existing, false, governance.ErrLibraryJobIdempotencyConflict + } + return existing, true, nil +} + +func (s *Store) deadLetterReleaseByFingerprint(ctx context.Context, instance, fingerprint string) (governance.LibraryDeadLetterRelease, error) { + var id string + if err := s.db.QueryRow(ctx, `SELECT id FROM library_dead_letter_releases WHERE instance=$1 AND fingerprint=$2`, + instance, fingerprint).Scan(&id); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return governance.LibraryDeadLetterRelease{}, errors.New("dead letter release not found") + } + return governance.LibraryDeadLetterRelease{}, fmt.Errorf("load dead letter release: %w", err) + } + return s.DeadLetterRelease(ctx, id) +} + +func (s *Store) DeadLetterRelease(ctx context.Context, id string) (governance.LibraryDeadLetterRelease, error) { + var release governance.LibraryDeadLetterRelease + var observation []byte + err := s.db.QueryRow(ctx, ` +SELECT id, instance, tenant_id, kb_id, dead_letter_id, fingerprint, task_type, scope, scope_id, related_id, + replay_queue, replay_task_id, replayed_at, replay_evidence, observation_digest, observation, note, actor, created_at +FROM library_dead_letter_releases WHERE id=$1`, id).Scan(&release.ID, &release.Instance, &release.TenantID, + &release.KnowledgeBaseID, &release.DeadLetterID, &release.Fingerprint, &release.TaskType, &release.Scope, + &release.ScopeID, &release.RelatedID, &release.ReplayQueue, &release.ReplayTaskID, &release.ReplayedAt, + &release.ReplayEvidence, &release.ObservationDigest, &observation, &release.Note, &release.Actor, &release.CreatedAt) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return governance.LibraryDeadLetterRelease{}, err + } + return governance.LibraryDeadLetterRelease{}, fmt.Errorf("load dead letter release: %w", err) + } + release.Observation = json.RawMessage(observation) + return release, nil +} + +// DeadLetterReleases 按实例倒序列出解除记录（审计视图）。 +func (s *Store) DeadLetterReleases(ctx context.Context, instance string, limit int) ([]governance.LibraryDeadLetterRelease, error) { + if instance == "" { + return nil, errors.New("dead letter releases require an instance") + } + if limit <= 0 || limit > 200 { + limit = 50 + } + rows, err := s.db.Query(ctx, ` +SELECT id FROM library_dead_letter_releases WHERE instance=$1 ORDER BY created_at DESC LIMIT $2`, instance, limit) + if err != nil { + return nil, fmt.Errorf("list dead letter releases: %w", err) + } + defer rows.Close() + ids := []string{} + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("list dead letter releases: %w", err) + } + ids = append(ids, id) + } + if rows.Err() != nil { + return nil, fmt.Errorf("list dead letter releases: %w", rows.Err()) + } + result := make([]governance.LibraryDeadLetterRelease, 0, len(ids)) + for _, id := range ids { + release, loadErr := s.DeadLetterRelease(ctx, id) + if loadErr != nil { + return nil, loadErr + } + result = append(result, release) + } + return result, nil +} + +// DeadLetterReleaseFingerprints 是封存执行侧需要的唯一投影：只给指纹，不给证据。 +// worker 把它原样传给 agent，CLI 再对实时行逐字节复核。 +func (s *Store) DeadLetterReleaseFingerprints(ctx context.Context, instance string) ([]string, error) { + if instance == "" { + return nil, errors.New("dead letter release fingerprints require an instance") + } + rows, err := s.db.Query(ctx, `SELECT fingerprint FROM library_dead_letter_releases WHERE instance=$1 ORDER BY created_at`, instance) + if err != nil { + return nil, fmt.Errorf("list dead letter release fingerprints: %w", err) + } + defer rows.Close() + result := []string{} + for rows.Next() { + var fingerprint string + if err := rows.Scan(&fingerprint); err != nil { + return nil, fmt.Errorf("list dead letter release fingerprints: %w", err) + } + result = append(result, fingerprint) + } + return result, rows.Err() +} + +var _ = time.Time{} diff --git a/internal/platform/postgres/migrations/000073_library_dead_letter_releases.down.sql b/internal/platform/postgres/migrations/000073_library_dead_letter_releases.down.sql new file mode 100644 index 0000000..25e895e --- /dev/null +++ b/internal/platform/postgres/migrations/000073_library_dead_letter_releases.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS library_dead_letter_releases; \ No newline at end of file diff --git a/internal/platform/postgres/migrations/000073_library_dead_letter_releases.up.sql b/internal/platform/postgres/migrations/000073_library_dead_letter_releases.up.sql new file mode 100644 index 0000000..4c609e3 --- /dev/null +++ b/internal/platform/postgres/migrations/000073_library_dead_letter_releases.up.sql @@ -0,0 +1,29 @@ +-- 死信解除：上游 task_dead_letters 没有"已解决"标记，purge/DELETE 都会销毁失败证据。 +-- 这里只记录"某条精确死信已被审计解除"，不复制也不删除上游行：解除绑定实时行的 +-- 逐字节指纹、真实重放引用、源侧观测摘要与操作人；行被改过或出现新行都不会命中。 +CREATE TABLE library_dead_letter_releases ( + id VARCHAR(64) PRIMARY KEY, + instance VARCHAR(128) NOT NULL, + tenant_id BIGINT NOT NULL, + kb_id VARCHAR(64) NOT NULL, + dead_letter_id BIGINT NOT NULL, + fingerprint VARCHAR(80) NOT NULL, + task_type VARCHAR(64) NOT NULL DEFAULT '', + scope VARCHAR(32) NOT NULL DEFAULT '', + scope_id VARCHAR(64) NOT NULL DEFAULT '', + related_id VARCHAR(64) NOT NULL DEFAULT '', + replay_queue VARCHAR(64) NOT NULL, + replay_task_id VARCHAR(128) NOT NULL, + replayed_at TIMESTAMPTZ NOT NULL, + replay_evidence TEXT NOT NULL, + observation_digest VARCHAR(80) NOT NULL, + observation JSONB NOT NULL, + note TEXT NOT NULL DEFAULT '', + actor VARCHAR(128) NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + -- 同一个实例里同一条死信只能解除一次；重复提交返回已存记录。 + UNIQUE (instance, fingerprint) +); + +CREATE INDEX idx_library_dead_letter_releases_instance + ON library_dead_letter_releases (instance, created_at DESC); \ No newline at end of file diff --git a/internal/platform/postgres/migrations_test.go b/internal/platform/postgres/migrations_test.go index 6e64155..f0c76f1 100644 --- a/internal/platform/postgres/migrations_test.go +++ b/internal/platform/postgres/migrations_test.go @@ -11,35 +11,40 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) { if err != nil { t.Fatalf("loadMigrations() error = %v", err) } - if len(migrations) != 72 { - t.Fatalf("migration count = %d, want 72", len(migrations)) + if len(migrations) != 73 { + t.Fatalf("migration count = %d, want 73", len(migrations)) + } + // 73 记录死信精确解除：保留上游行，只绑定指纹、重放引用与观测摘要。 + if release := migrations[len(migrations)-1]; 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)-1]; instance.Version != 72 || + if instance := migrations[len(migrations)-2]; instance.Version != 72 || !strings.Contains(instance.UpSQL, "library_instance_maintenance") { t.Fatalf("migration 72 = %#v", instance) } // 71 让维护记录能区分实例级维护与仅治理库互斥。 - if level := migrations[len(migrations)-2]; level.Version != 71 || + if level := migrations[len(migrations)-3]; 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)-3]; maint.Version != 70 || !strings.Contains(maint.UpSQL, "library_maintenance") { + if maint := migrations[len(migrations)-4]; maint.Version != 70 || !strings.Contains(maint.UpSQL, "library_maintenance") { t.Fatalf("migration 70 = %#v", maint) } - if tail := migrations[len(migrations)-4]; tail.Version != 69 || + if tail := migrations[len(migrations)-5]; tail.Version != 69 || !strings.Contains(tail.UpSQL, "library_jobs_write_target") { t.Fatalf("migration 69 = %#v", tail) } - if restore := migrations[len(migrations)-5]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") { + if restore := migrations[len(migrations)-6]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") { t.Fatalf("migration 68 = %#v", restore) } - if last := migrations[len(migrations)-6]; last.Version != 67 || + if last := migrations[len(migrations)-7]; 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)-7]; flow.Version != 66 || + if flow := migrations[len(migrations)-8]; flow.Version != 66 || !strings.Contains(flow.UpSQL, "CREATE TABLE library_versions") || !strings.Contains(flow.UpSQL, "library_jobs_active") { t.Fatalf("migration 66 = %#v", flow) diff --git a/internal/releaseadapter/v071/library_capture.go b/internal/releaseadapter/v071/library_capture.go index f3c72bb..51a5425 100644 --- a/internal/releaseadapter/v071/library_capture.go +++ b/internal/releaseadapter/v071/library_capture.go @@ -20,6 +20,9 @@ type LibraryCaptureRequest struct { // it differs from Compatibility, so a model/index change between the // preflight read and the maintenance window cannot be sealed silently. VerifyContract func(context.Context) error + // ReleasedDeadLetters 是治理平台已审计的精确解除指纹。它只把"这些确切的行" + // 从死信阻断里摘出来：实时行必须逐字节命中，否则照旧阻断。 + ReleasedDeadLetters []string } // CaptureLibrary connects maintenance to the actual PG/PIT/object archive path. @@ -33,7 +36,7 @@ func CaptureLibrary(ctx context.Context, freeze *libraryfreeze.Coordinator, arch return libraryfreeze.Result{}, errors.New("invalid library capture configuration") } result, err := freeze.RunPrepared(ctx, req.JobID, func(ctx context.Context) error { - if err := checkLibrarySourceSettlement(ctx, db); err != nil { + if err := checkLibrarySourceSettlementReleased(ctx, db, req.ReleasedDeadLetters); err != nil { return err } if err := sourceReady(ctx); err != nil { @@ -41,10 +44,10 @@ func CaptureLibrary(ctx context.Context, freeze *libraryfreeze.Coordinator, arch } // Check readiness before a maintenance window, then repeat the export inside // that window: work can arrive between preflight and the graceful stop. - _, err := ExportLibrary(ctx, db, req.Identity) + _, err := ExportLibraryWithReleases(ctx, db, req.Identity, req.ReleasedDeadLetters) return err }, func(ctx context.Context) (string, error) { - if err := checkLibrarySourceSettlement(ctx, db); err != nil { + if err := checkLibrarySourceSettlementReleased(ctx, db, req.ReleasedDeadLetters); err != nil { return "", err } if err := sourceReady(ctx); err != nil { @@ -55,7 +58,7 @@ func CaptureLibrary(ctx context.Context, freeze *libraryfreeze.Coordinator, arch return "", err } } - pg, err := ExportLibrary(ctx, db, req.Identity) + pg, err := ExportLibraryWithReleases(ctx, db, req.Identity, req.ReleasedDeadLetters) if err != nil { return "", err } diff --git a/internal/releaseadapter/v071/library_export.go b/internal/releaseadapter/v071/library_export.go index 7993325..c1ff3c1 100644 --- a/internal/releaseadapter/v071/library_export.go +++ b/internal/releaseadapter/v071/library_export.go @@ -15,6 +15,12 @@ type librarySnapshotDB interface { // It does not certify semantic quality, freeze Elasticsearch/objects, or authorize // publication. Those require a separately verified, complete library version. func ExportLibrary(ctx context.Context, db librarySnapshotDB, identity PostgresIdentity) (PostgresSnapshot, error) { + return ExportLibraryWithReleases(ctx, db, identity, nil) +} + +// ExportLibraryWithReleases 与 ExportLibrary 相同，但允许调用方带上精确解除的死信 +// 指纹（只影响死信这一项判定，其余结算条件不变）。 +func ExportLibraryWithReleases(ctx context.Context, db librarySnapshotDB, identity PostgresIdentity, released []string) (PostgresSnapshot, error) { if db == nil || identity.TenantID == 0 || !validSearchIdentifier(identity.KnowledgeBaseID) || identity.KnowledgeID != "" { return PostgresSnapshot{}, errors.New("invalid library snapshot scope") } @@ -40,7 +46,17 @@ func ExportLibrary(ctx context.Context, db librarySnapshotDB, identity PostgresI return PostgresSnapshot{}, errors.New("read library settlement") } if unsettled != 0 || pending != 0 || failed != 0 { - return PostgresSnapshot{}, errors.New("library has unfinished or failed processing; semantic acceptance is still required after settlement") + if failed != 0 && unsettled == 0 && pending == 0 { + rows, rowsErr := sourceDeadLetterRows(ctx, tx, identity.TenantID, identity.KnowledgeBaseID) + if rowsErr != nil { + return PostgresSnapshot{}, rowsErr + } + if releasedErr := requireReleased(rows, released); releasedErr != nil { + return PostgresSnapshot{}, releasedErr + } + } else { + return PostgresSnapshot{}, errors.New("library has unfinished or failed processing; semantic acceptance is still required after settlement") + } } exporter := PostgresExporter{db: tx} var snapshot PostgresSnapshot diff --git a/internal/releaseadapter/v071/library_export_test.go b/internal/releaseadapter/v071/library_export_test.go index 4293d20..4b8a012 100644 --- a/internal/releaseadapter/v071/library_export_test.go +++ b/internal/releaseadapter/v071/library_export_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "knowledge-governance/internal/librarydeadletter" "strings" "testing" @@ -102,9 +103,16 @@ func (d *libraryFixtureTx) Query(_ context.Context, q string, _ ...any) (pgx.Row if d.queryError { return nil, errors.New("fixture failure") } + // 死信行按真实形状返回：解除判定必须逐字节比对指纹，空结果会让"存在即阻断"失效。 + if strings.Contains(q, "to_jsonb(d)") && d.failed > 0 { + return &jsonRows{values: []json.RawMessage{json.RawMessage(fixtureDeadLetter)}}, nil + } return &jsonRows{values: []json.RawMessage{}}, nil } +// fixtureDeadLetter 与 PostgreSQL to_jsonb(task_dead_letters) 的键序/形状一致。 +const fixtureDeadLetter = `{"id": 1, "payload": {"knowledge_id": "deleted-doc"}, "scope": "knowledge", "scope_id": "deleted-doc", "failed_at": "2026-10-28T10:00:00+00:00", "last_error": "llm timeout", "task_type": "wiki:ingest", "tenant_id": 10000, "fail_count": 3, "related_id": "deleted-doc"}` + type libraryFixtureRow struct { d *libraryFixtureTx sql string @@ -128,3 +136,24 @@ func (r libraryFixtureRow) Scan(dest ...any) error { } return nil } + +// 解除只能按精确指纹生效：同一条死信换个字节（或没有解除）都必须继续阻断。 +func TestLibraryExportReleasesOnlyExactDeadLetterFingerprint(t *testing.T) { + db := newLibraryFixture() + db.tx.failed = 1 + identity := PostgresIdentity{TenantID: 10000, KnowledgeBaseID: "kb-safe"} + if _, err := ExportLibrary(context.Background(), db, identity); err == nil { + t.Fatal("unreleased dead letter did not block the export") + } + if _, err := ExportLibraryWithReleases(context.Background(), db, identity, + []string{"sha256:0000000000000000000000000000000000000000000000000000000000000000"}); err == nil { + t.Fatal("a stale fingerprint released the dead letter") + } + released := librarydeadletter.Fingerprint([]byte(fixtureDeadLetter)) + if _, err := ExportLibraryWithReleases(context.Background(), db, identity, []string{released}); err != nil { + t.Fatalf("exact released fingerprint still blocked: %v", err) + } + if !db.tx.committed { + t.Fatal("released export did not commit its snapshot") + } +} diff --git a/internal/releaseadapter/v071/library_preflight.go b/internal/releaseadapter/v071/library_preflight.go index abf2f31..2fcad04 100644 --- a/internal/releaseadapter/v071/library_preflight.go +++ b/internal/releaseadapter/v071/library_preflight.go @@ -24,6 +24,12 @@ func LibrarySourceSettled(ctx context.Context, db *pgxpool.Pool) error { return checkLibrarySourceSettlement(ctx, db) } +// LibrarySourceSettledWithReleases 是同一检查，但带上治理平台已审计的精确死信解除 +// 指纹：只有逐字节命中的行不再阻断。 +func LibrarySourceSettledWithReleases(ctx context.Context, db *pgxpool.Pool, released []string) error { + return checkLibrarySourceSettlementReleased(ctx, db, released) +} + // ValidateLibraryCandidateForPreflight runs the candidate validator used by // ArchiveLibraryCandidate (identity, bidirectional content coverage, generated // question coverage, resources and digests) without archiving anything. diff --git a/internal/releaseadapter/v071/library_source_settlement.go b/internal/releaseadapter/v071/library_source_settlement.go index 541a317..d5f687a 100644 --- a/internal/releaseadapter/v071/library_source_settlement.go +++ b/internal/releaseadapter/v071/library_source_settlement.go @@ -2,9 +2,15 @@ package v071 import ( "context" + "encoding/json" "errors" + "fmt" + "strings" + "time" "github.com/jackc/pgx/v5" + + "knowledge-governance/internal/librarydeadletter" ) // A maintenance window stops a shared writer, not a single KB. Check its entire @@ -13,7 +19,14 @@ import ( // This only establishes recorded processing settlement, never Wiki semantics or // the absence of unrecorded/remote work; CaptureLibrary still requires the // deployment's ingress and background-worker readiness callback. +// +// 死信不再"存在即阻断"：允许调用方带上一组精确解除指纹。每条实时行都要被逐字节 +// 命中，行被改过、出现新行或没有解除记录都仍然阻断。这里不删除任何上游行。 func checkLibrarySourceSettlement(ctx context.Context, db librarySnapshotDB) error { + return checkLibrarySourceSettlementReleased(ctx, db, nil) +} + +func checkLibrarySourceSettlementReleased(ctx context.Context, db librarySnapshotDB, released []string) error { tx, err := db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly}) if err != nil { return errors.New("begin library source settlement") @@ -23,16 +36,118 @@ func checkLibrarySourceSettlement(ctx context.Context, db librarySnapshotDB) err if err := tx.QueryRow(ctx, librarySourceSettlementSQL).Scan(&unsettled, &pending, &failed); err != nil { return errors.New("read library source settlement") } - if unsettled != 0 || pending != 0 || failed != 0 { + if unsettled != 0 || pending != 0 { return errors.New("library source has unfinished or failed processing") } + if failed != 0 { + rows, err := sourceDeadLetterRows(ctx, tx, 0, "") + if err != nil { + return err + } + if err := requireReleased(rows, released); err != nil { + return err + } + } if err := tx.Commit(ctx); err != nil { return errors.New("commit library source settlement") } return nil } -const librarySourceSettlementSQL = `SELECT /* library-source-settlement */ +const librarySourceSettlementSQL = `SELECT /* library-source-settlement */ (SELECT count(*) FROM knowledges WHERE deleted_at IS NULL AND (parse_status IS DISTINCT FROM 'completed' OR pending_subtasks_count IS NULL OR pending_subtasks_count<>0)), (SELECT count(*) FROM task_pending_ops), (SELECT count(*) FROM task_dead_letters)` + +// sourceDeadLetterRows 读原始行文本（to_jsonb 已规范化），指纹直接对字节求摘要。 +// tenant=0 表示整库（维护窗口的粒度），否则按知识库范围。 +func sourceDeadLetterRows(ctx context.Context, q interface { + Query(context.Context, string, ...any) (pgx.Rows, error) +}, tenant uint64, kb string) ([][]byte, error) { + statement := `SELECT to_jsonb(d)::text FROM task_dead_letters d` + args := []any{} + if tenant != 0 { + statement += ` WHERE d.tenant_id=$1 AND ((d.scope='knowledge_base' AND d.scope_id=$2) OR + (d.scope='knowledge' AND d.scope_id IN (SELECT id FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2)))` + args = append(args, tenant, kb) + } + statement += ` ORDER BY d.id` + rows, err := q.Query(ctx, statement, args...) + if err != nil { + return nil, errors.New("read library dead letters") + } + defer rows.Close() + var result [][]byte + for rows.Next() { + var raw []byte + if err := rows.Scan(&raw); err != nil { + return nil, errors.New("read library dead letters") + } + result = append(result, raw) + } + if rows.Err() != nil { + return nil, errors.New("read library dead letters") + } + return result, nil +} + +func releasedSet(released []string) map[string]bool { + set := make(map[string]bool, len(released)) + for _, fingerprint := range released { + set[fingerprint] = true + } + return set +} + +func requireReleased(rows [][]byte, released []string) error { + missing, err := librarydeadletter.Unreleased(rows, releasedSet(released)) + if err != nil { + return err + } + if len(missing) > 0 { + shown := missing + if len(shown) > 5 { + shown = shown[:5] + } + return fmt.Errorf("source has %d dead letter(s) without a verified release: %s", len(missing), strings.Join(shown, ", ")) + } + return nil +} + +// ObserveDeadLetters 生成主机侧只读观测：整库结算计数 + 每条死信原始行。 +// 解除记录绑定这份观测的字节摘要，平台侧不复制"自报成功"的结论。 +func ObserveDeadLetters(ctx context.Context, db librarySnapshotDB, project string, tenant uint64, kb string) ([]byte, error) { + if project == "" || kb == "" || tenant == 0 { + return nil, errors.New("observation requires project, tenant and knowledge base") + } + tx, err := db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly}) + if err != nil { + return nil, errors.New("begin dead letter observation") + } + defer tx.Rollback(context.WithoutCancel(ctx)) + var unsettled, pending, failed int64 + if err := tx.QueryRow(ctx, librarySourceSettlementSQL).Scan(&unsettled, &pending, &failed); err != nil { + return nil, errors.New("read library source settlement") + } + rows, err := sourceDeadLetterRows(ctx, tx, 0, "") + if err != nil { + return nil, err + } + observation := librarydeadletter.Observation{ + Project: project, TenantID: tenant, KnowledgeBaseID: kb, + CapturedAt: time.Now().UTC().Truncate(time.Second), + Settlement: librarydeadletter.Settlement{Unsettled: unsettled, PendingOps: pending}, + Rows: make([]json.RawMessage, 0, len(rows)), + } + for _, raw := range rows { + observation.Rows = append(observation.Rows, json.RawMessage(raw)) + } + payload, err := json.Marshal(observation) + if err != nil { + return nil, errors.New("encode dead letter observation") + } + if err := tx.Commit(ctx); err != nil { + return nil, errors.New("commit dead letter observation") + } + return payload, nil +}