Refactor discovery module: remove obsolete tests, update relay selection logic, and enhance relay state management
Kim committed
May 19, 2026 at 18:29 UTC
469f2be09d74c09b902126f0103bb16095cafd5a
16 files changed
+296
-688
cmd/portal-loadtest/chaos_test.go
deleted
-52
@@ -1,52 +0,0 @@
1
-package main
2
-
3
-import (
4
- "fmt"
5
- "math/rand"
6
- "testing"
7
- "time"
8
-
9
- "github.com/gosuda/portal-tunnel/v2/portal/discovery"
10
- "github.com/gosuda/portal-tunnel/v2/types"
11
-)
12
-
13
-func TestChaosMeshScenario(t *testing.T) {
14
- // Extreme Chaos Mesh: 100 relays, constant churn, massive RTT swings
15
- const numRelays = 100
16
- const rounds = 2000
17
-
18
- relayStates := make([]discovery.RelayState, numRelays)
19
- for i := range relayStates {
20
- relayStates[i] = discovery.RelayState{
21
- Descriptor: types.RelayDescriptor{APIHTTPSAddr: fmt.Sprintf("node-%d", i)},
22
- }
23
- }
24
-
25
- policy := discovery.MOLSRelayPolicy{}
26
- var history []string
27
- var latencies []time.Duration
28
- errors := 0
29
-
30
- start := time.Now()
31
- for r := 0; r < rounds; r++ {
32
- // Random Churn: Add/Remove relays
33
- for i := 0; i < numRelays; i++ {
34
- if rand.Float64() < 0.05 { // 5% churn per round
35
- relayStates[i].DiscoveryRTT = time.Duration(100+rand.Intn(900)) * time.Millisecond
36
- }
37
- }
38
-
39
- cs := discovery.ClientState{LocalAddress: "chaos-client"}
40
- res, _ := policy.SelectPriorityWithTrace(relayStates, cs)
41
-
42
- if len(res) == 0 {
43
- errors++
44
- } else {
45
- history = append(history, res[0])
46
- latencies = append(latencies, 100*time.Millisecond) // Mock
47
- }
48
- }
49
-
50
- m := CalculateMetrics(history, latencies, errors, rounds, time.Since(start))
51
- fmt.Printf("Chaos Mesh Metrics: %+v\n", m)
52
-}
cmd/portal-loadtest/main.go
+10
-11
@@ -1,5 +1,5 @@
1
// Command portal-loadtest is a Phase 1 uniformity probe that measures
2
-// how evenly the MOLS relay-selection policy distributes N synthetic clients
2
+// how evenly MOLS relay selection distributes N synthetic clients
3
// across K synthetic relays. It runs entirely in-process — no running
4
// portal-tunnel server is required.
5
//
@@ -58,15 +58,15 @@ func main() {
58
59
// Build K synthetic relay states. We construct discovery.RelayState values
60
// directly (not via RelaySet.InsertAnnounced) because the public announce
61
- // path requires real EVM-signed descriptors. MOLSRelayPolicy is called
61
+ // path requires real EVM-signed descriptors. Selection functions are called
62
// directly so that no signature gate runs.
63
//
64
// For priority mode: states without an observed descriptor (LastSeenAt zero)
65
- // are accepted into the auto pool by SelectPriorityWithTrace — the
66
- // expiry/protocol gates only fire when hasObservedDescriptor() is true.
65
+ // are accepted into the auto pool by SelectPriority; the expiry/protocol
66
+ // gates only fire when hasObservedDescriptor() is true.
67
//
68
- // For multi-hop mode: SelectMultiHopWithTrace requires hasObservedDescriptor,
69
- // a non-expired ExpiresAt, and HasOverlayPeer()==true. We populate those
68
+ // For multi-hop mode: SelectMultiHop requires hasObservedDescriptor, a
69
+ // non-expired ExpiresAt, and HasOverlayPeer()==true. We populate those
70
// fields with dummy-but-valid values using a far-future ExpiresAt and a
71
// syntactically valid WireGuard public key placeholder.
72
now := time.Now().UTC()
@@ -79,7 +79,7 @@ func main() {
79
},
80
}
81
if mode == "multihop" {
82
- // Populate the fields required by SelectMultiHopWithTrace's eligibility
82
+ // Populate the fields required by SelectMultiHop's eligibility
83
// gates: hasObservedDescriptor (LastSeenAt non-zero), valid ExpiresAt,
84
// and HasOverlayPeer() = SupportsOverlay && WireGuardPublicKey != "" &&
85
// WireGuardPort in [1, 65535].
@@ -96,18 +96,17 @@ func main() {
96
// Generate N synthetic client states with UNIQUE LocalAddress values.
97
// MOLS is deterministic on (LocalAddress, relayURL): duplicate addresses
98
// would make all clients pick identically, falsely appearing as 100% imbalance.
99
- policy := discovery.MOLSRelayPolicy{}
99
picks := make(map[string]int, *relays) // relay URL → count of clients that picked it first
100
for i := 0; i < *clients; i++ {
102
- cs := discovery.ClientState{
101
+ cs := discovery.RouteState{
102
LocalAddress: fmt.Sprintf("synthetic-client-%d", i),
103
MultiHopDepth: *multiHop,
104
}
105
var outputURLs []string
106
if mode == "multihop" {
108
- outputURLs, _ = policy.SelectMultiHopWithTrace(relayStates, cs)
107
+ outputURLs = discovery.SelectMultiHop(relayStates, cs)
108
} else {
110
- outputURLs, _ = policy.SelectPriorityWithTrace(relayStates, cs)
109
+ outputURLs = discovery.SelectPriority(relayStates, cs)
110
}
111
if len(outputURLs) == 0 {
112
// All relays were filtered; skip this client.
cmd/portal-loadtest/runner.go
deleted
-37
@@ -1,37 +0,0 @@
1
-package main
2
-
3
-import (
4
- "sort"
5
- "time"
6
-)
7
-
8
-type Metrics struct {
9
- Oscillations int
10
- P99Latency time.Duration
11
- SelectionTPS float64
12
- MemUsageMB float64
13
- ErrorRate float64
14
-}
15
-
16
-// CalculateMetrics aggregates raw observations into the 5 requested stress metrics.
17
-func CalculateMetrics(history []string, latencies []time.Duration, errors int, totalOps int, duration time.Duration) Metrics {
18
- oscillations := 0
19
- for i := 1; i < len(history); i++ {
20
- if history[i] != history[i-1] {
21
- oscillations++
22
- }
23
- }
24
-
25
- sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] })
26
- p99Idx := int(float64(len(latencies)) * 0.99)
27
- if p99Idx >= len(latencies) {
28
- p99Idx = len(latencies) - 1
29
- }
30
-
31
- return Metrics{
32
- Oscillations: oscillations,
33
- P99Latency: latencies[p99Idx],
34
- SelectionTPS: float64(totalOps) / duration.Seconds(),
35
- ErrorRate: float64(errors) / float64(totalOps),
36
- }
37
-}
cmd/portal-loadtest/stress_test.go
deleted
-93
@@ -1,93 +0,0 @@
1
-package main
2
-
3
-import (
4
- "fmt"
5
- "math/rand"
6
- "testing"
7
- "time"
8
-
9
- "github.com/gosuda/portal-tunnel/v2/portal/discovery"
10
- "github.com/gosuda/portal-tunnel/v2/types"
11
-)
12
-
13
-// TestStressScenarioMessyGrid validates relay selection under high fragmentation
14
-// (53 nodes) and intermittent RTT spikes, tracking priority oscillations.
15
-func TestStressScenarioMessyGrid(t *testing.T) {
16
- const clients = 500
17
- const numRelays = 53
18
-
19
- relayStates := make([]discovery.RelayState, numRelays)
20
- for i := range relayStates {
21
- relayStates[i] = discovery.RelayState{
22
- Descriptor: types.RelayDescriptor{
23
- APIHTTPSAddr: fmt.Sprintf("https://test-relay-%d.example", i),
24
- },
25
- }
26
- for j := 0; j < 100; j++ {
27
- relayStates[i].UpdateEWMARTT(100 * time.Millisecond)
28
- }
29
- }
30
-
31
- policy := discovery.MOLSRelayPolicy{}
32
- var lastTop string
33
- oscillations := 0
34
-
35
- for step := 0; step < 1440; step++ {
36
- for i := 0; i < numRelays; i++ {
37
- if rand.Float64() < 0.3 {
38
- relayStates[i].UpdateEWMARTT(900 * time.Millisecond)
39
- } else {
40
- relayStates[i].UpdateEWMARTT(100 * time.Millisecond)
41
- }
42
- }
43
-
44
- cs := discovery.ClientState{
45
- LocalAddress: fmt.Sprintf("client-%d", rand.Intn(clients)),
46
- }
47
- result, _ := policy.SelectPriorityWithTrace(relayStates, cs)
48
- if len(result) > 0 {
49
- if lastTop != "" && result[0] != lastTop {
50
- oscillations++
51
- }
52
- lastTop = result[0]
53
- }
54
- }
55
- fmt.Printf("Messy Grid Test: Total oscillations=%d\n", oscillations)
56
-}
57
-
58
-// TestStressScenarioMassiveScale validates performance and stability under
59
-// 256-node relay density.
60
-func TestStressScenarioMassiveScale(t *testing.T) {
61
- const clients = 2000
62
- const numRelays = 256
63
-
64
- relayStates := make([]discovery.RelayState, numRelays)
65
- for i := range relayStates {
66
- relayStates[i] = discovery.RelayState{
67
- Descriptor: types.RelayDescriptor{
68
- APIHTTPSAddr: fmt.Sprintf("https://test-relay-%d.example", i),
69
- },
70
- }
71
- for j := 0; j < 100; j++ {
72
- relayStates[i].UpdateEWMARTT(100 * time.Millisecond)
73
- }
74
- }
75
-
76
- policy := discovery.MOLSRelayPolicy{}
77
- start := time.Now()
78
-
79
- for i := 0; i < clients; i++ {
80
- cs := discovery.ClientState{
81
- LocalAddress: fmt.Sprintf("client-%d", i),
82
- }
83
- _, _ = policy.SelectPriorityWithTrace(relayStates, cs)
84
- }
85
-
86
- duration := time.Since(start)
87
- avg := duration / time.Duration(clients)
88
-
89
- fmt.Printf("Massive Scale Test: Avg selection time: %v\n", avg)
90
- if avg > 5*time.Millisecond {
91
- t.Errorf("Massive Scale Test: average selection time %v exceeds limit 5ms", avg)
92
- }
93
-}
portal/discovery/announce.go
+4
-4
@@ -11,8 +11,8 @@ import (
11
// represent multiple relays behind the same NAT or proxy, so the default
12
// receiver budget leaves room for shared egress while still bounding abuse.
13
const (
14
- DefaultAnnounceRatePerMinute = 30
15
- DefaultAnnounceBurst = 60
14
+ defaultAnnounceRatePerMinute = 30
15
+ defaultAnnounceBurst = 60
16
announceLimiterPruneInterval = 10 * time.Minute
17
announceLimiterIdleTTL = 30 * time.Minute
18
)
@@ -49,10 +49,10 @@ type announceBucket struct {
49
// to the defaults so the zero-config path is safe.
50
func NewAnnounceLimiter(ratePerMinute, burst int) *AnnounceLimiter {
51
if ratePerMinute <= 0 {
52
- ratePerMinute = DefaultAnnounceRatePerMinute
52
+ ratePerMinute = defaultAnnounceRatePerMinute
53
}
54
if burst <= 0 {
55
- burst = DefaultAnnounceBurst
55
+ burst = defaultAnnounceBurst
56
}
57
return &AnnounceLimiter{
58
buckets: make(map[string]*announceBucket),
portal/discovery/compare_bench_test.go
deleted
-31
@@ -1,31 +0,0 @@
1
-package discovery
2
-
3
-import (
4
- "testing"
5
- "time"
6
-
7
- "github.com/gosuda/portal-tunnel/v2/types"
8
-)
9
-
10
-func BenchmarkRankRelayPool(b *testing.B) {
11
- localAddr := "test-client-address"
12
- relays := make([]RelayState, 100)
13
- for i := 0; i < 100; i++ {
14
- relays[i] = RelayState{
15
- Descriptor: types.RelayDescriptor{APIHTTPSAddr: "test"},
16
- DiscoveryRTT: 100 * time.Millisecond,
17
- DiscoveryRTTAt: time.Now(),
18
- Confirmed: true,
19
- }
20
- // Add some dummy history
21
- for j := 0; j < 50; j++ {
22
- }
23
- }
24
-
25
- policy := MOLSRelayPolicy{}
26
-
27
- b.ResetTimer()
28
- for i := 0; i < b.N; i++ {
29
- policy.rankRelayPool(relays, localAddr)
30
- }
31
-}
portal/discovery/mols.go
+18
-56
@@ -1,6 +1,6 @@
1
package discovery
2
3
-// MOLSRelayPolicy ranks relays using a GF(64)-based MOLS grid with a
3
+// MOLS selection ranks relays using a GF(64)-based MOLS grid with a
4
// non-invasive adaptive partition over local load telemetry.
5
//
6
// Ordering Pipeline:
@@ -123,8 +123,6 @@ func isRelayFallback(state RelayState) bool {
123
return !state.DiscoveryRTTAt.IsZero() && state.DiscoveryRTT > molsFallbackRTTThreshold
124
}
125
126
-type MOLSRelayPolicy struct{}
127
-
126
type molsCandidate struct {
127
state RelayState
128
score int
@@ -146,7 +144,7 @@ func betterMOLSCandidate(a, b molsCandidate) bool {
144
return a.seq < b.seq
145
}
146
149
-func (p MOLSRelayPolicy) SelectAggregate(states []RelayState) []RelayState {
147
+func selectAggregate(states []RelayState) []RelayState {
148
out := make([]RelayState, 0, len(states))
149
for _, state := range states {
150
if !state.Banned {
@@ -156,7 +154,7 @@ func (p MOLSRelayPolicy) SelectAggregate(states []RelayState) []RelayState {
154
return out
155
}
156
159
-func (p MOLSRelayPolicy) SelectConfirmed(states []RelayState) []RelayState {
157
+func selectConfirmed(states []RelayState) []RelayState {
158
out := make([]RelayState, 0)
159
for _, state := range states {
160
if state.Confirmed {
@@ -166,31 +164,7 @@ func (p MOLSRelayPolicy) SelectConfirmed(states []RelayState) []RelayState {
164
return out
165
}
166
169
-func (p MOLSRelayPolicy) OnActiveConfirmed(state RelayState) RelayState {
170
- state.Confirmed = true
171
- state.activeFailures = 0
172
- state.suppressActiveUntil = time.Time{}
173
- return state
174
-}
175
-
176
-func (p MOLSRelayPolicy) OnUnconfirmed(state RelayState) RelayState {
177
- state.Confirmed = false
178
- return state
179
-}
180
-
181
-func (p MOLSRelayPolicy) OnDiscoveryConfirmed(state RelayState) RelayState {
182
- state.discoveryFailures = 0
183
- state.nextDiscoveryRefreshAt = time.Time{}
184
- state.unhealthySince = time.Time{}
185
- return state
186
-}
187
-
188
-func (p MOLSRelayPolicy) OnBanned(state RelayState) RelayState {
189
- state.Banned = true
190
- return state
191
-}
192
-
193
-func (p MOLSRelayPolicy) rankRelayPool(autoPool []RelayState, localAddress string) []string {
167
+func rankRelayPool(autoPool []RelayState, localAddress string) []string {
168
if len(autoPool) == 0 {
169
return nil
170
}
@@ -289,9 +263,9 @@ func (p MOLSRelayPolicy) rankRelayPool(autoPool []RelayState, localAddress strin
263
return append(activeURLs, fallbackURLs...)
264
}
265
292
-// SelectPriorityWithTrace is the telemetry-instrumented sibling of
266
+// selectPriorityWithTrace is the telemetry-instrumented sibling of
267
// SelectPriority. It returns the same ordered relay list plus a SelectionTrace.
294
-func (p MOLSRelayPolicy) SelectPriorityWithTrace(states []RelayState, cs ClientState) ([]string, telemetry.SelectionTrace) {
268
+func selectPriorityWithTrace(states []RelayState, cs RouteState) ([]string, telemetry.SelectionTrace) {
269
start := time.Now()
270
now := start.UTC()
271
@@ -311,7 +285,7 @@ func (p MOLSRelayPolicy) SelectPriorityWithTrace(states []RelayState, cs ClientS
285
}
286
}
287
314
- selected := p.SelectAggregate(states)
288
+ selected := selectAggregate(states)
289
if len(selected) == 0 {
290
trace.SelectionTook = time.Since(start)
291
return nil, trace
@@ -404,7 +378,7 @@ func (p MOLSRelayPolicy) SelectPriorityWithTrace(states []RelayState, cs ClientS
378
})
379
}
380
407
- autoURLs := p.rankRelayPool(autoPool, cs.LocalAddress)
381
+ autoURLs := rankRelayPool(autoPool, cs.LocalAddress)
382
maxActiveRelays := cs.MaxActiveRelays
383
if maxActiveRelays <= 0 {
384
maxActiveRelays = defaultMaxActiveRelays
@@ -440,15 +414,15 @@ func relayURLSet(states []RelayState) map[string]bool {
414
}
415
416
// SelectPriority returns the ordered list of relay URLs for a client using the
443
-// MOLS policy. It delegates to SelectPriorityWithTrace and discards the trace.
444
-func (p MOLSRelayPolicy) SelectPriority(states []RelayState, clientState ClientState) []string {
445
- out, _ := p.SelectPriorityWithTrace(states, clientState)
417
+// MOLS selection. It delegates to selectPriorityWithTrace and discards the trace.
418
+func SelectPriority(states []RelayState, routeState RouteState) []string {
419
+ out, _ := selectPriorityWithTrace(states, routeState)
420
return out
421
}
422
449
-// SelectMultiHopWithTrace is the telemetry-instrumented sibling of
423
+// selectMultiHopWithTrace is the telemetry-instrumented sibling of
424
// SelectMultiHop. It returns the same ordered relay list plus a SelectionTrace.
451
-func (p MOLSRelayPolicy) SelectMultiHopWithTrace(states []RelayState, cs ClientState) ([]string, telemetry.SelectionTrace) {
425
+func selectMultiHopWithTrace(states []RelayState, cs RouteState) ([]string, telemetry.SelectionTrace) {
426
start := time.Now()
427
now := start.UTC()
428
@@ -473,7 +447,7 @@ func (p MOLSRelayPolicy) SelectMultiHopWithTrace(states []RelayState, cs ClientS
447
}
448
}
449
476
- selected := p.SelectAggregate(states)
450
+ selected := selectAggregate(states)
451
if len(selected) == 0 {
452
trace.SelectionTook = time.Since(start)
453
return nil, trace
@@ -557,7 +531,7 @@ func (p MOLSRelayPolicy) SelectMultiHopWithTrace(states []RelayState, cs ClientS
531
})
532
}
533
560
- multiHop := p.rankRelayPool(autoPool, cs.LocalAddress)
534
+ multiHop := rankRelayPool(autoPool, cs.LocalAddress)
535
if len(multiHop) > cs.MultiHopDepth {
536
multiHop = multiHop[:cs.MultiHopDepth]
537
}
@@ -567,20 +541,8 @@ func (p MOLSRelayPolicy) SelectMultiHopWithTrace(states []RelayState, cs ClientS
541
}
542
543
// SelectMultiHop returns the ordered list of relay URLs for multi-hop routing.
570
-// It delegates to SelectMultiHopWithTrace and discards the trace.
571
-func (p MOLSRelayPolicy) SelectMultiHop(states []RelayState, clientState ClientState) []string {
572
- out, _ := p.SelectMultiHopWithTrace(states, clientState)
544
+// It delegates to selectMultiHopWithTrace and discards the trace.
545
+func SelectMultiHop(states []RelayState, routeState RouteState) []string {
546
+ out, _ := selectMultiHopWithTrace(states, routeState)
547
return out
548
}
575
-
576
-func rankRelayPool(autoPool []RelayState, localAddress string) []string {
577
- return MOLSRelayPolicy{}.rankRelayPool(autoPool, localAddress)
578
-}
579
-
580
-func selectPriority(states []RelayState, routeState RouteState) []string {
581
- return MOLSRelayPolicy{}.SelectPriority(states, routeState)
582
-}
583
-
584
-func selectMultiHop(states []RelayState, routeState RouteState) []string {
585
- return MOLSRelayPolicy{}.SelectMultiHop(states, routeState)
586
-}
portal/discovery/mols_test.go
+107
-49
@@ -4,6 +4,8 @@ import (
4
"fmt"
5
"testing"
6
"time"
7
+
8
+ "github.com/gosuda/portal-tunnel/v2/types"
9
)
10
11
// TestGF64MulIdentity checks that multiplying any element by 1 is the identity.
@@ -83,6 +85,32 @@ func TestMOLSScoreRowPermutation(t *testing.T) {
85
}
86
}
87
88
+func TestMOLSSelectPriorityMathematicalOrdering(t *testing.T) {
89
+ clientAddr := "192.168.0.10"
90
+ ingressIdx := hashToGF64(clientAddr)
91
+
92
+ relays := []string{
93
+ "https://relay-alpha.io",
94
+ "https://relay-beta.io",
95
+ "https://relay-gamma.io",
96
+ }
97
+
98
+ states := make([]RelayState, 0, len(relays))
99
+ for _, relayURL := range relays {
100
+ states = append(states, confirmedRelayState(t, relayURL))
101
+ }
102
+
103
+ selected := SelectPriority(states, RouteState{LocalAddress: clientAddr})
104
+
105
+ for i := 0; i < len(selected)-1; i++ {
106
+ scoreA := molsScore(int(ingressIdx), int(hashToGF64(selected[i])), int(molsBaseM1), int(molsBaseM2), molsOrder)
107
+ scoreB := molsScore(int(ingressIdx), int(hashToGF64(selected[i+1])), int(molsBaseM1), int(molsBaseM2), molsOrder)
108
+ if scoreA < scoreB {
109
+ t.Fatalf("selected[%d:%d] scores = %d < %d", i, i+1, scoreA, scoreB)
110
+ }
111
+ }
112
+}
113
+
114
// TestMOLSCongestionScoreRange checks that the Reverse-Siamese scores are in
115
// [1, 4096] and are the complement of the base scores.
116
func TestMOLSCongestionScoreRange(t *testing.T) {
@@ -161,10 +189,10 @@ func TestMOLSSelectPriorityKeepsExplicitRelaysOutsideAutoLimit(t *testing.T) {
189
relayA := "https://relay-a.example"
190
relayB := "https://relay-b.example"
191
164
- selected := selectPriority([]RelayState{
165
- bootstrapPolicyRelayState(explicitRelay),
166
- confirmedPolicyRelayState(t, relayA),
167
- confirmedPolicyRelayState(t, relayB),
192
+ selected := SelectPriority([]RelayState{
193
+ bootstrapRelayState(explicitRelay),
194
+ confirmedRelayState(t, relayA),
195
+ confirmedRelayState(t, relayB),
196
}, RouteState{
197
ExplicitRelayURLs: []string{explicitRelay},
198
MaxActiveRelays: 1,
@@ -182,15 +210,15 @@ func TestMOLSSelectPriorityKeepsExplicitRelaysOutsideAutoLimit(t *testing.T) {
210
// produce the same ordered output.
211
func TestMOLSSelectPriorityDeterministic(t *testing.T) {
212
states := []RelayState{
185
- confirmedPolicyRelayState(t, "https://relay-a.example"),
186
- confirmedPolicyRelayState(t, "https://relay-b.example"),
187
- confirmedPolicyRelayState(t, "https://relay-c.example"),
213
+ confirmedRelayState(t, "https://relay-a.example"),
214
+ confirmedRelayState(t, "https://relay-b.example"),
215
+ confirmedRelayState(t, "https://relay-c.example"),
216
}
217
routeState := RouteState{LocalAddress: "0x1234abcd"}
218
191
- first := selectPriority(states, routeState)
219
+ first := SelectPriority(states, routeState)
220
for range 5 {
193
- got := selectPriority(states, routeState)
221
+ got := SelectPriority(states, routeState)
222
if len(got) != len(first) {
223
t.Fatalf("non-deterministic length: %d vs %d", len(got), len(first))
224
}
@@ -207,22 +235,22 @@ func TestMOLSSelectPriorityDeterministic(t *testing.T) {
235
func TestMOLSSelectPriorityFallbackRelaysDemoted(t *testing.T) {
236
237
// Two healthy relays ensure molsMinActiveNodes is met without promoting fallbacks.
210
- healthy1 := confirmedPolicyRelayState(t, "https://relay-healthy-1.example")
238
+ healthy1 := confirmedRelayState(t, "https://relay-healthy-1.example")
239
healthy1.DiscoveryRTT = 100 * time.Millisecond
240
healthy1.DiscoveryRTTAt = time.Now()
241
healthy1.LoadFactor = 0.1 // Explicitly healthy
242
215
- healthy2 := confirmedPolicyRelayState(t, "https://relay-healthy-2.example")
243
+ healthy2 := confirmedRelayState(t, "https://relay-healthy-2.example")
244
healthy2.DiscoveryRTT = 150 * time.Millisecond
245
healthy2.DiscoveryRTTAt = time.Now()
246
healthy2.LoadFactor = 0.1 // Explicitly healthy
247
220
- fallback := confirmedPolicyRelayState(t, "https://relay-fallback.example")
248
+ fallback := confirmedRelayState(t, "https://relay-fallback.example")
249
fallback.DiscoveryRTT = molsFallbackRTTThreshold + time.Millisecond
250
fallback.DiscoveryRTTAt = time.Now()
251
fallback.LoadFactor = 0.1 // Explicitly healthy, but will be demoted by high RTT (isRelayFallback)
252
225
- selected := selectPriority([]RelayState{fallback, healthy1, healthy2}, RouteState{})
253
+ selected := SelectPriority([]RelayState{fallback, healthy1, healthy2}, RouteState{})
254
255
if len(selected) != 3 {
256
t.Fatalf("len(selected) = %d, want 3", len(selected))
@@ -238,14 +266,14 @@ func TestMOLSSelectPriorityFallbackRelaysDemoted(t *testing.T) {
266
// relays to maintain the minimum.
267
func TestMOLSSelectPriorityMinActiveNodesPromotesFallback(t *testing.T) {
268
241
- fallback1 := confirmedPolicyRelayState(t, "https://relay-fallback-1.example")
269
+ fallback1 := confirmedRelayState(t, "https://relay-fallback-1.example")
270
fallback1.DiscoveryRTT = molsFallbackRTTThreshold + time.Millisecond
271
fallback1.DiscoveryRTTAt = time.Now()
244
- fallback2 := confirmedPolicyRelayState(t, "https://relay-fallback-2.example")
272
+ fallback2 := confirmedRelayState(t, "https://relay-fallback-2.example")
273
fallback2.DiscoveryRTT = molsFallbackRTTThreshold + time.Millisecond
274
fallback2.DiscoveryRTTAt = time.Now()
275
248
- selected := selectPriority([]RelayState{fallback1, fallback2}, RouteState{})
276
+ selected := SelectPriority([]RelayState{fallback1, fallback2}, RouteState{})
277
278
// Both fallbacks should be promoted to meet the minimum of 2.
279
if len(selected) != 2 {
@@ -259,11 +287,11 @@ func TestMOLSSelectPriorityMinActiveNodesPromotesFallback(t *testing.T) {
287
func TestMOLSSelectPriorityCongestionSwitchChangesOrder(t *testing.T) {
288
289
// Two relays with different MOLS column indices so their scores differ.
262
- r1 := confirmedPolicyRelayState(t, "https://relay-one.example")
263
- r2 := confirmedPolicyRelayState(t, "https://relay-two.example")
290
+ r1 := confirmedRelayState(t, "https://relay-one.example")
291
+ r2 := confirmedRelayState(t, "https://relay-two.example")
292
293
// Normal mode: no RTT measurements, no congestion.
266
- normal := selectPriority([]RelayState{r1, r2}, RouteState{
294
+ normal := SelectPriority([]RelayState{r1, r2}, RouteState{
295
LocalAddress: "ingress-test",
296
})
297
@@ -276,7 +304,7 @@ func TestMOLSSelectPriorityCongestionSwitchChangesOrder(t *testing.T) {
304
r2c.DiscoveryRTT = rttHigh
305
r2c.DiscoveryRTTAt = time.Now()
306
279
- congested := selectPriority([]RelayState{r1c, r2c}, RouteState{
307
+ congested := SelectPriority([]RelayState{r1c, r2c}, RouteState{
308
LocalAddress: "ingress-test",
309
})
310
@@ -307,11 +335,11 @@ func TestMOLSSelectPriorityCongestionSwitchChangesOrder(t *testing.T) {
335
// mean RTT stays below the congestion threshold.
336
func TestMOLSSelectPriorityVariantGridActivatesOnHighCV(t *testing.T) {
337
310
- r1 := confirmedPolicyRelayState(t, "https://relay-one.example")
311
- r2 := confirmedPolicyRelayState(t, "https://relay-two.example")
338
+ r1 := confirmedRelayState(t, "https://relay-one.example")
339
+ r2 := confirmedRelayState(t, "https://relay-two.example")
340
341
// Normal mode: no RTT, no congestion, no CV.
314
- normalOrder := selectPriority([]RelayState{r1, r2}, RouteState{
342
+ normalOrder := SelectPriority([]RelayState{r1, r2}, RouteState{
343
LocalAddress: "ingress-cv",
344
})
345
@@ -333,7 +361,7 @@ func TestMOLSSelectPriorityVariantGridActivatesOnHighCV(t *testing.T) {
361
t.Fatalf("test precondition: avgRTT = %v, want <= %v", avgRTT, molsCongestionRTTThreshold)
362
}
363
336
- variantOrder := selectPriority([]RelayState{r1v, r2v}, RouteState{
364
+ variantOrder := SelectPriority([]RelayState{r1v, r2v}, RouteState{
365
LocalAddress: "ingress-cv",
366
})
367
@@ -354,9 +382,9 @@ func TestMOLSSelectPriorityVariantGridActivatesOnHighCV(t *testing.T) {
382
// property: each row is an independent permutation).
383
func TestMOLSSelectPriorityDifferentIngressDifferentOrder(t *testing.T) {
384
357
- r1 := confirmedPolicyRelayState(t, "https://relay-alpha.example")
358
- r2 := confirmedPolicyRelayState(t, "https://relay-beta.example")
359
- r3 := confirmedPolicyRelayState(t, "https://relay-gamma.example")
385
+ r1 := confirmedRelayState(t, "https://relay-alpha.example")
386
+ r2 := confirmedRelayState(t, "https://relay-beta.example")
387
+ r3 := confirmedRelayState(t, "https://relay-gamma.example")
388
states := []RelayState{r1, r2, r3}
389
390
// Collect orderings for a range of ingress addresses and check that at
@@ -366,7 +394,7 @@ func TestMOLSSelectPriorityDifferentIngressDifferentOrder(t *testing.T) {
394
"0xabc", "0xdef", "0x123", "0x456", "user@example.com", "relay.net",
395
}
396
for _, addr := range addresses {
369
- sel := selectPriority(states, RouteState{LocalAddress: addr})
397
+ sel := SelectPriority(states, RouteState{LocalAddress: addr})
398
key := ""
399
for _, u := range sel {
400
key += u + "|"
@@ -400,7 +428,7 @@ func TestMOLSSelectPriorityDifferentIngressDifferentOrder(t *testing.T) {
428
429
// TestMOLSSelectPriorityEmptyPoolReturnsNil checks the empty-input guard.
430
func TestMOLSSelectPriorityEmptyPoolReturnsNil(t *testing.T) {
403
- if got := selectPriority(nil, RouteState{}); got != nil {
431
+ if got := SelectPriority(nil, RouteState{}); got != nil {
432
t.Fatalf("SelectPriority(nil, ...) = %v, want nil", got)
433
}
434
}
@@ -411,10 +439,10 @@ func TestMOLSSelectPriorityMaxActiveRelaysLimitsAutoPool(t *testing.T) {
439
440
relays := make([]RelayState, 10)
441
for i := range relays {
414
- relays[i] = confirmedPolicyRelayState(t, fmt.Sprintf("https://relay-%d.example", i))
442
+ relays[i] = confirmedRelayState(t, fmt.Sprintf("https://relay-%d.example", i))
443
}
444
417
- selected := selectPriority(relays, RouteState{MaxActiveRelays: 3})
445
+ selected := SelectPriority(relays, RouteState{MaxActiveRelays: 3})
446
if len(selected) != 3 {
447
t.Fatalf("len(selected) = %d, want 3", len(selected))
448
}
@@ -424,39 +452,39 @@ func TestMOLSSelectPriorityZeroMaxActiveRelaysUsesDefault(t *testing.T) {
452
453
relays := make([]RelayState, 10)
454
for i := range relays {
427
- relays[i] = confirmedPolicyRelayState(t, fmt.Sprintf("https://relay-default-%d.example", i))
455
+ relays[i] = confirmedRelayState(t, fmt.Sprintf("https://relay-default-%d.example", i))
456
}
457
430
- selected := selectPriority(relays, RouteState{MaxActiveRelays: 0})
458
+ selected := SelectPriority(relays, RouteState{MaxActiveRelays: 0})
459
if len(selected) != defaultMaxActiveRelays {
460
t.Fatalf("len(selected) = %d, want %d", len(selected), defaultMaxActiveRelays)
461
}
462
}
463
464
func TestMOLSSelectPrioritySkipsExpiredAutoRelay(t *testing.T) {
437
- expired := confirmedPolicyRelayState(t, "https://relay-expired.example")
465
+ expired := confirmedRelayState(t, "https://relay-expired.example")
466
expired.Descriptor.ExpiresAt = time.Now().UTC().Add(-time.Minute)
467
440
- if selected := selectPriority([]RelayState{expired}, RouteState{}); len(selected) != 0 {
468
+ if selected := SelectPriority([]RelayState{expired}, RouteState{}); len(selected) != 0 {
469
t.Fatalf("SelectPriority(expired auto) = %v, want empty", selected)
470
}
471
}
472
473
func TestMOLSSelectPrioritySkipsBannedRelay(t *testing.T) {
446
- banned := confirmedPolicyRelayState(t, "https://relay-banned.example")
474
+ banned := confirmedRelayState(t, "https://relay-banned.example")
475
banned.Banned = true
476
449
- if selected := selectPriority([]RelayState{banned}, RouteState{}); len(selected) != 0 {
477
+ if selected := SelectPriority([]RelayState{banned}, RouteState{}); len(selected) != 0 {
478
t.Fatalf("SelectPriority(banned) = %v, want empty", selected)
479
}
480
}
481
482
func TestMOLSSelectPriorityKeepsExpiredExplicitRelay(t *testing.T) {
483
relayURL := "https://relay-explicit-expired.example"
456
- expired := confirmedPolicyRelayState(t, relayURL)
484
+ expired := confirmedRelayState(t, relayURL)
485
expired.Descriptor.ExpiresAt = time.Now().UTC().Add(-time.Minute)
486
459
- selected := selectPriority([]RelayState{expired}, RouteState{
487
+ selected := SelectPriority([]RelayState{expired}, RouteState{
488
ExplicitRelayURLs: []string{relayURL},
489
})
490
if len(selected) != 1 || selected[0] != relayURL {
@@ -465,20 +493,20 @@ func TestMOLSSelectPriorityKeepsExpiredExplicitRelay(t *testing.T) {
493
}
494
495
func TestMOLSSelectPrioritySkipsAutoRelayInBackoff(t *testing.T) {
468
- backingOff := confirmedPolicyRelayState(t, "https://relay-backoff.example")
496
+ backingOff := confirmedRelayState(t, "https://relay-backoff.example")
497
backingOff.suppressActiveUntil = time.Now().UTC().Add(time.Minute)
498
471
- if selected := selectPriority([]RelayState{backingOff}, RouteState{}); len(selected) != 0 {
499
+ if selected := SelectPriority([]RelayState{backingOff}, RouteState{}); len(selected) != 0 {
500
t.Fatalf("SelectPriority(backing off auto) = %v, want empty", selected)
501
}
502
}
503
504
func TestMOLSSelectPriorityKeepsDiscoveryBackoffRelay(t *testing.T) {
505
relayURL := "https://relay-discovery-backoff.example"
478
- backingOff := confirmedPolicyRelayState(t, relayURL)
506
+ backingOff := confirmedRelayState(t, relayURL)
507
backingOff.nextDiscoveryRefreshAt = time.Now().UTC().Add(time.Minute)
508
481
- selected := selectPriority([]RelayState{backingOff}, RouteState{})
509
+ selected := SelectPriority([]RelayState{backingOff}, RouteState{})
510
if len(selected) != 1 || selected[0] != relayURL {
511
t.Fatalf("SelectPriority(discovery backoff) = %v, want [%q]", selected, relayURL)
512
}
@@ -487,7 +515,7 @@ func TestMOLSSelectPriorityKeepsDiscoveryBackoffRelay(t *testing.T) {
515
func TestMOLSSelectPriorityKeepsUnobservedAutoSeed(t *testing.T) {
516
relayURL := "https://relay-seed.example"
517
490
- selected := selectPriority([]RelayState{bootstrapPolicyRelayState(relayURL)}, RouteState{})
518
+ selected := SelectPriority([]RelayState{bootstrapRelayState(relayURL)}, RouteState{})
519
if len(selected) != 1 || selected[0] != relayURL {
520
t.Fatalf("SelectPriority(unobserved seed) = %v, want [%q]", selected, relayURL)
521
}
@@ -603,20 +631,18 @@ func TestMOLSRTTStatsEmpty(t *testing.T) {
631
// TestMOLSSelectPriorityEWMAStabilityTransposition verifies that relays with
632
// high EWMA RTT are demoted relative to stable relays.
633
func TestMOLSSelectPriorityEWMAStabilityTransposition(t *testing.T) {
606
- policy := MOLSRelayPolicy{}
607
-
608
- relayStable := confirmedPolicyRelayState(t, "https://relay-stable.example")
634
+ relayStable := confirmedRelayState(t, "https://relay-stable.example")
635
relayStable.EWMARTT = 100 * time.Millisecond
636
relayStable.DiscoveryRTT = 100 * time.Millisecond
637
612
- relayUnstable := confirmedPolicyRelayState(t, "https://relay-unstable.example")
638
+ relayUnstable := confirmedRelayState(t, "https://relay-unstable.example")
639
relayUnstable.EWMARTT = 600 * time.Millisecond
640
relayUnstable.DiscoveryRTT = 600 * time.Millisecond
641
642
states := []RelayState{relayStable, relayUnstable}
643
644
// We force the same ingress so they are ranked together.
619
- selected := policy.SelectPriority(states, ClientState{LocalAddress: "test-ingress"})
645
+ selected := SelectPriority(states, RouteState{LocalAddress: "test-ingress"})
646
647
if len(selected) != 2 {
648
t.Fatalf("len(selected) = %d, want 2", len(selected))
@@ -627,3 +653,35 @@ func TestMOLSSelectPriorityEWMAStabilityTransposition(t *testing.T) {
653
t.Errorf("expected stable relay to be first, got %q", selected[0])
654
}
655
}
656
+
657
+func BenchmarkMOLSRankRelayPool(b *testing.B) {
658
+ localAddr := "test-client-address"
659
+ relays := make([]RelayState, 100)
660
+ for i := 0; i < 100; i++ {
661
+ relays[i] = RelayState{
662
+ Descriptor: types.RelayDescriptor{APIHTTPSAddr: "test"},
663
+ DiscoveryRTT: 100 * time.Millisecond,
664
+ DiscoveryRTTAt: time.Now(),
665
+ Confirmed: true,
666
+ }
667
+ }
668
+
669
+ b.ResetTimer()
670
+ for i := 0; i < b.N; i++ {
671
+ rankRelayPool(relays, localAddr)
672
+ }
673
+}
674
+
675
+func BenchmarkMOLSSelectPriorityMassiveScale(b *testing.B) {
676
+ const numRelays = 256
677
+ relayStates := make([]RelayState, numRelays)
678
+ for i := range relayStates {
679
+ relayStates[i] = RelayState{Descriptor: types.RelayDescriptor{APIHTTPSAddr: fmt.Sprintf("https://test-%d.example", i)}}
680
+ }
681
+
682
+ b.ResetTimer()
683
+ for i := 0; i < b.N; i++ {
684
+ routeState := RouteState{LocalAddress: fmt.Sprintf("client-%d", i)}
685
+ SelectPriority(relayStates, routeState)
686
+ }
687
+}
portal/discovery/policy_test.go
deleted
-227
@@ -1,227 +0,0 @@
1
-package discovery
2
-
3
-import (
4
- "testing"
5
- "time"
6
-
7
- "github.com/gosuda/portal-tunnel/v2/portal/auth"
8
- "github.com/gosuda/portal-tunnel/v2/portal/identity"
9
- "github.com/gosuda/portal-tunnel/v2/types"
10
-)
11
-
12
-func mustPolicyRelayDescriptor(t *testing.T, relayURL string) types.RelayDescriptor {
13
- t.Helper()
14
-
15
- signing, err := identity.ResolveSecp256k1Identity("")
16
- if err != nil {
17
- t.Fatalf("identity.ResolveSecp256k1Identity() error = %v", err)
18
- }
19
- authority, err := identity.NewLocalAuthority(signing)
20
- if err != nil {
21
- t.Fatalf("identity.NewLocalAuthority() error = %v", err)
22
- }
23
- now := time.Now().UTC()
24
- signed, err := auth.SignRelayDescriptor(types.RelayDescriptor{
25
- Address: signing.Address,
26
- Version: types.DiscoveryVersion,
27
- IssuedAt: now,
28
- ExpiresAt: now.Add(time.Hour),
29
- APIHTTPSAddr: relayURL,
30
- }, authority)
31
- if err != nil {
32
- t.Fatalf("SignRelayDescriptor() error = %v", err)
33
- }
34
- return signed
35
-}
36
-
37
-func bootstrapPolicyRelayState(relayURL string) RelayState {
38
- return RelayState{
39
- Descriptor: types.RelayDescriptor{
40
- APIHTTPSAddr: relayURL,
41
- },
42
- Bootstrap: true,
43
- }
44
-}
45
-
46
-func confirmedPolicyRelayState(t *testing.T, relayURL string) RelayState {
47
- t.Helper()
48
-
49
- return RelayState{
50
- Descriptor: mustPolicyRelayDescriptor(t, relayURL),
51
- Confirmed: true,
52
- LastSeenAt: time.Now().UTC(),
53
- }
54
-}
55
-
56
-func confirmedPolicyRelayStateWithRTT(t *testing.T, relayURL string, rtt time.Duration) RelayState {
57
- t.Helper()
58
-
59
- state := confirmedPolicyRelayState(t, relayURL)
60
- state.DiscoveryRTT = rtt
61
- state.DiscoveryRTTAt = time.Now().UTC()
62
- return state
63
-}
64
-
65
-func TestSelectPriorityMathematicalOrdering(t *testing.T) {
66
- clientAddr := "192.168.0.10"
67
- ingressIdx := hashToGF64(clientAddr)
68
-
69
- relays := []string{
70
- "https://relay-alpha.io",
71
- "https://relay-beta.io",
72
- "https://relay-gamma.io",
73
- }
74
-
75
- var states []RelayState
76
- for _, url := range relays {
77
- states = append(states, confirmedPolicyRelayState(t, url))
78
- }
79
-
80
- selected := selectPriority(states, RouteState{LocalAddress: clientAddr})
81
-
82
- for i := 0; i < len(selected)-1; i++ {
83
- scoreA := molsScore(int(ingressIdx), int(hashToGF64(selected[i])), int(molsBaseM1), int(molsBaseM2), 64)
84
- scoreB := molsScore(int(ingressIdx), int(hashToGF64(selected[i+1])), int(molsBaseM1), int(molsBaseM2), 64)
85
- if scoreA < scoreB {
86
- t.Errorf("Priority mismatch at index %d: %d < %d", i, scoreA, scoreB)
87
- }
88
- }
89
-}
90
-
91
-func TestSelectPriorityKeepsExplicitRelaysOutsideAutoLimit(t *testing.T) {
92
- explicitRelay := "https://relay-explicit.example"
93
- relayA := "https://relay-a.example"
94
- relayB := "https://relay-b.example"
95
-
96
- selected := selectPriority([]RelayState{
97
- bootstrapPolicyRelayState(explicitRelay),
98
- confirmedPolicyRelayState(t, relayA),
99
- confirmedPolicyRelayState(t, relayB),
100
- }, RouteState{
101
- LocalAddress: "127.0.0.1",
102
- ExplicitRelayURLs: []string{explicitRelay},
103
- MaxActiveRelays: 1,
104
- })
105
-
106
- if len(selected) < 2 {
107
- t.Fatalf("len(selected) = %d, want at least 2", len(selected))
108
- }
109
- if selected[0] != explicitRelay {
110
- t.Fatalf("selected[0] = %q, want %q", selected[0], explicitRelay)
111
- }
112
-}
113
-
114
-func TestSelectPriorityCongestionInversion(t *testing.T) {
115
- clientAddr := "10.0.0.1"
116
- ingressIdx := hashToGF64(clientAddr)
117
-
118
- r1, r2 := "https://r1.net", "https://r2.net"
119
- states := []RelayState{
120
- confirmedPolicyRelayStateWithRTT(t, r1, 800*time.Millisecond),
121
- confirmedPolicyRelayStateWithRTT(t, r2, 800*time.Millisecond),
122
- }
123
-
124
- selected := selectPriority(states, RouteState{LocalAddress: clientAddr})
125
-
126
- if len(selected) == 2 {
127
- s1 := molsCongestionScore(int(ingressIdx), int(hashToGF64(selected[0])), int(molsBaseM1), int(molsBaseM2), 64)
128
- s2 := molsCongestionScore(int(ingressIdx), int(hashToGF64(selected[1])), int(molsBaseM1), int(molsBaseM2), 64)
129
- if s1 < s2 {
130
- t.Errorf("Congestion priority failed: %d < %d", s1, s2)
131
- }
132
- }
133
-}
134
-
135
-func TestSelectPriorityFallbackPromotion(t *testing.T) {
136
- states := []RelayState{
137
- confirmedPolicyRelayStateWithRTT(t, "https://f1.com", 3*time.Second),
138
- confirmedPolicyRelayStateWithRTT(t, "https://f2.com", 4*time.Second),
139
- }
140
-
141
- selected := selectPriority(states, RouteState{LocalAddress: "1.1.1.1"})
142
-
143
- if len(selected) < molsMinActiveNodes {
144
- t.Errorf("Fallback promotion failed: got %d, want %d", len(selected), molsMinActiveNodes)
145
- }
146
-}
147
-
148
-func TestConfirmRelayURLResetsActiveFailures(t *testing.T) {
149
- relayURL := "https://error.io"
150
- set := NewRelaySet(nil)
151
- state := RelayState{
152
- Descriptor: types.RelayDescriptor{
153
- APIHTTPSAddr: relayURL,
154
- },
155
- activeFailures: 5,
156
- suppressActiveUntil: time.Now().UTC().Add(time.Minute),
157
- Confirmed: false,
158
- }
159
- set.mu.Lock()
160
- set.relays[relayURL] = state
161
- set.mu.Unlock()
162
-
163
- set.ConfirmRelayURL(relayURL)
164
-
165
- set.mu.RLock()
166
- state = set.relays[relayURL]
167
- set.mu.RUnlock()
168
- if !state.Confirmed {
169
- t.Fatal("Confirmed should be true")
170
- }
171
- if state.activeFailures != 0 {
172
- t.Errorf("activeFailures = %d, want 0", state.activeFailures)
173
- }
174
- if !state.suppressActiveUntil.IsZero() {
175
- t.Errorf("suppressActiveUntil = %v, want zero", state.suppressActiveUntil)
176
- }
177
-}
178
-
179
-func TestRecordDiscoveryFailureBackoff(t *testing.T) {
180
- relayURL := "https://error.io"
181
- set := NewRelaySet(nil)
182
- set.mu.Lock()
183
- set.relays[relayURL] = confirmedPolicyRelayState(t, relayURL)
184
- set.mu.Unlock()
185
- budget := 3
186
-
187
- start := time.Now()
188
- for i := 0; i < budget; i++ {
189
- backed, _, _ := set.RecordDiscoveryFailure(relayURL, budget)
190
- if i < budget-1 && backed {
191
- t.Fatal("Premature backoff")
192
- }
193
- }
194
-
195
- set.mu.RLock()
196
- state := set.relays[relayURL]
197
- set.mu.RUnlock()
198
- if !state.nextDiscoveryRefreshAt.After(start) {
199
- t.Fatal("discovery retry timer not scheduled")
200
- }
201
- if !state.suppressActiveUntil.IsZero() {
202
- t.Fatalf("suppressActiveUntil = %v, want zero", state.suppressActiveUntil)
203
- }
204
-}
205
-
206
-func TestRecordActiveFailureBackoff(t *testing.T) {
207
- relayURL := "https://error.io"
208
- set := NewRelaySet(nil)
209
- set.mu.Lock()
210
- set.relays[relayURL] = confirmedPolicyRelayState(t, relayURL)
211
- set.mu.Unlock()
212
- start := time.Now()
213
-
214
- backed, _, _ := set.RecordActiveFailure(relayURL, 1)
215
- if !backed {
216
- t.Fatal("active failure should back off at budget")
217
- }
218
- set.mu.RLock()
219
- state := set.relays[relayURL]
220
- set.mu.RUnlock()
221
- if !state.suppressActiveUntil.After(start) {
222
- t.Fatal("active suppression timer not scheduled")
223
- }
224
- if !state.nextDiscoveryRefreshAt.IsZero() {
225
- t.Fatalf("nextDiscoveryRefreshAt = %v, want zero", state.nextDiscoveryRefreshAt)
226
- }
227
-}
portal/discovery/qos_test.go
deleted
-65
@@ -1,65 +0,0 @@
1
-package discovery
2
-
3
-import (
4
- "fmt"
5
- "math/rand"
6
- "sort"
7
- "testing"
8
- "time"
9
-
10
- "github.com/gosuda/portal-tunnel/v2/types"
11
-)
12
-
13
-// TestQoSConsistency verifies p99 latency reduction and rank stability (oscillation).
14
-func TestQoSConsistency(t *testing.T) {
15
- rng := rand.New(rand.NewSource(42))
16
- numNodes := 20
17
- // 5 stable nodes (100ms), 5 jittery nodes (spikes to 600ms)
18
- nodes := make([]RelayState, numNodes)
19
- for i := 0; i < numNodes; i++ {
20
- nodes[i] = RelayState{Descriptor: types.RelayDescriptor{APIHTTPSAddr: fmt.Sprintf("node-%d", i)}}
21
- base := 100.0
22
- if i >= 5 {
23
- base = 300.0
24
- } // jittery nodes are slower on average
25
- for j := 0; j < 100; j++ {
26
- rtt := base
27
- if i >= 5 && rng.Float64() < 0.2 {
28
- rtt += 400.0
29
- }
30
- nodes[i].UpdateEWMARTT(time.Duration(rtt) * time.Millisecond)
31
- }
32
- }
33
-
34
- policy := MOLSRelayPolicy{}
35
- var history []string
36
- var latencies []float64
37
-
38
- // Simulate 1000 selection rounds
39
- for r := 0; r < 1000; r++ {
40
- selected := policy.rankRelayPool(nodes, "client-x")
41
- top := selected[0]
42
- history = append(history, top)
43
-
44
- // Find latency of top node
45
- for _, n := range nodes {
46
- if n.Descriptor.APIHTTPSAddr == top {
47
- latencies = append(latencies, float64(n.EWMARTT.Milliseconds()))
48
- }
49
- }
50
- }
51
-
52
- // 1. Oscillation Check: count changes in top-pick
53
- changes := 0
54
- for i := 1; i < len(history); i++ {
55
- if history[i] != history[i-1] {
56
- changes++
57
- }
58
- }
59
-
60
- // 2. Latency Check: p99
61
- sort.Float64s(latencies)
62
- p99 := latencies[990]
63
-
64
- fmt.Printf("QoS Results: Changes (Oscillations)=%d, p99 Latency=%vms\n", changes, p99)
65
-}
portal/discovery/refresher.go
+3
-5
@@ -184,7 +184,8 @@ func (r *Refresher) refreshHTTPS(ctx context.Context) error {
184
}
185
186
func (r *Refresher) refreshOverlay(ctx context.Context) error {
187
- states := r.relaySet.overlayPeerRelayStates()
187
+ now := time.Now().UTC()
188
+ states := r.relaySet.overlayPeerRelayStates(now)
189
if len(states) == 0 {
190
return nil
191
}
@@ -196,10 +197,7 @@ func (r *Refresher) refreshOverlay(ctx context.Context) error {
197
return err
198
}
199
relaySetChanged := false
199
- for _, state := range states {
200
- if !state.nextDiscoveryRefreshAt.IsZero() && state.nextDiscoveryRefreshAt.After(time.Now().UTC()) {
201
- continue
202
- }
200
+ for _, state := range r.relaySet.overlayRefreshCandidates(now) {
201
relay := state.Descriptor
202
recoveryFailures := r.directRecoveryFailures
203
if state.Bootstrap {
portal/discovery/relayset.go
+40
-18
@@ -335,7 +335,7 @@ func disposableRelayState(state RelayState) bool {
335
}
336
337
func (s *RelaySet) AggregateRelays() []RelayState {
338
- return MOLSRelayPolicy{}.SelectAggregate(s.currentRelayStates(time.Now().UTC()))
338
+ return selectAggregate(s.currentRelayStates(time.Now().UTC()))
339
}
340
341
func (s *RelaySet) AllRelays() []RelayState {
@@ -343,7 +343,7 @@ func (s *RelaySet) AllRelays() []RelayState {
343
}
344
345
func (s *RelaySet) ConfirmedRelays() []RelayState {
346
- return MOLSRelayPolicy{}.SelectConfirmed(s.currentRelayStates(time.Now().UTC()))
346
+ return selectConfirmed(s.currentRelayStates(time.Now().UTC()))
347
}
348
349
type Route struct {
@@ -400,14 +400,14 @@ func (s *RelaySet) PlanRoutes(explicitPath []string, routeState RouteState) ([]R
400
states := s.currentRelayStates(time.Now().UTC())
401
402
if routeState.MultiHopDepth > 1 {
403
- path := selectMultiHop(states, routeState)
403
+ path := SelectMultiHop(states, routeState)
404
if len(path) < routeState.MultiHopDepth {
405
return nil, fmt.Errorf("multi-hop-depth %d requires %d overlay relay candidates, got %d", routeState.MultiHopDepth, routeState.MultiHopDepth, len(path))
406
}
407
return []Route{NewRoute(path, false)}, nil
408
}
409
410
- relayURLs := selectPriority(states, routeState)
410
+ relayURLs := SelectPriority(states, routeState)
411
routes := make([]Route, 0, len(relayURLs))
412
for _, relayURL := range relayURLs {
413
routes = append(routes, NewRoute([]string{relayURL}, slices.Contains(routeState.ExplicitRelayURLs, relayURL)))
@@ -420,10 +420,10 @@ func (s *RelaySet) PlanRoutes(explicitPath []string, routeState RouteState) ([]R
420
// eligibility classification, and the scoring parameters used. Prometheus
421
// metrics are emitted from the trace before returning, and a sampled zerolog
422
// debug entry is written.
423
-func (s *RelaySet) PriorityRelaysWithTrace(clientState ClientState) ([]string, telemetry.SelectionTrace) {
423
+func (s *RelaySet) PriorityRelaysWithTrace(routeState RouteState) ([]string, telemetry.SelectionTrace) {
424
states := s.currentRelayStates(time.Now().UTC())
425
426
- result, trace := MOLSRelayPolicy{}.SelectPriorityWithTrace(states, clientState)
426
+ result, trace := selectPriorityWithTrace(states, routeState)
427
telemetry.EmitFromTrace(trace)
428
log.Debug().
429
Uint8("client_hash", trace.ClientHash).
@@ -437,10 +437,9 @@ func (s *RelaySet) PriorityRelaysWithTrace(clientState ClientState) ([]string, t
437
}
438
439
// PriorityRelays returns the ordered list of relay URLs for a client. It
440
-// delegates to PriorityRelaysWithTrace and discards the trace. The public
441
-// signature is unchanged through all phases.
442
-func (s *RelaySet) PriorityRelays(clientState ClientState) []string {
443
- out, _ := s.PriorityRelaysWithTrace(clientState)
440
+// delegates to PriorityRelaysWithTrace and discards the trace.
441
+func (s *RelaySet) PriorityRelays(routeState RouteState) []string {
442
+ out, _ := s.PriorityRelaysWithTrace(routeState)
443
return out
444
}
445
@@ -449,10 +448,10 @@ func (s *RelaySet) PriorityRelays(clientState ClientState) []string {
448
// eligibility classification, and the scoring parameters used. Prometheus
449
// metrics are emitted from the trace before returning, and a sampled zerolog
450
// debug entry is written.
452
-func (s *RelaySet) PriorityMultiHopWithTrace(clientState ClientState) ([]string, telemetry.SelectionTrace) {
451
+func (s *RelaySet) PriorityMultiHopWithTrace(routeState RouteState) ([]string, telemetry.SelectionTrace) {
452
states := s.currentRelayStates(time.Now().UTC())
453
455
- result, trace := MOLSRelayPolicy{}.SelectMultiHopWithTrace(states, clientState)
454
+ result, trace := selectMultiHopWithTrace(states, routeState)
455
telemetry.EmitFromTrace(trace)
456
log.Debug().
457
Uint8("client_hash", trace.ClientHash).
@@ -467,14 +466,37 @@ func (s *RelaySet) PriorityMultiHopWithTrace(clientState ClientState) ([]string,
466
467
// PriorityMultiHop returns the ordered list of relay URLs for multi-hop
468
// routing. It delegates to PriorityMultiHopWithTrace and discards the trace.
470
-// The public signature is unchanged through all phases.
471
-func (s *RelaySet) PriorityMultiHop(clientState ClientState) []string {
472
- out, _ := s.PriorityMultiHopWithTrace(clientState)
469
+func (s *RelaySet) PriorityMultiHop(routeState RouteState) []string {
470
+ out, _ := s.PriorityMultiHopWithTrace(routeState)
471
return out
472
}
473
476
-func (s *RelaySet) overlayPeerRelayStates() []RelayState {
477
- now := time.Now().UTC()
474
+func (s *RelaySet) overlayRefreshCandidates(now time.Time) []RelayState {
475
+ if now.IsZero() {
476
+ now = time.Now().UTC()
477
+ } else {
478
+ now = now.UTC()
479
+ }
480
+ states := s.overlayPeerRelayStates(now)
481
+ out := make([]RelayState, 0, len(states))
482
+ for _, state := range states {
483
+ if !state.nextDiscoveryRefreshAt.IsZero() && state.nextDiscoveryRefreshAt.After(now) {
484
+ continue
485
+ }
486
+ out = append(out, state)
487
+ }
488
+ if len(out) == 0 {
489
+ return nil
490
+ }
491
+ return out
492
+}
493
+
494
+func (s *RelaySet) overlayPeerRelayStates(now time.Time) []RelayState {
495
+ if now.IsZero() {
496
+ now = time.Now().UTC()
497
+ } else {
498
+ now = now.UTC()
499
+ }
500
states := s.currentRelayStates(now)
501
out := make([]RelayState, 0, len(states))
502
for _, state := range states {
@@ -490,7 +512,7 @@ func (s *RelaySet) overlayPeerRelayStates() []RelayState {
512
}
513
514
func (s *RelaySet) OverlayPeerDescriptor() []types.RelayDescriptor {
493
- states := s.overlayPeerRelayStates()
515
+ states := s.overlayPeerRelayStates(time.Now().UTC())
516
if len(states) == 0 {
517
return nil
518
}
portal/discovery/relayset_test.go
+112
-11
@@ -19,10 +19,31 @@ func relayStates(set *RelaySet) []RelayState {
19
return states
20
}
21
22
+func mustRelayDescriptor(t *testing.T, relayURL string) types.RelayDescriptor {
23
+ t.Helper()
24
+ now := time.Now().UTC().Truncate(time.Microsecond)
25
+ return mustSignedDescriptor(t, mustSigningIdentity(t), relayURL, now)
26
+}
27
+
28
+func confirmedRelayState(t *testing.T, relayURL string) RelayState {
29
+ t.Helper()
30
+ return RelayState{
31
+ Descriptor: mustRelayDescriptor(t, relayURL),
32
+ Confirmed: true,
33
+ LastSeenAt: time.Now().UTC(),
34
+ }
35
+}
36
+
37
+func bootstrapRelayState(relayURL string) RelayState {
38
+ state := newRelayState(relayURL)
39
+ state.Bootstrap = true
40
+ return state
41
+}
42
+
43
func TestApplyRelayDiscoveryResponsePreservesBootstrapFlag(t *testing.T) {
44
set := NewRelaySet([]string{"https://relay-a.example"})
45
25
- desc := mustPolicyRelayDescriptor(t, "https://relay-a.example")
46
+ desc := mustRelayDescriptor(t, "https://relay-a.example")
47
if _, err := set.ApplyRelayDiscoveryResponse(desc.APIHTTPSAddr, types.DiscoveryResponse{
48
ProtocolVersion: types.DiscoveryVersion,
49
Relays: []types.RelayDescriptor{desc},
@@ -44,7 +65,7 @@ func TestDescriptorsDropsExpiredSignedRelayDescriptor(t *testing.T) {
65
66
now := time.Now().UTC()
67
relayURL := "https://relay-stale.example"
47
- state := confirmedPolicyRelayState(t, relayURL)
68
+ state := confirmedRelayState(t, relayURL)
69
state.Descriptor.ExpiresAt = now.Add(-time.Minute)
70
state.LastSeenAt = now.Add(-6 * time.Hour)
71
state.Descriptor.SupportsUDP = true
@@ -63,7 +84,7 @@ func TestDescriptorsDropsExpiredSignedRelayDescriptor(t *testing.T) {
84
func TestApplyRelayDiscoveryResponseCollectsRelaysDespiteProtocolMismatch(t *testing.T) {
85
set := NewRelaySet(nil)
86
66
- desc := mustPolicyRelayDescriptor(t, "https://relay-mismatch.example")
87
+ desc := mustRelayDescriptor(t, "https://relay-mismatch.example")
88
changed, err := set.ApplyRelayDiscoveryResponse("", types.DiscoveryResponse{
89
ProtocolVersion: "5",
90
Relays: []types.RelayDescriptor{desc},
@@ -90,7 +111,7 @@ func TestApplyRelayDiscoveryResponseCollectsRelaysDespiteProtocolMismatch(t *tes
111
func TestApplyRelayDiscoveryResponseCollectsHintsWhenTargetDescriptorIsMissing(t *testing.T) {
112
set := NewRelaySet(nil)
113
93
- hinted := mustPolicyRelayDescriptor(t, "https://relay-hinted.example")
114
+ hinted := mustRelayDescriptor(t, "https://relay-hinted.example")
115
changed, err := set.ApplyRelayDiscoveryResponse("https://relay-source.example", types.DiscoveryResponse{
116
ProtocolVersion: "5",
117
Relays: []types.RelayDescriptor{hinted},
@@ -118,7 +139,7 @@ func TestApplyRelayDiscoveryResponseClearsDiscoveryRetryOnAuthoritativeSuccess(t
139
set := NewRelaySet(nil)
140
141
relayURL := "https://relay-source.example"
121
- desc := mustPolicyRelayDescriptor(t, relayURL)
142
+ desc := mustRelayDescriptor(t, relayURL)
143
set.mu.Lock()
144
state := RelayState{
145
Descriptor: desc,
@@ -159,7 +180,7 @@ func TestApplyRelayDiscoveryResponsePreservesDiscoveryRetryOnHint(t *testing.T)
180
set := NewRelaySet(nil)
181
182
relayURL := "https://relay-hinted.example"
162
- desc := mustPolicyRelayDescriptor(t, relayURL)
183
+ desc := mustRelayDescriptor(t, relayURL)
184
nextDiscoveryRefreshAt := time.Now().UTC().Add(time.Minute)
185
set.mu.Lock()
186
state := RelayState{
@@ -190,7 +211,7 @@ func TestConfirmRelayURLMarksRelayConfirmedWithoutChangingAggregateDescriptor(t
211
212
relayURL := "https://relay-confirmed.example"
213
state := RelayState{
193
- Descriptor: mustPolicyRelayDescriptor(t, relayURL),
214
+ Descriptor: mustRelayDescriptor(t, relayURL),
215
LastSeenAt: time.Now().UTC(),
216
}
217
@@ -211,11 +232,41 @@ func TestConfirmRelayURLMarksRelayConfirmedWithoutChangingAggregateDescriptor(t
232
}
233
}
234
235
+func TestConfirmRelayURLResetsActiveFailures(t *testing.T) {
236
+ relayURL := "https://error.io"
237
+ set := NewRelaySet(nil)
238
+ state := RelayState{
239
+ Descriptor: types.RelayDescriptor{
240
+ APIHTTPSAddr: relayURL,
241
+ },
242
+ activeFailures: 5,
243
+ suppressActiveUntil: time.Now().UTC().Add(time.Minute),
244
+ }
245
+ set.mu.Lock()
246
+ set.relays[relayURL] = state
247
+ set.mu.Unlock()
248
+
249
+ set.ConfirmRelayURL(relayURL)
250
+
251
+ set.mu.RLock()
252
+ state = set.relays[relayURL]
253
+ set.mu.RUnlock()
254
+ if !state.Confirmed {
255
+ t.Fatal("relay should be confirmed")
256
+ }
257
+ if state.activeFailures != 0 {
258
+ t.Fatalf("activeFailures = %d, want 0", state.activeFailures)
259
+ }
260
+ if !state.suppressActiveUntil.IsZero() {
261
+ t.Fatalf("suppressActiveUntil = %v, want zero", state.suppressActiveUntil)
262
+ }
263
+}
264
+
265
func TestUnconfirmRelayURLClearsLocalConfirmationOnly(t *testing.T) {
266
set := NewRelaySet(nil)
267
268
relayURL := "https://relay-confirmed.example"
218
- state := confirmedPolicyRelayState(t, relayURL)
269
+ state := confirmedRelayState(t, relayURL)
270
271
set.mu.Lock()
272
set.relays[relayURL] = state
@@ -231,11 +282,61 @@ func TestUnconfirmRelayURLClearsLocalConfirmationOnly(t *testing.T) {
282
}
283
}
284
285
+func TestRecordDiscoveryFailureBackoff(t *testing.T) {
286
+ relayURL := "https://discovery-error.example"
287
+ set := NewRelaySet(nil)
288
+ set.mu.Lock()
289
+ set.relays[relayURL] = confirmedRelayState(t, relayURL)
290
+ set.mu.Unlock()
291
+ budget := 3
292
+
293
+ start := time.Now()
294
+ for i := 0; i < budget; i++ {
295
+ backedOff, _, _ := set.RecordDiscoveryFailure(relayURL, budget)
296
+ if i < budget-1 && backedOff {
297
+ t.Fatal("discovery failure backed off before budget")
298
+ }
299
+ }
300
+
301
+ set.mu.RLock()
302
+ state := set.relays[relayURL]
303
+ set.mu.RUnlock()
304
+ if !state.nextDiscoveryRefreshAt.After(start) {
305
+ t.Fatal("discovery retry timer was not scheduled")
306
+ }
307
+ if !state.suppressActiveUntil.IsZero() {
308
+ t.Fatalf("suppressActiveUntil = %v, want zero", state.suppressActiveUntil)
309
+ }
310
+}
311
+
312
+func TestRecordActiveFailureBackoff(t *testing.T) {
313
+ relayURL := "https://active-error.example"
314
+ set := NewRelaySet(nil)
315
+ set.mu.Lock()
316
+ set.relays[relayURL] = confirmedRelayState(t, relayURL)
317
+ set.mu.Unlock()
318
+ start := time.Now()
319
+
320
+ backedOff, _, _ := set.RecordActiveFailure(relayURL, 1)
321
+ if !backedOff {
322
+ t.Fatal("active failure should back off at budget")
323
+ }
324
+ set.mu.RLock()
325
+ state := set.relays[relayURL]
326
+ set.mu.RUnlock()
327
+ if !state.suppressActiveUntil.After(start) {
328
+ t.Fatal("active suppression timer was not scheduled")
329
+ }
330
+ if !state.nextDiscoveryRefreshAt.IsZero() {
331
+ t.Fatalf("nextDiscoveryRefreshAt = %v, want zero", state.nextDiscoveryRefreshAt)
332
+ }
333
+}
334
+
335
func TestRecordDiscoveryFailurePoolBansLongUnhealthyRelay(t *testing.T) {
336
set := NewRelaySet(nil)
337
338
relayURL := "https://relay-unhealthy.example"
238
- state := confirmedPolicyRelayState(t, relayURL)
339
+ state := confirmedRelayState(t, relayURL)
340
state.unhealthySince = time.Now().UTC().Add(-AnnounceMaxValidity - time.Minute)
341
342
set.mu.Lock()
@@ -268,7 +369,7 @@ func TestRecordActiveFailureDoesNotPoolBanLongUnhealthyRelay(t *testing.T) {
369
set := NewRelaySet(nil)
370
371
relayURL := "https://relay-active-unhealthy.example"
271
- state := confirmedPolicyRelayState(t, relayURL)
372
+ state := confirmedRelayState(t, relayURL)
373
state.unhealthySince = time.Now().UTC().Add(-AnnounceMaxValidity - time.Minute)
374
375
set.mu.Lock()
@@ -301,7 +402,7 @@ func TestPoolBanRejectsDiscoveryUntilExpiry(t *testing.T) {
402
set := NewRelaySet(nil)
403
404
relayURL := "https://relay-quarantined.example"
304
- desc := mustPolicyRelayDescriptor(t, relayURL)
405
+ desc := mustRelayDescriptor(t, relayURL)
406
407
set.mu.Lock()
408
state := newRelayState(relayURL)
portal/discovery/relaystate.go
+1
-3
@@ -165,7 +165,7 @@ func (state RelayState) hasObservedDescriptor() bool {
165
type RouteState struct {
166
ExplicitRelayURLs []string
167
// MaxActiveRelays caps auto-selected relays. Zero or negative values use
168
- // the policy default of 3.
168
+ // the selection default of 3.
169
MaxActiveRelays int
170
MultiHopDepth int
171
RequireUDP bool
@@ -174,5 +174,3 @@ type RouteState struct {
174
// derive a deterministic row index into the GF(64) MOLS grid.
175
LocalAddress string
176
}
177
-
178
-type ClientState = RouteState
portal/discovery/stress_test.go
deleted
-25
@@ -1,25 +0,0 @@
1
-package discovery
2
-
3
-import (
4
- "fmt"
5
- "testing"
6
- "time"
7
-
8
- "github.com/gosuda/portal-tunnel/v2/types"
9
-)
10
-
11
-func TestStressScenarioMassiveScale(t *testing.T) {
12
- clients := 2000
13
- numRelays := 256
14
- relayStates := make([]RelayState, numRelays)
15
- for i := range relayStates {
16
- relayStates[i] = RelayState{Descriptor: types.RelayDescriptor{APIHTTPSAddr: fmt.Sprintf("https://test-%d.example", i)}}
17
- }
18
- policy := MOLSRelayPolicy{}
19
- start := time.Now()
20
- for i := 0; i < clients; i++ {
21
- cs := ClientState{LocalAddress: fmt.Sprintf("client-%d", i)}
22
- policy.SelectPriority(relayStates, cs)
23
- }
24
- fmt.Printf("Massive Scale Test: Avg selection time: %v\n", time.Since(start)/time.Duration(clients))
25
-}
sdk/expose_test.go
+1
-1
@@ -208,7 +208,7 @@ func TestExposureRemoveRelayStopsRunningListener(t *testing.T) {
208
if got := exposure.Config().RelayURLs; len(got) != 0 {
209
t.Fatalf("RelayURLs = %v, want empty", got)
210
}
211
- if got := exposure.relaySet.PriorityRelays(discovery.ClientState{}); len(got) != 0 {
211
+ if got := exposure.relaySet.PriorityRelays(discovery.RouteState{}); len(got) != 0 {
212
t.Fatalf("PriorityRelays() = %v, want empty", got)
213
}
214
relays := exposure.relaySet.AllRelays()