feat: release v0.27 unified HTTP limits and safe request shutdown
This commit is contained in:
@@ -1,5 +1,13 @@
|
||||
# Changelog
|
||||
|
||||
## v0.27.0-preview(2026-09-29)
|
||||
|
||||
- HTTP 配置新增 defaultMaxBodyBytes、maxBodyBytes、shutdownCancelTimeout;三种内置服务统一默认/路由/硬请求体限额,配置冲突在监听前报告完整路由与限额。native/HTTP2 不再以隐含 8 MiB 阻挡已放宽的 App 上传路由。
|
||||
- App 在所有服务入口跟踪完整请求处理链;关闭连接后有界等待处理器实际退出。超时仍有请求则返回 ErrResourcesRetained,跳过 App 清理;Runtime.Finish 继续停止后台工作但保留依赖,生成入口保留日志。
|
||||
- native 提前拒绝上传后有界半关闭连接,减少 TCP reset 吞掉错误响应;业务 HTTP 200 策略不再触发对被拒绝请求体的继续读取。
|
||||
- HTTP/2 Shutdown 同时等待连接和独立处理器,修复客户端断开后误判完成。增加真实 TLS/流式上传、取消收尾、拒绝晚到请求、资源保留和生成消费者回归,以及 httpSafety 示例。
|
||||
- 迁移:nethttp 手写普通路由新增默认 8 MiB 请求体上限,大请求显式覆盖;生成契约默认 1 MiB 保持。HTTP 停机多一个默认 5s 取消收尾窗口;外部宿主须将运行错误传给 Runtime.Finish,并遵守资源保留信号。
|
||||
|
||||
## v0.26.0-preview(2026-09-29)
|
||||
|
||||
- 新增独立 i18n:不可变命名空间词典、源/默认/启用语言、BCP 47 匹配、显式回退、参数/条件/基数复数和有界资源加载;不依赖 HTTP、业务或数据库。
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# 通用业务开发指南(v0.26.0-preview)
|
||||
# 通用业务开发指南(v0.27.0-preview)
|
||||
|
||||
从 [README](README.md) 创建服务后,业务主要维护 `.api`、配置、`svc.Init`、Logic、Repository 和项目中间件。本页说明通用接入能力;字段、错误码、权限、表结构和事务边界都由项目定义。
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# 生成项目与文件维护边界
|
||||
|
||||
v0.26.0-preview 提供受管理的生成项目。安装同版远程 CLI,生成的项目直接引用该版本,无需本地 replace。根目录入口支持 `go run .` 和 `go build .`;框架本身仍然是可直接引用的库。安装和私有仓库访问见 [README](README.md#准备环境与安装)。
|
||||
v0.27.0-preview 提供受管理的生成项目。安装同版远程 CLI,生成的项目直接引用该版本,无需本地 replace。根目录入口支持 `go run .` 和 `go build .`;框架本身仍然是可直接引用的库。安装和私有仓库访问见 [README](README.md#准备环境与安装)。
|
||||
|
||||
## 创建和日常开发
|
||||
|
||||
@@ -98,7 +98,7 @@ func (c *ServiceContext) AppOptions() []omni.Option {
|
||||
|
||||
最终配置校验包含自定义响应编码器与错误目录的组合。码类型/映射冲突、编码器已绑定另一份目录等问题在创建 App 前返回错误,进入入口的清理流程,不等到构造 App 时才 panic。
|
||||
|
||||
日志在 AppOptions 前初始化,其配置应放在 Config 中。退出信号先触发 HTTP 排空;达到 shutdownTimeout 后按框架规则强制关闭连接。请求仍依赖的后台任务/客户端不能直接绑定信号 context 提前关闭,应持有独立生命周期,使用 WorkContext 并登记成对的 OnStop/OnDrain,详见 [分阶段关闭](lifecycle/README.md#后台组件分阶段关闭)。业务必须响应请求 context 取消;忽略取消的 goroutine 无法保证在超时前结束。
|
||||
日志在 AppOptions 前初始化,其配置应放在 Config 中。退出信号先触发 HTTP 排空;达到 shutdownTimeout 后取消请求并关闭连接,再按 `http.shutdownCancelTimeout` 等待处理函数收尾。仍未结束时保留依赖并返回 `ErrResourcesRetained`,不会提前关闭数据库。请求仍依赖的后台任务/客户端不能直接绑定信号 context 提前关闭,应持有独立生命周期,使用 WorkContext 并登记成对的 OnStop/OnDrain,详见 [分阶段关闭](lifecycle/README.md#后台组件分阶段关闭)。业务必须响应请求 context 取消;忽略取消的 goroutine 无法保证在超时前结束。
|
||||
|
||||
## 依赖主动申请退出
|
||||
|
||||
@@ -109,7 +109,7 @@ func (c *ServiceContext) AppOptions() []omni.Option {
|
||||
c.RequestShutdown(err)
|
||||
```
|
||||
|
||||
这只是通知,立即返回,不等待 HTTP 排空,也不直接关闭数据库。生成入口停止接收新请求,等待在途请求完成;超过 `shutdownTimeout` 后取消请求并关闭连接,再执行 svc.Close,最后关闭日志。非空故障原因保留在返回错误链中,进程退出码为 1;`RequestShutdown(nil)` 或正常 SIGINT/SIGTERM 在没有其他故障时退出码为 0。
|
||||
这只是通知,立即返回,不等待 HTTP 排空,也不直接关闭数据库。生成入口停止接收新请求,等待在途请求完成;超过 `shutdownTimeout` 后取消请求并关闭连接,再等待 HTTP 取消收尾窗口。通过 `svc.Finish(runErr)` 执行后台停机;确认工作结束才关闭依赖与日志,无法确认时保留资源并非零退出。非空故障原因保留在返回错误链中,进程退出码为 1;`RequestShutdown(nil)` 或正常 SIGINT/SIGTERM 在没有其他故障时退出码为 0。
|
||||
|
||||
多个依赖可并发通知,关闭流程仅执行一次,首个非空故障原因保留。正常信号已开始排空时收到的故障也不会被清理动作覆盖。初始化期间的通知同样有效,不依赖 HTTP 监听已经启动。后台监控任务应通过 OnStop/OnDrain 停止并等待,再进入资源清理;关闭完成以后才报告的故障无法参与已经结束的进程结果。
|
||||
|
||||
@@ -215,7 +215,7 @@ get /download (DownloadReq)
|
||||
|
||||
## 已有服务接入步骤
|
||||
|
||||
每个独立服务生成一个项目,module 名称、上层目录和共享业务包由项目选择。接入时固定 CLI 与框架版本为 v0.26.0-preview:
|
||||
每个独立服务生成一个项目,module 名称、上层目录和共享业务包由项目选择。接入时固定 CLI 与框架版本为 v0.27.0-preview:
|
||||
|
||||
1. 先保留旧服务,整理实际路由及现有响应、鉴权和超时要求。在新目录执行 `go-omni new --api /path/to/service.api --module example.com/service ./service`;不对旧入口强制覆盖。
|
||||
2. 在 `.api` 中选择 `@app(defaultResponse: ...)`,显式保留原项目的字段、成功码、错误码和 `dataMode`。需要独立回执的路由单独声明 raw/produces。普通路由、404/405 和提前返回的中间件一起验收。
|
||||
|
||||
21
README.md
21
README.md
@@ -2,9 +2,11 @@
|
||||
|
||||
GoOmni 是面向多个业务项目复用的 Go 后端框架:用 `.api` 定义接口、请求和返回类型,生成项目目录、HTTP 接入代码及 OpenAPI,业务代码负责 Logic 和 Repository。也可以在已有项目中直接使用 `omni.New()`,或单独引入日志、存储等模块。
|
||||
|
||||
**v0.26 改动**:通用 [i18n](i18n/README.md),提供独立词典、语言匹配、回退、命名参数/条件/复数、可选 HTTP 与错误适配、检查/导出 CLI 及前端运行时。文案由项目维护;[完整示例](examples/i18n/README.md)可直接运行。CLI 与框架使用同版 v0.26.0-preview。同时修复 Go 1.22 下请求体限额错误的识别。
|
||||
**v0.27 改动**:统一三种内置服务实现的默认/路由/硬请求体限额,配置冲突在启动前报错;停机以处理函数实际退出为资源清理前提,超时仍有请求时保留共享依赖。见 [HTTP 策略](transport/README.md#统一请求体限额)、[停机规则](lifecycle/README.md#http-请求与依赖的关闭屏障)和[可运行示例](examples/httpSafety/README.md)。
|
||||
|
||||
**从这份 README 开始接入。** 当前版本 **`v0.26.0-preview`**,新生成项目采用根目录入口和 `config / svc / logic / handler`。可以固定版本开展业务接入;文件维护边界和已有服务迁移步骤见 [项目指南](PROJECT.md),版本变化与验收见 [发布说明](RELEASE.md)。
|
||||
**v0.26 改动**:通用 [i18n](i18n/README.md),提供独立词典、语言匹配、回退、命名参数/条件/复数、可选 HTTP 与错误适配、检查/导出 CLI 及前端运行时。文案由项目维护;[完整示例](examples/i18n/README.md)可直接运行。CLI 与框架使用同版 v0.27.0-preview。同时修复 Go 1.22 下请求体限额错误的识别。
|
||||
|
||||
**从这份 README 开始接入。** 当前版本 **`v0.27.0-preview`**,新生成项目采用根目录入口和 `config / svc / logic / handler`。可以固定版本开展业务接入;文件维护边界和已有服务迁移步骤见 [项目指南](PROJECT.md),版本变化与验收见 [发布说明](RELEASE.md)。
|
||||
|
||||
**v0.25 改动**:`database.InTx` 返回可识别的事务阶段与结果,保留业务/驱动及回滚错误链;提交报错保守标记为结果不确定,不等同于已回滚。[使用规则](database/README.md#事务阶段与结果)、[可运行故障示例](examples/transactions/README.md)。
|
||||
|
||||
@@ -156,7 +158,7 @@ module 地址是 `git.yuntuo.hk/k2safe/go-omni`。先确保本机 Go 和 Git 可
|
||||
|
||||
```sh
|
||||
export GOPRIVATE="$(go env GOPRIVATE),git.yuntuo.hk"
|
||||
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.26.0-preview
|
||||
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.27.0-preview
|
||||
export PATH="$(go env GOPATH)/bin:$PATH"
|
||||
go-omni --help
|
||||
```
|
||||
@@ -313,7 +315,7 @@ go-omni new --api catalog.api \
|
||||
在业务项目的 **go.mod 所在目录**执行:
|
||||
|
||||
```sh
|
||||
go get git.yuntuo.hk/k2safe/go-omni@v0.26.0-preview
|
||||
go get git.yuntuo.hk/k2safe/go-omni@v0.27.0-preview
|
||||
```
|
||||
|
||||
已有项目可直接复用现有目录、Service、Repository 和数据库连接,先选择一个接口接入。也可以只用 `logx`、`storage` 等包,暂时保留原来的 HTTP 框架。希望入口和 HTTP 接入都交给生成器的项目,按 [已有服务接入步骤](PROJECT.md#已有服务接入步骤) 在新目录生成后迁入业务代码。
|
||||
@@ -589,6 +591,9 @@ API 项目的数据库迁移需要单独维护并显式执行;启动不会自
|
||||
| `server.http.readHeaderTimeout/readTimeout` | 请求头 / 请求读取超时 |
|
||||
| `server.http.writeTimeout/idleTimeout` | 响应写入 / Keep-Alive 空闲超时 |
|
||||
| `server.http.maxHeaderBytes` | 请求头大小限制,字节数 |
|
||||
| `server.http.defaultMaxBodyBytes` | 普通 App 路由默认请求体上限,零默认 8 MiB;路由可覆盖 |
|
||||
| `server.http.maxBodyBytes` | 服务请求体硬上限,路由不得超过;零不添加低于框架 1 TiB 的硬上限 |
|
||||
| `server.http.shutdownCancelTimeout` | 请求取消后等待处理函数收尾,零默认 5s;未结束则保留资源 |
|
||||
| `server.shutdownTimeout` | 优雅退出等待时间,默认 10s |
|
||||
| `server.enableLogger` / `server.log.level` | 访问日志开关、日志级别 |
|
||||
| `server.trustedProxies` | 可信代理 IP/CIDR,文件数组 |
|
||||
@@ -616,13 +621,15 @@ server:
|
||||
maxHeaderBytes: 1048576
|
||||
```
|
||||
|
||||
代码配置使用 `omni.WithHTTPConfig(omni.HTTPConfig{ReadTimeout: 30*time.Second, WriteTimeout: 30*time.Second})`。`Run`、`RunContext`、`ServeContext` 会应用它;`NewServer` 会配置返回的标准 `http.Server`,包含可选的 Go 字段 `ErrorLog`。**正值覆盖、零值保留所选网络层的原有默认行为、负值报错**;不配置时升级行为不变。超时含义与支持范围见 [HTTP 配置](transport/README.md#http-配置v091-preview),不要把网络 I/O 超时当成业务执行时限。
|
||||
代码配置使用 `omni.WithHTTPConfig(omni.HTTPConfig{ReadTimeout: 30*time.Second, WriteTimeout: 30*time.Second})`。`Run`、`RunContext`、`ServeContext` 会应用它;`NewServer` 会配置返回的标准 `http.Server`,包含可选的 Go 字段 `ErrorLog`。**网络 I/O 参数正值覆盖、零值保留所选网络层的原有默认行为、负值报错**。请求体与取消收尾字段按上表解释;网络超时含义与支持范围见 [HTTP 配置](transport/README.md#http-配置v091-preview),不要把网络 I/O 超时当成业务执行时限。
|
||||
|
||||
外部宿主在注册完成后可用 `app.ValidateHTTP()` 预检配置与路由限额。上传路由使用 `app.HandleWithBodyLimit("POST", "/upload", 16<<20, nil, handler)` 或 `.api @request(maxBytes: 16777216)`;限额涵盖整个请求体。配置硬上限时至少允许该值;multipart 单文件、字段数以及幂等中间件缓存限制仍独立配置。新增普通路由默认 8 MiB;原 nethttp 手写大请求请显式设置默认值或路由上限,生成路由既有 `maxBytes`(默认 1 MiB)保持。
|
||||
|
||||
自研网络层通过 `RunWith(ctx, runner)` 接入监听与连接生命周期,通过 `NewWithRouter(router)` 替换路由;业务 Handler/Service 不需要导入具体适配器。细节与验收契约见 [网络适配层](transport/README.md)。
|
||||
|
||||
生成项目的依赖可调用 `svcCtx.RequestShutdown(err)` 主动申请退出,框架排空请求、清理依赖后保留故障原因并以非零状态退出。健康检查与 OpenAPI 的开关、路径通过 `server.endpoints` 配置,健康响应默认沿用项目公共体;未就绪保持 HTTP 503。完整配置和 v0.12 项目的升级步骤见 [项目指南](PROJECT.md#依赖主动申请退出)。
|
||||
|
||||
`Run` 处理 SIGINT/SIGTERM;`RunContext` 接收调用方控制的退出信号。服务先停止接收新请求,等待活跃请求完成,再清理资源;超时会取消请求并关闭连接,业务代码仍需响应 context 取消。
|
||||
`Run` 处理 SIGINT/SIGTERM;`RunContext` 接收调用方控制的退出信号。先停止接收新请求并等待排空,超时取消请求、关闭连接,再等待 `http.shutdownCancelTimeout` 收尾窗口。只有处理函数和它的 defer 全部退出才清理资源;仍有请求时返回 `lifecycle.ErrResourcesRetained`,跳过 App 清理钩子。生成入口通过 `deps.Finish(runErr)` 同步保留数据库等依赖及日志,非零退出。`App.Close`、`Engine`、`NewServer`、`RunWith` 共用这个完成屏障;业务自行启动的 goroutine、Hijack 后的工作需另行注册后台生命周期。
|
||||
|
||||
## 日常开发与升级
|
||||
|
||||
@@ -639,7 +646,7 @@ go test ./...
|
||||
|
||||
`main.go`、`Gen.go`、`docs/openapi.json` 和 manifest 由生成器管理,手动修改会报告冲突。Logic、Repository、中间件实现、配置和依赖装配属于项目,重复生成保留手写内容。契约变化会提示受影响的 Logic 文件和 DTO 字段,明确的签名错误会给出原签名与期望签名;生成后 `--check` 仍能检出未修复的问题。将 `.api`、`omni.gen.json`、生成文件和 `.goOmni/apiManifest.json` 一起提交,CI 同时运行 `generate --check` 与 `go test ./...`。详见 [契约变更提示与检查边界](PROJECT.md#契约变更提示)。
|
||||
|
||||
CLI 与业务 module 固定使用 `v0.26.0-preview`。v0.21 及更早项目先按 [配置迁移](CONFIG.md#v022-迁移旧模式已移除) 整理旧标签、文件与 Load,再升级依赖与 CLI 并重新生成,框架维护的 main/runtimeGen 自动更新;v0.13 及更早项目按 [升级预览](DEVELOPMENT.md#v013-项目升级) 处理,不直接覆盖手写入口或业务文件。使用过旧 mapping 的项目必须先完成 [移除迁移](COPY.md#从旧-mapping-迁移),旧配置或旧生成文件会阻止生成。升级后执行生成检查与业务回归。版本变化、迁移边界和验收记录见 [发布说明](RELEASE.md)。
|
||||
CLI 与业务 module 固定使用 `v0.27.0-preview`。v0.21 及更早项目先按 [配置迁移](CONFIG.md#v022-迁移旧模式已移除) 整理旧标签、文件与 Load,再升级依赖与 CLI 并重新生成,框架维护的 main/runtimeGen 自动更新;v0.13 及更早项目按 [升级预览](DEVELOPMENT.md#v013-项目升级) 处理,不直接覆盖手写入口或业务文件。使用过旧 mapping 的项目必须先完成 [移除迁移](COPY.md#从旧-mapping-迁移),旧配置或旧生成文件会阻止生成。升级后执行生成检查与业务回归。版本变化、迁移边界和验收记录见 [发布说明](RELEASE.md)。
|
||||
|
||||
## 按需深入
|
||||
|
||||
|
||||
19
RELEASE.md
19
RELEASE.md
@@ -1,3 +1,22 @@
|
||||
# v0.27.0-preview
|
||||
|
||||
新增统一请求体策略与请求完成屏障,无业务字段、数据库依赖或新第三方包。`server.http.defaultMaxBodyBytes` 控制普通路由默认值(零为 8 MiB),`maxBodyBytes` 为不可由路由突破的服务硬上限(零仅保留 1 TiB 框架上界)。已有显式路由和生成契约限额保持,冲突会在启动前报错;旧 nethttp 手写大请求需显式设置路由或默认上限。multipart 单文件、中间件缓存及代理限额仍各自配置。
|
||||
|
||||
HTTP 关闭连接后继续等待处理器实际退出,收尾窗口由 `server.http.shutdownCancelTimeout` 设置(零默认 5s)。超时仍有处理器运行时返回 `lifecycle.ErrResourcesRetained`,App/Runtime/生成入口保留共享依赖和日志;不再将连接关闭当成业务完成。`Runtime.Finish(runErr)` 会传播该信号,手写入口必须处理它,避免无条件关闭额外资源。该行为适用于全部 App 服务入口;HTTP/2 低层 Shutdown 同时等待独立处理器。
|
||||
|
||||
详见 [请求体策略](transport/README.md#统一请求体限额)、[停机规则](lifecycle/README.md#http-请求与依赖的关闭屏障)、[运行示例](examples/httpSafety/README.md)。CLI 与框架固定使用同版 v0.27.0-preview;模板最低兼容版本仍为 v0.22,HTTP 安全改进需要升级运行时。
|
||||
|
||||
```sh
|
||||
go get git.yuntuo.hk/k2safe/go-omni@v0.27.0-preview
|
||||
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.27.0-preview
|
||||
go-omni api generate --check
|
||||
go test ./...
|
||||
```
|
||||
|
||||
54 个测试包全量 race、Go 1.22 兼容检查、生成项目集成测试、vet/build 及真实进程关闭检查通过,见 [验收记录](reports/httpSafetyAcceptance.md)。
|
||||
|
||||
---
|
||||
|
||||
# v0.26.0-preview
|
||||
|
||||
新增可独立使用的 i18n 底层模块及可选 HTTP/错误适配。项目维护文案和语言设置,框架提供不可变词典、语言匹配、回退、参数与分支、检查和导出;没有绑定业务身份、页面框架、数据库或响应字段名。
|
||||
|
||||
69
app/app.go
69
app/app.go
@@ -33,6 +33,10 @@ type App struct {
|
||||
closeErr error
|
||||
clientIPResolver *httpx.ClientIPResolver
|
||||
responseEncoder response.Encoder
|
||||
requests requestDrain
|
||||
bodyLimits []routeBodyLimit
|
||||
validateOnce sync.Once
|
||||
validateErr error
|
||||
}
|
||||
|
||||
// Runner is the small lifecycle boundary used by an HTTP server
|
||||
@@ -100,7 +104,14 @@ func newApp(router httpx.Router, settings config.Config) *App {
|
||||
}
|
||||
middlewares = append(middlewares, middleware.Recovery())
|
||||
application := &App{router: router, config: settings, middlewares: middlewares, clientIPResolver: resolver, responseEncoder: responseEncoder}
|
||||
application.handler = corsPolicy.Wrap(router)
|
||||
application.requests.done = make(chan struct{})
|
||||
application.handler = corsPolicy.Wrap(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.ContentLength > application.hardBodyLimit() {
|
||||
application.adapt(func(*httpx.Context) error { return apierrors.New(413, "BODY_TOO_LARGE", "request body exceeds limit") }).ServeHTTP(w, r)
|
||||
return
|
||||
}
|
||||
router.ServeHTTP(w, r)
|
||||
}))
|
||||
if fallback, ok := router.(httpx.FallbackRouter); ok {
|
||||
adaptError := func(status int, code, message string) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -113,15 +124,31 @@ func newApp(router httpx.Router, settings config.Config) *App {
|
||||
}
|
||||
|
||||
// Router exposes the underlying routing contract for advanced registration.
|
||||
// Serve requests through App or Engine so application CORS policy is applied.
|
||||
// Serve through App or Engine for CORS, hard body limits and shutdown tracking.
|
||||
func (a *App) Router() httpx.Router { return a.router }
|
||||
|
||||
// Engine is kept as a compatibility name for the old API. It returns a
|
||||
// standard http.Handler, so callers do not depend on a router type.
|
||||
func (a *App) Engine() http.Handler { return a.handler }
|
||||
func (a *App) Engine() http.Handler { return a }
|
||||
|
||||
// ServeHTTP implements http.Handler for standard servers and HTTP tests.
|
||||
func (a *App) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
|
||||
ctx, work := a.requests.begin(request.Context())
|
||||
if work == nil {
|
||||
// Do not invoke encoders, middleware or loggers after their owners close.
|
||||
http.Error(writer, "server shutting down", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
defer a.requests.end(work)
|
||||
a.validateOnce.Do(func() { a.validateErr = a.ValidateHTTP() })
|
||||
if a.validateErr != nil {
|
||||
http.Error(writer, "invalid HTTP configuration", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
request = request.WithContext(ctx)
|
||||
if request.Body != nil {
|
||||
request.Body = http.MaxBytesReader(writer, request.Body, a.hardBodyLimit())
|
||||
}
|
||||
a.handler.ServeHTTP(writer, request)
|
||||
}
|
||||
|
||||
@@ -143,7 +170,8 @@ func (a *App) Use(middleware ...httpx.Middleware) *App {
|
||||
}
|
||||
|
||||
// OnShutdown registers a cleanup function owned by the application. Cleanup
|
||||
// functions run in reverse registration order when Close or RunContext ends.
|
||||
// functions run in reverse registration order after all handlers have returned.
|
||||
// Incomplete draining skips these callbacks and returns ErrResourcesRetained.
|
||||
// This keeps omni.New as the only public startup entry point while allowing
|
||||
// projects to attach databases, loggers and other resources to its lifecycle.
|
||||
func (a *App) OnShutdown(cleanup func() error) *App {
|
||||
@@ -153,9 +181,21 @@ func (a *App) OnShutdown(cleanup func() error) *App {
|
||||
return a
|
||||
}
|
||||
|
||||
// Close releases application resources. It is safe to call more than once.
|
||||
// Close stops admission, cancels active requests and waits for their handlers
|
||||
// to return before releasing resources. It is idempotent. If cancellation grace
|
||||
// expires, ErrResourcesRetained is permanent and no cleanup callbacks run.
|
||||
// Use RunContext/ServeContext for graceful draining before cancellation. Do not
|
||||
// call Close from a handler or cleanup callback; it would wait for itself.
|
||||
func (a *App) Close() error {
|
||||
a.closeOnce.Do(func() {
|
||||
timeout := a.config.HTTP.ShutdownCancelTimeout
|
||||
if timeout <= 0 {
|
||||
timeout = 5 * time.Second
|
||||
}
|
||||
if err := a.requests.cancelAndWait(timeout); err != nil {
|
||||
a.closeErr = err
|
||||
return
|
||||
}
|
||||
for index := len(a.closers) - 1; index >= 0; index-- {
|
||||
if err := a.closers[index](); err != nil {
|
||||
a.closeErr = errors.Join(a.closeErr, err)
|
||||
@@ -177,7 +217,7 @@ func (a *App) Run(address ...string) error {
|
||||
// Invalid HTTP options panic; RunContext/ServeContext return validation errors.
|
||||
// Use RunContext or ServeContext to apply the configured transport and TLS.
|
||||
func (a *App) NewServer(address string) *http.Server {
|
||||
if err := a.config.HTTP.Validate(); err != nil {
|
||||
if err := a.ValidateHTTP(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
if address == "" {
|
||||
@@ -187,7 +227,7 @@ func (a *App) NewServer(address string) *http.Server {
|
||||
}
|
||||
}
|
||||
cfg := a.config.HTTP
|
||||
return &http.Server{Addr: address, Handler: a.handler,
|
||||
return &http.Server{Addr: address, Handler: a,
|
||||
ReadHeaderTimeout: cfg.ReadHeaderTimeout, ReadTimeout: cfg.ReadTimeout,
|
||||
WriteTimeout: cfg.WriteTimeout, IdleTimeout: cfg.IdleTimeout,
|
||||
MaxHeaderBytes: cfg.MaxHeaderBytes, ErrorLog: cfg.ErrorLog}
|
||||
@@ -202,6 +242,8 @@ func (a *App) RunContext(ctx context.Context, address ...string) (runErr error)
|
||||
if ctx.Err() != nil {
|
||||
return a.Close()
|
||||
}
|
||||
stopWatching := a.watchShutdown(ctx)
|
||||
defer stopWatching()
|
||||
serve, err := a.prepareTransport()
|
||||
if err != nil {
|
||||
return errors.Join(err, a.Close())
|
||||
@@ -228,6 +270,11 @@ func (a *App) RunWith(ctx context.Context, runner Runner, address ...string) (ru
|
||||
if ctx.Err() != nil {
|
||||
return a.Close()
|
||||
}
|
||||
if err := a.ValidateHTTP(); err != nil {
|
||||
return errors.Join(err, a.Close())
|
||||
}
|
||||
stopWatching := a.watchShutdown(ctx)
|
||||
defer stopWatching()
|
||||
return errors.Join(runner.Run(ctx, a, a.listenAddress(address...)), a.Close())
|
||||
}
|
||||
|
||||
@@ -261,12 +308,16 @@ func (a *App) serveContext(ctx context.Context, server *http.Server, listener ne
|
||||
|
||||
select {
|
||||
case err := <-serveErr:
|
||||
// Serve returning only proves the accept loop ended. Cancel any remaining
|
||||
// connections before the application completion barrier and cleanup.
|
||||
_ = server.Close()
|
||||
closeErr := a.Close()
|
||||
if err == http.ErrServerClosed {
|
||||
return closeErr
|
||||
}
|
||||
return errors.Join(err, closeErr)
|
||||
case <-ctx.Done():
|
||||
a.requests.stop()
|
||||
shutdownTimeout := a.config.ShutdownTimeout
|
||||
if shutdownTimeout <= 0 {
|
||||
shutdownTimeout = 10 * time.Second
|
||||
@@ -275,8 +326,8 @@ func (a *App) serveContext(ctx context.Context, server *http.Server, listener ne
|
||||
defer cancel()
|
||||
shutdownErr := server.Shutdown(shutdownCtx)
|
||||
if shutdownErr != nil {
|
||||
// Shutdown leaves active connections open on timeout. Close them so
|
||||
// request contexts are canceled before releasing shared resources.
|
||||
// Closing sockets requests cancellation; App.Close separately waits
|
||||
// for the actual handler return before releasing shared resources.
|
||||
shutdownErr = errors.Join(shutdownErr, server.Close())
|
||||
}
|
||||
if err := <-serveErr; err != nil && err != http.ErrServerClosed {
|
||||
|
||||
63
app/body.go
63
app/body.go
@@ -1,11 +1,68 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"git.yuntuo.hk/k2safe/go-omni/httpx"
|
||||
"git.yuntuo.hk/k2safe/go-omni/response"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/config"
|
||||
"git.yuntuo.hk/k2safe/go-omni/httpx"
|
||||
"git.yuntuo.hk/k2safe/go-omni/internal/httplimits"
|
||||
"git.yuntuo.hk/k2safe/go-omni/response"
|
||||
)
|
||||
|
||||
type routeBodyLimit struct {
|
||||
method, path string
|
||||
maxBytes int64
|
||||
}
|
||||
|
||||
func (a *App) hardBodyLimit() int64 {
|
||||
if a.config.HTTP.MaxBodyBytes > 0 {
|
||||
return a.config.HTTP.MaxBodyBytes
|
||||
}
|
||||
return httplimits.MaxBodyBytes
|
||||
}
|
||||
|
||||
func (a *App) defaultBodyLimit() int64 {
|
||||
limit := a.config.HTTP.DefaultMaxBodyBytes
|
||||
if limit <= 0 {
|
||||
limit = httplimits.DefaultBodyBytes
|
||||
}
|
||||
if limit > a.hardBodyLimit() {
|
||||
limit = a.hardBodyLimit()
|
||||
}
|
||||
// Invalid configuration is rejected at startup or the first external request.
|
||||
if limit > httplimits.MaxBodyBytes {
|
||||
limit = httplimits.MaxBodyBytes
|
||||
}
|
||||
return limit
|
||||
}
|
||||
|
||||
// ValidateHTTP checks HTTP settings and registered route body limits without
|
||||
// opening a listener. Call after registration when hosting App/Engine externally.
|
||||
// Built-in serving entry points and RunWith perform this check automatically.
|
||||
func (a *App) ValidateHTTP() error {
|
||||
if err := a.config.HTTP.Validate(); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, route := range a.bodyLimits {
|
||||
if route.maxBytes > a.hardBodyLimit() {
|
||||
return &config.ConfigError{Field: "http.maxBodyBytes", Message: fmt.Sprintf("route %s %s maxBytes=%d exceeds service hard limit=%d; raise http.maxBodyBytes or lower the route limit", route.method, route.path, route.maxBytes, a.hardBodyLimit())}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func joinRoutePath(prefix, path string) string {
|
||||
if path == "" {
|
||||
if prefix == "" {
|
||||
return "/"
|
||||
}
|
||||
return prefix
|
||||
}
|
||||
return strings.TrimSuffix(prefix, "/") + "/" + strings.TrimPrefix(path, "/")
|
||||
}
|
||||
|
||||
func (a *App) bodyHandler(maxBytes int64, encoder response.Encoder, handler httpx.Handler, middleware ...httpx.Middleware) http.Handler {
|
||||
checked := httpx.LimitBody(maxBytes)(handler)
|
||||
if encoder == nil {
|
||||
@@ -25,8 +82,10 @@ func (a *App) bodyHandler(maxBytes int64, encoder response.Encoder, handler http
|
||||
// the response policy, including auth failures and parameter errors.
|
||||
func (a *App) HandleWithBodyLimit(method, path string, maxBytes int64, encoder response.Encoder, handler httpx.Handler, middleware ...httpx.Middleware) {
|
||||
a.router.Handle(method, path, a.bodyHandler(maxBytes, encoder, handler, middleware...))
|
||||
a.bodyLimits = append(a.bodyLimits, routeBodyLimit{method, path, maxBytes})
|
||||
}
|
||||
func (g *RouterGroup) HandleWithBodyLimit(method, path string, maxBytes int64, encoder response.Encoder, handler httpx.Handler, middleware ...httpx.Middleware) {
|
||||
combined := append(append([]httpx.Middleware{}, g.middlewares...), middleware...)
|
||||
g.group.Handle(method, path, g.app.bodyHandler(maxBytes, encoder, handler, combined...))
|
||||
g.app.bodyLimits = append(g.app.bodyLimits, routeBodyLimit{method, joinRoutePath(g.prefix, path), maxBytes})
|
||||
}
|
||||
|
||||
224
app/bodyPolicy_test.go
Normal file
224
app/bodyPolicy_test.go
Normal file
@@ -0,0 +1,224 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/config"
|
||||
"git.yuntuo.hk/k2safe/go-omni/httpx"
|
||||
"git.yuntuo.hk/k2safe/go-omni/internal/testutil"
|
||||
"git.yuntuo.hk/k2safe/go-omni/response"
|
||||
)
|
||||
|
||||
func serveBodyPolicy(t *testing.T, a *App, transport string) (string, *http.Client) {
|
||||
t.Helper()
|
||||
cert, key, clientTLS := testutil.CertificateFiles(t)
|
||||
a.config.Transport = transport
|
||||
a.config.TLS = config.TLSConfig{CertFile: cert, KeyFile: key}
|
||||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- a.ServeContext(ctx, l) }()
|
||||
client := &http.Client{Timeout: 5 * time.Second, Transport: &http.Transport{TLSClientConfig: clientTLS, ForceAttemptHTTP2: transport == config.TransportHTTP2}}
|
||||
t.Cleanup(func() {
|
||||
client.CloseIdleConnections()
|
||||
cancel()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Error("server did not stop")
|
||||
}
|
||||
})
|
||||
return "https://" + l.Addr().String(), client
|
||||
}
|
||||
|
||||
func consumeBody(c *httpx.Context) error {
|
||||
_, err := io.Copy(io.Discard, c.Request().Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return c.Text(200, "accepted")
|
||||
}
|
||||
|
||||
func TestUploadPolicyAcrossTransports(t *testing.T) {
|
||||
for _, transport := range []string{config.TransportNetHTTP, config.TransportNative, config.TransportHTTP2} {
|
||||
t.Run(transport, func(t *testing.T) {
|
||||
a := New()
|
||||
a.POST("/default", consumeBody)
|
||||
a.HandleWithBodyLimit("POST", "/large", 16<<20, nil, consumeBody)
|
||||
a.Group("/v1").Group("files").HandleWithBodyLimit("POST", "/large", 16<<20, nil, consumeBody)
|
||||
base, client := serveBodyPolicy(t, a, transport)
|
||||
payload := bytes.Repeat([]byte{'x'}, 9<<20)
|
||||
for _, chunked := range []bool{false, true} {
|
||||
for _, path := range []string{"/default", "/large", "/v1/files/large"} {
|
||||
var body io.Reader = bytes.NewReader(payload)
|
||||
if chunked {
|
||||
body = struct{ io.Reader }{body}
|
||||
}
|
||||
req, _ := http.NewRequest("POST", base+path, body)
|
||||
res, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := io.ReadAll(res.Body)
|
||||
res.Body.Close()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := 200
|
||||
if path == "/default" {
|
||||
want = 413
|
||||
}
|
||||
if res.StatusCode != want || (want == 413 && !strings.Contains(string(data), "BODY_TOO_LARGE")) {
|
||||
t.Fatalf("%s unknownLength=%v: %d %s", path, chunked, res.StatusCode, data)
|
||||
}
|
||||
if transport == config.TransportHTTP2 && res.ProtoMajor != 2 {
|
||||
t.Fatal("HTTP/2 not negotiated")
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfiguredBodyPolicyBeforeMiddleware(t *testing.T) {
|
||||
for _, transport := range []string{config.TransportNetHTTP, config.TransportNative, config.TransportHTTP2} {
|
||||
t.Run(transport, func(t *testing.T) {
|
||||
a := New(config.WithHTTPConfig(config.HTTPConfig{DefaultMaxBodyBytes: 32, MaxBodyBytes: 128}))
|
||||
var calls atomic.Int32
|
||||
a.Use(func(next httpx.Handler) httpx.Handler {
|
||||
return func(c *httpx.Context) error {
|
||||
data, err := io.ReadAll(c.Request().Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c.Request().Body = io.NopCloser(bytes.NewReader(data))
|
||||
return next(c)
|
||||
}
|
||||
})
|
||||
h := func(c *httpx.Context) error { calls.Add(1); return consumeBody(c) }
|
||||
a.POST("/default", h)
|
||||
a.HandleWithBodyLimit("POST", "/override", 64, nil, h)
|
||||
a.HandleWithBodyLimit("POST", "/hard", 128, nil, h)
|
||||
base, client := serveBodyPolicy(t, a, transport)
|
||||
for _, tc := range []struct {
|
||||
path string
|
||||
size, status int
|
||||
}{
|
||||
{"/default", 32, 200}, {"/default", 33, 413}, {"/override", 64, 200}, {"/override", 65, 413}, {"/hard", 128, 200}, {"/hard", 129, 413},
|
||||
} {
|
||||
for _, unknown := range []bool{false, true} {
|
||||
before := calls.Load()
|
||||
var body io.Reader = strings.NewReader(strings.Repeat("a", tc.size))
|
||||
if unknown {
|
||||
body = struct{ io.Reader }{body}
|
||||
}
|
||||
res, err := client.Post(base+tc.path, "application/octet-stream", body)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := io.ReadAll(res.Body)
|
||||
res.Body.Close()
|
||||
if err != nil || res.StatusCode != tc.status {
|
||||
t.Fatalf("%+v unknown=%v: %d %s %v", tc, unknown, res.StatusCode, data, err)
|
||||
}
|
||||
if tc.status == 413 && (calls.Load() != before || !strings.Contains(string(data), "BODY_TOO_LARGE")) {
|
||||
t.Fatal("oversize reached business or lost error", string(data))
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBodyHardLimitConflictBeforeServing(t *testing.T) {
|
||||
for _, entry := range []string{"RunContext", "ServeContext", "RunWith", "NewServer", "Engine"} {
|
||||
t.Run(entry, func(t *testing.T) {
|
||||
a := New(config.WithHTTPConfig(config.HTTPConfig{MaxBodyBytes: 8 << 20}))
|
||||
a.Group("/v1").Group("files").HandleWithBodyLimit("POST", "/upload/", 16<<20, nil, consumeBody)
|
||||
closed := 0
|
||||
a.OnShutdown(func() error { closed++; return nil })
|
||||
if err := a.ValidateHTTP(); err == nil {
|
||||
t.Fatal("external preflight accepted conflicting route limit")
|
||||
}
|
||||
var err error
|
||||
switch entry {
|
||||
case "RunContext":
|
||||
err = a.RunContext(context.Background(), "bad address")
|
||||
case "ServeContext":
|
||||
l, e := net.Listen("tcp", "127.0.0.1:0")
|
||||
if e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
err = a.ServeContext(context.Background(), l)
|
||||
if _, e = l.Accept(); !errors.Is(e, net.ErrClosed) {
|
||||
t.Fatal("listener leaked", e)
|
||||
}
|
||||
case "RunWith":
|
||||
err = a.RunWith(context.Background(), runnerFunc(func(context.Context, http.Handler, string) error { t.Error("runner started"); return nil }))
|
||||
case "NewServer":
|
||||
func() {
|
||||
defer func() {
|
||||
if p := recover(); p != nil {
|
||||
err, _ = p.(error)
|
||||
}
|
||||
}()
|
||||
a.NewServer("")
|
||||
}()
|
||||
_ = a.Close()
|
||||
case "Engine":
|
||||
w := httptest.NewRecorder()
|
||||
a.Engine().ServeHTTP(w, httptest.NewRequest("POST", "/v1/files/upload/", nil))
|
||||
if w.Code != 503 {
|
||||
t.Fatal(w.Code)
|
||||
}
|
||||
err = a.validateErr
|
||||
_ = a.Close()
|
||||
}
|
||||
for _, part := range []string{"POST /v1/files/upload/", "16777216", "8388608", "http.maxBodyBytes"} {
|
||||
if err == nil || !strings.Contains(err.Error(), part) {
|
||||
t.Fatalf("missing %q: %v", part, err)
|
||||
}
|
||||
}
|
||||
if closed != 1 {
|
||||
t.Fatal("cleanup count", closed)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDefaultBodyLimitRegistrationPaths(t *testing.T) {
|
||||
a := New(config.WithHTTPConfig(config.HTTPConfig{DefaultMaxBodyBytes: 4, MaxBodyBytes: 16}))
|
||||
h := func(*httpx.Context) error { t.Error("oversize reached business"); return nil }
|
||||
encoder, _ := response.New(response.Config{})
|
||||
a.Handle("POST", "/handle", h)
|
||||
a.HandleWithResponse("POST", "/response", encoder, h)
|
||||
a.Any("/any", h)
|
||||
g := a.Group("/group")
|
||||
g.POST("/post", h)
|
||||
g.Any("/any", h)
|
||||
g.HandleWithResponse("POST", "/response", encoder, h)
|
||||
for _, path := range []string{"/handle", "/response", "/any", "/group/post", "/group/any", "/group/response"} {
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest("POST", path, strings.NewReader("12345"))
|
||||
a.ServeHTTP(w, r)
|
||||
if w.Code != 413 {
|
||||
t.Fatal(path, w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
88
app/requests.go
Normal file
88
app/requests.go
Normal file
@@ -0,0 +1,88 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/lifecycle"
|
||||
)
|
||||
|
||||
// Admission and completion share a lock: once stopped, no request can enter
|
||||
// between observing zero active handlers and releasing their dependencies.
|
||||
type requestDrain struct {
|
||||
mu sync.Mutex
|
||||
stopped bool
|
||||
active map[*requestWork]struct{}
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
type requestWork struct{ cancel context.CancelFunc }
|
||||
|
||||
func (d *requestDrain) begin(ctx context.Context) (context.Context, *requestWork) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
if d.stopped {
|
||||
return ctx, nil
|
||||
}
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
w := &requestWork{cancel: cancel}
|
||||
if d.active == nil {
|
||||
d.active = make(map[*requestWork]struct{})
|
||||
}
|
||||
d.active[w] = struct{}{}
|
||||
return ctx, w
|
||||
}
|
||||
|
||||
func (d *requestDrain) end(w *requestWork) {
|
||||
w.cancel()
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
delete(d.active, w)
|
||||
if d.stopped && len(d.active) == 0 {
|
||||
close(d.done)
|
||||
}
|
||||
}
|
||||
|
||||
func (d *requestDrain) stop() {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
if d.stopped {
|
||||
return
|
||||
}
|
||||
d.stopped = true
|
||||
if len(d.active) == 0 {
|
||||
close(d.done)
|
||||
}
|
||||
}
|
||||
|
||||
func (d *requestDrain) cancelAndWait(timeout time.Duration) error {
|
||||
d.stop()
|
||||
d.mu.Lock()
|
||||
for w := range d.active {
|
||||
w.cancel()
|
||||
}
|
||||
d.mu.Unlock()
|
||||
timer := time.NewTimer(timeout)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-d.done:
|
||||
return nil
|
||||
case <-timer.C:
|
||||
select {
|
||||
case <-d.done:
|
||||
return nil
|
||||
default:
|
||||
}
|
||||
return fmt.Errorf("go-omni: HTTP handlers did not finish within cancellation grace %s: %w", timeout,
|
||||
errors.Join(context.DeadlineExceeded, lifecycle.ErrResourcesRetained))
|
||||
}
|
||||
}
|
||||
|
||||
// watchShutdown stops admission on the early signal without canceling accepted
|
||||
// work. The transport retains its graceful drain budget before Close cancels.
|
||||
func (a *App) watchShutdown(ctx context.Context) func() bool {
|
||||
return context.AfterFunc(ctx, a.requests.stop)
|
||||
}
|
||||
167
app/requests_test.go
Normal file
167
app/requests_test.go
Normal file
@@ -0,0 +1,167 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/config"
|
||||
"git.yuntuo.hk/k2safe/go-omni/httpx"
|
||||
"git.yuntuo.hk/k2safe/go-omni/internal/testutil"
|
||||
"git.yuntuo.hk/k2safe/go-omni/lifecycle"
|
||||
)
|
||||
|
||||
func TestShutdownWaitsForCancellationCleanup(t *testing.T) {
|
||||
for _, transport := range []string{config.TransportNetHTTP, config.TransportNative, config.TransportHTTP2} {
|
||||
t.Run(transport, func(t *testing.T) {
|
||||
cert, key, tlsClient := testutil.CertificateFiles(t)
|
||||
a := New(config.WithTransport(transport), config.WithTLS(cert, key), config.WithShutdownTimeout(30*time.Millisecond), config.WithHTTPConfig(config.HTTPConfig{ShutdownCancelTimeout: time.Second}))
|
||||
started, canceled, release := make(chan struct{}), make(chan struct{}), make(chan struct{})
|
||||
var once sync.Once
|
||||
unblock := func() { once.Do(func() { close(release) }) }
|
||||
defer unblock()
|
||||
var closed atomic.Int32
|
||||
a.OnShutdown(func() error { closed.Add(1); return nil })
|
||||
a.GET("/work", func(c *httpx.Context) error {
|
||||
close(started)
|
||||
<-c.Context().Done()
|
||||
close(canceled)
|
||||
<-release
|
||||
if closed.Load() != 0 {
|
||||
t.Error("dependency closed during cancellation cleanup")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- a.ServeContext(ctx, l) }()
|
||||
client := &http.Client{Timeout: 3 * time.Second, Transport: &http.Transport{TLSClientConfig: tlsClient, ForceAttemptHTTP2: true}}
|
||||
defer client.CloseIdleConnections()
|
||||
requestDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(requestDone)
|
||||
res, err := client.Get("https://" + l.Addr().String() + "/work")
|
||||
if err == nil {
|
||||
io.Copy(io.Discard, res.Body)
|
||||
res.Body.Close()
|
||||
}
|
||||
}()
|
||||
await(t, started)
|
||||
cancel()
|
||||
await(t, canceled)
|
||||
if closed.Load() != 0 {
|
||||
t.Fatal("dependency closed on cancellation")
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
t.Fatal("returned before handler cleanup", err)
|
||||
default:
|
||||
}
|
||||
// Even an in-memory invocation through Engine must now be rejected.
|
||||
w := httptest.NewRecorder()
|
||||
a.Engine().ServeHTTP(w, httptest.NewRequest("GET", "/work", nil))
|
||||
if w.Code != 503 {
|
||||
t.Fatal(w.Code)
|
||||
}
|
||||
unblock()
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, context.DeadlineExceeded) || errors.Is(err, lifecycle.ErrResourcesRetained) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("shutdown stuck")
|
||||
}
|
||||
await(t, requestDone)
|
||||
if closed.Load() != 1 {
|
||||
t.Fatal("cleanup count", closed.Load())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExternalServingEntriesRetainActiveResources(t *testing.T) {
|
||||
for _, entry := range []string{"App", "Engine", "NewServer", "RunWith"} {
|
||||
t.Run(entry, func(t *testing.T) {
|
||||
a := New(config.WithHTTPConfig(config.HTTPConfig{ShutdownCancelTimeout: 30 * time.Millisecond}))
|
||||
started, release, returned := make(chan struct{}), make(chan struct{}), make(chan struct{})
|
||||
var closed atomic.Int32
|
||||
a.OnShutdown(func() error { closed.Add(1); return nil })
|
||||
a.GET("/", func(*httpx.Context) error {
|
||||
close(started)
|
||||
<-release
|
||||
if closed.Load() != 0 {
|
||||
t.Error("closed while active")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
var h http.Handler = a
|
||||
if entry == "Engine" {
|
||||
h = a.Engine()
|
||||
}
|
||||
if entry == "NewServer" {
|
||||
h = a.NewServer("").Handler
|
||||
}
|
||||
invoke := func(h http.Handler) {
|
||||
defer close(returned)
|
||||
h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", "/", nil))
|
||||
}
|
||||
defer func() { close(release); await(t, returned) }()
|
||||
var err error
|
||||
if entry == "RunWith" {
|
||||
err = a.RunWith(context.Background(), runnerFunc(func(_ context.Context, h http.Handler, _ string) error { go invoke(h); await(t, started); return nil }))
|
||||
} else {
|
||||
go invoke(h)
|
||||
await(t, started)
|
||||
err = a.Close()
|
||||
}
|
||||
if !errors.Is(err, lifecycle.ErrResourcesRetained) || closed.Load() != 0 {
|
||||
t.Fatal("retention failed", err, closed.Load())
|
||||
}
|
||||
if err = a.Close(); !errors.Is(err, lifecycle.ErrResourcesRetained) {
|
||||
t.Fatal("repeat close lost retention", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestAdmissionAndCloseRace(t *testing.T) {
|
||||
for attempt := 0; attempt < 20; attempt++ {
|
||||
a := New()
|
||||
var closed atomic.Bool
|
||||
a.OnShutdown(func() error { closed.Store(true); return nil })
|
||||
a.GET("/", func(c *httpx.Context) error {
|
||||
if closed.Load() {
|
||||
t.Error("handler admitted after resource release")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
start := make(chan struct{})
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 20; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
a.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", "/", nil))
|
||||
}()
|
||||
}
|
||||
close(start)
|
||||
if err := a.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
}
|
||||
@@ -10,7 +10,7 @@ import (
|
||||
)
|
||||
|
||||
func (a *App) handle(routes httpx.RouterGroup, method, path string, handler httpx.Handler, middleware []httpx.Middleware) {
|
||||
routes.Handle(method, path, a.adapt(handler, middleware...))
|
||||
routes.Handle(method, path, a.bodyHandler(a.defaultBodyLimit(), nil, handler, middleware...))
|
||||
}
|
||||
|
||||
func (a *App) adapt(handler httpx.Handler, middleware ...httpx.Middleware) http.Handler {
|
||||
@@ -62,7 +62,7 @@ func (a *App) HandleWithResponse(method, path string, encoder response.Encoder,
|
||||
if encoder == nil {
|
||||
encoder = a.responseEncoder
|
||||
}
|
||||
a.router.Handle(method, path, a.adaptResponse(encoder, handler, middleware...))
|
||||
a.router.Handle(method, path, a.bodyHandler(a.defaultBodyLimit(), encoder, handler, middleware...))
|
||||
}
|
||||
|
||||
func (a *App) GET(path string, handler httpx.Handler, middleware ...httpx.Middleware) {
|
||||
@@ -86,18 +86,19 @@ func (a *App) DELETE(path string, handler httpx.Handler, middleware ...httpx.Mid
|
||||
}
|
||||
|
||||
func (a *App) Any(path string, handler httpx.Handler, middleware ...httpx.Middleware) {
|
||||
a.router.Any(path, a.adapt(handler, middleware...))
|
||||
a.router.Any(path, a.bodyHandler(a.defaultBodyLimit(), nil, handler, middleware...))
|
||||
}
|
||||
|
||||
// RouterGroup scopes a route prefix and optional middleware.
|
||||
type RouterGroup struct {
|
||||
prefix string
|
||||
group httpx.RouterGroup
|
||||
app *App
|
||||
middlewares []httpx.Middleware
|
||||
}
|
||||
|
||||
func (a *App) Group(path string, middleware ...httpx.Middleware) *RouterGroup {
|
||||
return &RouterGroup{group: a.router.Group(path), app: a, middlewares: middleware}
|
||||
return &RouterGroup{group: a.router.Group(path), app: a, middlewares: middleware, prefix: joinRoutePath("", path)}
|
||||
}
|
||||
|
||||
func (g *RouterGroup) Use(middleware ...httpx.Middleware) *RouterGroup {
|
||||
@@ -108,7 +109,7 @@ func (g *RouterGroup) Use(middleware ...httpx.Middleware) *RouterGroup {
|
||||
func (g *RouterGroup) Group(path string, middleware ...httpx.Middleware) *RouterGroup {
|
||||
combined := append([]httpx.Middleware{}, g.middlewares...)
|
||||
combined = append(combined, middleware...)
|
||||
return &RouterGroup{group: g.group.Group(path), app: g.app, middlewares: combined}
|
||||
return &RouterGroup{group: g.group.Group(path), app: g.app, middlewares: combined, prefix: joinRoutePath(g.prefix, path)}
|
||||
}
|
||||
|
||||
func (g *RouterGroup) route(method, path string, handler httpx.Handler, middleware []httpx.Middleware) {
|
||||
@@ -128,7 +129,7 @@ func (g *RouterGroup) HandleWithResponse(method, path string, encoder response.E
|
||||
}
|
||||
combined := append([]httpx.Middleware{}, g.middlewares...)
|
||||
combined = append(combined, middleware...)
|
||||
g.group.Handle(method, path, g.app.adaptResponse(encoder, handler, combined...))
|
||||
g.group.Handle(method, path, g.app.bodyHandler(g.app.defaultBodyLimit(), encoder, handler, combined...))
|
||||
}
|
||||
|
||||
func (g *RouterGroup) GET(path string, handler httpx.Handler, middleware ...httpx.Middleware) {
|
||||
@@ -154,5 +155,5 @@ func (g *RouterGroup) DELETE(path string, handler httpx.Handler, middleware ...h
|
||||
func (g *RouterGroup) Any(path string, handler httpx.Handler, middleware ...httpx.Middleware) {
|
||||
combined := append([]httpx.Middleware{}, g.middlewares...)
|
||||
combined = append(combined, middleware...)
|
||||
g.group.Any(path, g.app.adapt(handler, combined...))
|
||||
g.group.Any(path, g.app.bodyHandler(g.app.defaultBodyLimit(), nil, handler, combined...))
|
||||
}
|
||||
|
||||
@@ -10,13 +10,14 @@ import (
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/config"
|
||||
"git.yuntuo.hk/k2safe/go-omni/internal/httplimits"
|
||||
h2transport "git.yuntuo.hk/k2safe/go-omni/transport/http2"
|
||||
"git.yuntuo.hk/k2safe/go-omni/transport/native"
|
||||
)
|
||||
|
||||
// ServeContext serves a caller-created listener using the configured transport
|
||||
// and TLS. App takes ownership of the listener, including on startup failure,
|
||||
// and runs resource cleanup after the server has stopped.
|
||||
// and runs resource cleanup only after admitted handlers have returned.
|
||||
func (a *App) ServeContext(ctx context.Context, listener net.Listener) (runErr error) {
|
||||
defer func() { runErr = shutdownResult(ctx, runErr) }()
|
||||
if listener == nil {
|
||||
@@ -30,6 +31,8 @@ func (a *App) ServeContext(ctx context.Context, listener net.Listener) (runErr e
|
||||
_ = listener.Close()
|
||||
return a.Close()
|
||||
}
|
||||
stopWatching := a.watchShutdown(ctx)
|
||||
defer stopWatching()
|
||||
serve, err := a.prepareTransport()
|
||||
if err != nil {
|
||||
_ = listener.Close()
|
||||
@@ -39,6 +42,9 @@ func (a *App) ServeContext(ctx context.Context, listener net.Listener) (runErr e
|
||||
}
|
||||
|
||||
func (a *App) prepareTransport() (func(context.Context, net.Listener) error, error) {
|
||||
if err := a.ValidateHTTP(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := a.config.ValidateTransport(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -57,6 +63,9 @@ func (a *App) prepareTransport() (func(context.Context, net.Listener) error, err
|
||||
switch a.config.Transport {
|
||||
case config.TransportNative:
|
||||
server := native.New(a)
|
||||
// App enforces the service/route policy after routing. Keep only the
|
||||
// supported framing ceiling here, never a hidden smaller transport cap.
|
||||
server.MaxBodyBytes = httplimits.MaxBodyBytes
|
||||
server.ShutdownTimeout = shutdownTimeout
|
||||
cfg := a.config.HTTP
|
||||
if cfg.ReadHeaderTimeout > 0 {
|
||||
@@ -82,6 +91,7 @@ func (a *App) prepareTransport() (func(context.Context, net.Listener) error, err
|
||||
return server.Serve, nil
|
||||
case config.TransportHTTP2:
|
||||
server := h2transport.New(a)
|
||||
server.MaxBodyBytes = httplimits.MaxBodyBytes
|
||||
server.ShutdownTimeout = shutdownTimeout
|
||||
if a.config.HTTP.IdleTimeout > 0 {
|
||||
server.IdleTimeout = a.config.HTTP.IdleTimeout
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
"git.yuntuo.hk/k2safe/go-omni/config"
|
||||
"git.yuntuo.hk/k2safe/go-omni/httpx"
|
||||
"git.yuntuo.hk/k2safe/go-omni/internal/testutil"
|
||||
"git.yuntuo.hk/k2safe/go-omni/lifecycle"
|
||||
)
|
||||
|
||||
func TestConfiguredTransportLifecycle(t *testing.T) {
|
||||
@@ -269,7 +270,9 @@ func TestConfiguredShutdownDoesNotWaitForUncooperativeHandler(t *testing.T) {
|
||||
for _, transport := range []string{config.TransportNetHTTP, config.TransportNative, config.TransportHTTP2} {
|
||||
t.Run(transport, func(t *testing.T) {
|
||||
cert, key, tlsClient := testutil.CertificateFiles(t)
|
||||
application := New(config.WithTransport(transport), config.WithTLS(cert, key), config.WithShutdownTimeout(80*time.Millisecond))
|
||||
application := New(config.WithTransport(transport), config.WithTLS(cert, key), config.WithShutdownTimeout(80*time.Millisecond), config.WithHTTPConfig(config.HTTPConfig{ShutdownCancelTimeout: 40 * time.Millisecond}))
|
||||
var cleaned atomic.Int32
|
||||
application.OnShutdown(func() error { cleaned.Add(1); return nil })
|
||||
started := make(chan context.Context, 1)
|
||||
release, finished := make(chan struct{}), make(chan struct{})
|
||||
application.GET("/blocked", func(ctx *httpx.Context) error {
|
||||
@@ -304,11 +307,17 @@ func TestConfiguredShutdownDoesNotWaitForUncooperativeHandler(t *testing.T) {
|
||||
close(release)
|
||||
t.Fatal("handler did not start")
|
||||
}
|
||||
defer func() { close(release); await(t, finished) }()
|
||||
defer func() {
|
||||
close(release)
|
||||
await(t, finished)
|
||||
if err := application.Close(); !errors.Is(err, lifecycle.ErrResourcesRetained) || cleaned.Load() != 0 {
|
||||
t.Fatalf("late handler return must not trigger cleanup: %v count=%d", err, cleaned.Load())
|
||||
}
|
||||
}()
|
||||
cancel()
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
if !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, lifecycle.ErrResourcesRetained) || cleaned.Load() != 0 {
|
||||
t.Fatalf("shutdown: %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# go-omni CLI
|
||||
|
||||
v0.26 新增 `i18n check|export|types|extract`:检查独立语言资源、提取 Go 静态消息引用或已导出的错误目录,导出指定命名空间的前端包和 TypeScript 类型。输入只读,结果写 stdout;动态引用明确报告。完整命令见 [国际化工具链](../../i18n/README.md#cli-提取检查与导出)。`.api errorCatalog` 可声明 translation 引用,旧 runtime 在生成前拒绝,接入时使用同版 v0.26 CLI 与框架。
|
||||
v0.26 新增 `i18n check|export|types|extract`:检查独立语言资源、提取 Go 静态消息引用或已导出的错误目录,导出指定命名空间的前端包和 TypeScript 类型。输入只读,结果写 stdout;动态引用明确报告。完整命令见 [国际化工具链](../../i18n/README.md#cli-提取检查与导出)。`.api errorCatalog` 可声明 translation 引用,旧 runtime 在生成前拒绝,接入时使用当前同版 CLI 与框架。
|
||||
|
||||
v0.19 移除结构映射生成:`omni.gen.json` 只保留 Repository 和中间件选项。普通接口直接使用生成 DTO;需要类型转换时用普通 Go 或可选 [copyx](../../COPY.md)。旧 `mappings` / `mappingFiles` 必须按迁移步骤清理,CLI 不会静默删除仍在调用的旧生成文件。
|
||||
|
||||
@@ -16,7 +16,7 @@ v0.14 新增 `api format [--check]`、`api upgrade [--dry-run|--preview <空目
|
||||
|
||||
```sh
|
||||
# 仓库读取权限与 PATH 配置见根目录 README
|
||||
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.26.0-preview
|
||||
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.27.0-preview
|
||||
go-omni new --module example.com/hello ./hello
|
||||
|
||||
# 从自己的契约生成;service.api 及其相对 import 文件需要在本地存在
|
||||
@@ -51,14 +51,14 @@ main、Gen.go、OpenAPI 和 manifest 由生成器维护;业务文件按字节
|
||||
|
||||
## 版本与迁移
|
||||
|
||||
CLI 与框架依赖固定使用 v0.26.0-preview。远程安装的 CLI 默认把自身版本写入 go.mod,不添加本地 replace;`--framework-version` 可显式指定已发布的兼容版本。仅在开发框架源码时使用本地替换:
|
||||
CLI 与框架依赖固定使用 v0.27.0-preview。远程安装的 CLI 默认把自身版本写入 go.mod,不添加本地 replace;`--framework-version` 可显式指定已发布的兼容版本。仅在开发框架源码时使用本地替换:
|
||||
|
||||
```sh
|
||||
# 在 go-omni 源码仓库中执行
|
||||
go run ./cmd/go-omni new --module example.com/hello --framework-dir . ../hello
|
||||
```
|
||||
|
||||
当前模板与依赖以 v0.26.0-preview 为基线,继续使用 manifest v2;GORM 可选适配器从 v0.14.1-preview 提供。已有项目使用 [定点升级与预览](../../DEVELOPMENT.md#v013-项目升级):原始模板可自动替换,含业务修改的基础文件按预览合并;通用清理由 Runtime 负责。独立工具命令不会被删除。受管理项目通过 `.api` 模块导入增加业务,不使用旧 `add` 命令的手动路由装配。
|
||||
当前模板与依赖以 v0.27.0-preview 为基线,继续使用 manifest v2;GORM 可选适配器从 v0.14.1-preview 提供。已有项目使用 [定点升级与预览](../../DEVELOPMENT.md#v013-项目升级):原始模板可自动替换,含业务修改的基础文件按预览合并;通用清理由 Runtime 负责。独立工具命令不会被删除。受管理项目通过 `.api` 模块导入增加业务,不使用旧 `add` 命令的手动路由装配。
|
||||
|
||||
## 数据库迁移命令
|
||||
|
||||
|
||||
@@ -22,7 +22,7 @@ const frameworkModule = "git.yuntuo.hk/k2safe/go-omni"
|
||||
|
||||
// Runtime-only releases need not raise the managed template compatibility floor.
|
||||
const minimumVersion = "v0.22.0-preview"
|
||||
const currentVersion = "v0.26.0-preview"
|
||||
const currentVersion = "v0.27.0-preview"
|
||||
const generatedHeader = "// Code generated by go-omni api. DO NOT EDIT.\n"
|
||||
|
||||
type Options struct {
|
||||
|
||||
@@ -30,12 +30,12 @@ func main(){
|
||||
func run(ctx context.Context,path string)(runErr error){
|
||||
cfg,err:=config.Load(path);if err!=nil{return err}
|
||||
logger,err:=logx.New(cfg.Server.Log);if err!=nil{return err}
|
||||
// Unjoined background work may still log. The owning process exits on this
|
||||
// Unjoined HTTP or background work may still log. The owning process exits on this
|
||||
// error; do not close its borrowed logger before that work has finished.
|
||||
defer func(){if !errors.Is(runErr,lifecycle.ErrResourcesRetained){runErr=errors.Join(runErr,logger.Close())}}()
|
||||
deps,err:=svc.New(ctx,logger,cfg);if err!=nil{return err}
|
||||
// RunContext drains HTTP requests before returning. This scope owns dependencies
|
||||
// and closes them exactly once, before closing the borrowed logger.
|
||||
// Finish propagates incomplete HTTP draining: stop background work but retain
|
||||
// shared dependencies when any handler is still running. Otherwise close once.
|
||||
defer func(){runErr=deps.Finish(runErr)}()
|
||||
app,err:=handler.NewApp(deps);if err!=nil{return err}
|
||||
defer func(){runErr=errors.Join(runErr,app.Close())}()
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
)
|
||||
|
||||
var Opened, Closed, DownloadClosed atomic.Int32
|
||||
var SlowReturned atomic.Int32
|
||||
var Started = make(chan struct{}, 1)
|
||||
var Release = make(chan struct{})
|
||||
var Current atomic.Pointer[ServiceContext]
|
||||
|
||||
@@ -71,6 +71,7 @@ type countedReader struct{ io.Reader }
|
||||
|
||||
func (c *countedReader) Close() error { svc.DownloadClosed.Add(1); return nil }
|
||||
func (l *Logic) Slow(context.Context) (*types.Receipt, error) {
|
||||
defer svc.SlowReturned.Add(1)
|
||||
before := svc.Closed.Load()
|
||||
svc.Started <- struct{}{}
|
||||
<-svc.Release
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"example.com/managed/internal/handler"
|
||||
"example.com/managed/internal/svc"
|
||||
"git.yuntuo.hk/k2safe/go-omni/health"
|
||||
"git.yuntuo.hk/k2safe/go-omni/lifecycle"
|
||||
"git.yuntuo.hk/k2safe/go-omni/logx"
|
||||
)
|
||||
|
||||
@@ -229,3 +230,80 @@ func TestGeneratedEndpointConflictCleansUp(t *testing.T) {
|
||||
t.Fatal("collision leaked resources or started serving", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGeneratedIncompleteHTTPDrainRetainsDependencies(t *testing.T) {
|
||||
for _, transport := range []string{"nethttp", "native"} {
|
||||
t.Run(transport, func(t *testing.T) {
|
||||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
address := l.Addr().String()
|
||||
l.Close()
|
||||
path := filepath.Join(t.TempDir(), "config.json")
|
||||
logFile := filepath.Join(t.TempDir(), "service.log")
|
||||
data, _ := json.Marshal(map[string]any{"server": map[string]any{"address": address, "transport": transport, "shutdownTimeout": "30ms", "http": map[string]string{"shutdownCancelTimeout": "30ms"}, "log": map[string]any{"file": logFile, "rotation": map[string]any{"enabled": true}}}})
|
||||
if err = os.WriteFile(path, data, 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
svc.Started = make(chan struct{}, 1)
|
||||
svc.Release = make(chan struct{})
|
||||
before := svc.Closed.Load()
|
||||
returned := svc.SlowReturned.Load()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(ctx, path) }()
|
||||
waitReady(t, "http://"+address, done)
|
||||
client := &http.Client{Timeout: 3 * time.Second}
|
||||
defer client.CloseIdleConnections()
|
||||
requestDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(requestDone)
|
||||
res, err := client.Get("http://" + address + "/slow")
|
||||
if err == nil {
|
||||
io.Copy(io.Discard, res.Body)
|
||||
res.Body.Close()
|
||||
}
|
||||
}()
|
||||
await(t, svc.Started)
|
||||
// Capture the channel, because a later test installs its own fixture.
|
||||
release := svc.Release
|
||||
deps := svc.Current.Load()
|
||||
defer func() {
|
||||
close(release)
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for svc.SlowReturned.Load() == returned {
|
||||
if time.Now().After(deadline) {
|
||||
t.Error("handler did not exit after test release")
|
||||
break
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
// The test owns the controlled handler and has now joined it.
|
||||
_ = deps.Logger.Close()
|
||||
}()
|
||||
cancel()
|
||||
select {
|
||||
case err = <-done:
|
||||
if !errors.Is(err, lifecycle.ErrResourcesRetained) || !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("unbounded HTTP shutdown")
|
||||
}
|
||||
if err = deps.Close(); !errors.Is(err, lifecycle.ErrResourcesRetained) {
|
||||
t.Fatal("deferred Close bypassed retention", err)
|
||||
}
|
||||
if svc.Closed.Load() != before {
|
||||
t.Fatal("generated main closed client during active request")
|
||||
}
|
||||
deps.Logger.Info("retained.http.logger")
|
||||
logs, err := os.ReadFile(logFile)
|
||||
if err != nil || !strings.Contains(string(logs), "retained.http.logger") {
|
||||
t.Fatal("generated main closed logger during active request", err, string(logs))
|
||||
}
|
||||
await(t, requestDone)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,7 +10,7 @@ const frameworkModule = "git.yuntuo.hk/k2safe/go-omni"
|
||||
// Source builds can carry an unpublished VCS version (including +dirty).
|
||||
// Keep this fallback in sync with the next release; installed CLIs use their
|
||||
// downloaded module version, which also has a module checksum.
|
||||
const defaultFrameworkVersion = "v0.26.0-preview"
|
||||
const defaultFrameworkVersion = "v0.27.0-preview"
|
||||
|
||||
// This target is not a claim that the release already exists. Source users
|
||||
// select a local framework checkout until the module templates are published.
|
||||
|
||||
@@ -238,7 +238,8 @@ func WithLogConfig(settings logx.Config) Option {
|
||||
}
|
||||
}
|
||||
|
||||
// WithShutdownTimeout controls how long RunContext waits for active requests.
|
||||
// WithShutdownTimeout controls the graceful HTTP drain budget. After timeout,
|
||||
// HTTP.ShutdownCancelTimeout separately bounds the wait for canceled handlers.
|
||||
func WithShutdownTimeout(timeout time.Duration) Option {
|
||||
return func(cfg *Config) {
|
||||
if timeout > 0 {
|
||||
|
||||
@@ -3,17 +3,28 @@ package config
|
||||
import (
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/internal/httplimits"
|
||||
)
|
||||
|
||||
// HTTPConfig controls built-in HTTP servers. Positive values override the
|
||||
// selected transport's defaults; zero keeps its existing behavior. Timeouts
|
||||
// follow that transport's semantics, not a handler execution deadline.
|
||||
// HTTPConfig controls HTTP serving. Network timeouts follow the selected
|
||||
// transport's semantics, not a handler execution deadline. Body limits and
|
||||
// shutdown cancellation grace also apply to App used with an external server.
|
||||
type HTTPConfig struct {
|
||||
ReadHeaderTimeout time.Duration `json:"readHeaderTimeout"`
|
||||
ReadTimeout time.Duration `json:"readTimeout"`
|
||||
WriteTimeout time.Duration `json:"writeTimeout"`
|
||||
IdleTimeout time.Duration `json:"idleTimeout"`
|
||||
MaxHeaderBytes int `json:"maxHeaderBytes"`
|
||||
// DefaultMaxBodyBytes applies to App routes without an explicit limit.
|
||||
// Zero uses 8 MiB, or the hard limit if smaller. Route limits may override it.
|
||||
DefaultMaxBodyBytes int64 `json:"defaultMaxBodyBytes"`
|
||||
// MaxBodyBytes is an optional service hard limit. Zero adds no hard limit
|
||||
// below the framework's supported maximum of 1 TiB.
|
||||
MaxBodyBytes int64 `json:"maxBodyBytes"`
|
||||
// ShutdownCancelTimeout bounds the wait for handlers after cancellation.
|
||||
// Zero uses 5s. Resources are retained if handlers have not returned.
|
||||
ShutdownCancelTimeout time.Duration `json:"shutdownCancelTimeout"`
|
||||
// ErrorLog is a Go-only hook for the nethttp server's internal diagnostics.
|
||||
ErrorLog *log.Logger `json:"-"`
|
||||
}
|
||||
@@ -27,6 +38,7 @@ func (c HTTPConfig) Validate() error {
|
||||
{"readTimeout", c.ReadTimeout},
|
||||
{"writeTimeout", c.WriteTimeout},
|
||||
{"idleTimeout", c.IdleTimeout},
|
||||
{"shutdownCancelTimeout", c.ShutdownCancelTimeout},
|
||||
} {
|
||||
if field.value < 0 {
|
||||
return &ConfigError{Field: "http." + field.name, Message: "must be nonnegative"}
|
||||
@@ -35,11 +47,22 @@ func (c HTTPConfig) Validate() error {
|
||||
if c.MaxHeaderBytes < 0 {
|
||||
return &ConfigError{Field: "http.maxHeaderBytes", Message: "must be nonnegative"}
|
||||
}
|
||||
for _, field := range []struct {
|
||||
name string
|
||||
value int64
|
||||
}{{"defaultMaxBodyBytes", c.DefaultMaxBodyBytes}, {"maxBodyBytes", c.MaxBodyBytes}} {
|
||||
if field.value < 0 || field.value > httplimits.MaxBodyBytes {
|
||||
return &ConfigError{Field: "http." + field.name, Message: "must be 0..1 TiB (bytes)"}
|
||||
}
|
||||
}
|
||||
if c.MaxBodyBytes > 0 && c.DefaultMaxBodyBytes > c.MaxBodyBytes {
|
||||
return &ConfigError{Field: "http.defaultMaxBodyBytes", Message: "must not exceed http.maxBodyBytes"}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// WithHTTPConfig sets the built-in server's HTTP options. RunWith uses a
|
||||
// caller-owned Runner and does not apply these options to that runner.
|
||||
// WithHTTPConfig sets HTTP options. RunWith leaves network configuration to the
|
||||
// caller-owned Runner, but App body limits and cancellation grace still apply.
|
||||
func WithHTTPConfig(settings HTTPConfig) Option {
|
||||
return func(cfg *Config) { cfg.HTTP = settings }
|
||||
}
|
||||
|
||||
@@ -93,3 +93,40 @@ func TestHTTPInvalidAndUnsupportedSettings(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPBodyAndCancellationSettings(t *testing.T) {
|
||||
for _, cfg := range []config.HTTPConfig{
|
||||
{DefaultMaxBodyBytes: -1}, {MaxBodyBytes: -1}, {DefaultMaxBodyBytes: 1<<40 + 1}, {MaxBodyBytes: 1<<40 + 1},
|
||||
{DefaultMaxBodyBytes: 1024, MaxBodyBytes: 512}, {ShutdownCancelTimeout: -time.Second},
|
||||
} {
|
||||
if err := cfg.Validate(); err == nil {
|
||||
t.Fatalf("invalid HTTP config accepted: %+v", cfg)
|
||||
}
|
||||
}
|
||||
for _, transport := range []string{config.TransportNetHTTP, config.TransportNative, config.TransportHTTP2} {
|
||||
cfg := config.Default()
|
||||
cfg.Transport = transport
|
||||
cfg.TLS = config.TLSConfig{CertFile: "c", KeyFile: "k"}
|
||||
cfg.HTTP = config.HTTPConfig{DefaultMaxBodyBytes: 1 << 20, MaxBodyBytes: 16 << 20, ShutdownCancelTimeout: time.Second}
|
||||
if err := cfg.ValidateTransport(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
t.Setenv("OMNI_BODY_HARD", "16777216")
|
||||
for _, file := range []struct{ name, body string }{
|
||||
{"config.json", `{"http":{"defaultMaxBodyBytes":1048576,"maxBodyBytes":"${OMNI_BODY_HARD}","shutdownCancelTimeout":"250ms"}}`},
|
||||
{"config.yaml", "http:\n defaultMaxBodyBytes: 1048576\n maxBodyBytes: ${OMNI_BODY_HARD}\n shutdownCancelTimeout: 250ms\n"},
|
||||
} {
|
||||
path := filepath.Join(t.TempDir(), file.name)
|
||||
if err := os.WriteFile(path, []byte(file.body), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg, err := config.LoadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.HTTP.DefaultMaxBodyBytes != 1<<20 || cfg.HTTP.MaxBodyBytes != 16<<20 || cfg.HTTP.ShutdownCancelTimeout != 250*time.Millisecond {
|
||||
t.Fatalf("loaded %+v", cfg.HTTP)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
15
examples/httpSafety/README.md
Normal file
15
examples/httpSafety/README.md
Normal file
@@ -0,0 +1,15 @@
|
||||
# HTTP 限额与停机
|
||||
|
||||
从仓库根目录运行 `go run ./examples/httpSafety`,可用 `-addr 127.0.0.1:18080` 指定端口。仅使用本机 HTTP 和模拟收尾,不连接数据库或其他服务。
|
||||
|
||||
上传验证:
|
||||
|
||||
```sh
|
||||
dd if=/dev/zero bs=1048576 count=9 2>/dev/null | curl --data-binary @- http://127.0.0.1:8080/upload
|
||||
```
|
||||
|
||||
普通路由默认 1 MiB,上传路由允许 16 MiB,服务硬上限 20 MiB;以上请求返回已读取的 9437184 字节。将上传路由改为 21 MiB 后,启动会报告该路由与服务硬上限冲突。实际文件上传须额外预留 multipart 字段/边界开销,并配置文件大小限制。
|
||||
|
||||
停机验证:另一个终端执行 `curl http://127.0.0.1:8080/work`,随后在服务终端按 Ctrl-C。先等待 1s 排空,再取消请求,处理器模拟 100ms 收尾;日志中先出现 `request cleanup completed`,再出现资源关闭记录。因为用尽排空预算,本次退出仍返回超时错误。
|
||||
|
||||
若处理器不在取消收尾窗口内退出,App 和 `deps.Finish(runErr)` 返回 `lifecycle.ErrResourcesRetained` 并跳过资源关闭。真实项目拥有的日志也必须在该错误下保留,生成入口已处理。进程应非零退出;库不自行退出进程,也不强杀 goroutine。
|
||||
62
examples/httpSafety/main.go
Normal file
62
examples/httpSafety/main.go
Normal file
@@ -0,0 +1,62 @@
|
||||
// This example uses only local HTTP and simulated work; no external services.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"flag"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
omni "git.yuntuo.hk/k2safe/go-omni"
|
||||
"git.yuntuo.hk/k2safe/go-omni/lifecycle"
|
||||
)
|
||||
|
||||
func main() {
|
||||
address := flag.String("addr", "127.0.0.1:8080", "local listen address")
|
||||
flag.Parse()
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
if err := run(ctx, *address); err != nil {
|
||||
log.Print(err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func run(ctx context.Context, address string) (runErr error) {
|
||||
deps := lifecycle.NewRuntime(ctx)
|
||||
deps.OnClose(func() error { log.Print("shared resource closed after requests completed"); return nil })
|
||||
defer func() { runErr = deps.Finish(runErr) }()
|
||||
app := omni.New(omni.WithHTTPConfig(omni.HTTPConfig{
|
||||
DefaultMaxBodyBytes: 1 << 20,
|
||||
MaxBodyBytes: 20 << 20,
|
||||
ShutdownCancelTimeout: time.Second,
|
||||
}), omni.WithShutdownTimeout(time.Second))
|
||||
defer func() { runErr = errors.Join(runErr, app.Close()) }()
|
||||
app.HandleWithBodyLimit("POST", "/upload", 16<<20, nil, func(c *omni.Context) error {
|
||||
n, err := io.Copy(io.Discard, c.Request().Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return c.OK(map[string]int64{"bytes": n})
|
||||
})
|
||||
app.GET("/work", func(c *omni.Context) error {
|
||||
log.Print("request accepted")
|
||||
select {
|
||||
case <-time.After(10 * time.Second):
|
||||
return c.OK("finished")
|
||||
case <-c.Context().Done():
|
||||
// Simulate bounded rollback/cleanup after cancellation. Dependencies
|
||||
// remain available until this handler and its defers have returned.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
log.Print("request cleanup completed")
|
||||
return c.Context().Err()
|
||||
}
|
||||
})
|
||||
log.Printf("listening on %s; POST /upload or GET /work, then Ctrl-C", address)
|
||||
return app.RunContext(deps.Context(), address)
|
||||
}
|
||||
@@ -22,7 +22,7 @@
|
||||
"internal/svc/runtimeGen.go": "29f6afb6614881935099c1845ecb3f7902ee773caa077940f566cb8dfdbcdbb1",
|
||||
"internal/types/dtoManagedd28645a2af1cGen.go": "7586e86fa5d67051283178cc578c890f6c02fd2d5c383170a71a39c2e77ce344",
|
||||
"internal/types/typesGen.go": "372d67f23a313b3e77d2a3b7a236315db9721bc63350bf1f71db1293e57570d1",
|
||||
"main.go": "10cb6a84b2eb1b68dc10c2dca949c9dd7c68a9f78a8e6d8f56ce9336d4b2d125"
|
||||
"main.go": "e1cdd80f8ee090f1391cdfa799762c99b10d4e09715e9d4faefb2cb0291f1c05"
|
||||
},
|
||||
"contract": {
|
||||
"routes": {
|
||||
|
||||
@@ -2,7 +2,7 @@ module example.com/managed
|
||||
|
||||
go 1.22.0
|
||||
|
||||
require git.yuntuo.hk/k2safe/go-omni v0.26.0-preview
|
||||
require git.yuntuo.hk/k2safe/go-omni v0.27.0-preview
|
||||
|
||||
require (
|
||||
github.com/gabriel-vasile/mimetype v1.4.3 // indirect
|
||||
|
||||
@@ -41,7 +41,7 @@ func run(ctx context.Context, path string) (runErr error) {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Unjoined background work may still log. The owning process exits on this
|
||||
// Unjoined HTTP or background work may still log. The owning process exits on this
|
||||
// error; do not close its borrowed logger before that work has finished.
|
||||
defer func() {
|
||||
if !errors.Is(runErr, lifecycle.ErrResourcesRetained) {
|
||||
@@ -52,8 +52,8 @@ func run(ctx context.Context, path string) (runErr error) {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// RunContext drains HTTP requests before returning. This scope owns dependencies
|
||||
// and closes them exactly once, before closing the borrowed logger.
|
||||
// Finish propagates incomplete HTTP draining: stop background work but retain
|
||||
// shared dependencies when any handler is still running. Otherwise close once.
|
||||
defer func() { runErr = deps.Finish(runErr) }()
|
||||
app, err := handler.NewApp(deps)
|
||||
if err != nil {
|
||||
|
||||
@@ -3,7 +3,7 @@ module example.com/userService
|
||||
go 1.22.0
|
||||
|
||||
require (
|
||||
git.yuntuo.hk/k2safe/go-omni v0.26.0-preview
|
||||
git.yuntuo.hk/k2safe/go-omni v0.27.0-preview
|
||||
github.com/go-sql-driver/mysql v1.9.3
|
||||
)
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"net/http"
|
||||
|
||||
apierrors "git.yuntuo.hk/k2safe/go-omni/errors"
|
||||
"git.yuntuo.hk/k2safe/go-omni/internal/httplimits"
|
||||
"git.yuntuo.hk/k2safe/go-omni/logx"
|
||||
"git.yuntuo.hk/k2safe/go-omni/response"
|
||||
"git.yuntuo.hk/k2safe/go-omni/validate"
|
||||
@@ -284,7 +285,12 @@ func (c *Context) WriteError(err error) {
|
||||
if errors.As(err, &bodyErr) {
|
||||
err = bodyTooLarge()
|
||||
}
|
||||
var original *apierrors.APIError
|
||||
tooLarge := errors.As(err, &original) && original.Status == http.StatusRequestEntityTooLarge
|
||||
apiErr := c.ResolveError(err)
|
||||
if tooLarge || apiErr.Status == http.StatusRequestEntityTooLarge {
|
||||
httplimits.RejectBody(c.Raw())
|
||||
}
|
||||
c.failure = apiErr
|
||||
if writeErr := c.writeResult(response.Result{
|
||||
BusinessCode: apiErr.BusinessCode,
|
||||
|
||||
25
internal/httplimits/limits.go
Normal file
25
internal/httplimits/limits.go
Normal file
@@ -0,0 +1,25 @@
|
||||
// Package httplimits shares the application body policy across transports.
|
||||
package httplimits
|
||||
|
||||
import "net/http"
|
||||
|
||||
const (
|
||||
DefaultBodyBytes int64 = 8 << 20
|
||||
MaxBodyBytes int64 = 1 << 40
|
||||
)
|
||||
|
||||
// RejectBody tells cooperating transports not to drain a rejected body. The
|
||||
// semantic failure must survive encoders that map business failures to HTTP 200.
|
||||
func RejectBody(w http.ResponseWriter) {
|
||||
for {
|
||||
if reject, ok := w.(interface{ RequestBodyTooLarge() }); ok {
|
||||
reject.RequestBodyTooLarge()
|
||||
return
|
||||
}
|
||||
unwrap, ok := w.(interface{ Unwrap() http.ResponseWriter })
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
w = unwrap.Unwrap()
|
||||
}
|
||||
}
|
||||
@@ -35,14 +35,16 @@ c.OnDrain("jobs", jobs.Wait)
|
||||
|
||||
```yaml
|
||||
server:
|
||||
shutdownTimeout: 10s # HTTP 排空时限,保持原语义
|
||||
shutdownTimeout: 10s # HTTP 排空时限
|
||||
http:
|
||||
shutdownCancelTimeout: 5s # HTTP 取消后的收尾窗口
|
||||
lifecycle:
|
||||
stopTimeout: 5s
|
||||
drainTimeout: 10s
|
||||
cancelTimeout: 5s
|
||||
```
|
||||
|
||||
这些是各阶段的整体预算,不是每个回调分别等待这么久。零使用默认值,负值在启动前拒绝。可用 `GO_OMNI_LIFECYCLE_STOP_TIMEOUT`、`GO_OMNI_LIFECYCLE_DRAIN_TIMEOUT`、`GO_OMNI_LIFECYCLE_CANCEL_TIMEOUT` 覆盖;生成配置还支持完整的 `GO_OMNI_SERVER_LIFECYCLE_*` 路径,别名优先。配置应在 config 中设置,不放在创建 App 时才执行的 AppOptions 中。
|
||||
这些是各阶段的整体预算,不是每个回调分别等待这么久。零使用默认值,负值在启动前拒绝。环境变量仅通过配置文件显式引用,例如 `cancelTimeout: "${WORK_CANCEL_TIMEOUT:-5s}"`,不存在自动覆盖或别名优先。后台生命周期配置应在 config 中设置,不放在创建 App 时才执行的 AppOptions 中。
|
||||
|
||||
阶段超时会取消阶段 context 和 WorkContext,再给尚未返回的钩子一个 cancelTimeout 收尾窗口。若所有钩子都返回 nil,可以释放依赖,但仍返回超时错误,进程非零退出。task.Wait 收到取消会取消其任务并返回错误;这不证明全部任务已退出,因此仍按以下保守规则处理。
|
||||
|
||||
@@ -50,10 +52,24 @@ Stop 超时且仍未返回的组件,不会再并发调用它同名的 Drain;
|
||||
|
||||
钩子返回错误、panic、缺少匹配 Drain,或窗口结束后仍有钩子未返回时,Runtime 返回可由 `errors.Is(err, lifecycle.ErrResourcesRetained)` 判断的错误,**跳过全部 OnClose**;生成入口也保留借用的日志实例并非零退出,由进程退出释放剩余 OS 资源。不会悄悄重试 Close,也不会让晚到的后台结果触发自动关闭。错误链保留初始化/监听/后台故障及具体 HookError 的阶段、名称和原始原因。
|
||||
|
||||
手写入口必须处理这个错误并终止该服务,不能继续运行半关闭的 Runtime;嵌入式宿主需自行安排隔离/终止策略。Go 不能强杀不响应取消的 goroutine,有限等待与保证它已经结束不能同时无条件成立。OnClose 仍要求自身能返回;本版阶段时限不会强制中断资源 Close。HTTP 超时沿用现有传输层规则,业务处理器仍须响应请求取消。
|
||||
手写入口必须处理这个错误并终止该服务,不能继续运行半关闭的 Runtime;嵌入式宿主需自行安排隔离/终止策略。Go 不能强杀不响应取消的 goroutine,有限等待与保证它已经结束不能同时无条件成立。OnClose 仍要求自身能返回;阶段时限不会强制中断资源 Close。HTTP 超时也必须经过下述完成屏障,业务处理器仍须响应请求取消。
|
||||
|
||||
手写项目使用 `lifecycle.NewRuntimeWithConfig(ctx, cfg)`;完成 HTTP 排空后调用 `runtime.Finish(runErr)`。库本身不执行 os.Exit。
|
||||
|
||||
### HTTP 请求与依赖的关闭屏障
|
||||
|
||||
App 统计已进入且尚未退出的整个处理链,包括中间件、响应编码和 defer;停止接收与登记请求共用同步屏障。关闭连接、取消 context、HTTP/2 session 结束均不等于处理链结束。
|
||||
|
||||
`RunContext` / `ServeContext` 先按 `shutdownTimeout` 排空,超时取消并关闭连接;App 随后按 `http.shutdownCancelTimeout`(零默认 5s)等待处理器完成收尾。收尾成功则执行逆序 OnShutdown,仍保留原排空超时错误。收尾窗口结束时还有处理器运行,则永久返回 `ErrResourcesRetained` 并跳过所有 OnShutdown;后续 Close 或晚到的处理器返回不会触发补做清理。
|
||||
|
||||
`Runtime.Finish(runErr)` 收到该错误后,继续执行后台 OnStop/OnDrain 并取消 WorkContext,但跳过 OnClose;后续 `Runtime.Close` 也保持资源保留。生成入口通过这条错误链保留数据库等共享依赖和借用日志,并非零退出。手写入口必须传递 runErr 给 Finish,并在 `errors.Is(runErr, lifecycle.ErrResourcesRetained)` 时跳过自己拥有的额外资源清理,不能无条件 defer DB.Close/Logger.Close。
|
||||
|
||||
`App.Close` 本身会停止接收、取消请求并等待收尾,适用于外部 HTTP 宿主;正常优雅排空应先由宿主停止监听并 Shutdown,再调用 Close。不要在请求处理器或清理钩子内部调用 Close(会等待自身)。`App`、`Engine`、`NewServer().Handler` 和 `RunWith` 都经过统计;已停止的 App 返回简单 503,不再调用可能使用已关闭依赖的中间件或编码器。App 不支持关闭后重新启动。
|
||||
|
||||
外部 Runner 返回不构成请求完成证明,App 仍执行自己的完成屏障;Runner 负责关闭自己拥有的 listener/连接。原始 `Router().ServeHTTP` 绕过 App。处理器退出后继续运行的自建 goroutine、Hijack/WebSocket 后台工作,必须用 OnStop/OnDrain 注册,框架不会推断这些工作的生命周期。HTTP/2 低层 Shutdown 也会等待独立 handler goroutine;超时返回 error 仍不表示它已退出。
|
||||
|
||||
新行为增加一个有限 HTTP 收尾窗口,调用方总停机预算应计入它。已有生成项目升级框架即可获得 Runtime.Finish 的保护;重新生成可同步入口注释。可运行示例见 [httpSafety](../examples/httpSafety/README.md)。
|
||||
|
||||
### 升级
|
||||
|
||||
v0.14.x 受管理项目将 module 和 CLI 一起升级到 v0.15.0-preview,再运行 `go-omni api generate`、`go-omni api generate --check` 和业务测试。生成器更新 main/runtimeGen,保留 svc、Logic 和配置。已有配置不添加新字段也会使用默认时限,需要调整时再编辑 yaml。
|
||||
|
||||
@@ -11,10 +11,10 @@ import (
|
||||
"git.yuntuo.hk/k2safe/go-omni/internal/configdiag"
|
||||
)
|
||||
|
||||
// ErrResourcesRetained means background quiescence could not be established.
|
||||
// ErrResourcesRetained means HTTP or background quiescence was not established.
|
||||
// OnClose callbacks were deliberately skipped. Close is not retried; an owning
|
||||
// process should exit unsuccessfully instead of continuing with a stopped Runtime.
|
||||
var ErrResourcesRetained = errors.New("lifecycle: background shutdown incomplete; shared resources retained")
|
||||
var ErrResourcesRetained = errors.New("lifecycle: shutdown incomplete; shared resources retained")
|
||||
|
||||
type ShutdownConfig struct {
|
||||
StopTimeout time.Duration `json:"stopTimeout"`
|
||||
|
||||
@@ -68,6 +68,10 @@ func (r *Runtime) OnClose(cleanup func() error) {
|
||||
// skips resource callbacks and returns ErrResourcesRetained. Once release is
|
||||
// safe, all OnClose callbacks are attempted even if one fails or panics.
|
||||
func (r *Runtime) Close() error {
|
||||
return r.close(false)
|
||||
}
|
||||
|
||||
func (r *Runtime) close(retain bool) error {
|
||||
if r == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -108,7 +112,7 @@ func (r *Runtime) Close() error {
|
||||
drainErr := r.runPhase("drain", eligible, r.shutdown.DrainTimeout)
|
||||
r.cancelWork()
|
||||
r.closeErr = errors.Join(stopErr, drainErr)
|
||||
if unsafePhase(stopErr) || unsafePhase(drainErr) {
|
||||
if retain || unsafePhase(stopErr) || unsafePhase(drainErr) {
|
||||
r.closeErr = errors.Join(r.closeErr, ErrResourcesRetained)
|
||||
return
|
||||
}
|
||||
@@ -129,12 +133,14 @@ func runCleanup(fn func() error) (err error) {
|
||||
}
|
||||
|
||||
// Finish closes resources and preserves the operation, cleanup and shutdown
|
||||
// errors. Read the final cause only after monitors have been stopped by Close.
|
||||
// errors. An incoming ErrResourcesRetained (for example, unfinished HTTP work)
|
||||
// still runs stop/drain hooks, but permanently skips resource cleanup, including
|
||||
// subsequent Close calls. Read the final cause after stopping the monitors.
|
||||
func (r *Runtime) Finish(err error) error {
|
||||
if r == nil {
|
||||
return err
|
||||
}
|
||||
err = errors.Join(err, r.Close())
|
||||
err = errors.Join(err, r.close(errors.Is(err, ErrResourcesRetained)))
|
||||
if cause := r.Lifecycle.Err(); cause != nil && !errors.Is(err, cause) {
|
||||
err = errors.Join(err, cause)
|
||||
}
|
||||
|
||||
@@ -54,3 +54,28 @@ func TestRuntimeRetainsMonitorFaultDuringCleanup(t *testing.T) {
|
||||
}()
|
||||
r.OnClose(func() error { return nil })
|
||||
}
|
||||
|
||||
func TestFinishRetainsDependenciesAfterIncompleteHTTPDrain(t *testing.T) {
|
||||
r := NewRuntime(context.Background())
|
||||
var order []string
|
||||
r.OnStop("worker", func(context.Context) error { order = append(order, "stop"); return nil })
|
||||
r.OnDrain("worker", func(context.Context) error { order = append(order, "drain"); return nil })
|
||||
r.OnClose(func() error { t.Error("released dependency while HTTP work remains"); return nil })
|
||||
fault := errors.New("HTTP shutdown timeout")
|
||||
err := r.Finish(errors.Join(fault, ErrResourcesRetained))
|
||||
if !errors.Is(err, fault) || !errors.Is(err, ErrResourcesRetained) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(order, []string{"stop", "drain"}) {
|
||||
t.Fatal(order)
|
||||
}
|
||||
if r.WorkContext().Err() == nil {
|
||||
t.Fatal("background work was not canceled")
|
||||
}
|
||||
if err = r.Close(); !errors.Is(err, ErrResourcesRetained) {
|
||||
t.Fatal("deferred Close bypassed retention", err)
|
||||
}
|
||||
if err = r.Finish(nil); !errors.Is(err, ErrResourcesRetained) {
|
||||
t.Fatal("repeat Finish bypassed retention", err)
|
||||
}
|
||||
}
|
||||
|
||||
33
reports/httpSafetyAcceptance.md
Normal file
33
reports/httpSafetyAcceptance.md
Normal file
@@ -0,0 +1,33 @@
|
||||
# HTTP 限额与停机安全验收(2026-09-29)
|
||||
|
||||
对应 v0.27.0-preview;历史 v0.26.0-preview 标签保持。框架不包含业务文件规格、数据库实现或应用专属错误码。
|
||||
|
||||
## 行为验收
|
||||
|
||||
| 场景 | 结果 |
|
||||
| --- | --- |
|
||||
| nethttp/native/HTTP2:16 MiB 路由接收 9 MiB | 固定长度与未知长度均成功 |
|
||||
| 默认、路由覆盖、服务硬上限 | 边界值成功,超限进入 BODY_TOO_LARGE;读取中间件也受 reader 限额约束 |
|
||||
| 路由超过服务硬上限 | 启动前拒绝,诊断包含嵌套分组完整路径及两级字节数;外部宿主可调用 ValidateHTTP |
|
||||
| native 提前拒绝上传 | 先发出响应再有界半关闭,避免即时关闭触发 TCP reset;10 轮上传相关 race 回归通过 |
|
||||
| 错误目录将超限映射为 HTTP 200 | 业务码/HTTP 策略保持,不再继续读取被拒绝的请求体;客户端不发送 body 也能及时收到响应 |
|
||||
| 正常停机与取消后收尾 | 处理链和 defer 完成后才运行资源清理;收尾成功仍保留原排空超时错误 |
|
||||
| 不响应取消的处理器 | 有界返回 ErrResourcesRetained,跳过 App 清理;重复 Close、晚到返回不触发清理 |
|
||||
| App/Engine/NewServer/RunWith | 统一跟踪请求;停止后拒绝新请求,不执行依赖项目资源的处理链 |
|
||||
| HTTP/2 客户端断开 | Shutdown 同时等待连接和独立 handler,不把连接结束当成业务结束 |
|
||||
| 真实生成项目 | HTTP 未结束时 Runtime.Finish 仍停止后台工作,但保留客户端与日志;重复 deps.Close 不能绕过保护 |
|
||||
| 可运行示例 | 9 MiB 上传返回 9437184 字节;SIGTERM 后先请求收尾、再关闭资源;排空超时退出码为 1 |
|
||||
|
||||
## 检查
|
||||
|
||||
- `GO_OMNI_SCAFFOLD_INTEGRATION=1 go test -race -count=1 ./...`:54 个测试包通过,包含真实生成消费者的 race/build。
|
||||
- `GOTOOLCHAIN=go1.22.12 GO_OMNI_SCAFFOLD_INTEGRATION=1 go test -count=1 ./...`:54 个测试包通过。随后 native 拒绝连接收尾补充修复,在最终代码上重新执行 `GOTOOLCHAIN=go1.22.12 go test -count=1 ./app ./config ./lifecycle ./httpx ./transport/...`,8 个测试包通过。
|
||||
- `go test -race ./app ./transport/native -run 'Test(UploadPolicy|ConfiguredBody|ApplicationRejectsUnreadBody)' -count=10`:通过。
|
||||
- `go vet ./...`、`go build ./...`、`git diff --check`:通过。
|
||||
- 编译并实跑 `examples/httpSafety`,使用随机本机端口与实际 SIGTERM 验证上传和关闭顺序,无外部服务。
|
||||
|
||||
## 兼容边界
|
||||
|
||||
旧 nethttp 手写普通路由新增默认 8 MiB 限额,大请求应显式配置;生成契约默认 1 MiB 保持。取消后的等待默认 5s,需计入服务停机预算。手写资源拥有者须将运行错误传给 Runtime.Finish,并在资源保留信号下跳过自己的额外关闭操作。自建 goroutine、Hijack 后的工作仍须登记后台生命周期。
|
||||
|
||||
自定义 Runner、独立 native/http2.Server、反向代理的额外限额由各自拥有者配置;使用 WithTransport 获取统一装配。请求体、multipart 单文件和幂等缓存限额保持各自用途。已发送的流式响应无法撤回,先完成有界读取再执行业务。
|
||||
@@ -134,6 +134,7 @@ def main():
|
||||
sums = re.findall(r"checksum (h1:\S+); no local replace", evidence)
|
||||
assert len(sums) == len(tests) and all(s == dependency["Sum"] for s in sums), "remote consumer checksum mismatch"
|
||||
checks += ["initializationAndListenFailureCleanup", "drainBeforeDependencyClose", "generatedConsumerRaceTests", "errorParamsAndInheritance", "manualHooksAndScalars", "dependencyFatalShutdown", "fatalProcessExitCode", "initializationShutdownNotification", "configuredBuiltinEndpoints", "projectHealthEnvelope", "endpointConflictCleanup", "responseArrayNullableNull", "primitiveAndDynamicArrays", "strictRequestPolicies", "declarativeMultipartBinding", "backgroundStopDrainResourceOrder", "backgroundInitAndListenFailureCleanup", "backgroundTimeoutRetainsResourcesAndExits"]
|
||||
checks += ["generatedIncompleteHTTPDrainRetainsDependenciesAndLogger"]
|
||||
|
||||
# Run tests from the downloaded module, not the framework checkout.
|
||||
http_output = run(["go", "test", "-race", "-count=1", "-json", MODULE + "/httpc"], starter, env)
|
||||
@@ -152,7 +153,8 @@ def main():
|
||||
config_passed = {e.get("Test") for e in config_events if e.get("Action") == "pass"}
|
||||
assert {"TestExplicitReferencesAcrossFormats", "TestReferenceFallbackAndFixedValues",
|
||||
"TestReferenceDiagnosticsAndScalarLimits", "TestRemovedEnvTagsAndFieldSuggestion",
|
||||
"TestModuleFieldDiagnostics", "TestModuleDiagnosticsKeepReferenceAndCause"} <= config_passed
|
||||
"TestModuleFieldDiagnostics", "TestModuleDiagnosticsKeepReferenceAndCause",
|
||||
"TestHTTPBodyAndCancellationSettings"} <= config_passed
|
||||
run([cli, "api", "generate", "--check"], starter, env)
|
||||
run(["go", "run", ".", "-config", "etc/config.yaml", "-check-config"], starter, env)
|
||||
checks += ["remoteExplicitConfigurationRaceTests", "literalAndRequiredDefaultReferences",
|
||||
@@ -162,7 +164,12 @@ def main():
|
||||
"copyx": {"TestCheckTypesAndEmptyCollections", "TestCheckRulesAndConverterBoundaries", "TestCopierCheckConcurrentAndOptionsIsolation"},
|
||||
"logx": {"TestLoggerTimeZoneIsolation", "TestLoggerTimeZoneValidationAndDST", "TestLoggerTimeZoneConcurrentChildren"},
|
||||
"cors": {"TestPolicyPreflightAndActual", "TestPolicyValidation", "TestPolicyCredentialsWildcardAndDuplicates", "TestPolicyPreservesHTTPAndSnapshot"},
|
||||
"app": {"TestAppCORSRoutingAndErrors", "TestAppCORSConfigIsolation", "TestCORSTransports", "TestI18nRouteScopesEarlyFailuresAndReplay", "TestI18nExplicitProtocolsAndConcurrentApps", "TestI18nAdapterFailurePreservesBusiness"},
|
||||
"app": {"TestAppCORSRoutingAndErrors", "TestAppCORSConfigIsolation", "TestCORSTransports", "TestI18nRouteScopesEarlyFailuresAndReplay", "TestI18nExplicitProtocolsAndConcurrentApps", "TestI18nAdapterFailurePreservesBusiness",
|
||||
"TestUploadPolicyAcrossTransports", "TestConfiguredBodyPolicyBeforeMiddleware", "TestBodyHardLimitConflictBeforeServing", "TestDefaultBodyLimitRegistrationPaths",
|
||||
"TestShutdownWaitsForCancellationCleanup", "TestExternalServingEntriesRetainActiveResources", "TestRequestAdmissionAndCloseRace", "TestConfiguredShutdownDoesNotWaitForUncooperativeHandler"},
|
||||
"lifecycle": {"TestFinishRetainsDependenciesAfterIncompleteHTTPDrain"},
|
||||
"transport/http2": {"TestShutdownWaitsForHandlerAfterClientDisconnect"},
|
||||
"transport/native": {"TestApplicationRejectsUnreadBodyPromptly", "TestBodyRejectionWithBusinessHTTP200DoesNotDrain"},
|
||||
"contract": {"TestBodyPolicyPropagatesOuterLimit", "TestJSONClosingDelimiterPreservesReaderCause"},
|
||||
"i18n": {"TestIndependentImmutableBundlesAndFallback", "TestResourceAndParameterDiagnostics", "TestStrictLoadingAndExplicitOverlay", "TestPinnedCLDRSamples", "TestBrowserConformance", "TestSourceLocaleCanBeFallbackOnly", "TestTypeScriptConsumer"},
|
||||
"i18n/httpi18n": {"TestNegotiationAndExplicitResolvers"},
|
||||
@@ -176,6 +183,11 @@ def main():
|
||||
checks += ["remotePreciseModuleDiagnostics", "remoteJSONSuccessStatuses", "remoteRejectedStatusBodyCleanup",
|
||||
"remoteCopyTypePreflight", "remoteEmptyCollectionTypeRejection", "remoteLoggerTimeZoneIsolation",
|
||||
"remoteCORSPolicyRace", "remoteCORSErrorResponses", "remoteCORSTransports", "generatedCORSConfigOnly"]
|
||||
run(["go", "build", "-o", temp / "httpSafety", MODULE + "/examples/httpSafety"], starter, env)
|
||||
checks += ["remoteUnifiedBodyPolicyTransports", "remoteBodyLimitsBeforeMiddleware", "remoteHardLimitConflictBeforeServing",
|
||||
"remoteCanceledHandlerCleanupBarrier", "remoteExternalServingResourceRetention", "remoteAdmissionCloseRace",
|
||||
"remoteRuntimeFinishRetainsHTTPDependencies", "remoteHTTP2DisconnectedHandlerDrain",
|
||||
"remoteNativeEarly413AndBusinessHTTP200", "remoteHTTPSafetyExampleBuild"]
|
||||
transaction_output = run(["go", "run", MODULE + "/examples/transactions"], starter, env)
|
||||
assert "commitAcknowledgementLost: stage=commit outcome=unknown stored=true cleanup=false" in transaction_output
|
||||
assert "workFailure: stage=work outcome=rolledBack stored=false cleanup=true" in transaction_output
|
||||
@@ -231,6 +243,9 @@ def main():
|
||||
sources += ["database/transaction.go", "database/transaction_test.go", "database/transactionResponse_test.go", "database/mysqlConnector_test.go", "examples/transactions/main.go", "examples/transactions/demoDriver.go"]
|
||||
sources += [str(p.relative_to(ROOT)) for directory in ("i18n", "examples/i18n") for p in (ROOT / directory).rglob("*") if p.is_file()]
|
||||
sources += ["app/i18n_test.go", "errors/localization.go", "cmd/go-omni/i18n.go", "cmd/go-omni/apigen/i18nVersion.go", "cmd/go-omni/apigen/testdata/i18n/consumer_test.go", "contract/policy_test.go"]
|
||||
sources += ["app/body.go", "app/router.go", "app/requests.go", "app/bodyPolicy_test.go", "app/requests_test.go", "app/transport_test.go",
|
||||
"httpx/context.go", "internal/httplimits/limits.go", "lifecycle/runtime_test.go", "transport/http2/server.go", "transport/http2/lifecycle_test.go",
|
||||
"transport/native/server.go", "transport/native/earlyReject_test.go", "examples/httpSafety/main.go"]
|
||||
report = {"passed": True, "frameworkVersion": args.version, "frameworkSum": dependency["Sum"],
|
||||
"freshModuleCache": True, "noLocalReplace": True,
|
||||
"durationSeconds": round(time.monotonic() - started, 2), "checks": checks,
|
||||
|
||||
@@ -118,7 +118,7 @@ python3 scripts/verifyTransports.py --version v0.6.0-preview
|
||||
|
||||
第一条验证标准库适配器复用完整 Handler / Service / Repository。第二条在源码目录外创建无 replace 的消费模块,从远程版本导入同一套验收包;自研适配器也应使用同一个公共测试包。
|
||||
|
||||
当前替换边界分成两层:`transport/nethttp` 提供标准库路由,`transport/native` 进一步接管 TCP 监听、连接管理和 HTTP/1.1 协议 I/O。`App.Run` / `App.RunContext` 仍使用标准库 `net/http.Server`,需要显式调用 `App.RunWith(ctx, native.New(nil), address)` 才会启用 native 预览实现。native 当前通过标准 `http.Request` / `http.ResponseWriter` 桥接到业务层,后续可在不改 Handler / Service 的前提下逐步替换解析器和响应写回。
|
||||
当前替换边界分成两层:`transport/nethttp` 提供标准库路由,`transport/native` 进一步接管 TCP 监听、连接管理和 HTTP/1.1 协议 I/O。`App.Run` / `App.RunContext` / `ServeContext` 使用配置选中的 transport,默认 nethttp;推荐 `WithTransport(TransportNative)` 启用 native,框架会统一装配限额和 HTTP 配置。自定义 Runner 才使用 `RunWith`。native 通过标准 `http.Request` / `http.ResponseWriter` 桥接到业务层。
|
||||
|
||||
## HTTP 配置(v0.9.1-preview)
|
||||
|
||||
@@ -152,4 +152,28 @@ I/O 超时不会强制终止 CPU 计算或数据库调用;业务和下游调
|
||||
|
||||
YAML/JSON 字段与上述字段对应的 lowerCamelCase 一致,位于 `config.Config` 的 `http` 对象。脚手架将框架配置包在 `server` 下,因此生成项目填写 `server.http`;直接用 `config.LoadFile` 加载框架配置时填写根级 `http`。环境变量由文件显式声明,例如 `readTimeout: "${HTTP_READ_TIMEOUT:-30s}"`,不再按字段自动覆盖。`omni.New` 不自动读取文件;先 `config.LoadFile` / `config.LoadProject`,再 `WithConfig`。Go-only 的 `ErrorLog` 在文件加载之后设置。
|
||||
|
||||
`NewServer` 始终返回普通 `*http.Server`,应用 HTTP 字段但不负责加载 TLS、监听、退出或资源清理;非法 HTTP 值会 panic。`RunContext` / `ServeContext` 的非法配置则作为 error 返回并关闭 App 资源。手动修改返回服务器的字段仍然有效。项目自己的 Runner 通过 `RunWith` 接入时,配置由该 Runner 持有,这组 HTTP 配置不会覆盖它。
|
||||
`NewServer` 始终返回普通 `*http.Server`,应用 HTTP 字段但不负责加载 TLS、监听、退出或资源清理;非法 HTTP 值或路由限额冲突会 panic。`RunContext` / `ServeContext` 的非法配置则作为 error 返回并关闭 App 资源。手动修改返回服务器的字段仍然有效。`RunWith` 的网络配置由 Runner 持有;App 自身的请求体策略和取消收尾窗口仍生效,框架无法自动移除自定义 Runner、反向代理或业务中间件的其他限额。
|
||||
|
||||
## 统一请求体限额
|
||||
|
||||
三种内置 transport 通过 App 共用下列配置,全部以字节为单位,范围 0..1 TiB:
|
||||
|
||||
```yaml
|
||||
server:
|
||||
http:
|
||||
defaultMaxBodyBytes: 1048576 # 普通 App 路由默认 1 MiB
|
||||
maxBodyBytes: 20971520 # 所有请求硬上限 20 MiB
|
||||
shutdownCancelTimeout: 5s
|
||||
```
|
||||
|
||||
- `DefaultMaxBodyBytes`:未显式声明限额的 App/RouterGroup 路由默认值;零为 8 MiB,如果硬上限更小则取硬上限。显式默认值超过硬上限会报配置错误。
|
||||
- `HandleWithBodyLimit` / `.api @request(maxBytes: ...)`:覆盖普通路由默认值。生成契约既有默认 1 MiB 保持;服务默认值不会扩大契约限制。
|
||||
- `MaxBodyBytes`:服务硬上限;零仅保留框架可支持的 1 TiB 上限,不表示请求会预分配相应内存。路由超过硬上限会在监听前报错,包含完整路由、路由限额、服务限额。`RunWith` 也检查冲突;`NewServer` 在创建时检查,外部宿主在注册完成后调用 `app.ValidateHTTP()` 可预检;未经预检直接调用 App/Engine 时,首次请求拒绝非法配置。
|
||||
|
||||
App 在中间件读取请求之前安装流式 reader 限额,支持已知长度和未知长度。native/HTTP2 在 App 装配下只保留 1 TiB 协议上界,原来的隐含 8 MiB 不再挡住放宽的上传路由。请求进入 App 后使用 `BODY_TOO_LARGE` 错误路径;服务硬上限提前拒绝时使用 App 响应策略,路由限额使用该路由策略。native 提前拒绝后先发送响应再半关闭写方向,最多等待 500ms 避免未读上传触发 TCP reset 吞掉提示;不会等待上传完成。错误目录将 HTTP 状态映射为 200 时仍保留请求体拒绝语义。已提交的流式响应不能撤回;处理器应先完成有界读取/校验再执行不可逆业务。
|
||||
|
||||
底层协议错误(例如非法 framing 或超过 1 TiB)可能在路由前直接拒绝。直接使用 `native.Server` / `http2.Server` 的 `MaxBodyBytes` 默认仍为 8 MiB,由该服务器的拥有者配置;通过 `RunWith` 传入这些服务器也需自行协调其限额,推荐使用 `WithTransport` 获取统一装配。高级 `Router()` 原始注册绕过 App 路由默认值,但经 App/Engine 服务时仍有服务硬限额;直接调用 Router.ServeHTTP 会绕过整个 App 策略。
|
||||
|
||||
multipart 的整个请求包含文件、字段和边界,`MaxFileBytes` 与请求体上限分别控制;幂等中间件 `MaxRequestBytes` 控制缓存量,不随路由自动放大。代理层限额继续在代理配置中维护。框架不猜测业务文件规格或代理策略。
|
||||
|
||||
迁移:旧 nethttp 手写路由此前无统一默认 reader 限额,现在默认 8 MiB。大请求用路由覆盖或显式默认值;已有生成接口保留契约限额。示例见 [httpSafety](../examples/httpSafety/README.md)。
|
||||
|
||||
@@ -247,3 +247,29 @@ func TestCanceledServeContextWaitsForShutdownDeadline(t *testing.T) {
|
||||
}
|
||||
waitSignal(t, canceled)
|
||||
}
|
||||
|
||||
func TestShutdownWaitsForHandlerAfterClientDisconnect(t *testing.T) {
|
||||
started, canceled, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{}), make(chan struct{})
|
||||
s := New(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
|
||||
defer close(finished)
|
||||
close(started)
|
||||
<-r.Context().Done()
|
||||
close(canceled)
|
||||
<-release
|
||||
}))
|
||||
defer func() { close(release); waitSignal(t, finished) }()
|
||||
address, done := launchServer(t, s, context.Background())
|
||||
p := newFramePeer(t, address)
|
||||
p.get(t, 1, "/hold")
|
||||
waitSignal(t, started)
|
||||
p.Close()
|
||||
waitSignal(t, canceled)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond)
|
||||
defer cancel()
|
||||
if err := s.Shutdown(ctx); !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("disconnected handler reported as drained: %v", err)
|
||||
}
|
||||
// Serve can finish its bounded shutdown, but successful Shutdown must never
|
||||
// imply that a disconnected, still-running handler has completed.
|
||||
_ = serveResult(t, done)
|
||||
}
|
||||
|
||||
@@ -64,6 +64,7 @@ type Server struct {
|
||||
activeConns atomic.Int64
|
||||
activeRequests atomic.Int64
|
||||
wg sync.WaitGroup
|
||||
requestWG sync.WaitGroup
|
||||
}
|
||||
|
||||
// Stats is a point-in-time view of HTTP/2 transport admission and lifecycle.
|
||||
@@ -390,7 +391,8 @@ func (s *Server) tlsHandshakeTimeout() time.Duration {
|
||||
return defaultTLSHandshake
|
||||
}
|
||||
|
||||
// Shutdown stops accepting connections and waits for active HTTP/2 sessions.
|
||||
// Shutdown stops accepting connections/requests and waits for both HTTP/2
|
||||
// sessions and admitted handlers, even when the client has already disconnected.
|
||||
// If the context expires, remaining connections are closed so handlers can
|
||||
// observe cancellation through their request contexts.
|
||||
func (s *Server) Shutdown(ctx context.Context) error {
|
||||
@@ -436,6 +438,7 @@ func (s *Server) Shutdown(ctx context.Context) error {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
s.wg.Wait()
|
||||
s.requestWG.Wait()
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
@@ -476,6 +479,17 @@ type requestGate struct {
|
||||
}
|
||||
|
||||
func (g *requestGate) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
|
||||
// Synchronize admission with Shutdown before it starts waiting. Connection
|
||||
// completion alone does not join x/net/http2's separate handler goroutines.
|
||||
g.server.mu.Lock()
|
||||
if g.server.closing {
|
||||
g.server.mu.Unlock()
|
||||
http.Error(writer, "server shutting down", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
g.server.requestWG.Add(1)
|
||||
g.server.mu.Unlock()
|
||||
defer g.server.requestWG.Done()
|
||||
if g.slots != nil {
|
||||
select {
|
||||
case g.slots <- struct{}{}:
|
||||
|
||||
@@ -2,6 +2,7 @@ package native_test
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -11,6 +12,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
omni "git.yuntuo.hk/k2safe/go-omni"
|
||||
apierrors "git.yuntuo.hk/k2safe/go-omni/errors"
|
||||
"git.yuntuo.hk/k2safe/go-omni/transport/native"
|
||||
)
|
||||
|
||||
@@ -58,3 +61,41 @@ func TestApplicationRejectsUnreadBodyPromptly(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBodyRejectionWithBusinessHTTP200DoesNotDrain(t *testing.T) {
|
||||
app := omni.New(omni.WithErrors(apierrors.Catalog{DefaultStatus: 200, Definitions: map[string]apierrors.Definition{
|
||||
"BODY_TOO_LARGE": {Code: "LIMIT_REJECTED", Message: "body limit exceeded"},
|
||||
}}))
|
||||
app.HandleWithBodyLimit("POST", "/upload", 32, nil, func(*omni.Context) error { t.Error("oversize reached handler"); return nil })
|
||||
server := native.New(app)
|
||||
server.ReadTimeout = 10 * time.Second
|
||||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- server.Serve(context.Background(), l) }()
|
||||
defer stopServer(t, server, l, done)
|
||||
defer app.Close()
|
||||
c, err := net.DialTimeout("tcp", l.Addr().String(), time.Second)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer c.Close()
|
||||
c.SetDeadline(time.Now().Add(time.Second))
|
||||
// Intentionally withhold every body byte. The rejection must be sent before
|
||||
// any attempted drain, despite the project mapping its HTTP status to 200.
|
||||
_, err = io.WriteString(c, "POST /upload HTTP/1.1\r\nHost: local\r\nContent-Length: 1024\r\n\r\n")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
res, err := http.ReadResponse(bufio.NewReader(c), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := io.ReadAll(res.Body)
|
||||
res.Body.Close()
|
||||
if err != nil || res.StatusCode != 200 || !res.Close || !bytes.Contains(data, []byte("LIMIT_REJECTED")) {
|
||||
t.Fatalf("status=%d close=%v body=%s err=%v", res.StatusCode, res.Close, data, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -420,7 +420,9 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
|
||||
status = protocolErr.status
|
||||
message = protocolErr.message + "\n"
|
||||
}
|
||||
_ = writeSimple(state.conn, status, message, true, s.WriteTimeout)
|
||||
if writeSimple(state.conn, status, message, true, s.WriteTimeout) == nil && status == http.StatusRequestEntityTooLarge {
|
||||
closeRejectedBody(baseCtx, state.conn)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -482,13 +484,15 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
|
||||
}
|
||||
handler.ServeHTTP(writer, request)
|
||||
var bodyErr error
|
||||
if body, ok := request.Body.(*requestBody); ok && !body.done && body.err == nil && writer.status == http.StatusRequestEntityTooLarge {
|
||||
rejectedBody := false
|
||||
if body, ok := request.Body.(*requestBody); ok && !body.done && body.err == nil && (writer.bodyTooLarge || writer.status == http.StatusRequestEntityTooLarge) {
|
||||
// An application can reject size from the headers alone when its
|
||||
// body limit is below the transport limit. Waiting to
|
||||
// drain a slow or absent body would withhold the rejection response.
|
||||
// Abandon it and close; unread bytes can never become a next request.
|
||||
body.closed = true
|
||||
closeAfter = true
|
||||
rejectedBody = true
|
||||
} else {
|
||||
bodyErr = request.Body.Close()
|
||||
}
|
||||
@@ -538,11 +542,30 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
|
||||
s.writerPool.Put(writer)
|
||||
shouldClose = s.finishRequest(state) || shouldClose
|
||||
if shouldClose {
|
||||
if rejectedBody {
|
||||
closeRejectedBody(baseCtx, state.conn)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Half-close after flushing a rejection, then allow the peer to consume it
|
||||
// before closing a socket with unread upload data. As in net/http, immediate
|
||||
// full close can cause a TCP reset to hide the response on BSD systems. This
|
||||
// never waits for the body and cancellation bounds the additional grace.
|
||||
func closeRejectedBody(ctx context.Context, conn net.Conn) {
|
||||
if writer, ok := conn.(interface{ CloseWrite() error }); ok {
|
||||
_ = writer.CloseWrite()
|
||||
}
|
||||
timer := time.NewTimer(500 * time.Millisecond)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-timer.C:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}
|
||||
|
||||
// Publish the idle transition under the same lock that starts shutdown. Either
|
||||
// Shutdown observes an idle connection and closes it, or this request observes
|
||||
// shutdown and closes it instead of starting another Keep-Alive read.
|
||||
@@ -1055,6 +1078,7 @@ func writeSimple(conn net.Conn, status int, body string, closeAfter bool, timeou
|
||||
|
||||
type responseWriter struct {
|
||||
header http.Header
|
||||
bodyTooLarge bool
|
||||
status int
|
||||
wroteHeader bool
|
||||
body bytes.Buffer
|
||||
@@ -1081,6 +1105,7 @@ func (w *responseWriter) reset(method string) {
|
||||
}
|
||||
w.status = 0
|
||||
w.wroteHeader = false
|
||||
w.bodyTooLarge = false
|
||||
if w.body.Cap() > maxPooledResponseBytes {
|
||||
w.body = bytes.Buffer{}
|
||||
} else {
|
||||
@@ -1110,6 +1135,10 @@ func (w *responseWriter) configure(request *http.Request, conn net.Conn, closeAf
|
||||
|
||||
func (w *responseWriter) Header() http.Header { return w.header }
|
||||
|
||||
// RequestBodyTooLarge preserves semantic body rejection independently from a
|
||||
// project's chosen HTTP response status. Called through the optional writer hook.
|
||||
func (w *responseWriter) RequestBodyTooLarge() { w.bodyTooLarge = true }
|
||||
|
||||
func (w *responseWriter) WriteHeader(status int) {
|
||||
if w.wroteHeader {
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user