[tunnel] Add multi-ping edge selection with outlier removal

Enhance the tunnel endpoint latency selector to run multiple pings per
endpoint and discard outliers before selecting the best edge.

- Add pingsPerEndpoint option (default: 3) to LatencySelector
- Refactor probe() to reuse QUIC connection across multiple pings
- Implement trimmed mean aggregation (discards high/low values)
- Add comprehensive tests for aggregateLatencies function

This reduces susceptibility to noisy measurements from transient
network conditions when selecting tunnel endpoints.

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

Author: dilyevsky

pkg/cmd/tunnel/run.go

@@ -149,7 +149,9 @@ var tunnelRunCmd = &cobra.Command{
 		if err != nil {
 			return fmt.Errorf("unable to parse endpoint selection strategy: %w", err)
 		}
-		selectorOpts := []endpointselect.Option{}
+		selectorOpts := []endpointselect.Option{
+			endpointselect.WithPingsPerEndpoint(3),
+		}
 		if insecureSkipVerify {
 			selectorOpts = append(selectorOpts, endpointselect.WithInsecureSkipVerify(true))
 		}

pkg/tunnel/endpointselect/latency.go

@@ -21,15 +21,18 @@ const (
 	DefaultProbeTimeout = 3 * time.Second
 	// DefaultMaxConcurrent is the default maximum number of concurrent probes.
 	DefaultMaxConcurrent = 10
+	// DefaultPingsPerEndpoint is the default number of ping requests per endpoint.
+	DefaultPingsPerEndpoint = 3
 )
 
 // Option configures a LatencySelector.
 type Option func(*latencyOptions)
 
 type latencyOptions struct {
-	probeTimeout  time.Duration
-	maxConcurrent int
-	insecureSkip  bool
+	probeTimeout     time.Duration
+	maxConcurrent    int
+	insecureSkip     bool
+	pingsPerEndpoint int
 }
 
 // WithProbeTimeout sets the timeout for each endpoint probe.
@@ -53,6 +56,17 @@ func WithInsecureSkipVerify(skip bool) Option {
 	}
 }
 
+// WithPingsPerEndpoint sets the number of ping requests per endpoint.
+// The latencies are aggregated using a trimmed mean (removing outliers).
+func WithPingsPerEndpoint(n int) Option {
+	return func(o *latencyOptions) {
+		if n < 1 {
+			n = 1
+		}
+		o.pingsPerEndpoint = n
+	}
+}
+
 // LatencySelector selects endpoints based on QUIC handshake latency.
 type LatencySelector struct {
 	opts latencyOptions
@@ -61,8 +75,9 @@ type LatencySelector struct {
 // NewLatencySelector creates a new LatencySelector.
 func NewLatencySelector(opts ...Option) *LatencySelector {
 	options := latencyOptions{
-		probeTimeout:  DefaultProbeTimeout,
-		maxConcurrent: DefaultMaxConcurrent,
+		probeTimeout:     DefaultProbeTimeout,
+		maxConcurrent:    DefaultMaxConcurrent,
+		pingsPerEndpoint: DefaultPingsPerEndpoint,
 	}
 	for _, opt := range opts {
 		opt(&options)
@@ -153,17 +168,9 @@ func (s *LatencySelector) probeAll(ctx context.Context, endpoints []string) []Pr
 	return results
 }
 
-// probe measures the round-trip latency to a single endpoint by making
-// an HTTP/3 request to the /ping endpoint.
-func (s *LatencySelector) probe(ctx context.Context, addr string) ProbeResult {
-	result := ProbeResult{
-		Addr:     addr,
-		ProbedAt: time.Now(),
-	}
-
-	probeCtx, cancel := context.WithTimeout(ctx, s.opts.probeTimeout)
-	defer cancel()
-
+// dialEndpoint establishes a QUIC connection and creates an HTTP/3 client connection.
+// Returns the QUIC connection, HTTP/3 client connection, and any error.
+func (s *LatencySelector) dialEndpoint(ctx context.Context, addr string) (quic.Connection, *http3.ClientConn, error) {
 	// Extract hostname from address for TLS ServerName.
 	serverName := "proxy"
 	if host, _, err := net.SplitHostPort(addr); err == nil && net.ParseIP(host) == nil {
@@ -181,55 +188,130 @@ func (s *LatencySelector) probe(ctx context.Context, addr string) ProbeResult {
 		InitialPacketSize: 1350,
 	}
 
-	start := time.Now()
-
 	// Dial QUIC connection.
-	qConn, err := quic.DialAddr(probeCtx, addr, tlsConfig, quicConfig)
+	qConn, err := quic.DialAddr(ctx, addr, tlsConfig, quicConfig)
 	if err != nil {
-		result.Error = err
-		slog.Debug("Endpoint probe failed (QUIC dial)",
-			slog.String("addr", addr),
-			slog.Any("error", err))
-		return result
+		return nil, nil, err
 	}
-	defer qConn.CloseWithError(0, "probe complete")
 
-	// Make HTTP/3 request to /ping endpoint.
+	// Create HTTP/3 client connection.
 	tr := &http3.Transport{EnableDatagrams: true}
 	hConn := tr.NewClientConn(qConn)
 
-	req, err := http.NewRequestWithContext(probeCtx, "GET", "https://proxy/ping", nil)
+	return qConn, hConn, nil
+}
+
+// pingSingle performs a single HTTP/3 GET /ping request over an existing connection.
+// Returns the round-trip latency or an error.
+func (s *LatencySelector) pingSingle(ctx context.Context, hConn *http3.ClientConn) (time.Duration, error) {
+	req, err := http.NewRequestWithContext(ctx, "GET", "https://proxy/ping", nil)
 	if err != nil {
-		result.Error = err
-		slog.Debug("Endpoint probe failed (request creation)",
-			slog.String("addr", addr),
-			slog.Any("error", err))
-		return result
+		return 0, err
 	}
 
+	start := time.Now()
 	resp, err := hConn.RoundTrip(req)
+	if err != nil {
+		return 0, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return 0, fmt.Errorf("ping returned status %d", resp.StatusCode)
+	}
+
+	return time.Since(start), nil
+}
+
+// aggregateLatencies computes the trimmed mean of latencies by discarding
+// the highest and lowest values (if there are enough samples) and averaging the rest.
+// For 1-2 samples, returns the median. For 3+ samples, discards extremes.
+func aggregateLatencies(pings []time.Duration) time.Duration {
+	if len(pings) == 0 {
+		return 0
+	}
+	if len(pings) == 1 {
+		return pings[0]
+	}
+
+	// Sort the pings.
+	sorted := make([]time.Duration, len(pings))
+	copy(sorted, pings)
+	sort.Slice(sorted, func(i, j int) bool {
+		return sorted[i] < sorted[j]
+	})
+
+	if len(sorted) == 2 {
+		// Return median (average of two).
+		return (sorted[0] + sorted[1]) / 2
+	}
+
+	// Discard highest and lowest, average the rest.
+	trimmed := sorted[1 : len(sorted)-1]
+	var sum time.Duration
+	for _, d := range trimmed {
+		sum += d
+	}
+	return sum / time.Duration(len(trimmed))
+}
+
+// probe measures the round-trip latency to a single endpoint by making
+// multiple HTTP/3 requests to the /ping endpoint and aggregating the results.
+func (s *LatencySelector) probe(ctx context.Context, addr string) ProbeResult {
+	result := ProbeResult{
+		Addr:     addr,
+		ProbedAt: time.Now(),
+	}
+
+	probeCtx, cancel := context.WithTimeout(ctx, s.opts.probeTimeout)
+	defer cancel()
+
+	// Establish connection.
+	qConn, hConn, err := s.dialEndpoint(probeCtx, addr)
 	if err != nil {
 		result.Error = err
-		slog.Debug("Endpoint probe failed (HTTP/3 request)",
+		slog.Debug("Endpoint probe failed (QUIC dial)",
 			slog.String("addr", addr),
 			slog.Any("error", err))
 		return result
 	}
-	defer resp.Body.Close()
+	defer qConn.CloseWithError(0, "probe complete")
 
-	if resp.StatusCode != http.StatusOK {
-		result.Error = fmt.Errorf("ping returned status %d", resp.StatusCode)
-		slog.Debug("Endpoint probe failed (bad status)",
+	// Perform multiple pings and collect latencies.
+	var pings []time.Duration
+	for i := 0; i < s.opts.pingsPerEndpoint; i++ {
+		latency, err := s.pingSingle(probeCtx, hConn)
+		if err != nil {
+			slog.Debug("Endpoint ping failed",
+				slog.String("addr", addr),
+				slog.Int("ping", i+1),
+				slog.Any("error", err))
+			// Continue to collect as many pings as possible.
+			continue
+		}
+		pings = append(pings, latency)
+		slog.Debug("Endpoint ping succeeded",
 			slog.String("addr", addr),
-			slog.Int("status", resp.StatusCode))
+			slog.Int("ping", i+1),
+			slog.Duration("latency", latency))
+	}
+
+	// If no pings succeeded, return an error.
+	if len(pings) == 0 {
+		result.Error = fmt.Errorf("all %d pings failed", s.opts.pingsPerEndpoint)
+		slog.Debug("Endpoint probe failed (all pings failed)",
+			slog.String("addr", addr))
 		return result
 	}
 
-	result.Latency = time.Since(start)
+	// Aggregate latencies using trimmed mean.
+	result.Latency = aggregateLatencies(pings)
 
 	slog.Debug("Endpoint probe succeeded",
 		slog.String("addr", addr),
-		slog.Duration("latency", result.Latency))
+		slog.Int("successful_pings", len(pings)),
+		slog.Int("total_pings", s.opts.pingsPerEndpoint),
+		slog.Duration("aggregated_latency", result.Latency))
 
 	return result
 }

pkg/tunnel/endpointselect/selector_test.go

@@ -107,10 +107,88 @@ func TestLatencySelector_Select(t *testing.T) {
 			WithProbeTimeout(500*time.Millisecond),
 			WithMaxConcurrent(5),
 			WithInsecureSkipVerify(true),
+			WithPingsPerEndpoint(5),
 		)
 		assert.Equal(t, 500*time.Millisecond, s.opts.probeTimeout)
 		assert.Equal(t, 5, s.opts.maxConcurrent)
 		assert.True(t, s.opts.insecureSkip)
+		assert.Equal(t, 5, s.opts.pingsPerEndpoint)
+	})
+
+	t.Run("default pings per endpoint is 3", func(t *testing.T) {
+		s := NewLatencySelector()
+		assert.Equal(t, 3, s.opts.pingsPerEndpoint)
+	})
+
+	t.Run("pings per endpoint minimum is 1", func(t *testing.T) {
+		s := NewLatencySelector(WithPingsPerEndpoint(0))
+		assert.Equal(t, 1, s.opts.pingsPerEndpoint)
+
+		s = NewLatencySelector(WithPingsPerEndpoint(-5))
+		assert.Equal(t, 1, s.opts.pingsPerEndpoint)
+	})
+}
+
+func TestAggregateLatencies(t *testing.T) {
+	t.Run("empty slice returns zero", func(t *testing.T) {
+		result := aggregateLatencies([]time.Duration{})
+		assert.Equal(t, time.Duration(0), result)
+	})
+
+	t.Run("single value returns that value", func(t *testing.T) {
+		result := aggregateLatencies([]time.Duration{100 * time.Millisecond})
+		assert.Equal(t, 100*time.Millisecond, result)
+	})
+
+	t.Run("two values returns average", func(t *testing.T) {
+		result := aggregateLatencies([]time.Duration{
+			100 * time.Millisecond,
+			200 * time.Millisecond,
+		})
+		assert.Equal(t, 150*time.Millisecond, result)
+	})
+
+	t.Run("three values returns median (middle value)", func(t *testing.T) {
+		// With 3 pings: discards high and low, returns the one remaining (median).
+		result := aggregateLatencies([]time.Duration{
+			100 * time.Millisecond, // low - discarded
+			150 * time.Millisecond, // middle - kept
+			300 * time.Millisecond, // high - discarded
+		})
+		assert.Equal(t, 150*time.Millisecond, result)
+	})
+
+	t.Run("three values in different order returns same median", func(t *testing.T) {
+		// Verify sorting works correctly.
+		result := aggregateLatencies([]time.Duration{
+			300 * time.Millisecond, // high - discarded
+			100 * time.Millisecond, // low - discarded
+			150 * time.Millisecond, // middle - kept
+		})
+		assert.Equal(t, 150*time.Millisecond, result)
+	})
+
+	t.Run("five values discards outliers and averages middle three", func(t *testing.T) {
+		result := aggregateLatencies([]time.Duration{
+			50 * time.Millisecond,  // low - discarded
+			100 * time.Millisecond, // kept
+			150 * time.Millisecond, // kept
+			200 * time.Millisecond, // kept
+			500 * time.Millisecond, // high - discarded
+		})
+		// Average of 100, 150, 200 = 450/3 = 150ms
+		assert.Equal(t, 150*time.Millisecond, result)
+	})
+
+	t.Run("removes outliers from noisy measurements", func(t *testing.T) {
+		// Simulate realistic scenario: 2 normal pings and 1 outlier.
+		result := aggregateLatencies([]time.Duration{
+			25 * time.Millisecond,   // normal
+			30 * time.Millisecond,   // normal
+			500 * time.Millisecond,  // outlier (network blip)
+		})
+		// Should return 30ms (middle value after sorting: 25, 30, 500).
+		assert.Equal(t, 30*time.Millisecond, result)
 	})
 }