TsDB2/DB/core/wal_recovery.go

package core

import (
	"encoding/binary"
	"fmt"
	"hash/crc32"

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

// =============================================================================
// 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 // 整页镜像 (旧格式)
		baseLen  uint32 // 整页镜像入链长度 (补尾记录按它定位覆盖偏移)
		isRow    bool   // 是否为行级记录 (新格式)
		opType   byte   // 行级操作类型 / WALOpPageRest
		storeRow []byte // 行级数据 / 补尾数据
	}
	var entries []recoveryEntry

	var crcFailed bool // 链上任意一条条目 CRC 损坏即置位
	// malformed: 条目 dataLen 越出页容量 —— 长度字段本身不可信, 后续条目偏移
	// 全部无从解析, 与 CRC 失败一样停止收集。
	var malformed bool
	// chainBreakErr: 链中途页读失败 / 页类型非法。旧实现只 break, 会把"已提交却
	// 残缺"的链当完整事务重放 —— 与 CRC 失败拒绝重放的策略矛盾; 已提交链在循环外
	// 据此返回错误并保留链。未提交链维持丢弃语义: 事务从未发布, 残缺无害。
	var chainBreakErr error
	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 {
			chainBreakErr = fmt.Errorf("WAL 页 %d 读失败: %w", pid, err)
			break
		}

		hdr := page.HdrOf(buf)
		if hdr.Type != page.WALIndex {
			chainBreakErr = fmt.Errorf("WAL 页 %d 类型非法 (%d, 期望 %d)", pid, 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]

			// uint64 提升: dataLen 被损坏/恶意写成超大值 (如 0xFFFFFFFF) 时, uint32
			// 加法回绕会绕过越界检查, 下一行切片直接 panic。越界即停止收集, 已提交链
			// 在循环外按"拒绝部分重放"返回错误。
			if uint64(off)+page.WALEntryMeta+uint64(dataLen) > uint64(PageSize) {
				malformed = true
				remaining = 0
				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 {
				// 校验失败: 后续内容不可信, 停止收集
				crcFailed = true
				remaining = 0
				break
			}

			if opType == WALOpFullPage {
				// 整页镜像: 零填充到整页长度, 保证写回时文件偏移一致
				data := make([]byte, PageSize)
				copy(data, payload)
				entries = append(entries, recoveryEntry{pageID: targetPID, data: data, baseLen: dataLen})
			} else if opType == WALOpPageRest {
				// 整页镜像的页尾补片: 记录目标页与数据, 重放阶段按 baseLen 定位偏移
				rest := make([]byte, dataLen)
				copy(rest, payload)
				entries = append(entries, recoveryEntry{pageID: targetPID, opType: opType, storeRow: rest})
			} 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)
	}

	// 已提交链出现 CRC 损坏 / 条目越界 / 链中途断裂: 不能静默部分重放 (撕裂提交)
	// 或回收整链 (销毁唯一完好的副本) —— 报告错误让 Open 失败, WAL 链原样保留,
	// 由用户决定修复方式 (从备份恢复 / 手工截断 WAL 链放弃这笔事务)。
	// 未提交链维持丢弃语义: 事务从未发布, 残缺/损坏的链丢弃回收即可。
	if committed == WALStateCommitted {
		switch {
		case crcFailed:
			return fmt.Errorf("WAL 恢复: 已提交事务 (TxnID=%d) 的条目 CRC 校验失败, "+
				"拒绝部分重放或回收 —— WAL 链已原样保留, 请修复后重试", header.NextTxnID)
		case malformed:
			return fmt.Errorf("WAL 恢复: 已提交事务 (TxnID=%d) 含越界条目 (dataLen 超出页容量), "+
				"拒绝部分重放或回收 —— WAL 链已原样保留, 请修复后重试", header.NextTxnID)
		case chainBreakErr != nil:
			return fmt.Errorf("WAL 恢复: 已提交事务 (TxnID=%d) 的 WAL 链中途断裂 (%v), "+
				"拒绝部分重放或回收 —— WAL 链已原样保留, 请修复后重试", header.NextTxnID, chainBreakErr)
		}
	}

	// 根据 Committed 标记决定是否重放
	if committed == WALStateCommitted && len(entries) > 0 {
		// Phase 1: 先处理整页条目 (恢复页结构: B+树节点, IndexMeta, DataPage 等)。
		//
		// 头页镜像 (pageID==0) 推迟到数据页全部写回之后: 头页记着 WAL 链指针与空闲链,
		// 若先写头页、中途断电, 会出现"头页说 WAL 已清空、数据页却还是旧内容"的窗口
		// (与提交路径步骤 4/5 钉死的顺序同一道理)。当前提交端头页恒为最后一条
		// RecordPage, 本不会先写 —— 恢复侧不依赖这个巧合。
		var headerImage []byte
		covered := make(map[uint32]uint32, len(entries)) // pageID → 已写回的字节数 (补尾定位用)
		for _, e := range entries {
			if e.isRow {
				continue
			}
			if e.pageID == 0 {
				headerImage = e.data // RecordPage 已按页去重, 至多一条头页镜像
				continue
			}
			if e.opType == WALOpPageRest {
				// 页尾补片: Data = [覆盖偏移 u32][页尾原文]。必须跟在同页整页镜像之后
				// (写端 packEntries 保证); 缺前置镜像 / 偏移越界都说明链被篡改或损坏,
				// 按拒绝部分重放返回错误。
				if _, ok := covered[e.pageID]; !ok {
					return fmt.Errorf("WAL 恢复: 已提交事务含孤儿补尾记录 (页 %d 缺少前置整页镜像), "+
						"拒绝部分重放 —— WAL 链已原样保留, 请修复后重试", e.pageID)
				}
				if len(e.storeRow) < 4 {
					return fmt.Errorf("WAL 恢复: 已提交事务的补尾记录过短 (页 %d, %d 字节), "+
						"拒绝部分重放 —— WAL 链已原样保留, 请修复后重试", e.pageID, len(e.storeRow))
				}
				off := binary.LittleEndian.Uint32(e.storeRow[:4])
				rest := e.storeRow[4:]
				if off+uint32(len(rest)) > PageSize {
					return fmt.Errorf("WAL 恢复: 已提交事务的补尾记录越界 (页 %d: %d+%d > %d), "+
						"拒绝部分重放 —— WAL 链已原样保留, 请修复后重试", e.pageID, off, len(rest), PageSize)
				}
				restBuf := make([]byte, PageSize)
				if err := ff.readPageDisk(e.pageID, restBuf); err != nil {
					return err
				}
				copy(restBuf[off:], rest)
				if err := ff.WritePageDirect(e.pageID, restBuf); err != nil {
					return err
				}
				continue
			}
			if err := ff.WritePageDirect(e.pageID, e.data); err != nil {
				return err
			}
			covered[e.pageID] = e.baseLen
		}
		if headerImage != nil {
			if err := recoverHeader(ff, headerImage); 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 —— WAL 整页镜像重放阶段最后一步, 把头页写回磁盘。
//
// 关键: 写盘后必须同步 BufferPool 页 0 缓存。本函数用裸 WriteAt(0) 绕过 BufferPool,
// 如果不调 bp.UpdatePage(0, ...), Close 时 bp.Flush 会把 BufferPool 里陈旧的头页
// 缓存写回去 —— 刚好覆盖掉 recoverHeader 刚落盘的干净头页 (和 writeHeaderLocked 之前
// 犯的是同一个 bug, 之前 grep 只扫到了 file.go 的两个 WriteAt(0), 漏了这里)。
func recoverHeader(ff *File, data []byte) error {
	if ff.bp != nil {
		defer func() {
			if len(data) >= PageSize4K {
				ff.bp.UpdatePage(0, data[:PageSize4K])
			}
		}()
	}
	if len(data) < PageSize4K {
		_, err := ff.f.WriteAt(data, 0)
		return err
	}
	_, err := ff.f.WriteAt(data[:PageSize4K], 0)
	return err
}

// healOrphanWALPages 孤儿 WAL 链自愈。
//
// 背景: 正常提交里 WAL 页写入与头页指针发布是同句柄程序序, kill 拷出的文件
// 要么指针指向完整链 (RecoverWAL 重放), 要么链根本没写过 —— 但直写路径/旧版本
// 引擎可能留下"头页说没有 WAL、分配区间里却躺着类型=WALIndex 的链页"的中间态,
// 这些页不在空闲链上, 只能永久泄漏 (表现为 Analyzer 的 unaccounted/WAL 残留)。
//
// 判定刻意保守, 宁漏勿错:
//   - 只在 FirstWALPage==0 (没有待恢复事务) 时动手;
//   - 先实走空闲链, 必须与 header.FreePageCount 完全一致 (不成环、不越界、计数
//     相符) —— 这保证"不在空闲链上"这个判定本身可信, 不会把"其实在空闲链上、
//     只是头页版本错位走不到"的页二次回收 (二次回收会弄断空闲链);
//   - 链首 = TotalEntries>0 的 WAL 页 (续页该字段恒 0), 整链必须全部由
//     WALIndex 页组成、且不与空闲链/分配区间相交, 否则整条放弃;
//   - 回收不重放镜像: 若链属已提交事务, 其数据页回写 (wal.Commit 步骤 3) 先于
//     指针清零 (步骤 5) 落盘, 镜像已在盘上, 重放只是幂等覆盖; 若属未提交事务,
//     丢弃才是正确语义。统一 recycleChainWithCache 整条还链。
func healOrphanWALPages(ff *File, header *HeaderPage) error {
	if header.FirstWALPage != 0 {
		return nil // 有待恢复事务, 归 RecoverWAL 全权处理
	}
	// 空闲链自洽性: 实走长度必须与声明值一致
	freeSet := make(map[uint32]bool, header.FreePageCount)
	for pid := header.FreePageStart; pid != 0; {
		if freeSet[pid] {
			return nil // 成环: 空闲链已不可信, 放弃自愈
		}
		freeSet[pid] = true
		buf := make([]byte, PageSize)
		if err := ff.readPageDisk(pid, buf); err != nil {
			return nil // 链中途读失败: 同样放弃
		}
		pid = page.HdrOf(buf).NextPage
	}
	if uint32(len(freeSet)) != header.FreePageCount {
		return nil // 计数不符: 头页与链内容版本错位, freeSet 不可信
	}
	// 分配区间 = [2, NextFreePage), 与文件实际页数取交集兜底
	fi, err := ff.f.Stat()
	if err != nil {
		return nil
	}
	total := (fi.Size() - int64(PageSize4K)) / int64(PageSize)
	limit := int64(header.NextFreePage)
	if limit <= 2 || limit > total+1 {
		limit = total + 1 // 游标不可信时用文件大小
	}
	buf := make([]byte, PageSize)
	for p := uint32(2); int64(p) < limit; p++ {
		if freeSet[p] {
			continue
		}
		if err := ff.readPageDisk(p, buf); err != nil {
			continue
		}
		if page.HdrOf(buf).Type != page.WALIndex {
			continue
		}
		if page.WalHdrOf(buf).TotalEntries == 0 {
			continue // 续页/游离页, 跟着链首一并回收
		}
		// 找到一个链首: 先走完整链, 校验全部由 WAL 页组成且不越界
		// (cnt 上限防循环链: 链页互不相交, 正常不会到达 limit)
		head, tail, cnt := p, p, uint32(0)
		ok := true
		for q := head; q != 0; {
			if freeSet[q] || int64(q) >= limit || int64(cnt) >= limit {
				ok = false // 与空闲链/分配区间相交, 或链成环: 不是完整孤儿链
				break
			}
			tail = q
			if err := ff.readPageDisk(q, buf); err != nil {
				ok = false
				break
			}
			if page.HdrOf(buf).Type != page.WALIndex {
				ok = false
				break
			}
			cnt++
			q = page.HdrOf(buf).NextPage
		}
		if !ok || cnt == 0 {
			continue
		}
		var recErr error
		ff.withHeader(func() {
			recErr = recycleChainWithCache(ff, header, head, tail, cnt)
			if recErr == nil {
				recErr = ff.writeHeaderLocked(header)
			}
		})
		if recErr != nil {
			return fmt.Errorf("孤儿 WAL 链回收失败 (head=%d, %d 页): %w", head, cnt, recErr)
		}
	}
	return ff.sync()
}