fix(library): 资源范围精确关联 + 内容级校验 + 真实对象存储验收
build / build (push) Successful in 1m13s

为什么:父审指出资源范围未关联外层 resources,会把别的库/租户资源算进来;
Verify 只看行数与 ID,同 ID 改正文/改向量/损坏对象都能通过;集成测试还在用假对象存储。

- scopeWhere(resources)/resourcePaths 用 resource_bindings.resource_id=resources.id 精确关联,
  覆盖 knowledge/chunk 绑定并限制租户;资源部件名与资源 ID 对齐。
- Verify 增加逐行内容摘要、索引文档摘要、对象字节 SHA256 校验。
- 预备份一致性改为内容指纹(数量+摘要),同数量改写也会失败。
- 新增真实 MinIO 对象存储的本机验收用例,覆盖隔离/损坏/独立恢复。
This commit is contained in:
姚章浩
2026-10-28 16:23:28 +08:00
parent 6ca8481e8e
commit 0b9dbe12cf
7 changed files with 677 additions and 31 deletions
+23
View File
@@ -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
@@ -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)
@@ -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://<handle>:归档会据此要求对应资源字节必须一起封存。
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
@@ -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)
}
@@ -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)
@@ -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
}
@@ -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")