test: complete bounded HTTP2 TLS soak and flow control checks

transport/http2/admission_test.go

@@ -0,0 +1,116 @@
+package http2
+
+import (
+	"context"
+	"crypto/tls"
+	"io"
+	"net"
+	"net/http"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	xhttp2 "golang.org/x/net/http2"
+)
+
+func TestTLSConnectionLimitRecoversAfterHandshakeTimeout(t *testing.T) {
+	certificate, roots := testCertificate(t)
+	var handled atomic.Int32
+	server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		handled.Add(1)
+		_, _ = io.WriteString(writer, "ok")
+	}))
+	server.MaxConnections = 1
+	server.TLSHandshakeTimeout = time.Second
+	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.ServeTLS(context.Background(), listener, &tls.Config{
+			Certificates: []tls.Certificate{certificate}, MinVersion: tls.VersionTLS12,
+		})
+	}()
+	t.Cleanup(func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel()
+		if err := server.Shutdown(ctx); err != nil {
+			t.Errorf("shutdown: %v", err)
+		}
+		_ = listener.Close()
+		if err := serveResult(t, done); err != nil {
+			t.Errorf("serve: %v", err)
+		}
+	})
+
+	// Occupy the only connection slot without sending a TLS ClientHello.
+	held, err := net.DialTimeout("tcp", listener.Addr().String(), time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer held.Close()
+	waitHTTP2Stats(t, server, func(stats Stats) bool { return stats.ActiveConnections == 1 })
+
+	rejected, err := net.DialTimeout("tcp", listener.Addr().String(), time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rejected.Close()
+	if err := rejected.SetReadDeadline(time.Now().Add(250 * time.Millisecond)); err != nil {
+		t.Fatal(err)
+	}
+	var buffer [1]byte
+	if _, err := rejected.Read(buffer[:]); err == nil {
+		t.Fatal("connection above the cap was not closed")
+	} else if timeout, ok := err.(net.Error); ok && timeout.Timeout() {
+		t.Fatal("connection above the cap was not rejected promptly")
+	}
+	if stats := server.Stats(); stats.RejectedConnections != 1 || handled.Load() != 0 {
+		t.Fatalf("unexpected admission state: %+v handled=%d", stats, handled.Load())
+	}
+
+	// The handshake deadline, rather than a client disconnect, must release it.
+	waitHTTP2Stats(t, server, func(stats Stats) bool { return stats.ActiveConnections == 0 })
+	if err := held.SetReadDeadline(time.Now().Add(250 * time.Millisecond)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := held.Read(buffer[:]); err == nil {
+		t.Fatal("timed-out TLS connection remained open")
+	} else if timeout, ok := err.(net.Error); ok && timeout.Timeout() {
+		t.Fatal("timed-out TLS connection was not closed")
+	}
+
+	transport := &xhttp2.Transport{TLSClientConfig: &tls.Config{
+		RootCAs: roots, ServerName: "localhost", MinVersion: tls.VersionTLS12,
+		NextProtos: []string{"h2"},
+	}}
+	defer transport.CloseIdleConnections()
+	client := &http.Client{Transport: transport, Timeout: 3 * time.Second}
+	response, err := client.Get("https://" + listener.Addr().String() + "/health")
+	if err != nil {
+		t.Fatal(err)
+	}
+	body, readErr := io.ReadAll(response.Body)
+	response.Body.Close()
+	if readErr != nil || response.StatusCode != http.StatusOK || response.ProtoMajor != 2 || string(body) != "ok" {
+		t.Fatalf("TLS admission did not recover: status=%d protocol=%s body=%q err=%v", response.StatusCode, response.Proto, body, readErr)
+	}
+	transport.CloseIdleConnections()
+	waitHTTP2Stats(t, server, func(stats Stats) bool { return stats.ActiveConnections == 0 && stats.ActiveRequests == 0 })
+	if stats := server.Stats(); stats.AcceptedConnections != 3 || stats.RejectedConnections != 1 || stats.ProtocolErrors != 0 || handled.Load() != 1 {
+		t.Fatalf("unexpected recovered admission state: %+v handled=%d", stats, handled.Load())
+	}
+}
+
+func waitHTTP2Stats(t *testing.T, server *Server, predicate func(Stats) bool) {
+	t.Helper()
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if predicate(server.Stats()) {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatalf("HTTP/2 stats did not reach expected state: %+v", server.Stats())
+}

transport/http2/flowControl_test.go

@@ -0,0 +1,129 @@
+package http2
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	xhttp2 "golang.org/x/net/http2"
+	"golang.org/x/net/http2/hpack"
+)
+
+// A zero receive window simulates a client that has stopped consuming DATA.
+// Headers need no flow-control credit, so a 204 on another stream should still
+// complete while the large response waits for a WINDOW_UPDATE or cancellation.
+func TestSlowResponseStreamIsolationAndRecovery(t *testing.T) {
+	payload := strings.Repeat("x", 32<<10)
+	started := make(chan struct{}, 2)
+	written := make(chan error, 2)
+	server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		if request.URL.Path == "/health" {
+			writer.WriteHeader(http.StatusNoContent)
+			return
+		}
+		if request.URL.Path == "/ready" {
+			_, _ = io.WriteString(writer, "ready")
+			return
+		}
+		writer.Header().Set("Content-Length", fmt.Sprint(len(payload)))
+		writer.(http.Flusher).Flush()
+		started <- struct{}{}
+		_, 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: 0})
+	defer slow.Close()
+	slow.framer.ReadMetaHeaders = hpack.NewDecoder(4096, nil)
+	slow.get(t, 1, "/large")
+	waitSignal(t, started)
+
+	// The same TCP connection can still carry response headers for other streams.
+	slow.get(t, 3, "/health")
+	readNoContent(t, slow, 3)
+	select {
+	case err := <-written:
+		t.Fatalf("large response escaped the zero stream window: %v", err)
+	default:
+	}
+
+	// A second connection also remains usable while the first handler is blocked.
+	fast := newFramePeer(t, address)
+	defer fast.Close()
+	fast.framer.ReadMetaHeaders = hpack.NewDecoder(4096, nil)
+	fast.get(t, 1, "/ready")
+	if body := fast.body(t, 1); body != "ready" {
+		t.Fatalf("other connection response: %q", body)
+	}
+	if stats := server.Stats(); stats.ActiveConnections != 2 || stats.RejectedRequests != 0 {
+		t.Fatalf("slow response affected other traffic: %+v", stats)
+	}
+
+	// Restoring stream credit must resume the existing response without reconnecting.
+	if err := slow.framer.WriteWindowUpdate(1, uint32(len(payload))); err != nil {
+		t.Fatal(err)
+	}
+	if body := slow.body(t, 1); body != payload {
+		t.Fatalf("resumed response mismatch: got %d bytes, want %d", len(body), len(payload))
+	}
+	select {
+	case err := <-written:
+		if err != nil {
+			t.Fatalf("resumed write: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("resumed write did not return")
+	}
+
+	// A new zero-window stream must release its handler when the client cancels.
+	slow.get(t, 5, "/large")
+	waitSignal(t, started)
+	if err := slow.framer.WriteRSTStream(5, xhttp2.ErrCodeCancel); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case err := <-written:
+		if err == nil {
+			t.Fatal("canceled zero-window write unexpectedly succeeded")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("RST_STREAM did not unblock the writer")
+	}
+
+	slow.get(t, 7, "/health")
+	readNoContent(t, slow, 7)
+	deadline := time.Now().Add(time.Second)
+	for server.Stats().ActiveRequests != 0 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if stats := server.Stats(); stats.ActiveRequests != 0 || stats.RSTStreams != 1 || stats.ProtocolErrors != 0 || stats.RejectedRequests != 0 {
+		t.Fatalf("flow control did not recover: %+v", stats)
+	}
+}
+
+func readNoContent(t *testing.T, peer *framePeer, streamID uint32) {
+	t.Helper()
+	for {
+		switch frame := peer.read(t).(type) {
+		case *xhttp2.MetaHeadersFrame:
+			if frame.StreamID != streamID {
+				continue
+			}
+			if frame.PseudoValue("status") != "204" || !frame.StreamEnded() {
+				t.Fatalf("stream %d: expected complete 204, got %+v", streamID, frame)
+			}
+			return
+		case *xhttp2.DataFrame:
+			t.Fatalf("unexpected DATA on zero-window stream %d", frame.StreamID)
+		case *xhttp2.RSTStreamFrame:
+			t.Fatalf("unexpected server reset: %+v", frame)
+		}
+	}
+}

transport/http2/lifecycle_test.go

@@ -20,7 +20,7 @@ type framePeer struct {
 	framer *xhttp2.Framer
 }
 
-func newFramePeer(t *testing.T, address string) *framePeer {
+func newFramePeer(t *testing.T, address string, settings ...xhttp2.Setting) *framePeer {
 	t.Helper()
 	c, err := net.DialTimeout("tcp", address, time.Second)
 	if err != nil {
@@ -32,7 +32,7 @@ func newFramePeer(t *testing.T, address string) *framePeer {
 	if _, err = io.WriteString(c, xhttp2.ClientPreface); err != nil {
 		t.Fatal(err)
 	}
-	if err = p.framer.WriteSettings(); err != nil {
+	if err = p.framer.WriteSettings(settings...); err != nil {
 		t.Fatal(err)
 	}
 	return p