Series B — Adaptive per-resolver duplication

internal/client/async_runtime.go

@@ -544,6 +544,21 @@ func (c *Client) asyncPlanEncodeWorker(ctx context.Context, id int) {
 			if targetCount < 1 {
 				targetCount = 1
 			}
+			// Series B / U1.5: apply adaptive per-resolver duplication. The
+			// balancer returns `targetCount` unchanged for setup-class and
+			// non-stream-data traffic, so SetupPacketDuplicationCount keeps
+			// flooring those types. For data-like stream traffic the count
+			// is clamped to [MIN_DUP, MAX_DUP] based on the preferred
+			// resolver's LossScore (sustained-healthy window enforces
+			// conservatism on decreases). Wire site is the planner — the
+			// dispatcher continues to only forward dupCount.
+			targetCount = c.balancer.AdaptiveDuplicationCount(
+				task.opts.PacketType,
+				task.opts.StreamID,
+				targetCount,
+				c.cfg.MinDup,
+				c.cfg.MaxDup,
+			)
 
 			var (
 				conns []Connection

internal/client/balancer.go

@@ -11,6 +11,7 @@ package client
 
 import (
 	"encoding/binary"
+	"math"
 	"net"
 	"sync"
 	"sync/atomic"
@@ -51,6 +52,17 @@ type balancerStreamRouteState struct {
 	PreferredResolverKey string
 	ResendStreak         int
 	LastFailoverAt       time.Time
+
+	// Adaptive duplication state (Series B / U1.5). AdaptiveLastDup is the
+	// last effective dup count returned by AdaptiveDuplicationCount;
+	// AdaptiveHealthySinceAt is the wall-clock instant since which the
+	// preferred resolver's lossPermille has stayed below
+	// adaptiveDupHealthyPermille. Zero AdaptiveLastDup means the stream
+	// has not been sampled yet; zero AdaptiveHealthySinceAt means the
+	// healthy window has been reset (loss was observed). Both are mutated
+	// under mu while the balancer's outer mu.RLock is held by the caller.
+	AdaptiveLastDup        int
+	AdaptiveHealthySinceAt time.Time
 }
 
 type balancerResolverSampleKey struct {
@@ -105,8 +117,24 @@ type Balancer struct {
 	autoDisableTimeoutWindow time.Duration
 	onResolverDisabled       func(*Connection, string)
 	confirmResolverDown      func(*Connection, time.Duration) bool
+
+	// nowFunc abstracts time.Now for AdaptiveDuplicationCount so tests can
+	// drive the sustained-healthy window deterministically. Defaults to
+	// time.Now in NewBalancer.
+	nowFunc func() time.Time
 }
 
+// Adaptive-duplication tunables (Series B / U1.5). The sustained-healthy
+// window is intentionally LONGER than the default streamFailoverCooldown
+// (1 second, see NewBalancer) so the two hysteresis mechanisms don't
+// double-damp the same loss signal: a stream failover may already have
+// rotated the preferred resolver before the adaptive guard decides it is
+// safe to drop the dup count back down.
+const (
+	adaptiveDupHealthyPermille uint64        = 50              // ≤ 5% loss is "healthy"
+	adaptiveDupHealthyWindow   time.Duration = 3 * time.Second // > default streamFailoverCooldown
+)
+
 type connectionStats struct {
 	sent            atomic.Uint64
 	acked           atomic.Uint64
@@ -129,6 +157,7 @@ func NewBalancer(strategy int, log *logger.Logger) *Balancer {
 		streamRoutes:            make(map[uint16]*balancerStreamRouteState),
 		streamFailoverThreshold: 1,
 		streamFailoverCooldown:  time.Second,
+		nowFunc:                 time.Now,
 	}
 	for i := range b.pendingShards {
 		b.pendingShards[i].pending = make(map[balancerResolverSampleKey]balancerResolverSample)
@@ -137,6 +166,28 @@ func NewBalancer(strategy int, log *logger.Logger) *Balancer {
 	return b
 }
 
+// SetNowFunc swaps the balancer's clock for adaptive-duplication tests. Pass
+// nil to restore time.Now. Not safe to call concurrently with planner traffic;
+// it is intended for test setup only.
+func (b *Balancer) SetNowFunc(fn func() time.Time) {
+	if b == nil {
+		return
+	}
+	if fn == nil {
+		fn = time.Now
+	}
+	b.mu.Lock()
+	b.nowFunc = fn
+	b.mu.Unlock()
+}
+
+func (b *Balancer) now() time.Time {
+	if b == nil || b.nowFunc == nil {
+		return time.Now()
+	}
+	return b.nowFunc()
+}
+
 func (b *Balancer) SetStreamFailoverConfig(threshold int, cooldown time.Duration) {
 	if b == nil {
 		return
@@ -2044,6 +2095,153 @@ func (b *Balancer) lossScoreLocked(idx int) uint64 {
 	return (lost * 1000) / sent
 }
 
+// LossScore returns the per-resolver loss score in permille (parts per
+// thousand) for the resolver identified by serverKey. The returned shape is a
+// uint64 permille value in the range [0, 1000]: 0 means no loss observed,
+// higher values mean more loss. The neutral / initial-probation value is 200,
+// returned when the resolver is unknown or has fewer than 5 sent samples
+// (matches the existing locked helper).
+//
+// This exposes lossScoreLocked through a key-based public API so that callers
+// outside balancer.go (e.g. the planner in async_runtime.go) can read the loss
+// signal without taking the balancer's lock manually. Adaptive duplication
+// (see B3) consumes this score as the lossRate input via the conversion
+//
+//	lossRate := float64(LossScore(key)) / 1000.0
+//
+// B1 and B3 must use this identifier; do not introduce a parallel "LossRate"
+// shape.
+func (b *Balancer) LossScore(serverKey string) uint64 {
+	if b == nil {
+		return 200
+	}
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	idx, ok := b.indexByKey[serverKey]
+	if !ok {
+		return 200
+	}
+	return b.lossScoreLocked(idx)
+}
+
+// AdaptiveDuplicationCount is the planner-side helper that converts a baseline
+// duplication count into the adaptive per-resolver value before it is handed
+// to SelectTargets. It implements Series B / U1.5.
+//
+// Setup-class boundary: setup-class packets (PACKET_STREAM_SYN,
+// PACKET_PACKED_CONTROL_BLOCKS, PACKET_SOCKS5_SYN, PACKET_STREAM_CLOSE_READ,
+// PACKET_STREAM_CLOSE_WRITE) and any non-stream-data-like packet bypass this
+// function entirely — they keep `base`, which the caller already set via
+// runtimePacketDuplicationCount() so SetupPacketDuplicationCount continues to
+// floor the count. Adaptive duplication applies only to data-like stream
+// traffic (PACKET_STREAM_DATA, PACKET_STREAM_RESEND) on a non-zero stream.
+//
+// Formula (B1 + B2):
+//
+//	target = clamp(round(MIN_DUP + (MAX_DUP - MIN_DUP) * lossRate),
+//	               MIN_DUP, MAX_DUP)
+//
+// where lossRate = float64(LossScore(preferredResolver)) / 1000.0 — the same
+// LossScore identifier introduced in B1.
+//
+// Conservatism: dup counts are only allowed to DECREASE after the preferred
+// resolver's lossPermille has stayed at or below adaptiveDupHealthyPermille
+// for at least adaptiveDupHealthyWindow of wall-clock time. Increases apply
+// immediately (duplication-for-survivability under bursty loss). The window
+// is longer than the default streamFailoverCooldown so adaptive duplication
+// and the existing stream-failover hysteresis do not double-damp the same
+// signal.
+func (b *Balancer) AdaptiveDuplicationCount(packetType uint8, streamID uint16, base, minDup, maxDup int) int {
+	if b == nil {
+		return base
+	}
+	if streamID == 0 || !isBalancerStreamDataLike(packetType) {
+		return base
+	}
+	if base < 1 {
+		base = 1
+	}
+	if minDup < 1 {
+		minDup = 1
+	}
+	if maxDup < minDup {
+		maxDup = minDup
+	}
+
+	b.mu.RLock()
+	state := b.streamRoutes[streamID]
+	var preferredKey string
+	if state != nil {
+		state.mu.Lock()
+		preferredKey = state.PreferredResolverKey
+		state.mu.Unlock()
+	}
+	var lossPermille uint64 = 200
+	if preferredKey != "" {
+		if idx, ok := b.indexByKey[preferredKey]; ok {
+			lossPermille = b.lossScoreLocked(idx)
+		}
+	}
+	b.mu.RUnlock()
+
+	span := float64(maxDup - minDup)
+	lossRate := float64(lossPermille) / 1000.0
+	if lossRate > 1 {
+		lossRate = 1
+	}
+	target := int(math.Round(float64(minDup) + span*lossRate))
+	if target < minDup {
+		target = minDup
+	} else if target > maxDup {
+		target = maxDup
+	}
+
+	if minDup == maxDup {
+		return target
+	}
+	if state == nil {
+		return target
+	}
+
+	state.mu.Lock()
+	defer state.mu.Unlock()
+
+	now := b.now()
+	isHealthy := lossPermille <= adaptiveDupHealthyPermille
+
+	if state.AdaptiveLastDup == 0 {
+		state.AdaptiveLastDup = target
+		if isHealthy {
+			state.AdaptiveHealthySinceAt = now
+		} else {
+			state.AdaptiveHealthySinceAt = time.Time{}
+		}
+		return state.AdaptiveLastDup
+	}
+
+	if isHealthy {
+		if state.AdaptiveHealthySinceAt.IsZero() {
+			state.AdaptiveHealthySinceAt = now
+		}
+	} else {
+		state.AdaptiveHealthySinceAt = time.Time{}
+	}
+
+	if target > state.AdaptiveLastDup {
+		state.AdaptiveLastDup = target
+		return state.AdaptiveLastDup
+	}
+	if target < state.AdaptiveLastDup {
+		if !state.AdaptiveHealthySinceAt.IsZero() && now.Sub(state.AdaptiveHealthySinceAt) >= adaptiveDupHealthyWindow {
+			state.AdaptiveLastDup = target
+			// Reset the window so we don't continue dropping on every
+			// subsequent sample without re-validating sustained health.
+			state.AdaptiveHealthySinceAt = now
+		}
+	}
+	return state.AdaptiveLastDup
+}
+
 func (b *Balancer) latencyScoreLocked(idx int) uint64 {
 	if idx < 0 || idx >= len(b.stats) || b.stats[idx] == nil {
 		return 999000