XDRIVE transport: A bunch of optimizations for the Google Drive backend (#7340)

https://github.com/XTLS/Xray-core/pull/7217#issuecomment-6489554617
https://github.com/XTLS/Xray-core/issues/7208#issuecomment-6490366098
https://github.com/XTLS/Xray-core/issues/7208#issuecomment-6558792002
https://github.com/XTLS/Xray-core/issues/7208#issuecomment-6583177537
https://github.com/XTLS/Xray-core/issues/7208#issuecomment-6634955793
https://github.com/XTLS/Xray-core/issues/7208#issuecomment-6677620667

Syncs with https://github.com/anonymously-hidden/Xray-core/commit/f7a8d213ac69410c23b4484d6eeea8dc55b580bd

---------

Co-authored-by: anonymously-hidden <106820058+anonymously-hidden@users.noreply.github.com>
Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>

transport/internet/xdrive/client.go

@@ -438,9 +438,11 @@ func (b *doneBody) Close() error {
 type sessionTransport struct {
 	pool *connPool
 
-	mu     sync.Mutex
+	mu                sync.Mutex
-	client *splithttp.XmuxClient
+	client            *splithttp.XmuxClient
-	closed bool
+	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"
+	driveBoundary         = "xdrive-boundary"
-	drivePageSize    = 1000
+	drivePageSize         = 1000
-	driveMaxAttempts = 8
+	driveMaxAttempts      = 8
-	driveMaxInflight = 32
+	driveMaxInflight      = 64
-	driveInlineLimit = 12000
+	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 {
+		wg.Add(1)
-			return err
+		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
+	slotHuntFloor   = 10 * time.Millisecond
-	slotHuntCeiling = 200 * time.Millisecond
+	slotHuntCeiling = 25 * time.Millisecond
 	slotTailMin     = 150 * time.Millisecond
 	slotTailMax     = 800 * time.Millisecond
-	slotFetchAhead  = 6
+	slotFetchAhead  = 12
 
-	slotAckRetry     = 100 * time.Millisecond
+	slotAckRetry     = 30 * time.Millisecond
-	slotWatermarkTTL = 250 * 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)
+	defer func() {
-	if offset > 0 && !r.hold(ctx, time.Duration(offset)*r.pace(), f.promoted, nil) {
+		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 {
+		if err := l.storage.Delete(l.ctx, sessionPrefix(session)); err != nil && first == nil {
-			return err
+			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()