feat(library): 目标库隔离 ID 映射并保留旧库
build / build (push) Successful in 2m11s

同实例第二个目标 KB 与旧库共享源 UUID 时,计划/写入/校验/恢复按目标范围派生稳定身份;
计划持久化 target_id_mode,旧计划与写前镜像不二次映射;资源保持共享 ID。
This commit is contained in:
姚章浩
2026-11-04 12:07:27 +08:00
parent 6042955498
commit e0fc2c32b8
13 changed files with 1324 additions and 5 deletions
+18
View File
@@ -1,5 +1,23 @@
# Changelog
## 开发快照(目标库独立 ID 映射,保留旧库)
### Changed
- **同一源快照可导入同实例的第二个知识库**:写入目标前按 `sha256(目标 tenant + 目标 KB + 表 + 源身份)`
派生目标侧 UUID(knowledges / chunks / knowledge_tags / wiki_pages / wiki_folders /
wiki_page_issues / resource_bindings)。同一目标重复同步、重试与跨版本得到同一 ID,不同目标
或不同表得到不同 ID;旧库原有行不再与本次写入撞主键,也不会被删除。
- **引用字段一起改写**:chunks 的 parent/pre/next 与 relation/metadata 中的精确身份、
wiki_pages 的 chunk_refs / source_refs(保留 `id|title` 与顺序、重复)、folder_id、
resource_bindings.owner_id、索引文档物理 ID 与 chunk_id/knowledge_id/source_id/tag_id;
正文、标题、向量、引用顺序与重复项保持不变,不重新解析、不重新 Embedding。
- **resources 保持原 ID**:资源是多知识库共享的字节载体,继续走既有"内容相同 + 来源可审计"复用
与归属证明;只改绑定行的 id/owner_id,不改 resource_id。
- **写集合、校验、索引计划与恢复共用同一映射**:写集合登记目标身份,verify 按目标身份比对;
恢复重放目标写前镜像时不二次映射。引用指向未登记身份时直接拒绝,不做模糊替换。
## 开发快照(共享桶维护式更新:取消整桶独占前置)
### Changed
+48
View File
@@ -0,0 +1,48 @@
# 目标库隔离 ID 映射
同一 WeKnora 实例里可以有多个知识库。同步把源快照写进目标库时,如果沿用源库的 UUID 主键,
第二个目标库会与实例里已有库的行撞主键(真实证据:2 knowledges / 2 chunks / 23 wiki_pages /
17 wiki_folders 已存在于旧库),写入前的跨范围检查会拒绝整次同步。关闭检查等于让两个库共享身份,
删除/恢复/校验会互相误伤,因此身份必须在目标侧派生。
## 映射规则
```
digest = sha256("knowledge-governance/target-identity/v1" || tenant || kb || table || source_id)
target_id = 前 16 字节按 8-4-4-4-12 十六进制格式化(置 RFC 4122 variant/version 位)
```
这是**本产品自定义的稳定 ID 算法**(`target_id_mode=scoped-sha256-v1`),只保证确定性与目标隔离;
它不是标准 RFC 4122 UUIDv5(UUIDv5 用 SHA-1 + 标准 namespace),不要按 UUIDv5 互操作。
- 只绑定**稳定目标身份**(目标 tenant、目标 KB)与源身份,表名参与命名空间;
- 不绑定 job、时间、快照版本等瞬态值:同一目标重复同步/重试/跨版本得到同一 ID;
- 不同目标 KB、不同表得到不同 ID;
- 源归档字节不变,映射只存在于目标侧派生视图(写集合、写入行、索引文档、校验预期);
- 计划把策略持久化为 `target_id_mode=scoped-sha256-v1`:新进程按该字段重建映射,旧计划/写前镜像
没有该字段则按原 ID 回放,绝不二次映射。
## 参与映射的表与字段
| 表 | 身份 | 引用字段 |
|---|---|---|
| knowledges | id | — |
| chunks | id | knowledge_id, tag_id, parent_chunk_id, pre_chunk_id, next_chunk_id, relation_chunks, indirect_relation_chunks, metadata(仅精确匹配) |
| knowledge_tags | id | — |
| knowledge_tag_relations | (knowledge_id, tag_id) | 两个分量分别映射 |
| wiki_folders | id | parent_id |
| wiki_pages | id | folder_id, chunk_refs, source_refs(保留 `id|title`), page_metadata(仅精确匹配) |
| wiki_page_issues | id | page_id, knowledge_id, chunk_id |
| resource_bindings | id | owner_id(按 owner_type) |
| resources | **不映射** | resource_id 保持共享身份 |
| 索引文档 | 文档物理 ID | chunk_id, knowledge_id, source_id, tag_id |
正文、标题、摘要、向量、slug、in_links/out_links、pages_affected 不改写;本地自增列
(如 `wiki_log_entries.id`)仍由既有 LocalIDs 分配。引用指向未登记身份时拒绝(悬挂引用不静默放行)。
## 一致性边界
- 写集合登记的是目标身份;verify 的身份摘要与内容摘要都按同一 `rowTransform` 计算;
- 索引计划的目标文档 ID 与 `_source` 引用同步映射,before-image(目标写前镜像)保持原样;
- 恢复重放写前镜像时映射关闭,避免把目标原始 ID 再映射一次;
- 资源对象路径/location_hash 仍由既有资源改写逻辑处理,不因 ID 映射而变化。
+17 -1
View File
@@ -98,7 +98,10 @@ type WriteSet struct {
// Kind 区分发布执行计划与受限演练恢复计划。空值 = 历史发布计划。
// 演练计划必须显式为 restore-drill:它只授权 canary 回放,不含对象/索引
// 写项,读取方不能把它当发布计划接管。
Kind string `json:"kind,omitempty"`
Kind string `json:"kind,omitempty"`
// TargetIDMode 记录这份计划是否把源身份映射成目标隔离身份,以及用哪套算法。
// 空值 = 历史计划/写前镜像:按原 ID 回放,绝不能在恢复时二次映射。
TargetIDMode string `json:"target_id_mode,omitempty"`
Index []IndexWrite `json:"index,omitempty"`
Objects []ObjectWrite `json:"objects,omitempty"`
SchemaDigest string `json:"schema_digest,omitempty"`
@@ -203,7 +206,20 @@ func (w *WriteSet) computeDigest() (string, error) {
// Verify 校验摘要与必备身份。缺任一身份都必须失败:没有绑定就没有"这份记录属于
// 这个作业/目标/预备份"的证明,拿它去删目标数据无法追责。
const (
// TargetIDModeScoped 是目标隔离身份的持久契约标识。
// 算法是自定义的 sha256(namespace|tenant|kb|table|source) 16 字节派生,
// 只保证确定性与隔离,不自称标准 UUIDv5。
TargetIDModeScoped = "scoped-sha256-v1"
)
func (w *WriteSet) Verify() error {
if w.TargetIDMode != "" && w.TargetIDMode != TargetIDModeScoped {
return fmt.Errorf("unsupported target id mode %q", w.TargetIDMode)
}
if w.TargetIDMode == TargetIDModeScoped && (w.TenantID == 0 || w.KnowledgeBaseID == "") {
return errors.New("target id mode lacks target identity")
}
if w.ContractVersion != 0 && w.ContractVersion != 1 {
return errors.New("unsupported migration contract version")
}
@@ -73,6 +73,10 @@ func (t *ProductionLibraryTarget) planIndex(ctx context.Context, archive *librar
if err != nil {
return err
}
after, err = t.remapIndexMap(after)
if err != nil {
return err
}
ids := map[string]bool{}
for id := range before {
ids[id] = true
@@ -126,6 +130,10 @@ func (t *ProductionLibraryTarget) validateIndexPlanBinding(archive *librarysnaps
if err != nil {
return err
}
after, err = t.remapIndexMap(after)
if err != nil {
return err
}
ids := map[string]bool{}
for id := range before {
ids[id] = true
@@ -131,6 +131,12 @@ type ProductionLibraryTarget struct {
// 每次重建时都实时重读并比较:同名重建的目标/索引不可能复用旧证明。
restoreFingerprint string
restoreIndexUUID string
// targetIDs / mappingActive 是本作业的目标隔离映射;只对目标快照生效。
targetIDs *targetIDMapper
mappingActive bool
// targetIDMode 来自持久计划(或本次新建计划)。只有它等于 scoped 契约时
// 才允许映射:旧计划/写前镜像的 ID 已经是目标身份,重映射会写到别处。
targetIDMode string
// restoreIntentFault 是一次性文件写失败注入点(生产恒为 nil):在 prepared 已提交、
// mark-intent 尚未落盘处返回错误,用来复现"可补同内容"的窄故障区段。
restoreIntentFault func(libraryrelease.IndexWrite) error
@@ -356,6 +362,12 @@ func (t *ProductionLibraryTarget) Apply(ctx context.Context, stage libraryreleas
if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil {
return err
}
t.disableTargetMapping()
if t.targetMappingApplies(snapshotID) {
if err := t.enableTargetMapping(archive, manifest); err != nil {
return err
}
}
switch stage {
case libraryrelease.StageDatabase, libraryrelease.StageWiki:
kind := "database"
@@ -707,6 +719,15 @@ func (t *ProductionLibraryTarget) importIndex(ctx context.Context, archive *libr
if err != nil {
return err
}
if t.mappingActive && t.targetIDs != nil {
for i := range documents {
mappedID, mappedSource, err := t.targetIDs.remapIndexDocument(documents[i].ID, documents[i].Source)
if err != nil {
return err
}
documents[i] = librarysnapshot.IndexDocument{ID: mappedID, Source: mappedSource}
}
}
// 直接搬运原始文档与向量:不解析、不重新 Embedding。
identity := SearchIdentity{TenantID: t.scope.TenantID, KnowledgeBaseID: t.scope.KnowledgeBaseID}
write := t.search.BulkIndex
@@ -747,6 +768,12 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil {
return err
}
t.disableTargetMapping()
if t.targetMappingApplies(expect.SnapshotID) {
if err := t.enableTargetMapping(archive, manifest); err != nil {
return err
}
}
for _, part := range manifest.Parts {
if part.Kind != "database" && part.Kind != "wiki" {
continue
@@ -758,7 +785,7 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
return err
}
}
wantCount, wantDigest, err := archivedTableState(archive, part)
wantCount, wantDigest, err := t.archivedTableState(archive, part)
if err == nil && part.Name == "wiki_log_entries" && t.activePlan != nil && t.activeSnapshot == t.activePlan.TargetSnapshotID && !t.preserveChunkSequence {
ids := []string{}
for _, mapping := range t.activePlan.LocalIDs {
@@ -952,6 +979,25 @@ func (t *ProductionLibraryTarget) AdoptWriteSet(set libraryrelease.WriteSet) err
if set.ContractVersion != 0 && set.ContractVersion != migrationIdentityContractVersion {
return errors.New("unsupported migration identity contract")
}
t.targetIDMode = set.TargetIDMode
if t.targetIDMode == libraryrelease.TargetIDModeScoped {
// 新进程接管计划时没有内存里的映射器:按持久契约从目标快照重建,
// 否则 validatePlanBindings 会拿索引原 ID 去比映射后的计划而失败。
archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
if err != nil {
return err
}
manifest, err := archive.Load(set.TargetSnapshotID)
if err != nil {
_ = archive.Close()
return err
}
err = t.enableTargetMapping(archive, manifest)
_ = archive.Close()
if err != nil {
return err
}
}
if err := t.validatePlanBindings(set); err != nil {
return err
}
@@ -1192,6 +1238,8 @@ func (t *ProductionLibraryTarget) Restore(ctx context.Context, backup libraryrel
if err := backup.WriteSet.ValidateRestoreBinding(backup.SnapshotID, t.scope.TenantID, t.scope.KnowledgeBaseID); err != nil {
return err
}
// 恢复重放的是目标写前镜像,行里已经是目标原始 ID;再映射一次会写到别的身份上。
t.disableTargetMapping()
t.restoreScope = t.config.RestoreScope
t.restoreSourceSnapshot = backup.WriteSet.TargetSnapshotID
t.restoreBackupSnapshot = backup.SnapshotID
@@ -51,6 +51,11 @@ type manifestScope struct {
// 换成目标身份,资源行还要把定位符、存储后端、提供方换成目标实例自己的值。
func (t *ProductionLibraryTarget) retargetRow(table string, row map[string]json.RawMessage) error {
if !tableHasTenant(table) {
// 关系表没有 tenant/kb 列,但它的两个分量都是身份:范围改写跳过它时,
// 身份映射仍必须执行,否则写集合/写入会出现未映射元组。
if t.mappingActive && t.targetIDs != nil {
return t.targetIDs.remapRow(table, row)
}
return nil
}
if err := t.retargetTenantField(row); err != nil {
@@ -65,7 +70,14 @@ func (t *ProductionLibraryTarget) retargetRow(table string, row map[string]json.
return t.rewriteResourceIdentity(row)
}
if table == "knowledges" {
return t.retargetEmbeddingModelField(row)
if err := t.retargetEmbeddingModelField(row); err != nil {
return err
}
}
// 身份与引用最后换:前面的范围/资源/模型改写都按来源身份读字段,
// 先换 ID 会让它们找不到要改的列。
if t.mappingActive && t.targetIDs != nil {
return t.targetIDs.remapRow(table, row)
}
return nil
}
@@ -0,0 +1,468 @@
package v071
import (
"bytes"
"crypto/sha256"
"encoding/json"
"errors"
"fmt"
"sort"
"strconv"
"strings"
"knowledge-governance/internal/librarysnapshot"
)
// 目标隔离 ID 映射。
//
// 为什么必须映射:同一实例里可以存在多个知识库,源库的 UUID 主键被原样写进第二个
// 目标库时会与旧库的行撞主键(父证据:2 knowledges / 2 chunks / 23 wiki_pages /
// 17 wiki_folders 已存在于旧 KB),而写入前的"跨范围冲突"检查会(正确地)拒绝。
// 关掉检查等于让两个库共享身份,删除/恢复/校验都会互相误伤。
//
// 因此身份必须由"目标范围 + 表 + 源身份"确定性派生:
// - 同一目标重复同步/重试/跨版本 → 同一 ID(幂等、可恢复);
// - 不同目标 KB、不同表 → 不同 ID(隔离);
// - 源归档字节不变,映射只存在于目标侧派生视图。
//
// 资源表(resources)是多知识库共享的字节载体,ID 不重映射,沿用既有复用/归属证明;
// 绑定行的 owner_id 与自身 id 则按目标映射。
type targetIDMapper struct {
tenant uint64
kb string
ids map[string]string
reverse map[string]string
known map[string]map[string]bool
owner map[string]string
ambiguous map[string]bool
sequence int
// 不重映射的表:资源行是跨库共享对象,改了 ID 就等于复制一份字节,
// 现有复用契约要求"内容相同 + 来源可审计"才复用,不能靠换 ID 绕过。
skip map[string]bool
}
const targetIDNamespace = "knowledge-governance/target-identity/v1"
var errUnmappedTargetReference = errors.New("unmapped target reference")
func newTargetIDMapper(tenant uint64, kb string, _ any) *targetIDMapper {
return &targetIDMapper{
tenant: tenant, kb: kb,
ids: map[string]string{}, reverse: map[string]string{},
known: map[string]map[string]bool{}, owner: map[string]string{}, ambiguous: map[string]bool{},
skip: map[string]bool{"resources": true},
}
}
func (m *targetIDMapper) mapID(table, source string) string {
if source == "" || m.skip[table] {
return source
}
key := table + "\x00" + source
if mapped, ok := m.ids[key]; ok {
return mapped
}
sum := sha256.Sum256([]byte(targetIDNamespace + "\x00" + strconv.FormatUint(m.tenant, 10) + "\x00" + m.kb + "\x00" + table + "\x00" + source))
var raw [16]byte
copy(raw[:], sum[:16])
raw[6] = raw[6]&0x0f | 0x50 // version 5 semantics: name-based, deterministic
raw[8] = raw[8]&0x3f | 0x80 // RFC 4122 variant
mapped := fmt.Sprintf("%x-%x-%x-%x-%x", raw[0:4], raw[4:6], raw[6:8], raw[8:10], raw[10:16])
if previous, exists := m.reverse[table+"\x00"+mapped]; exists && previous != source {
// 位移后的 SHA-256 也撞同一张表是密码学事件;宁可停住也不静默合并两个身份。
panic("target id collision for " + table)
}
m.ids[key] = mapped
m.reverse[table+"\x00"+mapped] = source
m.sequence++
return mapped
}
func (m *targetIDMapper) registerSourceID(table, id string) {
if id == "" || m.skip[table] {
return
}
// 本地自增主键(如 wiki_log_entries.id)不是跨库身份:既不能当引用匹配,
// 也不能参与"残留源 ID"判断,否则普通整数列会被误报。
if contract, ok := migrationIdentityContracts[table]; ok && contract.Kind == "local-bigint" {
return
}
if m.known[table] == nil {
m.known[table] = map[string]bool{}
}
m.known[table][id] = true
if previous, exists := m.owner[id]; !exists {
m.owner[id] = table
} else if previous != table {
// 两张表出现同一字符串 ID:无法从值本身判定引用指向哪张表,
// 只能报歧义,不能 first-table-wins 猜一个。
m.ambiguous[id] = true
}
}
// remapRow 只改"身份与引用"字段;正文、向量、顺序、重复项一律原样保留。
func (m *targetIDMapper) remapRow(table string, row map[string]json.RawMessage) error {
if m.skip[table] {
return nil
}
switch table {
case "knowledges":
if err := m.remapScalar(row, "id", table); err != nil {
return err
}
case "chunks":
if err := m.remapScalar(row, "id", table); err != nil {
return err
}
if err := m.remapScalar(row, "knowledge_id", "knowledges"); err != nil {
return err
}
if err := m.remapScalar(row, "tag_id", "knowledge_tags"); err != nil {
return err
}
for _, column := range []string{"parent_chunk_id", "pre_chunk_id", "next_chunk_id"} {
if err := m.remapScalar(row, column, "chunks"); err != nil {
return err
}
}
for _, column := range []string{"relation_chunks", "indirect_relation_chunks"} {
if err := m.remapJSONExact(row, column); err != nil {
return err
}
}
case "knowledge_tags":
if err := m.remapScalar(row, "id", table); err != nil {
return err
}
case "knowledge_tag_relations":
if err := m.remapScalar(row, "knowledge_id", "knowledges"); err != nil {
return err
}
if err := m.remapScalar(row, "tag_id", "knowledge_tags"); err != nil {
return err
}
case "wiki_folders":
if err := m.remapScalar(row, "id", table); err != nil {
return err
}
if err := m.remapScalar(row, "parent_id", table); err != nil {
return err
}
case "wiki_pages":
if err := m.remapScalar(row, "id", table); err != nil {
return err
}
if err := m.remapScalar(row, "folder_id", "wiki_folders"); err != nil {
return err
}
if err := m.remapRefArray(row, "chunk_refs", "chunks"); err != nil {
return err
}
if err := m.remapSourceRefs(row); err != nil {
return err
}
case "wiki_page_issues":
if err := m.remapScalar(row, "id", table); err != nil {
return err
}
for _, column := range []string{"page_id", "knowledge_id", "chunk_id"} {
if err := m.remapScalar(row, column, m.tableFor(column)); err != nil {
return err
}
}
case "wiki_log_entries":
// id 是目标本地的 bigint,由既有 LocalIDs 分配;历史日志可能引用已删除文档,
// 这类引用只做"已知才换",不能当作悬挂拒绝(否则整库同步被历史日志卡死)。
m.remapScalarIfKnown(row, "knowledge_id", "knowledges")
if err := m.remapScalar(row, "page_id", "wiki_pages"); err != nil {
return err
}
case "resource_bindings":
if err := m.remapScalar(row, "id", table); err != nil {
return err
}
var ownerType string
if raw, ok := row["owner_type"]; ok {
_ = json.Unmarshal(raw, &ownerType)
}
switch ownerType {
case "knowledge":
if err := m.remapScalar(row, "owner_id", "knowledges"); err != nil {
return err
}
case "chunk":
if err := m.remapScalar(row, "owner_id", "chunks"); err != nil {
return err
}
default:
return fmt.Errorf("resource_bindings: unknown owner_type %q", ownerType)
}
// resource_id 不映射:资源是共享字节,绑定只换归属。
}
return m.rejectUnmapped(table, row)
}
func (m *targetIDMapper) tableFor(column string) string {
switch column {
case "page_id":
return "wiki_pages"
case "knowledge_id":
return "knowledges"
case "chunk_id":
return "chunks"
}
return column
}
func (m *targetIDMapper) remapScalar(row map[string]json.RawMessage, column, table string) error {
raw, ok := row[column]
if !ok {
return nil
}
var value string
if err := json.Unmarshal(raw, &value); err != nil {
return fmt.Errorf("%s: unreadable %s", table, column)
}
if value == "" {
return nil
}
if !m.known[table][value] {
return fmt.Errorf("%w: %s.%s=%s", errUnmappedTargetReference, table, column, value)
}
encoded, _ := json.Marshal(m.mapID(table, value))
row[column] = encoded
return nil
}
func (m *targetIDMapper) remapScalarIfKnown(row map[string]json.RawMessage, column, table string) {
raw, ok := row[column]
if !ok {
return
}
var value string
if err := json.Unmarshal(raw, &value); err != nil || value == "" || !m.known[table][value] {
return
}
encoded, _ := json.Marshal(m.mapID(table, value))
row[column] = encoded
}
func (m *targetIDMapper) remapRefArray(row map[string]json.RawMessage, column, table string) error {
raw, ok := row[column]
if !ok {
return nil
}
var values []string
if err := json.Unmarshal(raw, &values); err != nil {
return nil
}
for i, value := range values {
if !m.known[table][value] {
return fmt.Errorf("%w: %s.%s=%s", errUnmappedTargetReference, table, column, value)
}
values[i] = m.mapID(table, value)
}
encoded, _ := json.Marshal(values)
row[column] = encoded
return nil
}
func (m *targetIDMapper) remapSourceRefs(row map[string]json.RawMessage) error {
raw, ok := row["source_refs"]
if !ok {
return nil
}
var refs []string
if err := json.Unmarshal(raw, &refs); err != nil {
return nil
}
for i, ref := range refs {
id, suffix := ref, ""
if cut := strings.IndexByte(ref, '|'); cut >= 0 {
id, suffix = ref[:cut], ref[cut:]
}
if !m.known["knowledges"][id] {
return fmt.Errorf("%w: wiki_pages.source_refs=%s", errUnmappedTargetReference, id)
}
refs[i] = m.mapID("knowledges", id) + suffix
}
encoded, _ := json.Marshal(refs)
row["source_refs"] = encoded
return nil
}
// remapJSONExact 只在指定 JSON 列里替换"整体等于已知身份"的字符串;
// 不做子串替换(正文里出现同形状文本不是引用)。
func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column string) error {
raw, ok := row[column]
if !ok {
return nil
}
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.UseNumber() // >2^53 的整数必须原样保留,不能经 float64
var value any
if err := decoder.Decode(&value); err != nil {
return fmt.Errorf("chunks.%s 不可解析", column)
}
changed := false
var walk func(any) any
walk = func(v any) any {
switch typed := v.(type) {
case string:
if m.ambiguous[typed] {
return typed // 由调用方在扫描阶段报歧义
}
if table, ok := m.owner[typed]; ok && !m.skip[table] {
changed = true
return m.mapID(table, typed)
}
return typed
case []any:
for i := range typed {
typed[i] = walk(typed[i])
}
return typed
case map[string]any:
for key := range typed {
typed[key] = walk(typed[key])
}
return typed
}
return v
}
mapped := walk(value)
if !changed {
return nil
}
encoded, err := json.Marshal(mapped)
if err != nil {
return err
}
row[column] = encoded
return nil
}
// rejectUnmapped 保证"改写后不再残留任何已知源身份",并拒绝无法判定归属的悬挂引用。
func (m *targetIDMapper) rejectUnmapped(table string, row map[string]json.RawMessage) error {
for id := range m.ambiguous {
if strings.Contains(strings.Join(rawStrings(row), "\x00"), id) {
return fmt.Errorf("%w: ambiguous id %s appears in %s", errUnmappedTargetReference, id, table)
}
}
body := map[string]bool{"content": true, "summary": true, "description": true, "physical_path": true,
"pages_affected": true, "slug": true, "metadata": true, "page_metadata": true, "title": true, "aliases": true, "in_links": true, "out_links": true}
for column, raw := range row {
if body[column] {
continue
}
if leftover := m.findKnown(string(raw)); leftover != "" {
return fmt.Errorf("%w: %s.%s still references source id %s", errUnmappedTargetReference, table, column, leftover)
}
}
return nil
}
func rawStrings(row map[string]json.RawMessage) []string {
out := make([]string, 0, len(row))
for _, raw := range row {
out = append(out, string(raw))
}
return out
}
func (m *targetIDMapper) findKnown(value string) string {
if table, ok := m.owner[value]; ok && !m.skip[table] {
return value
}
// JSON 字符串里嵌着的 id 也要查(只查完整 id 子串,不做模糊匹配)。
for id, table := range m.owner {
if m.skip[table] || len(id) < 16 || !strings.Contains(value, id) {
continue
}
// 允许 it 出现在正文类字段里的情况已由 rejectUnmapped 排除;这里只对
// 明确带引用的字段生效,因此命中即说明漏改。
return id
}
return ""
}
func (m *targetIDMapper) remapIndexDocument(id string, source json.RawMessage) (string, json.RawMessage, error) {
var fields map[string]json.RawMessage
if err := json.Unmarshal(source, &fields); err != nil {
return "", nil, errors.New("index document source is not an object")
}
for column, table := range map[string]string{
"chunk_id": "chunks", "knowledge_id": "knowledges", "tag_id": "knowledge_tags",
} {
raw, ok := fields[column]
if !ok {
continue
}
var value string
if err := json.Unmarshal(raw, &value); err != nil || value == "" {
continue
}
if !m.known[table][value] {
return "", nil, fmt.Errorf("%w: index %s=%s", errUnmappedTargetReference, column, value)
}
encoded, _ := json.Marshal(m.mapID(table, value))
fields[column] = encoded
}
encoded, err := json.Marshal(fields)
if err != nil {
return "", nil, err
}
return m.mapID("index_chunks", id), encoded, nil
}
// targetIDMapperFromArchive 从目标快照里登记源身份,供引用映射与残留检查使用。
func targetIDMapperFromArchive(archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest, tenant uint64, kb string) (*targetIDMapper, error) {
mapper := newTargetIDMapper(tenant, kb, nil)
for _, part := range manifest.Parts {
switch part.Kind {
case "database", "wiki":
reader, err := archive.OpenPart(part)
if err != nil {
return nil, err
}
var rows []map[string]json.RawMessage
err = json.NewDecoder(reader).Decode(&rows)
reader.Close()
if err != nil {
return nil, err
}
for _, row := range rows {
if identity, err := parseRowIdentity(part.Name, row); err == nil {
for _, value := range identity.Values {
mapper.registerSourceID(part.Name, value)
}
}
// 引用值不登记为"实体存在":悬挂引用必须由主键表缺行来暴露。
}
case "index":
// 索引文档的 chunk_id/knowledge_id 同样只是引用,不能反过来证明实体存在。
reader, err := archive.OpenPart(part)
if err != nil {
return nil, err
}
documents, err := librarysnapshot.ReadIndexDocuments(reader)
reader.Close()
if err != nil {
return nil, err
}
_ = documents
}
}
return mapper, nil
}
// sortedSourceIDs 只用于稳定报告/调试,不参与映射。
func (m *targetIDMapper) sortedSourceIDs() []string {
out := make([]string, 0, len(m.ids))
for key := range m.ids {
out = append(out, strings.ReplaceAll(key, "\x00", "/"))
}
sort.Strings(out)
return out
}
@@ -0,0 +1,115 @@
package v071
import (
"context"
"testing"
"knowledge-governance/internal/libraryrelease"
)
// 同一隔离实例里已有旧库占用源快照的原始 UUID;目标换成另一个空知识库。
// 必须 Plan→Apply→Verify 成功、旧库逐字节保留,失败路径恢复目标 before-image。
func TestTargetIDMappingPlanApplyVerifyRestoreIsolated(t *testing.T) {
fx := openWriteSetFixture(t)
seedTargetData(t, fx)
ctx := fx.ctx
// 旧库占用与来源快照相同的身份(真实缺陷:原实现按原 ID 写入会撞主键/被归属检查拒绝)。
oldRows := []string{
`INSERT INTO knowledges VALUES('doc-src',10001,'` + wsOtherKB + `','completed',0,'',NULL)`,
`INSERT INTO chunks VALUES('chunk-src',10001,'` + wsOtherKB + `','doc-src','旧库正文',NULL)`,
`INSERT INTO wiki_pages VALUES('wiki-src',10001,'` + wsOtherKB + `','legacy-page','旧库 wiki',NULL,'published')`,
}
for _, statement := range oldRows {
if _, err := fx.pool.Exec(ctx, statement); err != nil {
t.Fatalf("seed old kb: %v\n%s", err, statement)
}
}
before := universeDump(t, fx.pool)
snapshot := sealSourceSnapshot(t, fx, nil)
target := fx.target(t)
backups := fx.backups(t, target)
current, err := target.Current(ctx)
if err != nil {
t.Fatal(err)
}
backup, err := backups.Prepare(ctx, snapshot, current)
if err != nil {
t.Fatal(err)
}
set, err := target.PlanWriteSet(ctx, libraryrelease.WriteSetPlan{JobID: "target-id-map-job", Target: "internal",
BackupSnapshotID: backup.SnapshotID, TargetSnapshotID: snapshot, Universe: backup.Universe})
if err != nil {
t.Fatalf("plan against a database that already owns the source ids: %v", err)
}
for _, row := range set.Rows {
if row.ID == "doc-src" || row.ID == "chunk-src" || row.ID == "wiki-src" {
t.Fatalf("plan still carries source identity %s for %s", row.ID, row.Table)
}
}
for _, stage := range []libraryrelease.Stage{libraryrelease.StageDatabase, libraryrelease.StageWiki,
libraryrelease.StageResources, libraryrelease.StageIndex} {
if err := target.Apply(ctx, stage, snapshot); err != nil {
t.Fatalf("apply %s: %v", stage, err)
}
}
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshot}); err != nil {
t.Fatalf("verify: %v", err)
}
// 旧库原行必须原样存在(内容与 ID 都不变)。
for _, check := range []struct{ query, want string }{
{`SELECT content FROM chunks WHERE id='chunk-src' AND knowledge_base_id='` + wsOtherKB + `'`, "旧库正文"},
{`SELECT content FROM wiki_pages WHERE id='wiki-src' AND knowledge_base_id='` + wsOtherKB + `'`, "旧库 wiki"},
} {
var got string
if err := fx.pool.QueryRow(ctx, check.query).Scan(&got); err != nil || got != check.want {
t.Fatalf("old kb row changed/lost (%s): %q %v", check.query, got, err)
}
}
// 新目标里必须是映射后的身份,不能出现源 UUID。
var sourceIDs int64
if err := fx.pool.QueryRow(ctx, `SELECT count(*) FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2 AND id IN ('doc-src','chunk-src')`,
wsTargetTenant, wsTargetKB).Scan(&sourceIDs); err != nil {
t.Fatal(err)
}
if sourceIDs != 0 {
t.Fatalf("source ids leaked into the new target: %d", sourceIDs)
}
var mapped int64
if err := fx.pool.QueryRow(ctx, `SELECT count(*) FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2 AND id<>'doc-src'`,
wsTargetTenant, wsTargetKB).Scan(&mapped); err != nil {
t.Fatal(err)
}
if mapped == 0 {
t.Fatal("no mapped knowledge row written")
}
// 第二次同步(全新 target 实例、只靠持久计划)不重复、不冲突。
second := fx.target(t)
if err := second.AdoptWriteSet(set); err != nil {
t.Fatalf("adopt persisted plan: %v", err)
}
if err := second.Apply(ctx, libraryrelease.StageDatabase, snapshot); err != nil {
t.Fatalf("second apply: %v", err)
}
var duplicated int64
if err := fx.pool.QueryRow(ctx, `SELECT count(*) FROM chunks WHERE tenant_id=$1 AND knowledge_base_id=$2`,
wsTargetTenant, wsTargetKB).Scan(&duplicated); err != nil {
t.Fatal(err)
}
if duplicated != 1 {
t.Fatalf("repeat sync duplicated target chunks: %d", duplicated)
}
// 失败恢复:按持久写集合回放目标 before-image,旧库不变、残留为 0。
ref := backup
ref.WriteSet = &set
if err := target.Restore(context.WithoutCancel(ctx), ref); err != nil {
t.Fatalf("restore: %v", err)
}
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: backup.SnapshotID}); err != nil {
t.Fatalf("verify restored before-image: %v", err)
}
after := universeDump(t, fx.pool)
if diff := diffUniverse(before, after); diff != "" {
t.Fatalf("restore did not return to before-image: %s", diff)
}
t.Logf("plan rows=%d index=%d objects=%d; old kb preserved; restore exact", len(set.Rows), len(set.Index), len(set.Objects))
}
@@ -0,0 +1,254 @@
package v071
import (
"encoding/json"
"strings"
"testing"
"knowledge-governance/internal/libraryrelease"
)
// 目标隔离映射的行为契约:同源对象在不同目标 KB 必须得到不同 ID;同一目标
// 跨作业/跨版本稳定;引用字段与身份字段一起换,正文不得被改。
func TestTargetIDMapperIsScopedAndStable(t *testing.T) {
a := newTargetIDMapper(10000, "kb-a", nil)
b := newTargetIDMapper(10000, "kb-b", nil)
first := a.mapID("chunks", "0d25c09c-5af7-4cee-8b6e-a0b129caeb63")
if first == "0d25c09c-5af7-4cee-8b6e-a0b129caeb63" {
t.Fatalf("mapping returned the source id unchanged")
}
if len(first) != 36 || strings.Count(first, "-") != 4 {
t.Fatalf("mapped id is not uuid-shaped: %q", first)
}
if again := a.mapID("chunks", "0d25c09c-5af7-4cee-8b6e-a0b129caeb63"); again != first {
t.Fatalf("mapping is not deterministic: %q != %q", first, again)
}
if other := b.mapID("chunks", "0d25c09c-5af7-4cee-8b6e-a0b129caeb63"); other == first {
t.Fatalf("different target KB produced the same id")
}
if cross := a.mapID("wiki_pages", "0d25c09c-5af7-4cee-8b6e-a0b129caeb63"); cross == first {
t.Fatalf("different tables collided on one id")
}
}
func rawTargetRow(t *testing.T, v any) map[string]json.RawMessage {
t.Helper()
encoded, err := json.Marshal(v)
if err != nil {
t.Fatal(err)
}
var row map[string]json.RawMessage
if err := json.Unmarshal(encoded, &row); err != nil {
t.Fatal(err)
}
return row
}
func TestTargetIDMapperRewritesChunkReferences(t *testing.T) {
m := newTargetIDMapper(10000, "kb-a", nil)
chunk := "11111111-1111-4111-8111-111111111111"
m.registerSourceID("chunks", chunk)
know := "22222222-2222-4222-8222-222222222222"
m.registerSourceID("knowledges", know)
parent := "33333333-3333-4333-8333-333333333333"
m.registerSourceID("chunks", parent)
row := rawTargetRow(t, map[string]any{
"id": chunk, "knowledge_id": know, "parent_chunk_id": parent,
"pre_chunk_id": parent, "next_chunk_id": parent,
"relation_chunks": []string{parent},
"indirect_relation_chunks": []any{map[string]any{"id": parent}},
"metadata": map[string]any{"source_chunk": parent},
"content": "正文里出现 " + parent + " 也不能被改",
})
if err := m.remapRow("chunks", row); err != nil {
t.Fatal(err)
}
if got := jsonString(t, row["id"]); got != m.mapID("chunks", chunk) {
t.Fatalf("chunk id not remapped: %s", got)
}
if got := jsonString(t, row["knowledge_id"]); got != m.mapID("knowledges", know) {
t.Fatalf("knowledge_id not remapped: %s", got)
}
if got := jsonString(t, row["parent_chunk_id"]); got != m.mapID("chunks", parent) {
t.Fatalf("parent_chunk_id not remapped: %s", got)
}
var content string
_ = json.Unmarshal(row["content"], &content)
if content != "正文里出现 "+parent+" 也不能被改" {
t.Fatalf("content was rewritten: %q", content)
}
var rel []string
_ = json.Unmarshal(row["relation_chunks"], &rel)
if rel[0] != m.mapID("chunks", parent) {
t.Fatalf("relation_chunks not remapped: %v", rel)
}
var meta map[string]any
_ = json.Unmarshal(row["metadata"], &meta)
if meta["source_chunk"] != parent {
t.Fatalf("non-reference metadata payload must stay byte-identical: %v", meta)
}
}
func TestTargetIDMapperRewritesWikiRefsPreservingShape(t *testing.T) {
m := newTargetIDMapper(10000, "kb-a", nil)
c1 := "44444444-4444-4444-8444-444444444444"
m.registerSourceID("chunks", c1)
c2 := "55555555-5555-4555-8555-555555555555"
m.registerSourceID("chunks", c2)
k := "66666666-6666-4666-8666-666666666666"
m.registerSourceID("knowledges", k)
m.registerSourceID("wiki_pages", "77777777-7777-4777-8777-777777777777")
m.registerSourceID("wiki_folders", "88888888-8888-4888-8888-888888888888")
row := rawTargetRow(t, map[string]any{
"id": "77777777-7777-4777-8777-777777777777", "folder_id": "88888888-8888-4888-8888-888888888888",
"chunk_refs": []string{c1, c1, c2},
"source_refs": []string{k + "|文档", k},
"content": "body", "title": "t",
})
if err := m.remapRow("wiki_pages", row); err != nil {
t.Fatal(err)
}
var refs []string
_ = json.Unmarshal(row["chunk_refs"], &refs)
if len(refs) != 3 || refs[0] != m.mapID("chunks", c1) || refs[1] != refs[0] || refs[2] != m.mapID("chunks", c2) {
t.Fatalf("chunk_refs shape/order/dupes not preserved: %v", refs)
}
var src []string
_ = json.Unmarshal(row["source_refs"], &src)
if src[0] != m.mapID("knowledges", k)+"|文档" || src[1] != m.mapID("knowledges", k) {
t.Fatalf("source_refs not remapped: %v", src)
}
var content string
_ = json.Unmarshal(row["content"], &content)
if content != "body" {
t.Fatal("wiki body changed")
}
}
func TestTargetIDMapperKeepsSharedResourceIdentity(t *testing.T) {
m := newTargetIDMapper(10000, "kb-a", nil)
res := "99999999-9999-4999-8999-999999999999"
m.registerSourceID("resource_bindings", "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb")
know := "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
m.registerSourceID("knowledges", know)
row := rawTargetRow(t, map[string]any{"id": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
"resource_id": res, "owner_id": know, "owner_type": "knowledge"})
if err := m.remapRow("resource_bindings", row); err != nil {
t.Fatal(err)
}
if got := jsonString(t, row["resource_id"]); got != res {
t.Fatalf("shared resource id must not be remapped: %s", got)
}
if got := jsonString(t, row["owner_id"]); got != m.mapID("knowledges", know) {
t.Fatalf("owner_id not remapped: %s", got)
}
}
func TestTargetIDMapperRejectsUnmappedReference(t *testing.T) {
m := newTargetIDMapper(10000, "kb-a", nil)
m.registerSourceID("chunks", "cccccccc-cccc-4ccc-8ccc-cccccccccccc")
row := rawTargetRow(t, map[string]any{"id": "dddddddd-dddd-4ddd-8ddd-dddddddddddd",
"chunk_refs": []string{"eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"}})
if err := m.remapRow("wiki_pages", row); err == nil {
t.Fatal("unmapped dangling reference was accepted")
}
}
func TestTargetIndexDocumentMapping(t *testing.T) {
m := newTargetIDMapper(10000, "kb-a", nil)
chunk := "12121212-1212-4212-8212-121212121212"
m.registerSourceID("chunks", chunk)
doc := []byte(`{"chunk_id":"` + chunk + `","knowledge_base_id":"old","content":"c"}`)
id, source, err := m.remapIndexDocument("doc-1", doc)
if err != nil {
t.Fatal(err)
}
if id == "doc-1" {
t.Fatal("index document id not remapped")
}
var out map[string]any
if err := json.Unmarshal(source, &out); err != nil {
t.Fatal(err)
}
if out["chunk_id"] != m.mapID("chunks", chunk) {
t.Fatalf("index chunk_id not remapped: %v", out["chunk_id"])
}
if out["content"] != "c" {
t.Fatal("index content changed")
}
}
func TestTargetIDMapperRewritesRelationTuple(t *testing.T) {
m := newTargetIDMapper(10000, "kb-a", nil)
know := "13131313-1313-4131-8131-131313131313"
tag := "14141414-1414-4141-8141-141414141414"
m.registerSourceID("knowledges", know)
m.registerSourceID("knowledge_tags", tag)
row := rawTargetRow(t, map[string]any{"knowledge_id": know, "tag_id": tag})
if err := m.remapRow("knowledge_tag_relations", row); err != nil {
t.Fatal(err)
}
if jsonString(t, row["knowledge_id"]) != m.mapID("knowledges", know) || jsonString(t, row["tag_id"]) != m.mapID("knowledge_tags", tag) {
t.Fatal("relation tuple not remapped")
}
}
func TestTargetMappingOnlyAppliesToTargetSnapshot(t *testing.T) {
plan := &libraryrelease.WriteSet{TargetSnapshotID: "target-seal", BackupSnapshotID: "backup-seal"}
target := &ProductionLibraryTarget{activePlan: plan}
if !target.targetMappingApplies("target-seal") {
t.Fatal("target snapshot must be mapped")
}
if target.targetMappingApplies("backup-seal") {
t.Fatal("backup/before-image must not be mapped")
}
target.preserveChunkSequence = true
if target.targetMappingApplies("target-seal") {
t.Fatal("restore replay must not be mapped")
}
}
func TestTargetIDMapperPreservesLargeNumbersAndRejectsAmbiguity(t *testing.T) {
m := newTargetIDMapper(10000, "kb-a", nil)
chunk := "15151515-1515-4515-8515-151515151515"
m.registerSourceID("chunks", chunk)
// 同列还有 9007199254740993(>2^53):必须逐字节保留。
row := rawTargetRow(t, map[string]any{"id": "16161616-1616-4616-8616-161616161616",
"relation_chunks": json.RawMessage(`["` + chunk + `",9007199254740993]`)})
m.registerSourceID("chunks", "16161616-1616-4616-8616-161616161616")
if err := m.remapRow("chunks", row); err != nil {
t.Fatal(err)
}
if got := string(row["relation_chunks"]); got != `["`+m.mapID("chunks", chunk)+`",9007199254740993]` {
t.Fatalf("relation payload not preserved: %s", got)
}
// 同一字符串 ID 同时出现在两张表:不能 first-table-wins 猜映射。
same := "17171717-1717-4717-8717-171717171717"
m.registerSourceID("chunks", same)
m.registerSourceID("knowledges", same)
bad := rawTargetRow(t, map[string]any{"id": "18181818-1818-4818-8818-181818181818",
"relation_chunks": []string{same}})
if err := m.remapRow("chunks", bad); err == nil {
t.Fatal("ambiguous cross-table id was mapped by guess")
}
}
func TestTargetIDMapperDoesNotInventEntitiesFromReferences(t *testing.T) {
// 引用值不能反向登记为实体:wiki page 的 chunk_refs 指向一个不存在的主键。
m := newTargetIDMapper(10000, "kb-a", nil)
m.registerSourceID("wiki_pages", "19191919-1919-4919-8919-191919191919")
row := rawTargetRow(t, map[string]any{"id": "19191919-1919-4919-8919-191919191919",
"chunk_refs": []string{"20202020-2020-4020-8020-202020202020"}})
if err := m.remapRow("wiki_pages", row); err == nil {
t.Fatal("dangling chunk_ref was accepted")
}
}
func jsonString(t *testing.T, raw json.RawMessage) string {
t.Helper()
var s string
if err := json.Unmarshal(raw, &s); err != nil {
t.Fatalf("not a string: %s", raw)
}
return s
}
@@ -0,0 +1,112 @@
package v071
import (
"encoding/json"
"errors"
"fmt"
"sort"
"knowledge-governance/internal/librarysnapshot"
)
// enableTargetMapping 为目标快照构建一次映射器。源身份集合来自归档本身,
// 因此引用字段的"未登记即悬挂"判断与真实数据一致,不靠猜测。
func (t *ProductionLibraryTarget) enableTargetMapping(archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
if t.scope.TenantID == 0 || t.scope.KnowledgeBaseID == "" {
return errors.New("目标范围身份不完整,拒绝生成目标 ID 映射")
}
mapper, err := targetIDMapperFromArchive(archive, manifest, t.scope.TenantID, t.scope.KnowledgeBaseID)
if err != nil {
return fmt.Errorf("构建目标 ID 映射失败: %w", err)
}
t.targetIDs = mapper
t.mappingActive = true
return nil
}
func (t *ProductionLibraryTarget) disableTargetMapping() {
t.targetIDs = nil
t.mappingActive = false
}
// targetMappingApplies 只在"把目标快照写进目标"时启用映射;
// 备份/恢复快照里的 ID 是目标自己的既有身份,二次映射会把恢复写坏。
func (t *ProductionLibraryTarget) targetMappingApplies(snapshotID string) bool {
if t.preserveChunkSequence || snapshotID == "" {
return false
}
if t.activePlan != nil {
return snapshotID == t.activePlan.TargetSnapshotID
}
return true
}
// archivedRowIdentitiesFor 返回写入后的目标身份集合(映射开启时)或归档原身份。
func (t *ProductionLibraryTarget) archivedRowIdentitiesFor(archive *librarysnapshot.Archive, part librarysnapshot.Part) ([]string, error) {
if !t.mappingActive {
return archivedRowIdentities(archive, part)
}
reader, err := archive.OpenPart(part)
if err != nil {
return nil, err
}
defer reader.Close()
decoder := json.NewDecoder(reader)
token, err := decoder.Token()
if err != nil || token != json.Delim('[') {
return nil, errors.New("归档部件不是行数组")
}
transform := t.rowTransform(part.Name)
ids := make([]string, 0, 64)
for decoder.More() {
var row map[string]json.RawMessage
if err := decoder.Decode(&row); err != nil {
return nil, err
}
if err := transform(row); err != nil {
return nil, err
}
identity, err := parseRowIdentity(part.Name, row)
if err != nil {
return nil, err
}
ids = append(ids, identity.key())
}
return ids, nil
}
// archivedTableState 是 archivedTableState 的目标映射版本:身份摘要按写入后的
// ID 计算,否则"写对了"也会被判成与快照不一致。
func (t *ProductionLibraryTarget) archivedTableState(archive *librarysnapshot.Archive, part librarysnapshot.Part) (int64, string, error) {
if !t.mappingActive {
return archivedTableState(archive, part)
}
ids, err := t.archivedRowIdentitiesFor(archive, part)
if err != nil {
if errors.Is(err, errMissingIdentity) {
return 0, "", err
}
return 0, "", err
}
sort.Strings(ids)
return int64(len(ids)), digestIDs(ids), nil
}
// remapIndexMap 把目标快照索引文档的物理 ID 与引用换成目标身份。
func (t *ProductionLibraryTarget) remapIndexMap(documents map[string]json.RawMessage) (map[string]json.RawMessage, error) {
if !t.mappingActive {
return documents, nil
}
out := make(map[string]json.RawMessage, len(documents))
for id, source := range documents {
mappedID, mappedSource, err := t.targetIDs.remapIndexDocument(id, source)
if err != nil {
return nil, err
}
if _, exists := out[mappedID]; exists {
return nil, errors.New("index target id collision")
}
out[mappedID] = mappedSource
}
return out, nil
}
@@ -0,0 +1,205 @@
package v071
import (
"encoding/json"
"errors"
"os"
"path/filepath"
"sort"
"testing"
"knowledge-governance/internal/librarysnapshot"
)
// 用真实封存做离线转换闭包证明:不依赖人工精简 fixture。
// 只读归档;不写数据库/对象存储,不联网。
func TestRealSealTargetMappingClosure(t *testing.T) {
root := os.Getenv("KG_LIBRARY_ARCHIVE_ROOT")
if root == "" {
root = filepath.Join(os.Getenv("HOME"), ".local/share/knowledge-governance/library")
}
const seal = "2c6db552d3b2119dc8e3c6d7c39cb4993a47396ca5bb4e38cfd05e947126c4a8"
if _, err := os.Stat(filepath.Join(root, "versions", seal+".json")); err != nil {
t.Skip("real seal not present on this host")
}
archive, err := librarysnapshot.Open(root, 256*1024*1024)
if err != nil {
t.Fatalf("open archive: %v", err)
}
defer archive.Close()
manifest, err := archive.Load(seal)
if err != nil {
t.Fatalf("load seal: %v", err)
}
target := &ProductionLibraryTarget{}
target.scope.TenantID = 10000
target.scope.KnowledgeBaseID = "a8e9c8ad-411c-4c53-8421-599d29e9b179"
target.snapshotTenant = manifest.TenantID
target.snapshotKB = manifest.KnowledgeBaseID
if err := target.enableTargetMapping(archive, manifest); err != nil {
t.Fatalf("enable mapping: %v", err)
}
mapper := target.targetIDs
t.Logf("mapping_active=%v known_chunks=%d known_know=%d known_pages=%d", target.mappingActive, len(mapper.known["chunks"]), len(mapper.known["knowledges"]), len(mapper.known["wiki_pages"]))
other := newTargetIDMapper(10000, "kb-other-target", nil)
mappedChunks := map[string]bool{}
mappedKnowledge := map[string]bool{}
counts := map[string]int{}
for _, part := range manifest.Parts {
if part.Kind != "database" && part.Kind != "wiki" {
continue
}
reader, err := archive.OpenPart(part)
if err != nil {
t.Fatal(err)
}
var rows []map[string]json.RawMessage
err = json.NewDecoder(reader).Decode(&rows)
reader.Close()
if err != nil {
t.Fatal(err)
}
transform := target.rowTransform(part.Name)
for _, row := range rows {
var sourceID string
if raw, ok := row["id"]; ok {
_ = json.Unmarshal(raw, &sourceID)
}
if err := transform(row); err != nil {
t.Fatalf("%s row %s: %v", part.Name, sourceID, err)
}
counts[part.Name]++
var mappedID string
if raw, ok := row["id"]; ok {
_ = json.Unmarshal(raw, &mappedID)
}
switch part.Name {
case "chunks":
if mappedID == sourceID {
t.Fatalf("chunk id was not remapped: %s", sourceID)
}
if other.mapID("chunks", sourceID) == mappedID {
t.Fatalf("different target reused mapped chunk id")
}
mappedChunks[mappedID] = true
case "knowledges":
mappedKnowledge[mappedID] = true
case "resources":
if mappedID != sourceID {
t.Fatalf("shared resource id must be preserved: %s -> %s", sourceID, mappedID)
}
case "wiki_log_entries":
// id 是目标本地 bigint,映射器不得改动其类型或值。
var number json.Number
if err := json.Unmarshal(row["id"], &number); err != nil {
t.Fatalf("log id must stay numeric: %s", row["id"])
}
}
}
}
// 引用闭包:所有 chunk_refs / source_refs 必须落在映射后的同库身份里。
for _, part := range manifest.Parts {
if part.Kind != "wiki" || part.Name != "wiki_pages" {
continue
}
reader, _ := archive.OpenPart(part)
var rows []map[string]json.RawMessage
_ = json.NewDecoder(reader).Decode(&rows)
reader.Close()
transform := target.rowTransform("wiki_pages")
for _, row := range rows {
if err := transform(row); err != nil {
t.Fatal(err)
}
var refs, sources []string
_ = json.Unmarshal(row["chunk_refs"], &refs)
_ = json.Unmarshal(row["source_refs"], &sources)
for _, ref := range refs {
if !mappedChunks[ref] {
t.Fatalf("chunk ref %s is not a mapped live chunk", ref)
}
}
for _, ref := range sources {
id := ref
if i := len(ref) - len("|"); i > 0 {
// "id|title" 形式:只比较前段
for j := 0; j < len(ref); j++ {
if ref[j] == '|' {
id = ref[:j]
break
}
}
}
if !mappedKnowledge[id] {
t.Fatalf("source ref %s is not a mapped knowledge", id)
}
}
}
}
// 索引文档:ID/引用都换到目标身份,正文与向量字段不动。
for _, part := range manifest.Parts {
if part.Kind != "index" {
continue
}
reader, _ := archive.OpenPart(part)
docs, err := librarysnapshot.ReadIndexDocuments(reader)
reader.Close()
if err != nil {
t.Fatal(err)
}
for _, document := range docs {
mappedID, source, err := mapper.remapIndexDocument(document.ID, document.Source)
if err != nil {
t.Fatal(err)
}
if mappedID == document.ID {
t.Fatal("index doc id not remapped")
}
var fields map[string]any
if err := json.Unmarshal(source, &fields); err != nil {
t.Fatal(err)
}
chunkID, _ := fields["chunk_id"].(string)
if !mappedChunks[chunkID] {
t.Fatalf("index chunk_id %s not in mapped chunk set", chunkID)
}
}
}
// 写集合身份必须用映射后的 ID:同一目标重复计算结果一致,且与源身份不同。
for _, part := range manifest.Parts {
if part.Kind != "database" && part.Kind != "wiki" {
continue
}
if part.Name != "chunks" && part.Name != "knowledges" && part.Name != "wiki_pages" {
continue
}
first, err := target.archivedRowIdentitiesFor(archive, part)
if err != nil {
t.Fatalf("%s identities: %v", part.Name, err)
}
second, err := target.archivedRowIdentitiesFor(archive, part)
if err != nil {
t.Fatal(err)
}
sort.Strings(first)
sort.Strings(second)
if len(first) != len(second) {
t.Fatalf("%s identity count changed between runs", part.Name)
}
for i := range first {
if first[i] != second[i] {
t.Fatalf("%s identity is not deterministic", part.Name)
}
}
}
want := map[string]int{"knowledges": 4, "chunks": 38, "wiki_pages": 31, "wiki_folders": 21, "wiki_log_entries": 11}
for table, n := range want {
if counts[table] != n {
t.Fatalf("%s rows=%d want %d", table, counts[table], n)
}
}
if _, err := os.Stat(filepath.Join(root, "versions", seal+".json")); err != nil {
t.Fatal(errors.New("archive changed during test"))
}
t.Logf("mapped rows=%v chunks=%d knowledges=%d", counts, len(mappedChunks), len(mappedKnowledge))
}
@@ -232,6 +232,15 @@ func (t *ProductionLibraryTarget) verifyIndexDigest(ctx context.Context, archive
if err != nil {
return err
}
if t.mappingActive && t.targetIDs != nil {
for i := range documents {
mappedID, mappedSource, err := t.targetIDs.remapIndexDocument(documents[i].ID, documents[i].Source)
if err != nil {
return err
}
documents[i] = librarysnapshot.IndexDocument{ID: mappedID, Source: mappedSource}
}
}
archived := make([]SearchDocument, 0, len(documents))
for _, document := range documents {
// 写入时 BulK Index 会把 tenant_id/knowledge_base_id 覆盖成本目标的身份,
@@ -116,6 +116,12 @@ func (t *ProductionLibraryTarget) PlanWriteSet(ctx context.Context, plan library
if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil {
return libraryrelease.WriteSet{}, err
}
// 新计划必须从计划阶段就使用目标隔离身份:后面的归属检查、索引计划、
// 写集合登记与 Apply/Verify 都要看到同一套 ID。
if err := t.enableTargetMapping(archive, manifest); err != nil {
return libraryrelease.WriteSet{}, err
}
t.targetIDMode = libraryrelease.TargetIDModeScoped
if err := validateArchivedLibraryClosure(archive, manifest); err != nil {
return libraryrelease.WriteSet{}, err
}
@@ -146,12 +152,12 @@ func (t *ProductionLibraryTarget) PlanWriteSet(ctx context.Context, plan library
SourceTenantID: manifest.TenantID, SourceKnowledgeBaseID: manifest.KnowledgeBaseID, JobID: plan.JobID, Target: plan.Target,
TenantID: t.scope.TenantID, KnowledgeBaseID: t.scope.KnowledgeBaseID,
BackupSnapshotID: plan.BackupSnapshotID, TargetSnapshotID: plan.TargetSnapshotID,
CreatedAt: t.now().UTC(), Universe: plan.Universe}
CreatedAt: t.now().UTC(), Universe: plan.Universe, TargetIDMode: libraryrelease.TargetIDModeScoped}
for _, part := range manifest.Parts {
if part.Kind != "database" && part.Kind != "wiki" {
continue
}
ids, err := archivedRowIdentities(archive, part)
ids, err := t.archivedRowIdentitiesFor(archive, part)
if err != nil {
return libraryrelease.WriteSet{}, err
}