test(demo): txn_crash_chaos 事务混沌压测 — 无清单验证事务原子性/持久性/回滚

子进程循环 'Begin → 插 5 行 → 按 decisionFor(txn) 确定性提交或回滚',
父进程随机时刻强杀 (杀点大量落在事务中途 / Commit 中途) 后重开校验:
每个可见事务必须完整 (seq 恰 0..4, 半截=原子性破坏); 已决提交事务必须
全部可见 (持久性); 回滚事务与被杀的在途事务必须零行; 逐行 XXH32 校验和;
TotalRows 与实际可见一致; 缺失 id 只能属于回滚/在途事务。决策由 splitmix64
确定性重算, 不写任何旁路记录。已抓获并修复 DDL 回滚元数据分叉 bug
(见 166840e)。默认 60 轮, 种子可复现。

demo/txn_crash_chaos/main.go

@@ -0,0 +1,379 @@
+package main
+
+// =============================================================================
+// txn_crash_chaos — 事务混沌压测
+//
+// 验证事务语义在随机断电下的正确性, 不需要清单文件 (manifest):
+//   - 每个事务写 txnSize 行 (列: id / txn / seq / 校验和 / 负载);
+//   - commit 还是 rollback 由 decisionFor(txn) 确定性决定 (同一 txn 双方各自
+//     重算, 父进程由此得知"哪些事务必须可见" —— 子进程不写任何旁路记录);
+//   - 父进程在随机时刻强杀子进程 (大量杀点落在事务中途 / Commit 中途),
+//     重开后从库内容反推并断言:
+//       1. 原子性: 每个可见事务必须完整 (seq 恰为 0..txnSize-1, 一行不多
+//          不少) —— 半截事务 = 提交边界被破坏;
+//       2. 持久性: 每个已决提交的事务 (≤ 最大可见事务号) 必须全部可见;
+//       3. 回滚: 任何回滚事务的行必须一行都不剩; 被杀时正在执行的事务
+//          (最大可见事务号 +1) 必须零行;
+//       4. 行完整性: 每行负载长度 + XXH32 校验和必须对上 (半条/错位/旧值);
+//       5. 元数据: TotalRows 与实际可见行一致; 缺失的 id 只能属于回滚
+//          事务或被杀的在途事务 (id 按 (txn-1)*txnSize+seq+1 连续发放)。
+//
+// 默认 60 轮 (env TSDB_CRASH_ROUNDS), 种子可复现 (env TSDB_CRASH_SEED)。
+// 失败保留原始现场 (evidence 副本) 并打印复现命令。
+// =============================================================================
+
+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"
+	envWindow = "TSDB_CRASH_WINDOW"
+	envClean  = "TSDB_CRASH_CLEAN_TXNS"
+
+	txnSize = 5 // 每个事务的行数
+	endGap  = 32 // 连续多少个 id 缺失判定为扫描终点 (回滚一次最多浪费 txnSize 个)
+)
+
+var killWindow = envInt(envWindow, 300)
+
+func main() {
+	if os.Getenv(envChild) == "1" {
+		runChild()
+		return
+	}
+	runParent()
+}
+
+// =============================================================================
+// 确定性决策与负载: 父子进程各自重算, 不写任何旁路记录
+// =============================================================================
+
+// decisionFor 某事务是提交还是回滚: 12.5% 回滚。splitmix64 纯位运算, 跨进程一致。
+func decisionFor(txn uint64) (rollback bool) {
+	s := txn*0x9E3779B97F4A7C15 + 0xD1B54A32D192ED03
+	s ^= s >> 12
+	s ^= s << 25
+	s ^= s >> 27
+	s *= 0x2545F4914F6CDD1D
+	return s%8 == 0
+}
+
+func payloadFor(id uint64) []byte {
+	buf := make([]byte, 128)
+	s := id*0x9E3779B97F4A7C15 + 0xD1B54A32D192ED03
+	for i := 0; i < len(buf); i += 8 {
+		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))) }
+
+// =============================================================================
+// 子进程: 循环 "Begin → 插 txnSize 行 → 按决策 Commit/Rollback", 直到被强杀
+// =============================================================================
+
+func runChild() {
+	path := os.Getenv(envDB)
+	db, err := core.Open(path)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "open:", err)
+		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("txn")},
+		{FieldType: types.FieldType_Uint64, FieldName: []byte("seq")},
+		{FieldType: types.FieldType_Uint64, FieldName: []byte("c")},
+		{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(128), FieldName: []byte("p")},
+	})
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "create table:", err)
+		os.Exit(2)
+	}
+	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
+		fmt.Fprintln(os.Stderr, "create index:", err)
+		os.Exit(2)
+	}
+
+	var id uint64
+	for txn := uint64(1); ; txn++ {
+		tx := db.BeginTransaction()
+		for seq := uint64(0); seq < txnSize; seq++ {
+			id++
+			if _, err := tbl.InsertRowWithID(id, []types.FieldValue{
+				{FieldName: []byte("id"), Value: id},
+				{FieldName: []byte("txn"), Value: txn},
+				{FieldName: []byte("seq"), Value: seq},
+				{FieldName: []byte("c"), Value: checksumFor(id)},
+				{FieldName: []byte("p"), Value: string(payloadFor(id))},
+			}); err != nil {
+				fmt.Fprintln(os.Stderr, "insert fail: id", id, "txn", txn, "seq", seq, "err:", err)
+				_ = tx.Rollback()
+				os.Exit(2)
+			}
+		}
+		if decisionFor(txn) {
+			if err := tx.Rollback(); err != nil {
+				fmt.Fprintln(os.Stderr, "rollback txn", txn, ":", err)
+				os.Exit(2)
+			}
+		} else {
+			if err := tx.Commit(); err != nil {
+				fmt.Fprintln(os.Stderr, "commit txn", txn, ":", err)
+				os.Exit(2)
+			}
+		}
+		// 干净模式: 跑满 N 个事务后正常关闭, 校验装置本身
+		if clean := envInt(envClean, 0); clean > 0 && int(txn) == clean {
+			if err := db.Close(); err != nil {
+				fmt.Fprintln(os.Stderr, "close:", err)
+				os.Exit(2)
+			}
+			os.Exit(0)
+		}
+	}
+}
+
+// =============================================================================
+// 父进程
+// =============================================================================
+
+func runParent() {
+	rounds := envInt(envRounds, 60)
+	baseSeed := int64(envInt(envSeed, 1))
+	rng := rand.New(rand.NewSource(baseSeed))
+
+	fmt.Printf("事务混沌压测: %d 轮, 种子 %d\n", rounds, baseSeed)
+	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-txn-chaos-")
+	if err != nil {
+		return err
+	}
+	path := filepath.Join(dir, "txn.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)
+	cmd.Stderr = os.Stderr // 子进程诊断直接透传 (干净模式排障用)
+	if clean := envInt(envClean, 0); clean > 0 {
+		cmd.Env = append(cmd.Env, envClean+"="+strconv.Itoa(clean))
+	}
+	if err := cmd.Start(); err != nil {
+		os.RemoveAll(dir)
+		return err
+	}
+	if envInt(envClean, 0) > 0 {
+		if err := cmd.Wait(); err != nil {
+			os.RemoveAll(dir)
+			return fmt.Errorf("干净模式子进程异常退出: %v", err)
+		}
+	} else {
+		time.Sleep(time.Duration(20+rng.Intn(killWindow-20)) * time.Millisecond)
+		_ = cmd.Process.Kill()
+		_ = cmd.Wait()
+	}
+
+	if _, err := os.Stat(path); os.IsNotExist(err) {
+		os.RemoveAll(dir)
+		return nil
+	}
+	evidence := filepath.Join(dir, "txn_evidence.db")
+	if err := copyFile(path, evidence); err != nil {
+		return err
+	}
+	if err := verify(path, dir); err != nil {
+		return err
+	}
+	return os.RemoveAll(dir)
+}
+
+// =============================================================================
+// 恢复后校验: 原子性 + 持久性 + 回滚 + 行完整性 + 元数据
+// =============================================================================
+
+type row struct {
+	txn uint64
+	seq uint64
+	c   uint64
+	p   []byte
+}
+
+func verify(path, dir string) error {
+	db, err := core.Open(path)
+	if err != nil {
+		return fmt.Errorf("崩溃后重开失败: %w (现场: %s)", err, dir)
+	}
+	defer db.Close()
+	tbl, ok := db.FindTable("t")
+	if !ok {
+		return nil // 建表前被杀, 无对象可验
+	}
+
+	// 逐 id 探测收集全部可见行 (id 连续发放, 回滚/在途事务造成洞)
+	rows := map[uint64]*row{}
+	var missRun, maxID uint64
+	for id := uint64(1); ; id++ {
+		vals, err := tbl.FindByPK(id)
+		if err != nil {
+			if missRun++; missRun >= endGap {
+				break
+			}
+			continue
+		}
+		missRun = 0
+		maxID = id
+		var rn row
+		for _, fv := range vals {
+			switch string(fv.FieldName) {
+			case "txn":
+				rn.txn, _ = fv.Value.(uint64)
+			case "seq":
+				rn.seq, _ = fv.Value.(uint64)
+			case "c":
+				rn.c, _ = fv.Value.(uint64)
+			case "p":
+				switch v := fv.Value.(type) {
+				case string:
+					rn.p = []byte(v)
+				case []byte:
+					rn.p = v
+				}
+			}
+		}
+		rows[id] = &rn
+	}
+
+	// 行完整性: 负载长度 + 校验和
+	for id, rn := range rows {
+		if len(rn.p) != 128 || rn.c != checksumFor(id) || uint64(core.XXH32(rn.p)) != rn.c {
+			return fmt.Errorf("行 id=%d 数据损坏 (len=%d c=%d want=%d) (现场: %s)",
+				id, len(rn.p), rn.c, checksumFor(id), dir)
+		}
+	}
+
+	// 按事务分组, 校验原子性 / 决策一致性
+	byTxn := map[uint64]map[uint64]uint64{} // txn → seq → id
+	var maxVisibleTxn uint64
+	for id, rn := range rows {
+		if byTxn[rn.txn] == nil {
+			byTxn[rn.txn] = map[uint64]uint64{}
+		}
+		if _, dup := byTxn[rn.txn][rn.seq]; dup {
+			return fmt.Errorf("事务 %d seq=%d 重复行 (id=%d) (现场: %s)", rn.txn, rn.seq, id, dir)
+		}
+		byTxn[rn.txn][rn.seq] = id
+		if rn.txn > maxVisibleTxn {
+			maxVisibleTxn = rn.txn
+		}
+	}
+	for txn, seqs := range byTxn {
+		if decisionFor(txn) {
+			return fmt.Errorf("回滚事务 %d 仍有 %d 行可见 (回滚失效) (现场: %s)", txn, len(seqs), dir)
+		}
+		if txn == maxVisibleTxn+1 {
+			return fmt.Errorf("在途事务 %d 在恢复后可见 %d 行 (提交边界被破坏) (现场: %s)", txn, len(seqs), dir)
+		}
+		// 原子性: 已决事务必须完整 (seq 恰为 0..txnSize-1)
+		if len(seqs) != txnSize {
+			return fmt.Errorf("事务 %d 不完整: %d/%d 行可见 (原子性被破坏) (现场: %s)", txn, len(seqs), txnSize, dir)
+		}
+		for seq := uint64(0); seq < txnSize; seq++ {
+			if _, ok := seqs[seq]; !ok {
+				return fmt.Errorf("事务 %d 缺 seq=%d (原子性被破坏) (现场: %s)", txn, seq, dir)
+			}
+		}
+	}
+	// 持久性: 每个已决提交且 ≤ 最大可见事务号的事务都必须可见
+	for txn := uint64(1); txn <= maxVisibleTxn; txn++ {
+		if !decisionFor(txn) {
+			if _, ok := byTxn[txn]; !ok {
+				return fmt.Errorf("已提交事务 %d 不可见 (持久性被破坏) (现场: %s)", txn, dir)
+			}
+		}
+	}
+
+	// 元数据: TotalRows 与实际可见行一致
+	if total := tbl.TotalRows(); total != uint64(len(rows)) {
+		return fmt.Errorf("TotalRows=%d 但实际可见=%d (现场: %s)", total, len(rows), dir)
+	}
+
+	// id 连续性: 每个缺失的 id 只能属于回滚事务或被杀的在途事务
+	for id := uint64(1); id <= maxID; id++ {
+		if rows[id] != nil {
+			continue
+		}
+		txn := (id-1)/txnSize + 1
+		if txn == maxVisibleTxn+1 {
+			continue // 在途事务, 允许整段缺失
+		}
+		if decisionFor(txn) {
+			continue // 回滚事务, 允许整段缺失
+		}
+		return fmt.Errorf("已提交事务 %d 的 id=%d 缺失 (现场: %s)", txn, id, dir)
+	}
+	return nil
+}
+
+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 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
+}