From 3b00bb55d189cbe480c7d15c8d14bcd09d3f2d52 Mon Sep 17 00:00:00 2001 From: k2safe Date: Tue, 29 Sep 2026 23:24:19 +0800 Subject: [PATCH] feat: release v0.27 unified HTTP limits and safe request shutdown --- CHANGELOG.md | 8 + DEVELOPMENT.md | 2 +- PROJECT.md | 8 +- README.md | 21 +- RELEASE.md | 19 ++ app/app.go | 69 +++++- app/body.go | 63 ++++- app/bodyPolicy_test.go | 224 ++++++++++++++++++ app/requests.go | 88 +++++++ app/requests_test.go | 167 +++++++++++++ app/router.go | 15 +- app/transport.go | 12 +- app/transport_test.go | 15 +- cmd/go-omni/README.md | 8 +- cmd/go-omni/apigen/generate.go | 2 +- cmd/go-omni/apigen/templates.go | 6 +- cmd/go-omni/apigen/testdata/managed/client.go | 1 + .../apigen/testdata/managed/operations.go | 1 + .../apigen/testdata/managed/shutdown_test.go | 78 ++++++ cmd/go-omni/scaffold/version.go | 2 +- config/config.go | 3 +- config/http.go | 33 ++- config/http_test.go | 37 +++ examples/httpSafety/README.md | 15 ++ examples/httpSafety/main.go | 62 +++++ .../managedService/.goOmni/apiManifest.json | 2 +- examples/managedService/go.mod | 2 +- examples/managedService/main.go | 6 +- examples/userService/go.mod | 2 +- httpx/context.go | 6 + internal/httplimits/limits.go | 25 ++ lifecycle/README.md | 22 +- lifecycle/phases.go | 4 +- lifecycle/runtime.go | 12 +- lifecycle/runtime_test.go | 25 ++ reports/httpSafetyAcceptance.md | 33 +++ scripts/verifyManagedRelease.py | 19 +- transport/README.md | 28 ++- transport/http2/lifecycle_test.go | 26 ++ transport/http2/server.go | 16 +- transport/native/earlyReject_test.go | 41 ++++ transport/native/server.go | 33 ++- 42 files changed, 1191 insertions(+), 70 deletions(-) create mode 100644 app/bodyPolicy_test.go create mode 100644 app/requests.go create mode 100644 app/requests_test.go create mode 100644 examples/httpSafety/README.md create mode 100644 examples/httpSafety/main.go create mode 100644 internal/httplimits/limits.go create mode 100644 reports/httpSafetyAcceptance.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 63fd254..f707188 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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、业务或数据库。 diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index e882cf2..5b3176a 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -1,4 +1,4 @@ -# 通用业务开发指南（v0.26.0-preview） +# 通用业务开发指南（v0.27.0-preview） 从 [README](README.md) 创建服务后，业务主要维护 `.api`、配置、`svc.Init`、Logic、Repository 和项目中间件。本页说明通用接入能力；字段、错误码、权限、表结构和事务边界都由项目定义。 diff --git a/PROJECT.md b/PROJECT.md index 0b20711..a3f6b72 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -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 和提前返回的中间件一起验收。 diff --git a/README.md b/README.md index 9c1c84c..5dff0c6 100644 --- a/README.md +++ b/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)。 ## 按需深入 diff --git a/RELEASE.md b/RELEASE.md index b8799a3..3508a95 100644 --- a/RELEASE.md +++ b/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/错误适配。项目维护文案和语言设置，框架提供不可变词典、语言匹配、回退、参数与分支、检查和导出；没有绑定业务身份、页面框架、数据库或响应字段名。 diff --git a/app/app.go b/app/app.go index 80be03c..e2a652d 100644 --- a/app/app.go +++ b/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 { diff --git a/app/body.go b/app/body.go index cae0eb8..114d352 100644 --- a/app/body.go +++ b/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}) } diff --git a/app/bodyPolicy_test.go b/app/bodyPolicy_test.go new file mode 100644 index 0000000..69b47f4 --- /dev/null +++ b/app/bodyPolicy_test.go @@ -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()) + } + } +} diff --git a/app/requests.go b/app/requests.go new file mode 100644 index 0000000..585ef42 --- /dev/null +++ b/app/requests.go @@ -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) +} diff --git a/app/requests_test.go b/app/requests_test.go new file mode 100644 index 0000000..31e235b --- /dev/null +++ b/app/requests_test.go @@ -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() + } +} diff --git a/app/router.go b/app/router.go index e196d37..4a4f836 100644 --- a/app/router.go +++ b/app/router.go @@ -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...)) } diff --git a/app/transport.go b/app/transport.go index 064adda..1e5e830 100644 --- a/app/transport.go +++ b/app/transport.go @@ -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 diff --git a/app/transport_test.go b/app/transport_test.go index 0f03245..3bb7751 100644 --- a/app/transport_test.go +++ b/app/transport_test.go @@ -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): diff --git a/cmd/go-omni/README.md b/cmd/go-omni/README.md index 51b855e..999b8d6 100644 --- a/cmd/go-omni/README.md +++ b/cmd/go-omni/README.md @@ -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` 命令的手动路由装配。 ## 数据库迁移命令 diff --git a/cmd/go-omni/apigen/generate.go b/cmd/go-omni/apigen/generate.go index 97cf376..6d5ac75 100644 --- a/cmd/go-omni/apigen/generate.go +++ b/cmd/go-omni/apigen/generate.go @@ -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 { diff --git a/cmd/go-omni/apigen/templates.go b/cmd/go-omni/apigen/templates.go index 8b2ad73..1762c9f 100644 --- a/cmd/go-omni/apigen/templates.go +++ b/cmd/go-omni/apigen/templates.go @@ -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())}() diff --git a/cmd/go-omni/apigen/testdata/managed/client.go b/cmd/go-omni/apigen/testdata/managed/client.go index 22d3b61..0a1b307 100644 --- a/cmd/go-omni/apigen/testdata/managed/client.go +++ b/cmd/go-omni/apigen/testdata/managed/client.go @@ -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] diff --git a/cmd/go-omni/apigen/testdata/managed/operations.go b/cmd/go-omni/apigen/testdata/managed/operations.go index 9d97873..680c7f4 100644 --- a/cmd/go-omni/apigen/testdata/managed/operations.go +++ b/cmd/go-omni/apigen/testdata/managed/operations.go @@ -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 diff --git a/cmd/go-omni/apigen/testdata/managed/shutdown_test.go b/cmd/go-omni/apigen/testdata/managed/shutdown_test.go index df2eda9..c821072 100644 --- a/cmd/go-omni/apigen/testdata/managed/shutdown_test.go +++ b/cmd/go-omni/apigen/testdata/managed/shutdown_test.go @@ -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) + }) + } +} diff --git a/cmd/go-omni/scaffold/version.go b/cmd/go-omni/scaffold/version.go index d8f7cb4..912c6d0 100644 --- a/cmd/go-omni/scaffold/version.go +++ b/cmd/go-omni/scaffold/version.go @@ -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. diff --git a/config/config.go b/config/config.go index 9670c1f..f7ee3e6 100644 --- a/config/config.go +++ b/config/config.go @@ -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 { diff --git a/config/http.go b/config/http.go index 1d7b460..dcd6d89 100644 --- a/config/http.go +++ b/config/http.go @@ -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 } } diff --git a/config/http_test.go b/config/http_test.go index 82041f0..d6b68f6 100644 --- a/config/http_test.go +++ b/config/http_test.go @@ -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) + } + } +} diff --git a/examples/httpSafety/README.md b/examples/httpSafety/README.md new file mode 100644 index 0000000..6dbb20c --- /dev/null +++ b/examples/httpSafety/README.md @@ -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。 diff --git a/examples/httpSafety/main.go b/examples/httpSafety/main.go new file mode 100644 index 0000000..2629361 --- /dev/null +++ b/examples/httpSafety/main.go @@ -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) +} diff --git a/examples/managedService/.goOmni/apiManifest.json b/examples/managedService/.goOmni/apiManifest.json index 020ff8b..c18966c 100644 --- a/examples/managedService/.goOmni/apiManifest.json +++ b/examples/managedService/.goOmni/apiManifest.json @@ -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": { diff --git a/examples/managedService/go.mod b/examples/managedService/go.mod index d51b964..9d1aff3 100644 --- a/examples/managedService/go.mod +++ b/examples/managedService/go.mod @@ -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 diff --git a/examples/managedService/main.go b/examples/managedService/main.go index 4c281f5..305ca0d 100644 --- a/examples/managedService/main.go +++ b/examples/managedService/main.go @@ -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 { diff --git a/examples/userService/go.mod b/examples/userService/go.mod index 7c6d1c4..e2febf3 100644 --- a/examples/userService/go.mod +++ b/examples/userService/go.mod @@ -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 ) diff --git a/httpx/context.go b/httpx/context.go index d4a8525..7fa6f8c 100644 --- a/httpx/context.go +++ b/httpx/context.go @@ -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, diff --git a/internal/httplimits/limits.go b/internal/httplimits/limits.go new file mode 100644 index 0000000..cdaca70 --- /dev/null +++ b/internal/httplimits/limits.go @@ -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() + } +} diff --git a/lifecycle/README.md b/lifecycle/README.md index ba386ba..aca72d0 100644 --- a/lifecycle/README.md +++ b/lifecycle/README.md @@ -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。 diff --git a/lifecycle/phases.go b/lifecycle/phases.go index d2c5cd3..5d6ee1a 100644 --- a/lifecycle/phases.go +++ b/lifecycle/phases.go @@ -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"` diff --git a/lifecycle/runtime.go b/lifecycle/runtime.go index 3c995c7..cde1700 100644 --- a/lifecycle/runtime.go +++ b/lifecycle/runtime.go @@ -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) } diff --git a/lifecycle/runtime_test.go b/lifecycle/runtime_test.go index 00c9180..3fbf060 100644 --- a/lifecycle/runtime_test.go +++ b/lifecycle/runtime_test.go @@ -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) + } +} diff --git a/reports/httpSafetyAcceptance.md b/reports/httpSafetyAcceptance.md new file mode 100644 index 0000000..1d7cf6b --- /dev/null +++ b/reports/httpSafetyAcceptance.md @@ -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 单文件和幂等缓存限额保持各自用途。已发送的流式响应无法撤回，先完成有界读取再执行业务。 diff --git a/scripts/verifyManagedRelease.py b/scripts/verifyManagedRelease.py index 8e9d460..9cf8d92 100644 --- a/scripts/verifyManagedRelease.py +++ b/scripts/verifyManagedRelease.py @@ -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, diff --git a/transport/README.md b/transport/README.md index a2bfc33..0e92880 100644 --- a/transport/README.md +++ b/transport/README.md @@ -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)。 diff --git a/transport/http2/lifecycle_test.go b/transport/http2/lifecycle_test.go index 10adcde..e48156f 100644 --- a/transport/http2/lifecycle_test.go +++ b/transport/http2/lifecycle_test.go @@ -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) +} diff --git a/transport/http2/server.go b/transport/http2/server.go index 939229d..b7f30f1 100644 --- a/transport/http2/server.go +++ b/transport/http2/server.go @@ -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{}{}: diff --git a/transport/native/earlyReject_test.go b/transport/native/earlyReject_test.go index dd59eed..83ea290 100644 --- a/transport/native/earlyReject_test.go +++ b/transport/native/earlyReject_test.go @@ -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) + } +} diff --git a/transport/native/server.go b/transport/native/server.go index e5e784c..0a61c91 100644 --- a/transport/native/server.go +++ b/transport/native/server.go @@ -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