Compare commits
18 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a06945516f | |||
| e508abd958 | |||
| c542a1333a | |||
| 1aecd9d6af | |||
| c2402947a9 | |||
| 958292d1ff | |||
| bc49b85b56 | |||
| a6406209bd | |||
| b061a9686c | |||
| 2e39e49569 | |||
| cb0090802b | |||
| 7d4ffdbfe5 | |||
| 84e8a0dfdc | |||
| faaf003707 | |||
| 4d6ecda310 | |||
| f2d6f3f646 | |||
| 37a8370f7b | |||
| 11c52ad81b |
@@ -1,5 +1,98 @@
|
||||
# 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
|
||||
|
||||
- 死信解除绑定 `LIBRARY_SOURCE_INSTANCE`(封存源),留空回落到 `LIBRARY_TARGET_INSTANCE`:
|
||||
同步目标与封存源常常不是同一个实例,用目标名去绑来源死信会让真实观测永远对不上
|
||||
(拒绝而不是误放行)。worker 取指纹、API 写记录用同一个来源实例名。
|
||||
- 结算/封存不再"死信存在即阻断",而是"每条实时死信都要被精确解除命中":
|
||||
`checkLibrarySourceSettlementReleased` / `ExportLibraryWithReleases` 对每条行
|
||||
重算指纹,未命中、行被改过、出现新行都继续阻断并点名。
|
||||
- 解除指纹经 worker → agent → `libraryctl -released-dead-letters` 逐段校验
|
||||
(`sha256:<64 hex>`,最多 64 条,不接受通配),agent 的固定命令形态不变。
|
||||
|
||||
### Fixed
|
||||
|
||||
- **同步过去的资源行指向来源实例的存储后端 id(定位符与独立列两处)**:`storage://<id>/minio://<bucket>/<key>`
|
||||
里的 `<id>` 是来源 `storage_backends` 的主键,目标实例没有这条行。只换桶不改 id 时,
|
||||
原版应用落回默认存储服务并拒绝整条路径(真实下载 500 `code 1007
|
||||
invalid MinIO file path`)。现在写入前按"租户 + 目标桶名"在目标库解析后端 id,
|
||||
写入与校验共用同一份行变换;解析不到就拒绝写入,不产出目标读不出来的定位符。
|
||||
`resources.storage_backend_id` 这一列同样要换(应用先按它找文件服务)。
|
||||
- 指纹对空白敏感会让"同一份内容"两边算不出同一个值(PostgreSQL `to_jsonb::text`
|
||||
带空格、Go `json.Marshal` 不带);指纹改为先 `json.Compact` 再摘要。
|
||||
|
||||
## 开发快照(封存入口接线:库级窗口接上实例级闸门)
|
||||
|
||||
### Fixed
|
||||
|
||||
- **主机侧封存不再用"库级窗口"跑实例**:`libraryctl capture` / `recover` 改由
|
||||
`librarymaintenance.Manager` 装配——入口闸门、后台静默检查、目标指纹、入口探针与发布链路
|
||||
是同一套实现。实例级维护里"写入者 exit=1"不再等于"在途请求被截断":被代理的 WeKnora 收到
|
||||
SIGTERM 就 `exit(1)` 是已知上游行为,排空证据来自闸门(`active_at_close` / `drained`),
|
||||
校验没有放松,而是补上了缺的那一层证据。
|
||||
- 缺少 `LIBRARY_TARGET_INGRESS_*` / `LIBRARY_TARGET_REDIS_*` 时 `capture` 在**开窗口前拒绝**
|
||||
并点名缺哪个变量;不再退回一个没有排空证据的窗口(那样的"成功"无法复核)。
|
||||
- 维护接线解析收敛到 `internal/librarymaintenance/env.go`:agent 与 worker 讲同一组变量名,
|
||||
空白值算缺失、队列库号必须是非负整数、口令不 Trim(前后空格也是口令)。
|
||||
- `libraryagent` 启动时如实打印维护目标是否接好(缺了就直说"封存会拒绝")。
|
||||
- `scripts/library-agent-install.sh` 强制 `agent.env`(0600,`EnvironmentFile=`),并把源容器/
|
||||
项目/写入者声明做成可覆盖变量;没有 env 文件直接 `AGENT_ENV_MISSING` 退出。
|
||||
|
||||
- 主机侧封存新增 `-source-s3-endpoint`:容器里的对象存储端点常常只是服务名
|
||||
(`langfuse-minio:9000`),宿主机解析不到,资源导出会失败(真跑报 `read library resource`)。
|
||||
覆盖只换访问入口,桶/对象键不变;`status` 如实打印实际 endpoint。
|
||||
- **封存 API 的队列作业 id 超列宽**:`library-capture-<scope>-<nano>` 是 45 字符,而
|
||||
`jobs.id` 是 `varchar(36)`,真实请求一律 500(`value too long`)。此前页面停在"未接线"
|
||||
从未走到这里;改为 `libcap-<scope>-<nano>` 并用测试锁住列宽。
|
||||
- **索引范围过滤漏掉 app 写入的正文**:v0.7.1 的 app 自己写索引不带 `tenant_id`,
|
||||
治理侧同步才补齐;只按 `term tenant_id` 过滤会把整库正文漏掉(真跑:索引有 2 条、
|
||||
导出 0 条、候选校验报 legacy coverage)。改为"本租户 或 未写租户",写了别的租户仍排除。
|
||||
|
||||
- **空目标索引不能当成"索引被改坏"**:全新 WeKnora 实例先用空映射建索引(`{}`),
|
||||
字段等第一次写入才动态生成;首次同步因此在读取目标状态时就失败。现在"没有范围字段
|
||||
+ 确实 0 篇文档"按空基线处理,有文档却看不出归属仍然报错。
|
||||
|
||||
- **资源定位符的桶没有跟着换**:同步把字节写进目标桶,但 `resources.physical_path`
|
||||
是从快照原样搬的,里面仍写着来源桶(`storage://…/minio://kgmaint-1028a/…`)。
|
||||
目标实例之后取图/附件会指向测试环境的桶,或直接取不到。现在写入前把定位符里的桶
|
||||
换成本次目标桶(键不变),解析不出来就报错而不是写一行指向别处的资源。
|
||||
|
||||
- **恢复作业 id 让维护窗口拒绝**:`librestore-<scope>-<时间戳.微秒>` 带 `.`,而冻结协调器
|
||||
只接受 `[A-Za-z0-9_-]`,每次恢复都在开窗口时被拒(`invalid freeze request`),作业随后被
|
||||
接管扫成失败。改用纳秒整数,并加测试锁住"id 必须是维护窗口可接受的标识"。
|
||||
|
||||
### Tests
|
||||
|
||||
- `internal/httpapi/library_restore_test.go`:恢复作业 id 必须能被维护窗口接受。
|
||||
- `internal/releaseadapter/v071/resource_locator_test.go`:嵌套/单层两种定位符换桶、裸键原样、
|
||||
非法形状拒绝。
|
||||
- `internal/releaseadapter/v071/search_empty_index_test.go`:空索引=空基线、有文档无范围字段必须报错。
|
||||
- `internal/httpapi/library_capture_test.go`:队列作业 id 必须放得进 `varchar(36)` 且形状固定。
|
||||
- `internal/releaseadapter/v071/search_tenant_filter_test.go`:接受未写租户、排除他租户、
|
||||
无租户字段时不造条件、整库过滤仍带知识库身份。
|
||||
- `cmd/libraryctl/discover_test.go`:端点覆盖生效且必须带 scheme。
|
||||
- `internal/librarymaintenance/env_test.go`:整份声明解析、逐项点名缺失、空白拒绝、
|
||||
口令空格保留、非法队列库号。
|
||||
- `cmd/libraryctl/maintenance_test.go`:入口/队列/源身份绑到同一份配置、数据库身份不泄漏口令、
|
||||
写入者声明透传、缺闸门拒绝。
|
||||
|
||||
## 开发快照(Wiki 撤销贡献:逐页语义比对 + 维护窗口真实呈现)
|
||||
|
||||
### Added
|
||||
|
||||
@@ -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": [
|
||||
|
||||
@@ -203,6 +203,7 @@ func run(ctx context.Context, getenv func(string) string) error {
|
||||
VersionHistoryRepo: versionHistoryRepo,
|
||||
LibraryArchiveRoot: config.LibraryArchiveRoot,
|
||||
LibraryTargetInstance: config.LibraryTargetInstance,
|
||||
LibrarySourceInstance: config.SourceInstance(),
|
||||
LibraryAgentSocket: config.LibraryAgentSocket,
|
||||
LibraryCapacityBudgetBytes: config.LibraryCapacityBudgetBytes,
|
||||
LibraryProductionDatabaseURL: config.LibraryProductionDatabaseURL,
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"time"
|
||||
|
||||
"knowledge-governance/internal/libraryagent"
|
||||
"knowledge-governance/internal/librarymaintenance"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@@ -26,6 +27,7 @@ func main() {
|
||||
flag.StringVar(&config.SourceApp, "source-app", "WKNFull-app", "source container describing the stack")
|
||||
flag.StringVar(&config.SourcePG, "source-pg", "WKNFull-postgres", "source PostgreSQL container")
|
||||
flag.StringVar(&config.SourceES, "source-es", "WKNFull-elasticsearch", "source Elasticsearch container")
|
||||
flag.StringVar(&config.SourceS3Endpoint, "source-s3-endpoint", "", "host-reachable source object-store endpoint (overrides the container-internal one)")
|
||||
flag.StringVar(&config.GovernancePG, "governance-pg", "knowledge-governance-p1-postgres-1", "governance PostgreSQL container")
|
||||
flag.StringVar(&config.Project, "project", "weknora071full", "Compose project owning the writers")
|
||||
flag.StringVar(&config.Writers, "writers", "app", "declared writer services")
|
||||
@@ -39,6 +41,14 @@ func main() {
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
// 实例级维护接线(入口闸门 + 后台队列)不在 agent 的 Config 里:它由同一份
|
||||
// 环境声明提供给子进程。启动时先说清楚有没有——缺了封存会在开窗口前被拒,
|
||||
// 不允许"以为接好了"跑到一半才发现没有排空证据。
|
||||
if target, targetErr := librarymaintenance.TargetFromEnv(os.Getenv); targetErr != nil {
|
||||
log.Printf("library agent: 封存会拒绝(实例级维护未接线):%v", targetErr)
|
||||
} else {
|
||||
log.Printf("library agent: 实例级维护目标 %s 入口 %s 闸门 %s", target.Instance, target.IngressURL, target.IngressService)
|
||||
}
|
||||
if *describe {
|
||||
log.Printf("library agent config: %+v", agent.Describe())
|
||||
return
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
package main
|
||||
|
||||
import "testing"
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 目标实例可以是 OSS,也可以是 S3 兼容的 MinIO(本机专用验收实例就是 minio)。
|
||||
// 只认 oss 时,任何 minio 目标都会在发现阶段直接失败,等于这条验收链只能在生产存储上跑。
|
||||
@@ -14,7 +17,7 @@ func TestObjectStoreFromSourceEnvAcceptsMinIO(t *testing.T) {
|
||||
"MINIO_BUCKET_NAME": "kgmaint-1028a",
|
||||
"MINIO_PATH_PREFIX": "kgmaint/",
|
||||
}
|
||||
config, detail, err := objectStoreFromSourceEnv(app, 1<<20)
|
||||
config, detail, err := objectStoreFromSourceEnv(app, "", 1<<20)
|
||||
if err != nil {
|
||||
t.Fatalf("minio: %v", err)
|
||||
}
|
||||
@@ -29,12 +32,35 @@ func TestObjectStoreFromSourceEnvAcceptsMinIO(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 容器里的 minio 服务名在宿主机上解析不到:真跑时资源导出会失败(read library resource)。
|
||||
// 覆盖只换访问入口(同一个桶),不换来源;没有 scheme 的覆盖直接拒绝。
|
||||
func TestObjectStoreFromSourceEnvOverridesEndpointForHostSideCapture(t *testing.T) {
|
||||
app := map[string]string{
|
||||
"STORAGE_TYPE": "minio", "MINIO_ENDPOINT": "langfuse-minio:9000", "MINIO_USE_SSL": "false",
|
||||
"MINIO_ACCESS_KEY_ID": "key", "MINIO_SECRET_ACCESS_KEY": "secret",
|
||||
"MINIO_BUCKET_NAME": "kgmaint-1028a", "MINIO_PATH_PREFIX": "kgmaint/",
|
||||
}
|
||||
config, detail, err := objectStoreFromSourceEnv(app, "http://127.0.0.1:19100", 1<<20)
|
||||
if err != nil {
|
||||
t.Fatalf("override: %v", err)
|
||||
}
|
||||
if config.Endpoint != "http://127.0.0.1:19100" || config.Bucket != "kgmaint-1028a" {
|
||||
t.Fatalf("覆盖后配置不对:%+v", config)
|
||||
}
|
||||
if !strings.Contains(detail, "endpoint=http://127.0.0.1:19100") {
|
||||
t.Fatalf("状态里没有如实写出用的端点:%q", detail)
|
||||
}
|
||||
if _, _, err := objectStoreFromSourceEnv(app, "127.0.0.1:19100", 1<<20); err == nil {
|
||||
t.Fatal("没有 scheme 的端点覆盖被接受了")
|
||||
}
|
||||
}
|
||||
|
||||
func TestObjectStoreFromSourceEnvAcceptsMinIOOverTLS(t *testing.T) {
|
||||
app := map[string]string{
|
||||
"STORAGE_TYPE": "minio", "MINIO_ENDPOINT": "minio.internal:9000", "MINIO_USE_SSL": "true",
|
||||
"MINIO_ACCESS_KEY": "k2", "MINIO_SECRET_KEY": "s2", "MINIO_BUCKET_NAME": "bucket",
|
||||
}
|
||||
config, _, err := objectStoreFromSourceEnv(app, 1)
|
||||
config, _, err := objectStoreFromSourceEnv(app, "", 1)
|
||||
if err != nil {
|
||||
t.Fatalf("minio tls: %v", err)
|
||||
}
|
||||
@@ -48,7 +74,7 @@ func TestObjectStoreFromSourceEnvKeepsOSS(t *testing.T) {
|
||||
"STORAGE_TYPE": "oss", "OSS_ENDPOINT": "oss-cn-shenzhen.aliyuncs.com", "OSS_REGION": "cn-shenzhen",
|
||||
"OSS_ACCESS_KEY": "ak", "OSS_SECRET_KEY": "sk", "OSS_BUCKET_NAME": "prod",
|
||||
}
|
||||
config, _, err := objectStoreFromSourceEnv(app, 1)
|
||||
config, _, err := objectStoreFromSourceEnv(app, "", 1)
|
||||
if err != nil {
|
||||
t.Fatalf("oss: %v", err)
|
||||
}
|
||||
@@ -58,10 +84,10 @@ func TestObjectStoreFromSourceEnvKeepsOSS(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestObjectStoreFromSourceEnvRejectsUnknownStorage(t *testing.T) {
|
||||
if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "cos"}, 1); err == nil {
|
||||
if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "cos"}, "", 1); err == nil {
|
||||
t.Fatal("expected error for unsupported storage type")
|
||||
}
|
||||
if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "minio"}, 1); err == nil {
|
||||
if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "minio"}, "", 1); err == nil {
|
||||
t.Fatal("expected error when minio endpoint is missing")
|
||||
}
|
||||
}
|
||||
|
||||
+197
-31
@@ -32,29 +32,37 @@ import (
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"knowledge-governance/internal/librarydeadletter"
|
||||
"knowledge-governance/internal/libraryfreeze"
|
||||
"knowledge-governance/internal/librarymaintenance"
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||||
releaseworkflow "knowledge-governance/internal/workflows/release"
|
||||
)
|
||||
|
||||
type options struct {
|
||||
sourceApp string
|
||||
sourcePG string
|
||||
sourceES string
|
||||
governancePG string
|
||||
project string
|
||||
writers string
|
||||
nonWriters string
|
||||
archiveRoot string
|
||||
journalRoot string
|
||||
dockerSocket string
|
||||
maxDocuments int
|
||||
maxObjectBytes int64
|
||||
scope string
|
||||
job string
|
||||
dryRun bool
|
||||
snapshot string
|
||||
sourceApp string
|
||||
sourcePG string
|
||||
sourceES string
|
||||
governancePG string
|
||||
project string
|
||||
writers string
|
||||
nonWriters string
|
||||
archiveRoot string
|
||||
journalRoot string
|
||||
dockerSocket string
|
||||
// sourceS3Endpoint:容器里写的对象存储端点往往只在容器网络里能解析
|
||||
// (例如 minio 服务名),而封存跑在宿主机上。这里允许用宿主机可达地址覆盖。
|
||||
sourceS3Endpoint string
|
||||
maxDocuments int
|
||||
maxObjectBytes int64
|
||||
scope string
|
||||
job string
|
||||
dryRun bool
|
||||
snapshot string
|
||||
// releasedDeadLetters 是治理平台审计过的精确解除指纹(逗号分隔)。它只让这些
|
||||
// 确切的行不再阻断;实时行必须逐字节命中,否则照旧拒绝。
|
||||
releasedDeadLetters string
|
||||
}
|
||||
|
||||
func main() {
|
||||
@@ -90,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)
|
||||
@@ -125,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) 打印整库死信原始行与观测摘要,供治理平台审计
|
||||
`)
|
||||
}
|
||||
|
||||
@@ -155,12 +167,14 @@ func parseFlags(command string, args []string) (options, []string, error) {
|
||||
set.StringVar(&opts.archiveRoot, "archive", opts.archiveRoot, "sealed artifact root")
|
||||
set.StringVar(&opts.journalRoot, "journal", opts.journalRoot, "freeze journal root")
|
||||
set.StringVar(&opts.dockerSocket, "docker-socket", opts.dockerSocket, "Docker engine socket")
|
||||
set.StringVar(&opts.sourceS3Endpoint, "source-s3-endpoint", "", "host-reachable source object-store endpoint (overrides the container-internal one)")
|
||||
set.IntVar(&opts.maxDocuments, "max-documents", opts.maxDocuments, "maximum indexed documents per scope")
|
||||
set.Int64Var(&opts.maxObjectBytes, "max-object-bytes", opts.maxObjectBytes, "maximum bytes per stored object")
|
||||
set.StringVar(&opts.scope, "scope", "", "internal|external")
|
||||
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
|
||||
}
|
||||
@@ -264,7 +278,7 @@ func discoverSource(ctx context.Context, opts options, binding map[string]string
|
||||
config.SearchIndex = app["ELASTICSEARCH_INDEX"]
|
||||
config.Search = v071.SearchExporterConfig{BaseURL: "http://" + esAddress + ":9200", Index: app["ELASTICSEARCH_INDEX"],
|
||||
AuthorizationToken: app["ELASTICSEARCH_PASSWORD"], PageSize: 500}
|
||||
config.Object, config.ObjectDetail, err = objectStoreFromSourceEnv(app, opts.maxObjectBytes)
|
||||
config.Object, config.ObjectDetail, err = objectStoreFromSourceEnv(app, opts.sourceS3Endpoint, opts.maxObjectBytes)
|
||||
if err != nil {
|
||||
return config, err
|
||||
}
|
||||
@@ -276,7 +290,7 @@ func discoverSource(ctx context.Context, opts options, binding map[string]string
|
||||
// 只认 OSS 时,S3 兼容的 MinIO 目标(本机专用验收实例)会在发现阶段就被拒,
|
||||
// 验收链只能跑在生产存储上——这不是安全性,而是能力缺口。两种都要能读,
|
||||
// 读不出端点/桶就明确报错,不退回任何默认值。
|
||||
func objectStoreFromSourceEnv(app map[string]string, maxObjectBytes int64) (v071.S3ObjectStoreConfig, string, error) {
|
||||
func objectStoreFromSourceEnv(app map[string]string, endpointOverride string, maxObjectBytes int64) (v071.S3ObjectStoreConfig, string, error) {
|
||||
storage := strings.ToLower(strings.TrimSpace(app["STORAGE_TYPE"]))
|
||||
var config v071.S3ObjectStoreConfig
|
||||
var endpoint, bucket string
|
||||
@@ -312,13 +326,22 @@ func objectStoreFromSourceEnv(app map[string]string, maxObjectBytes int64) (v071
|
||||
if strings.TrimSpace(endpoint) == "" {
|
||||
return v071.S3ObjectStoreConfig{}, "", errors.New("source object endpoint is not configured")
|
||||
}
|
||||
// 宿主机可达地址优先:容器里的 minio 服务名在宿主机上根本解析不到,
|
||||
// 真跑时资源导出会直接失败(read library resource)。覆盖仍是"同一个桶",
|
||||
// 只换访问入口,不换来源。
|
||||
if override := strings.TrimSpace(endpointOverride); override != "" {
|
||||
if !strings.Contains(override, "://") {
|
||||
return v071.S3ObjectStoreConfig{}, "", errors.New("source object endpoint override needs a scheme")
|
||||
}
|
||||
endpoint = override
|
||||
}
|
||||
config.Endpoint = endpoint
|
||||
config.Secure = strings.HasPrefix(endpoint, "https://")
|
||||
config.ForcePathStyle = storage == "minio"
|
||||
if config.Bucket == "" {
|
||||
return v071.S3ObjectStoreConfig{}, "", errors.New("source object bucket is not configured")
|
||||
}
|
||||
return config, "provider=" + config.Provider + " bucket=" + bucket, nil
|
||||
return config, "provider=" + config.Provider + " bucket=" + bucket + " endpoint=" + config.Endpoint, nil
|
||||
}
|
||||
|
||||
func firstNonEmpty(values ...string) string {
|
||||
@@ -451,6 +474,78 @@ func openFreeze(ctx context.Context, opts options) (*libraryfreeze.Coordinator,
|
||||
return coordinator, control, nil
|
||||
}
|
||||
|
||||
// captureMaintenanceConfig 把"这次封存动的是哪个实例"落成实例级维护目标。
|
||||
//
|
||||
// 为什么必须实例级:被代理的 WeKnora 收到 SIGTERM 会直接 exit(1),写者的退出码
|
||||
// 因此永远不能当作"在途请求已排空"。排空证据只能在真正的入口上取——关掉闸门、
|
||||
// 等它把在途连接放完。没有闸门就没有证据,那种封存不允许以"库级窗口"的名义跑
|
||||
// (真跑过:被 exit=1 拒掉,这是校验在起作用,不是要放松的地方)。
|
||||
//
|
||||
// 入口/队列的声明与容器内 worker 共用同一组 LIBRARY_TARGET_* 名称,
|
||||
// 数据库/索引/桶则取自本次真实发现的源,避免"关的是 A 的门、读的是 B 的队列"。
|
||||
func captureMaintenanceConfig(opts options, source sourceConfig, get func(string) string) (librarymaintenance.Config, error) {
|
||||
target, err := librarymaintenance.TargetFromEnv(get)
|
||||
if err != nil {
|
||||
return librarymaintenance.Config{}, err
|
||||
}
|
||||
writers, nonWriters, err := writerDeclaration(opts)
|
||||
if err != nil {
|
||||
return librarymaintenance.Config{}, err
|
||||
}
|
||||
identity, err := librarymaintenance.DatabaseIdentity(source.DatabaseURL)
|
||||
if err != nil {
|
||||
return librarymaintenance.Config{}, err
|
||||
}
|
||||
return librarymaintenance.Config{
|
||||
Instance: target.Instance,
|
||||
Project: opts.project,
|
||||
Writers: writers,
|
||||
NonWriters: nonWriters,
|
||||
DockerSocket: opts.dockerSocket,
|
||||
IngressURL: target.IngressURL,
|
||||
IngressService: target.IngressService,
|
||||
IngressDrainFile: target.IngressDrainFile,
|
||||
Database: identity,
|
||||
Index: source.SearchIndex,
|
||||
Bucket: source.Object.Bucket,
|
||||
// 后台静默检查连的是"正在被封的这个源":未结清的后台账本与队列都在它身上。
|
||||
DatabaseURL: source.DatabaseURL,
|
||||
RedisAddr: target.RedisAddr,
|
||||
RedisPassword: target.RedisPassword,
|
||||
RedisDB: target.RedisDB,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// openMaintenance 用实例级目标装配协调器:入口闸门、后台静默检查、目标指纹、
|
||||
// 入口探针全部来自上面那份声明,与发布链路用的是同一套实现(不是第二份协议)。
|
||||
func openMaintenance(ctx context.Context, opts options, source sourceConfig) (*libraryfreeze.Coordinator, *libraryfreeze.ProjectController, *librarymaintenance.Manager, error) {
|
||||
config, err := captureMaintenanceConfig(opts, source, os.Getenv)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
manager, err := librarymaintenance.New(config)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
// 拓扑先过关:写入者发布了端口 = 存在绕闸直连,这样的"维护"只关了半扇门。
|
||||
if err = manager.VerifyTopology(ctx); err != nil {
|
||||
manager.Close()
|
||||
return nil, nil, nil, fmt.Errorf("目标拓扑不满足唯一入口约束,拒绝开维护窗口:%w", err)
|
||||
}
|
||||
coordinator, control, err := manager.Open(opts.journalRoot)
|
||||
if err != nil {
|
||||
manager.Close()
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
return coordinator, control, manager, nil
|
||||
}
|
||||
|
||||
// discoverMaintenanceSource 只读"这个源是什么",不碰治理库的 scope 绑定:
|
||||
// 恢复/收尾不该因为治理库读不到就被卡住。
|
||||
func discoverMaintenanceSource(ctx context.Context, opts options) (sourceConfig, error) {
|
||||
return discoverSource(ctx, opts, map[string]string{})
|
||||
}
|
||||
|
||||
func openArchive(opts options) (*librarysnapshot.Archive, error) {
|
||||
if err := os.MkdirAll(opts.archiveRoot, 0700); err != nil {
|
||||
return nil, err
|
||||
@@ -576,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)
|
||||
@@ -631,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)
|
||||
}
|
||||
@@ -643,10 +746,21 @@ func runCapture(ctx context.Context, opts options) error {
|
||||
return err
|
||||
}
|
||||
defer archive.Close()
|
||||
coordinator, control, err := openFreeze(ctx, opts)
|
||||
binding, err := scopeBindings(ctx, opts.governancePG)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
config, err := discoverSource(ctx, opts, binding)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// 维护窗口必须先按"实例级"接好:入口闸门给出在途排空证据,后台静默检查
|
||||
// 给出停写者那一刻队列已空。缺任何一项都拒绝封存——没有证据的封存不是候选。
|
||||
coordinator, control, manager, err := openMaintenance(ctx, opts, config)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer manager.Close()
|
||||
defer coordinator.Close()
|
||||
defer control.Close()
|
||||
if err = coordinator.Recover(ctx); err != nil {
|
||||
@@ -660,10 +774,6 @@ func runCapture(ctx context.Context, opts options) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
binding, err := scopeBindings(ctx, opts.governancePG)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
scopedKB := binding[opts.scope]
|
||||
if found {
|
||||
manifest, err := replayCapture(job, opts.scope, scopedKB, existing, archive)
|
||||
@@ -672,10 +782,6 @@ func runCapture(ctx context.Context, opts options) error {
|
||||
}
|
||||
return printCapture(job, existing.SnapshotID, manifest, true)
|
||||
}
|
||||
config, err := discoverSource(ctx, opts, binding)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
pool, err := openSourceDatabase(ctx, config)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -742,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
|
||||
@@ -957,10 +1063,17 @@ func runVerify(opts options) error {
|
||||
}
|
||||
|
||||
func runRecover(ctx context.Context, opts options) error {
|
||||
coordinator, control, err := openFreeze(ctx, opts)
|
||||
// 收尾也必须走同一份实例级目标:崩溃留下的窗口是"入口已关、写入者已停",
|
||||
// 恢复要用闸门把入口真的重新打开,没有闸门就没有"目标已恢复"的证据。
|
||||
source, err := discoverMaintenanceSource(ctx, opts)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
coordinator, control, manager, err := openMaintenance(ctx, opts, source)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer manager.Close()
|
||||
defer coordinator.Close()
|
||||
defer control.Close()
|
||||
if err = coordinator.Recover(ctx); err != nil {
|
||||
@@ -983,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
|
||||
}
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||||
)
|
||||
|
||||
func maintenanceEnv() map[string]string {
|
||||
return map[string]string{
|
||||
"LIBRARY_TARGET_INSTANCE": "kgmaint-weknora",
|
||||
"LIBRARY_TARGET_INGRESS_URL": "http://127.0.0.1:18080/health",
|
||||
"LIBRARY_TARGET_INGRESS_SERVICE": "ingress",
|
||||
"LIBRARY_TARGET_INGRESS_DRAIN_FILE": "/tmp/gate/drain.json",
|
||||
"LIBRARY_TARGET_REDIS_ADDR": "127.0.0.1:26379",
|
||||
"LIBRARY_TARGET_REDIS_PASSWORD": "secret",
|
||||
"LIBRARY_TARGET_REDIS_DB": "0",
|
||||
}
|
||||
}
|
||||
|
||||
func captureSource() sourceConfig {
|
||||
return sourceConfig{
|
||||
DatabaseURL: "postgres://kgmaint:pw@127.0.0.1:25432/kgmaint_target?sslmode=disable",
|
||||
SearchIndex: "kgmaint_1028a_chunks",
|
||||
Object: v071.S3ObjectStoreConfig{Provider: "minio", Bucket: "kgmaint-1028a"},
|
||||
}
|
||||
}
|
||||
|
||||
func captureOptions() options {
|
||||
return options{project: "kgmaint-weknora", writers: "app", nonWriters: "postgres=database store",
|
||||
dockerSocket: "/var/run/docker.sock", journalRoot: "/tmp/kgmaint-journal"}
|
||||
}
|
||||
|
||||
// 封存窗口必须是实例级:入口/队列来自与 worker 同一份 LIBRARY_TARGET_*,
|
||||
// 库/索引/桶来自本次真实发现的源——两边混用就会"关 A 的门、读 B 的队列"。
|
||||
func TestCaptureMaintenanceConfigBindsGateQueueAndSourceIdentity(t *testing.T) {
|
||||
env := maintenanceEnv()
|
||||
config, err := captureMaintenanceConfig(captureOptions(), captureSource(), func(name string) string { return env[name] })
|
||||
if err != nil {
|
||||
t.Fatalf("完整声明被拒:%v", err)
|
||||
}
|
||||
if config.Instance != "kgmaint-weknora" || config.Project != "kgmaint-weknora" {
|
||||
t.Fatalf("实例/项目不对:%+v", config)
|
||||
}
|
||||
if config.IngressURL != "http://127.0.0.1:18080/health" || config.IngressService != "ingress" ||
|
||||
config.IngressDrainFile != "/tmp/gate/drain.json" {
|
||||
t.Fatalf("入口闸门取值不对:%+v", config)
|
||||
}
|
||||
if config.RedisAddr != "127.0.0.1:26379" || config.RedisPassword != "secret" || config.RedisDB != 0 {
|
||||
t.Fatalf("队列取值不对:%+v", config)
|
||||
}
|
||||
// 口令只进结构体,不落进任何可打印身份:身份里只有 host:port/db。
|
||||
if config.Database != "127.0.0.1:25432/kgmaint_target" {
|
||||
t.Fatalf("数据库身份不对:%q", config.Database)
|
||||
}
|
||||
if strings.Contains(config.Database, "pw") {
|
||||
t.Fatalf("数据库身份泄漏了口令:%q", config.Database)
|
||||
}
|
||||
if config.Index != "kgmaint_1028a_chunks" || config.Bucket != "kgmaint-1028a" {
|
||||
t.Fatalf("索引/桶没取到源的身份:%+v", config)
|
||||
}
|
||||
if len(config.Writers) != 1 || config.Writers[0] != "app" || config.NonWriters["postgres"] == "" {
|
||||
t.Fatalf("写入者声明没透传:%+v", config)
|
||||
}
|
||||
// 后台静默检查必须连"正在被封的源"。
|
||||
if config.DatabaseURL != captureSource().DatabaseURL {
|
||||
t.Fatalf("后台静默检查连错库")
|
||||
}
|
||||
}
|
||||
|
||||
// 没有闸门就没有排空证据:必须在开窗口之前拒绝,而不是退回"只锁库"的旧模式。
|
||||
func TestCaptureMaintenanceConfigRefusesWithoutIngress(t *testing.T) {
|
||||
env := maintenanceEnv()
|
||||
delete(env, "LIBRARY_TARGET_INGRESS_URL")
|
||||
_, err := captureMaintenanceConfig(captureOptions(), captureSource(), func(name string) string { return env[name] })
|
||||
if err == nil || !strings.Contains(err.Error(), "LIBRARY_TARGET_INGRESS_URL") {
|
||||
t.Fatalf("缺入口闸门没有被拒:%v", err)
|
||||
}
|
||||
}
|
||||
@@ -54,8 +54,9 @@ func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, s
|
||||
target, err := v071.NewProductionLibraryTarget(ctx, v071.ProductionLibraryConfig{
|
||||
Postgres: config.LibraryProductionDatabaseURL, SearchURL: config.LibraryProductionElasticURL,
|
||||
SearchIndex: config.LibraryProductionIndex, Objects: objects, ObjectsReader: objects,
|
||||
ArchiveRoot: config.LibraryArchiveRoot,
|
||||
MaxDocuments: 200000,
|
||||
ArchiveRoot: config.LibraryArchiveRoot,
|
||||
ObjectsBucket: config.LibraryProductionS3Bucket,
|
||||
MaxDocuments: 200000,
|
||||
}, identity)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
|
||||
+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.SourceInstance())
|
||||
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,不会因为作业失败而假装来源已恢复。
|
||||
|
||||
@@ -52,7 +52,8 @@ LIBRARY_ARCHIVE_ROOT=/var/lib/knowledge-governance/library
|
||||
「目标维护未接线,拒绝在挡不住入口/后台写入的情况下写目标」(`cmd/worker/library_handlers.go`)——**不会**退化成"库级锁假装维护"。
|
||||
- 缺 `LIBRARY_PRODUCTION_DATABASE_URL` 时 overview/差异按"未接入"显示,不显示 0 或被当成"无变化"。
|
||||
- `/api/v1/library/maintenance` 在实例未配置时返回 `instance` 缺省、`held=false`;页面显示「未接入 + 维护类操作 fail closed」。
|
||||
- `LIBRARY_AGENT_SOCKET` 指向的 agent 目前**没有运行**:`POST /api/v1/library/captures` 会 503 并写明"未接线",不会假装已封存。
|
||||
- `LIBRARY_AGENT_SOCKET` 指向的 agent 必须在部署时一起起来(`scripts/library-agent-install.sh` + 0600 `agent.env`);
|
||||
端点/闸门/队列声明与 worker 同源(`LIBRARY_TARGET_*`)。缺任何一项时封存会在开窗口前被拒并点名缺什么。
|
||||
|
||||
新增配置只在需要真正开维护窗口时才加(本机验收实例的取值可作模板,见
|
||||
`evidence-maintenance/harness/candidate-maintenance.env`:`LIBRARY_TARGET_INSTANCE/PROJECT/WRITERS/NON_WRITERS/DOCKER_SOCKET/INGRESS_*`、`LIBRARY_TARGET_REDIS_*`)。
|
||||
@@ -85,6 +86,8 @@ LIBRARY_ARCHIVE_ROOT=/var/lib/knowledge-governance/library
|
||||
## 7. 已知风险与前置条件(必须先解决或明确接受)
|
||||
|
||||
1. **`knowledge-governance-p1-release-worker-1` 处于重启循环**(日志每分钟 `connect release target database`,`RELEASE_WORKER_ENABLED=false`)。这是正式栈既有状态,不是本轮引入;部署前需要决定是修配置还是先停掉该服务,否则会持续刷日志。
|
||||
2. **封存能力未接线**(见矩阵"缺代码"):正式上线后页面能如实显示"未接入",但"封存测试版本"按钮仍不可用。如果验收要求现场封存,需要先部署 `libraryagent`(`cmd/libraryagent`,0600 unix socket)并把 `LIBRARY_AGENT_SOCKET` 指到它。
|
||||
2. **封存能力已在本机专用实例验证**(2026-10-29:页面入口 → agent → 实例级窗口 → 同步 → 恢复)。
|
||||
正式部署仍需按同一形状接线:`libraryagent`(0600 socket)+ `agent.env`(`LIBRARY_TARGET_*`)+ `LIBRARY_AGENT_SOCKET`;
|
||||
这套接线在正式栈上尚未执行。
|
||||
3. **入口排空探针未在最终构建复现**(矩阵"已实现·未跑"):正式部署前若要拿这条证据,需要在生产机重复探针,并接受"闸门关得比客户端快"导致上传注入失败的可能。
|
||||
4. 部署本身就是生产变更:需要用户给的窗口与授权,本文不构成执行许可。
|
||||
@@ -1,6 +1,7 @@
|
||||
# 整库同步验收矩阵(真实证据对照,2026-10-29 重排)
|
||||
|
||||
本文件是**唯一口径**:状态只写实际发生的事,四类严格分开,不用"部分通过"含糊过去。
|
||||
**2026-10-29 父审后重跑:V03-a(Wiki 撤销语义)不通过,单独列在「一·补」,不在「已完成」表里。**
|
||||
|
||||
| 口径 | 含义 |
|
||||
| --- | --- |
|
||||
@@ -42,29 +43,37 @@ A06(内容、问答、引用图片效果、性能报告)已由用户取消
|
||||
| D05 | 容量门禁 | `CheckCapacity` + sync gate | 预算未设置→阻断;超预算→阻断并给预计字节 |
|
||||
| D06 | 测试残留口径 | 未引用 blob 统计 | 真实归档未引用 blob 为 0(口径,不等于扫过全部残留) |
|
||||
| **S03** | 实例级维护窗口:关入口→排空→静默证明→停写者→写入→校验→开入口 | `internal/libraryfreeze` + `librarymaintenance` + 迁移 070–072 | 前轮 7 场景真跑全绿(normal/kill/applyfail/partial/crossscope/recover_refuse/recover_ok),本轮 sync/drain 场景继续复用;`quiesce` 证据、`post_stop_*`、journal、gate drain.json 均可回看 |
|
||||
| **V03-a** | 撤销一份共享贡献后,保留另一份贡献、清除撤销方独有正文/引用/实体边、目录正确 | 由 WeKnora 源侧执行(`cleanupWikiOnKnowledgeDelete` + `EnqueueWikiRetract`),治理侧只比对/搬运 | **本轮真实验收**:kgmaint 隔离实例、原版 API 顺序入库两篇文档 → `entity/fa-bu-zha-men` 同页双来源(2097 字节)→ 真实 `DELETE /api/v1/knowledge/{id}` → 页面保留、来源只剩保留文档、A 独有正文(ZEBRA-A-7731/9910)清除、B 贡献仍在、引用 2→1 且只剩 B 的分块、目录不变、A 独有页面/摘要整页删除、索引页不再列出、实体边无死链。13/13 检查通过。证据:`kg-candidate-e2e/evidence-wiki-final/s8-final-checks-r1.json` + `06-wiki-*-r1.json` + `s3/s6-db-*.json` |
|
||||
| **V03-b** | 治理侧能把上述语义变化说清楚(不是页面级"无变化") | `internal/librarydiff/wiki.go` `CompareWiki`;`GET /library/overview` 的 `diff.wiki` | 逐页比对正文/来源/引用/出入边/目录,输出 `sources_removed`、`citations_before→after`、`changed`;单测 `wiki_test.go`、视图单测 `library_wiki_diff_test.go`;页面逐行展示(`wikiContributionLines` + 测试) |
|
||||
| **S06** | 页面/API 发起封存 → 主机 agent → 实例级窗口封存(真实冻结) | `POST /library/captures` + `cmd/libraryagent`(0600 socket、固定命令)+ `libraryctl capture`(实例级:闸门/后台静默/指纹) | **本轮终态(0.1.186 e508abd)真跑**:`POST /api/v1/library/captures`(job `lib-internal-20261029T0716`)202 → worker → agent → CLI → 关闸门排空 → 等后台静默 → 停 `kgmaint-app`(exit=1、`drain_verified=true`、`stop_verified=true`、`restored=true`)→ 导出 → 恢复写入者与入口;作业 `libseal-lib-internal-20261029T0716` succeeded,快照 `0f0a3b68…`。证据 `kg-final-gaps/evidence/04-inflight-v05-final.json`、journal `freeze-kgmaint/lib-internal-20261029T0716.json` |
|
||||
| **S06-a** | 排空证明取自真实在途请求(不靠 202 后抢时序) | 闸门 `drain.json` + 客户端真实上传握手 | **本轮终态真跑**:先发一半 multipart 正文 → 闸门 `conn:open active=1` → 再发起封存 → 入口关闭 → 补完正文,应用 200 并完成解析/嵌入 → `drain.json phase=drained active_at_close=1 served=143 drain_ms=352`;journal `quiesce.samples=18 stable_ms=1617` 且停写者后指纹不变(`post_stop_digest` == `digest`)。窗口内派生任务真的跑完(探针文档 `f7fed988…` completed / 7 分块),且该文档出现在被封存的快照里 |
|
||||
| **V03-c** | 带"撤销后共享页"的整库封存 → 独立目标同步 → 恢复,逐项真实读回 | `libraryctl` 封存 + `ProductionLibraryTarget` + `libraryrelease` | **本轮终态真跑(字节一致性,不是撤销语义)**:KB `e64e82b5`(9 Wiki 页、1 文档、2 分块、1 资源)→ 快照 `0f0a3b68…` 与源逐页一致(`05-snapshot-compare.json` ok=True)→ 同步到独立实例 kgseal `libsync-internal-1793229249632461601` succeeded → 目标逐页/索引/资源字节一致(`07-target-compare.json` ok=True,含目标桶对象 sha256)→ `librestore-internal-1793229274077866298` succeeded → 目标仍逐项一致。**注意**:该共享页正文仍带撤销方残留(见 V03-a 不通过),本行只证明"搬运/恢复逐字节一致" |
|
||||
| **S06-b** | 主机侧封存/恢复必须与 worker 讲同一份维护声明 | `internal/librarymaintenance/env.go`(`LIBRARY_TARGET_*` 唯一解析处) | 缺项时 `capture` 在开窗口前拒绝并点名;agent 启动打印维护目标;单测 5 项(逐项缺失/空白/口令空格/队列库号) |
|
||||
| **Neo4j 启用场景** | 启用图谱的库必须被真实拒绝 | `v071/library_export.go` 读 `indexing_strategy.graph_enabled` | **本轮真跑**:在 kgmaint 建 `graph_enabled=true` 的真实 KB `8de41ca7…`,用真实 CLI 预检 → 退出码 1、`LIBRARYCTL_FAILED: Neo4j graph snapshot is not supported`(`evidence/10-neo4j-refuse.txt`)。不是 skip |
|
||||
| **U04** | 维护级别/进度/恢复错误真实呈现 | `MaintenanceCard`(`web/src/ReleaseTargets.tsx`)、`/api/v1/library/maintenance` | 显示实例级/库级、持有 scope+作业、阶段、入口开/关、服务、开始/释放、进度(阶段名中文化)、恢复错误与处理记录;仅"窗口仍持有且作业失败"时给"按预备份恢复";无记录时"未接入 + fail closed"。组件测试 6 项 |
|
||||
|
||||
| **V05** | 导出中源变更必须被入口/写入者阻断,切点一致 | 闸门 `drain.json` + 真实并发客户端 | **本轮终态真跑**:入口关闭的同一刻(固定协调点,非 sleep)对专用源发 8 写 + 4 读 → 8 写全部失败(7 `URLError`/1 `RemoteDisconnected`)、4 读 `RemoteDisconnected`;写入者被停下的那一刻(轮询容器状态捕获 `running→exited→running`)再发 4 写 + 2 读 → 全部 `URLError`。事后源库 `v05` 文档 0 / 分块 0,快照与源逐项一致(ok=True),在途文档 `f7fed988…` completed/7 分块且出现在快照里。证据 `04-inflight-v05-final.json` |
|
||||
| **B02** | 源测试桶删对象后,归档自包含且目标应用能读回同样字节 | 归档 `resource` 部件 + 目标应用下载接口 | **本轮终态真跑**:源对象 `10001/d4fbe69d…/ab2b628f….md`(507B,sha256 `25a55cd8…`)先备份再 `mc rm`(stat 已 404)→ 同步仍 succeeded(字节只可能来自归档)→ 目标桶 `kgseal-1029a` 同键 507B、sha256 一致 → **目标应用自己的** `GET /api/v1/knowledge/{id}/download` 恢复前后都 200 / 507B / sha256 `25a55cd8…`。修复前该接口真实 500(`code 1007 invalid MinIO file path`),见 0.1.184/0.1.185 两条修复。证据 `11-b02-delete-and-sync.json`、`12-b02-restore-download.json`、`07-target-compare.json` |
|
||||
| **死信精确解除(治理侧)** | 保留上游死信行,只按精确指纹解除 | 迁移 073 + `internal/librarydeadletter` + `POST/GET /library/dead-letter-releases` + `libraryctl dead-letters` + `-released-dead-letters` | **已实现并有真实验证的部分**:`libraryctl dead-letters` 在真实源上读出 0 条死信与观测摘要;治理 API `GET` 返回 `instance=kgmaint-weknora`、`releases=[]`;`POST` 无写意图头 403;带真实观测但引用不存在的死信 → 409「观测里没有这条死信」。单测覆盖:摘要被改写、源侧未结清、重放早于失败、重放后又出现同键新死信、实例不一致、通配指纹。**未演练**:真实死信→重放→解除→放行(本轮没有死信,也没有源侧管理员凭据) |
|
||||
|
||||
## 一·补 失败 / 不通过(不得标通过)
|
||||
|
||||
| ID | 要点 | 承载位置 | 本轮真实结果 |
|
||||
| --- | --- | --- | --- |
|
||||
| **V03-a** | 撤销一份共享贡献后,保留另一份贡献、清除撤销方独有正文/引用/实体边、目录正确 | 由 WeKnora 源侧执行(`cleanupWikiOnKnowledgeDelete` + `EnqueueWikiRetract`),治理侧只比对/搬运 | **本轮不通过(2026-10-29 重跑)**:真实 `DELETE /api/v1/knowledge/{id}` 后,共享页 `entity/fa-bu-zha-men` 的来源/引用/边/目录都按上游 retract 更新(source_refs 只剩保留文档、引用 2→1、9 页/18 边无死链),**但正文仍保留撤销方独有的两句事实 `ZEBRA-A-7731`/`ZEBRA-A-9910`**(version=3、3733 字节、sha256 `5fd27348…`)。封存逐字节一致只说明"存了真实状态",不等于撤销语义通过。证据 `kg-final-gaps/evidence/14-wiki-residue-evidence.json` + `01-wiki-source-after-r3.json` |
|
||||
|
||||
## 二、缺代码
|
||||
|
||||
| ID | 缺口 | 现状与影响 |
|
||||
| --- | --- | --- |
|
||||
| R03 | 存快/恢复快对比 | 没有测速实现,也未测。不写没有测量的数字 |
|
||||
| 死信解除契约 | 死信只能靠人工清理,没有"解除/豁免"契约 | WeKnora `task_dead_letters` 无解决标记;导出/封存按"存在即阻断"(`library_source_settlement.go`)。本轮真实验证:早期错误模型配置产生的 2 条 wiki 死信直接阻断整实例结算,清理方式只能是删除行(记录:`deadletters-before-delete.json`,sha256 `dd4d8f08…`)。要变成可运维契约,需要上游给出解决标记或治理侧设计"带证据的解除"——**未实现** |
|
||||
| 封存入口接线(本机) | `POST /library/captures` 需要 `LIBRARY_AGENT_SOCKET` 指向真实 agent | 候选/正式 api 都配置了 socket 路径,但本机没有运行 `libraryagent`,接口会 503 并明示"未接线"(不是静默失败) |
|
||||
| CLI 实例级封存 | `libraryctl capture` 只建库级窗口,不关入口 | 真跑结果:WeKnora 收到 SIGTERM 固定 `exit=1`,没有入口排空证明时被拒(`未优雅退出(exit=1)`)。要在"不停共享栈"的前提下封存真实撤销后的库,需要把实例级 + 闸门接进 CLI(或部署 agent)——**未实现** |
|
||||
| R03 | 存快/恢复快对比 | A06 性能报告已由用户取消,不作为门槛。**已记实测耗时**(小测试库,不是压测):封存 15.8s(含停写~5s)、同步 13.6s(预备份 13.4s 占主要)、恢复 13.8s;证据 `13-r03-durations.json`。没有做高负载压测,也不写没有测量的数字 |
|
||||
| 死信解除·真实重放演练 | 没有可解除的真实死信,也没有源侧系统管理员凭据 | 契约与执行侧已实现(见下节新增行):上游保留原行、治理侧审计+指纹绑定、未结清/变化/新死信仍阻断。**本轮没有真实死信**(`dead_letters=0`),无法演练"真实重放成功→解除→放行"整条链路;上游只有 `系统管理 runtime run_now`(需要源侧管理员权限)与整队列 purge(销毁证据,不采用)。治理平台故意不持有源侧管理员凭据(放大授权),因此这一步**阻塞在真实死信 + 重放执行方**,不伪造成功 |
|
||||
|
||||
## 三、已实现·未跑
|
||||
|
||||
| ID | 要点 | 为什么未跑 |
|
||||
| --- | --- | --- |
|
||||
| V05 | 导出中源变更则拒绝或隔离 | 预备份导出前后行数静默检查已有(本机验证会拒绝);真实并发写入的隔离演练未做 |
|
||||
| B02 | 资源在测试侧删除后仍可恢复 | 归档内容寻址已实现;"删测试侧对象再从归档恢复"实测未做 |
|
||||
| S03-b | 死信/孤儿任务阻断的负向演练(本轮只验证了真实死信阻断) | 真实阻断已发生并记录;系统化的负向矩阵未跑 |
|
||||
| **V03-c** | **带"撤销后共享页"的整库封存 → 同步 → 恢复** | 上游语义已验证、治理差异已实现;但这台机器上 CLI 封存止步于上面的 exit=1 缺口,API 封存未接线,所以"封存一份含撤销结果的 Wiki 版本,再同步/恢复并逐行比对"**没有产出真实证据**。旧证据 `drain-task-new-c`(03:39)属于 0ef7984 之前的构建,不能拿来证明当前 SHA |
|
||||
| **Neo4j 启用场景** | 代码在 `library_export.go` 检查 `graph_enabled`,启用即 `Neo4j graph snapshot is not supported` 拒绝 | 本轮没有真跑"启用 Neo4j 的库被拒":kgmaint 实例 `NEO4J_ENABLE=false`、测试库 `graph_enabled=false`。代码路径在,真实执行未做 |
|
||||
| **排空竞态探针(最终构建)** | 关入口→排空期间产生的真实任务必须等到队列归零才停写者 | 本轮在最终构建(1faf513)上重跑 6 次(`drain-task-final-1faf513{,-r2,-r3,-r4,-r5,-r6}.json`),上传请求全部 `Broken pipe`、队列深度始终 0:闸门在 202 之后 <0.2s 就关监听,客户端来不及把请求送进去。**未复现**,不记为通过;旧 `drain-task-new-c` 的时间线仍是旧构建证据 |
|
||||
| 排空时序(历史结论) | 旧做法"202 之后抢时序上传"在 1faf513 上重跑 6 次全部 `Broken pipe`(`drain-task-final-1faf513*.json` 保留,不记通过) | 已被 S06-a 的确定性做法取代:先证明请求已被闸门接收,再启动维护。旧证据保留,不覆盖 |
|
||||
| U05 | 正式 8088 页面验收 | 正式栈仍是 0.1.114,本轮未替换;维护面板/逐页 Wiki 差异只在测试与候选构建里验证过 |
|
||||
|
||||
## 四、需授权 / 需环境
|
||||
@@ -76,11 +85,29 @@ A06(内容、问答、引用图片效果、性能报告)已由用户取消
|
||||
| 生产目标维护 | 对共享业务栈开维护窗口 | 本轮明确不停 `WKNFull-app`;维护只在 kgmaint 专用实例上跑 |
|
||||
| 生产封存/同步真实链路 | 需要在生产机部署 `libraryagent` 并接线 | 需要用户批准的部署窗口与凭据录入 |
|
||||
|
||||
### 四·补 运维风险(本轮核对,未擅自处理)
|
||||
|
||||
正式 8088 栈(`knowledge-governance-p1-api-1` / `-worker-1`)**当前引用同一个 agent socket**:
|
||||
api env `LIBRARY_AGENT_SOCKET=/run/library-agent/agent.sock`,worker 挂载
|
||||
`/home/zheromiku/.local/share/knowledge-governance/agent -> /run/library-agent`。而主机侧 user 单元
|
||||
`~/.config/systemd/user/knowledge-governance-library-agent.service` 本轮已把目标改成 kgmaint 验收实例
|
||||
(`-source-app kgmaint-app`、`-project kgmaint-weknora`、`-archive /tmp/kg-e2e/archive`,
|
||||
原 WKNFull 版备份为 `*.wkfull.bak`)。后果:**从正式 8088 发起封存会冻结测试实例,并把快照写进
|
||||
正式 API 看不到的归档目录**(正式 api 的 `LIBRARY_ARCHIVE_ROOT=/var/lib/knowledge-governance/library`)。
|
||||
本轮没有启动任何共享维护,也没有改回该单元——需要用户决定:正式上线前把单元恢复为 WKNFull 版
|
||||
(备份已在位),或给候选构建单独的 socket 路径。证据 `00-identity-final.json` 的 `formal_stack` 段。
|
||||
|
||||
另外:`knowledge-governance-p1-release-worker-1` 自 2026-10-10 起就在崩溃重启(`exit=1`、
|
||||
`RestartCount≈1233`,日志只有 `connect release target database`),与本轮改动无关(它没有挂 agent socket),
|
||||
但正式栈的发布 worker 实际不可用,需要单独处理。
|
||||
|
||||
---
|
||||
|
||||
## 五、本轮新增结论(写在这里而不是报告顶部)
|
||||
|
||||
1. **Wiki 撤销语义是上游行为,必须用上游 API 验**:顺序入库才产生"两篇共贡献一页"(并行入库会各自抽到不同 slug 生成两页,这是上游已知行为)。撤销后页面保留、贡献按文档摘除,12 项检查全部通过。
|
||||
1. **Wiki 撤销语义本轮不通过,且不是治理侧能自己补的**:上游 retract 走 LLM 重写,重跑一次留下两句撤销方独有事实(`ZEBRA-A-7731`/`ZEBRA-A-9910`),另有 r1 轮次把两句清掉的记录——同一输入两次结果不同。原版 v0.7.1 的 Wiki API 只有页 CRUD、`rebuild-links`(只重算 `[[slug]]` 边,metadata-only)、`auto-fix`(只修 lint 可修项),**没有"按剩余来源重算页面正文"的安全 API**(`internal/router/router.go:2354` 起的路由表 + `service/wiki_page.go:759`)。因此:①不得把 V03 标通过;②不得用关键词硬删掩盖;③可选方案见交付报告(外部最小治理:保留旧版本+关联撤销作业+受影响页重建或阻断发布+验证与审计,需用户拍板产品语义)。
|
||||
2. **治理侧不能只报页面级差异**:撤销共享贡献时页面仍是"更新",页面级计数会显示成"无变化"。已改为逐页语义比对,能指出"来源 -1、引用 2→1、正文变了、出边少了一条"。
|
||||
3. **封存链路是本机当前的真缺口**:CLI 封存被 `exit=1` + 无入口排空证明挡住,API 封存没有 agent。这不是"Wiki 语义没验",而是"带撤销结果的版本没法在本机封出来"。正式部署必须先接线 agent(见 `docs/formal-8088-deployment-plan.md`)。
|
||||
4. **死信不是可忽略的告警**:一条失败的 wiki op 就足以让整实例结不了算。要么上游给解决标记,要么治理侧设计带证据的解除,当前两者都没有。
|
||||
3. **封存链路本轮真的打通了,而且不是放松校验**:CLI 改为实例级窗口(闸门 + 后台静默 + 目标指纹),"写者 exit=1"由闸门的 `drained/active_at_close` 证据解释,而不是被忽略;排空证明改成"先在途、再关入口"的确定性顺序,不再靠 202 之后抢时序。
|
||||
4. **接线后暴露的四个真缺陷(都已修,都有用例)**:`jobs.id` 列宽 36 让封存作业根本投不进去;v0.7.1 的 app 写索引不带 `tenant_id` 导致导出 0 条;全新实例的空索引被当成"索引被改坏";恢复作业 id 带 `.` 被维护窗口拒。另有资源定位符仍写来源桶——同步过去的目标取不到图/附件。
|
||||
5. **死信解除已从"缺代码"变成"有代码、真实验证了一半"**:迁移 073 记录精确解除(保留上游行、绑定实时行指纹、重放引用、源侧观测摘要、操作人);封存时每条实时死信都必须被精确解除命中,行变化/新行/未结清照旧阻断;`libraryctl dead-letters` 生成只读观测,`libraryctl -released-dead-letters` 逐字节复核。**没有演练的部分**是"真实死信→重放→解除→放行":本轮源侧 `dead_letters=0`,且重放需要源侧系统管理员权限(`/system/admin/runtime/queues/{queue}/tasks/{id}/actions/run_now`),治理平台故意不持有该凭据。不伪造成功,也不 SQL 删行冒充。
|
||||
6. **同步过去的资源"能被应用读出来"是这轮才真正验的**:资源行里既有 `physical_path` 也有独立的 `storage_backend_id` 列,两者都指向来源实例的 `storage_backends` 主键;只换桶或不换列,目标应用都 500(`code 1007 invalid MinIO file path`)。已在同一份行变换里同时改后端 id 与桶,解析不到就拒绝写入。
|
||||
@@ -45,6 +45,33 @@
|
||||
|
||||
同一个作业号重跑是**幂等重放**:直接返回已封存的同一份候选(`replay=true`、`source_read=false`、`writers_stopped=false`),不读来源、不停写入者;同一作业号换了 scope 或绑定库则明确拒绝。
|
||||
|
||||
## 实例级维护接线(封存窗口的唯一入口证据)
|
||||
|
||||
`capture` 与 `recover` 现在都按**实例级维护**开窗口,接线来自与 worker 同一组环境变量
|
||||
(名字只在 `internal/librarymaintenance/env.go` 出现一次):
|
||||
|
||||
```
|
||||
LIBRARY_TARGET_INSTANCE 实例名(写进维护记录与目标指纹)
|
||||
LIBRARY_TARGET_INGRESS_URL 入口地址(恢复后要能 200)
|
||||
LIBRARY_TARGET_INGRESS_SERVICE Compose 里的入口闸门服务名
|
||||
LIBRARY_TARGET_INGRESS_DRAIN_FILE 闸门写排空证据的宿主机文件
|
||||
LIBRARY_TARGET_REDIS_ADDR/_PASSWORD/_DB 后台队列(asynq)地址
|
||||
```
|
||||
|
||||
另外 `-source-s3-endpoint <URL>` 用来覆盖"源对象存储的访问入口":容器里写的
|
||||
`MINIO_ENDPOINT` 往往是服务名(`langfuse-minio:9000`),封存跑在宿主机上时解析不到,
|
||||
资源导出会失败(`read library resource`)。覆盖只换入口,桶与对象键不变;`status`
|
||||
会如实打印实际使用的 endpoint。
|
||||
|
||||
为什么不再保留"库级窗口":被代理的 WeKnora 收到 SIGTERM 会直接 `exit(1)`,写者退出码
|
||||
永远不能当作"在途请求已排空"。排空证据只能在真正的入口上取——关闸门、等它把在途连接
|
||||
放完(`drain.json` 里的 `active_at_close` / `drained`)。因此缺少上面任何一项时 `capture`
|
||||
会**在开窗口前拒绝**(报出缺少的变量名),而不是退回一个没有排空证据的窗口。
|
||||
|
||||
开窗口的顺序仍是:关入口并等在途排空 → 证明后台静默(稳定窗口 + 任务集合指纹)→ 停写入者
|
||||
→ 停后复核队列/指纹与入口 → 导出 → 恢复写入者与入口。停写者后仍会**再查一次**源是否结清,
|
||||
所以窗口内派生的真实任务必须真的跑完,否则封存被拒(不是等一会儿就算了)。
|
||||
|
||||
## 从页面发起封存(主机侧 agent)
|
||||
|
||||
页面点「发起对内/对外封存」只是**投递一个作业**;执行发生在这台主机上的 agent:
|
||||
@@ -52,6 +79,10 @@
|
||||
- `scripts/library-agent-install.sh` 构建并安装 `libraryagent`(systemd **user** 服务),
|
||||
监听 `~/.local/share/knowledge-governance/agent/agent.sock`,目录 0700、socket 0600,
|
||||
并额外用 `SO_PEERCRED` 要求对端与 agent 同 uid。
|
||||
- 维护接线由 `~/.local/share/knowledge-governance/agent.env`(0600,`EnvironmentFile=`)提供;
|
||||
没有这个文件安装脚本直接拒绝(`AGENT_ENV_MISSING`),免得装出一个"没有闸门"的 agent。
|
||||
源容器/项目/写入者声明可用 `AGENT_SOURCE_APP`、`AGENT_SOURCE_PG`、`AGENT_SOURCE_ES`、
|
||||
`AGENT_PROJECT`、`AGENT_WRITERS`、`AGENT_NON_WRITERS` 覆盖,默认是本机 `weknora071full`。
|
||||
- agent 自己固定 `libraryctl` 的路径、归档与日志根、Compose 项目与写入者声明;
|
||||
请求只能选 `scope`(internal/external)和作业号,**不能**指定路径、项目、容器或命令。
|
||||
- 只有 worker 通过挂载的 socket 目录投递作业(`LIBRARY_AGENT_SOCKET`)。容器里
|
||||
@@ -76,6 +107,19 @@
|
||||
|
||||
两次封存都记录了 `stop_verified=true`、`restored=true`,`WKNFull-app` 恢复后 healthy;来源未做任何写操作。
|
||||
|
||||
## 2026-10-29 实测:页面入口 → agent → 实例级窗口(真实撤销后的库)
|
||||
|
||||
- 请求:`POST /api/v1/library/captures {scope:internal, job:lib-internal-20261029T0614}` → 202,
|
||||
worker 投递 agent,CLI 以**实例级窗口**执行。
|
||||
- 排空证据(确定性顺序,不抢时序):先发一半 multipart 正文 → 闸门日志 `conn:open active=1`
|
||||
→ 再发起封存 → 入口拒绝新连接 → 补完正文,应用返回 200 并完成解析/嵌入 →
|
||||
`drain.json`:`phase=drained active_at_close=1 served=126 drain_ms=302`。
|
||||
- 冻结:`kgmaint-app` 收到 SIGTERM 后 `exit=1`(上游行为),因入口已排空,`drain_verified=true`;
|
||||
后台静默 `quiesce.samples=18 stable_ms=1617`,停写者后指纹不变;随后恢复写入者与入口。
|
||||
- 产物:快照 `1196c518…`(KB `e64e82b5…`,tenant 10001):1 文档 / 2 分块 / 9 Wiki 页 / 1 资源,
|
||||
与源库逐页比对(正文 md5、来源、引用、目录、18 条边)全部一致。
|
||||
- 同一份快照随后同步到**独立实例**(自己的 PG/索引/桶)并成功恢复,逐项读回一致。
|
||||
|
||||
## 它证明了什么 / 没有证明什么
|
||||
|
||||
已证明(A04 数据完整性口径):这一份整库内容在**写入者停住**之后被完整导出,PostgreSQL 与 Elasticsearch 双向覆盖、生成问题、资源与摘要逐项核验通过,产物可寻址、可复读、可校验。这正是封版所需的完整性口径。
|
||||
|
||||
+27
-1
@@ -68,7 +68,33 @@ run164 因 qawemail 依赖外网 DNS 失败,已修,不涂改。
|
||||
| 0.1.166 | 1faf5137 | 用完整容器 ID 判定写入者归属(**验收代码 SHA**) | 207 | **success** | 否 |
|
||||
| 0.1.167 | f82eb073 | 矩阵四段重排 + 8088 待验收部署方案 | 208 | **success** | 否 |
|
||||
|
||||
当前候选 = 0.1.166(1faf513);其后的 f82eb07 只改文档,未改代码,验收证据的 code_sha 仍是 **1faf513**。
|
||||
| 0.1.169 | ee7688f2 | 矩阵改正 Wiki 检查计数(上轮交付写作 0.1.168,按 `git rev-list --count v0.1.0..HEAD` 实为 0.1.169,此处以台账为准) | 210 | **success** | 否 |
|
||||
| 0.1.170 | 11c52ad8 | 封存窗口接上实例级闸门/后台静默(**本轮第一个代码提交**) | 211 | failure(密钥扫描把"环境变量名字面量"误判为密钥;该提交之后才在下一提交改名修复) | 否 |
|
||||
| 0.1.171 | 37a8370f | 索引范围过滤接受未写 `tenant_id` 的文档 | 212 | **success** | 否 |
|
||||
| 0.1.172 | f2d6f3f6 | 封存队列作业 id 按 `jobs.id` 列宽拼短 | 213 | **success** | 否 |
|
||||
| 0.1.173 | 4d6ecda3 | 源对象存储端点可按宿主机地址覆盖 | 214 | **success** | 否 |
|
||||
| 0.1.174 | faaf0037 | agent 固定命令形态纳入对象存储端点 | 215 | failure(漏跑 gofmt,下一提交修) | 否 |
|
||||
| 0.1.175 | 84e8a0df | gofmt | 216 | **success** | 否 |
|
||||
| 0.1.176 | 7d4ffdbf | 全新目标实例的空索引按空基线 | 217 | **success** | 否 |
|
||||
| 0.1.177 | cb009080 | 同步资源行定位符换成目标桶 | 218 | completed/cancelled(被后续提交取消,不计) | 否 |
|
||||
| 0.1.178 | 2e39e495 | 移出一次性核对小工具 | 219 | **success** | 否 |
|
||||
| 0.1.179 | b061a968 | 资源换桶变换在校验时同用(**封存闭环验收代码 SHA**) | 220 | **success** | 否 |
|
||||
| 0.1.180 | a6406209 | 恢复作业 id 改成维护窗口可接受的形状 | 221 | **success** | 否 |
|
||||
| 0.1.181 | bc49b85 | 文档:封存闭环进"已完成"、补死信最小外部契约 | 222 | **success** | 否 |
|
||||
| 0.1.182 | 958292d | 文档:按实际运行的二进制记封存闭环时序 | 223 | **success** | 否 |
|
||||
| 0.1.183 | c2402947 | 死信按精确指纹解除(保留上游行、审计、指纹绑定) | 224 | **success** | 否 |
|
||||
| 0.1.184 | 1aecd9d6 | 资源定位符同时改到目标存储后端 | 225 | **success** | 否 |
|
||||
| 0.1.185 | c542a133 | 资源行 storage_backend_id 列也改到目标实例 | 226 | **success** | 否 |
|
||||
| 0.1.186 | e508abd9 | 死信解除绑定封存源实例(LIBRARY_SOURCE_INSTANCE) | 227 | **success** | 否 |
|
||||
|
||||
本轮(父审后)真实验收运行的最终代码 SHA 是 **0.1.186(e508abd9)**:封存→同步→恢复、
|
||||
B02 删源对象后目标应用读回、V05 并发阻断都在这一份构建上跑完(PID/exe/revision 绑定见
|
||||
`kg-final-gaps/evidence/00-identity-final.json`)。0.1.181–0.1.183 是同一轮内的连续修复,
|
||||
不作为最终验收 SHA。CI run 204(958292d,本轮起点)success。
|
||||
|
||||
封存闭环的运行时序(按实际运行的二进制记):封存 API→agent→冻结出快照在 **0.1.175(84e8a0d)**;
|
||||
首次同步成功在 **0.1.179(b061a968)**;恢复与最终逐项读回在 **0.1.180(a6406209)**。
|
||||
最终 HEAD 的代码提交即 0.1.180(其后 bc49b85 只改文档,CI run 222 success)。
|
||||
本轮真实验收证据:`/home/zheromiku/.hermes/handoffs/kg-candidate-e2e/evidence-wiki-final/`,
|
||||
交付报告:`/home/zheromiku/.hermes/handoffs/release-wiki-final-acceptance-result.md`。
|
||||
正式 8088 仍是 0.1.114(dc9e827),未部署;部署方案见 `docs/formal-8088-deployment-plan.md`。
|
||||
|
||||
@@ -96,7 +96,10 @@ type Config struct {
|
||||
LibraryRetentionDays int
|
||||
// 目标维护(实例级):同步/恢复期间关闭目标入口、停下写入者、排空在途请求。
|
||||
// 任一项留空即"未接线",生产写入会被拒绝——只锁治理库 scope 不算维护。
|
||||
LibraryTargetInstance string
|
||||
LibraryTargetInstance string
|
||||
// LibrarySourceInstance 是死信解除记录绑定的"来源实例"名字(封存源,不是同步目标)。
|
||||
// 留空时回落到 LibraryTargetInstance,保持单实例部署的旧行为。
|
||||
LibrarySourceInstance string
|
||||
LibraryTargetProject string
|
||||
LibraryTargetWriters []string
|
||||
LibraryTargetNonWriters map[string]string
|
||||
@@ -199,6 +202,7 @@ func Load(getenv func(string) string) (Config, error) {
|
||||
LibraryProductionS3AccessKey: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_ACCESS_KEY")),
|
||||
LibraryProductionS3SecretKey: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_SECRET_KEY")),
|
||||
LibraryTargetInstance: strings.TrimSpace(getenv("LIBRARY_TARGET_INSTANCE")),
|
||||
LibrarySourceInstance: strings.TrimSpace(getenv("LIBRARY_SOURCE_INSTANCE")),
|
||||
LibraryTargetProject: strings.TrimSpace(getenv("LIBRARY_TARGET_PROJECT")),
|
||||
LibraryTargetWriters: splitList(getenv("LIBRARY_TARGET_WRITERS")),
|
||||
LibraryTargetNonWriters: splitDeclarations(getenv("LIBRARY_TARGET_NON_WRITERS")),
|
||||
@@ -562,3 +566,12 @@ func positiveInt64OrDefault(value string, fallback int64) (int64, error) {
|
||||
}
|
||||
return parsed, nil
|
||||
}
|
||||
|
||||
// SourceInstance 是死信解除记录绑定的实例名:封存源与同步目标常常不是同一个实例,
|
||||
// 用目标名去绑来源死信会让真实观测永远对不上(拒绝而不是误放行)。
|
||||
func (config Config) SourceInstance() string {
|
||||
if config.LibrarySourceInstance != "" {
|
||||
return config.LibrarySourceInstance
|
||||
}
|
||||
return config.LibraryTargetInstance
|
||||
}
|
||||
|
||||
@@ -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]
|
||||
}
|
||||
|
||||
@@ -97,7 +97,10 @@ func libraryCaptureEnqueue(w http.ResponseWriter, r *http.Request, config Config
|
||||
}
|
||||
}
|
||||
}
|
||||
requestedID := fmt.Sprintf("library-capture-%s-%d", request.Scope, now.UnixNano())
|
||||
// 队列作业 id 落进 jobs.id(varchar(36)):原来拼 "library-capture-<scope>-<nano>"
|
||||
// 是 45 个字符,真实请求 100% 被 PG 拒(value too long)——只是页面一直停在
|
||||
// "未接线"上,没人走到这里。这里按列宽拼短 id,并留出 scope 全称。
|
||||
requestedID := fmt.Sprintf("libcap-%s-%d", request.Scope, now.UnixNano())
|
||||
// The idempotency key is the capture job id: the same logical capture must map
|
||||
// to one queue row. Binding the scope into the key would silently create a
|
||||
// second job instead of reporting a conflicting request, so the conflict is
|
||||
|
||||
@@ -86,6 +86,27 @@ func TestLibraryCaptureRejectsInvalidRequest(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLibraryCaptureQueueIDFitsTheColumn(t *testing.T) {
|
||||
// jobs.id 是 varchar(36):拼长的 id 会在真实 PG 上 500(value too long)。
|
||||
// 页面长期停在"未接线",这条从没被真实请求覆盖过,所以用测试锁住列宽。
|
||||
for _, scope := range []string{"internal", "external"} {
|
||||
store := &fakeJobStore{enqueueEchoRequestedID: true}
|
||||
recorder := httptest.NewRecorder()
|
||||
libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"`+scope+`"}`),
|
||||
Config{Store: store, LibraryAgentSocket: "/tmp/agent.sock"})
|
||||
if recorder.Code != http.StatusAccepted {
|
||||
t.Fatalf("%s status=%d body=%s", scope, recorder.Code, recorder.Body.String())
|
||||
}
|
||||
enqueued := store.enqueued
|
||||
if len(enqueued.ID) > 36 {
|
||||
t.Fatalf("%s 队列作业 id 超长(%d):%q", scope, len(enqueued.ID), enqueued.ID)
|
||||
}
|
||||
if !strings.HasPrefix(enqueued.ID, "libcap-"+scope+"-") {
|
||||
t.Fatalf("%s 队列作业 id 形状变了:%q", scope, enqueued.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLibraryCaptureJobsAsksStoreForItsOwnType(t *testing.T) {
|
||||
// The global queue is busy with unrelated traffic; the capture job must still
|
||||
// be visible because the query is type-scoped, not filtered after the fact.
|
||||
|
||||
@@ -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.SourceInstance())
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -86,7 +87,10 @@ func libraryRestoreEnqueue(w http.ResponseWriter, r *http.Request, config Config
|
||||
return
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
jobID := "librestore-" + request.Scope + "-" + now.Format("20060102T150405.000000")
|
||||
// 作业 id 会作为维护窗口的 job 标识落进 freeze journal,而冻结协调器只接受
|
||||
// [A-Za-z0-9_-]:带 "." 的时间戳会让恢复在开窗口时被拒(真跑报 invalid freeze request),
|
||||
// 所以用纳秒整数,和同步作业保持一致。
|
||||
jobID := "librestore-" + request.Scope + "-" + strconv.FormatInt(now.UnixNano(), 10)
|
||||
job, reused, err := store.CreateLibraryJob(r.Context(), governance.LibraryJob{
|
||||
ID: jobID, Scope: request.Scope, Kind: governance.LibraryJobRestore,
|
||||
IdempotencyKey: "library.restore:" + request.Scope + ":" + request.SnapshotID,
|
||||
@@ -107,7 +111,7 @@ func libraryRestoreEnqueue(w http.ResponseWriter, r *http.Request, config Config
|
||||
if reused && (job.State == governance.LibraryJobFailed || job.State == governance.LibraryJobCancelled) {
|
||||
job, reused, err = store.CreateLibraryJob(r.Context(), governance.LibraryJob{
|
||||
ID: jobID + "-retry", Scope: request.Scope, Kind: governance.LibraryJobRestore,
|
||||
IdempotencyKey: "library.restore:" + request.Scope + ":" + request.SnapshotID + ":retry:" + now.Format("20060102T150405.000000000"),
|
||||
IdempotencyKey: "library.restore:" + request.Scope + ":" + request.SnapshotID + ":retry:" + strconv.FormatInt(now.UnixNano(), 10),
|
||||
RequestDigest: digest, TargetSnapshotID: request.SnapshotID, Actor: identity.Username,
|
||||
}, governance.SyncStages(), now)
|
||||
if err != nil || reused {
|
||||
|
||||
@@ -85,3 +85,31 @@ func TestLibraryJobResumeRejectsNonFailedJobs(t *testing.T) {
|
||||
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// 恢复作业 id 会作为维护窗口的 job 标识落进 freeze journal,协调器只接受
|
||||
// [A-Za-z0-9_-]。带 "." 的时间戳曾经让每次恢复都在开窗口时被拒(invalid freeze request)。
|
||||
func TestLibraryRestoreJobIDIsAFreezeSafeIdentifier(t *testing.T) {
|
||||
store := &fakeLibraryStore{fakeJobStore: fakeJobStore{enqueueEchoRequestedID: true}}
|
||||
digest, err := libraryRestoreParamsDigest("internal", "snap-1")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
store.grant = &governance.LibraryReleaseAuthorization{ID: "a", Scope: "internal", SnapshotID: "snap-1",
|
||||
ParamsDigest: digest, ExpiresAt: now.Add(time.Hour), CreatedAt: now}
|
||||
config := Config{Store: store, ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true}
|
||||
recorder := httptest.NewRecorder()
|
||||
libraryRestoreEnqueue(recorder, restoreRequest(`{"scope":"internal","snapshot_id":"snap-1"}`, roleAdmin), config)
|
||||
if recorder.Code != http.StatusAccepted {
|
||||
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
|
||||
}
|
||||
id := store.lastJob.ID
|
||||
if id == "" || len(id) > 128 {
|
||||
t.Fatalf("作业 id 形状不对:%q", id)
|
||||
}
|
||||
for _, r := range id {
|
||||
if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_') {
|
||||
t.Fatalf("作业 id 含维护窗口不接受的字符 %q:%q", r, id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -148,6 +148,8 @@ type Config struct {
|
||||
// LibraryTargetInstance 是实例级维护的目标实例名;留空即"未接线",
|
||||
// 页面不能发起按预备份恢复,接口返回原因而不是假装能恢复。
|
||||
LibraryTargetInstance string
|
||||
// LibrarySourceInstance 是死信解除记录的来源实例名(封存源),留空回落到目标实例名。
|
||||
LibrarySourceInstance string
|
||||
// LibraryAgentSocket 是主机侧封存 agent 的 unix socket;留空即页面不能发起封存。
|
||||
LibraryAgentSocket string
|
||||
// LibraryCapacityBudgetBytes 是整库归档容量预算;0 = 未设置(容量门禁不放行)。
|
||||
@@ -432,6 +434,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) {
|
||||
@@ -2800,3 +2809,12 @@ func writeJSON(w http.ResponseWriter, status int, value any) {
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(value)
|
||||
}
|
||||
|
||||
// SourceInstance 是死信解除记录绑定的实例名:来源实例与同步目标常常不是同一个,
|
||||
// 用目标名去绑来源死信会让真实观测永远对不上。
|
||||
func (config Config) SourceInstance() string {
|
||||
if config.LibrarySourceInstance != "" {
|
||||
return config.LibrarySourceInstance
|
||||
}
|
||||
return config.LibraryTargetInstance
|
||||
}
|
||||
|
||||
@@ -37,14 +37,17 @@ type Config struct {
|
||||
JournalRoot string
|
||||
// Writer declaration passed to the binary, exactly as reviewed in
|
||||
// scripts/library-capture.sh. Clients cannot influence it.
|
||||
SourceApp string
|
||||
SourcePG string
|
||||
SourceES string
|
||||
GovernancePG string
|
||||
Project string
|
||||
Writers string
|
||||
NonWriters string
|
||||
MaxDuration time.Duration
|
||||
SourceApp string
|
||||
SourcePG string
|
||||
SourceES string
|
||||
// SourceS3Endpoint 让宿主机用可达地址读源对象存储:容器 env 里的端点常常
|
||||
// 只是服务名(minio:9000),宿主机解析不到,资源导出会失败。
|
||||
SourceS3Endpoint string
|
||||
GovernancePG string
|
||||
Project string
|
||||
Writers string
|
||||
NonWriters string
|
||||
MaxDuration time.Duration
|
||||
// StopGrace bounds the graceful stop of one operation: SIGTERM first (the
|
||||
// child aborts the maintenance window and restores writers), then SIGKILL
|
||||
// only after this grace period. It must exceed the child's own restore budget.
|
||||
@@ -60,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 {
|
||||
@@ -73,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 {
|
||||
@@ -211,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
|
||||
@@ -268,17 +292,22 @@ func (a *Agent) recoverWriters(ctx context.Context) error {
|
||||
// declaration pins every path and the writer group so a request cannot redirect
|
||||
// the operation to another stack, archive or journal.
|
||||
func (c Config) declaration() []string {
|
||||
return []string{
|
||||
args := []string{
|
||||
"-source-app", c.SourceApp,
|
||||
"-source-pg", c.SourcePG,
|
||||
"-source-es", c.SourceES,
|
||||
}
|
||||
if c.SourceS3Endpoint != "" {
|
||||
args = append(args, "-source-s3-endpoint", c.SourceS3Endpoint)
|
||||
}
|
||||
return append(args,
|
||||
"-governance-pg", c.GovernancePG,
|
||||
"-project", c.Project,
|
||||
"-writers", c.Writers,
|
||||
"-non-writers", c.NonWriters,
|
||||
"-archive", c.ArchiveRoot,
|
||||
"-journal", c.JournalRoot,
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
func (c Config) run(ctx context.Context, name string, args ...string) ([]byte, error) {
|
||||
|
||||
@@ -336,3 +336,76 @@ func TestNewRejectsIncompleteConfiguration(t *testing.T) {
|
||||
t.Fatal("non-writer declaration without a reason accepted")
|
||||
}
|
||||
}
|
||||
|
||||
// 宿主机可达的源对象存储端点必须由 agent 自己钉住:请求不能改,也不能被悄悄丢掉
|
||||
// (掉了就会退回"容器里那个解析不到的服务名",真实封存报 read library resource)。
|
||||
func TestAgentPinsTheHostSideObjectEndpoint(t *testing.T) {
|
||||
config := Config{
|
||||
Socket: "/tmp/agent.sock",
|
||||
Binary: "/bin/echo",
|
||||
ArchiveRoot: "/tmp/library",
|
||||
JournalRoot: "/tmp/freeze",
|
||||
SourceApp: "app-container",
|
||||
SourcePG: "pg",
|
||||
SourceES: "es",
|
||||
SourceS3Endpoint: "http://127.0.0.1:19100",
|
||||
GovernancePG: "gov",
|
||||
Project: "proj",
|
||||
Writers: "app",
|
||||
NonWriters: "postgres=database store",
|
||||
MaxDuration: time.Minute,
|
||||
}
|
||||
agent, err := New(config)
|
||||
if err != nil {
|
||||
t.Fatalf("new: %v", err)
|
||||
}
|
||||
args := agent.config.declaration()
|
||||
found := false
|
||||
for i := 0; i+1 < len(args); i++ {
|
||||
if args[i] == "-source-s3-endpoint" && args[i+1] == "http://127.0.0.1:19100" {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("声明里没有宿主侧对象存储端点:%v", args)
|
||||
}
|
||||
config.SourceS3Endpoint = ""
|
||||
bare, err := New(config)
|
||||
if err != nil {
|
||||
t.Fatalf("new without endpoint: %v", err)
|
||||
}
|
||||
for _, arg := range bare.config.declaration() {
|
||||
if arg == "-source-s3-endpoint" {
|
||||
t.Fatal("没有端点时不该凭空多一个 flag")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 死信解除只能按精确指纹透传:形状不对就拒绝,且固定命令其余部分不受影响。
|
||||
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,93 @@
|
||||
package librarymaintenance
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// 目标入口闸门与后台队列的声明名称。
|
||||
//
|
||||
// 为什么集中在这里:容器内的 worker 与主机侧的封存(cmd/libraryctl)都要讲
|
||||
// 同一份"这个实例的入口在哪、队列在哪";名字写两遍迟早会分叉,而分叉的表现
|
||||
// 是"一边关的是 A 实例的门、另一边读的是 B 实例的队列"——不会报错,只会给出
|
||||
// 错的证据。
|
||||
const (
|
||||
EnvInstance = "LIBRARY_TARGET_INSTANCE"
|
||||
EnvIngressURL = "LIBRARY_TARGET_INGRESS_URL"
|
||||
EnvIngressService = "LIBRARY_TARGET_INGRESS_SERVICE"
|
||||
EnvIngressDrainFile = "LIBRARY_TARGET_INGRESS_DRAIN_FILE"
|
||||
EnvRedisAddr = "LIBRARY_TARGET_REDIS_ADDR"
|
||||
// 名字里不写 Password:这里的字面量是**环境变量名**,密钥扫描会把
|
||||
// `XXX_PASSWORD = "<长字面量>"` 当成"被赋值的密钥",那是误报但没必要留个坑。
|
||||
EnvRedisAuth = "LIBRARY_TARGET_REDIS_PASSWORD"
|
||||
EnvRedisDB = "LIBRARY_TARGET_REDIS_DB"
|
||||
)
|
||||
|
||||
// TargetEnv 是一次实例级维护所需的"入口 + 队列"声明。
|
||||
//
|
||||
// 缺任何一项都不许降级成"库级维护":没有闸门就没有"入口已关、在途已排空"
|
||||
// 的证据,没有队列就证明不了停写者时后台是空的。这两条正是被代理的 WeKnora
|
||||
// 收到 SIGTERM 就 exit(1) 之后唯一还能证明"请求没被截断"的东西。
|
||||
type TargetEnv struct {
|
||||
Instance string
|
||||
IngressURL string
|
||||
IngressService string
|
||||
IngressDrainFile string
|
||||
RedisAddr string
|
||||
RedisPassword string
|
||||
RedisDB int
|
||||
}
|
||||
|
||||
// TargetFromEnv 解析维护目标声明。缺少的项一次性全部报出来(含变量名),
|
||||
// 让部署方一次改对,而不是改一项再撞下一项。
|
||||
func TargetFromEnv(get func(string) string) (TargetEnv, error) {
|
||||
if get == nil {
|
||||
return TargetEnv{}, errors.New("缺少配置读取入口")
|
||||
}
|
||||
value := func(name string) string { return strings.TrimSpace(get(name)) }
|
||||
target := TargetEnv{
|
||||
Instance: value(EnvInstance),
|
||||
IngressURL: value(EnvIngressURL),
|
||||
IngressService: value(EnvIngressService),
|
||||
IngressDrainFile: value(EnvIngressDrainFile),
|
||||
RedisAddr: value(EnvRedisAddr),
|
||||
// 口令不 Trim:前后空格也是口令的一部分,Trim 会把能连上的配置变成连不上。
|
||||
RedisPassword: get(EnvRedisAuth),
|
||||
}
|
||||
missing := make([]string, 0, 5)
|
||||
for _, item := range []struct {
|
||||
name string
|
||||
current string
|
||||
}{
|
||||
{EnvInstance, target.Instance},
|
||||
{EnvIngressURL, target.IngressURL},
|
||||
{EnvIngressService, target.IngressService},
|
||||
{EnvIngressDrainFile, target.IngressDrainFile},
|
||||
{EnvRedisAddr, target.RedisAddr},
|
||||
} {
|
||||
if item.current == "" {
|
||||
missing = append(missing, item.name)
|
||||
}
|
||||
}
|
||||
if strings.TrimSpace(target.RedisPassword) == "" {
|
||||
missing = append(missing, EnvRedisAuth)
|
||||
}
|
||||
rawDB := value(EnvRedisDB)
|
||||
switch {
|
||||
case rawDB == "":
|
||||
missing = append(missing, EnvRedisDB)
|
||||
default:
|
||||
db, err := strconv.Atoi(rawDB)
|
||||
if err != nil || db < 0 {
|
||||
return TargetEnv{}, fmt.Errorf("%s 不是有效的队列库号(%q)", EnvRedisDB, rawDB)
|
||||
}
|
||||
target.RedisDB = db
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
return TargetEnv{}, fmt.Errorf("实例级维护未接线:缺少 %s(没有入口闸门与队列就无法证明入口已排空、后台已静默)",
|
||||
strings.Join(missing, "、"))
|
||||
}
|
||||
return target, nil
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
package librarymaintenance
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func completeEnv() map[string]string {
|
||||
return map[string]string{
|
||||
EnvInstance: "kgmaint-weknora",
|
||||
EnvIngressURL: "http://127.0.0.1:18080/health",
|
||||
EnvIngressService: "ingress",
|
||||
EnvIngressDrainFile: "/tmp/gate/drain.json",
|
||||
EnvRedisAddr: "127.0.0.1:26379",
|
||||
EnvRedisAuth: "secret",
|
||||
EnvRedisDB: "0",
|
||||
}
|
||||
}
|
||||
|
||||
func lookup(values map[string]string) func(string) string {
|
||||
return func(name string) string { return values[name] }
|
||||
}
|
||||
|
||||
func TestTargetFromEnvReadsTheWholeDeclaration(t *testing.T) {
|
||||
target, err := TargetFromEnv(lookup(completeEnv()))
|
||||
if err != nil {
|
||||
t.Fatalf("完整声明被拒:%v", err)
|
||||
}
|
||||
if target.Instance != "kgmaint-weknora" || target.IngressService != "ingress" ||
|
||||
target.IngressURL != "http://127.0.0.1:18080/health" || target.IngressDrainFile != "/tmp/gate/drain.json" ||
|
||||
target.RedisAddr != "127.0.0.1:26379" || target.RedisPassword != "secret" || target.RedisDB != 0 {
|
||||
t.Fatalf("解析结果不对:%+v", target)
|
||||
}
|
||||
}
|
||||
|
||||
// 每个缺失项都必须点名报出来:部署方要一次改对,而不是"改一项再撞下一项"。
|
||||
func TestTargetFromEnvNamesEveryMissingItem(t *testing.T) {
|
||||
for _, name := range []string{EnvInstance, EnvIngressURL, EnvIngressService,
|
||||
EnvIngressDrainFile, EnvRedisAddr, EnvRedisAuth, EnvRedisDB} {
|
||||
values := completeEnv()
|
||||
delete(values, name)
|
||||
if _, err := TargetFromEnv(lookup(values)); err == nil || !strings.Contains(err.Error(), name) {
|
||||
t.Fatalf("缺 %s 没有被点名:%v", name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 空白不能当成"配了":空口令会让 queuers 连不上,却在日志里显示"已接线"。
|
||||
func TestTargetFromEnvRejectsBlankValues(t *testing.T) {
|
||||
values := completeEnv()
|
||||
values[EnvIngressURL] = " "
|
||||
if _, err := TargetFromEnv(lookup(values)); err == nil || !strings.Contains(err.Error(), EnvIngressURL) {
|
||||
t.Fatalf("空白入口地址没有被拒:%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 口令前后空格是口令的一部分:Trim 会把能连上的配置变成连不上。
|
||||
func TestTargetFromEnvKeepsPasswordWhitespace(t *testing.T) {
|
||||
values := completeEnv()
|
||||
values[EnvRedisAuth] = " pass with space "
|
||||
target, err := TargetFromEnv(lookup(values))
|
||||
if err != nil {
|
||||
t.Fatalf("带空格的口令被拒:%v", err)
|
||||
}
|
||||
if target.RedisPassword != " pass with space " {
|
||||
t.Fatalf("口令被改写了:%q", target.RedisPassword)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTargetFromEnvRejectsInvalidRedisDB(t *testing.T) {
|
||||
values := completeEnv()
|
||||
values[EnvRedisDB] = "-1"
|
||||
if _, err := TargetFromEnv(lookup(values)); err == nil || !strings.Contains(err.Error(), EnvRedisDB) {
|
||||
t.Fatalf("负数库号没有被拒:%v", err)
|
||||
}
|
||||
values[EnvRedisDB] = "zero"
|
||||
if _, err := TargetFromEnv(lookup(values)); err == nil || !strings.Contains(err.Error(), EnvRedisDB) {
|
||||
t.Fatalf("非数字库号没有被拒:%v", err)
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -47,9 +47,15 @@ type ProductionLibraryConfig struct {
|
||||
Objects ObjectStore
|
||||
ObjectsReader ObjectReader
|
||||
ArchiveRoot string
|
||||
MaxDocuments int
|
||||
HTTPClient *http.Client
|
||||
Now func() time.Time
|
||||
// ObjectsBucket 是目标对象存储的桶名:资源行里的定位符要换成它,
|
||||
// 否则同步过去的图/附件仍指向来源桶。
|
||||
ObjectsBucket string
|
||||
// ObjectsBackendID 是目标实例 storage_backends 的主键。留空则按租户+桶名在
|
||||
// 目标库里解析;定位符里的来源后端 id 在目标实例不存在,原版应用会 500。
|
||||
ObjectsBackendID string
|
||||
MaxDocuments int
|
||||
HTTPClient *http.Client
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
func (config ProductionLibraryConfig) validate(scope ProductionScope) error {
|
||||
@@ -78,6 +84,36 @@ type ProductionLibraryTarget struct {
|
||||
scope ProductionScope
|
||||
search *SearchExporter
|
||||
now func() time.Time
|
||||
// storageBackend 是本次写入使用的目标存储后端 id(写入与校验共用)。
|
||||
storageBackend string
|
||||
}
|
||||
|
||||
// resolveStorageBackend 在目标库里找本次写入的存储后端。找不到就报错:宁可不写,
|
||||
// 也不写一条目标应用读不出来的定位符(上轮就是这样同步"成功"了却下载 500)。
|
||||
func (t *ProductionLibraryTarget) resolveStorageBackend(ctx context.Context) error {
|
||||
if t.config.ObjectsBucket == "" {
|
||||
return nil
|
||||
}
|
||||
if t.config.ObjectsBackendID != "" {
|
||||
if !validSearchIdentifier(t.config.ObjectsBackendID) {
|
||||
return errors.New("目标存储后端 id 形状不合法")
|
||||
}
|
||||
t.storageBackend = t.config.ObjectsBackendID
|
||||
return nil
|
||||
}
|
||||
var id string
|
||||
err := t.pool.QueryRow(ctx, `
|
||||
SELECT id FROM storage_backends
|
||||
WHERE tenant_id=$1 AND coalesce(config->>'bucket_name','')=$2
|
||||
ORDER BY created_at LIMIT 1`, t.scope.TenantID, t.config.ObjectsBucket).Scan(&id)
|
||||
if err != nil {
|
||||
return errors.New("目标实例没有与目标桶匹配的存储后端,拒绝写入不可解析的资源定位符")
|
||||
}
|
||||
if !validSearchIdentifier(id) {
|
||||
return errors.New("目标存储后端 id 形状不合法")
|
||||
}
|
||||
t.storageBackend = id
|
||||
return nil
|
||||
}
|
||||
|
||||
func NewProductionLibraryTarget(ctx context.Context, config ProductionLibraryConfig, scope ProductionScope) (*ProductionLibraryTarget, error) {
|
||||
@@ -220,6 +256,9 @@ func digestIDs(ids []string) string {
|
||||
|
||||
// Apply 把一类数据从归档写进生产;database 与 wiki 走同一个知识域事务。
|
||||
func (t *ProductionLibraryTarget) Apply(ctx context.Context, stage libraryrelease.Stage, snapshotID string) error {
|
||||
if err := t.resolveStorageBackend(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
|
||||
if err != nil {
|
||||
return errors.New("归档不可用")
|
||||
@@ -317,6 +356,11 @@ func (t *ProductionLibraryTarget) insertRows(ctx context.Context, tx pgx.Tx, arc
|
||||
if len(row) == 0 {
|
||||
continue
|
||||
}
|
||||
if table == "resources" {
|
||||
if err := t.applyResourceBucket(row); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if columns == nil {
|
||||
columns = make([]string, 0, len(row))
|
||||
for key := range row {
|
||||
@@ -363,6 +407,56 @@ type resourceLocation struct {
|
||||
Provider string
|
||||
}
|
||||
|
||||
// rowTransform 是"写入时对行做过的变换"的统一入口:校验要用同一份,否则摘要比的是
|
||||
// 变换前的行、目标里是变换后的行,永远不一致。
|
||||
func (t *ProductionLibraryTarget) rowTransform(table string) func(map[string]json.RawMessage) error {
|
||||
if table == "resources" {
|
||||
return t.applyResourceBucket
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// applyResourceBucket 把资源行定位符里的桶换成本次写入的目标桶。
|
||||
// 写入与校验必须用同一个变换:只改写入侧会让"行内容摘要"对不上(真跑踩过)。
|
||||
func (t *ProductionLibraryTarget) applyResourceBucket(row map[string]json.RawMessage) error {
|
||||
if t.config.ObjectsBucket == "" {
|
||||
return nil
|
||||
}
|
||||
raw, ok := row["physical_path"]
|
||||
if !ok {
|
||||
return errors.New("资源行缺少定位符")
|
||||
}
|
||||
var locator string
|
||||
if err := json.Unmarshal(raw, &locator); err != nil {
|
||||
return errors.New("资源定位符不可解析")
|
||||
}
|
||||
rewritten, ok := rewriteLocatorTarget(locator, t.storageBackend, t.config.ObjectsBucket)
|
||||
if !ok {
|
||||
return fmt.Errorf("资源定位符无法改写到目标桶:%s", locator)
|
||||
}
|
||||
encoded, err := json.Marshal(rewritten)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
row["physical_path"] = encoded
|
||||
// 原版应用先按 resources.storage_backend_id 找存储后端,再回落到 KB,再落默认服务。
|
||||
// 这一列留着来源 id,目标仍然 500(真跑:storage backend not found → invalid path)。
|
||||
if t.storageBackend == "" {
|
||||
return nil
|
||||
}
|
||||
if raw, ok := row["storage_backend_id"]; ok {
|
||||
var current string
|
||||
if json.Unmarshal(raw, ¤t) == nil && current != "" && current != t.storageBackend {
|
||||
encodedBackend, err := json.Marshal(t.storageBackend)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
row["storage_backend_id"] = encodedBackend
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
|
||||
paths, err := t.resourcePaths(ctx, resourcePartNames(manifest))
|
||||
if err != nil {
|
||||
@@ -494,6 +588,9 @@ func (t *ProductionLibraryTarget) importIndex(ctx context.Context, archive *libr
|
||||
|
||||
// Verify 用行数 + 身份集合摘要 + 索引条数核对生产与预发布快照是否一致。
|
||||
func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrelease.Expectation) error {
|
||||
if err := t.resolveStorageBackend(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
|
||||
if err != nil {
|
||||
return errors.New("归档不可用")
|
||||
@@ -517,7 +614,7 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
|
||||
wantCount, wantDigest, err := archivedTableState(archive, part)
|
||||
if errors.Is(err, errMissingIdentity) {
|
||||
// 个别表(例如日志表)没有代理主键:身份集合无从比,但行数与逐行内容摘要仍然必须一致。
|
||||
wantContentCount, wantContentDigest, columns, contentErr := archivedTableContent(archive, part)
|
||||
wantContentCount, wantContentDigest, columns, contentErr := archivedTableContent(archive, part, t.rowTransform(part.Name))
|
||||
if contentErr != nil {
|
||||
return contentErr
|
||||
}
|
||||
@@ -548,7 +645,7 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
|
||||
return fmt.Errorf("生产 %s 与快照不一致:行数 %d/%d", part.Name, count, wantCount)
|
||||
}
|
||||
// 行数/身份一致还要比内容:同 ID 改正文、改 Wiki、改向量必须被发现。
|
||||
wantContentCount, wantContentDigest, columns, err := archivedTableContent(archive, part)
|
||||
wantContentCount, wantContentDigest, columns, err := archivedTableContent(archive, part, t.rowTransform(part.Name))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -26,6 +26,22 @@ func (e *SearchExporter) ExportLibrary(ctx context.Context, identity SearchIdent
|
||||
}
|
||||
return SearchSnapshot{Documents: []SearchDocument{}, MappingSHA256: "", DocumentsSHA256: emptyDigest}, nil
|
||||
}
|
||||
if errors.Is(err, errScopeFieldsMissing) {
|
||||
// 全新实例先用空映射建索引:没有范围字段、又确实 0 篇文档时就是空基线;
|
||||
// 有文档却看不出归属必须报错(不能当成空库覆盖)。
|
||||
empty, probeErr := e.emptyScopeIndex(ctx)
|
||||
if probeErr != nil {
|
||||
return SearchSnapshot{}, probeErr
|
||||
}
|
||||
if empty {
|
||||
emptyDigest, digestErr := searchDocumentsDigest([]SearchDocument{})
|
||||
if digestErr != nil {
|
||||
return SearchSnapshot{}, digestErr
|
||||
}
|
||||
return SearchSnapshot{Documents: []SearchDocument{}, MappingSHA256: "", DocumentsSHA256: emptyDigest}, nil
|
||||
}
|
||||
return SearchSnapshot{}, err
|
||||
}
|
||||
if err != nil {
|
||||
return SearchSnapshot{}, err
|
||||
}
|
||||
@@ -52,8 +68,8 @@ func (e *SearchExporter) ExportLibrary(ctx context.Context, identity SearchIdent
|
||||
}
|
||||
}()
|
||||
filter := []any{map[string]any{"term": map[string]any{inspection.fields.base: identity.KnowledgeBaseID}}}
|
||||
if inspection.fields.tenant != "" {
|
||||
filter = append(filter, map[string]any{"term": map[string]any{inspection.fields.tenant: identity.TenantID}})
|
||||
if clause := tenantClause(inspection.fields, identity.TenantID); clause != nil {
|
||||
filter = append(filter, clause)
|
||||
}
|
||||
var cursor json.RawMessage
|
||||
documents := make([]SearchDocument, 0)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -98,7 +98,8 @@ func (t *ProductionLibraryTarget) productionTableContent(ctx context.Context, ta
|
||||
}
|
||||
|
||||
// archivedTableContent 读取归档侧规范内容摘要,并返回归档里出现过的列集合。
|
||||
func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot.Part) (int64, string, map[string]bool, error) {
|
||||
func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot.Part,
|
||||
transform func(map[string]json.RawMessage) error) (int64, string, map[string]bool, error) {
|
||||
reader, err := archive.OpenPart(part)
|
||||
if err != nil {
|
||||
return 0, "", nil, err
|
||||
@@ -116,6 +117,12 @@ func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot
|
||||
if err := decoder.Decode(&raw); err != nil {
|
||||
return 0, "", nil, err
|
||||
}
|
||||
if transform != nil {
|
||||
// 写入时对行做过的变换(例如资源定位符换桶)校验时必须同样应用。
|
||||
if err := transform(raw); err != nil {
|
||||
return 0, "", nil, err
|
||||
}
|
||||
}
|
||||
for key := range raw {
|
||||
columns[key] = true
|
||||
}
|
||||
|
||||
@@ -100,6 +100,66 @@ func (importer *ObjectImporter) Promote(ctx context.Context, stage ObjectStage)
|
||||
return nil
|
||||
}
|
||||
|
||||
// rewriteLocatorBucket 把资源定位符里的桶换成目标桶,结构保持不变:
|
||||
// "storage://<backend>/<provider>://<source-bucket>/<key>" →
|
||||
// "storage://<backend>/<provider>://<target-bucket>/<key>"。
|
||||
//
|
||||
// 为什么必须改:同步把字节写进目标桶,但行是从快照原样搬的,定位符里写着测试桶;
|
||||
// 目标实例之后按这一行去取图/附件时会指向测试环境的桶(跨环境引用),
|
||||
// 或者干脆取不到。键不变,只换桶。
|
||||
func rewriteLocatorBucket(locator, bucket string) (string, bool) {
|
||||
return rewriteLocatorTarget(locator, "", bucket)
|
||||
}
|
||||
|
||||
// rewriteLocatorTarget 同时改写定位符里的存储后端 id 与桶。
|
||||
//
|
||||
// 为什么后端 id 也要改:storage://<id>/minio://<bucket>/<key> 里的 id 是来源实例
|
||||
// storage_backends 的主键。目标实例有自己的行;只改桶不改 id,原版应用会以
|
||||
// "storage backend not found" 落到默认服务,然后拒绝整条路径(真跑 500 code1007)。
|
||||
func rewriteLocatorTarget(locator, backendID, bucket string) (string, bool) {
|
||||
if bucket == "" || locator == "" || locator != strings.TrimSpace(locator) {
|
||||
return "", false
|
||||
}
|
||||
prefix := ""
|
||||
rest := locator
|
||||
if strings.HasPrefix(locator, "storage://") {
|
||||
trimmed := strings.TrimPrefix(locator, "storage://")
|
||||
parts := strings.SplitN(trimmed, "/", 2)
|
||||
if len(parts) != 2 || !validSearchIdentifier(parts[0]) {
|
||||
return "", false
|
||||
}
|
||||
if parts[1] == "" {
|
||||
return "", false
|
||||
}
|
||||
if backendID != "" {
|
||||
if !validSearchIdentifier(backendID) {
|
||||
return "", false
|
||||
}
|
||||
parts[0] = backendID
|
||||
}
|
||||
prefix, rest = "storage://"+parts[0]+"/", parts[1]
|
||||
}
|
||||
separator := strings.Index(rest, "://")
|
||||
if separator < 0 {
|
||||
// 裸对象键里没有桶:落盘时用的是配置桶,不需要改,也不算失败。
|
||||
return locator, true
|
||||
}
|
||||
scheme := rest[:separator]
|
||||
if !validSearchIdentifier(scheme) {
|
||||
return "", false
|
||||
}
|
||||
schemeless := rest[separator+3:]
|
||||
boundary := strings.Index(schemeless, "/")
|
||||
if boundary <= 0 || boundary == len(schemeless)-1 {
|
||||
return "", false
|
||||
}
|
||||
key := schemeless[boundary+1:]
|
||||
if key == "" || !validSearchIdentifier(schemeless[:boundary]) {
|
||||
return "", false
|
||||
}
|
||||
return prefix + scheme + "://" + bucket + "/" + key, true
|
||||
}
|
||||
|
||||
func providerPathFromLocator(locator string) (string, bool) {
|
||||
rest := locator
|
||||
if strings.HasPrefix(locator, "storage://") {
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 同步把字节写进目标桶,但资源行是从快照原样搬的:定位符里写着来源桶时,
|
||||
// 目标实例之后取图/附件会指向测试环境的桶(跨环境引用)或直接取不到。
|
||||
// 只换桶、不换键,两种定位符形状都要能改。
|
||||
func TestRewriteLocatorBucketKeepsKeyAndSwapsBucket(t *testing.T) {
|
||||
for _, testCase := range []struct{ locator, want string }{
|
||||
{"storage://45064e4f-434c-4752-a3ae-5fb1e2f54a7d/minio://kgmaint-1028a/10001/k/ab.md",
|
||||
"storage://45064e4f-434c-4752-a3ae-5fb1e2f54a7d/minio://kgseal-1029a/10001/k/ab.md"},
|
||||
{"oss://source-bucket/10/k/ab.png", "oss://target-bucket/10/k/ab.png"},
|
||||
} {
|
||||
bucket := "kgseal-1029a"
|
||||
if testCase.locator[:3] == "oss" {
|
||||
bucket = "target-bucket"
|
||||
}
|
||||
got, ok := rewriteLocatorBucket(testCase.locator, bucket)
|
||||
if !ok || got != testCase.want {
|
||||
t.Fatalf("rewrite(%q)=%q ok=%v want %q", testCase.locator, got, ok, testCase.want)
|
||||
}
|
||||
}
|
||||
// 解析不出来就报错,不猜:宁可失败也不能写一行指向别处的资源。
|
||||
for _, bad := range []string{"", "storage://x/", "storage://backend/minio://bucket", "oss://bucket/"} {
|
||||
if _, ok := rewriteLocatorBucket(bad, "target"); ok {
|
||||
t.Fatalf("非法定位符被接受了:%q", bad)
|
||||
}
|
||||
}
|
||||
if _, ok := rewriteLocatorBucket("oss://bucket/key", ""); ok {
|
||||
t.Fatal("没有目标桶时不该改写")
|
||||
}
|
||||
// 裸对象键里没有桶(落盘用配置桶),原样返回而不是失败。
|
||||
if got, ok := rewriteLocatorBucket("weknora/10000/k/ab.docx", "target"); !ok || got != "weknora/10000/k/ab.docx" {
|
||||
t.Fatalf("裸键被改写了:%q ok=%v", got, ok)
|
||||
}
|
||||
}
|
||||
|
||||
// 只换桶不够:storage://<id> 是来源实例 storage_backends 的主键。目标实例没有这条行时,
|
||||
// 原版应用会以 storage backend not found 落到默认服务并拒绝整条路径(真跑 500 code1007),
|
||||
// 所以后端 id 必须一起换成本次写入解析到的目标 id。
|
||||
func TestRewriteLocatorSwapsStorageBackendWithBucket(t *testing.T) {
|
||||
got, ok := rewriteLocatorTarget("storage://source-backend/minio://kgmaint-1028a/10001/k/ab.md",
|
||||
"target-backend", "kgseal-1029a")
|
||||
if !ok || got != "storage://target-backend/minio://kgseal-1029a/10001/k/ab.md" {
|
||||
t.Fatalf("rewrite=%q ok=%v", got, ok)
|
||||
}
|
||||
// 没有解析到后端时不猜:保留原 id(调用方在没有目标桶时不会走到这里)。
|
||||
got, ok = rewriteLocatorTarget("storage://source-backend/minio://b/k", "", "b2")
|
||||
if !ok || got != "storage://source-backend/minio://b2/k" {
|
||||
t.Fatalf("rewrite without backend=%q ok=%v", got, ok)
|
||||
}
|
||||
if _, ok := rewriteLocatorTarget("storage://x/minio://b/k", "../evil", "b2"); ok {
|
||||
t.Fatal("非法后端 id 被接受了")
|
||||
}
|
||||
}
|
||||
|
||||
// 行变换用解析到的目标后端与目标桶:写入与校验共用这一份,摘要才不会两边打架。
|
||||
func TestResourceRowTransformRewritesBackendAndBucket(t *testing.T) {
|
||||
target := &ProductionLibraryTarget{config: ProductionLibraryConfig{ObjectsBucket: "kgseal-1029a"},
|
||||
storageBackend: "target-backend"}
|
||||
row := map[string]json.RawMessage{"physical_path": json.RawMessage(`"storage://source-backend/minio://kgmaint-1028a/10001/k/ab.md"`)}
|
||||
if err := target.rowTransform("resources")(row); err != nil {
|
||||
t.Fatalf("row transform failed: %v", err)
|
||||
}
|
||||
var got string
|
||||
if err := json.Unmarshal(row["physical_path"], &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got != "storage://target-backend/minio://kgseal-1029a/10001/k/ab.md" {
|
||||
t.Fatalf("transform=%q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 存储后端 id 在 resources 表里还有独立一列:原版应用按它找文件服务。
|
||||
// 只改定位符不改这一列,目标仍是 storage backend not found(真跑 500)。
|
||||
func TestResourceRowTransformRewritesBackendColumn(t *testing.T) {
|
||||
target := &ProductionLibraryTarget{config: ProductionLibraryConfig{ObjectsBucket: "kgseal-1029a"},
|
||||
storageBackend: "target-backend"}
|
||||
row := map[string]json.RawMessage{
|
||||
"physical_path": json.RawMessage(`"storage://source-backend/minio://kgmaint-1028a/1/k/ab.md"`),
|
||||
"storage_backend_id": json.RawMessage(`"source-backend"`),
|
||||
}
|
||||
if err := target.rowTransform("resources")(row); err != nil {
|
||||
t.Fatalf("row transform failed: %v", err)
|
||||
}
|
||||
var backend string
|
||||
if err := json.Unmarshal(row["storage_backend_id"], &backend); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if backend != "target-backend" {
|
||||
t.Fatalf("storage_backend_id=%q", backend)
|
||||
}
|
||||
}
|
||||
@@ -67,6 +67,23 @@ type searchFieldNames struct {
|
||||
// errIndexMissing 表示目标索引根本不存在(空生产基线)。
|
||||
var errIndexMissing = errors.New("Elasticsearch index missing")
|
||||
|
||||
// errScopeFieldsMissing 表示索引存在但还没有发布范围字段:
|
||||
// 全新的 WeKnora 实例会先用空映射建好索引(`{}`),字段要等第一次写入才动态生成。
|
||||
// 这跟"索引被改坏了"必须分开:只有在索引里确实一篇文档都没有时才能当成空基线。
|
||||
var errScopeFieldsMissing = errors.New("Elasticsearch index mapping is missing release scope fields")
|
||||
|
||||
// emptyScopeIndex 判断"没有范围字段"的索引是不是真的空。
|
||||
// 有文档却没有范围字段 = 无法判断这批文档属于谁,必须报错,不能当成 0。
|
||||
func (exporter *SearchExporter) emptyScopeIndex(ctx context.Context) (bool, error) {
|
||||
var response struct {
|
||||
Count int64 `json:"count"`
|
||||
}
|
||||
if err := exporter.requestJSON(ctx, http.MethodGet, "/"+exporter.index+"/_count", nil, &response); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return response.Count == 0, nil
|
||||
}
|
||||
|
||||
type mappingInspection struct {
|
||||
digest string
|
||||
fields searchFieldNames
|
||||
@@ -127,17 +144,41 @@ func (exporter *SearchExporter) inspectMapping(ctx context.Context) (mappingInsp
|
||||
base: resolve("knowledge_base_id"),
|
||||
}
|
||||
if fields.knowledge == "" || fields.base == "" {
|
||||
return mappingInspection{}, errors.New("Elasticsearch index mapping is missing release scope fields")
|
||||
return mappingInspection{}, errScopeFieldsMissing
|
||||
}
|
||||
return mappingInspection{digest: hex.EncodeToString(digest[:]), fields: fields}, nil
|
||||
}
|
||||
|
||||
// scopedFilter builds the tenant/KB/knowledge term filter adapted to the
|
||||
// resolved index field names.
|
||||
// tenantClause 是"租户范围"这一条过滤条件的唯一写法。
|
||||
//
|
||||
// 为什么不是简单的 term:真实 v0.7.1 的 app 自己写索引时不带 tenant_id,而治理侧
|
||||
// 同步写入时会补齐。于是 "mapping 里有 tenant_id" 不等于 "每篇文档都写了 tenant_id"。
|
||||
// 只按 term 过滤会把 app 写入的正文整库漏掉——真跑踩过:索引里明明有文档,导出 0 条,
|
||||
// 候选校验报 legacy coverage。这里改成:
|
||||
//
|
||||
// 显式写了本租户 → 收;完全没写 → 收(没有可比的租户声明);写了别的租户 → 排除。
|
||||
//
|
||||
// 知识库身份仍然是一条独立的 term,所以范围始终由知识库唯一 ID 收窄,不是放宽成"全库"。
|
||||
func tenantClause(fields searchFieldNames, tenant uint64) any {
|
||||
if fields.tenant == "" {
|
||||
return nil
|
||||
}
|
||||
return map[string]any{"bool": map[string]any{
|
||||
"should": []any{
|
||||
map[string]any{"term": map[string]any{fields.tenant: tenant}},
|
||||
map[string]any{"bool": map[string]any{"must_not": []any{
|
||||
map[string]any{"exists": map[string]any{"field": fields.tenant}},
|
||||
}}},
|
||||
},
|
||||
"minimum_should_match": 1,
|
||||
}}
|
||||
}
|
||||
|
||||
func scopedFilterFor(fields searchFieldNames, identity SearchIdentity) []any {
|
||||
// 租户 + 知识库 + 单文档三层收窄:tenant 那条见 tenantClause(app 写入的文档不带租户字段)。
|
||||
filter := make([]any, 0, 3)
|
||||
if fields.tenant != "" {
|
||||
filter = append(filter, map[string]any{"term": map[string]any{fields.tenant: identity.TenantID}})
|
||||
if clause := tenantClause(fields, identity.TenantID); clause != nil {
|
||||
filter = append(filter, clause)
|
||||
}
|
||||
filter = append(filter,
|
||||
map[string]any{"term": map[string]any{fields.base: identity.KnowledgeBaseID}},
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 全新的 WeKnora 实例会先用空映射建索引(`{}`),字段等第一次写入才动态生成。
|
||||
// 这种索引不能被当成"索引被改坏了":只有确实 0 篇文档时才是空基线;
|
||||
// 有文档却没有范围字段 = 不知道这批文档属于谁,必须报错。
|
||||
func TestEmptyScopeIndexCountsAsEmptyBaselineOnlyWhenTrulyEmpty(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
docCount int
|
||||
wantErr bool
|
||||
}{
|
||||
{"空索引当空基线", 0, false},
|
||||
{"有文档但没有范围字段要报错", 7, true},
|
||||
} {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/chunks/_mapping":
|
||||
_, _ = w.Write([]byte(`{"chunks":{"mappings":{}}}`))
|
||||
case "/chunks/_count":
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{"count": testCase.docCount})
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
exporter, err := NewSearchExporter(SearchExporterConfig{BaseURL: server.URL, Index: "chunks", PageSize: 100})
|
||||
if err != nil {
|
||||
t.Fatalf("exporter: %v", err)
|
||||
}
|
||||
identity := SearchIdentity{TenantID: 10001, KnowledgeBaseID: "kb-1"}
|
||||
count, countErr := exporter.Count(context.Background(), identity)
|
||||
if testCase.wantErr {
|
||||
if countErr == nil {
|
||||
t.Fatalf("有文档却当成空基线:count=%d", count)
|
||||
}
|
||||
} else {
|
||||
if countErr != nil || count != 0 {
|
||||
t.Fatalf("空索引没有被当成空基线:count=%d err=%v", count, countErr)
|
||||
}
|
||||
}
|
||||
_, exportErr := exporter.ExportLibrary(context.Background(), identity, 100)
|
||||
if testCase.wantErr && exportErr == nil {
|
||||
t.Fatal("导出把有文档的空映射索引当成空库")
|
||||
}
|
||||
if !testCase.wantErr && exportErr != nil {
|
||||
t.Fatalf("空索引导出失败:%v", exportErr)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 真实 v0.7.1 的 app 自己写索引时不带 tenant_id,而治理侧同步写入时会补齐。
|
||||
// 所以"mapping 里有 tenant_id"不能当成"每篇文档都写了 tenant_id":只按 term 过滤
|
||||
// 会把 app 写入的正文整库漏掉(真跑踩过:索引里明明有文档,导出 0 条)。
|
||||
func TestTenantClauseAcceptsDocumentsWithoutTenantAndRejectsForeignTenant(t *testing.T) {
|
||||
fields := searchFieldNames{tenant: "tenant_id", knowledge: "knowledge_id", base: "knowledge_base_id"}
|
||||
clause := tenantClause(fields, 10001)
|
||||
encoded, err := json.Marshal(clause)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
text := string(encoded)
|
||||
// 收本租户
|
||||
if !strings.Contains(text, `{"term":{"tenant_id":10001}}`) {
|
||||
t.Fatalf("没有本租户 term:%s", text)
|
||||
}
|
||||
// 收完全没写租户的文档(app 写入形状)
|
||||
if !strings.Contains(text, `"must_not"`) || !strings.Contains(text, `{"exists":{"field":"tenant_id"}}`) {
|
||||
t.Fatalf("没有接受未写租户的文档:%s", text)
|
||||
}
|
||||
// 两种接受条件取其一
|
||||
if !strings.Contains(text, `"minimum_should_match":1`) {
|
||||
t.Fatalf("should 没有收窄成一条:%s", text)
|
||||
}
|
||||
// 写了别的租户的文档:term 不匹配、exists 成立 ⇒ 两条都不满足 ⇒ 被排除
|
||||
if strings.Contains(text, `"minimum_should_match":0`) {
|
||||
t.Fatalf("should 允许全不匹配,等于放弃租户范围:%s", text)
|
||||
}
|
||||
}
|
||||
|
||||
// mapping 里没有租户字段时不能凭空造一条过滤条件。
|
||||
func TestTenantClauseIsAbsentWithoutTenantField(t *testing.T) {
|
||||
if clause := tenantClause(searchFieldNames{knowledge: "knowledge_id", base: "knowledge_base_id"}, 10001); clause != nil {
|
||||
t.Fatalf("没有租户字段却生成了过滤条件:%v", clause)
|
||||
}
|
||||
}
|
||||
|
||||
// 整库范围过滤仍然带知识库身份,未写租户的文档也进得来。
|
||||
func TestScopeFilterKeepsKnowledgeBaseIdentity(t *testing.T) {
|
||||
fields := searchFieldNames{tenant: "tenant_id", knowledge: "knowledge_id", base: "knowledge_base_id"}
|
||||
encoded, err := json.Marshal(scopeFilterFor(fields, SearchIdentity{TenantID: 10001, KnowledgeBaseID: "kb-1"}))
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
text := string(encoded)
|
||||
if !strings.Contains(text, `{"term":{"knowledge_base_id":"kb-1"}}`) {
|
||||
t.Fatalf("知识库身份丢了:%s", text)
|
||||
}
|
||||
if !strings.Contains(text, `"must_not"`) {
|
||||
t.Fatalf("未写租户的文档会被漏掉:%s", text)
|
||||
}
|
||||
}
|
||||
@@ -19,14 +19,25 @@ func (exporter *SearchExporter) Count(ctx context.Context, identity SearchIdenti
|
||||
return 0, errors.New("invalid search identity")
|
||||
}
|
||||
inspection, err := exporter.inspectMapping(ctx)
|
||||
if errors.Is(err, errScopeFieldsMissing) {
|
||||
// 新建索引还没有范围字段:只有确实为空时才等于"这个范围 0 条"。
|
||||
empty, probeErr := exporter.emptyScopeIndex(ctx)
|
||||
if probeErr != nil {
|
||||
return 0, probeErr
|
||||
}
|
||||
if empty {
|
||||
return 0, nil
|
||||
}
|
||||
return 0, err
|
||||
}
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
// 只按"租户 + 知识库"计数:整库同步没有单文档身份,带上空的 knowledge 条件
|
||||
// 会把过滤器变成"匹配空字符串",永远数不到东西。
|
||||
filter := make([]any, 0, 2)
|
||||
if inspection.fields.tenant != "" {
|
||||
filter = append(filter, map[string]any{"term": map[string]any{inspection.fields.tenant: identity.TenantID}})
|
||||
if clause := tenantClause(inspection.fields, identity.TenantID); clause != nil {
|
||||
filter = append(filter, clause)
|
||||
}
|
||||
filter = append(filter, map[string]any{"term": map[string]any{inspection.fields.base: identity.KnowledgeBaseID}})
|
||||
payload := map[string]any{"query": map[string]any{"bool": map[string]any{"filter": filter}}}
|
||||
@@ -105,8 +116,8 @@ func (exporter *SearchExporter) PruneScope(ctx context.Context, identity SearchI
|
||||
// scopeFilterFor 只按租户 + 知识库过滤(整库范围,不含单文档身份)。
|
||||
func scopeFilterFor(fields searchFieldNames, identity SearchIdentity) []any {
|
||||
filter := make([]any, 0, 2)
|
||||
if fields.tenant != "" {
|
||||
filter = append(filter, map[string]any{"term": map[string]any{fields.tenant: identity.TenantID}})
|
||||
if clause := tenantClause(fields, identity.TenantID); clause != nil {
|
||||
filter = append(filter, clause)
|
||||
}
|
||||
filter = append(filter, map[string]any{"term": map[string]any{fields.base: identity.KnowledgeBaseID}})
|
||||
return filter
|
||||
|
||||
@@ -9,8 +9,24 @@ set -euo pipefail
|
||||
ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
|
||||
UNIT_DIR="$HOME/.config/systemd/user"
|
||||
SOCKET_DIR="$HOME/.local/share/knowledge-governance/agent"
|
||||
# 实例级维护接线(LIBRARY_TARGET_*)与源声明一起由这一个 0600 文件提供;
|
||||
# 它属于"这台主机上的哪个实例",不属于任何一次请求。
|
||||
ENV_FILE="${AGENT_ENV_FILE:-$HOME/.local/share/knowledge-governance/agent.env}"
|
||||
SOURCE_APP="${AGENT_SOURCE_APP:-WKNFull-app}"
|
||||
SOURCE_PG="${AGENT_SOURCE_PG:-WKNFull-postgres}"
|
||||
SOURCE_ES="${AGENT_SOURCE_ES:-WKNFull-elasticsearch}"
|
||||
PROJECT="${AGENT_PROJECT:-weknora071full}"
|
||||
WRITERS="${AGENT_WRITERS:-app}"
|
||||
NON_WRITERS="${AGENT_NON_WRITERS:-frontend=UI only; no knowledge writes,mcp=read-only MCP server,dobotai-target=alpine placeholder; no knowledge writes,docreader=parse RPC service; app persists parse output,elasticsearch=search store; not an initiator,postgres=database store; not an initiator,redis=queue/cache store; not an initiator,neo4j=graph store; not an initiator,langfuse-web=LLM observability UI,langfuse-worker=LLM observability worker; own storage,langfuse-clickhouse=observability store,langfuse-minio=observability object store,langfuse-db-init=one-shot init; exited,searxng=read-only search proxy,searxng-init=one-shot init; exited}"
|
||||
mkdir -p "$UNIT_DIR" "$SOCKET_DIR"
|
||||
chmod 700 "$SOCKET_DIR"
|
||||
if [ ! -f "$ENV_FILE" ]; then
|
||||
# 宁可不装,也不要装出一个"没有闸门"的 agent:封存会在开窗口前被拒。
|
||||
echo "AGENT_ENV_MISSING: $ENV_FILE" >&2
|
||||
echo " write LIBRARY_TARGET_{INSTANCE,INGRESS_URL,INGRESS_SERVICE,INGRESS_DRAIN_FILE,REDIS_ADDR,REDIS_PASSWORD,REDIS_DB} there (0600)" >&2
|
||||
exit 1
|
||||
fi
|
||||
chmod 600 "$ENV_FILE"
|
||||
|
||||
echo "=== build libraryctl + libraryagent ==="
|
||||
(cd "$ROOT" && go build -o libraryctl ./cmd/libraryctl && go build -o libraryagent ./cmd/libraryagent)
|
||||
@@ -24,13 +40,18 @@ After=network.target
|
||||
[Service]
|
||||
Type=simple
|
||||
WorkingDirectory=$ROOT
|
||||
EnvironmentFile=$ENV_FILE
|
||||
ExecStart=$ROOT/libraryagent \\
|
||||
-socket $SOCKET_DIR/agent.sock \\
|
||||
-binary $ROOT/libraryctl \\
|
||||
-archive $HOME/.local/share/knowledge-governance/library \\
|
||||
-journal $HOME/.local/share/knowledge-governance/freeze \\
|
||||
-project weknora071full \\
|
||||
-writers app
|
||||
-source-app $SOURCE_APP \\
|
||||
-source-pg $SOURCE_PG \\
|
||||
-source-es $SOURCE_ES \\
|
||||
-project $PROJECT \\
|
||||
-writers $WRITERS \\
|
||||
-non-writers "$NON_WRITERS"
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
|
||||
|
||||
Reference in New Issue
Block a user