TsDB2/DB/core/wal_page_rest_test.go

package core

// =============================================================================
// WALOpPageRest 补尾记录回归测试
//
// 背景 (demo/wal_crash_chaos 随机断电压测实锤): 整页镜像按 useSize 截断存 WAL,
// 而 DataPage 的槽位目录在页尾 (useSize 之外) —— 崩溃恢复零填充把目录抹掉,
// 页上全部行经槽位不可达 (TotalRows 与实际可达行数分叉)。Heap/Blob 满页同理
// 会被"单条容量上限"截掉真实数据。修复 = FullPage + PageRest 成对入链, 恢复端
// 按显式偏移覆盖页尾。
// =============================================================================

import (
	"bytes"
	"encoding/binary"
	"testing"

	"git.ycl.pub/yl32940/TsDB2/DB/core/index/datapage"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
)

// stubHandleDataPage 内存态数据页: NewPage 自增分配, SavePage 存回 map。
// 用来造一张带真实槽位目录的 DataPage 镜像, 喂给 WALWriter.packEntries。
func stubHandleDataPage(t *testing.T, rows int) (full []byte, useSize, dirStart uint32) {
	t.Helper()
	pages := map[uint32][]byte{}
	next := uint32(1000)
	h := page.HandleDb{
		NewPage: func() uint32 { p := next; next++; return p },
		SavePage: func(pid uint32, buf []byte) error {
			b := make([]byte, len(buf))
			copy(b, buf)
			pages[pid] = b
			return nil
		},
		ReadPage: func(pid uint32, buf []byte) error {
			copy(buf, pages[pid])
			return nil
		},
	}
	dp := datapage.New(h)
	for i := 0; i < rows; i++ {
		row := make([]byte, 160) // 行: BE 键 + 填充
		binary.BigEndian.PutUint64(row[:8], uint64(295+i))
		if dp.AppendRowSlot(row) == 0 {
			t.Fatalf("AppendRowSlot 第 %d 行失败", i)
		}
	}
	dp.Save()
	full = pages[dp.PageID()]
	useSize = binary.LittleEndian.Uint32(full[16:])
	ds, ok := datapage.SlotDirStart(full)
	if !ok {
		t.Fatalf("造页失败: 槽位目录无效")
	}
	return full, useSize, ds
}

// TestPackEntriesDataPageTail 截断的 DataPage 必须展开成 FullPage+PageRest,
// 且补尾的显式偏移 = 槽位目录起始, 内容 = 目录区原文。
func TestPackEntriesDataPageTail(t *testing.T) {
	full, useSize, ds := stubHandleDataPage(t, 21)
	if ds <= useSize {
		t.Fatalf("前置条件: 目录应在 useSize 之外 (useSize=%d ds=%d)", useSize, ds)
	}

	w := &WALWriter{pageIndex: map[uint32]int{}}
	w.RecordPage(9, full)
	packed := w.packEntries()
	if len(packed) != 2 {
		t.Fatalf("应展开成 2 条 (FullPage+PageRest), 实际 %d", len(packed))
	}
	if packed[0].opType != WALOpFullPage || uint32(len(packed[0].payload)) != useSize {
		t.Fatalf("base: op=%d len=%d, 想要 FullPage/%d", packed[0].opType, len(packed[0].payload), useSize)
	}
	if !bytes.Equal(packed[0].payload, full[:useSize]) {
		t.Fatalf("base 内容 ≠ 页有效前缀")
	}
	rest := packed[1]
	if rest.opType != WALOpPageRest || rest.pageID != 9 {
		t.Fatalf("补尾: op=%d page=%d, 想要 PageRest/9", rest.opType, rest.pageID)
	}
	off := binary.LittleEndian.Uint32(rest.payload[:4])
	if off != ds {
		t.Fatalf("补尾偏移=%d, 想要目录起始 %d", off, ds)
	}
	if !bytes.Equal(rest.payload[4:], full[ds:]) {
		t.Fatalf("补尾内容 ≠ 目录区原文")
	}
}

// TestPackEntriesCapAndSkips 容量上限截掉的尾数据必须补 (Heap 满页);
// 未被截断的非数据页 (B+ 树节点) 与头页不产生补尾。
func TestPackEntriesCapAndSkips(t *testing.T) {
	// Heap 满页: useSize=16384 > 单条上限 → base 截到 16331, 补 [16331, 16384)
	heap := make([]byte, PageSize)
	page.HdrOf(heap).Type = page.HeapIndex
	page.HdrOf(heap).UseSize = PageSize
	w := &WALWriter{pageIndex: map[uint32]int{}}
	w.RecordPage(8, heap)
	packed := w.packEntries()
	if len(packed) != 2 {
		t.Fatalf("Heap 满页应产生补尾, 实际 %d 条", len(packed))
	}
	if uint32(len(packed[0].payload)) != PageSize-page.WALPageHdrSize-page.WALEntryMeta {
		t.Fatalf("base 应截到容量上限, 实际 %d", len(packed[0].payload))
	}
	if off := binary.LittleEndian.Uint32(packed[1].payload[:4]); off != uint32(len(packed[0].payload)) {
		t.Fatalf("补尾偏移=%d, 想要 %d", off, len(packed[0].payload))
	}
	if uint32(len(packed[1].payload)) != 4+(PageSize-16331) {
		t.Fatalf("补尾长度=%d, 想要 %d", len(packed[1].payload), 4+(PageSize-16331))
	}

	// B+ 树节点 (useSize 远小于上限): useSize 内完整入链, 无补尾
	bt := make([]byte, PageSize)
	page.HdrOf(bt).Type = page.IndexDataIndex
	page.HdrOf(bt).UseSize = 4442
	w2 := &WALWriter{pageIndex: map[uint32]int{}}
	w2.RecordPage(7, bt)
	if p2 := w2.packEntries(); len(p2) != 1 {
		t.Fatalf("B+ 节点不应产生补尾, 实际 %d 条", len(p2))
	}

	// 头页镜像 (4KB): 无补尾
	hdr := make([]byte, PageSize4K)
	w3 := &WALWriter{pageIndex: map[uint32]int{}}
	w3.RecordPage(0, hdr)
	if p3 := w3.packEntries(); len(p3) != 1 {
		t.Fatalf("头页镜像不应产生补尾, 实际 %d 条", len(p3))
	}
}

// TestRecoverWALPageRestRoundTrip 恢复端: FullPage+PageRest 重放后, 整页字节
// (含 useSize 之外的槽位目录) 必须与原始镜像逐字节一致 —— 这是压测抓到的
// 丢数据路径的直接回归。
func TestRecoverWALPageRestRoundTrip(t *testing.T) {
	path, ff := walHardeningDB(t)
	walPID := nextFreePage(t, ff)
	targetPID := walPID + 5 // 不与 WAL 首页相撞

	full, useSize, ds := stubHandleDataPage(t, 21)
	base := encodeWALEntry(targetPID, WALOpFullPage, full[:useSize], 0)
	restPayload := make([]byte, 4+PageSize-ds)
	binary.LittleEndian.PutUint32(restPayload, ds)
	copy(restPayload[4:], full[ds:])
	rest := encodeWALEntry(targetPID, WALOpPageRest, restPayload, 0)

	first := buildWALFirstPage(t, walPID, 0, 2, base, rest)
	publishWALChain(t, ff, first)
	_ = ff.Close()

	db, err := Open(path)
	if err != nil {
		t.Fatalf("重开 (重放 FullPage+PageRest): %v", err)
	}
	_ = db.Close()

	// 直读目标页, 与原始镜像逐字节比对 (重点是 useSize 之外的目录区)
	ff2, err := OpenFile(path)
	if err != nil {
		t.Fatalf("重开裸文件: %v", err)
	}
	defer ff2.Close()
	got := make([]byte, PageSize)
	if err := ff2.readPageDisk(targetPID, got); err != nil {
		t.Fatalf("读目标页: %v", err)
	}
	if !bytes.Equal(got, full) {
		t.Fatalf("重放后页 %d 与原始镜像不一致 (useSize=%d, 目录起始=%d)", targetPID, useSize, ds)
	}
}

// TestRecoverWALOrphanPageRestRejected 孤儿补尾 (没有同页前置整页镜像) 的已提交链
// 必须拒绝重放并保留 WAL —— 补尾只有叠在 base 写回的页上才有意义, 单独出现
// 说明链损坏/被篡改, 部分重放会把零填充页当真实数据。
func TestRecoverWALOrphanPageRestRejected(t *testing.T) {
	path, ff := walHardeningDB(t)
	walPID := nextFreePage(t, ff)

	restPayload := make([]byte, 8)
	binary.LittleEndian.PutUint32(restPayload, 100)
	rest := encodeWALEntry(5, WALOpPageRest, restPayload, 0)
	first := buildWALFirstPage(t, walPID, 0, 1, rest)
	publishWALChain(t, ff, first)
	_ = ff.Close()

	db, err := Open(path)
	if err == nil {
		_ = db.Close()
		t.Fatalf("孤儿补尾的已提交链必须让 Open 返回错误, 却成功了")
	}
	t.Logf("正确拒绝: %v", err)

	ff2, err := OpenFile(path)
	if err != nil {
		t.Fatalf("重开裸文件: %v", err)
	}
	hdr, err := ff2.ReadHeader()
	_ = ff2.Close()
	if err != nil {
		t.Fatalf("读头页: %v", err)
	}
	if hdr.FirstWALPage != walPID {
		t.Fatalf("WAL 链应原样保留: FirstWALPage=%d, 想要 %d", hdr.FirstWALPage, walPID)
	}
}