TsDB2/DB/core/pool/buffer_pool_test.go

package pool

import (
	"sync"
	"testing"
	"time"
)

// newTestPool 构造一个带内存"磁盘"的缓冲池，便于观察缓存与落盘行为。
func newTestPool(t *testing.T, capacity int) (*BufferPool, *diskStub) {
	t.Helper()
	d := &diskStub{pages: map[uint32][]byte{}}
	return NewBufferPool(capacity, d.read, d.write), d
}

type diskStub struct {
	mu    sync.Mutex
	pages map[uint32][]byte
}

func (d *diskStub) read(pageID uint32, buf []byte) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	if data, ok := d.pages[pageID]; ok {
		copy(buf, data)
	}
	return nil
}

func (d *diskStub) write(pageID uint32, buf []byte) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	data := make([]byte, len(buf))
	copy(data, buf)
	d.pages[pageID] = data
	return nil
}

// TestCollectDirtyKeepsDirtyPagesReadable 收集脏页快照之后, 脏页必须【仍然留在缓存里】
// 且内容可读 —— 这是"并发写 + 提交"崩溃的直接回归点。
//
// 旧实现把脏页移出缓存, 于是从收集完到 WAL 逐页回写完成(中间隔着两次 fsync)这段窗口
// 里, 这些页既不在缓存、也还没落盘: 并发的读一旦未命中就回源, 拿到的是旧内容, 对
// 一页本会话刚分配、盘上从没写过的页则直接是全零。btree 把全零页当成 EntryLen==0 的
// 节点, 接着就在 btPID 里切片越界 panic; 文件还没长到那个偏移时则读出错, 表现为
// "insert: load failed"。
//
// 注意旧版本的这条用例恰好把 bug 当成预期钉住了: 它断言"脏页回落到磁盘, 期望 0"。
func TestCollectDirtyKeepsDirtyPagesReadable(t *testing.T) {
	bp, _ := newTestPool(t, 64)

	_ = bp.WritePage(1, []byte{1}) // 脏页 (盘上还没有)
	bp.UpdatePage(2, []byte{9, 9}) // 干净页 (WAL 回填/回收页回填)
	_ = bp.WritePage(3, []byte{7}) // 脏页 (盘上还没有)

	if bp.Len() != 3 || bp.DirtyCount() != 2 {
		t.Fatalf("收集前期望池内 3 页 / 脏页 2, 实际 %d / %d", bp.Len(), bp.DirtyCount())
	}

	dirty := bp.CollectDirty()
	if len(dirty) != 2 {
		t.Fatalf("期望取出 2 个脏页, 实际 %d", len(dirty))
	}

	// 快照不等于"页离开了池子": 页数与脏页数都不该变。
	if bp.Len() != 3 {
		t.Fatalf("收集后页仍应全部留在池中, 实际 %d", bp.Len())
	}
	if bp.DirtyCount() != 2 {
		t.Fatalf("收集后脏页计数不该清零 (它们还没落盘), 实际 %d", bp.DirtyCount())
	}

	// 干净页 2 命中缓存
	out2 := make([]byte, 16)
	if err := bp.ReadPage(2, out2); err != nil {
		t.Fatalf("ReadPage(2): %v", err)
	}
	if out2[0] != 9 || out2[1] != 9 {
		t.Fatalf("干净页应命中缓存并返回回填值, 实际 %v", out2[:2])
	}

	// 脏页 1 / 3 也必须命中缓存并读到真实内容 —— 拿到 0 就说明又出现"两头都不在"。
	for _, c := range []struct {
		pageID uint32
		want   byte
	}{{1, 1}, {3, 7}} {
		out := make([]byte, 16)
		if err := bp.ReadPage(c.pageID, out); err != nil {
			t.Fatalf("ReadPage(%d): %v", c.pageID, err)
		}
		if out[0] != c.want {
			t.Fatalf("收集快照后读取页 %d 应当是 %d, 实际 %d (页被移出缓存就会出现这种全零读)",
				c.pageID, c.want, out[0])
		}
	}
}

// TestMarkCleanOnlyClearsUnchangedPages 提交成功收脏标记时必须按版本号逐页判定:
// 提交窗口里被写入线程重新改过的页要继续留脏。
//
// 反例就是 ClearAll 式的"提交成功 → 整个缓存标干净": 那些页并不在本笔 WAL 里, 标干净
// 等于宣布"已持久化" —— 新内容再也不会被写盘, 页被淘汰时静默消失。
func TestMarkCleanOnlyClearsUnchangedPages(t *testing.T) {
	bp, _ := newTestPool(t, 64)

	_ = bp.WritePage(1, []byte{1}) // 会被重新改脏
	_ = bp.WritePage(2, []byte{2}) // 保持原样
	dirty := bp.CollectDirty()
	if len(dirty) != 2 {
		t.Fatalf("期望 2 个脏页, 实际 %d", len(dirty))
	}

	// 模拟"提交进行中, 写入线程又改了页 1"
	_ = bp.WritePage(1, []byte{0xFF})

	// 提交成功 → WAL 回填步骤 (WritePageHome) 已在盘上写下快照内容
	bp.MarkClean(dirty)

	if bp.DirtyCount() != 1 {
		t.Fatalf("被重新改脏的页必须继续留脏, 期望脏页 1, 实际 %d", bp.DirtyCount())
	}
	out := make([]byte, 16)
	if err := bp.ReadPage(1, out); err != nil {
		t.Fatalf("ReadPage(1): %v", err)
	}
	if out[0] != 0xFF {
		t.Fatalf("页 1 应保留提交窗口里写入的新内容 0xFF, 实际 %d", out[0])
	}
	// 页 2 没被改过 → 转干净页 (内容不受影响)
	if entry, ok := bp.pages[2]; !ok || entry.dirty {
		t.Fatalf("未被改动的页应转成干净页, 实际 %+v (存在=%v)", entry, ok)
	}
}

// TestFlushWritesAndClearsDirty Flush 必须真的落盘, 并且把写出去的页转成干净页 ——
// CollectDirty 不再移出缓存, 少了收尾这一步, 同一批内容会被下一次提交再写一遍。
func TestFlushWritesAndClearsDirty(t *testing.T) {
	bp, disk := newTestPool(t, 64)

	_ = bp.WritePage(1, []byte{0x11})
	_ = bp.WritePage(2, []byte{0x22})

	bp.Flush()

	if bp.DirtyCount() != 0 {
		t.Fatalf("Flush 后不该还有脏页, 实际 %d", bp.DirtyCount())
	}
	for _, c := range []struct {
		pageID uint32
		want   byte
	}{{1, 0x11}, {2, 0x22}} {
		got, ok := disk.pages[c.pageID]
		if !ok {
			t.Fatalf("页 %d 没有被写盘", c.pageID)
		}
		if got[0] != c.want {
			t.Fatalf("页 %d 落盘内容应当是 %#x, 实际 %#x", c.pageID, c.want, got[0])
		}
	}
}

// TestWritePageThresholdCountsOnlyDirty 验证自动提交阈值只由脏页数决定，
// 常驻的干净页不会把阈值反复顶爆、导致提交频率失控。
func TestWritePageThresholdCountsOnlyDirty(t *testing.T) {
	bp, _ := newTestPool(t, 1000)

	triggers := 0
	bp.SetDirtyThreshold(10, func() { triggers++ })

	// 灌入大量干净页 (模拟大量回收页经 UpdatePage 回填缓存)
	for i := uint32(0); i < 200; i++ {
		bp.UpdatePage(1000+i, []byte{1})
	}

	// 9 张脏页 —— 未达阈值，不应触发
	for i := uint32(0); i < 9; i++ {
		_ = bp.WritePage(i, []byte{2})
	}
	if triggers != 0 {
		t.Fatalf("干净页不应触发阈值, 实际触发 %d 次", triggers)
	}

	// 第 10 张脏页 → 命中阈值，触发一次
	_ = bp.WritePage(9, []byte{2})
	if triggers != 1 {
		t.Fatalf("脏页达阈值应触发 1 次, 实际 %d 次", triggers)
	}
	if bp.DirtyCount() != 10 {
		t.Fatalf("期望脏页计数 10, 实际 %d", bp.DirtyCount())
	}
	if bp.Len() < 200 {
		t.Fatalf("干净页应保留在池中, 实际池内仅 %d 页", bp.Len())
	}
}

// countingRead 包装 diskStub.read 计数 (用于断言"命中缓存后不再走磁盘")。
type countingRead struct {
	d *diskStub
	n int
}

func (c *countingRead) read(pageID uint32, buf []byte) error {
	c.n++
	return c.d.read(pageID, buf)
}

// TestReadPageCachesCleanPages 读未命中必须把磁盘内容作为干净页载入缓存:
// 第二次读同一页必须命中缓存 (磁盘读次数不增加), 这是"LRU BufferPool"读路径
// 收益的主体 (B+ 树下降 / Hash 目录链 / 堆页 Load 都是重复读)。
func TestReadPageCachesCleanPages(t *testing.T) {
	d := &diskStub{pages: map[uint32][]byte{42: {0x2A}}}
	cr := &countingRead{d: d}
	bp := NewBufferPool(8, cr.read, d.write)

	buf := make([]byte, 1)
	if err := bp.ReadPage(42, buf); err != nil {
		t.Fatalf("第一次读: %v", err)
	}
	if buf[0] != 0x2A {
		t.Fatalf("读到 %d, 想要 0x2A", buf[0])
	}
	if cr.n != 1 {
		t.Fatalf("第一次读应走 1 次磁盘, 实际 %d", cr.n)
	}

	// 第二次读: 必须命中缓存
	if err := bp.ReadPage(42, buf); err != nil {
		t.Fatalf("第二次读: %v", err)
	}
	if cr.n != 1 {
		t.Fatalf("第二次读应命中缓存 (磁盘读仍为 1), 实际 %d", cr.n)
	}

	// 载入的是干净页: 不占脏页计数, 可被 LRU 淘汰
	if bp.DirtyCount() != 0 {
		t.Fatalf("读缓存页不应计脏, 实际脏页 %d", bp.DirtyCount())
	}
}

// TestLRUEvictsColdestPage LRU 淘汰必须按"最久未用"顺序: 最近读过的热页留下,
// 最冷的页被淘汰。旧实现按 map 随机序淘汰, 热页与冷页同等概率被丢。
func TestLRUEvictsColdestPage(t *testing.T) {
	d := &diskStub{pages: map[uint32][]byte{}}
	cr := &countingRead{d: d}
	bp := NewBufferPool(3, cr.read, d.write)

	// 三张干净页入池 (模拟读缓存), lru 序: [1, 2, 3]
	bp.UpdatePage(1, []byte{1})
	bp.UpdatePage(2, []byte{2})
	bp.UpdatePage(3, []byte{3})

	// 读页 1 → 它变热: lru 序 [2, 3, 1]
	if err := bp.ReadPage(1, make([]byte, 1)); err != nil {
		t.Fatalf("ReadPage(1): %v", err)
	}

	// 写页 4 (触发容量淘汰): 最冷的页 2 必须先被淘汰
	if err := bp.WritePage(4, []byte{4}); err != nil {
		t.Fatalf("WritePage(4): %v", err)
	}

	if _, ok := bp.pages[2]; ok {
		t.Fatalf("最久未用的页 2 应被淘汰, 却仍在池中")
	}
	for _, pid := range []uint32{1, 3, 4} {
		if _, ok := bp.pages[pid]; !ok {
			t.Fatalf("页 %d 应在池中 (淘汰顺序错误)", pid)
		}
	}

	// 淘汰后 pool 长度应等于 capacity (3)
	if bp.Len() != len(bp.pages) {
		t.Fatalf("pool.Len() (%d) 与页表 (%d) 长度不一致", bp.Len(), len(bp.pages))
	}

	// 当前时间戳序: 页 3 最久未用 → 读页 2 (miss, 重新入池) 应淘汰最冷的页 3
	cr.n = 0
	if err := bp.ReadPage(2, make([]byte, 1)); err != nil {
		t.Fatalf("ReadPage(2): %v", err)
	}
	if cr.n != 1 {
		t.Fatalf("被淘汰的页 2 应重新走磁盘 1 次, 实际 %d 次", cr.n)
	}
	if _, ok := bp.pages[3]; ok {
		t.Fatalf("页 3 现在是最冷的, 应被页 2 的入池挤掉")
	}
	if err := bp.ReadPage(1, make([]byte, 1)); err != nil {
		t.Fatalf("ReadPage(1): %v", err)
	}
	if cr.n != 1 {
		t.Fatalf("热页 1 应命中缓存, 实际磁盘读 %d 次", cr.n)
	}
}

// TestReadPageNeverOverwritesConcurrentDirty 读盘期间另一线程写入了同一页时,
// ReadPage 必须以池内 (更新) 内容为准 —— 绝不能用更旧的读盘结果覆盖脏页。
func TestReadPageNeverOverwritesConcurrentDirty(t *testing.T) {
	d := &diskStub{pages: map[uint32][]byte{7: {0x07}}}
	started := make(chan struct{}, 1)
	release := make(chan struct{})
	bp := NewBufferPool(8, func(pageID uint32, buf []byte) error {
		started <- struct{}{}
		<-release // 卡住读盘, 制造并发窗口
		return d.read(pageID, buf)
	}, d.write)

	out := make([]byte, 1)
	done := make(chan error, 1)
	go func() { done <- bp.ReadPage(7, out) }()

	<-started // 读盘进行中 (页 7 尚未入池)
	// 并发写入同页 (单写入者模型下不该发生, 但缓冲池是并发的最后防线)
	if err := bp.WritePage(7, []byte{0x99}); err != nil {
		t.Fatalf("WritePage(7): %v", err)
	}
	close(release)
	if err := <-done; err != nil {
		t.Fatalf("ReadPage(7): %v", err)
	}

	if out[0] != 0x99 {
		t.Fatalf("读到的是旧读盘内容 %#x, 池内新内容 0x99 被覆盖了", out[0])
	}
	if bp.DirtyCount() != 1 {
		t.Fatalf("页 7 必须保持脏, 实际脏页 %d", bp.DirtyCount())
	}
	// 再读一次: 必须命中且仍是新内容
	if err := bp.ReadPage(7, out); err != nil {
		t.Fatalf("重读: %v", err)
	}
	if out[0] != 0x99 {
		t.Fatalf("重读拿到 %#x", out[0])
	}
}

// TestWriteOverReadCachedPageKeepsLRUConsistent 读缓存的干净页被 WritePage 改脏后,
// 条目原地复用: lru 链表与页表保持一致, 不泄漏孤儿节点。
func TestWriteOverReadCachedPageKeepsLRUConsistent(t *testing.T) {
	d := &diskStub{pages: map[uint32][]byte{5: {0x05}}}
	bp := NewBufferPool(8, d.read, d.write)

	// 读入池 (干净页)
	if err := bp.ReadPage(5, make([]byte, 1)); err != nil {
		t.Fatalf("ReadPage(5): %v", err)
	}
	// 改脏
	if err := bp.WritePage(5, []byte{0x55}); err != nil {
		t.Fatalf("WritePage(5): %v", err)
	}

	if bp.DirtyCount() != 1 {
		t.Fatalf("改脏后脏页应计 1, 实际 %d", bp.DirtyCount())
	}
	if bp.Len() != len(bp.pages) {
		t.Fatalf("pool.Len() (%d) 与页表 (%d) 长度不一致", bp.Len(), len(bp.pages))
	}
	// 淘汰路径上它现在是脏页: 容量满时不得被淘汰
	bp.UpdatePage(100, []byte{1})
	if err := bp.WritePage(6, []byte{6}); err != nil {
		t.Fatalf("WritePage(6): %v", err)
	}
	if _, ok := bp.pages[5]; !ok {
		t.Fatalf("脏页 5 绝不能被容量淘汰")
	}
}

// TestReadPageConcurrentHot 验证多个 goroutine 同时读热页真的并行。
// mu 从 sync.Mutex 改成 sync.RWMutex + atomic touch 后, cache-hit 路径
// 不应该互相阻塞。8 个 goroutine 并发读同一页应该 ≈ 单次耗时, 而不是 8×。
//
// 注意: 纯无工作量的 RLock+atomic 是纳秒级, 在 Windows Go timer 精度 (~15ms)
// 下 baseline 和 elapsed 会同时被 round 成 0, 加速比算 NaN 或 0.00x —— 这不是
// "并发没生效", 是 timer 没东西可量。所以这里给每次 ReadPage 加一段 workload,
// 让每次访问约 50µs (和 TestBufferPoolHotConcurrencyWithWork 同套路)。
func TestReadPageConcurrentHot(t *testing.T) {
	d := &diskStub{pages: map[uint32][]byte{42: {0x2A}}}
	bp := NewBufferPool(64, d.read, d.write)

	// 先把页 42 读入池 (让它变热)
	buf := make([]byte, 1)
	if err := bp.ReadPage(42, buf); err != nil {
		t.Fatalf("warmup: %v", err)
	}

	const perRead = 50 * time.Microsecond // 每次读后睡一下, 让 timer 有东西量
	readers := 8
	baselineN := 20
	var baseline time.Duration
	for i := 0; i < baselineN; i++ {
		start := time.Now()
		_ = bp.ReadPage(42, buf)
		time.Sleep(perRead)
		baseline += time.Since(start)
	}
	baseline /= time.Duration(baselineN)

	start := time.Now()
	var wg sync.WaitGroup
	wg.Add(readers)
	errs := make([]error, readers)
	for i := 0; i < readers; i++ {
		go func(idx int) {
			defer wg.Done()
			out := make([]byte, 1)
			errs[idx] = bp.ReadPage(42, out)
			time.Sleep(perRead)
		}(i)
	}
	wg.Wait()
	elapsed := time.Since(start)

	for i, e := range errs {
		if e != nil {
			t.Fatalf("reader %d error: %v", i, e)
		}
	}

	speedup := float64(baseline*time.Duration(readers)) / float64(elapsed)
	t.Logf("单 ReadPage baseline (含 %v workload): %v", perRead, baseline)
	t.Logf("8 并发读同一热页: %v", elapsed)
	t.Logf("串行预估 (baseline × 8): %v", baseline*time.Duration(readers))
	t.Logf("加速比: %.2fx (并行 ≈ %.1f× 单次)", speedup, float64(elapsed)/float64(baseline))

	// RWMutex 的 cache-hit 路径应该真并行 —— 8 并发耗时不应是串行的 8 倍,
	// 至少小于串行预估的一半 (speedup >= 2.0)
	if elapsed >= baseline*time.Duration(readers) {
		t.Errorf("读没有真的并行! 耗时 %v >= 串行预估 %v (speedup=%.2f)", elapsed, baseline*time.Duration(readers), speedup)
	} else {
		t.Logf("ReadPage cache-hit 路径真正并行!")
	}
}

// TestBufferPoolHotConcurrencyDirect 直接测 BufferPool 热缓存并发读 —— 绕过 TsDB2 上层
// 模拟 8 个 goroutine 并发 ReadPage 不同页 + 同页 (热缓存, 应该全是 RLock)。
//
// 和 TestReadPageConcurrentHot 同套路: 纯 RLock+atomic 是纳秒级, 在 Windows timer
// 精度 (~15ms) 下 baseline/elapsed 会被 round 成 0 —— 加 workload 让 timer 有东西量。
func TestBufferPoolHotConcurrencyDirect(t *testing.T) {
	// 模拟: 先预热 50 页 (相当于 B+ 树节点 + 数据页全进池)
	d := &diskStub{pages: map[uint32][]byte{}}
	for i := uint32(1); i <= 50; i++ {
		d.pages[i] = []byte{byte(i)}
	}
	bp := NewBufferPool(100, d.read, d.write)

	// 预热: 把 50 页全读进池
	buf := make([]byte, 1)
	for i := uint32(1); i <= 50; i++ {
		_ = bp.ReadPage(i, buf)
	}
	t.Logf("预热后 BufferPool Len=%d, DirtyCount=%d", bp.Len(), bp.DirtyCount())

	const pageWork = 50 * time.Microsecond // 每页 50µs workload, 让 timer 有东西量
	const pagesPerScan = 50

	// baseline: 单 goroutine 读 50 页 + 每页 workload
	baselineN := 20
	var baseline time.Duration
	for i := 0; i < baselineN; i++ {
		start := time.Now()
		for p := uint32(1); p <= pagesPerScan; p++ {
			_ = bp.ReadPage(p, buf)
			time.Sleep(pageWork)
		}
		baseline += time.Since(start)
	}
	baseline /= time.Duration(baselineN)
	t.Logf("单 goroutine 读 %d 页 baseline (含 workload): %v", pagesPerScan, baseline)

	// 8 goroutine 并发: 每个读 50 页 + 每页 workload
	readers := 8
	start := time.Now()
	var wg sync.WaitGroup
	wg.Add(readers)
	errs := make([]error, readers)
	for i := 0; i < readers; i++ {
		go func(idx int) {
			defer wg.Done()
			out := make([]byte, 1)
			for p := uint32(1); p <= pagesPerScan; p++ {
				if e := bp.ReadPage(p, out); e != nil {
					errs[idx] = e
					return
				}
				time.Sleep(pageWork)
			}
		}(i)
	}
	wg.Wait()
	elapsed := time.Since(start)

	for i, e := range errs {
		if e != nil {
			t.Fatalf("reader %d error: %v", i, e)
		}
	}

	speedup := float64(baseline*time.Duration(readers)) / float64(elapsed)
	t.Logf("8 并发各读 %d 页: %v", pagesPerScan, elapsed)
	t.Logf("串行预估: %v", baseline*time.Duration(readers))
	t.Logf("加速比: %.2fx", speedup)

	if elapsed >= baseline*time.Duration(readers) {
		t.Errorf("BufferPool 热缓存没真并行! 耗时 %v >= 串行预估 %v (speedup=%.2f)", elapsed, baseline*time.Duration(readers), speedup)
	} else {
		t.Logf("BufferPool 热缓存并发加速比 %.2fx —— RLock 真并行!", speedup)
	}
}

// TestBufferPoolHotConcurrencyWithWork 测 BufferPool 热缓存 + CPU 密集工作的并发度
// 在 ReadPage cache-hit 路径加 sleep 模拟 B+ 树遍历的 CPU 工作
func TestBufferPoolHotConcurrencyWithWork(t *testing.T) {
	d := &diskStub{pages: map[uint32][]byte{}}
	for i := uint32(1); i <= 50; i++ {
		d.pages[i] = []byte{byte(i)}
	}
	bp := NewBufferPool(100, d.read, d.write)

	// 预热
	buf := make([]byte, 1)
	for i := uint32(1); i <= 50; i++ {
		_ = bp.ReadPage(i, buf)
	}

	// baseline: 单 goroutine 读 50 页 + 每页 sleep 50μs (模拟 CPU 工作)
	readers := 8
	const pageWork = 50 * time.Microsecond // 50μs per page (模拟 B+ 树节点处理)
	const pagesPerScan = 50

	baselineN := 20
	var baseline time.Duration
	for i := 0; i < baselineN; i++ {
		start := time.Now()
		for p := uint32(1); p <= pagesPerScan; p++ {
			_ = bp.ReadPage(p, buf)
			time.Sleep(pageWork) // 模拟 CPU 工作
		}
		baseline += time.Since(start)
	}
	baseline /= time.Duration(baselineN)
	t.Logf("单 goroutine 读 %d 页 + %v/页 sleep: baseline %v", pagesPerScan, pageWork, baseline)

	// 8 goroutine 并发
	start := time.Now()
	var wg sync.WaitGroup
	wg.Add(readers)
	errs := make([]error, readers)
	for i := 0; i < readers; i++ {
		go func(idx int) {
			defer wg.Done()
			out := make([]byte, 1)
			for p := uint32(1); p <= pagesPerScan; p++ {
				if e := bp.ReadPage(p, out); e != nil {
					errs[idx] = e
					return
				}
				time.Sleep(pageWork)
			}
		}(i)
	}
	wg.Wait()
	elapsed := time.Since(start)

	for i, e := range errs {
		if e != nil {
			t.Fatalf("reader %d error: %v", i, e)
		}
	}

	speedup := float64(baseline*time.Duration(readers)) / float64(elapsed)
	t.Logf("8 并发耗时: %v", elapsed)
	t.Logf("串行预估: %v", baseline*time.Duration(readers))
	t.Logf("加速比: %.2fx", speedup)

	if speedup < 2.0 {
		t.Errorf("并发加速比 %.2fx < 2x! BufferPool RLock 可能没真并行", speedup)
	} else {
		t.Logf("BufferPool RLock 真并行! 加速比 %.2fx", speedup)
	}
}

// TestEvictBurstAfterCommit 采样淘汰的正确性: Commit 把一批脏页变干净留在池里,
// 池子超出容量; 下一次写入必须在一轮内把容量压回来, 且被丢的只能是干净页。
// (这正是旧实现 O(n×淘汰数) 退化的场景 —— 500 次淘汰 × 全表扫; 采样后 O(K×淘汰数))
func TestEvictBurstAfterCommit(t *testing.T) {
	d := &diskStub{pages: map[uint32][]byte{}}
	cr := &countingRead{d: d}
	const poolCap = 1000
	bp := NewBufferPool(poolCap, cr.read, d.write)

	// 塞 1500 张脏页: 全脏时允许临时超容量, 所以能超过 1000
	for i := 1; i <= 1500; i++ {
		if err := bp.WritePage(uint32(i), []byte{byte(i)}); err != nil {
			t.Fatalf("WritePage(%d): %v", i, err)
		}
	}
	if got := bp.Len(); got != 1500 {
		t.Fatalf("全脏时应允许超容量, 实际 %d 页", got)
	}

	bp.ClearAll() // 模拟 Commit 后脏页变干净、留在池里

	// 下一次写入: WritePage 先淘汰再插入 —— 一轮把 1500 压到 999,
	// 插入新页后回到恰 capacity (这是池的稳态不变量)
	if err := bp.WritePage(2000, []byte{0x7f}); err != nil {
		t.Fatalf("触发淘汰的 WritePage: %v", err)
	}
	if got := bp.Len(); got != poolCap {
		t.Fatalf("淘汰+插入后应恰为容量 %d, 实际 %d", poolCap, got)
	}
	// 触发写入的那页必须还在
	if _, ok := bp.pages[2000]; !ok {
		t.Fatalf("刚写入的页 2000 不应被淘汰")
	}
	// 剩下的页都能读回 —— 没有脏页被静默丢掉
	for pid := range bp.pages {
		if err := bp.ReadPage(pid, make([]byte, 1)); err != nil {
			t.Fatalf("读回页 %d: %v", pid, err)
		}
	}
}

// TestEvictSkipsDirtyPages 样本里的脏页绝不被淘汰: 池满时若没有干净候选,
// 宁可临时超容量也不丢未落盘的修改。
func TestEvictSkipsDirtyPages(t *testing.T) {
	d := &diskStub{pages: map[uint32][]byte{}}
	cr := &countingRead{d: d}
	bp := NewBufferPool(4, cr.read, d.write)
	for i := 1; i <= 4; i++ {
		if err := bp.WritePage(uint32(i), []byte{byte(i)}); err != nil {
			t.Fatal(err)
		}
	}
	// 4 张全是脏页, 再写一张: 样本抽到的全是脏页 → 不做淘汰, 允许临时超容量
	if err := bp.WritePage(5, []byte{5}); err != nil {
		t.Fatal(err)
	}
	if got := bp.Len(); got != 5 {
		t.Fatalf("全脏时应允许临时超容量 (5 页), 实际 %d", got)
	}
}

// BenchmarkEvictBurst 一轮淘汰 ~2000 页的成本 (白盒直接测淘汰循环):
// 6000 页干净池 / 容量 4000, evictCleanForRoomLocked 一轮要淘汰 2001 页。
// 全表精确最小 = 2001 × 6000 ≈ 1200 万次条目扫描 (~0.5-1s);
// 采样 32 = 2001 × 32 ≈ 6.4 万次 (~0.3-0.6ms) —— 与池大小脱钩。
func BenchmarkEvictBurst(b *testing.B) {
	d := &diskStub{pages: map[uint32][]byte{}}
	cr := &countingRead{d: d}
	bp := NewBufferPool(4000, cr.read, d.write)
	fillN := 1000000 // 单调发号: 用 len 当页号会在第二轮与幸存页碰撞 (死循环), 必须独立递增
	fill := func(n int) {
		bp.mu.Lock()
		defer bp.mu.Unlock()
		for len(bp.pages) < n {
			pid := uint32(fillN)
			fillN++
			e := &pageEntry{data: []byte{1}, pageLen: 1}
			bp.stampLocked(e)
			bp.pages[pid] = e
			bp.touchLocked(e)
		}
	}
	fill(6000)
	b.ResetTimer()
	for n := 0; n < b.N; n++ {
		fill(6000) // 每轮先把池子重新撑到 6000, 再测一轮 2001 次采样淘汰
		bp.mu.Lock()
		bp.evictCleanForRoomLocked()
		bp.mu.Unlock()
	}
}