diff --git a/cmd/release-worker/config.go b/cmd/release-worker/config.go index 343b2a0..02853c6 100644 --- a/cmd/release-worker/config.go +++ b/cmd/release-worker/config.go @@ -15,8 +15,19 @@ import ( const maxReleaseConfigBytes = 1024 * 1024 type releaseRuntimeConfig struct { - ArchiveRoot string `json:"archive_root"` - Routes []releaseRouteConfig `json:"routes"` + ArchiveRoot string `json:"archive_root"` + // TargetSnapshot 是"写目标库前必须留档"的门禁配置。 + TargetSnapshot releaseTargetSnapshotConfig `json:"target_snapshot"` + Routes []releaseRouteConfig `json:"routes"` +} + +// releaseTargetSnapshotConfig 指向快照根目录（scripts/target_kb_snapshot.sh 的输出 +// 目录），并声明门禁对哪些项生效。 +type releaseTargetSnapshotConfig struct { + Root string `json:"root"` + MaxAgeHours int `json:"max_age_hours"` + // Policy：mapped（默认，只对 源库→目标库 映射的共享库强制）| always（一律强制）。 + Policy string `json:"policy"` } type releaseRouteConfig struct { @@ -98,6 +109,9 @@ func validateReleaseRuntimeConfig(config releaseRuntimeConfig) error { if !filepath.IsAbs(config.ArchiveRoot) || len(config.Routes) == 0 { return errors.New("release archive root and routes are required") } + if err := validateTargetSnapshotConfig(config.TargetSnapshot); err != nil { + return err + } seen := map[string]bool{} for _, route := range config.Routes { if route.TenantID == 0 || !validConfigIdentifier(route.KnowledgeBaseID) || !validConfigIdentifier(route.EmbeddingModel) || seen[route.KnowledgeBaseID] { @@ -135,6 +149,27 @@ func validObjectEndpoint(object releaseObjectConfig) bool { func absoluteFile(path string) bool { return filepath.IsAbs(path) && filepath.Clean(path) == path } +// validateTargetSnapshotConfig 校验快照门禁配置。留档目录缺省即视为"未接线"， +// 但一旦声明要留档（policy=always）就必须给出目录——否则是假门禁。 +func validateTargetSnapshotConfig(config releaseTargetSnapshotConfig) error { + if config.Policy != "" && config.Policy != "mapped" && config.Policy != "always" { + return errors.New("release target snapshot policy must be mapped or always") + } + if config.MaxAgeHours < 0 || config.MaxAgeHours > 720 { + return errors.New("release target snapshot max_age_hours must be between 0 and 720") + } + if config.Root == "" { + if config.Policy == "always" { + return errors.New("release target snapshot root is required when policy is always") + } + return nil + } + if !filepath.IsAbs(config.Root) || filepath.Clean(config.Root) != config.Root { + return errors.New("release target snapshot root must be an absolute clean path") + } + return nil +} + func validHTTPURL(value string) bool { parsed, err := url.Parse(value) return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") && parsed.Host != "" && parsed.User == nil && parsed.RawQuery == "" && parsed.Fragment == "" diff --git a/cmd/release-worker/runtime.go b/cmd/release-worker/runtime.go index 7666877..e4430af 100644 --- a/cmd/release-worker/runtime.go +++ b/cmd/release-worker/runtime.go @@ -13,6 +13,7 @@ import ( "knowledge-governance/internal/platform/weknora" v071 "knowledge-governance/internal/releaseadapter/v071" + "knowledge-governance/internal/releasesnapshot" releaseworkflow "knowledge-governance/internal/workflows/release" ) @@ -31,7 +32,7 @@ func buildReleaseRuntime(ctx context.Context, config releaseRuntimeConfig) (*rel runtime := &releaseRuntime{} publishers := make(map[string]releaseworkflow.Publisher, len(config.Routes)) for _, route := range config.Routes { - publisher, pools, err := buildReleaseRoute(ctx, config.ArchiveRoot, route) + publisher, pools, err := buildReleaseRoute(ctx, config.ArchiveRoot, config.TargetSnapshot, route) if err != nil { runtime.Close() return nil, err @@ -48,7 +49,7 @@ func buildReleaseRuntime(ctx context.Context, config releaseRuntimeConfig) (*rel return runtime, nil } -func buildReleaseRoute(ctx context.Context, archiveRoot string, route releaseRouteConfig) (releaseworkflow.Publisher, []*pgxpool.Pool, error) { +func buildReleaseRoute(ctx context.Context, archiveRoot string, snapshot releaseTargetSnapshotConfig, route releaseRouteConfig) (releaseworkflow.Publisher, []*pgxpool.Pool, error) { sourceDatabaseURL, err := readRestrictedValue(route.Source.DatabaseURLFile) if err != nil { return nil, nil, errors.New("read source database credential") @@ -158,7 +159,21 @@ func buildReleaseRoute(ctx context.Context, archiveRoot string, route releaseRou if err != nil { return cleanup(errors.New("create v0.7.1 document target")) } - publisher, err := releaseworkflow.NewSidecarPublisher(documentExporter, documentTarget) + var publisherOptions []releaseworkflow.PublisherOption + if snapshot.Root != "" { + // 写目标库前必须有覆盖该库的、未过期的快照留档（见 releasesnapshot.Guard）。 + maxAge := time.Duration(snapshot.MaxAgeHours) * time.Hour + if snapshot.MaxAgeHours == 0 { + maxAge = 24 * time.Hour + } + policy := releaseworkflow.SnapshotPolicyMapped + if snapshot.Policy == "always" { + policy = releaseworkflow.SnapshotPolicyAlways + } + publisherOptions = append(publisherOptions, releaseworkflow.WithTargetSnapshots( + releasesnapshot.Guard{Root: snapshot.Root, MaxAge: maxAge, VerifyLocalFiles: true}, policy)) + } + publisher, err := releaseworkflow.NewSidecarPublisher(documentExporter, documentTarget, publisherOptions...) if err != nil { return cleanup(errors.New("create v0.7.1 release publisher")) } diff --git a/compose.release-sim.yaml b/compose.release-sim.yaml index 2c27e6e..52cdaae 100644 --- a/compose.release-sim.yaml +++ b/compose.release-sim.yaml @@ -17,6 +17,9 @@ services: environment: RELEASE_CONFIG_FILE: /run/release-sim/release.sim.json PRODUCTION_RELEASE_AUTHORIZER_REF: yaozhanghao + volumes: + # 发布前快照留档目录（只读）：写目标库之前必须存在覆盖该库的新快照 + - ./backups/target-kb-snapshots:/var/lib/knowledge-governance/target-kb-snapshots:ro extra_hosts: - "host.docker.internal:host-gateway" - "prod-pg:host-gateway" diff --git a/config/release.sim.json b/config/release.sim.json index a209d5e..1369d21 100644 --- a/config/release.sim.json +++ b/config/release.sim.json @@ -1,5 +1,10 @@ { "archive_root": "/var/lib/knowledge-governance/release-bundles", + "target_snapshot": { + "root": "/var/lib/knowledge-governance/target-kb-snapshots", + "max_age_hours": 24, + "policy": "mapped" + }, "routes": [ { "tenant_id": 10000, diff --git a/config/release.v071.example.json b/config/release.v071.example.json index d0d43e9..4168147 100644 --- a/config/release.v071.example.json +++ b/config/release.v071.example.json @@ -1,5 +1,10 @@ { "archive_root": "/var/lib/knowledge-governance/release-bundles", + "target_snapshot": { + "root": "/var/lib/knowledge-governance/target-kb-snapshots", + "max_age_hours": 24, + "policy": "mapped" + }, "routes": [ { "tenant_id": 10000, diff --git a/docs/P6_RELEASE_WORKER_ACCEPTANCE.md b/docs/P6_RELEASE_WORKER_ACCEPTANCE.md index cdbb76f..b177f99 100644 --- a/docs/P6_RELEASE_WORKER_ACCEPTANCE.md +++ b/docs/P6_RELEASE_WORKER_ACCEPTANCE.md @@ -301,7 +301,33 @@ INCLUDE_CHUNKS=0 bash scripts/target_kb_snapshot.sh # 不导 chunks（快照 约定：对**共享真实知识库**发布、撤权、回滚前必须先跑一次快照，并把快照目录与 `MANIFEST.json` 的 sha256 记进发布单；`previous_production_revision=absent` 时必须写明 -"无前版可回退，只有这份快照"。待办：把"快照存在"从人工约定做成发布前置硬门禁。 +"无前版可回退，只有这份快照"。 + +#### 硬门禁（写目标库之前） + +"必须有快照"已从人工约定升级为**发布前置硬门禁**（`internal/releasesnapshot`）： + +- Worker 配置新增： + + ```json + "target_snapshot": { + "root": "/var/lib/knowledge-governance/target-kb-snapshots", + "max_age_hours": 24, + "policy": "mapped" + } + ``` + + `root` 由 `compose.release-sim.yaml` 以只读方式挂载快照目录；`policy=mapped`（默认） + 只对**源库 → 目标库**映射的项强制（同 ID 目标的本地演练不受影响），`always` 则一律强制， + 且此时 `root` 缺失直接判定配置非法。 +- `Publish` / `Delete` / `Rollback` 在**任何写目标库的动作之前**检查：存在覆盖目标库、 + 未过期（`max_age_hours`，默认 24h）、且含 `wiki_pages.jsonl` 及其 sha256 的快照； + 本地文件会再按哈希校验一次（"可恢复"的最低证明）。任一条不满足—— + **包括门禁本身未接线**——都以 `release_target_snapshot_missing`（不可重试）拒绝发布， + 且不触碰目标库一行。页面把该码显示为"缺少目标知识库的发布前快照留档，已拒绝发布"。 +- 云内执行：`bash scripts/target_kb_snapshot.sh` 默认在与 DB 同 VPC 的云主机上跑 + （凭据取自已部署的 WeKnora 容器，不经家用带宽），实测 **44 秒**产出 8 张表 + 清单； + 隧道模式保留为 `EXECUTION=tunnel` 兜底。 ## Elasticsearch 通道 diff --git a/internal/releasesnapshot/guard.go b/internal/releasesnapshot/guard.go new file mode 100644 index 0000000..fa71332 --- /dev/null +++ b/internal/releasesnapshot/guard.go @@ -0,0 +1,189 @@ +// Package releasesnapshot 实现"发布前必须有目标知识库快照"的硬门禁。 +// +// 背景：发布链本身不做发布前备份（DocumentTarget.Prepare 只归档本包、解析 +// previousRevision），而 wiki 通道在共享真实知识库上会覆盖整库页面。留档由 +// scripts/target_kb_snapshot.sh 产生（KB 维度只读行级 JSONL + MANIFEST.json）， +// 本包只负责在写目标库之前证明"这份现场确实存在且是新的"。 +package releasesnapshot + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + "time" +) + +const manifestKind = "weknora-target-kb-snapshot" + +// ErrNoFreshSnapshot 表示找不到覆盖目标库、且在有效期内的快照。 +var ErrNoFreshSnapshot = errors.New("no fresh target knowledge base snapshot") + +type manifestFile struct { + File string `json:"file"` + Rows int `json:"rows"` + // Bytes/SHA256 由快照脚本写入；本地文件会再校验一次。 + Bytes int64 `json:"bytes"` + SHA256 string `json:"sha256"` + Location string `json:"location"` +} + +type manifest struct { + Kind string `json:"kind"` + CreatedAt string `json:"created_at"` + KnowledgeBaseIDs []string `json:"knowledge_base_ids"` + Files []manifestFile `json:"files"` +} + +// Guard 在快照根目录下寻找覆盖目标库的有效快照。 +type Guard struct { + // Root 是快照根目录（每份快照一个子目录，内含 MANIFEST.json）。 + Root string + // MaxAge 是允许的快照最大年龄（0 表示不限制年龄，仅要求存在）。 + MaxAge time.Duration + // Now 便于测试注入时间。 + Now func() time.Time + // VerifyLocalFiles 为真时校验本地 JSONL 的 sha256（"可恢复"的最低证明）。 + VerifyLocalFiles bool +} + +// Check 要求 Root 下存在一份覆盖 knowledgeBaseID 的快照：类型正确、在有效期内、 +// 含 wiki_pages.jsonl 且其 sha256 非空（本地文件再按哈希校验一次）。 +// 找到即返回 nil；否则返回包装了 ErrNoFreshSnapshot 的原因。 +func (guard Guard) Check(knowledgeBaseID string) error { + if strings.TrimSpace(knowledgeBaseID) == "" { + return fmt.Errorf("%w: 目标知识库为空", ErrNoFreshSnapshot) + } + root := strings.TrimSpace(guard.Root) + if root == "" { + return fmt.Errorf("%w: 未配置快照根目录（RELEASE_SNAPSHOT_ROOT）", ErrNoFreshSnapshot) + } + entries, err := os.ReadDir(root) + if err != nil { + return fmt.Errorf("%w: 读取快照目录失败: %v", ErrNoFreshSnapshot, err) + } + now := guard.now() + names := make([]string, 0, len(entries)) + for _, entry := range entries { + if entry.IsDir() { + names = append(names, entry.Name()) + } + } + // 目录名以 UTC 时间戳开头（-），倒序即从最新开始找。 + sort.Sort(sort.Reverse(sort.StringSlice(names))) + var newestProblem error + for _, name := range names { + ok, problem := guard.inspect(filepath.Join(root, name), knowledgeBaseID, now) + if ok { + return nil + } + if problem != nil && newestProblem == nil { + newestProblem = problem + } + } + if newestProblem != nil { + return newestProblem + } + return fmt.Errorf("%w: 未找到覆盖目标库 %s 的快照", ErrNoFreshSnapshot, knowledgeBaseID) +} + +func (guard Guard) now() time.Time { + if guard.Now != nil { + return guard.Now().UTC() + } + return time.Now().UTC() +} + +// inspect 判断一份快照是否可用；ok=true 表示通过。 +func (guard Guard) inspect(directory, knowledgeBaseID string, now time.Time) (bool, error) { + path := filepath.Join(directory, "MANIFEST.json") + payload, err := os.ReadFile(path) + if err != nil { + return false, nil // 目录里没有清单（半成品），跳过 + } + var parsed manifest + if err := json.Unmarshal(payload, &parsed); err != nil { + return false, fmt.Errorf("%w: 快照清单无法解析（%s）", ErrNoFreshSnapshot, filepath.Base(directory)) + } + if parsed.Kind != manifestKind { + return false, nil + } + covered := false + for _, id := range parsed.KnowledgeBaseIDs { + if id == knowledgeBaseID { + covered = true + break + } + } + if !covered { + return false, nil + } + created, err := parseStamp(parsed.CreatedAt) + if err != nil { + return false, fmt.Errorf("%w: 快照时间戳无法识别（%s）", ErrNoFreshSnapshot, filepath.Base(directory)) + } + if guard.MaxAge > 0 && now.Sub(created) > guard.MaxAge { + return false, fmt.Errorf("%w: 最新快照已过期（%s，超过 %s）", + ErrNoFreshSnapshot, created.Format(time.RFC3339), guard.MaxAge) + } + wiki, present := findFile(parsed.Files, "wiki_pages.jsonl") + if !present || wiki.SHA256 == "" { + return false, fmt.Errorf("%w: 快照缺少 wiki_pages.jsonl 或哈希（%s）", + ErrNoFreshSnapshot, filepath.Base(directory)) + } + if guard.VerifyLocalFiles { + if err := verifyLocal(directory, wiki); err != nil { + return false, err + } + } + return true, nil +} + +// parseStamp 接受脚本写出的 20261009T025017Z，也接受 RFC3339。 +func parseStamp(value string) (time.Time, error) { + value = strings.TrimSpace(value) + if created, err := time.Parse("20060102T150405Z", value); err == nil { + return created.UTC(), nil + } + if created, err := time.Parse(time.RFC3339, value); err == nil { + return created.UTC(), nil + } + return time.Time{}, errors.New("unrecognized snapshot timestamp") +} + +func findFile(files []manifestFile, name string) (manifestFile, bool) { + for _, file := range files { + if file.File == name { + return file, true + } + } + return manifestFile{}, false +} + +// verifyLocal 校验本地 JSONL 的 sha256，作为"这份快照还能用"的最小证明。 +// 放在云主机上（location=cloud:...）的文件无法在这里校验，跳过。 +func verifyLocal(directory string, file manifestFile) error { + if strings.HasPrefix(file.Location, "cloud:") { + return nil + } + handle, err := os.Open(filepath.Join(directory, file.File)) + if err != nil { + return fmt.Errorf("%w: 快照文件不可读（%s）", ErrNoFreshSnapshot, file.File) + } + defer handle.Close() + digest := sha256.New() + if _, err := io.Copy(digest, handle); err != nil { + return fmt.Errorf("%w: 快照文件校验失败（%s）", ErrNoFreshSnapshot, file.File) + } + actual := hex.EncodeToString(digest.Sum(nil)) + if actual != file.SHA256 { + return fmt.Errorf("%w: 快照文件哈希不匹配（%s）", ErrNoFreshSnapshot, file.File) + } + return nil +} diff --git a/internal/releasesnapshot/guard_test.go b/internal/releasesnapshot/guard_test.go new file mode 100644 index 0000000..8fc89bc --- /dev/null +++ b/internal/releasesnapshot/guard_test.go @@ -0,0 +1,123 @@ +package releasesnapshot + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func writeSnapshot(t *testing.T, root, name, createdAt string, kbIDs []string, wikiContent string, hashOverride string) string { + t.Helper() + directory := filepath.Join(root, name) + if err := os.MkdirAll(directory, 0o755); err != nil { + t.Fatal(err) + } + if wikiContent != "" { + if err := os.WriteFile(filepath.Join(directory, "wiki_pages.jsonl"), []byte(wikiContent), 0o644); err != nil { + t.Fatal(err) + } + } + digest := sha256.Sum256([]byte(wikiContent)) + sha := hex.EncodeToString(digest[:]) + if hashOverride != "" { + sha = hashOverride + } + files := []map[string]any{} + if wikiContent != "" { + files = append(files, map[string]any{ + "file": "wiki_pages.jsonl", "rows": 1, "bytes": len(wikiContent), "sha256": sha, "location": "local", + }) + } + manifestPayload, err := json.Marshal(map[string]any{ + "kind": manifestKind, "created_at": createdAt, "knowledge_base_ids": kbIDs, "files": files, + }) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(directory, "MANIFEST.json"), manifestPayload, 0o644); err != nil { + t.Fatal(err) + } + return directory +} + +func fixedNow() func() time.Time { + return func() time.Time { return time.Date(2026, time.October, 9, 3, 0, 0, 0, time.UTC) } +} + +func TestGuardAcceptsFreshSnapshotCoveringTargetKnowledgeBase(t *testing.T) { + root := t.TempDir() + writeSnapshot(t, root, "20261009T025017Z-0b217e8a", "20261009T025017Z", []string{"kb-a", "kb-b"}, "{}\n", "") + guard := Guard{Root: root, MaxAge: 24 * time.Hour, Now: fixedNow(), VerifyLocalFiles: true} + if err := guard.Check("kb-a"); err != nil { + t.Fatalf("fresh snapshot must pass: %v", err) + } +} + +// 过期快照不算数：门禁的语义是"发布前刚留过档"。 +func TestGuardRejectsExpiredSnapshot(t *testing.T) { + root := t.TempDir() + writeSnapshot(t, root, "20261006T025017Z-0b217e8a", "20261006T025017Z", []string{"kb-a"}, "{}\n", "") + guard := Guard{Root: root, MaxAge: 24 * time.Hour, Now: fixedNow(), VerifyLocalFiles: true} + err := guard.Check("kb-a") + if !errors.Is(err, ErrNoFreshSnapshot) { + t.Fatalf("expired snapshot must be refused, got %v", err) + } +} + +// 不覆盖目标库的快照不能用（另一个库的快照救不了这个库）。 +func TestGuardRejectsSnapshotOfAnotherKnowledgeBase(t *testing.T) { + root := t.TempDir() + writeSnapshot(t, root, "20261009T025017Z-0b217e8a", "20261009T025017Z", []string{"kb-other"}, "{}\n", "") + guard := Guard{Root: root, Now: fixedNow()} + if err := guard.Check("kb-a"); !errors.Is(err, ErrNoFreshSnapshot) { + t.Fatalf("snapshot of another base must be refused, got %v", err) + } +} + +// 关键文件哈希对不上 = 这份快照不可恢复，必须拒绝。 +func TestGuardRejectsTamperedSnapshot(t *testing.T) { + root := t.TempDir() + writeSnapshot(t, root, "20261009T025017Z-0b217e8a", "20261009T025017Z", []string{"kb-a"}, "{}\n", + strings.Repeat("0", 64)) + guard := Guard{Root: root, Now: fixedNow(), VerifyLocalFiles: true} + if err := guard.Check("kb-a"); !errors.Is(err, ErrNoFreshSnapshot) { + t.Fatalf("tampered snapshot must be refused, got %v", err) + } +} + +// 没有 wiki_pages.jsonl 的快照不算"现场留档"。 +func TestGuardRejectsSnapshotWithoutWikiPages(t *testing.T) { + root := t.TempDir() + writeSnapshot(t, root, "20261009T025017Z-0b217e8a", "20261009T025017Z", []string{"kb-a"}, "", "") + guard := Guard{Root: root, Now: fixedNow()} + if err := guard.Check("kb-a"); !errors.Is(err, ErrNoFreshSnapshot) { + t.Fatalf("snapshot without wiki pages must be refused, got %v", err) + } +} + +// 根目录未配置（门禁未接线）必须拒绝，而不是放行。 +func TestGuardFailsClosedWithoutRoot(t *testing.T) { + guard := Guard{Root: "", Now: fixedNow()} + if err := guard.Check("kb-a"); !errors.Is(err, ErrNoFreshSnapshot) { + t.Fatalf("unconfigured guard must fail closed, got %v", err) + } +} + +// 半成品目录（清单写了一半/缺失）要被跳过，最新那份可用即可。 +func TestGuardSkipsHalfWrittenSnapshotAndPicksNewestValid(t *testing.T) { + root := t.TempDir() + if err := os.MkdirAll(filepath.Join(root, "20261009T030000Z-0b217e8a"), 0o755); err != nil { + t.Fatal(err) + } + writeSnapshot(t, root, "20261009T025017Z-0b217e8a", "20261009T025017Z", []string{"kb-a"}, "{}\n", "") + guard := Guard{Root: root, MaxAge: 24 * time.Hour, Now: fixedNow(), VerifyLocalFiles: true} + if err := guard.Check("kb-a"); err != nil { + t.Fatalf("valid snapshot behind a half-written dir must still pass: %v", err) + } +} diff --git a/internal/workflows/release/handler.go b/internal/workflows/release/handler.go index befe200..bdf550b 100644 --- a/internal/workflows/release/handler.go +++ b/internal/workflows/release/handler.go @@ -27,7 +27,10 @@ const ( ErrorCodeRollbackArtifactMissing = "release_rollback_artifact_missing" ErrorCodeRollbackFailed = "release_rollback_failed" ErrorCodeStoppedAfterFailure = "release_stopped_after_failure" - ErrorCodeInternal = "release_internal_error" + // ErrorCodeTargetSnapshotMissing：写目标库之前必须存在覆盖该库的、未过期的 + // 快照留档（scripts/target_kb_snapshot.sh 产出），否则拒绝发布。 + ErrorCodeTargetSnapshotMissing = "release_target_snapshot_missing" + ErrorCodeInternal = "release_internal_error" ) var ( @@ -305,7 +308,8 @@ func ValidErrorCode(code string) bool { switch code { case ErrorCodeContractMismatch, ErrorCodeBundleInvalid, ErrorCodeTargetUnavailable, ErrorCodeTargetWriteFailed, ErrorCodeVerificationFailed, ErrorCodeRollbackArtifactMissing, - ErrorCodeRollbackFailed, ErrorCodeStoppedAfterFailure, ErrorCodeInternal: + ErrorCodeRollbackFailed, ErrorCodeStoppedAfterFailure, ErrorCodeTargetSnapshotMissing, + ErrorCodeInternal: return true default: return false diff --git a/internal/workflows/release/sidecar_publisher.go b/internal/workflows/release/sidecar_publisher.go index 2a66368..9e4744e 100644 --- a/internal/workflows/release/sidecar_publisher.go +++ b/internal/workflows/release/sidecar_publisher.go @@ -34,13 +34,73 @@ type SidecarTarget interface { type SidecarPublisher struct { exporter BundleExporter target SidecarTarget + // snapshots 是"写目标库前必须存在目标库快照"的门禁；nil 表示未接线。 + snapshots TargetSnapshotGuard + snapshotPolicy SnapshotPolicy } -func NewSidecarPublisher(exporter BundleExporter, target SidecarTarget) (*SidecarPublisher, error) { +// TargetSnapshotGuard 由 internal/releasesnapshot.Guard 实现。 +type TargetSnapshotGuard interface { + Check(knowledgeBaseID string) error +} + +// SnapshotPolicy 决定门禁对哪些项生效。 +type SnapshotPolicy int + +const ( + // SnapshotPolicyMapped：只对"源库 → 目标库"映射的项强制（共享真实库）。 + SnapshotPolicyMapped SnapshotPolicy = iota + // SnapshotPolicyAlways：所有写入目标库的项都要有快照。 + SnapshotPolicyAlways +) + +// PublisherOption 用于给发布器加装配件（保持既有两参构造的兼容）。 +type PublisherOption func(*SidecarPublisher) error + +// WithTargetSnapshots 装上快照门禁：写入目标库之前先证明"现场已留档"。 +func WithTargetSnapshots(guard TargetSnapshotGuard, policy SnapshotPolicy) PublisherOption { + return func(publisher *SidecarPublisher) error { + if guard == nil { + return errors.New("target snapshot guard is required") + } + publisher.snapshots = guard + publisher.snapshotPolicy = policy + return nil + } +} + +func NewSidecarPublisher(exporter BundleExporter, target SidecarTarget, options ...PublisherOption) (*SidecarPublisher, error) { if exporter == nil || target == nil { return nil, errors.New("release sidecar dependencies are required") } - return &SidecarPublisher{exporter: exporter, target: target}, nil + publisher := &SidecarPublisher{exporter: exporter, target: target} + for _, option := range options { + if option == nil { + continue + } + if err := option(publisher); err != nil { + return nil, err + } + } + return publisher, nil +} + +// ensureTargetSnapshot 在任何写目标库的动作之前调用：目标库是共享真实库 +// （源库 → 目标库 映射）或策略要求"一律"时，必须存在覆盖该库的新快照。 +// 门禁未接线（root 未配置）时对这类项**拒绝放行**，而不是跳过校验。 +func (publisher *SidecarPublisher) ensureTargetSnapshot(item PlanItem) error { + required := publisher.snapshotPolicy == SnapshotPolicyAlways || item.TargetBase() != item.TestKnowledgeBaseID + if !required { + return nil + } + if publisher.snapshots == nil { + return newSidecarError(ErrorCodeTargetSnapshotMissing, false, + errors.New("发布前快照门禁未配置（release config target_snapshot.root）")) + } + if err := publisher.snapshots.Check(item.TargetBase()); err != nil { + return newSidecarError(ErrorCodeTargetSnapshotMissing, false, err) + } + return nil } // releaseItemMapping declares which production base this item must land in: the @@ -53,6 +113,9 @@ func releaseItemMapping(item PlanItem) releasebundle.KnowledgeBaseMapping { } func (publisher *SidecarPublisher) Publish(ctx context.Context, item PlanItem) (ItemResult, error) { + if err := publisher.ensureTargetSnapshot(item); err != nil { + return ItemResult{}, err + } bundle, err := publisher.exporter.Export(ctx, item) if err != nil { return ItemResult{}, newSidecarError(ErrorCodeTargetUnavailable, true, err) @@ -95,6 +158,9 @@ func (publisher *SidecarPublisher) Publish(ctx context.Context, item PlanItem) ( } func (publisher *SidecarPublisher) Rollback(ctx context.Context, item PlanItem) (ItemResult, error) { + if err := publisher.ensureTargetSnapshot(item); err != nil { + return ItemResult{}, err + } if !validRevision(item.PreviousProductionRevision) { return ItemResult{}, newSidecarError(ErrorCodeRollbackArtifactMissing, false, errors.New("missing rollback revision")) } @@ -112,6 +178,9 @@ func (publisher *SidecarPublisher) Rollback(ctx context.Context, item PlanItem) } func (publisher *SidecarPublisher) Delete(ctx context.Context, item PlanItem) (ItemResult, error) { + if err := publisher.ensureTargetSnapshot(item); err != nil { + return ItemResult{}, err + } bundle, err := publisher.exporter.Export(ctx, item) if err != nil { return ItemResult{}, newSidecarError(ErrorCodeTargetUnavailable, true, err) diff --git a/internal/workflows/release/sidecar_publisher_test.go b/internal/workflows/release/sidecar_publisher_test.go index 036a0c7..efb334f 100644 --- a/internal/workflows/release/sidecar_publisher_test.go +++ b/internal/workflows/release/sidecar_publisher_test.go @@ -215,3 +215,148 @@ func testManifest() releasebundle.Manifest { func testTargetContract() releasebundle.TargetContract { return releasebundle.TargetContract{WeKnoraVersion: releasebundle.WeKnoraVersionV071, MigrationVersion: releasebundle.WeKnoraMigrationV071, TenantID: 10000, KnowledgeBaseID: "kb-safe", RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "bge-m3", EmbeddingDimension: releasebundle.EmbeddingDimensionV071, ElasticsearchMappingSHA256: strings.Repeat("a", 64)} } + +// --- 发布前快照门禁 --------------------------------------------------------- + +type stubSnapshotGuard struct { + checked []string + err error +} + +func (guard *stubSnapshotGuard) Check(knowledgeBaseID string) error { + guard.checked = append(guard.checked, knowledgeBaseID) + return guard.err +} + +func mappedPlanItem() PlanItem { + item := testPlanItem() + item.TargetKnowledgeBaseID = "kb-production" + return item +} + +// 共享真实库（源库 → 目标库 映射）没有快照时必须拒绝，且不得碰目标库。 +func TestSidecarPublisherRefusesMappedItemWithoutSnapshot(t *testing.T) { + calls := []string{} + manifest, err := releasebundle.Seal(testManifest()) + if err != nil { + t.Fatal(err) + } + exporter := &fakeExporter{manifest: manifest, calls: &calls} + target := &fakeSidecarTarget{contract: testTargetContract(), calls: &calls, prepared: PreparedChange{PreviousRevision: "prod-v0", ProductionRevision: "prod-v1"}} + guard := &stubSnapshotGuard{err: errors.New("no fresh target knowledge base snapshot")} + publisher, err := NewSidecarPublisher(exporter, target, WithTargetSnapshots(guard, SnapshotPolicyMapped)) + if err != nil { + t.Fatal(err) + } + _, publishErr := publisher.Publish(context.Background(), mappedPlanItem()) + var itemErr *ItemError + if !errors.As(publishErr, &itemErr) || itemErr.Code != ErrorCodeTargetSnapshotMissing { + t.Fatalf("err=%v want code %s", publishErr, ErrorCodeTargetSnapshotMissing) + } + if len(guard.checked) != 1 || guard.checked[0] != "kb-production" { + t.Fatalf("guard must be asked about the production target: %v", guard.checked) + } + if len(calls) != 0 { + t.Fatalf("a refused release must not touch the target: %v", calls) + } +} + +// 门禁未接线（target_snapshot.root 未配置）时，映射项也要拒绝——不能静默放行。 +func TestSidecarPublisherFailsClosedWhenSnapshotGateUnconfigured(t *testing.T) { + calls := []string{} + manifest, err := releasebundle.Seal(testManifest()) + if err != nil { + t.Fatal(err) + } + exporter := &fakeExporter{manifest: manifest, calls: &calls} + target := &fakeSidecarTarget{contract: testTargetContract(), calls: &calls, prepared: PreparedChange{PreviousRevision: "prod-v0", ProductionRevision: "prod-v1"}} + publisher, err := NewSidecarPublisher(exporter, target) + if err != nil { + t.Fatal(err) + } + _, publishErr := publisher.Publish(context.Background(), mappedPlanItem()) + var itemErr *ItemError + if !errors.As(publishErr, &itemErr) || itemErr.Code != ErrorCodeTargetSnapshotMissing { + t.Fatalf("err=%v want code %s", publishErr, ErrorCodeTargetSnapshotMissing) + } + if len(calls) != 0 { + t.Fatalf("calls=%v", calls) + } +} + +// 同 ID 目标（历史语义 / 本地演练）不需要快照，策略为 mapped 时门禁不介入。 +func TestSidecarPublisherSkipsSnapshotGateForSameBaseTarget(t *testing.T) { + calls := []string{} + manifest, err := releasebundle.Seal(testManifest()) + if err != nil { + t.Fatal(err) + } + exporter := &fakeExporter{manifest: manifest, calls: &calls} + target := &fakeSidecarTarget{contract: testTargetContract(), calls: &calls, prepared: PreparedChange{PreviousRevision: "prod-v0", ProductionRevision: "prod-v1"}} + guard := &stubSnapshotGuard{err: errors.New("must not be consulted")} + publisher, err := NewSidecarPublisher(exporter, target, WithTargetSnapshots(guard, SnapshotPolicyMapped)) + if err != nil { + t.Fatal(err) + } + if _, err := publisher.Publish(context.Background(), testPlanItem()); err != nil { + t.Fatalf("same-base target must publish without a snapshot: %v", err) + } + if len(guard.checked) != 0 { + t.Fatalf("guard must not run for same-base targets: %v", guard.checked) + } +} + +// policy=always 时，同 ID 目标同样要求快照。 +func TestSidecarPublisherAlwaysPolicyChecksSameBaseTarget(t *testing.T) { + calls := []string{} + manifest, err := releasebundle.Seal(testManifest()) + if err != nil { + t.Fatal(err) + } + exporter := &fakeExporter{manifest: manifest, calls: &calls} + target := &fakeSidecarTarget{contract: testTargetContract(), calls: &calls, prepared: PreparedChange{PreviousRevision: "prod-v0", ProductionRevision: "prod-v1"}} + guard := &stubSnapshotGuard{} + publisher, err := NewSidecarPublisher(exporter, target, WithTargetSnapshots(guard, SnapshotPolicyAlways)) + if err != nil { + t.Fatal(err) + } + if _, err := publisher.Publish(context.Background(), testPlanItem()); err != nil { + t.Fatal(err) + } + if len(guard.checked) != 1 || guard.checked[0] != "kb-safe" { + t.Fatalf("always policy must check every target: %v", guard.checked) + } +} + +// 撤权（delete）与回滚同样先过门禁。 +func TestSidecarPublisherGatesDeleteAndRollbackToo(t *testing.T) { + for _, operation := range []string{"delete", "rollback"} { + calls := []string{} + manifest, err := releasebundle.Seal(testManifest()) + if err != nil { + t.Fatal(err) + } + exporter := &fakeExporter{manifest: manifest, calls: &calls} + target := &fakeSidecarTarget{contract: testTargetContract(), calls: &calls, prepared: PreparedChange{PreviousRevision: "prod-v0", ProductionRevision: "prod-v1"}} + guard := &stubSnapshotGuard{err: errors.New("stale")} + publisher, err := NewSidecarPublisher(exporter, target, WithTargetSnapshots(guard, SnapshotPolicyMapped)) + if err != nil { + t.Fatal(err) + } + item := mappedPlanItem() + item.Action = operation + var publishErr error + if operation == "delete" { + _, publishErr = publisher.Delete(context.Background(), item) + } else { + _, publishErr = publisher.Rollback(context.Background(), item) + } + var itemErr *ItemError + if !errors.As(publishErr, &itemErr) || itemErr.Code != ErrorCodeTargetSnapshotMissing { + t.Fatalf("%s: err=%v want %s", operation, publishErr, ErrorCodeTargetSnapshotMissing) + } + if len(calls) != 0 { + t.Fatalf("%s touched the target: %v", operation, calls) + } + } +} diff --git a/scripts/oss_bucket_settings.py b/scripts/oss_bucket_settings.py new file mode 100644 index 0000000..9c539d7 --- /dev/null +++ b/scripts/oss_bucket_settings.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 +"""目标 OSS 桶的保护设置：读取状态，或开启版本控制 / 跨区域复制检查。 + +只处理目标环境的桶（secrets/release-sim 下的 key 文件），凭据不打印。 + +用法： + python3 scripts/oss_bucket_settings.py status # 版本控制 / 复制 / 生命周期 + python3 scripts/oss_bucket_settings.py enable-versioning # 写操作：开启版本控制 +""" +import base64 +import hashlib +import hmac +import pathlib +import sys +import urllib.error +import urllib.request +from email.utils import formatdate + +ROOT = pathlib.Path(__file__).resolve().parents[1] +SECRETS = ROOT / "secrets" / "release-sim" +BUCKET = "dobot-weknora-test" +ENDPOINT = "oss-cn-shenzhen.aliyuncs.com" + + +def read_secret(name): + return (SECRETS / name).read_text().strip() + + +def request(method, subresource, body=""): + access_key = read_secret("target_object_access_key") + secret_key = read_secret("target_object_secret_key") + date = formatdate(usegmt=True) + content_type = "application/xml" if body else "" + canonical = f"/{BUCKET}/?{subresource}" + string_to_sign = f"{method}\n\n{content_type}\n{date}\n{canonical}" + signature = base64.b64encode( + hmac.new(secret_key.encode(), string_to_sign.encode(), hashlib.sha1).digest() + ).decode() + headers = {"Date": date, "Authorization": f"OSS {access_key}:{signature}"} + if content_type: + headers["Content-Type"] = content_type + http_request = urllib.request.Request( + f"https://{BUCKET}.{ENDPOINT}/?{subresource}", + data=body.encode() if body else None, headers=headers, method=method) + try: + with urllib.request.urlopen(http_request, timeout=20) as response: + return response.status, response.read().decode("utf-8", "replace") + except urllib.error.HTTPError as error: + return error.code, error.read().decode("utf-8", "replace") + + +def versioning_state(): + status, body = request("GET", "versioning") + if status != 200: + return f"HTTP {status}" + return "Enabled" if "Enabled" in body else "Disabled（未开启）" + + +def main(): + action = sys.argv[1] if len(sys.argv) > 1 else "status" + if action == "status": + print("versioning:", versioning_state()) + for name in ("replication", "lifecycle"): + status, body = request("GET", name) + summary = "none" if status == 404 else " ".join(body.split())[:120] + print(f"{name}: HTTP {status} {summary}") + return + if action == "enable-versioning": + print("before:", versioning_state()) + body = "Enabled" + status, response = request("PUT", "versioning", body) + print("PUT versioning:", f"HTTP {status}", " ".join(response.split())[:160]) + print("after:", versioning_state()) + return + print("unknown action:", action, file=sys.stderr) + sys.exit(2) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/scripts/target_kb_snapshot.sh b/scripts/target_kb_snapshot.sh index 4c24401..5ea8fc0 100755 --- a/scripts/target_kb_snapshot.sh +++ b/scripts/target_kb_snapshot.sh @@ -5,32 +5,47 @@ # 只做 SELECT + \copy，不写目标库任何一行。 # # 用法： -# bash scripts/target_kb_snapshot.sh # 用 production_kb_bindings 里已绑定的目标库 -# bash scripts/target_kb_snapshot.sh [] # 指定目标库 +# bash scripts/target_kb_snapshot.sh # 已绑定的目标库；云内执行（快） +# EXECUTION=tunnel bash scripts/target_kb_snapshot.sh # 走本机 SSH 隧道（慢，兜底） +# bash scripts/target_kb_snapshot.sh [] # 指定目标库 +# INCLUDE_CHUNKS=1 bash scripts/target_kb_snapshot.sh # 连 chunks 一起导（大，按需） +# PULL_CHUNKS=1 bash scripts/target_kb_snapshot.sh # 云内执行时把 chunks 结果也拉回本机 # # 环境变量： +# EXECUTION cloud（默认，与 DB 同 VPC 的云主机上跑）| tunnel（本机经 SSH 隧道） # SNAPSHOT_ROOT 输出根目录（默认 backups/target-kb-snapshots） -# TUNNEL_HOST 目标 PG 隧道地址（默认 172.17.0.1） -# TUNNEL_PORT 目标 PG 隧道端口（默认 25432） -# INCLUDE_CHUNKS 1=连 chunks 一起导出（较大，默认 1） +# INCLUDE_CHUNKS 1=连 chunks 一起导（默认 0：秒级完成；向量在 ES 侧，行可从源端重发） +# PULL_CHUNKS 1=云内执行时把 chunks 结果也拉回本机（默认 0，留在云主机） +# TUNNEL_HOST/TUNNEL_PORT 隧道模式的目标 PG 地址（默认 172.17.0.1:25432） +# SSH_CONFIG SSH 配置路径（默认 ~/.ssh/config.weknora-cloud-test） +# SSH_HOST 云主机别名（默认 weknora-test） +# APP_CONTAINER 云上提供 DB 凭据的容器（默认 weknora-test-v071-app-1） +# REMOTE_ROOT 云上快照目录（默认 /opt/weknora-test/kb-snapshots） # WITH_ES 1=同时记录目标 ES 的文档数（默认 1） # NOTE 附加到清单的备注（例如发布单号） # # 覆盖范围（PG 行级）：knowledge_bases / knowledges / knowledge_tags / -# knowledge_tag_relations / chunks / resources / resource_bindings / -# wiki_pages / wiki_folders（含软删行，不含表结构）。 +# knowledge_tag_relations / wiki_pages / wiki_folders / resources / +# resource_bindings（+ chunks，可选）。含软删行，不含表结构。 # -# 不覆盖：Elasticsearch 文档内容与向量、OSS 对象本体（只在 resources.jsonl 里 -# 记路径/大小/哈希做清单）、表结构与索引。整库级恢复请用阿里云 RDS 快照/PITR。 +# 不覆盖：Elasticsearch 文档与向量（见 ES 仓库 kb_backup 与 SLM 策略 kb-hourly/kb-daily）、 +# OSS 对象本体（只在 resources.jsonl 里记路径/大小/哈希做清单）、表结构与索引。 +# 整库级恢复请用阿里云 RDS 快照/PITR。 set -euo pipefail WORKDIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" cd "$WORKDIR" +EXECUTION="${EXECUTION:-cloud}" SNAPSHOT_ROOT="${SNAPSHOT_ROOT:-$WORKDIR/backups/target-kb-snapshots}" TUNNEL_HOST="${TUNNEL_HOST:-172.17.0.1}" TUNNEL_PORT="${TUNNEL_PORT:-25432}" -INCLUDE_CHUNKS="${INCLUDE_CHUNKS:-1}" +INCLUDE_CHUNKS="${INCLUDE_CHUNKS:-0}" +PULL_CHUNKS="${PULL_CHUNKS:-0}" +SSH_CONFIG="${SSH_CONFIG:-$HOME/.ssh/config.weknora-cloud-test}" +SSH_HOST="${SSH_HOST:-weknora-test}" +APP_CONTAINER="${APP_CONTAINER:-weknora-test-v071-app-1}" +REMOTE_ROOT="${REMOTE_ROOT:-/opt/weknora-test/kb-snapshots}" WITH_ES="${WITH_ES:-1}" NOTE="${NOTE:-}" @@ -53,34 +68,15 @@ done STAMP="$(date -u +%Y%m%dT%H%M%SZ)" PREFIX="$(printf '%s' "${KB_IDS[0]}" | cut -c1-8)" -OUT="$SNAPSHOT_ROOT/${STAMP}-${PREFIX}${PREFIX:+_}" -OUT="${OUT%_}" +OUT="$SNAPSHOT_ROOT/${STAMP}-${PREFIX}" mkdir -p "$OUT" -# ---------- 连接串（不打印） ---------- -SECRET_FILE="secrets/release-sim/target_database_url" -[ -f "$SECRET_FILE" ] || { echo "missing $SECRET_FILE"; exit 1; } -PGURL="$(python3 - "$SECRET_FILE" "$TUNNEL_HOST" "$TUNNEL_PORT" <<'PY' -import re, sys, pathlib -url = pathlib.Path(sys.argv[1]).read_text().strip() -url = re.sub(r'@[^/@]+:\d+/', '@%s:%s/' % (sys.argv[2], sys.argv[3]), url) -print(url) -PY -)" -ENDPOINT="$TUNNEL_HOST:$TUNNEL_PORT ($(python3 - "$SECRET_FILE" <<'PY' -import sys, pathlib, re, urllib.parse -url = pathlib.Path(sys.argv[1]).read_text().strip() -parsed = urllib.parse.urlparse(url) -print("host=%s db=%s" % (parsed.hostname or "?", (parsed.path or "/?").lstrip("/") or "?")) -PY -))" - -# ---------- 生成导出脚本（本身也是留档的一部分） ---------- KB_SQL="'$(printf '%s' "${KB_IDS[*]}" | sed "s/ /','/g")'" +# ---------- 生成导出脚本（本身也是留档的一部分） ---------- cat >"$OUT/export.sql" <"$OUT/export.sql" <>"$OUT/export.sql" </dev/null +rm -f "$REMOTE_DIR/export.sql" +if [ "$INCLUDE_CHUNKS" = "1" ] && [ -f "$REMOTE_DIR/chunks.jsonl" ]; then + gzip -f "$REMOTE_DIR/chunks.jsonl" +fi +python3 - "$REMOTE_DIR" <<'PY' +import hashlib, json, pathlib, sys +out = pathlib.Path(sys.argv[1]); files = [] +for path in sorted(out.glob("*.jsonl*")): + digest = hashlib.sha256(); size = 0 + with path.open("rb") as handle: + for block in iter(lambda: handle.read(1 << 20), b""): + digest.update(block); size += len(block) + files.append({"file": path.name, "bytes": size, "sha256": digest.hexdigest()}) +(out / "REMOTE_MANIFEST.json").write_text(json.dumps({"host_files": files}, indent=2) + "\n") +print("remote files:", " ".join(f"{f['file']}={f['bytes']}" for f in files)) +PY +exit 0 +REMOTE + # 小表拉回本机；chunks 默认留在云主机（PULL_CHUNKS=1 才拉） + scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/*.jsonl" "$OUT/" 2>/dev/null || true + scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/REMOTE_MANIFEST.json" "$OUT/" 2>/dev/null || true + if [ "$PULL_CHUNKS" = "1" ]; then + scp -q -F "$SSH_CONFIG" "$SSH_HOST:$REMOTE_DIR/chunks.jsonl.gz" "$OUT/" 2>/dev/null || true + fi +else + # ---------- 隧道模式（兜底）：本机经 SSH 隧道直连 ---------- + SECRET_FILE="secrets/release-sim/target_database_url" + [ -f "$SECRET_FILE" ] || { echo "missing $SECRET_FILE"; exit 1; } + PGURL="$(python3 - "$SECRET_FILE" "$TUNNEL_HOST" "$TUNNEL_PORT" <<'PY' +import re, sys, pathlib +url = pathlib.Path(sys.argv[1]).read_text().strip() +print(re.sub(r'@[^/@]+:\d+/', '@%s:%s/' % (sys.argv[2], sys.argv[3]), url)) +PY +)" + docker run --rm --network host -e PGURL="$PGURL" \ + -v "$OUT:/out" -v "$OUT/export.sql:/scripts/export.sql:ro" postgres:16 \ + sh -c 'psql "$PGURL" -q -f /scripts/export.sql' >/dev/null +fi # ---------- ES 文档数（只读，失败不阻塞） ---------- ES_SUMMARY="" if [ "$WITH_ES" = "1" ]; then - ES_URL="http://${TUNNEL_HOST}:29200" for id in "${KB_IDS[@]}"; do count="$(curl -s --max-time 15 -H 'Content-Type: application/json' \ - -X POST "$ES_URL/weknora_test_chunks/_count" \ - -d "{\"query\":{\"term\":{\"knowledge_base_id\":\"$id\"}}}" \ + -X POST "http://${TUNNEL_HOST}:29200/weknora_test_chunks/_count" \ + -d "{\"query\":{\"term\":{\"knowledge_base_id.keyword\":\"$id\"}}}" \ | python3 -c 'import json,sys; print(json.load(sys.stdin).get("count","?"))' 2>/dev/null || echo "?")" ES_SUMMARY="${ES_SUMMARY}${id}=${count} " done fi # ---------- 清单 + 说明 ---------- -python3 - "$OUT" "$STAMP" "$ENDPOINT" "$INCLUDE_CHUNKS" "$ES_SUMMARY" "$NOTE" "${KB_IDS[@]}" <<'PY' +python3 - "$OUT" "$STAMP" "$EXECUTION" "$INCLUDE_CHUNKS" "$ES_SUMMARY" "$NOTE" "$REMOTE_DIR" "${KB_IDS[@]}" <<'PY' import hashlib, json, pathlib, sys -out = pathlib.Path(sys.argv[1]); stamp = sys.argv[2]; endpoint = sys.argv[3] +out = pathlib.Path(sys.argv[1]); stamp, execution = sys.argv[2], sys.argv[3] include_chunks = sys.argv[4] == "1"; es_summary = sys.argv[5].strip(); note = sys.argv[6] -kb_ids = sys.argv[7:] +remote_dir = sys.argv[7]; kb_ids = sys.argv[8:] files = [] for path in sorted(out.glob("*.jsonl")): digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1 << 20), b""): digest.update(block) - rows = sum(1 for _ in path.open("rb")) - files.append({"file": path.name, "rows": rows, "bytes": path.stat().st_size, - "sha256": digest.hexdigest()}) + files.append({"file": path.name, "rows": sum(1 for _ in path.open("rb")), + "bytes": path.stat().st_size, "sha256": digest.hexdigest(), "location": "local"}) +for path in sorted(out.glob("chunks.jsonl.gz")): + digest = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(1 << 20), b""): + digest.update(block) + files.append({"file": path.name, "bytes": path.stat().st_size, + "sha256": digest.hexdigest(), "location": "local"}) +remote_manifest = out / "REMOTE_MANIFEST.json" +if remote_manifest.exists(): + for item in json.loads(remote_manifest.read_text()).get("host_files", []): + if not any(existing["file"] == item["file"] for existing in files): + item["location"] = f"cloud:{remote_dir}" + files.append(item) manifest = { "kind": "weknora-target-kb-snapshot", "created_at": stamp, - "target_endpoint": endpoint, + "execution": execution, "knowledge_base_ids": kb_ids, - "profile": "kb-rows" + ("" if include_chunks else " (chunks excluded)"), + "profile": "kb-rows" + ("" if include_chunks else " (chunks excluded: 向量在 ES 侧，行可从源端重发)"), "covers": ["knowledge_bases", "knowledges", "knowledge_tags", "knowledge_tag_relations", "wiki_pages", "wiki_folders", "resource_bindings", "resources"] + (["chunks"] if include_chunks else []), - "does_not_cover": ["Elasticsearch documents and vectors", "OSS object bodies (listed by reference in resources.jsonl)", - "table structure / indexes", "rows outside the listed knowledge bases"], + "does_not_cover": ["Elasticsearch documents and vectors (see ES repository kb_backup, SLM kb-hourly/kb-daily)", + "OSS object bodies (listed by reference in resources.jsonl)", + "table structure / indexes", "rows outside the listed knowledge bases", + "whole-database recovery (use Aliyun RDS snapshot / PITR)"], "es_document_counts": es_summary or None, + "remote_directory": remote_dir or None, "note": note or None, "files": files, } (out / "MANIFEST.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n") -summary = " ".join(f"{item['file']}={item['rows']}" for item in files) print("snapshot:", out) -print("files:", summary) -print("bytes:", sum(item["bytes"] for item in files)) -if es_summary: - print("es docs:", es_summary) +print("files:", " ".join(f"{item['file']}={item.get('rows', '-')}" for item in files)) +print("local bytes:", sum(item["bytes"] for item in files if item.get("location") == "local")) +print("es docs:", es_summary or "-") PY cat >"$OUT/README.md" <<'MD' @@ -163,12 +230,12 @@ cat >"$OUT/README.md" <<'MD' ## 它是什么 发布 / 撤权 / 回滚**之前**的目标库现场：KB 维度的 PG 行（含软删行），每张表一个 -JSONL（`to_jsonb(row)` 一行一条），附 `MANIFEST.json`（行数 / 字节 / sha256 / 覆盖范围） -与 `export.sql`（导出用的原始查询）。生成过程只有 SELECT + `\copy`，不修改目标库。 +JSONL（`to_jsonb(row)` 一行一条），附 `MANIFEST.json`（行数 / 字节 / sha256 / 覆盖范围 / +位置）与 `export.sql`（导出用的原始查询）。生成过程只有 SELECT + `\copy`，不修改目标库。 ## 它不是什么 -- 不含 Elasticsearch 文档与向量、不含 OSS 对象本体（`resources.jsonl` 里只有 - 路径 / 大小 / 哈希清单）、不含表结构与索引。 +- 不含 Elasticsearch 文档与向量（见 ES 仓库 `kb_backup` 与 SLM 策略 `kb-hourly`/`kb-daily`）、 + 不含 OSS 对象本体（`resources.jsonl` 里只有路径 / 大小 / 哈希清单）、不含表结构与索引。 - 不是整库级灾备。**整库级恢复请用阿里云 RDS 的快照 / 按时间点恢复（PITR）**： 恢复到临时实例后，再按 KB 导出需要的行，这样连索引和约束一起回到某个时间点。 @@ -176,12 +243,11 @@ JSONL（`to_jsonb(row)` 一行一条），附 `MANIFEST.json`（行数 / 字节 先确认目标库当前状态已被另一份快照覆盖（避免二次覆盖），然后按外键顺序灌回： ```bash -URL=<目标库连接串> # 例如经隧道：postgres://...@172.17.0.1:25432/ +URL=<目标库连接串> OUT=<本快照目录> docker run --rm --network host -e PGURL="$URL" -v "$OUT:/snap:ro" postgres:16 \ psql "$PGURL" -v ON_ERROR_STOP=1 <<'SQL' BEGIN; --- 每张表：把 JSONL 灌进临时表，再按类型还原成行 DO $$ DECLARE t text; BEGIN diff --git a/web/src/App.tsx b/web/src/App.tsx index 6ec3679..8130348 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -767,6 +767,7 @@ function releaseFailureMessage(code: string) { release_retryable: '遇到临时故障，系统将自动重试', release_manifest_mismatch: '发布结果与冻结的发布清单不一致，已降级为失败', release_rollback_failed: '自动回滚失败，需整体快照恢复（等待授权）', + release_target_snapshot_missing: '缺少目标知识库的发布前快照留档，已拒绝发布（先跑 scripts/target_kb_snapshot.sh）', } return labels[code] ?? '执行失败，请联系管理员查看技术记录' }