xdrive: drop V1 slot frame compatibility, keep only V2

Co-authored-by: RPRX <63339210+RPRX@users.noreply.github.com>

go.mod

@@ -26,13 +26,13 @@ require (
 	golang.org/x/crypto v0.58.0
 	golang.org/x/crypto/x509roots/fallback v0.0.0-20261020153306-9f2b28d4331a
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842
-	golang.org/x/net v0.61.0
+	golang.org/x/net v0.62.0
 	golang.org/x/sync v0.24.0
 	golang.org/x/sys v0.49.0
 	golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2
 	golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb
 	golang.zx2c4.com/wireguard/windows v1.1.1
-	google.golang.org/grpc v1.84.0
+	google.golang.org/grpc v1.85.0
 	google.golang.org/protobuf v1.36.12
 	h12.io/socks v1.0.3
 	layeh.com/gopher-luar v1.0.11
@@ -58,7 +58,7 @@ require (
 	go.yaml.in/yaml/v3 v3.0.5 // indirect
 	golang.org/x/text v0.43.0 // indirect
 	golang.org/x/tools v0.51.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260706201446-f0a921348800 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260817212433-ac3dfec99bb1 // indirect
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gvisor.dev/gvisor v0.0.0-20260122175437-89a5d21be8f0 // indirect

go.sum

@@ -109,8 +109,8 @@ golang.org/x/mod v0.5.1/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw=
-golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI=
+golang.org/x/net v0.62.0 h1:18ELalfM5+XNRwzM2nYqXsZQ9ByS1uhj2O/U7+5jqRY=
+golang.org/x/net v0.62.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA=
@@ -150,10 +150,10 @@ golang.zx2c4.com/wireguard/windows v1.1.1 h1:8/H97U1v1PNDNcBsMZgU3KFuND9MQdTsU2N
 golang.zx2c4.com/wireguard/windows v1.1.1/go.mod h1:+fbT3FFdX4zzYDLwJh5+HPEcNN/3HyNdzhNSVsQM+zs=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260706201446-f0a921348800 h1:qEHAMpSaUhtD0p3NbEEI83HwNGFxEwaSJ1G9PLnCBZE=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260706201446-f0a921348800/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
-google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0=
-google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260817212433-ac3dfec99bb1 h1:utmqiUzZAEINbLmRyAt1QjVkMyLQ+lvIwCpLaB1rZfw=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260817212433-ac3dfec99bb1/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
+google.golang.org/grpc v1.85.0 h1:BPBCfuVRQ1qvBhKDOq9uvQllMvO59rKsYhXpp7Dd8vw=
+google.golang.org/grpc v1.85.0/go.mod h1:0rQrOD58yjal/0EaVbGBANindixsCZNQPDV3VKtaKuQ=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 h1:rgSNvqscFZ1JgV/4wH5GOsZFSFkR2Eua9As3KIr2LlM=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2/go.mod h1:iMEtFwDlAhjDU9L5mY6U1XLwlIId/G3h+QcBHDIvrJ8=
 google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=

transport/internet/xdrive/slots.go

@@ -22,12 +22,12 @@ import (
 // reuses a slot only once that says what was in it before was taken
 
 const (
-	slotMagic     = "XDRVSL01"
+	slotMagic     = "XDRVSL02"
 	slotHeaderLen = len(slotMagic) + 8 + 1 + 4 + 4
 	// 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
+	slotDepthLimit  = 16
 	// inlineRampSegments segments of one size in a row double it, from
 	// slotInlineLimit up to segmentBytes
 	inlineRampSegments = 8
@@ -42,16 +42,23 @@ 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
+	slotHuntCeiling = 100 * time.Millisecond
 	slotTailMin     = 150 * time.Millisecond
 	slotTailMax     = 800 * time.Millisecond
-	slotFetchAhead  = 6
+	slotFetchAhead  = 10
 
-	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
+
+	// a failed write of a segment is tried slotWriteAttempts times
+	slotWriteAttempts = 3
+	// a read of the head that has not answered for slotHedgeTick is asked a
+	// second time, for slotHedgeBudget at most
+	slotHedgeTick   = 250 * time.Millisecond
+	slotHedgeBudget = time.Second
 )
 
 // slotStore is a storage of objects that are written and read again by id,
@@ -271,6 +278,11 @@ func (w *slotWriter) readyToFlush() bool {
 	grew := len(w.buf) > w.lastSize
 	w.lastSize = len(w.buf)
 
+	// slots are busy and the buffer small, so a bigger segment is worth the wait
+	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,6 +308,12 @@ func (w *slotWriter) flushLocked() error {
 		return w.err
 	}
 
+	// a ring three quarters busy is saturated by small segments, so widen at once
+	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
@@ -336,11 +354,31 @@ func (w *slotWriter) upload(seq int64, chunk []byte) {
 		w.fail(errors.New("failed to store segment ", seq).Base(err))
 		return
 	}
-	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)
+	var lastErr error
+	for attempt := range slotWriteAttempts {
+		if attempt > 0 {
+			select {
+			case <-w.ctx.Done():
+				return
+			case <-time.After(time.Duration(attempt) * 200 * time.Millisecond):
+			}
+			// the failed write may have gone through
+			if d, err := w.store.readDescription(w.ctx, w.slotID(seq)); err == nil {
+				if f, ok := decodeSlot(d); ok && f.seq >= seq {
+					return
+				}
+			}
+		}
+		if lastErr = w.write(seq, mode, chunk); lastErr == nil {
+			return
+		}
+		if w.ctx.Err() != nil {
+			return
+		}
 	}
+	w.fail(errors.New("failed to store segment ", seq).Base(lastErr))
+	// the slot is free, the reader learns there what happened
+	w.write(seq, slotErr, nil)
 }
 
 func (w *slotWriter) write(seq int64, mode byte, data []byte) error {
@@ -393,7 +431,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)
 		}
 
@@ -598,7 +636,13 @@ 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
+		if r.isHead(seq) {
+			description, err = r.readHedged(ctx, id)
+		} else {
+			description, err = r.store.readDescription(ctx, id)
+		}
 		frame, ok := decodeSlot(description)
 		switch {
 		case err == nil && ok && frame.seq == seq:
@@ -662,6 +706,55 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
 	}
 }
 
+// readHedged reads the description of the head, and asks a second time if the
+// first request is slow to answer, as one that is stuck costs more than one
+// too many. The first good answer is taken
+func (r *slotReader) readHedged(ctx context.Context, id string) (string, error) {
+	if slotHedgeTick <= 0 {
+		return r.store.readDescription(ctx, id)
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	type result struct {
+		description string
+		err         error
+	}
+	results := make(chan result, 2)
+	launch := func() {
+		go func() {
+			d, err := r.store.readDescription(ctx, id)
+			results <- result{d, err}
+		}()
+	}
+	launch()
+	outstanding := 1
+	deadline := time.Now().Add(slotHedgeBudget)
+	ticker := time.NewTicker(slotHedgeTick)
+	defer ticker.Stop()
+	var last result
+	for {
+		select {
+		case res := <-results:
+			outstanding--
+			if res.err == nil {
+				return res.description, nil
+			}
+			last = res
+			if outstanding == 0 {
+				return last.description, last.err
+			}
+		case <-ticker.C:
+			if outstanding < 2 && time.Now().Before(deadline) {
+				launch()
+				outstanding++
+			}
+		case <-ctx.Done():
+			return "", ctx.Err()
+		}
+	}
+}
+
 // pace is about how long segments take to come one after the other
 func (r *slotReader) pace() time.Duration {
 	r.mu.Lock()

transport/internet/xdrive/wal.go

@@ -53,6 +53,8 @@ type walWriter struct {
 	buf      []byte
 	seq      int64
 	lastSize int
+	segBytes int
+	ramped   int
 	held     int
 	closed   bool
 	err      error
@@ -66,6 +68,10 @@ func newWALWriter(ctx context.Context, storage Storage, prefix string, p params)
 		params:  p,
 		sem:     make(chan struct{}, p.concurrency),
 	}
+	// A session being set up is latency bound: small segments fit in a
+	// single inline create, so start there and double up to the configured
+	// size once full segments keep coming.
+	w.segBytes = min(p.segmentBytes, driveInlineLimit)
 	go w.flushLoop()
 	return w
 }
@@ -82,7 +88,7 @@ func (w *walWriter) Write(p []byte) (int, error) {
 	}
 
 	w.buf = append(w.buf, p...)
-	for len(w.buf) >= w.segmentBytes {
+	for len(w.buf) >= w.segBytes {
 		if err := w.flushLocked(); err != nil {
 			return 0, err
 		}
@@ -133,8 +139,15 @@ func (w *walWriter) flushLocked() error {
 	}
 
 	n := len(w.buf)
-	if n > w.segmentBytes {
-		n = w.segmentBytes
+	if n > w.segBytes {
+		n = w.segBytes
+	}
+	if n == w.segBytes && w.segBytes < w.segmentBytes {
+		w.ramped++
+		if w.ramped >= inlineRampSegments {
+			w.ramped = 0
+			w.segBytes = min(w.segBytes*2, w.segmentBytes)
+		}
 	}
 
 	chunk := make([]byte, n)
@@ -324,85 +337,75 @@ func (r *walReader) poll() (advanced, eof bool, err error) {
 		r.holeSince = time.Time{}
 	}
 
+	type result struct {
+		chunk []byte
+		err   error
+	}
+	outstanding := make(map[int64]chan result)
+
 	for {
 		if entry, ok := pending[r.seq]; ok && strings.HasSuffix(entry.Name, errSuffix) {
 			r.discard(r.prefix + "/" + entry.Name)
 			return advanced, false, errors.New("the peer could not store segment ", r.seq)
 		}
-
-		batch, done := r.nextBatch(pending)
-		if done {
+		if entry, ok := pending[r.seq]; ok && strings.HasSuffix(entry.Name, endSuffix) && len(outstanding) == 0 {
+			r.discard(r.prefix + "/" + entry.Name)
 			return advanced, true, nil
 		}
-		if len(batch) == 0 {
+
+		// keep the window of fetches full, so the next segments download
+		// while the head one is handed over
+		for i := int64(0); len(outstanding) < r.concurrency; i++ {
+			seq := r.seq + i
+			entry, ok := pending[seq]
+			if !ok || !strings.HasSuffix(entry.Name, segSuffix) {
+				break
+			}
+			if _, ok := outstanding[seq]; ok {
+				continue
+			}
+			ch := make(chan result, 1)
+			outstanding[seq] = ch
+			if entry.Inline != nil {
+				ch <- result{chunk: entry.Inline}
+				continue
+			}
+			go func(name string) {
+				chunk, err := r.storage.Get(r.ctx, r.prefix+"/"+name)
+				ch <- result{chunk, err}
+			}(entry.Name)
+		}
+
+		head, ok := outstanding[r.seq]
+		if !ok {
 			return advanced, false, nil
 		}
-
-		chunks, err := r.fetch(batch)
-		if err != nil {
-			if err == errNotFound {
+		var res result
+		select {
+		case res = <-head:
+		case <-r.ctx.Done():
+			return advanced, true, nil
+		}
+		delete(outstanding, r.seq)
+		if res.err != nil {
+			if res.err == errNotFound {
 				return advanced, false, nil
 			}
-			return advanced, false, err
+			return advanced, false, res.err
 		}
-
-		for i, chunk := range chunks {
-			select {
-			case r.ch <- chunk:
-			case <-r.ctx.Done():
-				return advanced, true, nil
-			}
-			r.seq++
-			advanced = true
-			r.discard(r.prefix + "/" + batch[i].Name)
+		select {
+		case r.ch <- res.chunk:
+		case <-r.ctx.Done():
+			return advanced, true, nil
 		}
+		name := pending[r.seq].Name
+		delete(pending, r.seq)
+		r.seq++
+		advanced = true
+		r.discard(r.prefix + "/" + name)
 	}
 }
 
-func (r *walReader) nextBatch(pending map[int64]Entry) (batch []Entry, done bool) {
-	for i := 0; i < r.concurrency; i++ {
-		entry, ok := pending[r.seq+int64(i)]
-		if !ok {
-			break
-		}
-		if !strings.HasSuffix(entry.Name, segSuffix) {
-			if i == 0 && strings.HasSuffix(entry.Name, endSuffix) {
-				r.discard(r.prefix + "/" + entry.Name)
-				return nil, true
-			}
-			break
-		}
-		batch = append(batch, entry)
-	}
-	return batch, false
-}
-
-func (r *walReader) fetch(batch []Entry) ([][]byte, error) {
-	chunks := make([][]byte, len(batch))
-	failures := make([]error, len(batch))
-
-	var wg sync.WaitGroup
-	for i, entry := range batch {
-		if entry.Inline != nil {
-			chunks[i] = entry.Inline
-			continue
-		}
-		wg.Add(1)
-		go func(i int, name string) {
-			defer wg.Done()
-			chunks[i], failures[i] = r.storage.Get(r.ctx, r.prefix+"/"+name)
-		}(i, entry.Name)
-	}
-	wg.Wait()
-
-	for i := range batch {
-		if failures[i] != nil {
-			return nil, failures[i]
-		}
-	}
-	return chunks, nil
-}
-
 func (r *walReader) discard(name string) {
 	select {
 	case r.discards <- name: