diff --git a/CHANGELOG.md b/CHANGELOG.md index 188729a..75e6e1a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,22 @@ # Changelog +## 开发快照 snapshot/v0.1.110 - 2026-10-28 + +### Fixed + +- 同一作业号重跑不再必然冲突：命令先读持久日志与已封存清单，按首次请求（时间、范围、契约、绑定库）幂等重放，`source_read=false`、`writers_stopped=false`；同一作业号换了范围或绑定库仍明确拒绝，已封存清单校验失败也拒绝。 +- 归档用量改按跨版本去重统计内容字节，并把清单、未被引用 blob、文件实际分配单独列出；不再把"各版本 unique 之和"当归档占用，也不再暗示可回收空间（真实两份清单：各版之和 55,309,695 字节，跨版本去重 55,309,693 字节，磁盘分配 55,373,824 字节）。 + +### Added + +- 冻结窗口内的契约复核：窗口内重新读取来源契约，与窗口前不一致即中止（不封存、恢复写入者）。 +- 写入者组之外的同镜像容器检查：若有不属于声明组的运行中容器使用声明写入者的镜像，拒绝封存。 +- 补齐 `GET /api/v1/library/snapshots`、`GET /api/v1/library/snapshots/{id}` 的 OpenAPI 契约（上一版遗漏，契约测试可检出）。 + +### Changed + +- 按用户确认取消 A06：文档、页面与接口说明不再把内容/问答/引用图片效果/性能测试列为封版或发布门槛；A04 数据完整性口径保留。 + ## 开发快照 snapshot/v0.1.108 - 2026-10-28 ### Added diff --git a/api/openapi.yaml b/api/openapi.yaml index 2f83c5f..c63ab08 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -812,6 +812,59 @@ } } }, + "/api/v1/library/snapshots": { + "get": { + "tags": [ + "Runtime" + ], + "operationId": "listLibrarySnapshots", + "summary": "列出已封存的整库完整版本（只读；含分项数量/字节/摘要、跨版本去重内容字节与文件系统事实）", + "responses": { + "200": { + "$ref": "#/components/responses/Envelope" + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "500": { + "$ref": "#/components/responses/InternalError" + } + } + } + }, + "/api/v1/library/snapshots/{id}": { + "get": { + "tags": [ + "Runtime" + ], + "operationId": "getLibrarySnapshot", + "summary": "读取单个已封存整库版本的分项与兼容契约（只读；不存在或校验失败返回 404）", + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "$ref": "#/components/responses/Envelope" + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "404": { + "description": "快照不存在或归档校验失败" + }, + "500": { + "$ref": "#/components/responses/InternalError" + } + } + } + }, "/api/v1/version-history": { "get": { "tags": [ diff --git a/cmd/libraryctl/main.go b/cmd/libraryctl/main.go index ebd5672..7a57212 100644 --- a/cmd/libraryctl/main.go +++ b/cmd/libraryctl/main.go @@ -224,11 +224,10 @@ type sourceConfig struct { ObjectDetail string } -func discoverSource(ctx context.Context, opts options) (sourceConfig, error) { +func discoverSource(ctx context.Context, opts options, binding map[string]string) (sourceConfig, error) { var config sourceConfig - binding, err := readRoutingBindings(ctx, opts.governancePG) - if err != nil { - return config, err + if binding == nil { + return config, errors.New("routing bindings are required") } config.Scopes = binding app, err := containerEnv(ctx, opts.sourceApp) @@ -385,7 +384,14 @@ func openArchive(opts options) (*librarysnapshot.Archive, error) { func runStatus(ctx context.Context, opts options) error { report := map[string]any{} - config, err := discoverSource(ctx, opts) + binding, bindingErr := readRoutingBindings(ctx, opts.governancePG) + var config sourceConfig + var err error + if bindingErr != nil { + err = bindingErr + } else { + config, err = discoverSource(ctx, opts, binding) + } if err != nil { report["source_discovery"] = err.Error() } else { @@ -471,7 +477,11 @@ func runPreflight(ctx context.Context, opts options) error { if opts.scope != "internal" && opts.scope != "external" { return errors.New("-scope must be internal or external") } - config, err := discoverSource(ctx, opts) + binding, err := readRoutingBindings(ctx, opts.governancePG) + if err != nil { + return err + } + config, err := discoverSource(ctx, opts, binding) if err != nil { return err } @@ -547,7 +557,41 @@ func runCapture(ctx context.Context, opts options) error { if job == "" { job = "lib-" + opts.scope + "-" + time.Now().UTC().Format("20060102T150405") } - config, err := discoverSource(ctx, opts) + archive, err := openArchive(opts) + if err != nil { + return err + } + defer archive.Close() + coordinator, control, err := openFreeze(ctx, opts) + if err != nil { + return err + } + defer coordinator.Close() + defer control.Close() + if err = coordinator.Recover(ctx); err != nil { + return fmt.Errorf("pending freeze requires recovery: %w", err) + } + // Replay is answered from the persisted journal + sealed archive: the same job + // re-used with the same scope/binding returns the original candidate without + // reading the source or stopping a writer, and a job re-used with different + // parameters is refused instead of silently sealing under a stale identity. + existing, found, err := coordinator.Inspect(job) + if err != nil { + return err + } + binding, err := readRoutingBindings(ctx, opts.governancePG) + if err != nil { + return err + } + scopedKB := binding[opts.scope] + if found { + manifest, err := replayCapture(job, opts.scope, scopedKB, existing, archive) + if err != nil { + return err + } + return printCapture(job, existing.SnapshotID, manifest, true) + } + config, err := discoverSource(ctx, opts, binding) if err != nil { return err } @@ -573,23 +617,24 @@ func runCapture(ctx context.Context, opts options) error { if err != nil { return err } - archive, err := openArchive(opts) - if err != nil { - return err + // The contract above is read before the maintenance window. Re-read it inside + // the stopped window and refuse if a model/index change happened in between: + // sealing a candidate whose contract no longer matches the source would hide + // exactly the change a later production sync cannot reproduce. + verifyContract := func(ctx context.Context) error { + live, err := compatibilityContract(ctx, pool, search, config.SearchIndex, tenant, kb) + if err != nil { + return err + } + if live != contract { + return errors.New("source contract changed between preflight and the frozen window; capture aborted") + } + return nil } - defer archive.Close() - coordinator, control, err := openFreeze(ctx, opts) - if err != nil { - return err - } - defer coordinator.Close() - defer control.Close() - if err = coordinator.Recover(ctx); err != nil { - return fmt.Errorf("pending freeze requires recovery: %w", err) - } - // sourceReady proves the *group* is complete and consistent. Recorded work - // settlement is enforced separately by CaptureLibrary; neither check can see - // unrecorded or remote work. + // sourceReady proves the declared writer group is complete and no other + // running container uses a declared writer image. Recorded work settlement is + // enforced separately by CaptureLibrary; neither check can see unrecorded or + // remote work, so this is not a cross-system consistency proof. sourceReady := func(ctx context.Context) error { items, err := control.Inventory(ctx) if err != nil { @@ -598,12 +643,21 @@ func runCapture(ctx context.Context, opts options) error { if len(items) == 0 { return errors.New("writer group is empty") } - return nil + declared := make(map[string]bool, len(items)) + for _, item := range items { + declared[item.ID] = true + } + images, err := writerImageIDs(ctx, items) + if err != nil { + return err + } + return checkNoUndeclaredWriter(ctx, images, declared) } result, err := v071.CaptureLibrary(ctx, coordinator, archive, pool, search, objects, sourceReady, v071.LibraryCaptureRequest{JobID: job, Scope: opts.scope, Identity: v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb}, - Compatibility: contract, CreatedAt: time.Now().UTC(), MaxDocuments: opts.maxDocuments}) + Compatibility: contract, CreatedAt: time.Now().UTC(), MaxDocuments: opts.maxDocuments, + VerifyContract: verifyContract}) if err != nil { printJSON(map[string]any{"job": job, "state": result.State, "snapshot_id": result.SnapshotID, "error": err.Error()}) return err @@ -612,6 +666,32 @@ func runCapture(ctx context.Context, opts options) error { if err != nil { return err } + return printCapture(job, result.SnapshotID, manifest, false) +} + +// replayCapture answers a repeated job from persisted state. It never reads the +// source and never stops a writer; conflicting parameters are refused. +func replayCapture(job, scope, expectedKB string, result libraryfreeze.Result, archive *librarysnapshot.Archive) (librarysnapshot.Manifest, error) { + if result.State != "succeeded" || result.SnapshotID == "" { + return librarysnapshot.Manifest{}, fmt.Errorf("job %q already exists with state %q: recover it or use a new job id", job, result.State) + } + manifest, err := archive.Load(result.SnapshotID) + if err != nil { + return librarysnapshot.Manifest{}, errors.New("captured library archive no longer verifies") + } + if manifest.ValidationPolicy != librarysnapshot.ArtifactAcceptancePolicy { + return librarysnapshot.Manifest{}, errors.New("existing job was sealed under a different validation policy") + } + if manifest.Scope != scope { + return librarysnapshot.Manifest{}, fmt.Errorf("job %q was captured for scope %q, not %q", job, manifest.Scope, scope) + } + if manifest.KnowledgeBaseID != expectedKB { + return librarysnapshot.Manifest{}, fmt.Errorf("job %q was captured for knowledge base %q, which is no longer the %q binding", job, manifest.KnowledgeBaseID, scope) + } + return manifest, nil +} + +func printCapture(job, snapshotID string, manifest librarysnapshot.Manifest, replay bool) error { parts := make([]map[string]any, 0, len(manifest.Parts)) var total int64 for _, part := range manifest.Parts { @@ -621,13 +701,58 @@ func runCapture(ctx context.Context, opts options) error { sort.Slice(parts, func(i, j int) bool { return fmt.Sprint(parts[i]["kind"])+fmt.Sprint(parts[i]["name"]) < fmt.Sprint(parts[j]["kind"])+fmt.Sprint(parts[j]["name"]) }) - printJSON(map[string]any{"job": job, "state": result.State, "snapshot_id": result.SnapshotID, + printJSON(map[string]any{"job": job, "state": "succeeded", "snapshot_id": snapshotID, "scope": manifest.Scope, "knowledge_base_id": manifest.KnowledgeBaseID, "tenant_id": manifest.TenantID, "compatibility": manifest.Compatibility, "parts": parts, "logical_bytes": total, - "validation_policy": manifest.ValidationPolicy, "writers_restored": true}) + "validation_policy": manifest.ValidationPolicy, "writers_restored": true, + "replay": replay, "source_read": !replay, "writers_stopped": !replay}) return nil } +func writerImageIDs(ctx context.Context, items []libraryfreeze.Service) ([]string, error) { + images := make([]string, 0, len(items)) + seen := map[string]bool{} + for _, item := range items { + image, err := containerImageOf(ctx, item.ID) + if err != nil { + return nil, err + } + if !seen[image] { + seen[image] = true + images = append(images, image) + } + } + return images, nil +} + +// checkNoUndeclaredWriter refuses to run when a running container uses a declared +// writer image but is not part of the declared group: such a container could keep +// writing during the maintenance window. +func checkNoUndeclaredWriter(ctx context.Context, images []string, declared map[string]bool) error { + for _, image := range images { + out, err := dockerOutput(ctx, "ps", "-q", "--filter", "ancestor="+image) + if err != nil { + return err + } + for _, id := range strings.Fields(out) { + if !declared[id] { + return fmt.Errorf("container %s uses a declared writer image but is not in the declared writer group", id[:12]) + } + } + } + return nil +} + +func containerImageOf(ctx context.Context, id string) (string, error) { + image, err := dockerOutput(ctx, "inspect", "--format", "{{.Image}}", id) + if err != nil { + return "", err + } + if image == "" { + return "", errors.New("declared writer has no image identity") + } + return image, nil +} func runList(opts options) error { archive, err := openArchive(opts) if err != nil { @@ -638,6 +763,10 @@ func runList(opts options) error { if err != nil { return err } + total, err := archive.ContentUsage() + if err != nil { + return err + } items := make([]map[string]any, 0, len(usage)) for _, item := range usage { manifest, err := archive.Load(item.VersionID) @@ -649,7 +778,12 @@ func runList(opts options) error { "logical_bytes": item.LogicalBytes, "unique_bytes": item.UniqueBytes, "shared_bytes": item.SharedBytes, "parts": len(manifest.Parts), "policy": manifest.ValidationPolicy}) } - printJSON(map[string]any{"archive": opts.archiveRoot, "snapshots": items}) + printJSON(map[string]any{"archive": opts.archiveRoot, "snapshots": items, + // 跨版本去重后的内容字节；文件系统事实单列，任何一项都不是可回收空间。 + "versions": total.Versions, "content_bytes": total.ContentBytes, + "blob_file_bytes": total.BlobFileBytes, "blob_allocated_bytes": total.BlobAllocatedBytes, + "unreferenced_blobs": total.UnreferencedBlobs, "unreferenced_bytes": total.UnreferencedBytes, + "manifest_file_bytes": total.ManifestFileBytes, "manifest_allocated_bytes": total.ManifestAllocatedBytes}) return nil } diff --git a/cmd/libraryctl/replay_test.go b/cmd/libraryctl/replay_test.go new file mode 100644 index 0000000..2835738 --- /dev/null +++ b/cmd/libraryctl/replay_test.go @@ -0,0 +1,126 @@ +package main + +import ( + "bytes" + "fmt" + "os" + "strings" + "testing" + "time" + + "knowledge-governance/internal/libraryfreeze" + "knowledge-governance/internal/librarysnapshot" +) + +// sealedFixture writes one complete sealed candidate and returns its ID. +func sealedFixture(t *testing.T, archive *librarysnapshot.Archive, scope, knowledgeBaseID string, createdAt time.Time) string { + t.Helper() + rows := func(count int) string { + if count == 0 { + return "[]" + } + items := make([]string, count) + for i := range items { + items[i] = fmt.Sprintf(`{"id":"00000000-0000-4000-8000-%012d"}`, i) + } + return "[" + strings.Join(items, ",") + "]" + } + parts := []struct { + kind, name string + count int64 + }{ + {"database", "knowledges", 4}, + {"database", "chunks", 38}, + {"database", "resources", 0}, + {"database", "resource_bindings", 0}, + {"database", "knowledge_tags", 0}, + {"database", "knowledge_tag_relations", 0}, + {"index", "chunks", 122}, + {"wiki", "wiki_pages", 31}, + {"wiki", "wiki_folders", 21}, + {"wiki", "wiki_page_issues", 0}, + {"wiki", "wiki_log_entries", 11}, + } + manifest := librarysnapshot.Manifest{ValidationPolicy: librarysnapshot.ArtifactAcceptancePolicy, + Format: librarysnapshot.Format, Scope: scope, TenantID: 10000, KnowledgeBaseID: knowledgeBaseID, + CreatedAt: createdAt, Compatibility: librarysnapshot.Compatibility{WeKnoraVersion: "v0.7.1", + SchemaVersion: 74, RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "text-embedding-v4", + EmbeddingDimension: 1024, IndexMappingSHA256: strings.Repeat("b", 64)}} + for _, part := range parts { + stored, err := archive.Put(bytes.NewReader([]byte(rows(int(part.count))))) + if err != nil { + t.Fatal(err) + } + stored.Kind, stored.Name, stored.Count = part.kind, part.name, part.count + manifest.Parts = append(manifest.Parts, stored) + } + id, err := archive.Seal(manifest) + if err != nil { + t.Fatal(err) + } + return id +} + +func fixtureArchive(t *testing.T) (*librarysnapshot.Archive, string) { + t.Helper() + root, err := os.MkdirTemp(os.TempDir(), "kglc-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(root) }) + if err = os.Chmod(root, 0700); err != nil { + t.Fatal(err) + } + archive, err := librarysnapshot.Open(root, 1024*1024) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { archive.Close() }) + return archive, root +} + +const fixtureKB = "0c84a098-d3ed-4b82-a165-e6de9691d6a8" + +// A repeated job with the same parameters must return the original candidate +// without touching the source or stopping a writer (this function has no access +// to either), instead of failing with a conflicting-request error. +func TestReplayCaptureReturnsOriginalCandidate(t *testing.T) { + archive, _ := fixtureArchive(t) + created := time.Date(2026, 10, 28, 3, 34, 57, 0, time.UTC) + id := sealedFixture(t, archive, "internal", fixtureKB, created) + manifest, err := replayCapture("lib-internal-1", "internal", fixtureKB, + libraryfreeze.Result{State: "succeeded", SnapshotID: id}, archive) + if err != nil { + t.Fatal(err) + } + if !manifest.CreatedAt.Equal(created) || manifest.Compatibility.EmbeddingModel != "text-embedding-v4" { + t.Fatalf("replay lost the original request identity: %+v", manifest) + } +} + +func TestReplayCaptureRejectsChangedParameters(t *testing.T) { + archive, root := fixtureArchive(t) + id := sealedFixture(t, archive, "internal", fixtureKB, time.Date(2026, 10, 28, 3, 34, 57, 0, time.UTC)) + // Different scope. + if _, err := replayCapture("lib-internal-1", "external", fixtureKB, + libraryfreeze.Result{State: "succeeded", SnapshotID: id}, archive); err == nil { + t.Fatal("scope conflict was accepted") + } + // Binding moved to another knowledge base. + if _, err := replayCapture("lib-internal-1", "internal", "aec3dfc9-4ad3-4a39-926b-5bfa5d6a9654", + libraryfreeze.Result{State: "succeeded", SnapshotID: id}, archive); err == nil { + t.Fatal("knowledge base binding change was accepted") + } + // A job that never succeeded must not be answered as success. + if _, err := replayCapture("lib-internal-1", "internal", fixtureKB, + libraryfreeze.Result{State: "failed"}, archive); err == nil { + t.Fatal("failed job replayed as success") + } + // A sealed candidate that no longer verifies must not be reported as replay. + if err := os.Remove(root + "/versions/" + id + ".json"); err == nil { + if _, err := replayCapture("lib-internal-1", "internal", fixtureKB, + libraryfreeze.Result{State: "succeeded", SnapshotID: id}, archive); err == nil { + t.Fatal("missing sealed candidate replayed as success") + } + } +} diff --git a/docs/library-capture.md b/docs/library-capture.md index 8a6d6f7..5260895 100644 --- a/docs/library-capture.md +++ b/docs/library-capture.md @@ -41,7 +41,11 @@ ./scripts/library-capture.sh recover ``` -产物默认写到 `~/.local/share/knowledge-governance/library`（0700），日志写到 `~/.local/share/knowledge-governance/freeze`（0700）。api 容器只以只读方式挂载前者，「整库版本」页据此展示；页面**没有**封存按钮，因为封存要维护窗口与来源凭据，只能在目标主机执行。 +产物默认写到 `~/.local/share/knowledge-governance/library`（0700），日志写到 `~/.local/share/knowledge-governance/freeze`（0700）。api 容器只以只读方式挂载前者，「整库版本」页据此展示。 + +同一个作业号重跑是**幂等重放**：直接返回已封存的同一份候选（`replay=true`、`source_read=false`、`writers_stopped=false`），不读来源、不停写入者；同一作业号换了 scope 或绑定库则明确拒绝。 + +页面侧的发起入口尚未接线（需要最小权限的主机执行边界，见进度文件），当前只能在目标主机执行命令。 ## 2026-10-28 实测（本机测试栈） @@ -57,13 +61,15 @@ ## 它证明了什么 / 没有证明什么 -已证明：这一份整库内容在**写入者停住**之后被完整导出，PostgreSQL 与 Elasticsearch 双向覆盖、生成问题、资源与摘要逐项核验通过，产物可寻址、可复读、可校验。 +已证明（A04 数据完整性口径）：这一份整库内容在**写入者停住**之后被完整导出，PostgreSQL 与 Elasticsearch 双向覆盖、生成问题、资源与摘要逐项核验通过，产物可寻址、可复读、可校验。这正是封版所需的完整性口径。 -没有证明（不要把候选当版本）： +用户已确认取消 A06：**不要求**内容、问答、引用/图片效果与性能测试及其报告，也不作为封版/生产发布门槛。因此这里不再把"没有做效果测试"列为缺口。 -- **内容/问答/性能验收**没有做，也没有把结果绑定到快照； -- **跨 PG / ES / 资源的一致切点**只是在维护窗口里取得，没有独立的切点证据； +仍然不成立（不许把完整性通过说成这些）： + +- **跨 PG / ES / 资源的一致切点**只在维护窗口内取得；写入者组完整性有检查，但没有独立的切点证据，不能当成已验收的一致性证明； - **Wiki 删除贡献的语义重算**没有做；Wiki 队列空不等于语义正确； - 对外库当前是空的（0 文档 1 Wiki 页），这份快照只说明"它当时确实是空的"； -- 生产备份/恢复、双闸授权、整库同步与失败续作都没有接线，本文不涉及； -- 上游父子分块漏正文索引的风险仍在：本次候选通过是因为库里现有数据已补齐，**以后再重解析**仍可能漏项，届时封存会因覆盖核验失败而阻断（这是设计如此）。 \ No newline at end of file +- 生产独立预备份、恢复演练、双闸授权、整库同步协调与失败续作都没有接线，本文不涉及； +- 从页面发起封存、持久作业互斥、启动恢复还没有接线（封存目前只能在目标主机执行命令）； +- 上游父子分块漏正文索引的风险仍在：本次通过是因为库里现有数据已补齐，**以后再重解析**仍可能漏项，届时封存会因覆盖核验失败而阻断（这是设计如此）。 \ No newline at end of file diff --git a/internal/httpapi/library_snapshots.go b/internal/httpapi/library_snapshots.go index 50d1c58..b9ce39d 100644 --- a/internal/httpapi/library_snapshots.go +++ b/internal/httpapi/library_snapshots.go @@ -15,8 +15,8 @@ import ( // 边界（刻意不做的事）： // - 不触发封存、不停/起写入者：封存需要维护窗口与来源凭据，只能由运维命令 // scripts/library-capture.sh 在目标主机执行； -// - 不把候选当版本：候选只证明"这一份内容被完整、可核验地封存了"，内容/问答/ -// 性能验收与生产发布是另外的、需要独立授权的事； +// - 不把完整性当发布：候选证明"这一份内容被完整、可核验地封存了"（A04 口径）， +// 而授权与生产同步是另外的、需要独立授权的步骤（A06 效果测试已由用户取消）； // - 归档根没接线时返回可用性=false 与原因，不返回空表冒充"没有快照"。 type librarySnapshotPart struct { Kind string `json:"kind"` @@ -47,12 +47,22 @@ type librarySnapshotList struct { Reason string `json:"reason,omitempty"` ArchiveRoot string `json:"archive_root,omitempty"` Snapshots []librarySnapshotView `json:"snapshots"` - StoredBytes int64 `json:"stored_bytes"` + // 归档级口径：ContentBytes 是**跨版本去重后、被版本引用的内容字节**。 + // 它不是磁盘占用（清单、未引用 blob、块对齐与元数据都不在里面）， + // 更不是可回收空间——清理需要保留策略、依赖保护与在线作业引用。 + Versions int `json:"versions"` + ContentBytes int64 `json:"content_bytes"` + BlobFileBytes int64 `json:"blob_file_bytes"` + BlobAllocatedBytes int64 `json:"blob_allocated_bytes"` + UnreferencedBlobs int `json:"unreferenced_blobs"` + UnreferencedBytes int64 `json:"unreferenced_bytes"` + ManifestFileBytes int64 `json:"manifest_file_bytes"` + ManifestAllocatedBytes int64 `json:"manifest_allocated_bytes"` // 页面必须能说清"封存由谁执行"，不能让人以为页面上有个没实现的按钮。 CaptureNote string `json:"capture_note"` } -const libraryCaptureNote = "封存由运维命令 scripts/library-capture.sh capture -scope internal|external 在目标主机执行（停写入者→导出→恢复）。页面只读展示已封存的候选；候选不等于验收通过或可发布。" +const libraryCaptureNote = "封存由运维命令 scripts/library-capture.sh capture -scope internal|external 在目标主机执行（停写入者→导出→恢复）。页面只读展示已封存的完整版本：它通过了数据完整性检查（身份、双向正文覆盖、生成问题、资源与摘要），这已满足封版所需的完整性口径；授权与生产同步仍是独立步骤。" func librarySnapshotsList(w http.ResponseWriter, r *http.Request, config Config) { view, err := readLibrarySnapshots(config.LibraryArchiveRoot) @@ -114,10 +124,24 @@ func readLibrarySnapshots(root string) (librarySnapshotList, error) { if err != nil { return view, errors.New("归档用量不可读") } + // Per-version UniqueBytes deduplicates only inside one version, so summing it + // over versions over-reports shared blobs. The archive-wide accounting below + // deduplicates across versions and keeps filesystem facts separate. + total, err := archive.ContentUsage() + if err != nil { + return view, errors.New("归档用量不可读") + } + view.Versions = total.Versions + view.ContentBytes = total.ContentBytes + view.BlobFileBytes = total.BlobFileBytes + view.BlobAllocatedBytes = total.BlobAllocatedBytes + view.UnreferencedBlobs = total.UnreferencedBlobs + view.UnreferencedBytes = total.UnreferencedBytes + view.ManifestFileBytes = total.ManifestFileBytes + view.ManifestAllocatedBytes = total.ManifestAllocatedBytes byID := map[string]librarysnapshot.Usage{} for _, item := range usage { byID[item.VersionID] = item - view.StoredBytes += item.UniqueBytes } for id, item := range byID { manifest, loadErr := archive.Load(id) diff --git a/internal/httpapi/library_snapshots_test.go b/internal/httpapi/library_snapshots_test.go index 788278c..d025736 100644 --- a/internal/httpapi/library_snapshots_test.go +++ b/internal/httpapi/library_snapshots_test.go @@ -101,10 +101,13 @@ func TestReadLibrarySnapshotsReportsSealedTotals(t *testing.T) { t.Fatalf("total %s = %d, want %d", key, snapshot.Totals[key], value) } } - // Stored bytes are deduplicated per blob, so they can be smaller than the - // logical sum but never larger. - if snapshot.LogicalBytes <= 0 || list.StoredBytes <= 0 || list.StoredBytes > snapshot.LogicalBytes { - t.Fatalf("bytes not reported: logical=%d stored=%d", snapshot.LogicalBytes, list.StoredBytes) + // The archive-wide content figure deduplicates blobs across versions, so it + // can be smaller than a version's logical sum but never larger. + if snapshot.LogicalBytes <= 0 || list.ContentBytes <= 0 || list.ContentBytes > snapshot.LogicalBytes { + t.Fatalf("bytes not reported: logical=%d content=%d", snapshot.LogicalBytes, list.ContentBytes) + } + if list.BlobAllocatedBytes < list.BlobFileBytes || list.ManifestFileBytes <= 0 { + t.Fatalf("filesystem facts not separated: %+v", list) } if snapshot.Compatibility.EmbeddingDimension != 1024 || snapshot.ValidationPolicy != librarysnapshot.ArtifactAcceptancePolicy { t.Fatalf("contract not reported: %+v", snapshot) diff --git a/internal/libraryfreeze/freeze.go b/internal/libraryfreeze/freeze.go index 4a652cc..9d6a17d 100644 --- a/internal/libraryfreeze/freeze.go +++ b/internal/libraryfreeze/freeze.go @@ -129,6 +129,25 @@ func (c *Coordinator) Recover(ctx context.Context) error { } return nil } + +// Inspect reports the persisted result of a job, if this coordinator ever ran it. +// A caller which re-uses a job ID needs the original request identity (time, +// scope, knowledge base, contract) to replay idempotently instead of guessing a +// fresh one, which would be rejected as a conflicting request. +func (c *Coordinator) Inspect(job string) (Result, bool, error) { + if !validID(job) { + return Result{}, false, errors.New("invalid freeze job id") + } + j, err := c.load(job) + if errors.Is(err, os.ErrNotExist) { + return Result{}, false, nil + } + if err != nil { + return Result{}, false, err + } + return j.Result, true, nil +} + func invokeCapture(ctx context.Context, capture func(context.Context) (string, error)) (id string, err error) { defer func() { if recover() != nil { diff --git a/internal/librarysnapshot/usage.go b/internal/librarysnapshot/usage.go index d8f1660..f2761be 100644 --- a/internal/librarysnapshot/usage.go +++ b/internal/librarysnapshot/usage.go @@ -6,9 +6,11 @@ import ( "errors" "io" "math" + "os" "slices" "sort" "strings" + "syscall" ) type Usage struct { @@ -97,6 +99,135 @@ func (a *Archive) Usage() ([]Usage, error) { return result, nil } +// ContentUsage is the archive-wide accounting of *content* bytes. +// +// It exists because per-version Usage() deduplicates only within one version: a +// blob shared by two versions is counted in both UniqueBytes, so summing those +// numbers over-reports the archive. ContentBytes here is deduplicated across +// every sealed version. +// +// It is still not filesystem occupancy: manifests, unreferenced blobs left by an +// interrupted Put and filesystem allocation (block rounding, metadata) are +// reported separately, and none of these numbers are reclaimable space. Cleanup +// needs retention/protection and live-job references from the coordinator. +type ContentUsage struct { + Versions int `json:"versions"` + ReferencedBlobs int `json:"referenced_blobs"` + ContentBytes int64 `json:"content_bytes"` + UnreferencedBlobs int `json:"unreferenced_blobs"` + UnreferencedBytes int64 `json:"unreferenced_bytes"` + BlobFileBytes int64 `json:"blob_file_bytes"` + BlobAllocatedBytes int64 `json:"blob_allocated_bytes"` + ManifestFileBytes int64 `json:"manifest_file_bytes"` + ManifestAllocatedBytes int64 `json:"manifest_allocated_bytes"` +} + +// ContentUsage walks the sealed versions and the blob directory once, with the +// same "inventory must not change under us" guard as Usage. +func (a *Archive) ContentUsage() (ContentUsage, error) { + var usage ContentUsage + ids, err := a.versionIDs() + if err != nil { + return usage, err + } + referenced := map[string]int64{} + for _, id := range ids { + m, err := a.Load(id) + if err != nil { + return usage, err + } + for _, p := range m.Parts { + referenced[p.SHA256] = p.Bytes + } + } + after, err := a.versionIDs() + if err != nil { + return usage, err + } + if !slices.Equal(ids, after) { + return usage, errors.New("snapshot inventory changed; retry usage read") + } + usage.Versions = len(ids) + usage.ReferencedBlobs = len(referenced) + for _, size := range referenced { + if usage.ContentBytes > math.MaxInt64-size { + return usage, errors.New("snapshot usage overflow") + } + usage.ContentBytes += size + } + blobs, err := directoryBytes(a.blobs, func(name string) error { + if !validDigest(name) { + return errors.New("unexpected archive blob entry") + } + return nil + }) + if err != nil { + return usage, err + } + usage.BlobFileBytes, usage.BlobAllocatedBytes = blobs.fileBytes, blobs.allocatedBytes + usage.UnreferencedBlobs = blobs.files - len(referenced) + usage.UnreferencedBytes = blobs.fileBytes + for _, size := range referenced { + usage.UnreferencedBytes -= size + } + if usage.UnreferencedBytes < 0 || usage.UnreferencedBlobs < 0 { + return usage, errors.New("archive blob inventory is inconsistent") + } + manifests, err := directoryBytes(a.versions, func(name string) error { + if strings.HasPrefix(name, ".pending-") { + return nil + } + if !strings.HasSuffix(name, ".json") { + return errors.New("unexpected archive version entry") + } + return nil + }) + if err != nil { + return usage, err + } + usage.ManifestFileBytes, usage.ManifestAllocatedBytes = manifests.fileBytes, manifests.allocatedBytes + return usage, nil +} + +type directoryUsage struct { + files int + fileBytes int64 + allocatedBytes int64 +} + +// directoryBytes reports apparent size and allocated size (st_blocks) so callers +// can tell content size from what the filesystem actually reserved. +func directoryBytes(root *os.Root, validate func(string) error) (directoryUsage, error) { + var usage directoryUsage + dir, err := root.Open(".") + if err != nil { + return usage, err + } + defer dir.Close() + entries, err := dir.ReadDir(-1) + if err != nil { + return usage, err + } + for _, entry := range entries { + if err := validate(entry.Name()); err != nil { + return usage, err + } + info, err := entry.Info() + if err != nil { + return usage, err + } + if !info.Mode().IsRegular() { + return usage, errors.New("archive entry is not a regular file") + } + usage.files++ + usage.fileBytes += info.Size() + if stat, ok := info.Sys().(*syscall.Stat_t); ok { + usage.allocatedBytes += stat.Blocks * 512 + } + } + return usage, nil +} + func (a *Archive) versionIDs() ([]string, error) { f, err := a.versions.Open(".") if err != nil { diff --git a/internal/librarysnapshot/usage_content_test.go b/internal/librarysnapshot/usage_content_test.go new file mode 100644 index 0000000..9746124 --- /dev/null +++ b/internal/librarysnapshot/usage_content_test.go @@ -0,0 +1,149 @@ +package librarysnapshot + +import ( + "bytes" + "fmt" + "os" + "strings" + "testing" + "time" +) + +// sealFixtureVersion seals one complete manifest whose parts are all derived +// from the same rows, so two versions of the same content share every blob. +func sealFixtureVersion(t *testing.T, archive *Archive, scope string, createdAt time.Time, documents int) string { + t.Helper() + rows := func(count int) string { + if count == 0 { + return "[]" + } + items := make([]string, count) + for i := range items { + items[i] = fmt.Sprintf(`{"id":"00000000-0000-4000-8000-%012d"}`, i) + } + return "[" + strings.Join(items, ",") + "]" + } + parts := []struct { + kind, name string + count int64 + body string + }{ + {"database", "knowledges", int64(documents), rows(documents)}, + {"database", "chunks", int64(documents), rows(documents)}, + {"database", "resources", 0, rows(0)}, + {"database", "resource_bindings", 0, rows(0)}, + {"database", "knowledge_tags", 0, rows(0)}, + {"database", "knowledge_tag_relations", 0, rows(0)}, + {"index", "chunks", int64(documents), rows(documents)}, + {"wiki", "wiki_pages", 0, rows(0)}, + {"wiki", "wiki_folders", 0, rows(0)}, + {"wiki", "wiki_page_issues", 0, rows(0)}, + {"wiki", "wiki_log_entries", 0, rows(0)}, + } + manifest := Manifest{ValidationPolicy: ArtifactAcceptancePolicy, Format: Format, Scope: scope, + TenantID: 10000, KnowledgeBaseID: "kb-safe", CreatedAt: createdAt, + Compatibility: Compatibility{WeKnoraVersion: "v0.7.1", SchemaVersion: 74, + RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "fixture-model", EmbeddingDimension: 1024, + IndexMappingSHA256: strings.Repeat("a", 64)}} + for _, part := range parts { + stored, err := archive.Put(bytes.NewReader([]byte(part.body))) + if err != nil { + t.Fatal(err) + } + stored.Kind, stored.Name, stored.Count = part.kind, part.name, part.count + manifest.Parts = append(manifest.Parts, stored) + } + id, err := archive.Seal(manifest) + if err != nil { + t.Fatal(err) + } + return id +} + +func fixtureArchive(t *testing.T) (*Archive, string) { + t.Helper() + root, err := os.MkdirTemp(os.TempDir(), "kgsnap-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(root) }) + if err = os.Chmod(root, 0700); err != nil { + t.Fatal(err) + } + archive, err := Open(root, 1024*1024) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { archive.Close() }) + return archive, root +} + +// Two sealed versions with identical content share every blob: summing the +// per-version UniqueBytes double counts them, which is the reporting bug this +// accounting fixes. +func TestContentUsageDeduplicatesAcrossVersions(t *testing.T) { + archive, _ := fixtureArchive(t) + sealFixtureVersion(t, archive, "internal", time.Date(2026, 10, 28, 1, 0, 0, 0, time.UTC), 4) + sealFixtureVersion(t, archive, "internal", time.Date(2026, 10, 28, 2, 0, 0, 0, time.UTC), 4) + perVersion, err := archive.Usage() + if err != nil { + t.Fatal(err) + } + if len(perVersion) != 2 { + t.Fatalf("expected two versions, got %d", len(perVersion)) + } + sum := perVersion[0].UniqueBytes + perVersion[1].UniqueBytes + total, err := archive.ContentUsage() + if err != nil { + t.Fatal(err) + } + if total.Versions != 2 { + t.Fatalf("versions=%d", total.Versions) + } + if total.ContentBytes <= 0 { + t.Fatal("no content bytes reported") + } + if total.ContentBytes*2 != sum { + t.Fatalf("shared content not deduplicated: content=%d sum-of-versions=%d", total.ContentBytes, sum) + } + if total.UnreferencedBlobs != 0 || total.UnreferencedBytes != 0 { + t.Fatalf("unexpected unreferenced blobs: %+v", total) + } + if total.BlobFileBytes != total.ContentBytes { + t.Fatalf("blob files=%d content=%d", total.BlobFileBytes, total.ContentBytes) + } + if total.BlobAllocatedBytes < total.BlobFileBytes { + t.Fatalf("allocated=%d smaller than apparent=%d", total.BlobAllocatedBytes, total.BlobFileBytes) + } + if total.ManifestFileBytes <= 0 || total.ManifestAllocatedBytes < total.ManifestFileBytes { + t.Fatalf("manifest accounting wrong: %+v", total) + } +} + +// A blob written by an interrupted capture is referenced by no manifest and must +// be reported separately, never folded into content bytes. +func TestContentUsageSeparatesUnreferencedBlobs(t *testing.T) { + archive, _ := fixtureArchive(t) + sealFixtureVersion(t, archive, "internal", time.Date(2026, 10, 28, 3, 0, 0, 0, time.UTC), 2) + before, err := archive.ContentUsage() + if err != nil { + t.Fatal(err) + } + stray, err := archive.Put(bytes.NewReader([]byte("interrupted-write"))) + if err != nil { + t.Fatal(err) + } + after, err := archive.ContentUsage() + if err != nil { + t.Fatal(err) + } + if after.ContentBytes != before.ContentBytes { + t.Fatal("unreferenced blob changed content bytes") + } + if after.UnreferencedBlobs != 1 || after.UnreferencedBytes != stray.Bytes { + t.Fatalf("unreferenced accounting wrong: %+v (stray=%d)", after, stray.Bytes) + } + if after.BlobFileBytes != before.BlobFileBytes+stray.Bytes { + t.Fatal("blob file bytes do not include the unreferenced blob") + } +} diff --git a/internal/releaseadapter/v071/library_capture.go b/internal/releaseadapter/v071/library_capture.go index f657ec2..f3c72bb 100644 --- a/internal/releaseadapter/v071/library_capture.go +++ b/internal/releaseadapter/v071/library_capture.go @@ -15,6 +15,11 @@ type LibraryCaptureRequest struct { Compatibility librarysnapshot.Compatibility CreatedAt time.Time MaxDocuments int + // VerifyContract, when set, runs inside the stopped window immediately before + // the export. The caller re-reads the source contract and returns an error if + // it differs from Compatibility, so a model/index change between the + // preflight read and the maintenance window cannot be sealed silently. + VerifyContract func(context.Context) error } // CaptureLibrary connects maintenance to the actual PG/PIT/object archive path. @@ -45,6 +50,11 @@ func CaptureLibrary(ctx context.Context, freeze *libraryfreeze.Coordinator, arch if err := sourceReady(ctx); err != nil { return "", err } + if req.VerifyContract != nil { + if err := req.VerifyContract(ctx); err != nil { + return "", err + } + } pg, err := ExportLibrary(ctx, db, req.Identity) if err != nil { return "", err diff --git a/internal/releaseadapter/v071/library_capture_test.go b/internal/releaseadapter/v071/library_capture_test.go index 865207e..6caef36 100644 --- a/internal/releaseadapter/v071/library_capture_test.go +++ b/internal/releaseadapter/v071/library_capture_test.go @@ -2,6 +2,7 @@ package v071 import ( "context" + "errors" "fmt" "net/http" "net/http/httptest" @@ -14,6 +15,8 @@ import ( "knowledge-governance/internal/librarysnapshot" ) +var errTestContractChanged = errors.New("source contract changed inside the maintenance window") + type captureWriter struct { running atomic.Bool db *libraryFixture @@ -38,6 +41,85 @@ func (c *captureWriter) Stop(context.Context, string) error { return nil } func (c *captureWriter) Start(context.Context, string) error { c.running.Store(true); return nil } + +// The contract is read before the window and must be re-checked inside it: a +// model or mapping change between those two moments must abort, restore writers +// and seal nothing. +func TestLibraryCaptureRechecksContractInsideFrozenWindow(t *testing.T) { + writer := &captureWriter{db: newLibraryFixture()} + writer.running.Store(true) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/chunks/_mapping": + fmt.Fprint(w, `{"chunks":{"mappings":{"properties":{"knowledge_base_id":{"type":"keyword"},"knowledge_id":{"type":"keyword"},"tenant_id":{"type":"long"}}}}}`) + case "/chunks/_pit": + fmt.Fprint(w, `{"id":"fixed-pit"}`) + case "/_search": + fmt.Fprint(w, `{"timed_out":false,"_shards":{"total":1,"successful":1,"failed":0},"hits":{"total":{"value":0,"relation":"eq"},"hits":[]}}`) + case "/_pit": + fmt.Fprint(w, `{"succeeded":true}`) + default: + http.Error(w, "unexpected", 400) + } + })) + defer server.Close() + search, err := NewSearchExporter(SearchExporterConfig{BaseURL: server.URL, Index: "chunks", PageSize: 10}) + if err != nil { + t.Fatal(err) + } + mapping, err := search.inspectMapping(context.Background()) + if err != nil { + t.Fatal(err) + } + root := t.TempDir() + _ = os.Chmod(root, 0700) + freeze, err := libraryfreeze.Open(root, "test-source", writer) + if err != nil { + t.Fatal(err) + } + defer freeze.Close() + archiveRoot := t.TempDir() + _ = os.Chmod(archiveRoot, 0700) + archive, err := librarysnapshot.Open(archiveRoot, 1024*1024) + if err != nil { + t.Fatal(err) + } + defer archive.Close() + req := LibraryCaptureRequest{JobID: "contract-job", Scope: "internal", + Identity: PostgresIdentity{TenantID: 10000, KnowledgeBaseID: "kb-safe"}, + Compatibility: librarysnapshot.Compatibility{WeKnoraVersion: "v0.7.1", SchemaVersion: 74, + RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "fixture-model", EmbeddingDimension: 1024, + IndexMappingSHA256: mapping.digest}, + CreatedAt: time.Date(2026, 10, 28, 0, 0, 0, 0, time.UTC), MaxDocuments: 100} + called := 0 + req.VerifyContract = func(context.Context) error { + called++ + if writer.running.Load() { + t.Error("contract re-check ran before writers stopped") + } + return errTestContractChanged + } + ready := func(context.Context) error { return nil } + result, err := CaptureLibrary(context.Background(), freeze, archive, dbFixture(t, writer), search, &libraryCandidateReader{}, ready, req) + if err == nil || result.State == "succeeded" || result.SnapshotID != "" { + t.Fatalf("contract change was sealed: %+v %v", result, err) + } + if called != 1 { + t.Fatalf("contract re-check called %d times", called) + } + if !writer.running.Load() { + t.Fatal("writer left stopped after a contract abort") + } + if usage, usageErr := archive.Usage(); usageErr != nil || len(usage) != 0 { + t.Fatalf("aborted capture left an artifact: %v %v", usage, usageErr) + } +} + +func dbFixture(t *testing.T, writer *captureWriter) *libraryFixture { + t.Helper() + return writer.db +} + func TestLibraryCaptureMaintenanceOwnsArchiveAndRestoresOnFailure(t *testing.T) { for _, mode := range []string{"success", "initial-busy", "late-pending", "search-failure", "other-library-busy", "other-library-late-pending"} { t.Run(mode, func(t *testing.T) { diff --git a/web/src/LibraryVersions.tsx b/web/src/LibraryVersions.tsx index ca9b1d3..e350715 100644 --- a/web/src/LibraryVersions.tsx +++ b/web/src/LibraryVersions.tsx @@ -10,7 +10,7 @@ import './libraryVersions.css' // 「整库版本」页：只读展示已封存的整库候选产物（对内/对外各一份最新）。 // 页面不提供封存按钮：封存要停真实写入者并读来源凭据，只能在目标主机用 -// scripts/library-capture.sh 执行。候选 != 验收通过，也 != 可发布。 +// scripts/library-capture.sh 执行。通过完整性检查 != 已发布，授权仍走发布管理。 const stamp = (value: string) => value && Number.isFinite(new Date(value).getTime()) ? new Date(value).toLocaleString('zh-CN', { @@ -74,8 +74,11 @@ export function LibraryVersions() {

- - 共 {snapshots.length} 份 · 归档占用 {formatBytes(page?.stored_bytes ?? 0)} + + 共 {snapshots.length} 份 · 被引用的去重内容 {formatBytes(page?.content_bytes ?? 0)}
+
+ + 磁盘事实（不是可回收空间）：清单 {formatBytes(page.manifest_file_bytes)} · blob 文件{' '} + {formatBytes(page.blob_file_bytes)}（实际分配 {formatBytes(page.blob_allocated_bytes)}）· 未被任何版本引用{' '} + {page.unreferenced_blobs} 个 / {formatBytes(page.unreferenced_bytes)} + +
+ {visible.length === 0 ? (

还没有封存产物。

) : ( @@ -201,7 +212,9 @@ export function LibraryVersions() { {metric(snapshot, 'index_documents')} {metric(snapshot, 'resources')} {metric(snapshot, 'wiki_pages')} - + {formatBytes(snapshot.logical_bytes)} diff --git a/web/src/librarySnapshots.ts b/web/src/librarySnapshots.ts index 493b578..0359039 100644 --- a/web/src/librarySnapshots.ts +++ b/web/src/librarySnapshots.ts @@ -42,7 +42,16 @@ export interface LibrarySnapshotList { reason: string archive_root: string snapshots: LibrarySnapshot[] - stored_bytes: number + // 归档级口径：content_bytes 是跨版本去重后、被版本引用的内容字节； + // 其余字段是文件系统事实，都不是可回收空间。 + versions: number + content_bytes: number + blob_file_bytes: number + blob_allocated_bytes: number + unreferenced_blobs: number + unreferenced_bytes: number + manifest_file_bytes: number + manifest_allocated_bytes: number capture_note: string } @@ -113,7 +122,14 @@ export async function fetchLibrarySnapshots(signal?: AbortSignal): Promise