From 5defe238e0b992ab4d0037f63d04606d9e16cd2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A7=9A=E7=AB=A0=E6=B5=A9?= Date: Thu, 29 Oct 2026 22:11:59 +0800 Subject: [PATCH] =?UTF-8?q?fix(library):=20=E8=B7=A8=E8=8C=83=E5=9B=B4?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E6=8C=89=E7=9B=AE=E6=A0=87=E8=8C=83=E5=9B=B4?= =?UTF-8?q?=E6=94=B9=E5=86=99=E5=B9=B6=E6=A0=A1=E9=AA=8C=E6=AE=8B=E7=95=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 真实事故：源 KB(10000/0c84a098) 同步到目标 KB(10001/e64e82b5) 时，归档行被原样 插入目标库，落在来源租户/知识库；资源阶段随即报"生产缺少资源 ... 的存储路径" （资源查询按目标租户过滤，看不见刚写进去的行）。回滚只按目标范围回放，于是页面 报"已恢复到预备份"，目标却多出 4 文件/38 分块/21 资源/4 绑定。 改： - 写入与校验共用 retargetRow：tenant_id / knowledge_base_id / 资源定位符(后端、 provider、桶、租户前缀) / storage_backend_id / provider / embedding_model_id 统一映射到目标范围；行里的来源身份对不上就失败，宁可不写。 - Verify 增加"本次写集合仍以来源身份残留"判定；Restore 增加整库指纹比对（预备份 时刻 vs 恢复后）与孤儿资源清理，回滚后用 Verify 复核，恢复工具退出 0 不再等于 恢复完成。 - 新增 cmd/libraryresidue：按归档身份精确、可备份、可复核地清理已知的范围外残留。 验证：make verify 全绿；真实 PG+ES 用例复现"源目标 KB 不同 + 目标非空 + 资源阶段 失败"，断言回滚后无来源范围残留、其它租户行未动、恢复后内容=预备份；正向用例断言 行落在目标范围且资源定位符指向目标存储。 --- cmd/libraryresidue/main.go | 115 ++++++ internal/libraryrelease/release.go | 21 +- internal/libraryrelease/release_test.go | 58 ++- .../v071/library_backup_store.go | 15 +- .../v071/library_production_target.go | 183 +++++++--- ...rary_production_target_integration_test.go | 339 ++++++++++++++++++ .../releaseadapter/v071/library_residue.go | 334 +++++++++++++++++ .../v071/library_scope_remap.go | 262 ++++++++++++++ .../v071/library_scope_remap_test.go | 194 ++++++++++ .../releaseadapter/v071/library_universe.go | 126 +++++++ .../v071/resource_locator_test.go | 11 +- 11 files changed, 1588 insertions(+), 70 deletions(-) create mode 100644 cmd/libraryresidue/main.go create mode 100644 internal/releaseadapter/v071/library_residue.go create mode 100644 internal/releaseadapter/v071/library_scope_remap.go create mode 100644 internal/releaseadapter/v071/library_scope_remap_test.go create mode 100644 internal/releaseadapter/v071/library_universe.go diff --git a/cmd/libraryresidue/main.go b/cmd/libraryresidue/main.go new file mode 100644 index 0000000..164aaaa --- /dev/null +++ b/cmd/libraryresidue/main.go @@ -0,0 +1,115 @@ +// command libraryresidue 处理"整库同步写到目标范围外"留下的残留行： +// +// 读指定快照（来源范围）→ 在目标里找同内容、仍带来源身份的行 +// → 先备份成 0600 文件并记录摘要 → 精确按归档身份删除 → 复核来源身份已空、 +// 目标内容与预备份一致、整库指纹里没有别的租户 +// +// 它不做范围广删，也不改任何别的东西：判据是"归档里确实有这些行、内容逐字段 +// 一致"，任何一行对不上就整体拒绝。默认是干跑（只报告、不删）。 +// +// 真实事故：跨范围同步把 4 文件/38 分块/21 资源/4 绑定写成来源租户的行， +// 回滚只按目标范围回放，于是"已恢复"却留下这些行。这个工具的用途就是在 +// 修复后的代码上线前，用可审计的方式清掉它们。 +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "log" + "os" + "strconv" + "strings" + + "knowledge-governance/internal/appconfig" + "knowledge-governance/internal/libraryrelease" + "knowledge-governance/internal/librarysnapshot" + "knowledge-governance/internal/librarystack" + v071 "knowledge-governance/internal/releaseadapter/v071" +) + +func main() { + scope := flag.String("scope", "internal", "internal|external") + snapshot := flag.String("snapshot", "", "sealed snapshot whose source scope is the residue (required)") + backupSnapshot := flag.String("backup-snapshot", "", "prebackup snapshot the target must be restored to (required)") + expectSource := flag.String("expect-source", "", "expected source scope as / (required guard against typos)") + out := flag.String("out", "", "directory for the residue backup (required)") + apply := flag.Bool("apply", false, "actually delete the residue rows (default: dry run)") + reportPath := flag.String("report", "", "write the JSON report here") + flag.Parse() + if *snapshot == "" || *backupSnapshot == "" || *out == "" || *expectSource == "" { + log.Fatal("snapshot, backup-snapshot, out and expect-source are required") + } + config, err := appconfig.Load(os.Getenv) + if err != nil { + log.Fatalf("load config: %v", err) + } + ctx := context.Background() + report, err := run(ctx, config, *scope, *snapshot, *backupSnapshot, *expectSource, *out, *apply) + if *reportPath != "" && report != nil { + raw, marshalErr := json.MarshalIndent(report, "", " ") + if marshalErr == nil { + raw = append(raw, '\n') + if writeErr := os.WriteFile(*reportPath, raw, 0o600); writeErr != nil { + log.Printf("report write failed: %v", writeErr) + } + } + } + if err != nil { + log.Fatalf("residue cleanup failed: %v", err) + } + log.Printf("residue cleanup ok: source=%s/%s tables=%v cleared=%d applied=%v verified=%v", + reportSource(report), report.SourceKB, report.Tables, report.Cleared, report.Applied, report.Verified) +} + +func reportSource(report *v071.ResidueReport) string { + return strconv.FormatUint(report.SourceTenant, 10) +} + +func run(ctx context.Context, config appconfig.Config, scope, snapshotID, backupSnapshotID, expectSource, out string, + apply bool) (*v071.ResidueReport, error) { + if scope != "internal" && scope != "external" { + return nil, fmt.Errorf("scope must be internal or external") + } + archive, err := librarysnapshot.Open(config.LibraryArchiveRoot, 256*1024*1024) + if err != nil { + return nil, fmt.Errorf("归档不可用: %w", err) + } + manifest, err := archive.Load(snapshotID) + archive.Close() + if err != nil { + return nil, fmt.Errorf("指定快照不可读: %w", err) + } + source := v071.ProductionScope{TenantID: manifest.TenantID, KnowledgeBaseID: manifest.KnowledgeBaseID} + want := strconv.FormatUint(source.TenantID, 10) + "/" + source.KnowledgeBaseID + // 范围写错会把"别人的行"当成残留：必须由调用方显式确认。 + if strings.TrimSpace(expectSource) != want { + return nil, fmt.Errorf("来源范围与 -expect-source 不一致：归档为 %s，参数为 %s", want, expectSource) + } + target, _, err := librarystack.OpenProductionStack(ctx, config, scope, backupSnapshotID, "") + if err != nil { + return nil, fmt.Errorf("装配目标失败: %w", err) + } + defer target.Close() + report, err := target.ClearForeignScope(ctx, source, snapshotID, out, apply) + if err != nil { + return report, err + } + report.Foreign, _ = target.ForeignScopes(ctx) + if !apply { + return report, nil + } + if err := target.VerifyResidueCleared(ctx, source); err != nil { + return report, err + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: backupSnapshotID}); err != nil { + return report, fmt.Errorf("清理后目标内容仍与预备份不一致（保持停写）: %w", err) + } + report.Verified = true + if len(report.Foreign) > 0 { + raw, _ := json.Marshal(report.Foreign) + return report, fmt.Errorf("清理后目标仍有其它租户的数据，需人工确认才能开放：%s", raw) + } + return report, nil +} diff --git a/internal/libraryrelease/release.go b/internal/libraryrelease/release.go index 2f9bcc0..3322bf2 100644 --- a/internal/libraryrelease/release.go +++ b/internal/libraryrelease/release.go @@ -59,6 +59,14 @@ type Expectation struct { Digests map[string]string } +// TableFingerprint 是一张知识域表在**整实例**上的行数与身份集合摘要（不限租户/ +// 知识库）。恢复后必须与预备份时刻逐表一致：只比目标范围会让"写到范围外"的 +// 残留被当成已恢复（真实事故：恢复报成功，目标仍多 67 行）。 +type TableFingerprint struct { + Count int64 `json:"count"` + Digest string `json:"digest"` +} + // BackupRef 指向一份可独立恢复的完整预备份。 type BackupRef struct { ID string `json:"id"` @@ -67,6 +75,9 @@ type BackupRef struct { SnapshotID string `json:"snapshot_id"` Bytes int64 `json:"bytes"` CreatedAt time.Time `json:"created_at"` + // Universe 是预备份时刻的整实例指纹，恢复后逐表比对。空表示这次没有采集 + // （例如 worker 被杀后用已持久化的快照接管），此时恢复只能靠内容校验兜底。 + Universe map[string]TableFingerprint `json:"universe,omitempty"` } type Target interface { @@ -224,8 +235,16 @@ func failAndRestore(ctx context.Context, outcome Outcome, target Target, recorde recorder.Stage(ctx, "rollback", "failed", outcome.Rollback) return outcome } + // "恢复工具跑完"不是"恢复完成"：必须再校验一次目标内容与预备份一致， + // 否则页面会把半恢复状态报成已恢复（真实事故就是这样放开了维护）。 + if verifyErr := target.Verify(ctx, Expectation{SnapshotID: outcome.Backup.SnapshotID}); verifyErr != nil { + outcome.State = "rollback_failed" + outcome.Rollback = "恢复后校验失败（目标可能仍是半写状态）：" + verifyErr.Error() + recorder.Stage(ctx, "rollback", "failed", outcome.Rollback) + return outcome + } outcome.State = "rolled_back" - outcome.Rollback = "已恢复到预备份 " + outcome.Backup.ID + outcome.Rollback = "已恢复到预备份 " + outcome.Backup.ID + "，并逐表/索引/资源复核一致" recorder.Stage(ctx, "rollback", "succeeded", outcome.Rollback) return outcome } diff --git a/internal/libraryrelease/release_test.go b/internal/libraryrelease/release_test.go index 3a513e8..824bcef 100644 --- a/internal/libraryrelease/release_test.go +++ b/internal/libraryrelease/release_test.go @@ -13,9 +13,14 @@ type fakeTarget struct { applyErr map[Stage]error verifyErr error restoreErr error - applied []Stage - restored int - currentN int + // 恢复之后校验是否通过：用来区分"恢复跑完"和"恢复已验证"。 + verifyPassesAfterRestore bool + applied []Stage + restored int + currentN int + verifyN int + // 记录最后一次 Verify 是否发生在 Restore 之后（恢复必须被校验兜底）。 + verifiedAfterRestore bool } func (f *fakeTarget) Current(context.Context) (CurrentState, error) { @@ -35,7 +40,17 @@ func (f *fakeTarget) Apply(_ context.Context, stage Stage, _ string) error { return nil } -func (f *fakeTarget) Verify(context.Context, Expectation) error { return f.verifyErr } +func (f *fakeTarget) Verify(context.Context, Expectation) error { + f.verifyN++ + f.verifiedAfterRestore = f.restored > 0 + if f.verifyErr == nil { + return nil + } + if f.verifyPassesAfterRestore && f.restored > 0 { + return nil + } + return f.verifyErr +} func (f *fakeTarget) Restore(context.Context, BackupRef) error { f.restored++ @@ -185,12 +200,39 @@ func TestRunRollsBackWhenStageFails(t *testing.T) { } } -func TestRunRollsBackWhenVerificationFails(t *testing.T) { - target := &fakeTarget{verifyErr: errors.New("生产分块数不一致")} - outcome := Run(context.Background(), request(time.Now()), target, &fakeBackups{}, nil) - if outcome.State != "rolled_back" || target.restored != 1 { +// 恢复跑完但恢复后校验仍失败：不能报 rolled_back。真实事故里页面就是在这种 +// 状态下显示"已恢复到预备份并验证"，而目标其实还留着 67 行残留。 +func TestRunDoesNotClaimRollbackWhenRecoveryVerificationFails(t *testing.T) { + target := &fakeTarget{applyErr: map[Stage]error{StageResources: errors.New("资源写失败")}, + verifyErr: errors.New("生产分块数不一致")} + recorder := &stageLog{} + outcome := Run(context.Background(), request(time.Now()), target, &fakeBackups{}, recorder) + if outcome.State != "rollback_failed" || target.restored != 1 { t.Fatalf("outcome = %+v restored=%d", outcome, target.restored) } + if !strings.Contains(outcome.Rollback, "恢复后校验失败") { + t.Fatalf("rollback message = %q", outcome.Rollback) + } + if !recorder.has("rollback:failed") { + t.Fatalf("stage log = %v", recorder.entries) + } +} + +// 只有"恢复到预备份且恢复后校验通过"才允许报 rolled_back。 +func TestRunReportsRolledBackOnlyAfterRecoveryVerifies(t *testing.T) { + target := &fakeTarget{applyErr: map[Stage]error{StageResources: errors.New("资源写失败")}, + verifyErr: errors.New("半写状态"), verifyPassesAfterRestore: true} + recorder := &stageLog{} + outcome := Run(context.Background(), request(time.Now()), target, &fakeBackups{}, recorder) + if outcome.State != "rolled_back" { + t.Fatalf("outcome = %+v", outcome) + } + if !target.verifiedAfterRestore { + t.Fatal("restore must be followed by a content verification") + } + if !recorder.has("rollback:succeeded") { + t.Fatalf("stage log = %v", recorder.entries) + } } func TestRunReportsRollbackFailureDistinctly(t *testing.T) { diff --git a/internal/releaseadapter/v071/library_backup_store.go b/internal/releaseadapter/v071/library_backup_store.go index 7cbffd0..6970b0a 100644 --- a/internal/releaseadapter/v071/library_backup_store.go +++ b/internal/releaseadapter/v071/library_backup_store.go @@ -113,6 +113,12 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l if err != nil { return libraryrelease.BackupRef{}, err } + // 整实例指纹:恢复之后要拿它证明"多写的、少写的都没有"。只按目标范围比, + // 写到范围外的行会被漏掉(真实事故:恢复报成功但目标多 67 行)。 + universeBefore, err := universeFingerprint(ctx, s.pool) + if err != nil { + return libraryrelease.BackupRef{}, err + } pg, err := ExportLibrary(ctx, s.pool, s.identity) if err != nil { return libraryrelease.BackupRef{}, fmt.Errorf("读取生产整库失败: %w", err) @@ -154,6 +160,13 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l return libraryrelease.BackupRef{}, fmt.Errorf("预备份期间生产 %s 内容发生变化，切点不一致", table) } } + universeAfter, err := universeFingerprint(ctx, s.pool) + if err != nil { + return libraryrelease.BackupRef{}, err + } + if diff := universeDiff(universeBefore, universeAfter); diff != "" { + return libraryrelease.BackupRef{}, fmt.Errorf("预备份期间整实例知识域发生变化（%s），切点不一致", diff) + } for table, want := range current.Counts { if table == "index" { continue @@ -171,7 +184,7 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l total += part.Bytes } return libraryrelease.BackupRef{ID: "backup-" + snapshotID, SnapshotID: snapshotID, Bytes: total, - CreatedAt: createdAt}, nil + CreatedAt: createdAt, Universe: universeBefore}, nil } func (s *ProductionBackupStore) rowCounts(ctx context.Context) (map[string]int64, error) { diff --git a/internal/releaseadapter/v071/library_production_target.go b/internal/releaseadapter/v071/library_production_target.go index 79a63fb..6a52d58 100644 --- a/internal/releaseadapter/v071/library_production_target.go +++ b/internal/releaseadapter/v071/library_production_target.go @@ -84,8 +84,16 @@ type ProductionLibraryTarget struct { scope ProductionScope search *SearchExporter now func() time.Time - // storageBackend 是本次写入使用的目标存储后端 id（写入与校验共用）。 - storageBackend string + // storageBackend / storageProvider 是本次写入使用的目标存储后端与提供方 + // （写入与校验共用，缺一不可：只换桶不换提供方，目标取文件仍会走来源实例的路）。 + storageBackend string + storageProvider string + // snapshotTenant / snapshotKB 是这次要写入的快照来源身份。写入时按它做 + // 来源→目标的范围映射；校验时按它反查"目标里还有没有来源范围的行"。 + snapshotTenant uint64 + snapshotKB string + // targetEmbeddingModel 是目标知识库绑定的 embedding 模型 id（文档行要换成它）。 + targetEmbeddingModel string } // resolveStorageBackend 在目标库里找本次写入的存储后端。找不到就报错：宁可不写， @@ -98,24 +106,39 @@ func (t *ProductionLibraryTarget) resolveStorageBackend(ctx context.Context) err if !validSearchIdentifier(t.config.ObjectsBackendID) { return errors.New("目标存储后端 id 形状不合法") } - t.storageBackend = t.config.ObjectsBackendID + provider, err := t.storageProviderOf(ctx, t.config.ObjectsBackendID) + if err != nil { + return err + } + t.storageBackend, t.storageProvider = t.config.ObjectsBackendID, provider return nil } - var id string + var id, provider string err := t.pool.QueryRow(ctx, ` -SELECT id FROM storage_backends +SELECT id, coalesce(provider,'') FROM storage_backends WHERE tenant_id=$1 AND coalesce(config->>'bucket_name','')=$2 -ORDER BY created_at LIMIT 1`, t.scope.TenantID, t.config.ObjectsBucket).Scan(&id) +ORDER BY created_at LIMIT 1`, t.scope.TenantID, t.config.ObjectsBucket).Scan(&id, &provider) if err != nil { return errors.New("目标实例没有与目标桶匹配的存储后端，拒绝写入不可解析的资源定位符") } if !validSearchIdentifier(id) { return errors.New("目标存储后端 id 形状不合法") } - t.storageBackend = id + t.storageBackend, t.storageProvider = id, provider return nil } +// storageProviderOf 读出目标存储后端的提供方（minio / oss / s3…）。资源行里的 +// 提供方必须与它一致，否则目标应用会按来源提供方去找一个不存在的存储服务。 +func (t *ProductionLibraryTarget) storageProviderOf(ctx context.Context, id string) (string, error) { + var provider string + if err := t.pool.QueryRow(ctx, `SELECT coalesce(provider,'') FROM storage_backends WHERE id=$1 AND tenant_id=$2`, + id, t.scope.TenantID).Scan(&provider); err != nil { + return "", errors.New("目标实例没有这个存储后端，拒绝写入不可解析的资源定位符") + } + return provider, nil +} + func NewProductionLibraryTarget(ctx context.Context, config ProductionLibraryConfig, scope ProductionScope) (*ProductionLibraryTarget, error) { if err := config.validate(scope); err != nil { return nil, err @@ -259,6 +282,7 @@ func (t *ProductionLibraryTarget) Apply(ctx context.Context, stage libraryreleas if err := t.resolveStorageBackend(ctx); err != nil { return err } + t.resolveEmbeddingModel(ctx) archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024) if err != nil { return errors.New("归档不可用") @@ -268,6 +292,9 @@ func (t *ProductionLibraryTarget) Apply(ctx context.Context, stage libraryreleas if err != nil { return errors.New("预发布快照不可读") } + if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil { + return err + } switch stage { case libraryrelease.StageDatabase, libraryrelease.StageWiki: kind := "database" @@ -356,10 +383,10 @@ func (t *ProductionLibraryTarget) insertRows(ctx context.Context, tx pgx.Tx, arc if len(row) == 0 { continue } - if table == "resources" { - if err := t.applyResourceBucket(row); err != nil { - return err - } + // 行必须先映射到目标范围再落库：原样插入会把行写进来源租户/知识库， + // 目标范围的删除、校验与恢复都看不见它们。 + if err := t.retargetRow(table, row); err != nil { + return err } if columns == nil { columns = make([]string, 0, len(row)) @@ -408,53 +435,9 @@ type resourceLocation struct { } // rowTransform 是"写入时对行做过的变换"的统一入口：校验要用同一份，否则摘要比的是 -// 变换前的行、目标里是变换后的行，永远不一致。 +// 变换前的行、目标里是变换后的行，永远不一致。范围映射同样必须两边一致。 func (t *ProductionLibraryTarget) rowTransform(table string) func(map[string]json.RawMessage) error { - if table == "resources" { - return t.applyResourceBucket - } - return nil -} - -// applyResourceBucket 把资源行定位符里的桶换成本次写入的目标桶。 -// 写入与校验必须用同一个变换：只改写入侧会让"行内容摘要"对不上（真跑踩过）。 -func (t *ProductionLibraryTarget) applyResourceBucket(row map[string]json.RawMessage) error { - if t.config.ObjectsBucket == "" { - return nil - } - raw, ok := row["physical_path"] - if !ok { - return errors.New("资源行缺少定位符") - } - var locator string - if err := json.Unmarshal(raw, &locator); err != nil { - return errors.New("资源定位符不可解析") - } - rewritten, ok := rewriteLocatorTarget(locator, t.storageBackend, t.config.ObjectsBucket) - if !ok { - return fmt.Errorf("资源定位符无法改写到目标桶：%s", locator) - } - encoded, err := json.Marshal(rewritten) - if err != nil { - return err - } - row["physical_path"] = encoded - // 原版应用先按 resources.storage_backend_id 找存储后端，再回落到 KB，再落默认服务。 - // 这一列留着来源 id，目标仍然 500（真跑：storage backend not found → invalid path）。 - if t.storageBackend == "" { - return nil - } - if raw, ok := row["storage_backend_id"]; ok { - var current string - if json.Unmarshal(raw, ¤t) == nil && current != "" && current != t.storageBackend { - encodedBackend, err := json.Marshal(t.storageBackend) - if err != nil { - return err - } - row["storage_backend_id"] = encodedBackend - } - } - return nil + return func(row map[string]json.RawMessage) error { return t.retargetRow(table, row) } } func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error { @@ -591,6 +574,7 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele if err := t.resolveStorageBackend(ctx); err != nil { return err } + t.resolveEmbeddingModel(ctx) archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024) if err != nil { return errors.New("归档不可用") @@ -600,6 +584,9 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele if err != nil { return errors.New("预发布快照不可读") } + if err := t.bindSnapshotScope(manifestScope{tenantID: manifest.TenantID, knowledgeBaseID: manifest.KnowledgeBaseID}); err != nil { + return err + } for _, part := range manifest.Parts { if part.Kind != "database" && part.Kind != "wiki" { continue @@ -668,6 +655,72 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele if err := t.verifyObjects(ctx, archive, manifest); err != nil { return err } + // 目标里不允许留下"快照来源范围"的行：那说明有写入逃出了目标范围， + // 而恢复只按目标范围回放，永远删不到它们（真实事故就是这样被漏掉的）。 + if err := t.assertNoSourceScopeRows(ctx, archive, manifest); err != nil { + return err + } + return nil +} + +// dropOrphanResources 清掉"本次恢复后已无任何归属"的目标租户资源行。 +// +// 判据必须同时满足：属于目标租户、不在预备份快照里、没有任何 resource_binding、 +// 也没有任何分块正文以 resource:// 引用它。这样的行对目标应用完全不可见 +// （没有知识库能按它取到文件），只可能是失败写入留下的残渣。 +func (t *ProductionLibraryTarget) dropOrphanResources(ctx context.Context, backupSnapshotID string) error { + archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024) + if err != nil { + return errors.New("归档不可用") + } + defer archive.Close() + manifest, err := archive.Load(backupSnapshotID) + if err != nil { + return errors.New("预备份快照不可读") + } + kept := []string{} + for _, part := range manifest.Parts { + if part.Kind != "database" || part.Name != "resources" { + continue + } + ids, err := archivedResourceIDs(archive, part) + if err != nil { + return err + } + kept = append(kept, ids...) + } + rows, err := t.pool.Query(ctx, ` +SELECT r.id FROM resources AS r +WHERE r.tenant_id=$1 + AND NOT (r.id = ANY($2::text[])) + AND NOT EXISTS (SELECT 1 FROM resource_bindings b WHERE b.resource_id=r.id) + AND NOT EXISTS (SELECT 1 FROM chunks c WHERE c.tenant_id=r.tenant_id AND c.deleted_at IS NULL + AND coalesce(r.handle,'') <> '' + AND position('resource://'||r.handle IN coalesce(c.content,'')) > 0)`, + t.scope.TenantID, kept) + if err != nil { + return fmt.Errorf("查找目标残留资源失败: %w", err) + } + orphans := make([]string, 0, 8) + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return err + } + orphans = append(orphans, id) + } + rows.Close() + if err := rows.Err(); err != nil { + return err + } + if len(orphans) == 0 { + return nil + } + if _, err := t.pool.Exec(ctx, `DELETE FROM resources WHERE tenant_id=$1 AND id = ANY($2::text[])`, + t.scope.TenantID, orphans); err != nil { + return fmt.Errorf("清理目标残留资源失败: %w", err) + } return nil } @@ -772,5 +825,21 @@ func (t *ProductionLibraryTarget) Restore(ctx context.Context, backup libraryrel return fmt.Errorf("恢复阶段 %s 失败: %w", stage, err) } } + // resources 是多知识库共享表，回放阶段不删它：本次写入新增、恢复后已经 + // 没有任何归属的资源行会残留。它们对应用不可见，但属于"多写的"。 + if err := t.dropOrphanResources(ctx, backup.SnapshotID); err != nil { + return err + } + // 恢复工具把四个阶段都跑完不算恢复完成：整实例必须回到预备份时刻的样子。 + // 只比目标范围会把"写到范围外"的残留当成已恢复（真实事故：恢复后仍多 67 行）。 + if len(backup.Universe) > 0 { + after, err := universeFingerprint(ctx, t.pool) + if err != nil { + return fmt.Errorf("恢复后读取整库指纹失败: %w", err) + } + if diff := universeDiff(backup.Universe, after); diff != "" { + return fmt.Errorf("恢复后目标与预备份时刻不一致：%s（很可能有写入逃出了预备份范围，不能当作已恢复）", diff) + } + } return nil } diff --git a/internal/releaseadapter/v071/library_production_target_integration_test.go b/internal/releaseadapter/v071/library_production_target_integration_test.go index d771eb5..51c8ed0 100644 --- a/internal/releaseadapter/v071/library_production_target_integration_test.go +++ b/internal/releaseadapter/v071/library_production_target_integration_test.go @@ -4,12 +4,15 @@ import ( "bytes" "context" "crypto/rand" + "crypto/sha256" "encoding/hex" "encoding/json" "errors" + "fmt" "io" "net/http" "os" + "strconv" "strings" "testing" "time" @@ -390,3 +393,339 @@ func sealTargetFixture(t *testing.T, archive *librarysnapshot.Archive, scope str } return id } + +// memoryObjectStore 是"真实字节"的资源存储替身：Put/Open/Stat/Copy 都按摘要核对， +// 可以强制写入失败，用来复现"资源阶段失败"这一半写场景。 +type memoryObjectStore struct { + objects map[string][]byte + failPut bool + puts []string +} + +func newMemoryObjectStore(objects map[string][]byte) *memoryObjectStore { + copied := map[string][]byte{} + for key, value := range objects { + copied[key] = append([]byte(nil), value...) + } + return &memoryObjectStore{objects: copied} +} + +func objectKey(provider, path string) string { return provider + "|" + path } + +func (s *memoryObjectStore) Put(_ context.Context, provider, path string, reader io.Reader, size int64) error { + if s.failPut { + return errors.New("对象存储不可写") + } + body, err := io.ReadAll(reader) + if err != nil || int64(len(body)) != size { + return errors.New("对象字节数与声明不符") + } + s.objects[objectKey(provider, path)] = body + s.puts = append(s.puts, path) + return nil +} + +func (s *memoryObjectStore) Open(_ context.Context, provider, path string) (io.ReadCloser, error) { + body, ok := s.objects[objectKey(provider, path)] + if !ok { + return nil, errors.New("对象不存在") + } + return io.NopCloser(bytes.NewReader(body)), nil +} + +func (s *memoryObjectStore) Stat(_ context.Context, provider, path string) (int64, string, error) { + body, ok := s.objects[objectKey(provider, path)] + if !ok { + return 0, "", errors.New("对象不存在") + } + digest := sha256.Sum256(body) + return int64(len(body)), hex.EncodeToString(digest[:]), nil +} + +func (s *memoryObjectStore) Copy(_ context.Context, provider, source, target string) error { + body, ok := s.objects[objectKey(provider, source)] + if !ok { + return errors.New("对象不存在") + } + s.objects[objectKey(provider, target)] = append([]byte(nil), body...) + return nil +} + +// 真实事故的形状：来源 KB 与目标 KB 不同（租户也不同）、目标非空、资源阶段写失败。 +// 旧代码把归档行原样写进目标（落在来源租户/知识库），随后的回滚只清目标范围， +// 于是"已恢复到预备份"却留下 4 文件 / 38 分块 / 21 资源 / 4 绑定。 +func TestLibraryCrossScopeSyncRollbackLeavesNoResidueIsolated(t *testing.T) { + dsn, base := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL"), os.Getenv("KG_LIBRARY_TEST_ELASTIC_URL") + if dsn == "" || base == "" || os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" { + t.Skip("requires isolated PG and ES opt-in") + } + cfg, err := pgxpool.ParseConfig(dsn) + if err != nil || !strings.HasPrefix(cfg.ConnConfig.Database, "kg_library_test_") { + t.Fatal("isolated database required") + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatalf("open isolated PG: %v", err) + } + defer pool.Close() + for _, statement := range []string{ + `DROP SCHEMA public CASCADE; CREATE SCHEMA public`, + libraryTestSchema, + `ALTER TABLE wiki_pages ADD COLUMN status text DEFAULT 'published'`, + `ALTER TABLE knowledge_tag_relations ADD COLUMN tenant_id bigint`, + `ALTER TABLE resources ADD COLUMN provider text, ADD COLUMN size bigint, ADD COLUMN storage_backend_id text`, + `ALTER TABLE resources ADD PRIMARY KEY (id)`, + `ALTER TABLE resource_bindings ADD COLUMN relation text`, + `INSERT INTO knowledge_bases VALUES('kb-target',10001,'{}',NULL),('kb-source',10000,'{}',NULL)`, + `CREATE TABLE storage_backends(id text PRIMARY KEY, tenant_id bigint, provider text, config jsonb, created_at timestamptz)`, + `INSERT INTO storage_backends VALUES('backend-target',10001,'minio','{"bucket_name":"kg-test-bucket"}',now())`, + // 目标非空：目标自己有文档/分块，另一个租户也有数据（回滚不得碰它）。 + `INSERT INTO knowledges VALUES('doc-target',10001,'kb-target','completed',0,'',NULL)`, + `INSERT INTO knowledges VALUES('doc-foreign',20000,'kb-other','completed',0,'',NULL)`, + `INSERT INTO chunks VALUES('chunk-target',10001,'kb-target','doc-target','target content',NULL)`, + `INSERT INTO chunks VALUES('chunk-foreign',20000,'kb-other','doc-foreign','foreign content',NULL)`, + `UPDATE resource_bindings SET tenant_id=10000 WHERE tenant_id IS NULL`, + } { + if _, err := pool.Exec(ctx, statement); err != nil { + t.Fatalf("fixture schema: %v", err) + } + } + + token := make([]byte, 8) + if _, err := rand.Read(token); err != nil { + t.Fatal(err) + } + index := "kg_library_test_" + hex.EncodeToString(token) + client := &http.Client{Timeout: 20 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }} + call := func(method, path, body string, want int) { + t.Helper() + request, err := http.NewRequestWithContext(ctx, method, base+path, strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + request.Header.Set("Content-Type", "application/json") + response, err := client.Do(request) + if err != nil { + t.Fatalf("fixture HTTP: %v", err) + } + defer response.Body.Close() + if response.StatusCode != want { + t.Fatalf("fixture %s %s => %d, want %d", method, path, response.StatusCode, want) + } + } + call("PUT", "/"+index, `{"mappings":{"properties":{"tenant_id":{"type":"long"},"knowledge_base_id":{"type":"keyword"},"knowledge_id":{"type":"keyword"},"chunk_id":{"type":"keyword"},"content":{"type":"text"},"embedding":{"type":"dense_vector","dims":2}}}}`, 200) + defer func() { + call("DELETE", "/"+index, "", 200) + call("HEAD", "/"+index, "", 404) + }() + // 目标当前索引自洽：目标分块必须有对应文档，否则预备份会（正确地）拒绝。 + call("PUT", "/"+index+"/_doc/chunk-target?refresh=true", + `{"tenant_id":10001,"knowledge_base_id":"kb-target","knowledge_id":"doc-target","chunk_id":"chunk-target","source_id":"chunk-target","content":"target content","embedding":[0.1,0.2]}`, 201) + + archiveRoot := t.TempDir() + if err := os.Chmod(archiveRoot, 0700); err != nil { + t.Fatal(err) + } + archive, err := librarysnapshot.Open(archiveRoot, 64*1024*1024) + if err != nil { + t.Fatal(err) + } + defer archive.Close() + snapshotID := sealCrossScopeFixture(t, archive, 10000, "kb-source") + + store := newMemoryObjectStore(nil) + target, err := NewProductionLibraryTarget(ctx, ProductionLibraryConfig{ + Postgres: dsn, SearchURL: base, SearchIndex: index, Objects: store, ObjectsReader: store, + ArchiveRoot: archiveRoot, ObjectsBucket: "kg-test-bucket", MaxDocuments: 100, HTTPClient: client, + }, ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb-target"}) + if err != nil { + t.Fatalf("target: %v", err) + } + defer target.Close() + mapping, err := target.Search().MappingSHA256(ctx) + if err != nil { + t.Fatal(err) + } + backups, err := NewProductionBackupStore(ProductionBackupConfig{ + Pool: target.Pool(), Search: target.Search(), Objects: store, ArchiveRoot: archiveRoot, + Scope: "internal", Identity: target.Identity(), MaxDocuments: 100, + Compatibility: librarysnapshot.Compatibility{WeKnoraVersion: "v0.7.1", SchemaVersion: 74, + RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "text-embedding-v4", + EmbeddingDimension: 1024, IndexMappingSHA256: mapping}, + }) + if err != nil { + t.Fatalf("backup store: %v", err) + } + now := time.Now().UTC() + request := libraryrelease.Request{Target: "internal", SnapshotID: snapshotID, + ParamsDigest: "digest-cross-scope", Now: now, + Grant: libraryrelease.Grant{Target: "internal", SnapshotID: snapshotID, ParamsDigest: "digest-cross-scope", + ExpiresAt: now.Add(10 * time.Minute), AuthorizedBy: "acceptance"}} + + // 1) 资源阶段失败 → 必须回滚且不留下来源范围的行。 + store.failPut = true + outcome := libraryrelease.Run(ctx, request, target, backups, nil) + if outcome.State != "rolled_back" { + t.Fatalf("outcome = %+v", outcome) + } + if outcome.FailedAt != libraryrelease.StageResources { + t.Fatalf("failed_at = %s, want resources", outcome.FailedAt) + } + if outcome.Backup.SnapshotID == "" || len(outcome.Backup.Universe) == 0 { + t.Fatalf("backup must carry the whole-instance fingerprint: %+v", outcome.Backup) + } + for _, table := range []string{"knowledges", "chunks", "wiki_pages"} { + var residue int64 + if err := pool.QueryRow(ctx, fmt.Sprintf(`SELECT count(*) FROM %s WHERE tenant_id=$1 AND knowledge_base_id=$2`, table), + 10000, "kb-source").Scan(&residue); err != nil { + t.Fatal(err) + } + if residue != 0 { + t.Fatalf("%s still carries %d source-scope rows after rollback", table, residue) + } + } + // 夹具在租户 10000 里本来就有自己的行（res-safe / binding-safe）：它们必须原样 + // 留下，只有这次同步写进去的行（res-source / 绑到 doc-source）才算残留。 + for query, want := range map[string]int64{ + `SELECT count(*) FROM resources WHERE id='res-source'`: 0, + `SELECT count(*) FROM resource_bindings WHERE tenant_id=10000 AND owner_id='doc-source'`: 0, + `SELECT count(*) FROM resources WHERE tenant_id=10000`: 1, + } { + var got int64 + if err := pool.QueryRow(ctx, query).Scan(&got); err != nil { + t.Fatal(err) + } + if got != want { + t.Fatalf("%s = %d, want %d", query, got, want) + } + } + var leftovers int64 + if err := pool.QueryRow(ctx, `SELECT count(*) FROM resources WHERE tenant_id=10001`).Scan(&leftovers); err != nil { + t.Fatal(err) + } + if leftovers != 0 { + t.Fatalf("aborted write left %d resource rows in the target tenant", leftovers) + } + var foreign int64 + if err := pool.QueryRow(ctx, `SELECT count(*) FROM chunks WHERE tenant_id=20000`).Scan(&foreign); err != nil { + t.Fatal(err) + } + // 夹具本来就有 1 行 other-scope(20000)，加上本用例的 chunk-foreign，共 2 行。 + if foreign != 2 { + t.Fatalf("cross-tenant rows must survive rollback, got %d", foreign) + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: outcome.Backup.SnapshotID}); err != nil { + t.Fatalf("rollback must verify against the prebackup: %v", err) + } + + // 2) 同样的跨范围同步，资源能写 → 行必须落在目标范围，资源定位符指向目标存储。 + store.failPut = false + succeed := libraryrelease.Run(ctx, request, target, backups, nil) + if succeed.State != "succeeded" { + t.Fatalf("second run outcome = %+v", succeed) + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err != nil { + t.Fatalf("verify after cross-scope sync: %v", err) + } + 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'`). + 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 { + t.Fatal(err) + } + if chunkKB != "kb-target" { + t.Fatalf("chunk landed in %s", chunkKB) + } + if err := pool.QueryRow(ctx, `SELECT provider, physical_path FROM resources WHERE id='res-source'`). + Scan(&provider, &locator); err != nil { + t.Fatal(err) + } + if provider != "minio" { + t.Fatalf("resource provider = %s, want minio", provider) + } + if !strings.Contains(locator, "storage://backend-target/minio://kg-test-bucket/10001/") { + t.Fatalf("resource locator was not retargeted: %s", locator) + } + if strings.Contains(locator, "weknora/10000") { + t.Fatalf("resource locator still points at the source tenant: %s", locator) + } + t.Logf("cross_scope_sync verified: snapshot=%s backup=%s locator=%s", snapshotID, succeed.Backup.SnapshotID, locator) +} + +// sealCrossScopeFixture 造一份"来源范围与目标不同"的整库快照：文档/分块/资源/绑定/ +// Wiki/索引都带来源身份，资源部件是可核对的真实字节。 +func sealCrossScopeFixture(t *testing.T, archive *librarysnapshot.Archive, sourceTenant uint64, sourceKB string) string { + t.Helper() + encode := func(value any) []byte { + encoded, err := json.Marshal(value) + if err != nil { + t.Fatal(err) + } + return encoded + } + const handle = "abcdefghijklmnopqrstuv" + objectBytes := []byte("12345678") + parts := make([]librarysnapshot.Part, 0, 12) + put := func(kind, name string, count int64, body []byte) { + part, err := archive.Put(bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + part.Kind, part.Name, part.Count = kind, name, count + parts = append(parts, part) + } + put("database", "knowledges", 1, encode([]map[string]any{{ + "id": "doc-source", "tenant_id": sourceTenant, "knowledge_base_id": sourceKB, + "parse_status": "completed", "pending_subtasks_count": 0, "file_path": ""}})) + put("database", "chunks", 1, encode([]map[string]any{{ + "id": "chunk-source", "tenant_id": sourceTenant, "knowledge_base_id": sourceKB, + "knowledge_id": "doc-source", "content": "见 resource://" + handle}})) + put("database", "knowledge_tags", 0, encode([]map[string]any{})) + put("database", "knowledge_tag_relations", 0, encode([]map[string]any{})) + put("database", "resources", 1, encode([]map[string]any{{ + "id": "res-source", "handle": handle, "tenant_id": sourceTenant, + "storage_backend_id": "backend-source", "provider": "oss", + "physical_path": "storage://backend-source/oss://source-bucket/weknora/" + + formatTenant(sourceTenant) + "/res-source/a.png", + "size": int64(len(objectBytes))}})) + put("database", "resource_bindings", 1, encode([]map[string]any{{ + "id": "bind-source", "resource_id": "res-source", "tenant_id": sourceTenant, + "owner_type": "knowledge", "owner_id": "doc-source", "relation": "attachment"}})) + put("resource", "res-source", 1, objectBytes) + put("wiki", "wiki_folders", 0, encode([]map[string]any{})) + put("wiki", "wiki_pages", 1, encode([]map[string]any{{ + "id": "wiki-source", "tenant_id": sourceTenant, "knowledge_base_id": sourceKB, + "slug": "index", "content": "wiki"}})) + for _, name := range []string{"wiki_page_issues", "wiki_log_entries"} { + put("wiki", name, 0, encode([]map[string]any{})) + } + put("index", "chunks", 1, encode([]map[string]any{{"id": "chunk-source", "source": map[string]any{ + "tenant_id": sourceTenant, "knowledge_base_id": sourceKB, "knowledge_id": "doc-source", + "chunk_id": "chunk-source", "source_id": "chunk-source", "content": "见 resource://" + handle, + "embedding": []float64{0.1, 0.2}}}})) + manifest := librarysnapshot.Manifest{ValidationPolicy: librarysnapshot.ArtifactAcceptancePolicy, + Format: librarysnapshot.Format, Scope: "internal", TenantID: sourceTenant, KnowledgeBaseID: sourceKB, + CreatedAt: time.Now().UTC(), Parts: parts, + Compatibility: librarysnapshot.Compatibility{ + WeKnoraVersion: "v0.7.1", SchemaVersion: 74, RetrieverEngine: "elasticsearch_v8", + EmbeddingModel: "text-embedding-v4", EmbeddingDimension: 1024, + IndexMappingSHA256: strings.Repeat("a", 64), + }} + id, err := archive.Seal(manifest) + if err != nil { + t.Fatalf("seal cross-scope fixture: %v", err) + } + return id +} + +func formatTenant(tenant uint64) string { return strconv.FormatUint(tenant, 10) } diff --git a/internal/releaseadapter/v071/library_residue.go b/internal/releaseadapter/v071/library_residue.go new file mode 100644 index 0000000..b692c38 --- /dev/null +++ b/internal/releaseadapter/v071/library_residue.go @@ -0,0 +1,334 @@ +package v071 + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + + "knowledge-governance/internal/librarysnapshot" +) + +// ResidueReport 是"清理范围外残留"的审计记录：清掉了什么、依据是哪份归档、 +// 备份写到哪儿、清理后目标处于什么状态。没有这份记录就不许动目标。 +type ResidueReport struct { + SnapshotID string `json:"snapshot_id"` + SourceTenant uint64 `json:"source_tenant"` + SourceKB string `json:"source_kb"` + Tables map[string]int `json:"tables"` + BackupPath string `json:"backup_path"` + BackupSHA256 string `json:"backup_sha256"` + Applied bool `json:"applied"` + Cleared int `json:"cleared"` + Verified bool `json:"verified"` + Foreign map[string]string `json:"foreign_scopes,omitempty"` +} + +// ClearForeignScope 清掉"归档里确实有、却仍以来源身份躺在目标里"的行。 +// +// 它只做这一件事，而且必须全部条件成立才允许删： +// 1. 目标这些行的内容与指定归档行逐字段一致（先证明它们是这次写入的副本， +// 不是目标自己的数据）； +// 2. 先把它们完整导出成 0600 的备份文件并记下摘要； +// 3. 删除只按"归档身份集合 + 来源租户"精确命中，不做范围广删。 +// +// 清完还要复核：来源身份不再存在、目标内容与预备份一致、整库指纹里没有别的租户。 +func (t *ProductionLibraryTarget) ClearForeignScope(ctx context.Context, source ProductionScope, snapshotID, + backupDir string, apply bool) (*ResidueReport, error) { + if source.TenantID == 0 || !validSearchIdentifier(source.KnowledgeBaseID) { + return nil, errors.New("来源范围不完整，拒绝清理") + } + if source.TenantID == t.scope.TenantID && source.KnowledgeBaseID == t.scope.KnowledgeBaseID { + return nil, errors.New("来源与目标范围相同，不需要清理") + } + if backupDir == "" { + return nil, errors.New("清理前必须指定备份目录") + } + if err := t.resolveStorageBackend(ctx); err != nil { + return nil, err + } + archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024) + if err != nil { + return nil, errors.New("归档不可用") + } + defer archive.Close() + manifest, err := archive.Load(snapshotID) + if err != nil { + return nil, errors.New("指定快照不可读") + } + report := &ResidueReport{SnapshotID: snapshotID, SourceTenant: source.TenantID, SourceKB: source.KnowledgeBaseID, + Tables: map[string]int{}, Applied: false} + backup := map[string][]map[string]json.RawMessage{} + digests := map[string]string{} + identities := map[string][]string{} + for _, part := range manifest.Parts { + if part.Kind != "database" && part.Kind != "wiki" { + continue + } + archived, archivedDigests, err := archivedRowsWithDigests(archive, part) + if err != nil { + return nil, err + } + // 事故里的残留行是"归档行 + 只把桶/后端换成目标"的旧变换结果：判等时必须 + // 同时接受原样与这一种变换，否则会把真残留当成"不是这次写的"而拒绝清理。 + if part.Name == "resources" { + for _, row := range archived { + legacy, legacyErr := t.legacyResourceRow(row) + if legacyErr != nil { + return nil, legacyErr + } + digest, digestErr := canonicalRowDigest(legacy) + if digestErr != nil { + return nil, digestErr + } + archivedDigests[digest] = true + } + } + if len(archived) == 0 { + continue + } + live, err := t.foreignScopeRows(ctx, part.Name, source) + if err != nil { + return nil, err + } + liveDigests := make(map[string]string, len(live)) + for _, row := range live { + digest, err := canonicalRowDigest(row) + if err != nil { + return nil, err + } + liveDigests[digest] = archiveRowIdentity(part.Name, row) + } + // 目标里这一范围的行必须都能在归档里找到同样内容：否则它们不是这次写入的 + // 副本，删了就是删别人的数据。 + rows := make([]map[string]json.RawMessage, 0, len(live)) + ids := make([]string, 0, len(live)) + for digest, identity := range liveDigests { + if _, ok := archivedDigests[digest]; !ok { + return nil, fmt.Errorf("目标 %s 的行 %s 与指定归档不一致，拒绝按本次写入清理", part.Name, identity) + } + ids = append(ids, identity) + } + for _, row := range live { + rows = append(rows, row) + } + sort.Strings(ids) + report.Tables[part.Name] = len(rows) + backup[part.Name] = rows + identities[part.Name] = ids + digests[part.Name] = combineRowDigests(sortedKeys(liveDigests)) + } + total := 0 + for _, count := range report.Tables { + total += count + } + if total == 0 { + report.Verified = true + return report, nil + } + if err := os.MkdirAll(backupDir, 0o700); err != nil { + return nil, fmt.Errorf("创建备份目录失败: %w", err) + } + payload, err := json.MarshalIndent(map[string]any{ + "snapshot_id": snapshotID, "source_tenant": source.TenantID, "source_kb": source.KnowledgeBaseID, + "target_tenant": t.scope.TenantID, "target_kb": t.scope.KnowledgeBaseID, + "tables": report.Tables, "rows": backup, "digests": digests, + }, "", " ") + if err != nil { + return nil, err + } + report.BackupPath = filepath.Join(backupDir, "foreign-scope-residue-"+snapshotID[:12]+".json") + if err := os.WriteFile(report.BackupPath, payload, 0o600); err != nil { + return nil, fmt.Errorf("写残留备份失败: %w", err) + } + sum := sha256.Sum256(payload) + report.BackupSHA256 = hex.EncodeToString(sum[:]) + if !apply { + return report, nil + } + // 删除顺序：子表先、父表后（resource_bindings → resources，chunks → knowledges）。 + for index := len(productionLibraryTables) - 1; index >= 0; index-- { + table := productionLibraryTables[index] + ids, ok := identities[table] + if !ok || len(ids) == 0 { + continue + } + statement := fmt.Sprintf(`DELETE FROM %s WHERE id = ANY($1::text[])`, table) + if !tableHasTenant(table) { + // knowledge_tag_relations 没有 id：身份就是父文档 + 标签。 + statement = `DELETE FROM knowledge_tag_relations WHERE (knowledge_id||':'||tag_id) = ANY($1::text[])` + } + if _, err := t.pool.Exec(ctx, statement, ids); err != nil { + return nil, fmt.Errorf("清理 %s 失败: %w", table, err) + } + } + report.Applied = true + report.Cleared = total + return report, nil +} + +// legacyResourceRow 复现旧实现写资源行时的变换：只换定位符里的存储后端与桶， +// 租户/知识库/provider 原样保留。判等要认这一种形状，事故残留才是"可证明"的。 +func (t *ProductionLibraryTarget) legacyResourceRow(row map[string]json.RawMessage) (map[string]json.RawMessage, error) { + copied := make(map[string]json.RawMessage, len(row)) + for key, value := range row { + copied[key] = value + } + if t.config.ObjectsBucket == "" { + return copied, nil + } + raw, ok := copied["physical_path"] + if !ok { + return nil, errors.New("资源行缺少定位符") + } + var locator string + if err := json.Unmarshal(raw, &locator); err != nil { + return nil, errors.New("资源定位符不可解析") + } + if rewritten, ok := rewriteLocatorTarget(locator, t.storageBackend, t.config.ObjectsBucket); ok { + encoded, err := json.Marshal(rewritten) + if err != nil { + return nil, err + } + copied["physical_path"] = encoded + } + if t.storageBackend == "" { + return copied, nil + } + if raw, ok := copied["storage_backend_id"]; ok { + var current string + if json.Unmarshal(raw, ¤t) == nil && current != "" && current != t.storageBackend { + encoded, err := json.Marshal(t.storageBackend) + if err != nil { + return nil, err + } + copied["storage_backend_id"] = encoded + } + } + return copied, nil +} + +// VerifyResidueCleared 复核清理结果：来源身份不再存在、其它租户的解释得清、 +// 且（可选）目标内容与预备份完全一致。 +func (t *ProductionLibraryTarget) VerifyResidueCleared(ctx context.Context, source ProductionScope) error { + for _, table := range productionLibraryTables { + if !tableHasTenant(table) { + continue + } + statement := fmt.Sprintf(`SELECT count(*) FROM %s WHERE tenant_id=$1`, table) + arguments := []any{source.TenantID} + if tablesWithKB[table] { + statement += " AND knowledge_base_id=$2" + arguments = append(arguments, source.KnowledgeBaseID) + } + var count int64 + if err := t.pool.QueryRow(ctx, statement, arguments...).Scan(&count); err != nil { + return fmt.Errorf("复核 %s 失败: %w", table, err) + } + if count > 0 { + return fmt.Errorf("目标 %s 仍有 %d 行属于来源范围 %d/%s", table, count, source.TenantID, source.KnowledgeBaseID) + } + } + return nil +} + +// ForeignScopes 列出目标里所有"不是目标租户"的租户，用于人工确认没有别的泄漏。 +func (t *ProductionLibraryTarget) ForeignScopes(ctx context.Context) (map[string]string, error) { + foreign := map[string]string{} + for _, table := range productionLibraryTables { + if !tableHasTenant(table) { + continue + } + rows, err := t.pool.Query(ctx, fmt.Sprintf( + `SELECT tenant_id, count(*) FROM %s WHERE tenant_id <> $1 GROUP BY 1`, table), t.scope.TenantID) + if err != nil { + return nil, fmt.Errorf("读取 %s 租户分布失败: %w", table, err) + } + for rows.Next() { + var tenant, count int64 + if err := rows.Scan(&tenant, &count); err != nil { + rows.Close() + return nil, err + } + foreign[fmt.Sprintf("%s:%d", table, tenant)] = fmt.Sprintf("%d 行", count) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + } + return foreign, nil +} + +// foreignScopeRows 读出目标里属于来源范围的行（resources / resource_bindings +// 只有租户列，按租户取；其余按租户+知识库取）。 +func (t *ProductionLibraryTarget) foreignScopeRows(ctx context.Context, table string, source ProductionScope) ( + []map[string]json.RawMessage, error) { + if !tableHasTenant(table) { + return nil, nil + } + statement := fmt.Sprintf(`SELECT to_jsonb(%s) FROM %s WHERE %s`, table, table, scopeWhere(table)) + arguments := []any{source.TenantID, source.KnowledgeBaseID} + if table == "resources" || table == "resource_bindings" { + statement = fmt.Sprintf(`SELECT to_jsonb(%s) FROM %s WHERE tenant_id=$1`, table, table) + arguments = arguments[:1] + } + rows, err := t.pool.Query(ctx, statement, arguments...) + if err != nil { + return nil, fmt.Errorf("读取目标 %s 来源范围行失败: %w", table, err) + } + defer rows.Close() + result := make([]map[string]json.RawMessage, 0, 8) + for rows.Next() { + var row map[string]json.RawMessage + if err := rows.Scan(&row); err != nil { + return nil, err + } + result = append(result, row) + } + return result, rows.Err() +} + +// archivedRowsWithDigests 读出一份归档部件的行与规范化内容摘要。 +func archivedRowsWithDigests(archive *librarysnapshot.Archive, part librarysnapshot.Part) ( + []map[string]json.RawMessage, map[string]bool, error) { + reader, err := archive.OpenPart(part) + if err != nil { + return nil, nil, err + } + defer reader.Close() + decoder := json.NewDecoder(reader) + token, err := decoder.Token() + if err != nil || token != json.Delim('[') { + return nil, nil, errors.New("归档部件不是行数组") + } + rows := make([]map[string]json.RawMessage, 0, 64) + digests := map[string]bool{} + for decoder.More() { + var row map[string]json.RawMessage + if err := decoder.Decode(&row); err != nil { + return nil, nil, err + } + digest, err := canonicalRowDigest(row) + if err != nil { + return nil, nil, err + } + digests[digest] = true + rows = append(rows, row) + } + return rows, digests, nil +} + +func sortedKeys(values map[string]string) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} diff --git a/internal/releaseadapter/v071/library_scope_remap.go b/internal/releaseadapter/v071/library_scope_remap.go new file mode 100644 index 0000000..23fe813 --- /dev/null +++ b/internal/releaseadapter/v071/library_scope_remap.go @@ -0,0 +1,262 @@ +package v071 + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" +) + +// 整库同步是"把来源知识库的内容搬到目标的租户/知识库"，但归档行带着来源身份 +// （tenant_id / knowledge_base_id）。原样插入会让行落在来源范围里：目标范围的 +// 删除、校验、预备份与恢复全都看不见它们（真实事故：同步失败后"已恢复到预备份"， +// 目标却多出 4 文件 / 38 分块 / 21 资源 / 4 绑定，行所属租户还是来源租户）。 +// +// 因此写入与校验必须共用这一个改写入口，并且对不上就失败——宁可不写，也不写 +// 一行目标范围外的数据。 + +// tablesWithKB 是真实 WeKnora v0.7.1 里带 knowledge_base_id 的知识域表。 +// resources / resource_bindings 是多知识库共享表，只有 tenant_id； +// knowledge_tag_relations 两列都没有，归属完全由父表 knowledges 决定。 +var tablesWithKB = map[string]bool{ + "knowledges": true, "chunks": true, "knowledge_tags": true, + "wiki_folders": true, "wiki_pages": true, "wiki_page_issues": true, "wiki_log_entries": true, +} + +// tableHasTenant 报告这张表是否自带租户列。 +func tableHasTenant(table string) bool { return table != "knowledge_tag_relations" } + +// bindSnapshotScope 记下快照的来源身份。没有来源身份就无法判断"这行要搬到哪个 +// 范围"，宁可失败也不要按行内容猜。 +func (t *ProductionLibraryTarget) bindSnapshotScope(manifest manifestScope) error { + if manifest.tenantID == 0 || !validSearchIdentifier(manifest.knowledgeBaseID) { + return errors.New("快照缺少来源范围身份（tenant_id / knowledge_base_id），无法确定目标范围映射") + } + t.snapshotTenant = manifest.tenantID + t.snapshotKB = manifest.knowledgeBaseID + return nil +} + +// manifestScope 只是把快照的两个范围字段抽出来，避免调用点重复解构。 +type manifestScope struct { + tenantID uint64 + knowledgeBaseID string +} + +// retargetRow 把一行归档数据改写成目标范围的数据：tenant_id / knowledge_base_id +// 换成目标身份，资源行还要把定位符、存储后端、提供方换成目标实例自己的值。 +func (t *ProductionLibraryTarget) retargetRow(table string, row map[string]json.RawMessage) error { + if !tableHasTenant(table) { + return nil + } + if err := t.retargetTenantField(row); err != nil { + return fmt.Errorf("%s：%w", table, err) + } + if tablesWithKB[table] { + if err := t.retargetKnowledgeBaseField(row); err != nil { + return fmt.Errorf("%s：%w", table, err) + } + } + if table == "resources" { + return t.rewriteResourceIdentity(row) + } + if table == "knowledges" { + return t.retargetEmbeddingModelField(row) + } + return nil +} + +// retargetEmbeddingModelField 把文档的 embedding 模型换成目标知识库绑定的那一个。 +// 来源模型 id 在目标实例里通常不存在（目标 models 表可能是空的），留着它等于让 +// 应用拿一个解析不到的模型去查——库里的向量与查询向量就不是同一套了。 +func (t *ProductionLibraryTarget) retargetEmbeddingModelField(row map[string]json.RawMessage) error { + if t.targetEmbeddingModel == "" { + return nil + } + if _, ok := row["embedding_model_id"]; !ok { + return nil + } + encoded, err := json.Marshal(t.targetEmbeddingModel) + if err != nil { + return errors.New("编码目标 embedding 模型失败") + } + row["embedding_model_id"] = encoded + return nil +} + +// resolveEmbeddingModel 读目标知识库自己的 embedding 模型 id。读不到不算错误 +// （目标可能确实没绑模型），此时保持来源值不动，由契约校验去发现不一致。 +func (t *ProductionLibraryTarget) resolveEmbeddingModel(ctx context.Context) { + var model string + if err := t.pool.QueryRow(ctx, + `SELECT coalesce(embedding_model_id,'') FROM knowledge_bases WHERE id=$1 AND tenant_id=$2`, + t.scope.KnowledgeBaseID, t.scope.TenantID).Scan(&model); err == nil { + t.targetEmbeddingModel = model + } +} + +func (t *ProductionLibraryTarget) retargetTenantField(row map[string]json.RawMessage) error { + raw, ok := row["tenant_id"] + if !ok { + return errors.New("行缺少 tenant_id，无法确定要写入哪个范围") + } + var current uint64 + if err := json.Unmarshal(raw, ¤t); err != nil { + return errors.New("行 tenant_id 不可解析") + } + if t.snapshotTenant == 0 { + return errors.New("快照来源租户未知，拒绝猜测目标租户") + } + if current != t.snapshotTenant { + return fmt.Errorf("行 tenant_id=%d 不属于快照来源租户 %d", current, t.snapshotTenant) + } + encoded, err := json.Marshal(t.scope.TenantID) + if err != nil { + return errors.New("编码目标租户失败") + } + row["tenant_id"] = encoded + return nil +} + +func (t *ProductionLibraryTarget) retargetKnowledgeBaseField(row map[string]json.RawMessage) error { + raw, ok := row["knowledge_base_id"] + if !ok { + return errors.New("行缺少 knowledge_base_id，无法确定要写入哪个知识库") + } + var current string + if err := json.Unmarshal(raw, ¤t); err != nil || current == "" { + return errors.New("行 knowledge_base_id 不可解析") + } + if t.snapshotKB == "" { + return errors.New("快照来源知识库未知，拒绝猜测目标知识库") + } + if current != t.snapshotKB { + return fmt.Errorf("行 knowledge_base_id=%s 不属于快照来源知识库 %s", current, t.snapshotKB) + } + encoded, err := json.Marshal(t.scope.KnowledgeBaseID) + if err != nil { + return errors.New("编码目标知识库失败") + } + row["knowledge_base_id"] = encoded + return nil +} + +// rewriteResourceIdentity 把资源行的定位符与提供方换成目标实例的值。 +// +// 只换桶不够：来源行的 scheme（oss/minio…）与对象键里的租户段也是来源实例的， +// 目标实例按它取文件会落到别的提供方或别的租户前缀上（真跑：storage backend +// not found → invalid path → 500 code1007）。 +func (t *ProductionLibraryTarget) rewriteResourceIdentity(row map[string]json.RawMessage) error { + raw, ok := row["physical_path"] + if !ok { + return errors.New("资源行缺少定位符") + } + var locator string + if err := json.Unmarshal(raw, &locator); err != nil { + return errors.New("资源定位符不可解析") + } + rewritten, err := retargetLocator(locator, t.storageBackend, t.storageProvider, t.config.ObjectsBucket, + t.snapshotTenant, t.scope.TenantID) + if err != nil { + return err + } + if rewritten != locator { + encoded, err := json.Marshal(rewritten) + if err != nil { + return errors.New("编码目标定位符失败") + } + row["physical_path"] = encoded + } + if t.storageBackend != "" { + if _, ok := row["storage_backend_id"]; ok { + encoded, err := json.Marshal(t.storageBackend) + if err != nil { + return errors.New("编码目标存储后端失败") + } + row["storage_backend_id"] = encoded + } + } + if t.storageProvider != "" { + if _, ok := row["provider"]; ok { + encoded, err := json.Marshal(t.storageProvider) + if err != nil { + return errors.New("编码目标提供方失败") + } + row["provider"] = encoded + } + } + return nil +} + +// retargetLocator 把 "storage:///:///" 整体换成目标 +// 实例的真实取值：后端 id、scheme/提供方、桶名，以及对象键里的租户段。 +// 桶名为空表示目标没有配置对象存储（测试夹具），此时不动定位符。 +func retargetLocator(locator, backendID, provider, bucket string, sourceTenant, targetTenant uint64) (string, error) { + if bucket == "" || locator == "" { + return locator, nil + } + if locator != strings.TrimSpace(locator) { + return "", errors.New("资源定位符带首尾空白") + } + prefix := "" + rest := locator + if strings.HasPrefix(locator, "storage://") { + trimmed := strings.TrimPrefix(locator, "storage://") + parts := strings.SplitN(trimmed, "/", 2) + if len(parts) != 2 || !validSearchIdentifier(parts[0]) || parts[1] == "" { + return "", errors.New("资源定位符结构不可解析") + } + if backendID != "" { + if !validSearchIdentifier(backendID) { + return "", errors.New("目标存储后端 id 形状不合法") + } + parts[0] = backendID + } + prefix, rest = "storage://"+parts[0]+"/", parts[1] + } + separator := strings.Index(rest, "://") + if separator < 0 { + // 裸对象键没有桶与租户段：落盘按配置桶，改写无意义。 + return locator, nil + } + scheme := rest[:separator] + if !validSearchIdentifier(scheme) { + return "", errors.New("资源定位符的提供方不可解析") + } + schemeless := rest[separator+3:] + boundary := strings.Index(schemeless, "/") + if boundary <= 0 || boundary == len(schemeless)-1 { + return "", errors.New("资源定位符缺少对象键") + } + key := schemeless[boundary+1:] + if provider != "" { + if !validSearchIdentifier(provider) { + return "", errors.New("目标提供方形状不合法") + } + scheme = provider + } + return prefix + scheme + "://" + bucket + "/" + retargetKeyTenant(key, sourceTenant, targetTenant), nil +} + +// retargetKeyTenant 把对象键里的来源租户段换成目标租户，并丢掉来源实例自己的 +// 路径前缀（例如 MINIO_PATH_PREFIX=weknora/），让键与目标实例的写法一致。 +// 找不到来源租户段时原样返回：此时键本身不含跨租户引用。 +func retargetKeyTenant(key string, sourceTenant, targetTenant uint64) string { + if sourceTenant == 0 || targetTenant == 0 || sourceTenant == targetTenant { + return key + } + source, target := strconv.FormatUint(sourceTenant, 10), strconv.FormatUint(targetTenant, 10) + segments := strings.Split(key, "/") + for index := 0; index < len(segments) && index < 2; index++ { + // 只有"后面还有内容"的段才是路径前缀；整键就是租户号时不该动它。 + if index >= len(segments)-1 || segments[index] != source { + continue + } + kept := segments[index:] + kept[0] = target + return strings.Join(kept, "/") + } + return key +} diff --git a/internal/releaseadapter/v071/library_scope_remap_test.go b/internal/releaseadapter/v071/library_scope_remap_test.go new file mode 100644 index 0000000..0ce8aef --- /dev/null +++ b/internal/releaseadapter/v071/library_scope_remap_test.go @@ -0,0 +1,194 @@ +package v071 + +import ( + "encoding/json" + "strings" + "testing" +) + +// 真实事故的形状：来源 KB 与目标 KB 不同（租户也不同），归档行原样写进目标， +// 目标范围的删除/校验/恢复都看不到它们。这里把"必须改写哪些列"钉死。 + +func scopeTargetFixture() *ProductionLibraryTarget { + return &ProductionLibraryTarget{ + scope: ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb-target"}, + config: ProductionLibraryConfig{ObjectsBucket: "kgseal-1029a"}, + storageBackend: "backend-target", + storageProvider: "minio", + snapshotTenant: 10000, + snapshotKB: "kb-source", + } +} + +func rawRow(t *testing.T, fields map[string]any) map[string]json.RawMessage { + t.Helper() + row := make(map[string]json.RawMessage, len(fields)) + for key, value := range fields { + encoded, err := json.Marshal(value) + if err != nil { + t.Fatalf("encode %s: %v", key, err) + } + row[key] = encoded + } + return row +} + +func fieldText(t *testing.T, row map[string]json.RawMessage, key string) string { + t.Helper() + var value string + if err := json.Unmarshal(row[key], &value); err != nil { + t.Fatalf("%s not a string: %v", key, err) + } + return value +} + +func fieldUint(t *testing.T, row map[string]json.RawMessage, key string) uint64 { + t.Helper() + var value uint64 + if err := json.Unmarshal(row[key], &value); err != nil { + t.Fatalf("%s not a number: %v", key, err) + } + return value +} + +func TestRetargetRowMapsSourceScopeToTarget(t *testing.T) { + target := scopeTargetFixture() + for _, table := range []string{"knowledges", "chunks", "knowledge_tags", "wiki_pages", "wiki_folders"} { + row := rawRow(t, map[string]any{"id": "row-1", "tenant_id": 10000, "knowledge_base_id": "kb-source"}) + if err := target.retargetRow(table, row); err != nil { + t.Fatalf("%s retarget: %v", table, err) + } + if got := fieldUint(t, row, "tenant_id"); got != 10001 { + t.Fatalf("%s tenant_id = %d, want 10001", table, got) + } + if got := fieldText(t, row, "knowledge_base_id"); got != "kb-target" { + t.Fatalf("%s knowledge_base_id = %s, want kb-target", table, got) + } + } + // 共享表只有租户列：不得凭空写一个 knowledge_base_id 出来。 + binding := rawRow(t, map[string]any{"id": "bind-1", "tenant_id": 10000, "owner_type": "knowledge", "owner_id": "row-1"}) + if err := target.retargetRow("resource_bindings", binding); err != nil { + t.Fatalf("resource_bindings retarget: %v", err) + } + if got := fieldUint(t, binding, "tenant_id"); got != 10001 { + t.Fatalf("binding tenant_id = %d, want 10001", got) + } + if _, ok := binding["knowledge_base_id"]; ok { + t.Fatal("resource_bindings must not gain a knowledge_base_id column") + } + // knowledge_tag_relations 没有范围列：归属走父表，改写必须是空操作。 + relation := rawRow(t, map[string]any{"knowledge_id": "row-1", "tag_id": "tag-1"}) + if err := target.retargetRow("knowledge_tag_relations", relation); err != nil { + t.Fatalf("knowledge_tag_relations retarget: %v", err) + } + if len(relation) != 2 { + t.Fatalf("knowledge_tag_relations must stay untouched, got %v", relation) + } +} + +// 文档行的 embedding 模型必须换成目标知识库绑定的那一个：来源模型 id 在目标 +// 实例里通常不存在，留着它检索时的查询向量与库里向量就不是同一套。 +func TestRetargetRowUsesTargetEmbeddingModel(t *testing.T) { + target := scopeTargetFixture() + target.targetEmbeddingModel = "model-target" + row := rawRow(t, map[string]any{"id": "doc-1", "tenant_id": 10000, "knowledge_base_id": "kb-source", + "embedding_model_id": "model-source"}) + if err := target.retargetRow("knowledges", row); err != nil { + t.Fatalf("retarget: %v", err) + } + if got := fieldText(t, row, "embedding_model_id"); got != "model-target" { + t.Fatalf("embedding_model_id = %s, want model-target", got) + } + // 目标没绑模型时不猜：保持来源值，交给契约校验去发现不一致。 + plain := scopeTargetFixture() + row = rawRow(t, map[string]any{"id": "doc-1", "tenant_id": 10000, "knowledge_base_id": "kb-source", + "embedding_model_id": "model-source"}) + if err := plain.retargetRow("knowledges", row); err != nil { + t.Fatalf("retarget without target model: %v", err) + } + if got := fieldText(t, row, "embedding_model_id"); got != "model-source" { + t.Fatalf("embedding_model_id = %s, want unchanged", got) + } +} + +func TestRetargetRowFailsClosedOutsideSnapshotScope(t *testing.T) { + target := scopeTargetFixture() + cases := map[string]map[string]any{ + "foreign tenant": {"id": "x", "tenant_id": 20000, "knowledge_base_id": "kb-source"}, + "foreign knowledge": {"id": "x", "tenant_id": 10000, "knowledge_base_id": "kb-other"}, + "missing tenant": {"id": "x", "knowledge_base_id": "kb-source"}, + "missing knowledge kb": {"id": "x", "tenant_id": 10000}, + } + for name, fields := range cases { + if err := target.retargetRow("knowledges", rawRow(t, fields)); err == nil { + t.Fatalf("%s: retarget must fail closed", name) + } + } + // 快照来源未知时不许猜目标范围。 + unknown := &ProductionLibraryTarget{scope: ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb-target"}} + if err := unknown.retargetRow("knowledges", rawRow(t, map[string]any{"id": "x", "tenant_id": 10000, "knowledge_base_id": "kb"})); err == nil { + t.Fatal("unknown snapshot scope must fail closed") + } +} + +func TestRetargetRowRewritesResourceStorageIdentity(t *testing.T) { + target := scopeTargetFixture() + row := rawRow(t, map[string]any{ + "id": "res-1", + "tenant_id": 10000, + "storage_backend_id": "backend-source", + "provider": "oss", + "physical_path": "storage://backend-source/oss://yzh-weknora-test/weknora/10000/res-1/a.png", + }) + if err := target.retargetRow("resources", row); err != nil { + t.Fatalf("retarget resource: %v", err) + } + if got := fieldUint(t, row, "tenant_id"); got != 10001 { + t.Fatalf("resource tenant = %d, want 10001", got) + } + want := "storage://backend-target/minio://kgseal-1029a/10001/res-1/a.png" + if got := fieldText(t, row, "physical_path"); got != want { + t.Fatalf("physical_path = %q, want %q", got, want) + } + if got := fieldText(t, row, "provider"); got != "minio" { + t.Fatalf("provider = %q, want minio", got) + } + if got := fieldText(t, row, "storage_backend_id"); got != "backend-target" { + t.Fatalf("storage_backend_id = %q, want backend-target", got) + } +} + +func TestRetargetKeyTenantDropsSourcePrefix(t *testing.T) { + cases := []struct{ key, want string }{ + {"weknora/10000/exports/a.png", "10001/exports/a.png"}, + {"10000/knowledge/file.md", "10001/knowledge/file.md"}, + {"other/prefix/a.png", "other/prefix/a.png"}, + {"10000", "10000"}, + } + for _, tc := range cases { + if got := retargetKeyTenant(tc.key, 10000, 10001); got != tc.want { + t.Fatalf("retargetKeyTenant(%q) = %q, want %q", tc.key, got, tc.want) + } + } + // 同租户不编码：键必须原样返回。 + if got := retargetKeyTenant("weknora/10000/a.png", 10000, 10000); got != "weknora/10000/a.png" { + t.Fatalf("same tenant must keep key, got %q", got) + } +} + +func TestRetargetLocatorRejectsMalformed(t *testing.T) { + if _, err := retargetLocator(" not-a-locator ", "backend", "minio", "bucket", 10000, 10001); err == nil { + t.Fatal("locator with surrounding spaces must be rejected") + } + if _, err := retargetLocator("storage://backend/oss://", "backend", "minio", "bucket", 10000, 10001); err == nil { + t.Fatal("locator without object key must be rejected") + } + // 没有目标桶（测试夹具）时不改写，保持原值可继续用于采集/校验。 + locator := "storage://backend/oss://bucket/key" + if got, err := retargetLocator(locator, "backend", "minio", "", 10000, 10001); err != nil || got != locator { + t.Fatalf("empty bucket must be a no-op, got %q err=%v", got, err) + } + if !strings.Contains(retargetKeyTenant("weknora/10000/a", 10000, 10001), "10001") { + t.Fatal("tenant segment must be replaced") + } +} diff --git a/internal/releaseadapter/v071/library_universe.go b/internal/releaseadapter/v071/library_universe.go new file mode 100644 index 0000000..4abf5fb --- /dev/null +++ b/internal/releaseadapter/v071/library_universe.go @@ -0,0 +1,126 @@ +package v071 + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + + "github.com/jackc/pgx/v5/pgxpool" + + "knowledge-governance/internal/libraryrelease" + "knowledge-governance/internal/librarysnapshot" +) + +// universeFingerprint 是"整实例"的知识域指纹：不限租户/知识库的行数 + 身份集合摘要。 +// +// 为什么不能只比目标范围：恢复只按目标范围删除/回放，写到范围外的残留它看不见 +// （真实事故：作业报"已恢复到预备份"，目标却多出 4 文件 / 38 分块 / 21 资源 / +// 4 绑定）。整实例指纹让"多写了什么"必然暴露，同时也能证明恢复没有误删别的 +// 知识库的行。 +func universeFingerprint(ctx context.Context, pool *pgxpool.Pool) (map[string]libraryrelease.TableFingerprint, error) { + result := make(map[string]libraryrelease.TableFingerprint, len(productionLibraryTables)) + for _, table := range productionLibraryTables { + identity := identityExpr(table) + statement := fmt.Sprintf( + `SELECT count(*), coalesce(md5(string_agg((%s)::text, E'\n' ORDER BY (%s)::text)), '') FROM %s`, + identity, identity, table) + var fingerprint libraryrelease.TableFingerprint + if err := pool.QueryRow(ctx, statement).Scan(&fingerprint.Count, &fingerprint.Digest); err != nil { + return nil, fmt.Errorf("读取整库 %s 指纹失败: %w", table, err) + } + result[table] = fingerprint + } + return result, nil +} + +// archivedRowIdentities 读出一份归档部件的行身份（与校验用的身份口径一致）。 +func archivedRowIdentities(archive *librarysnapshot.Archive, part librarysnapshot.Part) ([]string, error) { + 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("归档部件不是行数组") + } + ids := make([]string, 0, 64) + for decoder.More() { + var row map[string]json.RawMessage + if err := decoder.Decode(&row); err != nil { + return nil, err + } + identity := archiveRowIdentity(part.Name, row) + if identity == "" { + continue + } + ids = append(ids, identity) + } + return ids, nil +} + +// universeDiff 返回第一处不一致的表；空字符串表示整实例与参照完全一致。 +func universeDiff(want, got map[string]libraryrelease.TableFingerprint) string { + tables := make([]string, 0, len(want)) + for table := range want { + tables = append(tables, table) + } + sort.Strings(tables) + for _, table := range tables { + current, ok := got[table] + if !ok { + return fmt.Sprintf("%s（目标读不到）", table) + } + if current.Count != want[table].Count { + return fmt.Sprintf("%s 行数 %d，参照 %d", table, current.Count, want[table].Count) + } + if current.Digest != want[table].Digest { + return fmt.Sprintf("%s 身份集合与参照不同（同数量改写也会被这条拦住）", table) + } + } + return "" +} + +// assertNoSourceScopeRows 找出"这次要写的这些行，仍然以来源身份躺在目标里"。 +// +// 判据按归档里的身份集合逐表核对，而不是"这个租户/知识库有没有行"：同一个租户 +// 里本来就可能住着别的知识库（共享资源、别的 KB 的绑定），按范围数会把它们误 +// 报成残留。这里只认"本次写集合里、仍属于来源租户"的行——它们只可能来自逃出 +// 目标范围的写入，而恢复只按目标范围回放，永远删不到它们。 +func (t *ProductionLibraryTarget) assertNoSourceScopeRows(ctx context.Context, archive *librarysnapshot.Archive, + manifest librarysnapshot.Manifest) error { + if t.snapshotTenant == 0 { + return nil + } + if t.snapshotTenant == t.scope.TenantID && t.snapshotKB == t.scope.KnowledgeBaseID { + return nil + } + for _, part := range manifest.Parts { + if part.Kind != "database" && part.Kind != "wiki" { + continue + } + if !tableHasTenant(part.Name) { + continue + } + ids, err := archivedRowIdentities(archive, part) + if err != nil { + return err + } + if len(ids) == 0 { + continue + } + statement := fmt.Sprintf(`SELECT count(*) FROM %s WHERE tenant_id=$1 AND id = ANY($2::text[])`, part.Name) + var count int64 + if err := t.pool.QueryRow(ctx, statement, t.snapshotTenant, ids).Scan(&count); err != nil { + return fmt.Errorf("检查目标 %s 的来源范围残留失败: %w", part.Name, err) + } + if count > 0 { + return fmt.Errorf("目标 %s 残留 %d 行仍属于来源范围 %d/%s：有写入逃出了目标范围，恢复不会清掉它们", + part.Name, count, t.snapshotTenant, t.snapshotKB) + } + } + return nil +} diff --git a/internal/releaseadapter/v071/resource_locator_test.go b/internal/releaseadapter/v071/resource_locator_test.go index a504e97..b106509 100644 --- a/internal/releaseadapter/v071/resource_locator_test.go +++ b/internal/releaseadapter/v071/resource_locator_test.go @@ -60,8 +60,11 @@ func TestRewriteLocatorSwapsStorageBackendWithBucket(t *testing.T) { // 行变换用解析到的目标后端与目标桶：写入与校验共用这一份，摘要才不会两边打架。 func TestResourceRowTransformRewritesBackendAndBucket(t *testing.T) { target := &ProductionLibraryTarget{config: ProductionLibraryConfig{ObjectsBucket: "kgseal-1029a"}, - storageBackend: "target-backend"} - row := map[string]json.RawMessage{"physical_path": json.RawMessage(`"storage://source-backend/minio://kgmaint-1028a/10001/k/ab.md"`)} + storageBackend: "target-backend", snapshotTenant: 10001, snapshotKB: "kb", + scope: ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb"}} + row := map[string]json.RawMessage{ + "tenant_id": json.RawMessage(`10001`), + "physical_path": json.RawMessage(`"storage://source-backend/minio://kgmaint-1028a/10001/k/ab.md"`)} if err := target.rowTransform("resources")(row); err != nil { t.Fatalf("row transform failed: %v", err) } @@ -78,8 +81,10 @@ func TestResourceRowTransformRewritesBackendAndBucket(t *testing.T) { // 只改定位符不改这一列，目标仍是 storage backend not found（真跑 500）。 func TestResourceRowTransformRewritesBackendColumn(t *testing.T) { target := &ProductionLibraryTarget{config: ProductionLibraryConfig{ObjectsBucket: "kgseal-1029a"}, - storageBackend: "target-backend"} + storageBackend: "target-backend", snapshotTenant: 10001, snapshotKB: "kb", + scope: ProductionScope{TenantID: 10001, KnowledgeBaseID: "kb"}} row := map[string]json.RawMessage{ + "tenant_id": json.RawMessage(`10001`), "physical_path": json.RawMessage(`"storage://source-backend/minio://kgmaint-1028a/1/k/ab.md"`), "storage_backend_id": json.RawMessage(`"source-backend"`), }