TUN inbound: Only log IPv6 setup errors on Windows when DisabledComponents turns IPv6 off, with "misconfigtun" blocking IPv6 then

https://github.com/XTLS/Xray-core/pull/7341#issuecomment-6700092032

IPv6 entries are no longer left out. With IPv6 turned off by
DisabledComponents (0x10, the non-tunnel interfaces, the TUN among them,
as in 0xFF), IPv6 is set up as usual, and errors doing so, clearing its
DNS servers included, are only logged; otherwise they exit, also for a
config without IPv6 entries, which replaces the skips of #6874, #7325 and
#7332. "misconfigtun" then blocks IPv6 outside the TUN whatever the routes,
as IPv6 still works until the reboot that applies the value.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.github/docker/Dockerfile

@@ -9,8 +9,8 @@ ARG TARGETARCH
 RUN GOOS=$TARGETOS GOARCH=$TARGETARCH CGO_ENABLED=0 go build -o xray -trimpath -buildvcs=false -gcflags="all=-l=4" -ldflags "-s -w -buildid=" ./main
 
 # Download geodat into a staging directory
-ADD https://raw.githubusercontent.com/XTLS/Xray-rules-dat/release/geoip/geoip-full.dat /tmp/geodat/geoip.dat
-ADD https://raw.githubusercontent.com/XTLS/Xray-rules-dat/release/geosite/geosite-expand.dat /tmp/geodat/geosite.dat
+ADD https://github.com/XTLS/Xray-rules-dat/releases/latest/download/geoip-full.dat /tmp/geodat/geoip.dat
+ADD https://github.com/XTLS/Xray-rules-dat/releases/latest/download/geosite-expand.dat /tmp/geodat/geosite.dat
 
 RUN mkdir -p /tmp/empty
 

.github/docker/Dockerfile.usa

@@ -9,8 +9,8 @@ ARG TARGETARCH
 RUN GOOS=$TARGETOS GOARCH=$TARGETARCH CGO_ENABLED=0 go build -o xray -trimpath -buildvcs=false -gcflags="all=-l=4" -ldflags "-s -w -buildid=" ./main
 
 # Download geodat into a staging directory
-ADD https://raw.githubusercontent.com/XTLS/Xray-rules-dat/release/geoip/geoip-full.dat /tmp/geodat/geoip.dat
-ADD https://raw.githubusercontent.com/XTLS/Xray-rules-dat/release/geosite/geosite-expand.dat /tmp/geodat/geosite.dat
+ADD https://github.com/XTLS/Xray-rules-dat/releases/latest/download/geoip-full.dat /tmp/geodat/geoip.dat
+ADD https://github.com/XTLS/Xray-rules-dat/releases/latest/download/geosite-expand.dat /tmp/geodat/geosite.dat
 
 RUN mkdir -p /tmp/empty
 

app/proxyman/inbound/inbound.go

@@ -51,7 +51,17 @@ func (m *Manager) AddHandler(ctx context.Context, handler inbound.Handler) error
 	}
 
 	if m.running {
-		return handler.Start()
+		if err := handler.Start(); err != nil {
+			if len(tag) > 0 {
+				delete(m.taggedHandlers, tag)
+			} else {
+				m.untaggedHandlers = m.untaggedHandlers[:len(m.untaggedHandlers)-1]
+			}
+			if err := handler.Close(); err != nil {
+				errors.LogWarningInner(ctx, err, "failed to close handler ", tag)
+			}
+			return err
+		}
 	}
 
 	return nil

app/proxyman/outbound/handler.go

@@ -122,8 +122,8 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
 			if config.Concurrency < 0 {
 				h.mux = &mux.ClientManager{Enabled: false}
 			}
-			if config.Concurrency == 0 {
-				config.Concurrency = 8 // same as before
+			if config.Concurrency >= 0 && config.Concurrency <= 128 {
+				config.Concurrency = 128
 			}
 			if config.Concurrency > 0 {
 				h.mux = &mux.ClientManager{
@@ -134,7 +134,7 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
 							Dialer: h,
 							Strategy: mux.ClientStrategy{
 								MaxConcurrency: uint32(config.Concurrency),
-								MaxConnection:  128,
+								MaxConnection:  1024,
 							},
 						},
 					},
@@ -155,7 +155,7 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
 							Dialer: h,
 							Strategy: mux.ClientStrategy{
 								MaxConcurrency: uint32(config.XudpConcurrency),
-								MaxConnection:  128,
+								MaxConnection:  1024,
 							},
 						},
 					},

infra/conf/geodata.go

@@ -2,6 +2,7 @@ package conf
 
 import (
 	"net/url"
+	"strings"
 
 	"github.com/robfig/cron/v3"
 	"github.com/xtls/xray-core/app/geodata"
@@ -19,11 +20,20 @@ func (c *GeodataAssetConfig) Build() (*geodata.Asset, error) {
 		return nil, errors.New("invalid geodata asset url: ", c.URL).Base(err)
 	}
 	return &geodata.Asset{
-		Url:  c.URL,
+		Url:  migrateGeodataURL(c.URL),
 		File: c.File,
 	}, nil
 }
 
+func migrateGeodataURL(assetURL string) string {
+	if strings.HasSuffix(strings.ToLower(assetURL), ".dat") && (strings.HasPrefix(strings.ToLower(assetURL), "https://github.com/xtls/xray-rules-dat/") ||
+		strings.HasPrefix(strings.ToLower(assetURL), "https://raw.githubusercontent.com/xtls/xray-rules-dat/")) {
+		errors.PrintDeprecatedFeatureWarning(assetURL, assetURL+".gz")
+		return assetURL + ".gz"
+	}
+	return assetURL
+}
+
 func validateHTTPS(s string) error {
 	u, err := url.ParseRequestURI(s)
 	if err != nil {

proxy/proxy.go

@@ -13,6 +13,7 @@ import (
 	"math/big"
 	"runtime"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/pires/go-proxyproto"
@@ -301,6 +302,7 @@ type VisionWriter struct {
 	// internal
 	writeOnceUserUUID  []byte
 	directWriteCounter stats.Counter
+	access             sync.Mutex
 
 	testseed []uint32
 }
@@ -324,6 +326,9 @@ func NewVisionWriter(writer buf.Writer, trafficState *TrafficState, isUplink boo
 }
 
 func (w *VisionWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	w.access.Lock()
+	defer w.access.Unlock()
+
 	var isPadding *bool
 	var switchToDirectCopy *bool
 	var spliceReadyInbound *session.Inbound

proxy/tun/README.md

@@ -223,16 +223,16 @@ To make it start, wintun.dll specific for your Windows/arch must be present next
 
 After the start network adapter with the name you chose in the config will be created in the system, and exist while Xray is running. A Wintun adapter that already has that name is reused.
 
-With IPv6 disabled system-wide (the `DisabledComponents` registry value), the adapter has no IPv6, and the TUN starts with IPv4 only, unless `gateway` or `autoSystemRoutingTable` has IPv6 entries, which then keep it from starting.
+With IPv6 turned off by the `DisabledComponents` registry value (0x10 set, as in 0xFF), errors setting up IPv6 for `gateway`, `dns` and `autoSystemRoutingTable`, clearing the IPv6 DNS servers when `dns` has none included, are only logged. Otherwise, they keep the TUN from starting.
 
 When `dns` is set, those servers are applied to the adapter. Windows is kept from registering the TUN's addresses in DNS, and its DNS cache is flushed when the TUN starts and stops. Xray's own lookups through the system's DNS servers, like those of the `localhost` DNS server with `autoOutboundsInterface`, skip them, unless another interface uses them as well, as they would lead back into the TUN.
 
-Setting `autoSystemWfpBlockLeak` to at least `["dns"]` is recommended, to block DNS leaks. It needs `autoSystemRoutingTable` (the config is rejected otherwise), and adds Windows Filtering Platform filters that let traffic of every program but Xray itself leave only through the TUN, each kind chosen by a value:
+Setting `autoSystemWfpBlockLeak` to at least `["dns", "misconfigtun"]` is recommended, to block DNS leaks and the traffic of an IP version without routes. It needs `autoSystemRoutingTable` (the config is rejected otherwise), and adds Windows Filtering Platform filters that let traffic of every program but Xray itself leave only through the TUN, each kind chosen by a value:
 - `"dns"`: stands for `"53"`, `"853"` (DNS over TLS and QUIC) and `"windnsclient"`. Windows also queries the DNS servers of the other interfaces, through those interfaces whatever the routes say, and other programs reach a resolver on the local network (e.g. `192.168.1.1` from DHCP) through its more specific route.
 - `"windnsclient"`: Windows' DNS Client service, except for name resolution on the local network (LLMNR, mDNS). Since Windows 11 and Server 2022, it may query over HTTPS or TLS, to any port; earlier versions only query port 53, so there it does nothing.
 - `"tcp"`, `"udp"`: that protocol, to all ports, or with `/<port>` or `/<low>-<high>` (1 to 65535) to some, e.g. `"tcp/853"` or `"udp/10-20"`. The ports alone stand for both, e.g. `"53"` or `"10-20"`.
 - `"icmp"`: ICMP and ICMPv6, e.g. `ping`.
-- `"misconfigtun"`: an IP version without routes in `autoSystemRoutingTable`, which would bypass the TUN.
+- `"misconfigtun"`: an IP version without routes in `autoSystemRoutingTable`, which would bypass the TUN. With IPv6 turned off by `DisabledComponents`, it covers IPv6 whatever the routes, as IPv6 setup errors are then only logged, and IPv6 still works until the reboot that applies the value.
 - `"all"`: everything, in both directions. Access to the local network then breaks unless it is routed into the TUN as well, and so do incoming connections to other programs, other VPNs, and virtual machines whose NAT runs on the host.
 - `"allexceptlan"`: like `"all"`, but traffic to and from the local network stays allowed, e.g. for printers or casting to a TV: private, link-local, multicast and broadcast addresses (`10.0.0.0/8`, `172.16.0.0/12`, `192.168.0.0/16`, `169.254.0.0/16`, `224.0.0.0/4`, `255.255.255.255`, `fc00::/7`, `fe80::/10`, `ff00::/8`), not global IPv6 ones. It includes `"dns"`, and the other values apply there as well. `"all"` overrides it.
 

proxy/tun/tun_windows.go

@@ -23,6 +23,7 @@ import (
 	"github.com/xtls/xray-core/common/errors"
 	"github.com/xtls/xray-core/transport/internet"
 	"golang.org/x/sys/windows"
+	"golang.org/x/sys/windows/registry"
 	"golang.zx2c4.com/wintun"
 	"golang.zx2c4.com/wireguard/windows/tunnel/winipcfg"
 )
@@ -98,6 +99,19 @@ func open(name, desc string) (*wintun.Adapter, error) {
 	return nil, err
 }
 
+// ipv6Off tells whether DisabledComponents turns IPv6 off on the non-tunnel
+// interfaces (0x10, as 0xFF does), the TUN among them. Windows applies it at
+// boot only.
+func ipv6Off() bool {
+	key, err := registry.OpenKey(registry.LOCAL_MACHINE, `SYSTEM\CurrentControlSet\Services\Tcpip6\Parameters`, registry.QUERY_VALUE)
+	if err != nil {
+		return false
+	}
+	defer key.Close()
+	value, _, err := key.GetIntegerValue("DisabledComponents")
+	return err == nil && value&0x10 != 0
+}
+
 func (t *WindowsTun) Start() (err error) {
 	var address4, address6 bool
 	addresses := make([]netip.Prefix, 0, len(t.options.Gateway))
@@ -139,6 +153,17 @@ func (t *WindowsTun) Start() (err error) {
 		routesData = append(routesData, &r)
 	}
 
+	// With IPv6 off by DisabledComponents, setting it up may fail, clearing
+	// its DNS servers too, which is then only logged.
+	off6 := ipv6Off()
+	ignored := func(family winipcfg.AddressFamily, err error) bool {
+		if family != windows.AF_INET6 || !off6 {
+			return false
+		}
+		errors.LogErrorInner(context.Background(), err, "[tun] ignored, as DisabledComponents turns IPv6 off")
+		return true
+	}
+
 	var retryTimes int
 	var firstErr error
 startOver:
@@ -155,6 +180,9 @@ startOver:
 			err = t.luid.SetRoutesForFamily(family, routesData)
 			if err != nil {
 				firstErr = errors.New("unable to set routes").Base(err)
+				if ignored(family, firstErr) {
+					continue
+				}
 				if err == windows.ERROR_NOT_FOUND {
 					goto startOver
 				}
@@ -165,6 +193,9 @@ startOver:
 			err = t.luid.SetIPAddressesForFamily(family, addresses)
 			if err != nil {
 				firstErr = errors.New("unable to set ips").Base(err)
+				if ignored(family, firstErr) {
+					continue
+				}
 				if err == windows.ERROR_NOT_FOUND {
 					goto startOver
 				}
@@ -173,9 +204,7 @@ startOver:
 		}
 		ipif, err := t.luid.IPInterface(family)
 		if err != nil {
-			// With IPv6 disabled system-wide (DisabledComponents), the adapter has no
-			// IPv6 interface at all. Skip the family unless the config asks for it.
-			if err == windows.ERROR_NOT_FOUND && family == windows.AF_INET6 && !address6 && !route6 {
+			if ignored(family, err) {
 				continue
 			}
 			return err
@@ -194,6 +223,9 @@ startOver:
 		err = ipif.Set()
 		if err != nil {
 			firstErr = errors.New("unable to set metric and MTU").Base(err)
+			if ignored(family, firstErr) {
+				continue
+			}
 			if err == windows.ERROR_NOT_FOUND {
 				goto startOver
 			}
@@ -201,13 +233,10 @@ startOver:
 		}
 		err = t.luid.SetDNS(family, dns, nil)
 		if err != nil {
-			// Skip IPv6 the config does not ask for also when setting its DNS
-			// servers fails, as reported on Windows 11 24H2 with IPv6 disabled
-			// (Access is denied).
-			if family == windows.AF_INET6 && !address6 && !route6 {
+			firstErr = errors.New("unable to set DNS").Base(err)
+			if ignored(family, firstErr) {
 				continue
 			}
-			firstErr = errors.New("unable to set DNS").Base(err)
 			if err == windows.ERROR_NOT_FOUND {
 				goto startOver
 			}
@@ -264,7 +293,10 @@ startOver:
 	if err != nil {
 		return err
 	}
-	block := newLeakFilters(leaks, route4, route6)
+	// With IPv6 off by DisabledComponents, IPv6 setup failures are only logged,
+	// and IPv6 still works until the reboot that applies it. So "misconfigtun"
+	// then blocks IPv6 outside the TUN whatever the routes.
+	block := newLeakFilters(leaks, route4, route6 && !off6)
 	if route4 || route6 {
 		dnsClient := func(l leak) bool { return l.name == "windnsclient" }
 		if slices.ContainsFunc(block, dnsClient) && !dnsClientUsesAnyPort() {

transport/internet/xdrive/client.go

@@ -438,9 +438,11 @@ func (b *doneBody) Close() error {
 type sessionTransport struct {
 	pool *connPool
 
-	mu     sync.Mutex
-	client *splithttp.XmuxClient
-	closed bool
+	mu                sync.Mutex
+	client            *splithttp.XmuxClient
+	closed            bool
+	slowAt, retiredAt time.Time
+	movedAt           time.Time
 }
 
 func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
@@ -450,7 +452,7 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
 		return nil, errSessionClosed
 	}
 	client := t.client
-	if client == nil || client.LeftRequests.Add(-1) < 0 || !client.UnreusableAt.IsZero() && time.Now().After(client.UnreusableAt) {
+	if client == nil || client.XmuxConn.IsClosed() || 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()
@@ -458,6 +460,7 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
 		t.client.LeftRequests.Add(-1)
 		if client != nil {
 			client.DoneRunning()
+			t.movedAt = time.Now()
 		}
 		client = t.client
 	}
@@ -471,6 +474,22 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
 	return conn.roundTrip(r)
 }
 
+func (t *sessionTransport) stalled(began time.Time) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if began.Before(t.movedAt) {
+		return
+	}
+	now := time.Now()
+	slow := now.Sub(t.slowAt) < 3*time.Second
+	t.slowAt = now
+	if !slow || t.client == nil || now.Sub(t.retiredAt) < 2*time.Second {
+		return
+	}
+	t.retiredAt = now
+	t.client.XmuxConn.(*poolConn).Close()
+}
+
 func (t *sessionTransport) Close() {
 	t.mu.Lock()
 	defer t.mu.Unlock()

transport/internet/xdrive/client_test.go

@@ -987,3 +987,103 @@ func TestDriveXmuxOverH3(t *testing.T) {
 	}
 	waitFor(t, "the first connection of XMUX to close", func() bool { return server.closed.Load() == 1 })
 }
+
+type stallConn struct{}
+
+func (stallConn) RoundTrip(r *http.Request) (*http.Response, error) {
+	<-r.Context().Done()
+	return nil, r.Context().Err()
+}
+
+type okConn struct{}
+
+func (okConn) RoundTrip(r *http.Request) (*http.Response, error) {
+	return &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     make(http.Header),
+		Body:       io.NopCloser(strings.NewReader(`{"id":"file-1"}`)),
+	}, nil
+}
+
+func TestSessionRetiresStalledConn(t *testing.T) {
+	made := atomic.Int32{}
+	transport := &sessionTransport{pool: newConnPool(nil, func(connCtx context.Context) *http.Client {
+		if made.Add(1) == 1 {
+			return &http.Client{Transport: stallConn{}}
+		}
+		return &http.Client{Transport: okConn{}}
+	})}
+	storage := &driveStorage{
+		driveShared: &driveShared{
+			token:       "token",
+			tokenExpiry: time.Now().Add(time.Hour),
+			backoff:     time.Millisecond,
+		},
+		client:   &http.Client{Transport: transport},
+		inflight: make(chan struct{}, 1),
+		reserved: make(chan struct{}, 1),
+	}
+
+	ctx := context.WithValue(context.Background(), attemptKey{}, 100*time.Millisecond)
+	code, body, err := storage.do(ctx, http.MethodPost, "http://drive.invalid/files", "", nil)
+	if err != nil {
+		t.Fatalf("the first request over a stalled connection: %v", err)
+	}
+	if code != http.StatusOK || string(body) != `{"id":"file-1"}` {
+		t.Fatalf("the stalled connection gave %d %q", code, body)
+	}
+	if n := made.Load(); n != 2 {
+		t.Fatalf("a stalled connection took %d connections, want it retired and one made after it", n)
+	}
+	if _, _, err := storage.do(ctx, http.MethodPost, "http://drive.invalid/files", "", nil); err != nil {
+		t.Fatalf("the request after the retirement: %v", err)
+	}
+	if n := made.Load(); n != 2 {
+		t.Fatalf("%d connections were made, want the healthy one kept", n)
+	}
+}
+
+func TestSessionIgnoresStallOfRetiredConn(t *testing.T) {
+	var made atomic.Int32
+	transport := &sessionTransport{pool: newConnPool(nil, func(connCtx context.Context) *http.Client {
+		made.Add(1)
+		return &http.Client{Transport: okConn{}}
+	})}
+	send := func() {
+		req, err := http.NewRequest(http.MethodGet, "http://drive.invalid/files", nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp, err := transport.RoundTrip(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp.Body.Close()
+	}
+	send()
+	old := time.Now()
+	time.Sleep(2 * time.Millisecond)
+	transport.stalled(old)
+	transport.stalled(old)
+	send()
+	if n := made.Load(); n != 2 {
+		t.Fatalf("a stalled connection took %d connections, want it retired and one made after it", n)
+	}
+	transport.mu.Lock()
+	transport.slowAt, transport.retiredAt = time.Time{}, time.Time{}
+	transport.mu.Unlock()
+	transport.stalled(old)
+	transport.stalled(old)
+	send()
+	if n := made.Load(); n != 2 {
+		t.Fatalf("the stall of a retired connection took the healthy one too, %d connections", n)
+	}
+	time.Sleep(2 * time.Millisecond)
+	fresh := time.Now()
+	transport.stalled(fresh)
+	transport.stalled(fresh)
+	send()
+	if n := made.Load(); n != 3 {
+		t.Fatalf("a stall of the current connection made %d connections, want it retired", n)
+	}
+}

transport/internet/xdrive/drive.go

@@ -21,11 +21,12 @@ import (
 const (
 	flatSeparator = "~"
 
-	driveBoundary    = "xdrive-boundary"
-	drivePageSize    = 1000
-	driveMaxAttempts = 8
-	driveMaxInflight = 32
-	driveInlineLimit = 12000
+	driveBoundary         = "xdrive-boundary"
+	drivePageSize         = 1000
+	driveMaxAttempts      = 8
+	driveMaxInflight      = 64
+	driveReservedInflight = 8
+	driveInlineLimit      = 12000
 	// driveTimeout ends a request that makes no progress for that long
 	driveTimeout = 60 * time.Second
 )
@@ -36,13 +37,21 @@ var (
 	driveFilesURL       = "https://www.googleapis.com/drive/v3/files"
 	driveUploadURL      = "https://www.googleapis.com/upload/drive/v3/files?uploadType=multipart&fields=id,name"
 	driveInitialBackoff = 200 * time.Millisecond
+	driveCooldown       = 2 * time.Second
+	driveSlotRead       = 3 * time.Second
+	driveSlotWrite      = 5 * time.Second
 )
 
+type laneKey struct{}
+
+type attemptKey struct{}
+
 type driveStorage struct {
 	*driveShared
 	// a session has a client and inflight of its own, see session
 	client   *http.Client
 	inflight chan struct{}
+	reserved chan struct{}
 	release  func()
 }
 
@@ -65,6 +74,9 @@ type driveShared struct {
 
 	idMu sync.Mutex
 	ids  map[string]string
+
+	coolMu    sync.Mutex
+	coolUntil time.Time
 }
 
 func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config) (*driveStorage, error) {
@@ -100,6 +112,7 @@ func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config
 		},
 		client:   client,
 		inflight: make(chan struct{}, driveMaxInflight),
+		reserved: make(chan struct{}, driveReservedInflight),
 	}, nil
 }
 
@@ -111,6 +124,7 @@ func (s *driveStorage) session() Storage {
 		driveShared: s.driveShared,
 		client:      &http.Client{Transport: transport},
 		inflight:    make(chan struct{}, driveMaxInflight),
+		reserved:    make(chan struct{}, driveReservedInflight),
 		release:     transport.Close,
 	}
 }
@@ -219,6 +233,17 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
 	backoff := s.backoff
 	var lastErr error
 
+	s.coolMu.Lock()
+	until := s.coolUntil
+	s.coolMu.Unlock()
+	if d := time.Until(until); d > 0 {
+		select {
+		case <-ctx.Done():
+			return 0, nil, ctx.Err()
+		case <-time.After(d):
+		}
+	}
+
 	for attempt := 0; attempt < driveMaxAttempts; attempt++ {
 		if attempt > 0 {
 			select {
@@ -238,8 +263,12 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
 			continue
 		}
 
+		pool := s.inflight
+		if ctx.Value(laneKey{}) != nil {
+			pool = s.reserved
+		}
 		select {
-		case s.inflight <- struct{}{}:
+		case pool <- struct{}{}:
 		case <-ctx.Done():
 			return 0, nil, ctx.Err()
 		}
@@ -248,8 +277,14 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
 		if body != nil {
 			reader = bytes.NewReader(body)
 		}
-		req, err := http.NewRequestWithContext(ctx, method, target, reader)
+		attemptCtx, cancel := ctx, context.CancelFunc(func() {})
+		if d, ok := ctx.Value(attemptKey{}).(time.Duration); ok {
+			attemptCtx, cancel = context.WithTimeout(ctx, d)
+		}
+		req, err := http.NewRequestWithContext(attemptCtx, method, target, reader)
 		if err != nil {
+			cancel()
+			<-pool
 			return 0, nil, errors.New("failed to build a Drive request").Base(err)
 		}
 		req.Header.Set("Authorization", "Bearer "+token)
@@ -257,9 +292,17 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
 			req.Header.Set("Content-Type", contentType)
 		}
 
+		began := time.Now()
 		resp, err := s.client.Do(req)
-		<-s.inflight
+		<-pool
 		if err != nil {
+			timedOut := attemptCtx.Err() == context.DeadlineExceeded && ctx.Err() == nil
+			cancel()
+			if timedOut {
+				if t, ok := s.client.Transport.(*sessionTransport); ok {
+					t.stalled(began)
+				}
+			}
 			if ctx.Err() != nil {
 				return 0, nil, ctx.Err()
 			}
@@ -270,6 +313,7 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
 		}
 		payload, err := io.ReadAll(resp.Body)
 		resp.Body.Close()
+		cancel()
 		if err != nil {
 			if ctx.Err() != nil {
 				return 0, nil, ctx.Err()
@@ -286,13 +330,19 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
 		if resp.StatusCode == http.StatusForbidden && rateLimited(payload) {
 			lastErr = errors.New("Drive is rate limiting: ", string(payload))
 			errors.LogWarning(ctx, "rate limited by Drive, attempt ",
-				attempt+1, " of ", driveMaxAttempts)
+				attempt+1, " of ", driveMaxAttempts, ", ", method, " ", len(body),
+				" bytes, body read ", strings.Contains(target, "alt=media"),
+				", lane ", ctx.Value(laneKey{}) != nil, ": ", string(payload[:min(len(payload), 300)]))
+			s.cool()
 			continue
 		}
 		if retryableStatus(resp.StatusCode) {
 			lastErr = errors.New("Drive answered ", resp.StatusCode, ": ", string(payload))
 			errors.LogWarning(ctx, "retrying after Drive answered ", resp.StatusCode,
 				", attempt ", attempt+1, " of ", driveMaxAttempts)
+			if resp.StatusCode == http.StatusTooManyRequests {
+				s.cool()
+			}
 			continue
 		}
 		return resp.StatusCode, payload, nil
@@ -301,6 +351,12 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
 	return 0, nil, lastErr
 }
 
+func (s *driveStorage) cool() {
+	s.coolMu.Lock()
+	s.coolUntil = time.Now().Add(jitter(driveCooldown))
+	s.coolMu.Unlock()
+}
+
 func (s *driveStorage) invalidateToken() {
 	s.tokenMu.Lock()
 	s.token = ""
@@ -542,12 +598,27 @@ func (s *driveStorage) Delete(ctx context.Context, name string) error {
 	if err != nil {
 		return err
 	}
+	sem := make(chan struct{}, 8)
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	var first error
 	for childName, file := range children {
-		if err := s.deleteID(ctx, childName, file.id); err != nil {
-			return err
-		}
+		wg.Add(1)
+		sem <- struct{}{}
+		go func(name, id string) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			if err := s.deleteID(ctx, name, id); err != nil {
+				mu.Lock()
+				if first == nil {
+					first = err
+				}
+				mu.Unlock()
+			}
+		}(childName, file.id)
 	}
-	return nil
+	wg.Wait()
+	return first
 }
 
 func (s *driveStorage) List(ctx context.Context, prefix string) ([]Entry, error) {
@@ -604,6 +675,9 @@ func (s *driveStorage) makeSlot(ctx context.Context, name string) (string, error
 
 // setSlot implements slotStore, a description and a body go in one request
 func (s *driveStorage) setSlot(ctx context.Context, id, description string, body []byte) error {
+	if body == nil {
+		ctx = context.WithValue(ctx, attemptKey{}, driveSlotWrite)
+	}
 	metadata, err := json.Marshal(map[string]string{"description": description})
 	if err != nil {
 		return errors.New("failed to build the slot metadata").Base(err)
@@ -631,6 +705,7 @@ func (s *driveStorage) setSlot(ctx context.Context, id, description string, body
 
 // readDescription implements slotStore
 func (s *driveStorage) readDescription(ctx context.Context, id string) (string, error) {
+	ctx = context.WithValue(ctx, attemptKey{}, driveSlotRead)
 	status, payload, err := s.do(ctx, http.MethodGet,
 		s.filesURL+"/"+url.PathEscape(id)+"?fields=description&supportsAllDrives=true", "", nil)
 	switch {

transport/internet/xdrive/drive_live_test.go

@@ -7,8 +7,13 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"net/http"
+	"net/http/httptrace"
+	"net/url"
 	"os"
+	"sort"
 	"strconv"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -343,3 +348,130 @@ func TestLiveDriveListLag(t *testing.T) {
 	}
 	t.Logf("worst listing lag: %v", worst.Round(time.Millisecond))
 }
+
+func TestLiveDriveLatencyFloor(t *testing.T) {
+	config := liveDriveConfig(t)
+	root, err := newDriveStorage(&internet.MemoryStreamConfig{}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer root.Close()
+	store := root.session().(*driveStorage)
+	defer store.Close()
+
+	ctx := context.Background()
+	const probe = "latency-floor/probe"
+	id, err := store.makeSlot(ctx, probe)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.deleteID(ctx, flatten(probe), id)
+
+	small := newSlotFrame(1, slotInline, randomPayload(700)).encode()
+	large := newSlotFrame(1, slotInline, randomPayload(slotInlineLimit)).encode()
+	file := store.filesURL + "/" + url.PathEscape(id)
+
+	type sample struct {
+		total, ttfb time.Duration
+		reused      bool
+	}
+	measure := func(label string, op func(context.Context) error) []sample {
+		var out []sample
+		for i := 0; i < 28; i++ {
+			var wrote, first time.Time
+			var reused bool
+			trace := &httptrace.ClientTrace{
+				GotConn:              func(info httptrace.GotConnInfo) { reused = info.Reused },
+				WroteRequest:         func(httptrace.WroteRequestInfo) { wrote = time.Now() },
+				GotFirstResponseByte: func() { first = time.Now() },
+			}
+			start := time.Now()
+			err := op(httptrace.WithClientTrace(ctx, trace))
+			total := time.Since(start)
+			if err != nil {
+				t.Logf("latency-floor: %s sample failed: %v", label, err)
+			} else if i >= 3 {
+				var ttfb time.Duration
+				if !wrote.IsZero() && !first.IsZero() {
+					ttfb = first.Sub(wrote)
+				}
+				out = append(out, sample{total, ttfb, reused})
+			}
+			time.Sleep(250 * time.Millisecond)
+		}
+		return out
+	}
+	report := func(label string, s []sample) time.Duration {
+		if len(s) == 0 {
+			t.Logf("latency-floor: %-30s no samples", label)
+			return 0
+		}
+		totals := make([]time.Duration, len(s))
+		ttfbs := make([]time.Duration, len(s))
+		reused := 0
+		for i, x := range s {
+			totals[i], ttfbs[i] = x.total, x.ttfb
+			if x.reused {
+				reused++
+			}
+		}
+		sort.Slice(totals, func(a, b int) bool { return totals[a] < totals[b] })
+		sort.Slice(ttfbs, func(a, b int) bool { return ttfbs[a] < ttfbs[b] })
+		q := func(v []time.Duration, p float64) time.Duration { return v[min(int(p*float64(len(v))), len(v)-1)] }
+		t.Logf("latency-floor: %-30s n=%d min=%v p50=%v p95=%v ttfb_p50=%v reused=%d/%d",
+			label, len(s), totals[0], q(totals, 0.5), q(totals, 0.95), q(ttfbs, 0.5), reused, len(s))
+		return q(ttfbs, 0.5)
+	}
+	patch := func(target string, doc map[string]any) func(context.Context) error {
+		body, _ := json.Marshal(doc)
+		return func(ctx context.Context) error {
+			status, reply, err := store.do(ctx, http.MethodPatch, target, "application/json; charset=UTF-8", body)
+			if err == nil && status != http.StatusOK {
+				err = fmt.Errorf("status %d: %s", status, reply)
+			}
+			return err
+		}
+	}
+	get := func(target string) func(context.Context) error {
+		return func(ctx context.Context) error {
+			status, reply, err := store.do(ctx, http.MethodGet, target, "", nil)
+			if err == nil && status != http.StatusOK {
+				err = fmt.Errorf("status %d: %s", status, reply)
+			}
+			return err
+		}
+	}
+
+	proto := ""
+	netFloor := report("net floor (401, no auth)", measure("net", func(ctx context.Context) error {
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://www.googleapis.com/drive/v3/about?fields=user", nil)
+		if err != nil {
+			return err
+		}
+		resp, err := store.client.Do(req)
+		if err != nil {
+			return err
+		}
+		proto = resp.Proto
+		io.Copy(io.Discard, resp.Body)
+		return resp.Body.Close()
+	}))
+	t.Logf("latency-floor: negotiated protocol %s", proto)
+	readDesc := report("GET description (prod)", measure("get", func(ctx context.Context) error {
+		_, err := store.readDescription(ctx, id)
+		return err
+	}))
+	report("GET fields=id", measure("getid", get(file+"?fields=id&supportsAllDrives=true")))
+	report("GET description, no allDrives", measure("getnoall", get(file+"?fields=description")))
+	write1k := report("PATCH inline 1 KB (prod)", measure("p1k", func(ctx context.Context) error {
+		return store.setSlot(ctx, id, small, nil)
+	}))
+	write14k := report("PATCH inline 14 KB (prod)", measure("p14k", func(ctx context.Context) error {
+		return store.setSlot(ctx, id, large, nil)
+	}))
+	report("PATCH 1 KB, no allDrives", measure("pnoall", patch(file+"?fields=id", map[string]any{"description": small})))
+	report("PATCH name 200 B", measure("pname", patch(file+"?fields=id&supportsAllDrives=true", map[string]any{"name": strings.Repeat("n", 200)})))
+	report("PATCH appProperties 100 B", measure("pprop", patch(file+"?fields=id&supportsAllDrives=true", map[string]any{"appProperties": map[string]string{"k": strings.Repeat("a", 100)}})))
+	t.Logf("latency-floor: derived: write backend ~ %v, read backend ~ %v, 14 KB payload cost ~ %v",
+		write1k-netFloor, readDesc-netFloor, write14k-write1k)
+}

transport/internet/xdrive/params.go

@@ -13,7 +13,7 @@ const (
 	defaultConcurrency     = 8
 
 	maxSegmentBytes = 16 * 1024 * 1024
-	maxConcurrency  = 64
+	maxConcurrency  = 256
 )
 
 type params struct {

transport/internet/xdrive/slots.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/xtls/xray-core/common/errors"
@@ -27,7 +28,6 @@ const (
 	// slotInlineLimit is how large a segment is at most to go in the description
 	// of its slot, which takes one request less than a body
 	slotInlineLimit = 14 << 10
-	slotDepthLimit  = 8
 	// inlineRampSegments segments of one size in a row double it, from
 	// slotInlineLimit up to segmentBytes
 	inlineRampSegments = 8
@@ -41,19 +41,26 @@ const (
 var (
 	// the head is asked for between slotHuntFloor and slotHuntCeiling, a segment
 	// behind it at about the pace that segments come
-	slotHuntFloor   = 50 * time.Millisecond
-	slotHuntCeiling = 200 * time.Millisecond
+	slotHuntFloor   = 10 * time.Millisecond
+	slotHuntCeiling = 25 * time.Millisecond
 	slotTailMin     = 150 * time.Millisecond
 	slotTailMax     = 800 * time.Millisecond
-	slotFetchAhead  = 6
+	slotFetchAhead  = 12
 
-	slotAckRetry     = 100 * time.Millisecond
-	slotWatermarkTTL = 250 * time.Millisecond
+	slotAckRetry     = 30 * time.Millisecond
+	slotWatermarkTTL = 100 * time.Millisecond
 	slotStallTimeout = 60 * time.Second
 	slotCloseGrace   = 10 * time.Second
 	slotGoneTimeout  = 60 * time.Second
+	slotBodyHedge    = 2 * time.Second
+
+	slotReadAhead      int64 = 3
+	slotReadAheadPoll        = 40 * time.Millisecond
+	slotReadAheadBurst       = 500 * time.Millisecond
 )
 
+var slotWriteInflight = 12
+
 // slotStore is a storage of objects that are written and read again by id,
 // each with a description and a body
 type slotStore interface {
@@ -143,7 +150,7 @@ func decodeRing(b []byte) (ring, bool) {
 
 // makeRing makes the slots and watermarks of a session, all at once
 func makeRing(ctx context.Context, store slotStore, session string, p params) (ring, error) {
-	depth := min(max(p.concurrency, 2), slotDepthLimit)
+	depth := max(p.concurrency, 2)
 	r := ring{
 		Up:   slotTable{Slots: make([]string, depth)},
 		Down: slotTable{Slots: make([]string, depth)},
@@ -177,7 +184,7 @@ func makeRing(ctx context.Context, store slotStore, session string, p params) (r
 }
 
 func readWatermark(ctx context.Context, store slotStore, id string) (int64, error) {
-	d, err := store.readDescription(ctx, id)
+	d, err := store.readDescription(context.WithValue(ctx, laneKey{}, true), id)
 	if err != nil || d == "" {
 		return -1, err
 	}
@@ -216,7 +223,7 @@ func newSlotWriter(ctx context.Context, store slotStore, table slotTable, p para
 		store:    store,
 		table:    table,
 		params:   p,
-		slots:    make(chan struct{}, len(table.Slots)),
+		slots:    make(chan struct{}, min(len(table.Slots), slotWriteInflight)),
 		size:     min(slotInlineLimit, p.segmentBytes),
 		acked:    -1,
 		progress: time.Now(),
@@ -271,6 +278,10 @@ func (w *slotWriter) readyToFlush() bool {
 	grew := len(w.buf) > w.lastSize
 	w.lastSize = len(w.buf)
 
+	if len(w.slots) > 0 && w.held < maxCoalescedTicks && len(w.buf) < w.size/2 {
+		w.held++
+		return false
+	}
 	if grew && w.held < maxCoalescedTicks {
 		w.held++
 		return false
@@ -296,11 +307,19 @@ func (w *slotWriter) flushLocked() error {
 		return w.err
 	}
 
+	if w.size < w.segmentBytes && len(w.buf) >= w.size && len(w.slots)*4 > cap(w.slots)*3 {
+		w.size = w.segmentBytes
+		w.full = 0
+	}
+
 	n := min(len(w.buf), w.size)
 	if len(w.buf) > 4*slotInlineLimit {
 		// a burst goes at the full size at once
 		n = min(len(w.buf), w.segmentBytes)
 	}
+	if n > slotInlineLimit && len(w.buf) <= 2*cap(w.slots)*slotInlineLimit {
+		n = slotInlineLimit
+	}
 	if n == w.size && w.size < w.segmentBytes {
 		if w.full++; w.full == inlineRampSegments {
 			w.full = 0
@@ -332,15 +351,21 @@ func (w *slotWriter) upload(seq int64, chunk []byte) {
 	if len(chunk) <= slotInlineLimit {
 		mode = slotInline
 	}
+	began := time.Now()
 	if err := w.await(seq); err != nil {
 		w.fail(errors.New("failed to store segment ", seq).Base(err))
 		return
 	}
+	awaited := time.Since(began)
+	putAt := time.Now()
 	if err := w.write(seq, mode, chunk); err != nil {
 		w.fail(errors.New("failed to store segment ", seq).Base(err))
 		// the slot is free, the reader learns there what happened
 		w.write(seq, slotErr, nil)
+		return
 	}
+	errors.LogDebug(w.ctx, "xdrive write ", slotTag(w.table.Watermark), " seq=", seq, " bytes=", len(chunk),
+		" mode=", mode, " await=", awaited, " put=", time.Since(putAt))
 }
 
 func (w *slotWriter) write(seq int64, mode byte, data []byte) error {
@@ -393,7 +418,7 @@ func (w *slotWriter) await(seq int64) error {
 		w.ackMu.Lock()
 		still := time.Since(w.progress)
 		w.ackMu.Unlock()
-		if time.Since(start) > slotStallTimeout || closed && still > slotCloseGrace {
+		if min(time.Since(start), still) > slotStallTimeout || closed && still > slotCloseGrace {
 			return errors.New("the reader took nothing from slot ", w.slotID(seq), " for ", still).Base(err)
 		}
 
@@ -461,6 +486,7 @@ type slotReader struct {
 	ch   chan []byte
 	wake chan struct{}
 	acks chan struct{}
+	grow chan struct{}
 
 	mu      sync.Mutex
 	head    int64
@@ -472,7 +498,11 @@ type slotReader struct {
 	gap      time.Duration
 	activeAt time.Time
 	active   chan struct{}
+	frontier int64
 	err      error
+	reads    atomic.Int64
+	tailMin  time.Duration
+	tailMax  time.Duration
 }
 
 func newSlotReader(ctx context.Context, store slotStore, table slotTable, p params) *slotReader {
@@ -484,9 +514,12 @@ func newSlotReader(ctx context.Context, store slotStore, table slotTable, p para
 		ch:       make(chan []byte, p.concurrency),
 		wake:     make(chan struct{}, 1),
 		acks:     make(chan struct{}, 1),
+		grow:     make(chan struct{}, 1),
 		hitAt:    time.Now(),
 		activeAt: time.Now(),
 		active:   make(chan struct{}),
+		tailMin:  slotTailMin,
+		tailMax:  slotTailMax,
 	}
 	go r.run()
 	go r.ackLoop()
@@ -516,9 +549,20 @@ type slotFetch struct {
 	ready chan struct{}
 	// promoted is closed when the segment becomes the head
 	promoted chan struct{}
+	near     chan struct{}
 	body     []byte
 	end      bool
 	err      error
+	born     time.Time
+	readyAt  time.Time
+}
+
+func (f *slotFetch) enter() {
+	select {
+	case <-f.near:
+	default:
+		close(f.near)
+	}
 }
 
 func (r *slotReader) run() {
@@ -529,17 +573,20 @@ func (r *slotReader) run() {
 	fetches := make(map[int64]*slotFetch)
 	for {
 		r.mu.Lock()
-		head, started := r.head, r.started
+		head, started, frontier := r.head, r.started, r.frontier
 		r.mu.Unlock()
 
 		window := 1
 		if started {
-			window = min(slotFetchAhead, len(r.table.Slots))
+			window = min(max(int(frontier-head)+slotFetchAhead, slotFetchAhead), len(r.table.Slots))
 		}
 		for i := range window {
 			seq := head + int64(i)
 			if fetches[seq] == nil {
-				fetches[seq] = &slotFetch{ready: make(chan struct{}), promoted: make(chan struct{})}
+				fetches[seq] = &slotFetch{ready: make(chan struct{}), promoted: make(chan struct{}), near: make(chan struct{}), born: time.Now()}
+				if int64(i) <= slotReadAhead {
+					fetches[seq].enter()
+				}
 				go r.fetch(ctx, seq, i, fetches[seq])
 			}
 		}
@@ -547,6 +594,8 @@ func (r *slotReader) run() {
 		f := fetches[head]
 		select {
 		case <-f.ready:
+		case <-r.grow:
+			continue
 		case <-ctx.Done():
 			return
 		}
@@ -564,8 +613,14 @@ func (r *slotReader) run() {
 		r.head++
 		r.started = true
 		r.mu.Unlock()
+		errors.LogDebug(r.ctx, "xdrive read ", slotTag(r.table.Watermark), " seq=", head, " fetch=",
+			f.readyAt.Sub(f.born), " blocked=", time.Since(f.readyAt), " bytes=", len(f.body),
+			" reads=", r.reads.Load())
 		if next := fetches[head+1]; next != nil {
 			close(next.promoted)
+			if nf := fetches[head+1+slotReadAhead]; nf != nil {
+				nf.enter()
+			}
 		}
 		select {
 		case r.acks <- struct{}{}:
@@ -586,11 +641,35 @@ func (r *slotReader) isHead(seq int64) bool {
 	return r.head == seq
 }
 
+func (r *slotReader) bursting() bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return time.Since(r.hitAt) <= slotReadAheadBurst
+}
+
+func (r *slotReader) reach(seq int64) {
+	r.mu.Lock()
+	grown := seq > r.frontier
+	if grown {
+		r.frontier = seq
+	}
+	r.mu.Unlock()
+	if grown {
+		select {
+		case r.grow <- struct{}{}:
+		default:
+		}
+	}
+}
+
 // fetch asks for the slot of seq until it holds seq: the head often, a segment
-// behind it at the pace that segments come, as it would wait for the head anyway
+// behind it soon after it was asked for, then at the pace that segments come
 func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFetch) {
-	defer close(f.ready)
-	if offset > 0 && !r.hold(ctx, time.Duration(offset)*r.pace(), f.promoted, nil) {
+	defer func() {
+		f.readyAt = time.Now()
+		close(f.ready)
+	}()
+	if offset > 0 && !r.hold(ctx, r.tailMin, f.promoted, f.near) {
 		return
 	}
 
@@ -598,7 +677,17 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
 	backoff := slotHuntFloor
 	var goneSince, warnedAt time.Time
 	for {
-		description, err := r.store.readDescription(ctx, id)
+		var description string
+		var err error
+		r.mu.Lock()
+		hot := time.Since(r.activeAt) <= r.eagerWindow
+		r.mu.Unlock()
+		if hot && r.isHead(seq) {
+			description, err = r.readHead(ctx, id, seq)
+		} else {
+			r.reads.Add(1)
+			description, err = r.store.readDescription(ctx, id)
+		}
 		frame, ok := decodeSlot(description)
 		switch {
 		case err == nil && ok && frame.seq == seq:
@@ -606,6 +695,7 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
 			r.mu.Lock()
 			r.touch()
 			r.mu.Unlock()
+			r.reach(seq)
 			switch frame.mode {
 			case slotEnd:
 				f.end = true
@@ -618,7 +708,7 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
 				r.hit()
 				return
 			case slotMedia:
-				if body, err := r.store.readBody(ctx, id); err == nil && frame.holds(body) {
+				if body, err := r.readBodyHedged(ctx, id); err == nil && frame.holds(body) {
 					f.body = body
 					r.hit()
 					return
@@ -654,7 +744,15 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
 			// only the head is asked for until something comes
 			wait, a, b, backoff = 0, f.promoted, active, slotHuntFloor
 		default:
-			wait, a, backoff = min(max(r.pace(), slotTailMin), slotTailMax), f.promoted, slotHuntFloor
+			wait, a, b, backoff = min(max(r.pace(), r.tailMin), r.tailMax), f.promoted, f.near, slotHuntFloor
+			select {
+			case <-f.near:
+				b = nil
+				if r.bursting() {
+					wait = slotReadAheadPoll
+				}
+			default:
+			}
 		}
 		if !r.hold(ctx, wait, a, b) {
 			return
@@ -662,6 +760,106 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
 	}
 }
 
+func (r *slotReader) readHead(ctx context.Context, id string, seq int64) (string, error) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	ctx = context.WithValue(ctx, laneKey{}, true)
+
+	type result struct {
+		description string
+		err         error
+	}
+	results := make(chan result, 4)
+	outstanding := 0
+	launch := func() {
+		outstanding++
+		go func() {
+			r.reads.Add(1)
+			description, err := r.store.readDescription(ctx, id)
+			results <- result{description, err}
+		}()
+	}
+	launch()
+	ticker := time.NewTicker(slotHuntCeiling)
+	defer ticker.Stop()
+	var last result
+	seen := false
+	for {
+		select {
+		case res := <-results:
+			outstanding--
+			if res.err != nil {
+				return res.description, res.err
+			}
+			if frame, ok := decodeSlot(res.description); ok && frame.seq == seq {
+				return res.description, nil
+			}
+			last, seen = res, true
+		case <-ticker.C:
+			r.mu.Lock()
+			idle := time.Since(r.activeAt) > r.eagerWindow
+			r.mu.Unlock()
+			switch {
+			case idle && seen:
+				return last.description, nil
+			case !idle && outstanding < 3:
+				launch()
+			}
+		case <-r.wake:
+			if outstanding < 3 {
+				launch()
+			}
+		case <-ctx.Done():
+			return "", ctx.Err()
+		}
+	}
+}
+
+func (r *slotReader) readBodyHedged(ctx context.Context, id string) ([]byte, error) {
+	if slotBodyHedge <= 0 {
+		return r.store.readBody(ctx, id)
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	type result struct {
+		body []byte
+		err  error
+	}
+	results := make(chan result, 2)
+	launch := func() {
+		go func() {
+			body, err := r.store.readBody(ctx, id)
+			results <- result{body, err}
+		}()
+	}
+	launch()
+	outstanding := 1
+	timer := time.NewTimer(slotBodyHedge)
+	defer timer.Stop()
+	var last result
+	for {
+		select {
+		case res := <-results:
+			outstanding--
+			if res.err == nil {
+				return res.body, nil
+			}
+			last = res
+			if outstanding == 0 {
+				return nil, last.err
+			}
+		case <-timer.C:
+			if outstanding == 1 {
+				launch()
+				outstanding++
+			}
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+}
+
 // pace is about how long segments take to come one after the other
 func (r *slotReader) pace() time.Duration {
 	r.mu.Lock()
@@ -726,7 +924,7 @@ func (r *slotReader) ackLoop() {
 			if want <= acked {
 				break
 			}
-			if err := r.store.setSlot(r.ctx, r.table.Watermark, strconv.FormatInt(want, 10), nil); err == nil {
+			if err := r.store.setSlot(context.WithValue(r.ctx, laneKey{}, true), r.table.Watermark, strconv.FormatInt(want, 10), nil); err == nil {
 				acked, wait = want, slotAckRetry
 				continue
 			}
@@ -756,3 +954,10 @@ func (r *slotReader) Err() error {
 	}
 	return io.EOF
 }
+
+func slotTag(id string) string {
+	if len(id) > 6 {
+		return id[len(id)-6:]
+	}
+	return id
+}

transport/internet/xdrive/slots_test.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -304,6 +305,373 @@ func TestSlotStaleBody(t *testing.T) {
 	client, server := slotPair(t, &staleSlots{memSlots: newMemSlots(), served: make(map[string]bool)}, 2, 64<<10)
 	payload := randomPayload(200 << 10)
 	errs := make(chan error, 1)
-	writeAll(client, payload, true, errs)
+	go func() {
+		_, err := client.Write(payload)
+		if err == nil {
+			err = client.Close()
+		}
+		errs <- err
+	}()
 	readAll(t, server, payload)
 }
+
+func TestSlotBurstSplitsInline(t *testing.T) {
+	store := newMemSlots()
+	client, server := slotPair(t, store, 8, 512<<10)
+	payload := randomPayload(90 << 10)
+	errs := make(chan error, 1)
+	writeAll(client, payload, true, errs)
+	readAll(t, server, payload)
+	store.mu.Lock()
+	bodies := len(store.bodies)
+	store.mu.Unlock()
+	if bodies != 0 {
+		t.Fatalf("a burst of 90 KiB went in %d bodies, want none", bodies)
+	}
+}
+
+type stuckBodySlots struct {
+	*memSlots
+	calls atomic.Int32
+}
+
+func (s *stuckBodySlots) readBody(ctx context.Context, id string) ([]byte, error) {
+	if s.calls.Add(1) == 1 {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	return s.memSlots.readBody(ctx, id)
+}
+
+func TestSlotBodyHedge(t *testing.T) {
+	old := slotBodyHedge
+	slotBodyHedge = 50 * time.Millisecond
+	defer func() { slotBodyHedge = old }()
+
+	store := &stuckBodySlots{memSlots: newMemSlots()}
+	client, server := slotPair(t, store, 2, 64<<10)
+	payload := randomPayload(200 << 10)
+	errs := make(chan error, 1)
+	go func() {
+		_, err := client.Write(payload)
+		if err == nil {
+			err = client.Close()
+		}
+		errs <- err
+	}()
+	readAll(t, server, payload)
+	if store.calls.Load() < 2 {
+		t.Fatalf("the stuck body read was not hedged, %d reads", store.calls.Load())
+	}
+}
+
+func TestTakeSpareSkipsStale(t *testing.T) {
+	cfg := &Config{}
+	p := params{sessionTTL: 5 * time.Minute}
+	st1, err := newLocalStorage(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	st2, err := newLocalStorage(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	box := spareBoxFor(cfg)
+	box.ch <- spareRing{storage: st1, session: "old", at: time.Now().Add(-time.Hour)}
+	box.ch <- spareRing{storage: st2, session: "new", at: time.Now()}
+	got, ok := takeSpare(cfg, p)
+	if !ok || got.session != "new" {
+		t.Fatalf("takeSpare gave %q, %v, want the fresh one", got.session, ok)
+	}
+	if _, ok := takeSpare(cfg, p); ok {
+		t.Fatal("takeSpare returned a second spare")
+	}
+}
+
+func TestDropSpareDeletesSession(t *testing.T) {
+	st, err := newLocalStorage(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	name := sessionPrefix("gone") + "/" + uplinkDir + "/s0"
+	if err := st.Put(context.Background(), name, []byte("x")); err != nil {
+		t.Fatal(err)
+	}
+	dropSpare(spareRing{storage: st, session: "gone"})
+	if got, _ := st.List(context.Background(), sessionPrefix("gone")); len(got) != 0 {
+		t.Fatalf("%d entries left of a dropped spare", len(got))
+	}
+}
+
+type stuckHeadSlots struct {
+	*memSlots
+	calls atomic.Int32
+}
+
+func (s *stuckHeadSlots) readDescription(ctx context.Context, id string) (string, error) {
+	if s.calls.Add(1) == 1 {
+		<-ctx.Done()
+		return "", ctx.Err()
+	}
+	return s.memSlots.readDescription(ctx, id)
+}
+
+func TestSlotHeadReadOverlaps(t *testing.T) {
+	client, server := slotPair(t, &stuckHeadSlots{memSlots: newMemSlots()}, 4, 64<<10)
+	for _, c := range [][2]*Conn{{client, server}, {server, client}} {
+		c[0].Write([]byte("ping"))
+		if _, err := io.ReadFull(c[1], make([]byte, 4)); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestSlotBurstReadsTogether(t *testing.T) {
+	client, server := slotPair(t, newMemSlots(), 16, 512<<10)
+	payload := randomPayload(8 * slotInlineLimit)
+	go client.Write(payload)
+	start := time.Now()
+	got := make([]byte, len(payload))
+	if _, err := io.ReadFull(server, got); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatal("the burst came back changed")
+	}
+	if took := time.Since(start); took > time.Second {
+		t.Fatalf("a burst of 8 segments took %v to read", took)
+	}
+}
+
+type lagSlots struct{ *memSlots }
+
+func (s lagSlots) readDescription(ctx context.Context, id string) (string, error) {
+	select {
+	case <-ctx.Done():
+		return "", ctx.Err()
+	case <-time.After(200 * time.Millisecond):
+	}
+	return s.memSlots.readDescription(ctx, id)
+}
+
+func (s lagSlots) setSlot(ctx context.Context, id, description string, body []byte) error {
+	if _, ok := decodeSlot(description); ok {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(600 * time.Millisecond):
+		}
+	}
+	return s.memSlots.setSlot(ctx, id, description, body)
+}
+
+func TestSlotReadAheadBurst(t *testing.T) {
+	oldMin, oldMax := slotTailMin, slotTailMax
+	slotTailMin, slotTailMax = 150*time.Millisecond, 800*time.Millisecond
+	t.Cleanup(func() { slotTailMin, slotTailMax = oldMin, oldMax })
+
+	client, server := slotPair(t, lagSlots{newMemSlots()}, 16, 512<<10)
+	payload := randomPayload(8 * slotInlineLimit)
+	go client.Write(payload)
+	start := time.Now()
+	got := make([]byte, len(payload))
+	if _, err := io.ReadFull(server, got); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatal("the burst came back changed")
+	}
+	if took := time.Since(start); took > 1700*time.Millisecond {
+		t.Fatalf("a burst of 8 segments took %v to read", took)
+	}
+	client.Close()
+	server.Close()
+}
+
+func TestSlotLargeBurstInline(t *testing.T) {
+	store := newMemSlots()
+	client, server := slotPair(t, store, 16, 512<<10)
+	payload := randomPayload(10 * slotInlineLimit)
+	go client.Write(payload)
+	got := make([]byte, len(payload))
+	if _, err := io.ReadFull(server, got); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatal("the burst came back changed")
+	}
+	store.mu.Lock()
+	bodies := len(store.bodies)
+	store.mu.Unlock()
+	if bodies != 0 {
+		t.Fatalf("a burst of 140 KiB went in %d bodies, want none", bodies)
+	}
+}
+
+func TestSlotDepthFollowsConcurrency(t *testing.T) {
+	for _, c := range []struct{ set, want int }{{100, 100}, {maxConcurrency, maxConcurrency}, {1 << 20, maxConcurrency}} {
+		p := paramsFromConfig(&Config{Concurrency: uint32(c.set)})
+		r, err := makeRing(context.Background(), newMemSlots(), "session", p)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(r.Up.Slots) != c.want || len(r.Down.Slots) != c.want {
+			t.Fatalf("concurrency %d made %d and %d slots, want %d", c.set, len(r.Up.Slots), len(r.Down.Slots), c.want)
+		}
+	}
+}
+
+func TestSlotDeepRingRoundTrip(t *testing.T) {
+	client, server := slotPair(t, newMemSlots(), 64, 64<<10)
+	payload := randomPayload(2 << 20)
+	errs := make(chan error, 1)
+	writeAll(client, payload, true, errs)
+	readAll(t, server, payload)
+	if err := <-errs; err != nil {
+		t.Fatal(err)
+	}
+}
+
+type pollSlots struct {
+	*memSlots
+	pmu    sync.Mutex
+	polled map[string]bool
+}
+
+func (s *pollSlots) readDescription(ctx context.Context, id string) (string, error) {
+	s.pmu.Lock()
+	s.polled[id] = true
+	s.pmu.Unlock()
+	return s.memSlots.readDescription(ctx, id)
+}
+
+func TestSlotReaderReachesPastWindow(t *testing.T) {
+	store := &pollSlots{memSlots: newMemSlots(), polled: make(map[string]bool)}
+	p := paramsFromConfig(&Config{FlushIntervalMs: 1, PollIntervalMs: 1, MaxPollIntervalMs: 20, EagerWindowMs: 50, Concurrency: 64})
+	ring, err := makeRing(context.Background(), store, "session", p)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	put := func(seq int64) {
+		slot := ring.Down.Slots[seq%int64(len(ring.Down.Slots))]
+		if err := store.setSlot(ctx, slot, newSlotFrame(seq, slotInline, []byte{byte(seq)}).encode(), nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+	reader := newSlotReader(ctx, store, ring.Down, p)
+	next := func() []byte {
+		select {
+		case chunk := <-reader.chunks():
+			return chunk
+		case <-time.After(10 * time.Second):
+			t.Fatal("no segment came")
+			return nil
+		}
+	}
+	put(0)
+	if got := next(); !bytes.Equal(got, []byte{0}) {
+		t.Fatalf("segment 0 came as %v", got)
+	}
+	for seq := int64(2); seq <= 40; seq++ {
+		put(seq)
+	}
+	last := ring.Down.Slots[40]
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		reader.Wake()
+		store.pmu.Lock()
+		polled := store.polled[last]
+		store.pmu.Unlock()
+		if polled {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("segment 40 was never asked for while segment 1 was missing")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	put(1)
+	for seq := int64(1); seq <= 40; seq++ {
+		if got := next(); !bytes.Equal(got, []byte{byte(seq)}) {
+			t.Fatalf("segment %d came as %v", seq, got)
+		}
+	}
+}
+
+type countSlots struct {
+	*memSlots
+	cur  atomic.Int32
+	peak atomic.Int32
+}
+
+func (s *countSlots) setSlot(ctx context.Context, id, description string, body []byte) error {
+	if _, ok := decodeSlot(description); !ok {
+		return s.memSlots.setSlot(ctx, id, description, body)
+	}
+	n := s.cur.Add(1)
+	for {
+		p := s.peak.Load()
+		if n <= p || s.peak.CompareAndSwap(p, n) {
+			break
+		}
+	}
+	time.Sleep(20 * time.Millisecond)
+	s.cur.Add(-1)
+	return s.memSlots.setSlot(ctx, id, description, body)
+}
+
+func TestSlotWriteInflightCap(t *testing.T) {
+	store := &countSlots{memSlots: newMemSlots()}
+	client, server := slotPair(t, store, 48, 512<<10)
+	payload := randomPayload(60 * slotInlineLimit)
+	errs := make(chan error, 1)
+	go func() {
+		for off := 0; off < len(payload); off += slotInlineLimit {
+			if _, err := client.Write(payload[off : off+slotInlineLimit]); err != nil {
+				errs <- err
+				return
+			}
+		}
+		errs <- nil
+	}()
+	got := make([]byte, len(payload))
+	if _, err := io.ReadFull(server, got); err != nil {
+		t.Fatal(err)
+	}
+	if err := <-errs; err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatal("the payload came back changed")
+	}
+	if p := int(store.peak.Load()); p > slotWriteInflight {
+		t.Fatalf("%d segments were written at once, want at most %d", p, slotWriteInflight)
+	}
+}
+
+func TestSlotBurstInlineWaves(t *testing.T) {
+	for _, c := range []struct {
+		pieces int
+		media  bool
+	}{{20, false}, {30, true}} {
+		store := newMemSlots()
+		client, server := slotPair(t, store, 16, 512<<10)
+		payload := randomPayload(c.pieces * slotInlineLimit)
+		go client.Write(payload)
+		got := make([]byte, len(payload))
+		if _, err := io.ReadFull(server, got); err != nil {
+			t.Fatal(err)
+		}
+		if !bytes.Equal(got, payload) {
+			t.Fatal("the burst came back changed")
+		}
+		store.mu.Lock()
+		bodies := len(store.bodies)
+		store.mu.Unlock()
+		if (bodies > 0) != c.media {
+			t.Fatalf("a burst of %d pieces went in %d bodies, media wanted %v", c.pieces, bodies, c.media)
+		}
+	}
+}

transport/internet/xdrive/xdrive.go

@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/xtls/xray-core/common"
@@ -26,6 +27,11 @@ const (
 	downlinkDir  = "s2c"
 )
 
+const (
+	spareCount    = 2
+	spareIdleStop = 10 * time.Minute
+)
+
 func init() {
 	common.Must(internet.RegisterProtocolConfigCreator(protocolName, func() interface{} {
 		return new(Config)
@@ -78,30 +84,152 @@ func streamConfig(streamSettings *internet.MemoryStreamConfig) (*Config, error)
 	return config, nil
 }
 
+type spareRing struct {
+	storage Storage
+	session string
+	ring    ring
+	at      time.Time
+}
+
+type spareBox struct {
+	ch   chan spareRing
+	busy atomic.Bool
+	last atomic.Int64
+	stop atomic.Bool
+}
+
+var spareBoxes sync.Map
+
+func spareBoxFor(config *Config) *spareBox {
+	if b, ok := spareBoxes.Load(config); ok {
+		return b.(*spareBox)
+	}
+	b, _ := spareBoxes.LoadOrStore(config, &spareBox{ch: make(chan spareRing, spareCount)})
+	return b.(*spareBox)
+}
+
+func dropSpare(s spareRing) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	s.storage.Delete(ctx, sessionPrefix(s.session))
+	s.storage.Close()
+}
+
+func takeSpare(config *Config, p params) (spareRing, bool) {
+	box := spareBoxFor(config)
+	for {
+		select {
+		case s := <-box.ch:
+			if time.Since(s.at) < p.sessionTTL*2/3 {
+				return s, true
+			}
+			go dropSpare(s)
+		default:
+			return spareRing{}, false
+		}
+	}
+}
+
+func warmSpares(streamSettings *internet.MemoryStreamConfig, config *Config, p params) {
+	spareBoxFor(config).last.Store(time.Now().UnixNano())
+	fillSpares(streamSettings, config, p)
+}
+
+func fillSpares(streamSettings *internet.MemoryStreamConfig, config *Config, p params) {
+	box := spareBoxFor(config)
+	if !box.busy.CompareAndSwap(false, true) {
+		return
+	}
+	go func() {
+		defer box.busy.Store(false)
+		for n := len(box.ch); n > 0; n-- {
+			select {
+			case s := <-box.ch:
+				if time.Since(s.at) < p.sessionTTL/3 {
+					box.ch <- s
+				} else {
+					dropSpare(s)
+				}
+			default:
+			}
+		}
+		for len(box.ch) < cap(box.ch) {
+			for range 40 {
+				if box.stop.Load() {
+					return
+				}
+				time.Sleep(50 * time.Millisecond)
+			}
+			if box.stop.Load() {
+				return
+			}
+			storage, err := newStorage(streamSettings)
+			if err != nil {
+				return
+			}
+			storage = sessionStorage(storage)
+			slots, ok := storage.(slotStore)
+			if !ok {
+				storage.Close()
+				return
+			}
+			session, err := newSessionID()
+			if err != nil {
+				storage.Close()
+				return
+			}
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			r, err := makeRing(ctx, slots, session, p)
+			cancel()
+			if err != nil {
+				dropSpare(spareRing{storage: storage, session: session})
+				return
+			}
+			select {
+			case box.ch <- spareRing{storage: storage, session: session, ring: r, at: time.Now()}:
+			default:
+				storage.Close()
+				return
+			}
+		}
+		time.AfterFunc(p.sessionTTL/4, func() {
+			if !box.stop.Load() && time.Since(time.Unix(0, box.last.Load())) < spareIdleStop {
+				fillSpares(streamSettings, config, p)
+			}
+		})
+	}()
+}
+
 func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.MemoryStreamConfig) (stat.Connection, error) {
 	config, err := streamConfig(streamSettings)
 	if err != nil {
 		return nil, err
 	}
 
-	storage, err := newStorage(streamSettings)
-	if err != nil {
-		return nil, err
-	}
-	storage = sessionStorage(storage)
-
-	session, err := newSessionID()
-	if err != nil {
-		storage.Close()
-		return nil, err
-	}
-
 	p := paramsFromConfig(config)
+
+	spare, warm := takeSpare(config, p)
+	storage, session, r := spare.storage, spare.session, spare.ring
+	if !warm {
+		storage, err = newStorage(streamSettings)
+		if err != nil {
+			return nil, err
+		}
+		storage = sessionStorage(storage)
+
+		session, err = newSessionID()
+		if err != nil {
+			storage.Close()
+			return nil, err
+		}
+	}
+
 	var announce []byte
-	var r ring
 	slots, ok := storage.(slotStore)
 	if ok {
-		if r, err = makeRing(ctx, slots, session, p); err == nil {
+		if warm {
+			announce, _ = json.Marshal(r)
+		} else if r, err = makeRing(ctx, slots, session, p); err == nil {
 			announce, _ = json.Marshal(r)
 		} else {
 			ok = false
@@ -115,6 +243,7 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me
 	}
 
 	if ok {
+		warmSpares(streamSettings, config, p)
 		errors.LogInfo(ctx, "opened session ", session, " in ", len(r.Up.Slots), " slots")
 		return newSlotConn(context.Background(), slots, r.Up, r.Down, p, func() {
 			storage.Close()
@@ -320,12 +449,13 @@ func (l *Listener) collect() error {
 	}
 	l.mu.Unlock()
 
+	var first error
 	for _, session := range expired {
-		if err := l.storage.Delete(l.ctx, sessionPrefix(session)); err != nil {
-			return err
+		if err := l.storage.Delete(l.ctx, sessionPrefix(session)); err != nil && first == nil {
+			first = err
 		}
 	}
-	return nil
+	return first
 }
 
 func (l *Listener) Addr() net.Addr {

transport/internet/xdrive/xdrive_test.go

@@ -69,6 +69,13 @@ func pairWith(t *testing.T, streamSettings *internet.MemoryStreamConfig) (client
 	}
 
 	return client, server, func() {
+		if config, err := streamConfig(streamSettings); err == nil {
+			box := spareBoxFor(config)
+			box.stop.Store(true)
+			for box.busy.Load() {
+				time.Sleep(10 * time.Millisecond)
+			}
+		}
 		client.Close()
 		server.Close()
 		listener.Close()