TsDB2/DB/core/wal_hardening_test.go

package core

import (
	"encoding/binary"
	"errors"
	"hash/crc32"
	"path/filepath"
	"strings"
	"testing"

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

// =============================================================================
// WAL/恢复路径加固测试
//
// 构造方式: 先正常建库并关闭, 再用裸 File 句柄直接改写头页 / 伪造 WAL 页,
// 模拟"损坏现场"后重新 Open —— 断言必须返回错误而不是 panic / 静默吞掉。
// =============================================================================

// walHardeningDB 建一个最小合法库并正常关闭, 然后以裸文件句柄返回,
// 供测试绕过引擎的 WAL 写入逻辑、直接伪造损坏的 WAL 链。
func walHardeningDB(t *testing.T) (string, *File) {
	t.Helper()
	path := filepath.Join(t.TempDir(), "wal_hardening.db")
	db, err := Open(path)
	if err != nil {
		t.Fatalf("建库: %v", err)
	}
	db.SetAutoCommitDelay(0)
	if err := db.Close(); err != nil {
		t.Fatalf("关闭: %v", err)
	}
	ff, err := OpenFile(path)
	if err != nil {
		t.Fatalf("重开裸文件: %v", err)
	}
	return path, ff
}

// encodeWALEntry 按写入端格式 (见 wal.go writeWALPages) 编码一条 WAL 条目:
// [TargetPageID(4B)][DataLen(4B)][Checksum(4B)][OpType(1B)][Data]。
// dataLenOverride 非 0 时改写 DataLen 字段 (模拟损坏), Checksum 仍按页内
// 实际字节计算 —— 让越界检查成为唯一防线, 而不是靠 CRC 失败兜住。
func encodeWALEntry(targetPID uint32, opType byte, payload []byte, dataLenOverride uint32) []byte {
	e := make([]byte, page.WALEntryMeta+len(payload))
	binary.LittleEndian.PutUint32(e[0:], targetPID)
	dataLen := uint32(len(payload))
	if dataLenOverride != 0 {
		dataLen = dataLenOverride
	}
	binary.LittleEndian.PutUint32(e[4:], dataLen)
	cs := crc32.NewIEEE()
	_, _ = cs.Write([]byte{opType})
	_, _ = cs.Write(payload)
	binary.LittleEndian.PutUint32(e[8:], cs.Sum32())
	e[12] = opType
	copy(e[page.WALEntryMeta:], payload)
	return e
}

// buildWALFirstPage 构造一个已提交的 WAL 首页 (State=Committed), 条目区塞入
// 给定的编码后条目, 链指针指向 nextPID。
func buildWALFirstPage(t *testing.T, pid, nextPID uint32, totalEntries uint32, entries ...[]byte) []byte {
	t.Helper()
	buf := make([]byte, PageSize)
	wh := page.WalHdrOf(buf)
	wh.Init(pid, 0)
	wh.TxnID = 99
	wh.TotalEntries = totalEntries
	wh.State = WALStateCommitted
	wh.EntryCount = uint16(len(entries))
	wh.NextPage = nextPID
	off := page.WALPageHdrSize
	for _, e := range entries {
		copy(buf[off:], e)
		off += uint32(len(e))
	}
	return buf
}

// publishWALChain 把伪造的 WAL 首页写盘, 并把头页链指针指过去。
// 页号取头页 NextFreePage —— 空闲链/游标上的页, 不与已活的页 (头页/SchemaExt/
// TablePage) 相撞; 已提交链出错返回时该页不会被回收, 不影响后续断言。
func publishWALChain(t *testing.T, ff *File, firstBuf []byte) uint32 {
	t.Helper()
	hdr, err := ff.ReadHeader()
	if err != nil {
		t.Fatalf("读头页: %v", err)
	}
	pid := hdr.NextFreePage
	if err := ff.WritePageDirect(pid, firstBuf); err != nil {
		t.Fatalf("写 WAL 首页 %d: %v", pid, err)
	}
	hdr.FirstWALPage = pid
	hdr.LastWALPage = pid
	if err := ff.WriteHeader(hdr); err != nil {
		t.Fatalf("写头页: %v", err)
	}
	return pid
}

// nextFreePage 读头页取 NextFreePage (空闲页分配游标) —— 伪造 WAL 页从这里取号,
// 不与已活的页 (头页 / SchemaExt / TablePage) 相撞。
func nextFreePage(t *testing.T, ff *File) uint32 {
	t.Helper()
	hdr, err := ff.ReadHeader()
	if err != nil {
		t.Fatalf("读头页: %v", err)
	}
	return hdr.NextFreePage
}

// TestRecoverWALRejectsOversizedDataLen WAL 条目 dataLen=0xFFFFFFFF 时, 旧实现的
// uint32 越界检查被加法回绕绕过 (40+13+0xFFFFFFFF 回绕成 52 < PageSize), 下一行
// 切片直接 panic。修复后必须返回错误; 已提交链原样保留, 拒绝部分重放 / 回收。
func TestRecoverWALRejectsOversizedDataLen(t *testing.T) {
	path, ff := walHardeningDB(t)
	walPID := nextFreePage(t, ff)

	entry := encodeWALEntry(5, WALOpFullPage, []byte("partial-page-image"), 0xFFFFFFFF)
	first := buildWALFirstPage(t, walPID, 0, 1, entry)
	publishWALChain(t, ff, first)
	_ = ff.Close()

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

	// 保守处理: 错误返回后 WAL 链必须原样保留 (不回收), 重开仍报错
	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)
	}
}

// TestRecoverWALCommittedChainReadFailure 已提交 WAL 链中途页读失败 (链指针指向
// 文件之外的页号) 必须返回错误 —— 旧实现只 break, 残缺事务被当完整事务重放。
func TestRecoverWALCommittedChainReadFailure(t *testing.T) {
	path, ff := walHardeningDB(t)
	walPID := nextFreePage(t, ff)

	entry := encodeWALEntry(5, WALOpFullPage, []byte("page-five-image"), 0)
	// 首页声称有 2 条 / 还有下一页 0x100000, 但文件远没长到那里 —— 读下一页必失败
	first := buildWALFirstPage(t, walPID, 0x100000, 2, entry)
	publishWALChain(t, ff, first)
	_ = ff.Close()

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

// TestRecoverWALCommittedChainBadPageType 已提交 WAL 链中途混入非 WAL 页
// (页类型非法) 必须返回错误 —— 与读失败同属"链中途断裂", 不能静默部分重放。
func TestRecoverWALCommittedChainBadPageType(t *testing.T) {
	path, ff := walHardeningDB(t)
	walPID := nextFreePage(t, ff)
	badPID := walPID + 1

	entry := encodeWALEntry(5, WALOpFullPage, []byte("page-five-image"), 0)
	first := buildWALFirstPage(t, walPID, badPID, 2, entry)
	publishWALChain(t, ff, first)
	// 链的第二页: 全零页 (Type=0, 不是 WALIndex=7)
	if err := ff.WritePageDirect(badPID, make([]byte, PageSize)); err != nil {
		t.Fatalf("写坏页: %v", err)
	}
	_ = ff.Close()

	db, err := Open(path)
	if err == nil {
		_ = db.Close()
		t.Fatalf("已提交链中途页类型非法必须让 Open 返回错误, 却成功了")
	}
	t.Logf("正确拒绝: %v", err)
}

// TestFlushWriteFailureKeepsPagesDirty Flush 写盘失败: 失败页必须保持 dirty、
// 成功的页转干净, 且第一个错误要返回 —— 旧实现 _ = 吞掉错误还无条件全标干净,
// 写失败的页从此被当"已持久化", 淘汰即静默丢数据。
func TestFlushWriteFailureKeepsPagesDirty(t *testing.T) {
	writeErr := errors.New("disk full")
	fail := true
	bp := pool.NewBufferPool(8,
		func(uint32, []byte) error { return nil },
		func(pageID uint32, _ []byte) error {
			if fail && pageID == 2 {
				return writeErr
			}
			return nil
		})

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

	if err := bp.Flush(); !errors.Is(err, writeErr) {
		t.Fatalf("Flush 必须返回第一个写盘错误, 实际 %v", err)
	}
	if bp.DirtyCount() != 1 {
		t.Fatalf("失败页必须保持 dirty, 期望脏页 1, 实际 %d", bp.DirtyCount())
	}

	// 磁盘恢复后重试: 之前失败的页这次必须落盘并转干净
	fail = false
	if err := bp.Flush(); err != nil {
		t.Fatalf("恢复后重试 Flush: %v", err)
	}
	if bp.DirtyCount() != 0 {
		t.Fatalf("重试成功后不应再有脏页, 实际 %d", bp.DirtyCount())
	}
}

// TestBeginTransactionRejectsNestedTxn 已有活跃事务时再 BeginTransaction 必须
// panic: 旧实现直接覆盖 currentTxn, 前一笔事务的 Rollback 会丢弃错误的页集合。
func TestBeginTransactionRejectsNestedTxn(t *testing.T) {
	db := fixOpenDB(t)
	tx := db.BeginTransaction()

	func() {
		defer func() {
			r := recover()
			if r == nil {
				t.Fatalf("嵌套 BeginTransaction 必须 panic, 却没有")
			}
			msg, ok := r.(string)
			if !ok || !strings.Contains(msg, "活跃事务") {
				t.Fatalf("panic 信息应说明已有活跃事务, 实际 %v", r)
			}
			t.Logf("正确拒绝: %v", r)
		}()
		_ = db.BeginTransaction() // 必须 panic
	}()

	if err := tx.Rollback(); err != nil {
		t.Fatalf("rollback: %v", err)
	}
}