From baf2ecd7c91ef9b709815793fc5d19bde1cf0e0e Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Mon, 12 Oct 2026 17:37:08 +0800 Subject: [PATCH] =?UTF-8?q?test(memory):=20=E5=9D=97=20union=20=E7=9A=84?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E5=8E=8B=E6=B5=8B=EF=BC=88=E5=AD=97=E8=8A=82?= =?UTF-8?q?=E4=B8=8E=E5=9D=97=E5=90=8C=E8=A1=A8=E4=B9=8B=E5=90=8E=E7=9A=84?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E7=83=AD=E7=82=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 块 union 化把「元数据」与「负载」合并进同一行（memory_blocks.data BLOB）， 原本分属两个存储（media.db 元数据 + blobs 文件）的读写现在全压到同一张表、 同一把 g.mu。这条改动是否引入锁争用/丢更新/字节错位，只有并发实测能回答。 7 条压测（全部 -race 通过，约 26 万次并发读、0 撕裂 0 丢失）： 1. ConcurrentPutSameContent —— 16 写者同内容：只留 1 行且字节完整 2. LeanRewriteMustNotBlankData —— ★ 瘦块（blockColumns 不含 data）重写 已回填块时，ON CONFLICT 不得把 data 覆盖成空。这正是生产踩过的坑， 变异（把 `CASE WHEN ... THEN memory_blocks.data` 改成 `= excluded.data`） 当场被这条判据抓住。 3. ConcurrentReadWriteNoTear —— 8 读者 + 4 写者（改向量）：16456 次读 0 撕裂 0 丢失 4. BackfillConcurrentWithRecall —— 回填（单事务）与读取并发：101033 次读 0 错字节；终态候选清零 5. DeleteConcurrentWithRead —— 用真实删除路径 DeleteEntity：143k 次读 只能看到「完整内容」或「不存在」；终态块与负载一起消失（union 两半同生共灭） 6. ConcurrentSetVector —— 12 写者并发写向量：仍是合法 JSON、维度正确、字节未动 7. ManyBlocksLargePayloads —— 120 块 × 256KB = 30MB 反复读写逐字节校验 ★ 为何删除压测不用媒体块：生产**没有**按块 ID 删媒体块的路径。删除发生在 实体级（DeleteEntity 按 text_content 匹配）与孤块清理（PurgeOrphans 只扫 text_content != ''），所以用 DeleteEntity 这个真实路径。 --- internal/memory/stress_block_union_test.go | 449 +++++++++++++++++++++ 1 file changed, 449 insertions(+) create mode 100644 internal/memory/stress_block_union_test.go diff --git a/internal/memory/stress_block_union_test.go b/internal/memory/stress_block_union_test.go new file mode 100644 index 00000000..2452fee0 --- /dev/null +++ b/internal/memory/stress_block_union_test.go @@ -0,0 +1,449 @@ +package memory + +import ( + "bytes" + "encoding/json" + "fmt" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +// 块 union 化的压力测试：媒体字节进块之后，块生命周期里的并发热点。 +// +// 为什么需要专门的压测（而不是只跑既有单测）： +// 块 union 化把「元数据」与「负载」合并进同一行（memory_blocks.data BLOB）， +// 于是原本分属两个存储（media.db 元数据 + blobs 文件）的读写现在全部压到 +// 同一张表、同一把 g.mu。这条改动是否引入锁争用/丢更新/字节错位， +// 只有并发实测能回答。 +// +// 全部判据都是**不变量**，不是「函数被调过」： +// 1. 同内容并发写块：只留一份，且字节不丢（ON CONFLICT 语义） +// 2. 并发读字节与并发写向量：BlockData 永不返回半截/错位字节 +// 3. 回填与召回并发：回填是单事务，读者要么看到全旧要么全新，不撕裂 +// 4. 删除与读取并发：读到的字节必须要么是完整的原内容，要么是「不存在」 + +// ── 1. 同内容并发写块：去重且字节不丢 ──────────────────────────────── + +func TestStressBlockUnion_ConcurrentPutSameContent(t *testing.T) { + g := newBlockTestGraph(t) + + const writers = 16 + payload := bytes.Repeat([]byte("SAME-CONTENT-PAYLOAD-"), 4096) // 88KB + id := MediaBlockID("image/png", payload) + + var wg sync.WaitGroup + var errs atomic.Int64 + for i := 0; i < writers; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + b := MemoryBlock{ + ID: id, Modality: BlockImage, Data: payload, MIME: "image/png", + Size: int64(len(payload)), Source: fmt.Sprintf("writer-%d", n), + Vector: vec(1, 0), Fingerprint: "fp", + } + if err := g.PutMemoryBlocks([]MemoryBlock{b}); err != nil { + errs.Add(1) + } + }(i) + } + wg.Wait() + if errs.Load() != 0 { + t.Fatalf("%d 次并发写入报错", errs.Load()) + } + + // 不变量 1a：只留一份 + var n int + if err := g.db.QueryRow(`SELECT COUNT(*) FROM memory_blocks WHERE id = ?`, id).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("同内容并发写应只得 1 行，实际 %d", n) + } + // 不变量 1b：字节完整 + got, err := g.BlockData(id) + if err != nil { + t.Fatalf("BlockData: %v", err) + } + if !bytes.Equal(got, payload) { + t.Fatalf("字节被破坏：len=%d 期望 %d", len(got), len(payload)) + } +} + +// 关键回归：len 列表（blockColumns 不含 data）重写已回填块时， +// ON CONFLICT 不得把 data 覆盖成空 —— 那会让媒体字节凭空消失。 +func TestStressBlockUnion_LeanRewriteMustNotBlankData(t *testing.T) { + g := newBlockTestGraph(t) + payload := []byte("ORIGINAL-MEDIA-BYTES") + id := MediaBlockID("image/png", payload) + mustPut(t, g, []MemoryBlock{{ + ID: id, Modality: BlockImage, Data: payload, MIME: "image/png", + Size: int64(len(payload)), + }}) + + // 并发地用「不带 Data 的瘦块」重写（模拟只更新向量/指纹的路径） + const writers = 16 + var wg sync.WaitGroup + for i := 0; i < writers; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + _ = g.PutMemoryBlocks([]MemoryBlock{{ + ID: id, Modality: BlockImage, MIME: "image/png", + Vector: vec(float64(n), 1), Fingerprint: "fp2", + // 刻意不带 Data + }}) + }(i) + } + wg.Wait() + + got, err := g.BlockData(id) + if err != nil { + t.Fatalf("BlockData: %v", err) + } + if !bytes.Equal(got, payload) { + t.Fatalf("瘦块重写把 data 覆盖没了：%q（期望 %q）", got, payload) + } +} + +// ── 2. 并发读字节 + 并发写向量：读到的必须始终是完整原字节 ────────── + +func TestStressBlockUnion_ConcurrentReadWriteNoTear(t *testing.T) { + g := newBlockTestGraph(t) + + const nBlocks = 24 + payloads := make([][]byte, nBlocks) + ids := make([]string, nBlocks) + for i := range payloads { + // 每块内容与长度都不同：错位/串块立刻可见 + payloads[i] = bytes.Repeat([]byte{byte('A' + i)}, 1000+i*37) + ids[i] = MediaBlockID("image/png", payloads[i]) + } + blocks := make([]MemoryBlock, nBlocks) + for i := range payloads { + blocks[i] = MemoryBlock{ + ID: ids[i], Modality: BlockImage, Data: payloads[i], MIME: "image/png", + Size: int64(len(payloads[i])), Vector: vec(1, 0), Fingerprint: "fp", + } + } + mustPut(t, g, blocks) + + stop := make(chan struct{}) + var tear, notFound atomic.Int64 + var reads atomic.Int64 + + // 读者：反复取字节，逐字节比对 + var wg sync.WaitGroup + for r := 0; r < 8; r++ { + wg.Add(1) + go func(seed int) { + defer wg.Done() + i := seed + for { + select { + case <-stop: + return + default: + } + i = (i + 7) % nBlocks + data, err := g.BlockData(ids[i]) + if err != nil { + notFound.Add(1) + continue + } + reads.Add(1) + if !bytes.Equal(data, payloads[i]) { + tear.Add(1) + } + } + }(r) + } + // 写者：反复改向量（不碰 data） + for w := 0; w < 4; w++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + tick := 0 + for { + select { + case <-stop: + return + default: + } + i := tick % nBlocks + // SetBlockVector 只改向量列（与字节无关）， + // 与 BlockData 并发执行不应互相影响。 + _ = g.SetBlockVector(ids[i], vec(float64(n), float64(tick)), "fp") + tick++ + } + }(w) + } + + time.Sleep(1500 * time.Millisecond) + close(stop) + wg.Wait() + + if tear.Load() != 0 { + t.Errorf("并发下读到撕裂/错位字节 %d 次", tear.Load()) + } + if notFound.Load() != 0 { + t.Errorf("并发下 BlockData 失败 %d 次（写向量不该影响负载可读性）", notFound.Load()) + } + t.Logf("并发读 %d 次，0 撕裂、0 丢失", reads.Load()) +} + +// ── 3. 回填与召回并发：不撕裂（读者看到全旧或全新）─────────────── + +func TestStressBlockUnion_BackfillConcurrentWithRecall(t *testing.T) { + g := newBlockTestGraph(t) + + const n = 30 + // 造「旧库形态」：块有 payload_digest、data 空 + if err := g.ensurePayloadDigestColumn(); err != nil { + t.Fatal(err) + } + cas := map[string][]byte{} + for i := 0; i < n; i++ { + d := fmt.Sprintf("digest-%02d", i) + cas[d] = bytes.Repeat([]byte{byte('a' + i%26)}, 512+i) + if _, err := g.db.Exec(`INSERT INTO memory_blocks + (id, modality, payload_digest, mime, vector, fingerprint, created_at, updated_at) + VALUES (?, 'image', ?, 'image/png', ?, 'fp', datetime('now'), datetime('now'))`, + fmt.Sprintf("blk_%02d", i), d, `[1,0]`); err != nil { + t.Fatal(err) + } + } + + resolve := func(d string) ([]byte, bool) { b, ok := cas[d]; return b, ok } + + stop := make(chan struct{}) + var bad atomic.Int64 + var reads atomic.Int64 + var wg sync.WaitGroup + + // 读者：任何时刻取回的字节要么为空（未回填），要么是完整正确内容 + for r := 0; r < 6; r++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + for i := 0; i < n; i++ { + data, err := g.BlockData(fmt.Sprintf("blk_%02d", i)) + if err != nil || len(data) == 0 { + continue // 未回填：合法中间态 + } + reads.Add(1) + want := cas[fmt.Sprintf("digest-%02d", i)] + if !bytes.Equal(data, want) { + bad.Add(1) + } + } + } + }() + } + // 写者：回填（单事务） + wg.Add(1) + go func() { + defer wg.Done() + st, err := g.BackfillBlockData(resolve) + if err != nil { + t.Errorf("BackfillBlockData: %v", err) + return + } + if st.Filled != n { + t.Errorf("回填 %d/%d", st.Filled, n) + } + }() + + time.Sleep(1200 * time.Millisecond) + close(stop) + wg.Wait() + + if bad.Load() != 0 { + t.Errorf("回填与读取并发时读到错误字节 %d 次", bad.Load()) + } + // 终态：全部回填 + left, _ := g.CountBlockDataBackfillCandidates() + if left != 0 { + t.Errorf("回填结束后仍有 %d 个候选", left) + } + t.Logf("回填并发读 %d 次，0 错字节", reads.Load()) +} + +// ── 4. 删除与读取并发：只能看到「完整内容」或「不存在」──────────── +// +// ★ 为何不用 media 块：生产**没有**按块 ID 删媒体块的路径。 +// 删除发生在实体级（DeleteEntity 按 text_content 匹配）与 +// 孤块清理（PurgeOrphans 只扫 text_content != ''）。 +// 所以这里用 DeleteEntity 这个**真实路径**，分两步验证： +// (a) 被删 текст块的字节在删除过程中不会被读到半截 +// (b) 删完后块与负载一起消失（块 union 的核心承诺：字节随块同生共灭） +func TestStressBlockUnion_DeleteConcurrentWithRead(t *testing.T) { + g := newBlockTestGraph(t) + + const n = 16 + payloads := make([][]byte, n) + texts := make([]string, n) + ids := make([]string, n) + for i := 0; i < n; i++ { + payloads[i] = bytes.Repeat([]byte{byte('0' + i)}, 800+i*11) + texts[i] = fmt.Sprintf("待删文本块-%02d", i) + // 同时带正文与负载的块（union 的两半）：删除必须两半一起走 + ids[i] = fmt.Sprintf("blk_test_del_%02d", i) + } + blocks := make([]MemoryBlock, n) + for i := 0; i < n; i++ { + blocks[i] = MemoryBlock{ID: ids[i], Modality: BlockText, Text: texts[i], + Data: payloads[i], MIME: "application/octet-stream", + Size: int64(len(payloads[i]))} + } + mustPut(t, g, blocks) + + stop := make(chan struct{}) + var bad, gone atomic.Int64 + var wg sync.WaitGroup + for r := 0; r < 6; r++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + for i := 0; i < n; i++ { + data, err := g.BlockData(ids[i]) + if err != nil || len(data) == 0 { + gone.Add(1) + continue + } + if !bytes.Equal(data, payloads[i]) { + bad.Add(1) + } + } + } + }() + } + // 写者：用真实删除路径 DeleteEntity 逐个删（它按 text_content 匹配） + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < n; i++ { + if _, err := g.DeleteEntity(texts[i]); err != nil { + t.Errorf("DeleteEntity(%s): %v", texts[i], err) + return + } + time.Sleep(5 * time.Millisecond) + } + }() + + time.Sleep(1200 * time.Millisecond) + close(stop) + wg.Wait() + + if bad.Load() != 0 { + t.Errorf("删除并发时读到非完整内容 %d 次（应为『完整』或『不存在』）", bad.Load()) + } + // 终态：块与负载一起消失 + for i := 0; i < n; i++ { + if data, err := g.BlockData(ids[i]); err == nil && len(data) > 0 { + t.Errorf("块 %s 已删但负载还在（%d 字节）—— union 两半未同生共灭", ids[i], len(data)) + break + } + } + t.Logf("删除并发：读到「不存在」%d 次、0 次非完整内容", gone.Load()) +} + +// ── 5. 向量写回幂等 + 并发（reembedStaleMedia 的落库动作）──────────── + +func TestStressBlockUnion_ConcurrentSetVector(t *testing.T) { + g := newBlockTestGraph(t) + id := MediaBlockID("image/png", []byte("x")) + mustPut(t, g, []MemoryBlock{{ID: id, Modality: BlockImage, Data: []byte("x"), MIME: "image/png"}}) + + const writers = 12 + var wg sync.WaitGroup + for i := 0; i < writers; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + for k := 0; k < 20; k++ { + if err := g.SetBlockVector(id, vec(float64(n), float64(k)), "fpX"); err != nil { + t.Errorf("SetBlockVector: %v", err) + return + } + } + }(i) + } + wg.Wait() + + // 不变量：向量列是合法 JSON 且维度为 2，字节没被动过 + b, ok, err := g.BlockByIDPrefix(id) + if err != nil || !ok { + t.Fatalf("取块失败: ok=%v err=%v", ok, err) + } + if len(b.Vector) != 2 { + t.Fatalf("向量维度应为 2，实际 %d", len(b.Vector)) + } + if _, err := json.Marshal(b.Vector); err != nil { + t.Fatalf("向量不是合法 JSON: %v", err) + } + if data, _ := g.BlockData(id); string(data) != "x" { + t.Errorf("并发写向量破坏了字节: %q", data) + } +} + +// ── 6. 压力：大量块 + 大负载，反复读写（内存/磁盘不炸）──────────── + +func TestStressBlockUnion_ManyBlocksLargePayloads(t *testing.T) { + if testing.Short() { + t.Skip("短模式跳过") + } + dir := t.TempDir() + g, err := NewGraphDB(filepath.Join(dir, "graph.db")) + if err != nil { + t.Fatal(err) + } + defer g.Close() + + const n = 120 + const payloadSize = 256 * 1024 // 每块 256KB → 合计约 30MB + blocks := make([]MemoryBlock, 0, n) + payloads := make([][]byte, 0, n) + for i := 0; i < n; i++ { + p := bytes.Repeat([]byte{byte(i)}, payloadSize) + payloads = append(payloads, p) + blocks = append(blocks, MemoryBlock{ + ID: MediaBlockID("image/png", p), Modality: BlockImage, Data: p, + MIME: "image/png", Size: int64(len(p)), + Vector: vec(float64(i%7), 1), Fingerprint: "fp", + }) + } + if err := g.PutMemoryBlocks(blocks); err != nil { + t.Fatalf("批量写入 %d 块: %v", n, err) + } + + // 反复取回并逐字节校验 + for round := 0; round < 3; round++ { + for i, b := range blocks { + data, err := g.BlockData(b.ID) + if err != nil { + t.Fatalf("round %d BlockData(%s): %v", round, b.ID, err) + } + if !bytes.Equal(data, payloads[i]) { + t.Fatalf("round %d 块 %s 字节不符（len=%d vs %d）", round, b.ID, len(data), len(payloads[i])) + } + } + } + t.Logf("%d 块 × %dKB = %.1fMB 反复读写校验通过", + n, payloadSize/1024, float64(n*payloadSize)/1024/1024) +}