diff --git a/api/openapi.yaml b/api/openapi.yaml index aecfad0..b5d9f3e 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -1547,6 +1547,58 @@ } } }, + "/api/v1/library/baselines": { + "post": { + "tags": [ + "Runtime" + ], + "operationId": "startLibraryBaseline", + "summary": "建立目标权威基线:管理员在维护隔离下把非空目标整库捕获成快照、逐类校验后登记为生效版本;不清空目标、不手写版本", + "responses": { + "202": { + "description": "已创建基线作业(worker 在维护窗口内捕获并校验)" + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "403": { + "description": "不是管理员" + }, + "409": { + "description": "已有生效版本、目标为空、目标正在维护或已有在飞写作业" + }, + "500": { + "$ref": "#/components/responses/InternalError" + }, + "503": { + "description": "目标整库读不到,不能建立基线(不创建作业、不登记版本)" + } + }, + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": false, + "required": [ + "scope" + ], + "properties": { + "scope": { + "type": "string", + "enum": [ + "internal", + "external" + ] + } + } + } + } + } + } + } + }, "/api/v1/library/snapshots/{id}": { "get": { "tags": [ diff --git a/cmd/api/main.go b/cmd/api/main.go index 0a652f3..163a637 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -203,6 +203,12 @@ func run(ctx context.Context, getenv func(string) string) error { VersionHistoryRepo: versionHistoryRepo, LibraryArchiveRoot: config.LibraryArchiveRoot, LibraryTargetInstance: config.LibraryTargetInstance, + LibraryTargetProject: config.LibraryTargetProject, + LibraryTargetWriters: config.LibraryTargetWriters, + LibraryTargetIngressURL: config.LibraryTargetIngressURL, + LibraryTargetIngressService: config.LibraryTargetIngressService, + LibraryTargetDockerSocket: config.LibraryTargetDockerSocket, + LibraryTargetJournalRoot: config.LibraryTargetJournalRoot, LibrarySourceInstance: config.SourceInstance(), LibrarySourceProject: config.SourceProject(), LibraryAgentSocket: config.LibraryAgentSocket, diff --git a/cmd/worker/library_baseline.go b/cmd/worker/library_baseline.go new file mode 100644 index 0000000..0dab748 --- /dev/null +++ b/cmd/worker/library_baseline.go @@ -0,0 +1,227 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log" + "time" + + "knowledge-governance/internal/appconfig" + "knowledge-governance/internal/governance" + "knowledge-governance/internal/jobs" + "knowledge-governance/internal/libraryfreeze" + "knowledge-governance/internal/librarymaintenance" + "knowledge-governance/internal/libraryrelease" + postgresstore "knowledge-governance/internal/platform/postgres" +) + +// libraryBaselinePayload 是"建立目标权威基线"的队列载荷。 +type libraryBaselinePayload struct { + LibraryJobID string `json:"library_job_id"` + Scope string `json:"scope"` +} + +// runLibraryBaseline 给"非空但没有权威版本"的目标建立基线。 +// +// 真实链路:维护隔离(关入口/停写入者/证明排空)→ 把目标整库捕获成归档快照 +// → 用同一套 Verify 证明目标与这份快照逐类一致 → 登记 production_effective。 +// 它不写目标内容、不清空目标、不猜版本:基线只是"目标现在是什么"的忠实记录, +// 因此失败时重新开放目标永远是安全的(allowRestore 恒 true)。 +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 == "" || + (payload.Scope != "internal" && payload.Scope != "external") { + return errors.New("invalid library baseline payload") + } + now := time.Now().UTC() + recorder := libraryStageRecorder{store: store, jobID: payload.LibraryJobID, scope: payload.Scope} + fail := func(message, stage string) error { + return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, message, stage) + } + _ = 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") + } + } + + if !config.ProductionReleaseRequestsEnabled { + return fail("生产写入闸口关闭(PRODUCTION_RELEASE_REQUESTS_ENABLED=false):建立基线要动目标运行状态,闸口关闭时不执行", "capture") + } + target, backups, wireErr := buildLibraryBaselineStack(ctx, config, payload.Scope) + if wireErr != nil { + return fail(wireErr.Error(), "capture") + } + defer target.Close() + + manager, managerErr := openLibraryMaintenance(config) + if managerErr != nil { + return fail("目标维护未接线,拒绝在挡不住入口/后台写入的情况下捕获目标:"+managerErr.Error(), "capture") + } + defer manager.Close() + if topoErr := manager.VerifyTopology(ctx); topoErr != nil { + return fail("目标拓扑不满足唯一入口约束,拒绝开维护窗口:"+topoErr.Error(), "capture") + } + originalState, stateErr := manager.OriginalState(ctx) + if stateErr != nil { + return fail("读不到目标写入者的原运行状态,不能开维护窗口:"+stateErr.Error(), "capture") + } + identity := manager.Target() + // 实例级占用:internal/external 动的是同一批容器,必须按实例互斥。 + if _, instanceErr := store.AcquireInstanceMaintenance(ctx, postgresstore.LibraryInstanceMaintenance{ + Instance: identity.Instance, OwnerJobID: payload.LibraryJobID, OwnerScope: payload.Scope, + TargetFingerprint: identity.Fingerprint, Level: librarymaintenance.LevelInstance, + Services: identity.Writers, IngressState: "unknown", State: "acquired", + OriginalServiceState: originalState, + Note: "实例级维护:为建立权威基线捕获目标整库,捕获并校验后按原状态恢复", + }, now); instanceErr != nil { + return fail("目标实例正在维护中(跨对内/对外互斥):"+instanceErr.Error(), "capture") + } + if _, acquireErr := store.AcquireLibraryMaintenance(ctx, postgresstore.LibraryMaintenance{ + Scope: payload.Scope, OwnerJobID: payload.LibraryJobID, + TargetIdentity: identity.Fingerprint, + Level: "instance", + TargetFingerprint: identity.Fingerprint, + Services: identity.Writers, + IngressState: "unknown", + State: "acquired", + OriginalServiceState: originalState, + Note: "实例级维护:为建立权威基线捕获目标整库,捕获并校验后按原状态恢复", + }, now); acquireErr != nil { + _ = store.UpdateInstanceMaintenanceState(ctx, identity.Instance, payload.LibraryJobID, "needs_restore", + "scope 占用申请失败,实例状态未确认:"+acquireErr.Error(), now) + return fail("目标正在维护中:"+acquireErr.Error(), "capture") + } + + coordinator, control, openErr := manager.Open(config.LibraryTargetJournalRoot) + if openErr != nil { + _ = releaseWindowLocks(ctx, store, identity.Instance, payload.Scope, payload.LibraryJobID, + "未进入维护窗口(协调器不可用),目标未被停用", time.Now().UTC()) + return fail("维护协调器不可用:"+openErr.Error(), "capture") + } + defer control.Close() + defer coordinator.Close() + + var baselineID string + var counts map[string]int64 + verified := false + work := func(ctx context.Context, window *libraryfreeze.Window) (string, error) { + 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()) + } + // 作业行也要立刻带上这份快照身份: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", "逐表/索引/资源比对目标与刚封存的快照") + if verifyErr := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: baselineID}); verifyErr != nil { + return "", errors.New("目标内容与刚封存的快照不一致,拒绝登记基线:" + verifyErr.Error()) + } + verified = true + recorder.Stage(ctx, "verify", "succeeded", "行身份/逐行内容/索引文档/资源摘要与快照一致") + 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) + } + + // 校验通过 → 登记权威基线。证据里记的是这份快照本身、逐表行数与目标索引映射; + // 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) + return nil +} diff --git a/cmd/worker/library_handlers.go b/cmd/worker/library_handlers.go index 331c6eb..795817e 100644 --- a/cmd/worker/library_handlers.go +++ b/cmd/worker/library_handlers.go @@ -480,8 +480,13 @@ func resumeInterruptedLibraryJobs(ctx context.Context, store *postgresstore.Stor continue } queueType := librarySyncJobType - if job.Kind == governance.LibraryJobRestore { + switch job.Kind { + case governance.LibraryJobRestore: queueType = libraryRestoreJobType + case governance.LibraryJobBaseline: + // 基线作业被杀时窗口可能还停着目标:用同一个作业 id 接管那个窗口, + // 继续捕获/校验,而不是留下一个 held 目标等人工处理。 + queueType = libraryBaselineJobType } payload, _ := json.Marshal(map[string]string{"library_job_id": job.ID, "scope": job.Scope, "target_snapshot_id": job.TargetSnapshotID}) diff --git a/cmd/worker/library_wiring.go b/cmd/worker/library_wiring.go index b4a6dfc..74f4aa6 100644 --- a/cmd/worker/library_wiring.go +++ b/cmd/worker/library_wiring.go @@ -11,11 +11,15 @@ import ( v071 "knowledge-governance/internal/releaseadapter/v071" ) -// buildLibraryProductionStack 用部署注入的配置装配生产整库目标与独立预备份存储。 -// 任何一项缺失都返回明确原因,调用方据此把作业失败并把缺项写进阶段, -// 不用"未配置"的占位对象假装能跑。 -func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, scope, snapshotID, knownMapping string) ( - *v071.ProductionLibraryTarget, *v071.ProductionBackupStore, error) { +// productionScopeConfig 是"某个 scope 的生产整库目标"的部署注入配置。 +// 校验、错误措辞与 S3 参数只在这里有一份:同步、恢复与基线共用同一套装配, +// 不允许任何一条路径自己拼一份(那会让"目标是谁"随路径漂移)。 +type productionScopeConfig struct { + target v071.ProductionScope + s3 v071.S3ObjectStoreConfig +} + +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") @@ -38,26 +42,53 @@ func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, s missing = append(missing, "LIBRARY_PRODUCTION_S3_ENDPOINT/BUCKET/ACCESS_KEY/SECRET_KEY") } if len(missing) > 0 { - return nil, nil, fmt.Errorf("生产整库目标未配置:缺少 %s", strings.Join(missing, "、")) + return productionScopeConfig{}, fmt.Errorf("生产整库目标未配置:缺少 %s", strings.Join(missing, "、")) } - secure := strings.HasPrefix(config.LibraryProductionS3Endpoint, "https://") - objects, err := v071.NewS3ObjectStore(v071.S3ObjectStoreConfig{ - Provider: config.LibraryProductionS3Provider, Endpoint: config.LibraryProductionS3Endpoint, - Region: config.LibraryProductionS3Region, Bucket: config.LibraryProductionS3Bucket, - AccessKeyID: config.LibraryProductionS3AccessKey, SecretAccessKey: config.LibraryProductionS3SecretKey, - Secure: secure, ForcePathStyle: true, MaxObjectBytes: 256 * 1024 * 1024, - }) + 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, nil, errors.New("生产对象存储配置无效") + return nil, errors.New("生产对象存储配置无效") } - identity := v071.ProductionScope{TenantID: config.LibraryProductionTenantID, KnowledgeBaseID: kb} - target, err := v071.NewProductionLibraryTarget(ctx, v071.ProductionLibraryConfig{ + 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, - }, identity) + }, 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 用部署注入的配置装配生产整库目标与独立预备份存储。 +// 任何一项缺失都返回明确原因,调用方据此把作业失败并把缺项写进阶段, +// 不用"未配置"的占位对象假装能跑。 +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 } @@ -105,11 +136,43 @@ func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, s if knownMapping != "" { backupContract.IndexMappingSHA256 = knownMapping } - backups, err := v071.NewProductionBackupStore(v071.ProductionBackupConfig{ - Pool: target.Pool(), Search: target.Search(), Objects: target.Objects(), - ArchiveRoot: config.LibraryArchiveRoot, Scope: scope, Identity: target.Identity(), - Compatibility: backupContract, MaxDocuments: 200000, MaxObjectByte: 256 * 1024 * 1024, - }) + backups, err := newProductionBackupStore(config, scope, target, backupContract) + if err != nil { + target.Close() + return nil, nil, err + } + return target, backups, nil +} + +// 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 diff --git a/cmd/worker/main.go b/cmd/worker/main.go index e72aaf2..acdefac 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -93,6 +93,9 @@ const libraryRecoverJobType = "library.recover" // libraryRestoreJobType 是按指定整库版本恢复的队列类型;与同步分开、授权也分开。 const libraryRestoreJobType = "library.restore" +// libraryBaselineJobType 是"建立目标权威基线"的队列类型(与 API 侧常量同值)。 +const libraryBaselineJobType = "library.baseline" + func main() { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() @@ -326,6 +329,10 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st handlers[libraryRestoreJobType] = func(ctx context.Context, job jobs.Job) error { return runLibraryRelease(ctx, store, config, job, governance.LibraryJobRestore) } + // 建立目标权威基线:目标非空又没有生效版本时的唯一入口(只读捕获 + 校验 + 登记)。 + handlers[libraryBaselineJobType] = func(ctx context.Context, job jobs.Job) error { + return runLibraryBaseline(ctx, store, config, job) + } // A crash can leave the source writer stopped: recover before taking work, // and keep the failure visible instead of pretending the worker is ready. go func() { diff --git a/internal/governance/library_flow.go b/internal/governance/library_flow.go index ed8bb7b..7e6b440 100644 --- a/internal/governance/library_flow.go +++ b/internal/governance/library_flow.go @@ -31,6 +31,10 @@ const ( LibraryJobSync LibraryJobKind = "sync" // LibraryJobRestore 是按指定整库版本恢复:与同步分开,授权与阶段也独立。 LibraryJobRestore LibraryJobKind = "restore" + // LibraryJobBaseline 建立首次发布目标的权威基线:在维护隔离下把目标整库 + // 读出来封存成快照并逐类校验,再登记成 production_effective。 + // 它不写目标内容、不清空目标、不猜版本——基线只能来自真实捕获与校验。 + LibraryJobBaseline LibraryJobKind = "baseline" ) type LibraryJobState string @@ -134,6 +138,12 @@ func SealStages() []string { return []string{"export"} } +// BaselineStages 是"建立目标权威基线"的阶段:捕获目标整库 → 与快照逐类校验 +// → 登记生效版本。没有 rollback 阶段:基线过程不写目标内容。 +func BaselineStages() []string { + return []string{"capture", "verify", "finalize"} +} + // LibraryReleaseAuthorization 是生产发布授权:绑定目标、快照、参数摘要与有效期。 type LibraryReleaseAuthorization struct { ID string `json:"id"` diff --git a/internal/httpapi/library_baseline.go b/internal/httpapi/library_baseline.go new file mode 100644 index 0000000..a96e803 --- /dev/null +++ b/internal/httpapi/library_baseline.go @@ -0,0 +1,150 @@ +package httpapi + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "time" + + "knowledge-governance/internal/governance" + "knowledge-governance/internal/jobs" + postgresstore "knowledge-governance/internal/platform/postgres" +) + +// 建立目标权威基线:首次发布时目标已经有内容,却没有生效版本记录。 +// +// 唯一诚实的解法是"维护隔离 + 真实整库捕获 + 逐类校验 + 登记版本": +// - 不手写 production_effective,不伪造验收; +// - 不清空目标换库绕过(那会把"非空但未知"变成"空"); +// - 页面只投递作业,停写入者、捕获、校验都在 worker 里执行。 +// +// 这个入口只建作业,不签发发布授权:基线不等于发布,同步仍然要单独的 sync 授权。 +const libraryBaselineJobType = "library.baseline" + +type libraryBaselineRequest struct { + Scope string `json:"scope"` +} + +func libraryBaselineEnqueue(w http.ResponseWriter, r *http.Request, config Config) { + if r.Method != http.MethodPost { + writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"}) + return + } + identity, ok := sessionFrom(r.Context()) + if !ok || roleRank(identity.Role) < roleRank(roleAdmin) { + writeJSON(w, http.StatusForbidden, map[string]any{"error": "建立目标权威基线仅管理员可发起"}) + return + } + var request libraryBaselineRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&request); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"}) + return + } + if request.Scope != "internal" && request.Scope != "external" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "scope must be internal or external"}) + return + } + store, ok := libraryStore(config) + if !ok { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_baseline_unavailable", + "error": "版本登记存储不可用"}) + return + } + versions, err := store.LibraryVersions(r.Context(), request.Scope, 50) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read library versions"}) + return + } + for _, item := range versions { + if item.Kind == governance.LibraryVersionProductionEffective && item.SupersededAt == nil { + writeJSON(w, http.StatusConflict, map[string]any{"code": "library_baseline_exists", + "error": "该目标已有生效版本,不需要再建基线;换版本请走同步或恢复", + "snapshot_id": item.SnapshotID}) + return + } + } + // 目标必须真的读得到:读不到就不是"非空无权威版本",而是另一个问题, + // 不允许用建基线去掩盖连接/配置故障。 + empty, counts, probeErr := productionScopeState(r.Context(), config, request.Scope) + if probeErr != nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{"code": "library_baseline_unprobeable", + "error": "无法读取目标整库状态,不能建立基线:" + probeErr.Error()}) + return + } + if empty { + writeJSON(w, http.StatusConflict, map[string]any{"code": "library_baseline_target_empty", + "error": "目标整库为空,不需要基线:首次同步会在写入前生成完整预备份并登记生效版本", + "counts": counts}) + return + } + // 维护互斥与同步一致:基线要在目标上开维护窗口,目标正在维护时直接拒绝。 + if body, held := instanceMaintenanceGuard(r.Context(), config, request.Scope); held { + writeJSON(w, http.StatusConflict, body) + return + } + if maintenanceStore, ok := config.Store.(interface { + LibraryMaintenanceOf(ctx context.Context, scope string) (*postgresstore.LibraryMaintenance, error) + }); ok { + if held, err := maintenanceStore.LibraryMaintenanceOf(r.Context(), request.Scope); err == nil && held != nil && held.ReleasedAt == nil { + writeJSON(w, http.StatusConflict, map[string]any{"code": "library_target_in_maintenance", + "error": "目标正在维护中,验证完成前不接受新的写入请求", + "owner_job": held.OwnerJobID, "state": held.State}) + return + } + } + digest, err := governance.LibraryRequestDigest(request.Scope, governance.LibraryJobBaseline, + map[string]any{"scope": request.Scope}) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot digest request"}) + return + } + now := time.Now().UTC() + job, reused, err := store.CreateLibraryJob(r.Context(), governance.LibraryJob{ + ID: fmt.Sprintf("libbase-%s-%d", request.Scope, now.UnixNano()), Scope: request.Scope, + Kind: governance.LibraryJobBaseline, + IdempotencyKey: "library.baseline:" + request.Scope + ":" + now.Format("20060102T150405.000000000"), + RequestDigest: digest, Actor: identity.Username, + }, governance.BaselineStages(), now) + if errors.Is(err, governance.ErrLibraryJobActive) { + writeJSON(w, http.StatusConflict, map[string]any{"code": "library_write_active", + "error": "该目标已有在飞的写作业(同步/恢复/基线互斥),等它结束再发起"}) + return + } + if errors.Is(err, governance.ErrLibraryJobIdempotencyConflict) { + writeJSON(w, http.StatusConflict, map[string]any{"code": "library_baseline_conflict", + "error": "同一幂等键对应了不同的请求"}) + return + } + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot create library baseline job"}) + return + } + if reused { + writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": true}) + return + } + payload, err := json.Marshal(map[string]string{"library_job_id": job.ID, "scope": request.Scope}) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot build baseline payload"}) + return + } + queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{ + ID: governance.QueueJobIDFor(job.ID), Type: libraryBaselineJobType, Payload: payload, + AvailableAt: now, MaxAttempts: 1, IdempotencyKey: "library.baseline.queue:" + job.ID, + }, now) + if err != nil || queueJob == nil { + // 作业先落库、队列投递失败就把作业取消:不留下"看起来在跑"的空作业, + // 否则它会一直占着该 scope 的写互斥。 + _ = store.CancelLibraryJob(r.Context(), job.ID, "队列投递失败,未开始建立基线", now) + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue baseline job"}) + return + } + if attachErr := store.AttachLibraryQueueJob(r.Context(), job.ID, queueJob.ID, now); attachErr != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot attach baseline job"}) + return + } + writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": false, + "target_counts": counts}) +} diff --git a/internal/httpapi/library_baseline_test.go b/internal/httpapi/library_baseline_test.go new file mode 100644 index 0000000..ce83293 --- /dev/null +++ b/internal/httpapi/library_baseline_test.go @@ -0,0 +1,162 @@ +package httpapi + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "knowledge-governance/internal/governance" +) + +// 建立目标权威基线的入口契约: +// - 只有管理员能发起(它是生产侧维护动作); +// - 已有生效版本时拒绝(基线只解决首次发布); +// - 目标读不到时拒绝,且**不建作业、不登记版本**(不能用基线掩盖连接问题)。 + +func baselineRequest(body, role, username string) *http.Request { + request := httptest.NewRequest(http.MethodPost, "/api/v1/library/baselines", strings.NewReader(body)) + request.Header.Set("Content-Type", "application/json") + ctx := context.WithValue(request.Context(), sessionKey{}, sessionIdentity{Username: username, Role: role}) + return request.WithContext(ctx) +} + +func decodeBody(t *testing.T, recorder *httptest.ResponseRecorder) map[string]any { + t.Helper() + var payload map[string]any + if err := json.Unmarshal(recorder.Body.Bytes(), &payload); err != nil { + t.Fatalf("response is not JSON: %s", recorder.Body.String()) + } + return payload +} + +func TestLibraryBaselineRequiresAdmin(t *testing.T) { + store := &fakeLibraryStore{} + recorder := httptest.NewRecorder() + libraryBaselineEnqueue(recorder, baselineRequest(`{"scope":"internal"}`, roleEngineer, "eng"), Config{Store: store}) + if recorder.Code != http.StatusForbidden { + t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String()) + } + if store.createN != 0 { + t.Fatal("non-admin must not create a baseline job") + } +} + +func TestLibraryBaselineRefusesWhenEffectiveVersionExists(t *testing.T) { + store := &fakeLibraryStore{versions: []governance.LibraryVersion{{ + ID: "libver-1", Scope: "internal", Kind: governance.LibraryVersionProductionEffective, + SnapshotID: "snap-effective", CreatedAt: time.Now().UTC(), + }}} + recorder := httptest.NewRecorder() + libraryBaselineEnqueue(recorder, baselineRequest(`{"scope":"internal"}`, roleAdmin, "admin"), Config{Store: store}) + if recorder.Code != http.StatusConflict { + t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String()) + } + if body := decodeBody(t, recorder); body["code"] != "library_baseline_exists" || body["snapshot_id"] != "snap-effective" { + t.Fatalf("body=%v", body) + } + if store.createN != 0 { + t.Fatal("existing effective version must not create another baseline job") + } +} + +// 目标读不到(这里用"没配置目标库"复现):必须 503 并说明原因, +// 既不能建作业,也不能顺手写一条版本记录来"解决"未知状态。 +func TestLibraryBaselineRefusesWhenTargetUnreadable(t *testing.T) { + store := &fakeLibraryStore{} + recorder := httptest.NewRecorder() + libraryBaselineEnqueue(recorder, baselineRequest(`{"scope":"external"}`, roleAdmin, "admin"), Config{Store: store}) + if recorder.Code != http.StatusServiceUnavailable { + t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String()) + } + body := decodeBody(t, recorder) + if body["code"] != "library_baseline_unprobeable" || !strings.Contains(func() string { + text, _ := body["error"].(string) + return text + }(), "无法读取目标整库状态") { + t.Fatalf("body=%v", body) + } + if store.createN != 0 { + t.Fatal("unreadable target must not create a baseline job") + } + if len(store.versions) != 0 { + t.Fatal("unreadable target must not record any version") + } +} + +// 测试候选只认登记过 test_sealed 的那份快照:归档里同时存在更新的目标基线/ +// 预备份快照时,绝不能把它们当成"拟发布的测试版本"。 +func TestTestSealedCandidateComesFromRegistryNotNewestArchiveEntry(t *testing.T) { + sealed := librarySnapshotView{SnapshotID: "snap-sealed", CreatedAt: time.Date(2026, 10, 29, 8, 54, 0, 0, time.UTC)} + prebackup := librarySnapshotView{SnapshotID: "snap-prebackup", CreatedAt: time.Date(2026, 10, 29, 9, 30, 0, 0, time.UTC)} + byID := map[string]librarySnapshotView{sealed.SnapshotID: sealed, prebackup.SnapshotID: prebackup} + versions := []governance.LibraryVersion{ + {Kind: governance.LibraryVersionProductionPrebackup, SnapshotID: "snap-prebackup"}, + {Kind: governance.LibraryVersionTestSealed, SnapshotID: "snap-sealed"}, + } + candidate, reason := testSealedOf(versions, byID) + if reason != "" { + t.Fatalf("reason=%q", reason) + } + if candidate == nil || candidate.SnapshotID != "snap-sealed" { + t.Fatalf("candidate=%v, want the registered test seal", candidate) + } + + // 没有任何 test_sealed 登记:宁可说"还没有测试封存",也不拿别的快照顶替。 + none, reason := testSealedOf([]governance.LibraryVersion{ + {Kind: governance.LibraryVersionProductionPrebackup, SnapshotID: "snap-prebackup"}, + }, byID) + if none != nil || !strings.Contains(reason, "还没有该目标的测试封存") { + t.Fatalf("candidate=%v reason=%q", none, reason) + } + + // 登记过但它已经不在本机归档:说清是归档问题,不给候选。 + missing, reason := testSealedOf([]governance.LibraryVersion{ + {Kind: governance.LibraryVersionTestSealed, SnapshotID: "snap-gone"}, + }, byID) + if missing != nil || !strings.Contains(reason, "不在本机归档里") { + t.Fatalf("candidate=%v reason=%q", missing, reason) + } +} + +// 非空但无权威版本时,缺项必须指向真实入口,而不是一句"未知"。 +func TestLibrarySyncGateNamesTheBaselineEntryForNonEmptyTarget(t *testing.T) { + gate := librarySyncGateWithGrant(Config{ProductionReleaseRequestsEnabled: true, + ProductionBackupRestoreVerified: true, LibraryProductionDatabaseURL: "postgres://prod.invalid/db", + LibraryProductionInternalKB: "kb-int", LibraryCapacityBudgetBytes: 1 << 30}, + libraryKnownView{State: "unknown"}, libraryKnownView{State: "unknown"}, "snap-1", true) + if gate.SyncAllowed { + t.Fatal("unknown production version must block the sync") + } + joined := strings.Join(gate.Reasons, "|") + if !strings.Contains(joined, "目标非空且没有生效版本记录") || !strings.Contains(joined, "目标权威基线") { + t.Fatalf("reasons=%v", gate.Reasons) + } +} + +// 维护能力是配置事实:配置齐全时必须说"已接线",缺项要列出缺哪个变量。 +func TestLibraryMaintenanceCapabilityReportsWiringInsteadOfOneVagueSentence(t *testing.T) { + wired := libraryMaintenanceCapability(Config{ + LibraryTargetInstance: "kgseal-app", LibraryTargetProject: "kgseal-weknora", + LibraryTargetWriters: []string{"app"}, LibraryTargetIngressURL: "http://kgseal-ingress:18084/health", + LibraryTargetIngressService: "ingress", LibraryTargetJournalRoot: "/var/lib/knowledge-governance/journal", + LibraryTargetDockerSocket: "/var/run/docker.sock", + }) + if wired["configured"] != true { + t.Fatalf("capability=%v", wired) + } + if wired["instance"] != "kgseal-app" { + t.Fatalf("instance=%v", wired["instance"]) + } + broken := libraryMaintenanceCapability(Config{LibraryTargetInstance: "kgseal-app"}) + if broken["configured"] != false { + t.Fatalf("capability=%v", broken) + } + missing, _ := broken["missing"].([]string) + if len(missing) == 0 || !strings.Contains(strings.Join(missing, ","), "LIBRARY_TARGET_WRITERS") { + t.Fatalf("missing=%v", broken["missing"]) + } +} diff --git a/internal/httpapi/library_flow.go b/internal/httpapi/library_flow.go index 19f355f..191b427 100644 --- a/internal/httpapi/library_flow.go +++ b/internal/httpapi/library_flow.go @@ -19,6 +19,7 @@ import ( "knowledge-governance/internal/librarydiff" "knowledge-governance/internal/librarysnapshot" postgresstore "knowledge-governance/internal/platform/postgres" + v071 "knowledge-governance/internal/releaseadapter/v071" ) // 整库版本/状态/差异读取契约 + 生产同步的 fail-closed 入口。 @@ -56,30 +57,36 @@ type libraryKnownView struct { EmptyBaseline bool `json:"empty_baseline,omitempty"` } -// productionScopeEmpty 只读探测生产目标:真正连库数一遍,不用缓存或推断。 -func productionScopeEmpty(ctx context.Context, config Config, scope string) (bool, error) { +// productionScopeState 只读探测生产目标:真正连库数一遍,不用缓存或推断。 +// +// "空"必须是整库意义的空:只数 knowledges 会把"文档已删、分块/Wiki/资源还在" +// 的目标当成首次发布的空基线,从而绕过基线初始化。所以这里用适配层同一份 +// 表清单与范围过滤(v071.ScopeContentEmpty),并把逐表行数交回调用方作为证据。 +func productionScopeState(ctx context.Context, config Config, scope string) (bool, map[string]int64, error) { if strings.TrimSpace(config.LibraryProductionDatabaseURL) == "" { - return false, errors.New("生产整库目标未配置") + return false, nil, errors.New("生产整库目标未配置") } kb := config.LibraryProductionInternalKB if scope == "external" { kb = config.LibraryProductionExternalKB } if kb == "" { - return false, errors.New("生产知识库未配置") + return false, nil, errors.New("该范围的目标知识库未配置") } pool, err := pgxpool.New(ctx, config.LibraryProductionDatabaseURL) if err != nil { - return false, err + return false, nil, err } defer pool.Close() - var count int64 - err = pool.QueryRow(ctx, `SELECT count(*) FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2 AND deleted_at IS NULL`, - config.LibraryProductionTenantID, kb).Scan(&count) - if err != nil { - return false, err - } - return count == 0, nil + return v071.ScopeContentEmpty(ctx, pool, + v071.ProductionScope{TenantID: config.LibraryProductionTenantID, KnowledgeBaseID: kb}) +} + +// productionScopeEmpty 保留"只问空不空"的调用形状;逐表行数在需要举证时用 +// productionScopeState 取。 +func productionScopeEmpty(ctx context.Context, config Config, scope string) (bool, error) { + empty, _, err := productionScopeState(ctx, config, scope) + return empty, err } type libraryDiffView struct { @@ -508,6 +515,70 @@ func errString(err error) string { return err.Error() } +// shortSnapshot 只用于给人看的提示:完整摘要留给快照详情,别在原因句里塞 64 位。 +func shortSnapshot(id string) string { + if len(id) <= 12 { + return id + } + return id[:12] +} + +// testSealedOf 选"拟发布的测试封存":只有登记过 test_sealed 的那一份才算候选。 +// +// 归档是共享的:生产预备份与目标基线快照同样带着 scope 写进同一个归档根, +// 按"这个 scope 最新一份"挑会把目标自己的内容当成测试候选(页面会显示一份 +// 永远不该被发布的东西)。所以候选来自版本登记,再用归档确认它真的还在本地。 +func testSealedOf(versions []governance.LibraryVersion, byID map[string]librarySnapshotView) (*librarySnapshotView, string) { + for _, item := range versions { + if item.Kind != governance.LibraryVersionTestSealed { + continue + } + snapshot, ok := byID[item.SnapshotID] + if !ok { + return nil, "最近一次测试封存 " + shortSnapshot(item.SnapshotID) + " 不在本机归档里(归档被清理或换了归档根)" + } + return &snapshot, "" + } + return nil, "还没有该目标的测试封存" +} + +// libraryMaintenanceCapability 用配置事实回答"这台机器上维护是否接线"。 +// 它刻意只说配置:能停写入者/关入口/证明排空的真实能力在 worker 侧(docker socket、 +// 闸门、排空文件),API 若声称"能力可用"就是拿配置字符串冒充执行证据。 +func libraryMaintenanceCapability(config Config) map[string]any { + missing := make([]string, 0, 6) + if config.LibraryTargetInstance == "" { + missing = append(missing, "LIBRARY_TARGET_INSTANCE") + } + if config.LibraryTargetProject == "" { + missing = append(missing, "LIBRARY_TARGET_PROJECT") + } + if len(config.LibraryTargetWriters) == 0 { + missing = append(missing, "LIBRARY_TARGET_WRITERS") + } + if config.LibraryTargetIngressURL == "" { + missing = append(missing, "LIBRARY_TARGET_INGRESS_URL") + } + if config.LibraryTargetIngressService == "" { + missing = append(missing, "LIBRARY_TARGET_INGRESS_SERVICE") + } + if config.LibraryTargetJournalRoot == "" { + missing = append(missing, "LIBRARY_TARGET_JOURNAL_ROOT") + } + if config.LibraryTargetDockerSocket == "" { + missing = append(missing, "LIBRARY_TARGET_DOCKER_SOCKET") + } + return map[string]any{ + "configured": len(missing) == 0, + "missing": missing, + "instance": config.LibraryTargetInstance, + "project": config.LibraryTargetProject, + "writers": config.LibraryTargetWriters, + "ingress_service": config.LibraryTargetIngressService, + "note": "配置齐全只说明治理侧接线完整;停写入者、关入口与排空由 worker 在作业内校验并执行", + } +} + func librarySyncGate(config Config, production libraryKnownView, prebackup libraryKnownView, target string) libraryGateView { return librarySyncGateWithGrant(config, production, prebackup, target, false) } @@ -531,9 +602,11 @@ func librarySyncGateWithGrant(config Config, production libraryKnownView, prebac gate.Reasons = append(gate.Reasons, "生产整库目标未配置(LIBRARY_PRODUCTION_DATABASE_URL),无法核验生产当前状态") } // 首次发布:生产还没有"生效版本"记录,但只要生产目标可读且是空的, - // 这就是一个合法的"空基线",不应该被永久挡住。非空但无权威版本仍然阻断。 + // 这就是一个合法的"空基线",不应该被永久挡住。非空但无权威版本仍然阻断, + // 指向真实可行的入口(在维护隔离下捕获目标整库并校验后登记基线), + // 而不是让人去手写版本或清空目标。 if production.State != "known" && !production.EmptyBaseline { - gate.Reasons = append(gate.Reasons, "生产当前权威版本未知,不能启动写入") + gate.Reasons = append(gate.Reasons, "目标非空且没有生效版本记录:先建立目标权威基线(维护隔离下捕获整库并逐类校验),或确认目标确实是空库") } if target == "" { gate.Reasons = append(gate.Reasons, "还没有可同步的测试封存版本") @@ -569,21 +642,20 @@ func libraryOverviewHandler(w http.ResponseWriter, r *http.Request, config Confi // 可发起封存 = worker 真实探测到执行器可用且指向本实例;原因逐条给全。 overview.CaptureAvailable, overview.CaptureReason, overview.Executor = libraryExecutorAvailability(r.Context(), config) - // 测试封存来自归档:它只是测试侧事实。 + // 测试封存来自归档 + 版本登记:归档里**不止**测试封存——生产预备份与目标基线 + // 快照同样带 scope,如果只按"这个 scope 最新一份"挑,页面就会把目标自己的内容 + // 当成拟发布的测试候选。所以候选只认登记过 kind=test_sealed 的那一份。 snapshots, snapErr := readLibrarySnapshots(r.Context(), config.LibraryArchiveRoot) + byID := map[string]librarySnapshotView{} + archiveReadable := false if snapErr != nil { _, overview.TestSealedReason = libraryArchiveErrorStatus(snapErr, "归档不可读") } else if !snapshots.Available { overview.TestSealedReason = snapshots.Reason } else { - for index := range snapshots.Snapshots { - if snapshots.Snapshots[index].Scope == scope { - overview.TestSealed = &snapshots.Snapshots[index] - break - } - } - if overview.TestSealed == nil { - overview.TestSealedReason = "还没有该目标的测试封存" + archiveReadable = true + for _, snapshot := range snapshots.Snapshots { + byID[snapshot.SnapshotID] = snapshot } } @@ -593,6 +665,9 @@ func libraryOverviewHandler(w http.ResponseWriter, r *http.Request, config Confi writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read library versions"}) return } + if archiveReadable { + overview.TestSealed, overview.TestSealedReason = testSealedOf(versions, byID) + } overview.Production = libraryKnownView{State: "unknown", Reason: "还没有生产验证通过的生效版本记录"} overview.Prebackup = libraryKnownView{State: "unknown", Reason: "本次目标快照还没有生产预备份"} target := "" diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 1b1c4cf..bb53029 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -148,6 +148,15 @@ type Config struct { // LibraryTargetInstance 是实例级维护的目标实例名;留空即"未接线", // 页面不能发起按预备份恢复,接口返回原因而不是假装能恢复。 LibraryTargetInstance string + // 维护能力的其余配置项只用于如实展示"是否接线":未运行维护与没有维护能力 + // 是两件事,页面不能混成一句"未接入"。真正停写入者/关入口的能力由 worker + // 在执行作业时校验(socket、入口、排空文件都在 worker 侧)。 + LibraryTargetProject string + LibraryTargetWriters []string + LibraryTargetIngressURL string + LibraryTargetIngressService string + LibraryTargetDockerSocket string + LibraryTargetJournalRoot string // LibrarySourceInstance 是死信解除记录的来源实例名(封存源),留空回落到目标实例名。 LibrarySourceInstance string // LibraryAgentSocket 是主机侧封存 agent 的 unix socket;留空即页面不能发起封存。 @@ -479,6 +488,8 @@ func NewHandler(config Config) (http.Handler, error) { return } response := map[string]any{"held": held != nil && held.ReleasedAt == nil, "maintenance": held} + // 有没有维护能力是配置事实,和"此刻没有窗口"必须分开回答。 + response["capability"] = libraryMaintenanceCapability(config) // 实例级占用才是真正的互斥来源:跨 scope 也在这里体现。 if instanceStore, ok := config.Store.(interface { InstanceMaintenanceOf(ctx context.Context, instance string) (*postgresstore.LibraryInstanceMaintenance, error) @@ -527,6 +538,10 @@ func NewHandler(config Config) (http.Handler, error) { mux.HandleFunc("POST /api/v1/library/syncs", func(w http.ResponseWriter, r *http.Request) { librarySyncEnqueue(w, r, config) }) + // 首次发布的目标权威基线:管理员在维护隔离下捕获目标整库并校验后登记版本。 + mux.HandleFunc("POST /api/v1/library/baselines", func(w http.ResponseWriter, r *http.Request) { + libraryBaselineEnqueue(w, r, config) + }) // 全部 main 提交时间线(读本库缓存,不直接打 Gitea) mux.HandleFunc("GET /api/v1/version-history", func(w http.ResponseWriter, r *http.Request) { versionHistoryList(w, r, config) diff --git a/internal/platform/postgres/library_jobs.go b/internal/platform/postgres/library_jobs.go index 83c7b92..66ad5fb 100644 --- a/internal/platform/postgres/library_jobs.go +++ b/internal/platform/postgres/library_jobs.go @@ -261,7 +261,7 @@ func (s *Store) InterruptedLibraryJobs(ctx context.Context) ([]governance.Librar rows, err := s.db.Query(ctx, ` SELECT lj.id FROM library_jobs lj LEFT JOIN jobs j ON j.id=lj.queue_job_id -WHERE lj.kind IN ('sync','restore') +WHERE lj.kind IN ('sync','restore','baseline') AND ( -- 业务作业在跑、队列已经没了/停了/租约过期 (lj.state='running' AND (j.id IS NULL OR j.state IN ('failed','completed','cancelled','dead') diff --git a/internal/platform/postgres/migrations/000075_library_baseline_kind.down.sql b/internal/platform/postgres/migrations/000075_library_baseline_kind.down.sql new file mode 100644 index 0000000..a863f58 --- /dev/null +++ b/internal/platform/postgres/migrations/000075_library_baseline_kind.down.sql @@ -0,0 +1,8 @@ +DELETE FROM library_jobs WHERE kind='baseline'; +ALTER TABLE library_jobs DROP CONSTRAINT IF EXISTS library_jobs_kind_check; +ALTER TABLE library_jobs ADD CONSTRAINT library_jobs_kind_check + CHECK (kind IN ('seal','sync','restore')); +DROP INDEX IF EXISTS library_jobs_write_target; +CREATE UNIQUE INDEX library_jobs_write_target + ON library_jobs(scope) + WHERE state IN ('pending','running') AND kind IN ('sync','restore'); \ No newline at end of file diff --git a/internal/platform/postgres/migrations/000075_library_baseline_kind.up.sql b/internal/platform/postgres/migrations/000075_library_baseline_kind.up.sql new file mode 100644 index 0000000..741515e --- /dev/null +++ b/internal/platform/postgres/migrations/000075_library_baseline_kind.up.sql @@ -0,0 +1,12 @@ +-- 允许"建立目标权威基线"作为一种独立业务作业:首次发布的目标非空又没有生效版本时, +-- 只能通过真实捕获 + 校验登记基线,不能手写版本记录。 +-- 基线会在目标上开维护窗口(停写入者),所以它与 sync/restore 共用同一把写互斥: +-- 同一 scope 不允许同时存在两种写作业。 +ALTER TABLE library_jobs DROP CONSTRAINT IF EXISTS library_jobs_kind_check; +ALTER TABLE library_jobs ADD CONSTRAINT library_jobs_kind_check + CHECK (kind IN ('seal','sync','restore','baseline')); + +DROP INDEX IF EXISTS library_jobs_write_target; +CREATE UNIQUE INDEX library_jobs_write_target + ON library_jobs(scope) + WHERE state IN ('pending','running') AND kind IN ('sync','restore','baseline'); \ No newline at end of file diff --git a/internal/platform/postgres/migrations_test.go b/internal/platform/postgres/migrations_test.go index e15f24e..e638bba 100644 --- a/internal/platform/postgres/migrations_test.go +++ b/internal/platform/postgres/migrations_test.go @@ -11,45 +11,51 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) { if err != nil { t.Fatalf("loadMigrations() error = %v", err) } - if len(migrations) != 74 { - t.Fatalf("migration count = %d, want 74", len(migrations)) + if len(migrations) != 75 { + t.Fatalf("migration count = %d, want 75", len(migrations)) + } + // 75 允许"建立目标权威基线"作为独立业务作业,并与 sync/restore 共用写互斥。 + if baseline := migrations[len(migrations)-1]; baseline.Version != 75 || + !strings.Contains(baseline.UpSQL, "library_jobs_write_target") || + !strings.Contains(baseline.UpSQL, "'baseline'") { + t.Fatalf("migration 75 = %#v", baseline) } // 74 记录封存执行器的运行时状态:页面可用性来自真实观测,不来自配置字符串。 - if executor := migrations[len(migrations)-1]; executor.Version != 74 || + if executor := migrations[len(migrations)-2]; executor.Version != 74 || !strings.Contains(executor.UpSQL, "library_executor_status") { t.Fatalf("migration 74 = %#v", executor) } // 73 记录死信精确解除:保留上游行,只绑定指纹、重放引用与观测摘要。 - if release := migrations[len(migrations)-2]; release.Version != 73 || + if release := migrations[len(migrations)-3]; release.Version != 73 || !strings.Contains(release.UpSQL, "library_dead_letter_releases") { t.Fatalf("migration 73 = %#v", release) } // 72 让维护占用按实例互斥(internal/external 不再各有各的锁)。 - if instance := migrations[len(migrations)-3]; instance.Version != 72 || + if instance := migrations[len(migrations)-4]; instance.Version != 72 || !strings.Contains(instance.UpSQL, "library_instance_maintenance") { t.Fatalf("migration 72 = %#v", instance) } // 71 让维护记录能区分实例级维护与仅治理库互斥。 - if level := migrations[len(migrations)-4]; level.Version != 71 || + if level := migrations[len(migrations)-5]; level.Version != 71 || !strings.Contains(level.UpSQL, "level") || !strings.Contains(level.UpSQL, "ingress_state") { t.Fatalf("migration 71 = %#v", level) } - if maint := migrations[len(migrations)-5]; maint.Version != 70 || !strings.Contains(maint.UpSQL, "library_maintenance") { + if maint := migrations[len(migrations)-6]; maint.Version != 70 || !strings.Contains(maint.UpSQL, "library_maintenance") { t.Fatalf("migration 70 = %#v", maint) } - if tail := migrations[len(migrations)-6]; tail.Version != 69 || + if tail := migrations[len(migrations)-7]; tail.Version != 69 || !strings.Contains(tail.UpSQL, "library_jobs_write_target") { t.Fatalf("migration 69 = %#v", tail) } - if restore := migrations[len(migrations)-7]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") { + if restore := migrations[len(migrations)-8]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") { t.Fatalf("migration 68 = %#v", restore) } - if last := migrations[len(migrations)-8]; last.Version != 67 || + if last := migrations[len(migrations)-9]; last.Version != 67 || !strings.Contains(last.UpSQL, "library_release_authorizations") || !strings.Contains(last.UpSQL, "params_digest") { t.Fatalf("migration 67 = %#v", last) } - if flow := migrations[len(migrations)-9]; flow.Version != 66 || + if flow := migrations[len(migrations)-10]; flow.Version != 66 || !strings.Contains(flow.UpSQL, "CREATE TABLE library_versions") || !strings.Contains(flow.UpSQL, "library_jobs_active") { t.Fatalf("migration 66 = %#v", flow) diff --git a/internal/releaseadapter/v071/library_scope_probe.go b/internal/releaseadapter/v071/library_scope_probe.go new file mode 100644 index 0000000..d25ec8e --- /dev/null +++ b/internal/releaseadapter/v071/library_scope_probe.go @@ -0,0 +1,46 @@ +package v071 + +import ( + "context" + "errors" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// ScopeContentCounts 数一个知识库在各内容表里的行数(数据库侧事实)。 +// +// 为什么不能只数 knowledges:文档被删掉、分块与 Wiki/资源还在的库并不为空。 +// 只数一张表会把这种目标判成"首次发布的空库",从而让一条本该走基线初始化的 +// 路径绕过去;这里用与整库读写完全相同的表清单与过滤条件,保证"空"的定义只有一份。 +func ScopeContentCounts(ctx context.Context, pool *pgxpool.Pool, identity ProductionScope) (map[string]int64, error) { + if pool == nil || identity.TenantID == 0 || !validSearchIdentifier(identity.KnowledgeBaseID) { + return nil, errors.New("invalid scope identity") + } + counts := make(map[string]int64, len(productionLibraryTables)) + for _, table := range productionLibraryTables { + var count int64 + if err := pool.QueryRow(ctx, fmt.Sprintf(`SELECT count(*) FROM %s WHERE %s`, table, scopeWhere(table)), + identity.TenantID, identity.KnowledgeBaseID).Scan(&count); err != nil { + return nil, fmt.Errorf("count %s: %w", table, err) + } + counts[table] = count + } + return counts, nil +} + +// ScopeContentEmpty 只有每张内容表都是 0 才算空目标,并交回逐表行数作为证据。 +// 注意它**不**包含检索索引:索引由分块派生,分块为空而索引非空属于不一致, +// 应在执行侧(预备份/基线捕获)报出来,不能在这里被当成"空"。 +func ScopeContentEmpty(ctx context.Context, pool *pgxpool.Pool, identity ProductionScope) (bool, map[string]int64, error) { + counts, err := ScopeContentCounts(ctx, pool, identity) + if err != nil { + return false, nil, err + } + for _, count := range counts { + if count != 0 { + return false, counts, nil + } + } + return true, counts, nil +}