TsDB2/internal/orphancheck/orphancheck.go

// Package orphancheck 取证工具: 找出 VRTX 文件里"既不在空闲链、也不被任何结构引用"的孤儿页,
// 并按页类型出直方图 + 样例 hex。
//
// 两套口径并排输出:
//
//   - analyzer 口径: 与 analyzer/audit.go 的 runAuditStructural 同一批可达根
//     (表页链 / schema 扩展链 / 未恢复 WAL 链 / 每表 data-heap-index 链),
//     不含 blob 链 —— 它就是 analyzer 报的 "44 页既不属于任何表/索引, 也不在空闲链上" 的数字。
//   - blob 感知口径: 在上面基础上, 用引擎只读打开库、全表扫出行内 BlobRef,
//     把 blob 链页 (Type=9) 也计入可达 —— 剩余才是真正的泄漏页。
//
// 两口径的差集 = "analyzer 误报的孤儿" (活 blob 页)。若 blob 感知口径仍有孤儿,
// 按类型直方图指认泄漏路径。
//
// 所有页布局常量与 analyzer/main.go 逐一对照 (那边由 layout_assert.go 编译期钉在
// DB/core 的真实结构体上); 本包不 import analyzer (它是 package main), 常量独立维护。
package orphancheck

import (
	"encoding/binary"
	"encoding/hex"
	"fmt"
	"os"
	"sort"
	"strings"

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

const (
	pgsz  = 16384
	hdrsz = 4096
	phdr  = 20

	pgHead  = 0
	pgTable = 1
	pgData  = 2
	pgHeap  = 3
	pgIMeta = 4
	pgIData = 5
	pgOver  = 6
	pgWAL   = 7
	pgSExt  = 8
	pgBlob  = 9

	// PageHdr 字段偏移
	phNext = 12
	phUse  = 16

	// 头页字段偏移 (core/header_page.go v11 布局, 编译期断言钉死)
	hdrNextFree  = 12
	hdrTablePage = 48
	hdrFreeStart = 56
	hdrFreeEnd   = 60
	hdrFreeCount = 64
	hdrFirstWAL  = 68

	// 表页/表记录 (TableInfo=104B, TableNameSize=48)
	tinfoSZ  = 104
	tiFlag   = 48
	tiHPCnt  = 52
	tiFHeap  = 64
	tiFData  = 72
	tiAlloc  = 82
	tiFIdx   = 88
	sextHdr  = 4
	recOff   = phdr + sextHdr
	flagTomb = 0x80

	// IndexMeta 记录 (100B, IndexNameSize=64)
	metaSZ      = 100
	imNameSz    = 64
	imType      = 64
	imFcnt      = 65
	imFlds      = 66
	imEP        = 84
	imFlag      = 96
	imFlagDel   = 0x80
	imFlagUniq  = 0x01
	idxHdr      = 32
	ixTyp       = 20
	ixELen      = 22
	ixCnt       = 24
	ixLeaf      = 28

	// 索引类型 (page/base_page.go)
	itTime    = 1
	itPrimary = 2
	itStr     = 3
	itCompose = 4
	itBTree   = 5
	itHash    = 6

	// Hash 索引元数据/桶页偏移 (analyzer/audit.go)
	hmMagic   = 32
	hmDir     = 48
	hashMagic = 0x54484833
	hashSlot  = 12
	hashBktHd = 28
)

func le16(b []byte) uint16 { return binary.LittleEndian.Uint16(b) }
func le32(b []byte) uint32 { return binary.LittleEndian.Uint32(b) }
func be32(b []byte) uint32 { return binary.BigEndian.Uint32(b) }

// PageInfo 一个孤儿页的取证信息。
type PageInfo struct {
	ID       uint32
	TypeByte byte
	UseSize  uint32
	Next     uint32
	Head64   []byte // 页头 64 字节样例
}

// ChainInfo 一条 blob 链的取证信息。
type ChainInfo struct {
	Table     string
	PKID      uint64
	FirstPage uint32
	Pages     int
	BrokenAt  uint32 // 链中途断掉/成环/越界时的页号 (0=完整)
}

// Result 取证结果。
type Result struct {
	File          string
	TotalPages    int
	NextFreePage  uint32
	FreeCount     int
	FreeDeclared  uint32
	FreeVerdict   string
	WALPending    int // 头页 FirstWALPage != 0 时的未恢复 WAL 链页数 (只读打开跳过恢复)
	Notes         []string

	ReachableNoBlob int // analyzer 口径可达页数 (含页 1 / 空闲页不计)
	BlobPages       int // blob 链页总数 (含与结构页重叠数)

	// AnalyzerOrphans: 与 analyzer 报数完全同口径的孤儿 (不含 blob 链可达性)
	AnalyzerOrphans    []PageInfo
	AnalyzerByType     map[string]int
	AnalyzerWalLeft    int

	// Orphans: blob 感知口径的真正孤儿
	Orphans    []PageInfo
	ByType     map[string]int
	WalLeft    int
	Samples    map[string][]PageInfo // 每类型最多 N 个样例 (头 64B)
	Chains     []ChainInfo
	BrokenChains int
}

// TypeName 页类型字节 -> 名字 (9=BlobIndex 是 analyzer 不认识的, 这里显式命名)。
func TypeName(t byte) string {
	switch t {
	case pgHead:
		return "Free(0)"
	case pgTable:
		return "Table(1)"
	case pgData:
		return "Data(2)"
	case pgHeap:
		return "Heap(3)"
	case pgIMeta:
		return "IdxMeta(4)"
	case pgIData:
		return "IdxData(5)"
	case pgOver:
		return "Overflow(6,deprecated)"
	case pgWAL:
		return "WAL(7)"
	case pgSExt:
		return "SchemaExt(8)"
	case pgBlob:
		return "BlobIndex(9)"
	}
	return fmt.Sprintf("?(%d)", t)
}

type walker struct {
	f      *os.File
	total  int
	claim  map[uint32]string // pid -> role
	cross  [][2]string       // 重复认领记录 (诊断)
}

// page 读一页。每次新分配 16KB —— 调用方拿到的切片在下一次 page() 之前有效。
// 取证工具是 one-shot, 不需要 analyzer 那种复用缓冲 (那里要在热循环里省 GC);
// 换来的正确性: 链 walk 里随时再读子页/元页, 复用单缓冲会被覆盖 (aliasing bug)。
func (w *walker) page(pid uint32) ([]byte, error) {
	b := make([]byte, pgsz)
	if _, err := w.f.ReadAt(b, hdrsz+int64(pid-1)*pgsz); err != nil {
		return nil, err
	}
	return b, nil
}

// take 认领一页; 重复认领记入 cross (同角色不算)。
func (w *walker) take(pid uint32, role string) bool {
	if pid == 0 {
		return false
	}
	if r, ok := w.claim[pid]; ok {
		if r != role {
			w.cross = append(w.cross, [2]string{r, role})
		}
		return false
	}
	w.claim[pid] = role
	return true
}

// chain 沿 NextPage 走链并认领 (带越界/成环保护)。
func (w *walker) chain(start uint32, role string) int {
	n := 0
	seen := map[uint32]bool{}
	for pid := start; pid != 0; {
		if pid < 2 || int(pid) > w.total || seen[pid] {
			return n
		}
		seen[pid] = true
		if !w.take(pid, role) {
			return n
		}
		n++
		b, err := w.page(pid)
		if err != nil {
			return n
		}
		pid = le32(b[phNext : phNext+4])
	}
	return n
}

// idxDataPages 收下一页行存储索引叶条目引用的 DataPageID (与 analyzer 同逻辑)。
func (w *walker) idxDataPages(b []byte, it byte) int {
	if b[0] != pgIData {
		return 0
	}
	cn := int(le32(b[ixCnt : ixCnt+4]))
	el := int(le16(b[ixELen : ixELen+2]))
	if el <= 4 {
		el = 12
	}
	refOff := el - 4
	if it == itPrimary {
		refOff = 8
	}
	if refOff < 0 || refOff+4 > el {
		return 0
	}
	n := 0
	for i := 0; i < cn; i++ {
		o := idxHdr + i*el
		if o+el > len(b) {
			break
		}
		if p := be32(b[o+refOff : o+refOff+4]); p != 0 {
			if w.take(p, "data(idx)") {
				n++
			}
		}
	}
	return n
}

func isRowStore(it byte) bool { return it == itTime || it == itPrimary }
func isBTreeIdx(it byte) bool { return it == itPrimary || it == itBTree || it == itStr || it == itCompose }

// indexPages 枚举一个索引占用的全部页 (移植 analyzer/audit.go indexPages)。
func (w *walker) indexPages(it byte, entry uint32) (n, dp int) {
	if entry == 0 || int(entry) > w.total {
		return 0, 0
	}
	switch {
	case it == itHash:
		meta, err := w.page(entry)
		if err != nil || le32(meta[hmMagic:hmMagic+4]) != hashMagic {
			return 0, 0
		}
		w.take(entry, "index-meta")
		n++
		limit := phdr + ((pgsz-phdr)/hashSlot)*hashSlot
		for dir := le32(meta[hmDir : hmDir+4]); dir != 0; {
			if dir < 2 || int(dir) > w.total {
				break
			}
			if !w.take(dir, "index-dir") {
				break
			}
			n++
			b, err := w.page(dir)
			if err != nil {
				break
			}
			for o := phdr; o+hashSlot <= limit; o += hashSlot {
				if head := le32(b[o : o+4]); head != 0 {
					n += w.chain(head, "index-bucket")
				}
			}
			dir = le32(b[phNext : phNext+4])
		}
		return n, 0

	case isBTreeIdx(it):
		queue := []uint32{entry}
		seen := map[uint32]bool{}
		for len(queue) > 0 {
			pid := queue[0]
			queue = queue[1:]
			if seen[pid] {
				continue
			}
			seen[pid] = true
			if pid < 2 || int(pid) > w.total {
				continue
			}
			if !w.take(pid, "index-node") {
				continue
			}
			n++
			b, err := w.page(pid)
			if err != nil || b[0] != pgIData {
				continue
			}
			if b[ixLeaf] == 1 {
				if isRowStore(it) {
					dp += w.idxDataPages(b, it)
				}
				for nx := le32(b[phNext : phNext+4]); nx != 0; {
					if nx < 2 || int(nx) > w.total || seen[nx] {
						break
					}
					seen[nx] = true
					if !w.take(nx, "index-node") {
						break
					}
					n++
					nb, err := w.page(nx)
					if err != nil {
						break
					}
					if isRowStore(it) {
						dp += w.idxDataPages(nb, it)
					}
					nx = le32(nb[phNext : phNext+4])
				}
				continue
			}
			cn := int(le32(b[ixCnt : ixCnt+4]))
			el := int(le16(b[ixELen : ixELen+2]))
			if el <= 4 {
				el = 12
			}
			for i := 0; i < cn; i++ {
				o := idxHdr + i*el
				if o+el > len(b) {
					break
				}
				if child := be32(b[o+el-4 : o+el]); child != 0 {
					queue = append(queue, child)
				}
			}
		}
		return n, dp

	default:
		n = w.chain(entry, "index-chain")
		return n, 0
	}
}

// tableInfo 结构走查所需的最小表信息 (不含 schema)。
type tableInfo struct {
	name  string
	dp0   uint32
	hp0   uint32
	ip0   uint32
	index []idxInfo
}

type idxInfo struct {
	typ byte
	ep  uint32
}

// parseTables 移植 analyzer pTabs 的表链/表记录/索引元数据解析 (跳过 schema 重建)。
func (w *walker) parseTables() []tableInfo {
	var out []tableInfo
	hdr := make([]byte, hdrsz)
	if _, err := w.f.ReadAt(hdr, 0); err != nil {
		return out
	}
	for pid := le32(hdr[hdrTablePage : hdrTablePage+4]); pid != 0; {
		if pid < 2 || int(pid) > w.total {
			break
		}
		b, err := w.page(pid)
		if err != nil || b[0] != pgTable {
			break
		}
		us := le32(b[phUse : phUse+4])
		if us > pgsz {
			us = pgsz
		}
		for off := uint32(recOff); off+tinfoSZ <= us && off+tinfoSZ <= pgsz; {
			tb := b[off : off+tinfoSZ]
			alloc := uint32(le16(tb[tiAlloc : tiAlloc+2]))
			if tb[tiFlag]&flagTomb != 0 {
				off += tinfoSZ + alloc
				continue
			}
			t := tableInfo{
				name: strings.TrimRight(string(tb[:48]), "\x00"),
				dp0:  le32(tb[tiFData : tiFData+4]),
				hp0:  le32(tb[tiFHeap : tiFHeap+4]),
				ip0:  le32(tb[tiFIdx : tiFIdx+4]),
			}
			// 索引元数据链
			for ip := t.ip0; ip != 0; {
				if ip < 2 || int(ip) > w.total {
					break
				}
				ib, err := w.page(ip)
				if err != nil || ib[0] != pgIMeta {
					break
				}
				iu := le32(ib[phUse : phUse+4])
				if iu > pgsz {
					iu = pgsz
				}
				for io := uint32(phdr); io+metaSZ <= iu && io+metaSZ <= pgsz; io += metaSZ {
					im := ib[io : io+metaSZ]
					if im[imFlag]&imFlagDel != 0 {
						continue
					}
					ep := le32(im[imEP : imEP+4])
					if ep != 0 {
						if pb, err := w.page(ep); err != nil || pb[0] != pgIData {
							ep = 0
						}
					}
					if ep != 0 {
						t.index = append(t.index, idxInfo{typ: im[imType], ep: ep})
					}
				}
				ip = le32(ib[phNext : phNext+4])
			}
			out = append(out, t)
			off += tinfoSZ + alloc
		}
		nx := le32(b[phNext : phNext+4])
		if nx == 0 || nx == pid {
			break
		}
		pid = nx
	}
	return out
}

// Options 取证选项。
type Options struct {
	// EngineBlobScan 用引擎只读打开库扫行内 BlobRef (推荐; 只在副本上跑)。
	EngineBlobScan bool
	// SamplesPerType 每种类型的样例页数 (头 64B hex)。
	SamplesPerType int
}

// Run 对 dbPath 执行取证。调用方必须保证 dbPath 是副本 (本函数只读打开,
// 但"只读打开"本身不适合直接在唯一原件上做)。
func Run(dbPath string, opt Options) (*Result, error) {
	if opt.SamplesPerType <= 0 {
		opt.SamplesPerType = 3
	}
	f, err := os.Open(dbPath)
	if err != nil {
		return nil, err
	}
	defer f.Close()
	st, err := f.Stat()
	if err != nil {
		return nil, err
	}
	R := &Result{
		File:            dbPath,
		TotalPages:      int((st.Size() - hdrsz) / pgsz),
		ByType:          map[string]int{},
		AnalyzerByType:  map[string]int{},
		Samples:         map[string][]PageInfo{},
	}
	if R.TotalPages < 0 {
		return nil, fmt.Errorf("文件 %s 不足 4KB 头页", dbPath)
	}

	hdr := make([]byte, hdrsz)
	if _, err := f.ReadAt(hdr, 0); err != nil {
		return nil, err
	}
	nf := le32(hdr[hdrNextFree : hdrNextFree+4])
	R.NextFreePage = nf
	freeHead := le32(hdr[hdrFreeStart : hdrFreeStart+4])
	R.FreeDeclared = le32(hdr[hdrFreeCount : hdrFreeCount+4])
	freeEnd := le32(hdr[hdrFreeEnd : hdrFreeEnd+4])

	w := &walker{f: f, total: R.TotalPages, claim: map[uint32]string{}}

	// 1) 空闲链
	freeSet := map[uint32]bool{}
	{
		seen := map[uint32]bool{}
		bad := 0
		var tail uint32
		for pid := freeHead; pid != 0; {
			if int(pid) > w.total || seen[pid] {
				bad++
				break
			}
			seen[pid] = true
			freeSet[pid] = true
			tail = pid
			b, err := w.page(pid)
			if err != nil {
				bad++
				break
			}
			pid = le32(b[phNext : phNext+4])
		}
		R.FreeCount = len(freeSet)
		switch {
		case bad > 0:
			R.FreeVerdict = "损坏 (成环/越界)"
		case uint32(R.FreeCount) != R.FreeDeclared:
			R.FreeVerdict = fmt.Sprintf("计数不符 (实走 %d, header 声明 %d)", R.FreeCount, R.FreeDeclared)
		case freeHead != 0 && freeEnd != tail:
			R.FreeVerdict = fmt.Sprintf("尾指针不符 (实走尾 %d, header 声明 %d)", tail, freeEnd)
		default:
			R.FreeVerdict = "OK"
		}
	}

	// 2) 结构可达 (analyzer 同口径)
	w.take(1, "reserved")
	w.chain(le32(hdr[hdrTablePage:hdrTablePage+4]), "table-page")
	// schema 扩展链 (表页第 1 页 PHDR 之后, 大端)
	if tp0 := le32(hdr[hdrTablePage : hdrTablePage+4]); tp0 != 0 {
		if b, err := w.page(tp0); err == nil && len(b) >= phdr+4 {
			w.chain(be32(b[phdr:phdr+4]), "schema-ext")
		}
	}
	// 未恢复 WAL 链 (只读打开引擎不会重放, 链页仍是活的可达根)
	if w0 := le32(hdr[hdrFirstWAL : hdrFirstWAL+4]); w0 != 0 {
		R.WALPending = w.chain(w0, "wal-pending")
		R.Notes = append(R.Notes, fmt.Sprintf("头页 FirstWALPage=%d (%d 页): 上次未干净关闭或只读跳过恢复, 页归属可能不完整", w0, R.WALPending))
	}

	tables := w.parseTables()
	for i := range tables {
		t := &tables[i]
		w.chain(t.dp0, "data:"+t.name)
		w.chain(t.hp0, "heap:"+t.name)
		w.chain(t.ip0, "imeta:"+t.name)
		for _, idx := range t.index {
			w.indexPages(idx.typ, idx.ep)
		}
	}
	R.ReachableNoBlob = len(w.claim)

	// 3) 孤儿: analyzer 口径 (此时 claim 里还没有 blob 链)
	snapNoBlob := make(map[uint32]string, len(w.claim))
	for pid, role := range w.claim {
		snapNoBlob[pid] = role
	}
	R.AnalyzerOrphans, R.AnalyzerByType, R.AnalyzerWalLeft = collectOrphans(f, w, nf, freeSet, snapNoBlob)

	// 4) blob 链可达 (引擎只读扫描行内 BlobRef)
	blobClaimed := 0
	if opt.EngineBlobScan {
		chains, notes, claimed := scanBlobChains(dbPath, tables, w)
		R.Chains = chains
		R.Notes = append(R.Notes, notes...)
		blobClaimed = claimed
	}

	// 5) 孤儿: blob 感知口径 (在结构可达上叠加 blob 链), 并按类型留样
	R.Orphans, R.ByType, R.WalLeft = collectOrphans(f, w, nf, freeSet, w.claim)
	R.BlobPages = blobClaimed
	for _, ci := range R.Chains {
		if ci.BrokenAt != 0 {
			R.BrokenChains++
		}
	}
	perType := map[string]int{}
	for _, p := range R.Orphans {
		tn := TypeName(p.TypeByte)
		if perType[tn] < opt.SamplesPerType {
			R.Samples[tn] = append(R.Samples[tn], p)
			perType[tn]++
		}
	}

	if len(w.cross) > 0 {
		R.Notes = append(R.Notes, fmt.Sprintf("有 %d 次跨角色重复认领 (一页多个主人), 详见交叉引用", len(w.cross)))
	}
	return R, nil
}

// collectOrphans 算 [2, NextFreePage) 内不在空闲链、不在 claim 集合里的页,
// 返回 (全部孤儿, 按类型直方图, 其中 WAL 类型页数)。
func collectOrphans(f *os.File, w *walker, nf uint32, freeSet map[uint32]bool,
	claim map[uint32]string) ([]PageInfo, map[string]int, int) {
	var out []PageInfo
	byType := map[string]int{}
	walLeft := 0
	buf := make([]byte, pgsz)
	for pid := uint32(2); int(pid) < int(nf) && int(pid) <= w.total; pid++ {
		if _, ok := claim[pid]; ok {
			continue
		}
		if freeSet[pid] {
			continue
		}
		pi := PageInfo{ID: pid}
		if _, err := f.ReadAt(buf, hdrsz+int64(pid-1)*pgsz); err == nil {
			pi.TypeByte = buf[0]
			pi.UseSize = le32(buf[phUse : phUse+4])
			pi.Next = le32(buf[phNext : phNext+4])
			pi.Head64 = append([]byte(nil), buf[:64]...)
		}
		out = append(out, pi)
		byType[TypeName(pi.TypeByte)]++
		if pi.TypeByte == pgWAL {
			walLeft++
		}
	}
	return out, byType, walLeft
}

// scanBlobChains 引擎只读打开库, 全表扫出行内 BlobRef, 走链认领 blob 页。
// 返回 (链清单, 备注, 认领页数)。任何引擎错误都不致命 —— 降级为"未走 blob 链"。
func scanBlobChains(dbPath string, tables []tableInfo, w *walker) ([]ChainInfo, []string, int) {
	var notes []string
	db, err := core.OpenWithOptions(dbPath, core.OpenOptions{ReadOnly: true})
	if err != nil {
		return nil, []string{fmt.Sprintf("引擎只读打开失败, 跳过 blob 链走查: %v", err)}, 0
	}
	defer db.Close()

	before := len(w.claim)
	var chains []ChainInfo
	broken := 0
	refs := 0
	for i := range tables {
		t := &tables[i]
		tbl, ok := db.FindTable(t.name)
		if !ok {
			notes = append(notes, fmt.Sprintf("表 %s: 引擎 FindTable 失败, 该表 blob 列未走查", t.name))
			continue
		}
		perPK := map[uint64][]types.BlobRef{}
		scanErr := tbl.Query().ScanWithPKID(func(pkID uint64, vals []types.FieldValue) bool {
			for _, fv := range vals {
				if ref, ok := fv.Value.(types.BlobRef); ok && !ref.Empty() {
					perPK[pkID] = append(perPK[pkID], ref)
				}
			}
			return true
		})
		if scanErr != nil {
			notes = append(notes, fmt.Sprintf("表 %s: 扫描失败 (%v), blob 列可能漏走", t.name, scanErr))
		}
		for pkID, rs := range perPK {
			for _, ref := range rs {
				refs++
				ci := ChainInfo{Table: t.name, PKID: pkID, FirstPage: ref.FirstPageID}
				seen := map[uint32]bool{}
				for p := ref.FirstPageID; p != 0; {
					if p < 2 || int(p) > w.total || seen[p] {
						ci.BrokenAt = p
						broken++
						break
					}
					seen[p] = true
					ci.Pages++
					w.take(p, "blob:"+t.name)
					b, err := w.page(p)
					if err != nil {
						ci.BrokenAt = p
						broken++
						break
					}
					if b[0] != pgBlob {
						ci.BrokenAt = p
						broken++
						break
					}
					p = le32(b[phNext : phNext+4])
				}
				chains = append(chains, ci)
			}
		}
	}
	if refs == 0 {
		notes = append(notes, "全库行内没有任何非空 BlobRef (无 blob 列或无 blob 数据)")
	}
	if broken > 0 {
		notes = append(notes, fmt.Sprintf("有 %d 条 blob 链中途断裂/类型不符/越界 —— 链页泄漏或行内引用损坏", broken))
	}
	return chains, notes, len(w.claim) - before
}

// SampleHex 把样例页的头 64B 格式化成 hex 行。
func SampleHex(p PageInfo) string {
	return hex.Dump(p.Head64)
}

// SortedTypes 直方图按类型名排序输出。
func SortedTypes(m map[string]int) []string {
	ks := make([]string, 0, len(m))
	for k := range m {
		ks = append(ks, k)
	}
	sort.Strings(ks)
	return ks
}