From cef25d92a62ace37c5ecdeed07ba270dbe50c6cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A7=9A=E7=AB=A0=E6=B5=A9?= Date: Wed, 28 Oct 2026 19:55:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(worker):=20=E5=9B=9B=E7=B1=BB=E8=BF=9B?= =?UTF-8?q?=E5=BA=A6=E7=94=A8=20manifest=20=E7=9C=9F=E5=AE=9E=E6=80=BB?= =?UTF-8?q?=E9=87=8F=EF=BC=8C=E6=9C=AA=E7=9F=A5=E4=BF=9D=E6=8C=81=20NULL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么:P02 要求分项计数/字节来自真实清单,不能永远 NULL 或写 0。 --- cmd/worker/library_handlers.go | 8 ++++++ cmd/worker/main.go | 45 +++++++++++++++++++++++++++++++--- 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/cmd/worker/library_handlers.go b/cmd/worker/library_handlers.go index 2844fff..13873dc 100644 --- a/cmd/worker/library_handlers.go +++ b/cmd/worker/library_handlers.go @@ -13,6 +13,7 @@ import ( "knowledge-governance/internal/governance" "knowledge-governance/internal/jobs" "knowledge-governance/internal/libraryrelease" + "knowledge-governance/internal/librarysnapshot" postgresstore "knowledge-governance/internal/platform/postgres" ) @@ -37,6 +38,13 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a now := time.Now().UTC() _ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobRunning, "prebackup", "", "", "", "", now) recorder := libraryStageRecorder{store: store, jobID: payload.LibraryJobID, scope: payload.Scope} + // 四类进度的真实总量来自快照清单(manifest):读不到就保持 NULL,不写 0。 + if archive, openErr := librarysnapshot.Open(config.LibraryArchiveRoot, 256*1024*1024); openErr == nil { + if manifest, loadErr := archive.Load(payload.TargetSnapshotID); loadErr == nil { + recorder.totals = stageSizesFrom(manifest) + } + archive.Close() + } // 崩溃接管:作业里已经持久化过预备份快照,就直接复用它继续, // 绝不重新备份——重新备份会把"半写状态"当成新基线,覆盖真正可回滚的那份。 diff --git a/cmd/worker/main.go b/cmd/worker/main.go index a35737c..839840d 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -17,6 +17,7 @@ import ( "knowledge-governance/internal/governance" "knowledge-governance/internal/jobs" "knowledge-governance/internal/libraryagent" + "knowledge-governance/internal/librarysnapshot" postgresstore "knowledge-governance/internal/platform/postgres" "knowledge-governance/internal/platform/weknora" "knowledge-governance/internal/versionhistory" @@ -28,19 +29,55 @@ import ( // libraryStageRecorder 把执行阶段写进持久作业,页面刷新/worker 重启后仍可复现。 type libraryStageRecorder struct { - store *postgresstore.Store - jobID string - scope string + store *postgresstore.Store + jobID string + scope string + totals map[string]stageSize // mappingDigest 是本次读到的生产索引映射,随预备份一起登记成版本证据。 mappingDigest string } func (r libraryStageRecorder) Stage(ctx context.Context, stage, state, detail string) { - if err := r.store.UpdateLibraryStage(ctx, r.jobID, stage, state, nil, nil, nil, nil, detail, time.Now().UTC()); err != nil { + // 真实总量来自快照 manifest:已知就写真实数字,未知就保持 NULL(不写 0、不编百分比)。 + var total, completed, bytesTotal, bytesDone *int64 + if size, ok := r.totals[stage]; ok { + total, bytesTotal = int64Ptr(size.count), int64Ptr(size.bytes) + if state == "succeeded" { + completed, bytesDone = total, bytesTotal + } + } + if err := r.store.UpdateLibraryStage(ctx, r.jobID, stage, state, total, completed, bytesTotal, bytesDone, + detail, time.Now().UTC()); err != nil { log.Printf("library stage %s/%s update failed: %v", r.jobID, stage, err) } } +func int64Ptr(value int64) *int64 { return &value } + +// stageSize 是某一类数据在快照里的真实总量(行数/字节)。 +type stageSize struct { + count int64 + bytes int64 +} + +// stageSizesFrom 从 manifest 汇总四类进度的真实总量: +// database/resources/index/wiki 各自的行数与字节,全部来自归档部件清单。 +func stageSizesFrom(manifest librarysnapshot.Manifest) map[string]stageSize { + sizes := map[string]stageSize{} + kindFor := map[string]string{"database": "database", "resource": "resources", "index": "index", "wiki": "wiki"} + for _, part := range manifest.Parts { + stage, ok := kindFor[part.Kind] + if !ok { + continue + } + current := sizes[stage] + current.count += part.Count + current.bytes += part.Bytes + sizes[stage] = current + } + return sizes +} + // libraryCaptureJobType is shared with the API which enqueues it. const libraryCaptureJobType = "library.capture"