TsDB2/demo/wal_crash_chaos/main.go

package main

// =============================================================================
// wal_crash_chaos — 随机断电压测
//
// 父进程循环 N 轮:
//   1. 起子进程, 子进程向全新数据库按 "7 行一个显式事务" 循环提交
//      (负载为按行号确定性生成的 128 字节 + XXH32 校验和);
//   2. 随机 20~250ms 后强杀子进程 (不做任何清理, 脏页随进程蒸发, 模拟断电);
//   3. 父进程重开数据库 —— WAL 恢复 + 头页 CRC 校验在这一步发生 —— 然后断言:
//      - 库能正常打开, 无 panic / 无结构损坏;
//      - 已提交前缀 1..K 连续完整, 每行校验和对得上 (无丢页 / 无半条);
//      - 半截事务 (K 之后至多 6 行 + 在途) 完全不可见 (提交边界);
//      - 表元数据 TotalRows 与实际可见行一致。
//
// 默认 120 轮 (env TSDB_CRASH_ROUNDS 可调), 固定种子 (env TSDB_CRASH_SEED,
// 默认 1) 保证可复现 —— 失败时输出复现命令。任何一轮失败都以非零退出,
// 并把现场数据库留在临时目录路径打印出来。
//
// 这是 WAL 正确性的真实故障注入: 杀点覆盖 提交前 / 提交中(WAL 写链) /
// 提交后(数据页回写) / 两 fsync 之间 的全部时序窗口。
// =============================================================================

import (
	"encoding/binary"
	"fmt"
	"io"
	"math/rand"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"time"

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

const (
	envChild  = "TSDB_CRASH_CHAOS_CHILD"
	envDB     = "TSDB_CRASH_DB"
	envRounds = "TSDB_CRASH_ROUNDS"
	envSeed   = "TSDB_CRASH_SEED"

	payloadLen = 128
	batchSize  = 7 // 每个显式事务提交的行数; K(已提交前缀) 必须是它的整数倍
	probeExtra = 8 // 已提交前缀之后必须不存在的探测行数 (>= 半截事务最大 6 行 + 余量)
)

func main() {
	if os.Getenv(envChild) == "1" {
		runChild()
		return
	}
	runParent()
}

// =============================================================================
// 确定性负载: 父进程按行号重新生成, 不需要子进程留任何痕迹
// =============================================================================

func payloadFor(id uint64) []byte {
	buf := make([]byte, payloadLen)
	s := id*0x9E3779B97F4A7C15 + 0xD1B54A32D192ED03
	for i := 0; i < payloadLen; i += 8 {
		// splitmix64 步进: 纯位运算, 与平台无关
		s ^= s >> 12
		s ^= s << 25
		s ^= s >> 27
		binary.LittleEndian.PutUint64(buf[i:], s*0x2545F4914F6CDD1D)
	}
	return buf
}

func checksumFor(id uint64) uint64 {
	return uint64(core.XXH32(payloadFor(id)))
}

// =============================================================================
// 子进程: 建库, 然后无限循环 "插 7 行 → Commit", 直到被强杀
// =============================================================================

func runChild() {
	path := os.Getenv(envDB)
	db, err := core.Open(path)
	if err != nil {
		os.Exit(2)
	}
	db.SetAutoCommitDelay(0) // 关掉防抖: 提交边界完全由显式 Commit 决定

	tbl, err := db.CreateTable("t", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Uint64, FieldName: []byte("c")},
		{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(payloadLen), FieldName: []byte("p")},
	})
	if err != nil {
		os.Exit(2)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		os.Exit(2)
	}

	for i := uint64(1); ; i++ {
		if _, err := tbl.InsertRowWithID(i, []types.FieldValue{
			{FieldName: []byte("id"), Value: i},
			{FieldName: []byte("c"), Value: checksumFor(i)},
			{FieldName: []byte("p"), Value: string(payloadFor(i))},
		}); err != nil {
			os.Exit(2)
		}
		if i%batchSize == 0 {
			if err := db.Commit(); err != nil { // WAL 写链 + fsync ×2 发生在这里
				os.Exit(2)
			}
			// 干净模式: 写满 N 批后正常关闭 (校验压测装置本身: 走 Close 路径必须零损伤)
			if cleanBatches := envInt("TSDB_CRASH_CLEAN_BATCHES", 0); cleanBatches > 0 && int(i/batchSize) == cleanBatches {
				if err := db.Close(); err != nil {
					os.Exit(2)
				}
				os.Exit(0)
			}
		}
	}
}

// =============================================================================
// 父进程: 循环 起子进程 → 随机时刻强杀 → 重开校验
// =============================================================================

func runParent() {
	rounds := envInt(envRounds, 120)
	baseSeed := int64(envInt(envSeed, 1))
	rng := rand.New(rand.NewSource(baseSeed))

	fmt.Printf("随机断电压测: %d 轮, 种子 %d (复现: %s=<%s> %s=轮数 go run main.go)\n",
		rounds, baseSeed, envSeed, envRounds, envSeed)
	for r := 1; r <= rounds; r++ {
		if err := oneRound(rng); err != nil {
			fmt.Printf("❌ 第 %d 轮失败 (种子 %d): %v\n", r, baseSeed, err)
			fmt.Printf("   复现: %s=%d %s=%d go run main.go\n", envSeed, baseSeed, envRounds, r)
			os.Exit(1)
		}
		if r%20 == 0 {
			fmt.Printf("  进度 %d/%d 通过\n", r, rounds)
		}
	}
	fmt.Printf("✅ 随机断电压测通过: %d 轮, 每轮随机时刻强杀子进程后 —— 已提交前缀完整、校验和全对、未提交事务不可见\n", rounds)
}

func oneRound(rng *rand.Rand) error {
	dir, err := os.MkdirTemp("", "tsdb-chaos-")
	if err != nil {
		return err
	}
	path := filepath.Join(dir, "chaos.db")

	exe, err := os.Executable()
	if err != nil {
		os.RemoveAll(dir)
		return err
	}
	cmd := exec.Command(exe)
	cmd.Env = append(os.Environ(), envChild+"=1", envDB+"="+path)
	if cleanBatches := envInt("TSDB_CRASH_CLEAN_BATCHES", 0); cleanBatches > 0 {
		// 干净模式: 子进程写满 N 批后自行 Close 退出, 不杀 —— 校验压测装置本身
		cmd.Env = append(cmd.Env, "TSDB_CRASH_CLEAN_BATCHES="+strconv.Itoa(cleanBatches))
	}
	if err := cmd.Start(); err != nil {
		os.RemoveAll(dir)
		return err
	}
	if envInt("TSDB_CRASH_CLEAN_BATCHES", 0) > 0 {
		if err := cmd.Wait(); err != nil {
			os.RemoveAll(dir)
			return fmt.Errorf("干净模式子进程异常退出: %v", err)
		}
	} else {
		// 随机存活 20~250ms: 杀点均匀覆盖 提交前 / WAL 写链中 / 页回写中 / 两 fsync 之间
		time.Sleep(time.Duration(20+rng.Intn(230)) * time.Millisecond)
		_ = cmd.Process.Kill()
		_ = cmd.Wait() // 回收句柄, 确保文件锁释放后再重开
	}

	if _, err := os.Stat(path); os.IsNotExist(err) {
		os.RemoveAll(dir)
		return nil // 子进程在建文件前被杀, 本轮无对象可验
	}
	// 先留一份原始现场再校验: core.Open 会触发 WAL 重放 + Close 刷盘,
	// 直接开在原件上就把崩溃现场改掉了
	evidence := filepath.Join(dir, "chaos_evidence.db")
	if err := copyFile(path, evidence); err != nil {
		return err
	}
	if err := verify(path, dir); err != nil {
		return err // 现场保留在 dir (chaos.db=重放后, chaos_evidence.db=原始)
	}
	return os.RemoveAll(dir)
}

func copyFile(src, dst string) error {
	in, err := os.Open(src)
	if err != nil {
		return err
	}
	defer in.Close()
	out, err := os.Create(dst)
	if err != nil {
		return err
	}
	defer out.Close()
	_, err = io.Copy(out, in)
	return err
}

// =============================================================================
// 恢复后校验: 已提交前缀完整 + 校验和 + 未提交不可见 + 元数据一致
// =============================================================================

func verify(path, dir string) error {
	db, err := core.Open(path) // RecoverWAL + HeaderCrc 在这里跑
	if err != nil {
		return fmt.Errorf("崩溃后重开失败: %w (现场: %s)", err, dir)
	}
	defer db.Close()

	tbl, ok := db.FindTable("t")
	if !ok {
		// 子进程在建表窗口内被杀: 库结构完好即算过, 后续断言无对象
		return nil
	}

	// 已提交前缀: id 从 1 开始连续存在, 第一个洞即是边界 K
	var k uint64
	for k = 0; ; k++ {
		if _, err := tbl.FindByPK(k + 1); err != nil {
			break
		}
	}

	if total := tbl.TotalRows(); total != k {
		return fmt.Errorf("TotalRows=%d 但实际连续前缀=%d (存在不可达的洞行/幽灵行) (现场: %s)", total, k, dir)
	}

	// 提交边界: K 之后必须完全不可见 (半截事务 + 在途行)
	for id := k + 1; id <= k+probeExtra; id++ {
		if _, err := tbl.FindByPK(id); err == nil {
			return fmt.Errorf("未提交行 id=%d 在恢复后可见, 提交边界被破坏 (现场: %s)", id, dir)
		}
	}

	// 已提交行逐行校验: 负载长度 + XXH32 校验和, 抓半条/错位/旧值
	for id := uint64(1); id <= k; id++ {
		vals, err := tbl.FindByPK(id)
		if err != nil {
			return fmt.Errorf("已提交行 id=%d 读取失败: %v (现场: %s)", id, err, dir)
		}
		var c uint64
		var p []byte
		for _, fv := range vals {
			switch string(fv.FieldName) {
			case "c":
				c, _ = fv.Value.(uint64)
			case "p":
				// core 反序列化字符串返回 []byte (非 string), 两种都接
				switch v := fv.Value.(type) {
				case string:
					p = []byte(v)
				case []byte:
					p = v
				}
			}
		}
		if len(p) != payloadLen || c != checksumFor(id) || uint64(core.XXH32(p)) != c {
			return fmt.Errorf("已提交行 id=%d 数据损坏 (len=%d, c=%d, want=%d) (现场: %s)",
				id, len(p), c, checksumFor(id), dir)
		}
	}
	return nil
}

func envInt(key string, def int) int {
	if v := os.Getenv(key); v != "" {
		var n int
		if _, err := fmt.Sscanf(v, "%d", &n); err == nil && n > 0 {
			return n
		}
	}
	return def
}