fix(library): 幂等重放、跨版本去重用量与契约复核
build / build (push) Failing after 15s

- 同作业号重跑读持久日志+已封存清单按首次请求重放(source_read=false,
  writers_stopped=false),换范围/绑定库或清单校验失败即拒绝
- 归档用量按跨版本去重统计内容字节,清单/未引用 blob/实际分配单列,
  不再用各版本 unique 之和冒充归档占用
- 冻结窗口内复核来源契约,不一致即中止;拒绝组外同镜像运行容器
- 补齐整库只读接口的 OpenAPI 契约
- 按用户确认取消 A06:文档/页面/接口不再把效果测试列为封版门槛
This commit is contained in:
姚章浩
2026-10-28 12:01:09 +08:00
parent 20fc40a147
commit c5e033dd0e
15 changed files with 839 additions and 50 deletions
+17
View File
@@ -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
+53
View File
@@ -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": [
+162 -28
View File
@@ -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
}
+126
View File
@@ -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")
}
}
}
+13 -7
View File
@@ -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 页),这份快照只说明"它当时确实是空的";
- 生产备份/恢复、双闸授权、整库同步与失败续作都没有接线,本文不涉及;
- 上游父子分块漏正文索引的风险仍在:本次候选通过是因为库里现有数据已补齐,**以后再重解析**仍可能漏项,届时封存会因覆盖核验失败而阻断(这是设计如此)。
- 生产独立预备份、恢复演练、双闸授权、整库同步协调与失败续作都没有接线,本文不涉及;
- 从页面发起封存、持久作业互斥、启动恢复还没有接线(封存目前只能在目标主机执行命令);
- 上游父子分块漏正文索引的风险仍在:本次通过是因为库里现有数据已补齐,**以后再重解析**仍可能漏项,届时封存会因覆盖核验失败而阻断(这是设计如此)。
+29 -5
View File
@@ -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)
+7 -4
View File
@@ -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)
+19
View File
@@ -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 {
+131
View File
@@ -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 {
@@ -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")
}
}
@@ -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
@@ -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) {
+17 -4
View File
@@ -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() {
</p>
</div>
<div className="library-versions-actions">
<span className="library-versions-usage">
共 {snapshots.length} 份 · 归档占用 {formatBytes(page?.stored_bytes ?? 0)}
<span
className="library-versions-usage"
title="跨版本去重后被版本引用的内容字节;不是磁盘占用,也不是可回收空间"
>
共 {snapshots.length} 份 · 被引用的去重内容 {formatBytes(page?.content_bytes ?? 0)}
</span>
<button type="button" onClick={() => void load()} disabled={loading}>
{loading ? '刷新中…' : '刷新'}
@@ -173,6 +176,14 @@ export function LibraryVersions() {
<span className="library-versions-note">{page.capture_note}</span>
</div>
<div className="library-versions-toolbar library-versions-accounting">
<span>
磁盘事实(不是可回收空间):清单 {formatBytes(page.manifest_file_bytes)} · blob 文件{' '}
{formatBytes(page.blob_file_bytes)}(实际分配 {formatBytes(page.blob_allocated_bytes)})· 未被任何版本引用{' '}
{page.unreferenced_blobs} 个 / {formatBytes(page.unreferenced_bytes)}
</span>
</div>
{visible.length === 0 ? (
<p className="library-versions-empty">还没有封存产物。</p>
) : (
@@ -201,7 +212,9 @@ export function LibraryVersions() {
<td>{metric(snapshot, 'index_documents')}</td>
<td>{metric(snapshot, 'resources')}</td>
<td>{metric(snapshot, 'wiki_pages')}</td>
<td title={`逻辑 ${snapshot.logical_bytes} 字节 · 去重后 ${snapshot.unique_bytes} 字节`}>
<td
title={`本版逻辑 ${snapshot.logical_bytes} 字节 · 本版内去重 ${snapshot.unique_bytes} 字节(跨版本去重见上方合计)`}
>
{formatBytes(snapshot.logical_bytes)}
</td>
<td>
+18 -2
View File
@@ -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<Libra
reason: text(value.reason),
archive_root: text(value.archive_root),
snapshots,
stored_bytes: num(value.stored_bytes),
versions: num(value.versions),
content_bytes: num(value.content_bytes),
blob_file_bytes: num(value.blob_file_bytes),
blob_allocated_bytes: num(value.blob_allocated_bytes),
unreferenced_blobs: num(value.unreferenced_blobs),
unreferenced_bytes: num(value.unreferenced_bytes),
manifest_file_bytes: num(value.manifest_file_bytes),
manifest_allocated_bytes: num(value.manifest_allocated_bytes),
capture_note: text(value.capture_note),
}
}
+6
View File
@@ -165,6 +165,12 @@
max-width: 70ch;
line-height: 1.5;
}
.library-versions-accounting {
justify-content: flex-start;
font-size: 12px;
color: var(--ink-mute, #64748b);
line-height: 1.6;
}
.library-versions-empty {
margin: 0;
padding: 24px;