TsDB2/DB/core/blob_leak_test.go

package core_test

// 回归: 写入失败的行不得泄漏 blob 链页 (Type=9 孤儿页)。
//
// 两条历史泄漏路径:
//  1. serializeRowInternal 里 blob 列先落链, 后续列编码失败 → 链页无人引用;
//  2. indexMeta.InsertRow/InsertRows 唯一冲突报错 → 行没落任何索引, 链页无人引用。
// 修复后这些链页必须回到空闲链 (orphancheck 的 blob 感知口径孤儿数 == 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"
	"git.ycl.pub/yl32940/TsDB2/internal/orphancheck"
)

func blobLeakSchema() []types.FieldInfo {
	return []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("body")},
		{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(32), FieldName: []byte("tag")},
	}
}

func openBlobLeakDB(t *testing.T, path string) *core.DB {
	t.Helper()
	db, err := core.Open(path)
	if err != nil {
		t.Fatal(err)
	}
	db.SetAutoCommitDelay(0)
	tbl, err := db.CreateTable("docs", blobLeakSchema())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := tbl.CreateIndex("pk_docs", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatal(err)
	}
	return db
}

// 失败行 (序列化阶段: blob 列之后的字符串列超宽) 不得泄漏链页。
func TestBlobChainFreedOnSerializeError(t *testing.T) {
	path := t.TempDir() + "/leak1.db"
	db := openBlobLeakDB(t, path)

	// 先行一行成功数据, 作为"活链"对照。
	good, ok := db.FindTable("docs")
	if !ok {
		t.Fatal("FindTable docs: not found")
	}
	if _, err := good.InsertRowWithID(1, []types.FieldValue{
		{FieldName: []byte("id"), Value: uint64(1)},
		{FieldName: []byte("body"), Value: bytes.Repeat([]byte("a"), 40000)},
		{FieldName: []byte("tag"), Value: []byte("ok")},
	}); err != nil {
		t.Fatal(err)
	}

	// 失败行: blob 列先落链 (40KB), 随后 tag 超宽 → 整行写入中止。
	if _, err := good.InsertRowWithID(2, []types.FieldValue{
		{FieldName: []byte("id"), Value: uint64(2)},
		{FieldName: []byte("body"), Value: bytes.Repeat([]byte("b"), 40000)},
		{FieldName: []byte("tag"), Value: bytes.Repeat([]byte("x"), 100)},
	}); err == nil {
		t.Fatal("超宽 tag 应报错, 实际写入成功")
	}

	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}

	R, err := orphancheck.Run(path, orphancheck.Options{EngineBlobScan: true})
	if err != nil {
		t.Fatal(err)
	}
	if len(R.Orphans) > 0 {
		t.Fatalf("失败行泄漏 %d 页: %v (按类型 %v)", len(R.Orphans), idsOf(R.Orphans), R.ByType)
	}
	if R.BlobPages == 0 {
		t.Fatal("活 blob 链页数 = 0 —— 成功行的链也应被认领, 检查走查逻辑")
	}
}

// 唯一冲突行 (索引阶段预检报错, 零写入) 不得泄漏链页。
func TestBlobChainFreedOnUniqueConflict(t *testing.T) {
	path := t.TempDir() + "/leak2.db"
	db := openBlobLeakDB(t, path)
	tbl, ok := db.FindTable("docs")
	if !ok {
		t.Fatal("FindTable docs: not found")
	}
	if _, err := tbl.CreateIndex("uq_docs_tag", page.IndexTypeBTree, []byte{2}, true); err != nil {
		t.Fatal(err)
	}

	body1 := bytes.Repeat([]byte("a"), 40000)
	if _, err := tbl.InsertRowWithID(1, []types.FieldValue{
		{FieldName: []byte("id"), Value: uint64(1)},
		{FieldName: []byte("body"), Value: body1},
		{FieldName: []byte("tag"), Value: []byte("dup")},
	}); err != nil {
		t.Fatal(err)
	}
	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}

	// 冲突行: 唯一预检先于任何写入, blob 链已落 → 必须被回收。
	if _, err := tbl.InsertRowWithID(2, []types.FieldValue{
		{FieldName: []byte("id"), Value: uint64(2)},
		{FieldName: []byte("body"), Value: bytes.Repeat([]byte("b"), 40000)},
		{FieldName: []byte("tag"), Value: []byte("dup")},
	}); err == nil {
		t.Fatal("唯一冲突应报错, 实际写入成功")
	}

	// 批量路径同样不得泄漏。
	if _, err := tbl.InsertRowsWithIDs([]uint64{3}, [][]types.FieldValue{{
		{FieldName: []byte("id"), Value: uint64(3)},
		{FieldName: []byte("body"), Value: bytes.Repeat([]byte("c"), 40000)},
		{FieldName: []byte("tag"), Value: []byte("dup")},
	}}); err == nil {
		t.Fatal("批量唯一冲突应报错, 实际写入成功")
	}

	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}

	// 先行行的 blob 必须完好 (回收不得误伤活链)。
	got, err := tbl.BlobRead(types.BlobRef{
		FirstPageID: firstPageOf(t, tbl, 1), TotalLen: uint32(len(body1)),
	}, types.CompressNone)
	if err != nil {
		t.Fatalf("读回先行行的 blob: %v", err)
	}
	if !bytes.Equal(got, body1) {
		t.Fatalf("先行行的 blob 内容被损毁: 读回 %d 字节, 期望 %d", len(got), len(body1))
	}

	if err := db.Close(); err != nil {
		t.Fatal(err)
	}

	R, err := orphancheck.Run(path, orphancheck.Options{EngineBlobScan: true})
	if err != nil {
		t.Fatal(err)
	}
	if len(R.Orphans) > 0 {
		t.Fatalf("唯一冲突行泄漏 %d 页: %v (按类型 %v)", len(R.Orphans), idsOf(R.Orphans), R.ByType)
	}
}

// firstPageOf 取一行 body 列的链首页 (经行存储读回 BlobRef)。
func firstPageOf(t *testing.T, tbl *core.Table, pkID uint64) uint32 {
	t.Helper()
	vals, err := tbl.FindByPK(pkID)
	if err != nil {
		t.Fatalf("FindByPK(%d): %v", pkID, err)
	}
	for _, fv := range vals {
		if string(fv.FieldName) == "body" {
			ref, ok := fv.Value.(types.BlobRef)
			if !ok {
				t.Fatalf("body 列类型 %T, 期望 types.BlobRef", fv.Value)
			}
			return ref.FirstPageID
		}
	}
	t.Fatal("行里没有 body 列")
	return 0
}

func idsOf(ps []orphancheck.PageInfo) []uint32 {
	out := make([]uint32, len(ps))
	for i, p := range ps {
		out[i] = p.ID
	}
	return out
}