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  			conn, err := tr.opts.ListenPacket("udp", ":0")
   146  			if err != nil {
   147  				return err
   148  			}
   149  			tr.endpoint, err = quic.NewEndpoint(conn, quicConfig)
   150  			if err != nil {
   151  				conn.Close()
   152  			}
   153  		} else {
   154  			tr.endpoint, err = quic.Listen("udp", ":0", quicConfig)
   155  		}
   156  	}
   157  	return err
   158  }
   159  
   160  // dial creates a new HTTP/3 client connection.
   161  func (tr *transport) dial(ctx context.Context, target string, tlsConfig *tls.Config, stateHook func()) (*clientConn, error) {
   162  	tr.incInFlightDials()
   163  	defer tr.decInFlightDials()
   164  
   165  	if err := tr.initEndpoint(); err != nil {
   166  		return nil, err
   167  	}
   168  	qconn, err := tr.endpoint.Dial(ctx, "udp", target, newQUICConfig(tr.opts.QUICConfig, tlsConfig))
   169  	if err != nil {
   170  		return nil, err
   171  	}
   172  	return tr.newClientConn(ctx, qconn, stateHook)
   173  }
   174  
   175  // CloseIdleConnections is called by net/http.Transport.CloseIdleConnections
   176  // after all existing idle connections are closed using http3.clientConn.Close.
   177  //
   178  // When the transport has no active connections anymore, calling this method
   179  // will make the transport clean up any shared resources that are no longer
   180  // required, such as its QUIC endpoint.
   181  func (tr *transport) CloseIdleConnections() {
   182  	tr.mu.Lock()
   183  	defer tr.mu.Unlock()
   184  	if tr.endpoint == nil || len(tr.activeConns) > 0 || tr.inFlightDials > 0 {
   185  		return
   186  	}
   187  	tr.endpoint.Close(canceledCtx)
   188  	tr.endpoint = nil
   189  }
   190  
   191  // A clientConn is a client HTTP/3 connection.
   192  //
   193  // Multiple goroutines may invoke methods on a clientConn simultaneously.
   194  type clientConn struct {
   195  	tr           *transport
   196  	unregistered chan struct{} // closed when clientConn is unregistered from tr.
   197  
   198  	qconn *quic.Conn
   199  	genericConn
   200  
   201  	enc qpackEncoder
   202  	dec qpackDecoder
   203  
   204  	// Guarded by genericConn.mu
   205  	reserved int
   206  	active   int
   207  	closed   bool
   208  
   209  	stateHook func()
   210  }
   211  
   212  func (tr *transport) registerConn(cc *clientConn) {
   213  	tr.mu.Lock()
   214  	defer tr.mu.Unlock()
   215  	tr.activeConns[cc] = struct{}{}
   216  }
   217  
   218  func (tr *transport) unregisterConn(cc *clientConn) {
   219  	tr.mu.Lock()
   220  	defer tr.mu.Unlock()
   221  	delete(tr.activeConns, cc)
   222  	close(cc.unregistered)
   223  }
   224  
   225  func (tr *transport) newClientConn(ctx context.Context, qconn *quic.Conn, stateHook func()) (*clientConn, error) {
   226  	cc := &clientConn{
   227  		tr:           tr,
   228  		unregistered: make(chan struct{}),
   229  		qconn:        qconn,
   230  		stateHook:    stateHook,
   231  	}
   232  	tr.registerConn(cc)
   233  	cc.enc.init()
   234  
   235  	// Create control stream and send SETTINGS frame.
   236  	controlStream, err := newConnStream(ctx, cc.qconn, streamTypeControl)
   237  	if err != nil {
   238  		tr.unregisterConn(cc)
   239  		return nil, fmt.Errorf("http3: cannot create control stream: %v", err)
   240  	}
   241  	controlStream.writeSettings()
   242  	controlStream.Flush()
   243  
   244  	go func() {
   245  		cc.acceptStreams(qconn, cc)
   246  		cc.mu.Lock()
   247  		cc.closed = true
   248  		cc.mu.Unlock()
   249  		cc.maybeCallStateHook()
   250  		tr.unregisterConn(cc)
   251  	}()
   252  	return cc, nil
   253  }
   254  
   255  func (cc *clientConn) Close() error {
   256  	err := cc.qconn.Close()
   257  	// Wait until cc is actually unregistered from the transport before
   258  	// returning. Otherwise, a race condition might occur: CloseIdleConnections
   259  	// might be called before cc gets a chance to be unregistered; if so,
   260  	// CloseIdleConnections will unexpectedly not close its QUIC endpoint,
   261  	// thinking that there is still an active cc.
   262  	<-cc.unregistered
   263  	return err
   264  }
   265  
   266  func (cc *clientConn) Err() error {
   267  	cc.mu.Lock()
   268  	defer cc.mu.Unlock()
   269  	if cc.closed {
   270  		return errors.New("connection closed")
   271  	}
   272  	return nil
   273  }
   274  
   275  func (cc *clientConn) Reserve() error {
   276  	cc.mu.Lock()
   277  	defer cc.mu.Unlock()
   278  	if cc.closed {
   279  		return errors.New("connection closed")
   280  	}
   281  	cc.reserved++
   282  	return nil
   283  }
   284  
   285  func (cc *clientConn) Release() {
   286  	cc.mu.Lock()
   287  	defer cc.mu.Unlock()
   288  	// This is consistent with RoundTrip: both Release and RoundTrip will
   289  	// consume a reservation iff one exists.
   290  	if cc.reserved > 0 {
   291  		cc.reserved--
   292  	}
   293  }
   294  
   295  func (cc *clientConn) Available() int {
   296  	cc.mu.Lock()
   297  	defer cc.mu.Unlock()
   298  	if cc.closed {
   299  		return 0
   300  	}
   301  	// The general recommendation for HTTP/3 is to reuse the same connection
   302  	// for multiple requests rather than creating new connections. As of now,
   303  	// we don't have a good understanding of when one might want to create
   304  	// multiple HTTP/3 connections to the same server.
   305  	// Therefore, for ClientConn API, let HTTP/3 connections have no limit.
   306  	// Starting a new RoundTrip when we are at the connection limit will just
   307  	// block until a new max stream limit is received.
   308  	return math.MaxInt
   309  }
   310  
   311  func (cc *clientConn) InFlight() int {
   312  	cc.mu.Lock()
   313  	defer cc.mu.Unlock()
   314  	if cc.closed {
   315  		return 0
   316  	}
   317  	return cc.reserved + cc.active
   318  }
   319  
   320  func (cc *clientConn) maybeCallStateHook() {
   321  	if cc.stateHook != nil {
   322  		cc.stateHook()
   323  	}
   324  }
   325  
   326  func (cc *clientConn) handleControlStream(st *stream) error {
   327  	// "A SETTINGS frame MUST be sent as the first frame of each control stream [...]"
   328  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-7.2.4-2
   329  	if err := st.readSettings(func(settingsType, settingsValue int64) error {
   330  		switch settingsType {
   331  		case settingsMaxFieldSectionSize:
   332  			_ = settingsValue // TODO
   333  		case settingsQPACKMaxTableCapacity:
   334  			_ = settingsValue // TODO
   335  		case settingsQPACKBlockedStreams:
   336  			_ = settingsValue // TODO
   337  		default:
   338  			// Unknown settings types are ignored.
   339  		}
   340  		return nil
   341  	}); err != nil {
   342  		return err
   343  	}
   344  
   345  	for {
   346  		ftype, err := st.readFrameHeader()
   347  		if err != nil {
   348  			return err
   349  		}
   350  		switch ftype {
   351  		case frameTypeCancelPush:
   352  			// "If a CANCEL_PUSH frame is received that references a push ID
   353  			// greater than currently allowed on the connection,
   354  			// this MUST be treated as a connection error of type H3_ID_ERROR."
   355  			// https://www.rfc-editor.org/rfc/rfc9114.html#section-7.2.3-7
   356  			return &connectionError{
   357  				code:    errH3IDError,
   358  				message: "CANCEL_PUSH received when no MAX_PUSH_ID has been sent",
   359  			}
   360  		case frameTypeGoaway:
   361  			// TODO: Wait for requests to complete before closing connection.
   362  			return errH3NoError
   363  		default:
   364  			// Unknown frames are ignored.
   365  			if err := st.discardUnknownFrame(ftype); err != nil {
   366  				return err
   367  			}
   368  		}
   369  	}
   370  }
   371  
   372  func (cc *clientConn) handleEncoderStream(*stream) error {
   373  	// TODO
   374  	return nil
   375  }
   376  
   377  func (cc *clientConn) handleDecoderStream(*stream) error {
   378  	// TODO
   379  	return nil
   380  }
   381  
   382  func (cc *clientConn) handlePushStream(*stream) error {
   383  	// "A client MUST treat receipt of a push stream as a connection error
   384  	// of type H3_ID_ERROR when no MAX_PUSH_ID frame has been sent [...]"
   385  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-4.6-3
   386  	return &connectionError{
   387  		code:    errH3IDError,
   388  		message: "push stream created when no MAX_PUSH_ID has been sent",
   389  	}
   390  }
   391  
   392  func (cc *clientConn) handleRequestStream(st *stream) error {
   393  	// "Clients MUST treat receipt of a server-initiated bidirectional
   394  	// stream as a connection error of type H3_STREAM_CREATION_ERROR [...]"
   395  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-6.1-3
   396  	return &connectionError{
   397  		code:    errH3StreamCreationError,
   398  		message: "server created bidirectional stream",
   399  	}
   400  }
   401  
   402  // abort closes the connection with an error.
   403  func (cc *clientConn) abort(err error) {
   404  	if e, ok := err.(*connectionError); ok {
   405  		cc.qconn.Abort(&quic.ApplicationError{
   406  			Code:   uint64(e.code),
   407  			Reason: e.message,
   408  		})
   409  	} else {
   410  		cc.qconn.Abort(err)
   411  	}
   412  }
   413
View as plain text