TsDB2/DB/core/file.go

package core

import (
	"errors"
	"fmt"
	"os"
	"sync"
	"unsafe"

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

// File 只负责：打开、关闭、读、写 —— unsafe 全部收敛
type File struct {
	f  *os.File
	bp *pool.BufferPool // 页面缓冲池 (nil = 未启用)

	// headerMu 保护 HeaderPage (pageID=0) 的【内存字段】及其持久化。
	//
	// 头页是整个引擎唯一的全局可变结构: 写入路径要它分配 / 回收页号 (NextFreePage /
	// FreePageStart / FreePageEnd / FreePageCount), 提交路径要它发布 / 清空 WAL 链指针
	// (FirstWALPage / LastWALPage) 并推进 NextTxnID。两条路径并发时若不串行化:
	//   - 两个 goroutine 同时读到同一个 NextFreePage 再各自 +1 → 同一个页号发给两方,
	//     后写的一方覆盖先写的一方 → 静默丢数据;
	//   - WriteHeader 在拷贝整张头页时, 另一个 goroutine 正在改字段 → 数据竞争 (崩溃)。
	// 所有对 HeaderPage 字段的读写都必须落在 headerMu 内。
	//
	// 不能把 mutex 放进 HeaderPage 本身: 它按内存布局经 unsafe 直接落盘 (见 WriteHeader /
	// marshalHeader), 加字段会改变盘上格式。
	headerMu sync.Mutex

	// headerDirty 头页脏标记 (headerMu 保护): 自上次头页落盘以来 allocPage 是否
	// 又改过头页 (NextFreePage 游标 / 空闲链)。
	//
	// 旧实现在 allocPage 里每分配【一个】页就同步 pwrite 一次 4K 头页 —— 批量灌数
	// 期间上千张新页 (数据页 / 树节点 / 堆页) 就是上千次头页落盘, 实测占灌数 CPU
	// 的 ~16%。而这期间数据页本身全在缓冲池里 (未 Commit 根本不落盘), 头页急着
	// 持久化毫无收益 —— 崩溃后这批缓冲页本来就丢了, 头页游标回退一点反而自洽
	// (那些页号还没被任何盘上数据占用)。
	//
	// 落盘点收敛到三处:
	//   - Commit: WAL 打包写头页 (wal.go 的 writeHeaderLocked 链指针收尾);
	//   - 空闲链变更: realFreePage / recycleChain 本来就写头页;
	//   - Close: db.Close 先 Commit (写头页) 再刷缓冲池, 兜底用 FlushHeader。
	// 前两处之外若有人要把数据页直接写盘, 必须先让头页落盘 (页号 > 盘上
	// NextFreePage 的页一旦落盘, 重开时会被二次分配覆盖)。
	headerDirty bool

	// readOnly 抢救模式 (force-inspect) 标志: 一切落盘写 (数据页 / 头页 /
	// 截断) 返回 ErrReadOnly, 读路径不受影响。由 OpenWithOptions 在校验
	// 失败但用户显式要求抢救时置位。
	readOnly bool

	// fileLocked 是否已持有跨进程文件锁 (tryLock 置位, unlock / Close 清除)。
	// 仅内存状态, 不落盘。锁用来阻止两个进程同时打开同一个数据库文件
	// (sync.Mutex 只在单进程内有效, 跨进程只能靠 OS 文件锁, 见 file_lock_*.go)。
	fileLocked bool

	// lockOffset tryLock 时锁的起始偏移 (LockFileEx 必须用完全相同的区间解锁;
	// 文件 size 在锁持有期间可能因 RecoverWAL / Close.Flush 而变化, 所以
	// 解锁时不能再 stat 一次 —— 必须保存 tryLock 时的值)。
	lockOffset uint64
}

// ErrReadOnly 抢救模式 (force-inspect) 下所有写操作的返回错误。
var ErrReadOnly = errors.New("force-inspect 只读模式: 写操作被禁用")

// ErrDatabaseLocked 数据库文件已被另一个进程 (持有文件锁) 占用时, Open 返回的错误。
// 非阻塞: 拿不到锁立即返回本错误, 而不是无限等待 —— 调用方可稍后重试或确认是否有
// 其它进程正在使用该文件。
var ErrDatabaseLocked = errors.New("数据库已被另一个进程锁定(占用): 同一时刻只允许一个进程打开")

func OpenFile(path string) (*File, error) {
	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0644)
	if err != nil {
		return nil, err
	}
	return &File{f: f}, nil
}

// OpenFileReadOnly 以只读方式打开【已存在】的数据库文件 (不创建、不可写)。
func OpenFileReadOnly(path string) (*File, error) {
	f, err := os.Open(path) // os.O_RDONLY
	if err != nil {
		return nil, err
	}
	return &File{f: f, readOnly: true}, nil
}

func (ff *File) Close() error {
	if ff.bp != nil {
		ff.bp.Close() // 先刷盘再关文件
	}
	ff.unlock() // 显式释放跨进程文件锁 (若持有); 句柄关闭也会自动释放, 这里兜底
	return ff.f.Close()
}

// InitBufferPool 初始化页面缓冲池, 拦截 ReadPage/WritePage 实现内存缓存
// capacity: 最大缓存页数 (建议 10000~50000)
func (ff *File) InitBufferPool(capacity int) {
	ff.bp = pool.NewBufferPool(capacity, ff.readPageDisk, ff.writePageDisk)
}

// SetDirtyThreshold 设置脏页阈值, 超过时触发回调 (配合 WAL 自动 Commit)
func (ff *File) SetDirtyThreshold(threshold int, onThreshold func()) {
	if ff.bp != nil {
		ff.bp.SetDirtyThreshold(threshold, onThreshold)
	}
}

// DirtyOverThreshold 脏页数是否已达到阈值 (无缓冲池时 false)。
// beginOp 收尾用它判断"阈值在写操作期间被越过"并补提交。
func (ff *File) DirtyOverThreshold() bool {
	return ff.bp != nil && ff.bp.OverThreshold()
}

// SetOnPageWriteCallback 设置页写入回调（用于事务登记被修改的页）
func (ff *File) SetOnPageWriteCallback(callback func(pageID uint32)) {
	if ff.bp != nil {
		ff.bp.SetOnPageWriteCallback(callback)
	}
}

// LockWrites 锁定写入 —— 锁的是"页落盘"(BufferPool.WritePage), 不是写入动作本身。
func (ff *File) LockWrites() {
	if ff.bp != nil {
		ff.bp.LockWrites()
	}
}

// UnlockWrites 解锁写入 (见 LockWrites 的语义说明)
func (ff *File) UnlockWrites() {
	if ff.bp != nil {
		ff.bp.UnlockWrites()
	}
}

// SetBufferPoolCapacity 设置 BufferPool 容量（页数）
func (ff *File) SetBufferPoolCapacity(capacity int) {
	if ff.bp != nil {
		ff.bp.SetCapacity(capacity)
	}
}

// CollectDirty 收集所有脏页快照 (配合 WAL 使用; 页不会因此离开缓存, 见 pool.CollectDirty)
func (ff *File) CollectDirty() []pool.DirtyPage {
	if ff.bp != nil {
		return ff.bp.CollectDirty()
	}
	return nil
}

// MarkClean 把这批快照页收成干净页 (提交成功后调用, 与 CollectDirty 配对)
func (ff *File) MarkClean(pages []pool.DirtyPage) {
	if ff.bp != nil {
		ff.bp.MarkClean(pages)
	}
}

// FlushBufferPool 刷盘 (Close / DropTable 等绕过 WAL 的收尾路径使用)。
// 返回第一个写盘错误: 失败的页保持 dirty, 由调用方决定如何处置 (如 Close 上抛,
// DropTable 清缓存前显式丢弃)。
func (ff *File) FlushBufferPool() error {
	if ff.bp != nil {
		return ff.bp.Flush()
	}
	return nil
}

// RestoreDirty 提交失败时兜底: 保证这批页在缓存里仍然是脏页 (与 CollectDirty 配对)
func (ff *File) RestoreDirty(pages []pool.DirtyPage) {
	if ff.bp != nil {
		ff.bp.RestoreDirty(pages)
	}
}

// DiscardPages 丢弃指定页的缓存 (不落盘), 供事务 Rollback 回到磁盘状态
func (ff *File) DiscardPages(pageIDs map[uint32]struct{}) {
	if ff.bp != nil {
		ff.bp.DiscardPages(pageIDs)
	}
}

// -----------------------------------------------------------------------------------
// ✅ 动态长度：索引页 — 缓冲池拦截层
// -----------------------------------------------------------------------------------
func (ff *File) WritePage(pageID uint32, buf []byte) error {
	if ff.bp != nil {
		return ff.bp.WritePage(pageID, buf) // 写缓存, 延迟落盘
	}
	return ff.writePageDisk(pageID, buf)
}

func (ff *File) ReadPage(pageID uint32, buf []byte) error {
	if ff.bp != nil {
		return ff.bp.ReadPage(pageID, buf) // 先查缓存
	}
	return ff.readPageDisk(pageID, buf)
}

// WritePageDirect 直接写盘 (绕过 BufferPool)
//
// 因为绕过了缓冲池, 必须让缓存中该页的旧副本失效 —— 否则后续 ReadPage 会命中
// 与磁盘不一致的陈旧内容。典型场景: 页从空闲链复用时缓存里还留有旧副本, 被当作
// WAL 页直写后该副本已失真; 整链回收若只同步链尾页, 就会漏掉链内这些页。
//
// 失效必须排在【落盘之后】。反过来的话, 从删缓存到写完盘这一小段里, 这一页同样
// "两头都不在": 对一页本会话刚分配、盘上还从没写过的页, 并发读会拿到全零 (文件还
// 没长到那个偏移时直接读出错), 而 btree 把全零页当成 EntryLen==0 的节点, 随后就在
// btPID 里切片越界。先落盘再失效, 最坏也只是读到"马上要被覆盖的旧内容", 不会两头皆空。
func (ff *File) WritePageDirect(pageID uint32, buf []byte) error {
	if err := ff.writePageDisk(pageID, buf); err != nil {
		return err
	}
	if ff.bp != nil {
		ff.bp.DiscardPage(pageID)
	}
	return nil
}

// WritePageHome 把页镜像写回它自己的位置 (WAL 提交时的"回填"步骤)。
//
// 与 WritePageDirect 的唯一区别是【不动缓存副本】: 这份镜像是从缓存快照里来的,
// 缓存里那一份的内容只会等于它或新于它, 无差别失效反而会丢掉提交窗口里写入线程刚
// 改出来的新内容 (并发写 + 提交时就是如此)。页的脏/干净状态交给提交方用
// File.MarkClean 按版本号统一收尾 —— 没被改过的转干净, 改过的继续留脏。
func (ff *File) WritePageHome(pageID uint32, buf []byte) error {
	return ff.writePageDisk(pageID, buf)
}

// SyncCache 同步缓存页 (WAL 提交后调用, 保持缓存与磁盘一致)
func (ff *File) SyncCache(pageID uint32, data []byte) {
	if ff.bp != nil {
		ff.bp.UpdatePage(pageID, data)
	}
}

// ClearCache 清空所有缓存 (WAL 提交后调用)
func (ff *File) ClearCache() {
	if ff.bp != nil {
		ff.bp.ClearAll()
	}
}

// -----------------------------------------------------------------------------------
// ✅ 底层磁盘 I/O (缓冲池回源)
// -----------------------------------------------------------------------------------
func (ff *File) writePageDisk(pageID uint32, buf []byte) error {
	if ff.readOnly {
		return ErrReadOnly
	}
	if pageID == 0 {
		// 数据页 ID 从 1 起 (页 0 是头页, 走 writeHeaderLocked)。pageID-1 在
		// uint32 下会下溢成 0xFFFFFFFF, int64 转换后偏移 ~70TB —— 写错位置或直接报错
		return fmt.Errorf("writePageDisk: 非法数据页号 0 (数据页 ID 从 1 起)")
	}
	offset := int64(PageSize4K) + int64(pageID-1)*int64(len(buf))
	_, err := ff.f.WriteAt(buf, offset)
	return err
}

func (ff *File) readPageDisk(pageID uint32, buf []byte) error {
	if pageID == 0 {
		// 同 writePageDisk: uint32 下溢会让偏移飞到 ~70TB, 返回错误而不是读空气
		return fmt.Errorf("readPageDisk: 非法数据页号 0 (数据页 ID 从 1 起)")
	}
	offset := int64(PageSize4K) + int64(pageID-1)*int64(len(buf))
	_, err := ff.f.ReadAt(buf, offset)
	return err
}

// -----------------------------------------------------------------------------------
// ✅ 头读写（固定4K）
// -----------------------------------------------------------------------------------

// writeHeaderLocked 把头页 (4K) 落盘。调用方必须已持有 headerMu。
// 与 WriteHeader 的区别就是不重复加锁 —— 供 withHeader 内部使用, 避免自死锁。
//
// 这是头页出盘的【唯一汇聚点】: PrepareWrite (整页 XXH32) 在这里统一调用,
// 任何路径写头页 CRC 都自动最新; 对旧 v11 库 (HeaderCrc==0), 这次写入同时
// 完成"自动生成 CRC"的升级。
func (ff *File) writeHeaderLocked(h *HeaderPage) error {
	if ff.readOnly {
		return ErrReadOnly
	}
	h.PrepareWrite()
	var buf [PageSize4K]byte
	copy(buf[:], (*[PageSize4K]byte)(unsafe.Pointer(h))[:])
	_, err := ff.f.WriteAt(buf[:], 0)
	if err == nil {
		ff.headerDirty = false // 头页已是最新
		if ff.bp != nil {
			ff.bp.UpdatePage(0, buf[:])
		}
	}
	return err
}

// withHeader 在 headerMu 保护下执行 fn。
//
// fn 内可以安全地读 / 改 HeaderPage 的任何字段; 若还需要把头页落盘, 必须调用
// writeHeaderLocked (不能调用 WriteHeader —— 那会二次获取 headerMu, 直接死锁)。
//
// fn 内【不要】调用会同步触发提交的动作 (典型是经 BufferPool 缓存的 WritePage: 脏页
// 超过阈值时它会在当前 goroutine 里直接调 Commit, 而 Commit 的 WAL 写入又要 headerMu):
// 那会自死锁。需要这种写页请放到锁外, 见 allocPage。
func (ff *File) withHeader(fn func()) {
	ff.headerMu.Lock()
	defer ff.headerMu.Unlock()
	fn()
}

// WriteHeader 线程安全地把头页 (4K) 落盘。
func (ff *File) WriteHeader(h *HeaderPage) error {
	ff.headerMu.Lock()
	defer ff.headerMu.Unlock()
	return ff.writeHeaderLocked(h)
}

// FlushHeader 头页有未落盘的分配游标变更时把它写下去。
//
// 兜底用途: 把缓冲池里的脏页直接刷盘之前必须先调用 (db.Close / 任何绕过
// Commit 的 Flush)—— 否则盘上会出现页号超过头页 NextFreePage 的页, 重开
// 后被二次分配覆盖。Commit 路径自己会写头页, 通常到这里已是干净的。
func (ff *File) FlushHeader(h *HeaderPage) error {
	ff.headerMu.Lock()
	defer ff.headerMu.Unlock()
	if !ff.headerDirty {
		return nil
	}
	return ff.writeHeaderLocked(h)
}

// IsHeaderDirty 头页自上次落盘以来是否有未持久化的真实变更 (当前即分配游标移动)。
// db.Commit 据此决定要不要把头页镜像纳入本笔 WAL —— 无分配的纯改写提交不动头页。
func (ff *File) IsHeaderDirty() bool {
	ff.headerMu.Lock()
	defer ff.headerMu.Unlock()
	return ff.headerDirty
}

// allocPage 线程安全地分配一个页号: 在 headerMu 内摘空闲链 / 递增游标并推进头页,
// 随后把头页落盘。
//
// erase 决定是否把该页擦成空白页 (写缓存, 延迟落盘):
//   - 普通写入路径传 true: 复用的空闲页可能残留旧页类型, 必须擦干净, 否则按页号
//     读回来会被当成合法但类型错误的页。
//   - WAL 打包路径传 false: WAL 页紧接着就在内存里被整页重造 (16KB 全写), 不需要
//     擦页; 更要紧的是此时正处在 Commit 内部 (持有 commitMu), 而擦页会走
//     File.WritePage → BufferPool.WritePage, 脏页超阈值时同步回调 autoCommit →
//     再进一次 Commit → 卡死在已持有的 commitMu 上 (fatal: all goroutines are asleep)。
//
// 为什么 headerMu 只罩住"选号 + 推进头页", 擦页放在锁外: 擦页可能重入提交,
// 而提交的 WAL 打包又要 headerMu —— 锁内擦页同样是自死锁。
//
// 头页不在这里落盘 (只标脏): 分配的页本身只在缓冲池里, 未提交前盘上没有它们;
// 头页游标的持久化收敛到 Commit / 空闲链变更 / Close 三处 (见 headerDirty 注释)。
func (ff *File) allocPage(h *HeaderPage, erase bool) uint32 {
	ff.headerMu.Lock()
	// writePage 传 nil: 只推进头页游标, 不做任何可能重入提交的写。
	pid := h.newPageWithReader(ff.readPageDisk, nil)
	ff.headerDirty = true
	ff.headerMu.Unlock()

	if !erase {
		return pid
	}

	// 锁外擦页: 页头只留自述页号 (Type=0、UseSize=页头长), 其余字节全 0。
	clean := make([]byte, PageSize)
	page.HdrOf(clean).Init(0, pid)
	_ = ff.WritePage(pid, clean)
	return pid
}

func (ff *File) ReadHeader() (*HeaderPage, error) {
	buf := make([]byte, PageSize4K)
	_, err := ff.f.ReadAt(buf, 0)
	if err != nil {
		return nil, err
	}
	h := &HeaderPage{}
	copy((*[PageSize4K]byte)(unsafe.Pointer(h))[:], buf)
	return h, nil
}

func (ff *File) TruncateSize(size int64) error {
	if ff.readOnly {
		return ErrReadOnly
	}
	return ff.f.Truncate(size)
}