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 事务
// 支持两种行级格式:
//
//	新格式 (OpType != 0): 行级记录, 用 applyRowToPage 应用到页
//	旧格式 (OpType == 0): 整页镜像或旧行级格式, 直接写回
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 // 整页镜像 (旧格式)
		isRow    bool   // 是否为行级记录 (新格式)
		opType   byte   // 行级操作类型
		storeRow []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
		}

		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:])

			// 检测新行级格式: OpType 在 off+12, 值为 1/2/3
			potentialOpType := buf[off+12]
			isNewRowFormat := dataLen > 0 &&
				(potentialOpType == WALOpInsert || potentialOpType == WALOpDelete || potentialOpType == WALOpUpdate) &&
				targetPID != 0 // HeaderPage (pageID=0) 不会是行级记录

			if isNewRowFormat {
				// === 新行级格式: [PageID][DataLen][Checksum][OpType(1B)][Data] ===
				if off+walEntryMeta+1 > PageSize {
					break
				}
				opType := buf[off+12]
				rowDataLen := dataLen // dataLen 只是 StoreRow 的长度
				if off+13+rowDataLen > PageSize {
					break
				}

				// CRC32 覆盖 OpType + Data
				csBuf := make([]byte, 1+rowDataLen)
				csBuf[0] = opType
				copy(csBuf[1:], buf[off+13:off+13+rowDataLen])
				if crc32.ChecksumIEEE(csBuf) != checksum {
					remaining = 0
					break
				}

				storeRow := make([]byte, rowDataLen)
				copy(storeRow, buf[off+13:off+13+rowDataLen])

				entries = append(entries, recoveryEntry{
					pageID:   targetPID,
					isRow:    true,
					opType:   opType,
					storeRow: storeRow,
				})
				off += 13 + rowDataLen
			} else {
				// === 旧格式: 整页镜像或旧行级格式 ===
				if off+walEntryMeta+dataLen > PageSize {
					break
				}

				data := make([]byte, PageSize)
				copy(data, buf[off+12:off+12+dataLen])

				if crc32.ChecksumIEEE(data[:dataLen]) != checksum {
					remaining = 0
					break
				}

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

		pid = readNextWalPage(buf, pid == header.FirstWALPage)
	}

	// 根据 Committed 标记决定是否重放
	if committed == WALStateCommitted && len(entries) > 0 {
		// Phase 1: 先处理整页条目 (恢复页结构: B+树节点, IndexMeta, HeaderPage 等)
		for _, e := range entries {
			if e.isRow {
				continue
			}
			if e.pageID == 0 {
				if err := recoverHeader(ff, e.data); err != nil {
					return err
				}
			} else {
				if err := ff.WritePageDirect(e.pageID, e.data); err != nil {
					return err
				}
			}
		}

		// Phase 2: 处理行级条目 (用 applyRowToPage 应用到 DataPage)
		for _, e := range entries {
			if !e.isRow {
				continue
			}
			pageBuf := make([]byte, PageSize)
			if err := ff.readPageDisk(e.pageID, pageBuf); err != nil {
				continue // 页不存在, 跳过
			}
			applyRowToPage(pageBuf, e.opType, e.storeRow)
			if err := ff.WritePageDirect(e.pageID, pageBuf); err != nil {
				return err
			}
		}
	}
	// committed != Committed: 事务未完成, 丢弃所有 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
	}

	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
}