feat(discovery): enhance MOLS traffic distribution with P2C pressure optimization and stickiness

Author: gg582

portal/discovery/mols.go

@@ -35,6 +35,7 @@ const (
 	molsFallbackRTTThreshold   = 2 * time.Second
 	molsMinActiveNodes         = 2
 	defaultMaxActiveRelays     = 3
+	molsP2CPressureDelta       = 0.3
 )
 
 func molsScore(i, j, m1, m2, order int) int {
@@ -105,7 +106,8 @@ func molsCongestionScore(i, j, m1, m2, order int) int {
 	return (order*order + 1) - molsScore(i, (order-1)-j, m1, m2, order)
 }
 
-// hashToGridIndex maps an identity string to a stable FNV-1a hash. Callers
+// hashToGridIndex maps an identity string to a stable FNV-1a hash with a 2nd-stage
+// bit-mixing cascade (avalanche diffusion to eliminate clustering). Callers
 // fold it into the current grid order with % order; the folded index is not
 // stable across orders, so it is recomputed whenever the pool size changes.
 func hashToGridIndex(s string) uint32 {
@@ -114,6 +116,11 @@ func hashToGridIndex(s string) uint32 {
 		h ^= uint32(s[i])
 		h *= 16777619
 	}
+	h ^= h >> 16
+	h *= 0x85ebca6b
+	h ^= h >> 13
+	h *= 0xc2b2ae35
+	h ^= h >> 16
 	return h
 }
 
@@ -286,17 +293,33 @@ func RankRelayPool(autoPool []RelayState, localAddress string) []string {
 			return 0
 		})
 
-		tierOut := make([]string, 0, len(candidates))
+		var nonSaturated []molsCandidate
+		var saturated []molsCandidate
 		for _, candidate := range candidates {
-			if !candidate.state.IsSaturated {
-				tierOut = append(tierOut, candidate.state.Descriptor.APIHTTPSAddr)
+			if candidate.state.IsSaturated {
+				saturated = append(saturated, candidate)
+			} else {
+				nonSaturated = append(nonSaturated, candidate)
 			}
 		}
-		for _, candidate := range candidates {
-			if candidate.state.IsSaturated {
-				tierOut = append(tierOut, candidate.state.Descriptor.APIHTTPSAddr)
+
+		// P2C pressure optimization: If candidate 0 has significantly higher
+		// pressure than candidate 1, swap them to relieve node pressure without herd effect.
+		if len(nonSaturated) >= 2 {
+			p0 := nonSaturated[0].state.Pressure()
+			p1 := nonSaturated[1].state.Pressure()
+			if p0-p1 > molsP2CPressureDelta {
+				nonSaturated[0], nonSaturated[1] = nonSaturated[1], nonSaturated[0]
 			}
 		}
+
+		tierOut := make([]string, 0, len(candidates))
+		for _, candidate := range nonSaturated {
+			tierOut = append(tierOut, candidate.state.Descriptor.APIHTTPSAddr)
+		}
+		for _, candidate := range saturated {
+			tierOut = append(tierOut, candidate.state.Descriptor.APIHTTPSAddr)
+		}
 		return tierOut
 	}
 
@@ -327,8 +350,41 @@ func SelectPriority(states []RelayState, routeState RouteState) []string {
 	if maxActive <= 0 {
 		maxActive = defaultMaxActiveRelays
 	}
-	if len(auto) > maxActive {
-		auto = auto[:maxActive]
-	}
+	auto = applyActiveStickiness(auto, routeState.ActiveRelayURLs, maxActive)
 	return append(explicit, auto...)
 }
+
+// applyActiveStickiness preserves currently active relays that remain in the
+// ranked eligible pool to avoid connection churn.
+func applyActiveStickiness(ranked []string, activeRelayURLs []string, maxActive int) []string {
+	if len(activeRelayURLs) == 0 || len(ranked) <= maxActive {
+		if len(ranked) > maxActive {
+			return ranked[:maxActive]
+		}
+		return ranked
+	}
+	activeSet := make(map[string]struct{}, len(activeRelayURLs))
+	for _, u := range activeRelayURLs {
+		activeSet[u] = struct{}{}
+	}
+	selected := make([]string, 0, maxActive)
+	for _, u := range ranked {
+		if _, isActive := activeSet[u]; isActive {
+			selected = append(selected, u)
+			if len(selected) == maxActive {
+				break
+			}
+		}
+	}
+	if len(selected) < maxActive {
+		for _, u := range ranked {
+			if !slices.Contains(selected, u) {
+				selected = append(selected, u)
+				if len(selected) == maxActive {
+					break
+				}
+			}
+		}
+	}
+	return selected
+}

portal/discovery/mols_test.go

@@ -2,6 +2,7 @@ package discovery
 
 import (
 	"fmt"
+	"slices"
 	"testing"
 	"time"
 
@@ -102,6 +103,77 @@ func TestMOLSSelectPriorityKeepsUnobservedAutoSeed(t *testing.T) {
 	}
 }
 
+func TestHashToGridIndexDistribution(t *testing.T) {
+	const buckets = 7
+	counts := make(map[int]int, buckets)
+	for i := 0; i < 1000; i++ {
+		addr := fmt.Sprintf("192.168.1.%d:8080", i)
+		idx := int(hashToGridIndex(addr) % buckets)
+		counts[idx]++
+	}
+	// Every bucket must receive at least some items without starving
+	for b := 0; b < buckets; b++ {
+		if counts[b] == 0 {
+			t.Errorf("bucket %d received 0 items", b)
+		}
+	}
+}
+
+func TestMOLSP2CPressurePromotion(t *testing.T) {
+	relayA := confirmedRelayState(t, "https://relay-a.example")
+	relayB := confirmedRelayState(t, "https://relay-b.example")
+
+	// relayA: High load momentum and tail inflation (P90=100ms, P50=10ms)
+	relayA.LoadFactor = 0.75
+	relayA.EWMALoad = 0.75
+	relayA.LoadDelta = 0.2
+	for i := 0; i < 90; i++ {
+		relayA.RTTTracker.Add(10 * time.Millisecond)
+	}
+	for i := 0; i < 10; i++ {
+		relayA.RTTTracker.Add(100 * time.Millisecond)
+	}
+
+	// relayB: Low load and uniform RTT (P90=20ms, P50=20ms)
+	relayB.LoadFactor = 0.1
+	relayB.EWMALoad = 0.1
+	for i := 0; i < 100; i++ {
+		relayB.RTTTracker.Add(20 * time.Millisecond)
+	}
+
+	if relayA.Pressure() <= relayB.Pressure()+molsP2CPressureDelta {
+		t.Fatalf("relayA pressure (%.2f) should exceed relayB pressure (%.2f) + delta (%.2f)",
+			relayA.Pressure(), relayB.Pressure(), molsP2CPressureDelta)
+	}
+}
+
+func TestMOLSSelectPriorityActiveStickiness(t *testing.T) {
+	relays := make([]RelayState, 10)
+	for i := range relays {
+		relays[i] = confirmedRelayState(t, fmt.Sprintf("https://relay-stick-%d.example", i))
+	}
+
+	// First selection without active relays
+	firstPick := SelectPriority(relays, RouteState{MaxActiveRelays: 2})
+	if len(firstPick) != 2 {
+		t.Fatalf("len(firstPick) = %d, want 2", len(firstPick))
+	}
+
+	// Suppose relay 9 was currently connected and is healthy
+	activeRelay := "https://relay-stick-9.example"
+	secondPick := SelectPriority(relays, RouteState{
+		ActiveRelayURLs: []string{activeRelay},
+		MaxActiveRelays: 2,
+	})
+
+	if len(secondPick) != 2 {
+		t.Fatalf("len(secondPick) = %d, want 2", len(secondPick))
+	}
+	if !slices.Contains(secondPick, activeRelay) {
+		t.Fatalf("secondPick %v should contain activeRelay %q due to stickiness", secondPick, activeRelay)
+	}
+}
+
 func BenchmarkMOLSRankRelayPool(b *testing.B) {
 	localAddr := "test-client-address"
 	relays := make([]RelayState, 100)

portal/discovery/relayset.go

@@ -491,9 +491,7 @@ func (s *RelaySet) PlanRoutes(explicitPath []string, routeState RouteState) ([]R
 	if maxActive <= 0 {
 		maxActive = defaultMaxActiveRelays
 	}
-	if len(ranked) > maxActive {
-		ranked = ranked[:maxActive]
-	}
+	ranked = applyActiveStickiness(ranked, routeState.ActiveRelayURLs, maxActive)
 	routes := make([]Route, 0, len(ranked)+len(routeState.ExplicitRelayURLs))
 	for _, relayURL := range routeState.ExplicitRelayURLs {
 		eligible := true

portal/discovery/relaystate.go

@@ -200,6 +200,22 @@ func (state *RelayState) EvaluateSaturation() {
 	state.IsSaturated = saturated == 1
 }
 
+// Pressure computes the normalized pressure index using tail latency ratio
+// (P90/P50 inflation) and load momentum (EWMALoad + beta * LoadDelta).
+func (state RelayState) Pressure() float64 {
+	p50 := float64(state.RTTTracker.Get(0.50))
+	p90 := float64(state.RTTTracker.Get(0.90))
+
+	var tailInflation float64
+	if p50 > 0 && p90 > p50 {
+		tailInflation = (p90 - p50) / p50
+	}
+
+	const beta = 0.5
+	loadMomentum := state.EWMALoad + (beta * state.LoadDelta)
+	return tailInflation + loadMomentum
+}
+
 func (state *RelayState) UpdateEWMARTT(newRTT time.Duration) {
 	const alpha = 0.3
 	state.RTTDelta = absDuration(newRTT - state.DiscoveryRTT)
@@ -252,6 +268,9 @@ func (state RelayState) eligibleForMultiHop(routeState RouteState, now time.Time
 
 type RouteState struct {
 	ExplicitRelayURLs []string
+	// ActiveRelayURLs holds currently active connected relay URLs to enable
+	// connection-level stickiness and prevent listener churn during ranking updates.
+	ActiveRelayURLs []string
 	// MaxActiveRelays caps auto-selected listener entries. Zero or negative
 	// values use the selection default of 3. Multi-hop paths may use further
 	// eligible relays as non-entry hops.

sdk/expose.go

@@ -706,6 +706,7 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
 	cfg := e.Config()
 	routes, err := e.relaySet.PlanRoutes(append([]string(nil), cfg.MultiHop...), discovery.RouteState{
 		ExplicitRelayURLs: append([]string(nil), cfg.RelayURLs...),
+		ActiveRelayURLs:   e.ActiveRelayURLs(),
 		MaxActiveRelays:   cfg.MaxActiveRelays,
 		MultiHopDepth:     cfg.MultiHopDepth,
 		RequireUDP:        cfg.UDPEnabled,