TsDB2/analyzer/blobwalk_test.go

package main

// 回归: 带 blob 列的库, 结构审计不得把活 blob 链页报成孤儿页。
//
// 背景 (DockerManager 实测): 迁移后的库 analyzer 报"44 页既不属于任何表/索引,
// 也不在空闲链上" —— 那 44 页是行内 BlobRef 指向的活链页 (Type=9), 早期
// analyzer 的可达根不含 blob 链, 整链误报。本测试建一张带 blob 列的表
// (含跨页大 blob), 跑 runAuditStructural, 要求 Unaccounted == 0。

import (
	"bytes"
	"testing"

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

func seedBlobDB(t *testing.T) string {
	t.Helper()
	path := t.TempDir() + "/blob.db"
	db, err := core.Open(path)
	if err != nil {
		t.Fatal(err)
	}
	db.SetAutoCommitDelay(0)
	tbl, err := db.CreateTable("docs", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("body")},
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := tbl.CreateIndex("pk_docs", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatal(err)
	}
	// 小 blob (单页链) + 大 blob (跨页链, 确保多页链也被认领)。
	big := bytes.Repeat([]byte("0123456789abcdef"), 8*1024) // 128KB ≈ 8 个链页
	rows := [][]byte{
		[]byte("hello blob"),
		big,
		bytes.Repeat([]byte("z"), 20000), // 略超单页数据区, 强制跨页
	}
	for i, body := range rows {
		if _, err := tbl.InsertRowWithID(uint64(i+1), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i + 1)},
			{FieldName: []byte("body"), Value: body},
		}); err != nil {
			t.Fatal(err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	return path
}

func TestAuditBlobChainsReachable(t *testing.T) {
	path := seedBlobDB(t)

	if err := useDB(path); err != nil {
		t.Fatalf("useDB: %v", err)
	}
	invalidateAudit()
	mu.Lock()
	fp = path
	rHdr()
	tabs = pTabs()
	mu.Unlock()

	R, _, _, _, _ := runAuditStructural()
	if R.Reach.Wal > 0 {
		t.Skipf("库带着未恢复 WAL (%d 页), 页归属不完整, 本断言不适用", R.Reach.Wal)
	}
	if R.Reach.Unaccounted != 0 {
		for i := range R.Tables { t.Logf("表 %s BlobChains=%d BlobPages=%d", R.Tables[i].Name, R.Tables[i].BlobChains, R.Tables[i].BlobPages) }
		t.Fatalf("活 blob 链页被误报成孤儿: Unaccounted=%d byType=%v ids=%v",
			R.Reach.Unaccounted, R.Reach.UnaccountedByType, R.Reach.UnaccountedIDs)
	}
	if got := R.Reach.ByRole["blob"]; got <= 0 {
		t.Fatalf("ByRole[blob] = %d, 期望 > 0 —— blob 链根本没被走查", got)
	}
	var ta *TableAudit
	for i := range R.Tables {
		if R.Tables[i].Name == "docs" {
			ta = &R.Tables[i]
		}
	}
	if ta == nil {
		t.Fatal("审计结果里没有 docs 表")
	}
	if ta.BlobChains != 3 {
		t.Fatalf("BlobChains = %d, 期望 3 (每行一条链)", ta.BlobChains)
	}
	if ta.BlobPages < 10 {
		t.Fatalf("BlobPages = %d, 期望 >= 10 (128KB 大 blob 单独就 ~8 页)", ta.BlobPages)
	}
	// 释放全局句柄, 否则 Windows 下 t.TempDir 清理删不掉文件
	fileMu.Lock()
	if dbf != nil {
		dbf.Close()
		dbf = nil
	}
	fileMu.Unlock()
}