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) newClientConn(ctx context.Context, qconn *quic.Conn, stateHook func()) (*clientConn, error) {
   227  	cc := &clientConn{
   228  		tr:           tr,
   229  		unregistered: make(chan struct{}),
   230  		qconn:        qconn,
   231  		stateHook:    stateHook,
   232  	}
   233  	tr.registerConn(cc)
   234  	cc.enc.init()
   235  
   236  	// Create control stream and send SETTINGS frame.
   237  	controlStream, err := newConnStream(ctx, cc.qconn, streamTypeControl)
   238  	if err != nil {
   239  		tr.unregisterConn(cc)
   240  		return nil, fmt.Errorf("http3: cannot create control stream: %v", err)
   241  	}
   242  	controlStream.writeSettings()
   243  	controlStream.Flush()
   244  
   245  	go func() {
   246  		cc.acceptStreams(qconn, cc)
   247  		cc.mu.Lock()
   248  		cc.closed = true
   249  		cc.mu.Unlock()
   250  		cc.maybeCallStateHook()
   251  		tr.unregisterConn(cc)
   252  	}()
   253  	return cc, nil
   254  }
   255  
   256  func (cc *clientConn) Close() error {
   257  	err := cc.qconn.Close()
   258  	// Wait until cc is actually unregistered from the transport before
   259  	// returning. Otherwise, a race condition might occur: CloseIdleConnections
   260  	// might be called before cc gets a chance to be unregistered; if so,
   261  	// CloseIdleConnections will unexpectedly not close its QUIC endpoint,
   262  	// thinking that there is still an active cc.
   263  	<-cc.unregistered
   264  	return err
   265  }
   266  
   267  func (cc *clientConn) Err() error {
   268  	cc.mu.Lock()
   269  	defer cc.mu.Unlock()
   270  	if cc.closed {
   271  		return errors.New("connection closed")
   272  	}
   273  	return nil
   274  }
   275  
   276  func (cc *clientConn) Reserve() error {
   277  	cc.mu.Lock()
   278  	defer cc.mu.Unlock()
   279  	if cc.closed {
   280  		return errors.New("connection closed")
   281  	}
   282  	cc.reserved++
   283  	return nil
   284  }
   285  
   286  func (cc *clientConn) Release() {
   287  	cc.mu.Lock()
   288  	defer cc.mu.Unlock()
   289  	// This is consistent with RoundTrip: both Release and RoundTrip will
   290  	// consume a reservation iff one exists.
   291  	if cc.reserved > 0 {
   292  		cc.reserved--
   293  	}
   294  }
   295  
   296  func (cc *clientConn) Available() int {
   297  	cc.mu.Lock()
   298  	defer cc.mu.Unlock()
   299  	if cc.closed {
   300  		return 0
   301  	}
   302  	// The general recommendation for HTTP/3 is to reuse the same connection
   303  	// for multiple requests rather than creating new connections. As of now,
   304  	// we don't have a good understanding of when one might want to create
   305  	// multiple HTTP/3 connections to the same server.
   306  	// Therefore, for ClientConn API, let HTTP/3 connections have no limit.
   307  	// Starting a new RoundTrip when we are at the connection limit will just
   308  	// block until a new max stream limit is received.
   309  	return math.MaxInt
   310  }
   311  
   312  func (cc *clientConn) InFlight() int {
   313  	cc.mu.Lock()
   314  	defer cc.mu.Unlock()
   315  	if cc.closed {
   316  		return 0
   317  	}
   318  	return cc.reserved + cc.active
   319  }
   320  
   321  func (cc *clientConn) maybeCallStateHook() {
   322  	if cc.stateHook != nil {
   323  		cc.stateHook()
   324  	}
   325  }
   326  
   327  func (cc *clientConn) handleControlStream(st *stream) error {
   328  	// "A SETTINGS frame MUST be sent as the first frame of each control stream [...]"
   329  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-7.2.4-2
   330  	if err := st.readSettings(func(settingsType, settingsValue int64) error {
   331  		switch settingsType {
   332  		case settingsMaxFieldSectionSize:
   333  			_ = settingsValue // TODO
   334  		case settingsQPACKMaxTableCapacity:
   335  			_ = settingsValue // TODO
   336  		case settingsQPACKBlockedStreams:
   337  			_ = settingsValue // TODO
   338  		default:
   339  			// Unknown settings types are ignored.
   340  		}
   341  		return nil
   342  	}); err != nil {
   343  		return err
   344  	}
   345  
   346  	for {
   347  		ftype, err := st.readFrameHeader()
   348  		if err != nil {
   349  			return err
   350  		}
   351  		switch ftype {
   352  		case frameTypeCancelPush:
   353  			// "If a CANCEL_PUSH frame is received that references a push ID
   354  			// greater than currently allowed on the connection,
   355  			// this MUST be treated as a connection error of type H3_ID_ERROR."
   356  			// https://www.rfc-editor.org/rfc/rfc9114.html#section-7.2.3-7
   357  			return &connectionError{
   358  				code:    errH3IDError,
   359  				message: "CANCEL_PUSH received when no MAX_PUSH_ID has been sent",
   360  			}
   361  		case frameTypeGoaway:
   362  			// TODO: Wait for requests to complete before closing connection.
   363  			return errH3NoError
   364  		default:
   365  			// Unknown frames are ignored.
   366  			if err := st.discardUnknownFrame(ftype); err != nil {
   367  				return err
   368  			}
   369  		}
   370  	}
   371  }
   372  
   373  func (cc *clientConn) handleEncoderStream(*stream) error {
   374  	// TODO
   375  	return nil
   376  }
   377  
   378  func (cc *clientConn) handleDecoderStream(*stream) error {
   379  	// TODO
   380  	return nil
   381  }
   382  
   383  func (cc *clientConn) handlePushStream(*stream) error {
   384  	// "A client MUST treat receipt of a push stream as a connection error
   385  	// of type H3_ID_ERROR when no MAX_PUSH_ID frame has been sent [...]"
   386  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-4.6-3
   387  	return &connectionError{
   388  		code:    errH3IDError,
   389  		message: "push stream created when no MAX_PUSH_ID has been sent",
   390  	}
   391  }
   392  
   393  func (cc *clientConn) handleRequestStream(st *stream) error {
   394  	// "Clients MUST treat receipt of a server-initiated bidirectional
   395  	// stream as a connection error of type H3_STREAM_CREATION_ERROR [...]"
   396  	// https://www.rfc-editor.org/rfc/rfc9114.html#section-6.1-3
   397  	return &connectionError{
   398  		code:    errH3StreamCreationError,
   399  		message: "server created bidirectional stream",
   400  	}
   401  }
   402  
   403  // abort closes the connection with an error.
   404  func (cc *clientConn) abort(err error) {
   405  	if e, ok := err.(*connectionError); ok {
   406  		cc.qconn.Abort(&quic.ConnectionCloseError{
   407  			Code:   uint64(e.code),
   408  			Reason: e.message,
   409  		})
   410  	} else {
   411  		cc.qconn.Abort(err)
   412  	}
   413  }
   414
View as plain text