diff --git a/CHANGELOG.md b/CHANGELOG.md index 82fac25..4c03b72 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/internal/releaseadapter/v071/library_index_guard.go b/internal/releaseadapter/v071/library_index_guard.go index 7f2366c..11139ad 100644 --- a/internal/releaseadapter/v071/library_index_guard.go +++ b/internal/releaseadapter/v071/library_index_guard.go @@ -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 } diff --git a/internal/releaseadapter/v071/library_index_preflight_integration_test.go b/internal/releaseadapter/v071/library_index_preflight_integration_test.go new file mode 100644 index 0000000..be31a97 --- /dev/null +++ b/internal/releaseadapter/v071/library_index_preflight_integration_test.go @@ -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) + } +}