上游 task_dead_letters 没有解决标记,purge/DELETE 会销毁"失败过"的唯一证据。
治理侧新增解除记录(迁移 073):绑定实时行逐字节指纹、重放引用、源侧观测摘要与
操作人;行被改过、重放后又出现新死信、源侧未结清都不命中,封存照旧阻断。
指纹先压紧空白再摘要,避免 PG to_jsonb::text 与 json.Marshal 的空白差异。
试过:go test ./internal/{librarydeadletter,libraryagent,httpapi,releaseadapter/v071,
platform/postgres},make verify 15 项全绿。
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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": [
|
||||
|
||||
+72
-2
@@ -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 <id> (capture; default: lib-<scope>-<UTC timestamp>)
|
||||
-dry-run (capture: export + validate only, no archive, no stop)
|
||||
-snapshot <id> (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
|
||||
}
|
||||
|
||||
+9
-1
@@ -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,不会因为作业失败而假装来源已恢复。
|
||||
|
||||
@@ -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]
|
||||
}
|
||||
|
||||
@@ -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})
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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{}
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS library_dead_letter_releases;
|
||||
@@ -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);
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user