feat(worker): 四类进度用 manifest 真实总量,未知保持 NULL
build / build (push) Successful in 1m11s

为什么:P02 要求分项计数/字节来自真实清单,不能永远 NULL 或写 0。
This commit is contained in:
姚章浩
2026-10-28 19:55:43 +08:00
parent f722ea8727
commit cef25d92a6
2 changed files with 49 additions and 4 deletions
+8
View File
@@ -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()
}
// 崩溃接管:作业里已经持久化过预备份快照,就直接复用它继续,
// 绝不重新备份——重新备份会把"半写状态"当成新基线,覆盖真正可回滚的那份。
+41 -4
View File
@@ -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"