From 22f1c9ebfb982d615af064f099c9ff19fad7fca8 Mon Sep 17 00:00:00 2001 From: k2safe Date: Mon, 21 Sep 2026 03:47:55 +0800 Subject: [PATCH] feat: support bounded chunked request bodies --- CHANGELOG.md | 3 +- README.md | 2 +- ROADMAP.md | 2 +- transport/README.md | 2 +- transport/native/server.go | 133 +++++++++++++++++++++++++++++--- transport/native/server_test.go | 79 ++++++++++++++++--- 6 files changed, 195 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9cdf10b..fb1aa4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - 统一标准库与 native 的真实 TCP benchmark，增加固定并发档位、响应状态与读写错误检查。 - 新增独立进程并发与长连接内存测量工具，记录成功吞吐、P95/P99、服务端分配、空闲连接堆内存和回收情况；修正并行 `ns/op` 与客户端/服务端混合分配的解释。 - 加固 native HTTP/1.1 解析边界：校验 Host、重复 Content-Length、Connection token、Transfer-Encoding、控制字符和响应头名称；补充 HTTP/1.0 Keep-Alive、HEAD、204 以及异常请求测试。 +- native 支持带 trailers 的 chunked 请求体，并继续通过 `MaxBodyBytes`、读超时和头部上限控制解析内存与连接占用。 - 补充慢请求体超时、大请求体和同连接管线请求验收，确认请求体未完整读取时不会进入业务 Handler，连续请求不会串线。 - 补充 `MaxConnections` 满载拒绝与槽位释放验收，确认请求完成后连接容量可再次使用。 - 增加 8 并发、800 次建连/断连的状态池 churn 验收，覆盖连接对象回收与容量复用。 @@ -13,7 +14,7 @@ - 新增 `transport/native` 第一版自研 HTTP/1.1 服务端，接管 TCP 监听、连接、Keep-Alive、请求解析、响应写回和优雅关闭。 - 新增 `App.RunWith` / `omni.Runner` 启动边界；native 通过标准 `http.Handler` 接入，现有 Handler、Service 和 Repository 无需修改。 -- native 支持 `Content-Length` 请求体、请求头 / 请求体限制、读写超时和服务关闭时的请求取消；通过对象池和 `MaxConnections` / `MaxConcurrentRequests` 控制高并发内存；chunked 请求体暂返回 501。 +- native 支持 `Content-Length` / chunked 请求体、请求头 / 请求体限制、读写超时和服务关闭时的请求取消；通过对象池和 `MaxConnections` / `MaxConcurrentRequests` 控制高并发内存。 - `omni.New()` 默认仍使用只依赖标准库的 `transport/nethttp` 路由，native 作为显式启用的网络层预览实现。 ## v0.4.0 diff --git a/README.md b/README.md index 2523b03..12ad7ef 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,7 @@ if err := app.RunWith(ctx, native.New(nil), ":8080"); err != nil { } ``` -`transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 请求体、读写超时和优雅关闭；chunked 请求体暂返回 501。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。 +`transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 和 chunked 请求体、读写超时和优雅关闭；不支持的传输编码会被拒绝。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。 业务代码可以直接使用 `git.yuntuo.hk/k2safe/go-omni/logx` 打结构化日志，不需要依赖具体网络库： diff --git a/ROADMAP.md b/ROADMAP.md index 2b5317c..1805827 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -41,7 +41,7 @@ - 新增 `transport/native` 第一版自研 HTTP/1.1 服务端：自行管理 TCP listener、Keep-Alive、Content-Length 请求解析、响应写回、读写超时和优雅关闭；通过 `App.RunWith` 接入，业务 Handler / Service 保持不变。 - 标准库与 native 共用 TCP benchmark，固定 1 / 32 / 128 并发；相同应用、客户端、超时和响应检查，避免两个测试入口逐渐失去可比性。 - 新增独立服务进程压测 `internal/perfcheck`，记录成功吞吐、P50/P95/P99、服务端分配、采样堆峰值，以及 128 / 1,024 条空闲长连接的堆内存、goroutine 和断开后的回收情况。原始报告保存在 `performance/tcpBaseline.json`。 -- native 当前拒绝 chunked 请求体，仍使用标准 `http.Request` / `http.ResponseWriter` 作为桥接；解析器、连接抽象和生产压测将在后续阶段继续下沉。 +- native 已支持受 `MaxBodyBytes` 约束的 `Content-Length` / chunked 请求体，仍使用标准 `http.Request` / `http.ResponseWriter` 作为桥接；解析器、连接抽象和生产压测将在后续阶段继续下沉。 - 已补充慢请求体超时、大请求体、HTTP/1.0 Keep-Alive、HEAD/204 和管线请求验收；Linux、慢客户端规模、TLS 和持续运行压测仍待在目标环境验证。 - 已验证 `MaxConnections` 满载时快速拒绝新连接、请求完成后释放容量；后续仍需在目标部署环境观察文件描述符、内核 socket 缓冲和反向代理联动。 - 已用 8 个并发 worker 完成 800 次建连/断连 race 验收；真实部署仍需补文件描述符和跨平台 socket 行为观测。 diff --git a/transport/README.md b/transport/README.md index 8ad13cc..649b2a0 100644 --- a/transport/README.md +++ b/transport/README.md @@ -14,7 +14,7 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码 `transport/nethttp` 是当前内置的标准库路由实现。它提供一套可移植的路径、方法和错误回退规则；自研网络层可以直接实现同一个 `httpx.Router` 契约。 -`transport/native` 是第一版自研 HTTP/1.1 服务端：它自己管理 TCP listener、连接、Keep-Alive、请求解析、响应写回和优雅关闭，再把请求交给现有 `http.Handler`。当前只支持 `Content-Length` 请求体，chunked 请求会返回 501；它仍使用标准 `http.Request` / `http.ResponseWriter` 作为业务桥接，因此不会影响现有 Handler / Service。 +`transport/native` 是第一版自研 HTTP/1.1 服务端：它自己管理 TCP listener、连接、Keep-Alive、请求解析、响应写回和优雅关闭，再把请求交给现有 `http.Handler`。请求体支持 `Content-Length` 和 chunked，并统一受 `MaxBodyBytes` 限制；其他传输编码会被拒绝。它仍使用标准 `http.Request` / `http.ResponseWriter` 作为业务桥接，因此不会影响现有 Handler / Service。 服务端不是把连接交给一个全局串行 worker，而是每条连接独立处理请求；Keep-Alive 负责复用客户端连接，`Server` 内部的 `sync.Pool` 负责复用小型 Reader、响应对象和连接状态。Reader 默认只有 4 KiB，需要处理大请求头时按需增长。需要保护内存上限时，可以设置 `MaxConnections` 和 `MaxConcurrentRequests`；超出连接上限的连接会被关闭，超出请求并发上限的请求返回 503，不会无限排队。 diff --git a/transport/native/server.go b/transport/native/server.go index 206a062..82fde3b 100644 --- a/transport/native/server.go +++ b/transport/native/server.go @@ -2,8 +2,8 @@ // // It deliberately keeps the public handler boundary as net/http so existing // GoOmni business code remains unchanged while connection management and -// protocol I/O move out of net/http.Server. The parser currently supports -// Content-Length request bodies and rejects chunked requests. +// protocol I/O move out of net/http.Server. The parser supports Content-Length +// and chunked request bodies with configurable size limits. package native import ( @@ -493,14 +493,23 @@ func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBod return nil, true, &protocolError{status: http.StatusExpectationFailed, message: "unsupported expectation"} } } + chunked := false for _, value := range header.Values("Transfer-Encoding") { for _, coding := range strings.Split(value, ",") { coding = strings.TrimSpace(coding) - if coding != "" && !strings.EqualFold(coding, "identity") { - return nil, true, &protocolError{status: http.StatusNotImplemented, message: "transfer encoding is not supported"} + if coding == "" || strings.EqualFold(coding, "identity") { + continue } + if strings.EqualFold(coding, "chunked") && !chunked { + chunked = true + continue + } + return nil, true, &protocolError{status: http.StatusNotImplemented, message: "transfer encoding is not supported"} } } + if chunked && len(header.Values("Content-Length")) > 0 { + return nil, true, &protocolError{status: http.StatusBadRequest, message: "content length with chunked encoding"} + } contentLength := int64(0) contentLengthSeen := false for _, value := range header.Values("Content-Length") { @@ -514,17 +523,31 @@ func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBod contentLengthSeen = true } } - if contentLength > maxBodyBytes { + if !chunked && contentLength > maxBodyBytes { return nil, true, &protocolError{status: http.StatusRequestEntityTooLarge, message: "request body too large"} } - body := make([]byte, contentLength) - if contentLength > 0 { - if setBodyDeadline != nil { - setBodyDeadline() - } - if _, err := io.ReadFull(reader, body); err != nil { + var body []byte + var trailer http.Header + if chunked { + body, trailer, err = readChunkedBody(reader, maxBodyBytes, setBodyDeadline, readLine) + if err != nil { return nil, true, err } + // net/http uses -1 to indicate that the decoded length was not known + // from the request headers. The body is already buffered for the + // handler, but retaining this value keeps the request contract + // compatible with the standard server. + contentLength = -1 + } else { + body = make([]byte, contentLength) + if contentLength > 0 { + if setBodyDeadline != nil { + setBodyDeadline() + } + if _, err := io.ReadFull(reader, body); err != nil { + return nil, true, err + } + } } request := &http.Request{ Method: parts[0], @@ -538,6 +561,10 @@ func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBod Host: header.Get("Host"), RequestURI: target, RemoteAddr: conn.RemoteAddr().String(), + Trailer: trailer, + } + if chunked { + request.TransferEncoding = []string{"chunked"} } closeAfter := headerHasToken(header.Values("Connection"), "close") || (protoMajor == 1 && protoMinor == 0 && !headerHasToken(header.Values("Connection"), "keep-alive")) @@ -545,6 +572,90 @@ func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBod return request, closeAfter, nil } +func readChunkedBody(reader *bufio.Reader, maxBodyBytes int64, setBodyDeadline func(), readLine func() (string, error)) ([]byte, http.Header, error) { + var body bytes.Buffer + initialCapacity := int64(64 << 10) + if maxBodyBytes < initialCapacity { + initialCapacity = maxBodyBytes + } + if initialCapacity > 0 && initialCapacity <= int64(int(^uint(0)>>1)) { + body.Grow(int(initialCapacity)) + } + + for { + line, err := readLine() + if err != nil { + return nil, nil, err + } + chunkSizeText := line + if extension := strings.IndexByte(chunkSizeText, ';'); extension >= 0 { + chunkSizeText = chunkSizeText[:extension] + } + chunkSizeText = strings.TrimSpace(chunkSizeText) + if chunkSizeText == "" || !validHexNumber(chunkSizeText) { + return nil, nil, &protocolError{status: http.StatusBadRequest, message: "invalid chunk size"} + } + chunkSize, parseErr := strconv.ParseUint(chunkSizeText, 16, 63) + if parseErr != nil { + return nil, nil, &protocolError{status: http.StatusBadRequest, message: "invalid chunk size"} + } + if chunkSize == 0 { + trailer := make(http.Header) + for { + line, err = readLine() + if err != nil { + return nil, nil, err + } + if line == "" { + return body.Bytes(), trailer, nil + } + if strings.HasPrefix(line, " ") || strings.HasPrefix(line, "\t") { + return nil, nil, &protocolError{status: http.StatusBadRequest, message: "folded trailers are not supported"} + } + index := strings.IndexByte(line, ':') + if index <= 0 { + return nil, nil, &protocolError{status: http.StatusBadRequest, message: "malformed trailer"} + } + name := strings.TrimSpace(line[:index]) + value := strings.TrimSpace(line[index+1:]) + if !validToken(name) || !validHeaderValue(value) { + return nil, nil, &protocolError{status: http.StatusBadRequest, message: "malformed trailer"} + } + trailer.Add(name, value) + } + } + if int64(body.Len()) > maxBodyBytes-int64(chunkSize) { + return nil, nil, &protocolError{status: http.StatusRequestEntityTooLarge, message: "request body too large"} + } + if setBodyDeadline != nil { + setBodyDeadline() + } + if _, err := io.CopyN(&body, reader, int64(chunkSize)); err != nil { + return nil, nil, err + } + if setBodyDeadline != nil { + setBodyDeadline() + } + var suffix [2]byte + if _, err := io.ReadFull(reader, suffix[:]); err != nil { + return nil, nil, err + } + if suffix[0] != '\r' || suffix[1] != '\n' { + return nil, nil, &protocolError{status: http.StatusBadRequest, message: "malformed chunk terminator"} + } + } +} + +func validHexNumber(value string) bool { + for index := 0; index < len(value); index++ { + char := value[index] + if !((char >= '0' && char <= '9') || (char >= 'a' && char <= 'f') || (char >= 'A' && char <= 'F')) { + return false + } + } + return true +} + func parseProtocol(value string) (int, int, bool) { if value != "HTTP/1.1" && value != "HTTP/1.0" { return 0, 0, false diff --git a/transport/native/server_test.go b/transport/native/server_test.go index be9286c..9263748 100644 --- a/transport/native/server_test.go +++ b/transport/native/server_test.go @@ -53,7 +53,7 @@ func TestServerKeepAliveAndApplicationBridge(t *testing.T) { } } -func TestServerRejectsUnsupportedBodyAndLimits(t *testing.T) { +func TestServerRejectsMalformedBodyAndLimits(t *testing.T) { application := omni.NewWithRouter(netrouter.New()) application.POST("/body", func(ctx *omni.Context) error { return ctx.OK("ok") }) @@ -67,28 +67,85 @@ func TestServerRejectsUnsupportedBodyAndLimits(t *testing.T) { go func() { done <- server.Serve(context.Background(), listener) }() defer stopServer(t, server, listener, done) - for _, requestText := range []string{ - "POST /body HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\n\r\n4\r\ntest\r\n0\r\n\r\n", - "POST /body HTTP/1.1\r\nHost: test\r\nContent-Length: 5\r\n\r\ntest!", - } { + tests := []struct { + request string + status int + }{ + { + request: "POST /body HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\n\r\nZ\r\nbad\r\n0\r\n\r\n", + status: http.StatusBadRequest, + }, + { + request: "POST /body HTTP/1.1\r\nHost: test\r\nContent-Length: 5\r\n\r\ntest!", + status: http.StatusRequestEntityTooLarge, + }, + } + for _, test := range tests { conn, err := net.Dial("tcp", listener.Addr().String()) if err != nil { t.Fatal(err) } - _, _ = io.WriteString(conn, requestText) + _, _ = io.WriteString(conn, test.request) response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodPost}) if err != nil { conn.Close() t.Fatal(err) } response.Body.Close() - if response.StatusCode != http.StatusNotImplemented && response.StatusCode != http.StatusRequestEntityTooLarge { - t.Fatalf("unexpected protocol status %d", response.StatusCode) + if response.StatusCode != test.status { + t.Fatalf("expected status %d, got %d", test.status, response.StatusCode) } conn.Close() } } +func TestServerChunkedBodyAndTrailers(t *testing.T) { + application := omni.NewWithRouter(netrouter.New()) + application.POST("/body", func(ctx *omni.Context) error { + body, err := io.ReadAll(ctx.Request().Body) + if err != nil { + return err + } + return ctx.OK(map[string]string{ + "body": string(body), + "trailer": ctx.Request().Trailer.Get("X-Trace"), + "transfer": strings.Join(ctx.Request().TransferEncoding, ","), + }) + }) + server, listener, done := startServer(t, application) + defer stopServer(t, server, listener, done) + + request := "POST /body HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n" + + "4;source=wire\r\nWiki\r\n5\r\npedia\r\n0\r\nX-Trace: yes\r\n\r\n" + response := rawResponseMethod(t, listener.Addr().String(), http.MethodPost, request) + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil || response.StatusCode != http.StatusOK || !strings.Contains(string(body), `"body":"Wikipedia"`) || !strings.Contains(string(body), `"trailer":"yes"`) || !strings.Contains(string(body), `"transfer":"chunked"`) { + t.Fatalf("unexpected chunked response: status=%d body=%s error=%v", response.StatusCode, body, err) + } +} + +func TestServerChunkedBodyLimit(t *testing.T) { + application := omni.NewWithRouter(netrouter.New()) + application.POST("/body", func(ctx *omni.Context) error { return ctx.OK("unexpected") }) + server := native.New(application) + server.MaxBodyBytes = 4 + listener, 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(), listener) }() + defer stopServer(t, server, listener, done) + + request := "POST /body HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n" + response := rawResponseMethod(t, listener.Addr().String(), http.MethodPost, request) + defer response.Body.Close() + if response.StatusCode != http.StatusRequestEntityTooLarge { + t.Fatalf("expected chunked body limit status 413, got %d", response.StatusCode) + } +} + func TestServerSlowBodyTimesOutBeforeHandler(t *testing.T) { var handlerCalls atomic.Int32 application := omni.NewWithRouter(netrouter.New()) @@ -210,9 +267,9 @@ func TestServerProtocolBoundaries(t *testing.T) { status: http.StatusBadRequest, }, { - name: "chunked", - request: "POST /health HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\n\r\n", - status: http.StatusNotImplemented, + name: "malformed chunked", + request: "POST /health HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\n\r\nZ\r\n\r\n", + status: http.StatusBadRequest, }, { name: "expectation",