XDRIVE transport: Support h3 (Chrome CH by default) and XMUX as well; Send Chrome HTTP headers (User-Agent, etc.) by default (#7187)

https://github.com/XTLS/Xray-core/pull/7181#issuecomment-6229694351
https://github.com/XTLS/Xray-core/pull/7187#issuecomment-6250200781
https://github.com/XTLS/Xray-core/pull/7187#issuecomment-6314784068

infra/conf/transport_internet.go

@@ -90,6 +90,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
 		config.ProtocolName = protocol
 	}
 
+	var xdriveH3 bool
 	switch strings.ToLower(c.Security) {
 	case "", "none":
 	case "tls":
@@ -104,6 +105,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
 		tm := serial.ToTypedMessage(ts)
 		config.SecuritySettings = append(config.SecuritySettings, tm)
 		config.SecurityType = tm.Type
+		xdriveH3 = config.ProtocolName == "xdrive" && tlsSettings.ALPN != nil && slices.Equal(*tlsSettings.ALPN, StringList{"h3"})
 		if config.ProtocolName == "xdrive" && tlsSettings.ALPN != nil && slices.Equal(*tlsSettings.ALPN, StringList{"http/1.1"}) && tlsSettings.ECHConfigList == "" {
 			// a fingerprint puts its own ALPN into the hello, except Go TLS and the
 			// randomized ones, which take it from the config or send none
@@ -219,6 +221,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
 		})
 	}
 	if c.XDRIVESettings != nil {
+		c.XDRIVESettings.h3 = xdriveH3
 		xs, err := c.XDRIVESettings.Build()
 		if err != nil {
 			return nil, errors.New("Failed to build XDRIVE config.").Base(err)
@@ -260,5 +263,20 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
 		}
 	}
 
+	if xdriveH3 && c.FinalMask != nil {
+		if c.XDRIVESettings != nil && c.XDRIVESettings.Service == "google-drive" {
+			// noise only sends datagrams of its own ahead of the QUIC ones, the
+			// other masks change those or need a peer that undoes them
+			for _, mask := range c.FinalMask.Udp {
+				if strings.ToLower(mask.Type) != "noise" {
+					return nil, errors.New(`XDRIVE: over h3 to Google Drive the only "udp" mask that works is "noise", not "`, mask.Type, `"`)
+				}
+			}
+		}
+		if config.QuicParams != nil && (config.QuicParams.Congestion == "brutal" || config.QuicParams.Congestion == "force-brutal") {
+			return nil, errors.New(`XDRIVE: brutal is for Hysteria servers, over h3 "congestion" is "bbr" or "reno", not "`, config.QuicParams.Congestion, `"`)
+		}
+	}
+
 	return config, nil
 }

infra/conf/transport_method.go

@@ -997,6 +997,10 @@ type XDriveConfig struct {
 	EagerWindowMs     uint32          `json:"eagerWindowMs"`
 	HoleTimeoutMs     uint32          `json:"holeTimeoutMs"`
 	Template          json.RawMessage `json:"template"`
+	Xmux              XmuxConfig      `json:"xmux"`
+
+	// set by StreamConfig when "alpn" is ["h3"]
+	h3 bool
 }
 
 // Build implements Buildable.
@@ -1018,6 +1022,21 @@ func (c *XDriveConfig) Build() (proto.Message, error) {
 	default:
 		return nil, errors.New(`unknown "service": "`, c.Service, `", use "google-drive", "template" or "local"`)
 	}
+	if c.Xmux.MaxConnections.To > 0 && c.Xmux.MaxConcurrency.To > 0 {
+		return nil, errors.New("maxConnections cannot be specified together with maxConcurrency")
+	}
+	if c.Xmux == (XmuxConfig{}) {
+		c.Xmux.MaxConnections.From = 1
+		c.Xmux.MaxConnections.To = 1
+		if c.Service == "google-drive" && !c.h3 {
+			// Google gives an h2 connection 1 MiB of window, which a few sessions on
+			// a long rtt would share, an h3 connection has no such limit
+			c.Xmux.MaxConnections.From = 4
+			c.Xmux.MaxConnections.To = 4
+		}
+		c.Xmux.HMaxReusableSecs.From = 1800
+		c.Xmux.HMaxReusableSecs.To = 3000
+	}
 	config := &xdrive.Config{
 		RemoteFolder:      c.RemoteFolder,
 		Service:           c.Service,
@@ -1031,6 +1050,14 @@ func (c *XDriveConfig) Build() (proto.Message, error) {
 		EagerWindowMs:     c.EagerWindowMs,
 		HoleTimeoutMs:     c.HoleTimeoutMs,
 		Template:          string(c.Template),
+		Xmux: &splithttp.XmuxConfig{
+			MaxConcurrency:   newRangeConfig(c.Xmux.MaxConcurrency),
+			MaxConnections:   newRangeConfig(c.Xmux.MaxConnections),
+			CMaxReuseTimes:   newRangeConfig(c.Xmux.CMaxReuseTimes),
+			HMaxRequestTimes: newRangeConfig(c.Xmux.HMaxRequestTimes),
+			HMaxReusableSecs: newRangeConfig(c.Xmux.HMaxReusableSecs),
+			HKeepAlivePeriod: c.Xmux.HKeepAlivePeriod,
+		},
 	}
 	return config, nil
 }

infra/conf/transport_test.go

@@ -9,6 +9,7 @@ import (
 	. "github.com/xtls/xray-core/infra/conf"
 	"github.com/xtls/xray-core/transport/internet"
 	finalmaskcustom "github.com/xtls/xray-core/transport/internet/finalmask/header/custom"
+	"github.com/xtls/xray-core/transport/internet/splithttp"
 	"github.com/xtls/xray-core/transport/internet/xdrive"
 	"google.golang.org/protobuf/proto"
 )
@@ -424,6 +425,81 @@ func TestXDriveHTTP1ALPNWarning(t *testing.T) {
 	}
 }
 
+// over h3 Google Drive takes only the noise mask, and no service takes brutal
+func TestXDriveH3Checks(t *testing.T) {
+	const (
+		drive       = `"method": "xdrive", "xdriveSettings": {"remoteFolder": "f", "service": "google-drive", "secrets": ["a", "b", "c"]}`
+		local       = `"method": "xdrive", "xdriveSettings": {"remoteFolder": "/tmp/xdrive", "service": "local"}`
+		noise       = `{"type": "Noise", "settings": {"noise": [{"rand": "50-100"}]}}`
+		salamander  = `{"type": "salamander", "settings": {"password": "p"}}`
+		forceBrutal = `{"quicParams": {"congestion": "force-brutal", "brutalUp": "10 mbps"}}`
+	)
+	for _, c := range []struct {
+		alpn, transport, finalmask string
+		ok                         bool
+	}{
+		{`["h3"]`, drive, `{"udp": [` + noise + `]}`, true},
+		{`["h3"]`, drive, `{"udp": [` + salamander + `]}`, false},
+		{`["h3"]`, drive, `{"udp": [` + noise + `, ` + salamander + `]}`, false},
+		{`["h3"]`, local, `{"udp": [` + salamander + `]}`, true},
+		{`["h3"]`, `"method": "xdrive"`, `{"udp": [` + salamander + `]}`, true},
+		{`["h2"]`, drive, `{"udp": [` + salamander + `]}`, true},
+		{`null`, drive, `{"udp": [` + salamander + `]}`, true},
+		{`["h3"]`, drive, `null`, true},
+		{`["h3"]`, drive, `{"quicParams": {"congestion": "bbr"}}`, true},
+		{`["h3"]`, drive, `{"quicParams": {"congestion": "brutal"}}`, false},
+		{`["h3"]`, local, forceBrutal, false},
+		{`["h3", "h2"]`, drive, `{"quicParams": {"congestion": "brutal"}}`, true},
+		{`["h3"]`, `"method": "xhttp"`, forceBrutal, true},
+	} {
+		config := new(StreamConfig)
+		if err := json.Unmarshal([]byte(`{
+			`+c.transport+`,
+			"security": "tls",
+			"tlsSettings": {"alpn": `+c.alpn+`},
+			"finalmask": `+c.finalmask+`
+		}`), config); err != nil {
+			t.Fatalf("Unmarshal: %v", err)
+		}
+		if _, err := config.Build(); (err == nil) != c.ok {
+			t.Errorf("alpn %s, %s, finalmask %s: %v", c.alpn, c.transport, c.finalmask, err)
+		}
+	}
+}
+
+// one h3 connection is not held back by a window like h2, so h3 takes one by default
+func TestXDriveH3Connections(t *testing.T) {
+	for _, c := range []struct {
+		alpn string
+		want int32
+	}{
+		{`["h3"]`, 1},
+		{`["h2"]`, 4},
+		{`null`, 4},
+	} {
+		config := new(StreamConfig)
+		if err := json.Unmarshal([]byte(`{
+			"method": "xdrive",
+			"xdriveSettings": {"remoteFolder": "f", "service": "google-drive", "secrets": ["a", "b", "c"]},
+			"security": "tls",
+			"tlsSettings": {"alpn": `+c.alpn+`}
+		}`), config); err != nil {
+			t.Fatalf("Unmarshal: %v", err)
+		}
+		built, err := config.Build()
+		if err != nil {
+			t.Fatalf("alpn %s: %v", c.alpn, err)
+		}
+		xs, err := built.TransportSettings[0].Settings.GetInstance()
+		if err != nil {
+			t.Fatalf("alpn %s: %v", c.alpn, err)
+		}
+		if got := xs.(*xdrive.Config).Xmux.MaxConnections; got.From != c.want || got.To != c.want {
+			t.Errorf("alpn %s: maxConnections is %d-%d, want %d", c.alpn, got.From, got.To, c.want)
+		}
+	}
+}
+
 func TestXDriveTemplateNeedsTemplate(t *testing.T) {
 	config := new(XDriveConfig)
 	if err := json.Unmarshal([]byte(`{"remoteFolder": "f", "service": "template"}`), config); err != nil {
@@ -513,3 +589,50 @@ func TestXDriveInboundConfig(t *testing.T) {
 		}
 	}
 }
+
+func TestXDriveXmux(t *testing.T) {
+	defaults := func(connections int32) *splithttp.XmuxConfig {
+		return &splithttp.XmuxConfig{
+			MaxConcurrency:   &splithttp.RangeConfig{},
+			MaxConnections:   &splithttp.RangeConfig{From: connections, To: connections},
+			CMaxReuseTimes:   &splithttp.RangeConfig{},
+			HMaxRequestTimes: &splithttp.RangeConfig{},
+			HMaxReusableSecs: &splithttp.RangeConfig{From: 1800, To: 3000},
+		}
+	}
+	for _, c := range []struct {
+		input string
+		want  *splithttp.XmuxConfig
+	}{
+		{`{"service": "google-drive", "secrets": ["a", "b", "c"]}`, defaults(4)},
+		{`{"service": "template", "template": {}}`, defaults(1)},
+		{`{"service": "google-drive", "secrets": ["a", "b", "c"], "xmux": {"maxConcurrency": "1-2", "hMaxRequestTimes": 100, "hKeepAlivePeriod": -1}}`, &splithttp.XmuxConfig{
+			MaxConcurrency:   &splithttp.RangeConfig{From: 1, To: 2},
+			MaxConnections:   &splithttp.RangeConfig{},
+			CMaxReuseTimes:   &splithttp.RangeConfig{},
+			HMaxRequestTimes: &splithttp.RangeConfig{From: 100, To: 100},
+			HMaxReusableSecs: &splithttp.RangeConfig{},
+			HKeepAlivePeriod: -1,
+		}},
+	} {
+		config := new(XDriveConfig)
+		if err := json.Unmarshal([]byte(c.input), config); err != nil {
+			t.Fatalf("%s: Unmarshal: %v", c.input, err)
+		}
+		built, err := config.Build()
+		if err != nil {
+			t.Fatalf("%s: Build: %v", c.input, err)
+		}
+		if got := built.(*xdrive.Config).Xmux; !proto.Equal(got, c.want) {
+			t.Errorf("%s: xmux is %v, want %v", c.input, got, c.want)
+		}
+	}
+
+	config := new(XDriveConfig)
+	if err := json.Unmarshal([]byte(`{"service": "template", "template": {}, "xmux": {"maxConnections": 2, "maxConcurrency": 8}}`), config); err != nil {
+		t.Fatalf("Unmarshal: %v", err)
+	}
+	if _, err := config.Build(); err == nil {
+		t.Error("Build accepted maxConnections with maxConcurrency")
+	}
+}

transport/internet/xdrive/client.go

@@ -2,14 +2,25 @@ package xdrive
 
 import (
 	"context"
+	gotls "crypto/tls"
 	"io"
 	"net/http"
+	"reflect"
+	"runtime"
+	"slices"
 	"sync"
 	"time"
 
+	"github.com/apernet/quic-go"
+	"github.com/apernet/quic-go/http3"
 	"github.com/xtls/xray-core/common/errors"
 	"github.com/xtls/xray-core/common/net"
+	"github.com/xtls/xray-core/common/net/cnc"
+	"github.com/xtls/xray-core/common/utils"
 	"github.com/xtls/xray-core/transport/internet"
+	"github.com/xtls/xray-core/transport/internet/hysteria/congestion"
+	"github.com/xtls/xray-core/transport/internet/hysteria/congestion/bbr"
+	"github.com/xtls/xray-core/transport/internet/splithttp"
 	"github.com/xtls/xray-core/transport/internet/tls"
 )
 
@@ -39,6 +50,10 @@ func (t *serviceTransport) RoundTrip(r *http.Request) (*http.Response, error) {
 		cancel(nil)
 	}
 	r = r.WithContext(ctx)
+	// the headers of chrome as with XHTTP and DoH, net/http and http3 would send
+	// their own user agent after a hello of chrome
+	r.Header = r.Header.Clone()
+	utils.TryDefaultHeadersWith(r.Header, "fetch")
 	if r.Body != nil && r.Body != http.NoBody {
 		r.Body = &progressBody{ReadCloser: r.Body, alive: alive}
 	}
@@ -104,25 +119,40 @@ func (b *progressBody) Close() error {
 }
 
 // newServiceClient returns a client whose requests end after stall without
-// progress, see serviceTransport.RoundTrip
-func newServiceClient(streamSettings *internet.MemoryStreamConfig, stall time.Duration, maxConns int) *http.Client {
+// progress, see serviceTransport.RoundTrip, and whose connections close when
+// connCtx is done. Clients that share h1Hosts share which hosts answer over h1
+func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings *internet.MemoryStreamConfig, stall time.Duration, maxConns int) *http.Client {
 	var (
 		tlsConfig *tls.Config
 		sockopt   *internet.SocketConfig
 		fronting  *net.Destination
+		keepAlive = net.ChromeH2KeepAlivePeriod
+		// h3 keeps alive only when asked to, see below
+		h3KeepAlive time.Duration
 	)
 	if streamSettings != nil {
 		tlsConfig = tls.ConfigFromStreamSettings(streamSettings)
 		sockopt = streamSettings.SocketSettings
 		fronting = streamSettings.Destination
+		if config, ok := streamSettings.ProtocolSettings.(*Config); ok && config.Xmux != nil {
+			switch period := config.Xmux.HKeepAlivePeriod; {
+			case period > 0:
+				keepAlive = time.Duration(period) * time.Second
+				h3KeepAlive = keepAlive
+			case period < 0:
+				keepAlive = 0
+			}
+		}
 	}
 	if tlsConfig == nil {
 		tlsConfig = &tls.Config{}
 	}
 	fingerprint := tls.GetFingerprint(tlsConfig.Fingerprint)
 
-	dial := func(ctx context.Context, addr string) (net.Conn, net.Destination, error) {
-		host, err := net.ParseDestination("tcp:" + addr)
+	// fronting takes only the address and port, the network stays that of the
+	// dial, which is UDP for h3 though the destination is built as TCP
+	dial := func(ctx context.Context, network, addr string) (net.Conn, net.Destination, error) {
+		host, err := net.ParseDestination(network + ":" + addr)
 		if err != nil {
 			return nil, host, errors.New("bad address: ", addr).Base(err)
 		}
@@ -136,25 +166,36 @@ func newServiceClient(streamSettings *internet.MemoryStreamConfig, stall time.Du
 		}
 
 		var conn net.Conn
-		if streamSettings.FinalMask != nil {
-			conn, err = streamSettings.FinalMask.DialTCP(ctx, target)
-		} else {
+		switch {
+		case streamSettings.FinalMask == nil:
 			conn, err = internet.DialSystem(ctx, target, sockopt)
+		case target.Network == net.Network_UDP:
+			conn, err = streamSettings.FinalMask.DialUDP(ctx, target)
+		default:
+			conn, err = streamSettings.FinalMask.DialTCP(ctx, target)
 		}
 		if err != nil {
 			return nil, host, errors.New("failed to dial to dest").Base(err)
 		}
-		return conn, host, nil
+		// h3 closes its QUIC connection instead, as the server never hears of a UDP
+		// socket that closes
+		if network == "udp" {
+			return conn, host, nil
+		}
+		// CloseIdleConnections leaves an h2 connection open while a stream just
+		// canceled has not left it yet
+		stop := context.AfterFunc(connCtx, func() { conn.Close() })
+		return &ctxConn{Conn: conn, stop: stop}, host, nil
 	}
 
 	dialPlain := func(ctx context.Context, network, addr string) (net.Conn, error) {
-		conn, _, err := dial(ctx, addr)
+		conn, _, err := dial(ctx, "tcp", addr)
 		return conn, err
 	}
 
 	// http.Transport picks h2 or h1 by the ALPN in the ConnectionState of the conn
 	dialTLS := func(ctx context.Context, network, addr string) (net.Conn, error) {
-		conn, host, err := dial(ctx, addr)
+		conn, host, err := dial(ctx, "tcp", addr)
 		if err != nil {
 			return nil, err
 		}
@@ -185,7 +226,86 @@ func newServiceClient(streamSettings *internet.MemoryStreamConfig, stall time.Du
 			MaxIdleConns:        maxConns,
 			MaxIdleConnsPerHost: maxConns,
 			MaxConnsPerHost:     connsPerHost,
-			HTTP2:               &http.HTTP2Config{SendPingTimeout: net.ChromeH2KeepAlivePeriod},
+			HTTP2:               &http.HTTP2Config{SendPingTimeout: keepAlive},
+		}
+	}
+
+	var https http.RoundTripper = &secureTransport{h2: secure(1), h1: secure(maxConns), h1Hosts: h1Hosts}
+	// no server offers h3 by ALPN over TCP, so it is used only when asked for alone
+	if slices.Equal(tlsConfig.NextProtocol, []string{"h3"}) {
+		quicParams := streamSettings.QuicParams
+		if quicParams == nil {
+			quicParams = &internet.QuicParams{BbrProfile: string(bbr.ProfileStandard)}
+		}
+		// the defaults are those of quic-go, with no keepalive unless asked for, as
+		// in DoH: http3 never closes a connection that is idle or that it dropped
+		// after an error, so only the idle timeout or the end of connCtx ends it
+		quicConfig := &quic.Config{
+			InitialStreamReceiveWindow:     quicParams.InitStreamReceiveWindow,
+			MaxStreamReceiveWindow:         quicParams.MaxStreamReceiveWindow,
+			InitialConnectionReceiveWindow: quicParams.InitConnReceiveWindow,
+			MaxConnectionReceiveWindow:     quicParams.MaxConnReceiveWindow,
+			MaxIdleTimeout:                 time.Duration(quicParams.MaxIdleTimeout) * time.Second,
+			KeepAlivePeriod:                time.Duration(quicParams.KeepAlivePeriod) * time.Second,
+			MaxIncomingStreams:             quicParams.MaxIncomingStreams,
+			DisablePathMTUDiscovery:        quicParams.DisablePathMtuDiscovery || (runtime.GOOS != "linux" && runtime.GOOS != "windows" && runtime.GOOS != "darwin"),
+			ChromeParrot:                   !quicParams.DisableChromeParrot,
+		}
+		if quicConfig.KeepAlivePeriod == 0 {
+			quicConfig.KeepAlivePeriod = h3KeepAlive
+		}
+
+		https = &http3.Transport{
+			QUICConfig: quicConfig,
+			Dial: func(ctx context.Context, addr string, _ *gotls.Config, cfg *quic.Config) (*quic.Conn, error) {
+				switch quicParams.Congestion {
+				case "", "bbr", "reno":
+				default:
+					// brutal is for Hysteria servers, not services, and a panic here, in
+					// a goroutine of http3, would end the process
+					return nil, errors.New("unsupported congestion control over h3: ", quicParams.Congestion)
+				}
+
+				conn, host, err := dial(ctx, "udp", addr)
+				if err != nil {
+					return nil, err
+				}
+				var pktConn net.PacketConn
+				var udpAddr net.Addr
+				switch c := conn.(type) {
+				case *net.PacketConnWrapper:
+					pktConn = c.PacketConn
+					udpAddr = c.RemoteAddr()
+				case *cnc.Connection:
+					pktConn = &internet.FakePacketConn{Conn: c}
+					udpAddr = &net.UDPAddr{IP: []byte{0, 0, 0, 0}}
+				default:
+					conn.Close()
+					return nil, errors.New("failed to dial to dest: unexpected ", reflect.TypeOf(c))
+				}
+
+				// built for each dial with the host as SNI, as for h2: http3 sets the
+				// SNI on a copy, which a pinned CA and the ECH lookup do not see
+				gotlsConfig := tlsConfig.GetTLSConfig(tls.WithDestination(host))
+				// it is for servers, and ChromeParrot refuses a config that has it
+				gotlsConfig.GetCertificate = nil
+				tr := &quic.Transport{Conn: pktConn, DisableGSO: quicParams.DisableGSO}
+				if !quicParams.DisableChromeParrot {
+					tr.ConnectionIDGenerator = quic.ZeroLengthConnectionIDGenerator{}
+				}
+				qconn, err := tr.Dial(ctx, udpAddr, gotlsConfig, cfg)
+				if err != nil {
+					tr.Close()
+					pktConn.Close()
+					return nil, err
+				}
+				stop := context.AfterFunc(connCtx, func() { qconn.CloseWithError(quic.ApplicationErrorCode(http3.ErrCodeNoError), "") })
+				context.AfterFunc(qconn.Context(), func() { stop(); tr.Close(); pktConn.Close() })
+				if quicParams.Congestion != "reno" {
+					congestion.UseBBR(qconn, bbr.Profile(quicParams.BbrProfile))
+				}
+				return qconn, nil
+			},
 		}
 	}
 
@@ -198,18 +318,30 @@ func newServiceClient(streamSettings *internet.MemoryStreamConfig, stall time.Du
 				MaxIdleConnsPerHost: maxConns,
 				MaxConnsPerHost:     maxConns,
 			},
-			secure: &secureTransport{h2: secure(1), h1: secure(maxConns)},
+			secure: https,
 			stall:  stall,
 		},
 	}
 }
 
+// ctxConn is a connection that connCtx no longer closes once it is closed,
+// so that one that ends on its own leaves nothing behind in connCtx
+type ctxConn struct {
+	net.Conn
+	stop func() bool
+}
+
+func (c *ctxConn) Close() error {
+	c.stop()
+	return c.Conn.Close()
+}
+
 // secureTransport keeps a host to one connection, all that h2 needs, as without
 // a limit parallel requests would each dial their own. A host that answers over
 // h1 moves to a transport with more, since h1 carries one request at a time
 type secureTransport struct {
 	h2, h1  http.RoundTripper
-	h1Hosts sync.Map
+	h1Hosts *sync.Map
 }
 
 func (t *secureTransport) RoundTrip(r *http.Request) (*http.Response, error) {
@@ -222,3 +354,129 @@ func (t *secureTransport) RoundTrip(r *http.Request) (*http.Response, error) {
 	}
 	return resp, err
 }
+
+var errSessionClosed = errors.New("the session is closed")
+
+// connPool hands the sessions of a storage the connections of XMUX, each a
+// client of its own made by newClient
+type connPool struct {
+	mu   sync.Mutex
+	xmux *splithttp.XmuxManager
+}
+
+func newConnPool(config *splithttp.XmuxConfig, newClient func(connCtx context.Context) *http.Client) *connPool {
+	var xmuxConfig splithttp.XmuxConfig
+	if config != nil {
+		xmuxConfig = *config
+	}
+	return &connPool{xmux: splithttp.NewXmuxManager(xmuxConfig, func() splithttp.XmuxConn {
+		ctx, cancel := context.WithCancel(context.Background())
+		return &poolConn{next: newClient(ctx).Transport, cancel: cancel}
+	})}
+}
+
+// poolConn is a connection of XMUX, which closes once XMUX drops it and the
+// requests on it end
+type poolConn struct {
+	next   http.RoundTripper
+	cancel context.CancelFunc
+
+	mu       sync.Mutex
+	inflight int
+	closing  bool
+}
+
+func (c *poolConn) IsClosed() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.closing
+}
+
+func (c *poolConn) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.closing = true
+	if c.inflight == 0 {
+		c.cancel()
+	}
+	return nil
+}
+
+func (c *poolConn) done() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.inflight--
+	if c.closing && c.inflight == 0 {
+		c.cancel()
+	}
+}
+
+// roundTrip sends a request counted in inflight by the caller
+func (c *poolConn) roundTrip(r *http.Request) (*http.Response, error) {
+	resp, err := c.next.RoundTrip(r)
+	if err != nil {
+		c.done()
+		return nil, err
+	}
+	resp.Body = &doneBody{ReadCloser: resp.Body, done: sync.OnceFunc(c.done)}
+	return resp, nil
+}
+
+type doneBody struct {
+	io.ReadCloser
+	done func()
+}
+
+func (b *doneBody) Close() error {
+	err := b.ReadCloser.Close()
+	b.done()
+	return err
+}
+
+// sessionTransport sends the requests of one session over a connection of
+// XMUX, and over another once that one has no requests or time left
+type sessionTransport struct {
+	pool *connPool
+
+	mu     sync.Mutex
+	client *splithttp.XmuxClient
+	closed bool
+}
+
+func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
+	t.mu.Lock()
+	if t.closed {
+		t.mu.Unlock()
+		return nil, errSessionClosed
+	}
+	client := t.client
+	if client == nil || client.LeftRequests.Add(-1) < 0 || !client.UnreusableAt.IsZero() && time.Now().After(client.UnreusableAt) {
+		t.pool.mu.Lock()
+		t.client = t.pool.xmux.GetXmuxClient(r.Context())
+		t.pool.mu.Unlock()
+		t.client.AddRunning()
+		t.client.LeftRequests.Add(-1)
+		if client != nil {
+			client.DoneRunning()
+		}
+		client = t.client
+	}
+	// counted under mu, or another request of the session that moves on could
+	// close the connection before this one is sent
+	conn := client.XmuxConn.(*poolConn)
+	conn.mu.Lock()
+	conn.inflight++
+	conn.mu.Unlock()
+	t.mu.Unlock()
+	return conn.roundTrip(r)
+}
+
+func (t *sessionTransport) Close() {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.closed = true
+	if t.client != nil {
+		t.client.DoneRunning()
+		t.client = nil
+	}
+}

transport/internet/xdrive/client_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	gotls "crypto/tls"
+	"crypto/x509"
 	"encoding/base64"
 	"errors"
 	"io"
@@ -11,13 +12,22 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"slices"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/apernet/quic-go"
+	"github.com/apernet/quic-go/http3"
 	xnet "github.com/xtls/xray-core/common/net"
+	"github.com/xtls/xray-core/common/net/cnc"
+	"github.com/xtls/xray-core/common/protocol/tls/cert"
+	"github.com/xtls/xray-core/common/utils"
 	"github.com/xtls/xray-core/transport/internet"
+	"github.com/xtls/xray-core/transport/internet/finalmask"
+	"github.com/xtls/xray-core/transport/internet/finalmask/noise"
+	"github.com/xtls/xray-core/transport/internet/splithttp"
 	"github.com/xtls/xray-core/transport/internet/tls"
 )
 
@@ -75,7 +85,7 @@ func sniForSettings(t *testing.T, serverName string) string {
 	driveFilesURL = "https://www.googleapis.com/drive/v3/files"
 	defer func() { driveFilesURL = prev }()
 
-	client := newServiceClient(settings, 5*time.Second, 8)
+	client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8)
 	req, err := http.NewRequest(http.MethodGet, driveFilesURL, nil)
 	if err != nil {
 		t.Fatalf("request: %v", err)
@@ -145,7 +155,7 @@ func TestServiceClientEndsOnlyStalledRequests(t *testing.T) {
 		}
 	}))
 	defer server.Close()
-	client := newServiceClient(&internet.MemoryStreamConfig{}, stall, 8)
+	client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, stall, 8)
 
 	fetch := func(path string, body io.Reader) (int, error) {
 		req, err := http.NewRequest(http.MethodPost, server.URL+path, body)
@@ -217,7 +227,7 @@ func TestServiceClientStallOverTLS(t *testing.T) {
 			}
 			time.Sleep(5 * stall)
 		})
-		client := newServiceClient(server.settings(&tls.Config{}), stall, 8)
+		client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), stall, 8)
 
 		for _, path := range []string{"/stuck", "/silent"} {
 			start := time.Now()
@@ -240,12 +250,13 @@ func TestServiceClientStallOverTLS(t *testing.T) {
 }
 
 // tlsServer is a TLS server that keeps the last hello it got and counts its
-// connections
+// connections, and those closed
 type tlsServer struct {
 	*httptest.Server
-	mu    sync.Mutex
-	hello *gotls.ClientHelloInfo
-	conns atomic.Int32
+	mu     sync.Mutex
+	hello  *gotls.ClientHelloInfo
+	conns  atomic.Int32
+	closed atomic.Int32
 }
 
 func newTLSServer(t *testing.T, config *gotls.Config, handler http.HandlerFunc) *tlsServer {
@@ -260,8 +271,11 @@ func newTLSServer(t *testing.T, config *gotls.Config, handler http.HandlerFunc)
 	}
 	s.TLS = config
 	s.Config.ConnState = func(_ net.Conn, state http.ConnState) {
-		if state == http.StateNew {
+		switch state {
+		case http.StateNew:
 			s.conns.Add(1)
+		case http.StateClosed:
+			s.closed.Add(1)
 		}
 	}
 	s.StartTLS()
@@ -298,7 +312,7 @@ func TestServiceClientSpeaksWhatServerPicks(t *testing.T) {
 	for _, protos := range [][]string{chromeALPN, {"http/1.1"}} {
 		server := newTLSServer(t, &gotls.Config{NextProtos: protos}, func(w http.ResponseWriter, r *http.Request) {})
 		for _, config := range []*tls.Config{{}, {NextProtocol: []string{"h2"}}, {NextProtocol: []string{"http/1.1"}}, {Fingerprint: "unsafe"}} {
-			client := newServiceClient(server.settings(config), 5*time.Second, 8)
+			client := newServiceClient(context.Background(), new(sync.Map), server.settings(config), 5*time.Second, 8)
 			resp, err := client.Get(server.URL)
 			if err != nil {
 				t.Fatalf("server %v, alpn %v, fingerprint %q: %v", protos, config.NextProtocol, config.Fingerprint, err)
@@ -318,7 +332,7 @@ func TestServiceClientSpeaksWhatServerPicks(t *testing.T) {
 // offers certificate compression
 func TestServiceClientDefaultsToChrome(t *testing.T) {
 	server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) {})
-	client := newServiceClient(&internet.MemoryStreamConfig{}, 5*time.Second, 8)
+	client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8)
 	// the test certificate is not trusted, the hello is all that is needed
 	if resp, err := client.Get(server.URL); err == nil {
 		resp.Body.Close()
@@ -332,6 +346,33 @@ func TestServiceClientDefaultsToChrome(t *testing.T) {
 	}
 }
 
+// a request says it is chrome, as the hello does, unless it says what it is
+func TestServiceClientUserAgent(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		io.WriteString(w, r.UserAgent())
+	}))
+	defer server.Close()
+	client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8)
+	for ua, want := range map[string]string{"": utils.ChromeUA, "firefox": utils.FirefoxUA, "custom/1.0": "custom/1.0"} {
+		req, _ := http.NewRequest(http.MethodGet, server.URL, nil)
+		if ua != "" {
+			req.Header.Set("User-Agent", ua)
+		}
+		resp, err := client.Do(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		got, _ := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		if string(got) != want {
+			t.Errorf("with %q the server got %q, not %q", ua, got, want)
+		}
+		if req.Header.Get("User-Agent") != ua || len(req.Header) > min(len(ua), 1) {
+			t.Errorf("with %q the request was changed to %v", ua, req.Header)
+		}
+	}
+}
+
 // parallel requests share one h2 connection instead of dialing one each, the
 // first ones as the later ones, and next to a host that answers over h1
 func TestServiceClientOneConnection(t *testing.T) {
@@ -339,7 +380,7 @@ func TestServiceClientOneConnection(t *testing.T) {
 		time.Sleep(50 * time.Millisecond)
 	})
 	h1 := newTLSServer(t, &gotls.Config{NextProtos: []string{"http/1.1"}}, func(w http.ResponseWriter, r *http.Request) {})
-	client := newServiceClient(server.settings(&tls.Config{}), 5*time.Second, driveMaxInflight)
+	client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), 5*time.Second, driveMaxInflight)
 	get := func(url string) {
 		resp, err := client.Get(url)
 		if err != nil {
@@ -395,7 +436,7 @@ func TestServiceClientHTTP1(t *testing.T) {
 		server := newTLSServer(t, &gotls.Config{NextProtos: c.protos, EncryptedClientHelloKeys: echKeys}, func(w http.ResponseWriter, r *http.Request) {
 			time.Sleep(50 * time.Millisecond)
 		})
-		client := newServiceClient(server.settings(c.config), 5*time.Second, 8)
+		client := newServiceClient(context.Background(), new(sync.Map), server.settings(c.config), 5*time.Second, 8)
 		get := func() {
 			resp, err := client.Get(server.URL)
 			if err != nil {
@@ -426,7 +467,7 @@ func TestServiceClientHTTP1(t *testing.T) {
 	// without the keys the server gets the outer hello, which offers the ALPN of
 	// Chrome
 	outer := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) {})
-	if resp, err := newServiceClient(outer.settings(ech), 5*time.Second, 8).Get(outer.URL); err == nil {
+	if resp, err := newServiceClient(context.Background(), new(sync.Map), outer.settings(ech), 5*time.Second, 8).Get(outer.URL); err == nil {
 		resp.Body.Close()
 	}
 	if hello := outer.lastHello(); hello == nil {
@@ -462,7 +503,7 @@ func TestServiceClientHungHandshake(t *testing.T) {
 		server.Listener = &muteFirst{Listener: server.Listener}
 		server.StartTLS()
 		t.Cleanup(server.Close)
-		client := newServiceClient(server.settings(&tls.Config{Fingerprint: fingerprint}), stall, 8)
+		client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{Fingerprint: fingerprint}), stall, 8)
 
 		if _, err := client.Get(server.URL); err == nil {
 			t.Fatalf("fingerprint %q: a request got through a handshake that hung", fingerprint)
@@ -488,7 +529,7 @@ func TestServiceClientClosesFailedHandshake(t *testing.T) {
 
 	for _, fingerprint := range []string{"", "unsafe"} {
 		settings := &internet.MemoryStreamConfig{ProtocolName: protocolName, SecuritySettings: &tls.Config{Fingerprint: fingerprint}}
-		go newServiceClient(settings, 5*time.Second, 8).Get("https://" + ln.Addr().String())
+		go newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8).Get("https://" + ln.Addr().String())
 		conn, err := ln.Accept()
 		if err != nil {
 			t.Fatal(err)
@@ -503,6 +544,446 @@ func TestServiceClientClosesFailedHandshake(t *testing.T) {
 	}
 }
 
+// afterFuncs counts the functions that context.AfterFunc registers on it and
+// that are not stopped yet
+type afterFuncs struct {
+	context.Context
+	live atomic.Int32
+}
+
+// Value hides the cancelCtx below, which context.AfterFunc would use instead
+func (c *afterFuncs) Value(any) any { return nil }
+
+func (c *afterFuncs) AfterFunc(func()) func() bool {
+	c.live.Add(1)
+	return func() bool {
+		c.live.Add(-1)
+		return true
+	}
+}
+
+// a connection that ends on its own, here after Connection: close, leaves
+// nothing on the context that closes the connections of the client
+func TestServiceClientForgetsClosedConnections(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Connection", "close")
+	}))
+	defer server.Close()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	connCtx := &afterFuncs{Context: ctx}
+	client := newServiceClient(connCtx, new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8)
+
+	for range 4 {
+		resp, err := client.Get(server.URL)
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp.Body.Close()
+	}
+	waitFor(t, "the closed connections to leave the context", func() bool { return connCtx.live.Load() == 0 })
+}
+
 type roundTrip func(*http.Request) (*http.Response, error)
 
 func (f roundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
+
+// "hKeepAlivePeriod" of XMUX sets when an idle h2 connection pings, -1 never,
+// and h3 keeps alive only when it is set
+func TestServiceClientKeepAlive(t *testing.T) {
+	for period, want := range map[int64]time.Duration{0: xnet.ChromeH2KeepAlivePeriod, 20: 20 * time.Second, -1: 0} {
+		settings := &internet.MemoryStreamConfig{ProtocolSettings: &Config{Xmux: &splithttp.XmuxConfig{HKeepAlivePeriod: period}}}
+		client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8)
+		h2 := client.Transport.(*serviceTransport).secure.(*secureTransport).h2.(*http.Transport)
+		if got := h2.HTTP2.SendPingTimeout; got != want {
+			t.Errorf("hKeepAlivePeriod %d pings after %v, want %v", period, got, want)
+		}
+	}
+	for period, want := range map[int64]time.Duration{0: 0, 20: 20 * time.Second, -1: 0} {
+		settings := &internet.MemoryStreamConfig{
+			ProtocolSettings: &Config{Xmux: &splithttp.XmuxConfig{HKeepAlivePeriod: period}},
+			SecuritySettings: &tls.Config{NextProtocol: []string{"h3"}},
+		}
+		client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8)
+		if got := client.Transport.(*serviceTransport).secure.(*http3.Transport).QUICConfig.KeepAlivePeriod; got != want {
+			t.Errorf("hKeepAlivePeriod %d keeps h3 alive every %v, want %v", period, got, want)
+		}
+	}
+}
+
+// h3 is used when "alpn" asks for it alone, and only then
+func TestServiceClientH3Selection(t *testing.T) {
+	for _, c := range []struct {
+		alpn []string
+		h3   bool
+	}{
+		{nil, false},
+		{[]string{"h2"}, false},
+		{[]string{"http/1.1"}, false},
+		{[]string{"h3", "h2"}, false},
+		{[]string{"h2", "h3"}, false},
+		{[]string{"h3"}, true},
+	} {
+		client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{SecuritySettings: &tls.Config{NextProtocol: c.alpn}}, 5*time.Second, 8)
+		if _, h3 := client.Transport.(*serviceTransport).secure.(*http3.Transport); h3 != c.h3 {
+			t.Errorf("alpn %v: h3 %v, want %v", c.alpn, h3, c.h3)
+		}
+	}
+	if _, h3 := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8).Transport.(*serviceTransport).secure.(*http3.Transport); h3 {
+		t.Error("h3 without tlsSettings")
+	}
+}
+
+// h3Server is an http3 server that keeps the SNI of each connection, and counts
+// its connections and those closed
+type h3Server struct {
+	port         xnet.Port
+	hash, caHash [32]byte
+	mu           sync.Mutex
+	snis         []string
+	conns        atomic.Int32
+	closed       atomic.Int32
+}
+
+func newH3Server(t *testing.T, handler http.Handler) *h3Server {
+	t.Helper()
+
+	// ECDSA, as ChromeParrot offers no Ed25519, from a CA of its own
+	ca, caHash := cert.MustGenerate(nil, cert.Authority(true), cert.KeyUsage(x509.KeyUsageCertSign))
+	certificate, hash := cert.MustGenerate(ca, cert.DNSNames("www.googleapis.com", "oauth2.googleapis.com"))
+	key, err := x509.ParsePKCS8PrivateKey(certificate.PrivateKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	s := &h3Server{port: xnet.Port(conn.LocalAddr().(*net.UDPAddr).Port), hash: hash, caHash: caHash}
+	server := &http3.Server{
+		Handler: handler,
+		TLSConfig: http3.ConfigureTLSConfig(&gotls.Config{
+			Certificates: []gotls.Certificate{{Certificate: [][]byte{certificate.Certificate, ca.Certificate}, PrivateKey: key}},
+			GetConfigForClient: func(hello *gotls.ClientHelloInfo) (*gotls.Config, error) {
+				s.mu.Lock()
+				s.snis = append(s.snis, hello.ServerName)
+				s.mu.Unlock()
+				return nil, nil
+			},
+		}),
+		ConnContext: func(ctx context.Context, c *quic.Conn) context.Context {
+			s.conns.Add(1)
+			context.AfterFunc(c.Context(), func() { s.closed.Add(1) })
+			return ctx
+		},
+	}
+	go server.Serve(conn)
+	t.Cleanup(func() {
+		server.Close()
+		conn.Close()
+	})
+	return s
+}
+
+// settings sends every host to the server over h3, through a destination built
+// as TCP the way conf builds it, and pins the certificate of the server in config
+func (s *h3Server) settings(config *tls.Config) *internet.MemoryStreamConfig {
+	config.NextProtocol = []string{"h3"}
+	config.PinnedPeerCertSha256 = [][]byte{s.hash[:]}
+	return &internet.MemoryStreamConfig{
+		ProtocolName:     protocolName,
+		Destination:      &xnet.Destination{Address: xnet.LocalHostIP, Port: s.port, Network: xnet.Network_TCP},
+		SecuritySettings: config,
+	}
+}
+
+func (s *h3Server) seenSNIs() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return slices.Sorted(slices.Values(s.snis))
+}
+
+// one client serves two hosts, and each gets its own SNI unless serverName
+// says otherwise, which is also the name a pinned CA checks the certificate for
+func TestServiceClientH3SNI(t *testing.T) {
+	for _, c := range []struct {
+		serverName string
+		pinCA      bool
+		want       []string
+	}{
+		{"", false, []string{"oauth2.googleapis.com", "www.googleapis.com"}},
+		{"www.google.com", false, []string{"www.google.com", "www.google.com"}},
+		{"", true, []string{"oauth2.googleapis.com", "www.googleapis.com"}},
+	} {
+		server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+		settings := server.settings(&tls.Config{ServerName: c.serverName})
+		if c.pinCA {
+			settings.SecuritySettings.(*tls.Config).PinnedPeerCertSha256 = [][]byte{server.caHash[:]}
+		}
+		client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8)
+		for _, url := range []string{"https://www.googleapis.com/drive/v3/files", "https://oauth2.googleapis.com/token"} {
+			resp, err := client.Get(url)
+			if err != nil {
+				t.Fatalf("serverName %q, pinned CA %v, %s: %v", c.serverName, c.pinCA, url, err)
+			}
+			resp.Body.Close()
+			if resp.Proto != "HTTP/3.0" {
+				t.Errorf("serverName %q, %s: went over %s", c.serverName, url, resp.Proto)
+			}
+		}
+		if snis := server.seenSNIs(); !slices.Equal(snis, c.want) {
+			t.Errorf("serverName %q: SNIs %v, want %v", c.serverName, snis, c.want)
+		}
+	}
+}
+
+// a request that stands still ends over h3 too, and says so
+func TestServiceClientStallOverH3(t *testing.T) {
+	const stall = 200 * time.Millisecond
+	server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/stuck" {
+			w.Write([]byte("x"))
+			w.(http.Flusher).Flush()
+		}
+		time.Sleep(5 * stall)
+	}))
+	client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), stall, 8)
+
+	for _, path := range []string{"/stuck", "/silent"} {
+		start := time.Now()
+		resp, err := client.Get("https://www.googleapis.com" + path)
+		if err == nil {
+			if resp.Proto != "HTTP/3.0" {
+				t.Fatalf("%s went over %s", path, resp.Proto)
+			}
+			_, err = io.ReadAll(resp.Body)
+			resp.Body.Close()
+		}
+		if !errors.Is(err, errStalled) {
+			t.Errorf("%s gave %v, want %v", path, err, errStalled)
+		}
+		if took := time.Since(start); took > 4*stall {
+			t.Errorf("%s took %v to give up", path, took)
+		}
+	}
+}
+
+// brutal and force-brutal, which are for Hysteria servers, give an error and
+// no panic in a goroutine of http3, where XHTTP and DoH panic on brutal
+func TestServiceClientH3Congestion(t *testing.T) {
+	server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	for _, cc := range []string{"", "bbr", "reno", "brutal", "force-brutal"} {
+		settings := server.settings(&tls.Config{})
+		settings.QuicParams = &internet.QuicParams{Congestion: cc, BrutalUp: 1 << 20}
+		resp, err := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8).Get("https://www.googleapis.com/")
+		if err == nil {
+			resp.Body.Close()
+		}
+		if brutal := strings.HasSuffix(cc, "brutal"); brutal != (err != nil) || brutal && !strings.Contains(err.Error(), cc) {
+			t.Errorf("congestion %q gave %v", cc, err)
+		}
+	}
+}
+
+// recordConn keeps the datagrams sent through it
+type recordConn struct {
+	net.PacketConn
+	mu     sync.Mutex
+	sent   [][]byte
+	closed atomic.Bool
+}
+
+func (c *recordConn) WriteTo(p []byte, addr net.Addr) (int, error) {
+	c.mu.Lock()
+	c.sent = append(c.sent, bytes.Clone(p))
+	c.mu.Unlock()
+	return c.PacketConn.WriteTo(p, addr)
+}
+
+func (c *recordConn) Close() error {
+	c.closed.Store(true)
+	return c.PacketConn.Close()
+}
+
+// recorder dials UDP through recordConns and keeps them
+type recorder struct {
+	mu    sync.Mutex
+	conns []*recordConn
+}
+
+func (r *recorder) dialUDP(ctx context.Context, dest xnet.Destination) (net.PacketConn, net.Addr, error) {
+	conn, err := internet.DialSystem(ctx, dest, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	rec := &recordConn{PacketConn: conn.(*xnet.PacketConnWrapper).PacketConn}
+	r.mu.Lock()
+	r.conns = append(r.conns, rec)
+	r.mu.Unlock()
+	return rec, conn.RemoteAddr(), nil
+}
+
+// a connection ends with its socket once it is idle for its idle timeout, as
+// keepalive is off unless asked for, and so does one whose handshake failed,
+// and neither leaves anything on the context that closes the connections
+func TestServiceClientH3Closes(t *testing.T) {
+	server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	for _, pin := range [][]byte{server.hash[:], make([]byte, 32)} {
+		rec := new(recorder)
+		settings := server.settings(&tls.Config{})
+		settings.SecuritySettings.(*tls.Config).PinnedPeerCertSha256 = [][]byte{pin}
+		// ChromeParrot would keep the idle timeout at 30 s
+		settings.QuicParams = &internet.QuicParams{MaxIdleTimeout: 1, DisableChromeParrot: true}
+		settings.FinalMask = finalmask.NewFinalMask(nil, nil, nil, nil, rec.dialUDP, nil)
+		connCtx := &afterFuncs{Context: ctx}
+		resp, err := newServiceClient(connCtx, new(sync.Map), settings, 5*time.Second, 8).Get("https://www.googleapis.com/")
+		if err == nil {
+			resp.Body.Close()
+		}
+		if bytes.Equal(pin, server.hash[:]) != (err == nil) {
+			t.Fatalf("pin %x: %v", pin[:4], err)
+		}
+
+		closed := func() bool {
+			rec.mu.Lock()
+			defer rec.mu.Unlock()
+			return len(rec.conns) == 1 && rec.conns[0].closed.Load()
+		}
+		for deadline := time.Now().Add(5 * time.Second); !closed() && time.Now().Before(deadline); {
+			time.Sleep(10 * time.Millisecond)
+		}
+		if !closed() {
+			t.Errorf("pin %x: the socket was left open", pin[:4])
+		}
+		waitFor(t, "the closed connection to leave the context", func() bool { return connCtx.live.Load() == 0 })
+	}
+}
+
+// cncDialer gives UDP as a stream conn, the way dialerProxy does
+type cncDialer struct{}
+
+func (cncDialer) Dial(ctx context.Context, src xnet.Address, dest xnet.Destination, sockopt *internet.SocketConfig) (net.Conn, error) {
+	conn, err := net.Dial("udp", dest.NetAddr())
+	if err != nil {
+		return nil, err
+	}
+	return cnc.NewConnection(cnc.ConnectionInput(conn), cnc.ConnectionOutput(conn), cnc.ConnectionOnClose(conn)), nil
+}
+
+func (cncDialer) DestIpAddress() net.IP { return nil }
+
+// h3 goes over the stream conn of a dialerProxy too
+func TestServiceClientH3DialerProxy(t *testing.T) {
+	internet.UseAlternativeSystemDialer(cncDialer{})
+	defer internet.UseAlternativeSystemDialer(nil)
+
+	server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	resp, err := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), 5*time.Second, 8).Get("https://www.googleapis.com/")
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp.Body.Close()
+	if resp.Proto != "HTTP/3.0" {
+		t.Errorf("went over %s", resp.Proto)
+	}
+}
+
+// xdrive moves data over h3 to a fake Drive, with the noise of a UDP mask sent
+// ahead of the Initial of each connection
+func TestDriveOverH3(t *testing.T) {
+	drive := newFakeDrive(t)
+	server := newH3Server(t, drive.server.Config.Handler)
+	driveTokenURL = "https://oauth2.googleapis.com/token"
+	driveFilesURL = "https://www.googleapis.com/files"
+	driveUploadURL = "https://www.googleapis.com/upload"
+
+	rec := new(recorder)
+	settings := server.settings(&tls.Config{})
+	settings.ProtocolSettings = driveSettings().ProtocolSettings
+	settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{MaxConnections: &splithttp.RangeConfig{From: 4, To: 4}}
+	masks := []finalmask.UDPMask{&noise.Config{Items: []*noise.Item{{Packet: []byte("noise")}}}}
+	settings.FinalMask = finalmask.NewFinalMask(nil, masks, nil, nil, rec.dialUDP, nil)
+
+	client, srv, cleanup := pairWith(t, settings)
+	defer cleanup()
+
+	if _, err := client.Write([]byte("ping")); err != nil {
+		t.Fatalf("client write: %v", err)
+	}
+	expectRead(t, srv, "ping")
+	if _, err := srv.Write([]byte("pong")); err != nil {
+		t.Fatalf("server write: %v", err)
+	}
+	expectRead(t, client, "pong")
+
+	payload := make([]byte, 300000)
+	for i := range payload {
+		payload[i] = byte(i % 251)
+	}
+	go client.Write(payload)
+	if err := srv.SetReadDeadline(time.Now().Add(testPatience)); err != nil {
+		t.Fatalf("SetReadDeadline: %v", err)
+	}
+	got := make([]byte, len(payload))
+	if _, err := io.ReadFull(srv, got); err != nil {
+		t.Fatalf("ReadFull: %v", err)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatal("payload mismatch")
+	}
+
+	rec.mu.Lock()
+	defer rec.mu.Unlock()
+	// one for the listener to www.googleapis.com, one for the token to
+	// oauth2.googleapis.com and one of XMUX for each end of the session
+	if len(rec.conns) != 4 {
+		t.Errorf("%d connections went through the mask, want 4", len(rec.conns))
+	}
+	for i, conn := range rec.conns {
+		conn.mu.Lock()
+		sent := conn.sent
+		conn.mu.Unlock()
+		if len(sent) < 2 {
+			t.Errorf("connection %d sent %d datagrams", i, len(sent))
+			continue
+		}
+		// the noise, then a QUIC v1 Initial of 1250 bytes with no source
+		// connection ID, as ChromeParrot sends it
+		first, initial := sent[0], sent[1]
+		if string(first) != "noise" || len(initial) != 1250 || initial[0]&0xf0 != 0xc0 || !bytes.Equal(initial[1:5], []byte{0, 0, 0, 1}) || initial[6+int(initial[5])] != 0 {
+			t.Errorf("connection %d began with %q and %d bytes %x", i, first, len(initial), initial[:min(len(initial), 24)])
+		}
+	}
+}
+
+// over h3 a connection of XMUX is a QUIC connection, which closes once XMUX
+// drops it, so that the server hears of it
+func TestDriveXmuxOverH3(t *testing.T) {
+	drive := newFakeDrive(t)
+	server := newH3Server(t, drive.server.Config.Handler)
+	driveTokenURL = "https://www.googleapis.com/token"
+	driveFilesURL = "https://www.googleapis.com/files"
+	driveUploadURL = "https://www.googleapis.com/upload"
+	settings := server.settings(&tls.Config{})
+	settings.ProtocolSettings = driveSettings().ProtocolSettings
+	settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{HMaxRequestTimes: &splithttp.RangeConfig{From: 3, To: 3}}
+	storage, err := newDriveStorage(settings, settings.ProtocolSettings.(*Config))
+	if err != nil {
+		t.Fatalf("newDriveStorage: %v", err)
+	}
+	session := sessionStorage(storage)
+	defer session.Close()
+
+	for _, name := range []string{"a", "b", "c", "d"} {
+		if err := session.Put(context.Background(), "streams/abc/c2s/"+name, []byte("x")); err != nil {
+			t.Fatalf("Put: %v", err)
+		}
+	}
+	// one for the token, the first of XMUX for 3 requests and one more
+	if n := server.conns.Load(); n != 3 {
+		t.Errorf("the token and 4 requests with 3 to a connection took %d connections", n)
+	}
+	waitFor(t, "the first connection of XMUX to close", func() bool { return server.closed.Load() == 1 })
+}

transport/internet/xdrive/config.pb.go

@@ -7,6 +7,7 @@
 package xdrive
 
 import (
+	splithttp "github.com/xtls/xray-core/transport/internet/splithttp"
 	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
 	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 	reflect "reflect"
@@ -35,6 +36,7 @@ type Config struct {
 	EagerWindowMs     uint32                 `protobuf:"varint,10,opt,name=eager_window_ms,json=eagerWindowMs,proto3" json:"eager_window_ms,omitempty"`
 	HoleTimeoutMs     uint32                 `protobuf:"varint,11,opt,name=hole_timeout_ms,json=holeTimeoutMs,proto3" json:"hole_timeout_ms,omitempty"`
 	Template          string                 `protobuf:"bytes,12,opt,name=template,proto3" json:"template,omitempty"`
+	Xmux              *splithttp.XmuxConfig  `protobuf:"bytes,13,opt,name=xmux,proto3" json:"xmux,omitempty"`
 	unknownFields     protoimpl.UnknownFields
 	sizeCache         protoimpl.SizeCache
 }
@@ -153,11 +155,18 @@ func (x *Config) GetTemplate() string {
 	return ""
 }
 
+func (x *Config) GetXmux() *splithttp.XmuxConfig {
+	if x != nil {
+		return x.Xmux
+	}
+	return nil
+}
+
 var File_transport_internet_xdrive_config_proto protoreflect.FileDescriptor
 
 const file_transport_internet_xdrive_config_proto_rawDesc = "" +
 	"\n" +
-	"&transport/internet/xdrive/config.proto\x12\x1exray.transport.internet.xdrive\"\xcb\x03\n" +
+	"&transport/internet/xdrive/config.proto\x12\x1exray.transport.internet.xdrive\x1a)transport/internet/splithttp/config.proto\"\x8e\x04\n" +
 	"\x06Config\x12#\n" +
 	"\rremote_folder\x18\x01 \x01(\tR\fremoteFolder\x12\x18\n" +
 	"\aservice\x18\x02 \x01(\tR\aservice\x12\x18\n" +
@@ -171,7 +180,8 @@ const file_transport_internet_xdrive_config_proto_rawDesc = "" +
 	"\x0feager_window_ms\x18\n" +
 	" \x01(\rR\reagerWindowMs\x12&\n" +
 	"\x0fhole_timeout_ms\x18\v \x01(\rR\rholeTimeoutMs\x12\x1a\n" +
-	"\btemplate\x18\f \x01(\tR\btemplateB5Z3github.com/xtls/xray-core/transport/internet/xdriveb\x06proto3"
+	"\btemplate\x18\f \x01(\tR\btemplate\x12A\n" +
+	"\x04xmux\x18\r \x01(\v2-.xray.transport.internet.splithttp.XmuxConfigR\x04xmuxB5Z3github.com/xtls/xray-core/transport/internet/xdriveb\x06proto3"
 
 var (
 	file_transport_internet_xdrive_config_proto_rawDescOnce sync.Once
@@ -187,14 +197,16 @@ func file_transport_internet_xdrive_config_proto_rawDescGZIP() []byte {
 
 var file_transport_internet_xdrive_config_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
 var file_transport_internet_xdrive_config_proto_goTypes = []any{
-	(*Config)(nil), // 0: xray.transport.internet.xdrive.Config
+	(*Config)(nil),               // 0: xray.transport.internet.xdrive.Config
+	(*splithttp.XmuxConfig)(nil), // 1: xray.transport.internet.splithttp.XmuxConfig
 }
 var file_transport_internet_xdrive_config_proto_depIdxs = []int32{
-	0, // [0:0] is the sub-list for method output_type
-	0, // [0:0] is the sub-list for method input_type
-	0, // [0:0] is the sub-list for extension type_name
-	0, // [0:0] is the sub-list for extension extendee
-	0, // [0:0] is the sub-list for field type_name
+	1, // 0: xray.transport.internet.xdrive.Config.xmux:type_name -> xray.transport.internet.splithttp.XmuxConfig
+	1, // [1:1] is the sub-list for method output_type
+	1, // [1:1] is the sub-list for method input_type
+	1, // [1:1] is the sub-list for extension type_name
+	1, // [1:1] is the sub-list for extension extendee
+	0, // [0:1] is the sub-list for field type_name
 }
 
 func init() { file_transport_internet_xdrive_config_proto_init() }

transport/internet/xdrive/config.proto

@@ -3,6 +3,8 @@ syntax = "proto3";
 package xray.transport.internet.xdrive;
 option go_package = "github.com/xtls/xray-core/transport/internet/xdrive";
 
+import "transport/internet/splithttp/config.proto";
+
 message Config {
   string remote_folder = 1;
   string service = 2;
@@ -16,4 +18,5 @@ message Config {
   uint32 eager_window_ms = 10;
   uint32 hole_timeout_ms = 11;
   string template = 12;
+  xray.transport.internet.splithttp.XmuxConfig xmux = 13;
 }

transport/internet/xdrive/drive.go

@@ -39,11 +39,21 @@ var (
 )
 
 type driveStorage struct {
+	*driveShared
+	// a session has a client and inflight of its own, see session
+	client   *http.Client
+	inflight chan struct{}
+	release  func()
+}
+
+// driveShared is what the sessions of a storage have in common
+type driveShared struct {
+	conns        *connPool
 	folder       string
 	clientID     string
 	clientSecret string
 	refreshToken string
-	client       *http.Client
+	tokenClient  *http.Client
 	tokenURL     string
 	filesURL     string
 	uploadURL    string
@@ -53,8 +63,6 @@ type driveStorage struct {
 	token       string
 	tokenExpiry time.Time
 
-	inflight chan struct{}
-
 	idMu sync.Mutex
 	ids  map[string]string
 }
@@ -72,21 +80,41 @@ func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config
 		}
 	}
 
+	h1Hosts := new(sync.Map)
+	client := newServiceClient(context.Background(), h1Hosts, streamSettings, driveTimeout, driveMaxInflight)
 	return &driveStorage{
-		folder:       config.RemoteFolder,
-		clientID:     config.Secrets[0],
-		clientSecret: config.Secrets[1],
-		refreshToken: config.Secrets[2],
-		client:       newServiceClient(streamSettings, driveTimeout, driveMaxInflight),
-		tokenURL:     driveTokenURL,
-		filesURL:     driveFilesURL,
-		uploadURL:    driveUploadURL,
-		backoff:      driveInitialBackoff,
-		inflight:     make(chan struct{}, driveMaxInflight),
-		ids:          make(map[string]string),
+		driveShared: &driveShared{
+			conns: newConnPool(config.Xmux, func(connCtx context.Context) *http.Client {
+				return newServiceClient(connCtx, h1Hosts, streamSettings, driveTimeout, driveMaxInflight)
+			}),
+			folder:       config.RemoteFolder,
+			clientID:     config.Secrets[0],
+			clientSecret: config.Secrets[1],
+			refreshToken: config.Secrets[2],
+			tokenClient:  client,
+			tokenURL:     driveTokenURL,
+			filesURL:     driveFilesURL,
+			uploadURL:    driveUploadURL,
+			backoff:      driveInitialBackoff,
+			ids:          make(map[string]string),
+		},
+		client:   client,
+		inflight: make(chan struct{}, driveMaxInflight),
 	}, nil
 }
 
+// session returns a storage for one session, over the connections of XMUX
+// until Close, with the token and ids of s
+func (s *driveStorage) session() Storage {
+	transport := &sessionTransport{pool: s.conns}
+	return &driveStorage{
+		driveShared: s.driveShared,
+		client:      &http.Client{Transport: transport},
+		inflight:    make(chan struct{}, driveMaxInflight),
+		release:     transport.Close,
+	}
+}
+
 func flatten(name string) string {
 	return strings.ReplaceAll(name, "/", flatSeparator)
 }
@@ -115,7 +143,7 @@ func (s *driveStorage) accessToken(ctx context.Context) (string, error) {
 	}
 	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
 
-	resp, err := s.client.Do(req)
+	resp, err := s.tokenClient.Do(req)
 	if err != nil {
 		return "", errors.New("failed to refresh the access token").Base(err)
 	}
@@ -572,5 +600,8 @@ func (s *driveStorage) List(ctx context.Context, prefix string) ([]Entry, error)
 }
 
 func (s *driveStorage) Close() error {
+	if s.release != nil {
+		s.release()
+	}
 	return nil
 }

transport/internet/xdrive/drive_test.go

@@ -562,6 +562,12 @@ func TestDriveSharedClient(t *testing.T) {
 	if first != second {
 		t.Fatal("same settings did not share one storage")
 	}
+	// a session has an inflight of its own, as large
+	session := sessionStorage(first).(*driveStorage)
+	defer session.Close()
+	if session.inflight == first.(*driveStorage).inflight || cap(session.inflight) != driveMaxInflight {
+		t.Fatalf("a session shares inflight or has cap %d", cap(session.inflight))
+	}
 }
 
 func TestDriveOldServiceName(t *testing.T) {

transport/internet/xdrive/storage.go

@@ -48,6 +48,19 @@ func newStorage(streamSettings *internet.MemoryStreamConfig) (Storage, error) {
 	}
 }
 
+// sessionStorage returns the storage of one session, which takes connections
+// of XMUX until Close, with the token and cache of storage. Local storage has
+// no connections
+func sessionStorage(storage Storage) Storage {
+	switch s := storage.(type) {
+	case *driveStorage:
+		return s.session()
+	case *templateStorage:
+		return s.session()
+	}
+	return storage
+}
+
 var (
 	sharedMu sync.Mutex
 	shared   = make(map[string]Storage)

transport/internet/xdrive/template.go

@@ -62,12 +62,20 @@ type storageTemplate struct {
 }
 
 type templateStorage struct {
-	tmpl    *storageTemplate
-	client  *http.Client
-	folder  string
-	secrets []string
-
+	*templateShared
+	// a session has a client and inflight of its own, see session
+	client   *http.Client
 	inflight chan struct{}
+	release  func()
+}
+
+// templateShared is what the sessions of a storage have in common
+type templateShared struct {
+	conns       *connPool
+	tmpl        *storageTemplate
+	tokenClient *http.Client
+	folder      string
+	secrets     []string
 
 	tokenMu     sync.Mutex
 	token       string
@@ -102,15 +110,35 @@ func newTemplateStorage(streamSettings *internet.MemoryStreamConfig, config *Con
 		conc = maxTemplateConcurrency
 	}
 
+	h1Hosts := new(sync.Map)
+	client := newServiceClient(context.Background(), h1Hosts, streamSettings, driveTimeout, conc)
 	return &templateStorage{
-		tmpl:     tmpl,
-		client:   newServiceClient(streamSettings, driveTimeout, conc),
-		folder:   config.RemoteFolder,
-		secrets:  config.Secrets,
+		templateShared: &templateShared{
+			conns: newConnPool(config.Xmux, func(connCtx context.Context) *http.Client {
+				return newServiceClient(connCtx, h1Hosts, streamSettings, driveTimeout, conc)
+			}),
+			tmpl:        tmpl,
+			tokenClient: client,
+			folder:      config.RemoteFolder,
+			secrets:     config.Secrets,
+		},
+		client:   client,
 		inflight: make(chan struct{}, conc),
 	}, nil
 }
 
+// session returns a storage for one session, over the connections of XMUX
+// until Close, with the token of s
+func (s *templateStorage) session() Storage {
+	transport := &sessionTransport{pool: s.conns}
+	return &templateStorage{
+		templateShared: s.templateShared,
+		client:         &http.Client{Transport: transport},
+		inflight:       make(chan struct{}, cap(s.inflight)),
+		release:        transport.Close,
+	}
+}
+
 const maxTemplateConcurrency = 256
 
 func (s *templateStorage) baseVars() map[string]string {
@@ -224,7 +252,7 @@ func (s *templateStorage) accessToken(ctx context.Context) (string, error) {
 	}
 	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
 
-	resp, err := s.client.Do(req)
+	resp, err := s.tokenClient.Do(req)
 	if err != nil {
 		return "", errors.New("failed to fetch the token").Base(err)
 	}
@@ -474,5 +502,8 @@ func (s *templateStorage) List(ctx context.Context, prefix string) ([]Entry, err
 }
 
 func (s *templateStorage) Close() error {
+	if s.release != nil {
+		s.release()
+	}
 	return nil
 }

transport/internet/xdrive/template_test.go

@@ -113,7 +113,7 @@ func templateSettings(store *fakeStore, auth map[string]interface{}, secrets []s
 		"delete":  map[string]interface{}{"method": "DELETE", "url": base + "/folder/{name}"},
 		"list": map[string]interface{}{
 			"method":     "PROPFIND",
-			"url":        base + "/folder/",
+			"url":        base + "/folder/?prefix={prefix}",
 			"namesRegex": `<d:href>/folder/([^<]+)</d:href>`,
 		},
 		"retry": map[string]interface{}{"status": []int{429, 500, 502, 503}},
@@ -297,6 +297,12 @@ func TestTemplateConcurrency(t *testing.T) {
 	if cap(storage.inflight) != 4 {
 		t.Fatalf("inflight cap is %d, want 4 from the template", cap(storage.inflight))
 	}
+	// and so has a session, of its own
+	session := storage.session().(*templateStorage)
+	defer session.Close()
+	if session.inflight == storage.inflight || cap(session.inflight) != 4 {
+		t.Fatalf("a session shares inflight or has cap %d, want 4", cap(session.inflight))
+	}
 }
 
 func TestTemplateConcurrencyDefault(t *testing.T) {

transport/internet/xdrive/xdrive.go

@@ -87,6 +87,7 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me
 	if err != nil {
 		return nil, err
 	}
+	storage = sessionStorage(storage)
 
 	session, err := newSessionID()
 	if err != nil {
@@ -228,8 +229,10 @@ func (l *Listener) claim(session string) bool {
 }
 
 func (l *Listener) newSessionConn(session string) *Conn {
-	return newConn(l.ctx, l.storage,
+	storage := sessionStorage(l.storage)
+	return newConn(l.ctx, storage,
 		downlinkPrefix(session), uplinkPrefix(session), l.params, func() {
+			storage.Close()
 			l.mu.Lock()
 			delete(l.active, session)
 			l.mu.Unlock()

transport/internet/xdrive/xdrive_test.go

@@ -4,16 +4,21 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	gotls "crypto/tls"
 	"io"
+	"net/http"
 	"os"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/xtls/xray-core/common/net"
 	"github.com/xtls/xray-core/transport/internet"
+	"github.com/xtls/xray-core/transport/internet/splithttp"
 	"github.com/xtls/xray-core/transport/internet/stat"
+	"github.com/xtls/xray-core/transport/internet/tls"
 )
 
 const testPatience = 30 * time.Second
@@ -581,3 +586,388 @@ func TestInlinePayload(t *testing.T) {
 		t.Fatalf("called Get %d times for an inline payload", got)
 	}
 }
+
+// runSessions opens n sessions at once, both ends in this process, passes data
+// both ways over each and closes them
+func runSessions(t *testing.T, streamSettings *internet.MemoryStreamConfig, n int) {
+	t.Helper()
+
+	accepted := make(chan stat.Connection, 1)
+	listener, err := Serve(context.Background(), net.LocalHostIP, net.Port(0), streamSettings, func(conn stat.Connection) {
+		accepted <- conn
+	})
+	if err != nil {
+		t.Fatalf("Serve: %v", err)
+	}
+	defer listener.Close()
+
+	var ends []stat.Connection
+	defer func() {
+		for _, end := range ends {
+			end.Close()
+		}
+	}()
+	for range n {
+		client, err := Dial(context.Background(), net.Destination{}, streamSettings)
+		if err != nil {
+			t.Fatalf("Dial: %v", err)
+		}
+		ends = append(ends, client)
+		select {
+		case peer := <-accepted:
+			ends = append(ends, peer)
+		case <-time.After(testPatience):
+			t.Fatal("listener did not accept the session")
+		}
+	}
+	for i := 0; i < len(ends); i += 2 {
+		if _, err := ends[i].Write([]byte("ping")); err != nil {
+			t.Fatalf("client write: %v", err)
+		}
+		expectRead(t, ends[i+1], "ping")
+		if _, err := ends[i+1].Write([]byte("pong")); err != nil {
+			t.Fatalf("server write: %v", err)
+		}
+		expectRead(t, ends[i], "pong")
+	}
+}
+
+// sessionEnd is the session and end of it that a request to the fake store
+// comes from, or "" for the listener and the token
+func sessionEnd(r *http.Request) string {
+	name := strings.TrimPrefix(r.URL.Path, "/folder/")
+	if r.Method == "PROPFIND" {
+		name = r.URL.Query().Get("prefix")
+	}
+	write := r.Method == http.MethodPut
+	parts := strings.Split(name, flatSeparator)
+	switch {
+	case len(parts) > 2 && parts[0] == streamsDir:
+		// each end writes its own direction and reads the other
+		if (parts[2] == uplinkDir) == write {
+			return parts[1] + " client"
+		}
+		return parts[1] + " server"
+	case len(parts) == 2 && parts[0] == sessionsDir && write:
+		_, session, _ := strings.Cut(parts[1], "-")
+		return session + " client"
+	}
+	return ""
+}
+
+// with "maxConcurrency" 1 every end of a session has a connection of its own,
+// with "maxConnections" the ends share that many, and the sessions after them
+// take the same ones again. The listener and the token use none of them, and
+// the sessions get the token once
+func TestXmuxConnections(t *testing.T) {
+	const n = 3
+	for _, c := range []struct {
+		name   string
+		xmux   *splithttp.XmuxConfig
+		conns  int
+		shared bool
+	}{
+		{"maxConcurrency", &splithttp.XmuxConfig{MaxConcurrency: &splithttp.RangeConfig{From: 1, To: 1}}, 2 * n, false},
+		{"maxConnections", &splithttp.XmuxConfig{MaxConnections: &splithttp.RangeConfig{From: 2, To: 2}}, 2, true},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			resetSharedStorage()
+			t.Cleanup(resetSharedStorage)
+
+			store := &fakeStore{objects: make(map[string][]byte), needAuth: "Bearer tok-fake"}
+			var (
+				mu    sync.Mutex
+				conns = make(map[string]map[string]bool)
+			)
+			server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) {
+				end := sessionEnd(r)
+				mu.Lock()
+				if conns[end] == nil {
+					conns[end] = make(map[string]bool)
+				}
+				conns[end][r.RemoteAddr] = true
+				mu.Unlock()
+				store.handle(w, r)
+			})
+			store.server = server.Server
+			settings := templateSettings(store, map[string]interface{}{
+				"type":     "oauth2",
+				"tokenUrl": server.URL + "/token",
+				"form":     map[string]interface{}{"grant_type": "refresh_token", "refresh_token": "{secret0}"},
+				"header":   map[string]interface{}{"Authorization": "Bearer {token}"},
+			}, []string{"refresh"})
+			settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings
+			settings.ProtocolSettings.(*Config).Xmux = c.xmux
+
+			runSessions(t, settings, n)
+			mu.Lock()
+			first := conns
+			conns = make(map[string]map[string]bool)
+			mu.Unlock()
+			dialed := server.conns.Load()
+			runSessions(t, settings, n)
+			if got := server.conns.Load(); got != dialed {
+				t.Errorf("the next %d sessions dialed %d connections", n, got-dialed)
+			}
+
+			owner := make(map[string]string)
+			for end, addrs := range first {
+				if end == "" {
+					continue
+				}
+				if len(addrs) != 1 {
+					t.Errorf("%s used %d connections", end, len(addrs))
+				}
+				for addr := range addrs {
+					if owner[addr] != "" && !c.shared {
+						t.Errorf("%s and %s shared a connection", end, owner[addr])
+					}
+					owner[addr] = end
+				}
+			}
+			if len(owner) != c.conns {
+				t.Errorf("%d sessions had %d connections, want %d", 2*n, len(owner), c.conns)
+			}
+			for addr := range first[""] {
+				if owner[addr] != "" {
+					t.Errorf("the listener or the token went over the connection of %s", owner[addr])
+				}
+			}
+			store.mu.Lock()
+			defer store.mu.Unlock()
+			if store.tokens != 1 {
+				t.Errorf("the sessions got the token %d times", store.tokens)
+			}
+		})
+	}
+}
+
+func TestDriveXmuxConnections(t *testing.T) {
+	drive := newFakeDrive(t)
+	server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, drive.server.Config.Handler.ServeHTTP)
+	driveTokenURL = server.URL + "/token"
+	driveFilesURL = server.URL + "/files"
+	driveUploadURL = server.URL + "/upload"
+	settings := driveSettings()
+	settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings
+	settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{MaxConnections: &splithttp.RangeConfig{From: 3, To: 3}}
+
+	runSessions(t, settings, 3)
+	if got := server.conns.Load(); got != 4 {
+		t.Errorf("6 ends of sessions with 3 connections and the listener took %d connections", got)
+	}
+
+	drive.mu.Lock()
+	defer drive.mu.Unlock()
+	if drive.tokens != 1 {
+		t.Errorf("6 ends of sessions got the token %d times", drive.tokens)
+	}
+}
+
+// a session knows the hosts that answered the storage over h1, the polling of
+// the listener for one, so it sends requests in a row over one connection, not
+// first over one for h2 that then stays idle, and parallel ones over more
+func TestSessionKnowsHTTP1Hosts(t *testing.T) {
+	check := func(name string, server *tlsServer, storage Storage) {
+		put := func(s Storage, names ...string) {
+			for _, n := range names {
+				if err := s.Put(context.Background(), "streams/abc/c2s/"+n, []byte("x")); err != nil {
+					t.Errorf("%s: Put: %v", name, err)
+				}
+			}
+		}
+		put(storage, "a", "b")
+		before := server.conns.Load()
+		session := sessionStorage(storage)
+		defer session.Close()
+		put(session, "c", "d")
+		if n := server.conns.Load() - before; n != 1 {
+			t.Errorf("%s: 2 requests in a row of a session over h1 took %d connections", name, n)
+		}
+
+		before = server.conns.Load()
+		var wg sync.WaitGroup
+		for _, n := range []string{"e", "f", "g", "h"} {
+			wg.Go(func() { put(session, n) })
+		}
+		wg.Wait()
+		if server.conns.Load() == before {
+			t.Errorf("%s: 4 parallel requests of a session over h1 took one connection", name)
+		}
+	}
+	slow := func(next http.HandlerFunc) http.HandlerFunc {
+		return func(w http.ResponseWriter, r *http.Request) {
+			time.Sleep(20 * time.Millisecond)
+			next(w, r)
+		}
+	}
+
+	store := &fakeStore{objects: make(map[string][]byte)}
+	server := newTLSServer(t, &gotls.Config{NextProtos: []string{"http/1.1"}}, slow(store.handle))
+	store.server = server.Server
+	settings := templateSettings(store, map[string]interface{}{"type": "none"}, nil)
+	settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings
+	template, err := newTemplateStorage(settings, settings.ProtocolSettings.(*Config))
+	if err != nil {
+		t.Fatalf("newTemplateStorage: %v", err)
+	}
+	check("template", server, template)
+
+	drive := newFakeDrive(t)
+	server = newTLSServer(t, &gotls.Config{NextProtos: []string{"http/1.1"}}, slow(drive.server.Config.Handler.ServeHTTP))
+	driveTokenURL = server.URL + "/token"
+	driveFilesURL = server.URL + "/files"
+	settings = driveSettings()
+	settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings
+	google, err := newDriveStorage(settings, settings.ProtocolSettings.(*Config))
+	if err != nil {
+		t.Fatalf("newDriveStorage: %v", err)
+	}
+	check("drive", server, google)
+}
+
+// with "hMaxRequestTimes" a session moves on to a new connection after that
+// many requests, and the old one closes once the requests on it end, not before
+func TestXmuxRequestLimit(t *testing.T) {
+	store := &fakeStore{objects: make(map[string][]byte)}
+	arrived := make(chan struct{}, 1)
+	release := make(chan struct{})
+	server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodGet {
+			arrived <- struct{}{}
+			<-release
+		}
+		store.handle(w, r)
+	})
+	store.server = server.Server
+	settings := templateSettings(store, map[string]interface{}{"type": "none"}, nil)
+	settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings
+	settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{HMaxRequestTimes: &splithttp.RangeConfig{From: 3, To: 3}}
+	storage, err := newTemplateStorage(settings, settings.ProtocolSettings.(*Config))
+	if err != nil {
+		t.Fatalf("newTemplateStorage: %v", err)
+	}
+	session := sessionStorage(storage)
+	defer session.Close()
+	put := func(names ...string) {
+		for _, name := range names {
+			if err := session.Put(context.Background(), "streams/abc/c2s/"+name, []byte("x")); err != nil {
+				t.Fatalf("Put: %v", err)
+			}
+		}
+	}
+
+	put("a")
+	got := make(chan error, 1)
+	go func() {
+		_, err := session.Get(context.Background(), "streams/abc/c2s/a")
+		got <- err
+	}()
+	<-arrived
+	put("b", "c")
+	if n := server.conns.Load(); n != 2 {
+		t.Errorf("4 requests with 3 to a connection took %d connections", n)
+	}
+	if server.closed.Load() != 0 {
+		t.Error("the first connection closed under a request")
+	}
+	close(release)
+	if err := <-got; err != nil {
+		t.Errorf("Get on the first connection: %v", err)
+	}
+	waitFor(t, "the first connection to close", func() bool { return server.closed.Load() == 1 })
+
+	put("d", "e", "f")
+	if n := server.conns.Load(); n != 3 {
+		t.Errorf("7 requests with 3 to a connection took %d connections", n)
+	}
+	waitFor(t, "the second connection to close", func() bool { return server.closed.Load() == 2 })
+}
+
+// a session that fails to announce itself lets go of its connection, which the
+// next one takes with "maxConcurrency" 1
+func TestFailedDialReleasesConnection(t *testing.T) {
+	resetSharedStorage()
+	t.Cleanup(resetSharedStorage)
+
+	store := &fakeStore{objects: make(map[string][]byte)}
+	var refuse atomic.Bool
+	refuse.Store(true)
+	server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) {
+		if refuse.Load() {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		store.handle(w, r)
+	})
+	store.server = server.Server
+	settings := templateSettings(store, map[string]interface{}{"type": "none"}, nil)
+	settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings
+	settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{MaxConcurrency: &splithttp.RangeConfig{From: 1, To: 1}}
+
+	if _, err := Dial(context.Background(), net.Destination{}, settings); err == nil {
+		t.Fatal("Dial announced a session that the service refused")
+	}
+	refuse.Store(false)
+	conn, err := Dial(context.Background(), net.Destination{}, settings)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	defer conn.Close()
+	if n := server.conns.Load(); n != 1 {
+		t.Errorf("a failed Dial and the next took %d connections", n)
+	}
+}
+
+// a session gets the token over a connection of the storage, not over one of
+// XMUX
+func TestSessionTokenConnection(t *testing.T) {
+	check := func(name string, server *tlsServer, storage Storage) {
+		session := sessionStorage(storage)
+		defer session.Close()
+		if err := session.Put(context.Background(), "sessions/abc", nil); err != nil {
+			t.Fatalf("%s: Put: %v", name, err)
+		}
+		if n := server.conns.Load(); n != 2 {
+			t.Errorf("%s: the token and a Put of a session took %d connections, want 2", name, n)
+		}
+	}
+
+	store := &fakeStore{objects: make(map[string][]byte), needAuth: "Bearer tok-fake"}
+	server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, store.handle)
+	store.server = server.Server
+	settings := templateSettings(store, map[string]interface{}{
+		"type":     "oauth2",
+		"tokenUrl": server.URL + "/token",
+		"header":   map[string]interface{}{"Authorization": "Bearer {token}"},
+	}, nil)
+	settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings
+	template, err := newTemplateStorage(settings, settings.ProtocolSettings.(*Config))
+	if err != nil {
+		t.Fatalf("newTemplateStorage: %v", err)
+	}
+	check("template", server, template)
+
+	drive := newFakeDrive(t)
+	server = newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, drive.server.Config.Handler.ServeHTTP)
+	driveTokenURL = server.URL + "/token"
+	driveFilesURL = server.URL + "/files"
+	settings = driveSettings()
+	settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings
+	google, err := newDriveStorage(settings, settings.ProtocolSettings.(*Config))
+	if err != nil {
+		t.Fatalf("newDriveStorage: %v", err)
+	}
+	check("drive", server, google)
+}
+
+// local storage has no connections to keep apart
+func TestLocalSessionStorage(t *testing.T) {
+	storage, err := newLocalStorage(t.TempDir())
+	if err != nil {
+		t.Fatalf("newLocalStorage: %v", err)
+	}
+	if sessionStorage(storage) != Storage(storage) {
+		t.Fatal("a session of local storage got a storage of its own")
+	}
+}