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 }