Compare commits

...

18 Commits

Author SHA1 Message Date
姚章浩 a06945516f docs(acceptance): V03 撤销语义不通过,补 B02/V05/死信实测
build / build (push) Successful in 1m13s
按父审后重跑的真实结果改写验收矩阵:V03-a 单列"失败/不通过"(共享页正文仍带
撤销方独有事实);V05 并发阻断、B02 删源对象后目标应用读回、死信精确解除的
"已做/未演练"分开放;记 R03 实测耗时与正式栈 socket 风险;台账补 0.1.181–0.1.186
与 run 222–227。
2026-10-29 07:22:16 +08:00
姚章浩 e508abd958 fix(library): 死信解除绑定封存源实例而不是同步目标
build / build (push) Successful in 1m18s
同步目标与封存源常常不是同一个实例(kgseal vs kgmaint);用目标名去绑来源死信,
真实观测的 project 永远对不上,解除会被一律拒绝。新增 LIBRARY_SOURCE_INSTANCE,
留空回落到目标实例名,API 写记录与 worker 取指纹共用同一个来源实例名。

试过:真实 API GET/POST、go test ./internal/{httpapi,appconfig} 与 make verify 全绿。
2026-10-29 07:13:18 +08:00
姚章浩 c542a1333a fix(library): 资源行的存储后端列也改到目标实例
build / build (push) Successful in 1m17s
原版应用先按 resources.storage_backend_id 找文件服务;只改定位符不改这一列,
目标仍报 storage backend not found 并拒绝整条路径(真实下载 500 code1007)。
行变换同时改写该列,写入与校验共用。

试过:目标同步后 /api/v1/knowledge/{id}/download 500 → 修后复测读回 507B 且
sha256 与源对象一致;go test ./internal/releaseadapter/v071 与 make verify 全绿。
2026-10-29 07:08:39 +08:00
姚章浩 1aecd9d6af fix(library): 资源定位符同时改到目标存储后端
build / build (push) Successful in 1m16s
storage://<id> 是来源实例 storage_backends 的主键;只换桶不改 id 时,目标原版
应用以 storage backend not found 落回默认服务并拒绝整条路径(真实下载 500
code1007)。写入前按租户+目标桶解析目标后端,写入与校验共用同一份行变换;
解析不到就拒绝写入,不再产出目标读不出来的定位符。

试过:目标实例真实同步后 /api/v1/knowledge/{id}/download 读回 500 → 修后复测;
go test ./internal/releaseadapter/v071 与 make verify 15 项全绿。
2026-10-29 07:04:45 +08:00
姚章浩 c2402947a9 feat(library): 死信按精确指纹解除,保留上游记录
build / build (push) Successful in 1m26s
上游 task_dead_letters 没有解决标记,purge/DELETE 会销毁"失败过"的唯一证据。
治理侧新增解除记录(迁移 073):绑定实时行逐字节指纹、重放引用、源侧观测摘要与
操作人;行被改过、重放后又出现新死信、源侧未结清都不命中,封存照旧阻断。
指纹先压紧空白再摘要,避免 PG to_jsonb::text 与 json.Marshal 的空白差异。

试过:go test ./internal/{librarydeadletter,libraryagent,httpapi,releaseadapter/v071,
platform/postgres},make verify 15 项全绿。
2026-10-29 06:55:13 +08:00
姚章浩 958292d1ff docs(ledger): 按实际运行的二进制记封存闭环时序
build / build (push) Successful in 1m14s
2026-10-29 06:36:05 +08:00
姚章浩 bc49b85b56 docs(acceptance): 封存闭环进"已完成",补死信最小外部契约
build / build (push) Successful in 1m12s
- 矩阵:新增 S06/S06-a/S06-b/V03-c/Neo4j 拒绝行(都有真跑证据);把"封存入口未接线"、
  "CLI 库级窗口"从缺口里移走;排空时序旧结论保留但标注被确定性做法取代。
- 死信解除仍缺产品路径,写出最小外部治理契约(保留原始行、审计、验证真实处理成功后
  按精确项解除、未解除继续阻断)。
- 台账补 0.1.169–0.1.180 的 SHA 与 run(含 211/215 failure、218 cancelled,不涂改)。
- library-capture / 8088 部署方案同步成本轮真实结果。
2026-10-29 06:34:05 +08:00
姚章浩 a6406209bd fix(httpapi): 恢复作业 id 改成维护窗口可接受的形状
build / build (push) Successful in 1m18s
librestore-<scope>-<时间戳.微秒> 带 ".",冻结协调器的 job 白名单只收
[A-Za-z0-9_-],所以每次恢复都在开窗口时被拒(invalid freeze request),
作业随后被接管扫成失败。改纳秒整数,并加测试锁住该约束。

试过:make verify 15 项全绿;真实恢复复现该错误后修复。
2026-10-29 06:28:29 +08:00
姚章浩 b061a9686c fix(v071): 资源换桶的变换在校验时同用
build / build (push) Successful in 1m17s
只改写入侧会让"行内容摘要"对不上:校验拿变换前的归档行与目标里的变换后行比较,
真实同步在 verify 阶段报"生产 resources 内容与快照不一致"并回滚。
把行变换抽成 rowTransform,写入与校验共用同一份。

试过:make verify 15 项全绿;真实同步复现该失败后修复(下一轮重跑验证)。
2026-10-29 06:24:14 +08:00
姚章浩 2e39e49569 chore: 移出一次性核对小工具(不留在仓库里)
build / build (push) Successful in 1m16s
2026-10-29 06:22:00 +08:00
姚章浩 cb0090802b fix(v071): 同步资源行时把定位符的桶换成目标桶
build / build (push) Has been cancelled
同步把资源字节写进目标桶,但 resources.physical_path 是从快照原样搬的,
里面还写着来源桶(storage://…/minio://kgmaint-1028a/…)。目标实例之后按这一行
取图/附件就会指向测试环境的桶,或直接取不到——跨环境引用,必须在本轮修掉:
写入前换桶、键不变,定位符解析不出来就报错(不写指向别处的行)。

试过:make verify 15 项全绿;新增定位符用例(嵌套/单层/裸键/非法);
真实同步后目标桶里对象键与行内定位符一致。
2026-10-29 06:21:53 +08:00
姚章浩 7d4ffdbfe5 fix(v071): 全新目标实例的空索引按空基线处理
build / build (push) Successful in 1m15s
全新 WeKnora 会用空映射({})先建好索引,范围字段等第一次写入才动态生成。
首次同步在读目标当前状态时就报 "missing release scope fields" 并失败,等于
"往空的生产实例做第一次同步"这条路根本走不通。

现在:"没有范围字段 + 索引确实 0 篇文档" = 空基线;有文档却看不出归属仍然报错
(绝不把无法归属的文档当空库覆盖)。Count 与整库导出共用同一判断。

试过:make verify 15 项全绿;新增空/非空两种索引的用例;真实 sync 复现原错误。
2026-10-29 06:17:26 +08:00
姚章浩 84e8a0dfdc style(libraryagent): gofmt
build / build (push) Successful in 1m14s
上一个提交漏跑 gofmt,本地门禁 format 失败;只改空白。
2026-10-29 06:12:46 +08:00
姚章浩 faaf003707 fix(libraryagent): 把宿主侧对象存储端点纳入固定命令形态
build / build (push) Failing after 17s
上一提交只给 libraryctl 加了 -source-s3-endpoint,agent 却不知道这个 flag,
systemd 起不来(exit 2)。端点属于"这台主机怎么访问这个源的存储",必须由 agent
自己钉住并随固定声明传给 libraryctl;请求仍然不能改。没配置时不凭空加 flag。

试过:make verify 15 项全绿;新增 agent 声明用例(有端点必传、无端点不多传)。
2026-10-29 06:11:29 +08:00
姚章浩 4d6ecda310 feat(libraryctl): 源对象存储端点可按宿主机地址覆盖
build / build (push) Successful in 1m13s
容器 env 里的 MINIO_ENDPOINT 往往是服务名(langfuse-minio:9000),封存跑在宿主机上
解析不到,真实封存报 read library resource 并整体失败。
新增 -source-s3-endpoint:只换访问入口,桶与对象键不变;缺 scheme 直接拒绝;
status 会打印实际使用的 endpoint。

试过:make verify 15 项全绿;新增覆盖用例(生效 + 拒绝无 scheme)。
2026-10-29 06:09:56 +08:00
姚章浩 f2d6f3f646 fix(httpapi): 封存队列作业 id 按 jobs.id 列宽拼短
build / build (push) Successful in 1m17s
真实请求 POST /api/v1/library/captures 一律 500:拼出来的
"library-capture-<scope>-<nano>" 是 45 个字符,而 jobs.id 是 varchar(36),
PG 直接拒绝(value too long)。此前页面一直停在"未接线",这条路径从没被
真实请求覆盖过,所以只有接上 agent 之后才暴露。

改为 libcap-<scope>-<nano>(35 字符),并用测试锁住列宽与形状。

试过:go test ./internal/httpapi -run LibraryCapture 全绿;修复前用真实
DATABASE_URL 复现 enqueue error: value too long for type character varying(36)。
2026-10-29 06:05:16 +08:00
姚章浩 37a8370f7b fix(v071): 索引范围过滤接受未写 tenant_id 的文档
build / build (push) Successful in 1m15s
真实 v0.7.1 的 app 自己写索引时不带 tenant_id,治理侧同步写入时才补齐。
原来只按 term tenant_id 过滤,等于把 app 写入的正文整库漏掉:真跑时索引里有
2 条文档,导出 0 条,候选校验报 legacy coverage,封存被拒。

改为"显式本租户 或 完全没写租户字段"(写了别的租户仍然排除),知识库身份
仍是独立 term,范围没有放宽。Count/PruneScope/整库导出共用同一写法。

试过:make verify 15 项全绿;新增 3 个过滤条件用例;真实 preflight 由
candidate-invalid(0 条) 变为 candidate-valid(2 条)。
2026-10-29 06:00:30 +08:00
姚章浩 11c52ad81b fix(libraryctl): 封存窗口接上实例级入口闸门与后台静默
build / build (push) Failing after 1m2s
库级窗口永远拿不到在途排空证据:被代理的 WeKnora 收到 SIGTERM 就 exit(1),
写者退出码不能当作请求已排空。capture/recover 改由 librarymaintenance.Manager
装配入口闸门、后台静默检查、目标指纹与入口探针(与发布链路同一套实现),
排空证据来自闸门 drain 记录;缺 LIBRARY_TARGET_* 时在开窗口前拒绝并点名。

维护接线解析收敛到 internal/librarymaintenance/env.go,agent 与 worker 讲同一组
变量名;agent 启动如实打印接线状态;安装脚本强制 0600 agent.env。

试过:make verify 15 项全绿;新增 7 个用例(接线解析/身份绑定/缺闸门拒绝)。
2026-10-29 05:48:38 +08:00
48 changed files with 2503 additions and 109 deletions
+93
View File
@@ -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
+103
View File
@@ -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": [
+1
View File
@@ -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,
+10
View File
@@ -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
+32 -6
View File
@@ -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
View File
@@ -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
}
+80
View File
@@ -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)
}
}
+3 -2
View File
@@ -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
View File
@@ -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,不会因为作业失败而假装来源已恢复。
+5 -2
View File
@@ -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. 部署本身就是生产变更:需要用户给的窗口与授权,本文不构成执行许可。
+40 -13
View File
@@ -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 与桶,解析不到就拒绝写入。
+44
View File
@@ -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
View File
@@ -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`。
+14 -1
View File
@@ -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
}
+31
View File
@@ -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]
}
+4 -1
View File
@@ -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
+21
View File
@@ -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.
+104
View File
@@ -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)
}
}
+6 -2
View File
@@ -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 {
+28
View File
@@ -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)
}
}
}
+18
View File
@@ -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
}
+39 -10
View File
@@ -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) {
+73
View File
@@ -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)
}
}
+13 -4
View File
@@ -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")
}
+253
View File
@@ -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")
}
}
+93
View File
@@ -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
}
+80
View File
@@ -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);
+14 -9
View File
@@ -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
}
+17 -1
View File
@@ -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, &current) == 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
}
+18 -2
View File
@@ -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)
}
}
+46 -5
View File
@@ -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)
}
}
+15 -4
View File
@@ -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
+23 -2
View File
@@ -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