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 }