Mult Port + TCP Calibration (SSH DEAD)

core/internal/wire/protocol.go

@@ -26,14 +26,25 @@ const (
 	StatusWait  byte = 3
 	StatusEOF   byte = 4
 
-	ProbeUpload    byte = 1
-	ProbeDownload  byte = 2
-	ProbeKeepalive byte = 3
-	ProbeBatch     byte = 4
+	ProbeUpload        byte = 1
+	ProbeDownload      byte = 2
+	ProbeKeepalive     byte = 3
+	ProbeBatch         byte = 4
+	ProbeIperfUpload   byte = 5
+	ProbeIperfDownload byte = 6
 )
 
 var ProbeMagic = [4]byte{'D', 'T', 'P', '2'}
 
+// ProbeBurstCount is deliberately fixed at one. Startup calibration measures
+// the safe record size of a single DragonTCP lane, not aggregate throughput.
+// Multiple outstanding calibration records can make a constrained carrier look
+// artificially better or worse and can produce a false ceiling. Confirmation
+// retries are performed sequentially on fresh connections by the client.
+func ProbeBurstCount(chunk int) int {
+	return 1
+}
+
 type SessionID [16]byte
 
 type Request struct {

core/internal/wire/protocol_test.go

@@ -103,3 +103,11 @@ func BenchmarkWriteRequest1MiB(b *testing.B) {
 		})
 	}
 }
+
+func TestProbeBurstCountSinglePoller(t *testing.T) {
+	for _, chunk := range []int{0, 1, 1024, 128 * 1024, 512 * 1024, 1024 * 1024} {
+		if got := ProbeBurstCount(chunk); got != 1 {
+			t.Fatalf("ProbeBurstCount(%d)=%d, want 1", chunk, got)
+		}
+	}
+}

core/internal/xorchunk/chunk.go

@@ -7,6 +7,7 @@
 package xorchunk
 
 import (
+	"bytes"
 	"context"
 	"crypto/rand"
 	"encoding/hex"
@@ -29,7 +30,7 @@ var requestCounter atomic.Uint32
 
 // NewOptions builds the transport options from the values the CLI already
 // parses, keeping the struct fields unexported as in the original.
-func NewOptions(startSize, minSize, maxSize int, adaptive bool, adaptSuccesses int, adaptLog bool,
+func NewOptions(startSize, minSize, maxSize int, adaptive bool, adaptSuccesses, shrinkAfter int, adaptLog bool,
 	pollers, reconnectEvery int, pollDelay, txnTimeout time.Duration, tcpBuffer int) Options {
 	return Options{
 		startSize:      startSize,
@@ -37,6 +38,7 @@ func NewOptions(startSize, minSize, maxSize int, adaptive bool, adaptSuccesses i
 		maxSize:        maxSize,
 		adaptive:       adaptive,
 		adaptSuccesses: adaptSuccesses,
+		shrinkAfter:    shrinkAfter,
 		adaptLog:       adaptLog,
 		pollers:        pollers,
 		reconnectEvery: reconnectEvery,
@@ -47,19 +49,24 @@ func NewOptions(startSize, minSize, maxSize int, adaptive bool, adaptSuccesses i
 }
 
 type Options struct {
-	startSize      int
-	minSize        int
-	maxSize        int
-	adaptive       bool
-	adaptSuccesses int
-	adaptLog       bool
-	pollers        int
-	reconnectEvery int
-	pollDelay      time.Duration
-	txnTimeout     time.Duration
-	tcpBuffer      int
-	headerMask     byte
-	coverProfile   cover.Profile
+	startSize         int
+	minSize           int
+	maxSize           int
+	uploadStartSize   int
+	downloadStartSize int
+	uploadMaxSize     int
+	downloadMaxSize   int
+	adaptive          bool
+	adaptSuccesses    int
+	shrinkAfter       int
+	adaptLog          bool
+	pollers           int
+	reconnectEvery    int
+	pollDelay         time.Duration
+	txnTimeout        time.Duration
+	tcpBuffer         int
+	headerMask        byte
+	coverProfile      cover.Profile
 }
 
 // WithHeaderMask returns a copy using one fixed frame-magic profile. The mask
@@ -78,6 +85,38 @@ func (o Options) WithCoverProfile(profile cover.Profile) Options {
 	return o
 }
 
+// MinSize and MaxSize expose the configured X carrier calibration bounds.
+func (o Options) MinSize() int { return o.minSize }
+func (o Options) MaxSize() int { return o.maxSize }
+
+// WithCalibratedChunks locks X to the UP/DW sizes proven by the pre-tunnel
+// fake-iperf calibration. Once calibration succeeds, runtime adaptive sizing is
+// disabled for X: transport successes cannot grow the chunk and transport
+// failures cannot shrink it. Failed physical transactions reconnect/retry using
+// the same calibrated size.
+func (o Options) WithCalibratedChunks(upload, download int) Options {
+	if upload < o.minSize {
+		upload = o.minSize
+	}
+	if download < o.minSize {
+		download = o.minSize
+	}
+	if upload > o.maxSize {
+		upload = o.maxSize
+	}
+	if download > o.maxSize {
+		download = o.maxSize
+	}
+	o.uploadStartSize = upload
+	o.downloadStartSize = download
+	o.uploadMaxSize = upload
+	o.downloadMaxSize = download
+	// Calibration replaces runtime X chunk adaptation. The calibrated values are
+	// the operating sizes for this session, not merely adaptive ceilings.
+	o.adaptive = false
+	return o
+}
+
 func wireToken(token string) string {
 	if token == "" {
 		return "-"
@@ -93,28 +132,44 @@ type adaptiveSizer struct {
 	max            int
 	adaptive       bool
 	adaptSuccesses int
+	shrinkAfter    int
+	failures       int
 	successes      int
 	good           int
 	bad            int
 	logChanges     bool
 }
 
-func newAdaptiveSizer(name string, opts Options) *adaptiveSizer {
-	start := opts.startSize
+func newAdaptiveSizer(name string, start, ceiling int, opts Options) *adaptiveSizer {
+	if ceiling <= 0 || ceiling > opts.maxSize {
+		ceiling = opts.maxSize
+	}
+	if ceiling < opts.minSize {
+		ceiling = opts.minSize
+	}
+	if start <= 0 {
+		start = opts.startSize
+	}
 	if start < opts.minSize {
 		start = opts.minSize
 	}
-	if start > opts.maxSize {
-		start = opts.maxSize
+	if start > ceiling {
+		start = ceiling
 	}
 	return &adaptiveSizer{
 		name:           name,
 		current:        start,
 		min:            opts.minSize,
-		max:            opts.maxSize,
+		max:            ceiling,
 		adaptive:       opts.adaptive,
 		adaptSuccesses: opts.adaptSuccesses,
-		logChanges:     opts.adaptLog,
+		shrinkAfter: func() int {
+			if opts.shrinkAfter > 0 {
+				return opts.shrinkAfter
+			}
+			return 1
+		}(),
+		logChanges: opts.adaptLog,
 	}
 }
 
@@ -129,7 +184,7 @@ func (s *adaptiveSizer) Success(attempted int) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	if !s.adaptive || s.current >= s.max {
+	if !s.adaptive {
 		return
 	}
 	// Ignore stale successes from records that were already in flight when
@@ -137,6 +192,10 @@ func (s *adaptiveSizer) Success(attempted int) {
 	if attempted != s.current {
 		return
 	}
+	s.failures = 0
+	if s.current >= s.max {
+		return
+	}
 
 	if attempted > s.good {
 		s.good = attempted
@@ -201,6 +260,14 @@ func (s *adaptiveSizer) Failure(attempted int) (old, next int) {
 		return old, old
 	}
 	s.successes = 0
+	s.failures++
+	if s.failures < s.shrinkAfter {
+		if s.logChanges && s.shrinkAfter > 1 {
+			fmt.Printf("adaptive %s chunk: holding %d after failure %d/%d\n", s.name, old, s.failures, s.shrinkAfter)
+		}
+		return old, old
+	}
+	s.failures = 0
 
 	if s.bad == 0 || attempted < s.bad {
 		s.bad = attempted
@@ -373,6 +440,288 @@ func ProbeProfile(serverAddr, token string, opts Options) bool {
 	return opts.headerMask == 0 && strings.HasPrefix(string(resp), "ERR expected TUNNEL")
 }
 
+type calibrationProbeResult struct {
+	ok      bool
+	bytes   int
+	elapsed time.Duration
+	err     error
+}
+
+func (r calibrationProbeResult) mbps() float64 {
+	if r.bytes <= 0 || r.elapsed <= 0 {
+		return 0
+	}
+	return float64(r.bytes*8) / r.elapsed.Seconds() / 1_000_000
+}
+
+func calibrationPattern(n int) []byte {
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = byte((i*31 + 17) & 0xff)
+	}
+	return b
+}
+
+func calibrationBurstCount(chunk int) int {
+	// Calibration is strictly single-poller/single-outstanding-request. Runtime
+	// X traffic may use its normal concurrency after calibration completes.
+	return 1
+}
+
+func calibrationTimeout(opts Options) time.Duration {
+	t := opts.txnTimeout
+	if t < 8*time.Second {
+		t = 8 * time.Second
+	}
+	if t > 20*time.Second {
+		t = 20 * time.Second
+	}
+	return t
+}
+
+func isCalibrationTimeout(err error) bool {
+	if err == nil {
+		return false
+	}
+	if ne, ok := err.(net.Error); ok && ne.Timeout() {
+		return true
+	}
+	return strings.Contains(strings.ToLower(err.Error()), "i/o timeout") || strings.Contains(strings.ToLower(err.Error()), "timeout")
+}
+
+const calibrationDecisionAttempts = 3
+
+// confirmCalibrationResult uses a 2-of-3 decision near the carrier boundary.
+// Each retry gets a fresh X connection. Timeouts are treated as inconclusive
+// connection failures, not as evidence that the candidate chunk is too large.
+func confirmCalibrationResult(serverAddr, token string, opts Options, download bool, candidate int, first calibrationProbeResult, stage string) calibrationProbeResult {
+	name := "upload"
+	if download {
+		name = "download"
+	}
+	successes, failures := 0, 0
+	var lastSuccess, lastFailure, lastTimeout calibrationProbeResult
+
+	observe := func(r calibrationProbeResult) {
+		if r.ok {
+			successes++
+			lastSuccess = r
+			return
+		}
+		if isCalibrationTimeout(r.err) {
+			lastTimeout = r
+			return
+		}
+		failures++
+		lastFailure = r
+	}
+
+	observe(first)
+	for attempt := 2; attempt <= calibrationDecisionAttempts && successes < 2 && failures < 2; attempt++ {
+		r := probeCalibrationSize(serverAddr, token, opts, download, candidate)
+		observe(r)
+		result := "failure"
+		if r.ok {
+			result = "success"
+		} else if isCalibrationTimeout(r.err) {
+			result = "connection_timeout"
+		}
+		fmt.Printf("[D-TCP] phase=CALIBRATION fake_iperf=%s wire=x stage=%s chunk=%d confirmation=%d/%d result=%s\n", name, stage, candidate, attempt, calibrationDecisionAttempts, result)
+	}
+
+	if successes >= 2 {
+		return lastSuccess
+	}
+	if failures >= 2 {
+		return lastFailure
+	}
+	if lastTimeout.err != nil {
+		return lastTimeout
+	}
+	if successes > failures && lastSuccess.ok {
+		return lastSuccess
+	}
+	return lastFailure
+}
+
+// probeCalibrationSize runs a short repeated UP or DW transfer over one X
+// physical connection. It exercises the same UP/OK framing and XOR payload path
+// as real X traffic without opening a destination tunnel.
+func probeCalibrationSize(serverAddr, token string, opts Options, download bool, candidate int) calibrationProbeResult {
+	timeout := calibrationTimeout(opts)
+	lane := newTxnLane(serverAddr, opts.tcpBuffer, 0, timeout, opts.headerMask, opts.coverProfile)
+	defer lane.Close()
+	count := calibrationBurstCount(candidate)
+	started := time.Now()
+	total := 0
+	want := calibrationPattern(candidate)
+
+	for i := 0; i < count; i++ {
+		if download {
+			resp, err := lane.Do([]byte(fmt.Sprintf("CIPERFDW %s %d", wireToken(token), candidate)))
+			if err != nil {
+				return calibrationProbeResult{bytes: total, elapsed: time.Since(started), err: err}
+			}
+			if !bytes.Equal(resp, want) {
+				return calibrationProbeResult{bytes: total, elapsed: time.Since(started), err: fmt.Errorf("X download validation failed len=%d want=%d", len(resp), candidate)}
+			}
+			total += len(resp)
+			continue
+		}
+
+		prefix := []byte(fmt.Sprintf("CIPERFUP %s %d ", wireToken(token), candidate))
+		payload := make([]byte, len(prefix)+len(want))
+		copy(payload, prefix)
+		copy(payload[len(prefix):], want)
+		resp, err := lane.Do(payload)
+		if err != nil {
+			return calibrationProbeResult{bytes: total, elapsed: time.Since(started), err: err}
+		}
+		if string(resp) != "IPERFOK" {
+			return calibrationProbeResult{bytes: total, elapsed: time.Since(started), err: fmt.Errorf("X upload rejected: %s", string(resp))}
+		}
+		total += candidate
+	}
+	return calibrationProbeResult{ok: true, bytes: total, elapsed: time.Since(started)}
+}
+
+func calibrateMaximum(serverAddr, token string, opts Options, download bool, fine int) int {
+	if fine < 1 {
+		fine = 32
+	}
+	name := "upload"
+	if download {
+		name = "download"
+	}
+	candidate := opts.minSize
+	if candidate < 32 {
+		candidate = 32
+	}
+	if candidate > opts.maxSize {
+		candidate = opts.maxSize
+	}
+	good, bad := 0, 0
+
+	for {
+		r := probeCalibrationSize(serverAddr, token, opts, download, candidate)
+		if r.ok {
+			fmt.Printf("[D-TCP] phase=CALIBRATION fake_iperf=%s wire=x stage=ascend chunk=%d records=%d bytes=%d mbps=%.2f result=success\n", name, candidate, calibrationBurstCount(candidate), r.bytes, r.mbps())
+			good = candidate
+			if candidate >= opts.maxSize {
+				return opts.maxSize
+			}
+			next := candidate * 4
+			if candidate == opts.minSize && next < 512 && opts.maxSize >= 512 {
+				next = 512
+			}
+			if next > opts.maxSize {
+				next = opts.maxSize
+			}
+			if next <= candidate {
+				return good
+			}
+			fmt.Printf("[D-TCP] phase=CALIBRATION probe=%s wire=x stage=ascend chunk_upgrade=%d->%d\n", name, candidate, next)
+			candidate = next
+			continue
+		}
+		r = confirmCalibrationResult(serverAddr, token, opts, download, candidate, r, "ascend-confirm")
+		if r.ok {
+			fmt.Printf("[D-TCP] phase=CALIBRATION fake_iperf=%s wire=x stage=ascend chunk=%d result=recovered_after_retry\n", name, candidate)
+			good = candidate
+			if candidate >= opts.maxSize {
+				return opts.maxSize
+			}
+			next := candidate * 4
+			if candidate == opts.minSize && next < 512 && opts.maxSize >= 512 {
+				next = 512
+			}
+			if next > opts.maxSize {
+				next = opts.maxSize
+			}
+			if next <= candidate {
+				return good
+			}
+			fmt.Printf("[D-TCP] phase=CALIBRATION probe=%s wire=x stage=ascend chunk_upgrade=%d->%d\n", name, candidate, next)
+			candidate = next
+			continue
+		}
+		if isCalibrationTimeout(r.err) {
+			selected := good
+			if selected == 0 {
+				selected = opts.minSize
+			}
+			fmt.Printf("[D-TCP] phase=CALIBRATION fake_iperf=%s wire=x stage=ascend chunk=%d result=connection_timeout action=keep_known_good known_good=%d err=%v\n", name, candidate, selected, r.err)
+			return selected
+		}
+		fmt.Printf("[D-TCP] phase=CALIBRATION fake_iperf=%s wire=x stage=ascend chunk=%d records=%d bytes=%d mbps=%.2f result=failure err=%v\n", name, candidate, calibrationBurstCount(candidate), r.bytes, r.mbps(), r.err)
+		bad = candidate
+		if good == 0 {
+			return opts.minSize
+		}
+		break
+	}
+
+	for bad-good > fine {
+		next := good + (bad-good)/2
+		if next <= good || next >= bad {
+			break
+		}
+		r := probeCalibrationSize(serverAddr, token, opts, download, next)
+		r = confirmCalibrationResult(serverAddr, token, opts, download, next, r, "refine-confirm")
+		if r.ok {
+			fmt.Printf("[D-TCP] phase=CALIBRATION fake_iperf=%s wire=x stage=refine chunk=%d records=%d bytes=%d mbps=%.2f result=success\n", name, next, calibrationBurstCount(next), r.bytes, r.mbps())
+			good = next
+			continue
+		}
+		if isCalibrationTimeout(r.err) {
+			fmt.Printf("[D-TCP] phase=CALIBRATION fake_iperf=%s wire=x stage=refine chunk=%d result=connection_timeout action=keep_known_good known_good=%d err=%v\n", name, next, good, r.err)
+			return good
+		}
+		fmt.Printf("[D-TCP] phase=CALIBRATION fake_iperf=%s wire=x stage=refine chunk=%d records=%d bytes=%d mbps=%.2f result=failure err=%v\n", name, next, calibrationBurstCount(next), r.bytes, r.mbps(), r.err)
+		bad = next
+	}
+	fmt.Printf("[D-TCP] phase=CALIBRATION probe=%s wire=x stage=refine selected=%d failed_above=%d resolution=%d\n", name, good, bad, fine)
+	return good
+}
+
+func probePersistent(serverAddr, token string, opts Options) bool {
+	lane := newTxnLane(serverAddr, opts.tcpBuffer, 0, minDurationX(calibrationTimeout(opts), 2500*time.Millisecond), opts.headerMask, opts.coverProfile)
+	defer lane.Close()
+	for i := 0; i < 8; i++ {
+		resp, err := lane.Do([]byte("CPROBE " + wireToken(token)))
+		if err != nil || string(resp) != "PROBEOK" {
+			return false
+		}
+	}
+	return true
+}
+
+func minDurationX(a, b time.Duration) time.Duration {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+// Calibrate performs the X wire's pre-tunnel UP/DW fake-iperf calibration.
+// It ascends by 4x and only spends extra probes around the first failure, where
+// it resolves the highest stable boundary to the requested byte precision.
+func Calibrate(serverAddr, token string, opts Options, fine int) (upload, download int, persistent bool) {
+	if opts.minSize < 32 {
+		opts.minSize = 32
+	}
+	if opts.maxSize < opts.minSize {
+		opts.maxSize = opts.minSize
+	}
+	if opts.maxSize > protocol.MaxChunkPayload {
+		opts.maxSize = protocol.MaxChunkPayload
+	}
+	upload = calibrateMaximum(serverAddr, token, opts, false, fine)
+	download = calibrateMaximum(serverAddr, token, opts, true, fine)
+	persistent = probePersistent(serverAddr, token, opts)
+	return
+}
+
 type chunkResult struct {
 	seq   uint64
 	data  []byte
@@ -471,8 +820,8 @@ func Open(serverAddr, token, targetHost string, targetPort int, opts Options) (n
 		pending:    make(map[uint64][]byte, opts.pollers*2),
 	}
 	c.ack.Store(-1)
-	c.upSizer = newAdaptiveSizer("upload", opts)
-	c.downSizer = newAdaptiveSizer("download", opts)
+	c.upSizer = newAdaptiveSizer("upload", opts.uploadStartSize, opts.uploadMaxSize, opts)
+	c.downSizer = newAdaptiveSizer("download", opts.downloadStartSize, opts.downloadMaxSize, opts)
 
 	c.pushLane = newTxnLane(serverAddr, opts.tcpBuffer, opts.reconnectEvery, opts.txnTimeout, opts.headerMask, opts.coverProfile)
 
@@ -589,14 +938,18 @@ func (c *chunkConn) pullWorker(lane *txnLane) {
 			resp, err := lane.Do(payload)
 			if err != nil {
 				old, next := c.downSizer.Failure(limit)
-				if next == old && next == c.opts.minSize {
+				if !c.opts.adaptive || (next == old && next == c.opts.minSize) {
 					consecutiveMinFailures++
 				} else {
 					consecutiveMinFailures = 0
 				}
 				if consecutiveMinFailures >= 8 {
+					detail := fmt.Sprintf("download failed at minimum chunk %d", next)
+					if !c.opts.adaptive {
+						detail = fmt.Sprintf("download failed repeatedly at fixed calibrated chunk %d", next)
+					}
 					select {
-					case c.results <- chunkResult{seq: seq, err: fmt.Errorf("download failed at minimum chunk %d: %w", next, err)}:
+					case c.results <- chunkResult{seq: seq, err: fmt.Errorf("%s: %w", detail, err)}:
 					case <-c.ctx.Done():
 					}
 					return
@@ -773,12 +1126,15 @@ func (c *chunkConn) Write(p []byte) (int, error) {
 		resp, err := c.pushLane.Do(payload)
 		if err != nil {
 			old, next := c.upSizer.Failure(size)
-			if next == old && next == c.opts.minSize {
+			if !c.opts.adaptive || (next == old && next == c.opts.minSize) {
 				consecutiveMinFailures++
 			} else {
 				consecutiveMinFailures = 0
 			}
 			if consecutiveMinFailures >= 8 {
+				if !c.opts.adaptive {
+					return total, fmt.Errorf("upload failed repeatedly at fixed calibrated chunk %d: %w", next, err)
+				}
 				return total, fmt.Errorf("upload failed at minimum chunk %d: %w", next, err)
 			}
 			time.Sleep(30 * time.Millisecond)

core/internal/xorchunk/chunk_test.go

@@ -0,0 +1,236 @@
+package xorchunk
+
+import (
+	"bytes"
+	"net"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"dragontcp/internal/protocol"
+)
+
+func TestAdaptiveSizerShrinkBudget(t *testing.T) {
+	opts := Options{
+		startSize:   1024,
+		minSize:     32,
+		maxSize:     1024,
+		adaptive:    true,
+		shrinkAfter: 3,
+	}
+	s := newAdaptiveSizer("test", opts.startSize, opts.maxSize, opts)
+	for i := 1; i <= 2; i++ {
+		_, next := s.Failure(1024)
+		if next != 1024 {
+			t.Fatalf("failure %d reduced early to %d", i, next)
+		}
+	}
+	s.Success(1024)
+	for i := 1; i <= 2; i++ {
+		_, next := s.Failure(1024)
+		if next != 1024 {
+			t.Fatalf("post-success failure %d reduced early to %d", i, next)
+		}
+	}
+	_, next := s.Failure(1024)
+	if next != 512 {
+		t.Fatalf("third consecutive failure reduced to %d, want 512", next)
+	}
+}
+
+func startCalibrationTestServer(t *testing.T, threshold int) (string, func()) {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	stop := make(chan struct{})
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func(c net.Conn) {
+				defer c.Close()
+				for {
+					id, _, payload, err := protocol.ReadRequestFrameProfile(c, 0)
+					if err != nil {
+						return
+					}
+					if bytes.HasPrefix(payload, []byte("CIPERFUP ")) {
+						parts := bytes.SplitN(payload, []byte(" "), 4)
+						if len(parts) != 4 {
+							_ = protocol.WriteResponseFrame(c, id, []byte("ERR bad upload"))
+							continue
+						}
+						size, _ := strconv.Atoi(string(parts[2]))
+						if size > threshold || len(parts[3]) != size || !bytes.Equal(parts[3], calibrationPattern(size)) {
+							_ = protocol.WriteResponseFrame(c, id, []byte("ERR too large"))
+							continue
+						}
+						_ = protocol.WriteResponseFrame(c, id, []byte("IPERFOK"))
+						continue
+					}
+					if bytes.HasPrefix(payload, []byte("CIPERFDW ")) {
+						parts := strings.Fields(string(payload))
+						if len(parts) != 3 {
+							_ = protocol.WriteResponseFrame(c, id, []byte("ERR bad download"))
+							continue
+						}
+						size, _ := strconv.Atoi(parts[2])
+						if size > threshold {
+							_ = protocol.WriteResponseFrame(c, id, []byte("ERR too large"))
+							continue
+						}
+						_ = protocol.WriteResponseFrame(c, id, calibrationPattern(size))
+						continue
+					}
+					if bytes.HasPrefix(payload, []byte("CPROBE ")) {
+						_ = protocol.WriteResponseFrame(c, id, []byte("PROBEOK"))
+						continue
+					}
+					_ = protocol.WriteResponseFrame(c, id, []byte("ERR unsupported"))
+				}
+			}(conn)
+		}
+	}()
+	return ln.Addr().String(), func() {
+		close(stop)
+		_ = ln.Close()
+	}
+}
+
+func TestXCalibrationRefinesUploadAndDownloadTo32Bytes(t *testing.T) {
+	const threshold = 731237
+	addr, closeServer := startCalibrationTestServer(t, threshold)
+	defer closeServer()
+	opts := Options{
+		startSize:  1024 * 1024,
+		minSize:    32,
+		maxSize:    1024 * 1024,
+		txnTimeout: time.Second,
+	}
+	for _, tc := range []struct {
+		name     string
+		download bool
+	}{
+		{"upload", false},
+		{"download", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			got := calibrateMaximum(addr, "", opts, tc.download, 32)
+			if got > threshold {
+				t.Fatalf("calibrated size=%d exceeds threshold=%d", got, threshold)
+			}
+			if threshold-got > 32 {
+				t.Fatalf("calibrated size=%d is more than 32 bytes below threshold=%d", got, threshold)
+			}
+		})
+	}
+}
+
+func TestXCalibrationRetriesTransientFailure(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+
+	const (
+		threshold      = 4096
+		transientChunk = 2048
+	)
+	var transient atomic.Int32
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func(c net.Conn) {
+				defer c.Close()
+				for {
+					id, _, payload, err := protocol.ReadRequestFrameProfile(c, 0)
+					if err != nil {
+						return
+					}
+					if !bytes.HasPrefix(payload, []byte("CIPERFUP ")) {
+						_ = protocol.WriteResponseFrame(c, id, []byte("ERR unsupported"))
+						continue
+					}
+					parts := bytes.SplitN(payload, []byte(" "), 4)
+					if len(parts) != 4 {
+						_ = protocol.WriteResponseFrame(c, id, []byte("ERR bad upload"))
+						continue
+					}
+					size, _ := strconv.Atoi(string(parts[2]))
+					if size == transientChunk && transient.CompareAndSwap(0, 1) {
+						_ = protocol.WriteResponseFrame(c, id, []byte("ERR transient"))
+						return
+					}
+					if size > threshold || len(parts[3]) != size || !bytes.Equal(parts[3], calibrationPattern(size)) {
+						_ = protocol.WriteResponseFrame(c, id, []byte("ERR too large"))
+						continue
+					}
+					_ = protocol.WriteResponseFrame(c, id, []byte("IPERFOK"))
+				}
+			}(conn)
+		}
+	}()
+
+	opts := Options{
+		startSize:  16 * 1024,
+		minSize:    32,
+		maxSize:    16 * 1024,
+		txnTimeout: time.Second,
+	}
+	got := calibrateMaximum(ln.Addr().String(), "", opts, false, 32)
+	if transient.Load() != 1 {
+		t.Fatalf("transient failure count=%d, want 1", transient.Load())
+	}
+	if got < transientChunk {
+		t.Fatalf("X calibration collapsed below transiently failed %d-byte probe: got %d", transientChunk, got)
+	}
+	if got > threshold || threshold-got > 32 {
+		t.Fatalf("X calibrated size=%d, want within 32 bytes below threshold=%d", got, threshold)
+	}
+}
+
+func TestWithCalibratedChunksLocksIndependentXSizes(t *testing.T) {
+	opts := Options{minSize: 32, maxSize: 1024 * 1024, startSize: 1024 * 1024, adaptive: true}
+	opts = opts.WithCalibratedChunks(900000, 500000)
+	if opts.adaptive {
+		t.Fatal("X runtime adaptation remained enabled after calibration")
+	}
+
+	up := newAdaptiveSizer("upload", opts.uploadStartSize, opts.uploadMaxSize, opts)
+	down := newAdaptiveSizer("download", opts.downloadStartSize, opts.downloadMaxSize, opts)
+	if up.Current() != 900000 || up.max != 900000 {
+		t.Fatalf("upload current/max=%d/%d, want 900000", up.Current(), up.max)
+	}
+	if down.Current() != 500000 || down.max != 500000 {
+		t.Fatalf("download current/max=%d/%d, want 500000", down.Current(), down.max)
+	}
+
+	// The calibrated sizes are immutable during the X session. Neither a
+	// transport failure nor a long run of successes may move them.
+	if old, next := up.Failure(900000); old != 900000 || next != 900000 {
+		t.Fatalf("upload failure changed calibrated chunk: %d -> %d", old, next)
+	}
+	for i := 0; i < 1000; i++ {
+		up.Success(900000)
+		down.Success(500000)
+	}
+	if up.Current() != 900000 {
+		t.Fatalf("upload success changed calibrated chunk to %d", up.Current())
+	}
+	if old, next := down.Failure(500000); old != 500000 || next != 500000 {
+		t.Fatalf("download failure changed calibrated chunk: %d -> %d", old, next)
+	}
+	if down.Current() != 500000 {
+		t.Fatalf("download calibrated chunk changed to %d", down.Current())
+	}
+}