From a072da4a3147abdced0ed06a7ea82df0485a45be Mon Sep 17 00:00:00 2001 From: k2safe Date: Tue, 22 Sep 2026 22:25:35 +0800 Subject: [PATCH] feat: make HTTP2 shutdown graceful --- CHANGELOG.md | 2 +- ROADMAP.md | 2 +- transport/http2/lifecycle_test.go | 243 ++++++++++++++++++++++++++++++ transport/http2/server.go | 92 ++++++++--- transport/native/HTTP2.md | 6 +- 5 files changed, 319 insertions(+), 26 deletions(-) create mode 100644 transport/http2/lifecycle_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d51383..0c6def9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## Unreleased -- 新增独立 `transport/http2` 实验 Runner：支持 h2c/TLS h2、连接上限、全局请求并发闸门和生命周期统计；不改变 native HTTP/1.1 默认路径。 +- 新增独立 `transport/http2` 实验 Runner：支持 h2c/TLS h2、连接上限、全局请求并发闸门、GOAWAY 排空、RST_STREAM 单流取消和生命周期统计；不改变 native HTTP/1.1 默认路径。 - native TLS 明确限制 ALPN 为 `http/1.1`：缺省自动宣告该协议，误配置 `h2` 等未实现协议时在启动阶段报错。 - native `Server.Stats()` 新增连接和请求过载拒绝计数；新增可信代理配置，访问日志只在 socket 对端属于配置网段时解析 `X-Forwarded-For`。 - 新增 transport-neutral `metrics` 包，支持请求数、错误数、in-flight、累计耗时和状态分布，并可直接注册 `/metrics` 接口。 diff --git a/ROADMAP.md b/ROADMAP.md index 2253bbe..6f5e061 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -50,7 +50,7 @@ - 已验证 `MaxConcurrentRequests` 满载返回 503，活动请求结束后槽位释放，后续请求可以恢复处理。 - native `Stats()` 已记录连接和请求过载拒绝数；访问日志已支持可信代理网段配置，默认拒绝伪造的 `X-Forwarded-For`。 - native TLS 已明确只宣告 HTTP/1.1 ALPN；HTTP/2 暂由边缘代理终止或使用 `transport/nethttp`，已整理独立协议设计、流控和压测门槛，见 [native HTTP/2 进入条件](transport/native/HTTP2.md)，完成后再接入 native。 -- 已加入独立 `transport/http2` 实验 Runner，验证 h2c/TLS h2、连接上限、全局请求并发拒绝和恢复；它使用 `x/net/http2` 作为临时帧/流控实现，尚未达到 native 生产协议的 GOAWAY、流控和长测门槛。 +- 已加入独立 `transport/http2` 实验 Runner，验证 h2c/TLS h2、连接上限、全局请求并发拒绝和恢复，以及 GOAWAY 排空、RST_STREAM 单流取消和服务关闭超时；它使用 `x/net/http2` 作为临时帧/流控实现，尚未达到 native 生产协议的流控和长测门槛。 - 已用 16 个并发 worker 完成 3,200 次建连/断连 race 验收；MaxConnections 满载拒绝、槽位释放和 TLS 连接上限均已覆盖。 - 新增 `scripts/verifyLinux.py`，固定在 Go 1.25 Linux 容器内运行 race、vet、build 和可选 TLS/非 TLS perfcheck；perfcheck 记录堆、goroutine、连接和 Linux 文件描述符。 - Linux/arm64 已完成 30 秒、128 并发、512 空闲连接的 TLS 与非 TLS 持续回归，报告见 `performance/linuxSoak.json` 和 `performance/linuxSoakTLS.json`，均为 0 错误且连接、goroutine、FD 回收正常。 diff --git a/transport/http2/lifecycle_test.go b/transport/http2/lifecycle_test.go new file mode 100644 index 0000000..721c1c9 --- /dev/null +++ b/transport/http2/lifecycle_test.go @@ -0,0 +1,243 @@ +package http2 + +import ( + "bytes" + "context" + "errors" + "io" + "net" + "net/http" + "testing" + "time" + + xhttp2 "golang.org/x/net/http2" + "golang.org/x/net/http2/hpack" +) + +// A raw peer makes GOAWAY and RST_STREAM observable without client retries. +type framePeer struct { + net.Conn + framer *xhttp2.Framer +} + +func newFramePeer(t *testing.T, address string) *framePeer { + t.Helper() + c, err := net.DialTimeout("tcp", address, time.Second) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Close() }) + c.SetDeadline(time.Now().Add(5 * time.Second)) + p := &framePeer{c, xhttp2.NewFramer(c, c)} + if _, err = io.WriteString(c, xhttp2.ClientPreface); err != nil { + t.Fatal(err) + } + if err = p.framer.WriteSettings(); err != nil { + t.Fatal(err) + } + return p +} +func (p *framePeer) get(t *testing.T, id uint32, path string) { + t.Helper() + var b bytes.Buffer + e := hpack.NewEncoder(&b) + for _, field := range []hpack.HeaderField{{Name: ":method", Value: "GET"}, {Name: ":scheme", Value: "http"}, {Name: ":authority", Value: "localhost"}, {Name: ":path", Value: path}} { + if err := e.WriteField(field); err != nil { + t.Fatal(err) + } + } + if err := p.framer.WriteHeaders(xhttp2.HeadersFrameParam{StreamID: id, EndStream: true, EndHeaders: true, BlockFragment: b.Bytes()}); err != nil { + t.Fatal(err) + } +} +func (p *framePeer) read(t *testing.T) xhttp2.Frame { + t.Helper() + f, err := p.framer.ReadFrame() + if err != nil { + t.Fatal(err) + } + if s, ok := f.(*xhttp2.SettingsFrame); ok && !s.IsAck() { + if err := p.framer.WriteSettingsAck(); err != nil { + t.Fatal(err) + } + } + return f +} +func (p *framePeer) body(t *testing.T, id uint32) string { + t.Helper() + var b bytes.Buffer + for { + switch f := p.read(t).(type) { + case *xhttp2.DataFrame: + if f.StreamID != id { + t.Fatalf("unexpected stream %d", f.StreamID) + } + b.Write(f.Data()) + if f.StreamEnded() { + return b.String() + } + case *xhttp2.HeadersFrame: + if f.StreamID == id && f.StreamEnded() { + return b.String() + } + case *xhttp2.RSTStreamFrame: + t.Fatalf("unexpected RST_STREAM: %v", f) + } + } +} +func waitSignal(t *testing.T, signal <-chan struct{}) { + t.Helper() + select { + case <-signal: + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for event") + } +} +func launchServer(t *testing.T, s *Server, ctx context.Context) (string, <-chan error) { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- s.Serve(ctx, l) }() + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := s.Shutdown(ctx); err != nil { + t.Errorf("cleanup shutdown: %v", err) + } + l.Close() + }) + return l.Addr().String(), done +} +func serveResult(t *testing.T, done <-chan error) error { + t.Helper() + select { + case err := <-done: + return err + case <-time.After(3 * time.Second): + t.Fatal("Serve did not finish") + return nil + } +} +func TestShutdownSendsGoAwayAndDrains(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + defer close(release) + requestCtx := make(chan context.Context, 1) + s := New(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/hold" { + t.Error("new stream reached handler during drain") + return + } + requestCtx <- r.Context() + close(started) + select { + case <-release: + case <-r.Context().Done(): + return + } + io.WriteString(w, "completed") + })) + address, done := launchServer(t, s, context.Background()) + p := newFramePeer(t, address) + p.get(t, 1, "/hold") + waitSignal(t, started) + rctx := <-requestCtx + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + stopped := make(chan error, 1) + go func() { stopped <- s.Shutdown(ctx) }() + for { + if f, ok := p.read(t).(*xhttp2.GoAwayFrame); ok { + if f.ErrCode != xhttp2.ErrCodeNo || f.LastStreamID != 1 { + t.Fatalf("GOAWAY=%+v", f) + } + break + } + } + if err := rctx.Err(); err != nil { + t.Fatalf("graceful shutdown canceled active request: %v", err) + } + select { + case err := <-done: + t.Fatalf("Serve returned before drain: %v", err) + default: + } + p.get(t, 3, "/late") + release <- struct{}{} + if got := p.body(t, 1); got != "completed" { + t.Fatalf("body=%q", got) + } + p.Close() + if err := serveResult(t, stopped); err != nil { + t.Fatal(err) + } + if err := serveResult(t, done); err != nil { + t.Fatal(err) + } + if stats := s.Stats(); stats.ActiveConnections != 0 || stats.ActiveRequests != 0 { + t.Fatalf("not drained: %+v", stats) + } +} +func TestRSTStreamCancelsOnlyOneRequest(t *testing.T) { + started := make(chan struct{}, 2) + canceled := make(chan string, 2) + release := make(chan struct{}) + defer close(release) + s := New(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + started <- struct{}{} + select { + case <-r.Context().Done(): + canceled <- r.URL.Path + return + case <-release: + io.WriteString(w, r.URL.Path) + } + })) + s.MaxConcurrentRequests = 2 + address, _ := launchServer(t, s, context.Background()) + p := newFramePeer(t, address) + p.get(t, 1, "/cancel") + p.get(t, 3, "/survive") + waitSignal(t, started) + waitSignal(t, started) + if err := p.framer.WriteRSTStream(1, xhttp2.ErrCodeCancel); err != nil { + t.Fatal(err) + } + select { + case path := <-canceled: + if path != "/cancel" { + t.Fatalf("canceled %s", path) + } + case <-time.After(time.Second): + t.Fatal("RST did not cancel request") + } + release <- struct{}{} + if got := p.body(t, 3); got != "/survive" { + t.Fatalf("sibling body=%q", got) + } + p.get(t, 5, "/reused") + waitSignal(t, started) + release <- struct{}{} + if got := p.body(t, 5); got != "/reused" { + t.Fatalf("connection reuse body=%q", got) + } +} +func TestCanceledServeContextWaitsForShutdownDeadline(t *testing.T) { + started, canceled := make(chan struct{}), make(chan struct{}) + s := New(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { close(started); <-r.Context().Done(); close(canceled) })) + s.ShutdownTimeout = 100 * time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + address, done := launchServer(t, s, ctx) + p := newFramePeer(t, address) + p.get(t, 1, "/hold") + waitSignal(t, started) + cancel() + if err := serveResult(t, done); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Serve should report drain timeout, got %v", err) + } + waitSignal(t, canceled) +} diff --git a/transport/http2/server.go b/transport/http2/server.go index 6f2a3cd..db0de49 100644 --- a/transport/http2/server.go +++ b/transport/http2/server.go @@ -42,6 +42,9 @@ type Server struct { mu sync.Mutex handler http.Handler listener net.Listener + shutdownConfig *http.Server + shutdownDone chan struct{} + shutdownOnce sync.Once closing bool cancel context.CancelFunc conns map[net.Conn]struct{} @@ -176,9 +179,14 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error { s.mu.Unlock() return errors.New("go-omni/http2: server is already running") } - baseCtx, cancel := context.WithCancel(ctx) + // Keep the connection context independent from the caller's context. A + // canceled parent starts graceful shutdown; it must not cancel active + // streams before the GOAWAY/drain phase has completed. + baseCtx, cancel := context.WithCancel(context.Background()) s.cancel = cancel s.listener = listener + s.shutdownDone = make(chan struct{}) + s.shutdownOnce = sync.Once{} s.closing = false if s.conns == nil { s.conns = make(map[net.Conn]struct{}) @@ -200,24 +208,43 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error { MaxConcurrentStreams: s.MaxConcurrentStreams, IdleTimeout: s.IdleTimeout, } - shutdownDone := make(chan struct{}) - go func() { - select { - case <-ctx.Done(): - shutdownCtx, cancel := context.WithTimeout(context.Background(), s.shutdownTimeout()) - _ = s.Shutdown(shutdownCtx) - cancel() - case <-shutdownDone: - } - }() - - defer func() { - close(shutdownDone) + shutdownConfig := &http.Server{IdleTimeout: s.IdleTimeout} + if err := xhttp2.ConfigureServer(shutdownConfig, protocolServer); err != nil { cancel() s.mu.Lock() s.listener = nil s.cancel = nil s.mu.Unlock() + return err + } + s.mu.Lock() + s.shutdownConfig = shutdownConfig + s.mu.Unlock() + contextWakeDone := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + // Wake Accept. Serve performs the graceful drain synchronously so + // its return value reports a shutdown timeout to the caller. + _ = listener.Close() + case <-contextWakeDone: + } + }() + + defer func() { + close(contextWakeDone) + s.mu.Lock() + closing := s.closing + shutdownDone := s.shutdownDone + s.listener = nil + s.shutdownConfig = nil + s.cancel = nil + s.mu.Unlock() + if !closing { + cancel() + } else if shutdownDone != nil { + <-shutdownDone + } }() for { @@ -251,7 +278,13 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error { s.activeConns.Add(1) s.wg.Add(1) s.mu.Unlock() - go s.serveConn(baseCtx, protocolServer, gate, conn) + go s.serveConn(baseCtx, protocolServer, shutdownConfig, gate, conn) + } + if ctx.Err() != nil && !s.isClosing() { + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), s.shutdownTimeout()) + err := s.Shutdown(shutdownCtx) + shutdownCancel() + return err } return nil } @@ -263,7 +296,7 @@ func (s *Server) shutdownTimeout() time.Duration { return defaultShutdownTimeout } -func (s *Server) serveConn(ctx context.Context, protocolServer *xhttp2.Server, handler http.Handler, conn net.Conn) { +func (s *Server) serveConn(ctx context.Context, protocolServer *xhttp2.Server, baseConfig *http.Server, handler http.Handler, conn net.Conn) { defer s.wg.Done() defer func() { s.mu.Lock() @@ -283,11 +316,7 @@ func (s *Server) serveConn(ctx context.Context, protocolServer *xhttp2.Server, h _ = conn.SetDeadline(time.Time{}) } protocolServer.ServeConn(conn, &xhttp2.ServeConnOpts{ - Context: ctx, - Handler: handler, - BaseConfig: &http.Server{ - IdleTimeout: s.IdleTimeout, - }, + Context: ctx, Handler: handler, BaseConfig: baseConfig, }) } @@ -321,6 +350,7 @@ func (s *Server) Shutdown(ctx context.Context) error { } s.closing = true listener := s.listener + shutdownConfig := s.shutdownConfig cancel := s.cancel connections := make([]net.Conn, 0, len(s.conns)) for conn := range s.conns { @@ -330,7 +360,12 @@ func (s *Server) Shutdown(ctx context.Context) error { if listener != nil { _ = listener.Close() } - cancel() + // x/net/http2's configured shutdown callback sends GOAWAY and lets active + // streams finish. The base request context remains live until the timeout + // path below, where connections are forcibly closed. + if shutdownConfig != nil { + _ = shutdownConfig.Shutdown(context.Background()) + } done := make(chan struct{}) go func() { @@ -339,15 +374,28 @@ func (s *Server) Shutdown(ctx context.Context) error { }() select { case <-done: + cancel() + s.finishShutdown() return nil case <-ctx.Done(): for _, conn := range connections { _ = conn.Close() } + cancel() + s.finishShutdown() return ctx.Err() } } +func (s *Server) finishShutdown() { + s.mu.Lock() + done := s.shutdownDone + s.mu.Unlock() + if done != nil { + s.shutdownOnce.Do(func() { close(done) }) + } +} + func (s *Server) isClosing() bool { s.mu.Lock() closing := s.closing diff --git a/transport/native/HTTP2.md b/transport/native/HTTP2.md index 37415bc..8f8c14b 100644 --- a/transport/native/HTTP2.md +++ b/transport/native/HTTP2.md @@ -2,6 +2,8 @@ 当前 `transport/native` 是经过 Linux 长连接和反向代理验收的 HTTP/1.1 实现。HTTP/2 暂不直接打开，原因是它把一个 TCP 连接变成多个独立请求流，连接级的 Keep-Alive、超时和并发限制不能直接复用现有 HTTP/1.1 状态机。 +当前已提供独立的 `transport/http2` 实验 Runner。它已经验证 TLS/h2、h2c、本地连接和请求闸门、GOAWAY 排空以及 RST_STREAM 取消；这些结果用于验证生命周期设计，不代表 native 已经启用 HTTP/2。 + ## 目标边界 - HTTP/2 必须作为显式 opt-in 协议，不改变 `omni.New()`、`transport/nethttp` 或 native HTTP/1.1 的默认行为。 @@ -15,7 +17,7 @@ 2. TLS 协商得到 `h2` 后进入 HTTP/2 connection state；其他 ALPN 直接关闭，不能把 HTTP/2 preface 交给 HTTP/1.1 解析器。 3. 每个 stream 创建独立的 `http.Request` 和取消 context。`RST_STREAM` 只取消当前请求，TCP 断开才取消该连接上的所有请求。 4. `MaxConcurrentRequests` 是所有连接共享的全局请求闸门；`MaxConcurrentStreams` 限制单连接并发流；两者都必须在收到完整请求体前生效。 -5. 服务关闭时停止 accept，发送 GOAWAY 拒绝新流，等待活动流排空；超过 `ShutdownTimeout` 后关闭连接并取消剩余请求。 +5. 服务关闭时停止 accept，发送 GOAWAY 拒绝新流，等待活动流排空；超过 `ShutdownTimeout` 后关闭连接并取消剩余请求。实验 Runner 已覆盖这条顺序。 ## 必须可观测的状态 @@ -26,7 +28,7 @@ HTTP/2 版本需要在 native `Stats` 之外至少记录：活动 stream、被 - TLS ALPN：只协商 `h2`，HTTP/1.1 客户端仍由现有 native 路径处理；不支持的 ALPN 在握手前后都不会误入错误解析器。 - 多路复用：一个连接并发至少 128 个 stream，响应和 requestId 不串线，单个慢 stream 不阻塞其他 stream。 - 流控：慢读取、慢写、窗口耗尽和大 body 都能恢复；内存不会随未消费 body 无限增长。 -- 取消：RST_STREAM、客户端断开、服务关闭分别只影响应受影响的请求，并能在 metrics 中看到回收。 +- 取消：RST_STREAM、客户端断开、服务关闭分别只影响应受影响的请求，并能在 metrics 中看到回收；当前实验 Runner 已覆盖 RST_STREAM 与服务关闭。 - 背压：连接上限、全局 stream 上限、单连接 stream 上限和 header/body 上限均能快速拒绝并自动恢复。 - 代理联调：Nginx/Cloudflare 的 TLS、Keep-Alive、超时、GOAWAY 和优雅发布行为与 HTTP/1.1 回源结果一致。 - 长测：Linux amd64 至少 30 分钟 TLS、多连接、多 stream、慢客户端混合场景；要求 0 协议错误、0 请求串线，连接、stream、goroutine 和 FD 在停止后回收。