diff --git a/backend/internal/database/database.go b/backend/internal/database/database.go index b78aba0..cfd543e 100644 --- a/backend/internal/database/database.go +++ b/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) } diff --git a/backend/internal/database/mysql_pool.go b/backend/internal/database/mysql_pool.go new file mode 100644 index 0000000..60c1547 --- /dev/null +++ b/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) diff --git a/backend/internal/database/mysql_pool_test.go b/backend/internal/database/mysql_pool_test.go new file mode 100644 index 0000000..7a54d57 --- /dev/null +++ b/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("操作只能执行一次,事务内不能插入额外探测") + } + }) + } +} diff --git a/backend/internal/database/mysql_recovery_integration_test.go b/backend/internal/database/mysql_recovery_integration_test.go new file mode 100644 index 0000000..94db81d --- /dev/null +++ b/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) + } +}