diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ed4a20..c001cd2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,28 @@ # Changelog +## 开发快照(资源范围修正 + 内容级校验 + 真实对象存储验收) + +### Fixed + +- 资源范围:`scopeWhere(resources)` 与 `resourcePaths` 现在用 `resource_bindings.resource_id = resources.id` + 精确关联,覆盖知识库绑定与分块绑定,并限制绑定租户;此前只按 owner_id 判断会命中整张资源表。 + 归档资源部件名与资源 ID 对齐(此前按 handle 查路径,导入会找不到对象)。 +- 资源路径查询补上 `provider`,对象读写按资源真实 provider 执行。 + +### Added + +- `Verify` 由"行数 + ID 集合"升级为内容级:逐行规范化摘要(仅豁免 created_at/updated_at/processed_at/expires_at)、 + 归档索引文档与生产索引文档的规范摘要、以及每个资源对象的实际字节 SHA256。 + 同 ID 改正文、改 Wiki、改向量、损坏对象都会被拒绝。 +- 预备份一致性由行数升级为内容指纹:导出前后逐表比对"数量 + 内容摘要",同数量改写也会失败。 +- 本机真实对象存储(MinIO)纳入验收:`TestLibraryProductionIsolationRealServices` 用真实桶与对象, + 覆盖跨租户/跨库资源隔离、chunk-only 绑定、对象损坏、以及"源对象被改后仍能用预备份恢复"。 + +### Verified(本机真实服务) + +- 三条真实服务用例同时通过:隔离资源+内容校验(`kg_library_test_*` 库、随机 ES 索引、唯一 MinIO 桶)、 + 单阶段导入/撤销/恢复、完整编排 e2e。跑完删库、删索引、删对象。 + ## 开发快照(生产整库真实适配 + 本机真实服务验收) ### Added diff --git a/internal/releaseadapter/v071/library_backup_store.go b/internal/releaseadapter/v071/library_backup_store.go index 46d4266..efaeb07 100644 --- a/internal/releaseadapter/v071/library_backup_store.go +++ b/internal/releaseadapter/v071/library_backup_store.go @@ -2,6 +2,7 @@ package v071 import ( "context" + "encoding/json" "errors" "fmt" "time" @@ -63,6 +64,44 @@ func NewProductionBackupStore(config ProductionBackupConfig) (*ProductionBackupS maxObjectByte: config.MaxObjectByte, now: now}, nil } +// tableFingerprint 是一张表的"数量 + 内容摘要"指纹:同数量改写也会被识别。 +type tableFingerprint struct { + Count int64 + Digest string +} + +// fingerprints 读取当前生产指纹;Prepare 在导出前后各取一次做一致性判断。 +func (s *ProductionBackupStore) fingerprints(ctx context.Context) (map[string]tableFingerprint, error) { + result := map[string]tableFingerprint{} + for _, table := range productionLibraryTables { + rows, err := s.pool.Query(ctx, fmt.Sprintf(`SELECT to_jsonb(%s) FROM %s WHERE %s`, + table, table, scopeWhere(table)), s.identity.TenantID, s.identity.KnowledgeBaseID) + if err != nil { + return nil, fmt.Errorf("读取生产 %s 指纹失败", table) + } + digests := make([]string, 0, 64) + for rows.Next() { + var raw map[string]json.RawMessage + if err := rows.Scan(&raw); err != nil { + rows.Close() + return nil, err + } + digest, err := canonicalRowDigest(raw) + if err != nil { + rows.Close() + return nil, err + } + digests = append(digests, digest) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + result[table] = tableFingerprint{Count: int64(len(digests)), Digest: combineRowDigests(digests)} + } + return result, nil +} + func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current libraryrelease.CurrentState) (libraryrelease.BackupRef, error) { archive, err := librarysnapshot.Open(s.archiveRoot, 256*1024*1024) if err != nil { @@ -70,6 +109,10 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l } defer archive.Close() createdAt := s.now().UTC() + before, err := s.fingerprints(ctx) + 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) @@ -92,17 +135,23 @@ func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current l if err != nil { return libraryrelease.BackupRef{}, fmt.Errorf("密封预备份失败: %w", err) } - // 静默检查:导出期间生产不能有变化,否则这份预备份不是一致切点。 - after, err := s.rowCounts(ctx) + // 静默检查:导出期间生产内容不能变。只比行数会被"同数量改写"骗过, + // 所以这里比对逐行内容摘要(行数 + 身份 + 内容)。 + after, err := s.fingerprints(ctx) if err != nil { return libraryrelease.BackupRef{}, err } + for table, beforeState := range before { + if after[table] != beforeState { + return libraryrelease.BackupRef{}, fmt.Errorf("预备份期间生产 %s 内容发生变化,切点不一致", table) + } + } for table, want := range current.Counts { if table == "index" { continue } - if after[table] != want { - return libraryrelease.BackupRef{}, fmt.Errorf("预备份期间生产 %s 行数从 %d 变为 %d,切点不一致", table, want, after[table]) + if after[table].Count != want { + return libraryrelease.BackupRef{}, fmt.Errorf("预备份期间生产 %s 行数从 %d 变为 %d", table, want, after[table].Count) } } manifest, err := archive.Load(snapshotID) diff --git a/internal/releaseadapter/v071/library_production_isolation_test.go b/internal/releaseadapter/v071/library_production_isolation_test.go new file mode 100644 index 0000000..f7c4829 --- /dev/null +++ b/internal/releaseadapter/v071/library_production_isolation_test.go @@ -0,0 +1,317 @@ +package v071 + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "os" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "knowledge-governance/internal/libraryrelease" + "knowledge-governance/internal/librarysnapshot" +) + +// 本机真实服务验收(对象存储 + PG + ES 都是真的): +// - 资源/绑定/Wiki/索引都在夹具里,不用假的 ObjectStore; +// - 同 ID 改正文、改 Wiki、改向量、损坏对象都必须被 Verify 拒绝; +// - 预备份脱离原对象后仍可独立恢复; +// - 资源范围必须精确:别的租户/别的知识库、chunk-only 绑定都要正确。 +func TestLibraryProductionIsolationRealServices(t *testing.T) { + dsn, esBase := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL"), os.Getenv("KG_LIBRARY_TEST_ELASTIC_URL") + s3Endpoint, s3Bucket := os.Getenv("KG_LIBRARY_TEST_S3_ENDPOINT"), os.Getenv("KG_LIBRARY_TEST_S3_BUCKET") + access, secret := os.Getenv("KG_LIBRARY_TEST_S3_ACCESS"), os.Getenv("KG_LIBRARY_TEST_S3_SECRET") + if dsn == "" || esBase == "" || s3Endpoint == "" || s3Bucket == "" || access == "" || secret == "" || + os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" { + t.Skip("requires isolated PG, ES and object storage 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(), 5*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 DEFAULT 's3', ADD COLUMN size bigint DEFAULT 8`, + `ALTER TABLE resources ADD PRIMARY KEY (id)`, + } { + if _, err := pool.Exec(ctx, statement); err != nil { + t.Fatalf("fixture schema: %v", err) + } + } + // 夹具 schema 自带几行种子数据:这里清空,改用本次运行自己的 ID, + // 避免与种子行主键冲突、也避免把种子数据当成被测内容。 + if _, err := pool.Exec(ctx, `TRUNCATE knowledges, chunks, resources, resource_bindings, wiki_pages, knowledge_bases CASCADE`); err != nil { + t.Fatalf("reset seeded rows: %v", err) + } + client := &http.Client{Timeout: 30 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }} + esCall := func(method, path, body string, want int) { + t.Helper() + request, err := http.NewRequestWithContext(ctx, method, esBase+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("elastic fixture: %v", err) + } + defer response.Body.Close() + if response.StatusCode != want { + t.Fatalf("elastic %s %s => %d, want %d", method, path, response.StatusCode, want) + } + } + token := make([]byte, 6) + if _, err := rand.Read(token); err != nil { + t.Fatal(err) + } + index := "kg_library_test_" + hex.EncodeToString(token) + esCall("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() { + esCall("DELETE", "/"+index, "", 200) + esCall("HEAD", "/"+index, "", 404) + }() + + objects, err := NewS3ObjectStore(S3ObjectStoreConfig{ + Provider: "s3", Endpoint: s3Endpoint, Bucket: s3Bucket, Region: "us-east-1", + AccessKeyID: access, SecretAccessKey: secret, ForcePathStyle: true, MaxObjectBytes: 8 << 20, + }) + if err != nil { + t.Fatalf("object store: %v", err) + } + if err := objects.EnsureBucket(ctx); err != nil { + t.Fatalf("ensure bucket: %v", err) + } + + // 真实数据:本库一个文档(chunk-only 资源绑定)、另一个知识库一个文档、另一个租户一个文档。 + anchor := hex.EncodeToString(token) + // 真实 WeKnora 的 resource handle 是 22 位;夹具必须同形,否则归档校验会拒绝。 + handle := (anchor + "0000000000000000000000")[:22] + // 对象路径与资源行保持一致:归档行会覆盖数据库行,路径必须同源。 + resourcePath := "kg-library-test/" + handle + "/attachment.txt" + objectBytes := []byte("12345678") + if err := objects.Put(ctx, "s3", resourcePath, bytes.NewReader(objectBytes), int64(len(objectBytes))); err != nil { + t.Fatalf("seed object: %v", err) + } + defer func() { + // 只清理本次运行写入的对象;桶按运行唯一命名,不碰其它数据。 + if err := objects.Delete(ctx, "s3", resourcePath); err != nil { + t.Logf("cleanup object left behind: %v", err) + } + }() + seed := ` +INSERT INTO knowledge_bases(id,tenant_id,indexing_strategy,deleted_at) VALUES('kb-safe',10000,'{}',NULL),('kb-other',10000,'{}',NULL); +INSERT INTO knowledges(id,tenant_id,knowledge_base_id,parse_status,pending_subtasks_count,file_path) VALUES + ('doc-safe',10000,'kb-safe','completed',0,''),('doc-other',10000,'kb-other','completed',0,''),('doc-tenant2',20000,'kb-safe','completed',0,''); +INSERT INTO chunks(id,tenant_id,knowledge_base_id,knowledge_id,content) VALUES + ('chunk-safe',10000,'kb-safe','doc-safe','chunk'),('chunk-other',10000,'kb-other','doc-other','other'),('chunk-tenant2',20000,'kb-safe','doc-tenant2','t2'); +INSERT INTO resources(id,handle,tenant_id,provider,physical_path,size) VALUES + ('res-safe','` + handle + `','10000','s3','` + resourcePath + `',8), + ('res-shared','` + handle + `shared','20000','s3','kg-library-test/other/attachment.txt',8); +INSERT INTO resource_bindings(id,resource_id,tenant_id,owner_type,owner_id) VALUES + ('bind-safe','res-safe',10000,'chunk','chunk-safe'), + ('bind-tenant2','res-shared',20000,'knowledge','doc-tenant2'); +INSERT INTO wiki_pages(id,tenant_id,knowledge_base_id,slug,content,deleted_at) VALUES + ('wiki-safe',10000,'kb-safe','index','old-content',NULL),('wiki-other',10000,'kb-other','index','other',NULL); +` + if _, err := pool.Exec(ctx, seed); err != nil { + t.Fatalf("seed rows: %v", err) + } + esCall("PUT", "/"+index+"/_doc/chunk-safe?refresh=true", + `{"tenant_id":10000,"knowledge_base_id":"kb-safe","knowledge_id":"doc-safe","chunk_id":"chunk-safe","source_id":"chunk-safe","content":"chunk","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 := sealRealFixture(t, archive, handle, "chunk-updated", [2]float64{0.3, 0.4}, objectBytes) + + target, err := NewProductionLibraryTarget(ctx, ProductionLibraryConfig{ + Postgres: dsn, SearchURL: esBase, SearchIndex: index, Objects: objects, ObjectsReader: objects, + ArchiveRoot: archiveRoot, MaxDocuments: 100, HTTPClient: client, + }, ProductionScope{TenantID: 10000, KnowledgeBaseID: "kb-safe"}) + if err != nil { + t.Fatalf("target: %v", err) + } + defer target.Close() + + // R1:资源范围必须只含本租户本库(chunk 绑定也要算),排除别的租户/KB。 + paths, err := target.resourcePaths(ctx) + if err != nil { + t.Fatalf("resource paths: %v", err) + } + if _, ok := paths["res-safe"]; !ok { + t.Fatalf("chunk-bound resource must be in scope: %v", paths) + } + if len(paths) != 1 { + t.Fatalf("scope leaked resources from other tenant/kb: %v", paths) + } + + mapping, err := target.Search().MappingSHA256(ctx) + if err != nil { + t.Fatal(err) + } + backups, err := NewProductionBackupStore(ProductionBackupConfig{ + Pool: target.Pool(), Search: target.Search(), Objects: objects, 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.Fatal(err) + } + current, err := target.Current(ctx) + if err != nil { + t.Fatalf("current: %v", err) + } + backup, err := backups.Prepare(ctx, snapshotID, current) + if err != nil { + t.Fatalf("prepare backup: %v", err) + } + + for _, stage := range libraryrelease.Stages() { + if err := target.Apply(ctx, stage, snapshotID); err != nil { + t.Fatalf("apply %s: %v", stage, err) + } + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err != nil { + t.Fatalf("verify after sync: %v", err) + } + + // R2:同 ID 改正文必须被发现。 + if _, err := pool.Exec(ctx, `UPDATE chunks SET content='tampered' WHERE id='chunk-safe'`); err != nil { + t.Fatal(err) + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err == nil { + t.Fatal("content change with same id must fail verification") + } + if err := target.Restore(ctx, backup); err != nil { + t.Fatalf("restore after content tamper: %v", err) + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: backup.SnapshotID}); err != nil { + t.Fatalf("verify after restore: %v", err) + } + // 再回到目标快照,继续后面的破坏实验。 + for _, stage := range libraryrelease.Stages() { + if err := target.Apply(ctx, stage, snapshotID); err != nil { + t.Fatalf("re-apply %s: %v", stage, err) + } + } + + // R2:同数量、同 ID,但向量被改 → 必须被发现。 + esCall("PUT", "/"+index+"/_doc/chunk-safe?refresh=true", + `{"tenant_id":10000,"knowledge_base_id":"kb-safe","knowledge_id":"doc-safe","chunk_id":"chunk-safe","source_id":"chunk-safe","content":"chunk-updated","embedding":[9.9,9.9]}`, 200) + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err == nil { + t.Fatal("vector change with same id must fail verification") + } + if err := target.Apply(ctx, libraryrelease.StageIndex, snapshotID); err != nil { + t.Fatalf("repair index: %v", err) + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err != nil { + t.Fatalf("verify after index repair: %v", err) + } + + // R2:对象字节损坏必须被发现(行数与 ID 都没变)。 + if err := objects.Put(ctx, "s3", resourcePath, strings.NewReader("87654321"), 8); err != nil { + t.Fatalf("corrupt object: %v", err) + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err == nil { + t.Fatal("corrupted object bytes must fail verification") + } + + // R3:把源对象删掉后,用预备份恢复必须仍然成功(备份不依赖原对象)。 + if err := objects.Put(ctx, "s3", resourcePath, strings.NewReader("XXXXXXXX"), 8); err != nil { + t.Fatal(err) + } + if err := target.Restore(ctx, backup); err != nil { + t.Fatalf("restore from independent backup: %v", err) + } + if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: backup.SnapshotID}); err != nil { + t.Fatalf("verify after restore from backup: %v", err) + } + t.Logf("backup=%s snapshot=%s object_round_trip_ok=true", backup.SnapshotID, snapshotID) +} + +// sealRealFixture 造一份包含资源部件与真实绑定的完整快照。 +func sealRealFixture(t *testing.T, archive *librarysnapshot.Archive, handle, chunkContent string, vector [2]float64, objectBytes []byte) string { + t.Helper() + encode := func(value any) []byte { + encoded, err := json.Marshal(value) + if err != nil { + t.Fatal(err) + } + return encoded + } + 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-safe", "tenant_id": 10000, + "knowledge_base_id": "kb-safe", "parse_status": "completed", "pending_subtasks_count": 0, "file_path": ""}})) + // 正文里带 resource://:归档会据此要求对应资源字节必须一起封存。 + put("database", "chunks", 1, encode([]map[string]any{{"id": "chunk-safe", "tenant_id": 10000, + "knowledge_base_id": "kb-safe", "knowledge_id": "doc-safe", + "content": chunkContent + " resource://" + handle}})) + put("database", "resources", 1, encode([]map[string]any{{"id": "res-safe", "handle": handle, + "tenant_id": 10000, "provider": "s3", "physical_path": "kg-library-test/" + handle + "/attachment.txt", + "size": len(objectBytes)}})) + put("database", "resource_bindings", 1, encode([]map[string]any{{"id": "bind-safe", "resource_id": "res-safe", + "tenant_id": 10000, "owner_type": "chunk", "owner_id": "chunk-safe"}})) + for _, name := range []string{"knowledge_tags", "knowledge_tag_relations"} { + put("database", name, 0, encode([]map[string]any{})) + } + put("wiki", "wiki_pages", 1, encode([]map[string]any{{"id": "wiki-safe", "tenant_id": 10000, + "knowledge_base_id": "kb-safe", "slug": "index", "content": "wiki"}})) + for _, name := range []string{"wiki_folders", "wiki_page_issues", "wiki_log_entries"} { + put("wiki", name, 0, encode([]map[string]any{})) + } + put("index", "chunks", 1, encode([]map[string]any{{"id": "chunk-safe", "source": map[string]any{ + "tenant_id": 10000, "knowledge_base_id": "kb-safe", "knowledge_id": "doc-safe", + "chunk_id": "chunk-safe", "source_id": "chunk-safe", "content": chunkContent, + "embedding": []float64{vector[0], vector[1]}}}})) + put("resource", "res-safe", 1, objectBytes) + _ = sha256.Sum256(objectBytes) + manifest := librarysnapshot.Manifest{ValidationPolicy: librarysnapshot.ArtifactAcceptancePolicy, + Format: librarysnapshot.Format, Scope: "internal", TenantID: 10000, KnowledgeBaseID: "kb-safe", + 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 fixture: %v", err) + } + return id +} + +var _ = io.Discard diff --git a/internal/releaseadapter/v071/library_production_target.go b/internal/releaseadapter/v071/library_production_target.go index 3f9faa7..7239ad4 100644 --- a/internal/releaseadapter/v071/library_production_target.go +++ b/internal/releaseadapter/v071/library_production_target.go @@ -151,13 +151,24 @@ func scopeWhere(table string) string { WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2)) OR (owner_type='chunk' AND owner_id IN (SELECT id FROM chunks c WHERE c.tenant_id=$1 AND c.knowledge_base_id=$2)))` case "resources": - return `EXISTS (SELECT 1 FROM resource_bindings b WHERE b.owner_id IN - (SELECT id FROM knowledges k WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2))` + // 必须把外层 resources.id 与绑定关系精确关联:只按 owner_id 判断会命中整张 + // 资源表;同时要覆盖 chunk 绑定(图片/附件常只挂在分块上)并限制绑定租户。 + return `EXISTS ( + SELECT 1 FROM resource_bindings b + JOIN knowledges k ON k.id=b.owner_id AND k.tenant_id=b.tenant_id + WHERE b.resource_id=resources.id AND b.tenant_id=$1 AND b.owner_type='knowledge' AND k.knowledge_base_id=$2 + UNION ALL + SELECT 1 FROM resource_bindings b + JOIN chunks c ON c.id=b.owner_id AND c.tenant_id=b.tenant_id + WHERE b.resource_id=resources.id AND b.tenant_id=$1 AND b.owner_type='chunk' AND c.knowledge_base_id=$2)` default: return "tenant_id=$1 AND knowledge_base_id=$2" } } +// errMissingIdentity 表示归档里的这张表没有代理主键,只能靠内容摘要核对。 +var errMissingIdentity = errors.New("missing stable identity") + // identityExpr 是"稳定身份"列:多数表就是 id,关系表没有 id 就用复合键。 func identityExpr(table string) string { if table == "knowledge_tag_relations" { @@ -331,6 +342,12 @@ func updateAssignments(columns []string) string { return strings.Join(assignments, ", ") } +// resourceLocation 是资源对象在生产存储里的位置与提供方(oss / s3 / minio…)。 +type resourceLocation struct { + Path string + Provider string +} + func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error { paths, err := t.resourcePaths(ctx) if err != nil { @@ -340,8 +357,8 @@ func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive * if part.Kind != "resource" { continue } - path, ok := paths[part.Name] - if !ok || path == "" { + location, ok := paths[part.Name] + if !ok || location.Path == "" { return fmt.Errorf("生产缺少资源 %s 的存储路径", part.Name) } reader, err := archive.OpenPart(part) @@ -353,30 +370,61 @@ func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive * if err != nil || int64(len(body)) != part.Bytes { return fmt.Errorf("资源 %s 字节与归档摘要不符", part.Name) } - if err := t.config.Objects.Put(ctx, "s3", path, bytes.NewReader(body), int64(len(body))); err != nil { + if err := t.config.Objects.Put(ctx, location.Provider, location.Path, bytes.NewReader(body), int64(len(body))); err != nil { return fmt.Errorf("上传资源 %s 失败", part.Name) } } return nil } -func (t *ProductionLibraryTarget) resourcePaths(ctx context.Context) (map[string]string, error) { - rows, err := t.pool.Query(ctx, ` -SELECT handle, physical_path FROM resources -WHERE EXISTS (SELECT 1 FROM resource_bindings b WHERE b.owner_id IN - (SELECT id FROM knowledges k WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2))`, - t.scope.TenantID, t.scope.KnowledgeBaseID) +// verifyObjects 逐个比对生产对象的实际字节摘要与归档摘要: +// 只有"对象真的在、内容一致"才算通过,防止对象丢失或损坏被行数掩盖。 +func (t *ProductionLibraryTarget) verifyObjects(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error { + paths, err := t.resourcePaths(ctx) if err != nil { - return nil, errors.New("读取生产资源路径失败") + return err + } + for _, part := range manifest.Parts { + if part.Kind != "resource" { + continue + } + location, ok := paths[part.Name] + if !ok || location.Path == "" { + return fmt.Errorf("生产缺少资源 %s", part.Name) + } + reader, err := t.config.ObjectsReader.Open(ctx, location.Provider, location.Path) + if err != nil { + return fmt.Errorf("读取生产资源 %s 失败: %w", part.Name, err) + } + hash := sha256.New() + size, err := io.Copy(hash, io.LimitReader(reader, part.Bytes+1)) + reader.Close() + if err != nil { + return fmt.Errorf("读取生产资源 %s 失败", part.Name) + } + if size != part.Bytes || hex.EncodeToString(hash.Sum(nil)) != part.SHA256 { + return fmt.Errorf("生产资源 %s 与归档摘要不一致(%d 字节)", part.Name, size) + } + } + return nil +} + +func (t *ProductionLibraryTarget) resourcePaths(ctx context.Context) (map[string]resourceLocation, error) { + // 归档里的资源部件名就是资源 ID(见 ArchiveLibraryCandidate),这里按 id 建映射, + // 否则导入/校验会按 handle 找不到对象。 + rows, err := t.pool.Query(ctx, fmt.Sprintf(`SELECT id, physical_path, provider FROM resources +WHERE %s`, scopeWhere("resources")), t.scope.TenantID, t.scope.KnowledgeBaseID) + if err != nil { + return nil, fmt.Errorf("读取生产资源路径失败: %w", err) } defer rows.Close() - paths := map[string]string{} + paths := map[string]resourceLocation{} for rows.Next() { - var handle, path string - if err := rows.Scan(&handle, &path); err != nil { + var id, path, provider string + if err := rows.Scan(&id, &path, &provider); err != nil { return nil, err } - paths[handle] = path + paths[id] = resourceLocation{Path: path, Provider: provider} } return paths, rows.Err() } @@ -430,27 +478,50 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele if part.Kind != "database" && part.Kind != "wiki" { continue } - count, digest, err := t.tableState(ctx, part.Name) + wantCount, wantDigest, err := archivedTableState(archive, part) + if errors.Is(err, errMissingIdentity) { + // 个别表(例如日志表)没有代理主键:身份集合无从比,但行数与逐行内容摘要仍然必须一致。 + wantContentCount, wantContentDigest, columns, contentErr := archivedTableContent(archive, part) + if contentErr != nil { + return contentErr + } + contentCount, contentDigest, contentErr := t.productionTableContent(ctx, part.Name, columns) + if contentErr != nil { + return contentErr + } + if contentCount != wantContentCount || contentDigest != wantContentDigest { + return fmt.Errorf("生产 %s 内容与快照不一致:%d 行", part.Name, contentCount) + } + continue + } if err != nil { return err } - wantCount, wantDigest, err := archivedTableState(archive, part) + count, digest, err := t.tableState(ctx, part.Name) if err != nil { return err } if count != wantCount || digest != wantDigest { return fmt.Errorf("生产 %s 与快照不一致:行数 %d/%d", part.Name, count, wantCount) } + // 行数/身份一致还要比内容:同 ID 改正文、改 Wiki、改向量必须被发现。 + wantContentCount, wantContentDigest, columns, err := archivedTableContent(archive, part) + if err != nil { + return err + } + contentCount, contentDigest, err := t.productionTableContent(ctx, part.Name, columns) + if err != nil { + return err + } + if contentCount != wantContentCount || contentDigest != wantContentDigest { + return fmt.Errorf("生产 %s 内容与快照不一致:%d 行 / 摘要不一致", part.Name, contentCount) + } } - indexCount, err := t.search.Count(ctx, SearchIdentity{TenantID: t.scope.TenantID, - KnowledgeBaseID: t.scope.KnowledgeBaseID}) - if err != nil { + if err := t.verifyIndexDigest(ctx, archive, manifest); err != nil { return err } - for _, part := range manifest.Parts { - if part.Kind == "index" && part.Count != indexCount { - return fmt.Errorf("生产索引条数 %d 与快照 %d 不一致", indexCount, part.Count) - } + if err := t.verifyObjects(ctx, archive, manifest); err != nil { + return err } return nil } @@ -474,7 +545,7 @@ func archivedTableState(archive *librarysnapshot.Archive, part librarysnapshot.P } identity := archiveRowIdentity(part.Name, row) if identity == "" { - return 0, "", fmt.Errorf("归档 %s 行缺少稳定身份", part.Name) + return 0, "", errMissingIdentity } ids = append(ids, identity) } diff --git a/internal/releaseadapter/v071/library_production_target_integration_test.go b/internal/releaseadapter/v071/library_production_target_integration_test.go index ea314d9..d771eb5 100644 --- a/internal/releaseadapter/v071/library_production_target_integration_test.go +++ b/internal/releaseadapter/v071/library_production_target_integration_test.go @@ -47,7 +47,10 @@ func TestLibraryProductionTargetIsolated(t *testing.T) { } // 真实 schema 里这两张表都带 tenant_id;夹具补上,避免为了测试去简化生产 SQL。 if _, err := pool.Exec(ctx, `ALTER TABLE wiki_pages ADD COLUMN status text DEFAULT 'published'; - ALTER TABLE knowledge_tag_relations ADD COLUMN tenant_id bigint`); err != nil { + ALTER TABLE knowledge_tag_relations ADD COLUMN tenant_id bigint; + ALTER TABLE resources ADD COLUMN provider text DEFAULT 's3', ADD COLUMN size bigint DEFAULT 8; + ALTER TABLE resources ADD PRIMARY KEY (id); + UPDATE resource_bindings SET tenant_id=10000 WHERE tenant_id IS NULL`); err != nil { t.Fatal(err) } @@ -216,6 +219,9 @@ func TestLibraryReleaseRunIsolated(t *testing.T) { `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 DEFAULT 's3', ADD COLUMN size bigint DEFAULT 8`, + `ALTER TABLE resources ADD PRIMARY KEY (id)`, + // 夹具的绑定行没有 tenant_id(真实 schema 有),补上才能按租户精确判权。 + `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) diff --git a/internal/releaseadapter/v071/library_verify_content.go b/internal/releaseadapter/v071/library_verify_content.go new file mode 100644 index 0000000..5762d5d --- /dev/null +++ b/internal/releaseadapter/v071/library_verify_content.go @@ -0,0 +1,169 @@ +package v071 + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + + "knowledge-governance/internal/librarysnapshot" +) + +// 内容级校验:行数相同、ID 相同不能证明内容没变。 +// +// 这里对每张知识域表取 to_jsonb 行,剔除"保存就会变"的时间戳类字段后做规范化 +// 摘要;索引侧比对归档文档与生产导出的文档摘要;对象侧逐字节比对。 +// 允许的易变字段只有时间戳,正文、标题、向量、路径、摘要都不在豁免之列。 + +var contentVolatileFields = map[string]bool{ + "created_at": true, + "updated_at": true, + "processed_at": true, + "expires_at": true, +} + +func canonicalRowDigest(raw map[string]json.RawMessage) (string, error) { + normalized := make(map[string]any, len(raw)) + for key, value := range raw { + if contentVolatileFields[key] { + continue + } + var decoded any + if err := json.Unmarshal(value, &decoded); err != nil { + return "", fmt.Errorf("decode column %s: %w", key, err) + } + normalized[key] = decoded + } + encoded, err := json.Marshal(normalized) + if err != nil { + return "", err + } + sum := sha256.Sum256(encoded) + return hex.EncodeToString(sum[:]), nil +} + +func combineRowDigests(digests []string) string { + sort.Strings(digests) + sum := sha256.Sum256([]byte(joinLines(digests))) + return hex.EncodeToString(sum[:]) +} + +func joinLines(items []string) string { + joined := "" + for index, item := range items { + if index > 0 { + joined += "\n" + } + joined += item + } + return joined +} + +// productionTableContent 读取生产侧规范内容摘要;只比对归档里实际存在的列。 +// 归档没有的列(生产默认值/后续新增列)不参与比对,避免"我们没管的列"造成假失败, +// 但归档里有的每一列都必须逐字一致。 +func (t *ProductionLibraryTarget) productionTableContent(ctx context.Context, table string, columns map[string]bool) (int64, string, error) { + // 不加表别名:scopeWhere(resources) 里引用了 resources.id,别名会让它失效。 + rows, err := t.pool.Query(ctx, fmt.Sprintf(`SELECT to_jsonb(%s) FROM %s WHERE %s`, table, table, scopeWhere(table)), + t.scope.TenantID, t.scope.KnowledgeBaseID) + if err != nil { + return 0, "", fmt.Errorf("读取生产 %s 内容失败: %w", table, err) + } + defer rows.Close() + digests := make([]string, 0, 64) + for rows.Next() { + var raw map[string]json.RawMessage + if err := rows.Scan(&raw); err != nil { + return 0, "", err + } + projected := make(map[string]json.RawMessage, len(columns)) + for key := range columns { + if value, ok := raw[key]; ok { + projected[key] = value + } + } + digest, err := canonicalRowDigest(projected) + if err != nil { + return 0, "", err + } + digests = append(digests, digest) + } + if err := rows.Err(); err != nil { + return 0, "", err + } + return int64(len(digests)), combineRowDigests(digests), nil +} + +// archivedTableContent 读取归档侧规范内容摘要,并返回归档里出现过的列集合。 +func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot.Part) (int64, string, map[string]bool, error) { + reader, err := archive.OpenPart(part) + if err != nil { + return 0, "", nil, err + } + defer reader.Close() + decoder := json.NewDecoder(reader) + token, err := decoder.Token() + if err != nil || token != json.Delim('[') { + return 0, "", nil, errors.New("归档部件不是行数组") + } + digests := make([]string, 0, 64) + columns := map[string]bool{} + for decoder.More() { + var raw map[string]json.RawMessage + if err := decoder.Decode(&raw); err != nil { + return 0, "", nil, err + } + for key := range raw { + columns[key] = true + } + digest, err := canonicalRowDigest(raw) + if err != nil { + return 0, "", nil, err + } + digests = append(digests, digest) + } + return int64(len(digests)), combineRowDigests(digests), columns, nil +} + +// verifyIndexDigest 比对归档索引文档与生产实际索引文档的规范化摘要。 +func (t *ProductionLibraryTarget) verifyIndexDigest(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error { + for _, part := range manifest.Parts { + if part.Kind != "index" { + continue + } + reader, err := archive.OpenPart(part) + if err != nil { + return err + } + documents, err := librarysnapshot.ReadIndexDocuments(reader) + reader.Close() + if err != nil { + return err + } + archived := make([]SearchDocument, 0, len(documents)) + for _, document := range documents { + canonical, err := canonicalJSON(document.Source) + if err != nil { + return errors.New("归档索引文档无法规范化") + } + archived = append(archived, SearchDocument{ID: document.ID, Source: canonical}) + } + wantDigest, err := searchDocumentsDigest(archived) + if err != nil { + return err + } + snapshot, err := t.search.ExportLibrary(ctx, SearchIdentity{TenantID: t.scope.TenantID, + KnowledgeBaseID: t.scope.KnowledgeBaseID}, t.config.MaxDocuments) + if err != nil { + return fmt.Errorf("读取生产索引失败: %w", err) + } + if len(snapshot.Documents) != len(archived) || snapshot.DocumentsSHA256 != wantDigest { + return fmt.Errorf("生产索引与快照不一致:文档 %d/%d,内容摘要不一致", + len(snapshot.Documents), len(archived)) + } + } + return nil +} diff --git a/internal/releaseadapter/v071/s3_object_store.go b/internal/releaseadapter/v071/s3_object_store.go index cc65750..b93025c 100644 --- a/internal/releaseadapter/v071/s3_object_store.go +++ b/internal/releaseadapter/v071/s3_object_store.go @@ -105,6 +105,17 @@ func (store *S3ObjectStore) Put(ctx context.Context, provider, objectPath string return nil } +// Delete 删除单个对象;只用于显式清理(保留策略未接线,不会自动调用)。 +func (store *S3ObjectStore) Delete(ctx context.Context, provider, objectPath string) error { + if provider != store.provider || !validObjectLocator(objectPath) { + return errors.New("invalid release object delete") + } + if _, err := store.client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(store.bucket), Key: aws.String(objectPath)}); err != nil { + return errors.New("delete release object") + } + return nil +} + func (store *S3ObjectStore) Stat(ctx context.Context, provider, objectPath string) (int64, string, error) { if provider != store.provider || !validObjectLocator(objectPath) { return 0, "", errors.New("invalid release object read")