From f38bce5b3e0d794ed3c72e75359f611c94704564 Mon Sep 17 00:00:00 2001 From: cppla Date: Fri, 9 Oct 2026 10:19:30 +0800 Subject: [PATCH] fix: keep H2 failed response cleanup out of dial budgets --- docs/WEB_COVER.md | 9 + internal/tunnel/web_h2_client.go | 56 ++- .../web_h2_failed_response_cleanup_test.go | 426 ++++++++++++++++++ internal/tunnel/web_h2_lifecycle_test.go | 2 + 4 files changed, 475 insertions(+), 18 deletions(-) create mode 100644 internal/tunnel/web_h2_failed_response_cleanup_test.go diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index 696ac2d..c5e89a1 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -144,6 +144,15 @@ streams on the affected connection, while ordinary stream cancellation must preserve healthy siblings. See [timeout semantics and configuration](DEPLOYMENT.md) for partial-progress and cleanup boundaries. This setting is not in v1.0.1. +Failed H2 opens cancel their request immediately. An authenticated error or +canceled handoff can leave unread response data whose cleanup needs that same +shared write lock. Cleanup is client-owned and joined by client shutdown; it +does not extend the failed caller's wait or retire usable sibling streams. +The pending cleanup retains its opening reservation and continuation ticket +slot until the response closes, so the per-connection authentication window +also limits detached continuation cleanup. An invalid authentication proof +instead retires the physical connection before closing its response body. + There is no `autocar/2` ALPN or AutoCAR binary stream header on these paths. The web ALPNs are `h2`, `h3`, and `http/1.1`. Native and web transports remain separate modes and are not wire-compatible. diff --git a/internal/tunnel/web_h2_client.go b/internal/tunnel/web_h2_client.go index a6956e6..67a871f 100644 --- a/internal/tunnel/web_h2_client.go +++ b/internal/tunnel/web_h2_client.go @@ -73,8 +73,8 @@ type WebH2Client struct { sessions map[*webH2ClientSession]struct{} dial *webH2DialAttempt selection *webH2SelectionCohort - // Additions are made under mu before the closed gate. Close joins both - // unpublished physical setup and cleanup detached from sessions. + // Additions are made under mu before the closed gate. Close joins physical + // setup, opening reservations and cleanup detached from failed responses. workers sync.WaitGroup closeDone chan struct{} closeErr error @@ -103,7 +103,8 @@ type webH2ClientSession struct { authReady chan struct{} auth *webSessionClientAuth // opening protects a selected session while an authentication ticket waits - // for replay-window capacity, before RoundTrip owns an HTTP/2 stream. + // for replay-window capacity and until an unsuccessful response is closed. + // Each reservation also owns a worker so client Close joins its cleanup. opening int // active counts returned net.Conns until full close, including deadline // cancellation. Half-closes keep the opposite direction's ownership. @@ -308,17 +309,38 @@ func (c *WebH2Client) dialStreamOnce( return nil, err, false } session := reservation.session - defer c.releaseSessionReservation(session) var ( bearer string authClaims webAuthClaims exchange webSessionExchange ) + finish := func() { + reservation.auth.complete(exchange) + c.releaseSessionReservation(session) + } + cleanupTransferred := false + defer func() { + if !cleanupTransferred { + finish() + } + }() + // Closing an unread response may wait on the shared HTTP/2 write lock + // even after request cancellation. Transfer the already-registered opening + // reservation, rather than delaying this caller or closing usable siblings. + // Keep its short-ticket slot until cleanup completes: the authentication + // replay window then also bounds pending continuation cleanup per session. + failResponse := func(body io.ReadCloser) { + fail() + cleanupTransferred = true + go func() { + defer finish() + _ = body.Close() + }() + } if reservation.bootstrap { bearer, authClaims, err = c.auth.authorization(binding, c.claims) } else { bearer, exchange, err = reservation.auth.authorization(requestCtx, binding) - defer reservation.auth.complete(exchange) } if err != nil { fail() @@ -375,9 +397,9 @@ func (c *WebH2Client) dialStreamOnce( response.StatusCode, ) if !ok || !c.completeSessionBootstrap(session, sessionAuth) { - _ = response.Body.Close() fail() c.failSessionAuthentication(session) + _ = response.Body.Close() return nil, errors.New("tunnel: web-cover HTTP/2 server authentication failed"), false } } else if !reservation.auth.verifyResponseProof( @@ -386,44 +408,39 @@ func (c *WebH2Client) dialStreamOnce( exchange, response.StatusCode, ) { - _ = response.Body.Close() fail() c.failSessionAuthentication(session) + _ = response.Body.Close() return nil, errors.New("tunnel: web-cover HTTP/2 server authentication failed"), false } if response.StatusCode != http.StatusOK { - _ = response.Body.Close() - fail() + failResponse(response.Body) 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 // dial instead of returning an already-canceled stream. if !stopOpenTimeout() { - _ = response.Body.Close() - fail() + failResponse(response.Body) return nil, context.DeadlineExceeded, false } stopDial() if err := contextError(ctx); err != nil { - _ = response.Body.Close() - fail() + failResponse(response.Body) return nil, err, false } c.mu.Lock() if c.closed { c.mu.Unlock() - _ = response.Body.Close() - fail() + failResponse(response.Body) 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. if err := contextError(ctx); err != nil { c.mu.Unlock() - _ = response.Body.Close() - fail() + failResponse(response.Body) return nil, err, false } c.selected = true @@ -546,6 +563,7 @@ func (c *WebH2Client) reserveSession(ctx context.Context) (webH2SessionReservati // not when a background worker publishes a cold connection. session.authState = webH2ClientAuthBootstrapping session.opening++ + c.workers.Add(1) c.mu.Unlock() <-c.dialGate return webH2SessionReservation{session: session, bootstrap: true}, nil @@ -554,6 +572,7 @@ func (c *WebH2Client) reserveSession(ctx context.Context) (webH2SessionReservati case webH2ClientAuthReady: if canTake { session.opening++ + c.workers.Add(1) auth := session.auth c.mu.Unlock() <-c.dialGate @@ -651,6 +670,7 @@ func (c *WebH2Client) runSessionDial(attempt *webH2DialAttempt) { } func (c *WebH2Client) releaseSessionReservation(session *webH2ClientSession) { + defer c.workers.Done() c.mu.Lock() session.opening-- retired := c.cleanupIdleSessionsLocked() @@ -896,7 +916,7 @@ func closeWebH2Session(session *webH2ClientSession) error { } // Close prevents future dials, cancels active streams, and closes every pooled -// HTTP/2 connection. It joins owned setup and detached cleanup; concurrent +// HTTP/2 connection. It joins owned setup, pending opens and detached cleanup; concurrent // callers observe the same completed closure and error. func (c *WebH2Client) Close() error { c.mu.Lock() diff --git a/internal/tunnel/web_h2_failed_response_cleanup_test.go b/internal/tunnel/web_h2_failed_response_cleanup_test.go new file mode 100644 index 0000000..a7d772d --- /dev/null +++ b/internal/tunnel/web_h2_failed_response_cleanup_test.go @@ -0,0 +1,426 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" + "golang.org/x/net/http2" + "golang.org/x/net/http2/hpack" +) + +// This test gates the real bootstrap commit, not a synthetic response Body. +// The peer has sent a valid status-bound proof and unread DATA before its next +// SETTINGS makes the actual HTTP/2 writer block in a TLS write over net.Pipe. +// Releasing the commit lock must return the HTTP error without waiting for +// that unrelated write, without retiring the healthy physical connection, +// and without losing ownership of the still-pending response cleanup. +func TestWebH2FailedResponseCleanupDoesNotWaitForWriter(t *testing.T) { + for _, profile := range []FingerprintProfile{FingerprintNative, FingerprintChrome133, FingerprintChrome155} { + t.Run(string(profile), func(t *testing.T) { + for _, mode := range []string{"authenticated_error", "canceled_handoff", "client_close", "bootstrap_lost_selection"} { + t.Run(mode, func(t *testing.T) { testWebH2FailedResponseCleanup(t, profile, mode) }) + } + }) + } +} + +func testWebH2FailedResponseCleanup(t *testing.T, profile FingerprintProfile, mode string) { + status := http.StatusBadGateway + if mode == "canceled_handoff" { + status = http.StatusOK + } + f := newWebH2FailedResponseFixture(t, profile, status) + caller, cancel := context.WithCancelCause(f.ctx) + t.Cleanup(func() { cancel(context.Canceled) }) + var gate *webH2TCPHandoffContext + var dialCtx context.Context = caller + if mode == "canceled_handoff" { + release := make(chan struct{}) + gate = &webH2TCPHandoffContext{Context: caller, client: f.client, + entered: make(chan webH2TCPHandoffSnapshot, 1), release: release, fixture: f.ctx} + dialCtx = gate + f.releaseHandoff = sync.OnceFunc(func() { close(release) }) + t.Cleanup(f.releaseHandoff) + } + result, joined := f.dial(dialCtx) + f.wait(f.requestReceived, "first authenticated CONNECT headers") + var session *webH2ClientSession + var state tls.ConnectionState + var unlock func() + if gate == nil { + f.client.mu.Lock() + unlock = sync.OnceFunc(f.client.mu.Unlock) + t.Cleanup(unlock) + session = f.client.current + if session == nil || session.authState != webH2ClientAuthBootstrapping { + unlock() + t.Fatal("first CONNECT did not own a real bootstrapping session") + } + state = session.conn.ConnectionState() + close(f.respond) + } else { + close(f.respond) + select { + case snapshot := <-gate.entered: + session, state = snapshot.session, snapshot.state + if snapshot.opening != 1 || snapshot.active != 0 || snapshot.selected { + t.Fatal("handoff gate did not observe an opening, unreturned CONNECT") + } + case <-f.ctx.Done(): + t.Fatal("actual post-proof handoff checkpoint exceeded fixture budget") + } + unlock = f.releaseHandoff + } + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != webH2ALPN || len(state.VerifiedChains) == 0 { + unlock() + t.Fatal("fixture did not negotiate verified TLS 1.3 and H2") + } + f.wait(f.wire.started, "SETTINGS ACK entering the blocked TLS writer") + cause := errors.New("caller canceled after verified CONNECT proof") + if gate != nil { + cancel(cause) + f.wait(caller.Done(), "actual caller cancellation") + } + if mode == "bootstrap_lost_selection" { + // The proof remains valid. Only the selected session is invalidated + // under the held application mutex, so completeSessionBootstrap must + // reject its late result. This is not an invalid-MAC test. + f.client.current = nil + } + unlock() + + // This is an assertion deadline after a witnessed blocking event, + // not a sleep used to manufacture the scheduling order. The wire's + // production write timeout is deliberately much longer (30s). + select { + case <-joined: + case <-time.After(time.Second): + t.Error("failed-response DialContext waited for the stalled SETTINGS ACK writer") + f.release() + f.wait(joined, "failed DialContext after explicit fixture release") + } + got := <-result + if got.conn != nil { + _ = got.conn.Close() + t.Fatal("failed or canceled CONNECT returned a tunnel") + } + if mode == "bootstrap_lost_selection" { + if got.err == nil || !strings.Contains(got.err.Error(), "server authentication failed") { + t.Fatalf("late bootstrap result = %v, want authentication failure", got.err) + } + f.client.mu.Lock() + opening, active, registered := session.opening, session.active, len(f.client.sessions) + failed := session.authState == webH2ClientAuthFailed + f.client.mu.Unlock() + if !failed || opening != 0 || active != 0 || registered != 0 || f.wire.closes.Load() == 0 { + t.Errorf("rejected bootstrap failed/opening/active/registered/raw-closes=%t/%d/%d/%d/%d", failed, opening, active, registered, f.wire.closes.Load()) + } + return + } + var connectErr *WebConnectError + if gate != nil { + if !errors.Is(got.err, cause) || !gate.gated.Load() || gate.expired.Load() { + t.Fatalf("canceled handoff error=%v, gated/expired=%t/%t", got.err, gate.gated.Load(), gate.expired.Load()) + } + } else if !errors.As(got.err, &connectErr) || connectErr.StatusCode != http.StatusBadGateway { + t.Fatalf("authenticated HTTP error = %v, want status-bound 502", got.err) + } + if !f.released { + f.client.mu.Lock() + current, opening, active := f.client.current == session, session.opening, session.active + f.client.mu.Unlock() + if !current || opening != 1 || active != 0 || f.wire.closes.Load() != 0 { + t.Errorf("pending cleanup current/opening/active/raw-closes=%t/%d/%d/%d, want true/1/0/0", current, opening, active, f.wire.closes.Load()) + } + } + if mode == "client_close" { + closed := make(chan struct{}) + go func() { _ = f.client.Close(); close(closed) }() + f.joins = append(f.joins, closed) + f.wait(closed, "Client.Close joins pending response cleanup while ACK is blocked") + f.client.mu.Lock() + opening, active, registered := session.opening, session.active, len(f.client.sessions) + f.client.mu.Unlock() + if opening != 0 || active != 0 || registered != 0 || f.wire.closes.Load() == 0 { + t.Errorf("joined cleanup opening/active/registered/raw-closes=%d/%d/%d/%d, want 0/0/0/positive", opening, active, registered, f.wire.closes.Load()) + } + return + } + f.release() + f.wait(f.firstReset, "failed stream reset after writer release") + + // A second public DialContext must use the same physical TLS/H2 + // connection and its real short-ticket continuation authentication. + healthy, err := f.client.DialContext(f.ctx, "tcp", "owned.invalid:443") + if err != nil { + t.Fatalf("healthy continuation after failed response: %v", err) + } + defer healthy.Close() + if err := healthy.SetDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + if _, err := io.WriteString(healthy, "healthy sibling"); err != nil { + t.Fatal(err) + } + body := make([]byte, len("healthy sibling")) + if _, err := io.ReadFull(healthy, body); err != nil || string(body) != "healthy sibling" { + t.Fatalf("healthy continuation echo failed: %v", err) + } + f.client.mu.Lock() + same := f.client.current == session + f.client.mu.Unlock() + if !same || f.wire.closes.Load() != 0 { + t.Fatal("failed response cleanup replaced or closed the healthy physical session") + } +} + +type webH2FailedResponseResult struct { + conn net.Conn + err error +} + +type webH2FailedResponseFixture struct { + t *testing.T + ctx context.Context + client *WebH2Client + wire *webH2StalledWriter + requestReceived chan struct{} + respond chan struct{} + resume chan struct{} + firstReset chan struct{} + firstStatus int + releaseHandoff func() + released bool // owned only by the test goroutine + joins []<-chan struct{} +} + +func newWebH2FailedResponseFixture(t *testing.T, profile FingerprintProfile, status int) *webH2FailedResponseFixture { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + raw, peer := net.Pipe() + wire := &webH2StalledWriter{Conn: raw, started: make(chan struct{})} + serverTLS, clientTLS := testTLSConfigs(t) + serverTLS.NextProtos = []string{webH2ALPN} + client, err := NewWebH2Client(WebH2ClientConfig{ServerAddress: "127.0.0.1:443", Token: webTestToken, + TLSConfig: clientTLS, FingerprintProfile: profile, HandshakeTimeout: 5 * time.Second}) + if err != nil { + cancel() + _ = raw.Close() + _ = peer.Close() + t.Fatal(err) + } + var dialOnce sync.Once + client.dialer = transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + var result net.Conn + dialOnce.Do(func() { result = wire }) + if result == nil { + return nil, errors.New("fixture refuses a replacement physical dial") + } + return result, nil + }) + f := &webH2FailedResponseFixture{t: t, ctx: ctx, client: client, wire: wire, + requestReceived: make(chan struct{}), respond: make(chan struct{}), resume: make(chan struct{}), firstReset: make(chan struct{}), firstStatus: status} + serverJoined := make(chan struct{}) + f.joins = append(f.joins, serverJoined) + serverResult := make(chan error, 1) + t.Cleanup(func() { + cancel() + _ = raw.Close() + _ = peer.Close() + closed := make(chan struct{}) + go func() { _ = client.Close(); close(closed) }() + for _, joined := range append(f.joins, closed) { + select { + case <-joined: + case <-time.After(2 * time.Second): + t.Error("failed-response fixture did not join an owned worker") + } + } + select { + case err := <-serverResult: + if err != nil && !errors.Is(err, net.ErrClosed) && !errors.Is(err, io.ErrClosedPipe) && !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) { + t.Errorf("failed-response peer: %v", err) + } + default: + t.Error("failed-response peer result absent after cleanup join") + } + }) + go func() { + defer close(serverJoined) + serverResult <- f.serve(peer, serverTLS) + }() + return f +} + +func (f *webH2FailedResponseFixture) wait(done <-chan struct{}, name string) { + f.t.Helper() + select { + case <-done: + case <-f.ctx.Done(): + f.t.Fatalf("%s exceeded independent fixture budget", name) + } +} + +func (f *webH2FailedResponseFixture) release() { + if !f.released { + f.released = true + close(f.resume) + } +} + +func (f *webH2FailedResponseFixture) dial(ctx context.Context) (<-chan webH2FailedResponseResult, <-chan struct{}) { + result, joined := make(chan webH2FailedResponseResult, 1), make(chan struct{}) + f.joins = append(f.joins, joined) + go func() { + defer close(joined) + conn, err := f.client.DialContext(ctx, "tcp", "owned.invalid:443") + result <- webH2FailedResponseResult{conn, err} + }() + return result, joined +} + +func (f *webH2FailedResponseFixture) serve(peer net.Conn, serverTLS *tls.Config) error { + deadline, _ := f.ctx.Deadline() + _ = peer.SetDeadline(deadline) + conn := tls.Server(peer, serverTLS) + if err := conn.HandshakeContext(f.ctx); err != nil { + return err + } + preface := make([]byte, len(http2.ClientPreface)) + if _, err := io.ReadFull(conn, preface); err != nil { + return err + } + if string(preface) != http2.ClientPreface { + return errors.New("unexpected HTTP/2 preface") + } + fr := http2.NewFramer(conn, conn) + fr.ReadMetaHeaders = hpack.NewDecoder(4096, nil) + for range 2 { + if _, err := fr.ReadFrame(); err != nil { + return err + } + } + if err := fr.WriteSettings(); err != nil { + return err + } + key, err := deriveWebAuthKey(webTestToken) + if err != nil { + return err + } + verifier, err := newWebAuthVerifier(key, nil, 16) + if err != nil { + return err + } + var auth *webSessionServerAuth + var firstStream uint32 + startupAck := false + var encoded bytes.Buffer + encoder := hpack.NewEncoder(&encoded) + for { + frame, err := fr.ReadFrame() + if err != nil { + return err + } + switch frame := frame.(type) { + case *http2.MetaHeadersFrame: + // HEADERS can arrive before the initial SETTINGS ACK. Drain that + // ACK before pausing peer reads, so only the deliberately later + // SETTINGS can hold the client's writer during the response. + for !startupAck { + pending, err := fr.ReadFrame() + if err != nil { + return err + } + if settings, ok := pending.(*http2.SettingsFrame); ok && settings.IsAck() { + startupAck = true + } + } + binding := webAuthBinding{transport: webAuthTransportH2, method: frame.PseudoValue("method"), authority: frame.PseudoValue("authority")} + var bearer string + for _, field := range frame.RegularFields() { + if field.Name == "proxy-authorization" { + bearer = field.Value + } + } + status, proof := http.StatusOK, "" + if auth == nil { + claims, valid := verifier.verifyBearer(bearer, binding) + if !valid { + return errors.New("fixture received invalid full authentication") + } + status = f.firstStatus + proof, auth, err = issueWebSessionBootstrap(key, binding, claims, status, nil) + if err != nil { + return err + } + firstStream = frame.StreamID + close(f.requestReceived) + select { + case <-f.respond: + case <-f.ctx.Done(): + return f.ctx.Err() + } + } else { + exchange, valid := auth.verifyAuthorization(bearer, binding) + if !valid { + return errors.New("fixture received invalid continuation authentication") + } + proof = auth.responseProof(binding, exchange, status) + } + encoded.Reset() + for _, field := range []hpack.HeaderField{{Name: ":status", Value: strconv.Itoa(status)}, {Name: "proxy-authentication-info", Value: proof}} { + if err := encoder.WriteField(field); err != nil { + return err + } + } + if err := fr.WriteHeaders(http2.HeadersFrameParam{StreamID: frame.StreamID, BlockFragment: encoded.Bytes(), EndHeaders: true}); err != nil { + return err + } + if frame.StreamID == firstStream { + if err := fr.WriteData(firstStream, false, []byte("unread authenticated error")); err != nil { + return err + } + f.wire.signal.Store(true) + if err := fr.WriteSettings(); err != nil { + return err + } + select { + case <-f.resume: + case <-f.ctx.Done(): + return f.ctx.Err() + } + } + case *http2.SettingsFrame: + if !frame.IsAck() { + return fmt.Errorf("unexpected additional non-ACK client SETTINGS") + } + startupAck = true + case *http2.RSTStreamFrame: + if frame.StreamID == firstStream { + select { + case <-f.firstReset: + default: + close(f.firstReset) + } + } + case *http2.DataFrame: + if frame.StreamID != firstStream && len(frame.Data()) != 0 { + if err := fr.WriteData(frame.StreamID, frame.StreamEnded(), append([]byte(nil), frame.Data()...)); err != nil { + return err + } + } + } + } +} diff --git a/internal/tunnel/web_h2_lifecycle_test.go b/internal/tunnel/web_h2_lifecycle_test.go index 4d9eefc..bbc07b7 100644 --- a/internal/tunnel/web_h2_lifecycle_test.go +++ b/internal/tunnel/web_h2_lifecycle_test.go @@ -28,6 +28,7 @@ func TestWebH2RetirementAndCloseDoNotWaitForWireWrite(t *testing.T) { ctx: ctx, cancel: cancel, dialGate: make(chan struct{}, 1), sessions: map[*webH2ClientSession]struct{}{session: {}}, } + client.workers.Add(session.opening) // Leave the dial gate occupied so a second dial can only finish // through its deadline, without accessing the network. client.dialGate <- struct{}{} @@ -76,6 +77,7 @@ func TestWebH2RetirementKeepsOpeningReservation(t *testing.T) { wire, h2 := newWebH2StalledWriter(t) session := &webH2ClientSession{raw: wire, h2: h2, opening: 2} client := &WebH2Client{sessions: map[*webH2ClientSession]struct{}{session: {}}} + client.workers.Add(session.opening) client.releaseSessionReservation(session) if wire.closes.Load() != 0 || len(client.sessions) != 1 || session.opening != 1 { t.Fatal("retirement closed a session with another opening stream")