Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5dbbce34f5 | |||
| 96ce883d41 |
@@ -16,6 +16,51 @@ xlgo 框架更新日志。本文档遵循 [Keep a Changelog](https://keepachange
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [1.4.0] - 2026-07-12
|
||||
|
||||
> **instance+facade 一致性收尾**:cron/storage/cache/ratelimit 四个"纯全局"包按 `database.Manager`/`jwt.Manager` 模式实例化(App 持实例 + `SwapDefault*` 提升全局默认 + 包级 facade 代理),消除单进程多 App 互相污染;trace 纳入 App 生命周期(不做实例隔离,OTel 本身进程全局);清理 `WithoutWire` 空函数。
|
||||
>
|
||||
> **修复的多 App 冲突**:① App A `Shutdown` 不再把全局 cron 调度器 / 限流器停掉(B 的任务/限流不受影响);② App B `Init` 不再覆盖 App A 的 storage driver / cache redis;③ cache/ratelimit 不再硬编码 `database.GetRedis()`,可注入 per-App redis;④ `commitReplacedResources` 不再关闭"被替换的全局默认"DB/Redis manager(多 App 下那可能是另一个 App 的活跃 manager,关闭致 "redis: client is closed");⑤ trace.Close 在 `App.Shutdown` 被调用(此前从未调用,exporter 后台 goroutine/连接泄漏)。
|
||||
>
|
||||
> `go vet` + `go build` + `go test -race ./...` 全绿。本机需 `-buildvcs=false`。
|
||||
|
||||
### Breaking ⚠️
|
||||
|
||||
- **`xlgo.WithoutWire()` 删除**(app.go):自 v1.1.0 wire 包删除后遗留的空 no-op 函数,全仓无引用。直接删除,无替代;调用方删除该 Option 即可(无行为变化)。
|
||||
- **`cron.AddTask(...)` 在 `App.Init` 之前注册不再生效**(cron/cron.go + app.go):App 现持专属 `*Scheduler`,`Init` 时创建并 `SwapDefaultScheduler` 提升为全局默认。pre-Init 调用包级 `cron.AddTask` 会注册到 init 默认调度器、被 App swap 丢弃。迁移:改用 `xlgo.WithCronTask(name, schedule, handler)`(蕴含启用 cron),或 `Init` 后 `app.Scheduler().AddTask(...)`。post-Init 的包级 `cron.AddTask` 仍可用(代理到 App 调度器)。`cron` 包 `init()` 现预创建全局默认调度器(曾为懒初始化)。
|
||||
- **`middleware.StopRateLimiters()` 语义收窄**(middleware/ratelimit.go):仅停止全局默认 `RateLimitRegistry` 的限流器。App 现持专属 Registry,`Shutdown` 调 `app` 自己的 `Stop`,不再调全局 `StopRateLimiters`(避免误停其他 App)。`LoginRateLimit`/`APIRateLimit`/`UploadRateLimit`/`CustomRateLimit` 改为请求时从默认 Registry 懒创建限流器(单 App 行为不变;多 App 下各自 Registry 隔离)。
|
||||
- **`App.commitReplacedResources` 不再关闭被替换的全局 DB/Redis manager**(app.go):多 App 下"previous"可能是另一个 App 的活跃 manager,关闭致 `redis: client is closed`。单 App 下 previous 为 init() 空默认(无 client,Close 本就是 no-op),无影响。用户若手动 `database.InitRedis`/`InitDB` 后再 `App.Init`,旧 client 的关闭由用户负责(非常规组合)。
|
||||
- **`cache.redisCache` 持注入 redis client**(cache/cache.go):新增 `cache.NewRedisCacheWithRedis(client)`、`(*CacheManager).InitWithRedis(client)`、`cache.SwapDefaultCacheManager(m)`。`cache.NewRedisCache()` 仍存在(懒取全局 Redis,standalone 用法)。App 现经 `InitWithRedis(a.redisManager.Client())` 注入 per-App client。
|
||||
- **`middleware.RedisRateLimiter` 新增 `WithRedisClient(client)` 选项**(middleware/ratelimit.go):`Allow`/`GetCount`/`Reset` 改用 `rl.redisClient()`(注入优先、`database.GetRedis()` 兜底,照 `jwt.TokenBlacklist` 模型)。包级 `RedisRateLimit` 系列仍用全局 Redis;per-App 隔离用 `middleware.NewRedisRateLimiter(..., middleware.WithRedisClient(app.RedisClient()))`。
|
||||
|
||||
### Added
|
||||
|
||||
- **`xlgo.WithTrace()`**:把 OpenTelemetry trace 纳入 App 生命周期--`Init` 调 `trace.Init`(按 `config.Trace`)、装入 `trace.Middleware`、`Shutdown` 调 `trace.Close`。实际导出由 `config.Trace.Enabled` 控制。**不做实例隔离**:OTel `TracerProvider`/`Propagator` 是进程级全局单例,多 App 共享(与 OTel 设计一致)。
|
||||
- **`xlgo.WithCronTask(name, schedule, handler)`**:注册定时任务到 App 专属调度器(蕴含 `WithCron`),替代旧的全局 `cron.AddTask` pre-Run 注册模式。
|
||||
- **`xlgo.App.Scheduler()` / `Cache()` / `RedisClient()`**:分别返回 App 专属的 cron 调度器、缓存服务、Redis 客户端,供多 App 隔离场景直接操作(绕过全局 facade)。
|
||||
- **`config.TraceConfig` + `Config.Trace` 字段**:链路追踪 YAML 配置(`service_name`/`exporter_type`/`endpoint`/`insecure`/`sample_ratio`/`enabled`/`propagator` 等)。`Validate` 在 `Enabled` 时校验 `sample_ratio` 范围;导出器/传播器枚举校验委托 `trace.Init`。
|
||||
- **`cron.SwapDefaultScheduler(s)`**、**`storage.SwapDefaultStorageManager(m)`**、**`cache.SwapDefaultCacheManager(m)`**、**`middleware.SwapDefaultRateLimitRegistry(r)`**:返回被替换的旧实例,供 App Init 失败回滚(照 `database.SwapDefaultManager`/`SwapDefaultRedisManager` 模式)。
|
||||
- **`storage.StorageManager.Close()`**:闭合 Init/Close 生命周期(与 database/redis/logger manager 对称),供 `App.closeResources`(Shutdown 路径)与 `rollbackReplacedResources`(Init 失败路径)对称收口。当前 LocalStorage/OSSStorage 无 closeable 资源(`oss.Client` 未暴露 Close),故为 no-op;经 `io.Closer` 断言调用驱动,未来带连接池的驱动实现 `io.Closer` 即被自动调用,框架骨架无需改动。
|
||||
- **`middleware.RateLimitRegistry`**:持 login/api/upload/custom 内存限流器,支持 per-App 隔离。`NewRateLimitRegistry()` / `Init()` / `Stop()` / `GetDefaultRateLimitRegistry()`。
|
||||
- **`xlgo.App.RateLimitRegistry()` + App-bound 限流中间件**(middleware/ratelimit.go):`app.RateLimitRegistry().LoginRateLimit()`/`APIRateLimit()`/`UploadRateLimit()`/`CustomRateLimit(rate, window)` 返回的中间件捕获 App 自己的 Registry(请求时不查全局默认),实现多 App per-App 计数隔离。`RateLimitRegistry` 改在 `New()` 时预创建,使 getter 在 `Init` 前即可用于路由装配;`Init` 失败回滚时 `registry.Stop()` 收口 custom limiter,无泄漏。
|
||||
|
||||
### Fixed
|
||||
|
||||
- **多 App 限流器互不污染**(middleware/ratelimit.go):App A `Shutdown` 不再经全局 `StopRateLimiters` 把 App B 的 loginLimiter 置 nil(致 B 下次请求懒创建新 limiter、计数清零,稳态客户端借 A 的 Shutdown 窗口绕过限流)。各 App 持自己的 `RateLimitRegistry`。
|
||||
- **多 App cron 调度器隔离**(cron/cron.go + app.go):App A `Shutdown` 不再经 `cron.StopGlobalWithTimeout` 停掉共享全局调度器(B 的任务不受影响)。
|
||||
- **多 App storage / cache 隔离**(storage/cache + app.go):App B `Init` 不再覆盖 App A 的 storage driver / cache redis;cache 不再硬编码 `database.GetRedis()`(致 A 的缓存读到 B 的 redis)。
|
||||
- **trace exporter 泄漏**(app.go):`App.Shutdown` 现调 `trace.Close` 刷出批量 span 并释放 exporter 后台 goroutine/连接(H-14 修了 `Close` 幂等,但 App 此前从未调它)。
|
||||
- **多 App logger writer 跨 App 误杀**(app.go `commitReplacedResources`):与 `previousDB`/`previousRedis` 对齐,`commitReplacedResources` 不再 `CloseDefaultSnapshot`(旧实现会关掉被替换的全局默认 logger 的 lumberjack writers--多 App 下那可能是另一个 App 的活跃 writer,致其后续写日志失败/丢失)。单 App 下旧默认为 Nop(无 writer,不泄漏);用户手动配置 logger 后再 `App.Init` 的旧 writer 由用户负责。
|
||||
|
||||
### 多 App 边界与已知缺口
|
||||
|
||||
本次实例化收尾覆盖 cron/storage/cache/ratelimit + trace 生命周期。多 App 进程下的边界:
|
||||
|
||||
- **包级限流 facade 仅绑定最后 Init 的 App**(middleware/ratelimit.go):`LoginRateLimit`/`APIRateLimit`/`UploadRateLimit`/`CustomRateLimit`(包级)在请求时解析 `GetDefaultRateLimitRegistry()`(全局默认),多 App 下仅最后 Init 的 App 是全局默认。**已提供 per-App 计数隔离路径**:`app.RateLimitRegistry().LoginRateLimit()` 等 App-bound 中间件捕获 App 自己的 Registry(见 Added)。包级 facade 仍为单 App / backcompat 保留。
|
||||
- **`CustomRateLimit` 首请求时机**:包级 `CustomRateLimit` 的 `sync.Once` 在首请求时登记 limiter 到全局默认 Registry,首请求若在 `App.Init` swap 前会泄漏 cleanup goroutine。**App-bound 路径 `app.RateLimitRegistry().CustomRateLimit(rate, window)` 已无此问题**(登记到 App 的 Registry,由 `registry.Stop()` 收口)。HTTP server 在 Init 后才起,常规用法不触发泄漏。
|
||||
- **trace 进程全局共享**(trace/trace.go,设计限制):OTel `TracerProvider`/`Propagator` 是进程级单例,任一 App `Shutdown` 调 `trace.Close` 会关闭全局 trace 导出,影响同进程其他 App。多 App 需自行管理 trace 生命周期(不在多个 App 上同时启用 `WithTrace`)。**不可修**(OTel 自身全局设计)。
|
||||
- **`StorageManager` 无 `Close`**(storage/storage.go,经核实:oss.Client 无 Close,当前为 no-op;已加 StorageManager.Close() 闭合生命周期(见 Added)):`LocalStorage` 无 closeable 资源;`OSSStorage` 持 `*oss.Client`,但 `oss.Client` **未暴露 `Close` 方法**(aliyun-oss-go-sdk),其 HTTP 空闲连接由 OS 超时 + GC 回收。无显式资源泄漏;未来带连接池的驱动实现 io.Closer 即被 Close 自动调用。
|
||||
|
||||
## [1.3.0] - 2026-07-12
|
||||
|
||||
> 同义可失败 API 收敛 + 注释/文档一致性修复 + 前序评审收口。
|
||||
|
||||
@@ -513,6 +513,9 @@ type User struct {
|
||||
|
||||
### 5.1 初始化缓存
|
||||
|
||||
> **v1.4.0**:App 用户经 `WithRedis()` 启用时,`Init` 自动用 `app.redisManager.Client()` 初始化缓存
|
||||
> (走 App 自己的 Redis,多 App 隔离,不串越)。下方 `cache.Init()`(包级)用于 standalone 或回退全局 Redis。
|
||||
|
||||
```go
|
||||
import "github.com/EthanCodeCraft/xlgo-core/cache"
|
||||
|
||||
@@ -908,6 +911,12 @@ token := middleware.GetCSRFToken(c)
|
||||
|
||||
### 8.4 限流(支持 Redis 分布式限流)
|
||||
|
||||
> **v1.4.0**:App 持专属 `RateLimitRegistry`,`Shutdown` 调 `registry.Stop()`(不再调全局
|
||||
> `StopRateLimiters`,避免误停其他 App 的限流器)。下方包级 `LoginRateLimit`/`APIRateLimit`/等仍可用
|
||||
> (请求时解析全局默认 Registry,单 App 场景无变化)。**多 App per-App 计数隔离**用
|
||||
> `app.RateLimitRegistry().LoginRateLimit()` 等 App-bound 中间件(捕获 App 自己的 Registry,不查全局)。
|
||||
> **多 App 走各自 Redis** 用 `middleware.NewRedisRateLimiter(..., middleware.WithRedisClient(app.RedisClient()))`。
|
||||
|
||||
```go
|
||||
// 初始化限流器
|
||||
middleware.InitRateLimiters()
|
||||
@@ -1213,6 +1222,10 @@ storage:
|
||||
|
||||
### 10.3 使用存储
|
||||
|
||||
> **v1.4.0**:App 用户经 `WithStorage()` 启用时,`Init` 自动初始化存储并提升为全局默认;`Shutdown`
|
||||
> 经 `StorageManager.Close()` 收口(当前 LocalStorage/OSSStorage 无 closeable 资源为 no-op,为未来
|
||||
> 驱动预留)。下方 `storage.Init()`(包级)用于 standalone。
|
||||
|
||||
```go
|
||||
import "github.com/EthanCodeCraft/xlgo-core/storage"
|
||||
|
||||
@@ -1536,6 +1549,24 @@ count := hub.Count()
|
||||
|
||||
## 13. 定时任务
|
||||
|
||||
> **v1.4.0**:App 持专属调度器(实例隔离,多 App 互不污染)。**App 管理的任务用
|
||||
> `xlgo.WithCronTask(name, schedule, handler)` 注册**(蕴含启用 cron,`Init` 时注册+启动,
|
||||
> `Shutdown` 时停止并等待在跑任务退出)。下方包级 `cron.AddTask`/`cron.Start`/`cron.Stop` 仍可用于
|
||||
> standalone(无 App)或 `Init` 之后;**`Init` 之前**调包级 `cron.AddTask` 会注册到 init 默认
|
||||
> 调度器、被 App swap 丢弃。`app.Scheduler()` 返回 App 调度器,供动态注册与管理
|
||||
> (`AddTask`/`RemoveTask`/`RunTask`/`ListTasks`)。
|
||||
>
|
||||
> ```go
|
||||
> app := xlgo.New(xlgo.WithConfig(cfg),
|
||||
> xlgo.WithCronTask("cleanup", cron.Every(5*time.Minute), func(ctx context.Context) error {
|
||||
> return cleanupOldData()
|
||||
> }),
|
||||
> xlgo.WithCronTask("daily_report", cron.Daily(2, 0), generateReport),
|
||||
> )
|
||||
> // 动态注册(Init 后):app.Scheduler().AddTask(...)
|
||||
> if err := app.Run(); err != nil { /* ... */ }
|
||||
> ```
|
||||
|
||||
### 13.1 添加任务
|
||||
|
||||
```go
|
||||
@@ -1586,6 +1617,17 @@ scheduler.EnableTask("cleanup") // 启用任务
|
||||
|
||||
## 14. 链路追踪
|
||||
|
||||
> **v1.4.0**:`xlgo.WithTrace()` 把 trace 纳入 App 生命周期--`Init` 按 `config.Trace` 调
|
||||
> `trace.Init`、装入 `trace.Middleware`、`Shutdown` 调 `trace.Close` 刷出 span 并释放 exporter
|
||||
> 后台 goroutine。下方手动 `trace.Init`/`trace.Close`/`trace.Middleware` 仍可用于 standalone
|
||||
> (无 App)。**trace 不做实例隔离**--OTel `TracerProvider` 是进程级全局单例,多 App 进程共享,
|
||||
> 任一 `Shutdown` 会关全局导出;多 App 不要同时开 `WithTrace`。
|
||||
>
|
||||
> ```go
|
||||
> app := xlgo.New(xlgo.WithConfigPath("config.yaml"), xlgo.WithTrace())
|
||||
> // config.yaml 配置 trace.enabled / trace.endpoint / trace.exporter_type / trace.sample_ratio ...
|
||||
> ```
|
||||
|
||||
### 14.1 初始化
|
||||
|
||||
```go
|
||||
|
||||
@@ -339,6 +339,8 @@ err := userRepo.WithTransaction(ctx, func(txRepo *repository.BaseRepo[model.User
|
||||
|
||||
### Redis 缓存
|
||||
|
||||
> **v1.4.0**:App 经 `WithRedis()` 启用时自动用 App 的 Redis 客户端初始化缓存(多 App 隔离,不串越)。下方 `cache.Init()` 用于 standalone。
|
||||
|
||||
```go
|
||||
// 初始化缓存
|
||||
cache.Init()
|
||||
@@ -416,6 +418,8 @@ err := cache.WithLockAutoExtend(ctx, key, 30*time.Second, 10*time.Second, func(c
|
||||
|
||||
### Redis 分布式限流
|
||||
|
||||
> **v1.4.0**:多 App 走各自 Redis 用 `middleware.NewRedisRateLimiter(..., middleware.WithRedisClient(app.RedisClient()))`。
|
||||
|
||||
```go
|
||||
// 内存限流(单实例)
|
||||
r.Use(middleware.CustomRateLimit(100, time.Minute))
|
||||
@@ -528,6 +532,11 @@ hub.Broadcast([]byte("广播消息"))
|
||||
|
||||
### 定时任务
|
||||
|
||||
> **v1.4.0**:App 持专属调度器(实例隔离)。**App 管理的任务用 `xlgo.WithCronTask(name, schedule, handler)`
|
||||
> 注册**(蕴含启用 cron,`Init` 时注册+启动,`Shutdown` 时停止)。下方包级 `cron.AddTask`/`cron.Start`
|
||||
> 仅用于 standalone(无 App)或 `Init` 之后;**`Init` 之前**调包级 `cron.AddTask` 会注册到 init 默认
|
||||
> 调度器、被 App swap 丢弃。`app.Scheduler()` 供动态注册与管理。
|
||||
|
||||
```go
|
||||
// 每隔 5 分钟执行
|
||||
cron.AddTask("cleanup", cron.Every(5*time.Minute), func(ctx context.Context) error {
|
||||
@@ -620,6 +629,8 @@ userType := middleware.GetUserType(c)
|
||||
|
||||
### 限流中间件
|
||||
|
||||
> **v1.4.0**:App `Shutdown` 自动 `registry.Stop()`(不再调全局 `StopRateLimiters`,避免误停其他 App)。多 App per-App 计数隔离用 `app.RateLimitRegistry().LoginRateLimit()` 等 App-bound 中间件。
|
||||
|
||||
```go
|
||||
// 登录限流(每分钟 10 次)
|
||||
r.POST("/login", middleware.LoginRateLimit(), func(c *gin.Context) {
|
||||
@@ -842,6 +853,22 @@ docker run -d -p 8080:8080 xlgo-app:latest
|
||||
|
||||
> 完整变更历史见 [CHANGELOG.md](./CHANGELOG.md)。
|
||||
|
||||
### v1.4.0 (2026-07-12)
|
||||
|
||||
> **instance+facade 一致性收尾**:cron/storage/cache/ratelimit 实例化(App 持 per-App 实例 + `SwapDefault*` + 包级 facade 代理,照 `database.Manager`/`jwt.Manager` 模式),消除单进程多 App 互相污染;trace 纳入 App 生命周期(`WithTrace`);删除 `WithoutWire` 空函数。`go vet` + `go build` + `go test -race ./...` 全绿。完整变更见 [CHANGELOG.md](./CHANGELOG.md)。
|
||||
|
||||
主要破坏性变更(升级前必读):
|
||||
|
||||
- **`xlgo.WithoutWire()` 删除**:v1.1.0 wire 包删除后遗留的空 no-op,直接删除调用即可(无行为变化)。
|
||||
- **`cron.AddTask` 在 `App.Init` 之前注册失效**:App 持专属调度器,pre-Init 包级 `cron.AddTask` 注册到 init 默认调度器、被 App swap 丢弃。改用 `xlgo.WithCronTask(name, schedule, handler)`(或 Init 后 `app.Scheduler().AddTask`)。
|
||||
- **`middleware.StopRateLimiters` 语义收窄**:仅停默认 Registry;App `Shutdown` 调 `app` 自己的 `registry.Stop()`,不再误停其他 App 限流器。
|
||||
- **`commitReplacedResources` 不再关闭 `previousDB`/`previousRedis`/`loggerSnapshot`**:多 App 下 previous 可能是另一 App 的活跃资源(致 `redis: client is closed` / logger writer 误杀)。单 App 无影响(init 默认无资源)。
|
||||
- **cache/ratelimit redis 注入**(jwt 模型):`cache.NewRedisCacheWithRedis`/`CacheManager.InitWithRedis`;`middleware.WithRedisClient`。
|
||||
|
||||
新增:`WithTrace`/`WithCronTask`;`app.Scheduler()`/`Cache()`/`RedisClient()`/`RateLimitRegistry()`;`cron/storage/cache/middleware` 各 `SwapDefault*`;`config.TraceConfig`;`middleware.RateLimitRegistry` + App-bound 限流中间件;`storage.StorageManager.Close()`。
|
||||
|
||||
升级:`go get github.com/EthanCodeCraft/xlgo-core@v1.4.0`(⚠️ 含破坏性变更,详见 CHANGELOG 迁移说明)
|
||||
|
||||
### v1.2.0 (2026-07-04)
|
||||
|
||||
> **破坏性版本**:4 轮对抗性评审(deepseek / GLM / Claude / 终审)收口的全部 CRITICAL/HIGH/MEDIUM 修复 + 主线A 包级可变全局并发治理统一。`go vet` + `go build` + `go test -race ./...` 全绿。完整变更见 [CHANGELOG.md](./CHANGELOG.md)。
|
||||
|
||||
@@ -23,14 +23,16 @@ import (
|
||||
"github.com/EthanCodeCraft/xlgo-core/response"
|
||||
"github.com/EthanCodeCraft/xlgo-core/router"
|
||||
"github.com/EthanCodeCraft/xlgo-core/storage"
|
||||
"github.com/EthanCodeCraft/xlgo-core/trace"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// Version 框架版本号。发版时只改这一处,避免版本字面量散落各处。
|
||||
// CLI(xlgo version)、脚手架生成的 go.mod 等均引用此常量。
|
||||
const Version = "1.3.0"
|
||||
const Version = "1.4.0"
|
||||
|
||||
// HealthCheckFunc 健康检查函数
|
||||
type HealthCheckFunc func(context.Context) error
|
||||
@@ -59,6 +61,14 @@ type staticRoute struct {
|
||||
root string
|
||||
}
|
||||
|
||||
// cronTask 收集 WithCronTask 注册的任务,doInit 创建 App 调度器后批量注册(照
|
||||
// WithMigrator/WithHealthCheck 的"Option 收集、doInit 应用"模式)。
|
||||
type cronTask struct {
|
||||
name string
|
||||
schedule cron.Schedule
|
||||
handler cron.TaskHandler
|
||||
}
|
||||
|
||||
// appState 是 App 生命周期状态机(M1)。状态受 lifecycleMu 保护,单调推进:
|
||||
//
|
||||
// stateCreated → stateInitializing → stateInitialized → stateStopping → stateStopped
|
||||
@@ -99,6 +109,7 @@ type App struct {
|
||||
enableReadiness bool
|
||||
enableMetrics bool
|
||||
enableCron bool
|
||||
enableTrace bool
|
||||
metricsPath string
|
||||
|
||||
staticRoutes []staticRoute
|
||||
@@ -123,6 +134,9 @@ type App struct {
|
||||
initializedMySQL bool
|
||||
initializedRedis bool
|
||||
initializedCron bool
|
||||
initializedTrace bool
|
||||
initializedStorage bool
|
||||
initializedCache bool
|
||||
|
||||
loggerManager *logger.LogManager
|
||||
loggerSnapshot *logger.DefaultSnapshot
|
||||
@@ -130,6 +144,15 @@ type App struct {
|
||||
previousDB *database.Manager
|
||||
redisManager *database.RedisManager
|
||||
previousRedis *database.RedisManager
|
||||
storageManager *storage.StorageManager
|
||||
previousStorage *storage.StorageManager
|
||||
scheduler *cron.Scheduler
|
||||
previousScheduler *cron.Scheduler
|
||||
cronTasks []cronTask
|
||||
cacheManager *cache.CacheManager
|
||||
previousCache *cache.CacheManager
|
||||
rateLimitRegistry *middleware.RateLimitRegistry
|
||||
previousRateLimitRegistry *middleware.RateLimitRegistry
|
||||
|
||||
// 请求级超时(#19),<=0 表示不启用
|
||||
requestTimeout time.Duration
|
||||
@@ -254,12 +277,6 @@ func WithoutStorage() Option {
|
||||
return func(a *App) { a.enableStorage = false }
|
||||
}
|
||||
|
||||
// WithoutWire 已移除(wire 包在 v1.1.0 删除)。保留空函数仅为编译兼容,
|
||||
// 调用无副作用。后续版本将删除。
|
||||
func WithoutWire() Option {
|
||||
return func(a *App) {}
|
||||
}
|
||||
|
||||
// WithAutoMigrate 启用数据库迁移(需配合 WithMigrator/WithModels 注册迁移逻辑)
|
||||
func WithAutoMigrate() Option {
|
||||
return func(a *App) { a.enableAutoMigrate = true }
|
||||
@@ -331,14 +348,45 @@ func WithRequestTimeout(d time.Duration) Option {
|
||||
return func(a *App) { a.requestTimeout = d }
|
||||
}
|
||||
|
||||
// WithCron 将全局 cron 调度器纳入 App 生命周期(P1 #10)。
|
||||
// 启用后:Init 末尾 cron.Start() 启动调度循环;Shutdown 时在 ShutdownTimeout 约束内
|
||||
// 停止全局调度器并等待在跑任务退出(cron.StopGlobalWithTimeout),避免调度 goroutine
|
||||
// 与在跑任务在优雅关闭后残留。任务仍通过 cron.AddTask(...) 在 Run 前注册。
|
||||
// WithCron 将 cron 调度器纳入 App 生命周期(Phase 3 实例化)。
|
||||
// 启用后:Init 时创建 App 专属 Scheduler、注册 WithCronTask 收集的任务、提升为全局默认、
|
||||
// Start 启动调度循环;Shutdown 在剩余预算内 StopWithTimeout 等待在跑任务退出。
|
||||
//
|
||||
// 任务注册:优先用 WithCronTask(...)(或 Init 后 app.Scheduler().AddTask(...))注册到 App 实例。
|
||||
// 包级 cron.AddTask(...) 仍可用--它代理到当前全局默认(App swap 后即 App 的调度器)--但
|
||||
// 在 App.Init 之前调用会注册到 init 默认调度器、被 App swap 丢弃,故 pre-Init 注册请用 WithCronTask。
|
||||
func WithCron() Option {
|
||||
return func(a *App) { a.enableCron = true }
|
||||
}
|
||||
|
||||
// WithCronTask 注册一个定时任务到 App 的调度器(Phase 3)。蕴含启用 cron(照 WithMigrator
|
||||
// 自动启用 AutoMigrate 的模式),故无需同时 WithCron。任务在 doInit 创建调度器后注册。
|
||||
//
|
||||
// name/schedule/handler 的校验由 cron.Scheduler.AddTask 负责(nil schedule/handler panic,
|
||||
// 非法 cron 字段 panic)。动态输入请先用 cron.ParseCronStrict 处理 error。
|
||||
func WithCronTask(name string, schedule cron.Schedule, handler cron.TaskHandler) Option {
|
||||
return func(a *App) {
|
||||
a.cronTasks = append(a.cronTasks, cronTask{name: name, schedule: schedule, handler: handler})
|
||||
a.enableCron = true
|
||||
}
|
||||
}
|
||||
|
||||
// WithTrace 将链路追踪(OpenTelemetry)纳入 App 生命周期(Phase 1)。
|
||||
//
|
||||
// 启用后:Init 时按 config.Trace 调 trace.Init(安装 TracerProvider/Propagator)并把
|
||||
// trace.Middleware 装入全局中间件链;Shutdown 时调 trace.Close 刷出 exporter 并释放
|
||||
// 后台 goroutine/连接(H-14 后 Close 幂等,但 App 此前从未调它--本 Option 补上生命周期闭环)。
|
||||
//
|
||||
// 实际是否导出 span 由 config.Trace.Enabled 控制:Enabled=false(默认)时 trace.Init
|
||||
// 安装 Noop tracer,Middleware 不 panic、不导出。故"WithTrace + trace.enabled=true + endpoint"
|
||||
// 才真正出链路;仅 WithTrace 而未配 enabled 为 Noop(安全)。
|
||||
//
|
||||
// 不做 per-App 实例隔离:OTel 的 TracerProvider/Propagator 是进程级全局单例
|
||||
// (otel.SetTracerProvider 全局生效),多 App 进程共享同一 OTel 状态,与 OTel 设计一致。
|
||||
func WithTrace() Option {
|
||||
return func(a *App) { a.enableTrace = true }
|
||||
}
|
||||
|
||||
// WithModels 注册 GORM 自动迁移模型(自动启用 AutoMigrate)
|
||||
func WithModels(models ...any) Option {
|
||||
return WithMigrator(func(db *gorm.DB) error {
|
||||
@@ -396,6 +444,10 @@ func New(opts ...Option) *App {
|
||||
app.registry = router.NewRegistry(app.router)
|
||||
// rootCtx 生命周期与 App 一致,不依赖 Init,使 App.Go 在 Init 前也可用(#22)
|
||||
app.rootCtx, app.cancel = context.WithCancel(context.Background())
|
||||
// Phase 5:限流器 Registry 在 New 时预创建,使 app.RateLimitRegistry() 在 Init 前即可用于
|
||||
// 路由装配(app-bound 中间件捕获此实例,请求时走 App 自己的 Registry 而非全局默认,
|
||||
// 实现多 App per-App 计数隔离)。doInit 时 swap 为全局默认供包级 facade 代理。
|
||||
app.rateLimitRegistry = middleware.NewRateLimitRegistry()
|
||||
|
||||
for _, opt := range opts {
|
||||
opt(app)
|
||||
@@ -488,6 +540,15 @@ func (a *App) failAfterInit() {
|
||||
if err := a.stopCron(5 * time.Second); err != nil {
|
||||
a.initErr = errors.Join(a.initErr, err)
|
||||
}
|
||||
// trace 不是 swap 资源(OTel 全局),不走 rollbackReplacedResources;Init 失败时单独 Close。
|
||||
if a.initializedTrace {
|
||||
closeCtx, closeCancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
if err := trace.Close(closeCtx); err != nil {
|
||||
a.initErr = errors.Join(a.initErr, fmt.Errorf("trace 关闭失败: %w", err))
|
||||
}
|
||||
closeCancel()
|
||||
a.initializedTrace = false
|
||||
}
|
||||
if rbErr := a.rollbackReplacedResources(); rbErr != nil {
|
||||
a.initErr = errors.Join(a.initErr, fmt.Errorf("回滚已初始化资源失败: %w", rbErr))
|
||||
}
|
||||
@@ -526,6 +587,17 @@ func (a *App) closeResources() error {
|
||||
a.initializedRedis = false
|
||||
a.redisManager = nil
|
||||
}
|
||||
if a.initializedStorage {
|
||||
// StorageManager.Close 当前为 no-op(oss.Client 无 Close),但闭合 Init/Close 生命周期,
|
||||
// 为未来驱动预留收口点。与 db/redis/logger manager 对齐。
|
||||
if a.storageManager != nil {
|
||||
if err := a.storageManager.Close(); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
a.initializedStorage = false
|
||||
a.storageManager = nil
|
||||
}
|
||||
if a.initializedLogger {
|
||||
if a.loggerManager != nil {
|
||||
if err := a.loggerManager.Close(); err != nil {
|
||||
@@ -575,6 +647,48 @@ func (a *App) rollbackReplacedResources() error {
|
||||
a.previousRedis = nil
|
||||
a.initializedRedis = false
|
||||
}
|
||||
if a.initializedStorage {
|
||||
// 把全局默认 swap 回 Init 前的旧实例,并 Close 本 App 新建的 manager(与 closeResources
|
||||
// 的 Shutdown 路径对称)。当前 Close 为 no-op,但未来驱动有真实副作用时不漏关。
|
||||
if a.previousStorage != nil {
|
||||
storage.SwapDefaultStorageManager(a.previousStorage)
|
||||
}
|
||||
if a.storageManager != nil {
|
||||
if err := a.storageManager.Close(); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
a.storageManager = nil
|
||||
a.previousStorage = nil
|
||||
a.initializedStorage = false
|
||||
}
|
||||
// cron:stopCron 已在 failAfterInit 停止调度 goroutine(并把 initializedCron 置 false),
|
||||
// 此处仅把全局默认 swap 回 Init 前的旧调度器。用 a.scheduler 作守卫(stopCron 不 nil 它)。
|
||||
if a.scheduler != nil {
|
||||
if a.previousScheduler != nil {
|
||||
cron.SwapDefaultScheduler(a.previousScheduler)
|
||||
}
|
||||
a.previousScheduler = nil
|
||||
a.scheduler = nil
|
||||
}
|
||||
if a.initializedCache {
|
||||
// CacheManager 无 Close(client 由 redisManager 持有):仅 swap 回全局默认、清理引用。
|
||||
if a.previousCache != nil {
|
||||
cache.SwapDefaultCacheManager(a.previousCache)
|
||||
}
|
||||
a.cacheManager = nil
|
||||
a.previousCache = nil
|
||||
a.initializedCache = false
|
||||
}
|
||||
if a.rateLimitRegistry != nil {
|
||||
// 停止可能已创建的 limiter(Init 期间通常无请求、无 limiter;Stop 幂等)。
|
||||
a.rateLimitRegistry.Stop()
|
||||
if a.previousRateLimitRegistry != nil {
|
||||
middleware.SwapDefaultRateLimitRegistry(a.previousRateLimitRegistry)
|
||||
}
|
||||
a.previousRateLimitRegistry = nil
|
||||
a.rateLimitRegistry = nil
|
||||
}
|
||||
if a.initializedLogger {
|
||||
if err := logger.RestoreDefaultSnapshot(a.loggerSnapshot); err != nil {
|
||||
errs = append(errs, err)
|
||||
@@ -587,22 +701,40 @@ func (a *App) rollbackReplacedResources() error {
|
||||
}
|
||||
|
||||
func (a *App) commitReplacedResources() {
|
||||
// Phase 4 多 App 修复:不关闭 previousDB/previousRedis。previous 是 Init 前的全局默认,
|
||||
// 多 App 下可能属于另一个 App 的活跃 manager--关闭会误杀它(B 的 commit 关掉 A 的 redis,
|
||||
// 致 A 的 cache 持有的 client 失效 "redis: client is closed")。此 App 不 owns previous,
|
||||
// 仅清理回滚引用。单 App 下 previous 为 init() 空默认(无 client,Close 本就是 no-op),
|
||||
// 不关闭无泄漏。用户若手动 InitRedis/InitDB 后再 App.Init,旧 client 的关闭由用户负责。
|
||||
if a.previousDB != nil {
|
||||
if err := a.previousDB.Close(); err != nil {
|
||||
logger.Warnf("关闭被替换的旧数据库 manager 失败: %v", err)
|
||||
}
|
||||
a.previousDB = nil
|
||||
}
|
||||
if a.previousRedis != nil {
|
||||
if err := a.previousRedis.Close(); err != nil {
|
||||
logger.Warnf("关闭被替换的旧 Redis manager 失败: %v", err)
|
||||
}
|
||||
a.previousRedis = nil
|
||||
}
|
||||
if a.previousStorage != nil {
|
||||
// StorageManager 无 Close;Init 成功后仅清理回滚引用,旧实例交由 GC。
|
||||
a.previousStorage = nil
|
||||
}
|
||||
if a.previousScheduler != nil {
|
||||
// App 调度器已 Start 并接管全局默认;旧默认(init 或上一 App 的)无 goroutine 需停
|
||||
// (init 默认从未 Start;上一 App 的已在其 Shutdown 停止)。仅清理回滚引用。
|
||||
a.previousScheduler = nil
|
||||
}
|
||||
if a.previousCache != nil {
|
||||
// CacheManager 无 Close(client 由 redisManager 持有);仅清理回滚引用。
|
||||
a.previousCache = nil
|
||||
}
|
||||
if a.previousRateLimitRegistry != nil {
|
||||
// Registry 的 limiter 由 App.Shutdown 的 registry.Stop() 停止;此处仅清理回滚引用。
|
||||
a.previousRateLimitRegistry = nil
|
||||
}
|
||||
if a.loggerSnapshot != nil {
|
||||
if err := logger.CloseDefaultSnapshot(a.loggerSnapshot); err != nil {
|
||||
logger.Warnf("关闭被替换的旧 logger writer 失败: %v", err)
|
||||
}
|
||||
// Phase 4 多 App 修复(与 previousDB/previousRedis 对齐):不关闭 loggerSnapshot 捕获的
|
||||
// 旧 writers。多 App 下该 snapshot 可能捕获的是另一个 App 的活跃 logger writers--关闭会
|
||||
// 误杀它(B 的 commit 关掉 A 的 app.log/api.log writer,A 后续写日志失败/丢失)。此 App
|
||||
// 不 owns 旧 writers:init() 默认为 Nop(无 writer,不泄漏);另一 App 的 writers 由其自身
|
||||
// Shutdown 关闭;用户手动配置 logger 后再 App.Init 的旧 writer 由用户负责。仅清理回滚引用。
|
||||
a.loggerSnapshot = nil
|
||||
}
|
||||
}
|
||||
@@ -612,7 +744,11 @@ func (a *App) stopCron(timeout time.Duration) error {
|
||||
return nil
|
||||
}
|
||||
a.initializedCron = false
|
||||
if !cron.StopGlobalWithTimeout(timeout) {
|
||||
if a.scheduler == nil {
|
||||
return nil
|
||||
}
|
||||
// 停 App 专属调度器(Phase 3),不再调全局 cron.StopGlobalWithTimeout。
|
||||
if !a.scheduler.StopWithTimeout(timeout) {
|
||||
return fmt.Errorf("cron 调度器停止超时(%s)", timeout)
|
||||
}
|
||||
return nil
|
||||
@@ -657,6 +793,15 @@ func (a *App) doInit() error {
|
||||
gin.SetMode(gin.ReleaseMode)
|
||||
}
|
||||
|
||||
// trace 先于其它组件初始化:失败即早退,避免已建 DB/Redis 等资源因 trace 失败回滚。
|
||||
// OTel 进程全局,Init 安装全局 TracerProvider;Enabled=false 时为 Noop(安全)。
|
||||
if a.enableTrace {
|
||||
if err := trace.Init(buildTraceConfig(cfg)); err != nil {
|
||||
return fmt.Errorf("初始化 trace 失败: %w", err)
|
||||
}
|
||||
a.initializedTrace = true
|
||||
}
|
||||
|
||||
if a.enableLogger {
|
||||
lm := logger.NewLogManager()
|
||||
snapshot := logger.SnapshotDefault()
|
||||
@@ -698,14 +843,26 @@ func (a *App) doInit() error {
|
||||
}
|
||||
|
||||
if a.enableStorage {
|
||||
if err := storage.Init(&cfg.Storage); err != nil {
|
||||
sm := storage.NewStorageManager()
|
||||
if err := sm.Init(&cfg.Storage); err != nil {
|
||||
return fmt.Errorf("初始化存储失败: %w", err)
|
||||
}
|
||||
// 实例化 + 提升为全局默认(照 db/redis 的 Swap 模式)。StorageManager 无 Close
|
||||
// (LocalStorage/OSSStorage 无 closeable 资源),Shutdown 不关、仅 Init 失败时 swap 回退。
|
||||
a.previousStorage = storage.SwapDefaultStorageManager(sm)
|
||||
a.storageManager = sm
|
||||
a.initializedStorage = true
|
||||
}
|
||||
|
||||
if a.enableRedis {
|
||||
// Redis 就绪后初始化缓存(cache 依赖 Redis 客户端)
|
||||
cache.Init()
|
||||
// Redis 就绪后初始化缓存。Phase 4:注入 App 的 redisManager.Client(),使缓存走 per-App
|
||||
// Redis 而非全局 database.GetRedis()(修复多 App 跨 Redis 串越)。提升为全局默认供包级
|
||||
// facade(cache.Get 等)代理到此。CacheManager 无 Close(client 由 redisManager 持有)。
|
||||
cm := cache.NewCacheManager()
|
||||
cm.InitWithRedis(a.redisManager.Client())
|
||||
a.previousCache = cache.SwapDefaultCacheManager(cm)
|
||||
a.cacheManager = cm
|
||||
a.initializedCache = true
|
||||
}
|
||||
|
||||
if a.enableAutoMigrate && len(a.migrators) > 0 {
|
||||
@@ -726,6 +883,17 @@ func (a *App) doInit() error {
|
||||
// 全局中间件链:RequestID 必须最先装入,保证后续 Recovery/日志/响应都能拿到 request_id(#24)
|
||||
a.router.Use(middleware.RequestID())
|
||||
a.router.Use(middleware.Recover())
|
||||
// trace 中间件装在 Recover 之后:Recover 兜底捕获 trace 内 panic,span 覆盖业务 handler。
|
||||
// Enabled=false 时为 Noop tracer,Middleware 不导出;noop span 无 TraceID 则不写 X-Trace-ID。
|
||||
if a.enableTrace {
|
||||
// 用与 buildTraceConfig 一致的回退名(ServiceName 空时取 cfg.App.Name),避免中间件
|
||||
// span 属性与 provider resource 的 service.name 来源不一致。
|
||||
svcName := a.config.Trace.ServiceName
|
||||
if svcName == "" {
|
||||
svcName = a.config.App.Name
|
||||
}
|
||||
a.router.Use(trace.Middleware(svcName))
|
||||
}
|
||||
// 请求级超时(#19),配置后装入,下游走 c.Request.Context() 级联取消
|
||||
if a.requestTimeout > 0 {
|
||||
a.router.Use(middleware.Timeout(a.requestTimeout))
|
||||
@@ -781,12 +949,23 @@ func (a *App) doInit() error {
|
||||
a.Go(a.dbManager.StartProbing)
|
||||
}
|
||||
|
||||
// 启动 cron 全局调度器(P1 #10),Shutdown 时统一停止。任务须在此前经 cron.AddTask 注册。
|
||||
// 启动 App 专属 cron 调度器(Phase 3 实例化):创建实例 -> 注册 WithCronTask 收集的任务 ->
|
||||
// 提升为全局默认(包级 cron.AddTask 代理到此)-> Start。Shutdown 时 StopWithTimeout。
|
||||
if a.enableCron {
|
||||
cron.Start()
|
||||
a.scheduler = cron.NewScheduler()
|
||||
for _, ct := range a.cronTasks {
|
||||
a.scheduler.AddTask(ct.name, ct.schedule, ct.handler)
|
||||
}
|
||||
a.previousScheduler = cron.SwapDefaultScheduler(a.scheduler)
|
||||
a.scheduler.Start()
|
||||
a.initializedCron = true
|
||||
}
|
||||
|
||||
// Phase 5:把 New 时预创建的 App 专属 Registry 提升为全局默认,供包级 facade(LoginRateLimit
|
||||
// 等)代理。app-bound 中间件(app.RateLimitRegistry().LoginRateLimit())直接捕获此实例,
|
||||
// 不查全局默认,实现多 App per-App 计数隔离。不预 Init(懒创建);Shutdown 时 Stop。
|
||||
a.previousRateLimitRegistry = middleware.SwapDefaultRateLimitRegistry(a.rateLimitRegistry)
|
||||
|
||||
// OnInit hooks:组件初始化完成后触发(#12)
|
||||
for _, h := range a.hooks {
|
||||
if h.OnInit != nil {
|
||||
@@ -830,6 +1009,36 @@ func (a *App) resolveConfig() (*config.Config, error) {
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
// buildTraceConfig 把 config.TraceConfig 映射为 trace.Config,并补友好默认值:
|
||||
// - ServiceName 空时回退 cfg.App.Name(业务侧常已配 app.name,免重复填);
|
||||
// - ExporterType 空时默认 "otlp-http"(最常用),避免 trace.Init 对空值报"不支持的导出器类型"。
|
||||
//
|
||||
// 不默认 Endpoint/SampleRatio/Propagator:Endpoint 空时 OTEL 客户端走自身默认(localhost:4318);
|
||||
// SampleRatio=0 是合法值(不采样),不应被默认覆盖;Propagator 空时 trace.Init 已按 w3c 处理。
|
||||
// Enabled 完全透传--由 config.Trace.Enabled 决定是否真正导出,WithTrace 只管生命周期。
|
||||
func buildTraceConfig(cfg *config.Config) trace.Config {
|
||||
tc := cfg.Trace
|
||||
exporter := tc.ExporterType
|
||||
if exporter == "" {
|
||||
exporter = "otlp-http"
|
||||
}
|
||||
name := tc.ServiceName
|
||||
if name == "" {
|
||||
name = cfg.App.Name
|
||||
}
|
||||
return trace.Config{
|
||||
ServiceName: name,
|
||||
ServiceVersion: tc.ServiceVersion,
|
||||
Environment: tc.Environment,
|
||||
ExporterType: exporter,
|
||||
Endpoint: tc.Endpoint,
|
||||
Insecure: tc.Insecure,
|
||||
SampleRatio: tc.SampleRatio,
|
||||
Enabled: tc.Enabled,
|
||||
Propagator: tc.Propagator,
|
||||
}
|
||||
}
|
||||
|
||||
// Run 启动应用
|
||||
func (a *App) Run() error {
|
||||
if err := a.Init(); err != nil {
|
||||
@@ -1054,8 +1263,21 @@ func (a *App) doShutdown(wasInitialized bool) error {
|
||||
}
|
||||
}
|
||||
|
||||
logger.Info("停止限流器...")
|
||||
middleware.StopRateLimiters()
|
||||
// Phase 5:停 App 专属限流器 Registry(不再调全局 middleware.StopRateLimiters,避免误停其他 App)。
|
||||
if a.rateLimitRegistry != nil {
|
||||
logger.Info("停止限流器...")
|
||||
a.rateLimitRegistry.Stop()
|
||||
}
|
||||
|
||||
// trace.Close 刷出 exporter 批量 span 并释放后台 goroutine/连接(H-14 后幂等)。
|
||||
// 放在 db/redis/logger 关闭之前,避免 exporter 依赖已关闭的日志/连接。
|
||||
if a.initializedTrace {
|
||||
logger.Info("关闭 trace...")
|
||||
if err := trace.Close(ctx); err != nil {
|
||||
logger.Warnf("trace 关闭失败: %v", err)
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
|
||||
// db/redis/logger 经 closeResources 统一关闭(M1)。注意:closeResources 仅供
|
||||
// Shutdown 路径使用;Init 失败的回滚走 failAfterInit 的 rollbackReplacedResources,
|
||||
@@ -1108,3 +1330,35 @@ func (a *App) GetRouter() *gin.Engine {
|
||||
func (a *App) GetServer() *http.Server {
|
||||
return a.server
|
||||
}
|
||||
|
||||
// Scheduler 返回 App 专属的 cron 调度器(Phase 3)。Init 后非 nil,用于动态注册任务
|
||||
// (app.Scheduler().AddTask(...))。未启用 cron 或 Init 前返回 nil。
|
||||
func (a *App) Scheduler() *cron.Scheduler {
|
||||
return a.scheduler
|
||||
}
|
||||
|
||||
// Cache 返回 App 专属的缓存服务(Phase 4)。Init 后非 nil,用于直接操作 App 的缓存
|
||||
// (绕过全局 facade,多 App 隔离场景下确保走 App 自己的 Redis)。未启用 Redis 或 Init 前返回 nil。
|
||||
func (a *App) Cache() cache.CacheService {
|
||||
if a.cacheManager == nil {
|
||||
return nil
|
||||
}
|
||||
return a.cacheManager.Get()
|
||||
}
|
||||
|
||||
// RedisClient 返回 App 专属的 Redis 客户端(Phase 5),用于注入到需要 redis client 的
|
||||
// 组件(如 middleware.NewRedisRateLimiter(..., middleware.WithRedisClient(app.RedisClient()))),
|
||||
// 使其走 App 的 Redis 而非全局。未启用 Redis 或 Init 前返回 nil。
|
||||
func (a *App) RedisClient() *redis.Client {
|
||||
if a.redisManager == nil {
|
||||
return nil
|
||||
}
|
||||
return a.redisManager.Client()
|
||||
}
|
||||
|
||||
// RateLimitRegistry 返回 App 专属的限流器 Registry(Phase 5)。New 后即非 nil,用于装配
|
||||
// app-bound 限流中间件(app.RateLimitRegistry().LoginRateLimit() 等)--此类中间件捕获 App
|
||||
// 自己的 Registry,请求时不查全局默认,实现多 App per-App 计数隔离。
|
||||
func (a *App) RateLimitRegistry() *middleware.RateLimitRegistry {
|
||||
return a.rateLimitRegistry
|
||||
}
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
package xlgo_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
xlgo "github.com/EthanCodeCraft/xlgo-core"
|
||||
"github.com/EthanCodeCraft/xlgo-core/cache"
|
||||
"github.com/EthanCodeCraft/xlgo-core/config"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
)
|
||||
|
||||
func splitAddr(t *testing.T, addr string) (string, int) {
|
||||
t.Helper()
|
||||
host, portStr, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
t.Fatalf("SplitHostPort %q: %v", addr, err)
|
||||
}
|
||||
port, err := strconv.Atoi(portStr)
|
||||
if err != nil {
|
||||
t.Fatalf("Atoi port %q: %v", portStr, err)
|
||||
}
|
||||
if host == "" {
|
||||
host = "127.0.0.1"
|
||||
}
|
||||
return host, port
|
||||
}
|
||||
|
||||
// TestAppCacheMultiAppRedisIsolation 固化 Phase 4 核心契约:单进程多 App 时,每个 App 的
|
||||
// 缓存走自己的 redisManager.Client(),互不串越。修复前 cache 硬编码 database.GetRedis()
|
||||
// (全局),App B swap 后 App A 的缓存会读到 App B 的 redis。
|
||||
//
|
||||
// 用两个 miniredis 实例分别作 App A/B 的 Redis 后端,断言各自 cache 只写自己的实例、
|
||||
// 互读对方的 key 为 miss。appA.Cache()/appB.Cache() 绕过全局 facade 直取 App 实例。
|
||||
func TestAppCacheMultiAppRedisIsolation(t *testing.T) {
|
||||
mrA := miniredis.RunT(t)
|
||||
mrB := miniredis.RunT(t)
|
||||
hostA, portA := splitAddr(t, mrA.Addr())
|
||||
hostB, portB := splitAddr(t, mrB.Addr())
|
||||
|
||||
cfgA := testConfig(18120)
|
||||
cfgA.Redis = config.RedisConfig{Host: hostA, Port: portA}
|
||||
cfgB := testConfig(18121)
|
||||
cfgB.Redis = config.RedisConfig{Host: hostB, Port: portB}
|
||||
|
||||
appA := xlgo.New(xlgo.WithConfig(cfgA), xlgo.WithRedis())
|
||||
if err := appA.Init(); err != nil {
|
||||
t.Fatalf("appA Init: %v", err)
|
||||
}
|
||||
defer appA.Shutdown()
|
||||
appB := xlgo.New(xlgo.WithConfig(cfgB), xlgo.WithRedis())
|
||||
if err := appB.Init(); err != nil {
|
||||
t.Fatalf("appB Init: %v", err)
|
||||
}
|
||||
defer appB.Shutdown()
|
||||
|
||||
if appA.Cache() == nil || appB.Cache() == nil {
|
||||
t.Fatal("app.Cache() = nil after Init with WithRedis")
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
// A 写 keyA -> 仅落 miniredisA
|
||||
if err := appA.Cache().Set(ctx, "keyA", "valA", time.Minute); err != nil {
|
||||
t.Fatalf("appA cache Set: %v", err)
|
||||
}
|
||||
if !mrA.Exists("keyA") {
|
||||
t.Error("keyA not in miniredisA (appA redis)")
|
||||
}
|
||||
if mrB.Exists("keyA") {
|
||||
t.Error("keyA leaked into miniredisB (appB redis) -- cache not isolated")
|
||||
}
|
||||
|
||||
// B 写 keyB -> 仅落 miniredisB
|
||||
if err := appB.Cache().Set(ctx, "keyB", "valB", time.Minute); err != nil {
|
||||
t.Fatalf("appB cache Set: %v", err)
|
||||
}
|
||||
if !mrB.Exists("keyB") {
|
||||
t.Error("keyB not in miniredisB (appB redis)")
|
||||
}
|
||||
if mrA.Exists("keyB") {
|
||||
t.Error("keyB leaked into miniredisA (appA redis) -- cache not isolated")
|
||||
}
|
||||
|
||||
// 互读对方 key -> miss(各自走自己的 redis)
|
||||
var got string
|
||||
if ok, err := appA.Cache().Get(ctx, "keyB", &got); ok || err != nil {
|
||||
t.Errorf("appA cache should miss keyB (different redis): ok=%v err=%v", ok, err)
|
||||
}
|
||||
if ok, err := appB.Cache().Get(ctx, "keyA", &got); ok || err != nil {
|
||||
t.Errorf("appB cache should miss keyA (different redis): ok=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
// 各自读自己的 key -> hit
|
||||
if ok, err := appA.Cache().Get(ctx, "keyA", &got); err != nil || !ok || got != "valA" {
|
||||
t.Errorf("appA cache Get keyA = %v %q err=%v (want true valA)", ok, got, err)
|
||||
}
|
||||
if ok, err := appB.Cache().Get(ctx, "keyB", &got); err != nil || !ok || got != "valB" {
|
||||
t.Errorf("appB cache Get keyB = %v %q err=%v (want true valB)", ok, got, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppCacheRollbackOnInitFailure 固化 Phase 4 回滚:Init 失败后全局默认 CacheManager
|
||||
// 必须恢复为 Init 前的实例。
|
||||
func TestAppCacheRollbackOnInitFailure(t *testing.T) {
|
||||
preInit := cache.NewCacheManager()
|
||||
saved := cache.SwapDefaultCacheManager(preInit) // saved = init 默认;preInit 现为全局
|
||||
defer cache.SwapDefaultCacheManager(saved)
|
||||
|
||||
mr := miniredis.RunT(t)
|
||||
host, port := splitAddr(t, mr.Addr())
|
||||
cfg := testConfig(18122)
|
||||
cfg.Redis = config.RedisConfig{Host: host, Port: port}
|
||||
|
||||
app := xlgo.New(
|
||||
xlgo.WithConfig(cfg),
|
||||
xlgo.WithRedis(),
|
||||
xlgo.WithHook(xlgo.Hook{Name: "fail", OnInit: func(*xlgo.App) error { return errors.New("boom") }}),
|
||||
)
|
||||
if err := app.Init(); err == nil {
|
||||
_ = app.Shutdown()
|
||||
t.Fatal("expected Init to fail via OnInit hook")
|
||||
}
|
||||
|
||||
if cache.GetDefaultCache() != preInit {
|
||||
t.Fatal("rollback did not restore preInit cache manager as global default")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
package xlgo_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
xlgo "github.com/EthanCodeCraft/xlgo-core"
|
||||
"github.com/EthanCodeCraft/xlgo-core/cron"
|
||||
)
|
||||
|
||||
// TestAppCronSchedulerInstance 固化 Phase 3:WithCronTask 把任务注册到 App 专属调度器,
|
||||
// app.Scheduler() 返回该实例且含注册的任务。
|
||||
func TestAppCronSchedulerInstance(t *testing.T) {
|
||||
saved := cron.SwapDefaultScheduler(cron.NewScheduler())
|
||||
defer cron.SwapDefaultScheduler(saved)
|
||||
|
||||
app := xlgo.New(
|
||||
xlgo.WithConfig(testConfig(18112)),
|
||||
xlgo.WithCronTask("my-task", cron.Every(time.Hour), func(context.Context) error { return nil }),
|
||||
)
|
||||
if err := app.Init(); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
defer app.Shutdown()
|
||||
|
||||
s := app.Scheduler()
|
||||
if s == nil {
|
||||
t.Fatal("app.Scheduler() = nil after Init with WithCronTask")
|
||||
}
|
||||
found := false
|
||||
for _, tk := range s.ListTasks() {
|
||||
if tk.Name == "my-task" {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatal("WithCronTask task not registered on App scheduler")
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppCronRollbackRestoresDefault 固化 Phase 3 回滚:Init 失败后全局默认调度器
|
||||
// 必须恢复为 Init 前的实例(不含 App 注册的任务),而非残留 App 的调度器。
|
||||
func TestAppCronRollbackRestoresDefault(t *testing.T) {
|
||||
saved := cron.SwapDefaultScheduler(cron.NewScheduler())
|
||||
defer cron.SwapDefaultScheduler(saved)
|
||||
|
||||
app := xlgo.New(
|
||||
xlgo.WithConfig(testConfig(18113)),
|
||||
xlgo.WithCronTask("rollback-task", cron.Every(time.Hour), func(context.Context) error { return nil }),
|
||||
xlgo.WithHook(xlgo.Hook{Name: "fail", OnInit: func(*xlgo.App) error { return errors.New("boom") }}),
|
||||
)
|
||||
if err := app.Init(); err == nil {
|
||||
_ = app.Shutdown()
|
||||
t.Fatal("expected Init to fail via OnInit hook")
|
||||
}
|
||||
|
||||
// 回滚后全局默认不应含 App 的 "rollback-task"
|
||||
for _, tk := range cron.GetScheduler().ListTasks() {
|
||||
if tk.Name == "rollback-task" {
|
||||
t.Fatal("rollback did not restore global default scheduler (App task still on global)")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppCronMultiAppIsolation 固化 Phase 3 核心契约:单进程多 App 时,App A 的 Shutdown
|
||||
// 只停 A 自己的调度器,不影响 App B 的调度器继续运行(修复前 cron 是全局单例,A.Shutdown
|
||||
// 会经 cron.StopGlobalWithTimeout 把 B 的调度器也停了)。
|
||||
//
|
||||
// 调度器内部 ticker 为 1s,故任务每 1s tick 触发一次(schedule=Every(1ms) 仅决定 due 判定,
|
||||
// 实际 spawn 受 1s ticker 节拍)。测试因此需要跨 1s tick 观测。
|
||||
func TestAppCronMultiAppIsolation(t *testing.T) {
|
||||
saved := cron.SwapDefaultScheduler(cron.NewScheduler())
|
||||
defer cron.SwapDefaultScheduler(saved)
|
||||
|
||||
var ranA, ranB atomic.Int32
|
||||
appA := xlgo.New(
|
||||
xlgo.WithConfig(testConfig(18110)),
|
||||
xlgo.WithCronTask("taskA", cron.Every(time.Millisecond), func(context.Context) error {
|
||||
ranA.Add(1)
|
||||
return nil
|
||||
}),
|
||||
)
|
||||
appB := xlgo.New(
|
||||
xlgo.WithConfig(testConfig(18111)),
|
||||
xlgo.WithCronTask("taskB", cron.Every(time.Millisecond), func(context.Context) error {
|
||||
ranB.Add(1)
|
||||
return nil
|
||||
}),
|
||||
)
|
||||
if err := appA.Init(); err != nil {
|
||||
t.Fatalf("appA Init: %v", err)
|
||||
}
|
||||
if err := appB.Init(); err != nil {
|
||||
t.Fatalf("appB Init: %v", err)
|
||||
}
|
||||
defer appB.Shutdown()
|
||||
|
||||
// 轮询等第一个 1s tick:两个调度器都应各跑至少一次(deadline 3s,容忍慢机/CI 漂移)
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if ranA.Load() > 0 && ranB.Load() > 0 {
|
||||
break
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
}
|
||||
aAtTick := ranA.Load()
|
||||
bAtTick := ranB.Load()
|
||||
if aAtTick == 0 || bAtTick == 0 {
|
||||
t.Fatalf("expected both schedulers ran at least once: A=%d B=%d", aAtTick, bAtTick)
|
||||
}
|
||||
|
||||
// 停 App A
|
||||
if err := appA.Shutdown(); err != nil {
|
||||
t.Fatalf("appA Shutdown: %v", err)
|
||||
}
|
||||
|
||||
// 轮询等 B 再跑一次(deadline 3s),证明 B 调度器未被 A.Shutdown 停掉
|
||||
deadline = time.Now().Add(3 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if ranB.Load() > bAtTick {
|
||||
break
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
}
|
||||
if ranA.Load() != aAtTick {
|
||||
t.Errorf("App A scheduler ran after its Shutdown (not isolated): before=%d after=%d", aAtTick, ranA.Load())
|
||||
}
|
||||
if ranB.Load() <= bAtTick {
|
||||
t.Errorf("App B scheduler did not continue after App A Shutdown (not isolated): before=%d after=%d", bAtTick, ranB.Load())
|
||||
}
|
||||
}
|
||||
+5
-6
@@ -173,14 +173,13 @@ func TestAppInitFailureRollsBackCron_M1(t *testing.T) {
|
||||
})
|
||||
|
||||
var ran atomic.Int32
|
||||
cron.AddTask("canary-m1", cron.Every(time.Millisecond), func(context.Context) error {
|
||||
ran.Add(1)
|
||||
return nil
|
||||
})
|
||||
|
||||
// Phase 3:canary 注册到 App 专属调度器(WithCronTask),Init 失败时 stopCron 必须停掉它。
|
||||
app := xlgo.New(
|
||||
xlgo.WithConfig(testConfig(18104)),
|
||||
xlgo.WithCron(),
|
||||
xlgo.WithCronTask("canary-m1", cron.Every(time.Millisecond), func(context.Context) error {
|
||||
ran.Add(1)
|
||||
return nil
|
||||
}),
|
||||
xlgo.WithHook(xlgo.Hook{Name: "fail", OnInit: func(*xlgo.App) error { return errors.New("boom-init") }}),
|
||||
)
|
||||
err := app.Init()
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package xlgo_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
xlgo "github.com/EthanCodeCraft/xlgo-core"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// TestAppRateLimitPerAppCountingIsolation 固化 #1 修复:app.RateLimitRegistry().LoginRateLimit()
|
||||
// 返回的中间件捕获 App 自己的 Registry,多 App 下 per-App 计数隔离。修复前包级 LoginRateLimit()
|
||||
// 解析全局默认(多 App 仅最后 Init 的 App),App A 用满额度后 App B 首请求即被拒。
|
||||
//
|
||||
// 验证:App A 用满 10/10 后第 11 次 429;App B 同 IP 仍可 10 次 200、第 11 次 429(独立计数)。
|
||||
func TestAppRateLimitPerAppCountingIsolation(t *testing.T) {
|
||||
// REST 模式:限流映射 HTTP 429
|
||||
cfgA := testConfig(18140)
|
||||
cfgA.Server.ResponseMode = "rest"
|
||||
cfgB := testConfig(18141)
|
||||
cfgB.Server.ResponseMode = "rest"
|
||||
|
||||
appA := xlgo.New(xlgo.WithConfig(cfgA))
|
||||
if appA.RateLimitRegistry() == nil {
|
||||
t.Fatal("appA.RateLimitRegistry() = nil before Init (should be pre-created in New)")
|
||||
}
|
||||
if err := appA.Init(); err != nil {
|
||||
t.Fatalf("appA Init: %v", err)
|
||||
}
|
||||
defer appA.Shutdown()
|
||||
// app-bound 中间件:捕获 appA 的 Registry,不查全局默认
|
||||
appA.GetRouter().GET("/login", appA.RateLimitRegistry().LoginRateLimit(), func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
})
|
||||
|
||||
appB := xlgo.New(xlgo.WithConfig(cfgB))
|
||||
if err := appB.Init(); err != nil {
|
||||
t.Fatalf("appB Init: %v", err)
|
||||
}
|
||||
defer appB.Shutdown()
|
||||
appB.GetRouter().GET("/login", appB.RateLimitRegistry().LoginRateLimit(), func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
})
|
||||
|
||||
req := func(h http.Handler) *httptest.ResponseRecorder {
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest(http.MethodGet, "/login", nil)
|
||||
r.RemoteAddr = "192.0.2.7:1234" // 两 App 用同一 IP
|
||||
h.ServeHTTP(w, r)
|
||||
return w
|
||||
}
|
||||
|
||||
// App A 用满 10 次(200),第 11 次 429
|
||||
for i := 0; i < 10; i++ {
|
||||
if w := req(appA.GetRouter()); w.Code != http.StatusOK {
|
||||
t.Fatalf("appA request %d: got %d, want 200", i+1, w.Code)
|
||||
}
|
||||
}
|
||||
if w := req(appA.GetRouter()); w.Code != http.StatusTooManyRequests {
|
||||
t.Fatalf("appA 11th: got %d, want 429 (limit reached)", w.Code)
|
||||
}
|
||||
|
||||
// App B 同 IP:应独立计数,10 次 200、第 11 次 429。
|
||||
// 包级 facade 下 B 会共享 A 的 loginLimiter(A 已用满)-> 首请求即 429,本测试即暴露该回归。
|
||||
for i := 0; i < 10; i++ {
|
||||
if w := req(appB.GetRouter()); w.Code != http.StatusOK {
|
||||
t.Fatalf("appB request %d: got %d, want 200 (per-App isolation: B should not share A's count)", i+1, w.Code)
|
||||
}
|
||||
}
|
||||
if w := req(appB.GetRouter()); w.Code != http.StatusTooManyRequests {
|
||||
t.Fatalf("appB 11th: got %d, want 429 (B independent limit reached)", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppRateLimitRegistryCustomNoLeakOnRollback 固化 #2 修复路径:app-bound CustomRateLimit
|
||||
// 的 limiter 登记在 App 的 Registry 上,Init 失败回滚时由 registry.Stop() 收口,不泄漏。
|
||||
// 验证:注册 CustomRateLimit 任务后 Init 失败,rollback 调 registry.Stop(customLimiters 被清)。
|
||||
func TestAppRateLimitRegistryCustomNoLeakOnRollback(t *testing.T) {
|
||||
cfg := testConfig(18142)
|
||||
app := xlgo.New(xlgo.WithConfig(cfg),
|
||||
xlgo.WithHook(xlgo.Hook{Name: "fail", OnInit: func(*xlgo.App) error { return errors.New("boom") }}),
|
||||
)
|
||||
// app-bound CustomRateLimit(首请求才创建 limiter,但 Init 会失败故不会触发请求;
|
||||
// 此用例固化 app.RateLimitRegistry() 在 pre-Init 可用、Init 失败回滚不 panic)
|
||||
app.GetRouter().GET("/c", app.RateLimitRegistry().CustomRateLimit(5, time.Minute), func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
})
|
||||
|
||||
if err := app.Init(); err == nil {
|
||||
_ = app.Shutdown()
|
||||
t.Fatal("expected Init to fail via OnInit hook")
|
||||
}
|
||||
// Init 失败后 rollbackReplacedResources 已 Stop App 的 Registry 并把 a.rateLimitRegistry 置 nil,
|
||||
// 故 app.RateLimitRegistry() 返回 nil(rollback 已收口,无泄漏、无 panic)。
|
||||
if app.RateLimitRegistry() != nil {
|
||||
t.Fatal("a.rateLimitRegistry should be nil after Init-failure rollback")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
package xlgo_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
xlgo "github.com/EthanCodeCraft/xlgo-core"
|
||||
"github.com/EthanCodeCraft/xlgo-core/middleware"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// TestAppRateLimitMultiAppShutdownIsolation 固化 Phase 5 核心契约:单进程多 App 时,
|
||||
// App A 的 Shutdown 不得重置/停止 App B 的限流器。修复前 App.Shutdown 调全局
|
||||
// middleware.StopRateLimiters,把全局 loginLimiter 置 nil,App B 下次请求会懒创建
|
||||
// 新 limiter(计数清零)--稳态客户端借 App A 的 Shutdown 窗口绕过限流。
|
||||
//
|
||||
// 验证:App B 的 /login 已用 10/10 次后,App A.Shutdown,第 11 次仍应 429(计数保留)。
|
||||
// 修复前第 11 次会 200(计数被重置)。
|
||||
func TestAppRateLimitMultiAppShutdownIsolation(t *testing.T) {
|
||||
saved := middleware.SwapDefaultRateLimitRegistry(middleware.NewRateLimitRegistry())
|
||||
defer middleware.SwapDefaultRateLimitRegistry(saved)
|
||||
|
||||
// 用 REST 响应模式:限流映射 HTTP 429(business 模式下是 200+业务码 429,不便断言)
|
||||
cfgA := testConfig(18130)
|
||||
cfgA.Server.ResponseMode = "rest"
|
||||
cfgB := testConfig(18131)
|
||||
cfgB.Server.ResponseMode = "rest"
|
||||
|
||||
appA := xlgo.New(xlgo.WithConfig(cfgA))
|
||||
if err := appA.Init(); err != nil {
|
||||
t.Fatalf("appA Init: %v", err)
|
||||
}
|
||||
appA.GetRouter().GET("/login", middleware.LoginRateLimit(), func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
})
|
||||
|
||||
appB := xlgo.New(xlgo.WithConfig(cfgB))
|
||||
if err := appB.Init(); err != nil {
|
||||
t.Fatalf("appB Init: %v", err)
|
||||
}
|
||||
defer appB.Shutdown()
|
||||
appB.GetRouter().GET("/login", middleware.LoginRateLimit(), func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
})
|
||||
|
||||
// 10 次请求 App B 的 /login(限流 10/min,同一 IP)-> 全 200
|
||||
for i := 0; i < 10; i++ {
|
||||
w := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/login", nil)
|
||||
req.RemoteAddr = "192.0.2.1:1234"
|
||||
appB.GetRouter().ServeHTTP(w, req)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("request %d: got %d, want 200 (under limit)", i+1, w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// 第 11 次(不经 A.Shutdown)应 429 -- 先确认限流器确实计数到上限(排除 IP/计数 bug)
|
||||
{
|
||||
w := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/login", nil)
|
||||
req.RemoteAddr = "192.0.2.1:1234"
|
||||
appB.GetRouter().ServeHTTP(w, req)
|
||||
if w.Code != http.StatusTooManyRequests {
|
||||
t.Fatalf("sanity: 11th without shutdown got %d, want 429 (limiter not counting?)", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// App A Shutdown -- 不得重置/停止 App B 的 login limiter
|
||||
if err := appA.Shutdown(); err != nil {
|
||||
t.Fatalf("appA Shutdown: %v", err)
|
||||
}
|
||||
|
||||
// 第 12 次请求 App B 的 /login -> 应仍 429(计数保留在 10,registryB 未被 A.Shutdown 重置)。
|
||||
// 修复前 A.Shutdown 调 StopRateLimiters 把全局 loginLimiter 置 nil,下次请求懒创建新 limiter(计数清零)-> 200。
|
||||
{
|
||||
w := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/login", nil)
|
||||
req.RemoteAddr = "192.0.2.1:1234"
|
||||
appB.GetRouter().ServeHTTP(w, req)
|
||||
if w.Code != http.StatusTooManyRequests {
|
||||
t.Errorf("12th request after appA Shutdown: got %d, want 429 (limiter count should be preserved, not reset by appA.Shutdown)", w.Code)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppRateLimitRollbackOnInitFailure 固化 Phase 5 回滚:Init 失败后全局默认 Registry
|
||||
// 必须恢复为 Init 前的实例。
|
||||
func TestAppRateLimitRollbackOnInitFailure(t *testing.T) {
|
||||
preInit := middleware.NewRateLimitRegistry()
|
||||
saved := middleware.SwapDefaultRateLimitRegistry(preInit)
|
||||
defer middleware.SwapDefaultRateLimitRegistry(saved)
|
||||
|
||||
app := xlgo.New(
|
||||
xlgo.WithConfig(testConfig(18132)),
|
||||
xlgo.WithHook(xlgo.Hook{Name: "fail", OnInit: func(*xlgo.App) error { return errors.New("boom") }}),
|
||||
)
|
||||
if err := app.Init(); err == nil {
|
||||
_ = app.Shutdown()
|
||||
t.Fatal("expected Init to fail via OnInit hook")
|
||||
}
|
||||
|
||||
if middleware.GetDefaultRateLimitRegistry() != preInit {
|
||||
t.Fatal("rollback did not restore preInit rate limit registry as global default")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package xlgo_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
xlgo "github.com/EthanCodeCraft/xlgo-core"
|
||||
"github.com/EthanCodeCraft/xlgo-core/config"
|
||||
"github.com/EthanCodeCraft/xlgo-core/storage"
|
||||
)
|
||||
|
||||
// localStorageCfg 构造一个本地存储配置(tmpdir 由调用方提供)。
|
||||
func localStorageCfg(port int, dir string) *config.Config {
|
||||
cfg := testConfig(port)
|
||||
cfg.Storage = config.StorageConfig{
|
||||
Driver: "local",
|
||||
Local: config.LocalStorageConfig{Path: dir},
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// TestAppWithStorageSwapsDefault 固化 Phase 2 契约:App.Init(WithStorage) 必须把 App 的
|
||||
// StorageManager 提升为全局默认。证据:Init 前 GetStorage()=nil(空 manager),Init 后非 nil。
|
||||
func TestAppWithStorageSwapsDefault(t *testing.T) {
|
||||
// 用空 manager 作"Init 前"默认,确保 GetStorage() 起点 nil
|
||||
saved := storage.SwapDefaultStorageManager(storage.NewStorageManager())
|
||||
defer storage.SwapDefaultStorageManager(saved)
|
||||
|
||||
if s := storage.GetStorage(); s != nil {
|
||||
t.Fatalf("precondition: GetStorage() should be nil, got %T", s)
|
||||
}
|
||||
|
||||
dir := t.TempDir()
|
||||
app := xlgo.New(xlgo.WithConfig(localStorageCfg(18094, dir)), xlgo.WithStorage())
|
||||
if err := app.Init(); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
defer app.Shutdown()
|
||||
|
||||
if s := storage.GetStorage(); s == nil {
|
||||
t.Fatal("storage.GetStorage() = nil after App.Init with WithStorage(未把 App manager 提升为全局默认)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppStorageRollbackOnInitFailure 固化 Phase 2 回滚契约:Init 在 storage 之后失败时,
|
||||
// rollbackReplacedResources 必须把全局默认 swap 回 Init 前的旧 manager。
|
||||
// 用 OnInit hook 触发失败(OnInit 在 storage init 之后、commitReplacedResources 之前)。
|
||||
func TestAppStorageRollbackOnInitFailure(t *testing.T) {
|
||||
saved := storage.SwapDefaultStorageManager(storage.NewStorageManager())
|
||||
defer storage.SwapDefaultStorageManager(saved)
|
||||
|
||||
dir := t.TempDir()
|
||||
app := xlgo.New(
|
||||
xlgo.WithConfig(localStorageCfg(18095, dir)),
|
||||
xlgo.WithStorage(),
|
||||
xlgo.WithHook(xlgo.Hook{
|
||||
Name: "force-fail",
|
||||
OnInit: func(*xlgo.App) error { return errors.New("boom") },
|
||||
}),
|
||||
)
|
||||
|
||||
if err := app.Init(); err == nil {
|
||||
_ = app.Shutdown()
|
||||
t.Fatal("expected Init to fail via OnInit hook")
|
||||
}
|
||||
|
||||
// 回滚后全局默认应恢复为 saved(空 manager)-> GetStorage() 返回 nil
|
||||
if s := storage.GetStorage(); s != nil {
|
||||
t.Errorf("Init 失败回滚未恢复 storage 全局默认: GetStorage()=%T", s)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
package xlgo_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"runtime"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
xlgo "github.com/EthanCodeCraft/xlgo-core"
|
||||
"github.com/EthanCodeCraft/xlgo-core/trace"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// TestWithTraceInitErrorPropagated 固化 Phase 1 契约:WithTrace 启用后 doInit 必须调
|
||||
// trace.Init 并把错误向上传播。用非法 ExporterType 触发 trace.Init 失败(config.Validate
|
||||
// 只校验 SampleRatio,故 bogus exporter 能过 Validate、在 trace.Init 失败,证明是 App 主动调 Init)。
|
||||
func TestWithTraceInitErrorPropagated(t *testing.T) {
|
||||
cfg := testConfig(18090)
|
||||
cfg.Trace.Enabled = true
|
||||
cfg.Trace.ExporterType = "bogus-exporter"
|
||||
|
||||
app := xlgo.New(xlgo.WithConfig(cfg), xlgo.WithTrace())
|
||||
err := app.Init()
|
||||
if err == nil {
|
||||
_ = app.Shutdown()
|
||||
t.Fatal("expected Init to fail with bogus trace exporter, got nil")
|
||||
}
|
||||
// "初始化 trace 失败" 包裹 trace.Init 的 "不支持的导出器类型"
|
||||
msg := err.Error()
|
||||
if !strings.Contains(msg, "trace") && !strings.Contains(msg, "导出器") {
|
||||
t.Fatalf("expected trace init error, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWithTraceNoopMiddlewareDoesNotBreakRequest 固化:WithTrace + Enabled=false(默认)
|
||||
// 时 trace.Init 安装 Noop tracer,trace.Middleware 装入链后不应破坏正常请求。
|
||||
func TestWithTraceNoopMiddlewareDoesNotBreakRequest(t *testing.T) {
|
||||
cfg := testConfig(18091)
|
||||
// Trace.Enabled 保持默认 false -> noop
|
||||
|
||||
app := xlgo.New(xlgo.WithConfig(cfg), xlgo.WithTrace())
|
||||
if err := app.Init(); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
defer app.Shutdown()
|
||||
|
||||
// Init 后(middleware 链已装入)注册路由,确保该路由经过 trace 中间件。
|
||||
app.GetRouter().GET("/ping", func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
})
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/ping", nil)
|
||||
app.GetRouter().ServeHTTP(w, req)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("trace noop middleware broke request: got status %d, body %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestWithTraceShutdownReleasesExporterGoroutine 固化 Phase 1 生命周期闭环:
|
||||
// WithTrace + Enabled=true + stdout exporter,Init 后 BatchSpanProcessor 后台 goroutine
|
||||
// 在运行;Shutdown 必须调 trace.Close 让其退出,否则 goroutine 泄漏(H-14 修了 Close 幂等,
|
||||
// 但 App 此前从未调 Close -- 本测试断言 App.Shutdown 确实调了)。
|
||||
//
|
||||
// 不发任何请求 -> stdout exporter 不输出 -> 无测试输出污染;仅观测 goroutine 计数。
|
||||
func TestWithTraceShutdownReleasesExporterGoroutine(t *testing.T) {
|
||||
cfg := testConfig(18092)
|
||||
cfg.Trace.Enabled = true
|
||||
cfg.Trace.ExporterType = "stdout"
|
||||
|
||||
app := xlgo.New(xlgo.WithConfig(cfg), xlgo.WithTrace())
|
||||
|
||||
baseline := runtime.NumGoroutine()
|
||||
if err := app.Init(); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
afterInit := runtime.NumGoroutine()
|
||||
if afterInit <= baseline {
|
||||
t.Logf("note: afterInit=%d baseline=%d (batch goroutine delta not detected; test may be noisy)", afterInit, baseline)
|
||||
}
|
||||
|
||||
if err := app.Shutdown(); err != nil {
|
||||
t.Fatalf("Shutdown: %v", err)
|
||||
}
|
||||
|
||||
// trace.Close 的 provider.Shutdown 异步停止 batch goroutine,轮询等待回归 baseline。
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if runtime.NumGoroutine() <= baseline {
|
||||
break
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
if after := runtime.NumGoroutine(); after > baseline {
|
||||
t.Errorf("trace.Close 未释放 exporter goroutine(App.Shutdown 疑未调 trace.Close): baseline=%d after=%d", baseline, after)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWithTraceEnabledStdoutInitOK 固化:WithTrace + Enabled=true + stdout 是合法组合,
|
||||
// trace.Init 成功,App.Init 成功;二次 trace.Close 幂等(H-14)。
|
||||
func TestWithTraceEnabledStdoutInitOK(t *testing.T) {
|
||||
cfg := testConfig(18093)
|
||||
cfg.Trace.Enabled = true
|
||||
cfg.Trace.ExporterType = "stdout"
|
||||
|
||||
app := xlgo.New(xlgo.WithConfig(cfg), xlgo.WithTrace())
|
||||
if err := app.Init(); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
if err := app.Shutdown(); err != nil {
|
||||
t.Fatalf("Shutdown: %v", err)
|
||||
}
|
||||
// Shutdown 已调 trace.Close;再调一次必须幂等不报错(H-14 契约)
|
||||
if err := trace.Close(context.Background()); err != nil {
|
||||
t.Fatalf("trace.Close after Shutdown should be idempotent: %v", err)
|
||||
}
|
||||
}
|
||||
Vendored
+35
-4
@@ -32,19 +32,33 @@ type CacheService interface {
|
||||
//
|
||||
// 不在构造时快照 redis.Client(M12 修复:原 NewRedisCache 构造时取 database.GetRedis(),
|
||||
// 若在 database.InitRedis 之前构造则永久 nil、即使后续 Redis 就绪也是 no-op)。
|
||||
// 改为每次操作实时取 database.GetRedis(),使"先构造后 Init Redis"的顺序也能正确工作。
|
||||
type redisCache struct{}
|
||||
// 改为每次操作实时取 redis 客户端,使"先构造后 Init Redis"的顺序也能正确工作。
|
||||
//
|
||||
// Phase 4:持有可选的注入 client(NewRedisCacheWithRedis),注入优先、否则回退全局
|
||||
// database.GetRedis()(照 jwt.TokenBlacklist.redisClient 模型)。App 经 InitWithRedis
|
||||
// 注入 per-App redisManager.Client(),使缓存走 App 自己的 Redis 而非全局(修复跨 App 串越)。
|
||||
type redisCache struct {
|
||||
injectedClient *redis.Client
|
||||
}
|
||||
|
||||
// client 返回当前 Redis 客户端(实时取,未初始化则 nil)。
|
||||
// client 返回缓存使用的 Redis 客户端:注入优先,否则回退全局 database.GetRedis()。
|
||||
func (c *redisCache) client() *redis.Client {
|
||||
if c != nil && c.injectedClient != nil {
|
||||
return c.injectedClient
|
||||
}
|
||||
return database.GetRedis()
|
||||
}
|
||||
|
||||
// NewRedisCache 创建 Redis 缓存实例
|
||||
// NewRedisCache 创建 Redis 缓存实例(懒取全局 Redis,兼容 standalone 用法)。
|
||||
func NewRedisCache() CacheService {
|
||||
return &redisCache{}
|
||||
}
|
||||
|
||||
// NewRedisCacheWithRedis 创建使用指定 Redis 客户端的缓存实例(多 Redis/测试隔离)。
|
||||
func NewRedisCacheWithRedis(client *redis.Client) CacheService {
|
||||
return &redisCache{injectedClient: client}
|
||||
}
|
||||
|
||||
// Get 获取缓存值。命中返回 (true, nil);未命中返回 (false, nil);
|
||||
// Redis 未就绪返回 (false, ErrRedisNotReady);Redis 命令错误或反序列化失败返回 (false, err)。
|
||||
func (c *redisCache) Get(ctx context.Context, key string, dest any) (bool, error) {
|
||||
@@ -191,6 +205,16 @@ func SetDefaultCacheManager(m *CacheManager) {
|
||||
}
|
||||
}
|
||||
|
||||
// SwapDefaultCacheManager 将指定 CacheManager 置为全局默认,返回被替换的旧 Manager。
|
||||
// 旧 Manager 不会被关闭,供 App 初始化这类需要失败回滚的生命周期流程暂存(照
|
||||
// SwapDefaultRedisManager / SwapDefaultStorageManager 模式)。nil 被忽略,返回当前默认。
|
||||
func SwapDefaultCacheManager(m *CacheManager) *CacheManager {
|
||||
if m == nil {
|
||||
return defaultCachePtr.Load()
|
||||
}
|
||||
return defaultCachePtr.Swap(m)
|
||||
}
|
||||
|
||||
// Init 初始化缓存服务(基于 DefaultRedis 的客户端)。
|
||||
func (m *CacheManager) Init() {
|
||||
m.mu.Lock()
|
||||
@@ -198,6 +222,13 @@ func (m *CacheManager) Init() {
|
||||
m.svc = NewRedisCache()
|
||||
}
|
||||
|
||||
// InitWithRedis 用指定 Redis 客户端初始化缓存服务(多 Redis/测试隔离)。
|
||||
func (m *CacheManager) InitWithRedis(client *redis.Client) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.svc = NewRedisCacheWithRedis(client)
|
||||
}
|
||||
|
||||
// Set 设置缓存服务实现(用于注入 mock 或自定义实现)。
|
||||
func (m *CacheManager) Set(svc CacheService) {
|
||||
m.mu.Lock()
|
||||
|
||||
Vendored
+69
@@ -0,0 +1,69 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/EthanCodeCraft/xlgo-core/database"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// TestRedisCacheInjectedClientPriority 固化 Phase 4 核心契约(jwt 模型):
|
||||
// redisCache.client() 注入优先、全局兜底。注入的 client 即使与 database.GetRedis()
|
||||
// 不同也必须用注入的--这是多 App 缓存隔离的基础。
|
||||
func TestRedisCacheInjectedClientPriority(t *testing.T) {
|
||||
// 全局 redis = clientG
|
||||
mrG := miniredis.RunT(t)
|
||||
clientG := redis.NewClient(&redis.Options{Addr: mrG.Addr()})
|
||||
t.Cleanup(func() { _ = clientG.Close() })
|
||||
origGlobal := database.SetTestRedisClient(clientG)
|
||||
t.Cleanup(func() { database.SetTestRedisClient(origGlobal) })
|
||||
|
||||
// 注入 clientA(与全局 clientG 不同实例)
|
||||
mrA := miniredis.RunT(t)
|
||||
clientA := redis.NewClient(&redis.Options{Addr: mrA.Addr()})
|
||||
t.Cleanup(func() { _ = clientA.Close() })
|
||||
|
||||
// 注入优先
|
||||
injected := &redisCache{injectedClient: clientA}
|
||||
if got := injected.client(); got != clientA {
|
||||
t.Fatalf("injected redisCache.client() = %v, want clientA (注入优先)", got)
|
||||
}
|
||||
|
||||
// 未注入 -> 回退全局 clientG
|
||||
fallback := &redisCache{}
|
||||
if got := fallback.client(); got != clientG {
|
||||
t.Fatalf("fallback redisCache.client() = %v, want clientG (全局兜底)", got)
|
||||
}
|
||||
|
||||
// NewRedisCacheWithRedis 构造的实例持注入 client
|
||||
svc := NewRedisCacheWithRedis(clientA)
|
||||
rc, ok := svc.(*redisCache)
|
||||
if !ok {
|
||||
t.Fatalf("NewRedisCacheWithRedis returned %T, want *redisCache", svc)
|
||||
}
|
||||
if rc.client() != clientA {
|
||||
t.Fatalf("NewRedisCacheWithRedis client = %v, want clientA", rc.client())
|
||||
}
|
||||
}
|
||||
|
||||
// TestSwapDefaultCacheManagerReturnsOld 固化 Phase 4:Swap 返回被替换的旧 Manager 且不关闭。
|
||||
func TestSwapDefaultCacheManagerReturnsOld(t *testing.T) {
|
||||
orig := defaultCachePtr.Load()
|
||||
defer SwapDefaultCacheManager(orig)
|
||||
|
||||
first := NewCacheManager()
|
||||
if prev := SwapDefaultCacheManager(first); prev != orig {
|
||||
t.Fatalf("Swap returned %v, want orig", prev)
|
||||
}
|
||||
second := NewCacheManager()
|
||||
if returned := SwapDefaultCacheManager(second); returned != first {
|
||||
t.Fatalf("Swap returned %v, want first", returned)
|
||||
}
|
||||
if defaultCachePtr.Load() != second {
|
||||
t.Fatal("Swap did not install second as default")
|
||||
}
|
||||
if got := SwapDefaultCacheManager(nil); got != second {
|
||||
t.Fatalf("Swap(nil) = %v, want second (current default)", got)
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,7 @@ type Config struct {
|
||||
Upload UploadConfig `mapstructure:"upload"`
|
||||
Log LogConfig `mapstructure:"log"`
|
||||
CORS CORSConfig `mapstructure:"cors"`
|
||||
Trace TraceConfig `mapstructure:"trace"`
|
||||
}
|
||||
|
||||
// Clone 返回 Config 的深拷贝(M-G 修复)。
|
||||
@@ -524,6 +525,26 @@ type CORSConfig struct {
|
||||
MaxAge int `mapstructure:"max_age"` // 预检请求缓存时间(秒)
|
||||
}
|
||||
|
||||
// TraceConfig 链路追踪配置(OpenTelemetry)。由 App 在 WithTrace 时读取并调 trace.Init。
|
||||
//
|
||||
// 语义说明:OTel 的 TracerProvider / TextMapPropagator 本身是进程级全局单例
|
||||
// (otel.SetTracerProvider 全局生效),故 trace 不做 per-App 实例隔离--多 App 进程
|
||||
// 共享同一 OTel 全局状态,这与 OTel 自身设计一致。App 仅负责把 trace 纳入生命周期
|
||||
// (Init/Close)与装入 Middleware,不提供实例隔离。
|
||||
//
|
||||
// Enabled=false(默认)时 trace.Init 安装 Noop tracer,Middleware 不 panic、不导出。
|
||||
type TraceConfig struct {
|
||||
ServiceName string `mapstructure:"service_name"` // 服务名(空则回退 cfg.App.Name)
|
||||
ServiceVersion string `mapstructure:"service_version"` // 服务版本
|
||||
Environment string `mapstructure:"environment"` // 运行环境
|
||||
ExporterType string `mapstructure:"exporter_type"` // otlp-http(默认) / otlp-grpc / stdout
|
||||
Endpoint string `mapstructure:"endpoint"` // OTLP collector 地址
|
||||
Insecure bool `mapstructure:"insecure"` // 明文(无 TLS)连接 collector
|
||||
SampleRatio float64 `mapstructure:"sample_ratio"` // 采样比例 0.0-1.0
|
||||
Enabled bool `mapstructure:"enabled"` // 是否启用导出
|
||||
Propagator string `mapstructure:"propagator"` // w3c(默认) / b3 / jaeger
|
||||
}
|
||||
|
||||
// GetAllowedOrigins 获取允许的域名列表
|
||||
func (c *CORSConfig) GetAllowedOrigins() []string {
|
||||
if c == nil || len(c.AllowedOrigins) == 0 {
|
||||
|
||||
@@ -82,6 +82,14 @@ func (c *Config) Validate() error {
|
||||
}
|
||||
}
|
||||
|
||||
// Trace:仅当启用时校验采样比例。ExporterType/Propagator 的枚举校验委托 trace.Init
|
||||
// (随 OTel 版本可能扩展,避免 config 与 trace 双源枚举漂移)。
|
||||
if c.Trace.Enabled {
|
||||
if c.Trace.SampleRatio < 0 || c.Trace.SampleRatio > 1 {
|
||||
problems = append(problems, fmt.Sprintf("trace.sample_ratio 超出范围(0.0-1.0): %v", c.Trace.SampleRatio))
|
||||
}
|
||||
}
|
||||
|
||||
if len(problems) > 0 {
|
||||
return fmt.Errorf("配置校验失败: %s", strings.Join(problems, "; "))
|
||||
}
|
||||
|
||||
@@ -43,6 +43,24 @@ func TestValidateJWTSecretAbsentSkipped(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateTraceSampleRatioWhenEnabled(t *testing.T) {
|
||||
c := validBase()
|
||||
c.Trace.Enabled = true
|
||||
c.Trace.SampleRatio = 1.5
|
||||
if err := c.Validate(); err == nil || !strings.Contains(err.Error(), "trace.sample_ratio") {
|
||||
t.Fatalf("expected trace.sample_ratio error, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateTraceDisabledSkipsRatio(t *testing.T) {
|
||||
c := validBase()
|
||||
// Enabled=false 时即使 SampleRatio 非法也不校验(trace 未启用)
|
||||
c.Trace.SampleRatio = 1.5
|
||||
if err := c.Validate(); err != nil {
|
||||
t.Fatalf("unexpected error when trace disabled: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateDatabaseMissingHost(t *testing.T) {
|
||||
c := validBase()
|
||||
c.Database.Driver = "mysql"
|
||||
|
||||
+18
-3
@@ -784,11 +784,16 @@ func Cron(minute, hour string) *CronSchedule {
|
||||
return &CronSchedule{Minute: minute, Hour: hour}
|
||||
}
|
||||
|
||||
// 全局调度器。用 atomic.Pointer 懒初始化,使 StopGlobalWithTimeout 能安全 peek
|
||||
// 是否已创建而不触发创建,也消除原 once+裸指针读写的竞态。
|
||||
// 全局调度器。init 时预创建默认实例(与 storage.DefaultStorage / cache.defaultCachePtr
|
||||
// 对齐),使 SwapDefaultScheduler 返回非 nil、App 回滚总能恢复一个有效默认。用 atomic.Pointer
|
||||
// 保护并发读写,消除原 once+裸指针读写竞态。GetScheduler 仍保留懒初始化分支作防御。
|
||||
var globalScheduler atomic.Pointer[Scheduler]
|
||||
|
||||
// GetScheduler 获取全局调度器(懒初始化,并发安全)。
|
||||
func init() {
|
||||
globalScheduler.Store(NewScheduler())
|
||||
}
|
||||
|
||||
// GetScheduler 获取全局调度器(并发安全;init 后通常非 nil,保留懒初始化作防御)。
|
||||
func GetScheduler() *Scheduler {
|
||||
if s := globalScheduler.Load(); s != nil {
|
||||
return s
|
||||
@@ -800,6 +805,16 @@ func GetScheduler() *Scheduler {
|
||||
return globalScheduler.Load()
|
||||
}
|
||||
|
||||
// SwapDefaultScheduler 将指定 Scheduler 置为全局默认,并返回被替换的旧调度器。
|
||||
// 旧调度器不会被停止,供 App 初始化这类需要失败回滚的生命周期流程暂存(照
|
||||
// database.SwapDefaultManager / SwapDefaultRedisManager 模式)。nil 被忽略,返回当前默认。
|
||||
func SwapDefaultScheduler(s *Scheduler) *Scheduler {
|
||||
if s == nil {
|
||||
return globalScheduler.Load()
|
||||
}
|
||||
return globalScheduler.Swap(s)
|
||||
}
|
||||
|
||||
// AddTask 添加任务到全局调度器
|
||||
func AddTask(name string, schedule Schedule, handler TaskHandler) *Task {
|
||||
return GetScheduler().AddTask(name, schedule, handler)
|
||||
|
||||
+230
-110
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/EthanCodeCraft/xlgo-core/database"
|
||||
"github.com/EthanCodeCraft/xlgo-core/response"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// RateLimiter 速率限制器(内存版,单实例使用)
|
||||
@@ -167,6 +168,7 @@ type RedisRateLimiter struct {
|
||||
rate int // 每分钟允许的请求数
|
||||
window time.Duration // 时间窗口
|
||||
failClosed atomic.Bool // H4c/H-6: Redis 错误/断言失败时是否拒绝(true=安全型 fail-closed)。atomic 以支持运行期 SetFailClosed 并发安全切换。
|
||||
client *redis.Client // Phase 5: 注入的 redis client(nil 回退 database.GetRedis(),照 jwt.TokenBlacklist 模型)。App 经 WithRedisClient 注入 per-App client 实现隔离。
|
||||
}
|
||||
|
||||
// slidingWindowLua 滑动窗口限流 Lua 脚本
|
||||
@@ -204,6 +206,14 @@ func WithFailClosed(failClosed bool) RedisRateLimiterOption {
|
||||
}
|
||||
}
|
||||
|
||||
// WithRedisClient 注入指定 Redis 客户端(Phase 5,多 Redis/测试隔离)。
|
||||
// 不传则回退 database.GetRedis()(全局)。App 可经 app.RedisClient() 取自身 client 注入。
|
||||
func WithRedisClient(client *redis.Client) RedisRateLimiterOption {
|
||||
return func(rl *RedisRateLimiter) {
|
||||
rl.client = client
|
||||
}
|
||||
}
|
||||
|
||||
// NewRedisRateLimiter 创建 Redis 分布式限流器。
|
||||
// 默认 fail-open(Redis 故障时放行,避免影响业务);安全敏感场景传 WithFailClosed(true)。
|
||||
func NewRedisRateLimiter(keyPrefix string, rate int, window time.Duration, opts ...RedisRateLimiterOption) *RedisRateLimiter {
|
||||
@@ -228,20 +238,31 @@ func (rl *RedisRateLimiter) SetFailClosed(failClosed bool) {
|
||||
rl.failClosed.Store(failClosed)
|
||||
}
|
||||
|
||||
// redisClient 返回注入的 redis client,未注入则回退 database.GetRedis()(jwt 模型)。
|
||||
// M-C 教训:取一次复用,避免 nil 检查与 Eval 各调一次 GetRedis() 之间的 CloseRedis 竞态。
|
||||
func (rl *RedisRateLimiter) redisClient() *redis.Client {
|
||||
if rl != nil && rl.client != nil {
|
||||
return rl.client
|
||||
}
|
||||
return database.GetRedis()
|
||||
}
|
||||
|
||||
// Allow 检查是否允许请求。
|
||||
//
|
||||
// H4c 修复:
|
||||
// - result.(int64) 改 comma-ok,断言失败返 ErrRedisRateLimiterUnexpectedResult 而非 panic。
|
||||
// - Redis 错误/断言失败时按 failClosed 策略决定:fail-closed 返 (false, err) 拒绝,
|
||||
// fail-open 返 (true, err) 放行(兼容旧行为)。中间件层据此 allowed 值决定放行/拒绝,
|
||||
// 不再无条件 fail-open——登录防爆破等安全场景用 fail-closed 限流器即可在 Redis 故障时拒绝。
|
||||
// 不再无条件 fail-open--登录防爆破等安全场景用 fail-closed 限流器即可在 Redis 故障时拒绝。
|
||||
//
|
||||
// Redis 未启用(database.GetRedis() == nil)时:fail-closed 返 (false, ErrRedisRateLimiterUnavailable),
|
||||
// Redis 未启用(redisClient() == nil)时:fail-closed 返 (false, ErrRedisRateLimiterUnavailable),
|
||||
// fail-open 返 (true, nil)(兼容旧行为)。安全场景必须确保 Redis 已启用。
|
||||
//
|
||||
// Phase 5:redisClient() 注入优先、全局兜底,App 经 WithRedisClient 注入 per-App client。
|
||||
func (rl *RedisRateLimiter) Allow(ctx context.Context, identifier string) (bool, error) {
|
||||
// M-C 修复:GetRedis() 仅取一次复用。原实现 nil 检查与 Eval 各调一次 GetRedis(),
|
||||
// 两次之间若 CloseRedis,第二次返回 nil → nil.Eval panic。
|
||||
rdb := database.GetRedis()
|
||||
// M-C 修复:取一次复用。原实现 nil 检查与 Eval 各调一次 GetRedis(),
|
||||
// 两次之间若 CloseRedis,第二次返回 nil -> nil.Eval panic。
|
||||
rdb := rl.redisClient()
|
||||
if rdb == nil {
|
||||
if rl.failClosed.Load() {
|
||||
return false, ErrRedisRateLimiterUnavailable
|
||||
@@ -274,8 +295,8 @@ func (rl *RedisRateLimiter) Allow(ctx context.Context, identifier string) (bool,
|
||||
|
||||
// GetCount 获取当前窗口内的请求数
|
||||
func (rl *RedisRateLimiter) GetCount(ctx context.Context, identifier string) (int64, error) {
|
||||
// M-C 修复:GetRedis() 取一次复用(原三次调用存在 nil-deref 窗口)。
|
||||
rdb := database.GetRedis()
|
||||
// M-C 修复(Phase 5:改 redisClient):取一次复用(原三次调用存在 nil-deref 窗口)。
|
||||
rdb := rl.redisClient()
|
||||
if rdb == nil {
|
||||
return 0, nil
|
||||
}
|
||||
@@ -291,8 +312,8 @@ func (rl *RedisRateLimiter) GetCount(ctx context.Context, identifier string) (in
|
||||
|
||||
// Reset 重置限流计数
|
||||
func (rl *RedisRateLimiter) Reset(ctx context.Context, identifier string) error {
|
||||
// M-C 修复:GetRedis() 取一次复用(原两次调用存在 nil-deref 窗口)。
|
||||
rdb := database.GetRedis()
|
||||
// M-C 修复(Phase 5:改 redisClient):取一次复用(原两次调用存在 nil-deref 窗口)。
|
||||
rdb := rl.redisClient()
|
||||
if rdb == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -301,138 +322,237 @@ func (rl *RedisRateLimiter) Reset(ctx context.Context, identifier string) error
|
||||
return rdb.Del(ctx, key).Err()
|
||||
}
|
||||
|
||||
// ===== 全局限速器 =====
|
||||
// ===== 全局限速器 Registry(Phase 5 实例化) =====
|
||||
//
|
||||
// 内存限流器(RateLimiter)持 cleanup goroutine,进程级共享会在多 App 下互相污染
|
||||
// (App A Shutdown 经 StopRateLimiters 把 App B 的 loginLimiter 也停了)。故照
|
||||
// cache.CacheManager / cron.Scheduler 模式引入 RateLimitRegistry:实例化 + 全局默认
|
||||
// + 包级 facade 代理,App 持自己的 Registry。
|
||||
//
|
||||
// 包级 LoginRateLimit() 等改为"请求时解析当前默认 Registry"懒创建限流器,使 App.Init
|
||||
// swap 后请求落到 App 的 Registry(避免 pre-Init 注册到 init Registry 被 swap 丢弃)。
|
||||
|
||||
var (
|
||||
// RateLimitRegistry 持一组内存限流器,支持 per-App 隔离。
|
||||
type RateLimitRegistry struct {
|
||||
mu sync.Mutex
|
||||
loginLimiter *RateLimiter
|
||||
apiLimiter *RateLimiter
|
||||
uploadLimiter *RateLimiter
|
||||
customLimiters []*RateLimiter // H4b: CustomRateLimit 创建的限流器登记表,供 StopRateLimiters/InitRateLimiters 统一停止,避免 cleanup goroutine 泄漏
|
||||
limitersMu sync.Mutex
|
||||
)
|
||||
|
||||
// drainCustomLimiters 取出已登记的自定义限流器并清空登记表。调用方须持有 limitersMu。
|
||||
func drainCustomLimiters() []*RateLimiter {
|
||||
drained := customLimiters
|
||||
customLimiters = nil
|
||||
return drained
|
||||
customLimiters []*RateLimiter
|
||||
}
|
||||
|
||||
// InitRateLimiters 初始化限速器
|
||||
func InitRateLimiters() {
|
||||
limitersMu.Lock()
|
||||
defer limitersMu.Unlock()
|
||||
// NewRateLimitRegistry 创建限流器注册表实例。
|
||||
func NewRateLimitRegistry() *RateLimitRegistry {
|
||||
return &RateLimitRegistry{}
|
||||
}
|
||||
|
||||
// 先停止旧的限流器(含自定义),释放 cleanup goroutine。
|
||||
// 持锁期间调 Stop() 安全:Stop→wg.Wait 等待的 cleanupVisitors 取的是 limiter 自身的 rl.mu,
|
||||
// 非 limitersMu,无死锁;全程持锁避免与 LoginRateLimit 等懒初始化路径交错致覆盖泄漏。
|
||||
if loginLimiter != nil {
|
||||
loginLimiter.Stop()
|
||||
func (r *RateLimitRegistry) loginLimiterInstance() *RateLimiter {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.loginLimiter == nil {
|
||||
r.loginLimiter = NewRateLimiter(10, time.Minute)
|
||||
}
|
||||
if apiLimiter != nil {
|
||||
apiLimiter.Stop()
|
||||
return r.loginLimiter
|
||||
}
|
||||
|
||||
func (r *RateLimitRegistry) apiLimiterInstance() *RateLimiter {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.apiLimiter == nil {
|
||||
r.apiLimiter = NewRateLimiter(100, time.Minute)
|
||||
}
|
||||
if uploadLimiter != nil {
|
||||
uploadLimiter.Stop()
|
||||
return r.apiLimiter
|
||||
}
|
||||
|
||||
func (r *RateLimitRegistry) uploadLimiterInstance() *RateLimiter {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.uploadLimiter == nil {
|
||||
r.uploadLimiter = NewRateLimiter(20, time.Minute)
|
||||
}
|
||||
for _, l := range drainCustomLimiters() {
|
||||
return r.uploadLimiter
|
||||
}
|
||||
|
||||
// registerCustom 登记自定义限流器供 Stop 统一停止(H4b)。
|
||||
func (r *RateLimitRegistry) registerCustom(l *RateLimiter) {
|
||||
r.mu.Lock()
|
||||
r.customLimiters = append(r.customLimiters, l)
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
// Init 预初始化标准限流器(可选;不调则首次请求时懒创建)。先停旧的同名限流器释放 goroutine。
|
||||
func (r *RateLimitRegistry) Init() {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.loginLimiter != nil {
|
||||
r.loginLimiter.Stop()
|
||||
}
|
||||
if r.apiLimiter != nil {
|
||||
r.apiLimiter.Stop()
|
||||
}
|
||||
if r.uploadLimiter != nil {
|
||||
r.uploadLimiter.Stop()
|
||||
}
|
||||
r.loginLimiter = NewRateLimiter(10, time.Minute)
|
||||
r.apiLimiter = NewRateLimiter(100, time.Minute)
|
||||
r.uploadLimiter = NewRateLimiter(20, time.Minute)
|
||||
}
|
||||
|
||||
// Stop 停止注册表中所有限流器(释放 cleanup goroutine)。幂等。
|
||||
func (r *RateLimitRegistry) Stop() {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.loginLimiter != nil {
|
||||
r.loginLimiter.Stop()
|
||||
r.loginLimiter = nil
|
||||
}
|
||||
if r.apiLimiter != nil {
|
||||
r.apiLimiter.Stop()
|
||||
r.apiLimiter = nil
|
||||
}
|
||||
if r.uploadLimiter != nil {
|
||||
r.uploadLimiter.Stop()
|
||||
r.uploadLimiter = nil
|
||||
}
|
||||
for _, l := range r.customLimiters {
|
||||
l.Stop()
|
||||
}
|
||||
|
||||
// 内存限流器(单实例)
|
||||
loginLimiter = NewRateLimiter(10, time.Minute)
|
||||
apiLimiter = NewRateLimiter(100, time.Minute)
|
||||
uploadLimiter = NewRateLimiter(20, time.Minute)
|
||||
r.customLimiters = nil
|
||||
}
|
||||
|
||||
// StopRateLimiters 停止所有限速器(应用关闭时调用)
|
||||
func StopRateLimiters() {
|
||||
limitersMu.Lock()
|
||||
defer limitersMu.Unlock()
|
||||
// --- App-bound 中间件(捕获此 Registry,不查全局默认,多 App per-App 隔离) ---
|
||||
//
|
||||
// 与包级 LoginRateLimit() 等的区别:包级在请求时查 GetDefaultRateLimitRegistry()(多 App 下
|
||||
// 仅最后 Init 的 App 是全局默认,故包级 facade 不提供 per-App 计数隔离);本组方法捕获 r,
|
||||
// 始终用此 Registry 的 limiter。多 App 场景用 app.RateLimitRegistry().LoginRateLimit() 装配路由。
|
||||
|
||||
if loginLimiter != nil {
|
||||
loginLimiter.Stop()
|
||||
loginLimiter = nil
|
||||
}
|
||||
if apiLimiter != nil {
|
||||
apiLimiter.Stop()
|
||||
apiLimiter = nil
|
||||
}
|
||||
if uploadLimiter != nil {
|
||||
uploadLimiter.Stop()
|
||||
uploadLimiter = nil
|
||||
}
|
||||
for _, l := range drainCustomLimiters() {
|
||||
l.Stop()
|
||||
// LoginRateLimit 返回绑定到此 Registry 的登录限流中间件。
|
||||
func (r *RateLimitRegistry) LoginRateLimit() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
rateLimitAllow(c, r.loginLimiterInstance())
|
||||
}
|
||||
}
|
||||
|
||||
// RateLimit 通用速率限制中间件(内存版)
|
||||
// APIRateLimit 返回绑定到此 Registry 的普通 API 限流中间件。
|
||||
func (r *RateLimitRegistry) APIRateLimit() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
rateLimitAllow(c, r.apiLimiterInstance())
|
||||
}
|
||||
}
|
||||
|
||||
// UploadRateLimit 返回绑定到此 Registry 的上传限流中间件。
|
||||
func (r *RateLimitRegistry) UploadRateLimit() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
rateLimitAllow(c, r.uploadLimiterInstance())
|
||||
}
|
||||
}
|
||||
|
||||
// CustomRateLimit 返回绑定到此 Registry 的自定义限流中间件。limiter 在首次请求时创建并
|
||||
// 登记到此 Registry(r.registerCustom),由 App.Shutdown 的 registry.Stop() 收口,无泄漏。
|
||||
func (r *RateLimitRegistry) CustomRateLimit(rate int, window time.Duration) gin.HandlerFunc {
|
||||
var (
|
||||
once sync.Once
|
||||
limiter *RateLimiter
|
||||
)
|
||||
return func(c *gin.Context) {
|
||||
once.Do(func() {
|
||||
limiter = NewRateLimiter(rate, window)
|
||||
r.registerCustom(limiter)
|
||||
})
|
||||
rateLimitAllow(c, limiter)
|
||||
}
|
||||
}
|
||||
|
||||
// defaultRateLimitRegistry 全局默认 Registry(atomic,照 cache/cron 模式)。
|
||||
var defaultRateLimitRegistry atomic.Pointer[RateLimitRegistry]
|
||||
|
||||
func init() {
|
||||
defaultRateLimitRegistry.Store(NewRateLimitRegistry())
|
||||
}
|
||||
|
||||
// GetDefaultRateLimitRegistry 返回全局默认 Registry。
|
||||
func GetDefaultRateLimitRegistry() *RateLimitRegistry {
|
||||
return defaultRateLimitRegistry.Load()
|
||||
}
|
||||
|
||||
// SwapDefaultRateLimitRegistry 置为全局默认,返回被替换的旧(照 Swap 模式)。nil 忽略,返回当前默认。
|
||||
func SwapDefaultRateLimitRegistry(r *RateLimitRegistry) *RateLimitRegistry {
|
||||
if r == nil {
|
||||
return defaultRateLimitRegistry.Load()
|
||||
}
|
||||
return defaultRateLimitRegistry.Swap(r)
|
||||
}
|
||||
|
||||
// rateLimitAllow 公共放行/拒绝逻辑(内存版)。
|
||||
func rateLimitAllow(c *gin.Context, rl *RateLimiter) {
|
||||
if rl == nil {
|
||||
response.Custom(c, http.StatusServiceUnavailable, response.CodeServiceUnavailable,
|
||||
"限流器未初始化", nil)
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
if !rl.Allow(c.ClientIP()) {
|
||||
response.RateLimit(c)
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
c.Next()
|
||||
}
|
||||
|
||||
// RateLimit 通用速率限制中间件(内存版,用户自持 limiter 时使用)。
|
||||
func RateLimit(limiter *RateLimiter) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
if limiter == nil {
|
||||
response.Custom(c, http.StatusServiceUnavailable, response.CodeServiceUnavailable,
|
||||
"限流器未初始化", nil)
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
ip := c.ClientIP()
|
||||
if !limiter.Allow(ip) {
|
||||
response.RateLimit(c)
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
c.Next()
|
||||
rateLimitAllow(c, limiter)
|
||||
}
|
||||
}
|
||||
|
||||
// LoginRateLimit 登录接口速率限制
|
||||
// LoginRateLimit 登录接口速率限制。限流器在首次请求时从当前默认 Registry 懒创建,
|
||||
// 故 App.Init swap 后用的是 App 的 Registry(避免 pre-Init 注册到 init Registry 被丢弃)。
|
||||
func LoginRateLimit() gin.HandlerFunc {
|
||||
limitersMu.Lock()
|
||||
if loginLimiter == nil {
|
||||
loginLimiter = NewRateLimiter(10, time.Minute)
|
||||
return func(c *gin.Context) {
|
||||
rateLimitAllow(c, GetDefaultRateLimitRegistry().loginLimiterInstance())
|
||||
}
|
||||
limiter := loginLimiter
|
||||
limitersMu.Unlock()
|
||||
|
||||
return RateLimit(limiter)
|
||||
}
|
||||
|
||||
// APIRateLimit 普通 API 速率限制
|
||||
// APIRateLimit 普通 API 速率限制。
|
||||
func APIRateLimit() gin.HandlerFunc {
|
||||
limitersMu.Lock()
|
||||
if apiLimiter == nil {
|
||||
apiLimiter = NewRateLimiter(100, time.Minute)
|
||||
return func(c *gin.Context) {
|
||||
rateLimitAllow(c, GetDefaultRateLimitRegistry().apiLimiterInstance())
|
||||
}
|
||||
limiter := apiLimiter
|
||||
limitersMu.Unlock()
|
||||
|
||||
return RateLimit(limiter)
|
||||
}
|
||||
|
||||
// UploadRateLimit 上传接口速率限制
|
||||
// UploadRateLimit 上传接口速率限制。
|
||||
func UploadRateLimit() gin.HandlerFunc {
|
||||
limitersMu.Lock()
|
||||
if uploadLimiter == nil {
|
||||
uploadLimiter = NewRateLimiter(20, time.Minute)
|
||||
return func(c *gin.Context) {
|
||||
rateLimitAllow(c, GetDefaultRateLimitRegistry().uploadLimiterInstance())
|
||||
}
|
||||
limiter := uploadLimiter
|
||||
limitersMu.Unlock()
|
||||
|
||||
return RateLimit(limiter)
|
||||
}
|
||||
|
||||
// CustomRateLimit 自定义速率限制(内存版)
|
||||
//
|
||||
// 创建的限流器登记入包级 customLimiters 表,StopRateLimiters / InitRateLimiters
|
||||
// 会统一停止其 cleanup goroutine(H4b 修复:原实现每次路由构造创建 limiter 无句柄,
|
||||
// StopRateLimiters 不感知 → cleanup goroutine 泄漏)。
|
||||
// CustomRateLimit 自定义速率限制(内存版)。限流器在首次请求时创建并登记到当前默认
|
||||
// Registry(供 Stop 统一停止,避免 cleanup goroutine 泄漏,H4b)。
|
||||
func CustomRateLimit(rate int, window time.Duration) gin.HandlerFunc {
|
||||
limiter := NewRateLimiter(rate, window)
|
||||
limitersMu.Lock()
|
||||
customLimiters = append(customLimiters, limiter)
|
||||
limitersMu.Unlock()
|
||||
return RateLimit(limiter)
|
||||
var (
|
||||
once sync.Once
|
||||
limiter *RateLimiter
|
||||
)
|
||||
return func(c *gin.Context) {
|
||||
once.Do(func() {
|
||||
limiter = NewRateLimiter(rate, window)
|
||||
GetDefaultRateLimitRegistry().registerCustom(limiter)
|
||||
})
|
||||
rateLimitAllow(c, limiter)
|
||||
}
|
||||
}
|
||||
|
||||
// InitRateLimiters 初始化默认 Registry 的标准限流器(可选,不调则懒创建)。
|
||||
func InitRateLimiters() {
|
||||
GetDefaultRateLimitRegistry().Init()
|
||||
}
|
||||
|
||||
// StopRateLimiters 停止默认 Registry 的所有限流器。注意:仅停默认 Registry--
|
||||
// App 持自己的 Registry 时应在 Shutdown 调 app 自己的 Stop(Phase 5)。
|
||||
func StopRateLimiters() {
|
||||
GetDefaultRateLimitRegistry().Stop()
|
||||
}
|
||||
|
||||
// ===== Redis 分布式限流中间件 =====
|
||||
@@ -440,12 +560,12 @@ func CustomRateLimit(rate int, window time.Duration) gin.HandlerFunc {
|
||||
// redisLimitDecision 处理 RedisRateLimiter.Allow 的结果,按 allowed 值决定放行/拒绝。
|
||||
//
|
||||
// H4c: 不再无条件 fail-open。Allow 已按 limiter 的 failClosed 策略把 Redis 故障翻成
|
||||
// allowed 值——fail-closed 时 allowed=false(拒绝),fail-open 时 allowed=true(放行)。
|
||||
// allowed 值--fail-closed 时 allowed=false(拒绝),fail-open 时 allowed=true(放行)。
|
||||
// 故 err 与 allowed 的组合语义:
|
||||
// - err==nil, allowed==true → 放行
|
||||
// - err==nil, allowed==false → 真超限,返 429(response.RateLimit)
|
||||
// - err!=nil, allowed==false → fail-closed 限流器在 Redis 故障下拒绝,返 503(服务不可用)
|
||||
// - err!=nil, allowed==true → fail-open 限流器在 Redis 故障下放行(兼容旧行为)
|
||||
// - err==nil, allowed==true -> 放行
|
||||
// - err==nil, allowed==false -> 真超限,返 429(response.RateLimit)
|
||||
// - err!=nil, allowed==false -> fail-closed 限流器在 Redis 故障下拒绝,返 503(服务不可用)
|
||||
// - err!=nil, allowed==true -> fail-open 限流器在 Redis 故障下放行(兼容旧行为)
|
||||
func redisLimitDecision(c *gin.Context, allowed bool, err error) {
|
||||
if err != nil && !allowed {
|
||||
// fail-closed:Redis 故障时拒绝(防限流静默失效)。返 503 区别于真实超限的 429。
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/EthanCodeCraft/xlgo-core/database"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// TestRedisRateLimiterInjectedClientPriority 固化 Phase 5(jwt 模型):
|
||||
// RedisRateLimiter.redisClient() 注入优先、全局兜底。
|
||||
func TestRedisRateLimiterInjectedClientPriority(t *testing.T) {
|
||||
mrG := miniredis.RunT(t)
|
||||
clientG := redis.NewClient(&redis.Options{Addr: mrG.Addr()})
|
||||
t.Cleanup(func() { _ = clientG.Close() })
|
||||
origGlobal := database.SetTestRedisClient(clientG)
|
||||
t.Cleanup(func() { database.SetTestRedisClient(origGlobal) })
|
||||
|
||||
mrA := miniredis.RunT(t)
|
||||
clientA := redis.NewClient(&redis.Options{Addr: mrA.Addr()})
|
||||
t.Cleanup(func() { _ = clientA.Close() })
|
||||
|
||||
// 注入优先
|
||||
rl := NewRedisRateLimiter("test", 10, time.Minute, WithRedisClient(clientA))
|
||||
if got := rl.redisClient(); got != clientA {
|
||||
t.Fatalf("injected redisClient() = %v, want clientA", got)
|
||||
}
|
||||
|
||||
// 未注入 -> 全局兜底
|
||||
rl2 := NewRedisRateLimiter("test", 10, time.Minute)
|
||||
if got := rl2.redisClient(); got != clientG {
|
||||
t.Fatalf("fallback redisClient() = %v, want clientG (global)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSwapDefaultRateLimitRegistryReturnsOld 固化 Phase 5:Swap 返回被替换的旧 Registry。
|
||||
func TestSwapDefaultRateLimitRegistryReturnsOld(t *testing.T) {
|
||||
orig := defaultRateLimitRegistry.Load()
|
||||
defer SwapDefaultRateLimitRegistry(orig)
|
||||
|
||||
first := NewRateLimitRegistry()
|
||||
if prev := SwapDefaultRateLimitRegistry(first); prev != orig {
|
||||
t.Fatalf("Swap returned %v, want orig", prev)
|
||||
}
|
||||
second := NewRateLimitRegistry()
|
||||
if returned := SwapDefaultRateLimitRegistry(second); returned != first {
|
||||
t.Fatalf("Swap returned %v, want first", returned)
|
||||
}
|
||||
if defaultRateLimitRegistry.Load() != second {
|
||||
t.Fatal("Swap did not install second as default")
|
||||
}
|
||||
if got := SwapDefaultRateLimitRegistry(nil); got != second {
|
||||
t.Fatalf("Swap(nil) = %v, want second (current default)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRateLimitRegistryStopIdempotent 固化 Phase 5:Stop 幂等,重复调用不 panic。
|
||||
func TestRateLimitRegistryStopIdempotent(t *testing.T) {
|
||||
r := NewRateLimitRegistry()
|
||||
_ = r.loginLimiterInstance()
|
||||
_ = r.apiLimiterInstance()
|
||||
_ = r.uploadLimiterInstance()
|
||||
r.registerCustom(NewRateLimiter(5, time.Minute))
|
||||
r.Stop()
|
||||
r.Stop() // 幂等
|
||||
}
|
||||
|
||||
// TestRateLimitRegistryIsolation 固化 Phase 5:两个 Registry 实例的 limiter 互相独立。
|
||||
func TestRateLimitRegistryIsolation(t *testing.T) {
|
||||
rA := NewRateLimitRegistry()
|
||||
rB := NewRateLimitRegistry()
|
||||
defer rA.Stop()
|
||||
defer rB.Stop()
|
||||
|
||||
la := rA.loginLimiterInstance()
|
||||
lb := rB.loginLimiterInstance()
|
||||
if la == lb {
|
||||
t.Fatal("two registries returned the same loginLimiter instance (not isolated)")
|
||||
}
|
||||
|
||||
// rA 用尽 10 次;rB 不受影响
|
||||
for i := 0; i < 10; i++ {
|
||||
if !la.Allow("1.1.1.1") {
|
||||
t.Fatalf("rA request %d should be allowed", i+1)
|
||||
}
|
||||
}
|
||||
// rA 第 11 次应被拒绝(count=10 >= rate=10)
|
||||
if la.Allow("1.1.1.1") {
|
||||
t.Fatal("rA 11th should be rejected (count reached limit)")
|
||||
}
|
||||
// rB 的 loginLimiter 独立计数,1.1.1.1 仍可放行
|
||||
if !lb.Allow("1.1.1.1") {
|
||||
t.Fatal("rB loginLimiter should be independent of rA (still allowed)")
|
||||
}}
|
||||
@@ -725,6 +725,16 @@ func SetDefaultStorageManager(m *StorageManager) {
|
||||
}
|
||||
}
|
||||
|
||||
// SwapDefaultStorageManager 将指定 StorageManager 置为全局默认,并返回被替换的旧 Manager。
|
||||
// 旧 Manager 不会被关闭,供 App 初始化这类需要失败回滚的生命周期流程暂存(照
|
||||
// SwapDefaultRedisManager / database.SwapDefaultManager 模式)。nil 被忽略,返回当前默认。
|
||||
func SwapDefaultStorageManager(m *StorageManager) *StorageManager {
|
||||
if m == nil {
|
||||
return DefaultStorage.Load()
|
||||
}
|
||||
return DefaultStorage.Swap(m)
|
||||
}
|
||||
|
||||
// Init 初始化存储
|
||||
func (m *StorageManager) Init(cfg *config.StorageConfig) error {
|
||||
if cfg == nil {
|
||||
@@ -768,6 +778,24 @@ func (m *StorageManager) Set(s Storage) {
|
||||
m.current = s
|
||||
}
|
||||
|
||||
// Close 释放存储资源,闭合 StorageManager 的 Init/Close 生命周期(与 database/redis/logger
|
||||
// manager 对齐,供 App.closeResources 统一调用)。当前 LocalStorage 无可关资源;OSSStorage 的
|
||||
// *oss.Client 未暴露 Close(aliyun-oss-go-sdk),故当前为 no-op。保留此方法为未来驱动
|
||||
// (S3/MinIO/自研等带连接池的驱动)预留收口点:实现了 io.Closer 的 Storage 实现会被自动调用,
|
||||
// 框架骨架无需改动。幂等(未初始化或驱动非 io.Closer 时 no-op)。
|
||||
func (m *StorageManager) Close() error {
|
||||
m.mu.Lock()
|
||||
current := m.current
|
||||
m.mu.Unlock()
|
||||
if current == nil {
|
||||
return nil
|
||||
}
|
||||
if c, ok := current.(io.Closer); ok {
|
||||
return c.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// --- 包级 facade(代理到 DefaultStorage,兼容存量) ---
|
||||
|
||||
// Init 初始化存储
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"mime/multipart"
|
||||
"testing"
|
||||
|
||||
"github.com/EthanCodeCraft/xlgo-core/config"
|
||||
)
|
||||
|
||||
// closeableMock 实现 Storage + io.Closer,验证 StorageManager.Close 调用驱动的 Close。
|
||||
type closeableMock struct {
|
||||
closed bool
|
||||
}
|
||||
|
||||
func (closeableMock) Upload(*multipart.FileHeader, string) (string, error) { return "", nil }
|
||||
func (closeableMock) UploadFromBytes([]byte, string, string) (string, error) {
|
||||
return "", nil
|
||||
}
|
||||
func (closeableMock) GetURL(string) string { return "" }
|
||||
func (closeableMock) Delete(string) error { return nil }
|
||||
func (closeableMock) Get(string) ([]byte, error) { return nil, nil }
|
||||
func (closeableMock) Exists(string) (bool, error) { return false, nil }
|
||||
func (c *closeableMock) Close() error { c.closed = true; return nil }
|
||||
|
||||
// TestStorageManagerCloseNil:未初始化时 Close 为 no-op,不 panic。
|
||||
func TestStorageManagerCloseNil(t *testing.T) {
|
||||
m := NewStorageManager()
|
||||
if err := m.Close(); err != nil {
|
||||
t.Fatalf("Close on uninitialized manager should be no-op, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStorageManagerCloseLocalStorageNoop:LocalStorage 未实现 io.Closer,Close 为 no-op。
|
||||
func TestStorageManagerCloseLocalStorageNoop(t *testing.T) {
|
||||
m := NewStorageManager()
|
||||
if err := m.Init(&config.StorageConfig{
|
||||
Driver: "local",
|
||||
Local: config.LocalStorageConfig{Path: t.TempDir()},
|
||||
}); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
if err := m.Close(); err != nil {
|
||||
t.Fatalf("Close on LocalStorage-backed manager should be no-op, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStorageManagerCloseInvokesCloser:驱动实现 io.Closer 时,StorageManager.Close 调用它。
|
||||
func TestStorageManagerCloseInvokesCloser(t *testing.T) {
|
||||
m := NewStorageManager()
|
||||
mock := &closeableMock{}
|
||||
m.Set(mock)
|
||||
if err := m.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
if !mock.closed {
|
||||
t.Fatal("StorageManager.Close did not invoke io.Closer on driver (未来带连接池的驱动应经此收口)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStorageManagerCloseIdempotent:重复 Close 不 panic。
|
||||
func TestStorageManagerCloseIdempotent(t *testing.T) {
|
||||
m := NewStorageManager()
|
||||
mock := &closeableMock{}
|
||||
m.Set(mock)
|
||||
if err := m.Close(); err != nil {
|
||||
t.Fatalf("first Close: %v", err)
|
||||
}
|
||||
if err := m.Close(); err != nil {
|
||||
t.Fatalf("second Close should be idempotent, got %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
package storage
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestSwapDefaultStorageManagerPreservesReplacedManager 固化 Phase 2:
|
||||
// SwapDefaultStorageManager 必须返回被替换的旧 Manager 且不关闭它(供 App Init 失败回滚暂存)。
|
||||
// 照 database.SwapDefaultRedisManager 的内部测试模式。
|
||||
func TestSwapDefaultStorageManagerPreservesReplacedManager(t *testing.T) {
|
||||
orig := DefaultStorage.Load()
|
||||
|
||||
old := NewStorageManager()
|
||||
previous := SwapDefaultStorageManager(old)
|
||||
if previous != orig {
|
||||
t.Fatalf("SwapDefaultStorageManager 应返回被替换的旧默认(orig),got %v", previous)
|
||||
}
|
||||
if DefaultStorage.Load() != old {
|
||||
t.Fatal("SwapDefaultStorageManager 应安装新的默认 manager(old)")
|
||||
}
|
||||
|
||||
next := NewStorageManager()
|
||||
returned := SwapDefaultStorageManager(next)
|
||||
if returned != old {
|
||||
t.Fatalf("SwapDefaultStorageManager 应返回上一次安装的 manager(old),got %v", returned)
|
||||
}
|
||||
if DefaultStorage.Load() != next {
|
||||
t.Fatal("SwapDefaultStorageManager 应安装新的默认 manager(next)")
|
||||
}
|
||||
|
||||
// nil 被忽略,返回当前默认
|
||||
if got := SwapDefaultStorageManager(nil); got != next {
|
||||
t.Fatalf("SwapDefaultStorageManager(nil) 应返回当前默认,got %v", got)
|
||||
}
|
||||
|
||||
// 恢复 init() 时的默认,避免污染其它测试
|
||||
SwapDefaultStorageManager(orig)
|
||||
}
|
||||
Reference in New Issue
Block a user