transport.go

     1  // Copyright 2025 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package http3
     6  
     7  import (
     8  	"context"
     9  	"crypto/tls"
    10  	"errors"
    11  	"fmt"
    12  	"math"
    13  	"net"
    14  	"net/http"
    15  	"net/url"
    16  	"sync"
    17  
    18  	"golang.org/x/net/quic"
    19  )
    20  
    21  // A transport is an HTTP/3 transport.
    22  //
    23  // It does not manage a pool of connections,
    24  // and therefore does not implement net/http.RoundTripper.
    25  //
    26  // TODO: Provide a way to register an HTTP/3 transport with a net/http.transport's
    27  // connection pool.
    28  type transport struct {
    29  	tr1  *http.Transport
    30  	opts TransportOpts
    31  
    32  	mu sync.Mutex // Guards fields below.
    33  	// endpoint is the QUIC endpoint used by connections created by the
    34  	// transport. If CloseIdleConnections is called when activeConns is empty,
    35  	// endpoint will be unset. If unset, endpoint will be initialized by any
    36  	// call to dial.
    37  	endpoint      *quic.Endpoint
    38  	activeConns   map[*clientConn]struct{}
    39  	inFlightDials int
    40  }
    41  
    42  // netHTTPTransport implements the net/http.dialClientConner interface,
    43  // allowing our HTTP/3 transport to integrate with net/http.
    44  type netHTTPTransport struct {
    45  	*transport
    46  }
    47  
    48  // Registered is called to record successful registration with a net/http Transport.
    49  func (t netHTTPTransport) Registered(tr1 *http.Transport) {
    50  	t.transport.tr1 = tr1
    51  }
    52  
    53  // RoundTrip is defined since Transport.RegisterProtocol takes in a
    54  // RoundTripper. However, this method will never be used as net/http's
    55  // dialClientConner interface does not have a RoundTrip method and will only
    56  // use DialClientConn to create a new RoundTripper.
    57  func (t netHTTPTransport) RoundTrip(*http.Request) (*http.Response, error) {
    58  	panic("netHTTPTransport.RoundTrip should never be called")
    59  }
    60  
    61  // DialClientConn creates a new client connection to address.
    62  //
    63  // If proxy is non-nil, the connection should use the provided proxy.
    64  // If HTTP/3 proxies are not supported, DialClientConn should return
    65  // an error wrapping [errors.ErrUnsupported].
    66  //
    67  // The RoundTripper returned by DialClientConn must also implement the
    68  // http.genericClientConn interface.
    69  //
    70  // The client connection should arrange to call internalStateHook
    71  // when the connection closes, when requests complete, and when the
    72  // connection concurrency limit changes.
    73  //
    74  // The client connection must call the internal state hook when
    75  // the connection state changes asynchronously, such as when a request completes.
    76  //
    77  // The internal state hook need not be called after synchronous changes
    78  // to the state: Close, Reserve, Release, and RoundTrip calls
    79  // which don't start a request do not need to call the hook.
    80  func (t netHTTPTransport) DialClientConn(ctx context.Context, addr string, _ *url.URL, tlsConfig *tls.Config, stateHook func()) (http.RoundTripper, error) {
    81  	return t.transport.dial(ctx, addr, tlsConfig, stateHook)
    82  }
    83  
    84  type TransportOpts struct {
    85  	// ListenQUIC determines how the transport will open a QUIC endpoint.
    86  	// By default, quic.Listen("udp", addr, config) is used.
    87  	// ListenQUIC might be called multiple times.
    88  	ListenQUIC func(addr string, config *quic.Config) (*quic.Endpoint, error)
    89  
    90  	// ListenPacket specifies the function for creating a UDP listener.
    91  	// If ListenPacket is nil, then the transport listens using net.ListenPacket.
    92  	//
    93  	// If ListenQUIC and ListenPacket are both set, ListenQUIC takes priority.
    94  	ListenPacket func(network, addr string) (net.PacketConn, error)
    95  
    96  	// QUICConfig is the QUIC configuration used by the transport.
    97  	// QUICConfig may be nil and should not be modified after calling
    98  	// RegisterTransport.
    99  	//
   100  	// The QUICConfig's TLSConfig is not used.
   101  	// Set the TLSConfig on the net/http Transport instead.
   102  	QUICConfig *quic.Config
   103  }
   104  
   105  // RegisterTransport configures a net/http HTTP/1 Transport to use HTTP/3.
   106  func RegisterTransport(tr *http.Transport, opts TransportOpts) error {
   107  	tr3 := &transport{
   108  		opts:        opts,
   109  		activeConns: make(map[*clientConn]struct{}),
   110  	}
   111  	// RegisterProtocol will set tr3.tr1.
   112  	tr.RegisterProtocol("http/3", netHTTPTransport{tr3})
   113  	if tr3.tr1 != tr {
   114  		return errors.New("http3: net/http does not support HTTP/3")
   115  	}
   116  	return nil
   117  }
   118  
   119  func (tr *transport) incInFlightDials() {
   120  	tr.mu.Lock()
   121  	defer tr.mu.Unlock()
   122  	tr.inFlightDials++
   123  }
   124  
   125  func (tr *transport) decInFlightDials() {
   126  	tr.mu.Lock()
   127  	defer tr.mu.Unlock()
   128  	tr.inFlightDials--
   129  }
   130  
   131  func (tr *transport) initEndpoint() (err error) {
   132  	tr.mu.Lock()
   133  	defer tr.mu.Unlock()
   134  	// This might cause rare issues on Darwin. Unlike Linux, Darwin kernel
   135  	// seems to have the following behaviors:
   136  	// - After closing a UDP socket, the port that was bound to the socket
   137  	//   might not be immediately usable again.
   138  	// - When doing IPv6 dual-stack binding (e.g., bind to ":0"), it will
   139  	//   happily bind the IPv6 port, even when the IPv4 port is unavailable.
   140  	//
   141  	// When both of these are combined, in practice, it is possible for the
   142  	// following to happen:
   143  	// 1. Transport binds ":0", creating a dual-stack IPv6 UDP socket.
   144  	//    Everything works as expected.
   145  	// 2. At some point, CloseIdleConnections is called and the socket is
   146  	//    closed.
   147  	// 3. Soon after, a new dial is started, and a new dual-stack IPv6 socket
   148  	//    is coincidentally assigned the same port as the previous socket.
   149  	// 4. If the IPv4 port is still unavailable, Darwin's permissive binding
   150  	//    behavior will cause us to have a socket that silently is unable to
   151  	//    receive packets on its IPv4 address.
   152  	// 5. If the dial target is an IPv4 address, transport will be able to send
   153  	//    packets to the target, but will be unable to receive its reply.
   154  	//
   155  	// TransportOpts.ListenQUIC can technically be configured to avoid
   156  	// dual-stack binding to avoid this issue, and high socket churn is
   157  	// probably uncommon for regular use cases. However, finding a workaround
   158  	// for this eventually would be ideal.
   159  	if tr.endpoint == nil {
   160  		quicConfig := newQUICConfig(tr.opts.QUICConfig, tr.tr1.TLSClientConfig)
   161  		if tr.opts.ListenQUIC != nil {
   162  			tr.endpoint, err = tr.opts.ListenQUIC(":0", quicConfig)
   163  		} else if tr.opts.ListenPacket != nil {
   164  			var conn net.PacketConn
   165  			conn, err = tr.opts.ListenPacket("udp", ":0")
   166  			if err != nil {
   167  				return err
   168  			}
   169  			tr.endpoint, err = quic.NewEndpoint(conn, quicConfig)
   170  			if err != nil {
   171  				conn.Close()
   172  			}
   173  		} else {
   174  			tr.endpoint, err = quic.Listen("udp", ":0", quicConfig)
   175  		}
   176  	}
   177  	return err
   178  }
   179  
   180  // dial creates a new HTTP/3 client connection.
   181  func (tr *transport) dial(ctx context.Context, target string, tlsConfig *tls.Config, stateHook func()) (*clientConn, error) {
   182  	tr.incInFlightDials()
   183  	defer tr.decInFlightDials()
   184  
   185  	if err := tr.initEndpoint(); err != nil {
   186  		return nil, err
   187  	}
   188  	qconn, err := tr.endpoint.Dial(ctx, "udp", target, newQUICConfig(tr.opts.QUICConfig, tlsConfig))
   189  	if err != nil {
   190  		return nil, err
   191  	}
   192  	return tr.newClientConn(ctx, qconn, stateHook)
   193  }
   194  
   195  // CloseIdleConnections is called by net/http.Transport.CloseIdleConnections
   196  // after all existing idle connections are closed using http3.clientConn.Close.
   197  //
   198  // When the transport has no active connections anymore, calling this method
   199  // will make the transport clean up any shared resources that are no longer
   200  // required, such as its QUIC endpoint.
   201  func (tr *transport) CloseIdleConnections() {
   202  	tr.mu.Lock()
   203  	defer tr.mu.Unlock()
   204  	if tr.endpoint == nil || len(tr.activeConns) > 0 || tr.inFlightDials > 0 {
   205  		return
   206  	}
   207  	tr.endpoint.Close(canceledCtx)
   208  	tr.endpoint = nil
   209  }
   210  
   211  // A clientConn is a client HTTP/3 connection.
   212  //
   213  // Multiple goroutines may invoke methods on a clientConn simultaneously.
   214  type clientConn struct {
   215  	tr           *transport
   216  	unregistered chan struct{} // closed when clientConn is unregistered from tr.
   217  
   218  	qconn *quic.Conn
   219  	genericConn
   220  
   221  	enc qpackEncoder
   222  	dec qpackDecoder
   223  
   224  	// Guarded by genericConn.mu
   225  	reserved int
   226  	active   int
   227  	closed   bool
   228  
   229  	stateHook func()
   230  }
   231  
   232  func (tr *transport) registerConn(cc *clientConn) {
   233  	tr.mu.Lock()
   234  	defer tr.mu.Unlock()
   235  	tr.activeConns[cc] = struct{}{}
   236  }
   237  
   238  func (tr *transport) unregisterConn(cc *clientConn) {
   239  	tr.mu.Lock()
   240  	defer tr.mu.Unlock()
   241  	delete(tr.activeConns, cc)
   242  	close(cc.unregistered)
   243  }
   244  
   245  func (tr *transport) maxResponseHeaderBytes() int64 {
   246  	if tr.tr1 != nil && tr.tr1.MaxResponseHeaderBytes > 0 {
   247  		return tr.tr1.MaxResponseHeaderBytes
   248  	}
   249  	return 10 << 20 // Matches HTTP/1 and HTTP/2 default of 10 MB.
   250  }
   251  
   252  func (tr *transport) newClientConn(ctx context.Context, qconn *quic.Conn, stateHook func()) (*clientConn, error) {
   253  	cc := &clientConn{
   254  		tr:           tr,
   255  		unregistered: make(chan struct{}),
   256  		qconn:        qconn,
   257  		stateHook:    stateHook,
   258  	}
   259  	tr.registerConn(cc)
   260  	cc.enc.init()
   261  
   262  	// Create control stream and send SETTINGS frame.
   263  	controlStream, err := newConnStream(ctx, cc.qconn, streamTypeControl)
   264  	if err != nil {
   265  		tr.unregisterConn(cc)
   266  		return nil, fmt.Errorf("http3: cannot create control stream: %v", err)
   267  	}
   268  	controlStream.writeSettings(settingsMaxFieldSectionSize, tr.maxResponseHeaderBytes())
   269  	controlStream.Flush()
   270  
   271  	go func() {
   272  		cc.acceptStreams(qconn, cc)
   273  		cc.mu.Lock()
   274  		cc.closed = true
   275  		cc.mu.Unlock()
   276  		cc.maybeCallStateHook()
   277  		tr.unregisterConn(cc)
   278  	}()
   279  	return cc, nil
   280  }
   281  
   282  func (cc *clientConn) Close() error {
   283  	err := cc.qconn.Close()
   284  	// Wait until cc is actually unregistered from the transport before
   285  	// returning. Otherwise, a race condition might occur: CloseIdleConnections
   286  	// might be called before cc gets a chance to be unregistered; if so,
   287  	// CloseIdleConnections will unexpectedly not close its QUIC endpoint,
   288  	// thinking that there is still an active cc.
   289  	<-cc.unregistered
   290  	return err
   291  }
   292  
   293  func (cc *clientConn) Proto() int {
   294  	return 3
   295  }
   296  
   297  func (cc *clientConn) IsAlt() bool {
   298  	return true
   299  }
   300  
   301  func (cc *clientConn) IsReused() bool {
   302  	return false
   303  }
   304  
   305  func (cc *clientConn) NetConn() net.Conn {
   306  	return nil
   307  }
   308  
   309  func (cc *clientConn) CloseWithError(err error) {
   310  	cc.Close()
   311  }
   312  
   313  func (cc *clientConn) Err() error {
   314  	cc.mu.Lock()
   315  	defer cc.mu.Unlock()
   316  	if cc.closed {
   317  		return errors.New("connection closed")
   318  	}
   319  	return nil
   320  }
   321  
   322  func (cc *clientConn) Reserve() error {
   323  	cc.mu.Lock()
   324  	defer cc.mu.Unlock()
   325  	if cc.closed {
   326  		return errors.New("connection closed")
   327  	}
   328  	cc.reserved++
   329  	return nil
   330  }
   331  
   332  func (cc *clientConn) Release() {
   333  	cc.mu.Lock()
   334  	defer cc.mu.Unlock()
   335  	// This is consistent with RoundTrip: both Release and RoundTrip will
   336  	// consume a reservation iff one exists.
   337  	if cc.reserved > 0 {
   338  		cc.reserved--
   339  	}
   340  }
   341  
   342  func (cc *clientConn) Available() int {
   343  	cc.mu.Lock()
   344  	defer cc.mu.Unlock()
   345  	if cc.closed {
   346  		return 0
   347  	}
   348  	// The general recommendation for HTTP/3 is to reuse the same connection
   349  	// for multiple requests rather than creating new connections. As of now,
   350  	// we don't have a good understanding of when one might want to create
   351  	// multiple HTTP/3 connections to the same server.
   352  	// Therefore, for ClientConn API, let HTTP/3 connections have no limit.
   353  	// Starting a new RoundTrip when we are at the connection limit will just
   354  	// block until a new max stream limit is received.
   355  	return math.MaxInt
   356  }
   357  
   358  func (cc *clientConn) InFlight() int {
   359  	cc.mu.Lock()
   360  	defer cc.mu.Unlock()
   361  	if cc.closed {
   362  		return 0
   363  	}
   364  	return cc.reserved + cc.active
   365  }
   366  
   367  func (cc *clientConn) maybeCallStateHook() {
   368  	if cc.stateHook != nil {
   369  		cc.stateHook()
   370  	}
   371  }
   372  
   373  func (cc *clientConn) handleControlStream(st *stream) error {
   374  	// "A SETTINGS frame MUST be sent as the first frame of each control stream [...]"
   375  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-7.2.4-2
   376  	if err := st.readSettings(func(settingsType, settingsValue int64) error {
   377  		switch settingsType {
   378  		case settingsMaxFieldSectionSize:
   379  			_ = settingsValue // TODO
   380  		case settingsQPACKMaxTableCapacity:
   381  			_ = settingsValue // TODO
   382  		case settingsQPACKBlockedStreams:
   383  			_ = settingsValue // TODO
   384  		default:
   385  			// Unknown settings types are ignored.
   386  		}
   387  		return nil
   388  	}); err != nil {
   389  		return err
   390  	}
   391  
   392  	for {
   393  		ftype, err := st.readFrameHeader()
   394  		if err != nil {
   395  			return err
   396  		}
   397  		switch ftype {
   398  		case frameTypeCancelPush:
   399  			// "If a CANCEL_PUSH frame is received that references a push ID
   400  			// greater than currently allowed on the connection,
   401  			// this MUST be treated as a connection error of type H3_ID_ERROR."
   402  			// https://www.rfc-editor.org/rfc/rfc9114.html#section-7.2.3-7
   403  			return &connectionError{
   404  				code:    errH3IDError,
   405  				message: "CANCEL_PUSH received when no MAX_PUSH_ID has been sent",
   406  			}
   407  		case frameTypeGoaway:
   408  			// TODO: Wait for requests to complete before closing connection.
   409  			return errH3NoError
   410  		default:
   411  			// Unknown frames are ignored.
   412  			if err := st.discardUnknownFrame(ftype); err != nil {
   413  				return err
   414  			}
   415  		}
   416  	}
   417  }
   418  
   419  func (cc *clientConn) handleEncoderStream(*stream) error {
   420  	// TODO
   421  	return nil
   422  }
   423  
   424  func (cc *clientConn) handleDecoderStream(*stream) error {
   425  	// TODO
   426  	return nil
   427  }
   428  
   429  func (cc *clientConn) handlePushStream(*stream) error {
   430  	// "A client MUST treat receipt of a push stream as a connection error
   431  	// of type H3_ID_ERROR when no MAX_PUSH_ID frame has been sent [...]"
   432  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-4.6-3
   433  	return &connectionError{
   434  		code:    errH3IDError,
   435  		message: "push stream created when no MAX_PUSH_ID has been sent",
   436  	}
   437  }
   438  
   439  func (cc *clientConn) handleRequestStream(st *stream) error {
   440  	// "Clients MUST treat receipt of a server-initiated bidirectional
   441  	// stream as a connection error of type H3_STREAM_CREATION_ERROR [...]"
   442  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-6.1-3
   443  	return &connectionError{
   444  		code:    errH3StreamCreationError,
   445  		message: "server created bidirectional stream",
   446  	}
   447  }
   448  
   449  // abort closes the connection with an error.
   450  func (cc *clientConn) abort(err error) {
   451  	if e, ok := err.(*connectionError); ok {
   452  		cc.qconn.Abort(&quic.ConnectionCloseError{
   453  			Code:   uint64(e.code),
   454  			Reason: e.message,
   455  		})
   456  	} else {
   457  		cc.qconn.Abort(err)
   458  	}
   459  }
   460
View as plain text