项目文件夹

文件
杭州明婳科技 ffcb1a77b9 docs: 注释/文档一致性审查修复(A/B/C 三档共 23 项)
对 60 个非测试源文件做文档/注释/代码一致性审查,修复 23 项。全部为注释/文档
改动 + 1 处死代码删除,无运行逻辑变更,无 breaking change。

A 档·实质错误(10 项,会误导下游):
- app.go: closeResources 三连契约注释统一修正--failAfterInit 走
  rollbackReplacedResources 不走 closeResources,closeResources 仅供 doShutdown 复用
- database/manager.go: RandomPicker 注释从过时的 rand.Intn/issue 54899 改为 crypto/rand
- model/base.go: datetime(0) 与“亚秒精度”矛盾,改为 datetime(6)
- cache/keybuilder.go: 三处示例下划线分隔符改为冒号(与默认分隔符 : 一致)
- repository/repository.go: Delete 接口注释从绝对“软删除”改为条件性;
  UpdateFields 示例移除把字段名当 WHERE 条件的误用写法
- middleware/csrf.go: CSRFToken 注释从“用于 API 模式”改为 Cookie/双重提交模式说明
- examples/full/main.go: POST /users 补上“需 Authorization”标注;删除 _ = app 死代码
- utils/random.go: RandInt/RandInt64 补 min==max 边界说明(对齐 RandIntSecure 写法)

B 档·导出符号注释缺失/不完整(6 项):
- ws/ws.go: 5 个 Type* 常量、4 个 Err* 变量补注释
- console/console.go: 5 个 Level 常量补注释
- jwt/jwt.go: 2 个 Blacklist 常量补注释
- router/router.go: 包级 GroupWithMiddlewareGroup 补注释
- middleware/ratelimit.go: NewRateLimiter 补 panic 条件与 Stop 生命周期说明

C 档·描述不完整(7 项):
- utils/strings.go Substr、convert.go CalcPageCount/CalcOffset、datetime.go
  StartOfMonth/EndOfMonth、file.go CopyFile、validator.go IsChinese、crypto.go HashFile
  补全边界/默认行为/返回值语义
- response/error.go: WithDetail 补 nil 接收者行为说明

验证:go build / go vet / go test -race 全绿。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-12 18:47:12 +08:00

915 行
29 KiB
Go

此文件含有模棱两可的 Unicode 字符
此文件含有可能会与其他字符混淆的 Unicode 字符。 如果您是想特意这样的,可以安全地忽略该警告。 使用 Escape 按钮显示他们。
package database
import (
"context"
cryptorand "crypto/rand"
"database/sql"
"errors"
"fmt"
"math/big"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/EthanCodeCraft/xlgo-core/config"
"github.com/EthanCodeCraft/xlgo-core/logger"
"go.uber.org/zap"
"gorm.io/gorm"
gormlogger "gorm.io/gorm/logger"
)
type dbModeContextKey struct{}
// txContextKey 携带外层事务的 *gorm.DB,使 repository 等上层在调用时能 join 到外层事务
// H6c:外层 ctx 事务无法 join)。由 WithTx 注入、TxFromContext 读取。
type txContextKey struct{}
const (
dbModeMaster = "master"
dbModeReplica = "replica"
)
func normalizeContext(ctx context.Context) context.Context {
if ctx == nil {
return context.Background()
}
return ctx
}
// ReplicaPicker 从库选择策略
type ReplicaPicker interface {
Pick(replicas []*gorm.DB) *gorm.DB
}
// RoundRobinPicker 轮询选择从库
type RoundRobinPicker struct {
mu sync.Mutex
counter int
}
// Pick 轮询选择一个从库
func (p *RoundRobinPicker) Pick(replicas []*gorm.DB) *gorm.DB {
if len(replicas) == 0 {
return nil
}
p.mu.Lock()
idx := p.counter % len(replicas)
p.counter = (idx + 1) % len(replicas)
p.mu.Unlock()
return replicas[idx]
}
// RandomPicker 随机选择从库。
//
// 使用 crypto/rand 生成随机索引,并发安全且不可预测。len(replicas)<=0 返回 nil;
// crypto/rand 失败(极罕见,如熵池耗尽)时回退到 replicas[0],保证可用性。
type RandomPicker struct{}
// Pick 随机选择一个从库
func (p *RandomPicker) Pick(replicas []*gorm.DB) *gorm.DB {
if len(replicas) == 0 {
return nil
}
n, err := cryptorand.Int(cryptorand.Reader, big.NewInt(int64(len(replicas))))
if err != nil {
return replicas[0]
}
return replicas[int(n.Int64())]
}
// Manager 数据库管理器,持有主库与从库连接实例
type Manager struct {
cfg *config.Config
master *gorm.DB
replicas []*gorm.DB
picker ReplicaPicker
mu sync.Mutex
// opMu 串行化 InitDB / InitDBWithReplicas / Close 这类资源生命周期操作。
// 避免关闭已开始后初始化又发布新连接,或初始化发布后被并发 Close 置空。
opMu sync.Mutex
// #21 健康自愈
healthy atomic.Bool // 主库是否健康
replicaHealthy []atomic.Bool // 每个从库的健康标记,索引与 replicas 对齐
probeFailures int // 主库连续探活失败次数
probeMu sync.Mutex // 保护 probeFailures
replicaHealthSet bool // replicaHealthy 是否已按 replicas 长度初始化
}
// NewManager 创建数据库管理器
func NewManager(cfg *config.Config) *Manager {
return &Manager{cfg: cfg, picker: &RandomPicker{}}
}
// getCfg 在锁内读取 m.cfgP1 #11:消除与 InitDB 写 m.cfg 的数据竞争)。
func (m *Manager) getCfg() *config.Config {
m.mu.Lock()
defer m.mu.Unlock()
return m.cfg
}
// setCfg 在锁内写入 m.cfgP1 #11)。
func (m *Manager) setCfg(cfg *config.Config) {
m.mu.Lock()
m.cfg = cfg
m.mu.Unlock()
}
// SetPicker 设置从库选择策略
func (m *Manager) SetPicker(p ReplicaPicker) {
if p == nil {
return
}
m.mu.Lock()
m.picker = p
m.mu.Unlock()
}
// Picker 返回当前从库选择策略
func (m *Manager) Picker() ReplicaPicker {
m.mu.Lock()
defer m.mu.Unlock()
return m.picker
}
// Master 返回主库实例。
// 经 m.mu 锁保护读取,避免与 InitDB/Close 的写竞争返回已关闭/nil 池(C11d)。
func (m *Manager) Master() *gorm.DB {
m.mu.Lock()
defer m.mu.Unlock()
return m.master
}
// Replicas 返回所有从库实例的拷贝。
// 经 m.mu 锁保护读取并返回拷贝,避免调用方持活切片与 InitDBWithReplicas/Close 重置竞争(C11d)。
func (m *Manager) Replicas() []*gorm.DB {
m.mu.Lock()
defer m.mu.Unlock()
if m.replicas == nil {
return nil
}
out := make([]*gorm.DB, len(m.replicas))
copy(out, m.replicas)
return out
}
// Replica 按策略选择一个从库;无从库时返回主库。
// #21:启用探活后,自动过滤不健康的从库;全不健康时回退到全部从库(仍可服务)。
// 全程持 m.mu 锁,避免 replicas/master 与重建路径写竞争(C11d)。
func (m *Manager) Replica() *gorm.DB {
m.mu.Lock()
defer m.mu.Unlock()
if len(m.replicas) == 0 {
return m.master
}
pool := m.replicas
// 启用探活且至少有一个健康标记时,仅从健康从库中选取
if m.replicaHealthSet {
var healthy []*gorm.DB
for i, r := range m.replicas {
if i < len(m.replicaHealthy) && m.replicaHealthy[i].Load() {
healthy = append(healthy, r)
}
}
if len(healthy) > 0 {
pool = healthy
}
// healthy 为空时回退到全部 replicas,避免读流量完全中断
}
if m.picker != nil {
if db := m.picker.Pick(pool); db != nil {
return db
}
}
return pool[0]
}
// IsHealthy 返回主库当前健康状态(#21)。供 readiness/health 探针联动。
func (m *Manager) IsHealthy() bool {
return m.healthy.Load()
}
// initReplicaHealth 按 replicas 数量初始化健康标记(全部为健康)。
// 已初始化(replicaHealthSet=true)时早返回;重建从库前须先 resetReplicaHealth 重置,
// 否则健康切片长度与新 replicas 错位(C11a)。
func (m *Manager) initReplicaHealth() {
m.mu.Lock()
defer m.mu.Unlock()
if m.replicaHealthSet {
return
}
m.replicaHealthy = make([]atomic.Bool, len(m.replicas))
for i := range m.replicaHealthy {
m.replicaHealthy[i].Store(true)
}
m.replicaHealthSet = true
}
// ensureReplicaHealthLocked 按当前 replicas 重建健康标记。调用方须持有 m.mu。
func (m *Manager) ensureReplicaHealthLocked() {
m.replicaHealthy = make([]atomic.Bool, len(m.replicas))
for i := range m.replicaHealthy {
m.replicaHealthy[i].Store(true)
}
m.replicaHealthSet = true
}
// resetReplicaHealth 清空从库健康标记,使下次 initReplicaHealth 按新 replicas 长度重建。
// 重建从库(InitDBWithReplicas/Close 前必须调用,避免健康切片与新 replicas 长度错位(C11a)。
// 调用方须持有 m.mu。
func (m *Manager) resetReplicaHealth() {
m.replicaHealthy = nil
m.replicaHealthSet = false
}
// StartProbing 启动主库与从库的健康探活后台循环(#21)。
// 阻塞调用方,应通过 App.Go 在独立 goroutine 运行;ctx 取消时退出。
// 周期 ping 主库,连续失败达阈值后标记不健康(IsHealthy=false);
// 同时 ping 各从库,失败则从读流量剔除,恢复后自动重新纳入。
func (m *Manager) StartProbing(ctx context.Context) {
ctx = normalizeContext(ctx)
m.initReplicaHealth()
cfg := m.getCfg() // P1 #11:锁内快照,避免与 InitDB 写 m.cfg 竞态
interval := 30 * time.Second
if cfg != nil && cfg.Database.HealthCheckInterval > 0 {
interval = cfg.Database.HealthCheckInterval
}
threshold := 3
if cfg != nil && cfg.Database.HealthCheckFailureThreshold > 0 {
threshold = cfg.Database.HealthCheckFailureThreshold
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
m.probeOnce(ctx, threshold)
}
}
}
// probeOnce 执行一轮主库+从库探活并更新健康标记。
func (m *Manager) probeOnce(ctx context.Context, threshold int) {
// 主库
if err := m.HealthCheck(ctx); err != nil {
m.probeMu.Lock()
m.probeFailures++
if m.probeFailures >= threshold {
if m.healthy.Load() {
logger.Warnf("数据库主库连续探活失败 %d 次,标记为不健康: %v", m.probeFailures, err)
}
m.healthy.Store(false)
}
m.probeMu.Unlock()
} else {
m.probeMu.Lock()
if m.probeFailures >= threshold && !m.healthy.Load() {
logger.Info("数据库主库探活恢复,重新标记为健康")
}
m.probeFailures = 0
m.probeMu.Unlock()
m.healthy.Store(true)
}
// 从库
m.mu.Lock()
replicas := make([]*gorm.DB, len(m.replicas))
copy(replicas, m.replicas)
if len(replicas) > 0 && !m.replicaHealthSet {
m.ensureReplicaHealthLocked()
}
healthSet := m.replicaHealthSet
replicaHealthy := m.replicaHealthy // 快照切片头,避免与 resetReplicaHealth 写竞争
m.mu.Unlock()
if !healthSet {
return
}
for i, r := range replicas {
if r == nil {
continue
}
sqlDB, err := r.DB()
if err != nil {
if i < len(replicaHealthy) {
replicaHealthy[i].Store(false)
}
continue
}
// M11H-db-1):从库探活 ping 经 pingWithTimeout 受 3s 约束,挂起 DB 不无限阻塞探活 goroutine。
if err := pingWithTimeout(sqlDB, ctx); err != nil {
if i < len(replicaHealthy) && replicaHealthy[i].Load() {
logger.Warnf("数据库从库 #%d 探活失败,暂时剔除读流量: %v", i, err)
}
if i < len(replicaHealthy) {
replicaHealthy[i].Store(false)
}
} else {
if i < len(replicaHealthy) {
replicaHealthy[i].Store(true)
}
}
}
}
// FromContext 根据上下文选择数据库
func (m *Manager) FromContext(ctx context.Context) *gorm.DB {
ctx = normalizeContext(ctx)
mode, ok := ctx.Value(dbModeContextKey{}).(string)
if !ok {
return m.Replica()
}
switch mode {
case dbModeMaster:
return m.Master()
case dbModeReplica:
return m.Replica()
default:
return m.Replica()
}
}
// Open 打开主库连接
func (m *Manager) Open(ctx context.Context) error {
cfg := m.getCfg() // P1 #11:锁内读取,避免与 InitDB 写竞态
if cfg == nil {
return errors.New("数据库配置未设置")
}
return m.InitDB(ctx, cfg)
}
// OpenWithReplicas 打开主库与从库连接
func (m *Manager) OpenWithReplicas(ctx context.Context, replicaDSNs []string) error {
cfg := m.getCfg() // P1 #11:锁内读取
if cfg == nil {
return errors.New("数据库配置未设置")
}
return m.InitDBWithReplicas(ctx, cfg, replicaDSNs)
}
// closeDB 关闭 gorm.DB 底层连接池。nil 或未初始化(无 ConnPool)时返回 nil,不 panic。
// 用于重建/关闭路径释放旧池,避免直接覆盖致泄漏(C11b/C11c)。
func closeDB(db *gorm.DB) error {
if db == nil {
return nil
}
sqlDB, err := db.DB()
if err != nil {
return err
}
return sqlDB.Close()
}
func warnCloseDB(db *gorm.DB, context string) {
if err := closeDB(db); err != nil {
logger.Warnf("%s: %v", context, err)
}
}
// Close 关闭主库与全部从库连接,并重置从库健康状态。
// 字段置空在锁内完成(保证新读取得到 nil),实际关闭在锁外执行避免持锁阻塞。
func (m *Manager) Close() error {
m.opMu.Lock()
defer m.opMu.Unlock()
m.mu.Lock()
master := m.master
replicas := m.replicas
m.master = nil
m.replicas = nil
m.resetReplicaHealth()
m.healthy.Store(false)
m.mu.Unlock()
var errs []error
if err := closeDB(master); err != nil {
errs = append(errs, err)
}
for _, replica := range replicas {
if err := closeDB(replica); err != nil {
errs = append(errs, err)
}
}
return errors.Join(errs...)
}
// HealthCheck 健康检查,主库不可达时返回错误。
//
// M11 完整覆盖(H-db-1 修复):ping 经 pingWithTimeout 受 healthCheckTimeout(3s) 约束,
// 使后台探活(probeOnce)与 /health 端点(app.go 经本方法)都不会被挂起 DB(连接活但不
// 响应)无限阻塞。ctx 自带更短 deadline 时优先尊重 ctx。
func (m *Manager) HealthCheck(ctx context.Context) error {
ctx = normalizeContext(ctx)
m.mu.Lock()
db := m.master
m.mu.Unlock()
if db == nil {
return errors.New("数据库主库未初始化")
}
sqlDB, err := db.DB()
if err != nil {
return err
}
return pingWithTimeout(sqlDB, ctx)
}
// DefaultManager 默认数据库管理器,包级 facade 代理到它。
//
// 主线A 修复:改用 atomic.Pointer 保护读写,消除原裸指针(外部直接
// `database.DefaultManager = ...` 赋值与请求 goroutine 经 facade 读取)之间的数据竞争。
// 与 config.defaultManager / database.DefaultRedis / storage.DefaultStorage /
// cache.defaultCachePtr / jwt.defaultManager 对齐——框架内包级可变全局一律 atomic.Pointer。
//
// 类型由 *Manager 变更为 atomic.Pointer[Manager]breaking):下游若直接调用
// DefaultManager.Init/Master 等方法需改用 InitDB/GetDB 等 facade,或
// DefaultManager.Load().Init(...),或经 GetDefaultManager() 取实例后再调方法。
var DefaultManager atomic.Pointer[Manager]
func init() {
DefaultManager.Store(NewManager(nil))
}
// SwapDefaultManager 将指定 Manager 置为全局默认,并返回被替换的旧 Manager。
// 旧 Manager 不会被关闭,供 App 初始化这类需要失败回滚的生命周期流程暂存。
// nil 被忽略,以防 facade Load 到 nil panic。
func SwapDefaultManager(m *Manager) *Manager {
if m == nil {
return DefaultManager.Load()
}
return DefaultManager.Swap(m)
}
// SetDefaultManager 提升指定 Manager 为全局默认,并关闭被替换的旧 Manager。
// 用于多实例场景或测试注入。nil 被忽略以防 facade Load 到 nil panic。
// 若调用方需要保留旧 Manager 用于回滚,请使用 SwapDefaultManager。
func SetDefaultManager(m *Manager) {
if m == nil {
return
}
old := SwapDefaultManager(m)
if old != nil && old != m {
if err := old.Close(); err != nil {
logger.Warnf("关闭被替换的旧数据库 manager 失败: %v", err)
}
}
}
// GetDefaultManager 返回全局默认 Manageratomic 读取,并发安全)。
// 替代直接读 DefaultManager 包级变量(类型已改为 atomic.Pointer,直接读得到的是 atomic
// 值而非 *Manager)。需直接持有 Manager 调用其方法时用本函数或 DefaultManager.Load()。
func GetDefaultManager() *Manager {
return DefaultManager.Load()
}
// InitDB 初始化数据库连接(带重试机制),驱动由配置决定。
// ctx 控制 Ping 与重试等待;调用方取消 ctx 时初始化会尽快返回。
func (m *Manager) InitDB(ctx context.Context, cfg *config.Config) error {
ctx = normalizeContext(ctx)
m.opMu.Lock()
defer m.opMu.Unlock()
return m.initDB(ctx, cfg)
}
func (m *Manager) initDB(ctx context.Context, cfg *config.Config) error {
if cfg == nil {
return errors.New("数据库配置未设置")
}
if err := ctx.Err(); err != nil {
return fmt.Errorf("数据库初始化已取消: %w", err)
}
m.setCfg(cfg) // P1 #11:锁内写入,避免与 StartProbing/Open 读竞态
// GORM 日志配置
var gormLogLevel gormlogger.LogLevel
if cfg.IsDevelopment() {
gormLogLevel = gormlogger.Info
} else {
gormLogLevel = gormlogger.Warn
}
gormConfig := &gorm.Config{
Logger: gormlogger.Default.LogMode(gormLogLevel),
// H-db-1:禁用 gorm.Open 的自动 Pinggorm.go:204,ConnPool 为 *sql.DB 时会调
// pinger.Ping() 无超时)。挂起 DB(连接活但不响应)下该 Ping 无 ctx deadline 无限阻塞,
// 发生在 initDB 的 pingWithTimeout 之前,使启动卡死。框架改用 pingWithTimeout3s)自管
// 启动 ping,故禁用 gorm 无超时自动 ping。InitDBWithReplicas 的 replica 路径同理。
DisableAutomaticPing: true,
}
// M-config-2MySQL 启用 TLS 且配置自定义 CA 时,注册命名 TLS 配置,使 DSN 中 tls=<name> 生效。
// 失败 fail-fast 返回错误,绝不静默回退明文连接。非 MySQL / 未配 CA 为 no-op。
if err := ensureMySQLTLSRegistered(cfg); err != nil {
return err
}
// 重试配置
maxRetries := 5
retryDelay := time.Second
var lastErr error
for i := range maxRetries {
if err := ctx.Err(); err != nil {
return fmt.Errorf("数据库初始化已取消: %w", err)
}
// 连接主库:先打开到局部变量,仅 Ping 成功后才安装为 m.master,
// 避免 Ping 失败时下轮覆盖 m.master 泄漏旧池(C11b)。
db, err := gorm.Open(Dialector(cfg), gormConfig)
if err != nil {
lastErr = err
// 不可恢复的错误(认证失败、未知数据库、DSN 非法等)直接返回,不必重试
if !isTransientDBError(err) {
return fmt.Errorf("数据库连接失败(不可恢复): %w", err)
}
} else {
sqlDB, err := db.DB()
if err != nil {
lastErr = err
warnCloseDB(db, "关闭刚打开的数据库连接池失败") // C11b: 关闭刚打开的池,避免下轮泄漏
} else {
sqlDB.SetMaxIdleConns(cfg.Database.MaxIdleConns)
sqlDB.SetMaxOpenConns(cfg.Database.MaxOpenConns)
sqlDB.SetConnMaxLifetime(time.Hour)
if cfg.Database.ConnMaxIdleTime > 0 {
sqlDB.SetConnMaxIdleTime(cfg.Database.ConnMaxIdleTime)
}
// M11H-db-1):启动 ping 经 pingWithTimeout 受 3s 约束,挂起 DB 致 InitDB 重试失败而非无限阻塞。
if err := pingWithTimeout(sqlDB, ctx); err == nil {
// 成功:安装为新主库,关闭旧主库池(重建路径覆盖前先释放旧资源,C11b)
m.mu.Lock()
old := m.master
m.master = db
m.mu.Unlock()
m.healthy.Store(true) // Ping 通过才标记健康(#21)
warnCloseDB(old, "关闭旧数据库主库连接池失败")
logger.Info("数据库主库连接成功",
zap.String("driver", driverDescription(cfg.Database.Driver)),
zap.String("host", cfg.Database.Host),
zap.Int("port", cfg.Database.Port))
return nil
} else {
// Ping 失败(如服务端暂时不可达)视作可重试
lastErr = err
warnCloseDB(db, "关闭 Ping 失败的数据库连接池失败") // C11b: 关闭刚打开的池,避免下轮覆盖泄漏
}
}
}
logger.Warnf("数据库连接失败,第 %d/%d 次重试: %v", i+1, maxRetries, lastErr)
if i == maxRetries-1 {
break
}
select {
case <-ctx.Done():
return fmt.Errorf("数据库初始化已取消: %w", ctx.Err())
case <-time.After(retryDelay):
}
retryDelay *= 2
if retryDelay > 30*time.Second {
retryDelay = 30 * time.Second
}
}
return fmt.Errorf("数据库连接失败(重试 %d 次): %w", maxRetries, lastErr)
}
// isTransientDBError 判断数据库连接错误是否值得重试。
// 认证失败、未知数据库、非法 DSN/驱动等属于配置类错误,重试无意义,直接返回更友好。
// D5 修复:覆盖 MySQL 和 PostgreSQL 常见非瞬态错误。
func isTransientDBError(err error) bool {
if err == nil {
return true
}
msg := err.Error()
nonTransient := []string{
// MySQL
"Access denied", // 认证失败(用户名/密码错误)
"authentication plugin", // 认证插件不支持
"Unknown database", // 目标库不存在
"invalid DSN", // DSN 语法错误
"unknown driver", // 驱动未注册
"unsupported driver", // 驱动不支持
// PostgreSQLD5 修复;P1 #12:移除过宽的独立 "database" 子串——它会把
// "the database system is starting up" 等瞬态错误误判为非瞬态而放弃重试。
// PG 目标库不存在的消息形如 database "x" does not exist,用 "does not exist" 精确匹配)。
"password authentication failed", // pg 密码错误
"does not exist", // pg 目标库/角色不存在
"no pg_hba.conf entry", // pg_hba.conf 拒绝
}
for _, sub := range nonTransient {
if strings.Contains(msg, sub) {
return false
}
}
return true
}
// replicaMaxOpenConns 计算从库连接池 MaxOpenConnsH-A 修复)。
// masterMax>0 时取 max(1, masterMax/2)(从库适当减少,但绝不截断为 0);
// masterMax<=0(未配置/无限)时返回 0,与主库"无限"语义一致。
func replicaMaxOpenConns(masterMax int) int {
if masterMax <= 0 {
return 0
}
half := masterMax / 2
if half < 1 {
half = 1
}
return half
}
// InitDBWithReplicas 初始化数据库主从连接,驱动由配置决定
// replicaDSNs: 从库连接字符串列表(需与主库驱动匹配)
func (m *Manager) InitDBWithReplicas(ctx context.Context, cfg *config.Config, replicaDSNs []string) error {
ctx = normalizeContext(ctx)
m.opMu.Lock()
defer m.opMu.Unlock()
if cfg == nil {
return errors.New("数据库配置未设置")
}
// 先初始化主库
if err := m.initDB(ctx, cfg); err != nil {
return err
}
// C11c: 重建从库前关闭旧从库池;C11a: 重置健康状态,使下次 initReplicaHealth 按新 replicas 长度重建
m.mu.Lock()
oldReplicas := m.replicas
m.replicas = nil
m.resetReplicaHealth()
m.mu.Unlock()
for _, r := range oldReplicas {
warnCloseDB(r, "关闭旧数据库从库连接池失败")
}
// 初始化从库
if len(replicaDSNs) > 0 {
var gormLogLevel gormlogger.LogLevel
if cfg.IsDevelopment() {
gormLogLevel = gormlogger.Info
} else {
gormLogLevel = gormlogger.Warn
}
gormConfig := &gorm.Config{
Logger: gormlogger.Default.LogMode(gormLogLevel),
// H-db-1:禁用 gorm.Open 自动 Ping(同 initDB),框架用 pingWithTimeout 自管 replica 启动 ping。
DisableAutomaticPing: true,
}
// 先构建到局部切片,全部成功后再安装,避免部分构建期间外部读到中间态
var newReplicas []*gorm.DB
for i, dsn := range replicaDSNs {
if err := ctx.Err(); err != nil {
for _, r := range newReplicas {
warnCloseDB(r, "关闭已打开的数据库从库连接池失败")
}
return fmt.Errorf("数据库从库初始化已取消: %w", err)
}
replicaDB, err := gorm.Open(dialectorForDSN(cfg.Database.Driver, dsn), gormConfig)
if err != nil {
logger.Warnf("数据库从库 %d 连接失败: %v", i+1, err)
continue
}
sqlDB, err := replicaDB.DB()
if err != nil {
logger.Warnf("数据库从库 %d 获取连接池失败: %v", i+1, err)
warnCloseDB(replicaDB, "关闭刚打开的数据库从库连接池失败") // C11c: 关闭刚打开的池避免泄漏
continue
}
sqlDB.SetMaxIdleConns(cfg.Database.MaxIdleConns)
// H-A 修复:从库 MaxOpenConns 适当减少,但须避免截断为 0。
// database/sql 中 SetMaxOpenConns(0) 表示"无限制"——原实现 MaxOpenConns/2 在
// 配置为 1 时得 0,反而让从库连接池无上限(与"减少"意图相反、高并发下打爆 DB);
// 配置为 0(未配置/无限)时 0/2=0 恰好"无限",与主库一致,保持语义。
// 现规则:MaxOpenConns>0 时取 max(1, /2);<=0 时从库亦无限(0),与主库对齐。
sqlDB.SetMaxOpenConns(replicaMaxOpenConns(cfg.Database.MaxOpenConns))
sqlDB.SetConnMaxLifetime(time.Hour)
// M11H-db-1):从库启动 ping 经 pingWithTimeout 受 3s 约束,挂起 DB 不无限阻塞 InitDBWithReplicas。
if err := pingWithTimeout(sqlDB, ctx); err != nil {
logger.Warnf("数据库从库 %d Ping 失败: %v", i+1, err)
warnCloseDB(replicaDB, "关闭 Ping 失败的数据库从库连接池失败") // C11c: 关闭刚打开的池避免泄漏
continue
}
newReplicas = append(newReplicas, replicaDB)
logger.Info("数据库从库连接成功", zap.Int("index", i+1))
}
m.mu.Lock()
m.replicas = newReplicas
m.ensureReplicaHealthLocked()
m.mu.Unlock()
}
return nil
}
// InitDB 初始化数据库连接(带重试机制),驱动由配置决定。
func InitDB(ctx context.Context, cfg *config.Config) error {
return DefaultManager.Load().InitDB(ctx, cfg)
}
// InitDBWithReplicas 初始化数据库主从连接,驱动由配置决定
func InitDBWithReplicas(ctx context.Context, cfg *config.Config, replicaDSNs []string) error {
return DefaultManager.Load().InitDBWithReplicas(ctx, cfg, replicaDSNs)
}
// GetReadDB 获取读库实例(按策略选择从库)
func GetReadDB() *gorm.DB {
return DefaultManager.Load().Replica()
}
// GetWriteDB 获取写库实例(主库)
func GetWriteDB() *gorm.DB {
return DefaultManager.Load().Master()
}
// GetDB 获取数据库实例(默认主库,兼容旧代码)
func GetDB() *gorm.DB {
return DefaultManager.Load().Master()
}
// GetReplicas 获取所有从库实例
func GetReplicas() []*gorm.DB {
return DefaultManager.Load().Replicas()
}
// SetReplicaPicker 设置默认管理器的从库选择策略
func SetReplicaPicker(p ReplicaPicker) {
DefaultManager.Load().SetPicker(p)
}
// UseMaster 强制使用主库(用于事务或需要实时数据的场景)
func UseMaster(ctx context.Context) context.Context {
ctx = normalizeContext(ctx)
return context.WithValue(ctx, dbModeContextKey{}, dbModeMaster)
}
// UseReplica 强制使用从库(用于报表查询等场景)
func UseReplica(ctx context.Context) context.Context {
ctx = normalizeContext(ctx)
return context.WithValue(ctx, dbModeContextKey{}, dbModeReplica)
}
// GetDBFromContext 根据上下文选择数据库
func GetDBFromContext(ctx context.Context) *gorm.DB {
return DefaultManager.Load().FromContext(ctx)
}
// WithTx 将外层事务注入 ctx,使上层(如 repository.BaseRepo)在调用时能 join 到该事务
// 而非另开连接/路由到主从库(H6c)。
//
// 用法:
//
// err := database.TransactionWithContext(ctx, func(tx *gorm.DB) error {
// ctx2 := database.WithTx(ctx, tx)
// // 传 ctx2 给 repo 方法,repo 内部会优先使用该 tx
// return repo.FindByID(ctx2, id) // 此查询参与外层事务
// })
//
// 注意:tx 仅在该 ctx 的生命周期内有效;事务提交/回滚后不得再用该 ctx 携带的 tx。
func WithTx(ctx context.Context, tx *gorm.DB) context.Context {
ctx = normalizeContext(ctx)
if tx == nil {
return ctx
}
return context.WithValue(ctx, txContextKey{}, tx)
}
// TxFromContext 取出 ctx 携带的外层事务;无则返回 nil。
func TxFromContext(ctx context.Context) *gorm.DB {
ctx = normalizeContext(ctx)
if tx, ok := ctx.Value(txContextKey{}).(*gorm.DB); ok {
return tx
}
return nil
}
// AutoMigrate 自动迁移数据库表结构(由应用通过 WithMigrator/WithModels 注册)
func AutoMigrate() error {
logger.Info("数据库表结构迁移完成")
return nil
}
// Close 关闭所有数据库连接(主库与从库),等价于 CloseAll。
// 历史上仅关闭主库、遗留从库池泄漏(C11f),已修正为委托 CloseAll。
func Close() error {
return CloseAll()
}
// CloseAll 关闭所有数据库连接(包括从库)
func CloseAll() error {
return DefaultManager.Load().Close()
}
// Transaction 事务操作(自动使用主库)
func Transaction(fn func(tx *gorm.DB) error) error {
db := DefaultManager.Load().Master()
if db == nil {
return errors.New("数据库未初始化")
}
return db.Transaction(fn)
}
// TransactionWithContext 带上下文的事务操作
func TransactionWithContext(ctx context.Context, fn func(tx *gorm.DB) error) error {
ctx = normalizeContext(ctx)
db := DefaultManager.Load().Master()
if db == nil {
return errors.New("数据库未初始化")
}
return db.WithContext(ctx).Transaction(fn)
}
// ReadQuery 读查询。遵循 ctx 中的数据库路由标记;未指定时默认走从库。
func ReadQuery(ctx context.Context, model any, query string, args ...any) error {
ctx = normalizeContext(ctx)
db := GetDBFromContext(ctx)
if db == nil {
return errors.New("数据库未初始化")
}
return db.WithContext(ctx).Where(query, args...).Find(model).Error
}
// WriteQuery 在主库上执行查询并扫描到 model(强制主库,绕过从库延迟)。
// 注意:命名沿用历史,实际用 .Find() 扫描结果集(读取语义),并非写操作——
// 强制主库是为了读到刚写入的最新数据(read-your-writes)。命名误导见 M11。
func WriteQuery(ctx context.Context, model any, query string, args ...any) error {
ctx = normalizeContext(ctx)
db := DefaultManager.Load().Master()
if db == nil {
return errors.New("数据库未初始化")
}
return db.WithContext(ctx).Where(query, args...).Find(model).Error
}
// healthCheckTimeout 健康检查单次 ping 超时,避免探针被慢/挂起的 DB 长期阻塞(M11)。
const healthCheckTimeout = 3 * time.Second
// pingWithTimeout 带超时的 ping,ctx 由调用方传入时优先尊重其 deadline。
func pingWithTimeout(sqlDB *sql.DB, parent context.Context) error {
parent = normalizeContext(parent)
ctx, cancel := context.WithTimeout(parent, healthCheckTimeout)
defer cancel()
return sqlDB.PingContext(ctx)
}
// HealthCheck 健康检查(主库 + 从库),单次 ping 限 3s 超时(M11)。
func HealthCheck() map[string]bool {
result := make(map[string]bool)
ctx := context.Background()
m := DefaultManager.Load()
// 检查主库
if master := m.Master(); master != nil {
sqlDB, err := master.DB()
if err == nil && pingWithTimeout(sqlDB, ctx) == nil {
result["master"] = true
} else {
result["master"] = false
}
} else {
result["master"] = false
}
// 检查从库
for i, replica := range m.Replicas() {
if replica != nil {
sqlDB, err := replica.DB()
if err == nil && pingWithTimeout(sqlDB, ctx) == nil {
result[fmt.Sprintf("replica_%d", i+1)] = true
} else {
result[fmt.Sprintf("replica_%d", i+1)] = false
}
} else {
result[fmt.Sprintf("replica_%d", i+1)] = false
}
}
return result
}
// IsDBHealthy 返回主库探活健康状态(#21)。
// 与 HealthCheck()(实时 ping)不同,这是后台探活维护的缓存标记,
// 供 readiness 探针快速判断是否接流量,避免每次探针都同步 ping。
func IsDBHealthy() bool {
return DefaultManager.Load().IsHealthy()
}
// StartDBProbing 启动主库/从库探活后台循环(#21)。
// 阻塞,应通过 App.Go 在独立 goroutine 运行;ctx 取消时退出。
func StartDBProbing(ctx context.Context) {
DefaultManager.Load().StartProbing(ctx)
}