TsDB2/demo/txn_crash_chaos/main.go

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
}