TsDB2/DB/core/wal.go

package core

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"encoding/binary"
	"hash/crc32"
)

// =============================================================================
// WAL (Write-Ahead Log) 实现
//
// WAL 页布局 (16KB), 权威定义见 page.WALPageHdr:
//   [WALPageHdr(40B)] [WALEntry...]
//   首页与续页头部等长 —— 续页的 TxnID/TotalEntries/State 填 0, 换来整条链能用
//   同一个结构体解析, 读写都不再按 isFirst 分叉。
//
// ⭐ 页间链指针用 PageHdr.NextPage (与空闲链同一个字段), 不再自定义 NextWalPage ——
// 这样整条 WAL 链本身就是一条合法空闲链, 回收时一次拼接即可, 不必逐页遍历 (见 recycleChain)。
//
// WALEntry (见 page.WALEntryMeta): [TargetPageID(4B)] [DataLen(4B)] [Checksum(4B)] [OpType(1B)] [Data(DataLen)]
//   OpType == WALOpFullPage(0)          → Data 为整页镜像 (DataLen = 页有效长度)
//   OpType == Insert/Delete/Update      → Data 为行级数据
//   Checksum = CRC32(OpType || Data)
//
// ⚠️ OpType 必须显式存储: 早期实现靠"Data 的首字节"猜类型, 但整页镜像的首字节
// 就是 PageHdr.Type, 而 DataPage/TablePage/HeapPage 的 Type 恰好是 2/1/3,
// 与行级 OpType 冲突, 整页记录会被误判成行级记录而丢失。
//
// WAL 状态机 (State 字段):
//   0 = Mem (内存中，未写盘)
//   1 = Written (已 write()，未 fsync())
//   2 = Persisted (已 fsync()，但未标记提交)
//   3 = Committed (已提交，可安全恢复)
//
// 打包: 一个 WAL 页可容纳**多条** Entry —— 写端按「当前页剩余空间 >= 13B + DataLen」
// 紧凑追加, 放不下才开新页。整页镜像 (≈16KB) 仍会独占一页; 瘦页 (点更新/小表) 则能
// 在一页里合并几十条, 页尾不再被零填充浪费。
// 读端本就按 EntryCount 逐条解析 (见 wal_recovery.go), 所以打包只改写端。
// =============================================================================

// DataPage 磁盘布局见 page.DataPageHdr (头 40B), 行数据区从 page.DataPageHdrSize 开始。

// WAL 状态常量 (对应 State 字段)
const (
	WALStateMem       = 0 // 内存中，未写盘
	WALStateWritten   = 1 // 已 write()，未 fsync()
	WALStatePersisted = 2 // 已 fsync()，但未标记提交
	WALStateCommitted = 3 // 已提交，可安全恢复
)

// =============================================================================
// WAL 条目操作类型 (每个条目都显式带 1B OpType)
//
// INSERT: Data = storeRow [pkID(8B BE) + rowData]
// DELETE: Data = pkID(8B BE)
// UPDATE: Data = storeRow [pkID(8B BE) + rowData]  (新行数据)
// =============================================================================

const (
	WALOpFullPage = 0 // 整页镜像 (Data = 页内容)
	WALOpInsert   = 1 // 插入行
	WALOpDelete   = 2 // 删除行
	WALOpUpdate   = 3 // 更新行
)

// RowEntry 行级 WAL 记录
type RowEntry struct {
	PageID   uint32 // 目标 DataPage ID
	OpType   byte   // WALOpInsert / WALOpDelete / WALOpUpdate
	StoreRow []byte // INSERT/UPDATE: [pkID(8B BE) + rowData]; DELETE: pkID(8B BE)
}

// WALWriter 事务 WAL 写入器
type WALWriter struct {
	txnID   uint64
	entries []walEntry
	// pageIndex: pageID → entries 下标。RecordPage 的"同页只留最新镜像"去重用,
	// 免掉每次提交 O(条目数²) 的线性扫 (默认阈值 500 页/提交时是 12.5 万次比较)。
	pageIndex map[uint32]int
	file      *File
	header    *HeaderPage
	state     byte // 当前 WAL 状态 (Mem/Written/Persisted/Committed)
}

type walEntry struct {
	pageID     uint32
	data       []byte    // 完整页镜像 (isRowLevel=false 时使用)
	isRowLevel bool      // 是否为行级记录（true=存行数据，false=存整页）
	opType     byte      // 行级操作类型 (WALOpInsert/WALOpDelete/WALOpUpdate)
	rowEntry   *RowEntry // 行级记录
}

// BeginTxn 开始新事务
func (db *DB) BeginTxn() *WALWriter {
	// NextTxnID 是头页里的全局游标, 提交 / 事务提交都可能并发取号, 必须在 headerMu 内自增。
	var txnID uint64
	db.file.withHeader(func() {
		txnID = db.header.NextTxnID
		db.header.NextTxnID++
	})
	return &WALWriter{
		txnID:     txnID,
		entries:   make([]walEntry, 0, 16),
		pageIndex: make(map[uint32]int, 16),
		file:      db.file,
		header:    db.header,
		state:     WALStateMem, // 初始状态: Mem (内存中)
	}
}

// RecordPage 记录一个被修改的页 (同一 pageID 只保留最新镜像)。
// 去重走 pageIndex 映射, O(1); 旧实现对每条 entry 线性扫全部 entries,
// 一次 500 页的提交就是 12.5 万次比较。
func (w *WALWriter) RecordPage(pageID uint32, data []byte) {
	d := make([]byte, len(data))
	copy(d, data)
	if i, ok := w.pageIndex[pageID]; ok {
		w.entries[i].data = d
		return
	}
	w.pageIndex[pageID] = len(w.entries)
	w.entries = append(w.entries, walEntry{
		pageID:     pageID,
		data:       d,
		isRowLevel: false, // 原有逻辑：存整页
	})
}

// Rollback 回滚: 丢弃所有条目
func (w *WALWriter) Rollback() {
	w.entries = nil
	w.pageIndex = nil
	w.state = WALStateMem // 重置状态
}

// Commit 原子提交流程:
//  1. 将 entries 写入 WAL 页链 (State: Mem → Written)
//  2. 标记 WAL 首页 Committed = 1 (State → Committed)
//  3. fsync 一次: WAL 页与提交标记随这次 fsync 一起落盘
//     (排序要求的是"标记 durable ⇒ 数据已 durable", 同一次 fsync 前的顺序写出即满足)
//  4. 回写实际页 (redo; 整页镜像, 重复执行幂等)
//  5. 回收所有 WAL 页 + 清空 Header 链指针 + fsync
//
// ⭐ 关键: 「提交标记」必须先于「回写真实页」落盘。
// 若顺序反过来, 断电落在两者之间时真实页已被原地覆盖, 而提交标记仍是旧值,
// 恢复会丢弃整个 WAL, 留下半应用(损坏)的页 —— 这里没有 undo, 无法自愈。
//
// 缓冲池的脏标记【不在这里】收尾: 提交方拿到 success 之后调用 File.MarkClean
// (按版本号逐页判定), 因为只有提交方知道哪些页属于这次快照, 也才知道窗口里有没有
// 人把这些页又改了一遍。
func (w *WALWriter) Commit() error {
	if len(w.entries) == 0 {
		return nil
	}

	// === 步骤 1+2+3: 写 WAL 页链 → 标记首页已提交 → 一次 fsync ===
	// WAL 数据与提交标记在同一次 fsync 前依次写出: fsync 保证"标记 durable ⇒
	// 数据先于它 durable"的顺序不变 (排序要求的是先后持久, 不是两次 fsync)。
	// 旧实现这里分两次 fsync (写完数据一次、标记后再一次), 每次提交白付一轮 fsync 延迟。
	walPageIDs, firstBuf, err := w.writeWALPages()
	if err != nil {
		return err
	}

	// === 步骤 3: 标记 WAL 首页为已提交 (State: Written → Committed) ===
	// 这一步之后, 无论发生什么崩溃, 恢复都能重放整笔事务 (WAL 已完整且被标记)。
	if err := w.markCommitted(walPageIDs[0], firstBuf); err != nil {
		return err
	}
	// 状态转换: Written → Persisted → Committed (随下面这次 fsync 落盘)
	if err := w.file.sync(); err != nil {
		return err
	}
	w.state = WALStateCommitted

	// === 步骤 4: 回写实际页 (直接写盘, 绕过 BufferPool) ===
	// 放在提交标记之后: 中途断电可由恢复阶段重做 (整页镜像, 幂等)。
	//
	// 用 WritePageHome 而不是 WritePageDirect: 这里写出去的正是缓存快照本身, 缓存里
	// 那一份的内容只会等于它或新于它 (并发写入时), 无差别失效会丢掉更新的那一份。
	// 缓存状态由提交方在 Commit() 返回后用 File.MarkClean 按版本号收尾 —— 这也是
	// 这里不再 ClearCache 的原因: ClearAll 会把提交窗口里被重新改脏的页一并标干净,
	// 那些页并不在本笔 WAL 里, 标干净等于把它们的新内容永久丢掉。
	for _, e := range w.entries {
		if e.pageID == 0 {
			continue
		}
		if err := w.file.WritePageHome(e.pageID, e.data); err != nil {
			return err
		}
	}

	// 步骤 4.5: 回写后立刻 fsync, 钉死"数据页先于头页(步骤 5 清 WAL 指针)落盘"。
	// 没有这一次 fsync, OS 后台回刷可能把步骤 5 的头页先刷下去、数据页留在缓存 —
	// 断电后头页说"没有 WAL 可重放", 数据页却是旧值, 已提交数据静默丢失。
	if err := w.file.sync(); err != nil {
		return err
	}

	// === 步骤 5: 回收整条 WAL 链 (O(1) 拼接, 与链长无关) ===
	// 链内每页的 next 都写在 PageHdr.NextPage, 与空闲链同字段, 所以整条链
	// head…tail 可以直接挂到空闲链尾, 不必逐页遍历回收。
	// 链首/链尾/链长都已在 writeWALPages 中记录 (walPageIDs), 尾页 NextPage 必为 0。
	//
	// 还链 (改空闲链指针) 与清空 WAL 链指针是一次头页上的复合修改, 必须整体落在
	// headerMu 内, 否则并发写入线程可能看到"WAL 链已清空、空闲链却还没接好"的中间态。
	var recycleErr error
	w.file.withHeader(func() {
		recycleErr = recycleChainWithCache(w.file, w.header,
			walPageIDs[0], walPageIDs[len(walPageIDs)-1], uint32(len(walPageIDs)))
		if recycleErr != nil {
			return
		}
		w.header.FirstWALPage = 0
		w.header.LastWALPage = 0
		recycleErr = w.file.writeHeaderLocked(w.header)
	})
	if recycleErr != nil {
		return recycleErr
	}
	// 收尾必须 fsync, 它钉死一个关键的持久化顺序: "忘掉日志"(头页: WAL 链回收
	// + 指针清零) 不得先于"数据就位"(步骤 4 回写的真实数据页) 落盘。
	// 没有这次 fsync, 断电时 OS 可能只把头页刷下去、数据页留在缓存里 —— 重开时
	// 头页说"没有 WAL", 已提交的数据却是旧内容, 且 WAL 页已被标成空闲页即将被
	// 覆盖, 连重放救回的机会都没有 (静默丢已提交数据)。
	// 顺带的效果: 头页若仍指着已提交的 WAL, 下次打开会重复重放 —— 整页镜像幂等
	// 无害, 行级记录则会重复插入。
	if err := w.file.sync(); err != nil {
		return err
	}

	w.entries = nil
	w.pageIndex = nil
	return nil
}

// writeWALPages 将所有 entry 紧凑写入 WAL 页链, 返回 (分配的页 ID 列表, 首页缓冲区)。
// 状态转换: Mem → Written
//
// 打包策略: 一条 Entry 占 [13B 元数据 + DataLen]。当前页剩余空间够就继续追加,
// 不够才定型落盘并开新页 —— 一条 200B 的记录原本独占 16KB, 现在一页可放约 76 条。
//
// 所有 WAL 页头部等长 (page.WALPageHdrSize); walEntryPayload 已把单条记录截断到
// 「一定能放进空页」的长度, 因此开新页后必定写得下, 不会出现空页仍放不下的情况。
func (w *WALWriter) writeWALPages() ([]uint32, []byte, error) {
	var pageIDs []uint32
	var firstBuf []byte // 首页缓冲区: markCommitted 直接改它打提交标记, 免一次整页读回

	// 当前正在填充的 WAL 页
	var curBuf []byte
	var curPID uint32
	var curIsFirst bool
	var curOff uint32     // 已用字节数 (从页头之后开始)
	var curEntries uint16 // 已写入的条目数
	var opTypeBuf [1]byte // CRC 的 opType 段缓冲 (分段 Update 免每条一次临时分配)

	// flushCurrent 定型当前页 (回填 UseSize / EntryCount) 并落盘。
	// 链指针不在这里填 —— 只有"开新页"时才知道下一页是谁, 由调用处先回填再 flush。
	// ⭐ 全程用内存缓冲记录"是否首页", 不从磁盘读回上一页判定 —— 那种实现依赖
	// TxnID != 0, 而首个事务的 TxnID 恰好是 0, 会把链指针写到错误偏移。
	flushCurrent := func() error {
		if curBuf == nil {
			return nil
		}
		wh := page.WalHdrOf(curBuf)
		wh.UseSize = curOff
		wh.EntryCount = curEntries
		return w.file.WritePageDirect(curPID, curBuf)
	}

	for i := range w.entries {
		opType, targetPID, payload := w.walEntryPayload(w.entries[i])
		dataLen := uint32(len(payload))
		need := page.WALEntryMeta + dataLen

		// 当前页放不下 → 开新页
		if curBuf == nil || uint32(PageSize)-curOff < need {
			// 分配新页: headerMu 内摘空闲链 / 递增游标并推进头页。
			// erase=false —— 页紧接着就在内存里被整页重造 (curBuf 是全 0 的新切片,
			// flushCurrent 一次写满 16KB), 不需要擦页; 且此刻正处在 Commit 内
			// (持有 commitMu), 擦页会重入 autoCommit → Commit → 自死锁在 commitMu 上。
			pid := w.file.allocPage(w.header, false)

			// 先回填上一页的链指针, 再把上一页定型落盘。
			// 链指针统一写 PageHdr.NextPage (偏移 12) —— 与空闲链表同一个字段,
			// 所以整条 WAL 链本身就是一条合法空闲链, 回收时可一次拼接 (见 Commit 步骤 5)。
			if curBuf != nil {
				page.HdrOf(curBuf).NextPage = pid
				if err := flushCurrent(); err != nil {
					return nil, nil, err
				}
			}

			pageIDs = append(pageIDs, pid)
			curBuf = make([]byte, PageSize)
			if len(pageIDs) == 1 {
				// 首页缓冲留一份给 markCommitted: 它的内容在"开第二页回填链指针 +
				// flush"后就定稿了, 之后不再被本函数改动 —— 提交标记直接改这份内存
				// 缓冲写回, 免一次"整页读回只为改 1 字节 State"。
				firstBuf = curBuf
			}
			curPID = pid
			curIsFirst = len(pageIDs) == 1
			curEntries = 0

			var prevPID uint32
			if len(pageIDs) > 1 {
				prevPID = pageIDs[len(pageIDs)-2]
			}
			wh := page.WalHdrOf(curBuf)
			wh.Init(pid, prevPID)

			if curIsFirst {
				// === 首页: TxnID + TotalEntries + State ===
				wh.TxnID = w.txnID
				wh.TotalEntries = uint32(len(w.entries))
				wh.State = WALStateWritten // State = Written (已写盘，未 fsync)
			}
			// 首页与续页头部等长, 条目区起点一致
			curOff = page.WALPageHdrSize
		}

		// 追加当前 entry
		// 统一格式: [TargetPageID(4B)][DataLen(4B)][Checksum(4B)][OpType(1B)][Data]
		// CRC32 覆盖 OpType + Data, 整页/行级一致, 类型显式存储不再靠首字节猜测。
		// crc32.Update 分段计算与"拼成一条再 ChecksumIEEE"完全等价, 省掉每条一次
		// 1+DataLen 的临时缓冲分配。
		opTypeBuf[0] = opType
		crc := crc32.Update(0, crc32.IEEETable, opTypeBuf[:])
		crc = crc32.Update(crc, crc32.IEEETable, payload)

		binary.LittleEndian.PutUint32(curBuf[curOff:], targetPID)
		binary.LittleEndian.PutUint32(curBuf[curOff+4:], dataLen)
		binary.LittleEndian.PutUint32(curBuf[curOff+8:], crc)
		curBuf[curOff+12] = opType
		curOff += page.WALEntryMeta
		copy(curBuf[curOff:], payload)
		curOff += dataLen
		curEntries++
	}

	// 定型最后一页
	if err := flushCurrent(); err != nil {
		return nil, nil, err
	}

	// 更新 HeaderPage 的 WAL 链指针
	// 发布链指针是为了崩溃恢复能找回这笔事务, 必须在 headerMu 内对"首页是否已存在"
	// 与"尾页指向"做一次原子更新, 否则并发提交可能互相覆盖链尾。
	if len(pageIDs) > 0 {
		w.file.withHeader(func() {
			if w.header.FirstWALPage == 0 {
				w.header.FirstWALPage = pageIDs[0]
			}
			w.header.LastWALPage = pageIDs[len(pageIDs)-1]
			_ = w.file.writeHeaderLocked(w.header)
		})
	}

	// 状态转换: Mem → Written (已写盘，未 fsync)
	w.state = WALStateWritten

	return pageIDs, firstBuf, nil
}

// walEntryPayload 解析一条 entry 的目标页 / OpType / 载荷, 并把载荷截断到
// 「一定能放进一张空 WAL 页」的长度 (首页/续页头等长, 只需一个限制)。
//
// 这个截断是 writeWALPages 打包逻辑的前提: 开新页后必定能写下当前这条记录。
func (w *WALWriter) walEntryPayload(e walEntry) (byte, uint32, []byte) {
	opType := byte(WALOpFullPage)
	targetPID := e.pageID
	var payload []byte

	switch {
	case e.isRowLevel && e.rowEntry != nil:
		opType = e.rowEntry.OpType
		targetPID = e.rowEntry.PageID
		payload = e.rowEntry.StoreRow
	case e.isRowLevel:
		// 行级条目缺少 rowEntry: 视作空行级记录
		opType = e.opType
		payload = nil
	default:
		// 整页记录: 只存有效前缀 (整页 16KB 无法与条目元数据同放一页)
		storeLen := uint32(len(e.data))
		if e.pageID == 0 {
			// HeaderPage 只有前 4KB 有效; 必须与 recoverHeader 的解析长度一致
			storeLen = PageSize4K
		} else if len(e.data) >= int(page.PageHdrSize) {
			useSize := page.HdrOf(e.data).UseSize
			if useSize > 0 && useSize < uint32(len(e.data)) {
				storeLen = useSize
			}
		}
		if storeLen > uint32(len(e.data)) {
			storeLen = uint32(len(e.data))
		}
		payload = e.data[:storeLen]
	}

	// 防御: 单条记录不得超过空首页容量 (行级记录同样受此保护)
	if maxLen := uint32(PageSize) - page.WALPageHdrSize - page.WALEntryMeta; uint32(len(payload)) > maxLen {
		payload = payload[:maxLen]
	}
	return opType, targetPID, payload
}

// markCommitted 标记 WAL 首页为已提交 (State → Committed)。
// firstBuf 是 writeWALPages 手里刚写盘的首页内存缓冲 (内容与盘上逐字节一致),
// 直接改它的 State 字节整页写回 —— 旧实现为此整页读回 16KB, 只为改 1 字节。
func (w *WALWriter) markCommitted(firstPageID uint32, firstBuf []byte) error {
	if len(firstBuf) != int(PageSize) {
		// 防御: 缓冲缺失/长度不符时退回读盘路径 (正常不会发生)
		firstBuf = make([]byte, PageSize)
		// 必须直接读盘: WAL 页是用 WritePageDirect 写的, 走 BufferPool 可能读到陈旧缓存
		if err := w.file.readPageDisk(firstPageID, firstBuf); err != nil {
			return err
		}
	}
	page.WalHdrOf(firstBuf).State = WALStateCommitted // State = Committed
	if err := w.file.WritePageDirect(firstPageID, firstBuf); err != nil {
		return err
	}
	return nil
}

// sync 刷盘 (fsync)
func (ff *File) sync() error {
	return ff.f.Sync()
}

// =============================================================================
// applyRowToPage 将行级 WAL 记录应用到页缓冲区 (用于崩溃恢复)
//
// 操作类型:
//
//	WALOpInsert: 在 DataPage 中追加一行
//	WALOpDelete: 在 DataPage 中软删除指定 pkID 的行
//	WALOpUpdate: 在 DataPage 中按 pkID 原地更新行数据
//
// storeRow 格式: [pkID(8B BE) + rowData] (INSERT/UPDATE)
// storeRow 格式: pkID(8B BE)           (DELETE)
// =============================================================================
func applyRowToPage(buf []byte, opType byte, storeRow []byte) {
	const delFlag = uint32(0x80000000)

	if len(buf) < int(page.DataPageHdrSize) {
		return
	}
	// 读取 DataPage 元数据 (Count/Min/Max 紧随 PageHdr)
	dh := page.DataHdrOf(buf)
	count := dh.Count
	dataEnd := dh.UseSize
	if dataEnd < page.DataPageHdrSize || dataEnd > uint32(len(buf)) {
		return
	}

	switch opType {
	case WALOpInsert:
		// 追加行: [len(4B)] [storeRow]
		off := dataEnd
		if off+4+uint32(len(storeRow)) > uint32(len(buf)) {
			return // 页满, 放弃 (整页镜像已包含该行)
		}
		binary.LittleEndian.PutUint32(buf[off:], uint32(len(storeRow)))
		off += 4
		copy(buf[off:], storeRow)
		off += uint32(len(storeRow))
		// 更新 count 和 dataEnd (UseSize); 追加的是活行, 存活计数同步 +1
		dh.Count = count + 1
		dh.LiveCnt++
		dh.UseSize = off

	case WALOpDelete:
		// storeRow = pkID(8B BE)
		if len(storeRow) < 8 {
			return
		}
		keyPK := binary.BigEndian.Uint64(storeRow[:8])
		off := page.DataPageHdrSize
		for i := uint32(0); i < count; i++ {
			if off+4 > dataEnd {
				break
			}
			rawLen := binary.LittleEndian.Uint32(buf[off:])
			deleted := rawLen&delFlag != 0
			rowLen := rawLen & ^delFlag
			rowStart := off + 4
			if rowStart+rowLen > dataEnd {
				break
			}
			if !deleted && rowLen >= 8 {
				rowPK := binary.BigEndian.Uint64(buf[rowStart : rowStart+8])
				if rowPK == keyPK {
					// 设置软删除标志; 存活/失效统计同步维护 (页头持久化, 见 page.DataPageHdr)
					binary.LittleEndian.PutUint32(buf[off:], rowLen|delFlag)
					dh.LiveCnt--
					dh.Dead += 4 + rowLen
					return
				}
			}
			off = rowStart + rowLen
		}

	case WALOpUpdate:
		// storeRow = [pkID(8B BE) + rowData]
		if len(storeRow) < 8 {
			return
		}
		keyPK := binary.BigEndian.Uint64(storeRow[:8])
		off := page.DataPageHdrSize
		for i := uint32(0); i < count; i++ {
			if off+4 > dataEnd {
				break
			}
			rawLen := binary.LittleEndian.Uint32(buf[off:])
			deleted := rawLen&delFlag != 0
			rowLen := rawLen & ^delFlag
			rowStart := off + 4
			if rowStart+rowLen > dataEnd {
				break
			}
			if !deleted && rowLen >= 8 {
				rowPK := binary.BigEndian.Uint64(buf[rowStart : rowStart+8])
				if rowPK == keyPK {
					// 原地覆盖行数据
					copy(buf[rowStart:], storeRow)
					// 剩余字节置零
					for j := uint32(len(storeRow)); j < rowLen; j++ {
						buf[rowStart+j] = 0
					}
					return
				}
			}
			off = rowStart + rowLen
		}
	}
}