From af5e96a318d015e431aba6486340853330971d5b Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 16:25:43 +0800 Subject: [PATCH] fix(tunnel): preserve full HTTP/2 session reuse across Go versions --- docs/ARCHITECTURE.md | 9 + internal/tunnel/web_h2_capacity_go127.go | 22 + internal/tunnel/web_h2_capacity_legacy.go | 11 + .../tunnel/web_h2_capacity_release_test.go | 116 ++++ internal/tunnel/web_h2_capacity_test.go | 481 +++++++++++++ internal/tunnel/web_h2_client.go | 120 +++- .../tunnel/web_h2_goaway_capacity_test.go | 632 ++++++++++++++++++ 7 files changed, 1368 insertions(+), 23 deletions(-) create mode 100644 internal/tunnel/web_h2_capacity_go127.go create mode 100644 internal/tunnel/web_h2_capacity_legacy.go create mode 100644 internal/tunnel/web_h2_capacity_release_test.go create mode 100644 internal/tunnel/web_h2_capacity_test.go create mode 100644 internal/tunnel/web_h2_goaway_capacity_test.go diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 2b09920..7205a62 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -158,6 +158,15 @@ connection and wake waiters to select a replacement. Reconnects and the new H2 connection selected after GOAWAY run the full bootstrap again. QUIC migration or NAT rebinding that remains the same `*quic.Conn` retains authentication state. +The H2 client queues at a healthy peer's stream-concurrency limit instead of +opening another physical connection just because the existing one is full. +On Go 1.27, the pinned HTTP/2 wrapper exposes immediately available capacity, +so its live, non-closing state is a conservative queue hint, not proof of usability. +If the underlying transport rejects a ready session before any request header +is encoded, a live caller may reselect once within the same establishment budget. +Encoded requests are never replayed, and accepted sibling streams may drain on +the old connection. Transport state inspection never holds the client owner lock. + Stopping an opening cancellation watcher is not the connection handoff. H2 and H3 recheck the caller's cancellation and deadline under their final ownership lock before publishing a TCP stream. Canceling an already handed-off diff --git a/internal/tunnel/web_h2_capacity_go127.go b/internal/tunnel/web_h2_capacity_go127.go new file mode 100644 index 0000000..7550a53 --- /dev/null +++ b/internal/tunnel/web_h2_capacity_go127.go @@ -0,0 +1,22 @@ +//go:build go1.27 && !http2legacy + +package tunnel + +import "golang.org/x/net/http2" + +// x/net v0.58's Go1.27 wrapper reports immediately available slots, not whether +// its strict-concurrency RoundTrip can queue. State uses memory locks in this +// implementation. Call outside the owner mutex (as on the legacy path). +func webH2CanTakeRequest(conn *http2.ClientConn) bool { + if conn.CanTakeNewRequest() { + return true + } + state := conn.State() + // This is a conservative queue attempt, not an exact native usability + // test: active GOAWAY/PROTOCOL_ERROR retirement is not fully exposed by + // the wrapper. RoundTrip remains authoritative; only an entirely unencoded + // rejection with a live context may reselect once. Do not compare capacity + // counts: the wrapper synthesizes them from separate native snapshots and + // can report mismatched values while a stream finishes or starts. + return !state.Closed && !state.Closing +} diff --git a/internal/tunnel/web_h2_capacity_legacy.go b/internal/tunnel/web_h2_capacity_legacy.go new file mode 100644 index 0000000..4f8692a --- /dev/null +++ b/internal/tunnel/web_h2_capacity_legacy.go @@ -0,0 +1,11 @@ +//go:build !(go1.27 && !http2legacy) + +package tunnel + +import "golang.org/x/net/http2" + +// The legacy transport's strict-concurrency check already allows queuing and +// excludes draining connections without taking its wire-write mutex. +func webH2CanTakeRequest(conn *http2.ClientConn) bool { + return conn.CanTakeNewRequest() +} diff --git a/internal/tunnel/web_h2_capacity_release_test.go b/internal/tunnel/web_h2_capacity_release_test.go new file mode 100644 index 0000000..435febb --- /dev/null +++ b/internal/tunnel/web_h2_capacity_release_test.go @@ -0,0 +1,116 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "testing" + "time" +) + +// A live caller queued by the real peer's one-stream limit must receive the +// released slot on the original authenticated connection, not merely cancel +// cleanly while queued. This reuses the bounded owned-wire fixture unchanged. +func TestWebH2CapacityReleaseResumesQueuedAuthenticatedStream(t *testing.T) { + f := newWebH2CapacityFixture(t) + first := f.open() + f.ack(first, []byte("hold first H2 capacity slot\x00\xff")) + firstRequest := f.request(0) + f.client.mu.Lock() + session := f.client.current + ready := session != nil && session.authState == webH2ClientAuthReady && session.auth != nil + var authentication *webSessionClientAuth + if ready { + authentication = session.auth + } + f.client.mu.Unlock() + if !ready { + t.Fatal("first stream did not establish ready physical authentication") + } + state := session.conn.ConnectionState() + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != webH2ALPN || firstRequest.short || firstRequest.physical == nil { + t.Fatalf("first stream TLS/ALPN/bootstrap=%#x/%q/%v", state.Version, state.NegotiatedProtocol, firstRequest.short) + } + observer := f.observe(session.h2) + f.checkpoint(observer, false) + + caller, cancel := context.WithCancel(f.ctx) + defer cancel() + result := make(chan webH2CapacityDialResult, 1) + joined := make(chan struct{}) + f.register(joined) + go func() { + defer close(joined) + conn, err := f.client.DialContext(caller, "tcp", f.target.Addr().String()) + f.own(conn) + result <- webH2CapacityDialResult{conn: conn, err: err} + }() + // This requires an actual request waiting for peer capacity. If an old + // transport replacement is observed, retain the failure but continue the + // real release/echo/independent-cleanup controls rather than abandon them. + f.checkpoint(observer, true) + if got := f.physicalDials.Load(); got != 1 { + t.Errorf("live queued caller caused %d physical dials before release, want 1", got) + } + if got := f.entropy.nonceReads.Load(); got != 1 { + t.Errorf("live queued caller generated %d full-auth nonces before release, want 1", got) + } + if got := f.requestCount(); got != 1 { + t.Errorf("server handlers before release=%d, want only held first stream", got) + } + select { + case <-joined: + t.Error("second public DialContext finished before actual first stream capacity release") + default: + } + if err := caller.Err(); err != nil { + t.Fatalf("queued caller expired before capacity release: %v", err) + } + f.ack(first, []byte("first stream remains usable with a live waiter")) + f.finish(first, nil) + f.require(firstRequest.done, "first handler joined after actual FIN/capacity release") + f.targetResult() + + f.require(joined, "live queued caller resumed after first stream release") + second := <-result // Result publication precedes the joined channel close. + if second.err != nil || second.conn == nil { + t.Fatalf("released queued public DialContext: %v", second.err) + } + if err := second.conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + secondRequest := f.request(1) + // Establishment cancellation is detached only after successful handoff. + // Canceling now must not interrupt the complete payload or either FIN. + cancel() + payload := bytes.Repeat([]byte{'q', 'u', 'e', 'u', 'e', 'd', 0, 255}, 8192) + f.finish(second.conn, payload) + f.targetResult() + f.require(secondRequest.done, "resumed second handler completed after full FIN echo") + f.sameSession(session) + f.client.mu.Lock() + sameAuth := f.client.current == session && session.auth == authentication + f.client.mu.Unlock() + if !sameAuth { + t.Error("capacity release replaced original physical session or authentication key") + } + requests := f.requestsSnapshot() + if len(requests) != 2 { + t.Errorf("server requests=%d, want full bootstrap then one short continuation", len(requests)) + } else if requests[0].physical != requests[1].physical || requests[0].short || !requests[1].short { + t.Error("queued continuation did not preserve actual same-physical full/short authentication") + } + if got := f.physicalDials.Load(); got != 1 { + t.Errorf("physical dials=%d after queued payload, want 1", got) + } + if got := f.entropy.nonceReads.Load(); got != 1 { + t.Errorf("full-auth nonces=%d after queued payload, want 1", got) + } + if got := f.destinationDials.Load(); got != 2 { + t.Errorf("actual target dials=%d, want 2", got) + } + if len(f.admission.sem) != 0 { + t.Error("joined capacity-release handlers retained stream admission") + } + t.Log("real capacity=1 pending caller resumed on the original TLS13/h2 session; full/short auth, 64 KiB FIN echo, detached caller cancellation, and owned worker joins") +} diff --git a/internal/tunnel/web_h2_capacity_test.go b/internal/tunnel/web_h2_capacity_test.go new file mode 100644 index 0000000..1cf5772 --- /dev/null +++ b/internal/tunnel/web_h2_capacity_test.go @@ -0,0 +1,481 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" + "golang.org/x/net/http2" +) + +// An occupied, healthy HTTP/2 connection is not a failed physical connection. +// The peer really advertises one stream and the first authenticated stream +// remains open. The second caller must queue on that connection, not redial. +func TestWebH2CapacityQueuesWithoutReplacingAuthenticatedSession(t *testing.T) { + f := newWebH2CapacityFixture(t) + first := f.open() + f.ack(first, []byte("held authenticated stream\x00\xff")) + firstRequest := f.request(0) + f.client.mu.Lock() + session := f.client.current + ready := session != nil && session.authState == webH2ClientAuthReady && session.auth != nil + f.client.mu.Unlock() + if !ready { + t.Fatal("first stream did not establish a ready authenticated session") + } + state := session.conn.ConnectionState() + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != webH2ALPN || firstRequest.short || firstRequest.physical == nil { + t.Fatalf("first stream TLS/ALPN/bootstrap=%#x/%q/%v", state.Version, state.NegotiatedProtocol, firstRequest.short) + } + observer := f.observe(session.h2) + f.checkpoint(observer, false) + if got := f.physicalDials.Load(); got != 1 { + t.Fatalf("first stream used %d physical dials", got) + } + + cause := errors.New("capacity fixture private caller cancellation") + caller, cancel := context.WithCancelCause(f.ctx) + defer cancel(context.Canceled) + result := make(chan webH2CapacityDialResult, 1) + joined := make(chan struct{}) + f.register(joined) + go func() { + defer close(joined) + conn, err := f.client.DialContext(caller, "tcp", f.target.Addr().String()) + f.own(conn) + result <- webH2CapacityDialResult{conn: conn, err: err} + }() + // State.Pending is a real HTTP/2 request waiting for a concurrency slot. + // The alternative is an actual second physical DialContext invocation. + // A caller AfterFunc/Done registration is deliberately not a wait witness. + f.checkpoint(observer, true) + if got := f.physicalDials.Load(); got != 1 { + t.Errorf("occupied healthy connection caused %d physical dials, want 1", got) + } + if got := f.entropy.nonceReads.Load(); got != 1 { + t.Errorf("queued caller generated %d full-auth nonces, want 1", got) + } + if got := f.requestCount(); got != 1 { + t.Errorf("queued caller reached %d server handlers before capacity release, want 1", got) + } + cancel(cause) + f.require(joined, "second caller joined after cancellation") + second := <-result // Publication precedes the joined signal. + if second.conn != nil { + _ = second.conn.Close() + t.Error("canceled queued caller returned a connection") + } + if second.err != cause { + t.Errorf("queued caller error=%v, want exact private cancellation cause", second.err) + } + f.ack(first, []byte("first stream survives sibling cancel")) + f.sameSession(session) + f.finish(first, nil) + f.require(firstRequest.done, "first server handler completed after FIN") + f.targetResult() + + // After genuine capacity release, both independent live callers must use + // short authentication on the original physical connection and carry data. + for i := 0; i < 2; i++ { + conn := f.open() + payload := bytes.Repeat([]byte{byte(i), 'h', '2', 'q', 0, 255}, 2048) + f.finish(conn, payload) + f.targetResult() + f.sameSession(session) + } + for _, request := range f.requestsSnapshot() { + f.require(request.done, "observed server handler joined") + } + requests := f.requestsSnapshot() + if len(requests) != 3 { + t.Errorf("server handlers=%d, want initial full plus two healthy short requests", len(requests)) + } else { + for i, request := range requests { + if request.physical != firstRequest.physical || request.short != (i != 0) { + t.Errorf("request %d did not preserve physical auth/full-short sequence", i) + } + } + } + if got := f.physicalDials.Load(); got != 1 { + t.Errorf("physical dials=%d after healthy continuations, want 1", got) + } + if got := f.entropy.nonceReads.Load(); got != 1 { + t.Errorf("full-auth nonces=%d after healthy continuations, want 1", got) + } + if got := f.destinationDials.Load(); got != 3 { + t.Errorf("actual destination dials=%d, want 3", got) + } + if len(f.admission.sem) != 0 { + t.Error("completed handlers retained stream admission") + } + t.Log("capacity=1: actual pending-or-redial witness, exact caller cause, surviving first stream, two complete FIN echoes, full/short/short on one physical connection") +} + +type webH2CapacityDialResult struct { + conn net.Conn + err error +} + +type webH2CapacityRequest struct { + physical *webServerConnectionAuth + short bool + done chan struct{} +} + +type webH2CapacityObserver struct { + mu sync.Mutex + state http2.ClientConnState + changed chan struct{} +} + +type webH2CapacityFixture struct { + t *testing.T + ctx context.Context + cancel context.CancelFunc + client *WebH2Client + server *WebH2Server + target *net.TCPListener + admission *StreamAdmission + entropy webH2AuthEntropyCounter + physicalDials atomic.Int64 + destinationDials atomic.Int64 + extraPhysical chan struct{} + results chan error + mu sync.Mutex + closed bool + conns []net.Conn + joins []<-chan struct{} + requests []webH2CapacityRequest + serveErr error +} + +func newWebH2CapacityFixture(t *testing.T) *webH2CapacityFixture { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 6*time.Second) + f := &webH2CapacityFixture{t: t, ctx: ctx, cancel: cancel, extraPhysical: make(chan struct{}, 1), results: make(chan error, 4)} + t.Cleanup(f.cleanup) + var err error + f.target, err = net.ListenTCP("tcp4", &net.TCPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + targetDone := make(chan struct{}) + f.register(targetDone) + go func() { + defer close(targetDone) + for { + conn, err := f.target.AcceptTCP() + if err != nil { + return + } + if !f.own(conn) { + continue + } + done := make(chan struct{}) + f.register(done) + go func() { + defer close(done) + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + _, err := io.Copy(conn, conn) + if err == nil { + err = conn.CloseWrite() + } + select { + case f.results <- err: + case <-f.ctx.Done(): + } + }() + } + }() + f.admission, err = NewStreamAdmission(1) + if err != nil { + t.Fatal(err) + } + serverTLS, clientTLS := testTLSConfigs(t) + f.server, err = ListenWebH2(WebH2ServerConfig{Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, + StreamAdmission: f.admission, Cover: http.NotFoundHandler(), Dialer: transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != f.target.Addr().String() { + return nil, fmt.Errorf("fixture refused unowned target %q/%q", network, address) + } + f.destinationDials.Add(1) + conn, err := (&net.Dialer{}).DialContext(ctx, "tcp4", address) + f.own(conn) + return conn, err + })}) + if err != nil { + t.Fatal(err) + } + handler := f.server.server.Handler + f.server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + physical, _ := r.Context().Value(webServerConnectionAuthContextKey{}).(*webServerConnectionAuth) + _, short := parseWebSessionBearer(r.Header.Get("Proxy-Authorization")) + request := webH2CapacityRequest{physical: physical, short: short, done: make(chan struct{})} + f.mu.Lock() + if f.closed { + f.mu.Unlock() + return + } + f.requests = append(f.requests, request) + f.joins = append(f.joins, request.done) + f.mu.Unlock() + defer close(request.done) + handler.ServeHTTP(w, r) + }) + f.client, err = newWebH2ClientWithSigner(WebH2ClientConfig{ServerAddress: f.server.Addr().String(), Token: testToken, + TLSConfig: clientTLS, FingerprintProfile: FingerprintNative}, newWebAuthSigner(mustWebAuthKey(t, testToken), nil, &f.entropy), webAuthClaims{}) + if err != nil { + t.Fatal(err) + } + f.client.dialer = transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != f.server.Addr().String() { + return nil, fmt.Errorf("fixture refused unowned relay %q/%q", network, address) + } + if f.physicalDials.Add(1) > 1 { + select { + case f.extraPhysical <- struct{}{}: + default: + } + } + conn, err := (&net.Dialer{}).DialContext(ctx, "tcp4", address) + f.own(conn) + return conn, err + }) + serveDone := make(chan struct{}) + f.register(serveDone) + go func() { defer close(serveDone); f.serveErr = f.server.Serve(f.ctx) }() + return f +} + +func (f *webH2CapacityFixture) own(conn net.Conn) bool { + if conn == nil { + return false + } + f.mu.Lock() + if f.closed { + f.mu.Unlock() + _ = conn.Close() + return false + } + f.conns = append(f.conns, conn) + f.mu.Unlock() + return true +} + +func (f *webH2CapacityFixture) register(done <-chan struct{}) { + f.mu.Lock() + f.joins = append(f.joins, done) + f.mu.Unlock() +} + +func (f *webH2CapacityFixture) require(done <-chan struct{}, what string) { + f.t.Helper() + select { + case <-done: + case <-f.ctx.Done(): + f.t.Fatalf("%s: %v", what, context.Cause(f.ctx)) + } +} + +func (f *webH2CapacityFixture) open() net.Conn { + f.t.Helper() + ctx, cancel := context.WithTimeout(f.ctx, 2*time.Second) + defer cancel() + conn, err := f.client.DialContext(ctx, "tcp", f.target.Addr().String()) + f.own(conn) + if err != nil || conn == nil { + f.t.Fatalf("healthy public H2 DialContext: %v", err) + } + if err := conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + f.t.Fatal(err) + } + return conn +} + +func (f *webH2CapacityFixture) ack(conn net.Conn, payload []byte) { + f.t.Helper() + if _, err := io.Copy(conn, bytes.NewReader(payload)); err != nil { + f.t.Fatal(err) + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil || !bytes.Equal(got, payload) { + f.t.Fatalf("live held stream echo=%d bytes, error=%v", len(got), err) + } +} + +func (f *webH2CapacityFixture) finish(conn net.Conn, payload []byte) { + f.t.Helper() + if _, err := io.Copy(conn, bytes.NewReader(payload)); err != nil { + f.t.Fatal(err) + } + if err := conn.(interface{ CloseWrite() error }).CloseWrite(); err != nil { + f.t.Fatal(err) + } + got, err := io.ReadAll(conn) + _ = conn.Close() + if err != nil || !bytes.Equal(got, payload) { + f.t.Fatalf("FIN echo=%d/%d, error=%v", len(got), len(payload), err) + } +} + +func (f *webH2CapacityFixture) targetResult() { + f.t.Helper() + select { + case err := <-f.results: + if err != nil { + f.t.Fatalf("actual target EOF/echo: %v", err) + } + case <-f.ctx.Done(): + f.t.Fatal("actual target did not complete") + } +} + +func (f *webH2CapacityFixture) request(index int) webH2CapacityRequest { + f.t.Helper() + f.mu.Lock() + defer f.mu.Unlock() + if len(f.requests) <= index { + f.t.Fatal("successful CONNECT had no server handler observation") + } + return f.requests[index] +} + +func (f *webH2CapacityFixture) requestCount() int { return len(f.requestsSnapshot()) } + +func (f *webH2CapacityFixture) requestsSnapshot() []webH2CapacityRequest { + f.mu.Lock() + defer f.mu.Unlock() + return append([]webH2CapacityRequest(nil), f.requests...) +} + +func (f *webH2CapacityFixture) sameSession(want *webH2ClientSession) { + f.client.mu.Lock() + defer f.client.mu.Unlock() + if f.client.current != want || want.authState != webH2ClientAuthReady || want.auth == nil || len(f.client.sessions) != 1 { + f.t.Error("capacity/caller cancellation replaced or invalidated the ready physical session") + } +} + +func (f *webH2CapacityFixture) observe(conn *http2.ClientConn) *webH2CapacityObserver { + observer := &webH2CapacityObserver{changed: make(chan struct{}, 1)} + done := make(chan struct{}) + f.register(done) + go func() { + defer close(done) + tick := time.NewTicker(5 * time.Millisecond) + defer tick.Stop() + for { + // Legacy State may acquire a wire-write lock. Never hold client.mu + // here; independent cleanup closes the real raw socket before joining. + state := conn.State() + observer.mu.Lock() + observer.state = state + observer.mu.Unlock() + select { + case observer.changed <- struct{}{}: + default: + } + select { + case <-tick.C: + case <-f.ctx.Done(): + return + } + } + }() + return observer +} + +func (f *webH2CapacityFixture) checkpoint(observer *webH2CapacityObserver, pending bool) { + f.t.Helper() + limit := time.NewTimer(time.Second) + defer limit.Stop() + for { + observer.mu.Lock() + state := observer.state + observer.mu.Unlock() + if state.MaxConcurrentStreams == 1 && state.StreamsActive >= 1 && !state.Closed && !state.Closing && (!pending || state.StreamsPending >= 1) { + f.t.Logf("real H2 capacity witness: active=%d pending=%d max=%d", state.StreamsActive, state.StreamsPending, state.MaxConcurrentStreams) + return + } + select { + case <-observer.changed: + case <-f.extraPhysical: + f.t.Error("second actual physical DialContext entered while original H2 connection was occupied and healthy") + return // Continue cancellation, real healthy recovery and cleanup. + case <-limit.C: + f.t.Fatalf("no actual H2 capacity/pending witness: %+v", state) + case <-f.ctx.Done(): + f.t.Fatal("independent fixture expired at capacity checkpoint") + } + } +} + +func (f *webH2CapacityFixture) cleanup() { + f.mu.Lock() + f.closed = true + conns := append([]net.Conn(nil), f.conns...) + f.mu.Unlock() + f.cancel() + if f.target != nil { + _ = f.target.Close() + } + // Includes actual client raw sockets, not just returned logical streams. + // Raw closure releases any legacy State/HTTP2 write-lock dependency first. + for _, conn := range conns { + _ = conn.Close() + } + closed := make(chan struct{}) + var closeErr error + go func() { + defer close(closed) + if f.client != nil { + closeErr = f.client.Close() + } + if f.server != nil { + closeErr = errors.Join(closeErr, f.server.Close()) + } + }() + limit := time.NewTimer(2 * time.Second) + defer limit.Stop() + select { + case <-closed: + if !destinationDialOwnershipOnlyClosed(closeErr) { + f.t.Errorf("client/server cleanup: %v", closeErr) + } + case <-limit.C: + f.t.Error("client/server Close did not join") + return + } + for i := 0; ; i++ { + f.mu.Lock() + if i == len(f.joins) { + f.mu.Unlock() + break + } + done := f.joins[i] + f.mu.Unlock() + select { + case <-done: + case <-limit.C: + f.t.Errorf("owned worker %d did not join", i) + return + } + } + if f.serveErr != nil { + f.t.Errorf("owned Serve: %v", f.serveErr) + } + if f.admission != nil && len(f.admission.sem) != 0 { + f.t.Error("cleanup retained stream admission") + } + f.t.Log("all owned raw sockets, target/handler/caller/State/Serve workers and client/server Close joined") +} diff --git a/internal/tunnel/web_h2_client.go b/internal/tunnel/web_h2_client.go index fc19d68..17082ca 100644 --- a/internal/tunnel/web_h2_client.go +++ b/internal/tunnel/web_h2_client.go @@ -8,8 +8,10 @@ import ( "io" "net" "net/http" + "net/http/httptrace" "net/url" "sync" + "sync/atomic" "time" "github.com/cppla/autocar/internal/protocol" @@ -244,8 +246,9 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) // the returned net.Conn. The client's lifetime remains the stream parent. streamCtx, streamCancelCause := context.WithCancelCause(c.ctx) streamCancel := func() { streamCancelCause(context.Canceled) } + openDeadline := time.Now().Add(c.handshakeTimeout) openTimeoutDone := make(chan struct{}) - openTimeout := time.AfterFunc(c.handshakeTimeout, func() { + openTimeout := time.AfterFunc(time.Until(openDeadline), func() { streamCancelCause(context.DeadlineExceeded) close(openTimeoutDone) }) @@ -266,17 +269,42 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) stopOpenTimeout() streamCancel() } + for attempt := 0; ; attempt++ { + conn, err, unsent := c.dialStreamOnce(ctx, streamCtx, openDeadline, authority, binding, cancelStream, stopDial, stopOpenTimeout) + if err == nil { + return conn, nil + } + if attempt != 0 || !unsent { + cancelStream() + return nil, err + } + // A busy Go1.27 connection can also be draining: its public state + // does not expose every native retirement reason. Retry once only + // after RoundTrip rejected an entirely unencoded request. Keep the + // SAME caller link and establishment budget across both attempts. + } +} + +// Each attempt owns its pipe, stream cancellation and authentication exchange. +// Returning before reselection releases its opening reservation and short-ticket +// window slot; a draining physical connection can keep its existing siblings. +func (c *WebH2Client) dialStreamOnce( + ctx, streamCtx context.Context, openDeadline time.Time, + authority string, binding webAuthBinding, + cancelStream func(), stopDial func() bool, stopOpenTimeout func() bool, +) (net.Conn, error, bool) { + requestCtx, requestCancel := context.WithCancel(streamCtx) requestReader, requestWriter := io.Pipe() fail := func() { - cancelStream() + requestCancel() _ = requestWriter.Close() _ = requestReader.Close() } - reservation, err := c.reserveSession(streamCtx) + reservation, err := c.reserveSession(requestCtx) if err != nil { fail() - return nil, err + return nil, err, false } session := reservation.session defer c.releaseSessionReservation(session) @@ -288,7 +316,7 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) if reservation.bootstrap { bearer, authClaims, err = c.auth.authorization(binding, c.claims) } else { - bearer, exchange, err = reservation.auth.authorization(streamCtx, binding) + bearer, exchange, err = reservation.auth.authorization(requestCtx, binding) defer reservation.auth.complete(exchange) } if err != nil { @@ -296,8 +324,15 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) if reservation.bootstrap || errors.Is(err, errWebSessionSequenceExhausted) { c.failSessionAuthentication(session) } - return nil, err - } + return nil, err, false + } + var headersStarted atomic.Bool + requestCtx = httptrace.WithClientTrace(requestCtx, &httptrace.ClientTrace{ + // WroteHeaders is too late: a partial wire write can already have + // reached the peer. In the pinned HTTP/2 implementation this hook + // runs while encoding the first field, BEFORE writing any HEADERS. + WroteHeaderField: func(string, []string) { headersStarted.Store(true) }, + }) request := (&http.Request{ Method: http.MethodConnect, URL: &url.URL{Scheme: "https", Host: authority}, @@ -305,22 +340,30 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) Header: webConnectRequestHeaders(bearer), Body: requestReader, ContentLength: -1, - }).WithContext(streamCtx) + }).WithContext(requestCtx) request.Header.Set("Content-Type", "application/octet-stream") response, err := session.h2.RoundTrip(request) if err != nil { - cause := context.Cause(streamCtx) + cause := contextError(requestCtx) + // Direct ClientConn.RoundTrip joins its request writer on a + // non-context error. With a live request context, no writer can + // start encoding after this check. Request.Cancel is deliberately + // nil. Never retry bootstrap, cancellation, or any encoded request. + unsent := !reservation.bootstrap && cause == nil && contextError(ctx) == nil && time.Now().Before(openDeadline) && + !headersStarted.Load() && !session.h2.CanTakeNewRequest() fail() if reservation.bootstrap { c.failSessionAuthentication(session) + } else if unsent { + c.retireSessionSelection(session) } else { c.noteSessionFailure(session) } if cause != nil && !errors.Is(cause, context.Canceled) { - return nil, cause + return nil, cause, false } - return nil, fmt.Errorf("tunnel: web-cover HTTP/2 CONNECT: %w", err) + return nil, fmt.Errorf("tunnel: web-cover HTTP/2 CONNECT: %w", err), unsent } if reservation.bootstrap { sessionAuth, ok := acceptWebSessionBootstrap( @@ -334,7 +377,7 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) _ = response.Body.Close() fail() c.failSessionAuthentication(session) - return nil, errors.New("tunnel: web-cover HTTP/2 server authentication failed") + return nil, errors.New("tunnel: web-cover HTTP/2 server authentication failed"), false } } else if !reservation.auth.verifyResponseProof( response.Header.Values(webAuthResponseHeader), @@ -345,12 +388,12 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) _ = response.Body.Close() fail() c.failSessionAuthentication(session) - return nil, errors.New("tunnel: web-cover HTTP/2 server authentication failed") + return nil, errors.New("tunnel: web-cover HTTP/2 server authentication failed"), false } if response.StatusCode != http.StatusOK { _ = response.Body.Close() fail() - return nil, &WebConnectError{Transport: webAuthTransportH2, StatusCode: response.StatusCode} + return nil, &WebConnectError{Transport: webAuthTransportH2, StatusCode: response.StatusCode}, false } // Stop the establishment-only cancellation link before handing ownership // to the caller. If cancellation won the race with the response, fail the @@ -358,13 +401,13 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) if !stopOpenTimeout() { _ = response.Body.Close() fail() - return nil, context.DeadlineExceeded + return nil, context.DeadlineExceeded, false } stopDial() if err := contextError(ctx); err != nil { _ = response.Body.Close() fail() - return nil, err + return nil, err, false } c.mu.Lock() @@ -372,7 +415,7 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) c.mu.Unlock() _ = response.Body.Close() fail() - return nil, net.ErrClosed + return nil, net.ErrClosed, false } // Detaching the opening watcher is not the ownership handoff. A caller // canceled while queued on mu still owns only this request, not its session. @@ -380,7 +423,7 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) c.mu.Unlock() _ = response.Body.Close() fail() - return nil, err + return nil, err, false } c.selected = true session.active++ @@ -388,12 +431,12 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) conn := newWebH2Conn( response.Body, requestWriter, - cancelStream, + func() { requestCancel(); cancelStream() }, session.conn.LocalAddr(), session.conn.RemoteAddr(), ) conn.onClose = func() { c.releaseSessionStream(session) } - return conn, nil + return conn, nil, false } func normalizeWebH2Authority(address string) (string, error) { @@ -477,9 +520,27 @@ func (c *WebH2Client) reserveSession(ctx context.Context) (webH2SessionReservati continue } if session := c.current; session != nil { + state := session.authState + canTake := false + if state == webH2ClientAuthFresh || state == webH2ClientAuthReady { + // Even legacy State can wait behind a wire write. Keep every + // transport inspection outside mu so Close can release I/O. + c.mu.Unlock() + if state == webH2ClientAuthReady { + canTake = webH2CanTakeRequest(session.h2) + } else { + canTake = session.h2.CanTakeNewRequest() + } + c.mu.Lock() + if c.closed || contextError(ctx) != nil || c.current != session || session.authState != state { + c.mu.Unlock() + <-c.dialGate + continue + } + } switch session.authState { case webH2ClientAuthFresh: - if session.h2.CanTakeNewRequest() { + if canTake { // Claim bootstrap only when a live caller reserves a stream, // not when a background worker publishes a cold connection. session.authState = webH2ClientAuthBootstrapping @@ -490,7 +551,7 @@ func (c *WebH2Client) reserveSession(ctx context.Context) (webH2SessionReservati } c.current = nil case webH2ClientAuthReady: - if session.h2.CanTakeNewRequest() { + if canTake { session.opening++ auth := session.auth c.mu.Unlock() @@ -780,8 +841,21 @@ func (c *WebH2Client) failSessionAuthentication(session *webH2ClientSession) { } func (c *WebH2Client) noteSessionFailure(session *webH2ClientSession) { + canTake := webH2CanTakeRequest(session.h2) + c.mu.Lock() + if c.current == session && !canTake { + c.current = nil + } + retired := c.cleanupIdleSessionsLocked() + c.mu.Unlock() + c.closeRetiredSessions(retired) +} + +// A request rejected before encoding may select a new physical connection, +// but must not abort streams already accepted on the draining connection. +func (c *WebH2Client) retireSessionSelection(session *webH2ClientSession) { c.mu.Lock() - if c.current == session && !session.h2.CanTakeNewRequest() { + if c.current == session { c.current = nil } retired := c.cleanupIdleSessionsLocked() diff --git a/internal/tunnel/web_h2_goaway_capacity_test.go b/internal/tunnel/web_h2_goaway_capacity_test.go new file mode 100644 index 0000000..162e56a --- /dev/null +++ b/internal/tunnel/web_h2_goaway_capacity_test.go @@ -0,0 +1,632 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" + "golang.org/x/net/http2" +) + +// Capacity is not an unusable-connection signal, but an active GOAWAY is. +// A genuine client PING_ACK, ordered after the injected GOAWAY, establishes +// that the client read loop processed it before the next public DialContext. +func TestWebH2CapacityDoesNotQueueOnDrainingConnection(t *testing.T) { + f := newWebH2GoAwayFixture(t) + first := f.open() + f.ack(first, []byte("first authenticated stream remains open\x00\xff")) + initial := f.request(0) + f.client.mu.Lock() + old := f.client.current + ready := old != nil && old.authState == webH2ClientAuthReady && old.auth != nil + f.client.mu.Unlock() + if !ready || initial.short || initial.physical == nil { + t.Fatal("first stream did not establish physical-connection authentication") + } + state := old.conn.ConnectionState() + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != webH2ALPN || len(state.VerifiedChains) == 0 { + t.Fatalf("verified first TLS/ALPN=%#x/%q/chains%d", state.Version, state.NegotiatedProtocol, len(state.VerifiedChains)) + } + if err := initial.wire.goAwayAndPing(); err != nil { + t.Fatalf("actual GOAWAY/PING write: %v", err) + } + f.require(initial.wire.acked, "client processed GOAWAY then acknowledged following PING") + // The old physical connection is still carrying the admitted stream. + // Its continued echo rules out a connection-level shutdown as the witness. + f.ack(first, []byte("old sibling survives real GOAWAY")) + + second := f.open() // One public call must select/reselect a fresh physical connection. + secondRequest := f.request(1) + f.client.mu.Lock() + fresh := f.client.current + f.client.mu.Unlock() + if fresh == nil || fresh == old || fresh.conn == old.conn || secondRequest.wire == initial.wire || secondRequest.physical == initial.physical || secondRequest.short { + t.Error("new caller did not establish full authentication on a different physical connection") + } + f.finish(second, bytes.Repeat([]byte{'n', 'e', 'w', 0, 255, 1}, 2048)) + f.require(secondRequest.done, "new physical connection handler completed") + f.targetResult() + + // Retiring the selected connection must not kill an already admitted sibling. + f.ack(first, []byte("old sibling survives new authenticated connection")) + f.finish(first, nil) + f.require(initial.done, "old sibling completed its upload and reply FIN") + f.targetResult() + + third := f.open() + thirdRequest := f.request(2) + f.finish(third, bytes.Repeat([]byte{'s', 'h', 'o', 'r', 't', 0}, 2048)) + f.require(thirdRequest.done, "new physical connection short-auth handler completed") + f.targetResult() + f.client.mu.Lock() + sameFresh := f.client.current == fresh && fresh != nil && fresh.authState == webH2ClientAuthReady + f.client.mu.Unlock() + if !sameFresh || thirdRequest.wire != secondRequest.wire || thirdRequest.physical != secondRequest.physical || !thirdRequest.short { + t.Error("later healthy caller did not reuse the new physical connection with short authentication") + } + requests := f.requestsSnapshot() + if len(requests) != 3 { + t.Errorf("actual authenticated handlers=%d, want full/full/short", len(requests)) + } + if got := f.physicalDials.Load(); got != 2 { + t.Errorf("actual physical dials=%d, want 2", got) + } + if got := f.accepted.Load(); got != 2 { + t.Errorf("accepted physical TLS connections=%d, want 2", got) + } + if got := f.entropy.nonceReads.Load(); got != 2 { + t.Errorf("full-auth nonce reads=%d, want 2", got) + } + if got := f.destinationDials.Load(); got != 3 { + t.Errorf("actual owned destination dials=%d, want 3", got) + } + if len(f.admission.sem) != 0 { + t.Error("complete FIN exchanges retained stream admission") + } + t.Log("real TLS13/h2: GOAWAY then observed PING_ACK; old admitted sibling survives; one public caller reconnects; full/full/short on two physical connections; three complete FIN echoes") +} + +// The second request is really decoded and authenticated by the peer before +// it sends a protocol reset. Unlike an unencoded GOAWAY rejection, this is +// not safe to replay: a live context alone must not authorize another dial. +func TestWebH2EncodedProtocolResetDoesNotRetry(t *testing.T) { + f := newWebH2GoAwayFixtureConfigured(t, 2, true) + first := f.open() + f.ack(first, []byte("first sibling before encoded protocol reset")) + initial := f.request(0) + f.client.mu.Lock() + old := f.client.current + ready := old != nil && old.authState == webH2ClientAuthReady && old.auth != nil + f.client.mu.Unlock() + if !ready || initial.short || initial.physical == nil { + t.Fatal("first stream did not establish physical-connection authentication") + } + state := old.conn.ConnectionState() + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != webH2ALPN || len(state.VerifiedChains) == 0 { + t.Fatal("first stream did not verify TLS13/h2") + } + ctx, cancel := context.WithTimeout(f.ctx, 2*time.Second) + conn, err := f.client.DialContext(ctx, "tcp", f.target.Addr().String()) + callerLive := ctx.Err() == nil + cancel() + f.own(conn) + if conn != nil { + _ = conn.Close() + t.Error("encoded protocol reset returned a successful connection") + } + var reset http2.StreamError + if !errors.As(err, &reset) || reset.StreamID != 3 || reset.Code != http2.ErrCodeProtocol { + t.Errorf("encoded reset result=%v, want peer PROTOCOL_ERROR on stream 3", err) + } + if !callerLive { + t.Error("protocol reset result consumed the caller deadline instead of returning its stream error") + } + f.require(f.resetWritten, "actual protocol reset written after short authentication") + f.releaseReset() + secondRequest := f.request(1) + f.require(secondRequest.done, "reset request handler completed") + if !secondRequest.short || secondRequest.wire != initial.wire || secondRequest.physical != initial.physical { + t.Error("encoded second request did not use the first physical connection's short authentication") + } + if got := f.physicalDials.Load(); got != 1 { + t.Errorf("encoded rejected request automatically made %d physical dials, want 1", got) + } + if got := f.entropy.nonceReads.Load(); got != 1 { + t.Errorf("encoded rejected request automatically generated %d full-auth nonces, want 1", got) + } + if got := f.destinationDials.Load(); got != 2 || len(f.requestsSnapshot()) != 2 { + t.Error("encoded rejected request was retried at the authenticated peer") + } + f.ack(first, []byte("first sibling survives encoded protocol reset")) + f.finish(first, nil) + f.require(initial.done, "surviving first sibling completed its FIN exchange") + f.targetResult() + + // A later independent caller can recover on a new physical connection. + healthy := f.open() + recovery := f.request(2) + f.finish(healthy, bytes.Repeat([]byte{'r', 'e', 'c', 'o', 'v', 0}, 2048)) + f.require(recovery.done, "independent recovery handler completed") + f.targetResult() + if recovery.short || recovery.wire == initial.wire || recovery.physical == initial.physical { + t.Error("independent recovery did not establish fresh physical authentication") + } + if f.physicalDials.Load() != 2 || f.accepted.Load() != 2 || f.entropy.nonceReads.Load() != 2 || f.destinationDials.Load() != 3 || len(f.requestsSnapshot()) != 3 { + t.Error("independent healthy recovery did not use exactly one fresh physical dial/bootstrap/destination") + } + if len(f.admission.sem) != 0 { + t.Error("completed reset/recovery handlers retained stream admission") + } + t.Log("actual short-auth target dial entry then peer RST_STREAM(PROTOCOL_ERROR); live caller returns typed stream error with no retry; first sibling survives FIN; later independent caller recovers") +} + +type webH2GoAwayWireContextKey struct{} + +type webH2GoAwayRequest struct { + physical *webServerConnectionAuth + short bool + wire *webH2GoAwayTLSConn + done chan struct{} +} + +type webH2GoAwayFixture struct { + t *testing.T + ctx context.Context + cancel context.CancelFunc + client *WebH2Client + relay *net.TCPListener + target *net.TCPListener + admission *StreamAdmission + entropy webH2AuthEntropyCounter + physicalDials atomic.Int64 + accepted atomic.Int64 + destinationDials atomic.Int64 + results chan error + mu sync.Mutex + closed bool + conns []net.Conn + joins []<-chan struct{} + requests []webH2GoAwayRequest + workerErrors []error + resetWritten chan struct{} + resetRelease chan struct{} + resetReleaseOnce sync.Once +} + +func newWebH2GoAwayFixture(t *testing.T) *webH2GoAwayFixture { + return newWebH2GoAwayFixtureConfigured(t, 1, false) +} + +func newWebH2GoAwayFixtureConfigured(t *testing.T, maxStreams uint32, resetSecond bool) *webH2GoAwayFixture { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + f := &webH2GoAwayFixture{t: t, ctx: ctx, cancel: cancel, results: make(chan error, 4), resetWritten: make(chan struct{}), resetRelease: make(chan struct{})} + t.Cleanup(f.cleanup) + var err error + f.target, err = net.ListenTCP("tcp4", &net.TCPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + f.relay, err = net.ListenTCP("tcp4", &net.TCPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + f.admission, err = NewStreamAdmission(2) + if err != nil { + t.Fatal(err) + } + core, err := newServerCoreWithAdmission(testToken, transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != f.target.Addr().String() { + return nil, fmt.Errorf("fixture refused unowned destination %q/%q", network, address) + } + if sequence := f.destinationDials.Add(1); resetSecond && sequence == 2 { + // The real handler authenticates before invoking its target dialer. + // This peer reset therefore follows receipt of encoded short-ticket + // HEADERS, but no second actual destination socket is opened. + wire, _ := ctx.Value(webH2GoAwayWireContextKey{}).(*webH2GoAwayTLSConn) + if wire == nil { + return nil, errors.New("owned target dial omitted its physical TLS context") + } + if err := wire.protocolReset(); err != nil { + f.workerError(err) + return nil, err + } + close(f.resetWritten) + // Do not race the intended peer stream error with a subsequent + // signed 502: native RoundTrip favors simultaneously ready headers. + // This fixture gate is released only after the caller observes the + // reset, and is independently released on every cleanup path. + select { + case <-f.resetRelease: + case <-ctx.Done(): + } + return nil, errors.New("owned target refused after an actual encoded-request protocol reset") + } + conn, err := (&net.Dialer{}).DialContext(ctx, "tcp4", address) + f.own(conn) + return conn, err + }), 2*time.Second, 2*time.Second, 0, 2, f.admission) + if err != nil { + t.Fatal(err) + } + auth, err := newWebAuthVerifier(mustWebAuthKey(t, testToken), nil, 0) + if err != nil { + t.Fatal(err) + } + handler := &webTunnelHandler{core: core, auth: auth, cover: http.NotFoundHandler()} + serverTLS, clientTLS := testTLSConfigs(t) + serverTLS, err = webServerTLSConfig(serverTLS, webH2ALPN) + if err != nil { + t.Fatal(err) + } + f.client, err = newWebH2ClientWithSigner(WebH2ClientConfig{ServerAddress: f.relay.Addr().String(), Token: testToken, + TLSConfig: clientTLS, FingerprintProfile: FingerprintNative, DialTimeout: 2 * time.Second, HandshakeTimeout: 2 * time.Second}, + newWebAuthSigner(mustWebAuthKey(t, testToken), nil, &f.entropy), webAuthClaims{}) + if err != nil { + t.Fatal(err) + } + f.client.dialer = transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != f.relay.Addr().String() { + return nil, fmt.Errorf("fixture refused unowned relay %q/%q", network, address) + } + f.physicalDials.Add(1) + conn, err := (&net.Dialer{}).DialContext(ctx, "tcp4", address) + f.own(conn) + return conn, err + }) + targetDone := make(chan struct{}) + f.register(targetDone) + go func() { + defer close(targetDone) + for { + conn, err := f.target.AcceptTCP() + if err != nil { + return + } + if !f.own(conn) { + continue + } + done := make(chan struct{}) + f.register(done) + go func() { + defer close(done) + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(7 * time.Second)) + _, err := io.Copy(conn, conn) + if err == nil { + err = conn.CloseWrite() + } + select { + case f.results <- err: + case <-f.ctx.Done(): + } + }() + } + }() + relayDone := make(chan struct{}) + f.register(relayDone) + go func() { + defer close(relayDone) + for { + raw, err := f.relay.AcceptTCP() + if err != nil { + return + } + if !f.own(raw) { + continue + } + f.accepted.Add(1) + done := make(chan struct{}) + f.register(done) + go func() { + defer close(done) + defer raw.Close() + _ = raw.SetDeadline(time.Now().Add(7 * time.Second)) + conn := tls.Server(raw, serverTLS) + handshakeCtx, stop := context.WithTimeout(f.ctx, 2*time.Second) + err := conn.HandshakeContext(handshakeCtx) + stop() + if err != nil { + f.workerError(err) + return + } + state := conn.ConnectionState() + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != webH2ALPN { + f.workerError(errors.New("owned peer did not negotiate TLS13/h2")) + return + } + wire := &webH2GoAwayTLSConn{Conn: conn, acked: make(chan struct{})} + physical := newWebServerConnectionAuth(conn.Close) + ctx := context.WithValue(f.ctx, webTLSConnectionContextKey{}, conn) + ctx = context.WithValue(ctx, webServerConnectionAuthContextKey{}, physical) + ctx = context.WithValue(ctx, webH2GoAwayWireContextKey{}, wire) + observed := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, short := parseWebSessionBearer(r.Header.Get("Proxy-Authorization")) + request := webH2GoAwayRequest{physical: physical, short: short, wire: wire, done: make(chan struct{})} + f.mu.Lock() + if f.closed { + f.mu.Unlock() + return + } + f.requests = append(f.requests, request) + f.joins = append(f.joins, request.done) + f.mu.Unlock() + defer close(request.done) + handler.ServeHTTP(w, r) + }) + (&http2.Server{MaxConcurrentStreams: maxStreams}).ServeConn(wire, &http2.ServeConnOpts{Context: ctx, Handler: observed}) + }() + } + }() + return f +} + +func (f *webH2GoAwayFixture) own(conn net.Conn) bool { + if conn == nil { + return false + } + f.mu.Lock() + if f.closed { + f.mu.Unlock() + _ = conn.Close() + return false + } + f.conns = append(f.conns, conn) + f.mu.Unlock() + return true +} + +func (f *webH2GoAwayFixture) register(done <-chan struct{}) { + f.mu.Lock() + f.joins = append(f.joins, done) + f.mu.Unlock() +} + +func (f *webH2GoAwayFixture) workerError(err error) { + f.mu.Lock() + if !f.closed { + f.workerErrors = append(f.workerErrors, err) + } + f.mu.Unlock() +} + +func (f *webH2GoAwayFixture) require(done <-chan struct{}, what string) { + f.t.Helper() + select { + case <-done: + case <-f.ctx.Done(): + f.t.Fatalf("%s: %v", what, context.Cause(f.ctx)) + } +} + +func (f *webH2GoAwayFixture) open() net.Conn { + f.t.Helper() + ctx, cancel := context.WithTimeout(f.ctx, 2*time.Second) + defer cancel() + conn, err := f.client.DialContext(ctx, "tcp", f.target.Addr().String()) + f.own(conn) + if err != nil || conn == nil { + f.t.Fatalf("healthy public H2 DialContext: %v", err) + } + if err := conn.SetDeadline(time.Now().Add(6 * time.Second)); err != nil { + f.t.Fatal(err) + } + return conn +} + +func (f *webH2GoAwayFixture) ack(conn net.Conn, payload []byte) { + f.t.Helper() + if _, err := io.Copy(conn, bytes.NewReader(payload)); err != nil { + f.t.Fatal(err) + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil || !bytes.Equal(got, payload) { + f.t.Fatalf("held stream echo=%d bytes, error=%v", len(got), err) + } +} + +func (f *webH2GoAwayFixture) finish(conn net.Conn, payload []byte) { + f.t.Helper() + if _, err := io.Copy(conn, bytes.NewReader(payload)); err != nil { + f.t.Fatal(err) + } + if err := conn.(interface{ CloseWrite() error }).CloseWrite(); err != nil { + f.t.Fatal(err) + } + got, err := io.ReadAll(conn) + _ = conn.Close() + if err != nil || !bytes.Equal(got, payload) { + f.t.Fatalf("FIN echo=%d/%d, error=%v", len(got), len(payload), err) + } +} + +func (f *webH2GoAwayFixture) request(index int) webH2GoAwayRequest { + f.t.Helper() + f.mu.Lock() + defer f.mu.Unlock() + if len(f.requests) <= index { + f.t.Fatal("successful CONNECT had no actual handler observation") + } + return f.requests[index] +} + +func (f *webH2GoAwayFixture) requestsSnapshot() []webH2GoAwayRequest { + f.mu.Lock() + defer f.mu.Unlock() + return append([]webH2GoAwayRequest(nil), f.requests...) +} + +func (f *webH2GoAwayFixture) targetResult() { + f.t.Helper() + select { + case err := <-f.results: + if err != nil { + f.t.Fatalf("actual target EOF/echo: %v", err) + } + case <-f.ctx.Done(): + f.t.Fatal("actual target did not complete") + } +} + +func (f *webH2GoAwayFixture) releaseReset() { + f.resetReleaseOnce.Do(func() { close(f.resetRelease) }) +} + +func (f *webH2GoAwayFixture) cleanup() { + f.mu.Lock() + f.closed = true + conns := append([]net.Conn(nil), f.conns...) + f.mu.Unlock() + f.cancel() + f.releaseReset() + if f.relay != nil { + _ = f.relay.Close() + } + if f.target != nil { + _ = f.target.Close() + } + // Independently close the actual raw sockets before relying on Client.Close. + for _, conn := range conns { + _ = conn.Close() + } + closed := make(chan struct{}) + var closeErr error + go func() { + defer close(closed) + if f.client != nil { + closeErr = f.client.Close() + } + }() + limit := time.NewTimer(2 * time.Second) + defer limit.Stop() + select { + case <-closed: + if !destinationDialOwnershipOnlyClosed(closeErr) { + f.t.Errorf("client Close: %v", closeErr) + } + case <-limit.C: + f.t.Error("client Close did not join") + return + } + for i := 0; ; i++ { + f.mu.Lock() + if i == len(f.joins) { + f.mu.Unlock() + break + } + done := f.joins[i] + f.mu.Unlock() + select { + case <-done: + case <-limit.C: + f.t.Errorf("owned worker %d did not join", i) + return + } + } + f.mu.Lock() + errs := append([]error(nil), f.workerErrors...) + f.mu.Unlock() + for _, err := range errs { + f.t.Errorf("owned TLS worker: %v", err) + } + if f.admission != nil && len(f.admission.sem) != 0 { + f.t.Error("cleanup retained stream admission") + } + f.t.Log("owned real raw sockets, target/handler/TLS/ServeConn workers and client Close joined") +} + +// This test-only wrapper serializes injection with the server's own writes. +// It preserves all bytes and observes only the fixed PING_ACK payload while +// ServeConn remains the sole owner of reading and interpreting client frames. +type webH2GoAwayTLSConn struct { + *tls.Conn + writeMu sync.Mutex + readMu sync.Mutex + acked chan struct{} + ackOnce sync.Once + header [9]byte + headerN int + prefaceN int + remaining uint32 + pingACK bool + ping [8]byte + pingN int +} + +var webH2GoAwayPing = [8]byte{'g', 'o', 'a', 'w', 'a', 'y', 0, 255} + +func (c *webH2GoAwayTLSConn) Write(p []byte) (int, error) { + c.writeMu.Lock() + defer c.writeMu.Unlock() + return c.Conn.Write(p) +} + +func (c *webH2GoAwayTLSConn) goAwayAndPing() error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + framer := http2.NewFramer(c.Conn, nil) + if err := framer.WriteGoAway(1, http2.ErrCodeNo, nil); err != nil { + return err + } + return framer.WritePing(false, webH2GoAwayPing) +} + +func (c *webH2GoAwayTLSConn) protocolReset() error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + return http2.NewFramer(c.Conn, nil).WriteRSTStream(3, http2.ErrCodeProtocol) +} + +func (c *webH2GoAwayTLSConn) Read(p []byte) (int, error) { + c.readMu.Lock() + defer c.readMu.Unlock() + n, err := c.Conn.Read(p) + c.observe(p[:n]) + return n, err +} + +func (c *webH2GoAwayTLSConn) observe(p []byte) { + for len(p) > 0 { + if c.prefaceN < len(http2.ClientPreface) { + n := min(len(p), len(http2.ClientPreface)-c.prefaceN) + c.prefaceN += n + p = p[n:] + continue + } + if c.remaining != 0 { + n := min(len(p), int(c.remaining)) + if c.pingACK { + c.pingN += copy(c.ping[c.pingN:], p[:n]) + } + p = p[n:] + c.remaining -= uint32(n) + if c.remaining == 0 && c.pingACK && c.pingN == len(c.ping) && c.ping == webH2GoAwayPing { + c.ackOnce.Do(func() { close(c.acked) }) + } + continue + } + used := copy(c.header[c.headerN:], p) + c.headerN += used + // Consume exactly the copied header bytes, including fragmented headers. + p = p[used:] + if c.headerN != len(c.header) { + continue + } + c.remaining = uint32(c.header[0])<<16 | uint32(c.header[1])<<8 | uint32(c.header[2]) + stream := binary.BigEndian.Uint32(c.header[5:]) & 0x7fffffff + c.pingACK = c.header[3] == byte(http2.FramePing) && c.header[4]&byte(http2.FlagPingAck) != 0 && stream == 0 && c.remaining == 8 + c.pingN = 0 + c.headerN = 0 + } +}