From 4daf3b211c866d1096dd90dacec5d7362bc45d83 Mon Sep 17 00:00:00 2001 From: k2safe Date: Mon, 21 Sep 2026 04:35:13 +0800 Subject: [PATCH] feat: cancel native requests on disconnect --- CHANGELOG.md | 1 + ROADMAP.md | 1 + transport/README.md | 4 + transport/native/disconnect_test.go | 104 +++++++++++++++++++++++++ transport/native/server.go | 77 ++++++++++++++++--- transport/native/server_test.go | 115 ++++++++++++++++++++++++++++ 6 files changed, 291 insertions(+), 11 deletions(-) create mode 100644 transport/native/disconnect_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 9349738..d60f645 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ - 加固 native HTTP/1.1 解析边界：校验 Host、重复 Content-Length、Connection token、Transfer-Encoding、控制字符和响应头名称；补充 HTTP/1.0 Keep-Alive、HEAD、204 以及异常请求测试。 - native 支持带 trailers 的 chunked 请求体，并继续通过 `MaxBodyBytes`、读超时和头部上限控制解析内存与连接占用。 - native 响应实现 `http.Flusher`：HTTP/1.1 流式响应自动使用 chunked framing，普通响应继续自动生成 `Content-Length`。 +- native 在长 Handler 运行期间监测客户端断开并取消请求 context，流式写失败同步取消；短请求延迟启动监测，管线数据保留给下一次请求。补充断开、监测退出和长连接复用验收。 - 补充慢请求体超时、大请求体和同连接管线请求验收，确认请求体未完整读取时不会进入业务 Handler，连续请求不会串线。 - 补充 `MaxConnections` 满载拒绝与槽位释放验收，确认请求完成后连接容量可再次使用。 - 增加 8 并发、800 次建连/断连的状态池 churn 验收，覆盖连接对象回收与容量复用。 diff --git a/ROADMAP.md b/ROADMAP.md index ec0c4eb..c65aece 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -43,6 +43,7 @@ - 新增独立服务进程压测 `internal/perfcheck`，记录成功吞吐、P50/P95/P99、服务端分配、采样堆峰值，以及 128 / 1,024 条空闲长连接的堆内存、goroutine 和断开后的回收情况。原始报告保存在 `performance/tcpBaseline.json`。 - native 已支持受 `MaxBodyBytes` 约束的 `Content-Length` / chunked 请求体，仍使用标准 `http.Request` / `http.ResponseWriter` 作为桥接；解析器、连接抽象和生产压测将在后续阶段继续下沉。 - native 响应已支持标准 `http.Flusher`；HTTP/1.1 流式响应使用 chunked framing，普通响应继续走 `Content-Length`，HTTP/1.0 回退到关闭连接分帧。 +- native 长请求已补客户端断开及流式写失败取消，监测退出后再复用 Reader；已验收普通/流式断开、慢 Handler 后连续 Keep-Alive 和管线数据不被误消费。短请求延迟启动监测，已有管线数据时取消依赖后续写失败或 Handler 返回。 - 已补充慢请求体超时、大请求体、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 85a5d44..c92a4b9 100644 --- a/transport/README.md +++ b/transport/README.md @@ -18,6 +18,10 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码 业务 Handler 需要流式返回时可以断言 `http.Flusher` 并调用 `Flush()`；HTTP/1.1 会自动使用 chunked 响应，普通未 Flush 的响应仍自动生成 `Content-Length`。HTTP/1.0 没有 chunked framing，会退回到写完后关闭连接。 +Handler 运行超过约 5 ms 时，native 启动一次阻塞读取监测；读到 EOF 或连接错误会取消 `Request.Context()`，流式写入失败也会触发取消。Handler 返回后先停止监测，再复用连接和 Reader；普通短请求不会启动监测 goroutine。 + +如果客户端提前发送了下一条请求，监测器会保留管线数据并退出，避免预读不断占用内存。这种情况下，断开取消会延后到写失败或 Handler 返回；业务仍应为下游调用设置自己的超时。 + 服务端不是把连接交给一个全局串行 worker，而是每条连接独立处理请求；Keep-Alive 负责复用客户端连接，`Server` 内部的 `sync.Pool` 负责复用小型 Reader、响应对象和连接状态。Reader 默认只有 4 KiB，需要处理大请求头时按需增长。需要保护内存上限时，可以设置 `MaxConnections` 和 `MaxConcurrentRequests`；超出连接上限的连接会被关闭，超出请求并发上限的请求返回 503，不会无限排队。 这个设计借鉴了 Nginx 的资源边界，但没有照搬它的进程模型：Nginx 用少量 worker 进程和事件机制管理大量连接，并用 `worker_connections` 限制每个 worker 的同时连接数；Go runtime 自己已经用 epoll / kqueue 等网络轮询器唤醒阻塞的 goroutine，因此 native 采用“accept loop + 每连接轻量 goroutine + 请求并发闸门”的等价实现。这样业务 Handler 可以继续执行同步代码，慢请求不会阻塞其他连接；只有在真实压测证明 goroutine 调度或协议状态机成为瓶颈时，才需要继续下沉为手写事件循环。 diff --git a/transport/native/disconnect_test.go b/transport/native/disconnect_test.go new file mode 100644 index 0000000..eeccb07 --- /dev/null +++ b/transport/native/disconnect_test.go @@ -0,0 +1,104 @@ +package native + +import ( + "bufio" + "context" + "errors" + "io" + "net" + "net/http" + "strings" + "testing" + "time" +) + +type observedReadConn struct { + net.Conn + reading chan struct{} +} + +func (c *observedReadConn) Read(data []byte) (int, error) { + select { + case c.reading <- struct{}{}: + default: + } + return c.Conn.Read(data) +} + +func TestDisconnectWatchStopsBeforeReaderReuse(t *testing.T) { + server, client := net.Pipe() + defer server.Close() + defer client.Close() + conn := &observedReadConn{Conn: server, reading: make(chan struct{}, 1)} + reader := bufio.NewReader(conn) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + watch := newClientDisconnectWatch(reader, conn, cancel) + select { + case <-conn.reading: + case <-time.After(time.Second): + t.Fatal("disconnect watcher did not start reading") + } + watch.stop() + if ctx.Err() != nil { + t.Fatalf("interrupting the watcher canceled a healthy request: %v", ctx.Err()) + } + if err := server.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + writeDone := make(chan error, 1) + go func() { + _, err := io.WriteString(client, "GET /next HTTP/1.1\r\n") + writeDone <- err + }() + line, err := reader.ReadString('\n') + if err != nil || line != "GET /next HTTP/1.1\r\n" { + t.Fatalf("reader retained watcher data or deadline: line=%q err=%v", line, err) + } + if err := <-writeDone; err != nil { + t.Fatal(err) + } +} + +func TestDisconnectWatchPreservesPipelineAndWriteFailureCancels(t *testing.T) { + server, client := net.Pipe() + defer server.Close() + defer client.Close() + pipeline := "GET /next HTTP/1.1\r\nHost: test\r\n\r\n" + reader := bufio.NewReader(strings.NewReader(pipeline)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + watch := newClientDisconnectWatch(reader, server, cancel) + select { + case <-watch.done: + case <-time.After(time.Second): + t.Fatal("watcher should stop when pipelined data is available") + } + watch.stop() + if ctx.Err() != nil { + t.Fatal("pipeline data canceled the current request") + } + remaining, err := io.ReadAll(reader) + if err != nil || string(remaining) != pipeline { + t.Fatalf("pipeline bytes changed: data=%q err=%v", remaining, err) + } + // With pending pipeline data, EOF detection is deferred. A failed stream + // write must still cancel downstream work even when the handler ignores it. + client.Close() + writer := newResponseWriter(http.MethodGet) + writer.configure(&http.Request{ProtoMajor: 1, ProtoMinor: 1}, server, false, time.Second, cancel) + writer.Flush() + if !errors.Is(ctx.Err(), context.Canceled) { + t.Fatal("failed Flush did not cancel request context") + } + if _, err := writer.Write([]byte("more")); err == nil { + t.Fatal("failed Flush was not returned by subsequent Write") + } + if err := writer.writeTo(server, false, time.Second); err == nil { + t.Fatal("failed Flush was not returned on response completion") + } + writer.reset("") + if writer.cancel != nil || writer.writeErr != nil || writer.conn != nil { + t.Fatal("pooled response retained the previous request") + } +} diff --git a/transport/native/server.go b/transport/native/server.go index 7c5f9f6..4496b3d 100644 --- a/transport/native/server.go +++ b/transport/native/server.go @@ -334,7 +334,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { state.active.Store(true) writer := s.writerPool.Get().(*responseWriter) writer.reset(request.Method) - writer.configure(request, state.conn, closeAfter, s.WriteTimeout) + writer.configure(request, state.conn, closeAfter, s.WriteTimeout, cancel) requestSlot := false if slots := s.requestSlots; slots != nil { select { @@ -363,7 +363,9 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { } } handler := s.handler + disconnectWatch := newClientDisconnectWatch(reader, state.conn, cancel) handler.ServeHTTP(writer, request) + disconnectWatch.stop() cancel() closing := s.isClosing() || state.closeWhenIdle.Load() @@ -711,6 +713,48 @@ func isTimeout(err error) bool { return errors.As(err, &netErr) && netErr.Timeout() } +// Most handlers complete before this delay. A delayed, blocking read avoids a +// watcher goroutine for those requests and does not poll idle client sockets. +const clientDisconnectWatchDelay = 5 * time.Millisecond + +type clientDisconnectWatch struct { + conn net.Conn + done chan struct{} + stopped atomic.Bool + timer *time.Timer +} + +func newClientDisconnectWatch(reader *bufio.Reader, conn net.Conn, cancel context.CancelFunc) *clientDisconnectWatch { + watch := &clientDisconnectWatch{conn: conn, done: make(chan struct{})} + watch.timer = time.AfterFunc(clientDisconnectWatchDelay, func() { + defer close(watch.done) + if watch.stopped.Load() { + return + } + // Only this goroutine may read until stop joins it. Peek preserves + // pipelined bytes for the next request and bounds read-ahead to the + // existing reader buffer. Once data is available, stop watching: + // draining it to detect EOF would require an unbounded pipeline queue. + _, err := reader.Peek(1) + if err != nil && !(watch.stopped.Load() && isTimeout(err)) { + cancel() + } + }) + return watch +} + +func (w *clientDisconnectWatch) stop() { + if w.timer.Stop() { + return + } + w.stopped.Store(true) + // The watcher never changes deadlines: this interrupt cannot be + // overwritten by a callback starting concurrently with stop. + _ = w.conn.SetReadDeadline(time.Now()) + <-w.done + _ = w.conn.SetReadDeadline(time.Time{}) +} + func writeSimple(conn net.Conn, status int, body string, closeAfter bool, timeout time.Duration) error { writer := newResponseWriter("") writer.status = status @@ -735,6 +779,7 @@ type responseWriter struct { chunked bool streamingClose bool writeErr error + cancel context.CancelFunc } func newResponseWriter(method string) *responseWriter { @@ -762,9 +807,11 @@ func (w *responseWriter) reset(method string) { w.chunked = false w.streamingClose = false w.writeErr = nil + w.cancel = nil } -func (w *responseWriter) configure(request *http.Request, conn net.Conn, closeAfter bool, timeout time.Duration) { +func (w *responseWriter) configure(request *http.Request, conn net.Conn, closeAfter bool, timeout time.Duration, cancel context.CancelFunc) { + w.cancel = cancel w.protoMajor = request.ProtoMajor w.protoMinor = request.ProtoMinor w.conn = conn @@ -807,9 +854,7 @@ func (w *responseWriter) Flush() { w.status == http.StatusResetContent || w.status == http.StatusNotModified || (w.status >= 100 && w.status < 200) { return } - if err := w.beginStream(); err != nil { - w.writeErr = err - } + _ = w.recordWriteError(w.beginStream()) } func (w *responseWriter) beginStream() error { @@ -881,15 +926,25 @@ func (w *responseWriter) writeBytes(data []byte) error { if w.writeTimeout > 0 { _ = w.conn.SetWriteDeadline(time.Now().Add(w.writeTimeout)) } - return writeAll(w.conn, data) + return w.recordWriteError(writeAll(w.conn, data)) +} + +func (w *responseWriter) recordWriteError(err error) error { + if err != nil { + w.writeErr = err + if w.cancel != nil { + w.cancel() + } + } + return err } func (w *responseWriter) writeTo(conn net.Conn, closeAfter bool, timeout time.Duration) error { + if w.writeErr != nil { + _ = conn.SetWriteDeadline(time.Time{}) + return w.writeErr + } if w.streaming { - if w.writeErr != nil { - _ = conn.SetWriteDeadline(time.Time{}) - return w.writeErr - } if w.chunked { if err := w.writeBytes([]byte("0\r\n\r\n")); err != nil { w.writeErr = err @@ -979,7 +1034,7 @@ func (w *responseWriter) writeHeaders(conn net.Conn) error { } } output.WriteString("\r\n") - return writeAll(conn, output.Bytes()) + return w.recordWriteError(writeAll(conn, output.Bytes())) } func writeAll(conn net.Conn, data []byte) error { diff --git a/transport/native/server_test.go b/transport/native/server_test.go index efc0c02..1de97d5 100644 --- a/transport/native/server_test.go +++ b/transport/native/server_test.go @@ -496,6 +496,121 @@ func TestServerShutdownCancelsActiveHandler(t *testing.T) { } } +func TestServerCancelsHandlerWhenClientDisconnects(t *testing.T) { + for _, stream := range []bool{false, true} { + t.Run(fmt.Sprintf("stream=%t", stream), func(t *testing.T) { + started := make(chan struct{}) + canceled := make(chan struct{}) + application := omni.NewWithRouter(netrouter.New()) + application.GET("/wait", func(ctx *omni.Context) error { + if stream { + _, _ = io.WriteString(ctx.Raw(), "ready\n") + ctx.Raw().(http.Flusher).Flush() + } + close(started) + <-ctx.Context().Done() + close(canceled) + return nil + }) + server, listener, done := startServer(t, application) + defer stopServer(t, server, listener, done) + + conn, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(2 * time.Second)) + _, _ = io.WriteString(conn, "GET /wait HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n") + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("handler did not start") + } + if stream { + response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodGet}) + if err != nil { + t.Fatal(err) + } + first := make([]byte, len("ready\n")) + if _, err := io.ReadFull(response.Body, first); err != nil || string(first) != "ready\n" { + t.Fatalf("failed reading stream: %q err=%v", first, err) + } + // Close the socket before closing the unfinished response body. + conn.Close() + response.Body.Close() + } else { + conn.Close() + } + select { + case <-canceled: + case <-time.After(time.Second): + t.Fatal("client disconnect did not cancel request context") + } + }) + } +} + +func TestServerLongHandlerPreservesKeepAliveAndPipeline(t *testing.T) { + for _, pipeline := range []bool{false, true} { + t.Run(fmt.Sprintf("pipeline=%t", pipeline), func(t *testing.T) { + application := omni.New() + application.GET("/slow", func(ctx *omni.Context) error { + timer := time.NewTimer(25 * time.Millisecond) + defer timer.Stop() + select { + case <-timer.C: + return ctx.OK("slow done") + case <-ctx.Context().Done(): + return ctx.Context().Err() + } + }) + application.POST("/next", func(ctx *omni.Context) error { + body, err := io.ReadAll(ctx.Request().Body) + if err != nil { + return err + } + return ctx.OK(string(body)) + }) + server, listener, done := startServer(t, application) + defer stopServer(t, server, listener, done) + conn, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(3 * time.Second)) + reader := bufio.NewReader(conn) + for attempt := 0; attempt < 3; attempt++ { + slow := "GET /slow HTTP/1.1\r\nHost: test\r\n\r\n" + next := "POST /next HTTP/1.1\r\nHost: test\r\nContent-Length: 7\r\n\r\npayload" + if pipeline { + slow += next + } + if _, err := io.WriteString(conn, slow); err != nil { + t.Fatal(err) + } + for index, want := range []string{"slow done", "payload"} { + if index == 1 && !pipeline { + if _, err := io.WriteString(conn, next); err != nil { + t.Fatal(err) + } + } + response, err := http.ReadResponse(reader, &http.Request{Method: http.MethodGet}) + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(response.Body) + response.Body.Close() + if err != nil || response.StatusCode != http.StatusOK || !strings.Contains(string(body), want) { + t.Fatalf("attempt=%d response=%d status=%d body=%s err=%v", attempt, index, response.StatusCode, body, err) + } + } + } + }) + } +} + func TestServerConcurrentRequestLimit(t *testing.T) { started := make(chan struct{}) release := make(chan struct{})