TsDB2/DB/concurrent_test.go

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)
	}
}