Scert/backend/internal/database/mysql_recovery_integration_test.go

package database

import (
	"context"
	"database/sql"
	"math"
	"testing"
	"time"
)

func TestMySQLRealPoolRecovery(t *testing.T) {
	f := startRealMySQL(t)
	ctx := t.Context()
	t.Run("idle_pool_survives_restart", func(t *testing.T) {
		db := mysqlRecoveryDatabase(t, f)
		if db.DB.Stats().Idle != 1 {
			t.Fatal("重启前必须有真实空闲连接")
		}
		f.stop(t)
		f.start(t, false)
		// 不关闭应用池、不调用额外 Ping；第一条业务写入必须成功且只生效一次。
		if _, err := db.DB.ExecContext(ctx, "UPDATE pool_recovery SET value = value + 1 WHERE id = 1"); err != nil {
			t.Fatal("数据库重启后第一条写入失败", err)
		}
		mysqlAssertRecoveryValue(t, db.DB, 1)
	})
	t.Run("prepared_statement_survives_idle_disconnect", func(t *testing.T) {
		db := mysqlRecoveryDatabase(t, f)
		stmt, err := db.DB.PrepareContext(ctx, "UPDATE pool_recovery SET value = value + ? WHERE id = ?")
		if err != nil {
			t.Fatal(err)
		}
		defer stmt.Close()
		mysqlKillPoolConnection(t, f, db.DB)
		if _, err := stmt.ExecContext(ctx, 2, 1); err != nil {
			t.Fatal("空闲断线后预编译语句失败", err)
		}
		mysqlAssertRecoveryValue(t, db.DB, 2)
	})
	t.Run("query_and_migration_survive_idle_disconnect", func(t *testing.T) {
		db := mysqlRecoveryDatabase(t, f)
		mysqlKillPoolConnection(t, f, db.DB)
		mysqlAssertRecoveryValue(t, db.DB, 0)
		mysqlKillPoolConnection(t, f, db.DB)
		if err := Migrate(ctx, db.DB, "mysql"); err != nil {
			t.Fatal("空闲断线后首次迁移失败", err)
		}
		mysqlAssertRecoveryValue(t, db.DB, 0)
	})
	t.Run("parameter_conversion_and_commit", func(t *testing.T) {
		db := mysqlRecoveryDatabase(t, f)
		var value string
		if err := db.DB.QueryRowContext(ctx, "SELECT ?", uint64(math.MaxUint64)).Scan(&value); err != nil || value != "18446744073709551615" {
			t.Fatal("驱动无符号整数参数转换必须保留", err)
		}
		tx, err := db.DB.BeginTx(ctx, nil)
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		stmt, err := tx.PrepareContext(ctx, "UPDATE pool_recovery SET value = ? WHERE id = ?")
		if err != nil {
			t.Fatal(err)
		}
		defer stmt.Close()
		if _, err := stmt.ExecContext(ctx, 7, 1); err != nil {
			t.Fatal(err)
		}
		if err := tx.Commit(); err != nil {
			t.Fatal(err)
		}
		mysqlAssertRecoveryValue(t, db.DB, 7)
	})
	t.Run("transaction_options_and_rollback", func(t *testing.T) {
		db := mysqlRecoveryDatabase(t, f)
		writable, err := db.DB.BeginTx(ctx, nil)
		if err != nil {
			t.Fatal(err)
		}
		defer writable.Rollback()
		if _, err := writable.ExecContext(ctx, "UPDATE pool_recovery SET value = 40 WHERE id = 1"); err != nil {
			t.Fatal(err)
		}
		if err := writable.Rollback(); err != nil {
			t.Fatal(err)
		}
		mysqlKillPoolConnection(t, f, db.DB)
		tx, err := db.DB.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable, ReadOnly: true})
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		if _, err := tx.ExecContext(ctx, "UPDATE pool_recovery SET value = 99 WHERE id = 1"); err == nil {
			t.Fatal("只读事务不应接受写入")
		}
		if err := tx.Rollback(); err != nil {
			t.Fatal(err)
		}
		mysqlAssertRecoveryValue(t, db.DB, 0)
	})
	t.Run("active_transaction_is_not_replayed", func(t *testing.T) {
		db := mysqlRecoveryDatabase(t, f)
		tx, err := db.DB.BeginTx(ctx, nil)
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		if _, err := tx.ExecContext(ctx, "UPDATE pool_recovery SET value = 99 WHERE id = 1"); err != nil {
			t.Fatal(err)
		}
		var id int64
		if err := tx.QueryRowContext(ctx, "SELECT CONNECTION_ID()").Scan(&id); err != nil {
			t.Fatal(err)
		}
		if _, err := f.admin.ExecContext(ctx, "KILL CONNECTION ?", id); err != nil {
			t.Fatal(err)
		}
		if err := tx.Commit(); err == nil {
			t.Fatal("连接中断后的事务不能报告提交成功")
		}
		mysqlAssertRecoveryValue(t, db.DB, 0)
	})
	t.Run("database_down_is_bounded_and_then_recovers", func(t *testing.T) {
		db := mysqlRecoveryDatabase(t, f)
		f.stop(t)
		bounded, cancel := context.WithTimeout(ctx, 250*time.Millisecond)
		defer cancel()
		started := time.Now()
		_, err := db.DB.ExecContext(bounded, "UPDATE pool_recovery SET value = 99 WHERE id = 1")
		if err == nil || time.Since(started) > 3*time.Second {
			t.Fatal("数据库不可用时必须有界失败", err)
		}
		f.start(t, false)
		mysqlAssertRecoveryValue(t, db.DB, 0)
	})
}

func mysqlRecoveryDatabase(t *testing.T, f *realMySQL) *Database {
	t.Helper()
	db := f.database(t)
	db.DB.SetMaxOpenConns(1)
	if _, err := db.DB.ExecContext(t.Context(), "CREATE TABLE pool_recovery (id BIGINT PRIMARY KEY, value BIGINT NOT NULL)"); err != nil {
		t.Fatal(err)
	}
	if _, err := db.DB.ExecContext(t.Context(), "INSERT INTO pool_recovery VALUES (1, 0)"); err != nil {
		t.Fatal(err)
	}
	return db
}

func mysqlKillPoolConnection(t *testing.T, f *realMySQL, db *sql.DB) {
	t.Helper()
	var id int64
	if err := db.QueryRowContext(t.Context(), "SELECT CONNECTION_ID()").Scan(&id); err != nil {
		t.Fatal(err)
	}
	if _, err := f.admin.ExecContext(t.Context(), "KILL CONNECTION ?", id); err != nil {
		t.Fatal(err)
	}
}

func mysqlAssertRecoveryValue(t *testing.T, db *sql.DB, want int64) {
	t.Helper()
	var got int64
	if err := db.QueryRowContext(t.Context(), "SELECT value FROM pool_recovery WHERE id = 1").Scan(&got); err != nil {
		t.Fatal(err)
	}
	if got != want {
		t.Fatalf("写入次数或回滚结果不符合预期：得到 %d，期望 %d", got, want)
	}
}