fix(index): 预检使用目标映射 ID

PlanWriteSet 的 index preflight 改用与 plan/Apply 一致的目标物理 _id。保留 tenant/KB 归属、完整 _mget 与并发版本校验；legacy 路径不映射。负例：映射后 ID 被占、legacy 源 ID 冲突仍拒绝。
2026-11-04 15:12:08 +08:00
parent 537a6df46d
commit 92684c6f60
3 changed files with 223 additions and 0 deletions
@@ -1,5 +1,15 @@
 # Changelog

+## 开发快照 0.1.351（索引预检使用目标映射 ID）
+
+### Fixed
+
+- **目标库索引预检不再误报旧库占用源 `_id`**：`PlanWriteSet` 的早期 index preflight 与
+  plan/Apply 统一使用 `scoped-sha256-v1` 映射后的目标物理 ID；旧库占用源索引 ID 不再阻断
+  首次同步。仍逐 ID 校验目标 tenant/KB 归属、完整 `_mget` 响应与并发版本。
+- **负例保持**：映射后的目标 ID 真被其他 KB 占用仍拒绝；legacy/no-mapping 路径继续按原
+  `_id` 校验并拒绝冲突。资源不映射、共享资源行/字节保护不变。
+
 ## 开发快照 0.1.350（首次复用取消本库历史来源门槛）

 ### Changed
@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"sort"

 	"knowledge-governance/internal/librarysnapshot"
 )
@@ -90,6 +91,30 @@ func (t *ProductionLibraryTarget) preflightArchiveIndex(ctx context.Context, arc
 		if err != nil {
 			return err
 		}
+		if t.mappingActive {
+			// Plan/Apply 的索引身份已在 planIndex/ApplyIndex 统一映射；前置预检
+			// 必须对同一目标物理 _id 做归属检查，否则旧库占用源 _id 会被误报冲突。
+			mapped := make(map[string]json.RawMessage, len(docs))
+			for _, doc := range docs {
+				id, source, err := t.targetIDs.remapIndexDocument(doc.ID, doc.Source)
+				if err != nil {
+					return err
+				}
+				if _, exists := mapped[id]; exists {
+					return errors.New("index target id collision")
+				}
+				mapped[id] = source
+			}
+			ids := make([]string, 0, len(mapped))
+			for id := range mapped {
+				ids = append(ids, id)
+			}
+			sort.Strings(ids)
+			docs = make([]librarysnapshot.IndexDocument, 0, len(ids))
+			for _, id := range ids {
+				docs = append(docs, librarysnapshot.IndexDocument{ID: id, Source: mapped[id]})
+			}
+		}
 		if _, err := t.search.indexWriteGuards(ctx, docs, SearchIdentity{TenantID: t.scope.TenantID, KnowledgeBaseID: t.scope.KnowledgeBaseID}); err != nil {
 			return err
 		}
@@ -0,0 +1,188 @@
+package v071
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"testing"
+
+	"knowledge-governance/internal/libraryrelease"
+	"knowledge-governance/internal/librarysnapshot"
+)
+
+type esDocReadback struct {
+	ID     string `json:"_id"`
+	Found  bool   `json:"found"`
+	Source struct {
+		Tenant  uint64 `json:"tenant_id"`
+		KB      string `json:"knowledge_base_id"`
+		Content string `json:"content"`
+		ChunkID string `json:"chunk_id"`
+	} `json:"_source"`
+}
+
+func (fx *writeSetFixture) getIndexDoc(t *testing.T, id string) esDocReadback {
+	t.Helper()
+	request, err := http.NewRequestWithContext(fx.ctx, "GET", fmt.Sprintf("%s/%s/_doc/%s", fx.searchBase, fx.index, id), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	response, err := fx.client.Do(request)
+	if err != nil {
+		t.Fatalf("get index %s: %v", id, err)
+	}
+	defer response.Body.Close()
+	var doc esDocReadback
+	if response.StatusCode == 404 {
+		return doc // found=false
+	}
+	if response.StatusCode != 200 {
+		t.Fatalf("get index %s => %d", id, response.StatusCode)
+	}
+	if err := json.NewDecoder(response.Body).Decode(&doc); err != nil {
+		t.Fatal(err)
+	}
+	return doc
+}
+
+// 目标库为空、同 tenant 旧库占用源 DB 主键且占用源 ES 物理 _id：
+// PlanWriteSet 的早期 index preflight 必须用映射后的目标 _id，而不是把源 _id
+// 当成本目标写入预检；同时旧库 ES 文档/DB 行必须原样保留。
+func TestIndexPreflightUsesMappedTargetIDIsolated(t *testing.T) {
+	fx := openWriteSetFixture(t)
+	seedTargetData(t, fx)
+	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 resources VALUES('res-src','` + wsHandleSource + `',10001,'` + wsLocator("backend-target", "minio", wsBucket, 10001, "res-src") + `',NULL,'minio',8,'backend-target')`,
+		`INSERT INTO resource_bindings VALUES('bind-old-src',10001,'res-src','knowledge','doc-src',NULL)`,
+	}
+	for _, statement := range oldRows {
+		if _, err := fx.pool.Exec(fx.ctx, statement); err != nil {
+			t.Fatalf("seed old db: %v\n%s", err, statement)
+		}
+	}
+	// 旧库在共享索引里真实占用源物理 _id（同 tenant，不同 KB）。
+	fx.indexDoc(t, "chunk-src", wsTargetTenant, wsOtherKB, "doc-src", "旧库索引正文")
+	if err := fx.objects.s3.Put(fx.ctx, "minio", wsLocator("backend-target", "minio", wsBucket, wsTargetTenant, "res-src"),
+		strings.NewReader("12345678"), 8); err != nil {
+		t.Fatal(err)
+	}
+	oldIndex := fx.getIndexDoc(t, "chunk-src")
+	if !oldIndex.Found || oldIndex.Source.KB != wsOtherKB {
+		t.Fatalf("old index fixture missing: %+v", oldIndex)
+	}
+	beforeObj, err := fx.objects.s3.ObserveObject(fx.ctx, "minio", wsLocator("backend-target", "minio", wsBucket, wsTargetTenant, "res-src"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	snapshot := sealSourceSnapshot(t, fx, nil)
+	target := fx.target(t)
+	backups := fx.backups(t, target)
+	current, err := target.Current(fx.ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	backup, err := backups.Prepare(fx.ctx, snapshot, current)
+	if err != nil {
+		t.Fatal(err)
+	}
+	set, err := target.PlanWriteSet(fx.ctx, libraryrelease.WriteSetPlan{JobID: "index-preflight-map",
+		Target: "internal", BackupSnapshotID: backup.SnapshotID, TargetSnapshotID: snapshot, Universe: backup.Universe})
+	if err != nil {
+		t.Fatalf("index preflight must use mapped target id, got: %v", err)
+	}
+	mapped := expectedIndexID("chunk-src")
+	for _, row := range set.Index {
+		if row.ID == "chunk-src" {
+			t.Fatal("write set still carries source index id")
+		}
+		if row.ID == mapped {
+			// ok
+		}
+	}
+	// 计划持久化后由新实例重新 adopt，再走真实 Apply/Verify。
+	next := fx.target(t)
+	if err := next.AdoptWriteSet(set); err != nil {
+		t.Fatalf("adopt persisted plan: %v", err)
+	}
+	for _, stage := range []libraryrelease.Stage{libraryrelease.StageDatabase, libraryrelease.StageWiki,
+		libraryrelease.StageResources, libraryrelease.StageIndex} {
+		if err := next.Apply(fx.ctx, stage, snapshot); err != nil {
+			t.Fatalf("apply %s: %v", stage, err)
+		}
+	}
+	if err := next.Verify(fx.ctx, libraryrelease.Expectation{SnapshotID: snapshot}); err != nil {
+		t.Fatalf("verify: %v", err)
+	}
+
+	oldAfter := fx.getIndexDoc(t, "chunk-src")
+	if !oldAfter.Found || oldAfter.Source.KB != wsOtherKB || oldAfter.Source.Content != "旧库索引正文" || oldAfter.Source.ChunkID != "chunk-src" {
+		t.Fatalf("old KB index document changed: %+v", oldAfter)
+	}
+	var oldDB int64
+	if err := fx.pool.QueryRow(fx.ctx, `SELECT count(*) FROM chunks WHERE id='chunk-src' AND knowledge_base_id=$1 AND content='旧库正文'`, wsOtherKB).Scan(&oldDB); err != nil || oldDB != 1 {
+		t.Fatalf("old DB row changed: %d %v", oldDB, err)
+	}
+	newDoc := fx.getIndexDoc(t, mapped)
+	if !newDoc.Found || newDoc.Source.Tenant != wsTargetTenant || newDoc.Source.KB != wsTargetKB ||
+		!strings.Contains(newDoc.Source.Content, wsHandleTarget) && !strings.Contains(newDoc.Source.Content, "来源正文") {
+		t.Fatalf("mapped target index doc missing/wrong: %+v", newDoc)
+	}
+	if newDoc.Source.ChunkID != expectedTargetID(wsTargetTenant, wsTargetKB, "chunks", "chunk-src") {
+		t.Fatalf("mapped doc chunk_id is not target id: %s", newDoc.Source.ChunkID)
+	}
+	afterObj, err := fx.objects.s3.ObserveObject(fx.ctx, "minio", wsLocator("backend-target", "minio", wsBucket, wsTargetTenant, "res-src"))
+	if err != nil || beforeObj.Exists != afterObj.Exists || beforeObj.SHA256 != afterObj.SHA256 || beforeObj.Bytes != afterObj.Bytes {
+		t.Fatalf("shared resource object changed: %+v -> %+v %v", beforeObj, afterObj, err)
+	}
+}
+
+func TestIndexPreflightRejectsMappedIDOwnedByOtherKBIsolated(t *testing.T) {
+	fx := openWriteSetFixture(t)
+	seedTargetData(t, fx)
+	mapped := expectedIndexID("chunk-src")
+	fx.indexDoc(t, mapped, wsTargetTenant, wsOtherKB, "chunk-src", "别的库占用了映射后的物理 id")
+	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)
+	}
+	_, err = target.PlanWriteSet(fx.ctx, libraryrelease.WriteSetPlan{JobID: "mapped-index-collision", Target: "internal",
+		BackupSnapshotID: backup.SnapshotID, TargetSnapshotID: snapshot, Universe: backup.Universe})
+	if err == nil || !strings.Contains(err.Error(), mapped) || !strings.Contains(err.Error(), "not owned") {
+		t.Fatalf("mapped index id occupied by another KB must be rejected on that id, got %v", err)
+	}
+}
+
+func TestLegacyIndexPreflightStillRejectsSourceIDConflictIsolated(t *testing.T) {
+	fx := openWriteSetFixture(t)
+	seedTargetData(t, fx)
+	fx.indexDoc(t, "chunk-src", wsTargetTenant, wsOtherKB, "doc-src", "旧库源 id")
+	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.preflightArchiveIndex(fx.ctx, archive, manifest)
+	if err == nil || !strings.Contains(err.Error(), "chunk-src") || !strings.Contains(err.Error(), "not owned") {
+		t.Fatalf("legacy/no-mapping preflight must still reject source id conflict, got %v", err)
+	}
+}