TsDB2/migrate/orphan_pages_test.go

package migrate

// 回归: sqlite → VRTX 迁移 (含 VACUUM) 之后, blob 感知口径的孤儿页必须为 0。
//
// 样本刻意覆盖 DockerManager 案的要素:
//   - payload TEXT 列里有一行 >16360 字节 (动态列单值上限) —— 按三态分流落成
//     Blob 链式列, 迁移产生 Type=9 blob 链页;
//   - 一个带 UNIQUE 索引的表 + 普通表。
//
// 判据: orphancheck 的 analyzer 口径 (不走 blob 链) 会报出这些活链页 —— 这正是
// analyzer 用户看到的"孤儿页"; 而 blob 感知口径 (走行内 BlobRef → 链) 必须为 0,
// 非 0 即迁移/引擎在成功路径上泄漏了页。

import (
	"bytes"
	"database/sql"
	"fmt"
	"path/filepath"
	"testing"

	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/internal/orphancheck"
)

// buildSQLiteSource 造一个 sqlite 源库, 返回路径。
func buildSQLiteSource(t *testing.T) string {
	t.Helper()
	path := filepath.Join(t.TempDir(), "src.db")
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	stmts := []string{
		`CREATE TABLE configs (id INTEGER PRIMARY KEY, name VARCHAR(64), payload TEXT)`,
		`CREATE TABLE svc (
			id INTEGER PRIMARY KEY,
			host VARCHAR(128),
			port INTEGER,
			UNIQUE(host)
		)`,
	}
	for _, s := range stmts {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("建源表: %v", err)
		}
	}
	// 大 payload: 40KB > 16360 单值上限 → 迁移时落 Blob 链式列。
	big := bytes.Repeat([]byte("0123456789abcdef"), 2560) // 40960 字节
	for i := 1; i <= 20; i++ {
		var payload any
		if i == 1 {
			payload = big
		} else {
			payload = fmt.Sprintf(`{"i":%d,"s":"small"}`, i)
		}
		if _, err := db.Exec(`INSERT INTO configs (id, name, payload) VALUES (?, ?, ?)`,
			i, fmt.Sprintf("cfg-%d", i), payload); err != nil {
			t.Fatal(err)
		}
	}
	for i := 1; i <= 50; i++ {
		if _, err := db.Exec(`INSERT INTO svc (id, host, port) VALUES (?, ?, ?)`,
			i, fmt.Sprintf("host-%d.example", i), 10000+i); err != nil {
			t.Fatal(err)
		}
	}
	return path
}

// migrateToVRTX 跑一次官方迁移, 返回目标 VRTX 路径。
func migrateToVRTX(t *testing.T, srcPath string) string {
	t.Helper()
	dstPath := filepath.Join(t.TempDir(), "out.db")
	src, err := OpenSQLite(srcPath, nil)
	if err != nil {
		t.Fatalf("打开 sqlite 源: %v", err)
	}
	dst, err := OpenVRTX(dstPath, nil)
	if err != nil {
		t.Fatalf("打开 VRTX 目标: %v", err)
	}
	if _, err := Run(src, dst, quiet()); err != nil {
		t.Fatalf("迁移: %v", err)
	}
	if err := dst.Close(); err != nil {
		t.Fatalf("关闭目标: %v", err)
	}
	if err := src.Close(); err != nil {
		t.Fatalf("关闭源: %v", err)
	}
	return dstPath
}

func checkOrphans(t *testing.T, dbPath string) *orphancheck.Result {
	t.Helper()
	R, err := orphancheck.Run(dbPath, orphancheck.Options{EngineBlobScan: true, SamplesPerType: 3})
	if err != nil {
		t.Fatalf("取证 %s: %v", dbPath, err)
	}
	return R
}

func TestMigrateLeavesNoOrphanPages(t *testing.T) {
	src := buildSQLiteSource(t)
	dst := migrateToVRTX(t, src)

	R := checkOrphans(t, dst)
	t.Logf("analyzer 口径孤儿=%d %v; blob 感知真实孤儿=%d %v",
		len(R.AnalyzerOrphans), R.AnalyzerByType, len(R.Orphans), R.ByType)

	if len(R.AnalyzerOrphans) == 0 {
		t.Log("注意: 样本未产生 analyzer 口径孤儿 (blob 链页), 误报对照组失效 —— 加大样本")
	}
	if len(R.Orphans) > 0 {
		var ids []uint32
		for _, p := range R.Orphans {
			ids = append(ids, p.ID)
		}
		t.Fatalf("迁移后存在 %d 页真实孤儿 %v (按类型 %v) —— 成功路径泄漏页",
			len(R.Orphans), ids, R.ByType)
	}
}

func TestMigrateVacuumLeavesNoOrphanPages(t *testing.T) {
	src := buildSQLiteSource(t)
	dst := migrateToVRTX(t, src)

	db, err := core.Open(dst)
	if err != nil {
		t.Fatalf("打开迁移产物: %v", err)
	}
	if _, err := db.VACUUM(); err != nil {
		t.Fatalf("VACUUM: %v", err)
	}
	if err := db.Close(); err != nil {
		t.Fatalf("关闭: %v", err)
	}

	R := checkOrphans(t, dst)
	if len(R.Orphans) > 0 {
		var ids []uint32
		for _, p := range R.Orphans {
			ids = append(ids, p.ID)
		}
		t.Fatalf("迁移+VACUUM 后存在 %d 页真实孤儿 %v (按类型 %v)", len(R.Orphans), ids, R.ByType)
	}
}