TsDB2/DB/core/wal_recovery.go

package core

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

// =============================================================================
// 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:
//
//	WALOpFullPage (0): 整页镜像, 直接写回实际页
//	Insert/Delete/Update: 行级记录, 用 applyRowToPage 应用到页
func RecoverWAL(ff *File, header *HeaderPage) error {
	if header.FirstWALPage == 0 {
		return nil // 无待恢复 WAL
	}

	// 读取首页 (直接读盘: 避免读到 BufferPool 里的陈旧缓存)
	firstBuf := make([]byte, PageSize)
	if err := ff.readPageDisk(header.FirstWALPage, firstBuf); err != nil {
		return err
	}

	// 解析首页元数据
	firstHdr := page.WalHdrOf(firstBuf)
	totalEntries := firstHdr.TotalEntries
	committed := firstHdr.State

	// 收集所有 entry
	type recoveryEntry struct {
		pageID   uint32
		data     []byte // 整页镜像 (旧格式)
		isRow    bool   // 是否为行级记录 (新格式)
		opType   byte   // 行级操作类型
		storeRow []byte // 行级数据
	}
	var entries []recoveryEntry

	pid := header.FirstWALPage
	remaining := int(totalEntries)
	walTail := uint32(0) // 已遍历到的最后一页 (链正常读完时即真实链尾)
	walPages := uint32(0)

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

		hdr := page.HdrOf(buf)
		if hdr.Type != page.WALIndex {
			break
		}

		walTail = pid
		walPages++

		// 首页与续页头部等长 —— 条目区起点与 EntryCount 位置一致, 无需按 isFirst 分叉
		walHdr := page.WalHdrOf(buf)
		off := page.WALPageHdrSize
		entryCount := walHdr.EntryCount
		for i := uint16(0); i < entryCount && remaining > 0; i++ {
			if off+page.WALEntryMeta > PageSize {
				break
			}
			targetPID := binary.LittleEndian.Uint32(buf[off:])
			dataLen := binary.LittleEndian.Uint32(buf[off+4:])
			checksum := binary.LittleEndian.Uint32(buf[off+8:])
			opType := buf[off+12]

			if off+page.WALEntryMeta+dataLen > PageSize {
				break
			}
			payload := buf[off+page.WALEntryMeta : off+page.WALEntryMeta+dataLen]

			// CRC32 覆盖 OpType + Data
			csBuf := make([]byte, 1+dataLen)
			csBuf[0] = opType
			copy(csBuf[1:], payload)
			if crc32.ChecksumIEEE(csBuf) != checksum {
				// 校验失败: 后续内容不可信, 停止收集
				remaining = 0
				break
			}

			if opType == WALOpFullPage {
				// 整页镜像: 零填充到整页长度, 保证写回时文件偏移一致
				data := make([]byte, PageSize)
				copy(data, payload)
				entries = append(entries, recoveryEntry{pageID: targetPID, data: data})
			} else {
				storeRow := make([]byte, dataLen)
				copy(storeRow, payload)
				entries = append(entries, recoveryEntry{
					pageID:   targetPID,
					isRow:    true,
					opType:   opType,
					storeRow: storeRow,
				})
			}

			off += page.WALEntryMeta + dataLen
			remaining--
		}

		pid = readNextWalPage(buf)
	}

	// 根据 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)
		//
		// 去重: 若同一页已有整页镜像 (Phase 1 已完整恢复), 则跳过它的行级条目。
		// 否则整页镜像里的新行 + 行级 INSERT 会被应用两次, 造成行重复。
		fullPageApplied := make(map[uint32]bool, len(entries))
		for _, e := range entries {
			if !e.isRow {
				fullPageApplied[e.pageID] = true
			}
		}
		for _, e := range entries {
			if !e.isRow || fullPageApplied[e.pageID] {
				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 链
	//
	// 正常路径: 上面的收集循环正好把链读完 (pid == 0), 此时 walTail / walPages 即真实
	// 链尾 / 链长 —— 一次拼接 O(1) 还链, 与链长无关。
	// 异常路径 (CRC 失败或页类型不符而提前 break, pid != 0): 真实链尾未知, 退化为逐页回收,
	// 既不会漏页, 也避免把"半截链"拼进空闲链 —— 那截链尾的 NextPage 仍指向后继页, 会让
	// 空闲链遍历窜进未回收的 WAL 页, 造成链错乱。
	if pid == 0 && walPages > 0 {
		if err := recycleChainWithCache(ff, header, header.FirstWALPage, walTail, walPages); err != nil {
			return err
		}
	} else {
		walPID := header.FirstWALPage
		for walPID != 0 {
			buf := make([]byte, PageSize)
			if err := ff.readPageDisk(walPID, buf); err != nil {
				break
			}
			nextPID := readNextWalPage(buf)
			header.recyclePage(walPID, ff.WritePageDirect)
			walPID = nextPID
		}
	}

	ff.withHeader(func() {
		header.FirstWALPage = 0
		header.LastWALPage = 0
	})
	if err := ff.WriteHeader(header); err != nil {
		return err
	}
	// 恢复收尾 fsync: 保证已重放的真实页与"WAL 已清空"的 Header 一起落盘,
	// 否则恢复过程本身再断电会重复重放
	return ff.sync()
}

// readNextWalPage 读取 WAL 页的链指针 (统一存放在 PageHdr.NextPage, 与空闲链同字段)
func readNextWalPage(buf []byte) uint32 {
	return page.HdrOf(buf).NextPage
}

// 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
}