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