discovery: enhance MOLS selection with tiered policy, epoch salt, and virtual latency

Author: gg582

cmd/portal-loadtest/main.go

@@ -64,7 +64,7 @@ func main() {
 	picks := make(map[string]int, *relays) // relay URL → count of clients that picked it first
 	for i := 0; i < *clients; i++ {
 		localAddr := fmt.Sprintf("synthetic-client-%d", i)
-		outputURLs := discovery.RankRelayPool(relayStates, localAddr)
+		outputURLs := discovery.RankRelayPool(relayStates, localAddr, 0)
 		if len(outputURLs) == 0 {
 			// All relays were filtered; skip this client.
 			continue

portal/discovery/mols.go

@@ -21,6 +21,7 @@ import (
 	"cmp"
 	"math"
 	"slices"
+	"strconv"
 	"time"
 )
 
@@ -31,7 +32,7 @@ const (
 	molsVariantM2 uint8 = 11
 
 	molsCongestionRTTThreshold = 500 * time.Millisecond
-	molsCVThreshold            = 0.5
+	molsCVThreshold            = 0.6
 	molsFallbackRTTThreshold   = 2 * time.Second
 	molsMinActiveNodes         = 2
 	defaultMaxActiveRelays     = 3
@@ -132,7 +133,7 @@ func molsRTTStats(states []RelayState) (mean time.Duration, cv float64) {
 			continue
 		}
 		count++
-		sum += float64(state.DiscoveryRTT)
+		sum += float64(state.effectiveRTT())
 	}
 	if count == 0 {
 		return 0, 0
@@ -146,7 +147,7 @@ func molsRTTStats(states []RelayState) (mean time.Duration, cv float64) {
 		if state.DiscoveryRTTAt.IsZero() {
 			continue
 		}
-		d := float64(state.DiscoveryRTT) - avg
+		d := float64(state.effectiveRTT()) - avg
 		sq += d * d
 	}
 	stddev := math.Sqrt(sq / float64(count))
@@ -157,7 +158,7 @@ func molsRTTStats(states []RelayState) (mean time.Duration, cv float64) {
 }
 
 func isRelayFallback(state RelayState) bool {
-	return !state.DiscoveryRTTAt.IsZero() && state.DiscoveryRTT > molsFallbackRTTThreshold
+	return !state.DiscoveryRTTAt.IsZero() && state.effectiveRTT() > molsFallbackRTTThreshold
 }
 
 type molsCandidate struct {
@@ -201,9 +202,9 @@ func selectConfirmed(states []RelayState) []RelayState {
 	return out
 }
 
-// RankRelayPool ranks the autoPool of relay states using MOLS selection for the given local address.
+// RankRelayPool ranks the autoPool of relay states using MOLS selection for the given local address and epoch.
 // The returned slice contains relay URLs ordered by MOLS-derived priority with saturation partitioning.
-func RankRelayPool(autoPool []RelayState, localAddress string) []string {
+func RankRelayPool(autoPool []RelayState, localAddress string, epoch uint64) []string {
 	if len(autoPool) == 0 {
 		return nil
 	}
@@ -214,7 +215,11 @@ func RankRelayPool(autoPool []RelayState, localAddress string) []string {
 
 	order := len(autoPool)
 	m1, m2, _ := molsMultipliers(order, nonLinear)
-	ingressRow := int(hashToGridIndex(localAddress) % uint32(order))
+	ingressKey := localAddress
+	if epoch > 0 {
+		ingressKey = localAddress + "#" + strconv.FormatUint(epoch, 10)
+	}
+	ingressRow := int(hashToGridIndex(ingressKey) % uint32(order))
 
 	type relayHash struct {
 		url  string
@@ -259,10 +264,12 @@ func RankRelayPool(autoPool []RelayState, localAddress string) []string {
 
 	if len(activeStates) < molsMinActiveNodes && len(fallbackStates) > 0 {
 		slices.SortFunc(fallbackStates, func(a, b RelayState) int {
-			if a.DiscoveryRTT < b.DiscoveryRTT {
+			aRTT := a.effectiveRTT()
+			bRTT := b.effectiveRTT()
+			if aRTT < bRTT {
 				return -1
 			}
-			if a.DiscoveryRTT > b.DiscoveryRTT {
+			if aRTT > bRTT {
 				return 1
 			}
 			return 0
@@ -305,14 +312,27 @@ func RankRelayPool(autoPool []RelayState, localAddress string) []string {
 			}
 		}
 
-		// P2C pressure optimization: If candidate 0 has significantly higher
-		// pressure than candidate 1, swap them to relieve node pressure without herd effect.
+		// Pressure-aware partitioning: Candidates with significantly elevated pressure
+		// (pressure difference > molsP2CPressureDelta compared to minimum pressure) are
+		// demoted behind low-pressure candidates so they are pushed outside the MaxActiveRelays
+		// quota, achieving real active-set membership migration.
 		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]
+			minPressure := nonSaturated[0].state.Pressure()
+			for _, c := range nonSaturated[1:] {
+				if p := c.state.Pressure(); p < minPressure {
+					minPressure = p
+				}
 			}
+			var lowPressure []molsCandidate
+			var highPressure []molsCandidate
+			for _, c := range nonSaturated {
+				if c.state.Pressure()-minPressure > molsP2CPressureDelta {
+					highPressure = append(highPressure, c)
+				} else {
+					lowPressure = append(lowPressure, c)
+				}
+			}
+			nonSaturated = append(lowPressure, highPressure...)
 		}
 
 		tierOut := make([]string, 0, len(candidates))
@@ -347,29 +367,55 @@ func SelectPriority(states []RelayState, routeState RouteState) []string {
 		}
 		explicit = append(explicit, relayURL)
 	}
-	auto := RankRelayPool(filterCandidatePool(states, routeState, now, false), routeState.LocalAddress)
+	auto := RankRelayPool(filterCandidatePool(states, routeState, now, false), routeState.LocalAddress, routeState.SelectionEpoch)
 	maxActive := routeState.MaxActiveRelays
 	if maxActive <= 0 {
 		maxActive = defaultMaxActiveRelays
 	}
-	auto = applyActiveStickiness(auto, routeState.ActiveRelayURLs, maxActive)
+	auto = applyActiveStickiness(auto, routeState.ActiveRelayURLs, states, maxActive)
+	if len(auto) > maxActive {
+		auto = auto[: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]
-		}
+// applyActiveStickiness reorders the ranked candidates so that eligible healthy sticky
+// relays occupy the first maxActive positions without dropping remaining pool candidates.
+//
+// Priority cascade:
+//
+//	Layer 1: Sticky relays — preserves currently active connections ONLY if they remain in
+//	         the healthy tier (not saturated and not in fallback) to avoid zombie resurrection.
+//	Layer 2: Warm healthy candidates — fills remaining slots using top-ranked MOLS candidates.
+//	Trailing: Remaining ranked candidates are preserved to support multi-hop path building.
+func applyActiveStickiness(ranked []string, activeRelayURLs []string, states []RelayState, maxActive int) []string {
+	if len(ranked) == 0 || maxActive <= 0 {
+		return ranked
+	}
+	if len(activeRelayURLs) == 0 {
 		return ranked
 	}
+
+	stateMap := make(map[string]RelayState, len(states))
+	for _, s := range states {
+		stateMap[s.Descriptor.APIHTTPSAddr] = s
+	}
+
+	// Stickiness is ONLY granted to healthy nodes: non-saturated and non-fallback.
+	// Degraded/saturated nodes must migrate out rather than being resurrected.
 	activeSet := make(map[string]struct{}, len(activeRelayURLs))
 	for _, u := range activeRelayURLs {
-		activeSet[u] = struct{}{}
+		if s, ok := stateMap[u]; ok {
+			s.EvaluateSaturation()
+			if s.IsSaturated || isRelayFallback(s) {
+				continue
+			}
+			activeSet[u] = struct{}{}
+		}
 	}
-	selected := make([]string, 0, maxActive)
+
+	selected := make([]string, 0, len(ranked))
+	// Layer 1: Retain currently active sticky relays that remain healthy (capped at maxActive)
 	for _, u := range ranked {
 		if _, isActive := activeSet[u]; isActive {
 			selected = append(selected, u)
@@ -378,14 +424,10 @@ func applyActiveStickiness(ranked []string, activeRelayURLs []string, maxActive
 			}
 		}
 	}
-	if len(selected) < maxActive {
-		for _, u := range ranked {
-			if !slices.Contains(selected, u) {
-				selected = append(selected, u)
-				if len(selected) == maxActive {
-					break
-				}
-			}
+	// Layer 2 & Trailing: Append remaining candidates preserving their relative MOLS ranking
+	for _, u := range ranked {
+		if !slices.Contains(selected, u) {
+			selected = append(selected, u)
 		}
 	}
 	return selected

portal/discovery/mols_test.go

@@ -145,6 +145,13 @@ func TestMOLSP2CPressurePromotion(t *testing.T) {
 		t.Fatalf("relayA pressure (%.2f) should exceed relayB pressure (%.2f) + delta (%.2f)",
 			relayA.Pressure(), relayB.Pressure(), molsP2CPressureDelta)
 	}
+
+	// Behavioral assertion: In SelectPriority with MaxActiveRelays=1, low-pressure relayB
+	// must be promoted over high-pressure relayA regardless of initial MOLS order.
+	selected := SelectPriority([]RelayState{relayA, relayB}, RouteState{MaxActiveRelays: 1})
+	if len(selected) != 1 || selected[0] != "https://relay-b.example" {
+		t.Fatalf("SelectPriority with pressure delta = %v, want [%q]", selected, "https://relay-b.example")
+	}
 }
 
 func TestMOLSSelectPriorityActiveStickiness(t *testing.T) {
@@ -174,6 +181,130 @@ func TestMOLSSelectPriorityActiveStickiness(t *testing.T) {
 	}
 }
 
+func TestMOLSSelectPriorityEpochRotation(t *testing.T) {
+	relays := make([]RelayState, 10)
+	for i := range relays {
+		relays[i] = confirmedRelayState(t, fmt.Sprintf("https://relay-epoch-%d.example", i))
+	}
+
+	routeStateEpoch0 := RouteState{LocalAddress: "192.168.1.50:5000", SelectionEpoch: 0}
+	routeStateEpoch1 := RouteState{LocalAddress: "192.168.1.50:5000", SelectionEpoch: 1}
+
+	rank0a := SelectPriority(relays, routeStateEpoch0)
+	rank0b := SelectPriority(relays, routeStateEpoch0)
+	// Deterministic for same epoch
+	if !slices.Equal(rank0a, rank0b) {
+		t.Fatalf("rank0a != rank0b: %v vs %v", rank0a, rank0b)
+	}
+
+	rank1 := SelectPriority(relays, routeStateEpoch1)
+	// Rotation should yield a different primary ranking for non-trivial pool
+	if slices.Equal(rank0a, rank1) {
+		t.Fatalf("rank1 should differ from rank0, got identical %v", rank1)
+	}
+}
+
+func TestMOLSVirtualLatencyPenalty(t *testing.T) {
+	now := time.Now().UTC()
+	relayA := confirmedRelayState(t, "https://relay-a.example")
+	relayA.DiscoveryRTT = 50 * time.Millisecond
+	relayA.DiscoveryRTTAt = now
+	// 7 failures * 300ms = 2.1s penalty => EffectiveRTT = 2.15s (> 2s fallback threshold)
+	relayA.activeFailures = 7
+
+	relayB := confirmedRelayState(t, "https://relay-b.example")
+	relayB.DiscoveryRTT = 100 * time.Millisecond
+	relayB.DiscoveryRTTAt = now
+
+	relayC := confirmedRelayState(t, "https://relay-c.example")
+	relayC.DiscoveryRTT = 120 * time.Millisecond
+	relayC.DiscoveryRTTAt = now
+
+	relays := []RelayState{relayA, relayB, relayC}
+	selected := SelectPriority(relays, RouteState{MaxActiveRelays: 2})
+
+	// relayA should be demoted to fallback tier due to virtual latency, so active picks should be B and C
+	if slices.Contains(selected, "https://relay-a.example") {
+		t.Fatalf("selected %v should not contain relayA in top 2 due to virtual latency penalty", selected)
+	}
+	if len(selected) != 2 {
+		t.Fatalf("len(selected) = %d, want 2", len(selected))
+	}
+}
+
+func TestMOLSStickinessDoesNotResurrectSaturatedOrFallback(t *testing.T) {
+	now := time.Now().UTC()
+	activeSaturated := confirmedRelayState(t, "https://active-sat.example")
+	activeSaturated.IsSaturated = true
+	activeSaturated.LoadFactor = 0.95
+
+	activeFallback := confirmedRelayState(t, "https://active-fb.example")
+	activeFallback.DiscoveryRTT = 3 * time.Second
+	activeFallback.DiscoveryRTTAt = now
+
+	healthyA := confirmedRelayState(t, "https://healthy-a.example")
+	healthyA.DiscoveryRTT = 50 * time.Millisecond
+	healthyA.DiscoveryRTTAt = now
+
+	healthyB := confirmedRelayState(t, "https://healthy-b.example")
+	healthyB.DiscoveryRTT = 60 * time.Millisecond
+	healthyB.DiscoveryRTTAt = now
+
+	relays := []RelayState{activeSaturated, activeFallback, healthyA, healthyB}
+
+	// ActiveRelayURLs includes the saturated and fallback relays.
+	// Stickiness MUST NOT resurrect them over the healthy candidates.
+	selected := SelectPriority(relays, RouteState{
+		ActiveRelayURLs: []string{"https://active-sat.example", "https://active-fb.example"},
+		MaxActiveRelays: 2,
+	})
+
+	if len(selected) != 2 {
+		t.Fatalf("len(selected) = %d, want 2", len(selected))
+	}
+	for _, u := range selected {
+		if u == "https://active-sat.example" || u == "https://active-fb.example" {
+			t.Fatalf("selected %v should not contain demoted relays despite stickiness", selected)
+		}
+	}
+}
+
+func TestMOLSMultiHopEntryStickiness(t *testing.T) {
+	now := time.Now().UTC()
+	set := NewRelaySet(nil)
+	var activeEntry string
+	for i := 0; i < 5; i++ {
+		url := fmt.Sprintf("https://relay-mh-%d.example", i)
+		st := confirmedRelayState(t, url)
+		st.Descriptor.SupportsOverlay = true
+		st.Descriptor.ExpiresAt = now.Add(time.Hour)
+		st.Descriptor.WireGuardPublicKey = "wg-key"
+		st.Descriptor.WireGuardPort = 51820
+		st.LastSeenAt = now
+		set.relays[url] = st
+		if i == 4 {
+			activeEntry = url
+		}
+	}
+
+	routes, err := set.PlanRoutes(nil, RouteState{
+		ActiveRelayURLs: []string{activeEntry},
+		MultiHopDepth:   2,
+		MaxActiveRelays: 2,
+		LocalAddress:    "client",
+	})
+	if err != nil {
+		t.Fatalf("PlanRoutes failed: %v", err)
+	}
+	if len(routes) == 0 {
+		t.Fatalf("routes is empty")
+	}
+	// The first entry hop should preserve the active entry relay due to stickiness
+	if entry := routes[0].ListenerRelayURL(); entry != activeEntry {
+		t.Fatalf("first entry hop = %q, want activeEntry %q due to stickiness", entry, activeEntry)
+	}
+}
+
 func BenchmarkMOLSRankRelayPool(b *testing.B) {
 	localAddr := "test-client-address"
 	relays := make([]RelayState, 100)
@@ -188,7 +319,7 @@ func BenchmarkMOLSRankRelayPool(b *testing.B) {
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		RankRelayPool(relays, localAddr)
+		RankRelayPool(relays, localAddr, 0)
 	}
 }
 

portal/discovery/relayset.go

@@ -481,20 +481,18 @@ func (s *RelaySet) PlanRoutes(explicitPath []string, routeState RouteState) ([]R
 		}
 	}
 
-	ranked := RankRelayPool(filterCandidatePool(states, routeState, now, routeState.MultiHopDepth > 1), routeState.LocalAddress)
-	if routeState.MultiHopDepth > 1 {
-		maxActive := routeState.MaxActiveRelays
-		if maxActive <= 0 {
-			maxActive = defaultMaxActiveRelays
-		}
-		return buildMOLSPaths(ranked, routeState.MultiHopDepth, maxActive)
-	}
-
+	ranked := RankRelayPool(filterCandidatePool(states, routeState, now, routeState.MultiHopDepth > 1), routeState.LocalAddress, routeState.SelectionEpoch)
 	maxActive := routeState.MaxActiveRelays
 	if maxActive <= 0 {
 		maxActive = defaultMaxActiveRelays
 	}
-	ranked = applyActiveStickiness(ranked, routeState.ActiveRelayURLs, maxActive)
+	ranked = applyActiveStickiness(ranked, routeState.ActiveRelayURLs, states, maxActive)
+	if routeState.MultiHopDepth > 1 {
+		return buildMOLSPaths(ranked, routeState.MultiHopDepth, maxActive)
+	}
+	if len(ranked) > maxActive {
+		ranked = ranked[:maxActive]
+	}
 	routes := make([]Route, 0, len(ranked)+len(routeState.ExplicitRelayURLs))
 	for _, relayURL := range routeState.ExplicitRelayURLs {
 		eligible := true