父审发现三处会让目标卡死或状态对不上实际入口,逐条修: 1. 登记成功但没来得及收尾:崩溃重入会撞上"已有生效版本"直接失败, 目标留在维护里、占用也不释放。现在按「本作业+同一快照」识别为自己的 版本,只做收尾(恢复服务/开入口/释放两把占用),不重新捕获、不重写; 别人的版本仍然拒绝。捕获完成但登记失败则明确失败并写"未对目标写入"。 2. 重入覆盖原运行状态:崩溃时服务已被停住,再读一次只会读到 stopped, 写回去会让恢复阶段永远不启动写入者。AcquireInstance/LibraryMaintenance 在 owner 相同的重入时保留已持久化原状态;新作业接管已释放窗口用自己的观察。 真实 PostgreSQL 集成用例覆盖(先红后绿)。 3. 版本登记移进维护窗口:捕获/校验/登记要么一起成立、要么目标原样恢复; 记录失败时窗口已开放,治理状态同步释放,不再"显示维护中、实际已开"。 索引映射读不到时拒绝登记,不再用空摘要写证据。 顺带: - 基线窗口统一收尾抽成 settleBaselineWindow,四种结局各有测试(成功/健康失败/ held 保持维护/已恢复则释放),并用真实 DB 用例证明 baseline 与 sync/restore 共用写互斥、幂等重试不产生第二个作业。 - 新增 cmd/librarydrill:维护隔离 → 整库预备份 → 一致性校验 → canary 漂移检测 → 按预备份恢复 → 逐类复核 → 按原状态恢复服务。用与 worker 相同的 LIBRARY_* 配置,绝不触碰封存源。 - 目标/预备份/维护栈装配收成 internal/librarystack 一份(worker 与演练共用)。 验证:go build/vet/test 全过;TEST_DATABASE_URL 指向临时 PostgreSQL 跑 TestMaintenanceReentryKeepsOriginalServiceState(修复前失败:got app=stopped) 与 TestLibraryWriteMutexCoversBaselineJobs;本地门禁 15/15。
This commit is contained in:
@@ -1,5 +1,39 @@
|
||||
# Changelog
|
||||
|
||||
## 开发快照(基线崩溃续作、维护原状态保留与真实恢复演练工具)
|
||||
|
||||
### Fixed
|
||||
|
||||
- **基线作业"登记成功但没来得及收尾"不再被判死**:原来崩溃重入会在
|
||||
"已有生效版本"检查处直接失败,把目标留在维护里、占用也不释放。
|
||||
现在按「本作业 + 同一快照」识别为**自己的**版本:只做收尾(恢复服务、开入口、
|
||||
释放两把占用),不重新捕获、不重写版本;别人登记的生效版本仍然一律拒绝。
|
||||
捕获完成但登记失败的作业会明确失败并写"未对目标写入",不会假装成功。
|
||||
- **同一作业重入不再覆盖第一次记下的原运行状态**:崩溃接管时目标已被停住,
|
||||
再读一次只会读到 `stopped`,写回去会让恢复阶段以为写入者本来就没跑、
|
||||
永远不启动它。现在 `AcquireInstanceMaintenance` / `AcquireLibraryMaintenance`
|
||||
在 owner 相同的重入时保留已持久化的原状态;新作业接管已释放窗口仍用自己读到的值。
|
||||
(真实 PostgreSQL 集成用例覆盖,非内存假实现。)
|
||||
- **基线的版本登记移进维护窗口**:捕获、校验、登记三件事要么一起成立、要么目标
|
||||
原样恢复;记录失败时窗口已按原状态开放,治理记录同步释放,**不再出现
|
||||
"显示保持维护、实际入口已开"**。
|
||||
- **索引映射读不到时拒绝登记**:不再 `_` 忽略错误后用空摘要写进基线证据。
|
||||
- 基线窗口的统一收尾抽成 `settleBaselineWindow`:版本已登记走与同步相同的成功边界;
|
||||
窗口仍 held 才标 `needs_restore`;窗口已恢复则释放占用并失败。
|
||||
|
||||
### Added
|
||||
|
||||
- **`cmd/librarydrill`:目标实例真实恢复演练**(维护隔离 → 整库预备份 → 一致性校验
|
||||
→ canary 漂移检测 → 按预备份恢复 → 逐类复核 → 按原状态恢复服务)。它使用与
|
||||
worker 完全相同的 `LIBRARY_*` 配置与维护协调器,绝不触碰封存源;任一步失败都
|
||||
保持维护并写报告,不伪造通过。
|
||||
- `internal/librarystack`:目标/预备份/维护栈的唯一装配点(同步、恢复、基线、演练共用)。
|
||||
|
||||
### Changed
|
||||
|
||||
- 基线作业的"已有生效版本"判定改为三种状态(own / foreign / none),并在
|
||||
阶段里写明是否走"只收尾"续作。
|
||||
|
||||
## 开发快照(首次发布入口:同步授权、目标权威基线与维护能力)
|
||||
|
||||
### Added
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
// command librarydrill 在目标实例上做一次**真实**的恢复演练:
|
||||
//
|
||||
// 维护隔离(关入口/停写入者/证明排空)
|
||||
// → 捕获目标整库(真预备份,落归档)
|
||||
// → 校验目标 == 预备份(基线一致)
|
||||
// → 写入一个 canary(证明"校验确实能发现内容变化")
|
||||
// → 按预备份恢复
|
||||
// → 逐表/索引/资源/Wiki 再校验一次,必须完全一致
|
||||
// → 关窗口(按原运行状态恢复服务、开入口、验证健康)
|
||||
//
|
||||
// 它只读/写**被测目标自己**,绝不触碰封存源;所有凭据来自进程环境
|
||||
// (与 worker 同一份 LIBRARY_* 配置),不进命令行、不打印。
|
||||
// 任一步失败都不伪装成功:窗口保持维护、报告写明失败阶段。
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"knowledge-governance/internal/appconfig"
|
||||
"knowledge-governance/internal/libraryfreeze"
|
||||
"knowledge-governance/internal/libraryrelease"
|
||||
"knowledge-governance/internal/librarystack"
|
||||
)
|
||||
|
||||
type drillVerification struct {
|
||||
SnapshotID string `json:"snapshot_id"`
|
||||
Verified bool `json:"verified"`
|
||||
Error string `json:"error,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
type drillReport struct {
|
||||
Status string `json:"status"` // passed | failed | maintenance_held
|
||||
Scope string `json:"scope"`
|
||||
Target string `json:"target"`
|
||||
SnapshotID string `json:"snapshot_id"`
|
||||
Project string `json:"project"`
|
||||
Counts map[string]int64 `json:"counts"`
|
||||
BaselineCheck *drillVerification `json:"baseline_check"`
|
||||
DriftCheck *drillVerification `json:"drift_check"` // 必须失败:证明确实能发现差异
|
||||
RestoreCheck *drillVerification `json:"restore_check"`
|
||||
Canary string `json:"canary"`
|
||||
Failure string `json:"failure,omitempty"`
|
||||
Stage string `json:"stage,omitempty"`
|
||||
StartedAt time.Time `json:"started_at"`
|
||||
FinishedAt time.Time `json:"finished_at"`
|
||||
Until map[string]string `json:"until,omitempty"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
scope := flag.String("scope", "internal", "internal|external")
|
||||
reportPath := flag.String("report", "", "write the JSON report here (0600)")
|
||||
jobID := flag.String("job", "", "maintenance job id (default drill-<scope>-<timestamp>)")
|
||||
flag.Parse()
|
||||
report, err := run(*scope, *jobID)
|
||||
if *reportPath != "" && report != nil {
|
||||
if writeErr := writeReport(*reportPath, report); writeErr != nil {
|
||||
log.Printf("drill report write failed: %v", writeErr)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
log.Fatalf("drill failed: %v", err)
|
||||
}
|
||||
log.Printf("drill passed: snapshot=%s", report.SnapshotID)
|
||||
}
|
||||
|
||||
func run(scope, jobID string) (*drillReport, error) {
|
||||
if scope != "internal" && scope != "external" {
|
||||
return nil, errors.New("scope must be internal or external")
|
||||
}
|
||||
config, err := appconfig.Load(os.Getenv)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load config: %w", err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
report := &drillReport{Status: "failed", Scope: scope, Project: config.LibraryTargetProject,
|
||||
Target: config.LibraryTargetInstance, StartedAt: time.Now().UTC()}
|
||||
fail := func(stage string, cause error, held bool) (*drillReport, error) {
|
||||
report.Stage = stage
|
||||
report.Failure = cause.Error()
|
||||
report.FinishedAt = time.Now().UTC()
|
||||
if held {
|
||||
report.Status = "maintenance_held"
|
||||
}
|
||||
return report, cause
|
||||
}
|
||||
|
||||
target, backups, err := librarystack.OpenBaselineStack(ctx, config, scope)
|
||||
if err != nil {
|
||||
return fail("open_target", err, false)
|
||||
}
|
||||
defer target.Close()
|
||||
manager, err := librarystack.OpenMaintenance(config)
|
||||
if err != nil {
|
||||
return fail("open_maintenance", err, false)
|
||||
}
|
||||
defer manager.Close()
|
||||
if err := manager.VerifyTopology(ctx); err != nil {
|
||||
return fail("topology", err, false)
|
||||
}
|
||||
coordinator, control, err := manager.Open(config.LibraryTargetJournalRoot)
|
||||
if err != nil {
|
||||
return fail("open_coordinator", err, false)
|
||||
}
|
||||
defer control.Close()
|
||||
defer coordinator.Close()
|
||||
|
||||
if jobID == "" {
|
||||
jobID = "drill-" + scope + "-" + time.Now().UTC().Format("20060102T150405")
|
||||
}
|
||||
var (
|
||||
snapshotID string
|
||||
counts map[string]int64
|
||||
canary string
|
||||
baseline drillVerification
|
||||
drift drillVerification
|
||||
restored drillVerification
|
||||
)
|
||||
work := func(ctx context.Context, window *libraryfreeze.Window) (string, error) {
|
||||
start := time.Now()
|
||||
current, err := target.Current(ctx)
|
||||
if err != nil {
|
||||
return "", errors.New("读取目标当前整库状态失败:" + err.Error())
|
||||
}
|
||||
counts = current.Counts
|
||||
report.Counts = counts
|
||||
// 1) 真预备份(落归档、内容寻址、导出后复核内容未变)。
|
||||
backup, err := backups.Prepare(ctx, "", current)
|
||||
if err != nil {
|
||||
return "", errors.New("捕获目标整库失败:" + err.Error())
|
||||
}
|
||||
snapshotID = backup.SnapshotID
|
||||
report.SnapshotID = snapshotID
|
||||
if err := window.SetBackupSnapshot(ctx, snapshotID); err != nil {
|
||||
return "", errors.New("记录预备份身份失败:" + err.Error())
|
||||
}
|
||||
log.Printf("drill: snapshot captured snapshot=%s bytes=%d in=%s", snapshotID, backup.Bytes, time.Since(start))
|
||||
|
||||
// 2) 基线校验:预备份必须与目标完全一致。
|
||||
baseline = drillVerification{SnapshotID: snapshotID, At: time.Now().UTC()}
|
||||
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err != nil {
|
||||
baseline.Error = err.Error()
|
||||
report.BaselineCheck = &baseline
|
||||
return "", errors.New("预备份与目标不一致(预备份本身不可信):" + err.Error())
|
||||
}
|
||||
baseline.Verified = true
|
||||
report.BaselineCheck = &baseline
|
||||
|
||||
// 3) canary:改一行真实内容,校验**必须**失败(否则"校验"是摆设)。
|
||||
canary = "drill-canary-" + time.Now().UTC().Format("20060102T150405")
|
||||
identity := target.Identity()
|
||||
tag, err := target.Pool().Exec(ctx, `
|
||||
UPDATE knowledges SET title = title || ' [' || $3 || ']'
|
||||
WHERE id = (SELECT id FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2 AND deleted_at IS NULL
|
||||
ORDER BY id LIMIT 1)`, identity.TenantID, identity.KnowledgeBaseID, canary)
|
||||
if err != nil {
|
||||
return "", errors.New("写入 canary 失败:" + err.Error())
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return "", errors.New("目标里没有可改写的文档,无法做漂移检测(目标不应为空)")
|
||||
}
|
||||
report.Canary = canary
|
||||
drift = drillVerification{SnapshotID: snapshotID, At: time.Now().UTC()}
|
||||
if verifyErr := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); verifyErr == nil {
|
||||
drift.Error = "写入 canary 后校验仍然通过:校验没有真正比对内容"
|
||||
report.DriftCheck = &drift
|
||||
return "", errors.New("写入 canary 后校验仍然通过:校验没有真正比对内容,拒绝把这次演练当成通过")
|
||||
} else {
|
||||
drift.Error = verifyErr.Error()
|
||||
}
|
||||
report.DriftCheck = &drift
|
||||
log.Printf("drill: drift detected as expected (%s)", drift.Error)
|
||||
|
||||
// 4) 按预备份恢复。
|
||||
if err := target.Restore(ctx, libraryrelease.BackupRef{ID: "backup-" + snapshotID, SnapshotID: snapshotID}); err != nil {
|
||||
return "", errors.New("按预备份恢复失败:" + err.Error())
|
||||
}
|
||||
// 5) 恢复后逐类复核:必须与预备份完全一致。
|
||||
restored = drillVerification{SnapshotID: snapshotID, At: time.Now().UTC()}
|
||||
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: snapshotID}); err != nil {
|
||||
restored.Error = err.Error()
|
||||
report.RestoreCheck = &restored
|
||||
return "", errors.New("恢复后内容仍与预备份不一致:" + err.Error())
|
||||
}
|
||||
restored.Verified = true
|
||||
report.RestoreCheck = &restored
|
||||
log.Printf("drill: restored and verified snapshot=%s", snapshotID)
|
||||
return snapshotID, nil
|
||||
}
|
||||
|
||||
window, windowErr := coordinator.RunWindow(ctx, jobID,
|
||||
func(ctx context.Context) error { return manager.VerifyRunning(ctx) }, work,
|
||||
// 目标内容已被本演练改写过:只有"恢复并校验通过"才允许重新开放。
|
||||
func(error) bool { return restored.Verified })
|
||||
|
||||
report.Until = map[string]string{"job": jobID, "window_state": window.State}
|
||||
if windowErr != nil || window.State != "succeeded" {
|
||||
cause := windowErr
|
||||
if cause == nil {
|
||||
cause = fmt.Errorf("维护窗口状态 %s(未成功)", window.State)
|
||||
}
|
||||
if window.State == "held" {
|
||||
return fail("window", cause, true)
|
||||
}
|
||||
return fail("window", cause, false)
|
||||
}
|
||||
report.Status = "passed"
|
||||
report.FinishedAt = time.Now().UTC()
|
||||
return report, nil
|
||||
}
|
||||
|
||||
func writeReport(path string, report *drillReport) error {
|
||||
raw, err := json.MarshalIndent(report, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if dir := filepath.Dir(path); dir != "" {
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return os.WriteFile(path, append(raw, '\n'), 0o600)
|
||||
}
|
||||
+181
-110
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
@@ -23,12 +22,85 @@ type libraryBaselinePayload struct {
|
||||
Scope string `json:"scope"`
|
||||
}
|
||||
|
||||
// baselineVersionState 回答"这个 scope 现在有没有生效版本、是不是本作业登记的"。
|
||||
//
|
||||
// 三种情况必须分开,否则崩溃续作会被自己的版本挡住:
|
||||
// - own:本作业 + 同一快照已经登记过 → 只收尾,绝不重复捕获或重写版本;
|
||||
// - foreign:别人登记的生效版本 → 一律拒绝(不能用基线覆盖别人的版本);
|
||||
// - none:还没有生效版本 → 正常建基线。
|
||||
type baselineVersionState struct {
|
||||
Own bool
|
||||
Foreign *governance.LibraryVersion
|
||||
}
|
||||
|
||||
func classifyBaselineVersions(versions []governance.LibraryVersion, jobID, snapshotID string) baselineVersionState {
|
||||
for index := range versions {
|
||||
item := versions[index]
|
||||
if item.Kind != governance.LibraryVersionProductionEffective || item.SupersededAt != nil {
|
||||
continue
|
||||
}
|
||||
if jobID != "" && snapshotID != "" && item.SourceJobID == jobID && item.SnapshotID == snapshotID {
|
||||
return baselineVersionState{Own: true}
|
||||
}
|
||||
foreign := item
|
||||
return baselineVersionState{Foreign: &foreign}
|
||||
}
|
||||
return baselineVersionState{}
|
||||
}
|
||||
|
||||
// settleBaselineWindow 是基线窗口的**唯一**收尾入口,负责让治理状态与实际入口一致:
|
||||
//
|
||||
// - 版本已登记(本次刚写,或崩溃前由本作业写过)→ 走与同步一样的成功边界:
|
||||
// 目标健康、入口开放、两把维护锁释放之后才暴露 succeeded;
|
||||
// - 没有登记且窗口仍是 held(服务没回来)→ 保持维护并标 needs_restore,
|
||||
// 只有"按预备份恢复"能把它收掉;
|
||||
// - 没有登记但窗口已按原状态恢复 → 释放占用并把作业标失败,明确写"未对目标写入",
|
||||
// 绝不留着"显示维护中、实际已开放"的状态。
|
||||
func settleBaselineWindow(ctx context.Context, store libraryMaintenanceStore, health healthChecker,
|
||||
jobID, scope, instance, baselineID string, recorded bool, windowState string, windowErr error) error {
|
||||
if recorded {
|
||||
finishedAt := time.Now().UTC()
|
||||
return finishLibrarySuccess(ctx, store, health, instance, scope, jobID,
|
||||
successBoundary{Ready: true, Backup: baselineID, FinishedAt: finishedAt})
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
message := "未建立基线:维护窗口没有完成捕获与校验"
|
||||
if windowErr != nil {
|
||||
message += ":" + windowErr.Error()
|
||||
}
|
||||
if windowState == "held" {
|
||||
note := message + "(目标仍被停住、入口未恢复,保持维护)"
|
||||
_ = store.UpdateLibraryMaintenanceState(ctx, scope, jobID, "needs_restore", note, now)
|
||||
_ = store.UpdateInstanceMaintenanceState(ctx, instance, jobID, "needs_restore", note, now)
|
||||
_ = store.AdvanceLibraryJob(ctx, jobID, governance.LibraryJobFailed, "capture", note,
|
||||
"目标仍在维护:先按预备份恢复并校验", baselineID, "", now)
|
||||
return errors.New(note)
|
||||
}
|
||||
// 窗口已按原状态开放(本流程对目标内容只读):占用必须跟着释放,
|
||||
// 不能留下"界面说维护中、实际入口已开"的记录。
|
||||
if releaseErr := releaseWindowLocks(ctx, store, instance, scope, jobID,
|
||||
"未登记基线,目标内容未被写入,窗口已按原状态恢复", now); releaseErr != nil {
|
||||
note := message + ";维护占用释放失败:" + releaseErr.Error()
|
||||
_ = store.AdvanceLibraryJob(ctx, jobID, governance.LibraryJobFailed, "capture", note,
|
||||
"维护记录与实际入口不一致:先人工核对再重试", baselineID, "", now)
|
||||
return errors.New(note)
|
||||
}
|
||||
_ = store.AdvanceLibraryJob(ctx, jobID, governance.LibraryJobFailed, "capture", message,
|
||||
"未对目标执行任何写入;可直接重试建立基线", baselineID, "", now)
|
||||
return errors.New(message)
|
||||
}
|
||||
|
||||
// runLibraryBaseline 给"非空但没有权威版本"的目标建立基线。
|
||||
//
|
||||
// 真实链路:维护隔离(关入口/停写入者/证明排空)→ 把目标整库捕获成归档快照
|
||||
// → 用同一套 Verify 证明目标与这份快照逐类一致 → 登记 production_effective。
|
||||
// 它不写目标内容、不清空目标、不猜版本:基线只是"目标现在是什么"的忠实记录,
|
||||
// 因此失败时重新开放目标永远是安全的(allowRestore 恒 true)。
|
||||
// 真实链路(全部在同一个维护窗口内):关入口/停写入者/证明排空 → 把目标整库捕获成
|
||||
// 归档快照 → 用同一套 Verify 证明目标与这份快照逐类一致 → 读目标索引映射 →
|
||||
// 登记 production_effective → 关窗口(恢复服务、开入口)→ 释放两把占用 → 作业成功。
|
||||
//
|
||||
// 崩溃语义:
|
||||
// - 捕获后、登记前被杀:作业行里已有快照 ID,重入时**不重新捕获**,直接在窗口内
|
||||
// 复核这份快照再登记;
|
||||
// - 登记后、收尾前被杀:重入识别出"本作业+同一快照"已生效,只做收尾
|
||||
// (恢复服务/开入口/释放占用),不重复捕获、不重写版本;别人的版本仍然拒绝。
|
||||
func runLibraryBaseline(ctx context.Context, store *postgresstore.Store, config appconfig.Config, job jobs.Job) error {
|
||||
var payload libraryBaselinePayload
|
||||
if err := json.Unmarshal(job.Payload, &payload); err != nil || payload.LibraryJobID == "" ||
|
||||
@@ -40,21 +112,31 @@ func runLibraryBaseline(ctx context.Context, store *postgresstore.Store, config
|
||||
fail := func(message, stage string) error {
|
||||
return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, message, stage)
|
||||
}
|
||||
existing, jobErr := store.LibraryJob(ctx, payload.LibraryJobID)
|
||||
actor := ""
|
||||
persistedSnapshot := ""
|
||||
if jobErr == nil {
|
||||
actor = existing.Actor
|
||||
persistedSnapshot = existing.SnapshotID
|
||||
}
|
||||
_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobRunning, "capture", "", "", "", "", now)
|
||||
|
||||
// 已有生效版本时不再建基线:基线只解决"目标非空但无权威版本"的首次发布,
|
||||
// 版本切换必须走同步/恢复(它们有预备份、回滚与授权)。
|
||||
// 版本判定:别人的生效版本一律拒绝;本作业已登记过的快照只收尾。
|
||||
versions, versionErr := store.LibraryVersions(ctx, payload.Scope, 50)
|
||||
if versionErr != nil {
|
||||
return fail("读不到版本记录,无法确认是否需要建立基线;未对目标执行任何操作", "capture")
|
||||
}
|
||||
for _, item := range versions {
|
||||
if item.Kind == governance.LibraryVersionProductionEffective && item.SupersededAt == nil {
|
||||
return fail("该目标已有生效版本 "+item.SnapshotID+",不需要再建基线;换版本请走同步或恢复", "capture")
|
||||
}
|
||||
state := classifyBaselineVersions(versions, payload.LibraryJobID, persistedSnapshot)
|
||||
if state.Foreign != nil {
|
||||
return fail("该目标已有生效版本 "+shortSnapshotID(state.Foreign.SnapshotID)+
|
||||
"(作业 "+state.Foreign.SourceJobID+"),不需要再建基线;换版本请走同步或恢复", "capture")
|
||||
}
|
||||
finalizeOnly := state.Own
|
||||
if finalizeOnly {
|
||||
recorder.Stage(ctx, "finalize", "running", "续作:本作业已登记同一份基线快照,只做收尾(不重新捕获)")
|
||||
}
|
||||
|
||||
if !config.ProductionReleaseRequestsEnabled {
|
||||
if !finalizeOnly && !config.ProductionReleaseRequestsEnabled {
|
||||
return fail("生产写入闸口关闭(PRODUCTION_RELEASE_REQUESTS_ENABLED=false):建立基线要动目标运行状态,闸口关闭时不执行", "capture")
|
||||
}
|
||||
target, backups, wireErr := buildLibraryBaselineStack(ctx, config, payload.Scope)
|
||||
@@ -77,6 +159,7 @@ func runLibraryBaseline(ctx context.Context, store *postgresstore.Store, config
|
||||
}
|
||||
identity := manager.Target()
|
||||
// 实例级占用:internal/external 动的是同一批容器,必须按实例互斥。
|
||||
// 同一作业重入时 store 保留第一次记下的原运行状态(否则会把"已停"当原状)。
|
||||
if _, instanceErr := store.AcquireInstanceMaintenance(ctx, postgresstore.LibraryInstanceMaintenance{
|
||||
Instance: identity.Instance, OwnerJobID: payload.LibraryJobID, OwnerScope: payload.Scope,
|
||||
TargetFingerprint: identity.Fingerprint, Level: librarymaintenance.LevelInstance,
|
||||
@@ -89,7 +172,7 @@ func runLibraryBaseline(ctx context.Context, store *postgresstore.Store, config
|
||||
if _, acquireErr := store.AcquireLibraryMaintenance(ctx, postgresstore.LibraryMaintenance{
|
||||
Scope: payload.Scope, OwnerJobID: payload.LibraryJobID,
|
||||
TargetIdentity: identity.Fingerprint,
|
||||
Level: "instance",
|
||||
Level: librarymaintenance.LevelInstance,
|
||||
TargetFingerprint: identity.Fingerprint,
|
||||
Services: identity.Writers,
|
||||
IngressState: "unknown",
|
||||
@@ -111,117 +194,105 @@ func runLibraryBaseline(ctx context.Context, store *postgresstore.Store, config
|
||||
defer control.Close()
|
||||
defer coordinator.Close()
|
||||
|
||||
var baselineID string
|
||||
var counts map[string]int64
|
||||
verified := false
|
||||
var (
|
||||
baselineID = persistedSnapshot
|
||||
counts map[string]int64
|
||||
recorded = finalizeOnly
|
||||
)
|
||||
work := func(ctx context.Context, window *libraryfreeze.Window) (string, error) {
|
||||
current, currentErr := target.Current(ctx)
|
||||
if currentErr != nil {
|
||||
return "", errors.New("读取目标当前整库状态失败:" + currentErr.Error())
|
||||
if finalizeOnly {
|
||||
// 版本已在窗口内校验并登记过:不再捕获、不再重写,收尾由下方统一处理。
|
||||
return baselineID, nil
|
||||
}
|
||||
counts = map[string]int64{}
|
||||
for table, count := range current.Counts {
|
||||
counts[table] = count
|
||||
if baselineID == "" {
|
||||
current, currentErr := target.Current(ctx)
|
||||
if currentErr != nil {
|
||||
return "", errors.New("读取目标当前整库状态失败:" + currentErr.Error())
|
||||
}
|
||||
counts = map[string]int64{}
|
||||
for table, count := range current.Counts {
|
||||
counts[table] = count
|
||||
}
|
||||
// 预备份存储在导出后用同一份指纹复核"导出期间内容没变",
|
||||
// 所以这里把窗口内读到的 state 交进去,而不是事后再读一次。
|
||||
backup, backupErr := backups.Prepare(ctx, "", current)
|
||||
if backupErr != nil {
|
||||
return "", errors.New("捕获目标整库失败:" + backupErr.Error())
|
||||
}
|
||||
baselineID = backup.SnapshotID
|
||||
// 先把"这次窗口的基线快照"落盘:进程在这里被杀时,恢复器知道该回到哪一份,
|
||||
// 重入也据此跳过重新捕获。
|
||||
if markErr := window.SetBackupSnapshot(ctx, baselineID); markErr != nil {
|
||||
return "", errors.New("记录基线快照身份失败:" + markErr.Error())
|
||||
}
|
||||
if persistErr := store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobRunning, "",
|
||||
"", "", baselineID, "", time.Now().UTC()); persistErr != nil {
|
||||
return "", errors.New("持久化基线快照引用失败:" + persistErr.Error())
|
||||
}
|
||||
recorder.Stage(ctx, "capture", "succeeded", "已封存目标整库快照 "+baselineID)
|
||||
} else {
|
||||
recorder.Stage(ctx, "capture", "skipped", "复用作业已持久化的基线快照 "+baselineID)
|
||||
}
|
||||
// 预备份存储在导出后用同一份指纹复核"导出期间内容没变",所以这里
|
||||
// 必须把窗口内读到的 state 交进去,而不是事后再读一次。
|
||||
backup, backupErr := backups.Prepare(ctx, "", current)
|
||||
if backupErr != nil {
|
||||
return "", errors.New("捕获目标整库失败:" + backupErr.Error())
|
||||
if baselineID == "" {
|
||||
return "", errors.New("没有可校验的基线快照")
|
||||
}
|
||||
baselineID = backup.SnapshotID
|
||||
// 先把"这次窗口的基线快照"落盘:进程在这里被杀时,恢复器知道该回到哪一份,
|
||||
// 而不是把目标留在停机状态且没有任何可核对的基线。
|
||||
if markErr := window.SetBackupSnapshot(ctx, baselineID); markErr != nil {
|
||||
return "", errors.New("记录基线快照身份失败:" + markErr.Error())
|
||||
}
|
||||
// 作业行也要立刻带上这份快照身份:worker 被杀后接管/按预备份恢复都靠它,
|
||||
// 否则"目标停着、却不知道拿哪一份核对"会让目标一直卡在维护里。
|
||||
if persistErr := store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobRunning, "",
|
||||
"", "", baselineID, "", time.Now().UTC()); persistErr != nil {
|
||||
return "", errors.New("持久化基线快照引用失败:" + persistErr.Error())
|
||||
}
|
||||
recorder.Stage(ctx, "capture", "succeeded", "已封存目标整库快照 "+baselineID)
|
||||
recorder.Stage(ctx, "verify", "running", "逐表/索引/资源比对目标与刚封存的快照")
|
||||
recorder.Stage(ctx, "verify", "running", "逐表/索引/资源比对目标与快照 "+shortSnapshotID(baselineID))
|
||||
if verifyErr := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: baselineID}); verifyErr != nil {
|
||||
return "", errors.New("目标内容与刚封存的快照不一致,拒绝登记基线:" + verifyErr.Error())
|
||||
return "", errors.New("目标内容与基线快照不一致,拒绝登记基线:" + verifyErr.Error())
|
||||
}
|
||||
verified = true
|
||||
recorder.Stage(ctx, "verify", "succeeded", "行身份/逐行内容/索引文档/资源摘要与快照一致")
|
||||
// 索引映射是基线证据的一部分:读不到就不登记,绝不用空摘要冒充。
|
||||
mapping, mapErr := target.Search().MappingSHA256(ctx)
|
||||
if mapErr != nil {
|
||||
return "", errors.New("读不到目标索引映射,拒绝用空摘要登记基线:" + mapErr.Error())
|
||||
}
|
||||
if counts == nil {
|
||||
if current, currentErr := target.Current(ctx); currentErr == nil {
|
||||
counts = current.Counts
|
||||
}
|
||||
}
|
||||
// 版本登记放在窗口内:捕获、校验、登记三件事要么一起成立,要么目标原样恢复。
|
||||
recorder.Stage(ctx, "finalize", "running", "登记目标权威基线")
|
||||
if recordErr := store.RecordLibraryVersion(ctx, governance.LibraryVersion{
|
||||
ID: governance.LibraryVersionID(governance.LibraryVersionProductionEffective, payload.Scope, baselineID),
|
||||
Scope: payload.Scope, Kind: governance.LibraryVersionProductionEffective,
|
||||
SnapshotID: baselineID, SourceJobID: payload.LibraryJobID, CreatedBy: actor,
|
||||
Evidence: map[string]any{
|
||||
"action": "target_baseline_init",
|
||||
"origin": "target_baseline",
|
||||
"baseline_snapshot": baselineID,
|
||||
"index_mapping_sha256": mapping,
|
||||
"scope_content_counts": counts,
|
||||
"verification": "目标与快照逐表行身份/逐行内容/索引文档/资源摘要一致",
|
||||
"note": "基线是目标在维护窗口内的真实整库捕获,不是手写版本;发布仍需独立 sync 授权",
|
||||
},
|
||||
CreatedAt: time.Now().UTC(),
|
||||
}); recordErr != nil {
|
||||
return "", errors.New("登记基线版本失败:" + recordErr.Error())
|
||||
}
|
||||
recorded = true
|
||||
recorder.Stage(ctx, "finalize", "succeeded", "已登记目标权威基线 "+shortSnapshotID(baselineID))
|
||||
return baselineID, nil
|
||||
}
|
||||
|
||||
actor := ""
|
||||
if libJob, jobErr := store.LibraryJob(ctx, payload.LibraryJobID); jobErr == nil {
|
||||
actor = libJob.Actor
|
||||
}
|
||||
window, windowErr := coordinator.RunWindow(ctx, payload.LibraryJobID,
|
||||
func(ctx context.Context) error { return manager.VerifyRunning(ctx) }, work,
|
||||
// 本流程对目标内容只读:失败时按原状态重新开放永远安全。
|
||||
func(error) bool { return true })
|
||||
|
||||
now2 := time.Now().UTC()
|
||||
if !verified || window.State != "succeeded" {
|
||||
message := "未建立基线:维护窗口没有完成捕获与校验"
|
||||
if windowErr != nil {
|
||||
message += ":" + windowErr.Error()
|
||||
}
|
||||
if window.State == "held" {
|
||||
// 目标仍被占用:内容状态未知只能按维护保持,等"按预备份恢复"收尾。
|
||||
note := message + "(目标保持维护)"
|
||||
_ = store.UpdateLibraryMaintenanceState(ctx, payload.Scope, payload.LibraryJobID, "needs_restore", note, now2)
|
||||
_ = store.UpdateInstanceMaintenanceState(ctx, identity.Instance, payload.LibraryJobID, "needs_restore", note, now2)
|
||||
recorder.Stage(ctx, "capture", "failed", note)
|
||||
_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "capture", note,
|
||||
"按预备份恢复并校验后再重试", baselineID, "", now2)
|
||||
return errors.New(note)
|
||||
}
|
||||
// 没有写过目标:重新开放占用是安全的,窗口也已按原状态恢复。
|
||||
if releaseErr := releaseWindowLocks(ctx, store, identity.Instance, payload.Scope, payload.LibraryJobID,
|
||||
"未建立基线,目标未被改动", now2); releaseErr != nil {
|
||||
return fmt.Errorf("%s;维护占用释放失败: %w", message, releaseErr)
|
||||
}
|
||||
recorder.Stage(ctx, "capture", "failed", message)
|
||||
_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "capture", message,
|
||||
"未对目标执行任何写入,可直接重试", baselineID, "", now2)
|
||||
return errors.New(message)
|
||||
if settleErr := settleBaselineWindow(ctx, store, manager, payload.LibraryJobID, payload.Scope,
|
||||
identity.Instance, baselineID, recorded, window.State, windowErr); settleErr != nil {
|
||||
return settleErr
|
||||
}
|
||||
|
||||
// 校验通过 → 登记权威基线。证据里记的是这份快照本身、逐表行数与目标索引映射;
|
||||
// created_by 用发起人(管理员),不是 worker 自封。
|
||||
recorder.Stage(ctx, "finalize", "running", "登记目标权威基线")
|
||||
mapping, _ := target.Search().MappingSHA256(ctx)
|
||||
versionErr = store.RecordLibraryVersion(ctx, governance.LibraryVersion{
|
||||
ID: governance.LibraryVersionID(governance.LibraryVersionProductionEffective, payload.Scope, baselineID),
|
||||
Scope: payload.Scope, Kind: governance.LibraryVersionProductionEffective,
|
||||
SnapshotID: baselineID, SourceJobID: payload.LibraryJobID, CreatedBy: actor,
|
||||
Evidence: map[string]any{
|
||||
"action": "target_baseline_init",
|
||||
"origin": "target_baseline",
|
||||
"baseline_snapshot": baselineID,
|
||||
"index_mapping_sha256": mapping,
|
||||
"scope_content_counts": counts,
|
||||
"verification": "目标与快照逐表行身份/逐行内容/索引文档/资源摘要一致",
|
||||
"note": "基线是目标在维护窗口内的真实整库捕获,不是手写版本;发布仍需独立 sync 授权",
|
||||
},
|
||||
CreatedAt: now2,
|
||||
})
|
||||
if versionErr != nil {
|
||||
message := "目标已捕获并校验,但基线版本登记失败:" + versionErr.Error()
|
||||
recorder.Stage(ctx, "finalize", "failed", message)
|
||||
_ = store.UpdateLibraryMaintenanceState(ctx, payload.Scope, payload.LibraryJobID, "needs_restore", message, now2)
|
||||
_ = store.UpdateInstanceMaintenanceState(ctx, identity.Instance, payload.LibraryJobID, "needs_restore", message, now2)
|
||||
_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "finalize", message,
|
||||
"基线快照 "+baselineID+" 保留;修好治理库后可续作登记", baselineID, "", time.Now().UTC())
|
||||
return errors.New(message)
|
||||
}
|
||||
recorder.Stage(ctx, "finalize", "succeeded", "已登记目标权威基线 "+baselineID)
|
||||
|
||||
// 成功要等目标服务、入口与两把维护锁都落定才暴露。
|
||||
if finishErr := finishLibrarySuccess(ctx, store, manager, identity.Instance, payload.Scope,
|
||||
payload.LibraryJobID, successBoundary{Ready: true, Backup: baselineID, FinishedAt: time.Now().UTC()}); finishErr != nil {
|
||||
return finishErr
|
||||
}
|
||||
log.Printf("library baseline succeeded job=%s scope=%s snapshot=%s actor=%s",
|
||||
payload.LibraryJobID, payload.Scope, baselineID, actor)
|
||||
log.Printf("library baseline succeeded job=%s scope=%s snapshot=%s actor=%s finalize_only=%v",
|
||||
payload.LibraryJobID, payload.Scope, baselineID, actor, finalizeOnly)
|
||||
return nil
|
||||
}
|
||||
|
||||
// shortSnapshotID 只用于人看的日志/阶段说明:完整摘要留在版本记录与接口里。
|
||||
func shortSnapshotID(id string) string {
|
||||
if len(id) <= 12 {
|
||||
return id
|
||||
}
|
||||
return id[:12]
|
||||
}
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"knowledge-governance/internal/governance"
|
||||
)
|
||||
|
||||
// 崩溃续作的版本判定:本作业+同一快照 = 只收尾;别人的生效版本 = 拒绝;都没有 = 正常建基线。
|
||||
func TestClassifyBaselineVersions(t *testing.T) {
|
||||
const jobA, jobB, snap = "libbase-internal-1", "libbase-internal-2", "snap-abc"
|
||||
effective := func(sourceJob, snapshot string, superseded bool) governance.LibraryVersion {
|
||||
item := governance.LibraryVersion{
|
||||
Scope: "internal", Kind: governance.LibraryVersionProductionEffective,
|
||||
SnapshotID: snapshot, SourceJobID: sourceJob, CreatedAt: time.Now().UTC(),
|
||||
}
|
||||
if superseded {
|
||||
at := time.Now().UTC()
|
||||
item.SupersededAt = &at
|
||||
}
|
||||
return item
|
||||
}
|
||||
|
||||
if state := classifyBaselineVersions(nil, jobA, snap); state.Own || state.Foreign != nil {
|
||||
t.Fatalf("no versions must mean neither own nor foreign: %+v", state)
|
||||
}
|
||||
if state := classifyBaselineVersions([]governance.LibraryVersion{effective(jobA, snap, false)}, jobA, snap); !state.Own {
|
||||
t.Fatalf("same job + same snapshot must be recognized as own: %+v", state)
|
||||
}
|
||||
// 快照不同就不是"我登记的那份",不能当成可收尾。
|
||||
if state := classifyBaselineVersions([]governance.LibraryVersion{effective(jobA, "snap-other", false)}, jobA, snap); state.Own || state.Foreign == nil {
|
||||
t.Fatalf("different snapshot of same job must not be own: %+v", state)
|
||||
}
|
||||
// 别人的生效版本:即使快照相同也必须拒绝。
|
||||
foreign := classifyBaselineVersions([]governance.LibraryVersion{effective(jobB, snap, false)}, jobA, snap)
|
||||
if foreign.Own || foreign.Foreign == nil || foreign.Foreign.SourceJobID != jobB {
|
||||
t.Fatalf("foreign effective version must be reported: %+v", foreign)
|
||||
}
|
||||
// 已被顶替的旧版本不算生效版本。
|
||||
if state := classifyBaselineVersions([]governance.LibraryVersion{effective(jobB, snap, true)}, jobA, snap); state.Own || state.Foreign != nil {
|
||||
t.Fatalf("superseded version must not block a baseline: %+v", state)
|
||||
}
|
||||
}
|
||||
|
||||
// 版本已登记 + 健康:走成功边界,释放两把占用并把作业标成功。
|
||||
func TestSettleBaselineWindowFinishesWhenRecorded(t *testing.T) {
|
||||
store := &fakeFinishStore{}
|
||||
if err := settleBaselineWindow(context.Background(), store, fakeHealth{}, "job-1", "internal", "kgseal-app",
|
||||
"snap-abc", true, "succeeded", nil); err != nil {
|
||||
t.Fatalf("settle: %v", err)
|
||||
}
|
||||
if !containsCall(store.calls, "advance_succeeded") {
|
||||
t.Fatalf("job must be marked succeeded: %v", store.calls)
|
||||
}
|
||||
if containsCall(store.states, "needs_restore") {
|
||||
t.Fatalf("success path must not hold the target in maintenance: %v", store.states)
|
||||
}
|
||||
if !containsCall(store.calls, "scope_release") || !containsCall(store.calls, "instance_release") {
|
||||
t.Fatalf("both locks must be released: %v", store.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// 版本已登记但目标服务/入口没恢复:不能报成功,必须保持维护并标 needs_restore。
|
||||
func TestSettleBaselineWindowHoldsWhenHealthFails(t *testing.T) {
|
||||
store := &fakeFinishStore{}
|
||||
err := settleBaselineWindow(context.Background(), store, fakeHealth{err: errors.New("kgseal-ingress 未就绪")},
|
||||
"job-1", "internal", "kgseal-app", "snap-abc", true, "succeeded", nil)
|
||||
if err == nil {
|
||||
t.Fatal("unhealthy target must not report success")
|
||||
}
|
||||
if containsCall(store.calls, "advance_succeeded") {
|
||||
t.Fatalf("must not mark succeeded while target is unhealthy: %v", store.calls)
|
||||
}
|
||||
if countCall(store.states, "needs_restore") != 2 {
|
||||
t.Fatalf("must hold both records with needs_restore: %v", store.states)
|
||||
}
|
||||
}
|
||||
|
||||
// 没有登记、窗口仍是 held:只能保持维护(服务确实没回来),绝不能释放占用。
|
||||
func TestSettleBaselineWindowKeepsHeldWindowInMaintenance(t *testing.T) {
|
||||
store := &fakeFinishStore{}
|
||||
err := settleBaselineWindow(context.Background(), store, fakeHealth{}, "job-1", "internal", "kgseal-app",
|
||||
"snap-abc", false, "held", errors.New("停写入者后后台仍有任务"))
|
||||
if err == nil || !strings.Contains(err.Error(), "保持维护") {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
if containsCall(store.calls, "scope_release") || containsCall(store.calls, "instance_release") {
|
||||
t.Fatalf("held window must not release locks: %v", store.calls)
|
||||
}
|
||||
if countCall(store.states, "needs_restore") != 2 {
|
||||
t.Fatalf("held window must be recorded as needs_restore: %v", store.states)
|
||||
}
|
||||
if !containsCall(store.calls, "advance_failed") {
|
||||
t.Fatalf("job must fail: %v", store.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// 没有登记、但窗口已按原状态开放:状态必须跟着变成"已释放",
|
||||
// 不能留下"显示维护中、实际入口已开"。
|
||||
func TestSettleBaselineWindowReleasesWhenWindowRestored(t *testing.T) {
|
||||
store := &fakeFinishStore{}
|
||||
err := settleBaselineWindow(context.Background(), store, fakeHealth{}, "job-1", "internal", "kgseal-app",
|
||||
"snap-abc", false, "failed", errors.New("捕获目标整库失败"))
|
||||
if err == nil || !strings.Contains(err.Error(), "未建立基线") {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
if !containsCall(store.calls, "scope_release") || !containsCall(store.calls, "instance_release") {
|
||||
t.Fatalf("restored window must release both locks: %v", store.calls)
|
||||
}
|
||||
if containsCall(store.states, "needs_restore") {
|
||||
t.Fatalf("released window must not be marked as needs_restore: %v", store.states)
|
||||
}
|
||||
if !containsCall(store.calls, "advance_failed") {
|
||||
t.Fatalf("job must fail: %v", store.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// 释放失败时不能在作业上写"无需恢复":那会让界面与实际占用分叉。
|
||||
func TestSettleBaselineWindowReportsReleaseFailure(t *testing.T) {
|
||||
store := &fakeFinishStore{failOn: "scope_release"}
|
||||
err := settleBaselineWindow(context.Background(), store, fakeHealth{}, "job-1", "internal", "kgseal-app",
|
||||
"snap-abc", false, "failed", nil)
|
||||
if err == nil || !strings.Contains(err.Error(), "维护占用释放失败") {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func countCall(calls []string, want string) int {
|
||||
count := 0
|
||||
for _, call := range calls {
|
||||
if call == want {
|
||||
count++
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
func containsCall(calls []string, want string) bool {
|
||||
for _, call := range calls {
|
||||
if call == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -16,6 +16,8 @@ type fakeFinishStore struct {
|
||||
failOn string
|
||||
alreadyReleased bool
|
||||
advanced []governance.LibraryJobState
|
||||
// states 记录所有维护状态写入(测试要区分 verified 与 needs_restore)。
|
||||
states []string
|
||||
}
|
||||
|
||||
// 模拟"这一行已由同一个作业释放过"(真实 Store 返回 ErrMaintenanceReleased)。
|
||||
@@ -46,16 +48,18 @@ func (f *fakeFinishStore) SetLibraryMaintenanceIngress(context.Context, string,
|
||||
}
|
||||
return f.record("scope_ingress")
|
||||
}
|
||||
func (f *fakeFinishStore) UpdateInstanceMaintenanceState(context.Context, string, string, string, string, time.Time) error {
|
||||
func (f *fakeFinishStore) UpdateInstanceMaintenanceState(_ context.Context, _, _, state, _ string, _ time.Time) error {
|
||||
if err := f.releasedSentinel(); err != nil {
|
||||
return err
|
||||
}
|
||||
f.states = append(f.states, state)
|
||||
return f.record("instance_state")
|
||||
}
|
||||
func (f *fakeFinishStore) UpdateLibraryMaintenanceState(context.Context, string, string, string, string, time.Time) error {
|
||||
func (f *fakeFinishStore) UpdateLibraryMaintenanceState(_ context.Context, _, _, state, _ string, _ time.Time) error {
|
||||
if err := f.releasedSentinel(); err != nil {
|
||||
return err
|
||||
}
|
||||
f.states = append(f.states, state)
|
||||
return f.record("scope_state")
|
||||
}
|
||||
func (f *fakeFinishStore) ReleaseInstanceMaintenance(context.Context, string, string, time.Time) error {
|
||||
|
||||
@@ -66,32 +66,8 @@ type libraryReleasePayload struct {
|
||||
FinalizeOnly bool `json:"finalize_only,omitempty"`
|
||||
}
|
||||
|
||||
// openLibraryMaintenance 按部署注入的配置装配实例级维护目标。
|
||||
// 未接线时返回明确原因:只锁治理库 scope 不算维护,宁可拒绝写也不冒充。
|
||||
func openLibraryMaintenance(config appconfig.Config) (*librarymaintenance.Manager, error) {
|
||||
database, err := librarymaintenance.DatabaseIdentity(config.LibraryProductionDatabaseURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return librarymaintenance.New(librarymaintenance.Config{
|
||||
Instance: config.LibraryTargetInstance,
|
||||
Project: config.LibraryTargetProject,
|
||||
Writers: config.LibraryTargetWriters,
|
||||
NonWriters: config.LibraryTargetNonWriters,
|
||||
DockerSocket: config.LibraryTargetDockerSocket,
|
||||
IngressURL: config.LibraryTargetIngressURL,
|
||||
IngressService: config.LibraryTargetIngressService,
|
||||
IngressDrainFile: config.LibraryTargetIngressDrainFile,
|
||||
Database: database,
|
||||
Index: config.LibraryProductionIndex,
|
||||
Bucket: config.LibraryProductionS3Bucket,
|
||||
// 后台静默检查的运行时依赖:HTTP 排空不等于后台任务排空。
|
||||
DatabaseURL: config.LibraryProductionDatabaseURL,
|
||||
RedisAddr: config.LibraryTargetRedisAddr,
|
||||
RedisPassword: config.LibraryTargetRedisPassword,
|
||||
RedisDB: config.LibraryTargetRedisDB,
|
||||
})
|
||||
}
|
||||
// openLibraryMaintenance 定义在 library_wiring.go:维护目标的装配与目标整库栈共用
|
||||
// 同一个包(internal/librarystack),不在这里再拼一份。
|
||||
|
||||
// runLibraryRelease 执行"闸口 → 授权 → 关闭目标入口并停止写入者 → 预备份 →
|
||||
// 四类写入 → 校验 → 登记权威版本 → 恢复原运行状态和入口"的完整链路。
|
||||
|
||||
+13
-163
@@ -2,180 +2,30 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"knowledge-governance/internal/appconfig"
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
"knowledge-governance/internal/librarymaintenance"
|
||||
"knowledge-governance/internal/librarystack"
|
||||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||||
)
|
||||
|
||||
// productionScopeConfig 是"某个 scope 的生产整库目标"的部署注入配置。
|
||||
// 校验、错误措辞与 S3 参数只在这里有一份:同步、恢复与基线共用同一套装配,
|
||||
// 不允许任何一条路径自己拼一份(那会让"目标是谁"随路径漂移)。
|
||||
type productionScopeConfig struct {
|
||||
target v071.ProductionScope
|
||||
s3 v071.S3ObjectStoreConfig
|
||||
// 目标整库读写栈的装配只有一份实现(internal/librarystack):worker 的同步/恢复/
|
||||
// 基线与主机侧恢复演练共用它,避免"目标是谁、用哪份契约"随调用路径漂移。
|
||||
|
||||
// openLibraryMaintenance 按部署注入的配置装配实例级维护目标。
|
||||
// 未接线时返回明确原因:只锁治理库 scope 不算维护,宁可拒绝写也不冒充。
|
||||
func openLibraryMaintenance(config appconfig.Config) (*librarymaintenance.Manager, error) {
|
||||
return librarystack.OpenMaintenance(config)
|
||||
}
|
||||
|
||||
func productionScopeOf(ctx context.Context, config appconfig.Config, scope string) (productionScopeConfig, error) {
|
||||
missing := make([]string, 0, 4)
|
||||
if config.LibraryProductionDatabaseURL == "" {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_DATABASE_URL")
|
||||
}
|
||||
if config.LibraryProductionElasticURL == "" {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_ELASTIC_URL")
|
||||
}
|
||||
if config.LibraryProductionTenantID == 0 {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_TENANT_ID")
|
||||
}
|
||||
kb := config.LibraryProductionInternalKB
|
||||
if scope == "external" {
|
||||
kb = config.LibraryProductionExternalKB
|
||||
}
|
||||
if kb == "" {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_KB_"+strings.ToUpper(scope))
|
||||
}
|
||||
if config.LibraryProductionS3Endpoint == "" || config.LibraryProductionS3Bucket == "" ||
|
||||
config.LibraryProductionS3AccessKey == "" || config.LibraryProductionS3SecretKey == "" {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_S3_ENDPOINT/BUCKET/ACCESS_KEY/SECRET_KEY")
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
return productionScopeConfig{}, fmt.Errorf("生产整库目标未配置:缺少 %s", strings.Join(missing, "、"))
|
||||
}
|
||||
return productionScopeConfig{
|
||||
target: v071.ProductionScope{TenantID: config.LibraryProductionTenantID, KnowledgeBaseID: kb},
|
||||
s3: v071.S3ObjectStoreConfig{
|
||||
Provider: config.LibraryProductionS3Provider, Endpoint: config.LibraryProductionS3Endpoint,
|
||||
Region: config.LibraryProductionS3Region, Bucket: config.LibraryProductionS3Bucket,
|
||||
AccessKeyID: config.LibraryProductionS3AccessKey, SecretAccessKey: config.LibraryProductionS3SecretKey,
|
||||
Secure: strings.HasPrefix(config.LibraryProductionS3Endpoint, "https://"),
|
||||
ForcePathStyle: true, MaxObjectBytes: 256 * 1024 * 1024,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func newProductionTarget(ctx context.Context, config appconfig.Config, scope productionScopeConfig) (*v071.ProductionLibraryTarget, error) {
|
||||
objects, err := v071.NewS3ObjectStore(scope.s3)
|
||||
if err != nil {
|
||||
return nil, errors.New("生产对象存储配置无效")
|
||||
}
|
||||
return v071.NewProductionLibraryTarget(ctx, v071.ProductionLibraryConfig{
|
||||
Postgres: config.LibraryProductionDatabaseURL, SearchURL: config.LibraryProductionElasticURL,
|
||||
SearchIndex: config.LibraryProductionIndex, Objects: objects, ObjectsReader: objects,
|
||||
ArchiveRoot: config.LibraryArchiveRoot,
|
||||
ObjectsBucket: config.LibraryProductionS3Bucket,
|
||||
MaxDocuments: 200000,
|
||||
}, scope.target)
|
||||
}
|
||||
|
||||
func newProductionBackupStore(config appconfig.Config, scope string,
|
||||
target *v071.ProductionLibraryTarget, contract librarysnapshot.Compatibility) (*v071.ProductionBackupStore, error) {
|
||||
return v071.NewProductionBackupStore(v071.ProductionBackupConfig{
|
||||
Pool: target.Pool(), Search: target.Search(), Objects: target.Objects(),
|
||||
ArchiveRoot: config.LibraryArchiveRoot, Scope: scope, Identity: target.Identity(),
|
||||
Compatibility: contract, MaxDocuments: 200000, MaxObjectByte: 256 * 1024 * 1024,
|
||||
})
|
||||
}
|
||||
|
||||
// buildLibraryProductionStack 用部署注入的配置装配生产整库目标与独立预备份存储。
|
||||
// 任何一项缺失都返回明确原因,调用方据此把作业失败并把缺项写进阶段,
|
||||
// 不用"未配置"的占位对象假装能跑。
|
||||
// buildLibraryProductionStack 用"要同步的这份快照"做契约装配目标与预备份存储。
|
||||
func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, scope, snapshotID, knownMapping string) (
|
||||
*v071.ProductionLibraryTarget, *v071.ProductionBackupStore, error) {
|
||||
scopeConfig, err := productionScopeOf(ctx, config, scope)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
target, err := newProductionTarget(ctx, config, scopeConfig)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
// 契约以"要同步的这份快照"为准:生产必须向它看齐,而不是反过来。
|
||||
// 生产为空时索引可能根本不存在,这时不能拿生产映射当契约(会永远启动不了)。
|
||||
archive, err := librarysnapshot.Open(config.LibraryArchiveRoot, 256*1024*1024)
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, fmt.Errorf("归档不可用,无法读取快照契约: %w", err)
|
||||
}
|
||||
manifest, err := archive.Load(snapshotID)
|
||||
archive.Close()
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, fmt.Errorf("读取目标快照失败: %w", err)
|
||||
}
|
||||
contract := manifest.Compatibility
|
||||
if contract.EmbeddingDimension == 0 {
|
||||
contract.EmbeddingModel = config.LibraryEmbeddingModel
|
||||
contract.EmbeddingDimension = config.LibraryEmbeddingDimension
|
||||
}
|
||||
// 生产索引存在时必须与快照契约一致;不存在或为空基线时跳过比对
|
||||
// (空索引的映射没有可比性,非空之后仍然严格比对)。
|
||||
if mapping, mapErr := target.Search().MappingSHA256(ctx); mapErr == nil {
|
||||
// 映射基线:优先用"上一次生产生效版本登记的实际映射",
|
||||
// 它才是这份生产索引的既有事实;快照契约只管"来自哪份数据"。
|
||||
baseline := knownMapping
|
||||
if baseline == "" {
|
||||
// 没有历史基线时(首次接管这套生产索引),把当前实际映射当作基线并随
|
||||
// 预备份登记;之后任何改动都会与登记值不符而被拒绝。
|
||||
baseline = mapping
|
||||
}
|
||||
if baseline != "" && mapping != baseline {
|
||||
current, currentErr := target.Current(ctx)
|
||||
emptyBaseline := currentErr == nil && current.Counts["index"] == 0 && current.Counts["chunks"] == 0
|
||||
if !emptyBaseline {
|
||||
target.Close()
|
||||
return nil, nil, errors.New("生产索引映射与上次生效版本不一致(索引被改动过),先对齐索引映射")
|
||||
}
|
||||
}
|
||||
}
|
||||
// 预备份描述的是"生产现在的索引",不是"快照的来源索引":
|
||||
// 生产映射基线优先用上次生效版本登记的实际值。
|
||||
backupContract := contract
|
||||
if knownMapping != "" {
|
||||
backupContract.IndexMappingSHA256 = knownMapping
|
||||
}
|
||||
backups, err := newProductionBackupStore(config, scope, target, backupContract)
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return target, backups, nil
|
||||
return librarystack.OpenProductionStack(ctx, config, scope, snapshotID, knownMapping)
|
||||
}
|
||||
|
||||
// buildLibraryBaselineStack 装配"给目标自己建基线"所需的目标与捕获器。
|
||||
//
|
||||
// 与同步的差别只有契约来源:同步要拿**待发布快照**的兼容性做契约,而基线
|
||||
// 还没有快照——它要捕获的正是目标当前状态,所以契约取目标此刻的真实索引映射,
|
||||
// 绝不去借一份别人的快照来当契约(那会把基线登记成"与某份快照一致"的假象)。
|
||||
// buildLibraryBaselineStack 装配"给目标自己建基线"所需的目标与捕获器(契约取目标当前映射)。
|
||||
func buildLibraryBaselineStack(ctx context.Context, config appconfig.Config, scope string) (
|
||||
*v071.ProductionLibraryTarget, *v071.ProductionBackupStore, error) {
|
||||
scopeConfig, err := productionScopeOf(ctx, config, scope)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
target, err := newProductionTarget(ctx, config, scopeConfig)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
mapping, mapErr := target.Search().MappingSHA256(ctx)
|
||||
if mapErr != nil {
|
||||
target.Close()
|
||||
return nil, nil, fmt.Errorf("读取目标索引映射失败,无法确定基线契约: %w", mapErr)
|
||||
}
|
||||
// 只写能证明的字段:embedding 配置来自部署声明(与目标实例同一份配置),
|
||||
// 索引映射来自目标此刻的真实映射。目标实例的 WeKnora/schema 版本这里没有
|
||||
// 可靠来源,就不填——空白比编一个数字诚实,迁移/兼容判断不会因此放行任何东西。
|
||||
contract := librarysnapshot.Compatibility{
|
||||
EmbeddingModel: config.LibraryEmbeddingModel,
|
||||
EmbeddingDimension: config.LibraryEmbeddingDimension,
|
||||
IndexMappingSHA256: mapping,
|
||||
}
|
||||
backups, err := newProductionBackupStore(config, scope, target, contract)
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return target, backups, nil
|
||||
return librarystack.OpenBaselineStack(ctx, config, scope)
|
||||
}
|
||||
|
||||
@@ -67,7 +67,13 @@ func libraryBaselineEnqueue(w http.ResponseWriter, r *http.Request, config Confi
|
||||
}
|
||||
// 目标必须真的读得到:读不到就不是"非空无权威版本",而是另一个问题,
|
||||
// 不允许用建基线去掩盖连接/配置故障。
|
||||
empty, counts, probeErr := productionScopeState(r.Context(), config, request.Scope)
|
||||
probe := config.LibraryScopeProbe
|
||||
if probe == nil {
|
||||
probe = func(ctx context.Context, scope string) (bool, map[string]int64, error) {
|
||||
return productionScopeState(ctx, config, scope)
|
||||
}
|
||||
}
|
||||
empty, counts, probeErr := probe(r.Context(), request.Scope)
|
||||
if probeErr != nil {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_baseline_unprobeable",
|
||||
"error": "无法读取目标整库状态,不能建立基线:" + probeErr.Error()})
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"knowledge-governance/internal/governance"
|
||||
"knowledge-governance/internal/jobs"
|
||||
)
|
||||
|
||||
// nonEmptyTarget 模拟"目标整库非空且可读"的探测结果(真实实现直连目标库逐表计数)。
|
||||
func nonEmptyTarget(context.Context, string) (bool, map[string]int64, error) {
|
||||
return false, map[string]int64{"knowledges": 1, "chunks": 2}, nil
|
||||
}
|
||||
|
||||
// scopedProbe:internal 可读非空,external 没有目标 KB(与当前部署一致)。
|
||||
func scopedProbe(_ context.Context, scope string) (bool, map[string]int64, error) {
|
||||
if scope == "external" {
|
||||
return false, nil, errors.New("该范围的目标知识库未配置")
|
||||
}
|
||||
return nonEmptyTarget(context.Background(), scope)
|
||||
}
|
||||
|
||||
// baselineMutexStore 在 fake 之外**真的**实现"同一 scope 只允许一个在飞写作业"这条不变量,
|
||||
// 用来验证接口的重试/双击行为(真实索引由 integration test
|
||||
// TestLibraryWriteMutexCoversBaselineJobs 在 PostgreSQL 上证明)。
|
||||
type baselineMutexStore struct {
|
||||
fakeLibraryStore
|
||||
activeByScope map[string]string
|
||||
enqueues int
|
||||
}
|
||||
|
||||
func (s *baselineMutexStore) CreateLibraryJob(ctx context.Context, job governance.LibraryJob, stages []string, now time.Time) (governance.LibraryJob, bool, error) {
|
||||
if s.activeByScope == nil {
|
||||
s.activeByScope = map[string]string{}
|
||||
}
|
||||
if owner := s.activeByScope[job.Scope]; owner != "" {
|
||||
return governance.LibraryJob{}, false, governance.ErrLibraryJobActive
|
||||
}
|
||||
s.activeByScope[job.Scope] = job.ID
|
||||
return s.fakeLibraryStore.CreateLibraryJob(ctx, job, stages, now)
|
||||
}
|
||||
|
||||
func (s *baselineMutexStore) Enqueue(ctx context.Context, request jobs.EnqueueRequest, now time.Time) (*jobs.Job, error) {
|
||||
s.enqueues++
|
||||
return s.fakeLibraryStore.fakeJobStore.Enqueue(ctx, request, now)
|
||||
}
|
||||
|
||||
// 双击/重试:第一次建作业成功,第二次被互斥拒绝,且**不会**再投递一个队列作业。
|
||||
func TestLibraryBaselineDoubleSubmitCreatesOneJob(t *testing.T) {
|
||||
store := &baselineMutexStore{fakeLibraryStore: fakeLibraryStore{fakeJobStore: fakeJobStore{enqueueEchoRequestedID: true}}}
|
||||
config := Config{Store: store, LibraryScopeProbe: nonEmptyTarget}
|
||||
first := httptest.NewRecorder()
|
||||
libraryBaselineEnqueue(first, baselineRequest(`{"scope":"internal"}`, roleAdmin, "admin"), config)
|
||||
if first.Code != http.StatusAccepted {
|
||||
t.Fatalf("first status=%d body=%s", first.Code, first.Body.String())
|
||||
}
|
||||
if store.enqueues != 1 {
|
||||
t.Fatalf("first submit enqueues=%d, want 1", store.enqueues)
|
||||
}
|
||||
second := httptest.NewRecorder()
|
||||
libraryBaselineEnqueue(second, baselineRequest(`{"scope":"internal"}`, roleAdmin, "admin"), config)
|
||||
if second.Code != http.StatusConflict {
|
||||
t.Fatalf("second status=%d body=%s", second.Code, second.Body.String())
|
||||
}
|
||||
if body := decodeBody(t, second); body["code"] != "library_write_active" {
|
||||
t.Fatalf("second body=%v", body)
|
||||
}
|
||||
if store.enqueues != 1 {
|
||||
t.Fatalf("rejected submit must not enqueue again: enqueues=%d", store.enqueues)
|
||||
}
|
||||
}
|
||||
|
||||
// 不同 scope 不会被同一把锁挡住(互斥粒度是 scope/实例,不是全局)。
|
||||
func TestLibraryBaselineDifferentScopesAreNotBlocked(t *testing.T) {
|
||||
store := &baselineMutexStore{fakeLibraryStore: fakeLibraryStore{fakeJobStore: fakeJobStore{enqueueEchoRequestedID: true}}}
|
||||
config := Config{Store: store, LibraryScopeProbe: scopedProbe}
|
||||
internal := httptest.NewRecorder()
|
||||
libraryBaselineEnqueue(internal, baselineRequest(`{"scope":"internal"}`, roleAdmin, "admin"), config)
|
||||
if internal.Code != http.StatusAccepted {
|
||||
t.Fatalf("internal status=%d body=%s", internal.Code, internal.Body.String())
|
||||
}
|
||||
// external 在探针阶段就会被拒(该范围没有目标 KB),但绝不能是因为互斥。
|
||||
external := httptest.NewRecorder()
|
||||
libraryBaselineEnqueue(external, baselineRequest(`{"scope":"external"}`, roleAdmin, "admin"), config)
|
||||
if external.Code == http.StatusAccepted {
|
||||
t.Fatalf("external without a target KB must not be accepted: %s", external.Body.String())
|
||||
}
|
||||
body := decodeBody(t, external)
|
||||
if body["code"] == "library_write_active" {
|
||||
t.Fatalf("external must not be blocked by the internal scope's mutex: %v", body)
|
||||
}
|
||||
}
|
||||
@@ -173,6 +173,9 @@ type Config struct {
|
||||
LibraryProductionInternalKB string
|
||||
LibraryProductionExternalKB string
|
||||
LibraryProductionTenantID uint64
|
||||
// LibraryScopeProbe 让测试注入"目标整库是否为空"的探测;留空即用真实连接。
|
||||
// 它不是运行时开关:生产部署不设置它,那次探测仍然直连目标库逐表计数。
|
||||
LibraryScopeProbe func(ctx context.Context, scope string) (bool, map[string]int64, error)
|
||||
// LibraryRetentionVersions/Days 是保留策略;都为 0 = 未设置(不生成清理候选)。
|
||||
LibraryRetentionVersions int
|
||||
LibraryRetentionDays int
|
||||
|
||||
@@ -0,0 +1,201 @@
|
||||
// Package librarystack 是"目标整库读写栈"的唯一装配点:同步、恢复、基线,
|
||||
// 以及主机侧的真实恢复演练都从这里拿同一个目标/预备份存储/维护协调器。
|
||||
//
|
||||
// 为什么单独成包:装配一旦被复制两份,"目标是谁、用哪份契约"就会随调用路径漂移。
|
||||
// 这个包只做装配与配置校验,不包含任何业务编排。
|
||||
package librarystack
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"knowledge-governance/internal/appconfig"
|
||||
"knowledge-governance/internal/librarymaintenance"
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||||
)
|
||||
|
||||
// OpenMaintenance 按部署配置装配实例级维护目标;未接线时返回明确原因。
|
||||
func OpenMaintenance(config appconfig.Config) (*librarymaintenance.Manager, error) {
|
||||
database, err := librarymaintenance.DatabaseIdentity(config.LibraryProductionDatabaseURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return librarymaintenance.New(librarymaintenance.Config{
|
||||
Instance: config.LibraryTargetInstance,
|
||||
Project: config.LibraryTargetProject,
|
||||
Writers: config.LibraryTargetWriters,
|
||||
NonWriters: config.LibraryTargetNonWriters,
|
||||
DockerSocket: config.LibraryTargetDockerSocket,
|
||||
IngressURL: config.LibraryTargetIngressURL,
|
||||
IngressService: config.LibraryTargetIngressService,
|
||||
IngressDrainFile: config.LibraryTargetIngressDrainFile,
|
||||
Database: database,
|
||||
Index: config.LibraryProductionIndex,
|
||||
Bucket: config.LibraryProductionS3Bucket,
|
||||
DatabaseURL: config.LibraryProductionDatabaseURL,
|
||||
RedisAddr: config.LibraryTargetRedisAddr,
|
||||
RedisPassword: config.LibraryTargetRedisPassword,
|
||||
RedisDB: config.LibraryTargetRedisDB,
|
||||
})
|
||||
}
|
||||
|
||||
// scopeConfig 是"某个 scope 的生产整库目标"的部署注入配置。
|
||||
type scopeConfig struct {
|
||||
target v071.ProductionScope
|
||||
s3 v071.S3ObjectStoreConfig
|
||||
}
|
||||
|
||||
func scopeConfigOf(config appconfig.Config, scope string) (scopeConfig, error) {
|
||||
missing := make([]string, 0, 5)
|
||||
if config.LibraryProductionDatabaseURL == "" {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_DATABASE_URL")
|
||||
}
|
||||
if config.LibraryProductionElasticURL == "" {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_ELASTIC_URL")
|
||||
}
|
||||
if config.LibraryProductionTenantID == 0 {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_TENANT_ID")
|
||||
}
|
||||
kb := config.LibraryProductionInternalKB
|
||||
if scope == "external" {
|
||||
kb = config.LibraryProductionExternalKB
|
||||
}
|
||||
if kb == "" {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_KB_"+strings.ToUpper(scope))
|
||||
}
|
||||
if config.LibraryProductionS3Endpoint == "" || config.LibraryProductionS3Bucket == "" ||
|
||||
config.LibraryProductionS3AccessKey == "" || config.LibraryProductionS3SecretKey == "" {
|
||||
missing = append(missing, "LIBRARY_PRODUCTION_S3_ENDPOINT/BUCKET/ACCESS_KEY/SECRET_KEY")
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
return scopeConfig{}, fmt.Errorf("生产整库目标未配置:缺少 %s", strings.Join(missing, "、"))
|
||||
}
|
||||
return scopeConfig{
|
||||
target: v071.ProductionScope{TenantID: config.LibraryProductionTenantID, KnowledgeBaseID: kb},
|
||||
s3: v071.S3ObjectStoreConfig{
|
||||
Provider: config.LibraryProductionS3Provider, Endpoint: config.LibraryProductionS3Endpoint,
|
||||
Region: config.LibraryProductionS3Region, Bucket: config.LibraryProductionS3Bucket,
|
||||
AccessKeyID: config.LibraryProductionS3AccessKey, SecretAccessKey: config.LibraryProductionS3SecretKey,
|
||||
Secure: strings.HasPrefix(config.LibraryProductionS3Endpoint, "https://"),
|
||||
ForcePathStyle: true, MaxObjectBytes: 256 * 1024 * 1024,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func newTarget(ctx context.Context, config appconfig.Config, scope scopeConfig) (*v071.ProductionLibraryTarget, error) {
|
||||
objects, err := v071.NewS3ObjectStore(scope.s3)
|
||||
if err != nil {
|
||||
return nil, errors.New("生产对象存储配置无效")
|
||||
}
|
||||
return v071.NewProductionLibraryTarget(ctx, v071.ProductionLibraryConfig{
|
||||
Postgres: config.LibraryProductionDatabaseURL, SearchURL: config.LibraryProductionElasticURL,
|
||||
SearchIndex: config.LibraryProductionIndex, Objects: objects, ObjectsReader: objects,
|
||||
ArchiveRoot: config.LibraryArchiveRoot,
|
||||
ObjectsBucket: config.LibraryProductionS3Bucket,
|
||||
MaxDocuments: 200000,
|
||||
}, scope.target)
|
||||
}
|
||||
|
||||
func newBackups(config appconfig.Config, scope string, target *v071.ProductionLibraryTarget,
|
||||
contract librarysnapshot.Compatibility) (*v071.ProductionBackupStore, error) {
|
||||
return v071.NewProductionBackupStore(v071.ProductionBackupConfig{
|
||||
Pool: target.Pool(), Search: target.Search(), Objects: target.Objects(),
|
||||
ArchiveRoot: config.LibraryArchiveRoot, Scope: scope, Identity: target.Identity(),
|
||||
Compatibility: contract, MaxDocuments: 200000, MaxObjectByte: 256 * 1024 * 1024,
|
||||
})
|
||||
}
|
||||
|
||||
// OpenProductionStack 用部署注入的配置装配目标与独立预备份存储。
|
||||
// 契约以"要同步的这份快照"为准:生产必须向它看齐,而不是反过来。
|
||||
func OpenProductionStack(ctx context.Context, config appconfig.Config, scope, snapshotID, knownMapping string) (
|
||||
*v071.ProductionLibraryTarget, *v071.ProductionBackupStore, error) {
|
||||
scopeCfg, err := scopeConfigOf(config, scope)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
target, err := newTarget(ctx, config, scopeCfg)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
archive, err := librarysnapshot.Open(config.LibraryArchiveRoot, 256*1024*1024)
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, fmt.Errorf("归档不可用,无法读取快照契约: %w", err)
|
||||
}
|
||||
manifest, err := archive.Load(snapshotID)
|
||||
archive.Close()
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, fmt.Errorf("读取目标快照失败: %w", err)
|
||||
}
|
||||
contract := manifest.Compatibility
|
||||
if contract.EmbeddingDimension == 0 {
|
||||
contract.EmbeddingModel = config.LibraryEmbeddingModel
|
||||
contract.EmbeddingDimension = config.LibraryEmbeddingDimension
|
||||
}
|
||||
// 生产索引存在时必须与快照契约一致;不存在或为空基线时跳过比对
|
||||
// (空索引的映射没有可比性,非空之后仍然严格比对)。
|
||||
if mapping, mapErr := target.Search().MappingSHA256(ctx); mapErr == nil {
|
||||
// 映射基线:优先用"上一次生产生效版本登记的实际映射",
|
||||
// 它才是这份生产索引的既有事实;快照契约只管"来自哪份数据"。
|
||||
baseline := knownMapping
|
||||
if baseline == "" {
|
||||
baseline = mapping
|
||||
}
|
||||
if baseline != "" && mapping != baseline {
|
||||
current, currentErr := target.Current(ctx)
|
||||
emptyBaseline := currentErr == nil && current.Counts["index"] == 0 && current.Counts["chunks"] == 0
|
||||
if !emptyBaseline {
|
||||
target.Close()
|
||||
return nil, nil, errors.New("生产索引映射与上次生效版本不一致(索引被改动过),先对齐索引映射")
|
||||
}
|
||||
}
|
||||
}
|
||||
backupContract := contract
|
||||
if knownMapping != "" {
|
||||
backupContract.IndexMappingSHA256 = knownMapping
|
||||
}
|
||||
backups, err := newBackups(config, scope, target, backupContract)
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return target, backups, nil
|
||||
}
|
||||
|
||||
// OpenBaselineStack 装配"给目标自己建基线"所需的目标与捕获器。
|
||||
//
|
||||
// 与同步的差别只有契约来源:基线还没有快照,它要捕获的正是目标当前状态,
|
||||
// 所以契约取目标此刻的真实索引映射,绝不去借一份别人的快照当契约。
|
||||
func OpenBaselineStack(ctx context.Context, config appconfig.Config, scope string) (
|
||||
*v071.ProductionLibraryTarget, *v071.ProductionBackupStore, error) {
|
||||
scopeCfg, err := scopeConfigOf(config, scope)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
target, err := newTarget(ctx, config, scopeCfg)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
mapping, mapErr := target.Search().MappingSHA256(ctx)
|
||||
if mapErr != nil {
|
||||
target.Close()
|
||||
return nil, nil, fmt.Errorf("读取目标索引映射失败,无法确定基线契约: %w", mapErr)
|
||||
}
|
||||
// 只写能证明的字段:embedding 配置来自部署声明,索引映射来自目标此刻的真实映射;
|
||||
// 目标实例的 WeKnora/schema 版本这里没有可靠来源,就不填。
|
||||
contract := librarysnapshot.Compatibility{
|
||||
EmbeddingModel: config.LibraryEmbeddingModel,
|
||||
EmbeddingDimension: config.LibraryEmbeddingDimension,
|
||||
IndexMappingSHA256: mapping,
|
||||
}
|
||||
backups, err := newBackups(config, scope, target, contract)
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return target, backups, nil
|
||||
}
|
||||
@@ -65,7 +65,15 @@ ON CONFLICT (instance) DO UPDATE SET
|
||||
services = EXCLUDED.services,
|
||||
ingress_state = EXCLUDED.ingress_state,
|
||||
state = EXCLUDED.state,
|
||||
original_service_state = EXCLUDED.original_service_state,
|
||||
-- 同一作业重入(崩溃接管)时不许用"现在已停"覆盖第一次记下的原状态:
|
||||
-- 重入时服务已经被我们停住,再读一次只会读到 stopped,写回去会让恢复阶段
|
||||
-- 以为写入者本来就没跑,于是永远不启动它。
|
||||
original_service_state = CASE
|
||||
WHEN library_instance_maintenance.owner_job_id = EXCLUDED.owner_job_id
|
||||
AND library_instance_maintenance.original_service_state <> ''
|
||||
THEN library_instance_maintenance.original_service_state
|
||||
ELSE EXCLUDED.original_service_state
|
||||
END,
|
||||
note = EXCLUDED.note,
|
||||
updated_at = EXCLUDED.updated_at,
|
||||
released_at = NULL
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"knowledge-governance/internal/governance"
|
||||
)
|
||||
|
||||
// 迁移 075 之后,"建立基线"必须与同步/恢复共用同一把写互斥:
|
||||
// 同一 scope 只允许一个在飞写作业,且互斥由数据库唯一索引保证(不靠页面)。
|
||||
// 这条用例跑在真实 PostgreSQL 上(TEST_DATABASE_URL),不是内存假实现。
|
||||
func TestLibraryWriteMutexCoversBaselineJobs(t *testing.T) {
|
||||
databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
|
||||
if databaseURL == "" {
|
||||
t.Skip("TEST_DATABASE_URL is not set")
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, databaseURL)
|
||||
if err != nil {
|
||||
t.Fatalf("connect: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
if err := ApplyMigrations(ctx, pool); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
store := NewStore(pool)
|
||||
now := time.Now().UTC()
|
||||
digest, err := governance.LibraryRequestDigest("internal", governance.LibraryJobBaseline,
|
||||
map[string]any{"scope": "internal"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
baseline := governance.LibraryJob{
|
||||
ID: "libbase-integration-1", Scope: "internal", Kind: governance.LibraryJobBaseline,
|
||||
IdempotencyKey: "library.baseline:internal:1", RequestDigest: digest, Actor: "admin",
|
||||
}
|
||||
created, reused, err := store.CreateLibraryJob(ctx, baseline, governance.BaselineStages(), now)
|
||||
if err != nil || reused {
|
||||
t.Fatalf("first baseline job: err=%v reused=%v", err, reused)
|
||||
}
|
||||
if len(created.Stages) == 0 {
|
||||
t.Fatalf("stages not created: %+v", created)
|
||||
}
|
||||
|
||||
// 同一幂等键 + 同一请求 = 复用同一作业,不产生第二行。
|
||||
same, reused, err := store.CreateLibraryJob(ctx, baseline, governance.BaselineStages(), now)
|
||||
if err != nil || !reused || same.ID != baseline.ID {
|
||||
t.Fatalf("idempotent retry: err=%v reused=%v id=%q", err, reused, same.ID)
|
||||
}
|
||||
|
||||
// 另一个基线作业(不同幂等键)在同一 scope 上必须被互斥挡住。
|
||||
second := baseline
|
||||
second.ID = "libbase-integration-2"
|
||||
second.IdempotencyKey = "library.baseline:internal:2"
|
||||
if _, _, err := store.CreateLibraryJob(ctx, second, governance.BaselineStages(), now); !errors.Is(err, governance.ErrLibraryJobActive) {
|
||||
t.Fatalf("second baseline during first: err=%v", err)
|
||||
}
|
||||
|
||||
// 同步作业与基线共用同一把锁:cross-kind 也必须被挡。
|
||||
syncDigest, err := governance.LibraryRequestDigest("internal", governance.LibraryJobSync,
|
||||
map[string]any{"target_snapshot_id": "snap-1"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, _, err := store.CreateLibraryJob(ctx, governance.LibraryJob{
|
||||
ID: "libsync-integration-1", Scope: "internal", Kind: governance.LibraryJobSync,
|
||||
IdempotencyKey: "library.sync:internal:snap-1", RequestDigest: syncDigest, TargetSnapshotID: "snap-1",
|
||||
}, governance.SyncStages(), now); !errors.Is(err, governance.ErrLibraryJobActive) {
|
||||
t.Fatalf("sync during baseline: err=%v", err)
|
||||
}
|
||||
|
||||
// 第一个作业失败后互斥必须释放,新作业可以进(否则第一次失败就永久堵死)。
|
||||
if err := store.AdvanceLibraryJob(ctx, baseline.ID, governance.LibraryJobFailed, "capture", "test", "", "", "", now.Add(time.Second)); err != nil {
|
||||
t.Fatalf("advance: %v", err)
|
||||
}
|
||||
if _, _, err := store.CreateLibraryJob(ctx, second, governance.BaselineStages(), now.Add(2*time.Second)); err != nil {
|
||||
t.Fatalf("baseline after failure: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -81,7 +81,13 @@ ON CONFLICT (scope) DO UPDATE SET
|
||||
services = EXCLUDED.services,
|
||||
ingress_state = EXCLUDED.ingress_state,
|
||||
state = EXCLUDED.state,
|
||||
original_service_state = EXCLUDED.original_service_state,
|
||||
-- 同上:原运行状态只由第一次进入窗口时决定,重入不能覆盖。
|
||||
original_service_state = CASE
|
||||
WHEN library_maintenance.owner_job_id = EXCLUDED.owner_job_id
|
||||
AND library_maintenance.original_service_state <> ''
|
||||
THEN library_maintenance.original_service_state
|
||||
ELSE EXCLUDED.original_service_state
|
||||
END,
|
||||
note = EXCLUDED.note,
|
||||
updated_at = EXCLUDED.updated_at,
|
||||
released_at = NULL
|
||||
|
||||
@@ -72,6 +72,86 @@ func TestLibraryMaintenanceOwnershipAcrossRelease(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 同一作业重入(崩溃后的接管)必须保留第一次记下的"原运行状态":
|
||||
// 重入时目标已经被停住,此时再去读一次只会读到"已停",把它写回去会让恢复阶段
|
||||
// 以为写入者本来就没跑,于是永远不启动它 —— 目标就这么一直停着。
|
||||
func TestMaintenanceReentryKeepsOriginalServiceState(t *testing.T) {
|
||||
databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
|
||||
if databaseURL == "" {
|
||||
t.Skip("TEST_DATABASE_URL is not set")
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, databaseURL)
|
||||
if err != nil {
|
||||
t.Fatalf("connect: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
if err := ApplyMigrations(ctx, pool); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
store := NewStore(pool)
|
||||
now := time.Now().UTC()
|
||||
|
||||
first, err := store.AcquireLibraryMaintenance(ctx, LibraryMaintenance{
|
||||
Scope: "internal", OwnerJobID: "job-reentry", TargetIdentity: "instance=kgseal",
|
||||
Level: "instance", TargetFingerprint: "fp", Services: []string{"app"},
|
||||
State: "acquired", OriginalServiceState: "app=running", Note: "first",
|
||||
}, now)
|
||||
if err != nil {
|
||||
t.Fatalf("first acquire: %v", err)
|
||||
}
|
||||
if first.OriginalServiceState != "app=running" {
|
||||
t.Fatalf("first original state = %q", first.OriginalServiceState)
|
||||
}
|
||||
reentered, err := store.AcquireLibraryMaintenance(ctx, LibraryMaintenance{
|
||||
Scope: "internal", OwnerJobID: "job-reentry", Level: "instance",
|
||||
State: "applying", OriginalServiceState: "app=stopped", Note: "re-entry after crash",
|
||||
}, now.Add(time.Second))
|
||||
if err != nil {
|
||||
t.Fatalf("same-owner re-entry: %v", err)
|
||||
}
|
||||
if reentered.OriginalServiceState != "app=running" {
|
||||
t.Fatalf("re-entry overwrote the original state: got %q, want app=running", reentered.OriginalServiceState)
|
||||
}
|
||||
|
||||
instFirst, err := store.AcquireInstanceMaintenance(ctx, LibraryInstanceMaintenance{
|
||||
Instance: "kgseal-instance-reentry", OwnerJobID: "job-reentry", OwnerScope: "internal",
|
||||
TargetFingerprint: "fp", Level: "instance", Services: []string{"app"},
|
||||
State: "acquired", OriginalServiceState: "app=running", Note: "first",
|
||||
}, now)
|
||||
if err != nil {
|
||||
t.Fatalf("instance acquire: %v", err)
|
||||
}
|
||||
if instFirst.OriginalServiceState != "app=running" {
|
||||
t.Fatalf("instance first original state = %q", instFirst.OriginalServiceState)
|
||||
}
|
||||
instReentered, err := store.AcquireInstanceMaintenance(ctx, LibraryInstanceMaintenance{
|
||||
Instance: "kgseal-instance-reentry", OwnerJobID: "job-reentry", OwnerScope: "internal",
|
||||
Level: "instance", State: "applying", OriginalServiceState: "app=stopped",
|
||||
}, now.Add(2*time.Second))
|
||||
if err != nil {
|
||||
t.Fatalf("instance same-owner re-entry: %v", err)
|
||||
}
|
||||
if instReentered.OriginalServiceState != "app=running" {
|
||||
t.Fatalf("instance re-entry overwrote the original state: got %q", instReentered.OriginalServiceState)
|
||||
}
|
||||
|
||||
// 新作业接管已释放的窗口时,必须接受它自己读到的当前状态(不能沿用上一个作业的)。
|
||||
if err := store.ReleaseInstanceMaintenance(ctx, "kgseal-instance-reentry", "job-reentry", now.Add(3*time.Second)); err != nil {
|
||||
t.Fatalf("release instance: %v", err)
|
||||
}
|
||||
next, err := store.AcquireInstanceMaintenance(ctx, LibraryInstanceMaintenance{
|
||||
Instance: "kgseal-instance-reentry", OwnerJobID: "job-next", OwnerScope: "internal",
|
||||
Level: "instance", State: "acquired", OriginalServiceState: "app=stopped",
|
||||
}, now.Add(4*time.Second))
|
||||
if err != nil {
|
||||
t.Fatalf("next owner acquire: %v", err)
|
||||
}
|
||||
if next.OriginalServiceState != "app=stopped" {
|
||||
t.Fatalf("new owner must record its own observation: got %q", next.OriginalServiceState)
|
||||
}
|
||||
}
|
||||
|
||||
// 实例级占用必须跨 scope 互斥:internal 持有期间 external 也进不来。
|
||||
// (旧实现按 scope 建行,两个 scope 各持一个锁,会同时停/起同一个实例。)
|
||||
func TestInstanceMaintenanceExcludesAcrossScopes(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user