feat: cancel native requests on disconnect

2026-09-21 04:35:13 +08:00
parent 62b0f09546
commit 4daf3b211c
6 changed files with 291 additions and 11 deletions
--- 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 验收，覆盖连接对象回收与容量复用。
--- 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 行为观测。
--- 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 调度或协议状态机成为瓶颈时，才需要继续下沉为手写事件循环。
--- a/transport/native/disconnect_test.go
+++ 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")
+	}
+}
--- 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 {
--- 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{})