修复Windows下MySQL空闲连接恢复并验证写入不重放

backend/internal/database/database.go

@@ -55,7 +55,13 @@ func Open(ctx context.Context, cfg config.DatabaseConfig) (*Database, error) {
 	if driver == "postgres" {
 		sqlDriver = "pgx"
 	}
-	db, err := sql.Open(sqlDriver, dsn)
+	var db *sql.DB
+	var err error
+	if driver == "mysql" {
+		db, err = openMySQLPool(dsn)
+	} else {
+		db, err = sql.Open(sqlDriver, dsn)
+	}
 	if err != nil {
 		return nil, fmt.Errorf("打开数据库失败：%w", err)
 	}

backend/internal/database/mysql_pool.go

@@ -0,0 +1,83 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"fmt"
+	"runtime"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+)
+
+const mysqlReuseProbeTimeout = 2 * time.Second
+
+func openMySQLPool(dsn string) (*sql.DB, error) {
+	cfg, err := mysql.ParseDSN(dsn)
+	if err != nil {
+		return nil, fmt.Errorf("解析 MySQL 连接池配置失败")
+	}
+	connector, err := mysql.NewConnector(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("创建 MySQL 连接器失败")
+	}
+	// 驱动在 Windows 上没有空闲套接字检查；Linux 继续使用驱动原有检查，避免额外往返。
+	if runtime.GOOS == "windows" && cfg.CheckConnLiveness {
+		connector = &mysqlPoolConnector{Connector: connector}
+	}
+	return sql.OpenDB(connector), nil
+}
+
+// 保留锁定驱动实现的全部连接接口，不能因包装丢失事务选项、参数转换或上下文支持。
+type mysqlDriverConn interface {
+	driver.Conn
+	driver.ConnBeginTx
+	driver.ConnPrepareContext
+	driver.Execer
+	driver.ExecerContext
+	driver.Queryer
+	driver.QueryerContext
+	driver.NamedValueChecker
+	driver.Pinger
+	driver.SessionResetter
+	driver.Validator
+}
+
+type mysqlPoolConnector struct {
+	driver.Connector
+}
+
+func (c *mysqlPoolConnector) Connect(ctx context.Context) (driver.Conn, error) {
+	conn, err := c.Connector.Connect(ctx)
+	if err != nil {
+		return nil, err
+	}
+	full, ok := conn.(mysqlDriverConn)
+	if !ok {
+		_ = conn.Close()
+		return nil, fmt.Errorf("MySQL 驱动连接接口不兼容")
+	}
+	return &mysqlPoolConn{mysqlDriverConn: full}, nil
+}
+
+type mysqlPoolConn struct {
+	mysqlDriverConn
+}
+
+func (c *mysqlPoolConn) ResetSession(ctx context.Context) error {
+	if err := c.mysqlDriverConn.ResetSession(ctx); err != nil {
+		return err
+	}
+	probeCtx, cancel := context.WithTimeout(ctx, mysqlReuseProbeTimeout)
+	defer cancel()
+	if err := c.mysqlDriverConn.Ping(probeCtx); err != nil {
+		_ = c.mysqlDriverConn.Close()
+		// 仅在业务语句发送前淘汰连接；不得把执行或提交错误改成可重试错误。
+		return driver.ErrBadConn
+	}
+	return nil
+}
+
+var _ driver.Connector = (*mysqlPoolConnector)(nil)
+var _ mysqlDriverConn = (*mysqlPoolConn)(nil)

backend/internal/database/mysql_pool_test.go

@@ -0,0 +1,196 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"errors"
+	"testing"
+	"time"
+)
+
+type poolTestConn struct {
+	mysqlDriverConn
+	reset     func(context.Context) error
+	ping      func(context.Context) error
+	execErr   error
+	queryErr  error
+	commitErr error
+	closed    bool
+	probes    int
+	writes    int
+	queries   int
+	commits   int
+	options   driver.TxOptions
+}
+
+func (c *poolTestConn) ResetSession(ctx context.Context) error {
+	if c.reset != nil {
+		return c.reset(ctx)
+	}
+	return nil
+}
+
+func (c *poolTestConn) Ping(ctx context.Context) error {
+	c.probes++
+	if c.ping != nil {
+		return c.ping(ctx)
+	}
+	return ctx.Err()
+}
+
+func (c *poolTestConn) Close() error {
+	c.closed = true
+	return nil
+}
+
+func (c *poolTestConn) IsValid() bool { return !c.closed }
+
+func (c *poolTestConn) ExecContext(context.Context, string, []driver.NamedValue) (driver.Result, error) {
+	c.writes++
+	return driver.RowsAffected(1), c.execErr
+}
+
+func (c *poolTestConn) QueryContext(context.Context, string, []driver.NamedValue) (driver.Rows, error) {
+	c.queries++
+	return nil, c.queryErr
+}
+
+func (c *poolTestConn) BeginTx(_ context.Context, opts driver.TxOptions) (driver.Tx, error) {
+	c.options = opts
+	return &poolTestTx{conn: c}, nil
+}
+
+type poolTestTx struct{ conn *poolTestConn }
+
+func (tx *poolTestTx) Commit() error {
+	tx.conn.commits++
+	return tx.conn.commitErr
+}
+
+func (tx *poolTestTx) Rollback() error { return nil }
+
+type poolTestConnector struct {
+	driver.Connector
+	connections []*poolTestConn
+	calls       int
+}
+
+func (c *poolTestConnector) Connect(context.Context) (driver.Conn, error) {
+	if c.calls >= len(c.connections) {
+		return nil, errors.New("测试连接已耗尽")
+	}
+	conn := c.connections[c.calls]
+	c.calls++
+	return conn, nil
+}
+
+func TestMySQLPoolResetSession(t *testing.T) {
+	t.Run("healthy_probe_has_deadline", func(t *testing.T) {
+		conn := &poolTestConn{ping: func(ctx context.Context) error {
+			deadline, ok := ctx.Deadline()
+			if !ok || time.Until(deadline) <= 0 || time.Until(deadline) > mysqlReuseProbeTimeout {
+				t.Fatal("复用探测必须具有有限超时")
+			}
+			return nil
+		}}
+		if err := (&mysqlPoolConn{conn}).ResetSession(t.Context()); err != nil {
+			t.Fatal(err)
+		}
+		if conn.probes != 1 || conn.closed || conn.writes != 0 {
+			t.Fatal("健康连接不应关闭或执行业务语句")
+		}
+	})
+	t.Run("driver_rejects_before_probe", func(t *testing.T) {
+		conn := &poolTestConn{reset: func(context.Context) error { return driver.ErrBadConn }}
+		if err := (&mysqlPoolConn{conn}).ResetSession(t.Context()); !errors.Is(err, driver.ErrBadConn) || conn.probes != 0 {
+			t.Fatal("原驱动拒绝连接后不能继续探测", err)
+		}
+	})
+	t.Run("failed_probe_closes_connection", func(t *testing.T) {
+		conn := &poolTestConn{ping: func(context.Context) error { return errors.New("连接中断") }}
+		if err := (&mysqlPoolConn{conn}).ResetSession(t.Context()); !errors.Is(err, driver.ErrBadConn) || !conn.closed || conn.writes != 0 {
+			t.Fatal("探测失败必须在写入前丢弃连接", err)
+		}
+	})
+	t.Run("shorter_caller_deadline_wins", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+		defer cancel()
+		want, _ := ctx.Deadline()
+		conn := &poolTestConn{ping: func(probe context.Context) error {
+			got, ok := probe.Deadline()
+			if !ok || !got.Equal(want) {
+				t.Fatal("探测不能延长调用者截止时间")
+			}
+			<-probe.Done()
+			return probe.Err()
+		}}
+		if err := (&mysqlPoolConn{conn}).ResetSession(ctx); !errors.Is(err, driver.ErrBadConn) || !conn.closed {
+			t.Fatal("超时连接必须丢弃", err)
+		}
+	})
+	t.Run("canceled_context", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel()
+		conn := &poolTestConn{}
+		if err := (&mysqlPoolConn{conn}).ResetSession(ctx); !errors.Is(err, driver.ErrBadConn) || !conn.closed || conn.writes != 0 {
+			t.Fatal("取消后不能使用探测连接执行写入", err)
+		}
+	})
+}
+
+func TestMySQLPoolReplacesOnlyBeforeExecution(t *testing.T) {
+	stale, fresh := &poolTestConn{}, &poolTestConn{}
+	connector := &poolTestConnector{connections: []*poolTestConn{stale, fresh}}
+	db := sql.OpenDB(&mysqlPoolConnector{connector})
+	defer db.Close()
+	db.SetMaxOpenConns(1)
+	if err := db.PingContext(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	stale.ping = func(context.Context) error { return errors.New("空闲连接失效") }
+	if _, err := db.ExecContext(t.Context(), "UPDATE counter SET value = value + 1"); err != nil {
+		t.Fatal(err)
+	}
+	if connector.calls != 2 || !stale.closed || stale.writes != 0 || fresh.writes != 1 {
+		t.Fatal("只允许替换业务写入前失效的连接")
+	}
+}
+
+func TestMySQLPoolDoesNotReplayUncertainOperations(t *testing.T) {
+	uncertain := errors.New("响应丢失，执行结果不确定")
+	for _, operation := range []string{"exec", "query", "commit"} {
+		t.Run(operation, func(t *testing.T) {
+			conn := &poolTestConn{execErr: uncertain, queryErr: uncertain, commitErr: uncertain}
+			connector := &poolTestConnector{connections: []*poolTestConn{conn}}
+			db := sql.OpenDB(&mysqlPoolConnector{connector})
+			defer db.Close()
+			if err := db.PingContext(t.Context()); err != nil {
+				t.Fatal(err)
+			}
+			var err error
+			switch operation {
+			case "exec":
+				_, err = db.ExecContext(t.Context(), "UPDATE counter SET value = value + 1")
+			case "query":
+				_, err = db.QueryContext(t.Context(), "CALL change_and_return_counter()")
+			case "commit":
+				var tx *sql.Tx
+				tx, err = db.BeginTx(t.Context(), &sql.TxOptions{Isolation: sql.LevelSerializable, ReadOnly: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if conn.options.Isolation != driver.IsolationLevel(sql.LevelSerializable) || !conn.options.ReadOnly {
+					t.Fatal("事务隔离级别和只读选项必须原样传递")
+				}
+				err = tx.Commit()
+			}
+			if !errors.Is(err, uncertain) || errors.Is(err, driver.ErrBadConn) || connector.calls != 1 {
+				t.Fatal("结果不确定的操作必须保留原错误且不能重建连接重放", err)
+			}
+			if conn.writes+conn.queries+conn.commits != 1 || conn.probes != 2 {
+				t.Fatal("操作只能执行一次，事务内不能插入额外探测")
+			}
+		})
+	}
+}

backend/internal/database/mysql_recovery_integration_test.go

@@ -0,0 +1,171 @@
+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)
+	}
+}