feat: support database connectors and HTTP server configuration

This commit is contained in:
2026-09-24 02:18:18 +08:00
parent 6200864a64
commit 20e84e5ff8
22 changed files with 1403 additions and 63 deletions

View File

@@ -3,7 +3,7 @@
`database` 包提供基于标准库 `database/sql` 的连接池生命周期,不绑定 ORM 或驱动:
- `Config` / `ConfigFromEnv`:连接池参数和 `GO_OMNI_DB_*` 环境变量。
- `Open`:创建连接池、设置连接上限并在返回前执行 `PingContext`。
- `Open` / `OpenConnector`:通过 DSN 或标准 Connector 创建连接池、设置连接上限并在返回前执行 `PingContext`。
- `Health`:用于就绪检查,等待连接和 Ping 均受 `PingTimeout` 限制。
- `InTx`:统一事务提交和回滚;工作函数返回错误或 panic 时回滚。
- `Close`:由嵌入的 `*sql.DB` 提供,应用退出时显式调用。
@@ -40,7 +40,7 @@ GO_OMNI_DB_CONN_MAX_IDLE_TIME=5m
GO_OMNI_DB_PING_TIMEOUT=5s
```
`DSN` 必填,驱动默认是 `mysql`。数据库建库、迁移、表结构和具体 Repository 属于业务项目或可选模块,不会被核心包自动执行。
使用 `ConfigFromEnv` / `Open` 时 `DSN` 必填,环境变量入口的驱动默认是 `mysql`。数据库建库、迁移、表结构和具体 Repository 属于业务项目或可选模块,不会被核心包自动执行。
`MaxIdleConns=0` 会禁用空闲连接。初始化 Ping 失败会关闭已创建的连接池。
@@ -64,3 +64,61 @@ python3 scripts/verifyDatabase.py --version v0.2.0
`Config` 的 JSON/YAML 字段使用 lowerCamelCase,`DSN` 使用 `json:"-"`,连接密钥从环境变量加载。
API 脚手架 `--database mysql` 使用此流程,并要求正的 `maxOpenConns`,提供数据库 readiness 与幂等清理。
具体表结构、迁移和 Repository 留在业务项目,参考 `examples/userService`。
## Connector 接入(v0.9.1-preview)
项目已有 `database/sql/driver.Connector` 时使用 `OpenConnector`。这是标准驱动接口,核心包不增加具体驱动依赖。项目可以在同一进程创建多个使用不同证书、日志、拨号策略的连接池。
下面是 MySQL 的完整组装函数;连接地址、凭据、TLS 和驱动 Logger 都从调用方传入,不在框架中约定:
```go
package connection
import (
"context"
"crypto/tls"
"github.com/go-sql-driver/mysql"
"git.yuntuo.hk/k2safe/go-omni/database"
)
func Open(ctx context.Context, address, user, password, name string,
tlsConfig *tls.Config, logger mysql.Logger,
pool database.PoolConfig) (*database.DB, error) {
cfg := mysql.NewConfig()
cfg.Net, cfg.Addr = "tcp", address
cfg.User, cfg.Passwd, cfg.DBName = user, password, name
cfg.TLS = tlsConfig // 项目设置 RootCAs、ServerName、MinVersion 等
if logger != nil {
cfg.Logger = logger // 项目现有的 Print(...any) 日志适配器
}
connector, err := mysql.NewConnector(cfg)
if err != nil {
return nil, err
}
return database.OpenConnector(ctx, connector, pool)
}
```
调用时以 `database.DefaultPoolConfig()` 为起点,按项目需要调整。已有 `database.Config` 可调用 `.PoolConfig()` 提取池设置;`ConfigFromEnv` / `ApplyEnv` 仍是 DSN 入口,不用为 Connector 填假 DSN。`PoolConfig` 支持通过 `config.DecodeFile` 加载项目的 JSON/YAML 嵌套配置,环境覆盖由项目按自己的驱动配置组织。
| PoolConfig 字段 | 默认值 / 含义 |
| --- | --- |
| `MaxOpenConns` | `10`;`0` 表示不限制 |
| `MaxIdleConns` | `5`;`0` 禁止保留空闲连接 |
| `ConnMaxLifetime` | `30m`;`0` 不因连接年龄回收 |
| `ConnMaxIdleTime` | `5m`;`0` 不因空闲时长回收 |
| `PingTimeout` | `5s`;必须为正,约束启动 Ping 和 Health 等待 |
返回的 `*database.DB` 嵌入 `*sql.DB`,可使用 `QueryContext`、`ExecContext`、`Stats`,以及框架的 `Health`、`InTx`。需要标准数据库指针的 ORM/迁移器可取 `db.DB`;不要重复建立池。项目负责驱动支持的 context 取消与读写超时设置,框架不能强制中断不遵守 context 的自定义驱动。
**生命周期:** nil/带类型的 nil Connector、非法池参数、启动前已取消的 context 会在创建池前返回,Connector 仍归调用方。进入 `sql.OpenDB` 后由池持有;启动 Ping 失败自动关闭,成功后由调用方负责 `db.Close`,通常注册 `app.OnShutdown(db.Close)`。若 Connector 实现 `io.Closer`,遵循标准 `database/sql` 规则随池关闭;有资源的 Connector 不应跨多个池共享。资源初始化过程中、注册退出回调之前若业务校验失败,项目也须关闭池。
`db.PoolConfig()` 返回池配置快照。Connector 入口的 `db.Config()` 中 `Driver` / `DSN` 为空,不能用它重建 Connector。TLS、Logger、驱动版本检查、字符集、时区以及迁移规则保持由项目负责;不写入框架配置快照。
验收脚本使用独立本地 MySQL 8.4 容器(1 CPU / 768 MiB),自动生成测试 CA/证书与随机凭据,检查 TLS 信任/主机名、两池配置隔离、驱动日志、事务和 HTTP 超时,结束后清理本次资源:
```sh
python3 scripts/verifyConnector.py --local --report performance/connectorLocal.json
# 发布后通过全新 module cache 验证远程依赖,无本地 replace:
python3 scripts/verifyConnector.py --version v0.9.1-preview --report performance/connectorRemote.json
```

82
database/connector.go Normal file
View File

@@ -0,0 +1,82 @@
package database
import (
"context"
"database/sql"
"database/sql/driver"
"errors"
"reflect"
"time"
)
// PoolConfig contains only driver-independent settings. Connection credentials,
// TLS, logging and dial behavior belong to the application's Connector.
// Start from DefaultPoolConfig; zero PingTimeout is invalid.
type PoolConfig struct {
MaxOpenConns int `json:"maxOpenConns"`
MaxIdleConns int `json:"maxIdleConns"`
ConnMaxLifetime time.Duration `json:"connMaxLifetime"`
ConnMaxIdleTime time.Duration `json:"connMaxIdleTime"`
PingTimeout time.Duration `json:"pingTimeout"`
}
func DefaultPoolConfig() PoolConfig {
return PoolConfig{MaxOpenConns: 10, MaxIdleConns: 5, ConnMaxLifetime: 30 * time.Minute, ConnMaxIdleTime: 5 * time.Minute, PingTimeout: 5 * time.Second}
}
func (c PoolConfig) Validate() error {
if c.MaxOpenConns < 0 || c.MaxIdleConns < 0 {
return errors.New("database pool limits cannot be negative")
}
if c.MaxOpenConns > 0 && c.MaxIdleConns > c.MaxOpenConns {
return errors.New("max idle connections cannot exceed max open connections")
}
if c.ConnMaxLifetime < 0 || c.ConnMaxIdleTime < 0 || c.PingTimeout <= 0 {
return errors.New("database timeouts are invalid")
}
return nil
}
func (c PoolConfig) config() Config {
return Config{MaxOpenConns: c.MaxOpenConns, MaxIdleConns: c.MaxIdleConns, ConnMaxLifetime: c.ConnMaxLifetime, ConnMaxIdleTime: c.ConnMaxIdleTime, PingTimeout: c.PingTimeout}
}
// OpenConnector opens an independent pool using the supplied standard Connector.
// No registered driver name or DSN is required. It applies cfg and completes a
// startup Ping bounded by ctx and PingTimeout before returning.
//
// Validation failures and an already-canceled ctx leave the Connector untouched.
// Once sql.OpenDB is called, the pool owns its lifecycle: a failed startup closes
// the pool; on success the caller must Close the returned DB. As in database/sql,
// this also closes a Connector that implements io.Closer. Use a separate Connector
// for each pool if it owns resources. Nil ctx means context.Background.
func OpenConnector(ctx context.Context, connector driver.Connector, cfg PoolConfig) (*DB, error) {
if nilConnector(connector) {
return nil, errors.New("database connector is required")
}
if err := cfg.Validate(); err != nil {
return nil, err
}
if ctx == nil {
ctx = context.Background()
}
if err := ctx.Err(); err != nil {
return nil, err
}
return initializePool(ctx, sql.OpenDB(connector), cfg.config())
}
func nilConnector(connector driver.Connector) bool {
if connector == nil {
return true
}
v := reflect.ValueOf(connector)
switch v.Kind() {
case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
return v.IsNil()
}
return false
}
// PoolConfig returns a copy of the settings used to create this pool.
func (db *DB) PoolConfig() PoolConfig { return db.config.PoolConfig() }

336
database/connector_test.go Normal file
View File

@@ -0,0 +1,336 @@
package database_test
import (
"context"
"database/sql"
"database/sql/driver"
"errors"
"io"
"sync"
"sync/atomic"
"testing"
"time"
"git.yuntuo.hk/k2safe/go-omni/database"
)
type fixtureConnector struct {
identity string
connects, closedConnections, closedConnector, commits, rollbacks atomic.Int32
connect func(context.Context) error
ping func(context.Context) error
query func(context.Context) error
closeErr error
}
func (c *fixtureConnector) Connect(ctx context.Context) (driver.Conn, error) {
c.connects.Add(1)
if c.connect != nil {
if err := c.connect(ctx); err != nil {
return nil, err
}
}
return &fixtureConn{owner: c}, nil
}
func (c *fixtureConnector) Driver() driver.Driver { return fixtureDriver{} }
func (c *fixtureConnector) Close() error { c.closedConnector.Add(1); return c.closeErr }
type fixtureDriver struct{}
func (fixtureDriver) Open(string) (driver.Conn, error) {
return nil, errors.New("unexpected driver name/DSN path")
}
type fixtureConn struct{ owner *fixtureConnector }
func (c *fixtureConn) Prepare(string) (driver.Stmt, error) { return nil, errors.New("not used") }
func (c *fixtureConn) Close() error { c.owner.closedConnections.Add(1); return nil }
func (c *fixtureConn) Begin() (driver.Tx, error) { return fixtureTx{c.owner}, nil }
func (c *fixtureConn) Ping(ctx context.Context) error {
if c.owner.ping != nil {
return c.owner.ping(ctx)
}
return nil
}
func (c *fixtureConn) QueryContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) {
if c.owner.query != nil {
if err := c.owner.query(ctx); err != nil {
return nil, err
}
}
return &fixtureRows{identity: c.owner.identity}, nil
}
type fixtureTx struct{ owner *fixtureConnector }
func (tx fixtureTx) Commit() error { tx.owner.commits.Add(1); return nil }
func (tx fixtureTx) Rollback() error { tx.owner.rollbacks.Add(1); return nil }
type fixtureRows struct {
identity string
sent bool
}
func (r *fixtureRows) Columns() []string { return []string{"identity"} }
func (r *fixtureRows) Close() error { return nil }
func (r *fixtureRows) Next(values []driver.Value) error {
if r.sent {
return io.EOF
}
r.sent = true
values[0] = r.identity
return nil
}
func TestConnectorIndependentPoolsAndContext(t *testing.T) {
type key struct{}
ctx := context.WithValue(context.Background(), key{}, "startup")
cfg := database.DefaultPoolConfig()
cfg.MaxOpenConns = 2
cfg.MaxIdleConns = 2
var owners []*fixtureConnector
var pools []*database.DB
for _, identity := range []string{"alpha", "beta"} {
owner := &fixtureConnector{identity: identity}
// Startup Connect/Ping and each query preserve their caller context.
// database/sql may also open replacement connections in the background.
owner.connect = func(ctx context.Context) error {
if owner.connects.Load() == 1 && ctx.Value(key{}) != "startup" {
return errors.New("lost request context")
}
return nil
}
owner.ping = func(ctx context.Context) error {
if ctx.Value(key{}) != "startup" {
return errors.New("lost ping context")
}
return nil
}
owner.query = func(ctx context.Context) error {
if ctx.Value(key{}) != "startup" {
return errors.New("lost query context")
}
return nil
}
pool, err := database.OpenConnector(ctx, owner, cfg)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { pool.Close() })
if pool.PoolConfig() != cfg || pool.Config().Driver != "" || pool.Config().DSN != "" || pool.Stats().MaxOpenConnections != 2 {
t.Fatal("pool settings or DSN boundary")
}
owners = append(owners, owner)
pools = append(pools, pool)
}
var wg sync.WaitGroup
for i, pool := range pools {
for n := 0; n < 16; n++ {
wg.Add(1)
go func(i int, pool *database.DB) {
defer wg.Done()
var identity string
if err := pool.QueryRowContext(ctx, "identity").Scan(&identity); err != nil {
t.Error(err)
} else if identity != owners[i].identity {
t.Error("crossed connectors")
}
}(i, pool)
}
}
wg.Wait()
for i, pool := range pools {
if pool.Stats().OpenConnections > cfg.MaxOpenConns {
t.Fatal("pool cap not applied")
}
if err := pool.Close(); err != nil {
t.Fatal(err)
}
if err := pool.Close(); err != nil {
t.Fatal(err)
}
if owners[i].closedConnector.Load() != 1 || owners[i].closedConnections.Load() != owners[i].connects.Load() {
t.Fatal("close did not release owned resources exactly once")
}
if pool.Health(ctx) == nil {
t.Fatal("closed pool is healthy")
}
}
}
func TestConnectorRejectsInvalidInputWithoutTakingOwnership(t *testing.T) {
var typedNil *fixtureConnector
for _, connector := range []driver.Connector{nil, typedNil} {
if db, err := database.OpenConnector(nil, connector, database.DefaultPoolConfig()); err == nil || db != nil {
t.Fatal("nil connector accepted")
}
}
for _, mutate := range []func(*database.PoolConfig){
func(c *database.PoolConfig) { c.MaxOpenConns = -1 }, func(c *database.PoolConfig) { c.MaxIdleConns = -1 },
func(c *database.PoolConfig) { c.MaxIdleConns = c.MaxOpenConns + 1 }, func(c *database.PoolConfig) { c.PingTimeout = 0 },
func(c *database.PoolConfig) { c.ConnMaxLifetime = -1 }, func(c *database.PoolConfig) { c.ConnMaxIdleTime = -1 },
} {
cfg := database.DefaultPoolConfig()
mutate(&cfg)
owner := &fixtureConnector{}
if db, err := database.OpenConnector(nil, owner, cfg); err == nil || db != nil {
t.Fatal("invalid config accepted")
}
if owner.connects.Load() != 0 || owner.closedConnector.Load() != 0 {
t.Fatal("validation consumed connector")
}
}
owner := &fixtureConnector{}
ctx, cancel := context.WithCancel(context.Background())
cancel()
if db, err := database.OpenConnector(ctx, owner, database.DefaultPoolConfig()); !errors.Is(err, context.Canceled) || db != nil {
t.Fatalf("pre-canceled context: %v", err)
}
if owner.connects.Load() != 0 || owner.closedConnector.Load() != 0 {
t.Fatal("pre-cancellation consumed connector")
}
if err := database.DefaultConfig().Validate(); err == nil {
t.Fatal("legacy DSN requirements changed")
}
}
func TestConnectorStartupFailuresReleaseResources(t *testing.T) {
failure := errors.New("fixture unavailable")
closeFailure := errors.New("fixture close failed")
for _, phase := range []string{"connect", "ping"} {
t.Run(phase, func(t *testing.T) {
owner := &fixtureConnector{closeErr: closeFailure}
if phase == "connect" {
owner.connect = func(context.Context) error { return failure }
} else {
owner.ping = func(context.Context) error { return failure }
}
pool, err := database.OpenConnector(nil, owner, database.DefaultPoolConfig())
if pool != nil || !errors.Is(err, failure) || !errors.Is(err, closeFailure) {
t.Fatalf("lost failure or cleanup error: %v", err)
}
expectedConnections := int32(0)
if phase == "ping" {
expectedConnections = 1
}
if owner.closedConnector.Load() != 1 || owner.closedConnections.Load() != expectedConnections {
t.Fatal("startup failure leaked resources")
}
})
}
}
func TestConnectorTimeoutBoundsConnectAndPing(t *testing.T) {
for _, phase := range []string{"connect", "ping"} {
t.Run(phase, func(t *testing.T) {
owner := &fixtureConnector{}
cfg := database.DefaultPoolConfig()
cfg.PingTimeout = 20 * time.Millisecond
wait := func(ctx context.Context) error {
if _, ok := ctx.Deadline(); !ok {
return errors.New("missing deadline")
}
<-ctx.Done()
return ctx.Err()
}
if phase == "connect" {
owner.connect = wait
} else {
owner.ping = wait
}
pool, err := database.OpenConnector(context.Background(), owner, cfg)
if pool != nil || !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("%s timeout: %v", phase, err)
}
if owner.closedConnector.Load() != 1 {
t.Fatal("timeout leaked connector")
}
})
}
}
func TestConnectorHealthBoundsPoolWaitAndRecovers(t *testing.T) {
cfg := database.DefaultPoolConfig()
cfg.MaxOpenConns = 1
cfg.MaxIdleConns = 1
cfg.PingTimeout = 20 * time.Millisecond
owner := &fixtureConnector{}
pool, err := database.OpenConnector(nil, owner, cfg)
if err != nil {
t.Fatal(err)
}
defer pool.Close()
conn, err := pool.Conn(context.Background())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
if err = pool.Health(nil); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("waiting for busy pool was not bounded: %v", err)
}
if err = conn.Close(); err != nil {
t.Fatal(err)
}
if err = pool.Health(nil); err != nil {
t.Fatal(err)
}
}
func TestConnectorIdleZeroAndLifetime(t *testing.T) {
for _, mode := range []string{"noIdle", "expired"} {
t.Run(mode, func(t *testing.T) {
cfg := database.DefaultPoolConfig()
if mode == "noIdle" {
cfg.MaxIdleConns = 0
} else {
cfg.ConnMaxLifetime = time.Nanosecond
}
owner := &fixtureConnector{}
pool, err := database.OpenConnector(nil, owner, cfg)
if err != nil {
t.Fatal(err)
}
defer pool.Close()
if err = pool.Health(nil); err != nil {
t.Fatal(err)
}
if owner.connects.Load() < 2 || owner.closedConnections.Load() == 0 {
t.Fatal("pool did not retire connections")
}
if mode == "noIdle" && pool.Stats().Idle != 0 {
t.Fatal("zero idle not honored")
}
if mode == "expired" && pool.Stats().MaxLifetimeClosed == 0 {
t.Fatal("lifetime not honored")
}
})
}
}
func TestConnectorTransactionsKeepCommitAndRollbackSemantics(t *testing.T) {
owner := &fixtureConnector{}
pool, err := database.OpenConnector(nil, owner, database.DefaultPoolConfig())
if err != nil {
t.Fatal(err)
}
defer pool.Close()
ctx := context.Background()
if err = pool.InTx(ctx, nil, func(*sql.Tx) error { return nil }); err != nil {
t.Fatal(err)
}
failed := errors.New("business failed")
if err = pool.InTx(ctx, nil, func(*sql.Tx) error { return failed }); !errors.Is(err, failed) {
t.Fatal(err)
}
func() {
defer func() {
if recover() != failed {
t.Error("panic not preserved")
}
}()
_ = pool.InTx(ctx, nil, func(*sql.Tx) error { panic(failed) })
}()
if owner.commits.Load() != 1 || owner.rollbacks.Load() != 2 {
t.Fatal("transaction semantics changed")
}
}

View File

@@ -23,13 +23,14 @@ type Config struct {
PingTimeout time.Duration `json:"pingTimeout"`
}
func DefaultConfig() Config {
return Config{
MaxOpenConns: 10,
MaxIdleConns: 5,
ConnMaxLifetime: 30 * time.Minute,
ConnMaxIdleTime: 5 * time.Minute,
PingTimeout: 5 * time.Second,
func DefaultConfig() Config { return DefaultPoolConfig().config() }
// PoolConfig returns driver-independent connection pool settings.
func (c Config) PoolConfig() PoolConfig {
return PoolConfig{
MaxOpenConns: c.MaxOpenConns, MaxIdleConns: c.MaxIdleConns,
ConnMaxLifetime: c.ConnMaxLifetime, ConnMaxIdleTime: c.ConnMaxIdleTime,
PingTimeout: c.PingTimeout,
}
}
@@ -83,16 +84,7 @@ func (c Config) Validate() error {
if strings.TrimSpace(c.DSN) == "" {
return errors.New("database DSN is required")
}
if c.MaxOpenConns < 0 || c.MaxIdleConns < 0 {
return errors.New("database pool limits cannot be negative")
}
if c.MaxIdleConns > c.MaxOpenConns && c.MaxOpenConns > 0 {
return errors.New("max idle connections cannot exceed max open connections")
}
if c.ConnMaxLifetime < 0 || c.ConnMaxIdleTime < 0 || c.PingTimeout <= 0 {
return errors.New("database timeouts are invalid")
}
return nil
return c.PoolConfig().Validate()
}
// DB owns a connection pool and exposes the embedded database/sql methods.
@@ -112,23 +104,30 @@ func Open(ctx context.Context, cfg Config) (*DB, error) {
if err != nil {
return nil, fmt.Errorf("open database: %w", err)
}
if cfg.MaxOpenConns > 0 {
db.SetMaxOpenConns(cfg.MaxOpenConns)
}
// database/sql uses zero to disable idle connections; do not leave its
// default pool size in place when the caller explicitly chooses zero.
db.SetMaxIdleConns(cfg.MaxIdleConns)
return initializePool(ctx, db, cfg)
}
// initializePool is the shared ownership boundary for both creation paths.
// All configuration is validated before creating the sql.DB.
func initializePool(ctx context.Context, db *sql.DB, cfg Config) (*DB, error) {
db.SetMaxOpenConns(cfg.MaxOpenConns)
db.SetMaxIdleConns(cfg.MaxIdleConns) // Zero explicitly disables idle connections.
db.SetConnMaxLifetime(cfg.ConnMaxLifetime)
db.SetConnMaxIdleTime(cfg.ConnMaxIdleTime)
checkCtx, cancel := context.WithTimeout(ctx, cfg.PingTimeout)
defer cancel()
if err := db.PingContext(checkCtx); err != nil {
_ = db.Close()
return nil, fmt.Errorf("ping database: %w", err)
pingErr := fmt.Errorf("ping database: %w", err)
if closeErr := db.Close(); closeErr != nil {
return nil, errors.Join(pingErr, fmt.Errorf("close database after failed ping: %w", closeErr))
}
return nil, pingErr
}
return &DB{DB: db, config: cfg}, nil
}
// Config returns the original DSN configuration. Pools opened with a Connector
// have empty Driver/DSN fields; their connector cannot be reconstructed from Config.
func (db *DB) Config() Config { return db.config }
func (db *DB) Health(ctx context.Context) error {

View File

@@ -0,0 +1,180 @@
package database_test
import (
"context"
"crypto/tls"
"crypto/x509"
"database/sql"
"errors"
"net"
"os"
"sync/atomic"
"testing"
"time"
"git.yuntuo.hk/k2safe/go-omni/database"
"github.com/go-sql-driver/mysql"
)
type connectorDriverLog struct{ count atomic.Int32 }
func (l *connectorDriverLog) Print(...any) { l.count.Add(1) }
func TestMySQLConnectorDriverLoggerIsolation(t *testing.T) {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
stopped := make(chan struct{})
go func() {
defer close(stopped)
for {
c, err := listener.Accept()
if err != nil {
return
}
c.Close()
}
}()
defer func() { listener.Close(); <-stopped }()
logs := [2]connectorDriverLog{}
for i := range logs {
cfg := mysql.NewConfig()
cfg.Net = "tcp"
cfg.Addr = listener.Addr().String()
cfg.Timeout = time.Second
cfg.ReadTimeout = time.Second
cfg.Logger = &logs[i]
connector, err := mysql.NewConnector(cfg)
if err != nil {
t.Fatal(err)
}
before := [2]int32{logs[0].count.Load(), logs[1].count.Load()}
db, err := database.OpenConnector(context.Background(), connector, database.DefaultPoolConfig())
if err == nil {
db.Close()
t.Fatal("handshake to closed socket succeeded")
}
if logs[i].count.Load() <= before[i] || logs[1-i].count.Load() != before[1-i] {
t.Fatal("driver logger did not stay on its connector")
}
}
}
func TestMySQLConnectorTLSAndPoolIsolation(t *testing.T) {
if os.Getenv("GO_OMNI_CONNECTOR_MYSQL_TEST") != "1" {
t.Skip("use scripts/verifyConnector.py for isolated MySQL TLS verification")
}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
ca, err := os.ReadFile(os.Getenv("GO_OMNI_CONNECTOR_CA_FILE"))
if err != nil {
t.Fatal(err)
}
roots := x509.NewCertPool()
if !roots.AppendCertsFromPEM(ca) {
t.Fatal("invalid test CA")
}
cfg := mysql.NewConfig()
cfg.Net = "tcp"
cfg.Addr = os.Getenv("GO_OMNI_CONNECTOR_ADDR")
cfg.User = os.Getenv("GO_OMNI_CONNECTOR_USER")
cfg.Passwd = os.Getenv("GO_OMNI_CONNECTOR_PASSWORD")
cfg.DBName = os.Getenv("GO_OMNI_CONNECTOR_DATABASE")
cfg.Timeout = time.Second
cfg.ReadTimeout = 2 * time.Second
cfg.WriteTimeout = 2 * time.Second
cfg.TLS = &tls.Config{MinVersion: tls.VersionTLS12, ServerName: "localhost", RootCAs: roots}
pool := database.DefaultPoolConfig()
pool.MaxOpenConns = 2
pool.MaxIdleConns = 2
pool.PingTimeout = 2 * time.Second
open := func(cfg *mysql.Config) (*database.DB, error) {
connector, err := mysql.NewConnector(cfg)
if err != nil {
return nil, err
}
return database.OpenConnector(ctx, connector, pool)
}
cfg.Params = map[string]string{"time_zone": "'+00:00'"}
first, err := open(cfg)
if err != nil {
t.Fatal(err)
}
defer first.Close()
other := cfg.Clone()
other.Params["time_zone"] = "'+08:00'"
second, err := open(other)
if err != nil {
t.Fatal(err)
}
defer second.Close()
for i, db := range []*database.DB{first, second} {
var name, cipher, zone string
if err := db.QueryRowContext(ctx, "SHOW SESSION STATUS LIKE 'Ssl_cipher'").Scan(&name, &cipher); err != nil {
t.Fatal(err)
}
if cipher == "" {
t.Fatal("connection is not encrypted")
}
if err := db.QueryRowContext(ctx, "SELECT @@session.time_zone").Scan(&zone); err != nil {
t.Fatal(err)
}
if zone != []string{"+00:00", "+08:00"}[i] {
t.Fatalf("session config leaked: %s", zone)
}
if db.Stats().MaxOpenConnections != 2 {
t.Fatal("pool configuration ignored")
}
}
for _, bad := range []string{"untrustedCA", "wrongHostname"} {
t.Run(bad, func(t *testing.T) {
invalid := cfg.Clone()
if bad == "untrustedCA" {
invalid.TLS.RootCAs = x509.NewCertPool()
} else {
invalid.TLS.ServerName = "wrong.invalid"
}
db, err := open(invalid)
if err == nil {
db.Close()
t.Fatal("invalid server certificate accepted")
}
var verification *tls.CertificateVerificationError
if !errors.As(err, &verification) {
t.Fatalf("not a TLS verification failure: %v", err)
}
if err := first.Health(ctx); err != nil {
t.Fatalf("failed connector affected existing pool: %v", err)
}
})
}
if _, err := first.ExecContext(ctx, "CREATE TABLE connector_check (id INT PRIMARY KEY)"); err != nil {
t.Fatal(err)
}
if err := first.InTx(ctx, nil, func(tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, "INSERT INTO connector_check VALUES (1)")
return err
}); err != nil {
t.Fatal(err)
}
rollbackErr := errors.New("rollback requested")
if err := first.InTx(ctx, nil, func(tx *sql.Tx) error {
if _, err := tx.ExecContext(ctx, "INSERT INTO connector_check VALUES (2)"); err != nil {
return err
}
return rollbackErr
}); !errors.Is(err, rollbackErr) {
t.Fatal(err)
}
var count int
if err := second.QueryRowContext(ctx, "SELECT COUNT(*) FROM connector_check").Scan(&count); err != nil || count != 1 {
t.Fatalf("transaction count=%d err=%v", count, err)
}
if err := first.Close(); err != nil {
t.Fatal(err)
}
if err := second.Health(ctx); err != nil {
t.Fatal("closing first pool affected second:", err)
}
}