package DB import ( "fmt" "path/filepath" "sync" "testing" ) // ccRow 并发用例的行: 只有一列, 免得用例本身被别的机制干扰。 type ccRow struct { BaseModel Name string `vrtx:"maxlen:32"` } // TestConcurrentInsertAndCommit 一个 goroutine 写、另一个 goroutine 提交 —— 这是 // 「自动提交开着 + 业务自己再显式 Commit」以及各种后台提交线程的常见写法, 也是旧实现 // 会稳定 panic 的那条路。 // // 旧实现的窗口: DB.Commit 先 CollectDirty —— 把脏页【移出】缓冲池, 再走 WAL // (writeWALPages + 两次 fsync) 和逐页回写。从移出到回写完成这段时间里, 那些页 // 既不在缓存、也还没落到自己的位置: 并发的读一旦缓存未命中就回源, 要么读到旧内容, // 要么 (页在本会话刚分配、盘上还从没写过) 读到全零 —— 而 btree 把全零页当成 // 「EntryLen == 0 的节点」, 直接在 btPID 里切片越界 panic。 // // 修复后脏页在提交期间【留在缓存】, 只有等它确实回写成功、且期间没被再改过, 才转成 // 干净页。因此下面这个用例必须整轮跑完且一行不少。 func TestConcurrentInsertAndCommit(t *testing.T) { dbPath := filepath.Join(t.TempDir(), "concurrent.db") db, err := Open(dbPath) if err != nil { t.Fatalf("打开 %s: %v", dbPath, err) } defer db.Close() tbl, err := UseTable[ccRow](db, "cc") if err != nil { t.Fatalf("建表: %v", err) } // 关掉防抖定时器: 提交时机完全由下面那个 goroutine 决定, 用例才可复现。 db.SetAutoCommitDelay(0) const rows = 3000 // 写入侧与提交侧各自把 panic / error 交回来: goroutine 里 panic 会直接带走整个 // 进程, 测试根本跑不到断言 —— 必须在 goroutine 内 recover 才能把它变成失败。 fail := make(chan any, 1) report := func(v any) { select { case fail <- v: default: } } var writer sync.WaitGroup writer.Add(1) go func() { defer writer.Done() defer func() { if r := recover(); r != nil { report(fmt.Errorf("写入 goroutine panic: %v", r)) } }() for i := 0; i < rows; i++ { if err := tbl.Insert(&ccRow{Name: fmt.Sprintf("n%06d", i)}); err != nil { report(fmt.Errorf("第 %d 行写入失败: %w", i+1, err)) return } } }() stop := make(chan struct{}) var committer sync.WaitGroup committer.Add(1) go func() { defer committer.Done() defer func() { if r := recover(); r != nil { report(fmt.Errorf("提交 goroutine panic: %v", r)) } }() for { select { case <-stop: return default: } if err := db.Commit(); err != nil { report(fmt.Errorf("提交失败: %w", err)) return } } }() writer.Wait() close(stop) committer.Wait() select { case v := <-fail: t.Fatal(v) default: } // 行数必须一行不少: 少的那部分正是「提交与写入重叠时被吞掉 / 被当成干净页丢掉」的。 if err := db.Commit(); err != nil { t.Fatalf("收尾提交: %v", err) } n, err := tbl.Count() if err != nil { t.Fatalf("查行数: %v", err) } if n != rows { t.Fatalf("行数应当 %d, 实际 %d", rows, n) } // 关掉再开一遍: 确认提交出去的东西真的在盘上, 而不只是活在缓存里。 if err := db.Close(); err != nil { t.Fatalf("关闭: %v", err) } reopened, err := Open(dbPath) if err != nil { t.Fatalf("重开: %v", err) } defer reopened.Close() tbl2, err := UseTable[ccRow](reopened, "cc") if err != nil { t.Fatalf("重开表: %v", err) } if n2, err := tbl2.Count(); err != nil || n2 != rows { t.Fatalf("重开后行数应当 %d, 实际 %d (err=%v)", rows, n2, err) } } // TestWritePageDirectKeepsCacheCoherent 直写盘之后缓存副本必须失效 —— 且失效动作要 // 排在【落盘之后】。 // // 反了的话, WritePageDirect 先删缓存再写盘, 中间这一小段页同样"两头都不在": 对一页 // 本会话刚分配、盘上从没写过的页, 并发读会拿到全零。 func TestWritePageDirectKeepsCacheCoherent(t *testing.T) { dbPath := filepath.Join(t.TempDir(), "direct.db") db, err := Open(dbPath) if err != nil { t.Fatalf("打开 %s: %v", dbPath, err) } defer db.Close() tbl, err := UseTable[ccRow](db, "direct") if err != nil { t.Fatalf("建表: %v", err) } db.SetAutoCommitDelay(0) // 脏页留在缓存里, 好让下面读到的是缓存副本 for i := 0; i < 50; i++ { if err := tbl.Insert(&ccRow{Name: fmt.Sprintf("d%06d", i)}); err != nil { t.Fatalf("第 %d 行写入失败: %v", i+1, err) } } // 有脏页在缓存里时反复"直写盘 + 读回", 全程不允许出现读到全零页的情况。 // 这里直接调 ORM 的 Commit (内部就是 WAL 直写回填), 再立刻按 pk 读回。 for round := 0; round < 20; round++ { if err := db.Commit(); err != nil { t.Fatalf("第 %d 轮提交: %v", round+1, err) } got, err := tbl.Find(1) if err != nil { t.Fatalf("第 %d 轮按主键读回: %v", round+1, err) } if got == nil || got.Name != "d000000" { t.Fatalf("第 %d 轮读回的内容不对: %+v", round+1, got) } } } // TestConcurrentCommitDoesNotLoseDirtyPages 提交窗口里被再次改脏的页, 不能因为一次 // 「提交成功」就被顺手标成干净页。 // // ClearAll 那种「提交成功就把整个缓存标干净」的做法, 对并发写入是致命的: 窗口期内 // 新弄脏的页没有被写进本笔 WAL, 却被抹掉脏标记 —— 缓存里是新值, 盘上是旧值, 而引擎 // 以为两边一致; 该页一旦被淘汰, 新值就永久消失。 func TestConcurrentCommitDoesNotLoseDirtyPages(t *testing.T) { dbPath := filepath.Join(t.TempDir(), "dirty.db") db, err := Open(dbPath) if err != nil { t.Fatalf("打开 %s: %v", dbPath, err) } defer db.Close() tbl, err := UseTable[ccRow](db, "dirty") if err != nil { t.Fatalf("建表: %v", err) } db.SetAutoCommitDelay(0) // 先把表建出来并落一次盘, 让后面的写入都发生在"已提交结构"之上。 if err := tbl.Insert(&ccRow{Name: "base"}); err != nil { t.Fatalf("写入基线行: %v", err) } if err := db.Commit(); err != nil { t.Fatalf("基线提交: %v", err) } // 写入侧持续改同一批行 (Update 会反复弄脏同一批索引页/数据页, 最容易撞上窗口), // 提交侧同时不停提交。 const rounds = 400 fail := make(chan any, 1) report := func(v any) { select { case fail <- v: default: } } stop := make(chan struct{}) var committer sync.WaitGroup committer.Add(1) go func() { defer committer.Done() defer func() { if r := recover(); r != nil { report(fmt.Errorf("提交 goroutine panic: %v", r)) } }() for { select { case <-stop: return default: } if err := db.Commit(); err != nil { report(fmt.Errorf("提交失败: %w", err)) return } } }() for i := 1; i <= rounds; i++ { row := &ccRow{Name: fmt.Sprintf("v%06d", i)} if err := tbl.Insert(row); err != nil { report(fmt.Errorf("第 %d 行写入失败: %w", i, err)) break } } close(stop) committer.Wait() select { case v := <-fail: t.Fatal(v) default: } // 收尾: 提交 + 关闭重开。任何"被标成干净但其实没落盘"的页都会在这里现形。 if err := db.Commit(); err != nil { t.Fatalf("收尾提交: %v", err) } want := int64(rounds + 1) if n, err := tbl.Count(); err != nil || n != want { t.Fatalf("关闭前行数应当 %d, 实际 %d (err=%v)", want, n, err) } if err := db.Close(); err != nil { t.Fatalf("关闭: %v", err) } reopened, err := Open(dbPath) if err != nil { t.Fatalf("重开: %v", err) } defer reopened.Close() tbl2, err := UseTable[ccRow](reopened, "dirty") if err != nil { t.Fatalf("重开表: %v", err) } if n, err := tbl2.Count(); err != nil || n != want { t.Fatalf("重开后行数应当 %d, 实际 %d (err=%v)", want, n, err) } if _, err := tbl2.Find(1); err != nil { t.Fatalf("重开后按主键读回: %v", err) } }