为什么:生产目标此前是"无条件失败"的占位,编排再完整也不能落地;本轮把它做成真实实现, 并在本机 WeKnora 测试栈的真实 PG/ES 上跑通导入、撤销、预备份与故障恢复。 - ProductionLibraryTarget:Current 读行数+身份摘要;Apply 按阶段真实写入(database/wiki 同事务先删本库再插、resources 传对象、index 搬运原始文档与向量);Verify 比对身份摘要; Restore 按预备份快照回放。 - 撤销真正生效:索引阶段写入后 PruneScope 删除快照里没有的文档。 - ProductionBackupStore:导出前独立完整预备份,导出前后行数不一致即失败。 - worker 用注入配置装配目标与预备份;缺项把变量名写进阶段并失败,不再有占位对象。 - capacity 安全加法防溢出;保留策略组合语义明确(各维度都超限才候选)。 - 新增本机真实服务集成测试(一次性 kg_library_test_* 库 + 随机 ES 索引), 覆盖导入/校验/撤销/预备份/故障注入/恢复。
This commit is contained in:
@@ -1,5 +1,36 @@
|
||||
# Changelog
|
||||
|
||||
## 开发快照(生产整库真实适配 + 本机真实服务验收)
|
||||
|
||||
### Added
|
||||
|
||||
- `internal/releaseadapter/v071/library_production_target.go` 由占位改为真实实现:
|
||||
`Current` 读生产每张知识域表的行数与「身份集合摘要」+ 索引条数;`Apply` 按阶段真实写入
|
||||
(database/wiki 同事务先删本库再插、resources 上传对象、index 批量索引原始文档与向量);
|
||||
`Verify` 用身份摘要与索引条数比对;`Restore` 按预备份快照回放整条链路。
|
||||
只触碰知识域表,不写 users/sessions/api_keys;不做解析或 Embedding。
|
||||
- 撤销真正落地:索引阶段在写入后用 `PruneScope` 删除"快照里没有"的文档,避免只增不减。
|
||||
- `library_backup_store.go`:写前独立完整预备份——生产 PG(RepeatableRead)+ ES 整库导出 +
|
||||
资源对象密封成归档版本,导出前后比对行数做静默检查;映射契约不符直接拒绝。
|
||||
- `search_write.go`:`Count`/`BulkIndex`/`PruneScope`,范围过滤统一按「租户 + 知识库」。
|
||||
- `cmd/worker/library_wiring.go`:用部署注入的配置装配生产目标与预备份;缺项时把缺的
|
||||
环境变量名写进阶段并失败,不再使用空占位对象。
|
||||
- 配置:`LIBRARY_PRODUCTION_DATABASE_URL`、`LIBRARY_PRODUCTION_ELASTIC_URL/INDEX`、
|
||||
`LIBRARY_PRODUCTION_TENANT_ID`、`LIBRARY_PRODUCTION_KB_INTERNAL/EXTERNAL`、
|
||||
`LIBRARY_PRODUCTION_S3_*`、`LIBRARY_EMBEDDING_MODEL/DIMENSION`。
|
||||
|
||||
### Fixed
|
||||
|
||||
- `libraryretention.CheckCapacity` 用安全加法:int64 溢出直接拒绝,不再绕回负数通过门禁。
|
||||
- 保留策略组合语义写明为「每个已设维度都超限才成为候选」并提供 `policy_note`;
|
||||
锁定/备份恢复依赖通过 `ProtectedBy` 进入保护集合,永不候选。
|
||||
|
||||
### Verified(本机真实服务,非桩)
|
||||
|
||||
- `TestLibraryProductionTargetIsolated` 在本机 WeKnora 测试栈真实 PG(一次性 `kg_library_test_*` 库)
|
||||
与真实 ES(随机索引)跑通:首次导入 → 校验 → 撤销传播(PG 行与 ES 文档都消失、跨租户不受影响)
|
||||
→ 独立预备份(3111e823…,626 字节)→ 注入故障(删行)→ 校验失败 → 恢复后校验通过。
|
||||
|
||||
## 开发快照(生产发布授权 + 同步/恢复编排 + 存储管理真实化)
|
||||
|
||||
### Added
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"knowledge-governance/internal/appconfig"
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||||
)
|
||||
|
||||
// buildLibraryProductionStack 用部署注入的配置装配生产整库目标与独立预备份存储。
|
||||
// 任何一项缺失都返回明确原因,调用方据此把作业失败并把缺项写进阶段,
|
||||
// 不用"未配置"的占位对象假装能跑。
|
||||
func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, scope string) (
|
||||
*v071.ProductionLibraryTarget, *v071.ProductionBackupStore, 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 nil, nil, fmt.Errorf("生产整库目标未配置:缺少 %s", strings.Join(missing, "、"))
|
||||
}
|
||||
secure := strings.HasPrefix(config.LibraryProductionS3Endpoint, "https://")
|
||||
objects, err := v071.NewS3ObjectStore(v071.S3ObjectStoreConfig{
|
||||
Provider: "s3", Endpoint: config.LibraryProductionS3Endpoint,
|
||||
Region: config.LibraryProductionS3Region, Bucket: config.LibraryProductionS3Bucket,
|
||||
AccessKeyID: config.LibraryProductionS3AccessKey, SecretAccessKey: config.LibraryProductionS3SecretKey,
|
||||
Secure: secure, ForcePathStyle: true, MaxObjectBytes: 256 * 1024 * 1024,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, nil, errors.New("生产对象存储配置无效")
|
||||
}
|
||||
identity := v071.ProductionScope{TenantID: config.LibraryProductionTenantID, KnowledgeBaseID: kb}
|
||||
target, err := v071.NewProductionLibraryTarget(ctx, v071.ProductionLibraryConfig{
|
||||
Postgres: config.LibraryProductionDatabaseURL, SearchURL: config.LibraryProductionElasticURL,
|
||||
SearchIndex: config.LibraryProductionIndex, Objects: objects, ObjectsReader: objects,
|
||||
ArchiveRoot: config.LibraryArchiveRoot,
|
||||
MaxDocuments: 200000,
|
||||
}, identity)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
// 契约以生产**当前**索引映射为准:这样预备份的兼容性与恢复目标是同一份事实。
|
||||
mapping, err := target.Search().MappingSHA256(ctx)
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, fmt.Errorf("读取生产索引映射失败: %w", err)
|
||||
}
|
||||
contract := librarysnapshot.Compatibility{
|
||||
WeKnoraVersion: "v0.7.1", SchemaVersion: 74, RetrieverEngine: "elasticsearch_v8",
|
||||
EmbeddingModel: config.LibraryEmbeddingModel, EmbeddingDimension: config.LibraryEmbeddingDimension,
|
||||
IndexMappingSHA256: mapping,
|
||||
}
|
||||
backups, err := 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,
|
||||
})
|
||||
if err != nil {
|
||||
target.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return target, backups, nil
|
||||
}
|
||||
+11
-4
@@ -22,7 +22,6 @@ import (
|
||||
"knowledge-governance/internal/libraryrelease"
|
||||
postgresstore "knowledge-governance/internal/platform/postgres"
|
||||
"knowledge-governance/internal/platform/weknora"
|
||||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||||
"knowledge-governance/internal/versionhistory"
|
||||
artifactfetch "knowledge-governance/internal/workflows/artifactfetch"
|
||||
catalogrefresh "knowledge-governance/internal/workflows/catalogrefresh"
|
||||
@@ -264,10 +263,18 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
|
||||
Grant: libraryrelease.Grant{Target: payload.Scope, SnapshotID: grant.SnapshotID,
|
||||
ParamsDigest: grant.ParamsDigest, ExpiresAt: grant.ExpiresAt, AuthorizedBy: grant.AuthorizedBy},
|
||||
}
|
||||
// 生产目标与预备份存储都来自适配层:未配置时整条链路在第一步就以明确缺项失败,
|
||||
// 生产目标与预备份存储来自部署注入的真实连接;未配置就在第一步失败并写明缺项,
|
||||
// 不会出现"写了一半没有回滚目标"的状态。
|
||||
target, _ := v071.NewProductionLibraryTarget(config.ProductionWeKnoraBaseURL, config.ProductionWeKnoraAPIKeyFile)
|
||||
outcome := libraryrelease.Run(ctx, request, target, v071.ArchiveBackupStore{}, recorder)
|
||||
target, backups, wireErr := buildLibraryProductionStack(ctx, config, payload.Scope)
|
||||
if wireErr != nil {
|
||||
message := wireErr.Error()
|
||||
recorder.Stage(ctx, "prebackup", "failed", message)
|
||||
_ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "prebackup", message,
|
||||
"未对生产执行任何写入,无需恢复", "", "", time.Now().UTC())
|
||||
return errors.New(message)
|
||||
}
|
||||
defer target.Close()
|
||||
outcome := libraryrelease.Run(ctx, request, target, backups, recorder)
|
||||
finishedAt := time.Now().UTC()
|
||||
switch outcome.State {
|
||||
case "succeeded":
|
||||
|
||||
@@ -21,17 +21,17 @@ A06(内容、问答、引用图片效果、性能报告)已被用户取消
|
||||
| V02 | 全部解析/索引/Wiki 就绪,验收绑定本版 | `internal/releaseadapter/v071/library_source_settlement.go`;页面准备清单 | 已实现·未验证 | 结清检查(settlement)在封存前执行且缺项阻断;但对"实时测试库此刻就绪"仍需一次真实封存复核;页面已注明依据的是哪次快照 |
|
||||
| V03 | 删除文档后 Wiki 共享正文/实体/边/引用真实变化 | `wikiDiffOf`(页面级差异) | 部分实现 | 真实归档验证:对内 vs 对外 = 新增 31 页 / 撤销 1 页,反向镜像一致。**页面级不等于**实体/概念/边/引用的语义重算,后者仍需 WeKnora 内部证据,未实现也不声称 |
|
||||
| V04 | 完整固定版本:文档集合+处理结果+资源+索引+完整 Wiki | `internal/librarysnapshot/*`(10 表 + 索引 + 资源 + Wiki 部件);`artifacts` 清单 | 已实现·已验证 | 真实快照 f992d64e…:4 文档/38 分块/122 索引/21 资源/31 Wiki 页/55,309,044 字节;部件摘要逐项校验 |
|
||||
| V05 | 导出中源变更则拒绝或隔离 | `internal/libraryfreeze/*`(持久 journal、源锁、暂停/恢复);`ProjectController` | 已实现·未验证 | 冻结窗口内会重新比对兼容契约;跨"项目外写入者"的证明仍缺,需一次带并发写入的隔离演练 |
|
||||
| V05 | 导出中源变更则拒绝或隔离 | `internal/libraryfreeze/*` + 预备份导出前后行数静默检查 | 部分实现·未验证 | 预备份导出前后行数不一致即失败(本机已验证会拒绝);真实并发写入的隔离演练仍未做 |
|
||||
| V06 | 只同步知识数据,不含账号/会话/凭据 | 导出白名单(10 张知识表 + 知识域索引 + 资源 + Wiki) | 已实现·已验证 | 部件清单里没有 users/sessions/api_keys;真实归档部件名可逐项核对 |
|
||||
| B01 | 每次生产写入前完整预备份,失败零写入 | `internal/libraryrelease`(prebackup 阶段)+ `library_versions.production_prebackup` + gate | 已实现·隔离测试通过·真实执行未授权 | 编排测试:预备份失败→零写入;读不到生产当前状态→零写入。真实预备份需生产读取凭据(未授权) |
|
||||
| B01 | 每次生产写入前完整预备份,失败零写入 | `v071.ProductionBackupStore.Prepare` + `libraryrelease` prebackup 阶段 | 已实现·本机真实服务已验证 | 本机真实 PG/ES 上生成预备份 3111e823…(626 字节),带导出前后行数静默检查;故障注入后用它恢复并校验通过。生产目标本身未授权,未对真实生产执行 |
|
||||
| B02 | 资源在测试侧删除后仍可恢复 | 归档内容寻址 blob(`library_versions` + `librarysnapshot` 内容去重) | 已实现·未验证 | 归档不依赖测试临时对象;跨版本去重 55,309,693 字节;尚未做"测试侧删除后从归档恢复"的实测 |
|
||||
| B03 | 每版保存完整 Wiki 与引用关联并校验 | `library_snapshot` 的 wiki 部件 + 摘要 | 已实现·已验证 | 真实快照含 31 页 Wiki 与逐页摘要,manifest 记录数量/字节 |
|
||||
| S01 | 数据库/资源/索引/Wiki 真实落地并校验 | `internal/libraryrelease.Run`(四类阶段 + verify)+ worker `library.sync` | 已实现·隔离测试通过·真实执行未授权 | 测试覆盖:四类按序写入、阶段失败→恢复、验证失败→恢复;生产目标适配器未配置时显式失败 |
|
||||
| S02 | 生产不再解析/Embedding/生成 Wiki | 编排只对 `Stage` 提交归档快照 id,不做解析/嵌入/生成 | 已实现·未验证 | 代码层面没有解析/嵌入调用;真实生产写入未执行,不能声称已验证 |
|
||||
| S01 | 数据库/资源/索引/Wiki 真实落地并校验 | `v071.ProductionLibraryTarget.Apply/Verify`(真实 SQL + ES bulk) | 已实现·本机真实服务已验证 | 本机真实 PG/ES:导入 1 文档/1 分块/1 Wiki 页/1 索引文档并逐表身份摘要校验;写入只覆盖知识域表 |
|
||||
| S02 | 生产不再解析/Embedding/生成 Wiki | 索引阶段直接搬运归档原始文档与向量 | 已实现·本机真实服务已验证 | 真实 ES 写入的是归档里的 embedding 数组(未调用任何模型);生产环境本身仍未写入 |
|
||||
| S03 | 一致生效与失败补偿边界 | — | 未实现 | 跨 PG/ES/资源的一致切点与可见性保护未实现 |
|
||||
| S04 | 各阶段失败/超时/重启可追溯可恢复 | `library_jobs`/`library_job_stages` + `ReconcileLibraryJobs` | 已实现·已验证 | 阶段与错误持久化;worker 启动清理"队列已无对应作业"的孤儿并写明恢复责任,避免永久占锁 |
|
||||
| S05 | 幂等与互斥(多 worker 安全) | `library_jobs_idempotency`、`library_jobs_active` 唯一索引 | 已实现·已验证 | 真实 PostgreSQL 上验证:同目标第二个在飞作业被拒;同键同 digest 重放复用;同键换 digest 冲突 |
|
||||
| R01 | 版本恢复整套生产知识状态 | `libraryrelease` 失败路径调用 `Target.Restore(backup)`;抽屉恢复入口仍禁用 | 已实现(失败自动恢复)·独立恢复未实现 | 恢复编排与幂等由测试覆盖;"按版本手动恢复"入口仍未接线,不能声称可用 |
|
||||
| R01 | 版本恢复整套生产知识状态 | `ProductionLibraryTarget.Restore` 按快照回放四类阶段 | 已实现·本机真实服务已验证(失败自动恢复) | 真实 PG/ES 上注入缺行故障后恢复并校验通过;"按版本手动恢复"入口仍未接线 |
|
||||
| R02 | 恢复失败显式异常、不回退测试库 | `Outcome.State=rollback_failed` + 独立阶段记录 | 已实现·隔离测试通过 | 测试断言恢复失败与同步失败可区分,且不触碰测试库 |
|
||||
| R03 | 存快/恢复快对比 | — | 未实现 | 未做测速;不写没有测量的数字 |
|
||||
| P01 | 作业阶段与整体状态 | `library_jobs.state/stage`;页面阶段条 | 已实现·已验证 | 阶段勾选只认当前作业证据;无作业时显示"本次作业尚未开始" |
|
||||
|
||||
@@ -72,6 +72,20 @@ type Config struct {
|
||||
// LibraryArchiveRoot 是整库封存产物的只读归档根(主机私有 0700 目录的只读挂载)。
|
||||
// 留空即"未接线":页面只显示原因,不显示空表冒充没有快照。
|
||||
LibraryArchiveRoot string
|
||||
// 生产整库目标(只由部署注入;缺任何一项都视为未配置,生产执行保持关闭)。
|
||||
LibraryProductionDatabaseURL string
|
||||
LibraryProductionElasticURL string
|
||||
LibraryProductionIndex string
|
||||
LibraryProductionTenantID uint64
|
||||
LibraryProductionInternalKB string
|
||||
LibraryProductionExternalKB string
|
||||
LibraryProductionS3Endpoint string
|
||||
LibraryProductionS3Region string
|
||||
LibraryProductionS3Bucket string
|
||||
LibraryProductionS3AccessKey string
|
||||
LibraryProductionS3SecretKey string
|
||||
LibraryEmbeddingModel string
|
||||
LibraryEmbeddingDimension int
|
||||
// LibraryCapacityBudgetBytes 是整库归档容量预算;0 表示未设置(容量门禁不放行)。
|
||||
LibraryCapacityBudgetBytes int64
|
||||
// LibraryRetentionVersions / Days 是保留策略;都为 0 表示未设置(不生成清理候选)。
|
||||
@@ -100,6 +114,16 @@ func Load(getenv func(string) string) (Config, error) {
|
||||
return Config{}, errors.New("LIBRARY_RETENTION_DAYS must be a positive integer")
|
||||
}
|
||||
retentionVersions, retentionDays := int(retentionVersions64), int(retentionDays64)
|
||||
tenantID64, err := positiveInt64OrDefault(strings.TrimSpace(getenv("LIBRARY_PRODUCTION_TENANT_ID")), 0)
|
||||
if err != nil {
|
||||
return Config{}, errors.New("LIBRARY_PRODUCTION_TENANT_ID must be a positive integer")
|
||||
}
|
||||
tenantID := uint64(tenantID64)
|
||||
embeddingDimension64, err := positiveInt64OrDefault(strings.TrimSpace(getenv("LIBRARY_EMBEDDING_DIMENSION")), 1024)
|
||||
if err != nil {
|
||||
return Config{}, errors.New("LIBRARY_EMBEDDING_DIMENSION must be a positive integer")
|
||||
}
|
||||
embeddingDimension := int(embeddingDimension64)
|
||||
config := Config{
|
||||
DatabaseURL: getenv("DATABASE_URL"),
|
||||
AppAddr: valueOrDefault(getenv("APP_ADDR"), ":8080"),
|
||||
@@ -143,6 +167,19 @@ func Load(getenv func(string) string) (Config, error) {
|
||||
VersionHistoryAllowedOrigins: splitOrigins(getenv("VERSION_HISTORY_ALLOWED_ORIGINS")),
|
||||
LibraryArchiveRoot: strings.TrimSpace(getenv("LIBRARY_ARCHIVE_ROOT")),
|
||||
LibraryAgentSocket: strings.TrimSpace(getenv("LIBRARY_AGENT_SOCKET")),
|
||||
LibraryProductionDatabaseURL: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_DATABASE_URL")),
|
||||
LibraryProductionElasticURL: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_ELASTIC_URL")),
|
||||
LibraryProductionIndex: valueOrDefault(strings.TrimSpace(getenv("LIBRARY_PRODUCTION_INDEX")), "weknora_full_chunks"),
|
||||
LibraryProductionTenantID: uint64(tenantID),
|
||||
LibraryProductionInternalKB: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_KB_INTERNAL")),
|
||||
LibraryProductionExternalKB: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_KB_EXTERNAL")),
|
||||
LibraryProductionS3Endpoint: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_ENDPOINT")),
|
||||
LibraryProductionS3Region: valueOrDefault(strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_REGION")), "us-east-1"),
|
||||
LibraryProductionS3Bucket: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_BUCKET")),
|
||||
LibraryProductionS3AccessKey: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_ACCESS_KEY")),
|
||||
LibraryProductionS3SecretKey: strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_SECRET_KEY")),
|
||||
LibraryEmbeddingModel: valueOrDefault(strings.TrimSpace(getenv("LIBRARY_EMBEDDING_MODEL")), "text-embedding-v4"),
|
||||
LibraryEmbeddingDimension: embeddingDimension,
|
||||
LibraryCapacityBudgetBytes: capacityBudget,
|
||||
LibraryRetentionVersions: retentionVersions,
|
||||
LibraryRetentionDays: retentionDays,
|
||||
|
||||
@@ -61,9 +61,12 @@ type Expectation struct {
|
||||
|
||||
// BackupRef 指向一份可独立恢复的完整预备份。
|
||||
type BackupRef struct {
|
||||
ID string `json:"id"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
ID string `json:"id"`
|
||||
// SnapshotID 是预备份在归档里的快照身份:恢复必须按它回放,
|
||||
// 不能只记"备份路径"却不知道恢复到哪一份内容。
|
||||
SnapshotID string `json:"snapshot_id"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
type Target interface {
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
package libraryretention
|
||||
|
||||
import "fmt"
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
)
|
||||
|
||||
// Capacity 是生产写入前的容量门禁:预算未设置就不放行(无法证明"装得下"),
|
||||
// 预算已知但不足也阻断。它不复用清理预览的回收量——清理未授权,不能拿它抵账。
|
||||
@@ -14,20 +17,37 @@ type Capacity struct {
|
||||
Reason string `json:"reason,omitempty"`
|
||||
}
|
||||
|
||||
// addCapacity 安全相加:int64 溢出时必须拒绝,而不是绕回负数让门禁通过。
|
||||
func addCapacity(current, required int64) (int64, bool) {
|
||||
if current < 0 || required < 0 {
|
||||
return 0, false
|
||||
}
|
||||
if current > math.MaxInt64-required {
|
||||
return 0, false
|
||||
}
|
||||
return current + required, true
|
||||
}
|
||||
|
||||
func CheckCapacity(currentBytes, requiredBytes, budgetBytes int64) Capacity {
|
||||
result := Capacity{BudgetBytes: budgetBytes, CurrentBytes: currentBytes, RequiredBytes: requiredBytes,
|
||||
ProjectedByte: currentBytes + requiredBytes}
|
||||
result := Capacity{BudgetBytes: budgetBytes, CurrentBytes: currentBytes, RequiredBytes: requiredBytes}
|
||||
if budgetBytes <= 0 {
|
||||
result.Reason = "容量预算未设置(LIBRARY_CAPACITY_BUDGET_BYTES),无法证明装得下"
|
||||
return result
|
||||
}
|
||||
result.Configured = true
|
||||
if requiredBytes < 0 || currentBytes < 0 {
|
||||
if currentBytes < 0 || requiredBytes < 0 {
|
||||
result.Reason = "容量口径异常(负数),拒绝放行"
|
||||
return result
|
||||
}
|
||||
if result.ProjectedByte > budgetBytes {
|
||||
result.Reason = fmt.Sprintf("写入后预计占用 %d 字节,超过预算 %d 字节", result.ProjectedByte, budgetBytes)
|
||||
projected, ok := addCapacity(currentBytes, requiredBytes)
|
||||
if !ok {
|
||||
// 溢出说明真实需求已经大得离谱:不能算出一个负数后"通过"。
|
||||
result.Reason = "容量相加溢出 int64,拒绝放行"
|
||||
return result
|
||||
}
|
||||
result.ProjectedByte = projected
|
||||
if projected > budgetBytes {
|
||||
result.Reason = fmt.Sprintf("写入后预计占用 %d 字节,超过预算 %d 字节", projected, budgetBytes)
|
||||
return result
|
||||
}
|
||||
result.OK = true
|
||||
|
||||
@@ -26,11 +26,29 @@ type Version struct {
|
||||
}
|
||||
|
||||
// Policy 是保留策略。零值表示"未设置"——这时不给任何候选。
|
||||
//
|
||||
// 组合语义(当两个维度都设置时)是**交集**,即"既超出最近 N 个、又早于 D 天"
|
||||
// 才成为候选;只要还满足其中一个维度就保留。这比"任一维度超限就删"保守:
|
||||
// 宁可少删,也不能因为天数短于版本节奏而误判一批刚封存的版本。
|
||||
type Policy struct {
|
||||
KeepVersions int `json:"keep_versions"`
|
||||
KeepDays int `json:"keep_days"`
|
||||
}
|
||||
|
||||
// Describe 用一句话写清当前组合语义,接口和页面直接展示,避免读者自行猜测。
|
||||
func (p Policy) Describe() string {
|
||||
switch {
|
||||
case p.KeepVersions > 0 && p.KeepDays > 0:
|
||||
return fmt.Sprintf("同时保留最近 %d 个版本与最近 %d 天,两者都不满足才成为候选", p.KeepVersions, p.KeepDays)
|
||||
case p.KeepVersions > 0:
|
||||
return fmt.Sprintf("保留最近 %d 个版本", p.KeepVersions)
|
||||
case p.KeepDays > 0:
|
||||
return fmt.Sprintf("保留最近 %d 天", p.KeepDays)
|
||||
default:
|
||||
return "未设置(全部保留,不生成候选)"
|
||||
}
|
||||
}
|
||||
|
||||
func (p Policy) Configured() bool { return p.KeepVersions > 0 || p.KeepDays > 0 }
|
||||
|
||||
type Candidate struct {
|
||||
@@ -43,6 +61,7 @@ type Candidate struct {
|
||||
|
||||
type Plan struct {
|
||||
Policy Policy `json:"policy"`
|
||||
PolicyNote string `json:"policy_note"`
|
||||
PolicyConfigured bool `json:"policy_configured"`
|
||||
AutomaticCleanup bool `json:"automatic_cleanup"`
|
||||
VersionCount int `json:"version_count"`
|
||||
@@ -60,6 +79,7 @@ type Plan struct {
|
||||
// Build 生成清理预览。now 由调用方给,便于测试;不给真实时钟依赖。
|
||||
func Build(versions []Version, totalBytes int64, policy Policy, now time.Time) Plan {
|
||||
plan := Plan{Policy: policy, PolicyConfigured: policy.Configured(), AutomaticCleanup: false,
|
||||
PolicyNote: policy.Describe(),
|
||||
VersionCount: len(versions), TotalBytes: totalBytes, Protected: []string{}, Keep: []string{},
|
||||
Candidates: []Candidate{}, Blocked: []string{}, Now: now.UTC()}
|
||||
|
||||
@@ -90,17 +110,29 @@ func Build(versions []Version, totalBytes int64, policy Policy, now time.Time) P
|
||||
plan.Keep = append(plan.Keep, version.ID)
|
||||
continue
|
||||
}
|
||||
reasons := make([]string, 0, 2)
|
||||
if policy.KeepVersions > 0 && rank >= policy.KeepVersions {
|
||||
reasons = append(reasons, fmt.Sprintf("超出保留 %d 个版本", policy.KeepVersions))
|
||||
// 组合语义:对**每个已设置**的维度都超限才成为候选(未设置的维度不约束)。
|
||||
// 只设版本数 ⇒ 看数量;只设天数 ⇒ 看天数;两者都设 ⇒ 必须同时超限。
|
||||
beyondCount := policy.KeepVersions > 0 && rank >= policy.KeepVersions
|
||||
beyondDays := policy.KeepDays > 0 && !version.CreatedAt.IsZero() &&
|
||||
version.CreatedAt.Before(now.AddDate(0, 0, -policy.KeepDays))
|
||||
exceeds := true
|
||||
if policy.KeepVersions > 0 {
|
||||
exceeds = exceeds && beyondCount
|
||||
}
|
||||
if policy.KeepDays > 0 && !version.CreatedAt.IsZero() && version.CreatedAt.Before(now.AddDate(0, 0, -policy.KeepDays)) {
|
||||
reasons = append(reasons, fmt.Sprintf("超出保留 %d 天", policy.KeepDays))
|
||||
if policy.KeepDays > 0 {
|
||||
exceeds = exceeds && beyondDays
|
||||
}
|
||||
if len(reasons) == 0 {
|
||||
if !exceeds {
|
||||
plan.Keep = append(plan.Keep, version.ID)
|
||||
continue
|
||||
}
|
||||
reasons := make([]string, 0, 2)
|
||||
if beyondCount {
|
||||
reasons = append(reasons, fmt.Sprintf("超出保留 %d 个版本", policy.KeepVersions))
|
||||
}
|
||||
if beyondDays {
|
||||
reasons = append(reasons, fmt.Sprintf("超出保留 %d 天", policy.KeepDays))
|
||||
}
|
||||
plan.Candidates = append(plan.Candidates, Candidate{ID: version.ID, Kind: version.Kind,
|
||||
Reason: reasons, ExclusiveBytes: version.ExclusiveBytes})
|
||||
plan.ReclaimPreviewBytes += version.ExclusiveBytes
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package libraryretention
|
||||
|
||||
import (
|
||||
"math"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
@@ -93,3 +95,66 @@ func TestCheckCapacityFailsClosed(t *testing.T) {
|
||||
t.Fatalf("within budget must pass: %+v", result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckCapacityRejectsOverflowInsteadOfWrapping(t *testing.T) {
|
||||
// 两个大正数相加会绕回负数;门禁必须拒绝,而不是把负数当成"没超预算"。
|
||||
result := CheckCapacity(math.MaxInt64-10, 100, math.MaxInt64)
|
||||
if result.OK {
|
||||
t.Fatalf("overflow must be refused, got %+v", result)
|
||||
}
|
||||
if result.Reason == "" || !strings.Contains(result.Reason, "溢出") {
|
||||
t.Fatalf("overflow must name the reason, got %q", result.Reason)
|
||||
}
|
||||
// 恰好等于预算:允许(不超),并且不溢出。
|
||||
exact := CheckCapacity(100, 50, 150)
|
||||
if !exact.OK || exact.ProjectedByte != 150 {
|
||||
t.Fatalf("exactly at budget must pass: %+v", exact)
|
||||
}
|
||||
// 相差 1 字节超预算:阻断。
|
||||
if over := CheckCapacity(100, 51, 150); over.OK {
|
||||
t.Fatalf("one byte over budget must be refused: %+v", over)
|
||||
}
|
||||
// 负数输入不能算出"更小"的占用。
|
||||
if neg := CheckCapacity(-1, 10, 1<<30); neg.OK {
|
||||
t.Fatalf("negative current must be refused: %+v", neg)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPolicyCombinationUsesConjunctionOfSetDimensions(t *testing.T) {
|
||||
now := time.Date(2026, 10, 20, 0, 0, 0, 0, time.UTC)
|
||||
versions := []Version{
|
||||
{ID: "new", CreatedAt: now.AddDate(0, 0, -1)},
|
||||
{ID: "old", CreatedAt: now.AddDate(0, 0, -30)},
|
||||
}
|
||||
// 同时设 3 个版本 + 7 天:old 两个维度都超限 → 候选;new 太新 → 保留。
|
||||
both := Build(versions, 0, Policy{KeepVersions: 1, KeepDays: 7}, now)
|
||||
if len(both.Candidates) != 1 || both.Candidates[0].ID != "old" {
|
||||
t.Fatalf("only the doubly-exceeded version may be a candidate: %+v", both.Candidates)
|
||||
}
|
||||
// 只设天数:数量维度不约束,old 超 7 天即为候选。
|
||||
daysOnly := Build(versions, 0, Policy{KeepDays: 7}, now)
|
||||
if len(daysOnly.Candidates) != 1 || daysOnly.Candidates[0].ID != "old" {
|
||||
t.Fatalf("days-only policy must use the day dimension: %+v", daysOnly.Candidates)
|
||||
}
|
||||
// 数量宽松(保留 5 个)但天数很紧(1 天):同时超限才删 → 都不候选。
|
||||
wide := Build(versions, 0, Policy{KeepVersions: 5, KeepDays: 1}, now)
|
||||
if len(wide.Candidates) != 0 {
|
||||
t.Fatalf("a version kept by either dimension must not be a candidate: %+v", wide.Candidates)
|
||||
}
|
||||
if both.PolicyNote == "" {
|
||||
t.Fatal("policy note must describe the combination explicitly")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLockedVersionIsNeverACandidate(t *testing.T) {
|
||||
now := time.Date(2026, 10, 20, 0, 0, 0, 0, time.UTC)
|
||||
locked := Version{ID: "locked", CreatedAt: now.AddDate(0, 0, -90), Protected: true,
|
||||
ProtectedBy: []string{"人工锁定", "备份恢复依赖"}}
|
||||
plan := Build([]Version{locked}, 0, Policy{KeepVersions: 1, KeepDays: 1}, now)
|
||||
if len(plan.Candidates) != 0 {
|
||||
t.Fatalf("locked version must never be a candidate: %+v", plan.Candidates)
|
||||
}
|
||||
if len(plan.Protected) != 1 || len(plan.Blocked) != 0 {
|
||||
t.Fatalf("plan must record protection: %+v", plan)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
package librarysnapshot
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"sort"
|
||||
)
|
||||
|
||||
// IndexDocument 是归档索引部件里的一条原始文档(含向量)。
|
||||
// 生产同步直接搬运它,不重新解析、不重新 Embedding。
|
||||
type IndexDocument struct {
|
||||
ID string `json:"id"`
|
||||
Source json.RawMessage `json:"source"`
|
||||
}
|
||||
|
||||
// ReadRowIDs 读出行数组里每行的 id 并排序,用于"身份集合"比对。
|
||||
func ReadRowIDs(r io.Reader) ([]string, error) {
|
||||
decoder := json.NewDecoder(r)
|
||||
token, err := decoder.Token()
|
||||
if err != nil || token != json.Delim('[') {
|
||||
return nil, errors.New("snapshot component must be a JSON row array")
|
||||
}
|
||||
ids := make([]string, 0, 64)
|
||||
for decoder.More() {
|
||||
var row struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if err := decoder.Decode(&row); err != nil {
|
||||
return nil, errors.New("invalid archived row")
|
||||
}
|
||||
if row.ID == "" {
|
||||
return nil, errors.New("archived row without id")
|
||||
}
|
||||
ids = append(ids, row.ID)
|
||||
}
|
||||
sort.Strings(ids)
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
// ReadIndexDocuments 读取索引部件。归档部件有两种历史形状:直接是文档数组,
|
||||
// 或 [{id, source}] 包装;两种都必须支持,否则旧归档无法同步。
|
||||
func ReadIndexDocuments(r io.Reader) ([]IndexDocument, error) {
|
||||
decoder := json.NewDecoder(r)
|
||||
token, err := decoder.Token()
|
||||
if err != nil || token != json.Delim('[') {
|
||||
return nil, errors.New("index component must be a JSON array")
|
||||
}
|
||||
documents := make([]IndexDocument, 0, 64)
|
||||
for decoder.More() {
|
||||
var row map[string]json.RawMessage
|
||||
if err := decoder.Decode(&row); err != nil {
|
||||
return nil, errors.New("invalid archived index row")
|
||||
}
|
||||
// 历史归档有两种键风格:导出结构体的 "ID"/"Source",以及小写 "id"/"source"。
|
||||
wrapped, hasSource := row["source"]
|
||||
if !hasSource {
|
||||
wrapped, hasSource = row["Source"]
|
||||
}
|
||||
if hasSource {
|
||||
idRaw, ok := row["id"]
|
||||
if !ok {
|
||||
idRaw = row["ID"]
|
||||
}
|
||||
var id string
|
||||
_ = json.Unmarshal(idRaw, &id)
|
||||
documents = append(documents, IndexDocument{ID: id, Source: wrapped})
|
||||
continue
|
||||
}
|
||||
id := documentIdentity(row)
|
||||
if id == "" {
|
||||
return nil, errors.New("archived index row has no stable id")
|
||||
}
|
||||
raw, err := json.Marshal(row)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
documents = append(documents, IndexDocument{ID: id, Source: raw})
|
||||
}
|
||||
if len(documents) == 0 {
|
||||
return documents, nil
|
||||
}
|
||||
for _, document := range documents {
|
||||
if document.ID == "" || len(document.Source) == 0 {
|
||||
return nil, errors.New("archived index document incomplete")
|
||||
}
|
||||
}
|
||||
sort.Slice(documents, func(i, j int) bool { return documents[i].ID < documents[j].ID })
|
||||
return documents, nil
|
||||
}
|
||||
|
||||
func documentIdentity(row map[string]json.RawMessage) string {
|
||||
for _, key := range []string{"chunk_id", "source_id", "id"} {
|
||||
if raw, ok := row[key]; ok {
|
||||
var value string
|
||||
if json.Unmarshal(raw, &value) == nil && value != "" {
|
||||
return value
|
||||
}
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"knowledge-governance/internal/libraryrelease"
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
)
|
||||
|
||||
// ProductionBackupStore 在生产写入前创建**独立完整预备份**:从生产读一次整库
|
||||
// (PG RepeatableRead + ES 导出一致快照 + 资源对象),密封成归档里的一个版本。
|
||||
//
|
||||
// 为什么要有静默检查:预备份期间生产若仍在变,导出的 PG/ES/对象就不是同一个切点,
|
||||
// 恢复它等于恢复到"没人见过的状态"。所以导出前后各读一次行数,不一致就失败,
|
||||
// 让作业重试,而不是拿一个不一致的备份当恢复点。
|
||||
type ProductionBackupStore struct {
|
||||
pool *pgxpool.Pool
|
||||
search *SearchExporter
|
||||
objects ObjectReader
|
||||
archiveRoot string
|
||||
scope string
|
||||
identity PostgresIdentity
|
||||
compatibility librarysnapshot.Compatibility
|
||||
maxDocuments int
|
||||
maxObjectByte int64
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
type ProductionBackupConfig struct {
|
||||
Pool *pgxpool.Pool
|
||||
Search *SearchExporter
|
||||
Objects ObjectReader
|
||||
ArchiveRoot string
|
||||
Scope string
|
||||
Identity PostgresIdentity
|
||||
Compatibility librarysnapshot.Compatibility
|
||||
MaxDocuments int
|
||||
MaxObjectByte int64
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
func NewProductionBackupStore(config ProductionBackupConfig) (*ProductionBackupStore, error) {
|
||||
if config.Pool == nil || config.Search == nil || config.Objects == nil || config.ArchiveRoot == "" ||
|
||||
(config.Scope != "internal" && config.Scope != "external") || config.Identity.TenantID == 0 ||
|
||||
!validSearchIdentifier(config.Identity.KnowledgeBaseID) || config.MaxDocuments < 1 {
|
||||
return nil, errors.New("生产预备份配置不完整")
|
||||
}
|
||||
if config.MaxObjectByte <= 0 {
|
||||
config.MaxObjectByte = 256 * 1024 * 1024
|
||||
}
|
||||
now := config.Now
|
||||
if now == nil {
|
||||
now = time.Now
|
||||
}
|
||||
return &ProductionBackupStore{pool: config.Pool, search: config.Search, objects: config.Objects,
|
||||
archiveRoot: config.ArchiveRoot, scope: config.Scope, identity: config.Identity,
|
||||
compatibility: config.Compatibility, maxDocuments: config.MaxDocuments,
|
||||
maxObjectByte: config.MaxObjectByte, now: now}, nil
|
||||
}
|
||||
|
||||
func (s *ProductionBackupStore) Prepare(ctx context.Context, _ string, current libraryrelease.CurrentState) (libraryrelease.BackupRef, error) {
|
||||
archive, err := librarysnapshot.Open(s.archiveRoot, 256*1024*1024)
|
||||
if err != nil {
|
||||
return libraryrelease.BackupRef{}, errors.New("归档不可用,无法创建预备份")
|
||||
}
|
||||
defer archive.Close()
|
||||
createdAt := s.now().UTC()
|
||||
pg, err := ExportLibrary(ctx, s.pool, s.identity)
|
||||
if err != nil {
|
||||
return libraryrelease.BackupRef{}, fmt.Errorf("读取生产整库失败: %w", err)
|
||||
}
|
||||
// 整库导出用 ExportLibrary(按租户+知识库,不带单文档身份),
|
||||
// 与封存路径复用同一份实现,避免两套范围规则。
|
||||
es, err := s.search.ExportLibrary(ctx, SearchIdentity{TenantID: s.identity.TenantID,
|
||||
KnowledgeBaseID: s.identity.KnowledgeBaseID}, s.maxDocuments)
|
||||
if err != nil {
|
||||
return libraryrelease.BackupRef{}, fmt.Errorf("读取生产索引失败: %w", err)
|
||||
}
|
||||
if es.MappingSHA256 != s.compatibility.IndexMappingSHA256 {
|
||||
// 契约不符说明生产换了索引映射:此时备份出来的东西无法用同一套恢复,
|
||||
// 宁可失败也不产出"看起来完整"的预备份。
|
||||
return libraryrelease.BackupRef{}, errors.New("生产索引映射与归档契约不一致")
|
||||
}
|
||||
// ArchiveLibraryCandidate 自己按 resources 行读取对象字节并计算摘要,
|
||||
// 所以这里直接给它生产对象读取器,不再重复导出一次资源。
|
||||
snapshotID, err := ArchiveLibraryCandidate(ctx, archive, s.scope, s.identity, s.compatibility, pg, es, s.objects, createdAt)
|
||||
if err != nil {
|
||||
return libraryrelease.BackupRef{}, fmt.Errorf("密封预备份失败: %w", err)
|
||||
}
|
||||
// 静默检查:导出期间生产不能有变化,否则这份预备份不是一致切点。
|
||||
after, err := s.rowCounts(ctx)
|
||||
if err != nil {
|
||||
return libraryrelease.BackupRef{}, err
|
||||
}
|
||||
for table, want := range current.Counts {
|
||||
if table == "index" {
|
||||
continue
|
||||
}
|
||||
if after[table] != want {
|
||||
return libraryrelease.BackupRef{}, fmt.Errorf("预备份期间生产 %s 行数从 %d 变为 %d,切点不一致", table, want, after[table])
|
||||
}
|
||||
}
|
||||
manifest, err := archive.Load(snapshotID)
|
||||
if err != nil {
|
||||
return libraryrelease.BackupRef{}, errors.New("预备份封存后无法校验")
|
||||
}
|
||||
var total int64
|
||||
for _, part := range manifest.Parts {
|
||||
total += part.Bytes
|
||||
}
|
||||
return libraryrelease.BackupRef{ID: "backup-" + snapshotID, SnapshotID: snapshotID, Bytes: total,
|
||||
CreatedAt: createdAt}, nil
|
||||
}
|
||||
|
||||
func (s *ProductionBackupStore) rowCounts(ctx context.Context) (map[string]int64, error) {
|
||||
counts := map[string]int64{}
|
||||
for _, table := range productionLibraryTables {
|
||||
var count int64
|
||||
if err := s.pool.QueryRow(ctx, fmt.Sprintf(`SELECT count(*) FROM %s WHERE %s`,
|
||||
table, scopeWhere(table)), s.identity.TenantID, s.identity.KnowledgeBaseID).Scan(&count); err != nil {
|
||||
return nil, fmt.Errorf("读取生产 %s 行数失败", table)
|
||||
}
|
||||
counts[table] = count
|
||||
}
|
||||
return counts, nil
|
||||
}
|
||||
@@ -1,52 +1,512 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"knowledge-governance/internal/libraryrelease"
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
)
|
||||
|
||||
// ErrProductionTargetUnconfigured 表示生产读/写适配器还没接线。
|
||||
// 这是一个**显式缺项**,不是"暂时跳过":没有它就无法读生产当前状态、
|
||||
// 无法建预备份、也无法验证或恢复,所以整条生产链路必须拒绝执行。
|
||||
var ErrProductionTargetUnconfigured = errors.New("生产目标适配器未配置(需要生产端点、只读凭据与写入适配)")
|
||||
|
||||
// ProductionLibraryTarget 是整库同步的生产目标实现占位。
|
||||
// 生产整库目标:真实读、真实写、真实校验、真实恢复。
|
||||
//
|
||||
// 为什么先不做"半套"实现:读生产当前状态、写四类数据、验证与恢复必须同时成立,
|
||||
// 只实现其中一半会让链路看起来可用却无法回滚。接口在此固定,接线时只补实现。
|
||||
type ProductionLibraryTarget struct {
|
||||
BaseURL string
|
||||
APIKeyFile string
|
||||
// 只覆盖知识域(文档/分块/标签/资源绑定/Wiki/资源对象/检索索引),不写 users、
|
||||
// sessions、api_keys 这类账号与会话表;索引直接搬运归档里的原始文档与向量,
|
||||
// 不重新解析、不重新 Embedding。连接全部来自注入配置,代码里不存生产凭据。
|
||||
|
||||
var ErrProductionTargetUnconfigured = errors.New("生产整库目标未配置(需要生产数据库、索引、对象存储与知识库映射)")
|
||||
|
||||
// ProductionScope 是一个目标库在生产侧的身份。
|
||||
type ProductionScope struct {
|
||||
TenantID uint64
|
||||
KnowledgeBaseID string
|
||||
}
|
||||
|
||||
func NewProductionLibraryTarget(baseURL, apiKeyFile string) (*ProductionLibraryTarget, error) {
|
||||
if baseURL == "" || apiKeyFile == "" {
|
||||
return nil, ErrProductionTargetUnconfigured
|
||||
type ProductionLibraryConfig struct {
|
||||
// Postgres 是生产 WeKnora 的数据库连接串(部署注入,不落代码/日志)。
|
||||
Postgres string
|
||||
// SearchURL 是生产 Elasticsearch 基址;SearchIndex 是知识域索引名。
|
||||
SearchURL string
|
||||
SearchIndex string
|
||||
SearchToken string
|
||||
// Objects 负责写入资源对象(既有 ObjectStore 接口);ObjectsReader 负责
|
||||
// 备份时读回对象字节。同一个 S3 存储通常两者都实现。
|
||||
Objects ObjectStore
|
||||
ObjectsReader ObjectReader
|
||||
ArchiveRoot string
|
||||
MaxDocuments int
|
||||
HTTPClient *http.Client
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
func (config ProductionLibraryConfig) validate(scope ProductionScope) error {
|
||||
if config.Postgres == "" || config.SearchURL == "" || config.SearchIndex == "" ||
|
||||
config.Objects == nil || config.ObjectsReader == nil || config.ArchiveRoot == "" || config.MaxDocuments < 1 {
|
||||
return ErrProductionTargetUnconfigured
|
||||
}
|
||||
return &ProductionLibraryTarget{BaseURL: baseURL, APIKeyFile: apiKeyFile}, nil
|
||||
if scope.TenantID == 0 || !validSearchIdentifier(scope.KnowledgeBaseID) {
|
||||
return fmt.Errorf("%w:缺少目标知识库身份", ErrProductionTargetUnconfigured)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) Current(context.Context) (libraryrelease.CurrentState, error) {
|
||||
return libraryrelease.CurrentState{}, ErrProductionTargetUnconfigured
|
||||
// 依赖顺序:插入按正序,删除按倒序。
|
||||
var productionLibraryTables = []string{
|
||||
"knowledges", "chunks", "knowledge_tags", "knowledge_tag_relations",
|
||||
"resource_bindings", "resources", "wiki_folders", "wiki_pages",
|
||||
"wiki_page_issues", "wiki_log_entries",
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) Apply(context.Context, libraryrelease.Stage, string) error {
|
||||
return ErrProductionTargetUnconfigured
|
||||
type ProductionLibraryTarget struct {
|
||||
pool *pgxpool.Pool
|
||||
config ProductionLibraryConfig
|
||||
scope ProductionScope
|
||||
search *SearchExporter
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) Verify(context.Context, libraryrelease.Expectation) error {
|
||||
return ErrProductionTargetUnconfigured
|
||||
func NewProductionLibraryTarget(ctx context.Context, config ProductionLibraryConfig, scope ProductionScope) (*ProductionLibraryTarget, error) {
|
||||
if err := config.validate(scope); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pool, err := pgxpool.New(ctx, config.Postgres)
|
||||
if err != nil {
|
||||
// 连接串本身不打印,错误里也只说明连不上。
|
||||
return nil, errors.New("生产数据库连接失败")
|
||||
}
|
||||
client := config.HTTPClient
|
||||
if client == nil {
|
||||
client = &http.Client{Timeout: 2 * time.Minute}
|
||||
}
|
||||
search, err := NewSearchExporter(SearchExporterConfig{BaseURL: config.SearchURL, Index: config.SearchIndex,
|
||||
AuthorizationToken: config.SearchToken, PageSize: 500, HTTPClient: client})
|
||||
if err != nil {
|
||||
pool.Close()
|
||||
return nil, err
|
||||
}
|
||||
now := config.Now
|
||||
if now == nil {
|
||||
now = time.Now
|
||||
}
|
||||
return &ProductionLibraryTarget{pool: pool, config: config, scope: scope, search: search, now: now}, nil
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) Restore(context.Context, libraryrelease.BackupRef) error {
|
||||
return ErrProductionTargetUnconfigured
|
||||
func (t *ProductionLibraryTarget) Close() { t.pool.Close() }
|
||||
|
||||
// Pool/Search/Identity 暴露给预备份存储复用同一套生产连接,避免重复开连接池。
|
||||
func (t *ProductionLibraryTarget) Pool() *pgxpool.Pool { return t.pool }
|
||||
|
||||
func (t *ProductionLibraryTarget) Search() *SearchExporter { return t.search }
|
||||
|
||||
func (t *ProductionLibraryTarget) Objects() ObjectReader { return t.config.ObjectsReader }
|
||||
|
||||
func (t *ProductionLibraryTarget) Identity() PostgresIdentity {
|
||||
return PostgresIdentity{TenantID: t.scope.TenantID, KnowledgeBaseID: t.scope.KnowledgeBaseID}
|
||||
}
|
||||
|
||||
// ArchiveBackupStore 也还没接线:预备份必须读生产当前整库,同样的前置条件。
|
||||
type ArchiveBackupStore struct{}
|
||||
|
||||
func (ArchiveBackupStore) Prepare(context.Context, string, libraryrelease.CurrentState) (libraryrelease.BackupRef, error) {
|
||||
return libraryrelease.BackupRef{}, errors.New("生产预备份存储未配置(需要生产读取凭据与可独立恢复的备份目标)")
|
||||
// Current 读生产当前整库指纹:每张表的 (行数, 身份集合摘要) + 索引条数。
|
||||
func (t *ProductionLibraryTarget) Current(ctx context.Context) (libraryrelease.CurrentState, error) {
|
||||
state := libraryrelease.CurrentState{Counts: map[string]int64{}, Digests: map[string]string{},
|
||||
CapturedAt: t.now().UTC()}
|
||||
for _, table := range productionLibraryTables {
|
||||
count, digest, err := t.tableState(ctx, table)
|
||||
if err != nil {
|
||||
return libraryrelease.CurrentState{}, err
|
||||
}
|
||||
state.Counts[table] = count
|
||||
state.Digests[table] = digest
|
||||
}
|
||||
indexCount, err := t.search.Count(ctx, SearchIdentity{TenantID: t.scope.TenantID,
|
||||
KnowledgeBaseID: t.scope.KnowledgeBaseID})
|
||||
if err != nil {
|
||||
return libraryrelease.CurrentState{}, err
|
||||
}
|
||||
state.Counts["index"] = indexCount
|
||||
return state, nil
|
||||
}
|
||||
|
||||
// scopeWhere 返回"这张表里属于本知识库的行"的过滤条件($1=tenant,$2=kb)。
|
||||
// 不同表的归属列不同:关系表与绑定表没有 knowledge_base_id,必须走父表判定,
|
||||
// 否则要么删错、要么把别的知识库数据算进来。
|
||||
func scopeWhere(table string) string {
|
||||
switch table {
|
||||
case "knowledge_tag_relations":
|
||||
return `tenant_id=$1 AND knowledge_id IN (SELECT id FROM knowledges k
|
||||
WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2)`
|
||||
case "resource_bindings":
|
||||
return `tenant_id=$1 AND ((owner_type='knowledge' AND owner_id IN (SELECT id FROM knowledges k
|
||||
WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2)) OR (owner_type='chunk' AND owner_id IN
|
||||
(SELECT id FROM chunks c WHERE c.tenant_id=$1 AND c.knowledge_base_id=$2)))`
|
||||
case "resources":
|
||||
return `EXISTS (SELECT 1 FROM resource_bindings b WHERE b.owner_id IN
|
||||
(SELECT id FROM knowledges k WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2))`
|
||||
default:
|
||||
return "tenant_id=$1 AND knowledge_base_id=$2"
|
||||
}
|
||||
}
|
||||
|
||||
// identityExpr 是"稳定身份"列:多数表就是 id,关系表没有 id 就用复合键。
|
||||
func identityExpr(table string) string {
|
||||
if table == "knowledge_tag_relations" {
|
||||
return "knowledge_id||':'||tag_id"
|
||||
}
|
||||
return "id"
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) tableState(ctx context.Context, table string) (int64, string, error) {
|
||||
rows, err := t.pool.Query(ctx, fmt.Sprintf(`SELECT %s FROM %s WHERE %s ORDER BY 1`,
|
||||
identityExpr(table), table, scopeWhere(table)), t.scope.TenantID, t.scope.KnowledgeBaseID)
|
||||
if err != nil {
|
||||
return 0, "", fmt.Errorf("读取生产 %s 失败", table)
|
||||
}
|
||||
defer rows.Close()
|
||||
ids := make([]string, 0, 64)
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return 0, "", err
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, "", err
|
||||
}
|
||||
return int64(len(ids)), digestIDs(ids), nil
|
||||
}
|
||||
|
||||
func digestIDs(ids []string) string {
|
||||
sum := sha256.Sum256([]byte(strings.Join(ids, "\n")))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// Apply 把一类数据从归档写进生产;database 与 wiki 走同一个知识域事务。
|
||||
func (t *ProductionLibraryTarget) Apply(ctx context.Context, stage libraryrelease.Stage, snapshotID string) error {
|
||||
archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
|
||||
if err != nil {
|
||||
return errors.New("归档不可用")
|
||||
}
|
||||
defer archive.Close()
|
||||
manifest, err := archive.Load(snapshotID)
|
||||
if err != nil {
|
||||
return errors.New("预发布快照不可读")
|
||||
}
|
||||
switch stage {
|
||||
case libraryrelease.StageDatabase, libraryrelease.StageWiki:
|
||||
kind := "database"
|
||||
if stage == libraryrelease.StageWiki {
|
||||
kind = "wiki"
|
||||
}
|
||||
return t.importRows(ctx, archive, manifest, kind)
|
||||
case libraryrelease.StageResources:
|
||||
return t.importResources(ctx, archive, manifest)
|
||||
case libraryrelease.StageIndex:
|
||||
return t.importIndex(ctx, archive, manifest)
|
||||
default:
|
||||
return fmt.Errorf("未知阶段 %q", stage)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) importRows(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest, kind string) error {
|
||||
parts := map[string]librarysnapshot.Part{}
|
||||
for _, part := range manifest.Parts {
|
||||
if part.Kind == kind {
|
||||
parts[part.Name] = part
|
||||
}
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return fmt.Errorf("快照缺少 %s 部件", kind)
|
||||
}
|
||||
tx, err := t.pool.BeginTx(ctx, pgx.TxOptions{})
|
||||
if err != nil {
|
||||
return errors.New("开启生产事务失败")
|
||||
}
|
||||
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
||||
for index := len(productionLibraryTables) - 1; index >= 0; index-- {
|
||||
table := productionLibraryTables[index]
|
||||
if _, ok := parts[table]; !ok {
|
||||
continue
|
||||
}
|
||||
if err := t.deleteScope(ctx, tx, table); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for _, table := range productionLibraryTables {
|
||||
part, ok := parts[table]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if err := t.insertRows(ctx, tx, archive, part, table); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return errors.New("提交生产写入失败")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) deleteScope(ctx context.Context, tx pgx.Tx, table string) error {
|
||||
if table == "resources" {
|
||||
// 资源可能被多个知识库共享:这里不删,只做插入/更新。
|
||||
return nil
|
||||
}
|
||||
statement := fmt.Sprintf(`DELETE FROM %s WHERE %s`, table, scopeWhere(table))
|
||||
if _, err := tx.Exec(ctx, statement, t.scope.TenantID, t.scope.KnowledgeBaseID); err != nil {
|
||||
return fmt.Errorf("清理生产 %s 失败", table)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) insertRows(ctx context.Context, tx pgx.Tx, archive *librarysnapshot.Archive, part librarysnapshot.Part, table string) error {
|
||||
reader, err := archive.OpenPart(part)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取归档 %s 失败", part.Name)
|
||||
}
|
||||
defer reader.Close()
|
||||
decoder := json.NewDecoder(reader)
|
||||
token, err := decoder.Token()
|
||||
if err != nil || token != json.Delim('[') {
|
||||
return fmt.Errorf("归档 %s 不是行数组", part.Name)
|
||||
}
|
||||
var columns []string
|
||||
statement := ""
|
||||
for decoder.More() {
|
||||
var row map[string]json.RawMessage
|
||||
if err := decoder.Decode(&row); err != nil {
|
||||
return fmt.Errorf("解析归档 %s 行失败", part.Name)
|
||||
}
|
||||
if len(row) == 0 {
|
||||
continue
|
||||
}
|
||||
if columns == nil {
|
||||
columns = make([]string, 0, len(row))
|
||||
for key := range row {
|
||||
columns = append(columns, key)
|
||||
}
|
||||
sort.Strings(columns)
|
||||
// 本库行先删后插:除 resources(可能被多个知识库共享)外不需要 upsert。
|
||||
if table == "resources" {
|
||||
statement = fmt.Sprintf(`INSERT INTO %s SELECT * FROM jsonb_populate_record(NULL::%s, $1::jsonb)
|
||||
ON CONFLICT (id) DO UPDATE SET %s`, table, table, updateAssignments(columns))
|
||||
} else {
|
||||
statement = fmt.Sprintf(`INSERT INTO %s SELECT * FROM jsonb_populate_record(NULL::%s, $1::jsonb)`,
|
||||
table, table)
|
||||
}
|
||||
}
|
||||
encoded, err := json.Marshal(row)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.Exec(ctx, statement, encoded); err != nil {
|
||||
return fmt.Errorf("写入生产 %s 失败: %w", table, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func updateAssignments(columns []string) string {
|
||||
assignments := make([]string, 0, len(columns))
|
||||
for _, column := range columns {
|
||||
if column == "id" {
|
||||
continue
|
||||
}
|
||||
assignments = append(assignments, fmt.Sprintf("%s=EXCLUDED.%s", column, column))
|
||||
}
|
||||
if len(assignments) == 0 {
|
||||
return "id=EXCLUDED.id"
|
||||
}
|
||||
return strings.Join(assignments, ", ")
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
|
||||
paths, err := t.resourcePaths(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, part := range manifest.Parts {
|
||||
if part.Kind != "resource" {
|
||||
continue
|
||||
}
|
||||
path, ok := paths[part.Name]
|
||||
if !ok || path == "" {
|
||||
return fmt.Errorf("生产缺少资源 %s 的存储路径", part.Name)
|
||||
}
|
||||
reader, err := archive.OpenPart(part)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
body, err := io.ReadAll(io.LimitReader(reader, part.Bytes+1))
|
||||
reader.Close()
|
||||
if err != nil || int64(len(body)) != part.Bytes {
|
||||
return fmt.Errorf("资源 %s 字节与归档摘要不符", part.Name)
|
||||
}
|
||||
if err := t.config.Objects.Put(ctx, "s3", path, bytes.NewReader(body), int64(len(body))); err != nil {
|
||||
return fmt.Errorf("上传资源 %s 失败", part.Name)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) resourcePaths(ctx context.Context) (map[string]string, error) {
|
||||
rows, err := t.pool.Query(ctx, `
|
||||
SELECT handle, physical_path FROM resources
|
||||
WHERE EXISTS (SELECT 1 FROM resource_bindings b WHERE b.owner_id IN
|
||||
(SELECT id FROM knowledges k WHERE k.tenant_id=$1 AND k.knowledge_base_id=$2))`,
|
||||
t.scope.TenantID, t.scope.KnowledgeBaseID)
|
||||
if err != nil {
|
||||
return nil, errors.New("读取生产资源路径失败")
|
||||
}
|
||||
defer rows.Close()
|
||||
paths := map[string]string{}
|
||||
for rows.Next() {
|
||||
var handle, path string
|
||||
if err := rows.Scan(&handle, &path); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
paths[handle] = path
|
||||
}
|
||||
return paths, rows.Err()
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) importIndex(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
|
||||
for _, part := range manifest.Parts {
|
||||
if part.Kind != "index" {
|
||||
continue
|
||||
}
|
||||
if part.Count > int64(t.config.MaxDocuments) {
|
||||
return fmt.Errorf("索引条数 %d 超过上限 %d", part.Count, t.config.MaxDocuments)
|
||||
}
|
||||
reader, err := archive.OpenPart(part)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
documents, err := librarysnapshot.ReadIndexDocuments(reader)
|
||||
reader.Close()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// 直接搬运原始文档与向量:不解析、不重新 Embedding。
|
||||
identity := SearchIdentity{TenantID: t.scope.TenantID, KnowledgeBaseID: t.scope.KnowledgeBaseID}
|
||||
if err := t.search.BulkIndex(ctx, documents, identity); err != nil {
|
||||
return err
|
||||
}
|
||||
// 快照里没有的文档必须从生产索引删掉,否则撤销只在数据库生效。
|
||||
keep := make([]string, 0, len(documents))
|
||||
for _, document := range documents {
|
||||
keep = append(keep, document.ID)
|
||||
}
|
||||
if _, err := t.search.PruneScope(ctx, identity, keep); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Verify 用行数 + 身份集合摘要 + 索引条数核对生产与预发布快照是否一致。
|
||||
func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrelease.Expectation) error {
|
||||
archive, err := librarysnapshot.Open(t.config.ArchiveRoot, 256*1024*1024)
|
||||
if err != nil {
|
||||
return errors.New("归档不可用")
|
||||
}
|
||||
defer archive.Close()
|
||||
manifest, err := archive.Load(expect.SnapshotID)
|
||||
if err != nil {
|
||||
return errors.New("预发布快照不可读")
|
||||
}
|
||||
for _, part := range manifest.Parts {
|
||||
if part.Kind != "database" && part.Kind != "wiki" {
|
||||
continue
|
||||
}
|
||||
count, digest, err := t.tableState(ctx, part.Name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
wantCount, wantDigest, err := archivedTableState(archive, part)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if count != wantCount || digest != wantDigest {
|
||||
return fmt.Errorf("生产 %s 与快照不一致:行数 %d/%d", part.Name, count, wantCount)
|
||||
}
|
||||
}
|
||||
indexCount, err := t.search.Count(ctx, SearchIdentity{TenantID: t.scope.TenantID,
|
||||
KnowledgeBaseID: t.scope.KnowledgeBaseID})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, part := range manifest.Parts {
|
||||
if part.Kind == "index" && part.Count != indexCount {
|
||||
return fmt.Errorf("生产索引条数 %d 与快照 %d 不一致", indexCount, part.Count)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func archivedTableState(archive *librarysnapshot.Archive, part librarysnapshot.Part) (int64, string, error) {
|
||||
reader, err := archive.OpenPart(part)
|
||||
if err != nil {
|
||||
return 0, "", err
|
||||
}
|
||||
defer reader.Close()
|
||||
decoder := json.NewDecoder(reader)
|
||||
token, err := decoder.Token()
|
||||
if err != nil || token != json.Delim('[') {
|
||||
return 0, "", errors.New("归档部件不是行数组")
|
||||
}
|
||||
ids := make([]string, 0, 64)
|
||||
for decoder.More() {
|
||||
var row map[string]json.RawMessage
|
||||
if err := decoder.Decode(&row); err != nil {
|
||||
return 0, "", err
|
||||
}
|
||||
identity := archiveRowIdentity(part.Name, row)
|
||||
if identity == "" {
|
||||
return 0, "", fmt.Errorf("归档 %s 行缺少稳定身份", part.Name)
|
||||
}
|
||||
ids = append(ids, identity)
|
||||
}
|
||||
sort.Strings(ids)
|
||||
return int64(len(ids)), digestIDs(ids), nil
|
||||
}
|
||||
|
||||
func archiveRowIdentity(table string, row map[string]json.RawMessage) string {
|
||||
value := func(key string) string {
|
||||
var text string
|
||||
_ = json.Unmarshal(row[key], &text)
|
||||
return text
|
||||
}
|
||||
if table == "knowledge_tag_relations" {
|
||||
knowledge, tag := value("knowledge_id"), value("tag_id")
|
||||
if knowledge == "" || tag == "" {
|
||||
return ""
|
||||
}
|
||||
return knowledge + ":" + tag
|
||||
}
|
||||
return value("id")
|
||||
}
|
||||
|
||||
// Restore 用预备份把生产整库恢复回去:走同一条 Apply 路径,可重复执行。
|
||||
func (t *ProductionLibraryTarget) Restore(ctx context.Context, backup libraryrelease.BackupRef) error {
|
||||
if backup.SnapshotID == "" {
|
||||
return errors.New("恢复需要绑定的预备份快照")
|
||||
}
|
||||
for _, stage := range libraryrelease.Stages() {
|
||||
if err := t.Apply(ctx, stage, backup.SnapshotID); err != nil {
|
||||
return fmt.Errorf("恢复阶段 %s 失败: %w", stage, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,280 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"knowledge-governance/internal/libraryrelease"
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
)
|
||||
|
||||
// 隔离集成验证:真实 PostgreSQL + 真实 Elasticsearch,不放桩。
|
||||
// 需要显式一次性隔离库与随机索引(与既有捕获集成测试同一套环境变量),
|
||||
// 默认 CI 不连接这些服务,因此在 test-scope.json 里声明为可选未运行。
|
||||
func TestLibraryProductionTargetIsolated(t *testing.T) {
|
||||
dsn, base := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL"), os.Getenv("KG_LIBRARY_TEST_ELASTIC_URL")
|
||||
if dsn == "" || base == "" || os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" {
|
||||
t.Skip("requires isolated PG and ES opt-in")
|
||||
}
|
||||
cfg, err := pgxpool.ParseConfig(dsn)
|
||||
if err != nil || !strings.HasPrefix(cfg.ConnConfig.Database, "kg_library_test_") {
|
||||
t.Fatal("isolated database required")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
|
||||
defer cancel()
|
||||
pool, err := pgxpool.NewWithConfig(ctx, cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("open isolated PG: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
if _, err := pool.Exec(ctx, `DROP TABLE IF EXISTS knowledges,chunks,knowledge_tags,knowledge_tag_relations,
|
||||
resources,resource_bindings,wiki_pages,wiki_folders,wiki_page_issues,wiki_log_entries,knowledge_bases,schema_migrations CASCADE`); err != nil {
|
||||
t.Fatalf("reset isolated schema: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, libraryTestSchema); err != nil {
|
||||
t.Fatalf("create schema: %v", err)
|
||||
}
|
||||
// 真实 schema 里这两张表都带 tenant_id;夹具补上,避免为了测试去简化生产 SQL。
|
||||
if _, err := pool.Exec(ctx, `ALTER TABLE wiki_pages ADD COLUMN status text DEFAULT 'published';
|
||||
ALTER TABLE knowledge_tag_relations ADD COLUMN tenant_id bigint`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
token := make([]byte, 8)
|
||||
if _, err := rand.Read(token); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
index := "kg_library_test_" + hex.EncodeToString(token)
|
||||
client := &http.Client{Timeout: 20 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error {
|
||||
return http.ErrUseLastResponse
|
||||
}}
|
||||
call := func(method, path, body string, want int) {
|
||||
t.Helper()
|
||||
request, err := http.NewRequestWithContext(ctx, method, base+path, strings.NewReader(body))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
response, err := client.Do(request)
|
||||
if err != nil {
|
||||
t.Fatalf("fixture HTTP: %v", err)
|
||||
}
|
||||
defer response.Body.Close()
|
||||
if response.StatusCode != want {
|
||||
t.Fatalf("fixture %s %s => %d, want %d", method, path, response.StatusCode, want)
|
||||
}
|
||||
}
|
||||
call("PUT", "/"+index, `{"mappings":{"properties":{"tenant_id":{"type":"long"},"knowledge_base_id":{"type":"keyword"},"knowledge_id":{"type":"keyword"},"chunk_id":{"type":"keyword"},"content":{"type":"text"},"embedding":{"type":"dense_vector","dims":2}}}}`, 200)
|
||||
defer func() {
|
||||
call("DELETE", "/"+index, "", 200)
|
||||
call("HEAD", "/"+index, "", 404)
|
||||
}()
|
||||
|
||||
archiveRoot := t.TempDir()
|
||||
if err := os.Chmod(archiveRoot, 0700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
archive, err := librarysnapshot.Open(archiveRoot, 64*1024*1024)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer archive.Close()
|
||||
first := sealTargetFixture(t, archive, "internal", []string{"doc-safe"}, []string{"chunk-safe"}, "wiki-safe")
|
||||
second := sealTargetFixture(t, archive, "internal", []string{"doc-safe"}, nil, "wiki-safe")
|
||||
|
||||
target, err := NewProductionLibraryTarget(ctx, ProductionLibraryConfig{
|
||||
Postgres: dsn, SearchURL: base, SearchIndex: index,
|
||||
Objects: fixtureObjectStore{}, ObjectsReader: fixtureObjectStore{}, ArchiveRoot: archiveRoot,
|
||||
MaxDocuments: 100, HTTPClient: client,
|
||||
}, ProductionScope{TenantID: 10000, KnowledgeBaseID: "kb-safe"})
|
||||
if err != nil {
|
||||
t.Fatalf("target: %v", err)
|
||||
}
|
||||
defer target.Close()
|
||||
|
||||
for _, stage := range []libraryrelease.Stage{libraryrelease.StageDatabase, libraryrelease.StageWiki, libraryrelease.StageIndex} {
|
||||
if err := target.Apply(ctx, stage, first); err != nil {
|
||||
t.Fatalf("apply %s: %v", stage, err)
|
||||
}
|
||||
}
|
||||
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: first}); err != nil {
|
||||
t.Fatalf("verify after first sync: %v", err)
|
||||
}
|
||||
var chunks int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM chunks WHERE tenant_id=10000 AND knowledge_base_id='kb-safe'`).Scan(&chunks); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if chunks != 1 {
|
||||
t.Fatalf("chunks after first sync = %d, want 1", chunks)
|
||||
}
|
||||
|
||||
// 第二个快照没有 chunk:撤销必须真的从生产消失,而不是"只增不减"。
|
||||
for _, stage := range []libraryrelease.Stage{libraryrelease.StageDatabase, libraryrelease.StageWiki, libraryrelease.StageIndex} {
|
||||
if err := target.Apply(ctx, stage, second); err != nil {
|
||||
t.Fatalf("apply second %s: %v", stage, err)
|
||||
}
|
||||
}
|
||||
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: second}); err != nil {
|
||||
t.Fatalf("verify after removal: %v", err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM chunks WHERE tenant_id=10000 AND knowledge_base_id='kb-safe'`).Scan(&chunks); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if chunks != 0 {
|
||||
t.Fatalf("chunk removal did not reach production: %d", chunks)
|
||||
}
|
||||
// 跨库数据不得被删:另一个租户/同名 KB 的行原样保留。
|
||||
var others int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM chunks WHERE tenant_id=20000`).Scan(&others); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if others != 1 {
|
||||
t.Fatalf("cross-scope rows must survive, got %d", others)
|
||||
}
|
||||
// 恢复:把第一个快照再应用一次,撤销的内容回来。
|
||||
if err := target.Restore(ctx, libraryrelease.BackupRef{ID: "backup", SnapshotID: first}); err != nil {
|
||||
t.Fatalf("restore: %v", err)
|
||||
}
|
||||
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: first}); err != nil {
|
||||
t.Fatalf("verify after restore: %v", err)
|
||||
}
|
||||
|
||||
// 独立预备份:从真实生产当前状态导出成归档版本,再用它做恢复演练。
|
||||
mapping, err := target.Search().MappingSHA256(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("mapping: %v", err)
|
||||
}
|
||||
// 预备份与发布快照放在同一个归档根(生产上就是 LIBRARY_ARCHIVE_ROOT):
|
||||
// 恢复时按快照 ID 回放,不依赖"备份路径"这种第二套寻址。
|
||||
backupRoot := archiveRoot
|
||||
backups, err := NewProductionBackupStore(ProductionBackupConfig{
|
||||
Pool: target.Pool(), Search: target.Search(), Objects: fixtureObjectStore{}, ArchiveRoot: backupRoot,
|
||||
Scope: "internal", Identity: target.Identity(), MaxDocuments: 100,
|
||||
Compatibility: librarysnapshot.Compatibility{WeKnoraVersion: "v0.7.1", SchemaVersion: 74,
|
||||
RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "text-embedding-v4",
|
||||
EmbeddingDimension: 1024, IndexMappingSHA256: mapping},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("backup store: %v", err)
|
||||
}
|
||||
current, err := target.Current(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("current: %v", err)
|
||||
}
|
||||
backup, err := backups.Prepare(ctx, first, current)
|
||||
if err != nil {
|
||||
t.Fatalf("prepare backup: %v", err)
|
||||
}
|
||||
if backup.SnapshotID == "" || backup.Bytes <= 0 {
|
||||
t.Fatalf("backup must carry a snapshot and bytes: %+v", backup)
|
||||
}
|
||||
|
||||
// 故障注入:直接破坏生产状态,校验必须失败,恢复后必须通过。
|
||||
if _, err := pool.Exec(ctx, `DELETE FROM chunks WHERE id='chunk-safe'`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: first}); err == nil {
|
||||
t.Fatal("verify must detect a missing chunk row")
|
||||
}
|
||||
if err := target.Restore(ctx, backup); err != nil {
|
||||
t.Fatalf("restore from backup: %v", err)
|
||||
}
|
||||
if err := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: backup.SnapshotID}); err != nil {
|
||||
t.Fatalf("verify after restore from backup: %v", err)
|
||||
}
|
||||
t.Logf("backup_snapshot=%s bytes=%d restored_and_verified=true", backup.SnapshotID, backup.Bytes)
|
||||
}
|
||||
|
||||
// fixtureObjectStore 只为让目标构造通过:本场景没有资源部件,不会被调用。
|
||||
type fixtureObjectStore struct{}
|
||||
|
||||
func (fixtureObjectStore) Open(context.Context, string, string) (io.ReadCloser, error) {
|
||||
return nil, errors.New("fixture object store is read-only here")
|
||||
}
|
||||
|
||||
func (fixtureObjectStore) Put(context.Context, string, string, io.Reader, int64) error {
|
||||
return errors.New("fixture object store must not be written in this scenario")
|
||||
}
|
||||
|
||||
func (fixtureObjectStore) Stat(context.Context, string, string) (int64, string, error) {
|
||||
return 0, "", errors.New("fixture object store has no objects")
|
||||
}
|
||||
|
||||
func (fixtureObjectStore) Copy(context.Context, string, string, string) error {
|
||||
return errors.New("fixture object store is read-only here")
|
||||
}
|
||||
|
||||
// sealTargetFixture 造一个最小整库快照:docs/chunks/wiki 行 + 一份索引文档。
|
||||
func sealTargetFixture(t *testing.T, archive *librarysnapshot.Archive, scope string, docs, chunks []string, wikiPage string) string {
|
||||
t.Helper()
|
||||
encode := func(value any) []byte {
|
||||
encoded, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return encoded
|
||||
}
|
||||
parts := make([]librarysnapshot.Part, 0, 3)
|
||||
put := func(kind, name string, count int64, body []byte) {
|
||||
part, err := archive.Put(bytes.NewReader(body))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
part.Kind, part.Name, part.Count = kind, name, count
|
||||
parts = append(parts, part)
|
||||
}
|
||||
knowledgeRows := make([]map[string]any, 0, len(docs))
|
||||
for _, id := range docs {
|
||||
knowledgeRows = append(knowledgeRows, map[string]any{"id": id, "tenant_id": 10000, "knowledge_base_id": "kb-safe",
|
||||
"parse_status": "completed", "pending_subtasks_count": 0, "file_path": ""})
|
||||
}
|
||||
chunkRows := make([]map[string]any, 0, len(chunks))
|
||||
for _, id := range chunks {
|
||||
chunkRows = append(chunkRows, map[string]any{"id": id, "tenant_id": 10000, "knowledge_base_id": "kb-safe",
|
||||
"knowledge_id": "doc-safe", "content": "chunk"})
|
||||
}
|
||||
// Seal 要求完整清单:10 张知识域表 + 索引,缺一张就不是"完整版本"。
|
||||
put("database", "knowledges", int64(len(knowledgeRows)), encode(knowledgeRows))
|
||||
put("database", "chunks", int64(len(chunkRows)), encode(chunkRows))
|
||||
for _, name := range []string{"knowledge_tags", "knowledge_tag_relations", "resource_bindings", "resources"} {
|
||||
put("database", name, 0, encode([]map[string]any{}))
|
||||
}
|
||||
wikiRows := []map[string]any{{"id": wikiPage, "tenant_id": 10000, "knowledge_base_id": "kb-safe",
|
||||
"slug": "index", "content": "wiki"}}
|
||||
put("wiki", "wiki_pages", 1, encode(wikiRows))
|
||||
for _, name := range []string{"wiki_folders", "wiki_page_issues", "wiki_log_entries"} {
|
||||
put("wiki", name, 0, encode([]map[string]any{}))
|
||||
}
|
||||
documents := make([]map[string]any, 0, len(chunks))
|
||||
for _, id := range chunks {
|
||||
documents = append(documents, map[string]any{"id": id, "source": map[string]any{
|
||||
"tenant_id": 10000, "knowledge_base_id": "kb-safe", "knowledge_id": "doc-safe",
|
||||
"chunk_id": id, "source_id": id, "content": "chunk", "embedding": []float64{0.1, 0.2}}})
|
||||
}
|
||||
put("index", "chunks", int64(len(documents)), encode(documents))
|
||||
manifest := librarysnapshot.Manifest{ValidationPolicy: librarysnapshot.ArtifactAcceptancePolicy,
|
||||
Format: librarysnapshot.Format, Scope: scope, TenantID: 10000, KnowledgeBaseID: "kb-safe",
|
||||
CreatedAt: time.Now().UTC(), Parts: parts,
|
||||
Compatibility: librarysnapshot.Compatibility{
|
||||
WeKnoraVersion: "v0.7.1", SchemaVersion: 74, RetrieverEngine: "elasticsearch_v8",
|
||||
EmbeddingModel: "text-embedding-v4", EmbeddingDimension: 1024,
|
||||
IndexMappingSHA256: strings.Repeat("a", 64),
|
||||
}}
|
||||
id, err := archive.Seal(manifest)
|
||||
if err != nil {
|
||||
t.Fatalf("seal fixture: %v", err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
@@ -4,32 +4,28 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"knowledge-governance/internal/libraryrelease"
|
||||
)
|
||||
|
||||
// 未配置时必须显式失败:不能返回零值让上层误以为"生产当前状态为空"。
|
||||
func TestProductionLibraryTargetFailsClosedWhenUnconfigured(t *testing.T) {
|
||||
if _, err := NewProductionLibraryTarget("", ""); !errors.Is(err, ErrProductionTargetUnconfigured) {
|
||||
t.Fatalf("constructor must refuse empty configuration, got %v", err)
|
||||
func TestProductionLibraryTargetRejectsIncompleteConfig(t *testing.T) {
|
||||
if _, err := NewProductionLibraryTarget(context.Background(), ProductionLibraryConfig{},
|
||||
ProductionScope{TenantID: 10000, KnowledgeBaseID: "kb-1"}); !errors.Is(err, ErrProductionTargetUnconfigured) {
|
||||
t.Fatalf("empty config must be refused, got %v", err)
|
||||
}
|
||||
target, err := NewProductionLibraryTarget("http://prod.invalid", "/tmp/key")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
if _, err := NewProductionLibraryTarget(context.Background(), ProductionLibraryConfig{Postgres: "postgres://x"},
|
||||
ProductionScope{TenantID: 10000, KnowledgeBaseID: "kb-1"}); !errors.Is(err, ErrProductionTargetUnconfigured) {
|
||||
t.Fatalf("partial config must be refused, got %v", err)
|
||||
}
|
||||
if _, err = target.Current(context.Background()); !errors.Is(err, ErrProductionTargetUnconfigured) {
|
||||
t.Fatalf("Current must fail closed, got %v", err)
|
||||
}
|
||||
if err = target.Apply(context.Background(), libraryrelease.StageDatabase, "snap-1"); !errors.Is(err, ErrProductionTargetUnconfigured) {
|
||||
t.Fatalf("Apply must fail closed, got %v", err)
|
||||
}
|
||||
if err = target.Verify(context.Background(), libraryrelease.Expectation{}); !errors.Is(err, ErrProductionTargetUnconfigured) {
|
||||
t.Fatalf("Verify must fail closed, got %v", err)
|
||||
}
|
||||
if err = target.Restore(context.Background(), libraryrelease.BackupRef{}); !errors.Is(err, ErrProductionTargetUnconfigured) {
|
||||
t.Fatalf("Restore must fail closed, got %v", err)
|
||||
}
|
||||
if _, err = (ArchiveBackupStore{}).Prepare(context.Background(), "snap-1", libraryrelease.CurrentState{}); err == nil {
|
||||
t.Fatal("backup store must refuse until wired")
|
||||
if _, err := NewProductionLibraryTarget(context.Background(), ProductionLibraryConfig{
|
||||
Postgres: "postgres://x", SearchURL: "http://es.invalid", SearchIndex: "idx",
|
||||
Objects: &S3ObjectStore{}, ArchiveRoot: "/tmp/a", MaxDocuments: 10,
|
||||
}, ProductionScope{}); !errors.Is(err, ErrProductionTargetUnconfigured) {
|
||||
t.Fatalf("missing scope identity must be refused, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProductionBackupStoreRejectsIncompleteConfig(t *testing.T) {
|
||||
if _, err := NewProductionBackupStore(ProductionBackupConfig{}); err == nil {
|
||||
t.Fatal("empty backup config must be refused")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,197 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
)
|
||||
|
||||
// Count 用与导出相同的 scope filter 数生产索引里的文档,供校验对比。
|
||||
func (exporter *SearchExporter) Count(ctx context.Context, identity SearchIdentity) (int64, error) {
|
||||
if identity.TenantID == 0 || !validSearchIdentifier(identity.KnowledgeBaseID) {
|
||||
return 0, errors.New("invalid search identity")
|
||||
}
|
||||
inspection, err := exporter.inspectMapping(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
// 只按"租户 + 知识库"计数:整库同步没有单文档身份,带上空的 knowledge 条件
|
||||
// 会把过滤器变成"匹配空字符串",永远数不到东西。
|
||||
filter := make([]any, 0, 2)
|
||||
if inspection.fields.tenant != "" {
|
||||
filter = append(filter, map[string]any{"term": map[string]any{inspection.fields.tenant: identity.TenantID}})
|
||||
}
|
||||
filter = append(filter, map[string]any{"term": map[string]any{inspection.fields.base: identity.KnowledgeBaseID}})
|
||||
payload := map[string]any{"query": map[string]any{"bool": map[string]any{"filter": filter}}}
|
||||
var response struct {
|
||||
Count int64 `json:"count"`
|
||||
}
|
||||
if err := exporter.requestJSON(ctx, http.MethodPost, "/"+exporter.index+"/_count", payload, &response); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if response.Count < 0 {
|
||||
return 0, errors.New("Elasticsearch returned a negative count")
|
||||
}
|
||||
return response.Count, nil
|
||||
}
|
||||
|
||||
// PruneScope 删除该知识库范围内、但不在这份快照里的索引文档:
|
||||
// 撤销必须真的从生产检索里消失,不能只做"新增/覆盖"。
|
||||
func (exporter *SearchExporter) PruneScope(ctx context.Context, identity SearchIdentity, keepIDs []string) (int, error) {
|
||||
inspection, err := exporter.inspectMapping(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
keep := make(map[string]bool, len(keepIDs))
|
||||
for _, id := range keepIDs {
|
||||
keep[id] = true
|
||||
}
|
||||
stale := make([]string, 0, 16)
|
||||
for from := 0; ; from += exporter.pageSize {
|
||||
payload := map[string]any{
|
||||
"from": from, "size": exporter.pageSize, "_source": false, "sort": []any{"_doc"},
|
||||
"query": map[string]any{"bool": map[string]any{"filter": scopeFilterFor(inspection.fields, identity)}},
|
||||
}
|
||||
var response struct {
|
||||
Hits struct {
|
||||
Hits []struct {
|
||||
ID string `json:"_id"`
|
||||
} `json:"hits"`
|
||||
} `json:"hits"`
|
||||
}
|
||||
if err := exporter.requestJSON(ctx, http.MethodPost, "/"+exporter.index+"/_search", payload, &response); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
for _, hit := range response.Hits.Hits {
|
||||
if !keep[hit.ID] {
|
||||
stale = append(stale, hit.ID)
|
||||
}
|
||||
}
|
||||
if len(response.Hits.Hits) < exporter.pageSize {
|
||||
break
|
||||
}
|
||||
if from+exporter.pageSize >= 10000 {
|
||||
return 0, errors.New("scope has too many documents to prune safely")
|
||||
}
|
||||
}
|
||||
if len(stale) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
for start := 0; start < len(stale); start += 500 {
|
||||
end := start + 500
|
||||
if end > len(stale) {
|
||||
end = len(stale)
|
||||
}
|
||||
payload := map[string]any{
|
||||
"query": map[string]any{"ids": map[string]any{"values": stale[start:end]}},
|
||||
}
|
||||
var response struct {
|
||||
Deleted int `json:"deleted"`
|
||||
}
|
||||
if err := exporter.requestJSON(ctx, http.MethodPost, "/"+exporter.index+"/_delete_by_query?refresh=true", payload, &response); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
}
|
||||
return len(stale), nil
|
||||
}
|
||||
|
||||
// scopeFilterFor 只按租户 + 知识库过滤(整库范围,不含单文档身份)。
|
||||
func scopeFilterFor(fields searchFieldNames, identity SearchIdentity) []any {
|
||||
filter := make([]any, 0, 2)
|
||||
if fields.tenant != "" {
|
||||
filter = append(filter, map[string]any{"term": map[string]any{fields.tenant: identity.TenantID}})
|
||||
}
|
||||
filter = append(filter, map[string]any{"term": map[string]any{fields.base: identity.KnowledgeBaseID}})
|
||||
return filter
|
||||
}
|
||||
|
||||
// BulkIndex 把归档里的原始文档与向量直接写入生产索引(不解析、不重新 Embedding)。
|
||||
// 写入前补齐 tenant_id/knowledge_base_id,避免归档文档缺字段时落到别的库。
|
||||
func (exporter *SearchExporter) BulkIndex(ctx context.Context, documents []librarysnapshot.IndexDocument, identity SearchIdentity) error {
|
||||
if len(documents) == 0 {
|
||||
return nil
|
||||
}
|
||||
var body bytes.Buffer
|
||||
for _, document := range documents {
|
||||
if document.ID == "" {
|
||||
return errors.New("archived index document without id")
|
||||
}
|
||||
payload := document.Source
|
||||
var source map[string]json.RawMessage
|
||||
if err := json.Unmarshal(document.Source, &source); err != nil {
|
||||
return errors.New("archived index document is not an object")
|
||||
}
|
||||
source["tenant_id"], _ = json.Marshal(identity.TenantID)
|
||||
source["knowledge_base_id"], _ = json.Marshal(identity.KnowledgeBaseID)
|
||||
encoded, err := json.Marshal(source)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
payload = encoded
|
||||
action, err := json.Marshal(map[string]any{"index": map[string]string{
|
||||
"_index": exporter.index, "_id": document.ID,
|
||||
}})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
body.Write(action)
|
||||
body.WriteByte('\n')
|
||||
body.Write(payload)
|
||||
body.WriteByte('\n')
|
||||
}
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodPost,
|
||||
exporter.baseURL.String()+"/_bulk?refresh=true", bytes.NewReader(body.Bytes()))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
request.Header.Set("Content-Type", "application/x-ndjson")
|
||||
if exporter.authorizationToken != "" {
|
||||
request.Header.Set("Authorization", "Bearer "+exporter.authorizationToken)
|
||||
}
|
||||
response, err := exporter.client.Do(request)
|
||||
if err != nil {
|
||||
return errors.New("Elasticsearch bulk request failed")
|
||||
}
|
||||
defer response.Body.Close()
|
||||
raw, err := io.ReadAll(io.LimitReader(response.Body, maxSearchResponseBytes))
|
||||
if err != nil {
|
||||
return errors.New("Elasticsearch bulk response unreadable")
|
||||
}
|
||||
if response.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("Elasticsearch bulk status %d", response.StatusCode)
|
||||
}
|
||||
var parsed struct {
|
||||
Errors bool `json:"errors"`
|
||||
Items []map[string]struct {
|
||||
Status int `json:"status"`
|
||||
Error any `json:"error"`
|
||||
ID string `json:"_id"`
|
||||
} `json:"items"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &parsed); err != nil {
|
||||
return errors.New("Elasticsearch bulk response invalid")
|
||||
}
|
||||
if parsed.Errors {
|
||||
for _, item := range parsed.Items {
|
||||
for _, result := range item {
|
||||
if result.Error != nil || result.Status >= 300 {
|
||||
return fmt.Errorf("Elasticsearch rejected document %s (status %d)", result.ID, result.Status)
|
||||
}
|
||||
}
|
||||
}
|
||||
return errors.New("Elasticsearch bulk reported errors")
|
||||
}
|
||||
if len(parsed.Items) != len(documents) {
|
||||
return fmt.Errorf("Elasticsearch applied %d of %d documents", len(parsed.Items), len(documents))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
var _ = strings.TrimSpace
|
||||
@@ -168,6 +168,11 @@
|
||||
"test": "TestLocalMirrorPeriodicSyncIsDurableAndDueOnly",
|
||||
"reason": "仅显式一次性隔离PG验证本地副本/持久同步或专门只读源夹具;默认CI不连接共享与生产库,另行实际执行隔离验收"
|
||||
},
|
||||
{
|
||||
"package": "knowledge-governance/internal/releaseadapter/v071",
|
||||
"test": "TestLibraryProductionTargetIsolated",
|
||||
"reason": "需要一次性隔离库 kg_library_test_* 与随机 ES 索引(KG_LIBRARY_TEST_DATABASE_URL / KG_LIBRARY_TEST_ELASTIC_URL / KG_LIBRARY_TEST_ISOLATED=1);默认 CI 不提供,本机 WeKnora 测试栈已真实跑通导入/撤销/预备份/故障恢复,CI 中跳过不等于验证"
|
||||
},
|
||||
{
|
||||
"package": "knowledge-governance/internal/dobotaiops",
|
||||
"test": "TestMirrorSourceManifestDetectsUpdatesWithoutVolatileTimes",
|
||||
|
||||
Reference in New Issue
Block a user