From fa2be5e0166f1fa581a2b0c7ee2fc29c44ac2676 Mon Sep 17 00:00:00 2001 From: k2safe Date: Wed, 23 Sep 2026 10:32:28 +0800 Subject: [PATCH] test: verify HTTP2 connection window recovery and shutdown --- CHANGELOG.md | 1 + transport/http2/flowControl_test.go | 147 +++++++++++++++++++++++++++- 2 files changed, 147 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a8d1a4..a7c0737 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ - 新增可配置时长的 HTTP/2 TLS soak：默认 3 秒，Linux 验收可通过 `GO_OMNI_HTTP2_SOAK_DURATION=30m` 覆盖，混合验证 128 路 stream、慢 handler、请求体、Keep-Alive 和退出回收。 - HTTP/2 soak 每 15 秒输出请求/失败数、Go 堆、Linux RSS、goroutine、FD 和协议统计；校验响应协议、内容和独立请求 ID，不再忽略压测期限附近已发出请求的失败。 - 补充 HTTP/2 慢响应流窗口为零时的隔离与恢复验收：同连接 204、另一连接带 body 响应、WINDOW_UPDATE 恢复发送及 RST_STREAM 释放请求槽位；补充 TLS 握手未完成时的连接上限拒绝、握手超时回收和正常请求恢复。 +- 补充连接级接收窗口耗尽测试：保持 stream 信用充足，消费完 65,535 字节连接窗口后确认大响应暂停，同连接 204 和另一条连接仍可完成，连接级 WINDOW_UPDATE 恢复后继续复用；补充零窗口阻塞写在关闭超时后解除阻塞、取消请求与连接回收的验收。 - 新增 `scripts/verifyHTTP2.py`，固定在 Go 1.25 Linux 容器中运行 HTTP/2 TLS 长测；默认使用普通模式，`--race` 默认运行 60 秒。容器默认限制为 2 核、2 GiB 内存、256 个进程/线程，带总时限、中断清理和日志大小限制。 - 完成 Linux amd64 单 TLS 连接、128 个 worker 的 30 分钟 HTTP/2 长测：58,102,476 次请求，0 失败与协议错误，采样 RSS 峰值约 22.6 MiB（包含客户端和服务端），退出后连接/请求/FD 回收；保留原始日志和 120 份采样。多连接、持续慢客户端与 HTTP/2 回源联调仍是后续门槛。 - native TLS 明确限制 ALPN 为 `http/1.1`：缺省自动宣告该协议，误配置 `h2` 等未实现协议时在启动阶段报错。 diff --git a/transport/http2/flowControl_test.go b/transport/http2/flowControl_test.go index 6824b15..1bac63e 100644 --- a/transport/http2/flowControl_test.go +++ b/transport/http2/flowControl_test.go @@ -1,6 +1,8 @@ package http2 import ( + "context" + "errors" "fmt" "io" "net/http" @@ -108,6 +110,149 @@ func TestSlowResponseStreamIsolationAndRecovery(t *testing.T) { } } +// Leave stream credit available while exhausting the shared connection window. +// DATA must pause, but headers and traffic on a different TCP connection must +// remain usable until connection-level credit is restored. +func TestConnectionWindowIsolationAndRecovery(t *testing.T) { + const initialConnectionWindow = 65535 + payload := strings.Repeat("x", 96<<10) + written := make(chan error, 1) + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + switch request.URL.Path { + case "/health": + writer.WriteHeader(http.StatusNoContent) + case "/ready": + _, _ = io.WriteString(writer, "ready") + default: + writer.Header().Set("Content-Length", fmt.Sprint(len(payload))) + writer.(http.Flusher).Flush() + _, err := io.WriteString(writer, payload) + written <- err + } + })) + server.MaxConnections = 2 + server.MaxConcurrentRequests = 2 + address, stop := startServer(t, server) + defer stop() + + slow := newFramePeer(t, address, xhttp2.Setting{ID: xhttp2.SettingInitialWindowSize, Val: 128 << 10}) + defer slow.Close() + slow.framer.ReadMetaHeaders = hpack.NewDecoder(4096, nil) + slow.get(t, 1, "/large") + var received strings.Builder + for received.Len() < initialConnectionWindow { + switch frame := slow.read(t).(type) { + case *xhttp2.DataFrame: + if frame.StreamID != 1 || frame.StreamEnded() { + t.Fatalf("unexpected DATA before connection credit was restored: %+v", frame) + } + received.Write(frame.Data()) + case *xhttp2.RSTStreamFrame: + t.Fatalf("unexpected server reset: %+v", frame) + } + } + if received.Len() != initialConnectionWindow { + t.Fatalf("server exceeded connection window: got %d bytes", received.Len()) + } + + slow.get(t, 3, "/health") + readNoContent(t, slow, 3) + select { + case err := <-written: + t.Fatalf("large response escaped exhausted connection window: %v", err) + default: + } + + fast := newFramePeer(t, address) + defer fast.Close() + fast.get(t, 1, "/ready") + if body := fast.body(t, 1); body != "ready" { + t.Fatalf("other connection response: %q", body) + } + + remaining := len(payload) - received.Len() + if err := slow.framer.WriteWindowUpdate(0, uint32(remaining)); err != nil { + t.Fatal(err) + } + received.WriteString(slow.body(t, 1)) + if received.String() != payload { + t.Fatalf("resumed connection response mismatch: got %d bytes, want %d", received.Len(), len(payload)) + } + select { + case err := <-written: + if err != nil { + t.Fatalf("resumed connection write: %v", err) + } + case <-time.After(time.Second): + t.Fatal("connection WINDOW_UPDATE did not unblock writer") + } + + if err := slow.framer.WriteWindowUpdate(0, uint32(len("ready"))); err != nil { + t.Fatal(err) + } + slow.get(t, 5, "/ready") + if body := slow.body(t, 5); body != "ready" { + t.Fatalf("reused connection response: %q", body) + } + waitHTTP2Stats(t, server, func(stats Stats) bool { return stats.ActiveRequests == 0 }) + stats := server.Stats() + if stats.AcceptedConnections != 2 || stats.ActiveConnections != 2 || stats.RejectedConnections != 0 || stats.RejectedRequests != 0 || stats.ProtocolErrors != 0 || stats.RSTStreams != 0 { + t.Fatalf("connection flow control did not recover: %+v", stats) + } +} + +func TestShutdownReleasesFlowControlledWriter(t *testing.T) { + started := make(chan struct{}) + written := make(chan error, 1) + requestContext := make(chan context.Context, 1) + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + requestContext <- request.Context() + writer.(http.Flusher).Flush() + close(started) + _, err := io.WriteString(writer, strings.Repeat("x", 32<<10)) + written <- err + })) + address, done := launchServer(t, server, context.Background()) + peer := newFramePeer(t, address, xhttp2.Setting{ID: xhttp2.SettingInitialWindowSize, Val: 0}) + defer peer.Close() + peer.get(t, 1, "/blocked") + waitSignal(t, started) + requestCtx := <-requestContext + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + stopped := make(chan error, 1) + go func() { stopped <- server.Shutdown(ctx) }() + for { + if frame, ok := peer.read(t).(*xhttp2.GoAwayFrame); ok { + if frame.ErrCode != xhttp2.ErrCodeNo || frame.LastStreamID != 1 { + t.Fatalf("unexpected GOAWAY for blocked writer: %+v", frame) + } + break + } + } + if err := serveResult(t, stopped); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("shutdown with exhausted receive window: %v", err) + } + select { + case err := <-written: + if err == nil { + t.Fatal("blocked write unexpectedly succeeded without window credit") + } + case <-time.After(time.Second): + t.Fatal("shutdown deadline did not release flow-controlled writer") + } + if requestCtx.Err() == nil { + t.Fatal("forced shutdown did not cancel request context") + } + if err := serveResult(t, done); err != nil { + t.Fatal(err) + } + waitHTTP2Stats(t, server, func(stats Stats) bool { return stats.ActiveRequests == 0 && stats.ActiveConnections == 0 }) + if stats := server.Stats(); stats.GoAwayFrames != 1 || stats.ProtocolErrors != 0 { + t.Fatalf("blocked connection did not close cleanly: %+v", stats) + } +} + func readNoContent(t *testing.T, peer *framePeer, streamID uint32) { t.Helper() for { @@ -121,7 +266,7 @@ func readNoContent(t *testing.T, peer *framePeer, streamID uint32) { } return case *xhttp2.DataFrame: - t.Fatalf("unexpected DATA on zero-window stream %d", frame.StreamID) + t.Fatalf("unexpected DATA while receive window is exhausted on stream %d", frame.StreamID) case *xhttp2.RSTStreamFrame: t.Fatalf("unexpected server reset: %+v", frame) }