fix(library): 目标隔离映射兼容旧计划与真实回归
build / build (push) Successful in 1m41s

空 mode 计划不映射;索引 source_id 按语义映射;relation 按字段表解析。
隔离回归按持久映射 ID 断言,DS02 故障注入真实命中。
This commit is contained in:
姚章浩
2026-11-04 12:40:12 +08:00
parent e0fc2c32b8
commit 0d9091e464
15 changed files with 428 additions and 91 deletions
@@ -93,16 +93,20 @@ func TestR10IncomingRelationClosureIsolated(t *testing.T) {
if result.State != "succeeded" {
t.Fatalf("nonempty tuple publication: %+v", result)
}
wantID := "doc-src:tag-src"
wantID := expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-src") + ":" +
expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-src")
if mode == "valid-existing" {
wantID = "doc-target:tag-target"
wantID = expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-target") + ":" +
expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-target")
}
found := false
for _, row := range result.Backup.WriteSet.Rows {
if row.Table == "knowledge_tag_relations" && row.ID == wantID {
found = true
if row.Existing != (mode == "valid-existing") {
t.Fatal("tuple existence misclassified", row)
// 映射后所有 tuple 都是目标新身份;即使源 ID 恰是目标库已存在的
// doc-target/tag-target,它们会被派生成本次的新 ID,写前不存在。
if row.Existing {
t.Fatal("mapped tuple misclassified as pre-existing", row)
}
}
}
@@ -171,13 +171,17 @@ func TestR10SequenceOrderingReplayIsolated(t *testing.T) {
if e := fx.pool.QueryRow(fx.ctx, `SELECT array_agg(id ORDER BY chunk_index ASC) FROM chunks WHERE knowledge_base_id='kb-target'`).Scan(&chunkOrder); e != nil {
t.Fatal(e)
}
if !reflect.DeepEqual(tagOrder, []string{"tag-src", "tag-src-2"}) || !reflect.DeepEqual(chunkOrder, []string{"chunk-src", "chunk-src-2"}) {
wantTags := []string{expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-src"),
expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-src-2")}
wantChunks := []string{expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src"),
expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src-2")}
if !reflect.DeepEqual(tagOrder, wantTags) || !reflect.DeepEqual(chunkOrder, wantChunks) {
t.Fatal("application sort changed after numeric remapping", tagOrder, chunkOrder)
}
for _, m := range result.Backup.WriteSet.LocalIDs {
if m.Table == "chunks" {
var id string
if e := fx.pool.QueryRow(fx.ctx, `SELECT id FROM chunks WHERE tenant_id=$1 AND seq_id=$2::bigint`, wsTargetTenant, m.Target).Scan(&id); e != nil || (m.Source == "100" && id != "chunk-src") || (m.Source == "2" && id != "chunk-src-2") {
if e := fx.pool.QueryRow(fx.ctx, `SELECT id FROM chunks WHERE tenant_id=$1 AND seq_id=$2::bigint`, wsTargetTenant, m.Target).Scan(&id); e != nil || (m.Source == "100" && id != expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src")) || (m.Source == "2" && id != expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src-2")) {
t.Fatal("seq-ID consumer disagrees with frozen map", e)
}
}
@@ -221,7 +221,7 @@ func r11Source(t *testing.T, fx *writeSetFixture, tweak func(map[string][]map[st
}
})
}
func r11Reject(t *testing.T, fx *writeSetFixture, snapshot string) {
func r11RejectNatural(t *testing.T, fx *writeSetFixture, snapshot string) {
t.Helper()
before := r10State(t, fx)
target := fx.target(t)
@@ -232,6 +232,60 @@ func r11Reject(t *testing.T, fx *writeSetFixture, snapshot string) {
}
r10Unchanged(t, fx, before)
}
func r11Reject(t *testing.T, fx *writeSetFixture, snapshot, table, sourceID string) {
t.Helper()
before := r10State(t, fx)
var beforeRow string
if e := fx.pool.QueryRow(fx.ctx, `SELECT to_jsonb(x)::text FROM `+table+` AS x WHERE id=$1`, sourceID).Scan(&beforeRow); e != nil {
t.Fatalf("read foreign row before publish: %v", e)
}
beforeState := map[string]string{"row:" + table: beforeRow}
target := fx.target(t)
out := libraryrelease.Run(fx.ctx, requestFor(snapshot), target, fx.backups(t, target), nil)
r10Save(t, "outcome", out)
if table == "resources" {
// 资源是共享字节,不参与 UUID 映射:同 ID 属于他库必须继续拒绝。
if out.State != "failed" || out.FailedAt != "" {
t.Errorf("shared resource id collision must be rejected pre-stage: %+v", out)
}
r10Unchanged(t, fx, before)
return
}
if out.State != "succeeded" {
t.Fatalf("mapped identity publish should succeed: %+v", out)
}
// 旧库原行必须保留,且计划里不得再出现源 ID。
for _, row := range out.Backup.WriteSet.Rows {
if row.Table == table && row.ID == sourceID {
t.Fatalf("writeset still carries source identity %s/%s", table, sourceID)
}
}
mapped := expectedTargetID(wsTargetTenant, wsTargetKB, table, sourceID)
var foreign, targetRows int
if e := fx.pool.QueryRow(fx.ctx, `SELECT count(*) FROM `+table+` WHERE id=$1`, sourceID).Scan(&foreign); e != nil || foreign != 1 {
t.Fatalf("foreign row %s/%s changed or lost: %d %v", table, sourceID, foreign, e)
}
// 旧库行必须逐列字节不变(不只是"还在")。
beforeRow, afterRow := "", ""
if e := fx.pool.QueryRow(fx.ctx, `SELECT to_jsonb(x)::text FROM `+table+` AS x WHERE id=$1`, sourceID).Scan(&afterRow); e != nil {
t.Fatalf("read foreign row %s: %v", table, e)
}
beforeRow = beforeState["row:"+table]
if beforeRow != "" && beforeRow != afterRow {
t.Fatalf("foreign row %s/%s content changed", table, sourceID)
}
if e := fx.pool.QueryRow(fx.ctx, `SELECT count(*) FROM `+table+` WHERE id=$1`, mapped).Scan(&targetRows); e != nil || targetRows != 1 {
t.Fatalf("mapped row %s/%s missing: %d %v", table, mapped, targetRows, e)
}
if tablesWithKB[table] {
var tenant int64
var kb string
if e := fx.pool.QueryRow(fx.ctx, `SELECT tenant_id, knowledge_base_id FROM `+table+` WHERE id=$1`, mapped).Scan(&tenant, &kb); e != nil || tenant != int64(wsTargetTenant) || kb != wsTargetKB {
t.Fatalf("mapped row %s landed at %d/%s", table, tenant, kb)
}
}
}
func TestR11StableIdentityMatrixIsolated(t *testing.T) {
for _, table := range []string{"knowledges", "chunks", "knowledge_tags", "resources", "resource_bindings", "wiki_folders", "wiki_pages", "wiki_page_issues"} {
for _, foreign := range []string{"same-tenant-other-kb", "other-tenant"} {
@@ -250,7 +304,6 @@ func TestR11StableIdentityMatrixIsolated(t *testing.T) {
switch table {
case "chunks":
row["knowledge_id"] = "doc-other"
row["seq_id"] = int64(900000001)
case "resources":
delete(row, "knowledge_base_id")
row["handle"] = strings.Repeat("x", 22)
@@ -268,7 +321,7 @@ func TestR11StableIdentityMatrixIsolated(t *testing.T) {
}
r11Insert(t, fx, table, row)
snapshot := r11Source(t, fx, nil)
r11Reject(t, fx, snapshot)
r11Reject(t, fx, snapshot, table, id)
})
}
}
@@ -328,7 +381,7 @@ func TestR11NaturalKeyMatrixIsolated(t *testing.T) {
p[table] = append(p[table], cp)
})
if mode != "valid" && !strings.HasSuffix(mode, "positive") {
r11Reject(t, fx, snapshot)
r11RejectNatural(t, fx, snapshot)
return
}
before := r10State(t, fx)
@@ -97,7 +97,7 @@ func TestR11CatalogBindingIsolated(t *testing.T) {
"trigger": `CREATE FUNCTION r11_noop() RETURNS trigger LANGUAGE plpgsql AS 'BEGIN RETURN NEW; END';CREATE TRIGGER r11_noop BEFORE INSERT ON wiki_pages FOR EACH ROW EXECUTE FUNCTION r11_noop()`,
}
r10Exec(t, fx, statements[mode])
r11Reject(t, fx, snapshot)
r11RejectNatural(t, fx, snapshot)
// An identically extended source/target must not make an unreviewed FK safe.
if strings.Contains(mode, "fk") {
now, _, e := target.migrationCatalog(fx.ctx)
@@ -157,7 +157,7 @@ func TestR11ReferenceAndCursorIsolated(t *testing.T) {
}
})
if mode != "valid" && mode != "linked-valid" && mode != "legacy-pages" && mode != "historical-log" {
r11Reject(t, fx, snapshot)
r11RejectNatural(t, fx, snapshot)
return
}
target := fx.target(t)
@@ -23,10 +23,10 @@ func r12ESDoc(t *testing.T, fx *writeSetFixture, id string) map[string]any {
return recoveryIndexReadback(t, fx)[id]
}
func r12PutDoc(t *testing.T, fx *writeSetFixture, id string, tenant uint64, kb, knowledge, content string, embedding []float64) {
func r12PutDoc(t *testing.T, fx *writeSetFixture, id, chunkID string, tenant uint64, kb, knowledge, content string, embedding []float64) {
t.Helper()
body, err := json.Marshal(map[string]any{"tenant_id": tenant, "knowledge_base_id": kb,
"knowledge_id": knowledge, "chunk_id": id, "source_id": id, "content": content, "embedding": embedding})
"knowledge_id": knowledge, "chunk_id": chunkID, "source_id": chunkID, "content": content, "embedding": embedding})
if err != nil {
t.Fatal(err)
}
@@ -158,7 +158,9 @@ func TestR12IndexOracleDetectsContentVectorAndMetadataChangesIsolated(t *testing
if err := target.Verify(fx.ctx, expect); err != nil {
t.Fatalf("baseline oracle must pass: %v", err)
}
published := r12ESDoc(t, fx, "chunk-src")
docID := expectedIndexID("chunk-src")
chunkID := expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src")
published := r12ESDoc(t, fx, docID)
if published == nil {
t.Fatal("published document missing")
}
@@ -168,29 +170,29 @@ func TestR12IndexOracleDetectsContentVectorAndMetadataChangesIsolated(t *testing
originalKnowledge, _ := source["knowledge_id"].(string)
// 1) 正文被改。
r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, "被篡改的正文", []float64{0.1, 0.2})
r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, "被篡改的正文", []float64{0.1, 0.2})
if err := target.Verify(fx.ctx, expect); err == nil {
t.Fatal("正文篡改没有让独立 oracle 变红")
}
r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2})
r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2})
if err := target.Verify(fx.ctx, expect); err != nil {
t.Fatalf("恢复正文后 oracle 必须通过:%v", err)
}
// 2) 向量被改(同维不同值)。
r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.9, 0.8})
r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.9, 0.8})
if err := target.Verify(fx.ctx, expect); err == nil {
t.Fatal("向量篡改没有让独立 oracle 变红")
}
r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2})
r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2})
if err := target.Verify(fx.ctx, expect); err != nil {
t.Fatalf("恢复向量后 oracle 必须通过:%v", err)
}
// 3) metadata(knowledge_id)被改。
r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, "doc-target", originalContent, []float64{0.1, 0.2})
r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, "doc-target", originalContent, []float64{0.1, 0.2})
if err := target.Verify(fx.ctx, expect); err == nil {
t.Fatal("metadata 篡改没有让独立 oracle 变红")
}
r12PutDoc(t, fx, "chunk-src", wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2})
r12PutDoc(t, fx, docID, chunkID, wsTargetTenant, wsTargetKB, originalKnowledge, originalContent, []float64{0.1, 0.2})
if err := target.Verify(fx.ctx, expect); err != nil {
t.Fatalf("恢复 metadata 后 oracle 必须通过:%v", err)
}
@@ -903,7 +903,8 @@ func ds02Check(t *testing.T, fx *writeSetFixture, spec ds02Spec, expected map[st
diffs = append(diffs, fmt.Sprintf("index docs: want %d got %d", len(spec.rows["chunks"])+1, len(docs)))
}
for _, chunk := range spec.rows["chunks"] {
id := fmt.Sprint(chunk["id"])
sourceID := fmt.Sprint(chunk["id"])
id := expectedIndexID(sourceID)
doc, ok := docs[id]
if !ok {
diffs = append(diffs, "index doc missing: "+id)
@@ -911,8 +912,11 @@ func ds02Check(t *testing.T, fx *writeSetFixture, spec ds02Spec, expected map[st
}
source := doc
want := map[string]any{"tenant_id": json.Number(strconv.FormatUint(wsTargetTenant, 10)),
"knowledge_base_id": wsTargetKB, "knowledge_id": chunk["knowledge_id"], "chunk_id": id,
"source_id": id, "content": chunk["content"], "embedding": spec.vectors[id]}
"knowledge_base_id": wsTargetKB,
"knowledge_id": expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", fmt.Sprint(chunk["knowledge_id"])),
"chunk_id": expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", sourceID),
"source_id": expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", sourceID),
"content": chunk["content"], "embedding": spec.vectors[sourceID]}
for key, value := range want {
if fmt.Sprint(source[key]) != fmt.Sprint(value) {
diffs = append(diffs, fmt.Sprintf("index doc %s.%s: want %v got %v", id, key, value, source[key]))
@@ -971,6 +975,9 @@ func ds02Check(t *testing.T, fx *writeSetFixture, spec ds02Spec, expected map[st
}
copied["tenant_id"] = json.Number(strconv.FormatUint(wsTargetTenant, 10))
copied["knowledge_base_id"] = wsTargetKB
if knowledge, ok := copied["knowledge_id"].(string); ok && knowledge != "" {
copied["knowledge_id"] = expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", knowledge)
}
raw, _ := json.Marshal(copied)
wantTuples = append(wantTuples, string(raw))
}
@@ -1103,6 +1110,18 @@ func TestDS02FullShapeOracleIsolated(t *testing.T) {
spec := ds02SourceSpec(t, fx)
expectation := ds02Expected(t, fx, spec)
expected, paths := expectation.Rows, expectation.Paths
// 独立 oracle:只对"来源快照将写进目标范围"的期望行按 scoped-sha256-v1
// 推导目标身份;随后追加的同租户/外库保留行仍是原身份,不能映射。
for table := range expected {
for i, row := range expected[table] {
expected[table][i] = expectedTargetRow(table, row)
}
}
for table, bySeq := range expectation.RowsBySource {
for seq, row := range bySeq {
bySeq[seq] = expectedTargetRow(table, row)
}
}
for _, table := range []string{"resources", "resource_bindings"} {
expected[table] = append(expected[table], ds02SameTenantRows(t, fx, table)...)
}
@@ -1112,6 +1131,7 @@ func TestDS02FullShapeOracleIsolated(t *testing.T) {
ds02Read(t, fx, `SELECT to_jsonb(x)::text FROM knowledge_tag_relations AS x
WHERE knowledge_id NOT IN (SELECT id FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2)`,
wsTargetTenant, wsTargetKB)...)
sequencesBefore := ds02SequenceSummary(t, fx)
snapshot := ds02Seal(t, fx, spec)
recorder, inventory := ds02AttachEgress(t, fx)
@@ -1282,8 +1302,28 @@ func TestDS02FullShapeOracleIsolated(t *testing.T) {
foreignPath := wsLocator("backend-target", "minio", fx.bucket, 10001, "res-other")
foreignDoc := `{"tenant_id":20000,"knowledge_base_id":"kb-foreign","knowledge_id":"doc-foreign","chunk_id":"chunk-foreign","source_id":"chunk-foreign","content":"外部正文","embedding":[0.1,0.2]}`
foreignDocTampered := `{"tenant_id":20000,"knowledge_base_id":"kb-foreign","knowledge_id":"doc-foreign","chunk_id":"chunk-foreign","source_id":"chunk-foreign","content":"外部正文篡改","embedding":[0.1,0.2]}`
chunkDoc := `{"tenant_id":10001,"knowledge_base_id":"kb-target","knowledge_id":"doc-src","chunk_id":"chunk-src","source_id":"chunk-src","content":"来源正文一 resource://` + wsHandleSource + `","embedding":[0.5,1.5]}`
chunkDocTampered := `{"tenant_id":10001,"knowledge_base_id":"kb-target","knowledge_id":"doc-src","chunk_id":"chunk-src","source_id":"chunk-src","content":"来源正文一 resource://` + wsHandleSource + `","embedding":[9.5,9.5]}`
mappedDoc := expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-src")
mappedTag := expectedTargetID(wsTargetTenant, wsTargetKB, "knowledge_tags", "tag-src")
mappedChunk := expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src")
mappedIndex := expectedIndexID("chunk-src")
chunkJSON := func(embedding string) string {
return fmt.Sprintf(`{"tenant_id":%d,"knowledge_base_id":%q,"knowledge_id":%q,"chunk_id":%q,"source_id":%q,"content":%q,"embedding":%s}`,
wsTargetTenant, wsTargetKB, mappedDoc, mappedChunk, mappedChunk, content, embedding)
}
chunkDoc, chunkDocTampered := chunkJSON("[0.5,1.5]"), chunkJSON("[9.5,9.5]")
var originalChunk string
if err := fx.pool.QueryRow(fx.ctx, `SELECT content FROM chunks WHERE id=$1`, mappedChunk).Scan(&originalChunk); err != nil {
t.Fatalf("read published chunk content: %v", err)
}
execOne := func(sql string, args ...any) {
tag, err := fx.pool.Exec(fx.ctx, sql, args...)
if err != nil {
t.Fatalf("mutation sql: %v", err)
}
if tag.RowsAffected() != 1 {
t.Fatalf("mutation sql affected %d rows, want 1", tag.RowsAffected())
}
}
mutations := []struct {
name string
marker string
@@ -1291,27 +1331,25 @@ func TestDS02FullShapeOracleIsolated(t *testing.T) {
restore func()
}{
{"content", "chunks:", func() {
r10Exec(t, fx, `UPDATE chunks SET content=content||'!MUT' WHERE id='chunk-src'`)
execOne(`UPDATE chunks SET content=content||'!MUT' WHERE id=$1`, mappedChunk)
}, func() {
if _, err := fx.pool.Exec(fx.ctx, `UPDATE chunks SET content=$1 WHERE id='chunk-src'`, content); err != nil {
t.Fatal(err)
}
execOne(`UPDATE chunks SET content=$1 WHERE id=$2`, originalChunk, mappedChunk)
}},
{"vector", "index doc chunk-src.embedding", func() {
putDoc("chunk-src", chunkDocTampered)
{"vector", "index doc " + expectedIndexID("chunk-src") + ".embedding", func() {
putDoc(mappedIndex, chunkDocTampered)
}, func() {
putDoc("chunk-src", chunkDoc)
putDoc(mappedIndex, chunkDoc)
}},
{"reference", "reference relation", func() {
r10Exec(t, fx, `UPDATE knowledge_tag_relations SET tag_id='missing-tag' WHERE knowledge_id='doc-src' AND tag_id='tag-src'`)
execOne(`UPDATE knowledge_tag_relations SET tag_id='missing-tag' WHERE knowledge_id=$1 AND tag_id=$2`, mappedDoc, mappedTag)
}, func() {
r10Exec(t, fx, `UPDATE knowledge_tag_relations SET tag_id='tag-src' WHERE knowledge_id='doc-src' AND tag_id='missing-tag'`)
execOne(`UPDATE knowledge_tag_relations SET tag_id=$1 WHERE knowledge_id=$2 AND tag_id='missing-tag'`, mappedTag, mappedDoc)
}},
// 这是日志内容变异,不是本地号映射变异:映射 RED 只在账本语义副本上做。
{"log-content", "wiki log", func() {
r10Exec(t, fx, `UPDATE wiki_log_entries SET action='tampered' WHERE tenant_id=10001 AND knowledge_base_id='kb-target' AND knowledge_id='doc-src'`)
execOne(`UPDATE wiki_log_entries SET action='tampered' WHERE tenant_id=10001 AND knowledge_base_id='kb-target' AND knowledge_id=$1`, mappedDoc)
}, func() {
r10Exec(t, fx, `UPDATE wiki_log_entries SET action='ingest' WHERE tenant_id=10001 AND knowledge_base_id='kb-target' AND knowledge_id='doc-src'`)
execOne(`UPDATE wiki_log_entries SET action='ingest' WHERE tenant_id=10001 AND knowledge_base_id='kb-target' AND knowledge_id=$1`, mappedDoc)
}},
{"foreign-index-source", "foreign index doc _source changed", func() {
putDoc("chunk-foreign", foreignDocTampered)
@@ -25,7 +25,7 @@ func TestLibraryChunkSequenceIsolated(t *testing.T) {
t.Fatalf("cross-database publish must allocate local sequence: %+v", outcome)
}
var allocated int64
if err := fx.pool.QueryRow(fx.ctx, `SELECT seq_id FROM chunks WHERE id='chunk-src'`).Scan(&allocated); err != nil {
if err := fx.pool.QueryRow(fx.ctx, `SELECT seq_id FROM chunks WHERE id=$1`, expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src")).Scan(&allocated); err != nil {
t.Fatal(err)
}
if allocated == occupied || allocated <= 0 {
@@ -577,7 +577,7 @@ func TestContractKnownIndexRecoveryAfterDeleteCompensationFaultIsolated(t *testi
var base wireCounts
target.restoreMarkerFault = func(c libraryrelease.IndexWrite) error {
if c.ID != "chunk-src" {
if c.ID != expectedIndexID("chunk-src") {
return nil
}
base = wire.counts()
@@ -588,7 +588,7 @@ func TestContractKnownIndexRecoveryAfterDeleteCompensationFaultIsolated(t *testi
t.Fatalf("state = %q (rollback=%q error=%q), want rollback_failed", outcome.State, outcome.Rollback, outcome.Error)
}
plan := outcome.Backup.WriteSet
change := indexChangeOf(t, plan, "chunk-src")
change := indexChangeOf(t, plan, expectedIndexID("chunk-src"))
if len(change.Before) != 0 || len(change.After) == 0 {
t.Fatalf("fixture change is not a created document: %+v", change)
}
@@ -597,7 +597,7 @@ func TestContractKnownIndexRecoveryAfterDeleteCompensationFaultIsolated(t *testi
if refused := wire.since(base); refused.deletes != 0 || refused.clean != 0 || refused.creates != 0 {
t.Fatalf("boundary fault dispatched %+v; refusal must send nothing", refused)
}
if _, ok := recoveryIndexReadback(t, fx)["chunk-src"]; !ok {
if _, ok := recoveryIndexReadback(t, fx)[expectedIndexID("chunk-src")]; !ok {
t.Fatal("precondition: the created document must still be present after the refused delete")
}
@@ -186,17 +186,21 @@ func TestLibraryProductionTargetIsolated(t *testing.T) {
if err := target.Restore(ctx, secondBackup); err != nil {
t.Fatalf("restore: %v", err)
}
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: first}); err != nil {
t.Fatalf("verify after restore: %v", err)
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: secondBackup.SnapshotID}); err != nil {
t.Fatalf("verify after restore to second job before-image: %v", err)
}
// Freeze repair evidence before fault injection, never from damaged state.
backup := prepare(first)
// 故障注入:直接破坏生产状态,校验必须失败,恢复后必须通过。
if _, err := pool.Exec(ctx, `DELETE FROM chunks WHERE id='chunk-safe'`); err != nil {
deleted, err := pool.Exec(ctx, `DELETE FROM chunks WHERE id=$1`, expectedTargetID(10000, "kb-safe", "chunks", "chunk-safe"))
if err != nil {
t.Fatal(err)
}
if deleted.RowsAffected() != 1 {
t.Fatalf("fault injection deleted %d rows, want 1", deleted.RowsAffected())
}
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: first}); err == nil {
t.Fatal("verify must detect a missing chunk row")
}
@@ -650,14 +654,15 @@ func TestLibraryCrossScopeSyncRollbackLeavesNoResidueIsolated(t *testing.T) {
}
var knowledgeKB, chunkKB, provider, locator string
var tenant int64
if err := pool.QueryRow(ctx, `SELECT tenant_id, knowledge_base_id FROM knowledges WHERE id='doc-source'`).
if err := pool.QueryRow(ctx, `SELECT tenant_id, knowledge_base_id FROM knowledges WHERE id=$1`,
expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-source")).
Scan(&tenant, &knowledgeKB); err != nil {
t.Fatalf("source document must exist in the target scope: %v", err)
}
if tenant != 10001 || knowledgeKB != "kb-target" {
t.Fatalf("cross-scope row landed at %d/%s", tenant, knowledgeKB)
}
if err := pool.QueryRow(ctx, `SELECT knowledge_base_id FROM chunks WHERE id='chunk-source'`).Scan(&chunkKB); err != nil {
if err := pool.QueryRow(ctx, `SELECT knowledge_base_id FROM chunks WHERE id=$1`, expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-source")).Scan(&chunkKB); err != nil {
t.Fatal(err)
}
if chunkKB != "kb-target" {
@@ -28,14 +28,13 @@ import (
// 资源表(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
tenant uint64
kb string
ids map[string]string
reverse map[string]string
known map[string]map[string]bool
owner map[string]string
sequence int
// 不重映射的表:资源行是跨库共享对象,改了 ID 就等于复制一份字节,
// 现有复用契约要求"内容相同 + 来源可审计"才复用,不能靠换 ID 绕过。
@@ -50,7 +49,7 @@ 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{},
known: map[string]map[string]bool{}, owner: map[string]string{},
skip: map[string]bool{"resources": true},
}
}
@@ -85,19 +84,17 @@ func (m *targetIDMapper) registerSourceID(table, id string) {
}
// 本地自增主键(如 wiki_log_entries.id)不是跨库身份:既不能当引用匹配,
// 也不能参与"残留源 ID"判断,否则普通整数列会被误报。
if contract, ok := migrationIdentityContracts[table]; ok && contract.Kind == "local-bigint" {
if contract, ok := migrationIdentityContracts[table]; ok && (contract.Kind == "local-bigint" || contract.Kind == "tuple") {
// 本地自增/复合元组都不是可被别的表引用的实体身份:
// 关系表的两个分量必须由 knowledges/knowledge_tags 自己登记。
return
}
if m.known[table] == nil {
m.known[table] = map[string]bool{}
}
m.known[table][id] = true
if previous, exists := m.owner[id]; !exists {
if _, exists := m.owner[id]; !exists {
m.owner[id] = table
} else if previous != table {
// 两张表出现同一字符串 ID:无法从值本身判定引用指向哪张表,
// 只能报歧义,不能 first-table-wins 猜一个。
m.ambiguous[id] = true
}
}
@@ -127,7 +124,9 @@ func (m *targetIDMapper) remapRow(table string, row map[string]json.RawMessage)
}
}
for _, column := range []string{"relation_chunks", "indirect_relation_chunks"} {
if err := m.remapJSONExact(row, column); err != nil {
// 这两列是"分块引用":目标表由字段语义决定(chunks),不由字符串
// 在别的表里也出现来决定;跨表同 ID 是合法数据,不是歧义。
if err := m.remapJSONExact(row, column, "chunks"); err != nil {
return err
}
}
@@ -166,6 +165,9 @@ func (m *targetIDMapper) remapRow(table string, row map[string]json.RawMessage)
if err := m.remapScalar(row, "id", table); err != nil {
return err
}
if err := m.remapRefArray(row, "suspected_knowledge_ids", "knowledges"); 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
@@ -294,7 +296,7 @@ func (m *targetIDMapper) remapSourceRefs(row map[string]json.RawMessage) error {
// remapJSONExact 只在指定 JSON 列里替换"整体等于已知身份"的字符串;
// 不做子串替换(正文里出现同形状文本不是引用)。
func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column string) error {
func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column, table string) error {
raw, ok := row[column]
if !ok {
return nil
@@ -303,35 +305,51 @@ func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column s
decoder.UseNumber() // >2^53 的整数必须原样保留,不能经 float64
var value any
if err := decoder.Decode(&value); err != nil {
return fmt.Errorf("chunks.%s 不可解析", column)
return fmt.Errorf("%s.%s 不可解析", table, column)
}
changed := false
var walk func(any) any
walk = func(v any) any {
var walk func(any) (any, error)
walk = func(v any) (any, error) {
switch typed := v.(type) {
case string:
if m.ambiguous[typed] {
return typed // 由调用方在扫描阶段报歧义
}
if table, ok := m.owner[typed]; ok && !m.skip[table] {
if m.known[table][typed] {
changed = true
return m.mapID(table, typed)
return m.mapID(table, typed), nil
}
return typed
for other, ids := range m.known {
if other == table || m.skip[other] {
continue
}
if ids[typed] {
return nil, fmt.Errorf("%w: %s.%s references %s entity %s", errUnmappedTargetReference, table, column, other, typed)
}
}
return typed, nil
case []any:
for i := range typed {
typed[i] = walk(typed[i])
mapped, err := walk(typed[i])
if err != nil {
return nil, err
}
typed[i] = mapped
}
return typed
return typed, nil
case map[string]any:
for key := range typed {
typed[key] = walk(typed[key])
mapped, err := walk(typed[key])
if err != nil {
return nil, err
}
typed[key] = mapped
}
return typed
return typed, nil
}
return v
return v, nil
}
mapped, err := walk(value)
if err != nil {
return err
}
mapped := walk(value)
if !changed {
return nil
}
@@ -345,11 +363,6 @@ func (m *targetIDMapper) remapJSONExact(row map[string]json.RawMessage, column s
// 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 {
@@ -409,6 +422,21 @@ func (m *targetIDMapper) remapIndexDocument(id string, source json.RawMessage) (
encoded, _ := json.Marshal(m.mapID(table, value))
fields[column] = encoded
}
// source_id 在不同索引器版本里既可能是 chunk id,也可能是外部文档 id:
// 只有能确定是分块/文档身份时才映射,opaque 外部 id 原样保留。
if raw, ok := fields["source_id"]; ok {
var value string
if json.Unmarshal(raw, &value) == nil && value != "" {
switch {
case m.known["chunks"][value]:
encoded, _ := json.Marshal(m.mapID("chunks", value))
fields["source_id"] = encoded
case m.known["knowledges"][value]:
encoded, _ := json.Marshal(m.mapID("knowledges", value))
fields["source_id"] = encoded
}
}
}
encoded, err := json.Marshal(fields)
if err != nil {
return "", nil, err
@@ -2,8 +2,11 @@ package v071
import (
"context"
"strings"
"testing"
"knowledge-governance/internal/librarysnapshot"
"knowledge-governance/internal/libraryrelease"
)
@@ -113,3 +116,56 @@ func TestTargetIDMappingPlanApplyVerifyRestoreIsolated(t *testing.T) {
}
t.Logf("plan rows=%d index=%d objects=%d; old kb preserved; restore exact", len(set.Rows), len(set.Index), len(set.Objects))
}
// 映射后的目标 ID 若真被另一个库占用,仍必须拒绝(不能只取消旧冲突预期)。
func TestTargetIDMappedIdentityCollisionStillRejectedIsolated(t *testing.T) {
fx := openWriteSetFixture(t)
seedTargetData(t, fx)
mapped := expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", "doc-src")
if _, err := fx.pool.Exec(fx.ctx, `INSERT INTO knowledges VALUES($1,10001,$2,'completed',0,'',NULL)`, mapped, wsOtherKB); err != nil {
t.Fatal(err)
}
snapshot := sealSourceSnapshot(t, fx, nil)
target := fx.target(t)
current, err := target.Current(fx.ctx)
if err != nil {
t.Fatal(err)
}
backup, err := fx.backups(t, target).Prepare(fx.ctx, snapshot, current)
if err != nil {
t.Fatal(err)
}
if _, err := target.PlanWriteSet(fx.ctx, libraryrelease.WriteSetPlan{JobID: "mapped-collision", Target: "internal",
BackupSnapshotID: backup.SnapshotID, TargetSnapshotID: snapshot, Universe: backup.Universe}); err == nil {
t.Fatal("mapped identity occupied by another KB was accepted")
}
}
// 旧空 mode 计划没有映射契约:真实写入路径必须按原 ID 落库并因此与旧库冲突,
// 证明新进程不会给历史计划隐式重映射。
func TestLegacyPlanWithoutModeDoesNotRemapIsolated(t *testing.T) {
fx := openWriteSetFixture(t)
seedTargetData(t, fx)
if _, err := fx.pool.Exec(fx.ctx, `INSERT INTO knowledges VALUES('doc-src',10001,$1,'completed',0,'',NULL)`, wsOtherKB); err != nil {
t.Fatal(err)
}
snapshot := sealSourceSnapshot(t, fx, nil)
target := fx.target(t)
archive, err := librarysnapshot.Open(fx.archiveRoot, 256*1024*1024)
if err != nil {
t.Fatal(err)
}
defer archive.Close()
manifest, err := archive.Load(snapshot)
if err != nil {
t.Fatal(err)
}
target.disableTargetMapping()
if err := target.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil {
t.Fatal(err)
}
err = target.importRows(context.Background(), archive, manifest, "database")
if err == nil || !strings.Contains(strings.ToLower(err.Error()), "duplicate") && !strings.Contains(err.Error(), "冲突") {
t.Fatalf("legacy path must write original ids and fail on the old-kb collision, got %v", err)
}
}
@@ -195,7 +195,7 @@ func TestTargetIDMapperRewritesRelationTuple(t *testing.T) {
func TestTargetMappingOnlyAppliesToTargetSnapshot(t *testing.T) {
plan := &libraryrelease.WriteSet{TargetSnapshotID: "target-seal", BackupSnapshotID: "backup-seal"}
target := &ProductionLibraryTarget{activePlan: plan}
target := &ProductionLibraryTarget{activePlan: plan, targetIDMode: libraryrelease.TargetIDModeScoped}
if !target.targetMappingApplies("target-seal") {
t.Fatal("target snapshot must be mapped")
}
@@ -206,6 +206,11 @@ func TestTargetMappingOnlyAppliesToTargetSnapshot(t *testing.T) {
if target.targetMappingApplies("target-seal") {
t.Fatal("restore replay must not be mapped")
}
target.preserveChunkSequence = false
target.targetIDMode = ""
if target.targetMappingApplies("target-seal") {
t.Fatal("legacy plan without target id mode must not be mapped")
}
}
func TestTargetIDMapperPreservesLargeNumbersAndRejectsAmbiguity(t *testing.T) {
@@ -226,10 +231,14 @@ func TestTargetIDMapperPreservesLargeNumbersAndRejectsAmbiguity(t *testing.T) {
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",
m.registerSourceID("chunks", "18181818-1818-4818-8818-181818181818")
shared := 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")
if err := m.remapRow("chunks", shared); err != nil {
t.Fatalf("typed chunk reference rejected for cross-table same id: %v", err)
}
if got := string(shared["relation_chunks"]); got != `["`+m.mapID("chunks", same)+`"]` {
t.Fatalf("relation must resolve by field semantics: %s", got)
}
}
@@ -6,6 +6,8 @@ import (
"fmt"
"sort"
"knowledge-governance/internal/libraryrelease"
"knowledge-governance/internal/librarysnapshot"
)
@@ -32,13 +34,15 @@ func (t *ProductionLibraryTarget) disableTargetMapping() {
// targetMappingApplies 只在"把目标快照写进目标"时启用映射;
// 备份/恢复快照里的 ID 是目标自己的既有身份,二次映射会把恢复写坏。
func (t *ProductionLibraryTarget) targetMappingApplies(snapshotID string) bool {
if t.preserveChunkSequence || snapshotID == "" {
// 只有持久计划显式声明 scoped 契约时才映射。空 mode 是历史计划:
// 它登记的就是原身份,重映射会把恢复/重试写到别的 ID 上。
if t.targetIDMode != libraryrelease.TargetIDModeScoped || t.preserveChunkSequence || snapshotID == "" {
return false
}
if t.activePlan != nil {
return snapshotID == t.activePlan.TargetSnapshotID
if t.activePlan == nil {
return false
}
return true
return snapshotID == t.activePlan.TargetSnapshotID
}
// archivedRowIdentitiesFor 返回写入后的目标身份集合(映射开启时)或归档原身份。
@@ -261,6 +261,9 @@ func (t *ProductionLibraryTarget) verifyIndexDigest(ctx context.Context, archive
}
archived = append(archived, SearchDocument{ID: document.ID, Source: canonical})
}
// 目标 ID 映射会改变文档 ID 的排序位置;摘要按 ID 排序后的集合计算,
// 否则少一个/多一个的差异会被误报成"数量一致但摘要不同"。
sort.Slice(archived, func(i, j int) bool { return archived[i].ID < archived[j].ID })
wantDigest, err := searchDocumentsDigest(archived)
if err != nil {
return err
@@ -0,0 +1,131 @@
package v071
import (
"crypto/sha256"
"fmt"
"strings"
"testing"
)
// 测试侧独立实现的目标 ID 推导:公式来自产品文档(scoped-sha256-v1),
// 不复用生产 mapper/rowTransform,避免"用实现证明实现"。
func expectedTargetID(tenant uint64, kb, table, source string) string {
sum := sha256.Sum256([]byte("knowledge-governance/target-identity/v1\x00" +
fmt.Sprintf("%d", tenant) + "\x00" + kb + "\x00" + table + "\x00" + source))
var raw [16]byte
copy(raw[:], sum[:16])
raw[6] = raw[6]&0x0f | 0x50
raw[8] = raw[8]&0x3f | 0x80
return fmt.Sprintf("%x-%x-%x-%x-%x", raw[0:4], raw[4:6], raw[6:8], raw[8:10], raw[10:16])
}
func expectedIndexID(source string) string {
return expectedTargetID(wsTargetTenant, wsTargetKB, "index_chunks", source)
}
// expectedTargetRow 按已知引用字段语义改写一行期望数据,供独立 oracle 使用。
func expectedTargetRow(table string, row map[string]any) map[string]any {
out := make(map[string]any, len(row))
for k, v := range row {
out[k] = v
}
str := func(key string) (string, bool) { s, ok := out[key].(string); return s, ok }
mapped := func(key, targetTable string) {
if s, ok := str(key); ok && s != "" {
out[key] = expectedTargetID(wsTargetTenant, wsTargetKB, targetTable, s)
}
}
switch table {
case "knowledges":
mapped("id", "knowledges")
case "chunks":
mapped("id", "chunks")
mapped("knowledge_id", "knowledges")
mapped("tag_id", "knowledge_tags")
for _, key := range []string{"parent_chunk_id", "pre_chunk_id", "next_chunk_id"} {
mapped(key, "chunks")
}
for _, key := range []string{"relation_chunks", "indirect_relation_chunks"} {
if list, ok := out[key].([]any); ok {
for i, item := range list {
if s, ok := item.(string); ok {
list[i] = expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", s)
}
}
}
}
case "knowledge_tags":
mapped("id", "knowledge_tags")
case "knowledge_tag_relations":
mapped("knowledge_id", "knowledges")
mapped("tag_id", "knowledge_tags")
case "wiki_folders":
mapped("id", "wiki_folders")
mapped("parent_id", "wiki_folders")
case "wiki_pages":
mapped("id", "wiki_pages")
mapped("folder_id", "wiki_folders")
if list, ok := out["chunk_refs"].([]any); ok {
for i, item := range list {
if s, ok := item.(string); ok {
list[i] = expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", s)
}
}
}
if list, ok := out["source_refs"].([]any); ok {
for i, item := range list {
if s, ok := item.(string); ok {
id, suffix := s, ""
if cut := strings.IndexByte(s, '|'); cut >= 0 {
id, suffix = s[:cut], s[cut:]
}
list[i] = expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", id) + suffix
}
}
}
case "wiki_page_issues":
mapped("id", "wiki_page_issues")
mapped("knowledge_id", "knowledges")
mapped("page_id", "wiki_pages")
if list, ok := out["suspected_knowledge_ids"].([]any); ok {
for i, item := range list {
if s, ok := item.(string); ok {
list[i] = expectedTargetID(wsTargetTenant, wsTargetKB, "knowledges", s)
}
}
}
case "wiki_log_entries":
mapped("knowledge_id", "knowledges")
mapped("page_id", "wiki_pages")
case "resource_bindings":
mapped("id", "resource_bindings")
if ownerType, _ := out["owner_type"].(string); ownerType == "knowledge" {
mapped("owner_id", "knowledges")
} else if ownerType == "chunk" {
mapped("owner_id", "chunks")
}
}
return out
}
// expectedIndexSource 按语义改写索引文档里的引用(正文/向量不动)。
func expectedIndexSource(t *testing.T, source map[string]any) map[string]any {
t.Helper()
out := make(map[string]any, len(source))
for k, v := range source {
out[k] = v
}
for key, targetTable := range map[string]string{"chunk_id": "chunks", "knowledge_id": "knowledges", "tag_id": "knowledge_tags"} {
if s, ok := out[key].(string); ok && s != "" {
out[key] = expectedTargetID(wsTargetTenant, wsTargetKB, targetTable, s)
}
}
if s, ok := out["tenant_id"]; ok {
if number, ok := s.(float64); ok && number == float64(wsSourceTenant) {
out["tenant_id"] = float64(wsTargetTenant)
}
}
out["tenant_id"] = float64(wsTargetTenant)
out["knowledge_base_id"] = wsTargetKB
return out
}