TsDB2/DB/core/wal_recovery.go

package core

import (
	"TsDB2/DB/core/page"
	"encoding/binary"
	"hash/crc32"
	"unsafe"
)

// =============================================================================
// WAL 崩溃恢复
//
// 在 DB.Open 时调用 RecoverWAL:
//  1. 从 FirstWALPage 遍历 WAL 链
//  2. 读首页: TxnID + TotalEntries + Committed
//  3. 收集全部 Entry (按 TotalEntries 计数)
//  4. 每条 Entry 验证 CRC32
//  5. Committed==1 -> 重放所有页镜像到实际位置
//  6. Committed==0 -> 丢弃 (事务未完成)
//  7. 回收所有 WAL 页, 重置 HeaderPage
// =============================================================================

// RecoverWAL 恢复未完成的 WAL 事务
func RecoverWAL(ff *File, header *HeaderPage) error {
	if header.FirstWALPage == 0 {
		return nil // 无待恢复 WAL
	}

	// 读取首页
	firstBuf := make([]byte, PageSize)
	if err := ff.ReadPage(header.FirstWALPage, firstBuf); err != nil {
		return err
	}

	// 解析首页元数据
	_ = binary.LittleEndian.Uint64(firstBuf[walPHDR:]) // TxnID (用于调试)
	totalEntries := binary.LittleEndian.Uint32(firstBuf[walPHDR+8:])
	committed := firstBuf[walPHDR+12]

	// 收集所有 entry
	type recoveryEntry struct {
		pageID uint32
		data   []byte
	}
	var entries []recoveryEntry

	pid := header.FirstWALPage
	remaining := int(totalEntries)

	for pid != 0 && remaining > 0 {
		buf := make([]byte, PageSize)
		if err := ff.ReadPage(pid, buf); err != nil {
			break
		}

		// 验证页类型
		hdr := (*page.PageHdr)(unsafe.Pointer(&buf[0]))
		if hdr.Type != page.WALIndex {
			break
		}

		// 确定条目起始偏移
		var off uint32
		if pid == header.FirstWALPage {
			off = walFirstHdr // 首页
		} else {
			off = walContHdr // 续页
		}

		// 读取本页的 entry
		entryCount := readEntryCount(buf, pid == header.FirstWALPage)
		for i := uint16(0); i < entryCount && remaining > 0; i++ {
			if off+walEntryMeta > PageSize {
				break
			}
			targetPID := binary.LittleEndian.Uint32(buf[off:])
			dataLen := binary.LittleEndian.Uint32(buf[off+4:])
			checksum := binary.LittleEndian.Uint32(buf[off+8:])

			if off+walEntryMeta+dataLen > PageSize {
				break
			}

			data := make([]byte, PageSize)         // 完整页缓冲区, 未用部分零填充
			copy(data, buf[off+12:off+12+dataLen]) // 只读 dataLen 字节

			// 验证 CRC32 (只校验实际存储的 dataLen 字节, 与写入时一致)
			if crc32.ChecksumIEEE(data[:dataLen]) != checksum {
				// 校验失败, 停止恢复 (数据可能损坏)
				remaining = 0
				break
			}

			entries = append(entries, recoveryEntry{targetPID, data})
			off += walEntryMeta + dataLen
			remaining--
		}

		// 下一个 WAL 页
		pid = readNextWalPage(buf, pid == header.FirstWALPage)
	}

	// 根据 Committed 标记决定是否重放
	if committed == 1 && len(entries) > 0 {
		// 事务已提交但实际页可能未全部写入 -> 重放补齐
		for _, e := range entries {
			if e.pageID == 0 {
				// HeaderPage 特殊处理: 直接写回
				if err := recoverHeader(ff, e.data); err != nil {
					return err
				}
			} else {
				if err := ff.WritePageDirect(e.pageID, e.data); err != nil {
					return err
				}
			}
		}
	}
	// committed == 0: 事务未完成, 丢弃所有 entry (不回写)

	// 回收所有 WAL 页
	walPID := header.FirstWALPage
	for walPID != 0 {
		buf := make([]byte, PageSize)
		if err := ff.ReadPage(walPID, buf); err != nil {
			break
		}
		isFirst := walPID == header.FirstWALPage
		nextPID := readNextWalPage(buf, isFirst)
		header.recyclePage(walPID, ff.WritePageDirect)
		walPID = nextPID
	}

	// 重置 WAL 指针
	header.FirstWALPage = 0
	header.LastWALPage = 0
	return ff.WriteHeader(header)
}

// readEntryCount 读取 WAL 页的 EntryCount
func readEntryCount(buf []byte, isFirst bool) uint16 {
	if isFirst {
		return binary.LittleEndian.Uint16(buf[walPHDR+17:])
	}
	return binary.LittleEndian.Uint16(buf[walPHDR+4:])
}

// readNextWalPage 读取 WAL 页的 NextWalPage
func readNextWalPage(buf []byte, isFirst bool) uint32 {
	if isFirst {
		return binary.LittleEndian.Uint32(buf[walPHDR+13:])
	}
	return binary.LittleEndian.Uint32(buf[walPHDR:])
}

// recoverHeader 恢复 HeaderPage
func recoverHeader(ff *File, data []byte) error {
	// data 是完整的 16KB 页镜像, 但 HeaderPage 只有 4KB
	// 取前 4KB 写入文件头
	if len(data) < PageSize4K {
		_, err := ff.f.WriteAt(data, 0)
		return err
	}
	_, err := ff.f.WriteAt(data[:PageSize4K], 0)
	return err
}