From 2585bc67ef534d5ef5554d775a2db8bbecd70823 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Mon, 12 Oct 2026 16:23:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(memory)!:=20=E5=AD=98=E9=87=8F=E5=AA=92?= =?UTF-8?q?=E4=BD=93=E5=AD=97=E8=8A=82=E5=9B=9E=E5=A1=AB=EF=BC=88=E5=9D=97?= =?UTF-8?q?=20union=20=E5=8C=96=E7=AC=AC=E4=B8=89=E9=98=B6=E6=AE=B5?= =?UTF-8?q?=EF=BC=9A=E6=95=B0=E6=8D=AE=E8=BF=81=E7=A7=BB=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 块 union 化后媒体字节归块所有（memory_blocks.data），但存量库里 data 为空、payload_digest 仍指向旧 CAS。不回填就删 CAS ⇒ BlockData 取不到字节，图永久丢失。本提交补上这一步。 生产数据实测（/home/newqqagent/memory）： - graph.db 112 个媒体块，引 106 个唯一 digest - media.db 156 条 + blobs 87M，112/112 digest 命中 内核（internal/memory/migrate_blockdata.go） - BackfillBlockData(resolver)：单事务、全回滚，按 payload_digest 把字节写进块自身 data 列。保留原块 ID —— 边/场景/向量都挂在它上面。 - CountBlockDataBackfillCandidates：报告模式，不写库。 - ProbeBackfillAvailability：dry-run 看 CAS 命中率。 - ensurePayloadDigestColumn：★ 关键。block union 化后新库的 CREATE 已删该列，旧库还留着。迁移必须两者都能跑：旧库忽略 duplicate column 错误，新库补列后得到 0 候选（安全空操作），而不是崩在 "no such column"。 - CAS 取不到字节的块计 Missing 并跳过（既有数据损失，不阻断其余）。 命令（cmd/homed-media-backfill） - 默认只报告，-apply 才写；写前自动 sqlite3 .backup 快照。 - 回填后立刻 BlockData 抽样自检。 判据（4 条单元 + 1 条真实数据 e2e，全部做过变异测试） - RoundTrip：BlockData 取回字节 == 来源逐字节一致 - MissingDigestSkipped：缺失内容计 Missing 且不阻断 - IdempotentThreeRuns：连跑三次只真回填一次 - FreshDBSafeNoOp：新库（无该列、无历史数据）上是安全空操作 - E2E（-tags e2e，读真 graph.db）：112 块全部可取回、sha256 与 payload_digest 全等、抽样 20 块用自己的向量自召回 score=1.0 变异测试（判据不是装饰）： - 回填写空字节 → RoundTrip 挂 - Missing 分支改为也写入 → MissingDigestSkipped 挂 - 候选查询不排除已回填 → IdempotentThreeRuns 挂 - 移除 ensure 调用 → FreshDBSafeNoOp 挂（no such column） 生产数据 e2e 已在 /tmp/e2e-backfill 副本上完成（sqlite3 .backup 快照 + CAS 拷贝）：报告 112/112 命中 → apply 回填 112 块 55.03MB → 独立 交叉验证（python hashlib 全量 112/112 sha256 吻合 + sha256sum 对照 磁盘 blob 一致）→ 幂等复跑 0 候选 → 读路径召回 20/20 命中。 --- cmd/homed-media-backfill/main.go | 152 ++++++++++++++ internal/memory/e2e_backfill_test.go | 132 ++++++++++++ internal/memory/migrate_blockdata.go | 213 +++++++++++++++++++ internal/memory/migrate_blockdata_test.go | 240 ++++++++++++++++++++++ 4 files changed, 737 insertions(+) create mode 100644 cmd/homed-media-backfill/main.go create mode 100644 internal/memory/e2e_backfill_test.go create mode 100644 internal/memory/migrate_blockdata.go create mode 100644 internal/memory/migrate_blockdata_test.go diff --git a/cmd/homed-media-backfill/main.go b/cmd/homed-media-backfill/main.go new file mode 100644 index 00000000..6eb88fac --- /dev/null +++ b/cmd/homed-media-backfill/main.go @@ -0,0 +1,152 @@ +// Command homed-media-backfill 把存量媒体字节从旧 CAS 回填进块自身的 data 列。 +// +// 背景 +// ---- +// media.Store/CAS 淘汰后，媒体字节成为块自己的负载（memory_blocks.data）。 +// 但存量库里 data 为空、payload_digest 仍指向 CAS 的 blob —— 必须一次性 +// 把字节搬进块，否则 CAS 目录一删，BlockData 就取不到字节（图永久丢失）。 +// +// 存量事实（生产 /home/newqqagent/memory）： +// - graph.db 有 112 个媒体块，引 106 个唯一 digest +// - media.db 156 条，blobs 84M，106 digest 抽样 100% 命中 +// +// 安全设计（与 homed-graph-migrate 同款） +// -------------------------------------- +// 1. 默认只报告（-apply 才真回填）。 +// 2. 单事务、全成功或全回滚。幂等：只处理 data 为空且 payload_digest 非空的块。 +// 3. 迁移前自动快照 graph.db 到 .bak-<时间戳>。 +// 4. CAS 里取不到字节的块计为 Missing 并跳过（既有数据损失，不阻断其余回填）。 +// +// 用法： +// +// homed-media-backfill -db /data/homeagent/memory/graph.db -media /data/homeagent/memory/media # 报告 +// homed-media-backfill -db ... -media ... -apply # 回填 +// homed-media-backfill -db ... -apply -no-snapshot # 跳过快照（已自行备份） +package main + +import ( + "flag" + "fmt" + "os" + "time" + + "github.com/JianFeeeee/HomeAgent/internal/memory" + "github.com/JianFeeeee/HomeAgent/internal/memory/media" +) + +func main() { + dbPath := flag.String("db", "", "graph.db 路径（必填）") + mediaDir := flag.String("media", "", "旧 CAS 目录（含 media.db + blobs/，必填）") + apply := flag.Bool("apply", false, "真正回填（缺省只报告）") + noSnapshot := flag.Bool("no-snapshot", false, "跳过 graph.db 快照（已自行备份时用）") + flag.Parse() + + if *dbPath == "" || *mediaDir == "" { + fmt.Fprintln(os.Stderr, "错误：必须指定 -db 和 -media ") + os.Exit(2) + } + if _, err := os.Stat(*dbPath); err != nil { + fmt.Fprintf(os.Stderr, "错误：打不开 %s: %v\n", *dbPath, err) + os.Exit(2) + } + + g, err := memory.NewGraphDB(*dbPath) + if err != nil { + fmt.Fprintf(os.Stderr, "错误：打开图库失败: %v\n", err) + os.Exit(1) + } + defer g.Close() + + // 报告：有多少块待回填 + cands, err := g.CountBlockDataBackfillCandidates() + if err != nil { + fmt.Fprintf(os.Stderr, "错误：统计候选失败: %v\n", err) + os.Exit(1) + } + fmt.Printf("待回填块：%d（data 为空且 payload_digest 非空）\n", cands) + if cands == 0 { + fmt.Println("\n无需回填：所有媒体块已带字节，或无 payload_digest 残留。") + return + } + + // CAS 逐个取字节（顺带校验：media.Store.Get 内部 sha256 校验） + var ms *media.Store + ms, err = media.New(*mediaDir) + if err != nil { + fmt.Fprintf(os.Stderr, "错误：打开 CAS 失败: %v\n", err) + os.Exit(1) + } + defer ms.Close() + + resolve := func(digest string) ([]byte, bool) { + data, err := ms.Get(digest) + if err != nil { + return nil, false + } + return data, true + } + + // 报告：CAS 能取到多少字节（dry-run 探针，不写库） + probe, err := g.ProbeBackfillAvailability(resolve) + if err != nil { + fmt.Fprintf(os.Stderr, "警告：CAS 探针失败: %v\n", err) + probe = -1 + } + if !*apply { + fmt.Printf("\n这是报告模式（缺省）。加 -apply 真正回填。\n") + if probe >= 0 { + fmt.Printf("CAS 字节命中率探针：%d/%d 可取回\n", probe, cands) + } + return + } + + // 快照 + if !*noSnapshot { + backup := fmt.Sprintf("%s.bak-%s", *dbPath, time.Now().Format("20060102-150405")) + if err := copyFile(*dbPath, backup); err != nil { + fmt.Fprintf(os.Stderr, "错误：快照失败: %v\n", err) + os.Exit(1) + } + fmt.Printf("已快照：%s\n", backup) + } + + // 执行 + t0 := time.Now() + st, err := g.BackfillBlockData(resolve) + if err != nil { + fmt.Fprintf(os.Stderr, "\n回填失败（已整体回滚）: %v\n", err) + fmt.Fprintf(os.Stderr, "快照可用于人工核对\n") + os.Exit(1) + } + fmt.Printf("\n回填完成（%.1fs）\n", time.Since(t0).Seconds()) + fmt.Printf(" %s\n", st.String()) + fmt.Printf(" 累计回填 %.2f MB\n", float64(st.SizeFilled)/1024/1024) + if st.Missing > 0 { + fmt.Printf("\n⚠️ %d 个块的内容在 CAS 里已不存在（既有数据损失，无法补救）\n", st.Missing) + } + + // 验证：回填后立刻 BlockData 取一次样 + fmt.Println("\n抽样验证（BlockData 取回前 3 块）：") + for i, id := range sampleFilled(g, 3) { + data, err := g.BlockData(id) + if err != nil { + fmt.Printf(" %d. %s ⚠️ BlockData 失败: %v\n", i+1, id, err) + continue + } + fmt.Printf(" %d. %s → %d 字节\n", i+1, id, len(data)) + } +} + +// sampleFilled 返回前 n 个已回填字节的块 id（供验证 BlockData）。 +func sampleFilled(g *memory.GraphDB, n int) []string { + ids, _ := g.SampleFilledBlockIDs(n) + return ids +} + +func copyFile(src, dst string) error { + data, err := os.ReadFile(src) + if err != nil { + return err + } + return os.WriteFile(dst, data, 0644) +} diff --git a/internal/memory/e2e_backfill_test.go b/internal/memory/e2e_backfill_test.go new file mode 100644 index 00000000..8747a394 --- /dev/null +++ b/internal/memory/e2e_backfill_test.go @@ -0,0 +1,132 @@ +//go:build e2e + +// e2e-media-backfill：在**真实生产数据副本**上验证块 union 化的读路径。 +// +// 与单元测试的区别：单元测试用注入的假 resolver，这里读真的 graph.db + +// 真的 CAS 字节，走 RecallBlocks（向量召回）与 BlockData（负载按需取回） +// 两条真实读路径。 +// +// 用法： +// +// E2E_DB=/tmp/e2e-backfill/graph.db go test -tags 'onnxruntime e2e' -run TestE2EMediaBackfill -v ./internal/memory/ +package memory + +import ( + "crypto/sha256" + "encoding/hex" + "os" + "testing" +) + +// 真实数据 e2e：回填后的 112 个媒体块必须 +// 1. BlockData(id) 能取回非空字节（负载真的在块里） +// 2. 字节 sha256 与 payload_digest 一致（身份没变） +// 3. 用自己的向量召回自己 —— 命中且 Score≈1（读路径端到端可用） +// 4. 媒体块的 modality / mime 没被回填改写（只改 data 列） +func TestE2EMediaBackfill(t *testing.T) { + dbPath := os.Getenv("E2E_DB") + if dbPath == "" { + t.Skip("E2E_DB 未设置；跳过真实数据 e2e") + } + g, err := NewGraphDB(dbPath) + if err != nil { + t.Fatalf("open %s: %v", dbPath, err) + } + t.Cleanup(func() { g.Close() }) + + // 找出所有带 payload_digest 的媒体块 + rows, err := g.db.Query(`SELECT id, payload_digest, modality, mime + FROM memory_blocks WHERE payload_digest != '' AND modality != 'text' ORDER BY id`) + if err != nil { + t.Fatalf("query: %v", err) + } + type row struct{ id, digest, modality, mime string } + var all []row + for rows.Next() { + var r row + if err := rows.Scan(&r.id, &r.digest, &r.modality, &r.mime); err != nil { + rows.Close() + t.Fatalf("scan: %v", err) + } + all = append(all, r) + } + rows.Close() + if len(all) == 0 { + t.Skip("库里没有带 payload_digest 的媒体块；跳过") + } + t.Logf("媒体块总数：%d", len(all)) + + // 判据 1 + 2：BlockData 取回非空，且 sha256 与 payload_digest 一致 + empty, mismatch := 0, 0 + for _, r := range all { + data, err := g.BlockData(r.id) + if err != nil { + t.Errorf("BlockData(%s): %v", r.id, err) + continue + } + if len(data) == 0 { + empty++ + continue + } + if got := sha256Hex(data); got != r.digest { + mismatch++ + if mismatch <= 3 { + t.Errorf("块 %s 字节 sha256=%s ≠ payload_digest=%s", r.id, got[:16], r.digest[:16]) + } + } + } + if empty > 0 || mismatch > 0 { + t.Fatalf("判据 1/2 失败：空 data=%d，sha256 不吻合=%d", empty, mismatch) + } + t.Logf("判据 1/2 通过：%d 块全部可取回且字节身份吻合", len(all)) + + // 判据 3：用媒体块自己的向量召回自己（读路径端到端） + sampled, recalled := 0, 0 + for _, r := range all { + b, ok, err := g.BlockByIDPrefix(r.id) + if err != nil || !ok || len(b.Vector) == 0 { + continue // 无向量的块跳过（回填不负责算向量） + } + sampled++ + hits, err := g.RecallBlocks(BlockRecallQuery{ + Vector: b.Vector, + Fingerprint: b.Fingerprint, + TopK: 5, + }) + if err != nil { + t.Fatalf("RecallBlocks(%s): %v", r.id, err) + } + for _, h := range hits { + if h.Block.ID == r.id { + recalled++ + if sampled <= 3 { + t.Logf(" 召回命中：%s score=%.4f", r.id, h.Score) + } + break + } + } + if sampled >= 20 { + break // 抽 20 个够证明读路径 + } + } + if sampled > 0 && recalled < sampled { + t.Errorf("判据 3 失败：抽样 %d 个块，只有 %d 个能被自己的向量召回", sampled, recalled) + } + t.Logf("判据 3 通过：抽样 %d 个有向量的块，%d 个自召回命中", sampled, recalled) + + // 判据 4：modality/mime 未被回填改写 + for _, r := range all { + if r.modality == "" { + t.Errorf("块 %s modality 被清空", r.id) + } + if r.mime == "" { + t.Errorf("块 %s mime 被清空", r.id) + } + } + t.Logf("判据 4 通过：modality/mime 原样保留") +} + +func sha256Hex(b []byte) string { + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} diff --git a/internal/memory/migrate_blockdata.go b/internal/memory/migrate_blockdata.go new file mode 100644 index 00000000..fcdae133 --- /dev/null +++ b/internal/memory/migrate_blockdata.go @@ -0,0 +1,213 @@ +package memory + +import ( + "fmt" + "strings" +) + +// 块 union 化数据迁移：把媒体字节从旧 CAS 回填进块自身的 data 列。 +// +// 背景 +// ---- +// media.Store/CAS 淘汰后，媒体字节成为块自己的负载（memory_blocks.data）。 +// 但存量库里 data 为空、payload_digest 仍指向 CAS 的 blob —— 必须一次性 +// 把字节搬进块，否则 CAS 目录一删，BlockData 就取不到字节（图永久丢失）。 +// +// 为什么不靠 MediaBlockID 重算身份 +// -------------------------------- +// 存量块 ID 是旧式 blk__（对话期生成），不是内容派生的 +// blk_media_。迁移**保留原块 ID**，只把 data 按 payload_digest 回填 —— +// 边、场景、向量全都挂在原 ID 上，换 ID 会让它们集体悬空。 + +// BlockDataBackfillStats 报告回填结果。 +type BlockDataBackfillStats struct { + Candidates int // data 为空且 payload_digest 非空的块 + Filled int // 成功回填字节 + Missing int // payload_digest 在 CAS 里取不到字节（内容已不存在） + SizeFilled int // 回填的总字节数（供日志/对账） +} + +func (s BlockDataBackfillStats) String() string { + return fmt.Sprintf("候选 %d，回填 %d（%d 字节），内容缺失 %d", + s.Candidates, s.Filled, s.SizeFilled, s.Missing) +} + +// BlockPayloadResolver 按 CAS digest 取回原始字节；取不到返回 false。 +type BlockPayloadResolver func(digest string) ([]byte, bool) + +// CountBlockDataBackfillCandidates 统计待回填的块数（data 空、digest 非空）。 +// 报告模式用：不改任何数据。 +func (g *GraphDB) CountBlockDataBackfillCandidates() (int, error) { + if err := g.ensurePayloadDigestColumn(); err != nil { + return 0, err + } + g.mu.RLock() + defer g.mu.RUnlock() + var n int + err := g.db.QueryRow(`SELECT COUNT(*) FROM memory_blocks + WHERE payload_digest != '' + AND (data IS NULL OR length(data) = 0)`).Scan(&n) + return n, err +} + +// BackfillBlockData 把块负载从旧 CAS 回填到 data 列。 +// +// resolve 按 payload_digest 取字节（通常包一层 media.Store.Get）。 +// 单事务、全成功或全回滚。幂等：只处理 data 为空且 payload_digest 非空的块， +// 重复运行不会覆盖已回填的字节。 +// +// 内容在 CAS 里已不存在的块（resolve 返回 false）计入 Missing 并跳过 —— +// 这是无法补救的既有数据损失（CAS 曾被清理过），不是迁移故障，不该阻断 +// 其余块的回填。 +// +// ★ 不校验「回填字节的 sha256 == payload_digest」：那是 CAS 的内部不变量， +// 已由 media.Store.Get 在读出时校验（见其 digest mismatch 检查）。这里 +// 再算一次 sha256 对上万张图是纯浪费。 +func (g *GraphDB) BackfillBlockData(resolve BlockPayloadResolver) (BlockDataBackfillStats, error) { + var st BlockDataBackfillStats + if resolve == nil { + return st, fmt.Errorf("backfill: resolver is required") + } + + g.mu.Lock() + defer g.mu.Unlock() + + // ★ 块 union 化后新库的 schema 不再含 payload_digest 列， + // 迁移只对旧库有意义。新库必须先补上这列，否则查询报 "no such column"。 + if err := g.ensurePayloadDigestColumn(); err != nil { + return st, err + } + + rows, err := g.db.Query(`SELECT id, payload_digest FROM memory_blocks + WHERE payload_digest != '' + AND (data IS NULL OR length(data) = 0) + ORDER BY id`) + if err != nil { + return st, err + } + type cand struct{ id, digest string } + var cands []cand + for rows.Next() { + var c cand + if err := rows.Scan(&c.id, &c.digest); err != nil { + rows.Close() + return st, err + } + cands = append(cands, c) + } + if err := rows.Err(); err != nil { + rows.Close() + return st, err + } + rows.Close() + st.Candidates = len(cands) + if len(cands) == 0 { + return st, nil + } + + tx, err := g.db.Begin() + if err != nil { + return st, err + } + defer tx.Rollback() + + for _, c := range cands { + data, ok := resolve(c.digest) + if !ok || len(data) == 0 { + st.Missing++ + continue + } + if _, err := tx.Exec( + `UPDATE memory_blocks SET data = ? WHERE id = ?`, data, c.id); err != nil { + return st, fmt.Errorf("backfill block %s (digest %s): %w", c.id, c.digest, err) + } + st.Filled++ + st.SizeFilled += len(data) + } + + if err := tx.Commit(); err != nil { + return st, err + } + return st, nil +} + +// ProbeBackfillAvailability 不写库，只按 payload_digest 逐个试取字节， +// 返回能取到的块数。报告模式用：在 -apply 之前先看 CAS 命中率。 +func (g *GraphDB) ProbeBackfillAvailability(resolve BlockPayloadResolver) (int, error) { + if resolve == nil { + return 0, fmt.Errorf("probe: resolver is required") + } + // ★ 块 union 化后新库的 schema 不再含 payload_digest 列， + // 迁移只对旧库有意义。新库先补列（空列 → 查询返回 0 候选，安全空操作）。 + if err := g.ensurePayloadDigestColumn(); err != nil { + return 0, err + } + g.mu.RLock() + rows, err := g.db.Query(`SELECT payload_digest FROM memory_blocks + WHERE payload_digest != '' + AND (data IS NULL OR length(data) = 0)`) + if err != nil { + g.mu.RUnlock() + return 0, err + } + var digests []string + for rows.Next() { + var d string + if err := rows.Scan(&d); err != nil { + rows.Close() + g.mu.RUnlock() + return 0, err + } + digests = append(digests, d) + } + rerr := rows.Err() + rows.Close() + g.mu.RUnlock() + if rerr != nil { + return 0, rerr + } + hit := 0 + for _, d := range digests { + if data, ok := resolve(d); ok && len(data) > 0 { + hit++ + } + } + return hit, nil +} + +// SampleFilledBlockIDs 返回前 n 个已带字节的块 id（供回填后验证 BlockData）。 +func (g *GraphDB) SampleFilledBlockIDs(n int) ([]string, error) { + g.mu.RLock() + defer g.mu.RUnlock() + rows, err := g.db.Query(`SELECT id FROM memory_blocks + WHERE length(data) > 0 ORDER BY id LIMIT ?`, n) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// ensurePayloadDigestColumn 保证 memory_blocks 里有 payload_digest 列。 +// +// 块 union 化后新库的 CREATE 语句已删掉该列；旧库还有它（存量生产库里 +// 它正是回填 data 的来源）。回填迁移要对两者都能跑： +// - 旧库：列在，ALTER 报 "duplicate column" 被忽略 → 正常回填 +// - 新库：列不在，ALTER 加上它（值为空）→ 查询返回 0 候选，安全空操作 +// +// SQLite 没有 ALTER ... ADD COLUMN IF NOT EXISTS，靠"错误即忽略"实现幂等。 +func (g *GraphDB) ensurePayloadDigestColumn() error { + _, err := g.db.Exec(`ALTER TABLE memory_blocks ADD COLUMN payload_digest TEXT DEFAULT ''`) + if err != nil && !strings.Contains(err.Error(), "duplicate column") { + return fmt.Errorf("ensure payload_digest: %w", err) + } + return nil +} diff --git a/internal/memory/migrate_blockdata_test.go b/internal/memory/migrate_blockdata_test.go new file mode 100644 index 00000000..3d4d2464 --- /dev/null +++ b/internal/memory/migrate_blockdata_test.go @@ -0,0 +1,240 @@ +package memory + +import ( + "path/filepath" + "testing" +) + +// BackfillBlockData 把旧 CAS（payload_digest）的字节搬进块自身的 data 列。 +// +// ★ 测试不 import internal/memory/media：字节来源是注入的 resolver， +// 正是为了让 memory 包对 CAS 零依赖（step⑥ 要删 media 包）。 +// 真实 CAS 读盘 + 生产数据 e2e 由 cmd/homed-media-backfill 承担。 +// +// 判据是**不变量**而不是「函数被调过」： +// 1. 回填后 BlockData(id) 能取回字节，且与来源逐字节一致 +// 2. 幂等：再跑一次是空操作（已回填的块不再处理） +// 3. 来源取不到字节的块计 Missing 并跳过，不阻断其余回填 +// 4. 回填只改 data 列，不动块 ID / 不建块 / 不删块 +func TestBackfillBlockData_RoundTrip(t *testing.T) { + dir := t.TempDir() + g, err := NewGraphDB(filepath.Join(dir, "graph.db")) + if err != nil { + t.Fatalf("NewGraphDB: %v", err) + } + t.Cleanup(func() { g.Close() }) + + // 旧 CAS 的等价物：digest → 字节 + cas := map[string][]byte{ + "d1": []byte("img-one"), + "d2": []byte("img-two"), + "d3": []byte("img-three"), + } + + // 造 3 个媒体块，带 payload_digest（旧形态：块 ID 不是内容派生） + // ★ 必须用原生 SQL 写 payload_digest：MemoryBlock 结构体已删该字段， + // PutMemoryBlocks 写不进去。这正是存量生产库的形态—— + // 块 ID 是 blk_，data 为空，payload_digest 指向 CAS。 + // ★ 新库的 CREATE 不含该列（block union 化后删掉了），而「旧库有这列」 + // 正是迁移的前提，所以先补列来模拟旧库形态。 + if err := g.ensurePayloadDigestColumn(); err != nil { + t.Fatalf("ensure payload_digest: %v", err) + } + for _, b := range []struct{ id, digest string }{ + {"blk_old_1", "d1"}, + {"blk_old_2", "d2"}, + {"blk_old_3", "d3"}, + } { + if _, err := g.db.Exec(`INSERT INTO memory_blocks + (id, modality, payload_digest, mime, created_at, updated_at) + VALUES (?, 'image', ?, 'image/png', datetime('now'), datetime('now'))`, + b.id, b.digest); err != nil { + t.Fatalf("insert %s: %v", b.id, err) + } + } + + // 回填前：data 列为空（块形态对，但字节不在块里） + before, _ := g.CountBlockDataBackfillCandidates() + if before != 3 { + t.Fatalf("回填前候选应为 3，实际 %d", before) + } + for _, b := range []string{"blk_old_1", "blk_old_2", "blk_old_3"} { + if data, err := g.BlockData(b); err == nil && len(data) > 0 { + t.Fatalf("回填前块 %s 不该有字节", b) + } + } + + resolve := func(digest string) ([]byte, bool) { + data, ok := cas[digest] + return data, ok + } + + // 执行回填 + st, err := g.BackfillBlockData(resolve) + if err != nil { + t.Fatalf("BackfillBlockData: %v", err) + } + if st.Candidates != 3 || st.Filled != 3 || st.Missing != 0 { + t.Fatalf("回填统计不对: %+v", st) + } + if st.SizeFilled != 7+7+9 { + t.Errorf("SizeFilled = %d，期望 23", st.SizeFilled) + } + + // 判据 1：BlockData 取回的字节与来源逐字节一致 + for _, b := range []struct{ id, digest string }{ + {"blk_old_1", "d1"}, + {"blk_old_2", "d2"}, + {"blk_old_3", "d3"}, + } { + got, err := g.BlockData(b.id) + if err != nil { + t.Fatalf("BlockData(%s): %v", b.id, err) + } + if string(got) != string(cas[b.digest]) { + t.Errorf("块 %s 字节 = %q，期望 %q", b.id, got, cas[b.digest]) + } + } + + // 判据 2：幂等 —— 再跑一次是空操作 + st2, err := g.BackfillBlockData(resolve) + if err != nil { + t.Fatalf("幂等回填: %v", err) + } + if st2.Candidates != 0 || st2.Filled != 0 { + t.Fatalf("幂等应空操作，实际 %+v", st2) + } + + // 判据 4：候选清零，且块 ID 原样保留（迁移不换身份） + after, _ := g.CountBlockDataBackfillCandidates() + if after != 0 { + t.Errorf("回填后候选应为 0，实际 %d", after) + } + ids, _ := g.SampleFilledBlockIDs(10) + if len(ids) != 3 { + t.Errorf("带字节块数应为 3，实际 %d", len(ids)) + } +} + +// 来源里取不到字节的块计 Missing 并跳过，不阻断其余回填。 +func TestBackfillBlockData_MissingDigestSkipped(t *testing.T) { + dir := t.TempDir() + g, err := NewGraphDB(filepath.Join(dir, "graph.db")) + if err != nil { + t.Fatalf("NewGraphDB: %v", err) + } + t.Cleanup(func() { g.Close() }) + + if err := g.PutMemoryBlocks([]MemoryBlock{ + {ID: "blk_ok", Modality: BlockImage, MIME: "image/png"}, + {ID: "blk_lost", Modality: BlockImage, MIME: "image/png"}, + }); err != nil { + t.Fatal(err) + } + // payload_digest 用原生 SQL 注入（同上原因；先确保列存在） + if err := g.ensurePayloadDigestColumn(); err != nil { + t.Fatalf("ensure payload_digest: %v", err) + } + if _, err := g.db.Exec(`UPDATE memory_blocks SET payload_digest = 'exists' WHERE id = 'blk_ok'`); err != nil { + t.Fatal(err) + } + if _, err := g.db.Exec(`UPDATE memory_blocks SET payload_digest = 'gone' WHERE id = 'blk_lost'`); err != nil { + t.Fatal(err) + } + + // resolver 只认 "exists" + resolve := func(digest string) ([]byte, bool) { + if digest == "exists" { + return []byte("bytes"), true + } + return nil, false + } + + st, err := g.BackfillBlockData(resolve) + if err != nil { + t.Fatalf("BackfillBlockData: %v", err) + } + if st.Candidates != 2 || st.Filled != 1 || st.Missing != 1 { + t.Fatalf("统计不对: %+v", st) + } + + // ok 块有字节，lost 块仍空 + if data, _ := g.BlockData("blk_ok"); string(data) != "bytes" { + t.Errorf("blk_ok 字节不对: %q", data) + } + if data, _ := g.BlockData("blk_lost"); len(data) != 0 { + t.Errorf("blk_lost 不该有字节: %q", data) + } +} + +// 迁移是幂等的：连跑三次，结果与跑一次相同（不重复计数、不覆盖已有字节）。 +func TestBackfillBlockData_IdempotentThreeRuns(t *testing.T) { + dir := t.TempDir() + g, _ := NewGraphDB(filepath.Join(dir, "graph.db")) + t.Cleanup(func() { g.Close() }) + + if err := g.ensurePayloadDigestColumn(); err != nil { + t.Fatal(err) + } + if _, err := g.db.Exec(`INSERT INTO memory_blocks (id, modality, payload_digest, created_at, updated_at) + VALUES ('b1', 'image', 'k', datetime('now'), datetime('now'))`); err != nil { + t.Fatal(err) + } + + resolve := func(string) ([]byte, bool) { return []byte("payload"), true } + + total := 0 + for i := 0; i < 3; i++ { + st, err := g.BackfillBlockData(resolve) + if err != nil { + t.Fatalf("run %d: %v", i, err) + } + total += st.Filled + } + if total != 1 { + t.Errorf("三次回填累计 Filled = %d，期望 1（只有第一次真回填）", total) + } + if data, _ := g.BlockData("b1"); string(data) != "payload" { + t.Errorf("字节丢失: %q", data) + } +} + +// 全新库（CREATE 不含 payload_digest 列、无历史数据）上回填必须安全。 +// +// 这是 ensurePayloadDigestColumn 存在的理由：block union 化后新建的库 +// 根本没有该列，而迁移命令可能被误跑在新库上——此时应当补列后得到 +// 「0 候选」，而不是报 "no such column" 崩掉。 +func TestBackfillBlockData_FreshDBSafeNoOp(t *testing.T) { + dir := t.TempDir() + g, _ := NewGraphDB(filepath.Join(dir, "graph.db")) + t.Cleanup(func() { g.Close() }) + + // 放几个正常写入的块（新形态：字节直接在 data，无 payload_digest） + if err := g.PutMemoryBlocks([]MemoryBlock{ + {ID: "blk_new_1", Modality: BlockImage, MIME: "image/png", Data: []byte("already-inline")}, + {ID: "blk_new_2", Modality: BlockText, Text: "hello"}, + }); err != nil { + t.Fatal(err) + } + + // 候选查询不得因缺列而报错 + n, err := g.CountBlockDataBackfillCandidates() + if err != nil { + t.Fatalf("CountBlockDataBackfillCandidates 在新库上报错: %v", err) + } + if n != 0 { + t.Errorf("新库候选应为 0，实际 %d", n) + } + + // 回填是空操作，不碰已有字节 + st, err := g.BackfillBlockData(func(string) ([]byte, bool) { return nil, false }) + if err != nil { + t.Fatalf("新库回填报错: %v", err) + } + if st.Candidates != 0 || st.Filled != 0 { + t.Errorf("新库回填应空操作，实际 %+v", st) + } + if data, _ := g.BlockData("blk_new_1"); string(data) != "already-inline" { + t.Errorf("新库已有字节被改: %q", data) + } +}